Configuring async queues for high-volume claims ingestion

import asyncio
import os
import gc
import time
import itertools
import logging
from typing import Dict, Any, Optional
from dataclasses import dataclass
from collections import deque

logger = logging.getLogger(__name__)

@dataclass
class ClaimPayload:
    raw_bytes: memoryview
    routing_data: Dict[str, Any]
    priority: int = 100
    is_valid: bool = True
    reject_code: Optional[str] = None

class ClaimsIngestionController:
    def __init__(self, max_queue_size: int = 5000, memory_threshold_pct: float = 75.0, worker_count: int = 4):
        self.queue = asyncio.Queue(maxsize=max_queue_size)
        self.priority_queue = asyncio.PriorityQueue(maxsize=max_queue_size)
        self.dlq = asyncio.Queue()
        self.memory_threshold = memory_threshold_pct
        self.worker_count = worker_count
        self._shutdown = asyncio.Event()
        # No maxlen: a bounded deque would silently evict the oldest claim once
        # full, dropping payloads before they are flushed to adjudication.
        self._batch_buffer: deque = deque()
        self._last_flush = time.monotonic()
        # Monotonic tiebreaker so equal-priority items never force a comparison
        # of (uncomparable) ClaimPayload objects inside the PriorityQueue.
        self._seq = itertools.count()

    def _get_rss_mb(self) -> float:
        try:
            with open("/proc/self/status", "r") as f:
                for line in f:
                    if line.startswith("VmRSS:"):
                        return int(line.split()[1]) / 1024
        except FileNotFoundError:
            return 0.0
        return 0.0

    def _check_memory_pressure(self) -> bool:
        try:
            total_mem_mb = os.sysconf("SC_PAGE_SIZE") * os.sysconf("SC_PHYS_PAGES") / (1024 * 1024)
            if total_mem_mb == 0:
                return False
            return (self._get_rss_mb() / total_mem_mb) * 100 > self.memory_threshold
        except (AttributeError, ValueError):
            return False

    async def parse_and_validate(self, raw_stream: bytes) -> ClaimPayload:
        mv = memoryview(raw_stream)
        routing: Dict[str, Any] = {}
        is_valid = True
        reject_code = None

        # Synchronous boundary validation for critical NCPDP segments
        if b"407-D7" not in raw_stream:
            is_valid = False
            reject_code = "75"
        elif b"440-E4" not in raw_stream:
            is_valid = False
            reject_code = "70"

        # Priority extraction via B4 transaction code
        priority = 100
        if b"B4=01" in raw_stream or b"B4=02" in raw_stream:
            priority = 10

        return ClaimPayload(
            raw_bytes=mv,
            routing_data=routing,
            priority=priority,
            is_valid=is_valid,
            reject_code=reject_code
        )

    async def ingest(self, raw_stream: bytes):
        if self._check_memory_pressure():
            logger.warning("Memory pressure threshold breached. Applying backpressure.")
            await asyncio.sleep(0.5)
            return

        claim = await self.parse_and_validate(raw_stream)

        if not claim.is_valid:
            await self.dlq.put(claim)
            logger.info(f"Routed to DLQ. Reject Code: {claim.reject_code}")
            return

        try:
            if claim.priority <= 10:
                await asyncio.wait_for(
                    self.priority_queue.put((claim.priority, next(self._seq), claim)),
                    timeout=2.0,
                )
            else:
                await asyncio.wait_for(self.queue.put(claim), timeout=2.0)
        except asyncio.TimeoutError:
            logger.error("Queue full. Backpressure triggered. Routing to DLQ.")
            await self.dlq.put(claim)

    async def _process_worker(self, worker_id: int):
        while not self._shutdown.is_set():
            try:
                if not self.priority_queue.empty():
                    _, _, claim = await self.priority_queue.get()
                    source_queue = self.priority_queue
                else:
                    claim = await asyncio.wait_for(self.queue.get(), timeout=5.0)
                    source_queue = self.queue

                # Downstream: full payload hydration, formulary check, adjudication
                self._batch_buffer.append(claim)
                source_queue.task_done()

                current_time = time.monotonic()
                if len(self._batch_buffer) >= 5000 or (current_time - self._last_flush) >= 30.0:
                    await self._flush_batch()

                if self._check_memory_pressure():
                    gc.collect()
            except asyncio.TimeoutError:
                continue

    async def _flush_batch(self):
        if not self._batch_buffer:
            return
        batch = list(self._batch_buffer)
        self._batch_buffer.clear()
        self._last_flush = time.monotonic()
        logger.info(f"Flushing batch of {len(batch)} claims to adjudication API.")
        # Implement HTTP/gRPC dispatch here

    async def start_workers(self):
        tasks = [asyncio.create_task(self._process_worker(i)) for i in range(self.worker_count)]
        await asyncio.gather(*tasks)

    async def run(self):
        await self.start_workers()