Remove arrowheads from backbone connection

Ny Funktionalitet för kopplingar och signalstyrka

.dockerignore

@@ -1,2 +1,3 @@
 .env
 node_modules
+.git

Dockerfile

@@ -1,12 +1,29 @@
-FROM node:lts-alpine
-
-# add project files to /app
-ADD ./ /app
-WORKDIR /app
+FROM node:lts-alpine AS build
 
 RUN apk add --no-cache openssl
 
-# install node dependencies
-RUN npm install
+WORKDIR /app
+
+# Copy only package files and install deps
+# This layer will be cached as long as package*.json don't change
+COPY package*.json package-lock.json* ./
+RUN --mount=type=cache,target=/root/.npm npm ci --omit=dev
+
+# Copy the rest of your source
+COPY . .
+
+# Pre-generate prisma client
+RUN node_modules/.bin/prisma generate
+
+FROM node:lts-alpine
+
+RUN apk add --no-cache openssl
+
+USER node:node
+
+WORKDIR /app
+
+COPY --from=build --chown=node:node /app .
+
 
 EXPOSE 8080

docker-compose.yml

@@ -30,6 +30,18 @@ services:
       DATABASE_URL: "mysql://root:password@database:3306/meshtastic-map?connection_limit=100"
       MAP_OPTS: "" # add any custom index.js options here
 
+  # publishes mqtt packets via websocket
+  meshtastic-ws:
+    container_name: meshtastic-ws
+    build:
+      context: .
+      dockerfile: ./Dockerfile
+    command: /app/docker/ws.sh
+    ports:
+      - 8081:8081/tcp
+    environment:
+      WS_OPTS: ""
+
   # runs the database to store everything from mqtt
   database:
     container_name: database

docker/ws.sh

@@ -0,0 +1,5 @@
+#!/bin/sh
+
+echo "Starting websocket publisher"
+exec node src/ws.js ${WS_OPTS}
+

package.json

@@ -9,17 +9,18 @@
   "author": "",
   "license": "ISC",
   "dependencies": {
-    "@prisma/client": "^6.12.0",
+    "@prisma/client": "^6.16.2",
     "command-line-args": "^6.0.1",
     "command-line-usage": "^7.0.3",
     "compression": "^1.8.1",
     "cors": "^2.8.5",
-    "express": "^5.0.0",
-    "mqtt": "^5.13.2",
-    "protobufjs": "^7.5.3"
+    "express": "^5.2.1",
+    "mqtt": "^5.14.1",
+    "protobufjs": "^7.5.4",
+    "ws": "^8.18.3"
   },
   "devDependencies": {
-    "jest": "^30.0.4",
-    "prisma": "^6.12.0"
+    "jest": "^30.1.3",
+    "prisma": "^6.16.2"
   }
 }

prisma/migrations/20250805210540_add_node_pulickey_and_unmessagable/migration.sql

@@ -0,0 +1,3 @@
+-- AlterTable
+ALTER TABLE `nodes` ADD COLUMN `is_unmessagable` BOOLEAN NULL,
+    ADD COLUMN `public_key` VARCHAR(191) NULL;

prisma/migrations/20250810082255_add_node_max_hops/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE `nodes` ADD COLUMN `max_hops` INTEGER NULL;

prisma/migrations/20250813193012_add_node_channel_id/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE `nodes` ADD COLUMN `channel_id` VARCHAR(191) NULL;

prisma/migrations/20250925171652_add_chutil_channel_id/migration.sql

@@ -0,0 +1,5 @@
+-- AlterTable
+ALTER TABLE `channel_utilization_stats` ADD COLUMN `channel_id` VARCHAR(191) NULL;
+
+-- CreateIndex
+CREATE INDEX `channel_utilization_stats_channel_id_idx` ON `channel_utilization_stats`(`channel_id`);

prisma/migrations/20250929184048_add_ok_to_mqtt_to_text_messages/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE `text_messages` ADD COLUMN `ok_to_mqtt` BOOLEAN NULL;

prisma/migrations/20260106151912_add_edges/migration.sql

@@ -0,0 +1,23 @@
+-- CreateTable
+CREATE TABLE `edges` (
+    `id` BIGINT NOT NULL AUTO_INCREMENT,
+    `from_node_id` BIGINT NOT NULL,
+    `to_node_id` BIGINT NOT NULL,
+    `snr` INTEGER NOT NULL,
+    `from_latitude` INTEGER NULL,
+    `from_longitude` INTEGER NULL,
+    `to_latitude` INTEGER NULL,
+    `to_longitude` INTEGER NULL,
+    `packet_id` BIGINT NOT NULL,
+    `channel_id` VARCHAR(191) NULL,
+    `gateway_id` BIGINT NULL,
+    `source` VARCHAR(191) NOT NULL,
+    `created_at` DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
+    `updated_at` DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
+
+    INDEX `edges_from_node_id_idx`(`from_node_id`),
+    INDEX `edges_to_node_id_idx`(`to_node_id`),
+    INDEX `edges_created_at_idx`(`created_at`),
+    INDEX `edges_from_node_id_to_node_id_idx`(`from_node_id`, `to_node_id`),
+    PRIMARY KEY (`id`)
+) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

prisma/schema.prisma

