Update: [AEA-5931] - Add S3 Notifications Lambda (#352)

## Summary

Add notification to Slack when files update in S3

### Details

1. Add new Simple Queue Service (SQS)
   - SQS triggers when the S3 bucket updates
   - 60 second buffer for incoming messages
2. SQS triggers lambdas
   - syncKnowledgeBaseFunction
   - preprocessingFunction
   - notifyS3UploadFunction (new)
3. New lambda (notifyS3UploadFunction)
   - Collects any files changed over 60 seconds
   - Sends a message to all *private* channels it is installed in
4. Moved syncKnowledgeBaseFunction to handle SQS events 
   - S3 events are provided as a "message" inside SQS events

---------

Co-authored-by: Bence Gadanyi <bence.gadanyi1@nhs.net>

Author: kieran-wilkinson-4

.github/workflows/cdk_package_code.yml

@@ -68,6 +68,7 @@ jobs:
         run: |
           poetry show --only=slackBotFunction | grep -E "^[a-zA-Z]" | awk '{print $1"=="$2}' > requirements_slackBotFunction
           poetry show --only=syncKnowledgeBaseFunction | grep -E "^[a-zA-Z]" | awk '{print $1"=="$2}' > requirements_syncKnowledgeBaseFunction
+          poetry show --only=notifyS3UploadFunction | grep -E "^[a-zA-Z]" | awk '{print $1"=="$2}' > requirements_notifyS3UploadFunction
           poetry show --only=preprocessingFunction | grep -E "^[a-zA-Z]" | awk '{print $1"=="$2}' > requirements_preprocessingFunction
           poetry show --only=bedrockLoggingConfigFunction | grep -E "^[a-zA-Z]" | awk '{print $1"=="$2}' > requirements_bedrockLoggingConfigFunction
           if [ ! -s requirements_slackBotFunction ] || [ "$(grep -c -v '^[[:space:]]*$' requirements_slackBotFunction)" -eq 0 ]; then \
@@ -78,6 +79,10 @@ jobs:
             echo "Error: requirements_syncKnowledgeBaseFunction is empty or contains only blank lines"; \
             exit 1; \
           fi
+          if [ ! -s requirements_notifyS3UploadFunction ] || [ "$(grep -c -v '^[[:space:]]*$' requirements_notifyS3UploadFunction)" -eq 0 ]; then \
+            echo "Error: requirements_notifyS3UploadFunction is empty or contains only blank lines"; \
+            exit 1; \
+          fi
           if [ ! -s requirements_preprocessingFunction ] || [ "$(grep -c -v '^[[:space:]]*$' requirements_preprocessingFunction)" -eq 0 ]; then \
             echo "Error: requirements_preprocessingFunction is empty or contains only blank lines"; \
             exit 1; \
@@ -88,10 +93,14 @@ jobs:
           fi
           mkdir -p .dependencies/slackBotFunction/python
           mkdir -p .dependencies/syncKnowledgeBaseFunction/python
+          mkdir -p .dependencies/notifyS3UploadFunction/python
           mkdir -p .dependencies/preprocessingFunction/python
+          mkdir -p .dependencies/bedrockLoggingConfigFunction/python
           pip3 install -r requirements_slackBotFunction -t .dependencies/slackBotFunction/python
           pip3 install -r requirements_syncKnowledgeBaseFunction -t .dependencies/syncKnowledgeBaseFunction/python
           pip3 install -r requirements_preprocessingFunction -t .dependencies/preprocessingFunction/python
+          pip3 install -r requirements_notifyS3UploadFunction -t .dependencies/notifyS3UploadFunction/python
+          pip3 install -r requirements_bedrockLoggingConfigFunction -t .dependencies/bedrockLoggingConfigFunction/python
           rm -rf .dependencies/preprocessingFunction/python/magika* .dependencies/preprocessingFunction/python/onnxruntime*
           cp packages/preprocessingFunction/magika_shim.py .dependencies/preprocessingFunction/python/magika.py
           find .dependencies/preprocessingFunction/python -type d -name "tests" -exec rm -rf {} + 2>/dev/null || true
@@ -101,10 +110,6 @@ jobs:
           find .dependencies/preprocessingFunction/python -type f \( -name "*.pyc" -o -name "*.pyo" -o -name "*.so.debug" \) -delete
           find .dependencies/preprocessingFunction/python -type f -name "*.md" ! -name "README.md" -delete
           find .dependencies/preprocessingFunction/python -name "*.txt" -size +10k -delete
-          mkdir -p .dependencies/bedrockLoggingConfigFunction/python
-          pip3 install -r requirements_slackBotFunction -t .dependencies/slackBotFunction/python
-          pip3 install -r requirements_syncKnowledgeBaseFunction -t .dependencies/syncKnowledgeBaseFunction/python
-          pip3 install -r requirements_bedrockLoggingConfigFunction -t .dependencies/bedrockLoggingConfigFunction/python
 
       - name: "Tar files"
         run: |

.vscode/eps-assist-me.code-workspace

@@ -15,6 +15,14 @@
 		{
 			"name": "packages/syncKnowledgeBaseFunction",
 			"path": "../packages/syncKnowledgeBaseFunction"
+		},
+		{
+			"name": "packages/notifyS3UploadFunction",
+			"path": "../packages/notifyS3UploadFunction"
+		},
+		{
+			"name": "packages/preprocessingFunction",
+			"path": "../packages/preprocessingFunction"
 		}
 	],
 	"settings": {

Makefile

@@ -48,6 +48,7 @@ lint-flake8:
 test:
 	cd packages/slackBotFunction && PYTHONPATH=. COVERAGE_FILE=coverage/.coverage poetry run python -m pytest
 	cd packages/syncKnowledgeBaseFunction && PYTHONPATH=. COVERAGE_FILE=coverage/.coverage poetry run python -m pytest
+	cd packages/notifyS3UploadFunction && PYTHONPATH=. COVERAGE_FILE=coverage/.coverage poetry run python -m pytest
 	cd packages/preprocessingFunction && PYTHONPATH=. COVERAGE_FILE=coverage/.coverage poetry run python -m pytest
 	cd packages/bedrockLoggingConfigFunction && PYTHONPATH=. COVERAGE_FILE=coverage/.coverage poetry run python -m pytest
 
@@ -109,6 +110,7 @@ cdk-synth: cdk-synth-pr cdk-synth-non-pr
 cdk-synth-non-pr:
 	mkdir -p .dependencies/slackBotFunction
 	mkdir -p .dependencies/syncKnowledgeBaseFunction
+	mkdir -p .dependencies/notifyS3UploadFunction
 	mkdir -p .dependencies/preprocessingFunction
 	mkdir -p .dependencies/bedrockLoggingConfigFunction
 	mkdir -p .local_config
@@ -129,6 +131,7 @@ cdk-synth-non-pr:
 cdk-synth-pr:
 	mkdir -p .dependencies/slackBotFunction
 	mkdir -p .dependencies/syncKnowledgeBaseFunction
+	mkdir -p .dependencies/notifyS3UploadFunction
 	mkdir -p .dependencies/preprocessingFunction
 	mkdir -p .dependencies/bedrockLoggingConfigFunction
 	mkdir -p .local_config

packages/cdk/constructs/S3LambdaNotification.ts

@@ -0,0 +1,77 @@
+import {Construct} from "constructs"
+import {Duration, RemovalPolicy} from "aws-cdk-lib"
+import {Queue, QueueEncryption} from "aws-cdk-lib/aws-sqs"
+import {Key} from "aws-cdk-lib/aws-kms"
+import {SqsEventSource} from "aws-cdk-lib/aws-lambda-event-sources"
+import {LambdaFunction} from "./LambdaFunction"
+
+export interface SimpleQueueServiceProps {
+  readonly stackName: string
+  readonly queueName: string
+  readonly batchDelay: number
+  readonly functions: Array<LambdaFunction>
+}
+
+export class SimpleQueueService extends Construct {
+  public queue: Queue
+  public deadLetterQueue: Queue
+  public kmsKey: Key
+
+  constructor(scope: Construct, id: string, props: SimpleQueueServiceProps) {
+    super(scope, id)
+
+    const name = `${props.stackName}-${props.queueName}`.toLocaleLowerCase()
+
+    const kmsKey = new Key(this, `${name}-queue-key`, {
+      enableKeyRotation: true,
+      description: `KMS key for ${props.queueName} queue and dead-letter queue encryption`,
+      removalPolicy: RemovalPolicy.DESTROY
+    })
+
+    // Create a Dead-Letter Queue (DLQ) for handling failed messages, to help with debugging
+    const deadLetterQueue = new Queue(this, `${name}-dlq`, {
+      queueName: `${name}-dlq`,
+      retentionPeriod: Duration.days(14), // Max 14
+      encryption: QueueEncryption.KMS,
+      encryptionMasterKey: kmsKey,
+      visibilityTimeout: Duration.seconds(60),
+      enforceSSL: true
+    })
+
+    // Create the main SQS Queue with DLQ configured
+    const queue = new Queue(this, name,
+      {
+        queueName: name,
+        encryption: QueueEncryption.KMS,
+        encryptionMasterKey: kmsKey,
+        deadLetterQueue: {
+          queue: deadLetterQueue,
+          maxReceiveCount: 3 // Move to DLQ after 3 failed attempts
+        },
+        deliveryDelay: Duration.seconds(0),
+        visibilityTimeout: Duration.seconds(60),
+        enforceSSL: true
+      }
+    )
+
+    // Add queues as event source for the notify function and sync knowledge base function
+    // While batching, the messages will be sent if maxBatchingWindow is reached or batchSize is reached
+    // Set (very) large batch size to improve wait efficiency of batching window
+    const eventSource = new SqsEventSource(queue, {
+      maxBatchingWindow: Duration.seconds(props.batchDelay),
+      batchSize: 1000,
+      reportBatchItemFailures: true
+    })
+
+    props.functions.forEach(fn => {
+      fn.function.addEventSource(eventSource)
+      queue.grantConsumeMessages(fn.function)
+    })
+
+    // Grant the Lambda function permissions to consume messages from the queue
+
+    this.kmsKey = kmsKey
+    this.queue = queue
+    this.deadLetterQueue = deadLetterQueue
+  }
+}

packages/cdk/constructs/SimpleQueueService.ts

@@ -28,6 +28,18 @@ export const nagSuppressions = (stack: Stack, account: string) => {
     ]
   )
 
