Connettore: IMAP (caselle di posta)

Legge i messaggi di una cartella IMAP e agisce su di loro: aggiunge o toglie flag, li sposta in un'altra cartella. Basato su imapflow e mailparser. Codice in src/connectors/imap/. Nasce per portare su ETL procedure oggi affidate a un client di posta.

Funzione	Tipo	Cosa fa
`imapSource`	`Source<MessaggioImap>`	I messaggi di una cartella, con allegati ed EML grezzo
`imapSessione`	—	La connessione condivisa dalle action, aperta al primo uso
`imapFlagAction`	`Action<I, I>`	Aggiunge o toglie flag e keyword, e passa avanti il record
`imapSpostaAction`	`Action<I, O>`	Sposta in una cartella (creata se manca), con l'esito in uscita
`oggettoContiene`	`(oggetto, testo) => boolean`	Il confronto dei filtri sull'oggetto, per chi scrive un'action propria

La cartella è la coda: la sorgente non marca e non sposta nulla, lo fanno le action a valle. Un messaggio spostato, o marcato con la keyword del filtro, non si rilegge.

Configurazione

loadImapConfig(prefix?, env?) legge le variabili con prefisso (default IMAP_); loadImapRegistry(env?) le risolve per istanza: con IMAP_INSTANCES=ordini,fatture ogni casella ha le sue IMAP_<ID>_*, senza lista valgono le variabili base come istanza default.

Variabile	Default	Note
`IMAP_HOST`	—	obbligatoria
`IMAP_USER`	—	obbligatoria
`IMAP_SECURE`	`true`	TLS implicito; `false` usa STARTTLS se il server lo offre
`IMAP_PORT`	993 con TLS, 143 senza
`IMAP_PASSWORD`	—	una fra password e token, non entrambe
`IMAP_ACCESS_TOKEN`	—	OAuth2 (XOAUTH2), per esempio Microsoft 365
`IMAP_AUTH`	dedotto	`password` / `oauth2`; `oauth2` senza token vuole la funzione
`IMAP_OAUTH_INSTANCE`	`default`	quale identità `OAUTH_*` rinnova il token

OAuth2. Né Microsoft 365 né Gmail accettano più la password dell'account via IMAP (su Gmail resta la password per le app, che però è legata a un utente e alla sua verifica in due passaggi). Il token scade in un'ora: rinnovarlo non è compito del connettore, che riceve una funzione — client: { accessToken: () => ottieniToken() } — con la precedenza sul token dell'env. Con IMAP_AUTH=oauth2 il token nell'env non serve; senza token né funzione, la connessione non parte (ConfigError).
Quella funzione la costruisce il modulo oauth: fornitoriOauth(loadOauthRegistry(env)) risolve l'identità indicata da IMAP_OAUTH_INSTANCE e scambia il refresh token quando serve. Procedura completa nel runbook oauth-google-posta.md.
Niente credenziali nei log né nel DAG. Il logger di imapflow è spento (logga i comandi, e al login le credenziali); i descriptor mostrano host, porta e cartelle, mai utente e password.

Source — `imapSource`

const config = loadImapRegistry().get("ordini");

imapSource(config, {
  instanceId: "ordini", // mostrato nel DAG
  cartella: "INBOX", // default
  filtro: { nonLetti: true, senzaKeyword: "NoevaFatto" }, // facoltativo
  maxMessages: 50, // facoltativo, i primi per UID
});

Il filtro combina, in AND:

Campo	Che cosa tiene
`nonLetti`	solo i messaggi senza `\Seen`
`senzaKeyword`	solo i messaggi senza quella keyword: quelli già lavorati, se la pipeline li marca
`oggettoContiene`	solo i messaggi con quel testo nell'oggetto, maiuscole ignorate
`oggettoNonContiene`	esclude i messaggi con quel testo nell'oggetto, maiuscole ignorate

I due filtri sull'oggetto si decidono sull'oggetto decodificato, dopo la lettura: oggettoContiene restringe prima la ricerca sul server (SEARCH SUBJECT), ma Gmail cerca per parole e ammetterebbe un «test» e un «noeva» sparsi; oggettoNonContiene non passa dal server per la stessa ragione. Con un filtro sull'oggetto maxMessages conta i messaggi emessi, non quelli scartati. Un testo vuoto è un ConfigError: ammetterebbe o escluderebbe tutto.

Un record per messaggio (MessaggioImap):

Campo	Note
`cartella`, `uid`, `uidValidity`	la chiave naturale; `uidValidity` è una stringa (è un `bigint`)
`messageId`	con le parentesi angolari, o `null`
`from`, `to`, `cc`	`{ name, address }`
`subject`, `date`	`date` in ISO 8601
`flags`	flag e keyword al momento della lettura
`text`, `html`	il corpo; i `cid:` dell'HTML restano come sono
`attachments`	`filename`, `contentType`, `size`, `content`, `contentId`, `inline`
`eml`	il messaggio grezzo, per archiviarlo così com'è
`error`	`null`, o perché il messaggio non si è potuto interpretare

Una connessione sua, separata da quella delle action. Blocca la cartella in sola lettura, cerca gli UID all'inizio, scarica un messaggio alla volta e lo interpreta prima di emetterlo: in memoria c'è un messaggio per volta, allegati compresi. La connessione si chiude a fine lettura, anche su errore o abort.
Mai i messaggi \Deleted: sono in attesa di EXPUNGE, quindi fuori dalla coda.
Un messaggio illeggibile resta un record, con error valorizzato e l'EML grezzo quando c'è: la sorgente non ha dead-letter, e la pipeline decide cosa farne (per esempio spostarlo in revisione). Un messaggio sparito fra la ricerca e la lettura si salta.
Errori di rete, TLS, credenziali o cartella inesistente: ExtractError leggibile, senza credenziali, che ferma la run.

