Deduplicating Genesys Cloud EventBridge Events in Python

What You Will Build

• A Python Lambda function that receives events from Genesys Cloud via AWS EventBridge, identifies duplicates using a deterministic ID, and processes only unique events.
• This tutorial uses the AWS Lambda runtime and the boto3 SDK to interact with downstream resources.
• The implementation is written in Python 3.9+ using standard library modules for hashing and time management.

Prerequisites

• An AWS account with permissions to create Lambda functions and EventBridge rules.
• A Genesys Cloud organization with an active EventBridge integration configured.
• Python 3.9 or higher installed locally for testing.
• Dependencies: boto3 (for AWS SDK), requests (for local testing if needed).

Authentication Setup

This integration relies on the server-to-server trust established between Genesys Cloud and AWS EventBridge. No OAuth tokens are passed within the EventBridge event payload. The authentication is handled at the infrastructure level via the EventBridge Partner Event Source.

However, if your Lambda function needs to call the Genesys Cloud API to fetch additional context (e.g., user details), you will need to implement a token refresh mechanism. For the scope of deduplication, we assume the event payload contains all necessary data.

Required Scopes (if calling GC API from Lambda):

• analytics:conversation:read
• user:read

Implementation

Step 1: Understanding the Event Structure

Genesys Cloud sends events to EventBridge in the aws.partner/genesys.cloud source format. A critical field for deduplication is the detail object, which contains the specific payload from Genesys Cloud.

The event structure looks like this:

{
  "version": "0",
  "id": "e1cb837c-9395-4c40-b245-b08833002175",
  "detail-type": "Genesys Cloud Event",
  "source": "aws.partner/genesys.cloud",
  "account": "123456789012",
  "time": "2023-10-27T10:00:00Z",
  "region": "us-east-1",
  "resources": [],
  "detail": {
    "organizationId": "abc123",
    "eventType": "conversation",
    "eventTime": "2023-10-27T09:59:50Z",
    "data": {
      "conversationId": "conv-12345678-1234-1234-1234-123456789012",
      "type": "chat",
      "state": "closed"
    }
  }
}

Key Insight: The id field in the EventBridge envelope is unique per delivery attempt, but Genesys Cloud may retry sending the same event if it does not receive a confirmation. Therefore, you cannot deduplicate based on the EventBridge id. You must deduplicate based on the content of the detail object.

Step 2: Creating the Deduplication Key

To deduplicate, we need a deterministic hash of the event content. We will create a function that extracts the relevant fields from the detail object and generates a SHA-256 hash.

Why SHA-256? It provides a sufficient collision resistance for practical purposes and is standard in the Python hashlib library.

import hashlib
import json
import logging

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def generate_event_signature(event_detail: dict) -> str:
    """
    Generates a deterministic SHA-256 hash of the event detail.
    This hash serves as the unique identifier for deduplication.
    
    Args:
        event_detail (dict): The 'detail' object from the EventBridge event.
        
    Returns:
        str: Hexadecimal string of the SHA-256 hash.
    """
    # Sort keys to ensure consistent hashing regardless of JSON key order
    canonical_json = json.dumps(event_detail, sort_keys=True, default=str)
    return hashlib.sha256(canonical_json.encode('utf-8')).hexdigest()

Critical Note: The sort_keys=True parameter is essential. JSON objects are unordered maps. If Genesys Cloud sends the same event with keys in a different order (which can happen across retries or different microservices), the hash will differ unless the keys are sorted before hashing.

Step 3: Implementing the Deduplication Store

For a production-grade solution, you need a persistent store to track processed events. For this tutorial, we will use Amazon DynamoDB as it provides low-latency reads/writes and is cost-effective for high-volume event streams.

DynamoDB Table Schema:

• Partition Key: event_hash (String)
• Sort Key: processed_at (String, ISO 8601 format)
• Item: Contains the full event detail for audit purposes.

We will create a helper class to interact with DynamoDB.

import boto3
from botocore.exceptions import ClientError
from datetime import datetime, timezone

