// Legacy sync event system. It should stay unaware of core EventV2 execution;

// the only temporary V2 coupling here is exposing versioned core event schemas

// in effectPayloads() so existing HTTP/SDK schema generation remains stable.

// Remove that registry read when event schemas are generated from core directly.

import { Database } from "@/storage/db"

import { eq } from "drizzle-orm"

import { GlobalBus } from "@/bus/global"

import { Bus as ProjectBus } from "@/bus"

import { BusEvent } from "@/bus/bus-event"

import type { InstanceContext } from "@/project/instance-context"

import { EventSequenceTable, EventTable } from "./event.sql"

import type { WorkspaceID } from "@/control-plane/schema"

import { EventID } from "./schema"

import { Context, Effect, Layer, Schema as EffectSchema } from "effect"

import type { DeepMutable } from "@opencode-ai/core/schema"

import { EventV2 } from "@opencode-ai/core/event"

import { serviceUse } from "@/effect/service-use"

import { InstanceState } from "@/effect/instance-state"

import { RuntimeFlags } from "@/effect/runtime-flags"

import { attachWith } from "@/effect/run-service"

// Keep `Event["data"]` mutable because projectors mutate the persisted shape

// when writing to the database. Bus payloads (`Properties`) stay readonly —

// subscribers only read.

export type Definition<

Type extends string = string,

Schema extends EffectSchema.Top = EffectSchema.Top,

BusSchema extends EffectSchema.Top = Schema,

> = {

type: Type

version: number

aggregate: string

schema: Schema

// Bus event payload schema. Defaults to `schema` unless `busSchema` was

// passed at definition time (see `session.updated`, whose projector

// expands the persisted data to a `{ sessionID, info }` bus payload).

properties: BusSchema

}

export type Event<Def extends Definition = Definition> = {

id: string

seq: number

aggregateID: string

data: DeepMutable<EffectSchema.Schema.Type<Def["schema"]>>

}

export type Properties<Def extends Definition = Definition> = EffectSchema.Schema.Type<Def["properties"]>

export type SerializedEvent<Def extends Definition = Definition> = Event<Def> & { type: string }

type ProjectorFunc = (db: Database.TxOrDb, data: unknown, event: Event) => void

type ConvertEvent = (type: string, data: Event["data"]) => unknown | Promise<unknown>

type PublishContext = {

instance?: InstanceContext

workspace?: WorkspaceID

}

export interface Interface {

readonly run: <Def extends Definition>(

def: Def,

data: Event<Def>["data"],

options?: { publish?: boolean },

) => Effect.Effect<void>

readonly replay: (event: SerializedEvent, options?: { publish: boolean; ownerID?: string }) => Effect.Effect<void>

readonly replayAll: (

events: SerializedEvent[],

options?: { publish: boolean; ownerID?: string },

) => Effect.Effect<string | undefined>

readonly remove: (aggregateID: string) => Effect.Effect<void>

readonly claim: (aggregateID: string, ownerID: string) => Effect.Effect<void>

}

export class Service extends Context.Service<Service, Interface>()("@opencode/SyncEvent") {}

