1import {
2 CommandId,
3 type OrchestrationV2Run,
4 OrchestrationV2DomainEvent,
5 type OrchestrationV2ProviderThread,
6 OrchestrationV2StoredEvent,
7 ProviderThreadId,
8 RunAttemptId,
9 RunId,
10 RuntimeRequestId,
11 NodeId,
12 type ProjectId,
13 ThreadId,
14} from "@t3tools/contracts";
15import * as Context from "effect/Context";
16import * as DateTime from "effect/DateTime";
17import * as Effect from "effect/Effect";
18import * as Layer from "effect/Layer";
19import * as Option from "effect/Option";
20import * as PubSub from "effect/PubSub";
21import * as Semaphore from "effect/Semaphore";
22import * as Schema from "effect/Schema";
23import * as Stream from "effect/Stream";
24import * as SqlClient from "effect/sql/SqlClient";
25
26import { replayAndBufferProjectedLiveEvents } from "./LiveStreamBudget.ts";
27import type { UnsequencedProjectEvent } from "../persistence/OrchestrationEventStore.ts";
28import { projectDomainEventForWire } from "./WireProjection.ts";
29
30import * as CommandReceiptStore from "./CommandReceiptStore.ts";
31import * as EffectOutbox from "./EffectOutbox.ts";
32import * as EventStore from "./EventStore.ts";
33import * as ProjectionStore from "./ProjectionStore.ts";
34import * as ProjectStore from "./ProjectStore.ts";
35import * as TurnItemPositionStore from "./TurnItemPositionStore.ts";
36
37/**
38 * ERRORS
39 */
40export class EventSinkWriteError extends Schema.TaggedError<EventSinkWriteError>()(
41 "EventSinkWriteError",
42 {
43 eventCount: Schema.Number,
44 commandId: Schema.optional(CommandId),
45 cause: Schema.optional(Schema.Defect()),
46 },
47) {
48 override get message(): string {
49 return `Failed to write ${this.eventCount} orchestration V2 event(s).`;
50 }
51}
52
53export class EventSinkStreamError extends Schema.TaggedError<EventSinkStreamError>()(
54 "EventSinkStreamError",
55 {
56 threadId: Schema.optional(ThreadId),
57 afterSequence: Schema.optional(Schema.Number),
58 cause: Schema.optional(Schema.Defect()),
59 },
60) {
61 override get message(): string {
62 return this.threadId === undefined
63 ? "Failed to stream orchestration V2 events."
64 : `Failed to stream orchestration V2 events for thread ${this.threadId}.`;
65 }
66}
67
68export const EventSinkV2Error = Schema.Union([EventSinkWriteError, EventSinkStreamError]);
69export type EventSinkV2Error = typeof EventSinkV2Error.Type;
70
71/**
72 * SERVICE DEFINITION
73 */
74export interface EventSinkV2Shape {
75 readonly write: (input: {
76 readonly guardPendingUserInputCancellations?: boolean;
77 readonly commandId?: CommandId;
78 readonly events: ReadonlyArray<OrchestrationV2DomainEvent>;
79 }) => Effect.Effect<ReadonlyArray<OrchestrationV2StoredEvent>, EventSinkV2Error>;
80 readonly writeWithEffects: (input: {
81 readonly guardPendingUserInputCancellations?: boolean;
82 readonly commandId?: CommandId;
83 readonly events: ReadonlyArray<OrchestrationV2DomainEvent>;
84 readonly effects: ReadonlyArray<EffectOutbox.PendingOrchestrationEffectV2>;
85 }) => Effect.Effect<ReadonlyArray<OrchestrationV2StoredEvent>, EventSinkV2Error>;
86 readonly writeIfRunCurrent: (input: {
87 readonly guardPendingUserInputCancellations?: boolean;
88 readonly commandId?: CommandId;
89 readonly threadId: ThreadId;
90 readonly runId: RunId;
91 readonly activeAttemptId: RunAttemptId;
92 readonly expectedStatus: OrchestrationV2Run["status"];
93 readonly events: ReadonlyArray<OrchestrationV2DomainEvent>;
94 readonly effects?: ReadonlyArray<EffectOutbox.PendingOrchestrationEffectV2>;
95 }) => Effect.Effect<
96 {
97 readonly committed: boolean;
98 readonly storedEvents: ReadonlyArray<OrchestrationV2StoredEvent>;
99 },
100 EventSinkV2Error
101 >;
102 /**
103 * Atomically commit only when the provider thread is still owned by the
104 * expected run attempt and ordinal. Used for late post-terminal
105 * provider_thread updates so a completed or superseded attempt cannot clobber
106 * a newer attempt that already claimed the thread.
107 */
108 readonly writeIfProviderThreadOwner: (input: {
109 readonly guardPendingUserInputCancellations?: boolean;
110 readonly commandId?: CommandId;
111 readonly providerThreadId: ProviderThreadId;
112 readonly runId: RunId;
113 readonly activeAttemptId: RunAttemptId;
114 readonly expectedLastRunOrdinal: number;
115 readonly events: ReadonlyArray<OrchestrationV2DomainEvent>;
116 }) => Effect.Effect<
117 {
118 readonly committed: boolean;
119 readonly storedEvents: ReadonlyArray<OrchestrationV2StoredEvent>;
120 },
121 EventSinkV2Error
122 >;
123 /**
124 * Atomically commit only while the stored provider thread still equals
125 * `expected`, the row its writer read, so a read-modify-write of the row
126 * cannot overwrite an update published in between. The whole row is
127 * compared because not every writer moves `updatedAt`.
128 */
129 readonly writeIfProviderThreadUnchanged: (input: {
130 readonly expected: OrchestrationV2ProviderThread;
131 readonly events: ReadonlyArray<OrchestrationV2DomainEvent>;
132 }) => Effect.Effect<
133 {
134 readonly committed: boolean;
135 readonly storedEvents: ReadonlyArray<OrchestrationV2StoredEvent>;
136 },
137 EventSinkV2Error
138 >;
139 readonly commitCommand: (input: {
140 readonly commandId: CommandId;
141 readonly threadId: ThreadId;
142 readonly commandType: string;
143 readonly acceptedAt: DateTime.Utc;
144 readonly events: ReadonlyArray<OrchestrationV2DomainEvent>;
145 readonly effects: ReadonlyArray<EffectOutbox.PendingOrchestrationEffectV2>;
146 readonly cancelUnsettledEffects?: {
147 readonly effectTypes: ReadonlyArray<EffectOutbox.OrchestrationEffectRequestV2["type"]>;
148 readonly reason: string;
149 };
150 }) => Effect.Effect<
151 {
152 readonly receipt: CommandReceiptStore.CommandReceiptV2;
153 readonly storedEvents: ReadonlyArray<OrchestrationV2StoredEvent>;
154 readonly committed: boolean;
155 readonly cancelledEffectCount: number;
156 },
157 EventSinkV2Error
158 >;
159 readonly commitRejectedCommand: (input: {
160 readonly commandId: CommandId;
161 readonly threadId: ThreadId;
162 readonly commandType: string;
163 readonly rejectedAt: DateTime.Utc;
164 readonly error: string;
165 }) => Effect.Effect<CommandReceiptStore.CommandReceiptV2, EventSinkV2Error>;
166 /**
167 * Append a project event, fold it into its row and record the receipt in one
168 * transaction. A reused command id commits nothing and returns its receipt.
169 */
170 readonly commitProjectCommand: (input: {
171 readonly commandId: CommandId;
172 readonly projectId: ProjectId;
173 readonly commandType: string;
174 readonly acceptedAt: DateTime.Utc;
175 readonly event: UnsequencedProjectEvent;
176 }) => Effect.Effect<
177 { readonly receipt: CommandReceiptStore.ProjectCommandReceiptV2; readonly committed: boolean },
178 EventSinkV2Error
179 >;
180 /** Record a rejected project command, or return the receipt its command id already has. */
181 readonly commitRejectedProjectCommand: (input: {
182 readonly commandId: CommandId;
183 readonly projectId: ProjectId;
184 readonly commandType: string;
185 readonly rejectedAt: DateTime.Utc;
186 readonly error: string;
187 }) => Effect.Effect<CommandReceiptStore.ProjectCommandReceiptV2, EventSinkV2Error>;
188 readonly stream: (input?: {
189 readonly threadId?: ThreadId;
190 readonly afterSequence?: number;
191 /** Filter before queuing live events so a busy worker retains only the events it handles. */
192 readonly eventType?: OrchestrationV2DomainEvent["type"];
193 /** Bound RPC subscribers; internal workers must not drop their subscription under load. */
194 readonly bounded?: boolean;
195 }) => Stream.Stream<OrchestrationV2StoredEvent, EventSinkV2Error>;
196 readonly latestSequence: (input?: {
197 readonly threadId?: ThreadId;
198 }) => Effect.Effect<number, EventSinkV2Error>;
199 readonly readByCommandId: (input: {
200 readonly commandId: CommandId;
201 }) => Stream.Stream<OrchestrationV2StoredEvent, EventSinkV2Error>;
202}
203
204export class EventSinkV2 extends Context.Service<EventSinkV2, EventSinkV2Shape>()(
205 "t3/orchestration-v2/EventSink/EventSinkV2",
206) {}
207
208/**
209 * IMPLEMENTATIONS
210 */
211const layerBase: Layer.Layer<
212 EventSinkV2,
213 never,
214 | CommandReceiptStore.CommandReceiptStoreV2
215 | EffectOutbox.EffectOutboxV2
216 | EventStore.EventStoreV2
217 | ProjectionStore.ProjectionStoreV2
218 | ProjectStore.ProjectStoreV2
219 | SqlClient.SqlClient
220 | TurnItemPositionStore.TurnItemPositionStoreV2
221> = Layer.effect(
222 EventSinkV2,
223 Effect.gen(function* () {
224 const sql = yield* SqlClient.SqlClient;
225 const commandReceipts = yield* CommandReceiptStore.CommandReceiptStoreV2;
226 const effectOutbox = yield* EffectOutbox.EffectOutboxV2;
227 const eventStore = yield* EventStore.EventStoreV2;
228 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
229 const projectStore = yield* ProjectStore.ProjectStoreV2;
230 const turnItemPositions = yield* TurnItemPositionStore.TurnItemPositionStoreV2;
231 const liveEvents = yield* PubSub.unbounded<OrchestrationV2StoredEvent>();
232 const liveEventsByType = new Map<
233 OrchestrationV2DomainEvent["type"],
234 PubSub.PubSub<OrchestrationV2StoredEvent>
235 >();
236 const publishLiveEvents = (events: ReadonlyArray<OrchestrationV2StoredEvent>) =>
237 Effect.gen(function* () {
238 yield* PubSub.publishAll(liveEvents, events);
239 for (const [type, pubsub] of liveEventsByType) {
240 yield* PubSub.publishAll(
241 pubsub,
242 events.filter((stored) => stored.event.type === type),
243 );
244 }
245 });
246 const publishStoredEvents = (events: ReadonlyArray<OrchestrationV2StoredEvent>) =>
247 eventStore.publishCommitted(events).pipe(Effect.andThen(publishLiveEvents(events)));
248
249 // Transactions commit one at a time, but each writer publishes after its
250 // commit. If a writer is descheduled in between, a later commit reaches
251 // subscribers first, and clients drop any event at or below the newest
252 // sequence they have applied. So a writer takes this lane as the last step
253 // of its transaction and holds it until it has published. Publishing never
254 // waits, so a writer that holds the transaction while it waits for the
255 // lane is not blocked for long.
256 const publishLane = yield* Semaphore.make(1);
257 const commitThenPublish = <A, E, R>(
258 transaction: Effect.Effect<A, E, R>,
259 publish: (committed: A) => Effect.Effect<void>,
260 ) =>
261 Effect.suspend(() => {
262 let holdsLane = false;
263 const takeLane = publishLane.take(1).pipe(
264 Effect.andThen(
265 Effect.sync(() => {
266 holdsLane = true;
267 }),
268 ),
269 Effect.uninterruptible,
270 );
271 return sql
272 .withTransaction(Effect.tap(transaction, () => takeLane))
273 .pipe(
274 Effect.tap(publish),
275 Effect.ensuring(
276 Effect.suspend(() => (holdsLane ? publishLane.release(1) : Effect.void)),
277 ),
278 );
279 });
280
281 // A user can answer after terminal normalization reads the pending request.
282 // Recheck inside the write transaction so stale cleanup cannot erase answers.
283 const guardUserInputCancellations = (events: ReadonlyArray<OrchestrationV2DomainEvent>) =>
284 Effect.gen(function* () {
285 const staleRequests = new Set<RuntimeRequestId>();
286 const staleNodes = new Set<NodeId>();
287 for (const event of events) {
288 if (
289 event.type !== "runtime-request.updated" ||
290 event.payload.kind !== "user_input" ||
291 event.payload.status !== "cancelled"
292 )
293 continue;
294 const current = yield* projectionStore.getRuntimeRequest(
295 event.threadId,
296 event.payload.id,
297 );
298 if (
299 current?.status !== "pending" ||
300 current.kind !== "user_input" ||
301 current.providerTurnId !== event.payload.providerTurnId ||
302 current.responseCapability.type === "message"
303 ) {
304 staleRequests.add(event.payload.id);
305 staleNodes.add(event.payload.nodeId);
306 }
307 }
308 return events.filter((event) => {
309 switch (event.type) {
310 case "runtime-request.updated":
311 return event.payload.status !== "cancelled" || !staleRequests.has(event.payload.id);
312 case "node.updated":
313 return event.payload.status !== "cancelled" || !staleNodes.has(event.payload.id);
314 case "turn-item.updated":
315 return (
316 event.payload.type !== "user_input_request" ||
317 event.payload.status !== "cancelled" ||
318 !staleRequests.has(event.payload.requestId)
319 );
320 default:
321 return true;
322 }
323 });
324 });
325
326 const normalizeEvents = (events: ReadonlyArray<OrchestrationV2DomainEvent>) => {
327 const runOrdinals = new Map(
328 events.flatMap((event) =>
329 event.type === "run.created" || event.type === "run.updated"
330 ? [[event.payload.id, event.payload.ordinal] as const]
331 : [],
332 ),
333 );
334 return Effect.forEach(
335 events,
336 (event): Effect.Effect<OrchestrationV2DomainEvent, unknown> =>
337 event.type === "turn-item.updated"
338 ? turnItemPositions
339 .normalize(
340 event.payload,
341 event.payload.runId === null ? undefined : runOrdinals.get(event.payload.runId),
342 )
343 .pipe(Effect.map((payload) => ({ ...event, payload })))
344 : Effect.succeed(event),
345 { concurrency: 1 },
346 );
347 };
348
349 const applyStoredEvents = (storedEvents: ReadonlyArray<OrchestrationV2StoredEvent>) =>
350 Effect.gen(function* () {
351 yield* Effect.forEach(storedEvents, (stored) => projectionStore.apply(stored.event), {
352 concurrency: 1,
353 });
354 const sequence = storedEvents.at(-1)?.sequence;
355 if (sequence !== undefined) {
356 const now = DateTime.formatIso(yield* DateTime.now);
357 yield* sql`
358 INSERT INTO orchestration_v2_projection_metadata (
359 projection_name,
360 schema_version,
361 last_sequence,
362 updated_at
363 )
364 VALUES (
365 'thread-projections',
366 ${ProjectionStore.ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION},
367 ${sequence},
368 ${now}
369 )
370 ON CONFLICT(projection_name)
371 DO UPDATE SET
372 schema_version = excluded.schema_version,
373 last_sequence = excluded.last_sequence,
374 updated_at = excluded.updated_at
375 `;
376 }
377 });
378
379 const writeEffect = Effect.fn("orchestrationV2.EventSink.write")(function* (
380 input: Parameters<EventSinkV2Shape["writeWithEffects"]>[0],
381 ) {
382 yield* Effect.annotateCurrentSpan({
383 "orchestration_v2.command_id": input.commandId ?? null,
384 "orchestration_v2.event_count": input.events.length,
385 "orchestration_v2.thread_id": input.events[0]?.threadId ?? null,
386 });
387
388 return yield* commitThenPublish(
389 Effect.gen(function* () {
390 const normalized = yield* normalizeEvents(
391 input.guardPendingUserInputCancellations === true
392 ? yield* guardUserInputCancellations(input.events)
393 : input.events,
394 );
395 const committed = yield* eventStore.append({
396 ...(input.commandId === undefined ? {} : { commandId: input.commandId }),
397 events: normalized,
398 });
399 yield* applyStoredEvents(committed);
400 yield* effectOutbox.enqueue(input.effects);
401 return committed;
402 }).pipe(
403 // Every streamed agent event is a write, so its statements would
404 // dominate the trace file. This span and its sql.transaction span
405 // still time each write, including the commit.
406 Effect.withTracerEnabled(false),
407 ),
408 (storedEvents) =>
409 Effect.gen(function* () {
410 if (input.effects.length > 0) {
411 yield* effectOutbox.notifyAvailable(input.effects.length);
412 }
413 yield* publishStoredEvents(storedEvents);
414 }),
415 );
416 });
417
418 const writeIfRunCurrentEffect = Effect.fn("orchestrationV2.EventSink.writeIfRunCurrent")(
419 function* (input: Parameters<EventSinkV2Shape["writeIfRunCurrent"]>[0]) {
420 yield* Effect.annotateCurrentSpan({
421 "orchestration_v2.command_id": input.commandId ?? null,
422 "orchestration_v2.event_count": input.events.length,
423 "orchestration_v2.run_id": input.runId,
424 "orchestration_v2.thread_id": input.threadId,
425 });
426
427 return yield* commitThenPublish(
428 Effect.gen(function* () {
429 const rows = yield* sql<{
430 readonly status: string;
431 readonly active_attempt_id: string | null;
432 }>`
433 SELECT
434 status,
435 json_extract(payload_json, '$.activeAttemptId') AS active_attempt_id
436 FROM orchestration_v2_projection_runs
437 WHERE run_id = ${input.runId}
438 AND thread_id = ${input.threadId}
439 LIMIT 1
440 `;
441 const current = rows[0];
442 if (
443 current === undefined ||
444 current.status !== input.expectedStatus ||
445 current.active_attempt_id !== input.activeAttemptId
446 ) {
447 return {
448 committed: false as const,
449 storedEvents: [] as ReadonlyArray<OrchestrationV2StoredEvent>,
450 };
451 }
452
453 const normalized = yield* normalizeEvents(
454 input.guardPendingUserInputCancellations === true
455 ? yield* guardUserInputCancellations(input.events)
456 : input.events,
457 );
458 const storedEvents = yield* eventStore.append({
459 ...(input.commandId === undefined ? {} : { commandId: input.commandId }),
460 events: normalized,
461 });
462 yield* applyStoredEvents(storedEvents);
463 yield* effectOutbox.enqueue(input.effects ?? []);
464 return { committed: true as const, storedEvents };
465 }),
466 (result) =>
467 Effect.gen(function* () {
468 if (!result.committed) return;
469 if (input.effects !== undefined && input.effects.length > 0) {
470 yield* effectOutbox.notifyAvailable(input.effects.length);
471 }
472 yield* publishStoredEvents(result.storedEvents);
473 }),
474 );
475 },
476 );
477
478 const writeIfProviderThreadOwnerEffect = Effect.fn(
479 "orchestrationV2.EventSink.writeIfProviderThreadOwner",
480 )(function* (input: Parameters<EventSinkV2Shape["writeIfProviderThreadOwner"]>[0]) {
481 yield* Effect.annotateCurrentSpan({
482 "orchestration_v2.command_id": input.commandId ?? null,
483 "orchestration_v2.event_count": input.events.length,
484 "orchestration_v2.provider_thread_id": input.providerThreadId,
485 "orchestration_v2.run_id": input.runId,
486 "orchestration_v2.active_attempt_id": input.activeAttemptId,
487 "orchestration_v2.expected_last_run_ordinal": input.expectedLastRunOrdinal,
488 });
489
490 return yield* commitThenPublish(
491 Effect.gen(function* () {
492 const rows = yield* sql<{
493 readonly active_attempt_id: string | null;
494 readonly last_run_ordinal: number | null;
495 }>`
496 SELECT
497 json_extract(r.payload_json, '$.activeAttemptId') AS active_attempt_id,
498 p.last_run_ordinal
499 FROM orchestration_v2_projection_provider_threads p
500 JOIN orchestration_v2_projection_runs r
501 ON r.run_id = ${input.runId}
502 AND r.thread_id = p.thread_id
503 WHERE p.provider_thread_id = ${input.providerThreadId}
504 LIMIT 1
505 `;
506 const current = rows[0];
507 if (
508 current === undefined ||
509 current.active_attempt_id !== input.activeAttemptId ||
510 current.last_run_ordinal !== input.expectedLastRunOrdinal
511 ) {
512 return {
513 committed: false as const,
514 storedEvents: [] as ReadonlyArray<OrchestrationV2StoredEvent>,
515 };
516 }
517
518 const normalized = yield* normalizeEvents(
519 input.guardPendingUserInputCancellations === true
520 ? yield* guardUserInputCancellations(input.events)
521 : input.events,
522 );
523 const storedEvents = yield* eventStore.append({
524 ...(input.commandId === undefined ? {} : { commandId: input.commandId }),
525 events: normalized,
526 });
527 yield* applyStoredEvents(storedEvents);
528 return { committed: true as const, storedEvents };
529 }),
530 (result) => (result.committed ? publishStoredEvents(result.storedEvents) : Effect.void),
531 );
532 });
533
534 const writeIfProviderThreadUnchangedEffect = Effect.fn(
535 "orchestrationV2.EventSink.writeIfProviderThreadUnchanged",
536 )(function* (input: Parameters<EventSinkV2Shape["writeIfProviderThreadUnchanged"]>[0]) {
537 yield* Effect.annotateCurrentSpan({
538 "orchestration_v2.event_count": input.events.length,
539 "orchestration_v2.provider_thread_id": input.expected.id,
540 });
541 const expectedJson = yield* ProjectionStore.encodeProviderThreadPayload(input.expected);
542
543 return yield* commitThenPublish(
544 Effect.gen(function* () {
545 const rows = yield* sql<{ readonly payload_json: string }>`
546 SELECT payload_json
547 FROM orchestration_v2_projection_provider_threads
548 WHERE provider_thread_id = ${input.expected.id}
549 LIMIT 1
550 `;
551 if (rows[0]?.payload_json !== expectedJson) {
552 return {
553 committed: false as const,
554 storedEvents: [] as ReadonlyArray<OrchestrationV2StoredEvent>,
555 };
556 }
557
558 const normalized = yield* normalizeEvents(input.events);
559 const storedEvents = yield* eventStore.append({ events: normalized });
560 yield* applyStoredEvents(storedEvents);
561 return { committed: true as const, storedEvents };
562 }),
563 (result) => (result.committed ? publishStoredEvents(result.storedEvents) : Effect.void),
564 );
565 });
566
567 const existingCommandResult = (commandId: CommandId) =>
568 Effect.gen(function* () {
569 const existing = yield* commandReceipts.getByCommandId(commandId);
570 if (Option.isNone(existing)) {
571 return yield* Effect.die(
572 new Error(`Command receipt ${commandId} disappeared during its transaction.`),
573 );
574 }
575 const storedEvents = yield* eventStore.readByCommandId({ commandId }).pipe(
576 Stream.runCollect,
577 Effect.map((events): ReadonlyArray<OrchestrationV2StoredEvent> => Array.from(events)),
578 );
579 return { receipt: existing.value, storedEvents };
580 });
581
582 const commitCommandEffect = Effect.fn("orchestrationV2.EventSink.commitCommand")(function* (
583 input: Parameters<EventSinkV2Shape["commitCommand"]>[0],
584 ) {
585 const result = yield* commitThenPublish(
586 Effect.gen(function* () {
587 const reserved = yield* commandReceipts.insertIfAbsent({
588 commandId: input.commandId,
589 threadId: input.threadId,
590 commandType: input.commandType,
591 acceptedAt: input.acceptedAt,
592 resultSequence: 0,
593 status: "accepted",
594 error: null,
595 });
596 if (!reserved) {
597 const existing = yield* existingCommandResult(input.commandId);
598 return { ...existing, committed: false as const, cancelledEffectIds: [] };
599 }
600
601 const normalized = yield* normalizeEvents(input.events);
602 const storedEvents = yield* eventStore.append({
603 commandId: input.commandId,
604 events: normalized,
605 });
606 const sequence = storedEvents.at(-1)?.sequence;
607 if (sequence === undefined) {
608 return yield* Effect.die(
609 new Error(`Command ${input.commandId} produced no orchestration events.`),
610 );
611 }
612 yield* applyStoredEvents(storedEvents);
613 yield* effectOutbox.enqueue(input.effects);
614 const receipt: CommandReceiptStore.CommandReceiptV2 = {
615 commandId: input.commandId,
616 threadId: input.threadId,
617 commandType: input.commandType,
618 acceptedAt: input.acceptedAt,
619 resultSequence: sequence,
620 status: "accepted",
621 error: null,
622 };
623 yield* commandReceipts.upsert(receipt);
624 const cancelledEffectIds =
625 input.cancelUnsettledEffects === undefined
626 ? []
627 : yield* effectOutbox.cancelUnsettled({
628 threadId: input.threadId,
629 ...input.cancelUnsettledEffects,
630 });
631 return { receipt, storedEvents, committed: true as const, cancelledEffectIds };
632 }),
633 (result) =>
634 Effect.gen(function* () {
635 yield* effectOutbox.signalCancellations(result.cancelledEffectIds);
636 if (result.committed && input.effects.length > 0) {
637 yield* effectOutbox.notifyAvailable(input.effects.length);
638 }
639 if (result.committed) yield* publishStoredEvents(result.storedEvents);
640 }),
641 );
642 return {
643 receipt: result.receipt,
644 storedEvents: result.storedEvents,
645 committed: result.committed,
646 cancelledEffectCount: result.cancelledEffectIds.length,
647 };
648 });
649
650 const commitRejectedCommandEffect = Effect.fn(
651 "orchestrationV2.EventSink.commitRejectedCommand",
652 )(function* (input: Parameters<EventSinkV2Shape["commitRejectedCommand"]>[0]) {
653 return yield* sql.withTransaction(
654 Effect.gen(function* () {
655 const sequence = yield* eventStore.latestSequence({ threadId: input.threadId });
656 const receipt: CommandReceiptStore.CommandReceiptV2 = {
657 commandId: input.commandId,
658 threadId: input.threadId,
659 commandType: input.commandType,
660 acceptedAt: input.rejectedAt,
661 resultSequence: sequence,
662 status: "rejected",
663 error: input.error,
664 };
665 const inserted = yield* commandReceipts.insertIfAbsent(receipt);
666 if (inserted) {
667 return receipt;
668 }
669 const existing = yield* commandReceipts.getByCommandId(input.commandId);
670 return Option.getOrElse(existing, () => receipt);
671 }),
672 );
673 });
674
675 const existingProjectReceipt = (commandId: CommandId) =>
676 commandReceipts.getProjectByCommandId(commandId).pipe(
677 Effect.flatMap(
678 Option.match({
679 onNone: () =>
680 Effect.fail(`Command ${commandId} was already used by a thread command.` as const),
681 onSome: Effect.succeed,
682 }),
683 ),
684 );
685
686 const commitProjectCommandEffect = Effect.fn("orchestrationV2.EventSink.commitProjectCommand")(
687 function* (input: Parameters<EventSinkV2Shape["commitProjectCommand"]>[0]) {
688 const result = yield* commitThenPublish(
689 Effect.gen(function* () {
690 const reserved: CommandReceiptStore.ProjectCommandReceiptV2 = {
691 commandId: input.commandId,
692 projectId: input.projectId,
693 commandType: input.commandType,
694 acceptedAt: input.acceptedAt,
695 resultSequence: 0,
696 status: "accepted",
697 error: null,
698 };
699 if (!(yield* commandReceipts.insertIfAbsent(reserved))) {
700 return { receipt: yield* existingProjectReceipt(input.commandId), event: undefined };
701 }
702 const event = yield* eventStore.appendProjectEvent(input.event);
703 yield* projectStore.apply(event);
704 const receipt = { ...reserved, resultSequence: event.sequence };
705 yield* commandReceipts.upsert(receipt);
706 return { receipt, event };
707 }),
708 (result) =>
709 result.event === undefined ? Effect.void : eventStore.publishCommitted([result.event]),
710 );
711 return { receipt: result.receipt, committed: result.event !== undefined };
712 },
713 );
714
715 const commitRejectedProjectCommandEffect = Effect.fn(
716 "orchestrationV2.EventSink.commitRejectedProjectCommand",
717 )(function* (input: Parameters<EventSinkV2Shape["commitRejectedProjectCommand"]>[0]) {
718 return yield* sql.withTransaction(
719 Effect.gen(function* () {
720 const receipt: CommandReceiptStore.ProjectCommandReceiptV2 = {
721 commandId: input.commandId,
722 projectId: input.projectId,
723 commandType: input.commandType,
724 acceptedAt: input.rejectedAt,
725 resultSequence: yield* eventStore.latestApplicationSequence,
726 status: "rejected",
727 error: input.error,
728 };
729 return (yield* commandReceipts.insertIfAbsent(receipt))
730 ? receipt
731 : yield* existingProjectReceipt(input.commandId);
732 }),
733 );
734 });
735
736 const catchUp = (input: {
737 readonly afterSequence: number;
738 readonly throughSequence: number;
739 readonly threadId?: ThreadId;
740 readonly eventType?: OrchestrationV2DomainEvent["type"];
741 }): Stream.Stream<OrchestrationV2StoredEvent, unknown> => {
742 const pageSize = 256;
743 const loop = (afterSequence: number): Stream.Stream<OrchestrationV2StoredEvent, unknown> =>
744 Stream.unwrap(
745 eventStore
746 .read({
747 afterSequence,
748 throughSequence: input.throughSequence,
749 ...(input.threadId === undefined ? {} : { threadId: input.threadId }),
750 ...(input.eventType === undefined ? {} : { eventType: input.eventType }),
751 limit: pageSize,
752 })
753 .pipe(
754 Stream.runCollect,
755 Effect.map((chunk) => Array.from(chunk)),
756 Effect.map((events) => {
757 if (events.length === 0) {
758 return Stream.empty;
759 }
760 const current = Stream.fromIterable(events);
761 const last = events.at(-1)?.sequence ?? input.throughSequence;
762 return events.length < pageSize || last >= input.throughSequence
763 ? current
764 : Stream.concat(current, loop(last));
765 }),
766 ),
767 );
768 return loop(input.afterSequence);
769 };
770
771 const stream = (input?: Parameters<EventSinkV2Shape["stream"]>[0]) => {
772 const afterSequence = input?.afterSequence ?? 0;
773 const matches = (stored: OrchestrationV2StoredEvent) =>
774 (input?.threadId === undefined || stored.event.threadId === input.threadId) &&
775 (input?.eventType === undefined || stored.event.type === input.eventType);
776 const replay = (throughSequence: number) =>
777 catchUp({
778 afterSequence,
779 throughSequence,
780 ...(input?.threadId === undefined ? {} : { threadId: input.threadId }),
781 ...(input?.eventType === undefined ? {} : { eventType: input.eventType }),
782 }).pipe(Stream.filter(matches));
783 return Stream.unwrap(
784 Effect.gen(function* () {
785 let pubsub = liveEvents;
786 if (input?.eventType !== undefined) {
787 const existing = liveEventsByType.get(input.eventType);
788 if (existing !== undefined) {
789 pubsub = existing;
790 } else {
791 const created = yield* PubSub.unbounded<OrchestrationV2StoredEvent>();
792 pubsub = liveEventsByType.get(input.eventType) ?? created;
793 liveEventsByType.set(input.eventType, pubsub);
794 }
795 }
796 if (input?.bounded === true) {
797 return replayAndBufferProjectedLiveEvents({
798 subscribe: PubSub.subscribe(pubsub),
799 latestSequence: eventStore.latestSequence(),
800 afterSequence,
801 filter: matches,
802 replay,
803 project: (stored) => ({ ...stored, event: projectDomainEventForWire(stored.event) }),
804 });
805 }
806 const subscription = yield* PubSub.subscribe(pubsub);
807 const highWater = yield* eventStore.latestSequence();
808 const live = Stream.fromSubscription(subscription).pipe(
809 Stream.filter((stored) => stored.sequence > Math.max(highWater, afterSequence)),
810 Stream.filter(matches),
811 );
812 return Stream.concat(replay(highWater), live);
813 }),
814 );
815 };
816
817 return EventSinkV2.of({
818 write: (input) =>
819 writeEffect({ ...input, effects: [] }).pipe(
820 Effect.mapError(
821 (cause) =>
822 new EventSinkWriteError({
823 eventCount: input.events.length,
824 ...(input.commandId === undefined ? {} : { commandId: input.commandId }),
825 cause,
826 }),
827 ),
828 ),
829 writeWithEffects: (input) =>
830 writeEffect(input).pipe(
831 Effect.mapError(
832 (cause) =>
833 new EventSinkWriteError({
834 eventCount: input.events.length,
835 ...(input.commandId === undefined ? {} : { commandId: input.commandId }),
836 cause,
837 }),
838 ),
839 ),
840 writeIfRunCurrent: (input) =>
841 writeIfRunCurrentEffect(input).pipe(
842 Effect.mapError(
843 (cause) =>
844 new EventSinkWriteError({
845 eventCount: input.events.length,
846 ...(input.commandId === undefined ? {} : { commandId: input.commandId }),
847 cause,
848 }),
849 ),
850 ),
851 writeIfProviderThreadOwner: (input) =>
852 writeIfProviderThreadOwnerEffect(input).pipe(
853 Effect.mapError(
854 (cause) =>
855 new EventSinkWriteError({
856 eventCount: input.events.length,
857 ...(input.commandId === undefined ? {} : { commandId: input.commandId }),
858 cause,
859 }),
860 ),
861 ),
862 writeIfProviderThreadUnchanged: (input) =>
863 writeIfProviderThreadUnchangedEffect(input).pipe(
864 Effect.mapError(
865 (cause) => new EventSinkWriteError({ eventCount: input.events.length, cause }),
866 ),
867 ),
868 commitCommand: (input) =>
869 commitCommandEffect(input).pipe(
870 Effect.mapError(
871 (cause) =>
872 new EventSinkWriteError({
873 commandId: input.commandId,
874 eventCount: input.events.length,
875 cause,
876 }),
877 ),
878 ),
879 commitRejectedCommand: (input) =>
880 commitRejectedCommandEffect(input).pipe(
881 Effect.mapError(
882 (cause) =>
883 new EventSinkWriteError({
884 commandId: input.commandId,
885 eventCount: 0,
886 cause,
887 }),
888 ),
889 ),
890 commitProjectCommand: (input) =>
891 commitProjectCommandEffect(input).pipe(
892 Effect.mapError(
893 (cause) =>
894 new EventSinkWriteError({ commandId: input.commandId, eventCount: 1, cause }),
895 ),
896 ),
897 commitRejectedProjectCommand: (input) =>
898 commitRejectedProjectCommandEffect(input).pipe(
899 Effect.mapError(
900 (cause) =>
901 new EventSinkWriteError({ commandId: input.commandId, eventCount: 0, cause }),
902 ),
903 ),
904 stream: (input) =>
905 stream(input).pipe(
906 Stream.mapError(
907 (cause) =>
908 new EventSinkStreamError({
909 ...(input?.threadId === undefined ? {} : { threadId: input.threadId }),
910 ...(input?.afterSequence === undefined
911 ? {}
912 : { afterSequence: input.afterSequence }),
913 cause,
914 }),
915 ),
916 ),
917 latestSequence: (input) =>
918 eventStore.latestSequence(input).pipe(
919 Effect.mapError(
920 (cause) =>
921 new EventSinkStreamError({
922 ...(input?.threadId === undefined ? {} : { threadId: input.threadId }),
923 cause,
924 }),
925 ),
926 ),
927 readByCommandId: (input) =>
928 eventStore.readByCommandId(input).pipe(
929 Stream.mapError(
930 (cause) =>
931 new EventSinkStreamError({
932 cause,
933 }),
934 ),
935 ),
936 } satisfies EventSinkV2Shape);
937 }),
938);
939
940/**
941 * Event sink layer for application compositions that already own the
942 * persistence services. Keeping the outbox instance shared with the worker is
943 * important because enqueue notifications are in-memory wakeups backed by the
944 * durable SQL queue.
945 */
946export const layerFromStores = layerBase;
947
948export const layer: Layer.Layer<
949 EventSinkV2,
950 never,
951 EventStore.EventStoreV2 | ProjectionStore.ProjectionStoreV2 | SqlClient.SqlClient
952> = layerBase.pipe(
953 Layer.provide(
954 Layer.mergeAll(
955 CommandReceiptStore.layer,
956 EffectOutbox.layer,
957 ProjectStore.layer,
958 TurnItemPositionStore.layer,
959 ),
960 ),
961);