Pipeline: crm-leads-sync

Copia (upsert) le prime 100 righe di public.crm_leads da un'istanza Postgres sorgente a una tabella di landing sull'istanza sink. Nasce come sync one-off / demo del registry connettori multi-istanza: usa due istanze Postgres (source e sink) risolte dallo stesso registry.

Flusso dati

0/3

Scegli una scena e premi ▶ per guardarla scorrere, oppure → per andare avanti un passo alla volta.

Lo scopo dei nodi

Nodo	Tipo	Scopo
crm_leads	source	Legge dal database di origine un primo gruppo di lead del CRM, sempre gli stessi a ogni run. La copia è una prova del passaggio di dati fra due database, non un trasferimento completo.
guard	try-catch	Protegge la copia lead per lead: un lead che non si riesce a controllare o a salvare finisce fra gli errori, e la run prosegue con gli altri.
validate	transform	Controlla che ogni lead sia completo e coerente prima di copiarlo. Un lead che non passa il controllo finisce fra gli errori invece di arrivare a destinazione, dove entrano solo lead utilizzabili.
crm_leads_import	sink	Salva i lead nell'area d'appoggio del database di destinazione, aggiornando quelli già copiati invece di duplicarli. L'area d'appoggio non è collegata agli altri dati dell'applicazione, così la copia non li altera.
error-report	sink	Conta i lead che non è stato possibile copiare e ne riporta il numero nel report, senza salvarli. Dice a colpo d'occhio se la copia è stata completa.

Il grafo qui sopra è disegnato dalla topologia reale della pipeline (getSpec()): nodi, archi, dead-letter e ramo catch sono quelli eseguiti. Premi ▶ per far passare un record, o → per avanzare un passo alla volta.

Il grafo usa il DSL graph() (src/core/graph-builder.ts): il nodo guard (.tryCatch) apre lo scope di gestione errori; il ramo protetto (validate → sink) si collega dalla porta di default, il ramo di gestione dalla porta catch (.sink("guard", errorReport, { fromPort: "catch" })). Tutti i record catturati confluiscono nell'unico nodo error-report, che non persiste nulla: conta i record andati in errore, li logga (crm-leads-sync.error-report) e ne espone il numero come volume del nodo (written).

Semantica del guard = per-record: ogni record che fallisce nel ramo protetto è deviato sul ramo catch (col suo raw) e lo stream continua — tutte le righe valide vengono caricate e tutte le invalide contate (guard.caught = N), con status: "partial". Per il comportamento "fermati al primo errore del branch" si usa .tryCatch(..., { mode: "circuit-breaker" }).

Gli errori della source non sono record: il guard cattura solo gli errori nati nei nodi a valle. Se la lettura di public.crm_leads si interrompe a metà, la run è failed, non partial: un caricamento parziale non deve sembrare una giornata con qualche lead invalido.

Sorgenti

• Istanza source, query SELECT … FROM public.crm_leads ORDER BY id LIMIT 100. status e social_profiles sono proiettati come testo (::text) così ogni campo viaggia come primitivo; il sink li re-inserisce come literal non tipizzati e Postgres li coerce a enum / jsonb / uuid / numeric sul lato destinazione.

Sink

• Istanza sink — tabella public.crm_leads_import creata IF NOT EXISTS (16 colonne, PK id, senza vincoli FK). La landing table evita di violare l'integrità referenziale di una eventuale crm_leads applicativa sul sink (i workspace/utenti di produzione potrebbero non esistere lì).

Configurazione (.env, git-ignored)

Il registry Postgres richiede le due istanze:

PG_INSTANCES=source,sink

# sorgente (es. Supabase pooler → SSL)
PG_SOURCE_HOST=<host>
PG_SOURCE_PORT=5432
PG_SOURCE_USER=<user>
PG_SOURCE_PASSWORD=<password>
PG_SOURCE_DATABASE=<db>
PG_SOURCE_SSL=true

