Merge pull request #33 from ajvpot/release-ingest

add ingest code: monorepo + one-command deploy
2026-08-08 01:22:43 +02:00 · 2026-05-29 01:51:20 -04:00
parent 815d465566 6be73b04a6
commit 49847c46af
115 changed files with 4157 additions and 276 deletions
@@ -0,0 +1,46 @@
+# MeshExplorer unified stack configuration.
+# Copy this file to .env and fill in the values, then run:
+#   docker compose up --build
+# (add `--profile bot` to also start the Discord relay).
+
+# ─── ClickHouse ──────────────────────────────────────────────────────────────
+# The read/write "default" user is used by the ingest daemon and the migration
+# runner. Set a real password before deploying.
+CLICKHOUSE_DB=default
+CLICKHOUSE_USER=default
+CLICKHOUSE_PASSWORD=changeme
+
+# Read-only user used by the web app and the Discord bot. This account is only
+# reachable on the internal docker network; the default matches ingest/clickhouse/users.xml.
+CLICKHOUSE_READONLY_USER=readonly
+CLICKHOUSE_READONLY_PASSWORD=readonly
+
+# ─── MeshCore MQTT ingest ────────────────────────────────────────────────────
+# JSON array of MQTT brokers to subscribe to for meshcore packets. Each entry:
+#   { "url": "...", "username": "...", "password": "...", "topics": ["meshcore/#"] }
+# "topics" is optional and defaults to ["meshcore/#"]. The ingest daemon exits
+# with an error if this is empty, so configure at least one broker.
+MQTT_BROKERS=[{"url":"tcp://mqtt.example.com:1883","username":"CHANGE_ME","password":"CHANGE_ME","topics":["meshcore/#"]}]
+MQTT_CLIENT_ID=meshcore-ingest
+
+# ─── Web app ─────────────────────────────────────────────────────────────────
+# Base URL for client-side API calls. Leave empty to use relative URLs.
+NEXT_PUBLIC_API_URL=
+
+# ─── Discord relay bot (optional, --profile bot) ─────────────────────────────
+# Required when running the bot. Create a webhook in your Discord server.
+DISCORD_WEBHOOK_URL=
+# Optional: post into a specific thread instead of the channel.
+DISCORD_THREAD_ID=
+# Region filter for messages (e.g. seattle).
+MESH_REGION=seattle
+# Poll interval (ms) and batch size.
+POLL_INTERVAL=300
+MAX_ROWS_PER_POLL=50
+# Comma-separated base64 private keys used to decrypt channel messages.
+PRIVATE_KEYS=
+
+# ─── Grafana ─────────────────────────────────────────────────────────────────
+# Admin password for the bundled Grafana (published on 127.0.0.1:3000). A
+# ClickHouse datasource is auto-provisioned using the read-only user above.
+GRAFANA_ADMIN_PASSWORD=admin
@@ -1,41 +1,14 @@
-# See https://help.github.com/articles/ignoring-files/ for more about ignoring files.
+# Local environment configuration (never commit real secrets)
+.env
+.env.*
+!.env.example

