@ddd-ts-kit/typeorm

Les adaptateurs TypeORM des ports du noyau et des événements : une unité de travail que les repositories ne voient jamais, le store de l'outbox transactionnelle, le store des événements traités, leurs tables sous forme d'objets EntitySchema, et un schéma de base de données par contexte délimité (bounded context). Les mêmes pièces que @ddd-ts-kit/drizzle, pour les projets sur TypeORM.

npm i @ddd-ts-kit/typeorm typeorm pg

Tout ici utilise EntitySchema, pas des classes décorées : aucun import de reflect-metadata, pas d'experimentalDecorators, et les lignes restent des objets simples qu'un mapper transforme en agrégats. PostgreSQL d'abord ; voir Couverture des dialectes pour MySQL et SQLite.

Sa place dans une application

Rien d'ici n'est importé depuis domain/ ni application/. Un gestionnaire (handler) dépend du port de repository de son domaine ; l'adaptateur de ce port, dans infrastructure/, exécute ses instructions sur le TypeOrmContext ; la plateforme construit l'unité de travail une fois et la lie.

L'idée à retenir : toute instruction passe par manager(). Dans une unité de travail, c'est l'EntityManager transactionnel de cette unité de travail : la ligne de l'agrégat, l'enregistrement d'outbox de l'événement qu'il a publié et la marque d'événement traité d'un consommateur sont validés ou annulés ensemble. Hors d'une unité de travail (une requête, un script), c'est dataSource.manager.

Schéma

Les tables sont des objets EntitySchema : ajoutez-les aux entities de la source de données et typeorm migration:generate écrit leur DDL dans une migration de votre projet.

import { outboxEntity, processedEventEntity, schemaNameOf } from '@ddd-ts-kit/typeorm';

export const outbox = outboxEntity({ schema: 'messaging' }); // messaging.outbox
export const processedEvents = processedEventEntity({ schema: 'messaging' });

// contexts/billing-core/infrastructure/persistence/invoice-row.schema.ts
import { EntitySchema } from 'typeorm';

export interface InvoiceRow {
  id: string;
  amount: number;
  status: 'draft' | 'issued';
  version: number;
}

export const invoices = new EntitySchema<InvoiceRow>({
  name: 'Invoice',
  tableName: 'invoices',
  schema: schemaNameOf('billing-core'), // billing_core
  columns: {
    id: { type: 'text', primary: true },
    amount: { type: 'int' },
    status: { type: 'text' },
    version: { type: 'int' },
  },
});

Les options des deux fabriques : schema (aucune : le schéma par défaut de la connexion), dialect (postgres par défaut, mysql, sqlite : il choisit les types des colonnes JSON et horodatage), tableName, et name, le nom de l'entité. Le nom de l'entité vaut par défaut OutboxRecord ou ProcessedEvent, suffixé du schéma (OutboxRecord_messaging), si bien que chaque contexte peut enregistrer sa propre outbox dans la même source de données.

outbox est indexée pour le relais (les enregistrements en attente par prochaine tentative, index partiel sur PostgreSQL et SQLite) ; processed_events a (consumer, event_id) pour clé primaire. TypeORM ne crée pas les schémas : créez-les dans la migration (CREATE SCHEMA IF NOT EXISTS).

Racine de composition

createTypeOrmUnitOfWork(dataSource) donne un seul objet qui est deux ports : le UnitOfWork du noyau, et TypeOrmContext, la poignée sur laquelle les repositories et les stores exécutent leurs requêtes. Liez-le aux deux :

import { DataSource } from 'typeorm';
import { defineContextModule } from '@ddd-ts-kit/di';
import { IntegrationEventPublisher, ProcessedEventStore, UnitOfWork } from '@ddd-ts-kit/kernel';
import { OutboxPublisher, OutboxStore } from '@ddd-ts-kit/events';
import {
  createTypeOrmUnitOfWork,
  TypeOrmContext,
  TypeOrmOutboxStore,
  TypeOrmProcessedEventStore,
} from '@ddd-ts-kit/typeorm';

export const dataSource = new DataSource({
  type: 'postgres',
  url: process.env.DATABASE_URL,
  entities: [outbox, processedEvents, invoices],
});
const unitOfWork = createTypeOrmUnitOfWork(dataSource);

