1import {
2 type ApplicationProjectEvent,
3 type ApplicationStoredEvent,
4 CommandId,
5 OrchestrationV2DomainEvent,
6 OrchestrationV2StoredEvent,
7 ThreadId,
8} from "@t3tools/contracts";
9import * as Context from "effect/Context";
10import * as Effect from "effect/Effect";
11import * as Layer from "effect/Layer";
12import * as Schema from "effect/Schema";
13import * as Stream from "effect/Stream";
14import type * as SqlClient from "effect/sql/SqlClient";
15import * as OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts";
16
17export class EventStoreAppendEventsError extends Schema.TaggedError<EventStoreAppendEventsError>()(
18 "EventStoreAppendEventsError",
19 {
20 eventCount: Schema.Number,
21 cause: Schema.optional(Schema.Defect()),
22 },
23) {
24 override get message(): string {
25 return `Failed to append ${this.eventCount} agent orchestration event(s).`;
26 }
27}
28
29export class EventStoreReadEventsError extends Schema.TaggedError<EventStoreReadEventsError>()(
30 "EventStoreReadEventsError",
31 {
32 afterSequence: Schema.optional(Schema.Number),
33 threadId: Schema.optional(ThreadId),
34 cause: Schema.optional(Schema.Defect()),
35 },
36) {
37 override get message(): string {
38 return this.threadId === undefined
39 ? "Failed to read agent orchestration events."
40 : `Failed to read agent orchestration events for thread ${this.threadId}.`;
41 }
42}
43
44export const EventStoreV2Error = Schema.Union([
45 EventStoreAppendEventsError,
46 EventStoreReadEventsError,
47]);
48export type EventStoreV2Error = typeof EventStoreV2Error.Type;
49
50export interface EventStoreV2Shape {
51 readonly append: (input: {
52 readonly commandId?: CommandId;
53 readonly events: ReadonlyArray<OrchestrationV2DomainEvent>;
54 }) => Effect.Effect<ReadonlyArray<OrchestrationV2StoredEvent>, EventStoreV2Error>;
55 readonly appendProjectEvent: (
56 event: OrchestrationEventStore.UnsequencedProjectEvent,
57 ) => Effect.Effect<ApplicationProjectEvent, EventStoreV2Error>;
58 readonly read: (input?: {
59 readonly afterSequence?: number;
60 readonly throughSequence?: number;
61 readonly threadId?: ThreadId;
62 readonly eventType?: OrchestrationV2DomainEvent["type"];
63 readonly limit?: number;
64 }) => Stream.Stream<OrchestrationV2StoredEvent, EventStoreV2Error>;
65 readonly readByCommandId: (input: {
66 readonly commandId: CommandId;
67 }) => Stream.Stream<OrchestrationV2StoredEvent, EventStoreV2Error>;
68 readonly latestSequence: (input?: {
69 readonly threadId?: ThreadId;
70 }) => Effect.Effect<number, EventStoreV2Error>;
71 /** Latest sequence across project and V2 thread events. */
72 readonly latestApplicationSequence: Effect.Effect<number, EventStoreV2Error>;
73 readonly publishCommitted: (events: ReadonlyArray<ApplicationStoredEvent>) => Effect.Effect<void>;
74}
75
76export class EventStoreV2 extends Context.Service<EventStoreV2, EventStoreV2Shape>()(
77 "t3/orchestration-v2/EventStore/EventStoreV2",
78) {}
79
80const layerBase: Layer.Layer<EventStoreV2, never, OrchestrationEventStore.OrchestrationEventStore> =
81 Layer.effect(
82 EventStoreV2,
83 Effect.gen(function* () {
84 const applicationEvents = yield* OrchestrationEventStore.OrchestrationEventStore;
85
86 const read: EventStoreV2Shape["read"] = (input) =>
87 applicationEvents
88 .readAgentEvents({
89 ...(input?.afterSequence === undefined ? {} : { afterSequence: input.afterSequence }),
90 ...(input?.throughSequence === undefined
91 ? {}
92 : { throughSequence: input.throughSequence }),
93 ...(input?.threadId === undefined ? {} : { threadId: input.threadId }),
94 ...(input?.eventType === undefined ? {} : { eventType: input.eventType }),
95 ...(input?.limit === undefined ? {} : { limit: input.limit }),
96 })
97 .pipe(
98 Stream.mapError(
99 (cause) =>
100 new EventStoreReadEventsError({
101 ...(input?.afterSequence === undefined
102 ? {}
103 : { afterSequence: input.afterSequence }),
104 ...(input?.threadId === undefined ? {} : { threadId: input.threadId }),
105 cause,
106 }),
107 ),
108 );
109
110 return EventStoreV2.of({
111 append: (input) =>
112 applicationEvents.appendAgentEvents(input).pipe(
113 Effect.mapError(
114 (cause) =>
115 new EventStoreAppendEventsError({
116 eventCount: input.events.length,
117 cause,
118 }),
119 ),
120 ),
121 appendProjectEvent: (event) =>
122 applicationEvents
123 .appendProjectEvent(event)
124 .pipe(
125 Effect.mapError((cause) => new EventStoreAppendEventsError({ eventCount: 1, cause })),
126 ),
127 read,
128 readByCommandId: ({ commandId }) =>
129 applicationEvents
130 .readAgentEvents({ commandId })
131 .pipe(Stream.mapError((cause) => new EventStoreReadEventsError({ cause }))),
132 latestSequence: (input) =>
133 applicationEvents.latestAgentSequence(input?.threadId).pipe(
134 Effect.mapError(
135 (cause) =>
136 new EventStoreReadEventsError({
137 ...(input?.threadId === undefined ? {} : { threadId: input.threadId }),
138 cause,
139 }),
140 ),
141 ),
142 latestApplicationSequence: applicationEvents.latestApplicationSequence.pipe(
143 Effect.mapError((cause) => new EventStoreReadEventsError({ cause })),
144 ),
145 publishCommitted: applicationEvents.publishCommitted,
146 });
147 }),
148 );
149
150export const layer: Layer.Layer<EventStoreV2, never, SqlClient.SqlClient> = layerBase.pipe(
151 Layer.provide(OrchestrationEventStore.layer),
152);
153
154export const layerFromOrchestrationEventStore = layerBase;