class DeduplicationStore:
    def __init__(self, table_name: str):
        self.dynamodb = boto3.resource('dynamodb')
        self.table = self.dynamodb.Table(table_name)

    def is_duplicate(self, event_hash: str, ttl_seconds: int = 3600) -> bool:
        """
        Checks if an event hash already exists in the table.
        If it exists, returns True. If not, adds it to the table and returns False.
        
        Args:
            event_hash (str): The SHA-256 hash of the event.
            ttl_seconds (int): Time-to-live for the deduplication record. 
                               Defaults to 1 hour.
            
        Returns:
            bool: True if the event was a duplicate, False if it was new.
        """
        current_time = datetime.now(timezone.utc).isoformat()
        
        try:
            # Use a conditional write to check existence and insert in one atomic operation
            # If the item already exists, the condition fails and raises a ClientError
            self.table.put_item(
                Item={
                    'event_hash': event_hash,
                    'processed_at': current_time,
                    'ttl': int(datetime.now(timezone.utc).timestamp()) + ttl_seconds
                },
                ConditionExpression="attribute_not_exists(event_hash)"
            )
            return False # New event
        except ClientError as e:
            if e.response['Error']['Code'] == 'ConditionalCheckFailedException':
                return True # Duplicate found
            else:
                logger.error(f"Unexpected DynamoDB error: {e}")
                raise

    def cleanup_expired(self, batch_size: int = 25):
        """
        Optional: Periodic cleanup of expired items if TTL is not enabled on the table.
        For production, enable TTL on the 'ttl' attribute in DynamoDB settings.
        """
        # Implementation omitted for brevity. 
        # In production, enable DynamoDB TTL feature on the 'ttl' attribute.
        pass

Step 4: The Lambda Handler

Now we combine the signature generation and the store into the main Lambda handler.

import json
import logging
from typing import Any, Dict

logger = logging.getLogger()
logger.setLevel(logging.INFO)

# Initialize the store globally to reuse connections across invocations
DEDUP_TABLE_NAME = "GenesysEventDeduplication"
store = DeduplicationStore(DEDUP_TABLE_NAME)

def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
    """
    AWS Lambda handler for processing Genesys Cloud events from EventBridge.
    
    Args:
        event (dict): The EventBridge event.
        context (object): The Lambda runtime context.
        
    Returns:
        dict: Standard Lambda response.
    """
    logger.info(f"Received event ID: {event.get('id')}")
    
    # 1. Validate the event source
    if event.get('source') != 'aws.partner/genesys.cloud':
        logger.warning("Ignoring event from unknown source")
        return {
            'statusCode': 200,
            'body': json.dumps('Ignored non-Genesys event')
        }
    
    # 2. Extract the detail payload
    detail = event.get('detail')
    if not detail:
        logger.error("Event missing 'detail' field")
        raise ValueError("Invalid event structure: missing 'detail'")
    
    # 3. Generate the deduplication key
    event_hash = generate_event_signature(detail)
    logger.info(f"Generated event hash: {event_hash}")
    
    # 4. Check for duplicates
    is_dup = store.is_duplicate(event_hash, ttl_seconds=3600)
    
    if is_dup:
        logger.info(f"Duplicate event detected: {event_hash}. Skipping processing.")
        return {
            'statusCode': 200,
            'body': json.dumps('Duplicate event skipped')
        }
    
    # 5. Process the unique event
    try:
        process_event(detail)
        logger.info(f"Successfully processed event: {event_hash}")
    except Exception as e:
        logger.error(f"Failed to process event {event_hash}: {str(e)}")
        # Re-raise to trigger retry or DLQ
        raise
    
    return {
        'statusCode': 200,
        'body': json.dumps('Event processed successfully')
    }

def process_event(detail: Dict[str, Any]) -> None:
    """
    Placeholder for your business logic.
    
    Args:
        detail (dict): The Genesys Cloud event detail.
    """
    event_type = detail.get('eventType')
    data = detail.get('data', {})
    
    logger.info(f"Processing event type: {event_type}")
    
    # Example: Send to another service, update a database, trigger a workflow
    if event_type == 'conversation':
        conversation_id = data.get('conversationId')
        logger.info(f"Handling conversation event for ID: {conversation_id}")
        # Your logic here
        pass
    else:
        logger.warning(f"Unhandled event type: {event_type}")

Step 5: Handling Edge Cases and Retries

Scenario: Network Timeout During Processing
If process_event fails after the event has been added to the DynamoDB table, the event will be marked as processed. If Lambda retries the invocation (due to a timeout), the is_duplicate check will return True, and the event will be skipped.

Risk: If the failure occurred before the business logic completed but after the DynamoDB write, you lose the event.

Mitigation: Use a two-phase commit pattern or a transactional outbox pattern if eventual consistency is not acceptable. For most real-time analytics or notification use cases, the simple “write then process” pattern above is sufficient because Genesys Cloud will not re-send the exact same event payload indefinitely. It typically retries only on network errors.

Scenario: High Throughput
If you expect >1000 events per second, DynamoDB conditional writes may become a bottleneck. Consider:

1. Increasing DynamoDB WCU.
2. Using DynamoDB Streams to process events asynchronously after ingestion.
3. Using a distributed cache like ElastiCache (Redis) with SETNX (Set if Not Exists) for faster deduplication, with a shorter TTL (e.g., 5 minutes).

