From 041d9c1f45d8cb733c3f9233b7cfb7cd53c83b80 Mon Sep 17 00:00:00 2001 From: MrAlders0n <55921894+MrAlders0n@users.noreply.github.com> Date: Mon, 5 Oct 2026 18:35:41 -0400 Subject: [PATCH] release: v2.0.2 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Release the changes on `dev` as v2.0.2: authenticated MeshMapper imports and the traffic-sorted observer directory, including rolling seven-day counts from available history. Bump the API version and regenerate Swagger. MeshMapper integration requires an API key. Regional admins’ existing per-region or grouped-region coverage keys also work for these APIs. Deployments should set `MESHMAPPER_API_KEY` and restart Beacon. Validation: `go build ./...`, `gofmt -l .` (empty), `go vet ./...`, `go test ./...`, `govulncheck ./...`, and race tests for MeshMapper and scopestore passed in a clean checkout. PostgreSQL integration tests require CI because Docker is unavailable locally. --- README.md | 45 ++ cmd/beacon/main.go | 7 +- config.yaml.example | 10 +- db/meshmapper_integration_test.go | 4 +- db/observer_directory.go | 31 ++ db/observer_directory_integration_test.go | 396 ++++++++++++++++ db/queries/observer_directory.sql | 91 ++++ db/sqlc/mock/querier.go | 17 +- db/sqlc/observer_directory.sql.go | 140 ++++++ db/sqlc/querier.go | 3 + docs/docs.go | 227 +++++++++- docs/historical-stats.md | 46 +- docs/swagger.json | 227 +++++++++- docs/swagger.yaml | 163 ++++++- env.example | 6 + internal/api/handlers/observer_directory.go | 140 ++++++ .../api/handlers/observer_directory_test.go | 165 +++++++ internal/api/handlers/observers.go | 1 + internal/api/handlers/stub_reader_test.go | 8 + internal/api/observer_directory.go | 48 ++ internal/api/reader.go | 1 + internal/api/router/admin_config_test.go | 2 +- internal/backup/export.go | 14 + internal/backup/export_test.go | 31 ++ internal/cache/cache_test.go | 4 + internal/cache/reader.go | 4 + internal/config/config.go | 5 + internal/config/meshmapper.go | 1 + internal/config/meshmapper_test.go | 44 ++ internal/meshmapper/auth_test.go | 427 ++++++++++++++++++ internal/meshmapper/channels.go | 17 +- internal/meshmapper/channels_test.go | 4 +- internal/meshmapper/client.go | 67 +++ internal/meshmapper/directory.go | 19 +- internal/meshmapper/groups_test.go | 2 +- internal/meshmapper/scopes.go | 89 ++-- internal/meshmapper/scopes_test.go | 10 +- internal/meshmapper/zones.go | 32 +- internal/meshmapper/zones_test.go | 14 +- 39 files changed, 2442 insertions(+), 120 deletions(-) create mode 100644 db/observer_directory.go create mode 100644 db/observer_directory_integration_test.go create mode 100644 db/queries/observer_directory.sql create mode 100644 db/sqlc/observer_directory.sql.go create mode 100644 internal/api/handlers/observer_directory.go create mode 100644 internal/api/handlers/observer_directory_test.go create mode 100644 internal/api/observer_directory.go create mode 100644 internal/meshmapper/auth_test.go create mode 100644 internal/meshmapper/client.go diff --git a/README.md b/README.md index dca6275..efd9530 100644 --- a/README.md +++ b/README.md @@ -63,6 +63,51 @@ and the fully annotated `config.yaml` lives in beacon-docs as well; the copy her starter. Migrations run on startup. The API listens on `LISTEN_ADDR` (default `:8080`) and Swagger is at `http://localhost:8080/swagger/index.html`. +### MeshMapper authentication + +MeshMapper APIs require either a regional API key or a grouped-region API key +covering multiple regions. Your local MeshMapper regional or grouped-region admin +can generate these keys. + +Set `MESHMAPPER_API_KEY` in each deployment's private environment or +gitignored `.env`; it overrides `meshmapper.api_key` in `config.yaml`, including +when empty. The examples leave the key empty. Never put it in browser settings +or use the mobile app's App key. Restart Beacon after changing the key. + +A regional deployment needs access to its IATA; a group deployment needs access +to every member IATA, including members added by `import_groups`. API keys do not +remove rate limits or change the `get_zones.php?country=...` request shape. +IP exemptions are not authentication. + +Beacon sends `X-API-Key` to `get_zones.php`, `get_geojson.php`, `get_scopes.php`, +and `get_channels.php`. The shared client also supports that header for +`get_repeaters.php`; this repository does not currently fetch repeaters. +Requests are restricted to those HTTPS MeshMapper endpoints and never follow +redirects. Coverage consumers keep their separate keys, scopes, quotas, and +supported `?key=` authentication; this client does not call `coverage.php`. + +Missing keys are reported as unconfigured without stopping Beacon or deleting +cached imports. Failed refreshes record `last_error`, retain the last successful +data and freshness timestamp, and report 401 as an authentication failure or +403 as a permission failure. Inspect the `meshmapper.zones`, `meshmapper.scopes`, +and `meshmapper.channels` logs for freshness and the next attempt. No anonymous +fallback occurs. Existing refresh limits still apply, including longer +`Retry-After` delays on 429/503. + +Use the environment setting when backups are enabled. Backup exports preserve +saved YAML verbatim and refuse a nonempty `meshmapper.api_key` in that file so +the credential cannot enter a backup download. Clear that YAML value after +moving the key to the environment; environment secrets are excluded from backups. + +Deployment coordination is required with MeshMapper Server before enforcement: +confirm `X-API-Key` support, obtain and verify the integration key's API and IATA +permissions, and confirm these requests do not consume Coverage quota. Header +support must not be assumed deployed. The first four endpoints can enforce +authentication before the app rollout; repeater enforcement waits for app 1.4.1 +and its forced update. Key issuance, permissions, quotas, and enforcement switches +remain MeshMapper Server work, tracked in +[MeshMapper_Server#431](https://github.com/MeshMapper/MeshMapper_Server/issues/431). + To run the web frontend against it, see [Running the full stack locally](https://github.com/MeshCore-Beacon/beacon-docs/blob/main/CONTRIBUTING.md#running-the-full-stack-locally). diff --git a/cmd/beacon/main.go b/cmd/beacon/main.go index e3af349..62172bb 100644 --- a/cmd/beacon/main.go +++ b/cmd/beacon/main.go @@ -45,7 +45,7 @@ import ( var version = "dev" // @title MeshCore Beacon API -// @version 2.0.1 +// @version 2.0.2 // @description MeshCore network observation backend. Ingests LoRa packets from MQTT brokers, stores in PostgreSQL, and streams live events via WebSocket. // @description REST requests share a configurable per-client rate limit (default 300/minute and a 300-request one-second burst cap). Exceeded limits return HTTP 429 with error.code=rate_limited and a Retry-After header in seconds. CORS preflights and WebSocket upgrades do not consume this API budget. // @description WebSocket upgrade attempts at /ws have a configurable per-client limit (default 10/minute, including failed handshakes). Rate exhaustion returns HTTP 429 with Retry-After before upgrade. An accepted socket exceeding the concurrent cap closes with code 1013 before hello; established connections remain open. @@ -197,8 +197,11 @@ func main() { scopes.Load(scopeEntries) scopes.SetManualMembers(cfg.ManualScopeMembers()) slog.Info(fmt.Sprintf("loaded %d transport scopes", len(scopeEntries)), "component", "startup") - directory := meshmapper.NewDirectory(store) + directory := meshmapper.NewDirectory(store, cfg.MeshMapper.APIKey) if cfg.MeshMapper.Scopes.Enabled || cfg.MeshMapper.Zones.Enabled || cfg.MeshMapper.Channels.Enabled { + if cfg.MeshMapper.APIKey == "" { + slog.Warn("MeshMapper integration unconfigured: set MESHMAPPER_API_KEY or meshmapper.api_key; retaining cached data", "component", "meshmapper") + } restoreCtx, cancelRestore := context.WithTimeout(ctx, 10*time.Second) err = directory.Restore(restoreCtx) cancelRestore() diff --git a/config.yaml.example b/config.yaml.example index 98ec8f7..0a61765 100644 --- a/config.yaml.example +++ b/config.yaml.example @@ -46,7 +46,14 @@ ratelimit: # A 429 returns error.code=rate_limited and Retry-After (1 or 60 seconds). burst: 300 -# MeshMapper integrations (optional, no API key). Import transport scopes and +# MeshMapper integrations (optional, require a regional or grouped-region API key). +# Your local MeshMapper regional or grouped-region admin can generate the key. +# A grouped-region key covers multiple regions. +# MESHMAPPER_API_KEY overrides meshmapper.api_key, even when empty. Use the +# environment for secrets, especially with backups; never use the mobile App key. +# Missing keys leave imports unconfigured; cached data and the rest of Beacon remain. +# 401/403 failures retain cached data and report authentication/permission errors. +# Import transport scopes and # IATA border outlines from MeshMapper instead of maintaining scopes: and # iatas.*.borderFile by hand. # @@ -64,6 +71,7 @@ ratelimit: # Region-filtered scope lists and scope stats show imported names only under the # IATAs whose catalogue lists them; IATAs without a catalogue show manual scopes only. #meshmapper: +# api_key: "" # scopes: # enabled: false # refresh_interval: 24h # 1h-24h (MeshMapper allows one get_scopes.php per region per 55m); one source per 15s tick diff --git a/db/meshmapper_integration_test.go b/db/meshmapper_integration_test.go index 5b9af83..072a2b4 100644 --- a/db/meshmapper_integration_test.go +++ b/db/meshmapper_integration_test.go @@ -83,7 +83,7 @@ func TestMeshMapperCataloguePostgres(t *testing.T) { t.Fatal(row, err) } scopes := scopestore.New() - _, err = meshmapper.New(ctx, config.MeshMapperScopesConfig{Enabled: true}, store, meshmapper.NewDirectory(store), scopes, manualRows) + _, err = meshmapper.New(ctx, config.MeshMapperScopesConfig{Enabled: true}, store, meshmapper.NewDirectory(store, ""), scopes, manualRows) if err != nil || len(scopes.Entries()) != 2 { t.Fatal("restart did not restore imported membership", err) } @@ -106,7 +106,7 @@ func TestMeshMapperCataloguePostgres(t *testing.T) { if err := store.SaveScopeCatalogue(ctx, "YOW", url, update, nil); err != nil { t.Fatal(err) } - _, err = meshmapper.New(ctx, config.MeshMapperScopesConfig{Enabled: true}, store, meshmapper.NewDirectory(store), scopes, manualRows) + _, err = meshmapper.New(ctx, config.MeshMapperScopesConfig{Enabled: true}, store, meshmapper.NewDirectory(store, ""), scopes, manualRows) if err != nil || len(scopes.Entries()) != 1 { t.Fatal("removed imported membership remained active", err) } diff --git a/db/observer_directory.go b/db/observer_directory.go new file mode 100644 index 0000000..c76a9fd --- /dev/null +++ b/db/observer_directory.go @@ -0,0 +1,31 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package db + +import ( + "context" + "encoding/json" + "time" + + sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" + "github.com/MeshCore-Beacon/beacon-server/internal/api" +) + +func (s *Store) ListObserverDirectory(ctx context.Context, query api.ObserverDirectoryQuery) (*api.ObserverDirectory, error) { + ctx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + raw, err := s.q.ListObserverDirectory(ctx, sqlc.ListObserverDirectoryParams{ + Column1: ts(time.UnixMilli(query.Since)), Column2: ts(time.UnixMilli(query.Until)), Column3: query.IATAs, + Column4: query.Broker, Column5: query.Name, Column6: query.Status, Column7: query.Scope, Column8: query.Type, Column9: query.Sort, + Column10: query.Cursor, Column11: query.Limit, Column12: query.MatchNone, + }) + if err != nil { + return nil, err + } + var page api.ObserverDirectory + if err := json.Unmarshal([]byte(raw), &page); err != nil { + return nil, err + } + return &page, nil +} diff --git a/db/observer_directory_integration_test.go b/db/observer_directory_integration_test.go new file mode 100644 index 0000000..cf61e78 --- /dev/null +++ b/db/observer_directory_integration_test.go @@ -0,0 +1,396 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package db + +import ( + "context" + "fmt" + "slices" + "testing" + "time" + + sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" + "github.com/MeshCore-Beacon/beacon-server/internal/api" + "github.com/jackc/pgx/v5" +) + +func TestObserverDirectoryPagesPostgres(t *testing.T) { + ctx, pool := schemaPool(t) + if err := RunMigrations(ctx, pool); err != nil { + t.Fatal(err) + } + _, err := pool.Exec(ctx, `INSERT INTO iata_codes(iata) VALUES ('YVR'),('YYJ'); + INSERT INTO observers(id,public_key,display_name,observer_type,last_iata,last_seen) SELECT lpad(to_hex(i),32,'0')::uuid,int4send(i),'Observer '||lpad(i::text,3,'0'),CASE WHEN i%2=0 THEN 'type-a' ELSE 'type-b' END,'YVR',now() FROM generate_series(1,205)i; + INSERT INTO analytics_rollup_hours(hour,status) VALUES ('2026-01-01 00:00:00+00','complete'); + INSERT INTO analytics_hourly_observer_identity(hour,iata,observer_id,observation_count) SELECT '2026-01-01 00:00:00+00','YVR',id,CASE WHEN display_name IN ('Observer 001','Observer 002') THEN 100 ELSE 1 END FROM observers WHERE display_name <> 'Observer 205';`) + if err != nil { + t.Fatal(err) + } + tx, err := pool.BeginTx(ctx, pgx.TxOptions{AccessMode: pgx.ReadOnly}) + if err != nil { + t.Fatal(err) + } + defer tx.Rollback(ctx) + store := &Store{q: sqlc.New(tx)} + q := api.ObserverDirectoryQuery{IATAs: []string{"YVR"}, Sort: "traffic", Since: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC).UnixMilli(), Until: time.Date(2026, 1, 1, 1, 0, 0, 0, time.UTC).UnixMilli(), Limit: 2} + first, err := store.ListObserverDirectory(ctx, q) + if err != nil { + t.Fatal(err) + } + if len(first.Items) != 2 || *first.Items[0].DisplayName != "Observer 001" || *first.Items[1].DisplayName != "Observer 002" || !first.HasMore || *first.MaxObservationCount != 100 { + t.Fatalf("first %+v", first) + } + if len(first.ObserverTypes) != 2 { + t.Fatalf("facets %+v", first.ObserverTypes) + } + seen := map[string]bool{} + page := first + for { + for _, item := range page.Items { + if seen[item.ID.String()] { + t.Fatal("duplicate observer") + } + seen[item.ID.String()] = true + if item.IATA != "YVR" || *item.DisplayName == "Changed" { + t.Fatal("metadata moved between pages") + } + } + if !page.HasMore { + break + } + q.Cursor = *page.NextCursor + q.Limit = 37 + page, err = store.ListObserverDirectory(ctx, q) + if err != nil { + t.Fatal(err) + } + if *page.MaxObservationCount != 100 || page.WindowEnd != first.WindowEnd { + t.Fatal("unchanged data produced different metrics") + } + } + if len(seen) != 205 { + t.Fatalf("got %d observers", len(seen)) + } + if *page.Items[len(page.Items)-1].ObservationCount != 0 { + t.Fatal("missing explicit zero") + } + q.Cursor = 0 + if _, err := pool.Exec(ctx, `UPDATE analytics_hourly_observer_identity SET observation_count=900; UPDATE observers SET display_name='Changed'`); err != nil { + t.Fatal(err) + } + refreshed, err := store.ListObserverDirectory(ctx, q) + if err != nil { + t.Fatal(err) + } + if *refreshed.MaxObservationCount != 900 || *refreshed.Items[0].DisplayName != "Changed" { + t.Fatal("directory did not reflect current data") + } + q.Cursor = 1000 + empty, err := store.ListObserverDirectory(ctx, q) + if err != nil || len(empty.Items) != 0 || empty.HasMore || empty.NextCursor != nil { + t.Fatalf("terminal page %+v, %v", empty, err) + } + +} + +func TestObserverDirectoryFiltersAndCoveragePostgres(t *testing.T) { + ctx, pool := schemaPool(t) + if err := RunMigrations(ctx, pool); err != nil { + t.Fatal(err) + } + _, err := pool.Exec(ctx, `INSERT INTO iata_codes(iata) VALUES ('YVR'),('YYJ'); + INSERT INTO observers(id,public_key,display_name,observer_type,last_iata,last_seen,last_status_at) VALUES + ('00000000-0000-0000-0000-000000000001','\x01','Zulu','a','YVR',now(),now()), + ('00000000-0000-0000-0000-000000000002','\x02','alpha','b','YVR',now()-interval '1 hour',now()-interval '1 hour'), + ('00000000-0000-0000-0000-000000000003','\x03','alpha','a','YVR',now(),now()), + ('00000000-0000-0000-0000-000000000004','\x04','Other','b','YYJ',now(),now()); + INSERT INTO observer_brokers(observer_id,broker_name) VALUES ('00000000-0000-0000-0000-000000000001','mqtt1'); + INSERT INTO transport_scopes(id,name,transport_key,key_fingerprint) VALUES (1,'#test','\x01','\x01'); + INSERT INTO observer_scopes(observer_id,scope_id) VALUES ('00000000-0000-0000-0000-000000000001',1); + INSERT INTO analytics_rollup_hours(hour,status) VALUES ('2026-01-01 00:00:00+00','complete'); + INSERT INTO analytics_hourly_observer_identity(hour,iata,observer_id,observation_count) VALUES + ('2026-01-01 00:00:00+00','YVR','00000000-0000-0000-0000-000000000001',5), + ('2026-01-01 00:00:00+00','YYJ','00000000-0000-0000-0000-000000000001',999), + ('2026-01-01 00:00:00+00','YVR','00000000-0000-0000-0000-000000000002',8), + ('2025-12-31 00:00:00+00','YVR','00000000-0000-0000-0000-000000000003',100);`) + if err != nil { + t.Fatal(err) + } + store := New(pool, 0, 0) + base := api.ObserverDirectoryQuery{IATAs: []string{"YVR"}, Sort: "traffic", Since: 1767225600000, Until: 1767229200000, Limit: 1} + cases := []struct { + name string + apply func(*api.ObserverDirectoryQuery) + ids []byte + }{ + {"traffic", func(q *api.ObserverDirectoryQuery) {}, []byte{2, 1, 3}}, + {"name ties", func(q *api.ObserverDirectoryQuery) { q.Sort = "name" }, []byte{2, 3, 1}}, + {"name filter", func(q *api.ObserverDirectoryQuery) { q.Name = "zUL" }, []byte{1}}, + {"type", func(q *api.ObserverDirectoryQuery) { q.Type = "a" }, []byte{1, 3}}, + {"broker", func(q *api.ObserverDirectoryQuery) { q.Broker = "mqtt1" }, []byte{1}}, + {"scope", func(q *api.ObserverDirectoryQuery) { q.Scope = "#test" }, []byte{1}}, + {"status", func(q *api.ObserverDirectoryQuery) { q.Status = "offline" }, []byte{2}}, + {"empty region", func(q *api.ObserverDirectoryQuery) { q.MatchNone = true }, []byte{}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + q := base + tc.apply(&q) + page, err := store.ListObserverDirectory(ctx, q) + if err != nil { + t.Fatal(err) + } + var ids []byte + for { + for _, o := range page.Items { + ids = append(ids, o.ID[15]) + if o.ID[15] == 1 && *o.ObservationCount != 5 { + t.Fatal("count included other IATA") + } + if o.ID[15] == 3 && *o.ObservationCount != 0 { + t.Fatal("count included other hour") + } + } + if !page.HasMore { + break + } + q.Cursor = *page.NextCursor + page, err = New(pool, 0, 0).ListObserverDirectory(ctx, q) + if err != nil { + t.Fatal(err) + } + } + if !slices.Equal(ids, tc.ids) { + t.Fatalf("ids %v want %v", ids, tc.ids) + } + if tc.name == "type" && len(page.ObserverTypes) != 2 { + t.Fatal("type filter hid alternatives") + } + if tc.name == "name ties" && *page.MaxObservationCount != 8 { + t.Fatal("bar maximum followed page instead of full result") + } + }) + } + for _, status := range []string{"partial", "missing"} { + if _, err := pool.Exec(ctx, `UPDATE analytics_rollup_hours SET status=$1`, status); err != nil { + t.Fatal(err) + } + page, err := store.ListObserverDirectory(ctx, base) + if err != nil { + t.Fatal(err) + } + want := "partial" + if status == "missing" { + want = "unavailable" + } + if page.Coverage.Status != want || page.EffectiveSort != "traffic" || page.MaxObservationCount == nil || *page.MaxObservationCount != 0 || page.Items[0].ObservationCount == nil || *page.Items[0].ObservationCount != 0 { + t.Fatalf("incomplete history suppressed numeric counts or traffic sorting: %+v", page) + } + } +} + +func directoryFixture(t *testing.T) (context.Context, pgx.Tx, *Store) { + t.Helper() + ctx, tx := retentionTx(t) + applyBaseline(t, ctx, tx) + if _, err := tx.Exec(ctx, `INSERT INTO observers(id,public_key,display_name,observer_type,last_iata,last_seen,last_status_at) VALUES + ('00000000-0000-0000-0000-000000000001','\x01','Zulu','a','YVR',now(),now()), + ('00000000-0000-0000-0000-000000000002','\x02','alpha','b','YVR',now()-interval '1 hour',now()-interval '1 hour'), + ('00000000-0000-0000-0000-000000000003','\x03','alpha','a','YVR',now(),now()), + ('00000000-0000-0000-0000-000000000004','\x04','Other','b','YYJ',now(),now()); + INSERT INTO observer_brokers(observer_id,broker_name) VALUES ('00000000-0000-0000-0000-000000000001','mqtt1'); + INSERT INTO transport_scopes(id,name,transport_key,key_fingerprint) VALUES (1,'#test','\x01','\x01'); + INSERT INTO observer_scopes(observer_id,scope_id) VALUES ('00000000-0000-0000-0000-000000000001',1);`); err != nil { + t.Fatal(err) + } + return ctx, tx, &Store{q: sqlc.New(tx)} +} + +func directoryObservation(t *testing.T, ctx context.Context, tx pgx.Tx, observer int, iata string, at time.Time) { + t.Helper() + _, err := tx.Exec(ctx, `WITH packet AS ( + INSERT INTO packets(packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,first_heard_at,last_heard_at) + VALUES (decode(md5(gen_random_uuid()::text),'hex'),4,0,1,'\x00','\x00',$1,$1) RETURNING packet_hash + ) INSERT INTO packet_observations(packet_hash,observer_id,iata,heard_at,path_length_byte,hash_size,hop_count) + SELECT packet_hash,$2,$3,$1,0,1,0 FROM packet`, at, fmt.Sprintf("00000000-0000-0000-0000-%012d", observer), iata) + if err != nil { + t.Fatal(err) + } +} + +func TestObserverDirectoryMixedHistoryPostgres(t *testing.T) { + ctx, tx, store := directoryFixture(t) + _, err := tx.Exec(ctx, ` + INSERT INTO analytics_rollup_hours(hour,status) VALUES + ('2026-01-01 00:00:00+00','complete'), ('2026-01-01 01:00:00+00','complete'), + ('2026-01-01 02:00:00+00','partial'), ('2026-01-01 03:00:00+00','missing'), + ('2026-01-01 04:00:00+00','complete'); + INSERT INTO analytics_hourly_observer_identity(hour,iata,observer_id,observation_count) VALUES + ('2026-01-01 00:00:00+00','YVR','00000000-0000-0000-0000-000000000001',900), + ('2026-01-01 01:00:00+00','YVR','00000000-0000-0000-0000-000000000001',10), + ('2026-01-01 01:00:00+00','YVR','00000000-0000-0000-0000-000000000002',12), + ('2026-01-01 02:00:00+00','YVR','00000000-0000-0000-0000-000000000001',900), + ('2026-01-01 03:00:00+00','YVR','00000000-0000-0000-0000-000000000001',900), + ('2026-01-01 04:00:00+00','YVR','00000000-0000-0000-0000-000000000001',900);`) + if err != nil { + t.Fatal(err) + } + for _, row := range []struct { + id int + iata, at string + }{ + {1, "YVR", "00:30:00.122"}, {1, "YVR", "00:30:00.123"}, {2, "YVR", "00:59:59.999"}, + {1, "YVR", "01:30:00"}, {2, "YVR", "01:35:00"}, + {1, "YVR", "02:15:00"}, {1, "YVR", "02:20:00"}, {1, "YYJ", "02:25:00"}, + {2, "YVR", "03:15:00"}, {4, "YVR", "03:20:00"}, + {1, "YVR", "04:20:00.455"}, {1, "YVR", "04:20:00.456"}, {1, "YVR", "05:00:00"}, + } { + at, err := time.Parse(time.RFC3339Nano, "2026-01-01T"+row.at+"Z") + if err != nil { + t.Fatal(err) + } + directoryObservation(t, ctx, tx, row.id, row.iata, at) + } + since := time.Date(2026, 1, 1, 0, 30, 0, 123000000, time.UTC).UnixMilli() + until := time.Date(2026, 1, 1, 4, 20, 0, 456000000, time.UTC).UnixMilli() + base := api.ObserverDirectoryQuery{IATAs: []string{"YVR"}, Sort: "traffic", Since: since, Until: until, Limit: 1} + for _, tc := range []struct { + name string + apply func(*api.ObserverDirectoryQuery) + ids []byte + maximum int64 + }{ + {"traffic", func(*api.ObserverDirectoryQuery) {}, []byte{2, 1, 3}, 14}, + {"name", func(q *api.ObserverDirectoryQuery) { q.Sort = "name" }, []byte{2, 3, 1}, 14}, + {"type", func(q *api.ObserverDirectoryQuery) { q.Type = "a" }, []byte{1, 3}, 14}, + {"broker", func(q *api.ObserverDirectoryQuery) { q.Broker = "mqtt1" }, []byte{1}, 14}, + {"status", func(q *api.ObserverDirectoryQuery) { q.Status = "offline" }, []byte{2}, 14}, + {"name filter", func(q *api.ObserverDirectoryQuery) { q.Name = "ALP" }, []byte{2, 3}, 14}, + {"scope", func(q *api.ObserverDirectoryQuery) { q.Scope = "#test" }, []byte{1}, 14}, + {"all IATAs", func(q *api.ObserverDirectoryQuery) { q.IATAs = nil }, []byte{1, 2, 4, 3}, 15}, + {"multiple IATAs", func(q *api.ObserverDirectoryQuery) { q.IATAs = []string{"YVR", "YYJ"} }, []byte{1, 2, 4, 3}, 15}, + {"empty region", func(q *api.ObserverDirectoryQuery) { q.MatchNone = true }, nil, 0}, + } { + t.Run(tc.name, func(t *testing.T) { + q := base + tc.apply(&q) + var ids []byte + for { + page, err := store.ListObserverDirectory(ctx, q) + if err != nil { + t.Fatal(err) + } + if page.MaxObservationCount == nil || *page.MaxObservationCount != tc.maximum || page.EffectiveSort != q.Sort || page.WindowStart != since || page.WindowEnd != until { + t.Fatalf("page metrics %+v", page) + } + if page.Coverage.Status != "partial" || page.Coverage.ExpectedHours != 5 || page.Coverage.CompleteHours != 1 { + t.Fatalf("coverage %+v", page.Coverage) + } + for _, item := range page.Items { + ids = append(ids, item.ID[15]) + want := int64(14) + if item.ID[15] == 3 { + want = 0 + } + if item.ID[15] == 4 { + want = 1 + } + if item.ID[15] == 1 && tc.maximum == 15 { + want = 15 + } + if item.ObservationCount == nil || *item.ObservationCount != want { + t.Fatalf("observer %d count %v, want %d", item.ID[15], item.ObservationCount, want) + } + } + if !page.HasMore { + if page.NextCursor != nil { + t.Fatal("terminal page has cursor") + } + break + } + if page.NextCursor == nil || *page.NextCursor <= q.Cursor { + t.Fatal("cursor did not advance") + } + q.Cursor = *page.NextCursor + } + if !slices.Equal(ids, tc.ids) { + t.Fatalf("ids %v, want %v", ids, tc.ids) + } + }) + } + // Raw retention removes both the first boundary slice and the rolled hour's raw rows. + if _, err := tx.Exec(ctx, `DELETE FROM packet_observations WHERE heard_at < '2026-01-01 02:00:00+00'; UPDATE analytics_raw_state SET raw_deleted_before='2026-01-01 01:30:00+00'`); err != nil { + t.Fatal(err) + } + q := base + q.Limit = 10 + page, err := store.ListObserverDirectory(ctx, q) + if err != nil { + t.Fatal(err) + } + if len(page.Items) != 3 || page.Items[0].ObservationCount == nil || *page.Items[0].ObservationCount != 13 || page.Items[1].ObservationCount == nil || *page.Items[1].ObservationCount != 13 { + t.Fatalf("retention lost rolled history or included an out-of-window bucket: %+v", page) + } +} + +func TestObserverDirectoryRawStartupAndZeroPostgres(t *testing.T) { + ctx, tx, store := directoryFixture(t) + now := time.Now().UTC().Truncate(time.Millisecond) + q := api.ObserverDirectoryQuery{IATAs: []string{"YVR"}, Sort: "traffic", Since: now.Add(-7 * 24 * time.Hour).UnixMilli(), Until: now.UnixMilli(), Limit: 10} + page, err := store.ListObserverDirectory(ctx, q) + if err != nil { + t.Fatal(err) + } + if page.Coverage.Status != "unavailable" || page.EffectiveSort != "traffic" || page.MaxObservationCount == nil || *page.MaxObservationCount != 0 || len(page.Items) != 3 { + t.Fatalf("empty history: %+v", page) + } + for _, item := range page.Items { + if item.ObservationCount == nil || *item.ObservationCount != 0 { + t.Fatal("no history must return numeric zero") + } + } + directoryObservation(t, ctx, tx, 1, "YVR", now.Add(-6*24*time.Hour)) + directoryObservation(t, ctx, tx, 1, "YVR", now.Add(-time.Millisecond)) + directoryObservation(t, ctx, tx, 1, "YVR", now.Add(-7*24*time.Hour-time.Millisecond)) + directoryObservation(t, ctx, tx, 1, "YVR", now) + directoryObservation(t, ctx, tx, 1, "YVR", now.Add(time.Millisecond)) + page, err = store.ListObserverDirectory(ctx, q) + if err != nil { + t.Fatal(err) + } + if page.Coverage.Status != "partial" || page.Coverage.CompleteHours != 0 || page.Coverage.PartialHours != 2 || page.Coverage.MissingHours == 0 || page.EffectiveSort != "traffic" || page.Items[0].ID[15] != 1 || page.MaxObservationCount == nil || *page.MaxObservationCount != 2 || page.Items[0].ObservationCount == nil || *page.Items[0].ObservationCount != 2 { + t.Fatalf("available raw history and live tail were not counted: %+v", page) + } +} + +func TestObserverDirectoryExactSlicePostgres(t *testing.T) { + ctx, tx, store := directoryFixture(t) + at := time.Date(2026, 1, 1, 0, 30, 0, 123000000, time.UTC) + if _, err := tx.Exec(ctx, `SET LOCAL TIME ZONE 'America/Toronto'; + INSERT INTO analytics_rollup_hours(hour,status) VALUES ('2026-01-01 00:00:00+00','complete'); + INSERT INTO analytics_hourly_observer_identity(hour,iata,observer_id,observation_count) + VALUES ('2026-01-01 00:00:00+00','YVR','00000000-0000-0000-0000-000000000001',900)`); err != nil { + t.Fatal(err) + } + for _, delta := range []time.Duration{-time.Millisecond, 0, time.Millisecond} { + directoryObservation(t, ctx, tx, 1, "YVR", at.Add(delta)) + } + q := api.ObserverDirectoryQuery{IATAs: []string{"YVR"}, Sort: "traffic", Since: at.UnixMilli(), Until: at.Add(time.Millisecond).UnixMilli(), Limit: 10} + page, err := store.ListObserverDirectory(ctx, q) + if err != nil { + t.Fatal(err) + } + if page.WindowStart != q.Since || page.WindowEnd != q.Until || page.MaxObservationCount == nil || *page.MaxObservationCount != 1 || page.Items[0].ObservationCount == nil || *page.Items[0].ObservationCount != 1 || page.Coverage.ExpectedHours != 1 || page.Coverage.PartialHours != 1 || page.Coverage.CompleteHours != 0 { + t.Fatalf("exact slice included a whole rollup or wrong boundary: %+v", page) + } + // A refresh advances the exclusive bound to include the next arrival. + q.Until++ + page, err = store.ListObserverDirectory(ctx, q) + if err != nil { + t.Fatal(err) + } + if page.Items[0].ObservationCount == nil || *page.Items[0].ObservationCount != 2 || page.MaxObservationCount == nil || *page.MaxObservationCount != 2 { + t.Fatalf("refresh missed new traffic: %+v", page) + } +} diff --git a/db/queries/observer_directory.sql b/db/queries/observer_directory.sql new file mode 100644 index 0000000..93ec5f9 --- /dev/null +++ b/db/queries/observer_directory.sql @@ -0,0 +1,91 @@ +-- Copyright 2026 Beacon Contributors +-- SPDX-License-Identifier: AGPL-3.0-or-later + +-- name: ListObserverDirectory :one +WITH hours AS MATERIALIZED ( + SELECT g.h, coalesce(r.status, 'missing') AS status, + coalesce(r.status = 'complete', false) AND g.h >= $1::timestamptz + AND g.h + interval '1 hour' <= $2::timestamptz AS use_rollup + FROM generate_series(date_trunc('hour', $1::timestamptz, 'UTC'), $2::timestamptz, interval '1 hour') g(h) + LEFT JOIN analytics_rollup_hours r ON r.hour = g.h + WHERE g.h < $2::timestamptz +), candidates AS MATERIALIZED ( + SELECT o.id, o.display_name, o.observer_type, + coalesce(o.last_iata, '')::text AS iata, + CASE WHEN greatest(coalesce(o.last_status_at, o.last_seen), o.last_seen) > now() - interval '5 minutes' + THEN 'online' ELSE 'offline' END AS status, + CASE WHEN o.radio_freq_mhz IS NOT NULL AND o.radio_bw_khz IS NOT NULL AND o.radio_sf IS NOT NULL + THEN concat(o.radio_freq_mhz, ',', o.radio_bw_khz, ',', o.radio_sf) END AS radio, + coalesce((SELECT jsonb_agg(ts.name ORDER BY ts.name) FROM observer_scopes os JOIN transport_scopes ts ON ts.id = os.scope_id WHERE os.observer_id = o.id), '[]'::jsonb) AS scopes + FROM observers o + WHERE NOT $12::boolean AND (coalesce(cardinality($3::bpchar[]), 0) = 0 OR o.last_iata = ANY($3::bpchar[])) + AND ($4::text = '' OR EXISTS (SELECT 1 FROM observer_brokers ob WHERE ob.observer_id = o.id AND ob.broker_name = $4)) + AND ($5::text = '' OR o.display_name ILIKE '%' || $5 || '%') + AND ($6::text = '' OR CASE WHEN greatest(coalesce(o.last_status_at, o.last_seen), o.last_seen) > now() - interval '5 minutes' THEN 'online' ELSE 'offline' END = $6) + AND ($7::text = '' OR EXISTS (SELECT 1 FROM observer_scopes os JOIN transport_scopes ts ON ts.id = os.scope_id WHERE os.observer_id = o.id AND ts.name = $7)) +), raw_traffic AS MATERIALIZED ( + -- Disjoint hour slices keep raw scans bounded and never recount a completed whole hour. + SELECT h.h, r.observer_id, r.observations + FROM hours h + CROSS JOIN LATERAL ( + SELECT o.observer_id, count(*)::bigint AS observations + FROM packet_observations o + WHERE o.heard_at >= greatest(h.h, $1::timestamptz) + AND o.heard_at < least(h.h + interval '1 hour', $2::timestamptz) + AND (coalesce(cardinality($3::bpchar[]), 0) = 0 OR o.iata = ANY($3::bpchar[])) + GROUP BY o.observer_id + ) r + JOIN candidates c ON c.id = r.observer_id + WHERE NOT h.use_rollup +), coverage AS ( + SELECT count(*)::integer AS expected, + count(*) FILTER (WHERE h.use_rollup)::integer AS complete, + count(*) FILTER (WHERE NOT h.use_rollup AND (h.status IN ('complete', 'partial') OR r.h IS NOT NULL))::integer AS partial, + count(*) FILTER (WHERE NOT h.use_rollup AND h.status = 'missing' AND r.h IS NULL)::integer AS missing + FROM hours h LEFT JOIN (SELECT DISTINCT h FROM raw_traffic) r ON r.h = h.h +), traffic AS ( + SELECT observer_id, sum(observations)::bigint AS observations + FROM ( + SELECT a.observer_id, a.observation_count AS observations + FROM analytics_hourly_observer_identity a + JOIN hours h ON h.h = a.hour AND h.use_rollup + JOIN candidates c ON c.id = a.observer_id + WHERE (coalesce(cardinality($3::bpchar[]), 0) = 0 OR a.iata = ANY($3::bpchar[])) + UNION ALL + SELECT observer_id, observations FROM raw_traffic + ) counts + GROUP BY observer_id +), listed AS ( + SELECT c.*, coalesce(t.observations, 0)::bigint AS observations + FROM candidates c LEFT JOIN traffic t ON t.observer_id = c.id + WHERE $8::text = '' OR c.observer_type = $8 +), paged AS ( + SELECT * FROM listed + ORDER BY CASE WHEN $9::text = 'traffic' THEN observations END DESC NULLS LAST, + lower(coalesce(display_name, '')) COLLATE "C", id + LIMIT $11::integer OFFSET $10::bigint +), packed AS ( + SELECT coalesce(jsonb_agg( + jsonb_strip_nulls(jsonb_build_object('id', id, 'displayName', display_name, 'observerType', observer_type, 'iata', iata, 'status', status, 'radio', radio, 'scopes', scopes)) + || jsonb_build_object('observationCount', observations) + ORDER BY CASE WHEN $9::text = 'traffic' THEN observations END DESC NULLS LAST, + lower(coalesce(display_name, '')) COLLATE "C", id + ), '[]'::jsonb) AS items + FROM paged +), totals AS ( + SELECT count(*) AS n, max(observations) AS maximum FROM listed +) +SELECT jsonb_build_object( + 'items', p.items, + 'generatedAt', (extract(epoch FROM now())*1000)::bigint, + 'hasMore', t.n > $10::bigint + $11::integer, + 'nextCursor', CASE WHEN t.n > $10::bigint + $11::integer THEN $10::bigint + $11::integer END, + 'windowStart', (extract(epoch FROM $1::timestamptz)*1000)::bigint, + 'windowEnd', (extract(epoch FROM $2::timestamptz)*1000)::bigint, + 'sort', $9::text, 'effectiveSort', $9::text, + 'maxObservationCount', coalesce(t.maximum, 0), + 'observerTypes', (SELECT coalesce(jsonb_agg(t.observer_type ORDER BY t.observer_type), '[]'::jsonb) FROM (SELECT DISTINCT observer_type FROM candidates WHERE observer_type IS NOT NULL AND observer_type <> '') t), + 'coverage', jsonb_build_object('status', CASE WHEN v.complete = v.expected THEN 'complete' WHEN v.complete > 0 OR v.partial > 0 THEN 'partial' ELSE 'unavailable' END, + 'expectedHours', v.expected, 'completeHours', v.complete, 'partialHours', v.partial, 'missingHours', v.missing) +)::text AS page +FROM packed p CROSS JOIN coverage v CROSS JOIN totals t; diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index b34664f..114058a 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. @@ -1466,6 +1466,21 @@ func (mr *MockQuerierMockRecorder) ListObserverAdverts(ctx, arg any) *gomock.Cal return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListObserverAdverts", reflect.TypeOf((*MockQuerier)(nil).ListObserverAdverts), ctx, arg) } +// ListObserverDirectory mocks base method. +func (m *MockQuerier) ListObserverDirectory(ctx context.Context, arg db.ListObserverDirectoryParams) (string, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ListObserverDirectory", ctx, arg) + ret0, _ := ret[0].(string) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// ListObserverDirectory indicates an expected call of ListObserverDirectory. +func (mr *MockQuerierMockRecorder) ListObserverDirectory(ctx, arg any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListObserverDirectory", reflect.TypeOf((*MockQuerier)(nil).ListObserverDirectory), ctx, arg) +} + // ListObservers mocks base method. func (m *MockQuerier) ListObservers(ctx context.Context, arg db.ListObserversParams) ([]db.ListObserversRow, error) { m.ctrl.T.Helper() diff --git a/db/sqlc/observer_directory.sql.go b/db/sqlc/observer_directory.sql.go new file mode 100644 index 0000000..f6a878c --- /dev/null +++ b/db/sqlc/observer_directory.sql.go @@ -0,0 +1,140 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: observer_directory.sql + +package db + +import ( + "context" + + "github.com/jackc/pgx/v5/pgtype" +) + +const listObserverDirectory = `-- name: ListObserverDirectory :one + +WITH hours AS MATERIALIZED ( + SELECT g.h, coalesce(r.status, 'missing') AS status, + coalesce(r.status = 'complete', false) AND g.h >= $1::timestamptz + AND g.h + interval '1 hour' <= $2::timestamptz AS use_rollup + FROM generate_series(date_trunc('hour', $1::timestamptz, 'UTC'), $2::timestamptz, interval '1 hour') g(h) + LEFT JOIN analytics_rollup_hours r ON r.hour = g.h + WHERE g.h < $2::timestamptz +), candidates AS MATERIALIZED ( + SELECT o.id, o.display_name, o.observer_type, + coalesce(o.last_iata, '')::text AS iata, + CASE WHEN greatest(coalesce(o.last_status_at, o.last_seen), o.last_seen) > now() - interval '5 minutes' + THEN 'online' ELSE 'offline' END AS status, + CASE WHEN o.radio_freq_mhz IS NOT NULL AND o.radio_bw_khz IS NOT NULL AND o.radio_sf IS NOT NULL + THEN concat(o.radio_freq_mhz, ',', o.radio_bw_khz, ',', o.radio_sf) END AS radio, + coalesce((SELECT jsonb_agg(ts.name ORDER BY ts.name) FROM observer_scopes os JOIN transport_scopes ts ON ts.id = os.scope_id WHERE os.observer_id = o.id), '[]'::jsonb) AS scopes + FROM observers o + WHERE NOT $12::boolean AND (coalesce(cardinality($3::bpchar[]), 0) = 0 OR o.last_iata = ANY($3::bpchar[])) + AND ($4::text = '' OR EXISTS (SELECT 1 FROM observer_brokers ob WHERE ob.observer_id = o.id AND ob.broker_name = $4)) + AND ($5::text = '' OR o.display_name ILIKE '%' || $5 || '%') + AND ($6::text = '' OR CASE WHEN greatest(coalesce(o.last_status_at, o.last_seen), o.last_seen) > now() - interval '5 minutes' THEN 'online' ELSE 'offline' END = $6) + AND ($7::text = '' OR EXISTS (SELECT 1 FROM observer_scopes os JOIN transport_scopes ts ON ts.id = os.scope_id WHERE os.observer_id = o.id AND ts.name = $7)) +), raw_traffic AS MATERIALIZED ( + -- Disjoint hour slices keep raw scans bounded and never recount a completed whole hour. + SELECT h.h, r.observer_id, r.observations + FROM hours h + CROSS JOIN LATERAL ( + SELECT o.observer_id, count(*)::bigint AS observations + FROM packet_observations o + WHERE o.heard_at >= greatest(h.h, $1::timestamptz) + AND o.heard_at < least(h.h + interval '1 hour', $2::timestamptz) + AND (coalesce(cardinality($3::bpchar[]), 0) = 0 OR o.iata = ANY($3::bpchar[])) + GROUP BY o.observer_id + ) r + JOIN candidates c ON c.id = r.observer_id + WHERE NOT h.use_rollup +), coverage AS ( + SELECT count(*)::integer AS expected, + count(*) FILTER (WHERE h.use_rollup)::integer AS complete, + count(*) FILTER (WHERE NOT h.use_rollup AND (h.status IN ('complete', 'partial') OR r.h IS NOT NULL))::integer AS partial, + count(*) FILTER (WHERE NOT h.use_rollup AND h.status = 'missing' AND r.h IS NULL)::integer AS missing + FROM hours h LEFT JOIN (SELECT DISTINCT h FROM raw_traffic) r ON r.h = h.h +), traffic AS ( + SELECT observer_id, sum(observations)::bigint AS observations + FROM ( + SELECT a.observer_id, a.observation_count AS observations + FROM analytics_hourly_observer_identity a + JOIN hours h ON h.h = a.hour AND h.use_rollup + JOIN candidates c ON c.id = a.observer_id + WHERE (coalesce(cardinality($3::bpchar[]), 0) = 0 OR a.iata = ANY($3::bpchar[])) + UNION ALL + SELECT observer_id, observations FROM raw_traffic + ) counts + GROUP BY observer_id +), listed AS ( + SELECT c.id, c.display_name, c.observer_type, c.iata, c.status, c.radio, c.scopes, coalesce(t.observations, 0)::bigint AS observations + FROM candidates c LEFT JOIN traffic t ON t.observer_id = c.id + WHERE $8::text = '' OR c.observer_type = $8 +), paged AS ( + SELECT id, display_name, observer_type, iata, status, radio, scopes, observations FROM listed + ORDER BY CASE WHEN $9::text = 'traffic' THEN observations END DESC NULLS LAST, + lower(coalesce(display_name, '')) COLLATE "C", id + LIMIT $11::integer OFFSET $10::bigint +), packed AS ( + SELECT coalesce(jsonb_agg( + jsonb_strip_nulls(jsonb_build_object('id', id, 'displayName', display_name, 'observerType', observer_type, 'iata', iata, 'status', status, 'radio', radio, 'scopes', scopes)) + || jsonb_build_object('observationCount', observations) + ORDER BY CASE WHEN $9::text = 'traffic' THEN observations END DESC NULLS LAST, + lower(coalesce(display_name, '')) COLLATE "C", id + ), '[]'::jsonb) AS items + FROM paged +), totals AS ( + SELECT count(*) AS n, max(observations) AS maximum FROM listed +) +SELECT jsonb_build_object( + 'items', p.items, + 'generatedAt', (extract(epoch FROM now())*1000)::bigint, + 'hasMore', t.n > $10::bigint + $11::integer, + 'nextCursor', CASE WHEN t.n > $10::bigint + $11::integer THEN $10::bigint + $11::integer END, + 'windowStart', (extract(epoch FROM $1::timestamptz)*1000)::bigint, + 'windowEnd', (extract(epoch FROM $2::timestamptz)*1000)::bigint, + 'sort', $9::text, 'effectiveSort', $9::text, + 'maxObservationCount', coalesce(t.maximum, 0), + 'observerTypes', (SELECT coalesce(jsonb_agg(t.observer_type ORDER BY t.observer_type), '[]'::jsonb) FROM (SELECT DISTINCT observer_type FROM candidates WHERE observer_type IS NOT NULL AND observer_type <> '') t), + 'coverage', jsonb_build_object('status', CASE WHEN v.complete = v.expected THEN 'complete' WHEN v.complete > 0 OR v.partial > 0 THEN 'partial' ELSE 'unavailable' END, + 'expectedHours', v.expected, 'completeHours', v.complete, 'partialHours', v.partial, 'missingHours', v.missing) +)::text AS page +FROM packed p CROSS JOIN coverage v CROSS JOIN totals t +` + +type ListObserverDirectoryParams struct { + Column1 pgtype.Timestamptz `json:"column_1"` + Column2 pgtype.Timestamptz `json:"column_2"` + Column3 []string `json:"column_3"` + Column4 string `json:"column_4"` + Column5 string `json:"column_5"` + Column6 string `json:"column_6"` + Column7 string `json:"column_7"` + Column8 string `json:"column_8"` + Column9 string `json:"column_9"` + Column10 int64 `json:"column_10"` + Column11 int32 `json:"column_11"` + Column12 bool `json:"column_12"` +} + +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later +func (q *Queries) ListObserverDirectory(ctx context.Context, arg ListObserverDirectoryParams) (string, error) { + row := q.db.QueryRow(ctx, listObserverDirectory, + arg.Column1, + arg.Column2, + arg.Column3, + arg.Column4, + arg.Column5, + arg.Column6, + arg.Column7, + arg.Column8, + arg.Column9, + arg.Column10, + arg.Column11, + arg.Column12, + ) + var page string + err := row.Scan(&page) + return page, err +} diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index f3c54d3..8bb7be5 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -221,6 +221,9 @@ type Querier interface { // Pass cursor=0 to start from the beginning, or the last seen id for pagination. // Keep missing-origin adverts; the generated key field expects a string, not NULL. ListObserverAdverts(ctx context.Context, arg ListObserverAdvertsParams) ([]ListObserverAdvertsRow, error) + // Copyright 2026 Beacon Contributors + // SPDX-License-Identifier: AGPL-3.0-or-later + ListObserverDirectory(ctx context.Context, arg ListObserverDirectoryParams) (string, error) // Pass cursor=0 to start from the beginning, or the last seen observer's rownum for pagination. // Note: observers use UUID PKs so we order by last_seen and use a keyset on last_seen+id. ListObservers(ctx context.Context, arg ListObserversParams) ([]ListObserversRow, error) diff --git a/docs/docs.go b/docs/docs.go index 6fc5abc..227df2a 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -1430,6 +1430,127 @@ const docTemplate = `{ } } }, + "/observers/directory": { + "get": { + "description": "Defaults to the trailing seven days through request time. Counts combine completed whole-hour analytics with retained observations for other hours and exact boundary slices. Incomplete history still returns numeric counts and the requested sort; coverage describes available history. Repeat filters and the returned windowStart/windowEnd as since/until with the cursor on later pages. Results may move as data changes; refresh without bounds for a new rolling window.", + "produces": [ + "application/json" + ], + "tags": [ + "Observers" + ], + "summary": "List observers with windowed traffic counts", + "parameters": [ + { + "type": "string", + "description": "Single IATA", + "name": "iata", + "in": "query" + }, + { + "type": "string", + "description": "Comma-separated IATAs", + "name": "iatas", + "in": "query" + }, + { + "type": "integer", + "description": "Region ID", + "name": "regionId", + "in": "query" + }, + { + "type": "string", + "description": "Region slug", + "name": "region", + "in": "query" + }, + { + "type": "string", + "description": "Observer type", + "name": "type", + "in": "query" + }, + { + "type": "string", + "description": "Broker membership", + "name": "broker", + "in": "query" + }, + { + "type": "string", + "description": "online or offline", + "name": "status", + "in": "query" + }, + { + "type": "string", + "description": "Case-insensitive partial name", + "name": "name", + "in": "query" + }, + { + "type": "string", + "description": "Transport scope membership", + "name": "scope", + "in": "query" + }, + { + "type": "string", + "description": "traffic (default) or name", + "name": "sort", + "in": "query" + }, + { + "type": "integer", + "format": "int64", + "description": "Inclusive epoch ms, preserved exactly; default seven days before until; maximum window 31 days", + "name": "since", + "in": "query" + }, + { + "type": "integer", + "format": "int64", + "description": "Exclusive epoch ms, preserved exactly; default request time", + "name": "until", + "in": "query" + }, + { + "type": "integer", + "format": "int64", + "description": "nextCursor returned by preceding page", + "name": "cursor", + "in": "query" + }, + { + "type": "integer", + "description": "Page size, default 50, maximum 200", + "name": "limit", + "in": "query" + } + ], + "responses": { + "200": { + "description": "OK", + "schema": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverDirectory" + } + }, + "400": { + "description": "Bad Request", + "schema": { + "$ref": "#/definitions/internal_api_handlers.APIError" + } + }, + "500": { + "description": "Internal Server Error", + "schema": { + "$ref": "#/definitions/internal_api_handlers.APIError" + } + } + } + } + }, "/observers/{observerId}": { "get": { "produces": [ @@ -4401,6 +4522,110 @@ const docTemplate = `{ } } }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverDirectory": { + "type": "object", + "properties": { + "coverage": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverDirectoryCoverage" + }, + "effectiveSort": { + "type": "string" + }, + "generatedAt": { + "type": "integer" + }, + "hasMore": { + "type": "boolean" + }, + "items": { + "type": "array", + "items": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverDirectoryItem" + } + }, + "maxObservationCount": { + "description": "Maximum over all matching observers, not only this page; zero when none match.", + "type": "integer" + }, + "nextCursor": { + "type": "integer" + }, + "observerTypes": { + "type": "array", + "items": { + "type": "string" + } + }, + "sort": { + "type": "string" + }, + "windowEnd": { + "type": "integer" + }, + "windowStart": { + "type": "integer" + } + } + }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverDirectoryCoverage": { + "type": "object", + "properties": { + "completeHours": { + "type": "integer" + }, + "expectedHours": { + "type": "integer" + }, + "missingHours": { + "type": "integer" + }, + "partialHours": { + "type": "integer" + }, + "status": { + "type": "string" + } + } + }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverDirectoryItem": { + "type": "object", + "properties": { + "displayName": { + "description": "friendly name from /status messages", + "type": "string" + }, + "iata": { + "description": "most recently heard IATA", + "type": "string" + }, + "id": { + "type": "string" + }, + "observationCount": { + "description": "Available observations in [windowStart, windowEnd), including zero with missing history.", + "type": "integer" + }, + "observerType": { + "description": "e.g. \"meshcoretomqtt\", \"meshcoreha\"", + "type": "string" + }, + "radio": { + "description": "friendly radio param string: freqMhz,BwKhz,SF e.g. \"910.525,62.5,7\"", + "type": "string" + }, + "scopes": { + "description": "list of observer forwarded scopes matched to config", + "type": "array", + "items": { + "type": "string" + } + }, + "status": { + "description": "\"online\" or \"offline\" derived from last_status_at", + "type": "string" + } + } + }, "github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverSummary": { "type": "object", "properties": { @@ -5811,7 +6036,7 @@ const docTemplate = `{ // SwaggerInfo holds exported Swagger Info so clients can modify it var SwaggerInfo = &swag.Spec{ - Version: "2.0.1", + Version: "2.0.2", Host: "localhost:8080", BasePath: "/api/v1", Schemes: []string{"http", "https"}, diff --git a/docs/historical-stats.md b/docs/historical-stats.md index c039ad5..9aa6db7 100644 --- a/docs/historical-stats.md +++ b/docs/historical-stats.md @@ -2,10 +2,10 @@ ## The rule -Historical stats never scan `packet_observations` per request or per refresh. They read -hourly rollup tables filled once per closed hour by one background task. Raw rows are read -only for per-entity drill-downs (a packet, a trace, a route), observer comparison and -sub-hour observer activity. +Historical stats read hourly rollup tables filled once per closed hour by one background +task. Raw rows are read for per-entity drill-downs (a packet, a trace, a route), observer +comparison, sub-hour observer activity, and bounded observer-directory slices that cannot +use a completed whole-hour rollup. ## Rollup contract @@ -32,7 +32,8 @@ sub-hour observer activity. newest deleted `last_heard_at`. Cleanup and the rollup never write the same row. - **Partial hours.** A not-yet-complete hour starting at or before `raw_deleted_before + 30 min` lost raw rows before it was rolled. It is marked `partial` and never rolled; readers - report it without values. + report it without values, except the observer directory, which counts any retained raw + observations in the requested slice. - **Re-rolls.** Hours are not re-rolled periodically. A message stored by channel backfill (at startup, and when keys are imported at runtime) queues every hour its packet was heard in `analytics_dirty_hours`, in the same statement as the message insert. Each pass re-rolls a @@ -52,10 +53,43 @@ sub-hour observer activity. | Max path entries | Max | | SNR / RSSI averages | Σ sum / Σ samples, never an average of averages | -Windows snap to UTC hours on the server, and responses report the effective window. +Historical stats windows snap to UTC hours on the server, and responses report the +effective window. The observer directory preserves exact millisecond bounds instead. `/stats/overview` spans the 24 newest eligible hours; its `windowHours` counts the complete ones its totals cover. +## Observer directory + +`GET /api/v1/observers/directory` defaults to `[request time - 7 days, request time)`. +Explicit `since` and `until` are epoch milliseconds and remain exact; windows must be +positive and at most 31 days. If only `until` is supplied, `since` defaults to seven days +before it. If only `since` is supplied, `until` defaults to request time. + +Counts combine complete rollups for whole UTC hours fully inside the window with retained +raw observations for every other hour and the exact boundary slices. These sources never +overlap. Completed whole-hour history survives raw retention; a sliced boundary whose raw +rows have expired cannot be reconstructed from its whole-hour total. + +Every observer receives a numeric `observationCount`, including zero. `maxObservationCount` +is the maximum across all matching observers, independent of pagination, and is zero when +none match. Incomplete history never suppresses these counts or changes `effectiveSort`: +traffic sorting uses available counts, while explicit name sorting remains supported. +Both sorts break name ties by observer ID. + +`coverage` is conservative and informational. `expectedHours` counts intersecting UTC +hours, including boundary slices. `completeHours` counts the whole hours read from complete +rollups. Other hours are `partialHours` when there is a partial/boundary rollup or matching +retained raw traffic; otherwise they are `missingHours`. Raw-only traffic does not establish +complete history. Status is `complete` only when all intersecting hours are complete, +`partial` when any complete or partial hour exists, and `unavailable` otherwise. Zero counts +with incomplete coverage mean no available observations, not proof that no traffic occurred. + +The response uses `Cache-Control: no-store`. For continuation requests, send `nextCursor`, +the same filters and sort, and the exact returned `windowStart`/`windowEnd` as `since`/`until`. +These bounds fix the time window, not a database snapshot; late arrivals and metadata changes +can still move results. Refresh without bounds to obtain a new rolling window and include +new traffic. + ## IATA sets A packet heard in YVR and YYZ appears in both IATAs' per-IATA rows, so summing per-IATA diff --git a/docs/swagger.json b/docs/swagger.json index 9deb9fd..29cd6b8 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -15,7 +15,7 @@ "license": { "name": "AGPL-3-or-later" }, - "version": "2.0.1" + "version": "2.0.2" }, "host": "localhost:8080", "basePath": "/api/v1", @@ -1428,6 +1428,127 @@ } } }, + "/observers/directory": { + "get": { + "description": "Defaults to the trailing seven days through request time. Counts combine completed whole-hour analytics with retained observations for other hours and exact boundary slices. Incomplete history still returns numeric counts and the requested sort; coverage describes available history. Repeat filters and the returned windowStart/windowEnd as since/until with the cursor on later pages. Results may move as data changes; refresh without bounds for a new rolling window.", + "produces": [ + "application/json" + ], + "tags": [ + "Observers" + ], + "summary": "List observers with windowed traffic counts", + "parameters": [ + { + "type": "string", + "description": "Single IATA", + "name": "iata", + "in": "query" + }, + { + "type": "string", + "description": "Comma-separated IATAs", + "name": "iatas", + "in": "query" + }, + { + "type": "integer", + "description": "Region ID", + "name": "regionId", + "in": "query" + }, + { + "type": "string", + "description": "Region slug", + "name": "region", + "in": "query" + }, + { + "type": "string", + "description": "Observer type", + "name": "type", + "in": "query" + }, + { + "type": "string", + "description": "Broker membership", + "name": "broker", + "in": "query" + }, + { + "type": "string", + "description": "online or offline", + "name": "status", + "in": "query" + }, + { + "type": "string", + "description": "Case-insensitive partial name", + "name": "name", + "in": "query" + }, + { + "type": "string", + "description": "Transport scope membership", + "name": "scope", + "in": "query" + }, + { + "type": "string", + "description": "traffic (default) or name", + "name": "sort", + "in": "query" + }, + { + "type": "integer", + "format": "int64", + "description": "Inclusive epoch ms, preserved exactly; default seven days before until; maximum window 31 days", + "name": "since", + "in": "query" + }, + { + "type": "integer", + "format": "int64", + "description": "Exclusive epoch ms, preserved exactly; default request time", + "name": "until", + "in": "query" + }, + { + "type": "integer", + "format": "int64", + "description": "nextCursor returned by preceding page", + "name": "cursor", + "in": "query" + }, + { + "type": "integer", + "description": "Page size, default 50, maximum 200", + "name": "limit", + "in": "query" + } + ], + "responses": { + "200": { + "description": "OK", + "schema": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverDirectory" + } + }, + "400": { + "description": "Bad Request", + "schema": { + "$ref": "#/definitions/internal_api_handlers.APIError" + } + }, + "500": { + "description": "Internal Server Error", + "schema": { + "$ref": "#/definitions/internal_api_handlers.APIError" + } + } + } + } + }, "/observers/{observerId}": { "get": { "produces": [ @@ -4399,6 +4520,110 @@ } } }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverDirectory": { + "type": "object", + "properties": { + "coverage": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverDirectoryCoverage" + }, + "effectiveSort": { + "type": "string" + }, + "generatedAt": { + "type": "integer" + }, + "hasMore": { + "type": "boolean" + }, + "items": { + "type": "array", + "items": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverDirectoryItem" + } + }, + "maxObservationCount": { + "description": "Maximum over all matching observers, not only this page; zero when none match.", + "type": "integer" + }, + "nextCursor": { + "type": "integer" + }, + "observerTypes": { + "type": "array", + "items": { + "type": "string" + } + }, + "sort": { + "type": "string" + }, + "windowEnd": { + "type": "integer" + }, + "windowStart": { + "type": "integer" + } + } + }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverDirectoryCoverage": { + "type": "object", + "properties": { + "completeHours": { + "type": "integer" + }, + "expectedHours": { + "type": "integer" + }, + "missingHours": { + "type": "integer" + }, + "partialHours": { + "type": "integer" + }, + "status": { + "type": "string" + } + } + }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverDirectoryItem": { + "type": "object", + "properties": { + "displayName": { + "description": "friendly name from /status messages", + "type": "string" + }, + "iata": { + "description": "most recently heard IATA", + "type": "string" + }, + "id": { + "type": "string" + }, + "observationCount": { + "description": "Available observations in [windowStart, windowEnd), including zero with missing history.", + "type": "integer" + }, + "observerType": { + "description": "e.g. \"meshcoretomqtt\", \"meshcoreha\"", + "type": "string" + }, + "radio": { + "description": "friendly radio param string: freqMhz,BwKhz,SF e.g. \"910.525,62.5,7\"", + "type": "string" + }, + "scopes": { + "description": "list of observer forwarded scopes matched to config", + "type": "array", + "items": { + "type": "string" + } + }, + "status": { + "description": "\"online\" or \"offline\" derived from last_status_at", + "type": "string" + } + } + }, "github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverSummary": { "type": "object", "properties": { diff --git a/docs/swagger.yaml b/docs/swagger.yaml index 9844449..c7b70d6 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -687,6 +687,79 @@ definitions: description: exclusive, epoch milliseconds type: integer type: object + github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverDirectory: + properties: + coverage: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverDirectoryCoverage' + effectiveSort: + type: string + generatedAt: + type: integer + hasMore: + type: boolean + items: + items: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverDirectoryItem' + type: array + maxObservationCount: + description: Maximum over all matching observers, not only this page; zero + when none match. + type: integer + nextCursor: + type: integer + observerTypes: + items: + type: string + type: array + sort: + type: string + windowEnd: + type: integer + windowStart: + type: integer + type: object + github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverDirectoryCoverage: + properties: + completeHours: + type: integer + expectedHours: + type: integer + missingHours: + type: integer + partialHours: + type: integer + status: + type: string + type: object + github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverDirectoryItem: + properties: + displayName: + description: friendly name from /status messages + type: string + iata: + description: most recently heard IATA + type: string + id: + type: string + observationCount: + description: Available observations in [windowStart, windowEnd), including + zero with missing history. + type: integer + observerType: + description: e.g. "meshcoretomqtt", "meshcoreha" + type: string + radio: + description: 'friendly radio param string: freqMhz,BwKhz,SF e.g. "910.525,62.5,7"' + type: string + scopes: + description: list of observer forwarded scopes matched to config + items: + type: string + type: array + status: + description: '"online" or "offline" derived from last_status_at' + type: string + type: object github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverSummary: properties: displayName: @@ -1635,7 +1708,7 @@ info: name: AGPL-3-or-later termsOfService: https://github.com/MeshCore-Beacon/beacon-server title: MeshCore Beacon API - version: 2.0.1 + version: 2.0.2 paths: /admin/accounts: get: @@ -2739,6 +2812,94 @@ paths: summary: Get observer telemetry history tags: - Observers + /observers/directory: + get: + description: Defaults to the trailing seven days through request time. Counts + combine completed whole-hour analytics with retained observations for other + hours and exact boundary slices. Incomplete history still returns numeric + counts and the requested sort; coverage describes available history. Repeat + filters and the returned windowStart/windowEnd as since/until with the cursor + on later pages. Results may move as data changes; refresh without bounds for + a new rolling window. + parameters: + - description: Single IATA + in: query + name: iata + type: string + - description: Comma-separated IATAs + in: query + name: iatas + type: string + - description: Region ID + in: query + name: regionId + type: integer + - description: Region slug + in: query + name: region + type: string + - description: Observer type + in: query + name: type + type: string + - description: Broker membership + in: query + name: broker + type: string + - description: online or offline + in: query + name: status + type: string + - description: Case-insensitive partial name + in: query + name: name + type: string + - description: Transport scope membership + in: query + name: scope + type: string + - description: traffic (default) or name + in: query + name: sort + type: string + - description: Inclusive epoch ms, preserved exactly; default seven days before + until; maximum window 31 days + format: int64 + in: query + name: since + type: integer + - description: Exclusive epoch ms, preserved exactly; default request time + format: int64 + in: query + name: until + type: integer + - description: nextCursor returned by preceding page + format: int64 + in: query + name: cursor + type: integer + - description: Page size, default 50, maximum 200 + in: query + name: limit + type: integer + produces: + - application/json + responses: + "200": + description: OK + schema: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ObserverDirectory' + "400": + description: Bad Request + schema: + $ref: '#/definitions/internal_api_handlers.APIError' + "500": + description: Internal Server Error + schema: + $ref: '#/definitions/internal_api_handlers.APIError' + summary: List observers with windowed traffic counts + tags: + - Observers /packets: get: parameters: diff --git a/env.example b/env.example index c7a152e..d3fb5ca 100644 --- a/env.example +++ b/env.example @@ -10,6 +10,12 @@ LISTEN_ADDR=:8080 # If set, overrides auth.api_key (even when empty). # BEACON_API_KEY= +# Regional or grouped-region API key generated by your local MeshMapper admin. +# A grouped-region key covers multiple regions. Never use the mobile App key. +# Overrides meshmapper.api_key, even when empty. Required for enabled imports. +# Keep this in the deployment environment when using saved-config backups. +# MESHMAPPER_API_KEY= + # Optional application logging overrides (defaults: info / text). # LOG_LEVEL=info # LOG_FORMAT=json diff --git a/internal/api/handlers/observer_directory.go b/internal/api/handlers/observer_directory.go new file mode 100644 index 0000000..64549e1 --- /dev/null +++ b/internal/api/handlers/observer_directory.go @@ -0,0 +1,140 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package handlers + +import ( + "net/http" + "slices" + "strconv" + "strings" + "time" + + "github.com/MeshCore-Beacon/beacon-server/internal/api" +) + +// listObserverDirectory godoc +// +// @Summary List observers with windowed traffic counts +// @Description Defaults to the trailing seven days through request time. Counts combine completed whole-hour analytics with retained observations for other hours and exact boundary slices. Incomplete history still returns numeric counts and the requested sort; coverage describes available history. Repeat filters and the returned windowStart/windowEnd as since/until with the cursor on later pages. Results may move as data changes; refresh without bounds for a new rolling window. +// @Tags Observers +// @Produce json +// @Param iata query string false "Single IATA" +// @Param iatas query string false "Comma-separated IATAs" +// @Param regionId query int false "Region ID" +// @Param region query string false "Region slug" +// @Param type query string false "Observer type" +// @Param broker query string false "Broker membership" +// @Param status query string false "online or offline" +// @Param name query string false "Case-insensitive partial name" +// @Param scope query string false "Transport scope membership" +// @Param sort query string false "traffic (default) or name" +// @Param since query int64 false "Inclusive epoch ms, preserved exactly; default seven days before until; maximum window 31 days" +// @Param until query int64 false "Exclusive epoch ms, preserved exactly; default request time" +// @Param cursor query int64 false "nextCursor returned by preceding page" +// @Param limit query int false "Page size, default 50, maximum 200" +// @Success 200 {object} api.ObserverDirectory +// @Failure 400 {object} handlers.APIError +// @Failure 500 {object} handlers.APIError +// @Router /observers/directory [get] +func listObserverDirectory(reader api.Reader) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + values := r.URL.Query() + allowed := map[string]bool{"iata": true, "iatas": true, "regionId": true, "region": true, "type": true, "broker": true, "status": true, "name": true, "scope": true, "sort": true, "since": true, "until": true, "cursor": true, "limit": true} + for k, v := range values { + if !allowed[k] || len(v) != 1 { + respondError(w, 400, "unknown or repeated directory parameter") + return + } + } + q := api.ObserverDirectoryQuery{} + limit, err := parseLimit(r, 50) + if err != nil { + respondError(w, 400, err.Error()) + return + } + q.Limit = limit + if values.Has("cursor") { + q.Cursor, err = strconv.ParseInt(values.Get("cursor"), 10, 32) + if err != nil || q.Cursor < 0 || q.Cursor > 2147483447 { + respondError(w, 400, "invalid directory cursor") + return + } + } + if q.Cursor > 0 && (!values.Has("since") || !values.Has("until")) { + respondError(w, 400, "continuation requires since and until from the first page") + return + } + q.Sort = values.Get("sort") + if q.Sort == "" { + q.Sort = "traffic" + } + if q.Sort != "traffic" && q.Sort != "name" { + respondError(w, 400, "sort must be traffic or name") + return + } + q.Type = values.Get("type") + q.Broker = values.Get("broker") + q.Status = values.Get("status") + q.Name = strings.ToLower(values.Get("name")) + q.Scope = values.Get("scope") + if q.Status != "" && q.Status != "online" && q.Status != "offline" { + respondError(w, 400, "status must be online or offline") + return + } + for _, v := range []string{q.Type, q.Broker, q.Name, q.Scope} { + if len(v) > 200 { + respondError(w, 400, "directory filters must be at most 200 bytes") + return + } + } + now := time.Now().UTC().Truncate(time.Millisecond) + until := now + if values.Has("until") { + n, e := strconv.ParseInt(values.Get("until"), 10, 64) + if e != nil || n <= 0 || n > now.UnixMilli() { + respondError(w, 400, "until must be a past epoch millisecond timestamp") + return + } + until = time.UnixMilli(n).UTC() + } + since := until.Add(-7 * 24 * time.Hour) + if values.Has("since") { + n, e := strconv.ParseInt(values.Get("since"), 10, 64) + if e != nil || n <= 0 { + respondError(w, 400, "since must be positive epoch milliseconds") + return + } + since = time.UnixMilli(n).UTC() + } + if !since.Before(until) || until.Sub(since) > 31*24*time.Hour { + respondError(w, 400, "directory window must be positive and at most 31 days") + return + } + q.Since = since.UnixMilli() + q.Until = until.UnixMilli() + q.IATAs = parseIATAs(r) + if values.Get("regionId") != "" || values.Get("region") != "" { + iatas, err := resolveRegionIATAs(r.Context(), values.Get("regionId"), values.Get("region"), reader) + if err != nil { + respondRegionError(w, err) + return + } + q.IATAs = append(q.IATAs, iatas...) + q.MatchNone = len(q.IATAs) == 0 + } + if len(q.IATAs) > 200 { + respondError(w, 400, "too many IATA filters") + return + } + slices.Sort(q.IATAs) + q.IATAs = slices.Compact(q.IATAs) + page, err := reader.ListObserverDirectory(r.Context(), q) + w.Header().Set("Cache-Control", "no-store") + if err != nil { + respondError(w, 500, "failed to get observer directory") + return + } + respond(w, 200, page) + } +} diff --git a/internal/api/handlers/observer_directory_test.go b/internal/api/handlers/observer_directory_test.go new file mode 100644 index 0000000..74ca877 --- /dev/null +++ b/internal/api/handlers/observer_directory_test.go @@ -0,0 +1,165 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package handlers + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http/httptest" + "slices" + "testing" + "time" + + "github.com/MeshCore-Beacon/beacon-server/internal/api" +) + +func TestDirectoryRejectsInvalidPagination(t *testing.T) { + for _, q := range []string{"sort=bogus", "cursor=1", "cursor=-1", "limit=0", "snapshot=bad", "until=no", "since=0", "unknown=1", "snapshot=00000000-0000-0000-0000-000000000001&sort=name", "status=maybe", "sort=name&sort=traffic", "cursor=2147483448", "since=1767229200000&until=1767225600000", "since=1764543600000&until=1767229200000", "until=9223372036854775807"} { + t.Run(q, func(t *testing.T) { + r := stubReader{listObserverDirectory: func(context.Context, api.ObserverDirectoryQuery) (*api.ObserverDirectory, error) { + t.Fatal("invalid query reached store") + return nil, nil + }} + w := httptest.NewRecorder() + ObserversRouter(r).ServeHTTP(w, httptest.NewRequest("GET", "/directory?"+q, nil)) + if w.Code != 400 { + t.Fatalf("status %d: %s", w.Code, w.Body.String()) + } + }) + } +} + +func TestDirectoryContinuationAndErrors(t *testing.T) { + for _, tc := range []struct { + err error + code int + }{{nil, 200}, {errors.New("database unavailable"), 500}} { + r := stubReader{listObserverDirectory: func(_ context.Context, q api.ObserverDirectoryQuery) (*api.ObserverDirectory, error) { + if q.Sort != "name" || q.Name != "alpha" || q.Since != 1767225600123 || q.Until != 1767229200789 || q.Cursor != 50 || q.Limit != 25 { + t.Fatalf("query %+v", q) + } + return &api.ObserverDirectory{}, tc.err + }} + w := httptest.NewRecorder() + ObserversRouter(r).ServeHTTP(w, httptest.NewRequest("GET", "/directory?sort=name&name=alpha&since=1767225600123&until=1767229200789&cursor=50&limit=25", nil)) + if w.Code != tc.code { + t.Fatalf("status %d: %s", w.Code, w.Body.String()) + } + } +} + +func TestDirectoryNormalizesFirstPage(t *testing.T) { + r := stubReader{listObserverDirectory: func(_ context.Context, q api.ObserverDirectoryQuery) (*api.ObserverDirectory, error) { + if q.Sort != "traffic" || q.Limit != 200 || q.Name != "alpha" || q.Since != 1767225600999 || q.Until != 1767229200999 || len(q.IATAs) != 2 || q.IATAs[0] != "YVR" || q.IATAs[1] != "YYJ" { + t.Fatalf("query %+v", q) + } + return &api.ObserverDirectory{}, nil + }} + w := httptest.NewRecorder() + ObserversRouter(r).ServeHTTP(w, httptest.NewRequest("GET", "/directory?iatas=yyj,yvr,YVR&name=ALPHA&since=1767225600999&until=1767229200999&limit=201", nil)) + if w.Code != 200 || w.Header().Get("Cache-Control") != "no-store" { + t.Fatalf("response %d %s", w.Code, w.Body.String()) + } +} + +func TestDirectoryRollingWeekAndContinuation(t *testing.T) { + var queries []api.ObserverDirectoryQuery + r := stubReader{listObserverDirectory: func(_ context.Context, q api.ObserverDirectoryQuery) (*api.ObserverDirectory, error) { + queries = append(queries, q) + return &api.ObserverDirectory{WindowStart: q.Since, WindowEnd: q.Until}, nil + }} + router := ObserversRouter(r) + before := time.Now().UnixMilli() + w := httptest.NewRecorder() + router.ServeHTTP(w, httptest.NewRequest("GET", "/directory?iatas=YVR,YYJ&limit=1", nil)) + after := time.Now().UnixMilli() + if w.Code != 200 || len(queries) != 1 { + t.Fatalf("response %d: %s", w.Code, w.Body.String()) + } + q := queries[0] + if q.Until < before || q.Until > after || q.Until-q.Since != (7*24*time.Hour).Milliseconds() { + t.Fatalf("window [%d,%d) does not span seven days through the request time", q.Since, q.Until) + } + var page api.ObserverDirectory + if err := json.Unmarshal(w.Body.Bytes(), &page); err != nil { + t.Fatal(err) + } + w = httptest.NewRecorder() + router.ServeHTTP(w, httptest.NewRequest("GET", fmt.Sprintf("/directory?iatas=YVR,YYJ&limit=1&cursor=1&since=%d&until=%d", page.WindowStart, page.WindowEnd), nil)) + if w.Code != 200 || len(queries) != 2 { + t.Fatalf("continuation %d: %s", w.Code, w.Body.String()) + } + if next := queries[1]; next.Since != q.Since || next.Until != q.Until || next.Sort != "traffic" || !slices.Equal(next.IATAs, q.IATAs) { + t.Fatalf("continuation changed the window or filters: %+v", next) + } +} + +func TestDirectoryExactWindowValidation(t *testing.T) { + const until int64 = 1767229200123 + for _, tc := range []struct { + name, query string + code int + since int64 + }{ + {"until only defaults to seven days", fmt.Sprintf("until=%d", until), 200, until - 604800000}, + {"millisecond window", fmt.Sprintf("since=%d&until=%d", until-1, until), 200, until - 1}, + {"31 days", fmt.Sprintf("since=%d&until=%d", until-2678400000, until), 200, until - 2678400000}, + {"over 31 days", fmt.Sprintf("since=%d&until=%d", until-2678400001, until), 400, 0}, + {"empty window", fmt.Sprintf("since=%d&until=%d", until, until), 400, 0}, + {"repeated bound", fmt.Sprintf("since=1&since=2&until=%d", until), 400, 0}, + } { + t.Run(tc.name, func(t *testing.T) { + r := stubReader{listObserverDirectory: func(_ context.Context, q api.ObserverDirectoryQuery) (*api.ObserverDirectory, error) { + if tc.code != 200 || q.Since != tc.since || q.Until != until { + t.Fatalf("unexpected query: %+v", q) + } + return &api.ObserverDirectory{}, nil + }} + w := httptest.NewRecorder() + ObserversRouter(r).ServeHTTP(w, httptest.NewRequest("GET", "/directory?"+tc.query, nil)) + if w.Code != tc.code { + t.Fatalf("response %d: %s", w.Code, w.Body.String()) + } + }) + } +} + +func TestDirectoryRegionFilters(t *testing.T) { + for _, tc := range []struct { + query string + iatas []string + none bool + }{ + {"region=west", []string{"YVR", "YYJ"}, false}, + {"regionId=7&iata=YOW", []string{"YOW", "YVR", "YYJ"}, false}, + {"region=empty", nil, true}, + } { + t.Run(tc.query, func(t *testing.T) { + r := stubReader{ + getRegion: func(context.Context, int32) (*api.Region, error) { + return &api.Region{IATAs: []string{"YYJ", "YVR"}}, nil + }, + getRegionBySlug: func(_ context.Context, slug string) (*api.Region, error) { + if slug == "empty" { + return &api.Region{}, nil + } + return &api.Region{IATAs: []string{"YYJ", "YVR"}}, nil + }, + listObserverDirectory: func(_ context.Context, q api.ObserverDirectoryQuery) (*api.ObserverDirectory, error) { + if !slices.Equal(q.IATAs, tc.iatas) || q.MatchNone != tc.none { + t.Fatalf("query %+v", q) + } + return &api.ObserverDirectory{}, nil + }, + } + w := httptest.NewRecorder() + ObserversRouter(r).ServeHTTP(w, httptest.NewRequest("GET", "/directory?"+tc.query, nil)) + if w.Code != 200 { + t.Fatalf("response %d: %s", w.Code, w.Body.String()) + } + }) + } +} diff --git a/internal/api/handlers/observers.go b/internal/api/handlers/observers.go index 089c6e4..b3b0107 100644 --- a/internal/api/handlers/observers.go +++ b/internal/api/handlers/observers.go @@ -25,6 +25,7 @@ import ( func ObserversRouter(reader api.Reader) http.Handler { r := chi.NewRouter() r.Get("/", listObservers(reader)) + r.Get("/directory", listObserverDirectory(reader)) r.Route("/{observerId}", func(r chi.Router) { r.Get("/", getObserver(reader)) r.Get("/adverts", listObserverAdverts(reader)) diff --git a/internal/api/handlers/stub_reader_test.go b/internal/api/handlers/stub_reader_test.go index efb5cb8..941937e 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 { + listObserverDirectory func(context.Context, api.ObserverDirectoryQuery) (*api.ObserverDirectory, error) getStatsSeries func(context.Context, time.Time, time.Time, []string) (*api.StatsSeries, error) 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) @@ -417,3 +418,10 @@ func (s stubReader) GetRouteEvidence(ctx context.Context, iata, key string, q ap } return nil, nil } + +func (s stubReader) ListObserverDirectory(ctx context.Context, q api.ObserverDirectoryQuery) (*api.ObserverDirectory, error) { + if s.listObserverDirectory != nil { + return s.listObserverDirectory(ctx, q) + } + return nil, nil +} diff --git a/internal/api/observer_directory.go b/internal/api/observer_directory.go new file mode 100644 index 0000000..511d3f1 --- /dev/null +++ b/internal/api/observer_directory.go @@ -0,0 +1,48 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package api + +type ObserverDirectoryQuery struct { + MatchNone bool + IATAs []string + Type string + Broker string + Status string + Name string + Scope string + Sort string + Since int64 + Until int64 + Cursor int64 + Limit int32 +} + +type ObserverDirectoryItem struct { + ObserverSummary + // Available observations in [windowStart, windowEnd), including zero with missing history. + ObservationCount *int64 `json:"observationCount"` +} + +// Coverage counts intersecting UTC hours. Complete hours use whole-hour rollups; +// partial hours have partial/boundary rollups or available raw observations. +type ObserverDirectoryCoverage struct { + Status string `json:"status"` + ExpectedHours int `json:"expectedHours"` + CompleteHours int `json:"completeHours"` + PartialHours int `json:"partialHours"` + MissingHours int `json:"missingHours"` +} + +type ObserverDirectory struct { + Page[ObserverDirectoryItem] + GeneratedAt int64 `json:"generatedAt"` + WindowStart int64 `json:"windowStart"` + WindowEnd int64 `json:"windowEnd"` + Sort string `json:"sort"` + EffectiveSort string `json:"effectiveSort"` + Coverage ObserverDirectoryCoverage `json:"coverage"` + // Maximum over all matching observers, not only this page; zero when none match. + MaxObservationCount *int64 `json:"maxObservationCount"` + ObserverTypes []string `json:"observerTypes"` +} diff --git a/internal/api/reader.go b/internal/api/reader.go index dc71e63..15b8693 100644 --- a/internal/api/reader.go +++ b/internal/api/reader.go @@ -23,6 +23,7 @@ type Page[T any] struct { } type Reader interface { + ListObserverDirectory(context.Context, ObserverDirectoryQuery) (*ObserverDirectory, error) // 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. diff --git a/internal/api/router/admin_config_test.go b/internal/api/router/admin_config_test.go index 9acd524..4ae50d8 100644 --- a/internal/api/router/admin_config_test.go +++ b/internal/api/router/admin_config_test.go @@ -17,7 +17,7 @@ import ( func TestAdminConfig(t *testing.T) { const key = "admin-key-sentinel" - for _, variable := range []string{"POSTGRES_DSN", "MQTT_BROKER_1_URL", "MQTT_BROKER_1_USERNAME", "MQTT_BROKER_1_PASSWORD", "REDIS_PASSWORD"} { + for _, variable := range []string{"POSTGRES_DSN", "MQTT_BROKER_1_URL", "MQTT_BROKER_1_USERNAME", "MQTT_BROKER_1_PASSWORD", "REDIS_PASSWORD", "MESHMAPPER_API_KEY"} { t.Setenv(variable, "environment-secret-sentinel") } for _, custom := range []bool{false, true} { diff --git a/internal/backup/export.go b/internal/backup/export.go index e7b1a56..456a5d6 100644 --- a/internal/backup/export.go +++ b/internal/backup/export.go @@ -19,6 +19,8 @@ import ( "path/filepath" "strings" "time" + + "gopkg.in/yaml.v3" ) const ( @@ -229,6 +231,18 @@ func readConfig(path string) ([]byte, error) { if err != nil || len(data) > maxConfigBytes { return nil, errors.New("cannot read saved config within the 1 MiB limit") } + var credentials struct { + MeshMapper struct { + APIKey string `yaml:"api_key"` + } `yaml:"meshmapper"` + } + if err := yaml.Unmarshal(data, &credentials); err != nil { + return nil, errors.New("cannot inspect saved config for MeshMapper credentials") + } + // Backups preserve YAML verbatim, so require the deployment secret outside it. + if credentials.MeshMapper.APIKey != "" { + return nil, errors.New("backup unavailable: move meshmapper.api_key to MESHMAPPER_API_KEY and clear the saved YAML value") + } return data, nil } diff --git a/internal/backup/export_test.go b/internal/backup/export_test.go index a3481af..4dba97f 100644 --- a/internal/backup/export_test.go +++ b/internal/backup/export_test.go @@ -24,6 +24,37 @@ import ( const testSQL = "CREATE TABLE fixture (id bigint PRIMARY KEY, name text);\nINSERT INTO fixture VALUES (1, 'café');\n" +func TestBackupDoesNotExportMeshMapperAPIKey(t *testing.T) { + for _, saved := range []string{ + "meshmapper:\n api_key: backup-secret-canary\n", + "shared: &integration {api_key: backup-secret-canary}\nmeshmapper: *integration\n", + "meshmapper:\n <<: {api_key: backup-secret-canary}\n", + "meshmapper:\n api_key: [backup-secret-canary]\n", + } { + path := filepath.Join(t.TempDir(), "config.yaml") + if err := os.WriteFile(path, []byte(saved), 0600); err != nil { + t.Fatal(err) + } + data, err := readConfig(path) + if err == nil || data != nil { + t.Fatal("saved MeshMapper key could enter a backup response") + } + if strings.Contains(err.Error(), "backup-secret-canary") { + t.Fatal("backup error exposed the key") + } + } + t.Setenv("MESHMAPPER_API_KEY", "environment-secret-canary") + saved := []byte("# Saved config remains verbatim.\nmeshmapper:\n api_key: ''\n") + path := filepath.Join(t.TempDir(), "config.yaml") + if err := os.WriteFile(path, saved, 0600); err != nil { + t.Fatal(err) + } + data, err := readConfig(path) + if err != nil || !bytes.Equal(data, saved) { + t.Fatal("environment key changed the saved-config export") + } +} + // TestDumpProcess supplies an actual subprocess, including failure and cancellation. func TestDumpProcess(t *testing.T) { mode := os.Getenv("BEACON_BACKUP_TEST_PROCESS") diff --git a/internal/cache/cache_test.go b/internal/cache/cache_test.go index b5dcc7f..e00b124 100644 --- a/internal/cache/cache_test.go +++ b/internal/cache/cache_test.go @@ -446,3 +446,7 @@ func TestCachedReader_InvalidateObserver(t *testing.T) { func (s *stubReader) GetRouteEvidence(_ context.Context, _, _ string, _ api.RouteEvidenceQuery) (*api.RouteEvidence, error) { return nil, nil } + +func (s *stubReader) ListObserverDirectory(context.Context, api.ObserverDirectoryQuery) (*api.ObserverDirectory, error) { + return nil, nil +} diff --git a/internal/cache/reader.go b/internal/cache/reader.go index e491152..e032daf 100644 --- a/internal/cache/reader.go +++ b/internal/cache/reader.go @@ -378,3 +378,7 @@ func (cr *CachedReader) SearchCrossIATARoutes(ctx context.Context, fromHash, fro func (cr *CachedReader) ListTraceTags(ctx context.Context, iatas []string, scope, traceType string, since, until time.Time, cursor time.Time, cursorTag string, limit int32) ([]api.TraceTagSummary, error) { return cr.inner.ListTraceTags(ctx, iatas, scope, traceType, since, until, cursor, cursorTag, limit) } + +func (cr *CachedReader) ListObserverDirectory(ctx context.Context, query api.ObserverDirectoryQuery) (*api.ObserverDirectory, error) { + return cr.inner.ListObserverDirectory(ctx, query) +} diff --git a/internal/config/config.go b/internal/config/config.go index 94b109e..0b29211 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -393,6 +393,7 @@ func Load(path string) (*Config, error) { data, err := os.ReadFile(path) if err != nil { if os.IsNotExist(err) { + cfg.MeshMapper.APIKey = strings.TrimSpace(os.Getenv("MESHMAPPER_API_KEY")) cfg.Auth.APIKey = os.Getenv("BEACON_API_KEY") if err := cfg.validateAPIKey(); err != nil { return nil, err @@ -407,6 +408,10 @@ func Load(path string) (*Config, error) { if value, set := os.LookupEnv("BEACON_API_KEY"); set { cfg.Auth.APIKey = value } + if value, set := os.LookupEnv("MESHMAPPER_API_KEY"); set { + cfg.MeshMapper.APIKey = value + } + cfg.MeshMapper.APIKey = strings.TrimSpace(cfg.MeshMapper.APIKey) if err := cfg.validateAPIKey(); err != nil { return nil, err } diff --git a/internal/config/meshmapper.go b/internal/config/meshmapper.go index eacf32e..134b9f6 100644 --- a/internal/config/meshmapper.go +++ b/internal/config/meshmapper.go @@ -18,6 +18,7 @@ const ( ) type MeshMapperConfig struct { + APIKey string `yaml:"api_key" json:"-"` Scopes MeshMapperScopesConfig `yaml:"scopes"` Zones MeshMapperZonesConfig `yaml:"zones"` Channels MeshMapperChannelsConfig `yaml:"channels"` diff --git a/internal/config/meshmapper_test.go b/internal/config/meshmapper_test.go index 181db8f..15dcce7 100644 --- a/internal/config/meshmapper_test.go +++ b/internal/config/meshmapper_test.go @@ -4,6 +4,7 @@ package config import ( + "encoding/json" "os" "path/filepath" "strings" @@ -11,6 +12,49 @@ import ( "time" ) +func TestLoadMeshMapperAPIKey(t *testing.T) { + for _, tc := range []struct { + name, env, want string + file, set bool + }{ + {"missing", "", "", false, false}, + {"YAML", "", "yaml-integration-test-key", true, false}, + {"environment overrides", "env-integration-test-key", "env-integration-test-key", true, true}, + {"empty environment overrides", "", "", true, true}, + {"environment without file", "env-integration-test-key", "env-integration-test-key", false, true}, + {"trim whitespace", " env-integration-test-key\n", "env-integration-test-key", true, true}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Setenv("MESHMAPPER_API_KEY", tc.env) + if !tc.set { + if err := os.Unsetenv("MESHMAPPER_API_KEY"); err != nil { + t.Fatal(err) + } + } + path := filepath.Join(t.TempDir(), "config.yaml") + if tc.file { + if err := os.WriteFile(path, []byte("meshmapper:\n api_key: yaml-integration-test-key\n scopes:\n enabled: true\n"), 0600); err != nil { + t.Fatal(err) + } + } + cfg, err := Load(path) + if err != nil { + t.Fatal(err) + } + if cfg.MeshMapper.APIKey != tc.want { + t.Fatal("MeshMapper key not loaded with expected precedence") + } + encoded, err := json.Marshal(cfg) + if err != nil { + t.Fatal(err) + } + if strings.Contains(string(encoded), "integration-test-key") || strings.Contains(Resolve(cfg).String(), "integration-test-key") { + t.Fatal("MeshMapper key exposed in JSON or startup summary") + } + }) + } +} + func TestMeshMapperConfig(t *testing.T) { valid := "meshmapper:\n scopes:\n enabled: true\n" for _, tc := range []struct { diff --git a/internal/meshmapper/auth_test.go b/internal/meshmapper/auth_test.go new file mode 100644 index 0000000..fe08754 --- /dev/null +++ b/internal/meshmapper/auth_test.go @@ -0,0 +1,427 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package meshmapper + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "log/slog" + "net/http" + "net/http/httptest" + "net/url" + "slices" + "strings" + "testing" + "time" + + "github.com/MeshCore-Beacon/beacon-server/internal/config" + "github.com/MeshCore-Beacon/beacon-server/internal/keystore" + "github.com/MeshCore-Beacon/beacon-server/internal/scopestore" +) + +func TestDirectoryMissingKeyDoesNotRequest(t *testing.T) { + calls := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls++ + _, _ = w.Write([]byte(`{"country":"CA","zones":[]}`)) + })) + defer server.Close() + store := newZoneListMemory() + dir := NewDirectory(store, "") + dir.listURL = server.URL + "/get_zones.php" + now := time.Now().UTC() + if _, _, err := dir.List(context.Background(), "CA", now); err != nil { + t.Fatal(err) + } + if calls != 0 { + t.Fatal("missing key sent an anonymous request") + } + if !strings.Contains(store.rows["CA"].LastError, "unconfigured") { + t.Fatal("missing key not reported as unconfigured") + } + if _, fetched, err := dir.List(context.Background(), "CA", now.Add(PollInterval)); fetched || err != nil { + t.Fatal("missing key retried immediately") + } +} + +func TestAuthenticationStatusAndRetryAfter(t *testing.T) { + now := time.Date(2026, 10, 4, 12, 0, 0, 0, time.UTC) + for _, tc := range []struct { + status int + want string + }{ + {401, "authentication"}, {403, "permission"}, + } { + var retry time.Time + if got := statusProblem(tc.status, http.Header{}, now, &retry); !strings.Contains(got, tc.want) { + t.Errorf("status %d not identified: %s", tc.status, got) + } + } + for _, status := range []int{429, 503} { + var retry time.Time + deadline := now.Add(48 * time.Hour) + h := http.Header{"Retry-After": []string{deadline.Format(http.TimeFormat)}} + statusProblem(status, h, now, &retry) + if !retry.Equal(deadline) { + t.Errorf("status %d ignored HTTP-date Retry-After", status) + } + } +} + +// Parser and cache fixtures use localhost; authenticated requests are tested separately. +func newTestDirectory(store DirectoryStore) *Directory { + d := NewDirectory(store, "") + d.client.Transport = http.DefaultTransport + return d +} + +type roundTripFunc func(*http.Request) (*http.Response, error) + +func (f roundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } + +// Only the socket destination changes; the production credential checks see the real URL. +func routeClient(t *testing.T, client *http.Client, server string) { + t.Helper() + u, err := url.Parse(server) + if err != nil { + t.Fatal(err) + } + client.Transport.(*authenticatedTransport).base = roundTripFunc(func(r *http.Request) (*http.Response, error) { + out := r.Clone(r.Context()) + out.URL.Scheme, out.URL.Host = u.Scheme, u.Host + return http.DefaultTransport.RoundTrip(out) + }) +} + +func TestAuthenticatedClientEndpointsAndRedirects(t *testing.T) { + const key = "operator-integration-test-secret" + var paths []string + redirect := "" + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + paths = append(paths, r.URL.Path) + if r.Header.Get("X-API-Key") != key { + t.Error("request missing configured key") + } + if r.URL.Query().Get("key") != "" { + t.Error("key sent in query") + } + if r.Header.Get("If-None-Match") != `"saved"` { + t.Error("conditional header lost") + } + if r.URL.Path == "/get_zones.php" && r.URL.RawQuery != "country=CA" { + t.Error("country parameter changed") + } + if redirect != "" { + http.Redirect(w, r, redirect, http.StatusFound) + return + } + w.WriteHeader(http.StatusNotModified) + })) + defer server.Close() + client := newClient(key) + routeClient(t, client, server.URL) + for _, path := range []string{"get_zones.php?country=CA", "get_geojson.php", "get_scopes.php", "get_channels.php", "get_repeaters.php"} { + host := "yow.meshmapper.net" + if strings.HasPrefix(path, "get_zones") { + host = "meshmapper.net" + } + status, _, _, err := get(context.Background(), client, "Beacon-Test", "https://"+host+"/"+path, `"saved"`, MaxBody) + if err != nil || status != 304 { + t.Fatalf("%s: status=%d, error=%v", path, status, err) + } + } + if len(paths) != 5 { + t.Fatal("not all five endpoints requested") + } + for _, destination := range []string{"https://unrelated.example/get_scopes.php", "https://other.meshmapper.net/get_scopes.php", "http://yow.meshmapper.net/get_scopes.php", "/get_scopes.php"} { + redirect = destination + before := len(paths) + status, _, _, err := get(context.Background(), client, "Beacon-Test", "https://yow.meshmapper.net/get_scopes.php", `"saved"`, MaxBody) + if err != nil || status != 302 || len(paths) != before+1 { + t.Fatal("redirect followed") + } + } +} + +func TestAuthenticatedClientRejectsUntrustedTargets(t *testing.T) { + client := newClient("operator-integration-test-secret") + calls := 0 + client.Transport.(*authenticatedTransport).base = roundTripFunc(func(*http.Request) (*http.Response, error) { + calls++ + return nil, errors.New("unexpected network call") + }) + for _, endpoint := range []string{ + "https://unrelated.example/get_scopes.php", "http://yow.meshmapper.net/get_scopes.php", + "https://yow.meshmapper.net.evil.example/get_scopes.php", "https://evilmeshmapper.net/get_scopes.php", + "https://user@yow.meshmapper.net/get_scopes.php", "https://yow.meshmapper.net:8443/get_scopes.php", + "https://yow.meshmapper.net/coverage.php?key=coverage-test-key", "https://yow.meshmapper.net/other.php", + "https://yow.meshmapper.net/%67et_scopes.php", "https://yow.meshmapper.net/get_scopes.php#fragment", + } { + _, _, _, err := get(context.Background(), client, "Beacon-Test", endpoint, "", MaxBody) + if !errors.Is(err, errUntrustedEndpoint) { + t.Errorf("untrusted endpoint accepted: %s", endpoint) + } + } + if calls != 0 { + t.Fatal("credentials could reach an untrusted target or Coverage API") + } +} + +func TestAuthenticatedClientInvalidKeyAndTransportErrors(t *testing.T) { + for _, key := range []string{"", "bad\r\nkey", "bad\x00key", "bad key"} { + client := newClient(key) + client.Transport.(*authenticatedTransport).base = roundTripFunc(func(*http.Request) (*http.Response, error) { + t.Error("unconfigured client made a network request") + return nil, errors.New("unexpected request") + }) + _, _, _, err := get(context.Background(), client, "Beacon-Test", ZonesURL+"?country=CA", "", MaxBody) + if !strings.Contains(requestProblem(err), "unconfigured") { + t.Fatal("unusable key not identified") + } + } + client := newClient("reflected-test-secret") + client.Transport.(*authenticatedTransport).base = roundTripFunc(func(*http.Request) (*http.Response, error) { + return nil, errors.New("reflected-test-secret") + }) + _, _, _, err := get(context.Background(), client, "Beacon-Test", ZonesURL+"?country=CA", "", MaxBody) + if requestProblem(err) != "request failed" { + t.Fatal("transport diagnostics not sanitized") + } +} + +type authSite struct { + *httptest.Server + key string + status map[string]int + after string + calls map[string]int +} + +func newAuthSite(t *testing.T, key string, iatas []string) *authSite { + t.Helper() + f := &authSite{key: key, status: map[string]int{}, calls: map[string]int{}} + f.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + f.calls[r.URL.Path]++ + if r.Header.Get("X-API-Key") != f.key { + t.Error("importer omitted configured key") + } + if r.URL.Query().Has("key") { + t.Error("credential in URL") + } + w.Header().Set("ETag", `"auth-v1"`) + w.Header().Set("Retry-After", f.after) + status := f.status[r.URL.Path] + if status != 0 { + w.WriteHeader(status) + _, _ = w.Write([]byte("upstream reflected " + f.key)) + return + } + iata := strings.ToUpper(strings.Split(r.Host, ".")[0]) + switch r.URL.Path { + case "/get_zones.php": + country := r.URL.Query().Get("country") + if country != "CA" && country != "US" { + t.Error("missing or changed country") + } + var zones []map[string]any + for _, code := range iatas { + if (code == "JFK") != (country == "US") { + continue + } + zones = append(zones, map[string]any{"code": code, "url": "https://" + strings.ToLower(code) + ".meshmapper.net/", "has_boundary": true}) + } + _ = json.NewEncoder(w).Encode(map[string]any{"country": country, "zones": zones}) + case "/get_geojson.php": + _, _ = w.Write([]byte(boundaryBody(iata, square))) + case "/get_scopes.php": + _, _ = w.Write(bytes.ReplaceAll(catalogue(strings.ToLower(iata)), []byte("YOW"), []byte(iata))) + case "/get_channels.php": + _, _ = w.Write(channelList(iata, "#"+strings.ToLower(iata))) + default: + t.Error("unexpected endpoint") + } + })) + t.Cleanup(f.Close) + return f +} + +type authHarness struct { + dir *Directory + zones *Zones + scopes *Importer + channels *Channels + lists *zoneListMemory + bounds *zoneMemoryStore + catalogues *memoryStore + channelLists *channelMemoryStore +} + +func newAuthHarness(t *testing.T, site *authSite, key string, iatas []string, saved *authHarness) *authHarness { + t.Helper() + h := &authHarness{} + if saved != nil { + h.lists, h.bounds, h.catalogues, h.channelLists = saved.lists, saved.bounds, saved.catalogues, saved.channelLists + } else { + h.lists = newZoneListMemory() + h.bounds = &zoneMemoryStore{iatas: iatas, rows: map[string]Boundary{}} + h.catalogues, h.channelLists = newMemoryStore(iatas...), newChannelMemoryStore(iatas...) + } + h.dir = NewDirectory(h.lists, key) + routeClient(t, h.dir.client, site.URL) + ctx := context.Background() + if err := h.dir.Restore(ctx); err != nil { + t.Fatal(err) + } + h.zones = NewZones(config.MeshMapperZonesConfig{Enabled: true}, h.bounds, h.dir) + if err := h.zones.Restore(ctx); err != nil { + t.Fatal(err) + } + var err error + h.scopes, err = New(ctx, config.MeshMapperScopesConfig{Enabled: true}, h.catalogues, h.dir, scopestore.New(), nil) + if err != nil { + t.Fatal(err) + } + h.channels, err = NewChannels(ctx, config.MeshMapperChannelsConfig{Enabled: true}, h.channelLists, h.dir, keystore.NewMapKeyStore(nil)) + if err != nil { + t.Fatal(err) + } + return h +} + +func (h *authHarness) tick(t *testing.T, n int) { + t.Helper() + for range n { + for _, refresh := range []func(context.Context) error{h.zones.Refresh, h.scopes.Refresh, h.channels.Refresh} { + if err := refresh(context.Background()); err != nil { + t.Fatal(err) + } + } + } +} + +func TestAuthenticatedImportersDeploymentScopes(t *testing.T) { + for _, tc := range []struct { + name string + iatas []string + }{ + {"regional", []string{"YOW"}}, {"group", []string{"YOW", "YUL"}}, {"global", []string{"YOW", "JFK"}}, + } { + t.Run(tc.name, func(t *testing.T) { + key := tc.name + "-operator-test-key" + f := newAuthSite(t, key, tc.iatas) + h := newAuthHarness(t, f, key, tc.iatas, nil) + h.tick(t, 5) + for _, iata := range tc.iatas { + if h.bounds.rows[iata].Feature == nil || len(h.channelLists.members[iata]) != 1 || !slices.Contains(names(h.scopes.scopes), "#"+strings.ToLower(iata)) { + t.Fatalf("%s data not imported", iata) + } + } + countries := 1 + if tc.name == "global" { + countries = 2 + } + if f.calls["/get_zones.php"] != countries { + t.Fatal("country lists not shared between importers") + } + for _, endpoint := range []string{"/get_geojson.php", "/get_scopes.php", "/get_channels.php"} { + if f.calls[endpoint] != len(tc.iatas) { + t.Fatalf("%s unexpected request count", endpoint) + } + } + }) + } +} + +func TestAuthenticatedRefreshFailuresKeepCachedData(t *testing.T) { + for _, tc := range []struct { + name string + status int + after, want string + wait time.Duration + }{ + {"invalid key", 401, "", "authentication", 0}, + {"insufficient regional permissions", 403, "", "permission", 0}, + {"rate limit", 429, "172800", "HTTP 429", 48 * time.Hour}, + {"unavailable", 503, "172800", "HTTP 503", 48 * time.Hour}, + {"missing key", 0, "", "unconfigured", 0}, + } { + t.Run(tc.name, func(t *testing.T) { + const key = "failure-test-secret-canary" + var logs bytes.Buffer + oldLogger := slog.Default() + slog.SetDefault(slog.New(slog.NewTextHandler(&logs, nil))) + t.Cleanup(func() { slog.SetDefault(oldLogger) }) + f := newAuthSite(t, key, []string{"YOW"}) + h := newAuthHarness(t, f, key, []string{"YOW"}, nil) + h.tick(t, 2) + goodBoundary := h.bounds.rows["YOW"] + goodScope, goodChannel := h.scopes.sources[0].cache, h.channels.sources[0].cache + goodList := h.lists.rows["CA"] + if tc.name == "missing key" { + h = newAuthHarness(t, f, "", []string{"YOW"}, h) + } + for _, path := range []string{"/get_geojson.php", "/get_scopes.php", "/get_channels.php", "/get_zones.php"} { + f.status[path] = tc.status + } + f.after = tc.after + h.zones.regions[0].b.NextAttempt = time.Time{} + h.scopes.sources[0].cache.NextAttempt = time.Time{} + h.channels.sources[0].cache.NextAttempt = time.Time{} + h.tick(t, 1) + list := h.dir.lists["CA"] + list.fetchedAt, list.nextAttempt = time.Time{}, time.Time{} + if _, _, err := h.dir.List(context.Background(), "CA", time.Now().UTC()); err != nil { + t.Fatal(err) + } + boundary, scope, channel, savedList := h.bounds.rows["YOW"], h.scopes.sources[0].cache, h.channels.sources[0].cache, h.lists.rows["CA"] + for _, cache := range []Cache{scope, channel, + {LastError: boundary.LastError, AttemptedAt: boundary.AttemptedAt, NextAttempt: boundary.NextAttempt}, + {LastError: savedList.LastError, AttemptedAt: savedList.AttemptedAt, NextAttempt: savedList.NextAttempt}, + } { + if !strings.Contains(cache.LastError, tc.want) { + t.Fatalf("failure not recorded: %s", cache.LastError) + } + if tc.wait != 0 && cache.NextAttempt.Sub(cache.AttemptedAt) != tc.wait { + t.Fatal("Retry-After not retained") + } + } + if !bytes.Equal(boundary.Feature, goodBoundary.Feature) || !boundary.CheckedAt.Equal(goodBoundary.CheckedAt) || boundary.ETag != goodBoundary.ETag || + !bytes.Equal(scope.Payload, goodScope.Payload) || !scope.CheckedAt.Equal(goodScope.CheckedAt) || scope.ETag != goodScope.ETag || + !bytes.Equal(channel.Payload, goodChannel.Payload) || !channel.CheckedAt.Equal(goodChannel.CheckedAt) || channel.ETag != goodChannel.ETag || + !bytes.Equal(savedList.Payload, goodList.Payload) || !savedList.FetchedAt.Equal(goodList.FetchedAt) || savedList.ETag != goodList.ETag { + t.Fatal("failed refresh replaced data or advanced freshness") + } + for path, calls := range f.calls { + want := 2 + if tc.name == "missing key" { + want = 1 + } + if calls != want { + t.Fatalf("%s: unexpected retries or missing-key requests: %d", path, calls) + } + } + before := fmt.Sprint(f.calls) + restarted := newAuthHarness(t, f, key, []string{"YOW"}, h) + restarted.tick(t, 5) + if fmt.Sprint(f.calls) != before { + t.Fatal("restart retried before backoff elapsed") + } + if !restarted.lists.rows["CA"].AttemptedAt.Equal(savedList.AttemptedAt) || + !restarted.scopes.sources[0].cache.AttemptedAt.Equal(scope.AttemptedAt) || + !restarted.channels.sources[0].cache.AttemptedAt.Equal(channel.AttemptedAt) || + !restarted.zones.regions[0].b.AttemptedAt.Equal(boundary.AttemptedAt) { + t.Fatal("restart attempted a refresh before persisted backoff elapsed") + } + if restarted.zones.regions[0].b.Feature == nil || len(restarted.channels.sources[0].entries) != 1 || len(names(restarted.scopes.scopes)) != 1 || len(restarted.dir.lists["CA"].zones) != 1 { + t.Fatal("restart lost last successful cache") + } + if strings.Contains(logs.String(), key) { + t.Fatal("key leaked to logs") + } + }) + } +} diff --git a/internal/meshmapper/channels.go b/internal/meshmapper/channels.go index 33a07e3..4a95e31 100644 --- a/internal/meshmapper/channels.go +++ b/internal/meshmapper/channels.go @@ -64,10 +64,7 @@ type Channels struct { // Disabled, it clears MeshMapper's region membership so config scoping alone applies. func NewChannels(ctx context.Context, cfg config.MeshMapperChannelsConfig, store ChannelStore, dir *Directory, keys ChannelKeys) (*Channels, error) { c := &Channels{store: store, keys: keys, dir: dir, channelsURL: channelsEndpoint, interval: cfg.Interval(), seen: map[string]bool{}, - client: &http.Client{ - Timeout: requestTimeout, - CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }, - }} + client: dir.client} if !cfg.Enabled { if err := store.ClearChannelMembers(ctx); err != nil { return nil, fmt.Errorf("clear MeshMapper channel members: %w", err) @@ -186,19 +183,21 @@ func (c *Channels) refresh(ctx context.Context, s *channelSource, now time.Time) } s.cache.AttemptedAt, s.cache.LastError = now, attempt.LastError status, body, header, err := get(ctx, c.client, "Beacon-MeshMapper-Channels/1", s.url, etag, MaxChannelBody) - if err != nil { - return err + if ctx.Err() != nil { + return ctx.Err() } var entries []keystore.Entry var retryAt time.Time - switch status { - case http.StatusOK: + switch { + case err != nil: + update.LastError = requestProblem(err) + case status == http.StatusOK: if entries, err = decodeChannels(body, s.iata); err != nil { update.LastError = "invalid response" } else { update.Payload = body } - case http.StatusNotModified: + case status == http.StatusNotModified: if len(s.cache.Payload) == 0 { update.LastError = "304 without cached catalogue" } diff --git a/internal/meshmapper/channels_test.go b/internal/meshmapper/channels_test.go index 29773aa..1380927 100644 --- a/internal/meshmapper/channels_test.go +++ b/internal/meshmapper/channels_test.go @@ -115,7 +115,7 @@ type channelHarness struct { func newChannelHarness(t *testing.T, f *channelSite, store *channelMemoryStore, lists *zoneListMemory, enabled bool) *channelHarness { t.Helper() - dir := NewDirectory(lists) + dir := newTestDirectory(lists) dir.listURL = f.URL + "/get_zones.php" if err := dir.Restore(context.Background()); err != nil { t.Fatal(err) @@ -237,7 +237,7 @@ func TestChannelsRequestCutShortStillCounts(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { <-r.Context().Done() })) defer server.Close() store := newChannelMemoryStore("YOW") - c, err := NewChannels(context.Background(), config.MeshMapperChannelsConfig{Enabled: true}, store, NewDirectory(newZoneListMemory()), keystore.NewMapKeyStore(nil)) + c, err := NewChannels(context.Background(), config.MeshMapperChannelsConfig{Enabled: true}, store, newTestDirectory(newZoneListMemory()), keystore.NewMapKeyStore(nil)) if err != nil { t.Fatal(err) } diff --git a/internal/meshmapper/client.go b/internal/meshmapper/client.go new file mode 100644 index 0000000..0ab1853 --- /dev/null +++ b/internal/meshmapper/client.go @@ -0,0 +1,67 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package meshmapper + +import ( + "errors" + "net/http" + "net/url" + "strings" +) + +var ( + errMissingKey = errors.New("unconfigured: MeshMapper API key is missing") + errInvalidKey = errors.New("unconfigured: MeshMapper API key contains invalid characters") + errUntrustedEndpoint = errors.New("untrusted MeshMapper endpoint") +) + +func newClient(apiKey string) *http.Client { + return &http.Client{ + Timeout: requestTimeout, + CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }, + Transport: &authenticatedTransport{key: apiKey, base: http.DefaultTransport}, + } +} + +type authenticatedTransport struct { + key string + base http.RoundTripper +} + +func (t *authenticatedTransport) RoundTrip(req *http.Request) (*http.Response, error) { + if t.key == "" { + return nil, errMissingKey + } + if strings.ContainsFunc(t.key, func(r rune) bool { return r <= ' ' || r >= 127 }) { + return nil, errInvalidKey + } + if !trustedEndpoint(req.URL) || (req.Host != "" && req.Host != req.URL.Host) { + return nil, errUntrustedEndpoint + } + // Clone so credentials never enter a caller's request or redirect headers. + request := req.Clone(req.Context()) + request.Header.Set("X-API-Key", t.key) + return t.base.RoundTrip(request) +} + +func trustedEndpoint(u *url.URL) bool { + if u.Scheme != "https" || u.User != nil || u.Fragment != "" || u.Opaque != "" || + (u.Host != "meshmapper.net" && !zoneSite.MatchString(u.Host)) || u.EscapedPath() != u.Path { + return false + } + switch u.Path { + case "/get_zones.php", "/get_geojson.php", "/get_scopes.php", "/get_channels.php", "/get_repeaters.php": + return true + } + return false +} + +func requestProblem(err error) string { + for _, problem := range []error{errMissingKey, errInvalidKey, errUntrustedEndpoint} { + if errors.Is(err, problem) { + return problem.Error() + } + } + return "request failed" // Transport errors may contain URLs or credentials. +} diff --git a/internal/meshmapper/directory.go b/internal/meshmapper/directory.go index 5113f02..c50ed21 100644 --- a/internal/meshmapper/directory.go +++ b/internal/meshmapper/directory.go @@ -63,11 +63,8 @@ type Directory struct { version int // bumped whenever a list is replaced } -func NewDirectory(store DirectoryStore) *Directory { - return &Directory{store: store, listURL: ZonesURL, lists: map[string]*zoneList{}, client: &http.Client{ - Timeout: requestTimeout, - CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }, - }} +func NewDirectory(store DirectoryStore, apiKey string) *Directory { + return &Directory{store: store, listURL: ZonesURL, lists: map[string]*zoneList{}, client: newClient(apiKey)} } // Restore loads saved lists so a restart doesn't spend each country's daily request. @@ -130,23 +127,25 @@ func (d *Directory) fetch(ctx context.Context, list *zoneList, country string, c } endpoint := d.listURL + "?country=" + url.QueryEscape(country) status, body, header, err := get(ctx, d.client, "Beacon-MeshMapper-Zones/1", endpoint, etag, MaxZoneList) - if err != nil { - return err + if ctx.Err() != nil { + return ctx.Err() } problem := "" var payload json.RawMessage var retryAt time.Time var zones map[string]zoneEntry var groups map[string]zoneGroup - switch status { - case http.StatusOK: + switch { + case err != nil: + problem = requestProblem(err) + case status == http.StatusOK: var decodeErr error if zones, groups, decodeErr = decodeZones(body, country); decodeErr != nil { problem = "invalid response" } else { payload = body } - case http.StatusNotModified: + case status == http.StatusNotModified: if !cached { problem = "304 without cached list" } diff --git a/internal/meshmapper/groups_test.go b/internal/meshmapper/groups_test.go index fa1d853..5517f9e 100644 --- a/internal/meshmapper/groups_test.go +++ b/internal/meshmapper/groups_test.go @@ -39,7 +39,7 @@ func newGroupHarness(t *testing.T, f *fakeMeshMapper, store *zoneMemoryStore, im store.rows = map[string]Boundary{} } h := &groupHarness{f: f, store: store} - dir := NewDirectory(newZoneListMemory()) + dir := newTestDirectory(newZoneListMemory()) dir.listURL = f.URL + "/get_zones.php" h.z = NewZones(config.MeshMapperZonesConfig{Enabled: true, ImportGroups: importGroups}, store, dir) h.z.boundsURL = func(site string) (string, bool) { return site + "get_geojson.php", true } diff --git a/internal/meshmapper/scopes.go b/internal/meshmapper/scopes.go index 7677018..35dba76 100644 --- a/internal/meshmapper/scopes.go +++ b/internal/meshmapper/scopes.go @@ -8,11 +8,9 @@ import ( "context" "encoding/json" "fmt" - "io" "log/slog" "net/http" "slices" - "strconv" "strings" "time" "unicode" @@ -78,10 +76,7 @@ func (i *Importer) SetCacheInvalidator(fn func(context.Context)) { i.onChange = // New restores validated snapshots before ingestion, without making HTTP requests. func New(ctx context.Context, cfg config.MeshMapperScopesConfig, store Store, dir *Directory, scopes *scopestore.ScopeStore, manual []scopestore.Entry) (*Importer, error) { i := &Importer{store: store, scopes: scopes, manual: manual, dir: dir, scopesURL: scopesEndpoint, seen: map[string]bool{}, - interval: cfg.Interval(), client: &http.Client{ - Timeout: requestTimeout, - CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }, - }} + interval: cfg.Interval(), client: dir.client} if !cfg.Enabled { return i, nil } @@ -194,14 +189,9 @@ func (i *Importer) Refresh(ctx context.Context) (err error) { func (i *Importer) refresh(ctx context.Context, s *source, now time.Time) error { update := Cache{AttemptedAt: now, NextAttempt: now.Add(i.interval)} - request, err := http.NewRequestWithContext(ctx, http.MethodGet, s.url, nil) - if err != nil { - return err - } - request.Header.Set("Accept", "application/json") - request.Header.Set("User-Agent", "Beacon-MeshMapper-Scopes/1") - if len(s.cache.Payload) > 0 && s.cache.ETag != "" { - request.Header.Set("If-None-Match", s.cache.ETag) + etag := "" + if len(s.cache.Payload) > 0 { + etag = s.cache.ETag } // Recorded first: an abandoned request may still have used the region's call. attempt := Cache{AttemptedAt: now, NextAttempt: now.Add(min(i.interval, failureRetry)), LastError: "no response"} @@ -210,54 +200,41 @@ func (i *Importer) refresh(ctx context.Context, s *source, now time.Time) error return fmt.Errorf("persist scope catalogue %s: %w", s.iata, err) } s.cache.AttemptedAt, s.cache.LastError = now, attempt.LastError - response, err := i.client.Do(request) + status, body, header, err := get(ctx, i.client, "Beacon-MeshMapper-Scopes/1", s.url, etag, MaxBody) + if ctx.Err() != nil { + return ctx.Err() + } var entries []scopestore.Entry var generated time.Time var retryAfter time.Time - if err != nil { - if ctx.Err() != nil { - return ctx.Err() + switch { + case err != nil: + update.LastError = requestProblem(err) + case status == http.StatusOK: + if entries, generated, err = decode(body, s.iata); err != nil { + update.LastError = "invalid response" + } else { + update.Payload = body } - update.LastError = "request failed" // do not persist untrusted response bodies or URLs from errors - } else { - defer response.Body.Close() - switch response.StatusCode { - case http.StatusOK: - body, readErr := io.ReadAll(io.LimitReader(response.Body, MaxBody+1)) - if readErr == nil { - entries, generated, readErr = decode(body, s.iata) - } - if readErr != nil { - update.LastError = "invalid response" - } else { - update.Payload = body - } - case http.StatusNotModified: - if len(s.cache.Payload) == 0 || s.cache.ETag == "" { - update.LastError = "304 without cached catalogue" - } - default: - update.LastError = fmt.Sprintf("HTTP %d", response.StatusCode) - if response.StatusCode == http.StatusTooManyRequests || response.StatusCode == http.StatusServiceUnavailable { - // The published contract uses seconds. Ignore invalid/overflowing headers. - if seconds, parseErr := strconv.ParseInt(response.Header.Get("Retry-After"), 10, 64); parseErr == nil && seconds > 0 && seconds <= int64((1<<63-1)/time.Second) { - retryAfter = now.Add(time.Duration(seconds) * time.Second) - } - } + case status == http.StatusNotModified: + if len(s.cache.Payload) == 0 || s.cache.ETag == "" { + update.LastError = "304 without cached catalogue" } - if update.LastError == "" { - etag := response.Header.Get("ETag") - if len(etag) > 256 || strings.ContainsAny(etag, "\r\n") { - update.LastError = "invalid ETag" - update.Payload = nil - entries = nil - } else { - update.ETag = etag - if response.StatusCode == http.StatusNotModified && etag == "" { - update.ETag = s.cache.ETag - } - update.CheckedAt = now + default: + update.LastError = statusProblem(status, header, now, &retryAfter) + } + if update.LastError == "" { + etag := header.Get("ETag") + if len(etag) > 256 || strings.ContainsAny(etag, "\r\n") { + update.LastError = "invalid ETag" + update.Payload = nil + entries = nil + } else { + update.ETag = etag + if status == http.StatusNotModified && etag == "" { + update.ETag = s.cache.ETag } + update.CheckedAt = now } } if update.LastError != "" { diff --git a/internal/meshmapper/scopes_test.go b/internal/meshmapper/scopes_test.go index 804410a..7e56ca2 100644 --- a/internal/meshmapper/scopes_test.go +++ b/internal/meshmapper/scopes_test.go @@ -75,7 +75,7 @@ func (s *memoryStore) SaveScopeCatalogue(_ context.Context, iata, url string, ne // newImporter restores an importer for YOW whose source points at url. func newImporter(t *testing.T, cfg config.MeshMapperScopesConfig, store *memoryStore, url string, scopes *scopestore.ScopeStore, manual ...scopestore.Entry) *Importer { t.Helper() - imp, err := New(context.Background(), cfg, store, NewDirectory(newZoneListMemory()), scopes, manual) + imp, err := New(context.Background(), cfg, store, newTestDirectory(newZoneListMemory()), scopes, manual) if err != nil { t.Fatal(err) } @@ -260,7 +260,7 @@ func TestImportedSourcesOverlapWithoutGlobalMembership(t *testing.T) { store.rows[sourceKey{iata, iata}] = Cache{Payload: []byte(strings.ReplaceAll(string(catalogue("can")), "YOW", iata)), AttemptedAt: time.Now()} } scopes := scopestore.New() - _, err := New(ctx, config.MeshMapperScopesConfig{Enabled: true}, store, NewDirectory(newZoneListMemory()), scopes, nil) + _, err := New(ctx, config.MeshMapperScopesConfig{Enabled: true}, store, newTestDirectory(newZoneListMemory()), scopes, nil) if err != nil || len(scopes.Entries()) != 1 || !reflect.DeepEqual(scopes.Entries()[0].IATAs, []string{"YOW", "YVR"}) { t.Fatal("latest saved catalogue per IATA not restored", scopes.Entries(), err) } @@ -269,7 +269,7 @@ func TestImportedSourcesOverlapWithoutGlobalMembership(t *testing.T) { } // Disabled mode never calls the store. disabled := scopestore.New() - if _, err = New(ctx, config.MeshMapperScopesConfig{}, nil, NewDirectory(newZoneListMemory()), disabled, nil); err != nil || len(disabled.Entries()) != 0 { + if _, err = New(ctx, config.MeshMapperScopesConfig{}, nil, newTestDirectory(newZoneListMemory()), disabled, nil); err != nil || len(disabled.Entries()) != 0 { t.Fatal(err) } } @@ -360,7 +360,7 @@ func newFakeSites(t *testing.T) *fakeSites { func newDiscoveringImporter(t *testing.T, f *fakeSites, store *memoryStore, scopes *scopestore.ScopeStore) *Importer { t.Helper() - dir := NewDirectory(newZoneListMemory()) + dir := newTestDirectory(newZoneListMemory()) dir.listURL = f.URL + "/get_zones.php" imp, err := New(context.Background(), config.MeshMapperScopesConfig{Enabled: true}, store, dir, scopes, nil) if err != nil { @@ -448,7 +448,7 @@ func newSweep(t *testing.T, store *memoryStore, fail ...string) (*Importer, *[]s _, _ = w.Write([]byte(strings.ReplaceAll(string(catalogue("x")), "YOW", iata))) })) t.Cleanup(server.Close) - dir := NewDirectory(newZoneListMemory()) + dir := newTestDirectory(newZoneListMemory()) dir.listURL = server.URL + "/get_zones.php" imp, err := New(context.Background(), config.MeshMapperScopesConfig{Enabled: true}, store, dir, scopestore.New(), nil) if err != nil { diff --git a/internal/meshmapper/zones.go b/internal/meshmapper/zones.go index 4bec389..6893470 100644 --- a/internal/meshmapper/zones.go +++ b/internal/meshmapper/zones.go @@ -87,10 +87,7 @@ type Zones struct { func NewZones(cfg config.MeshMapperZonesConfig, store ZoneStore, dir *Directory) *Zones { return &Zones{store: store, enabled: cfg.Enabled, importGroups: cfg.Enabled && cfg.ImportGroups, seen: map[string]bool{}, interval: cfg.Interval(), dir: dir, - boundsURL: boundaryEndpoint, client: &http.Client{ - Timeout: requestTimeout, - CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }, - }} + boundsURL: boundaryEndpoint, client: dir.client} } // OnChange receives each IATA whose served border changed, for cache invalidation. @@ -228,11 +225,13 @@ func (z *Zones) refresh(ctx context.Context, r *zoneRegion, zones map[string]zon } r.b.URL, r.b.AttemptedAt, r.b.LastError = attempt.URL, now, attempt.LastError status, body, header, err := get(ctx, z.client, "Beacon-MeshMapper-Zones/1", endpoint, etag, MaxBoundaryBody) - if err != nil { - return err + if ctx.Err() != nil { + return ctx.Err() } - switch status { - case http.StatusOK: + switch { + case err != nil: + update.LastError = requestProblem(err) + case status == http.StatusOK: feature, decodeErr := decodeBoundary(body, r.b.IATA) switch { case decodeErr != nil: @@ -242,7 +241,7 @@ func (z *Zones) refresh(ctx context.Context, r *zoneRegion, zones map[string]zon default: update.Feature = feature } - case http.StatusNotModified: + case status == http.StatusNotModified: if r.b.Feature == nil { update.LastError = "304 without cached boundary" } @@ -287,7 +286,7 @@ func (z *Zones) refresh(ctx context.Context, r *zoneRegion, zones map[string]zon return nil } -// get returns transport errors only when ctx ended; other failures become status 0. +// get never returns response bodies for failed requests. func get(ctx context.Context, client *http.Client, agent, endpoint, etag string, limit int64) (int, []byte, http.Header, error) { request, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil) if err != nil { @@ -300,10 +299,7 @@ func get(ctx context.Context, client *http.Client, agent, endpoint, etag string, } response, err := client.Do(request) if err != nil { - if ctx.Err() != nil { - return 0, nil, nil, ctx.Err() - } - return 0, nil, http.Header{}, nil + return 0, nil, nil, err } defer response.Body.Close() if response.StatusCode != http.StatusOK { @@ -321,9 +317,17 @@ func statusProblem(status int, header http.Header, now time.Time, retryAt *time. if status == 0 { return "request failed" // don't persist untrusted error text } + if status == http.StatusUnauthorized { + return "authentication failed (HTTP 401)" + } + if status == http.StatusForbidden { + return "permission denied (HTTP 403)" + } if status == http.StatusTooManyRequests || status == http.StatusServiceUnavailable { if seconds, err := strconv.ParseInt(header.Get("Retry-After"), 10, 64); err == nil && seconds > 0 && seconds <= int64((1<<63-1)/time.Second) { *retryAt = now.Add(time.Duration(seconds) * time.Second) + } else if deadline, err := http.ParseTime(header.Get("Retry-After")); err == nil && deadline.After(now) { + *retryAt = deadline } } return fmt.Sprintf("HTTP %d", status) diff --git a/internal/meshmapper/zones_test.go b/internal/meshmapper/zones_test.go index b4bb35f..f42720a 100644 --- a/internal/meshmapper/zones_test.go +++ b/internal/meshmapper/zones_test.go @@ -227,7 +227,7 @@ func newZoneHarness(t *testing.T, f *fakeMeshMapper, store *zoneMemoryStore, ena if store.lists == nil { store.lists = newZoneListMemory() } - dir := NewDirectory(store.lists) + dir := newTestDirectory(store.lists) dir.listURL = f.URL + "/get_zones.php" if err := dir.Restore(context.Background()); err != nil { t.Fatal(err) @@ -488,17 +488,17 @@ func TestDecodeBoundaryRejectsUnusableOutlines(t *testing.T) { func TestClientsAllowMeshMapperTimeout(t *testing.T) { ctx := context.Background() - scopes, err := New(ctx, config.MeshMapperScopesConfig{}, nil, NewDirectory(nil), scopestore.New(), nil) + scopes, err := New(ctx, config.MeshMapperScopesConfig{}, nil, newTestDirectory(nil), scopestore.New(), nil) if err != nil { t.Fatal(err) } - channels, err := NewChannels(ctx, config.MeshMapperChannelsConfig{}, newChannelMemoryStore(), NewDirectory(nil), keystore.NewMapKeyStore(nil)) + channels, err := NewChannels(ctx, config.MeshMapperChannelsConfig{}, newChannelMemoryStore(), newTestDirectory(nil), keystore.NewMapKeyStore(nil)) if err != nil { t.Fatal(err) } for name, c := range map[string]*http.Client{ - "directory": NewDirectory(nil).client, - "zones": NewZones(config.MeshMapperZonesConfig{}, nil, nil).client, + "directory": newTestDirectory(nil).client, + "zones": NewZones(config.MeshMapperZonesConfig{}, nil, newTestDirectory(nil)).client, "scopes": scopes.client, "channels": channels.client, } { @@ -515,7 +515,7 @@ func TestZoneListCutShortStillCounts(t *testing.T) { f := newFakeMeshMapper(t) f.hang.Store(true) lists := newZoneListMemory() - dir := NewDirectory(lists) + dir := newTestDirectory(lists) dir.listURL = f.URL + "/get_zones.php" now := time.Now().UTC() ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) @@ -542,7 +542,7 @@ func TestZoneListFetchDoesNotBlockOtherCallers(t *testing.T) { })) defer server.Close() defer close(release) - dir := NewDirectory(newZoneListMemory()) + dir := newTestDirectory(newZoneListMemory()) dir.listURL = server.URL now := time.Now().UTC() go func() { _, _, _ = dir.List(context.Background(), "CA", now) }()