Aller au contenu

Histoire ING-004 : Transaction Commit + reject path + SHA-256 dedup + ImportCommitted

Module : ingestion Slice : Tier 1 sub-slice ING-001-c (architecture §11) Side : [BACKEND] Version cible : V0.0.0 Priorité : 4 Dépend de : ING-002 (scheduler + client MinIO + ING_UPLOAD_RECEIVED), ING-003 (parser + validator + phone normalizer), TENANT-001 (tables de base customer + contract + agence par défaut), AUDIT extension catalogue OQ-X10 couvre ING_COMMITTED / ING_PARSE_FAILED / ING_DUPLICATE_FILE / ING_CUSTOMER_* / ING_CONTRACT_* Peut être développé en même temps que : Aucun (ferme le chemin heureux de la v0.0.0) Ordre de fusion : Après ING-002 et ING-003 ; avant ING-005 et ING-006 Complexité estimée : M Histoires d'utilisateurs de la DRP : ING-US-01, ING-US-03 (v0.0.0 cron substitute for upload + auto-commit on clean parse) ; rule V1 ING-R-003 (dedup), R-012 (inter-file dedup key), R-014 (all-or-nothing), R-015/016 (upserts) Wireframe : N/A - backend uniquement

Objectif

Fermer le chemin heureux de la v0.0.0 : remplacer le placeholder "move to processed" de ING-002 par le FSM complet par fichier - déduplication SHA-256 par rapport aux importations précédentes, validation en une seule transaction des insertions customer/contract + ingestion_import_row par ligne, rejet de tout ou rien en cas d'échec de parse avec sidecar .errors.json, l'insertion de fichiers dupliqués avec originating_import_id, plus les émissions d'audit correspondantes et ImportCommitted pour OP. Après cette histoire, le dépôt d'un fichier propre à 02:00 Africa/Abidjan provoque l'atterrissage des données du pipeline de renouvellement.

Backend Scope

