diff --git a/ingest/.gitignore b/ingest/.gitignore new file mode 100644 index 0000000..257437e --- /dev/null +++ b/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 \ No newline at end of file diff --git a/ingest/Dockerfile b/ingest/Dockerfile new file mode 100644 index 0000000..c42c667 --- /dev/null +++ b/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"] diff --git a/ingest/README.md b/ingest/README.md new file mode 100644 index 0000000..87544d4 --- /dev/null +++ b/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 +``` diff --git a/ingest/clickhouse/Dockerfile b/ingest/clickhouse/Dockerfile new file mode 100644 index 0000000..a26050c --- /dev/null +++ b/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 \ No newline at end of file diff --git a/ingest/clickhouse/config.xml b/ingest/clickhouse/config.xml new file mode 100644 index 0000000..d556b7c --- /dev/null +++ b/ingest/clickhouse/config.xml @@ -0,0 +1,5 @@ + + + true + + diff --git a/ingest/clickhouse/meshcore_decrypt_udf.c b/ingest/clickhouse/meshcore_decrypt_udf.c new file mode 100644 index 0000000..2a77a14 --- /dev/null +++ b/ingest/clickhouse/meshcore_decrypt_udf.c @@ -0,0 +1,99 @@ +#include +#include +#include +#include +#include +#include + +#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; +} \ No newline at end of file diff --git a/ingest/clickhouse/meshcore_decrypt_udf.xml b/ingest/clickhouse/meshcore_decrypt_udf.xml new file mode 100644 index 0000000..05a95ce --- /dev/null +++ b/ingest/clickhouse/meshcore_decrypt_udf.xml @@ -0,0 +1,30 @@ + + + + meshcore_try_decrypt + executable + /usr/local/share/clickhouse-udf-bins/meshcore_decrypt_udf + + String + mac + + + String + encrypted_data + + + String + channel_hash_raw + + MsgPack + Nullable(Array(String)) + result + 1 + 30 + 10 + 10000 + 10000 + false + true + + \ No newline at end of file diff --git a/ingest/clickhouse/udf-config.xml b/ingest/clickhouse/udf-config.xml new file mode 100644 index 0000000..54c59b1 --- /dev/null +++ b/ingest/clickhouse/udf-config.xml @@ -0,0 +1,4 @@ + + /usr/local/share/clickhouse-udfs/*.xml + /usr/local/share/clickhouse-udf-bins/ + \ No newline at end of file diff --git a/ingest/clickhouse/users.xml b/ingest/clickhouse/users.xml new file mode 100644 index 0000000..6837afb --- /dev/null +++ b/ingest/clickhouse/users.xml @@ -0,0 +1,33 @@ + + + + + readonly + + ::/0 + + readonly + default + + + + + + + 1 + 0 + 10000000000 + 300 + 10000000 + 500000000 + + + + 0 + 180 + + + + + diff --git a/ingest/cmd/meshcoreingest/main.go b/ingest/cmd/meshcoreingest/main.go new file mode 100644 index 0000000..e7c747a --- /dev/null +++ b/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() +} diff --git a/ingest/cmd/meshcoreingest/main_test.go b/ingest/cmd/meshcoreingest/main_test.go new file mode 100644 index 0000000..8009d74 --- /dev/null +++ b/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) + } + } + }) + } +} diff --git a/ingest/go.mod b/ingest/go.mod new file mode 100644 index 0000000..7f10b0e --- /dev/null +++ b/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 +) diff --git a/ingest/go.sum b/ingest/go.sum new file mode 100644 index 0000000..184c8b7 --- /dev/null +++ b/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= diff --git a/ingest/internal/ingestcommon/ingest.go b/ingest/internal/ingestcommon/ingest.go new file mode 100644 index 0000000..560db22 --- /dev/null +++ b/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 +} diff --git a/ingest/internal/migrate/main.go b/ingest/internal/migrate/main.go new file mode 100644 index 0000000..621e554 --- /dev/null +++ b/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 +} diff --git a/ingest/migrations/001_create_meshcore_packets.sql b/ingest/migrations/001_create_meshcore_packets.sql new file mode 100644 index 0000000..743e875 --- /dev/null +++ b/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; diff --git a/ingest/migrations/002_create_unified_nodeinfo_view.sql b/ingest/migrations/002_create_unified_nodeinfo_view.sql new file mode 100644 index 0000000..692a390 --- /dev/null +++ b/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; diff --git a/.dockerignore b/meshexplorer/.dockerignore similarity index 100% rename from .dockerignore rename to meshexplorer/.dockerignore diff --git a/.gitignore b/meshexplorer/.gitignore similarity index 100% rename from .gitignore rename to meshexplorer/.gitignore diff --git a/Dockerfile b/meshexplorer/Dockerfile similarity index 100% rename from Dockerfile rename to meshexplorer/Dockerfile diff --git a/Dockerfile.bot b/meshexplorer/Dockerfile.bot similarity index 100% rename from Dockerfile.bot rename to meshexplorer/Dockerfile.bot diff --git a/README.md b/meshexplorer/README.md similarity index 100% rename from README.md rename to meshexplorer/README.md diff --git a/components.json b/meshexplorer/components.json similarity index 100% rename from components.json rename to meshexplorer/components.json diff --git a/docker-compose.yml b/meshexplorer/docker-compose.yml similarity index 100% rename from docker-compose.yml rename to meshexplorer/docker-compose.yml diff --git a/eslint.config.mjs b/meshexplorer/eslint.config.mjs similarity index 100% rename from eslint.config.mjs rename to meshexplorer/eslint.config.mjs diff --git a/next.config.ts b/meshexplorer/next.config.ts similarity index 100% rename from next.config.ts rename to meshexplorer/next.config.ts diff --git a/package-lock.json b/meshexplorer/package-lock.json similarity index 100% rename from package-lock.json rename to meshexplorer/package-lock.json diff --git a/package.json b/meshexplorer/package.json similarity index 100% rename from package.json rename to meshexplorer/package.json diff --git a/postcss.config.mjs b/meshexplorer/postcss.config.mjs similarity index 100% rename from postcss.config.mjs rename to meshexplorer/postcss.config.mjs diff --git a/public/favicon.svg b/meshexplorer/public/favicon.svg similarity index 100% rename from public/favicon.svg rename to meshexplorer/public/favicon.svg diff --git a/public/file.svg b/meshexplorer/public/file.svg similarity index 100% rename from public/file.svg rename to meshexplorer/public/file.svg diff --git a/public/globe.svg b/meshexplorer/public/globe.svg similarity index 100% rename from public/globe.svg rename to meshexplorer/public/globe.svg diff --git a/public/next.svg b/meshexplorer/public/next.svg similarity index 100% rename from public/next.svg rename to meshexplorer/public/next.svg diff --git a/public/vercel.svg b/meshexplorer/public/vercel.svg similarity index 100% rename from public/vercel.svg rename to meshexplorer/public/vercel.svg diff --git a/public/window.svg b/meshexplorer/public/window.svg similarity index 100% rename from public/window.svg rename to meshexplorer/public/window.svg diff --git a/scripts/README.md b/meshexplorer/scripts/README.md similarity index 100% rename from scripts/README.md rename to meshexplorer/scripts/README.md diff --git a/scripts/discord-bot.ts b/meshexplorer/scripts/discord-bot.ts similarity index 100% rename from scripts/discord-bot.ts rename to meshexplorer/scripts/discord-bot.ts diff --git a/scripts/lib/discord.ts b/meshexplorer/scripts/lib/discord.ts similarity index 100% rename from scripts/lib/discord.ts rename to meshexplorer/scripts/lib/discord.ts diff --git a/src/app/(app)/api-docs/page.tsx b/meshexplorer/src/app/(app)/api-docs/page.tsx similarity index 100% rename from src/app/(app)/api-docs/page.tsx rename to meshexplorer/src/app/(app)/api-docs/page.tsx diff --git a/src/app/(app)/layout.tsx b/meshexplorer/src/app/(app)/layout.tsx similarity index 100% rename from src/app/(app)/layout.tsx rename to meshexplorer/src/app/(app)/layout.tsx diff --git a/src/app/(app)/meshcore/node/[publicKey]/page.tsx b/meshexplorer/src/app/(app)/meshcore/node/[publicKey]/page.tsx similarity index 100% rename from src/app/(app)/meshcore/node/[publicKey]/page.tsx rename to meshexplorer/src/app/(app)/meshcore/node/[publicKey]/page.tsx diff --git a/src/app/(app)/messages/page.tsx b/meshexplorer/src/app/(app)/messages/page.tsx similarity index 100% rename from src/app/(app)/messages/page.tsx rename to meshexplorer/src/app/(app)/messages/page.tsx diff --git a/src/app/(app)/page.tsx b/meshexplorer/src/app/(app)/page.tsx similarity index 100% rename from src/app/(app)/page.tsx rename to meshexplorer/src/app/(app)/page.tsx diff --git a/src/app/(app)/search/page.tsx b/meshexplorer/src/app/(app)/search/page.tsx similarity index 100% rename from src/app/(app)/search/page.tsx rename to meshexplorer/src/app/(app)/search/page.tsx diff --git a/src/app/(app)/stats/page.tsx b/meshexplorer/src/app/(app)/stats/page.tsx similarity index 100% rename from src/app/(app)/stats/page.tsx rename to meshexplorer/src/app/(app)/stats/page.tsx diff --git a/src/app/(embed)/embed/map/page.tsx b/meshexplorer/src/app/(embed)/embed/map/page.tsx similarity index 100% rename from src/app/(embed)/embed/map/page.tsx rename to meshexplorer/src/app/(embed)/embed/map/page.tsx diff --git a/src/app/(embed)/layout.tsx b/meshexplorer/src/app/(embed)/layout.tsx similarity index 100% rename from src/app/(embed)/layout.tsx rename to meshexplorer/src/app/(embed)/layout.tsx diff --git a/src/app/api/chat/route.ts b/meshexplorer/src/app/api/chat/route.ts similarity index 100% rename from src/app/api/chat/route.ts rename to meshexplorer/src/app/api/chat/route.ts diff --git a/src/app/api/map/route.ts b/meshexplorer/src/app/api/map/route.ts similarity index 100% rename from src/app/api/map/route.ts rename to meshexplorer/src/app/api/map/route.ts diff --git a/src/app/api/meshcore/node/[publicKey]/neighbors/route.ts b/meshexplorer/src/app/api/meshcore/node/[publicKey]/neighbors/route.ts similarity index 100% rename from src/app/api/meshcore/node/[publicKey]/neighbors/route.ts rename to meshexplorer/src/app/api/meshcore/node/[publicKey]/neighbors/route.ts diff --git a/src/app/api/meshcore/node/[publicKey]/route.ts b/meshexplorer/src/app/api/meshcore/node/[publicKey]/route.ts similarity index 100% rename from src/app/api/meshcore/node/[publicKey]/route.ts rename to meshexplorer/src/app/api/meshcore/node/[publicKey]/route.ts diff --git a/src/app/api/meshcore/profilepicture.png/route.ts b/meshexplorer/src/app/api/meshcore/profilepicture.png/route.ts similarity index 100% rename from src/app/api/meshcore/profilepicture.png/route.ts rename to meshexplorer/src/app/api/meshcore/profilepicture.png/route.ts diff --git a/src/app/api/meshcore/search/route.ts b/meshexplorer/src/app/api/meshcore/search/route.ts similarity index 100% rename from src/app/api/meshcore/search/route.ts rename to meshexplorer/src/app/api/meshcore/search/route.ts diff --git a/src/app/api/meshcore/stream/chat/route.ts b/meshexplorer/src/app/api/meshcore/stream/chat/route.ts similarity index 100% rename from src/app/api/meshcore/stream/chat/route.ts rename to meshexplorer/src/app/api/meshcore/stream/chat/route.ts diff --git a/src/app/api/meshcore/stream/packets/route.ts b/meshexplorer/src/app/api/meshcore/stream/packets/route.ts similarity index 100% rename from src/app/api/meshcore/stream/packets/route.ts rename to meshexplorer/src/app/api/meshcore/stream/packets/route.ts diff --git a/src/app/api/neighbors/all/route.ts b/meshexplorer/src/app/api/neighbors/all/route.ts similarity index 100% rename from src/app/api/neighbors/all/route.ts rename to meshexplorer/src/app/api/neighbors/all/route.ts diff --git a/src/app/api/stats/nodes-over-time/route.ts b/meshexplorer/src/app/api/stats/nodes-over-time/route.ts similarity index 100% rename from src/app/api/stats/nodes-over-time/route.ts rename to meshexplorer/src/app/api/stats/nodes-over-time/route.ts diff --git a/src/app/api/stats/popular-channels/route.ts b/meshexplorer/src/app/api/stats/popular-channels/route.ts similarity index 100% rename from src/app/api/stats/popular-channels/route.ts rename to meshexplorer/src/app/api/stats/popular-channels/route.ts diff --git a/src/app/api/stats/repeater-prefixes/route.ts b/meshexplorer/src/app/api/stats/repeater-prefixes/route.ts similarity index 100% rename from src/app/api/stats/repeater-prefixes/route.ts rename to meshexplorer/src/app/api/stats/repeater-prefixes/route.ts diff --git a/src/app/api/stats/total-nodes/route.ts b/meshexplorer/src/app/api/stats/total-nodes/route.ts similarity index 100% rename from src/app/api/stats/total-nodes/route.ts rename to meshexplorer/src/app/api/stats/total-nodes/route.ts diff --git a/src/app/globals.css b/meshexplorer/src/app/globals.css similarity index 100% rename from src/app/globals.css rename to meshexplorer/src/app/globals.css diff --git a/src/app/layout.tsx b/meshexplorer/src/app/layout.tsx similarity index 100% rename from src/app/layout.tsx rename to meshexplorer/src/app/layout.tsx diff --git a/src/app/not-found.tsx b/meshexplorer/src/app/not-found.tsx similarity index 100% rename from src/app/not-found.tsx rename to meshexplorer/src/app/not-found.tsx diff --git a/src/components/AdvertDetails.tsx b/meshexplorer/src/components/AdvertDetails.tsx similarity index 100% rename from src/components/AdvertDetails.tsx rename to meshexplorer/src/components/AdvertDetails.tsx diff --git a/src/components/ChatBox.tsx b/meshexplorer/src/components/ChatBox.tsx similarity index 100% rename from src/components/ChatBox.tsx rename to meshexplorer/src/components/ChatBox.tsx diff --git a/src/components/ChatMessageItem.tsx b/meshexplorer/src/components/ChatMessageItem.tsx similarity index 100% rename from src/components/ChatMessageItem.tsx rename to meshexplorer/src/components/ChatMessageItem.tsx diff --git a/src/components/ConfigContext.tsx b/meshexplorer/src/components/ConfigContext.tsx similarity index 100% rename from src/components/ConfigContext.tsx rename to meshexplorer/src/components/ConfigContext.tsx diff --git a/src/components/ContactQRCode.tsx b/meshexplorer/src/components/ContactQRCode.tsx similarity index 100% rename from src/components/ContactQRCode.tsx rename to meshexplorer/src/components/ContactQRCode.tsx diff --git a/src/components/Header.tsx b/meshexplorer/src/components/Header.tsx similarity index 100% rename from src/components/Header.tsx rename to meshexplorer/src/components/Header.tsx diff --git a/src/components/InfoModal.tsx b/meshexplorer/src/components/InfoModal.tsx similarity index 100% rename from src/components/InfoModal.tsx rename to meshexplorer/src/components/InfoModal.tsx diff --git a/src/components/MapIcons.tsx b/meshexplorer/src/components/MapIcons.tsx similarity index 100% rename from src/components/MapIcons.tsx rename to meshexplorer/src/components/MapIcons.tsx diff --git a/src/components/MapLayerSettings.tsx b/meshexplorer/src/components/MapLayerSettings.tsx similarity index 100% rename from src/components/MapLayerSettings.tsx rename to meshexplorer/src/components/MapLayerSettings.tsx diff --git a/src/components/MapView.tsx b/meshexplorer/src/components/MapView.tsx similarity index 100% rename from src/components/MapView.tsx rename to meshexplorer/src/components/MapView.tsx diff --git a/src/components/MapWithChatClient.tsx b/meshexplorer/src/components/MapWithChatClient.tsx similarity index 100% rename from src/components/MapWithChatClient.tsx rename to meshexplorer/src/components/MapWithChatClient.tsx diff --git a/src/components/Modal.tsx b/meshexplorer/src/components/Modal.tsx similarity index 100% rename from src/components/Modal.tsx rename to meshexplorer/src/components/Modal.tsx diff --git a/src/components/NodeCard.tsx b/meshexplorer/src/components/NodeCard.tsx similarity index 100% rename from src/components/NodeCard.tsx rename to meshexplorer/src/components/NodeCard.tsx diff --git a/src/components/NodeLinkWithHover.tsx b/meshexplorer/src/components/NodeLinkWithHover.tsx similarity index 100% rename from src/components/NodeLinkWithHover.tsx rename to meshexplorer/src/components/NodeLinkWithHover.tsx diff --git a/src/components/PathVisualization.tsx b/meshexplorer/src/components/PathVisualization.tsx similarity index 100% rename from src/components/PathVisualization.tsx rename to meshexplorer/src/components/PathVisualization.tsx diff --git a/src/components/QueryProvider.tsx b/meshexplorer/src/components/QueryProvider.tsx similarity index 100% rename from src/components/QueryProvider.tsx rename to meshexplorer/src/components/QueryProvider.tsx diff --git a/src/components/RefreshButton.tsx b/meshexplorer/src/components/RefreshButton.tsx similarity index 100% rename from src/components/RefreshButton.tsx rename to meshexplorer/src/components/RefreshButton.tsx diff --git a/src/components/RegionSelector.tsx b/meshexplorer/src/components/RegionSelector.tsx similarity index 100% rename from src/components/RegionSelector.tsx rename to meshexplorer/src/components/RegionSelector.tsx diff --git a/src/components/SearchInput.tsx b/meshexplorer/src/components/SearchInput.tsx similarity index 100% rename from src/components/SearchInput.tsx rename to meshexplorer/src/components/SearchInput.tsx diff --git a/src/components/SearchResults.tsx b/meshexplorer/src/components/SearchResults.tsx similarity index 100% rename from src/components/SearchResults.tsx rename to meshexplorer/src/components/SearchResults.tsx diff --git a/src/contexts/RegionContext.tsx b/meshexplorer/src/contexts/RegionContext.tsx similarity index 100% rename from src/contexts/RegionContext.tsx rename to meshexplorer/src/contexts/RegionContext.tsx diff --git a/src/hooks/useAllNeighbors.ts b/meshexplorer/src/hooks/useAllNeighbors.ts similarity index 100% rename from src/hooks/useAllNeighbors.ts rename to meshexplorer/src/hooks/useAllNeighbors.ts diff --git a/src/hooks/useChatMessages.ts b/meshexplorer/src/hooks/useChatMessages.ts similarity index 100% rename from src/hooks/useChatMessages.ts rename to meshexplorer/src/hooks/useChatMessages.ts diff --git a/src/hooks/useConfigWithRegion.ts b/meshexplorer/src/hooks/useConfigWithRegion.ts similarity index 100% rename from src/hooks/useConfigWithRegion.ts rename to meshexplorer/src/hooks/useConfigWithRegion.ts diff --git a/src/hooks/useIntersectionObserver.ts b/meshexplorer/src/hooks/useIntersectionObserver.ts similarity index 100% rename from src/hooks/useIntersectionObserver.ts rename to meshexplorer/src/hooks/useIntersectionObserver.ts diff --git a/src/hooks/useLocalStorage.ts b/meshexplorer/src/hooks/useLocalStorage.ts similarity index 100% rename from src/hooks/useLocalStorage.ts rename to meshexplorer/src/hooks/useLocalStorage.ts diff --git a/src/hooks/useMapLayerSettings.ts b/meshexplorer/src/hooks/useMapLayerSettings.ts similarity index 100% rename from src/hooks/useMapLayerSettings.ts rename to meshexplorer/src/hooks/useMapLayerSettings.ts diff --git a/src/hooks/useMapPosition.ts b/meshexplorer/src/hooks/useMapPosition.ts similarity index 100% rename from src/hooks/useMapPosition.ts rename to meshexplorer/src/hooks/useMapPosition.ts diff --git a/src/hooks/useMeshcoreSearch.ts b/meshexplorer/src/hooks/useMeshcoreSearch.ts similarity index 100% rename from src/hooks/useMeshcoreSearch.ts rename to meshexplorer/src/hooks/useMeshcoreSearch.ts diff --git a/src/hooks/useMessageDecryption.ts b/meshexplorer/src/hooks/useMessageDecryption.ts similarity index 100% rename from src/hooks/useMessageDecryption.ts rename to meshexplorer/src/hooks/useMessageDecryption.ts diff --git a/src/hooks/useNeighbors.ts b/meshexplorer/src/hooks/useNeighbors.ts similarity index 100% rename from src/hooks/useNeighbors.ts rename to meshexplorer/src/hooks/useNeighbors.ts diff --git a/src/hooks/useNodeData.ts b/meshexplorer/src/hooks/useNodeData.ts similarity index 100% rename from src/hooks/useNodeData.ts rename to meshexplorer/src/hooks/useNodeData.ts diff --git a/src/hooks/useQueryParams.ts b/meshexplorer/src/hooks/useQueryParams.ts similarity index 100% rename from src/hooks/useQueryParams.ts rename to meshexplorer/src/hooks/useQueryParams.ts diff --git a/src/hooks/useStats.ts b/meshexplorer/src/hooks/useStats.ts similarity index 100% rename from src/hooks/useStats.ts rename to meshexplorer/src/hooks/useStats.ts diff --git a/src/lib/api.ts b/meshexplorer/src/lib/api.ts similarity index 100% rename from src/lib/api.ts rename to meshexplorer/src/lib/api.ts diff --git a/src/lib/clickhouse/actions.ts b/meshexplorer/src/lib/clickhouse/actions.ts similarity index 100% rename from src/lib/clickhouse/actions.ts rename to meshexplorer/src/lib/clickhouse/actions.ts diff --git a/src/lib/clickhouse/clickhouse.ts b/meshexplorer/src/lib/clickhouse/clickhouse.ts similarity index 100% rename from src/lib/clickhouse/clickhouse.ts rename to meshexplorer/src/lib/clickhouse/clickhouse.ts diff --git a/src/lib/clickhouse/streaming.ts b/meshexplorer/src/lib/clickhouse/streaming.ts similarity index 100% rename from src/lib/clickhouse/streaming.ts rename to meshexplorer/src/lib/clickhouse/streaming.ts diff --git a/src/lib/meshcore-map-nodeutils.ts b/meshexplorer/src/lib/meshcore-map-nodeutils.ts similarity index 100% rename from src/lib/meshcore-map-nodeutils.ts rename to meshexplorer/src/lib/meshcore-map-nodeutils.ts diff --git a/src/lib/meshcore.ts b/meshexplorer/src/lib/meshcore.ts similarity index 100% rename from src/lib/meshcore.ts rename to meshexplorer/src/lib/meshcore.ts diff --git a/src/lib/node-utils.ts b/meshexplorer/src/lib/node-utils.ts similarity index 100% rename from src/lib/node-utils.ts rename to meshexplorer/src/lib/node-utils.ts diff --git a/src/lib/pathUtils.ts b/meshexplorer/src/lib/pathUtils.ts similarity index 100% rename from src/lib/pathUtils.ts rename to meshexplorer/src/lib/pathUtils.ts diff --git a/src/lib/regionFilters.ts b/meshexplorer/src/lib/regionFilters.ts similarity index 100% rename from src/lib/regionFilters.ts rename to meshexplorer/src/lib/regionFilters.ts diff --git a/src/lib/regions.ts b/meshexplorer/src/lib/regions.ts similarity index 100% rename from src/lib/regions.ts rename to meshexplorer/src/lib/regions.ts diff --git a/src/lib/utils.ts b/meshexplorer/src/lib/utils.ts similarity index 100% rename from src/lib/utils.ts rename to meshexplorer/src/lib/utils.ts diff --git a/src/middleware.ts b/meshexplorer/src/middleware.ts similarity index 100% rename from src/middleware.ts rename to meshexplorer/src/middleware.ts diff --git a/src/types/map.ts b/meshexplorer/src/types/map.ts similarity index 100% rename from src/types/map.ts rename to meshexplorer/src/types/map.ts diff --git a/tsconfig.json b/meshexplorer/tsconfig.json similarity index 100% rename from tsconfig.json rename to meshexplorer/tsconfig.json