Connettore: File (storage × codec)

Astrazione componibile per leggere/scrivere file su qualunque storage in qualunque formato. Codice: src/connectors/file/.

Modello

Due concern ortogonali che si compongono:

• StorageProvider — stream di byte: list(prefix), get(path), put(path, data). Capacità facoltative, dichiarate in capabilities: replace (sostituzione atomica) e fingerprint (impronta del contenuto); oggi le offre solo filesystemStorage, gli altri storage non cambiano. (filesystem, S3, Azure Blob, GCS, FTP, SFTP — ognuno nella sua issue/connettore)
• FileCodec<T> — byte ↔ record: decode(bytes), encode(records). (CSV, JSON/NDJSON, XLSX, XML, Parquet — ognuno nella sua issue/connettore)

import { fileSource, fileSink } from "@noeva/etl-core/connectors/file";

// "CSV da SFTP" = sftpStorage + csvCodec; "Parquet su S3" = s3Storage + parquetCodec
const source = fileSource(storage, codec, { prefix: "in/" }); // o { path: "in/file" }
const sink = fileSink(storage, codec, { path: "out/result.csv" });

• fileSource: legge un path singolo o tutti i file di un prefix (concatenando i record); errori → ExtractError.
• fileSink: serializza i record col codec e li scrive sullo storage; written = record passati; errori → LoadError.

I codec anche come transform

Un codec non è legato allo storage: quando i byte arrivano da un'altra source (una cartella Noeva, un allegato di posta) o vanno a un altro sink insieme ai metadati del file, si usa in mezzo al grafo con le transform parse(name, codec, …) e serialize(name, codec, …). parse decodifica i byte portati da ogni record in una riga per riga del file; serialize compone un file per gruppo, anche con un codec diverso per ogni file. Dettagli, errori e tetto di memoria nel nodo transform.

Un codec può dichiarare columns, le colonne dei record decodificati quando le conosce prima di leggere il file: oggi csvCodec con header: false e columns, e xlsxCodec con columns. parse le mostra nel DAG.

Storage disponibili

Filesystem locale — filesystemStorage

Legge/scrive file su disco (node:fs, nessuna dipendenza). I path sono relativi a baseDir (sandbox: un path che ne esce è rifiutato). list è ricorsiva e restituisce path relativi con separatore /. get streamma i byte; put crea le directory mancanti.

Offre le due capacità facoltative dell'interfaccia (capabilities: ["atomic-replace", "fingerprint"]):

• replace(path, data) sostituisce il file in modo atomico: scrive su un temporaneo nella stessa cartella (.<nome>.noeva-tmp-<casuale>, nascosto da list) e lo rinomina sul destinatario. Se lo stream fallisce a metà il temporaneo è rimosso e l'originale resta intatto.
• fingerprint(path) dà un'impronta opaca (dimensione:mtime:inode) che cambia se il file cambia, anche per una sostituzione atomica; undefined se il file non esiste. Serve a chi scrive per accorgersi di una modifica concorrente.

import {
  filesystemStorage,
  csvCodec,
  fileSource,
} from "@noeva/etl-core/connectors/file";

const storage = filesystemStorage({ baseDir: "/dati/import" });
const source = fileSource(storage, csvCodec(), { path: "ordini.csv" }); // o { prefix: "in/" }

Storage remoti: S3, SFTP, FTP

I tre storage remoti prendono la config di un'istanza dal registry (ADR-0030) e costruiscono il client da sé: la pipeline non importa @aws-sdk/client-s3, ssh2-sftp-client né basic-ftp (ADR-0093 §4). La sessione è a noleggio (ADR-0103): si apre alla prima operazione (list, get, put) e si chiude quando non ne resta nessuna in corso, anche su errore e abort. Una fileSource su un prefisso apre una sessione per tutti i file, una fileSink una per la sua scrittura; a fine run non resta niente aperto. Una sessione su cui un'operazione fallisce o si interrompe a metà non si riusa: l'operazione successiva ne apre una nuova.

Opzioni comuni ai tre: name (default s3:<bucket>, sftp, ftp) e timeoutMs (default 30 secondi: per SFTP il tempo per stabilire la sessione, per FTP quello di ogni risposta del server, per S3 quello di connessione e di inattività di una richiesta). creaClient è la porta dei test, con un client finto (ClientS3, ClientSftp, ClientFtp): non serve per usare lo storage.

