authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-22 11:07:39-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-02-26 13:20:36-08:00
log5e3678c0af5023d02d938931b9efbb75b7c53f7e
tree9ceec8c187276a0c3793fcd357f90394e30bb513
parent0e4f1485c87a234b33cad6c197625d414bfd070c

fmt

5 files changed, 102 insertions(+), 111 deletions(-)

apps/server/src/orchestration/Layer.ts+98-103
......@@ -162,130 +162,125 @@ function mapCommandToEvent(command: OrchestrationCommand): Omit<OrchestrationEve
162162
163163const decodeUnknownCommand = Schema.decodeUnknownEffect(OrchestrationCommandSchema);
164164
165const makeOrchestrationEngine =
166 Effect.gen(function* () {
167 const logger = createLogger("orchestration");
168 const eventStore = yield* OrchestrationEventRepository;
165const makeOrchestrationEngine = Effect.gen(function* () {
166 const logger = createLogger("orchestration");
167 const eventStore = yield* OrchestrationEventRepository;
169168
170 let readModel = createEmptyReadModel(new Date().toISOString());
169 let readModel = createEmptyReadModel(new Date().toISOString());
171170
172 const commandQueue = yield* Queue.unbounded<CommandEnvelope>();
173 const readModelPubSub = yield* PubSub.unbounded<OrchestrationReadModel>();
174 const eventPubSub = yield* PubSub.unbounded<OrchestrationEvent>();
171 const commandQueue = yield* Queue.unbounded<CommandEnvelope>();
172 const readModelPubSub = yield* PubSub.unbounded<OrchestrationReadModel>();
173 const eventPubSub = yield* PubSub.unbounded<OrchestrationEvent>();
175174
176 const readModelListeners = new Set<(snapshot: OrchestrationReadModel) => void>();
177 const domainEventListeners = new Set<(event: OrchestrationEvent) => void>();
175 const readModelListeners = new Set<(snapshot: OrchestrationReadModel) => void>();
176 const domainEventListeners = new Set<(event: OrchestrationEvent) => void>();
178177
179 const notifyReadModelListeners = (snapshot: OrchestrationReadModel) =>
180 Effect.forEach(readModelListeners, (listener) =>
181 Effect.try({
182 try: () => listener(snapshot),
183 catch: toListenerCallbackError("read-model"),
184 }),
185 ).pipe(Effect.asVoid);
178 const notifyReadModelListeners = (snapshot: OrchestrationReadModel) =>
179 Effect.forEach(readModelListeners, (listener) =>
180 Effect.try({
181 try: () => listener(snapshot),
182 catch: toListenerCallbackError("read-model"),
183 }),
184 ).pipe(Effect.asVoid);
186185
187 const notifyDomainEventListeners = (event: OrchestrationEvent) =>
188 Effect.forEach(domainEventListeners, (listener) =>
189 Effect.try({
190 try: () => listener(event),
191 catch: toListenerCallbackError("domain-event"),
192 }),
193 ).pipe(Effect.asVoid);
186 const notifyDomainEventListeners = (event: OrchestrationEvent) =>
187 Effect.forEach(domainEventListeners, (listener) =>
188 Effect.try({
189 try: () => listener(event),
190 catch: toListenerCallbackError("domain-event"),
191 }),
192 ).pipe(Effect.asVoid);
194193
195 const processEnvelope = (envelope: CommandEnvelope): Effect.Effect<void> =>
196 Effect.gen(function* () {
197 const eventBase = mapCommandToEvent(envelope.command);
198 const savedEvent = yield* eventStore.append(eventBase);
199 readModel = yield* reduceEvent(readModel, savedEvent);
194 const processEnvelope = (envelope: CommandEnvelope): Effect.Effect<void> =>
195 Effect.gen(function* () {
196 const eventBase = mapCommandToEvent(envelope.command);
197 const savedEvent = yield* eventStore.append(eventBase);
198 readModel = yield* reduceEvent(readModel, savedEvent);
200199
201 const snapshot = readModel;
202 yield* Effect.all([
203 PubSub.publish(eventPubSub, savedEvent),
204 PubSub.publish(readModelPubSub, snapshot),
205 ]);
200 const snapshot = readModel;
201 yield* Effect.all([
202 PubSub.publish(eventPubSub, savedEvent),
203 PubSub.publish(readModelPubSub, snapshot),
204 ]);
206205
207 yield* notifyDomainEventListeners(savedEvent);
208 yield* notifyReadModelListeners(snapshot);
206 yield* notifyDomainEventListeners(savedEvent);
207 yield* notifyReadModelListeners(snapshot);
209208
210 yield* Deferred.succeed(envelope.result, { sequence: savedEvent.sequence });
211 }).pipe(
212 Effect.catch((error) => Deferred.fail(envelope.result, error).pipe(Effect.asVoid)),
213 );
209 yield* Deferred.succeed(envelope.result, { sequence: savedEvent.sequence });
210 }).pipe(Effect.catch((error) => Deferred.fail(envelope.result, error).pipe(Effect.asVoid)));
214211
215 const bootstrapReadModel: Effect.Effect<void, OrchestrationDispatchError> = Effect.gen(
216 function* () {
217 const existingEvents = yield* eventStore.readAll();
218 for (const event of existingEvents) {
219 readModel = yield* reduceEvent(readModel, event);
220 }
221 },
222 );
212 const bootstrapReadModel: Effect.Effect<void, OrchestrationDispatchError> = Effect.gen(
213 function* () {
214 const existingEvents = yield* eventStore.readAll();
215 for (const event of existingEvents) {
216 readModel = yield* reduceEvent(readModel, event);
217 }
218 },
219 );
223220
224 yield* bootstrapReadModel;
221 yield* bootstrapReadModel;
225222
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 });
223 const worker = Effect.forever(Queue.take(commandQueue).pipe(Effect.flatMap(processEnvelope)));
224 yield* Effect.forkScoped(worker);
225 logger.info("orchestration engine started", {
226 sequence: readModel.sequence,
227 });
231228
232 const getSnapshot: OrchestrationEngineShape["getSnapshot"] = () =>
233 Effect.sync((): OrchestrationReadModel => readModel);
229 const getSnapshot: OrchestrationEngineShape["getSnapshot"] = () =>
230 Effect.sync((): OrchestrationReadModel => readModel);
234231
235 const replayEvents: OrchestrationEngineShape["replayEvents"] = (fromSequenceExclusive) =>
236 eventStore.readFromSequence(fromSequenceExclusive);
232 const replayEvents: OrchestrationEngineShape["replayEvents"] = (fromSequenceExclusive) =>
233 eventStore.readFromSequence(fromSequenceExclusive);
237234
238 const dispatch: OrchestrationEngineShape["dispatch"] = (command) =>
239 Effect.gen(function* () {
240 const result = yield* Deferred.make<{ sequence: number }, OrchestrationDispatchError>();
241 yield* Queue.offer(commandQueue, { command, result });
242 return yield* Deferred.await(result);
243 });
235 const dispatch: OrchestrationEngineShape["dispatch"] = (command) =>
236 Effect.gen(function* () {
237 const result = yield* Deferred.make<{ sequence: number }, OrchestrationDispatchError>();
238 yield* Queue.offer(commandQueue, { command, result });
239 return yield* Deferred.await(result);
240 });
244241
245 const dispatchUnknown: OrchestrationEngineShape["dispatchUnknown"] = (command) =>
246 Effect.gen(function* () {
247 const payload =
248 typeof command === "string"
249 ? yield* Effect.try({
250 try: () => JSON.parse(command) as unknown,
251 catch: toOrchestrationJsonParseError,
252 })
253 : command;
242 const dispatchUnknown: OrchestrationEngineShape["dispatchUnknown"] = (command) =>
243 Effect.gen(function* () {
244 const payload =
245 typeof command === "string"
246 ? yield* Effect.try({
247 try: () => JSON.parse(command) as unknown,
248 catch: toOrchestrationJsonParseError,
249 })
250 : command;
254251
255 const decoded = yield* decodeUnknownCommand(payload).pipe(
256 Effect.mapError(toOrchestrationCommandDecodeError),
257 );
252 const decoded = yield* decodeUnknownCommand(payload).pipe(
253 Effect.mapError(toOrchestrationCommandDecodeError),
254 );
258255
259 return yield* dispatch(decoded);
260 });
256 return yield* dispatch(decoded);
257 });
261258
262 const subscribeToReadModel: OrchestrationEngineShape["subscribeToReadModel"] = (callback) =>
263 Effect.sync(() => {
264 readModelListeners.add(callback);
265 return () => {
266 readModelListeners.delete(callback);
267 };
268 });
259 const subscribeToReadModel: OrchestrationEngineShape["subscribeToReadModel"] = (callback) =>
260 Effect.sync(() => {
261 readModelListeners.add(callback);
262 return () => {
263 readModelListeners.delete(callback);
264 };
265 });
269266
270 const subscribeToDomainEvents: OrchestrationEngineShape["subscribeToDomainEvents"] = (
271 callback,
272 ) =>
273 Effect.sync(() => {
274 domainEventListeners.add(callback);
275 return () => {
276 domainEventListeners.delete(callback);
277 };
278 });
267 const subscribeToDomainEvents: OrchestrationEngineShape["subscribeToDomainEvents"] = (callback) =>
268 Effect.sync(() => {
269 domainEventListeners.add(callback);
270 return () => {
271 domainEventListeners.delete(callback);
272 };
273 });
279274
280 return {
281 getSnapshot,
282 replayEvents,
283 dispatchUnknown,
284 dispatch,
285 subscribeToReadModel,
286 subscribeToDomainEvents,
287 } satisfies OrchestrationEngineShape;
288 });
275 return {
276 getSnapshot,
277 replayEvents,
278 dispatchUnknown,
279 dispatch,
280 subscribeToReadModel,
281 subscribeToDomainEvents,
282 } satisfies OrchestrationEngineShape;
283});
289284
290285export const OrchestrationEngineLive = Layer.effect(
291286 OrchestrationEngineService,
apps/server/src/orchestration/reducer.ts+1-1
......@@ -36,7 +36,7 @@ function decodeForEvent<A>(
3636): Effect.Effect<A, OrchestrationReducerDecodeError> {
3737 return Schema.decodeUnknownEffect(schema)(value).pipe(
3838 Effect.mapError(toReducerDecodeError(`${eventType}:${field}`)),
39 ) as Effect.Effect<A, OrchestrationReducerDecodeError>
39 ) as Effect.Effect<A, OrchestrationReducerDecodeError>;
4040}
4141
4242const ThreadCreatedPayloadSchema = Schema.Struct({
apps/server/src/persistence/Layers/OrchestrationEvents.test.ts+1-2
......@@ -3,7 +3,7 @@ import os from "node:os";
33import path from "node:path";
44import { DatabaseSync } from "node:sqlite";
55
6import { ManagedRuntime, Effect } from "effect";
6import { ManagedRuntime, Effect } from "effect";
77import { afterEach, describe, expect, it } from "vitest";
88
99import { OrchestrationEventRepository } from "../Services/OrchestrationEvents.ts";
......@@ -57,7 +57,6 @@ describe("OrchestrationEventRepository", () => {
5757 const stateDir = makeTempDir("t3code-event-store-migrations-");
5858 const dbPath = path.join(stateDir, "orchestration.sqlite");
5959
60
6160 const firstRuntime = ManagedRuntime.make(makeSqliteOrchestrationEventRepositoryLive(dbPath));
6261 const first = await firstRuntime.runPromise(Effect.service(OrchestrationEventRepository));
6362 await firstRuntime.runPromise(first.readAll());
apps/server/src/persistence/Layers/Projects.ts+2-2
......@@ -381,7 +381,7 @@ const makeRepository = Effect.gen(function* () {
381381 Effect.mapError(toPersistenceSqlError("ProjectRepository.remove:findById")),
382382 );
383383 if (Option.isNone(existing)) {
384 return yield* (new ProjectNotFoundError({ projectId: input.id }));
384 return yield* new ProjectNotFoundError({ projectId: input.id });
385385 }
386386
387387 yield* deleteProject({ id: input.id }).pipe(
......@@ -396,7 +396,7 @@ const makeRepository = Effect.gen(function* () {
396396 Effect.mapError(toPersistenceSqlError("ProjectRepository.updateScripts:findById")),
397397 );
398398 if (Option.isNone(existing)) {
399 return yield* (new ProjectNotFoundError({ projectId: input.id }));
399 return yield* new ProjectNotFoundError({ projectId: input.id });
400400 }
401401
402402 const nextScripts = yield* parseProjectScripts(
apps/server/src/persistence/Layers/Sqlite.ts-3
......@@ -5,8 +5,6 @@ import * as SqlClient from "effect/unstable/sql/SqlClient";
55import * as SqliteClient from "@effect/sql-sqlite-node/SqliteClient";
66import { Effect, Layer, ServiceMap } from "effect";
77
8
9
108import { runMigrations } from "../Migrations.ts";
119
1210export interface PersistenceConfigShape {
......@@ -38,7 +36,6 @@ export function makeSqlitePersistenceLive(dbPath: string) {
3836 ).pipe(Layer.provide(SqliteClientLive));
3937}
4038
41
4239export const SqlitePersistenceLive = Layer.unwrap(
4340 Effect.map(Effect.service(PersistenceConfig), ({ dbPath }) => makeSqlitePersistenceLive(dbPath)),
4441);