Documentazione

Tema

Guida al progetto — noeva-etl ​

Guida di orientamento pratica: come funziona il progetto e come svilupparci. Per la visione di prodotto vedi Technical Overview; per le regole vincolanti vedi ../CLAUDE.md.

1. Cos'è in breve ​

noeva-etl è un motore ETL code-first: le pipeline di integrazione dati (estrazione → trasformazione → caricamento) sono scritte in TypeScript, testate con vitest e documentate — non job grafici stile Talend, ma software vero.

Il servizio è stateless: non ha un proprio database, esegue una pipeline e restituisce un GraphRunReport (JSON). Si distribuisce come immagine Docker con entrypoint la CLI etl, che si avvia in tre modi:

  • etl serve — il default dell'immagine: nello stesso processo espone la topologia dei grafi via HTTP (per noeva-etl-ui) e avvia i worker. Un solo comando fa girare la piattaforma, in dev come in Docker (ADR-0028, runbooks/graph-server.md).
  • etl worker (src/worker/) — solo il daemon: resta in polling sulla tabella processes di Noeva e dispatcha le pipeline i cui job vengono creati dal sistema Noeva stesso (workspace/scheduled job), invece che da un cron esterno. Il polling passa sempre tramite noeva-server-api, mai un client Supabase diretto (ADR-0019, che supera ADR-0017; runbooks/etl-worker.md). Mai insieme a un serve nello stesso ambiente, o i job vengono presi due volte.
  • etl run <pipeline> — batch: una esecuzione, report su stdout, poi termina.

In tutti e tre i casi il nucleo che esegue la pipeline è lo stesso: cambia solo chi la fa partire e dove finisce il report.

2. Le primitive del core ​

Tutto il framework (src/core/, senza logica di dominio) ruota attorno a quattro concetti base, mappati ai componenti Talend:

PrimitivaFileCosa fa
Source<T>src/core/connector.tsestrae record da un sistema esterno come AsyncIterable<T> (streaming, mai tutto in memoria)
Transform<I,O>src/core/transform.tstrasforma uno stream I in uno stream O; pura, senza side-effect esterni
Sink<T>src/core/connector.tscarica i record su un sistema di destinazione; deve essere idempotente (upsert/MERGE)
GraphPipelinesrc/core/graph-builder.tsil processo: un grafo di nodi costruito col DSL graph(name)…build()

Il grafo è l'unico modo di costruire un processo: il vecchio DSL lineare (pipeline().pipe(), RunReport, run(ctx, { deadLetter })) è stato rimosso (ADR-0036). Anche una catena semplice è un grafo con un nodo per stadio:

ts
graph("orders-daily")
  .source("orders", source)
  .transform("orders", "raw-contract", validate("raw-contract", rawSchema), {
    deadLetter: rejects,
  })
  .transform("raw-contract", "to-fact", mapTransform("to-fact", toFact))
  .transform("to-fact", "fact-contract", validate("fact-contract", factSchema))
  .sink("fact-contract", sink)
  .build();

I sei tipi di nodo:

NodoMetodoCosa fa
source.source(id, source)punto d'ingresso di uno stream
transform.transform(from, id, t, opts)trasformazione pura; accetta deadLetter
lookup.lookup(from, id, config, opts)query SQL in sola lettura in mezzo al grafo, a batch (ADR-0041); accetta deadLetter
join.join(id, { build, probe, … })unico fan-in: join a chiave inner/left, lato build in memoria e probe in streaming; accetta deadLetter
try-catch.tryCatch(from, id, opts)apre uno scope di gestione errori sui nodi a valle, con una porta catch
sink.sink(from, sink)scrittura verso l'esterno; l'id del nodo è sink.name

Il fan-out invece è libero: basta agganciare più nodi allo stesso from, e il runner duplica lo stream (tee). La forma del grafo è validata a runtime su .build() (validateGraph() in src/core/graph.ts): cicli, id sconosciuti o più di un ingresso su un nodo diverso da join esplodono come GraphValidationError alla costruzione — vedi ADR-0016 per il perché. Esempio di riferimento completo: src/pipelines/orders-enriched-daily/.

Altri moduli core:

  • context.ts — RunContext: env, params, logger, metriche, AbortSignal.
  • observability.ts — logger JSON strutturato + contatori.
  • errors.ts — gerarchia EtlError (ValidationError, ExtractError, LoadError, ConfigError, GraphValidationError, JoinError, LookupError) per distinguere errori di dominio da bug; il contesto porta anche nodeId.
  • transform.ts — gli helper mapTransform, filterTransform, validate(name, schema) (contratto Zod).
  • graph-runner.ts — esegue il grafo e produce il GraphRunReport.
  • lineage.ts, progress.ts, schema-columns.ts — identità dei record, report di avanzamento per nodo, colonne dei nodi per la UI.

