ingest: meshcore MQTT->ClickHouse stack

Vendor the ingest service under ingest/ and move the web app under meshexplorer/.
The ingest builds the meshcoreingest daemon and the goose migration runner,
applies the meshcore ClickHouse schema (packets, adverts, unified node view),
and loads its MQTT broker list and ClickHouse settings entirely from environment
variables (MQTT_BROKERS as a JSON array, CLICKHOUSE_*). No credentials are baked
into the source.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

ingest/.gitignore

@@ -0,0 +1,59 @@
+# Binaries for programs and plugins
+*.exe
+*.exe~
+*.dll
+*.so
+*.dylib
+
+# Test binary, built with `go test -c`
+*.test
+
+# Output of the go coverage tool, specifically when used with LiteIDE
+*.out
+
+# Dependency directories (remove the comment below to include it)
+# vendor/
+
+# Go workspace file
+go.work
+
+# Build artifacts
+clickhouse-meshingest
+clickhouse-meshingest.exe
+
+# IDE files
+.vscode/
+.idea/
+*.swp
+*.swo
+*~
+
+# OS generated files
+.DS_Store
+.DS_Store?
+._*
+.Spotlight-V100
+.Trashes
+ehthumbs.db
+Thumbs.db
+
+# Log files
+*.log
+
+# Configuration files with sensitive data
+config.json
+.env
+
+# Temporary files
+tmp/
+temp/
+
+mqtt/internal/data/
+mqtt/external/data/
+
+tiles/
+cache/
+
+splat*
+
+.env

ingest/Dockerfile

@@ -0,0 +1,57 @@
+# syntax=docker/dockerfile:1
+FROM golang:1.24-alpine AS builder
+
+# Install build dependencies
+RUN apk add --no-cache git ca-certificates tzdata
+
+# Set working directory
+WORKDIR /app
+
+# Copy go mod files for dependency caching
+COPY go.mod go.sum ./
+
+# Download dependencies (cached by BuildKit)
+RUN --mount=type=cache,target=/go/pkg/mod \
+    go mod download
+
+# Copy source code
+COPY . .
+
+# Build the meshcore ingest daemon
+RUN --mount=type=cache,target=/go/pkg/mod \
+    --mount=type=cache,target=/root/.cache/go-build \
+    CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -o meshcoreingest ./cmd/meshcoreingest
+
+# Build the migration runner
+RUN --mount=type=cache,target=/go/pkg/mod \
+    --mount=type=cache,target=/root/.cache/go-build \
+    CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -o migrate ./internal/migrate
+
+# Final stage
+FROM alpine:latest
+
+# Install runtime dependencies
+RUN apk --no-cache add ca-certificates tzdata
+
+# Create non-root user
+RUN addgroup -g 1001 -S appgroup && \
+    adduser -u 1001 -S appuser -G appgroup
+
+# Set working directory
+WORKDIR /app
+
+# Copy binaries from builder stage
+COPY --from=builder /app/meshcoreingest .
+COPY --from=builder /app/migrate .
+
+# Bundle migrations so the migrate service can apply them on startup
+COPY --from=builder /app/migrations ./migrations
+
+# Change ownership to non-root user
+RUN chown -R appuser:appgroup /app
+
+# Switch to non-root user
+USER appuser
+
+# Run the meshcore ingest daemon by default; the migrate service overrides this.
+CMD ["./meshcoreingest"]

ingest/README.md

