Persistence Adapters

import { DrizzleAdapter } from "@noddde/drizzle";
import { wireDomain } from "@noddde/engine";

const adapter = new DrizzleAdapter(db);

const domain = await wireDomain(definition, {
  persistenceAdapter: adapter,
  aggregates: {
    Room: { persistence: "event-sourced" },
    Inventory: {}, // defaults to state-stored from adapter
  },
});

yarn add @noddde/drizzle drizzle-orm
# Plus your database driver, e.g.:
yarn add better-sqlite3  # or: pg, mysql2

// SQLite
import {
  events,
  aggregateStates,
  sagaStates,
  snapshots,
  outbox,
} from "@noddde/drizzle/sqlite";

// PostgreSQL (uses serial PK, jsonb for payloads)
import {
  events,
  aggregateStates,
  sagaStates,
  snapshots,
  outbox,
} from "@noddde/drizzle/pg";

// MySQL (uses int auto-increment, varchar(255), json)
import {
  events,
  aggregateStates,
  sagaStates,
  snapshots,
  outbox,
} from "@noddde/drizzle/mysql";

import Database from "better-sqlite3";
import { drizzle } from "drizzle-orm/better-sqlite3";
import { DrizzleAdapter } from "@noddde/drizzle";
import { everyNEvents } from "@noddde/core";
import { defineDomain, wireDomain } from "@noddde/engine";

const db = drizzle(new Database("app.db"));
const adapter = new DrizzleAdapter(db);

const bankingDomain = defineDomain({
  writeModel: { aggregates: { BankAccount } },
  readModel: { projections: { BankAccount: BankAccountProjection } },
});

const domain = await wireDomain(bankingDomain, {
  persistenceAdapter: adapter,
  aggregates: {
    BankAccount: {
      persistence: "event-sourced",
      snapshots: { strategy: everyNEvents(100) },
    },
  },
});

import { DrizzleAdapter } from "@noddde/drizzle";
import { myEvents, mySagas } from "./schema"; // your own Drizzle tables

const db = drizzle(pool);
const adapter = new DrizzleAdapter(db, {
  tables: { eventStore: myEvents, sagaStore: mySagas },
});

const adapter = new DrizzleAdapter(db);

const domain = await wireDomain(definition, {
  persistenceAdapter: adapter,
  aggregates: {
    Payment: { persistence: "event-sourced" },
    Order: { persistence: adapter.stateStored(orders) },
    BankAccount: { persistence: adapter.stateStored(bankAccounts) },
  },
});

adapter.stateStored(orders, {
  aggregateId: orders.orderId,
  state: orders.orderData,
  version: orders.rev,
});

yarn add @noddde/prisma @prisma/client
yarn add -D prisma

model NodddeEvent {
  id             Int      @id @default(autoincrement())
  aggregateName  String   @map("aggregate_name")
  aggregateId    String   @map("aggregate_id")
  sequenceNumber Int      @map("sequence_number")
  eventName      String   @map("event_name")
  payload        String
  metadata       String?
  createdAt      DateTime @map("created_at")

  @@unique([aggregateName, aggregateId, sequenceNumber])
  @@map("noddde_events")
}

model NodddeAggregateState {
  aggregateName String @map("aggregate_name")
  aggregateId   String @map("aggregate_id")
  state         String
  version       Int    @default(0)

  @@id([aggregateName, aggregateId])
  @@map("noddde_aggregate_states")
}

model NodddeSagaState {
  sagaName String @map("saga_name")
  sagaId   String @map("saga_id")
  state    String

  @@id([sagaName, sagaId])
  @@map("noddde_saga_states")
}

model NodddeSnapshot {
  aggregateName String @map("aggregate_name")
  aggregateId   String @map("aggregate_id")
  state         String
  version       Int    @default(0)

  @@id([aggregateName, aggregateId])
  @@map("noddde_snapshots")
}

// Only needed if using the outbox pattern
model NodddeOutbox {
  id            String    @id
  event         String
  aggregateName String?   @map("aggregate_name")
  aggregateId   String?   @map("aggregate_id")
  createdAt     DateTime  @map("created_at")
  publishedAt   DateTime? @map("published_at")

  @@map("noddde_outbox")
}

import { PrismaClient } from "@prisma/client";
import { PrismaAdapter } from "@noddde/prisma";
import { everyNEvents } from "@noddde/core";
import { defineDomain, wireDomain } from "@noddde/engine";

const prisma = new PrismaClient();
const adapter = new PrismaAdapter(prisma);

const bankingDomain = defineDomain({
  writeModel: { aggregates: { BankAccount } },
  readModel: { projections: { BankAccount: BankAccountProjection } },
});

const domain = await wireDomain(bankingDomain, {
  persistenceAdapter: adapter,
  aggregates: {
    BankAccount: {
      persistence: "event-sourced",
      snapshots: { strategy: everyNEvents(100) },
    },
  },
});

