KMS-676: Consumer - Create Metadata Fix SQS Queue and SNS Topic (CDK)

.eslintignore

@@ -1,2 +1,3 @@
 dist
 .eslintrc
+extern/

.gitignore

@@ -47,3 +47,4 @@ data/scheme-size*
 notes/
 archive-processor/downloaded-rdf
 archive-processor/local-kms-csv
+extern

bin/env/local_env.sh

@@ -17,5 +17,5 @@ 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 SAM_WARM_CONTAINERS="${SAM_WARM_CONTAINERS:-EAGER}"
+export SAM_WARM_CONTAINERS="${SAM_WARM_CONTAINERS:-LAZY}"
 export SAM_LOCAL_WATCH="${SAM_LOCAL_WATCH:-false}"

cdk/app/lib/CmrEventProcessingStack.ts

@@ -1,16 +1,10 @@
-import * as path from 'path'
-
 import * as cdk from 'aws-cdk-lib'
-import * as iam from 'aws-cdk-lib/aws-iam'
-import * as lambda from 'aws-cdk-lib/aws-lambda'
-import * as eventsources from 'aws-cdk-lib/aws-lambda-event-sources'
-import { NodejsFunction } from 'aws-cdk-lib/aws-lambda-nodejs'
 import * as sns from 'aws-cdk-lib/aws-sns'
-import * as subscriptions from 'aws-cdk-lib/aws-sns-subscriptions'
-import * as sqs from 'aws-cdk-lib/aws-sqs'
 import { Construct } from 'constructs'
 
+import { CmrKeywordEventsListenerSetup } from './helper/CmrKeywordEventsListenerSetup'
 import { LogForwardingSetup } from './helper/LogForwardingSetup'
