Connettore: File (storage × codec)

Astrazione componibile per leggere/scrivere file su qualunque storage in qualunque formato (vedi ADR-0007). Codice: src/connectors/file/.

Modello

Due concern ortogonali che si compongono:

• StorageProvider — stream di byte: list(prefix), get(path), put(path, data). (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 "../connectors/file/index.js";

// "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.

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.

import {
  filesystemStorage,
  csvCodec,
  fileSource,
} from "../connectors/file/index.js";

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

Amazon S3 (e S3-compatibili) — s3Storage

StorageProvider su S3 via AWS SDK v3 (list paginata, get streaming, put multipart in streaming con lib-storage). Vedi ADR-0009. Il S3Client è iniettato; loadS3ClientConfig("S3_") lo costruisce da env (S3_REGION, opzionale S3_ENDPOINT+forcePathStyle per MinIO, credenziali opzionali).

import { S3Client } from "@aws-sdk/client-s3";
import {
  s3Storage,
  loadS3ClientConfig,
  csvCodec,
  fileSource,
} from "../connectors/file/index.js";

const client = new S3Client(loadS3ClientConfig("S3_"));
const storage = s3Storage({ client, bucket: "il-mio-bucket" });
const source = fileSource(storage, csvCodec(), { prefix: "import/" });

SFTP — sftpStorage

StorageProvider su SFTP via ssh2-sftp-client (vedi ADR-0011). list(prefix) elenca la directory prefix (non ricorsivo), get streamma il download, put carica creando le directory mancanti. Client iniettato; loadSftpConfig("SFTP_") legge host/porta/ utente + una tra password e chiave privata.

import Client from "ssh2-sftp-client";
import {
  sftpStorage,
  loadSftpConfig,
  csvCodec,
  fileSource,
} from "../connectors/file/index.js";

const client = new Client();
await client.connect(loadSftpConfig("SFTP_"));
try {
  const storage = sftpStorage({ client });
  const source = fileSource(storage, csvCodec(), { prefix: "upload/" });
} finally {
  await client.end();
}

FTP/FTPS — ftpStorage

StorageProvider su FTP/FTPS via basic-ftp (vedi ADR-0015). list(prefix) elenca la directory prefix (non ricorsivo), get streamma il download, put carica creando le directory mancanti. Client iniettato; loadFtpConfig("FTP_") legge host/porta/ utente/password + secure (false/true/implicit, default false — FTPS è opt-in esplicito).

import { Client } from "basic-ftp";
import {
  ftpStorage,
  loadFtpConfig,
  csvCodec,
  fileSource,
} from "../connectors/file/index.js";

const client = new Client();
await client.access(loadFtpConfig("FTP_"));
try {
  const storage = ftpStorage({ client });
  const source = fileSource(storage, csvCodec(), { prefix: "upload/" });
} finally {
  client.close();
}

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 "../connectors/file/index.js";
const storage = memoryStorage({ "in/data.jsonl": '{"id":1}\n' });

Codec disponibili

CSV — csvCodec

CSV conforme a RFC 4180 (csv-parse/csv-stringify, vedi ADR-0008). Decode in streaming; i valori restano stringhe (tipizzazione a valle con validate/Zod). Gestisce BOM, quoting (virgole e a-capo nei campi), delimitatori custom.

import { csvCodec } from "../connectors/file/index.js";
csvCodec(); // header dalla prima riga, delimitatore ","
csvCodec({ delimiter: ";", columns: ["id", "nome"] });

Opzioni: delimiter, header (default true), columns (nomi senza header / ordine in encode).

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)) e gira con relax_column_count: i campi in eccesso rispetto all'header vengono scartati in silenzio. 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 "../connectors/file/index.js";
csvRawCodec({ delimiter: ";", quote: false }); // decode: niente quoting
csvRawCodec({ delimiter: ";", bom: true }); // encode: BOM UTF-8 per Excel

Opzioni: delimiter, quote (false disattiva il quoting in decode: un tracciato scritto senza quoting può contenere virgolette come normale testo), bom (antepone il BOM UTF-8 in encode).

⚠️ In encode il quoting è invece conforme e attivo: un campo che contiene il separatore esce quotato. Il file prodotto quindi non ha il difetto di quello letto — è un cambiamento di formato, da dichiarare a chi consuma il file.

Excel/XLSX — xlsxCodec

Lettura/scrittura di file .xlsx (exceljs, vedi ADR-0010). 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 "../connectors/file/index.js";
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 "../connectors/file/index.js";
jsonCodec(); // NDJSON
jsonCodec({ mode: "array" });

XML — xmlCodec

fast-xml-parser (vedi ADR-0012). 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 "../connectors/file/index.js";
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, vedi ADR-0013). decode legge l'intero file; encode richiede uno schema (esplicito o inferito dal primo record).

import { parquetCodec } from "../connectors/file/index.js";
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).