Replaying NICE CXone EventBridge Failed Events from Dead Letter Queues with Python

What You Will Build

• A Python service that extracts failed events from CXone EventBridge dead letter queues, reconstructs them with retry metadata and destination overrides, and replays them using atomic PUT operations.
• This implementation uses the CXone EventBridge REST API (/api/v2/eventbridge/dlq/events and /api/v2/eventbridge/topics/{topicId}/events) with OAuth2 authentication.
• The tutorial covers Python 3.10+ using httpx for HTTP transport, pydantic for schema validation, and standard logging for audit trails.

Prerequisites

• OAuth2 client credentials registered in CXone with scopes: eventbridge:read, eventbridge:write
• CXone EventBridge API version v2
• Python 3.10 or higher
• External dependencies: pip install httpx pydantic python-dotenv
• Network access to your CXone region endpoint (e.g., https://api.nicecxone.com or region-specific variant)

Authentication Setup

CXone uses standard OAuth2 client credentials flow for server-to-server API access. The token endpoint requires your client ID, client secret, and the exact scopes needed for EventBridge operations.

import httpx
import os
from typing import Optional
from dataclasses import dataclass

@dataclass
class CxoneTokenResponse:
    access_token: str
    token_type: str
    expires_in: int
    scope: str

def fetch_cxone_token(client_id: str, client_secret: str, region: str = "api.nicecxone.com") -> CxoneTokenResponse:
    """
    Retrieves an OAuth2 access token from CXone.
    Required scopes: eventbridge:read, eventbridge:write
    """
    token_url = f"https://{region}/oauth2/token"
    payload = {
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret,
        "scope": "eventbridge:read eventbridge:write"
    }
    
    with httpx.Client(timeout=10.0) as client:
        response = client.post(token_url, data=payload)
        response.raise_for_status()
        data = response.json()
        
        return CxoneTokenResponse(
            access_token=data["access_token"],
            token_type=data["token_type"],
            expires_in=data["expires_in"],
            scope=data["scope"]
        )

# Example request/response cycle
# POST /oauth2/token
# Headers: Content-Type: application/x-www-form-urlencoded
# Body: grant_type=client_credentials&client_id=YOUR_ID&client_secret=YOUR_SECRET&scope=eventbridge:read%20eventbridge:write
# Response 200 OK:
# {
#   "access_token": "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...",
#   "token_type": "Bearer",
#   "expires_in": 86400,
#   "scope": "eventbridge:read eventbridge:write"
# }

Token caching is required for production systems. Implement a simple TTL cache or use a framework like botocore-style credential providers. The example below assumes tokens are refreshed before expiration.

Implementation

Step 1: Retrieve Dead Letter Queue Events with Pagination

CXone EventBridge exposes failed events through the DLQ endpoint. The API supports pagination via limit and nextPageToken. You must handle pagination to process all failed events.

import httpx
import logging
from typing import List, Dict, Any, Optional

logger = logging.getLogger(__name__)

def fetch_dlq_events(
    base_url: str,
    token: str,
    topic_id: str,
    limit: int = 50
) -> List[Dict[str, Any]]:
    """
    Retrieves failed events from CXone EventBridge DLQ.
    OAuth Scope: eventbridge:read
    """
    all_events: List[Dict[str, Any]] = []
    next_token: Optional[str] = None
    endpoint = f"{base_url}/api/v2/eventbridge/dlq/events"
    headers = {
        "Authorization": f"Bearer {token}",
        "Accept": "application/json"
    }
    
    while True:
        params = {"topicId": topic_id, "limit": limit}
        if next_token:
            params["nextPageToken"] = next_token
            
        with httpx.Client(timeout=15.0) as client:
            try:
                response = client.get(endpoint, headers=headers, params=params)
                if response.status_code == 429:
                    retry_after = int(response.headers.get("Retry-After", 5))
                    logger.warning("Rate limited (429). Retrying in %d seconds.", retry_after)
                    continue
                response.raise_for_status()
            except httpx.HTTPStatusError as e:
                logger.error("DLQ fetch failed: %s", e.response.status_code)
                raise
                
        data = response.json()
        events = data.get("events", [])
        all_events.extend(events)
        
        next_token = data.get("nextPageToken")
        if not next_token or len(events) < limit:
            break
            
    return all_events

# Example response structure from /api/v2/eventbridge/dlq/events
# {
#   "events": [
#     {
#       "eventId": "evt_8f3a2b1c-9d4e-5f6a-7b8c-9d0e1f2a3b4c",
#       "topicId": "topic_pub_conversations",
#       "payload": {"conversationId": "conv_123", "type": "conversation:start", "timestamp": "2024-05-10T14:30:00Z"},
#       "failureReason": "HTTP 502 Bad Gateway",
#       "retryCount": 3,
#       "firstFailureTimestamp": "2024-05-10T14:30:05Z",
#       "lastFailureTimestamp": "2024-05-10T14:35:00Z"
#     }
#   ],
#   "nextPageToken": "eyJwYWdlIjoxfQ=="
# }

Step 2: Construct Replay Payloads with Retry Matrices and Destination Overrides

Failed events must be reconstructed with explicit retry tracking and destination routing directives. CXone EventBridge accepts custom headers and payload extensions for routing.

import uuid
from datetime import datetime, timezone

def build_replay_payload(
    original_event: Dict[str, Any],
    destination_override: Optional[str] = None,
    max_retries: int = 5
) -> Dict[str, Any]:
    """
    Constructs a replay-ready event payload with idempotency keys,
    retry matrices, and destination override directives.
    """
    current_retry = original_event.get("retryCount", 0) + 1
    
    if current_retry > max_retries:
        raise ValueError(f"Event {original_event['eventId']} exceeded maximum retry window of {max_retries}.")
        
    idempotency_key = f"replay-{original_event['eventId']}-v{current_retry}"
    
    replay_payload = {
        "eventId": original_event["eventId"],
        "idempotencyKey": idempotency_key,
        "originalPayload": original_event["payload"],
        "retryMatrix": {
            "currentAttempt": current_retry,
            "maxAllowed": max_retries,
            "firstFailure": original_event.get("firstFailureTimestamp"),
            "lastFailure": original_event.get("lastFailureTimestamp"),
            "failureReason": original_event.get("failureReason")
        },
        "routingDirective": {
            "destinationOverride": destination_override,
            "preserveOriginalTopic": destination_override is None
        },
        "replayMetadata": {
            "initiatedAt": datetime.now(timezone.utc).isoformat(),
            "clientVersion": "1.0.0"
        }
    }
    
    return replay_payload

# Example constructed payload
# {
#   "eventId": "evt_8f3a2b1c-9d4e-5f6a-7b8c-9d0e1f2a3b4c",
#   "idempotencyKey": "replay-evt_8f3a2b1c-9d4e-5f6a-7b8c-9d0e1f2a3b4c-v4",
#   "originalPayload": {"conversationId": "conv_123", "type": "conversation:start", "timestamp": "2024-05-10T14:30:00Z"},
#   "retryMatrix": {
#     "currentAttempt": 4,
#     "maxAllowed": 5,
#     "firstFailure": "2024-05-10T14:30:05Z",
#     "lastFailure": "2024-05-10T14:35:00Z",
#     "failureReason": "HTTP 502 Bad Gateway"
#   },
#   "routingDirective": {
#     "destinationOverride": "https://secure-replay-endpoint.example.com/webhook",
#     "preserveOriginalTopic": false
#   },
#   "replayMetadata": {
#     "initiatedAt": "2024-05-11T09:15:22.103Z",
#     "clientVersion": "1.0.0"
#   }
# }

Step 3: Validate Replay Schemas Against Bus Constraints and Maximum Replay Windows

CXone EventBridge enforces payload size limits (typically 256KB) and schema constraints. You must validate before submission to prevent bus rejection and infinite loop failures.

from pydantic import BaseModel, Field, field_validator
from datetime import datetime
from typing import Optional

class RetryMatrix(BaseModel):
    currentAttempt: int
    maxAllowed: int
    firstFailure: Optional[str]
    lastFailure: Optional[str]
    failureReason: Optional[str]

class RoutingDirective(BaseModel):
    destinationOverride: Optional[str]
    preserveOriginalTopic: bool

class ReplayPayloadSchema(BaseModel):
    eventId: str
    idempotencyKey: str
    originalPayload: dict
    retryMatrix: RetryMatrix
    routingDirective: RoutingDirective
    replayMetadata: dict

    @field_validator("retryMatrix")
    @classmethod
    def validate_retry_window(cls, v: RetryMatrix) -> RetryMatrix:
        if v.currentAttempt > v.maxAllowed:
            raise ValueError("Replay attempt exceeds maximum allowed retry window.")
        return v

    @field_validator("originalPayload")
    @classmethod
    def validate_payload_size(cls, v: dict) -> dict:
        import json
        size_bytes = len(json.dumps(v).encode("utf-8"))
        if size_bytes > 262144:  # 256KB limit
            raise ValueError("Original payload exceeds EventBridge maximum size of 256KB.")
        return v

def validate_replay_payload(payload: Dict[str, Any]) -> ReplayPayloadSchema:
    """
    Validates the constructed replay payload against CXone EventBridge constraints.
    """
    try:
        validated = ReplayPayloadSchema(**payload)
        return validated
    except Exception as e:
        logger.error("Schema validation failed: %s", str(e))
        raise

# Example validation flow
# Input: constructed replay payload
# Output: Validated ReplayPayloadSchema object
# Throws ValueError if retry window exceeded or payload size > 256KB

Step 4: Execute Atomic PUT Operations with Idempotency and Error Classification

Replay submission uses an atomic PUT request to the topic event endpoint. CXone EventBridge supports idempotency via the Idempotency-Key header. You must implement error classification to distinguish between transient failures and permanent corruption.

import time
import json

def classify_error(status_code: int, response_text: str) -> str:
    """
    Classifies HTTP errors for replay routing decisions.
    """
    if status_code in (400, 422):
        return "SCHEMA_VIOLATION"
    elif status_code == 409:
        return "DUPLICATE_DETECTED"
    elif status_code == 413:
        return "PAYLOAD_TOO_LARGE"
    elif status_code >= 500:
        return "SERVER_TRANSIENT"
    return "UNKNOWN_FAILURE"

def replay_event_to_topic(
    base_url: str,
    token: str,
    topic_id: str,
    payload: Dict[str, Any],
    max_backoff_retries: int = 3
) -> Dict[str, Any]:
    """
    Submits the replay payload to CXone EventBridge via atomic PUT.
    OAuth Scope: eventbridge:write
    """
    endpoint = f"{base_url}/api/v2/eventbridge/topics/{topic_id}/events"
    idempotency_key = payload.get("idempotencyKey", str(uuid.uuid4()))
    
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
        "Idempotency-Key": idempotency_key,
        "Accept": "application/json"
    }
    
    with httpx.Client(timeout=20.0) as client:
        for attempt in range(1, max_backoff_retries + 1):
            try:
                response = client.put(endpoint, headers=headers, json=payload)
                
                if response.status_code == 429:
                    wait_time = min(2 ** attempt, 30)
                    logger.warning("Rate limited on replay. Backing off %d seconds.", wait_time)
                    time.sleep(wait_time)
                    continue
                    
                if response.status_code in (400, 409, 413, 422):
                    error_class = classify_error(response.status_code, response.text)
                    logger.error("Fatal replay error [%s]: %s", error_class, response.text)
                    return {"status": "FAILED", "classification": error_class, "response": response.json()}
                    
                response.raise_for_status()
                return {"status": "SUCCESS", "response": response.json()}
                
            except httpx.HTTPError as e:
                logger.error("Network error on attempt %d: %s", attempt, str(e))
                if attempt == max_backoff_retries:
                    raise
                    
    return {"status": "EXHAUSTED", "classification": "MAX_RETRIES_EXCEEDED"}

