[fabric/sqlite-store] Implement SQLiteEventStore

package.json

@@ -9,18 +9,18 @@
     "apps/**/*"
   ],
   "devDependencies": {
-    "@eslint/js": "^9.10.0",
+    "@eslint/js": "^9.12.0",
     "@types/eslint": "^9.6.1",
     "@types/eslint__js": "^8.42.3",
     "cross-env": "^7.0.3",
-    "eslint": "^9.10.0",
+    "eslint": "^9.12.0",
     "husky": "^9.1.6",
     "lint-staged": "^15.2.10",
     "prettier": "^3.3.3",
     "tsx": "^4.19.1",
-    "typescript": "^5.6.2",
+    "typescript": "^5.6.3",
-    "typescript-eslint": "^8.6.0",
+    "typescript-eslint": "^8.8.1",
-    "zx": "^8.1.7"
+    "zx": "^8.1.9"
   },
   "scripts": {
     "lint": "eslint . --fix --report-unused-disable-directives",

packages/fabric/core/package.json

@@ -1,5 +1,7 @@
 {
   "name": "@fabric/core",
+  "private": true,
+  "sideEffects": false,
   "type": "module",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -9,19 +11,15 @@
   "files": [
     "dist"
   ],
-  "private": true,
   "packageManager": "yarn@4.1.1",
   "devDependencies": {
-    "@types/validator": "^13.12.2",
+    "@vitest/coverage-v8": "^2.1.2",
-    "typescript": "^5.6.2",
+    "typescript": "^5.6.3",
-    "vitest": "^2.1.1"
+    "vitest": "^2.1.2"
   },
   "scripts": {
     "test": "vitest",
+    "coverage": "vitest run --coverage",
     "build": "tsc -p tsconfig.build.json"
-  },
-  "sideEffects": false,
-  "dependencies": {
-    "validator": "^13.12.0"
   }
 }

packages/fabric/core/src/result/async-result.ts

@@ -1,6 +1,7 @@
 /* eslint-disable @typescript-eslint/no-explicit-any */
 import { TaggedError } from "../error/tagged-error.js";
 import { UnexpectedError } from "../error/unexpected-error.js";
+import { MaybePromise } from "../types/maybe-promise.js";
 import { Result } from "./result.js";
 
 /**
@@ -14,7 +15,7 @@ export type AsyncResult<
 
 export namespace AsyncResult {
   export async function tryFrom<T, TError extends TaggedError>(
-    fn: () => Promise<T>,
+    fn: () => MaybePromise<T>,
     errorMapper: (error: any) => TError,
   ): AsyncResult<T, TError> {
     try {
@@ -25,8 +26,8 @@ export namespace AsyncResult {
   }
 
   export async function from<T>(
-    fn: () => Promise<T>,
+    fn: () => MaybePromise<T>,
-  ): AsyncResult<T, UnexpectedError> {
+  ): AsyncResult<T, never> {
-    return tryFrom(fn, (error) => new UnexpectedError(error));
+    return tryFrom(fn, (error) => new UnexpectedError(error) as never);
   }
 }

packages/fabric/core/src/run/run.ts

@@ -21,6 +21,18 @@ export namespace Run {
     fn2: (value: T1) => AsyncResult<T2, TE2>,
     fn3: (value: T2) => AsyncResult<T3, TE3>,
   ): AsyncResult<T3, TE1 | TE2 | TE3>;
+  // prettier-ignore
+  export async function seq<
+    T1, TE1 extends TaggedError,
+    T2, TE2 extends TaggedError,
+    T3, TE3 extends TaggedError,
+    T4, TE4 extends TaggedError,
+  >(
+    fn1: () => AsyncResult<T1, TE1>,
+    fn2: (value: T1) => AsyncResult<T2, TE2>,
+    fn3: (value: T2) => AsyncResult<T3, TE3>,
+    fn4: (value: T3) => AsyncResult<T4, TE4>,
+  ): AsyncResult<T4, TE1 | TE2 | TE3 | TE4>;
   export async function seq(
     ...fns: ((...args: any[]) => AsyncResult<any, any>)[]
   ): AsyncResult<any, any> {

packages/fabric/core/src/types/record.ts

@@ -0,0 +1 @@
+export type EmptyRecord = Record<string, never>;

packages/fabric/core/vitest.config.ts

@@ -3,7 +3,7 @@ import { defineConfig } from "vitest/config";
 export default defineConfig({
   test: {
     coverage: {
-      exclude: ["**/index.ts"],
+      exclude: ["dist/**", "vitest.config.ts", "**/index.ts", "**/*.spec.ts"],
     },
     passWithNoTests: true,
   },

