Routing barcode scan logs to central inventory

Route fragmented barcode scan logs from dispensing stations to a central DEA-compliant inventory ledger without duplicate posting, using stateless dispatch, idempotency keys, and cryptographic audit trails in Python.

The problem: fragmented scans, duplicated ledger postings

Barcode scans originate at the edge — dispensing stations, automated dispensing cabinets (ADCs), and receiving docks — and each emitter retries independently when the network blips. The failure this page solves is narrow and costly: the same physical scan posts twice to the central ledger, inflating an on-hand count for a Schedule II substance and triggering a reconciliation discrepancy that surfaces as a DEA audit finding under 21 CFR § 1304.04. The mirror failure is just as damaging — a scan that never posts because the broker dropped it silently. Both produce inventory drift, and drift on controlled substances is a compliance liability, not a rounding error.

This recipe is part of the Barcode Scan Log Routing Logic subsystem, which itself sits inside the Data Ingestion & Inventory Sync Workflows framework. The goal here is a single, copy-paste-ready router that turns an at-least-once delivery stream into an effectively exactly-once central posting, with a tamper-evident trail behind every event. We assume scans have already been parsed; if your payloads still carry raw codes, normalize them first with the NDC-11 vs NDC-10 Parsing Standards rules so identical drugs always hash identically.

From at-least-once scan emitters to exactly-once central posting A left-to-right data flow. Three edge emitters — a dispensing station, an automated dispensing cabinet, and a receiving dock — each retry independently and fan into a Normalize and validate stage that enforces a canonical 11-digit NDC against a strict jsonschema. Valid events reach an idempotency gate backed by a Redis SET NX cache with a 72-hour TTL keyed on a SHA-256 digest; a duplicate digest branches down to a suppressed sink and is never re-posted. A unique event passes to a header-keyed router that builds a facility-by-transaction-by-schedule partition key. The router first appends a SHA-256 record to an append-only write-ahead log, then dispatches: Schedule II through V events go to a strict FIFO exactly-once queue governed by 21 CFR 1304.04, while OTC events go to a throughput-optimized batch queue. Dispensing station Auto dispensing cabinet (ADC) Receiving dock POS · retries on blip dock receipts at-least-once delivery Normalize jsonschema validate canonical NDC-11 Idempotency gate Redis SET NX · TTL 72h SHA-256 key duplicate Suppressed no second post unique Header router facility·type·schedule partition key 1 · audit first Append-only WAL SHA-256 · before dispatch 2 · dispatch Schedule II–V FIFO · exactly-once §1304.04 OTC batch throughput-optimized eventual consistency

Prerequisites & environment

  • Python 3.11+ (uses datetime.UTC-style timezone-aware timestamps and modern type hints).
  • jsonschema for synchronous payload validation at the ingestion edge.
  • hashlib (standard library) for SHA-256 idempotency and audit hashing — no third-party crypto required.
  • A message broker client (Kafka, RabbitMQ, or SQS) exposing a publish(topic, key, value, headers) method.
  • A distributed cache (Redis) for the deduplication window, and an append-only store (WAL or object lock bucket) for the audit trail.

You should already understand DEA schedule classification — Schedule II–V events demand strict FIFO and exactly-once semantics, while OTC items tolerate eventual consistency. If schedule lookup is not yet wired in, build it from the DEA Schedule II-V Classification Mapping engine before deploying this router. Persistent connectivity gaps at the edge are handled separately by Fallback Routing for Offline Sync; this router assumes the broker is reachable and focuses on de-duplication once events arrive.

Implementation

The router does four things in order: validate against a strict schema, derive a deterministic idempotency key, classify the DEA schedule to pick a routing key, and append an audit record before dispatch. Schema validation is synchronous and at the edge — schema drift is the leading cause of reconciliation gaps, so a malformed payload must never propagate downstream.

python
import hashlib
import json
import logging
import time
from dataclasses import dataclass
from typing import Any, Dict, Optional
from datetime import datetime, timezone
from jsonschema import validate, ValidationError

# Structured logging — audit format, no PHI in the log line
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)sZ | %(levelname)s | %(message)s",
    datefmt="%Y-%m-%dT%H:%M:%S",
)
logger = logging.getLogger("pharmacy_scan_router")

