noeva-etl — Technical Overview

Documento condivisibile. Spiega cos'è, per cosa è pensato, cosa può fare, come si espande e qual è la direzione del prodotto.

---

1. In una frase

noeva-etl è una piattaforma BI/ETL code-first in TypeScript: si costruiscono pipeline di integrazione dati (estrazione → trasformazione → caricamento) come software versionato, testato e documentato, non come job grafici opachi. Il nucleo di esecuzione di ogni pipeline è semplice, stateless e portabile: esegue una pipeline e restituisce un risultato completo (report di run), senza possedere database né stato proprio.

Il servizio si invoca in tre modi (vedi §5): batch (etl run, un container = una esecuzione, esito su stdout), daemon (etl worker, poller sulla tabella processes di Noeva, per l'integrazione con l'orchestrazione job del prodotto) e server + daemon (etl serve, il default dell'immagine: espone i grafi e le run all'app desktop noeva-etl-ui e avvia i worker nello stesso processo).

2. Per cosa è stato pensato

Sostituire l'approccio "ETL a scatola nera" (es. Talend) con uno code-first:

• Versionabile — ogni pipeline è codice in git, rivedibile in Pull Request.
• Testabile — connettori, trasformazioni e pipeline hanno test automatici (gate in CI).
• Documentato — ogni pipeline/decisione ha la sua scheda/ADR, aggiornati nello stesso commit.
• Portabile — un'unica immagine container gira sull'infra del cliente (Docker locale, ECS proprio o gestito da noi).
• Agnostico — nessun lock-in: sorgenti e destinazioni sono intercambiabili; il servizio non impone un DB o un cloud.

Target d'uso: integrazioni dati ricorrenti e affidabili (sincronizzazioni, fatti per la BI, normalizzazioni cross-database) dove correttezza, ripetibilità e tracciabilità contano.

3. Cosa può fare oggi