# Example request/response
# PUT /api/v2/eventbridge/topics/topic_pub_conversations/events
# Headers: Authorization: Bearer <token>, Content-Type: application/json, Idempotency-Key: replay-evt_...-v4
# Body: {validated replay payload}
# Response 201 Created:
# {
#   "eventId": "evt_8f3a2b1c-9d4e-5f6a-7b8c-9d0e1f2a3b4c",
#   "status": "queued",
#   "routingStatus": "delivered",
#   "processedAt": "2024-05-11T09:15:23.450Z"
# }

Step 5: Synchronize Monitoring, Track Latency, and Generate Audit Logs

Production replay systems require latency tracking, success rate calculation, and audit logging for compliance. You will sync replay events to an external monitoring webhook and maintain a structured audit trail.

import json
from dataclasses import dataclass, asdict
from typing import Dict, Any

@dataclass
class ReplayAuditRecord:
    event_id: str
    action: str
    status: str
    error_classification: Optional[str]
    latency_ms: float
    timestamp: str
    idempotency_key: str
    destination: str

def send_monitoring_alert(webhook_url: str, audit_record: ReplayAuditRecord) -> None:
    """
    Synchronizes replay status with external monitoring via webhook callback.
    """
    alert_payload = {
        "source": "cxone-eventbridge-replayer",
        "type": "replay_status_update",
        "data": asdict(audit_record)
    }
    
    try:
        with httpx.Client(timeout=5.0) as client:
            client.post(
                webhook_url,
                json=alert_payload,
                headers={"Content-Type": "application/json"}
            )
    except httpx.RequestError as e:
        logger.warning("Failed to send monitoring alert: %s", str(e))