Action — `imapSessione`, `imapFlagAction`, `imapSpostaAction`

const sessione = imapSessione(config, { instanceId: "ordini" });
try {
  await graph("posta")
    .source("mail", imapSource(config, { instanceId: "ordini" }))
    // … elaborazione …
    .action(
      "verifica",
      "marca",
      imapFlagAction<Verificata>(sessione, {
        messaggio: (r) => r.mail, // RiferimentoMessaggio: un MessaggioImap va bene
        aggiungi: ["NoevaFatto"],
        togli: ["\\Seen"],
      }),
      { deadLetter },
    )
    .action(
      "marca",
      "archivia",
      imapSpostaAction<Verificata, Archiviata>(sessione, {
        messaggio: (r) => r.mail,
        destinazione: (r) => (r.esito === "AVANTI" ? "Done" : "Review"),
        merge: (r, esito) => ({ ...r, archiviata: esito }),
      }),
      { deadLetter },
    )
    .sink("archivia", report)
    .build()
    .run(ctx);
} finally {
  await sessione.close(); // chi compone la pipeline chiude la sessione
}

Una connessione condivisa dalle action della sessione, aperta al primo uso e riaperta se è caduta. Le action girano una alla volta per record e ogni operazione prende e rilascia il lock della sua cartella: operazioni di action diverse si mettono in fila. Il grafo non la chiude: va chiusa a fine run. Più operazioni concorrenti su una connessione caduta ne riaprono una sola. sessione.usa(stage, operazione, corpo) e sessione.descriptor(details) servono a chi scrive un'action IMAP propria.
UIDVALIDITY verificata. Se la cartella è stata ricreata, gli UID letti non indicano più gli stessi messaggi: è un errore del record, e non si tocca nulla.
imapFlagAction aggiunge e toglie flag di sistema (\Seen, \Flagged) e keyword; è idempotente. Un messaggio che non c'è più è un errore del record.
imapSpostaAction crea la destinazione se manca (una volta per action), poi sposta: MOVE, o COPY + \Deleted + EXPUNGE sui server che non lo hanno. L'esito (EsitoSpostamento) dice la cartella, il nuovo UID se il server lo comunica, e giaSpostato.
Spostamento idempotente. Se l'UID non è più nella cartella di partenza, o è \Deleted (un COPY + \Deleted interrotto prima dell'EXPUNGE), si cerca il Message-ID nella destinazione: se c'è, l'esito è giaSpostato: true (un retry dopo uno spostamento riuscito); se no, o se il messaggio non ha Message-ID, è un errore del record. Il confronto è sul Message-ID identico, verificato sull'envelope: SEARCH HEADER cerca per sottostringa. ⚠️ Due mail diverse con lo stesso Message-ID non si distinguono: se una è già nella destinazione, anche l'altra risulta giaSpostato.
Errori: LoadError leggibili, con lo stage del nodo, quindi dead-letter o ramo catch. Un ConfigError (token OAuth2 mancante) passa com'è.
Una connessione fallita non resta in memoria: il record successivo riprova.

Descriptor (UI del DAG)

Kind imap per sorgente e action, con connection = instanceId se passato. Dettagli: host e porta, ognuno con la sua variabile e la sua origine (IMAP_HOST · env, IMAP_PORT · default), poi, per nodo, cartella / filtro / maxMessages (sorgente), aggiunge / toglie (flag), destinazione (spostamento, (dal record) se è una funzione). Il filtro elenca i criteri attivi: non letti, senza <keyword>, oggetto con "…", oggetto senza "…". In fondo c'è credenziale: la variante in uso (password · impostata, access token · impostato, OAuth · identità default) o, se manca, le alternative ammesse. Nel piano, con OAuth2, la riga dice anche se l'identità OAUTH_* è completa (· impostata) o quali sue variabili mancano; un IMAP_USER mancante, che non ha una riga sua, lo dice la stessa riga (… · manca IMAP_USER). Mai utente né segreti. Contratto generale: node-descriptor.md.

Le variabili sono dichiarate in MANIFEST_IMAP (config.ts): loadImapConfig le risolve, e con variabili mancanti il ConfigError le elenca tutte in mancanti. Per il piano, istanzaImapPerIlPiano(id, env) dà la casella risolta o l'esempio con le variabili della casella vera, e loadImapInstance(id, env) carica la sola casella id, senza le altre del registry. Per etl env check, istanzaImapDichiarata(id) è la casella da passare ad ambienteDellaPipeline nell'environment della pipeline.

Test

Unit colocati in src/connectors/imap/*.test.ts, con un server IMAP in memoria (ServerImapFinto in src/testing/imap-finto.ts, esportato da @noeva/etl-core/testing): config, traduzione degli errori, parsing di EML sintetici, sorgente (filtri, maxMessages, chiusura su errore e abort, log senza contenuti), action (UIDVALIDITY, già spostato, \Deleted, Message-ID solo identico, cartella creata, riconnessione concorrente) e un grafo sorgente → flag → sposta con dead-letter.
Integrazione in tests/integration/imap.integration.test.ts contro GreenMail (docker compose up -d imap-test, utente etltest / etltest, IMAP su 3143): lettura con allegati inline e non, keyword aggiunta e tolta, spostamento in una cartella nuova e già spostato, credenziali errate. Si salta senza IMAP_HOST. Il fallback COPY + \Deleted + EXPUNGE è di imapflow e non è coperto: GreenMail supporta MOVE.

Connettore: IMAP (caselle di posta) ​

Configurazione ​

Source — imapSource ​

Action — imapSessione, imapFlagAction, imapSpostaAction ​