@generazioneai/events

Eventi durevoli fra i servizi della piattaforma (ADR-009, spec «Fondamenta enterprise» §1): l'evento si scrive nella stessa transazione del dato (outbox), un relay lo pubblica su NATS JetStream, chi lo riceve lo applica una volta sola (inbox) con riprove e dead letter. Consegna «almeno una volta», effetto «una volta sola». I comandi restano su NATS core, invariati.

npm i @generazioneai/events

Peer: @nestjs/common, @nestjs/core, reflect-metadata; drizzle-orm (opzionale) solo per @generazioneai/events/drizzle. Client NATS: @nats-io/transport-node + @nats-io/jetstream, su una connessione propria (diversa da quella del trasporto Nest).

Nel servizio

1. Copia nello schema le tabelle: prisma/events.prisma (Prisma) o sql/001-events-outbox-inbox.sql + -rientro.sql (HR: registro SQL). Non cambiare i nomi di tabelle e colonne: la libreria li usa in SQL.
2. Registra il modulo:

EventsModule.forRoot({
  service: 'id',               // id | hr | train | ops | notify | ai | sign | bill | audit | checkflow
  prisma: PrismaService,       // la classe (risolta da Nest) o l'istanza (con Drizzle: `store`, sotto)
  // facoltativi:
  signer: (p) => signEventToken(ctx, { eventId: p.eventId, subject: p.subject }),            // authz 0.7
  verifier: (token, p) => verifyEventToken(token, { eventId: p.eventId, subject: p.subject, mode: p.mode }),
  traceparent: () => currentTraceparent(),                                                   // @generazioneai/telemetry
})

3. Pubblica dentro la transazione del dato:

await this.prisma.$transaction(async (tx) => {
  const emp = await tx.employment.update(…);
  await this.outbox.enqueue(tx, {                          // OutboxService
    subject: 'evt.id.employment.changed',                  // evt.<servizio>.<entità>.<fatto>, il servizio è il proprio
    aggregate: { type: 'Employment', id: emp.id, version: emp.version },
    juridicalId: emp.juridicalId,
    data: { juridicalId, individualId, juridicalIndividualId, status: 'ENDED' }, // solo id e stati
  });
});

Fuori da una transazione lancia EVENTS_OUTSIDE_TX. Il riconoscimento è fatto dal database, non dalla forma del client (in Prisma 7 anche il tx ha $transaction): la prima enqueue su un tx esegue SAVEPOINT (esiste solo dentro un blocco di transazione, altrimenti Postgres risponde 25P01) e RELEASE; il tx verificato non si riprova. Con EVENTS_MODE=off (solo sviluppo locale e test, dalla 0.3) non lancia mai, non scrive nulla e restituisce null.

4. Consuma:

@OnEvent('evt.id.employment.changed', { consumer: 'hr.employment-changed', maxAckPending: 50 })
async onEmploymentChanged(evt: EventEnvelope<EmploymentChanged>, tx: PrismaTx) {
  if (!evt.data.individualId) throw new EventValidationError('manca individualId'); // subito in dead letter
  await tx.… // gli effetti nella transazione della libreria
}

La libreria apre la transazione, inserisce (consumer, eventId) in inbox_processed (già presente: salta il gestore), chiama il gestore col tx, fa COMMIT e poi ack. Il gestore deve restare dentro il tx per gli effetti sul database; gli effetti fuori dal database (chiamate NATS, email) devono essere idempotenti.

Con Drizzle (@generazioneai/events/drizzle, dalla 0.2.0)

Tutto quello che la libreria chiede al database sta dietro OutboxStore (src/store.ts): Prisma è lo store di serie (PrismaOutboxStore, invariato), Drizzle è createDrizzleOutboxStore({ db }). Outbox, relay, consumer, firma, metriche e riprove sono gli stessi; cambia solo chi parla con Postgres.

1. Le tabelle: ri-esporta lo schema da quello del servizio e lascia a drizzle-kit la migrazione (drizzle-kit generate):

// schema/events.ts
export { outboxEvents, inboxProcessed } from '@generazioneai/events/drizzle';