@@ -0,0 +1,59 @@
+# MeshCore Ingest
+
+A Go service that ingests MeshCore MQTT messages into ClickHouse, plus the
+ClickHouse image and SQL migrations for the schema.
+
+This directory is normally run as part of the full stack via the
+[root `docker compose`](../README.md). The notes below cover running and
+developing it on its own.
+
+## Components
+
+- `cmd/meshcoreingest` — the ingest daemon. Subscribes to MeshCore MQTT topics
+  and writes raw packets into the `meshcore_packets` table.
+- `internal/ingestcommon` — shared MQTT + ClickHouse connection/daemon logic.
+- `internal/migrate` — a [goose](https://github.com/pressly/goose) based
+  migration runner (ClickHouse dialect).
+- `migrations/` — the ClickHouse schema: the `meshcore_packets` table, the decoded
+  `meshcore_adverts` / `meshcore_adverts_latest` / `meshcore_public_channel_messages`
+  views, and the `unified_latest_nodeinfo` view consumed by the web app.
+- `clickhouse/` — a thin ClickHouse server image plus the read-only user used by
+  the web app.
+
+## Configuration
+
+All configuration is via environment variables (no credentials are baked into the
+source):
+
+| Variable | Description |
+|----------|-------------|
+| `MQTT_BROKERS` | JSON array of brokers: `[{"url","username","password","topics"}]`. `topics` defaults to `["meshcore/#"]`. Required; the daemon exits if unset. |
+| `MQTT_CLIENT_ID` | MQTT client id prefix (default `meshcore-ingest`). |
+| `CLICKHOUSE_HOST` / `CLICKHOUSE_PORT` | ClickHouse address (native protocol, default `127.0.0.1:9000`). |
+| `CLICKHOUSE_DB` / `CLICKHOUSE_USER` / `CLICKHOUSE_PASSWORD` | ClickHouse database and read/write credentials. |
+
+## Building
+
+```bash
+go build ./...
+go test ./...
+```
+
+## Running migrations
+
+```bash
+go run ./internal/migrate \
+  -host localhost -port 9000 \
+  -username default -password "$CLICKHOUSE_PASSWORD" \
+  -path migrations -action up
+```
+
+Actions: `up`, `down`, `reset`, `status`, `version`.
+
+## Running the ingest daemon
+
+```bash
+export MQTT_BROKERS='[{"url":"tcp://mqtt.example.com:1883","username":"u","password":"p","topics":["meshcore/#"]}]'
+export CLICKHOUSE_HOST=localhost CLICKHOUSE_PORT=9000 CLICKHOUSE_PASSWORD=...
+go run ./cmd/meshcoreingest
+```

ingest/clickhouse/Dockerfile

@@ -0,0 +1,26 @@
+FROM clickhouse/clickhouse-server:latest as builder
+
+RUN apt-get update && \
+    apt-get install -y build-essential libssl-dev libmsgpack-dev && \
+    rm -rf /var/lib/apt/lists/*
+
+WORKDIR /build
+COPY meshcore_decrypt_udf.c ./
+RUN gcc -O2 -o meshcore_decrypt_udf meshcore_decrypt_udf.c -lssl -lcrypto -lmsgpackc
+
+FROM clickhouse/clickhouse-server:latest
+
+# Copy UDF binary to /usr/local/share/clickhouse-udf-bins
+COPY --from=builder /build/meshcore_decrypt_udf /usr/local/share/clickhouse-udf-bins/meshcore_decrypt_udf
+
+# Copy UDF XML config directly to the location and name matching ClickHouse config pattern
+COPY meshcore_decrypt_udf.xml /etc/clickhouse-server/meshcore_decrypt_function.xml
+
+# Copy users configuration file
+COPY users.xml /etc/clickhouse-server/users.d/readonly_user.xml
+
+# Copy access control configuration
+COPY config.xml /etc/clickhouse-server/config.d/access_control.xml
+
+RUN chmod +x /usr/local/share/clickhouse-udf-bins/meshcore_decrypt_udf
+# Remove custom entrypoint, use default 

ingest/clickhouse/config.xml

@@ -0,0 +1,5 @@
+<clickhouse>
+    <access_control_improvements>
+        <settings_constraints_replace_previous>true</settings_constraints_replace_previous>
+    </access_control_improvements>
+</clickhouse>

ingest/clickhouse/meshcore_decrypt_udf.c

@@ -0,0 +1,99 @@
+#include <stdio.h>
+#include <string.h>
+#include <openssl/aes.h>
+#include <openssl/hmac.h>
+#include <openssl/sha.h>
+#include <msgpack.h>
+
+#define CIPHER_MAC_SIZE 2
+#define CIPHER_KEY_SIZE 16
+#define MAX_PAYLOAD 184
+
+// Default "public" channel key
+static const unsigned char public_channel_private_key[16] = {
+    0x8B, 0x33, 0x87, 0xE5, 0xCD, 0xEA, 0x6A, 0xC9, 
+    0xE5, 0xED, 0xBA, 0xA1, 0x15, 0xCD, 0x72, 0x0F
+};
+
+static int decrypt_msg(const unsigned char *key, const unsigned char *data, size_t len, unsigned char *out, size_t *out_len) {
+    if (len <= CIPHER_MAC_SIZE) return 0;
+    
+    unsigned char mac[SHA256_DIGEST_LENGTH];
+    HMAC(EVP_sha256(), key, CIPHER_KEY_SIZE, data + CIPHER_MAC_SIZE, 
+         len - CIPHER_MAC_SIZE, mac, NULL);
+    
+    if (memcmp(mac, data, CIPHER_MAC_SIZE) != 0) return 0;
+    
+    AES_KEY aes_key;
+    AES_set_decrypt_key(key, 128, &aes_key);
+    
+    size_t blocks = (len - CIPHER_MAC_SIZE) / 16;
+    for (size_t i = 0; i < blocks; i++) {
+        AES_decrypt(data + CIPHER_MAC_SIZE + (i * 16), out + (i * 16), &aes_key);
+    }
+    *out_len = blocks * 16;
+    return 1;
+}
+
+int main() {
+    unsigned char buf[8192];
+    size_t len = fread(buf, 1, sizeof(buf), stdin);
+    if (len == 0) return 0;
+    
+    msgpack_unpacked result;
+    msgpack_unpacked_init(&result);
+    
+    size_t off = 0;
+    msgpack_unpack_return ret = msgpack_unpack_next(&result, (char*)buf, len, &off);
+    
+    if (ret == MSGPACK_UNPACK_SUCCESS && result.data.type == MSGPACK_OBJECT_ARRAY) {
+        msgpack_object_array arr = result.data.via.array;
+        msgpack_sbuffer sbuf;
+        msgpack_sbuffer_init(&sbuf);
+        msgpack_packer pk;
+        msgpack_packer_init(&pk, &sbuf, msgpack_sbuffer_write);
+        
+        msgpack_pack_array(&pk, arr.size);
+        
+        for (size_t i = 0; i < arr.size; i++) {
+            msgpack_object row = arr.ptr[i];
+            if (row.type == MSGPACK_OBJECT_ARRAY && row.via.array.size >= 3) {
+                msgpack_object mac_obj = row.via.array.ptr[0];
+                msgpack_object enc_obj = row.via.array.ptr[1];
+                msgpack_object hash_obj = row.via.array.ptr[2];
+                if (mac_obj.type == MSGPACK_OBJECT_STR && enc_obj.type == MSGPACK_OBJECT_STR && hash_obj.type == MSGPACK_OBJECT_STR && hash_obj.via.str.size > 0) {
+                    // Concatenate mac + encrypted_data
+                    size_t mac_len = mac_obj.via.str.size;
+                    size_t enc_len = enc_obj.via.str.size;
+                    unsigned char mac_and_enc[MAX_PAYLOAD];
+                    if (mac_len + enc_len > MAX_PAYLOAD) { msgpack_pack_nil(&pk); continue; }
+                    memcpy(mac_and_enc, mac_obj.via.str.ptr, mac_len);
+                    memcpy(mac_and_enc + mac_len, enc_obj.via.str.ptr, enc_len);
+                    unsigned char decrypted[MAX_PAYLOAD];
+                    size_t decrypted_len;
+                    if (decrypt_msg(public_channel_private_key, mac_and_enc, mac_len + enc_len, decrypted, &decrypted_len)) {
+                        msgpack_pack_array(&pk, 2);
+                        msgpack_pack_str(&pk, decrypted_len);
+                        msgpack_pack_str_body(&pk, (char*)decrypted, decrypted_len);
+                        unsigned char full_hash[SHA256_DIGEST_LENGTH];
+                        SHA256(public_channel_private_key, CIPHER_KEY_SIZE, full_hash);
+                        msgpack_pack_str(&pk, SHA256_DIGEST_LENGTH);
+                        msgpack_pack_str_body(&pk, (char*)full_hash, SHA256_DIGEST_LENGTH);
+                    } else {
+                        msgpack_pack_nil(&pk);
+                    }
+                } else {
+                    msgpack_pack_nil(&pk);
+                }
+            } else {
+                msgpack_pack_nil(&pk);
+            }
+        }
+        
+        fwrite(sbuf.data, 1, sbuf.size, stdout);
+        msgpack_sbuffer_destroy(&sbuf);
+    }
+    
+    msgpack_unpacked_destroy(&result);
+    return 0;
+} 

ingest/clickhouse/meshcore_decrypt_udf.xml

@@ -0,0 +1,30 @@
+<?xml version="1.0"?>
+<clickhouse>
+    <function>
+        <name>meshcore_try_decrypt</name>
+        <type>executable</type>
+        <command>/usr/local/share/clickhouse-udf-bins/meshcore_decrypt_udf</command>
+        <argument>
+            <type>String</type>
+            <name>mac</name>
+        </argument>
+        <argument>
+            <type>String</type>
+            <name>encrypted_data</name>
+        </argument>
+        <argument>
+            <type>String</type>
+            <name>channel_hash_raw</name>
+        </argument>
+        <format>MsgPack</format>
+        <return_type>Nullable(Array(String))</return_type>
+        <return_name>result</return_name>
+        <execute_direct>1</execute_direct>
+        <max_command_execution_time>30</max_command_execution_time>
+        <command_termination_timeout>10</command_termination_timeout>
+        <command_read_timeout>10000</command_read_timeout>
+        <command_write_timeout>10000</command_write_timeout>
+        <send_chunk_header>false</send_chunk_header>
+        <deterministic>true</deterministic>
+    </function>
+</clickhouse> 

ingest/clickhouse/udf-config.xml

@@ -0,0 +1,4 @@
+<clickhouse>
+    <user_defined_executable_functions_config>/usr/local/share/clickhouse-udfs/*.xml</user_defined_executable_functions_config>
+    <user_scripts_path>/usr/local/share/clickhouse-udf-bins/</user_scripts_path>
+</clickhouse> 

ingest/clickhouse/users.xml

@@ -0,0 +1,33 @@
+<clickhouse>
+    <users>
+        <!-- Readonly user configuration. Used by the web UI / discord bot.
+             Only reachable on the internal docker network, so the password is fine inline. -->
+        <readonly>
+            <password>readonly</password>
+            <networks incl="networks" replace="replace">
+                <ip>::/0</ip>
+            </networks>
+            <profile>readonly</profile>
+            <quota>default</quota>
+        </readonly>
+    </users>
+    
+    <profiles>
+        <!-- Profile for readonly user -->
+        <readonly>
+            <readonly>1</readonly>
+            <allow_ddl>0</allow_ddl>
+            <max_memory_usage>10000000000</max_memory_usage>
+            <max_execution_time>300</max_execution_time>
+            <max_rows_to_read>10000000</max_rows_to_read>
+            <max_bytes_to_read>500000000</max_bytes_to_read>
+            <constraints>
+                <max_execution_time>
+                    <changeable_in_readonly/>
+                    <min>0</min>
+                    <max>180</max>
+                </max_execution_time>
+            </constraints>
+        </readonly>
+    </profiles>
+</clickhouse>

ingest/cmd/meshcoreingest/main.go

@@ -0,0 +1,334 @@
+package main
+
+import (
+	"encoding/hex"
+	"encoding/json"
+	"log"
+	"os"
+	"os/signal"
+	"strings"
+	"syscall"
+	"time"
+
+	"github.com/ClickHouse/ch-go/proto"
+	"github.com/ajvpot/clickhouse-meshingest/internal/ingestcommon"
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
+)
+
+func parseMeshCoreRawMessage(payload []byte) (origin string, originPubkey []byte, meshTimestamp time.Time, packet []byte, err error) {
+	type RawPacket struct {
+		Origin    string `json:"origin"`
+		OriginID  string `json:"origin_id"`
+		Timestamp string `json:"timestamp"`
+		Type      string `json:"type"`
+		Data      string `json:"data"`
+	}
+	var pkt RawPacket
+	if err = json.Unmarshal(payload, &pkt); err != nil {
+		return "", nil, time.Time{}, nil, err
+	}
+	packet, err = hex.DecodeString(pkt.Data)
+	if err != nil {
+		return "", nil, time.Time{}, nil, err
+	}
+
+	// Clean origin and origin_id by removing carriage returns and newlines
+	cleanOrigin := strings.ReplaceAll(strings.ReplaceAll(pkt.Origin, "\r", ""), "\n", "")
+	cleanOriginID := strings.ReplaceAll(strings.ReplaceAll(pkt.OriginID, "\r", ""), "\n", "")
+
+	// Decode origin_id hex to binary bytes for compact storage
+	var originPubkeyBytes []byte
+	if cleanOriginID != "" {
+		if decoded, decErr := hex.DecodeString(cleanOriginID); decErr == nil {
+			originPubkeyBytes = decoded
+		} else {
+			originPubkeyBytes = []byte(cleanOriginID)
+		}
+	}
+	ts := pkt.Timestamp
+	if len(ts) > 0 && ts[len(ts)-1] != 'Z' && !strings.ContainsAny(ts[len(ts)-6:], "+-") {
+		ts = ts + "Z"
+	}
+	meshTimestamp, err = time.Parse(time.RFC3339Nano, ts)
+	if err != nil {
+		return "", nil, time.Time{}, nil, err
+	}
+	return cleanOrigin, originPubkeyBytes, meshTimestamp, packet, nil
+}
+
+func parseMeshCorePacketsMessage(payload []byte) (origin string, originPubkey []byte, meshTimestamp time.Time, packet []byte, err error) {
+	type PacketMessage struct {
+		Origin     string `json:"origin"`
+		OriginID   string `json:"origin_id"`
+		Timestamp  string `json:"timestamp"`
+		Type       string `json:"type"`
+		Direction  string `json:"direction"`
+		Time       string `json:"time"`
+		Date       string `json:"date"`
+		Len        string `json:"len"`
+		PacketType string `json:"packet_type"`
+		Route      string `json:"route"`
+		PayloadLen string `json:"payload_len"`
+		Raw        string `json:"raw"`
+		SNR        string `json:"SNR"`
+		RSSI       string `json:"RSSI"`
+		Score      string `json:"score"`
+		Duration   string `json:"duration"`
+		Hash       string `json:"hash"`
+	}
+	var pkt PacketMessage
+	if err = json.Unmarshal(payload, &pkt); err != nil {
+		return "", nil, time.Time{}, nil, err
+	}
+
+	// Decode the raw hex packet data
+	packet, err = hex.DecodeString(pkt.Raw)
+	if err != nil {
+		return "", nil, time.Time{}, nil, err
+	}
+
+	// Clean origin and origin_id by removing carriage returns and newlines
+	cleanOrigin := strings.ReplaceAll(strings.ReplaceAll(pkt.Origin, "\r", ""), "\n", "")
+	cleanOriginID := strings.ReplaceAll(strings.ReplaceAll(pkt.OriginID, "\r", ""), "\n", "")
+
+	// Decode origin_id hex to binary bytes for compact storage
+	var originPubkeyBytes []byte
+	if cleanOriginID != "" {
+		if decoded, decErr := hex.DecodeString(cleanOriginID); decErr == nil {
+			originPubkeyBytes = decoded
+		} else {
+			originPubkeyBytes = []byte(cleanOriginID)
+		}
+	}
+
+	// Parse timestamp
+	ts := pkt.Timestamp
+	if len(ts) > 0 && ts[len(ts)-1] != 'Z' && !strings.ContainsAny(ts[len(ts)-6:], "+-") {
+		ts = ts + "Z"
+	}
+	meshTimestamp, err = time.Parse(time.RFC3339Nano, ts)
+	if err != nil {
+		return "", nil, time.Time{}, nil, err
+	}
+
+	return cleanOrigin, originPubkeyBytes, meshTimestamp, packet, nil
+}
+
+// extractBaseTopic extracts the base topic from a full MQTT topic path
+// For topics like:
+// - meshcore/raw -> returns "meshcore"
+// - meshcore/salish/raw -> returns "meshcore/salish"
+// - meshcore/salish/a/raw -> returns "meshcore/salish/a"
+// - meshcore/salish/a/b/raw -> returns "meshcore/salish/a/b"
+// - meshcore/binary/[gatewayId] -> returns "meshcore"
+// - meshcore/salish/binary/[gatewayId] -> returns "meshcore/salish"
+// - meshcore/packets -> returns "meshcore"
+// - meshcore/salish/packets -> returns "meshcore/salish"
+// - meshcore/SEA/[gatewayId]/packets -> returns "meshcore/SEA"
+func extractBaseTopic(topic string) string {
+	topicParts := strings.Split(topic, "/")
+
+	if len(topicParts) < 2 {
+		return topic
+	}
+
+	// Check if it's a binary topic (last 2 parts are "binary" and gateway ID)
+	if len(topicParts) >= 3 && topicParts[len(topicParts)-2] == "binary" {
+		// Remove the last 2 parts: "binary" and "[gatewayId]"
+		return strings.Join(topicParts[:len(topicParts)-2], "/")
+	}
+
+	// Check if it's a raw topic (last part is "raw")
+	if topicParts[len(topicParts)-1] == "raw" {
+		// Remove the last 1 part: "raw"
+		return strings.Join(topicParts[:len(topicParts)-1], "/")
+	}
+
+	// Check if it's a packets topic with gateway ID (last part is "packets" and second-to-last looks like a gateway ID)
+	if len(topicParts) >= 4 && topicParts[len(topicParts)-1] == "packets" {
+		gatewayID := topicParts[len(topicParts)-2]
+		// If the second-to-last part is a long hex string (likely a gateway ID), remove both parts
+		if len(gatewayID) >= 40 && isHexString(gatewayID) {
+			// Remove the last 2 parts: "[gatewayId]" and "packets"
+			return strings.Join(topicParts[:len(topicParts)-2], "/")
+		}
+	}
+
+	// Check if it's a packets topic (last part is "packets")
+	if topicParts[len(topicParts)-1] == "packets" {
+		// Remove the last 1 part: "packets"
+		return strings.Join(topicParts[:len(topicParts)-1], "/")
+	}
+
+	// Fallback: if no "raw", "binary", or "packets" pattern found, return the first part
+	return topicParts[0]
+}
+
+// isHexString checks if a string contains only hexadecimal characters
+func isHexString(s string) bool {
+	for _, c := range s {
+		if !((c >= '0' && c <= '9') || (c >= 'a' && c <= 'f') || (c >= 'A' && c <= 'F')) {
+			return false
+		}
+	}
+	return true
+}
+
+func handleMeshCoreMessage(client mqtt.Client, msg mqtt.Message, d *ingestcommon.Daemon) {
+	broker := ""
+	clientOptions := client.OptionsReader()
+	if servers := clientOptions.Servers(); len(servers) > 0 {
+		broker = servers[0].String()
+	}
+
+	if broker == "tcp://mqtt-internal:1883" {
+		broker = "tcp://mqtt.w0z.is:1883"
+	}
+
+	zap.L().Debug("Received MeshCore packet", zap.String("topic", msg.Topic()), zap.String("broker", broker))
+
+	// Split topic once for efficient processing
+	topicParts := strings.Split(msg.Topic(), "/")
+
+	// Extract base topic (e.g., meshcore, meshcore/salish, meshcore/salish/a, etc.)
+	// This ensures we store only the base topic in the database, not the full topic path
+	baseTopic := extractBaseTopic(msg.Topic())
+
+	startTime := time.Now()
+
+	// Handle meshcore/raw and meshcore/*/raw topics (including multi-level paths like meshcore/salish/a/raw)
+	if len(topicParts) >= 2 && topicParts[len(topicParts)-1] == "raw" {
+		origin, originPubkey, meshTimestamp, decoded, err := parseMeshCoreRawMessage(msg.Payload())
+		if err != nil {
+			zap.L().Warn("Failed to parse meshcore/*/raw message", zap.Error(err))
+			return
+		}
+		query := `INSERT INTO meshcore_packets (ingest_timestamp, origin, origin_pubkey, broker, topic, mesh_timestamp, packet) VALUES (?, ?, ?, ?, ?, ?, ?)`
+		ingestTime := time.Now()
+		err = d.CHConn.Exec(d.Ctx, query, proto.ToDateTime64(ingestTime, proto.PrecisionMilli), origin, string(originPubkey), broker, baseTopic, meshTimestamp, string(decoded))
+		if err != nil {
+			zap.L().Warn("Failed to insert MeshCore packet into ClickHouse", zap.Error(err))
+			return
+		}
+		zap.L().Info("Successfully ingested MeshCore RAW packet", zap.String("broker", broker), zap.String("topic", baseTopic), zap.Duration("duration", time.Since(startTime)))
+		return
+	}
+
+	// Handle meshcore/*/[gatewayId]/packets topics (e.g., meshcore/SEA/[gatewayId]/packets)
+	if len(topicParts) >= 4 && topicParts[len(topicParts)-1] == "packets" {
+		gatewayID := topicParts[len(topicParts)-2]
+		// Check if the second-to-last part is a gateway ID (long hex string)
+		if len(gatewayID) >= 40 && isHexString(gatewayID) {
+			origin, originPubkey, meshTimestamp, decoded, err := parseMeshCorePacketsMessage(msg.Payload())
+			if err != nil {
+				zap.L().Warn("Failed to parse meshcore/*/[gatewayId]/packets message", zap.Error(err))
+				return
+			}
+			query := `INSERT INTO meshcore_packets (ingest_timestamp, origin, origin_pubkey, broker, topic, mesh_timestamp, packet) VALUES (?, ?, ?, ?, ?, ?, ?)`
+			ingestTime := time.Now()
+			err = d.CHConn.Exec(d.Ctx, query, proto.ToDateTime64(ingestTime, proto.PrecisionMilli), origin, string(originPubkey), broker, baseTopic, meshTimestamp, string(decoded))
+			if err != nil {
+				zap.L().Warn("Failed to insert MeshCore packet into ClickHouse", zap.Error(err))
+				return
+			}
+			zap.L().Info("Successfully ingested MeshCore PACKETS packet (with gateway ID)", zap.String("broker", broker), zap.String("topic", baseTopic), zap.String("gatewayID", gatewayID), zap.Duration("duration", time.Since(startTime)))
+			return
+		}
+	}
+
+	// Handle meshcore/packets and meshcore/*/packets topics (including multi-level paths)
+	if len(topicParts) >= 2 && topicParts[len(topicParts)-1] == "packets" {
+		origin, originPubkey, meshTimestamp, decoded, err := parseMeshCorePacketsMessage(msg.Payload())
+		if err != nil {
+			zap.L().Warn("Failed to parse meshcore/*/packets message", zap.Error(err))
+			return
+		}
+		query := `INSERT INTO meshcore_packets (ingest_timestamp, origin, origin_pubkey, broker, topic, mesh_timestamp, packet) VALUES (?, ?, ?, ?, ?, ?, ?)`
+		ingestTime := time.Now()
+		err = d.CHConn.Exec(d.Ctx, query, proto.ToDateTime64(ingestTime, proto.PrecisionMilli), origin, string(originPubkey), broker, baseTopic, meshTimestamp, string(decoded))
+		if err != nil {
+			zap.L().Warn("Failed to insert MeshCore packet into ClickHouse", zap.Error(err))
+			return
+		}
+		zap.L().Info("Successfully ingested MeshCore PACKETS packet", zap.String("broker", broker), zap.String("topic", baseTopic), zap.Duration("duration", time.Since(startTime)))
+		return
+	}
+
+	// Handle meshcore/*/binary/[gatewayid] topics (including multi-level paths)
+	if len(topicParts) >= 3 && topicParts[len(topicParts)-2] == "binary" {
+		// Extract gateway ID from the last part of the topic
+		gatewayID := topicParts[len(topicParts)-1]
+		// Decode gatewayID hex to binary bytes for compact storage; fall back to raw bytes on error
+		originPubkeyBytes, decErr := hex.DecodeString(gatewayID)
+		if decErr != nil {
+			originPubkeyBytes = []byte{}
+		}
+
+		query := `INSERT INTO meshcore_packets (ingest_timestamp, origin, origin_pubkey, broker, topic, mesh_timestamp, packet) VALUES (?, ?, ?, ?, ?, ?, ?)`
+		ingestTime := time.Now()
+		err := d.CHConn.Exec(d.Ctx, query, proto.ToDateTime64(ingestTime, proto.PrecisionMilli), gatewayID, string(originPubkeyBytes), broker, baseTopic, time.Time{}, string(msg.Payload()))
+		if err != nil {
+			zap.L().Warn("Failed to insert MeshCore binary packet into ClickHouse", zap.Error(err))
+			return
+		}
+		zap.L().Info("Successfully ingested MeshCore BINARY packet", zap.String("broker", broker), zap.String("topic", baseTopic), zap.String("gatewayID", gatewayID), zap.Duration("duration", time.Since(startTime)))
+		return
+	}
+}
+
+func loadConfig() (*ingestcommon.Config, error) {
+	brokers, err := ingestcommon.LoadMQTTBrokersFromEnv()
+	if err != nil {
+		return nil, err
+	}
+	config := &ingestcommon.Config{
+		MQTTBrokers:  brokers,
+		MQTTClientID: ingestcommon.GetEnvOrDefault("MQTT_CLIENT_ID", "meshcore-ingest"),
+	}
+	config.ClickHouse.Host = ingestcommon.GetEnvOrDefault("CLICKHOUSE_HOST", "127.0.0.1")
+	config.ClickHouse.Port = ingestcommon.GetEnvIntOrDefault("CLICKHOUSE_PORT", 9000)
+	config.ClickHouse.Database = ingestcommon.GetEnvOrDefault("CLICKHOUSE_DB", "default")
+	config.ClickHouse.Username = ingestcommon.GetEnvOrDefault("CLICKHOUSE_USER", "default")
+	config.ClickHouse.Password = ingestcommon.GetEnvOrDefault("CLICKHOUSE_PASSWORD", "")
+	return config, nil
+}
+
+func main() {
+	zapConfig := zap.NewProductionConfig()
+	zapConfig.Level = zap.NewAtomicLevelAt(zap.InfoLevel)
+	zapConfig.EncoderConfig.EncodeTime = zapcore.ISO8601TimeEncoder
+	logger, err := zapConfig.Build()
+	if err != nil {
+		log.Fatalf("Failed to initialize logger: %v", err)
+	}
+	defer logger.Sync()
+	zap.ReplaceGlobals(logger)
+	config, err := loadConfig()
+	if err != nil {
+		log.Fatalf("Failed to load configuration: %v", err)
+	}
+	daemon := ingestcommon.NewDaemon(config, handleMeshCoreMessage)
+	if err := daemon.ConnectClickHouse(); err != nil {
+		log.Fatalf("Failed to connect to ClickHouse: %v", err)
+	}
+	defer daemon.CHConn.Close()
+	if err := daemon.ConnectMQTT(); err != nil {
+		log.Fatalf("Failed to connect to MQTT brokers: %v", err)
+	}
+	defer func() {
+		for _, client := range daemon.MQTTClients {
+			if client != nil {
+				client.Disconnect(250)
+			}
+		}
+	}()
+	go daemon.MonitorConnections()
+	sigChan := make(chan os.Signal, 1)
+	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
+	zap.L().Info("MeshCore ingest daemon started. Press Ctrl+C to stop.")
+	<-sigChan
+	zap.L().Info("Shutting down daemon...")
+	daemon.Cancel()
+}