export const persistenceModule = defineContextModule({
  name: 'persistence',
  register(builder) {
    builder.rebind(UnitOfWork).toValue(unitOfWork);
    builder.bind(TypeOrmContext).toValue(unitOfWork);
    builder
      .bind(OutboxStore)
      .toFactory((resolve) => new TypeOrmOutboxStore(resolve(TypeOrmContext), { entity: outbox }));
    builder
      .rebind(ProcessedEventStore)
      .toFactory((resolve) => new TypeOrmProcessedEventStore(resolve(TypeOrmContext), { entity: processedEvents }));
    builder.rebind(IntegrationEventPublisher).toClass(OutboxPublisher, [OutboxStore]);
  },
  onStart: async () => {
    await dataSource.initialize();
  },
  onStop: () => dataSource.destroy(),
});

const app = createApplication({ modules: [kernelModule(), persistenceModule, contextModule('billing', billingDeclarations)] });

const relay = new OutboxRelay({ store: app.resolve(OutboxStore), publisher: broker, clock: app.resolve(Clock) });
relay.start({ intervalMs: 500 });

Placez le module après kernelModule() : rebind remplace ses valeurs par défaut en mémoire. Le bus de commandes du module du noyau exécute chaque commande dans le UnitOfWork lié ; les gestionnaires n'en savent rien. contextModule de @ddd-ts-kit/di transforme les classes déclarées d'un contexte en son module.

Un projet écrit par la CLI contient le même câblage en deux fichiers : platform/database/persistence-over.ts construit l'unité de travail et les deux stores une fois, et platform/platform.module.ts les lie, le publisher de l'outbox étant passé à kernelModule({ eventPublisher }).

Passez à un store l'entité même que la source de données enregistre. Sans entity, un store utilise outboxEntity() ou processedEventEntity(), que TypeORM fait correspondre par son nom à l'entité par défaut dans le schéma par défaut ; la classe se lie alors directement : builder.bind(OutboxStore).toClass(TypeOrmOutboxStore, [TypeOrmContext]), builder.rebind(ProcessedEventStore).toClass(TypeOrmProcessedEventStore, [TypeOrmContext]).

unitOfWork.run(work) ouvre dataSource.transaction et garde l'EntityManager transactionnel dans un AsyncLocalStorage ; un run imbriqué rejoint la transaction englobante (sans point de sauvegarde). Comme append utilise le manager courant, un événement publié par un gestionnaire qui échoue ensuite est annulé avec l'écriture métier : c'est toute la raison d'être de l'outbox. createTypeOrmUnitOfWork(dataSource, 'REPEATABLE READ') fixe le niveau d'isolation.

run repose sur withinUnitOfWork du noyau. Les événements du domaine des agrégats enregistrés à l'intérieur attendent que work ait rendu la main, puis sont distribués dans la transaction, avant la validation (commit) : leurs gestionnaires voient chaque enregistrement, et ce qu'ils écrivent est validé ou annulé avec. Si work, un gestionnaire ou la validation échoue, il ne reste rien : aucune ligne, aucun enregistrement d'outbox, aucun gestionnaire exécuté après la validation. Les gestionnaires déclarés after-commit s'exécutent une fois la transaction validée. Un run imbriqué rejoint l'unité de travail en cours : une seule distribution, à la fin de la plus externe. Un run appelé directement (un consommateur, un script) suit le même déroulé qu'une commande. unitOfWorkContract de @ddd-ts-kit/testing/contracts le prouve, avec une contrainte différée pour la validation qui échoue ; sur un vrai PostgreSQL (DDD_TS_KIT_TEST_DATABASE_URL), il prouve aussi que les lignes d'un gestionnaire before-commit sont invisibles depuis une autre connexion et visibles des gestionnaires after-commit. Le README de @ddd-ts-kit/kernel dessine ce déroulé (Événements du domaine et unité de travail).

TypeOrmContext est une classe abstraite : elle est donc son propre jeton d'injection. Un projet à deux sources de données donne à chacune sa propre classe (export abstract class Ledger extends TypeOrmContext {} dans platform/database/) et lie chacune à son unité de travail.

Repository

Le port vit dans le domaine, l'adaptateur dans l'infrastructure. Un repository reçoit le contexte, jamais une transaction : manager() est le manager transactionnel de l'unité de travail s'il y en a une, et dataSource.manager sinon.