def process_replay_pipeline(
    base_url: str,
    token: str,
    topic_id: str,
    dlq_events: List[Dict[str, Any]],
    monitoring_webhook: str,
    destination_override: Optional[str] = None
) -> Dict[str, int]:
    """
    Orchestrates the full replay pipeline with latency tracking and audit logging.
    """
    stats = {"success": 0, "failed": 0, "skipped": 0}
    audit_log: List[ReplayAuditRecord] = []
    
    for event in dlq_events:
        start_time = time.time()
        event_id = event.get("eventId", "unknown")
        
        try:
            # Step 2: Construct payload
            replay_payload = build_replay_payload(event, destination_override)
            
            # Step 3: Validate schema
            validate_replay_payload(replay_payload)
            
            # Step 4: Execute replay
            result = replay_event_to_topic(base_url, token, topic_id, replay_payload)
            latency_ms = (time.time() - start_time) * 1000
            
            status = result.get("status", "UNKNOWN")
            classification = result.get("classification")
            
            audit_record = ReplayAuditRecord(
                event_id=event_id,
                action="REPLAY_PUT",
                status=status,
                error_classification=classification,
                latency_ms=round(latency_ms, 2),
                timestamp=datetime.now(timezone.utc).isoformat(),
                idempotency_key=replay_payload["idempotencyKey"],
                destination=destination_override or topic_id
            )
            
            if status == "SUCCESS":
                stats["success"] += 1
            else:
                stats["failed"] += 1
                
            audit_log.append(audit_record)
            send_monitoring_alert(monitoring_webhook, audit_record)
            
        except Exception as e:
            logger.error("Pipeline failure for event %s: %s", event_id, str(e))
            stats["failed"] += 1
            
    # Write audit log to file for compliance
    with open("replay_audit_log.json", "w") as f:
        json.dump([asdict(r) for r in audit_log], f, indent=2)
        
    return stats

