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 readonly latestApplicationSequence: Effect.Effect<number, EventStoreV2Error>;
72 readonly publishCommitted: (events: ReadonlyArray<ApplicationStoredEvent>) => Effect.Effect<void>;
73}
74
75export class EventStoreV2 extends Context.Service<EventStoreV2, EventStoreV2Shape>()(
76 "t3/orchestration-v2/EventStore/EventStoreV2",
77) {}
78
79const layerBase: Layer.Layer<EventStoreV2, never, OrchestrationEventStore.OrchestrationEventStore> =
80 Layer.effect(
81 EventStoreV2,
82 Effect.gen(function* () {
83 const applicationEvents = yield* OrchestrationEventStore.OrchestrationEventStore;
84
85 const read: EventStoreV2Shape["read"] = (input) =>
86 applicationEvents
87 .readAgentEvents({
88 ...(input?.afterSequence === undefined ? {} : { afterSequence: input.afterSequence }),
89 ...(input?.throughSequence === undefined
90 ? {}
91 : { throughSequence: input.throughSequence }),
92 ...(input?.threadId === undefined ? {} : { threadId: input.threadId }),
93 ...(input?.eventType === undefined ? {} : { eventType: input.eventType }),
94 ...(input?.limit === undefined ? {} : { limit: input.limit }),
95 })
96 .pipe(
97 Stream.mapError(
98 (cause) =>
99 new EventStoreReadEventsError({
100 ...(input?.afterSequence === undefined
101 ? {}
102 : { afterSequence: input.afterSequence }),
103 ...(input?.threadId === undefined ? {} : { threadId: input.threadId }),
104 cause,
105 }),
106 ),
107 );
108
109 return EventStoreV2.of({
110 append: Effect.fn("EventStoreV2.append")(function (input) {
111 return applicationEvents.appendAgentEvents(input).pipe(
112 Effect.mapError(
113 (cause) =>
114 new EventStoreAppendEventsError({
115 eventCount: input.events.length,
116 cause,
117 }),
118 ),
119 );
120 }),
121 appendProjectEvent: Effect.fn("EventStoreV2.appendProjectEvent")(function (event) {
122 return applicationEvents
123 .appendProjectEvent(event)
124 .pipe(
125 Effect.mapError((cause) => new EventStoreAppendEventsError({ eventCount: 1, cause })),
126 );
127 }),
128 read,
129 readByCommandId: ({ commandId }) =>
130 applicationEvents
131 .readAgentEvents({ commandId })
132 .pipe(Stream.mapError((cause) => new EventStoreReadEventsError({ cause }))),
133 latestSequence: Effect.fn("EventStoreV2.latestSequence")(function (input) {
134 return applicationEvents.latestAgentSequence(input?.threadId).pipe(
135 Effect.mapError(
136 (cause) =>
137 new EventStoreReadEventsError({
138 ...(input?.threadId === undefined ? {} : { threadId: input.threadId }),
139 cause,
140 }),
141 ),
142 );
143 }),
144 latestApplicationSequence: applicationEvents.latestApplicationSequence.pipe(
145 Effect.mapError((cause) => new EventStoreReadEventsError({ cause })),
146 ),
147 publishCommitted: applicationEvents.publishCommitted,
148 });
149 }),
150 );
151
152export const layer: Layer.Layer<EventStoreV2, never, SqlClient.SqlClient> = layerBase.pipe(
153 Layer.provide(OrchestrationEventStore.layer),
154);
155
156export const layerFromOrchestrationEventStore = layerBase;