Per etl env check, environment(env) dichiara l'istanza con istanzaS3Dichiarata(id), istanzaSftpDichiarata(id) o istanzaFtpDichiarata(id): gli stessi manifest (MANIFEST_S3, MANIFEST_SFTP, MANIFEST_FTP) che legge la run.

basic-ftp non regge due operazioni insieme sullo stesso client: due nodi che lavorano in parallelo sullo stesso server FTP usano due ftpStorage distinti.

Amazon S3 (e S3-compatibili) — s3Storage

StorageProvider su S3 via AWS SDK v3 (list paginata, get streaming, put multipart in streaming con lib-storage). La config dell'istanza arriva da loadS3Registry(env).get(id) (S3_REGION, opzionale S3_ENDPOINT + forcePathStyle per MinIO e S3-compatibili, credenziali opzionali S3_ACCESS_KEY_ID / S3_SECRET_ACCESS_KEY, altrimenti la catena AWS standard). Il bucket lo dice la pipeline. I retry sono quelli standard dell'SDK.

import {
  s3Storage,
  loadS3Registry,
  csvCodec,
  fileSource,
} from "@noeva/etl-core/connectors/file";

const storage = s3Storage(loadS3Registry(env).get("dwh"), {
  bucket: "il-mio-bucket",
});
const source = fileSource(storage, csvCodec(), {
  prefix: "import/",
  instanceId: "dwh",
});

SFTP — sftpStorage

StorageProvider su SFTP via ssh2-sftp-client. list(prefix) elenca la directory prefix (non ricorsivo), get streamma il download, put carica creando le directory mancanti. La config dell'istanza arriva da loadSftpRegistry(env).get(id): SFTP_HOST, SFTP_PORT (default 22), SFTP_USER e una tra SFTP_PASSWORD e SFTP_PRIVATE_KEY (con entrambe vince la chiave).

import {
  sftpStorage,
  loadSftpRegistry,
  csvCodec,
  fileSource,
} from "@noeva/etl-core/connectors/file";

const storage = sftpStorage(loadSftpRegistry(env).get("gestionale"));
const source = fileSource(storage, csvCodec(), {
  prefix: "upload/",
  instanceId: "gestionale",
});

FTP/FTPS — ftpStorage

StorageProvider su FTP/FTPS via basic-ftp. list(prefix) elenca la directory prefix (non ricorsivo), get streamma il download, put carica creando le directory mancanti. La config dell'istanza arriva da loadFtpRegistry(env).get(id): FTP_HOST, FTP_PORT (default 21), FTP_USER, FTP_PASSWORD e FTP_SECURE (false/true/implicit, default false — FTPS è opt-in esplicito).

import {
  ftpStorage,
  loadFtpRegistry,
  csvCodec,
  fileSource,
} from "@noeva/etl-core/connectors/file";

const storage = ftpStorage(loadFtpRegistry(env).get("default"));
const source = fileSource(storage, csvCodec(), {
  prefix: "upload/",
  instanceId: "default",
});

In-memory — memoryStorage

StorageProvider in-memory (come arraySource/arraySink), utile nei test e per piccole pipeline. Accetta un seed (stringhe o byte) ed espone .files (Map path→byte) per le asserzioni.

import { memoryStorage } from "@noeva/etl-core/connectors/file";
const storage = memoryStorage({ "in/data.jsonl": '{"id":1}\n' });

Una cartella come coda — localFolderSource

Non è uno StorageProvider: l'unità è il file, non il record, e la cartella stessa fa da coda. localFolderSource legge i file direttamente contenuti in una cartella locale o di rete (E:\…, \\server\share\…), in ordine di nome e già letti per intero, così chi li elabora può spostarli a lavoro finito. moveLocalFile e writeLocalFile sono i due complementi.

import {
  localFolderSource,
  moveLocalFile,
  writeLocalFile,
} from "@noeva/etl-core/connectors/file";

localFolderSource({
  dir: "D:\\Import\\Da_elaborare",
  extensions: [".csv"], // senza distinguere maiuscole e minuscole
  minAgeMs: 60_000, // un file più recente può essere ancora in copia
  maxFiles: 50,
}); // .writeReport → quali file ha lasciato in cartella, e perché
await moveLocalFile(percorso, "D:\\Import\\Elaborati"); // crea la cartella; EXDEV → copia+cancella