# Example audit log entry
# {
#   "event_id": "evt_8f3a2b1c-9d4e-5f6a-7b8c-9d0e1f2a3b4c",
#   "action": "REPLAY_PUT",
#   "status": "SUCCESS",
#   "error_classification": null,
#   "latency_ms": 342.15,
#   "timestamp": "2024-05-11T09:15:23.450Z",
#   "idempotency_key": "replay-evt_8f3a2b1c-9d4e-5f6a-7b8c-9d0e1f2a3b4c-v4",
#   "destination": "topic_pub_conversations"
# }

Complete Working Example

import os
import time
import json
import logging
import uuid
from typing import List, Dict, Any, Optional
from datetime import datetime, timezone
from dataclasses import dataclass, asdict

import httpx
from pydantic import BaseModel, Field, field_validator

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

@dataclass
class CxoneTokenResponse:
    access_token: str
    token_type: str
    expires_in: int
    scope: str

@dataclass
class ReplayAuditRecord:
    event_id: str
    action: str
    status: str
    error_classification: Optional[str]
    latency_ms: float
    timestamp: str
    idempotency_key: str
    destination: str

class RetryMatrix(BaseModel):
    currentAttempt: int
    maxAllowed: int
    firstFailure: Optional[str]
    lastFailure: Optional[str]
    failureReason: Optional[str]

