fix(dynamodb): honor MICROSECOND ApproximateCreationDateTimePrecision for Kinesis destinations and streams false enablement

compatibility-tests/compat-terraform/test/dynamodb-kinesis-precision-tf/main.tf

@@ -0,0 +1,44 @@
+# Verifies that a DynamoDB Kinesis streaming destination keeps the precision it was created with.
+#
+# EnableKinesisStreamingDestination ignored ApproximateCreationDateTimePrecision and
+# DescribeKinesisStreamingDestination always reported MILLISECOND. The attribute is ForceNew
+# in the provider, so a MICROSECOND destination was destroyed and recreated on every apply.
+#
+# The table deliberately leaves DynamoDB Streams off: enabling the destination used to switch
+# them on, which aws_dynamodb_table then planned to switch off again on every run.
+
+variable "precision" {
+  type    = string
+  default = "MICROSECOND"
+}
+
+resource "aws_kinesis_stream" "events" {
+  name             = "floci-ddb-kinesis-precision"
+  shard_count      = 1
+  retention_period = 24
+}
+
+resource "aws_dynamodb_table" "items" {
+  name         = "floci-ddb-kinesis-precision"
+  billing_mode = "PAY_PER_REQUEST"
+  hash_key     = "pk"
+
+  attribute {
+    name = "pk"
+    type = "S"
+  }
+}
+
+resource "aws_dynamodb_kinesis_streaming_destination" "items" {
+  stream_arn                               = aws_kinesis_stream.events.arn
+  table_name                               = aws_dynamodb_table.items.name
+  approximate_creation_date_time_precision = var.precision
+}
+
+output "table_name" {
+  value = aws_dynamodb_table.items.name
+}
+
+output "stream_arn" {
+  value = aws_kinesis_stream.events.arn
+}

compatibility-tests/compat-terraform/test/dynamodb-kinesis-precision-tf/provider.tf

@@ -0,0 +1,29 @@
+terraform {
+  required_providers {
+    aws = {
+      source  = "hashicorp/aws"
+      version = "~> 6.0"
+    }
+  }
+}
+
+variable "endpoint" {
+  type    = string
+  default = "http://localhost:4566"
+}
+
+provider "aws" {
+  region     = "us-east-1"
+  access_key = "test"
+  secret_key = "test"
+
+  skip_credentials_validation = true
+  skip_metadata_api_check     = true
+  skip_requesting_account_id  = true
+  s3_use_path_style           = true
+
+  endpoints {
+    dynamodb = var.endpoint
+    kinesis  = var.endpoint
+  }
+}

compatibility-tests/compat-terraform/test/dynamodb_kinesis_precision.bats