@@ -21,6 +21,8 @@ model Node {
   hardware_model Int
   role           Int
   is_licensed    Boolean?
+  public_key     String?
+  is_unmessagable Boolean?
 
   firmware_version       String?
   region                 Int?
@@ -52,7 +54,10 @@ model Node {
   mqtt_connection_state_updated_at DateTime?
 
   ok_to_mqtt  Boolean?
-  is_backbone  Boolean?
+  is_backbone Boolean?
+  max_hops    Int?
+
+  channel_id String?
 
   created_at DateTime @default(now())
   updated_at DateTime @default(now()) @updatedAt
@@ -234,6 +239,7 @@ model TextMessage {
   rx_snr    Decimal?
   rx_rssi   Int?
   hop_limit Int?
+  ok_to_mqtt  Boolean?
 
   created_at DateTime @default(now())
   updated_at DateTime @default(now()) @updatedAt
@@ -336,7 +342,33 @@ model ChannelUtilizationStats {
   id                      BigInt    @id @default(autoincrement())
   recorded_at             DateTime? @default(now())
   avg_channel_utilization Decimal?
+  channel_id              String?
 
+  @@index([channel_id])
   @@index([recorded_at])
   @@map("channel_utilization_stats")
+}
+
+model Edge {
+  id             BigInt  @id @default(autoincrement())
+  from_node_id   BigInt
+  to_node_id     BigInt
+  snr            Int
+  from_latitude  Int?
+  from_longitude Int?
+  to_latitude    Int?
+  to_longitude   Int?
+  packet_id      BigInt
+  channel_id     String?
+  gateway_id     BigInt?
+  source         String
+
+  created_at DateTime @default(now())
+  updated_at DateTime @default(now()) @updatedAt
+
+  @@index(from_node_id)
+  @@index(to_node_id)
+  @@index(created_at)
+  @@index([from_node_id, to_node_id])
+  @@map("edges")
 }

src/admin.js

@@ -1,6 +1,7 @@
 // node src/admin.js --purge-node-id 123
 // node src/admin.js --purge-node-id '!AABBCCDD'
 
+require('./utils/logger');
 const commandLineArgs = require("command-line-args");
 const commandLineUsage = require("command-line-usage");
 

src/index.js

@@ -1,3 +1,4 @@
+require('./utils/logger');
 const path = require('path');
 const express = require('express');
 const compression = require('compression');
@@ -146,6 +147,23 @@ app.get('/api', async (req, res) => {
             "path": "/api/v1/nodes/:nodeId/traceroutes",
             "description": "Trace routes for a meshtastic node",
         },
+        {
+            "path": "/api/v1/traceroutes",
+            "description": "Recent traceroute edges across all nodes",
+            "params": {
+                "time_from": "Only include traceroutes updated after this unix timestamp (milliseconds)",
+                "time_to": "Only include traceroutes updated before this unix timestamp (milliseconds)"
+            }
+        },
+        {
+            "path": "/api/v1/connections",
+            "description": "Aggregated edges between nodes from traceroutes",
+            "params": {
+                "node_id": "Only include connections involving this node id",
+                "time_from": "Only include edges created after this unix timestamp (milliseconds)",
+                "time_to": "Only include edges created before this unix timestamp (milliseconds)"
+            }
+        },
         {
             "path": "/api/v1/nodes/:nodeId/position-history",
             "description": "Position history for a meshtastic node",
@@ -211,6 +229,10 @@ app.get('/api/v1/nodes', async (req, res) => {
             where: {
                 role: role,
                 hardware_model: hardwareModel,
+                // Since we removed retention; only include nodes that have been updated in the last 30 days
+                updated_at: {
+                    gte: new Date(Date.now() - 30 * 24 * 60 * 60 * 1000) // within last 30 days
+                }
             },
         });
 
@@ -566,6 +588,313 @@ app.get('/api/v1/nodes/:nodeId/traceroutes', async (req, res) => {
     }
 });
 
+// Aggregated recent traceroute edges (global), filtered by updated_at
+// Returns deduplicated edges with the latest SNR and timestamp.
+// GET /api/v1/nodes/traceroutes?time_from=...&time_to=...
+app.get('/api/v1/traceroutes', async (req, res) => {
+    try {
+
+        const timeFrom = req.query.time_from ? parseInt(req.query.time_from) : undefined;
+        const timeTo = req.query.time_to ? parseInt(req.query.time_to) : undefined;
+
+        // Pull recent traceroutes within the time window. We only want replies (want_response=false)
+        // and those that were actually gated to MQTT (gateway_id not null)
+        const traces = await prisma.traceRoute.findMany({
+            where: {
+                want_response: false,
+                gateway_id: { not: null },
+                updated_at: {
+                    gte: timeFrom ? new Date(timeFrom) : undefined,
+                    lte: timeTo ? new Date(timeTo) : undefined,
+                },
+            },
+            orderBy: { id: 'desc' },
+            take: 5000, // cap to keep response bounded; UI can page/adjust time window if needed
+        });
+
+        // Normalize JSON fields that may be strings (depending on driver)
+        const normalized = traces.map((t) => {
+            const trace = { ...t };
+            if (typeof trace.route === 'string') {
+                try { trace.route = JSON.parse(trace.route); } catch(_) {}
+            }
+            if (typeof trace.route_back === 'string') {
+                try { trace.route_back = JSON.parse(trace.route_back); } catch(_) {}
+            }
+            if (typeof trace.snr_towards === 'string') {
+                try { trace.snr_towards = JSON.parse(trace.snr_towards); } catch(_) {}
+            }
+            if (typeof trace.snr_back === 'string') {
+                try { trace.snr_back = JSON.parse(trace.snr_back); } catch(_) {}
+            }
+            return trace;
+        });
+
+        // Build edges from both forward (towards) and reverse (back) paths.
+        // Forward path: to → route[] → from, using snr_towards
+        // Reverse path: from → route_back[] → to, using snr_back
+        const edgeKey = (a, b) => `${String(a)}->${String(b)}`;
+        const edges = new Map();
+
+        function upsertEdgesFromPath(trace, pathNodes, pathSnrs) {
+            for (let i = 0; i < pathNodes.length - 1; i++) {
+                const hopFrom = pathNodes[i];
+                const hopTo = pathNodes[i + 1];
+                const snr = typeof (pathSnrs && pathSnrs[i]) === 'number' ? pathSnrs[i] : null;
+
+                // Skip edges without SNR data
+                if (snr === null) continue;
+
+                const key = edgeKey(hopFrom, hopTo);
+                const existing = edges.get(key);
+                if (!existing) {
+                    edges.set(key, {
+                        from: hopFrom,
+                        to: hopTo,
+                        snr: snr,
+                        updated_at: trace.updated_at,
+                        channel_id: trace.channel_id ?? null,
+                        gateway_id: trace.gateway_id ?? null,
+                        traceroute_from: trace.from,  // original initiator
+                        traceroute_to: trace.to,      // original target
+                    });
+                } else if (new Date(trace.updated_at) > new Date(existing.updated_at)) {
+                    existing.snr = snr;
+                    existing.updated_at = trace.updated_at;
+                    existing.channel_id = trace.channel_id ?? existing.channel_id;
+                    existing.gateway_id = trace.gateway_id ?? existing.gateway_id;
+                    existing.traceroute_from = trace.from;
+                    existing.traceroute_to = trace.to;
+                }
+            }
+        }
+
+        for (const tr of normalized) {
+            // Forward path
+            const forwardPath = [];
+            if (tr.to != null) forwardPath.push(Number(tr.to));
+            if (Array.isArray(tr.route)) {
+                for (const hop of tr.route) {
+                    if (hop != null) forwardPath.push(Number(hop));
+                }
+            }
+            if (tr.from != null) forwardPath.push(Number(tr.from));
+            const forwardSnrs = Array.isArray(tr.snr_towards) ? tr.snr_towards : [];
+            upsertEdgesFromPath(tr, forwardPath, forwardSnrs);
+
+            // Reverse path
+            const reversePath = [];
+            if (tr.from != null) reversePath.push(Number(tr.from));
+            if (Array.isArray(tr.route_back)) {
+                for (const hop of tr.route_back) {
+                    if (hop != null) reversePath.push(Number(hop));
+                }
+            }
+            if (tr.to != null) reversePath.push(Number(tr.to));
+            const reverseSnrs = Array.isArray(tr.snr_back) ? tr.snr_back : [];
+            upsertEdgesFromPath(tr, reversePath, reverseSnrs);
+        }
+
+        res.json({
+            traceroute_edges: Array.from(edges.values()),
+        });
+
+    } catch (err) {
+        console.error(err);
+        res.status(500).json({
+            message: "Something went wrong, try again later.",
+        });
+    }
+});
+
+// Aggregated edges endpoint
+// GET /api/v1/connections?node_id=...&time_from=...&time_to=...
+app.get('/api/v1/connections', async (req, res) => {
+    try {
+        const nodeId = req.query.node_id ? parseInt(req.query.node_id) : undefined;
+        const timeFrom = req.query.time_from ? parseInt(req.query.time_from) : undefined;
+        const timeTo = req.query.time_to ? parseInt(req.query.time_to) : undefined;
+
+        // Query edges from database
+        const edges = await prisma.edge.findMany({
+            where: {
+                created_at: {
+                    ...(timeFrom && { gte: new Date(timeFrom) }),
+                    ...(timeTo && { lte: new Date(timeTo) }),
+                },
+                // Only include edges where both nodes have positions
+                from_latitude: { not: null },
+                from_longitude: { not: null },
+                to_latitude: { not: null },
+                to_longitude: { not: null },
+                // If node_id is provided, filter edges where either from_node_id or to_node_id matches
+                ...(nodeId !== undefined && {
+                    OR: [
+                        { from_node_id: nodeId },
+                        { to_node_id: nodeId },
+                    ],
+                }),
+            },
+            orderBy: [
+                { created_at: 'desc' },
+                { packet_id: 'desc' },
+            ],
+        });
+
+        // Collect all unique node IDs from edges
+        const nodeIds = new Set();
+        for (const edge of edges) {
+            nodeIds.add(edge.from_node_id);
+            nodeIds.add(edge.to_node_id);
+        }
+
+        // Fetch current positions for all nodes
+        const nodes = await prisma.node.findMany({
+            where: {
+                node_id: { in: Array.from(nodeIds) },
+            },
+            select: {
+                node_id: true,
+                latitude: true,
+                longitude: true,
+            },
+        });
+
+        // Create a map of current node positions
+        const nodePositions = new Map();
+        for (const node of nodes) {
+            nodePositions.set(node.node_id, {
+                latitude: node.latitude,
+                longitude: node.longitude,
+            });
+        }
+
+        // Filter edges: only include edges where both nodes are still at the same location
+        const validEdges = edges.filter(edge => {
+            const fromCurrentPos = nodePositions.get(edge.from_node_id);
+            const toCurrentPos = nodePositions.get(edge.to_node_id);
+
+            // Skip if either node doesn't exist or doesn't have a current position
+            if (!fromCurrentPos || !toCurrentPos || 
+                fromCurrentPos.latitude === null || fromCurrentPos.longitude === null ||
+                toCurrentPos.latitude === null || toCurrentPos.longitude === null) {
+                return false;
+            }
+
+            // Check if stored positions match current positions
+            const fromMatches = fromCurrentPos.latitude === edge.from_latitude && 
+                               fromCurrentPos.longitude === edge.from_longitude;
+            const toMatches = toCurrentPos.latitude === edge.to_latitude && 
+                             toCurrentPos.longitude === edge.to_longitude;
+
+            return fromMatches && toMatches;
+        });
+
+        // Normalize node pairs: always use min/max to treat A->B and B->A as same connection
+        const connectionsMap = new Map();
+
+        for (const edge of validEdges) {
+            const nodeA = edge.from_node_id < edge.to_node_id ? edge.from_node_id : edge.to_node_id;
+            const nodeB = edge.from_node_id < edge.to_node_id ? edge.to_node_id : edge.from_node_id;
+            const key = `${nodeA}-${nodeB}`;
+
+            if (!connectionsMap.has(key)) {
+                connectionsMap.set(key, {
+                    node_a: nodeA,
+                    node_b: nodeB,
+                    direction_ab: [], // A -> B edges
+                    direction_ba: [], // B -> A edges
+                });
+            }
+
+            const connection = connectionsMap.get(key);
+            const isAB = edge.from_node_id === nodeA;
+
+            // Add edge to appropriate direction
+            if (isAB) {
+                connection.direction_ab.push({
+                    snr: edge.snr,
+                    snr_db: edge.snr / 4, // Convert to dB
+                    created_at: edge.created_at,
+                    packet_id: edge.packet_id,
+                    source: edge.source,
+                });
+            } else {
+                connection.direction_ba.push({
+                    snr: edge.snr,
+                    snr_db: edge.snr / 4,
+                    created_at: edge.created_at,
+                    packet_id: edge.packet_id,
+                    source: edge.source,
+                });
+            }
+        }
+
+        // Aggregate each connection
+        const connections = Array.from(connectionsMap.values()).map(conn => {
+            // Deduplicate edges by packet_id for each direction (keep first occurrence, which is most recent)
+            const dedupeByPacketId = (edges) => {
+                const seen = new Set();
+                return edges.filter(edge => {
+                    if (seen.has(edge.packet_id)) {
+                        return false;
+                    }
+                    seen.add(edge.packet_id);
+                    return true;
+                });
+            };
+
+            const deduplicatedAB = dedupeByPacketId(conn.direction_ab);
+            const deduplicatedBA = dedupeByPacketId(conn.direction_ba);
+
+            // Calculate average SNR for A->B (using deduplicated edges)
+            const avgSnrAB = deduplicatedAB.length > 0
+                ? deduplicatedAB.reduce((sum, e) => sum + e.snr_db, 0) / deduplicatedAB.length
+                : null;
+
+            // Calculate average SNR for B->A (using deduplicated edges)
+            const avgSnrBA = deduplicatedBA.length > 0
+                ? deduplicatedBA.reduce((sum, e) => sum + e.snr_db, 0) / deduplicatedBA.length
+                : null;
+
+            // Get last 5 edges for each direction (already sorted by created_at DESC, packet_id DESC, now deduplicated)
+            const last5AB = deduplicatedAB.slice(0, 5);
+            const last5BA = deduplicatedBA.slice(0, 5);
+
+            // Determine worst average SNR
+            const worstAvgSnrDb = [avgSnrAB, avgSnrBA]
+                .filter(v => v !== null)
+                .reduce((min, val) => val < min ? val : min, Infinity);
+
+            return {
+                node_a: conn.node_a,
+                node_b: conn.node_b,
+                direction_ab: {
+                    avg_snr_db: avgSnrAB,
+                    last_5_edges: last5AB,
+                    total_count: deduplicatedAB.length, // Use deduplicated count
+                },
+                direction_ba: {
+                    avg_snr_db: avgSnrBA,
+                    last_5_edges: last5BA,
+                    total_count: deduplicatedBA.length, // Use deduplicated count
+                },
+                worst_avg_snr_db: worstAvgSnrDb !== Infinity ? worstAvgSnrDb : null,
+            };
+        }).filter(conn => conn.worst_avg_snr_db !== null); // Only return connections with at least one direction
+
+        res.json({
+            connections: connections,
+        });
+
+    } catch (err) {
+        console.error(err);
+        res.status(500).json({
+            message: "Something went wrong, try again later.",
+        });
+    }
+});
+
 app.get('/api/v1/nodes/:nodeId/position-history', async (req, res) => {
     try {
 
@@ -787,3 +1116,23 @@ const listener = app.listen(port, () => {
     const port = listener.address().port;
     console.log(`Server running at http://127.0.0.1:${port}`);
 });
+
+// Graceful shutdown handlers
+function gracefulShutdown(signal) {
+    console.log(`Received ${signal}. Starting graceful shutdown...`);
+    
+    // Stop accepting new connections
+    listener.close(async (err) => {        
+        console.log('HTTP server closed');
+        await prisma.$disconnect();
+        console.log('Database connections closed');
+        console.log('Graceful shutdown completed');
+        process.exit(0);
+    });
+}
+
+// Handle SIGTERM (Docker, systemd, etc.)
+process.on('SIGTERM', () => gracefulShutdown('SIGTERM'));
+
+// Handle SIGINT (Ctrl+C)
+process.on('SIGINT', () => gracefulShutdown('SIGINT'));

src/mqtt.js

@@ -1,3 +1,4 @@
+require('./utils/logger');
 const crypto = require("crypto");
 const path = require("path");
 const mqtt = require("mqtt");
@@ -263,8 +264,9 @@ const User = root.lookupType("User");
 const Waypoint = root.lookupType("Waypoint");
 
 // run automatic purge if configured
+let purgeInterval = null;
 if(purgeIntervalSeconds){
-    setInterval(async () => {
+    purgeInterval = setInterval(async () => {
         await purgeUnheardNodes();
         await purgeOldDeviceMetrics();
         await purgeOldEnvironmentMetrics();
@@ -740,6 +742,13 @@ client.on("message", async (topic, message) => {
             }
 
         }
+        
+        // check if bitfield is available, then set ok-to-mqtt
+        // else leave undefined to let Prisma ignore it.
+        let isOkToMqtt
+        if(bitfield != null){
+            isOkToMqtt = Boolean(bitfield & BITFIELD_OK_TO_MQTT_MASK);
+        }
 
         // create service envelope in db
         if(collectServiceEnvelopes){
@@ -817,6 +826,7 @@ client.on("message", async (topic, message) => {
                         rx_snr: envelope.packet.rxSnr,
                         rx_rssi: envelope.packet.rxRssi,
                         hop_limit: envelope.packet.hopLimit,
+                        ok_to_mqtt: isOkToMqtt,
                     },
                 });
             } catch (e) {
@@ -917,7 +927,6 @@ client.on("message", async (topic, message) => {
         else if(portnum === 4) {
 
             const user = User.decode(envelope.packet.decoded.payload);
-            let isOkToMqtt = null
 
             if(logKnownPacketTypes) {
                 console.log("NODEINFO_APP", {
@@ -926,11 +935,6 @@ client.on("message", async (topic, message) => {
                 });
             }
 
-            // check if bitfield is available, then check if ok-to-mqtt
-            if(bitfield != null){
-                isOkToMqtt = Boolean(bitfield & BITFIELD_OK_TO_MQTT_MASK);
-            }
-
             // create or update node in db
             try {
                 await prisma.node.upsert({
@@ -944,15 +948,19 @@ client.on("message", async (topic, message) => {
                         hardware_model: user.hwModel,
                         is_licensed: user.isLicensed === true,
                         role: user.role,
-                        // Since packages beeing forwarded by older firmwars dropps the bitfield
-                        // We only want to set form nodes that have the bitfield set.
-                        // That way we can get a more correct reading firmware status in the mesh.
-                        // This works since we had the old code:
-                        // firmware_version: (bitfield != null) ? '2.5.0 or newer' : '2.4.3 or older',
-                        ...(bitfield != null && {
-                            firmware_version: '2.5.0 or newer',
-                            ok_to_mqtt: isOkToMqtt,
-                        }),                            
+                        is_unmessagable: user.isUnmessagable,
+                        ok_to_mqtt: isOkToMqtt,
+                        max_hops: envelope.packet.hopStart,
+                        channel_id: envelope.channelId,
+
+                        firmware_version: '<2.5.0',
+                        ...(user.publicKey != '' && {
+                            firmware_version: '>2.5.0',
+                             public_key: user.publicKey?.toString("base64"),
+                        }),
+                        ...(user.isUnmessagable != null && {
+                            firmware_version: '>2.6.8',
+                        }),
                     },
                     update: {
                         long_name: user.longName,
@@ -960,14 +968,28 @@ client.on("message", async (topic, message) => {
                         hardware_model: user.hwModel,
                         is_licensed: user.isLicensed === true,
                         role: user.role,
-                        ...(bitfield != null && {
-                            firmware_version: '2.5.0 or newer',
-                            ok_to_mqtt: isOkToMqtt,
+                        is_unmessagable: user.isUnmessagable,
+                        ok_to_mqtt: isOkToMqtt,
+                        max_hops: envelope.packet.hopStart,
+                        channel_id: envelope.channelId,
+                        
+                        firmware_version: '<2.5.0',
+                        ...(user.publicKey != '' && {
+                            firmware_version: '>2.5.0',
+                            public_key: user.publicKey?.toString("base64"),
+                        }),
+                        ...(user.isUnmessagable != null && {
+                            firmware_version: '>2.6.8',
                         }),
                     },
                 });
             } catch (e) {
-                console.error(e);
+                // Ignore MySQL error 1020 "Record has changed since last read" - this is a race condition
+                // that occurs when multiple packets arrive concurrently for the same node
+                const errorMessage = e.message || String(e);
+                if (!errorMessage.includes('Record has changed since last read')) {
+                    console.error(e);
+                }
             }
 
             // Keep track of the names a node has been using.
@@ -990,7 +1012,12 @@ client.on("message", async (topic, message) => {
                     } 
                   });                  
             } catch (e) {
-                console.error(e);
+                // Ignore MySQL error 1020 "Record has changed since last read" - this is a race condition
+                // that occurs when multiple packets arrive concurrently for the same node
+                const errorMessage = e.message || String(e);
+                if (!errorMessage.includes('Record has changed since last read')) {
+                    console.error(e);
+                }
             }
         }
 
@@ -1067,6 +1094,70 @@ client.on("message", async (topic, message) => {
                 console.error(e);
             }
 
+            // Extract edges from neighbour info
+            try {
+                const toNodeId = envelope.packet.from;
+                const neighbors = neighbourInfo.neighbors || [];
+                const packetId = envelope.packet.id;
+                const channelId = envelope.channelId;
+                const gatewayId = envelope.gatewayId ? convertHexIdToNumericId(envelope.gatewayId) : null;
+                const edgesToCreate = [];
+
+                for(const neighbour of neighbors) {
+                    // Skip if no node ID
+                    if(!neighbour.nodeId) {
+                        continue;
+                    }
+
+                    // Skip if SNR is invalid (0 or null/undefined)
+                    // Note: SNR can be negative, so we check for 0 specifically
+                    if(neighbour.snr === 0 || neighbour.snr == null) {
+                        continue;
+                    }
+
+                    const fromNodeId = neighbour.nodeId;
+                    const snr = neighbour.snr;
+
+                    // Fetch node positions from Node table
+                    const [fromNode, toNode] = await Promise.all([
+                        prisma.node.findUnique({
+                            where: { node_id: fromNodeId },
+                            select: { latitude: true, longitude: true },
+                        }),
+                        prisma.node.findUnique({
+                            where: { node_id: toNodeId },
+                            select: { latitude: true, longitude: true },
+                        }),
+                    ]);
+
+                    // Create edge record
+                    edgesToCreate.push({
+                        from_node_id: fromNodeId,
+                        to_node_id: toNodeId,
+                        snr: snr,
+                        from_latitude: fromNode?.latitude ?? null,
+                        from_longitude: fromNode?.longitude ?? null,
+                        to_latitude: toNode?.latitude ?? null,
+                        to_longitude: toNode?.longitude ?? null,
+                        packet_id: packetId,
+                        channel_id: channelId,
+                        gateway_id: gatewayId,
+                        source: "NEIGHBORINFO_APP",
+                    });
+                }
+
+                // Bulk insert edges
+                if(edgesToCreate.length > 0) {
+                    await prisma.edge.createMany({
+                        data: edgesToCreate,
+                        skipDuplicates: true, // Skip if exact duplicate exists
+                    });
+                }
+            } catch (e) {
+                // Log error but don't crash - edge extraction is non-critical
+                console.error("Error extracting edges from neighbour info:", e);
+            }
+
             // don't store all neighbour infos, but we want to update the existing node above
             if(!collectNeighbourInfo){
                 return;
@@ -1309,6 +1400,160 @@ client.on("message", async (topic, message) => {
                 console.error(e);
             }
 
+            // Extract edges from traceroute (only for response packets)
+            if(!envelope.packet.decoded.wantResponse) {
+                try {
+                    const route = routeDiscovery.route || [];
+                    const snrTowards = routeDiscovery.snrTowards || [];
+                    const originNodeId = envelope.packet.to;
+                    const destinationNodeId = envelope.packet.from;
+                    const packetId = envelope.packet.id;
+                    const channelId = envelope.channelId;
+                    const gatewayId = envelope.gatewayId ? convertHexIdToNumericId(envelope.gatewayId) : null;
+
+                    // Determine number of edges: route.length + 1
+                    const numEdges = route.length + 1;
+                    const edgesToCreate = [];
+
+                    // Extract edges from the route path
+                    for(let i = 0; i < numEdges; i++) {
+                        // Get SNR for this edge
+                        if(i >= snrTowards.length) {
+                            // Array length mismatch - skip this edge
+                            continue;
+                        }
+
+                        const snr = snrTowards[i];
+                        
+                        // Skip if SNR is -128 (no SNR recorded)
+                        if(snr === -128) {
+                            continue;
+                        }
+
+                        // Determine from_node and to_node
+                        let fromNodeId, toNodeId;
+
+                        if(route.length === 0) {
+                            // Empty route: direct connection (to -> from)
+                            fromNodeId = originNodeId;
+                            toNodeId = destinationNodeId;
+                        } else if(i === 0) {
+                            // First edge: origin -> route[0]
+                            fromNodeId = originNodeId;
+                            toNodeId = route[0];
+                        } else if(i === route.length) {
+                            // Last edge: route[route.length-1] -> destination
+                            fromNodeId = route[route.length - 1];
+                            toNodeId = destinationNodeId;
+                        } else {
+                            // Middle edge: route[i-1] -> route[i]
+                            fromNodeId = route[i - 1];
+                            toNodeId = route[i];
+                        }
+
+                        // Fetch node positions from Node table
+                        const [fromNode, toNode] = await Promise.all([
+                            prisma.node.findUnique({
+                                where: { node_id: fromNodeId },
+                                select: { latitude: true, longitude: true },
+                            }),
+                            prisma.node.findUnique({
+                                where: { node_id: toNodeId },
+                                select: { latitude: true, longitude: true },
+                            }),
+                        ]);
+
+                        // Create edge record (skip if nodes don't exist, but still create edge with null positions)
+                        edgesToCreate.push({
+                            from_node_id: fromNodeId,
+                            to_node_id: toNodeId,
+                            snr: snr,
+                            from_latitude: fromNode?.latitude ?? null,
+                            from_longitude: fromNode?.longitude ?? null,
+                            to_latitude: toNode?.latitude ?? null,
+                            to_longitude: toNode?.longitude ?? null,
+                            packet_id: packetId,
+                            channel_id: channelId,
+                            gateway_id: gatewayId,
+                            source: "TRACEROUTE_APP",
+                        });
+                    }
+
+                    // Extract edges from route_back path
+                    const routeBack = routeDiscovery.routeBack || [];
+                    const snrBack = routeDiscovery.snrBack || [];
+                    
+                    if(routeBack.length > 0) {
+                        // Number of edges in route_back equals route_back.length
+                        for(let i = 0; i < routeBack.length; i++) {
+                            // Get SNR for this edge
+                            if(i >= snrBack.length) {
+                                // Array length mismatch - skip this edge
+                                continue;
+                            }
+
+                            const snr = snrBack[i];
+                            
+                            // Skip if SNR is -128 (no SNR recorded)
+                            if(snr === -128) {
+                                continue;
+                            }
+
+                            // Determine from_node and to_node
+                            let fromNodeId, toNodeId;
+
+                            if(i === 0) {
+                                // First edge: from -> route_back[0]
+                                fromNodeId = destinationNodeId; // 'from' in the packet
+                                toNodeId = routeBack[0];
+                            } else {
+                                // Subsequent edges: route_back[i-1] -> route_back[i]
+                                fromNodeId = routeBack[i - 1];
+                                toNodeId = routeBack[i];
+                            }
+
+                            // Fetch node positions from Node table
+                            const [fromNode, toNode] = await Promise.all([
+                                prisma.node.findUnique({
+                                    where: { node_id: fromNodeId },
+                                    select: { latitude: true, longitude: true },
+                                }),
+                                prisma.node.findUnique({
+                                    where: { node_id: toNodeId },
+                                    select: { latitude: true, longitude: true },
+                                }),
+                            ]);
+
+                            // Create edge record
+                            edgesToCreate.push({
+                                from_node_id: fromNodeId,
+                                to_node_id: toNodeId,
+                                snr: snr,
+                                from_latitude: fromNode?.latitude ?? null,
+                                from_longitude: fromNode?.longitude ?? null,
+                                to_latitude: toNode?.latitude ?? null,
+                                to_longitude: toNode?.longitude ?? null,
+                                packet_id: packetId,
+                                channel_id: channelId,
+                                gateway_id: gatewayId,
+                                source: "TRACEROUTE_APP",
+                            });
+                        }
+                    }
+
+                    // Bulk insert edges
+                    if(edgesToCreate.length > 0) {
+                        await prisma.edge.createMany({
+                            data: edgesToCreate,
+                            skipDuplicates: true, // Skip if exact duplicate exists
+                        });
+                    }
+                } catch (e) {
+                    // Log error but don't crash - edge extraction is non-critical
+                    console.error("Error extracting edges from traceroute:", e);
+                }
+            }
+
         }
 
         else if(portnum === 73) {
@@ -1413,6 +1658,7 @@ client.on("message", async (topic, message) => {
                     || portnum === 0 // ignore UNKNOWN_APP
                     || portnum === 1 // ignore TEXT_MESSAGE_APP
                     || portnum === 5 // ignore ROUTING_APP
+                    || portnum === 6 // ignore ADMIN_APP
                     || portnum === 34 // ignore PAXCOUNTER_APP
                     || portnum === 65 // ignore STORE_FORWARD_APP
                     || portnum === 66 // ignore RANGE_TEST_APP
@@ -1429,6 +1675,32 @@ client.on("message", async (topic, message) => {
         }
 
     } catch(e) {
-        // ignore errors
+        console.log("error", e);
     }
 });
+
+// Graceful shutdown handlers
+function gracefulShutdown(signal) {
+    console.log(`Received ${signal}. Starting graceful shutdown...`);
+    
+    // Clear the purge interval if it exists
+    if(purgeInterval) {
+        clearInterval(purgeInterval);
+        console.log('Purge interval cleared');
+    }
+    
+    // Close MQTT client
+    client.end(false, async () => {
+        console.log('MQTT client disconnected');
+        await prisma.$disconnect();
+        console.log('Database connections closed');
+        console.log('Graceful shutdown completed');
+        process.exit(0);
+    });
+}
+
+// Handle SIGTERM (Docker, systemd, etc.)
+process.on('SIGTERM', () => gracefulShutdown('SIGTERM'));
+
+// Handle SIGINT (Ctrl+C)
+process.on('SIGINT', () => gracefulShutdown('SIGINT'));

src/stats.js

@@ -4,7 +4,7 @@ const router = express.Router();
 const protobufjs = require("protobufjs");
 
 // create prisma db client
-const { PrismaClient } = require("@prisma/client");
+const { Prisma, PrismaClient } = require("@prisma/client");
 const prisma = new PrismaClient();
 
 // load protobufs
@@ -21,6 +21,12 @@ router.get('/hardware-models', async (req, res) => {
         // get nodes from db
         const results = await prisma.node.groupBy({
             by: ['hardware_model'],
+            where: {
+                 // Since we removed retention; only include nodes that have been updated in the last 30 days
+                updated_at: {
+                    gte: new Date(Date.now() - 30 * 24 * 60 * 60 * 1000) // within last 30 days
+                }
+            },
             orderBy: {
                 _count: {
                     hardware_model: 'desc',
@@ -64,19 +70,19 @@ router.get('/messages-per-hour', async (req, res) => {
             orderBy: { created_at: 'asc' }
         });
 
-        // Pre-fill `uniqueCounts` with zeros for all hours
+        // Pre-fill `uniqueCounts` with zeros for all hours, including the current hour
         const uniqueCounts = Object.fromEntries(
             Array.from({ length: hours }, (_, i) => {
-                const hourTime = new Date(now.getTime() - (hours - i) * 60 * 60 * 1000);
-                const hourString = hourTime.toISOString().slice(0, 13); // YYYY-MM-DD HH
+                const hourTime = new Date(now.getTime() - (hours - 1 - i) * 60 * 60 * 1000);
+                const hourString = hourTime.toISOString().slice(0, 13) + ":00:00.000Z"; // zero out the minutes and seconds
                 return [hourString, 0];
             })
         );
 
         // Populate actual message counts
         messages.forEach(({ created_at }) => {
-            const hourString = created_at.toISOString().slice(0, 13); // YYYY-MM-DD HH
-            uniqueCounts[hourString]++;
+            const hourString = created_at.toISOString().slice(0, 13) + ":00:00.000Z"; // zero out the minutes and seconds
+            uniqueCounts[hourString] = (uniqueCounts[hourString] ?? 0) + 1;
         });
 
         // Convert to final result format
@@ -91,7 +97,9 @@ router.get('/messages-per-hour', async (req, res) => {
 
 router.get('/most-active-nodes', async (req, res) => {
     try {
-        const result = await prisma.$queryRaw`
+        const channelId = req.query.channel_id;
+        const result = await prisma.$queryRaw(
+            Prisma.sql`
         SELECT n.long_name, COUNT(*) AS count
         FROM (
             SELECT DISTINCT \`from\`, packet_id
@@ -101,12 +109,14 @@ router.get('/most-active-nodes', async (req, res) => {
                 AND packet_id IS NOT NULL
                 AND portnum != 73
                 AND \`to\` != 1
+                ${channelId ? Prisma.sql`AND channel_id = ${channelId}` : Prisma.sql``}
         ) AS unique_packets
         JOIN nodes n ON unique_packets.from = n.node_id
         GROUP BY n.long_name
         ORDER BY count DESC
         LIMIT 25;
-        `;
+            `
+        );
   
       res.set('Cache-Control', 'public, max-age=600'); // 10 min cache
       res.json(result);
@@ -119,6 +129,7 @@ router.get('/most-active-nodes', async (req, res) => {
 
 router.get('/portnum-counts', async (req, res) => {
     const nodeId = req.query.nodeId ? parseInt(req.query.nodeId, 10) : null;
+    const channelId = req.query.channel_id;
     const hours = 24;
     const now = new Date();
     const startTime = new Date(now.getTime() - hours * 60 * 60 * 1000);
@@ -128,6 +139,7 @@ router.get('/portnum-counts', async (req, res) => {
             where: {
                 created_at: { gte: startTime },
                 ...(Number.isInteger(nodeId) ? { from: nodeId } : {}),
+                ...(channelId ? { channel_id: channelId } : {}),
                 packet_id: { not: null },
                 to: { not: 1 }, // Filter out NODENUM_BROADCAST_NO_LORA
                 OR: [
@@ -185,21 +197,48 @@ router.get('/battery-stats', async (req, res) => {
 });
 
 router.get('/channel-utilization-stats', async (req, res) => {
-    const days = parseInt(req.query.days || '1', 10);
-
+    const days = parseInt(req.query.days || '1', 10);  
+    const channelId = req.query.channel_id; // optional string
+  
     try {
-        const stats = await prisma.$queryRaw`
-            SELECT recorded_at, avg_channel_utilization
-            FROM channel_utilization_stats
-            WHERE recorded_at >= NOW() - INTERVAL ${days} DAY
-            ORDER BY recorded_at DESC;
-        `;
-
-        res.json(stats);
+      const stats = await prisma.$queryRaw(
+        Prisma.sql`
+          SELECT recorded_at, channel_id, avg_channel_utilization
+          FROM channel_utilization_stats
+          WHERE recorded_at >= NOW() - INTERVAL ${days} DAY
+          ${channelId ? Prisma.sql`AND channel_id = ${channelId}` : Prisma.sql``}
+          ORDER BY recorded_at DESC;
+        `
+      );
+  
+      res.json(stats);
     } catch (err) {
-        console.error('Error fetching channel utilization stats:', err);
-        res.status(500).json({ error: 'Internal server error' });
+      console.error('Error fetching channel utilization stats:', err);
+      res.status(500).json({ error: 'Internal server error' });
     }
-});
+  });
+
+  router.get('/channel-utilization', async (req, res) => {
+    const channelId = req.query.channel_id;
+  
+    try {
+      const snapshot = await prisma.$queryRaw(
+        Prisma.sql`
+          SELECT recorded_at, channel_id, avg_channel_utilization
+          FROM channel_utilization_stats
+          WHERE recorded_at = (
+            SELECT MAX(recorded_at) FROM channel_utilization_stats
+          )
+          ${channelId ? Prisma.sql`AND channel_id = ${channelId}` : Prisma.sql``}
+          ORDER BY channel_id;
+        `
+      );
+  
+      res.json(snapshot);
+    } catch (err) {
+      console.error('Error fetching latest channel utilization:', err);
+      res.status(500).json({ error: 'Internal server error' });
+    }
+  });
 
 module.exports = router;

src/utils/logger.js

@@ -0,0 +1,33 @@
+// Override console methods to add formatted timestamps
+const originalLog = console.log;
+const originalError = console.error;
+const originalWarn = console.warn;
+const originalInfo = console.info;
+
+function formatTimestamp() {
+    const now = new Date();
+    const year = now.getFullYear();
+    const month = String(now.getMonth() + 1).padStart(2, '0');
+    const day = String(now.getDate()).padStart(2, '0');
+    const hours = String(now.getHours()).padStart(2, '0');
+    const minutes = String(now.getMinutes()).padStart(2, '0');
+    const seconds = String(now.getSeconds()).padStart(2, '0');
+    return `${year}-${month}-${day} ${hours}:${minutes}:${seconds}`;
+}
+
+console.log = function(...args) {
+    originalLog(`${formatTimestamp()} [Info]`, ...args);
+};
+
+console.error = function(...args) {
+    originalError(`${formatTimestamp()} [Error]`, ...args);
+};
+
+console.warn = function(...args) {
+    originalWarn(`${formatTimestamp()} [Warn]`, ...args);
+};
+
+console.info = function(...args) {
+    originalInfo(`${formatTimestamp()} [Info]`, ...args);
+};
+

src/ws.js

@@ -0,0 +1,314 @@
+require('./utils/logger');
+const crypto = require("crypto");
+const path = require("path");
+const http = require("http");
+const mqtt = require("mqtt");
+const protobufjs = require("protobufjs");
+const commandLineArgs = require("command-line-args");
+const commandLineUsage = require("command-line-usage");
+const { WebSocketServer } = require("ws");
+
+const optionsList = [
+    {
+        name: 'help',
+        alias: 'h',
+        type: Boolean,
+        description: 'Display this usage guide.'
+    },
+    {
+        name: "mqtt-broker-url",
+        type: String,
+        description: "MQTT Broker URL (e.g: mqtt://mqtt.meshtastic.org)",
+    },
+    {
+        name: "mqtt-username",
+        type: String,
+        description: "MQTT Username (e.g: meshdev)",
+    },
+    {
+        name: "mqtt-password",
+        type: String,
+        description: "MQTT Password (e.g: large4cats)",
+    },
+    {
+        name: "mqtt-client-id",
+        type: String,
+        description: "MQTT Client ID (e.g: map.example.com)",
+    },
+    {
+        name: "mqtt-topic",
+        type: String,
+        multiple: true,
+        typeLabel: '<topic> ...',
+        description: "MQTT Topic to subscribe to (e.g: msh/#)",
+    },
+    {
+        name: "decryption-keys",
+        type: String,
+        multiple: true,
+        typeLabel: '<base64DecryptionKey> ...',
+        description: "Decryption keys encoded in base64 to use when decrypting service envelopes.",
+    },
+    {
+        name: "ws-port",
+        type: Number,
+        description: "WebSocket server port (default: 8081)",
+    },
+];
+
+// parse command line args
+const options = commandLineArgs(optionsList);
+
+// show help
+if(options.help){
+    const usage = commandLineUsage([
+        {
+            header: 'Meshtastic WebSocket Publisher',
+            content: 'Publishes real-time Meshtastic packets via WebSocket.',
+        },
+        {
+            header: 'Options',
+            optionList: optionsList,
+        },
+    ]);
+    console.log(usage);
+    process.exit(0);
+}
+
+// get options and fallback to default values
+const mqttBrokerUrl = options["mqtt-broker-url"] ?? "mqtt://mqtt.meshtastic.org";
+const mqttUsername = options["mqtt-username"] ?? "meshdev";
+const mqttPassword = options["mqtt-password"] ?? "large4cats";
+const mqttClientId = options["mqtt-client-id"] ?? null;
+const mqttTopics = options["mqtt-topic"] ?? ["msh/#"];
+const decryptionKeys = options["decryption-keys"] ?? [
+    "1PG7OiApB1nwvP+rz05pAQ==", // add default "AQ==" decryption key
+];
+const wsPort = options["ws-port"] ?? 8081;
+
+// create mqtt client
+const client = mqtt.connect(mqttBrokerUrl, {
+    username: mqttUsername,
+    password: mqttPassword,
+    clientId: mqttClientId,
+});
+
+// load protobufs
+const root = new protobufjs.Root();
+root.resolvePath = (origin, target) => path.join(__dirname, "protobufs", target);
+root.loadSync('meshtastic/mqtt.proto');
+const Data = root.lookupType("Data");
+const ServiceEnvelope = root.lookupType("ServiceEnvelope");
+const RouteDiscovery = root.lookupType("RouteDiscovery");
+
+// create HTTP server for WebSocket
+const server = http.createServer();
+const wss = new WebSocketServer({ server });
+
+// track connected clients
+const clients = new Set();
+
+wss.on('connection', (ws) => {
+    clients.add(ws);
+    console.log(`WebSocket client connected. Total clients: ${clients.size}`);
+
+    ws.on('close', () => {
+        clients.delete(ws);
+        console.log(`WebSocket client disconnected. Total clients: ${clients.size}`);
+    });
+
+    ws.on('error', (error) => {
+        console.error('WebSocket error:', error);
+        clients.delete(ws);
+    });
+});
+
+// broadcast message to all connected clients
+function broadcast(message) {
+    const messageStr = JSON.stringify(message);
+    clients.forEach((client) => {
+        if (client.readyState === 1) { // WebSocket.OPEN
+            try {
+                client.send(messageStr);
+            } catch (error) {
+                console.error('Error sending message to client:', error);
+            }
+        }
+    });
+}
+
+function createNonce(packetId, fromNode) {
+    // Expand packetId to 64 bits
+    const packetId64 = BigInt(packetId);
+
+    // Initialize block counter (32-bit, starts at zero)
+    const blockCounter = 0;
+
+    // Create a buffer for the nonce
+    const buf = Buffer.alloc(16);
+
+    // Write packetId, fromNode, and block counter to the buffer
+    buf.writeBigUInt64LE(packetId64, 0);
+    buf.writeUInt32LE(fromNode, 8);
+    buf.writeUInt32LE(blockCounter, 12);
+
+    return buf;
+}
+
+/**
+ * References:
+ * https://github.com/crypto-smoke/meshtastic-go/blob/develop/radio/aes.go#L42
+ * https://github.com/pdxlocations/Meshtastic-MQTT-Connect/blob/main/meshtastic-mqtt-connect.py#L381
+ */
+function decrypt(packet) {
+    // attempt to decrypt with all available decryption keys
+    for(const decryptionKey of decryptionKeys){
+        try {
+            // convert encryption key to buffer
+            const key = Buffer.from(decryptionKey, "base64");
+
+            // create decryption iv/nonce for this packet
+            const nonceBuffer = createNonce(packet.id, packet.from);
+
+            // determine algorithm based on key length
+            var algorithm = null;
+            if(key.length === 16){
+                algorithm = "aes-128-ctr";
+            } else if(key.length === 32){
+                algorithm = "aes-256-ctr";
+            } else {
+                // skip this key, try the next one...
+                console.error(`Skipping decryption key with invalid length: ${key.length}`);
+                continue;
+            }
+
+            // create decipher
+            const decipher = crypto.createDecipheriv(algorithm, key, nonceBuffer);
+
+            // decrypt encrypted packet
+            const decryptedBuffer = Buffer.concat([decipher.update(packet.encrypted), decipher.final()]);
+
+            // parse as data message
+            return Data.decode(decryptedBuffer);
+
+        } catch(e){}
+    }
+
+    // couldn't decrypt
+    return null;
+}
+
+/**
+ * converts hex id to numeric id, for example: !FFFFFFFF to 4294967295
+ * @param hexId a node id in hex format with a prepended "!"
+ * @returns {bigint} the node id in numeric form
+ */
+function convertHexIdToNumericId(hexId) {
+    return BigInt('0x' + hexId.replaceAll("!", ""));
+}
+
+// subscribe to everything when connected
+client.on("connect", () => {
+    console.log("Connected to MQTT broker");
+    for(const mqttTopic of mqttTopics){
+        client.subscribe(mqttTopic);
+        console.log(`Subscribed to MQTT topic: ${mqttTopic}`);
+    }
+});
+
+// handle message received
+client.on("message", async (topic, message) => {
+    try {
+        // decode service envelope
+        const envelope = ServiceEnvelope.decode(message);
+        if(!envelope.packet){
+            return;
+        }
+
+        // attempt to decrypt encrypted packets
+        const isEncrypted = envelope.packet.encrypted?.length > 0;
+        if(isEncrypted){
+            const decoded = decrypt(envelope.packet);
+            if(decoded){
+                envelope.packet.decoded = decoded;
+            }
+        }
+
+        // get portnum from decoded packet
+        const portnum = envelope.packet?.decoded?.portnum;
+
+        // check if we can see the decrypted packet data
+        if(envelope.packet.decoded == null){
+            return;
+        }
+
+        // handle traceroutes (portnum 70)
+        if(portnum === 70) {
+            try {
+                const routeDiscovery = RouteDiscovery.decode(envelope.packet.decoded.payload);
+                
+                const traceroute = {
+                    type: "traceroute",
+                    data: {
+                        to: envelope.packet.to,
+                        from: envelope.packet.from,
+                        want_response: envelope.packet.decoded.wantResponse,
+                        route: routeDiscovery.route,
+                        snr_towards: routeDiscovery.snrTowards,
+                        route_back: routeDiscovery.routeBack,
+                        snr_back: routeDiscovery.snrBack,
+                        channel_id: envelope.channelId,
+                        gateway_id: envelope.gatewayId ? Number(convertHexIdToNumericId(envelope.gatewayId)) : null,
+                        packet_id: envelope.packet.id,
+                    }
+                };
+                broadcast(traceroute);
+            } catch (e) {
+                console.error("Error processing traceroute:", e);
+            }
+        }
+
+    } catch(e) {
+        console.error("Error processing MQTT message:", e);
+    }
+});
+
+// start WebSocket server
+server.listen(wsPort, () => {
+    console.log(`WebSocket server running on port ${wsPort}`);
+});
+
+// Graceful shutdown handlers
+function gracefulShutdown(signal) {
+    console.log(`Received ${signal}. Starting graceful shutdown...`);
+    
+    // Close all WebSocket connections
+    clients.forEach((client) => {
+        client.close();
+    });
+    clients.clear();
+    
+    // Close WebSocket server
+    wss.close(() => {
+        console.log('WebSocket server closed');
+    });
+    
+    // Close HTTP server
+    server.close(() => {
+        console.log('HTTP server closed');
+    });
+    
+    // Close MQTT client
+    client.end(false, () => {
+        console.log('MQTT client disconnected');
+        console.log('Graceful shutdown completed');
+        process.exit(0);
+    });
+}
+
+// Handle SIGTERM (Docker, systemd, etc.)
+process.on('SIGTERM', () => gracefulShutdown('SIGTERM'));
+
+// Handle SIGINT (Ctrl+C)
+process.on('SIGINT', () => gracefulShutdown('SIGINT'));
+