KMS-703: Move metadata-correction auditing from RDF4J to Amazon DocumentDB Serverless

.gitignore

@@ -49,3 +49,4 @@ notes/
 archive-processor/downloaded-rdf
 archive-processor/local-kms-csv
 extern
+serverless/certs/us-east-1-bundle.pem

README.md

@@ -39,9 +39,15 @@ npm run localstack:start
 By default, `start-local` enables Redis with the local container settings from `bin/env/local_env.sh`, so the normal local startup path is:
 ```bash
 npm run redis:start
+npm run documentdb:start
 npm run start-local
 ```
 
+The local MongoDB 8 container provides a DocumentDB-compatible metadata-correction audit store.
+It listens on `localhost:27018` for host scripts and uses the shared KMS Docker network for SAM.
+Starting it also creates or verifies the metadata-correction audit indexes used in AWS.
+Stop it with `npm run documentdb:stop`.
+
 If you do not need Redis for your local test, start local with Redis disabled:
 ```bash
 REDIS_ENABLED=false npm run start-local
@@ -71,7 +77,8 @@ Local development intentionally splits responsibilities between SAM and LocalSta
 
 - SAM runs the API Gateway and Lambda side of KMS locally.
 - LocalStack emulates AWS-managed services that SAM does not model end-to-end for this repo, especially SNS and SQS.
-- RDF4J and Redis remain separate local services because they are not AWS services.
+- RDF4J, Redis, and the MongoDB-compatible audit database remain separate local services because
+  they are not modeled by SAM.
 
 We do not run the entire application stack inside LocalStack because the existing SAM flow is simpler for day-to-day Lambda/API development, while LocalStack is most useful here for the managed messaging pieces. For keyword event processing, `npm run start-local` also starts `scripts/localstack/run_bridge.sh`, which runs `scripts/localstack/bridge.js`.
 
@@ -297,7 +304,7 @@ Internally, the correction flow is now object-first:
   key construction
 - XML and UMM delegates work from `oldKeywordObject` / `newKeywordObject`
 - joined `oldKeywordPath` / `newKeywordPath` strings are now primarily boundary values for Redis,
-  logs, and audit records
+  logs, and audit documents
 
 The important distinction is:
 
@@ -440,6 +447,48 @@ Resolved corrections are also object-first now:
 Audit logging still derives `oldKeywordPath` / `newKeywordPath` strings for readability, but the
 runtime correction and delegate flow works from normalized keyword objects.
 
+Each collection-correction run is stored as one audit document. Its `statusHistory` records the
+`checked`, `pending`, and terminal `applied` or `failed` transitions. The audit document also
+links to the current CMR collection record, records the prior and resulting CMR revision IDs, and
+stores a bounded unified diff between the original native metadata and the corrected writeback
+payload.
+The audit API is:
+
+- `GET /metadata_correction_audit` for newest-first, token-paginated audit searches. Supported
+  filters include collection, keyword UUID, action, scheme, status, native format, KMS version,
+  source, and date range. Supplied actions and schemes must be recognized KMS values, limits must
+  be integers from 1 through 250, and `startDate` must not be after `endDate`. List results contain
+  compact collection, status, and old-to-new keyword path summaries. Add `?includeDiff=true` to
+  include each available native-metadata diff in the list results.
+- `GET /metadata_correction_audit/{runId}` for the complete audit document. Add
+  `?includeDiff=true` when the native-metadata diff is needed; it is omitted by default to keep
+  routine responses small.
+- Add `?format=html` to either endpoint for a self-contained browser view. The list is a compact
+  status and keyword-change summary with links to each run; add `&includeDiff=true` if list-level
+  native diffs are needed. The detail view includes run context, lifecycle history, diagnostics,
+  and the colored side-by-side native metadata diff. HTML lists default to 10 records per page.
+
+Publisher events carry the published KMS version through the queue into this document. Manual
+correction endpoints look up the current published version before starting the run, so the
+metadata-correction consumer and audit API do not query RDF4J.
+
+Deployed Lambdas use the public AWS `us-east-1` CA bundle to validate DocumentDB TLS connections.
+The Bamboo deployment downloads the current bundle from the
+[AWS certificate trust store](https://truststore.pki.rds.amazonaws.com/us-east-1/us-east-1-bundle.pem)
+before building the deployment image. For an AWS deployment outside Bamboo, download it manually:
+
+```bash
+curl --fail --location \
+  https://truststore.pki.rds.amazonaws.com/us-east-1/us-east-1-bundle.pem \
+  --output serverless/certs/us-east-1-bundle.pem
+```
+
+AWS documents the CA-bundle download requirement in
+[Connecting programmatically to Amazon DocumentDB](https://docs.aws.amazon.com/documentdb/latest/developerguide/connect_programmatically.html).
+That example uses the global bundle; KMS uses the equivalent regional bundle listed for
+US East (N. Virginia) in the
+[AWS regional certificate bundle table](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/UsingWithRDS.SSL.html).
+
 ## Setting up the RDF Database for local development
 In order to run KMS locally, you first need to setup a RDF database.
 ### Prerequisites
@@ -499,6 +548,9 @@ export bamboo_CMR_WRITER_TOKEN=[optional complete bearer authorization value use
 export bamboo_CMR_WRITEBACK_PROVIDERS=[optional provider id, comma-separated list, or ALL]
 export bamboo_CMR_WRITEBACK_VALIDATE_KEYWORDS=[true|false; defaults to false]
 export bamboo_CMR_WRITEBACK_VALIDATE_UMM_C=[true|false; defaults to false]
+export bamboo_CMR_WRITEBACK_TIMEOUT_MS=[optional timeout in milliseconds; defaults to 25000]
+export bamboo_METADATA_CORRECTION_RUNS_PER_MINUTE=[optional positive integer correction run rate]
+export bamboo_METADATA_CORRECTION_SERVICE_RESERVED_CONCURRENCY=[optional; defaults to 5]
 export bamboo_CORS_ORIGIN=[comma separated list of cors origins]
 export bamboo_RDF4J_CONTAINER_MEMORY_LIMIT=[7168 for sit|uat, 14336 for prod]
 export bamboo_RDF4J_INSTANCE_TYPE=["M5.LARGE" for sit|uat, "R5.LARGE" for prod]
@@ -521,6 +573,11 @@ Notes:
 - Leave `bamboo_CMR_WRITEBACK_PROVIDERS` empty to disable provider rollout for CMR writeback.
 - Set `bamboo_CMR_WRITEBACK_VALIDATE_KEYWORDS` and `bamboo_CMR_WRITEBACK_VALIDATE_UMM_C`
   to `true` to reject writebacks that still fail CMR keyword or UMM-C validation.
+- `bamboo_CMR_WRITEBACK_TIMEOUT_MS` is capped at 45000 milliseconds so the worker can record a
+  failed audit before its 60-second Lambda timeout.
+- Setting `bamboo_METADATA_CORRECTION_RUNS_PER_MINUTE` enables queue pacing and forces the
+  metadata-correction worker concurrency to `1`. When it is unset, pacing is disabled and
+  `bamboo_METADATA_CORRECTION_SERVICE_RESERVED_CONCURRENCY` controls concurrency.
 - If you are not deploying into an existing API Gateway, set `bamboo_EXISTING_API_ID` and `bamboo_ROOT_RESOURCE_ID` to empty strings.
 - If `bamboo_RDF4J_BACKUP_VAULT_NAME` is set, `SnapshotStack` imports that existing backup vault. This is useful when `rdf4jSnapshotStack` is being recreated after an RDF4J recovery event and you need the new stack to reuse an existing vault instead of trying to create the same vault name again.
 - If `bamboo_RDF4J_BACKUP_VAULT_NAME` is not set, `SnapshotStack` creates the default `rdf4j-backup-vault`.

bin/deploy-bamboo.sh

@@ -18,6 +18,14 @@ config="`jq '.edl.uid = $newValue' --arg newValue $bamboo_EDL_UID <<< $config`"
 # overwrite static.config.json with new values
 echo $config > tmp.$$.json && mv tmp.$$.json static.config.json
 
+# Download the current public AWS CA bundle before it is packaged with the Lambdas.
+documentDbCaBundleUrl='https://truststore.pki.rds.amazonaws.com/us-east-1/us-east-1-bundle.pem'
+documentDbCaBundlePath='serverless/certs/us-east-1-bundle.pem'
+mkdir -p "$(dirname "$documentDbCaBundlePath")"
+curl --fail --silent --show-error --location \
+  "$documentDbCaBundleUrl" \
+  --output "$documentDbCaBundlePath"
+
 # Set up Docker image
 #####################
 
@@ -70,6 +78,8 @@ dockerRun() {
         --env "CMR_WRITEBACK_PROVIDERS=${bamboo_CMR_WRITEBACK_PROVIDERS:-}" \
         --env "CMR_WRITEBACK_VALIDATE_KEYWORDS=${bamboo_CMR_WRITEBACK_VALIDATE_KEYWORDS:-false}" \
         --env "CMR_WRITEBACK_VALIDATE_UMM_C=${bamboo_CMR_WRITEBACK_VALIDATE_UMM_C:-false}" \
+        --env "CMR_WRITEBACK_TIMEOUT_MS=${bamboo_CMR_WRITEBACK_TIMEOUT_MS:-25000}" \
+        --env "METADATA_CORRECTION_RUNS_PER_MINUTE=${bamboo_METADATA_CORRECTION_RUNS_PER_MINUTE:-}" \
         --env "METADATA_CORRECTION_SERVICE_RESERVED_CONCURRENCY=${bamboo_METADATA_CORRECTION_SERVICE_RESERVED_CONCURRENCY:-5}" \
         --env "BLOCK_PUBLISH_ON_KEYWORD_DIFF_FAILURE=${bamboo_BLOCK_PUBLISH_ON_KEYWORD_DIFF_FAILURE:-false}" \
         --env "KEYWORD_SYNC_ALARM_EMAILS=${bamboo_KEYWORD_SYNC_ALARM_EMAILS:-}" \

bin/documentdb/start.sh

@@ -0,0 +1,65 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+PROJECT_ROOT="$(cd "${SCRIPT_DIR}/../.." && pwd)"
+# shellcheck source=bin/env/local_env.sh
+source "${SCRIPT_DIR}/../env/local_env.sh"
+
+IMAGE="${DOCUMENTDB_LOCAL_IMAGE:-mongo:8.0}"
+CONTAINER_PORT="27017"
+VOLUME_NAME="${DOCUMENTDB_LOCAL_VOLUME:-kms-documentdb-local-data}"
+
+if ! docker network inspect "${KMS_DOCKER_NETWORK}" >/dev/null 2>&1; then
+  docker network create "${KMS_DOCKER_NETWORK}" >/dev/null
+  echo "Created docker network '${KMS_DOCKER_NETWORK}'"
+fi
+
+existing_id="$(docker ps -aq --filter "name=^${DOCUMENTDB_CONTAINER_NAME}$")"
+if [[ -n "${existing_id}" ]]; then
+  running_id="$(docker ps -q --filter "name=^${DOCUMENTDB_CONTAINER_NAME}$")"
+  if [[ -n "${running_id}" ]]; then
+    echo "Mongo container '${DOCUMENTDB_CONTAINER_NAME}' is already running"
+  else
+    docker start "${DOCUMENTDB_CONTAINER_NAME}" >/dev/null
+    echo "Started existing Mongo container '${DOCUMENTDB_CONTAINER_NAME}'"
+  fi
+else
+  docker run -d \
+    --name "${DOCUMENTDB_CONTAINER_NAME}" \
+    --network "${KMS_DOCKER_NETWORK}" \
+    --network-alias "${DOCUMENTDB_CONTAINER_NAME}" \
+    -p "${DOCUMENTDB_HOST_PORT}:${CONTAINER_PORT}" \
+    -v "${VOLUME_NAME}:/data/db" \
+    "${IMAGE}" >/dev/null
+
+  echo "Started Mongo container '${DOCUMENTDB_CONTAINER_NAME}' on ${DOCUMENTDB_HOST_PORT}->${CONTAINER_PORT}"
+fi
+
+mongo_ready=false
+for _ in {1..30}; do
+  if docker exec "${DOCUMENTDB_CONTAINER_NAME}" \
+    mongosh --quiet --eval 'db.runCommand({ ping: 1 }).ok' 2>/dev/null | grep -q '1'; then
+    mongo_ready=true
+    break
+  fi
+
+  sleep 1
+done
+
+if [[ "${mongo_ready}" != "true" ]]; then
+  echo "Mongo container '${DOCUMENTDB_CONTAINER_NAME}' did not become ready" >&2
+  exit 1
+fi
+
+(
+  cd "${PROJECT_ROOT}"
+  DOCUMENTDB_URI="mongodb://127.0.0.1:${DOCUMENTDB_HOST_PORT}/?directConnection=true" \
+    ./node_modules/.bin/vite-node \
+      --config vite.config.js \
+      scripts/local/initialize_metadata_correction_audit.mjs
+)
+
+echo "Connect from host using mongodb://localhost:${DOCUMENTDB_HOST_PORT}"
+echo "Connect from SAM using ${DOCUMENTDB_URI}"

bin/documentdb/stop.sh

@@ -0,0 +1,22 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+# shellcheck source=bin/env/local_env.sh
+source "${SCRIPT_DIR}/../env/local_env.sh"
+
+container_id="$(docker ps -aq --filter "name=^${DOCUMENTDB_CONTAINER_NAME}$")"
+if [[ -z "${container_id}" ]]; then
+  echo "Mongo container '${DOCUMENTDB_CONTAINER_NAME}' does not exist"
+  exit 0
+fi
+
+running_id="$(docker ps -q --filter "name=^${DOCUMENTDB_CONTAINER_NAME}$")"
+if [[ -z "${running_id}" ]]; then
+  echo "Mongo container '${DOCUMENTDB_CONTAINER_NAME}' is already stopped"
+  exit 0
+fi
+
+docker stop "${DOCUMENTDB_CONTAINER_NAME}" >/dev/null
+echo "Stopped Mongo container '${DOCUMENTDB_CONTAINER_NAME}'"

bin/env/local_env.sh

@@ -19,5 +19,8 @@ export LOCALSTACK_CONTAINER_NAME="${LOCALSTACK_CONTAINER_NAME:-kms-localstack}"
 export LOCALSTACK_IMAGE="${LOCALSTACK_IMAGE:-localstack/localstack:3.8.1}"
 export LOCALSTACK_PORT="${LOCALSTACK_PORT:-4566}"
 export AWS_ENDPOINT_URL="${AWS_ENDPOINT_URL:-http://localstack:${LOCALSTACK_PORT}}"
+export DOCUMENTDB_URI="${DOCUMENTDB_URI:-mongodb://kms-documentdb-local:27017/?directConnection=true}"
+export DOCUMENTDB_CONTAINER_NAME="${DOCUMENTDB_CONTAINER_NAME:-kms-documentdb-local}"
+export DOCUMENTDB_HOST_PORT="${DOCUMENTDB_HOST_PORT:-27018}"
 export SAM_WARM_CONTAINERS="${SAM_WARM_CONTAINERS:-LAZY}"
 export SAM_LOCAL_WATCH="${SAM_LOCAL_WATCH:-false}"

bin/start-local.sh

@@ -25,7 +25,7 @@ clearStaleSAMContainers() {
     echo "Clearing stale SAM containers..."
 
     docker ps --format '{{.ID}} {{.Image}}' \
-      | awk '$2 ~ /public\.ecr\.aws\/lambda\/nodejs:22-rapid-/ { print $1 }' \
+      | awk '$2 ~ /public\.ecr\.aws\/lambda\/nodejs:[0-9]+-rapid-/ { print $1 }' \
       | xargs -r docker rm -f >/dev/null 2>&1 || true
 }
 

cdk/app/lib/CmrEventProcessingStack.ts

@@ -1,5 +1,6 @@
 import * as cdk from 'aws-cdk-lib'
 import * as ec2 from 'aws-cdk-lib/aws-ec2'
+import * as secretsmanager from 'aws-cdk-lib/aws-secretsmanager'
 import * as sns from 'aws-cdk-lib/aws-sns'
 import { Construct } from 'constructs'
 
@@ -14,18 +15,20 @@ import { VpcSetup } from './helper/VpcSetup'
 export interface CmrEventProcessingStackProps extends cdk.StackProps {
   cmrBaseUrl: string
   cmrSystemTokenParameterName?: string
+  cmrWritebackTimeoutMs?: string
   metadataCorrectionRequestDelayMs?: string
+  metadataCorrectionRunsPerMinute?: string
   metadataCorrectionServiceReservedConcurrency?: string
   cmrWriterToken?: string
   cmrWritebackProviders?: string
   cmrWritebackValidateKeywords?: string
   cmrWritebackValidateUmmC?: string
+  metadataCorrectionAuditClientSecurityGroup?: ec2.ISecurityGroup
+  metadataCorrectionAuditEnvironment: Record<string, string>
+  metadataCorrectionAuditSecret?: secretsmanager.ISecret
   redisEnabled?: string
   redisHost?: string
   redisPort?: string
-  rdf4jPassword: string
-  rdf4jServiceUrl: string
-  rdf4jUserName: string
   prefix: string
   stage: string
   topicArn: string
@@ -66,20 +69,23 @@ export class CmrEventProcessingStack extends cdk.Stack {
     const metadataCorrectionSetup = new MetadataCorrectionSetup(this, 'MetadataCorrection', {
       cmrBaseUrl: props.cmrBaseUrl,
       cmrSystemTokenParameterName: props.cmrSystemTokenParameterName,
+      cmrWritebackTimeoutMs: props.cmrWritebackTimeoutMs,
       metadataCorrectionRequestDelayMs: props.metadataCorrectionRequestDelayMs,
+      metadataCorrectionRunsPerMinute: props.metadataCorrectionRunsPerMinute,
       metadataCorrectionServiceReservedConcurrency:
         props.metadataCorrectionServiceReservedConcurrency,
       cmrWriterToken: props.cmrWriterToken,
       cmrWritebackProviders: props.cmrWritebackProviders,
       cmrWritebackValidateKeywords: props.cmrWritebackValidateKeywords,
       cmrWritebackValidateUmmC: props.cmrWritebackValidateUmmC,
+      metadataCorrectionAuditClientSecurityGroup:
+        props.metadataCorrectionAuditClientSecurityGroup,
+      metadataCorrectionAuditEnvironment: props.metadataCorrectionAuditEnvironment,
+      metadataCorrectionAuditSecret: props.metadataCorrectionAuditSecret,
       prefix: props.prefix,
       redisEnabled: props.redisEnabled,
       redisHost: props.redisHost,
       redisPort: props.redisPort,
-      rdf4jPassword: props.rdf4jPassword,
-      rdf4jServiceUrl: props.rdf4jServiceUrl,
-      rdf4jUserName: props.rdf4jUserName,
       stage: props.stage,
       securityGroup: this.securityGroup,
       useLocalstack,

cdk/app/lib/KmsStack.ts

@@ -3,6 +3,7 @@ import * as cdk from 'aws-cdk-lib'
 import * as apigateway from 'aws-cdk-lib/aws-apigateway'
 import * as ec2 from 'aws-cdk-lib/aws-ec2'
 import * as iam from 'aws-cdk-lib/aws-iam'
+import * as secretsmanager from 'aws-cdk-lib/aws-secretsmanager'
 import * as sns from 'aws-cdk-lib/aws-sns'
 import { Construct } from 'constructs'
 
@@ -19,12 +20,16 @@ import { VpcSetup } from './helper/VpcSetup'
  */
 export interface KmsStackProps extends cdk.StackProps {
   cmrSystemTokenParameterName?: string
+  cmrWritebackTimeoutMs?: string
   cmrWritebackProviders?: string
   cmrWritebackValidateKeywords?: string
   cmrWritebackValidateUmmC?: string
   cmrWriterToken?: string
   existingApiId: string | undefined
   keywordSyncAlarmEmails?: string[]
+  metadataCorrectionAuditClientSecurityGroup?: ec2.ISecurityGroup
+  metadataCorrectionAuditEnvironment: Record<string, string>
+  metadataCorrectionAuditSecret?: secretsmanager.ISecret
   prefix: string
   rootResourceId: string | undefined
   stage: string
@@ -124,6 +129,8 @@ export class KmsStack extends cdk.Stack {
     )
     this.lambdaRole = iamSetup.lambdaRole
 
+    props.metadataCorrectionAuditSecret?.grantRead(this.lambdaRole)
+
     this.keywordEventsTopic = new sns.Topic(this, 'KeywordEventsTopic', {
       topicName: keywordEventsTopicName
     })
@@ -175,12 +182,16 @@ export class KmsStack extends cdk.Stack {
       lambdaRole: this.lambdaRole,
       metadataCorrectionEnvironment: {
         CMR_SYSTEM_TOKEN_PARAMETER_NAME: props.cmrSystemTokenParameterName || '',
+        CMR_WRITEBACK_TIMEOUT_MS: props.cmrWritebackTimeoutMs || '',
         CMR_WRITER_TOKEN: props.cmrWriterToken || '',
         CMR_WRITEBACK_PROVIDERS: props.cmrWritebackProviders || '',
         CMR_WRITEBACK_VALIDATE_KEYWORDS: props.cmrWritebackValidateKeywords || '',
         CMR_WRITEBACK_VALIDATE_UMM_C: props.cmrWritebackValidateUmmC || '',
-        METADATA_CORRECTION_REQUESTS_TOPIC_ARN: metadataCorrectionRequestsTopicArn
+        METADATA_CORRECTION_REQUESTS_TOPIC_ARN: metadataCorrectionRequestsTopicArn,
+        ...props.metadataCorrectionAuditEnvironment
       },
+      metadataCorrectionAuditClientSecurityGroup:
+        props.metadataCorrectionAuditClientSecurityGroup,
       prefix,
       securityGroup: this.securityGroup,
       stage: this.stage,