Nodo: source

Il punto d'ingresso di un flusso: estrae record da un sistema esterno e li emette in streaming, uno alla volta, senza mai caricarli tutti in memoria. Ogni grafo ne ha almeno uno.

Metodo	`.source(id, source, { after?, purpose?, writeReport? })`
Ingressi	0
Uscite	≥1 (fan-out libero)
Dead-letter	no: un errore di estrazione ferma la run
Codice	`Source<T>` in `src/core/connector.ts`

Quando usarlo

Per ogni sistema da cui il processo parte: una tabella, un file, una cartella, una casella di posta, un workspace Noeva. Se invece i dati esterni servono ad arricchire record che scorrono già nel grafo, e vanno letti solo per le chiavi presenti, è un lookup.

Il contratto

interface Source<T> {
  readonly name: string;
  extract(ctx: RunContext, incremental?: IncrementalExtract): AsyncIterable<T>;
  readonly descriptor?: ConnectorDescriptor; // che cos'è e dove punta, per il DAG
  readonly columns?: readonly PlanColumn[]; // campi emessi, se non c'è uno schema Zod
  readonly watermark?: SourceWatermark; // presente se la source è incrementale
}

La source si inietta nella funzione che costruisce il grafo, non si istanzia dentro: così la pipeline si prova con arraySource e getSpec() mostra la topologia senza credenziali.

Source pronte

Funzione	Sistema	Scheda
`postgresSource`	PostgreSQL (cursor)	postgres
`mssqlSource`	SQL Server	sqlserver
`mysqlSource`	MySQL, MariaDB	mysql
`fileSource`, `localFolderSource`	File (storage × codec)	file
`excelSource`	Excel: uno o più fogli, a flusso	excel
`noevaFileSource`, `noevaFolderSource`	Workspace Noeva	noeva
`imapSource`	Casella di posta	imap
`jiraIssues`	Jira Cloud (issue di una JQL)	jira
`zohoCrmRecords`	Zoho CRM (record di un modulo)	zoho
`odooRecords`, `odooConteggi`	Odoo (record di un modello)	odoo
`arraySource(name, rows)`	In memoria: test e configurazione	—

Esempio

graph("fatture-giornaliere")
  .source("fatture", sources.fatture) // Source<unknown>: il contratto lo fa il nodo dopo
  .source("clienti", sources.clienti)
  .transform(
    "fatture",
    "fattura-grezza",
    validate("fattura-grezza", fatturaGrezzaSchema),
  )
  .transform(
    "clienti",
    "cliente-grezzo",
    validate("cliente-grezzo", clienteGrezzoSchema),
  );
// …

Una source emette record non ancora validati: conviene tiparla Source<unknown> e mettere un validate() subito dopo, con la sua dead-letter. È quello il contratto d'ingresso.

Estrazione incrementale

Una source è incrementale se dichiara watermark: la colonna monotona della sorgente (una data di modifica, un id crescente), il suo tipo e il valore da cui parte la prima run. Il runner fa il resto:

all'avvio legge lo stato di pipeline e passa a extract il valore da cui leggere;
durante l'estrazione la source segnala il watermark di ogni riga, e il runner ne tiene il massimo: è ciò che la source propone;
a run conclusa, dopo i sink e i post-task, committa il massimo, e solo se l'esito lo consente.

postgresSource({
  pool,
  instanceId: "erp",
  // Il predicato si scrive per esteso, con `>=` e il segnaposto `:watermark`.
  query:
    "SELECT id, importo, updated_at FROM ordini WHERE updated_at >= :watermark ORDER BY updated_at",
  watermark: {
    column: "updated_at", // colonna del risultato: va selezionata
    type: "timestamp", // oppure "integer"
    initial: "1970-01-01 00:00:00",
    lookback: 5 * 60_000, // opzionale, in millisecondi: copre le transazioni lunghe
  },
});

Esito della run	Il watermark avanza?
`success`	sì
`partial`	no, salvo `graph(nome, { commitOn: "partial" })`, e solo se ogni scarto è una dead-letter per record
`failed`	mai

Regole da tenere a mente:

Il confine è >=, non >: due run consecutive senza dati nuovi rileggono le righe al confine. Per questo una source incrementale scrive solo verso sink idempotenti.
Senza il segnaposto :watermark la costruzione fallisce (source SQL): una query incrementale non degrada in silenzio in un full scan.
Il watermark è testo, nella precisione del database: i microsecondi di Postgres non si perdono passando da una Date.
Lo store lo inietta chi lancia la run (ctx.stateStore): un file locale per etl run, noeva-server-api per il worker e per etl run --workspace <id>. Un grafo con source incrementali non parte senza. Nei test si usa MemoryPipelineStateStore.
Backfill: params.watermark.<nodeId> sostituisce in lettura il valore salvato, e non fa arretrare lo stato. Reset: etl state reset, l'unico modo di farlo arretrare.

Oggi sono incrementali postgresSource, mssqlSource, mysqlSource e jiraIssues con incremental (il filtro su updated lo scrive il connettore: vedi jira, «Modalità incrementale»). Operatività, diagnosi e procedure: runbook.

Partire dopo un altro nodo: `after`

Con after la source ha un gate: extract() non viene chiamato finché i nodi elencati non sono riusciti. Sono task o sink:

.source("ordini", ordini, { after: ["svuota-staging"] }) // dopo un task: pre-task
.source("fatti", fatti, { after: ["carica-dimensioni"] }) // dopo un sink: dimensioni prima dei fatti

Più nodi in after valgono in AND.
Tutto ciò che sta a valle aspetta con lei: un lookup sui fatti legge le dimensioni già caricate, e un sink a valle non chiama load finché il gate non si apre.
Se un predecessore non è riuscito la source è saltata: extract() non viene mai chiamato, e i nodi a valle sono saltati con lei. Vedi task.
build() rifiuta un after che crea un ciclo (il sink alimentato dalla source stessa) o un deadlock su un join o un union a valle.

Errori

extract è un contratto opaco: non isola il singolo record. Un errore di estrazione (ExtractError) interrompe la run, anche sotto un try-catch, perché la source sta a monte di ogni scope. Il retry non si applica alle source: rileggere non è sempre senza effetti.

Nel DAG e nel report

Il nodo mostra il descriptor (tipo di connettore, istanza, puntamenti; mai segreti) e le colonne dichiarate in columns. Nel report, extracted conta i record emessi. Una source saltata ha skipped: { because } e startedAt: null. Una source incrementale ha anche una voce in state: da dove ha letto, cosa propone, se ha committato.

Anti-pattern

Array.fromAsync(source.extract(ctx)), o qualunque cosa che legga tutto in memoria: si perde lo streaming e con lui la backpressure.
Istanziare il client o il pool dentro build…Pipeline: va iniettato.

Nodo: source ​

Quando usarlo ​

Il contratto ​

Source pronte ​

Esempio ​

Estrazione incrementale ​

Partire dopo un altro nodo: after ​

Errori ​

Nel DAG e nel report ​

Anti-pattern ​