packages/fabric/domain/package.json

@@ -1,5 +1,7 @@
 {
   "name": "@fabric/domain",
+  "private": true,
+  "sideEffects": false,
   "type": "module",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -10,12 +12,11 @@
   "files": [
     "dist"
   ],
-  "private": true,
   "packageManager": "yarn@4.1.1",
   "devDependencies": {
-    "@fabric/sqlite-store": "workspace:^",
+    "@vitest/coverage-v8": "^2.1.2",
-    "typescript": "^5.6.2",
+    "typescript": "^5.6.3",
-    "vitest": "^2.1.1"
+    "vitest": "^2.1.2"
   },
   "dependencies": {
     "@fabric/core": "workspace:^",
@@ -23,6 +24,7 @@
   },
   "scripts": {
     "test": "vitest",
+    "coverage": "vitest run --coverage",
     "build": "tsc -p tsconfig.build.json"
   }
 }

packages/fabric/domain/src/events/event-store.ts

@@ -1,21 +1,31 @@
-import { AsyncResult, PosixDate } from "@fabric/core";
+import { AsyncResult, MaybePromise, PosixDate } from "@fabric/core";
 import { StoreQueryError } from "../errors/query-error.js";
-import { EventsFromStream, EventStream } from "./event-stream.js";
+import { UUID } from "../types/uuid.js";
+import { Event, EventFromKey } from "./event.js";
 import { StoredEvent } from "./stored-event.js";
 
