Configuring MQTT QoS levels for telemetry drops

import paho.mqtt.client as mqtt
import json
import time
import sqlite3
import threading
import logging
from pathlib import Path
from dataclasses import dataclass
from typing import Dict

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

# Edge gateway constraint boundaries
MAX_BUFFER_SIZE = 5000
QOS_2_TIMEOUT_SEC = 15
LOCAL_DB_PATH = Path("/var/lib/edge-gateway/telemetry_fallback.db")

@dataclass
class TelemetryPacket:
    topic: str
    payload: bytes
    qos: int
    timestamp: float
    critical: bool  # True for QoS 2 routing

class EdgeMQTTRouter:
    def __init__(self, broker_host: str, broker_port: int = 1883):
        # Configure client for memory-constrained environments
        self.client = mqtt.Client(client_id=f"edge-gw-{int(time.time())}")
        self.client.max_inflight_messages_set(100)
        self.client.max_queued_messages_set(MAX_BUFFER_SIZE)
        
        self.client.on_connect = self._on_connect
        self.client.on_publish = self._on_publish
        self.client.on_disconnect = self._on_disconnect
        self.client.connect(broker_host, broker_port, keepalive=60)
        self.client.loop_start()

        self._lock = threading.Lock()
        self._pending_qos2: Dict[int, float] = {}  # msg_id -> publish_time
        self._db_conn = sqlite3.connect(str(LOCAL_DB_PATH), check_same_thread=False)
        self._init_db()

    def _init_db(self):
        # WAL mode reduces write-lock contention on constrained storage
        self._db_conn.execute("PRAGMA journal_mode=WAL")
        self._db_conn.execute("PRAGMA cache_size=-2000")  # ~2MB cache
        self._db_conn.execute("""
            CREATE TABLE IF NOT EXISTS fallback_queue (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                topic TEXT NOT NULL,
                payload BLOB NOT NULL,
                qos INTEGER NOT NULL,
                timestamp REAL NOT NULL,
                retries INTEGER DEFAULT 0
            )
        """)
        self._db_conn.commit()

    def _on_connect(self, client, userdata, flags, rc):
        if rc == 0:
            logging.info("Broker connection established. Draining fallback queue.")
            self._drain_fallback()
        else:
            logging.error(f"Connection failed with code {rc}")

    def _on_publish(self, client, userdata, mid):
        with self._lock:
            if mid in self._pending_qos2:
                del self._pending_qos2[mid]

    def _on_disconnect(self, client, userdata, rc):
        logging.warning(f"Disconnected from broker (rc={rc}). Queuing to local storage.")

    def route_packet(self, packet: TelemetryPacket) -> bool:
        # Dynamic QoS downgrade under backpressure
        if self._is_backpressure_active():
            packet.qos = 0 if not packet.critical else 1

        if packet.qos == 2:
            info = self.client.publish(packet.topic, packet.payload, qos=2)
            if info.rc == mqtt.MQTT_ERR_SUCCESS:
                with self._lock:
                    self._pending_qos2[info.mid] = time.time()
                return True
            else:
                self._persist_to_db(packet)
                return False
        else:
            self.client.publish(packet.topic, packet.payload, qos=packet.qos)
            return True

    def _is_backpressure_active(self) -> bool:
        now = time.time()
        timed_out = [mid for mid, ts in self._pending_qos2.items() if now - ts > QOS_2_TIMEOUT_SEC]
        with self._lock:
            for mid in timed_out:
                del self._pending_qos2[mid]
            return len(self._pending_qos2) > MAX_BUFFER_SIZE

    def _persist_to_db(self, packet: TelemetryPacket):
        self._db_conn.execute(
            "INSERT INTO fallback_queue (topic, payload, qos, timestamp) VALUES (?, ?, ?, ?)",
            (packet.topic, packet.payload, packet.qos, packet.timestamp)
        )
        self._db_conn.commit()

    def _drain_fallback(self):
        cursor = self._db_conn.execute("SELECT id, topic, payload, qos FROM fallback_queue ORDER BY timestamp ASC LIMIT 100")
        rows = cursor.fetchall()
        for row in rows:
            self.client.publish(row[1], row[2], qos=row[3])
            self._db_conn.execute("DELETE FROM fallback_queue WHERE id = ?", (row[0],))
        self._db_conn.commit()

    def close(self):
        self.client.loop_stop()
        self.client.disconnect()
        self._db_conn.close()