diff --git a/cmd/ingestor/client_uplink.go b/cmd/ingestor/client_uplink.go new file mode 100644 index 000000000..2680f2019 --- /dev/null +++ b/cmd/ingestor/client_uplink.go @@ -0,0 +1,221 @@ +package main + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "log" + "regexp" + "strings" + "time" +) + +// Uplink pings (CoreDrive RX v1.21.0+, docs/client-rx-coverage.md): a mobile +// companion sends a channel message on a dedicated hashtag channel carrying +// only "cdrx ", listens for repeaters re-broadcasting it, and reports the +// ping on meshcore/client/{PUBLIC_KEY}/uplink. Two facts come out of it: +// - echo: the companion heard a repeater re-broadcast the ping, so a repeater +// heard the companion (reported in the payload) +// - arrival: an observer received the ping, found through its packet hash +// +// The hash comes from the payload when there was an echo. Without one, the +// ingestor learns it by decrypting the ping itself when an observer reports +// it (noteUplinkTransmission), keyed by the tag in the text. + +var ( + uplinkTagRe = regexp.MustCompile(`^[0-9a-f]{8}$`) + uplinkHashRe = regexp.MustCompile(`^[0-9a-f]{16}$`) + uplinkHopRe = regexp.MustCompile(`^[0-9a-f]{2,8}$`) + uplinkTextTagRe = regexp.MustCompile(`(?:^|: )cdrx ([0-9a-f]{8})$`) +) + +// maxUplinkEchoes bounds the stored echo list: one ping is re-broadcast by a +// handful of repeaters in a 10 s window, never by hundreds. +const maxUplinkEchoes = 32 + +// UplinkEcho is one re-broadcast of the ping the companion heard. +type UplinkEcho struct { + FirstHop string `json:"first_hop"` + LastHop string `json:"last_hop"` + Hops int `json:"hops"` + SNR *float64 `json:"snr,omitempty"` + RSSI *float64 `json:"rssi,omitempty"` +} + +// ClientUplinkPing is one stored ping. +type ClientUplinkPing struct { + RxPubkey string + Tag string + SentAt string + IngestedAt string + Channel string + Scope string + PktHash string // empty when no echo was heard + Echoes []UplinkEcho + Lat, Lon *float64 + PosAccM *float64 +} + +// firstHops returns the distinct first hops of the echoes, in order heard. +func (p *ClientUplinkPing) firstHops() string { + seen := map[string]bool{} + var out []string + for _, e := range p.Echoes { + if !seen[e.FirstHop] { + seen[e.FirstHop] = true + out = append(out, e.FirstHop) + } + } + return strings.Join(out, ",") +} + +func handleClientUplink(store *Store, tag, rxPubkey string, msg map[string]interface{}) { + rxPubkey = strings.ToLower(strings.TrimSpace(rxPubkey)) + if !clientPubkeyRe.MatchString(rxPubkey) { + log.Printf("MQTT [%s] uplink: invalid pubkey %.8q, dropping", tag, rxPubkey) + return + } + pingTag, _ := msg["tag"].(string) + pingTag = strings.ToLower(strings.TrimSpace(pingTag)) + if !uplinkTagRe.MatchString(pingTag) { + log.Printf("MQTT [%s] uplink: invalid tag %.12q, dropping", tag, pingTag) + return + } + p := &ClientUplinkPing{RxPubkey: rxPubkey, Tag: pingTag, IngestedAt: time.Now().UTC().Format(time.RFC3339)} + rxTime, _ := resolveRxTimeCore(msg, tag) + p.SentAt = rxTime.Format(rxTimeMillisLayout) + p.Channel = clampString(msg["channel"], 32) + p.Scope = clampString(msg["scope"], 32) + if h, _ := msg["pkt_hash"].(string); uplinkHashRe.MatchString(strings.ToLower(h)) { + p.PktHash = strings.ToLower(h) + } + if raw, ok := msg["echoes"].([]interface{}); ok { + for _, r := range raw { + if len(p.Echoes) >= maxUplinkEchoes { + break + } + m, ok := r.(map[string]interface{}) + if !ok { + continue + } + first, _ := m["first_hop"].(string) + last, _ := m["last_hop"].(string) + first, last = strings.ToLower(first), strings.ToLower(last) + if !uplinkHopRe.MatchString(first) || !uplinkHopRe.MatchString(last) { + continue + } + e := UplinkEcho{FirstHop: first, LastHop: last} + if h, ok := toFloat64(m["hops"]); ok && h >= 1 && h <= 64 { + e.Hops = int(h) + } + if v, ok := toFloat64(m["snr"]); ok { + e.SNR = &v + } + if v, ok := toFloat64(m["rssi"]); ok { + e.RSSI = &v + } + p.Echoes = append(p.Echoes, e) + } + } + if gps, ok := msg["gps"].(map[string]interface{}); ok { + lat, latOK := toFloat64(gps["lat"]) + lon, lonOK := toFloat64(gps["lon"]) + if latOK && lonOK && lat >= -90 && lat <= 90 && lon >= -180 && lon <= 180 { + p.Lat, p.Lon = &lat, &lon + if acc, ok := toFloat64(gps["acc_m"]); ok { + p.PosAccM = &acc + } + } + } + if _, err := store.InsertClientUplinkPing(p); err != nil { + log.Printf("MQTT [%s] client_uplink_pings insert: %v", tag, err) + } +} + +func clampString(v interface{}, max int) string { + s, _ := v.(string) + s = strings.TrimSpace(s) + if len(s) > max { + s = s[:max] + } + return s +} + +// InsertClientUplinkPing stores one ping. Idempotent on (rx_pubkey, tag): a +// redelivered message is stored once. +func (s *Store) InsertClientUplinkPing(p *ClientUplinkPing) (bool, error) { + echoes, err := json.Marshal(p.Echoes) + if err != nil { + return false, err + } + if p.Echoes == nil { + echoes = []byte("[]") + } + var hash interface{} + if p.PktHash != "" { + hash = p.PktHash + } + res, err := s.db.Exec(` + INSERT INTO client_uplink_pings + (rx_pubkey, tag, sent_at, ingested_at, channel, scope, pkt_hash, echo_count, first_hops, echoes_json, lat, lon, pos_acc_m) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?) + ON CONFLICT(rx_pubkey, tag) DO NOTHING`, + p.RxPubkey, p.Tag, p.SentAt, p.IngestedAt, p.Channel, p.Scope, hash, len(p.Echoes), p.firstHops(), string(echoes), p.Lat, p.Lon, p.PosAccM) + if err != nil { + return false, err + } + n, _ := res.RowsAffected() + return n > 0, nil +} + +// uplinkChannel is the ping channel the ingestor decrypts to learn tag → hash. +type uplinkChannel struct { + keyHex string + hashByte int +} + +// newUplinkChannel derives the hashtag channel's key and hash byte the way the +// firmware does: key = SHA-256("#name")[:16], hash = SHA-256(key)[0] +// (helpers/BaseChatMesh.cpp:908). nil for an empty name. +func newUplinkChannel(name string) *uplinkChannel { + name = strings.TrimSpace(name) + if name == "" { + return nil + } + k := sha256.Sum256([]byte(name)) + key := k[:16] + h := sha256.Sum256(key) + return &uplinkChannel{keyHex: hex.EncodeToString(key), hashByte: int(h[0])} +} + +// noteUplinkTransmission links a ping's tag to the packet hash an observer +// reported. Only GRP_TXT the ingestor could not decrypt with its own channel +// keys reaches here, on the ping channel's hash byte; that byte is shared with +// unrelated channels, so the MAC check in decryptChannelMessage is what tells +// a ping from them. +func noteUplinkTransmission(store *Store, ch *uplinkChannel, decoded *DecodedPacket, pktHash string) { + if ch == nil || decoded == nil || pktHash == "" { + return + } + pl := decoded.Payload + if pl.Type != "GRP_TXT" || pl.ChannelHash != ch.hashByte || pl.EncryptedData == "" || pl.MAC == "" { + return + } + res, err := decryptChannelMessage(pl.EncryptedData, pl.MAC, ch.keyHex) + if err != nil { + return + } + text := res.Message + if res.Sender != "" { + text = res.Sender + ": " + res.Message + } + m := uplinkTextTagRe.FindStringSubmatch(text) + if m == nil { + return + } + if _, err := store.db.Exec(` + INSERT INTO client_uplink_tx (tag, pkt_hash, sender, seen_at) VALUES (?,?,?,?) + ON CONFLICT(tag) DO NOTHING`, m[1], pktHash, res.Sender, time.Now().UTC().Format(time.RFC3339)); err != nil { + log.Printf("[uplink] client_uplink_tx insert: %v", err) + } +} diff --git a/cmd/ingestor/client_uplink_test.go b/cmd/ingestor/client_uplink_test.go new file mode 100644 index 000000000..e168ef32e --- /dev/null +++ b/cmd/ingestor/client_uplink_test.go @@ -0,0 +1,201 @@ +package main + +import ( + "crypto/aes" + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "testing" + "time" +) + +func uplinkCount(t *testing.T, s *Store, table string) int { + t.Helper() + var n int + if err := s.db.QueryRow(`SELECT COUNT(*) FROM ` + table).Scan(&n); err != nil { + t.Fatal(err) + } + return n +} + +// uplinkMsg is a /uplink payload as CoreDrive RX v1.21.0 sends it +// (src/publisher.js buildUplinkPayload), an hour in the past so +// resolveRxTimeCore keeps the timestamp. +func uplinkMsg() map[string]interface{} { + return map[string]interface{}{ + "type": "UPLINK_PING", "timestamp": time.Now().UTC().Add(-time.Hour).Format(rxTimeMillisLayout), + "tag": "78277c0e", "channel": "#coredrive", "scope": "#be", "pkt_hash": "b76c6a2777f9b0e9", + "echoes": []interface{}{ + map[string]interface{}{"first_hop": "1403", "last_hop": "1403", "hops": float64(1), "snr": 12.25, "rssi": float64(-100)}, + map[string]interface{}{"first_hop": "c0cd", "last_hop": "c0cd", "hops": float64(1), "snr": -2.25, "rssi": float64(-120)}, + map[string]interface{}{"first_hop": "1403", "last_hop": "e470", "hops": float64(2), "snr": -8.25, "rssi": float64(-126)}, + }, + "gps": map[string]interface{}{"lat": 50.998, "lon": 5.384, "acc_m": 6.0}, + } +} + +func TestHandleClientUplinkWritesRow(t *testing.T) { + s := newTestStore(t) + handleClientUplink(s, "test", testCompanionPK, uplinkMsg()) + var tag, hash, firstHops, echoes string + var count int + var lat float64 + err := s.db.QueryRow(`SELECT tag, pkt_hash, echo_count, first_hops, echoes_json, lat FROM client_uplink_pings`).Scan(&tag, &hash, &count, &firstHops, &echoes, &lat) + if err != nil { + t.Fatal(err) + } + if tag != "78277c0e" || hash != "b76c6a2777f9b0e9" || count != 3 || lat != 50.998 { + t.Fatalf("row = %s %s %d %v", tag, hash, count, lat) + } + if firstHops != "1403,c0cd" { + t.Errorf("first_hops = %q, want the distinct first hops 1403,c0cd", firstHops) + } +} + +func TestHandleClientUplinkNoEchoNoHash(t *testing.T) { + s := newTestStore(t) + m := uplinkMsg() + delete(m, "pkt_hash") + m["echoes"] = []interface{}{} + handleClientUplink(s, "test", testCompanionPK, m) + var hash *string + var count int + if err := s.db.QueryRow(`SELECT pkt_hash, echo_count FROM client_uplink_pings`).Scan(&hash, &count); err != nil { + t.Fatal(err) + } + if hash != nil || count != 0 { + t.Fatalf("pkt_hash = %v, echo_count = %d; want NULL and 0", hash, count) + } +} + +func TestHandleClientUplinkRejectsAndSanitises(t *testing.T) { + s := newTestStore(t) + bad := uplinkMsg() + bad["tag"] = "not-a-tag" + handleClientUplink(s, "test", testCompanionPK, bad) + handleClientUplink(s, "test", "NOT-HEX!", uplinkMsg()) + if n := uplinkCount(t, s, "client_uplink_pings"); n != 0 { + t.Fatalf("rows = %d, want 0", n) + } + m := uplinkMsg() + m["pkt_hash"] = "xyz" + m["echoes"] = []interface{}{map[string]interface{}{"first_hop": "zz", "last_hop": "1403"}, "junk", + map[string]interface{}{"first_hop": "1403", "last_hop": "1403", "hops": float64(1)}} + handleClientUplink(s, "test", testCompanionPK, m) + var hash *string + var count int + if err := s.db.QueryRow(`SELECT pkt_hash, echo_count FROM client_uplink_pings`).Scan(&hash, &count); err != nil { + t.Fatal(err) + } + if hash != nil || count != 1 { + t.Fatalf("pkt_hash = %v, echo_count = %d; want a dropped bad hash and only the valid echo", hash, count) + } +} + +func TestHandleClientUplinkIdempotent(t *testing.T) { + s := newTestStore(t) + m := uplinkMsg() + handleClientUplink(s, "test", testCompanionPK, m) + handleClientUplink(s, "test", testCompanionPK, m) + if n := uplinkCount(t, s, "client_uplink_pings"); n != 1 { + t.Fatalf("rows = %d, want 1", n) + } +} + +// grpTxtPayload encrypts text the way the firmware does (Mesh.cpp:540-557, +// Utils.cpp:127-145) and returns it decoded as the ingestor would see it +// without the key. +func grpTxtPayload(t *testing.T, channel, text string) *DecodedPacket { + t.Helper() + ch := newUplinkChannel(channel) + key, _ := hex.DecodeString(ch.keyHex) + plain := append([]byte{1, 2, 3, 4, 0}, []byte(text)...) + for len(plain)%16 != 0 { + plain = append(plain, 0) + } + block, _ := aes.NewCipher(key) + ct := make([]byte, len(plain)) + for i := 0; i < len(plain); i += 16 { + block.Encrypt(ct[i:i+16], plain[i:i+16]) + } + secret := make([]byte, 32) + copy(secret, key) + h := hmac.New(sha256.New, secret) + h.Write(ct) + mac := h.Sum(nil)[:2] + buf := append([]byte{byte(ch.hashByte)}, mac...) + buf = append(buf, ct...) + pl := decodeGrpTxt(buf, nil) + return &DecodedPacket{Payload: pl} +} + +func TestNewUplinkChannelMatchesFirmware(t *testing.T) { + // #coredrive hashes to channel byte 0x76, as seen on air (CoreScope live, + // transmission b76c6a2777f9b0e9). + if ch := newUplinkChannel("#coredrive"); ch.hashByte != 0x76 { + t.Fatalf("hash byte = %#x, want 0x76", ch.hashByte) + } + if newUplinkChannel(" ") != nil { + t.Fatal("empty name must give no channel") + } +} + +func TestNoteUplinkTransmissionLinksTagToHash(t *testing.T) { + s := newTestStore(t) + ch := newUplinkChannel("#coredrive") + noteUplinkTransmission(s, ch, grpTxtPayload(t, "#coredrive", "BE-BRE-ON8AR | Erwin: cdrx 78277c0e"), "b76c6a2777f9b0e9") + var hash, sender string + if err := s.db.QueryRow(`SELECT pkt_hash, sender FROM client_uplink_tx WHERE tag = '78277c0e'`).Scan(&hash, &sender); err != nil { + t.Fatal(err) + } + if hash != "b76c6a2777f9b0e9" || sender != "BE-BRE-ON8AR | Erwin" { + t.Fatalf("hash %s sender %q", hash, sender) + } + // Other text on the channel, another channel's packet, or no channel set: nothing. + noteUplinkTransmission(s, ch, grpTxtPayload(t, "#coredrive", "someone: hello"), "1111111111111111") + noteUplinkTransmission(s, ch, grpTxtPayload(t, "#elsewhere", "x: cdrx aaaaaaaa"), "2222222222222222") + noteUplinkTransmission(s, nil, grpTxtPayload(t, "#coredrive", "x: cdrx bbbbbbbb"), "3333333333333333") + if n := uplinkCount(t, s, "client_uplink_tx"); n != 1 { + t.Fatalf("client_uplink_tx rows = %d, want 1", n) + } +} + +func TestClientUplinkConfig(t *testing.T) { + if (&Config{}).ClientUplinkEnabled() || (&Config{}).UplinkChannel() != nil { + t.Fatal("absent config must be off") + } + c := &Config{ClientUplink: &ClientUplinkConfig{Enabled: true}} + if ch := c.UplinkChannel(); ch == nil || ch.hashByte != 0x76 { + t.Fatalf("default channel should be #coredrive, got %+v", ch) + } +} + +func TestClientUplinkGate(t *testing.T) { + payload := []byte(`{"type":"UPLINK_PING","timestamp":"` + time.Now().UTC().Add(-time.Hour).Format(rxTimeMillisLayout) + `","tag":"0a1b2c3d","echoes":[]}`) + msg := &mockMessage{topic: "meshcore/client/" + testCompanionPK + "/uplink", payload: payload} + off := newTestStore(t) + handleMessage(off, "test", MQTTSource{Name: "test"}, msg, nil, nil, &Config{}) + if n := uplinkCount(t, off, "client_uplink_pings"); n != 0 { + t.Fatalf("feature off: rows = %d", n) + } + on := newTestStore(t) + handleMessage(on, "test", MQTTSource{Name: "test"}, msg, nil, nil, &Config{ClientUplink: &ClientUplinkConfig{Enabled: true}}) + if n := uplinkCount(t, on, "client_uplink_pings"); n != 1 { + t.Fatalf("feature on: rows = %d, want 1", n) + } +} + +func TestPruneOldClientUplink(t *testing.T) { + s := newTestStore(t) + handleClientUplink(s, "test", testCompanionPK, uplinkMsg()) + if _, err := s.db.Exec(`INSERT INTO client_uplink_tx (tag, pkt_hash, seen_at) VALUES ('aaaaaaaa','x','2026-01-01T00:00:00Z')`); err != nil { + t.Fatal(err) + } + n, err := s.pruneOldClientUplinkAt(time.Now().UTC().Add(-24 * time.Hour)) + if err != nil { + t.Fatal(err) + } + if n != 1 || uplinkCount(t, s, "client_uplink_pings") != 1 || uplinkCount(t, s, "client_uplink_tx") != 0 { + t.Fatalf("pruned %d; the recent ping must stay, the old tag link go", n) + } +} diff --git a/cmd/ingestor/config.go b/cmd/ingestor/config.go index f1c0409ec..62fb06817 100644 --- a/cmd/ingestor/config.go +++ b/cmd/ingestor/config.go @@ -65,6 +65,7 @@ type Config struct { ClientRfSamples *ClientRfSamplesConfig `json:"clientRfSamples,omitempty"` ClientRegions *ClientRegionsConfig `json:"clientRegions,omitempty"` ClientNodeNames *ClientNodeNamesConfig `json:"clientNodeNames,omitempty"` + ClientUplink *ClientUplinkConfig `json:"clientUplink,omitempty"` GeoFilter *GeoFilterConfig `json:"geo_filter,omitempty"` // PathTrust configures the minimum path-hash prefix length trusted as // mapping/topology evidence (issue #1784). @@ -236,6 +237,39 @@ func (c *Config) ClientNodeNamesEnabled() bool { return c.ClientNodeNames != nil && c.ClientNodeNames.Enabled } +// ClientUplinkConfig controls the opt-in uplink ping stream +// (meshcore/client//uplink). Channel is the hashtag channel the pings +// are sent on; the ingestor decrypts it to link a ping to its packet hash when +// the companion heard no echo. Default "#coredrive". +type ClientUplinkConfig struct { + Enabled bool `json:"enabled"` + Channel string `json:"channel,omitempty"` + + once sync.Once + ch *uplinkChannel +} + +// ClientUplinkEnabled reports whether uplink pings are recorded. Default false. +func (c *Config) ClientUplinkEnabled() bool { + return c.ClientUplink != nil && c.ClientUplink.Enabled +} + +// UplinkChannel returns the ping channel's derived key, computed once. +func (c *Config) UplinkChannel() *uplinkChannel { + if !c.ClientUplinkEnabled() { + return nil + } + u := c.ClientUplink + u.once.Do(func() { + name := u.Channel + if strings.TrimSpace(name) == "" { + name = "#coredrive" + } + u.ch = newUplinkChannel(name) + }) + return u.ch +} + // AutoRegionKeysConfig controls the opt-in derivation of region keys from the // names repeaters declare over RF (node_declared_regions), on top of the // explicit hashRegions list. @@ -325,6 +359,9 @@ type RetentionConfig struct { // ClientNodeNamesDays is the retention window (by observed_at) for the // client_node_names table; 0 disables. ClientNodeNamesDays int `json:"clientNodeNamesDays"` + // ClientUplinkDays is the retention window (by sent_at / seen_at) for + // client_uplink_pings and client_uplink_tx; 0 disables. + ClientUplinkDays int `json:"clientUplinkDays"` } // PacketDaysOrZero returns the configured retention.packetDays or 0 @@ -379,6 +416,14 @@ func (c *Config) ClientRegionsDaysOrZero() int { return 0 } +// ClientUplinkDaysOrZero returns retention.clientUplinkDays or 0 (disabled). +func (c *Config) ClientUplinkDaysOrZero() int { + if c.Retention != nil && c.Retention.ClientUplinkDays > 0 { + return c.Retention.ClientUplinkDays + } + return 0 +} + // ClientNodeNamesDaysOrZero returns retention.clientNodeNamesDays or 0 // (disabled) if not set. func (c *Config) ClientNodeNamesDaysOrZero() int { diff --git a/cmd/ingestor/db.go b/cmd/ingestor/db.go index 7e896b131..9b2e1a13c 100644 --- a/cmd/ingestor/db.go +++ b/cmd/ingestor/db.go @@ -494,6 +494,38 @@ func applySchema(db *sql.DB) error { ); CREATE INDEX IF NOT EXISTS idx_cnn_target ON client_node_names(target, observed_at); CREATE INDEX IF NOT EXISTS idx_cnn_prune ON client_node_names(observed_at); + + -- Uplink pings (cmd/ingestor/client_uplink.go): one row per ping a mobile + -- companion sent. pkt_hash is set when the companion heard an echo; + -- client_uplink_tx fills it in for pings without one, from the ingestor's + -- own decryption of the ping channel. The server joins both against + -- transmissions/observations to see whether an observer received it. + CREATE TABLE IF NOT EXISTS client_uplink_pings ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + rx_pubkey TEXT NOT NULL, + tag TEXT NOT NULL, + sent_at TEXT NOT NULL, + ingested_at TEXT NOT NULL, + channel TEXT NOT NULL DEFAULT '', + scope TEXT NOT NULL DEFAULT '', + pkt_hash TEXT, + echo_count INTEGER NOT NULL DEFAULT 0, + first_hops TEXT NOT NULL DEFAULT '', + echoes_json TEXT NOT NULL DEFAULT '[]', + lat REAL, + lon REAL, + pos_acc_m REAL, + UNIQUE(rx_pubkey, tag) + ); + CREATE INDEX IF NOT EXISTS idx_cup_latlon ON client_uplink_pings(lat, lon); + CREATE INDEX IF NOT EXISTS idx_cup_sent ON client_uplink_pings(sent_at); + CREATE TABLE IF NOT EXISTS client_uplink_tx ( + tag TEXT PRIMARY KEY, + pkt_hash TEXT NOT NULL, + sender TEXT NOT NULL DEFAULT '', + seen_at TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_cut_seen ON client_uplink_tx(seen_at); ` if _, err := db.Exec(schema); err != nil { return fmt.Errorf("base schema: %w", err) diff --git a/cmd/ingestor/main.go b/cmd/ingestor/main.go index d84812066..d34e98857 100644 --- a/cmd/ingestor/main.go +++ b/cmd/ingestor/main.go @@ -362,6 +362,16 @@ func main() { } } + // Uplink-ping retention: bounds client_uplink_pings and client_uplink_tx. + clientUplinkDays := cfg.ClientUplinkDaysOrZero() + if clientUplinkDays > 0 { + if n, err := store.PruneOldClientUplink(clientUplinkDays); err != nil { + log.Printf("[prune] client_uplink: %v", err) + } else if n > 0 { + log.Printf("[prune] startup pruned %d client_uplink rows older than %d days", n, clientUplinkDays) + } + } + // Node-name retention: bounds the opt-in client_node_names table. // 0 = disabled. clientNodeNamesDays := cfg.ClientNodeNamesDaysOrZero() @@ -455,7 +465,7 @@ func main() { // node_declared_regions retention (Task 6) rather than starting a second // ticker — the four flags are independent (0 disables each separately), // so the ticker itself must run when any is set. - if clientRxDays > 0 || clientRxObsDays > 0 || clientRfDays > 0 || clientRegionsDays > 0 || clientNodeNamesDays > 0 { + if clientRxDays > 0 || clientRxObsDays > 0 || clientRfDays > 0 || clientRegionsDays > 0 || clientNodeNamesDays > 0 || clientUplinkDays > 0 { clientRxRetentionTicker := time.NewTicker(24 * time.Hour) go func() { for range clientRxRetentionTicker.C { @@ -487,6 +497,13 @@ func main() { store.RunIncrementalVacuum(vacuumPages) } } + if clientUplinkDays > 0 { + if n, err := store.PruneOldClientUplink(clientUplinkDays); err != nil { + log.Printf("[prune] client_uplink: %v", err) + } else if n > 0 { + store.RunIncrementalVacuum(vacuumPages) + } + } if clientNodeNamesDays > 0 { if n, err := store.PruneOldClientNodeNames(clientNodeNamesDays); err != nil { log.Printf("[prune] client_node_names: %v", err) @@ -884,7 +901,8 @@ func handleMessage(store *Store, tag string, source MQTTSource, m mqtt.Message, // Mobile client topics: meshcore/client/{PUBLIC_KEY}/packets (RX coverage), // meshcore/client/{PUBLIC_KEY}/rf (RF environment samples), and // meshcore/client/{PUBLIC_KEY}/regions (declared-region answers), and - // meshcore/client/{PUBLIC_KEY}/names (a repeater's own name). A + // meshcore/client/{PUBLIC_KEY}/names (a repeater's own name), and + // meshcore/client/{PUBLIC_KEY}/uplink (uplink pings). A // roaming companion reports where it directly heard a node, its own // radio's counters, or a repeater's declared region list; all three are // handled in isolation from the observer/observations path. EMQX ACL @@ -929,6 +947,10 @@ func handleMessage(store *Store, tag string, source MQTTSource, m mqtt.Message, if cfg.ClientNodeNamesEnabled() { handleClientNodeNames(store, tag, parts[2], msg) } + case "uplink": + if cfg.ClientUplinkEnabled() { + handleClientUplink(store, tag, parts[2], msg) + } } return } @@ -1219,6 +1241,8 @@ func handleMessage(store *Store, tag string, source MQTTSource, m mqtt.Message, if _, err := store.InsertTransmission(pktData); err != nil { log.Printf("MQTT [%s] db insert error: %v", tag, err) } + // An uplink ping an observer received: learn its tag → hash. + noteUplinkTransmission(store, cfg.UplinkChannel(), decoded, pktData.Hash) } // Upsert observer diff --git a/cmd/ingestor/maintenance.go b/cmd/ingestor/maintenance.go index 376bb4490..54b143a0c 100644 --- a/cmd/ingestor/maintenance.go +++ b/cmd/ingestor/maintenance.go @@ -437,3 +437,26 @@ func (s *Store) pruneOldClientNodeNamesAt(cutoffInstant time.Time) (int64, error n, _ := res.RowsAffected() return n, nil } + +// PruneOldClientUplink deletes uplink pings (by sent_at) and tag links (by +// seen_at) older than `days`. 0 disables. +func (s *Store) PruneOldClientUplink(days int) (int64, error) { + if days <= 0 { + return 0, nil + } + return s.pruneOldClientUplinkAt(time.Now().UTC().AddDate(0, 0, -days)) +} + +func (s *Store) pruneOldClientUplinkAt(cutoffInstant time.Time) (int64, error) { + res, err := s.db.Exec(`DELETE FROM client_uplink_pings WHERE sent_at < ?`, cutoffInstant.Format(rxTimeMillisLayout)) + if err != nil { + return 0, fmt.Errorf("prune client_uplink_pings: %w", err) + } + n, _ := res.RowsAffected() + res, err = s.db.Exec(`DELETE FROM client_uplink_tx WHERE seen_at < ?`, cutoffInstant.Format(time.RFC3339)) + if err != nil { + return n, fmt.Errorf("prune client_uplink_tx: %w", err) + } + m, _ := res.RowsAffected() + return n + m, nil +} diff --git a/config.example.json b/config.example.json index ddc6abff2..55b829ceb 100644 --- a/config.example.json +++ b/config.example.json @@ -17,7 +17,8 @@ "clientRfDays": 30, "clientRegionsDays": 30, "clientNodeNamesDays": 0, - "_comment": "nodeDays: nodes not seen in N days moved to inactive_nodes (default 7). observerDays: observers not sending data in N days are removed (-1 = keep forever, default 14). observerPurgeDays: observers already marked inactive AND not seen in N days are hard-deleted, but only when no observation, observer_metrics or dropped_packets row still references them (0 = disabled, the default). Set it above both observerDays and packetDays — below those the reference guards keep every row anyway. packetDays: transmissions older than N days are deleted (0 = disabled). clientRxDays: mobile client-RX coverage rows (client_receptions/client_observers) older than N days are deleted (0 = disabled) — bounds the opt-in coverage tables (#1727). clientRxObsDays: diagnostic client_rx_observations rows (by rx_at) older than N days are deleted (0 = disabled) — bounds the opt-in clientRxObservations table. clientRfDays: RF environment sample rows (client_rf_samples, by sampled_at) older than N days are deleted (0 = disabled) — bounds the opt-in clientRfSamples table; unset means unbounded growth at roughly 240 rows/hour per driver. clientRegionsDays: region-discovery answer rows (node_declared_regions, by observed_at) older than N days are deleted (0 = disabled) — bounds the opt-in clientRegions table. clientNodeNamesDays: repeater-reported node names (client_node_names, by observed_at) older than N days are deleted (0 = disabled, keep them): one row per answer, and a pruned name is asked for again by the next companion that hears the repeater. NOTE (#1283): all retention fields are consumed by the INGESTOR process. The server is read-only and never prunes." + "clientUplinkDays": 90, + "_comment": "nodeDays: nodes not seen in N days moved to inactive_nodes (default 7). observerDays: observers not sending data in N days are removed (-1 = keep forever, default 14). observerPurgeDays: observers already marked inactive AND not seen in N days are hard-deleted, but only when no observation, observer_metrics or dropped_packets row still references them (0 = disabled, the default). Set it above both observerDays and packetDays — below those the reference guards keep every row anyway. packetDays: transmissions older than N days are deleted (0 = disabled). clientRxDays: mobile client-RX coverage rows (client_receptions/client_observers) older than N days are deleted (0 = disabled) — bounds the opt-in coverage tables (#1727). clientRxObsDays: diagnostic client_rx_observations rows (by rx_at) older than N days are deleted (0 = disabled) — bounds the opt-in clientRxObservations table. clientRfDays: RF environment sample rows (client_rf_samples, by sampled_at) older than N days are deleted (0 = disabled) — bounds the opt-in clientRfSamples table; unset means unbounded growth at roughly 240 rows/hour per driver. clientRegionsDays: region-discovery answer rows (node_declared_regions, by observed_at) older than N days are deleted (0 = disabled) — bounds the opt-in clientRegions table. clientNodeNamesDays: repeater-reported node names (client_node_names, by observed_at) older than N days are deleted (0 = disabled, keep them): one row per answer, and a pruned name is asked for again by the next companion that hears the repeater. clientUplinkDays: uplink pings (client_uplink_pings by sent_at, client_uplink_tx by seen_at) older than N days are deleted (0 = disabled). NOTE (#1283): all retention fields are consumed by the INGESTOR process. The server is read-only and never prunes." }, "db": { "vacuumOnStartup": false, @@ -391,6 +392,8 @@ "_comment_compression": "Opt-in HTTP gzip middleware + WebSocket permessage-deflate. Both default to false — enable ONLY when your upstream reverse proxy is NOT already compressing. gzip: enables the gzipMiddleware wrapper around the HTTP handler. websocket: sets gorilla websocket Upgrader.EnableCompression. level: gzip compression level 1..9 (1=BestSpeed, 9=BestCompression, default 6). minSizeBytes: advisory minimum response size below which compression would not pay off. contentTypes: MIME allow-list — only responses with these Content-Type values are compressed. Already-compressed types (image/*, video/*, audio/*, application/zip, application/x-gzip, application/pdf, application/octet-stream) are always skipped, as are responses whose handler already set Content-Encoding. Omit contentTypes to use the built-in default allow-list.", "clientRegions": { "enabled": false }, "clientNodeNames": { "enabled": false }, + "clientUplink": { "enabled": false, "channel": "#coredrive" }, + "_comment_clientUplink": "Opt-in uplink pings (CoreDrive RX v1.21.0+): a mobile companion sends cdrx on the hashtag channel below and reports the ping, with any echoes it heard, on meshcore/client//uplink, stored in client_uplink_pings. The ingestor decrypts that channel itself to link a ping to its packet hash when no echo was heard (client_uplink_tx); the channel does not need to be in hashChannels. Default OFF. TOP-LEVEL FLAG. Requires the broker ACL to allow /uplink. Retention: retention.clientUplinkDays.", "_comment_clientNodeNames": "Opt-in repeater names: when this instance has no name for a repeater a mobile companion hears directly, the companion asks the repeater for its own name (ANON_REQ_TYPE_OWNER) and relays it on meshcore/client//names, stored in client_node_names. /api/nodes/resolve and the coverage tooltips fall back to it when nodes has no name; an advert name always wins. Default OFF. TOP-LEVEL FLAG, same as clientRegions. GPS is optional. Requires the broker ACL to allow meshcore/client/{pubkey}/names for that publisher (see docs/client-rx-coverage.md). Retention: see retention.clientNodeNamesDays.", "_comment_clientRegions": "Opt-in region-discovery answers: a mobile companion asks each repeater in direct RF range which regions it declares flood-allowed, and relays the answer on meshcore/client//regions, stored in node_declared_regions — the declared side of a comparison against the OBSERVED transported_scopes already surfaced elsewhere. Default OFF, independent of clientRxCoverage/clientRxObservations/clientRfSamples: a deployment can run any combination. TOP-LEVEL FLAG, same as clientRfSamples. An empty regions list is a valid, stored answer ('nothing flood-allowed'); a repeater that never answers produces no row at all — silence carries no meaning, since the repeater ignores non-DIRECT requests. GPS is optional here (unlike the coverage/RF streams): a declared-regions answer is meaningful even without a position, so a missing fix is stored as NULL lat/lon rather than dropping the row. Requires an ACL-capable broker binding meshcore/client/{pubkey}/regions to that publisher — see docs/client-regions.md. Retention: see retention.clientRegionsDays; leaving it unset means unbounded growth.", "clientRxCoverage": { "enabled": false }, diff --git a/docs/client-rx-coverage.md b/docs/client-rx-coverage.md index e9480cd14..92a314e8c 100644 --- a/docs/client-rx-coverage.md +++ b/docs/client-rx-coverage.md @@ -206,6 +206,37 @@ toggle under Display shows or hides them (on by default). from one side only is placed towards that side. The ingestor stores the receptions; the server computes the estimate on request and caches it for 5 minutes. +### Uplink pings: `meshcore/client/{PUBLIC_KEY}/uplink` + +CoreDrive RX v1.21.0 and later can test whether the mesh hears the companion. With its `uplinkPing` +on, it sends `cdrx ` on a hashtag channel (`#coredrive` by default, usually region-scoped), +listens 10 s for repeaters re-broadcasting it, and publishes the ping here. Accepted only when +`clientUplink.enabled` is set; stored in `client_uplink_pings`. + +```json +{ + "origin": "", + "origin_id": "", + "timestamp": "2026-10-16T14:11:17.000Z", + "type": "UPLINK_PING", + "tag": "78277c0e", + "channel": "#coredrive", + "scope": "#be", + "pkt_hash": "b76c6a2777f9b0e9", + "echoes": [{ "first_hop": "1403", "last_hop": "1403", "hops": 1, "snr": 12.25, "rssi": -100 }], + "gps": { "lat": 50.998, "lon": 5.384, "acc_m": 6 } +} +``` + +- `tag` must be 8 hex characters. `pkt_hash` (16 hex) is present when an echo was heard; it is the + same hash `transmissions.hash` uses, so the ping can be matched with what observers received. +- Without an echo, the ingestor fills the link in itself: it decrypts GRP_TXT on the ping channel + (`clientUplink.channel`) and stores tag to hash in `client_uplink_tx`. The channel's hash byte is + shared with unrelated channels; the MAC check and the `cdrx ` text tell a ping apart. +- Echoes keep their first hop (the repeater that heard the companion) and last hop (the one the + companion heard it from). Invalid echoes are dropped; at most 32 are kept. +- The broker ACL must allow `/uplink`. + ## Capture HARD RULE — only what was heard directly The app and ingestor record **only the node the companion physically received**, never upstream