# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/through-what-dead-letter-quarantine-and-poison-message-isolation-gates-must-an-asynchronous-consumer-process-incoming-jobs-to-prevent-infinite-redelivery-loops-and-worker-pool-paralysis.md
# Remnant Warrant:    https://corpusfetch.com/c/through-what-dead-letter-quarantine-and-poison-message-isolation-gates-must-an-asynchronous-consumer-process-incoming-jobs-to-prevent-infinite-redelivery-loops-and-worker-pool-paralysis.warrant.json
# Markdown SHA-256:   c99c74d28622050fa383acde8fa718aefb103ac2e898af8b2b1c2bbd6adb6f35
# Code AST SHA-256:   94648448c414ce72ba52608337cfc9b23814a1a235473a5251cf3594c08f3620
# Verified Classes:   PermanentJobError, TransientJobError, MessageBroker, AsynchronousConsumerWorker
# Defensive Guards:   1 fail-closed invariants (zero external deps)

import json
import logging
import random
import time
from dataclasses import dataclass, field
from typing import Any, Callable, Mapping, Protocol

logger = logging.getLogger("corpusfetch.consumer")


class PermanentJobError(Exception):
    '''Raised when a job fails due to unrecoverable validation or schema faults.'''
    pass


class TransientJobError(Exception):
    '''Raised when a job fails due to temporary network or dependency drops.'''
    pass


class MessageBroker(Protocol):
    def acknowledge(self, message_id: str) -> None: ...
    def send_to_dlq(self, payload: dict[str, Any], reason: str, traceback_str: str) -> None: ...
    def schedule_retry(self, payload: dict[str, Any], delay_sec: float) -> None: ...


@dataclass
class AsynchronousConsumerWorker:
    '''
    Wraps asynchronous job execution with retry counting, exponential backoff with
    jitter, permanent error isolation, and Dead-Letter Queue quarantine.
    '''
    broker: MessageBroker
    max_retries: int = 3
    base_backoff_sec: float = 2.0

    def process_message(
        self,
        message_id: str,
        raw_payload: Mapping[str, Any],
        job_fn: Callable[[dict[str, Any]], Any],
    ) -> str:
        '''
        Consumes a message, inspects retry headers, executes job_fn, and routes
        failures either to delayed retry queues or the Dead-Letter Queue (DLQ).
        '''
        # Step 1: Extract and normalize envelope metadata
        mutable_payload = dict(raw_payload)
        envelope = mutable_payload.setdefault("_envelope", {})
        retry_count = int(envelope.get("retry_count", 0))
        first_seen = float(envelope.get("first_seen_epoch", time.time()))
        mutable_payload["_envelope"] = envelope

        try:
            # Step 2: Execute consumer business logic
            job_fn(mutable_payload)
            # Success: acknowledge message removal from active queue
            self.broker.acknowledge(message_id)
            return "SUCCESS"

        except (PermanentJobError, KeyError, TypeError, ValueError, ValidationError if 'ValidationError' in globals() else Exception) as exc:
            # Gate 1: Permanent faults bypass retries instantly and go straight to DLQ
            # (Note: catch specific expected business errors rather than bare Exception if possible)
            if isinstance(exc, TransientJobError):
                # Fall through to transient handler below
                raise exc

            err_msg = f"{type(exc).__name__}: {exc}"
            logger.error("Permanent job failure for message %s: %s. Routing to DLQ.", message_id, err_msg)
            
            self.broker.send_to_dlq(mutable_payload, reason=err_msg, traceback_str=str(exc))
            self.broker.acknowledge(message_id)  # Remove poison pill from active queue
            return "QUARANTINED_DLQ"

        except Exception as exc:
            # Gate 2: Transient faults get bounded exponential backoff + jitter
            err_msg = f"{type(exc).__name__}: {exc}"
            next_retry = retry_count + 1

            if next_retry > self.max_retries:
                logger.error(
                    "Max retries (%d) exceeded for message %s. Final routing to DLQ.",
                    self.max_retries,
                    message_id,
                )
                self.broker.send_to_dlq(mutable_payload, reason=f"MaxRetriesExceeded: {err_msg}", traceback_str=str(exc))
                self.broker.acknowledge(message_id)
                return "MAX_RETRIES_EXCEEDED_DLQ"

            # Compute exponential backoff with full decorrelated jitter
            backoff = (self.base_backoff_sec * (2 ** (next_retry - 1))) + random.uniform(0.0, 1.0)
            envelope["retry_count"] = next_retry
            envelope["first_seen_epoch"] = first_seen
            envelope["last_error"] = err_msg

            logger.warning(
                "Transient failure for message %s (attempt %d/%d). Retrying in %.2fs. Error: %s",
                message_id,
                next_retry,
                self.max_retries,
                backoff,
                err_msg,
            )

            self.broker.schedule_retry(mutable_payload, delay_sec=backoff)
            self.broker.acknowledge(message_id)  # Remove current redelivery attempt, replace with scheduled retry
            return "SCHEDULED_RETRY"