import { Repository } from '@ddd-ts-kit/kernel';

export abstract class InvoiceRepository extends Repository<Invoice> {}

import { Repository } from '@ddd-ts-kit/kernel/declarations';
import { insertIgnoringConflict, optimisticUpdate, TypeOrmContext } from '@ddd-ts-kit/typeorm';

@Repository(InvoiceRepository)
export class TypeOrmInvoiceRepository implements InvoiceRepository {
  constructor(private readonly context: TypeOrmContext) {}

  async findById(id: string): Promise<Invoice | undefined> {
    const row = await this.context.manager().findOneBy(invoices, { id });

    if (row === null) {
      return undefined;
    }

    const invoice = Invoice.rehydrate(row);
    invoice.version = row.version;

    return invoice;
  }

  async save(invoice: Invoice): Promise<void> {
    const manager = this.context.manager();

    await optimisticUpdate(invoice, async ({ expected, next }) => {
      const row = { ...invoice.state, version: next };

      if (expected === 0) {
        return (await insertIgnoringConflict(manager, invoices, row)) ? 1 : 0;
      }

      const updated = await manager.update(invoices, { id: invoice.id, version: expected }, row);

      return updated.affected ?? 0;
    });
  }
}

Dans le style explicite, sans le décorateur, la même liaison tient en une ligne de <context>.module.ts : builder.bind(InvoiceRepository).toClass(TypeOrmInvoiceRepository, [TypeOrmContext]).

optimisticUpdate donne la version attendue et la version à écrire, lève une ConcurrencyError (une DomainError de code CONCURRENCY_CONFLICT) quand l'instruction gardée n'a écrit aucune ligne, et fait avancer invoice.version sinon. insertIgnoringConflict insère sauf si la clé existe et dit s'il l'a fait ; contrairement à une violation d'unicité interceptée, il laisse utilisable la transaction PostgreSQL qui l'entoure. N'utilisez pas manager.save pour un agrégat : il lit avant d'écrire et choisit seul entre insertion et mise à jour, ce qui défait la garde de version. Le @VersionColumn de TypeORM n'est pas utilisé pour la même raison : la version appartient à l'agrégat.

// Au lieu de : une lecture, puis une insertion ou une mise à jour choisie par TypeORM, sans garde de version
await manager.save(invoices, { ...invoice.state, version: invoice.version + 1 });

// Écrivez : l'instruction gardée, qui n'écrit aucune ligne quand une autre transaction a enregistré avant
await optimisticUpdate(invoice, async ({ expected, next }) => {
  // un premier enregistrement (expected 0) insère plutôt avec insertIgnoringConflict, comme dans le repository ci-dessus
  const updated = await manager.update(invoices, { id: invoice.id, version: expected }, { ...invoice.state, version: next });

  return updated.affected ?? 0;
});

Ce que fait optimisticUpdate, seul, du nombre de lignes que renvoie l'écriture :

import { ConcurrencyError, optimisticUpdate } from '@ddd-ts-kit/typeorm';

// Ce qu'un repository passe à optimisticUpdate : l'agrégat, et une écriture qui répond combien de lignes elle a écrites.
const invoice = { id: 'inv-1', version: 3 }; // lu en version 3

await optimisticUpdate(invoice, async ({ expected, next }) => {
  // expected = 3 : la garde de l'UPDATE (WHERE id = 'inv-1' AND version = 3)
  // next = 4 : la version avec laquelle la ligne est écrite
  return expected === 3 && next === 4 ? 1 : 0;
});
// invoice.version vaut maintenant 4 : le même agrégat peut être enregistré de nouveau

try {
  // Une autre transaction a enregistré inv-1 entre-temps : l'UPDATE gardé ne trouve aucune ligne.
  await optimisticUpdate(invoice, async () => 0);
} catch (error) {
  if (error instanceof ConcurrencyError) {
    console.log(error.code); // CONCURRENCY_CONFLICT ; invoice.version vaut toujours 4
  }
}

La commande qui enregistrait échoue avec cette erreur et son unité de travail est annulée : rien de ce qu'elle a écrit ne reste, et l'appelant relit l'agrégat avant de réessayer.

Colonnes d'audit, restriction au tenant et pages

