New event store and projection store abstractions (#20)

Author: lazamar

package-lock.json

@@ -4,6 +4,7 @@
   "main": "dist/index.js",
   "scripts": {
     "build": "tsc",
+    "watch": "tsc --watch",
     "start": "tsx src/index.ts",
     "test": "tsx tests/unit/main.ts",
     "format": "prettier --write .",
@@ -14,6 +15,7 @@
     "class-validator": "^0.14.2",
     "express": "^4.18.2",
     "fluture": "^14.0.0",
+    "luxon": "^3.7.2",
     "minio": "^8.0.5",
     "mongodb": "^5.4.0",
     "nodemailer": "^7.0.5",
@@ -29,6 +31,7 @@
   },
   "devDependencies": {
     "@types/express": "^4.17.21",
+    "@types/luxon": "^3.7.1",
     "@types/node": "^24.3.0",
     "@types/nodemailer": "^7.0.1",
     "@types/pg": "^8.6.6",

package.json

@@ -0,0 +1,59 @@
+export { handleCommand };
+
+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';
+
+type Projections = {};
+type Services = {};
+
+type CommandController<Command> = {
+  decoder: Decoder<Command>;
+  handler: (v: {
+    command: Command;
+    store: EventStore;
+    projections: Projections;
+    services: Services;
+  }) => Future<Response, Response>;
+};
+
+function handleCommand<Command>(
+  withEventStore: <T>(f: (store: EventStore) => T) => T,
+  services: Services,
+  projections: Projections,
+  { decoder, handler }: CommandController<Command>,
+): express.Handler {
+  return router.route((req) =>
+    decodeCommand(decoder, req).chain((command) =>
+      withEventStore((store) =>
+        handler({
+          command,
+          store,
+          projections,
+          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);
+}

src/app/commandHandler.ts

@@ -0,0 +1,87 @@
+import {
+  Aggregate,
+  Id,
+  CreationEvent,
+  TransformationEvent,
+} from '@/lib/eventSourcing/event';
+
+import * as s from '@/lib/json/schema';
+
+class User implements Aggregate<User> {
+  constructor(
+    readonly aggregateId: Id<User>,
+    readonly aggregateVersion: number,
+    readonly name: string,
+  ) {}
+}
+
+export class CreateUser implements CreationEvent<CreateUser, User> {
+  static type: 'CreateUserr' = 'CreateUserr';
+  static schemaArgs = s.object({
+    type: s.stringLiteral(CreateUser.type),
+    aggregateId: Id.schema(),
+    name: s.string,
+  });
+  static schema = CreateUser.schemaArgs.dimap(
+    (v) => new CreateUser(v),
+    (v) => v.values,
+  );
+
+  schema = CreateUser.schema;
+  constructor(readonly values: s.Infer<typeof CreateUser.schemaArgs>) {}
+  createAggregate() {
+    return new User(new Id('wat'), 0, this.values.name);
+  }
+}
+
+export class AddName implements TransformationEvent<AddName, User> {
+  static type: 'AddName' = 'AddName';
+  constructor(readonly values: s.Infer<typeof AddName.schemaArgs>) {}
+
+  static schemaArgs = s.object({
+    type: s.stringLiteral(AddName.type),
+    aggregateId: Id.schema(),
+    name: s.string,
+  });
+
+  static schema = AddName.schemaArgs.dimap(
+    (v) => new AddName(v),
+    (v) => v.values,
+  );
+  readonly schema = AddName.schema;
+
+  transformAggregate(agg: User): User {
+    const u = new User(
+      agg.aggregateId,
+      agg.aggregateVersion + 1,
+      this.values.name,
+    );
+    return u;
+  }
+}
+
+export class RemoveName implements TransformationEvent<RemoveName, User> {
+  static type: 'RemoveName' = 'RemoveName';
+  constructor(readonly values: s.Infer<typeof RemoveName.schemaArgs>) {}
+
+  static schemaArgs = s.object({
+    type: s.stringLiteral(RemoveName.type),
+    aggregateId: Id.schema(),
+    name: s.string,
+  });
+
+  static schema = RemoveName.schemaArgs.dimap(
+    (v) => new RemoveName(v),
+    (v) => v.values,
+  );
+  readonly schema = RemoveName.schema;
+
+  transformAggregate(agg: User): User {
+    const u = new User(
+      agg.aggregateId,
+      agg.aggregateVersion + 1,
+      this.values.name,
+    );
+    return u;
+  }
+}