Orchestrating Parallel Filing Batches Across Multiple Jurisdictions #

from __future__ import annotations

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

import aiohttp
from aiohttp import ClientSession, TCPConnector

# Structured JSON logging — every line is a parseable audit/observability event.
logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger("orchestration.parallel_dispatch")


class FilingStatus(Enum):
    ACCEPTED = "accepted"      # registry returned 200
    REJECTED = "rejected"      # portal validation failure — terminal, never retried
    FALLOUT = "fallout"        # transient retries exhausted — routed to manual queue


@dataclass(frozen=True)
class JurisdictionConfig:
    code: str                 # e.g. "US-DE"
    max_concurrency: int      # per-jurisdiction semaphore ceiling, tuned to portal tolerance
    session_timeout_sec: int
    base_url: str


@dataclass(frozen=True)
class FilingEntity:
    entity_id: str
    jurisdiction: str
    payload: dict[str, str]
    idempotency_key: str      # registry-side dedup contract; stable across retries


@dataclass(frozen=True)
class AuditRecord:
    entity_id: str
    jurisdiction: str
    status: FilingStatus
    trace_id: str
    payload_sha256: str       # tamper-evident proof of exactly what was transmitted
    timestamp: float
    error_detail: Optional[str] = None


class ParallelDispatcher:
    def __init__(self, configs: dict[str, JurisdictionConfig], *, max_retries: int = 3) -> None:
        self.configs = configs
        self.max_retries = max_retries
        # One semaphore per jurisdiction: Delaware and California run in parallel,
        # but neither ever exceeds the concurrency its own portal will tolerate.
        self.semaphores = {
            code: asyncio.Semaphore(cfg.max_concurrency) for code, cfg in configs.items()
        }
        self.session: Optional[ClientSession] = None
        self.fallout_queue: asyncio.Queue[AuditRecord] = asyncio.Queue()

    async def __aenter__(self) -> "ParallelDispatcher":
        # limit_per_host keeps the connection pool jurisdiction-isolated to avoid
        # cookie/CSRF-token collisions across portals sharing the same connector.
        connector = TCPConnector(limit_per_host=8, ttl_dns_cache=300)
        self.session = ClientSession(
            connector=connector, timeout=aiohttp.ClientTimeout(total=45)
        )
        return self

    async def __aexit__(self, *exc: object) -> None:
        if self.session:
            await self.session.close()

    @staticmethod
    def _payload_hash(payload: dict[str, str]) -> str:
        # Canonical JSON (sorted keys) so the hash is reproducible at audit time.
        canonical = json.dumps(payload, sort_keys=True, ensure_ascii=False)
        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

    def _headers(self, entity: FilingEntity) -> dict[str, str]:
        return {
            "Content-Type": "application/json; charset=utf-8",
            "X-Idempotency-Key": entity.idempotency_key,  # registry rejects dup submissions
            "Cache-Control": "no-store",                  # never trust a cached 304 status
            "User-Agent": "ComplianceOrchestrator/2.1",
        }

    async def _dispatch_one(self, entity: FilingEntity, cfg: JurisdictionConfig) -> AuditRecord:
        trace_id = f"{entity.entity_id}-{int(time.time())}"
        payload_hash = self._payload_hash(entity.payload)
        last_error: Optional[str] = None

        for attempt in range(self.max_retries):
            try:
                # The semaphore is the whole game: it bounds in-flight requests to
                # this jurisdiction so a burst never trips its rate limiter.
                async with self.semaphores[entity.jurisdiction]:
                    async with self.session.post(  # type: ignore[union-attr]
                        f"{cfg.base_url}/submit",
                        headers=self._headers(entity),
                        json=entity.payload,
                        timeout=aiohttp.ClientTimeout(total=cfg.session_timeout_sec),
                    ) as resp:
                        if resp.status == 200:
                            return AuditRecord(entity.entity_id, entity.jurisdiction,
                                               FilingStatus.ACCEPTED, trace_id,
                                               payload_hash, time.time())
                        if resp.status == 429:  # transient: back off and retry
                            await asyncio.sleep(2 ** attempt + 0.5)
                            continue
                        if resp.status in (400, 409, 422):  # statutory/data — terminal
                            body = await resp.text()
                            return AuditRecord(entity.entity_id, entity.jurisdiction,
                                               FilingStatus.REJECTED, trace_id,
                                               payload_hash, time.time(),
                                               f"Portal validation failed: {body[:500]}")
                        raise RuntimeError(f"Unexpected status {resp.status}")
            except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
                last_error = str(exc)               # transient transport fault
                await asyncio.sleep(2 ** attempt)

        # Retries exhausted: never silently drop — preserve for forensic replay.
        record = AuditRecord(entity.entity_id, entity.jurisdiction, FilingStatus.FALLOUT,
                             trace_id, payload_hash, time.time(),
                             f"Max retries exceeded. Last error: {last_error}")
        await self.fallout_queue.put(record)
        return record

    async def run(self, source: AsyncGenerator[FilingEntity, None]) -> list[AuditRecord]:
        tasks: list[asyncio.Task[AuditRecord]] = []
        async for entity in source:
            cfg = self.configs.get(entity.jurisdiction)
            if cfg is None:  # unknown jurisdiction is a data defect, not a filing
                logger.warning(json.dumps({"event": "unknown_jurisdiction",
                                           "entity_id": entity.entity_id,
                                           "jurisdiction": entity.jurisdiction}))
                continue
            tasks.append(asyncio.create_task(self._dispatch_one(entity, cfg)))

        records = await asyncio.gather(*tasks)
        for r in records:  # emit one structured audit line per filing decision
            logger.info(json.dumps({
                "level": "ERROR" if r.status is FilingStatus.REJECTED else "INFO",
                "trace_id": r.trace_id, "entity_id": r.entity_id,
                "jurisdiction": r.jurisdiction, "status": r.status.value,
                "payload_sha256": r.payload_sha256, "ts": r.timestamp,
                "error": r.error_detail,
            }))
        return records