Configuring edge gateways for offline cold chain data caching

import hashlib
import json
import logging
import sqlite3
import threading
from datetime import datetime, timezone
from typing import Dict, List

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)


class OfflineColdChainCache:
    """Thread-safe write-once buffer for cold chain telemetry.

    The cache reuses a single long-lived connection (SQLite caches statement
    plans on a per-connection basis) and serializes all access through an
    RLock. ``payload_hash`` is UNIQUE so reboot-time replay of a buffered
    batch cannot duplicate records.
    """

    def __init__(self, db_path: str = "/var/lib/pharma/edge_cache.db"):
        self.db_path = db_path
        self._lock = threading.RLock()
        self._conn = sqlite3.connect(self.db_path, timeout=10.0, check_same_thread=False)
        self._conn.execute("PRAGMA journal_mode=WAL;")
        self._conn.execute("PRAGMA synchronous=NORMAL;")
        self._conn.execute("PRAGMA foreign_keys=ON;")
        self._init_db()

    def _init_db(self) -> None:
        with self._lock:
            self._conn.executescript("""
                CREATE TABLE IF NOT EXISTS telemetry_buffer (
                    local_sequence_id INTEGER PRIMARY KEY AUTOINCREMENT,
                    sensor_uuid TEXT NOT NULL,
                    recorded_utc REAL NOT NULL,
                    raw_value REAL,
                    calibrated_value REAL,
                    unit TEXT,
                    sync_status INTEGER NOT NULL DEFAULT 0,
                    payload_hash TEXT NOT NULL UNIQUE,
                    created_at REAL DEFAULT (strftime('%s', 'now'))
                );
                CREATE INDEX IF NOT EXISTS idx_sync_status ON telemetry_buffer(sync_status);
                CREATE INDEX IF NOT EXISTS idx_recorded_utc ON telemetry_buffer(recorded_utc);
                CREATE TRIGGER IF NOT EXISTS telemetry_buffer_no_delete
                  BEFORE DELETE ON telemetry_buffer
                  BEGIN SELECT RAISE(ABORT, 'telemetry_buffer is append-only'); END;
            """)
            self._conn.commit()

    def insert_payload(
        self, sensor_uuid: str, recorded_utc: float, raw: float, calibrated: float, unit: str,
    ) -> bool:
        """Insert one payload. Hash is over fields the server will receive,
        so the server can independently recompute and verify it."""
        payload_str = json.dumps(
            {
                "sensor_uuid": sensor_uuid,
                "recorded_utc": recorded_utc,
                "raw": raw,
                "calibrated": calibrated,
                "unit": unit,
            },
            sort_keys=True,
            separators=(",", ":"),
        )
        payload_hash = hashlib.sha256(payload_str.encode("utf-8")).hexdigest()

        with self._lock:
            try:
                cursor = self._conn.execute(
                    "INSERT OR IGNORE INTO telemetry_buffer "
                    "(sensor_uuid, recorded_utc, raw_value, calibrated_value, unit, payload_hash) "
                    "VALUES (?, ?, ?, ?, ?, ?)",
                    (sensor_uuid, recorded_utc, raw, calibrated, unit, payload_hash),
                )
                self._conn.commit()
                if cursor.rowcount == 0:
                    logger.debug("Duplicate payload suppressed: %s", payload_hash)
                    return False
                logger.info("Buffered payload for %s [seq=%s]", sensor_uuid, cursor.lastrowid)
                return True
            except sqlite3.Error as e:
                logger.error("Database write failed: %s", e)
                self._conn.rollback()
                return False

    def get_pending_sync(self, batch_size: int = 500) -> List[Dict]:
        with self._lock:
            rows = self._conn.execute(
                "SELECT local_sequence_id, sensor_uuid, recorded_utc, raw_value, "
                "calibrated_value, unit, payload_hash FROM telemetry_buffer "
                "WHERE sync_status = 0 ORDER BY recorded_utc ASC LIMIT ?",
                (batch_size,),
            ).fetchall()
        return [
            {
                "local_sequence_id": r[0],
                "sensor_uuid": r[1],
                "recorded_utc": r[2],
                "raw_value": r[3],
                "calibrated_value": r[4],
                "unit": r[5],
                "payload_hash": r[6],
            }
            for r in rows
        ]

    def mark_synced(self, sequence_ids: List[int]) -> None:
        if not sequence_ids:
            return
        with self._lock:
            try:
                placeholders = ",".join("?" for _ in sequence_ids)
                self._conn.execute(
                    f"UPDATE telemetry_buffer SET sync_status = 1 "
                    f"WHERE local_sequence_id IN ({placeholders}) AND sync_status = 0",
                    sequence_ids,
                )
                self._conn.commit()
                logger.info("Marked %d records as synced.", len(sequence_ids))
            except sqlite3.Error as e:
                logger.error("Sync status update failed: %s", e)
                self._conn.rollback()

    def close(self) -> None:
        with self._lock:
            self._conn.close()


# Caller-side example: capture the recorded_utc at the sensor read site so
# the hash is reproducible by both gateway and server.
# cache = OfflineColdChainCache()
# now_utc = datetime.now(timezone.utc).timestamp()
# cache.insert_payload("SENSOR-001", now_utc, raw=2.41, calibrated=2.43, unit="C")

sqlite3 /var/lib/pharma/edge_cache.db \
  "SELECT COUNT(*) FROM telemetry_buffer WHERE sync_status=0 AND recorded_utc < strftime('%s', 'now') - 86400;"