Fix: [AEA-6316] - Slack Notification on Knowledgebase Sync

.github/workflows/cdk_package_code.yml

@@ -54,7 +54,6 @@ 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 \
@@ -65,10 +64,6 @@ 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; \
@@ -79,13 +74,11 @@ 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

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

@@ -16,10 +16,6 @@
 			"name": "packages/syncKnowledgeBaseFunction",
 			"path": "../packages/syncKnowledgeBaseFunction"
 		},
-		{
-			"name": "packages/notifyS3UploadFunction",
-			"path": "../packages/notifyS3UploadFunction"
-		},
 		{
 			"name": "packages/preprocessingFunction",
 			"path": "../packages/preprocessingFunction"

Makefile

@@ -38,7 +38,6 @@ 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
 
@@ -83,7 +82,6 @@ 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
@@ -104,7 +102,6 @@ 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/LambdaFunction.ts

@@ -29,6 +29,8 @@ export interface LambdaFunctionProps {
   readonly logRetentionInDays: number
   readonly logLevel: string
   readonly dependencyLocation?: string
+  readonly reservedConcurrentExecutions?: number
+  readonly timeout_in_seconds?: Duration
 }
 
 // Lambda Insights layer for enhanced monitoring
@@ -131,7 +133,7 @@ export class LambdaFunction extends Construct {
     const lambdaFunction = new LambdaFunctionResource(this, props.functionName, {
       runtime: Runtime.PYTHON_3_14,
       memorySize: 256,
-      timeout: Duration.seconds(50),
+      timeout: props.timeout_in_seconds ?? Duration.seconds(50),
       architecture: Architecture.X86_64,
       handler: props.handler,
       code: Code.fromAsset(props.packageBasePath, {
@@ -149,7 +151,8 @@ export class LambdaFunction extends Construct {
         POWERTOOLS_LOG_LEVEL: props.logLevel
       },
       logGroup,
-      layers: layers
+      layers: layers,
+      reservedConcurrentExecutions: props.reservedConcurrentExecutions
     })
 
     // Suppress CFN guard rules for Lambda function

packages/cdk/constructs/SimpleQueueService.ts

@@ -8,10 +8,13 @@ import {LambdaFunction} from "./LambdaFunction"
 export interface SimpleQueueServiceProps {
   readonly stackName: string
   readonly queueName: string
-  readonly batchDelay: number
   readonly functions: Array<LambdaFunction>
 }
 
+/**
+ * AWS Simple Queue Service
+ * @see {@link https://aws.amazon.com/sqs/}
+ */
 export class SimpleQueueService extends Construct {
   public queue: Queue
   public deadLetterQueue: Queue
@@ -31,10 +34,9 @@ export class SimpleQueueService extends Construct {
     // 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
+      retentionPeriod: Duration.days(14), // Max
       encryption: QueueEncryption.KMS,
       encryptionMasterKey: kmsKey,
-      visibilityTimeout: Duration.seconds(60),
       enforceSSL: true
     })
 
@@ -46,30 +48,28 @@ export class SimpleQueueService extends Construct {
         encryptionMasterKey: kmsKey,
         deadLetterQueue: {
           queue: deadLetterQueue,
-          maxReceiveCount: 3 // Move to DLQ after 3 failed attempts
+          maxReceiveCount: 1 // Move to DLQ after a failed attempt
         },
-        deliveryDelay: Duration.seconds(0),
-        visibilityTimeout: Duration.seconds(60),
+        deliveryDelay: Duration.seconds(10),
+        visibilityTimeout: Duration.hours(1), // Really high visibility to prevent multiple calls
         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
+      maxBatchingWindow: Duration.seconds(30),
+      batchSize: 20
     })
 
     props.functions.forEach(fn => {
       fn.function.addEventSource(eventSource)
+      fn.function.addEnvironment("SQS_URL", queue.queueUrl)
+
       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/nagSuppressions.ts

@@ -28,18 +28,6 @@ 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,
@@ -174,6 +162,18 @@ export const nagSuppressions = (stack: Stack, account: string) => {
     ]
   )
 
+  // Suppress wildcard permissions for Preprocessing policy
+  safeAddNagSuppression(
+    stack,
+    "/EpsAssistMeStack/RuntimePolicies/SyncKnowledgeBasePolicy/Resource",
+    [
+      {
+        id: "AwsSolutions-IAM5",
+        reason: "Preprocessing Lambda needs wildcard permissions to read/write any file in raw/ and processed/ prefixes."
+      }
+    ]
+  )
+
   // Suppress secrets without rotation
   safeAddNagSuppressionGroup(
     stack,
@@ -460,6 +460,7 @@ export const nagSuppressions = (stack: Stack, account: string) => {
       }
     ]
   )
+
   // Suppress BedrockLogging Provider framework runtime version
   safeAddNagSuppression(
     stack,

packages/cdk/resources/DatabaseTables.ts

@@ -8,6 +8,7 @@ export interface TablesProps {
 
 export class DatabaseTables extends Construct {
   public readonly slackBotStateTable: DynamoDbTable
+  public readonly knowledgeSyncStateTable: DynamoDbTable
 
   constructor(scope: Construct, id: string, props: TablesProps) {
     super(scope, id)
@@ -24,5 +25,18 @@ export class DatabaseTables extends Construct {
       },
       timeToLiveAttribute: "ttl"
     })
+
+    this.knowledgeSyncStateTable = new DynamoDbTable(this, "KnowledgeSyncStateTable", {
+      tableName: `${props.stackName}-KnowledgeSyncState`,
+      partitionKey: {
+        name: "user_channel_composite",
+        type: AttributeType.STRING
+      },
+      sortKey: {
+        name: "last_ts",
+        type: AttributeType.STRING
+      },
+      timeToLiveAttribute: "ttl"
+    })
   }
 }

packages/cdk/resources/Functions.ts

@@ -36,14 +36,13 @@ export interface FunctionsProps {
   readonly mainSlackBotLambdaExecutionRoleArn : string
   readonly ragModelId: string
   readonly reformulationModelId: string
-  readonly notifyS3UploadFunctionPolicy: ManagedPolicy
   readonly docsBucketName: string
+  readonly knowledgeSyncStateTable: TableV2
 }
 
 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) {
@@ -130,29 +129,17 @@ export class Functions extends Construct {
       dependencyLocation: ".dependencies/syncKnowledgeBaseFunction",
       environmentVariables: {
         "KNOWLEDGEBASE_ID": props.knowledgeBaseId,
-        "DATA_SOURCE_ID": props.dataSourceId
-      },
-      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"
+        "SLACK_BOT_ACTIVE": `${!props.isPullRequest}`,
+        "DATA_SOURCE_ID": props.dataSourceId,
+        "KNOWLEDGE_SYNC_STATE_TABLE": props.knowledgeSyncStateTable.tableName
       },
-      additionalPolicies: [props.notifyS3UploadFunctionPolicy]
+      additionalPolicies: [props.syncKnowledgeBaseManagedPolicy],
+      reservedConcurrentExecutions: 1
     })
 
     this.slackBotLambda = slackBotLambda
     this.preprocessingFunction = preprocessingFunction
     this.syncKnowledgeBaseFunction = syncKnowledgeBaseFunction
-    this.notifyS3UploadFunction = notifyS3UploadFunction
   }
 }

