From 3e92bf5f42d0d8dde03bec424c891f60821050c9 Mon Sep 17 00:00:00 2001 From: Dylan Slack Date: Fri, 23 Oct 2026 12:21:50 +0200 Subject: [PATCH 1/2] feat(projectors): project runs into a Postgres event log without blocking append Durable Object projectors are scheduled with waitUntil after the log commits. postgresEventLog stores each non-ephemeral envelope and a run index with status, launch links, and tenant attributes, and can rebuild that sink from the log. Co-authored-by: Cursor --- .changeset/postgres-event-log.md | 7 + docs/production.md | 22 + packages/cloudflare/src/event-store.ts | 14 +- packages/projectors/src/event-log.test.ts | 150 +++++ packages/projectors/src/event-log.ts | 531 ++++++++++++++++++ packages/projectors/src/postgres.ts | 9 + .../projectors/src/with-projectors.test.ts | 36 ++ packages/projectors/src/with-projectors.ts | 31 +- 8 files changed, 790 insertions(+), 10 deletions(-) create mode 100644 .changeset/postgres-event-log.md create mode 100644 packages/projectors/src/event-log.test.ts create mode 100644 packages/projectors/src/event-log.ts diff --git a/.changeset/postgres-event-log.md b/.changeset/postgres-event-log.md new file mode 100644 index 0000000..8261e3d --- /dev/null +++ b/.changeset/postgres-event-log.md @@ -0,0 +1,7 @@ +--- +'@swirls/looms': minor +--- + +feat(projectors): project runs into a Postgres event log without blocking append + +Durable Object projectors are scheduled with `waitUntil` after the log commits. `postgresEventLog` stores each non-ephemeral envelope and a run index with status, launch links, and tenant attributes, and can rebuild that sink from the log. diff --git a/docs/production.md b/docs/production.md index 866dee0..362789e 100644 --- a/docs/production.md +++ b/docs/production.md @@ -13,6 +13,28 @@ coalesced. A deployment, eviction, or alarm retry may repeat a wake; event preconditions make that safe. Cross-object queries and transactions are not provided. Build global run lists with a projector. +On a Durable Object, projector delivery is scheduled with `waitUntil` after the +append commits, so projector latency is not part of the run step. The cursor is +durable. A delivery that fails, or an isolate that stops before `waitUntil` +finishes, is retried on the next alarm or request. + +`postgresEventLog` is the generic sink: one row per non-ephemeral envelope, +keyed by `(run_id, seq)`, and a `looms_run_index` row with status, timestamps, +the parent run, the launching effect, and the run's attributes. `tenant`, +`project`, and `org` are copied out of string attributes so row-level security +can index them: + +```sql +ALTER TABLE looms_run_index ENABLE ROW LEVEL SECURITY; +CREATE POLICY tenant_isolation ON looms_run_index + USING (tenant = current_setting('looms.tenant', true)); +``` + +Pass Hyperdrive's connection string as `postgresEventLog({ url })`. Writes are +idempotent. `rebuild(source)` replays a run from seq 1; run that while live +delivery is paused, then let the cursor catch up. Ephemeral events are left out +of the log. + All built-in stores reject an event over 1 MiB and a snapshot over 8 MiB, measured as serialized UTF-8, before storage is mutated. The stable error names are `EventPayloadTooLargeError` and `SnapshotPayloadTooLargeError`. diff --git a/packages/cloudflare/src/event-store.ts b/packages/cloudflare/src/event-store.ts index d706d9d..037ca51 100644 --- a/packages/cloudflare/src/event-store.ts +++ b/packages/cloudflare/src/event-store.ts @@ -1,4 +1,4 @@ -import type { Effect } from 'effect' +import { Effect } from 'effect' import type { EventStore } from '@looms/core' import { sqliteEventStore, sqlStorageExec } from '@looms/core/sqlite-store' @@ -21,7 +21,9 @@ export interface DurableObjectEventStore extends EventStore { /** * Creates an EventStore backed by a Durable Object's embedded SQLite database (`ctx.storage.sql`). - * Optionally attaches asynchronous projectors (such as S2). + * Optionally attaches projectors. Their delivery is scheduled with `waitUntil`, so an append + * returns when the log commits. A crashed delivery is retried from the durable cursor on the next + * alarm or request. */ export function durableObjectEventStore( ctx: DurableObjectState, @@ -42,7 +44,13 @@ export function durableObjectEventStore( { scheduleRetry: options.scheduleProjectorRetry }, ) - const wrapped = withProjectors(store, options.projectors, { delivery }) + const wrapped = withProjectors(store, options.projectors, { + delivery, + scheduleDelivery: (_runId, work) => { + ctx.waitUntil(Effect.runPromise(work)) + }, + }) + return Object.assign(wrapped, { projectorDelivery: delivery }) } diff --git a/packages/projectors/src/event-log.test.ts b/packages/projectors/src/event-log.test.ts new file mode 100644 index 0000000..14dd837 --- /dev/null +++ b/packages/projectors/src/event-log.test.ts @@ -0,0 +1,150 @@ +import { describe, expect, test } from 'bun:test' + +import { Effect } from 'effect' + +import { createEvent, makeMemoryEventStore, type EventEnvelope, type EventInput } from '@looms/core' + +import { memoryEventLog, POSTGRES_EVENT_LOG_SCHEMA_SQL } from './event-log' + +function event(input: EventInput, seq: number): EventEnvelope { + return { ...createEvent('run_1', input), seq } +} + +const started = event( + { + type: 'runtime.run.started', + payload: { + rootThreadId: 'thr_1', + kind: 'workflow', + definitionName: 'checkout', + definitionVersion: 'v1', + input: null, + attributes: { tenant: 'acme', project: 'billing', org: 'hq' }, + launchedBy: { + runId: 'run_parent', + threadId: 'thr_parent', + effectId: 'eff_1', + causationId: 'evt_cause', + causationSeq: 4, + }, + }, + correlationId: 'run_root', + threadId: null, + origin: { type: 'system' }, + }, + 1, +) + +describe('event log', () => { + test('keeps non-ephemeral envelopes and the run index', async () => { + const log = memoryEventLog() + + await log.project([ + started, + event( + { + type: 'runtime.run.completed', + payload: { output: { ok: true }, error: null }, + threadId: null, + origin: { type: 'system' }, + }, + 2, + ), + ]) + + expect(log.events('run_1').map((entry) => entry.seq)).toEqual([1, 2]) + + expect(log.run('run_1')).toMatchObject({ + status: 'completed', + kind: 'workflow', + definitionName: 'checkout', + definitionVersion: 'v1', + rootThreadId: 'thr_1', + parentRunId: 'run_parent', + launchedByEffectId: 'eff_1', + correlationId: 'run_root', + tenant: 'acme', + project: 'billing', + org: 'hq', + attributes: { tenant: 'acme', project: 'billing', org: 'hq' }, + }) + }) + + test('skips ephemeral events and does not regress a finished run when start is replayed', async () => { + const log = memoryEventLog() + + const delta = event( + { + type: 'agent.turn.text_delta', + payload: { text: 'hi' }, + threadId: 'thr_1', + origin: { type: 'system' }, + ephemeral: true, + }, + 2, + ) + + const failed = event( + { + type: 'runtime.run.completed', + payload: { output: null, error: 'nope' }, + threadId: null, + origin: { type: 'system' }, + }, + 3, + ) + + await log.project([started, delta, failed]) + await log.project([started, delta, failed]) + + expect(log.events('run_1').map((entry) => entry.type)).toEqual([ + 'runtime.run.started', + 'runtime.run.completed', + ]) + + expect(log.run('run_1')?.status).toBe('failed') + expect(log.run('run_1')?.tenant).toBe('acme') + }) + + test('rebuilds the sink from the source log', async () => { + const source = await Effect.runPromise(makeMemoryEventStore) + + await Effect.runPromise( + source.append('run_1', [ + { + type: 'runtime.run.started', + payload: { + kind: 'workflow', + definitionName: 'checkout', + input: null, + }, + threadId: null, + origin: { type: 'system' }, + }, + { + type: 'runtime.run.cancelled', + payload: { reason: 'stop' }, + threadId: null, + origin: { type: 'system' }, + }, + ]), + ) + + const log = memoryEventLog() + const written = await log.rebuild(source) + + expect(written).toBe(2) + expect(log.run('run_1')?.status).toBe('cancelled') + + const again = await log.rebuild(source) + expect(again).toBe(2) + expect(log.events('run_1')).toHaveLength(2) + }) + + test('the postgres schema keys events by run and seq and indexes tenant columns', () => { + expect(POSTGRES_EVENT_LOG_SCHEMA_SQL).toContain('PRIMARY KEY (run_id, seq)') + expect(POSTGRES_EVENT_LOG_SCHEMA_SQL).toContain('looms_run_index_tenant') + expect(POSTGRES_EVENT_LOG_SCHEMA_SQL).toContain('looms_run_index_project') + expect(POSTGRES_EVENT_LOG_SCHEMA_SQL).toContain('looms_run_index_org') + }) +}) diff --git a/packages/projectors/src/event-log.ts b/packages/projectors/src/event-log.ts new file mode 100644 index 0000000..308f603 --- /dev/null +++ b/packages/projectors/src/event-log.ts @@ -0,0 +1,531 @@ +import { DateTime, Effect, Option, Predicate, Schema } from 'effect' +import createPostgresClient, { type Sql, type TransactionSql } from 'postgres' + +import { + completedRunStatus, + isJsonObject, + readRunLaunchSource, + type EventEnvelope, + type EventOrigin, + type EventStore, + type EventStoreError, + type JsonValue, + type RunStatus, +} from '@looms/core' + +import type { Projector } from './projector' +import { runProjectorPromise } from './projector' + +const decodeAttributes = Schema.decodeUnknownOption(Schema.Record(Schema.String, Schema.String)) + +export const POSTGRES_EVENT_LOG_SCHEMA_SQL = ` +CREATE TABLE IF NOT EXISTS looms_run_events ( + run_id TEXT NOT NULL, + seq BIGINT NOT NULL, + event_id TEXT NOT NULL, + ts TIMESTAMPTZ NOT NULL, + type TEXT NOT NULL, + payload JSONB NOT NULL, + thread_id TEXT, + parent_thread_id TEXT, + causation_id TEXT, + correlation_id TEXT, + effect_id TEXT, + origin JSONB NOT NULL, + idempotency_key TEXT, + PRIMARY KEY (run_id, seq) +); + +CREATE TABLE IF NOT EXISTS looms_run_index ( + run_id TEXT PRIMARY KEY, + status TEXT NOT NULL, + kind TEXT, + definition_name TEXT, + definition_version TEXT, + root_thread_id TEXT, + parent_run_id TEXT, + launched_by_effect_id TEXT, + correlation_id TEXT, + attributes JSONB NOT NULL DEFAULT '{}'::jsonb, + tenant TEXT, + project TEXT, + org TEXT, + started_at TIMESTAMPTZ, + updated_at TIMESTAMPTZ NOT NULL, + completed_at TIMESTAMPTZ +); + +CREATE INDEX IF NOT EXISTS looms_run_index_tenant ON looms_run_index (tenant); +CREATE INDEX IF NOT EXISTS looms_run_index_project ON looms_run_index (project); +CREATE INDEX IF NOT EXISTS looms_run_index_org ON looms_run_index (org); +CREATE INDEX IF NOT EXISTS looms_run_index_parent ON looms_run_index (parent_run_id); +CREATE INDEX IF NOT EXISTS looms_run_index_status ON looms_run_index (status); +`.trim() + +export interface EventLogEntry { + readonly runId: string + readonly seq: number + readonly eventId: string + readonly ts: number + readonly type: string + readonly payload: JsonValue + readonly threadId: string | null + readonly parentThreadId: string | null + readonly causationId: string | null + readonly correlationId: string | null + readonly effectId: string | null + readonly origin: EventOrigin + readonly idempotencyKey: string | null +} + +export interface RunLogRow { + readonly runId: string + readonly status: RunStatus + readonly kind: string | null + readonly definitionName: string | null + readonly definitionVersion: string | null + readonly rootThreadId: string | null + readonly parentRunId: string | null + readonly launchedByEffectId: string | null + readonly correlationId: string | null + readonly attributes: { readonly [key: string]: string } + readonly tenant: string | null + readonly project: string | null + readonly org: string | null + readonly startedAt: number | null + readonly updatedAt: number + readonly completedAt: number | null +} + +export type EventLogOp = + | { readonly type: 'insert'; readonly entry: EventLogEntry } + | { + readonly type: 'start' + readonly runId: string + readonly kind: string | null + readonly definitionName: string | null + readonly definitionVersion: string | null + readonly rootThreadId: string | null + readonly parentRunId: string | null + readonly launchedByEffectId: string | null + readonly correlationId: string | null + readonly attributes: { readonly [key: string]: string } + readonly tenant: string | null + readonly project: string | null + readonly org: string | null + readonly at: number + } + | { + readonly type: 'finish' + readonly runId: string + readonly status: 'completed' | 'failed' | 'cancelled' + readonly at: number + } + | { readonly type: 'touch'; readonly runId: string; readonly at: number } + +export interface EventLogProjector extends Projector { + /** Replays the source log into this sink. Safe to repeat; do not run it beside live delivery. */ + rebuild( + source: EventStore, + options?: { readonly runId?: string; readonly batchSize?: number }, + ): Promise +} + +export interface MemoryEventLog extends EventLogProjector { + events(runId: string): readonly EventLogEntry[] + run(runId: string): RunLogRow | null +} + +interface RunRecord { + runId: string + status: RunStatus + kind: string | null + definitionName: string | null + definitionVersion: string | null + rootThreadId: string | null + parentRunId: string | null + launchedByEffectId: string | null + correlationId: string | null + attributes: { [key: string]: string } + tenant: string | null + project: string | null + org: string | null + startedAt: number | null + updatedAt: number + completedAt: number | null +} + +const terminal = new Set(['completed', 'failed', 'cancelled']) + +export function eventLogOps(event: EventEnvelope): readonly EventLogOp[] { + if (event.ephemeral) { + return [] + } + + const entry: EventLogEntry = { + runId: event.runId, + seq: event.seq, + eventId: event.id, + ts: event.ts, + type: event.type, + payload: event.payload, + threadId: event.threadId, + parentThreadId: event.parentThreadId ?? null, + causationId: event.causationId ?? null, + correlationId: event.correlationId ?? null, + effectId: event.effectId ?? null, + origin: event.origin, + idempotencyKey: event.idempotencyKey ?? null, + } + + const payload = isJsonObject(event.payload) ? event.payload : {} + + switch (event.type) { + case 'runtime.run.started': + return [{ type: 'insert', entry }, startOp(event, payload)] + case 'runtime.run.completed': + return [ + { type: 'insert', entry }, + { + type: 'finish', + runId: event.runId, + status: completedRunStatus(event.payload), + at: event.ts, + }, + ] + case 'runtime.run.cancelled': + return [ + { type: 'insert', entry }, + { type: 'finish', runId: event.runId, status: 'cancelled', at: event.ts }, + ] + default: + return [ + { type: 'insert', entry }, + { type: 'touch', runId: event.runId, at: event.ts }, + ] + } +} + +export function memoryEventLog(): MemoryEventLog { + const events = new Map() + const runs = new Map() + + const projector = createEventLogProjector('memory-event-log', { + init: () => Promise.resolve(), + apply: (batch) => { + for (const event of batch) { + for (const op of eventLogOps(event)) { + applyMemoryOp(events, runs, op) + } + } + + return Promise.resolve() + }, + }) + + return { + ...projector, + events: (runId) => + [...events.values()] + .filter((entry) => entry.runId === runId) + .toSorted((left, right) => left.seq - right.seq), + run: (runId) => { + const row = runs.get(runId) + return row ? { ...row, attributes: { ...row.attributes } } : null + }, + } +} + +export function postgresEventLog(options: { url: string } | { sql: Sql }): EventLogProjector { + const created = 'url' in options + const connected = created ? createPostgresClient(options.url) : options.sql + + return createEventLogProjector('postgres-event-log', postgresStore(connected, created)) +} + +function createEventLogProjector( + name: string, + store: { + init: () => Promise + apply: (events: readonly EventEnvelope[]) => Promise + dispose?: () => Promise + }, +): EventLogProjector { + return { + name, + version: '1', + init: store.init, + project: (events) => store.apply(events), + dispose: store.dispose, + rebuild: (source, options) => Effect.runPromise(rebuildEventLog(store, source, options)), + } +} + +function rebuildEventLog( + store: { init: () => Promise; apply: (events: readonly EventEnvelope[]) => Promise }, + source: EventStore, + options?: { readonly runId?: string; readonly batchSize?: number }, +): Effect.Effect { + const batchSize = options?.batchSize ?? 100 + + return Effect.gen(function* () { + yield* Effect.promise(() => store.init()) + const runIds = options?.runId ? [options.runId] : yield* source.listRuns + let written = 0 + + for (const runId of runIds) { + let fromSeq = 1 + + while (fromSeq > 0) { + const events = yield* source.read(runId, { fromSeq, limit: batchSize }) + const last = events.at(-1) + + if (!last) { + break + } + + yield* Effect.promise(() => store.apply(events)) + written += events.filter((event) => !event.ephemeral).length + fromSeq = events.length < batchSize ? 0 : last.seq + 1 + } + } + + return written + }) +} + +function applyMemoryOp( + events: Map, + runs: Map, + op: EventLogOp, +): void { + switch (op.type) { + case 'insert': + events.set(`${op.entry.runId}:${op.entry.seq}`, op.entry) + return + case 'start': + runs.set(op.runId, mergeStart(runs.get(op.runId), op)) + return + case 'finish': + runs.set(op.runId, mergeFinish(runs.get(op.runId), op)) + return + + case 'touch': { + const existing = runs.get(op.runId) + + if (!existing || existing.updatedAt >= op.at) { + return + } + + runs.set(op.runId, { ...existing, updatedAt: op.at }) + return + } + + default: { + const unexpected: never = op + return unexpected + } + } +} + +function mergeStart( + existing: RunRecord | undefined, + op: Extract, +): RunRecord { + const status = existing && terminal.has(existing.status) ? existing.status : 'running' + + return { + runId: op.runId, + status, + kind: existing?.kind ?? op.kind, + definitionName: existing?.definitionName ?? op.definitionName, + definitionVersion: existing?.definitionVersion ?? op.definitionVersion, + rootThreadId: existing?.rootThreadId ?? op.rootThreadId, + parentRunId: existing?.parentRunId ?? op.parentRunId, + launchedByEffectId: existing?.launchedByEffectId ?? op.launchedByEffectId, + correlationId: existing?.correlationId ?? op.correlationId, + attributes: + existing && Object.keys(existing.attributes).length > 0 + ? existing.attributes + : { ...op.attributes }, + tenant: existing?.tenant ?? op.tenant, + project: existing?.project ?? op.project, + org: existing?.org ?? op.org, + startedAt: existing?.startedAt ?? op.at, + updatedAt: Math.max(existing?.updatedAt ?? 0, op.at), + completedAt: existing?.completedAt ?? null, + } +} + +function mergeFinish( + existing: RunRecord | undefined, + op: Extract, +): RunRecord { + return { + runId: op.runId, + status: op.status, + kind: existing?.kind ?? null, + definitionName: existing?.definitionName ?? null, + definitionVersion: existing?.definitionVersion ?? null, + rootThreadId: existing?.rootThreadId ?? null, + parentRunId: existing?.parentRunId ?? null, + launchedByEffectId: existing?.launchedByEffectId ?? null, + correlationId: existing?.correlationId ?? null, + attributes: existing?.attributes ?? {}, + tenant: existing?.tenant ?? null, + project: existing?.project ?? null, + org: existing?.org ?? null, + startedAt: existing?.startedAt ?? null, + updatedAt: Math.max(existing?.updatedAt ?? 0, op.at), + completedAt: existing?.completedAt ?? op.at, + } +} + +function utcDate(ms: number): Date { + return DateTime.toDateUtc(DateTime.makeUnsafe(ms)) +} + +function startOp(event: EventEnvelope, payload: { [key: string]: JsonValue }): EventLogOp { + const launchedBy = readRunLaunchSource( + payload.launchedBy === undefined ? undefined : payload.launchedBy, + ) + + const attributes = readAttributes(payload.attributes) + + return { + type: 'start', + runId: event.runId, + kind: readString(payload, 'kind'), + definitionName: readString(payload, 'definitionName'), + definitionVersion: readString(payload, 'definitionVersion'), + rootThreadId: readString(payload, 'rootThreadId'), + parentRunId: launchedBy?.runId ?? null, + launchedByEffectId: launchedBy?.effectId ?? null, + correlationId: event.correlationId ?? null, + attributes, + tenant: attributes.tenant ?? null, + project: attributes.project ?? null, + org: attributes.org ?? null, + at: event.ts, + } +} + +function readString(payload: { [key: string]: JsonValue }, key: string): string | null { + const value = payload[key] + return Predicate.isString(value) ? value : null +} + +function readAttributes(value: JsonValue | undefined): { [key: string]: string } { + if (value === undefined) { + return {} + } + + return Option.getOrElse(decodeAttributes(value), () => ({})) +} + +function postgresStore(sql: Sql, ownsClient: boolean) { + return { + init: () => + runProjectorPromise('postgres-event-log', () => + sql.unsafe(POSTGRES_EVENT_LOG_SCHEMA_SQL), + ).then(() => undefined), + apply: (events: readonly EventEnvelope[]) => { + const ops = events.flatMap(eventLogOps) + + if (ops.length === 0) { + return Promise.resolve() + } + + return runProjectorPromise('postgres-event-log', () => + sql.begin((tx) => + Effect.runPromise( + Effect.forEach(ops, (op) => Effect.promise(() => Promise.resolve(applySqlOp(tx, op))), { + concurrency: 1, + discard: true, + }), + ), + ), + ).then(() => undefined) + }, + dispose: () => + ownsClient + ? runProjectorPromise('postgres-event-log', () => sql.end({ timeout: 5 })).then( + () => undefined, + ) + : Promise.resolve(), + } +} + +function applySqlOp(sql: TransactionSql, op: EventLogOp): Promise { + switch (op.type) { + case 'insert': + return sql` + INSERT INTO looms_run_events ( + run_id, seq, event_id, ts, type, payload, thread_id, parent_thread_id, + causation_id, correlation_id, effect_id, origin, idempotency_key + ) + VALUES ( + ${op.entry.runId}, ${op.entry.seq}, ${op.entry.eventId}, ${utcDate(op.entry.ts)}, + ${op.entry.type}, ${sql.json(op.entry.payload)}, ${op.entry.threadId}, + ${op.entry.parentThreadId}, ${op.entry.causationId}, ${op.entry.correlationId}, + ${op.entry.effectId}, ${sql.json(op.entry.origin)}, ${op.entry.idempotencyKey} + ) + ON CONFLICT (run_id, seq) DO NOTHING + `.then(() => undefined) + case 'start': + return sql` + INSERT INTO looms_run_index ( + run_id, status, kind, definition_name, definition_version, root_thread_id, + parent_run_id, launched_by_effect_id, correlation_id, attributes, + tenant, project, org, started_at, updated_at, completed_at + ) + VALUES ( + ${op.runId}, 'running', ${op.kind}, ${op.definitionName}, ${op.definitionVersion}, + ${op.rootThreadId}, ${op.parentRunId}, ${op.launchedByEffectId}, ${op.correlationId}, + ${sql.json(op.attributes)}, ${op.tenant}, ${op.project}, ${op.org}, + ${utcDate(op.at)}, ${utcDate(op.at)}, NULL + ) + ON CONFLICT (run_id) DO UPDATE SET + status = CASE + WHEN looms_run_index.status IN ('completed', 'failed', 'cancelled') THEN looms_run_index.status + ELSE EXCLUDED.status + END, + kind = COALESCE(looms_run_index.kind, EXCLUDED.kind), + definition_name = COALESCE(looms_run_index.definition_name, EXCLUDED.definition_name), + definition_version = COALESCE(looms_run_index.definition_version, EXCLUDED.definition_version), + root_thread_id = COALESCE(looms_run_index.root_thread_id, EXCLUDED.root_thread_id), + parent_run_id = COALESCE(looms_run_index.parent_run_id, EXCLUDED.parent_run_id), + launched_by_effect_id = COALESCE(looms_run_index.launched_by_effect_id, EXCLUDED.launched_by_effect_id), + correlation_id = COALESCE(looms_run_index.correlation_id, EXCLUDED.correlation_id), + attributes = CASE + WHEN looms_run_index.attributes = '{}'::jsonb THEN EXCLUDED.attributes + ELSE looms_run_index.attributes + END, + tenant = COALESCE(looms_run_index.tenant, EXCLUDED.tenant), + project = COALESCE(looms_run_index.project, EXCLUDED.project), + org = COALESCE(looms_run_index.org, EXCLUDED.org), + started_at = COALESCE(looms_run_index.started_at, EXCLUDED.started_at), + updated_at = GREATEST(looms_run_index.updated_at, EXCLUDED.updated_at) + `.then(() => undefined) + case 'finish': + return sql` + INSERT INTO looms_run_index (run_id, status, attributes, updated_at, completed_at) + VALUES (${op.runId}, ${op.status}, '{}'::jsonb, ${utcDate(op.at)}, ${utcDate(op.at)}) + ON CONFLICT (run_id) DO UPDATE SET + status = EXCLUDED.status, + updated_at = GREATEST(looms_run_index.updated_at, EXCLUDED.updated_at), + completed_at = COALESCE(looms_run_index.completed_at, EXCLUDED.completed_at) + `.then(() => undefined) + case 'touch': + return sql` + UPDATE looms_run_index + SET updated_at = GREATEST(updated_at, ${utcDate(op.at)}) + WHERE run_id = ${op.runId} + `.then(() => undefined) + + default: { + const unexpected: never = op + throw unexpected + } + } +} diff --git a/packages/projectors/src/postgres.ts b/packages/projectors/src/postgres.ts index 431acec..9f7e73b 100644 --- a/packages/projectors/src/postgres.ts +++ b/packages/projectors/src/postgres.ts @@ -12,6 +12,15 @@ import { runProjectorPromise } from './projector' import { POSTGRES_SCHEMA_SQL } from './schema' export { POSTGRES_SCHEMA_SQL } from './schema' +export { + memoryEventLog, + postgresEventLog, + POSTGRES_EVENT_LOG_SCHEMA_SQL, + type EventLogEntry, + type EventLogProjector, + type MemoryEventLog, + type RunLogRow, +} from './event-log' export type PostgresProjectorOptions = { url: string } | { sql: Sql } diff --git a/packages/projectors/src/with-projectors.test.ts b/packages/projectors/src/with-projectors.test.ts index 3efc523..86c799e 100644 --- a/packages/projectors/src/with-projectors.test.ts +++ b/packages/projectors/src/with-projectors.test.ts @@ -140,6 +140,42 @@ describe('withProjectors', () => { await pending }) + test('a scheduled delivery returns from append before the projector runs', async () => { + const base = await Effect.runPromise(makeMemoryEventStore) + let started = false + const scheduled: Array> = [] + + const store = withProjectors( + base, + [ + { + name: 'slow', + version: '1', + project: async () => { + started = true + }, + }, + ], + { + scheduleDelivery: (_runId, delivery) => { + scheduled.push(delivery) + }, + }, + ) + + await Effect.runPromise(store.append('run_parallel', [testEvent()])) + expect(started).toBe(false) + expect(scheduled).toHaveLength(1) + + const delivery = scheduled[0] + + if (delivery) { + await Effect.runPromise(delivery) + } + + expect(started).toBe(true) + }) + test('projects independent explicit handlers concurrently', async () => { const release = Promise.withResolvers() const started: string[] = [] diff --git a/packages/projectors/src/with-projectors.ts b/packages/projectors/src/with-projectors.ts index 769531c..d2b33b5 100644 --- a/packages/projectors/src/with-projectors.ts +++ b/packages/projectors/src/with-projectors.ts @@ -15,6 +15,12 @@ export interface WithProjectorsOptions { onError?: ProjectorErrorHandler /** Durable cursor runner. When present, appends trigger catch-up from its persisted cursor. */ delivery?: ProjectorDelivery + /** + * Runs projection after the append commits, without making the append wait. + * The callback has to keep the effect alive, for example with a Durable Object's `waitUntil`. + * Without it, projection still finishes before `append` returns. + */ + scheduleDelivery?: (runId: string, delivery: Effect.Effect) => void } function defaultOnError(error: Error, projector: Projector): void { @@ -40,6 +46,7 @@ export function withProjectors( ): EventStore { const onError = options?.onError ?? defaultOnError const delivery = options?.delivery + const scheduleDelivery = options?.scheduleDelivery const project = (runId: string, sequences: readonly number[]): Effect.Effect => { if (delivery) { @@ -73,13 +80,23 @@ export function withProjectors( const wrapped: EventStore = { append: (runId, events, appendOptions) => - store - .append(runId, events, appendOptions) - .pipe( - Effect.tap((result) => - projectors.length === 0 ? Effect.void : project(runId, result.sequences), - ), - ), + store.append(runId, events, appendOptions).pipe( + Effect.tap((result) => { + if (projectors.length === 0) { + return Effect.void + } + + const work = project(runId, result.sequences) + + if (!scheduleDelivery) { + return work + } + + return Effect.sync(() => { + scheduleDelivery(runId, work) + }) + }), + ), read: (runId, readOptions) => store.read(runId, readOptions), tail: (runId) => store.tail(runId), bounds: store.bounds ? (runId) => store.bounds!(runId) : undefined, From e9e0c7cd7edd86b27087cfb1258883d5a0f86617 Mon Sep 17 00:00:00 2001 From: Dylan Slack Date: Fri, 23 Oct 2026 14:08:43 +0200 Subject: [PATCH 2/2] perf(projectors): write each event-log batch in a fixed number of statements postgresEventLog issued one statement per event plus one per run-index change. A batch is now one multi-row event insert per 1,000 rows, one upsert for runs that started or finished, and one unnest update for runs that only gained events, all in one transaction. Index changes are folded per run first, with the same rules as the memory sink. New tests run the Postgres sink on PGlite through its socket server. They check that it matches the memory sink row for row, that it writes a fixed number of statements, and that a touch-only batch ignores unknown runs. Co-authored-by: Cursor --- bun.lock | 20 ++ docs/production.md | 5 +- packages/projectors/package.json | 2 + .../projectors/src/event-log-postgres.test.ts | 198 +++++++++++++++ packages/projectors/src/event-log.ts | 229 ++++++++++++------ 5 files changed, 376 insertions(+), 78 deletions(-) create mode 100644 packages/projectors/src/event-log-postgres.test.ts diff --git a/bun.lock b/bun.lock index d5649cb..9fb90c7 100644 --- a/bun.lock +++ b/bun.lock @@ -397,6 +397,8 @@ "effect": "4.0.0-rc.112", }, "devDependencies": { + "@electric-sql/pglite": "^0.5.11", + "@electric-sql/pglite-socket": "^0.2.14", "@types/bun": "^1.3.9", "postgres": "^3.4.0", "typescript": "7.0.2", @@ -687,6 +689,24 @@ "@effect/tsgo-win32-x64": ["@effect/tsgo-win32-x64@0.40.0", "", { "os": "win32", "cpu": "x64" }, "sha512-tA7NgEkZB0mFEt/Y/gBRoRDz8ui8hvUi1jE23lSxo3iaSvHlQaTi65Xkw2niryAMSvnYV9TWrY0C8F+jG6P2kw=="], + "@electric-sql/pglite": ["@electric-sql/pglite@0.5.11", "", {}, "sha512-SWuqsTjA4DDWqgZ05PC1RJaq8e00p5KACxbqaikDFQ8dIGA0SLRzuPniky7IW+q/c5RDNMdEXwv7mFBWIx9Hxw=="], + + "@electric-sql/pglite-age": ["@electric-sql/pglite-age@0.0.12", "", { "peerDependencies": { "@electric-sql/pglite": "0.5.11" } }, "sha512-h3w0ZbR/thNxbRiHrakK5XOJE5bbyIGT0mknREwiE2ELCLAEVv10x3jhnIkXTU5YJ5WzTZj2NbrFcAl/iHj3XQ=="], + + "@electric-sql/pglite-pg_hashids": ["@electric-sql/pglite-pg_hashids@0.0.12", "", { "peerDependencies": { "@electric-sql/pglite": "0.5.11" } }, "sha512-2vXH899hpmiKMhI1NvR9wxfyyW2SyvAMlBOwm8t02BRMpHIkXtl++fsBUTibL0TzTMAVLRFts53czelLPN0rDw=="], + + "@electric-sql/pglite-pg_ivm": ["@electric-sql/pglite-pg_ivm@0.0.12", "", { "peerDependencies": { "@electric-sql/pglite": "0.5.11" } }, "sha512-I+WQ0Hb7u7ZxMmznazZ15C0T6c4OKRuKVc5LaKkPNU+Cv7/aZmqRfPOHVu4MY65HC65dU3YmxT0A+tdWHcUF8w=="], + + "@electric-sql/pglite-pg_textsearch": ["@electric-sql/pglite-pg_textsearch@0.0.13", "", { "peerDependencies": { "@electric-sql/pglite": "0.5.11" } }, "sha512-TQX6Opop6fRkfgbZcuUDZ4UYm7x07uQ5JznVJWG9QL3H3hx3qQ1CbqG6eSHzVEjGxxFqoH+2UJs0l/XR/Y3esA=="], + + "@electric-sql/pglite-pg_uuidv7": ["@electric-sql/pglite-pg_uuidv7@0.0.12", "", { "peerDependencies": { "@electric-sql/pglite": "0.5.11" } }, "sha512-Q+JvI5fSE0bws+qNa9hzoZEqhGFql/LjTZ5Fq1ojKx2hd99RNkLRcpDUNjqeyluCpZS5Aq/PsLjb5JBUevkXGw=="], + + "@electric-sql/pglite-pgtap": ["@electric-sql/pglite-pgtap@0.0.12", "", { "peerDependencies": { "@electric-sql/pglite": "0.5.11" } }, "sha512-qp8jE4sx+fNoCkcyd/cL8od+Nir4hFVphSh4/vH7P3cFCH6/sayWjQas9OyzPL0JHe7XQRFquyOvgwtUzbJgXQ=="], + + "@electric-sql/pglite-pgvector": ["@electric-sql/pglite-pgvector@0.0.12", "", { "peerDependencies": { "@electric-sql/pglite": "0.5.11" } }, "sha512-n3TJqjWI9FXwFrGhuTVMghef8SL3E6nb4lxZnxYFW3UTg/7/hMUWfz7v2y0+idWzEsQ/raRpMNCsw9zvSdC/SA=="], + + "@electric-sql/pglite-socket": ["@electric-sql/pglite-socket@0.2.14", "", { "peerDependencies": { "@electric-sql/pglite": "0.5.11", "@electric-sql/pglite-age": "0.0.12", "@electric-sql/pglite-pg_hashids": "0.0.12", "@electric-sql/pglite-pg_ivm": "0.0.12", "@electric-sql/pglite-pg_textsearch": "0.0.13", "@electric-sql/pglite-pg_uuidv7": "0.0.12", "@electric-sql/pglite-pgtap": "0.0.12", "@electric-sql/pglite-pgvector": "0.0.12" }, "bin": { "pglite-server": "dist/scripts/server.js" } }, "sha512-n7JO9FzxQHdF5oBKzovf/LuRUrEUpOgGP/tmfxmOB5lqWPySTKkCb6FBkzj7kztWGJu9eB0FoilqxvOJ+9+0fQ=="], + "@emnapi/core": ["@emnapi/core@1.11.2", "", { "dependencies": { "@emnapi/wasi-threads": "1.2.2", "tslib": "^2.4.0" } }, "sha512-TC8MkTuZUtcTSiFeuC0ksCh9QIJ5+F21MvZ4Wn4ORfYaFJ/0dsiudv5tVkejgwZlwQ39jL9WWDe2lz8x0WglOA=="], "@emnapi/runtime": ["@emnapi/runtime@1.11.2", "", { "dependencies": { "tslib": "^2.4.0" } }, "sha512-kyOl3X0DuTiT1h2ft8r2fYO8JYtU9a9Xis/zBSiGArNaagCOWx90N1k2wxp18czFDH+OgcWGb5ZP/XMt3dcyPA=="], diff --git a/docs/production.md b/docs/production.md index 362789e..11d0730 100644 --- a/docs/production.md +++ b/docs/production.md @@ -30,7 +30,10 @@ CREATE POLICY tenant_isolation ON looms_run_index USING (tenant = current_setting('looms.tenant', true)); ``` -Pass Hyperdrive's connection string as `postgresEventLog({ url })`. Writes are +Pass Hyperdrive's connection string as `postgresEventLog({ url })`. Each +delivered batch is one transaction of about three statements, however many +events and runs it holds: one multi-row event insert per 1,000 events, one +run-index upsert, and one update for runs that only gained events. Writes are idempotent. `rebuild(source)` replays a run from seq 1; run that while live delivery is paused, then let the cursor catch up. Ephemeral events are left out of the log. diff --git a/packages/projectors/package.json b/packages/projectors/package.json index fcdefc1..7ff418a 100644 --- a/packages/projectors/package.json +++ b/packages/projectors/package.json @@ -43,6 +43,8 @@ "effect": "4.0.0-rc.112" }, "devDependencies": { + "@electric-sql/pglite": "^0.5.11", + "@electric-sql/pglite-socket": "^0.2.14", "@types/bun": "^1.3.9", "postgres": "^3.4.0", "typescript": "7.0.2" diff --git a/packages/projectors/src/event-log-postgres.test.ts b/packages/projectors/src/event-log-postgres.test.ts new file mode 100644 index 0000000..636e152 --- /dev/null +++ b/packages/projectors/src/event-log-postgres.test.ts @@ -0,0 +1,198 @@ +import { afterAll, beforeAll, describe, expect, test } from 'bun:test' + +import { PGlite } from '@electric-sql/pglite' +import { PGLiteSocketServer } from '@electric-sql/pglite-socket' +import createPostgresClient, { type Sql } from 'postgres' + +import { createEvent, type EventEnvelope, type EventInput, type RunStatus } from '@looms/core' + +import { memoryEventLog, postgresEventLog, type RunLogRow } from './event-log' + +const PORT = 54_329 +const RUN_STATUSES: readonly RunStatus[] = ['running', 'completed', 'failed', 'cancelled'] + +let db: PGlite +let server: PGLiteSocketServer +let sql: Sql +const statements: string[] = [] + +beforeAll(async () => { + db = await PGlite.create() + server = new PGLiteSocketServer({ db, port: PORT, host: '127.0.0.1' }) + await server.start() + + sql = createPostgresClient({ + host: '127.0.0.1', + port: PORT, + max: 1, + onnotice: () => undefined, + debug: (_connection, query) => { + statements.push(query) + }, + }) +}) + +afterAll(async () => { + await sql.end({ timeout: 1 }) + await server.stop() + await db.close() +}) + +function event(runId: string, seq: number, input: EventInput): EventEnvelope { + return { ...createEvent(runId, input), seq, ts: 1_700_000_000_000 + seq * 1_000 } +} + +function started(runId: string): EventEnvelope { + return event(runId, 1, { + type: 'runtime.run.started', + payload: { + rootThreadId: 'thr_1', + kind: 'workflow', + definitionName: 'checkout', + definitionVersion: 'v1', + input: { text: 'it\'s "quoted"' }, + attributes: { tenant: 'acme', project: 'billing' }, + launchedBy: { + runId: 'run_parent', + threadId: 'thr_parent', + effectId: 'eff_1', + causationId: 'evt_cause', + causationSeq: 4, + }, + }, + correlationId: 'run_root', + threadId: null, + origin: { type: 'system' }, + }) +} + +function step(runId: string, seq: number): EventEnvelope { + return event(runId, seq, { + type: 'workflow.node.completed', + payload: 'plain string payload', + threadId: 'thr_1', + origin: { type: 'system' }, + }) +} + +function failed(runId: string, seq: number): EventEnvelope { + return event(runId, seq, { + type: 'runtime.run.completed', + payload: { output: null, error: 'nope' }, + threadId: null, + origin: { type: 'system' }, + }) +} + +interface IndexRow { + readonly run_id: string + readonly status: string + readonly kind: string | null + readonly definition_name: string | null + readonly definition_version: string | null + readonly root_thread_id: string | null + readonly parent_run_id: string | null + readonly launched_by_effect_id: string | null + readonly correlation_id: string | null + readonly attributes: { readonly [key: string]: string } + readonly tenant: string | null + readonly project: string | null + readonly org: string | null + readonly started_at: Date | null + readonly updated_at: Date + readonly completed_at: Date | null +} + +function fromIndexRow(row: IndexRow): RunLogRow { + return { + runId: row.run_id, + status: RUN_STATUSES.find((status) => status === row.status) ?? 'running', + kind: row.kind, + definitionName: row.definition_name, + definitionVersion: row.definition_version, + rootThreadId: row.root_thread_id, + parentRunId: row.parent_run_id, + launchedByEffectId: row.launched_by_effect_id, + correlationId: row.correlation_id, + attributes: row.attributes, + tenant: row.tenant, + project: row.project, + org: row.org, + startedAt: row.started_at?.getTime() ?? null, + updatedAt: row.updated_at.getTime(), + completedAt: row.completed_at?.getTime() ?? null, + } +} + +function indexRow(runId: string): Promise { + return sql`SELECT * FROM looms_run_index WHERE run_id = ${runId}`.then(([row]) => + row ? fromIndexRow(row) : null, + ) +} + +describe('postgres event log', () => { + test('matches the memory sink for starts, steps, a replay, and a finish', async () => { + const postgres = postgresEventLog({ sql }) + const memory = memoryEventLog() + await postgres.init?.() + + const batches = [ + [started('run_a'), step('run_a', 2), started('run_b')], + [step('run_a', 3), failed('run_a', 4), step('run_b', 2)], + [started('run_a'), step('run_a', 3)], + ] + + for (const batch of batches) { + await postgres.project(batch) + await memory.project(batch) + } + + for (const runId of ['run_a', 'run_b']) { + expect(await indexRow(runId)).toEqual(memory.run(runId)) + } + + const events = await sql<{ seq: string; payload: unknown }[]>` + SELECT seq, payload FROM looms_run_events WHERE run_id = 'run_a' ORDER BY seq + ` + + expect(events.map((row) => Number(row.seq))).toEqual([1, 2, 3, 4]) + expect(events[1]?.payload).toBe('plain string payload') + expect(events[0]?.payload).toMatchObject({ input: { text: 'it\'s "quoted"' } }) + }) + + test('writes a batch with a fixed number of statements', async () => { + const postgres = postgresEventLog({ sql }) + await postgres.init?.() + + const batch = Array.from({ length: 50 }, (_, index) => [ + started(`run_many_${index}`), + step(`run_many_${index}`, 2), + step(`run_many_${index}`, 3), + ]).flat() + + statements.length = 0 + await postgres.project(batch) + + const writes = statements.filter((query) => /^\s*(INSERT|UPDATE)/i.test(query)) + + const [count] = await sql<{ count: string }[]>` + SELECT count(*) FROM looms_run_index WHERE run_id LIKE 'run_many_%' + ` + + expect(writes).toHaveLength(2) + expect(Number(count?.count)).toBe(50) + }) + + test('a touch-only batch updates existing runs and ignores unknown ones', async () => { + const postgres = postgresEventLog({ sql }) + await postgres.init?.() + await postgres.project([started('run_touch')]) + + statements.length = 0 + await postgres.project([step('run_touch', 9), step('run_unknown', 2)]) + + expect(statements.filter((query) => /^\s*UPDATE/i.test(query))).toHaveLength(1) + expect((await indexRow('run_touch'))?.updatedAt).toBe(1_700_000_009_000) + expect(await indexRow('run_unknown')).toBeNull() + }) +}) diff --git a/packages/projectors/src/event-log.ts b/packages/projectors/src/event-log.ts index 308f603..f395997 100644 --- a/packages/projectors/src/event-log.ts +++ b/packages/projectors/src/event-log.ts @@ -437,14 +437,7 @@ function postgresStore(sql: Sql, ownsClient: boolean) { } return runProjectorPromise('postgres-event-log', () => - sql.begin((tx) => - Effect.runPromise( - Effect.forEach(ops, (op) => Effect.promise(() => Promise.resolve(applySqlOp(tx, op))), { - concurrency: 1, - discard: true, - }), - ), - ), + sql.begin((tx) => applySqlOps(tx, ops)), ).then(() => undefined) }, dispose: () => @@ -456,76 +449,158 @@ function postgresStore(sql: Sql, ownsClient: boolean) { } } -function applySqlOp(sql: TransactionSql, op: EventLogOp): Promise { - switch (op.type) { - case 'insert': - return sql` - INSERT INTO looms_run_events ( - run_id, seq, event_id, ts, type, payload, thread_id, parent_thread_id, - causation_id, correlation_id, effect_id, origin, idempotency_key - ) - VALUES ( - ${op.entry.runId}, ${op.entry.seq}, ${op.entry.eventId}, ${utcDate(op.entry.ts)}, - ${op.entry.type}, ${sql.json(op.entry.payload)}, ${op.entry.threadId}, - ${op.entry.parentThreadId}, ${op.entry.causationId}, ${op.entry.correlationId}, - ${op.entry.effectId}, ${sql.json(op.entry.origin)}, ${op.entry.idempotencyKey} - ) - ON CONFLICT (run_id, seq) DO NOTHING - `.then(() => undefined) - case 'start': - return sql` - INSERT INTO looms_run_index ( - run_id, status, kind, definition_name, definition_version, root_thread_id, - parent_run_id, launched_by_effect_id, correlation_id, attributes, - tenant, project, org, started_at, updated_at, completed_at - ) - VALUES ( - ${op.runId}, 'running', ${op.kind}, ${op.definitionName}, ${op.definitionVersion}, - ${op.rootThreadId}, ${op.parentRunId}, ${op.launchedByEffectId}, ${op.correlationId}, - ${sql.json(op.attributes)}, ${op.tenant}, ${op.project}, ${op.org}, - ${utcDate(op.at)}, ${utcDate(op.at)}, NULL - ) - ON CONFLICT (run_id) DO UPDATE SET - status = CASE - WHEN looms_run_index.status IN ('completed', 'failed', 'cancelled') THEN looms_run_index.status - ELSE EXCLUDED.status - END, - kind = COALESCE(looms_run_index.kind, EXCLUDED.kind), - definition_name = COALESCE(looms_run_index.definition_name, EXCLUDED.definition_name), - definition_version = COALESCE(looms_run_index.definition_version, EXCLUDED.definition_version), - root_thread_id = COALESCE(looms_run_index.root_thread_id, EXCLUDED.root_thread_id), - parent_run_id = COALESCE(looms_run_index.parent_run_id, EXCLUDED.parent_run_id), - launched_by_effect_id = COALESCE(looms_run_index.launched_by_effect_id, EXCLUDED.launched_by_effect_id), - correlation_id = COALESCE(looms_run_index.correlation_id, EXCLUDED.correlation_id), - attributes = CASE - WHEN looms_run_index.attributes = '{}'::jsonb THEN EXCLUDED.attributes - ELSE looms_run_index.attributes - END, - tenant = COALESCE(looms_run_index.tenant, EXCLUDED.tenant), - project = COALESCE(looms_run_index.project, EXCLUDED.project), - org = COALESCE(looms_run_index.org, EXCLUDED.org), - started_at = COALESCE(looms_run_index.started_at, EXCLUDED.started_at), - updated_at = GREATEST(looms_run_index.updated_at, EXCLUDED.updated_at) - `.then(() => undefined) - case 'finish': - return sql` - INSERT INTO looms_run_index (run_id, status, attributes, updated_at, completed_at) - VALUES (${op.runId}, ${op.status}, '{}'::jsonb, ${utcDate(op.at)}, ${utcDate(op.at)}) - ON CONFLICT (run_id) DO UPDATE SET - status = EXCLUDED.status, - updated_at = GREATEST(looms_run_index.updated_at, EXCLUDED.updated_at), - completed_at = COALESCE(looms_run_index.completed_at, EXCLUDED.completed_at) - `.then(() => undefined) - case 'touch': - return sql` - UPDATE looms_run_index - SET updated_at = GREATEST(updated_at, ${utcDate(op.at)}) - WHERE run_id = ${op.runId} - `.then(() => undefined) +/** Keeps each statement under Postgres's 65,535 bind parameters (13 per event row). */ +const INSERT_CHUNK_ROWS = 1_000 - default: { - const unexpected: never = op - throw unexpected +type StartOp = Extract +type FinishOp = Extract + +/** Every index change a batch makes to one run, folded the way `applyMemoryOp` folds them. */ +interface RunIndexPatch { + readonly runId: string + start: StartOp | null + finish: FinishOp | null + completedAt: number | null + updatedAt: number +} + +function runIndexPatches(ops: readonly EventLogOp[]): RunIndexPatch[] { + const patches = new Map() + + for (const op of ops) { + if (op.type === 'insert') { + continue + } + + const patch = patches.get(op.runId) ?? { + runId: op.runId, + start: null, + finish: null, + completedAt: null, + updatedAt: op.at, } + + patch.updatedAt = Math.max(patch.updatedAt, op.at) + + if (op.type === 'start') { + patch.start ??= op + } + + if (op.type === 'finish') { + patch.finish = op + patch.completedAt ??= op.at + } + + patches.set(op.runId, patch) + } + + return [...patches.values()] +} + +function applySqlOps(sql: TransactionSql, ops: readonly EventLogOp[]): Promise { + const entries = ops.flatMap((op) => (op.type === 'insert' ? [op.entry] : [])) + const patches = runIndexPatches(ops) + const upserts = patches.filter((patch) => patch.start !== null || patch.finish !== null) + const touches = patches.filter((patch) => patch.start === null && patch.finish === null) + const statements: Promise[] = [] + + for (let index = 0; index < entries.length; index += INSERT_CHUNK_ROWS) { + statements.push(insertEvents(sql, entries.slice(index, index + INSERT_CHUNK_ROWS))) + } + + if (upserts.length > 0) { + statements.push(upsertRuns(sql, upserts)) } + + if (touches.length > 0) { + statements.push(touchRuns(sql, touches)) + } + + return Promise.all(statements).then(() => undefined) +} + +function insertEvents(sql: TransactionSql, entries: readonly EventLogEntry[]): Promise { + const rows = entries.map((entry) => ({ + run_id: entry.runId, + seq: entry.seq, + event_id: entry.eventId, + ts: utcDate(entry.ts), + type: entry.type, + payload: sql.json(entry.payload), + thread_id: entry.threadId, + parent_thread_id: entry.parentThreadId, + causation_id: entry.causationId, + correlation_id: entry.correlationId, + effect_id: entry.effectId, + origin: sql.json(entry.origin), + idempotency_key: entry.idempotencyKey, + })) + + return sql` + INSERT INTO looms_run_events ${sql(rows)} + ON CONFLICT (run_id, seq) DO NOTHING + `.then(() => undefined) +} + +// A finish in the batch sets the status; otherwise a finished run keeps its status and a +// started run becomes running. Start fields fill only columns that are still empty. +function upsertRuns(sql: TransactionSql, patches: readonly RunIndexPatch[]): Promise { + const rows = patches.map(({ runId, start, finish, completedAt, updatedAt }) => ({ + run_id: runId, + status: finish?.status ?? 'running', + kind: start?.kind ?? null, + definition_name: start?.definitionName ?? null, + definition_version: start?.definitionVersion ?? null, + root_thread_id: start?.rootThreadId ?? null, + parent_run_id: start?.parentRunId ?? null, + launched_by_effect_id: start?.launchedByEffectId ?? null, + correlation_id: start?.correlationId ?? null, + attributes: sql.json(start?.attributes ?? {}), + tenant: start?.tenant ?? null, + project: start?.project ?? null, + org: start?.org ?? null, + started_at: start ? utcDate(start.at) : null, + updated_at: utcDate(updatedAt), + completed_at: completedAt === null ? null : utcDate(completedAt), + })) + + return sql` + INSERT INTO looms_run_index ${sql(rows)} + ON CONFLICT (run_id) DO UPDATE SET + status = CASE + WHEN EXCLUDED.completed_at IS NOT NULL THEN EXCLUDED.status + WHEN looms_run_index.status IN ('completed', 'failed', 'cancelled') THEN looms_run_index.status + ELSE EXCLUDED.status + END, + kind = COALESCE(looms_run_index.kind, EXCLUDED.kind), + definition_name = COALESCE(looms_run_index.definition_name, EXCLUDED.definition_name), + definition_version = COALESCE(looms_run_index.definition_version, EXCLUDED.definition_version), + root_thread_id = COALESCE(looms_run_index.root_thread_id, EXCLUDED.root_thread_id), + parent_run_id = COALESCE(looms_run_index.parent_run_id, EXCLUDED.parent_run_id), + launched_by_effect_id = COALESCE(looms_run_index.launched_by_effect_id, EXCLUDED.launched_by_effect_id), + correlation_id = COALESCE(looms_run_index.correlation_id, EXCLUDED.correlation_id), + attributes = CASE + WHEN looms_run_index.attributes = '{}'::jsonb THEN EXCLUDED.attributes + ELSE looms_run_index.attributes + END, + tenant = COALESCE(looms_run_index.tenant, EXCLUDED.tenant), + project = COALESCE(looms_run_index.project, EXCLUDED.project), + org = COALESCE(looms_run_index.org, EXCLUDED.org), + started_at = COALESCE(looms_run_index.started_at, EXCLUDED.started_at), + updated_at = GREATEST(looms_run_index.updated_at, EXCLUDED.updated_at), + completed_at = COALESCE(looms_run_index.completed_at, EXCLUDED.completed_at) + `.then(() => undefined) +} + +function touchRuns(sql: TransactionSql, patches: readonly RunIndexPatch[]): Promise { + const runIds = patches.map((patch) => patch.runId) + const touchedAt = patches.map((patch) => DateTime.formatIso(DateTime.makeUnsafe(patch.updatedAt))) + + return sql` + UPDATE looms_run_index AS run + SET updated_at = GREATEST(run.updated_at, touched.at) + FROM unnest(${sql.array(runIds)}::text[], ${sql.array(touchedAt)}::timestamptz[]) + AS touched (run_id, at) + WHERE run.run_id = touched.run_id + `.then(() => undefined) }