-export interface EventStore<TEventStream extends EventStream> {
+export interface EventStore<TEvents extends Event> {
   /**
    * Store a new event in the event store.
    */
-  append<
+  append<T extends TEvents>(
-    TStreamKey extends TEventStream["name"],
-    T extends EventsFromStream<TEventStream, TStreamKey>,
-  >(
-    streamName: TStreamKey,
     event: T,
   ): AsyncResult<StoredEvent<T>, StoreQueryError>;
+
+  getEventsFromStream(
+    streamId: UUID,
+  ): AsyncResult<StoredEvent<TEvents>[], StoreQueryError>;
+
+  subscribe<TEventKey extends TEvents["type"]>(
+    events: TEventKey[],
+    subscriber: EventSubscriber<EventFromKey<TEvents, TEventKey>>,
+  ): void;
 }
 
+export type EventSubscriber<TEvents extends Event = Event> = (
+  event: StoredEvent<TEvents>,
+) => MaybePromise<void>;
+
 export interface EventFilterOptions {
   fromDate?: PosixDate;
   toDate?: PosixDate;

packages/fabric/domain/src/events/event-stream.ts

@@ -1,16 +0,0 @@
-import { AsyncResult } from "@fabric/core";
-import { UUID } from "../types/uuid.js";
-import { Event } from "./event.js";
-import { StoredEvent } from "./stored-event.js";
-
-export interface EventStream<
-  TName extends string = string,
-  TEvent extends Event = Event,
-> {
-  id: UUID;
-  name: TName;
-  append<T extends TEvent>(event: Event): AsyncResult<StoredEvent<T>>;
-}
-
-export type EventsFromStream<T extends EventStream, TKey extends string> =
-  T extends EventStream<TKey, infer TEvent> ? TEvent : never;

packages/fabric/domain/src/events/event.ts

@@ -10,3 +10,8 @@ export interface Event<TTag extends string = string, TPayload = any> {
   readonly streamId: UUID;
   readonly payload: TPayload;
 }
+
+export type EventFromKey<
+  TEvents extends Event,
+  TKey extends TEvents["type"],
+> = Extract<TEvents, { type: TKey }>;

packages/fabric/domain/src/events/index.ts

@@ -1,4 +1,3 @@
 export * from "./event-store.js";
-export * from "./event-stream.js";
 export * from "./event.js";
 export * from "./stored-event.js";

packages/fabric/domain/src/files/is-in-memory-file.ts

@@ -1,19 +1,14 @@
 import { isRecord } from "@fabric/core";
-import validator from "validator";
 import { InMemoryFile } from "./in-memory-file.js";
 
-const { isBase64, isMimeType } = validator;
-
 export function isInMemoryFile(value: unknown): value is InMemoryFile {
   try {
     return (
       isRecord(value) &&
       "data" in value &&
       typeof value.data === "string" &&
-      isBase64(value.data.split(",")[1]) &&
       "mimeType" in value &&
       typeof value.mimeType === "string" &&
-      isMimeType(value.mimeType) &&
       "name" in value &&
       typeof value.name === "string" &&
       "sizeInBytes" in value &&

packages/fabric/domain/src/index.ts

@@ -6,3 +6,4 @@ export * from "./security/index.js";
 export * from "./services/index.js";
 export * from "./types/index.js";
 export * from "./use-case/index.js";
+export * from "./utils/index.js";

packages/fabric/domain/src/utils/index.ts

@@ -1 +1,2 @@
+export * from "./json-utils.js";
 export * from "./sort-by-dependencies.js";

packages/fabric/domain/vitest.config.ts

@@ -3,7 +3,7 @@ import { defineConfig } from "vitest/config";
 export default defineConfig({
   test: {
     coverage: {
-      exclude: ["**/index.ts"],
+      exclude: ["dist/**", "vitest.config.ts", "**/index.ts", "**/*.spec.ts"],
     },
     passWithNoTests: true,
   },

packages/fabric/sqlite-store/package.json

@@ -1,5 +1,7 @@
 {
   "name": "@fabric/sqlite-store",
+  "private": true,
+  "sideEffects": false,
   "type": "module",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -9,11 +11,11 @@
   "files": [
     "dist"
   ],
-  "private": true,
   "packageManager": "yarn@4.1.1",
   "devDependencies": {
-    "typescript": "^5.6.2",
+    "@vitest/coverage-v8": "^2.1.2",
-    "vitest": "^2.1.1"
+    "typescript": "^5.6.3",
+    "vitest": "^2.1.2"
   },
   "dependencies": {
     "@fabric/core": "workspace:^",
@@ -22,6 +24,7 @@
   },
   "scripts": {
     "test": "vitest",
+    "coverage": "vitest run --coverage",
     "build": "tsc -p tsconfig.build.json"
   }
 }

packages/fabric/sqlite-store/src/events/event-store.spec.ts

@@ -0,0 +1,77 @@
+import { PosixDate, Run } from "@fabric/core";
+import { Event } from "@fabric/domain";
+import { UUIDGeneratorMock } from "@fabric/domain/mocks";
+import { afterEach, beforeEach, describe, expect, it, vitest } from "vitest";
+import { SQLiteEventStore } from "./event-store.js";
+
+describe("Event Store", () => {
+  type UserCreated = Event<"UserCreated", { name: string }>;
+  type UserUpdated = Event<"UserUpdated", { name: string }>;
+  type UserDeleted = Event<"UserDeleted", void>;
+
+  type UserEvents = UserCreated | UserUpdated | UserDeleted;
+
+  let store: SQLiteEventStore<UserEvents>;
+
+  beforeEach(async () => {
+    store = new SQLiteEventStore(":memory:");
+    await Run.UNSAFE(() => store.migrate());
+  });
+
+  afterEach(async () => {
+    await Run.UNSAFE(() => store.close());
+  });
+
+  it("Should append an event", async () => {
+    const newUUID = UUIDGeneratorMock.generate();
+
+    const userCreated: UserCreated = {
+      type: "UserCreated",
+      id: newUUID,
+      streamId: newUUID,
+      payload: { name: "test" },
+    };
+
+    await Run.UNSAFE(() => store.append(userCreated));
+
+    const events = await Run.UNSAFE(() => store.getEventsFromStream(newUUID));
+
+    expect(events).toHaveLength(1);
+
+    expect(events[0]).toEqual({
+      id: newUUID,
+      streamId: newUUID,
+      type: "UserCreated",
+      version: BigInt(1),
+      timestamp: expect.any(PosixDate),
+      payload: { name: "test" },
+    });
+  });
+
+  it("should notify subscribers on append", async () => {
+    const newUUID = UUIDGeneratorMock.generate();
+
+    const userCreated: UserCreated = {
+      type: "UserCreated",
+      id: newUUID,
+      streamId: newUUID,
+      payload: { name: "test" },
+    };
+
+    const subscriber = vitest.fn();
+
+    store.subscribe(["UserCreated"], subscriber);
+
+    await Run.UNSAFE(() => store.append(userCreated));
+
+    expect(subscriber).toHaveBeenCalledTimes(1);
+    expect(subscriber).toHaveBeenCalledWith({
+      id: newUUID,
+      streamId: newUUID,
+      type: "UserCreated",
+      version: BigInt(1),
+      timestamp: expect.any(PosixDate),
+      payload: { name: "test" },
+    });
+  });
+});

packages/fabric/sqlite-store/src/events/event-store.ts

@@ -0,0 +1,176 @@
+import { AsyncResult, MaybePromise, PosixDate, Run } from "@fabric/core";
+import {
+  Event,
+  EventFromKey,
+  EventStore,
+  EventSubscriber,
+  JSONUtils,
+  StoredEvent,
+  StoreQueryError,
+  UUID,
+} from "@fabric/domain";
+import { SQLiteDatabase } from "../sqlite/sqlite-database.js";
+
+export class SQLiteEventStore<TEvents extends Event>
+  implements EventStore<TEvents>
+{
+  private db: SQLiteDatabase;
+
+  private streamVersions = new Map<UUID, bigint>();
+
+  private eventSubscribers = new Map<
+    TEvents["type"],
+    EventSubscriber<TEvents>[]
+  >();
+
+  constructor(private readonly dbPath: string) {
+    this.db = new SQLiteDatabase(dbPath);
+  }
+
+  async migrate(): AsyncResult<void, StoreQueryError> {
+    return AsyncResult.tryFrom(
+      async () => {
+        await this.db.init();
+        await this.db.run(
+          `CREATE TABLE IF NOT EXISTS events (
+              id TEXT PRIMARY KEY,
+              type TEXT NOT NULL,
+              streamId TEXT NOT NULL,
+              version INTEGER NOT NULL,
+              timestamp NUMERIC NOT NULL,
+              payload TEXT NOT NULL,
+              UNIQUE(streamId, version)
+            )`,
+        );
+      },
+      (error) => new StoreQueryError(error.message, { error }),
+    );
+  }
+
+  async getEventsFromStream(
+    streamId: UUID,
+  ): AsyncResult<StoredEvent<TEvents>[], StoreQueryError> {
+    return AsyncResult.tryFrom(
+      async () => {
+        const events = await this.db.allPrepared(
+          `SELECT * FROM events WHERE streamId = $id`,
+          {
+            $id: streamId,
+          },
+          (event) => ({
+            id: event.id,
+            streamId: event.streamId,
+            type: event.type,
+            version: BigInt(event.version),
+            timestamp: new PosixDate(event.timestamp),
+            payload: JSONUtils.parse(event.payload),
+          }),
+        );
+        return events;
+      },
+      (error) => new StoreQueryError(error.message, { error }),
+    );
+  }
+
+  async append<T extends TEvents>(
+    event: T,
+  ): AsyncResult<StoredEvent<T>, StoreQueryError> {
+    return Run.seq(
+      () => this.getLastVersion(event.streamId),
+      (version) =>
+        AsyncResult.from(() => {
+          this.streamVersions.set(event.streamId, version + 1n);
+          return version;
+        }),
+      (version) => this.storeEvent(event.streamId, version + 1n, event),
+      (storedEvent) =>
+        AsyncResult.from(async () => {
+          await this.notifySubscribers(storedEvent);
+          return storedEvent;
+        }),
+    );
+  }
+
+  private async notifySubscribers(
+    event: StoredEvent<TEvents>,
+  ): AsyncResult<void> {
+    return AsyncResult.from(async () => {
+      const subscribers = this.eventSubscribers.get(event.type) || [];
+      await Promise.all(subscribers.map((subscriber) => subscriber(event)));
+    });
+  }
+
+  private async getLastVersion(
+    streamId: UUID,
+  ): AsyncResult<bigint, StoreQueryError> {
+    return AsyncResult.tryFrom(
+      async () => {
+        const { lastVersion } = await this.db.onePrepared(
+          `SELECT max(version) as lastVersion FROM events WHERE streamId = $id`,
+          {
+            $id: streamId,
+          },
+        );
+
+        return !lastVersion ? 0n : BigInt(lastVersion);
+      },
+      (error) =>
+        new StoreQueryError(`Error getting last version:${error.message}`, {
+          error,
+        }),
+    );
+  }
+
+  subscribe<TEventKey extends TEvents["type"]>(
+    events: TEventKey[],
+    subscriber: (
+      event: StoredEvent<EventFromKey<TEvents, TEventKey>>,
+    ) => MaybePromise<void>,
+  ): void {
+    events.forEach((event) => {
+      const subscribers = this.eventSubscribers.get(event) || [];
+      const newSubscribers = [
+        ...subscribers,
+        subscriber,
+      ] as EventSubscriber<TEvents>[];
+      this.eventSubscribers.set(event, newSubscribers);
+    });
+  }
+
+  close(): AsyncResult<void, StoreQueryError> {
+    return AsyncResult.tryFrom(
+      () => this.db.close(),
+      (error) => new StoreQueryError(error.message, { error }),
+    );
+  }
+
+  private storeEvent<T extends Event>(
+    streamId: UUID,
+    version: bigint,
+    event: T,
+  ): AsyncResult<StoredEvent<T>, StoreQueryError> {
+    return AsyncResult.tryFrom(
+      async () => {
+        const storedEvent: StoredEvent<T> = {
+          ...event,
+          version: version,
+          timestamp: new PosixDate(),
+        };
+        await this.db.runPrepared(
+          `INSERT INTO events (id, streamId, type, version, timestamp, payload) 
+          VALUES ($id, $streamId, $type, $version, $timestamp, $payload)`,
+          {
+            $id: storedEvent.id,
+            $streamId: streamId,
+            $type: storedEvent.type,
+            $version: storedEvent.version.toString(),
+            $timestamp: storedEvent.timestamp.timestamp,
+            $payload: JSON.stringify(storedEvent.payload),
+          },
+        );
+        return storedEvent;
+      },
+      (error) => new StoreQueryError("Error appending event", { error }),
+    );
+  }
+}

packages/fabric/sqlite-store/src/events/index.ts

@@ -0,0 +1 @@
+export * from "./event-store.js";

packages/fabric/sqlite-store/src/sqlite/sqlite-database.ts

@@ -79,7 +79,11 @@ export class SQLiteDatabase {
           if (err) {
             reject(err);
           } else {
-            resolve(transformer ? rows.map(transformer) : rows);
+            try {
+              resolve(transformer ? rows.map(transformer) : rows);
+            } catch (e) {
+              reject(e);
+            }
           }
         },
       );
@@ -100,7 +104,11 @@ export class SQLiteDatabase {
           if (err) {
             reject(err);
           } else {
-            resolve(transformer ? rows.map(transformer)[0] : rows[0]);
+            try {
+              resolve(transformer ? rows.map(transformer)[0] : rows[0]);
+            } catch (e) {
+              reject(e);
+            }
           }
         },
       );

packages/fabric/sqlite-store/src/state/query-builder.ts

@@ -14,7 +14,7 @@ import {
 } from "@fabric/domain";
 import { filterToParams, filterToSQL } from "../sqlite/filter-to-sql.js";
 import { transformRow } from "../sqlite/sql-to-value.js";
-import { SQLiteDatabase } from "../sqlite/sqlite-wrapper.js";
+import { SQLiteDatabase } from "../sqlite/sqlite-database.js";
 
 export class QueryBuilder<T> implements StoreQuery<T> {
   constructor(

packages/fabric/sqlite-store/src/state/state-store.ts

@@ -16,7 +16,7 @@ import {
   recordToSQLParams,
   recordToSQLSet,
 } from "../sqlite/record-utils.js";
-import { SQLiteDatabase } from "../sqlite/sqlite-wrapper.js";
+import { SQLiteDatabase } from "../sqlite/sqlite-database.js";
 import { QueryBuilder } from "./query-builder.js";
 
 export class SQLiteStateStore<TModel extends Model>
@@ -26,7 +26,7 @@ export class SQLiteStateStore<TModel extends Model>
   private db: SQLiteDatabase;
 
   constructor(
-    private dbPath: string,
+    private readonly dbPath: string,
     models: TModel[],
   ) {
     this.schema = models.reduce((acc, model: TModel) => {

packages/fabric/sqlite-store/vitest.config.ts

@@ -3,7 +3,7 @@ import { defineConfig } from "vitest/config";
 export default defineConfig({
   test: {
     coverage: {
-      exclude: ["**/index.ts"],
+      exclude: ["dist/**", "vitest.config.ts", "**/index.ts", "**/*.spec.ts"],
     },
     passWithNoTests: true,
   },

packages/templates/domain/package.json

@@ -1,16 +1,18 @@
 {
   "name": "@ulthar/template-domain",
+  "private": true,
+  "sideEffects": false,
   "type": "module",
   "module": "dist/index.js",
   "main": "dist/index.js",
   "files": [
     "dist"
   ],
-  "private": true,
   "packageManager": "yarn@4.1.1",
   "devDependencies": {
-    "typescript": "^5.6.2",
+    "@vitest/coverage-v8": "^2.1.2",
-    "vitest": "^2.1.1"
+    "typescript": "^5.6.3",
+    "vitest": "^2.1.2"
   },
   "dependencies": {
     "@fabric/core": "workspace:^",
@@ -18,6 +20,7 @@
   },
   "scripts": {
     "test": "vitest",
+    "coverage": "vitest run --coverage",
     "build": "tsc -p tsconfig.build.json"
   }
 }

packages/templates/domain/vitest.config.ts

@@ -3,7 +3,7 @@ import { defineConfig } from "vitest/config";
 export default defineConfig({
   test: {
     coverage: {
-      exclude: ["**/index.ts"],
+      exclude: ["dist/**", "vitest.config.ts", "**/index.ts", "**/*.spec.ts"],
     },
     passWithNoTests: true,
   },

packages/templates/lib/package.json

@@ -1,5 +1,7 @@
 {
   "name": "@ulthar/lib-template",
+  "private": true,
+  "sideEffects": false,
   "type": "module",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -9,17 +11,18 @@
   "files": [
     "dist"
   ],
-  "private": true,
   "packageManager": "yarn@4.1.1",
   "devDependencies": {
-    "typescript": "^5.6.2",
+    "@vitest/coverage-v8": "^2.1.2",
-    "vitest": "^2.1.1"
+    "typescript": "^5.6.3",
+    "vitest": "^2.1.2"
   },
   "dependencies": {
     "@fabric/core": "workspace:^"
   },
   "scripts": {
     "test": "vitest",
+    "coverage": "vitest run --coverage",
     "build": "tsc -p tsconfig.build.json"
   }
 }

packages/templates/lib/vitest.config.ts

@@ -3,7 +3,7 @@ import { defineConfig } from "vitest/config";
 export default defineConfig({
   test: {
     coverage: {
-      exclude: ["**/index.ts"],
+      exclude: ["dist/**", "vitest.config.ts", "**/index.ts", "**/*.spec.ts"],
     },
     passWithNoTests: true,
   },