export const layer = Layer.effect(Service)(

Effect.gen(function* () {

const flags = yield* RuntimeFlags.Service

const bus = yield* ProjectBus.Service

const replay: Interface["replay"] = Effect.fn("SyncEvent.replay")(function* (event, options) {

const def = registry.get(event.type)

if (!def) {

throw new Error(`Unknown event type: ${event.type}`)

}

const row = Database.use((db) =>

db

.select({ seq: EventSequenceTable.seq, ownerID: EventSequenceTable.owner_id })

.from(EventSequenceTable)

.where(eq(EventSequenceTable.aggregate_id, event.aggregateID))

.get(),

)

const latest = row?.seq ?? -1

if (event.seq <= latest) return

if (row?.ownerID && row.ownerID !== options?.ownerID) {

return

}

const expected = latest + 1

if (event.seq !== expected) {

throw new Error(

`Sequence mismatch for aggregate "${event.aggregateID}": expected ${expected}, got ${event.seq}`,

)

}

const publish = !!options?.publish

const context = publish

? {

instance: yield* InstanceState.context,

workspace: yield* InstanceState.workspaceID,

}

: undefined

process(def, event, {

bus,

publish,

context,

ownerID: options?.ownerID,

experimentalWorkspaces: flags.experimentalWorkspaces,

})

})

const replayAll: Interface["replayAll"] = Effect.fn("SyncEvent.replayAll")(function* (events, options) {

const source = events[0]?.aggregateID

if (!source) return undefined

if (events.some((item) => item.aggregateID !== source)) {

throw new Error("Replay events must belong to the same session")

}

const start = events[0].seq

for (const [i, item] of events.entries()) {

const seq = start + i

if (item.seq !== seq) {

throw new Error(`Replay sequence mismatch at index ${i}: expected ${seq}, got ${item.seq}`)

}

}

for (const item of events) {

yield* replay(item, options)

}

return source

})

const run: Interface["run"] = Effect.fn("SyncEvent.run")(function* (def, data, options) {

const agg = (data as Record<string, string>)[def.aggregate]

// This should never happen: we've enforced it via typescript in

// the definition

if (agg == null) {

throw new Error(`SyncEvent.run: "${def.aggregate}" required but not found: ${JSON.stringify(data)}`)

}

if (def.version !== versions.get(def.type)) {

throw new Error(`SyncEvent.run: running old versions of events is not allowed: ${def.type}`)

}

const { publish = true } = options || {}

const context = publish

? {

instance: yield* InstanceState.context,

workspace: yield* InstanceState.workspaceID,

}

: undefined

// Note that this is an "immediate" transaction which is critical.

// We need to make sure we can safely read and write with nothing

// else changing the data from under us

Database.transaction(

(tx) => {

const id = EventID.ascending()

const row = tx

.select({ seq: EventSequenceTable.seq })

.from(EventSequenceTable)

.where(eq(EventSequenceTable.aggregate_id, agg))

.get()

const seq = row?.seq != null ? row.seq + 1 : 0

const event = { id, seq, aggregateID: agg, data }

process(def, event, { bus, publish, context, experimentalWorkspaces: flags.experimentalWorkspaces })

},

{

behavior: "immediate",

},

)

})

const remove: Interface["remove"] = Effect.fn("SyncEvent.remove")(function* (aggregateID) {

Database.transaction((tx) => {

tx.delete(EventSequenceTable).where(eq(EventSequenceTable.aggregate_id, aggregateID)).run()

tx.delete(EventTable).where(eq(EventTable.aggregate_id, aggregateID)).run()

})

})

const claim: Interface["claim"] = Effect.fn("SyncEvent.claim")((aggregateID, ownerID) =>

Effect.sync(() =>

Database.use((db) =>

db

.update(EventSequenceTable)

.set({ owner_id: ownerID })

.where(eq(EventSequenceTable.aggregate_id, aggregateID))

.run(),

),

),

)

return Service.of({

run,

replay,

replayAll,

remove,

claim,

})

}),

)

export const defaultLayer = layer.pipe(Layer.provide([ProjectBus.defaultLayer, RuntimeFlags.defaultLayer]))

export const use = serviceUse(Service)

export const registry = new Map<string, Definition>()

let projectors: Map<string, ProjectorFunc> | undefined

const versions = new Map<string, number>()

let frozen = false

let convertEvent: ConvertEvent

export function reset() {

frozen = false

projectors = undefined

convertEvent = (_, data) => data

}

export function init(input: { projectors: Array<[Definition, ProjectorFunc]>; convertEvent?: ConvertEvent }) {

projectors = new Map(input.projectors.map(([def, func]) => [versionedType(def.type, def.version), func]))

for (let entry of EventV2.registry.values()) {

if (!entry.version || !entry.aggregate) continue

register({

type: entry.type,

version: entry.version,

aggregate: entry.aggregate,

properties: entry.data,

schema: entry.data,

})

}

// Install all the latest event defs to the bus. We only ever emit

// latest versions from code, and keep around old versions for

// replaying. Replaying does not go through the bus, and it

// simplifies the bus to only use unversioned latest events

for (let [type, version] of versions.entries()) {

let def = registry.get(versionedType(type, version))!

BusEvent.define(def.type, def.properties)

}

// Freeze the system so it clearly errors if events are defined

// after `init` which would cause bugs

frozen = true

convertEvent = input.convertEvent ?? ((_, data) => data)

}

