fix(etl): refresh amount validation on current develop

CHANGELOG.md

@@ -17,6 +17,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - ETL request errors now use RFC 9457 `application/problem+json` responses with a stable `errorCode`, fixed type URI, explicit 400/401/404/409/413/422/503/500 taxonomy, and no internal exception text in client responses.
 - ETL requests now enforce bounded UTF-8 payload and record-count limits, prevalidate and transform the complete batch before the first JDBC call, and commit accepted records inside one Spring transaction.
 - ETL transformations now preserve comma/colon-bearing values, use locale-independent text conversion and deterministic `BigDecimal` amount formatting, and retry only transient Spring data-access failures.
+- ETL amount integrity: invalid `AMOUNT` values fail closed before persistence instead of being rewritten to `0.00`, preserving the distinction from genuine zero values.
 - Product branding: user-facing docs and suggested image tags use **mightyETL** (formerly xtrmETL).
   - Legacy Java packages (`com.xtrmetl.*`), Maven `artifactId` `xtrmETL`, and some env/topic defaults remain for compatibility.
   - See `docs/rebrand-name-matrix.md`.

docs/etl/bounded-atomic-batches.md

@@ -2,12 +2,12 @@
 
 ## Purpose
 
-`etl-service` accepts JSON arrays at `POST /api/etl/process`. The service validates and transforms the complete request before the first database write, then executes all accepted writes inside one Spring transaction.
+`etl-service` accepts JSON arrays at `POST /api/etl/process`. The service validates and transforms the complete request before the first `processed_data` target write, then executes all accepted target writes inside one Spring transaction. When an `Idempotency-Key` is supplied, the successful replay-ledger insert shares that transaction with the target writes.
 
 This closes three production risks in the legacy implementation:
 
 - unbounded one-task-per-record fan-out on the JVM common pool;
-- partial database writes when a later record is malformed or rejected;
+- partial target-data writes when a later record is malformed or rejected;
 - delimiter and locale corruption caused by serializing fields to text and splitting them again.
 
 ## Admission contract
@@ -20,9 +20,11 @@ A request is accepted only when all of the following are true:
 4. Every array element is a JSON object.
 5. Every record has a trimmed, non-blank JSON string `id` containing no ISO control, Unicode format-control, Unicode line-separator, or paragraph-separator characters and no more than 256 Unicode code points; numeric JSON identifier types are rejected.
 6. Every record's field names remain unique after locale-independent uppercase normalization.
-7. Every record can be transformed before the first JDBC call.
+7. Every record can be transformed before any target-table write.
 
-Duplicate field names are rejected rather than accepting parser-dependent “first value” or “last value” semantics. Case variants and other field names that normalize to the same output key are also rejected, preventing ambiguous transformed records such as two `ID` or two `NAME` entries. A rejected request performs no database writes.
+Duplicate field names are rejected rather than accepting parser-dependent “first value” or “last value” semantics. Case variants and other field names that normalize to the same output key are also rejected, preventing ambiguous transformed records such as two `ID` or two `NAME` entries. A rejected request performs no `processed_data` target writes and creates no successful idempotency replay record.
+
+Idempotent requests may acquire the request lock and read the replay ledger before batch validation so an already committed request can be replayed without repeating target work. Those control-plane JDBC operations are not target-data writes. A new request with an invalid record still fails before any `processed_data` mutation or successful ledger insert.
 
 ## Transaction contract
 