const adapter = new PrismaAdapter(prisma);

const domain = await wireDomain(definition, {
  persistenceAdapter: adapter,
  aggregates: {
    Order: {
      persistence: adapter.stateStored("order", {
        aggregateId: "orderId",
        state: "orderData",
        version: "rev",
      }),
    },
  },
});

const adapter = new PrismaAdapter(prisma, { dialect: "postgresql" });

yarn add @noddde/typeorm typeorm reflect-metadata
# Plus your database driver, e.g.:
yarn add pg

import {
  NodddeEventEntity,
  NodddeAggregateStateEntity,
  NodddeSagaStateEntity,
  NodddeSnapshotEntity,
  NodddeOutboxEntryEntity, // only needed if using the outbox pattern
} from "@noddde/typeorm";

const dataSource = new DataSource({
  type: "postgres",
  url: process.env.DATABASE_URL,
  entities: [
    NodddeEventEntity,
    NodddeAggregateStateEntity,
    NodddeSagaStateEntity,
    NodddeSnapshotEntity,
    NodddeOutboxEntryEntity,
  ],
  synchronize: true, // use migrations in production
});
await dataSource.initialize();

import { TypeORMAdapter } from "@noddde/typeorm";
import { everyNEvents } from "@noddde/core";
import { defineDomain, wireDomain } from "@noddde/engine";

const adapter = new TypeORMAdapter(dataSource);

const bankingDomain = defineDomain({
  writeModel: { aggregates: { BankAccount } },
  readModel: { projections: { BankAccount: BankAccountProjection } },
});

const domain = await wireDomain(bankingDomain, {
  persistenceAdapter: adapter,
  aggregates: {
    BankAccount: {
      persistence: "event-sourced",
      snapshots: { strategy: everyNEvents(100) },
    },
  },
});

import { TypeORMAdapter } from "@noddde/typeorm";
import { OrderEntity } from "./entities"; // your own TypeORM entity

const adapter = new TypeORMAdapter(dataSource);

const domain = await wireDomain(definition, {
  persistenceAdapter: adapter,
  aggregates: {
    Order: {
      persistence: adapter.stateStored(OrderEntity, {
        aggregateId: "orderId",
        state: "orderData",
        version: "rev",
      }),
    },
  },
});

import type {
  PersistenceAdapter,
  EventSourcedAggregatePersistence,
  StateStoredAggregatePersistence,
  SagaPersistence,
  UnitOfWorkFactory,
  SnapshotStore,
  OutboxStore,
  IdempotencyStore,
  AggregateLocker,
} from "@noddde/core";

interface PersistenceAdapter {
  // --- Required ---
  unitOfWorkFactory: UnitOfWorkFactory;

  // --- Optional (engine errors at runtime if needed but missing) ---
  eventSourcedPersistence?: EventSourcedAggregatePersistence;
  stateStoredPersistence?: StateStoredAggregatePersistence;
  sagaPersistence?: SagaPersistence;
  snapshotStore?: SnapshotStore;
  outboxStore?: OutboxStore;
  idempotencyStore?: IdempotencyStore;
  aggregateLocker?: AggregateLocker;

  // --- Lifecycle hooks (optional) ---
  init?(): Promise<void>; // called by Domain.init()
  close?(): Promise<void>; // called by Domain.shutdown()
}

type UnitOfWorkFactory = () => UnitOfWork;

interface UnitOfWork {
  /** Buffer a write operation for deferred execution. */
  enlist(operation: () => Promise<void>): void;
  /** Schedule events for publishing after commit. */
  deferPublish(...events: Event[]): void;
  /** Execute all operations atomically, return deferred events. */
  commit(): Promise<Event[]>;
  /** Discard all operations and events. */
  rollback(): Promise<void>;
}

interface EventSourcedAggregatePersistence {
  save(
    aggregateName: string,
    aggregateId: string,
    events: Event[],
    expectedVersion: number,
  ): Promise<void>;
  load(aggregateName: string, aggregateId: string): Promise<Event[]>;
}

interface StateStoredAggregatePersistence {
  save(
    aggregateName: string,
    aggregateId: string,
    state: any,
    expectedVersion: number,
  ): Promise<void>;
  load(
    aggregateName: string,
    aggregateId: string,
  ): Promise<{ state: any; version: number } | null>;
}

interface SagaPersistence {
  save(sagaName: string, sagaId: string, state: any): Promise<void>;
  load(sagaName: string, sagaId: string): Promise<any | undefined | null>;
}

interface SnapshotStore {
  load(aggregateName: string, aggregateId: string): Promise<Snapshot | null>;
  save(
    aggregateName: string,
    aggregateId: string,
    snapshot: Snapshot,
  ): Promise<void>;
}
// Snapshot = { state: any; version: number }