# destinazione
PG_SINK_HOST=127.0.0.1
PG_SINK_PORT=54322
PG_SINK_USER=<user>
PG_SINK_PASSWORD=<password>
PG_SINK_DATABASE=<db>

Nota (breaking): PG_INSTANCES è globale per il connettore Postgres. Impostandola a source,sink si rimuove l'istanza implicita default: altre pipeline che usano l'istanza default vanno riconfigurate aggiungendo il proprio id alla lista e le rispettive PG_<ID>_*.

Nel piano

getSpec() usa connettori stub, senza connessione. Host, porta e database di crm_leads e crm_leads_import vengono dalla risoluzione delle istanze source e sink, ognuno con la sua variabile (PG_SOURCE_HOST, PG_SINK_DATABASE…) e la sua origine; l'istanza è in connection, e in fondo c'è la credenziale, che dice solo se la password è impostata.

Le due istanze ricadono sull'esempio ognuna da sola: un sink incompleto mostra postgres.esempio con le variabili mancante, e il source resta risolto. Se PG_INSTANCES non dichiara source o sink, i loro nodi portano anche configurazione: non valida: da controllare PG_INSTANCES.

Le variabili si vedono prima di lanciarla con pnpm run etl env check: la pipeline dichiara le due istanze che la run apre (environment, con istanzaPgDichiarata), quindi anche PG_INSTANCES, utente e password di PG_SOURCE_* e PG_SINK_*, che il piano non mostra. I pool li apre il connettore con createPgPool: la pipeline non importa il driver.

Esecuzione

pnpm run etl run crm-leads-sync

Idempotenza

Upsert su id (ON CONFLICT (id) DO UPDATE): rieseguire la pipeline non duplica le righe — il conteggio sulla landing table resta stabile.

Test

• pipeline.test.ts — cattura per-record: le righe valide arrivano alla landing, le N in errore finiscono sul ramo catch del guard; il nodo di report le conta senza persisterle.
• run.test.ts — il piano: host, porta e database delle due istanze con variabile e origine, ricaduta sull'esempio per istanza.
• Integrazione cross-DB (PG → SQL Server): pg-mssql-cross-db.integration.test.ts — lo stesso grafo con una sorgente Postgres e un sink SQL Server, su tabelle sintetiche create dal test: caricamento e idempotenza (MERGE) end-to-end.

Note

• Non copia le tabelle referenziate (workspaces/users/companies): per popolare la vera public.crm_leads servirebbe portare anche i parent o bypassare i FK.
• Dati reali dei clienti: non vanno committati; restano nel Postgres locale.

Versioni

La storia della pipeline, dalla versione più recente: la voce di changelog di ogni versione (src/pipelines/crm-leads-sync/changelog.ts) e, sotto, cosa è cambiato nel piano rispetto alla precedente. Vedi versioni di pipeline.

v32026-10-09 · ETL-193, ETL-195

Modifiche

• Nessun cambio di comportamento: il piano è identico alla v2. Gli import seguono l'infrastruttura (changelog, configurazione delle istanze, report di volume) uscita da `src/pipelines/` (ADR-0093 §1), e la pipeline si dichiara nel catalogo di `etl.config.ts` invece che in un registro (ADR-0093 §2)

Il piano non cambia rispetto alla v2.

v22026-10-06 · ETL-153

Aggiunte

• Host, porta e database delle istanze `source` e `sink`, ognuno con variabile e origine (ADR-0089)
• Riga `credenziale` con lo stato della password, mai il valore
• L'istanza del registry sta in `connection`, come negli altri connettori

Modifiche

• Il piano ricade sulla configurazione d'esempio per istanza: un sink incompleto non nasconde il source

Rimozioni

• Il dettaglio `instance`, sostituito da `connection`

Il piano rispetto alla v1:

Nodi

• ~crm_leadsconnettore
• ~crm_leads_importconnettore

v12026-10-05 · ETL-120

Aggiunte

• Versione iniziale sotto versionamento (ADR-0086)

Prima versione: non c'è una versione precedente da confrontare.