Les formes de persistance standard du noyau, sur TypeORM. auditColumns(dialect?) et tenantColumn() sont des options de colonnes à étaler dans les columns d'un EntitySchema : leurs propriétés sont celles des AuditStamp et TenantScoped du noyau, si bien qu'un tampon s'étale tel quel dans une insertion, et les colonnes portent les mêmes noms dans tous les projets (created_at, created_by, updated_at, updated_by, tenant_id) : c'est ce qui permet d'écrire une seule fois un tableau de bord, un export ou le package d'audit.

import type { AuditStamp, TenantScoped } from '@ddd-ts-kit/kernel';
import { auditColumns, schemaNameOf, tenantColumn } from '@ddd-ts-kit/typeorm';

interface InvoiceRow extends VersionedInvoice, TenantScoped, AuditStamp {}

export const invoices = new EntitySchema<InvoiceRow>({
  name: 'Invoice',
  tableName: 'invoices',
  schema: schemaNameOf('billing-core'),
  columns: {
    id: { type: 'text', primary: true },
    amount: { type: 'int' },
    status: { type: 'text' },
    version: { type: 'int' },
    ...tenantColumn(),
    ...auditColumns(),
  },
  indices: [{ name: 'invoices_tenant_amount_idx', columns: ['tenantId', 'amount', 'id'] }],
});

import { Clock, ExecutionScope, stampCreation, stampUpdate, tenantScopeOf } from '@ddd-ts-kit/kernel';
import { Repository } from '@ddd-ts-kit/kernel/declarations';

@Repository(InvoiceRepository)
export class TypeOrmInvoiceRepository implements InvoiceRepository {
  constructor(
    private readonly context: TypeOrmContext,
    private readonly scope: ExecutionScope,
    private readonly clock: Clock,
  ) {}

  async findById(id: string): Promise<Invoice | undefined> {
    const row = await this.context.manager().findOneBy(invoices, { id, ...tenantScopeOf(this.scope) });

    return row === null ? undefined : InvoiceMapper.toDomain(row);
  }

  async save(invoice: Invoice): Promise<void> {
    await optimisticUpdate(invoice, async ({ expected, next }) => {
      const state = { ...InvoiceMapper.toRow(invoice), version: next };

      return expected === 0 ? this.insert(state) : this.update(state, expected);
    });
  }

  private async insert(state: VersionedInvoice): Promise<number> {
    const row = { ...state, ...tenantScopeOf(this.scope), ...stampCreation(this.scope, this.clock) };

    return (await insertIgnoringConflict(this.context.manager(), invoices, row)) ? 1 : 0;
  }

  private async update(state: VersionedInvoice, expected: number): Promise<number> {
    const guarded = { id: state.id, version: expected, ...tenantScopeOf(this.scope) };
    const row = { ...state, ...stampUpdate(this.scope, this.clock) };

    return (await this.context.manager().update(invoices, guarded, row)).affected ?? 0;
  }
}

Les @CreateDateColumn et @UpdateDateColumn de TypeORM ne sont pas utilisés : ils prennent l'heure du serveur de base de données, pas celle de la Clock, et ne savent rien de qui agit.

Une requête de recherche prend la PageRequest du noyau et renvoie une Page, dans ses deux variantes :

export abstract class InvoiceQueries {
  abstract search(request: PageRequest<'amount'>): Promise<Page<InvoiceSummary>>;
}

import { LessThan, MoreThan, type FindOptionsWhere } from 'typeorm';
import { ExecutionScope, encodeCursor, pageOf, tenantScopeOf } from '@ddd-ts-kit/kernel';
import { Repository } from '@ddd-ts-kit/kernel/declarations';

@Repository(InvoiceQueries)
export class TypeOrmInvoiceQueries implements InvoiceQueries {
  constructor(
    private readonly context: TypeOrmContext,
    private readonly scope: ExecutionScope,
  ) {}

  async search(request: PageRequest<'amount'>): Promise<Page<InvoiceSummary>> {
    return request.kind === 'offset' ? this.byOffset(request) : this.byCursor(request);
  }

  private async byOffset(request: OffsetPageRequest<'amount'>): Promise<Page<InvoiceSummary>> {
    const [rows, total] = await this.context.manager().findAndCount(invoices, {
      where: tenantScopeOf(this.scope),
      order: totalOrder(directionOf(request)),
      take: request.limit,
      skip: request.offset,
    });

    return pageOf(rows.map(toSummary), { total });
  }