+import { MetadataCorrectionSetup } from './helper/MetadataCorrectionSetup'
 
 /**
  * Properties for the CMR event processing stack.
@@ -29,10 +23,10 @@ export interface CmrEventProcessingStackProps extends cdk.StackProps {
  * that will process those events for downstream CMR business logic.
  */
 export class CmrEventProcessingStack extends cdk.Stack {
-  public readonly keywordEventsQueueUrlOutput: cdk.CfnOutput
+  private readonly logForwardingSetup?: LogForwardingSetup
 
   /**
-   * Creates the CMR queue, subscription, listener, and queue output.
+   * Creates the CMR listener resources and metadata correction messaging resources.
    *
    * @param {Construct} scope - Parent construct.
    * @param {string} id - Stack identifier.
@@ -42,58 +36,32 @@ export class CmrEventProcessingStack extends cdk.Stack {
     super(scope, id, props)
 
     const useLocalstack = this.node.tryGetContext('useLocalstack') === 'true'
-    const queueName = `${props.prefix}-${props.stage}-cmr-keyword-events`
     const topic = sns.Topic.fromTopicArn(this, 'KeywordEventsTopic', props.topicArn)
 
-    const queue = new sqs.Queue(this, 'CmrKeywordEventsQueue', {
-      queueName
-    })
-
-    topic.addSubscription(new subscriptions.SqsSubscription(queue))
-
-    const listenerRole = new iam.Role(this, 'CmrKeywordEventsProcessorRole', {
-      assumedBy: new iam.ServicePrincipal('lambda.amazonaws.com'),
-      managedPolicies: [
-        iam.ManagedPolicy.fromAwsManagedPolicyName('service-role/AWSLambdaBasicExecutionRole')
-      ]
+    const metadataCorrectionSetup = new MetadataCorrectionSetup(this, 'MetadataCorrection', {
+      prefix: props.prefix,
+      stage: props.stage
     })
 
-    const listenerLambda = new NodejsFunction(this, `${props.prefix}-cmr-keyword-events-processor`, {
-      functionName: `${props.prefix}-${props.stage}-cmr-keyword-events-processor`,
-      entry: path.join(__dirname, '../../../serverless/src/cmrKeywordEventsListener/handler.js'),
-      handler: 'cmrKeywordEventsListener',
-      runtime: lambda.Runtime.NODEJS_22_X,
-      timeout: cdk.Duration.seconds(30),
-      memorySize: 1024,
-      role: listenerRole,
-      depsLockFilePath: path.join(__dirname, '../../../package-lock.json'),
-      projectRoot: path.join(__dirname, '../../..')
+    const listenerSetup = new CmrKeywordEventsListenerSetup(this, 'CmrKeywordEventsListener', {
+      prefix: props.prefix,
+      stage: props.stage,
+      keywordEventsTopic: topic,
+      metadataCorrectionRequestsTopic: metadataCorrectionSetup.metadataCorrectionRequestsTopic
     })
 
-    listenerLambda.addEventSource(new eventsources.SqsEventSource(queue, {
-      batchSize: 1
-    }))
-
-    queue.grantConsumeMessages(listenerLambda)
-
     // Set up CloudWatch Logs forwarding to Splunk via NGAP SecLog account
     // Skip log forwarding for localstack deployments
     if (!useLocalstack) {
-      // eslint-disable-next-line no-new
-      new LogForwardingSetup(this, 'LogForwarding', {
+      this.logForwardingSetup = new LogForwardingSetup(this, 'LogForwarding', {
         prefix: props.prefix,
         stage: props.stage,
         logDestinationArn: props.logDestinationArn,
         lambdas: {
-          'cmrKeywordEventsListener/handler.js::cmr-keyword-events-processor': listenerLambda
+          'cmrKeywordEventsListener/handler.js::cmr-keyword-events-processor': listenerSetup.listenerLambda,
+          'metadataCorrectionService/handler.js::metadata-correction-service': metadataCorrectionSetup.metadataCorrectionServiceLambda
         }
       })
     }
-
-    this.keywordEventsQueueUrlOutput = new cdk.CfnOutput(this, 'CmrKeywordEventsQueueUrl', {
-      description: 'Queue URL for CMR keyword event processing',
-      exportName: `${props.prefix}-CmrKeywordEventsQueueUrl`,
-      value: queue.queueUrl
-    })
   }
 }

cdk/app/lib/helper/CmrKeywordEventsListenerSetup.ts

@@ -0,0 +1,77 @@
+import * as path from 'path'
+
+import * as cdk from 'aws-cdk-lib'
+import * as eventsources from 'aws-cdk-lib/aws-lambda-event-sources'
+import { NodejsFunction } from 'aws-cdk-lib/aws-lambda-nodejs'
+import * as sns from 'aws-cdk-lib/aws-sns'
+import * as subscriptions from 'aws-cdk-lib/aws-sns-subscriptions'
+import * as sqs from 'aws-cdk-lib/aws-sqs'
+import { Construct } from 'constructs'
+
+import { NODE_LAMBDA_RUNTIME } from './NodeLambdaRuntime'
+
+/**
+ * Properties for the CMR keyword events listener infrastructure.
+ */
+interface CmrKeywordEventsListenerSetupProps {
+  prefix: string
+  stage: string
+  keywordEventsTopic: sns.ITopic
+  metadataCorrectionRequestsTopic: sns.ITopic
+}
+
+/**
+ * Creates the CMR keyword events queue, listener Lambda, and related wiring.
+ */
+export class CmrKeywordEventsListenerSetup extends Construct {
+  public readonly queue: sqs.Queue
+
+  public readonly listenerLambda: NodejsFunction
+
+  public readonly queueUrlOutput: cdk.CfnOutput
+
+  /**
+   * @param {Construct} scope - Parent construct.
+   * @param {string} id - Construct identifier.
+   * @param {CmrKeywordEventsListenerSetupProps} props - Listener configuration.
+   */
+  constructor(scope: Construct, id: string, props: CmrKeywordEventsListenerSetupProps) {
+    super(scope, id)
+
+    const queueName = `${props.prefix}-${props.stage}-cmr-keyword-events`
+    const projectRoot = path.join(__dirname, '../../../..')
+
+    this.queue = new sqs.Queue(this, 'CmrKeywordEventsQueue', {
+      queueName
+    })
+
+    props.keywordEventsTopic.addSubscription(new subscriptions.SqsSubscription(this.queue))
+
+    this.listenerLambda = new NodejsFunction(this, `${props.prefix}-cmr-keyword-events-processor`, {
+      functionName: `${props.prefix}-${props.stage}-cmr-keyword-events-processor`,
+      entry: path.join(projectRoot, 'serverless/src/cmrKeywordEventsListener/handler.js'),
+      handler: 'cmrKeywordEventsListener',
+      runtime: NODE_LAMBDA_RUNTIME,
+      timeout: cdk.Duration.seconds(30),
+      memorySize: 1024,
+      environment: {
+        METADATA_CORRECTION_REQUESTS_TOPIC_ARN: props.metadataCorrectionRequestsTopic.topicArn
+      },
+      depsLockFilePath: path.join(projectRoot, 'package-lock.json'),
+      projectRoot
+    })
+
+    this.listenerLambda.addEventSource(new eventsources.SqsEventSource(this.queue, {
+      batchSize: 1
+    }))
+
+    this.queue.grantConsumeMessages(this.listenerLambda)
+    props.metadataCorrectionRequestsTopic.grantPublish(this.listenerLambda)
+
+    this.queueUrlOutput = new cdk.CfnOutput(this, 'CmrKeywordEventsQueueUrl', {
+      description: 'Queue URL for CMR keyword event processing',
+      exportName: `${props.prefix}-CmrKeywordEventsQueueUrl`,
+      value: this.queue.queueUrl
+    })
+  }
+}

cdk/app/lib/helper/KmsLambdaFunctions.ts

@@ -11,6 +11,7 @@ import { NodejsFunction, NodejsFunctionProps } from 'aws-cdk-lib/aws-lambda-node
 import { Construct } from 'constructs'
 
 import { ApiResources } from './ApiResources'
+import { NODE_LAMBDA_RUNTIME } from './NodeLambdaRuntime'
 
 /**
  * Interface for LambdaFunctions constructor properties
@@ -582,7 +583,7 @@ export class LambdaFunctions {
         functionName: `${this.props.prefix}-${this.props.stage}-${functionName}`,
         entry: path.join(__dirname, '../../../../serverless/src', handlerPath),
         handler: handlerName,
-        runtime: lambda.Runtime.NODEJS_22_X,
+        runtime: NODE_LAMBDA_RUNTIME,
         timeout,
         memorySize,
         role: this.props.lambdaRole,

cdk/app/lib/helper/MetadataCorrectionSetup.ts

@@ -0,0 +1,144 @@
+import * as path from 'path'
+
+import * as cdk from 'aws-cdk-lib'
+import * as eventsources from 'aws-cdk-lib/aws-lambda-event-sources'
+import { NodejsFunction } from 'aws-cdk-lib/aws-lambda-nodejs'
+import * as sns from 'aws-cdk-lib/aws-sns'
+import * as subscriptions from 'aws-cdk-lib/aws-sns-subscriptions'
+import * as sqs from 'aws-cdk-lib/aws-sqs'
+import { Construct } from 'constructs'
+
+import { NODE_LAMBDA_RUNTIME } from './NodeLambdaRuntime'
+
+/**
+ * Properties for metadata correction infrastructure.
+ */
+interface MetadataCorrectionSetupProps {
+  prefix: string
+  stage: string
+}
+
+/**
+ * Creates the metadata correction SNS/SQS/Lambda plumbing and exports its endpoints.
+ */
+export class MetadataCorrectionSetup extends Construct {
+  public readonly metadataCorrectionRequestsTopic: sns.Topic
+
+  public readonly metadataCorrectionRequestsQueue: sqs.Queue
+
+  public readonly metadataCorrectionRequestsDlq: sqs.Queue
+
+  public readonly metadataCorrectionServiceLambda: NodejsFunction
+
+  public readonly metadataCorrectionRequestsTopicArnOutput: cdk.CfnOutput
+
+  public readonly metadataCorrectionRequestsQueueUrlOutput: cdk.CfnOutput
+
+  public readonly metadataCorrectionRequestsQueueArnOutput: cdk.CfnOutput
+
+  public readonly metadataCorrectionRequestsDlqUrlOutput: cdk.CfnOutput
+
+  public readonly metadataCorrectionRequestsDlqArnOutput: cdk.CfnOutput
+
+  /**
+   * @param {Construct} scope - Parent construct.
+   * @param {string} id - Construct identifier.
+   * @param {MetadataCorrectionSetupProps} props - Metadata correction configuration.
+   */
+  constructor(scope: Construct, id: string, props: MetadataCorrectionSetupProps) {
+    super(scope, id)
+
+    const metadataCorrectionRequestsBaseName = `${props.prefix}-${props.stage}-metadata-correction-requests`
+    const metadataCorrectionRequestsName = `${metadataCorrectionRequestsBaseName}.fifo`
+    const projectRoot = path.join(__dirname, '../../../..')
+
+    // TODO: Create a follow-up ticket for DLQ handling. This DLQ is only the
+    // redrive target today; before adding a consumer, decide whether failures
+    // should be alarmed on, manually inspected, or redriven by an operator.
+    this.metadataCorrectionRequestsDlq = new sqs.Queue(this, 'MetadataCorrectionRequestsDlq', {
+      queueName: `${metadataCorrectionRequestsBaseName}-dlq.fifo`,
+      fifo: true,
+      contentBasedDeduplication: true,
+      retentionPeriod: cdk.Duration.days(14)
+    })
+
+    this.metadataCorrectionRequestsQueue = new sqs.Queue(this, 'MetadataCorrectionRequestsQueue', {
+      queueName: metadataCorrectionRequestsName,
+      fifo: true,
+      contentBasedDeduplication: true,
+      deadLetterQueue: {
+        queue: this.metadataCorrectionRequestsDlq,
+        maxReceiveCount: 3
+      },
+      retentionPeriod: cdk.Duration.days(14),
+      visibilityTimeout: cdk.Duration.minutes(5)
+    })
+
+    this.metadataCorrectionRequestsTopic = new sns.Topic(this, 'MetadataCorrectionRequestsTopic', {
+      contentBasedDeduplication: true,
+      fifo: true,
+      topicName: metadataCorrectionRequestsName
+    })
+
+    this.metadataCorrectionRequestsTopic.addSubscription(new subscriptions.SqsSubscription(
+      this.metadataCorrectionRequestsQueue,
+      {
+        rawMessageDelivery: true
+      }
+    ))
+
+    this.metadataCorrectionServiceLambda = new NodejsFunction(
+      this,
+      `${props.prefix}-metadata-correction-service`,
+      {
+        functionName: `${props.prefix}-${props.stage}-metadata-correction-service`,
+        entry: path.join(projectRoot, 'serverless/src/metadataCorrectionService/handler.js'),
+        handler: 'metadataCorrectionService',
+        runtime: NODE_LAMBDA_RUNTIME,
+        timeout: cdk.Duration.seconds(30),
+        memorySize: 1024,
+        depsLockFilePath: path.join(projectRoot, 'package-lock.json'),
+        projectRoot
+      }
+    )
+
+    this.metadataCorrectionServiceLambda.addEventSource(new eventsources.SqsEventSource(
+      this.metadataCorrectionRequestsQueue,
+      {
+        batchSize: 1
+      }
+    ))
+
+    this.metadataCorrectionRequestsQueue.grantConsumeMessages(this.metadataCorrectionServiceLambda)
+
+    this.metadataCorrectionRequestsTopicArnOutput = new cdk.CfnOutput(this, 'MetadataCorrectionRequestsTopicArn', {
+      description: 'SNS topic ARN for metadata correction request publishing',
+      exportName: `${props.prefix}-MetadataCorrectionRequestsTopicArn`,
+      value: this.metadataCorrectionRequestsTopic.topicArn
+    })
+
+    this.metadataCorrectionRequestsQueueUrlOutput = new cdk.CfnOutput(this, 'MetadataCorrectionRequestsQueueUrl', {
+      description: 'Queue URL for metadata correction request processing',
+      exportName: `${props.prefix}-MetadataCorrectionRequestsQueueUrl`,
+      value: this.metadataCorrectionRequestsQueue.queueUrl
+    })
+
+    this.metadataCorrectionRequestsQueueArnOutput = new cdk.CfnOutput(this, 'MetadataCorrectionRequestsQueueArn', {
+      description: 'Queue ARN for metadata correction request processing',
+      exportName: `${props.prefix}-MetadataCorrectionRequestsQueueArn`,
+      value: this.metadataCorrectionRequestsQueue.queueArn
+    })
+
+    this.metadataCorrectionRequestsDlqUrlOutput = new cdk.CfnOutput(this, 'MetadataCorrectionRequestsDlqUrl', {
+      description: 'DLQ URL for failed metadata correction request processing',
+      exportName: `${props.prefix}-MetadataCorrectionRequestsDlqUrl`,
+      value: this.metadataCorrectionRequestsDlq.queueUrl
+    })
+
+    this.metadataCorrectionRequestsDlqArnOutput = new cdk.CfnOutput(this, 'MetadataCorrectionRequestsDlqArn', {
+      description: 'DLQ ARN for failed metadata correction request processing',
+      exportName: `${props.prefix}-MetadataCorrectionRequestsDlqArn`,
+      value: this.metadataCorrectionRequestsDlq.queueArn
+    })
+  }
+}

cdk/app/lib/helper/NodeLambdaRuntime.ts

@@ -0,0 +1,9 @@
+import * as lambda from 'aws-cdk-lib/aws-lambda'
+
+/**
+ * Shared runtime for all Node.js Lambda functions in the CDK app.
+ *
+ * Keeping this in one place makes Node runtime upgrades explicit and consistent
+ * across stacks and helper constructs.
+ */
+export const NODE_LAMBDA_RUNTIME = lambda.Runtime.NODEJS_22_X