Aller au contenu

Story ING-004: Commit transaction + reject path + SHA-256 dedup + ImportCommitted

Module: ingestion Slice: Tier 1 sub-slice ING-001-c (architecture §11) Side: [BACKEND] Version target: V0.0.0 Priority: 4 Depends on: ING-002 (scheduler + MinIO client + ING_UPLOAD_RECEIVED), ING-003 (parser + validator + phone normalizer), TENANT-001 (customer + contract base tables + default agency), AUDIT catalog extension OQ-X10 covers ING_COMMITTED / ING_PARSE_FAILED / ING_DUPLICATE_FILE / ING_CUSTOMER_* / ING_CONTRACT_* Can develop concurrently with: None (closes the v0.0.0 happy-path) Merge order: After ING-002 and ING-003; before ING-005 and ING-006 Estimated complexity: M PRD User Stories: 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 only

Objective

Close the v0.0.0 happy path: replace ING-002's placeholder "move to processed" with the full per-file FSM — SHA-256 dedup against prior imports, single-transaction commit of customer/contract upserts + ingestion_import_row per line, parse-failure all-or-nothing reject with sidecar .errors.json, duplicate-file insertion with originating_import_id, plus the corresponding audit emissions and ImportCommitted for OP. After this story, dropping a clean file at 02:00 Africa/Abidjan causes the renewal pipeline's data to land.

Backend Scope