  private async byCursor(request: CursorPageRequest<'amount'>): Promise<Page<InvoiceSummary>> {
    const direction = directionOf(request);
    const tenant = tenantScopeOf(this.scope);
    const rows = await this.context.manager().find(invoices, {
      where: request.cursor === undefined ? tenant : beyond(positionOf(request.cursor), direction, tenant),
      order: totalOrder(direction),
      take: request.limit + 1,
    });
    const items = rows.slice(0, request.limit);
    const last = items.at(-1);
    const more = rows.length > request.limit && last !== undefined;

    return pageOf(items.map(toSummary), more ? { nextCursor: encodeCursor({ amount: last.amount, id: last.id }) } : {});
  }
}

const directionOf = (request: PageRequest<'amount'>) => request.sort[0]?.direction ?? 'asc';

// La clé primaire suit le champ de tri, dans le même sens : l'ordre est total.
const totalOrder = (direction: SortDirection) => ({ amount: direction, id: direction });

// Les lignes après une position, en comparaison de tuples. Une liste de conditions est un OR en TypeORM,
// donc chaque alternative répète le tenant.
function beyond(
  { amount, id }: { amount: number; id: string },
  direction: SortDirection,
  tenant: TenantScoped,
): FindOptionsWhere<InvoiceRow>[] {
  const past = direction === 'asc' ? MoreThan : LessThan;

  return [
    { ...tenant, amount: past(amount) },
    { ...tenant, amount, id: past(id) },
  ];
}

positionOf(cursor) décode avec decodeCursor et vérifie que amount est un nombre et id une chaîne, en levant InvalidPageRequestError sinon.

Un adaptateur de lecture sur la base de données se déclare avec @Repository, comme ceux que génère la CLI : sa nature est repository. Déclaré avec @Adapter, il serait external, et une plateforme générée entoure chaque adaptateur external et gateway d'une nouvelle tentative (retry) et d'une limite de temps, ce qu'une instruction de l'unité de travail ne doit pas recevoir.

Les règles que tiennent ces exemples, et que les tests de ce package vérifient sur une vraie base de données :

Les colonnes d'audit sont remplies par l'adaptateur, jamais par l'agrégat : stampCreation(scope, clock) à l'insertion, stampUpdate(scope, clock) à la mise à jour, qui laisse created_at et created_by intacts. L'acteur est celui de la portée d'exécution : l'utilisateur de la requête HTTP, ou system:<consumer> pour un événement consommé.
Le tenant fait partie de chaque instruction : écrit à l'insertion, et dans la condition de chaque lecture et de chaque mise à jour, garde de version comprise. Une ligne d'un autre tenant n'est pas trouvée, et une mise à jour qui la vise n'écrit rien, ce que optimisticUpdate signale par une ConcurrencyError. Un travail qui ne nomme aucun tenant lève NoCurrentTenantError avant tout SQL.
Une liste a un ordre total. La clé primaire suit le champ de tri, dans le même sens, pour départager : sans elle, des lignes égales sur le champ de tri n'ont pas d'ordre défini, et une page pourrait en répéter ou en sauter une. L'index couvre (tenant_id, <sort field>, id).
Un curseur est une position, pas un numéro de page : la valeur de tri et l'id de la dernière ligne, écrits par encodeCursor, comparés comme un tuple (amount > :amount OR (amount = :amount AND id > :id)). Une ligne de plus que limit est lue pour savoir si une page suit, donc aucun comptage n'est nécessaire, et une ligne insérée avant le curseur ne décale pas la page suivante comme elle le fait avec un offset. Un curseur revient d'un client : decodeCursor refuse ce qu'il n'a pas écrit, et l'adaptateur vérifie les types de la position avant qu'ils n'atteignent une instruction.
Une page par offset répond total, une page par curseur nextCursor. Proposez la forme par offset pour de petites listes que l'on parcourt, et la forme par curseur pour tout ce qui grandit.

Critères

