Barcode Scan Log Routing Logic

Deterministic routing of barcode scan logs is the operational backbone of perpetual pharmacy inventory and controlled substance accountability. In high-throughput dispensing environments, every scan event—receiving, dispensing, waste, transfer, or return—must traverse a validated, idempotent pipeline before mutating inventory state. Failure to enforce strict routing boundaries directly compromises DEA 21 CFR §1304.21 recordkeeping requirements, violates HIPAA Security Rule §164.312(b) audit controls, and breaks FDA DSCSA traceability mandates. This workflow operates as a core subsystem within broader Data Ingestion & Inventory Sync Workflows, ensuring that raw scanner telemetry is transformed into compliant, reconcilable inventory deltas.

1. Payload Validation & Schema Enforcement

Routing cannot begin without strict schema validation. Pharmacy scanners emit heterogeneous payloads (GS1-128, NDC-11, proprietary POS formats). All inbound events must be normalized against a canonical schema before queue admission. Validation failures must be rejected at the ingress layer to prevent poison messages from contaminating downstream state machines.

import hashlib
import logging
from datetime import datetime, timezone
from enum import Enum
from pydantic import BaseModel, Field, field_validator

logger = logging.getLogger(__name__)

class ScanType(str, Enum):
    RECEIVE = "receive"
    DISPENSE = "dispense"
    WASTE = "waste"
    RETURN = "return"
    ADJUST = "adjust"

class ScanLogPayload(BaseModel):
    scan_uuid: str = Field(..., min_length=36, max_length=36)
    ndc: str = Field(..., pattern=r"^\d{11}$")
    lot_number: str = Field(..., min_length=1, max_length=32)
    expiration_date: datetime
    scan_type: ScanType
    quantity: int = Field(..., gt=0)
    scanner_id: str = Field(..., min_length=4, max_length=16)
    operator_id: str = Field(..., min_length=6, max_length=20)
    facility_uuid: str = Field(..., min_length=36, max_length=36)
    timestamp_utc: datetime
    dea_schedule: int = Field(..., ge=2, le=5)

    @field_validator("timestamp_utc", mode="before")
    @classmethod
    def enforce_utc(cls, v: datetime) -> datetime:
        if v.tzinfo is None:
            return v.replace(tzinfo=timezone.utc)
        return v.astimezone(timezone.utc)

    @field_validator("operator_id", mode="before")
    @classmethod
    def pseudonymize_phi(cls, v: str) -> str:
        # HIPAA §164.514(b): De-identify at routing edge. 
        # Store deterministic hash suffix for audit correlation without exposing PII.
        return f"OP_{hashlib.sha256(v.encode()).hexdigest()[:8].upper()}"

Compliance Boundary: Validation occurs synchronously at the API gateway or message broker consumer. Invalid NDC formats, missing lot/expiry, or out-of-range DEA schedules trigger an immediate 400 Bad Request with a structured rejection payload routed to a quarantine sink. No partial or malformed payloads enter the routing queue. This aligns with DSCSA §202 traceability requirements and prevents state corruption in downstream JSON Schema Validation for Drug Records pipelines.

2. Idempotency & Deduplication Routing

Duplicate scans from RF interference, operator double-taps, or network retries will corrupt perpetual inventory counts. Idempotency must be enforced atomically before routing decisions are evaluated. The routing layer implements a Redis-backed idempotency registry using deterministic composite keys.

import redis
import hashlib

# Defined elsewhere on this page (see the surrounding blocks):
# - ScanLogPayload
# - logger

class IdempotencyRouter:
    def __init__(self, redis_client: redis.Redis, ttl_seconds: int = 86400):
        self.redis = redis_client
        self.ttl = ttl_seconds
        self._key_prefix = "scan:idemp:"

    def _build_idempotency_key(self, payload: ScanLogPayload) -> str:
        # Composite key: UUID + NDC + Lot + Qty + Type + Operator Hash
        # Guarantees uniqueness for identical operational actions within TTL window
        raw = f"{payload.scan_uuid}|{payload.ndc}|{payload.lot_number}|{payload.quantity}|{payload.scan_type.value}"
        return f"{self._key_prefix}{hashlib.sha256(raw.encode()).hexdigest()}"

    def check_and_acquire(self, payload: ScanLogPayload) -> bool:
        key = self._build_idempotency_key(payload)
        # SETNX with TTL: atomic check-and-set prevents race conditions
        acquired = self.redis.set(key, "ACQUIRED", nx=True, ex=self.ttl)
        if not acquired:
            logger.warning(f"Duplicate scan detected: {payload.scan_uuid} | Key: {key}")
            return False
        return True

