authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-22 01:43:53-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-02-26 13:20:36-08:00
logb152462b2cd5f26f40ae174765ede2c62381d282
tree3fd5a6879fa53d5d21cf33ebf2beb79364272287
parente219efa22a4a8a6cc1424af54b04c58be9a8b68d

more effect: error handling

11 files changed, 846 insertions(+), 260 deletions(-)

apps/server/src/orchestration/Errors.ts created+95
...@@ -0,0 +1,95 @@
1import { ParseResult, Schema } from "effect";
2
3import type { OrchestrationEventRepositoryError } from "../persistence/Errors";
4
5export 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
17export 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
29export 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
42export 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
55export type OrchestrationDispatchError =
56 | OrchestrationEventRepositoryError
57 | OrchestrationReducerDecodeError
58 | OrchestrationListenerCallbackError;
59
60export type OrchestrationEngineError =
61 | OrchestrationDispatchError
62 | OrchestrationCommandJsonParseError
63 | OrchestrationCommandDecodeError;
64
65export function toOrchestrationCommandDecodeError(error: ParseResult.ParseError) {
66 return new OrchestrationCommandDecodeError({
67 issue: ParseResult.TreeFormatter.formatErrorSync(error),
68 cause: error,
69 });
70}
71
72export 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
81export function toOrchestrationJsonParseError(cause: unknown) {
82 return new OrchestrationCommandJsonParseError({
83 detail: `Failed to parse orchestration command JSON: ${cause}`,
84 cause,
85 });
86}
87
88export 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 {...@@ -10,6 +10,7 @@ import {
10 OrchestrationEventRepository,10 OrchestrationEventRepository,
11 type OrchestrationEventRepositoryShape,11 type OrchestrationEventRepositoryShape,
12} from "../persistence/Services/OrchestrationEvents";12} from "../persistence/Services/OrchestrationEvents";
13import { PersistenceSqlError } from "../persistence/Errors";
1314
14import { Layer, ManagedRuntime } from "effect";15import { Layer, ManagedRuntime } from "effect";
1516
...@@ -375,7 +376,12 @@ describe("OrchestrationEngine", () => {...@@ -375,7 +376,12 @@ describe("OrchestrationEngine", () => {
375 append(event) {376 append(event) {
376 if (shouldFailFirstAppend) {377 if (shouldFailFirstAppend) {
377 shouldFailFirstAppend = false;378 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 );
379 }385 }
380 const savedEvent = {386 const savedEvent = {
381 ...event,387 ...event,
apps/server/src/orchestration/Layer.ts+115-119
...@@ -4,33 +4,24 @@ import type {...@@ -4,33 +4,24 @@ import type {
4 OrchestrationReadModel,4 OrchestrationReadModel,
5} from "@t3tools/contracts";5} from "@t3tools/contracts";
6import { OrchestrationCommandSchema } from "@t3tools/contracts";6import { OrchestrationCommandSchema } from "@t3tools/contracts";
7import {7import { Deferred, Effect, Layer, PubSub, Queue, Schema } from "effect";
8 Cause,
9 Deferred,
10 Effect,
11 Either,
12 Fiber,
13 Layer,
14 PubSub,
15 Queue,
16 Runtime,
17 Schema,
18} from "effect";
198
20import { createLogger } from "../logger";9import { createLogger } from "../logger";
21import { OrchestrationEventRepository } from "../persistence/Services/OrchestrationEvents";10import { OrchestrationEventRepository } from "../persistence/Services/OrchestrationEvents";
11import {
12 toListenerCallbackError,
13 toOrchestrationCommandDecodeError,
14 toOrchestrationJsonParseError,
15 type OrchestrationDispatchError,
16} from "./Errors";
22import { createEmptyReadModel, reduceEvent } from "./reducer";17import { createEmptyReadModel, reduceEvent } from "./reducer";
23import { OrchestrationEngineService, type OrchestrationEngineShape } from "./Service";18import { OrchestrationEngineService, type OrchestrationEngineShape } from "./Service";
2419
25type CommandEnvelope = {20type CommandEnvelope = {
26 command: OrchestrationCommand;21 command: OrchestrationCommand;
27 result: Deferred.Deferred<{ sequence: number }, Error>;22 result: Deferred.Deferred<{ sequence: number }, OrchestrationDispatchError>;
28};23};
2924
30function asError(error: unknown, fallbackMessage: string): Error {
31 return error instanceof Error ? error : new Error(fallbackMessage);
32}
33
34function mapCommandToEvent(command: OrchestrationCommand): Omit<OrchestrationEvent, "sequence"> {25function mapCommandToEvent(command: OrchestrationCommand): Omit<OrchestrationEvent, "sequence"> {
35 const eventId = crypto.randomUUID();26 const eventId = crypto.randomUUID();
36 switch (command.type) {27 switch (command.type) {
...@@ -169,129 +160,134 @@ function mapCommandToEvent(command: OrchestrationCommand): Omit<OrchestrationEve...@@ -169,129 +160,134 @@ function mapCommandToEvent(command: OrchestrationCommand): Omit<OrchestrationEve
169 }160 }
170}161}
171162
172const decodeUnknownCommand = Schema.decodeUnknownEither(OrchestrationCommandSchema);163const decodeUnknownCommand = Schema.decodeUnknown(OrchestrationCommandSchema);
173164
174export const makeOrchestrationEngine = Effect.gen(function* () {165const makeOrchestrationEngine = () =>
175 const logger = createLogger("orchestration");166 Effect.gen(function* () {
176 const eventStore = yield* OrchestrationEventRepository;167 const logger = createLogger("orchestration");
177 const runtime = Runtime.defaultRuntime;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>();172 const commandQueue = yield* Queue.unbounded<CommandEnvelope>();
182 const readModelPubSub = yield* PubSub.unbounded<OrchestrationReadModel>();173 const readModelPubSub = yield* PubSub.unbounded<OrchestrationReadModel>();
183 const eventPubSub = yield* PubSub.unbounded<OrchestrationEvent>();174 const eventPubSub = yield* PubSub.unbounded<OrchestrationEvent>();
184175
185 const readModelListeners = new Set<(snapshot: OrchestrationReadModel) => void>();176 const readModelListeners = new Set<(snapshot: OrchestrationReadModel) => void>();
186 const domainEventListeners = new Set<(event: OrchestrationEvent) => void>();177 const domainEventListeners = new Set<(event: OrchestrationEvent) => void>();
187178
188 const processEnvelope = (envelope: CommandEnvelope): Effect.Effect<void> =>179 const notifyReadModelListeners = (snapshot: OrchestrationReadModel) =>
189 Effect.gen(function* () {180 Effect.forEach(readModelListeners, (listener) =>
190 const eventBase = mapCommandToEvent(envelope.command);181 Effect.try({
191 const savedEvent = yield* eventStore.append(eventBase);182 try: () => listener(snapshot),
192 readModel = yield* reduceEvent(readModel, savedEvent);183 catch: toListenerCallbackError("read-model"),
184 }),
185 ).pipe(Effect.asVoid);
193186
194 const snapshot = readModel;187 const notifyDomainEventListeners = (event: OrchestrationEvent) =>
195 yield* Effect.all([188 Effect.forEach(domainEventListeners, (listener) =>
196 PubSub.publish(eventPubSub, savedEvent),189 Effect.try({
197 PubSub.publish(readModelPubSub, snapshot),190 try: () => listener(event),
198 ]);191 catch: toListenerCallbackError("domain-event"),
192 }),
193 ).pipe(Effect.asVoid);
199194
200 for (const listener of domainEventListeners) {195 const processEnvelope = (envelope: CommandEnvelope): Effect.Effect<void> =>
201 listener(savedEvent);196 Effect.gen(function* () {
202 }197 const eventBase = mapCommandToEvent(envelope.command);
203 for (const listener of readModelListeners) {198 const savedEvent = yield* eventStore.append(eventBase);
204 listener(snapshot);199 readModel = yield* reduceEvent(readModel, savedEvent);
205 }
206200
207 yield* Deferred.succeed(envelope.result, { sequence: savedEvent.sequence });201 const snapshot = readModel;
208 }).pipe(202 yield* Effect.all([
209 Effect.catchAllCause((cause) =>203 PubSub.publish(eventPubSub, savedEvent),
210 Deferred.fail(204 PubSub.publish(readModelPubSub, snapshot),
211 envelope.result,205 ]);
212 asError(Cause.squash(cause), "Unknown command processing error"),
213 ).pipe(Effect.asVoid),
214 ),
215 );
216206
217 const bootstrapReadModel: Effect.Effect<void, unknown> = Effect.gen(function* () {207 yield* notifyDomainEventListeners(savedEvent);
218 const existingEvents = yield* eventStore.readAll();208 yield* notifyReadModelListeners(snapshot);
219 for (const event of existingEvents) {
220 readModel = yield* reduceEvent(readModel, event);
221 }
222 });
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)));215 const bootstrapReadModel: Effect.Effect<void, OrchestrationDispatchError> = Effect.gen(
227 const workerFiber: Fiber.RuntimeFiber<void, unknown> = Runtime.runFork(runtime)(worker);216 function* () {
228 logger.info("orchestration engine started", {217 const existingEvents = yield* eventStore.readAll();
229 sequence: readModel.sequence,218 for (const event of existingEvents) {
230 });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"] = () =>232 const getSnapshot: OrchestrationEngineShape["getSnapshot"] = () =>
235 Effect.sync((): OrchestrationReadModel => readModel);233 Effect.sync((): OrchestrationReadModel => readModel);
236234
237 const replayEvents: OrchestrationEngineShape["replayEvents"] = (fromSequenceExclusive) =>235 const replayEvents: OrchestrationEngineShape["replayEvents"] = (fromSequenceExclusive) =>
238 eventStore.readFromSequence(fromSequenceExclusive);236 eventStore.readFromSequence(fromSequenceExclusive);
239237
240 const dispatch: OrchestrationEngineShape["dispatch"] = (command) =>238 const dispatch: OrchestrationEngineShape["dispatch"] = (command) =>
241 Effect.gen(function* () {239 Effect.gen(function* () {
242 const result = yield* Deferred.make<{ sequence: number }, Error>();240 const result = yield* Deferred.make<{ sequence: number }, OrchestrationDispatchError>();
243 yield* Queue.offer(commandQueue, { command, result });241 yield* Queue.offer(commandQueue, { command, result });
244 return yield* Deferred.await(result);242 return yield* Deferred.await(result);
245 }).pipe(Effect.mapError((error) => asError(error, "Queue offer failed")));243 });
246244
247 const dispatchUnknown: OrchestrationEngineShape["dispatchUnknown"] = (command) =>245 const dispatchUnknown: OrchestrationEngineShape["dispatchUnknown"] = (command) =>
248 Effect.gen(function* () {246 Effect.gen(function* () {
249 let payload: unknown = command;247 const payload =
250 if (typeof command === "string") {248 typeof command === "string"
251 try {249 ? yield* Effect.try({
252 payload = JSON.parse(command) as unknown;250 try: () => JSON.parse(command) as unknown,
253 } catch {251 catch: toOrchestrationJsonParseError,
254 return yield* Effect.fail(252 })
255 new Error("Invalid orchestration command: payload is not valid JSON"),253 : command;
256 );254
257 }255 const decoded = yield* decodeUnknownCommand(payload).pipe(
258 }256 Effect.mapError(toOrchestrationCommandDecodeError),
259 const decoded = decodeUnknownCommand(payload);
260 if (Either.isLeft(decoded)) {
261 return yield* Effect.fail(
262 new Error(`Invalid orchestration command: ${decoded.left.toString()}`),
263 );257 );
264 }
265 return yield* dispatch(decoded.right);
266 });
267258
268 const subscribeToReadModel: OrchestrationEngineShape["subscribeToReadModel"] = (callback) =>259 return yield* dispatch(decoded);
269 Effect.sync(() => {260 });
270 readModelListeners.add(callback);
271 return () => {
272 readModelListeners.delete(callback);
273 };
274 });
275261
276 const subscribeToDomainEvents: OrchestrationEngineShape["subscribeToDomainEvents"] = (callback) =>262 const subscribeToReadModel: OrchestrationEngineShape["subscribeToReadModel"] = (callback) =>
277 Effect.sync(() => {263 Effect.sync(() => {
278 domainEventListeners.add(callback);264 readModelListeners.add(callback);
279 return () => {265 return () => {
280 domainEventListeners.delete(callback);266 readModelListeners.delete(callback);
281 };267 };
282 });268 });
283269
284 return {270 const subscribeToDomainEvents: OrchestrationEngineShape["subscribeToDomainEvents"] = (
285 getSnapshot,271 callback,
286 replayEvents,272 ) =>
287 dispatchUnknown,273 Effect.sync(() => {
288 dispatch,274 domainEventListeners.add(callback);
289 subscribeToReadModel,275 return () => {
290 subscribeToDomainEvents,276 domainEventListeners.delete(callback);
291 } satisfies OrchestrationEngineShape;277 };
292});278 });
279
280 return {
281 getSnapshot,
282 replayEvents,
283 dispatchUnknown,
284 dispatch,
285 subscribeToReadModel,
286 subscribeToDomainEvents,
287 } satisfies OrchestrationEngineShape;
288 });
293289
294export const OrchestrationEngineLive = Layer.scoped(290export const OrchestrationEngineLive = Layer.scoped(
295 OrchestrationEngineService,291 OrchestrationEngineService,
296 makeOrchestrationEngine,292 makeOrchestrationEngine(),
297);293);
apps/server/src/orchestration/Service.ts+73-3
...@@ -1,3 +1,11 @@...@@ -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 */
1import type {9import type {
2 OrchestrationCommand,10 OrchestrationCommand,
3 OrchestrationEvent,11 OrchestrationEvent,
...@@ -6,19 +14,81 @@ import type {...@@ -6,19 +14,81 @@ import type {
6import { Context } from "effect";14import { Context } from "effect";
7import type { Effect } from "effect";15import type { Effect } from "effect";
816
17import type { OrchestrationDispatchError, OrchestrationEngineError } from "./Errors";
18import type { OrchestrationEventRepositoryError } from "../persistence/Errors";
19
9export interface OrchestrationEngineShape {20export interface OrchestrationEngineShape {
21 /**
22 * Read the current in-memory orchestration snapshot.
23 *
24 * @returns Effect containing the latest read model.
25 */
10 readonly getSnapshot: () => Effect.Effect<OrchestrationReadModel>;26 readonly getSnapshot: () => Effect.Effect<OrchestrationReadModel>;
11 readonly replayEvents: (fromSequenceExclusive: number) => Effect.Effect<OrchestrationEvent[]>;27
12 readonly dispatchUnknown: (command: unknown) => Effect.Effect<{ sequence: number }, Error>;28 /**
13 readonly dispatch: (command: OrchestrationCommand) => Effect.Effect<{ sequence: number }, Error>;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 */
14 readonly subscribeToReadModel: (66 readonly subscribeToReadModel: (
15 callback: (snapshot: OrchestrationReadModel) => void,67 callback: (snapshot: OrchestrationReadModel) => void,
16 ) => Effect.Effect<() => void>;68 ) => 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 */
17 readonly subscribeToDomainEvents: (76 readonly subscribeToDomainEvents: (
18 callback: (event: OrchestrationEvent) => void,77 callback: (event: OrchestrationEvent) => void,
19 ) => Effect.Effect<() => void>;78 ) => Effect.Effect<() => void>;
20}79}
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 */
22export class OrchestrationEngineService extends Context.Tag("orchestration/Engine")<92export class OrchestrationEngineService extends Context.Tag("orchestration/Engine")<
23 OrchestrationEngineService,93 OrchestrationEngineService,
24 OrchestrationEngineShape94 OrchestrationEngineShape
apps/server/src/orchestration/reducer.ts+113-66
...@@ -9,7 +9,6 @@ import type {...@@ -9,7 +9,6 @@ import type {
9import {9import {
10 OrchestrationGitReadModelSchema,10 OrchestrationGitReadModelSchema,
11 OrchestrationMessageSchema,11 OrchestrationMessageSchema,
12 OrchestrationReadModelSchema,
13 OrchestrationSessionSchema,12 OrchestrationSessionSchema,
14 OrchestrationThreadSchema,13 OrchestrationThreadSchema,
15 OrchestrationTurnDiffFileSchema,14 OrchestrationTurnDiffFileSchema,
...@@ -17,6 +16,8 @@ import {...@@ -17,6 +16,8 @@ import {
17} from "@t3tools/contracts";16} from "@t3tools/contracts";
18import { Effect, Schema } from "effect";17import { Effect, Schema } from "effect";
1918
19import { toReducerDecodeError, type OrchestrationReducerDecodeError } from "./Errors";
20
20type ThreadPatch = Partial<Omit<OrchestrationThread, "id" | "projectId">>;21type ThreadPatch = Partial<Omit<OrchestrationThread, "id" | "projectId">>;
2122
22function updateThread(23function updateThread(
...@@ -27,12 +28,16 @@ function updateThread(...@@ -27,12 +28,16 @@ function updateThread(
27 return threads.map((thread) => (thread.id === threadId ? { ...thread, ...patch } : thread));28 return threads.map((thread) => (thread.id === threadId ? { ...thread, ...patch } : thread));
28}29}
2930
30const decodeReadModelSync = Schema.decodeUnknownSync(OrchestrationReadModelSchema);31function decodeForEvent<A>(
31const decodeMessage = Schema.decodeUnknown(OrchestrationMessageSchema);32 schema: Schema.Schema<A>,
32const decodeSession = Schema.decodeUnknown(OrchestrationSessionSchema);33 value: unknown,
33const decodeThread = Schema.decodeUnknown(OrchestrationThreadSchema);34 eventType: string,
34const decodeGitReadModel = Schema.decodeUnknown(OrchestrationGitReadModelSchema);35 field: string,
35const decodeTurnDiffSummary = Schema.decodeUnknown(OrchestrationTurnDiffSummarySchema);36): Effect.Effect<A, OrchestrationReducerDecodeError> {
37 return Schema.decodeUnknown(schema)(value).pipe(
38 Effect.mapError(toReducerDecodeError(`${eventType}:${field}`)),
39 );
40}
3641
37const ThreadCreatedPayloadSchema = Schema.Struct({42const ThreadCreatedPayloadSchema = Schema.Struct({
38 id: Schema.String,43 id: Schema.String,
...@@ -44,7 +49,6 @@ const ThreadCreatedPayloadSchema = Schema.Struct({...@@ -44,7 +49,6 @@ const ThreadCreatedPayloadSchema = Schema.Struct({
44 createdAt: Schema.String,49 createdAt: Schema.String,
45 updatedAt: Schema.String,50 updatedAt: Schema.String,
46});51});
47const decodeThreadCreatedPayload = Schema.decodeUnknown(ThreadCreatedPayloadSchema);
4852
49const ThreadMetaUpdatedPayloadSchema = Schema.Struct({53const ThreadMetaUpdatedPayloadSchema = Schema.Struct({
50 threadId: Schema.String,54 threadId: Schema.String,
...@@ -54,7 +58,6 @@ const ThreadMetaUpdatedPayloadSchema = Schema.Struct({...@@ -54,7 +58,6 @@ const ThreadMetaUpdatedPayloadSchema = Schema.Struct({
54 worktreePath: Schema.optional(Schema.NullOr(Schema.String)),58 worktreePath: Schema.optional(Schema.NullOr(Schema.String)),
55 updatedAt: Schema.String,59 updatedAt: Schema.String,
56});60});
57const decodeThreadMetaUpdatedPayload = Schema.decodeUnknown(ThreadMetaUpdatedPayloadSchema);
5861
59const MessageSentPayloadSchema = Schema.Struct({62const MessageSentPayloadSchema = Schema.Struct({
60 id: Schema.String,63 id: Schema.String,
...@@ -64,13 +67,11 @@ const MessageSentPayloadSchema = Schema.Struct({...@@ -64,13 +67,11 @@ const MessageSentPayloadSchema = Schema.Struct({
64 createdAt: Schema.String,67 createdAt: Schema.String,
65 streaming: Schema.Boolean,68 streaming: Schema.Boolean,
66});69});
67const decodeMessageSentPayload = Schema.decodeUnknown(MessageSentPayloadSchema);
6870
69const ThreadSessionSetPayloadSchema = Schema.Struct({71const ThreadSessionSetPayloadSchema = Schema.Struct({
70 threadId: Schema.String,72 threadId: Schema.String,
71 session: OrchestrationSessionSchema,73 session: OrchestrationSessionSchema,
72});74});
73const decodeThreadSessionSetPayload = Schema.decodeUnknown(ThreadSessionSetPayloadSchema);
7475
75const ThreadTurnDiffCompletedPayloadSchema = Schema.Struct({76const ThreadTurnDiffCompletedPayloadSchema = Schema.Struct({
76 threadId: Schema.String,77 threadId: Schema.String,
...@@ -81,30 +82,26 @@ const ThreadTurnDiffCompletedPayloadSchema = Schema.Struct({...@@ -81,30 +82,26 @@ const ThreadTurnDiffCompletedPayloadSchema = Schema.Struct({
81 assistantMessageId: Schema.optional(Schema.String),82 assistantMessageId: Schema.optional(Schema.String),
82 checkpointTurnCount: Schema.optional(Schema.Number),83 checkpointTurnCount: Schema.optional(Schema.Number),
83});84});
84const decodeThreadTurnDiffCompletedPayload = Schema.decodeUnknown(
85 ThreadTurnDiffCompletedPayloadSchema,
86);
8785
88const ThreadRevertedPayloadSchema = Schema.Struct({86const ThreadRevertedPayloadSchema = Schema.Struct({
89 threadId: Schema.String,87 threadId: Schema.String,
90 turnCount: Schema.Number,88 turnCount: Schema.Number,
91 messageCount: Schema.Number,89 messageCount: Schema.Number,
92});90});
93const decodeThreadRevertedPayload = Schema.decodeUnknown(ThreadRevertedPayloadSchema);
9491
95export function createEmptyReadModel(nowIso: string): OrchestrationReadModel {92export function createEmptyReadModel(nowIso: string): OrchestrationReadModel {
96 return decodeReadModelSync({93 return {
97 sequence: 0,94 sequence: 0,
98 threads: [],95 threads: [],
99 gitByProjectId: {},96 gitByProjectId: {},
100 updatedAt: nowIso,97 updatedAt: nowIso,
101 });98 };
102}99}
103100
104export function reduceEvent(101export function reduceEvent(
105 model: OrchestrationReadModel,102 model: OrchestrationReadModel,
106 event: OrchestrationEvent,103 event: OrchestrationEvent,
107): Effect.Effect<OrchestrationReadModel, unknown> {104): Effect.Effect<OrchestrationReadModel, OrchestrationReducerDecodeError> {
108 const nextBase: OrchestrationReadModel = {105 const nextBase: OrchestrationReadModel = {
109 ...model,106 ...model,
110 sequence: event.sequence,107 sequence: event.sequence,
...@@ -114,25 +111,35 @@ export function reduceEvent(...@@ -114,25 +111,35 @@ export function reduceEvent(
114 switch (event.type) {111 switch (event.type) {
115 case "thread.created":112 case "thread.created":
116 return Effect.gen(function* () {113 return Effect.gen(function* () {
117 const payload = yield* decodeThreadCreatedPayload(event.payload);114 const payload = yield* decodeForEvent(
118 const thread: OrchestrationThread = yield* decodeThread({115 ThreadCreatedPayloadSchema,
119 id: payload.id,116 event.payload,
120 projectId: payload.projectId,117 event.type,
121 title: payload.title,118 "payload",
122 model: payload.model,119 );
123 branch: payload.branch,120 const thread: OrchestrationThread = yield* decodeForEvent(
124 worktreePath: payload.worktreePath,121 OrchestrationThreadSchema,
125 createdAt: payload.createdAt,122 {
126 updatedAt: payload.updatedAt,123 id: payload.id,
127 latestTurnId: null,124 projectId: payload.projectId,
128 latestTurnStartedAt: null,125 title: payload.title,
129 latestTurnCompletedAt: null,126 model: payload.model,
130 latestTurnDurationMs: null,127 branch: payload.branch,
131 messages: [],128 worktreePath: payload.worktreePath,
132 session: null,129 createdAt: payload.createdAt,
133 turnDiffSummaries: [],130 updatedAt: payload.updatedAt,
134 error: null,131 latestTurnId: null,
135 });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 );
136 const existing = nextBase.threads.find((entry) => entry.id === thread.id);143 const existing = nextBase.threads.find((entry) => entry.id === thread.id);
137 return {144 return {
138 ...nextBase,145 ...nextBase,
...@@ -147,7 +154,12 @@ export function reduceEvent(...@@ -147,7 +154,12 @@ export function reduceEvent(
147 threads: nextBase.threads.filter((thread) => thread.id !== event.aggregateId),154 threads: nextBase.threads.filter((thread) => thread.id !== event.aggregateId),
148 });155 });
149 case "thread.meta-updated":156 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(
151 Effect.map((payload) => {163 Effect.map((payload) => {
152 const thread = nextBase.threads.find((entry) => entry.id === payload.threadId);164 const thread = nextBase.threads.find((entry) => entry.id === payload.threadId);
153 if (!thread) {165 if (!thread) {
...@@ -167,19 +179,29 @@ export function reduceEvent(...@@ -167,19 +179,29 @@ export function reduceEvent(
167 );179 );
168 case "message.sent":180 case "message.sent":
169 return Effect.gen(function* () {181 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 );
171 const targetThread = nextBase.threads.find((thread) => thread.id === payload.threadId);188 const targetThread = nextBase.threads.find((thread) => thread.id === payload.threadId);
172 if (!targetThread) {189 if (!targetThread) {
173 return nextBase;190 return nextBase;
174 }191 }
175192
176 const message: OrchestrationMessage = yield* decodeMessage({193 const message: OrchestrationMessage = yield* decodeForEvent(
177 id: payload.id,194 OrchestrationMessageSchema,
178 role: payload.role,195 {
179 text: payload.text,196 id: payload.id,
180 createdAt: payload.createdAt,197 role: payload.role,
181 streaming: payload.streaming,198 text: payload.text,
182 });199 createdAt: payload.createdAt,
200 streaming: payload.streaming,
201 },
202 event.type,
203 "message",
204 );
183 const existingMessage = targetThread.messages.find((entry) => entry.id === message.id);205 const existingMessage = targetThread.messages.find((entry) => entry.id === message.id);
184 const nextMessages = existingMessage206 const nextMessages = existingMessage
185 ? targetThread.messages.map((entry) =>207 ? targetThread.messages.map((entry) =>
...@@ -208,16 +230,26 @@ export function reduceEvent(...@@ -208,16 +230,26 @@ export function reduceEvent(
208 });230 });
209 case "thread.session-set":231 case "thread.session-set":
210 return Effect.gen(function* () {232 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 );
212 const thread = nextBase.threads.find((entry) => entry.id === payload.threadId);239 const thread = nextBase.threads.find((entry) => entry.id === payload.threadId);
213 if (!thread) {240 if (!thread) {
214 return nextBase;241 return nextBase;
215 }242 }
216243
217 const session: OrchestrationSession = yield* decodeSession({244 const session: OrchestrationSession = yield* decodeForEvent(
218 ...payload.session,245 OrchestrationSessionSchema,
219 threadId: payload.threadId,246 {
220 });247 ...payload.session,
248 threadId: payload.threadId,
249 },
250 event.type,
251 "session",
252 );
221253
222 return {254 return {
223 ...nextBase,255 ...nextBase,
...@@ -230,24 +262,34 @@ export function reduceEvent(...@@ -230,24 +262,34 @@ export function reduceEvent(
230 });262 });
231 case "thread.turn-diff-completed":263 case "thread.turn-diff-completed":
232 return Effect.gen(function* () {264 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 );
234 const thread = nextBase.threads.find((entry) => entry.id === payload.threadId);271 const thread = nextBase.threads.find((entry) => entry.id === payload.threadId);
235 if (!thread) {272 if (!thread) {
236 return nextBase;273 return nextBase;
237 }274 }
238275
239 const nextSummary = yield* decodeTurnDiffSummary({276 const nextSummary = yield* decodeForEvent(
240 turnId: payload.turnId,277 OrchestrationTurnDiffSummarySchema,
241 completedAt: payload.completedAt,278 {
242 ...(payload.status !== undefined ? { status: payload.status } : {}),279 turnId: payload.turnId,
243 files: payload.files,280 completedAt: payload.completedAt,
244 ...(payload.assistantMessageId !== undefined281 ...(payload.status !== undefined ? { status: payload.status } : {}),
245 ? { assistantMessageId: payload.assistantMessageId }282 files: payload.files,
246 : {}),283 ...(payload.assistantMessageId !== undefined
247 ...(payload.checkpointTurnCount !== undefined284 ? { assistantMessageId: payload.assistantMessageId }
248 ? { checkpointTurnCount: payload.checkpointTurnCount }285 : {}),
249 : {}),286 ...(payload.checkpointTurnCount !== undefined
250 });287 ? { checkpointTurnCount: payload.checkpointTurnCount }
288 : {}),
289 },
290 event.type,
291 "turnDiffSummary",
292 );
251 const turnDiffSummaries = [293 const turnDiffSummaries = [
252 ...thread.turnDiffSummaries.filter((summary) => summary.turnId !== payload.turnId),294 ...thread.turnDiffSummaries.filter((summary) => summary.turnId !== payload.turnId),
253 nextSummary,295 nextSummary,
...@@ -264,7 +306,7 @@ export function reduceEvent(...@@ -264,7 +306,7 @@ export function reduceEvent(
264 };306 };
265 });307 });
266 case "thread.reverted":308 case "thread.reverted":
267 return decodeThreadRevertedPayload(event.payload).pipe(309 return decodeForEvent(ThreadRevertedPayloadSchema, event.payload, event.type, "payload").pipe(
268 Effect.map((payload) => {310 Effect.map((payload) => {
269 const thread = nextBase.threads.find((entry) => entry.id === payload.threadId);311 const thread = nextBase.threads.find((entry) => entry.id === payload.threadId);
270 if (!thread) {312 if (!thread) {
...@@ -311,7 +353,12 @@ export function reduceEvent(...@@ -311,7 +353,12 @@ export function reduceEvent(
311 }),353 }),
312 );354 );
313 case "git.read-model-upsert":355 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(
315 Effect.map((gitReadModel: OrchestrationGitReadModel) => ({362 Effect.map((gitReadModel: OrchestrationGitReadModel) => ({
316 ...nextBase,363 ...nextBase,
317 gitByProjectId: {364 gitByProjectId: {
apps/server/src/persistence/Errors.ts created+126
...@@ -0,0 +1,126 @@
1import { ParseResult, Schema } from "effect";
2
3export 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
16export 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
29export 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
42export 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
55export 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
67export 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
79export type ProjectRepositoryError =
80 | PersistenceSqlError
81 | PersistenceDecodeError
82 | PersistenceSerializationError
83 | ProjectValidationError
84 | ProjectPathMissingError
85 | ProjectNotFoundError;
86
87export type OrchestrationEventRepositoryError =
88 | PersistenceSqlError
89 | PersistenceDecodeError
90 | PersistenceSerializationError;
91
92export 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
101export 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
110export 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
119export 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", () => {...@@ -80,9 +80,8 @@ describe("OrchestrationEventRepository", () => {
80 )80 )
81 .get() as { name: string } | undefined;81 .get() as { name: string } | undefined;
82 expect(tableRow?.name).toBeDefined();82 expect(tableRow?.name).toBeDefined();
83 if (!tableRow) {83 expect(tableRow).toBeDefined();
84 throw new Error("Expected migrations table to exist");84 if (!tableRow) return;
85 }
8685
87 const migrationRows = db86 const migrationRows = db
88 .prepare(87 .prepare(
apps/server/src/persistence/Layers/OrchestrationEvents.ts+59-23
...@@ -4,13 +4,20 @@ import * as SqlClient from "@effect/sql/SqlClient";...@@ -4,13 +4,20 @@ import * as SqlClient from "@effect/sql/SqlClient";
4import * as SqlSchema from "@effect/sql/SqlSchema";4import * as SqlSchema from "@effect/sql/SqlSchema";
5import { Effect, Layer, Schema } from "effect";5import { Effect, Layer, Schema } from "effect";
66
7import {
8 toPersistenceDecodeCauseError,
9 toPersistenceDecodeError,
10 toPersistenceSerializationError,
11 toPersistenceSqlError,
12 type OrchestrationEventRepositoryError,
13} from "../Errors";
7import { makeSqlitePersistenceLive } from "./Sqlite";14import { makeSqlitePersistenceLive } from "./Sqlite";
8import {15import {
9 OrchestrationEventRepository,16 OrchestrationEventRepository,
10 type OrchestrationEventRepositoryShape,17 type OrchestrationEventRepositoryShape,
11} from "../Services/OrchestrationEvents";18} from "../Services/OrchestrationEvents";
1219
13const decodeEvent = Schema.decodeUnknownSync(OrchestrationEventSchema);20const decodeEvent = Schema.decodeUnknown(OrchestrationEventSchema);
1421
15const EventRowSchema = Schema.Struct({22const EventRowSchema = Schema.Struct({
16 sequence: Schema.Number,23 sequence: Schema.Number,
...@@ -40,17 +47,27 @@ const ReadFromSequenceRequestSchema = Schema.Struct({...@@ -40,17 +47,27 @@ const ReadFromSequenceRequestSchema = Schema.Struct({
40 limit: Schema.Number,47 limit: Schema.Number,
41});48});
4249
43function eventRowToOrchestrationEvent(row: EventRow): OrchestrationEvent {50function eventRowToOrchestrationEvent(
44 return decodeEvent({51 row: EventRow,
45 sequence: row.sequence,52 operation: string,
46 eventId: row.eventId,53): Effect.Effect<OrchestrationEvent, OrchestrationEventRepositoryError> {
47 type: row.type,54 return Effect.try({
48 aggregateType: row.aggregateType,55 try: () => JSON.parse(row.payloadJson) as unknown,
49 aggregateId: row.aggregateId,56 catch: toPersistenceDecodeCauseError(`${operation}:parsePayloadJson`),
50 occurredAt: row.occurredAt,57 }).pipe(
51 commandId: row.commandId,58 Effect.flatMap((payload) =>
52 payload: JSON.parse(row.payloadJson) as unknown,59 decodeEvent({
53 });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 );
54}71}
5572
56const makeRepository = Effect.gen(function* () {73const makeRepository = Effect.gen(function* () {
...@@ -113,23 +130,42 @@ const makeRepository = Effect.gen(function* () {...@@ -113,23 +130,42 @@ const makeRepository = Effect.gen(function* () {
113 });130 });
114131
115 const append: OrchestrationEventRepositoryShape["append"] = (event) =>132 const append: OrchestrationEventRepositoryShape["append"] = (event) =>
116 appendEventRow({133 Effect.try({
117 eventId: event.eventId,134 try: () => JSON.stringify(event.payload),
118 type: event.type,135 catch: toPersistenceSerializationError("OrchestrationEventRepository.append:encodePayload"),
119 aggregateType: event.aggregateType,136 }).pipe(
120 aggregateId: event.aggregateId,137 Effect.flatMap((payloadJson) =>
121 occurredAt: event.occurredAt,138 appendEventRow({
122 commandId: event.commandId,139 eventId: event.eventId,
123 payloadJson: JSON.stringify(event.payload),140 type: event.type,
124 }).pipe(Effect.map(eventRowToOrchestrationEvent), Effect.orDie);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
126 const readFromSequence: OrchestrationEventRepositoryShape["readFromSequence"] = (155 const readFromSequence: OrchestrationEventRepositoryShape["readFromSequence"] = (
127 sequenceExclusive,156 sequenceExclusive,
128 limit = 1_000,157 limit = 1_000,
129 ) =>158 ) =>
130 readEventRowsFromSequence({ sequenceExclusive, limit }).pipe(159 readEventRowsFromSequence({ sequenceExclusive, limit }).pipe(
131 Effect.map((rows) => rows.map(eventRowToOrchestrationEvent)),160 Effect.mapError(toPersistenceSqlError("OrchestrationEventRepository.readFromSequence:query")),
132 Effect.orDie,161 Effect.flatMap((rows) =>
162 Effect.forEach(rows, (row) =>
163 eventRowToOrchestrationEvent(
164 row,
165 "OrchestrationEventRepository.readFromSequence:rowToEvent",
166 ),
167 ),
168 ),
133 );169 );
134170
135 return {171 return {
apps/server/src/persistence/Layers/Projects.ts+154-37
...@@ -8,6 +8,7 @@ import type {...@@ -8,6 +8,7 @@ import type {
8 ProjectListResult,8 ProjectListResult,
9 ProjectRecord,9 ProjectRecord,
10 ProjectScript,10 ProjectScript,
11 ProjectUpdateScriptsInput,
11 ProjectUpdateScriptsResult,12 ProjectUpdateScriptsResult,
12} from "@t3tools/contracts";13} from "@t3tools/contracts";
13import {14import {
...@@ -22,6 +23,16 @@ import * as SqlClient from "@effect/sql/SqlClient";...@@ -22,6 +23,16 @@ import * as SqlClient from "@effect/sql/SqlClient";
22import * as SqlSchema from "@effect/sql/SqlSchema";23import * as SqlSchema from "@effect/sql/SqlSchema";
23import { Effect, Layer, Option, Schema } from "effect";24import { Effect, Layer, Option, Schema } from "effect";
2425
26import {
27 PersistenceDecodeError,
28 ProjectNotFoundError,
29 ProjectPathMissingError,
30 ProjectValidationError,
31 toPersistenceDecodeCauseError,
32 toPersistenceSerializationError,
33 toPersistenceSqlError,
34 type ProjectRepositoryError,
35} from "../Errors";
25import { ProjectRepository, type ProjectRepositoryShape } from "../Services/Projects";36import { ProjectRepository, type ProjectRepositoryShape } from "../Services/Projects";
2637
27const ProjectRowSchema = Schema.Struct({38const ProjectRowSchema = Schema.Struct({
...@@ -98,25 +109,101 @@ function inferProjectName(cwd: string): string {...@@ -98,25 +109,101 @@ function inferProjectName(cwd: string): string {
98 return name.length > 0 ? name : "project";109 return name.length > 0 ? name : "project";
99}110}
100111
101function parseScriptsJson(raw: string): ProjectScript[] {112function parseProjectScripts(
102 try {113 value: unknown,
103 const parsed = JSON.parse(raw) as unknown;114 operation: string,
104 const scripts = projectScriptsSchema.parse(parsed);115): Effect.Effect<ProjectScript[], PersistenceDecodeError> {
105 return normalizeProjectScripts(scripts);116 const parsed = projectScriptsSchema.safeParse(value);
106 } catch {117 if (!parsed.success) {
107 return [];118 return Effect.fail(
119 new PersistenceDecodeError({
120 operation,
121 issue: parsed.error.message,
122 cause: (parsed.error as { cause?: unknown }).cause,
123 }),
124 );
108 }125 }
126 return Effect.succeed(normalizeProjectScripts(parsed.data));
109}127}
110128
111function rowToProjectRecord(row: ProjectRow): ProjectRecord {129function decodeScriptsJson(
112 return projectRecordSchema.parse({130 raw: string,
113 id: row.id,131 operation: string,
114 cwd: row.cwd,132): Effect.Effect<ProjectScript[], ProjectRepositoryError> {
115 name: row.name,133 return Effect.try({
116 scripts: parseScriptsJson(row.scriptsJson),134 try: () => JSON.parse(raw) as unknown,
117 createdAt: row.createdAt,135 catch: toPersistenceDecodeCauseError(`${operation}:parseScriptsJson`),
118 updatedAt: row.updatedAt,136 }).pipe(Effect.flatMap((value) => parseProjectScripts(value, `${operation}:decodeScripts`)));
119 });137}
138
139function 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
167function 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
181function 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
195function 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);
120}207}
121208
122const makeRepository = Effect.gen(function* () {209const makeRepository = Effect.gen(function* () {
...@@ -231,24 +318,31 @@ const makeRepository = Effect.gen(function* () {...@@ -231,24 +318,31 @@ const makeRepository = Effect.gen(function* () {
231318
232 const list: ProjectRepositoryShape["list"] = () =>319 const list: ProjectRepositoryShape["list"] = () =>
233 listProjectRows(undefined).pipe(320 listProjectRows(undefined).pipe(
234 Effect.map((rows): ProjectListResult => rows.map(rowToProjectRecord)),321 Effect.mapError(toPersistenceSqlError("ProjectRepository.list:query")),
235 Effect.map((projects) =>322 Effect.flatMap((rows) =>
236 projects.map((project) => ({ ...project, scripts: cloneScripts(project.scripts) })),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) })),
237 ),330 ),
238 Effect.orDie,
239 );331 );
240332
241 const add: ProjectRepositoryShape["add"] = (rawInput) =>333 const add: ProjectRepositoryShape["add"] = (rawInput) =>
242 Effect.gen(function* () {334 Effect.gen(function* () {
243 const input = projectAddInputSchema.parse(rawInput as ProjectAddInput);335 const input = yield* decodeAddInput(rawInput);
244 const normalizedCwd = normalizeCwd(input.cwd);336 const normalizedCwd = normalizeCwd(input.cwd);
245 if (!isDirectory(normalizedCwd)) {337 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 }));
247 }339 }
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 );
250 if (Option.isSome(existing)) {344 if (Option.isSome(existing)) {
251 const project = rowToProjectRecord(existing.value);345 const project = yield* rowToProjectRecord(existing.value, "ProjectRepository.add:existing");
252 const result: ProjectAddResult = {346 const result: ProjectAddResult = {
253 project: {347 project: {
254 ...project,348 ...project,
...@@ -267,9 +361,9 @@ const makeRepository = Effect.gen(function* () {...@@ -267,9 +361,9 @@ const makeRepository = Effect.gen(function* () {
267 scriptsJson: "[]",361 scriptsJson: "[]",
268 createdAt: now,362 createdAt: now,
269 updatedAt: now,363 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");
273 const result: ProjectAddResult = {367 const result: ProjectAddResult = {
274 project: {368 project: {
275 ...created,369 ...created,
...@@ -282,27 +376,48 @@ const makeRepository = Effect.gen(function* () {...@@ -282,27 +376,48 @@ const makeRepository = Effect.gen(function* () {
282376
283 const remove: ProjectRepositoryShape["remove"] = (rawInput) =>377 const remove: ProjectRepositoryShape["remove"] = (rawInput) =>
284 Effect.gen(function* () {378 Effect.gen(function* () {
285 const input = projectRemoveInputSchema.parse(rawInput);379 const input = yield* decodeRemoveInput(rawInput);
286 yield* deleteProject({ id: input.id }).pipe(Effect.orDie);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 );
287 });390 });
288391
289 const updateScripts: ProjectRepositoryShape["updateScripts"] = (rawInput) =>392 const updateScripts: ProjectRepositoryShape["updateScripts"] = (rawInput) =>
290 Effect.gen(function* () {393 Effect.gen(function* () {
291 const input = projectUpdateScriptsInputSchema.parse(rawInput);394 const input = yield* decodeUpdateScriptsInput(rawInput);
292 const existing = yield* findProjectById({ id: input.id }).pipe(Effect.orDie);395 const existing = yield* findProjectById({ id: input.id }).pipe(
396 Effect.mapError(toPersistenceSqlError("ProjectRepository.updateScripts:findById")),
397 );
293 if (Option.isNone(existing)) {398 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 }));
295 }400 }
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 });
298 const nextUpdatedAt = new Date().toISOString();410 const nextUpdatedAt = new Date().toISOString();
299 const updatedRow = yield* updateProjectScripts({411 const updatedRow = yield* updateProjectScripts({
300 id: input.id,412 id: input.id,
301 scriptsJson: JSON.stringify(nextScripts),413 scriptsJson,
302 updatedAt: nextUpdatedAt,414 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 );
306 const result: ProjectUpdateScriptsResult = {421 const result: ProjectUpdateScriptsResult = {
307 project: {422 project: {
308 ...updated,423 ...updated,
...@@ -314,17 +429,19 @@ const makeRepository = Effect.gen(function* () {...@@ -314,17 +429,19 @@ const makeRepository = Effect.gen(function* () {
314429
315 const pruneMissing: ProjectRepositoryShape["pruneMissing"] = () =>430 const pruneMissing: ProjectRepositoryShape["pruneMissing"] = () =>
316 listProjectRows(undefined).pipe(431 listProjectRows(undefined).pipe(
432 Effect.mapError(toPersistenceSqlError("ProjectRepository.pruneMissing:query")),
317 Effect.flatMap((rows) =>433 Effect.flatMap((rows) =>
318 Effect.forEach(rows, (row) => {434 Effect.forEach(rows, (row) => {
319 const normalizedCwd = normalizeCwd(row.cwd);435 const normalizedCwd = normalizeCwd(row.cwd);
320 if (isDirectory(normalizedCwd)) {436 if (isDirectory(normalizedCwd)) {
321 return Effect.void;437 return Effect.void;
322 }438 }
323 return deleteProject({ id: row.id }).pipe(Effect.orDie);439 return deleteProject({ id: row.id }).pipe(
440 Effect.mapError(toPersistenceSqlError("ProjectRepository.pruneMissing:delete")),
441 );
324 }),442 }),
325 ),443 ),
326 Effect.asVoid,444 Effect.asVoid,
327 Effect.orDie,
328 );445 );
329446
330 return {447 return {
apps/server/src/persistence/Services/OrchestrationEvents.ts+44-3
...@@ -1,18 +1,59 @@...@@ -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 */
1import type { OrchestrationEvent } from "@t3tools/contracts";9import type { OrchestrationEvent } from "@t3tools/contracts";
2import { Context } from "effect";10import { Context } from "effect";
3import type { Effect } from "effect";11import type { Effect } from "effect";
412
13import type { OrchestrationEventRepositoryError } from "../Errors";
14
5export interface OrchestrationEventRepositoryShape {15export 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 */
6 readonly append: (22 readonly append: (
7 event: Omit<OrchestrationEvent, "sequence">,23 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 */
9 readonly readFromSequence: (33 readonly readFromSequence: (
10 sequenceExclusive: number,34 sequenceExclusive: number,
11 limit?: number,35 limit?: number,
12 ) => Effect.Effect<OrchestrationEvent[]>;36 ) => Effect.Effect<OrchestrationEvent[], OrchestrationEventRepositoryError>;
13 readonly readAll: () => Effect.Effect<OrchestrationEvent[]>;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>;
14}44}
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 */
16export class OrchestrationEventRepository extends Context.Tag("orchestration/EventRepository")<57export class OrchestrationEventRepository extends Context.Tag("orchestration/EventRepository")<
17 OrchestrationEventRepository,58 OrchestrationEventRepository,
18 OrchestrationEventRepositoryShape59 OrchestrationEventRepositoryShape
apps/server/src/persistence/Services/Projects.ts+58-5
...@@ -1,3 +1,11 @@...@@ -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 */
1import type {9import type {
2 ProjectAddInput,10 ProjectAddInput,
3 ProjectAddResult,11 ProjectAddResult,
...@@ -9,16 +17,61 @@ import type {...@@ -9,16 +17,61 @@ import type {
9import { Context } from "effect";17import { Context } from "effect";
10import type { Effect } from "effect";18import type { Effect } from "effect";
1119
20import type { ProjectRepositoryError } from "../Errors";
21
12export interface ProjectRepositoryShape {22export interface ProjectRepositoryShape {
13 readonly list: () => Effect.Effect<ProjectListResult>;23 /**
14 readonly add: (input: ProjectAddInput) => Effect.Effect<ProjectAddResult, Error>;24 * List all persisted projects.
15 readonly remove: (input: ProjectRemoveInput) => Effect.Effect<void, Error>;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 */
16 readonly updateScripts: (52 readonly updateScripts: (
17 input: ProjectUpdateScriptsInput,53 input: ProjectUpdateScriptsInput,
18 ) => Effect.Effect<ProjectUpdateScriptsResult, Error>;54 ) => Effect.Effect<ProjectUpdateScriptsResult, ProjectRepositoryError>;
19 readonly pruneMissing: () => Effect.Effect<void>;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>;
20}62}
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 */
22export class ProjectRepository extends Context.Tag("persistence/ProjectRepository")<75export class ProjectRepository extends Context.Tag("persistence/ProjectRepository")<
23 ProjectRepository,76 ProjectRepository,
24 ProjectRepositoryShape77 ProjectRepositoryShape