diff --git a/apps/server/src/orchestration/Layer.ts b/apps/server/src/orchestration/Layer.ts index 5c08697cc2c7cccd0bd8360c54eb5dcb7371856d..bd36e701c24b3d519c30a877af946c3befd8d340 100644 --- a/apps/server/src/orchestration/Layer.ts +++ b/apps/server/src/orchestration/Layer.ts @@ -162,131 +162,126 @@ 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 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.catch((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); - } - }, - ); - - 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); - }); - - const subscribeToReadModel: OrchestrationEngineShape["subscribeToReadModel"] = (callback) => - Effect.sync(() => { - readModelListeners.add(callback); - return () => { - readModelListeners.delete(callback); - }; - }); - - const subscribeToDomainEvents: OrchestrationEngineShape["subscribeToDomainEvents"] = ( - callback, - ) => - Effect.sync(() => { - domainEventListeners.add(callback); - return () => { - domainEventListeners.delete(callback); - }; - }); - - return { - getSnapshot, - replayEvents, - dispatchUnknown, - dispatch, - subscribeToReadModel, - subscribeToDomainEvents, - } satisfies OrchestrationEngineShape; +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.catch((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); + } + }, + ); + + 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); + }); + + const subscribeToReadModel: OrchestrationEngineShape["subscribeToReadModel"] = (callback) => + Effect.sync(() => { + readModelListeners.add(callback); + return () => { + readModelListeners.delete(callback); + }; + }); + + 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.effect( OrchestrationEngineService, makeOrchestrationEngine, diff --git a/apps/server/src/orchestration/reducer.ts b/apps/server/src/orchestration/reducer.ts index 8fa8187af3bd7d31ee600b79f6bce863b1a1de80..5ba4334e0fc7ced6e0cdf891fd7e5c9a171825d3 100644 --- a/apps/server/src/orchestration/reducer.ts +++ b/apps/server/src/orchestration/reducer.ts @@ -36,7 +36,7 @@ function decodeForEvent( ): Effect.Effect { return Schema.decodeUnknownEffect(schema)(value).pipe( Effect.mapError(toReducerDecodeError(`${eventType}:${field}`)), - ) as Effect.Effect + ) as Effect.Effect; } const ThreadCreatedPayloadSchema = Schema.Struct({ diff --git a/apps/server/src/persistence/Layers/OrchestrationEvents.test.ts b/apps/server/src/persistence/Layers/OrchestrationEvents.test.ts index 0959c691e39b95bc337a4cc3481bfcae1fa8ff36..6b11a973e2358c0d8806bad4a350d35bae1a10ed 100644 --- a/apps/server/src/persistence/Layers/OrchestrationEvents.test.ts +++ b/apps/server/src/persistence/Layers/OrchestrationEvents.test.ts @@ -3,7 +3,7 @@ import os from "node:os"; import path from "node:path"; import { DatabaseSync } from "node:sqlite"; -import { ManagedRuntime, Effect } from "effect"; +import { ManagedRuntime, Effect } from "effect"; import { afterEach, describe, expect, it } from "vitest"; import { OrchestrationEventRepository } from "../Services/OrchestrationEvents.ts"; @@ -57,7 +57,6 @@ describe("OrchestrationEventRepository", () => { const stateDir = makeTempDir("t3code-event-store-migrations-"); const dbPath = path.join(stateDir, "orchestration.sqlite"); - const firstRuntime = ManagedRuntime.make(makeSqliteOrchestrationEventRepositoryLive(dbPath)); const first = await firstRuntime.runPromise(Effect.service(OrchestrationEventRepository)); await firstRuntime.runPromise(first.readAll()); diff --git a/apps/server/src/persistence/Layers/Projects.ts b/apps/server/src/persistence/Layers/Projects.ts index bbda8fa644a7cad6ed147105c5a4195a13e99475..e38ef34587655a5ff80c57c43264b1b8410f4eb3 100644 --- a/apps/server/src/persistence/Layers/Projects.ts +++ b/apps/server/src/persistence/Layers/Projects.ts @@ -381,7 +381,7 @@ const makeRepository = Effect.gen(function* () { Effect.mapError(toPersistenceSqlError("ProjectRepository.remove:findById")), ); if (Option.isNone(existing)) { - return yield* (new ProjectNotFoundError({ projectId: input.id })); + return yield* new ProjectNotFoundError({ projectId: input.id }); } yield* deleteProject({ id: input.id }).pipe( @@ -396,7 +396,7 @@ const makeRepository = Effect.gen(function* () { Effect.mapError(toPersistenceSqlError("ProjectRepository.updateScripts:findById")), ); if (Option.isNone(existing)) { - return yield* (new ProjectNotFoundError({ projectId: input.id })); + return yield* new ProjectNotFoundError({ projectId: input.id }); } const nextScripts = yield* parseProjectScripts( diff --git a/apps/server/src/persistence/Layers/Sqlite.ts b/apps/server/src/persistence/Layers/Sqlite.ts index c8ea33b12ebb6bc1ee6c3c659112cf57b1b196bf..53a8b7d558291e3096cc08bae3086299db2318b6 100644 --- a/apps/server/src/persistence/Layers/Sqlite.ts +++ b/apps/server/src/persistence/Layers/Sqlite.ts @@ -5,8 +5,6 @@ import * as SqlClient from "effect/unstable/sql/SqlClient"; import * as SqliteClient from "@effect/sql-sqlite-node/SqliteClient"; import { Effect, Layer, ServiceMap } from "effect"; - - import { runMigrations } from "../Migrations.ts"; export interface PersistenceConfigShape { @@ -38,7 +36,6 @@ export function makeSqlitePersistenceLive(dbPath: string) { ).pipe(Layer.provide(SqliteClientLive)); } - export const SqlitePersistenceLive = Layer.unwrap( Effect.map(Effect.service(PersistenceConfig), ({ dbPath }) => makeSqlitePersistenceLive(dbPath)), );