Operational Guardrails: The SETNX (SET if Not eXists) pattern guarantees exactly-once processing semantics for the routing window. When a duplicate is detected, the payload is silently acknowledged and routed to a metrics sink rather than the mutation pipeline. This approach directly supports Automating scan log deduplication workflows by eliminating phantom inventory deltas before they reach the ledger. For enterprise deployments, Redis cluster mode with consistent hashing ensures linear scalability under peak dispensing loads.

3. Deterministic Routing & Queue Partitioning

Once validated and deduplicated, payloads are routed to partitioned message queues based on regulatory priority and operational velocity. Controlled substances require strict ordering and accelerated processing to satisfy DEA audit trails, while OTC and general inventory can tolerate batched, asynchronous reconciliation.

import json
from dataclasses import dataclass
from typing import Protocol
from datetime import datetime, timezone

# Defined elsewhere on this page (see the surrounding blocks):
# - ScanLogPayload

class MessageBroker(Protocol):
    def publish(self, queue: str, payload: dict) -> None: ...

@dataclass(frozen=True)
class RoutingRule:
    schedule_threshold: int
    queue_name: str
    priority: int

ROUTING_TABLE = [
    RoutingRule(schedule_threshold=2, queue_name="controlled.substance.high", priority=1),
    RoutingRule(schedule_threshold=3, queue_name="controlled.substance.standard", priority=2),
    RoutingRule(schedule_threshold=5, queue_name="general.inventory", priority=3),
]

def route_scan_event(broker: MessageBroker, payload: ScanLogPayload) -> None:
    rule = next(
        (r for r in ROUTING_TABLE if payload.dea_schedule <= r.schedule_threshold),
        ROUTING_TABLE[-1]
    )
    
    envelope = {
        "metadata": {
            "routed_at_utc": datetime.now(timezone.utc).isoformat(),
            "routing_rule": rule.queue_name,
            "compliance_tier": "DEA_SCHEDULE_II" if payload.dea_schedule == 2 else "GENERAL"
        },
        "payload": payload.model_dump(mode="json")
    }
    
    broker.publish(rule.queue_name, json.dumps(envelope))

Compliance Mapping: DEA 21 CFR §1304.21 requires that records for Schedule II substances be maintained separately and be readily retrievable. By partitioning queues at the routing layer, the system enforces logical segregation without requiring physical database sharding. High-priority queues are configured with zero-message-loss guarantees (e.g., RabbitMQ durable=true, Kafka acks=all), while general inventory routes to Async Batch Processing for Inventory Updates for cost-optimized throughput.

4. Secure Telemetry & Immutable Audit Trails

Every routing decision, rejection, and successful dispatch must generate an immutable audit record. The routing layer emits structured telemetry to a write-once, append-only log sink. This satisfies HIPAA §164.312(b) audit controls and provides forensic traceability for DEA inspections.

import structlog

# Defined elsewhere on this page (see the surrounding blocks):
# - ScanLogPayload

audit_logger = structlog.get_logger("pharmacy.audit")

def log_routing_decision(payload: ScanLogPayload, success: bool, reason: str = "") -> None:
    audit_logger.info(
        "scan_routing_event",
        scan_uuid=payload.scan_uuid,
        ndc=payload.ndc,
        scan_type=payload.scan_type.value,
        dea_schedule=payload.dea_schedule,
        routing_success=success,
        rejection_reason=reason,
        event_timestamp=payload.timestamp_utc.isoformat(),
        # Explicitly exclude operator PII per HIPAA minimum necessary standard
    )

Audit Readiness: The structured log format ensures that compliance officers can reconstruct the exact lifecycle of any scan event. Logs are shipped to a tamper-evident storage tier (e.g., AWS CloudTrail, immutable S3 buckets with Object Lock) and cross-referenced with EDI 852 & 846 Parsing Pipelines during nightly reconciliation cycles. Any discrepancy between routed scan counts and EDI inventory adjustments triggers an automated variance report.

5. Enterprise Scaling & Failure Isolation

At enterprise scale, routing logic must handle backpressure, network partitions, and scanner firmware drift. The architecture implements circuit breakers at the consumer level, dead-letter queues (DLQ) for unrecoverable payloads, and exponential backoff for transient broker failures. Routing state is kept stateless; all persistence is delegated to Redis and the message broker, enabling horizontal scaling of routing workers without coordination overhead.

When deploying across multi-facility health systems, facility UUIDs are used to shard routing consumers geographically, reducing cross-region latency and ensuring that local dispensing workflows remain operational during WAN degradation. This isolation pattern is critical for maintaining continuous DEA accountability during infrastructure outages.

By enforcing strict validation, atomic idempotency, deterministic partitioning, and immutable audit logging, the routing layer transforms raw scanner telemetry into a compliant, high-fidelity inventory signal. This foundation enables downstream automation teams to build reliable dispensing logic, automated cycle counts, and real-time controlled substance dashboards without compromising regulatory posture.

Related Pages

• Deduplicating barcode scan logs in high-volume environments