New implementation

.gitignore

@@ -2,3 +2,4 @@
 node_modules/
 dist/
 *.log
+tags

src/app/postgresEventStore.ts → src/app/eventStore.ts

@@ -1,4 +1,4 @@
-export { initialize, PostgresEventStore };
+export { initialize, PostgresEventStore, type WithEventStore };
 
 import { Json } from '@/lib/json/types';
 import {
@@ -7,26 +7,28 @@ import {
   Aggregate,
   CreationEvent,
   TransformationEvent,
-  EventClass,
   EventInfo,
 } from '@/lib/eventSourcing/event';
 import {
   EventStore,
-  Hydrator,
+  Schemas,
   Constructor,
   EventData,
-  schema_EventData,
 } from '@/lib/eventSourcing/eventStore';
 import { PostgresTransaction } from '@/lib/postgres';
 import { log } from '@/common/util/Logger';
-import { IdGenerator } from '@/common/util/IdGenerator';
-import { encode } from '@/lib/json/schema';
 import { POSIX } from '@/lib/time';
+import { Future } from '@/lib/Future';
+
+type WithEventStore = <E, T>(
+  onError: (e: Error) => E,
+  f: (s: EventStore) => Future<E, T>,
+) => Future<E, T>;
 
 class PostgresEventStore implements EventStore {
   constructor(
     private transaction: PostgresTransaction,
-    private readonly hydrator: Hydrator,
+    private readonly schemas: Schemas,
     private readonly eventStoreTable: string,
   ) {}
 
@@ -35,22 +37,22 @@ class PostgresEventStore implements EventStore {
     aggregateId: Id<T>,
   ): Promise<{ aggregate: T; lastEvent: EventInfo }> {
     const events = await this.findAll(aggregateId);
-    const { lastEvent, aggregate } = this.hydrator
+    const { lastEvent, aggregate } = this.schemas
       .hydrate(cls, events)
       .unwrap((e) => e);
 
     return { aggregate, lastEvent };
   }
 
-  async save<E extends Event<T>, T extends Aggregate<T>>(args: {
+  async emit<T extends Aggregate<T>>(args: {
     aggregate: Constructor<T>;
-    event: CreationEvent<E, T> | TransformationEvent<E, T>;
+    event: CreationEvent<T> | TransformationEvent<T>;
     event_id?: Id<Event<T>>;
     correlation_id?: Id<Event<T>>;
     causation_id?: Id<Event<T>>;
   }): Promise<void> {
     const event = args.event;
-    const event_id = args.event_id || new Id(IdGenerator.generateRandomId());
+    const event_id = args.event_id || Id.random();
     let info: EventInfo;
     switch (true) {
       case event instanceof CreationEvent: {
@@ -121,14 +123,14 @@ class PostgresEventStore implements EventStore {
     }
   }
 
-  private async insert<E extends EventClass<E, any>>(edata: EventData<E>) {
+  private async insert<E extends Event<any>>(edata: EventData<E>) {
     const sql = `
       INSERT INTO ${this.eventStoreTable} (
           event_id, aggregate_id, causation_id, correlation_id,
           aggregate_version, json_payload, json_metadata, recorded_on, event_name
       ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)`;
 
-    const serialized = encode(schema_EventData(edata.event.schema), edata);
+    const serialized = this.schemas.encode(edata);
 
     const values = [
       // @ts-ignore

src/app/events.ts

@@ -0,0 +1,21 @@
+/*
+  Schemas for all application events
+*/
+export { schemas };
+
+import { Schemas, CSchema, TSchema } from '@/lib/eventSourcing/eventStore';
+import { ApplicationSubmitted } from '@/domain/cookingClub/membership2/events/membership/applicationSubmitted';
+import { ApplicationEvaluated } from '@/domain/cookingClub/membership2/events/membership/applicationEvaluated';
+
+const schemas = new Schemas([
+  new CSchema(
+    ApplicationSubmitted.aggregate,
+    ApplicationSubmitted.schema,
+    ApplicationSubmitted.type,
+  ),
+  new TSchema(
+    ApplicationEvaluated.aggregate,
+    ApplicationEvaluated.schema,
+    ApplicationEvaluated.type,
+  ),
+]);

src/app/handleCommand.ts

@@ -0,0 +1,71 @@
+export { handleCommand, type CommandController, type CommandHandler };
+
+import { Response } from '@/lib/router';
+import { EventStore } from '@/lib/eventSourcing/eventStore';
+import { Decoder, decode } from '@/lib/json/decoder';
+import * as express from 'express';
+import * as router from '@/lib/router';
+import { Future } from '@/lib/Future';
+import { Result, Failure } from '@/lib/Result';
+import { Repositories, Projections, allProjections } from '@/app/projections';
+import { Services } from '@/app/services';
+import { WithProjectionStore } from '@/app/projectionStore';
+import { WithEventStore } from '@/app/eventStore';
+
+type CommandHandler<Command> = (v: {
+  command: Command;
+  store: EventStore;
+  projections: Projections;
+  services: Services;
+}) => Future<Response, Response>;
+
+type CommandController<Command> = {
+  decoder: Decoder<Command>;
+  handler: CommandHandler<Command>;
+};
+
+const onStoreError = (_: Error): Response =>
+  router.json({
+    status: 500,
+    content: { message: 'Internal Server Error' },
+  });
+
+function handleCommand<Command>(
+  withEventStore: WithEventStore,
+  withProjectionStore: WithProjectionStore,
+  services: Services,
+  repositories: Repositories,
+  { decoder, handler }: CommandController<Command>,
+): express.Handler {
+  return router.route((req) =>
+    decodeCommand(decoder, req).chain((command) =>
+      withProjectionStore(onStoreError, (projectionStore) =>
+        withEventStore(onStoreError, (store) =>
+          handler({
+            command,
+            store,
+            projections: allProjections(repositories, projectionStore),
+            services,
+          }),
+        ),
+      ),
+    ),
+  );
+}
+
+function decodeCommand<C>(
+  decoder: Decoder<C>,
+  req: express.Request,
+): Future<Response, C> {
+  const decoded: Result<string, C> = decode(decoder, req.body);
+  if (decoded instanceof Failure) {
+    return Future.reject(
+      router.json({
+        status: 400,
+        content: { message: `Unable to decode command: ${decoded.error}` },
+      }),
+    );
+  }
+
+  return Future.resolve(decoded.value);
+}