criteriaToTypeOrm(builder, plan, { alias }) ajoute un plan de critères du noyau à un SelectQueryBuilder sur l'entité d'un modèle de lecture : les filtres et le périmètre de l'acteur, le keyset d'une page par curseur, l'ordre total terminé par le champ qui départage, la fenêtre. predicateToTypeOrm(builder, predicate, { alias }) ajoute les filtres seuls, pour un comptage. Les champs du modèle de lecture sont les propriétés de l'entité (properties renomme celles qui diffèrent). Chaque valeur est un paramètre nommé ; le seul texte composé est alias.property avec un opérateur d'une liste fermée, tous deux venant de l'adaptateur. Une recherche textuelle est LIKE :p ESCAPE :e, avec %, _ et \ échappés.

async search(criteria: ScopedCriteria<AppointmentRow>): Promise<Page<AppointmentRow>> {
  const plan = planCriteria(appointmentCriteria, criteria);

  if (plan.empty) {
    return pageOfPlan(plan, []);
  }

  const builder = this.context.manager().createQueryBuilder(appointmentList, 'row');

  return pageOfPlan(plan, await criteriaToTypeOrm(builder, plan, { alias: 'row' }).getMany());
}

Une page par curseur est lue avec une ligne de plus que demandé : c'est ainsi que pageOfPlan sait si une autre suit ; une page par offset prend son total de predicateToTypeOrm(...).getCount(). Un champ déclaré nullable est trié NULLS LAST en ordre croissant et NULLS FIRST en ordre décroissant. Deux critères dans une même requête prennent des noms de paramètres distincts avec l'option parameters.

Chaînes. Une colonne de texte est triée et comparée par intervalle selon la collation de la colonne ; les critères trient les chaînes par point de code. Déclarez avec collation: 'C' les colonnes de texte triées ou comparées avec gt, lt ou between : l'index sert alors le tri, et l'ordre est le même que dans le fake. Les décimaux sont une colonne numeric et une chaîne dans la ligne ; ils reviennent à l'échelle de la colonne. Les instants sont timestamptz et une Date. Un entier est integer : un bigint revient du pilote sous forme de chaîne.

Modèles de lecture

export const appointmentList = new EntitySchema<AppointmentRow & ReadModelStamps>({
  name: 'AppointmentRow',
  tableName: 'appointment_list',
  schema: 'scheduling',
  columns: {
    id: { type: 'text', primary: true },
    status: { type: 'text' },
    startsAt: { name: 'starts_at', type: 'timestamp with time zone', precision: 3 },
    ...readModelColumns(), // last_event_id, last_event_at, last_event_version, deleted
  },
});

// infrastructure/persistence/type-orm-appointment-list.repository.ts
@Repository(AppointmentList)
export class TypeOrmAppointmentList extends TypeOrmReadModelStore<AppointmentRow> implements AppointmentList {
  constructor(context: TypeOrmContext) {
    super(context, { entity: appointmentList, definition: appointmentCriteria });
  }
}

Chaque écriture passe par TableReadModelStore du noyau : une ligne n'est écrite que si son événement est plus récent que celui qu'elle porte, sur l'exécuteur de l'unité de travail, si bien que la ligne et la marque d'événement traité sont validées ensemble. Une ligne supprimée reste dans la table avec deleted posé : une requête à vous sur la table écarte ces lignes, et un index unique aussi (WHERE NOT deleted). store.table est ce que prend ProjectionRebuild ; la reconstruction est réservée à PostgreSQL (une copie <table>__rebuild faite LIKE ... INCLUDING ALL, les clés écrites entre-temps dans <table>__touched, l'échange par renommage en une transaction) et ses garanties et limites sont dans le README de @ddd-ts-kit/kernel (Reconstruction). Le store passe criteriaContract et readModelStoreContract de @ddd-ts-kit/testing/contracts, sur PGlite et sur PostgreSQL 18. Les critères et les modèles de lecture n'ont jamais été exécutés sur MySQL ni SQLite.

La table s'écrit par son entité ; la copie d'une reconstruction, qui n'en a pas, s'écrit colonne par colonne avec le même query builder.

Outbox et idempotence

