Custom Adapters

import { MongoClient, type ClientSession } from "mongodb";
import type { UnitOfWork, UnitOfWorkFactory, Event } from "@noddde/core";

/**
 * Shared mutable ref so persistence classes see the active transaction session.
 * This is the bridge between the UoW and persistence classes.
 */
interface SessionStore {
  current: ClientSession | null;
}

class MongoUnitOfWork implements UnitOfWork {
  private operations: (() => Promise<void>)[] = [];
  private events: Event[] = [];
  private sealed = false;

  constructor(
    private readonly client: MongoClient,
    private readonly sessionStore: SessionStore,
  ) {}

  enlist(op: () => Promise<void>) {
    if (this.sealed) throw new Error("UnitOfWork already completed");
    this.operations.push(op);
  }

  deferPublish(...events: Event[]) {
    if (this.sealed) throw new Error("UnitOfWork already completed");
    this.events.push(...events);
  }

  async commit(): Promise<Event[]> {
    if (this.sealed) throw new Error("UnitOfWork already completed");
    this.sealed = true;

    const session = this.client.startSession();
    try {
      session.startTransaction();
      this.sessionStore.current = session;

      for (const op of this.operations) {
        await op();
      }

      await session.commitTransaction();
      return this.events;
    } catch (err) {
      await session.abortTransaction();
      throw err;
    } finally {
      this.sessionStore.current = null;
      await session.endSession();
    }
  }

  async rollback(): Promise<void> {
    if (this.sealed) throw new Error("UnitOfWork already completed");
    this.sealed = true;
    this.operations = [];
    this.events = [];
  }
}

import type { Db, Collection } from "mongodb";
import { ConcurrencyError } from "@noddde/core";
import type { EventSourcedAggregatePersistence, Event } from "@noddde/core";

class MongoEventSourcedPersistence implements EventSourcedAggregatePersistence {
  private readonly collection: Collection;

  constructor(
    db: Db,
    private readonly sessionStore: SessionStore,
  ) {
    this.collection = db.collection("noddde_events");
  }

  /** Returns session options if inside a UoW, or empty object otherwise. */
  private get sessionOpts() {
    return this.sessionStore.current
      ? { session: this.sessionStore.current }
      : {};
  }

  async save(
    aggregateName: string,
    aggregateId: string,
    events: Event[],
    expectedVersion: number,
  ) {
    if (events.length === 0) return;

    const docs = events.map((event, i) => ({
      aggregateName,
      aggregateId,
      sequenceNumber: expectedVersion + i + 1,
      eventName: event.name,
      payload: event.payload,
      metadata: event.metadata ?? null,
      createdAt: event.metadata?.timestamp
        ? new Date(event.metadata.timestamp)
        : new Date(),
    }));

    try {
      await this.collection.insertMany(docs, this.sessionOpts);
    } catch (err: any) {
      if (err.code === 11000) {
        // duplicate key
        throw new ConcurrencyError(
          aggregateName,
          aggregateId,
          expectedVersion,
          -1,
        );
      }
      throw err;
    }
  }

  async load(aggregateName: string, aggregateId: string): Promise<Event[]> {
    const docs = await this.collection
      .find({ aggregateName, aggregateId }, this.sessionOpts)
      .sort({ sequenceNumber: 1 })
      .toArray();

    return docs.map((doc) => ({
      name: doc.eventName,
      payload: doc.payload,
      ...(doc.metadata ? { metadata: doc.metadata } : {}),
    }));
  }
}

import type { StateStoredAggregatePersistence } from "@noddde/core";

class MongoStateStoredPersistence implements StateStoredAggregatePersistence {
  private readonly collection: Collection;

  constructor(
    db: Db,
    private readonly sessionStore: SessionStore,
  ) {
    this.collection = db.collection("noddde_aggregate_states");
  }

  private get sessionOpts() {
    return this.sessionStore.current
      ? { session: this.sessionStore.current }
      : {};
  }

  async save(
    aggregateName: string,
    aggregateId: string,
    state: any,
    expectedVersion: number,
  ) {
    if (expectedVersion === 0) {
      // New aggregate: insert with version 1
      try {
        await this.collection.insertOne(
          { aggregateName, aggregateId, state, version: 1 },
          this.sessionOpts,
        );
      } catch (err: any) {
        if (err.code === 11000) {
          throw new ConcurrencyError(aggregateName, aggregateId, 0, -1);
        }
        throw err;
      }
    } else {
      // Existing aggregate: update only if version matches
      const result = await this.collection.findOneAndUpdate(
        { aggregateName, aggregateId, version: expectedVersion },
        { $set: { state, version: expectedVersion + 1 } },
        this.sessionOpts,
      );
      if (!result) {
        throw new ConcurrencyError(
          aggregateName,
          aggregateId,
          expectedVersion,
          -1,
        );
      }
    }
  }

  async load(aggregateName: string, aggregateId: string) {
    const doc = await this.collection.findOne(
      { aggregateName, aggregateId },
      this.sessionOpts,
    );
    return doc ? { state: doc.state, version: doc.version } : null;
  }
}

import type { PersistenceAdapter } from "@noddde/core";
import { MongoClient, type Db } from "mongodb";

class MongoAdapter implements PersistenceAdapter {
  readonly unitOfWorkFactory;
  readonly eventSourcedPersistence;
  readonly stateStoredPersistence;
  readonly sagaPersistence;
  // Add optional stores as needed:
  // readonly snapshotStore;
  // readonly outboxStore;

  private readonly sessionStore: SessionStore = { current: null };

  constructor(
    private readonly client: MongoClient,
    db: Db,
  ) {
    this.unitOfWorkFactory = () =>
      new MongoUnitOfWork(client, this.sessionStore);
    this.eventSourcedPersistence = new MongoEventSourcedPersistence(
      db,
      this.sessionStore,
    );
    this.stateStoredPersistence = new MongoStateStoredPersistence(
      db,
      this.sessionStore,
    );
    this.sagaPersistence = new MongoSagaPersistence(db, this.sessionStore);
  }

  /** Create required indexes on first use. */
  async init() {
    const db = this.client.db();
    await db
      .collection("noddde_events")
      .createIndex(
        { aggregateName: 1, aggregateId: 1, sequenceNumber: 1 },
        { unique: true },
      );
    await db
      .collection("noddde_aggregate_states")
      .createIndex({ aggregateName: 1, aggregateId: 1 }, { unique: true });
  }

  async close() {
    await this.client.close();
  }
}

import { defineDomain } from "@noddde/core";
import { wireDomain } from "@noddde/engine";
import { MongoClient } from "mongodb";

const client = new MongoClient(process.env.MONGODB_URI!);
await client.connect();
const db = client.db("my-app");

const adapter = new MongoAdapter(client, db);

const domain = await wireDomain(
  defineDomain({
    writeModel: { aggregates: { Order } },
    readModel: { projections: {} },
  }),
  { persistenceAdapter: adapter },
);