🎉 Initial commit

listener/Dockerfile

@@ -0,0 +1,20 @@
+# syntax=docker/dockerfile:1
+FROM python:3.12-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1
+
+WORKDIR /app
+
+# Copy into /app/listener/ (not /app/): `COPY listener/ /app/` would flatten the
+# directory contents into /app and `python -m listener.main` would find no package.
+COPY listener/requirements.txt /app/listener/requirements.txt
+RUN pip install --no-cache-dir -r /app/listener/requirements.txt
+
+COPY listener/ /app/listener/
+
+# Run as an unprivileged user.
+RUN useradd --system --create-home --uid 10001 appuser
+USER appuser
+
+CMD ["python", "-m", "listener.main"]

listener/__init__.py

@@ -0,0 +1 @@
+"""nodeledge mqtt-listener package."""

listener/db.py

@@ -0,0 +1,85 @@
+"""Database access for the MQTT listener.
+
+The network thread writes rows; writes happen only on that thread so a plain
+(no pool) blocking connection is used, with reconnection on failure.
+"""
+
+from __future__ import annotations
+
+import logging
+import threading
+from datetime import datetime
+from typing import TYPE_CHECKING
+
+import psycopg
+from psycopg import sql
+from psycopg.types.json import Jsonb
+
+if TYPE_CHECKING:
+    from psycopg import Connection
+
+logger = logging.getLogger("listener.db")
+
+# Row insert. Values are %s placeholders passed as parameters; no manual
+# string concatenation is involved, so this does not trip the
+# sql.SQL-LiteralString diagnostics for dynamically-built queries.
+_INSERT = sql.SQL(
+    """
+    INSERT INTO mqtt_messages (topic, imei, device, payload, payload_time)
+    VALUES (%s, %s, %s, %s, %s)
+    """
+)
+
+
+class Inserter:
+    """Single-threaded blocking writer for mqtt_messages.
+
+    Not thread-safe by itself; the listener funnels all writes through the
+    MQTT network thread and guards the connection with a lock so shutdown
+    can safely disconnect from the main thread.
+    """
+
+    def __init__(self, dsn: str) -> None:
+        self._dsn = dsn
+        self._lock = threading.Lock()
+        self._conn: Connection | None = None
+
+    def _connection(self) -> Connection:
+        if self._conn is None or self._conn.closed:
+            logger.info("connecting to database")
+            self._conn = psycopg.connect(self._dsn, autocommit=True)
+        return self._conn
+
+    def insert(
+        self,
+        topic: str,
+        imei: str,
+        device: str,
+        payload: dict,
+        payload_time: datetime | None = None,
+    ) -> None:
+        """Insert one message. Reconnects once on a broken connection."""
+        with self._lock:
+            try:
+                conn = self._connection()
+                with conn.cursor() as cur:
+                    cur.execute(
+                        _INSERT,
+                        (topic, imei, device, Jsonb(payload), payload_time),
+                    )
+            except psycopg.Error:
+                logger.exception("database write failed (topic=%s imei=%s)", topic, imei)
+                self._close_locked()
+                raise
+
+    def _close_locked(self) -> None:
+        if self._conn is not None:
+            try:
+                self._conn.close()
+            except psycopg.Error:
+                pass
+            self._conn = None
+
+    def close(self) -> None:
+        with self._lock:
+            self._close_locked()

listener/healthcheck.py