ingest/cmd/meshcoreingest/main_test.go

@@ -0,0 +1,267 @@
+package main
+
+import (
+	"encoding/hex"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestParseMeshCoreRawMessage_Valid(t *testing.T) {
+	payload := []byte(`{
+		"origin": "WW7STR/PugetMesh Cougar^",
+		"origin_id": "CB1F3E60913AC96ECCD9AE1053BB8646C5AA0A8EDBAC6D17822D06B0E054F6A5",
+		"timestamp": "2025-07-02T00:13:53.335723",
+		"type": "RAW",
+		"data": "050258C454C59A5A6A081C07D97E4A4C22FE575584F2BCE8267D76E0AAAC55283350EDCE7AF4ECBDCF12C599017ACC03D659F6C2A2AEF684B66774501A77BD4F361C6E5436BB6625"
+	}`)
+	origin, originPubkey, meshTimestamp, packet, err := parseMeshCoreRawMessage(payload)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if origin != "WW7STR/PugetMesh Cougar^" {
+		t.Errorf("unexpected origin: %s", origin)
+	}
+	if hex.EncodeToString(originPubkey) != strings.ToLower("CB1F3E60913AC96ECCD9AE1053BB8646C5AA0A8EDBAC6D17822D06B0E054F6A5") {
+		t.Errorf("unexpected origin_pubkey: %s", hex.EncodeToString(originPubkey))
+	}
+	if meshTimestamp.Format(time.RFC3339Nano) != "2025-07-02T00:13:53.335723Z" {
+		t.Errorf("unexpected meshTimestamp: %s", meshTimestamp)
+	}
+	if len(packet) == 0 {
+		t.Error("packet should not be empty")
+	}
+}
+
+func TestParseMeshCoreRawMessage_InvalidHex(t *testing.T) {
+	payload := []byte(`{
+		"origin": "WW7STR/PugetMesh Cougar^",
+		"origin_id": "CB1F3E60913AC96ECCD9AE1053BB8646C5AA0A8EDBAC6D17822D06B0E054F6A5",
+		"timestamp": "2025-07-02T00:13:53.335723",
+		"type": "RAW",
+		"data": "nothex"
+	}`)
+	_, _, _, _, err := parseMeshCoreRawMessage(payload)
+	if err == nil {
+		t.Error("expected error for invalid hex")
+	}
+}
+
+func TestParseMeshCoreRawMessage_InvalidTimestamp(t *testing.T) {
+	payload := []byte(`{
+		"origin": "WW7STR/PugetMesh Cougar^",
+		"origin_id": "CB1F3E60913AC96ECCD9AE1053BB8646C5AA0A8EDBAC6D17822D06B0E054F6A5",
+		"timestamp": "notatime",
+		"type": "RAW",
+		"data": "050258C454C59A5A6A081C07D97E4A4C22FE575584F2BCE8267D76E0AAAC55283350EDCE7AF4ECBDCF12C599017ACC03D659F6C2A2AEF684B66774501A77BD4F361C6E5436BB6625"
+	}`)
+	_, _, _, _, err := parseMeshCoreRawMessage(payload)
+	if err == nil {
+		t.Error("expected error for invalid timestamp")
+	}
+}
+
+func TestParseMeshCoreRawMessage_InvalidJSON(t *testing.T) {
+	payload := []byte(`notjson`)
+	_, _, _, _, err := parseMeshCoreRawMessage(payload)
+	if err == nil {
+		t.Error("expected error for invalid JSON")
+	}
+}
+
+func TestParseMeshCoreRawMessage_WithOriginID(t *testing.T) {
+	payload := []byte(`{
+		"origin": "HAX!peater^",
+		"origin_id": "CB1F3E60913AC96ECCD9AE1053BB8646C5AA0A8EDBAC6D17822D06B0E054F6A5",
+		"timestamp": "2025-08-18T17:27:28.053612",
+		"type": "RAW",
+		"data": "0500857E3083FDA8C09D70B84A77460B03F3408A1C24B478BA397BB9563CDDB09FE48CB9F0B04BEE7976EAD62B894E5FE91D9F000FC4B6437F46AB94CC6FE0936A97675698D9"
+	}`)
+	origin, originPubkey, meshTimestamp, packet, err := parseMeshCoreRawMessage(payload)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if origin != "HAX!peater^" {
+		t.Errorf("unexpected origin: %s", origin)
+	}
+	if hex.EncodeToString(originPubkey) != strings.ToLower("CB1F3E60913AC96ECCD9AE1053BB8646C5AA0A8EDBAC6D17822D06B0E054F6A5") {
+		t.Errorf("unexpected origin_pubkey: %s", hex.EncodeToString(originPubkey))
+	}
+	if meshTimestamp.Format(time.RFC3339Nano) != "2025-08-18T17:27:28.053612Z" {
+		t.Errorf("unexpected meshTimestamp: %s", meshTimestamp)
+	}
+	if len(packet) == 0 {
+		t.Error("packet should not be empty")
+	}
+}
+
+func TestExtractBaseTopic(t *testing.T) {
+	tests := []struct {
+		name     string
+		topic    string
+		expected string
+	}{
+		// Raw topic cases
+		{
+			name:     "meshcore/raw",
+			topic:    "meshcore/raw",
+			expected: "meshcore",
+		},
+		{
+			name:     "meshcore/salish/raw",
+			topic:    "meshcore/salish/raw",
+			expected: "meshcore/salish",
+		},
+		{
+			name:     "meshcore/salish/a/raw",
+			topic:    "meshcore/salish/a/raw",
+			expected: "meshcore/salish/a",
+		},
+		{
+			name:     "meshcore/salish/a/b/raw",
+			topic:    "meshcore/salish/a/b/raw",
+			expected: "meshcore/salish/a/b",
+		},
+		// Binary topic cases
+		{
+			name:     "meshcore/binary/gateway123",
+			topic:    "meshcore/binary/gateway123",
+			expected: "meshcore",
+		},
+		{
+			name:     "meshcore/salish/binary/gateway456",
+			topic:    "meshcore/salish/binary/gateway456",
+			expected: "meshcore/salish",
+		},
+		{
+			name:     "meshcore/salish/a/binary/gateway789",
+			topic:    "meshcore/salish/a/binary/gateway789",
+			expected: "meshcore/salish/a",
+		},
+		// Edge cases
+		{
+			name:     "single part topic",
+			topic:    "meshcore",
+			expected: "meshcore",
+		},
+		{
+			name:     "empty topic",
+			topic:    "",
+			expected: "",
+		},
+		{
+			name:     "topic with only raw",
+			topic:    "raw",
+			expected: "raw",
+		},
+		{
+			name:     "topic with only binary",
+			topic:    "binary",
+			expected: "binary",
+		},
+		// Complex nested cases
+		{
+			name:     "deep nested raw",
+			topic:    "meshcore/region/state/city/neighborhood/raw",
+			expected: "meshcore/region/state/city/neighborhood",
+		},
+		{
+			name:     "deep nested binary",
+			topic:    "meshcore/region/state/city/neighborhood/binary/gateway999",
+			expected: "meshcore/region/state/city/neighborhood",
+		},
+		// Mixed cases that shouldn't match our patterns
+		{
+			name:     "topic ending with other than raw or binary",
+			topic:    "meshcore/salish/other",
+			expected: "meshcore",
+		},
+		{
+			name:     "topic with raw in middle",
+			topic:    "meshcore/raw/salish",
+			expected: "meshcore",
+		},
+		{
+			name:     "topic with binary in middle",
+			topic:    "meshcore/binary/salish",
+			expected: "meshcore",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := extractBaseTopic(tt.topic)
+			if result != tt.expected {
+				t.Errorf("extractBaseTopic(%q) = %q, want %q", tt.topic, result, tt.expected)
+			}
+		})
+	}
+}
+
+func TestExtractGatewayID(t *testing.T) {
+	tests := []struct {
+		name       string
+		topic      string
+		expected   string
+		shouldHave bool
+	}{
+		{
+			name:       "meshcore/binary/gateway123",
+			topic:      "meshcore/binary/gateway123",
+			expected:   "gateway123",
+			shouldHave: true,
+		},
+		{
+			name:       "meshcore/salish/binary/gateway456",
+			topic:      "meshcore/salish/binary/gateway456",
+			expected:   "gateway456",
+			shouldHave: true,
+		},
+		{
+			name:       "meshcore/salish/a/binary/gateway789",
+			topic:      "meshcore/salish/a/binary/gateway789",
+			expected:   "gateway789",
+			shouldHave: true,
+		},
+		{
+			name:       "deep nested binary",
+			topic:      "meshcore/region/state/city/neighborhood/binary/gateway999",
+			expected:   "gateway999",
+			shouldHave: true,
+		},
+		{
+			name:       "not a binary topic",
+			topic:      "meshcore/raw",
+			expected:   "",
+			shouldHave: false,
+		},
+		{
+			name:       "not a binary topic 2",
+			topic:      "meshcore/salish/other",
+			expected:   "",
+			shouldHave: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			topicParts := strings.Split(tt.topic, "/")
+			var gatewayID string
+
+			// Match the exact logic from the main function
+			if len(topicParts) >= 3 && topicParts[len(topicParts)-2] == "binary" {
+				gatewayID = topicParts[len(topicParts)-1]
+			}
+
+			if tt.shouldHave {
+				if gatewayID != tt.expected {
+					t.Errorf("extractGatewayID(%q) = %q, want %q", tt.topic, gatewayID, tt.expected)
+				}
+			} else {
+				if gatewayID != "" {
+					t.Errorf("extractGatewayID(%q) = %q, but should not have gateway ID", tt.topic, gatewayID)
+				}
+			}
+		})
+	}
+}

