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