node-listen

In sintesi

Bridge tra il meccanismo nativo LISTEN/NOTIFY di PostgreSQL e il bus messaggi NATS, con un control plane HTTP per la gestione runtime dei mapping canale → topic. Il servizio (a) apre una connessione pg.Client dedicata, si sottoscrive ai canali Postgres attivi e ripubblica ogni NOTIFY come messaggio NATS, (b) gestisce il mapping canale postgres → topic NATS tramite la tabella nodelisten.listenchannelmapping e una API REST admin con hot-reload immediato (LISTEN/UNLISTEN sulla connessione attiva senza riavvio), (c) costruisce il subject finale come {NATS_NAMESPACE}.{toppicnats}, dove NATS_NAMESPACE resta sempre implicito e non è override-abile via API. L'ingress pubblico /listen-admin/* è autenticato da nginx con propagazione di X-User-Id ai fini dell'audit log.

Funzionalità principali

• Bridge Postgres LISTEN/NOTIFY → NATS publish: ogni NOTIFY ricevuto sui canali abilitati viene normalizzato in un NatsEnvelope e pubblicato sul subject NATS corrispondente
• Tabella nodelisten.listenchannelmapping come source of truth dei mapping canale Postgres → topic NATS, con flag abilitato, descrizione e metadata datacreazione/datamodifica
• 6 endpoint REST CRUD + reload (GET /channels, GET /channels/{id}, POST /channels, PATCH /channels/{id}, DELETE /channels/{id}, POST /channels/reload) per gestione runtime, con validazione Zod e OpenAPI completo
• Hot-reload runtime: la creazione di un mapping abilitato esegue LISTEN <ch> immediato sulla connessione Postgres attiva; la disabilitazione o l'eliminazione esegue UNLISTEN <ch> immediato; il cambio di toppicnats invalida la cache in-memory senza riconnessione
• {NATS_NAMESPACE} implicito non override-abile via API: il namespace applicativo è gestito esclusivamente dalla configurazione del deploy, lo schema Zod degli endpoint rifiuta payload che tentino di forzarlo
• Seed automatico da POSTGRES_LISTEN_CHANNELS al primo bootstrap: se la tabella è vuota, il ChannelMappingSeeder popola i mapping a partire dall'env (backwards compatibility con la configurazione legacy)
• Auth nginx con audit log: tutte le mutation (POST/PATCH/DELETE/reload) registrano actorUserId ricavato dall'header X-User-Id propagato da nginx tramite auth_request verso node-user-auth
• Riconnessione Postgres con ri-LISTEN automatico: alla perdita della sessione l'adapter pianifica un retry con backoff; al ripristino tutti i canali abilitato=true vengono ri-sottoscritti leggendo lo stato corrente della tabella

Architettura

Stack: Fastify v5 + @pzeta/fastify-utils v1.0.12 (plugin auth, healthcheck, openapi, security) · Inversify per la composition root · driver pg 8.x come client LISTEN/NOTIFY e come repository della tabella di mapping · client nats 2.x come publisher · Zod per la validazione di request/response e della configurazione · @pzeta/log per logging strutturato con correlation IDs.

Layout DDD (src/):

Pattern adottati: Hexagonal/Ports & Adapters (porte applicative isolano dominio da pg e nats), Observer (il ChannelRegistry notifica il PostgresListenerAdapter su create/update/delete così da emettere LISTEN/UNLISTEN puntuali), Repository per la persistenza dei mapping, Factory per la composition root, Bootstrapper con sequenza deterministica (config → logger → DI → migration → seeder → server → orchestrator → shutdown hooks).

Flusso dati — propagazione eventi:

PostgreSQL NOTIFY <ch> → pg.Client.on('notification')
  → PostgresListenerAdapter → HandleNotificationUseCase
  → ChannelRegistry.resolveTopic(ch)        // canale Postgres → toppicnats
  → NatsEnvelope (channel, payload, processId, timestamp, source)
  → NatsPublisher.publish({NATS_NAMESPACE}.{toppicnats}, envelope)
  → NATS

Flusso dati — control plane:

HTTP /listen-admin/channels  → nginx auth_request → node-user-auth
  → Fastify (X-User-Id propagato)
  → ChannelRoutes (Zod validation, OpenAPI)
  → Use case CRUD (audit log con actorUserId)
  → PostgresChannelMappingRepository (tabella nodelisten.listenchannelmapping)
  → ChannelRegistry.invalidate(...)         // Observer
  → PostgresListenerAdapter.applyDiff()     // LISTEN/UNLISTEN runtime

Casi d'uso

• Onboarding live di un nuovo canale: il team DB aggiunge un trigger che emette NOTIFY ordine_creato_v2, '<json>'; il team backend chiama POST /listen-admin/channels con { "channelpostgres": "ordine_creato_v2", "toppicnats": "orders.created", "descrizione": "Pubblicazione ordini v2" }; entro pochi secondi node-listen esegue LISTEN ordine_creato_v2 sulla connessione attiva e inizia a pubblicare su {NATS_NAMESPACE}.orders.created. Zero downtime, zero rolling restart
• Disabilitazione temporanea di un canale rumoroso: PATCH /listen-admin/channels/{id} con { "abilitato": false } esegue UNLISTEN immediato senza eliminare la riga, preservando descrizione e mapping per la riabilitazione futura
• Rebrand del subject NATS senza modifiche SQL: PATCH /listen-admin/channels/{id} con { "toppicnats": "orders.v3.created" } invalida la cache del registry e ridiriga i NOTIFY futuri sul nuovo subject; i nomi dei canali Postgres (e quindi i trigger SQL) restano invariati
• Audit operativo da UI admin: GET /listen-admin/channels?abilitato=true restituisce la lista paginata dei mapping attivi con descrizione, datacreazione, datamodifica per dashboard di osservabilità o reportistica
• Boot iniziale con configurazione legacy: alla prima accensione del cluster, se nodelisten.listenchannelmapping è vuota, il ChannelMappingSeeder legge POSTGRES_LISTEN_CHANNELS dall'env e popola la tabella creando un mapping 1:1 (channelpostgres == toppicnats); le accensioni successive non riapplicano il seed, garantendo che le modifiche operative via API non vengano sovrascritte