ingest/go.mod

@@ -0,0 +1,38 @@
+module github.com/ajvpot/clickhouse-meshingest
+
+go 1.24
+
+toolchain go1.24.4
+
+require (
+	github.com/ClickHouse/ch-go v0.66.1
+	github.com/ClickHouse/clickhouse-go/v2 v2.37.2
+	github.com/eclipse/paho.mqtt.golang v1.5.0
+	github.com/pressly/goose/v3 v3.24.3
+	go.uber.org/zap v1.27.0
+)
+
+require (
+	github.com/andybalholm/brotli v1.1.1 // indirect
+	github.com/go-faster/city v1.0.1 // indirect
+	github.com/go-faster/errors v0.7.1 // indirect
+	github.com/google/uuid v1.6.0 // indirect
+	github.com/gorilla/websocket v1.5.3 // indirect
+	github.com/klauspost/compress v1.18.0 // indirect
+	github.com/kr/pretty v0.3.1 // indirect
+	github.com/mfridman/interpolate v0.0.2 // indirect
+	github.com/paulmach/orb v0.11.1 // indirect
+	github.com/pierrec/lz4/v4 v4.1.22 // indirect
+	github.com/segmentio/asm v1.2.0 // indirect
+	github.com/sethvargo/go-retry v0.3.0 // indirect
+	github.com/shopspring/decimal v1.4.0 // indirect
+	go.opentelemetry.io/otel v1.36.0 // indirect
+	go.opentelemetry.io/otel/trace v1.36.0 // indirect
+	go.uber.org/multierr v1.11.0 // indirect
+	golang.org/x/exp v0.0.0-20250606033433-dcc06ee1d476 // indirect
+	golang.org/x/net v0.41.0 // indirect
+	golang.org/x/sync v0.15.0 // indirect
+	golang.org/x/sys v0.33.0 // indirect
+	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+)

