Files
meshcore-analyzer/cmd/ingestor
Marcel Verdult 498fbc0321 fix: ingestor uses ingest-time now() instead of observer receive time (#1233)
## Problem
The ingestor stamps every stored packet with its own ingest-time
`time.Now()`
(`BuildPacketData` in `db.go`; channel/DM paths in `main.go`),
discarding the
observer receive time the uploader already puts in the MQTT envelope's
`timestamp` field. `MQTTPacketMessage` had no `Timestamp` field and
`handleMessage` parsed every envelope field except that one.

Observers that buffer packets offline and upload hours later get every
buffered packet displayed at upload time, not receive time — a 5-hour
deferred upload shows packets 5 hours late. Retained messages and broker
backlog hit the same skew.

## Why the envelope timestamp is trustworthy
Uploaders stamp `timestamp` when the radio receives the frame and freeze
it;
the MQTT *message* is published late, but the `timestamp` *field* is not
re-stamped at publish. A buffered packet uploaded hours late still
carries
its true receive time.

## Fix
New `resolveRxTime` helper reads `msg["timestamp"]` and falls back to
`time.Now()` only when it is missing, unparseable, or implausibly in the
future. Applied to all three ingest paths (raw packet, channel, DM). No
wire-format change — the field already exists.

Channel/DM dedup hashes intentionally stay on ingest time, since those
bridge
messages carry no real packet hash and need ingest-unique input.

## Observer/node last_seen correction
Packet timestamps must reflect receive time, but observer/node
`last_seen`
must not. `InsertTransmission` fed `data.Timestamp` (now rxTime) into
`observers.last_seen` and `UpsertNode`'s `last_seen`, so a buffered
upload
could drag both fields backwards, and retained-message replay on MQTT
reconnect could flash long-offline observers as Online.

- `UpsertObserverAt` takes an explicit `lastSeen`; the status-packet and
BLE
companion handlers pass the resolved rxTime. `UpsertObserver` keeps its
  wall-clock behaviour for other callers.
- All three `last_seen` writes are guarded with
`MAX(MIN(existing, ingestNow), rxTime)`: `last_seen` never moves
backwards
  from a stale retained message, and never locks in a future value.

## Naive UTC+N timestamps
`resolveRxTime` rejects a timestamp only when it is >14h ahead (UTC+14
is the
maximum standard offset — anything further is a genuine clock error). A
timestamp that is merely in the future is soft-clamped to ingest time: a
future rxTime means a live packet from a UTC+N observer whose naive
local
clock parses as-if UTC, not a buffered packet, so ingest time is correct
and
no future timestamp reaches the DB.

For buffered packets from naive-clock uploaders a bounded residual
offset
remains (equal to the observer's UTC offset); uploaders emitting
zone-aware
ISO8601 everywhere would be the full cure but is a separate format
change.

## Test
`cmd/ingestor/rxtime_test.go` covers `parseEnvelopeTime` (zone-aware,
naive,
microseconds, garbage, empty) and `resolveRxTime` (plausible past used
verbatim, missing/garbage/future → ingest-time fallback). The existing
`TestBuildPacketData` is updated to supply an envelope timestamp and
assert it
propagates, since `BuildPacketData` no longer self-stamps.
2026-05-23 11:22:51 -07:00
..

MeshCore MQTT Ingestor (Go)

Standalone MQTT ingestion service for CoreScope. Connects to MQTT brokers, decodes raw MeshCore packets, and writes to the same SQLite database used by the Node.js web server.

This is the first step of a larger Go rewrite — separating MQTT ingestion from the web server.

Architecture

MQTT Broker(s)  →  Go Ingestor  →  SQLite DB  ←  Node.js Web Server
                    (this binary)     (shared)
  • Single static binary — no runtime dependencies, no CGO
  • SQLite via modernc.org/sqlite (pure Go)
  • MQTT via github.com/eclipse/paho.mqtt.golang
  • Runs alongside the Node.js server — they share the DB file
  • Does NOT serve HTTP/WebSocket — that stays in Node.js

Build

Requires Go 1.22+.

cd cmd/ingestor
go build -o corescope-ingestor .

Cross-compile for Linux (e.g., for the production VM):

GOOS=linux GOARCH=amd64 go build -o corescope-ingestor .

Run

./corescope-ingestor -config /path/to/config.json

The config file uses the same format as the Node.js config.json. The ingestor reads the mqttSources array (or legacy mqtt object) and dbPath fields.

Environment Variables

Variable Description Default
DB_PATH SQLite database path data/meshcore.db
MQTT_BROKER Single MQTT broker URL (overrides config)
MQTT_TOPIC MQTT topic (used with MQTT_BROKER) meshcore/#
CORESCOPE_INGESTOR_STATS Path to the per-second stats JSON file consumed by the server's /api/perf/io and /api/perf/write-sources endpoints (#1120) /tmp/corescope-ingestor-stats.json

Stats file (CORESCOPE_INGESTOR_STATS)

Every second the ingestor publishes a JSON snapshot of its counters (tx_inserted, obs_inserted, walCommits, backfillUpdates.*, etc.) plus a procIO block sampled from /proc/self/io (read/write/cancelled bytes per second + syscall counts). The server reads this file and surfaces the data on the Perf page so operators can self-diagnose write-volume anomalies.

The writer uses O_NOFOLLOW | O_CREAT | O_TRUNC mode 0o600, so a pre-planted symlink at the path cannot be used to clobber an arbitrary file.

Security note: the default lives in /tmp, which is world-writable on most hosts (sticky bit only protects deletion, not creation). On shared/multi-tenant hosts, override CORESCOPE_INGESTOR_STATS to point at a private directory (e.g. /var/lib/corescope/ingestor-stats.json) that only the corescope user can write to.

Minimal Config

{
  "dbPath": "data/meshcore.db",
  "mqttSources": [
    {
      "name": "local",
      "broker": "mqtt://localhost:1883",
      "topics": ["meshcore/#"]
    }
  ]
}

Full Config (same as Node.js)

The ingestor reads these fields from the existing config.json:

  • mqttSources[] — array of MQTT broker connections
    • name — display name for logging
    • broker — MQTT URL (mqtt://, mqtts://)
    • username / password — auth credentials
    • topics — array of topic patterns to subscribe
    • iataFilter — optional regional filter
  • mqtt — legacy single-broker config (auto-converted to mqttSources)
  • dbPath — SQLite DB path (default: data/meshcore.db)

Test

cd cmd/ingestor
go test -v ./...

What It Does

  1. Connects to configured MQTT brokers with auto-reconnect
  2. Subscribes to mesh packet topics (e.g., meshcore/+/+/packets)
  3. Receives raw hex packets via JSON messages ({ "raw": "...", "SNR": ..., "RSSI": ... })
  4. Decodes MeshCore packet headers, paths, and payloads (ported from decoder.js)
  5. Computes content hashes (path-independent, SHA-256-based)
  6. Writes to SQLite: transmissions + observations tables
  7. Upserts nodes from decoded ADVERT packets (with validation)
  8. Upserts observers from MQTT topic metadata

Schema Compatibility

The Go ingestor creates the same v3 schema as the Node.js server:

  • transmissions — deduplicated by content hash
  • observations — per-observer sightings with observer_idx (rowid reference)
  • nodes — mesh nodes discovered from adverts
  • observers — MQTT feed sources

Both processes can write to the same DB concurrently (SQLite WAL mode).

What's Not Ported (Yet)

  • Companion bridge format (Format 2 — meshcore/advertisement, channel messages, etc.)
  • Channel key decryption (GRP_TXT encrypted payload decryption)
  • WebSocket broadcast to browsers
  • In-memory packet store
  • Cache invalidation

These stay in the Node.js server for now.

Files

cmd/ingestor/
  main.go          — entry point, MQTT connect, message handler
  decoder.go       — MeshCore packet decoder (ported from decoder.js)
  decoder_test.go  — decoder tests (25 tests, golden fixtures)
  db.go            — SQLite writer (schema-compatible with db.js)
  db_test.go       — DB tests (schema validation, insert/upsert, E2E)
  config.go        — config struct + loader
  util.go          — shared utilities
  go.mod / go.sum  — Go module definition