class RoutingDirective(BaseModel):
    destinationOverride: Optional[str]
    preserveOriginalTopic: bool

class ReplayPayloadSchema(BaseModel):
    eventId: str
    idempotencyKey: str
    originalPayload: dict
    retryMatrix: RetryMatrix
    routingDirective: RoutingDirective
    replayMetadata: dict

    @field_validator("retryMatrix")
    @classmethod
    def validate_retry_window(cls, v: RetryMatrix) -> RetryMatrix:
        if v.currentAttempt > v.maxAllowed:
            raise ValueError("Replay attempt exceeds maximum allowed retry window.")
        return v

    @field_validator("originalPayload")
    @classmethod
    def validate_payload_size(cls, v: dict) -> dict:
        size_bytes = len(json.dumps(v).encode("utf-8"))
        if size_bytes > 262144:
            raise ValueError("Original payload exceeds EventBridge maximum size of 256KB.")
        return v

def fetch_cxone_token(client_id: str, client_secret: str, region: str = "api.nicecxone.com") -> CxoneTokenResponse:
    token_url = f"https://{region}/oauth2/token"
    payload = {
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret,
        "scope": "eventbridge:read eventbridge:write"
    }
    with httpx.Client(timeout=10.0) as client:
        response = client.post(token_url, data=payload)
        response.raise_for_status()
        data = response.json()
        return CxoneTokenResponse(
            access_token=data["access_token"],
            token_type=data["token_type"],
            expires_in=data["expires_in"],
            scope=data["scope"]
        )

def fetch_dlq_events(base_url: str, token: str, topic_id: str, limit: int = 50) -> List[Dict[str, Any]]:
    all_events: List[Dict[str, Any]] = []
    next_token: Optional[str] = None
    endpoint = f"{base_url}/api/v2/eventbridge/dlq/events"
    headers = {"Authorization": f"Bearer {token}", "Accept": "application/json"}
    
    while True:
        params = {"topicId": topic_id, "limit": limit}
        if next_token:
            params["nextPageToken"] = next_token
        with httpx.Client(timeout=15.0) as client:
            try:
                response = client.get(endpoint, headers=headers, params=params)
                if response.status_code == 429:
                    retry_after = int(response.headers.get("Retry-After", 5))
                    logger.warning("Rate limited (429). Retrying in %d seconds.", retry_after)
                    continue
                response.raise_for_status()
            except httpx.HTTPStatusError as e:
                logger.error("DLQ fetch failed: %s", e.response.status_code)
                raise
        data = response.json()
        events = data.get("events", [])
        all_events.extend(events)
        next_token = data.get("nextPageToken")
        if not next_token or len(events) < limit:
            break
    return all_events