Capacità	Dettaglio
Primitive componibili	Source<T> (estrazione), Transform<I,O> (trasformazione), Sink<T> (caricamento), GraphPipeline (DSL a grafo graph()...build())
Nodi del grafo	source, transform, lookup (lettura a metà grafo da un DB o da un'API REST), ai (domanda a un modello per record), action (scrittura per record con un record in uscita), route (smistamento esclusivo su porte), join (fan-in a chiave), union (fan-in di flussi con la stessa forma), try-catch (ramo di gestione errori), sink, task (operazione senza record, prima o dopo il caricamento, ordinata dagli edge di controllo after)
Streaming	I connettori lavorano su AsyncIterable: dataset grandi senza caricarli in memoria
Contratti dati (Zod)	Validazione di schema su input e output di ogni pipeline (validate)
Trasformazioni	map, filter, mappatura dichiarativa di colonne mapColumns (select, rename, cast, colonne calcolate, con lineage di colonna), espansione 1:N (flatMapTransform, explode), aggregazione stateful riusabile groupBy (init/reduce/finalize), deduplica dedup, passaggio fra formato largo e lungo unpivot/pivot, ordinamento stabile sort, lotti di dimensione fissa chunk, decodifica e codifica di file in mezzo al grafo parse/serialize, hash di riga hashRow e chiave surrogata surrogateKey, mascheramento e pseudonimizzazione dei dati personali mask, profilo delle colonne nel report del nodo profile, controlli sul dataset assert
Gestione errori	dead-letter per nodo (record invalidi instradati altrove, la run prosegue), ramo catch di un nodo try-catch, o fail-fast esplicito, che al primo errore fatale annulla anche i rami indipendenti — mai scarti silenziosi
Idempotenza	Sink con upsert/MERGE su chiave naturale: rieseguire non duplica
Connettori reali	PostgreSQL (cursor streaming + upsert ON CONFLICT o SCD2 + lookup + task SQL), SQL Server (streaming + MERGE o SCD2 + lookup + task SQL), MySQL e MariaDB (streaming + ON DUPLICATE KEY UPDATE + lookup + task SQL), file (filesystem/S3/SFTP/FTP × CSV/JSON/XML/XLSX/Parquet), Excel (excelSource, excelSink con append/upsert/replace e excelTask: modifica workbook .xlsx/.xlsm conservando grafici, pivot e macro, solo su filesystem), Noeva (file e cartelle dei workspace + upload con elaborazione AI come action + lookup su stato e testo estratto), HTTP (lookup e action su API REST, httpLookup e httpAction), Jira Cloud (jiraIssues: le issue di una JQL su API REST v3, in colonne normalizzate, e jiraIssueLookup per chiave; changelog, commenti e worklog completi come lista; jiraCreateIssue, jiraUpdateIssue, jiraTransition, jiraUpsertIssue per ID esterno, jiraAddComment e jiraAddWorklog con chiave di deduplica e jiraLinkIssues come action, testo in markdown), Zoho CRM (zohoCrmRecords: i record di un modulo con i soli campi dichiarati, in ordine di id, con la Records API v8 o, oltre i 100.000 record, con COQL), Odoo (odooRecords: i record di un modello con i soli campi dichiarati, in ordine di id, via XML-RPC in sola lettura, al più una chiamata al secondo; odooConteggi per i controlli), IMAP, SMTP e SendGrid (posta in lettura e invio, con OAuth2 dove la casella lo richiede), processo esterno (processoEsternoAction: un programma lanciato per record in un nodo action, senza shell)
Cross-database	Una pipeline può leggere da un DB e scrivere su un altro (es. PG → SQL Server)
GraphRunReport	Risultato auto-descrittivo di ogni run (vedi §6)
CLI & container	etl list / etl run <pipeline> / etl serve / etl runs list / etl state show / etl versions snapshot (anche --realign) / etl versions check / etl versions rollback / etl env check / etl catalog check / etl docs build / etl ci / etl init / etl agent sync / etl upgrade; immagine Docker
App desktop	noeva-etl-ui (Tauri): DAG delle pipeline, avvio manuale, storico ed esito delle run, nodo per nodo — mostra, non costruisce
Worker daemon	etl worker — poller su processes via noeva-server-api (claim/progress HTTP, mai Supabase diretto), dispatch verso il catalogo di etl.config.ts; una istanza serve N workspace

Esempi inclusi: le pipeline della sezione «Esempi» di questa guida, come orders-enriched-daily (join a chiave, fan-out, try/catch) e crm-leads-sync (copia fra due istanze Postgres).

4. Come funziona (architettura)

Ogni pipeline è un grafo di nodi (DAG), costruito in codice col DSL graph()…build():

flowchart LR
  ordini["source<br/>ordini"] --> valida["transform<br/>validate (contratto)"]
  clienti["source<br/>clienti"] --> join["join<br/>(a chiave)"]
  valida -.record invalido.-> scarti["dead-letter"]
  valida --> join
  join --> lookup["lookup<br/>(lettura DB o API)"]
  lookup --> guard["try-catch"]
  guard --> dwh["sink<br/>DWH"]
  guard --> file["sink<br/>file"]
  guard -.catch.-> errori["ramo errori"]

• Core (src/core) — primitive + runner; non contiene logica di dominio.
• Dependency Injection — la definizione della pipeline è agnostica: sorgente e sink concreti si iniettano all'avvio. Per questo lo stesso codice gira su DB diversi e si testa con connettori in-memory.
• Streaming pull-based — i record scorrono dai source ai sink senza essere caricati tutti in memoria; è il ritmo di scrittura a regolare quello di lettura.
• Tre percorsi per un record che fallisce: dead-letter sul nodo (il record va altrove, la run prosegue), ramo catch di un nodo try-catch (gli errori di un intero tratto del grafo finiscono su nodi di gestione dedicati), fail-fast (la run si ferma — obbligatorio per le aggregazioni stateful).
• Osservabilità — logger JSON strutturato + metriche; ogni run produce un GraphRunReport con i conteggi di ogni nodo.

Stack: TypeScript ESM strict, Node ≥ 26, pnpm, tsup, vitest, ESLint/Prettier, Zod. Dipendenze runtime: i driver dei database (pg, mssql, mysql2), le librerie per storage e formati dei file (AWS SDK, SFTP, FTP, CSV, XLSX, XML, Parquet), express per il server locale di etl serve. Nessuna dipendenza Supabase: il claim/report sulla tabella processes passa da noeva-server-api via HTTP (fetch nativo).

5. Come si usa e si distribuisce

# in un repo cliente, in sviluppo
pnpm etl list
pnpm etl run <pipeline>

# server + daemon: grafi e run via HTTP per l'app desktop, più i worker
pnpm etl serve

# solo daemon: polling continuo su processes (worker_type da env; uno o più
# workspace — NOEVA_WORKSPACE_IDS per il multi-workspace)
pnpm etl worker

• Configurazione solo da variabili d'ambiente (PG*, MSSQL_*, MYSQL_*, ETL_ENV, …); nessun segreto nel codice o nell'immagine.
• Tre modalità di invocazione: batch (etl run, un container = una esecuzione, poi termina — vedi il deploy); daemon (etl worker, resta in polling su processes finché non riceve SIGTERM/SIGINT — vedi il worker); server + daemon (etl serve, il CMD di default dell'immagine: un solo container per tutta la piattaforma — vedi etl serve).
• App desktop noeva-etl-ui: si distribuisce come applicazione Tauri, uguale per tutti i deploy.
• Modello di deploy: il deploy esegue un repo cliente, come immagine container o sotto PM2 dai sorgenti. Vedi il deploy.

6. Il report di run è il "prodotto"

Il nucleo di esecuzione non persiste nulla: esegue e restituisce un risultato completo. È il chiamante a conservarlo: la CLI tiene le ultime run in locale (etl runs list), il worker lo scrive nel job Noeva, ed etl serve conserva nello stesso storico locale le run tracciate che lancia lui.

• stdout = report in JSON (un oggetto, parsabile con jq); stderr = log.
• exit code: 0 se success/partial, ≠0 se failed.
• Anche in errore emette un report (status: failed) con dettagli e conteggi parziali.
• Contiene: stato, tempi, versione, topologia reale del grafo (nodi e archi, con i conteggi e i tempi di ogni nodo), parametri, volumi (extracted/written/rejected), categorie di scarto, metriche, errore — senza segreti né payload/PII.

Chi invoca decide cosa farne (archiviarlo, mandarlo in alert, ignorarlo). Vedi run-report.md.

7. Come si espande

Crescita per composizione, senza toccare il core. Comandi di scaffold coerenti generano struttura + test + doc:

• Nuova pipeline → /new-pipeline <nome> (schema, definizione, changelog, snapshot del piano, run, test, scheda doc).
• Nuovo connettore → /new-connector <nome> [source|sink] (verso un nuovo sistema: API, file/CSV, altri DB, code…).
• Nuova trasformazione → /new-transform <nome> (logica riusabile, pura e testabile).

Aggiungere un sistema = scrivere un Source/Sink che rispetta l'interfaccia; tutte le pipeline esistenti possono usarlo subito (DI). La logica di dominio vive nelle trasformazioni, riusabili tra pipeline.

8. Perché ci si può fidare (governance)

• Test obbligatori con soglie di coverage come gate di CI (GitHub Actions).
• Documentazione nello stesso commit del codice (schede pipeline, runbook, decisioni), e etl docs build che non costruisce il sito con una scheda o uno scopo mancante o un link rotto.
• Decisioni tracciate: il core e ogni repo cliente registrano le loro scelte strutturali.
• Assistenti di qualità versionati nel kit dell'agente: review ETL, audit dei contratti, sync della doc, autore dei test. Vedi Skill e subagent.

9. Direzione del prodotto (chiara e semplice)

Un motore ETL piccolo, code-first e agnostico, che esegue pipeline affidabili e restituisce un risultato completo — e cresce per connettori e pipeline, non per complessità.

Principi guida (stabili):

1. Semplice e stateless — il servizio è una funzione pipeline + config → report.
2. Agnostico — nessun DB interno, nessun cloud obbligato, nessun push integrato.
3. Code-first — niente builder grafici; le pipeline restano codice testato e documentato.
4. Portabile — un'immagine, eseguibile ovunque, configurata da env.
5. Espandibile per composizione — nuovi connettori/trasformazioni/pipeline, core stabile.

Cosa NON è (non-obiettivi, per scelta)

• ❌ Non è un pipeline builder grafico: l'app desktop noeva-etl-ui mostra i DAG e le run, ma le pipeline si scrivono in codice.
• ❌ Non è uno strumento di visualizzazione BI (per i grafici si collega Metabase/Grafana/Power BI alle tabelle prodotte).
• ❌ Non possiede un database di stato (storico/audit = archiviazione dei report, a cura del chiamante). Lo stato di pipeline delle source incrementali, cioè i loro watermark, vive in uno store che il chiamante inietta: noeva-server-api per il worker, un file locale per la CLI.

Evoluzioni possibili (senza tradire i principi)

• Scheduling esterno (cron / ECS Scheduled Task) — l'idempotenza rende sicure retry e sovrapposizioni.
• Nuovi connettori (data warehouse, code di messaggi) e nuovi resolver per il nodo lookup (oggi PostgreSQL, SQL Server, Noeva, API REST con httpLookup e Jira Cloud con jiraIssueLookup e i lookup delle liste Jira; un nuovo DBMS riusa connectors/shared/sql-lookup.ts).
• Orchestrazione per dipendenze/backfill complessi (es. Temporal TS), solo se e quando servirà.
• Persistenza opzionale dei report come adapter lato chiamante, mai dentro il servizio.

---

Riferimenti: guida allo sviluppo · connettori · report di run · deployment · etl worker · etl serve.