TypeOrmOutboxStore.claimBatch sélectionne les lignes dues FOR UPDATE SKIP LOCKED (setLock('pessimistic_write').setOnLocked('skip_locked')) et les prend à bail dans la même courte transaction en repoussant next_attempt_at (leaseMs, 30 s par défaut) : plusieurs relais se partagent ainsi le travail sans s'attendre, et les enregistrements d'un relais mort reviennent quand le bail expire. La livraison est au moins une fois et, par défaut, sans ordre : ordering: 'per-key' de OutboxRelay, muni d'un keyOf (en général l'identifiant de l'agrégat), garde dans l'ordre les enregistrements d'une même clé, avec un seul relais (L'ordre par clé).
Là où la base n'a pas de SKIP LOCKED (SQLite), ou quand vous passez claim: 'compare-and-set', la prise lit les lignes dues et prend chacune par une mise à jour gardée par le nombre de tentatives lu : même garantie, aucune clause de verrou, une instruction par enregistrement. La valeur par défaut est skip-locked pour les pilotes postgres, aurora-postgres, cockroachdb, mysql et aurora-mysql, compare-and-set pour tous les autres.
TypeOrmProcessedEventStore.markProcessed est une insertion qui ignore le conflit sur (consumer, event_id). Exécutez le consommateur dans l'unité de travail : la marque est alors validée avec les effets du gestionnaire, et disparaît avec eux s'il échoue.

Les options du store de l'outbox, et le relais qui le lit :

import { OutboxRelay } from '@ddd-ts-kit/events';
import type { Clock, IntegrationEventPublisher } from '@ddd-ts-kit/kernel';
import { outboxEntity, TypeOrmOutboxStore, type TypeOrmUnitOfWork } from '@ddd-ts-kit/typeorm';

// Depuis la racine de composition : createTypeOrmUnitOfWork(dataSource), l'adaptateur du broker, l'horloge.
declare const unitOfWork: TypeOrmUnitOfWork;
declare const broker: IntegrationEventPublisher;
declare const clock: Clock;

export const outbox = outboxEntity({ schema: 'messaging' }); // l'entité que la source de données enregistre

const store = new TypeOrmOutboxStore(unitOfWork, {
  entity: outbox,
  leaseMs: 60_000, // un enregistrement pris par un relais mort redevient dû au bout d'une minute
  claim: 'compare-and-set', // au lieu de la valeur par défaut du pilote, par exemple sur MySQL 5.7
});

const relay = new OutboxRelay({ store, publisher: broker, clock });

relay.start({ intervalMs: 500 });

Un consommateur sur le store des événements traités : consume de @ddd-ts-kit/events exécute la marque et le gestionnaire dans une seule unité de travail (Consommer de façon idempotente).

import { z } from 'zod';
import { consume, defineIntegrationEvent } from '@ddd-ts-kit/events';
import { zodSchema } from '@ddd-ts-kit/validation/zod';
import { processedEventEntity, TypeOrmProcessedEventStore, type TypeOrmUnitOfWork } from '@ddd-ts-kit/typeorm';

// Depuis la racine de composition et le contexte : l'unité de travail, et ce que fait le consommateur.
declare const unitOfWork: TypeOrmUnitOfWork;
declare function issueInvoice(orderId: string): Promise<void>;

// La déclaration du producteur, dans son langage publié.
const orderPlaced = defineIntegrationEvent({
  type: 'ordering.order.placed',
  version: 1,
  description: 'An order was placed by a customer.',
  payload: zodSchema(z.object({ orderId: z.string() })),
});

const processed = new TypeOrmProcessedEventStore(unitOfWork, {
  entity: processedEventEntity({ schema: 'messaging' }),
});

// La marque dans messaging.processed_events et les écritures d'issueInvoice sont validées en une transaction :
// une nouvelle livraison est ignorée, et un événement dont le gestionnaire a levé une erreur est retraité quand il revient.
export const invoiceOnOrderPlaced = consume(
  orderPlaced,
  { consumer: 'billing.invoice-on-order-placed', processed, unitOfWork },
  async (event) => {
    await issueInvoice(event.payload.orderId);
  },
);

Event store

const dataSource = new DataSource({ type: 'postgres', entities: [...eventStoreEntities({ schema: 'scheduling' })] });

const { events, snapshots, checkpoints, subjectKeys } = createTypeOrmEventSourcing(unitOfWork, { schema: 'scheduling' });

eventStoreEntities déclare les tables sous forme d'EntitySchema, si bien que typeorm migration:generate les crée avec les colonnes d'eventStoreTablesSql du noyau. Les stores ne les chargent jamais comme entités. Les stores de createTypeOrmEventSourcing refusent une source de données qui n'est pas PostgreSQL : leur première instruction lève une TypeError qui nomme son type.

