node-excel-import

In sintesi

Microservizio di importazione massiva di dati tabellari (Excel xlsx/xls e CSV) verso il database del tenant ditta. Espone una pipeline asincrona — upload → parsing → validazione → trasformazione → arricchimento → persistenza — con tracking dello stato del job (pending → processing → completed/failed) e supporto a lookup table riusabili per mapping codici. Nel grafo dei servizi è la porta d'ingresso "ETL leggero" per popolare risorse PostgreSQL a partire da fogli di calcolo prodotti dagli operatori o da sistemi esterni. Espone eventi NATS al completamento dei job e si registra come modulo orchestrator per workflow inter-servizio. Permette anche il download delle righe rifiutate come file Excel/CSV per la correzione manuale.

Funzionalità principali

• Upload e parsing di file xlsx, xls e csv via multipart/form-data, con selezione del foglio, salto righe header, limite righe e validazione schema tramite Zod
• Preview senza commit: ispezione dei primi N record (max 100) e dei nomi dei fogli, utile al frontend per costruire il mapping colonne prima dell'import effettivo
• Column mapping dichiarativo da colonna sorgente a campo target, con flag required per i campi obbligatori
• Trasformazioni dati componibili in catena (uppercase, lowercase, trim, parse_number, parse_date, format_date, format_currency, round, regex_replace, default_value, required, conditional, lookup, formula) applicate per ordine configurabile
• Lookup table persistite su PostgreSQL con CRUD completo, ricerca exact match e fuzzy match (distanza di Levenshtein con estensione pg_trgm)
• Risoluzione External ID: una colonna del foglio può fungere da chiave naturale; il servizio costruisce in batch il mapping External ID → primary key per decidere automaticamente fra INSERT e UPDATE (upsert)
• Data enrichment opzionale verso provider HTTP esterni configurabili, con throttling a batch per non saturare i servizi remoti
• Job tracking completo: contatori processedRows/totalRows/rowsCreated/rowsUpdated/rowsFailed, errori e timestamp, interrogabili per id, per stato o come elenco dei più recenti
• Auto-migration PostgreSQL idempotente al bootstrap, con pg_advisory_lock per gestire più repliche sullo stesso schema
• Download righe rifiutate via GET /import/jobs/{jobId}/rejected-rows?format=xlsx|csv&disposition=attachment|inline: produce un file con le sole righe scartate, preservando le colonne originali e aggiungendo una colonna _errors con il messaggio di errore (pensato per il flusso "correggi e re-importa")
• Pubblicazione eventi NATS su JetStream al termine del job — subject services.excel-import.job.completed.{tenant}.{jobId} e services.excel-import.job.failed.{tenant}.{jobId} con payload allineato alla convenzione PZeta delle altre flotte event-driven
• Registrazione come modulo orchestrator (@pzeta/orchestrator-node) con due task NATS request/reply: start-import-job (parzialmente integrato: lo scaricamento del file da node-storage via idallegato non è ancora completato e il task restituisce un errore esplicito) e get-import-status (operativo, espone lo stato corrente del job)

Architettura

Stack: Fastify v5 · Inversify (DI con constructor injection e decoratori @injectable()/@inject()) · @fastify/multipart per gli upload · exceljs per parsing Excel · zod per la validazione di request e righe importate · date-fns per le trasformazioni temporali · pg (driver nativo) verso PostgreSQL · @pzeta/fastify-utils (loggingPlugin, errorHandlerPlugin, healthcheckPlugin, metricsPlugin, openapiPlugin, securityPlugin con rate limit) · @pzeta/log per logging strutturato e correlation ID · @pzeta/orchestrator-node per la registrazione del modulo presso node-orchestrator · nats.js (driver ufficiale) per JetStream pub/sub.

Layout DDD (src/):

Pattern adottati: Facade (ImportService su ImportOrchestrator + ImportJobService), Repository con port-and-adapter sul layer dominio, Chain of Responsibility per le trasformazioni, elaborazione asincrona del job in background con aggiornamento incrementale dello stato, schema importexport autocreato e versionato tramite file 001–005 in migrations/, Publisher sul port IJobEventPublisher con implementazione swappabile (Noop → JetStream in produzione, InMemory nei test) e modulo orchestrator NATS-first che riusa l'ImportService come backend dei task handlers.

Autenticazione: il servizio si affida a nginx (auth_request → node-user-auth) e non implementa identity in proprio. Il rate limit pubblico (createRateLimitConfig.forPublicApi, finestra di 15 minuti) escludelusivamente /health e /ready. CORS è guidato da CORS_ORIGINS (whitelist).

Casi d'uso