SCAN_EVENT_SCHEMA = {
    "type": "object",
    "required": ["scanner_id", "timestamp", "ndc", "quantity", "transaction_type", "facility_id"],
    "properties": {
        "scanner_id": {"type": "string", "pattern": "^[A-Z0-9-]{8,24}$"},
        "timestamp": {"type": "string", "format": "date-time"},
        "ndc": {"type": "string", "pattern": "^[0-9]{11}$"},  # canonical NDC-11 only
        "quantity": {"type": "integer", "minimum": 1},
        "transaction_type": {"enum": ["receipt", "dispense", "return", "waste"]},
        "facility_id": {"type": "string", "pattern": "^[A-Z]{2}-[0-9]{4}$"},
    },
    "additionalProperties": False,
}

DEA_SCHEDULE_MAP = {
    "00400012345": "II", "00400012346": "III",
    "00400012347": "IV", "00400012348": "V",
}


@dataclass(frozen=True)
class ScanRouter:
    broker_client: Any   # Kafka/RabbitMQ/SQS client with .publish(...)
    audit_store: Any     # Append-only WAL / object-lock store with .append(...)
    dedupe_cache: Any    # Redis-like client with .set(key, val, nx=..., ex=...)
    dedupe_ttl_seconds: int = 72 * 3600  # match the DEA reconciliation window
    max_retries: int = 3
    base_backoff: float = 0.5

    def _idempotency_key(self, event: Dict[str, Any]) -> str:
        # Canonical serialization: sorted keys + tight separators so two
        # logically-identical payloads always hash to the same digest.
        payload_bytes = json.dumps(event, sort_keys=True, separators=(",", ":")).encode("utf-8")
        composite = f"{event['scanner_id']}|{event['timestamp']}|{hashlib.sha256(payload_bytes).hexdigest()}"
        return hashlib.sha256(composite.encode("utf-8")).hexdigest()

    def _classify_schedule(self, ndc: str) -> str:
        return DEA_SCHEDULE_MAP.get(ndc, "OTC")

    def _is_duplicate(self, key: str) -> bool:
        # SET NX returns False/None if the key already exists -> duplicate scan.
        stored = self.dedupe_cache.set(key, "1", nx=True, ex=self.dedupe_ttl_seconds)
        return not stored

    def _route_event(self, event: Dict[str, Any], routing_key: str, idem_key: str) -> None:
        headers = {
            "x-idempotency-key": idem_key,
            "x-facility-id": event["facility_id"],
            "x-transaction-type": event["transaction_type"],
            "x-dea-schedule": self._classify_schedule(event["ndc"]),
            "x-correlation-id": hashlib.sha256(
                f"{event['scanner_id']}-{event['timestamp']}".encode()
            ).hexdigest(),
        }
        attempt = 0
        while attempt < self.max_retries:
            try:
                self.broker_client.publish(
                    topic="pharmacy.scan.logs",
                    key=routing_key,
                    value=event,
                    headers=headers,
                )
                logger.info("routed routing_key=%s idem=%s", routing_key, idem_key)
                return
            except Exception as exc:  # broker transient failure
                attempt += 1
                time.sleep(self.base_backoff * (2 ** attempt))
                logger.warning("publish failed attempt=%d/%d err=%s", attempt, self.max_retries, exc)
        logger.error("retry exhausted -> dead-letter routing_key=%s", routing_key)
        self.broker_client.publish(
            topic="pharmacy.scan.dlq",
            key=routing_key,
            value={"original_event": event, "failure_reason": "retry_exhaustion",
                   "failure_timestamp": datetime.now(timezone.utc).isoformat()},
        )

    def process_scan(self, raw_payload: str) -> Optional[str]:
        try:
            event = json.loads(raw_payload)
            validate(instance=event, schema=SCAN_EVENT_SCHEMA)
        except (json.JSONDecodeError, ValidationError) as exc:
            logger.error("schema rejection err=%s", exc)
            return None

        idem_key = self._idempotency_key(event)
        if self._is_duplicate(idem_key):
            logger.info("duplicate suppressed idem=%s", idem_key)
            return idem_key  # already accounted for; do not post again

        schedule = self._classify_schedule(event["ndc"])
        routing_key = f"{event['facility_id']}.{event['transaction_type']}.{schedule}"

        # Audit BEFORE dispatch: the trail must record intent even if the broker fails.
        self.audit_store.append({
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "event_hash": idem_key,
            "routing_key": routing_key,
            "status": "dispatched",
        })
        self._route_event(event, routing_key, idem_key)
        return idem_key

The routing key facility_id || transaction_type || dea_schedule is also the broker partition key, which guarantees strict ordering per facility and transaction class — non-negotiable for controlled-substance reconciliation. Schedule II–V keys land on high-priority FIFO consumers; OTC keys feed throughput-optimized batch consumers documented in Async Batch Processing for Inventory Updates. Header-only routing keeps the broker out of deep payload inspection, holding P95 dispatch latency under 100 ms.