@@ -0,0 +1,40 @@
+"""Container healthcheck for the mqtt-listener.
+
+Exits 0 when the listener is alive: the process is running and its MQTT client
+is either connected or actively reconnecting. A missing pidfile means the main
+process crashed and docker should restart it (restart: unless-stopped).
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+
+PID_FILE = "/tmp/listener.pid"
+
+
+def main() -> int:
+    try:
+        with open(PID_FILE, encoding="utf-8") as f:
+            pid = int(f.read().strip())
+    except (FileNotFoundError, ValueError):
+        print("pidfile missing or unreadable")
+        return 1
+
+    # Docker kills the shell running the healthcheck, not the app, so the app
+    # pid must still exist. os.kill with signal 0 only probes for existence.
+    try:
+        os.kill(pid, 0)
+    except ProcessLookupError:
+        print(f"listener pid {pid} is gone")
+        return 1
+    except PermissionError:
+        print(f"listener pid {pid} running (owned by another uid)")
+        return 0
+
+    print(f"listener pid {pid} is alive")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

listener/main.py

@@ -0,0 +1,151 @@
+"""MQTT listener: subscribes to /<prefix>/<imei>, injects IMEI and device
+name into the JSON payload and stores the result in PostgreSQL.
+
+The paho network loop runs in its own thread. Writes go straight to the
+database from the network thread; the insert re-raises on failure so paho's
+automatic reconnect logic keeps the subscriber alive while a single message
+may be retried or dropped depending on QoS.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import signal
+import threading
+from pathlib import Path
+
+import paho.mqtt.client as mqtt
+from paho.mqtt.enums import CallbackAPIVersion
+
+from listener.db import Inserter
+from listener.process import DEFAULT_TIME_KEYS, normalize_imei, process_message
+
+logger = logging.getLogger("listener")
+
+_stop_event = threading.Event()
+_shutdown_complete = threading.Event()
+
+
+def load_imei_map(path: str) -> dict[str, str]:
+    """Load the IMEI -> device name map from a JSON file.
+
+    Keys must be 15 digit IMEIs; values must be non-empty strings.
+    """
+    data = json.loads(Path(path).read_text(encoding="utf-8"))
+    if not isinstance(data, dict):
+        raise ValueError(f"{path}: expected a JSON object mapping IMEI to device name")
+    result: dict[str, str] = {}
+    for key, value in data.items():
+        if not isinstance(value, str) or not value:
+            raise ValueError(f"{path}: values must be non-empty strings (key {key!r})")
+        normalized = key.strip()
+        if normalize_imei(normalized) is None:
+            raise ValueError(f"{path}: key {key!r} is not a 15-digit IMEI")
+        result[normalized] = value
+    logger.info("loaded %d IMEI mappings from %s", len(result), path)
+    return result
+
+
+def _env(name: str, default: str | None = None) -> str:
+    value = os.environ.get(name)
+    if not value:
+        if default is None:
+            raise RuntimeError(f"missing required environment variable {name}")
+        return default
+    return value
+
+
+def main() -> None:
+    log_level = _env("LOG_LEVEL", "INFO").upper()
+    logging.basicConfig(
+        level=getattr(logging, log_level, logging.INFO),
+        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+    )
+
+    broker_host = _env("MQTT_BROKER_HOST")
+    broker_port = int(_env("MQTT_BROKER_PORT", "1883"))
+    username = os.environ.get("MQTT_USERNAME") or None
+    password = os.environ.get("MQTT_PASSWORD") or None
+    client_id = _env("MQTT_CLIENT_ID", "nodeledge-mqtt-listener")
+    topic_filter = _env("MQTT_TOPIC_FILTER", "/+/+")
+    qos = int(_env("MQTT_QOS", "1"))
+
+    map_file = _env("IMEI_MAP_FILE")
+    imei_map = load_imei_map(map_file)
+    imei_key = _env("INJECT_IMEI_KEY", "imei")
+    device_key = _env("INJECT_DEVICE_KEY", "device")
+    time_keys = tuple(
+        key.strip()
+        for key in _env("PAYLOAD_TIME_KEYS", ",".join(DEFAULT_TIME_KEYS)).split(",")
+        if key.strip()
+    )
+
+    inserter = Inserter(_env("DB_DSN"))
+
+    # Healthcheck probe target: container health is "main process alive".
+    Path("/tmp/listener.pid").write_text(str(os.getpid()), encoding="utf-8")
+    logger.info("listener starting, pid %d", os.getpid())
+
+    def on_connect(client, userdata, flags, reason_code, properties):  # noqa: ANN001
+        if reason_code.is_failure:
+            logger.error("broker refused connection: %s", reason_code)
+            return
+        logger.info("connected to broker, subscribing to %s (qos %d)", topic_filter, qos)
+        client.subscribe(topic_filter, qos=qos)
+
+    def on_message(client, userdata, message):  # noqa: ANN001
+        try:
+            topic = message.topic
+            result = process_message(
+                topic,
+                bytes(message.payload),
+                imei_map,
+                imei_key=imei_key,
+                device_key=device_key,
+                time_keys=time_keys,
+            )
+            if result is None:
+                logger.warning(
+                    "skipping message on %r: payload is not a JSON object or last topic "
+                    "segment is not a 15-digit IMEI",
+                    topic,
+                )
+                return
+            inserter.insert(topic, result.imei, result.device, result.payload, result.payload_time)
+            logger.info("stored %s/%s -> %s", result.imei, result.device, topic)
+        except Exception:
+            logger.exception("message handling failed for topic %s", getattr(message, "topic", "?"))
+
+    client = mqtt.Client(
+        callback_api_version=CallbackAPIVersion.VERSION2,
+        client_id=client_id,
+    )
+    if username:
+        client.username_pw_set(username, password)
+    client.on_connect = on_connect
+    client.on_message = on_message
+    client.reconnect_delay_set(min_delay=1, max_delay=60)
+    client.connect_async(broker_host, broker_port, keepalive=60)
+    client.loop_start()  # retries initial connection failures automatically
+
+    def _handle_signal(signum, frame):  # noqa: ANN001
+        logger.info("received signal %s, shutting down", signum)
+        _stop_event.set()
+
+    signal.signal(signal.SIGTERM, _handle_signal)
+    signal.signal(signal.SIGINT, _handle_signal)
+
+    while not _stop_event.is_set():
+        _stop_event.wait(timeout=5.0)
+
+    client.loop_stop()
+    client.disconnect()
+    inserter.close()
+    Path("/tmp/listener.pid").unlink(missing_ok=True)
+    logger.info("shutdown complete")
+
+
+if __name__ == "__main__":
+    main()