def build_replay_payload(original_event: Dict[str, Any], destination_override: Optional[str] = None, max_retries: int = 5) -> Dict[str, Any]:
    current_retry = original_event.get("retryCount", 0) + 1
    if current_retry > max_retries:
        raise ValueError(f"Event {original_event['eventId']} exceeded maximum retry window of {max_retries}.")
    idempotency_key = f"replay-{original_event['eventId']}-v{current_retry}"
    return {
        "eventId": original_event["eventId"],
        "idempotencyKey": idempotency_key,
        "originalPayload": original_event["payload"],
        "retryMatrix": {
            "currentAttempt": current_retry,
            "maxAllowed": max_retries,
            "firstFailure": original_event.get("firstFailureTimestamp"),
            "lastFailure": original_event.get("lastFailureTimestamp"),
            "failureReason": original_event.get("failureReason")
        },
        "routingDirective": {
            "destinationOverride": destination_override,
            "preserveOriginalTopic": destination_override is None
        },
        "replayMetadata": {
            "initiatedAt": datetime.now(timezone.utc).isoformat(),
            "clientVersion": "1.0.0"
        }
    }

def validate_replay_payload(payload: Dict[str, Any]) -> ReplayPayloadSchema:
    return ReplayPayloadSchema(**payload)

def classify_error(status_code: int, response_text: str) -> str:
    if status_code in (400, 422): return "SCHEMA_VIOLATION"
    elif status_code == 409: return "DUPLICATE_DETECTED"
    elif status_code == 413: return "PAYLOAD_TOO_LARGE"
    elif status_code >= 500: return "SERVER_TRANSIENT"
    return "UNKNOWN_FAILURE"

def replay_event_to_topic(base_url: str, token: str, topic_id: str, payload: Dict[str, Any], max_backoff_retries: int = 3) -> Dict[str, Any]:
    endpoint = f"{base_url}/api/v2/eventbridge/topics/{topic_id}/events"
    idempotency_key = payload.get("idempotencyKey", str(uuid.uuid4()))
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
        "Idempotency-Key": idempotency_key,
        "Accept": "application/json"
    }
    with httpx.Client(timeout=20.0) as client:
        for attempt in range(1, max_backoff_retries + 1):
            try:
                response = client.put(endpoint, headers=headers, json=payload)
                if response.status_code == 429:
                    wait_time = min(2 ** attempt, 30)
                    logger.warning("Rate limited on replay. Backing off %d seconds.", wait_time)
                    continue
                if response.status_code in (400, 409, 413, 422):
                    error_class = classify_error(response.status_code, response.text)
                    return {"status": "FAILED", "classification": error_class, "response": response.json()}
                response.raise_for_status()
                return {"status": "SUCCESS", "response": response.json()}
            except httpx.HTTPError as e:
                logger.error("Network error on attempt %d: %s", attempt, str(e))
                if attempt == max_backoff_retries:
                    raise
    return {"status": "EXHAUSTED", "classification": "MAX_RETRIES_EXCEEDED"}

def send_monitoring_alert(webhook_url: str, audit_record: ReplayAuditRecord) -> None:
    alert_payload = {"source": "cxone-eventbridge-replayer", "type": "replay_status_update", "data": asdict(audit_record)}
    try:
        with httpx.Client(timeout=5.0) as client:
            client.post(webhook_url, json=alert_payload, headers={"Content-Type": "application/json"})
    except httpx.RequestError as e:
        logger.warning("Failed to send monitoring alert: %s", str(e))

