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 it is allowed to mutate inventory state. This routing subsystem operates within the Data Ingestion & Inventory Sync Workflows architecture, which in turn implements the event-sourced ledger and cryptographic boundaries defined by the platform’s Core Architecture & DEA Compliance Frameworks. Its single responsibility is narrow but unforgiving: transform raw, heterogeneous scanner telemetry into compliant, reconcilable inventory deltas without ever losing, duplicating, or misclassifying a controlled-substance movement.

Regulatory Context & Compliance Boundaries

Routing logic is not a convenience layer; it is the point at which strict-liability recordkeeping is either preserved or broken. Three federal regimes converge on the scan-routing edge:

The governing principle is that a malformed, duplicated, or unscheduled event must never reach the ledger. Once a phantom delta is committed, the perpetual record the DEA requires is already corrupt before any business logic runs, and reconstructing the true count requires a manual, reportable investigation. The routing layer therefore enforces four hard boundaries in sequence — validate, deduplicate, classify, and audit — and treats any failure of the first three as a routing rejection rather than a recoverable warning.

Scan Event Specification

Pharmacy scanners emit heterogeneous payloads — GS1-128 linear barcodes, GS1 DataMatrix 2D codes, and proprietary point-of-sale strings — but the router accepts only one canonical envelope. Every inbound event is normalized to the field set below before it is eligible for queue admission. NDC values are normalized to the 11-digit form using the rules defined in NDC-11 vs NDC-10 Parsing Standards; the DEA schedule is resolved against the DEA Schedule II-V Classification Mapping engine.

Deterministic Routing Workflow

The router is a strict state machine. A scan event advances only when the prior boundary returns success; any rejection terminates routing and diverts the payload to a side channel. The canonical sequence is:

1. Ingress validation — the raw payload is parsed into the canonical schema. A failure (bad NDC, missing lot/expiry, out-of-range schedule) yields an immediate 400 Bad Request and the event is routed to a quarantine sink. No partial payload proceeds.
2. Idempotency acquisition — a deterministic composite key is computed and claimed atomically. If the key already exists, the event is a duplicate: it is silently acknowledged and routed to a metrics sink, never to the mutation pipeline.
3. DEA-tier classification — the validated, unique event is matched against the routing table and assigned to a regulatory-priority queue.
4. Dispatch — the event is published to its queue inside a versioned envelope carrying routing metadata and the compliance tier.
5. Audit emission — every decision at steps 1–4 (accept, reject, deduplicate, dispatch) emits one structured, PHI-free audit record to the append-only sink.

State transitions are total: every event terminates in exactly one of committed-to-queue, quarantined, or deduplicated, and each terminal state is independently auditable.

Production Implementation

1. Payload Validation & Schema Enforcement

Routing cannot begin without strict schema validation. All inbound events are normalized against a canonical Pydantic model before queue admission; validation failures are rejected at the ingress layer to prevent poison messages from contaminating downstream state machines. Operator identity is pseudonymized in-model so PHI never crosses the routing edge.

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:
        # 45 CFR § 164.514(b): de-identify at the routing edge.
        # Store a deterministic hash suffix for audit correlation
        # without exposing PII downstream.
        return f"OP_{hashlib.sha256(v.encode()).hexdigest()[:8].upper()}"

Compliance boundary. Validation runs synchronously at the API gateway or broker consumer. Invalid ndc formats, missing lot/expiry, or out-of-range dea_schedule values trigger an immediate 400 Bad Request with a structured rejection payload routed to a quarantine sink. This mirrors the boundary enforced by JSON Schema Validation for Drug Records, which performs the same structural gate for distributor and EDI-sourced records.

2. Idempotency & Deduplication Routing