ingest/go.sum

@@ -0,0 +1,154 @@
+github.com/ClickHouse/ch-go v0.66.1 h1:LQHFslfVYZsISOY0dnOYOXGkOUvpv376CCm8g7W74A4=
+github.com/ClickHouse/ch-go v0.66.1/go.mod h1:NEYcg3aOFv2EmTJfo4m2WF7sHB/YFbLUuIWv9iq76xY=
+github.com/ClickHouse/clickhouse-go/v2 v2.37.2 h1:wRLNKoynvHQEN4znnVHNLaYnrqVc9sGJmGYg+GGCfto=
+github.com/ClickHouse/clickhouse-go/v2 v2.37.2/go.mod h1:pH2zrBGp5Y438DMwAxXMm1neSXPPjSI7tD4MURVULw8=
+github.com/andybalholm/brotli v1.1.1 h1:PR2pgnyFznKEugtsUo0xLdDop5SKXd5Qf5ysW+7XdTA=
+github.com/andybalholm/brotli v1.1.1/go.mod h1:05ib4cKhjx3OQYUY22hTVd34Bc8upXjOLL2rKwwZBoA=
+github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
+github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
+github.com/eclipse/paho.mqtt.golang v1.5.0 h1:EH+bUVJNgttidWFkLLVKaQPGmkTUfQQqjOsyvMGvD6o=
+github.com/eclipse/paho.mqtt.golang v1.5.0/go.mod h1:du/2qNQVqJf/Sqs4MEL77kR8QTqANF7XU7Fk0aOTAgk=
+github.com/go-faster/city v1.0.1 h1:4WAxSZ3V2Ws4QRDrscLEDcibJY8uf41H6AhXDrNDcGw=
+github.com/go-faster/city v1.0.1/go.mod h1:jKcUJId49qdW3L1qKHH/3wPeUstCVpVSXTM6vO3VcTw=
+github.com/go-faster/errors v0.7.1 h1:MkJTnDoEdi9pDabt1dpWf7AA8/BaSYZqibYyhZ20AYg=
+github.com/go-faster/errors v0.7.1/go.mod h1:5ySTjWFiphBs07IKuiL69nxdfd5+fzh1u7FPGZP2quo=
+github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
+github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
+github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
+github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
+github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
+github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
+github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
+github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
+github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
+github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
+github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
+github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
+github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
+github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
+github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
+github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
+github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
+github.com/mfridman/interpolate v0.0.2 h1:pnuTK7MQIxxFz1Gr+rjSIx9u7qVjf5VOoM/u6BbAxPY=
+github.com/mfridman/interpolate v0.0.2/go.mod h1:p+7uk6oE07mpE/Ik1b8EckO0O4ZXiGAfshKBWLUM9Xg=
+github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe/go.mod h1:wL8QJuTMNUDYhXwkmfOly8iTdp5TEcJFWZD2D7SIkUc=
+github.com/ncruces/go-strftime v0.1.9 h1:bY0MQC28UADQmHmaF5dgpLmImcShSi2kHU9XLdhx/f4=
+github.com/ncruces/go-strftime v0.1.9/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls=
+github.com/paulmach/orb v0.11.1 h1:3koVegMC4X/WeiXYz9iswopaTwMem53NzTJuTF20JzU=
+github.com/paulmach/orb v0.11.1/go.mod h1:5mULz1xQfs3bmQm63QEJA6lNGujuRafwA5S/EnuLaLU=
+github.com/paulmach/protoscan v0.2.1/go.mod h1:SpcSwydNLrxUGSDvXvO0P7g7AuhJ7lcKfDlhJCDw2gY=
+github.com/pierrec/lz4/v4 v4.1.22 h1:cKFw6uJDK+/gfw5BcDL0JL5aBsAFdsIT18eRtLj7VIU=
+github.com/pierrec/lz4/v4 v4.1.22/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
+github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA=
+github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/pressly/goose/v3 v3.24.3 h1:DSWWNwwggVUsYZ0X2VitiAa9sKuqtBfe+Jr9zFGwWlM=
+github.com/pressly/goose/v3 v3.24.3/go.mod h1:v9zYL4xdViLHCUUJh/mhjnm6JrK7Eul8AS93IxiZM4E=
+github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
+github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
+github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=
+github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
+github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
+github.com/segmentio/asm v1.2.0 h1:9BQrFxC+YOHJlTlHGkTrFWf59nbL3XnCoFLTwDCI7ys=
+github.com/segmentio/asm v1.2.0/go.mod h1:BqMnlJP91P8d+4ibuonYZw9mfnzI9HfxselHZr5aAcs=
+github.com/sethvargo/go-retry v0.3.0 h1:EEt31A35QhrcRZtrYFDTBg91cqZVnFL2navjDrah2SE=
+github.com/sethvargo/go-retry v0.3.0/go.mod h1:mNX17F0C/HguQMyMyJxcnU471gOZGxCLyYaFyAZraas=
+github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp81k=
+github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
+github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
+github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
+github.com/xdg-go/scram v1.1.1/go.mod h1:RaEWvsqvNKKvBPvcKeFjrG2cJqOkHTiyTpzz23ni57g=
+github.com/xdg-go/stringprep v1.0.3/go.mod h1:W3f5j4i+9rC0kuIEJL0ky1VpHXQU3ocBgklLGvcBnW8=
+github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU=
+github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E=
+github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d/go.mod h1:rHwXgn7JulP+udvsHwJoVG1YGAP6VLg4y9I5dyZdqmA=
+github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
+github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
+go.mongodb.org/mongo-driver v1.11.4/go.mod h1:PTSz5yu21bkT/wXpkS7WR5f0ddqw5quethTUn9WM+2g=
+go.opentelemetry.io/otel v1.36.0 h1:UumtzIklRBY6cI/lllNZlALOF5nNIzJVb16APdvgTXg=
+go.opentelemetry.io/otel v1.36.0/go.mod h1:/TcFMXYjyRNh8khOAO9ybYkqaDBb/70aVwkNML4pP8E=
+go.opentelemetry.io/otel/trace v1.36.0 h1:ahxWNuqZjpdiFAyrIoQ4GIiAIhxAunQR6MUoKrsNd4w=
+go.opentelemetry.io/otel/trace v1.36.0/go.mod h1:gQ+OnDZzrybY4k4seLzPAWNwVBBVlF2szhehOBB/tGA=
+go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
+go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
+go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
+go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
+go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8=
+go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
+golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
+golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
+golang.org/x/exp v0.0.0-20250606033433-dcc06ee1d476 h1:bsqhLWFR6G6xiQcb+JoGqdKdRU6WzPWmK8E0jxTjzo4=
+golang.org/x/exp v0.0.0-20250606033433-dcc06ee1d476/go.mod h1:3//PLf8L/X+8b4vuAfHzxeRUl04Adcb341+IGKfnqS8=
+golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
+golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
+golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
+golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
+golang.org/x/net v0.41.0 h1:vBTly1HeNPEn3wtREYfy4GZ/NECgw2Cnl+nK6Nz3uvw=
+golang.org/x/net v0.41.0/go.mod h1:B/K4NNqkfmg07DQYrbwvSluqCJOOXwUjeb/5lOisjbA=
+golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.15.0 h1:KWH3jNZsfyT6xfAfKiz6MRNmd46ByHDYaZ7KSkCtdW8=
+golang.org/x/sync v0.15.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.33.0 h1:q3i8TbbEz+JRD9ywIRlyRAQbM0qF7hu24q3teo2hbuw=
+golang.org/x/sys v0.33.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
+golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
+golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
+golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
+google.golang.org/protobuf v1.27.1/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+modernc.org/libc v1.65.0 h1:e183gLDnAp9VJh6gWKdTy0CThL9Pt7MfcR/0bgb7Y1Y=
+modernc.org/libc v1.65.0/go.mod h1:7m9VzGq7APssBTydds2zBcxGREwvIGpuUBaKTXdm2Qs=
+modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU=
+modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg=
+modernc.org/memory v1.10.0 h1:fzumd51yQ1DxcOxSO+S6X7+QTuVU+n8/Aj7swYjFfC4=
+modernc.org/memory v1.10.0/go.mod h1:/JP4VbVC+K5sU2wZi9bHoq2MAkCnrt2r98UGeSK7Mjw=
+modernc.org/sqlite v1.37.0 h1:s1TMe7T3Q3ovQiK2Ouz4Jwh7dw4ZDqbebSDTlSJdfjI=
+modernc.org/sqlite v1.37.0/go.mod h1:5YiWv+YviqGMuGw4V+PNplcyaJ5v+vQd7TQOgkACoJM=