• Non ricorsiva, ed è voluto: chi elabora sposta i file fatti o scartati in una sottocartella, e da lì non vengono più letti.
• Una cartella che non esiste è un ExtractError, non una cartella vuota: una share non montata sembrerebbe altrimenti una giornata senza file. Un singolo file sparito o bloccato (ENOENT, EBUSY, EPERM, EACCES) si salta con un warning, e la run successiva lo ritrova.
• L'età minima conta l'evento più recente fra modifica, creazione e cambio di stato: Explorer e robocopy conservano la data di modifica dell'originale, e un file a metà copia sembrerebbe vecchio. minAgeMs vale 0 se non impostato: nessuna attesa, nemmeno per un file datato nel futuro (orologio della share avanti), che altrimenti resterebbe indietro a ogni run.
• Perché un file è rimasto lì. La sorgente espone writeReport, da passare al nodo sorgente (anche combinato con altri report). Restituisce cartella (risolta: quella letta davvero), fileNonLetti e, se ce ne sono, elencoNonLetti con { file, motivo, dettaglio } per ognuno. I motivi: estensione, troppo-recente, data-nel-futuro (data avanti di almeno un secondo rispetto al server), non-leggibile, oltre-max-files, non-file. Le sottocartelle non compaiono. L'elenco si ferma a 50 voci (LIMITE_ELENCO_NON_LETTI), le altre si contano in elencoNonLettiOmessi. Lo stato è per run: la stessa sorgente può servirne più di una.
• writeLocalFile è atomica (<nome>.part e poi rename): chi preleva i file non vede mai un file scritto a metà, purché ignori i .part. Gli spostamenti riprovano qualche volta sugli errori transitori di Windows (antivirus, indicizzatore).
• Differenza da filesystemStorage: quello elenca ricorsivamente sotto una radice ed è pensato per fileSource + codec.

Codec disponibili

CSV — csvCodec

CSV conforme a RFC 4180 (csv-parse/csv-stringify). Decode in streaming; i valori restano stringhe (tipizzazione a valle con validate/Zod). Gestisce BOM, quoting (virgole e a-capo nei campi), delimitatori custom. Un errore dello stream a monte (un download interrotto, una source che fallisce) fa rigettare l'iterazione del codec, in decode come in encode, invece di lasciarla appesa; vale anche per csvRawCodec.

import { csvCodec } from "@noeva/etl-core/connectors/file";
csvCodec(); // header dalla prima riga, delimitatore ","
csvCodec({ delimiter: ";", columns: ["id", "nome"] });

Opzioni: delimiter, header (default true), columns (nomi senza header / ordine in encode), e per il decode unmappedFields e missingFields.

Righe e intestazioni che non combaciano. Il decode produce oggetti, quindi ogni campo deve trovare una chiave propria nel record:

• unmappedFields — i campi che non la trovano: quelli oltre l'ultima colonna, e quelli sotto un nome di colonna duplicato (resterebbe solo l'ultimo valore) o riservato (__proto__, nessun valore). "drop" (default) li perde in silenzio, "error" fa rigettare il decode.
• missingFields — le colonne che mancano a una riga corta, sempre ammessa. "omit" (default) le lascia assenti dal record, "empty" le mette a stringa vuota.

Chi riscrive il file che ha letto usa unmappedFields: "error", perché per lui un campo perso è un dato cancellato (esempio: cross-workspace-csv-merge).

csvCodec({ unmappedFields: "error" }); // "1,a,zzz" sotto "id,val" → errore, non { id, val }
csvCodec({ missingFields: "empty" }); // "1" sotto "id,val" → { id: "1", val: "" }

CSV a campi posizionali — csvRawCodec

Per tracciati non conformi a RFC 4180: separatore fisso, nessun header, nessun quoting. Produce readonly string[] invece di oggetti.

Esiste perché csvCodec produce sempre oggetti (columns: header ? true : (columns ?? true)): i campi in eccesso rispetto all'header o vengono scartati in silenzio (default) o fanno fallire il decode (unmappedFields: "error"), ma non arrivano mai a valle. Su un tracciato dove una descrizione può contenere il separatore, quei campi sono esattamente l'informazione che serve per accorgersi dello sfasamento e ripararlo.

import { csvRawCodec } from "@noeva/etl-core/connectors/file";
csvRawCodec({ delimiter: ";", quote: false }); // niente quoting, in lettura e in scrittura
csvRawCodec({ delimiter: ";", bom: true }); // encode: BOM UTF-8 per Excel
csvRawCodec({ delimiter: ";", quote: false, eol: "\r\n" }); // encode: riproduce un tracciato Windows
csvRawCodec({ delimiter: ";", quote: false, encoding: "latin1" }); // byte restituiti identici, qualunque codifica