+  // Suppress wildcard log permissions for NotifyS3UploadFunction Lambda
+  safeAddNagSuppression(
+    stack,
+    "/EpsAssistMeStack/Functions/NotifyS3UploadFunction/LambdaPutLogsManagedPolicy/Resource",
+    [
+      {
+        id: "AwsSolutions-IAM5",
+        reason: "Wildcard permissions are required for log stream access under known paths."
+      }
+    ]
+  )
+
   // Suppress wildcard log permissions for Preprocessing Lambda
   safeAddNagSuppression(
     stack,

packages/cdk/nagSuppressions.ts

@@ -36,12 +36,14 @@ export interface FunctionsProps {
   readonly mainSlackBotLambdaExecutionRoleArn : string
   readonly ragModelId: string
   readonly queryReformulationModelId: string
+  readonly notifyS3UploadFunctionPolicy: ManagedPolicy
   readonly docsBucketName: string
 }
 
 export class Functions extends Construct {
   public readonly slackBotLambda: LambdaFunction
   public readonly syncKnowledgeBaseFunction: LambdaFunction
+  public readonly notifyS3UploadFunction: LambdaFunction
   public readonly preprocessingFunction: LambdaFunction
 
   constructor(scope: Construct, id: string, props: FunctionsProps) {
@@ -133,8 +135,24 @@ export class Functions extends Construct {
       additionalPolicies: [props.syncKnowledgeBaseManagedPolicy]
     })
 
+    const notifyS3UploadFunction = new LambdaFunction(this, "NotifyS3UploadFunction", {
+      stackName: props.stackName,
+      functionName: `${props.stackName}-S3UpdateFunction`,
+      packageBasePath: "packages/notifyS3UploadFunction",
+      handler: "app.handler.handler",
+      logRetentionInDays: props.logRetentionInDays,
+      logLevel: props.logLevel,
+      dependencyLocation: ".dependencies/notifyS3UploadFunction",
+      environmentVariables: {
+        "SLACK_BOT_TOKEN_PARAMETER": props.slackBotTokenParameter.parameterName,
+        "SLACK_BOT_ACTIVE_ON_PRS": "false"
+      },
+      additionalPolicies: [props.notifyS3UploadFunctionPolicy]
+    })
+
     this.slackBotLambda = slackBotLambda
     this.preprocessingFunction = preprocessingFunction
     this.syncKnowledgeBaseFunction = syncKnowledgeBaseFunction
+    this.notifyS3UploadFunction = notifyS3UploadFunction
   }
 }

