Connettori

I connettori sono le implementazioni reali di Source (estrazione) e Sink (caricamento) verso sistemi esterni. Vivono in src/connectors/<sistema>/ e si iniettano nelle pipeline (mai istanziati dentro la definizione).

Indice

Connettore	Source	Sink	Doc
PostgreSQL	✅ streaming (cursor)	✅ upsert idempotente (ON CONFLICT), o SCD2	postgres.md
SQL Server	✅ streaming (pause/resume)	✅ upsert idempotente (MERGE), o SCD2	sqlserver.md
MySQL e MariaDB	✅ streaming (mysql2)	✅ upsert idempotente (ON DUPLICATE KEY UPDATE)	mysql.md
File (storage × codec)	✅ fileSource	✅ fileSink	file.md
— storage: filesystem · S3 · SFTP · FTP/FTPS · memory	(vedi file.md)	(vedi file.md)	file.md
— codec: CSV (anche grezzo) · JSON/NDJSON · XML · XLSX · Parquet	(vedi file.md)	(vedi file.md)	file.md
Excel (workbook .xlsx/.xlsm)	✅ uno o più fogli, a flusso, colonne per titolo	✅ append, upsert per chiave e replace su workbook nuovo o esistente, che conserva grafici, pivot e macro; più excelTask (fogli e colonne)	excel.md
Noeva (file dei workspace)	✅ file singolo · cartella	✅ file singolo · multi-file, overwrite/new-version	noeva.md
IMAP (caselle di posta)	✅ messaggi di una cartella	— (action: flag, spostamento)	imap.md
SMTP (notifiche email)	— (action: invio con esito)	✅ una mail per record, isolabile	smtp.md
SendGrid (mail via API REST; mailAction/mailSink per scegliere il trasporto)	— (action: invio con esito)	✅ una mail per record, isolabile	sendgrid.md
HTTP (API REST)	— (lookup: httpLookup)	— (action: httpAction)	http.md
Jira Cloud (issue)	✅ JQL; lookup per chiave e liste	— (action: jiraCreateIssue, jiraUpdateIssue, jiraTransition, jiraUpsertIssue, jiraAddComment, jiraAddWorklog, jiraLinkIssues)	jira.md
Odoo (record di un modello, conteggi)	✅ un modello, campi dichiarati	—	odoo.md
Zoho CRM (record di un modulo)	✅ un modulo, campi dichiarati	—	zoho.md
OAuth2 (identità per IMAP, SMTP e Zoho)	—	—	oauth.md
Processo esterno (un programma sulla macchina del worker)	—	— (action: processoEsternoAction)	process.md

Noeva offre anche un nodo action (noevaUploadAction: upload con elaborazione AI, fileId in uscita), due resolver di lookup non SQL (noevaElaborazioneLookup, noevaContenutoLookup) e l'estrazione AI noevaAiExtract, la configurazione completa di un nodo ai che estrae campi validati con Zod: vedi noeva.md § Elaborazione AI.

Oltre a Source e Sink, ogni connettore SQL fornisce un resolver del nodo lookup, che interroga il DB in sola lettura a metà grafo, a batch di chiavi:

DBMS	Resolver	Parametri bindati	Doc
PostgreSQL	postgresLookup	$1, $2, …	postgres.md § Lookup
SQL Server	mssqlLookup	@k0, @k1, …	sqlserver.md § Lookup
MySQL, MariaDB	mysqlLookup	?, ?, …	mysql.md § Lookup

Ogni connettore SQL fornisce anche un task, per il nodo task del grafo: statement statici in una transazione, senza flusso di record. Serve a svuotare uno staging prima del caricamento e a scambiare le tabelle dopo.

DBMS	Task	Parametri bindati	Doc
PostgreSQL	postgresSqlTask	@nome → $1, $2, …	postgres.md § Task
SQL Server	mssqlSqlTask	@nome	sqlserver.md § Task
MySQL, MariaDB	mysqlSqlTask	@nome → ?	mysql.md § Task

La parte comune vive in src/connectors/shared/sql-task.ts. Un task con destructive: true cancella o sostituisce dati: prima di usarlo in produzione leggi il runbook.

Per un'API REST c'è httpLookup (http.md): una richiesta per batch o una GET per chiave, risposta validata con Zod, credenziali dal registry HTTP_INSTANCES. Per scrivere sulla stessa API c'è httpAction (http.md § Scritture), un nodo action che prosegue con l'esito della chiamata.