export function versionedType<A extends string>(type: A): A

export function versionedType<A extends string, B extends number>(type: A, version: B): `${A}/${B}`

export function versionedType(type: string, version?: number) {

return version ? `${type}.${version}` : type

}

export function define<

Type extends string,

Agg extends string,

Schema extends EffectSchema.Top,

BusSchema extends EffectSchema.Top = Schema,

>(input: {

type: Type

version: number

aggregate: Agg

schema: Schema

busSchema?: BusSchema

}): Definition<Type, Schema, BusSchema> {

if (frozen) {

throw new Error("Error defining sync event: sync system has been frozen")

}

const def = {

type: input.type,

version: input.version,

aggregate: input.aggregate,

schema: input.schema,

properties: (input.busSchema ?? input.schema) as BusSchema,

}

register(def)

return def

}

export function project<Def extends Definition>(

def: Def,

func: (db: Database.TxOrDb, data: Event<Def>["data"], event: Event<Def>) => void,

): [Definition, ProjectorFunc] {

return [def, func as ProjectorFunc]

}

function register(def: Definition) {

versions.set(def.type, Math.max(def.version, versions.get(def.type) || 0))

registry.set(versionedType(def.type, def.version), def)

}

function process<Def extends Definition>(

def: Def,

event: Event<Def>,

options: {

bus: ProjectBus.Interface

publish: boolean

context?: PublishContext

ownerID?: string

experimentalWorkspaces: boolean

},

) {

if (projectors == null) {

throw new Error("No projectors available. Call `SyncEvent.init` to install projectors")

}

const projector = projectors.get(versionedType(def.type, def.version))

if (!projector) {

if (!def.type.includes("next")) throw new Error(`Projector not found for event: ${def.type}`)

return

}

Database.transaction((tx) => {

projector(tx, event.data, event)

if (options.experimentalWorkspaces) {

tx.insert(EventSequenceTable)

.values({

aggregate_id: event.aggregateID,

seq: event.seq,

owner_id: options?.ownerID,

})

.onConflictDoUpdate({

target: EventSequenceTable.aggregate_id,

set: { seq: event.seq },

})

.run()

tx.insert(EventTable)

.values({

id: event.id,

seq: event.seq,

aggregate_id: event.aggregateID,

type: versionedType(def.type, def.version),

data: event.data as Record<string, unknown>,

})

.run()

}

Database.effect(() => {

if (options?.publish) {

if (!options.context?.instance) {

throw new Error("SyncEvent.process: publish requires instance context")

}

const result = convertEvent(def.type, event.data)

const publish = (data: unknown) =>

Effect.runPromise(

attachWith(options.bus.publish(def, data as Properties<Def>, { id: event.id }), {

instance: options.context?.instance,

workspace: options.context?.workspace,

}),

)

if (result instanceof Promise) {

void result.then(publish)

} else {

void publish(result)

}

GlobalBus.emit("event", {

directory: options.context.instance.directory,

project: options.context.instance.project.id,

workspace: options.context.workspace,

payload: {

type: "sync",

syncEvent: {

type: versionedType(def.type, def.version),

...event,

},

},

})

}

})

})

}

export function effectPayloads() {

return [

...registry

.entries()

.map(([type, def]) =>

EffectSchema.Struct({

type: EffectSchema.Literal("sync"),

name: EffectSchema.Literal(type),

id: EffectSchema.String,

seq: EffectSchema.Finite,

aggregateID: EffectSchema.Literal(def.aggregate),

data: def.schema,

}).annotate({ identifier: `SyncEvent.${type}` }),

)

.toArray(),

...EventV2.registry

.values()

.filter(

(definition) =>

definition.version !== undefined && !registry.has(versionedType(definition.type, definition.version)),

)

.map((definition) =>

EffectSchema.Struct({

type: EffectSchema.Literal("sync"),

name: EffectSchema.Literal(versionedType(definition.type, definition.version!)),

id: EffectSchema.String,

seq: EffectSchema.Finite,

aggregateID: EffectSchema.Literal(definition.aggregate!),

data: definition.data,

}).annotate({ identifier: `SyncEvent.${definition.type}` }),

)

.toArray(),

]

}

export * as SyncEvent from "."