Connettore: Noeva (file dei workspace)

Legge e scrive i file dei workspace Noeva passando da noeva-server-api, mai da Supabase diretto. Codice in src/connectors/noeva/. Decisioni: ADR-0014 (connettore dedicato, file singolo), ADR-0022 (workspace ammessi), ADR-0042 (cartella e multi-file).

È una famiglia a sé, non uno storage del connettore File: le operazioni che servono (risolvere un file per nome o per somiglianza, nuova versione contro sovrascrittura, marcatori nella metadata) non stanno nel contratto path-based list/get/put. I byte però sono gli stessi: per decodificarli si compone un codec del connettore File a valle.

Funzione	Tipo	Cosa fa
noevaFileSource	Source<NoevaFileContent>	Un file, risolto per id, nome esatto o ricerca fuzzy
noevaFolderSource	Source<NoevaFileContent>	Tutti i file non ancora elaborati di una cartella
noevaFileSink	Sink<Uint8Array>	Un file dal nome fisso, da uno stream di byte
noevaFilesSink	Sink<NoevaOutputFile>	Un file per record, ciascuno col suo nome e la sua metadata

Configurazione

Il client (NoevaHttpClient) si costruisce da una NoevaClientConfig:

Variabile	Obbligatoria	Note
NOEVA_API_BASE_URL	sì	es. https://api.noeva.ai; lo slash finale viene tolto
NOEVA_API_KEY	sì	API key service-role del workspace, mai una chiave personale
NOEVA_WORKSPACE_ID	sì	il workspace; con più workspace si usa NOEVA_WORKSPACE_IDS
NOEVA_WORKSPACE_IDS	no	lista separata da virgole; ha la precedenza su NOEVA_WORKSPACE_ID

La chiave deve essere service-role perché la ricerca file dell'API, con una chiave personale, vede solo i file di quell'utente. Il client la manda come Authorization: Bearer … e il workspace come header x-noeva-workspace.

Workspace ammessi. Le pipeline non costruiscono il client a mano: passano dal WorkspaceRegistry, che contiene solo i workspace configurati e fa fail-fast su qualunque altro. Così una pipeline non tocca mai un workspace non previsto (ADR-0022). L'API key è una sola per tutti i workspace, cambia solo l'id.

const configs = loadEtlWorkerConfigs(env).map((c) => c.apiClientConfig);
const registry = buildWorkspaceRegistry(configs);
const client = registry.client(workspaceId); // ConfigError se non configurato

loadNoevaClientConfig("NOEVA_") legge un solo workspace, ed è comodo nei test di integrazione.

Source — noevaFileSource

Risolve un solo file e ne emette metadati e byte (in streaming). Va indicata una fra tre opzioni:

• fileId — diretto, nessuna ricerca.
• fileName — match esatto sul nome. Nessun match è un ExtractError.
• fuzzyQuery — best-match per somiglianza (fuse.js, lato client, perché l'API cerca solo per sottostringa). fuzzyThreshold va da 0 (solo match perfetto) a 1 (accetta tutto), default 0.4. Nessun candidato sopra soglia, o due candidati a pari punteggio, è un ExtractError: il connettore non tira a indovinare.

noevaFileSource(client, { fileName: "test-etl.csv" });

Source — noevaFolderSource

Emette tutti i file non ancora elaborati di una cartella, uno per record. I byte di ogni file si scaricano solo quando vengono consumati: la cartella non finisce in memoria.

noevaFolderSource(client, {
  folderId,
  includeSubfolders: false, // default
  skipIfMetadataKey: "elaborato_da_etl",
  maxFiles: 50, // default
});

• skipIfMetadataKey — un file la cui metadata contiene già questa chiave è considerato elaborato e viene saltato. È il modo di rendere incrementale una pipeline che lavora su una cartella: la run scrive il marcatore quando carica l'output (vedi noevaFilesSink), la run successiva lo trova e salta il file.
• maxFiles — tetto per run (default 50, DEFAULT_MAX_FILES): protegge la prima run su una cartella con anni di storico. I file oltre il tetto passano alla run successiva.
• includeSubfolders — di default solo la cartella indicata.

Costo: il marcatore sta nella metadata, che l'elenco della cartella non restituisce. Serve quindi una GET /api/files/:fileId per ogni file della cartella, anche quelli già elaborati. A pochi file al giorno è irrilevante; su una cartella molto grande andrà ripensato (ADR-0042).

⚠️ Il nome della chiave del marcatore è un identificatore stabile: cambiarlo fa risultare non elaborato tutto lo storico della cartella.

Sink — noevaFileSink

Concatena lo stream di byte in un solo file dal nome fisso (fileName), per esempio l'output di un codec.

noevaFileSink(client, {
  fileName: "report.csv",
  mode: "new-version",
  mimeType: "text/csv", // default application/octet-stream
  folderId, // opzionale
});

Sink — noevaFilesSink

Carica un file per record: ogni NoevaOutputFile porta fileName, bytes, mimeType? e metadata?. È il sink per le pipeline che riscrivono N file d'ingresso.

noevaFilesSink(client, { mode: "new-version", folderId });

In più rispetto a noevaFileSink:

• loadOne: il sink è isolabile per-record. Dentro uno scope try-catch il fallimento dell'upload di un file finisce sul ramo catch con il file in mano, e gli altri passano (ADR-0032).
• La metadata viene fusa, non sostituita. L'API, a ogni upload, rimpiazza l'intera metadata del file: il sink rilegge quella esistente e la rimanda insieme alle chiavi nuove. Senza, ogni nuova versione cancellerebbe il marcatore scritto dalla run precedente.

Comportamenti comuni ai due sink

• mode — overwrite sostituisce il file; new-version ne crea una versione nuova.
• Idempotenza di new-version: prima di caricare, il sink scarica l'ultima versione e ne confronta lo SHA-256 col contenuto nuovo; se coincidono salta l'upload (written: 0). L'API non espone un checksum, quindi l'hash si calcola lato client. Conseguenza: un output identico al precedente non riceve il marcatore nella metadata, e il file verrà rielaborato alla run successiva — stesso esito, nessuna versione nuova.
• folderId — dopo l'upload il file viene assegnato alla cartella con una chiamata dedicata (addFileToFolder), perché l'upload da solo non lo fa. Su un file già nella cartella è un no-op.
• File grandi: se l'API sceglie l'upload TUS (file oltre la soglia del presigned upload) il sink fallisce con un LoadError: oggi è supportato solo l'upload presigned.
• Ogni errore di rete o di API diventa ExtractError (source) o LoadError (sink), con uno stage che identifica il connettore. Un errore della source aborta la run (una source non ha dead-letter e sta a monte di ogni guard); un errore di un sink dentro uno scope try-catch finisce sul ramo catch.

Descriptor (UI del DAG)

Tutti espongono baseUrl e workspace — mai l'API key, che il client non espone nemmeno con un getter — più i puntamenti propri:

kind	Da	Dettagli in più
noeva-file	noevaFileSource	fileId / fileName / fuzzyQuery
noeva-file	noevaFileSink	fileName, folderId
noeva-folder	noevaFolderSource	folderId, includeSubfolders, skipIfMetadataKey, maxFiles
noeva-files	noevaFilesSink	mode, folderId

Contratto generale: node-descriptor.md.

Test

• Unit colocati in src/connectors/noeva/*.test.ts: client (envelope delle risposte), config, ricerca fuzzy, source e sink (hash-compare, fusione della metadata, folderId), descriptor con test anti-segreto, registry dei workspace.
• Integrazione manuale contro un workspace reale: tests/integration/noeva-file-version-bump.integration.test.ts, vedi il runbook. bumpVersion (esportato dal connettore) è l'helper che quel test usa per modificare il JSON prima di ricaricarlo.