Duplicate scans from RF interference, operator double-taps, or network retries will corrupt perpetual inventory counts. Idempotency is enforced atomically before any routing decision is evaluated, using a Redis-backed registry keyed on a deterministic composite of the event’s operational identity.

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.
        # Guarantees uniqueness for identical operational actions
        # within the TTL window.
        raw = (
            f"{payload.scan_uuid}|{payload.ndc}|{payload.lot_number}"
            f"|{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)
        # SET NX 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("Duplicate scan: %s | key=%s", payload.scan_uuid, key)
            return False
        return True

Operational guardrails. The SET NX (set-if-not-exists) pattern guarantees exactly-once processing semantics for the routing window. When a duplicate is detected, the payload is acknowledged and routed to a metrics sink rather than the mutation pipeline, eliminating phantom inventory deltas before they reach the ledger. For enterprise deployments, Redis cluster mode with consistent hashing keeps the registry linearly scalable under peak dispensing load. The same deterministic-key discipline is reused downstream by Async Batch Processing for Inventory Updates to guarantee idempotent ledger commitment.

3. Deterministic Routing & Queue Partitioning

Once validated and deduplicated, payloads are routed to partitioned queues by regulatory priority. Controlled substances require strict ordering and accelerated processing to satisfy DEA audit trails; OTC and general inventory 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: str) -> 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) -> str:
    rule = next(
        (r for r in ROUTING_TABLE if payload.dea_schedule <= r.schedule_threshold),
        ROUTING_TABLE[-1],
    )
    envelope = {
        "schema_version": 1,
        "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))
    return rule.queue_name

Compliance mapping. 21 CFR § 1304.21 requires that records for Schedule II substances be maintained separately and remain readily retrievable. Partitioning queues at the routing layer enforces that logical segregation without physical database sharding. High-priority queues are configured for zero message loss (RabbitMQ durable=true, Kafka acks=all); the general-inventory queue 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 generates an immutable audit record. The router emits structured, PHI-free telemetry to a write-once, append-only sink, satisfying the 45 CFR § 164.312(b) audit-control requirement and providing 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(),
        # operator PII is deliberately excluded — minimum-necessary standard
    )

Compliance Mapping & Audit Boundaries

The router’s outputs are the first entries in the append-only ledger chain. Each terminal state maps to a distinct, independently retrievable audit artifact:

Audit records ship to a tamper-evident tier — AWS CloudTrail or S3 Object Lock in compliance mode — so that no actor, including platform administrators, can alter or delete a routing decision within its retention window. Because operator_id is pseudonymized in the schema layer, the audit stream is safe to replicate to lower-trust analytics environments without re-identification risk. Compliance officers reconstruct the exact lifecycle of any scan from scan_uuid alone, joining queue-commit, quarantine, and dedup records into a single chain of custody.

Error Handling & Offline Resilience

At scale, routing logic must absorb backpressure, network partitions, and scanner firmware drift without dropping a controlled-substance movement. The router keeps no business state of its own — all persistence is delegated to Redis and the broker — so workers scale horizontally with no coordination overhead. The resilience controls are:

• Quarantine queue — schema-invalid payloads are preserved verbatim for manual or automated remediation; they are never discarded, because a rejected Schedule II scan is still a recordkeeping event.
• Dead-letter queue (DLQ) — payloads that pass validation but fail repeated dispatch are moved to a DLQ with their full audit context, then replayed by the shared machinery described in Error Handling & Retry Mechanisms.
• Circuit breakers & backoff — consumer-level breakers and exponential backoff prevent a degraded broker from amplifying retries into a self-inflicted outage.
• Offline fallback — when the central broker is unreachable, scans are buffered locally with their idempotency keys intact and replayed in order on reconnection. This deferred-commitment path is owned by Fallback Routing for Offline Sync, which preserves DEA accountability during WAN degradation by sharding consumers on facility_uuid so local dispensing continues uninterrupted.

Because idempotency keys are deterministic and survive the outage window, a scan buffered offline and a scan replayed online resolve to the same key — so no count drifts even when a movement is recorded twice across the partition.

Downstream Integration

The router is the upstream source for most of the synchronization backbone. Its dispatched envelopes are consumed by several sibling subsystems:

• The central reconciliation path is implemented in the child guide Routing barcode scan logs to central inventory, which turns dispatched envelopes into committed ledger deltas across dispensing stations, automated dispensing cabinets, and receiving docks.
• General-inventory envelopes feed Async Batch Processing for Inventory Updates for batched, idempotent commitment.
• Routed scan counts are cross-referenced against EDI 852 & 846 Parsing Pipelines during nightly reconciliation; any variance between scan-derived movement and EDI inventory adjustment triggers an automated discrepancy report and, where thresholds are exceeded, a diversion alert.

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 that downstream automation can trust without re-validating provenance.

Frequently Asked Questions

Related

• Up: Data Ingestion & Inventory Sync Workflows — the parent architecture this routing subsystem belongs to.
• Routing barcode scan logs to central inventory — the focused implementation walkthrough for central reconciliation.
• Async Batch Processing for Inventory Updates — downstream idempotent commitment of general-inventory events.
• Error Handling & Retry Mechanisms — shared dead-letter and replay machinery.
• Fallback Routing for Offline Sync — deferred commitment under disconnected operation.
• DEA Schedule II-V Classification Mapping — the schedule-resolution engine that drives queue partitioning.