Something went wrong. Try again.
opencode v2 mirror
Something went wrong. Try again.
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909export * as Bus from "./bus.js"
import { Cause, Clock, Context, Effect, Layer, Option, PubSub, Schema, Stream } from "effect"import { Event } from "@opencode/schema/event"import type { EventLog } from "@opencode/schema/event-log"import { and, asc, eq, gt, lte, sql } from "drizzle-orm"import { Database } from "./database/database.js"import { EventSequenceTable, EventTable } from "./event/sql.js"import type { Location } from "@opencode/schema/location"import { KeyedMutex } from "./effect/keyed-mutex.js"import { makeGlobalNode } from "@opencode/util/effect/app-node"import { isDeepStrictEqual } from "node:util"import { Durable } from "@opencode/schema/durable-event-manifest"import { SessionEvent } from "@opencode/schema/session-event"import type { SessionID } from "@opencode/schema/session-id"import { AbsolutePath } from "@opencode/schema/schema"
export type Subscriber<D extends Event.Definition = Event.Definition> = (event: Event.Payload<D>) => Effect.Effect<void>export type Unsubscribe = Effect.Effect<void>
export const latestSequence = Effect.fn("Bus.latestSequence")(function* ( db: Database.Interface["db"], aggregateID: string,) { const row = yield* db .select({ seq: EventSequenceTable.seq }) .from(EventSequenceTable) .where(eq(EventSequenceTable.aggregate_id, aggregateID)) .get() .pipe(Effect.orDie) return row?.seq ?? -1})
export const reserveSequence = Effect.fn("Bus.reserveSequence")(function* ( db: Database.Interface["db"], aggregateID: string, seq: number,) { yield* db .insert(EventSequenceTable) .values([{ aggregate_id: aggregateID, seq }]) .onConflictDoUpdate({ target: EventSequenceTable.aggregate_id, set: { seq: sql`max(${EventSequenceTable.seq}, ${seq})` }, }) .run() .pipe(Effect.orDie)})
export type SerializedEvent = { readonly id: Event.ID readonly type: string readonly created?: number readonly seq: number readonly aggregateID: string readonly data: Record<string, unknown>}
export class InvalidDurableEventError extends Schema.TaggedError<InvalidDurableEventError>()( "Bus.InvalidDurableEvent", { type: Schema.String, message: Schema.String, },) {}
const envelope = (aggregateID: string, seq: number, version: number) => ({ aggregateID, seq: Event.Seq.make(seq), version: Event.Version.make(version),})
const decodeSerializedEvent = (event: SerializedEvent): Event.Payload => { const definition = Durable.get(event.type) if (!definition?.durable) { throw new InvalidDurableEventError({ type: event.type, message: `Unknown durable event type ${event.type}` }) } return { id: event.id, created: event.created ?? 0, type: definition.type, durable: envelope(event.aggregateID, event.seq, definition.durable.version), data: Schema.decodeUnknownSync(definition.data)(event.data), }}
export const versionedType = Event.versionedTypeexport const durable = Event.durableexport const ephemeral = Event.ephemeral
export interface PublishOptions { readonly id?: Event.ID readonly metadata?: Record<string, unknown> readonly location?: Location.Ref readonly global?: boolean /** Local operational projection committed atomically with a new durable event. Not replayed or serialized. */ readonly commit?: (seq: number) => Effect.Effect<void>}
export type PublishInput<D extends Event.DurableDefinition = Event.DurableDefinition> = readonly [ definition: D, data: Event.Data<D>, options?: PublishOptions,]
export type PublishResult<I extends readonly PublishInput[]> = { readonly [K in keyof I]: I[K] extends PublishInput<infer D> ? Event.Payload<D> : never}
/** Marker/event union emitted by `log`. */export type LogItem = Event.Payload | EventLog.Synced
export const isSynced = (item: LogItem): item is EventLog.Synced => item.type === "log.synced"
export type SubscribePayload<D extends readonly Event.Definition[]> = D[number] extends infer Item ? Item extends Event.Definition ? Event.Payload<Item> : never : never
export interface Subscribe { /** * Volatile live channel: every event published from now on, nothing before or * across a disconnect. Consumers that need reliability combine it with `log`. * With an ambient Location, delivery is restricted to that Location and global * events. Unlocated Session events use the Session's owner at publication time. * Session moves reach both the old and new Location, without changing the event. */ (): Stream.Stream<Event.Payload> <D extends Event.Definition>(definition: D): Stream.Stream<Event.Payload<D>> <const D extends readonly [Event.Definition, ...Event.Definition[]]>( definitions: D, ): Stream.Stream<SubscribePayload<D>>}
const isDefinition = (input: Event.Definition | readonly Event.Definition[]): input is Event.Definition => !Array.isArray(input)
export interface Interface { readonly publish: <D extends Event.Definition>( definition: D, data: Event.Data<D>, options?: PublishOptions, ) => Effect.Effect<Event.Payload<D>> readonly publishAll: <const I extends readonly [PublishInput, ...PublishInput[]]>( events: I, ) => Effect.Effect<PublishResult<I>> readonly subscribe: Subscribe /** * Durable, ordered per-aggregate log read. Forked aggregates may reserve an * inherited prefix before their first child-authored event. `follow: false` * completes at the end of the log; `follow: true` replays then transitions * to live. Both modes emit one `Synced` marker at the captured replay * watermark. */ readonly log: (input: { readonly aggregateID: string readonly after?: number readonly follow?: boolean }) => Stream.Stream<LogItem> /** @deprecated Use `subscribe()` and consume the returned stream. */ readonly listen: (listener: Subscriber) => Effect.Effect<Unsubscribe> readonly project: <D extends Event.Definition>(definition: D, projector: Subscriber<D>) => Effect.Effect<void> readonly replay: ( event: SerializedEvent, options?: { readonly publish?: boolean; readonly ownerID?: string; readonly strictOwner?: boolean }, ) => Effect.Effect<void> 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/Bus") {}
interface Options { readonly beforeAggregateRead?: (aggregateID: string) => Effect.Effect<void> /** Maximum durable rows read per page while replaying or tailing an aggregate log. */ readonly logReadPageSize?: number /** Retain durable event payloads for historical log reads and replay. */ readonly persist?: boolean}
export function configured(options?: Options) { return makeGlobalNode({ service: Service, deps: [Database.node], layer: Layer.effect( Service, Effect.gen(function* () { // Deferred import: a static one would close the module cycle // bus → location → project → bus and hit the node bindings in TDZ. const { Location } = yield* Effect.promise(() => import("./location.js")) const { SessionTable } = yield* Effect.promise(() => import("./session/sql.js")) const pubsub = { live: yield* PubSub.unbounded<Event.Payload>(), durable: new Map<string, Set<PubSub.PubSub<void>>>(), typed: new Map<string, PubSub.PubSub<Event.Payload>>(), } const projectors = new Map<string, Subscriber[]>() const listeners = new Array<Subscriber>() const durableLocks = KeyedMutex.makeUnsafe<string>() const { db } = yield* Database.Service const logReadPageSize = options?.logReadPageSize ?? 512 const persist = options?.persist ?? false const sessions = new Map<SessionID, Location.Ref>() // Keep routing separate from the public event, and retain its snapshot // while a slow subscriber drains events queued before a move or deletion. const routes = new WeakMap<Event.Payload, readonly Location.Ref[]>()
const isSessionEvent = (event: Event.Payload): event is SessionEvent.Event => Object.hasOwn(SessionEvent.All.cases, event.type)
const prepareRoutes = Effect.fnUntraced(function* (events: readonly Event.Payload[]) { const updates = new Map<SessionID, Location.Ref | undefined>() const resolved = new Map<Event.Payload, readonly Location.Ref[]>() for (const event of events) { if (!isSessionEvent(event)) continue const id = event.data.sessionID if (event.type === "session.created") { updates.set(id, event.data.location) resolved.set(event, [event.location ?? event.data.location]) continue } if (event.location && event.type !== "session.forked" && event.type !== "session.moved") { if (event.type === "session.deleted") updates.set(id, undefined) continue } const owner = event.type === "session.forked" ? event.data.parentID : id let ref = updates.has(owner) ? updates.get(owner) : sessions.get(owner) if (!ref && !updates.has(owner)) { const row = yield* db .select({ directory: SessionTable.directory, workspaceID: SessionTable.workspace_id }) .from(SessionTable) .where(eq(SessionTable.id, owner)) .get() .pipe(Effect.orDie) ref = row ? { directory: AbsolutePath.make(row.directory), workspaceID: row.workspaceID ?? undefined } : undefined updates.set(owner, ref) } if (event.type === "session.moved") { // Both owners need the transition, even if the producer supplied // an envelope location. Later events use only the destination. updates.set(id, event.data.location) resolved.set(event, ref ? [ref, event.data.location] : [event.data.location]) continue } if (event.type === "session.forked") updates.set(id, ref) resolved.set(event, event.location ? [event.location] : ref ? [ref] : []) if (event.type === "session.deleted") updates.set(id, undefined) } // Apply only after the projection transaction commits. A failed move // must not redirect events away from the Session's actual location. return () => { for (const [id, ref] of updates) { if (ref) sessions.set(id, ref) else sessions.delete(id) } for (const [event, ref] of resolved) routes.set(event, ref) } })
const getOrCreate = (definition: Event.Definition) => Effect.gen(function* () { const existing = pubsub.typed.get(definition.type) if (existing) return existing const created = yield* PubSub.unbounded<Event.Payload>() pubsub.typed.set(definition.type, created) return created })
yield* Effect.addFinalizer(() => Effect.gen(function* () { yield* PubSub.shutdown(pubsub.live) yield* Effect.forEach( pubsub.durable.values(), (pubsubs) => Effect.forEach(pubsubs, PubSub.shutdown, { discard: true }), { discard: true }, ) yield* Effect.forEach(pubsub.typed.values(), PubSub.shutdown, { discard: true }) }), )
function commitDurableEvent( definition: Event.Definition, event: Event.Payload, input?: { readonly seq: number readonly aggregateID: string readonly ownerID?: string readonly strictOwner?: boolean }, commit?: (seq: number) => Effect.Effect<void>, ) { return Effect.gen(function* () { const durable = definition.durable if (!durable) return yield* Effect.void const aggregateID = (event.data as Record<string, unknown>)[durable.aggregate] if (typeof aggregateID !== "string") return yield* Effect.die( new InvalidDurableEventError({ type: event.type, message: `Expected string aggregate field ${durable.aggregate}`, }), ) if (input && input.aggregateID !== aggregateID) { yield* Effect.die( new InvalidDurableEventError({ type: event.type, message: `Aggregate mismatch: expected ${input.aggregateID}, got ${aggregateID}`, }), ) } const list = projectors.get(versionedType(definition.type, durable.version)) ?? [] return yield* Effect.uninterruptible( Effect.gen(function* () { const committed = yield* db .transaction( () => Effect.gen(function* () { const row = yield* db .select({ seq: EventSequenceTable.seq, ownerID: EventSequenceTable.owner_id }) .from(EventSequenceTable) .where(eq(EventSequenceTable.aggregate_id, aggregateID)) .get() .pipe(Effect.orDie) const latest = row?.seq ?? -1 const encoded = Schema.encodeUnknownSync(definition.data)(event.data) as Record<string, unknown> if (input?.strictOwner && row?.ownerID && row.ownerID !== input.ownerID) { yield* Effect.die( new InvalidDurableEventError({ type: event.type, message: `Replay owner mismatch for aggregate ${aggregateID}: expected ${row.ownerID}, got ${input.ownerID ?? "none"}`, }), ) } if (input && input.seq <= latest) { if (!persist) return const stored = yield* db .select() .from(EventTable) .where(and(eq(EventTable.aggregate_id, aggregateID), eq(EventTable.seq, input.seq))) .get() .pipe(Effect.orDie) if ( stored?.id === event.id && stored.type === versionedType(definition.type, durable.version) && stored.created === (event.created ?? 0) && isDeepStrictEqual(stored.data, encoded) ) { if (input.ownerID && row?.ownerID == null) { yield* db .update(EventSequenceTable) .set({ owner_id: input.ownerID }) .where(eq(EventSequenceTable.aggregate_id, aggregateID)) .run() .pipe(Effect.orDie) } return } yield* Effect.die( new InvalidDurableEventError({ type: event.type, message: `Replay diverged at aggregate ${aggregateID} sequence ${input.seq}`, }), ) } if (input && row?.ownerID && row.ownerID !== input.ownerID) { return } const seq = input?.seq ?? latest + 1 if (input && seq !== latest + 1) { yield* Effect.die( new InvalidDurableEventError({ type: event.type, message: `Sequence mismatch for aggregate ${aggregateID}: expected ${latest + 1}, got ${seq}`, }), ) } if (persist) { const stored = yield* db .select({ aggregateID: EventTable.aggregate_id, seq: EventTable.seq }) .from(EventTable) .where(eq(EventTable.id, event.id)) .get() .pipe(Effect.orDie) if (stored) yield* Effect.die( new InvalidDurableEventError({ type: event.type, message: `Event ${event.id} already exists at aggregate ${stored.aggregateID} sequence ${stored.seq}`, }), ) } const committed = { ...event, durable: { aggregateID, seq, version: durable.version }, } as Event.Payload const route = yield* prepareRoutes([committed]) for (const projector of list) { yield* projector(committed) } if (commit) yield* commit(seq) yield* db .insert(EventSequenceTable) .values([{ aggregate_id: aggregateID, seq, owner_id: input?.ownerID }]) .onConflictDoUpdate({ target: EventSequenceTable.aggregate_id, set: { seq: sql`max(${EventSequenceTable.seq}, ${seq})`, ...(input?.ownerID && row?.ownerID == null ? { owner_id: input.ownerID } : {}), }, }) .run() .pipe(Effect.orDie) if (persist) yield* db .insert(EventTable) .values([ { id: event.id, aggregate_id: aggregateID, seq, created: event.created ?? 0, type: versionedType(definition.type, durable.version), data: encoded, }, ]) .run() .pipe(Effect.orDie) return { aggregateID, seq, event: committed, route } }), { behavior: "immediate" }, ) .pipe(Effect.orDie) if (committed) { committed.route() yield* Effect.forEach( pubsub.durable.get(committed.aggregateID) ?? [], (wake) => PubSub.publish(wake, undefined), { discard: true }, ) } return committed }), ) }) }
function publishEvent<D extends Event.Definition>( definition: D, event: Event.Payload<D>, commit?: PublishOptions["commit"], ) { return Effect.gen(function* () { if (!definition.durable && commit) return yield* Effect.die( new InvalidDurableEventError({ type: event.type, message: "Local commit hooks require a durable event", }), ) if (definition.durable) { const aggregateID = (event.data as Record<string, unknown>)[definition.durable.aggregate] if (typeof aggregateID !== "string") return yield* commitDurableEvent(definition, event as Event.Payload, undefined, commit).pipe( Effect.as(event), ) return yield* durableLocks.withLock(aggregateID)( Effect.gen(function* () { const committed = yield* commitDurableEvent(definition, event as Event.Payload, undefined, commit) if (!committed) return event event = committed.event as Event.Payload<D> yield* notify(event as Event.Payload, true) return event }), ) } const route = yield* prepareRoutes([event as Event.Payload]) route() yield* notify(event as Event.Payload, false) return event }) }
const observe = (event: Event.Payload, observer: (event: Event.Payload) => Effect.Effect<void>) => Effect.suspend(() => observer(event)).pipe( Effect.catchCauseIf( (cause) => !Cause.hasInterrupts(cause), (cause) => Effect.logError("Event listener failed", { eventID: event.id, eventType: event.type, cause }), ), )
function notify(event: Event.Payload, isolateListeners: boolean) { return Effect.gen(function* () { yield* Effect.forEach( listeners, (listener) => (isolateListeners ? observe(event, listener) : listener(event)), { discard: true }, ) const typed = pubsub.typed.get(event.type) if (typed) yield* PubSub.publish(typed, event) yield* PubSub.publish(pubsub.live, event) }) }
function publish<D extends Event.Definition>(definition: D, data: Event.Data<D>, options?: PublishOptions) { return Effect.gen(function* () { const serviceLocation = Option.getOrUndefined(yield* Effect.serviceOption(Location.Service)) const location = options?.global ? undefined : (options?.location ?? (serviceLocation ? { directory: serviceLocation.directory, workspaceID: serviceLocation.workspaceID } : undefined)) return yield* publishEvent( definition, { id: options?.id ?? Event.ID.create(), created: yield* Clock.currentTimeMillis, ...(options?.metadata ? { metadata: options.metadata } : {}), type: definition.type, ...(location ? { location } : {}), data, } as Event.Payload<D>, options?.commit, ) }) }
function publishAll<const I extends readonly [PublishInput, ...PublishInput[]]>(events: I) { return Effect.gen(function* () { const serviceLocation = Option.getOrUndefined(yield* Effect.serviceOption(Location.Service)) const payloads = yield* Effect.forEach(events, ([definition, data, options]) => Effect.gen(function* () { const aggregateID = (data as Record<string, unknown>)[definition.durable.aggregate] if (typeof aggregateID !== "string") { return yield* Effect.die( new InvalidDurableEventError({ type: definition.type, message: `Expected string aggregate field ${definition.durable.aggregate}`, }), ) } const location = options?.global ? undefined : (options?.location ?? (serviceLocation ? { directory: serviceLocation.directory, workspaceID: serviceLocation.workspaceID } : undefined)) return { definition, aggregateID, commit: options?.commit, event: { id: options?.id ?? Event.ID.create(), created: yield* Clock.currentTimeMillis, ...(options?.metadata ? { metadata: options.metadata } : {}), type: definition.type, ...(location ? { location } : {}), data, } as Event.Payload, } }), ) const aggregateID = payloads[0].aggregateID if (payloads.some((item) => item.aggregateID !== aggregateID)) { return yield* Effect.die( new InvalidDurableEventError({ type: payloads[0].definition.type, message: "Published events must belong to the same aggregate", }), ) } return yield* durableLocks.withLock(aggregateID)( Effect.uninterruptible( Effect.gen(function* () { const committed = yield* db .transaction( () => Effect.gen(function* () { const row = yield* db .select({ seq: EventSequenceTable.seq }) .from(EventSequenceTable) .where(eq(EventSequenceTable.aggregate_id, aggregateID)) .get() .pipe(Effect.orDie) const firstSeq = (row?.seq ?? -1) + 1 const finalSeq = firstSeq + payloads.length - 1 const queued = payloads.map((item, index) => ({ ...item.event, durable: envelope(aggregateID, firstSeq + index, item.definition.durable.version), })) const route = yield* prepareRoutes(queued) const rows = new Array<typeof EventTable.$inferInsert>() const ids = new Set<Event.ID>() for (const [index, item] of payloads.entries()) { const seq = firstSeq + index const encoded = Schema.encodeUnknownSync(item.definition.data)(item.event.data) as Record< string, unknown > if (persist) { if (ids.has(item.event.id)) yield* Effect.die( new InvalidDurableEventError({ type: item.event.type, message: `Event ${item.event.id} appears more than once in the batch`, }), ) ids.add(item.event.id) const stored = yield* db .select({ aggregateID: EventTable.aggregate_id, seq: EventTable.seq }) .from(EventTable) .where(eq(EventTable.id, item.event.id)) .get() .pipe(Effect.orDie) if (stored) yield* Effect.die( new InvalidDurableEventError({ type: item.event.type, message: `Event ${item.event.id} already exists at aggregate ${stored.aggregateID} sequence ${stored.seq}`, }), ) } const event = queued[index] for (const projector of projectors.get( versionedType(item.definition.type, item.definition.durable.version), ) ?? []) { yield* projector(event) } if (item.commit) yield* item.commit(seq) if (persist) rows.push({ id: event.id, aggregate_id: aggregateID, seq, created: event.created, type: versionedType(item.definition.type, item.definition.durable.version), data: encoded, }) } yield* db .insert(EventSequenceTable) .values([{ aggregate_id: aggregateID, seq: finalSeq }]) .onConflictDoUpdate({ target: EventSequenceTable.aggregate_id, set: { seq: finalSeq } }) .run() .pipe(Effect.orDie) if (persist) yield* db.insert(EventTable).values(rows).run().pipe(Effect.orDie) return { events: queued, route } }), { behavior: "immediate" }, ) .pipe(Effect.orDie) committed.route() yield* Effect.forEach( pubsub.durable.get(aggregateID) ?? [], (wake) => PubSub.publish(wake, undefined), { discard: true, }, ) yield* Effect.forEach(committed.events, (event) => notify(event, true), { discard: true }) return committed.events as PublishResult<I> }), ), ) }) }
function replay( event: SerializedEvent, options?: { readonly publish?: boolean; readonly ownerID?: string; readonly strictOwner?: boolean }, ) { return Effect.gen(function* () { const definition = Durable.get(event.type) if (!definition?.durable) return yield* Effect.die( new InvalidDurableEventError({ type: event.type, message: `Unknown durable event type ${event.type}` }), ) yield* durableLocks.withLock(event.aggregateID)( Effect.gen(function* () { const payload = { id: event.id, created: event.created ?? 0, type: definition.type, data: Schema.decodeUnknownSync(definition.data)(event.data), } as Event.Payload const committed = yield* commitDurableEvent(definition, payload, { seq: event.seq, aggregateID: event.aggregateID, ownerID: options?.ownerID, strictOwner: options?.strictOwner, }) if (committed && options?.publish) { yield* notify(committed.event, true) } }), ) }) }
function remove(aggregateID: string) { return db .transaction(() => Effect.gen(function* () { yield* db.delete(EventSequenceTable).where(eq(EventSequenceTable.aggregate_id, aggregateID)).run() yield* db.delete(EventTable).where(eq(EventTable.aggregate_id, aggregateID)).run() }), ) .pipe( Effect.tap(() => Effect.sync(() => sessions.delete(aggregateID as SessionID))), Effect.orDie, ) }
function claim(aggregateID: string, ownerID: string) { return db .update(EventSequenceTable) .set({ owner_id: ownerID }) .where(eq(EventSequenceTable.aggregate_id, aggregateID)) .run() .pipe(Effect.orDie) }
const local = <A extends Event.Payload>(stream: Stream.Stream<A>) => Stream.unwrap( Effect.serviceOption(Location.Service).pipe( Effect.map((location) => Option.match(location, { onNone: () => stream, onSome: (location) => { const matches = (ref: Location.Ref) => ref.directory === location.directory && ref.workspaceID === location.workspaceID return stream.pipe( Stream.filter((event) => { const refs = routes.get(event) if (refs) return refs.some(matches) return !event.location || matches(event.location) }), ) }, }), ), ), )
function subscribe(): Stream.Stream<Event.Payload> function subscribe<D extends Event.Definition>(definition: D): Stream.Stream<Event.Payload<D>> function subscribe<const D extends readonly [Event.Definition, ...Event.Definition[]]>( definitions: D, ): Stream.Stream<SubscribePayload<D>> function subscribe(input?: Event.Definition | readonly Event.Definition[]): Stream.Stream<Event.Payload> { if (input === undefined) return streamLive() if (isDefinition(input)) { return local(Stream.unwrap(getOrCreate(input).pipe(Effect.map((pubsub) => Stream.fromPubSub(pubsub))))) } const types = new Set(input.map((definition) => definition.type)) return streamLive().pipe(Stream.filter((event) => types.has(event.type))) }
const streamLive = (): Stream.Stream<Event.Payload> => local(Stream.fromPubSub(pubsub.live))
const readAfter = ( aggregateID: string, after: number, input: { readonly through: number; readonly limit: number }, ) => (options?.beforeAggregateRead?.(aggregateID) ?? Effect.void).pipe( Effect.andThen( Effect.suspend(() => { const query = db .select() .from(EventTable) .where( and( eq(EventTable.aggregate_id, aggregateID), gt(EventTable.seq, after), lte(EventTable.seq, input.through), ), ) .orderBy(asc(EventTable.seq)) return query.limit(input.limit).all() }), ), Effect.orDie, // Skip types missing from the durable manifest instead of failing the // read: the aggregate may hold events this process cannot decode. The // raw tail seq keeps cursors advancing across the resulting gaps. Effect.map((rows) => ({ seq: rows.at(-1)?.seq, events: rows.flatMap((event) => { if (!Durable.get(event.type)?.durable) return [] return [ decodeSerializedEvent({ id: event.id, created: event.created, aggregateID: event.aggregate_id, seq: event.seq, type: event.type, data: event.data, }), ] }), })), )
const subscribeDurable = (aggregateID: string) => Effect.gen(function* () { const wake = yield* PubSub.sliding<void>(1) const subscription = yield* PubSub.subscribe(wake) yield* Effect.acquireRelease( Effect.sync(() => { const wakes = pubsub.durable.get(aggregateID) ?? new Set() wakes.add(wake) pubsub.durable.set(aggregateID, wakes) }), () => Effect.sync(() => { const wakes = pubsub.durable.get(aggregateID) wakes?.delete(wake) if (wakes?.size === 0) pubsub.durable.delete(aggregateID) }).pipe(Effect.andThen(PubSub.shutdown(wake))), ) return subscription })
const log = (input: { readonly aggregateID: string readonly after?: number readonly follow?: boolean }): Stream.Stream<LogItem> => Stream.unwrap( Effect.gen(function* () { let sequence = input.after ?? -1 const readThrough = (through: number): Stream.Stream<Event.Payload> => Stream.paginate(sequence, (cursor) => readAfter(input.aggregateID, cursor, { through, limit: logReadPageSize }).pipe( Effect.tap((page) => Effect.sync(() => { sequence = page.seq ?? sequence }), ), Effect.map( (page) => [ page.events, page.seq !== undefined && page.seq < through ? Option.some(page.seq) : Option.none<number>(), ] as const, ), ), ) // Subscribing before the historical read means events committed during // replay either appear in the read or arrive through a post-marker wake. const wakes = input.follow ? yield* subscribeDurable(input.aggregateID) : undefined const target = yield* latestSequence(db, input.aggregateID) const marker: EventLog.Synced = { type: "log.synced", aggregateID: input.aggregateID, ...(target >= 0 ? { seq: Event.Seq.make(target) } : {}), } const replay: Stream.Stream<LogItem> = readThrough(target).pipe(Stream.concat(Stream.make(marker))) if (!wakes) return replay const live: Stream.Stream<LogItem> = Stream.fromSubscription(wakes).pipe( Stream.mapEffect(() => latestSequence(db, input.aggregateID)), Stream.filter((target) => target > sequence), Stream.flatMap((target) => readThrough(target)), ) return Stream.concat(replay, live) }), )
const listen = (listener: Subscriber): Effect.Effect<Unsubscribe> => Effect.sync(() => { listeners.push(listener) return Effect.sync(() => { const index = listeners.indexOf(listener) if (index >= 0) listeners.splice(index, 1) }) })
const project = <D extends Event.Definition>(definition: D, projector: Subscriber<D>): Effect.Effect<void> => Effect.sync(() => { const key = definition.durable ? versionedType(definition.type, definition.durable.version) : definition.type const list = projectors.get(key) ?? [] list.push((event) => projector(event as Event.Payload<D>)) projectors.set(key, list) })
return Service.of({ publish, publishAll, subscribe, log, listen, project, replay, remove, claim, }) }), ), })}
export const node = configured()