diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7d374920..da907fe7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -55,7 +55,7 @@ jobs: - name: Verify stats and endpoint queries against PostgreSQL 16 env: BEACON_TEST_POSTGRES_DSN: postgres://postgres:backup-ci-only@127.0.0.1:5432/postgres?sslmode=disable - run: go test ./db -run '^Test(Signal|Paths|PacketEndpointsResolveLive|AnalyticsRetention|AnalyticsRetentionConcurrent|DeleteOldPacketsBatches)Postgres$' -count=1 -v + run: go test ./db -run '^Test(Signal|Paths|PacketEndpointsResolveLive|ObserverMetrics|RouteEvidence|RouteEvidenceIndex|AnalyticsRetention|AnalyticsRetentionConcurrent|DeleteOldPacketsBatches)Postgres$' -count=1 -v - name: Verify backup command against PostgreSQL 16 run: | diff --git a/README.md b/README.md index 823c04be..3ecf2d55 100644 --- a/README.md +++ b/README.md @@ -550,3 +550,24 @@ sub-hour observer activity, exact observer comparison and current entity/scope counts continue to describe retained raw data or current entities, rather than claiming archived packet detail. Archived summaries expire without waiting for new packet deletions. A failed archive leaves its entire raw batch intact. + +### Observer monitoring metrics + +Observer activity returns buckets in `[windowStart, windowEnd)`, including the current partial bucket for live requests, plus +`generatedAt`, `source` (`raw` or `hourly`) and `summary`. `recordedPackets` is the +sum of stored observations in those buckets; repeated broker delivery of the +same retained packet/observer pair counts once. Unknown payload types appear as +`-1` rather than disappearing. Freshness fields are measured at `generatedAt` even for historical `until` requests; `recordedPackets` alone follows the selected window. `lastCompleteHour` uses the previous complete UTC +hour and includes its own start/end; `latestRecordedAt` is the latest retained +reception timestamp. Missing records do not prove downtime. The optional `until` +(epoch milliseconds, within the last 30 days) aligns two observers' charts. + +The existing observer `observationCount` remains a legacy cumulative presence +counter for compatibility, including status/neighbour events. It is not a +period packet total. Broker presence and packet-arrival timestamps are now +updated separately; this cannot reconstruct previously overwritten timestamps. +The initial analytics archive migration includes unknown-type activity. Existing +development previews that used the earlier draft require a separate operator repair; +radio samples already discarded for those legacy rows cannot be recovered. + +Saved-route evidence: see [the operator guide](https://github.com/MeshCore-Beacon/beacon-docs/blob/main/app_documentation/saved-route-evidence.md). diff --git a/db/migrations/041_route_evidence_index.sql b/db/migrations/041_route_evidence_index.sql new file mode 100644 index 00000000..71088cc5 --- /dev/null +++ b/db/migrations/041_route_evidence_index.sql @@ -0,0 +1,9 @@ +-- Copyright 2026 Beacon Contributors +-- SPDX-License-Identifier: AGPL-3.0-or-later + +-- One statement outside a transaction. RunMigrations recovers interrupted builds; +-- avoid IF NOT EXISTS, which would silently accept an invalid index. +-- A compact digest narrows candidates; queries still compare the complete bytes. +CREATE INDEX CONCURRENTLY idx_observations_route_evidence +ON packet_observations (iata, hash_size, (decode(md5(path_bytes), 'hex')), heard_at DESC, id DESC) +WHERE path_bytes IS NOT NULL AND payload_type IS NOT NULL AND payload_type <> 9 AND hop_count >= 2; diff --git a/db/observer_metrics_integration_test.go b/db/observer_metrics_integration_test.go new file mode 100644 index 00000000..91dbeca1 --- /dev/null +++ b/db/observer_metrics_integration_test.go @@ -0,0 +1,95 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later +package db + +import ( + "encoding/json" + sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" + "strings" + "testing" + "time" +) + +func TestObserverMetricsPostgres(t *testing.T) { + ctx, tx := retentionTx(t) + analyticsTables(t, ctx, tx) + // Use the real observer DDL without depending on a preinstalled public schema. + initial, err := migrationFiles.ReadFile("migrations/001_initial_schema.sql") + if err != nil { + t.Fatal(err) + } + ddl := string(initial) + start := strings.Index(ddl, "CREATE TABLE observers (") + end := strings.Index(ddl[start:], "\n);") + start + 3 + if _, err := tx.Exec(ctx, "DROP TABLE observers; "+ddl[start:end]+" ALTER TABLE observers ADD COLUMN region_scope TEXT;"); err != nil { + t.Fatal(err) + } + if _, err := tx.Exec(ctx, ` + CREATE TABLE observer_brokers (observer_id uuid,broker_name text,last_seen timestamptz,last_packet_at timestamptz,PRIMARY KEY(observer_id,broker_name)); + INSERT INTO observers (id,public_key) VALUES ('00000000-0000-0000-0000-000000000001','\x01'); + INSERT INTO packets(packet_hash,payload_type,last_heard_at,first_heard_at) SELECT int4send(i),4,NOW()-interval '5 days',NOW()-interval '5 days' FROM generate_series(1,3) i; + INSERT INTO packet_observations(packet_hash,observer_id,iata,heard_at,path_length_byte,hash_size,hop_count,payload_type,snr,rssi) + SELECT packet_hash,'00000000-0000-0000-0000-000000000001','YOW',date_trunc('hour',NOW())-interval '5 days',0,1,0,CASE WHEN packet_hash=int4send(1) THEN 4 END,CASE WHEN packet_hash IN (int4send(1),int4send(2)) THEN 'NaN'::real ELSE 0 END,-100 FROM packets; + `); err != nil { + t.Fatal(err) + } + applyStatsMigration(t, ctx, tx, "039_analytics_retention.sql") + store := &Store{q: sqlc.New(tx)} + id := uuid.MustParse("00000000-0000-0000-0000-000000000001") + if err := store.DeleteOldPackets(ctx, time.Now().Add(-72*time.Hour)); err != nil { + t.Fatal(err) + } + // The initial archive migration retains unknown payloads without a second rebuild. + if err := store.RefreshObserverActivity(ctx); err != nil { + t.Fatal(err) + } + until := time.Now().UTC().Truncate(time.Hour) + activity, err := store.GetObserverActivity(ctx, id, 7*24*time.Hour, time.Hour, until) + if err != nil { + t.Fatal(err) + } + if _, err := json.Marshal(activity); err != nil { + t.Fatalf("non-finite sample leaked into JSON: %v", err) + } + if activity.Summary.RecordedPackets != 3 || len(activity.PayloadTypes) != 2 || activity.Summary.LatestRecordedAt != nil || activity.Summary.LastCompleteHour != 0 { + t.Fatalf("archive summary: %+v types=%+v", activity.Summary, activity.PayloadTypes) + } + if activity.WindowEnd != until.UnixMilli() || activity.WindowStart != until.Add(-7*24*time.Hour).UnixMilli() || activity.Source != "hourly" { + t.Fatalf("window: %+v", activity) + } + if _, err := tx.Exec(ctx, `INSERT INTO packets(packet_hash,last_heard_at) VALUES ('\xaa',NOW()); + INSERT INTO packet_observations(packet_hash,observer_id,iata,heard_at,path_length_byte,hash_size,hop_count,snr,rssi) VALUES ('\xaa','00000000-0000-0000-0000-000000000001','YOW',date_trunc('hour',NOW())-interval '30 minutes',0,1,0,'NaN'::real,-100); + INSERT INTO packet_observations(packet_hash,observer_id,iata,heard_at) VALUES ('\xaa','00000000-0000-0000-0000-000000000001','YOW',NOW()) ON CONFLICT DO NOTHING;`); err != nil { + t.Fatal(err) + } + raw, err := store.GetObserverActivity(ctx, id, time.Hour, 15*time.Minute, until) + if err != nil { + t.Fatal(err) + } + if _, err := json.Marshal(raw); err != nil { + t.Fatalf("raw non-finite sample leaked into JSON: %v", err) + } + if raw.Summary.RecordedPackets != 1 || raw.Summary.LastCompleteHour != 1 || raw.Summary.LatestRecordedAt == nil || raw.PayloadTypes[0].PayloadType != -1 { + t.Fatalf("raw/duplicate: %+v", raw) + } + if err := store.UpsertObserverBroker(ctx, id, "one", false); err != nil { + t.Fatal(err) + } + var packet pgtype.Timestamptz + tx.QueryRow(ctx, "SELECT last_packet_at FROM observer_brokers").Scan(&packet) + if packet.Valid { + t.Fatal("status created a packet timestamp") + } + if err := store.UpsertObserverBroker(ctx, id, "one", true); err != nil { + t.Fatal(err) + } + if err := store.TouchObserverBrokers(ctx, []uuid.UUID{id}, []string{"one"}, []time.Time{until.Add(time.Hour)}, []time.Time{{}}); err != nil { + t.Fatal(err) + } + tx.QueryRow(ctx, "SELECT last_packet_at FROM observer_brokers").Scan(&packet) + if !packet.Valid { + t.Fatal("status erased packet timestamp") + } +} diff --git a/db/observer_retention_integration_test.go b/db/observer_retention_integration_test.go index 8672ea42..8e64ea9c 100644 --- a/db/observer_retention_integration_test.go +++ b/db/observer_retention_integration_test.go @@ -100,7 +100,7 @@ SELECT id,1 FROM observers WHERE display_name='expired';`) if err != nil { t.Fatal(err) } - if err := coalescer.UpsertObserverBroker(ctx, oldID, "fixture"); err != nil { + if err := coalescer.UpsertObserverBroker(ctx, oldID, "fixture", true); err != nil { t.Fatal(err) } // A future cutoff simulates this cached observer having aged past retention. @@ -112,7 +112,7 @@ SELECT id,1 FROM observers WHERE display_name='expired';`) if err != nil || newID == oldID { t.Fatalf("returning observer reused a deleted ID: %v", err) } - if err := coalescer.UpsertObserverBroker(ctx, newID, "fixture"); err != nil { + if err := coalescer.UpsertObserverBroker(ctx, newID, "fixture", true); err != nil { t.Fatal(err) } var brokers int diff --git a/db/observer_window_test.go b/db/observer_window_test.go new file mode 100644 index 00000000..205d1ef8 --- /dev/null +++ b/db/observer_window_test.go @@ -0,0 +1,55 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package db + +import ( + "context" + "testing" + "time" + + sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" + mockdb "github.com/MeshCore-Beacon/beacon-server/db/sqlc/mock" + "github.com/google/uuid" + "go.uber.org/mock/gomock" +) + +func TestObserverActivityLiveAndFixedWindow(t *testing.T) { + for _, interval := range []time.Duration{15 * time.Minute, time.Hour, 6 * time.Hour} { + for _, fixed := range []bool{false, true} { + t.Run(interval.String()+"/"+map[bool]string{false: "live", true: "fixed"}[fixed], func(t *testing.T) { + mock := mockdb.NewMockQuerier(gomock.NewController(t)) + id := uuid.New() + mock.EXPECT().GetObserverByID(gomock.Any(), id).Return(sqlc.Observer{}, nil) + mock.EXPECT().GetObserverActivityLiveSummary(gomock.Any(), gomock.Any()).Return(sqlc.GetObserverActivityLiveSummaryRow{}, nil) + if interval < time.Hour { + mock.EXPECT().GetObserverActivityRaw(gomock.Any(), gomock.Any()).Return(nil, nil) + mock.EXPECT().GetObserverActivityRawPayloadTypes(gomock.Any(), gomock.Any()).Return(nil, nil) + } else { + mock.EXPECT().GetObserverActivityHourly(gomock.Any(), gomock.Any()).Return(nil, nil) + mock.EXPECT().GetObserverActivityHourlyPayloadTypes(gomock.Any(), gomock.Any()).Return(nil, nil) + } + before := time.Now().UTC() + var until time.Time + if fixed { + until = before.Add(-24*time.Hour + 123*time.Millisecond) + } + got, err := (&Store{q: mock}).GetObserverActivity(context.Background(), id, 48*time.Hour, interval, until) + if err != nil { + t.Fatal(err) + } + if fixed { + if got.WindowEnd != until.Truncate(interval).UnixMilli() { + t.Fatalf("fixed end was not aligned: %d", got.WindowEnd) + } + } else if got.WindowEnd < before.UnixMilli() || got.WindowEnd > time.Now().UnixMilli() { + t.Fatalf("live end dropped the partial bucket: %d", got.WindowEnd) + } + // Freshness is always live, even when the activity window is historical. + if got.Summary.LastCompleteHourEnd != time.UnixMilli(got.GeneratedAt).UTC().Truncate(time.Hour).UnixMilli() { + t.Fatal("freshness was anchored to the historical activity window") + } + }) + } + } +} diff --git a/db/observers.go b/db/observers.go index a3448cf1..fc10efad 100644 --- a/db/observers.go +++ b/db/observers.go @@ -8,6 +8,7 @@ import ( "encoding/hex" "fmt" "log/slog" + "math" "time" sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" @@ -230,12 +231,38 @@ func (s *Store) GetObserverTelemetryBucketed(ctx context.Context, observerID uui // GetObserverActivity returns bucketed heard-activity for an observer over the trailing window. // Buckets of an hour or coarser come from the hourly rollup; anything finer reads observations directly. // Range and Interval are left empty for the handler to fill. -func (s *Store) GetObserverActivity(ctx context.Context, observerID uuid.UUID, window, interval time.Duration) (*api.ObserverActivity, error) { +func (s *Store) GetObserverActivity(ctx context.Context, observerID uuid.UUID, window, interval time.Duration, until time.Time) (*api.ObserverActivity, error) { obs, err := s.q.GetObserverByID(ctx, observerID) if err != nil { return nil, err } - activity := &api.ObserverActivity{} + now := time.Now().UTC() + if until.IsZero() { + until = now + } else { + until = until.UTC().Truncate(interval) + } + since := until.Add(-window) + if !since.Equal(since.Truncate(interval)) { + since = since.Truncate(interval).Add(interval) + } + activity := &api.ObserverActivity{WindowStart: since.UnixMilli(), WindowEnd: until.UnixMilli(), GeneratedAt: now.UnixMilli(), Source: "raw", Summary: &api.ObserverActivitySummary{}} + if interval >= time.Hour { + activity.Source = "hourly" + } + live, err := s.q.GetObserverActivityLiveSummary(ctx, sqlc.GetObserverActivityLiveSummaryParams{ + ObserverID: observerID, GeneratedAt: pgtype.Timestamptz{Time: now, Valid: true}, + HourStart: pgtype.Timestamptz{Time: now.Truncate(time.Hour).Add(-time.Hour), Valid: true}, HourEnd: pgtype.Timestamptz{Time: now.Truncate(time.Hour), Valid: true}}) + if err != nil { + return nil, err + } + activity.Summary.LastCompleteHour = live.LastCompleteHour + if live.LatestRecordedAt.Valid { + v := live.LatestRecordedAt.Time.UnixMilli() + activity.Summary.LatestRecordedAt = &v + } + activity.Summary.LastCompleteHourStart = now.Truncate(time.Hour).Add(-time.Hour).UnixMilli() + activity.Summary.LastCompleteHourEnd = now.Truncate(time.Hour).UnixMilli() // radio is non-nil only when airtime is actually costable, so radio != null implies costed buckets if obs.RadioSf != nil && obs.RadioBwKhz != nil && obs.RadioCr != nil && *obs.RadioSf >= 7 && *obs.RadioSf <= 12 && *obs.RadioBwKhz > 0 && *obs.RadioCr > 0 { @@ -247,12 +274,6 @@ func (s *Store) GetObserverActivity(ctx context.Context, observerID uuid.UUID, w PreambleSymbols: lora.PreambleSymbols(int(*obs.RadioSf)), } } - // round the window start up to a bucket boundary so the first bucket is never a partial one - start := time.Now().Add(-window).UTC() - since := start.Truncate(interval) - if since.Before(start) { - since = since.Add(interval) - } sinceTS := pgtype.Timestamptz{Time: since, Valid: true} binWidth := pgtype.Interval{Microseconds: interval.Microseconds(), Valid: true} @@ -261,6 +282,7 @@ func (s *Store) GetObserverActivity(ctx context.Context, observerID uuid.UUID, w ObserverID: observerID, Column2: sinceTS, Column3: binWidth, + Until: pgtype.Timestamptz{Time: until, Valid: true}, }) if err != nil { return nil, err @@ -268,6 +290,7 @@ func (s *Store) GetObserverActivity(ctx context.Context, observerID uuid.UUID, w activity.Points = make([]api.ObserverActivityPoint, 0, len(rows)) for _, r := range rows { p := api.ObserverActivityPoint{T: r.Bucket.Time.UnixMilli(), Observations: r.Observations} + activity.Summary.RecordedPackets += r.Observations if r.AirtimeN > 0 { airtime := r.AirtimeMs p.AirtimeMs = &airtime @@ -275,7 +298,12 @@ func (s *Store) GetObserverActivity(ctx context.Context, observerID uuid.UUID, w if r.SnrN > 0 { avg := r.SnrSum / float32(r.SnrN) min := r.SnrMin - p.SNRAvg, p.SNRMin = &avg, &min + if !math.IsNaN(float64(avg)) && !math.IsInf(float64(avg), 0) { + p.SNRAvg = &avg + } + if !math.IsNaN(float64(min)) && !math.IsInf(float64(min), 0) { + p.SNRMin = &min + } } if r.RssiN > 0 { avg := float32(r.RssiSum) / float32(r.RssiN) @@ -286,18 +314,17 @@ func (s *Store) GetObserverActivity(ctx context.Context, observerID uuid.UUID, w typeRows, err := s.q.GetObserverActivityHourlyPayloadTypes(ctx, sqlc.GetObserverActivityHourlyPayloadTypesParams{ ObserverID: observerID, Column2: sinceTS, + Until: pgtype.Timestamptz{Time: until, Valid: true}, }) if err != nil { return nil, err } activity.PayloadTypes = make([]api.PayloadBreakdownItem, 0, len(typeRows)) for _, v := range typeRows { - if v.PayloadType == nil { - continue - } + payloadType := v.PayloadType activity.PayloadTypes = append(activity.PayloadTypes, api.PayloadBreakdownItem{ - PayloadType: *v.PayloadType, - PayloadTypeName: api.PayloadTypeName(*v.PayloadType), + PayloadType: payloadType, + PayloadTypeName: api.PayloadTypeName(payloadType), Count: v.Count, }) } @@ -308,6 +335,7 @@ func (s *Store) GetObserverActivity(ctx context.Context, observerID uuid.UUID, w ObserverID: observerID, Column2: sinceTS, Column3: binWidth, + Until: pgtype.Timestamptz{Time: until, Valid: true}, }) if err != nil { return nil, err @@ -315,13 +343,19 @@ func (s *Store) GetObserverActivity(ctx context.Context, observerID uuid.UUID, w activity.Points = make([]api.ObserverActivityPoint, 0, len(rows)) for _, r := range rows { p := api.ObserverActivityPoint{T: r.Bucket.Time.UnixMilli(), Observations: r.Observations} + activity.Summary.RecordedPackets += r.Observations if r.AirtimeN > 0 { airtime := r.AirtimeMs p.AirtimeMs = &airtime } if r.SnrN > 0 { avg, min := r.SnrAvg, r.SnrMin - p.SNRAvg, p.SNRMin = &avg, &min + if !math.IsNaN(float64(avg)) && !math.IsInf(float64(avg), 0) { + p.SNRAvg = &avg + } + if !math.IsNaN(float64(min)) && !math.IsInf(float64(min), 0) { + p.SNRMin = &min + } } if r.RssiN > 0 { avg := r.RssiAvg @@ -332,18 +366,20 @@ func (s *Store) GetObserverActivity(ctx context.Context, observerID uuid.UUID, w typeRows, err := s.q.GetObserverActivityRawPayloadTypes(ctx, sqlc.GetObserverActivityRawPayloadTypesParams{ ObserverID: observerID, Column2: sinceTS, + Until: pgtype.Timestamptz{Time: until, Valid: true}, }) if err != nil { return nil, err } activity.PayloadTypes = make([]api.PayloadBreakdownItem, 0, len(typeRows)) for _, v := range typeRows { - if v.PayloadType == nil { - continue + payloadType := int16(-1) + if v.PayloadType != nil { + payloadType = *v.PayloadType } activity.PayloadTypes = append(activity.PayloadTypes, api.PayloadBreakdownItem{ - PayloadType: *v.PayloadType, - PayloadTypeName: api.PayloadTypeName(*v.PayloadType), + PayloadType: payloadType, + PayloadTypeName: api.PayloadTypeName(payloadType), Count: v.Count, }) } @@ -424,10 +460,11 @@ func (s *Store) GetObserverRadio(ctx context.Context, observerID uuid.UUID) (ing return settings, nil } -func (s *Store) UpsertObserverBroker(ctx context.Context, observerID uuid.UUID, brokerName string) error { +func (s *Store) UpsertObserverBroker(ctx context.Context, observerID uuid.UUID, brokerName string, isPacket bool) error { params := sqlc.UpsertObserverBrokerParams{ ObserverID: observerID, BrokerName: brokerName, + IsPacket: isPacket, } return s.q.UpsertObserverBroker(ctx, params) } diff --git a/db/observers_test.go b/db/observers_test.go index 76ccf548..f6c67d3d 100644 --- a/db/observers_test.go +++ b/db/observers_test.go @@ -569,6 +569,7 @@ func f32(v float32) *float32 { return &v } func TestGetObserverActivity_HourlyFoldNoSignal(t *testing.T) { ctrl := gomock.NewController(t) mock := mockdb.NewMockQuerier(ctrl) + mock.EXPECT().GetObserverActivityLiveSummary(gomock.Any(), gomock.Any()).Return(sqlc.GetObserverActivityLiveSummaryRow{}, nil).AnyTimes() observerID := uuid.MustParse("00000000-0000-0000-0000-000000000001") mock.EXPECT().GetObserverByID(gomock.Any(), observerID). @@ -587,12 +588,12 @@ func TestGetObserverActivity_HourlyFoldNoSignal(t *testing.T) { }}, nil) mock.EXPECT().GetObserverActivityHourlyPayloadTypes(gomock.Any(), gomock.Any()). Return([]sqlc.GetObserverActivityHourlyPayloadTypesRow{ - {PayloadType: i16(4), Count: 9}, - {PayloadType: nil, Count: 3}, + {PayloadType: 4, Count: 9}, + {PayloadType: -1, Count: 3}, }, nil) store := &Store{q: mock} - got, err := store.GetObserverActivity(context.Background(), observerID, 24*time.Hour, time.Hour) + got, err := store.GetObserverActivity(context.Background(), observerID, 24*time.Hour, time.Hour, time.Time{}) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -615,8 +616,8 @@ func TestGetObserverActivity_HourlyFoldNoSignal(t *testing.T) { if p.RSSIAvg != nil { t.Errorf("expected nil RSSIAvg, got %v", *p.RSSIAvg) } - if len(got.PayloadTypes) != 1 || got.PayloadTypes[0].PayloadType != 4 || got.PayloadTypes[0].Count != 9 { - t.Fatalf("expected one payload type 4 with count 9, got %+v", got.PayloadTypes) + if len(got.PayloadTypes) != 2 || got.PayloadTypes[0].PayloadType != 4 || got.PayloadTypes[0].Count != 9 || got.PayloadTypes[1].PayloadType != -1 || got.PayloadTypes[1].Count != 3 { + t.Fatalf("expected known and unknown payloads to reconcile, got %+v", got.PayloadTypes) } if got.PayloadTypes[0].PayloadTypeName != api.PayloadTypeName(4) { t.Errorf("expected payload type name %q, got %q", api.PayloadTypeName(4), got.PayloadTypes[0].PayloadTypeName) @@ -638,6 +639,7 @@ func TestGetObserverActivity_HourlyFoldNoSignal(t *testing.T) { func TestGetObserverActivity_HourlyFoldWeightedAverages(t *testing.T) { ctrl := gomock.NewController(t) mock := mockdb.NewMockQuerier(ctrl) + mock.EXPECT().GetObserverActivityLiveSummary(gomock.Any(), gomock.Any()).Return(sqlc.GetObserverActivityLiveSummaryRow{}, nil).AnyTimes() observerID := uuid.MustParse("00000000-0000-0000-0000-000000000001") mock.EXPECT().GetObserverByID(gomock.Any(), observerID). @@ -658,7 +660,7 @@ func TestGetObserverActivity_HourlyFoldWeightedAverages(t *testing.T) { Return([]sqlc.GetObserverActivityHourlyPayloadTypesRow{}, nil) store := &Store{q: mock} - got, err := store.GetObserverActivity(context.Background(), observerID, 7*24*time.Hour, 6*time.Hour) + got, err := store.GetObserverActivity(context.Background(), observerID, 7*24*time.Hour, 6*time.Hour, time.Time{}) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -689,6 +691,7 @@ func TestGetObserverActivity_HourlyFoldWeightedAverages(t *testing.T) { func TestGetObserverActivity_RawPath(t *testing.T) { ctrl := gomock.NewController(t) mock := mockdb.NewMockQuerier(ctrl) + mock.EXPECT().GetObserverActivityLiveSummary(gomock.Any(), gomock.Any()).Return(sqlc.GetObserverActivityLiveSummaryRow{}, nil).AnyTimes() observerID := uuid.MustParse("00000000-0000-0000-0000-000000000001") mock.EXPECT().GetObserverByID(gomock.Any(), observerID). @@ -709,7 +712,7 @@ func TestGetObserverActivity_RawPath(t *testing.T) { Return([]sqlc.GetObserverActivityRawPayloadTypesRow{{PayloadType: i16(1), Count: 5}}, nil) store := &Store{q: mock} - got, err := store.GetObserverActivity(context.Background(), observerID, 6*time.Hour, 15*time.Minute) + got, err := store.GetObserverActivity(context.Background(), observerID, 6*time.Hour, 15*time.Minute, time.Time{}) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -737,6 +740,7 @@ func TestGetObserverActivity_RawPath(t *testing.T) { func TestGetObserverActivity_RawFoldNoSignal(t *testing.T) { ctrl := gomock.NewController(t) mock := mockdb.NewMockQuerier(ctrl) + mock.EXPECT().GetObserverActivityLiveSummary(gomock.Any(), gomock.Any()).Return(sqlc.GetObserverActivityLiveSummaryRow{}, nil).AnyTimes() observerID := uuid.MustParse("00000000-0000-0000-0000-000000000001") mock.EXPECT().GetObserverByID(gomock.Any(), observerID). @@ -757,7 +761,7 @@ func TestGetObserverActivity_RawFoldNoSignal(t *testing.T) { Return([]sqlc.GetObserverActivityRawPayloadTypesRow{}, nil) store := &Store{q: mock} - got, err := store.GetObserverActivity(context.Background(), observerID, 6*time.Hour, 15*time.Minute) + got, err := store.GetObserverActivity(context.Background(), observerID, 6*time.Hour, 15*time.Minute, time.Time{}) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -779,6 +783,7 @@ func TestGetObserverActivity_RawFoldNoSignal(t *testing.T) { func TestGetObserverActivity_RawPathEmpty(t *testing.T) { ctrl := gomock.NewController(t) mock := mockdb.NewMockQuerier(ctrl) + mock.EXPECT().GetObserverActivityLiveSummary(gomock.Any(), gomock.Any()).Return(sqlc.GetObserverActivityLiveSummaryRow{}, nil).AnyTimes() observerID := uuid.MustParse("00000000-0000-0000-0000-000000000001") mock.EXPECT().GetObserverByID(gomock.Any(), observerID). @@ -789,7 +794,7 @@ func TestGetObserverActivity_RawPathEmpty(t *testing.T) { Return([]sqlc.GetObserverActivityRawPayloadTypesRow{}, nil) store := &Store{q: mock} - got, err := store.GetObserverActivity(context.Background(), observerID, 6*time.Hour, 15*time.Minute) + got, err := store.GetObserverActivity(context.Background(), observerID, 6*time.Hour, 15*time.Minute, time.Time{}) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -804,6 +809,7 @@ func TestGetObserverActivity_RawPathEmpty(t *testing.T) { func TestGetObserverActivity_HourlyPathAtOneHour(t *testing.T) { ctrl := gomock.NewController(t) mock := mockdb.NewMockQuerier(ctrl) + mock.EXPECT().GetObserverActivityLiveSummary(gomock.Any(), gomock.Any()).Return(sqlc.GetObserverActivityLiveSummaryRow{}, nil).AnyTimes() observerID := uuid.MustParse("00000000-0000-0000-0000-000000000001") mock.EXPECT().GetObserverByID(gomock.Any(), observerID). @@ -814,7 +820,7 @@ func TestGetObserverActivity_HourlyPathAtOneHour(t *testing.T) { Return([]sqlc.GetObserverActivityHourlyPayloadTypesRow{}, nil) store := &Store{q: mock} - got, err := store.GetObserverActivity(context.Background(), observerID, 24*time.Hour, time.Hour) + got, err := store.GetObserverActivity(context.Background(), observerID, 24*time.Hour, time.Hour, time.Time{}) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -826,13 +832,14 @@ func TestGetObserverActivity_HourlyPathAtOneHour(t *testing.T) { func TestGetObserverActivity_UnknownObserver(t *testing.T) { ctrl := gomock.NewController(t) mock := mockdb.NewMockQuerier(ctrl) + mock.EXPECT().GetObserverActivityLiveSummary(gomock.Any(), gomock.Any()).Return(sqlc.GetObserverActivityLiveSummaryRow{}, nil).AnyTimes() observerID := uuid.MustParse("00000000-0000-0000-0000-000000000002") mock.EXPECT().GetObserverByID(gomock.Any(), observerID). Return(sqlc.Observer{}, pgx.ErrNoRows) store := &Store{q: mock} - got, err := store.GetObserverActivity(context.Background(), observerID, time.Hour, time.Hour) + got, err := store.GetObserverActivity(context.Background(), observerID, time.Hour, time.Hour, time.Time{}) if !errors.Is(err, pgx.ErrNoRows) { t.Fatalf("expected pgx.ErrNoRows, got %v", err) } @@ -858,6 +865,7 @@ func TestGetObserverActivity_RadioNilWhenIncomplete(t *testing.T) { t.Run(tc.name, func(t *testing.T) { ctrl := gomock.NewController(t) mock := mockdb.NewMockQuerier(ctrl) + mock.EXPECT().GetObserverActivityLiveSummary(gomock.Any(), gomock.Any()).Return(sqlc.GetObserverActivityLiveSummaryRow{}, nil).AnyTimes() observerID := uuid.MustParse("00000000-0000-0000-0000-000000000001") mock.EXPECT().GetObserverByID(gomock.Any(), observerID). @@ -868,7 +876,7 @@ func TestGetObserverActivity_RadioNilWhenIncomplete(t *testing.T) { Return([]sqlc.GetObserverActivityHourlyPayloadTypesRow{}, nil) store := &Store{q: mock} - got, err := store.GetObserverActivity(context.Background(), observerID, 24*time.Hour, time.Hour) + got, err := store.GetObserverActivity(context.Background(), observerID, 24*time.Hour, time.Hour, time.Time{}) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -882,6 +890,7 @@ func TestGetObserverActivity_RadioNilWhenIncomplete(t *testing.T) { func TestGetObserverActivity_SinceAlignedToInterval(t *testing.T) { ctrl := gomock.NewController(t) mock := mockdb.NewMockQuerier(ctrl) + mock.EXPECT().GetObserverActivityLiveSummary(gomock.Any(), gomock.Any()).Return(sqlc.GetObserverActivityLiveSummaryRow{}, nil).AnyTimes() observerID := uuid.MustParse("00000000-0000-0000-0000-000000000001") window, interval := 24*time.Hour, 6*time.Hour @@ -897,12 +906,12 @@ func TestGetObserverActivity_SinceAlignedToInterval(t *testing.T) { mock.EXPECT().GetObserverActivityHourlyPayloadTypes(gomock.Any(), gomock.Any()). Return([]sqlc.GetObserverActivityHourlyPayloadTypesRow{}, nil) - before := time.Now().Add(-window) + before := time.Now().UTC().Truncate(interval).Add(-window) store := &Store{q: mock} - if _, err := store.GetObserverActivity(context.Background(), observerID, window, interval); err != nil { + if _, err := store.GetObserverActivity(context.Background(), observerID, window, interval, time.Time{}); err != nil { t.Fatalf("unexpected error: %v", err) } - after := time.Now().Add(-window) + after := time.Now().UTC().Truncate(interval).Add(-window) if gotSince.UnixNano()%int64(interval) != 0 { t.Errorf("since %s is not aligned to %s", gotSince, interval) @@ -910,7 +919,7 @@ func TestGetObserverActivity_SinceAlignedToInterval(t *testing.T) { if gotSince.Before(before) { t.Errorf("since %s is before the window start %s", gotSince, before) } - if !gotSince.Before(after.Add(interval)) { + if gotSince.After(after.Add(interval)) { t.Errorf("since %s is more than one interval past the window start %s", gotSince, after) } if !gotInterval.Valid || gotInterval.Microseconds != interval.Microseconds() { diff --git a/db/presence.go b/db/presence.go index ea2d6fea..41222669 100644 --- a/db/presence.go +++ b/db/presence.go @@ -20,11 +20,12 @@ func (s *Store) TouchObservers(ctx context.Context, ids []uuid.UUID, seen []time }) } -func (s *Store) TouchObserverBrokers(ctx context.Context, ids []uuid.UUID, brokers []string, seen []time.Time) error { +func (s *Store) TouchObserverBrokers(ctx context.Context, ids []uuid.UUID, brokers []string, seen, packets []time.Time) error { return s.q.TouchObserverBrokers(ctx, sqlc.TouchObserverBrokersParams{ Column1: ids, Column2: brokers, Column3: toTimestamptzs(seen), + Column4: nullableTimes(packets), }) } @@ -42,3 +43,11 @@ func toTimestamptzs(ts []time.Time) []pgtype.Timestamptz { } return out } + +func nullableTimes(ts []time.Time) []pgtype.Timestamptz { + out := make([]pgtype.Timestamptz, len(ts)) + for i, t := range ts { + out[i] = pgtype.Timestamptz{Time: t, Valid: !t.IsZero()} + } + return out +} diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 29690a16..05ceacde 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -275,14 +275,14 @@ SELECT COALESCE(AVG(rssi) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0)), 0)::real AS rssi_avg, COUNT(rssi) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::bigint AS rssi_n FROM packet_observations -WHERE observer_id = $1 AND heard_at >= $2::timestamptz +WHERE observer_id = $1 AND heard_at >= $2::timestamptz AND heard_at < @until::timestamptz GROUP BY bucket ORDER BY bucket; -- name: GetObserverActivityRawPayloadTypes :many SELECT payload_type, COUNT(*)::bigint AS count FROM packet_observations -WHERE observer_id = $1 AND heard_at >= $2::timestamptz AND payload_type IS NOT NULL +WHERE observer_id = $1 AND heard_at >= $2::timestamptz AND heard_at < @until::timestamptz GROUP BY payload_type ORDER BY count DESC; @@ -299,14 +299,14 @@ SELECT COALESCE(SUM(rssi_sum), 0)::bigint AS rssi_sum, SUM(rssi_n)::bigint AS rssi_n FROM mv_observer_activity_hourly -WHERE observer_id = $1 AND bucket >= $2::timestamptz +WHERE observer_id = $1 AND bucket >= $2::timestamptz AND bucket < @until::timestamptz GROUP BY 1 ORDER BY 1; -- name: GetObserverActivityHourlyPayloadTypes :many SELECT payload_type, SUM(observations)::bigint AS count FROM mv_observer_activity_hourly -WHERE observer_id = $1 AND bucket >= $2::timestamptz +WHERE observer_id = $1 AND bucket >= $2::timestamptz AND bucket < @until::timestamptz GROUP BY payload_type ORDER BY count DESC; @@ -363,19 +363,20 @@ RETURNING o.id; -- name: UpsertObserverBroker :exec INSERT INTO observer_brokers (observer_id, broker_name, last_seen, last_packet_at) -VALUES ($1, $2, NOW(), NOW()) +VALUES ($1, $2, NOW(), CASE WHEN @is_packet::boolean THEN NOW() END) ON CONFLICT (observer_id, broker_name) DO UPDATE SET - last_seen = NOW(), - last_packet_at = NOW(); + last_seen = NOW(), + last_packet_at = COALESCE(EXCLUDED.last_packet_at, observer_brokers.last_packet_at); -- name: TouchObserverBrokers :exec UPDATE observer_brokers ob SET last_seen = GREATEST(ob.last_seen, v.seen), - last_packet_at = GREATEST(ob.last_packet_at, v.seen) + last_packet_at = GREATEST(ob.last_packet_at, v.packet) FROM ( SELECT unnest($1::uuid[]) AS observer_id, unnest($2::text[]) AS broker_name, - unnest($3::timestamptz[]) AS seen + unnest($3::timestamptz[]) AS seen, + unnest($4::timestamptz[]) AS packet ) v WHERE ob.observer_id = v.observer_id AND ob.broker_name = v.broker_name; @@ -1571,3 +1572,17 @@ WITH target AS MATERIALIZED ( ) SELECT EXISTS(SELECT 1 FROM target) AS found, EXISTS(SELECT 1 FROM changed) AS deactivated; + +-- name: GetObserverActivityLiveSummary :one +-- Two indexed ranges, bounded to one observer; no legacy presence counters. +WITH latest AS ( + SELECT heard_at FROM packet_observations + WHERE observer_id = @observer_id::uuid AND heard_at <= @generated_at::timestamptz + ORDER BY heard_at DESC LIMIT 1 +), hourly AS ( + SELECT COUNT(*)::bigint AS n FROM packet_observations + WHERE observer_id = @observer_id::uuid + AND heard_at >= @hour_start::timestamptz AND heard_at < @hour_end::timestamptz +) +SELECT (SELECT heard_at FROM latest)::timestamptz AS latest_recorded_at, + hourly.n AS last_complete_hour FROM hourly; diff --git a/db/queries/route_evidence.sql b/db/queries/route_evidence.sql new file mode 100644 index 00000000..dcc056fd --- /dev/null +++ b/db/queries/route_evidence.sql @@ -0,0 +1,23 @@ +-- name: GetRouteEvidenceRoute :one +SELECT * FROM known_routes +WHERE iata = @iata::bpchar AND path_key = @path_key::bytea; + +-- name: ListRouteEvidence :many +-- Leading index equalities and the time/ID boundary bound both custom and generic plans. +-- Full-byte equality is required even when the compact digest matches. TRACE path bytes +-- carry readings; unclassified legacy observations cannot be safely called ordinary paths. +SELECT po.id, po.packet_hash, po.observer_id, o.display_name AS observer_name, + po.heard_at, po.payload_type, po.rssi, po.snr +FROM packet_observations po +JOIN observers o ON o.id = po.observer_id +WHERE po.iata = @iata::bpchar + AND po.hash_size = @hash_size::smallint + AND decode(md5(po.path_bytes), 'hex') = @path_digest::bytea + AND po.path_bytes = @path_bytes::bytea + AND po.hop_count >= 2 + AND po.hop_count = @hop_count::smallint + AND po.path_bytes IS NOT NULL AND po.payload_type IS NOT NULL AND po.payload_type <> 9 + AND po.heard_at >= @since::timestamptz AND po.heard_at < @until::timestamptz + AND (po.heard_at, po.id) < (@before_at::timestamptz, @before_id::bigint) +ORDER BY po.heard_at DESC, po.id DESC +LIMIT @page_limit::integer; diff --git a/db/route_evidence.go b/db/route_evidence.go new file mode 100644 index 00000000..fc7209e0 --- /dev/null +++ b/db/route_evidence.go @@ -0,0 +1,93 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later +package db + +import ( + "bytes" + "context" + "crypto/md5" + "encoding/hex" + "math" + "time" + + sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" + "github.com/MeshCore-Beacon/beacon-server/internal/api" + "github.com/jackc/pgx/v5/pgtype" +) + +func savedRoutePath(prefixes [][]byte, count int32) (int16, []byte, bool) { + if count < 2 || count > 63 || len(prefixes) != int(count) { + return 0, nil, false + } + width := len(prefixes[0]) + if width < 1 || width > 3 { + return 0, nil, false + } + for _, p := range prefixes { + if len(p) != width { + return 0, nil, false + } + } + return int16(width), bytes.Join(prefixes, nil), true +} + +func (s *Store) GetRouteEvidence(ctx context.Context, iata, key string, query api.RouteEvidenceQuery) (*api.RouteEvidence, error) { + if !api.ValidRouteEvidenceKey(iata, key) || !api.ValidRouteEvidenceWindow(query.Since, query.Until) || query.Limit < 1 || query.Limit > 200 { + return nil, api.ErrRouteEvidenceInput + } + if c := query.Cursor; c != nil { + if c.IATA != iata || c.PathKey != key || !c.Since.Equal(query.Since) || !c.Until.Equal(query.Until) || c.ID <= 0 || c.HeardAt.Before(query.Since) || !c.HeardAt.Before(query.Until) { + return nil, api.ErrRouteEvidenceInput + } + } + keyBytes, _ := hex.DecodeString(key) + row, err := s.q.GetRouteEvidenceRoute(ctx, sqlc.GetRouteEvidenceRouteParams{Iata: iata, PathKey: keyBytes}) + if err != nil { + return nil, err + } + nodes, err := s.GetNodesByIDs(ctx, row.NodeIds) + if err != nil { + return nil, err + } + route := toKnownRoutes([]knownRouteRow{{ID: row.ID, NodeIds: row.NodeIds, HashPrefix: row.HashPrefix, Iata: row.Iata, HopCount: row.HopCount, FirstSeen: row.FirstSeen, LastSeen: row.LastSeen, ObservationCount: row.ObservationCount}}, nodes)[0] + route.PathKey = hex.EncodeToString(row.PathKey) + out := &api.RouteEvidence{Page: api.Page[api.RouteObservation]{Items: []api.RouteObservation{}}, Route: route, WindowStart: query.Since.UnixMilli(), WindowEnd: query.Until.UnixMilli(), GeneratedAt: time.Now().UnixMilli(), MatchType: "saved_path_prefixes"} + width, path, valid := savedRoutePath(row.HashPrefix, row.HopCount) + if !valid { + return out, nil + } + out.MatchAvailable = true + out.HashSize = width + out.PathBytes = hex.EncodeToString(path) + beforeAt, beforeID := query.Until, int64(math.MaxInt64) + if query.Cursor != nil { + beforeAt, beforeID = query.Cursor.HeardAt, query.Cursor.ID + } + ts := func(t time.Time) pgtype.Timestamptz { return pgtype.Timestamptz{Time: t, Valid: true} } + digest := md5.Sum(path) + rows, err := s.q.ListRouteEvidence(ctx, sqlc.ListRouteEvidenceParams{Iata: iata, HashSize: width, HopCount: int16(row.HopCount), PathDigest: digest[:], PathBytes: path, Since: ts(query.Since), Until: ts(query.Until), BeforeAt: ts(beforeAt), BeforeID: beforeID, PageLimit: query.Limit + 1}) + if err != nil { + return nil, err + } + out.HasMore = len(rows) > int(query.Limit) + if out.HasMore { + rows = rows[:query.Limit] + } + for _, r := range rows { + payload := int16(-1) + if r.PayloadType != nil { + payload = *r.PayloadType + } + snr := r.Snr + if snr != nil && (math.IsNaN(float64(*snr)) || math.IsInf(float64(*snr), 0)) { + snr = nil + } + out.Items = append(out.Items, api.RouteObservation{ID: r.ID, PacketHash: hex.EncodeToString(r.PacketHash), ObserverID: r.ObserverID, ObserverName: r.ObserverName, HeardAt: r.HeardAt.Time.UnixMilli(), PayloadType: payload, PayloadTypeName: api.PayloadTypeName(payload), RSSI: r.Rssi, SNR: snr}) + } + if out.HasMore { + last := rows[len(rows)-1] + cursor := (api.RouteEvidenceCursor{IATA: iata, PathKey: key, Since: query.Since, Until: query.Until, HeardAt: last.HeardAt.Time, ID: last.ID}).String() + out.NextPageCursor = &cursor + } + return out, nil +} diff --git a/db/route_evidence_integration_test.go b/db/route_evidence_integration_test.go new file mode 100644 index 00000000..f4a3e177 --- /dev/null +++ b/db/route_evidence_integration_test.go @@ -0,0 +1,214 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later +package db + +import ( + "context" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "os" + "strings" + "testing" + "time" + + sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" + "github.com/MeshCore-Beacon/beacon-server/internal/api" + "github.com/google/uuid" + "github.com/jackc/pgx/v5" +) + +type evidenceCapture struct { + pgx.Tx + statement string + args []any +} + +func (c *evidenceCapture) Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error) { + if strings.Contains(sql, "-- name: ListRouteEvidence") { + c.statement = sql + c.args = args + } + return c.Tx.Query(ctx, sql, args...) +} + +func TestRouteEvidencePostgres(t *testing.T) { + ctx, tx := retentionTx(t) + analyticsTables(t, ctx, tx) + ddl, err := migrationFiles.ReadFile("migrations/024_known_routes_pathkey.sql") + if err != nil { + t.Fatal(err) + } + text := string(ddl) + start := strings.Index(text, "CREATE TABLE known_routes_new (") + end := start + strings.Index(text[start:], "\n);") + 3 + table := strings.ReplaceAll(text[start:end], "known_routes_new", "known_routes") + _, err = tx.Exec(ctx, `CREATE TABLE iata_codes (iata char(3) PRIMARY KEY); INSERT INTO iata_codes VALUES ('YOW'),('YVR'); + ALTER TABLE nodes ADD COLUMN latitude double precision, ADD COLUMN longitude double precision;`+table+` + INSERT INTO observers(id,display_name) VALUES ('00000000-0000-0000-0000-000000000001','One'),('00000000-0000-0000-0000-000000000002','Two'); + INSERT INTO nodes(id,public_key,name) VALUES ('00000000-0000-0000-0000-000000000001','\xaa','A'),('00000000-0000-0000-0000-000000000002','\xbb','B'); + INSERT INTO packets(packet_hash) SELECT int4send(i) FROM generate_series(1,10) i;`) + if err != nil { + t.Fatal(err) + } + index, err := migrationFiles.ReadFile("migrations/041_route_evidence_index.sql") + if err != nil { + t.Fatal(err) + } + if _, err := tx.Exec(ctx, strings.Replace(string(index), "CREATE INDEX CONCURRENTLY", "CREATE INDEX", 1)); err != nil { + t.Fatal(err) + } + capture := &evidenceCapture{Tx: tx} + store := &Store{q: sqlc.New(capture)} + ids := []uuid.UUID{uuid.MustParse("00000000-0000-0000-0000-000000000001"), uuid.MustParse("00000000-0000-0000-0000-000000000002")} + if err := store.UpsertKnownRoute(ctx, ids, [][]byte{{0xaa}, {0xbb}}, "YOW", 2); err != nil { + t.Fatal(err) + } + key := hex.EncodeToString(routePathKey(ids)) + since := time.Now().UTC().Truncate(time.Millisecond).Add(-time.Hour) + until := since.Add(time.Hour) + at := since.Add(time.Minute + 123456*time.Microsecond) + _, err = tx.Exec(ctx, `INSERT INTO packet_observations(id,packet_hash,observer_id,iata,heard_at,hash_size,hop_count,path_bytes,payload_type,snr) + SELECT i,int4send(i), '00000000-0000-0000-0000-000000000001',CASE WHEN i=5 THEN 'YVR' ELSE 'YOW' END, + CASE WHEN i=9 THEN $2::timestamptz WHEN i=1 THEN $1::timestamptz-interval '2 microseconds' ELSE $1::timestamptz END, + CASE WHEN i=4 THEN 2 ELSE 1 END,CASE WHEN i=4 THEN 1 WHEN i=10 THEN 3 ELSE 2 END, + CASE WHEN i=6 THEN '\xabbb'::bytea WHEN i=10 THEN '\xaabbcc'::bytea ELSE '\xaabb'::bytea END, + CASE WHEN i=7 THEN 9 WHEN i=8 THEN NULL ELSE 4 END,CASE WHEN i=1 THEN 'NaN'::real ELSE 0 END FROM generate_series(1,10) i; + `, at, until) + if err != nil { + t.Fatal(err) + } + if _, err = tx.Exec(ctx, `INSERT INTO packet_observations(id,packet_hash,observer_id,iata,heard_at,hash_size,hop_count,path_bytes,payload_type) VALUES (11,int4send(1),'00000000-0000-0000-0000-000000000002','YOW',$1,1,2,'\xaabb',4)`, at.Add(-time.Microsecond)); err != nil { + t.Fatal(err) + } + query := api.RouteEvidenceQuery{Since: since, Until: until, Limit: 2} + first, err := store.GetRouteEvidence(ctx, "YOW", key, query) + if err != nil { + t.Fatal(err) + } + if !first.HasMore || first.NextPageCursor == nil || len(first.Items) != 2 || first.Items[0].ID != 3 || first.Items[1].ID != 2 || first.Route.PathKey != key || first.HashSize != 1 || first.PathBytes != "aabb" { + t.Fatalf("first page: %+v", first) + } + query.Cursor, err = api.ParseRouteEvidenceCursor(*first.NextPageCursor) + if err != nil { + t.Fatal(err) + } + second, err := store.GetRouteEvidence(ctx, "YOW", key, query) + if err != nil { + t.Fatal(err) + } + if second.HasMore || len(second.Items) != 2 || second.Items[0].ID != 11 || second.Items[1].ID != 1 || second.Items[1].SNR != nil { + t.Fatalf("precise second page: %+v", second) + } + if _, err = json.Marshal(second); err != nil { + t.Fatalf("non-finite JSON: %v", err) + } + query.Cursor.IATA = "YVR" + if _, err = store.GetRouteEvidence(ctx, "YOW", key, query); !errors.Is(err, api.ErrRouteEvidenceInput) { + t.Fatal("cross-route cursor accepted") + } + query.Cursor = nil + if _, err = store.GetRouteEvidence(ctx, "YVR", key, query); !errors.Is(err, pgx.ErrNoRows) { + t.Fatal("unknown route did not return not-found") + } + if _, err = tx.Exec(ctx, "DELETE FROM packets"); err != nil { + t.Fatal(err) + } + empty, err := store.GetRouteEvidence(ctx, "YOW", key, query) + if err != nil || !empty.MatchAvailable || len(empty.Items) != 0 || empty.HasMore { + t.Fatalf("expired evidence: %+v %v", empty, err) + } + + // Many unrelated paths must not turn this into a raw-table or route-ID scan. + started := time.Now() + _, err = tx.Exec(ctx, `INSERT INTO packets(packet_hash) SELECT int4send(i) FROM generate_series(100,200099) i;`) + if err != nil { + t.Fatal(err) + } + _, err = tx.Exec(ctx, `INSERT INTO packet_observations(packet_hash,observer_id,iata,heard_at,hash_size,hop_count,path_bytes,payload_type) + SELECT packet_hash,'00000000-0000-0000-0000-000000000001','YOW',$1,1,2, + CASE WHEN get_byte(packet_hash,3)=0 THEN '\xaabb'::bytea ELSE packet_hash END,4 FROM packets;`, at) + if err != nil { + t.Fatal(err) + } + if _, err = tx.Exec(ctx, "ANALYZE packet_observations; ANALYZE observers;"); err != nil { + t.Fatal(err) + } + for _, mode := range []string{"force_custom_plan", "force_generic_plan"} { + if _, err = tx.Exec(ctx, "SET LOCAL plan_cache_mode = "+mode); err != nil { + t.Fatal(err) + } + result, err := store.GetRouteEvidence(ctx, "YOW", key, query) + if err != nil || len(result.Items) != 2 { + t.Fatalf("indexed read: %v", err) + } + var plan string + if err = tx.QueryRow(ctx, "EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) "+capture.statement, capture.args...).Scan(&plan); err != nil { + t.Fatal(err) + } + if !strings.Contains(plan, "idx_observations_route_evidence") { + t.Fatalf("missing evidence index: %s", plan) + } + t.Logf("%s plan: %s", mode, plan) + } + t.Logf("200000-row fixture insert/analyze/read: %s", time.Since(started)) + if _, err = tx.Exec(ctx, `UPDATE known_routes SET hash_prefix=ARRAY['\xaa'::bytea]`); err != nil { + t.Fatal(err) + } + unavailable, err := store.GetRouteEvidence(ctx, "YOW", key, query) + if err != nil || unavailable.MatchAvailable || len(unavailable.Items) != 0 { + t.Fatalf("malformed stored path: %+v %v", unavailable, err) + } +} + +func TestRouteEvidenceIndexPostgres(t *testing.T) { + dsn := os.Getenv("BEACON_TEST_POSTGRES_DSN") + if dsn == "" { + t.Skip("set BEACON_TEST_POSTGRES_DSN") + } + ctx, cancel := context.WithTimeout(context.Background(), time.Minute) + defer cancel() + conn, err := pgx.Connect(ctx, dsn) + if err != nil { + t.Fatal(err) + } + defer conn.Close(context.Background()) + schema := fmt.Sprintf("route_evidence_%d", time.Now().UnixNano()) + ident := pgx.Identifier{schema}.Sanitize() + if _, err = conn.Exec(ctx, "CREATE SCHEMA "+ident+"; SET search_path TO "+ident); err != nil { + t.Fatal(err) + } + defer func() { _, _ = conn.Exec(context.Background(), "DROP SCHEMA "+ident+" CASCADE") }() + if _, err = conn.Exec(ctx, `CREATE TABLE packet_observations (id bigint,iata char(3),hash_size smallint,path_bytes bytea,heard_at timestamptz,payload_type smallint,hop_count smallint)`); err != nil { + t.Fatal(err) + } + migration, err := migrationFiles.ReadFile("migrations/041_route_evidence_index.sql") + if err != nil { + t.Fatal(err) + } + for i := 0; i < 2; i++ { + if err = applyMigration(ctx, conn, string(migration)); err != nil { + t.Fatalf("migration/retry: %v", err) + } + } + if _, err = conn.Exec(ctx, `UPDATE pg_index SET indisvalid=false WHERE indexrelid='idx_observations_route_evidence'::regclass`); err != nil { + t.Fatal(err) + } + if err = applyMigration(ctx, conn, string(migration)); err != nil { + t.Fatal(err) + } + var valid bool + if err = conn.QueryRow(ctx, `SELECT indisvalid FROM pg_index WHERE indexrelid='idx_observations_route_evidence'::regclass`).Scan(&valid); err != nil || !valid { + t.Fatalf("invalid index after recovery: %v", err) + } + if _, err = conn.Exec(ctx, `INSERT INTO packet_observations(id,iata,hash_size,path_bytes,heard_at,payload_type,hop_count) +SELECT n,'YOW',1,'\xaabb',NOW(),4,n FROM generate_series(0,2) n; +ANALYZE packet_observations`); err != nil { + t.Fatal(err) + } + var indexed float32 + if err = conn.QueryRow(ctx, `SELECT reltuples FROM pg_class WHERE oid='idx_observations_route_evidence'::regclass`).Scan(&indexed); err != nil || indexed != 1 { + t.Fatalf("0/1-hop rows entered the route index: %v %v", indexed, err) + } +} diff --git a/db/route_evidence_test.go b/db/route_evidence_test.go new file mode 100644 index 00000000..868b565b --- /dev/null +++ b/db/route_evidence_test.go @@ -0,0 +1,23 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later +package db + +import ( + "bytes" + "testing" +) + +func TestSavedRoutePath(t *testing.T) { + size, path, ok := savedRoutePath([][]byte{{0xaa}, {0xbb}}, 2) + if !ok || size != 1 || !bytes.Equal(path, []byte{0xaa, 0xbb}) { + t.Fatal("valid path lost") + } + for _, prefixes := range [][][]byte{nil, {{1}}, {{1}, {2, 3}}, {{}, {2}}, {{1, 2, 3, 4}, {5, 6, 7, 8}}} { + if _, _, ok := savedRoutePath(prefixes, int32(len(prefixes))); ok { + t.Fatalf("accepted malformed path %x", prefixes) + } + } + if _, _, ok := savedRoutePath([][]byte{{1}, {2}}, 3); ok { + t.Fatal("accepted inconsistent hop count") + } +} diff --git a/db/routes.go b/db/routes.go index 187e07ec..090d2a00 100644 --- a/db/routes.go +++ b/db/routes.go @@ -118,6 +118,7 @@ func (s *Store) SearchKnownRoutes(ctx context.Context, iata, fromHash, toHash st hops = append(hops, hop) } items = append(items, api.KnownRoute{ + PathKey: hex.EncodeToString(routePathKey(r.NodeIds)), ID: r.ID, IATA: r.Iata, HopCount: int32(len(hops)), @@ -348,6 +349,7 @@ func toKnownRoutes(rows []knownRouteRow, nodes map[uuid.UUID]*api.ResolvedNode) hops = append(hops, hop) } items = append(items, api.KnownRoute{ + PathKey: hex.EncodeToString(routePathKey(r.NodeIds)), ID: r.ID, IATA: r.Iata, HopCount: r.HopCount, diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index aeae62b1..9d3fb184 100644 --- a/db/sqlc/mock/querier.go +++ b/db/sqlc/mock/querier.go @@ -3,7 +3,7 @@ // // Generated by this command: // -// mockgen -source=db/sqlc/querier.go -destination=db/sqlc/mock/querier.go -package=mockdb +// mockgen -source db/sqlc/querier.go -destination db/sqlc/mock/querier.go -package mockdb // // Package mockdb is a generated GoMock package. @@ -384,6 +384,21 @@ func (mr *MockQuerierMockRecorder) GetObserverActivityHourlyPayloadTypes(ctx, ar return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetObserverActivityHourlyPayloadTypes", reflect.TypeOf((*MockQuerier)(nil).GetObserverActivityHourlyPayloadTypes), ctx, arg) } +// GetObserverActivityLiveSummary mocks base method. +func (m *MockQuerier) GetObserverActivityLiveSummary(ctx context.Context, arg db.GetObserverActivityLiveSummaryParams) (db.GetObserverActivityLiveSummaryRow, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetObserverActivityLiveSummary", ctx, arg) + ret0, _ := ret[0].(db.GetObserverActivityLiveSummaryRow) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// GetObserverActivityLiveSummary indicates an expected call of GetObserverActivityLiveSummary. +func (mr *MockQuerierMockRecorder) GetObserverActivityLiveSummary(ctx, arg any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetObserverActivityLiveSummary", reflect.TypeOf((*MockQuerier)(nil).GetObserverActivityLiveSummary), ctx, arg) +} + // GetObserverActivityRaw mocks base method. func (m *MockQuerier) GetObserverActivityRaw(ctx context.Context, arg db.GetObserverActivityRawParams) ([]db.GetObserverActivityRawRow, error) { m.ctrl.T.Helper() @@ -669,6 +684,21 @@ func (mr *MockQuerierMockRecorder) GetRegionIATAs(ctx, regionID any) *gomock.Cal return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRegionIATAs", reflect.TypeOf((*MockQuerier)(nil).GetRegionIATAs), ctx, regionID) } +// GetRouteEvidenceRoute mocks base method. +func (m *MockQuerier) GetRouteEvidenceRoute(ctx context.Context, arg db.GetRouteEvidenceRouteParams) (db.KnownRoute, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetRouteEvidenceRoute", ctx, arg) + ret0, _ := ret[0].(db.KnownRoute) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// GetRouteEvidenceRoute indicates an expected call of GetRouteEvidenceRoute. +func (mr *MockQuerierMockRecorder) GetRouteEvidenceRoute(ctx, arg any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRouteEvidenceRoute", reflect.TypeOf((*MockQuerier)(nil).GetRouteEvidenceRoute), ctx, arg) +} + // GetScopeByName mocks base method. func (m *MockQuerier) GetScopeByName(ctx context.Context, name string) (db.GetScopeByNameRow, error) { m.ctrl.T.Helper() @@ -1208,6 +1238,21 @@ func (mr *MockQuerierMockRecorder) ListRegions(ctx any) *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListRegions", reflect.TypeOf((*MockQuerier)(nil).ListRegions), ctx) } +// ListRouteEvidence mocks base method. +func (m *MockQuerier) ListRouteEvidence(ctx context.Context, arg db.ListRouteEvidenceParams) ([]db.ListRouteEvidenceRow, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ListRouteEvidence", ctx, arg) + ret0, _ := ret[0].([]db.ListRouteEvidenceRow) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// ListRouteEvidence indicates an expected call of ListRouteEvidence. +func (mr *MockQuerierMockRecorder) ListRouteEvidence(ctx, arg any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListRouteEvidence", reflect.TypeOf((*MockQuerier)(nil).ListRouteEvidence), ctx, arg) +} + // ListTraceTags mocks base method. func (m *MockQuerier) ListTraceTags(ctx context.Context, arg db.ListTraceTagsParams) ([]db.ListTraceTagsRow, error) { m.ctrl.T.Helper() diff --git a/db/sqlc/models.go b/db/sqlc/models.go index 8a24d538..47cce44e 100644 --- a/db/sqlc/models.go +++ b/db/sqlc/models.go @@ -244,7 +244,7 @@ type MvHourlyIataStat struct { type MvObserverActivityHourly struct { ObserverID uuid.UUID `json:"observer_id"` - PayloadType *int16 `json:"payload_type"` + PayloadType int16 `json:"payload_type"` Bucket pgtype.Timestamptz `json:"bucket"` Observations int64 `json:"observations"` AirtimeMs float32 `json:"airtime_ms"` diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index 38e6c971..a5770646 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -58,6 +58,8 @@ type Querier interface { // Hour-or-coarser buckets summed from the hourly rollup; same COALESCE-plus-count shape as the raw query. GetObserverActivityHourly(ctx context.Context, arg GetObserverActivityHourlyParams) ([]GetObserverActivityHourlyRow, error) GetObserverActivityHourlyPayloadTypes(ctx context.Context, arg GetObserverActivityHourlyPayloadTypesParams) ([]GetObserverActivityHourlyPayloadTypesRow, error) + // Two indexed ranges, bounded to one observer; no legacy presence counters. + GetObserverActivityLiveSummary(ctx context.Context, arg GetObserverActivityLiveSummaryParams) (GetObserverActivityLiveSummaryRow, error) // Sub-hour activity buckets straight off idx_observations_observer; no join to packets. // Aggregates are COALESCEd and paired with a count column: sqlc types a cast expression as // NOT NULL, so the counts are what tell the store a bucket had no costed or no signal rows. @@ -85,6 +87,7 @@ type Querier interface { GetRegion(ctx context.Context, id int32) (GetRegionRow, error) GetRegionBySlug(ctx context.Context, slug string) (GetRegionBySlugRow, error) GetRegionIATAs(ctx context.Context, regionID int32) ([]string, error) + GetRouteEvidenceRoute(ctx context.Context, arg GetRouteEvidenceRouteParams) (KnownRoute, error) GetScopeByName(ctx context.Context, name string) (GetScopeByNameRow, error) GetScopeNames(ctx context.Context) ([]string, error) // Aggregate matching observations once, separately from node memberships to avoid @@ -196,6 +199,10 @@ type Querier interface { // REGIONS // ============================================================ ListRegions(ctx context.Context) ([]ListRegionsRow, error) + // Leading index equalities and the time/ID boundary bound both custom and generic plans. + // Full-byte equality is required even when the compact digest matches. TRACE path bytes + // carry readings; unclassified legacy observations cannot be safely called ordinary paths. + ListRouteEvidence(ctx context.Context, arg ListRouteEvidenceParams) ([]ListRouteEvidenceRow, error) // ============================================================ // TRACES // ============================================================ diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 6a0eee13..054bc001 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -650,7 +650,7 @@ SELECT COALESCE(SUM(rssi_sum), 0)::bigint AS rssi_sum, SUM(rssi_n)::bigint AS rssi_n FROM mv_observer_activity_hourly -WHERE observer_id = $1 AND bucket >= $2::timestamptz +WHERE observer_id = $1 AND bucket >= $2::timestamptz AND bucket < $4::timestamptz GROUP BY 1 ORDER BY 1 ` @@ -659,6 +659,7 @@ type GetObserverActivityHourlyParams struct { ObserverID uuid.UUID `json:"observer_id"` Column2 pgtype.Timestamptz `json:"column_2"` Column3 pgtype.Interval `json:"column_3"` + Until pgtype.Timestamptz `json:"until"` } type GetObserverActivityHourlyRow struct { @@ -675,7 +676,12 @@ type GetObserverActivityHourlyRow struct { // Hour-or-coarser buckets summed from the hourly rollup; same COALESCE-plus-count shape as the raw query. func (q *Queries) GetObserverActivityHourly(ctx context.Context, arg GetObserverActivityHourlyParams) ([]GetObserverActivityHourlyRow, error) { - rows, err := q.db.Query(ctx, getObserverActivityHourly, arg.ObserverID, arg.Column2, arg.Column3) + rows, err := q.db.Query(ctx, getObserverActivityHourly, + arg.ObserverID, + arg.Column2, + arg.Column3, + arg.Until, + ) if err != nil { return nil, err } @@ -707,7 +713,7 @@ func (q *Queries) GetObserverActivityHourly(ctx context.Context, arg GetObserver const getObserverActivityHourlyPayloadTypes = `-- name: GetObserverActivityHourlyPayloadTypes :many SELECT payload_type, SUM(observations)::bigint AS count FROM mv_observer_activity_hourly -WHERE observer_id = $1 AND bucket >= $2::timestamptz +WHERE observer_id = $1 AND bucket >= $2::timestamptz AND bucket < $3::timestamptz GROUP BY payload_type ORDER BY count DESC ` @@ -715,15 +721,16 @@ ORDER BY count DESC type GetObserverActivityHourlyPayloadTypesParams struct { ObserverID uuid.UUID `json:"observer_id"` Column2 pgtype.Timestamptz `json:"column_2"` + Until pgtype.Timestamptz `json:"until"` } type GetObserverActivityHourlyPayloadTypesRow struct { - PayloadType *int16 `json:"payload_type"` - Count int64 `json:"count"` + PayloadType int16 `json:"payload_type"` + Count int64 `json:"count"` } func (q *Queries) GetObserverActivityHourlyPayloadTypes(ctx context.Context, arg GetObserverActivityHourlyPayloadTypesParams) ([]GetObserverActivityHourlyPayloadTypesRow, error) { - rows, err := q.db.Query(ctx, getObserverActivityHourlyPayloadTypes, arg.ObserverID, arg.Column2) + rows, err := q.db.Query(ctx, getObserverActivityHourlyPayloadTypes, arg.ObserverID, arg.Column2, arg.Until) if err != nil { return nil, err } @@ -742,6 +749,45 @@ func (q *Queries) GetObserverActivityHourlyPayloadTypes(ctx context.Context, arg return items, nil } +const getObserverActivityLiveSummary = `-- name: GetObserverActivityLiveSummary :one +WITH latest AS ( + SELECT heard_at FROM packet_observations + WHERE observer_id = $1::uuid AND heard_at <= $2::timestamptz + ORDER BY heard_at DESC LIMIT 1 +), hourly AS ( + SELECT COUNT(*)::bigint AS n FROM packet_observations + WHERE observer_id = $1::uuid + AND heard_at >= $3::timestamptz AND heard_at < $4::timestamptz +) +SELECT (SELECT heard_at FROM latest)::timestamptz AS latest_recorded_at, + hourly.n AS last_complete_hour FROM hourly +` + +type GetObserverActivityLiveSummaryParams struct { + ObserverID uuid.UUID `json:"observer_id"` + GeneratedAt pgtype.Timestamptz `json:"generated_at"` + HourStart pgtype.Timestamptz `json:"hour_start"` + HourEnd pgtype.Timestamptz `json:"hour_end"` +} + +type GetObserverActivityLiveSummaryRow struct { + LatestRecordedAt pgtype.Timestamptz `json:"latest_recorded_at"` + LastCompleteHour int64 `json:"last_complete_hour"` +} + +// Two indexed ranges, bounded to one observer; no legacy presence counters. +func (q *Queries) GetObserverActivityLiveSummary(ctx context.Context, arg GetObserverActivityLiveSummaryParams) (GetObserverActivityLiveSummaryRow, error) { + row := q.db.QueryRow(ctx, getObserverActivityLiveSummary, + arg.ObserverID, + arg.GeneratedAt, + arg.HourStart, + arg.HourEnd, + ) + var i GetObserverActivityLiveSummaryRow + err := row.Scan(&i.LatestRecordedAt, &i.LastCompleteHour) + return i, err +} + const getObserverActivityRaw = `-- name: GetObserverActivityRaw :many SELECT date_bin($3::interval, heard_at, TIMESTAMPTZ 'epoch')::timestamptz AS bucket, @@ -754,7 +800,7 @@ SELECT COALESCE(AVG(rssi) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0)), 0)::real AS rssi_avg, COUNT(rssi) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::bigint AS rssi_n FROM packet_observations -WHERE observer_id = $1 AND heard_at >= $2::timestamptz +WHERE observer_id = $1 AND heard_at >= $2::timestamptz AND heard_at < $4::timestamptz GROUP BY bucket ORDER BY bucket ` @@ -763,6 +809,7 @@ type GetObserverActivityRawParams struct { ObserverID uuid.UUID `json:"observer_id"` Column2 pgtype.Timestamptz `json:"column_2"` Column3 pgtype.Interval `json:"column_3"` + Until pgtype.Timestamptz `json:"until"` } type GetObserverActivityRawRow struct { @@ -781,7 +828,12 @@ type GetObserverActivityRawRow struct { // Aggregates are COALESCEd and paired with a count column: sqlc types a cast expression as // NOT NULL, so the counts are what tell the store a bucket had no costed or no signal rows. func (q *Queries) GetObserverActivityRaw(ctx context.Context, arg GetObserverActivityRawParams) ([]GetObserverActivityRawRow, error) { - rows, err := q.db.Query(ctx, getObserverActivityRaw, arg.ObserverID, arg.Column2, arg.Column3) + rows, err := q.db.Query(ctx, getObserverActivityRaw, + arg.ObserverID, + arg.Column2, + arg.Column3, + arg.Until, + ) if err != nil { return nil, err } @@ -813,7 +865,7 @@ func (q *Queries) GetObserverActivityRaw(ctx context.Context, arg GetObserverAct const getObserverActivityRawPayloadTypes = `-- name: GetObserverActivityRawPayloadTypes :many SELECT payload_type, COUNT(*)::bigint AS count FROM packet_observations -WHERE observer_id = $1 AND heard_at >= $2::timestamptz AND payload_type IS NOT NULL +WHERE observer_id = $1 AND heard_at >= $2::timestamptz AND heard_at < $3::timestamptz GROUP BY payload_type ORDER BY count DESC ` @@ -821,6 +873,7 @@ ORDER BY count DESC type GetObserverActivityRawPayloadTypesParams struct { ObserverID uuid.UUID `json:"observer_id"` Column2 pgtype.Timestamptz `json:"column_2"` + Until pgtype.Timestamptz `json:"until"` } type GetObserverActivityRawPayloadTypesRow struct { @@ -829,7 +882,7 @@ type GetObserverActivityRawPayloadTypesRow struct { } func (q *Queries) GetObserverActivityRawPayloadTypes(ctx context.Context, arg GetObserverActivityRawPayloadTypesParams) ([]GetObserverActivityRawPayloadTypesRow, error) { - rows, err := q.db.Query(ctx, getObserverActivityRawPayloadTypes, arg.ObserverID, arg.Column2) + rows, err := q.db.Query(ctx, getObserverActivityRawPayloadTypes, arg.ObserverID, arg.Column2, arg.Until) if err != nil { return nil, err } @@ -4422,11 +4475,12 @@ func (q *Queries) SetPacketDecrypted(ctx context.Context, packetHash []byte) err const touchObserverBrokers = `-- name: TouchObserverBrokers :exec UPDATE observer_brokers ob SET last_seen = GREATEST(ob.last_seen, v.seen), - last_packet_at = GREATEST(ob.last_packet_at, v.seen) + last_packet_at = GREATEST(ob.last_packet_at, v.packet) FROM ( SELECT unnest($1::uuid[]) AS observer_id, unnest($2::text[]) AS broker_name, - unnest($3::timestamptz[]) AS seen + unnest($3::timestamptz[]) AS seen, + unnest($4::timestamptz[]) AS packet ) v WHERE ob.observer_id = v.observer_id AND ob.broker_name = v.broker_name ` @@ -4435,10 +4489,16 @@ type TouchObserverBrokersParams struct { Column1 []uuid.UUID `json:"column_1"` Column2 []string `json:"column_2"` Column3 []pgtype.Timestamptz `json:"column_3"` + Column4 []pgtype.Timestamptz `json:"column_4"` } func (q *Queries) TouchObserverBrokers(ctx context.Context, arg TouchObserverBrokersParams) error { - _, err := q.db.Exec(ctx, touchObserverBrokers, arg.Column1, arg.Column2, arg.Column3) + _, err := q.db.Exec(ctx, touchObserverBrokers, + arg.Column1, + arg.Column2, + arg.Column3, + arg.Column4, + ) return err } @@ -4943,22 +5003,23 @@ func (q *Queries) UpsertObserver(ctx context.Context, publicKey []byte) (Observe const upsertObserverBroker = `-- name: UpsertObserverBroker :exec INSERT INTO observer_brokers (observer_id, broker_name, last_seen, last_packet_at) -VALUES ($1, $2, NOW(), NOW()) +VALUES ($1, $2, NOW(), CASE WHEN $3::boolean THEN NOW() END) ON CONFLICT (observer_id, broker_name) DO UPDATE SET - last_seen = NOW(), - last_packet_at = NOW() + last_seen = NOW(), + last_packet_at = COALESCE(EXCLUDED.last_packet_at, observer_brokers.last_packet_at) ` type UpsertObserverBrokerParams struct { ObserverID uuid.UUID `json:"observer_id"` BrokerName string `json:"broker_name"` + IsPacket bool `json:"is_packet"` } // ============================================================ // OBSERVER BROKERS // ============================================================ func (q *Queries) UpsertObserverBroker(ctx context.Context, arg UpsertObserverBrokerParams) error { - _, err := q.db.Exec(ctx, upsertObserverBroker, arg.ObserverID, arg.BrokerName) + _, err := q.db.Exec(ctx, upsertObserverBroker, arg.ObserverID, arg.BrokerName, arg.IsPacket) return err } diff --git a/db/sqlc/route_evidence.sql.go b/db/sqlc/route_evidence.sql.go new file mode 100644 index 00000000..0099c47b --- /dev/null +++ b/db/sqlc/route_evidence.sql.go @@ -0,0 +1,126 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: route_evidence.sql + +package db + +import ( + "context" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" +) + +const getRouteEvidenceRoute = `-- name: GetRouteEvidenceRoute :one +SELECT id, path_key, node_ids, hash_prefix, iata, hop_count, first_seen, last_seen, observation_count, last_reconfirmed_at FROM known_routes +WHERE iata = $1::bpchar AND path_key = $2::bytea +` + +type GetRouteEvidenceRouteParams struct { + Iata string `json:"iata"` + PathKey []byte `json:"path_key"` +} + +func (q *Queries) GetRouteEvidenceRoute(ctx context.Context, arg GetRouteEvidenceRouteParams) (KnownRoute, error) { + row := q.db.QueryRow(ctx, getRouteEvidenceRoute, arg.Iata, arg.PathKey) + var i KnownRoute + err := row.Scan( + &i.ID, + &i.PathKey, + &i.NodeIds, + &i.HashPrefix, + &i.Iata, + &i.HopCount, + &i.FirstSeen, + &i.LastSeen, + &i.ObservationCount, + &i.LastReconfirmedAt, + ) + return i, err +} + +const listRouteEvidence = `-- name: ListRouteEvidence :many +SELECT po.id, po.packet_hash, po.observer_id, o.display_name AS observer_name, + po.heard_at, po.payload_type, po.rssi, po.snr +FROM packet_observations po +JOIN observers o ON o.id = po.observer_id +WHERE po.iata = $1::bpchar + AND po.hash_size = $2::smallint + AND decode(md5(po.path_bytes), 'hex') = $3::bytea + AND po.path_bytes = $4::bytea + AND po.hop_count >= 2 + AND po.hop_count = $5::smallint + AND po.path_bytes IS NOT NULL AND po.payload_type IS NOT NULL AND po.payload_type <> 9 + AND po.heard_at >= $6::timestamptz AND po.heard_at < $7::timestamptz + AND (po.heard_at, po.id) < ($8::timestamptz, $9::bigint) +ORDER BY po.heard_at DESC, po.id DESC +LIMIT $10::integer +` + +type ListRouteEvidenceParams struct { + Iata string `json:"iata"` + HashSize int16 `json:"hash_size"` + PathDigest []byte `json:"path_digest"` + PathBytes []byte `json:"path_bytes"` + HopCount int16 `json:"hop_count"` + Since pgtype.Timestamptz `json:"since"` + Until pgtype.Timestamptz `json:"until"` + BeforeAt pgtype.Timestamptz `json:"before_at"` + BeforeID int64 `json:"before_id"` + PageLimit int32 `json:"page_limit"` +} + +type ListRouteEvidenceRow struct { + ID int64 `json:"id"` + PacketHash []byte `json:"packet_hash"` + ObserverID uuid.UUID `json:"observer_id"` + ObserverName *string `json:"observer_name"` + HeardAt pgtype.Timestamptz `json:"heard_at"` + PayloadType *int16 `json:"payload_type"` + Rssi *int16 `json:"rssi"` + Snr *float32 `json:"snr"` +} + +// Leading index equalities and the time/ID boundary bound both custom and generic plans. +// Full-byte equality is required even when the compact digest matches. TRACE path bytes +// carry readings; unclassified legacy observations cannot be safely called ordinary paths. +func (q *Queries) ListRouteEvidence(ctx context.Context, arg ListRouteEvidenceParams) ([]ListRouteEvidenceRow, error) { + rows, err := q.db.Query(ctx, listRouteEvidence, + arg.Iata, + arg.HashSize, + arg.PathDigest, + arg.PathBytes, + arg.HopCount, + arg.Since, + arg.Until, + arg.BeforeAt, + arg.BeforeID, + arg.PageLimit, + ) + if err != nil { + return nil, err + } + defer rows.Close() + items := []ListRouteEvidenceRow{} + for rows.Next() { + var i ListRouteEvidenceRow + if err := rows.Scan( + &i.ID, + &i.PacketHash, + &i.ObserverID, + &i.ObserverName, + &i.HeardAt, + &i.PayloadType, + &i.Rssi, + &i.Snr, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} diff --git a/docs/docs.go b/docs/docs.go index 43ed843a..93bc6f52 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -1468,6 +1468,12 @@ const docTemplate = `{ "description": "Bucket size: 5m, 15m, 1h, 6h or 24h (default 15m)", "name": "interval", "in": "query" + }, + { + "type": "integer", + "description": "Optional exclusive end in epoch milliseconds; aligned down to a complete bucket, at most 30 days old. Activity uses this window; summary freshness/latestRecordedAt and the last complete hour are measured at response generation time, independently of until.", + "name": "until", + "in": "query" } ], "responses": { @@ -2125,6 +2131,92 @@ const docTemplate = `{ } } }, + "/routes/{iata}/{pathKey}/observations": { + "get": { + "description": "Matches the saved IATA, complete path bytes and hash width. Excludes TRACE and unclassified reports; other widths and search-result subsegments are not included. Byte matches do not prove hop identities or delivery. The route and its historical counter can outlive raw evidence. Cursors retain full timestamp precision and pin route/window scope.", + "produces": [ + "application/json" + ], + "tags": [ + "Routes" + ], + "summary": "Get retained reports matching a full saved route prefix sequence", + "parameters": [ + { + "type": "string", + "description": "Three-character IATA code", + "name": "iata", + "in": "path", + "required": true + }, + { + "type": "string", + "description": "Stable 32-hex pathKey from a known-route response", + "name": "pathKey", + "in": "path", + "required": true + }, + { + "type": "string", + "description": "Server-anchored duration (default 24h, max 720h); exclusive with since/until/pageCursor", + "name": "range", + "in": "query" + }, + { + "type": "integer", + "description": "Window start epoch ms; provide with until (default last 24h)", + "name": "since", + "in": "query" + }, + { + "type": "integer", + "description": "Exclusive window end epoch ms; maximum span 30d, no future end", + "name": "until", + "in": "query" + }, + { + "type": "string", + "description": "Opaque precise cursor from nextPageCursor; window is pinned", + "name": "pageCursor", + "in": "query" + }, + { + "maximum": 200, + "minimum": 1, + "type": "integer", + "description": "Default 50; positive, capped at 200", + "name": "limit", + "in": "query" + } + ], + "responses": { + "200": { + "description": "OK", + "schema": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.RouteEvidence" + } + }, + "400": { + "description": "Bad Request", + "schema": { + "$ref": "#/definitions/internal_api_handlers.APIError" + } + }, + "404": { + "description": "Not Found", + "schema": { + "$ref": "#/definitions/internal_api_handlers.APIError" + } + }, + "500": { + "description": "Internal Server Error", + "schema": { + "$ref": "#/definitions/internal_api_handlers.APIError" + } + } + } + } + }, "/scopes": { "get": { "produces": [ @@ -3595,6 +3687,10 @@ const docTemplate = `{ }, "observationCount": { "type": "integer" + }, + "pathKey": { + "description": "stable identity within this IATA; use for route evidence links", + "type": "string" } } }, @@ -3924,6 +4020,7 @@ const docTemplate = `{ "type": "integer" }, "observationCount": { + "description": "legacy cumulative presence counter; includes non-packet events", "type": "integer" }, "observerType": { @@ -3979,6 +4076,10 @@ const docTemplate = `{ "github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivity": { "type": "object", "properties": { + "generatedAt": { + "description": "response computation time, not proof of continuous coverage", + "type": "integer" + }, "interval": { "type": "string" }, @@ -3999,6 +4100,21 @@ const docTemplate = `{ }, "range": { "type": "string" + }, + "source": { + "description": "raw or hourly; missing records do not prove an outage", + "type": "string" + }, + "summary": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivitySummary" + }, + "windowEnd": { + "description": "exclusive end, epoch ms; live requests include the current partial bucket", + "type": "integer" + }, + "windowStart": { + "description": "inclusive complete-bucket start, epoch ms", + "type": "integer" } } }, @@ -4046,6 +4162,27 @@ const docTemplate = `{ } } }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivitySummary": { + "type": "object", + "properties": { + "lastCompleteHour": { + "type": "integer" + }, + "lastCompleteHourEnd": { + "type": "integer" + }, + "lastCompleteHourStart": { + "type": "integer" + }, + "latestRecordedAt": { + "type": "integer" + }, + "recordedPackets": { + "description": "stored observations within windowStart/windowEnd", + "type": "integer" + } + } + }, "github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverBroker": { "type": "object", "properties": { @@ -4798,6 +4935,50 @@ const docTemplate = `{ } } }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.RouteEvidence": { + "type": "object", + "properties": { + "generatedAt": { + "type": "integer" + }, + "hasMore": { + "type": "boolean" + }, + "hashSize": { + "type": "integer" + }, + "items": { + "type": "array", + "items": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.RouteObservation" + } + }, + "matchAvailable": { + "type": "boolean" + }, + "matchType": { + "type": "string" + }, + "nextCursor": { + "type": "integer" + }, + "nextPageCursor": { + "type": "string" + }, + "pathBytes": { + "type": "string" + }, + "route": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.KnownRoute" + }, + "windowEnd": { + "type": "integer" + }, + "windowStart": { + "type": "integer" + } + } + }, "github_com_MeshCore-Beacon_beacon-server_internal_api.RouteHop": { "type": "object", "properties": { @@ -4818,6 +4999,38 @@ const docTemplate = `{ } } }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.RouteObservation": { + "type": "object", + "properties": { + "heardAt": { + "type": "integer" + }, + "id": { + "type": "integer" + }, + "observerId": { + "type": "string" + }, + "observerName": { + "type": "string" + }, + "packetHash": { + "type": "string" + }, + "payloadType": { + "type": "integer" + }, + "payloadTypeName": { + "type": "string" + }, + "rssi": { + "type": "integer" + }, + "snr": { + "type": "number" + } + } + }, "github_com_MeshCore-Beacon_beacon-server_internal_api.ScopeDetail": { "type": "object", "properties": { diff --git a/docs/swagger.json b/docs/swagger.json index 8a83f4c3..617277a7 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -1466,6 +1466,12 @@ "description": "Bucket size: 5m, 15m, 1h, 6h or 24h (default 15m)", "name": "interval", "in": "query" + }, + { + "type": "integer", + "description": "Optional exclusive end in epoch milliseconds; aligned down to a complete bucket, at most 30 days old. Activity uses this window; summary freshness/latestRecordedAt and the last complete hour are measured at response generation time, independently of until.", + "name": "until", + "in": "query" } ], "responses": { @@ -2123,6 +2129,92 @@ } } }, + "/routes/{iata}/{pathKey}/observations": { + "get": { + "description": "Matches the saved IATA, complete path bytes and hash width. Excludes TRACE and unclassified reports; other widths and search-result subsegments are not included. Byte matches do not prove hop identities or delivery. The route and its historical counter can outlive raw evidence. Cursors retain full timestamp precision and pin route/window scope.", + "produces": [ + "application/json" + ], + "tags": [ + "Routes" + ], + "summary": "Get retained reports matching a full saved route prefix sequence", + "parameters": [ + { + "type": "string", + "description": "Three-character IATA code", + "name": "iata", + "in": "path", + "required": true + }, + { + "type": "string", + "description": "Stable 32-hex pathKey from a known-route response", + "name": "pathKey", + "in": "path", + "required": true + }, + { + "type": "string", + "description": "Server-anchored duration (default 24h, max 720h); exclusive with since/until/pageCursor", + "name": "range", + "in": "query" + }, + { + "type": "integer", + "description": "Window start epoch ms; provide with until (default last 24h)", + "name": "since", + "in": "query" + }, + { + "type": "integer", + "description": "Exclusive window end epoch ms; maximum span 30d, no future end", + "name": "until", + "in": "query" + }, + { + "type": "string", + "description": "Opaque precise cursor from nextPageCursor; window is pinned", + "name": "pageCursor", + "in": "query" + }, + { + "maximum": 200, + "minimum": 1, + "type": "integer", + "description": "Default 50; positive, capped at 200", + "name": "limit", + "in": "query" + } + ], + "responses": { + "200": { + "description": "OK", + "schema": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.RouteEvidence" + } + }, + "400": { + "description": "Bad Request", + "schema": { + "$ref": "#/definitions/internal_api_handlers.APIError" + } + }, + "404": { + "description": "Not Found", + "schema": { + "$ref": "#/definitions/internal_api_handlers.APIError" + } + }, + "500": { + "description": "Internal Server Error", + "schema": { + "$ref": "#/definitions/internal_api_handlers.APIError" + } + } + } + } + }, "/scopes": { "get": { "produces": [ @@ -3593,6 +3685,10 @@ }, "observationCount": { "type": "integer" + }, + "pathKey": { + "description": "stable identity within this IATA; use for route evidence links", + "type": "string" } } }, @@ -3922,6 +4018,7 @@ "type": "integer" }, "observationCount": { + "description": "legacy cumulative presence counter; includes non-packet events", "type": "integer" }, "observerType": { @@ -3977,6 +4074,10 @@ "github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivity": { "type": "object", "properties": { + "generatedAt": { + "description": "response computation time, not proof of continuous coverage", + "type": "integer" + }, "interval": { "type": "string" }, @@ -3997,6 +4098,21 @@ }, "range": { "type": "string" + }, + "source": { + "description": "raw or hourly; missing records do not prove an outage", + "type": "string" + }, + "summary": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivitySummary" + }, + "windowEnd": { + "description": "exclusive end, epoch ms; live requests include the current partial bucket", + "type": "integer" + }, + "windowStart": { + "description": "inclusive complete-bucket start, epoch ms", + "type": "integer" } } }, @@ -4044,6 +4160,27 @@ } } }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivitySummary": { + "type": "object", + "properties": { + "lastCompleteHour": { + "type": "integer" + }, + "lastCompleteHourEnd": { + "type": "integer" + }, + "lastCompleteHourStart": { + "type": "integer" + }, + "latestRecordedAt": { + "type": "integer" + }, + "recordedPackets": { + "description": "stored observations within windowStart/windowEnd", + "type": "integer" + } + } + }, "github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverBroker": { "type": "object", "properties": { @@ -4796,6 +4933,50 @@ } } }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.RouteEvidence": { + "type": "object", + "properties": { + "generatedAt": { + "type": "integer" + }, + "hasMore": { + "type": "boolean" + }, + "hashSize": { + "type": "integer" + }, + "items": { + "type": "array", + "items": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.RouteObservation" + } + }, + "matchAvailable": { + "type": "boolean" + }, + "matchType": { + "type": "string" + }, + "nextCursor": { + "type": "integer" + }, + "nextPageCursor": { + "type": "string" + }, + "pathBytes": { + "type": "string" + }, + "route": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.KnownRoute" + }, + "windowEnd": { + "type": "integer" + }, + "windowStart": { + "type": "integer" + } + } + }, "github_com_MeshCore-Beacon_beacon-server_internal_api.RouteHop": { "type": "object", "properties": { @@ -4816,6 +4997,38 @@ } } }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.RouteObservation": { + "type": "object", + "properties": { + "heardAt": { + "type": "integer" + }, + "id": { + "type": "integer" + }, + "observerId": { + "type": "string" + }, + "observerName": { + "type": "string" + }, + "packetHash": { + "type": "string" + }, + "payloadType": { + "type": "integer" + }, + "payloadTypeName": { + "type": "string" + }, + "rssi": { + "type": "integer" + }, + "snr": { + "type": "number" + } + } + }, "github_com_MeshCore-Beacon_beacon-server_internal_api.ScopeDetail": { "type": "object", "properties": { diff --git a/docs/swagger.yaml b/docs/swagger.yaml index f03b978b..3898b5de 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -272,6 +272,9 @@ definitions: type: integer observationCount: type: integer + pathKey: + description: stable identity within this IATA; use for route evidence links + type: string type: object github_com_MeshCore-Beacon_beacon-server_internal_api.Node: properties: @@ -521,6 +524,7 @@ definitions: description: epoch ms type: integer observationCount: + description: legacy cumulative presence counter; includes non-packet events type: integer observerType: description: e.g. "meshcoretomqtt", "meshcoreha" @@ -560,6 +564,9 @@ definitions: type: object github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivity: properties: + generatedAt: + description: response computation time, not proof of continuous coverage + type: integer interval: type: string payloadTypes: @@ -574,6 +581,18 @@ definitions: $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivityRadio' range: type: string + source: + description: raw or hourly; missing records do not prove an outage + type: string + summary: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivitySummary' + windowEnd: + description: exclusive end, epoch ms; live requests include the current partial + bucket + type: integer + windowStart: + description: inclusive complete-bucket start, epoch ms + type: integer type: object github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivityPoint: properties: @@ -604,6 +623,20 @@ definitions: sf: type: integer type: object + github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivitySummary: + properties: + lastCompleteHour: + type: integer + lastCompleteHourEnd: + type: integer + lastCompleteHourStart: + type: integer + latestRecordedAt: + type: integer + recordedPackets: + description: stored observations within windowStart/windowEnd + type: integer + type: object github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverBroker: properties: lastPacketAt: @@ -1123,6 +1156,35 @@ definitions: description: hex-encoded prefix used for resolution type: string type: object + github_com_MeshCore-Beacon_beacon-server_internal_api.RouteEvidence: + properties: + generatedAt: + type: integer + hasMore: + type: boolean + hashSize: + type: integer + items: + items: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.RouteObservation' + type: array + matchAvailable: + type: boolean + matchType: + type: string + nextCursor: + type: integer + nextPageCursor: + type: string + pathBytes: + type: string + route: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.KnownRoute' + windowEnd: + type: integer + windowStart: + type: integer + type: object github_com_MeshCore-Beacon_beacon-server_internal_api.RouteHop: properties: hashBytes: @@ -1135,6 +1197,27 @@ definitions: nodeId: type: string type: object + github_com_MeshCore-Beacon_beacon-server_internal_api.RouteObservation: + properties: + heardAt: + type: integer + id: + type: integer + observerId: + type: string + observerName: + type: string + packetHash: + type: string + payloadType: + type: integer + payloadTypeName: + type: string + rssi: + type: integer + snr: + type: number + type: object github_com_MeshCore-Beacon_beacon-server_internal_api.ScopeDetail: properties: iataCount: @@ -2401,6 +2484,13 @@ paths: in: query name: interval type: string + - description: Optional exclusive end in epoch milliseconds; aligned down to + a complete bucket, at most 30 days old. Activity uses this window; summary + freshness/latestRecordedAt and the last complete hour are measured at response + generation time, independently of until. + in: query + name: until + type: integer produces: - application/json responses: @@ -2766,6 +2856,69 @@ paths: summary: List known routes tags: - Routes + /routes/{iata}/{pathKey}/observations: + get: + description: Matches the saved IATA, complete path bytes and hash width. Excludes + TRACE and unclassified reports; other widths and search-result subsegments + are not included. Byte matches do not prove hop identities or delivery. The + route and its historical counter can outlive raw evidence. Cursors retain + full timestamp precision and pin route/window scope. + parameters: + - description: Three-character IATA code + in: path + name: iata + required: true + type: string + - description: Stable 32-hex pathKey from a known-route response + in: path + name: pathKey + required: true + type: string + - description: Server-anchored duration (default 24h, max 720h); exclusive with + since/until/pageCursor + in: query + name: range + type: string + - description: Window start epoch ms; provide with until (default last 24h) + in: query + name: since + type: integer + - description: Exclusive window end epoch ms; maximum span 30d, no future end + in: query + name: until + type: integer + - description: Opaque precise cursor from nextPageCursor; window is pinned + in: query + name: pageCursor + type: string + - description: Default 50; positive, capped at 200 + in: query + maximum: 200 + minimum: 1 + name: limit + type: integer + produces: + - application/json + responses: + "200": + description: OK + schema: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.RouteEvidence' + "400": + description: Bad Request + schema: + $ref: '#/definitions/internal_api_handlers.APIError' + "404": + description: Not Found + schema: + $ref: '#/definitions/internal_api_handlers.APIError' + "500": + description: Internal Server Error + schema: + $ref: '#/definitions/internal_api_handlers.APIError' + summary: Get retained reports matching a full saved route prefix sequence + tags: + - Routes /routes/cross: get: parameters: diff --git a/internal/api/handlers/observers.go b/internal/api/handlers/observers.go index 43147b5a..66bddd8d 100644 --- a/internal/api/handlers/observers.go +++ b/internal/api/handlers/observers.go @@ -256,6 +256,7 @@ var activityIntervals = map[string]time.Duration{ // @Param observerId path string true "Observer UUID" // @Param range query string false "Trailing window as a Go duration, max 720h (default 24h); max 48h when interval is under 1h" // @Param interval query string false "Bucket size: 5m, 15m, 1h, 6h or 24h (default 15m)" +// @Param until query int false "Optional exclusive end in epoch milliseconds; aligned down to a complete bucket, at most 30 days old. Activity uses this window; summary freshness/latestRecordedAt and the last complete hour are measured at response generation time, independently of until." // @Success 200 {object} api.ObserverActivity // @Failure 400 {object} handlers.APIError // @Failure 404 {object} handlers.APIError @@ -295,7 +296,17 @@ func getObserverActivity(reader api.Reader) http.HandlerFunc { respondError(w, http.StatusBadRequest, "range/interval exceeds 1000 buckets") return } - activity, err := reader.GetObserverActivity(r.Context(), observerID, window, interval) + until := time.Time{} + if values, ok := r.URL.Query()["until"]; ok { + n, err := strconv.ParseInt(values[0], 10, 64) + now := time.Now() + if len(values) != 1 || err != nil || n < 0 || n > now.UnixMilli() || n < now.Add(-720*time.Hour).UnixMilli() { + respondError(w, http.StatusBadRequest, "until must be one past epoch-millisecond timestamp within 30 days") + return + } + until = time.UnixMilli(n) + } + activity, err := reader.GetObserverActivity(r.Context(), observerID, window, interval, until) if err != nil { if errors.Is(err, pgx.ErrNoRows) { respondError(w, http.StatusNotFound, "observer not found") diff --git a/internal/api/handlers/observers_test.go b/internal/api/handlers/observers_test.go index 1bf865fa..cfef1c9c 100644 --- a/internal/api/handlers/observers_test.go +++ b/internal/api/handlers/observers_test.go @@ -200,7 +200,7 @@ func TestGetObserverActivity_Defaults(t *testing.T) { var gotWindow, gotInterval time.Duration r := chi.NewRouter() r.Get("/observers/{observerId}/activity", getObserverActivity(stubReader{ - getObserverActivity: func(_ context.Context, _ uuid.UUID, window, interval time.Duration) (*api.ObserverActivity, error) { + getObserverActivity: func(_ context.Context, _ uuid.UUID, window, interval time.Duration, _ time.Time) (*api.ObserverActivity, error) { gotWindow, gotInterval = window, interval return &api.ObserverActivity{Points: []api.ObserverActivityPoint{}}, nil }, @@ -232,7 +232,7 @@ func TestGetObserverActivity_CustomRangeAndInterval(t *testing.T) { var gotWindow, gotInterval time.Duration r := chi.NewRouter() r.Get("/observers/{observerId}/activity", getObserverActivity(stubReader{ - getObserverActivity: func(_ context.Context, _ uuid.UUID, window, interval time.Duration) (*api.ObserverActivity, error) { + getObserverActivity: func(_ context.Context, _ uuid.UUID, window, interval time.Duration, _ time.Time) (*api.ObserverActivity, error) { gotWindow, gotInterval = window, interval return &api.ObserverActivity{Points: []api.ObserverActivityPoint{}}, nil }, @@ -264,7 +264,7 @@ func TestGetObserverActivity_SubHourRangeLimit(t *testing.T) { r := chi.NewRouter() r.Get("/observers/{observerId}/activity", getObserverActivity(stubReader{ - getObserverActivity: func(_ context.Context, _ uuid.UUID, _, _ time.Duration) (*api.ObserverActivity, error) { + getObserverActivity: func(_ context.Context, _ uuid.UUID, _, _ time.Duration, _ time.Time) (*api.ObserverActivity, error) { t.Fatal("reader should not be called") return nil, nil }, @@ -286,7 +286,7 @@ func TestGetObserverActivity_SubHourRangeLimit(t *testing.T) { var gotWindow, gotInterval time.Duration ok := chi.NewRouter() ok.Get("/observers/{observerId}/activity", getObserverActivity(stubReader{ - getObserverActivity: func(_ context.Context, _ uuid.UUID, window, interval time.Duration) (*api.ObserverActivity, error) { + getObserverActivity: func(_ context.Context, _ uuid.UUID, window, interval time.Duration, _ time.Time) (*api.ObserverActivity, error) { gotWindow, gotInterval = window, interval return &api.ObserverActivity{Points: []api.ObserverActivityPoint{}}, nil }, @@ -309,6 +309,10 @@ func TestGetObserverActivity_BadRequests(t *testing.T) { query string }{ {"invalid uuid", "not-a-uuid", ""}, + {"bad end", "00000000-0000-0000-0000-000000000001", "?until=banana"}, + {"negative end", "00000000-0000-0000-0000-000000000001", "?until=-1"}, + {"duplicate end", "00000000-0000-0000-0000-000000000001", "?until=1&until=2"}, + {"future end", "00000000-0000-0000-0000-000000000001", "?until=253402300799999"}, {"unparseable range", "00000000-0000-0000-0000-000000000001", "?range=banana"}, {"range too long", "00000000-0000-0000-0000-000000000001", "?range=721h"}, {"zero range", "00000000-0000-0000-0000-000000000001", "?range=0"}, @@ -319,7 +323,7 @@ func TestGetObserverActivity_BadRequests(t *testing.T) { t.Run(tt.name, func(t *testing.T) { r := chi.NewRouter() r.Get("/observers/{observerId}/activity", getObserverActivity(stubReader{ - getObserverActivity: func(_ context.Context, _ uuid.UUID, _, _ time.Duration) (*api.ObserverActivity, error) { + getObserverActivity: func(_ context.Context, _ uuid.UUID, _, _ time.Duration, _ time.Time) (*api.ObserverActivity, error) { t.Fatal("reader should not be called") return nil, nil }, @@ -338,7 +342,7 @@ func TestGetObserverActivity_NotFound(t *testing.T) { observerID := uuid.MustParse("00000000-0000-0000-0000-000000000001") r := chi.NewRouter() r.Get("/observers/{observerId}/activity", getObserverActivity(stubReader{ - getObserverActivity: func(_ context.Context, _ uuid.UUID, _, _ time.Duration) (*api.ObserverActivity, error) { + getObserverActivity: func(_ context.Context, _ uuid.UUID, _, _ time.Duration, _ time.Time) (*api.ObserverActivity, error) { return nil, pgx.ErrNoRows }, })) @@ -364,7 +368,7 @@ func TestGetObserverActivity_ReaderError(t *testing.T) { observerID := uuid.MustParse("00000000-0000-0000-0000-000000000001") r := chi.NewRouter() r.Get("/observers/{observerId}/activity", getObserverActivity(stubReader{ - getObserverActivity: func(_ context.Context, _ uuid.UUID, _, _ time.Duration) (*api.ObserverActivity, error) { + getObserverActivity: func(_ context.Context, _ uuid.UUID, _, _ time.Duration, _ time.Time) (*api.ObserverActivity, error) { return nil, errors.New("boom") }, })) @@ -381,7 +385,7 @@ func TestGetObserverActivity_NullsSerialise(t *testing.T) { observerID := uuid.MustParse("00000000-0000-0000-0000-000000000001") r := chi.NewRouter() r.Get("/observers/{observerId}/activity", getObserverActivity(stubReader{ - getObserverActivity: func(_ context.Context, _ uuid.UUID, _, _ time.Duration) (*api.ObserverActivity, error) { + getObserverActivity: func(_ context.Context, _ uuid.UUID, _, _ time.Duration, _ time.Time) (*api.ObserverActivity, error) { return &api.ObserverActivity{ PayloadTypes: []api.PayloadBreakdownItem{}, Points: []api.ObserverActivityPoint{{T: 1757376000000, Observations: 3}}, diff --git a/internal/api/handlers/route_evidence.go b/internal/api/handlers/route_evidence.go new file mode 100644 index 00000000..6c156fef --- /dev/null +++ b/internal/api/handlers/route_evidence.go @@ -0,0 +1,118 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later +package handlers + +import ( + "errors" + "net/http" + "strconv" + "strings" + "time" + + "github.com/MeshCore-Beacon/beacon-server/internal/api" + "github.com/go-chi/chi/v5" + "github.com/jackc/pgx/v5" +) + +func routeEvidenceQuery(r *http.Request, iata, key string, now time.Time) (api.RouteEvidenceQuery, error) { + q := api.RouteEvidenceQuery{} + values := r.URL.Query() + for _, name := range []string{"pageCursor", "since", "until", "range", "limit"} { + if len(values[name]) > 1 { + return q, api.ErrRouteEvidenceInput + } + } + if values.Has("cursor") { + return q, api.ErrRouteEvidenceInput + } + limit, err := parseLimit(r, 50) + if err != nil { + return q, err + } + q.Limit = limit + if values.Has("pageCursor") { + c, err := api.ParseRouteEvidenceCursor(values.Get("pageCursor")) + if err != nil || c.IATA != iata || c.PathKey != key { + return q, api.ErrRouteEvidenceInput + } + q.Cursor = c + q.Since = c.Since + q.Until = c.Until + } else { + q.Until = now.UTC().Truncate(time.Millisecond) + q.Since = q.Until.Add(-24 * time.Hour) + } + if values.Has("range") { + if values.Has("pageCursor") || values.Has("since") || values.Has("until") { + return q, api.ErrRouteEvidenceInput + } + duration, err := time.ParseDuration(values.Get("range")) + if err != nil || duration < time.Millisecond || duration > api.MaxRouteEvidenceWindow || duration%time.Millisecond != 0 { + return q, api.ErrRouteEvidenceInput + } + q.Since = q.Until.Add(-duration) + } + if values.Has("since") || values.Has("until") { + if !values.Has("since") || !values.Has("until") { + return q, api.ErrRouteEvidenceInput + } + since, e1 := strconv.ParseInt(values.Get("since"), 10, 64) + until, e2 := strconv.ParseInt(values.Get("until"), 10, 64) + if e1 != nil || e2 != nil || since < 0 || until < 0 || since > 253402300799999 || until > 253402300799999 { + return q, api.ErrRouteEvidenceInput + } + q.Since = time.UnixMilli(since).UTC() + q.Until = time.UnixMilli(until).UTC() + if q.Cursor != nil && (!q.Since.Equal(q.Cursor.Since) || !q.Until.Equal(q.Cursor.Until)) { + return q, api.ErrRouteEvidenceInput + } + } + if !api.ValidRouteEvidenceWindow(q.Since, q.Until) || q.Until.After(now) { + return q, api.ErrRouteEvidenceInput + } + return q, nil +} + +// getRouteEvidence godoc +// +// @Summary Get retained reports matching a full saved route prefix sequence +// @Description Matches the saved IATA, complete path bytes and hash width. Excludes TRACE and unclassified reports; other widths and search-result subsegments are not included. Byte matches do not prove hop identities or delivery. The route and its historical counter can outlive raw evidence. Cursors retain full timestamp precision and pin route/window scope. +// @Tags Routes +// @Produce json +// @Param iata path string true "Three-character IATA code" +// @Param pathKey path string true "Stable 32-hex pathKey from a known-route response" +// @Param range query string false "Server-anchored duration (default 24h, max 720h); exclusive with since/until/pageCursor" +// @Param since query int false "Window start epoch ms; provide with until (default last 24h)" +// @Param until query int false "Exclusive window end epoch ms; maximum span 30d, no future end" +// @Param pageCursor query string false "Opaque precise cursor from nextPageCursor; window is pinned" +// @Param limit query int false "Default 50; positive, capped at 200" minimum(1) maximum(200) +// @Success 200 {object} api.RouteEvidence +// @Failure 400 {object} handlers.APIError +// @Failure 404 {object} handlers.APIError +// @Failure 500 {object} handlers.APIError +// @Router /routes/{iata}/{pathKey}/observations [get] +func getRouteEvidence(reader api.Reader) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + iata, key := strings.ToUpper(chi.URLParam(r, "iata")), strings.ToLower(chi.URLParam(r, "pathKey")) + if !api.ValidRouteEvidenceKey(iata, key) { + respondError(w, 400, "invalid IATA or route pathKey") + return + } + query, err := routeEvidenceQuery(r, iata, key, time.Now()) + if err != nil { + respondError(w, 400, "invalid route evidence window, limit or page cursor") + return + } + result, err := reader.GetRouteEvidence(r.Context(), iata, key, query) + switch { + case errors.Is(err, pgx.ErrNoRows), err == nil && result == nil: + respondError(w, 404, "saved route not found") + case errors.Is(err, api.ErrRouteEvidenceInput): + respondError(w, 400, "invalid route evidence request") + case err != nil: + respondError(w, 500, "internal server error") + default: + respond(w, 200, result) + } + } +} diff --git a/internal/api/handlers/route_evidence_test.go b/internal/api/handlers/route_evidence_test.go new file mode 100644 index 00000000..8340d85b --- /dev/null +++ b/internal/api/handlers/route_evidence_test.go @@ -0,0 +1,78 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later +package handlers + +import ( + "context" + "errors" + "net/http/httptest" + "net/url" + "testing" + "time" + + "github.com/MeshCore-Beacon/beacon-server/internal/api" + "github.com/jackc/pgx/v5" +) + +const evidenceKey = "f097439148601d9f3291c474f82fa64c" + +func TestRouteEvidenceQuery(t *testing.T) { + now := time.Date(2026, 9, 27, 12, 0, 0, 0, time.UTC) + q, err := routeEvidenceQuery(httptest.NewRequest("GET", "/", nil), "YOW", evidenceKey, now) + if err != nil || q.Limit != 50 || q.Until != now || q.Until.Sub(q.Since) != 24*time.Hour { + t.Fatalf("defaults: %+v %v", q, err) + } + week, err := routeEvidenceQuery(httptest.NewRequest("GET", "/?range=168h", nil), "YOW", evidenceKey, now) + if err != nil || !week.Until.Equal(now) || week.Until.Sub(week.Since) != 7*24*time.Hour { + t.Fatalf("server anchored range: %+v %v", week, err) + } + c := api.RouteEvidenceCursor{IATA: "YOW", PathKey: evidenceKey, Since: q.Since, Until: q.Until, HeardAt: q.Since.Add(time.Hour + 123*time.Microsecond), ID: 3} + next, err := routeEvidenceQuery(httptest.NewRequest("GET", "/?pageCursor="+url.QueryEscape(c.String())+"&limit=999", nil), "YOW", evidenceKey, now.Add(time.Hour)) + if err != nil || next.Limit != 200 || !next.Until.Equal(now) || !next.Cursor.HeardAt.Equal(c.HeardAt) { + t.Fatalf("pinned cursor: %+v %v", next, err) + } + for _, query := range []string{"?limit=0", "?limit=-1", "?limit=a", "?limit=1&limit=2", "?since=0", "?until=10", "?since=2&until=1", "?since=0&until=2678400000", "?since=0&until=99999999999999999", "?pageCursor=bad", "?pageCursor=", "?cursor=123", "?pageCursor=" + url.QueryEscape(c.String()) + "&since=0&until=10"} { + if _, err := routeEvidenceQuery(httptest.NewRequest("GET", "/"+query, nil), "YOW", evidenceKey, now); err == nil { + t.Errorf("accepted %s", query) + } + } + if _, err := routeEvidenceQuery(httptest.NewRequest("GET", "/?pageCursor="+url.QueryEscape(c.String()), nil), "YVR", evidenceKey, now); err == nil { + t.Fatal("cursor crossed IATA") + } + for _, query := range []string{"range=0h", "range=-1h", "range=721h", "range=1ns", "range=7d", "range=24h&range=168h", "range=24h&since=0&until=10", "range=24h&pageCursor=" + url.QueryEscape(c.String())} { + if _, err := routeEvidenceQuery(httptest.NewRequest("GET", "/?"+query, nil), "YOW", evidenceKey, now); err == nil { + t.Errorf("accepted %s", query) + } + } +} + +func TestRouteEvidenceHandler(t *testing.T) { + for _, tc := range []struct { + path string + err error + status int + }{ + {"/yow/" + evidenceKey + "/observations", nil, 200}, + {"/YOW/bad/observations", nil, 400}, + {"/LONG/" + evidenceKey + "/observations", nil, 400}, + {"/YOW/" + evidenceKey + "/observations", pgx.ErrNoRows, 404}, + {"/YOW/" + evidenceKey + "/observations", errors.New("database unavailable"), 500}, + } { + called := false + router := RoutesRouter(stubReader{getRouteEvidence: func(_ context.Context, iata, key string, q api.RouteEvidenceQuery) (*api.RouteEvidence, error) { + called = true + if iata != "YOW" || key != evidenceKey || q.Limit != 50 { + t.Fatal("wrong reader scope") + } + return &api.RouteEvidence{Page: api.Page[api.RouteObservation]{Items: []api.RouteObservation{}}}, tc.err + }}) + w := httptest.NewRecorder() + router.ServeHTTP(w, httptest.NewRequest("GET", tc.path, nil)) + if w.Code != tc.status { + t.Fatalf("%s: %d %s", tc.path, w.Code, w.Body.String()) + } + if tc.status == 400 && called { + t.Fatal("invalid request reached database") + } + } +} diff --git a/internal/api/handlers/routes.go b/internal/api/handlers/routes.go index a5bdd76c..77f1117e 100644 --- a/internal/api/handlers/routes.go +++ b/internal/api/handlers/routes.go @@ -22,6 +22,7 @@ func RoutesRouter(reader api.Reader) http.Handler { r.Get("/", listKnownRoutes(reader)) r.Get("/cross", searchCrossIATARoutes(reader)) r.Get("/search", searchKnownRoutes(reader)) + r.Get("/{iata}/{pathKey}/observations", getRouteEvidence(reader)) return r } diff --git a/internal/api/handlers/stub_reader_test.go b/internal/api/handlers/stub_reader_test.go index 42200c1a..f9f885a6 100644 --- a/internal/api/handlers/stub_reader_test.go +++ b/internal/api/handlers/stub_reader_test.go @@ -16,6 +16,7 @@ import ( // Unset fields return zero values. Use it for both validation tests // (leave all fields nil) and happy path tests (set only what you need). type stubReader struct { + getRouteEvidence func(context.Context, string, string, api.RouteEvidenceQuery) (*api.RouteEvidence, error) getObserverComparison func(context.Context, uuid.UUID, uuid.UUID, time.Time, time.Time, []string) (*api.ObserverComparison, error) listIATAs func(ctx context.Context) ([]api.IATA, error) getIATA func(ctx context.Context, iata string) (*api.IATA, error) @@ -32,7 +33,7 @@ type stubReader struct { getObserver func(ctx context.Context, observerID uuid.UUID) (*api.Observer, error) getObserverTelemetry func(ctx context.Context, observerID uuid.UUID, since, until time.Time, afterID int64) (*api.ObserverTelemetry, error) getObserverTelemetryBucketed func(ctx context.Context, observerID uuid.UUID, since, until time.Time, bucketHours int32) ([]api.ObserverTelemetryPoint, error) - getObserverActivity func(ctx context.Context, observerID uuid.UUID, window, interval time.Duration) (*api.ObserverActivity, error) + getObserverActivity func(ctx context.Context, observerID uuid.UUID, window, interval time.Duration, until time.Time) (*api.ObserverActivity, error) getObserverScopes func(ctx context.Context, observerID uuid.UUID) ([]string, error) listObserverAdverts func(ctx context.Context, observerID uuid.UUID, cursor int64, limit int32) (api.Page[api.AdvertObservation], error) listNodes func(ctx context.Context, nodeType int16, iatas []string, supportsMultibytePaths, supportsMultibyteTraces *bool, pubkey []byte, pubkeyPrefix, name, scope string, cursor int64, limit int32, includeNeighbors bool) (api.Page[api.NodeSummary], error) @@ -186,9 +187,9 @@ func (s stubReader) GetObserverTelemetryBucketed(ctx context.Context, observerID return nil, nil } -func (s stubReader) GetObserverActivity(ctx context.Context, observerID uuid.UUID, window, interval time.Duration) (*api.ObserverActivity, error) { +func (s stubReader) GetObserverActivity(ctx context.Context, observerID uuid.UUID, window, interval time.Duration, until time.Time) (*api.ObserverActivity, error) { if s.getObserverActivity != nil { - return s.getObserverActivity(ctx, observerID, window, interval) + return s.getObserverActivity(ctx, observerID, window, interval, until) } return nil, nil } @@ -409,3 +410,10 @@ func (s stubReader) GetNodesByIDs(ctx context.Context, ids []uuid.UUID) (map[uui } return nil, nil } + +func (s stubReader) GetRouteEvidence(ctx context.Context, iata, key string, q api.RouteEvidenceQuery) (*api.RouteEvidence, error) { + if s.getRouteEvidence != nil { + return s.getRouteEvidence(ctx, iata, key, q) + } + return nil, nil +} diff --git a/internal/api/observers.go b/internal/api/observers.go index 206387b7..2da680c1 100644 --- a/internal/api/observers.go +++ b/internal/api/observers.go @@ -44,8 +44,8 @@ type Observer struct { LastStatusAt *int64 `json:"lastStatusAt,omitempty"` // epoch ms FirstSeen int64 `json:"firstSeen"` // epoch ms LastSeen int64 `json:"lastSeen"` // epoch ms - ObservationCount int64 `json:"observationCount"` - Brokers []ObserverBroker `json:"brokers"` // broker names this observer has been seen on + ObservationCount int64 `json:"observationCount"` // legacy cumulative presence counter; includes non-packet events + Brokers []ObserverBroker `json:"brokers"` // broker names this observer has been seen on } // ObserverTelemetryPoint is a single telemetry snapshot for an observer. @@ -88,8 +88,25 @@ type ObserverActivityPoint struct { RSSIAvg *float32 `json:"rssiAvg"` } +// ObserverActivitySummary describes stored packet records, never MQTT presence events. +// Freshness and lastCompleteHour are measured at generatedAt, even with an explicit until; +// only recordedPackets follows the selected activity window. +type ObserverActivitySummary struct { + RecordedPackets int64 `json:"recordedPackets"` // stored observations within windowStart/windowEnd + LastCompleteHour int64 `json:"lastCompleteHour"` + LastCompleteHourStart int64 `json:"lastCompleteHourStart"` + LastCompleteHourEnd int64 `json:"lastCompleteHourEnd"` + LatestRecordedAt *int64 `json:"latestRecordedAt"` +} + // ObserverActivity is the per-observer heard-activity response. type ObserverActivity struct { + WindowStart int64 `json:"windowStart"` // inclusive complete-bucket start, epoch ms + WindowEnd int64 `json:"windowEnd"` // exclusive end, epoch ms; live requests include the current partial bucket + GeneratedAt int64 `json:"generatedAt"` // response computation time, not proof of continuous coverage + Source string `json:"source"` // raw or hourly; missing records do not prove an outage + Summary *ObserverActivitySummary `json:"summary,omitempty"` + Range string `json:"range"` Interval string `json:"interval"` Radio *ObserverActivityRadio `json:"radio"` diff --git a/internal/api/reader.go b/internal/api/reader.go index 8aa3f1af..d46ec2d9 100644 --- a/internal/api/reader.go +++ b/internal/api/reader.go @@ -23,6 +23,8 @@ type Page[T any] struct { } type Reader interface { + // GetRouteEvidence returns the full saved route and exact-prefix retained ordinary reports. + GetRouteEvidence(ctx context.Context, iata, pathKey string, query RouteEvidenceQuery) (*RouteEvidence, error) // ListIATAs returns all known IATA codes with display name and coordinates. // IATAs are auto-created on first packet arrival from that location. ListIATAs(ctx context.Context) ([]IATA, error) @@ -100,7 +102,7 @@ type Reader interface { // GetObserverActivity returns bucketed heard-activity for an observer over the trailing window. // interval >= 1h is served from the hourly rollup. Returns pgx.ErrNoRows for an unknown observer. // Range and Interval on the result are left empty for the handler to fill. - GetObserverActivity(ctx context.Context, observerID uuid.UUID, window, interval time.Duration) (*ObserverActivity, error) + GetObserverActivity(ctx context.Context, observerID uuid.UUID, window, interval time.Duration, until time.Time) (*ObserverActivity, error) // GetObserverScopes returns the names of all transport scopes an observer has // been seen forwarding packets for, ordered alphabetically. diff --git a/internal/api/route_evidence.go b/internal/api/route_evidence.go new file mode 100644 index 00000000..cadc056a --- /dev/null +++ b/internal/api/route_evidence.go @@ -0,0 +1,110 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later +package api + +import ( + "encoding/hex" + "errors" + "fmt" + "strconv" + "strings" + "time" + + "github.com/google/uuid" +) + +const MaxRouteEvidenceWindow = 30 * 24 * time.Hour + +var ErrRouteEvidenceInput = errors.New("invalid route evidence request") + +// RouteObservation is a retained report reference, not a packet body or proof of hop identity. +type RouteObservation struct { + ID int64 `json:"id"` + PacketHash string `json:"packetHash"` + ObserverID uuid.UUID `json:"observerId"` + ObserverName *string `json:"observerName,omitempty"` + HeardAt int64 `json:"heardAt"` + PayloadType int16 `json:"payloadType"` + PayloadTypeName string `json:"payloadTypeName"` + RSSI *int16 `json:"rssi,omitempty"` + SNR *float32 `json:"snr,omitempty"` +} + +// RouteEvidence uses nextPageCursor rather than the lossy legacy numeric cursor. +type RouteEvidence struct { + Page[RouteObservation] + Route KnownRoute `json:"route"` + WindowStart int64 `json:"windowStart"` + WindowEnd int64 `json:"windowEnd"` + GeneratedAt int64 `json:"generatedAt"` + MatchType string `json:"matchType"` + MatchAvailable bool `json:"matchAvailable"` + HashSize int16 `json:"hashSize"` + PathBytes string `json:"pathBytes"` + NextPageCursor *string `json:"nextPageCursor,omitempty"` +} + +type RouteEvidenceQuery struct { + Since time.Time + Until time.Time + Cursor *RouteEvidenceCursor + Limit int32 +} + +// RouteEvidenceCursor binds the precise (heard_at DESC,id DESC) boundary to its route/window. +type RouteEvidenceCursor struct { + IATA string + PathKey string + Since time.Time + Until time.Time + HeardAt time.Time + ID int64 +} + +func ValidRouteEvidenceKey(iata, key string) bool { + if len(iata) != 3 || len(key) != 32 { + return false + } + for _, c := range iata { + if !(c >= 'A' && c <= 'Z') && !(c >= '0' && c <= '9') { + return false + } + } + _, err := hex.DecodeString(key) + return err == nil && key == strings.ToLower(key) +} + +func ValidRouteEvidenceWindow(since, until time.Time) bool { + return !since.Before(time.Unix(0, 0)) && until.After(since) && until.Sub(since) <= MaxRouteEvidenceWindow +} + +func (c RouteEvidenceCursor) String() string { + return fmt.Sprintf("v1:%s:%s:%d:%d:%d:%d", c.IATA, c.PathKey, c.Since.UnixMilli(), c.Until.UnixMilli(), c.HeardAt.UnixMicro(), c.ID) +} + +func ParseRouteEvidenceCursor(raw string) (*RouteEvidenceCursor, error) { + if len(raw) > 200 { + return nil, ErrRouteEvidenceInput + } + p := strings.Split(raw, ":") + if len(p) != 7 || p[0] != "v1" || !ValidRouteEvidenceKey(p[1], p[2]) { + return nil, ErrRouteEvidenceInput + } + values := [4]int64{} + for i, s := range p[3:] { + n, err := strconv.ParseInt(s, 10, 64) + if err != nil || n < 0 || strconv.FormatInt(n, 10) != s { + return nil, ErrRouteEvidenceInput + } + values[i] = n + } + // Bound before converting; this also keeps millisecond/microsecond comparisons safe. + if values[0] > 253402300799999 || values[1] > 253402300799999 || values[2] > 253402300799999999 || values[3] <= 0 { + return nil, ErrRouteEvidenceInput + } + c := &RouteEvidenceCursor{IATA: p[1], PathKey: p[2], Since: time.UnixMilli(values[0]).UTC(), Until: time.UnixMilli(values[1]).UTC(), HeardAt: time.UnixMicro(values[2]).UTC(), ID: values[3]} + if !ValidRouteEvidenceWindow(c.Since, c.Until) || c.HeardAt.Before(c.Since) || !c.HeardAt.Before(c.Until) { + return nil, ErrRouteEvidenceInput + } + return c, nil +} diff --git a/internal/api/route_evidence_test.go b/internal/api/route_evidence_test.go new file mode 100644 index 00000000..42c9a73a --- /dev/null +++ b/internal/api/route_evidence_test.go @@ -0,0 +1,25 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later +package api + +import ( + "strings" + "testing" + "time" +) + +func TestRouteEvidenceCursor(t *testing.T) { + c := RouteEvidenceCursor{IATA: "YOW", PathKey: "f097439148601d9f3291c474f82fa64c", Since: time.UnixMilli(1000).UTC(), Until: time.UnixMilli(3000).UTC(), HeardAt: time.UnixMicro(2000123).UTC(), ID: 9223372036854775806} + got, err := ParseRouteEvidenceCursor(c.String()) + if err != nil || *got != c { + t.Fatalf("cursor round-trip: %+v %v", got, err) + } + if got.HeardAt.Nanosecond() != 123000 { + t.Fatal("lost microsecond precision") + } + for _, bad := range []string{"", strings.Repeat("x", 300), "v2:YOW:x:1:2:3:4", strings.Replace(c.String(), "YOW", "TOOLONG", 1), strings.Replace(c.String(), c.PathKey, "bad", 1), strings.Replace(c.String(), "2000123", "4000000", 1), strings.TrimSuffix(c.String(), "9223372036854775806") + "0"} { + if _, err := ParseRouteEvidenceCursor(bad); err == nil { + t.Errorf("accepted %q", bad) + } + } +} diff --git a/internal/api/routes.go b/internal/api/routes.go index 3bc7eced..3f1bee1e 100644 --- a/internal/api/routes.go +++ b/internal/api/routes.go @@ -16,6 +16,7 @@ type RouteHop struct { // have been confirmed as high confidence. type KnownRoute struct { ID int64 `json:"id"` + PathKey string `json:"pathKey,omitempty"` // stable identity within this IATA; use for route evidence links IATA string `json:"iata"` HopCount int32 `json:"hopCount"` Hops []RouteHop `json:"hops"` diff --git a/internal/cache/cache_test.go b/internal/cache/cache_test.go index e58f0544..8184a17b 100644 --- a/internal/cache/cache_test.go +++ b/internal/cache/cache_test.go @@ -148,7 +148,7 @@ func (s *stubReader) GetObserverTelemetryBucketed(_ context.Context, _ uuid.UUID return nil, nil } -func (s *stubReader) GetObserverActivity(_ context.Context, _ uuid.UUID, _, _ time.Duration) (*api.ObserverActivity, error) { +func (s *stubReader) GetObserverActivity(_ context.Context, _ uuid.UUID, _, _ time.Duration, _ time.Time) (*api.ObserverActivity, error) { return nil, nil } @@ -418,3 +418,7 @@ func TestCachedReader_InvalidateObserver(t *testing.T) { t.Error("expected observer scopes key to be deleted") } } + +func (s *stubReader) GetRouteEvidence(_ context.Context, _, _ string, _ api.RouteEvidenceQuery) (*api.RouteEvidence, error) { + return nil, nil +} diff --git a/internal/cache/observer_window_test.go b/internal/cache/observer_window_test.go new file mode 100644 index 00000000..512e5800 --- /dev/null +++ b/internal/cache/observer_window_test.go @@ -0,0 +1,47 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package cache + +import ( + "context" + "testing" + "time" + + "github.com/MeshCore-Beacon/beacon-server/internal/api" + "github.com/google/uuid" +) + +type activityWindowReader struct { + stubReader + ends []time.Time +} + +func (r *activityWindowReader) GetObserverActivity(_ context.Context, _ uuid.UUID, _, _ time.Duration, until time.Time) (*api.ObserverActivity, error) { + r.ends = append(r.ends, until) + return &api.ObserverActivity{WindowEnd: until.UnixMilli()}, nil +} + +func TestObserverActivityCacheUsesEffectiveWindow(t *testing.T) { + client, redis := newTestClient(t) + inner := &activityWindowReader{} + reader := &CachedReader{inner: inner, c: client} + id := uuid.New() + anchor := time.Date(2026, 9, 29, 12, 0, 0, 0, time.UTC) + for _, offset := range []time.Duration{time.Second, time.Minute, 59 * time.Minute} { + if _, err := reader.GetObserverActivity(context.Background(), id, 24*time.Hour, time.Hour, anchor.Add(offset)); err != nil { + t.Fatal(err) + } + } + if len(inner.ends) != 1 || !inner.ends[0].Equal(anchor) || len(redis.Keys()) != 1 { + t.Fatalf("equivalent fixed windows missed the cache: %v, %v", inner.ends, redis.Keys()) + } + for _, until := range []time.Time{anchor.Add(time.Hour), {}, {}} { + if _, err := reader.GetObserverActivity(context.Background(), id, 24*time.Hour, time.Hour, until); err != nil { + t.Fatal(err) + } + } + if len(inner.ends) != 3 || !inner.ends[2].IsZero() || len(redis.Keys()) != 3 { + t.Fatalf("distinct/live windows aliased: %v", inner.ends) + } +} diff --git a/internal/cache/reader.go b/internal/cache/reader.go index 65551da8..a570df9e 100644 --- a/internal/cache/reader.go +++ b/internal/cache/reader.go @@ -372,10 +372,13 @@ func (cr *CachedReader) GetObserverTelemetryBucketed(ctx context.Context, observ } // GetObserverActivity implements [api.Reader]. -func (cr *CachedReader) GetObserverActivity(ctx context.Context, observerID uuid.UUID, window, interval time.Duration) (*api.ObserverActivity, error) { - key := keyObserverActivityPrefix + observerID.String() + ":" + window.String() + ":" + interval.String() +func (cr *CachedReader) GetObserverActivity(ctx context.Context, observerID uuid.UUID, window, interval time.Duration, until time.Time) (*api.ObserverActivity, error) { + if !until.IsZero() { + until = until.UTC().Truncate(interval) + } + key := keyObserverActivityPrefix + observerID.String() + ":" + window.String() + ":" + interval.String() + ":" + until.UTC().Format(time.RFC3339Nano) return getOrSet(ctx, cr.c, key, observerActivityTTL, func() (*api.ObserverActivity, error) { - return cr.inner.GetObserverActivity(ctx, observerID, window, interval) + return cr.inner.GetObserverActivity(ctx, observerID, window, interval, until) }) } @@ -464,6 +467,11 @@ func (cr *CachedReader) SearchKnownRoutes(ctx context.Context, iata, fromHash, t return cr.inner.SearchKnownRoutes(ctx, iata, fromHash, toHash) } +// Precise cursor/window combinations are intentionally passed through, like route lists. +func (cr *CachedReader) GetRouteEvidence(ctx context.Context, iata, key string, query api.RouteEvidenceQuery) (*api.RouteEvidence, error) { + return cr.inner.GetRouteEvidence(ctx, iata, key, query) +} + // SearchCrossIATARoutes implements [api.Reader]. func (cr *CachedReader) SearchCrossIATARoutes(ctx context.Context, fromHash, fromIATA, toHash, toIATA string) ([]api.CrossIATARoute, error) { return cr.inner.SearchCrossIATARoutes(ctx, fromHash, fromIATA, toHash, toIATA) diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index 998c2051..0c4d7f82 100644 --- a/internal/ingest/ingest.go +++ b/internal/ingest/ingest.go @@ -89,7 +89,7 @@ type DB interface { UpsertObserver(ctx context.Context, pubkey []byte) (uuid.UUID, string, error) // UpsertObserverBroker records that this observer was seen on brokerName. - UpsertObserverBroker(ctx context.Context, observerID uuid.UUID, brokerName string) error + UpsertObserverBroker(ctx context.Context, observerID uuid.UUID, brokerName string, isPacket bool) error // UpsertIATA auto-creates an iata_codes row if it doesn't exist yet. UpsertIATA(ctx context.Context, iata string) error diff --git a/internal/ingest/ingest_test.go b/internal/ingest/ingest_test.go index ea206c5d..2090f0e5 100644 --- a/internal/ingest/ingest_test.go +++ b/internal/ingest/ingest_test.go @@ -201,8 +201,10 @@ func (s *stubDB) SetNodeCapability(_ context.Context, nodeID uuid.UUID, paths, t func (s *stubDB) UpsertObserver(_ context.Context, _ []byte) (uuid.UUID, string, error) { return uuid.Nil, "", nil } -func (s *stubDB) UpsertObserverBroker(_ context.Context, _ uuid.UUID, _ string) error { return nil } -func (s *stubDB) UpsertIATA(_ context.Context, _ string) error { return nil } +func (s *stubDB) UpsertObserverBroker(_ context.Context, _ uuid.UUID, _ string, _ bool) error { + return nil +} +func (s *stubDB) UpsertIATA(_ context.Context, _ string) error { return nil } func (s *stubDB) UpsertPacket(_ context.Context, _ UpsertPacketParams) (bool, error) { return false, nil } diff --git a/internal/ingest/neighbors.go b/internal/ingest/neighbors.go index f96a32a1..3def0c9a 100644 --- a/internal/ingest/neighbors.go +++ b/internal/ingest/neighbors.go @@ -48,7 +48,7 @@ func (w *Worker) handleNeighbors(ctx context.Context, iata, pubkeyHex string, ra w.log.Error(fmt.Sprintf("db: upsert observer failed in neighbors from %s", pubkeyHex), "error", err) return } - if err := w.db.UpsertObserverBroker(ctx, observerID, w.cfg.BrokerName); err != nil { + if err := w.db.UpsertObserverBroker(ctx, observerID, w.cfg.BrokerName, false); err != nil { w.log.Error(fmt.Sprintf("db: upsert observer broker failed in neighbors from %s", pubkeyHex), "error", err) } diff --git a/internal/ingest/packet.go b/internal/ingest/packet.go index 92f3da78..bcc9ca58 100644 --- a/internal/ingest/packet.go +++ b/internal/ingest/packet.go @@ -324,7 +324,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ w.log.Error(fmt.Sprintf("db: upsert observer failed with packet from %s/%s", iata, pubkeyHex), "error", err) return } - err = w.db.UpsertObserverBroker(ctx, id, w.cfg.BrokerName) + err = w.db.UpsertObserverBroker(ctx, id, w.cfg.BrokerName, true) if err != nil { w.log.Error(fmt.Sprintf("db: update observer broker failed with packet from %s/%s", iata, pubkeyHex), "error", err) return diff --git a/internal/ingest/status.go b/internal/ingest/status.go index 31c80647..3c693db6 100644 --- a/internal/ingest/status.go +++ b/internal/ingest/status.go @@ -87,7 +87,7 @@ func (w *Worker) handleStatus(ctx context.Context, pubkeyHex string, raw []byte) if w.onObserverUpsert != nil { w.onObserverUpsert(ctx, id) } - if err := w.db.UpsertObserverBroker(ctx, id, w.cfg.BrokerName); err != nil { + if err := w.db.UpsertObserverBroker(ctx, id, w.cfg.BrokerName, false); err != nil { w.log.Error(fmt.Sprintf("db: upsert observer broker failed in status from %s", pubkeyHex), "error", err) } params := UpdateObserverStatusParams{ diff --git a/internal/presence/broker_freshness_test.go b/internal/presence/broker_freshness_test.go new file mode 100644 index 00000000..887a6e7c --- /dev/null +++ b/internal/presence/broker_freshness_test.go @@ -0,0 +1,50 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later +package presence + +import ( + "context" + "github.com/google/uuid" + "testing" + "time" +) + +type brokerTestStore struct { + Store + initial []bool + seen, packets []time.Time +} + +func (s *brokerTestStore) UpsertObserverBroker(_ context.Context, _ uuid.UUID, _ string, packet bool) error { + s.initial = append(s.initial, packet) + return nil +} +func (s *brokerTestStore) TouchObserverBrokers(_ context.Context, _ []uuid.UUID, _ []string, seen, packets []time.Time) error { + s.seen = seen + s.packets = packets + return nil +} + +func TestBrokerPresenceDoesNotInventPacketFreshness(t *testing.T) { + ctx := context.Background() + store := &brokerTestStore{} + c := New(store, time.Second, time.Minute) + now := time.Now() + c.now = func() time.Time { return now } + id := uuid.New() + c.UpsertObserverBroker(ctx, id, "one", false) + c.UpsertObserverBroker(ctx, id, "one", false) + c.Flush(ctx) + if len(store.initial) != 1 || store.initial[0] || !store.packets[0].IsZero() { + t.Fatal("status-only activity invented a packet") + } + now = now.Add(time.Minute) + packetTime := now + c.UpsertObserverBroker(ctx, id, "one", true) + now = now.Add(time.Minute) + c.UpsertObserverBroker(ctx, id, "one", false) + c.Flush(ctx) + if !store.seen[0].Equal(now) || !store.packets[0].Equal(packetTime) { + t.Fatal("later status changed packet freshness") + } +} diff --git a/internal/presence/coalescer.go b/internal/presence/coalescer.go index 9315f433..517f1217 100644 --- a/internal/presence/coalescer.go +++ b/internal/presence/coalescer.go @@ -31,7 +31,7 @@ type Store interface { // TouchObserverBrokers applies coalesced last_seen/last_packet_at bumps // for the given (observer, broker) pairs in one statement. - TouchObserverBrokers(ctx context.Context, ids []uuid.UUID, brokers []string, seen []time.Time) error + TouchObserverBrokers(ctx context.Context, ids []uuid.UUID, brokers []string, seen, packets []time.Time) error // TouchPackets applies coalesced last_heard_at bumps for the given packet // hashes in one statement. @@ -51,6 +51,8 @@ type observerBump struct { count int32 } +type brokerBump struct{ seen, packet time.Time } + type brokerKey struct { id uuid.UUID broker string @@ -68,7 +70,7 @@ type Coalescer struct { identities map[string]identity // pubkey -> observer row dirtyObservers map[uuid.UUID]observerBump knownBrokers map[brokerKey]struct{} - dirtyBrokers map[brokerKey]time.Time + dirtyBrokers map[brokerKey]brokerBump knownIATAs map[string]struct{} packetsSeen map[string]time.Time // hash -> last observation dirtyPackets map[string]time.Time @@ -86,7 +88,7 @@ func New(store Store, flushInterval, packetTTL time.Duration) *Coalescer { identities: make(map[string]identity), dirtyObservers: make(map[uuid.UUID]observerBump), knownBrokers: make(map[brokerKey]struct{}), - dirtyBrokers: make(map[brokerKey]time.Time), + dirtyBrokers: make(map[brokerKey]brokerBump), knownIATAs: make(map[string]struct{}), packetsSeen: make(map[string]time.Time), dirtyPackets: make(map[string]time.Time), @@ -122,17 +124,22 @@ func (c *Coalescer) UpsertObserver(ctx context.Context, pubkey []byte) (uuid.UUI // UpsertObserverBroker writes through the first time a pair is seen and // records a bump afterwards. -func (c *Coalescer) UpsertObserverBroker(ctx context.Context, observerID uuid.UUID, brokerName string) error { +func (c *Coalescer) UpsertObserverBroker(ctx context.Context, observerID uuid.UUID, brokerName string, isPacket bool) error { key := brokerKey{id: observerID, broker: brokerName} c.mu.Lock() if _, ok := c.knownBrokers[key]; ok { - c.dirtyBrokers[key] = c.now() + bump := c.dirtyBrokers[key] + bump.seen = c.now() + if isPacket { + bump.packet = bump.seen + } + c.dirtyBrokers[key] = bump c.mu.Unlock() return nil } c.mu.Unlock() - if err := c.Store.UpsertObserverBroker(ctx, observerID, brokerName); err != nil { + if err := c.Store.UpsertObserverBroker(ctx, observerID, brokerName, isPacket); err != nil { return err } c.mu.Lock() @@ -248,7 +255,7 @@ func (c *Coalescer) Flush(ctx context.Context) { brokers := c.dirtyBrokers packets := c.dirtyPackets c.dirtyObservers = make(map[uuid.UUID]observerBump) - c.dirtyBrokers = make(map[brokerKey]time.Time) + c.dirtyBrokers = make(map[brokerKey]brokerBump) c.dirtyPackets = make(map[string]time.Time) cutoff := c.now().Add(-c.packetTTL) for key, seen := range c.packetsSeen { @@ -266,12 +273,14 @@ func (c *Coalescer) Flush(ctx context.Context) { ids := make([]uuid.UUID, 0, len(brokers)) names := make([]string, 0, len(brokers)) seen := make([]time.Time, 0, len(brokers)) + packetTimes := make([]time.Time, 0, len(brokers)) for key, ts := range brokers { ids = append(ids, key.id) names = append(names, key.broker) - seen = append(seen, ts) + seen = append(seen, ts.seen) + packetTimes = append(packetTimes, ts.packet) } - if err := c.Store.TouchObserverBrokers(ctx, ids, names, seen); err != nil { + if err := c.Store.TouchObserverBrokers(ctx, ids, names, seen, packetTimes); err != nil { slog.Error(fmt.Sprintf("presence: flush observer brokers failed (%d rows dropped)", len(ids)), "component", "presence", "error", err) } } diff --git a/sqlc.yaml b/sqlc.yaml index b5e86b89..032c9911 100644 --- a/sqlc.yaml +++ b/sqlc.yaml @@ -30,6 +30,3 @@ sql: go_type: {type: "string", pointer: true} - column: "mv_top_advertisers_by_iata.name" go_type: {type: "string", pointer: true} - - - column: "mv_observer_activity_hourly.payload_type" - go_type: {type: "int16", pointer: true}