Fallback sync architecture for disconnected POS systems

SELECT 
  queue_depth,
  MAX(created_at) AS oldest_pending,
  COUNT(CASE WHEN schedule = 'II' THEN 1 END) AS c2_pending,
  SUM(CASE WHEN sync_status = 'PENDING' THEN 1 ELSE 0 END) AS total_pending
FROM offline_transaction_buffer
WHERE sync_status = 'PENDING'
GROUP BY queue_depth;

import hashlib
import hmac
import json
import logging
import sqlite3
import uuid
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from typing import Dict, Optional, List
from urllib3.util.retry import Retry
from requests.adapters import HTTPAdapter
import requests

# Configure HIPAA-compliant audit logger (no PHI/PII in logs)
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
    handlers=[logging.FileHandler("/var/log/pharmacy/sync_audit.log")]
)
logger = logging.getLogger("deafallback.sync_engine")

@dataclass
class TransactionPayload:
    ndc: str
    quantity: int
    schedule: str
    prescription_id: str
    pharmacist_npi: str
    dispensed_at_utc: str
    idempotency_key: str

class OfflineTransactionBuffer:
    """SQLite-backed FIFO queue with cryptographic hashing and WAL mode for crash resilience."""
    
    def __init__(self, db_path: str, encryption_key: Optional[bytes] = None):
        self.db_path = db_path
        self.conn = sqlite3.connect(db_path, timeout=30)
        self.conn.execute("PRAGMA journal_mode=WAL;")
        self.conn.execute("PRAGMA synchronous=NORMAL;")
        self._init_schema()
        
    def _init_schema(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS offline_queue (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                sequence_id TEXT UNIQUE NOT NULL,
                idempotency_key TEXT UNIQUE NOT NULL,
                payload_hash TEXT NOT NULL,
                payload_json TEXT NOT NULL,
                schedule TEXT NOT NULL,
                sync_status TEXT DEFAULT 'PENDING',
                created_at_utc TEXT NOT NULL,
                retry_count INTEGER DEFAULT 0
            );
            CREATE INDEX IF NOT EXISTS idx_sync_status ON offline_queue(sync_status);
            CREATE INDEX IF NOT EXISTS idx_schedule ON offline_queue(schedule);
        """)
        self.conn.commit()

    def enqueue(self, payload: TransactionPayload) -> str:
        """Persist transaction with SHA-256 hash and monotonic sequence ID."""
        canonical = json.dumps(asdict(payload), sort_keys=True, separators=(",", ":"))
        payload_hash = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
        sequence_id = f"{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')}_{uuid.uuid4().hex[:8]}"
        
        self.conn.execute(
            """INSERT INTO offline_queue 
               (sequence_id, idempotency_key, payload_hash, payload_json, schedule, created_at_utc)
               VALUES (?, ?, ?, ?, ?, ?)""",
            (sequence_id, payload.idempotency_key, payload_hash, canonical, payload.schedule, payload.dispensed_at_utc)
        )
        self.conn.commit()
        logger.info(f"Enqueued {payload.schedule} transaction: {sequence_id}")
        return sequence_id

    def fetch_pending(self, limit: int = 50) -> List[Dict]:
        """Retrieve FIFO ordered pending records for sync."""
        cursor = self.conn.execute(
            """SELECT id, sequence_id, payload_json, schedule, retry_count 
               FROM offline_queue 
               WHERE sync_status = 'PENDING' 
               ORDER BY id ASC LIMIT ?""",
            (limit,)
        )
        return [dict(zip(["id", "sequence_id", "payload_json", "schedule", "retry_count"], row)) for row in cursor.fetchall()]

    def mark_synced(self, record_ids: List[int]):
        """Update status to SYNCED after successful upstream acknowledgment."""
        placeholders = ",".join("?" * len(record_ids))
        self.conn.execute(f"UPDATE offline_queue SET sync_status = 'SYNCED' WHERE id IN ({placeholders})", record_ids)
        self.conn.commit()

class SyncEngine:
    """Handles deterministic reconciliation, retry with jitter, and DEA audit trail generation."""
    
    def __init__(self, api_base_url: str, auth_token: str, buffer: OfflineTransactionBuffer):
        self.api_base = api_base_url.rstrip("/")
        self.session = requests.Session()
        self.session.headers.update({"Authorization": f"Bearer {auth_token}", "Content-Type": "application/json"})
        
        # Exponential backoff with jitter for network resilience
        retry_strategy = Retry(
            total=5,
            backoff_factor=1.5,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["POST"]
        )
        self.session.mount("https://", HTTPAdapter(max_retries=retry_strategy))
        self.buffer = buffer

    def _verify_payload_integrity(self, payload_json: str, expected_hash: str) -> bool:
        """Validate SHA-256 integrity before transmission (21 CFR Part 11.10 compliance)."""
        actual_hash = hashlib.sha256(payload_json.encode("utf-8")).hexdigest()
        return hmac.compare_digest(actual_hash, expected_hash)

    def sync_batch(self) -> Dict[str, int]:
        """Execute deterministic reconciliation with schedule-aware routing."""
        pending = self.buffer.fetch_pending()
        if not pending:
            return {"synced": 0, "failed": 0}
            
        synced_ids = []
        failed_ids = []
        
        for record in pending:
            try:
                payload_data = json.loads(record["payload_json"])
                # Schedule II requires explicit audit boundary logging
                if record["schedule"] == "II":
                    logger.info(f"DEA C2 reconciliation initiated: {record['sequence_id']}")
                    
                response = self.session.post(
                    f"{self.api_base}/api/v1/transactions/sync",
                    json=payload_data,
                    headers={"X-Idempotency-Key": payload_data["idempotency_key"]},
                    timeout=15
                )
                response.raise_for_status()
                synced_ids.append(record["id"])
                
            except requests.exceptions.RequestException as e:
                logger.warning(f"Sync failed for {record['sequence_id']}: {str(e)}")
                failed_ids.append(record["id"])
                
        self.buffer.mark_synced(synced_ids)
        logger.info(f"Batch reconciliation complete. Synced: {len(synced_ids)}, Failed: {len(failed_ids)}")
        return {"synced": len(synced_ids), "failed": len(failed_ids)}