Il dettaglio operativo (porte, regole di validazione, registry, metadati per la UI) è nella skill etl-graph (.claude/skills/etl-graph/).

3. Come gira una pipeline ​

Il grafo è pull-based: la domanda parte dai sink e risale verso le source, quindi è il ritmo di scrittura a regolare quello di lettura (backpressure gratis). I record scorrono in streaming, mai tutti in memoria — tranne il lato build di un join.

Quando un record fallisce ci sono tre percorsi, dal più locale al più drastico:

  • Dead-letter per nodo — { deadLetter: sink } su un transform, un lookup o un join. Il record che solleva EtlError esce come envelope DeadLetter su quel sink e lo stream continua. Adatto alle trasformazioni stateless (map/filter/validate).
  • Nodo try-catch — protegge tutti i nodi raggiungibili dalla sua porta out, sink compresi. Di default è per-record: ogni record che fallisce va sul ramo catch, che è fatto di nodi veri e può loggare, rimappare o instradare altrove. In circuit-breaker al primo errore la sub-stream si ferma (ADR-0029, ADR-0031).
  • Fail-fast — il default, senza dead-letter né guard: l'errore aborta la run. È l'unica scelta sensata per le trasformazioni stateful (es. groupBy in src/transforms/group-by.ts): un'aggregazione spezzata per record darebbe un risultato falso.

Un errore che non è un EtlError (un TypeError, un bug) aborta sempre la run, anche dentro un guard: un bug non va scaricato in un sink di scarto, va visto. Qualunque sia la scelta, nessun record sparisce in silenzio: ogni scarto è contato nel report.

Ogni run() produce un GraphRunReport (src/core/graph-runner.ts):

  • stato — success, partial (arrivata in fondo con scarti o errori catturati) o failed;
  • totali — extracted (somma delle source), written (somma dei sink: su un fan-out lo stesso record scritto due volte conta due), rejected, rejectedReasons;
  • per nodo — nodes[] ed edges[] con la topologia reale e i contatori di ogni nodo, caught sui guard, tempi di attraversamento;
  • metriche ed errore — mai segreti o payload.

È il "prodotto" del servizio: chi invoca la pipeline decide se archiviarlo, alertare, ecc. Schema completo in conventions/run-report.md.

4. Cosa c'è già pronto ​

Connettori (src/connectors/):

  • PostgreSQL — source con cursor streaming, sink con upsert ON CONFLICT.
  • SQL Server — source streaming, sink con MERGE.
  • File (src/connectors/file/) — architettura a due assi componibili:
    • storage: dove vivono i file (locale, S3, SFTP, FTP/FTPS);
    • codec: come si (de)serializzano — CSV (anche grezzo, csvRawCodec), JSON/NDJSON, XML, XLSX, Parquet. Storage e codec si combinano liberamente (es. CSV su S3, Parquet su SFTP).
  • Noeva (src/connectors/noeva/) — file dei workspace Noeva via noeva-server-api: lettura di un file o di una cartella intera, scrittura di uno o più file in overwrite o new-version (ADR-0014, ADR-0042).

Trasformazioni riusabili (src/transforms/): group-by.ts (aggregazione stateful generica init/reduce/finalize).

Pipeline di esempio:

  • src/pipelines/langfuse-sessions-daily/ — legge le sessioni da Postgres, le aggrega per progetto/giorno con groupBy, carica i fatti su SQL Server. È l'esempio end-to-end cross-database, idempotente, da usare come riferimento per un grafo a catena (una source, un sink; schema, pipeline, run, test).
  • src/pipelines/orders-enriched-daily/ — dataset demo in-memory, esempio di riferimento per una pipeline a grafo: due sorgenti, join a chiave, fan-out su due sink, dead-letter isolato su due nodi indipendenti (vedi docs/pipelines/orders-enriched-daily.md).

Le pipeline disponibili si registrano in src/pipelines/registry.ts e sono eseguibili via CLI (src/cli.ts, entrypoint pnpm run etl).

