import type { OrchestrationEvent, OrchestrationReadModel, ProjectId, ThreadId,} from "@t3tools/contracts";import { OrchestrationCommand } from "@t3tools/contracts";import { Deferred, Effect, Layer, Option, PubSub, Queue, Schema, Stream } from "effect";import * as SqlClient from "effect/unstable/sql/SqlClient"; import { toPersistenceSqlError } from "../../persistence/Errors.ts";import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts";import { OrchestrationCommandReceiptRepository } from "../../persistence/Services/OrchestrationCommandReceipts.ts";import { OrchestrationCommandInvariantError, OrchestrationCommandPreviouslyRejectedError, type OrchestrationDispatchError,} from "../Errors.ts";import { decideOrchestrationCommand } from "../decider.ts";import { createEmptyReadModel, projectEvent } from "../projector.ts";import { OrchestrationProjectionPipeline } from "../Services/ProjectionPipeline.ts";import { OrchestrationEngineService, type OrchestrationEngineShape,} from "../Services/OrchestrationEngine.ts"; interface CommandEnvelope { command: OrchestrationCommand; result: Deferred.Deferred<{ sequence: number }, OrchestrationDispatchError>;} function commandToAggregateRef(command: OrchestrationCommand): { readonly aggregateKind: "project" | "thread"; readonly aggregateId: ProjectId | ThreadId;} { switch (command.type) { case "project.create": case "project.meta.update": case "project.delete": return { aggregateKind: "project", aggregateId: command.projectId, }; default: return { aggregateKind: "thread", aggregateId: command.threadId, }; }} const makeOrchestrationEngine = Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; const eventStore = yield* OrchestrationEventStore; const commandReceiptRepository = yield* OrchestrationCommandReceiptRepository; const projectionPipeline = yield* OrchestrationProjectionPipeline; let readModel = createEmptyReadModel(new Date().toISOString()); const commandQueue = yield* Queue.unbounded<CommandEnvelope>(); const eventPubSub = yield* PubSub.unbounded<OrchestrationEvent>(); const processEnvelope = (envelope: CommandEnvelope): Effect.Effect<void> => { const dispatchStartSequence = readModel.snapshotSequence; const reconcileReadModelAfterDispatchFailure = Effect.gen(function* () { const persistedEvents = yield* Stream.runCollect( eventStore.readFromSequence(dispatchStartSequence), ).pipe(Effect.map((chunk): OrchestrationEvent[] => Array.from(chunk))); if (persistedEvents.length === 0) { return; } let nextReadModel = readModel; for (const persistedEvent of persistedEvents) { nextReadModel = yield* projectEvent(nextReadModel, persistedEvent); } readModel = nextReadModel; for (const persistedEvent of persistedEvents) { yield* PubSub.publish(eventPubSub, persistedEvent); } }); return Effect.gen(function* () { const existingReceipt = yield* commandReceiptRepository.getByCommandId({ commandId: envelope.command.commandId, }); if (Option.isSome(existingReceipt)) { if (existingReceipt.value.status === "accepted") { yield* Deferred.succeed(envelope.result, { sequence: existingReceipt.value.resultSequence, }); return; } yield* Deferred.fail( envelope.result, new OrchestrationCommandPreviouslyRejectedError({ commandId: envelope.command.commandId, detail: existingReceipt.value.error ?? "Previously rejected.", }), ); return; } const eventBase = yield* decideOrchestrationCommand({ command: envelope.command, readModel, }); const eventBases = Array.isArray(eventBase) ? eventBase : [eventBase]; const committedCommand = yield* sql .withTransaction( Effect.gen(function* () { const committedEvents: OrchestrationEvent[] = []; let nextReadModel = readModel; for (const nextEvent of eventBases) { const savedEvent = yield* eventStore.append(nextEvent); nextReadModel = yield* projectEvent(nextReadModel, savedEvent); yield* projectionPipeline.projectEvent(savedEvent); committedEvents.push(savedEvent); } const lastSavedEvent = committedEvents.at(-1) ?? null; if (lastSavedEvent === null) { return yield* new OrchestrationCommandInvariantError({ commandType: envelope.command.type, detail: "Command produced no events.", }); } yield* commandReceiptRepository.upsert({ commandId: envelope.command.commandId, aggregateKind: lastSavedEvent.aggregateKind, aggregateId: lastSavedEvent.aggregateId, acceptedAt: lastSavedEvent.occurredAt, resultSequence: lastSavedEvent.sequence, status: "accepted", error: null, }); return { committedEvents, lastSequence: lastSavedEvent.sequence, nextReadModel, } as const; }), ) .pipe( Effect.catchTag("SqlError", (sqlError) => Effect.fail( toPersistenceSqlError("OrchestrationEngine.processEnvelope:transaction")(sqlError), ), ), ); readModel = committedCommand.nextReadModel; for (const event of committedCommand.committedEvents) { yield* PubSub.publish(eventPubSub, event); } yield* Deferred.succeed(envelope.result, { sequence: committedCommand.lastSequence }); }).pipe( Effect.catch((error) => Effect.gen(function* () { yield* reconcileReadModelAfterDispatchFailure.pipe( Effect.catch(() => Effect.logWarning( "failed to reconcile orchestration read model after dispatch failure", ).pipe( Effect.annotateLogs({ commandId: envelope.command.commandId, snapshotSequence: readModel.snapshotSequence, }), ), ), ); if (Schema.is(OrchestrationCommandInvariantError)(error)) { const aggregateRef = commandToAggregateRef(envelope.command); yield* commandReceiptRepository .upsert({ commandId: envelope.command.commandId, aggregateKind: aggregateRef.aggregateKind, aggregateId: aggregateRef.aggregateId, acceptedAt: new Date().toISOString(), resultSequence: readModel.snapshotSequence, status: "rejected", error: error.message, }) .pipe(Effect.catch(() => Effect.void)); } yield* Deferred.fail(envelope.result, error); }), ), ); }; yield* projectionPipeline.bootstrap; // bootstrap in-memory read model from event store yield* Stream.runForEach(eventStore.readAll(), (event) => Effect.gen(function* () { readModel = yield* projectEvent(readModel, event); }), ); const worker = Effect.forever(Queue.take(commandQueue).pipe(Effect.flatMap(processEnvelope))); yield* Effect.forkScoped(worker); yield* Effect.logDebug("orchestration engine started").pipe( Effect.annotateLogs({ sequence: readModel.snapshotSequence }), ); const getReadModel: OrchestrationEngineShape["getReadModel"] = () => Effect.sync((): OrchestrationReadModel => readModel); const readEvents: OrchestrationEngineShape["readEvents"] = (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); }); return { getReadModel, readEvents, dispatch, // Each access creates a fresh PubSub subscription so that multiple // consumers (wsServer, ProviderRuntimeIngestion, CheckpointReactor, etc.) // each independently receive all domain events. get streamDomainEvents(): OrchestrationEngineShape["streamDomainEvents"] { return Stream.fromPubSub(eventPubSub); }, } satisfies OrchestrationEngineShape;}); export const OrchestrationEngineLive = Layer.effect( OrchestrationEngineService, makeOrchestrationEngine,);
import type {
OrchestrationEvent,
OrchestrationReadModel,
ProjectId,
ThreadId,
} from "@t3tools/contracts";
import { OrchestrationCommand } from "@t3tools/contracts";
import { Deferred, Effect, Layer, Option, PubSub, Queue, Schema, Stream } from "effect";
import * as SqlClient from "effect/unstable/sql/SqlClient";
import { toPersistenceSqlError } from "../../persistence/Errors.ts";
import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts";
import { OrchestrationCommandReceiptRepository } from "../../persistence/Services/OrchestrationCommandReceipts.ts";
import {
OrchestrationCommandInvariantError,
OrchestrationCommandPreviouslyRejectedError,
type OrchestrationDispatchError,
} from "../Errors.ts";
import { decideOrchestrationCommand } from "../decider.ts";
import { createEmptyReadModel, projectEvent } from "../projector.ts";
import { OrchestrationProjectionPipeline } from "../Services/ProjectionPipeline.ts";
import {
OrchestrationEngineService,
type OrchestrationEngineShape,
} from "../Services/OrchestrationEngine.ts";
interface CommandEnvelope {
command: OrchestrationCommand;
result: Deferred.Deferred<{ sequence: number }, OrchestrationDispatchError>;
}
function commandToAggregateRef(command: OrchestrationCommand): {
readonly aggregateKind: "project" | "thread";
readonly aggregateId: ProjectId | ThreadId;
} {
switch (command.type) {
case "project.create":
case "project.meta.update":
case "project.delete":
return {
aggregateKind: "project",
aggregateId: command.projectId,
};
default:
return {
aggregateKind: "thread",
aggregateId: command.threadId,
};
}
}
const makeOrchestrationEngine = Effect.gen(function* () {
const sql = yield* SqlClient.SqlClient;
const eventStore = yield* OrchestrationEventStore;
const commandReceiptRepository = yield* OrchestrationCommandReceiptRepository;
const projectionPipeline = yield* OrchestrationProjectionPipeline;
let readModel = createEmptyReadModel(new Date().toISOString());
const commandQueue = yield* Queue.unbounded<CommandEnvelope>();
const eventPubSub = yield* PubSub.unbounded<OrchestrationEvent>();
const processEnvelope = (envelope: CommandEnvelope): Effect.Effect<void> => {
const dispatchStartSequence = readModel.snapshotSequence;
const reconcileReadModelAfterDispatchFailure = Effect.gen(function* () {
const persistedEvents = yield* Stream.runCollect(
eventStore.readFromSequence(dispatchStartSequence),
).pipe(Effect.map((chunk): OrchestrationEvent[] => Array.from(chunk)));
if (persistedEvents.length === 0) {
return;
}
let nextReadModel = readModel;
for (const persistedEvent of persistedEvents) {
nextReadModel = yield* projectEvent(nextReadModel, persistedEvent);
}
readModel = nextReadModel;
for (const persistedEvent of persistedEvents) {
yield* PubSub.publish(eventPubSub, persistedEvent);
}
});
return Effect.gen(function* () {
const existingReceipt = yield* commandReceiptRepository.getByCommandId({
commandId: envelope.command.commandId,
});
if (Option.isSome(existingReceipt)) {
if (existingReceipt.value.status === "accepted") {
yield* Deferred.succeed(envelope.result, {
sequence: existingReceipt.value.resultSequence,
});
return;
}
yield* Deferred.fail(
envelope.result,
new OrchestrationCommandPreviouslyRejectedError({
commandId: envelope.command.commandId,
detail: existingReceipt.value.error ?? "Previously rejected.",
}),
);
return;
}
const eventBase = yield* decideOrchestrationCommand({
command: envelope.command,
readModel,
});
const eventBases = Array.isArray(eventBase) ? eventBase : [eventBase];
const committedCommand = yield* sql
.withTransaction(
Effect.gen(function* () {
const committedEvents: OrchestrationEvent[] = [];
let nextReadModel = readModel;
for (const nextEvent of eventBases) {
const savedEvent = yield* eventStore.append(nextEvent);
nextReadModel = yield* projectEvent(nextReadModel, savedEvent);
yield* projectionPipeline.projectEvent(savedEvent);
committedEvents.push(savedEvent);
}
const lastSavedEvent = committedEvents.at(-1) ?? null;
if (lastSavedEvent === null) {
return yield* new OrchestrationCommandInvariantError({
commandType: envelope.command.type,
detail: "Command produced no events.",
});
}
yield* commandReceiptRepository.upsert({
commandId: envelope.command.commandId,
aggregateKind: lastSavedEvent.aggregateKind,
aggregateId: lastSavedEvent.aggregateId,
acceptedAt: lastSavedEvent.occurredAt,
resultSequence: lastSavedEvent.sequence,
status: "accepted",
error: null,
});
return {
committedEvents,
lastSequence: lastSavedEvent.sequence,
nextReadModel,
} as const;
}),
)
.pipe(
Effect.catchTag("SqlError", (sqlError) =>
Effect.fail(
toPersistenceSqlError("OrchestrationEngine.processEnvelope:transaction")(sqlError),
),
),
);
readModel = committedCommand.nextReadModel;
for (const event of committedCommand.committedEvents) {
yield* PubSub.publish(eventPubSub, event);
}
yield* Deferred.succeed(envelope.result, { sequence: committedCommand.lastSequence });
}).pipe(
Effect.catch((error) =>
Effect.gen(function* () {
yield* reconcileReadModelAfterDispatchFailure.pipe(
Effect.catch(() =>
Effect.logWarning(
"failed to reconcile orchestration read model after dispatch failure",
).pipe(
Effect.annotateLogs({
commandId: envelope.command.commandId,
snapshotSequence: readModel.snapshotSequence,
}),
),
),
);
if (Schema.is(OrchestrationCommandInvariantError)(error)) {
const aggregateRef = commandToAggregateRef(envelope.command);
yield* commandReceiptRepository
.upsert({
commandId: envelope.command.commandId,
aggregateKind: aggregateRef.aggregateKind,
aggregateId: aggregateRef.aggregateId,
acceptedAt: new Date().toISOString(),
resultSequence: readModel.snapshotSequence,
status: "rejected",
error: error.message,
})
.pipe(Effect.catch(() => Effect.void));
}
yield* Deferred.fail(envelope.result, error);
}),
),
);
};
yield* projectionPipeline.bootstrap;
// bootstrap in-memory read model from event store
yield* Stream.runForEach(eventStore.readAll(), (event) =>
Effect.gen(function* () {
readModel = yield* projectEvent(readModel, event);
}),
);
const worker = Effect.forever(Queue.take(commandQueue).pipe(Effect.flatMap(processEnvelope)));
yield* Effect.forkScoped(worker);
yield* Effect.logDebug("orchestration engine started").pipe(
Effect.annotateLogs({ sequence: readModel.snapshotSequence }),
);
const getReadModel: OrchestrationEngineShape["getReadModel"] = () =>
Effect.sync((): OrchestrationReadModel => readModel);
const readEvents: OrchestrationEngineShape["readEvents"] = (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);
});
return {
getReadModel,
readEvents,
dispatch,
// Each access creates a fresh PubSub subscription so that multiple
// consumers (wsServer, ProviderRuntimeIngestion, CheckpointReactor, etc.)
// each independently receive all domain events.
get streamDomainEvents(): OrchestrationEngineShape["streamDomainEvents"] {
return Stream.fromPubSub(eventPubSub);
},
} satisfies OrchestrationEngineShape;
});
export const OrchestrationEngineLive = Layer.effect(
OrchestrationEngineService,
makeOrchestrationEngine,
);