@@ -0,0 +1,95 @@
+#!/usr/bin/env bats
+# DynamoDB Kinesis Streaming Destination Precision Compatibility Test
+#
+# EnableKinesisStreamingDestination ignored ApproximateCreationDateTimePrecision, and
+# DescribeKinesisStreamingDestination always reported MILLISECOND. The provider marks
+# approximate_creation_date_time_precision as ForceNew, so a MICROSECOND destination read back
+# as MILLISECOND and was destroyed and recreated on every apply. Enabling the destination also
+# switched on DynamoDB Streams for the table, so aws_dynamodb_table planned to turn it back off
+# on every run. This verifies the precision reads back as written, a second plan is clean for
+# both resources, and changing the precision replaces the destination once.
+
+setup_file() {
+    load 'test_helper/common-setup'
+
+    DDB_KINESIS_TF_DIR="$(cd "$(dirname "$BATS_TEST_FILENAME")/dynamodb-kinesis-precision-tf" && pwd)"
+    cd "$DDB_KINESIS_TF_DIR"
+
+    echo "# === DynamoDB Kinesis Destination Precision Test ===" >&3
+    echo "# Endpoint: $FLOCI_ENDPOINT" >&3
+    echo "# Config: $DDB_KINESIS_TF_DIR" >&3
+
+    rm -rf .terraform .terraform.lock.hcl terraform.tfstate* 2>/dev/null || true
+
+    echo "# --- terraform init ---" >&3
+    run terraform init -input=false -no-color
+    if [ "$status" -ne 0 ]; then
+        echo "# terraform init failed: $output" >&3
+        return 1
+    fi
+
+    echo "# --- terraform apply (MICROSECOND) ---" >&3
+    run terraform apply -var="endpoint=${FLOCI_ENDPOINT}" -input=false -auto-approve -no-color
+    if [ "$status" -ne 0 ]; then
+        echo "# terraform apply failed: $output" >&3
+        return 1
+    fi
+}
+
+teardown_file() {
+    load 'test_helper/common-setup'
+
+    DDB_KINESIS_TF_DIR="$(cd "$(dirname "$BATS_TEST_FILENAME")/dynamodb-kinesis-precision-tf" && pwd)"
+    cd "$DDB_KINESIS_TF_DIR"
+
+    terraform destroy -var="endpoint=${FLOCI_ENDPOINT}" -input=false -auto-approve -no-color || true
+    rm -rf .terraform .terraform.lock.hcl terraform.tfstate* 2>/dev/null || true
+}
+
+setup() {
+    load 'test_helper/common-setup'
+    DDB_KINESIS_TF_DIR="$(cd "$(dirname "$BATS_TEST_FILENAME")/dynamodb-kinesis-precision-tf" && pwd)"
+}
+
+describe_precision() {
+    local table_name stream_arn
+    table_name=$(terraform -chdir="$DDB_KINESIS_TF_DIR" output -raw table_name)
+    stream_arn=$(terraform -chdir="$DDB_KINESIS_TF_DIR" output -raw stream_arn)
+    aws_cmd dynamodb describe-kinesis-streaming-destination --table-name "$table_name" \
+        --query "KinesisDataStreamDestinations[?StreamArn=='${stream_arn}'].ApproximateCreationDateTimePrecision | [0]" \
+        --output text
+}
+
+@test "DynamoDB Kinesis precision: destination reads back as MICROSECOND" {
+    run describe_precision
+    assert_success
+    assert_output "MICROSECOND"
+}
+
+# The critical assertion: before the fix the precision read back as MILLISECOND, and because
+# the attribute is ForceNew every plan proposed replacing the destination. The table here does
+# not enable streams, so this also catches stream_enabled drift on aws_dynamodb_table.
+@test "DynamoDB Kinesis precision: second plan reports no changes" {
+    cd "$DDB_KINESIS_TF_DIR"
+    run terraform plan -var="endpoint=${FLOCI_ENDPOINT}" -input=false -no-color -detailed-exitcode
+    assert_success
+    assert_output --partial "No changes"
+}
+
+# A real precision change is still ForceNew: it must replace the destination once, then settle.
+@test "DynamoDB Kinesis precision: changing the precision replaces once and re-plans clean" {
+    cd "$DDB_KINESIS_TF_DIR"
+    run terraform apply -var="endpoint=${FLOCI_ENDPOINT}" -var="precision=MILLISECOND" \
+        -input=false -auto-approve -no-color
+    assert_success
+    assert_output --partial "1 added, 0 changed, 1 destroyed"
+
+    run describe_precision
+    assert_success
+    assert_output "MILLISECOND"
+
+    run terraform plan -var="endpoint=${FLOCI_ENDPOINT}" -var="precision=MILLISECOND" \
+        -input=false -no-color -detailed-exitcode
+    assert_success
+    assert_output --partial "No changes"
+}

docs/services/dynamodb.md

