> Add post-processing decompressor lambda with documentation updates

Bob Strahan · Bob Strahan · commit ff29858c32b7 · 2025-12-10T17:11:08.000Z
diff --git a/docs/post-processing-lambda-hook.md b/docs/post-processing-lambda-hook.md
@@ -15,18 +15,58 @@ The GenAIIDP solution supports an optional post-processing Lambda hook that enab
 2. **EventBridge Integration**
    - The solution automatically publishes a custom event to EventBridge
    - Event contains complete document processing details and output locations
-   - EventBridge rule matches the event pattern and triggers your Lambda function
-
-3. **Custom Processing**
-   - Your Lambda function receives the document processing event
-   - Function can access extraction results, confidence scores, and metadata
+   - EventBridge rule matches the event pattern and triggers the decompression lambda
+
+3. **Document Decompression** (New as of compression feature)
+   - An intermediate lambda function in the IDP stack receives the event
+   - If the document is compressed, it decompresses it using `idp_common.models.Document.load_document()`
+   - The decompressed document is reconstructed into the original payload format
+   - The decompression lambda then invokes your custom post-processing lambda
+
+4. **Custom Processing**
+   - Your Lambda function receives the document processing event with a **decompressed** document
+   - No need to import `idp_common` or handle decompression logic
+   - Function can directly access extraction results, confidence scores, and metadata
    - Implement custom logic for notifications, data transformation, or system integration
 
-4. **Error Handling**
+5. **Error Handling**
    - Built-in retry logic for transient failures
    - Dead letter queue for failed invocations
    - CloudWatch monitoring and alerting
 
+## Architecture
+
+```mermaid
+flowchart LR
+    SF[Step Function<br/>Workflow<br/>SUCCEEDED] --> EB[EventBridge<br/>Rule]
+    EB --> DL[Decompression<br/>Lambda<br/>in Stack]
+    DL --> |Check if<br/>compressed| DL
+    DL --> |If compressed:<br/>decompress from<br/>WorkingBucket| S3[(S3<br/>Working<br/>Bucket)]
+    DL --> |Invoke with<br/>decompressed<br/>payload| CPL[Your Custom<br/>Post-Processing<br/>Lambda]
+    DL --> |Handle<br/>errors| DLQ[Dead Letter<br/>Queue]
+    
+    style DL fill:#90EE90
+    style CPL fill:#87CEEB
+    style S3 fill:#FFE4B5
+```
+
+**Key Components:**
+- **EventBridge Rule**: Triggers on StepFunction SUCCEEDED status
+- **Decompression Lambda** (in IDP stack): Handles document decompression transparently
+- **Your Custom Lambda** (external): Receives decompressed payload, no `idp_common` dependency needed
+- **S3 Working Bucket**: Temporary storage for compressed documents
+- **Dead Letter Queue**: Captures failed invocations for retry/analysis
+
+## Important Notes
+
+> **Backward Compatibility**: The decompression lambda was introduced to handle document compression feature added in a previous release. Your custom post-processing lambda receives the document in its **original uncompressed format**, maintaining backward compatibility.
+
+> **No External Dependencies**: Your custom post-processing lambda does **not** need to import the `idp_common` package. The IDP stack's decompression lambda handles all compression/decompression internally.
+
+> **Payload Format**: The event payload structure remains unchanged from the original implementation. All document data, including sections, pages, and attributes, is provided in the standard format documented below.
+
+> **Performance**: The decompression step adds minimal latency (typically < 1 second) and is transparent to your custom lambda function.
+
 ## Configuration
 
 Configure the post-processing hook during stack deployment:
