""" CMS submission tracking — durable status rows + per-attempt audit logging. Per the CMS ACCESS API Operations Manual v0.9.12, clients must store submission IDs and request context so polling can resume after an interruption or timeout. Each submitted operation (HTTP 202) gets a CMSApiSubmissionStatus row; every $submission-status check is logged as its own CMSApiRequest/CMSApiResponse pair. The service modules are submit-only: they POST the operation, record the durable row, and return immediately — never polling inline. All $submission-status checks happen in exactly one place, the management command `poll_cms_submissions` (one check per due row per run), which `cms_polling_worker` loops every CMS_POLL_LOOP_SECONDS. When a row reaches a final result, poll_cms_submissions applies it to the business record via the per-category handlers in the service modules. Concurrent runs coordinate through claim_submission_row(): an atomic conditional UPDATE that only one caller can win per row, so two overlapping runs (e.g. worker vs manual --force) do not issue the same status check at the same time. A force run can still re-claim a row another run just claimed; that residual duplicate is harmless because every final-result handler is idempotent under its own row lock / conditional update. """ import json import logging from datetime import timedelta from django.conf import settings as django_settings from django.db.models import F from django.utils import timezone from access_core.cms_integration.models import ( CMSApiRequest, CMSApiResponse, CMSApiSubmissionStatus, ) from access_core.services.cms_client import CmsApiError logger = logging.getLogger(__name__) def _poll_interval() -> int: return getattr(django_settings, "CMS_POLL_INTERVAL_SECONDS", 5) def record_submission( *, api_category: str, endpoint: str, submission_id: str, target_table: str | None = None, target_record_id=None, ) -> CMSApiSubmissionStatus: """Create the durable status row for a submitted (HTTP 202) operation.""" return CMSApiSubmissionStatus.objects.create( api_category=api_category, source_endpoint=endpoint, status=CMSApiSubmissionStatus.Status.PENDING, submission_id=submission_id, next_poll_at=timezone.now() + timedelta(seconds=_poll_interval()), target_table=target_table, target_record_id=target_record_id, ) def _log_poll_attempt( *, api_category: str, submission_id: str, status_code: int, response_data: dict ) -> None: """Log one $submission-status GET as a CMSApiRequest/CMSApiResponse pair.""" request_row = CMSApiRequest.objects.create( api_category=api_category, endpoint="$submission-status", method="GET", request_payload={"submission_id": submission_id}, request_headers={"Accept": "application/fhir+json"}, status_code=status_code, ) CMSApiResponse.objects.create( request=request_row, response_payload=json.loads(json.dumps(response_data, default=str)), response_headers={"Content-Type": "application/fhir+json"}, status_code=status_code, ) def note_failed_attempt(request_row: CMSApiRequest, exc: CmsApiError) -> None: """Attach a failed CMS API attempt to an already-created request row. Submit paths create the CMSApiRequest row *before* POSTing so that a failed request leaves an audit pair like successful ones do; this sets the status and stores the error body as its CMSApiResponse, mirroring what _log_poll_attempt does for $submission-status GETs. """ status_code = getattr(exc, "status_code", None) or 0 request_row.status_code = status_code request_row.save(update_fields=["status_code"]) response_data = getattr(exc, "response_data", None) or {"error": str(exc)} CMSApiResponse.objects.create( request=request_row, response_payload=json.loads(json.dumps(response_data, default=str)), response_headers={}, status_code=status_code, ) def _is_terminal_error(exc: CmsApiError) -> bool: """4xx (except 429) are permanent for a submission; 5xx/429/connection are transient.""" code = exc.status_code if code is None: return False return 400 <= code < 500 and code != 429 def poll_submission_once(client, status_row: CMSApiSubmissionStatus) -> dict | None: """ Perform a single $submission-status check. - Logs the attempt as a CMSApiRequest/CMSApiResponse pair. - Updates status_row: IN_PROGRESS while processing (next_poll_at set), COMPLETE when a final result arrives, ERROR on terminal 4xx. Transient failures (5xx/429/connection) reschedule next_poll_at and return None; terminal errors mark the row ERROR and re-raise. Returns the final response dict when complete, otherwise None. """ # In-memory count drives backoff math below; the database column is # incremented atomically by _save_poll_row (F expression), so concurrent # runs can never lose an update. status_row.poll_attempts += 1 try: result = client.get_submission_status("Patient", status_row.submission_id or "") except CmsApiError as exc: _log_poll_attempt( api_category=status_row.api_category, submission_id=status_row.submission_id or "", status_code=exc.status_code or 0, response_data=exc.response_data or {}, ) if _is_terminal_error(exc): logger.error( f"Submission {status_row.submission_id} got terminal error " f"(HTTP {exc.status_code}): {exc}" ) status_row.status = CMSApiSubmissionStatus.Status.ERROR status_row.next_poll_at = None _save_poll_row(status_row) raise logger.warning( f"Transient error checking submission {status_row.submission_id} " f"(HTTP {exc.status_code}); will retry." ) delay = _retry_delay(status_row) status_row.status = CMSApiSubmissionStatus.Status.IN_PROGRESS status_row.next_poll_at = timezone.now() + timedelta(seconds=delay) _save_poll_row(status_row) return None body = {k: v for k, v in result.items() if not k.startswith("_")} _log_poll_attempt( api_category=status_row.api_category, submission_id=status_row.submission_id or "", status_code=result["_http_status"], response_data=body, ) if result.get("_complete"): logger.info(f"Submission {status_row.submission_id} complete.") status_row.status = CMSApiSubmissionStatus.Status.COMPLETE status_row.next_poll_at = None _save_poll_row(status_row) return result # Still processing (HTTP 202, or 200 without a final-result marker). delay = _retry_delay(status_row) status_row.status = CMSApiSubmissionStatus.Status.IN_PROGRESS status_row.next_poll_at = timezone.now() + timedelta(seconds=delay) _save_poll_row(status_row) return None def _save_poll_row(status_row: CMSApiSubmissionStatus) -> None: """Persist a poll attempt's status/next_poll_at and bump poll_attempts. The attempt count is incremented with an F() expression inside the UPDATE so concurrent runs can never lose each other's increments (a model save would write the stale in-memory value). """ CMSApiSubmissionStatus.objects.filter(pk=status_row.pk).update( status=status_row.status, next_poll_at=status_row.next_poll_at, poll_attempts=F("poll_attempts") + 1, ) def _retry_delay(status_row: CMSApiSubmissionStatus) -> int: """Exponential backoff capped at 300s, based on attempts already made.""" base = _poll_interval() if not base: return 0 return min(base * (2**status_row.poll_attempts), 300) OPEN_STATUSES = [ CMSApiSubmissionStatus.Status.PENDING, CMSApiSubmissionStatus.Status.IN_PROGRESS, ] def open_target_ids(api_category: str): """target_record_id values of submissions in this category still awaiting a result.""" return CMSApiSubmissionStatus.objects.filter( api_category=api_category, status__in=OPEN_STATUSES ).values_list("target_record_id", flat=True) def claim_submission_row(status_row_pk, *, force: bool = False) -> bool: """Atomically claim an open submission row for this poll run. The conditional UPDATE is the lock: Postgres re-checks the WHERE clause against the freshly updated row when concurrent updates collide, so at most one caller wins a given row per run. Winning marks the row IN_PROGRESS and pushes next_poll_at forward; if the winner crashes before its status check completes, the row simply becomes due again after the interval and polling resumes (durable-row guarantee preserved). Returns True if this caller won the claim, False if another run already has it (or the row is no longer open / not yet due without force). """ now = timezone.now() qs = CMSApiSubmissionStatus.objects.filter( pk=status_row_pk, status__in=OPEN_STATUSES ) if not force: qs = qs.filter(next_poll_at__lte=now) return ( qs.update( status=CMSApiSubmissionStatus.Status.IN_PROGRESS, next_poll_at=now + timedelta(seconds=_poll_interval()), ) == 1 )