Entities

  • ingestion_import (new — first introduced here): per architecture §2.1 ERD. Fields: 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 (for 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 (denormalized per architecture §2.2), row_number int, customer_id UUID FK, contract_id UUID FK, action enum (CREATED, UPDATED, SKIPPED), committed_at timestamptz.
  • Additive columns on shared tables (per architecture §2.3, coordinated with TENANT): on 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 (body filled by this story): 

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 (body filled by this story — additive ALTERs on TENANT-owned tables; ownership confirmed 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);

Service Layer

  • IngestionCommitTransaction (@Transactional): given a ValidatedFile { importId, tenantId, filename, sha256, mime, size, inboxKey, parseResult }:
  • INSERT ingestion_import with status=PICKED_UP, picked_up_at=now(). (This INSERT happens in its own short TX before parse — per architecture §5.1 stage 3 and §6.1; it enables IngestionStartupRecovery in ING-005.)
  • After parse:
    • If parseResult.anomalies non-empty → 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 containing the same payload; MinIO COPY+DELETE inbox→/<tid>/rejected/<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 per source row with action CREATED/UPDATED; UPDATE ingestion_import → status=COMMITTED, rows_created, rows_updated, rows_skipped, row_count_total, finished_at. Inside the same TX, append outbox rows for ING_COMMITTED (single, with counters) + ING_CUSTOMER_CREATED/UPDATED (per row — PiiHasher applied to nom and telephone, name_hash + phone_hash in details, no plaintext) + ING_CONTRACT_CREATED/UPDATED (per row, {source_contract_id, customer_id, date_echeance, montant_du}) + publish ImportCommitted via spring-modulith-events-jdbc with @TransactionalEventListener(AFTER_COMMIT) delivery to OP. After TX commit: MinIO COPY+DELETE inbox→/<tid>/processed/<yyyy-MM-dd>__<filename>.
  • Sha256DedupChecker.findExisting(tenantId, sha256) → Optional<UUID>: SELECT against the UNIQUE (tenant_id, file_hash_sha256). If found, the IngestionService takes the DUPLICATE branch: 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/. No business write.
  • ImportCommitted (record): { importId, tenantId, rowsCreated, rowsUpdated, rowsSkipped, occurredAt, correlationId } where 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.

API Endpoints

None new. (The local-profile /internal/ingestion/tick from ING-002 unchanged.)

Validation Rules

  • All row-level validation lives in ING-003.
  • This slice enforces:
  • (tenant_id, file_hash_sha256) UNIQUE at the DB level (DUPLICATE branch's INSERT must NOT collide because it uses a new UUIDv7 import id; the UNIQUE is enforced via the dedup pre-check, not the new row).
  • (tenant_id, source_customer_id) and (tenant_id, customer_id, source_contract_id) UNIQUE at the DB level → upsert resolves via INSERT ... ON CONFLICT (...) DO UPDATE.
  • All-or-nothing: any anomaly in parseResult.anomalies ⇒ zero business writes (V1 ING-R-014).

Multi-Tenant Considerations

  • Critical rule #1: every SELECT/INSERT/UPDATE/UPSERT here carries WHERE tenant_id = ? or (tenant_id, …) columns. The default-agency UUID for contract.agency_id is resolved per-tenant via TenantConfigCache.findById(tenantId).defaultAgencyId().
  • Same DDL serves SHARED (v0.0.0) and BYO (v0.0.4) — datasource resolver routes the business TX to the per-tenant DB; audit + outbox always go to the platform DB (architecture §4.2, D12).
  • Country-profile (country_code) is not consulted in this slice (no payment / tax / regulatory output). Cron context already carries tenant.countryCode from ING-002 for forward-compat.

Audit & Logging

Per architecture §7.4, all events emitted via spring-modulith outbox (D4 + D7 — outbox failure must NOT roll back the business commit): - ING_COMMITTED { rows_created, rows_updated, rows_skipped, row_count_total } — once per committed import. - ING_PARSE_FAILED { reason_code, anomaly_counts_by_code, sample_row_numbers (first 5) } — once per rejected import; no row content, no PII. - ING_DUPLICATE_FILE { originating_import_id, originating_committed_at, file_hash_sha256 } — once per duplicate drop. - ING_CUSTOMER_CREATED / ING_CUSTOMER_UPDATED — per row. PiiHasher applied to nom and telephone. ING_CUSTOMER_UPDATED carries changed_fields[] only (no old/new values). - ING_CONTRACT_CREATED / ING_CONTRACT_UPDATED — per row. montant_du + date_echeance allowed in details (commercial, not PII). - All emissions: actor_type=SYSTEM, actor_id=ingestion-cron, correlation_id=import_id (architecture §1.3). - ING_DOB_DRIFT_DETECTED is OUT of v0.0.0 (V0.0.2).

Frontend Scope

N/A — backend only.

UI States (ALL REQUIRED)

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 —

Responsive Behavior

N/A.

Interactions

N/A.

Acceptance Criteria

Architecture §11 ING-001 binding 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>`.

Compliance Rules

  • V1 ING-R-003 (SHA-256 per (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) — all VÉRIFIÉ.
  • Architecture §7.4: PiiHasher applied to nom/telephone in audit details; never store plaintext PII in audit_log.details.
  • Architecture §7.6 A2 (audit failure does NOT roll back business commit) — outbox decoupling per D7.
  • Critical rule #1 (tenant isolation), #2 (XOF integer), #3 (audit catalog).
  • ARTCI feature flag is enforced in ING-005, not here — this slice's audit emissions still run on dev/test/staging.

Standards & 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).

Testing Requirements

Unit Tests

  • Sha256DedupCheckerTest — match / no-match / cross-tenant non-match.
  • IngestionCommitTransactionTest — happy-path counters, all-or-nothing semantics with mocked anomalies, upsert keys honored.

Integration Tests (TestContainers, both DB profiles)

  • IngestionHappyPathIT — AC-001 end-to-end with the clean-5-rows.csv fixture; assert ImportCommitted reaches a stub OP listener (@TransactionalEventListener(AFTER_COMMIT)).
  • IngestionRejectPathIT — AC-002 with the tainted XLSX; assert sidecar JSON shape + absence of PII (regex assertion forbidding any +225… or capitalized name strings in the audit row).
  • IngestionDuplicateIT — AC-003; assert new DUPLICATE row + originating link + audit + move.
  • IngestionTenantIsolationIT — same SHA-256 dropped in tenant A and B; assert both COMMIT, no cross-pollution (architecture §13.3 sample).
  • IngestionAuditOutboxIT — simulate a transient platform-DB failure on outbox INSERT; assert business commit still succeeds (D7).

What QA Will Validate

  • Full smoke verification command from architecture §14.3 (steps 1–5) passes end-to-end.
  • psql … SELECT status, rows_created, finished_at FROM ingestion_import ORDER BY picked_up_at DESC LIMIT 1 returns COMMITTED | 5 | <ts> after dropping clean-5-rows.csv.

Out of Scope

  • Startup crash recovery for stuck PICKED_UP rows + feature-flag prod gate — ING-005.
  • The cross-cutting integration test pack (large-file, magic-byte regression, etc.) — ING-006.
  • code_agence-based agency resolution — V0.0.2 ING-003.
  • DOB-required validation and ING_DOB_DRIFT_DETECTED — V0.0.2.
  • Atelier (ingestion_staged_row, PATCH /rows/{id}) — V0.0.3.
  • UI upload endpoint, source-file download, history page — V0.0.3.
  • Cron parallelism across tenants — V1 (bounded parallelism=1 in v0.0.0).
  • Per-row PiiHasher key rotation — V1.

Definition of Done

  • [ ] All backend endpoints implemented and returning correct responses (none added; ING-002 endpoint unchanged)
  • [ ] All frontend pages render correctly at 360 / 768 / 1024 px (N/A — backend only)
  • [ ] All 5 UI states implemented (N/A — backend only; sidecar JSON shape documented)
  • [ ] French micro-copy matches design spec exactly (anomaly enum names = French copy; verbatim in sidecar)
  • [ ] All acceptance criteria pass manually
  • [ ] Unit + integration tests written and passing
  • [ ] Tenant isolation verified (tenant_id on every query; cross-tenant IT green; SHARED + BYO smoke green)
  • [ ] [CUSTOMER stories] payload + Slow-3G + resumability (N/A — backend only)
  • [ ] Audit log entries present for every required event (ING_COMMITTED, ING_PARSE_FAILED, ING_DUPLICATE_FILE, ING_CUSTOMER_CREATED/UPDATED, ING_CONTRACT_CREATED/UPDATED — all verified in IT)
  • [ ] No silent failures on network errors (MinIO COPY+DELETE retried at most once per tick; idempotent via the FSM — architecture §5.1 stages 5a/5b)