-# dependencies
-/node_modules
-/.pnp
-.pnp.*
-.yarn/*
-!.yarn/patches
-!.yarn/plugins
-!.yarn/releases
-!.yarn/versions
-
-# testing
-/coverage
-
-# next.js
-/.next/
-/out/
-
-# production
-/build
-
-# misc
+# OS files
 .DS_Store
-*.pem

-# debug
-npm-debug.log*
-yarn-debug.log*
-yarn-error.log*
-.pnpm-debug.log*
-
-# env files (can opt-in for committing if needed)
-.env*
-
-# vercel
-.vercel
-
-# typescript
-*.tsbuildinfo
-next-env.d.ts
+# Go build artifacts from the ingest module
+/ingest/meshcoreingest
+/ingest/migrate
+/ingest/clickhouse-meshingest
+*.test
+*.out
@@ -1,143 +1,82 @@
 # MeshExplorer

-MeshExplorer is a real-time map, chat client, and packet analysis tool for mesh networks using MeshCore and Meshtastic. It enables users to visualize mesh nodes on a map, communicate via chat, and analyze packet data in real time.
+A real-time map, chat client, and packet-analysis tool for **MeshCore** mesh
+networks. This repository ships the whole stack so it can be brought up with a
+single `docker compose up`:

-## Features
- Real-time map of mesh network nodes (MeshCore and Meshtastic)
- Integrated chat client for mesh channels
- Packet analysis and inspection tools
- Customizable map layers and clustering
- Modern, responsive UI
+| Component | Path | Description |
+|-----------|------|-------------|
+| Web app | [`meshexplorer/`](./meshexplorer) | Next.js UI + API (map, chat, stats, packet analysis) |
+| Ingest + DB | [`ingest/`](./ingest) | Go MeshCore MQTT→ClickHouse ingest, ClickHouse image, and SQL migrations |
+| Discord relay | [`meshexplorer/`](./meshexplorer) (`Dockerfile.bot`) | Optional bot that relays MeshCore channel messages to Discord |
+| Grafana | [`grafana/`](./grafana) | Dashboards with a pre-provisioned ClickHouse datasource (read-only user), on `127.0.0.1:3000` |

-## Getting Started
+## Architecture

-First, run the development server:
+```
+                MQTT brokers (you configure)
+                          │
+                          ▼
+   ┌──────────────┐   ┌──────────────┐
+   │ meshcoreingest│──▶│  ClickHouse  │◀── migrate (one-shot, applies schema)
+   └──────────────┘   └──────┬───────┘
+                             │ (readonly user)
+                  ┌──────────┴──────────┐
+                  ▼              ▼              ▼
+            meshexplorer     discord-bot      grafana
+            (web UI :3001)   (--profile bot)  (:3000)
+```
+
+## Quick start
+
+Requirements: Docker + Docker Compose.

 ```bash
-npm run dev
-# or
-yarn dev
-# or
-pnpm dev
-# or
-bun dev
+cp .env.example .env
+# Edit .env — at minimum set:
+#   CLICKHOUSE_PASSWORD   (read/write user, used by ingest + migrations)
+#   MQTT_BROKERS          (JSON array of meshcore MQTT brokers to ingest from)
+# Optional, for the Discord relay: DISCORD_WEBHOOK_URL (+ run with --profile bot)
+
+docker compose up --build
 ```

-Open [http://localhost:3000](http://localhost:3000) with your browser to see the result.
+Then open <http://localhost:3001>.

-You can start editing the page by modifying `app/page.tsx`. The page auto-updates as you edit the file.
+Startup order is handled automatically: ClickHouse becomes healthy → `migrate`
+applies the schema and exits → `meshcoreingest` and `meshexplorer` start.

-## Environment Variables
-
-### ClickHouse Database Configuration
-
-The application connects to ClickHouse using the following environment variables:
-
- `CLICKHOUSE_HOST` - ClickHouse server hostname (default: `localhost`)
- `CLICKHOUSE_PORT` - ClickHouse server port (default: `8123`)
- `CLICKHOUSE_USER` - ClickHouse username (default: `default`)
- `CLICKHOUSE_PASSWORD` - ClickHouse password (default: `password`)
-
-### `NEXT_PUBLIC_API_URL`
-
-This environment variable allows you to override the API base URL for frontend development purposes. When set, all API calls will be made to the specified URL instead of using relative URLs.
-
-**Use case**: This is useful when you want to develop the frontend without direct access to the ClickHouse database, by pointing to a remote API endpoint.
-
-**Example**:
-```bash
-NEXT_PUBLIC_API_URL=https://map.w0z.is
-```
-
-**Important**: When this environment variable is set, the local API routes (`/api/*`) will not work. Make sure the remote API endpoint provides the same API structure and endpoints.
-
-**Default behavior**: If not set, the application uses relative URLs and works with the local Next.js API routes.
-
-### CORS Support
-
-The application includes middleware (`middleware.ts`) that automatically adds CORS headers to all API routes. This allows:
-
- Cross-origin requests from localhost to production APIs
- Cross-protocol requests (HTTP on localhost to HTTPS in production)
- Preflight OPTIONS requests are handled automatically
-
-The middleware applies the following CORS headers to all `/api/*` routes:
- `Access-Control-Allow-Origin: *`
- `Access-Control-Allow-Methods: GET, POST, PUT, DELETE, OPTIONS`
- `Access-Control-Allow-Headers: Content-Type, Authorization, X-Requested-With`
- `Access-Control-Allow-Credentials: true`
-
-## Learn More
-
- [Next.js Documentation](https://nextjs.org/docs) - learn about Next.js features and API.
- [MeshCore](https://github.com/your-org/meshcore) - mesh network backend
- [Meshtastic](https://meshtastic.org/) - open source mesh communication project
-
-## Docker Deployment
-
-The application includes Docker support for easy deployment. The Docker configuration is set up to connect to ClickHouse running on the Docker host.
-
-### Prerequisites
-
- Docker and Docker Compose installed
- ClickHouse running on the Docker host (default port 8123)
- External Docker network `shared-network` must exist (see setup instructions below)
-
-### External Network Setup
-
-The application requires an external Docker network called `shared-network` to communicate with ClickHouse. You must create this network before running the application:
+To also run the Discord relay:

 ```bash
-docker network create shared-network
+docker compose --profile bot up --build
 ```

-**Note**: If the network already exists, this command will show an error but can be safely ignored.
+## Configuration

-### Running with Docker Compose
+All configuration is via environment variables in `.env` (see
+[`.env.example`](./.env.example) for the full list and defaults). Highlights:

-1. **Create the required external network (if not already created):**
-   ```bash
-   docker network create shared-network
-   ```
+- **ClickHouse** — two accounts. The read/write `default` user
+  (`CLICKHOUSE_PASSWORD`) is used by the ingest daemon and the migration runner;
+  the `readonly` user (`CLICKHOUSE_READONLY_PASSWORD`) is used by the web app and
+  the Discord bot. ClickHouse is only published to `127.0.0.1` for debugging and
+  is otherwise reachable only on the internal `meshnet` network.
+- **MQTT_BROKERS** — a JSON array; each entry is
+  `{ "url", "username", "password", "topics" }` (`topics` defaults to
+  `["meshcore/#"]`). The ingest daemon exits with a clear error if this is unset,
+  so configure at least one broker.

-2. **Build and start the application:**
-   ```bash
-   docker-compose up --build
-   ```
+## Development

-3. **Access the application:**
-   Open [http://localhost:3001](http://localhost:3001) in your browser.
+Each component can be run on its own:

-### Docker Configuration
+- Web app: see [`meshexplorer/README.md`](./meshexplorer/README.md)
+  (`npm install && npm run dev`).
+- Ingest: see [`ingest/README.md`](./ingest/README.md) (`go build ./...`).

-The `docker-compose.yml` file is configured with:
- **Port mapping**: Container port 3000 → Host port 3001
- **ClickHouse connection**: Uses `clickhouse` hostname to connect to ClickHouse via the shared network
- **External network**: Requires `shared-network` to be created externally
- **Environment variables**: Pre-configured for typical ClickHouse setup
+## Security notes

-### Customizing ClickHouse Connection
-
-You can customize the ClickHouse connection by modifying the environment variables in `docker-compose.yml`:
-
-```yaml
-environment:
-  - CLICKHOUSE_HOST=your-clickhouse-host
-  - CLICKHOUSE_PORT=8123
-  - CLICKHOUSE_USER=your-username
-  - CLICKHOUSE_PASSWORD=your-password
-```
-
-### Building with BuildKit
-
-For faster builds with caching, enable BuildKit:
-
-```bash
-DOCKER_BUILDKIT=1 docker-compose up --build
-```
-
-## Deploy
-
-The easiest way to deploy your Next.js app is to use the [Vercel Platform](https://vercel.com/new?utm_medium=default-template&filter=next.js&utm_source=create-next-app&utm_campaign=create-next-app-readme) from the creators of Next.js.
-
-Check out the [Next.js deployment documentation](https://nextjs.org/docs/app/building-your-application/deploying) for more details.
+- `.env` is gitignored — keep real credentials out of version control.
+- If you previously used the bundled defaults, rotate any secrets before going
+  to production.
@@ -1,59 +1,157 @@
-version: '3.8'
-
 services:
+  # ClickHouse database (custom image bundles the meshcore decrypt UDF)
+  clickhouse:
+    build: ./ingest/clickhouse
+    environment:
+      - CLICKHOUSE_DB=${CLICKHOUSE_DB:-default}
+      - CLICKHOUSE_USER=${CLICKHOUSE_USER:-default}
+      - CLICKHOUSE_PASSWORD=${CLICKHOUSE_PASSWORD:-}
+      - CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT=1
+    ports:
+      # Published to localhost only, for debugging. Not required by the stack.
+      - "127.0.0.1:8123:8123"
+      - "127.0.0.1:9000:9000"
+    volumes:
+      - clickhouse-data:/var/lib/clickhouse
+    healthcheck:
+      test: ["CMD-SHELL", "clickhouse-client --user $$CLICKHOUSE_USER --password $$CLICKHOUSE_PASSWORD --query 'SELECT 1' || exit 1"]
+      interval: 5s
+      timeout: 5s
+      retries: 30
+      start_period: 10s
+    restart: unless-stopped
+    networks:
+      - meshnet
+
+  # One-shot migration runner (goose). Applies the meshcore schema then exits.
+  migrate:
+    build: ./ingest
+    command:
+      - "./migrate"
+      - "-host"
+      - "clickhouse"
+      - "-port"
+      - "9000"
+      - "-database"
+      - "${CLICKHOUSE_DB:-default}"
+      - "-username"
+      - "${CLICKHOUSE_USER:-default}"
+      - "-password"
+      - "${CLICKHOUSE_PASSWORD:-}"
+      - "-path"
+      - "migrations"
+      - "-action"
+      - "up"
+    depends_on:
+      clickhouse:
+        condition: service_healthy
+    restart: "no"
+    networks:
+      - meshnet
+
+  # MeshCore MQTT -> ClickHouse ingest daemon
+  meshcoreingest:
+    build: ./ingest
+    command: ["./meshcoreingest"]
+    environment:
+      - CLICKHOUSE_HOST=clickhouse
+      - CLICKHOUSE_PORT=9000
+      - CLICKHOUSE_DB=${CLICKHOUSE_DB:-default}
+      - CLICKHOUSE_USER=${CLICKHOUSE_USER:-default}
+      - CLICKHOUSE_PASSWORD=${CLICKHOUSE_PASSWORD:-}
+      - MQTT_BROKERS=${MQTT_BROKERS}
+      - MQTT_CLIENT_ID=${MQTT_CLIENT_ID:-meshcore-ingest}
+    depends_on:
+      clickhouse:
+        condition: service_healthy
+      migrate:
+        condition: service_completed_successfully
+    restart: unless-stopped
+    networks:
+      - meshnet
+
+  # MeshExplorer web app (Next.js). Reads ClickHouse over HTTP (8123) as the
+  # readonly user.
  meshexplorer:
    build:
-      context: .
+      context: ./meshexplorer
      dockerfile: Dockerfile
-    ports:
-      - "3001:3000"
    environment:
-      # Next.js Configuration
      - NODE_ENV=production
      - PORT=3000
      - HOSTNAME=0.0.0.0
-      
-      # ClickHouse Database Configuration
-      - CLICKHOUSE_HOST=${CLICKHOUSE_HOST:-clickhouse}
-      - CLICKHOUSE_PORT=${CLICKHOUSE_PORT:-8123}
-      - CLICKHOUSE_USER=${CLICKHOUSE_USER:-default}
-      - CLICKHOUSE_PASSWORD=${CLICKHOUSE_PASSWORD:-password}
-      
-      # Next.js API Configuration
+      - CLICKHOUSE_HOST=clickhouse
+      - CLICKHOUSE_PORT=8123
+      - CLICKHOUSE_USER=${CLICKHOUSE_READONLY_USER:-readonly}
+      - CLICKHOUSE_PASSWORD=${CLICKHOUSE_READONLY_PASSWORD:-readonly}
      - NEXT_PUBLIC_API_URL=${NEXT_PUBLIC_API_URL:-}
+    ports:
+      - "3001:3000"
+    depends_on:
+      clickhouse:
+        condition: service_healthy
+      migrate:
+        condition: service_completed_successfully
    restart: unless-stopped
    init: true
    networks:
-      - shared-network
+      - meshnet

+  # MeshCore -> Discord relay bot. Optional: enabled with the "bot" profile.
+  #   docker compose --profile bot up
  discord-bot:
    build:
-      context: .
+      context: ./meshexplorer
      dockerfile: Dockerfile.bot
+    profiles: ["bot"]
    environment:
-      # Node.js Configuration
      - NODE_ENV=production
-      
-      # ClickHouse Database Configuration
-      - CLICKHOUSE_HOST=${CLICKHOUSE_HOST:-clickhouse}
-      - CLICKHOUSE_PORT=${CLICKHOUSE_PORT:-8123}
-      - CLICKHOUSE_USER=${CLICKHOUSE_USER:-default}
-      - CLICKHOUSE_PASSWORD=${CLICKHOUSE_PASSWORD:-password}
-      
-      # Discord Bot Configuration
+      - CLICKHOUSE_HOST=clickhouse
+      - CLICKHOUSE_PORT=8123
+      - CLICKHOUSE_USER=${CLICKHOUSE_READONLY_USER:-readonly}
+      - CLICKHOUSE_PASSWORD=${CLICKHOUSE_READONLY_PASSWORD:-readonly}
      - DISCORD_WEBHOOK_URL=${DISCORD_WEBHOOK_URL}
      - DISCORD_THREAD_ID=${DISCORD_THREAD_ID:-}
      - MESH_REGION=${MESH_REGION:-seattle}
-      - POLL_INTERVAL=${POLL_INTERVAL:-1000}
+      - POLL_INTERVAL=${POLL_INTERVAL:-300}
      - MAX_ROWS_PER_POLL=${MAX_ROWS_PER_POLL:-50}
      - PRIVATE_KEYS=${PRIVATE_KEYS:-}
+    depends_on:
+      clickhouse:
+        condition: service_healthy
+      migrate:
+        condition: service_completed_successfully
    restart: unless-stopped
    init: true
    networks:
-      - shared-network
+      - meshnet
+
+  # Grafana dashboards, wired to ClickHouse via the read-only user.
+  grafana:
+    image: grafana/grafana:12.1.1
+    environment:
+      - GF_SECURITY_ADMIN_PASSWORD=${GRAFANA_ADMIN_PASSWORD:-admin}
+      - GF_INSTALL_PLUGINS=grafana-clickhouse-datasource
+      - GF_USERS_DEFAULT_THEME=dark
+      # Consumed by the provisioned ClickHouse datasource (grafana/provisioning).
+      - CLICKHOUSE_READONLY_USER=${CLICKHOUSE_READONLY_USER:-readonly}
+      - CLICKHOUSE_READONLY_PASSWORD=${CLICKHOUSE_READONLY_PASSWORD:-readonly}
+    ports:
+      - "127.0.0.1:3000:3000"
+    volumes:
+      - grafana-data:/var/lib/grafana
+      - ./grafana/provisioning:/etc/grafana/provisioning:ro
    depends_on:
-      - meshexplorer
+      clickhouse:
+        condition: service_healthy
+    restart: unless-stopped
+    networks:
+      - meshnet
+
+volumes:
+  clickhouse-data:
+  grafana-data:

 networks:
-  shared-network:
-    external: true 
+  meshnet:
+    driver: bridge
@@ -0,0 +1,13 @@
+apiVersion: 1
+
+providers:
+  - name: MeshCore
+    orgId: 1
+    type: file
+    disableDeletion: false
+    # Allow edits in the UI without them being reverted on the next scan.
+    allowUiUpdates: true
+    updateIntervalSeconds: 30
+    options:
+      path: /etc/grafana/provisioning/dashboards/json
+      foldersFromFilesStructure: false
@@ -0,0 +1,17 @@
+apiVersion: 1
+
+datasources:
+  - name: ClickHouse
+    type: grafana-clickhouse-datasource
+    # Pinned uid so the provisioned MeshCore dashboard's panel references resolve.
+    uid: clickhouse
+    access: proxy
+    isDefault: true
+    jsonData:
+      host: clickhouse
+      port: 9000
+      protocol: native
+      username: $CLICKHOUSE_READONLY_USER
+    secureJsonData:
+      password: $CLICKHOUSE_READONLY_PASSWORD
+    editable: false
@@ -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
@@ -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"]
@@ -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
+```
@@ -0,0 +1,10 @@
+FROM clickhouse/clickhouse-server:25.6.2.5
+
+# Read-only user used by the web app / discord bot
+COPY users.xml /etc/clickhouse-server/users.d/readonly_user.xml
+
+# Access control configuration
+COPY config.xml /etc/clickhouse-server/config.d/access_control.xml
+
+# System log retention (TTLs + tame text_log) so diagnostics don't grow unbounded
+COPY system-logs.xml /etc/clickhouse-server/config.d/system-logs.xml
@@ -0,0 +1,5 @@
+<clickhouse>
+    <access_control_improvements>
+        <settings_constraints_replace_previous>true</settings_constraints_replace_previous>
+    </access_control_improvements>
+</clickhouse>
@@ -0,0 +1,47 @@
+<clickhouse>
+    <!-- Keep ClickHouse's internal system log tables from growing unbounded.
+         text_log defaults to capturing Trace-level execution messages (Aggregator,
+         MergeTreeSequentialSource, ...) which is enormous; cap it at warning. The query
+         profiler (trace_log) is disabled in users.xml. Everything gets a short TTL so a
+         long-running server can't accumulate hundreds of GB of diagnostics. -->
+    <text_log>
+        <level>warning</level>
+        <ttl>event_date + INTERVAL 3 DAY DELETE</ttl>
+    </text_log>
+    <trace_log>
+        <ttl>event_date + INTERVAL 3 DAY DELETE</ttl>
+    </trace_log>
+    <query_log>
+        <ttl>event_date + INTERVAL 14 DAY DELETE</ttl>
+    </query_log>
+    <query_thread_log>
+        <ttl>event_date + INTERVAL 7 DAY DELETE</ttl>
+    </query_thread_log>
+    <query_views_log>
+        <ttl>event_date + INTERVAL 7 DAY DELETE</ttl>
+    </query_views_log>
+    <processors_profile_log>
+        <ttl>event_date + INTERVAL 7 DAY DELETE</ttl>
+    </processors_profile_log>
+    <part_log>
+        <ttl>event_date + INTERVAL 14 DAY DELETE</ttl>
+    </part_log>
+    <metric_log>
+        <ttl>event_date + INTERVAL 7 DAY DELETE</ttl>
+    </metric_log>
+    <asynchronous_metric_log>
+        <ttl>event_date + INTERVAL 7 DAY DELETE</ttl>
+    </asynchronous_metric_log>
+    <query_metric_log>
+        <ttl>event_date + INTERVAL 7 DAY DELETE</ttl>
+    </query_metric_log>
+    <latency_log>
+        <ttl>event_date + INTERVAL 7 DAY DELETE</ttl>
+    </latency_log>
+    <error_log>
+        <ttl>event_date + INTERVAL 14 DAY DELETE</ttl>
+    </error_log>
+    <asynchronous_insert_log>
+        <ttl>event_date + INTERVAL 7 DAY DELETE</ttl>
+    </asynchronous_insert_log>
+</clickhouse>
@@ -0,0 +1,44 @@
+<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>
+        <!-- Disable the query profiler for the default profile so trace_log stays empty
+             (it otherwise samples every query every 1s -> billions of rows over time). -->
+        <default>
+            <query_profiler_real_time_period_ns>0</query_profiler_real_time_period_ns>
+            <query_profiler_cpu_time_period_ns>0</query_profiler_cpu_time_period_ns>
+        </default>
+        <!-- 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>
+            <!-- No row/byte read caps: the map/stats views scan the full meshcore_packets
+                 table, which grows over time. memory + execution-time limits remain the
+                 guardrails. (0 = unlimited.) -->
+            <max_rows_to_read>0</max_rows_to_read>
+            <max_bytes_to_read>0</max_bytes_to_read>
+            <query_profiler_real_time_period_ns>0</query_profiler_real_time_period_ns>
+            <query_profiler_cpu_time_period_ns>0</query_profiler_cpu_time_period_ns>
+            <constraints>
+                <max_execution_time>
+                    <changeable_in_readonly/>
+                    <min>0</min>
+                    <max>180</max>
+                </max_execution_time>
+            </constraints>
+        </readonly>
+    </profiles>
+</clickhouse>
@@ -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/meshexplorer/ingest/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()
+}
@@ -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)
+				}
+			}
+		})
+	}
+}
@@ -0,0 +1,38 @@
+module github.com/ajvpot/meshexplorer/ingest
+
+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
+)
@@ -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=
@@ -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
+}
@@ -0,0 +1,117 @@
+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)
+	}
+
+	// ClickHouse is the target database; goose defaults to postgres otherwise.
+	if err := goose.SetDialect("clickhouse"); err != nil {
+		log.Fatalf("Failed to set goose dialect: %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
+}
@@ -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;
@@ -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;
@@ -0,0 +1,41 @@
+# See https://help.github.com/articles/ignoring-files/ for more about ignoring files.
+
+# dependencies
+/node_modules
+/.pnp
+.pnp.*
+.yarn/*
+!.yarn/patches
+!.yarn/plugins
+!.yarn/releases
+!.yarn/versions
+
+# testing
+/coverage
+
+# next.js
+/.next/
+/out/
+
+# production
+/build
+
+# misc
+.DS_Store
+*.pem
+
+# debug
+npm-debug.log*
+yarn-debug.log*
+yarn-error.log*
+.pnpm-debug.log*
+
+# env files (can opt-in for committing if needed)
+.env*
+
+# vercel
+.vercel
+
+# typescript
+*.tsbuildinfo
+next-env.d.ts
@@ -0,0 +1,97 @@
+# MeshExplorer
+
+MeshExplorer is a real-time map, chat client, and packet analysis tool for MeshCore mesh networks. It enables users to visualize mesh nodes on a map, communicate via chat, and analyze packet data in real time.
+
+> This directory contains the web app only. For the full stack (ClickHouse,
+> migrations, ingest, and this app) see the [root README](../README.md) and the
+> unified `docker compose` setup.
+
+## Features
+- Real-time map of mesh network nodes (MeshCore)
+- Integrated chat client for mesh channels
+- Packet analysis and inspection tools
+- Customizable map layers and clustering
+- Modern, responsive UI
+
+## Getting Started
+
+First, run the development server:
+
+```bash
+npm run dev
+# or
+yarn dev
+# or
+pnpm dev
+# or
+bun dev
+```
+
+Open [http://localhost:3000](http://localhost:3000) with your browser to see the result.
+
+You can start editing the page by modifying `app/page.tsx`. The page auto-updates as you edit the file.
+
+## Environment Variables
+
+### ClickHouse Database Configuration
+
+The application connects to ClickHouse using the following environment variables:
+
+- `CLICKHOUSE_HOST` - ClickHouse server hostname (default: `localhost`)
+- `CLICKHOUSE_PORT` - ClickHouse server port (default: `8123`)
+- `CLICKHOUSE_USER` - ClickHouse username (default: `default`)
+- `CLICKHOUSE_PASSWORD` - ClickHouse password (default: `password`)
+
+### `NEXT_PUBLIC_API_URL`
+
+This environment variable allows you to override the API base URL for frontend development purposes. When set, all API calls will be made to the specified URL instead of using relative URLs.
+
+**Use case**: This is useful when you want to develop the frontend without direct access to the ClickHouse database, by pointing to a remote API endpoint.
+
+**Example**:
+```bash
+NEXT_PUBLIC_API_URL=https://map.w0z.is
+```
+
+**Important**: When this environment variable is set, the local API routes (`/api/*`) will not work. Make sure the remote API endpoint provides the same API structure and endpoints.
+
+**Default behavior**: If not set, the application uses relative URLs and works with the local Next.js API routes.
+
+### CORS Support
+
+The application includes middleware (`middleware.ts`) that automatically adds CORS headers to all API routes. This allows:
+
+- Cross-origin requests from localhost to production APIs
+- Cross-protocol requests (HTTP on localhost to HTTPS in production)
+- Preflight OPTIONS requests are handled automatically
+
+The middleware applies the following CORS headers to all `/api/*` routes:
+- `Access-Control-Allow-Origin: *`
+- `Access-Control-Allow-Methods: GET, POST, PUT, DELETE, OPTIONS`
+- `Access-Control-Allow-Headers: Content-Type, Authorization, X-Requested-With`
+- `Access-Control-Allow-Credentials: true`
+
+## Learn More
+
+- [Next.js Documentation](https://nextjs.org/docs) - learn about Next.js features and API.
+- [MeshCore](https://meshcore.co.uk/) - mesh network firmware
+
+## Docker Deployment
+
+This app is deployed as part of the unified stack at the repository root. From
+the repo root:
+
+```bash
+cp .env.example .env   # then edit
+docker compose up --build
+```
+
+The web app is served on <http://localhost:3001>. ClickHouse, migrations, and the
+MeshCore ingest are started for you. See the [root README](../README.md) for the
+full list of services and environment variables.
+
+## Deploy
+
+The easiest way to deploy your Next.js app is to use the [Vercel Platform](https://vercel.com/new?utm_medium=default-template&filter=next.js&utm_source=create-next-app&utm_campaign=create-next-app-readme) from the creators of Next.js.
+
+Check out the [Next.js deployment documentation](https://nextjs.org/docs/app/building-your-application/deploying) for more details.
@@ -16,7 +16,7 @@ const geistMono = Geist_Mono({

 export const metadata: Metadata = {
  title: "MeshExplorer",
-  description: "A real-time map, chat client, and packet analysis tool for mesh networks using MeshCore and Meshtastic.",
+  description: "A real-time map, chat client, and packet analysis tool for MeshCore mesh networks.",
  icons: {
    icon: { url: "/favicon.svg", type: "image/svg+xml" },
    apple: { url: "/favicon.svg", type: "image/svg+xml" }
@@ -15,7 +15,7 @@ export default function InfoModal({ isOpen, onClose }: InfoModalProps) {
          <div>
            <p className="text-sm text-gray-600 dark:text-gray-300 mb-3">
              {getAppName()} is a web-based visualization tool for exploring and monitoring mesh networks. 
-              It provides real-time mapping of network nodes, message tracking, and statistical analysis for MeshCore and Meshtastic networks.
+              It provides real-time mapping of network nodes, message tracking, and statistical analysis for MeshCore networks.
            </p>
          </div>
          
@@ -23,19 +23,13 @@ interface PopupContentProps {
 // Individual node marker component
 export function NodeMarker({ node, showNodeNames = true, isSelected = false, isLoadingNeighbors = false }: NodeMarkerProps) {
  const getMarkerClass = () => {
-    let baseClass = "custom-node-marker";
-    
-    if (node.type === "meshtastic") {
-      baseClass += " custom-node-marker--green";
-    } else if (node.type === "meshcore") {
-      baseClass += " custom-node-marker--blue custom-node-marker--top";
-    }
-    
+    let baseClass = "custom-node-marker custom-node-marker--blue custom-node-marker--top";
+
    // Only add loading class when actually loading neighbors
    if (isLoadingNeighbors) {
      baseClass += " custom-node-marker--loading";
    }
-    
+
    return baseClass;
  };

@@ -43,7 +37,7 @@ export function NodeMarker({ node, showNodeNames = true, isSelected = false, isL
    <div className="custom-node-marker-container">
      {showNodeNames && node.short_name && (
        <div className="custom-node-label">
-          {node.type === "meshcore" ? getNameIconLabel(node.name || node.short_name) : node.short_name}
+          {getNameIconLabel(node.name || node.short_name)}
        </div>
      )}
      <div className={getMarkerClass()}></div>
@@ -51,29 +45,11 @@ export function NodeMarker({ node, showNodeNames = true, isSelected = false, isL
  );
 }

-// Cluster marker component with pie chart
+// Cluster marker component
 export function ClusterMarker({ children }: ClusterMarkerProps) {
-  let meshtasticCount = 0;
-  let meshcoreCount = 0;
-
  // Convert children to array if it's not already
  const childrenArray = Array.isArray(children) ? children : [children];
-
-  childrenArray.forEach((marker: any) => {
-    const node = marker.options && marker.options.nodeData;
-    if (node?.type === 'meshtastic') meshtasticCount++;
-    else if (node?.type === 'meshcore') meshcoreCount++;
-  });
-
-  const total = meshtasticCount + meshcoreCount;
-  const percentMeshcore = total ? meshcoreCount / total : 0;
-  const percentMeshtastic = total ? meshtasticCount / total : 0;
-
-  // Pie chart SVG calculations
-  const r = 13.5;
-  const c = 2 * Math.PI * r;
-  const meshcoreArc = percentMeshcore * c;
-  const meshtasticArc = percentMeshtastic * c;
+  const total = childrenArray.length;

  return (
    <div style={{
@@ -94,35 +70,11 @@ export function ClusterMarker({ children }: ClusterMarkerProps) {
          r="13.5"
          cx="15"
          cy="15"
-          fill="#fff"
+          fill="#2563eb"
          stroke="#fff"
          strokeWidth="3"
          opacity="0.7"
        />
-        <circle
-          r="13.5"
-          cx="15"
-          cy="15"
-          fill="transparent"
-          stroke="#2563eb"
-          strokeWidth="27"
-          strokeDasharray={`${meshcoreArc} ${c - meshcoreArc}`}
-          strokeDashoffset="0"
-          transform="rotate(-90 15 15)"
-          opacity="0.7"
-        />
-        <circle
-          r="13.5"
-          cx="15"
-          cy="15"
-          fill="transparent"
-          stroke="#22c55e"
-          strokeWidth="27"
-          strokeDasharray={`${meshtasticArc} ${c - meshtasticArc}`}
-          strokeDashoffset={`-${meshcoreArc}`}
-          transform="rotate(-90 15 15)"
-          opacity="0.7"
-        />
      </svg>
      <span style={{
        position: "absolute",
@@ -149,9 +101,9 @@ export function ClusterMarker({ children }: ClusterMarkerProps) {
 export function PopupContent({ node, target = '_self' }: PopupContentProps) {
  return (
    <div>
-      <div><b>ID:</b> {node.type === "meshcore" ? formatPublicKey(node.node_id) : node.node_id}</div>
+      <div><b>ID:</b> {formatPublicKey(node.node_id)}</div>
      <div><b>Full Name:</b> {node.name ?? "-"}</div>
-      <div><b>Short Name:</b> {node.type === "meshcore" && node.short_name ? getNameIconLabel(node.name || node.short_name) : (node.short_name ?? "-")}</div>
+      <div><b>Short Name:</b> {node.short_name ? getNameIconLabel(node.name || node.short_name) : "-"}</div>
      <div><b>Type:</b> {node.type ?? "-"}</div>
      <div><b>Lat:</b> {node.latitude}</div>
      <div><b>Lng:</b> {node.longitude}</div>
@@ -172,30 +124,26 @@ export function PopupContent({ node, target = '_self' }: PopupContentProps) {
      ) : (
        <div><b>First seen:</b> -</div>
      )}
-      {node.type === "meshcore" && (
-        <div style={{marginTop: '8px', paddingTop: '8px', borderTop: '1px solid #e5e7eb'}}>
-          <a 
-            href={`/meshcore/node/${node.node_id}`}
-            target={target}
-            style={{
-              display: 'inline-block',
-              padding: '4px 8px',
-              backgroundColor: '#3b82f6',
-              color: 'white',
-              textDecoration: 'none',
-              borderRadius: '4px',
-              fontSize: '12px',
-              fontWeight: '500'
-            }}
-            onMouseOver={(e) => e.currentTarget.style.backgroundColor = '#2563eb'}
-            onMouseOut={(e) => e.currentTarget.style.backgroundColor = '#3b82f6'}
-          >
-            View Node Details →
-          </a>
-        </div>
-      )}
+      <div style={{marginTop: '8px', paddingTop: '8px', borderTop: '1px solid #e5e7eb'}}>
+        <a
+          href={`/meshcore/node/${node.node_id}`}
+          target={target}
+          style={{
+            display: 'inline-block',
+            padding: '4px 8px',
+            backgroundColor: '#3b82f6',
+            color: 'white',
+            textDecoration: 'none',
+            borderRadius: '4px',
+            fontSize: '12px',
+            fontWeight: '500'
+          }}
+          onMouseOver={(e) => e.currentTarget.style.backgroundColor = '#2563eb'}
+          onMouseOut={(e) => e.currentTarget.style.backgroundColor = '#3b82f6'}
+        >
+          View Node Details →
+        </a>
+      </div>
    </div>
  );
 }
-
- 
@@ -78,11 +78,11 @@ export default function MapLayerSettingsComponent({ onSettingsChange }: MapLayer
            <label key={nodeType.key} className="flex items-center gap-2 mb-1 ml-6 cursor-pointer">
              <input
                type="checkbox"
-                checked={settings.nodeTypes.includes(nodeType.key as "meshcore" | "meshtastic")}
+                checked={settings.nodeTypes.includes(nodeType.key as "meshcore")}
                onChange={(e) => {
                  const currentTypes = settings.nodeTypes;
                  if (e.target.checked) {
-                    updateSetting('nodeTypes', [...currentTypes, nodeType.key as "meshcore" | "meshtastic"]);
+                    updateSetting('nodeTypes', [...currentTypes, nodeType.key as "meshcore"]);
                  } else {
                    updateSetting('nodeTypes', currentTypes.filter(t => t !== nodeType.key));
                  }
@@ -1,7 +1,7 @@
 "use client";
 import { useLocalStorage } from './useLocalStorage';

-type NodeType = "meshcore" | "meshtastic";
+type NodeType = "meshcore";

 export interface MapLayerSettings {
  showNodes: boolean;
@@ -41,5 +41,4 @@ export const TILE_LAYERS = [

 export const NODE_TYPE_OPTIONS = [
  { key: "meshcore", label: "Meshcore" },
-  { key: "meshtastic", label: "Meshtastic" },
 ];
--- a/Show More
+++ b/Show More