diff --git a/src/lambda/post_processing_decompressor/index.py b/src/lambda/post_processing_decompressor/index.py
@@ -0,0 +1,109 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: MIT-0
+
+import boto3
+import json
+import os
+import logging
+from idp_common.models import Document
+from typing import Dict, Any
+
+logger = logging.getLogger()
+logger.setLevel(os.environ.get("LOG_LEVEL", "INFO"))
+
+lambda_client = boto3.client('lambda')
+
+CUSTOM_POST_PROCESSOR_ARN = os.environ['CUSTOM_POST_PROCESSOR_ARN']
+WORKING_BUCKET = os.environ['WORKING_BUCKET']
+
+
+def handler(event: Dict[str, Any], context) -> Dict[str, Any]:
+    """
+    Decompresses the document from StepFunction output and invokes custom post-processor lambda.
+    
+    This lambda acts as an intermediary between EventBridge and the custom post-processing lambda,
+    handling document decompression so external lambdas don't need to import idp_common.
+    
+    Args:
+        event: EventBridge event containing StepFunction execution details
+        context: Lambda context
+        
+    Returns:
+        Response from custom post-processor lambda invocation
+    """
+    logger.info(f"Processing event for custom post-processor invocation")
+    
+    try:
+        # Extract output data from event
+        output_data = None
+        if event.get('detail', {}).get('output'):
+            output_data = json.loads(event['detail']['output'])
+        
+        if not output_data:
+            logger.error("No output data found in event")
+            raise ValueError("Missing output data in event")
+        
+        # Extract document data - handle both Pattern 1 and Pattern 2/3 structures
+        document_data = None
+        if 'document' in output_data:
+            # Pattern 2/3 structure: at root level
+            document_data = output_data['document']
+            logger.info("Found document in output_data['document']")
+        elif 'Result' in output_data and 'document' in output_data.get('Result', {}):
+            # Pattern 1 structure: wrapped in Result
+            document_data = output_data['Result']['document']
+            logger.info("Found document in output_data['Result']['document']")
+        else:
+            logger.warning("Document not found in expected locations, using entire output")
+            document_data = output_data
+        
+        # Check if document is compressed
+        is_compressed = isinstance(document_data, dict) and document_data.get('compressed', False)
+        
+        if is_compressed:
+            logger.info(f"Document is compressed, decompressing from S3 URI: {document_data.get('s3_uri', 'N/A')}")
+            
+            # Decompress document using idp_common
+            processed_doc = Document.load_document(document_data, WORKING_BUCKET, logger)
+            
+            logger.info(f"Decompressed document: {processed_doc.num_pages} pages, "
+                       f"{len(processed_doc.sections)} sections")
+            
+            # Reconstruct output_data with decompressed document
+            if 'document' in output_data:
+                output_data['document'] = processed_doc.to_dict()
+            elif 'Result' in output_data and 'document' in output_data.get('Result', {}):
+                output_data['Result']['document'] = processed_doc.to_dict()
+            else:
+                output_data = processed_doc.to_dict()
+            
+            # Update event with decompressed payload
+            event['detail']['output'] = json.dumps(output_data)
+            
+            logger.info("Document decompressed successfully")
+        else:
+            logger.info("Document is not compressed, passing through as-is")
+        
+        # Invoke custom post-processor lambda with decompressed payload
+        logger.info(f"Invoking custom post-processor: {CUSTOM_POST_PROCESSOR_ARN}")
+        
+        response = lambda_client.invoke(
+            FunctionName=CUSTOM_POST_PROCESSOR_ARN,
+            InvocationType='Event',  # Async invocation
+            Payload=json.dumps(event)
+        )
+        
+        logger.info(f"Custom post-processor invoked successfully. StatusCode: {response['StatusCode']}")
+        
+        return {
+            'statusCode': 200,
+            'body': json.dumps({
+                'message': 'Successfully invoked custom post-processor',
+                'customProcessorArn': CUSTOM_POST_PROCESSOR_ARN,
+                'documentCompressed': is_compressed
+            })
+        }
+        
+    except Exception as e:
+        logger.error(f"Error processing event: {str(e)}", exc_info=True)
+        raise
diff --git a/src/lambda/post_processing_decompressor/requirements.txt b/src/lambda/post_processing_decompressor/requirements.txt
@@ -0,0 +1,2 @@
+# Post-processing decompressor lambda dependencies
+# Uses idp_common for document decompression
diff --git a/template.yaml b/template.yaml
@@ -3268,6 +3268,94 @@ Resources:
   ##########################################################################
   # Optional Post Processing Lambda Hook
   ##########################################################################