Oppure copia sql/002-events-outbox-inbox-snake.sql (+ -rientro), equivalente SQL (colonne aggregate_type, published_at, …; timestamptz). Per riusare le colonne in camelCase di 001 (e di Prisma): createDrizzleOutboxStore({ db, columns: 'camel' }). Sono tabelle senza azienda: il relay legge le righe di tutte, quindi con una RLS sui tenant_id vanno nell'elenco delle eccezioni. 2. Il modulo:

EventsModule.forRoot({
  service: 'checkflow',
  store: createDrizzleOutboxStore({ db }),   // db = il client di drizzle (postgres-js, node-postgres, PGlite); niente `prisma`
  createStream: false,                       // CheckFlow non ha uno stream suo (eccezione E6): pubblica su `audit.checkflow.*` e consuma `evt.id.*`
})

3. Pubblica con il tx di Drizzle, nella stessa transazione del dato; fuori da una transazione lancia EVENTS_OUTSIDE_TX (stessa prova col SAVEPOINT):

await db.transaction(async (tx) => {
  await tx.update(invoices)…;
  await outbox.enqueue(tx, { subject: 'audit.checkflow.login', aggregate: { type: 'Session', id, version: 1 }, juridicalId, data: { … } });
});

4. Il gestore di @OnEvent riceve il tx di Drizzle (la transazione della libreria, la stessa dell'inbox): async on(evt: EventEnvelope<X>, tx: DrizzleExecutor).

Cose da sapere:

• Driver: JSON e UUID viaggiano come testo con il cast sul server, perché con postgres-js un oggetto passato a una colonna jsonb si serializza due volte; le righe si leggono sia come array (postgres-js) sia come { rows } (node-postgres). Provati tutti e due in CI.
• db può essere un oggetto con transaction() ed execute(): un servizio con la RLS ci mette un wrapper che apre la transazione nello scope giusto. Il relay e i consumer aprono db.transaction(...) senza azienda.
• Drizzle non ha il timeout di transazione di Prisma: relayBatchSize (100 di serie) e i 10 s di ogni pubblicazione sono il solo limite di quanto il relay tiene aperta la transazione.
• Lo store è scelto una volta in forRoot; con store e prisma insieme vince store.

Contratto

• Busta EventEnvelope<T> = { v: 1, id (UUID v7), subject, aggregateType, aggregateId, aggregateVersion, occurredAt, juridicalId?, data: T }.
• Intestazioni: Nats-Msg-Id (= id), Gen-Tenant, Gen-Event-Version (1), Gen-Sig (JWT della firma interna, audience events), traceparent.
• Subject ammessi in enqueue: evt.<proprio servizio>.… (stream EVT_<SVC>) e audit.<proprio servizio>[.…] (stream AUDIT, creato da genai-audit; NATS sceglie lo stream dal subject, quindi lo stream deve esistere). Altri prefissi: EVENTS_INVALID. @OnEvent su audit.… si lega a AUDIT.
• Stream EVT_<SVC> (evt.<svc>.>): lo crea il proprietario all'avvio (R3, file, 7 giorni o 5 GiB, discard: old, duplicate_window 10 m; streamReplicas o EVENTS_STREAM_REPLICAS per sviluppo). Il tetto (streamMaxBytes o EVENTS_STREAM_MAX_BYTES) JetStream lo riserva sul max_file_store del server: con un NATS piccolo (sviluppo, staging, CI) va abbassato, se no il secondo stream non si crea e l'avvio degli eventi resta a riprovare. DLQ (dlq.>) e AUDIT li crea genai-audit: se DLQ non esiste il messaggio non va perso, resta in coda.
• Consumer: pull, durevole, ack_wait 30 s (con working() ogni 3 s mentre il gestore lavora), max_deliver 8, backoff 10 s, 1 m, 5 m, 30 m, 2 h, 6 h, 12 h, filter_subject stretto. Il nome del durable è il nome logico con i punti sostituiti da _ (NATS non li ammette): hr.employment-changed → hr_employment-changed; nella dead letter e nelle metriche resta il nome logico. Lo stream si deduce dal subject (evt.id.… → EVT_ID).
• Riprove e dead letter: errore del gestore → nak con l'attesa crescente; all'ottava consegna (o subito per EventValidationError, busta non valida, firma rifiutata) copia in dlq.<stream>.<consumer> con Gen-Dlq-Reason, Gen-Dlq-Stream, Gen-Dlq-Consumer, Gen-Dlq-Seq, Gen-Dlq-Deliveries, Gen-Original-Id, Gen-Original-Subject e il corpo originale, poi term. Per i pod morti a metà ascolta l'avviso $JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.<stream>.> (un solo ricevente per servizio, gruppo di coda) e copia il messaggio per numero di sequenza; il Nats-Msg-Id della copia è dlq-<stream>-<consumer>-<seq>, quindi i due percorsi non duplicano.
• Modalità (0.3, decisione del 2026-11-10): EVENTS_MODE vale sempre on e EVENTS_SIGNATURE sempre enforce. Solo in sviluppo locale e nei test (SKILLERA_ENV dev/local, oppure NODE_ENV development/test, mai dentro un cluster) contano ancora i valori di prima; altrove un valore diverso si ignora con un avviso al primo uso, mai un errore all'avvio.
• Firma (EVENTS_SIGNATURE, di fatto enforce): rifiuta (dead letter) le firme assenti o errate. Firma chi ha il signer, verifica chi ha il verifier: senza uno dei due una riga nel log all'avvio (gli eventi partono senza Gen-Sig, o si accettano senza verifica), mai un errore. In sviluppo: off = niente; shadow = firma, verifica e conta gli errori senza rifiutare. La libreria non importa authz: signer e verifier si iniettano (con authz 0.7: signEventToken / verifyEventToken).
• Connessione: EVENTS_NATS_URL (ripiego NATS_URL); NATS_NKEY_SEED → autenticazione nkey; connection: {…} per altre opzioni di connect. Con NATS giù l'app parte lo stesso e l'outbox accumula.

Garanzie e limiti

• Pubblicazione esattamente una volta nella finestra di 10 minuti (Nats-Msg-Id); oltre, il doppione lo scarta l'inbox di chi consuma.
• Ordine: il relay legge per id (UUID v7), ma transazioni che committano fuori ordine, più repliche (SKIP LOCKED) e le riprove non lo garantiscono. Chi ha bisogno dell'ordine confronta aggregateVersion. I consumer elaborano un messaggio alla volta per consumer.
• Il relay tiene aperta una transazione mentre pubblica (fino a relayBatchSize righe, 100); per questo usa il timeout di 60 s di $transaction.
• JetStream non è uno scheduler: le code con regole di dominio restano nei servizi.

Metriche

EventsMetrics.render() produce testo Prometheus: events_outbox_pending, events_outbox_oldest_seconds, events_published_total{subject}, events_delivered_total{consumer}, events_nak_total{consumer}, events_term_total{consumer}, events_consumer_lag{consumer}, più events_duplicates_total{consumer} e events_signature_invalid_total{mode}. Il servizio le aggiunge al suo endpoint (o telemetry le legge).

Test

npm test -w @generazioneai/events (prima npm run test:prepare -w @generazioneai/events, genera il client Prisma 7 vero delle prove): le prove unitarie girano sempre; quelle che vogliono solo Postgres (drizzle.spec.ts, store-prisma.spec.ts: lo store sotto Prisma e sotto Drizzle, con postgres-js e node-postgres) vogliono EVENTS_TEST_DATABASE_URL; quelle di integrazione (integration.spec.ts, prisma.spec.ts, drizzle-flow.spec.ts: il flusso intero con JetStream) vogliono anche EVENTS_TEST_NATS_URL (un nats-server con JetStream), altrimenti sono saltate (con EVENTS_REQUIRE_INTEGRATION=1 mancano e il test fallisce). In locale:

docker run -d --rm --name evt-nats -p 14222:4222 nats:2.12 -js
docker run -d --rm --name evt-pg -e POSTGRES_PASSWORD=postgres -p 15432:5432 postgres:18
EVENTS_TEST_NATS_URL=nats://127.0.0.1:14222 EVENTS_TEST_DATABASE_URL=postgresql://postgres:[email protected]:15432/postgres npm test -w @generazioneai/events