ingest/internal/ingestcommon/ingest.go

@@ -0,0 +1,307 @@
+package ingestcommon
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"math"
+	"os"
+	"strconv"
+	"time"
+
+	"github.com/ClickHouse/clickhouse-go/v2"
+	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
+	mqtt "github.com/eclipse/paho.mqtt.golang"
+	"go.uber.org/zap"
+)
+
+type MQTTBrokerConfig struct {
+	URL      string   `json:"url"`
+	Username string   `json:"username"`
+	Password string   `json:"password"`
+	Topics   []string `json:"topics"`
+}
+
+// LoadMQTTBrokersFromEnv parses the MQTT_BROKERS environment variable, which must
+// contain a JSON array of broker configs, e.g.:
+//
+//	[{"url":"tcp://host:1883","username":"u","password":"p","topics":["meshcore/#"]}]
+//
+// It returns an error when the variable is unset/empty or cannot be parsed so the
+// daemon fails fast with a clear message rather than silently connecting to nothing.
+// Brokers without an explicit topic list default to subscribing to "meshcore/#".
+func LoadMQTTBrokersFromEnv() ([]MQTTBrokerConfig, error) {
+	raw := os.Getenv("MQTT_BROKERS")
+	if raw == "" {
+		return nil, fmt.Errorf("MQTT_BROKERS is not set; provide a JSON array of broker configs (see .env.example)")
+	}
+	var brokers []MQTTBrokerConfig
+	if err := json.Unmarshal([]byte(raw), &brokers); err != nil {
+		return nil, fmt.Errorf("failed to parse MQTT_BROKERS as JSON: %w", err)
+	}
+	if len(brokers) == 0 {
+		return nil, fmt.Errorf("MQTT_BROKERS contained no brokers")
+	}
+	for i := range brokers {
+		if brokers[i].URL == "" {
+			return nil, fmt.Errorf("MQTT_BROKERS[%d] is missing a url", i)
+		}
+		if len(brokers[i].Topics) == 0 {
+			brokers[i].Topics = []string{"meshcore/#"}
+		}
+	}
+	return brokers, nil
+}
+
+type Config struct {
+	MQTTBrokers  []MQTTBrokerConfig `json:"mqtt_brokers"`
+	MQTTClientID string             `json:"mqtt_client_id"`
+	ClickHouse   struct {
+		Host     string `json:"host"`
+		Port     int    `json:"port"`
+		Database string `json:"database"`
+		Username string `json:"username"`
+		Password string `json:"password"`
+	} `json:"clickhouse"`
+}
+
+type MessageHandler func(client mqtt.Client, msg mqtt.Message, d *Daemon)
+
+type Daemon struct {
+	Config         *Config
+	MQTTClients    []mqtt.Client
+	BrokerStatus   map[string]bool
+	CHConn         driver.Conn
+	Ctx            context.Context
+	Cancel         context.CancelFunc
+	MessageHandler MessageHandler
+}
+
+func NewDaemon(config *Config, handler MessageHandler) *Daemon {
+	ctx, cancel := context.WithCancel(context.Background())
+	return &Daemon{
+		Config:         config,
+		BrokerStatus:   make(map[string]bool),
+		Ctx:            ctx,
+		Cancel:         cancel,
+		MessageHandler: handler,
+	}
+}
+
+func (d *Daemon) connectToBroker(broker MQTTBrokerConfig, maxRetries int) (mqtt.Client, error) {
+	baseDelay := time.Second
+	maxDelay := 30 * time.Second
+
+	for attempt := 0; attempt <= maxRetries; attempt++ {
+		opts := mqtt.NewClientOptions()
+		opts.AddBroker(broker.URL)
+		opts.SetClientID(fmt.Sprintf("%s-%d", d.Config.MQTTClientID, attempt))
+		opts.SetDefaultPublishHandler(func(client mqtt.Client, msg mqtt.Message) {
+			d.MessageHandler(client, msg, d)
+		})
+		opts.SetAutoReconnect(true)
+		opts.SetConnectRetry(true)
+		opts.SetConnectTimeout(10 * time.Second)
+		opts.SetMaxReconnectInterval(30 * time.Second)
+		opts.SetKeepAlive(30 * time.Second)
+		opts.SetPingTimeout(10 * time.Second)
+		opts.SetCleanSession(false)
+		opts.SetResumeSubs(true)
+
+		opts.SetOnConnectHandler(func(client mqtt.Client) {
+			zap.L().Debug("Connected to MQTT broker", zap.String("broker", broker.URL))
+		})
+		opts.SetConnectionLostHandler(func(client mqtt.Client, err error) {
+			zap.L().Warn("Connection lost to MQTT broker", zap.String("broker", broker.URL), zap.Error(err))
+			d.BrokerStatus[broker.URL] = false
+		})
+
+		if broker.Username != "" {
+			opts.SetUsername(broker.Username)
+			if broker.Password != "" {
+				opts.SetPassword(broker.Password)
+			}
+		}
+
+		client := mqtt.NewClient(opts)
+		if token := client.Connect(); token.Wait() && token.Error() != nil {
+			zap.L().Warn("Failed to connect to MQTT broker",
+				zap.String("broker", broker.URL),
+				zap.Int("attempt", attempt+1),
+				zap.Int("maxRetries", maxRetries+1),
+				zap.Error(token.Error()))
+
+			if attempt < maxRetries {
+				delay := time.Duration(float64(baseDelay) * math.Pow(2, float64(attempt)))
+				if delay > maxDelay {
+					delay = maxDelay
+				}
+				zap.L().Debug("Retrying connection", zap.String("broker", broker.URL), zap.Duration("delay", delay))
+				time.Sleep(delay)
+				continue
+			}
+			return nil, fmt.Errorf("failed to connect to MQTT broker %s after %d attempts: %w", broker.URL, maxRetries+1, token.Error())
+		}
+
+		for _, topic := range broker.Topics {
+			if token := client.Subscribe(topic, 0, nil); token.Wait() && token.Error() != nil {
+				zap.L().Warn("Failed to subscribe to topic",
+					zap.String("topic", topic),
+					zap.String("broker", broker.URL),
+					zap.Int("attempt", attempt+1),
+					zap.Int("maxRetries", maxRetries+1),
+					zap.Error(token.Error()))
+
+				if attempt < maxRetries {
+					client.Disconnect(250)
+					delay := time.Duration(float64(baseDelay) * math.Pow(2, float64(attempt)))
+					if delay > maxDelay {
+						delay = maxDelay
+					}
+					zap.L().Debug("Retrying subscription", zap.String("broker", broker.URL), zap.Duration("delay", delay))
+					time.Sleep(delay)
+					continue
+				}
+				client.Disconnect(250)
+				return nil, fmt.Errorf("failed to subscribe to topic %s on broker %s after %d attempts: %w", topic, broker.URL, maxRetries+1, token.Error())
+			}
+		}
+
+		zap.L().Debug("Successfully connected to MQTT broker and subscribed to topics",
+			zap.Strings("topics", broker.Topics),
+			zap.String("broker", broker.URL))
+		return client, nil
+	}
+	return nil, fmt.Errorf("unexpected error in connectToBroker for %s", broker.URL)
+}
+
+func (d *Daemon) ConnectMQTT() error {
+	maxRetries := 5
+	successfulConnections := 0
+	totalBrokers := len(d.Config.MQTTBrokers)
+
+	zap.L().Debug("Attempting to connect to MQTT brokers", zap.Int("totalBrokers", totalBrokers))
+
+	for _, broker := range d.Config.MQTTBrokers {
+		client, err := d.connectToBroker(broker, maxRetries)
+		if err != nil {
+			zap.L().Warn("Failed to connect to broker", zap.String("broker", broker.URL), zap.Error(err))
+			d.BrokerStatus[broker.URL] = false
+			continue
+		}
+		d.MQTTClients = append(d.MQTTClients, client)
+		d.BrokerStatus[broker.URL] = true
+		successfulConnections++
+	}
+
+	if successfulConnections == 0 {
+		return fmt.Errorf("failed to connect to any MQTT brokers")
+	}
+
+	zap.L().Debug("Successfully connected to MQTT brokers",
+		zap.Int("successfulConnections", successfulConnections),
+		zap.Int("totalBrokers", totalBrokers))
+	return nil
+}
+
+func (d *Daemon) ConnectClickHouse() error {
+	settings := clickhouse.Settings{
+		"max_execution_time":            60,
+		"async_insert":                  1,
+		"wait_for_async_insert":         1,
+		"async_insert_busy_timeout_ms":  2500,    // 2.5 seconds - flush if buffer is busy for this long
+		"async_insert_max_data_size":    1048576, // 1MB - flush when buffer reaches this size
+		"async_insert_max_query_number": 5000,    // flush after this many insert queries accumulate
+	}
+
+	conn, err := clickhouse.Open(&clickhouse.Options{
+		Addr: []string{fmt.Sprintf("%s:%d", d.Config.ClickHouse.Host, d.Config.ClickHouse.Port)},
+		Auth: clickhouse.Auth{
+			Database: d.Config.ClickHouse.Database,
+			Username: d.Config.ClickHouse.Username,
+			Password: d.Config.ClickHouse.Password,
+		},
+		Settings: settings,
+		Debug:    false,
+	})
+	if err != nil {
+		return fmt.Errorf("failed to connect to ClickHouse: %w", err)
+	}
+	d.CHConn = conn
+	zap.L().Info("Connected to ClickHouse",
+		zap.String("host", d.Config.ClickHouse.Host),
+		zap.Int("port", d.Config.ClickHouse.Port),
+		zap.Reflect("settings", settings))
+	return nil
+}
+
+func (d *Daemon) MonitorConnections() {
+	ticker := time.NewTicker(30 * time.Second)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-d.Ctx.Done():
+			return
+		case <-ticker.C:
+			d.checkAndReconnectBrokers()
+		}
+	}
+}
+
+func (d *Daemon) checkAndReconnectBrokers() {
+	for i, broker := range d.Config.MQTTBrokers {
+		if i >= len(d.MQTTClients) {
+			client, err := d.connectToBroker(broker, 3)
+			if err != nil {
+				zap.L().Warn("Background reconnection failed for broker", zap.String("broker", broker.URL), zap.Error(err))
+				continue
+			}
+			d.MQTTClients = append(d.MQTTClients, client)
+			d.BrokerStatus[broker.URL] = true
+			zap.L().Debug("Successfully reconnected to broker", zap.String("broker", broker.URL))
+			continue
+		}
+		client := d.MQTTClients[i]
+		if client == nil {
+			newClient, err := d.connectToBroker(broker, 3)
+			if err != nil {
+				zap.L().Warn("Background reconnection failed for broker", zap.String("broker", broker.URL), zap.Error(err))
+				d.BrokerStatus[broker.URL] = false
+				continue
+			}
+			d.MQTTClients[i] = newClient
+			d.BrokerStatus[broker.URL] = true
+			zap.L().Debug("Successfully reconnected to broker", zap.String("broker", broker.URL))
+			continue
+		}
+		if !client.IsConnected() {
+			d.BrokerStatus[broker.URL] = false
+			newClient, err := d.connectToBroker(broker, 3)
+			if err != nil {
+				zap.L().Warn("Background reconnection failed for broker", zap.String("broker", broker.URL), zap.Error(err))
+				continue
+			}
+			d.MQTTClients[i] = newClient
+			d.BrokerStatus[broker.URL] = true
+			zap.L().Debug("Successfully reconnected to broker", zap.String("broker", broker.URL))
+		} else {
+			d.BrokerStatus[broker.URL] = true
+		}
+	}
+}
+
+func GetEnvOrDefault(key, defaultValue string) string {
+	if value := os.Getenv(key); value != "" {
+		return value
+	}
+	return defaultValue
+}
+
+func GetEnvIntOrDefault(key string, defaultValue int) int {
+	if value := os.Getenv(key); value != "" {
+		if intValue, err := strconv.Atoi(value); err == nil {
+			return intValue
+		}
+	}
+	return defaultValue
+}