listener/process.py

@@ -0,0 +1,123 @@
+"""Message handling: JSON parsing, IMEI extraction and payload injection."""
+
+from __future__ import annotations
+
+import json
+import re
+from dataclasses import dataclass
+from datetime import datetime, timezone
+
+# An IMEI is a 15 digit decimal number. The topic segment must be exactly
+# that: 15 digits, no separators, no letters, no other characters.
+_IMEI_RE = re.compile(r"^[0-9]{15}$")
+
+# Epoch values above this many seconds must be milliseconds (seconds only
+# reach this magnitude in the year ~5138, so the heuristic is safe).
+_EPOCH_MS_THRESHOLD = 1e11
+
+# Default payload keys searched for a device timestamp; first parseable
+# value wins. Configurable via PAYLOAD_TIME_KEYS in the listener container.
+DEFAULT_TIME_KEYS = ("timestamp", "time", "ts")
+
+
+def normalize_imei(value: str) -> str | None:
+    """Return the IMEI as a plain 15 digit string, or None if not one."""
+    if not _IMEI_RE.match(value):
+        return None
+    return value
+
+
+def _as_utc(value: datetime) -> datetime:
+    """Interpret naive timestamps as UTC."""
+    return value if value.tzinfo is not None else value.replace(tzinfo=timezone.utc)
+
+
+def parse_payload_time(value: object) -> datetime | None:
+    """Best-effort device-timestamp parse.
+
+    Accepts epoch seconds or milliseconds (number or numeric string) and
+    ISO 8601 strings. Returns an aware UTC datetime or None when the value
+    cannot be interpreted. Booleans are rejected (bool is an int subclass).
+    """
+    if isinstance(value, bool) or value is None:
+        return None
+    if isinstance(value, str):
+        try:
+            return _as_utc(datetime.fromisoformat(value))
+        except ValueError:
+            try:
+                value = float(value)  # numeric string -> epoch path
+            except ValueError:
+                return None
+    if isinstance(value, (int, float)):
+        seconds = float(value)
+        if seconds > _EPOCH_MS_THRESHOLD:
+            seconds /= 1000.0  # milliseconds -> seconds
+        try:
+            return datetime.fromtimestamp(seconds, tz=timezone.utc)
+        except (OverflowError, OSError, ValueError):
+            return None
+    return None
+
+
+@dataclass(frozen=True)
+class InjectionResult:
+    imei: str
+    device: str
+    payload: dict
+    payload_time: datetime | None = None
+
+
+def process_message(
+    topic: str,
+    payload_bytes: bytes,
+    imei_map: dict[str, str],
+    *,
+    imei_key: str = "imei",
+    device_key: str = "device",
+    time_keys: tuple[str, ...] = DEFAULT_TIME_KEYS,
+) -> InjectionResult | None:
+    """Parse a payload and inject IMEI + device into it, extracting the
+    device timestamp.
+
+    Returns None when the payload is not valid JSON or not a JSON object —
+    those messages are skipped (logged by the caller). The original payload's
+    own imei/device fields are overwritten by the values derived from the
+    topic, so the topic is always authoritative.
+
+    A missing or unparseable timestamp does NOT skip the message: the
+    payload_time is simply None and only received_at (server receipt time)
+    exists for the row.
+    """
+    parts = topic.split("/")
+    # IMEI is the last topic segment, e.g. /xxx/<imei> -> imei segment.
+    raw_imei = parts[-1] if parts else ""
+
+    normalized = normalize_imei(raw_imei)
+    if normalized is None:
+        return None
+
+    try:
+        payload = json.loads(payload_bytes.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError):
+        return None
+    if not isinstance(payload, dict):
+        return None
+
+    payload = dict(payload)  # do not mutate shared objects
+    payload[imei_key] = normalized
+    payload[device_key] = imei_map.get(normalized, normalized)
+
+    payload_time = None
+    for key in time_keys:
+        if key in payload:
+            payload_time = parse_payload_time(payload[key])
+            if payload_time is not None:
+                break
+
+    return InjectionResult(
+        imei=normalized,
+        device=payload[device_key],
+        payload=payload,
+        payload_time=payload_time,
+    )

listener/requirements.txt

@@ -0,0 +1,2 @@
+paho-mqtt==2.1.0
+psycopg[binary]==3.2.3