""" HTTP client for Amazon's Shuttle ConsumeConsultation API. POSTs the mapped consultation response with a cached Cognito Bearer token and returns a typed DeliveryResult. Retry contract (spec §7-§8): - 408 / 429 / 5xx / network / timeout are transient: retried with exponential backoff plus +-25% jitter, same payload and trackingId. - 401: refresh the token once and retry immediately; a second 401 is a hard auth failure (never loop on auth). - 403, 400, and HTTP 200 with body status ERRORED are terminal: the payload or configuration must be fixed before resending (with a new trackingId) — the client never retries them. Every request and response is also logged as one structured JSON line carrying the identifiers Amazon traces calls by. """ import asyncio import json import logging import random import time from dataclasses import dataclass from enum import Enum from typing import Any, Dict, Optional import aiohttp from .amazon_auth import AmazonAuthError, CognitoTokenProvider logger = logging.getLogger(__name__) class DeliveryOutcome(Enum): ACKNOWLEDGED = "acknowledged" ERRORED = "errored" # Amazon rejected the payload; fix + resend manually AUTH_FAILED = "auth_failed" # credentials/scope problem; alert, no retry TRANSIENT_EXHAUSTED = "transient_exhausted" # retries used up @dataclass class DeliveryResult: outcome: DeliveryOutcome http_status: Optional[int] = None error_message: Optional[str] = None acknowledgement_timestamp: Optional[int] = None attempts: int = 0 @property def acknowledged(self) -> bool: return self.outcome is DeliveryOutcome.ACKNOWLEDGED def _log_consume_event( event: str, payload: Dict[str, Any], *, http_status: Optional[int] = None, body_status: Optional[str] = None, acknowledgement_timestamp: Optional[int] = None, error_message: Optional[str] = None, attempt: Optional[int] = None, ) -> None: """One JSON line per request/response with the fields Amazon traces by (spec FAQ: ids, attempt number, HTTP + body status, ack timestamp, errorMessage, IST epoch ms).""" message_ids = payload.get("messageIdentifiers") or {} consultation = (payload.get("data") or {}).get("consultationResponse") or {} record = { "event": event, "primaryReferenceId": message_ids.get("primaryReferenceId"), "secondaryReferenceId": message_ids.get("secondaryReferenceId"), "trackingId": message_ids.get("trackingId"), "providerName": message_ids.get("providerName"), "customerCallAttemptNumber": consultation.get("customerCallAttemptNumber"), "attempt": attempt, "httpStatus": http_status, "status": body_status, "acknowledgementTimeStamp": acknowledgement_timestamp, "errorMessage": error_message, "timestampIst": int(time.time() * 1000), } logger.info(json.dumps({k: v for k, v in record.items() if v is not None})) class AmazonConsumeClient: def __init__(self, config, token_provider: Optional[CognitoTokenProvider] = None): self._config = config self._token_provider = token_provider or CognitoTokenProvider.from_config( config ) self._url = config.amazon_consultation_response_url self._timeout = aiohttp.ClientTimeout( total=getattr(config, "amazon_api_timeout_seconds", 30.0) ) self._attempts = max(1, getattr(config, "amazon_api_retry_attempts", 3)) self._base_delay = getattr(config, "amazon_api_retry_delay_seconds", 1.0) async def send_consultation_response( self, payload: Dict[str, Any] ) -> DeliveryResult: """Deliver one consultation response; never raises on delivery failure — the caller routes on the returned outcome.""" attempts_used = 0 refreshed_after_401 = False async with aiohttp.ClientSession(timeout=self._timeout) as session: attempt = 0 while attempt < self._attempts: attempt += 1 attempts_used = attempt try: token = await self._token_provider.get_token() except AmazonAuthError as e: if e.retryable and attempt < self._attempts: await self._backoff(attempt) continue return DeliveryResult( outcome=DeliveryOutcome.AUTH_FAILED, error_message=str(e), attempts=attempts_used, ) _log_consume_event("consume_api_request", payload, attempt=attempt) try: async with session.post( self._url, json=payload, headers={ "Authorization": f"Bearer {token}", "Content-Type": "application/json", }, ) as response: result = await self._interpret_response( response, payload, attempt, attempts_used ) except (aiohttp.ClientError, asyncio.TimeoutError) as e: logger.error( "Amazon Consume API call failed (attempt %d/%d): %s", attempt, self._attempts, e, ) result = "transient" if isinstance(result, DeliveryResult): return result if result == "refresh_token": if refreshed_after_401: return DeliveryResult( outcome=DeliveryOutcome.AUTH_FAILED, http_status=401, error_message="401 persisted after token refresh", attempts=attempts_used, ) refreshed_after_401 = True self._token_provider.invalidate() # The refresh-and-retry is Amazon's prescribed 401 # handling, not one of the transient retries. attempt -= 1 continue # transient if attempt < self._attempts: await self._backoff(attempt) return DeliveryResult( outcome=DeliveryOutcome.TRANSIENT_EXHAUSTED, error_message=f"no acknowledgement after {attempts_used} attempts", attempts=attempts_used, ) async def _interpret_response(self, response, payload, attempt, attempts_used): """DeliveryResult for a terminal response, 'refresh_token' for a first 401, 'transient' for a retryable status.""" status = response.status if status == 200: try: body = await response.json(content_type=None) except Exception: body = {} body_status = (body or {}).get("status") ack_ts = (body or {}).get("acknowledgementTimeStamp") error_message = (body or {}).get("errorMessage") _log_consume_event( "consume_api_response", payload, http_status=status, body_status=body_status, acknowledgement_timestamp=ack_ts, error_message=error_message, attempt=attempt, ) if body_status == "ACKNOWLEDGED": return DeliveryResult( outcome=DeliveryOutcome.ACKNOWLEDGED, http_status=status, acknowledgement_timestamp=ack_ts, attempts=attempts_used, ) return DeliveryResult( outcome=DeliveryOutcome.ERRORED, http_status=status, error_message=error_message or f"unexpected body status {body_status}", acknowledgement_timestamp=ack_ts, attempts=attempts_used, ) error_text = await response.text() _log_consume_event( "consume_api_response", payload, http_status=status, error_message=error_text[:300], attempt=attempt, ) if status == 401: return "refresh_token" if status == 403: return DeliveryResult( outcome=DeliveryOutcome.AUTH_FAILED, http_status=status, error_message=f"forbidden (wrong scope?): {error_text[:300]}", attempts=attempts_used, ) if status in (408, 429) or status >= 500: logger.error( "Amazon Consume API HTTP %d (attempt %d/%d): %s", status, attempt, self._attempts, error_text[:300], ) return "transient" # 400 and any other unexpected 4xx: the payload must be fixed. return DeliveryResult( outcome=DeliveryOutcome.ERRORED, http_status=status, error_message=error_text[:300], attempts=attempts_used, ) async def _backoff(self, attempt: int) -> None: delay = self._base_delay * (2 ** (attempt - 1)) delay *= random.uniform(0.75, 1.25) logger.info("Retrying Amazon Consume API call in %.2fs", delay) await asyncio.sleep(delay)