History

more effect: error handling

Julius Marminge committed 8 months ago

Commit 2c8e35c1 parent 9ef01eb11 files · +846 −260Browse files

Unified · Split

11 files
  1. apps/server/src/orchestration/Errors.ts +95 −0
  2. apps/server/src/orchestration/Layer.test.ts +7 −1
  3. apps/server/src/orchestration/Layer.ts +115 −119
  4. apps/server/src/orchestration/Service.ts +73 −3
  5. apps/server/src/orchestration/reducer.ts +113 −66
  6. apps/server/src/persistence/Errors.ts +126 −0
  7. apps/server/src/persistence/Layers/OrchestrationEvents.test.ts +2 −3
  8. apps/server/src/persistence/Layers/OrchestrationEvents.ts +59 −23
  9. apps/server/src/persistence/Layers/Projects.ts +154 −37
  10. apps/server/src/persistence/Services/OrchestrationEvents.ts +44 −3
  11. apps/server/src/persistence/Services/Projects.ts +58 −5
apps/server/src/orchestration/Errors.ts+95−0
@@ -0,0 +1,95 @@
1+import { ParseResult, Schema } from "effect";
2+
3+import type { OrchestrationEventRepositoryError } from "../persistence/Errors";
4+
5+export class OrchestrationCommandJsonParseError extends Schema.TaggedError<OrchestrationCommandJsonParseError>()(
6+ "OrchestrationCommandJsonParseError",
7+ {
8+ detail: Schema.String,
9+ cause: Schema.optional(Schema.Defect),
10+ },
11+) {
12+ override get message(): string {
13+ return `Invalid orchestration command JSON: ${this.detail}`;
14+ }
15+}
16+
17+export class OrchestrationCommandDecodeError extends Schema.TaggedError<OrchestrationCommandDecodeError>()(
18+ "OrchestrationCommandDecodeError",
19+ {
20+ issue: Schema.String,
21+ cause: Schema.optional(Schema.Defect),
22+ },
23+) {
24+ override get message(): string {
25+ return `Invalid orchestration command payload: ${this.issue}`;
26+ }
27+}
28+
29+export class OrchestrationReducerDecodeError extends Schema.TaggedError<OrchestrationReducerDecodeError>()(
30+ "OrchestrationReducerDecodeError",
31+ {
32+ eventType: Schema.String,
33+ issue: Schema.String,
34+ cause: Schema.optional(Schema.Defect),
35+ },
36+) {
37+ override get message(): string {
38+ return `Reducer decode failed for ${this.eventType}: ${this.issue}`;
39+ }
40+}
41+
42+export class OrchestrationListenerCallbackError extends Schema.TaggedError<OrchestrationListenerCallbackError>()(
43+ "OrchestrationListenerCallbackError",
44+ {
45+ listener: Schema.Literal("read-model", "domain-event"),
46+ detail: Schema.String,
47+ cause: Schema.optional(Schema.Defect),
48+ },
49+) {
50+ override get message(): string {
51+ return `Orchestration ${this.listener} listener failed: ${this.detail}`;
52+ }
53+}
54+
55+export type OrchestrationDispatchError =
56+ | OrchestrationEventRepositoryError
57+ | OrchestrationReducerDecodeError
58+ | OrchestrationListenerCallbackError;
59+
60+export type OrchestrationEngineError =
61+ | OrchestrationDispatchError
62+ | OrchestrationCommandJsonParseError
63+ | OrchestrationCommandDecodeError;
64+
65+export function toOrchestrationCommandDecodeError(error: ParseResult.ParseError) {
66+ return new OrchestrationCommandDecodeError({
67+ issue: ParseResult.TreeFormatter.formatErrorSync(error),
68+ cause: error,
69+ });
70+}
71+
72+export function toReducerDecodeError(eventType: string) {
73+ return (error: ParseResult.ParseError): OrchestrationReducerDecodeError =>
74+ new OrchestrationReducerDecodeError({
75+ eventType,
76+ issue: ParseResult.TreeFormatter.formatErrorSync(error),
77+ cause: error,
78+ });
79+}
80+
81+export function toOrchestrationJsonParseError(cause: unknown) {
82+ return new OrchestrationCommandJsonParseError({
83+ detail: `Failed to parse orchestration command JSON: ${cause}`,
84+ cause,
85+ });
86+}
87+
88+export function toListenerCallbackError(listener: "read-model" | "domain-event") {
89+ return (cause: unknown): OrchestrationListenerCallbackError =>
90+ new OrchestrationListenerCallbackError({
91+ listener,
92+ detail: `Failed to invoke orchestration ${listener} listener: ${cause}`,
93+ cause,
94+ });
95+}
apps/server/src/orchestration/Layer.test.ts+7−1
@@ -10,6 +10,7 @@ import { · 9 unmodified lines
1010 OrchestrationEventRepository,
1111 type OrchestrationEventRepositoryShape,
1212 } from "../persistence/Services/OrchestrationEvents";
13+import { PersistenceSqlError } from "../persistence/Errors";
1314
1415 import { Layer, ManagedRuntime } from "effect";
1516
@@ -375,7 +376,12 @@ describe("OrchestrationEngine", () => { · 359 unmodified lines
375376 append(event) {
376377 if (shouldFailFirstAppend) {
377378 shouldFailFirstAppend = false;
378− return Effect.die(new Error("append failed"));
379+ return Effect.fail(
380+ new PersistenceSqlError({
381+ operation: "test.append",
382+ detail: "append failed",
383+ }),
384+ );
379385 }
380386 const savedEvent = {
381387 ...event,
apps/server/src/orchestration/Layer.ts+115−119
@@ -4,33 +4,24 @@ import type { · 3 unmodified lines
44 OrchestrationReadModel,
55 } from "@t3tools/contracts";
66 import { OrchestrationCommandSchema } from "@t3tools/contracts";
7−import {
8− Cause,
9− Deferred,
10− Effect,
11− Either,
12− Fiber,
13− Layer,
14− PubSub,
15− Queue,
16− Runtime,
17− Schema,
18−} from "effect";
7+import { Deferred, Effect, Layer, PubSub, Queue, Schema } from "effect";
198
209 import { createLogger } from "../logger";
2110 import { OrchestrationEventRepository } from "../persistence/Services/OrchestrationEvents";
11+import {
12+ toListenerCallbackError,
13+ toOrchestrationCommandDecodeError,
14+ toOrchestrationJsonParseError,
15+ type OrchestrationDispatchError,
16+} from "./Errors";
2217 import { createEmptyReadModel, reduceEvent } from "./reducer";
2318 import { OrchestrationEngineService, type OrchestrationEngineShape } from "./Service";
2419
2520 type CommandEnvelope = {
2621 command: OrchestrationCommand;
27− result: Deferred.Deferred<{ sequence: number }, Error>;
22+ result: Deferred.Deferred<{ sequence: number }, OrchestrationDispatchError>;
2823 };
2924
30−function asError(error: unknown, fallbackMessage: string): Error {
31− return error instanceof Error ? error : new Error(fallbackMessage);
32−}
33−
3425 function mapCommandToEvent(command: OrchestrationCommand): Omit<OrchestrationEvent, "sequence"> {
3526 const eventId = crypto.randomUUID();
3627 switch (command.type) {
@@ -169,129 +160,134 @@ function mapCommandToEvent(command: OrchestrationCommand): Omit<OrchestrationEve · 132 unmodified lines
169160 }
170161 }
171162
172−const decodeUnknownCommand = Schema.decodeUnknownEither(OrchestrationCommandSchema);
163+const decodeUnknownCommand = Schema.decodeUnknown(OrchestrationCommandSchema);
173164
174−export const makeOrchestrationEngine = Effect.gen(function* () {
175− const logger = createLogger("orchestration");
176− const eventStore = yield* OrchestrationEventRepository;
177− const runtime = Runtime.defaultRuntime;
165+const makeOrchestrationEngine = () =>
166+ Effect.gen(function* () {
167+ const logger = createLogger("orchestration");
168+ const eventStore = yield* OrchestrationEventRepository;
178169
179− let readModel = createEmptyReadModel(new Date().toISOString());
170+ let readModel = createEmptyReadModel(new Date().toISOString());
180171
181− const commandQueue = yield* Queue.unbounded<CommandEnvelope>();
182− const readModelPubSub = yield* PubSub.unbounded<OrchestrationReadModel>();
183− const eventPubSub = yield* PubSub.unbounded<OrchestrationEvent>();
172+ const commandQueue = yield* Queue.unbounded<CommandEnvelope>();
173+ const readModelPubSub = yield* PubSub.unbounded<OrchestrationReadModel>();
174+ const eventPubSub = yield* PubSub.unbounded<OrchestrationEvent>();
184175
185− const readModelListeners = new Set<(snapshot: OrchestrationReadModel) => void>();
186− const domainEventListeners = new Set<(event: OrchestrationEvent) => void>();
176+ const readModelListeners = new Set<(snapshot: OrchestrationReadModel) => void>();
177+ const domainEventListeners = new Set<(event: OrchestrationEvent) => void>();
187178
188− const processEnvelope = (envelope: CommandEnvelope): Effect.Effect<void> =>
189− Effect.gen(function* () {
190− const eventBase = mapCommandToEvent(envelope.command);
191− const savedEvent = yield* eventStore.append(eventBase);
192− readModel = yield* reduceEvent(readModel, savedEvent);
179+ const notifyReadModelListeners = (snapshot: OrchestrationReadModel) =>
180+ Effect.forEach(readModelListeners, (listener) =>
181+ Effect.try({
182+ try: () => listener(snapshot),
183+ catch: toListenerCallbackError("read-model"),
184+ }),
185+ ).pipe(Effect.asVoid);
193186
194− const snapshot = readModel;
195− yield* Effect.all([
196− PubSub.publish(eventPubSub, savedEvent),
197− PubSub.publish(readModelPubSub, snapshot),
198− ]);
187+ const notifyDomainEventListeners = (event: OrchestrationEvent) =>
188+ Effect.forEach(domainEventListeners, (listener) =>
189+ Effect.try({
190+ try: () => listener(event),
191+ catch: toListenerCallbackError("domain-event"),
192+ }),
193+ ).pipe(Effect.asVoid);
199194
200− for (const listener of domainEventListeners) {
201− listener(savedEvent);
202− }
203− for (const listener of readModelListeners) {
204− listener(snapshot);
205− }
195+ const processEnvelope = (envelope: CommandEnvelope): Effect.Effect<void> =>
196+ Effect.gen(function* () {
197+ const eventBase = mapCommandToEvent(envelope.command);
198+ const savedEvent = yield* eventStore.append(eventBase);
199+ readModel = yield* reduceEvent(readModel, savedEvent);
206200
207− yield* Deferred.succeed(envelope.result, { sequence: savedEvent.sequence });
208− }).pipe(
209− Effect.catchAllCause((cause) =>
210− Deferred.fail(
211− envelope.result,
212− asError(Cause.squash(cause), "Unknown command processing error"),
213− ).pipe(Effect.asVoid),
214− ),
215− );
201+ const snapshot = readModel;
202+ yield* Effect.all([
203+ PubSub.publish(eventPubSub, savedEvent),
204+ PubSub.publish(readModelPubSub, snapshot),
205+ ]);
216206
217− const bootstrapReadModel: Effect.Effect<void, unknown> = Effect.gen(function* () {
218− const existingEvents = yield* eventStore.readAll();
219− for (const event of existingEvents) {
220− readModel = yield* reduceEvent(readModel, event);
221− }
222− });
207+ yield* notifyDomainEventListeners(savedEvent);
208+ yield* notifyReadModelListeners(snapshot);
223209
224− yield* bootstrapReadModel;
210+ yield* Deferred.succeed(envelope.result, { sequence: savedEvent.sequence });
211+ }).pipe(
212+ Effect.catchAll((error) => Deferred.fail(envelope.result, error).pipe(Effect.asVoid)),
213+ );
225214
226− const worker = Effect.forever(Queue.take(commandQueue).pipe(Effect.flatMap(processEnvelope)));
227− const workerFiber: Fiber.RuntimeFiber<void, unknown> = Runtime.runFork(runtime)(worker);
228− logger.info("orchestration engine started", {
229− sequence: readModel.sequence,
230− });
215+ const bootstrapReadModel: Effect.Effect<void, OrchestrationDispatchError> = Effect.gen(
216+ function* () {
217+ const existingEvents = yield* eventStore.readAll();
218+ for (const event of existingEvents) {
219+ readModel = yield* reduceEvent(readModel, event);
220+ }
221+ },
222+ );
223+
224+ yield* bootstrapReadModel;
231225
232− yield* Effect.addFinalizer(() => Fiber.interrupt(workerFiber).pipe(Effect.orDie));
226+ const worker = Effect.forever(Queue.take(commandQueue).pipe(Effect.flatMap(processEnvelope)));
227+ yield* Effect.forkScoped(worker);
228+ logger.info("orchestration engine started", {
229+ sequence: readModel.sequence,
230+ });
233231
234− const getSnapshot: OrchestrationEngineShape["getSnapshot"] = () =>
235− Effect.sync((): OrchestrationReadModel => readModel);
232+ const getSnapshot: OrchestrationEngineShape["getSnapshot"] = () =>
233+ Effect.sync((): OrchestrationReadModel => readModel);
236234
237− const replayEvents: OrchestrationEngineShape["replayEvents"] = (fromSequenceExclusive) =>
238− eventStore.readFromSequence(fromSequenceExclusive);
235+ const replayEvents: OrchestrationEngineShape["replayEvents"] = (fromSequenceExclusive) =>
236+ eventStore.readFromSequence(fromSequenceExclusive);
239237
240− const dispatch: OrchestrationEngineShape["dispatch"] = (command) =>
241− Effect.gen(function* () {
242− const result = yield* Deferred.make<{ sequence: number }, Error>();
243− yield* Queue.offer(commandQueue, { command, result });
244− return yield* Deferred.await(result);
245− }).pipe(Effect.mapError((error) => asError(error, "Queue offer failed")));
238+ const dispatch: OrchestrationEngineShape["dispatch"] = (command) =>
239+ Effect.gen(function* () {
240+ const result = yield* Deferred.make<{ sequence: number }, OrchestrationDispatchError>();
241+ yield* Queue.offer(commandQueue, { command, result });
242+ return yield* Deferred.await(result);
243+ });
246244
247− const dispatchUnknown: OrchestrationEngineShape["dispatchUnknown"] = (command) =>
248− Effect.gen(function* () {
249− let payload: unknown = command;
250− if (typeof command === "string") {
251− try {
252− payload = JSON.parse(command) as unknown;
253− } catch {
254− return yield* Effect.fail(
255− new Error("Invalid orchestration command: payload is not valid JSON"),
256− );
257− }
258− }
259− const decoded = decodeUnknownCommand(payload);
260− if (Either.isLeft(decoded)) {
261− return yield* Effect.fail(
262− new Error(`Invalid orchestration command: ${decoded.left.toString()}`),
245+ const dispatchUnknown: OrchestrationEngineShape["dispatchUnknown"] = (command) =>
246+ Effect.gen(function* () {
247+ const payload =
248+ typeof command === "string"
249+ ? yield* Effect.try({
250+ try: () => JSON.parse(command) as unknown,
251+ catch: toOrchestrationJsonParseError,
252+ })
253+ : command;
254+
255+ const decoded = yield* decodeUnknownCommand(payload).pipe(
256+ Effect.mapError(toOrchestrationCommandDecodeError),
263257 );
264− }
265− return yield* dispatch(decoded.right);
266− });
267258
268− const subscribeToReadModel: OrchestrationEngineShape["subscribeToReadModel"] = (callback) =>
269− Effect.sync(() => {
270− readModelListeners.add(callback);
271− return () => {
272− readModelListeners.delete(callback);
273− };
274− });
259+ return yield* dispatch(decoded);
260+ });
275261
276− const subscribeToDomainEvents: OrchestrationEngineShape["subscribeToDomainEvents"] = (callback) =>
277− Effect.sync(() => {
278− domainEventListeners.add(callback);
279− return () => {
280− domainEventListeners.delete(callback);
281− };
282− });
262+ const subscribeToReadModel: OrchestrationEngineShape["subscribeToReadModel"] = (callback) =>
263+ Effect.sync(() => {
264+ readModelListeners.add(callback);
265+ return () => {
266+ readModelListeners.delete(callback);
267+ };
268+ });
283269
284− return {
285− getSnapshot,
286− replayEvents,
287− dispatchUnknown,
288− dispatch,
289− subscribeToReadModel,
290− subscribeToDomainEvents,
291− } satisfies OrchestrationEngineShape;
292−});
270+ const subscribeToDomainEvents: OrchestrationEngineShape["subscribeToDomainEvents"] = (
271+ callback,
272+ ) =>
273+ Effect.sync(() => {
274+ domainEventListeners.add(callback);
275+ return () => {
276+ domainEventListeners.delete(callback);
277+ };
278+ });
279+
280+ return {
281+ getSnapshot,
282+ replayEvents,
283+ dispatchUnknown,
284+ dispatch,
285+ subscribeToReadModel,
286+ subscribeToDomainEvents,
287+ } satisfies OrchestrationEngineShape;
288+ });
293289
294290 export const OrchestrationEngineLive = Layer.scoped(
295291 OrchestrationEngineService,
296− makeOrchestrationEngine,
292+ makeOrchestrationEngine(),
297293 );
apps/server/src/orchestration/Service.ts+73−3
@@ -1,3 +1,11 @@
1+/**
2+ * OrchestrationEngineService - Service interface for orchestration command handling.
3+ *
4+ * Uses Effect `Context.Tag` for dependency injection. Command dispatch,
5+ * replay, and unknown-input decoding all return typed domain errors.
6+ *
7+ * @module OrchestrationEngineService
8+ */
19 import type {
210 OrchestrationCommand,
311 OrchestrationEvent,
@@ -6,19 +14,81 @@ import type { · 2 unmodified lines
614 import { Context } from "effect";
715 import type { Effect } from "effect";
816
17+import type { OrchestrationDispatchError, OrchestrationEngineError } from "./Errors";
18+import type { OrchestrationEventRepositoryError } from "../persistence/Errors";
19+
920 export interface OrchestrationEngineShape {
21+ /**
22+ * Read the current in-memory orchestration snapshot.
23+ *
24+ * @returns Effect containing the latest read model.
25+ */
1026 readonly getSnapshot: () => Effect.Effect<OrchestrationReadModel>;
11− readonly replayEvents: (fromSequenceExclusive: number) => Effect.Effect<OrchestrationEvent[]>;
12− readonly dispatchUnknown: (command: unknown) => Effect.Effect<{ sequence: number }, Error>;
13− readonly dispatch: (command: OrchestrationCommand) => Effect.Effect<{ sequence: number }, Error>;
27+
28+ /**
29+ * Replay persisted orchestration events from an exclusive sequence cursor.
30+ *
31+ * @param fromSequenceExclusive - Sequence cursor (exclusive).
32+ * @returns Effect containing ordered events.
33+ */
34+ readonly replayEvents: (
35+ fromSequenceExclusive: number,
36+ ) => Effect.Effect<OrchestrationEvent[], OrchestrationEventRepositoryError>;
37+
38+ /**
39+ * Validate and dispatch an unknown command payload.
40+ *
41+ * Accepts either parsed values or raw JSON strings.
42+ *
43+ * @param command - Unknown command payload.
44+ * @returns Effect containing the sequence of the persisted event.
45+ */
46+ readonly dispatchUnknown: (
47+ command: unknown,
48+ ) => Effect.Effect<{ sequence: number }, OrchestrationEngineError>;
49+
50+ /**
51+ * Dispatch a validated orchestration command.
52+ *
53+ * @param command - Valid orchestration command.
54+ * @returns Effect containing the sequence of the persisted event.
55+ */
56+ readonly dispatch: (
57+ command: OrchestrationCommand,
58+ ) => Effect.Effect<{ sequence: number }, OrchestrationDispatchError>;
59+
60+ /**
61+ * Subscribe to read model updates.
62+ *
63+ * @param callback - Handler invoked on every read model update.
64+ * @returns Effect containing an unsubscribe function.
65+ */
1466 readonly subscribeToReadModel: (
1567 callback: (snapshot: OrchestrationReadModel) => void,
1668 ) => Effect.Effect<() => void>;
69+
70+ /**
71+ * Subscribe to domain event fan-out.
72+ *
73+ * @param callback - Handler invoked for every persisted domain event.
74+ * @returns Effect containing an unsubscribe function.
75+ */
1776 readonly subscribeToDomainEvents: (
1877 callback: (event: OrchestrationEvent) => void,
1978 ) => Effect.Effect<() => void>;
2079 }
2180
81+/**
82+ * OrchestrationEngineService - Context tag for orchestration engine access.
83+ *
84+ * @example
85+ * ```ts
86+ * const program = Effect.gen(function* () {
87+ * const engine = yield* OrchestrationEngineService
88+ * return yield* engine.getSnapshot()
89+ * })
90+ * ```
91+ */
2292 export class OrchestrationEngineService extends Context.Tag("orchestration/Engine")<
2393 OrchestrationEngineService,
2494 OrchestrationEngineShape
apps/server/src/orchestration/reducer.ts+113−66
@@ -9,7 +9,6 @@ import type { · 8 unmodified lines
99 import {
1010 OrchestrationGitReadModelSchema,
1111 OrchestrationMessageSchema,
12− OrchestrationReadModelSchema,
1312 OrchestrationSessionSchema,
1413 OrchestrationThreadSchema,
1514 OrchestrationTurnDiffFileSchema,
@@ -17,6 +16,8 @@ import { · 1 unmodified line
1716 } from "@t3tools/contracts";
1817 import { Effect, Schema } from "effect";
1918
19+import { toReducerDecodeError, type OrchestrationReducerDecodeError } from "./Errors";
20+
2021 type ThreadPatch = Partial<Omit<OrchestrationThread, "id" | "projectId">>;
2122
2223 function updateThread(
@@ -27,12 +28,16 @@ function updateThread( · 4 unmodified lines
2728 return threads.map((thread) => (thread.id === threadId ? { ...thread, ...patch } : thread));
2829 }
2930
30−const decodeReadModelSync = Schema.decodeUnknownSync(OrchestrationReadModelSchema);
31−const decodeMessage = Schema.decodeUnknown(OrchestrationMessageSchema);
32−const decodeSession = Schema.decodeUnknown(OrchestrationSessionSchema);
33−const decodeThread = Schema.decodeUnknown(OrchestrationThreadSchema);
34−const decodeGitReadModel = Schema.decodeUnknown(OrchestrationGitReadModelSchema);
35−const decodeTurnDiffSummary = Schema.decodeUnknown(OrchestrationTurnDiffSummarySchema);
31+function decodeForEvent<A>(
32+ schema: Schema.Schema<A>,
33+ value: unknown,
34+ eventType: string,
35+ field: string,
36+): Effect.Effect<A, OrchestrationReducerDecodeError> {
37+ return Schema.decodeUnknown(schema)(value).pipe(
38+ Effect.mapError(toReducerDecodeError(`${eventType}:${field}`)),
39+ );
40+}
3641
3742 const ThreadCreatedPayloadSchema = Schema.Struct({
3843 id: Schema.String,
@@ -44,7 +49,6 @@ const ThreadCreatedPayloadSchema = Schema.Struct({ · 5 unmodified lines
4449 createdAt: Schema.String,
4550 updatedAt: Schema.String,
4651 });
47−const decodeThreadCreatedPayload = Schema.decodeUnknown(ThreadCreatedPayloadSchema);
4852
4953 const ThreadMetaUpdatedPayloadSchema = Schema.Struct({
5054 threadId: Schema.String,
@@ -54,7 +58,6 @@ const ThreadMetaUpdatedPayloadSchema = Schema.Struct({ · 3 unmodified lines
5458 worktreePath: Schema.optional(Schema.NullOr(Schema.String)),
5559 updatedAt: Schema.String,
5660 });
57−const decodeThreadMetaUpdatedPayload = Schema.decodeUnknown(ThreadMetaUpdatedPayloadSchema);
5861
5962 const MessageSentPayloadSchema = Schema.Struct({
6063 id: Schema.String,
@@ -64,13 +67,11 @@ const MessageSentPayloadSchema = Schema.Struct({ · 3 unmodified lines
6467 createdAt: Schema.String,
6568 streaming: Schema.Boolean,
6669 });
67−const decodeMessageSentPayload = Schema.decodeUnknown(MessageSentPayloadSchema);
6870
6971 const ThreadSessionSetPayloadSchema = Schema.Struct({
7072 threadId: Schema.String,
7173 session: OrchestrationSessionSchema,
7274 });
73−const decodeThreadSessionSetPayload = Schema.decodeUnknown(ThreadSessionSetPayloadSchema);
7475
7576 const ThreadTurnDiffCompletedPayloadSchema = Schema.Struct({
7677 threadId: Schema.String,
@@ -81,30 +82,26 @@ const ThreadTurnDiffCompletedPayloadSchema = Schema.Struct({ · 4 unmodified lines
8182 assistantMessageId: Schema.optional(Schema.String),
8283 checkpointTurnCount: Schema.optional(Schema.Number),
8384 });
84−const decodeThreadTurnDiffCompletedPayload = Schema.decodeUnknown(
85− ThreadTurnDiffCompletedPayloadSchema,
86−);
8785
8886 const ThreadRevertedPayloadSchema = Schema.Struct({
8987 threadId: Schema.String,
9088 turnCount: Schema.Number,
9189 messageCount: Schema.Number,
9290 });
93−const decodeThreadRevertedPayload = Schema.decodeUnknown(ThreadRevertedPayloadSchema);
9491
9592 export function createEmptyReadModel(nowIso: string): OrchestrationReadModel {
96− return decodeReadModelSync({
93+ return {
9794 sequence: 0,
9895 threads: [],
9996 gitByProjectId: {},
10097 updatedAt: nowIso,
101− });
98+ };
10299 }
103100
104101 export function reduceEvent(
105102 model: OrchestrationReadModel,
106103 event: OrchestrationEvent,
107−): Effect.Effect<OrchestrationReadModel, unknown> {
104+): Effect.Effect<OrchestrationReadModel, OrchestrationReducerDecodeError> {
108105 const nextBase: OrchestrationReadModel = {
109106 ...model,
110107 sequence: event.sequence,
@@ -114,25 +111,35 @@ export function reduceEvent( · 3 unmodified lines
114111 switch (event.type) {
115112 case "thread.created":
116113 return Effect.gen(function* () {
117− const payload = yield* decodeThreadCreatedPayload(event.payload);
118− const thread: OrchestrationThread = yield* decodeThread({
119− id: payload.id,
120− projectId: payload.projectId,
121− title: payload.title,
122− model: payload.model,
123− branch: payload.branch,
124− worktreePath: payload.worktreePath,
125− createdAt: payload.createdAt,
126− updatedAt: payload.updatedAt,
127− latestTurnId: null,
128− latestTurnStartedAt: null,
129− latestTurnCompletedAt: null,
130− latestTurnDurationMs: null,
131− messages: [],
132− session: null,
133− turnDiffSummaries: [],
134− error: null,
135− });
114+ const payload = yield* decodeForEvent(
115+ ThreadCreatedPayloadSchema,
116+ event.payload,
117+ event.type,
118+ "payload",
119+ );
120+ const thread: OrchestrationThread = yield* decodeForEvent(
121+ OrchestrationThreadSchema,
122+ {
123+ id: payload.id,
124+ projectId: payload.projectId,
125+ title: payload.title,
126+ model: payload.model,
127+ branch: payload.branch,
128+ worktreePath: payload.worktreePath,
129+ createdAt: payload.createdAt,
130+ updatedAt: payload.updatedAt,
131+ latestTurnId: null,
132+ latestTurnStartedAt: null,
133+ latestTurnCompletedAt: null,
134+ latestTurnDurationMs: null,
135+ messages: [],
136+ session: null,
137+ turnDiffSummaries: [],
138+ error: null,
139+ },
140+ event.type,
141+ "thread",
142+ );
136143 const existing = nextBase.threads.find((entry) => entry.id === thread.id);
137144 return {
138145 ...nextBase,
@@ -147,7 +154,12 @@ export function reduceEvent( · 8 unmodified lines
147154 threads: nextBase.threads.filter((thread) => thread.id !== event.aggregateId),
148155 });
149156 case "thread.meta-updated":
150− return decodeThreadMetaUpdatedPayload(event.payload).pipe(
157+ return decodeForEvent(
158+ ThreadMetaUpdatedPayloadSchema,
159+ event.payload,
160+ event.type,
161+ "payload",
162+ ).pipe(
151163 Effect.map((payload) => {
152164 const thread = nextBase.threads.find((entry) => entry.id === payload.threadId);
153165 if (!thread) {
@@ -167,19 +179,29 @@ export function reduceEvent( · 13 unmodified lines
167179 );
168180 case "message.sent":
169181 return Effect.gen(function* () {
170− const payload = yield* decodeMessageSentPayload(event.payload);
182+ const payload = yield* decodeForEvent(
183+ MessageSentPayloadSchema,
184+ event.payload,
185+ event.type,
186+ "payload",
187+ );
171188 const targetThread = nextBase.threads.find((thread) => thread.id === payload.threadId);
172189 if (!targetThread) {
173190 return nextBase;
174191 }
175192
176− const message: OrchestrationMessage = yield* decodeMessage({
177− id: payload.id,
178− role: payload.role,
179− text: payload.text,
180− createdAt: payload.createdAt,
181− streaming: payload.streaming,
182− });
193+ const message: OrchestrationMessage = yield* decodeForEvent(
194+ OrchestrationMessageSchema,
195+ {
196+ id: payload.id,
197+ role: payload.role,
198+ text: payload.text,
199+ createdAt: payload.createdAt,
200+ streaming: payload.streaming,
201+ },
202+ event.type,
203+ "message",
204+ );
183205 const existingMessage = targetThread.messages.find((entry) => entry.id === message.id);
184206 const nextMessages = existingMessage
185207 ? targetThread.messages.map((entry) =>
@@ -208,16 +230,26 @@ export function reduceEvent( · 22 unmodified lines
208230 });
209231 case "thread.session-set":
210232 return Effect.gen(function* () {
211− const payload = yield* decodeThreadSessionSetPayload(event.payload);
233+ const payload = yield* decodeForEvent(
234+ ThreadSessionSetPayloadSchema,
235+ event.payload,
236+ event.type,
237+ "payload",
238+ );
212239 const thread = nextBase.threads.find((entry) => entry.id === payload.threadId);
213240 if (!thread) {
214241 return nextBase;
215242 }
216243
217− const session: OrchestrationSession = yield* decodeSession({
218− ...payload.session,
219− threadId: payload.threadId,
220− });
244+ const session: OrchestrationSession = yield* decodeForEvent(
245+ OrchestrationSessionSchema,
246+ {
247+ ...payload.session,
248+ threadId: payload.threadId,
249+ },
250+ event.type,
251+ "session",
252+ );
221253
222254 return {
223255 ...nextBase,
@@ -230,24 +262,34 @@ export function reduceEvent( · 6 unmodified lines
230262 });
231263 case "thread.turn-diff-completed":
232264 return Effect.gen(function* () {
233− const payload = yield* decodeThreadTurnDiffCompletedPayload(event.payload);
265+ const payload = yield* decodeForEvent(
266+ ThreadTurnDiffCompletedPayloadSchema,
267+ event.payload,
268+ event.type,
269+ "payload",
270+ );
234271 const thread = nextBase.threads.find((entry) => entry.id === payload.threadId);
235272 if (!thread) {
236273 return nextBase;
237274 }
238275
239− const nextSummary = yield* decodeTurnDiffSummary({
240− turnId: payload.turnId,
241− completedAt: payload.completedAt,
242− ...(payload.status !== undefined ? { status: payload.status } : {}),
243− files: payload.files,
244− ...(payload.assistantMessageId !== undefined
245− ? { assistantMessageId: payload.assistantMessageId }
246− : {}),
247− ...(payload.checkpointTurnCount !== undefined
248− ? { checkpointTurnCount: payload.checkpointTurnCount }
249− : {}),
250− });
276+ const nextSummary = yield* decodeForEvent(
277+ OrchestrationTurnDiffSummarySchema,
278+ {
279+ turnId: payload.turnId,
280+ completedAt: payload.completedAt,
281+ ...(payload.status !== undefined ? { status: payload.status } : {}),
282+ files: payload.files,
283+ ...(payload.assistantMessageId !== undefined
284+ ? { assistantMessageId: payload.assistantMessageId }
285+ : {}),
286+ ...(payload.checkpointTurnCount !== undefined
287+ ? { checkpointTurnCount: payload.checkpointTurnCount }
288+ : {}),
289+ },
290+ event.type,
291+ "turnDiffSummary",
292+ );
251293 const turnDiffSummaries = [
252294 ...thread.turnDiffSummaries.filter((summary) => summary.turnId !== payload.turnId),
253295 nextSummary,
@@ -264,7 +306,7 @@ export function reduceEvent( · 10 unmodified lines
264306 };
265307 });
266308 case "thread.reverted":
267− return decodeThreadRevertedPayload(event.payload).pipe(
309+ return decodeForEvent(ThreadRevertedPayloadSchema, event.payload, event.type, "payload").pipe(
268310 Effect.map((payload) => {
269311 const thread = nextBase.threads.find((entry) => entry.id === payload.threadId);
270312 if (!thread) {
@@ -311,7 +353,12 @@ export function reduceEvent( · 40 unmodified lines
311353 }),
312354 );
313355 case "git.read-model-upsert":
314− return decodeGitReadModel(event.payload).pipe(
356+ return decodeForEvent(
357+ OrchestrationGitReadModelSchema,
358+ event.payload,
359+ event.type,
360+ "payload",
361+ ).pipe(
315362 Effect.map((gitReadModel: OrchestrationGitReadModel) => ({
316363 ...nextBase,
317364 gitByProjectId: {
apps/server/src/persistence/Errors.ts+126−0
@@ -0,0 +1,126 @@
1+import { ParseResult, Schema } from "effect";
2+
3+export class PersistenceSqlError extends Schema.TaggedError<PersistenceSqlError>()(
4+ "PersistenceSqlError",
5+ {
6+ operation: Schema.String,
7+ detail: Schema.String,
8+ cause: Schema.optional(Schema.Defect),
9+ },
10+) {
11+ override get message(): string {
12+ return `SQL error in ${this.operation}: ${this.detail}`;
13+ }
14+}
15+
16+export class PersistenceDecodeError extends Schema.TaggedError<PersistenceDecodeError>()(
17+ "PersistenceDecodeError",
18+ {
19+ operation: Schema.String,
20+ issue: Schema.String,
21+ cause: Schema.optional(Schema.Defect),
22+ },
23+) {
24+ override get message(): string {
25+ return `Decode error in ${this.operation}: ${this.issue}`;
26+ }
27+}
28+
29+export class PersistenceSerializationError extends Schema.TaggedError<PersistenceSerializationError>()(
30+ "PersistenceSerializationError",
31+ {
32+ operation: Schema.String,
33+ detail: Schema.String,
34+ cause: Schema.optional(Schema.Defect),
35+ },
36+) {
37+ override get message(): string {
38+ return `Serialization error in ${this.operation}: ${this.detail}`;
39+ }
40+}
41+
42+export class ProjectValidationError extends Schema.TaggedError<ProjectValidationError>()(
43+ "ProjectValidationError",
44+ {
45+ operation: Schema.String,
46+ issue: Schema.String,
47+ cause: Schema.optional(Schema.Defect),
48+ },
49+) {
50+ override get message(): string {
51+ return `Project validation failed in ${this.operation}: ${this.issue}`;
52+ }
53+}
54+
55+export class ProjectPathMissingError extends Schema.TaggedError<ProjectPathMissingError>()(
56+ "ProjectPathMissingError",
57+ {
58+ cwd: Schema.String,
59+ cause: Schema.optional(Schema.Defect),
60+ },
61+) {
62+ override get message(): string {
63+ return `Project path does not exist: ${this.cwd}`;
64+ }
65+}
66+
67+export class ProjectNotFoundError extends Schema.TaggedError<ProjectNotFoundError>()(
68+ "ProjectNotFoundError",
69+ {
70+ projectId: Schema.String,
71+ cause: Schema.optional(Schema.Defect),
72+ },
73+) {
74+ override get message(): string {
75+ return `Project not found: ${this.projectId}`;
76+ }
77+}
78+
79+export type ProjectRepositoryError =
80+ | PersistenceSqlError
81+ | PersistenceDecodeError
82+ | PersistenceSerializationError
83+ | ProjectValidationError
84+ | ProjectPathMissingError
85+ | ProjectNotFoundError;
86+
87+export type OrchestrationEventRepositoryError =
88+ | PersistenceSqlError
89+ | PersistenceDecodeError
90+ | PersistenceSerializationError;
91+
92+export function toPersistenceSqlError(operation: string) {
93+ return (cause: unknown): PersistenceSqlError =>
94+ new PersistenceSqlError({
95+ operation,
96+ detail: `Failed to execute ${operation}`,
97+ cause,
98+ });
99+}
100+
101+export function toPersistenceDecodeError(operation: string) {
102+ return (error: ParseResult.ParseError): PersistenceDecodeError =>
103+ new PersistenceDecodeError({
104+ operation,
105+ issue: ParseResult.TreeFormatter.formatErrorSync(error),
106+ cause: error,
107+ });
108+}
109+
110+export function toPersistenceDecodeCauseError(operation: string) {
111+ return (cause: unknown): PersistenceDecodeError =>
112+ new PersistenceDecodeError({
113+ operation,
114+ issue: `Failed to execute ${operation}`,
115+ cause,
116+ });
117+}
118+
119+export function toPersistenceSerializationError(operation: string) {
120+ return (cause: unknown): PersistenceSerializationError =>
121+ new PersistenceSerializationError({
122+ operation,
123+ detail: `Failed to execute ${operation}`,
124+ cause,
125+ });
126+}
apps/server/src/persistence/Layers/OrchestrationEvents.test.ts+2−3
@@ -80,9 +80,8 @@ describe("OrchestrationEventRepository", () => { · 79 unmodified lines
8080 )
8181 .get() as { name: string } | undefined;
8282 expect(tableRow?.name).toBeDefined();
83− if (!tableRow) {
84− throw new Error("Expected migrations table to exist");
85− }
83+ expect(tableRow).toBeDefined();
84+ if (!tableRow) return;
8685
8786 const migrationRows = db
8887 .prepare(
apps/server/src/persistence/Layers/OrchestrationEvents.ts+59−23
@@ -4,13 +4,20 @@ import * as SqlClient from "@effect/sql/SqlClient"; · 3 unmodified lines
44 import * as SqlSchema from "@effect/sql/SqlSchema";
55 import { Effect, Layer, Schema } from "effect";
66
7+import {
8+ toPersistenceDecodeCauseError,
9+ toPersistenceDecodeError,
10+ toPersistenceSerializationError,
11+ toPersistenceSqlError,
12+ type OrchestrationEventRepositoryError,
13+} from "../Errors";
714 import { makeSqlitePersistenceLive } from "./Sqlite";
815 import {
916 OrchestrationEventRepository,
1017 type OrchestrationEventRepositoryShape,
1118 } from "../Services/OrchestrationEvents";
1219
13−const decodeEvent = Schema.decodeUnknownSync(OrchestrationEventSchema);
20+const decodeEvent = Schema.decodeUnknown(OrchestrationEventSchema);
1421
1522 const EventRowSchema = Schema.Struct({
1623 sequence: Schema.Number,
@@ -40,17 +47,27 @@ const ReadFromSequenceRequestSchema = Schema.Struct({ · 23 unmodified lines
4047 limit: Schema.Number,
4148 });
4249
43−function eventRowToOrchestrationEvent(row: EventRow): OrchestrationEvent {
44− return decodeEvent({
45− sequence: row.sequence,
46− eventId: row.eventId,
47− type: row.type,
48− aggregateType: row.aggregateType,
49− aggregateId: row.aggregateId,
50− occurredAt: row.occurredAt,
51− commandId: row.commandId,
52− payload: JSON.parse(row.payloadJson) as unknown,
53− });
50+function eventRowToOrchestrationEvent(
51+ row: EventRow,
52+ operation: string,
53+): Effect.Effect<OrchestrationEvent, OrchestrationEventRepositoryError> {
54+ return Effect.try({
55+ try: () => JSON.parse(row.payloadJson) as unknown,
56+ catch: toPersistenceDecodeCauseError(`${operation}:parsePayloadJson`),
57+ }).pipe(
58+ Effect.flatMap((payload) =>
59+ decodeEvent({
60+ sequence: row.sequence,
61+ eventId: row.eventId,
62+ type: row.type,
63+ aggregateType: row.aggregateType,
64+ aggregateId: row.aggregateId,
65+ occurredAt: row.occurredAt,
66+ commandId: row.commandId,
67+ payload,
68+ }).pipe(Effect.mapError(toPersistenceDecodeError(`${operation}:decodeEvent`))),
69+ ),
70+ );
5471 }
5572
5673 const makeRepository = Effect.gen(function* () {
@@ -113,23 +130,42 @@ const makeRepository = Effect.gen(function* () { · 56 unmodified lines
113130 });
114131
115132 const append: OrchestrationEventRepositoryShape["append"] = (event) =>
116− appendEventRow({
117− eventId: event.eventId,
118− type: event.type,
119− aggregateType: event.aggregateType,
120− aggregateId: event.aggregateId,
121− occurredAt: event.occurredAt,
122− commandId: event.commandId,
123− payloadJson: JSON.stringify(event.payload),
124− }).pipe(Effect.map(eventRowToOrchestrationEvent), Effect.orDie);
133+ Effect.try({
134+ try: () => JSON.stringify(event.payload),
135+ catch: toPersistenceSerializationError("OrchestrationEventRepository.append:encodePayload"),
136+ }).pipe(
137+ Effect.flatMap((payloadJson) =>
138+ appendEventRow({
139+ eventId: event.eventId,
140+ type: event.type,
141+ aggregateType: event.aggregateType,
142+ aggregateId: event.aggregateId,
143+ occurredAt: event.occurredAt,
144+ commandId: event.commandId,
145+ payloadJson,
146+ }).pipe(
147+ Effect.mapError(toPersistenceSqlError("OrchestrationEventRepository.append:insert")),
148+ Effect.flatMap((row) =>
149+ eventRowToOrchestrationEvent(row, "OrchestrationEventRepository.append:rowToEvent"),
150+ ),
151+ ),
152+ ),
153+ );
125154
126155 const readFromSequence: OrchestrationEventRepositoryShape["readFromSequence"] = (
127156 sequenceExclusive,
128157 limit = 1_000,
129158 ) =>
130159 readEventRowsFromSequence({ sequenceExclusive, limit }).pipe(
131− Effect.map((rows) => rows.map(eventRowToOrchestrationEvent)),
132− Effect.orDie,
160+ Effect.mapError(toPersistenceSqlError("OrchestrationEventRepository.readFromSequence:query")),
161+ Effect.flatMap((rows) =>
162+ Effect.forEach(rows, (row) =>
163+ eventRowToOrchestrationEvent(
164+ row,
165+ "OrchestrationEventRepository.readFromSequence:rowToEvent",
166+ ),
167+ ),
168+ ),
133169 );
134170
135171 return {
apps/server/src/persistence/Layers/Projects.ts+154−37
@@ -8,6 +8,7 @@ import type { · 7 unmodified lines
88 ProjectListResult,
99 ProjectRecord,
1010 ProjectScript,
11+ ProjectUpdateScriptsInput,
1112 ProjectUpdateScriptsResult,
1213 } from "@t3tools/contracts";
1314 import {
@@ -22,6 +23,16 @@ import * as SqlClient from "@effect/sql/SqlClient"; · 8 unmodified lines
2223 import * as SqlSchema from "@effect/sql/SqlSchema";
2324 import { Effect, Layer, Option, Schema } from "effect";
2425
26+import {
27+ PersistenceDecodeError,
28+ ProjectNotFoundError,
29+ ProjectPathMissingError,
30+ ProjectValidationError,
31+ toPersistenceDecodeCauseError,
32+ toPersistenceSerializationError,
33+ toPersistenceSqlError,
34+ type ProjectRepositoryError,
35+} from "../Errors";
2536 import { ProjectRepository, type ProjectRepositoryShape } from "../Services/Projects";
2637
2738 const ProjectRowSchema = Schema.Struct({
@@ -98,25 +109,101 @@ function inferProjectName(cwd: string): string { · 70 unmodified lines
98109 return name.length > 0 ? name : "project";
99110 }
100111
101−function parseScriptsJson(raw: string): ProjectScript[] {
102− try {
103− const parsed = JSON.parse(raw) as unknown;
104− const scripts = projectScriptsSchema.parse(parsed);
105− return normalizeProjectScripts(scripts);
106− } catch {
107− return [];
112+function parseProjectScripts(
113+ value: unknown,
114+ operation: string,
115+): Effect.Effect<ProjectScript[], PersistenceDecodeError> {
116+ const parsed = projectScriptsSchema.safeParse(value);
117+ if (!parsed.success) {
118+ return Effect.fail(
119+ new PersistenceDecodeError({
120+ operation,
121+ issue: parsed.error.message,
122+ cause: (parsed.error as { cause?: unknown }).cause,
123+ }),
124+ );
108125 }
126+ return Effect.succeed(normalizeProjectScripts(parsed.data));
109127 }
110128
111−function rowToProjectRecord(row: ProjectRow): ProjectRecord {
112− return projectRecordSchema.parse({
113− id: row.id,
114− cwd: row.cwd,
115− name: row.name,
116− scripts: parseScriptsJson(row.scriptsJson),
117− createdAt: row.createdAt,
118− updatedAt: row.updatedAt,
119− });
129+function decodeScriptsJson(
130+ raw: string,
131+ operation: string,
132+): Effect.Effect<ProjectScript[], ProjectRepositoryError> {
133+ return Effect.try({
134+ try: () => JSON.parse(raw) as unknown,
135+ catch: toPersistenceDecodeCauseError(`${operation}:parseScriptsJson`),
136+ }).pipe(Effect.flatMap((value) => parseProjectScripts(value, `${operation}:decodeScripts`)));
137+}
138+
139+function rowToProjectRecord(
140+ row: ProjectRow,
141+ operation: string,
142+): Effect.Effect<ProjectRecord, ProjectRepositoryError> {
143+ return decodeScriptsJson(row.scriptsJson, `${operation}:scripts`).pipe(
144+ Effect.flatMap((scripts) => {
145+ const parsed = projectRecordSchema.safeParse({
146+ id: row.id,
147+ cwd: row.cwd,
148+ name: row.name,
149+ scripts,
150+ createdAt: row.createdAt,
151+ updatedAt: row.updatedAt,
152+ });
153+ if (!parsed.success) {
154+ return Effect.fail(
155+ new PersistenceDecodeError({
156+ operation: `${operation}:projectRecord`,
157+ issue: parsed.error.message,
158+ cause: (parsed.error as { cause?: unknown }).cause,
159+ }),
160+ );
161+ }
162+ return Effect.succeed(parsed.data);
163+ }),
164+ );
165+}
166+
167+function decodeAddInput(rawInput: ProjectAddInput) {
168+ const parsed = projectAddInputSchema.safeParse(rawInput);
169+ if (!parsed.success) {
170+ return Effect.fail(
171+ new ProjectValidationError({
172+ operation: "ProjectRepository.add",
173+ issue: parsed.error.message,
174+ cause: (parsed.error as { cause?: unknown }).cause,
175+ }),
176+ );
177+ }
178+ return Effect.succeed(parsed.data);
179+}
180+
181+function decodeUpdateScriptsInput(rawInput: ProjectUpdateScriptsInput) {
182+ const parsed = projectUpdateScriptsInputSchema.safeParse(rawInput);
183+ if (!parsed.success) {
184+ return Effect.fail(
185+ new ProjectValidationError({
186+ operation: "ProjectRepository.updateScripts",
187+ issue: parsed.error.message,
188+ cause: (parsed.error as { cause?: unknown }).cause,
189+ }),
190+ );
191+ }
192+ return Effect.succeed(parsed.data);
193+}
194+
195+function decodeRemoveInput(rawInput: unknown) {
196+ const parsed = projectRemoveInputSchema.safeParse(rawInput);
197+ if (!parsed.success) {
198+ return Effect.fail(
199+ new ProjectValidationError({
200+ operation: "ProjectRepository.remove",
201+ issue: parsed.error.message,
202+ cause: (parsed.error as { cause?: unknown }).cause,
203+ }),
204+ );
205+ }
206+ return Effect.succeed(parsed.data);
120207 }
121208
122209 const makeRepository = Effect.gen(function* () {
@@ -231,24 +318,31 @@ const makeRepository = Effect.gen(function* () { · 108 unmodified lines
231318
232319 const list: ProjectRepositoryShape["list"] = () =>
233320 listProjectRows(undefined).pipe(
234− Effect.map((rows): ProjectListResult => rows.map(rowToProjectRecord)),
235− Effect.map((projects) =>
236− projects.map((project) => ({ ...project, scripts: cloneScripts(project.scripts) })),
321+ Effect.mapError(toPersistenceSqlError("ProjectRepository.list:query")),
322+ Effect.flatMap((rows) =>
323+ Effect.forEach(rows, (row) =>
324+ rowToProjectRecord(row, "ProjectRepository.list:rowToProject"),
325+ ),
326+ ),
327+ Effect.map(
328+ (projects): ProjectListResult =>
329+ projects.map((project) => ({ ...project, scripts: cloneScripts(project.scripts) })),
237330 ),
238− Effect.orDie,
239331 );
240332
241333 const add: ProjectRepositoryShape["add"] = (rawInput) =>
242334 Effect.gen(function* () {
243− const input = projectAddInputSchema.parse(rawInput as ProjectAddInput);
335+ const input = yield* decodeAddInput(rawInput);
244336 const normalizedCwd = normalizeCwd(input.cwd);
245337 if (!isDirectory(normalizedCwd)) {
246− return yield* Effect.fail(new Error(`Project path does not exist: ${normalizedCwd}`));
338+ return yield* Effect.fail(new ProjectPathMissingError({ cwd: normalizedCwd }));
247339 }
248340
249− const existing = yield* findProjectByCwd({ cwd: normalizedCwd }).pipe(Effect.orDie);
341+ const existing = yield* findProjectByCwd({ cwd: normalizedCwd }).pipe(
342+ Effect.mapError(toPersistenceSqlError("ProjectRepository.add:findByCwd")),
343+ );
250344 if (Option.isSome(existing)) {
251− const project = rowToProjectRecord(existing.value);
345+ const project = yield* rowToProjectRecord(existing.value, "ProjectRepository.add:existing");
252346 const result: ProjectAddResult = {
253347 project: {
254348 ...project,
@@ -267,9 +361,9 @@ const makeRepository = Effect.gen(function* () { · 12 unmodified lines
267361 scriptsJson: "[]",
268362 createdAt: now,
269363 updatedAt: now,
270− }).pipe(Effect.orDie);
364+ }).pipe(Effect.mapError(toPersistenceSqlError("ProjectRepository.add:insert")));
271365
272− const created = rowToProjectRecord(row);
366+ const created = yield* rowToProjectRecord(row, "ProjectRepository.add:created");
273367 const result: ProjectAddResult = {
274368 project: {
275369 ...created,
@@ -282,27 +376,48 @@ const makeRepository = Effect.gen(function* () { · 6 unmodified lines
282376
283377 const remove: ProjectRepositoryShape["remove"] = (rawInput) =>
284378 Effect.gen(function* () {
285− const input = projectRemoveInputSchema.parse(rawInput);
286− yield* deleteProject({ id: input.id }).pipe(Effect.orDie);
379+ const input = yield* decodeRemoveInput(rawInput);
380+ const existing = yield* findProjectById({ id: input.id }).pipe(
381+ Effect.mapError(toPersistenceSqlError("ProjectRepository.remove:findById")),
382+ );
383+ if (Option.isNone(existing)) {
384+ return yield* Effect.fail(new ProjectNotFoundError({ projectId: input.id }));
385+ }
386+
387+ yield* deleteProject({ id: input.id }).pipe(
388+ Effect.mapError(toPersistenceSqlError("ProjectRepository.remove:delete")),
389+ );
287390 });
288391
289392 const updateScripts: ProjectRepositoryShape["updateScripts"] = (rawInput) =>
290393 Effect.gen(function* () {
291− const input = projectUpdateScriptsInputSchema.parse(rawInput);
292− const existing = yield* findProjectById({ id: input.id }).pipe(Effect.orDie);
394+ const input = yield* decodeUpdateScriptsInput(rawInput);
395+ const existing = yield* findProjectById({ id: input.id }).pipe(
396+ Effect.mapError(toPersistenceSqlError("ProjectRepository.updateScripts:findById")),
397+ );
293398 if (Option.isNone(existing)) {
294− return yield* Effect.fail(new Error(`Project not found: ${input.id}`));
399+ return yield* Effect.fail(new ProjectNotFoundError({ projectId: input.id }));
295400 }
296401
297− const nextScripts = normalizeProjectScripts(projectScriptsSchema.parse(input.scripts));
402+ const nextScripts = yield* parseProjectScripts(
403+ input.scripts,
404+ "ProjectRepository.updateScripts:decodeScripts",
405+ );
406+ const scriptsJson = yield* Effect.try({
407+ try: () => JSON.stringify(nextScripts),
408+ catch: toPersistenceSerializationError("ProjectRepository.updateScripts:encodeScripts"),
409+ });
298410 const nextUpdatedAt = new Date().toISOString();
299411 const updatedRow = yield* updateProjectScripts({
300412 id: input.id,
301− scriptsJson: JSON.stringify(nextScripts),
413+ scriptsJson,
302414 updatedAt: nextUpdatedAt,
303− }).pipe(Effect.orDie);
415+ }).pipe(Effect.mapError(toPersistenceSqlError("ProjectRepository.updateScripts:update")));
304416
305− const updated = rowToProjectRecord(updatedRow);
417+ const updated = yield* rowToProjectRecord(
418+ updatedRow,
419+ "ProjectRepository.updateScripts:updated",
420+ );
306421 const result: ProjectUpdateScriptsResult = {
307422 project: {
308423 ...updated,
@@ -314,17 +429,19 @@ const makeRepository = Effect.gen(function* () { · 5 unmodified lines
314429
315430 const pruneMissing: ProjectRepositoryShape["pruneMissing"] = () =>
316431 listProjectRows(undefined).pipe(
432+ Effect.mapError(toPersistenceSqlError("ProjectRepository.pruneMissing:query")),
317433 Effect.flatMap((rows) =>
318434 Effect.forEach(rows, (row) => {
319435 const normalizedCwd = normalizeCwd(row.cwd);
320436 if (isDirectory(normalizedCwd)) {
321437 return Effect.void;
322438 }
323− return deleteProject({ id: row.id }).pipe(Effect.orDie);
439+ return deleteProject({ id: row.id }).pipe(
440+ Effect.mapError(toPersistenceSqlError("ProjectRepository.pruneMissing:delete")),
441+ );
324442 }),
325443 ),
326444 Effect.asVoid,
327− Effect.orDie,
328445 );
329446
330447 return {
apps/server/src/persistence/Services/OrchestrationEvents.ts+44−3
@@ -1,18 +1,59 @@
1+/**
2+ * OrchestrationEventRepository - Repository interface for orchestration events.
3+ *
4+ * Uses Effect `Context.Tag` for dependency injection and exposes typed
5+ * persistence/decode errors for event append and replay operations.
6+ *
7+ * @module OrchestrationEventRepository
8+ */
19 import type { OrchestrationEvent } from "@t3tools/contracts";
210 import { Context } from "effect";
311 import type { Effect } from "effect";
412
13+import type { OrchestrationEventRepositoryError } from "../Errors";
14+
515 export interface OrchestrationEventRepositoryShape {
16+ /**
17+ * Persist a new orchestration event.
18+ *
19+ * @param event - Event payload without sequence (assigned by storage).
20+ * @returns Effect containing the stored event with assigned sequence.
21+ */
622 readonly append: (
723 event: Omit<OrchestrationEvent, "sequence">,
8− ) => Effect.Effect<OrchestrationEvent>;
24+ ) => Effect.Effect<OrchestrationEvent, OrchestrationEventRepositoryError>;
25+
26+ /**
27+ * Replay events after the provided sequence.
28+ *
29+ * @param sequenceExclusive - Sequence cursor (exclusive).
30+ * @param limit - Maximum number of events to return.
31+ * @returns Effect containing ordered events.
32+ */
933 readonly readFromSequence: (
1034 sequenceExclusive: number,
1135 limit?: number,
12− ) => Effect.Effect<OrchestrationEvent[]>;
13− readonly readAll: () => Effect.Effect<OrchestrationEvent[]>;
36+ ) => Effect.Effect<OrchestrationEvent[], OrchestrationEventRepositoryError>;
37+
38+ /**
39+ * Read all events from the beginning of the stream.
40+ *
41+ * @returns Effect containing all stored events.
42+ */
43+ readonly readAll: () => Effect.Effect<OrchestrationEvent[], OrchestrationEventRepositoryError>;
1444 }
1545
46+/**
47+ * OrchestrationEventRepository - Context tag for orchestration event persistence.
48+ *
49+ * @example
50+ * ```ts
51+ * const program = Effect.gen(function* () {
52+ * const events = yield* OrchestrationEventRepository
53+ * return yield* events.readAll()
54+ * })
55+ * ```
56+ */
1657 export class OrchestrationEventRepository extends Context.Tag("orchestration/EventRepository")<
1758 OrchestrationEventRepository,
1859 OrchestrationEventRepositoryShape
apps/server/src/persistence/Services/Projects.ts+58−5
@@ -1,3 +1,11 @@
1+/**
2+ * ProjectRepository - Repository interface for project persistence.
3+ *
4+ * Uses Effect `Context.Tag` for dependency injection and returns typed
5+ * domain errors for all operations.
6+ *
7+ * @module ProjectRepository
8+ */
19 import type {
210 ProjectAddInput,
311 ProjectAddResult,
@@ -9,16 +17,61 @@ import type { · 5 unmodified lines
917 import { Context } from "effect";
1018 import type { Effect } from "effect";
1119
20+import type { ProjectRepositoryError } from "../Errors";
21+
1222 export interface ProjectRepositoryShape {
13− readonly list: () => Effect.Effect<ProjectListResult>;
14− readonly add: (input: ProjectAddInput) => Effect.Effect<ProjectAddResult, Error>;
15− readonly remove: (input: ProjectRemoveInput) => Effect.Effect<void, Error>;
23+ /**
24+ * List all persisted projects.
25+ *
26+ * @returns Effect containing all project records.
27+ */
28+ readonly list: () => Effect.Effect<ProjectListResult, ProjectRepositoryError>;
29+
30+ /**
31+ * Add a project by cwd, or return the existing project when the cwd already exists.
32+ *
33+ * @param input - Project creation payload.
34+ * @returns Effect containing the project and whether it was newly created.
35+ */
36+ readonly add: (input: ProjectAddInput) => Effect.Effect<ProjectAddResult, ProjectRepositoryError>;
37+
38+ /**
39+ * Remove a persisted project.
40+ *
41+ * @param input - Project removal payload.
42+ * @returns Effect containing void.
43+ */
44+ readonly remove: (input: ProjectRemoveInput) => Effect.Effect<void, ProjectRepositoryError>;
45+
46+ /**
47+ * Update the scripts for an existing project.
48+ *
49+ * @param input - Script update payload.
50+ * @returns Effect containing the updated project.
51+ */
1652 readonly updateScripts: (
1753 input: ProjectUpdateScriptsInput,
18− ) => Effect.Effect<ProjectUpdateScriptsResult, Error>;
19− readonly pruneMissing: () => Effect.Effect<void>;
54+ ) => Effect.Effect<ProjectUpdateScriptsResult, ProjectRepositoryError>;
55+
56+ /**
57+ * Remove projects whose cwd no longer exists on disk.
58+ *
59+ * @returns Effect containing void.
60+ */
61+ readonly pruneMissing: () => Effect.Effect<void, ProjectRepositoryError>;
2062 }
2163
64+/**
65+ * ProjectRepository - Context tag for project persistence dependency injection.
66+ *
67+ * @example
68+ * ```ts
69+ * const program = Effect.gen(function* () {
70+ * const projects = yield* ProjectRepository
71+ * return yield* projects.list()
72+ * })
73+ * ```
74+ */
2275 export class ProjectRepository extends Context.Tag("persistence/ProjectRepository")<
2376 ProjectRepository,
2477 ProjectRepositoryShape