ingest/internal/migrate/main.go

@@ -0,0 +1,112 @@
+package main
+
+import (
+	"database/sql"
+	"flag"
+	"fmt"
+	"log"
+	"os"
+
+	"github.com/ClickHouse/clickhouse-go/v2"
+	"github.com/pressly/goose/v3"
+)
+
+func main() {
+	var (
+		host           = flag.String("host", "localhost", "ClickHouse host")
+		port           = flag.Int("port", 9000, "ClickHouse port")
+		database       = flag.String("database", "default", "ClickHouse database")
+		username       = flag.String("username", "default", "ClickHouse username")
+		password       = flag.String("password", "", "ClickHouse password")
+		migrationsPath = flag.String("path", "migrations", "Path to migration files")
+		action         = flag.String("action", "up", "Migration action: up, down, reset, status, version")
+		version        = flag.Int("version", 0, "Version for down action")
+	)
+	flag.Parse()
+
+	// Create ClickHouse connection
+	conn := clickhouse.Connector(&clickhouse.Options{
+		Addr: []string{fmt.Sprintf("%s:%d", *host, *port)},
+		Auth: clickhouse.Auth{
+			Database: *database,
+			Username: *username,
+			Password: *password,
+		},
+		Settings: clickhouse.Settings{
+			"max_execution_time": 60,
+		},
+		Debug: false,
+	})
+
+	// Open database connection
+	db := sql.OpenDB(conn)
+	defer db.Close()
+
+	// Test connection
+	if err := db.Ping(); err != nil {
+		log.Fatalf("Failed to connect to ClickHouse: %v", err)
+	}
+
+	// Set goose database
+	goose.SetBaseFS(os.DirFS(*migrationsPath))
+
+	// Execute migration action
+	switch *action {
+	case "up":
+		if err := goose.Up(db, "."); err != nil {
+			log.Fatalf("Failed to run migrations up: %v", err)
+		}
+		fmt.Println("Migrations applied successfully")
+	case "down":
+		if *version == 0 {
+			// Down one migration
+			if err := goose.Down(db, "."); err != nil {
+				log.Fatalf("Failed to run migrations down: %v", err)
+			}
+			fmt.Println("Migration rolled back successfully")
+		} else {
+			// Down to specific version
+			if err := goose.DownTo(db, ".", int64(*version)); err != nil {
+				log.Fatalf("Failed to run migrations down to version %d: %v", *version, err)
+			}
+			fmt.Printf("Migrations rolled back to version %d successfully\n", *version)
+		}
+	case "reset":
+		if err := goose.Reset(db, "."); err != nil {
+			log.Fatalf("Failed to reset migrations: %v", err)
+		}
+		fmt.Println("Migrations reset successfully")
+	case "status":
+		if err := showStatus(db); err != nil {
+			log.Fatalf("Failed to show status: %v", err)
+		}
+	case "version":
+		version, err := goose.GetDBVersion(db)
+		if err != nil {
+			log.Fatalf("Failed to get migration version: %v", err)
+		}
+		fmt.Printf("Current migration version: %d\n", version)
+	default:
+		fmt.Printf("Unknown action: %s\n", *action)
+		fmt.Println("Available actions: up, down, reset, status, version")
+		os.Exit(1)
+	}
+}
+
+func showStatus(db *sql.DB) error {
+	// Get current version
+	currentVersion, err := goose.GetDBVersion(db)
+	if err != nil {
+		return err
+	}
+
+	fmt.Println("Migration Status:")
+	fmt.Println("==================")
+	fmt.Printf("Current version: %d\n", currentVersion)
+
+	// Note: goose doesn't provide a direct way to list all available migrations
+	// from the source, so we can't show pending migrations without additional file system scanning
+	fmt.Println("\nNote: Use 'version' action to see current migration state")
+
+	return nil
+}