Complete Working Example

Here is the full, copy-pasteable Python module. Save this as lambda_function.py.

import hashlib
import json
import logging
import boto3
from botocore.exceptions import ClientError
from datetime import datetime, timezone
from typing import Any, Dict

# Configure logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# Constants
DEDUP_TABLE_NAME = "GenesysEventDeduplication"
DEDUP_TTL_SECONDS = 3600  # 1 hour

# Global DynamoDB resource for connection reuse
dynamodb = boto3.resource('dynamodb')
dedup_table = dynamodb.Table(DEDUP_TABLE_NAME)

def generate_event_signature(event_detail: dict) -> str:
    """
    Generates a deterministic SHA-256 hash of the event detail.
    """
    try:
        canonical_json = json.dumps(event_detail, sort_keys=True, default=str)
        return hashlib.sha256(canonical_json.encode('utf-8')).hexdigest()
    except Exception as e:
        logger.error(f"Failed to generate signature: {e}")
        raise

def check_and_insert_dedup(event_hash: str) -> bool:
    """
    Checks if the event hash exists. If not, inserts it.
    Returns True if duplicate, False if new.
    """
    current_time = datetime.now(timezone.utc).isoformat()
    ttl_value = int(datetime.now(timezone.utc).timestamp()) + DEDUP_TTL_SECONDS
    
    try:
        dedup_table.put_item(
            Item={
                'event_hash': event_hash,
                'processed_at': current_time,
                'ttl': ttl_value
            },
            ConditionExpression="attribute_not_exists(event_hash)"
        )
        return False
    except ClientError as e:
        if e.response['Error']['Code'] == 'ConditionalCheckFailedException':
            return True
        logger.error(f"DynamoDB error: {e}")
        raise

def process_event(detail: Dict[str, Any]) -> None:
    """
    Implement your business logic here.
    """
    event_type = detail.get('eventType')
    data = detail.get('data', {})
    
    logger.info(f"Processing {event_type} event with ID: {data.get('id', 'N/A')}")
    
    # Example: Log to CloudWatch
    logger.info(f"Event Data: {json.dumps(data)}")
    
    # Example: Call another AWS service
    # s3 = boto3.client('s3')
    # s3.put_object(Bucket='my-bucket', Key=f'events/{event_hash}.json', Body=json.dumps(detail))

def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
    """
    Main Lambda handler.
    """
    logger.info("Start processing event")
    
    # 1. Validate Source
    if event.get('source') != 'aws.partner/genesys.cloud':
        return {'statusCode': 200, 'body': 'Ignored'}
    
    # 2. Extract Detail
    detail = event.get('detail')
    if not detail:
        raise ValueError("Missing 'detail' in event")
    
    # 3. Generate Hash
    event_hash = generate_event_signature(detail)
    
    # 4. Deduplication Check
    is_duplicate = check_and_insert_dedup(event_hash)
    
    if is_duplicate:
        logger.info(f"Duplicate event {event_hash} ignored")
        return {'statusCode': 200, 'body': 'Duplicate'}
    
    # 5. Process
    try:
        process_event(detail)
    except Exception as e:
        logger.error(f"Processing failed: {e}")
        raise
    
    return {'statusCode': 200, 'body': 'Success'}

Common Errors & Debugging

Error: ConditionalCheckFailedException

Cause: The DynamoDB put_item with ConditionExpression failed because the item already exists.
Fix: This is expected behavior for deduplication. Ensure your code catches this specific ClientError and returns True for is_duplicate.

Error: Hash Mismatch Despite Same Content

Cause: Floating-point precision differences or non-deterministic JSON serialization (e.g., different key orders).
Fix: Always use sort_keys=True in json.dumps. If your data contains floats, consider rounding them to a fixed precision before hashing.

Error: Lambda Timeout

Cause: The process_event function takes longer than the Lambda timeout setting.
Fix: Increase the Lambda timeout in the AWS console. If processing is heavy, consider offloading to SQS or Step Functions. Note that if the Lambda times out after the DynamoDB write, the event is considered processed. To avoid data loss, ensure your business logic is idempotent or use a transactional pattern.

Error: EventBridge Delivery Failure

Cause: Genesys Cloud does not receive a 200 OK response from the EventBridge rule target.
Fix: Ensure your Lambda returns a valid JSON response with statusCode: 200. If the Lambda throws an unhandled exception, it returns a non-200 status, triggering retries from EventBridge. This will cause the same event to be re-processed, but our deduplication logic will handle it.

Official References

• AWS EventBridge Partner Events Documentation
• Genesys Cloud EventBridge Integration Guide
• Boto3 DynamoDB Documentation
• Python hashlib Documentation