interface OutboxStore {
  save(entries: OutboxEntry[]): Promise<void>;
  loadUnpublished(batchSize?: number): Promise<OutboxEntry[]>;
  markPublished(ids: string[]): Promise<void>;
  markPublishedByEventIds(eventIds: string[]): Promise<void>;
  deletePublished(olderThan?: Date): Promise<void>;
}

interface AggregateLocker {
  acquire(
    aggregateName: string,
    aggregateId: string,
    timeoutMs?: number,
  ): Promise<void>;
  release(aggregateName: string, aggregateId: string): Promise<void>;
}

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 { wireDomain, defineDomain } 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 },
);

CREATE TABLE IF NOT EXISTS noddde_events (
  id              SERIAL PRIMARY KEY,
  aggregate_name  TEXT NOT NULL,
  aggregate_id    TEXT NOT NULL,
  sequence_number INTEGER NOT NULL,
  event_name      TEXT NOT NULL,
  payload         JSONB NOT NULL,
  metadata        JSONB,                              -- event metadata envelope (nullable for backcompat)
  created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW()   -- populated from event.metadata.timestamp
);
CREATE UNIQUE INDEX IF NOT EXISTS noddde_events_stream_version_idx
  ON noddde_events (aggregate_name, aggregate_id, sequence_number);

CREATE TABLE IF NOT EXISTS noddde_aggregate_states (
  aggregate_name TEXT NOT NULL,
  aggregate_id   TEXT NOT NULL,
  state          JSONB NOT NULL,
  version        INTEGER NOT NULL DEFAULT 0,
  PRIMARY KEY (aggregate_name, aggregate_id)
);

CREATE TABLE IF NOT EXISTS noddde_saga_states (
  saga_name TEXT NOT NULL,
  saga_id   TEXT NOT NULL,
  state     JSONB NOT NULL,
  PRIMARY KEY (saga_name, saga_id)
);

CREATE TABLE IF NOT EXISTS noddde_snapshots (
  aggregate_name TEXT NOT NULL,
  aggregate_id   TEXT NOT NULL,
  state          JSONB NOT NULL,
  version        INTEGER NOT NULL,
  PRIMARY KEY (aggregate_name, aggregate_id)
);

CREATE TABLE IF NOT EXISTS noddde_outbox (
  id             TEXT PRIMARY KEY,
  event          JSONB NOT NULL,
  aggregate_name TEXT,
  aggregate_id   TEXT,
  created_at     TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  published_at   TIMESTAMPTZ
);

CREATE TABLE IF NOT EXISTS noddde_events (
  id              INT AUTO_INCREMENT PRIMARY KEY,
  aggregate_name  VARCHAR(255) NOT NULL,
  aggregate_id    VARCHAR(255) NOT NULL,
  sequence_number INT NOT NULL,
  event_name      VARCHAR(255) NOT NULL,
  payload         JSON NOT NULL,
  metadata        JSON,
  created_at      TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
  UNIQUE INDEX noddde_events_stream_version_idx (aggregate_name, aggregate_id, sequence_number)
);

CREATE TABLE IF NOT EXISTS noddde_aggregate_states (
  aggregate_name VARCHAR(255) NOT NULL,
  aggregate_id   VARCHAR(255) NOT NULL,
  state          TEXT NOT NULL,
  version        INT NOT NULL DEFAULT 0,
  PRIMARY KEY (aggregate_name, aggregate_id)
);

CREATE TABLE IF NOT EXISTS noddde_saga_states (
  saga_name VARCHAR(255) NOT NULL,
  saga_id   VARCHAR(255) NOT NULL,
  state     TEXT NOT NULL,
  PRIMARY KEY (saga_name, saga_id)
);

CREATE TABLE IF NOT EXISTS noddde_snapshots (
  aggregate_name VARCHAR(255) NOT NULL,
  aggregate_id   VARCHAR(255) NOT NULL,
  state          TEXT NOT NULL,
  version        INT NOT NULL,
  PRIMARY KEY (aggregate_name, aggregate_id)
);

CREATE TABLE IF NOT EXISTS noddde_outbox (
  id             VARCHAR(255) PRIMARY KEY,
  event          JSON NOT NULL,
  aggregate_name VARCHAR(255),
  aggregate_id   VARCHAR(255),
  created_at     TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
  published_at   TIMESTAMP(3)
);

CREATE TABLE IF NOT EXISTS noddde_events (
  id              INTEGER PRIMARY KEY AUTOINCREMENT,
  aggregate_name  TEXT NOT NULL,
  aggregate_id    TEXT NOT NULL,
  sequence_number INTEGER NOT NULL,
  event_name      TEXT NOT NULL,
  payload         TEXT NOT NULL,                -- JSON stored as text
  metadata        TEXT,                         -- JSON stored as text
  created_at      TEXT NOT NULL                 -- ISO 8601 string (SQLite has no native timestamp)
);
CREATE UNIQUE INDEX IF NOT EXISTS noddde_events_stream_version_idx
  ON noddde_events (aggregate_name, aggregate_id, sequence_number);

