KMS-660: Publish keyword events to SNS after a successful publish (#99)

Author: cgokey

.gitignore

@@ -44,3 +44,4 @@ loadtest/locust/**/summary.csv~
 loadtest/locust/
 scripts/load/hammer_endpoints_sequential.js
 data/scheme-size*
+notes/

README.md

@@ -27,17 +27,22 @@ Prerequisites:
 - Docker
 - aws-sam-cli (`brew install aws-sam-cli`)
 
-To start local server (including rdf4j database server, cdk synth and sam)
-First, make sure to start LocalStack:
+To start local server, first make sure to start LocalStack:
 ```
 npm run localstack:start
 ```
 
-Then, you can start the local server:
-```
+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 start-local
 ```
 
+If you do not need Redis for your local test, start local with Redis disabled:
+```bash
+REDIS_ENABLED=false npm run start-local
+```
+
 To run local server with SAM watch mode enabled
 ```
 npm run start-local:watch
@@ -51,35 +56,11 @@ Local development intentionally splits responsibilities between SAM and LocalSta
 - 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.
 
-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/local/run_localstack_cmr_keyword_events_bridge.js`], which polls the LocalStack SQS queue and forwards messages into the local CMR consumer handler. This bridge exists because `sam local start-api` does not emulate SQS event source mappings the way AWS does in deployed environments.
+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`.
 
-### Testing keyword event publishing in SIT
+This bridge exists because `sam local start-api` does not emulate EventBridge targets or SQS event source mappings the way AWS does in deployed environments.
 
-After deploying to SIT, you can exercise the keyword event publisher with:
-
-```bash
-curl -H "Authorization: Bearer $TOKEN" -X POST https://cmr.sit.earthdata.nasa.gov/kms/keyword-events/test \
-  -H 'Content-Type: application/json' \
-  -d '{
-    "EventType": "UPDATED",
-    "Scheme": "sciencekeywords",
-    "UUID": "4f81c61c-f100-4bc4-9664-d9b70d2f162f",
-    "OldKeywordPath": "Instruments > Solar/Space Observing Instruments > Passive Remote Sensing",
-    "NewKeywordPath": "Instruments > Earth Remote Sensing Instruments > Passive Remote Sensing",
-    "Timestamp": "2026-03-19T10:41:57.720Z",
-    "MetadataSpecification": {
-      "Name": "Kms-Keyword-Event",
-      "URL": "https://cdn.earthdata.nasa.gov/kms-keyword-event/v1.0",
-      "Version": "1.0"
-    }
-  }'
-```
-
-Expected result:
-- the API returns `200`
-- the response includes the SNS topic ARN and message id
-- the CMR event processor is invoked from the subscribed queue in AWS
-- `EventType` must be one of `INSERTED`, `UPDATED`, or `DELETED`
+For bridge implementation details and extension guidance, see `scripts/localstack/README.md`
 
 ### Optional: Enable Redis cache in local SAM/LocalStack
 

bin/deploy-bamboo.sh

@@ -63,6 +63,7 @@ dockerRun() {
         --env "RDF4J_PASSWORD=$bamboo_RDF4J_PASSWORD" \
         --env "EDL_PASSWORD=$bamboo_EDL_PASSWORD" \
         --env "CMR_BASE_URL=$bamboo_CMR_BASE_URL" \
+        --env "BLOCK_PUBLISH_ON_KEYWORD_DIFF_FAILURE=${bamboo_BLOCK_PUBLISH_ON_KEYWORD_DIFF_FAILURE:-false}" \
         --env "CORS_ORIGIN=$bamboo_CORS_ORIGIN" \
         --env "RDF4J_INSTANCE_TYPE=$bamboo_RDF4J_INSTANCE_TYPE" \
         --env "RDF4J_CONTAINER_MEMORY_LIMIT=$bamboo_RDF4J_CONTAINER_MEMORY_LIMIT" \

bin/env/local_env.sh

@@ -2,9 +2,16 @@
 
 # Shared defaults for local SAM/CDK workflows.
 export RDF4J_SERVICE_URL="${RDF4J_SERVICE_URL:-http://rdf4j-server:8080}"
+export RDF4J_HOST_SERVICE_URL="${RDF4J_HOST_SERVICE_URL:-http://localhost:8081}"
+export RDF4J_USER_NAME="${RDF4J_USER_NAME:-rdf4j}"
+export RDF4J_PASSWORD="${RDF4J_PASSWORD:-rdf4j}"
+export RDF4J_CONTAINER_MEMORY_LIMIT="${RDF4J_CONTAINER_MEMORY_LIMIT:-2048}"
 export REDIS_ENABLED="${REDIS_ENABLED:-true}"
 export REDIS_HOST="${REDIS_HOST:-kms-redis-local}"
 export REDIS_PORT="${REDIS_PORT:-6379}"
+export REDIS_FAIL_FAST="${REDIS_FAIL_FAST:-true}"
+export REDIS_HOST_SERVICE_HOST="${REDIS_HOST_SERVICE_HOST:-localhost}"
+export REDIS_HOST_PORT="${REDIS_HOST_PORT:-6380}"
 export KMS_DOCKER_NETWORK="${KMS_DOCKER_NETWORK:-kms-network}"
 export LOCALSTACK_CONTAINER_NAME="${LOCALSTACK_CONTAINER_NAME:-kms-localstack}"
 export LOCALSTACK_IMAGE="${LOCALSTACK_IMAGE:-localstack/localstack:3.8.1}"

bin/localstack/start.sh

@@ -6,12 +6,31 @@ SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
 # shellcheck source=bin/env/local_env.sh
 source "${SCRIPT_DIR}/../env/local_env.sh"
 
+REQUIRED_SERVICES="sns,sqs,events"
+
 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=^${LOCALSTACK_CONTAINER_NAME}$")"
+if [[ -n "${existing_id}" ]]; then
+  configured_services="$(
+    docker inspect \
+      --format '{{range .Config.Env}}{{println .}}{{end}}' \
+      "${LOCALSTACK_CONTAINER_NAME}" \
+      | grep '^SERVICES=' \
+      | cut -d= -f2- \
+      || true
+  )"
+
+  if [[ ",${configured_services}," != *",events,"* ]]; then
+    docker rm -f "${LOCALSTACK_CONTAINER_NAME}" >/dev/null
+    echo "Recreating LocalStack container '${LOCALSTACK_CONTAINER_NAME}' to enable services: ${REQUIRED_SERVICES}"
+    existing_id=""
+  fi
+fi
+
 if [[ -n "${existing_id}" ]]; then
   running_id="$(docker ps -q --filter "name=^${LOCALSTACK_CONTAINER_NAME}$")"
   if [[ -n "${running_id}" ]]; then
@@ -29,11 +48,11 @@ docker run -d \
   --network "${KMS_DOCKER_NETWORK}" \
   --network-alias "localstack" \
   -p "${LOCALSTACK_PORT}:4566" \
-  -e SERVICES="sns,sqs" \
+  -e SERVICES="${REQUIRED_SERVICES}" \
   -e AWS_DEFAULT_REGION="us-east-1" \
   -e EDGE_PORT="4566" \
   "${LOCALSTACK_IMAGE}" >/dev/null
 
 echo "Started LocalStack container '${LOCALSTACK_CONTAINER_NAME}' on ${LOCALSTACK_PORT}->4566"
-echo "SNS/SQS endpoint for SAM/Lambda containers: ${AWS_ENDPOINT_URL}"
-echo "SNS/SQS endpoint from host: http://localhost:${LOCALSTACK_PORT}"
+echo "SNS/SQS/EventBridge endpoint for SAM/Lambda containers: ${AWS_ENDPOINT_URL}"
+echo "SNS/SQS/EventBridge endpoint from host: http://localhost:${LOCALSTACK_PORT}"

bin/start-local.sh

@@ -5,20 +5,28 @@ PROJECT_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)"
 # shellcheck source=bin/env/local_env.sh
 source "${SCRIPT_DIR}/env/local_env.sh"
 
-LOCAL_CONSUMER_PID=""
+LOCAL_BRIDGE_PID=""
 
 # Function to clean up processes and containers
 cleanup() {
     echo "Cleaning up..."
 
-    if [ -n "${LOCAL_CONSUMER_PID}" ] && kill -0 "${LOCAL_CONSUMER_PID}" >/dev/null 2>&1; then
-      kill "${LOCAL_CONSUMER_PID}" >/dev/null 2>&1 || true
-      wait "${LOCAL_CONSUMER_PID}" 2>/dev/null || true
+    if [ -n "${LOCAL_BRIDGE_PID}" ] && kill -0 "${LOCAL_BRIDGE_PID}" >/dev/null 2>&1; then
+      kill "${LOCAL_BRIDGE_PID}" >/dev/null 2>&1 || true
+      wait "${LOCAL_BRIDGE_PID}" 2>/dev/null || true
     fi
 
     exit 0
 }
 
+clearStaleSAMContainers() {
+    echo "Clearing stale SAM containers..."
+
+    docker ps --format '{{.ID}} {{.Image}}' \
+      | awk '$2 ~ /public\.ecr\.aws\/lambda\/nodejs:22-rapid-/ { print $1 }' \
+      | xargs -r docker rm -f >/dev/null 2>&1 || true
+}
+
 # Set up trap to call cleanup function on Ctrl+C
 trap cleanup SIGINT
 
@@ -27,8 +35,10 @@ if [ "$SAM_LOCAL_WATCH" = "true" ]; then
   SAM_WATCH_ARGS+=(--beta-features --watch)
 fi
 
-vite-node --config "${PROJECT_ROOT}/vite.config.js" "${PROJECT_ROOT}/scripts/local/run_localstack_cmr_keyword_events_bridge.js" &
-LOCAL_CONSUMER_PID=$!
+clearStaleSAMContainers
+
+"${PROJECT_ROOT}/scripts/localstack/run_bridge.sh" &
+LOCAL_BRIDGE_PID=$!
 
 # Synthesize the CDK stack
 cd cdk

cdk/app/lib/KmsStack.ts

@@ -24,6 +24,7 @@ export interface KmsStackProps extends cdk.StackProps {
   environment: {
     CMR_BASE_URL: string
     EDL_PASSWORD: string
+    BLOCK_PUBLISH_ON_KEYWORD_DIFF_FAILURE?: string
     LOG_LEVEL: string
     REDIS_ENABLED?: string
     REDIS_HOST?: string

cdk/app/lib/helper/KmsLambdaFunctions.ts

@@ -27,6 +27,7 @@ interface LambdaFunctionsProps {
   environment: {
     CMR_BASE_URL: string;
     EDL_PASSWORD: string;
+    BLOCK_PUBLISH_ON_KEYWORD_DIFF_FAILURE?: string;
     REDIS_ENABLED?: string;
     REDIS_HOST?: string;
     REDIS_PORT?: string;
@@ -133,7 +134,6 @@ export class LambdaFunctions {
     this.createTreeOperationApiLambdas(scope)
     this.createNightlyCachePrimeCron(scope)
     this.createCrudOperationApiLambdas(scope)
-    this.createKeywordEventTestPublishApiLambda(scope)
     this.createPublishEventBridgeWiring(scope)
     this.createExportRdfCrons(scope)
   }
@@ -400,24 +400,6 @@ export class LambdaFunctions {
     )
   }
 
-  /**
-   * Creates the keyword events test publish endpoint.
-   * Queue-based event consumption lives in CmrEventProcessingStack.
-   * @param {Construct} scope - The scope in which to define these constructs
-   * @private
-   */
-  private createKeywordEventTestPublishApiLambda(scope: Construct) {
-    this.createApiLambda(
-      scope,
-      'keywordEventsTestPublish/handler.js',
-      'keyword-events-test-publish',
-      'keywordEventsTestPublish',
-      '/keyword-events/test',
-      'POST',
-      true
-    )
-  }
-
   /**
    * Creates EventBridge wiring for publish events and cache-prime target execution.
    *

cdk/bin/main.ts

@@ -160,6 +160,7 @@ async function main() {
       RDF_BUCKET_NAME: process.env.RDF_BUCKET_NAME || 'kms-rdf-backup',
       CMR_BASE_URL: process.env.CMR_BASE_URL || 'https://cmr.earthdata.nasa.gov',
       EDL_PASSWORD: process.env.EDL_PASSWORD || '',
+      BLOCK_PUBLISH_ON_KEYWORD_DIFF_FAILURE: process.env.BLOCK_PUBLISH_ON_KEYWORD_DIFF_FAILURE,
       LOG_LEVEL: process.env.LOG_LEVEL || 'INFO',
       REDIS_ENABLED: redisEnabledValue,
       REDIS_HOST: useLocalstack ? localRedisHost : redisStack?.endpointAddress,

scripts/local/create_unique_keyword.sh

@@ -0,0 +1,47 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+API_BASE_URL="${API_BASE_URL:-http://127.0.0.1:3013}"
+VERSION="${VERSION:-draft}"
+SCHEME_ID="${SCHEME_ID:-sciencekeywords}"
+BROADER_ID="${BROADER_ID:-1eb0ea0a-312c-4d74-8d42-6f1ad758f999}"
+LABEL_PREFIX="${LABEL_PREFIX:-LOCAL TEST KEYWORD}"
+TIMESTAMP_SUFFIX="$(date +%s)"
+CONCEPT_ID="$(uuidgen | tr '[:upper:]' '[:lower:]')"
+PREF_LABEL="${LABEL_PREFIX} ${TIMESTAMP_SUFFIX}"
+REFERENCE_URL="${REFERENCE_URL:-https://example.com/local-test-keyword}"
+DEFINITION_TEXT="${DEFINITION_TEXT:-Local test keyword created for publish verification ${TIMESTAMP_SUFFIX}}"
+
+read -r -d '' RDF_XML <<EOF || true
+<rdf:RDF
+  xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#"
+  xmlns:skos="http://www.w3.org/2004/02/skos/core#"
+  xmlns:gcmd="https://gcmd.earthdata.nasa.gov/kms#"
+  xmlns:dcterms="http://purl.org/dc/terms/"
+  xml:base="https://gcmd.earthdata.nasa.gov/kms/concept/">
+  <skos:Concept rdf:about="${CONCEPT_ID}">
+    <skos:prefLabel xml:lang="en">${PREF_LABEL}</skos:prefLabel>
+    <skos:definition xml:lang="en">${DEFINITION_TEXT}</skos:definition>
+    <gcmd:reference gcmd:text="${REFERENCE_URL}" xml:lang="en"/>
+    <skos:inScheme rdf:resource="https://gcmd.earthdata.nasa.gov/kms/concepts/concept_scheme/${SCHEME_ID}"/>
+    <skos:broader rdf:resource="${BROADER_ID}"/>
+  </skos:Concept>
+</rdf:RDF>
+EOF
+
+echo "Creating keyword:"
+echo "  conceptId: ${CONCEPT_ID}"
+echo "  prefLabel: ${PREF_LABEL}"
+echo "  scheme: ${SCHEME_ID}"
+echo "  broader: ${BROADER_ID}"
+
+curl \
+  --fail \
+  --silent \
+  --show-error \
+  -X POST "${API_BASE_URL}/concept?version=${VERSION}" \
+  -H 'Content-Type: application/rdf+xml' \
+  --data-binary "${RDF_XML}"
+
+echo