Worker daemon (src/worker/) — poller sulla tabella processes:

  • config.ts — loadEtlWorkerConfig legge il singolo workspace da NOEVA_WORKSPACE_ID, NOEVA_API_BASE_URL, NOEVA_API_KEY (obbligatorie), ETL_WORKER_TYPE (default "etl"), ETL_POLL_INTERVAL_MS (default 2000); fail-fast se mancante/invalida. loadEtlWorkerConfigs (plurale) legge invece la lista NOEVA_WORKSPACE_IDS (workspace separati da virgola, stessaNOEVA_API_KEY per tutti) — così una singola istanza serve N workspace (ADR-0021) — con fallback al singolo NOEVA_* se la variabile non è impostata.
  • run-workers.ts — startWorkers: crea un EtlProcessWorker per config workspace e ne gestisce lo shutdown combinato; è ciò che rende l'istanza multi-workspace (un poller indipendente per workspace).
  • processes-api-client.ts — ProcessesApiClient, client HTTP verso noeva-server-api (stessi header Authorization: Bearer/x-noeva-workspace del connettore workspace files): claimNextEtlProcess (POST /api/processes/etl/claim, CAS+filtro per worker_type/workspace_id fatti server-side) e reportProgress (PATCH /api/processes/etl/:processId/progress). Il worker non ha mai credenziali Supabase dirette.
  • dispatch.ts — risolve process.function in registry.ts (match esatto, fail-fast se sconosciuta), esegue la pipeline e mappa il GraphRunReport sull'esito riportato via ProcessesApiClient.reportProgress (completed/failed).
  • etl-process-worker.ts — il loop start()/stop(), un job alla volta per poller, shutdown cooperativo su SIGTERM/SIGINT. Con più workspace, i poller girano in parallelo e sono isolati fra loro.

Dettagli operativi (come inserire un job di test, monitoraggio, recovery, config multi-workspace): runbooks/etl-worker.md. Decisioni architetturali: il polling passa da noeva-server-api invece di un client Supabase diretto nel worker (ADR-0019, supersede ADR-0017); una singola istanza serve N workspace via N poller (ADR-0021).

5. Setup e comandi ​

Requisiti: Node ≥ 20, pnpm 9. Package manager: solo pnpm, mai npm/yarn.

bash
pnpm install

pnpm run check          # gate completo: typecheck + lint + format + test:cov + build + sito doc, varianti pubblica e app (= CI)
pnpm run test:watch     # TDD
pnpm run test:cov       # coverage con soglie (gate: lines/statements/functions 80%, branches 75%)
pnpm run lint:fix       # autofix lint
pnpm run format         # autofix formattazione
pnpm run docs:dev       # sito di documentazione in locale, variante pubblica (ADR-0043)
pnpm run docs:dev:app   # variante dell'app desktop, con le schede pipeline (ADR-0044)

pnpm run etl list                          # elenca le pipeline registrate
pnpm run etl run langfuse-sessions-daily   # esegue una pipeline (legge .env)
pnpm run etl:serve                         # grafi via HTTP + worker (vedi runbooks/graph-server.md)
pnpm run etl worker                        # solo il daemon su processes (vedi runbooks/etl-worker.md)
pnpm run etl runs list                     # storico locale delle run (ADR-0039)
./dev-desktop.sh                           # dev completo: etl serve + app desktop Tauri

pnpm run test:integration   # test contro sistemi reali (DB, S3, SFTP), separati dalla suite rapida

Config solo da variabili d'ambiente (.env, git-ignored): PG*, MSSQL_*, ETL_ENV, credenziali S3/SFTP, ecc. Ogni variabile va documentata in .env.example e nella scheda della pipeline che la usa. Per etl worker servono inoltre NOEVA_WORKSPACE_ID, NOEVA_API_BASE_URL, NOEVA_API_KEY (obbligatorie) e ETL_WORKER_TYPE/ETL_POLL_INTERVAL_MS (opzionali) — nessuna credenziale Supabase: il worker parla solo con noeva-server-api. Per servire più workspace con una sola istanza, in alternativa a NOEVA_WORKSPACE_ID si imposta NOEVA_WORKSPACE_IDS alla lista dei workspace (separati da virgola); la NOEVA_API_KEY è una sola e vale per tutti (vedi il runbook).

Distribuzione: docker build -t noeva-etl ., poi ./deploy-ecr.sh (build multi-arch + push su ECR) e ./run-ecr.sh <pipeline> per l'esecuzione da ECR in locale. L'immagine parte di default con serve. La UI non passa da ECR: è un'app desktop Tauri (ADR-0025). Dettagli: runbooks/deployment.md.