Chaque instruction s'exécute sur l'exécuteur du contexte : un ajout rejoint la transaction de l'unité de travail, si bien que le flux et les enregistrements d'outbox que publient ses événements sont validés ensemble (eventStoreUnitOfWorkContract). Les ajouts prennent la tête du store, une ligne verrouillée jusqu'à la fin de leur transaction : les positions sont données dans l'ordre des validations, sans trou, et deux ajouts ne s'exécutent jamais côte à côte, quels que soient leurs flux. PostgreSQL seulement. Les stores sont ceux du noyau (PostgresEventStore et ses compagnons), tenus ici à eventStoreContract ; comment un agrégat en event sourcing les utilise : le README de @ddd-ts-kit/kernel, section Event sourcing.

Couverture des dialectes

Ce qui suit concerne ce package. La CLI ne génère que pour PostgreSQL : init --orm typeorm refuse --database mysql et --database sqlite.

Base de données	État
PostgreSQL	Couvert par la suite automatique : suites de contrat de `@ddd-ts-kit/testing`, contrat de repository, annulation, unité de travail imbriquée, prises concurrentes. Elle tourne par défaut sur PGlite servi par le protocole réseau de PostgreSQL, et sur un vrai serveur quand `DDD_TS_KIT_TEST_DATABASE_URL` est définie (dernière exécution : PostgreSQL 18).
MySQL 8	Hors de la suite automatique. Les suites de contrat de l'outbox et des événements traités et l'annulation d'`append` ont été exécutées une fois à la main contre MySQL 8.4 avec `mysql2`, entités construites avec `dialect: 'mysql'`, avec `skip-locked` et `compare-and-set`.
SQLite	Hors de la suite automatique. Les mêmes suites ont été exécutées une fois à la main avec le pilote `better-sqlite3`, entités construites avec `dialect: 'sqlite'`. Pas de `SKIP LOCKED` : la prise est `compare-and-set`. Pas de schémas : omettez `schema` et nommez les tables par contexte.
MariaDB, CockroachDB, SQL Server, Oracle, MongoDB	Jamais exécuté.

Limites

Testé avec TypeORM 1.1. La plage de la peer dependency est ^1.0.0 : TypeORM 0.3 n'est pas accepté.
Un EntitySchema est statique : les types de colonnes dépendent du dialect que vous passez, pas de la source de données dans laquelle il est enregistré.
Une charge utile JSON null est stockée en NULL SQL et relue comme null.
Sur MySQL 5.7, il n'y a pas de SKIP LOCKED : passez claim: 'compare-and-set'.
synchronize: true est réservé aux tests. En production, les tables viennent de vos migrations.
Par défaut, l'ordre n'est pas garanti, pas même par agrégat : un événement en échec est retenté plus tard pendant que ses successeurs partent. ordering: 'per-key' de OutboxRelay, muni d'un keyOf, ordonne les enregistrements d'une même clé, avec un seul relais en marche. Sans keyOf, rien n'est ordonné.

Tester un adaptateur

L'unité de travail de ce package passe unitOfWorkContract, et ses stores outboxStoreContract et processedEventStoreContract, de @ddd-ts-kit/testing/contracts. Tenez vos repositories à repositoryContract de la même façon, à côté de leur fake en mémoire. Les tests de ce package n'ont besoin d'aucun conteneur :

npx vitest run packages/typeorm
DDD_TS_KIT_TEST_DATABASE_URL=postgresql://user:password@localhost:5432/postgres npx vitest run packages/typeorm

Avec la variable définie, chaque fichier de spec crée sa propre base de données sur ce serveur et la supprime ensuite, et les tests qui ont besoin de deux vraies connexions (un relais qui saute les lignes verrouillées par un autre) s'exécutent au lieu d'être ignorés.

Voir aussi

Persistance : choisir un ORM, et ce que chaque package d'adaptateurs a vérifié
Agrégats : la concurrence optimiste, vue depuis le domaine
L'outbox transactionnelle : pourquoi l'enregistrement d'outbox est validé avec l'écriture métier
Livraison et idempotence : au moins une fois, et ce qu'un consommateur en fait
@ddd-ts-kit/drizzle : les mêmes pièces sur Drizzle, la persistance par défaut du kit