Orchestrating Parallel Filing Batches Across Multiple Jurisdictions #

import asyncio
import hashlib
import json
import logging
import time
from dataclasses import dataclass
from enum import Enum
from typing import AsyncGenerator, Dict, Optional

import aiohttp
from aiohttp import ClientSession, TCPConnector

# Structured JSON logging configuration
logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger("compliance.orchestrator")

class FilingStatus(Enum):
    PENDING = "pending"
    SUBMITTED = "submitted"
    ACCEPTED = "accepted"
    REJECTED = "rejected"
    FALLOUT = "fallout"

@dataclass(frozen=True)
class JurisdictionConfig:
    code: str
    max_concurrency: int
    session_timeout_sec: int
    base_url: str
    requires_captcha_threshold: int = 3

@dataclass(frozen=True)
class FilingEntity:
    entity_id: str
    jurisdiction: str
    payload: Dict[str, str]
    idempotency_key: str

@dataclass
class AuditRecord:
    entity_id: str
    jurisdiction: str
    status: FilingStatus
    trace_id: str
    payload_hash: str
    timestamp: float
    error_detail: Optional[str] = None

    def to_json(self) -> str:
        return json.dumps(self.__dict__, default=str)

class StructuredAuditLogger:
    @staticmethod
    def record(record: AuditRecord) -> None:
        log_entry = {
            "level": "INFO" if record.status != FilingStatus.REJECTED else "ERROR",
            "trace_id": record.trace_id,
            "entity_id": record.entity_id,
            "jurisdiction": record.jurisdiction,
            "status": record.status.value,
            "payload_sha256": record.payload_hash,
            "ts": record.timestamp,
            "error": record.error_detail
        }
        logger.info(json.dumps(log_entry))

class BatchOrchestrator:
    def __init__(self, configs: Dict[str, JurisdictionConfig]):
        self.configs = configs
        self.semaphores = {code: asyncio.Semaphore(cfg.max_concurrency) for code, cfg in configs.items()}
        self.session: Optional[ClientSession] = None
        self.audit_logger = StructuredAuditLogger()
        self.fallout_queue: asyncio.Queue[AuditRecord] = asyncio.Queue()

    async def initialize(self) -> None:
        connector = TCPConnector(limit_per_host=100, ttl_dns_cache=300)
        self.session = ClientSession(connector=connector, timeout=aiohttp.ClientTimeout(total=45))

    async def close(self) -> None:
        if self.session:
            await self.session.close()

    def _compute_payload_hash(self, payload: Dict[str, str]) -> str:
        canonical = json.dumps(payload, sort_keys=True, ensure_ascii=False)
        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

    def _invalidate_cache_headers(self, headers: Dict[str, str]) -> Dict[str, str]:
        """Force portal cache bypass and prevent stale session tokens."""
        headers.update({
            "Cache-Control": "no-cache, no-store, must-revalidate",
            "Pragma": "no-cache",
            "If-None-Match": "*",
            "Accept-Encoding": "identity"
        })
        return headers

    async def _submit_with_fallback(
        self, entity: FilingEntity, config: JurisdictionConfig
    ) -> AuditRecord:
        trace_id = f"{entity.entity_id}-{int(time.time())}"
        payload_hash = self._compute_payload_hash(entity.payload)
        headers = self._invalidate_cache_headers({
            "Content-Type": "application/json; charset=utf-8",
            "X-Idempotency-Key": entity.idempotency_key,
            "User-Agent": "ComplianceOrchestrator/2.1"
        })

        attempt = 0
        max_retries = 3
        last_error = None

        while attempt < max_retries:
            try:
                async with self.semaphores[entity.jurisdiction]:
                    async with self.session.post(
                        f"{config.base_url}/submit",
                        headers=headers,
                        json=entity.payload,
                        timeout=aiohttp.ClientTimeout(total=config.session_timeout_sec)
                    ) as resp:
                        if resp.status == 200:
                            return AuditRecord(
                                entity_id=entity.entity_id,
                                jurisdiction=entity.jurisdiction,
                                status=FilingStatus.ACCEPTED,
                                trace_id=trace_id,
                                payload_hash=payload_hash,
                                timestamp=time.time()
                            )
                        elif resp.status == 429:
                            await asyncio.sleep(2 ** attempt + 0.5)
                            attempt += 1
                            continue
                        elif resp.status in (400, 409, 422):
                            body = await resp.text()
                            return AuditRecord(
                                entity_id=entity.entity_id,
                                jurisdiction=entity.jurisdiction,
                                status=FilingStatus.REJECTED,
                                trace_id=trace_id,
                                payload_hash=payload_hash,
                                timestamp=time.time(),
                                error_detail=f"Portal validation failed: {body}"
                            )
                        else:
                            raise RuntimeError(f"Unexpected status {resp.status}")
            except (aiohttp.ClientError, asyncio.TimeoutError) as e:
                last_error = str(e)
                await asyncio.sleep(2 ** attempt)
                attempt += 1

        # Fallback chain: route to manual processing queue
        record = AuditRecord(
            entity_id=entity.entity_id,
            jurisdiction=entity.jurisdiction,
            status=FilingStatus.FALLOUT,
            trace_id=trace_id,
            payload_hash=payload_hash,
            timestamp=time.time(),
            error_detail=f"Max retries exceeded. Routed to manual queue. Last error: {last_error}"
        )
        await self.fallout_queue.put(record)
        return record

    async def stream_entities(self, source: AsyncGenerator[FilingEntity, None]) -> None:
        tasks = []
        async for entity in source:
            config = self.configs.get(entity.jurisdiction)
            if not config:
                logger.warning(f"Unknown jurisdiction {entity.jurisdiction}. Skipping.")
                continue
            tasks.append(asyncio.create_task(self._submit_with_fallback(entity, config)))
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        for res in results:
            if isinstance(res, AuditRecord):
                self.audit_logger.record(res)
            else:
                logger.error(f"Unhandled orchestration exception: {res}")