@@ -32,7 +34,9 @@ After admission and transformation succeed for the full batch, mightyETL inserts
 INSERT INTO processed_data (data) VALUES (?)
 ```
 
-`EtlService.processData` is a Spring `@Transactional` boundary. A runtime database failure in any record rolls back earlier writes from that request. An H2-backed integration test verifies both successful commit and rollback of an earlier insert when a later row violates a target constraint.
+`EtlService.processData` is a Spring `@Transactional` boundary. A runtime database failure in any record rolls back earlier target writes from that request. An H2-backed integration test verifies both successful commit and rollback of an earlier insert when a later row violates a target constraint.
+
+`EtlService.processDataIdempotently` uses the same batch transformation and target-write path. Its transaction-scoped request lock and replay-ledger lookup occur first to preserve replay semantics; for a new request, the target writes and successful ledger insert commit or roll back together.
 
 Only `TransientDataAccessException` failures are retried. Invalid input and deterministic target constraints fail immediately instead of repeating the same batch.
 
@@ -44,10 +48,13 @@ Fields are transformed directly from the Jackson JSON tree; values are not split
 - `NAME` values use locale-independent uppercase conversion.
 - `EMAIL` values use locale-independent lowercase conversion.
 - `AMOUNT` values use `BigDecimal`, `HALF_UP`, scale `2`, and `toPlainString()`.
-- Invalid, excessive-precision, or extreme-scale amounts retain the legacy fallback `0.00` without expanding attacker-controlled exponents into huge strings.
+- Malformed, blank, excessive-precision, or extreme-scale `AMOUNT` values are rejected as invalid records before any target-table write instead of being rewritten to a valid-looking zero.
+- Batch admission remains all-or-nothing: if any record has an invalid `AMOUNT`, the request performs no `processed_data` target writes and creates no successful idempotency replay record.
 - Nested arrays and objects are retained as compact JSON instead of collapsing to empty text.
 - Response lines remain `Processed: <id>` in input order. Identifier whitespace, ISO control, Unicode format-control, Unicode line-separator characters, and length are bounded so one record cannot inject, visually reorder, conceal, or amplify response lines.
 
+Rows created by historic releases that rewrote invalid amounts to zero cannot be distinguished from genuine zero values after the fact without independent source-system evidence. Reconcile affected historical data from an authoritative source before making accounting or compliance decisions.
+
 ## Configuration
 
 Preferred keys use the mightyETL product namespace; compatibility keys remain available:
@@ -73,8 +80,10 @@ Values outside the supported range fail configuration binding instead of silentl
 - Keep the payload limit aligned with gateway and ingress body-size limits. The service-level check occurs after the MVC stack has materialized the request string and is not a substitute for edge enforcement.
 - Keep the record limit below the transaction size that the target database can commit within the request timeout and lock budget.
 - Monitor request latency, transaction duration, rollback rate, database pool wait time, and rejected payload/record-limit errors before raising either limit.
+- Monitor invalid-record rejection rates and reconcile upstream amount-format changes before retrying rejected inputs.
+- Do not log raw amount values or request payloads when diagnosing rejections; use bounded request/error metadata instead.
 - Use descriptive string identifiers. Numeric JSON identifier types are rejected to keep identifier contracts explicit and stable across systems.
 
 ## Remaining boundary
 
-The current API is synchronous and stores a text representation in `processed_data.data`. High-volume ingestion, asynchronous job state, idempotency keys, durable retry queues, and typed target schemas remain separate product milestones; this change does not claim those capabilities.
+The direct `/api/etl/process` path remains synchronous and stores a text representation in `processed_data.data`. Principal-scoped idempotent replay is documented in `docs/etl/idempotent-retries.md`, and durable job intake is documented in `docs/etl/durable-job-intake.md`; neither changes the target-write validation boundary described here. Completed durable worker execution, high-volume ingestion orchestration, and typed target schemas remain separate product milestones.

etl-service/src/main/java/com/xtrmetl/etl/service/EtlService.java

@@ -383,6 +383,12 @@ private String transformRecord(JsonNode record, int index) {
     }
 
     private String transformValue(String key, @Nullable JsonNode valueNode) {
+        if ("AMOUNT".equals(key)) {
+            if (valueNode == null || !(valueNode.isNumber() || valueNode.isTextual())) {
+                throw invalidRecord();
+            }
+            return formatAmount(valueNode.asText());
+        }
         if (valueNode == null || valueNode.isNull()) {
             return "null";
         }
@@ -394,24 +400,24 @@ private String transformValue(String key, @Nullable JsonNode valueNode) {
         return switch (key) {
             case "NAME" -> value.toUpperCase(Locale.ROOT);
             case "EMAIL" -> value.toLowerCase(Locale.ROOT);
-            case "AMOUNT" -> formatAmount(value);
             default -> value;
         };
     }
 
     private String formatAmount(String value) {
+        final BigDecimal amount;
         try {
-            BigDecimal amount = new BigDecimal(value.trim());
-            int scale = amount.scale();
-            if (amount.precision() > MAX_AMOUNT_PRECISION
-                    || scale < -MAX_AMOUNT_ABSOLUTE_SCALE
-                    || scale > MAX_AMOUNT_ABSOLUTE_SCALE) {
-                return "0.00";
-            }
-            return amount.setScale(2, RoundingMode.HALF_UP).toPlainString();
+            amount = new BigDecimal(value.trim());
         } catch (NumberFormatException exception) {
-            return "0.00";
+            throw invalidRecord();
+        }
+        int scale = amount.scale();
+        if (amount.precision() > MAX_AMOUNT_PRECISION
+                || scale < -MAX_AMOUNT_ABSOLUTE_SCALE
+                || scale > MAX_AMOUNT_ABSOLUTE_SCALE) {
+            throw invalidRecord();
         }
+        return amount.setScale(2, RoundingMode.HALF_UP).toPlainString();
     }
 
     private record StoredIdempotencyRecord(String requestDigest, String responseBody) {

etl-service/src/test/java/com/xtrmetl/etl/documentation/EtlBatchDocsAlignmentTest.java

@@ -34,8 +34,8 @@ void runbookStatesAdmissionRollbackAndIngressBoundaries() throws IOException {
         String runbook = read("docs/etl/bounded-atomic-batches.md");
         String normalizedRunbook = runbook.toLowerCase(Locale.ROOT);
 
-        assertTrue(runbook.contains("A rejected request performs no database writes"));
-        assertTrue(runbook.contains("rolls back earlier writes"));
+        assertTrue(runbook.contains("A rejected request performs no `processed_data` target writes"));
+        assertTrue(runbook.contains("rolls back earlier target writes"));
         assertTrue(runbook.contains("not a substitute for edge enforcement"));
         assertTrue(normalizedRunbook.contains("numeric json identifier types are rejected"));
         assertTrue(runbook.contains("no more than 256 Unicode code points"));
@@ -45,6 +45,22 @@ void runbookStatesAdmissionRollbackAndIngressBoundaries() throws IOException {
         assertTrue(runbook.contains("Duplicate field names are rejected"));
     }
 
+    @Test
+    void runbookRejectsInvalidAmountsInsteadOfManufacturingZero() throws IOException {
+        String runbook = read("docs/etl/bounded-atomic-batches.md");
+
+        assertTrue(runbook.contains(
+                "Malformed, blank, excessive-precision, or extreme-scale `AMOUNT` values are rejected"
+        ));
+        assertTrue(runbook.contains("before any target-table write"));
+        assertTrue(runbook.contains(
+                "Idempotent requests may acquire the request lock and read the replay ledger before batch validation"
+        ));
+        assertTrue(runbook.contains("creates no successful idempotency replay record"));
+        assertFalse(runbook.contains("before the first JDBC call"));
+        assertFalse(runbook.contains("legacy fallback `0.00`"));
+    }
+
     @Test
     void configurationAndChangelogUseTheSameLimits() throws IOException {
         String application = read("etl-service/src/main/resources/application.yml");
@@ -56,6 +72,15 @@ void configurationAndChangelogUseTheSameLimits() throws IOException {
         assertTrue(changelog.contains("retry only transient Spring data-access failures"));
     }
 
+    @Test
+    void changelogRecordsFailClosedAmountIntegrity() throws IOException {
+        String changelog = read("CHANGELOG.md");
+
+        assertTrue(changelog.contains("invalid `AMOUNT` values fail closed"));
+        assertTrue(changelog.contains("genuine zero"));
+        assertFalse(changelog.contains("invalid `AMOUNT` values to `0.00`"));
+    }
+
     private static String read(String relativePath) throws IOException {
         return Files.readString(projectRoot().resolve(relativePath), StandardCharsets.UTF_8);
     }

etl-service/src/test/java/com/xtrmetl/etl/service/EtlServiceAmountIntegrityTest.java

@@ -0,0 +1,71 @@
+package com.xtrmetl.etl.service;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.springframework.jdbc.core.JdbcTemplate;
+
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verifyNoInteractions;
+
+/**
+ * Proves that invalid monetary input is rejected rather than converted into a legitimate zero.
+ */
+class EtlServiceAmountIntegrityTest {
+
+    private JdbcTemplate jdbcTemplate;
+    private EtlService etlService;
+
+    @BeforeEach
+    void setUp() {
+        jdbcTemplate = mock(JdbcTemplate.class);
+        EtlBatchProperties properties = new EtlBatchProperties();
+        properties.setMaxPayloadBytes(65_536);
+        properties.setMaxBatchRecords(100);
+        etlService = new EtlService(jdbcTemplate, new ObjectMapper(), properties);
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {
+            "",
+            "not-a-number",
+            "123456789012345678901234567890123456789",
+            "0.0000000000000000001",
+            "1E+19"
+    })
+    void rejectsInvalidOrUnsupportedAmountsBeforeJdbc(String amount) {
+        EtlRequestException exception = assertThrows(
+                EtlRequestException.class,
+                () -> etlService.processData(jsonRecord("record_alpha", amount))
+        );
+
+        assertSame(EtlRequestError.INVALID_RECORD, exception.error());
+        verifyNoInteractions(jdbcTemplate);
+    }
+
+    @Test
+    void invalidAmountMakesTheWholeBatchFailBeforeAnyJdbcWrite() {
+        String payload = """
+                [
+                  {"id":"record_alpha","amount":"10.00"},
+                  {"id":"record_beta","amount":"not-a-number"}
+                ]
+                """;
+
+        EtlRequestException exception = assertThrows(
+                EtlRequestException.class,
+                () -> etlService.processData(payload)
+        );
+
+        assertSame(EtlRequestError.INVALID_RECORD, exception.error());
+        verifyNoInteractions(jdbcTemplate);
+    }
+
+    private static String jsonRecord(String id, String amount) {
+        return "[{\"id\":\"" + id + "\",\"amount\":\"" + amount + "\"}]";
+    }
+}

etl-service/src/test/java/com/xtrmetl/etl/service/EtlServiceAmountJsonTypeBoundaryTest.java

@@ -0,0 +1,57 @@
+package com.xtrmetl.etl.service;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.springframework.jdbc.core.JdbcTemplate;
+
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+
+/**
+ * Exercises the JSON type boundary for monetary input at the real ETL transformation path.
+ */
+class EtlServiceAmountJsonTypeBoundaryTest {
+
+    private JdbcTemplate jdbcTemplate;
+    private EtlService etlService;
+
+    @BeforeEach
+    void setUp() {
+        jdbcTemplate = mock(JdbcTemplate.class);
+        EtlBatchProperties properties = new EtlBatchProperties();
+        properties.setMaxPayloadBytes(65_536);
+        properties.setMaxBatchRecords(100);
+        etlService = new EtlService(jdbcTemplate, new ObjectMapper(), properties);
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {"null", "{}", "[]", "true"})
+    void rejectsNonNumericJsonAmountsBeforeJdbc(String rawAmount) {
+        EtlRequestException exception = assertThrows(
+                EtlRequestException.class,
+                () -> etlService.processData(recordWithRawAmount(rawAmount))
+        );
+
+        assertSame(EtlRequestError.INVALID_RECORD, exception.error());
+        verifyNoInteractions(jdbcTemplate);
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {"0", "\"0\""})
+    void acceptsRealZeroAsNumberOrNumericString(String rawAmount) {
+        etlService.processData(recordWithRawAmount(rawAmount));
+
+        verify(jdbcTemplate).update(anyString(), eq("ID:record_alpha,AMOUNT:0.00,"));
+    }
+
+    private static String recordWithRawAmount(String rawAmount) {
+        return "[{\"id\":\"record_alpha\",\"amount\":" + rawAmount + "}]";
+    }
+}

etl-service/src/test/java/com/xtrmetl/etl/service/EtlServiceBatchSafetyTest.java

@@ -274,15 +274,16 @@ void usesLocaleIndependentTextAndDecimalTransformations() {
     }
 
     @Test
-    void boundsExtremeDecimalInputsWithoutHugePlainStringExpansion() {
+    void rejectsExtremeDecimalInputsWithoutHugePlainStringExpansion() {
         EtlService service = service();
 
-        service.processData("[{\"id\":\"record_alpha\",\"amount\":\"1E+1000000\"}]");
-
-        verify(jdbcTemplate).update(
-                "INSERT INTO processed_data (data) VALUES (?)",
-                "ID:record_alpha,AMOUNT:0.00,"
+        EtlRequestException exception = assertThrows(
+                EtlRequestException.class,
+                () -> service.processData("[{\"id\":\"record_alpha\",\"amount\":\"1E+1000000\"}]")
         );
+
+        assertSame(EtlRequestError.INVALID_RECORD, exception.error());
+        verifyNoInteractions(jdbcTemplate);
     }
 
     @Test

etl-service/src/test/java/com/xtrmetl/etl/service/EtlServiceTest.java

@@ -129,12 +129,16 @@ void formatsAmountsDeterministically(String input, String expected) {
         }
 
         @Test
-        void fallsBackToZeroForInvalidAmount() {
-            etlService.processData(
-                    "[{\"id\":\"record_alpha\",\"amount\":\"not-a-number\"}]"
+        void rejectsInvalidAmountInsteadOfManufacturingZero() {
+            EtlRequestException exception = assertThrows(
+                    EtlRequestException.class,
+                    () -> etlService.processData(
+                            "[{\"id\":\"record_alpha\",\"amount\":\"not-a-number\"}]"
+                    )
             );
 
-            verify(jdbcTemplate).update(anyString(), contains("AMOUNT:0.00"));
+            assertSame(EtlRequestError.INVALID_RECORD, exception.error());
+            verifyNoInteractions(jdbcTemplate);
         }
 
         @Test
@@ -154,19 +158,18 @@ void retainsUnknownAndUnicodeFields() {
         }
 
         @Test
-        void preservesEmptyOptionalValues() {
+        void preservesEmptyOptionalNameAndEmailValues() {
             etlService.processData("""
                     [{
                       "id":"record_alpha",
                       "name":"",
-                      "email":"",
-                      "amount":""
+                      "email":""
                     }]
                     """);
 
             verify(jdbcTemplate).update(
                     anyString(),
-                    eq("ID:record_alpha,NAME:,EMAIL:,AMOUNT:0.00,")
+                    eq("ID:record_alpha,NAME:,EMAIL:,")
             );
         }
     }