Verification & testing

Two properties must hold: the idempotency key is stable across re-serialization, and a replayed scan is suppressed. The following assertions cover both, plus the audit-hash contract.

python
class _FakeCache:
    def __init__(self): self._store = set()
    def set(self, key, val, nx=False, ex=None):
        if nx and key in self._store:
            return False
        self._store.add(key)
        return True


class _FakeBroker:
    def __init__(self): self.published = []
    def publish(self, topic, key, value, headers=None):
        self.published.append((topic, key, value))


class _FakeAudit:
    def __init__(self): self.entries = []
    def append(self, entry): self.entries.append(entry)


def test_dedup_and_stability():
    broker, audit, cache = _FakeBroker(), _FakeAudit(), _FakeCache()
    router = ScanRouter(broker, audit, cache)
    payload = '{"scanner_id":"RX-0001AAAA","timestamp":"2026-06-28T14:03:09Z",' \
              '"ndc":"00400012345","quantity":2,"transaction_type":"dispense",' \
              '"facility_id":"TX-0042"}'

    first = router.process_scan(payload)
    # Same scan, keys re-ordered by an upstream re-serializer -> identical digest.
    reordered = '{"facility_id":"TX-0042","transaction_type":"dispense","quantity":2,' \
                '"ndc":"00400012345","timestamp":"2026-06-28T14:03:09Z","scanner_id":"RX-0001AAAA"}'
    second = router.process_scan(reordered)

    assert first == second                       # stable idempotency key
    assert len(broker.published) == 1            # second call suppressed, posted once
    assert audit.entries[0]["routing_key"] == "TX-0042.dispense.II"
    assert router.process_scan('{"ndc":"bad"}') is None  # schema rejection

A clean run logs one routed, one duplicate suppressed, and exactly one audit append with a 64-character SHA-256 event_hash. To validate compliance posture, diff the audit store against committed ledger rows on a fixed window — every dispatched hash should map to exactly one committed posting. The SHA-256 primitive used here is the one described in the Python hashlib documentation.

Gotchas & compliance pitfalls

  • Serialization-order collisions and misses. The idempotency key is only stable because the payload is re-serialized with sort_keys=True and tight separators. If you hash the raw inbound string instead, an upstream that re-orders keys produces a different digest and the duplicate slips through. Always canonicalize before hashing.
  • NDC padding splits the hash. A scan carrying NDC-10 and a later scan of the same drug in NDC-11 hash differently and both post. Force canonical 11-digit NDCs at the schema boundary (the pattern above is ^[0-9]{11}$) using the NDC-11 vs NDC-10 Parsing Standards rules — never dedupe on mixed-width codes.
  • TTL shorter than the partition. If the Redis dedupe TTL expires before a delayed redelivery arrives during a long network partition, the cache misses and the event double-posts. Pin the TTL to the DEA reconciliation window (24–72 h) and fall back to a uniqueness constraint on event_hash at the ledger’s primary key so the database is the last line of defense.
  • additionalProperties: False vs. firmware updates. A scanner firmware push that adds a field will spike the schema rejection rate to 100% for that fleet. Version your schema, quarantine rejects to a forensic bucket rather than dropping them, and route parse failures through Error Handling & Retry Mechanisms instead of swallowing them.
  • Schedule II FIFO is not optional. Routing a Schedule II waste event before its paired dispense because they landed on different partitions corrupts the reconstruction. Keep dea_schedule in the partition key so controlled-substance events for one facility stay strictly ordered.

Frequently Asked Questions

Why dedupe in a cache instead of relying on the database?

The cache catches the overwhelming majority of duplicates within milliseconds, before the event ever touches the central ledger, keeping write amplification low. The database uniqueness constraint on event_hash is the durable backstop for the rare case where the cache TTL has expired. Use both: the cache for speed, the constraint for guarantees.

Does suppressing a duplicate violate DEA recordkeeping?

No. Suppression prevents a second posting of the same physical event; the original posting and its audit hash remain intact and immutable. The audit store records every dispatched intent, so an inspector can still see that a redelivery occurred without it inflating on-hand counts under 21 CFR § 1304.04.

How does this interact with offline dispensing stations?

This router assumes the broker is reachable. Stations that go offline buffer locally and replay on reconnect — those replays are exactly the at-least-once stream the idempotency key is designed to absorb. The buffering and deferred-validation logic lives in Fallback Routing for Offline Sync.