def run_replay_service(
    client_id: str,
    client_secret: str,
    region: str,
    topic_id: str,
    monitoring_webhook: str,
    destination_override: Optional[str] = None
) -> Dict[str, int]:
    base_url = f"https://{region}"
    token_resp = fetch_cxone_token(client_id, client_secret, region)
    dlq_events = fetch_dlq_events(base_url, token_resp.access_token, topic_id)
    
    stats = {"success": 0, "failed": 0, "skipped": 0}
    audit_log: List[ReplayAuditRecord] = []
    
    for event in dlq_events:
        start_time = time.time()
        event_id = event.get("eventId", "unknown")
        try:
            replay_payload = build_replay_payload(event, destination_override)
            validate_replay_payload(replay_payload)
            result = replay_event_to_topic(base_url, token_resp.access_token, topic_id, replay_payload)
            latency_ms = (time.time() - start_time) * 1000
            status = result.get("status", "UNKNOWN")
            classification = result.get("classification")
            audit_record = ReplayAuditRecord(
                event_id=event_id, action="REPLAY_PUT", status=status,
                error_classification=classification, latency_ms=round(latency_ms, 2),
                timestamp=datetime.now(timezone.utc).isoformat(),
                idempotency_key=replay_payload["idempotencyKey"],
                destination=destination_override or topic_id
            )
            if status == "SUCCESS":
                stats["success"] += 1
            else:
                stats["failed"] += 1
            audit_log.append(audit_record)
            send_monitoring_alert(monitoring_webhook, audit_record)
        except Exception as e:
            logger.error("Pipeline failure for event %s: %s", event_id, str(e))
            stats["failed"] += 1
            
    with open("replay_audit_log.json", "w") as f:
        json.dump([asdict(r) for r in audit_log], f, indent=2)
    return stats

if __name__ == "__main__":
    results = run_replay_service(
        client_id=os.getenv("CXONE_CLIENT_ID"),
        client_secret=os.getenv("CXONE_CLIENT_SECRET"),
        region=os.getenv("CXONE_REGION", "api.nicecxone.com"),
        topic_id=os.getenv("CXONE_TOPIC_ID"),
        monitoring_webhook=os.getenv("MONITORING_WEBHOOK_URL"),
        destination_override=os.getenv("DESTINATION_OVERRIDE_URL")
    )
    logger.info("Replay pipeline completed. Stats: %s", results)

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: Expired OAuth token, incorrect client credentials, or missing eventbridge:read/eventbridge:write scopes.
• Fix: Verify token expiration timestamps. Implement automatic token refresh before the expires_in window closes. Ensure the client ID and secret match a CXone application with EventBridge permissions.
• Code Fix: Add token TTL tracking in fetch_cxone_token and cache the token until expires_in - 60 seconds.

Error: 403 Forbidden

• Cause: The OAuth client lacks permission to access the specified topic or DLQ, or the region endpoint is incorrect.
• Fix: Confirm the topic ID exists in your CXone organization. Verify the OAuth client is granted eventbridge:read and eventbridge:write in the CXone admin console. Check that the region URL matches your deployment.

Error: 429 Too Many Requests

• Cause: Exceeding CXone EventBridge rate limits during DLQ polling or replay submission.
• Fix: Implement exponential backoff. The provided code includes Retry-After header parsing and backoff logic in fetch_dlq_events and replay_event_to_topic. Increase the initial backoff delay if cascading failures occur.

Error: 400 Bad Request or 422 Unprocessable Entity

• Cause: Payload schema mismatch, missing required fields, or exceeding the 256KB payload limit.
• Fix: Run payloads through ReplayPayloadSchema validation before submission. Ensure originalPayload is a valid JSON object. Check that retryMatrix.currentAttempt does not exceed maxAllowed.

Error: 409 Conflict

• Cause: Duplicate idempotency key submission. CXone EventBridge rejects replay attempts that reuse an existing idempotency key within the retention window.
• Fix: Increment the version suffix in the idempotency key (replay-{eventId}-v{N}). The code handles this by appending the retry count to the key.

Official References

• NICE CXone EventBridge API Documentation
• CXone OAuth2 Authentication Guide
• EventBridge Dead Letter Queue Management
• Idempotency and Retry Best Practices for CXone APIs