Per leggere Jira Cloud c'è jiraIssues (jira.md, ADR-0100): le issue di una JQL, pagina dopo pagina, con i siti nel registry JIRA_INSTANCES e il trasporto condiviso di http. Ogni campo dichiarato, per id o per nome, è una colonna normalizzata, con il testo ADF convertito da adfInTesto (ADR-0101). Per arricchire record che hanno già una chiave (ABC-42) c'è jiraIssueLookup, un nodo lookup con le issue nella stessa forma; changelog, commenti e worklog completi di una issue arrivano come lista con jiraChangelogLookup, jiraCommentsLookup e jiraWorklogsLookup (jira.md § Le liste). Per scriverci ci sono i nodi action jiraCreateIssue, jiraUpdateIssue, jiraTransition, che porta una issue in uno stato per nome, jiraUpsertIssue, che crea o aggiorna per ID esterno, jiraAddComment e jiraAddWorklog, che aggiungono commenti e worklog senza doppioni con una chiave di deduplica, e jiraLinkIssues, che collega due issue senza duplicare il link (jira.md § Scritture, ADR-0102), con il testo in markdown convertito in ADF da marklassian.

Per leggere Zoho CRM c'è zohoCrmRecords (zoho.md, ADR-0106): i record di un modulo con i soli campi dichiarati, che sono anche la regola di privacy, in ordine di id: con la Records API fino a 100.000 record, con la Query API (COQL) a chiavi crescenti di id oltre. Le organizzazioni stanno nel registry ZOHO_INSTANCES, il token lo rinnova un'identità OAuth2 e il trasporto è quello condiviso di http.

Per leggere Odoo c'è odooRecords (odoo.md, ADR-0108): i record di un modello con i soli campi dichiarati, a pagine in ordine di id, dopo un fields_get che rifiuta i campi assenti o non stored; odooConteggi dà i search_count da confrontare con le righe caricate. Solo metodi di lettura, al più una chiamata al secondo per database e mai due in parallelo, come vuole l'Acceptable Use Policy di Odoo. I database stanno nel registry ODOO_INSTANCES; il protocollo è XML-RPC, JSON-2 arriverà con Odoo 19.

Per lanciare un programma su dei file del record, con file e un riepilogo JSON in uscita, c'è processoEsternoAction (process.md): un nodo action senza shell, con ambiente da allowlist, timeout obbligatorio e una cartella temporanea per record.

La query è la stessa sui due database — segnaposto @chiavi, stesso controllo di sola lettura, stessa normalizzazione delle chiavi — perché la parte comune vive in src/connectors/shared/sql-lookup.ts. Un nuovo connettore SQL ottiene il lookup aggiungendo solo la sintassi dei parametri e la chiamata al driver.

Regole comuni

• DI del client: il Pool/connessione è iniettato dal chiamante, che ne gestisce il ciclo di vita. Fanno eccezione gli storage remoti di file (S3, SFTP, FTP): prendono la config dell'istanza e aprono e chiudono la sessione da sé (ADR-0103).
• Streaming: i Source restituiscono AsyncIterable, mai array interi.
• Idempotenza: i Sink usano upsert/merge/swap sulla chiave naturale.
• Segreti: solo in .env (vedi .env.example), mai nel codice/log.
• Descriptor UI: ogni Source/Sink espone un descriptor non sensibile (tipo + puntamenti) per la visualizzazione del nodo nel DAG — whitelist esplicita, mai segreti. Contratto e how-to: ../conventions/node-descriptor.md.
• Errori: ExtractError/LoadError/TaskError/ConfigError da src/core/errors.ts. ConfigError.mancanti elenca in forma strutturata tutte le variabili mancanti.
• Variabili d'ambiente: i prefissi base delle istanze sono costanti di src/connectors/shared/prefissi.ts (PREFISSO_PG, PREFISSO_MSSQL, PREFISSO_IMAP…), e il prefisso di un'istanza si ricava con instanceEnvPrefix. Le variabili di un connettore si dichiarano in un manifest e si risolvono con risolviVariabili, che dà per ognuna nome e origine. Gli stessi manifest, con ambienteDellaPipeline, dicono a etl env check quali variabili legge una pipeline. Oggi lo usano i loader di IMAP, SMTP, HTTP, Jira, OAuth2, del client Noeva, Postgres, SQL Server, MySQL, SendGrid e degli storage S3, SFTP e FTP (MANIFEST_IMAP, MANIFEST_SMTP, MANIFEST_HTTP, MANIFEST_JIRA, MANIFEST_OAUTH, MANIFEST_NOEVA, MANIFEST_PG, MANIFEST_MSSQL, MANIFEST_MYSQL, MANIFEST_SENDGRID, MANIFEST_S3, MANIFEST_SFTP, MANIFEST_FTP); gli altri connettori migrano un tipo alla volta. La cartella locale non ha un loader suo: la indica la pipeline, che la dichiara nel proprio manifest. Vedi node-descriptor.md.
• Test: logica pura in unit (CI); I/O reale in tests/integration/ (pnpm run test:integration).