Opzioni:

• delimiter: separatore di campo.
• quote: false disattiva il quoting. In decode un tracciato scritto senza quoting può contenere virgolette come normale testo. In encode i campi escono così come sono, uniti dal separatore: è il modo di riprodurre un tracciato per un consumatore che non accetta altro. Un campo che contiene il separatore sposta le colonne che lo seguono, esattamente come nel file letto, quindi chi scrive valori propri deve garantire che non lo contengano.
• bom: antepone il BOM UTF-8 in encode.
• eol: fine riga in encode (default \n). Il parser accetta entrambi i fine riga, quindi per restituire quello dell'ingresso va rilevato dai byte.
• encoding: "utf8" (default) o "latin1". In utf8 un byte non valido diventa U+FFFD, e il file riscritto non è più quello letto. "latin1" è un trasporto: ogni byte diventa il carattere con lo stesso codice e torna lo stesso byte, quindi un file letto e riscritto esce identico sia UTF-8 sia Windows-1252. Il prezzo: un'accentata UTF-8 arriva come due caratteri (é come é), e in encode un carattere oltre U+00FF fa fallire la scrittura. Il BOM UTF-8 dell'ingresso si toglie comunque; bom: true con latin1 è rifiutato.

⚠️ Per default, in encode il quoting è conforme e attivo: un campo che contiene il separatore esce quotato. Il file prodotto quindi non ha il difetto di quello letto, ma ha una forma diversa: va dichiarato a chi lo consuma.

Excel/XLSX — xlsxCodec

Lettura/scrittura di file .xlsx (exceljs). A differenza del CSV preserva i tipi (numeri, date, booleani) e gestisce formule/hyperlink/rich text. Il decode carica il file (lo zip XLSX richiede il contenuto intero): adatto a export di dimensioni tipiche.

import { xlsxCodec } from "@noeva/etl-core/connectors/file";
xlsxCodec(); // primo foglio, header dalla prima riga
xlsxCodec({ sheet: "Ordini" }); // foglio per nome
xlsxCodec({ sheetIndex: 2, header: false, columns: ["id", "nome"] });

Opzioni: sheet / sheetIndex (default 1), header (default true), columns.

JSON / NDJSON — jsonCodec

JSON nativo, nessuna dipendenza. mode: "ndjson" (default) — un record per riga, streaming reale; mode: "array" — un unico array [ ... ] (il decode legge l'intero file). I tipi JSON (numeri, booleani, null, oggetti annidati) sono preservati.

import { jsonCodec } from "@noeva/etl-core/connectors/file";
jsonCodec(); // NDJSON
jsonCodec({ mode: "array" });

XML — xmlCodec

fast-xml-parser. Estrae un record per ogni occorrenza di recordTag; gli attributi sono mappati con prefisso @_. decode legge l'intero documento (XML non streamabile riga per riga).

import { xmlCodec } from "@noeva/etl-core/connectors/file";
xmlCodec({ recordTag: "item" }); // <root><item>…</item></root>
xmlCodec({ rootTag: "orders", recordTag: "order" }); // encode

Opzioni: recordTag (default record), rootTag (default root, per l'encode).

Parquet — parquetCodec

Formato colonnare tipato (@dsnp/parquetjs). decode legge l'intero file; encode richiede uno schema (esplicito o inferito dal primo record).

import { parquetCodec } from "@noeva/etl-core/connectors/file";
parquetCodec(); // schema inferito dal primo record
parquetCodec({ schema: { id: { type: "INT64" }, nome: { type: "UTF8" } } });

Opzioni: schema (mappa campo → tipo Parquet UTF8/DOUBLE/BOOLEAN/INT64/…).

Stato

Disponibili: interfacce + fileSource/fileSink (#21), codec CSV (#22), storage filesystem + memoryStorage (#27), storage Amazon S3 (#28), SFTP (#29), FTP/FTPS (#30), codec XLSX (#23), JSON/NDJSON (#24), XML (#25), Parquet (#26). In arrivo (issue dedicate): storage Azure/GCS — ognuno con il proprio ADR per la dipendenza.

Test

src/connectors/file/file-connector.test.ts: round-trip (write→read) via memoryStorage + codec di prova, lettura per prefisso, e gestione errori (ExtractError/LoadError).