• Caricamento anagrafiche batch: l'operatore carica un .xlsx di clienti/fornitori; il frontend chiama POST /import/preview per mostrare le prime righe e configurare il mapping colonne, poi POST /import/upload con columnMappings e enableTransformation: true per normalizzare maiuscole, codice fiscale e date
• Aggiornamento massivo via External ID: il foglio contiene una colonna codiceesterno che mappa a record già esistenti — l'orchestrator costruisce in batch la corrispondenza External ID → ID interno e produce un mix INSERT/UPDATE invece di duplicare le righe
• Normalizzazione tramite lookup: una colonna "tipo pagamento" testuale viene risolta contro una lookup table (POST /lookup/tables/{name}/lookup/{key}) per ottenere il codice interno; eventuali valori non noti vengono individuati via fuzzy match e segnalati nei warning del job
• Arricchimento da provider esterno: per ogni riga il DataEnrichmentService chiama un endpoint HTTP configurato (es. validazione partita IVA, geocoding indirizzi) a batch, con pausa configurabile tra i batch per rispettare i rate limit del provider
• Monitoraggio import lunghi: il frontend polla GET /import/jobs/{jobId} per mostrare la progress bar (processedRows/totalRows) e gli aggregati finali (rowsCreated/rowsUpdated/rowsFailed) a fine job
• Workflow orchestrator event-driven: node-scheduler invoca un workflow che chiama il task start-import-job, attende l'evento services.excel-import.job.completed.{tenant}.> su JetStream, e propaga il risultato (numero righe importate, durata, eventuale idallegato con le righe rifiutate) a node-notification per produrre una notifica all'operatore
• Correzione manuale righe rifiutate: a fine job l'operatore scarica /import/jobs/{jobId}/rejected-rows?format=xlsx ottenendo un foglio con le sole righe scartate e una colonna _errors con il dettaglio del motivo; le corregge in Excel, salva e re-importa il file ripulito tramite POST /import/upload senza dover ricaricare l'intero dataset

Identità & esposizione

Endpoint operazionali

Endpoint convenzionali esposti da tutti i microservizi PZeta basati su @pzeta/fastify-utils:

Configurazione

Variabili d'ambiente rilevanti per chi integra il servizio (per la lista completa vedi .env.default del repo):

Il moduleId esposto all'orchestrator coincide con il name del package.json (node-excel-import) e non è configurabile tramite env: per pubblicare il modulo con un identificativo diverso occorre modificare il package o il blocco di bootstrap in src/index.ts.

Lo schema PostgreSQL importexport viene creato e versionato automaticamente al bootstrap dai file in migrations/ (sequenza 001–005); l'esecuzione è protetta da pg_advisory_lock per consentire l'avvio simultaneo di più repliche.

Note eventing NATS

Il servizio non è più HTTP-only: quando ORCHESTRATOR_ENABLED=true il processo apre una connessione JetStream verso NATS_URL, garantisce l'esistenza dello stream NATS_STREAM_NAME (subject filter services.excel-import.job.>, retention Limits, storage File, max_age derivato da NATS_STREAM_MAX_AGE_DAYS) e si registra come modulo presso node-orchestrator tramite ModuleSDK di @pzeta/orchestrator-node. Con ORCHESTRATOR_ENABLED=false (default attuale) il publisher viene sostituito da un NoopJobEventPublisher e il binding all'orchestrator non viene attivato, mantenendo il comportamento HTTP-only legacy.

Eventi pubblicati

Al termine di ogni job (ImportOrchestrator) il servizio pubblica un evento idempotente su JetStream:

La convenzione di subject (services.excel-import.job come prefisso, segmenti completed/failed, {tenant} e {jobId} come ultimi token) è allineata a quanto consumato dal NATSInboundGateway di node-notification: di conseguenza è sufficiente sottoscrivere services.excel-import.job.completed.> o services.excel-import.job.failed.> (con filtro per tenant se necessario) per reagire al ciclo di vita degli import.

Task orchestrator esposti

Il modulo registra due task handler request/reply sul bus NATS dell'orchestrator (definiti in OrchestratorModule.ts):

Convenzione lato scanner

Il file nats.json esposto dallo scanner dei microservizi rileva la costante services.excel-import.job definita in JetStreamJobEventPublisher.ts come subject prefix del servizio, riconoscendo i due subject *.completed.{tenant}.{jobId} e *.failed.{tenant}.{jobId} come canali outbound. Il binding di consumo (chi reagisce agli eventi) non è di competenza di questo servizio, ma del sottoscrittore (node-notification, node-scheduler, futuri consumer).

Dipendenze e dipendenti

Dipende da (servizi che questo servizio chiama):

• node-user-auth

Consumato da (chi chiama questo servizio):

• frontend Vue

Infrastruttura (PostgreSQL, NATS, Redis, MinIO) non è elencata qui — vedi sezione Architettura del singolo servizio.