ingest/migrations/001_create_meshcore_packets.sql

@@ -0,0 +1,133 @@
+-- +goose Up
+CREATE TABLE IF NOT EXISTS meshcore_packets (
+    ingest_timestamp DateTime64(3) COMMENT 'Timestamp when the packet was ingested into the database',
+    -- todo change to low cardinality
+    origin String COMMENT 'Origin field from meshcore/raw',
+    origin_pubkey String COMMENT 'Public key of the origin node (raw packets) or gateway ID (binary packets)',
+    broker LowCardinality(String) COMMENT 'Message broker identifier',
+    topic LowCardinality(String) COMMENT 'Message topic',
+    mesh_timestamp DateTime64(6) COMMENT 'Timestamp from meshcore/raw',
+    packet String COMMENT 'Raw MeshCore packet as a string',
+    -- Extract the first byte as header
+    header UInt8 ALIAS reinterpretAsUInt8(substring(packet, 1, 1)) COMMENT 'Header byte: routing type, payload type, payload version',
+    -- Use bitwise operations on header as per MeshCore spec
+    route_type UInt8 ALIAS bitAnd(header, 0x03) COMMENT 'Route Type (bits 0-1 of header): 0=ROUTE_TYPE_TRANSPORT_FLOOD, 1=ROUTE_TYPE_FLOOD, 2=ROUTE_TYPE_DIRECT, 3=ROUTE_TYPE_TRANSPORT_DIRECT',
+    payload_type UInt8 ALIAS bitAnd(bitShiftRight(header, 2), 0x0F) COMMENT 'Payload Type (bits 2-5 of header): 0=PAYLOAD_TYPE_REQ, 1=PAYLOAD_TYPE_RESPONSE, 2=PAYLOAD_TYPE_TXT_MSG, 3=PAYLOAD_TYPE_ACK, 4=PAYLOAD_TYPE_ADVERT, 5=PAYLOAD_TYPE_GRP_TXT, 6=PAYLOAD_TYPE_GRP_DATA, 7=PAYLOAD_TYPE_ANON_REQ, 8=PAYLOAD_TYPE_PATH, 9=PAYLOAD_TYPE_TRACE, 15=PAYLOAD_TYPE_RAW_CUSTOM',
+    payload_version UInt8 ALIAS bitAnd(bitShiftRight(header, 6), 0x03) COMMENT 'Payload Version (bits 6-7 of header): 0=V1, 1=V2, 2=V3, 3=V4',
+    path_len UInt8 ALIAS reinterpretAsUInt8(substring(packet, 2, 1)) COMMENT 'Length of the path field in bytes (byte 2 of packet)',
+    path String ALIAS hex(substring(packet, 3, path_len)) COMMENT 'Routing path as hex string (variable length, starts at byte 3, length path_len)',
+    payload String ALIAS substring(packet, 3 + path_len, length(packet) - 2 - path_len) COMMENT 'Payload (starts after path, up to 184 bytes)',
+    packet_hash String ALIAS hex(substring(SHA256(concat(
+        reinterpretAsFixedString(toUInt8(payload_type)),
+        CASE WHEN payload_type = 9 THEN reinterpretAsFixedString(toUInt32(path_len)) ELSE '' END,
+        payload
+    )), 1, 8)) COMMENT 'Packet hash calculated using SHA-256: payload_type + path_len (TRACE only) + payload_data, truncated to 8 bytes'
+) ENGINE = ReplacingMergeTree(ingest_timestamp)
+ORDER BY (ingest_timestamp, origin, mesh_timestamp, packet)
+PARTITION BY toYYYYMM(ingest_timestamp)
+COMMENT 'Table for storing MeshCore packets with extracted fields via alias columns.';
+
+-- View to decode node info packets (payload_type = 4)
+CREATE OR REPLACE VIEW meshcore_adverts AS
+SELECT
+    ingest_timestamp,
+    origin,
+    origin_pubkey,
+    mesh_timestamp,
+    packet,
+    path_len,
+    path,
+    broker,
+    topic,
+    hex(substring(payload, 1, 32)) AS public_key,
+    reinterpretAsUInt32(substring(payload, 33, 4)) AS adv_timestamp,
+    hex(substring(payload, 37, 64)) AS signature,
+    reinterpretAsUInt8(substring(payload, 101, 1)) AS appdata_flags,
+    bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x01) = 0x01 AS is_chat_node,
+    bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x02) = 0x02 AS is_repeater,
+    bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x03) = 0x03 AS is_room_server,
+    bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x10) = 0x10 AS has_location,
+    bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x20) = 0x20 AS has_feature1,
+    bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x40) = 0x40 AS has_feature2,
+    bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x80) = 0x80 AS has_name,
+    CASE WHEN bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x10) = 0x10
+        THEN reinterpretAsInt32(substring(payload, 102, 4))
+        ELSE NULL
+    END AS latitude_i,
+    CASE WHEN bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x10) = 0x10
+        THEN reinterpretAsInt32(substring(payload, 106, 4))
+        ELSE NULL
+    END AS longitude_i,
+    latitude_i * 1e-6 AS latitude,
+    longitude_i * 1e-6 AS longitude,
+    substring(
+        payload,
+        102
+        + multiIf(bitAnd(reinterpretAsUInt8(substring(payload, 101, 1)), 0x10) = 0x10, 8, 0)
+    ) AS     node_name,
+    hex(substring(payload, 1, 1)) AS node_hash,
+    packet_hash
+FROM meshcore_packets
+WHERE payload_type = 4;
+
+CREATE OR REPLACE VIEW meshcore_adverts_latest AS
+SELECT
+    public_key,
+    min(ingest_timestamp) AS first_heard,
+    max(ingest_timestamp) AS last_seen,
+    argMax(broker, ingest_timestamp) AS broker,
+    argMax(topic, ingest_timestamp) AS topic,
+    argMax(origin, ingest_timestamp) AS origin,
+    argMax(mesh_timestamp, ingest_timestamp) AS mesh_timestamp,
+    argMax(packet, ingest_timestamp) AS packet,
+    argMax(path_len, ingest_timestamp) AS path_len,
+    argMax(path, ingest_timestamp) AS path,
+    argMax(adv_timestamp, ingest_timestamp) AS adv_timestamp,
+    argMax(signature, ingest_timestamp) AS signature,
+    argMax(appdata_flags, ingest_timestamp) AS appdata_flags,
+    argMax(is_chat_node, ingest_timestamp) AS is_chat_node,
+    argMax(is_repeater, ingest_timestamp) AS is_repeater,
+    argMax(is_room_server, ingest_timestamp) AS is_room_server,
+    argMax(has_location, ingest_timestamp) AS has_location,
+    argMax(has_feature1, ingest_timestamp) AS has_feature1,
+    argMax(has_feature2, ingest_timestamp) AS has_feature2,
+    argMax(has_name, ingest_timestamp) AS has_name,
+    argMax(latitude_i, ingest_timestamp) AS latitude_i,
+    argMax(longitude_i, ingest_timestamp) AS longitude_i,
+    argMax(latitude, ingest_timestamp) AS latitude,
+    argMax(longitude, ingest_timestamp) AS longitude,
+    argMax(node_name, ingest_timestamp) AS node_name,
+    argMax(node_hash, ingest_timestamp) AS node_hash,
+    argMax(packet_hash, ingest_timestamp) AS packet_hash
+FROM meshcore_adverts
+GROUP BY public_key
+ORDER BY last_seen DESC;
+
+-- View to decode public channel group text messages (payload_type = 5)
+CREATE OR REPLACE VIEW meshcore_public_channel_messages AS
+SELECT
+    max(ingest_timestamp) AS ingest_timestamp,
+    min(mesh_timestamp) AS mesh_timestamp,
+    groupArray(origin) AS origins,
+    any(packet) AS packet,
+    any(path_len) AS path_len,
+    hex(substring(payload, 1, 1)) AS channel_hash,
+    hex(substring(payload, 2, 2)) AS mac,
+    substring(payload, 4) AS encrypted_message,
+    count() AS message_count,
+    groupArray((origin, hex(path))) AS origin_path_array, --deprecated
+    groupArray((origin, hex(origin_pubkey), hex(path))) AS origin_key_path_array, --deprecated
+    groupArray((broker, topic)) AS topic_broker_array, --deprecated
+    groupArray((origin, hex(origin_pubkey), hex(path), broker, topic)) AS origin_path_info,
+    any(packet_hash) AS message_id
+FROM meshcore_packets
+WHERE payload_type = 5
+GROUP BY payload
+ORDER BY ingest_timestamp DESC;
+
+
+-- +goose Down
+DROP VIEW IF EXISTS meshcore_public_channel_messages;
+DROP VIEW IF EXISTS meshcore_adverts_latest;
+DROP VIEW IF EXISTS meshcore_adverts;
+DROP TABLE IF EXISTS meshcore_packets;

ingest/migrations/002_create_unified_nodeinfo_view.sql

@@ -0,0 +1,19 @@
+-- +goose Up
+-- Unified latest node info view.
+-- MeshCore is the only supported protocol, so this is a thin projection over
+-- meshcore_adverts_latest that the web app (getNodePositions) reads from.
+CREATE OR REPLACE VIEW unified_latest_nodeinfo AS
+SELECT
+    public_key AS node_id,
+    node_name AS name,
+    substringUTF8(node_name, 1, 1) AS short_name,
+    latitude AS latitude,
+    longitude AS longitude,
+    first_heard AS first_seen,
+    last_seen,
+    'meshcore' AS type
+FROM meshcore_adverts_latest
+ORDER BY last_seen DESC;
+
+-- +goose Down
+DROP VIEW IF EXISTS unified_latest_nodeinfo;