CREATE TABLE IF NOT EXISTS noddde_aggregate_states (
  aggregate_name TEXT NOT NULL,
  aggregate_id   TEXT NOT NULL,
  state          TEXT NOT NULL,
  version        INTEGER NOT NULL DEFAULT 0,
  PRIMARY KEY (aggregate_name, aggregate_id)
);

CREATE TABLE IF NOT EXISTS noddde_saga_states (
  saga_name TEXT NOT NULL,
  saga_id   TEXT NOT NULL,
  state     TEXT NOT NULL,
  PRIMARY KEY (saga_name, saga_id)
);

CREATE TABLE IF NOT EXISTS noddde_snapshots (
  aggregate_name TEXT NOT NULL,
  aggregate_id   TEXT NOT NULL,
  state          TEXT NOT NULL,
  version        INTEGER NOT NULL,
  PRIMARY KEY (aggregate_name, aggregate_id)
);

CREATE TABLE IF NOT EXISTS noddde_outbox (
  id             TEXT PRIMARY KEY,
  event          TEXT NOT NULL,
  aggregate_name TEXT,
  aggregate_id   TEXT,
  created_at     TEXT NOT NULL,
  published_at   TEXT
);

IF NOT EXISTS (SELECT * FROM sysobjects WHERE name='noddde_events' AND xtype='U')
CREATE TABLE noddde_events (
  id              INT IDENTITY(1,1) PRIMARY KEY,
  aggregate_name  NVARCHAR(255) NOT NULL,
  aggregate_id    NVARCHAR(255) NOT NULL,
  sequence_number INT NOT NULL,
  event_name      NVARCHAR(255) NOT NULL,
  payload         NVARCHAR(MAX) NOT NULL,
  metadata        NVARCHAR(MAX),
  created_at      DATETIME2 NOT NULL DEFAULT GETUTCDATE()
);
CREATE UNIQUE INDEX noddde_events_stream_version_idx
  ON noddde_events (aggregate_name, aggregate_id, sequence_number);

IF NOT EXISTS (SELECT * FROM sysobjects WHERE name='noddde_aggregate_states' AND xtype='U')
CREATE TABLE noddde_aggregate_states (
  aggregate_name NVARCHAR(255) NOT NULL,
  aggregate_id   NVARCHAR(255) NOT NULL,
  state          NVARCHAR(MAX) NOT NULL,
  version        INT NOT NULL DEFAULT 0,
  PRIMARY KEY (aggregate_name, aggregate_id)
);

IF NOT EXISTS (SELECT * FROM sysobjects WHERE name='noddde_saga_states' AND xtype='U')
CREATE TABLE noddde_saga_states (
  saga_name NVARCHAR(255) NOT NULL,
  saga_id   NVARCHAR(255) NOT NULL,
  state     NVARCHAR(MAX) NOT NULL,
  PRIMARY KEY (saga_name, saga_id)
);

IF NOT EXISTS (SELECT * FROM sysobjects WHERE name='noddde_snapshots' AND xtype='U')
CREATE TABLE noddde_snapshots (
  aggregate_name NVARCHAR(255) NOT NULL,
  aggregate_id   NVARCHAR(255) NOT NULL,
  state          NVARCHAR(MAX) NOT NULL,
  version        INT NOT NULL,
  PRIMARY KEY (aggregate_name, aggregate_id)
);

IF NOT EXISTS (SELECT * FROM sysobjects WHERE name='noddde_outbox' AND xtype='U')
CREATE TABLE noddde_outbox (
  id             NVARCHAR(255) PRIMARY KEY,
  event          NVARCHAR(MAX) NOT NULL,
  aggregate_name NVARCHAR(255),
  aggregate_id   NVARCHAR(255),
  created_at     DATETIME2 NOT NULL DEFAULT GETUTCDATE(),
  published_at   DATETIME2
);

-- Example: dedicated table for an Order aggregate
CREATE TABLE IF NOT EXISTS orders (
  aggregate_id TEXT PRIMARY KEY,
  state        JSONB NOT NULL,       -- or TEXT for MySQL/SQLite
  version      INTEGER NOT NULL DEFAULT 0
);

// Drizzle: pass the Drizzle table object
adapter.stateStored(ordersTable);

// Prisma: pass the camelCase model delegate name
adapter.stateStored("order", {
  aggregateId: "orderId",
  state: "data",
  version: "rev",
});

// TypeORM: pass the entity class
adapter.stateStored(OrderEntity, {
  aggregateId: "orderId",
  state: "data",
  version: "rev",
});