Identità & esposizione

Endpoint operazionali

Endpoint convenzionali esposti da tutti i microservizi PZeta basati su @pzeta/fastify-utils:

Configurazione

Variabili d'ambiente rilevanti per chi integra il servizio (lista completa in .env.example del repo). Tutte le variabili sono validate all'avvio tramite schema Zod: configurazione invalida = fail-fast.

L'utente PostgreSQL configurato necessita del privilegio CONNECT sul database, del permesso di eseguire LISTEN sui canali e dei diritti SELECT/INSERT/UPDATE/DELETE sullo schema nodelisten (creato dalla migration 001_initial_listenchannelmapping con pg_advisory_lock per evitare race su deploy paralleli). Nessun accesso ad altri schemi applicativi è richiesto.

Note eventing NATS

Sezione centrale del servizio: la trasformazione NOTIFY → subject è la responsabilità funzionale di node-listen, ortogonale al control plane HTTP.

Modello di comunicazione: node-listen è producer puro di subject NATS. Non si sottoscrive ad alcun subject e non implementa request/reply: il flusso eventi è strettamente unidirezionale (PostgreSQL → NATS). L'unico canale di "ingresso" del servizio è l'API REST /listen-admin/channels per la gestione dei mapping.

Naming dei subject: per ogni mapping (channelpostgres, toppicnats) con abilitato = true nella tabella nodelisten.listenchannelmapping, le notifiche ricevute sul canale Postgres vengono pubblicate su:

{NATS_NAMESPACE}.{toppicnats}

dove NATS_NAMESPACE è impostato dall'env del deploy ed è implicito (non viene mai esposto né accettato dagli endpoint REST). A differenza della versione legacy del servizio — in cui valeva la convenzione 1:1 {NATS_NAMESPACE}.{channelName} — il toppicnats è ora indipendente dal nome del canale Postgres: lo stesso canale può essere rinominato a livello NATS senza modificare i trigger SQL, e topic NATS gerarchici (es. orders.v2.created) sono utilizzabili anche se il nome del canale Postgres deve restare un identificatore semplice. Il subject finale è validato dal value object NatsSubject (no spazi, no segmenti vuoti, no leading/trailing dot).

Subject dinamici e scanner statico: poiché i subject derivano dalla tabella DB e non da costanti nel codice, lo scanner statico della documentazione non è in grado di rilevarli a build-time. È atteso che nats.json per questo servizio sia vuoto: la lista effettiva dei subject pubblicati nel deploy corrente è ottenibile via GET /listen-admin/channels?abilitato=true (campo toppicnats).

Schema del messaggio (NatsEnvelope): il payload originale del NOTIFY deve essere JSON valido; viene parsato in NotificationPayload e racchiuso in un envelope con metadata. La forma serializzata pubblicata su NATS è:

{
  "channel": "ordine_creato_v2",
  "payload": { "idordine": 1234, "stato": "nuovo", "...": "..." },
  "processId": 42891,
  "timestamp": "2026-05-14T18:54:06.706Z",
  "source": "node-listen"
}

Il campo channel riporta il nome del canale Postgres di origine (utile ai consumer per tracciare la sorgente anche dopo un rebrand del topic NATS); processId è il PID PostgreSQL del backend che ha emesso il NOTIFY ed è utile per il tracing; timestamp viene generato dal servizio al momento della ricezione (non al momento della scrittura DB).

Resilienza e semantica di delivery: il publish è best-effort fire-and-forget. Il client NATS è configurato con maxReconnectAttempts: -1, quindi tenta indefinitamente la riconnessione, ma le notifiche ricevute mentre il bus è irraggiungibile vengono perse (l'errore viene loggato). Non è attivo JetStream e non è implementato un buffer locale: la garanzia di delivery dipende dall'idempotenza dei consumer e dalla disponibilità del bus. Per workflow critici, la sorgente del trigger PostgreSQL dovrebbe persistere lo stato in una tabella outbox prima del NOTIFY così da poter ricostruire l'evento.

Riconnessione PostgreSQL e stato della tabella: alla perdita della sessione LISTEN (error o end), l'adapter pianifica un retry con backoff esponenziale. Al ripristino della connessione, l'adapter ri-legge nodelisten.listenchannelmapping filtrando abilitato = true ed emette LISTEN per ciascun canale corrente: eventuali modifiche operative effettuate via API durante il downtime (es. nuovi mapping, disabilitazioni) vengono quindi applicate automaticamente al primo reconnect. I NOTIFY emessi mentre la sessione era down sono persi (semantica nativa di LISTEN/NOTIFY PostgreSQL).

Dipendenze e dipendenti

Dipende da (servizi che questo servizio chiama):

• node-user-auth

Consumato da (chi chiama questo servizio):

• node-orchestrator
• node-notification

Infrastruttura (PostgreSQL, NATS, Redis, MinIO) non è elencata qui — vedi sezione Architettura del singolo servizio.