+  
+  # Decompression Lambda - handles document decompression before invoking custom post-processor
+  PostProcessingDecompressorDLQ:
+    Type: AWS::SQS::Queue
+    Condition: ShouldEnablePostProcessingLambdaHook
+    Properties:
+      KmsMasterKeyId: !Ref CustomerManagedEncryptionKey
+      VisibilityTimeout: 30
+      MessageRetentionPeriod: 345600 # 4 days
+
+  PostProcessingDecompressorDLQPolicy:
+    Type: AWS::SQS::QueuePolicy
+    Condition: ShouldEnablePostProcessingLambdaHook
+    Properties:
+      Queues:
+        - !Ref PostProcessingDecompressorDLQ
+      PolicyDocument:
+        Version: "2012-10-17"
+        Statement:
+          - Sid: EnforceSSLOnly
+            Effect: Deny
+            Principal: "*"
+            Action: "sqs:*"
+            Resource: !GetAtt PostProcessingDecompressorDLQ.Arn
+            Condition:
+              Bool:
+                "aws:SecureTransport": false
+
+  PostProcessingDecompressor:
+    Type: AWS::Serverless::Function
+    Condition: ShouldEnablePostProcessingLambdaHook
+    Metadata:
+      cfn_nag:
+        rules_to_suppress:
+          - id: W89
+            reason: "Function does not require VPC access as it only interacts with AWS services via APIs"
+          - id: W92
+            reason: "Function does not require reserved concurrency as it scales based on demand"
+    # checkov:skip=CKV_AWS_117: "Function does not require VPC access as it only interacts with AWS services via APIs"
+    # checkov:skip=CKV_AWS_115: "Function does not require reserved concurrency as it scales based on demand"
+    # checkov:skip=CKV_AWS_173: "Environment variables do not contain sensitive data - only configuration values like feature flags and non-sensitive settings"
+    Properties:
+      PermissionsBoundary:
+        !If [
+          HasPermissionsBoundary,
+          !Ref PermissionsBoundaryArn,
+          !Ref AWS::NoValue,
+        ]
+      CodeUri: src/lambda/post_processing_decompressor/
+      Handler: index.handler
+      Runtime: python3.12
+      Timeout: 300
+      MemorySize: 512
+      Tracing: Active
+      DeadLetterQueue:
+        Type: SQS
+        TargetArn: !GetAtt PostProcessingDecompressorDLQ.Arn
+      LoggingConfig:
+        LogGroup: !Ref PostProcessingDecompressorLogGroup
+      Environment:
+        Variables:
+          LOG_LEVEL: !Ref LogLevel
+          CUSTOM_POST_PROCESSOR_ARN: !Ref PostProcessingLambdaHookFunctionArn
+          WORKING_BUCKET: !Ref WorkingBucket
+      Policies:
+        - S3CrudPolicy:
+            BucketName: !Ref WorkingBucket
+        - Statement:
+            - Effect: Allow
+              Action:
+                - lambda:InvokeFunction
+              Resource: !Ref PostProcessingLambdaHookFunctionArn
+            - Effect: Allow
+              Action:
+                - kms:Encrypt
+                - kms:Decrypt
+                - kms:ReEncrypt*
+                - kms:GenerateDataKey*
+                - kms:DescribeKey
+              Resource: !GetAtt CustomerManagedEncryptionKey.Arn
+
+  PostProcessingDecompressorLogGroup:
+    Type: AWS::Logs::LogGroup
+    Condition: ShouldEnablePostProcessingLambdaHook
+    Properties:
+      KmsKeyId: !GetAtt CustomerManagedEncryptionKey.Arn
+      RetentionInDays: !Ref LogRetentionDays
+
   PostProcessingLambdaHookRule:
     Type: AWS::Events::Rule
     Condition: ShouldEnablePostProcessingLambdaHook
@@ -3290,17 +3378,17 @@ Resources:
             - SUCCEEDED
       State: ENABLED
       Targets:
-        - Arn: !Ref PostProcessingLambdaHookFunctionArn
-          Id: PostProcessingLambdaHookFunction
+        - Arn: !GetAtt PostProcessingDecompressor.Arn
+          Id: PostProcessingDecompressor
           RetryPolicy:
             MaximumRetryAttempts: 3
 
-  PostProcessingLambdaHookPermission:
+  PostProcessingDecompressorPermission:
     Type: AWS::Lambda::Permission
     Condition: ShouldEnablePostProcessingLambdaHook
     Properties:
       Action: lambda:InvokeFunction
-      FunctionName: !Ref PostProcessingLambdaHookFunctionArn
+      FunctionName: !Ref PostProcessingDecompressor
       Principal: !Sub "events.${AWS::URLSuffix}"
       SourceArn: !GetAtt PostProcessingLambdaHookRule.Arn
 

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,2 @@`
	`1`	`+# Post-processing decompressor lambda dependencies`
	`2`	`+# Uses idp_common for document decompression`