`services/claims/outbox_worker.py`

63 lines
"""Outbox worker — drains pending outbox_event rows to the state, completing the pattern.

queue_state_submission writes a 'state_claim_submission_requested' outbox row in the same
transaction as the SUBMITTING transition. This worker delivers it with AT-LEAST-ONCE
semantics + bounded retry:

  - success            -> mark 'sent'
  - transient failure  -> leave 'pending', bump attempts (retried next drain)
  - attempts exhausted -> mark 'failed' (claim NOT lost — stays SUBMITTING for ops review)

At-least-once means a row may be delivered more than once (e.g. adapter succeeded but the
'sent' write didn't land). So "claim already SUBMITTED_TO_STATE" is treated as SUCCESS, not
an error — otherwise a retry after a partial failure would loop forever.
"""
from __future__ import annotations

import json
from uuid import UUID

from services.claims.claim_workflow import ClaimStatus, complete_state_submission


class OutboxWorker:
    def __init__(self, repo, adapter, max_attempts: int = 3) -> None:
        self.repo = repo
        self.adapter = adapter
        self.max_attempts = max_attempts

    def drain_once(self) -> dict:
        sent = failed = retried = 0
        for row in self.repo.fetch_pending_outbox():
            attempts = row["attempts"] + 1
            try:
                self._dispatch(row["event_type"], json.loads(row["payload"]))
                self.repo.mark_outbox(row["outbox_event_id"], "sent", attempts)
                sent += 1
            except Exception as exc:  # noqa: BLE001 - worker isolates each row's failure
                if attempts >= self.max_attempts:
                    self.repo.mark_outbox(row["outbox_event_id"], "failed", attempts, str(exc))
                    failed += 1
                else:
                    self.repo.mark_outbox(row["outbox_event_id"], "pending", attempts, str(exc))
                    retried += 1
        return {"sent": sent, "failed": failed, "retried": retried}

    def _dispatch(self, event_type: str, payload: dict) -> None:
        if event_type == "state_claim_submission_requested":
            self._deliver_submission(payload)
        else:
            raise NotImplementedError(f"no outbox handler for {event_type!r}")

    def _deliver_submission(self, payload: dict) -> None:
        claim_id = UUID(payload["claim_id"])
        key = payload["idempotency_key"]
        try:
            complete_state_submission(self.repo, self.adapter, claim_id, key)
        except ValueError:
            # complete raises if the claim isn't awaiting submission. If it already reached
            # SUBMITTED_TO_STATE, a prior attempt delivered it — idempotent success.
            if self.repo.get_for_update(claim_id).status == ClaimStatus.SUBMITTED_TO_STATE:
                return
            raise