1/**
2 * Historical name for the shared application event store.
3 *
4 * Owns durable append/replay access for project events and V2 agent-thread
5 * events under one global sequence. It does not reduce events into read models
6 * or apply command validation rules.
7 *
8 * Uses Effect `Context.Service` for dependency injection and exposes typed
9 * persistence/decode errors for event append and replay operations.
10 *
11 * @module OrchestrationEventStore
12 */
13import {
14 ApplicationEventMetadata,
15 ApplicationProjectEvent,
16 type ApplicationStoredEvent,
17 CommandId,
18 EventId,
19 IsoDateTime,
20 isKnownOrchestrationV2EventType,
21 NonNegativeInt,
22 OrchestrationV2DomainEventJson,
23 OrchestrationV2StoredEvent,
24 ProjectId,
25 StoredProjectIcon,
26 ThreadId,
27 type OrchestrationV2DomainEvent,
28} from "@t3tools/contracts";
29import * as Context from "effect/Context";
30import * as SqlClient from "effect/sql/SqlClient";
31import * as SqlSchema from "effect/sql/SqlSchema";
32import * as Effect from "effect/Effect";
33import * as DateTime from "effect/DateTime";
34import * as Layer from "effect/Layer";
35import * as PubSub from "effect/PubSub";
36import * as Option from "effect/Option";
37import * as Schema from "effect/Schema";
38import * as Stream from "effect/Stream";
39
40import { replayAndBufferProjectedLiveEvents } from "../orchestration-v2/LiveStreamBudget.ts";
41
42import {
43 toPersistenceDecodeError,
44 toPersistenceSqlError,
45 type OrchestrationEventStoreError,
46} from "./Errors.ts";
47
48/** A project event before the store assigns its sequence. */
49export type UnsequencedProjectEvent = ApplicationProjectEvent extends infer Event
50 ? Event extends ApplicationProjectEvent
51 ? Omit<Event, "sequence">
52 : never
53 : never;
54
55/** OrchestrationEventStore - Service tag for the shared application event log. */
56export class OrchestrationEventStore extends Context.Service<
57 OrchestrationEventStore,
58 {
59 /** Append one project event to the shared application log. */
60 readonly appendProjectEvent: (
61 event: UnsequencedProjectEvent,
62 ) => Effect.Effect<ApplicationProjectEvent, OrchestrationEventStoreError>;
63
64 /** Append V2 agent events to the same globally ordered application log. */
65 readonly appendAgentEvents: (input: {
66 readonly commandId?: CommandId;
67 readonly events: ReadonlyArray<OrchestrationV2DomainEvent>;
68 }) => Effect.Effect<ReadonlyArray<OrchestrationV2StoredEvent>, OrchestrationEventStoreError>;
69
70 /**
71 * Read only V2 thread events from the application log.
72 *
73 * Reads in fixed-size sequence pages until the filtered range is exhausted;
74 * `limit` caps raw rows across pages, including skipped unknown types.
75 *
76 * A row whose `event_type` this build does not know fails the read unless
77 * `skipUnknownEventTypes` is set, which logs and drops it instead. Only
78 * client thread replay opts in; anything that builds server state stays strict.
79 */
80 readonly readAgentEvents: (input?: {
81 readonly afterSequence?: number;
82 readonly throughSequence?: number;
83 readonly threadId?: ThreadId;
84 readonly commandId?: CommandId;
85 readonly eventType?: OrchestrationV2DomainEvent["type"];
86 readonly limit?: number;
87 readonly skipUnknownEventTypes?: boolean;
88 }) => Stream.Stream<OrchestrationV2StoredEvent, OrchestrationEventStoreError>;
89
90 /** Measure one thread's bounded replay without loading or decoding its payloads. */
91 readonly getAgentReplayStats: (input: {
92 readonly threadId: ThreadId;
93 readonly afterSequence: number;
94 readonly throughSequence: number;
95 readonly maxEvents: number;
96 }) => Effect.Effect<
97 {
98 readonly eventCount: number;
99 /** UTF-8 bytes in persisted payload JSON before decoding or wire projection. */
100 readonly rawPayloadBytes: number;
101 readonly hasCreateEvent: boolean;
102 },
103 OrchestrationEventStoreError
104 >;
105
106 /**
107 * Measure the retained application-event range `(afterSequence, throughSequence]`
108 * without loading or decoding its payloads.
109 */
110 readonly getReplayStats: (input: {
111 readonly afterSequence: number;
112 readonly throughSequence: number;
113 }) => Effect.Effect<
114 {
115 readonly eventCount: number;
116 /** UTF-8 bytes in persisted payload JSON before decoding or wire projection. */
117 readonly rawPayloadBytes: number;
118 },
119 OrchestrationEventStoreError
120 >;
121
122 readonly latestAgentSequence: (
123 threadId?: ThreadId,
124 ) => Effect.Effect<number, OrchestrationEventStoreError>;
125
126 readonly latestApplicationSequence: Effect.Effect<number, OrchestrationEventStoreError>;
127 readonly getThreadSettlementTime: (
128 threadId: ThreadId,
129 ) => Effect.Effect<DateTime.Utc | null, OrchestrationEventStoreError>;
130
131 /**
132 * Read the finite retained application-event range `(afterSequence, throughSequence]`.
133 *
134 * A row whose `event_type` this build does not know fails the read unless
135 * `skipUnknownEventTypes` is set, which logs and drops it instead. Only
136 * client shell replay opts in; anything that builds server state stays strict.
137 */
138 readonly readApplicationEvents: (input: {
139 readonly afterSequence: number;
140 readonly throughSequence: number;
141 readonly skipUnknownEventTypes?: boolean;
142 }) => Stream.Stream<ApplicationStoredEvent, OrchestrationEventStoreError>;
143
144 /** Publish only after the surrounding event/projection transaction commits. */
145 readonly publishCommitted: (
146 events: ReadonlyArray<ApplicationStoredEvent>,
147 ) => Effect.Effect<void>;
148
149 /** Race-free replay-to-live stream for project and V2 thread events. */
150 readonly streamApplicationEvents: (input?: {
151 readonly afterSequence?: number;
152 }) => Stream.Stream<ApplicationStoredEvent, OrchestrationEventStoreError>;
153 /** Project transport events before bounding replay and the live tail. */
154 readonly streamProjectedApplicationEvents: <A extends { readonly sequence: number }>(input: {
155 readonly afterSequence?: number;
156 readonly project: (event: ApplicationStoredEvent) => A;
157 }) => Stream.Stream<A, OrchestrationEventStoreError>;
158 }
159>()("t3/persistence/OrchestrationEventStore") {}
160
161const encodeProjectIcon = Schema.encodeSync(StoredProjectIcon);
162const decodeProjectEvent = Schema.decodeUnknownEffect(ApplicationProjectEvent);
163const UnknownFromJsonString = Schema.fromJsonString(Schema.Unknown);
164const EventMetadataFromJsonString = Schema.fromJsonString(ApplicationEventMetadata);
165const ProjectEventType = Schema.Literals([
166 "project.created",
167 "project.meta-updated",
168 "project.deleted",
169]);
170const ActorKind = Schema.Literals(["client", "server", "provider"]);
171
172const AppendProjectEventRequestSchema = Schema.Struct({
173 eventId: EventId,
174 streamId: ProjectId,
175 type: ProjectEventType,
176 causationEventId: Schema.NullOr(EventId),
177 correlationId: Schema.NullOr(CommandId),
178 actorKind: ActorKind,
179 occurredAt: IsoDateTime,
180 commandId: Schema.NullOr(CommandId),
181 payloadJson: UnknownFromJsonString,
182 metadataJson: EventMetadataFromJsonString,
183});
184
185const ProjectEventPersistedRowSchema = Schema.Struct({
186 sequence: NonNegativeInt,
187 eventId: EventId,
188 type: ProjectEventType,
189 aggregateKind: Schema.Literal("project"),
190 aggregateId: ProjectId,
191 occurredAt: IsoDateTime,
192 commandId: Schema.NullOr(CommandId),
193 causationEventId: Schema.NullOr(EventId),
194 correlationId: Schema.NullOr(CommandId),
195 payload: UnknownFromJsonString,
196 metadata: EventMetadataFromJsonString,
197});
198
199const READ_PAGE_SIZE = 500;
200
201interface ApplicationEventRow {
202 readonly sequence: number;
203 readonly event_id: string;
204 readonly command_id: string | null;
205 readonly aggregate_kind: "project" | "thread";
206 readonly stream_id: string;
207 readonly event_type: string;
208 readonly occurred_at: string;
209 readonly payload_json: string;
210 readonly metadata_json: string;
211 readonly application_event_version: number;
212 readonly causation_event_id: string | null;
213 readonly correlation_id: string | null;
214}
215
216const decodeV2EventJson = Schema.decodeUnknownEffect(OrchestrationV2DomainEventJson);
217const encodeV2EventJson = Schema.encodeEffect(OrchestrationV2DomainEventJson);
218const decodeV2StoredEvent = Schema.decodeUnknownEffect(OrchestrationV2StoredEvent);
219const decodeJson = Schema.decodeUnknownEffect(Schema.fromJsonString(Schema.Unknown));
220const encodeJson = Schema.encodeEffect(Schema.fromJsonString(Schema.Unknown));
221
222function metadataForV2Event(event: OrchestrationV2DomainEvent): Record<string, unknown> {
223 return {
224 ...(event.runId === undefined ? {} : { runId: event.runId }),
225 ...(event.nodeId === undefined ? {} : { nodeId: event.nodeId }),
226 ...(event.driver === undefined ? {} : { driver: event.driver }),
227 ...(event.providerInstanceId === undefined
228 ? {}
229 : { providerInstanceId: event.providerInstanceId }),
230 ...(event.rawEventId === undefined ? {} : { rawEventId: event.rawEventId }),
231 };
232}
233
234const rowToV2StoredEvent = Effect.fn("OrchestrationEventStore.rowToV2StoredEvent")(function* (
235 row: ApplicationEventRow,
236) {
237 const payload = yield* decodeJson(row.payload_json);
238 const metadata = yield* decodeJson(row.metadata_json);
239 const values =
240 typeof metadata === "object" && metadata !== null ? (metadata as Record<string, unknown>) : {};
241 const event = yield* decodeV2EventJson({
242 id: row.event_id,
243 threadId: row.stream_id,
244 type: row.event_type,
245 occurredAt: row.occurred_at,
246 payload,
247 ...(values.runId === undefined ? {} : { runId: values.runId }),
248 ...(values.nodeId === undefined ? {} : { nodeId: values.nodeId }),
249 ...(values.driver === undefined ? {} : { driver: values.driver }),
250 ...(values.providerInstanceId === undefined
251 ? {}
252 : { providerInstanceId: values.providerInstanceId }),
253 ...(values.rawEventId === undefined ? {} : { rawEventId: values.rawEventId }),
254 });
255 return yield* decodeV2StoredEvent({
256 sequence: row.sequence,
257 commandId: row.command_id,
258 event,
259 });
260});
261
262const rowToProjectEvent = Effect.fn("OrchestrationEventStore.rowToProjectEvent")(function* (
263 row: ApplicationEventRow,
264) {
265 return yield* decodeProjectEvent({
266 sequence: row.sequence,
267 eventId: row.event_id,
268 type: row.event_type,
269 aggregateKind: row.aggregate_kind,
270 aggregateId: row.stream_id,
271 occurredAt: row.occurred_at,
272 commandId: row.command_id,
273 causationEventId: row.causation_event_id,
274 correlationId: row.correlation_id,
275 payload: yield* decodeJson(row.payload_json),
276 metadata: yield* decodeJson(row.metadata_json),
277 });
278});
279
280function rowToApplicationStoredEvent(
281 row: ApplicationEventRow,
282): Effect.Effect<ApplicationStoredEvent, Schema.SchemaError> {
283 return row.aggregate_kind === "project" ? rowToProjectEvent(row) : rowToV2StoredEvent(row);
284}
285
286const isKnownProjectEventType = Schema.is(ProjectEventType);
287
288/** Filter only unknown discriminants; known types still undergo strict payload decoding. */
289const keepReplayRow = (row: ApplicationEventRow, skipUnknownEventTypes?: boolean) =>
290 skipUnknownEventTypes !== true ||
291 (row.aggregate_kind === "project"
292 ? isKnownProjectEventType(row.event_type)
293 : isKnownOrchestrationV2EventType(row.event_type))
294 ? Effect.succeed(true)
295 : Effect.logWarning("Skipping an application event row with an unknown event type", {
296 aggregateKind: row.aggregate_kind,
297 sequence: row.sequence,
298 eventTypeLength: row.event_type.length,
299 }).pipe(Effect.as(false));
300
301function inferActorKind(event: UnsequencedProjectEvent): typeof ActorKind.Type {
302 if (event.commandId !== null && event.commandId.startsWith("provider:")) {
303 return "provider";
304 }
305 if (event.commandId !== null && event.commandId.startsWith("server:")) {
306 return "server";
307 }
308 if (
309 event.metadata.providerTurnId !== undefined ||
310 event.metadata.providerItemId !== undefined ||
311 event.metadata.adapterKey !== undefined
312 ) {
313 return "provider";
314 }
315 if (event.commandId === null) {
316 return "server";
317 }
318 return "client";
319}
320
321function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: string) {
322 return (cause: unknown): OrchestrationEventStoreError =>
323 Schema.isSchemaError(cause)
324 ? toPersistenceDecodeError(decodeOperation)(cause)
325 : toPersistenceSqlError(sqlOperation)(cause);
326}
327
328const makeEventStore = Effect.gen(function* () {
329 const sql = yield* SqlClient.SqlClient;
330 const committedEvents = yield* PubSub.unbounded<ApplicationStoredEvent>();
331 // Sequence of the latest row storing each subagent's full prompt. The INSERT re-checks it.
332 const promptHints = new Map<string, number>();
333 // Puts back a subagent prompt that appendAgentEvents stored once, behind `promptSequence`.
334 const restoredPayloadJson = sql`CASE
335 WHEN orchestration_events.event_type IN ('subagent.updated', 'turn-item.updated')
336 AND json_type(orchestration_events.payload_json, '$.promptSequence') = 'integer'
337 THEN json_set(
338 json_remove(orchestration_events.payload_json, '$.promptSequence'),
339 '$.prompt',
340 (SELECT json_extract(anchor.payload_json, '$.prompt')
341 FROM orchestration_events AS anchor
342 WHERE anchor.sequence = json_extract(orchestration_events.payload_json, '$.promptSequence'))
343 )
344 ELSE orchestration_events.payload_json
345 END`;
346
347 const appendProjectEventRow = SqlSchema.findOne({
348 Request: AppendProjectEventRequestSchema,
349 Result: ProjectEventPersistedRowSchema,
350 execute: (request) =>
351 sql`
352 INSERT INTO orchestration_events (
353 event_id,
354 aggregate_kind,
355 stream_id,
356 stream_version,
357 event_type,
358 occurred_at,
359 command_id,
360 causation_event_id,
361 correlation_id,
362 actor_kind,
363 payload_json,
364 metadata_json,
365 application_event_version
366 )
367 VALUES (
368 ${request.eventId},
369 'project',
370 ${request.streamId},
371 COALESCE(
372 (
373 SELECT stream_version + 1
374 FROM orchestration_events
375 WHERE aggregate_kind = 'project'
376 AND stream_id = ${request.streamId}
377 ORDER BY stream_version DESC
378 LIMIT 1
379 ),
380 0
381 ),
382 ${request.type},
383 ${request.occurredAt},
384 ${request.commandId},
385 ${request.causationEventId},
386 ${request.correlationId},
387 ${request.actorKind},
388 ${request.payloadJson},
389 ${request.metadataJson},
390 2
391 )
392 RETURNING
393 sequence,
394 event_id AS "eventId",
395 event_type AS "type",
396 aggregate_kind AS "aggregateKind",
397 stream_id AS "aggregateId",
398 occurred_at AS "occurredAt",
399 command_id AS "commandId",
400 causation_event_id AS "causationEventId",
401 correlation_id AS "correlationId",
402 payload_json AS "payload",
403 metadata_json AS "metadata"
404 `,
405 });
406
407 const appendProjectEvent: OrchestrationEventStore["Service"]["appendProjectEvent"] = (event) =>
408 appendProjectEventRow({
409 eventId: event.eventId,
410 streamId: event.aggregateId,
411 type: event.type,
412 causationEventId: event.causationEventId,
413 correlationId: event.correlationId,
414 actorKind: inferActorKind(event),
415 occurredAt: event.occurredAt,
416 commandId: event.commandId,
417 payloadJson:
418 event.type === "project.deleted" || !event.payload.projectIcon
419 ? event.payload
420 : { ...event.payload, projectIcon: encodeProjectIcon(event.payload.projectIcon) },
421 metadataJson: event.metadata,
422 }).pipe(
423 Effect.flatMap((row) => decodeProjectEvent(row)),
424 Effect.mapError(
425 toPersistenceSqlOrDecodeError(
426 "OrchestrationEventStore.appendProjectEvent:insert",
427 "OrchestrationEventStore.appendProjectEvent:decode",
428 ),
429 ),
430 );
431
432 const readApplicationRows = (input: {
433 readonly afterSequence: number;
434 readonly throughSequence?: number;
435 readonly threadId?: ThreadId;
436 readonly commandId?: CommandId;
437 readonly eventType?: OrchestrationV2DomainEvent["type"];
438 readonly onlyAgentEvents?: boolean;
439 readonly limit: number;
440 }) =>
441 sql<ApplicationEventRow>`
442 SELECT
443 sequence,
444 event_id,
445 command_id,
446 aggregate_kind,
447 stream_id,
448 event_type,
449 occurred_at,
450 ${restoredPayloadJson} AS payload_json,
451 metadata_json,
452 application_event_version,
453 causation_event_id,
454 correlation_id
455 FROM orchestration_events
456 ${
457 // The sequence range otherwise wins and a command lookup scans every
458 // V2 event, which blocks the server for seconds on large histories.
459 input.commandId !== undefined
460 ? sql`INDEXED BY idx_orch_events_command_id`
461 : input.onlyAgentEvents === true
462 ? sql``
463 : sql`INDEXED BY idx_orchestration_events_application_high_water`
464 }
465 WHERE sequence > ${input.afterSequence}
466 AND NOT (payload_json = 'null' AND EXISTS (
467 SELECT 1 FROM transcript_archives a WHERE a.thread_id = orchestration_events.stream_id AND orchestration_events.sequence <= a.through_sequence
468 ))
469 AND sequence <= ${input.throughSequence ?? Number.MAX_SAFE_INTEGER}
470 AND (
471 ${
472 input.onlyAgentEvents === true
473 ? sql`application_event_version = 2 AND aggregate_kind = 'thread'`
474 : sql`aggregate_kind = 'project'
475 OR (application_event_version = 2 AND aggregate_kind = 'thread')`
476 }
477 )
478 AND ${sql.and([
479 ...(input.threadId === undefined ? [] : [sql`stream_id = ${input.threadId}`]),
480 ...(input.commandId === undefined ? [] : [sql`command_id = ${input.commandId}`]),
481 ...(input.eventType === undefined ? [] : [sql`event_type = ${input.eventType}`]),
482 ])}
483 ORDER BY sequence ASC
484 LIMIT ${input.limit}
485 `;
486
487 const appendAgentEvents: OrchestrationEventStore["Service"]["appendAgentEvents"] = (input) =>
488 Effect.forEach(
489 input.events,
490 (event) =>
491 Effect.gen(function* () {
492 const encoded = yield* encodeV2EventJson(event);
493 const subagent =
494 encoded.type === "subagent.updated"
495 ? encoded.payload
496 : encoded.type === "turn-item.updated" && encoded.payload.type === "subagent"
497 ? encoded.payload
498 : undefined;
499 const promptKey = subagent === undefined ? undefined : `${event.type}:${subagent.id}`;
500 const hint = promptKey === undefined ? undefined : promptHints.get(promptKey);
501 const payloadJson = yield* encodeJson(encoded.payload);
502 // Progress ticks repeat an unchanged prompt, so store it once and point
503 // later rows at it. Keep an empty prompt so older servers still decode
504 // the row. ProjectionMaintenance never deletes these anchor rows. The
505 // hint is re-checked here because its transaction may have rolled back.
506 const storedPayload =
507 hint !== undefined && subagent !== undefined
508 ? sql`CASE WHEN EXISTS (
509 SELECT 1 FROM orchestration_events
510 WHERE sequence = ${hint}
511 AND json_type(payload_json, '$.promptSequence') IS NULL
512 AND json_extract(payload_json, '$.prompt') = ${subagent.prompt}
513 ) THEN json_set(${payloadJson}, '$.prompt', '', '$.promptSequence', CAST(${hint} AS INTEGER))
514 ELSE ${payloadJson} END`
515 : sql`${payloadJson}`;
516 const rows = yield* sql<{
517 readonly sequence: number;
518 readonly stores_full_prompt?: number;
519 }>`
520 INSERT INTO orchestration_events (
521 event_id,
522 aggregate_kind,
523 stream_id,
524 stream_version,
525 event_type,
526 occurred_at,
527 command_id,
528 causation_event_id,
529 correlation_id,
530 actor_kind,
531 payload_json,
532 metadata_json,
533 application_event_version
534 )
535 VALUES (
536 ${event.id},
537 'thread',
538 ${event.threadId},
539 COALESCE(
540 (
541 SELECT MAX(stream_version) + 1
542 FROM orchestration_events
543 WHERE aggregate_kind = 'thread' AND stream_id = ${event.threadId}
544 ),
545 0
546 ),
547 ${event.type},
548 ${encoded.occurredAt},
549 ${input.commandId ?? null},
550 NULL,
551 ${input.commandId ?? null},
552 ${event.rawEventId === undefined ? "server" : "provider"},
553 ${storedPayload},
554 ${yield* encodeJson(metadataForV2Event(event))},
555 2
556 )
557 RETURNING sequence${
558 hint === undefined
559 ? sql``
560 : sql`, json_type(payload_json, '$.promptSequence') IS NULL AS stores_full_prompt`
561 }
562 `;
563 const row = rows[0];
564 if (promptKey !== undefined && row !== undefined) {
565 // Delete first so Map order tracks recent use and eviction drops idle subagents.
566 promptHints.delete(promptKey);
567 promptHints.set(
568 promptKey,
569 hint === undefined || row.stores_full_prompt === 1 ? row.sequence : hint,
570 );
571 if (promptHints.size > 4096) {
572 const oldest = promptHints.keys().next().value;
573 if (oldest !== undefined) promptHints.delete(oldest);
574 }
575 }
576 return yield* decodeV2StoredEvent({
577 sequence: rows[0]?.sequence,
578 commandId: input.commandId ?? null,
579 event,
580 });
581 }),
582 { concurrency: 1 },
583 ).pipe(
584 Effect.mapError(
585 toPersistenceSqlOrDecodeError(
586 "OrchestrationEventStore.appendAgentEvents:insert",
587 "OrchestrationEventStore.appendAgentEvents:decode",
588 ),
589 ),
590 );
591
592 const readAgentEvents: OrchestrationEventStore["Service"]["readAgentEvents"] = (input) => {
593 const totalLimit =
594 input?.limit === undefined ? Number.MAX_SAFE_INTEGER : Math.max(0, Math.floor(input.limit));
595 if (totalLimit === 0) {
596 return Stream.empty;
597 }
598 return Stream.paginate(
599 { cursor: input?.afterSequence ?? 0, remaining: totalLimit },
600 ({ cursor, remaining }) => {
601 const pageLimit = Math.min(remaining, READ_PAGE_SIZE);
602 return readApplicationRows({
603 afterSequence: cursor,
604 ...(input?.throughSequence === undefined
605 ? {}
606 : { throughSequence: input.throughSequence }),
607 ...(input?.threadId === undefined ? {} : { threadId: input.threadId }),
608 ...(input?.commandId === undefined ? {} : { commandId: input.commandId }),
609 ...(input?.eventType === undefined ? {} : { eventType: input.eventType }),
610 onlyAgentEvents: true,
611 limit: pageLimit,
612 }).pipe(
613 Effect.mapError(toPersistenceSqlError("OrchestrationEventStore.readAgentEvents:query")),
614 Effect.map((rows) => {
615 const last = rows.at(-1);
616 const nextRemaining = remaining - rows.length;
617 return [
618 rows,
619 last === undefined ||
620 rows.length < pageLimit ||
621 nextRemaining <= 0 ||
622 (input?.throughSequence !== undefined && last.sequence >= input.throughSequence)
623 ? Option.none()
624 : Option.some({ cursor: last.sequence, remaining: nextRemaining }),
625 ] as const;
626 }),
627 );
628 },
629 ).pipe(
630 Stream.filterEffect((row) => keepReplayRow(row, input?.skipUnknownEventTypes)),
631 Stream.mapEffect((row) =>
632 rowToV2StoredEvent(row).pipe(
633 Effect.mapError(
634 toPersistenceDecodeError("OrchestrationEventStore.readAgentEvents:decode"),
635 ),
636 ),
637 ),
638 );
639 };
640
641 const getAgentReplayStats: OrchestrationEventStore["Service"]["getAgentReplayStats"] = (input) =>
642 sql<{
643 readonly eventCount: number;
644 readonly rawPayloadBytes: number;
645 readonly hasCreateEvent: number;
646 }>`
647 SELECT
648 COUNT(*) AS "eventCount",
649 COALESCE(SUM(octet_length(payload_json)), 0) AS "rawPayloadBytes",
650 COALESCE(MAX(event_type = 'thread.created'), 0) * (1 - COALESCE(MAX(payload_json = 'null'), 0)) AS "hasCreateEvent"
651 FROM (
652 SELECT ${restoredPayloadJson} AS payload_json, event_type
653 FROM orchestration_events
654 WHERE aggregate_kind = 'thread'
655 AND stream_id = ${input.threadId}
656 AND application_event_version = 2
657 AND sequence > ${input.afterSequence}
658 AND sequence <= ${input.throughSequence}
659 ORDER BY sequence ASC
660 LIMIT ${Math.max(0, Math.floor(input.maxEvents)) + 1}
661 )
662 `.pipe(
663 Effect.mapError(toPersistenceSqlError("OrchestrationEventStore.getAgentReplayStats:query")),
664 Effect.map((rows) => ({
665 eventCount: rows[0]?.eventCount ?? 0,
666 rawPayloadBytes: rows[0]?.rawPayloadBytes ?? 0,
667 hasCreateEvent: (rows[0]?.hasCreateEvent ?? 0) !== 0,
668 })),
669 );
670
671 const getReplayStats: OrchestrationEventStore["Service"]["getReplayStats"] = (input) =>
672 sql<{ readonly eventCount: number; readonly rawPayloadBytes: number }>`
673 SELECT
674 COUNT(*) AS "eventCount",
675 CASE WHEN MAX(payload_json = 'null') THEN 9007199254740991 ELSE COALESCE(SUM(octet_length(${restoredPayloadJson})), 0) END AS "rawPayloadBytes"
676 FROM orchestration_events INDEXED BY idx_orchestration_events_application_high_water
677 WHERE sequence > ${input.afterSequence}
678 AND sequence <= ${input.throughSequence}
679 AND (
680 aggregate_kind = 'project'
681 OR (application_event_version = 2 AND aggregate_kind = 'thread')
682 )
683 `.pipe(
684 Effect.mapError(toPersistenceSqlError("OrchestrationEventStore.getReplayStats:query")),
685 Effect.map((rows) => ({
686 eventCount: rows[0]?.eventCount ?? 0,
687 rawPayloadBytes: rows[0]?.rawPayloadBytes ?? 0,
688 })),
689 );
690
691 const latestAgentSequence: OrchestrationEventStore["Service"]["latestAgentSequence"] = (
692 threadId,
693 ) =>
694 sql<{ readonly sequence: number | null }>`
695 SELECT MAX(sequence) AS sequence
696 FROM orchestration_events
697 ${
698 threadId === undefined
699 ? sql``
700 : sql`INDEXED BY idx_orchestration_events_agent_stream_sequence`
701 }
702 WHERE application_event_version = 2
703 AND aggregate_kind = 'thread'
704 ${threadId === undefined ? sql`` : sql`AND stream_id = ${threadId}`}
705 `.pipe(
706 Effect.map((rows) => rows[0]?.sequence ?? 0),
707 Effect.mapError(toPersistenceSqlError("OrchestrationEventStore.latestAgentSequence:query")),
708 );
709
710 const getThreadSettlementTime: OrchestrationEventStore["Service"]["getThreadSettlementTime"] = (
711 threadId,
712 ) =>
713 sql<{ readonly settled_at: string | null }>`
714 SELECT CASE WHEN event_type = 'thread.settled' THEN occurred_at
715 ELSE json_extract(payload_json, '$.settledAt') END AS settled_at
716 FROM orchestration_events INDEXED BY idx_orchestration_events_agent_stream_sequence
717 WHERE application_event_version = 2 AND aggregate_kind = 'thread' AND stream_id = ${threadId}
718 AND (event_type = 'thread.settled' OR
719 (event_type = 'thread.created' AND json_extract(payload_json, '$.settledOverride') = 'settled'))
720 AND sequence > COALESCE((
721 SELECT MAX(sequence) FROM orchestration_events INDEXED BY idx_orchestration_events_agent_stream_sequence
722 WHERE application_event_version = 2 AND aggregate_kind = 'thread' AND stream_id = ${threadId}
723 AND event_type LIKE 'thread.%'
724 AND json_type(payload_json, '$.settledOverride') IS NOT NULL
725 AND json_extract(payload_json, '$.settledOverride') IS NOT 'settled'
726 ), 0)
727 ORDER BY sequence ASC LIMIT 1
728 `.pipe(
729 Effect.map((rows) =>
730 rows[0]?.settled_at == null ? null : DateTime.makeUnsafe(rows[0].settled_at),
731 ),
732 Effect.mapError(
733 toPersistenceSqlError("OrchestrationEventStore.getThreadSettlementTime:query"),
734 ),
735 );
736
737 // The OR planner otherwise scans every V2 event instead of seeking the final sequence.
738 const latestApplicationSequence = sql<{ readonly sequence: number | null }>`
739 SELECT MAX(sequence) AS sequence
740 FROM orchestration_events INDEXED BY idx_orchestration_events_application_high_water
741 WHERE aggregate_kind = 'project'
742 OR (application_event_version = 2 AND aggregate_kind = 'thread')
743 `.pipe(
744 Effect.map((rows) => rows[0]?.sequence ?? 0),
745 Effect.mapError(
746 toPersistenceSqlError("OrchestrationEventStore.latestApplicationSequence:query"),
747 ),
748 );
749
750 // Advance over raw rows so even a full page of unknown types cannot end replay early.
751 const catchUpApplicationEvents = (input: {
752 readonly afterSequence: number;
753 readonly throughSequence: number;
754 readonly skipUnknownEventTypes?: boolean;
755 }): Stream.Stream<ApplicationStoredEvent, OrchestrationEventStoreError> =>
756 Stream.paginate(input.afterSequence, (afterSequence) =>
757 readApplicationRows({
758 afterSequence,
759 throughSequence: input.throughSequence,
760 limit: READ_PAGE_SIZE,
761 }).pipe(
762 Effect.mapError(
763 toPersistenceSqlError("OrchestrationEventStore.readApplicationEvents:query"),
764 ),
765 Effect.map((rows) => {
766 const last = rows.at(-1);
767 return [
768 rows,
769 last === undefined ||
770 rows.length < READ_PAGE_SIZE ||
771 last.sequence >= input.throughSequence
772 ? Option.none()
773 : Option.some(last.sequence),
774 ] as const;
775 }),
776 ),
777 ).pipe(
778 Stream.filterEffect((row) => keepReplayRow(row, input.skipUnknownEventTypes)),
779 Stream.mapEffect((row) =>
780 rowToApplicationStoredEvent(row).pipe(
781 Effect.mapError(
782 toPersistenceDecodeError("OrchestrationEventStore.readApplicationEvents:decode"),
783 ),
784 ),
785 ),
786 );
787
788 const streamProjectedApplicationEvents: OrchestrationEventStore["Service"]["streamProjectedApplicationEvents"] =
789 (input) =>
790 replayAndBufferProjectedLiveEvents({
791 subscribe: PubSub.subscribe(committedEvents),
792 latestSequence: latestApplicationSequence,
793 afterSequence: input.afterSequence ?? 0,
794 project: input.project,
795 replay: (throughSequence) =>
796 catchUpApplicationEvents({
797 afterSequence: input?.afterSequence ?? 0,
798 throughSequence,
799 }),
800 }).pipe(
801 Stream.catchTag("LiveStreamBufferError", (cause) =>
802 Stream.fail(
803 toPersistenceSqlError("OrchestrationEventStore.streamApplicationEvents:buffer")(cause),
804 ),
805 ),
806 );
807
808 const streamApplicationEvents: OrchestrationEventStore["Service"]["streamApplicationEvents"] = (
809 input,
810 ) => streamProjectedApplicationEvents({ ...input, project: (event) => event });
811
812 return {
813 appendProjectEvent,
814 appendAgentEvents,
815 readAgentEvents,
816 getAgentReplayStats,
817 getReplayStats,
818 latestAgentSequence,
819 latestApplicationSequence,
820 getThreadSettlementTime,
821 readApplicationEvents: catchUpApplicationEvents,
822 publishCommitted: (events) => PubSub.publishAll(committedEvents, events).pipe(Effect.asVoid),
823 streamApplicationEvents,
824 streamProjectedApplicationEvents,
825 } satisfies OrchestrationEventStore["Service"];
826});
827
828export const layer = Layer.effect(OrchestrationEventStore, makeEventStore);