packages/cdk/resources/Functions.ts

@@ -21,6 +21,7 @@ export interface RuntimePoliciesProps {
 export class RuntimePolicies extends Construct {
   public readonly slackBotPolicy: ManagedPolicy
   public readonly syncKnowledgeBasePolicy: ManagedPolicy
+  public readonly notifyS3UploadFunctionPolicy: ManagedPolicy
   public readonly preprocessingPolicy: ManagedPolicy
 
   constructor(scope: Construct, id: string, props: RuntimePoliciesProps) {
@@ -51,11 +52,15 @@ export class RuntimePolicies extends Construct {
       resources: [props.knowledgeBaseArn]
     })
 
+    const slackBotPolicyResources = [
+      `arn:aws:ssm:${props.region}:${props.account}:parameter${props.slackBotTokenParameterName}`,
+      `arn:aws:ssm:${props.region}:${props.account}:parameter${props.slackSigningSecretParameterName}`
+    ]
+
     const slackBotSSMPolicy = new PolicyStatement({
       actions: ["ssm:GetParameter"],
       resources: [
-        `arn:aws:ssm:${props.region}:${props.account}:parameter${props.slackBotTokenParameterName}`,
-        `arn:aws:ssm:${props.region}:${props.account}:parameter${props.slackSigningSecretParameterName}`
+        ...slackBotPolicyResources
       ]
     })
 
@@ -136,6 +141,24 @@ export class RuntimePolicies extends Construct {
       statements: [syncKnowledgeBasePolicy]
     })
 
+    // Create managed policy for S3UpdateNotification Lambda function
+    const notifyS3UploadFunctionPolicy = new PolicyStatement({
+      actions: [
+        "ssm:GetParameter",
+        "sqs:ReceiveMessage",
+        "sqs:DeleteMessage"
+      ],
+      resources: [
+        props.knowledgeBaseArn,
+        ...slackBotPolicyResources
+      ]
+    })
+
+    this.notifyS3UploadFunctionPolicy = new ManagedPolicy(this, "notifyS3UploadFunctionPolicy", {
+      description: "Policy for S3UpdateNotification Lambda to access SSM parameters",
+      statements: [notifyS3UploadFunctionPolicy]
+    })
+
     //policy for the preprocessing lambda
     const preprocessingS3Policy = new PolicyStatement({
       actions: [