6. Come sviluppare — workflow consigliato ​

  1. Capire il requisito. Se ambiguo, chiedere prima di scrivere codice.
  2. Aprire un worktree per l'attività: .claude/worktrees/<descrizione>/ con un branch feat/* o fix/* da origin/dev. Ogni attività ha la sua cartella, così se ne portano avanti più d'una in parallelo e la cartella principale resta su dev. Comandi e insidie: conventions/git-workflow.md.
  3. Cercare riuso prima di scrivere codice nuovo: un connettore/storage/codec esistente copre già il caso? Una libreria matura risolve il problema meglio di codice hand-rolled?
  4. Scaffold con i comandi dedicati (generano struttura + test + doc coerenti, vedi §7):
    • /new-pipeline <nome> — nuova pipeline
    • /new-connector <nome> [source|sink] — nuovo sistema esterno
    • /new-transform <nome> — nuova logica di trasformazione
  5. TDD: test prima (rosso) → implementazione minima (verde) → refactor. Per un bug fix: prima un test che lo riproduce, poi la correzione.
  6. Verificare: pnpm run check. Per una review mirata ETL: /etl-review o l'agente etl-reviewer (idempotenza, contratti, dead-letter, osservabilità).
  7. Documentare nello stesso commit: scheda pipeline (docs/pipelines/<nome>.md), ADR se la scelta è strutturale (nuova dipendenza, nuovo pattern), runbook se cambia un processo operativo. Se hai aggiunto, modificato o rimosso una funzionalità, a lavoro finito esegui il doc sync (/doc-sync o l'agente doc-syncer) e chiudi ogni deriva prima della PR.
  8. Commit/PR: mai direttamente su dev/staging/main — si committa nel worktree, sul suo branch, e si apre la PR verso dev. Dopo il merge si rimuovono worktree e branch. Promozione dev → staging → main sempre via PR, mai saltando un tier.

7. Aggiungere le cose più comuni ​

Una nuova pipeline ​

Una cartella src/pipelines/<nome>/ con: schema.ts (contratti Zod input/output), pipeline.ts (definizione con graph()...build(), sorgente/sink iniettati, non hardcoded), run.ts (wiring dei connettori concreti + lettura env, escluso dalla coverage perché è solo DI), pipeline.test.ts (end-to-end con arraySource/arraySink: record validi caricati, invalidi al dead-letter, metriche corrette). Registrarla in src/pipelines/registry.ts e creare docs/pipelines/<nome>.md.

Un nuovo connettore (Source o Sink) ​

Implementare l'interfaccia Source<T> o Sink<T> (src/core/connector.ts) verso il nuovo sistema (API, file, altro DB, coda). Grazie alla dependency injection, tutte le pipeline esistenti possono usarlo subito senza toccare il core. Per un nuovo storage/codec file, seguire il pattern a due assi già in src/connectors/file/ (storage e codec si combinano liberamente). Aggiungere un ADR se introduce una nuova libreria/dipendenza (vedi docs/architecture/adr/ per i precedenti: es. 0009 per S3/aws-sdk, 0011 per SFTP/ssh2, 0013 per Parquet). Documentare in docs/connectors/README.md.

Una nuova trasformazione ​

Funzione pura in src/transforms/, senza side-effect verso sistemi esterni (quelli stanno solo in Source/Sink), così è testabile in isolamento. Riusabile tra pipeline diverse. Se stateful (come group-by.ts), documentare chiaramente che va usata in fail-fast: niente dead-letter sul suo nodo (vedi §3).

8. Cosa NON fare (vincoli non negoziabili) ​

  • Niente logica di dominio in src/core/ — il core resta agnostico e stabile.
  • Niente any in src/ — dati esterni entrano come unknown, si restringono con Zod.
  • Niente scarti silenziosi — ogni record invalido ha un percorso esplicito: dead-letter o fail-fast, mai un semplice skip.
  • Sink non idempotenti — un retry/riesecuzione non deve mai duplicare dati (upsert su chiave naturale, MERGE, o write-then-swap).
  • Niente segreti nel codice, nei log o nel report — solo variabili d'ambiente, documentate in .env.example.
  • Codice senza test o senza doc aggiornata non è "done" — vedi la checklist completa in ../CLAUDE.md §5.
  • Mai committare/pushare direttamente su dev, staging, main.

9. Mappa della documentazione ​

10. Strumenti dell'agente Claude Code ​

  • Slash command: /new-pipeline, /new-connector, /new-transform, /etl-review, /doc-sync.
  • Skill etl-graph — come costruire, cablare e configurare un grafo: nodi e porte, gestione errori, registry, metadati per la UI, pattern dato (idempotenza, incrementale, dead-letter, backpressure).
  • Subagent read-only in .claude/agents/: graph-auditor (struttura del grafo e cablaggio), etl-reviewer (correttezza dato), schema-contract-auditor (contratti Zod), doc-syncer, project-pulse. Scrivono file solo graph-builder (scaffold di una pipeline già progettata) e test-author (solo test).

Noeva è un marchio registrato di 4D S.R.L.

© 2026 4D S.R.L. · Gruppo4D