packages/cdk/resources/RuntimePolicies.ts

@@ -8,6 +8,8 @@ export interface RuntimePoliciesProps {
   readonly slackSigningSecretParameterName: string
   readonly slackBotStateTableArn: string
   readonly slackBotStateTableKmsKeyArn: string
+  readonly knowledgeSyncStateTableArn: string
+  readonly knowledgeSyncStateTableKmsKeyArn: string
   readonly knowledgeBaseArn: string
   readonly guardrailArn: string
   readonly dataSourceArn: string
@@ -21,7 +23,6 @@ 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) {
@@ -123,40 +124,59 @@ export class RuntimePolicies extends Construct {
       ]
     })
 
-    // Create managed policy for SyncKnowledgeBase Lambda function
-    const syncKnowledgeBasePolicy = new PolicyStatement({
+    const syncKnowledgeBaseBedrockPolicy = new PolicyStatement({
       actions: [
         "bedrock:StartIngestionJob",
         "bedrock:GetIngestionJob",
         "bedrock:ListIngestionJobs"
       ],
       resources: [
-        props.knowledgeBaseArn,
-        props.dataSourceArn
+        props.knowledgeBaseArn
       ]
     })
 
-    this.syncKnowledgeBasePolicy = new ManagedPolicy(this, "SyncKnowledgeBasePolicy", {
-      description: "Policy for SyncKnowledgeBase Lambda to trigger ingestion jobs",
-      statements: [syncKnowledgeBasePolicy]
-    })
-
-    // Create managed policy for S3UpdateNotification Lambda function
-    const notifyS3UploadFunctionPolicy = new PolicyStatement({
+    const syncKnowledgeBaseSSMPolicy = new PolicyStatement({
       actions: [
-        "ssm:GetParameter",
-        "sqs:ReceiveMessage",
-        "sqs:DeleteMessage"
+        "ssm:GetParameter"
       ],
       resources: [
-        props.knowledgeBaseArn,
         ...slackBotPolicyResources
       ]
     })
 
-    this.notifyS3UploadFunctionPolicy = new ManagedPolicy(this, "notifyS3UploadFunctionPolicy", {
-      description: "Policy for S3UpdateNotification Lambda to access SSM parameters",
-      statements: [notifyS3UploadFunctionPolicy]
+    const knowledgeSyncDynamoDbPolicy = new PolicyStatement({
+      actions: [
+        "dynamodb:GetItem",
+        "dynamodb:PutItem",
+        "dynamodb:DeleteItem",
+        "dynamodb:Query",
+        "dynamodb:Scan",
+        "dynamodb:BatchGetItem",
+        "dynamodb:BatchWriteItem",
+        "dynamodb:UpdateItem"
+      ],
+      resources: [props.knowledgeSyncStateTableArn]
+    })
+
+    const knowledgeSyncKmsPolicy = new PolicyStatement({
+      actions: [
+        "kms:Encrypt",
+        "kms:Decrypt",
+        "kms:ReEncrypt",
+        "kms:GenerateDataKey",
+        "kms:DescribeKey"
+      ],
+      resources: [props.knowledgeSyncStateTableKmsKeyArn]
+    })
+
+    this.syncKnowledgeBasePolicy = new ManagedPolicy(this, "SyncKnowledgeBasePolicy", {
+      description: "Policy for SyncKnowledgeBase Lambda to trigger ingestion jobs",
+      statements: [
+        syncKnowledgeBaseBedrockPolicy,
+        syncKnowledgeBaseSSMPolicy,
+        knowledgeSyncDynamoDbPolicy,
+        knowledgeSyncKmsPolicy
+      ]
     })
 
     //policy for the preprocessing lambda

packages/cdk/resources/S3LambdaNotification.ts

@@ -25,11 +25,7 @@ export class S3LambdaNotification extends Construct {
     const queue = new SimpleQueueService(this, `${props.stackName}-${queueName}`, {
       stackName: props.stackName,
       queueName: queueName,
-      batchDelay: 100,
       functions: [
-        // Temporarily only trigger sync, as only one can run at once on SQS
-        //    - if notifications is successful, sync won't run
-        // props.functions.notifyS3UploadFunction,
         props.functions.syncKnowledgeBaseFunction
       ]
     })