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/bun.lock b/bun.lock index 3601304..26d0ceb 100644 --- a/bun.lock +++ b/bun.lock @@ -398,6 +398,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", @@ -688,6 +690,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 8cdffdd..9a6843c 100644 --- a/docs/production.md +++ b/docs/production.md @@ -13,6 +13,31 @@ 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 })`. 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. + Live subscribers on a Durable Object should open a WebSocket (`Upgrade: websocket` on `GET /runs/:id/events`). The object hibernates the socket, answers `ping` with `pong` without waking, and fans out events when the log is appended. SSE stays available, including on `looms.serve`, but its keepalive keeps a Durable Object awake. Writes stay on HTTP. A browser cannot send `Authorization` on the upgrade, so pass a read grant (see `grantAuth` below); `@looms/client` puts `grant` on the socket URL. An idle-socket billed-duration check on a real Cloudflare account has not been run. A workflow can launch children with `childExecution: 'run'`. Each child is its own run, addressed by a deterministic id, and reports `runtime.child.settled` back to the parent. Cancelling the parent cancels those children. `childClose: 'wait'` records `runtime.child.close.requested` and cancels the parent only after the children reach a terminal status. `childClose: 'terminate'` cancels them as the parent closes. Children stay threads in the parent run unless `childExecution` is set. diff --git a/packages/cloudflare/src/event-store.ts b/packages/cloudflare/src/event-store.ts index 5263b7b..34d8476 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 { snapshotStoreOf, withBlobOffload, type BlobStore, type EventStore } from '@looms/core' import { sqliteEventStore, sqlStorageExec } from '@looms/core/sqlite-store' @@ -23,7 +23,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, @@ -49,7 +51,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/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.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..f395997 --- /dev/null +++ b/packages/projectors/src/event-log.ts @@ -0,0 +1,606 @@ +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) => applySqlOps(tx, ops)), + ).then(() => undefined) + }, + dispose: () => + ownsClient + ? runProjectorPromise('postgres-event-log', () => sql.end({ timeout: 5 })).then( + () => undefined, + ) + : Promise.resolve(), + } +} + +/** Keeps each statement under Postgres's 65,535 bind parameters (13 per event row). */ +const INSERT_CHUNK_ROWS = 1_000 + +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) +} 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,