diff --git a/cmd/server/db.go b/cmd/server/db.go index ded9c37e..fa72d319 100644 --- a/cmd/server/db.go +++ b/cmd/server/db.go @@ -3174,8 +3174,10 @@ func (db *DB) getChannelScopeRegions(since string) (map[string][]string, error) // "#wardriving") over the requested window: message volume over time, who's // actively sending, which repeater first relayed each message (raw hash // prefixes — the caller resolves names via /api/resolve-hops), which -// observer stations actually heard the traffic, and signal quality (SNR/RSSI) -// over the same time buckets as the activity series. See WardrivingObserverCoverage +// observer stations actually heard the traffic, signal quality (SNR/RSSI) +// over the same time buckets as the activity series, and each sender's +// messages grouped into distinct sessions/runs (see buildWardrivingSessions). +// See WardrivingObserverCoverage // doc for why observer coverage — not sender GPS — is the reliable half of // a "where did this reach" picture: MeshMapper's #wardriving messages carry // an anonymous per-session token by default, not the sender's live @@ -3372,9 +3374,148 @@ func (db *DB) GetWardrivingStats(window, channel string) (*WardrivingStatsRespon resp.AvgRSSI = &v } + sessions, err := db.buildWardrivingSessions(channel, since) + if err != nil { + return nil, err + } + resp.Sessions = sessions + return resp, nil } +// wardrivingSessionGapMinutes is the max gap between two consecutive +// messages from the same sender before buildWardrivingSessions treats them +// as separate wardriving runs rather than one continuous session. +const wardrivingSessionGapMinutes = 15.0 + +// buildWardrivingSessions groups each sender's messages (ordered by time) +// into runs, splitting on any gap over wardrivingSessionGapMinutes. For +// each session it also computes how many distinct entry-point repeaters +// and observers were involved, by unioning the per-transmission +// observation data across every message in that session. +func (db *DB) buildWardrivingSessions(channel, since string) ([]WardrivingSession, error) { + msgRows, err := db.conn.Query(` + SELECT id, json_extract(decoded_json, '$.sender') AS sender, first_seen + FROM transmissions + WHERE channel_hash = ? AND payload_type = 5 AND first_seen >= ? + AND json_extract(decoded_json, '$.sender') IS NOT NULL + AND json_extract(decoded_json, '$.sender') != '' + ORDER BY sender, first_seen ASC + `, channel, since) + if err != nil { + return nil, fmt.Errorf("wardriving sessions message query: %w", err) + } + type txInfo struct { + id int64 + sender string + ts time.Time + } + var txs []txInfo + for msgRows.Next() { + var id int64 + var sender, tsStr string + if err := msgRows.Scan(&id, &sender, &tsStr); err != nil { + continue + } + ts, err := time.Parse(time.RFC3339, tsStr) + if err != nil { + continue + } + txs = append(txs, txInfo{id: id, sender: sender, ts: ts}) + } + msgRows.Close() + if err := msgRows.Err(); err != nil { + return nil, fmt.Errorf("wardriving sessions message iteration: %w", err) + } + + // Per-transmission entry-point prefixes and observer IDs, so each + // session can report how many distinct ones it touched. + var perTxQuery string + if db.isV3 { + perTxQuery = ` + SELECT o.transmission_id, json_extract(o.path_json, '$[0]'), obs.rowid + FROM observations o + JOIN transmissions t ON t.id = o.transmission_id + JOIN observers obs ON obs.rowid = o.observer_idx + WHERE t.channel_hash = ? AND t.payload_type = 5 AND t.first_seen >= ?` + } else { + perTxQuery = ` + SELECT o.transmission_id, json_extract(o.path_json, '$[0]'), obs.id + FROM observations o + JOIN transmissions t ON t.id = o.transmission_id + JOIN observers obs ON obs.id = o.observer_id + WHERE t.channel_hash = ? AND t.payload_type = 5 AND t.first_seen >= ?` + } + perTxRows, err := db.conn.Query(perTxQuery, channel, since) + if err != nil { + return nil, fmt.Errorf("wardriving sessions per-tx query: %w", err) + } + txPrefixes := make(map[int64]map[string]bool) + txObservers := make(map[int64]map[string]bool) + for perTxRows.Next() { + var txID int64 + var prefix sql.NullString + var observerID string + if err := perTxRows.Scan(&txID, &prefix, &observerID); err != nil { + continue + } + if prefix.Valid && prefix.String != "" { + if txPrefixes[txID] == nil { + txPrefixes[txID] = make(map[string]bool) + } + txPrefixes[txID][prefix.String] = true + } + if txObservers[txID] == nil { + txObservers[txID] = make(map[string]bool) + } + txObservers[txID][observerID] = true + } + perTxRows.Close() + if err := perTxRows.Err(); err != nil { + return nil, fmt.Errorf("wardriving sessions per-tx iteration: %w", err) + } + + sessions := make([]WardrivingSession, 0) + var cur *WardrivingSession + var curPrefixes, curObservers map[string]bool + var lastTS time.Time + flush := func() { + if cur == nil { + return + } + cur.EntryPointCount = len(curPrefixes) + cur.ObserverCount = len(curObservers) + start, errS := time.Parse(time.RFC3339, cur.StartTime) + end, errE := time.Parse(time.RFC3339, cur.EndTime) + if errS == nil && errE == nil { + cur.DurationMinutes = end.Sub(start).Minutes() + } + sessions = append(sessions, *cur) + } + for _, tx := range txs { + newSession := cur == nil || cur.Sender != tx.sender || tx.ts.Sub(lastTS).Minutes() > wardrivingSessionGapMinutes + if newSession { + flush() + cur = &WardrivingSession{Sender: tx.sender, StartTime: tx.ts.UTC().Format(time.RFC3339)} + curPrefixes = make(map[string]bool) + curObservers = make(map[string]bool) + } + cur.EndTime = tx.ts.UTC().Format(time.RFC3339) + cur.MessageCount++ + for p := range txPrefixes[tx.id] { + curPrefixes[p] = true + } + for o := range txObservers[tx.id] { + curObservers[o] = true + } + lastTS = tx.ts + } + flush() + + sort.Slice(sessions, func(i, j int) bool { return sessions[i].StartTime > sessions[j].StartTime }) + return sessions, nil +} + // GetMatchedRegionNames returns the set of scope_name values that have ever // matched at least one transmission still in retention (NULL and empty-string // "unknown" rows are excluded). Used to diff against the operator's diff --git a/cmd/server/openapi.go b/cmd/server/openapi.go index 7e05fec8..d0107233 100644 --- a/cmd/server/openapi.go +++ b/cmd/server/openapi.go @@ -102,7 +102,7 @@ func routeDescriptions() map[string]routeMeta { "GET /api/analytics/subpaths-bulk": {Summary: "Bulk subpath analysis", Tag: "analytics"}, "GET /api/analytics/subpath-detail": {Summary: "Subpath detail", Tag: "analytics"}, "GET /api/analytics/neighbor-graph": {Summary: "Neighbor graph", Description: "Full neighbor affinity graph for visualization.", Tag: "analytics"}, - "GET /api/analytics/wardriving": {Summary: "Wardriving channel analytics", Description: "Activity/entry-point/coverage/signal analytics for the #wardriving channel (or another channel via ?channel=): message volume over time, top senders, path[0] entry-point hash-prefix tallies (resolve names via /api/resolve-hops), per-observer coverage (observer's known IATA-derived coordinates, not the sender's — MeshMapper's wardriving messages carry an anonymous session token by default, not live GPS), and average SNR/RSSI over the same time buckets as the activity series. Cached 30s per window+channel.", Tag: "analytics", + "GET /api/analytics/wardriving": {Summary: "Wardriving channel analytics", Description: "Activity/entry-point/coverage/signal/session analytics for the #wardriving channel (or another channel via ?channel=): message volume over time, top senders, path[0] entry-point hash-prefix tallies (resolve names via /api/resolve-hops), per-observer coverage (observer's known IATA-derived coordinates, not the sender's — MeshMapper's wardriving messages carry an anonymous session token by default, not live GPS), average SNR/RSSI over the same time buckets as the activity series, and each sender's messages grouped into distinct sessions/runs (split on a 15-minute gap). Cached 30s per window+channel.", Tag: "analytics", QueryParams: []paramMeta{ {Name: "window", Description: "Time window: 1h, 24h (default), or 7d", Type: "string"}, {Name: "channel", Description: "Channel name to analyze (default #wardriving)", Type: "string"}, diff --git a/cmd/server/types.go b/cmd/server/types.go index 09eb245f..7cfdd2bf 100644 --- a/cmd/server/types.go +++ b/cmd/server/types.go @@ -262,6 +262,20 @@ type WardrivingSignalPoint struct { ObservationCount int `json:"observationCount"` } +// WardrivingSession groups one sender's messages into a distinct "run": +// consecutive messages no more than wardrivingSessionGapMinutes apart. A +// bigger gap starts a new session, on the theory the sender paused, went +// out of range, or ended one wardriving trip and started another later. +type WardrivingSession struct { + Sender string `json:"sender"` + StartTime string `json:"startTime"` + EndTime string `json:"endTime"` + DurationMinutes float64 `json:"durationMinutes"` + MessageCount int `json:"messageCount"` + EntryPointCount int `json:"entryPointCount"` // distinct path[0] entry-point prefixes seen during the session + ObserverCount int `json:"observerCount"` // distinct observers that heard any message in the session +} + type WardrivingStatsResponse struct { Window string `json:"window"` Channel string `json:"channel"` @@ -273,6 +287,7 @@ type WardrivingStatsResponse struct { SignalTimeSeries []WardrivingSignalPoint `json:"signalTimeSeries"` AvgSNR *float64 `json:"avgSnr,omitempty"` AvgRSSI *float64 `json:"avgRssi,omitempty"` + Sessions []WardrivingSession `json:"sessions"` } // ─── Health ──────────────────────────────────────────────────────────────────── diff --git a/cmd/server/wardriving_stats_test.go b/cmd/server/wardriving_stats_test.go index f830a711..706bf732 100644 --- a/cmd/server/wardriving_stats_test.go +++ b/cmd/server/wardriving_stats_test.go @@ -177,6 +177,113 @@ func TestHandleWardrivingStats(t *testing.T) { } } +// TestHandleWardrivingStats_Sessions covers session/run grouping: +// consecutive messages within wardrivingSessionGapMinutes (15) from the +// same sender merge into one session; a bigger gap starts a new one. +func TestHandleWardrivingStats_Sessions(t *testing.T) { + srv, router := setupTestServer(t) + if _, err := srv.db.conn.Exec(`DELETE FROM transmissions`); err != nil { + t.Fatalf("clear transmissions: %v", err) + } + if _, err := srv.db.conn.Exec(`DELETE FROM observations`); err != nil { + t.Fatalf("clear observations: %v", err) + } + + base := time.Now().UTC().Add(-2 * time.Hour) + t0 := base // Alice, session A, msg 1 + t1 := base.Add(5 * time.Minute) // Alice, session A, msg 2 (5min gap — same session) + t2 := base.Add(40 * time.Minute) // Alice, session B, msg 3 (35min gap from t1 — new session) + t3 := base.Add(10 * time.Minute) // Bob, single-message session + + insertTx := func(hash, sender string, ts time.Time) int64 { + res, err := srv.db.conn.Exec( + `INSERT INTO transmissions (raw_hex,hash,first_seen,route_type,payload_type,channel_hash,decoded_json) VALUES (?,?,?,1,5,'#wardriving',?)`, + "aa", hash, ts.Format(time.RFC3339), `{"sender":"`+sender+`","text":"`+sender+`: MM:x"}`, + ) + if err != nil { + t.Fatalf("insert tx %s: %v", hash, err) + } + id, _ := res.LastInsertId() + return id + } + tx0 := insertTx("s1", "Alice", t0) + tx1 := insertTx("s2", "Alice", t1) + tx2 := insertTx("s3", "Alice", t2) + tx3 := insertTx("s4", "Bob", t3) + + insertObserver := func(id, name string) int64 { + res, err := srv.db.conn.Exec(`INSERT INTO observers (id, name) VALUES (?,?)`, id, name) + if err != nil { + t.Fatalf("insert observer %s: %v", id, err) + } + rowid, _ := res.LastInsertId() + return rowid + } + o1 := insertObserver("obsO1", "ObsOne") + o2 := insertObserver("obsO2", "ObsTwo") + + insertObs := func(txID, observerIdx int64, pathJSON string) { + if _, err := srv.db.conn.Exec( + `INSERT INTO observations (transmission_id, observer_idx, snr, rssi, path_json, timestamp) VALUES (?,?,1.0,-90,?,?)`, + txID, observerIdx, pathJSON, time.Now().Unix(), + ); err != nil { + t.Fatalf("insert observation: %v", err) + } + } + insertObs(tx0, o1, `["EEEE"]`) + insertObs(tx1, o1, `["FFFF"]`) + insertObs(tx2, o2, `["EEEE"]`) + insertObs(tx3, o1, `["GGGG"]`) + + req := httptest.NewRequest("GET", "/api/analytics/wardriving?window=24h", nil) + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + if w.Code != 200 { + t.Fatalf("status=%d body=%s", w.Code, w.Body.String()) + } + var resp WardrivingStatsResponse + if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil { + t.Fatalf("decode: %v body=%s", err, w.Body.String()) + } + + if len(resp.Sessions) != 3 { + t.Fatalf("Sessions = %+v, want 3 (Alice session A, Alice session B, Bob session)", resp.Sessions) + } + + // Ordered most-recent-first by StartTime: Alice-B (t2), Bob (t3), Alice-A (t0). + aliceB, bob, aliceA := resp.Sessions[0], resp.Sessions[1], resp.Sessions[2] + + if aliceA.Sender != "Alice" || aliceA.MessageCount != 2 { + t.Errorf("Alice session A = %+v, want {Alice, 2 messages}", aliceA) + } + if aliceA.DurationMinutes < 4.9 || aliceA.DurationMinutes > 5.1 { + t.Errorf("Alice session A DurationMinutes = %v, want ~5.0", aliceA.DurationMinutes) + } + if aliceA.EntryPointCount != 2 { + t.Errorf("Alice session A EntryPointCount = %d, want 2 (EEEE, FFFF)", aliceA.EntryPointCount) + } + if aliceA.ObserverCount != 1 { + t.Errorf("Alice session A ObserverCount = %d, want 1 (only ObsOne heard it)", aliceA.ObserverCount) + } + + if aliceB.Sender != "Alice" || aliceB.MessageCount != 1 { + t.Errorf("Alice session B = %+v, want {Alice, 1 message}", aliceB) + } + if aliceB.EntryPointCount != 1 || aliceB.ObserverCount != 1 { + t.Errorf("Alice session B = %+v, want {1 entry point, 1 observer}", aliceB) + } + + if bob.Sender != "Bob" || bob.MessageCount != 1 { + t.Errorf("Bob session = %+v, want {Bob, 1 message}", bob) + } + + // The 35-minute gap between t1 and t2 must NOT merge into one session — + // this is the core behavior under test. + if aliceA.MessageCount+aliceB.MessageCount != 3 { + t.Errorf("Alice's 3 messages should split into two sessions (2 + 1), got %d + %d", aliceA.MessageCount, aliceB.MessageCount) + } +} + // TestHandleWardrivingStats_InvalidWindow mirrors the existing scope-stats // window validation. func TestHandleWardrivingStats_InvalidWindow(t *testing.T) { @@ -211,9 +318,9 @@ func TestHandleWardrivingStats_EmptyChannel(t *testing.T) { if resp.TotalMessages != 0 { t.Errorf("TotalMessages = %d, want 0", resp.TotalMessages) } - if resp.TopSenders == nil || resp.EntryPoints == nil || resp.Observers == nil || resp.TimeSeries == nil || resp.SignalTimeSeries == nil { - t.Errorf("expected empty (non-nil) slices, got TopSenders=%v EntryPoints=%v Observers=%v TimeSeries=%v SignalTimeSeries=%v", - resp.TopSenders, resp.EntryPoints, resp.Observers, resp.TimeSeries, resp.SignalTimeSeries) + if resp.TopSenders == nil || resp.EntryPoints == nil || resp.Observers == nil || resp.TimeSeries == nil || resp.SignalTimeSeries == nil || resp.Sessions == nil { + t.Errorf("expected empty (non-nil) slices, got TopSenders=%v EntryPoints=%v Observers=%v TimeSeries=%v SignalTimeSeries=%v Sessions=%v", + resp.TopSenders, resp.EntryPoints, resp.Observers, resp.TimeSeries, resp.SignalTimeSeries, resp.Sessions) } if resp.AvgSNR != nil || resp.AvgRSSI != nil { t.Errorf("expected nil AvgSNR/AvgRSSI for a channel with no observations, got AvgSNR=%v AvgRSSI=%v", resp.AvgSNR, resp.AvgRSSI) diff --git a/public/analytics.js b/public/analytics.js index 31b052bc..49f0b052 100644 --- a/public/analytics.js +++ b/public/analytics.js @@ -5426,6 +5426,7 @@ function destroy() { _stopRolesRefresh(); _stopScopesRefresh(); _stopForeignTraf { label: 'Observers Reached', value: (d.observers || []).length.toLocaleString(), note: null }, { label: 'Avg SNR', value: (d.avgSnr != null ? d.avgSnr.toFixed(1) + ' dB' : '—'), note: null }, { label: 'Avg RSSI', value: (d.avgRssi != null ? d.avgRssi.toFixed(1) + ' dBm' : '—'), note: null }, + { label: 'Sessions', value: (d.sessions || []).length.toLocaleString(), note: '15min+ gap starts a new one' }, ].map(function(c) { return '
' + c.value + '
' + '
' + c.label + '
' + @@ -5493,6 +5494,33 @@ function destroy() { _stopRolesRefresh(); _stopScopesRefresh(); _stopForeignTraf ''; } + function formatSessionDuration(mins) { + if (mins < 1) return '<1m'; + if (mins < 60) return Math.round(mins) + 'm'; + var h = Math.floor(mins / 60), m = Math.round(mins % 60); + return h + 'h ' + m + 'm'; + } + + // Sessions — each sender's messages grouped into runs (backend splits + // on any gap over 15 minutes). Pre-sorted most-recent-first by the API. + function sessionsHtml(sessions) { + if (!sessions || sessions.length === 0) { + return '

No wardriving sessions in this window.

'; + } + var rows = sessions.map(function(s) { + return '' + esc(s.sender) + '' + + '' + (typeof timeAgo === 'function' ? timeAgo(s.startTime) : s.startTime) + '' + + '' + formatSessionDuration(s.durationMinutes) + '' + + '' + s.messageCount.toLocaleString() + '' + + '' + s.entryPointCount.toLocaleString() + '' + + '' + s.observerCount.toLocaleString() + ''; + }).join(''); + return '' + + '' + + '' + rows + '' + + '
SenderStartedDurationMessagesEntry PointsObservers
'; + } + function sendersHtml(senders, totalMessages) { if (!senders || senders.length === 0) { return '

No wardriving messages in this window.

'; @@ -5592,6 +5620,9 @@ function destroy() { _stopRolesRefresh(); _stopScopesRefresh(); _stopForeignTraf '

Top Senders

' + '

Who\'s actively wardriving in this window, by message count.

' + '
' + sendersHtml(d.topSenders, d.totalMessages) + '
' + + '

Sessions

' + + '

Each sender\'s messages grouped into distinct runs — a gap of more than 15 minutes starts a new session.

' + + '
' + sessionsHtml(d.sessions) + '
' + '

Entry Points

' + '

Which local repeater first relayed each wardriving message — the hop closest to the origin (path[0]) across every observed copy.

' + '
' + entryHtml + '
' + diff --git a/test-analytics-wardriving-tab.js b/test-analytics-wardriving-tab.js index 58c3f714..de03bea1 100644 --- a/test-analytics-wardriving-tab.js +++ b/test-analytics-wardriving-tab.js @@ -137,6 +137,11 @@ function makeWardrivingResponse(overrides) { ], avgSnr: 5.5, avgRssi: -72.5, + sessions: [ + { sender: 'Alice', startTime: '2026-07-20T09:00:00Z', endTime: '2026-07-20T09:00:00Z', durationMinutes: 0, messageCount: 1, entryPointCount: 1, observerCount: 1 }, + { sender: 'Bob', startTime: '2026-07-20T08:30:00Z', endTime: '2026-07-20T08:30:00Z', durationMinutes: 0, messageCount: 1, entryPointCount: 1, observerCount: 1 }, + { sender: 'Alice', startTime: '2026-07-20T08:00:00Z', endTime: '2026-07-20T08:05:00Z', durationMinutes: 5, messageCount: 1, entryPointCount: 2, observerCount: 1 }, + ], }, overrides); } @@ -191,6 +196,20 @@ function makeApiStub(wardrivingResp, resolveHopsResp) { assert.ok(el.innerHTML.includes('66.7%'), 'Alice row should show 66.7% of total messages'); }); + await testAsync('Sessions table lists each run with duration, entry points, and observers', async () => { + const ctx = makeAnalyticsSandbox(makeApiStub(makeWardrivingResponse())); + const el = fakeEl(); + await ctx.window._analyticsRenderWardrivingTab(el); + const startIdx = el.innerHTML.indexOf('id="wardrivingSessions"'); + const endIdx = el.innerHTML.indexOf('id="wardrivingEntryPoints"'); + const section = el.innerHTML.slice(startIdx, endIdx); + // 3 sessions in the fixture: two Alice rows, one Bob row. + assert.strictEqual((section.match(/Alice/g) || []).length, 2, 'both Alice sessions should render as separate rows'); + assert.ok(section.includes('Bob'), 'Bob session should render'); + assert.ok(section.includes('5m'), 'the 5-minute session should show its duration'); + assert.ok(el.innerHTML.includes('
3
Sessions
'), 'the Sessions stat card should show the session count (3)'); + }); + await testAsync('Entry Points resolves unique_prefix repeaters and folds ambiguous into one bucket', async () => { const ctx = makeAnalyticsSandbox(makeApiStub(makeWardrivingResponse(), { resolved: { @@ -200,8 +219,8 @@ function makeApiStub(wardrivingResp, resolveHopsResp) { })); const el = fakeEl(); await ctx.window._analyticsRenderWardrivingTab(el); - const startIdx = el.innerHTML.indexOf('Entry Points'); - const endIdx = el.innerHTML.indexOf('Coverage by Observer'); + const startIdx = el.innerHTML.indexOf('id="wardrivingEntryPoints"'); + const endIdx = el.innerHTML.indexOf('id="wardrivingObservers"'); const section = el.innerHTML.slice(startIdx, endIdx); assert.ok(section.includes('GatewayRepeater'), 'unique_prefix resolution should show the real repeater name'); assert.ok(!section.includes('BestGuessRepeater'), 'a non-unique_prefix resolution must not be shown as a specific named repeater'); @@ -227,11 +246,12 @@ function makeApiStub(wardrivingResp, resolveHopsResp) { await testAsync('shows empty-state messages when the window has no wardriving activity', async () => { const ctx = makeAnalyticsSandbox(makeApiStub(makeWardrivingResponse({ totalMessages: 0, topSenders: [], entryPoints: [], observers: [], timeSeries: [], - signalTimeSeries: [], avgSnr: null, avgRssi: null, + signalTimeSeries: [], avgSnr: null, avgRssi: null, sessions: [], }))); const el = fakeEl(); await ctx.window._analyticsRenderWardrivingTab(el); assert.ok(el.innerHTML.includes('No wardriving messages in this window'), 'senders empty state should show'); + assert.ok(el.innerHTML.includes('No wardriving sessions in this window'), 'sessions empty state should show'); assert.ok(el.innerHTML.includes('No wardriving messages with a relay path'), 'entry points empty state should show'); assert.ok(el.innerHTML.includes('No observer has heard wardriving traffic'), 'observers empty state should show'); assert.ok(el.innerHTML.includes('Insufficient data points to chart'), 'signal chart empty state should show');