fix(cdc): snapshot canonical change record maps

cdc-service/src/main/java/com/xtrmetl/cdc/spi/CanonicalChangeRecord.java

@@ -1,15 +1,20 @@
 package com.xtrmetl.cdc.spi;
 
 import java.util.Collections;
+import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Objects;
 
 /**
- * Product-neutral change event for any-to-any routing (source → targets).
+ * Immutable product-neutral change event for any-to-any CDC routing.
  *
- * <p>Debezium envelopes are adapted via {@link DebeziumChangeRecordMapper}.
- * Not yet wired into the live Kafka publish path — canonical form is for
- * upcoming multi-target routing.</p>
+ * <p>The constructor takes shallow snapshots of the supplied row maps so later
+ * caller mutations cannot change this record's contents, equality, or hash
+ * identity. Null map references become empty maps, while null database values
+ * inside a supplied map are preserved. Debezium envelopes are adapted via
+ * {@link DebeziumChangeRecordMapper}. The canonical form is not yet wired into
+ * the live Kafka publication path and remains the planned multi-target routing
+ * value object.</p>
  */
 public final class CanonicalChangeRecord {
 
@@ -22,6 +27,18 @@ public final class CanonicalChangeRecord {
     private final Map<String, Object> after;
     private final Map<String, Object> pk;
 
+    /**
+     * Creates an immutable canonical CDC record from scalar metadata and row snapshots.
+     *
+     * @param sourceId stable source connector identifier
+     * @param op Debezium-style operation code
+     * @param schema source database schema, when available
+     * @param table source table name, when available
+     * @param tsEpochMs source event timestamp in epoch milliseconds
+     * @param before row values before the change, or {@code null} when unavailable
+     * @param after row values after the change, or {@code null} when unavailable
+     * @param pk primary-key values, or {@code null} when unavailable
+     */
     public CanonicalChangeRecord(
             String sourceId,
             String op,
@@ -37,39 +54,86 @@ public CanonicalChangeRecord(
         this.schema = schema;
         this.table = table;
         this.tsEpochMs = tsEpochMs;
-        this.before = before == null ? Map.of() : Collections.unmodifiableMap(before);
-        this.after = after == null ? Map.of() : Collections.unmodifiableMap(after);
-        this.pk = pk == null ? Map.of() : Collections.unmodifiableMap(pk);
+        this.before = snapshot(before);
+        this.after = snapshot(after);
+        this.pk = snapshot(pk);
     }
 
+    private static Map<String, Object> snapshot(Map<String, Object> source) {
+        if (source == null || source.isEmpty()) {
+            return Map.of();
+        }
+        return Collections.unmodifiableMap(new LinkedHashMap<>(source));
+    }
+
+    /**
+     * Returns the stable source connector identifier.
+     *
+     * @return source connector identifier
+     */
     public String getSourceId() {
         return sourceId;
     }
 
+    /**
+     * Returns the source operation code.
+     *
+     * @return operation code
+     */
     public String getOp() {
         return op;
     }
 
+    /**
+     * Returns the source database schema, when available.
+     *
+     * @return source schema
+     */
     public String getSchema() {
         return schema;
     }
 
+    /**
+     * Returns the source table, when available.
+     *
+     * @return source table
+     */
     public String getTable() {
         return table;
     }
 
+    /**
+     * Returns the source event timestamp in epoch milliseconds.
+     *
+     * @return source event timestamp
+     */
     public long getTsEpochMs() {
         return tsEpochMs;
     }
 
+    /**
+     * Returns the immutable construction-time snapshot of values before the change.
+     *
+     * @return immutable before-values map
+     */
     public Map<String, Object> getBefore() {
         return before;
     }
 
+    /**
+     * Returns the immutable construction-time snapshot of values after the change.
+     *
+     * @return immutable after-values map
+     */
     public Map<String, Object> getAfter() {
         return after;
     }
 
+    /**
+     * Returns the immutable construction-time snapshot of primary-key values.
+     *
+     * @return immutable primary-key map
+     */
     public Map<String, Object> getPk() {
         return pk;
     }

cdc-service/src/test/java/com/xtrmetl/cdc/spi/CanonicalChangeRecordTest.java

@@ -0,0 +1,86 @@
+package com.xtrmetl.cdc.spi;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class CanonicalChangeRecordTest {
+
+    @Test
+    void snapshotsMutableInputMaps() {
+        Map<String, Object> before = new LinkedHashMap<>();
+        before.put("status", "old");
+        Map<String, Object> after = new LinkedHashMap<>();
+        after.put("status", "new");
+        Map<String, Object> pk = new LinkedHashMap<>();
+        pk.put("id", 7L);
+
+        CanonicalChangeRecord record = new CanonicalChangeRecord(
+                "postgres-debezium",
+                "u",
+                "public",
+                "orders",
+                123L,
+                before,
+                after,
+                pk
+        );
+        int originalHashCode = record.hashCode();
+
+        before.put("status", "mutated");
+        after.clear();
+        pk.put("id", 99L);
+
+        assertEquals("old", record.getBefore().get("status"));
+        assertEquals("new", record.getAfter().get("status"));
+        assertEquals(7L, record.getPk().get("id"));
+        assertEquals(originalHashCode, record.hashCode());
+    }
+
+    @Test
+    void exposesUnmodifiableMaps() {
+        CanonicalChangeRecord record = new CanonicalChangeRecord(
+                "postgres-debezium",
+                "c",
+                "public",
+                "orders",
+                123L,
+                Map.of(),
+                Map.of("status", "new"),
+                Map.of("id", 7L)
+        );
+
+        assertThrows(UnsupportedOperationException.class,
+                () -> record.getAfter().put("status", "mutated"));
+        assertThrows(UnsupportedOperationException.class,
+                () -> record.getPk().remove("id"));
+    }
+
+    @Test
+    void preservesNullEntriesForDatabaseRows() {
+        Map<String, Object> after = new LinkedHashMap<>();
+        after.put("optional_column", null);
+
+        CanonicalChangeRecord record = new CanonicalChangeRecord(
+                "postgres-debezium",
+                "c",
+                "public",
+                "orders",
+                123L,
+                null,
+                after,
+                null
+        );
+
+        assertTrue(record.getAfter().containsKey("optional_column"));
+        assertNull(record.getAfter().get("optional_column"));
+        assertTrue(record.getBefore().isEmpty());
+        assertTrue(record.getPk().isEmpty());
+    }
+}