From 7650d9b742800d6896fd7cb8d1d9f5b137d19b29 Mon Sep 17 00:00:00 2001 From: MrAlders0n Date: Thu, 10 Sep 2026 06:37:45 -0400 Subject: [PATCH] feat(observers): per-observer activity endpoint GET /observers/{id}/activity returns bucketed observation counts, LoRa airtime, SNR/RSSI aggregates and a payload-type breakdown for one observer. Airtime is computed at ingest from the raw frame length and stored on the observation; intervals of 1h and up are served from an hourly matview so no request path joins back to packets. --- db/migrations/030_observation_airtime.sql | 2 + .../031_observation_airtime_backfill.sql | 30 ++ db/migrations/032_mv_observer_activity.sql | 24 ++ db/observers.go | 124 ++++++ db/observers_test.go | 367 ++++++++++++++++++ db/packets.go | 1 + db/queries/queries.sql | 58 ++- db/sqlc/mock/querier.go | 74 ++++ db/sqlc/models.go | 15 + db/sqlc/querier.go | 9 + db/sqlc/queries.sql.go | 233 ++++++++++- db/stats.go | 4 + docs/docs.go | 128 ++++++ docs/swagger.json | 128 ++++++ docs/swagger.yaml | 85 ++++ internal/api/handlers/observers.go | 76 ++++ internal/api/handlers/observers_test.go | 228 +++++++++++ internal/api/handlers/stub_reader_test.go | 8 + internal/api/observers.go | 28 ++ internal/api/reader.go | 5 + internal/background/tasks.go | 2 + internal/background/tasks_test.go | 7 +- internal/cache/cache_test.go | 4 + internal/cache/reader.go | 12 + internal/ingest/frame_length_test.go | 145 +++++++ internal/ingest/packet.go | 10 + internal/lora/airtime.go | 56 +++ internal/lora/airtime_test.go | 122 ++++++ 28 files changed, 1977 insertions(+), 8 deletions(-) create mode 100644 db/migrations/030_observation_airtime.sql create mode 100644 db/migrations/031_observation_airtime_backfill.sql create mode 100644 db/migrations/032_mv_observer_activity.sql create mode 100644 internal/ingest/frame_length_test.go create mode 100644 internal/lora/airtime.go create mode 100644 internal/lora/airtime_test.go diff --git a/db/migrations/030_observation_airtime.sql b/db/migrations/030_observation_airtime.sql new file mode 100644 index 0000000..1a7c29e --- /dev/null +++ b/db/migrations/030_observation_airtime.sql @@ -0,0 +1,2 @@ +-- Time-on-air per observation so per-observer activity never joins back to packets. +ALTER TABLE packet_observations ADD COLUMN airtime_ms REAL; diff --git a/db/migrations/031_observation_airtime_backfill.sql b/db/migrations/031_observation_airtime_backfill.sql new file mode 100644 index 0000000..0a74f7d --- /dev/null +++ b/db/migrations/031_observation_airtime_backfill.sql @@ -0,0 +1,30 @@ +-- Separate from 030 so the ALTER's ACCESS EXCLUSIVE lock is released before this backfill runs. + +-- Backfill only: mirrors internal/lora.TimeOnAirMs (RadioLib getTimeOnAir, CRC on, explicit header). +CREATE FUNCTION lora_airtime_ms_backfill(frame_len int, sf int, bw real, cr int) RETURNS real +LANGUAGE sql IMMUTABLE AS $$ + SELECT ( + (CASE WHEN sf <= 8 THEN 32 ELSE 16 END) + 4.25 + + 8 + GREATEST(ceil((8*frame_len - 4*sf + 44)::double precision + / (4*(sf - 2*(CASE WHEN power(2, sf)/bw >= 16 THEN 1 ELSE 0 END)))) * cr, 0) + ) * (power(2, sf)/bw) +$$; + +-- Last 7 days like 015; parallelism off so the join spills to disk, not /dev/shm. +SET max_parallel_workers_per_gather = 0; + +UPDATE packet_observations po +SET airtime_ms = lora_airtime_ms_backfill( + 1 + CASE WHEN p.transport_codes_present THEN 4 ELSE 0 END + 1 + + COALESCE(octet_length(po.path_bytes), 0) + octet_length(p.raw_payload), + po.spread_factor, po.bandwidth_khz, po.coding_rate) +FROM packets p +WHERE p.packet_hash = po.packet_hash + AND po.heard_at > NOW() - INTERVAL '7 days' + AND po.airtime_ms IS NULL + AND po.spread_factor BETWEEN 7 AND 12 + AND po.bandwidth_khz > 0 + AND po.coding_rate BETWEEN 5 AND 8; + +RESET max_parallel_workers_per_gather; +DROP FUNCTION lora_airtime_ms_backfill(int, int, real, int); diff --git a/db/migrations/032_mv_observer_activity.sql b/db/migrations/032_mv_observer_activity.sql new file mode 100644 index 0000000..3640759 --- /dev/null +++ b/db/migrations/032_mv_observer_activity.sql @@ -0,0 +1,24 @@ +-- Hourly per-observer rollup so 7d/30d activity sums a few thousand rows instead of scanning +-- a month of observations. payload_type is a key so the breakdown comes from the same rows. +-- sqlc types the cast aggregates as non-null even though airtime_ms/snr_*/rssi_sum are NULL for +-- empty groups, so readers must gate on the matching *_n counts. +CREATE MATERIALIZED VIEW mv_observer_activity_hourly AS +SELECT + observer_id, + payload_type, + date_trunc('hour', heard_at)::timestamptz AS bucket, + COUNT(*)::bigint AS observations, + SUM(airtime_ms)::real AS airtime_ms, + COUNT(airtime_ms)::bigint AS airtime_n, + SUM(snr) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::real AS snr_sum, + COUNT(snr) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::bigint AS snr_n, + MIN(snr) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::real AS snr_min, + SUM(rssi) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::bigint AS rssi_sum, + COUNT(rssi) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::bigint AS rssi_n +FROM packet_observations +WHERE heard_at > NOW() - INTERVAL '30 days' + AND payload_type IS NOT NULL +GROUP BY observer_id, payload_type, date_trunc('hour', heard_at); + +CREATE UNIQUE INDEX idx_mv_observer_activity_hourly + ON mv_observer_activity_hourly(observer_id, payload_type, bucket); diff --git a/db/observers.go b/db/observers.go index c8036ca..1a71329 100644 --- a/db/observers.go +++ b/db/observers.go @@ -13,6 +13,7 @@ import ( sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" "github.com/MeshCore-Beacon/beacon-server/internal/api" "github.com/MeshCore-Beacon/beacon-server/internal/ingest" + "github.com/MeshCore-Beacon/beacon-server/internal/lora" "github.com/google/uuid" "github.com/jackc/pgx/v5/pgtype" ) @@ -226,6 +227,129 @@ func (s *Store) GetObserverTelemetryBucketed(ctx context.Context, observerID uui return points, nil } +// 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) { + obs, err := s.q.GetObserverByID(ctx, observerID) + if err != nil { + return nil, err + } + activity := &api.ObserverActivity{} + // 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 { + activity.Radio = &api.ObserverActivityRadio{ + FreqMHz: obs.RadioFreqMhz, + SF: *obs.RadioSf, + BWKHz: *obs.RadioBwKhz, + CR: *obs.RadioCr, + 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} + + if interval >= time.Hour { + rows, err := s.q.GetObserverActivityHourly(ctx, sqlc.GetObserverActivityHourlyParams{ + ObserverID: observerID, + Column2: sinceTS, + Column3: binWidth, + }) + if err != nil { + return nil, err + } + activity.Points = make([]api.ObserverActivityPoint, 0, len(rows)) + for _, r := range rows { + p := api.ObserverActivityPoint{T: r.Bucket.Time.UnixMilli(), Observations: r.Observations} + if r.AirtimeN > 0 { + airtime := r.AirtimeMs + p.AirtimeMs = &airtime + } + if r.SnrN > 0 { + avg := r.SnrSum / float32(r.SnrN) + min := r.SnrMin + p.SNRAvg, p.SNRMin = &avg, &min + } + if r.RssiN > 0 { + avg := float32(r.RssiSum) / float32(r.RssiN) + p.RSSIAvg = &avg + } + activity.Points = append(activity.Points, p) + } + typeRows, err := s.q.GetObserverActivityHourlyPayloadTypes(ctx, sqlc.GetObserverActivityHourlyPayloadTypesParams{ + ObserverID: observerID, + Column2: sinceTS, + }) + if err != nil { + return nil, err + } + activity.PayloadTypes = make([]api.PayloadBreakdownItem, 0, len(typeRows)) + for _, v := range typeRows { + if v.PayloadType == nil { + continue + } + activity.PayloadTypes = append(activity.PayloadTypes, api.PayloadBreakdownItem{ + PayloadType: *v.PayloadType, + PayloadTypeName: api.PayloadTypeName(*v.PayloadType), + Count: v.Count, + }) + } + return activity, nil + } + + rows, err := s.q.GetObserverActivityRaw(ctx, sqlc.GetObserverActivityRawParams{ + ObserverID: observerID, + Column2: sinceTS, + Column3: binWidth, + }) + if err != nil { + return nil, err + } + activity.Points = make([]api.ObserverActivityPoint, 0, len(rows)) + for _, r := range rows { + p := api.ObserverActivityPoint{T: r.Bucket.Time.UnixMilli(), Observations: 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 r.RssiN > 0 { + avg := r.RssiAvg + p.RSSIAvg = &avg + } + activity.Points = append(activity.Points, p) + } + typeRows, err := s.q.GetObserverActivityRawPayloadTypes(ctx, sqlc.GetObserverActivityRawPayloadTypesParams{ + ObserverID: observerID, + Column2: sinceTS, + }) + if err != nil { + return nil, err + } + activity.PayloadTypes = make([]api.PayloadBreakdownItem, 0, len(typeRows)) + for _, v := range typeRows { + if v.PayloadType == nil { + continue + } + activity.PayloadTypes = append(activity.PayloadTypes, api.PayloadBreakdownItem{ + PayloadType: *v.PayloadType, + PayloadTypeName: api.PayloadTypeName(*v.PayloadType), + Count: v.Count, + }) + } + return activity, nil +} + func (s *Store) ListObserverAdverts(ctx context.Context, observerID uuid.UUID, cursor int64, limit int32) (api.Page[api.AdvertObservation], error) { rows, err := s.q.ListObserverAdverts(ctx, sqlc.ListObserverAdvertsParams{ ObserverID: observerID, diff --git a/db/observers_test.go b/db/observers_test.go index f6f7572..76ccf54 100644 --- a/db/observers_test.go +++ b/db/observers_test.go @@ -11,7 +11,9 @@ import ( sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" mockdb "github.com/MeshCore-Beacon/beacon-server/db/sqlc/mock" + "github.com/MeshCore-Beacon/beacon-server/internal/api" "github.com/google/uuid" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgtype" "go.uber.org/mock/gomock" ) @@ -550,3 +552,368 @@ func TestIsObserverByPubkey_NotFound(t *testing.T) { t.Error("expected false for missing observer") } } + +func activityObserver(sf, cr *int16, bw, freq *float32) sqlc.Observer { + return sqlc.Observer{ + ID: uuid.MustParse("00000000-0000-0000-0000-000000000001"), + RadioFreqMhz: freq, + RadioSf: sf, + RadioBwKhz: bw, + RadioCr: cr, + } +} + +func i16(v int16) *int16 { return &v } +func f32(v float32) *float32 { return &v } + +func TestGetObserverActivity_HourlyFoldNoSignal(t *testing.T) { + ctrl := gomock.NewController(t) + mock := mockdb.NewMockQuerier(ctrl) + observerID := uuid.MustParse("00000000-0000-0000-0000-000000000001") + + mock.EXPECT().GetObserverByID(gomock.Any(), observerID). + Return(activityObserver(i16(10), i16(5), f32(62.5), f32(910.525)), nil) + mock.EXPECT().GetObserverActivityHourly(gomock.Any(), gomock.Any()). + Return([]sqlc.GetObserverActivityHourlyRow{{ + Bucket: pgtype.Timestamptz{Time: time.UnixMilli(1700000000000), Valid: true}, + Observations: 12, + AirtimeMs: 0, + AirtimeN: 0, + SnrSum: 0, + SnrN: 0, + SnrMin: 0, + RssiSum: 0, + RssiN: 0, + }}, nil) + mock.EXPECT().GetObserverActivityHourlyPayloadTypes(gomock.Any(), gomock.Any()). + Return([]sqlc.GetObserverActivityHourlyPayloadTypesRow{ + {PayloadType: i16(4), Count: 9}, + {PayloadType: nil, Count: 3}, + }, nil) + + store := &Store{q: mock} + got, err := store.GetObserverActivity(context.Background(), observerID, 24*time.Hour, time.Hour) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(got.Points) != 1 { + t.Fatalf("expected 1 point, got %d", len(got.Points)) + } + p := got.Points[0] + if p.T != 1700000000000 { + t.Errorf("expected T 1700000000000, got %d", p.T) + } + if p.Observations != 12 { + t.Errorf("expected 12 observations, got %d", p.Observations) + } + if p.AirtimeMs != nil { + t.Errorf("expected nil AirtimeMs, got %v", *p.AirtimeMs) + } + if p.SNRAvg != nil || p.SNRMin != nil { + t.Errorf("expected nil SNR fields, got %v %v", p.SNRAvg, p.SNRMin) + } + 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 got.PayloadTypes[0].PayloadTypeName != api.PayloadTypeName(4) { + t.Errorf("expected payload type name %q, got %q", api.PayloadTypeName(4), got.PayloadTypes[0].PayloadTypeName) + } + if got.Radio == nil { + t.Fatal("expected radio") + } + if got.Radio.SF != 10 || got.Radio.BWKHz != 62.5 || got.Radio.CR != 5 || got.Radio.PreambleSymbols != 16 { + t.Errorf("unexpected radio %+v", *got.Radio) + } + if got.Radio.FreqMHz == nil || *got.Radio.FreqMHz != 910.525 { + t.Errorf("unexpected freq %v", got.Radio.FreqMHz) + } + if got.Range != "" || got.Interval != "" { + t.Errorf("expected range/interval left for the handler, got %q %q", got.Range, got.Interval) + } +} + +func TestGetObserverActivity_HourlyFoldWeightedAverages(t *testing.T) { + ctrl := gomock.NewController(t) + mock := mockdb.NewMockQuerier(ctrl) + observerID := uuid.MustParse("00000000-0000-0000-0000-000000000001") + + mock.EXPECT().GetObserverByID(gomock.Any(), observerID). + Return(activityObserver(i16(7), i16(5), f32(250), nil), nil) + mock.EXPECT().GetObserverActivityHourly(gomock.Any(), gomock.Any()). + Return([]sqlc.GetObserverActivityHourlyRow{{ + Bucket: pgtype.Timestamptz{Time: time.UnixMilli(1700000000000), Valid: true}, + Observations: 8, + AirtimeMs: 123.5, + AirtimeN: 3, + SnrSum: 30, + SnrN: 4, + SnrMin: -3.5, + RssiSum: -400, + RssiN: 4, + }}, nil) + mock.EXPECT().GetObserverActivityHourlyPayloadTypes(gomock.Any(), gomock.Any()). + Return([]sqlc.GetObserverActivityHourlyPayloadTypesRow{}, nil) + + store := &Store{q: mock} + got, err := store.GetObserverActivity(context.Background(), observerID, 7*24*time.Hour, 6*time.Hour) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + p := got.Points[0] + if p.AirtimeMs == nil || *p.AirtimeMs != 123.5 { + t.Errorf("expected AirtimeMs 123.5, got %v", p.AirtimeMs) + } + if p.SNRAvg == nil || *p.SNRAvg != 7.5 { + t.Errorf("expected SNRAvg 7.5, got %v", p.SNRAvg) + } + if p.SNRMin == nil || *p.SNRMin != -3.5 { + t.Errorf("expected SNRMin -3.5, got %v", p.SNRMin) + } + if p.RSSIAvg == nil || *p.RSSIAvg != -100 { + t.Errorf("expected RSSIAvg -100, got %v", p.RSSIAvg) + } + if got.PayloadTypes == nil { + t.Error("expected non-nil PayloadTypes slice") + } + if got.Radio.FreqMHz != nil { + t.Errorf("expected nil freq, got %v", *got.Radio.FreqMHz) + } + if got.Radio.PreambleSymbols != 32 { + t.Errorf("expected 32 preamble symbols at SF7, got %d", got.Radio.PreambleSymbols) + } +} + +func TestGetObserverActivity_RawPath(t *testing.T) { + ctrl := gomock.NewController(t) + mock := mockdb.NewMockQuerier(ctrl) + observerID := uuid.MustParse("00000000-0000-0000-0000-000000000001") + + mock.EXPECT().GetObserverByID(gomock.Any(), observerID). + Return(activityObserver(i16(10), i16(5), f32(62.5), nil), nil) + mock.EXPECT().GetObserverActivityRaw(gomock.Any(), gomock.Any()). + Return([]sqlc.GetObserverActivityRawRow{{ + Bucket: pgtype.Timestamptz{Time: time.UnixMilli(1700000000000), Valid: true}, + Observations: 5, + AirtimeMs: 50, + AirtimeN: 5, + SnrAvg: 4.25, + SnrMin: 1.5, + SnrN: 2, + RssiAvg: -95.5, + RssiN: 0, + }}, nil) + mock.EXPECT().GetObserverActivityRawPayloadTypes(gomock.Any(), gomock.Any()). + 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) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + p := got.Points[0] + if p.Observations != 5 { + t.Errorf("expected 5 observations, got %d", p.Observations) + } + if p.AirtimeMs == nil || *p.AirtimeMs != 50 { + t.Errorf("expected AirtimeMs 50, got %v", p.AirtimeMs) + } + if p.SNRAvg == nil || *p.SNRAvg != 4.25 { + t.Errorf("expected SNRAvg 4.25, got %v", p.SNRAvg) + } + if p.SNRMin == nil || *p.SNRMin != 1.5 { + t.Errorf("expected SNRMin 1.5, got %v", p.SNRMin) + } + if p.RSSIAvg != nil { + t.Errorf("expected nil RSSIAvg with rssi_n 0, got %v", *p.RSSIAvg) + } + if len(got.PayloadTypes) != 1 || got.PayloadTypes[0].Count != 5 { + t.Errorf("unexpected payload types %+v", got.PayloadTypes) + } +} + +func TestGetObserverActivity_RawFoldNoSignal(t *testing.T) { + ctrl := gomock.NewController(t) + mock := mockdb.NewMockQuerier(ctrl) + observerID := uuid.MustParse("00000000-0000-0000-0000-000000000001") + + mock.EXPECT().GetObserverByID(gomock.Any(), observerID). + Return(activityObserver(i16(10), i16(5), f32(62.5), nil), nil) + mock.EXPECT().GetObserverActivityRaw(gomock.Any(), gomock.Any()). + Return([]sqlc.GetObserverActivityRawRow{{ + Bucket: pgtype.Timestamptz{Time: time.UnixMilli(1700000000000), Valid: true}, + Observations: 7, + AirtimeMs: 0, + AirtimeN: 0, + SnrAvg: 0, + SnrMin: 0, + SnrN: 0, + RssiAvg: 0, + RssiN: 0, + }}, nil) + mock.EXPECT().GetObserverActivityRawPayloadTypes(gomock.Any(), gomock.Any()). + Return([]sqlc.GetObserverActivityRawPayloadTypesRow{}, nil) + + store := &Store{q: mock} + got, err := store.GetObserverActivity(context.Background(), observerID, 6*time.Hour, 15*time.Minute) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + p := got.Points[0] + if p.Observations != 7 { + t.Errorf("expected 7 observations, got %d", p.Observations) + } + if p.AirtimeMs != nil { + t.Errorf("expected nil AirtimeMs with airtime_n 0, got %v", *p.AirtimeMs) + } + if p.SNRAvg != nil || p.SNRMin != nil { + t.Errorf("expected nil SNR fields with snr_n 0, got %v %v", p.SNRAvg, p.SNRMin) + } + if p.RSSIAvg != nil { + t.Errorf("expected nil RSSIAvg with rssi_n 0, got %v", *p.RSSIAvg) + } +} + +func TestGetObserverActivity_RawPathEmpty(t *testing.T) { + ctrl := gomock.NewController(t) + mock := mockdb.NewMockQuerier(ctrl) + observerID := uuid.MustParse("00000000-0000-0000-0000-000000000001") + + mock.EXPECT().GetObserverByID(gomock.Any(), observerID). + Return(activityObserver(i16(10), i16(5), f32(62.5), nil), nil) + mock.EXPECT().GetObserverActivityRaw(gomock.Any(), gomock.Any()). + Return([]sqlc.GetObserverActivityRawRow{}, nil) + mock.EXPECT().GetObserverActivityRawPayloadTypes(gomock.Any(), gomock.Any()). + Return([]sqlc.GetObserverActivityRawPayloadTypesRow{}, nil) + + store := &Store{q: mock} + got, err := store.GetObserverActivity(context.Background(), observerID, 6*time.Hour, 15*time.Minute) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got.Points == nil || len(got.Points) != 0 { + t.Errorf("expected empty non-nil points, got %#v", got.Points) + } + if got.PayloadTypes == nil || len(got.PayloadTypes) != 0 { + t.Errorf("expected empty non-nil payload types, got %#v", got.PayloadTypes) + } +} + +func TestGetObserverActivity_HourlyPathAtOneHour(t *testing.T) { + ctrl := gomock.NewController(t) + mock := mockdb.NewMockQuerier(ctrl) + observerID := uuid.MustParse("00000000-0000-0000-0000-000000000001") + + mock.EXPECT().GetObserverByID(gomock.Any(), observerID). + Return(activityObserver(i16(10), i16(5), f32(62.5), nil), nil) + mock.EXPECT().GetObserverActivityHourly(gomock.Any(), gomock.Any()). + Return([]sqlc.GetObserverActivityHourlyRow{}, nil) + mock.EXPECT().GetObserverActivityHourlyPayloadTypes(gomock.Any(), gomock.Any()). + Return([]sqlc.GetObserverActivityHourlyPayloadTypesRow{}, nil) + + store := &Store{q: mock} + got, err := store.GetObserverActivity(context.Background(), observerID, 24*time.Hour, time.Hour) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got.Points == nil || len(got.Points) != 0 { + t.Errorf("expected empty non-nil points, got %#v", got.Points) + } +} + +func TestGetObserverActivity_UnknownObserver(t *testing.T) { + ctrl := gomock.NewController(t) + mock := mockdb.NewMockQuerier(ctrl) + 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) + if !errors.Is(err, pgx.ErrNoRows) { + t.Fatalf("expected pgx.ErrNoRows, got %v", err) + } + if got != nil { + t.Errorf("expected nil result, got %+v", got) + } +} + +func TestGetObserverActivity_RadioNilWhenIncomplete(t *testing.T) { + cases := []struct { + name string + sf, cr *int16 + bw *float32 + }{ + {name: "sf nil", sf: nil, cr: i16(5), bw: f32(62.5)}, + {name: "sf zero", sf: i16(0), cr: i16(5), bw: f32(62.5)}, + {name: "sf below lora range", sf: i16(6), cr: i16(5), bw: f32(62.5)}, + {name: "sf above lora range", sf: i16(13), cr: i16(5), bw: f32(62.5)}, + {name: "bw zero", sf: i16(10), cr: i16(5), bw: f32(0)}, + {name: "cr nil", sf: i16(10), cr: nil, bw: f32(62.5)}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + mock := mockdb.NewMockQuerier(ctrl) + observerID := uuid.MustParse("00000000-0000-0000-0000-000000000001") + + mock.EXPECT().GetObserverByID(gomock.Any(), observerID). + Return(activityObserver(tc.sf, tc.cr, tc.bw, nil), nil) + mock.EXPECT().GetObserverActivityHourly(gomock.Any(), gomock.Any()). + Return([]sqlc.GetObserverActivityHourlyRow{}, nil) + mock.EXPECT().GetObserverActivityHourlyPayloadTypes(gomock.Any(), gomock.Any()). + Return([]sqlc.GetObserverActivityHourlyPayloadTypesRow{}, nil) + + store := &Store{q: mock} + got, err := store.GetObserverActivity(context.Background(), observerID, 24*time.Hour, time.Hour) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got.Radio != nil { + t.Errorf("expected nil radio, got %+v", *got.Radio) + } + }) + } +} + +func TestGetObserverActivity_SinceAlignedToInterval(t *testing.T) { + ctrl := gomock.NewController(t) + mock := mockdb.NewMockQuerier(ctrl) + observerID := uuid.MustParse("00000000-0000-0000-0000-000000000001") + window, interval := 24*time.Hour, 6*time.Hour + + var gotSince time.Time + var gotInterval pgtype.Interval + mock.EXPECT().GetObserverByID(gomock.Any(), observerID). + Return(activityObserver(i16(10), i16(5), f32(62.5), nil), nil) + mock.EXPECT().GetObserverActivityHourly(gomock.Any(), gomock.Any()). + DoAndReturn(func(_ context.Context, arg sqlc.GetObserverActivityHourlyParams) ([]sqlc.GetObserverActivityHourlyRow, error) { + gotSince, gotInterval = arg.Column2.Time, arg.Column3 + return nil, nil + }) + mock.EXPECT().GetObserverActivityHourlyPayloadTypes(gomock.Any(), gomock.Any()). + Return([]sqlc.GetObserverActivityHourlyPayloadTypesRow{}, nil) + + before := time.Now().Add(-window) + store := &Store{q: mock} + if _, err := store.GetObserverActivity(context.Background(), observerID, window, interval); err != nil { + t.Fatalf("unexpected error: %v", err) + } + after := time.Now().Add(-window) + + if gotSince.UnixNano()%int64(interval) != 0 { + t.Errorf("since %s is not aligned to %s", gotSince, interval) + } + if gotSince.Before(before) { + t.Errorf("since %s is before the window start %s", gotSince, before) + } + if !gotSince.Before(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() { + t.Errorf("expected interval %d us, got %+v", interval.Microseconds(), gotInterval) + } +} diff --git a/db/packets.go b/db/packets.go index b03eda6..ddf0ad2 100644 --- a/db/packets.go +++ b/db/packets.go @@ -586,6 +586,7 @@ func (s *Store) InsertObservation(ctx context.Context, o ingest.InsertObservatio SourceBroker: &o.SourceBroker, PayloadType: &o.PayloadType, ResolvedEndpoints: o.ResolvedEndpoints, + AirtimeMs: o.AirtimeMs, } row, err := s.q.InsertObservation(ctx, params) if errors.Is(err, pgx.ErrNoRows) { diff --git a/db/queries/queries.sql b/db/queries/queries.sql index da3fa9b..7c9a402 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -260,6 +260,56 @@ WHERE observer_id = $1 GROUP BY bucket ORDER BY bucket ASC; +-- name: GetObserverActivityRaw :many +-- 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. +SELECT + date_bin($3::interval, heard_at, TIMESTAMPTZ 'epoch')::timestamptz AS bucket, + COUNT(*)::bigint AS observations, + COALESCE(SUM(airtime_ms), 0)::real AS airtime_ms, + COUNT(airtime_ms)::bigint AS airtime_n, + COALESCE(AVG(snr) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0)), 0)::real AS snr_avg, + COALESCE(MIN(snr) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0)), 0)::real AS snr_min, + COUNT(snr) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::bigint AS snr_n, + 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 +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 +GROUP BY payload_type +ORDER BY count DESC; + +-- name: GetObserverActivityHourly :many +-- Hour-or-coarser buckets summed from the hourly rollup; same COALESCE-plus-count shape as the raw query. +SELECT + date_bin($3::interval, bucket, TIMESTAMPTZ 'epoch')::timestamptz AS bucket, + SUM(observations)::bigint AS observations, + COALESCE(SUM(airtime_ms), 0)::real AS airtime_ms, + SUM(airtime_n)::bigint AS airtime_n, + COALESCE(SUM(snr_sum), 0)::real AS snr_sum, + SUM(snr_n)::bigint AS snr_n, + COALESCE(MIN(snr_min), 0)::real AS snr_min, + 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 +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 +GROUP BY payload_type +ORDER BY count DESC; + -- name: ListObserverAdverts :many -- Returns advert packets (payload_type=4) heard by a specific observer. -- Pass cursor=0 to start from the beginning, or the last seen id for pagination. @@ -626,9 +676,10 @@ INSERT INTO packet_observations ( coding_rate, source_broker, payload_type, - resolved_endpoints + resolved_endpoints, + airtime_ms ) VALUES ( - $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18 + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19 ) ON CONFLICT (packet_hash, observer_id) DO NOTHING RETURNING *; @@ -1330,6 +1381,9 @@ REFRESH MATERIALIZED VIEW CONCURRENTLY mv_top_advertisers_by_iata; -- name: RefreshRadioPresets :exec REFRESH MATERIALIZED VIEW CONCURRENTLY mv_radio_presets; +-- name: RefreshObserverActivity :exec +REFRESH MATERIALIZED VIEW CONCURRENTLY mv_observer_activity_hourly; + -- name: ReconfirmRoutes :exec -- Checks the $1 least-recently-reconfirmed routes: deletes those with a departed -- hop node or a hop prefix now matching >1 node in that IATA (length-aware: diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index a1a5e7a..e43434b 100644 --- a/db/sqlc/mock/querier.go +++ b/db/sqlc/mock/querier.go @@ -292,6 +292,66 @@ func (mr *MockQuerierMockRecorder) GetNodesByIDs(ctx, dollar_1 any) *gomock.Call return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetNodesByIDs", reflect.TypeOf((*MockQuerier)(nil).GetNodesByIDs), ctx, dollar_1) } +// GetObserverActivityHourly mocks base method. +func (m *MockQuerier) GetObserverActivityHourly(ctx context.Context, arg db.GetObserverActivityHourlyParams) ([]db.GetObserverActivityHourlyRow, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetObserverActivityHourly", ctx, arg) + ret0, _ := ret[0].([]db.GetObserverActivityHourlyRow) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// GetObserverActivityHourly indicates an expected call of GetObserverActivityHourly. +func (mr *MockQuerierMockRecorder) GetObserverActivityHourly(ctx, arg any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetObserverActivityHourly", reflect.TypeOf((*MockQuerier)(nil).GetObserverActivityHourly), ctx, arg) +} + +// GetObserverActivityHourlyPayloadTypes mocks base method. +func (m *MockQuerier) GetObserverActivityHourlyPayloadTypes(ctx context.Context, arg db.GetObserverActivityHourlyPayloadTypesParams) ([]db.GetObserverActivityHourlyPayloadTypesRow, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetObserverActivityHourlyPayloadTypes", ctx, arg) + ret0, _ := ret[0].([]db.GetObserverActivityHourlyPayloadTypesRow) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// GetObserverActivityHourlyPayloadTypes indicates an expected call of GetObserverActivityHourlyPayloadTypes. +func (mr *MockQuerierMockRecorder) GetObserverActivityHourlyPayloadTypes(ctx, arg any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetObserverActivityHourlyPayloadTypes", reflect.TypeOf((*MockQuerier)(nil).GetObserverActivityHourlyPayloadTypes), ctx, arg) +} + +// GetObserverActivityRaw mocks base method. +func (m *MockQuerier) GetObserverActivityRaw(ctx context.Context, arg db.GetObserverActivityRawParams) ([]db.GetObserverActivityRawRow, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetObserverActivityRaw", ctx, arg) + ret0, _ := ret[0].([]db.GetObserverActivityRawRow) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// GetObserverActivityRaw indicates an expected call of GetObserverActivityRaw. +func (mr *MockQuerierMockRecorder) GetObserverActivityRaw(ctx, arg any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetObserverActivityRaw", reflect.TypeOf((*MockQuerier)(nil).GetObserverActivityRaw), ctx, arg) +} + +// GetObserverActivityRawPayloadTypes mocks base method. +func (m *MockQuerier) GetObserverActivityRawPayloadTypes(ctx context.Context, arg db.GetObserverActivityRawPayloadTypesParams) ([]db.GetObserverActivityRawPayloadTypesRow, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetObserverActivityRawPayloadTypes", ctx, arg) + ret0, _ := ret[0].([]db.GetObserverActivityRawPayloadTypesRow) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// GetObserverActivityRawPayloadTypes indicates an expected call of GetObserverActivityRawPayloadTypes. +func (mr *MockQuerierMockRecorder) GetObserverActivityRawPayloadTypes(ctx, arg any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetObserverActivityRawPayloadTypes", reflect.TypeOf((*MockQuerier)(nil).GetObserverActivityRawPayloadTypes), ctx, arg) +} + // GetObserverBrokers mocks base method. func (m *MockQuerier) GetObserverBrokers(ctx context.Context, observerID uuid.UUID) ([]db.GetObserverBrokersRow, error) { m.ctrl.T.Helper() @@ -1098,6 +1158,20 @@ func (mr *MockQuerierMockRecorder) RefreshHourlyStats(ctx any) *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RefreshHourlyStats", reflect.TypeOf((*MockQuerier)(nil).RefreshHourlyStats), ctx) } +// RefreshObserverActivity mocks base method. +func (m *MockQuerier) RefreshObserverActivity(ctx context.Context) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "RefreshObserverActivity", ctx) + ret0, _ := ret[0].(error) + return ret0 +} + +// RefreshObserverActivity indicates an expected call of RefreshObserverActivity. +func (mr *MockQuerierMockRecorder) RefreshObserverActivity(ctx any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RefreshObserverActivity", reflect.TypeOf((*MockQuerier)(nil).RefreshObserverActivity), ctx) +} + // RefreshPayloadBreakdown mocks base method. func (m *MockQuerier) RefreshPayloadBreakdown(ctx context.Context) error { m.ctrl.T.Helper() diff --git a/db/sqlc/models.go b/db/sqlc/models.go index 5bed269..9f6ef21 100644 --- a/db/sqlc/models.go +++ b/db/sqlc/models.go @@ -77,6 +77,20 @@ type MvHourlyIataStat struct { ActiveObservers int64 `json:"active_observers"` } +type MvObserverActivityHourly struct { + ObserverID uuid.UUID `json:"observer_id"` + PayloadType *int16 `json:"payload_type"` + Bucket pgtype.Timestamptz `json:"bucket"` + Observations int64 `json:"observations"` + AirtimeMs float32 `json:"airtime_ms"` + AirtimeN int64 `json:"airtime_n"` + SnrSum float32 `json:"snr_sum"` + SnrN int64 `json:"snr_n"` + SnrMin float32 `json:"snr_min"` + RssiSum int64 `json:"rssi_sum"` + RssiN int64 `json:"rssi_n"` +} + type MvPayloadBreakdownByIatum struct { Iata string `json:"iata"` PayloadType *int16 `json:"payload_type"` @@ -293,6 +307,7 @@ type PacketObservation struct { SourceBroker *string `json:"source_broker"` PayloadType *int16 `json:"payload_type"` ResolvedEndpoints []byte `json:"resolved_endpoints"` + AirtimeMs *float32 `json:"airtime_ms"` } type Region struct { diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index 4d1bdd1..7a64432 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -50,6 +50,14 @@ type Querier interface { // Returns the neighbors of a node with details, ordered by most recently seen. GetNodeNeighbors(ctx context.Context, nodeID uuid.UUID) ([]GetNodeNeighborsRow, error) GetNodesByIDs(ctx context.Context, dollar_1 []uuid.UUID) ([]GetNodesByIDsRow, error) + // 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) + // 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. + GetObserverActivityRaw(ctx context.Context, arg GetObserverActivityRawParams) ([]GetObserverActivityRawRow, error) + GetObserverActivityRawPayloadTypes(ctx context.Context, arg GetObserverActivityRawPayloadTypesParams) ([]GetObserverActivityRawPayloadTypesRow, error) GetObserverBrokers(ctx context.Context, observerID uuid.UUID) ([]GetObserverBrokersRow, error) GetObserverByID(ctx context.Context, id uuid.UUID) (Observer, error) GetObserverByPubkey(ctx context.Context, publicKey []byte) (Observer, error) @@ -192,6 +200,7 @@ type Querier interface { // 1/2/3/4-byte hop prefixes check prefix_1/2/3/4), and stamps the survivors. ReconfirmRoutes(ctx context.Context, limit int32) error RefreshHourlyStats(ctx context.Context) error + RefreshObserverActivity(ctx context.Context) error RefreshPayloadBreakdown(ctx context.Context) error RefreshRadioPresets(ctx context.Context) error RefreshTopAdvertisers(ctx context.Context) error diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 0dce6ba..45641d4 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -515,6 +515,216 @@ func (q *Queries) GetNodesByIDs(ctx context.Context, dollar_1 []uuid.UUID) ([]Ge return items, nil } +const getObserverActivityHourly = `-- name: GetObserverActivityHourly :many +SELECT + date_bin($3::interval, bucket, TIMESTAMPTZ 'epoch')::timestamptz AS bucket, + SUM(observations)::bigint AS observations, + COALESCE(SUM(airtime_ms), 0)::real AS airtime_ms, + SUM(airtime_n)::bigint AS airtime_n, + COALESCE(SUM(snr_sum), 0)::real AS snr_sum, + SUM(snr_n)::bigint AS snr_n, + COALESCE(MIN(snr_min), 0)::real AS snr_min, + 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 +GROUP BY 1 +ORDER BY 1 +` + +type GetObserverActivityHourlyParams struct { + ObserverID uuid.UUID `json:"observer_id"` + Column2 pgtype.Timestamptz `json:"column_2"` + Column3 pgtype.Interval `json:"column_3"` +} + +type GetObserverActivityHourlyRow struct { + Bucket pgtype.Timestamptz `json:"bucket"` + Observations int64 `json:"observations"` + AirtimeMs float32 `json:"airtime_ms"` + AirtimeN int64 `json:"airtime_n"` + SnrSum float32 `json:"snr_sum"` + SnrN int64 `json:"snr_n"` + SnrMin float32 `json:"snr_min"` + RssiSum int64 `json:"rssi_sum"` + RssiN int64 `json:"rssi_n"` +} + +// 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) + if err != nil { + return nil, err + } + defer rows.Close() + items := []GetObserverActivityHourlyRow{} + for rows.Next() { + var i GetObserverActivityHourlyRow + if err := rows.Scan( + &i.Bucket, + &i.Observations, + &i.AirtimeMs, + &i.AirtimeN, + &i.SnrSum, + &i.SnrN, + &i.SnrMin, + &i.RssiSum, + &i.RssiN, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +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 +GROUP BY payload_type +ORDER BY count DESC +` + +type GetObserverActivityHourlyPayloadTypesParams struct { + ObserverID uuid.UUID `json:"observer_id"` + Column2 pgtype.Timestamptz `json:"column_2"` +} + +type GetObserverActivityHourlyPayloadTypesRow struct { + 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) + if err != nil { + return nil, err + } + defer rows.Close() + items := []GetObserverActivityHourlyPayloadTypesRow{} + for rows.Next() { + var i GetObserverActivityHourlyPayloadTypesRow + if err := rows.Scan(&i.PayloadType, &i.Count); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const getObserverActivityRaw = `-- name: GetObserverActivityRaw :many +SELECT + date_bin($3::interval, heard_at, TIMESTAMPTZ 'epoch')::timestamptz AS bucket, + COUNT(*)::bigint AS observations, + COALESCE(SUM(airtime_ms), 0)::real AS airtime_ms, + COUNT(airtime_ms)::bigint AS airtime_n, + COALESCE(AVG(snr) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0)), 0)::real AS snr_avg, + COALESCE(MIN(snr) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0)), 0)::real AS snr_min, + COUNT(snr) FILTER (WHERE NOT (COALESCE(rssi, 0) = 0 AND COALESCE(snr, 0) = 0))::bigint AS snr_n, + 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 +GROUP BY bucket +ORDER BY bucket +` + +type GetObserverActivityRawParams struct { + ObserverID uuid.UUID `json:"observer_id"` + Column2 pgtype.Timestamptz `json:"column_2"` + Column3 pgtype.Interval `json:"column_3"` +} + +type GetObserverActivityRawRow struct { + Bucket pgtype.Timestamptz `json:"bucket"` + Observations int64 `json:"observations"` + AirtimeMs float32 `json:"airtime_ms"` + AirtimeN int64 `json:"airtime_n"` + SnrAvg float32 `json:"snr_avg"` + SnrMin float32 `json:"snr_min"` + SnrN int64 `json:"snr_n"` + RssiAvg float32 `json:"rssi_avg"` + RssiN int64 `json:"rssi_n"` +} + +// 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. +func (q *Queries) GetObserverActivityRaw(ctx context.Context, arg GetObserverActivityRawParams) ([]GetObserverActivityRawRow, error) { + rows, err := q.db.Query(ctx, getObserverActivityRaw, arg.ObserverID, arg.Column2, arg.Column3) + if err != nil { + return nil, err + } + defer rows.Close() + items := []GetObserverActivityRawRow{} + for rows.Next() { + var i GetObserverActivityRawRow + if err := rows.Scan( + &i.Bucket, + &i.Observations, + &i.AirtimeMs, + &i.AirtimeN, + &i.SnrAvg, + &i.SnrMin, + &i.SnrN, + &i.RssiAvg, + &i.RssiN, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +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 +GROUP BY payload_type +ORDER BY count DESC +` + +type GetObserverActivityRawPayloadTypesParams struct { + ObserverID uuid.UUID `json:"observer_id"` + Column2 pgtype.Timestamptz `json:"column_2"` +} + +type GetObserverActivityRawPayloadTypesRow struct { + PayloadType *int16 `json:"payload_type"` + Count int64 `json:"count"` +} + +func (q *Queries) GetObserverActivityRawPayloadTypes(ctx context.Context, arg GetObserverActivityRawPayloadTypesParams) ([]GetObserverActivityRawPayloadTypesRow, error) { + rows, err := q.db.Query(ctx, getObserverActivityRawPayloadTypes, arg.ObserverID, arg.Column2) + if err != nil { + return nil, err + } + defer rows.Close() + items := []GetObserverActivityRawPayloadTypesRow{} + for rows.Next() { + var i GetObserverActivityRawPayloadTypesRow + if err := rows.Scan(&i.PayloadType, &i.Count); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const getObserverBrokers = `-- name: GetObserverBrokers :many SELECT broker_name, last_seen, last_packet_at FROM observer_brokers @@ -1722,12 +1932,13 @@ INSERT INTO packet_observations ( coding_rate, source_broker, payload_type, - resolved_endpoints + resolved_endpoints, + airtime_ms ) VALUES ( - $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18 + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19 ) ON CONFLICT (packet_hash, observer_id) DO NOTHING -RETURNING id, packet_hash, observer_id, iata, heard_at, path_length_byte, hash_size, hop_count, path_bytes, rssi, snr, propagation_time_ms, radio_freq_mhz, spread_factor, bandwidth_khz, coding_rate, source_broker, payload_type, resolved_endpoints +RETURNING id, packet_hash, observer_id, iata, heard_at, path_length_byte, hash_size, hop_count, path_bytes, rssi, snr, propagation_time_ms, radio_freq_mhz, spread_factor, bandwidth_khz, coding_rate, source_broker, payload_type, resolved_endpoints, airtime_ms ` type InsertObservationParams struct { @@ -1749,6 +1960,7 @@ type InsertObservationParams struct { SourceBroker *string `json:"source_broker"` PayloadType *int16 `json:"payload_type"` ResolvedEndpoints []byte `json:"resolved_endpoints"` + AirtimeMs *float32 `json:"airtime_ms"` } // ============================================================ @@ -1774,6 +1986,7 @@ func (q *Queries) InsertObservation(ctx context.Context, arg InsertObservationPa arg.SourceBroker, arg.PayloadType, arg.ResolvedEndpoints, + arg.AirtimeMs, ) var i PacketObservation err := row.Scan( @@ -1796,6 +2009,7 @@ func (q *Queries) InsertObservation(ctx context.Context, arg InsertObservationPa &i.SourceBroker, &i.PayloadType, &i.ResolvedEndpoints, + &i.AirtimeMs, ) return i, err } @@ -2562,7 +2776,7 @@ func (q *Queries) ListNodes(ctx context.Context, arg ListNodesParams) ([]ListNod } const listObservationsForPacket = `-- name: ListObservationsForPacket :many -SELECT po.id, po.packet_hash, po.observer_id, po.iata, po.heard_at, po.path_length_byte, po.hash_size, po.hop_count, po.path_bytes, po.rssi, po.snr, po.propagation_time_ms, po.radio_freq_mhz, po.spread_factor, po.bandwidth_khz, po.coding_rate, po.source_broker, po.payload_type, po.resolved_endpoints, o.display_name AS observer_name +SELECT po.id, po.packet_hash, po.observer_id, po.iata, po.heard_at, po.path_length_byte, po.hash_size, po.hop_count, po.path_bytes, po.rssi, po.snr, po.propagation_time_ms, po.radio_freq_mhz, po.spread_factor, po.bandwidth_khz, po.coding_rate, po.source_broker, po.payload_type, po.resolved_endpoints, po.airtime_ms, o.display_name AS observer_name FROM packet_observations po LEFT JOIN observers o ON o.id = po.observer_id WHERE po.packet_hash = $1 @@ -2589,6 +2803,7 @@ type ListObservationsForPacketRow struct { SourceBroker *string `json:"source_broker"` PayloadType *int16 `json:"payload_type"` ResolvedEndpoints []byte `json:"resolved_endpoints"` + AirtimeMs *float32 `json:"airtime_ms"` ObserverName *string `json:"observer_name"` } @@ -2621,6 +2836,7 @@ func (q *Queries) ListObservationsForPacket(ctx context.Context, packetHash []by &i.SourceBroker, &i.PayloadType, &i.ResolvedEndpoints, + &i.AirtimeMs, &i.ObserverName, ); err != nil { return nil, err @@ -3482,6 +3698,15 @@ func (q *Queries) RefreshHourlyStats(ctx context.Context) error { return err } +const refreshObserverActivity = `-- name: RefreshObserverActivity :exec +REFRESH MATERIALIZED VIEW CONCURRENTLY mv_observer_activity_hourly +` + +func (q *Queries) RefreshObserverActivity(ctx context.Context) error { + _, err := q.db.Exec(ctx, refreshObserverActivity) + return err +} + const refreshPayloadBreakdown = `-- name: RefreshPayloadBreakdown :exec REFRESH MATERIALIZED VIEW CONCURRENTLY mv_payload_breakdown_by_iata ` diff --git a/db/stats.go b/db/stats.go index ffbcabb..b8dc915 100644 --- a/db/stats.go +++ b/db/stats.go @@ -303,3 +303,7 @@ func (s *Store) RefreshTopObservers(ctx context.Context) error { func (s *Store) RefreshRadioPresets(ctx context.Context) error { return s.q.RefreshRadioPresets(ctx) } + +func (s *Store) RefreshObserverActivity(ctx context.Context) error { + return s.q.RefreshObserverActivity(ctx) +} diff --git a/docs/docs.go b/docs/docs.go index 3a97dd9..9d56007 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -905,6 +905,64 @@ const docTemplate = `{ } } }, + "/observers/{observerId}/activity": { + "get": { + "produces": [ + "application/json" + ], + "tags": [ + "Observers" + ], + "summary": "Get observer heard-activity history", + "parameters": [ + { + "type": "string", + "description": "Observer UUID", + "name": "observerId", + "in": "path", + "required": true + }, + { + "type": "string", + "description": "Trailing window as a Go duration, max 720h (default 24h); max 48h when interval is under 1h", + "name": "range", + "in": "query" + }, + { + "type": "string", + "description": "Bucket size: 5m, 15m, 1h, 6h or 24h (default 15m)", + "name": "interval", + "in": "query" + } + ], + "responses": { + "200": { + "description": "OK", + "schema": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivity" + } + }, + "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" + } + } + } + } + }, "/observers/{observerId}/adverts": { "get": { "produces": [ @@ -3004,6 +3062,76 @@ const docTemplate = `{ } } }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivity": { + "type": "object", + "properties": { + "interval": { + "type": "string" + }, + "payloadTypes": { + "type": "array", + "items": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.PayloadBreakdownItem" + } + }, + "points": { + "type": "array", + "items": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivityPoint" + } + }, + "radio": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivityRadio" + }, + "range": { + "type": "string" + } + } + }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivityPoint": { + "type": "object", + "properties": { + "airtimeMs": { + "type": "number" + }, + "observations": { + "type": "integer" + }, + "rssiAvg": { + "type": "number" + }, + "snrAvg": { + "type": "number" + }, + "snrMin": { + "type": "number" + }, + "t": { + "description": "bucket start, epoch ms", + "type": "integer" + } + } + }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivityRadio": { + "type": "object", + "properties": { + "bwKhz": { + "type": "number" + }, + "cr": { + "type": "integer" + }, + "freqMhz": { + "type": "number" + }, + "preambleSymbols": { + "type": "integer" + }, + "sf": { + "type": "integer" + } + } + }, "github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverBroker": { "type": "object", "properties": { diff --git a/docs/swagger.json b/docs/swagger.json index c9f19cf..ede5784 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -903,6 +903,64 @@ } } }, + "/observers/{observerId}/activity": { + "get": { + "produces": [ + "application/json" + ], + "tags": [ + "Observers" + ], + "summary": "Get observer heard-activity history", + "parameters": [ + { + "type": "string", + "description": "Observer UUID", + "name": "observerId", + "in": "path", + "required": true + }, + { + "type": "string", + "description": "Trailing window as a Go duration, max 720h (default 24h); max 48h when interval is under 1h", + "name": "range", + "in": "query" + }, + { + "type": "string", + "description": "Bucket size: 5m, 15m, 1h, 6h or 24h (default 15m)", + "name": "interval", + "in": "query" + } + ], + "responses": { + "200": { + "description": "OK", + "schema": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivity" + } + }, + "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" + } + } + } + } + }, "/observers/{observerId}/adverts": { "get": { "produces": [ @@ -3002,6 +3060,76 @@ } } }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivity": { + "type": "object", + "properties": { + "interval": { + "type": "string" + }, + "payloadTypes": { + "type": "array", + "items": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.PayloadBreakdownItem" + } + }, + "points": { + "type": "array", + "items": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivityPoint" + } + }, + "radio": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivityRadio" + }, + "range": { + "type": "string" + } + } + }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivityPoint": { + "type": "object", + "properties": { + "airtimeMs": { + "type": "number" + }, + "observations": { + "type": "integer" + }, + "rssiAvg": { + "type": "number" + }, + "snrAvg": { + "type": "number" + }, + "snrMin": { + "type": "number" + }, + "t": { + "description": "bucket start, epoch ms", + "type": "integer" + } + } + }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivityRadio": { + "type": "object", + "properties": { + "bwKhz": { + "type": "number" + }, + "cr": { + "type": "integer" + }, + "freqMhz": { + "type": "number" + }, + "preambleSymbols": { + "type": "integer" + }, + "sf": { + "type": "integer" + } + } + }, "github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverBroker": { "type": "object", "properties": { diff --git a/docs/swagger.yaml b/docs/swagger.yaml index 147830b..7efd8ad 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -480,6 +480,52 @@ definitions: uptimeSeconds: type: integer type: object + github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivity: + properties: + interval: + type: string + payloadTypes: + items: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.PayloadBreakdownItem' + type: array + points: + items: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivityPoint' + type: array + radio: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivityRadio' + range: + type: string + type: object + github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivityPoint: + properties: + airtimeMs: + type: number + observations: + type: integer + rssiAvg: + type: number + snrAvg: + type: number + snrMin: + type: number + t: + description: bucket start, epoch ms + type: integer + type: object + github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivityRadio: + properties: + bwKhz: + type: number + cr: + type: integer + freqMhz: + type: number + preambleSymbols: + type: integer + sf: + type: integer + type: object github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverBroker: properties: lastPacketAt: @@ -1739,6 +1785,45 @@ paths: summary: Get observer detail tags: - Observers + /observers/{observerId}/activity: + get: + parameters: + - description: Observer UUID + in: path + name: observerId + required: true + type: string + - description: Trailing window as a Go duration, max 720h (default 24h); max + 48h when interval is under 1h + in: query + name: range + type: string + - description: 'Bucket size: 5m, 15m, 1h, 6h or 24h (default 15m)' + in: query + name: interval + type: string + produces: + - application/json + responses: + "200": + description: OK + schema: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverActivity' + "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 observer heard-activity history + tags: + - Observers /observers/{observerId}/adverts: get: parameters: diff --git a/internal/api/handlers/observers.go b/internal/api/handlers/observers.go index d559cbd..01491b5 100644 --- a/internal/api/handlers/observers.go +++ b/internal/api/handlers/observers.go @@ -4,6 +4,7 @@ package handlers import ( + "errors" "net/http" "strconv" "time" @@ -11,6 +12,7 @@ import ( "github.com/MeshCore-Beacon/beacon-server/internal/api" "github.com/go-chi/chi/v5" "github.com/google/uuid" + "github.com/jackc/pgx/v5" ) // ObserversRouter mounts all /observers routes onto a subrouter. @@ -18,6 +20,7 @@ import ( // GET /observers → listObservers // GET /observers/{observerId} → getObserver // GET /observers/{observerId}/telemetry → getObserverTelemetry +// GET /observers/{observerId}/activity → getObserverActivity // GET /observers/{observerId}/adverts → listObserverAdverts func ObserversRouter(reader api.Reader) http.Handler { r := chi.NewRouter() @@ -26,6 +29,7 @@ func ObserversRouter(reader api.Reader) http.Handler { r.Get("/", getObserver(reader)) r.Get("/adverts", listObserverAdverts(reader)) r.Get("/telemetry", getObserverTelemetry(reader)) + r.Get("/activity", getObserverActivity(reader)) }) return r } @@ -232,3 +236,75 @@ func getObserverTelemetry(reader api.Reader) http.HandlerFunc { respond(w, http.StatusOK, telemetry) } } + +// activityIntervals is the fixed bucket set; every value divides 24h so bucket +// starts stay aligned to the clock regardless of when the window began. +var activityIntervals = map[string]time.Duration{ + "5m": 5 * time.Minute, + "15m": 15 * time.Minute, + "1h": time.Hour, + "6h": 6 * time.Hour, + "24h": 24 * time.Hour, +} + +// getObserverActivity godoc +// +// @Summary Get observer heard-activity history +// @Tags Observers +// @Produce json +// @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)" +// @Success 200 {object} api.ObserverActivity +// @Failure 400 {object} handlers.APIError +// @Failure 404 {object} handlers.APIError +// @Failure 500 {object} handlers.APIError +// @Router /observers/{observerId}/activity [get] +func getObserverActivity(reader api.Reader) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + observerID, err := uuid.Parse(chi.URLParam(r, "observerId")) + if err != nil { + respondError(w, http.StatusBadRequest, "invalid observer ID") + return + } + rangeParam := r.URL.Query().Get("range") + if rangeParam == "" { + rangeParam = "24h" + } + window, err := time.ParseDuration(rangeParam) + if err != nil || window <= 0 || window > 720*time.Hour { + respondError(w, http.StatusBadRequest, "invalid range, use a duration up to 720h e.g. 24h, 168h, 720h") + return + } + intervalParam := r.URL.Query().Get("interval") + if intervalParam == "" { + intervalParam = "15m" + } + interval, ok := activityIntervals[intervalParam] + if !ok { + respondError(w, http.StatusBadRequest, "invalid interval, use 5m, 15m, 1h, 6h or 24h") + return + } + // the raw path scans live observations, so keep sub-hour windows short + if interval < time.Hour && window > 48*time.Hour { + respondError(w, http.StatusBadRequest, "range must be 48h or less for intervals under 1h") + return + } + if window/interval > 1000 { + respondError(w, http.StatusBadRequest, "range/interval exceeds 1000 buckets") + return + } + activity, err := reader.GetObserverActivity(r.Context(), observerID, window, interval) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + respondError(w, http.StatusNotFound, "observer not found") + return + } + respondError(w, http.StatusInternalServerError, "internal server error") + return + } + activity.Range = rangeParam + activity.Interval = intervalParam + respond(w, http.StatusOK, activity) + } +} diff --git a/internal/api/handlers/observers_test.go b/internal/api/handlers/observers_test.go index 148a06f..0d4a028 100644 --- a/internal/api/handlers/observers_test.go +++ b/internal/api/handlers/observers_test.go @@ -5,6 +5,8 @@ package handlers import ( "context" + "encoding/json" + "errors" "net/http" "net/http/httptest" "testing" @@ -13,6 +15,7 @@ import ( "github.com/MeshCore-Beacon/beacon-server/internal/api" "github.com/go-chi/chi/v5" "github.com/google/uuid" + "github.com/jackc/pgx/v5" ) func TestGetObserverTelemetry_InvalidUUID(t *testing.T) { @@ -168,3 +171,228 @@ func TestListObserverAdverts_InvalidUUID(t *testing.T) { t.Errorf("expected 400, got %d", w.Code) } } + +func TestGetObserverActivity_Defaults(t *testing.T) { + observerID := uuid.MustParse("00000000-0000-0000-0000-000000000001") + 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) { + gotWindow, gotInterval = window, interval + return &api.ObserverActivity{Points: []api.ObserverActivityPoint{}}, nil + }, + })) + req := httptest.NewRequest(http.MethodGet, "/observers/"+observerID.String()+"/activity", nil) + w := httptest.NewRecorder() + r.ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", w.Code) + } + if gotWindow != 24*time.Hour { + t.Errorf("expected window 24h, got %s", gotWindow) + } + if gotInterval != 15*time.Minute { + t.Errorf("expected interval 15m, got %s", gotInterval) + } + var body api.ObserverActivity + if err := json.Unmarshal(w.Body.Bytes(), &body); err != nil { + t.Fatalf("decode body: %v", err) + } + if body.Range != "24h" || body.Interval != "15m" { + t.Errorf("expected range 24h interval 15m, got %q/%q", body.Range, body.Interval) + } +} + +func TestGetObserverActivity_CustomRangeAndInterval(t *testing.T) { + observerID := uuid.MustParse("00000000-0000-0000-0000-000000000001") + 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) { + gotWindow, gotInterval = window, interval + return &api.ObserverActivity{Points: []api.ObserverActivityPoint{}}, nil + }, + })) + req := httptest.NewRequest(http.MethodGet, "/observers/"+observerID.String()+"/activity?range=168h&interval=1h", nil) + w := httptest.NewRecorder() + r.ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", w.Code) + } + if gotWindow != 168*time.Hour { + t.Errorf("expected window 168h, got %s", gotWindow) + } + if gotInterval != time.Hour { + t.Errorf("expected interval 1h, got %s", gotInterval) + } + var body api.ObserverActivity + if err := json.Unmarshal(w.Body.Bytes(), &body); err != nil { + t.Fatalf("decode body: %v", err) + } + if body.Range != "168h" || body.Interval != "1h" { + t.Errorf("expected range 168h interval 1h, got %q/%q", body.Range, body.Interval) + } +} + +func TestGetObserverActivity_SubHourRangeLimit(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) { + t.Fatal("reader should not be called") + return nil, nil + }, + })) + req := httptest.NewRequest(http.MethodGet, "/observers/"+observerID.String()+"/activity?range=168h&interval=15m", nil) + w := httptest.NewRecorder() + r.ServeHTTP(w, req) + if w.Code != http.StatusBadRequest { + t.Fatalf("expected 400, got %d (%s)", w.Code, w.Body.String()) + } + var errBody map[string]APIError + if err := json.Unmarshal(w.Body.Bytes(), &errBody); err != nil { + t.Fatalf("decode body: %v", err) + } + if errBody["error"].Message != "range must be 48h or less for intervals under 1h" { + t.Errorf("unexpected error message %q", errBody["error"].Message) + } + + 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) { + gotWindow, gotInterval = window, interval + return &api.ObserverActivity{Points: []api.ObserverActivityPoint{}}, nil + }, + })) + req = httptest.NewRequest(http.MethodGet, "/observers/"+observerID.String()+"/activity?range=48h&interval=15m", nil) + w = httptest.NewRecorder() + ok.ServeHTTP(w, req) + if w.Code != http.StatusOK { + t.Fatalf("expected 200 at 48h/15m, got %d (%s)", w.Code, w.Body.String()) + } + if gotWindow != 48*time.Hour || gotInterval != 15*time.Minute { + t.Errorf("expected 48h/15m, got %s/%s", gotWindow, gotInterval) + } +} + +func TestGetObserverActivity_BadRequests(t *testing.T) { + tests := []struct { + name string + path string + query string + }{ + {"invalid uuid", "not-a-uuid", ""}, + {"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"}, + {"invalid interval", "00000000-0000-0000-0000-000000000001", "?interval=2h"}, + {"too many buckets", "00000000-0000-0000-0000-000000000001", "?range=720h&interval=5m"}, + } + for _, tt := range tests { + 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) { + t.Fatal("reader should not be called") + return nil, nil + }, + })) + req := httptest.NewRequest(http.MethodGet, "/observers/"+tt.path+"/activity"+tt.query, nil) + w := httptest.NewRecorder() + r.ServeHTTP(w, req) + if w.Code != http.StatusBadRequest { + t.Errorf("expected 400, got %d (%s)", w.Code, w.Body.String()) + } + }) + } +} + +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) { + return nil, pgx.ErrNoRows + }, + })) + req := httptest.NewRequest(http.MethodGet, "/observers/"+observerID.String()+"/activity", nil) + w := httptest.NewRecorder() + r.ServeHTTP(w, req) + + if w.Code != http.StatusNotFound { + t.Fatalf("expected 404, got %d", w.Code) + } + var body struct { + Error APIError `json:"error"` + } + if err := json.Unmarshal(w.Body.Bytes(), &body); err != nil { + t.Fatalf("decode body: %v", err) + } + if body.Error.Code != "not_found" { + t.Errorf("expected code not_found, got %q", body.Error.Code) + } +} + +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) { + return nil, errors.New("boom") + }, + })) + req := httptest.NewRequest(http.MethodGet, "/observers/"+observerID.String()+"/activity", nil) + w := httptest.NewRecorder() + r.ServeHTTP(w, req) + + if w.Code != http.StatusInternalServerError { + t.Errorf("expected 500, got %d", w.Code) + } +} + +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) { + return &api.ObserverActivity{ + PayloadTypes: []api.PayloadBreakdownItem{}, + Points: []api.ObserverActivityPoint{{T: 1757376000000, Observations: 3}}, + }, nil + }, + })) + req := httptest.NewRequest(http.MethodGet, "/observers/"+observerID.String()+"/activity", nil) + w := httptest.NewRecorder() + r.ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", w.Code) + } + var body map[string]any + if err := json.Unmarshal(w.Body.Bytes(), &body); err != nil { + t.Fatalf("decode body: %v", err) + } + radio, ok := body["radio"] + if !ok { + t.Fatal("expected radio key in body") + } + if radio != nil { + t.Errorf("expected radio null, got %v", radio) + } + points, ok := body["points"].([]any) + if !ok || len(points) != 1 { + t.Fatalf("expected 1 point, got %v", body["points"]) + } + point := points[0].(map[string]any) + airtime, ok := point["airtimeMs"] + if !ok { + t.Fatal("expected airtimeMs key in point") + } + if airtime != nil { + t.Errorf("expected airtimeMs null, got %v", airtime) + } +} diff --git a/internal/api/handlers/stub_reader_test.go b/internal/api/handlers/stub_reader_test.go index 2b97ed2..4cd3d87 100644 --- a/internal/api/handlers/stub_reader_test.go +++ b/internal/api/handlers/stub_reader_test.go @@ -31,6 +31,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) 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) @@ -169,6 +170,13 @@ 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) { + if s.getObserverActivity != nil { + return s.getObserverActivity(ctx, observerID, window, interval) + } + return nil, nil +} + func (s stubReader) GetObserverScopes(ctx context.Context, observerID uuid.UUID) ([]string, error) { if s.getObserverScopes != nil { return s.getObserverScopes(ctx, observerID) diff --git a/internal/api/observers.go b/internal/api/observers.go index 94d1078..afe800f 100644 --- a/internal/api/observers.go +++ b/internal/api/observers.go @@ -67,3 +67,31 @@ type ObserverTelemetry struct { Interval string `json:"interval"` Points []ObserverTelemetryPoint `json:"points"` } + +// ObserverActivityRadio echoes the observer's current radio params plus the preamble the airtime math assumes. +type ObserverActivityRadio struct { + FreqMHz *float32 `json:"freqMhz"` + SF int16 `json:"sf"` + BWKHz float32 `json:"bwKhz"` + CR int16 `json:"cr"` + PreambleSymbols int `json:"preambleSymbols"` +} + +// ObserverActivityPoint is one bucket of what an observer heard. +type ObserverActivityPoint struct { + T int64 `json:"t"` // bucket start, epoch ms + Observations int64 `json:"observations"` + AirtimeMs *float32 `json:"airtimeMs"` + SNRAvg *float32 `json:"snrAvg"` + SNRMin *float32 `json:"snrMin"` + RSSIAvg *float32 `json:"rssiAvg"` +} + +// ObserverActivity is the per-observer heard-activity response. +type ObserverActivity struct { + Range string `json:"range"` + Interval string `json:"interval"` + Radio *ObserverActivityRadio `json:"radio"` + PayloadTypes []PayloadBreakdownItem `json:"payloadTypes"` + Points []ObserverActivityPoint `json:"points"` +} diff --git a/internal/api/reader.go b/internal/api/reader.go index b8b353a..dc3e1dc 100644 --- a/internal/api/reader.go +++ b/internal/api/reader.go @@ -97,6 +97,11 @@ type Reader interface { // GetObserverTelemetryBucketed returns telemetry points for an observer bucketed into N-hour intervals. GetObserverTelemetryBucketed(ctx context.Context, observerID uuid.UUID, since, until time.Time, bucketHours int32) ([]ObserverTelemetryPoint, error) + // 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) + // GetObserverScopes returns the names of all transport scopes an observer has // been seen forwarding packets for, ordered alphabetically. GetObserverScopes(ctx context.Context, observerID uuid.UUID) ([]string, error) diff --git a/internal/background/tasks.go b/internal/background/tasks.go index a19b6e4..b29760a 100644 --- a/internal/background/tasks.go +++ b/internal/background/tasks.go @@ -21,6 +21,7 @@ type viewRefresher interface { RefreshTopTalkers(context.Context) error RefreshTopAdvertisers(context.Context) error RefreshRadioPresets(context.Context) error + RefreshObserverActivity(context.Context) error } // ViewRefreshTask returns a Task that refreshes all materialized views. @@ -41,6 +42,7 @@ func ViewRefreshTask(store viewRefresher, interval time.Duration) Task { {"top talkers", store.RefreshTopTalkers}, {"top advertisers", store.RefreshTopAdvertisers}, {"radio presets", store.RefreshRadioPresets}, + {"observer activity", store.RefreshObserverActivity}, } { if err := view.refresh(ctx); err != nil { errs = append(errs, fmt.Errorf("%s: %w", view.name, err)) diff --git a/internal/background/tasks_test.go b/internal/background/tasks_test.go index b84973a..ca9fb99 100644 --- a/internal/background/tasks_test.go +++ b/internal/background/tasks_test.go @@ -111,6 +111,9 @@ func (s *refreshStub) RefreshTopAdvertisers(ctx context.Context) error { func (s *refreshStub) RefreshRadioPresets(ctx context.Context) error { return s.refresh(ctx, "radio presets") } +func (s *refreshStub) RefreshObserverActivity(ctx context.Context) error { + return s.refresh(ctx, "observer activity") +} func TestViewRefreshTask(t *testing.T) { first, second := errors.New("first failure"), errors.New("second failure") @@ -125,8 +128,8 @@ func TestViewRefreshTask(t *testing.T) { store := &refreshStub{errs: tc.errs} task := ViewRefreshTask(store, time.Minute) err := task.Run(context.Background()) - if len(store.calls) != 7 { - t.Fatalf("refreshed %d views, want 7", len(store.calls)) + if len(store.calls) != 8 { + t.Fatalf("refreshed %d views, want 8", len(store.calls)) } if len(tc.errs) == 0 && err != nil { t.Fatal(err) diff --git a/internal/cache/cache_test.go b/internal/cache/cache_test.go index eb1ad14..68c968b 100644 --- a/internal/cache/cache_test.go +++ b/internal/cache/cache_test.go @@ -121,6 +121,10 @@ func (s *stubReader) GetObserverTelemetryBucketed(_ context.Context, _ uuid.UUID return nil, nil } +func (s *stubReader) GetObserverActivity(_ context.Context, _ uuid.UUID, _, _ time.Duration) (*api.ObserverActivity, error) { + return nil, nil +} + func (s *stubReader) GetPacket(_ context.Context, _ []byte) (*api.Packet, error) { return nil, nil } func (s *stubReader) GetChannel(_ context.Context, _ int32) (*api.Channel, error) { return nil, nil } diff --git a/internal/cache/reader.go b/internal/cache/reader.go index 04dc5b7..4c69bd2 100644 --- a/internal/cache/reader.go +++ b/internal/cache/reader.go @@ -41,8 +41,12 @@ const ( keyNodesByIDsPrefix = "beacon:nodes:ids:" keyObserverPrefix = "beacon:observer:" keyObserverScopesPrefix = "beacon:observer:scopes:" + keyObserverActivityPrefix = "beacon:observer:activity:" ) +// observerActivityTTL is deliberately short: activity is a live chart, not reference data. +const observerActivityTTL = 60 * time.Second + // CachedReader wraps an api.Reader with a Redis caching layer. // It implements api.Reader and is a drop-in replacement for db.Store // at the wiring point in main.go. @@ -352,6 +356,14 @@ func (cr *CachedReader) GetObserverTelemetryBucketed(ctx context.Context, observ return cr.inner.GetObserverTelemetryBucketed(ctx, observerID, since, until, bucketHours) } +// 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() + return getOrSet(ctx, cr.c, key, observerActivityTTL, func() (*api.ObserverActivity, error) { + return cr.inner.GetObserverActivity(ctx, observerID, window, interval) + }) +} + // GetPacket implements [api.Reader]. func (cr *CachedReader) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet, error) { return cr.inner.GetPacket(ctx, packetHash) diff --git a/internal/ingest/frame_length_test.go b/internal/ingest/frame_length_test.go new file mode 100644 index 0000000..d69601d --- /dev/null +++ b/internal/ingest/frame_length_test.go @@ -0,0 +1,145 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package ingest + +import ( + "context" + "testing" + + "github.com/MeshCore-Beacon/beacon-server/internal/lora" + "github.com/google/uuid" + "github.com/meshcore-go/meshcore-go" +) + +// frameCaptureDB records what handlePacket stored so a test can rebuild the +// frame from the persisted columns and compare it to the bytes off the wire. +type frameCaptureDB struct { + *stubDB + radio RadioSettings + packets []UpsertPacketParams + observed []InsertObservationParams +} + +func (s *frameCaptureDB) UpsertPacket(_ context.Context, p UpsertPacketParams) (bool, error) { + s.packets = append(s.packets, p) + return true, nil +} + +func (s *frameCaptureDB) InsertObservation(_ context.Context, o InsertObservationParams) (bool, error) { + s.observed = append(s.observed, o) + return true, nil +} + +func (s *frameCaptureDB) GetObserverRadio(context.Context, uuid.UUID) (RadioSettings, error) { + return s.radio, nil +} + +// grpTxtPayload is a stand-in encrypted payload; its bytes never need decoding here. +func grpTxtPayload(t *testing.T) []byte { + t.Helper() + return buildGrpTxtPacket(t, 0x1a, make([]byte, 16)).Payload +} + +// pathBytes returns n distinct hashes of hashSize bytes each. +func pathBytes(hashCount, hashSize int) []byte { + path := make([]byte, 0, hashCount*hashSize) + for i := range hashCount { + for j := range hashSize { + path = append(path, byte(0x10+i*hashSize+j)) + } + } + return path +} + +// pathLengthByte packs hash size and hop count the way MeshCore does. +func pathLengthByte(hashCount, hashSize int) uint8 { + return uint8(hashSize-1)<<6 | uint8(hashCount) +} + +// TestFrameLengthReconstruction proves lora.FrameLength rebuilds the on-air byte +// count from the stored columns alone, across the path/transport-code shapes. +func TestFrameLengthReconstruction(t *testing.T) { + for _, tc := range []struct { + name string + packet *meshcore.Packet + wantTransportCodes bool + }{ + { + name: "flood with 1-byte hash path", + packet: &meshcore.Packet{ + Header: meshcore.MakeHeader(meshcore.RouteTypeFlood, meshcore.PayloadTypeGrpTxt, 0), + PathLength: pathLengthByte(2, 1), + Path: pathBytes(2, 1), + Payload: grpTxtPayload(t), + }, + }, + { + name: "transport flood with transport codes", + packet: &meshcore.Packet{ + Header: meshcore.MakeHeader(meshcore.RouteTypeTransportFlood, meshcore.PayloadTypeGrpTxt, 0), + PathLength: pathLengthByte(1, 1), + Path: pathBytes(1, 1), + Payload: grpTxtPayload(t), + TransportCode1: 0xbeef, + TransportCode2: 0xf00d, + }, + wantTransportCodes: true, + }, + { + name: "flood with 2-byte hash path", + packet: &meshcore.Packet{ + Header: meshcore.MakeHeader(meshcore.RouteTypeFlood, meshcore.PayloadTypeGrpTxt, 0), + PathLength: pathLengthByte(3, 2), + Path: pathBytes(3, 2), + Payload: grpTxtPayload(t), + }, + }, + } { + t.Run(tc.name, func(t *testing.T) { + raw, err := tc.packet.ToBytes() + if err != nil { + t.Fatalf("packet to bytes: %v", err) + } + w, base := newTestWorker() + db := &frameCaptureDB{stubDB: base, radio: RadioSettings{FreqMHz: 910.525, SF: 7, BWKHz: 62.5, CR: 5}} + w.db = db + + w.handlePacket(context.Background(), "YOW", "0102", packetEnvelope(t, tc.packet)) + + if len(db.packets) != 1 || len(db.observed) != 1 { + t.Fatalf("stored %d packets and %d observations, want 1 of each", len(db.packets), len(db.observed)) + } + up, obs := db.packets[0], db.observed[0] + // Pin the captured shape: without this a lost transport-code capture would + // quietly turn this case into the plain-flood one and still pass. + if gotCodes := len(up.TransportCodes) == 4; gotCodes != tc.wantTransportCodes { + t.Fatalf("captured %d transport-code bytes, want transport codes = %v", len(up.TransportCodes), tc.wantTransportCodes) + } + got := lora.FrameLength(len(up.TransportCodes) == 4, len(obs.PathBytes), len(up.RawPayload)) + if got != len(raw) { + t.Errorf("FrameLength = %d, want %d (wire bytes)", got, len(raw)) + } + if obs.AirtimeMs == nil { + t.Error("AirtimeMs is nil, want a cost for SF7/62.5/CR5") + } + }) + } +} + +// TestObservationAirtimeUnknownRadio covers the observer that never reported its +// radio: the columns are zero, so there is no airtime to store. +func TestObservationAirtimeUnknownRadio(t *testing.T) { + w, base := newTestWorker() + db := &frameCaptureDB{stubDB: base} // zero RadioSettings + w.db = db + + w.handlePacket(context.Background(), "YOW", "0102", packetEnvelope(t, buildGrpTxtPacket(t, 0x1a, make([]byte, 16)))) + + if len(db.observed) != 1 { + t.Fatalf("stored %d observations, want 1", len(db.observed)) + } + if got := db.observed[0].AirtimeMs; got != nil { + t.Errorf("AirtimeMs = %v, want nil when the radio settings are unknown", *got) + } +} diff --git a/internal/ingest/packet.go b/internal/ingest/packet.go index 4a16a04..be8d007 100644 --- a/internal/ingest/packet.go +++ b/internal/ingest/packet.go @@ -16,6 +16,7 @@ import ( "time" "github.com/MeshCore-Beacon/beacon-server/internal/api" + "github.com/MeshCore-Beacon/beacon-server/internal/lora" "github.com/google/uuid" "github.com/meshcore-go/meshcore-go" ) @@ -56,6 +57,7 @@ type InsertObservationParams struct { SourceBroker string PayloadType int16 ResolvedEndpoints json.RawMessage + AirtimeMs *float32 // nil when the observer never reported costable radio settings } // RadioSettings holds the radio configuration for an observer, populated from @@ -787,6 +789,13 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ resolvedEndpoints = nil // optional enrichment must not discard the observation } } + // Airtime is costed from the frame as received; zero radio columns mean the + // observer never reported its settings, so there is nothing to cost. + var airtimeMs *float32 + if ms, ok := lora.TimeOnAirMs(len(hexBytes), int(radio.SF), float64(radio.BWKHz), int(radio.CR)); ok { + cost := float32(ms) + airtimeMs = &cost + } oParams := InsertObservationParams{ PacketHash: packetHash[:], ObserverID: id, @@ -806,6 +815,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ SourceBroker: w.cfg.BrokerName, PayloadType: int16(packet.PayloadType()), ResolvedEndpoints: resolvedEndpoints, + AirtimeMs: airtimeMs, } inserted, err := w.db.InsertObservation(ctx, oParams) if err != nil { diff --git a/internal/lora/airtime.go b/internal/lora/airtime.go new file mode 100644 index 0000000..1bd8a15 --- /dev/null +++ b/internal/lora/airtime.go @@ -0,0 +1,56 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +// Package lora computes LoRa airtime for MeshCore frames, so an observation +// can be costed in milliseconds of channel occupancy. +package lora + +import "math" + +const ( + minSF = 7 + maxSF = 12 + minCR = 5 + maxCR = 8 +) + +// PreambleSymbols mirrors MeshCore's preambleLengthForSF: 32 symbols at SF<=8, 16 above. +func PreambleSymbols(sf int) int { + if sf <= 8 { + return 32 + } + return 16 +} + +// FrameLength is the on-air byte count of a MeshCore frame rebuilt from its stored parts. +func FrameLength(transportCodes bool, pathLen, payloadLen int) int { + n := 1 + 1 + pathLen + payloadLen // header byte + path length byte + if transportCodes { + n += 4 + } + return n +} + +// TimeOnAirMs is the Semtech LoRa time-on-air (RadioLib getTimeOnAir: CRC on, explicit header, +// LDRO auto at t_sym >= 16 ms). ok is false when any parameter is outside the costable range. +func TimeOnAirMs(frameLen, sf int, bwKHz float64, cr int) (ms float64, ok bool) { + if frameLen <= 0 || sf < minSF || sf > maxSF || bwKHz <= 0 || cr < minCR || cr > maxCR { + return 0, false + } + tSym := math.Exp2(float64(sf)) / bwKHz + de := 0 + if tSym >= 16 { + de = 1 + } + n := float64(PreambleSymbols(sf)) + 4.25 + float64(symbols(frameLen, sf, cr, de)) + return n * tSym, true +} + +// symbols is the Semtech payload symbol count with explicit header and CRC on. +func symbols(frameLen, sf, cr, de int) int { + n := math.Ceil(float64(8*frameLen-4*sf+44)/float64(4*(sf-2*de))) * float64(cr) + if n < 0 { + n = 0 + } + return 8 + int(n) +} diff --git a/internal/lora/airtime_test.go b/internal/lora/airtime_test.go new file mode 100644 index 0000000..795357b --- /dev/null +++ b/internal/lora/airtime_test.go @@ -0,0 +1,122 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package lora + +import ( + "math" + "testing" +) + +func TestPreambleSymbols(t *testing.T) { + cases := []struct { + sf int + want int + }{ + {7, 32}, + {8, 32}, + {9, 16}, + {12, 16}, + } + for _, tc := range cases { + if got := PreambleSymbols(tc.sf); got != tc.want { + t.Errorf("PreambleSymbols(%d) = %d, want %d", tc.sf, got, tc.want) + } + } +} + +func TestFrameLength(t *testing.T) { + cases := []struct { + name string + transportCodes bool + pathLen int + payloadLen int + want int + }{ + {"no transport codes, no path", false, 0, 10, 12}, + {"transport codes with path", true, 3, 10, 19}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := FrameLength(tc.transportCodes, tc.pathLen, tc.payloadLen); got != tc.want { + t.Errorf("FrameLength(%v, %d, %d) = %d, want %d", tc.transportCodes, tc.pathLen, tc.payloadLen, got, tc.want) + } + }) + } +} + +func TestSymbols(t *testing.T) { + cases := []struct { + name string + frameLen int + sf int + cr int + de int + want int + }{ + {"sf7 no ldro", 50, 7, 5, 0, 83}, + {"sf10 ldro", 50, 10, 5, 1, 73}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := symbols(tc.frameLen, tc.sf, tc.cr, tc.de); got != tc.want { + t.Errorf("symbols(%d, %d, %d, %d) = %d, want %d", tc.frameLen, tc.sf, tc.cr, tc.de, got, tc.want) + } + }) + } +} + +func TestTimeOnAirMs(t *testing.T) { + cases := []struct { + name string + frameLen int + sf int + bwKHz float64 + cr int + want float64 + }{ + {"sf7 bw62.5", 50, 7, 62.5, 5, 244.224}, + {"sf10 bw62.5 ldro", 50, 10, 62.5, 5, 1527.808}, + {"sf7 bw250", 50, 7, 250, 5, 61.056}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got, ok := TimeOnAirMs(tc.frameLen, tc.sf, tc.bwKHz, tc.cr) + if !ok { + t.Fatalf("TimeOnAirMs(%d, %d, %v, %d) not costable, want ok", tc.frameLen, tc.sf, tc.bwKHz, tc.cr) + } + if math.Abs(got-tc.want) > 0.001 { + t.Errorf("TimeOnAirMs(%d, %d, %v, %d) = %v, want %v", tc.frameLen, tc.sf, tc.bwKHz, tc.cr, got, tc.want) + } + }) + } +} + +func TestTimeOnAirMsNotCostable(t *testing.T) { + cases := []struct { + name string + frameLen int + sf int + bwKHz float64 + cr int + }{ + {"sf zero", 50, 0, 62.5, 5}, + {"sf below range", 50, 6, 62.5, 5}, + {"sf above range", 50, 13, 62.5, 5}, + {"bw zero", 50, 7, 0, 5}, + {"cr below range", 50, 7, 62.5, 4}, + {"cr above range", 50, 7, 62.5, 9}, + {"empty frame", 0, 7, 62.5, 5}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got, ok := TimeOnAirMs(tc.frameLen, tc.sf, tc.bwKHz, tc.cr) + if ok { + t.Errorf("TimeOnAirMs(%d, %d, %v, %d) = %v, ok; want not costable", tc.frameLen, tc.sf, tc.bwKHz, tc.cr, got) + } + if got != 0 { + t.Errorf("TimeOnAirMs(%d, %d, %v, %d) = %v, want 0 ms when not costable", tc.frameLen, tc.sf, tc.bwKHz, tc.cr, got) + } + }) + } +}