Entités

  • ingestion_import (nouveau - introduit ici pour la première fois) : selon l'architecture §2.1 ERD. Champs : id UUIDv7 PK, tenant_id UUID FK, filename varchar(255), file_hash_sha256 char(64), file_mime varchar, file_size_bytes int, file_minio_inbox_key text, file_minio_processed_key text NULL, file_minio_rejected_key text NULL, status enum (PICKED_UP, PARSE_FAILED, COMMITTED, DUPLICATE), row_count_total int NULL, rows_created int, rows_updated int, rows_skipped int, parse_error_detail jsonb NULL, originating_import_id UUID NULL (pour DUPLICATE), picked_up_at timestamptz, finished_at timestamptz NULL.
  • ingestion_import_row (new - first introduced here) : id UUIDv7 PK, import_id UUID FK, tenant_id UUID (dénormalisé par l'architecture §2.2), row_number int, customer_id UUID FK, contract_id UUID FK, action enum (CREATED, UPDATED, SKIPPED), committed_at timestamptz.
  • Colonnes additives sur les tables partagées (selon l'architecture §2.3, coordonnées avec TENANT) : sur customer → source_customer_id varchar + UNIQUE (tenant_id, source_customer_id). On contract → source_contract_id varchar, last_import_id UUID NULL, UNIQUE (tenant_id, customer_id, source_contract_id).

Migrations

backend/src/main/resources/db/migration/tenant/V010__ingestion_create_tables.sql (corps rempli par cette histoire) : 

CREATE TABLE ingestion_import (
  id                       UUID PRIMARY KEY,
  tenant_id                UUID NOT NULL,
  filename                 VARCHAR(255) NOT NULL,
  file_hash_sha256         CHAR(64) NOT NULL,
  file_mime                VARCHAR(120) NOT NULL,
  file_size_bytes          INTEGER NOT NULL,
  file_minio_inbox_key     TEXT NOT NULL,
  file_minio_processed_key TEXT,
  file_minio_rejected_key  TEXT,
  status                   VARCHAR(20) NOT NULL CHECK (status IN ('PICKED_UP','PARSE_FAILED','COMMITTED','DUPLICATE')),
  row_count_total          INTEGER,
  rows_created             INTEGER NOT NULL DEFAULT 0,
  rows_updated             INTEGER NOT NULL DEFAULT 0,
  rows_skipped             INTEGER NOT NULL DEFAULT 0,
  parse_error_detail       JSONB,
  originating_import_id    UUID,
  picked_up_at             TIMESTAMPTZ NOT NULL,
  finished_at              TIMESTAMPTZ
);
CREATE UNIQUE INDEX ux_ingestion_import_tenant_hash ON ingestion_import (tenant_id, file_hash_sha256);
CREATE INDEX ix_ingestion_import_tenant_status_picked ON ingestion_import (tenant_id, status, picked_up_at DESC);

CREATE TABLE ingestion_import_row (
  id           UUID PRIMARY KEY,
  import_id    UUID NOT NULL REFERENCES ingestion_import(id),
  tenant_id    UUID NOT NULL,
  row_number   INTEGER NOT NULL,
  customer_id  UUID NOT NULL,
  contract_id  UUID NOT NULL,
  action       VARCHAR(10) NOT NULL CHECK (action IN ('CREATED','UPDATED','SKIPPED')),
  committed_at TIMESTAMPTZ NOT NULL
);
CREATE INDEX ix_iir_import ON ingestion_import_row (import_id);
CREATE INDEX ix_iir_tenant_customer ON ingestion_import_row (tenant_id, customer_id);

V011__ingestion_customer_contract_columns.sql (corps rempli par cette histoire - ALTERs additifs sur des tables appartenant à TENANT ; propriété confirmée via OQ-X12) : 

ALTER TABLE customer ADD COLUMN source_customer_id VARCHAR(120) NOT NULL;
CREATE UNIQUE INDEX ux_customer_tenant_source ON customer (tenant_id, source_customer_id);

ALTER TABLE contract ADD COLUMN source_contract_id VARCHAR(120) NOT NULL;
ALTER TABLE contract ADD COLUMN last_import_id UUID;
CREATE UNIQUE INDEX ux_contract_tenant_customer_source ON contract (tenant_id, customer_id, source_contract_id);
CREATE INDEX ix_contract_tenant_echeance ON contract (tenant_id, date_echeance);

Couche de services

  • IngestionCommitTransaction (@Transactional) : étant donné un ValidatedFile { importId, tenantId, filename, sha256, mime, size, inboxKey, parseResult } :
  • INSERT ingestion_import avec status=PICKED_UP, picked_up_at=now(). (Cet INSERT se produit dans son propre TX court avant l'analyse - selon l'architecture §5.1 stage 3 et §6.1 ; il active IngestionStartupRecovery dans ING-005)
  • Après l'analyse :
    • Si parseResult.anomalies non vide → call rejectAndAudit(importId, anomalies) outside any open TX (UPDATE → PARSE_FAILED, populate parse_error_detail with {reason_code, anomaly_counts_by_code, sample_row_numbers (first 5)}, set finished_at) ; emit ING_PARSE_FAILED ; write sidecar JSON <inboxKey>.errors.json contenant le même payload ; MinIO COPY+DELETE inbox→/<tid>/rejeté/<yyyy-MM-dd>T<HH-mm-ss>__<filename>.
    • Else → call commitAndAudit(importId, parseResult) in a single business TX : upsert each customer keyed on (tenant_id, source_customer_id) ; upsert each contract keyed on (tenant_id, customer_id, source_contract_id) setting agency_id=<tenant's default agency UUID> (TENANT-001) and last_import_id=importId ; INSERT ingestion_import_row par ligne source avec action CREATED/UPDATED ; UPDATE ingestion_import → status=COMMITTED, rows_created, rows_updated, rows_skipped, row_count_total, finished_at. Dans le même TX, ajouter les lignes de la boîte de sortie pour ING_COMMITTED (unique, avec compteurs) + ING_CUSTOMER_CREATED/UPDATED (par ligne - PiiHasher appliqué à nom et telephone, name_hash + phone_hash dans les détails, pas de texte en clair) + ING_CONTRACT_CREATED/UPDATED (par ligne, {source_contract_id, customer_id, date_echeance, montant_du}) + publier ImportCommitted via spring Modulith-events-jdbc avec @TransactionalEventListener(AFTER_COMMIT) livraison à OP. Après le commit TX : MinIO COPY+DELETE inbox→/<tid>/processed/<yyyy-MM-dd>__<filename>.
  • Sha256DedupChecker.findExisting(tenantId, sha256) → Optional<UUID> : SELECT contre l'UNIQUE (tenant_id, file_hash_sha256). S'il en trouve, le service d'ingestion prend la branche DUPLICATE : INSERT new ingestion_import row with status=DUPLICATE, originating_import_id=<found>, emit ING_DUPLICATE_FILE { originating_import_id, originating_committed_at, file_hash_sha256 }, MinIO COPY+DELETE inbox→/<tid>/duplicate/. Pas d'écriture d'entreprise.
  • ImportCommitted (enregistrement) : { importId, tenantId, rowsCreated, rowsUpdated, rowsSkipped, occurredAt, correlationId } où correlationId == importId (architecture §1.3 ING-ADR-09).
  • IngestionService.processInbox (extends ING-002 skeleton) : wires the FSM pickup → dedup-check → parse → branch(commit/reject/duplicate) → move.

Les endpoints de l'API

Aucun nouveau. (Le profil local /internal/ingestion/tick de ING-002 inchangé)

Règles de validation

  • Toute la validation au niveau des lignes se trouve dans l'ING-003.
  • Cette tranche applique :
  • (tenant_id, file_hash_sha256) UNIQUE au niveau de la base de données (l'INSERT de la branche DUPLICATE ne doit PAS entrer en collision parce qu'il utilise un nouvel identifiant d'importation UUIDv7 ; l'UNIQUE est appliqué via le pré-contrôle de déduplication, pas la nouvelle ligne).
  • (tenant_id, source_customer_id) et (tenant_id, customer_id, source_contract_id) UNIQUE au niveau de la DB → upsert se résout via INSERT ... ON CONFLICT (...) DO UPDATE.
  • Tout ou rien : toute anomalie dans parseResult.anomalies ⇒ zéro écriture métier (V1 ING-R-014).

Considérations multi-tenant

  • Règle critique #1 : chaque SELECT/INSERT/UPDATE/UPSERT contient ici les colonnes WHERE tenant_id = ? ou (tenant_id, ...). L'UUID de l'agence par défaut pour contract.agency_id est résolu par locataire via TenantConfigCache.findById(tenantId).defaultAgencyId().
  • La même DDL sert SHARED (v0.0.0) et BYO (v0.0.4) - le résolveur de source de données achemine le business TX vers la base de données par locataire ; audit + outbox vont toujours vers la base de données de la plateforme (architecture §4.2, D12).
  • Le profil pays (country_code) n'est pas consulté dans cette tranche (pas de paiement / taxe / sortie réglementaire). Le contexte Cron contient déjà le tenant.countryCode de l'ING-002 pour le forward-compat.

Audit & Logging

Selon l'architecture §7.4, tous les événements émis via la boîte de sortie de Spring Modulith (D4 + D7 - l'échec de la boîte de sortie ne doit PAS faire reculer le business commit) : - ING_COMMITTED { rows_created, rows_updated, rows_skipped, row_count_total } - une fois par import commit. - ING_PARSE_FAILED { reason_code, anomaly_counts_by_code, sample_row_numbers (first 5) } - une fois par importation rejetée ; pas de contenu de ligne, pas de PII. - ING_DUPLICATE_FILE { originating_import_id, originating_committed_at, file_hash_sha256 } - une fois par duplicate drop. - ING_CUSTOMER_CREATED / ING_CUSTOMER_UPDATED - par ligne. PiiHasher appliqué à nom et telephone. ING_CUSTOMER_UPDATED porte uniquement changed_fields[] (pas de valeurs anciennes/nouvelles). - ING_CONTRACT_CREATED / ING_CONTRACT_UPDATED - par ligne. montant_du + date_echeance autorisés dans les détails (commerciaux, pas PII). - Toutes les émissions : actor_type=SYSTEM, actor_id=ingestion-cron, correlation_id=import_id (architecture §1.3). - ING_DOB_DRIFT_DETECTED est OUT de la v0.0.0 (V0.0.2).

Portée du frontend

N/A - backend uniquement.

États de l'interface utilisateur (TOUS OBLIGATOIRES)

State Behavior French Copy
Loading N/A — backend only —
Empty N/A — backend only —
Error The .errors.json sidecar is the operator-facing artefact (no UI yet). JSON shape: { "reason_code", "anomaly_counts_by_code", "sample_row_numbers", "file_hash_sha256", "filename", "rejected_at" }. Anomaly code names ARE the French copy (TELEPHONE_INVALIDE, MONTANT_ABSENT, etc.)
Offline / low network N/A — backend only —
Success N/A — backend only —

Comportement réactif

N/A.

Interactions

N/A.

Critères d'acceptation

Architecture §11 ING-001 liant ACs 1-3 :

AC-001: Happy path — clean 5-row CSV commits within 15 s
Given a clean 5-row CSV in `/<tenantA>/inbox/clean-5-rows.csv`
When IngestionService.tick() runs
Then within ≤15 seconds the rows are committed to `customer` + `contract`
  And `ingestion_import.status = COMMITTED`
  And `ING_COMMITTED` is present in `audit_log` with `correlation_id = import_id`
  And the file is in `/<tenantA>/processed/<yyyy-MM-dd>__clean-5-rows.csv`
  And an `ImportCommitted` was published with rowsCreated=5, rowsUpdated=0, rowsSkipped=0.

AC-002: All-or-nothing reject — one bad row aborts the file
Given a 5-row XLSX with one TELEPHONE_INVALIDE row in `/<tenantA>/inbox/`
When IngestionService.tick() runs
Then zero rows are committed to `customer` and `contract`
  And `ingestion_import.status = PARSE_FAILED` with `parse_error_detail` summarising anomaly counts by code + the first 5 row numbers
  And `ING_PARSE_FAILED` is in `audit_log` (no PII in details — name/phone never appear)
  And the file is in `/<tenantA>/rejected/<ts>__<filename>` accompanied by `<filename>.errors.json` sidecar.

AC-003: Duplicate by SHA-256 — no business write, audit + move only
Given a file already committed for tenant A (`ingestion_import` row exists with status=COMMITTED for that sha256)
When the same byte-identical file is re-dropped in `/<tenantA>/inbox/`
Then no INSERT/UPDATE occurs on `customer` or `contract`
  And a new `ingestion_import` row is inserted with `status=DUPLICATE` and `originating_import_id` set to the prior COMMITTED row
  And `ING_DUPLICATE_FILE` is in `audit_log` with `{originating_import_id, originating_committed_at, file_hash_sha256}`
  And the file is in `/<tenantA>/duplicate/<ts>__<filename>`.

Règles de conformité

  • V1 ING-R-003 (SHA-256 par (tenant_id, file_hash_sha256)), R-012 (inter-file dedup key on contract), R-014 (all-or-nothing reject), R-015 (customer upsert key), R-016 (contract upsert key) - tous VÉRIFIÉ.
  • Architecture §7.4 : PiiHasher appliqué à nom/téléphone dans les détails de l'audit ; ne jamais stocker les IIP en clair dans audit_log.details.
  • Architecture §7.6 A2 (l'échec de l'audit n'annule PAS l'engagement de l'entreprise) - découplage de la boîte d'envoi selon D7.
  • Règles critiques #1 (isolation du locataire), #2 (XOF integer), #3 (audit catalog).
  • L'indicateur de fonctionnalité ARTCI est appliqué dans ING-005, pas ici - les émissions d'audit de cette tranche sont toujours exécutées sur dev/test/staging.

Normes et conventions

  • docs/standards/critical-rules.md (#1, #2, #3).
  • docs/standards/database-guidelines.md § upsert pattern (INSERT ... ON CONFLICT ... DO UPDATE), § JSONB columns, § UUIDv7.
  • docs/standards/multi-datasource-patterns.md § business-TX-on-tenant-DB + outbox-on-platform-DB (D12).
  • docs/standards/multi-tenant-model.md § DB profile coverage.
  • docs/standards/java-spring-guidelines.md § @Transactional, @TransactionalEventListener(AFTER_COMMIT).

Exigences de test

Tests unitaires

  • Sha256DedupCheckerTest - match / no-match / cross-tenant non-match.
  • IngestionCommitTransactionTest - compteurs de chemin heureux, sémantique tout ou rien avec des anomalies simulées, clés upsert honorées.

Tests d'intégration (TestContainers, les deux profils DB)

  • IngestionHappyPathIT - AC-001 de bout en bout avec la fixation clean-5-rows.csv ; assert ImportCommitted atteint un stub OP listener (@TransactionalEventListener(AFTER_COMMIT)).
  • IngestionRejectPathIT - AC-002 avec le XLSX contaminé ; assert sidecar JSON shape + absence de PII (regex assertion interdisant tout +225... ou chaîne de nom en majuscule dans la ligne d'audit).
  • IngestionDuplicateIT - AC-003 ; assert new DUPLICATE row + originating link + audit + move.
  • IngestionTenantIsolationIT - même SHA-256 abandonné dans le locataire A et B ; affirmer que les deux COMMIT, pas de pollution croisée (exemple d'architecture §13.3).
  • IngestionAuditOutboxIT - simule une défaillance transitoire de la plateforme-DB sur l'INSERT de l'outbox ; affirme que le commit de l'entreprise réussit toujours (D7).

Ce que l'AQ validera

  • La commande de vérification complète de l'architecture §14.3 (étapes 1-5) réussit de bout en bout.
  • psql ... SELECT status, rows_created, finished_at FROM ingestion_import ORDER BY pick_up_at DESC LIMIT 1 renvoie COMMITTED | 5 | <ts> after dropping clean-5-rows.csv.

Hors du champ d'application

  • Récupération des crashs au démarrage pour les lignes PICKED_UP bloquées + feature-flag prod gate - ING-005.
  • Le pack de tests d'intégration transversaux (gros fichier, régression par octets magiques, etc.) - ING-006.
  • résolution des agences basée sur le "code_agence" - V0.0.2 ING-003.
  • Validation du DOB et ING_DOB_DRIFT_DETECTED - V0.0.2.
  • Atelier (ingestion_staged_row, PATCH /rows/{id}) - V0.0.3.
  • Endpoint de téléchargement de l'interface utilisateur, téléchargement du fichier source, page d'historique - V0.0.3.
  • Parallélisme Cron entre les locataires - V1 (bounded parallelism=1 dans la v0.0.0).
  • Rotation des clés PiiHasher par rangée - V1.

Définition de Fait

  • [Tous les endpoints du backend sont implémentés et renvoient des réponses correctes (aucun ajouté ; endpoint ING-002 inchangé)
  • [Toutes les pages du frontend s'affichent correctement à 360 / 768 / 1024 px (N/A - backend uniquement)
  • [Les 5 états de l'interface utilisateur sont implémentés (N/A - backend uniquement ; la forme JSON du sidecar est documentée)
  • [ ] La micro-copie française correspond exactement aux spécifications de conception (noms des listes d'anomalies = copie française ; verbatim dans le sidecar)
  • [Tous les critères d'acceptation sont passés manuellement
  • [Tests unitaires et d'intégration écrits et réussis
  • [ ] Isolation des locataires vérifiée (tenant_id sur chaque requête ; vert IT inter-locataires ; vert fumée SHARED + BYO)
  • [ ] [Histoires de clients] payload + Slow-3G + resumability (N/A - backend uniquement)
  • [Les entrées du journal d'audit sont présentes pour chaque événement requis (ING_COMMITTED, ING_PARSE_FAILED, ING_DUPLICATE_FILE, ING_CUSTOMER_CREATED/UPDATED, ING_CONTRACT_CREATED/UPDATED - tous vérifiés dans le service informatique)
  • [Pas de défaillances silencieuses sur des erreurs de réseau (MinIO COPY+DELETE retenté au plus une fois par tick ; idempotent via le FSM - architecture §5.1 stages 5a/5b)