From b152462b2cd5f26f40ae174765ede2c62381d282 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sun, 22 Feb 2026 01:43:53 -0800 Subject: [PATCH] more effect: error handling --- apps/server/src/orchestration/Errors.ts | 95 +++++++ apps/server/src/orchestration/Layer.test.ts | 8 +- apps/server/src/orchestration/Layer.ts | 264 +++++++++--------- apps/server/src/orchestration/Service.ts | 76 ++++- apps/server/src/orchestration/reducer.ts | 179 +++++++----- apps/server/src/persistence/Errors.ts | 126 +++++++++ .../Layers/OrchestrationEvents.test.ts | 5 +- .../persistence/Layers/OrchestrationEvents.ts | 82 ++++-- .../server/src/persistence/Layers/Projects.ts | 191 ++++++++++--- .../Services/OrchestrationEvents.ts | 47 +++- .../src/persistence/Services/Projects.ts | 63 ++++- 11 files changed, 861 insertions(+), 275 deletions(-) create mode 100644 apps/server/src/orchestration/Errors.ts create mode 100644 apps/server/src/persistence/Errors.ts diff --git a/apps/server/src/orchestration/Errors.ts b/apps/server/src/orchestration/Errors.ts new file mode 100644 index 0000000000000000000000000000000000000000..0f199e2baf45ea632e195fee0abf5bceb403500c --- /dev/null +++ b/apps/server/src/orchestration/Errors.ts @@ -0,0 +1,95 @@ +import { ParseResult, Schema } from "effect"; + +import type { OrchestrationEventRepositoryError } from "../persistence/Errors"; + +export class OrchestrationCommandJsonParseError extends Schema.TaggedError()( + "OrchestrationCommandJsonParseError", + { + detail: Schema.String, + cause: Schema.optional(Schema.Defect), + }, +) { + override get message(): string { + return `Invalid orchestration command JSON: ${this.detail}`; + } +} + +export class OrchestrationCommandDecodeError extends Schema.TaggedError()( + "OrchestrationCommandDecodeError", + { + issue: Schema.String, + cause: Schema.optional(Schema.Defect), + }, +) { + override get message(): string { + return `Invalid orchestration command payload: ${this.issue}`; + } +} + +export class OrchestrationReducerDecodeError extends Schema.TaggedError()( + "OrchestrationReducerDecodeError", + { + eventType: Schema.String, + issue: Schema.String, + cause: Schema.optional(Schema.Defect), + }, +) { + override get message(): string { + return `Reducer decode failed for ${this.eventType}: ${this.issue}`; + } +} + +export class OrchestrationListenerCallbackError extends Schema.TaggedError()( + "OrchestrationListenerCallbackError", + { + listener: Schema.Literal("read-model", "domain-event"), + detail: Schema.String, + cause: Schema.optional(Schema.Defect), + }, +) { + override get message(): string { + return `Orchestration ${this.listener} listener failed: ${this.detail}`; + } +} + +export type OrchestrationDispatchError = + | OrchestrationEventRepositoryError + | OrchestrationReducerDecodeError + | OrchestrationListenerCallbackError; + +export type OrchestrationEngineError = + | OrchestrationDispatchError + | OrchestrationCommandJsonParseError + | OrchestrationCommandDecodeError; + +export function toOrchestrationCommandDecodeError(error: ParseResult.ParseError) { + return new OrchestrationCommandDecodeError({ + issue: ParseResult.TreeFormatter.formatErrorSync(error), + cause: error, + }); +} + +export function toReducerDecodeError(eventType: string) { + return (error: ParseResult.ParseError): OrchestrationReducerDecodeError => + new OrchestrationReducerDecodeError({ + eventType, + issue: ParseResult.TreeFormatter.formatErrorSync(error), + cause: error, + }); +} + +export function toOrchestrationJsonParseError(cause: unknown) { + return new OrchestrationCommandJsonParseError({ + detail: `Failed to parse orchestration command JSON: ${cause}`, + cause, + }); +} + +export function toListenerCallbackError(listener: "read-model" | "domain-event") { + return (cause: unknown): OrchestrationListenerCallbackError => + new OrchestrationListenerCallbackError({ + listener, + detail: `Failed to invoke orchestration ${listener} listener: ${cause}`, + cause, + }); +} diff --git a/apps/server/src/orchestration/Layer.test.ts b/apps/server/src/orchestration/Layer.test.ts index 87afdcea4c38d1385ed23e4d5d56275150eae1bf..76a028567a8bada6d6c8f286770858676184064b 100644 --- a/apps/server/src/orchestration/Layer.test.ts +++ b/apps/server/src/orchestration/Layer.test.ts @@ -10,6 +10,7 @@ import { OrchestrationEventRepository, type OrchestrationEventRepositoryShape, } from "../persistence/Services/OrchestrationEvents"; +import { PersistenceSqlError } from "../persistence/Errors"; import { Layer, ManagedRuntime } from "effect"; @@ -375,7 +376,12 @@ describe("OrchestrationEngine", () => { append(event) { if (shouldFailFirstAppend) { shouldFailFirstAppend = false; - return Effect.die(new Error("append failed")); + return Effect.fail( + new PersistenceSqlError({ + operation: "test.append", + detail: "append failed", + }), + ); } const savedEvent = { ...event, diff --git a/apps/server/src/orchestration/Layer.ts b/apps/server/src/orchestration/Layer.ts index 9974132fee21b7aa2ceebba925f897e1eb5568f3..3c9590501c7eb32b699c8fcbd5efbd07e97898fe 100644 --- a/apps/server/src/orchestration/Layer.ts +++ b/apps/server/src/orchestration/Layer.ts @@ -4,33 +4,24 @@ import type { OrchestrationReadModel, } from "@t3tools/contracts"; import { OrchestrationCommandSchema } from "@t3tools/contracts"; -import { - Cause, - Deferred, - Effect, - Either, - Fiber, - Layer, - PubSub, - Queue, - Runtime, - Schema, -} from "effect"; +import { Deferred, Effect, Layer, PubSub, Queue, Schema } from "effect"; import { createLogger } from "../logger"; import { OrchestrationEventRepository } from "../persistence/Services/OrchestrationEvents"; +import { + toListenerCallbackError, + toOrchestrationCommandDecodeError, + toOrchestrationJsonParseError, + type OrchestrationDispatchError, +} from "./Errors"; import { createEmptyReadModel, reduceEvent } from "./reducer"; import { OrchestrationEngineService, type OrchestrationEngineShape } from "./Service"; type CommandEnvelope = { command: OrchestrationCommand; - result: Deferred.Deferred<{ sequence: number }, Error>; + result: Deferred.Deferred<{ sequence: number }, OrchestrationDispatchError>; }; -function asError(error: unknown, fallbackMessage: string): Error { - return error instanceof Error ? error : new Error(fallbackMessage); -} - function mapCommandToEvent(command: OrchestrationCommand): Omit { const eventId = crypto.randomUUID(); switch (command.type) { @@ -169,129 +160,134 @@ function mapCommandToEvent(command: OrchestrationCommand): Omit(); - const readModelPubSub = yield* PubSub.unbounded(); - const eventPubSub = yield* PubSub.unbounded(); - - const readModelListeners = new Set<(snapshot: OrchestrationReadModel) => void>(); - const domainEventListeners = new Set<(event: OrchestrationEvent) => void>(); - - const processEnvelope = (envelope: CommandEnvelope): Effect.Effect => - Effect.gen(function* () { - const eventBase = mapCommandToEvent(envelope.command); - const savedEvent = yield* eventStore.append(eventBase); - readModel = yield* reduceEvent(readModel, savedEvent); - - const snapshot = readModel; - yield* Effect.all([ - PubSub.publish(eventPubSub, savedEvent), - PubSub.publish(readModelPubSub, snapshot), - ]); - - for (const listener of domainEventListeners) { - listener(savedEvent); - } - for (const listener of readModelListeners) { - listener(snapshot); - } - - yield* Deferred.succeed(envelope.result, { sequence: savedEvent.sequence }); - }).pipe( - Effect.catchAllCause((cause) => - Deferred.fail( - envelope.result, - asError(Cause.squash(cause), "Unknown command processing error"), - ).pipe(Effect.asVoid), - ), - ); - - const bootstrapReadModel: Effect.Effect = Effect.gen(function* () { - const existingEvents = yield* eventStore.readAll(); - for (const event of existingEvents) { - readModel = yield* reduceEvent(readModel, event); - } - }); - - yield* bootstrapReadModel; - - const worker = Effect.forever(Queue.take(commandQueue).pipe(Effect.flatMap(processEnvelope))); - const workerFiber: Fiber.RuntimeFiber = Runtime.runFork(runtime)(worker); - logger.info("orchestration engine started", { - sequence: readModel.sequence, - }); - - yield* Effect.addFinalizer(() => Fiber.interrupt(workerFiber).pipe(Effect.orDie)); - - const getSnapshot: OrchestrationEngineShape["getSnapshot"] = () => - Effect.sync((): OrchestrationReadModel => readModel); - - const replayEvents: OrchestrationEngineShape["replayEvents"] = (fromSequenceExclusive) => - eventStore.readFromSequence(fromSequenceExclusive); - - const dispatch: OrchestrationEngineShape["dispatch"] = (command) => - Effect.gen(function* () { - const result = yield* Deferred.make<{ sequence: number }, Error>(); - yield* Queue.offer(commandQueue, { command, result }); - return yield* Deferred.await(result); - }).pipe(Effect.mapError((error) => asError(error, "Queue offer failed"))); - - const dispatchUnknown: OrchestrationEngineShape["dispatchUnknown"] = (command) => - Effect.gen(function* () { - let payload: unknown = command; - if (typeof command === "string") { - try { - payload = JSON.parse(command) as unknown; - } catch { - return yield* Effect.fail( - new Error("Invalid orchestration command: payload is not valid JSON"), - ); +const decodeUnknownCommand = Schema.decodeUnknown(OrchestrationCommandSchema); + +const makeOrchestrationEngine = () => + Effect.gen(function* () { + const logger = createLogger("orchestration"); + const eventStore = yield* OrchestrationEventRepository; + + let readModel = createEmptyReadModel(new Date().toISOString()); + + const commandQueue = yield* Queue.unbounded(); + const readModelPubSub = yield* PubSub.unbounded(); + const eventPubSub = yield* PubSub.unbounded(); + + const readModelListeners = new Set<(snapshot: OrchestrationReadModel) => void>(); + const domainEventListeners = new Set<(event: OrchestrationEvent) => void>(); + + const notifyReadModelListeners = (snapshot: OrchestrationReadModel) => + Effect.forEach(readModelListeners, (listener) => + Effect.try({ + try: () => listener(snapshot), + catch: toListenerCallbackError("read-model"), + }), + ).pipe(Effect.asVoid); + + const notifyDomainEventListeners = (event: OrchestrationEvent) => + Effect.forEach(domainEventListeners, (listener) => + Effect.try({ + try: () => listener(event), + catch: toListenerCallbackError("domain-event"), + }), + ).pipe(Effect.asVoid); + + const processEnvelope = (envelope: CommandEnvelope): Effect.Effect => + Effect.gen(function* () { + const eventBase = mapCommandToEvent(envelope.command); + const savedEvent = yield* eventStore.append(eventBase); + readModel = yield* reduceEvent(readModel, savedEvent); + + const snapshot = readModel; + yield* Effect.all([ + PubSub.publish(eventPubSub, savedEvent), + PubSub.publish(readModelPubSub, snapshot), + ]); + + yield* notifyDomainEventListeners(savedEvent); + yield* notifyReadModelListeners(snapshot); + + yield* Deferred.succeed(envelope.result, { sequence: savedEvent.sequence }); + }).pipe( + Effect.catchAll((error) => Deferred.fail(envelope.result, error).pipe(Effect.asVoid)), + ); + + const bootstrapReadModel: Effect.Effect = Effect.gen( + function* () { + const existingEvents = yield* eventStore.readAll(); + for (const event of existingEvents) { + readModel = yield* reduceEvent(readModel, event); } - } - const decoded = decodeUnknownCommand(payload); - if (Either.isLeft(decoded)) { - return yield* Effect.fail( - new Error(`Invalid orchestration command: ${decoded.left.toString()}`), + }, + ); + + yield* bootstrapReadModel; + + const worker = Effect.forever(Queue.take(commandQueue).pipe(Effect.flatMap(processEnvelope))); + yield* Effect.forkScoped(worker); + logger.info("orchestration engine started", { + sequence: readModel.sequence, + }); + + const getSnapshot: OrchestrationEngineShape["getSnapshot"] = () => + Effect.sync((): OrchestrationReadModel => readModel); + + const replayEvents: OrchestrationEngineShape["replayEvents"] = (fromSequenceExclusive) => + eventStore.readFromSequence(fromSequenceExclusive); + + const dispatch: OrchestrationEngineShape["dispatch"] = (command) => + Effect.gen(function* () { + const result = yield* Deferred.make<{ sequence: number }, OrchestrationDispatchError>(); + yield* Queue.offer(commandQueue, { command, result }); + return yield* Deferred.await(result); + }); + + const dispatchUnknown: OrchestrationEngineShape["dispatchUnknown"] = (command) => + Effect.gen(function* () { + const payload = + typeof command === "string" + ? yield* Effect.try({ + try: () => JSON.parse(command) as unknown, + catch: toOrchestrationJsonParseError, + }) + : command; + + const decoded = yield* decodeUnknownCommand(payload).pipe( + Effect.mapError(toOrchestrationCommandDecodeError), ); - } - return yield* dispatch(decoded.right); - }); - const subscribeToReadModel: OrchestrationEngineShape["subscribeToReadModel"] = (callback) => - Effect.sync(() => { - readModelListeners.add(callback); - return () => { - readModelListeners.delete(callback); - }; - }); + return yield* dispatch(decoded); + }); - const subscribeToDomainEvents: OrchestrationEngineShape["subscribeToDomainEvents"] = (callback) => - Effect.sync(() => { - domainEventListeners.add(callback); - return () => { - domainEventListeners.delete(callback); - }; - }); + const subscribeToReadModel: OrchestrationEngineShape["subscribeToReadModel"] = (callback) => + Effect.sync(() => { + readModelListeners.add(callback); + return () => { + readModelListeners.delete(callback); + }; + }); - return { - getSnapshot, - replayEvents, - dispatchUnknown, - dispatch, - subscribeToReadModel, - subscribeToDomainEvents, - } satisfies OrchestrationEngineShape; -}); + const subscribeToDomainEvents: OrchestrationEngineShape["subscribeToDomainEvents"] = ( + callback, + ) => + Effect.sync(() => { + domainEventListeners.add(callback); + return () => { + domainEventListeners.delete(callback); + }; + }); + + return { + getSnapshot, + replayEvents, + dispatchUnknown, + dispatch, + subscribeToReadModel, + subscribeToDomainEvents, + } satisfies OrchestrationEngineShape; + }); export const OrchestrationEngineLive = Layer.scoped( OrchestrationEngineService, - makeOrchestrationEngine, + makeOrchestrationEngine(), ); diff --git a/apps/server/src/orchestration/Service.ts b/apps/server/src/orchestration/Service.ts index 9959f897a4cfa0e6de8ee66d0b5405789892c14b..57eee9dabff917b28a6ab5a8d9a98372b68cbb77 100644 --- a/apps/server/src/orchestration/Service.ts +++ b/apps/server/src/orchestration/Service.ts @@ -1,3 +1,11 @@ +/** + * OrchestrationEngineService - Service interface for orchestration command handling. + * + * Uses Effect `Context.Tag` for dependency injection. Command dispatch, + * replay, and unknown-input decoding all return typed domain errors. + * + * @module OrchestrationEngineService + */ import type { OrchestrationCommand, OrchestrationEvent, @@ -6,19 +14,81 @@ import type { import { Context } from "effect"; import type { Effect } from "effect"; +import type { OrchestrationDispatchError, OrchestrationEngineError } from "./Errors"; +import type { OrchestrationEventRepositoryError } from "../persistence/Errors"; + export interface OrchestrationEngineShape { + /** + * Read the current in-memory orchestration snapshot. + * + * @returns Effect containing the latest read model. + */ readonly getSnapshot: () => Effect.Effect; - readonly replayEvents: (fromSequenceExclusive: number) => Effect.Effect; - readonly dispatchUnknown: (command: unknown) => Effect.Effect<{ sequence: number }, Error>; - readonly dispatch: (command: OrchestrationCommand) => Effect.Effect<{ sequence: number }, Error>; + + /** + * Replay persisted orchestration events from an exclusive sequence cursor. + * + * @param fromSequenceExclusive - Sequence cursor (exclusive). + * @returns Effect containing ordered events. + */ + readonly replayEvents: ( + fromSequenceExclusive: number, + ) => Effect.Effect; + + /** + * Validate and dispatch an unknown command payload. + * + * Accepts either parsed values or raw JSON strings. + * + * @param command - Unknown command payload. + * @returns Effect containing the sequence of the persisted event. + */ + readonly dispatchUnknown: ( + command: unknown, + ) => Effect.Effect<{ sequence: number }, OrchestrationEngineError>; + + /** + * Dispatch a validated orchestration command. + * + * @param command - Valid orchestration command. + * @returns Effect containing the sequence of the persisted event. + */ + readonly dispatch: ( + command: OrchestrationCommand, + ) => Effect.Effect<{ sequence: number }, OrchestrationDispatchError>; + + /** + * Subscribe to read model updates. + * + * @param callback - Handler invoked on every read model update. + * @returns Effect containing an unsubscribe function. + */ readonly subscribeToReadModel: ( callback: (snapshot: OrchestrationReadModel) => void, ) => Effect.Effect<() => void>; + + /** + * Subscribe to domain event fan-out. + * + * @param callback - Handler invoked for every persisted domain event. + * @returns Effect containing an unsubscribe function. + */ readonly subscribeToDomainEvents: ( callback: (event: OrchestrationEvent) => void, ) => Effect.Effect<() => void>; } +/** + * OrchestrationEngineService - Context tag for orchestration engine access. + * + * @example + * ```ts + * const program = Effect.gen(function* () { + * const engine = yield* OrchestrationEngineService + * return yield* engine.getSnapshot() + * }) + * ``` + */ export class OrchestrationEngineService extends Context.Tag("orchestration/Engine")< OrchestrationEngineService, OrchestrationEngineShape diff --git a/apps/server/src/orchestration/reducer.ts b/apps/server/src/orchestration/reducer.ts index 1b05441b1f5e7ce6aa89e37081a8a0bf94f1961d..01f9e8c09755bf9cb5bf3b57ea81a35d4b8fb7c3 100644 --- a/apps/server/src/orchestration/reducer.ts +++ b/apps/server/src/orchestration/reducer.ts @@ -9,7 +9,6 @@ import type { import { OrchestrationGitReadModelSchema, OrchestrationMessageSchema, - OrchestrationReadModelSchema, OrchestrationSessionSchema, OrchestrationThreadSchema, OrchestrationTurnDiffFileSchema, @@ -17,6 +16,8 @@ import { } from "@t3tools/contracts"; import { Effect, Schema } from "effect"; +import { toReducerDecodeError, type OrchestrationReducerDecodeError } from "./Errors"; + type ThreadPatch = Partial>; function updateThread( @@ -27,12 +28,16 @@ function updateThread( return threads.map((thread) => (thread.id === threadId ? { ...thread, ...patch } : thread)); } -const decodeReadModelSync = Schema.decodeUnknownSync(OrchestrationReadModelSchema); -const decodeMessage = Schema.decodeUnknown(OrchestrationMessageSchema); -const decodeSession = Schema.decodeUnknown(OrchestrationSessionSchema); -const decodeThread = Schema.decodeUnknown(OrchestrationThreadSchema); -const decodeGitReadModel = Schema.decodeUnknown(OrchestrationGitReadModelSchema); -const decodeTurnDiffSummary = Schema.decodeUnknown(OrchestrationTurnDiffSummarySchema); +function decodeForEvent( + schema: Schema.Schema, + value: unknown, + eventType: string, + field: string, +): Effect.Effect { + return Schema.decodeUnknown(schema)(value).pipe( + Effect.mapError(toReducerDecodeError(`${eventType}:${field}`)), + ); +} const ThreadCreatedPayloadSchema = Schema.Struct({ id: Schema.String, @@ -44,7 +49,6 @@ const ThreadCreatedPayloadSchema = Schema.Struct({ createdAt: Schema.String, updatedAt: Schema.String, }); -const decodeThreadCreatedPayload = Schema.decodeUnknown(ThreadCreatedPayloadSchema); const ThreadMetaUpdatedPayloadSchema = Schema.Struct({ threadId: Schema.String, @@ -54,7 +58,6 @@ const ThreadMetaUpdatedPayloadSchema = Schema.Struct({ worktreePath: Schema.optional(Schema.NullOr(Schema.String)), updatedAt: Schema.String, }); -const decodeThreadMetaUpdatedPayload = Schema.decodeUnknown(ThreadMetaUpdatedPayloadSchema); const MessageSentPayloadSchema = Schema.Struct({ id: Schema.String, @@ -64,13 +67,11 @@ const MessageSentPayloadSchema = Schema.Struct({ createdAt: Schema.String, streaming: Schema.Boolean, }); -const decodeMessageSentPayload = Schema.decodeUnknown(MessageSentPayloadSchema); const ThreadSessionSetPayloadSchema = Schema.Struct({ threadId: Schema.String, session: OrchestrationSessionSchema, }); -const decodeThreadSessionSetPayload = Schema.decodeUnknown(ThreadSessionSetPayloadSchema); const ThreadTurnDiffCompletedPayloadSchema = Schema.Struct({ threadId: Schema.String, @@ -81,30 +82,26 @@ const ThreadTurnDiffCompletedPayloadSchema = Schema.Struct({ assistantMessageId: Schema.optional(Schema.String), checkpointTurnCount: Schema.optional(Schema.Number), }); -const decodeThreadTurnDiffCompletedPayload = Schema.decodeUnknown( - ThreadTurnDiffCompletedPayloadSchema, -); const ThreadRevertedPayloadSchema = Schema.Struct({ threadId: Schema.String, turnCount: Schema.Number, messageCount: Schema.Number, }); -const decodeThreadRevertedPayload = Schema.decodeUnknown(ThreadRevertedPayloadSchema); export function createEmptyReadModel(nowIso: string): OrchestrationReadModel { - return decodeReadModelSync({ + return { sequence: 0, threads: [], gitByProjectId: {}, updatedAt: nowIso, - }); + }; } export function reduceEvent( model: OrchestrationReadModel, event: OrchestrationEvent, -): Effect.Effect { +): Effect.Effect { const nextBase: OrchestrationReadModel = { ...model, sequence: event.sequence, @@ -114,25 +111,35 @@ export function reduceEvent( switch (event.type) { case "thread.created": return Effect.gen(function* () { - const payload = yield* decodeThreadCreatedPayload(event.payload); - const thread: OrchestrationThread = yield* decodeThread({ - id: payload.id, - projectId: payload.projectId, - title: payload.title, - model: payload.model, - branch: payload.branch, - worktreePath: payload.worktreePath, - createdAt: payload.createdAt, - updatedAt: payload.updatedAt, - latestTurnId: null, - latestTurnStartedAt: null, - latestTurnCompletedAt: null, - latestTurnDurationMs: null, - messages: [], - session: null, - turnDiffSummaries: [], - error: null, - }); + const payload = yield* decodeForEvent( + ThreadCreatedPayloadSchema, + event.payload, + event.type, + "payload", + ); + const thread: OrchestrationThread = yield* decodeForEvent( + OrchestrationThreadSchema, + { + id: payload.id, + projectId: payload.projectId, + title: payload.title, + model: payload.model, + branch: payload.branch, + worktreePath: payload.worktreePath, + createdAt: payload.createdAt, + updatedAt: payload.updatedAt, + latestTurnId: null, + latestTurnStartedAt: null, + latestTurnCompletedAt: null, + latestTurnDurationMs: null, + messages: [], + session: null, + turnDiffSummaries: [], + error: null, + }, + event.type, + "thread", + ); const existing = nextBase.threads.find((entry) => entry.id === thread.id); return { ...nextBase, @@ -147,7 +154,12 @@ export function reduceEvent( threads: nextBase.threads.filter((thread) => thread.id !== event.aggregateId), }); case "thread.meta-updated": - return decodeThreadMetaUpdatedPayload(event.payload).pipe( + return decodeForEvent( + ThreadMetaUpdatedPayloadSchema, + event.payload, + event.type, + "payload", + ).pipe( Effect.map((payload) => { const thread = nextBase.threads.find((entry) => entry.id === payload.threadId); if (!thread) { @@ -167,19 +179,29 @@ export function reduceEvent( ); case "message.sent": return Effect.gen(function* () { - const payload = yield* decodeMessageSentPayload(event.payload); + const payload = yield* decodeForEvent( + MessageSentPayloadSchema, + event.payload, + event.type, + "payload", + ); const targetThread = nextBase.threads.find((thread) => thread.id === payload.threadId); if (!targetThread) { return nextBase; } - const message: OrchestrationMessage = yield* decodeMessage({ - id: payload.id, - role: payload.role, - text: payload.text, - createdAt: payload.createdAt, - streaming: payload.streaming, - }); + const message: OrchestrationMessage = yield* decodeForEvent( + OrchestrationMessageSchema, + { + id: payload.id, + role: payload.role, + text: payload.text, + createdAt: payload.createdAt, + streaming: payload.streaming, + }, + event.type, + "message", + ); const existingMessage = targetThread.messages.find((entry) => entry.id === message.id); const nextMessages = existingMessage ? targetThread.messages.map((entry) => @@ -208,16 +230,26 @@ export function reduceEvent( }); case "thread.session-set": return Effect.gen(function* () { - const payload = yield* decodeThreadSessionSetPayload(event.payload); + const payload = yield* decodeForEvent( + ThreadSessionSetPayloadSchema, + event.payload, + event.type, + "payload", + ); const thread = nextBase.threads.find((entry) => entry.id === payload.threadId); if (!thread) { return nextBase; } - const session: OrchestrationSession = yield* decodeSession({ - ...payload.session, - threadId: payload.threadId, - }); + const session: OrchestrationSession = yield* decodeForEvent( + OrchestrationSessionSchema, + { + ...payload.session, + threadId: payload.threadId, + }, + event.type, + "session", + ); return { ...nextBase, @@ -230,24 +262,34 @@ export function reduceEvent( }); case "thread.turn-diff-completed": return Effect.gen(function* () { - const payload = yield* decodeThreadTurnDiffCompletedPayload(event.payload); + const payload = yield* decodeForEvent( + ThreadTurnDiffCompletedPayloadSchema, + event.payload, + event.type, + "payload", + ); const thread = nextBase.threads.find((entry) => entry.id === payload.threadId); if (!thread) { return nextBase; } - const nextSummary = yield* decodeTurnDiffSummary({ - turnId: payload.turnId, - completedAt: payload.completedAt, - ...(payload.status !== undefined ? { status: payload.status } : {}), - files: payload.files, - ...(payload.assistantMessageId !== undefined - ? { assistantMessageId: payload.assistantMessageId } - : {}), - ...(payload.checkpointTurnCount !== undefined - ? { checkpointTurnCount: payload.checkpointTurnCount } - : {}), - }); + const nextSummary = yield* decodeForEvent( + OrchestrationTurnDiffSummarySchema, + { + turnId: payload.turnId, + completedAt: payload.completedAt, + ...(payload.status !== undefined ? { status: payload.status } : {}), + files: payload.files, + ...(payload.assistantMessageId !== undefined + ? { assistantMessageId: payload.assistantMessageId } + : {}), + ...(payload.checkpointTurnCount !== undefined + ? { checkpointTurnCount: payload.checkpointTurnCount } + : {}), + }, + event.type, + "turnDiffSummary", + ); const turnDiffSummaries = [ ...thread.turnDiffSummaries.filter((summary) => summary.turnId !== payload.turnId), nextSummary, @@ -264,7 +306,7 @@ export function reduceEvent( }; }); case "thread.reverted": - return decodeThreadRevertedPayload(event.payload).pipe( + return decodeForEvent(ThreadRevertedPayloadSchema, event.payload, event.type, "payload").pipe( Effect.map((payload) => { const thread = nextBase.threads.find((entry) => entry.id === payload.threadId); if (!thread) { @@ -311,7 +353,12 @@ export function reduceEvent( }), ); case "git.read-model-upsert": - return decodeGitReadModel(event.payload).pipe( + return decodeForEvent( + OrchestrationGitReadModelSchema, + event.payload, + event.type, + "payload", + ).pipe( Effect.map((gitReadModel: OrchestrationGitReadModel) => ({ ...nextBase, gitByProjectId: { diff --git a/apps/server/src/persistence/Errors.ts b/apps/server/src/persistence/Errors.ts new file mode 100644 index 0000000000000000000000000000000000000000..940357abe34a19f350bcdd64eec8ed6e9cc974d1 --- /dev/null +++ b/apps/server/src/persistence/Errors.ts @@ -0,0 +1,126 @@ +import { ParseResult, Schema } from "effect"; + +export class PersistenceSqlError extends Schema.TaggedError()( + "PersistenceSqlError", + { + operation: Schema.String, + detail: Schema.String, + cause: Schema.optional(Schema.Defect), + }, +) { + override get message(): string { + return `SQL error in ${this.operation}: ${this.detail}`; + } +} + +export class PersistenceDecodeError extends Schema.TaggedError()( + "PersistenceDecodeError", + { + operation: Schema.String, + issue: Schema.String, + cause: Schema.optional(Schema.Defect), + }, +) { + override get message(): string { + return `Decode error in ${this.operation}: ${this.issue}`; + } +} + +export class PersistenceSerializationError extends Schema.TaggedError()( + "PersistenceSerializationError", + { + operation: Schema.String, + detail: Schema.String, + cause: Schema.optional(Schema.Defect), + }, +) { + override get message(): string { + return `Serialization error in ${this.operation}: ${this.detail}`; + } +} + +export class ProjectValidationError extends Schema.TaggedError()( + "ProjectValidationError", + { + operation: Schema.String, + issue: Schema.String, + cause: Schema.optional(Schema.Defect), + }, +) { + override get message(): string { + return `Project validation failed in ${this.operation}: ${this.issue}`; + } +} + +export class ProjectPathMissingError extends Schema.TaggedError()( + "ProjectPathMissingError", + { + cwd: Schema.String, + cause: Schema.optional(Schema.Defect), + }, +) { + override get message(): string { + return `Project path does not exist: ${this.cwd}`; + } +} + +export class ProjectNotFoundError extends Schema.TaggedError()( + "ProjectNotFoundError", + { + projectId: Schema.String, + cause: Schema.optional(Schema.Defect), + }, +) { + override get message(): string { + return `Project not found: ${this.projectId}`; + } +} + +export type ProjectRepositoryError = + | PersistenceSqlError + | PersistenceDecodeError + | PersistenceSerializationError + | ProjectValidationError + | ProjectPathMissingError + | ProjectNotFoundError; + +export type OrchestrationEventRepositoryError = + | PersistenceSqlError + | PersistenceDecodeError + | PersistenceSerializationError; + +export function toPersistenceSqlError(operation: string) { + return (cause: unknown): PersistenceSqlError => + new PersistenceSqlError({ + operation, + detail: `Failed to execute ${operation}`, + cause, + }); +} + +export function toPersistenceDecodeError(operation: string) { + return (error: ParseResult.ParseError): PersistenceDecodeError => + new PersistenceDecodeError({ + operation, + issue: ParseResult.TreeFormatter.formatErrorSync(error), + cause: error, + }); +} + +export function toPersistenceDecodeCauseError(operation: string) { + return (cause: unknown): PersistenceDecodeError => + new PersistenceDecodeError({ + operation, + issue: `Failed to execute ${operation}`, + cause, + }); +} + +export function toPersistenceSerializationError(operation: string) { + return (cause: unknown): PersistenceSerializationError => + new PersistenceSerializationError({ + operation, + detail: `Failed to execute ${operation}`, + cause, + }); +} diff --git a/apps/server/src/persistence/Layers/OrchestrationEvents.test.ts b/apps/server/src/persistence/Layers/OrchestrationEvents.test.ts index b9768172b286a07ae63b6b153f8a4751081de42b..e9ca28dff0426c196f9b7dba869fa61e1df7b391 100644 --- a/apps/server/src/persistence/Layers/OrchestrationEvents.test.ts +++ b/apps/server/src/persistence/Layers/OrchestrationEvents.test.ts @@ -80,9 +80,8 @@ describe("OrchestrationEventRepository", () => { ) .get() as { name: string } | undefined; expect(tableRow?.name).toBeDefined(); - if (!tableRow) { - throw new Error("Expected migrations table to exist"); - } + expect(tableRow).toBeDefined(); + if (!tableRow) return; const migrationRows = db .prepare( diff --git a/apps/server/src/persistence/Layers/OrchestrationEvents.ts b/apps/server/src/persistence/Layers/OrchestrationEvents.ts index 29f256582697364e481d742ef0c62218c751e30d..d3a6841105907dd5d9bbd79b970330bcb195c455 100644 --- a/apps/server/src/persistence/Layers/OrchestrationEvents.ts +++ b/apps/server/src/persistence/Layers/OrchestrationEvents.ts @@ -4,13 +4,20 @@ import * as SqlClient from "@effect/sql/SqlClient"; import * as SqlSchema from "@effect/sql/SqlSchema"; import { Effect, Layer, Schema } from "effect"; +import { + toPersistenceDecodeCauseError, + toPersistenceDecodeError, + toPersistenceSerializationError, + toPersistenceSqlError, + type OrchestrationEventRepositoryError, +} from "../Errors"; import { makeSqlitePersistenceLive } from "./Sqlite"; import { OrchestrationEventRepository, type OrchestrationEventRepositoryShape, } from "../Services/OrchestrationEvents"; -const decodeEvent = Schema.decodeUnknownSync(OrchestrationEventSchema); +const decodeEvent = Schema.decodeUnknown(OrchestrationEventSchema); const EventRowSchema = Schema.Struct({ sequence: Schema.Number, @@ -40,17 +47,27 @@ const ReadFromSequenceRequestSchema = Schema.Struct({ limit: Schema.Number, }); -function eventRowToOrchestrationEvent(row: EventRow): OrchestrationEvent { - return decodeEvent({ - sequence: row.sequence, - eventId: row.eventId, - type: row.type, - aggregateType: row.aggregateType, - aggregateId: row.aggregateId, - occurredAt: row.occurredAt, - commandId: row.commandId, - payload: JSON.parse(row.payloadJson) as unknown, - }); +function eventRowToOrchestrationEvent( + row: EventRow, + operation: string, +): Effect.Effect { + return Effect.try({ + try: () => JSON.parse(row.payloadJson) as unknown, + catch: toPersistenceDecodeCauseError(`${operation}:parsePayloadJson`), + }).pipe( + Effect.flatMap((payload) => + decodeEvent({ + sequence: row.sequence, + eventId: row.eventId, + type: row.type, + aggregateType: row.aggregateType, + aggregateId: row.aggregateId, + occurredAt: row.occurredAt, + commandId: row.commandId, + payload, + }).pipe(Effect.mapError(toPersistenceDecodeError(`${operation}:decodeEvent`))), + ), + ); } const makeRepository = Effect.gen(function* () { @@ -113,23 +130,42 @@ const makeRepository = Effect.gen(function* () { }); const append: OrchestrationEventRepositoryShape["append"] = (event) => - appendEventRow({ - eventId: event.eventId, - type: event.type, - aggregateType: event.aggregateType, - aggregateId: event.aggregateId, - occurredAt: event.occurredAt, - commandId: event.commandId, - payloadJson: JSON.stringify(event.payload), - }).pipe(Effect.map(eventRowToOrchestrationEvent), Effect.orDie); + Effect.try({ + try: () => JSON.stringify(event.payload), + catch: toPersistenceSerializationError("OrchestrationEventRepository.append:encodePayload"), + }).pipe( + Effect.flatMap((payloadJson) => + appendEventRow({ + eventId: event.eventId, + type: event.type, + aggregateType: event.aggregateType, + aggregateId: event.aggregateId, + occurredAt: event.occurredAt, + commandId: event.commandId, + payloadJson, + }).pipe( + Effect.mapError(toPersistenceSqlError("OrchestrationEventRepository.append:insert")), + Effect.flatMap((row) => + eventRowToOrchestrationEvent(row, "OrchestrationEventRepository.append:rowToEvent"), + ), + ), + ), + ); const readFromSequence: OrchestrationEventRepositoryShape["readFromSequence"] = ( sequenceExclusive, limit = 1_000, ) => readEventRowsFromSequence({ sequenceExclusive, limit }).pipe( - Effect.map((rows) => rows.map(eventRowToOrchestrationEvent)), - Effect.orDie, + Effect.mapError(toPersistenceSqlError("OrchestrationEventRepository.readFromSequence:query")), + Effect.flatMap((rows) => + Effect.forEach(rows, (row) => + eventRowToOrchestrationEvent( + row, + "OrchestrationEventRepository.readFromSequence:rowToEvent", + ), + ), + ), ); return { diff --git a/apps/server/src/persistence/Layers/Projects.ts b/apps/server/src/persistence/Layers/Projects.ts index dc50541bc0cb1adc8b9d1f2df87d2aa15a1a1a96..1504545def06626d4f1f4fbb1fa328a016baa5a4 100644 --- a/apps/server/src/persistence/Layers/Projects.ts +++ b/apps/server/src/persistence/Layers/Projects.ts @@ -8,6 +8,7 @@ import type { ProjectListResult, ProjectRecord, ProjectScript, + ProjectUpdateScriptsInput, ProjectUpdateScriptsResult, } from "@t3tools/contracts"; import { @@ -22,6 +23,16 @@ import * as SqlClient from "@effect/sql/SqlClient"; import * as SqlSchema from "@effect/sql/SqlSchema"; import { Effect, Layer, Option, Schema } from "effect"; +import { + PersistenceDecodeError, + ProjectNotFoundError, + ProjectPathMissingError, + ProjectValidationError, + toPersistenceDecodeCauseError, + toPersistenceSerializationError, + toPersistenceSqlError, + type ProjectRepositoryError, +} from "../Errors"; import { ProjectRepository, type ProjectRepositoryShape } from "../Services/Projects"; const ProjectRowSchema = Schema.Struct({ @@ -98,25 +109,101 @@ function inferProjectName(cwd: string): string { return name.length > 0 ? name : "project"; } -function parseScriptsJson(raw: string): ProjectScript[] { - try { - const parsed = JSON.parse(raw) as unknown; - const scripts = projectScriptsSchema.parse(parsed); - return normalizeProjectScripts(scripts); - } catch { - return []; +function parseProjectScripts( + value: unknown, + operation: string, +): Effect.Effect { + const parsed = projectScriptsSchema.safeParse(value); + if (!parsed.success) { + return Effect.fail( + new PersistenceDecodeError({ + operation, + issue: parsed.error.message, + cause: (parsed.error as { cause?: unknown }).cause, + }), + ); } + return Effect.succeed(normalizeProjectScripts(parsed.data)); } -function rowToProjectRecord(row: ProjectRow): ProjectRecord { - return projectRecordSchema.parse({ - id: row.id, - cwd: row.cwd, - name: row.name, - scripts: parseScriptsJson(row.scriptsJson), - createdAt: row.createdAt, - updatedAt: row.updatedAt, - }); +function decodeScriptsJson( + raw: string, + operation: string, +): Effect.Effect { + return Effect.try({ + try: () => JSON.parse(raw) as unknown, + catch: toPersistenceDecodeCauseError(`${operation}:parseScriptsJson`), + }).pipe(Effect.flatMap((value) => parseProjectScripts(value, `${operation}:decodeScripts`))); +} + +function rowToProjectRecord( + row: ProjectRow, + operation: string, +): Effect.Effect { + return decodeScriptsJson(row.scriptsJson, `${operation}:scripts`).pipe( + Effect.flatMap((scripts) => { + const parsed = projectRecordSchema.safeParse({ + id: row.id, + cwd: row.cwd, + name: row.name, + scripts, + createdAt: row.createdAt, + updatedAt: row.updatedAt, + }); + if (!parsed.success) { + return Effect.fail( + new PersistenceDecodeError({ + operation: `${operation}:projectRecord`, + issue: parsed.error.message, + cause: (parsed.error as { cause?: unknown }).cause, + }), + ); + } + return Effect.succeed(parsed.data); + }), + ); +} + +function decodeAddInput(rawInput: ProjectAddInput) { + const parsed = projectAddInputSchema.safeParse(rawInput); + if (!parsed.success) { + return Effect.fail( + new ProjectValidationError({ + operation: "ProjectRepository.add", + issue: parsed.error.message, + cause: (parsed.error as { cause?: unknown }).cause, + }), + ); + } + return Effect.succeed(parsed.data); +} + +function decodeUpdateScriptsInput(rawInput: ProjectUpdateScriptsInput) { + const parsed = projectUpdateScriptsInputSchema.safeParse(rawInput); + if (!parsed.success) { + return Effect.fail( + new ProjectValidationError({ + operation: "ProjectRepository.updateScripts", + issue: parsed.error.message, + cause: (parsed.error as { cause?: unknown }).cause, + }), + ); + } + return Effect.succeed(parsed.data); +} + +function decodeRemoveInput(rawInput: unknown) { + const parsed = projectRemoveInputSchema.safeParse(rawInput); + if (!parsed.success) { + return Effect.fail( + new ProjectValidationError({ + operation: "ProjectRepository.remove", + issue: parsed.error.message, + cause: (parsed.error as { cause?: unknown }).cause, + }), + ); + } + return Effect.succeed(parsed.data); } const makeRepository = Effect.gen(function* () { @@ -231,24 +318,31 @@ const makeRepository = Effect.gen(function* () { const list: ProjectRepositoryShape["list"] = () => listProjectRows(undefined).pipe( - Effect.map((rows): ProjectListResult => rows.map(rowToProjectRecord)), - Effect.map((projects) => - projects.map((project) => ({ ...project, scripts: cloneScripts(project.scripts) })), + Effect.mapError(toPersistenceSqlError("ProjectRepository.list:query")), + Effect.flatMap((rows) => + Effect.forEach(rows, (row) => + rowToProjectRecord(row, "ProjectRepository.list:rowToProject"), + ), + ), + Effect.map( + (projects): ProjectListResult => + projects.map((project) => ({ ...project, scripts: cloneScripts(project.scripts) })), ), - Effect.orDie, ); const add: ProjectRepositoryShape["add"] = (rawInput) => Effect.gen(function* () { - const input = projectAddInputSchema.parse(rawInput as ProjectAddInput); + const input = yield* decodeAddInput(rawInput); const normalizedCwd = normalizeCwd(input.cwd); if (!isDirectory(normalizedCwd)) { - return yield* Effect.fail(new Error(`Project path does not exist: ${normalizedCwd}`)); + return yield* Effect.fail(new ProjectPathMissingError({ cwd: normalizedCwd })); } - const existing = yield* findProjectByCwd({ cwd: normalizedCwd }).pipe(Effect.orDie); + const existing = yield* findProjectByCwd({ cwd: normalizedCwd }).pipe( + Effect.mapError(toPersistenceSqlError("ProjectRepository.add:findByCwd")), + ); if (Option.isSome(existing)) { - const project = rowToProjectRecord(existing.value); + const project = yield* rowToProjectRecord(existing.value, "ProjectRepository.add:existing"); const result: ProjectAddResult = { project: { ...project, @@ -267,9 +361,9 @@ const makeRepository = Effect.gen(function* () { scriptsJson: "[]", createdAt: now, updatedAt: now, - }).pipe(Effect.orDie); + }).pipe(Effect.mapError(toPersistenceSqlError("ProjectRepository.add:insert"))); - const created = rowToProjectRecord(row); + const created = yield* rowToProjectRecord(row, "ProjectRepository.add:created"); const result: ProjectAddResult = { project: { ...created, @@ -282,27 +376,48 @@ const makeRepository = Effect.gen(function* () { const remove: ProjectRepositoryShape["remove"] = (rawInput) => Effect.gen(function* () { - const input = projectRemoveInputSchema.parse(rawInput); - yield* deleteProject({ id: input.id }).pipe(Effect.orDie); + const input = yield* decodeRemoveInput(rawInput); + const existing = yield* findProjectById({ id: input.id }).pipe( + Effect.mapError(toPersistenceSqlError("ProjectRepository.remove:findById")), + ); + if (Option.isNone(existing)) { + return yield* Effect.fail(new ProjectNotFoundError({ projectId: input.id })); + } + + yield* deleteProject({ id: input.id }).pipe( + Effect.mapError(toPersistenceSqlError("ProjectRepository.remove:delete")), + ); }); const updateScripts: ProjectRepositoryShape["updateScripts"] = (rawInput) => Effect.gen(function* () { - const input = projectUpdateScriptsInputSchema.parse(rawInput); - const existing = yield* findProjectById({ id: input.id }).pipe(Effect.orDie); + const input = yield* decodeUpdateScriptsInput(rawInput); + const existing = yield* findProjectById({ id: input.id }).pipe( + Effect.mapError(toPersistenceSqlError("ProjectRepository.updateScripts:findById")), + ); if (Option.isNone(existing)) { - return yield* Effect.fail(new Error(`Project not found: ${input.id}`)); + return yield* Effect.fail(new ProjectNotFoundError({ projectId: input.id })); } - const nextScripts = normalizeProjectScripts(projectScriptsSchema.parse(input.scripts)); + const nextScripts = yield* parseProjectScripts( + input.scripts, + "ProjectRepository.updateScripts:decodeScripts", + ); + const scriptsJson = yield* Effect.try({ + try: () => JSON.stringify(nextScripts), + catch: toPersistenceSerializationError("ProjectRepository.updateScripts:encodeScripts"), + }); const nextUpdatedAt = new Date().toISOString(); const updatedRow = yield* updateProjectScripts({ id: input.id, - scriptsJson: JSON.stringify(nextScripts), + scriptsJson, updatedAt: nextUpdatedAt, - }).pipe(Effect.orDie); + }).pipe(Effect.mapError(toPersistenceSqlError("ProjectRepository.updateScripts:update"))); - const updated = rowToProjectRecord(updatedRow); + const updated = yield* rowToProjectRecord( + updatedRow, + "ProjectRepository.updateScripts:updated", + ); const result: ProjectUpdateScriptsResult = { project: { ...updated, @@ -314,17 +429,19 @@ const makeRepository = Effect.gen(function* () { const pruneMissing: ProjectRepositoryShape["pruneMissing"] = () => listProjectRows(undefined).pipe( + Effect.mapError(toPersistenceSqlError("ProjectRepository.pruneMissing:query")), Effect.flatMap((rows) => Effect.forEach(rows, (row) => { const normalizedCwd = normalizeCwd(row.cwd); if (isDirectory(normalizedCwd)) { return Effect.void; } - return deleteProject({ id: row.id }).pipe(Effect.orDie); + return deleteProject({ id: row.id }).pipe( + Effect.mapError(toPersistenceSqlError("ProjectRepository.pruneMissing:delete")), + ); }), ), Effect.asVoid, - Effect.orDie, ); return { diff --git a/apps/server/src/persistence/Services/OrchestrationEvents.ts b/apps/server/src/persistence/Services/OrchestrationEvents.ts index d785733358f4875525c74bbdc9ad894ed80f8080..745b54902daa6f3b64ecbbecbb0b4c67ade6317c 100644 --- a/apps/server/src/persistence/Services/OrchestrationEvents.ts +++ b/apps/server/src/persistence/Services/OrchestrationEvents.ts @@ -1,18 +1,59 @@ +/** + * OrchestrationEventRepository - Repository interface for orchestration events. + * + * Uses Effect `Context.Tag` for dependency injection and exposes typed + * persistence/decode errors for event append and replay operations. + * + * @module OrchestrationEventRepository + */ import type { OrchestrationEvent } from "@t3tools/contracts"; import { Context } from "effect"; import type { Effect } from "effect"; +import type { OrchestrationEventRepositoryError } from "../Errors"; + export interface OrchestrationEventRepositoryShape { + /** + * Persist a new orchestration event. + * + * @param event - Event payload without sequence (assigned by storage). + * @returns Effect containing the stored event with assigned sequence. + */ readonly append: ( event: Omit, - ) => Effect.Effect; + ) => Effect.Effect; + + /** + * Replay events after the provided sequence. + * + * @param sequenceExclusive - Sequence cursor (exclusive). + * @param limit - Maximum number of events to return. + * @returns Effect containing ordered events. + */ readonly readFromSequence: ( sequenceExclusive: number, limit?: number, - ) => Effect.Effect; - readonly readAll: () => Effect.Effect; + ) => Effect.Effect; + + /** + * Read all events from the beginning of the stream. + * + * @returns Effect containing all stored events. + */ + readonly readAll: () => Effect.Effect; } +/** + * OrchestrationEventRepository - Context tag for orchestration event persistence. + * + * @example + * ```ts + * const program = Effect.gen(function* () { + * const events = yield* OrchestrationEventRepository + * return yield* events.readAll() + * }) + * ``` + */ export class OrchestrationEventRepository extends Context.Tag("orchestration/EventRepository")< OrchestrationEventRepository, OrchestrationEventRepositoryShape diff --git a/apps/server/src/persistence/Services/Projects.ts b/apps/server/src/persistence/Services/Projects.ts index 24288b3a673da920821736f6ece217c4255e7ecc..f16e5c49cdddb59262d1b11176c0a465820525c0 100644 --- a/apps/server/src/persistence/Services/Projects.ts +++ b/apps/server/src/persistence/Services/Projects.ts @@ -1,3 +1,11 @@ +/** + * ProjectRepository - Repository interface for project persistence. + * + * Uses Effect `Context.Tag` for dependency injection and returns typed + * domain errors for all operations. + * + * @module ProjectRepository + */ import type { ProjectAddInput, ProjectAddResult, @@ -9,16 +17,61 @@ import type { import { Context } from "effect"; import type { Effect } from "effect"; +import type { ProjectRepositoryError } from "../Errors"; + export interface ProjectRepositoryShape { - readonly list: () => Effect.Effect; - readonly add: (input: ProjectAddInput) => Effect.Effect; - readonly remove: (input: ProjectRemoveInput) => Effect.Effect; + /** + * List all persisted projects. + * + * @returns Effect containing all project records. + */ + readonly list: () => Effect.Effect; + + /** + * Add a project by cwd, or return the existing project when the cwd already exists. + * + * @param input - Project creation payload. + * @returns Effect containing the project and whether it was newly created. + */ + readonly add: (input: ProjectAddInput) => Effect.Effect; + + /** + * Remove a persisted project. + * + * @param input - Project removal payload. + * @returns Effect containing void. + */ + readonly remove: (input: ProjectRemoveInput) => Effect.Effect; + + /** + * Update the scripts for an existing project. + * + * @param input - Script update payload. + * @returns Effect containing the updated project. + */ readonly updateScripts: ( input: ProjectUpdateScriptsInput, - ) => Effect.Effect; - readonly pruneMissing: () => Effect.Effect; + ) => Effect.Effect; + + /** + * Remove projects whose cwd no longer exists on disk. + * + * @returns Effect containing void. + */ + readonly pruneMissing: () => Effect.Effect; } +/** + * ProjectRepository - Context tag for project persistence dependency injection. + * + * @example + * ```ts + * const program = Effect.gen(function* () { + * const projects = yield* ProjectRepository + * return yield* projects.list() + * }) + * ``` + */ export class ProjectRepository extends Context.Tag("persistence/ProjectRepository")< ProjectRepository, ProjectRepositoryShape -- 2.54.0