@@ -209,6 +209,15 @@ When a table has an **ACTIVE** Kinesis streaming destination (see
 including TTL expirations, is forwarded to the destination stream as a Kinesis record in the
 AWS CDC envelope (`eventName`, `dynamodb.Keys`, `NewImage`/`OldImage`, `ApproximateCreationDateTime`).
 
+`ApproximateCreationDateTime` follows the destination's
+`EnableKinesisStreamingConfiguration.ApproximateCreationDateTimePrecision`: epoch milliseconds for
+`MILLISECOND` (the default) and epoch microseconds for `MICROSECOND`. The precision is returned by
+`DescribeKinesisStreamingDestination` and stamped on each record as
+`dynamodb.ApproximateCreationDateTimePrecision`.
+
+Enabling a Kinesis streaming destination does not change the table's DynamoDB Streams setting.
+Kinesis forwarding works whether or not `StreamSpecification.StreamEnabled` is set.
+
 ### Delivery contract
 
 Forwarding is **bounded best-effort with in-process retry**. A write is never blocked or failed by

src/main/java/io/github/hectorvent/floci/services/dynamodb/DynamoDbJsonHandler.java

@@ -1996,6 +1996,18 @@ private Response handleDeleteResourcePolicy(JsonNode request, String region) {
     private Response handleEnableKinesisStreamingDestination(JsonNode request, String region) {
         String tableName = request.path("TableName").asText();
         String streamArn = request.path("StreamArn").asText();
+        JsonNode precisionNode = request.path("EnableKinesisStreamingConfiguration")
+                .path("ApproximateCreationDateTimePrecision");
+        String precision = precisionNode.isMissingNode() || precisionNode.isNull()
+                ? KinesisStreamingDestination.PRECISION_MILLISECOND : precisionNode.asText();
+        if (!KinesisStreamingDestination.PRECISION_MILLISECOND.equals(precision)
+                && !KinesisStreamingDestination.PRECISION_MICROSECOND.equals(precision)) {
+            throw new AwsException("ValidationException",
+                    "1 validation error detected: Value '" + precision
+                            + "' at 'enableKinesisStreamingConfiguration.approximateCreationDateTimePrecision' "
+                            + "failed to satisfy constraint: Member must satisfy enum value set: [MILLISECOND, MICROSECOND]",
+                    400);
+        }
 
         TableDefinition table = dynamoDbService.describeTable(tableName, region);
         String resolvedTableName = table.getTableName();
@@ -2017,25 +2029,22 @@ private Response handleEnableKinesisStreamingDestination(JsonNode request, Strin
         if (existing.isPresent()) {
             existing.get().setDestinationStatus("ACTIVE");
             existing.get().setDestinationStatusDescription("Kinesis streaming is enabled for this table");
+            existing.get().setApproximateCreationDateTimePrecision(precision);
         } else {
-            table.getKinesisStreamingDestinations().add(new KinesisStreamingDestination(streamArn));
-        }
-
-        if (!table.isStreamEnabled()) {
-            StreamDescription sd = dynamoDbStreamService.enableStream(
-                    resolvedTableName, table.getTableArn(), "NEW_AND_OLD_IMAGES", region);
-            table.setStreamEnabled(true);
-            table.setStreamArn(sd.getStreamArn());
-            table.setStreamViewType("NEW_AND_OLD_IMAGES");
+            table.getKinesisStreamingDestinations().add(new KinesisStreamingDestination(streamArn, precision));
         }
 
+        // DynamoDB Streams is left as the caller configured it: Kinesis forwarding does not depend on it, and
+        // turning it on here showed up as stream_enabled drift on aws_dynamodb_table that never converged.
         dynamoDbService.persistTable(resolvedTableName, table, region);
 
         ObjectNode response = objectMapper.createObjectNode();
         response.put("TableName", resolvedTableName);
         response.put("StreamArn", streamArn);
         response.put("DestinationStatus", "ACTIVE");
         response.put("DestinationStatusDescription", "Kinesis streaming is enabled for this table");
+        response.putObject("EnableKinesisStreamingConfiguration")
+                .put("ApproximateCreationDateTimePrecision", precision);
         return Response.ok(response).build();
     }
 

src/main/java/io/github/hectorvent/floci/services/dynamodb/KinesisStreamingForwarder.java

@@ -17,7 +17,10 @@
 import java.time.Instant;
 import java.util.ArrayDeque;
 import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executors;
@@ -141,23 +144,40 @@ public void forward(String eventName, JsonNode oldItem, JsonNode newItem,
         if (destinations == null || destinations.isEmpty()) {
             return;
         }
+        // Snapshot each active destination's precision once: Enable/Disable mutate these objects in place on
+        // request threads, and the payload chosen for a destination must match the one serialized for it.
+        Map<String, String> precisionByStreamArn = new LinkedHashMap<>();
+        for (KinesisStreamingDestination dest : destinations) {
+            if ("ACTIVE".equals(dest.getDestinationStatus())) {
+                precisionByStreamArn.putIfAbsent(dest.getStreamArn(), dest.getApproximateCreationDateTimePrecision());
+            }
+        }
+        if (precisionByStreamArn.isEmpty()) {
+            return;
+        }
+
         Instant now = Instant.now();
         JsonNode sourceItem = newItem != null ? newItem : oldItem;
         ObjectNode keys = buildKeys(sourceItem, table);
 
-        byte[] data;
+        // One serialized payload per precision in use: every destination sees the same event (same eventID),
+        // stamped at the resolution its ApproximateCreationDateTimePrecision asks for.
+        String eventId = UUID.randomUUID().toString();
+        Map<String, byte[]> dataByPrecision = new HashMap<>(2);
         String partitionKey;
         try {
-            ObjectNode payload = buildPayload(eventName, keys, newItem, oldItem, table.getTableName(), region, now);
-            data = objectMapper.writeValueAsBytes(payload);
+            for (String precision : precisionByStreamArn.values()) {
+                if (!dataByPrecision.containsKey(precision)) {
+                    ObjectNode payload = buildPayload(eventId, eventName, keys, newItem, oldItem,
+                            table.getTableName(), region, now, precision);
+                    dataByPrecision.put(precision, objectMapper.writeValueAsBytes(payload));
+                }
+            }
             partitionKey = extractPartitionKey(keys, table);
         } catch (Exception e) {
             // Serialization is a deterministic terminal failure: nothing can be forwarded for this event.
-            for (KinesisStreamingDestination dest : destinations) {
-                if (!"ACTIVE".equals(dest.getDestinationStatus())) {
-                    continue;
-                }
-                DestinationState st = stateFor(ownerAccountId, region, table.getTableName(), dest.getStreamArn());
+            for (String streamArn : precisionByStreamArn.keySet()) {
+                DestinationState st = stateFor(ownerAccountId, region, table.getTableName(), streamArn);
                 synchronized (st) {
                     st.dropped++;
                     recordError(st, e);
@@ -167,14 +187,12 @@ public void forward(String eventName, JsonNode oldItem, JsonNode newItem,
             return;
         }
 
-        for (KinesisStreamingDestination dest : destinations) {
-            if (!"ACTIVE".equals(dest.getDestinationStatus())) {
-                continue;
-            }
-            String streamName = extractStreamName(dest.getStreamArn());
-            DestinationState st = stateFor(ownerAccountId, region, table.getTableName(), dest.getStreamArn());
+        for (Map.Entry<String, String> target : precisionByStreamArn.entrySet()) {
+            String streamArn = target.getKey();
+            String streamName = extractStreamName(streamArn);
+            DestinationState st = stateFor(ownerAccountId, region, table.getTableName(), streamArn);
             PendingRecord rec = new PendingRecord(ownerAccountId, region, streamName, table.getTableName(),
-                    data, partitionKey, now);
+                    dataByPrecision.get(target.getValue()), partitionKey, now);
             synchronized (st) {
                 if (st.queue.size() >= MAX_BUFFERED) {
                     st.queue.pollFirst(); // drop OLDEST
@@ -413,22 +431,22 @@ List<DestinationForwardingStats> forwardingStats() {
         return out;
     }
 
-    // ──────────────────────────── Payload construction (unchanged) ────────────────────────────
+    // ──────────────────────────── Payload construction ────────────────────────────
 
-    private ObjectNode buildPayload(String eventName, JsonNode keys,
+    private ObjectNode buildPayload(String eventId, String eventName, JsonNode keys,
                                     JsonNode newImage, JsonNode oldImage,
-                                    String tableName, String region, Instant timestamp) {
+                                    String tableName, String region, Instant timestamp, String precision) {
         ObjectNode payload = objectMapper.createObjectNode();
         payload.put("awsRegion", region);
-        payload.put("eventID", UUID.randomUUID().toString());
+        payload.put("eventID", eventId);
         payload.put("eventName", eventName);
         payload.putNull("userIdentity");
         payload.put("recordFormat", "application/json");
         payload.put("tableName", tableName);
         payload.put("eventSource", "aws:dynamodb");
 
         ObjectNode dynamodb = objectMapper.createObjectNode();
-        dynamodb.put("ApproximateCreationDateTime", timestamp.toEpochMilli());
+        dynamodb.put("ApproximateCreationDateTime", approximateCreationDateTime(timestamp, precision));
         if (keys != null) {
             dynamodb.set("Keys", keys);
         }
@@ -439,12 +457,20 @@ private ObjectNode buildPayload(String eventName, JsonNode keys,
             dynamodb.set("OldImage", oldImage);
         }
         dynamodb.put("SizeBytes", 0);
-        dynamodb.put("ApproximateCreationDateTimePrecision", "MILLISECOND");
+        dynamodb.put("ApproximateCreationDateTimePrecision", precision);
         payload.set("dynamodb", dynamodb);
 
         return payload;
     }
 
+    static long approximateCreationDateTime(Instant timestamp, String precision) {
+        if (KinesisStreamingDestination.PRECISION_MICROSECOND.equals(precision)) {
+            return Math.addExact(Math.multiplyExact(timestamp.getEpochSecond(), 1_000_000L),
+                    timestamp.getNano() / 1_000L);
+        }
+        return timestamp.toEpochMilli();
+    }
+
     private ObjectNode buildKeys(JsonNode item, TableDefinition table) {
         ObjectNode keys = objectMapper.createObjectNode();
         if (item == null) {