diff --git a/README.md b/README.md index 8052707..855b9b8 100644 --- a/README.md +++ b/README.md @@ -202,6 +202,8 @@ even when the same bearer-authenticated request works with curl. # zones: # enabled: true # import_groups: true # MeshMapper zone groups become regions +# channels: +# enabled: true # import each region's public hashtag channels # Optional IATA overrides — auto-created on first packet arrival, # only needed if you want to customise display name or coordinates. @@ -229,9 +231,12 @@ regions: channel_keys: # Hashtag channels: Beacon derives the PSK from the tag name automatically. # secret = SHA256("#tag")[:16], channel_hash = SHA256(secret)[0] - # Tag names should be provided without the # prefix. + # Tag names should be provided without the # prefix. Plain names show under + # every region; {name, region: } limits a channel to one region. hashtags: - meshcore + - name: vancouver-mesh + region: western-canada # Explicit keys: channel hash (hex) and key (hex), with optional display name. # The public MeshCore channel key is included in config.yaml.example. diff --git a/cmd/beacon/main.go b/cmd/beacon/main.go index 2194636..4e4cf9a 100644 --- a/cmd/beacon/main.go +++ b/cmd/beacon/main.go @@ -199,7 +199,7 @@ func main() { scopes.SetManualMembers(cfg.ManualScopeMembers()) slog.Info(fmt.Sprintf("loaded %d transport scopes", len(scopeEntries)), "component", "startup") directory := meshmapper.NewDirectory(store) - if cfg.MeshMapper.Scopes.Enabled || cfg.MeshMapper.Zones.Enabled { + if cfg.MeshMapper.Scopes.Enabled || cfg.MeshMapper.Zones.Enabled || cfg.MeshMapper.Channels.Enabled { restoreCtx, cancelRestore := context.WithTimeout(ctx, 10*time.Second) err = directory.Restore(restoreCtx) cancelRestore() @@ -250,7 +250,8 @@ func main() { entries := make(map[string][]keystore.Entry) // Hashtag-derived channels: secret = SHA256("#tag")[:16], hash = SHA256(secret)[0] - for _, tag := range cfg.ChannelKeys.Hashtags { + for _, hashtag := range cfg.ChannelKeys.Hashtags { + tag := hashtag.Name secret, channelHash, fingerprint := keystore.DeriveHashtagKey(tag) hashHex := fmt.Sprintf("%02x", channelHash) entry := keystore.Entry{ @@ -285,6 +286,27 @@ func main() { keys := keystore.NewMapKeyStore(entries) + // Restored before the boot backfill so saved MeshMapper keys decrypt history too. + restoreCtx, cancelRestore = context.WithTimeout(ctx, 10*time.Second) + channelImporter, err := meshmapper.NewChannels(restoreCtx, cfg.MeshMapper.Channels, store, directory, keys) + cancelRestore() + if err != nil { + slog.Error("failed to restore MeshMapper channels", "component", "startup", "error", err) + os.Exit(1) + } + channelImporter.OnNewKeys(func(hashes [][]byte) { + // Off the refresh tick: a backfill can outlast it, and history isn't urgent. + go func() { + backfillCtx, cancel := context.WithTimeout(ctx, 10*time.Minute) + defer cancel() + if n, err := ingest.BackfillChannelHashes(backfillCtx, store, keys, hashes); err != nil { + slog.Error("MeshMapper channel backfill failed", "component", "meshmapper.channels", "error", err) + } else if n > 0 { + slog.Info(fmt.Sprintf("backfilled %d channel message(s) for imported keys", n), "component", "meshmapper.channels") + } + }() + }) + // ── Backfill channel messages ──────────────────────────────────────────── // Packets whose channel key wasn't yet configured at ingest time were stored as // hash-only channels and never decrypted. Retry them now against the keystore we just @@ -366,6 +388,9 @@ func main() { if cfg.MeshMapper.Zones.Enabled { tasks = append(tasks, background.Task{Name: "meshmapper.zones", Interval: meshmapper.PollInterval, Run: zones.Refresh}) } + if cfg.MeshMapper.Channels.Enabled { + tasks = append(tasks, background.Task{Name: "meshmapper.channels", Interval: meshmapper.PollInterval, Run: channelImporter.Refresh}) + } profiles := configureProfiling(ctx, pool) defer profiles.Stop() for i := range tasks { diff --git a/config.yaml.example b/config.yaml.example index e315c9e..dce503d 100644 --- a/config.yaml.example +++ b/config.yaml.example @@ -83,6 +83,18 @@ ratelimit: # enabled: false # import_groups: false # refresh_interval: 24h # 24h-168h (one get_geojson.php per region per 23.5h); one request per 15s tick, ETag-conditional +# +# Public hashtag channels (get_channels.php). Every known IATA that MeshMapper +# lists is looked up on its site, the same way as scopes. Each key and hash is +# checked against the name before use; a list with any mismatch is rejected and +# the last good one kept. Imported channels decrypt like configured ones (history +# is backfilled when a key first appears) and are listed under that IATA when +# /channels is filtered by region. Saved in PostgreSQL across restarts; failures +# wait out the 23.5h limit. Turning it off removes the region listing on restart. +# Logged under component=meshmapper.channels. +# channels: +# enabled: false +# refresh_interval: 24h # 24h-168h (one get_channels.php per region per 23.5h) iatas: YVR: @@ -138,8 +150,12 @@ channel_keys: # secret = SHA256("#tag")[:16], channel_hash = SHA256(secret)[0] # Hashtag names should be provided without the # prefix. # Beacon prepends # automatically before deriving the key. + # A plain name is listed under every region; add region: (a configured + # region) to list it only there when /channels is filtered by region. hashtags: - meshcore + # - name: ottawa-mesh + # region: eastern-canada # Explicit keys: provide the channel hash (hex) and key (hex) directly. # An optional name can be set for display purposes. @@ -147,6 +163,7 @@ channel_keys: "11": key: "8b3387e9c5cdea6ac9e5edbaa115cd72" name: "Public" + # region: eastern-canada # optional, as for hashtags # Regional transport scopes for matching TRANSPORT_FLOOD packets. # Plain names have # prepended automatically (e.g. "bc" → "#bc"). diff --git a/db/channels.go b/db/channels.go index 9082e5a..a4718b5 100644 --- a/db/channels.go +++ b/db/channels.go @@ -65,6 +65,31 @@ func (s *Store) ListUndecryptedGroupTextPackets(ctx context.Context) ([]ingest.U return packets, nil } +// ListUndecryptedGroupTextPacketsByHash limits the boot backfill to channels that just gained a key. +func (s *Store) ListUndecryptedGroupTextPacketsByHash(ctx context.Context, hashes [][]byte) ([]ingest.UndecryptedPacket, error) { + rows, err := s.q.ListUndecryptedGroupTextPacketsByHash(ctx, hashes) + if err != nil { + return nil, err + } + packets := make([]ingest.UndecryptedPacket, 0, len(rows)) + for _, v := range rows { + packets = append(packets, ingest.UndecryptedPacket{PacketHash: v.PacketHash, RawPayload: v.RawPayload}) + } + return packets, nil +} + +// SetChannelConfigScopes replaces every configured channel's region placement; +// an empty region means Beacon-wide. +func (s *Store) SetChannelConfigScopes(ctx context.Context, fingerprints [][]byte, regions []string) error { + if err := s.q.DeleteChannelConfigScopes(ctx); err != nil { + return err + } + if len(fingerprints) == 0 { + return nil + } + return s.q.AddChannelConfigScopes(ctx, sqlc.AddChannelConfigScopesParams{Fingerprints: fingerprints, Regions: regions}) +} + func (s *Store) UpsertChannelIATA(ctx context.Context, channelHash []byte, iata string, heardAt time.Time) error { return s.q.UpsertChannelIATA(ctx, sqlc.UpsertChannelIATAParams{ ChannelHash: channelHash, diff --git a/db/meshmapper.go b/db/meshmapper.go index 6ff3fa0..0940513 100644 --- a/db/meshmapper.go +++ b/db/meshmapper.go @@ -176,3 +176,55 @@ func (s *Store) PruneImportedRegions(ctx context.Context, keep []string) ([]stri } return s.q.PruneImportedRegions(ctx, keep) } + +func (s *Store) ListChannelCatalogues(ctx context.Context) ([]meshmapper.Catalogue, error) { + rows, err := s.q.ListChannelCatalogues(ctx) + if err != nil { + return nil, err + } + out := make([]meshmapper.Catalogue, 0, len(rows)) + for _, row := range rows { + c := meshmapper.Catalogue{IATA: row.Iata, URL: row.Url, Cache: meshmapper.Cache{Payload: row.Payload, CheckedAt: row.CheckedAt.Time, + AttemptedAt: row.AttemptedAt.Time, NextAttempt: row.NextAttempt.Time, LastError: row.LastError}} + if row.Etag != nil { + c.ETag = *row.Etag + } + out = append(out, c) + } + return out, nil +} + +// SaveChannelCatalogue records a fetch; a new payload also replaces the IATA's +// channel members, which are written first so a failure never outlives its snapshot. +func (s *Store) SaveChannelCatalogue(ctx context.Context, iata, url string, cache meshmapper.Cache, fingerprints [][]byte) error { + if cache.Payload != nil { + if err := s.SetChannelMembers(ctx, iata, fingerprints); err != nil { + return err + } + } + params := sqlc.SaveChannelCatalogueParams{ + Iata: iata, Url: url, Payload: cache.Payload, LastError: cache.LastError, + CheckedAt: pgtype.Timestamptz{Time: cache.CheckedAt, Valid: !cache.CheckedAt.IsZero()}, + AttemptedAt: pgtype.Timestamptz{Time: cache.AttemptedAt, Valid: true}, + NextAttempt: pgtype.Timestamptz{Time: cache.NextAttempt, Valid: true}, + } + if !cache.CheckedAt.IsZero() { + params.Etag = &cache.ETag + } + return s.q.SaveChannelCatalogue(ctx, params) +} + +// SetChannelMembers makes fingerprints the IATA's exact MeshMapper channel list. +func (s *Store) SetChannelMembers(ctx context.Context, iata string, fingerprints [][]byte) error { + if fingerprints == nil { + fingerprints = [][]byte{} + } + if err := s.q.DeleteChannelMembersNotIn(ctx, sqlc.DeleteChannelMembersNotInParams{Iata: iata, Keep: fingerprints}); err != nil { + return err + } + return s.q.AddChannelMembers(ctx, sqlc.AddChannelMembersParams{Iata: iata, Fingerprints: fingerprints}) +} + +func (s *Store) ClearChannelMembers(ctx context.Context) error { + return s.q.DeleteAllChannelMembers(ctx) +} diff --git a/db/meshmapper_integration_test.go b/db/meshmapper_integration_test.go index 7dd8c50..f80c0f7 100644 --- a/db/meshmapper_integration_test.go +++ b/db/meshmapper_integration_test.go @@ -6,6 +6,7 @@ package db import ( "context" "os" + "slices" "strings" "testing" "time" @@ -328,3 +329,108 @@ func TestMeshMapperRegionsPostgres(t *testing.T) { t.Fatal("prune touched configured regions", pruned, err) } } + +func TestChannelRegionScopingPostgres(t *testing.T) { + dsn := os.Getenv("BEACON_TEST_POSTGRES_DSN") + if dsn == "" { + t.Skip("set BEACON_TEST_POSTGRES_DSN") + } + ctx, cancel := context.WithTimeout(context.Background(), time.Minute) + defer cancel() + pool, err := pgxpool.New(ctx, dsn) + if err != nil { + t.Fatal(err) + } + if err := RunMigrations(ctx, pool); err != nil { + pool.Close() + t.Fatal(err) + } + pool.Close() + conn, err := pgx.Connect(ctx, dsn) + if err != nil { + t.Fatal(err) + } + defer conn.Close(context.Background()) + tx, err := conn.Begin(ctx) + if err != nil { + t.Fatal(err) + } + defer tx.Rollback(context.Background()) + for _, table := range []string{"regions", "region_iatas", "channels", "channel_iatas", "channel_config_scopes", + "meshmapper_channel_members", "meshmapper_channel_catalogues"} { + if _, err := tx.Exec(ctx, "CREATE TEMP TABLE "+table+" (LIKE public."+table+" INCLUDING ALL) ON COMMIT DROP"); err != nil { + t.Fatal(err) + } + } + store := &Store{q: sqlc.New(tx)} + for slug, iata := range map[string]string{"east": "YOW", "west": "YYZ"} { + id, err := store.UpsertRegion(ctx, slug, slug, "", 0, nil, nil, nil) + if err != nil { + t.Fatal(err) + } + if err := store.SetRegionIATAs(ctx, id, []string{iata}); err != nil { + t.Fatal(err) + } + } + fp := func(b byte) []byte { return []byte{b, b, b, b, b, b, b, b} } + for name, b := range map[string]byte{"#mm-yow": 1, "#config-east": 2, "#global": 3, "#toronto": 4} { + if _, err := store.UpsertChannel(ctx, []byte{b}, fp(b), name, name[1:]); err != nil { + t.Fatal(err) + } + } + if err := store.UpsertChannelIATA(ctx, []byte{4}, "YOW", time.Now()); err != nil { + t.Fatal(err) + } + if err := store.SetChannelConfigScopes(ctx, [][]byte{fp(2), fp(3)}, []string{"east", ""}); err != nil { + t.Fatal(err) + } + now := time.Now().UTC() + for iata, b := range map[string]byte{"YOW": 1, "YYZ": 4} { + update := meshmapper.Cache{Payload: []byte(`{}`), CheckedAt: now, AttemptedAt: now, NextAttempt: now.Add(24 * time.Hour)} + if err := store.SaveChannelCatalogue(ctx, iata, "https://x.meshmapper.net/get_channels.php", update, [][]byte{fp(b)}); err != nil { + t.Fatal(err) + } + } + names := func(iatas ...string) string { + t.Helper() + page, err := store.ListChannels(ctx, 50, nil, iatas, 0, nil) + if err != nil { + t.Fatal(err) + } + var out []string + for _, c := range page.Items { + out = append(out, *c.Name) + } + slices.Sort(out) + return strings.Join(out, ",") + } + if got := names("YOW"); got != "#config-east,#global,#mm-yow" { + t.Fatal("YOW channels:", got) + } + if got := names("YYZ"); got != "#global,#toronto" { + t.Fatal("YYZ channels:", got) + } + if got := names(); got != "#config-east,#global,#mm-yow,#toronto" { + t.Fatal("unfiltered channels:", got) + } + // A 304 keeps membership; a new payload replaces it. + if err := store.SaveChannelCatalogue(ctx, "YOW", "https://x.meshmapper.net/get_channels.php", meshmapper.Cache{AttemptedAt: now, NextAttempt: now}, nil); err != nil { + t.Fatal(err) + } + if got := names("YOW"); got != "#config-east,#global,#mm-yow" { + t.Fatal("304 dropped membership:", got) + } + if err := store.SaveChannelCatalogue(ctx, "YOW", "https://x.meshmapper.net/get_channels.php", meshmapper.Cache{Payload: []byte(`{}`), AttemptedAt: now, NextAttempt: now}, [][]byte{}); err != nil { + t.Fatal(err) + } + if got := names("YOW"); got != "#config-east,#global" { + t.Fatal("empty list kept membership:", got) + } + cats, err := store.ListChannelCatalogues(ctx) + if err != nil || len(cats) != 2 { + t.Fatal(cats, err) + } + if err := store.ClearChannelMembers(ctx); err != nil || names("YYZ") != "#global" { + t.Fatal("members not cleared", err) + } +} diff --git a/db/migrations/001_baseline.sql b/db/migrations/001_baseline.sql index 515df56..06e4335 100644 --- a/db/migrations/001_baseline.sql +++ b/db/migrations/001_baseline.sql @@ -564,6 +564,11 @@ CREATE TABLE analytics_top_talkers_by_iata ( last_sent timestamp with time zone ); +CREATE TABLE channel_config_scopes ( + key_fingerprint bytea NOT NULL, + region_slug text +); + CREATE TABLE channel_iatas ( channel_hash bytea NOT NULL, iata character(3) NOT NULL, @@ -643,6 +648,22 @@ ALTER TABLE known_routes ALTER COLUMN id ADD GENERATED BY DEFAULT AS IDENTITY ( CACHE 1 ); +CREATE TABLE meshmapper_channel_catalogues ( + iata text NOT NULL, + url text NOT NULL, + payload jsonb, + etag text, + checked_at timestamp with time zone, + attempted_at timestamp with time zone NOT NULL, + next_attempt timestamp with time zone NOT NULL, + last_error text DEFAULT ''::text NOT NULL +); + +CREATE TABLE meshmapper_channel_members ( + iata text NOT NULL, + key_fingerprint bytea NOT NULL +); + CREATE TABLE meshmapper_scope_catalogues ( iata text NOT NULL, url text NOT NULL, @@ -1027,6 +1048,12 @@ ALTER TABLE ONLY iata_codes ALTER TABLE ONLY known_routes ADD CONSTRAINT known_routes_pkey PRIMARY KEY (iata, path_key); +ALTER TABLE ONLY meshmapper_channel_catalogues + ADD CONSTRAINT meshmapper_channel_catalogues_pkey PRIMARY KEY (iata, url); + +ALTER TABLE ONLY meshmapper_channel_members + ADD CONSTRAINT meshmapper_channel_members_pkey PRIMARY KEY (iata, key_fingerprint); + ALTER TABLE ONLY meshmapper_scope_catalogues ADD CONSTRAINT meshmapper_scope_catalogues_pkey PRIMARY KEY (iata, url); @@ -1142,6 +1169,10 @@ CREATE INDEX idx_channel_messages_channel ON channel_messages USING btree (chann CREATE INDEX idx_channel_messages_sent_brin ON channel_messages USING brin (sent_at); +CREATE INDEX idx_channel_config_scopes_fingerprint ON channel_config_scopes USING btree (key_fingerprint); + +CREATE INDEX idx_meshmapper_channel_members_fingerprint ON meshmapper_channel_members USING btree (key_fingerprint); + CREATE INDEX idx_channels_hash ON channels USING btree (channel_hash); CREATE UNIQUE INDEX idx_channels_hash_no_key ON channels USING btree (channel_hash) WHERE (key_fingerprint IS NULL); diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 2e99933..addcddb 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -863,6 +863,19 @@ RETURNING id; SELECT packet_hash, raw_payload FROM packets WHERE payload_type = 5 AND decrypted IS NOT TRUE; +-- name: ListUndecryptedGroupTextPacketsByHash :many +-- Like ListUndecryptedGroupTextPackets, limited to channels that just gained a key. +SELECT packet_hash, raw_payload FROM packets +WHERE payload_type = 5 AND decrypted IS NOT TRUE AND channel_hash = ANY(@hashes::bytea[]); + +-- name: DeleteChannelConfigScopes :exec +DELETE FROM channel_config_scopes; + +-- name: AddChannelConfigScopes :exec +-- An empty region places the channel Beacon-wide. +INSERT INTO channel_config_scopes (key_fingerprint, region_slug) +SELECT unnest(@fingerprints::bytea[]), NULLIF(unnest(@regions::text[]), ''); + -- name: UpsertChannelIATA :exec -- Refreshes at most hourly so repeat hears don't churn the row. INSERT INTO channel_iatas (channel_hash, iata, last_heard) @@ -880,15 +893,19 @@ ON CONFLICT (trace_tag, iata) DO UPDATE SET WHERE EXCLUDED.last_heard > trace_iatas.last_heard + INTERVAL '1 hour'; -- name: ListChannels :many --- Channels ordered by last seen, optionally filtered by hash and/or IATAs --- (membership via channel_iatas). NULL hash / empty array skip those filters. +-- Channels ordered by last seen, optionally filtered by hash and/or IATAs. +-- A channel belongs to an IATA when MeshMapper lists it there or config scopes it +-- to a region containing it (or Beacon-wide). NULL hash / empty array skip those filters. -- Pass cursor=0 to start from the beginning (cursor is last_seen epoch ms). SELECT c.* FROM channels c WHERE (@channel_hash::bytea IS NULL OR c.channel_hash = @channel_hash) - AND (COALESCE(cardinality(@iatas::bpchar[]), 0) = 0 OR c.channel_hash IN ( - SELECT ci.channel_hash FROM channel_iatas ci - WHERE ci.iata = ANY(@iatas::bpchar[]) - )) + AND (COALESCE(cardinality(@iatas::bpchar[]), 0) = 0 + OR EXISTS (SELECT 1 FROM meshmapper_channel_members m + WHERE m.key_fingerprint = c.key_fingerprint AND m.iata::bpchar = ANY(@iatas::bpchar[])) + OR EXISTS (SELECT 1 FROM channel_config_scopes s + WHERE s.key_fingerprint = c.key_fingerprint AND (s.region_slug IS NULL OR s.region_slug IN ( + SELECT r.slug FROM regions r JOIN region_iatas ri ON ri.region_id = r.id + WHERE ri.iata = ANY(@iatas::bpchar[]))))) AND (@cursor_ts::timestamptz IS NULL OR c.last_seen < @cursor_ts) ORDER BY c.last_seen DESC, c.id DESC LIMIT @page_limit; @@ -899,9 +916,13 @@ LIMIT @page_limit; SELECT c.* FROM channels c WHERE (c.last_seen, c.id) < (@cursor_ts::timestamptz, @cursor_id::integer) AND (@channel_hash::bytea IS NULL OR c.channel_hash = @channel_hash) - AND (COALESCE(cardinality(@iatas::bpchar[]), 0) = 0 OR c.channel_hash IN ( - SELECT ci.channel_hash FROM channel_iatas ci WHERE ci.iata = ANY(@iatas::bpchar[]) - )) + AND (COALESCE(cardinality(@iatas::bpchar[]), 0) = 0 + OR EXISTS (SELECT 1 FROM meshmapper_channel_members m + WHERE m.key_fingerprint = c.key_fingerprint AND m.iata::bpchar = ANY(@iatas::bpchar[])) + OR EXISTS (SELECT 1 FROM channel_config_scopes s + WHERE s.key_fingerprint = c.key_fingerprint AND (s.region_slug IS NULL OR s.region_slug IN ( + SELECT r.slug FROM regions r JOIN region_iatas ri ON ri.region_id = r.id + WHERE ri.iata = ANY(@iatas::bpchar[]))))) ORDER BY c.last_seen DESC, c.id DESC LIMIT @page_limit; @@ -1628,6 +1649,33 @@ ON CONFLICT (iata, url) DO UPDATE SET next_attempt = EXCLUDED.next_attempt, last_error = EXCLUDED.last_error; +-- name: ListChannelCatalogues :many +SELECT * FROM meshmapper_channel_catalogues ORDER BY iata, attempted_at; + +-- name: SaveChannelCatalogue :exec +-- NULL payload/etag/checked_at retain the last good list after an error or 304. +INSERT INTO meshmapper_channel_catalogues (iata, url, payload, etag, checked_at, attempted_at, next_attempt, last_error) +VALUES (@iata, @url, sqlc.narg(payload)::jsonb, sqlc.narg(etag)::text, + sqlc.narg(checked_at)::timestamptz, @attempted_at, @next_attempt, @last_error) +ON CONFLICT (iata, url) DO UPDATE SET + payload = COALESCE(EXCLUDED.payload, meshmapper_channel_catalogues.payload), + etag = COALESCE(EXCLUDED.etag, meshmapper_channel_catalogues.etag), + checked_at = COALESCE(EXCLUDED.checked_at, meshmapper_channel_catalogues.checked_at), + attempted_at = EXCLUDED.attempted_at, + next_attempt = EXCLUDED.next_attempt, + last_error = EXCLUDED.last_error; + +-- name: DeleteChannelMembersNotIn :exec +DELETE FROM meshmapper_channel_members WHERE iata = @iata AND NOT (key_fingerprint = ANY(@keep::bytea[])); + +-- name: AddChannelMembers :exec +INSERT INTO meshmapper_channel_members (iata, key_fingerprint) +SELECT @iata, unnest(@fingerprints::bytea[]) +ON CONFLICT (iata, key_fingerprint) DO NOTHING; + +-- name: DeleteAllChannelMembers :exec +DELETE FROM meshmapper_channel_members; + -- name: ListZoneBoundaries :many SELECT * FROM meshmapper_zone_boundaries ORDER BY iata; diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index 81e4ff4..7612b07 100644 --- a/db/sqlc/mock/querier.go +++ b/db/sqlc/mock/querier.go @@ -44,6 +44,34 @@ func (m *MockQuerier) EXPECT() *MockQuerierMockRecorder { return m.recorder } +// AddChannelConfigScopes mocks base method. +func (m *MockQuerier) AddChannelConfigScopes(ctx context.Context, arg db.AddChannelConfigScopesParams) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "AddChannelConfigScopes", ctx, arg) + ret0, _ := ret[0].(error) + return ret0 +} + +// AddChannelConfigScopes indicates an expected call of AddChannelConfigScopes. +func (mr *MockQuerierMockRecorder) AddChannelConfigScopes(ctx, arg any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddChannelConfigScopes", reflect.TypeOf((*MockQuerier)(nil).AddChannelConfigScopes), ctx, arg) +} + +// AddChannelMembers mocks base method. +func (m *MockQuerier) AddChannelMembers(ctx context.Context, arg db.AddChannelMembersParams) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "AddChannelMembers", ctx, arg) + ret0, _ := ret[0].(error) + return ret0 +} + +// AddChannelMembers indicates an expected call of AddChannelMembers. +func (mr *MockQuerierMockRecorder) AddChannelMembers(ctx, arg any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddChannelMembers", reflect.TypeOf((*MockQuerier)(nil).AddChannelMembers), ctx, arg) +} + // AddRegionIATAs mocks base method. func (m *MockQuerier) AddRegionIATAs(ctx context.Context, arg db.AddRegionIATAsParams) error { m.ctrl.T.Helper() @@ -103,6 +131,48 @@ func (mr *MockQuerierMockRecorder) DeactivateAccount(ctx, id any) *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeactivateAccount", reflect.TypeOf((*MockQuerier)(nil).DeactivateAccount), ctx, id) } +// DeleteAllChannelMembers mocks base method. +func (m *MockQuerier) DeleteAllChannelMembers(ctx context.Context) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "DeleteAllChannelMembers", ctx) + ret0, _ := ret[0].(error) + return ret0 +} + +// DeleteAllChannelMembers indicates an expected call of DeleteAllChannelMembers. +func (mr *MockQuerierMockRecorder) DeleteAllChannelMembers(ctx any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteAllChannelMembers", reflect.TypeOf((*MockQuerier)(nil).DeleteAllChannelMembers), ctx) +} + +// DeleteChannelConfigScopes mocks base method. +func (m *MockQuerier) DeleteChannelConfigScopes(ctx context.Context) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "DeleteChannelConfigScopes", ctx) + ret0, _ := ret[0].(error) + return ret0 +} + +// DeleteChannelConfigScopes indicates an expected call of DeleteChannelConfigScopes. +func (mr *MockQuerierMockRecorder) DeleteChannelConfigScopes(ctx any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteChannelConfigScopes", reflect.TypeOf((*MockQuerier)(nil).DeleteChannelConfigScopes), ctx) +} + +// DeleteChannelMembersNotIn mocks base method. +func (m *MockQuerier) DeleteChannelMembersNotIn(ctx context.Context, arg db.DeleteChannelMembersNotInParams) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "DeleteChannelMembersNotIn", ctx, arg) + ret0, _ := ret[0].(error) + return ret0 +} + +// DeleteChannelMembersNotIn indicates an expected call of DeleteChannelMembersNotIn. +func (mr *MockQuerierMockRecorder) DeleteChannelMembersNotIn(ctx, arg any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteChannelMembersNotIn", reflect.TypeOf((*MockQuerier)(nil).DeleteChannelMembersNotIn), ctx, arg) +} + // DeleteOldChannelIATAs mocks base method. func (m *MockQuerier) DeleteOldChannelIATAs(ctx context.Context, lastHeard pgtype.Timestamptz) error { m.ctrl.T.Helper() @@ -1027,6 +1097,21 @@ func (mr *MockQuerierMockRecorder) ListAllChannelMessages(ctx, arg any) *gomock. return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListAllChannelMessages", reflect.TypeOf((*MockQuerier)(nil).ListAllChannelMessages), ctx, arg) } +// ListChannelCatalogues mocks base method. +func (m *MockQuerier) ListChannelCatalogues(ctx context.Context) ([]db.MeshmapperChannelCatalogue, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ListChannelCatalogues", ctx) + ret0, _ := ret[0].([]db.MeshmapperChannelCatalogue) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// ListChannelCatalogues indicates an expected call of ListChannelCatalogues. +func (mr *MockQuerierMockRecorder) ListChannelCatalogues(ctx any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListChannelCatalogues", reflect.TypeOf((*MockQuerier)(nil).ListChannelCatalogues), ctx) +} + // ListChannelMessages mocks base method. func (m *MockQuerier) ListChannelMessages(ctx context.Context, arg db.ListChannelMessagesParams) ([]db.ListChannelMessagesRow, error) { m.ctrl.T.Helper() @@ -1342,6 +1427,21 @@ func (mr *MockQuerierMockRecorder) ListUndecryptedGroupTextPackets(ctx any) *gom return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListUndecryptedGroupTextPackets", reflect.TypeOf((*MockQuerier)(nil).ListUndecryptedGroupTextPackets), ctx) } +// ListUndecryptedGroupTextPacketsByHash mocks base method. +func (m *MockQuerier) ListUndecryptedGroupTextPacketsByHash(ctx context.Context, hashes [][]byte) ([]db.ListUndecryptedGroupTextPacketsByHashRow, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ListUndecryptedGroupTextPacketsByHash", ctx, hashes) + ret0, _ := ret[0].([]db.ListUndecryptedGroupTextPacketsByHashRow) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// ListUndecryptedGroupTextPacketsByHash indicates an expected call of ListUndecryptedGroupTextPacketsByHash. +func (mr *MockQuerierMockRecorder) ListUndecryptedGroupTextPacketsByHash(ctx, hashes any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListUndecryptedGroupTextPacketsByHash", reflect.TypeOf((*MockQuerier)(nil).ListUndecryptedGroupTextPacketsByHash), ctx, hashes) +} + // ListZoneBoundaries mocks base method. func (m *MockQuerier) ListZoneBoundaries(ctx context.Context) ([]db.MeshmapperZoneBoundary, error) { m.ctrl.T.Helper() @@ -1661,6 +1761,20 @@ func (mr *MockQuerierMockRecorder) ResolvePathHashesP4(ctx, arg any) *gomock.Cal return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ResolvePathHashesP4", reflect.TypeOf((*MockQuerier)(nil).ResolvePathHashesP4), ctx, arg) } +// SaveChannelCatalogue mocks base method. +func (m *MockQuerier) SaveChannelCatalogue(ctx context.Context, arg db.SaveChannelCatalogueParams) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "SaveChannelCatalogue", ctx, arg) + ret0, _ := ret[0].(error) + return ret0 +} + +// SaveChannelCatalogue indicates an expected call of SaveChannelCatalogue. +func (mr *MockQuerierMockRecorder) SaveChannelCatalogue(ctx, arg any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SaveChannelCatalogue", reflect.TypeOf((*MockQuerier)(nil).SaveChannelCatalogue), ctx, arg) +} + // SaveScopeCatalogue mocks base method. func (m *MockQuerier) SaveScopeCatalogue(ctx context.Context, arg db.SaveScopeCatalogueParams) error { m.ctrl.T.Helper() diff --git a/db/sqlc/models.go b/db/sqlc/models.go index 5ddfc8b..e08581c 100644 --- a/db/sqlc/models.go +++ b/db/sqlc/models.go @@ -188,6 +188,11 @@ type Channel struct { MessageCount *int64 `json:"message_count"` } +type ChannelConfigScope struct { + KeyFingerprint []byte `json:"key_fingerprint"` + RegionSlug *string `json:"region_slug"` +} + type ChannelIata struct { ChannelHash []byte `json:"channel_hash"` Iata string `json:"iata"` @@ -234,6 +239,22 @@ type KnownRoute struct { LastReconfirmedAt pgtype.Timestamptz `json:"last_reconfirmed_at"` } +type MeshmapperChannelCatalogue struct { + Iata string `json:"iata"` + Url string `json:"url"` + Payload []byte `json:"payload"` + Etag *string `json:"etag"` + CheckedAt pgtype.Timestamptz `json:"checked_at"` + AttemptedAt pgtype.Timestamptz `json:"attempted_at"` + NextAttempt pgtype.Timestamptz `json:"next_attempt"` + LastError string `json:"last_error"` +} + +type MeshmapperChannelMember struct { + Iata string `json:"iata"` + KeyFingerprint []byte `json:"key_fingerprint"` +} + type MeshmapperScopeCatalogue struct { Iata string `json:"iata"` Url string `json:"url"` diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index 00d07da..98290d7 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -13,6 +13,9 @@ import ( ) type Querier interface { + // An empty region places the channel Beacon-wide. + AddChannelConfigScopes(ctx context.Context, arg AddChannelConfigScopesParams) error + AddChannelMembers(ctx context.Context, arg AddChannelMembersParams) error AddRegionIATAs(ctx context.Context, arg AddRegionIATAsParams) error // Hop prefixes that match >1 node in an IATA, per width. Computed once per reconfirm run. AmbiguousPrefixes(ctx context.Context) ([]AmbiguousPrefixesRow, error) @@ -20,6 +23,9 @@ type Querier interface { // Lock the current row before deciding the outcome, including when another // deactivation commits while this statement is waiting for its row lock. DeactivateAccount(ctx context.Context, id uuid.UUID) (DeactivateAccountRow, error) + DeleteAllChannelMembers(ctx context.Context) error + DeleteChannelConfigScopes(ctx context.Context) error + DeleteChannelMembersNotIn(ctx context.Context, arg DeleteChannelMembersNotInParams) error // Keeps the channel IATA filter in step with packet retention. DeleteOldChannelIATAs(ctx context.Context, lastHeard pgtype.Timestamptz) error // Deletes nodes not seen since the given cutoff. node_iatas and node_neighbors cascade- @@ -144,6 +150,7 @@ type Querier interface { // Pass empty string for iata or scope to skip those filters. // Pass cursor=0 to start from the beginning. ListAllChannelMessages(ctx context.Context, arg ListAllChannelMessagesParams) ([]ListAllChannelMessagesRow, error) + ListChannelCatalogues(ctx context.Context) ([]MeshmapperChannelCatalogue, error) // Returns messages for a channel identified by integer ID. // Pass a zero/null timestamp for since to return all messages up to limit. // Pass empty string for iata to skip IATA filtering. @@ -154,8 +161,9 @@ type Querier interface { // Pass empty string for iata or scope to skip those filters. // Pass cursor=0 to start from the beginning. ListChannelMessagesByHash(ctx context.Context, arg ListChannelMessagesByHashParams) ([]ListChannelMessagesByHashRow, error) - // Channels ordered by last seen, optionally filtered by hash and/or IATAs - // (membership via channel_iatas). NULL hash / empty array skip those filters. + // Channels ordered by last seen, optionally filtered by hash and/or IATAs. + // A channel belongs to an IATA when MeshMapper lists it there or config scopes it + // to a region containing it (or Beacon-wide). NULL hash / empty array skip those filters. // Pass cursor=0 to start from the beginning (cursor is last_seen epoch ms). ListChannels(ctx context.Context, arg ListChannelsParams) ([]Channel, error) // Keep the non-null tuple boundary separate from the legacy optional cursor so @@ -224,6 +232,8 @@ type Querier interface { // to the config after they'd already been ingested -- see // internal/ingest.BackfillChannelMessages. ListUndecryptedGroupTextPackets(ctx context.Context) ([]ListUndecryptedGroupTextPacketsRow, error) + // Like ListUndecryptedGroupTextPackets, limited to channels that just gained a key. + ListUndecryptedGroupTextPacketsByHash(ctx context.Context, hashes [][]byte) ([]ListUndecryptedGroupTextPacketsByHashRow, error) ListZoneBoundaries(ctx context.Context) ([]MeshmapperZoneBoundary, error) ListZoneLists(ctx context.Context) ([]MeshmapperZoneList, error) PruneImportedRegions(ctx context.Context, keep []string) ([]string, error) @@ -265,6 +275,8 @@ type Querier interface { ResolvePathHashesP2(ctx context.Context, arg ResolvePathHashesP2Params) ([]ResolvePathHashesP2Row, error) ResolvePathHashesP3(ctx context.Context, arg ResolvePathHashesP3Params) ([]ResolvePathHashesP3Row, error) ResolvePathHashesP4(ctx context.Context, arg ResolvePathHashesP4Params) ([]ResolvePathHashesP4Row, error) + // NULL payload/etag/checked_at retain the last good list after an error or 304. + SaveChannelCatalogue(ctx context.Context, arg SaveChannelCatalogueParams) error // One statement commits the validated snapshot and its lookup identities together. // Empty arrays insert nothing. NULL payload/checked_at retain last-known-good data // after an error or 304. Imported names never replace existing manual metadata. diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index d7ee331..d2264de 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -13,6 +13,38 @@ import ( "github.com/jackc/pgx/v5/pgtype" ) +const addChannelConfigScopes = `-- name: AddChannelConfigScopes :exec +INSERT INTO channel_config_scopes (key_fingerprint, region_slug) +SELECT unnest($1::bytea[]), NULLIF(unnest($2::text[]), '') +` + +type AddChannelConfigScopesParams struct { + Fingerprints [][]byte `json:"fingerprints"` + Regions []string `json:"regions"` +} + +// An empty region places the channel Beacon-wide. +func (q *Queries) AddChannelConfigScopes(ctx context.Context, arg AddChannelConfigScopesParams) error { + _, err := q.db.Exec(ctx, addChannelConfigScopes, arg.Fingerprints, arg.Regions) + return err +} + +const addChannelMembers = `-- name: AddChannelMembers :exec +INSERT INTO meshmapper_channel_members (iata, key_fingerprint) +SELECT $1, unnest($2::bytea[]) +ON CONFLICT (iata, key_fingerprint) DO NOTHING +` + +type AddChannelMembersParams struct { + Iata string `json:"iata"` + Fingerprints [][]byte `json:"fingerprints"` +} + +func (q *Queries) AddChannelMembers(ctx context.Context, arg AddChannelMembersParams) error { + _, err := q.db.Exec(ctx, addChannelMembers, arg.Iata, arg.Fingerprints) + return err +} + const addRegionIATAs = `-- name: AddRegionIATAs :exec INSERT INTO region_iatas (region_id, iata) SELECT $1, unnest($2::bpchar[]) @@ -110,6 +142,38 @@ func (q *Queries) DeactivateAccount(ctx context.Context, id uuid.UUID) (Deactiva return i, err } +const deleteAllChannelMembers = `-- name: DeleteAllChannelMembers :exec +DELETE FROM meshmapper_channel_members +` + +func (q *Queries) DeleteAllChannelMembers(ctx context.Context) error { + _, err := q.db.Exec(ctx, deleteAllChannelMembers) + return err +} + +const deleteChannelConfigScopes = `-- name: DeleteChannelConfigScopes :exec +DELETE FROM channel_config_scopes +` + +func (q *Queries) DeleteChannelConfigScopes(ctx context.Context) error { + _, err := q.db.Exec(ctx, deleteChannelConfigScopes) + return err +} + +const deleteChannelMembersNotIn = `-- name: DeleteChannelMembersNotIn :exec +DELETE FROM meshmapper_channel_members WHERE iata = $1 AND NOT (key_fingerprint = ANY($2::bytea[])) +` + +type DeleteChannelMembersNotInParams struct { + Iata string `json:"iata"` + Keep [][]byte `json:"keep"` +} + +func (q *Queries) DeleteChannelMembersNotIn(ctx context.Context, arg DeleteChannelMembersNotInParams) error { + _, err := q.db.Exec(ctx, deleteChannelMembersNotIn, arg.Iata, arg.Keep) + return err +} + const deleteOldChannelIATAs = `-- name: DeleteOldChannelIATAs :exec DELETE FROM channel_iatas WHERE last_heard < $1 ` @@ -2352,6 +2416,39 @@ func (q *Queries) ListAllChannelMessages(ctx context.Context, arg ListAllChannel return items, nil } +const listChannelCatalogues = `-- name: ListChannelCatalogues :many +SELECT iata, url, payload, etag, checked_at, attempted_at, next_attempt, last_error FROM meshmapper_channel_catalogues ORDER BY iata, attempted_at +` + +func (q *Queries) ListChannelCatalogues(ctx context.Context) ([]MeshmapperChannelCatalogue, error) { + rows, err := q.db.Query(ctx, listChannelCatalogues) + if err != nil { + return nil, err + } + defer rows.Close() + items := []MeshmapperChannelCatalogue{} + for rows.Next() { + var i MeshmapperChannelCatalogue + if err := rows.Scan( + &i.Iata, + &i.Url, + &i.Payload, + &i.Etag, + &i.CheckedAt, + &i.AttemptedAt, + &i.NextAttempt, + &i.LastError, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const listChannelMessages = `-- name: ListChannelMessages :many SELECT DISTINCT ON (cm.id) cm.id, cm.channel_id, cm.packet_hash, cm.sender_name, cm.sender_pubkey, cm.content, cm.sent_at, encode(cm.packet_hash, 'hex') as packet_hash_hex, c.channel_hash, ts.name AS scope_name, p.transport_codes_present, (SELECT COUNT(*) FROM packet_observations po2 WHERE po2.packet_hash = cm.packet_hash) AS observation_count @@ -2523,10 +2620,13 @@ func (q *Queries) ListChannelMessagesByHash(ctx context.Context, arg ListChannel const listChannels = `-- name: ListChannels :many SELECT c.id, c.channel_hash, c.key_fingerprint, c.name, c.hashtag, c.is_hashtag, c.is_public, c.key_known, c.first_seen, c.last_seen, c.message_count FROM channels c WHERE ($1::bytea IS NULL OR c.channel_hash = $1) - AND (COALESCE(cardinality($2::bpchar[]), 0) = 0 OR c.channel_hash IN ( - SELECT ci.channel_hash FROM channel_iatas ci - WHERE ci.iata = ANY($2::bpchar[]) - )) + AND (COALESCE(cardinality($2::bpchar[]), 0) = 0 + OR EXISTS (SELECT 1 FROM meshmapper_channel_members m + WHERE m.key_fingerprint = c.key_fingerprint AND m.iata::bpchar = ANY($2::bpchar[])) + OR EXISTS (SELECT 1 FROM channel_config_scopes s + WHERE s.key_fingerprint = c.key_fingerprint AND (s.region_slug IS NULL OR s.region_slug IN ( + SELECT r.slug FROM regions r JOIN region_iatas ri ON ri.region_id = r.id + WHERE ri.iata = ANY($2::bpchar[]))))) AND ($3::timestamptz IS NULL OR c.last_seen < $3) ORDER BY c.last_seen DESC, c.id DESC LIMIT $4 @@ -2539,8 +2639,9 @@ type ListChannelsParams struct { PageLimit int32 `json:"page_limit"` } -// Channels ordered by last seen, optionally filtered by hash and/or IATAs -// (membership via channel_iatas). NULL hash / empty array skip those filters. +// Channels ordered by last seen, optionally filtered by hash and/or IATAs. +// A channel belongs to an IATA when MeshMapper lists it there or config scopes it +// to a region containing it (or Beacon-wide). NULL hash / empty array skip those filters. // Pass cursor=0 to start from the beginning (cursor is last_seen epoch ms). func (q *Queries) ListChannels(ctx context.Context, arg ListChannelsParams) ([]Channel, error) { rows, err := q.db.Query(ctx, listChannels, @@ -2583,9 +2684,13 @@ const listChannelsAfter = `-- name: ListChannelsAfter :many SELECT c.id, c.channel_hash, c.key_fingerprint, c.name, c.hashtag, c.is_hashtag, c.is_public, c.key_known, c.first_seen, c.last_seen, c.message_count FROM channels c WHERE (c.last_seen, c.id) < ($1::timestamptz, $2::integer) AND ($3::bytea IS NULL OR c.channel_hash = $3) - AND (COALESCE(cardinality($4::bpchar[]), 0) = 0 OR c.channel_hash IN ( - SELECT ci.channel_hash FROM channel_iatas ci WHERE ci.iata = ANY($4::bpchar[]) - )) + AND (COALESCE(cardinality($4::bpchar[]), 0) = 0 + OR EXISTS (SELECT 1 FROM meshmapper_channel_members m + WHERE m.key_fingerprint = c.key_fingerprint AND m.iata::bpchar = ANY($4::bpchar[])) + OR EXISTS (SELECT 1 FROM channel_config_scopes s + WHERE s.key_fingerprint = c.key_fingerprint AND (s.region_slug IS NULL OR s.region_slug IN ( + SELECT r.slug FROM regions r JOIN region_iatas ri ON ri.region_id = r.id + WHERE ri.iata = ANY($4::bpchar[]))))) ORDER BY c.last_seen DESC, c.id DESC LIMIT $5 ` @@ -3967,6 +4072,37 @@ func (q *Queries) ListUndecryptedGroupTextPackets(ctx context.Context) ([]ListUn return items, nil } +const listUndecryptedGroupTextPacketsByHash = `-- name: ListUndecryptedGroupTextPacketsByHash :many +SELECT packet_hash, raw_payload FROM packets +WHERE payload_type = 5 AND decrypted IS NOT TRUE AND channel_hash = ANY($1::bytea[]) +` + +type ListUndecryptedGroupTextPacketsByHashRow struct { + PacketHash []byte `json:"packet_hash"` + RawPayload []byte `json:"raw_payload"` +} + +// Like ListUndecryptedGroupTextPackets, limited to channels that just gained a key. +func (q *Queries) ListUndecryptedGroupTextPacketsByHash(ctx context.Context, hashes [][]byte) ([]ListUndecryptedGroupTextPacketsByHashRow, error) { + rows, err := q.db.Query(ctx, listUndecryptedGroupTextPacketsByHash, hashes) + if err != nil { + return nil, err + } + defer rows.Close() + items := []ListUndecryptedGroupTextPacketsByHashRow{} + for rows.Next() { + var i ListUndecryptedGroupTextPacketsByHashRow + if err := rows.Scan(&i.PacketHash, &i.RawPayload); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const listZoneBoundaries = `-- name: ListZoneBoundaries :many SELECT iata, url, feature, etag, checked_at, attempted_at, next_attempt, last_error FROM meshmapper_zone_boundaries ORDER BY iata ` @@ -4574,6 +4710,45 @@ func (q *Queries) ResolvePathHashesP4(ctx context.Context, arg ResolvePathHashes return items, nil } +const saveChannelCatalogue = `-- name: SaveChannelCatalogue :exec +INSERT INTO meshmapper_channel_catalogues (iata, url, payload, etag, checked_at, attempted_at, next_attempt, last_error) +VALUES ($1, $2, $3::jsonb, $4::text, + $5::timestamptz, $6, $7, $8) +ON CONFLICT (iata, url) DO UPDATE SET + payload = COALESCE(EXCLUDED.payload, meshmapper_channel_catalogues.payload), + etag = COALESCE(EXCLUDED.etag, meshmapper_channel_catalogues.etag), + checked_at = COALESCE(EXCLUDED.checked_at, meshmapper_channel_catalogues.checked_at), + attempted_at = EXCLUDED.attempted_at, + next_attempt = EXCLUDED.next_attempt, + last_error = EXCLUDED.last_error +` + +type SaveChannelCatalogueParams struct { + Iata string `json:"iata"` + Url string `json:"url"` + Payload []byte `json:"payload"` + Etag *string `json:"etag"` + CheckedAt pgtype.Timestamptz `json:"checked_at"` + AttemptedAt pgtype.Timestamptz `json:"attempted_at"` + NextAttempt pgtype.Timestamptz `json:"next_attempt"` + LastError string `json:"last_error"` +} + +// NULL payload/etag/checked_at retain the last good list after an error or 304. +func (q *Queries) SaveChannelCatalogue(ctx context.Context, arg SaveChannelCatalogueParams) error { + _, err := q.db.Exec(ctx, saveChannelCatalogue, + arg.Iata, + arg.Url, + arg.Payload, + arg.Etag, + arg.CheckedAt, + arg.AttemptedAt, + arg.NextAttempt, + arg.LastError, + ) + return err +} + const saveScopeCatalogue = `-- name: SaveScopeCatalogue :exec WITH inserted AS ( INSERT INTO transport_scopes (name, transport_key, key_fingerprint, imported_only) diff --git a/docs/docs.go b/docs/docs.go index 4538138..6e59a48 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -597,13 +597,13 @@ const docTemplate = `{ }, { "type": "string", - "description": "Filter by IATA code", + "description": "Filter by IATA code: channels MeshMapper lists there, config channels scoped to a region containing it, and Beacon-wide config channels", "name": "iata", "in": "query" }, { "type": "string", - "description": "Filter by IATA code(s), comma-separated e.g. YOW or YOW,YYZ", + "description": "Filter by IATA code(s), comma-separated e.g. YOW or YOW,YYZ; same membership rule as iata", "name": "iatas", "in": "query" }, diff --git a/docs/swagger.json b/docs/swagger.json index 4cd7655..843fd6b 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -595,13 +595,13 @@ }, { "type": "string", - "description": "Filter by IATA code", + "description": "Filter by IATA code: channels MeshMapper lists there, config channels scoped to a region containing it, and Beacon-wide config channels", "name": "iata", "in": "query" }, { "type": "string", - "description": "Filter by IATA code(s), comma-separated e.g. YOW or YOW,YYZ", + "description": "Filter by IATA code(s), comma-separated e.g. YOW or YOW,YYZ; same membership rule as iata", "name": "iatas", "in": "query" }, diff --git a/docs/swagger.yaml b/docs/swagger.yaml index 4ed5e38..dda333f 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -1903,11 +1903,13 @@ paths: in: query name: hash type: string - - description: Filter by IATA code + - description: 'Filter by IATA code: channels MeshMapper lists there, config + channels scoped to a region containing it, and Beacon-wide config channels' in: query name: iata type: string - - description: Filter by IATA code(s), comma-separated e.g. YOW or YOW,YYZ + - description: Filter by IATA code(s), comma-separated e.g. YOW or YOW,YYZ; + same membership rule as iata in: query name: iatas type: string diff --git a/internal/api/handlers/channels.go b/internal/api/handlers/channels.go index 81ee792..46f312d 100644 --- a/internal/api/handlers/channels.go +++ b/internal/api/handlers/channels.go @@ -35,8 +35,8 @@ func ChannelsRouter(reader api.Reader) http.Handler { // @Tags Channels // @Produce json // @Param hash query string false "Single-byte channel hash (hex)" -// @Param iata query string false "Filter by IATA code" -// @Param iatas query string false "Filter by IATA code(s), comma-separated e.g. YOW or YOW,YYZ" +// @Param iata query string false "Filter by IATA code: channels MeshMapper lists there, config channels scoped to a region containing it, and Beacon-wide config channels" +// @Param iatas query string false "Filter by IATA code(s), comma-separated e.g. YOW or YOW,YYZ; same membership rule as iata" // @Param cursor query int false "last_seen epoch ms of last item for pagination; 0 starts from the beginning" // @Param pageCursor query string false "Opaque nextPageCursor from a previous response; preserves timestamp ties and precision. Cannot be combined with a positive cursor; cursor=0 is allowed." // @Param limit query int false "Max results (default 50); must be positive, values above 200 are clamped" minimum(1) maximum(200) diff --git a/internal/config/channels.go b/internal/config/channels.go new file mode 100644 index 0000000..09216c3 --- /dev/null +++ b/internal/config/channels.go @@ -0,0 +1,66 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package config + +import ( + "encoding/hex" + "fmt" + "strings" + + "github.com/MeshCore-Beacon/beacon-server/internal/keystore" + "gopkg.in/yaml.v3" +) + +// HashtagConfig is a hashtag channel; a plain string in YAML is a Beacon-wide one. +type HashtagConfig struct { + Name string `yaml:"name"` + Region string `yaml:"region"` // optional region slug +} + +func (h *HashtagConfig) UnmarshalYAML(node *yaml.Node) error { + if node.Kind == yaml.ScalarNode { + return node.Decode(&h.Name) + } + type plain HashtagConfig + return node.Decode((*plain)(h)) +} + +// ChannelScope places a configured channel under a region, or Beacon-wide when Region is empty. +type ChannelScope struct { + Fingerprint []byte + Region string +} + +func (c *Config) validateChannelKeys() error { + for i, h := range c.ChannelKeys.Hashtags { + if strings.TrimSpace(h.Name) == "" { + return fmt.Errorf("channel_keys.hashtags[%d] needs a name", i) + } + if h.Region != "" && c.region(h.Region) == nil { + return fmt.Errorf("channel #%s region %q is not a configured region", h.Name, h.Region) + } + } + for hash, k := range c.ChannelKeys.Keys { + if k.Region != "" && c.region(k.Region) == nil { + return fmt.Errorf("channel key %s region %q is not a configured region", hash, k.Region) + } + } + return nil +} + +// ChannelScopes lists every configured channel's region placement. Keys that +// aren't valid hex are skipped; startup already warns about them. +func (c *Config) ChannelScopes() []ChannelScope { + var scopes []ChannelScope + for _, h := range c.ChannelKeys.Hashtags { + _, _, fingerprint := keystore.DeriveHashtagKey(h.Name) + scopes = append(scopes, ChannelScope{Fingerprint: fingerprint, Region: h.Region}) + } + for _, k := range c.ChannelKeys.Keys { + if key, err := hex.DecodeString(k.Key); err == nil { + scopes = append(scopes, ChannelScope{Fingerprint: keystore.Fingerprint(key), Region: k.Region}) + } + } + return scopes +} diff --git a/internal/config/channels_test.go b/internal/config/channels_test.go new file mode 100644 index 0000000..4865670 --- /dev/null +++ b/internal/config/channels_test.go @@ -0,0 +1,88 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package config + +import ( + "bytes" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/MeshCore-Beacon/beacon-server/internal/keystore" +) + +func loadText(t *testing.T, text string) (*Config, error) { + t.Helper() + p := filepath.Join(t.TempDir(), "config.yaml") + if err := os.WriteFile(p, []byte(text), 0600); err != nil { + t.Fatal(err) + } + return Load(p) +} + +func TestChannelKeysRegionScoping(t *testing.T) { + text := `regions: + - slug: east + iatas: [YOW] +channel_keys: + hashtags: + - meshcore + - name: ottawa-mesh + region: east + keys: + "11": + key: "8b3387e9c5cdea6ac9e5edbaa115cd72" + name: Public +` + cfg, err := loadText(t, text) + if err != nil { + t.Fatal(err) + } + if got := cfg.ChannelKeys.Hashtags; len(got) != 2 || got[0].Name != "meshcore" || got[0].Region != "" || got[1].Name != "ottawa-mesh" || got[1].Region != "east" { + t.Fatalf("hashtags parsed as %+v", got) + } + scopes := cfg.ChannelScopes() + if len(scopes) != 3 { + t.Fatalf("scopes: %+v", scopes) + } + _, _, ottawa := keystore.DeriveHashtagKey("ottawa-mesh") + found := false + for _, s := range scopes { + if bytes.Equal(s.Fingerprint, ottawa) { + found = s.Region == "east" + } + } + if !found { + t.Fatal("region-scoped hashtag missing its region", scopes) + } + + for name, bad := range map[string]string{ + "unknown hashtag region": strings.Replace(text, "region: east", "region: west", 1), + "empty hashtag": strings.Replace(text, "- meshcore", `- ""`, 1), + "unknown key region": text + " region: west\n", + } { + if _, err := loadText(t, bad); err == nil { + t.Errorf("%s accepted", name) + } + } +} + +func TestMeshMapperChannelsConfig(t *testing.T) { + valid := "meshmapper:\n channels:\n enabled: true\n" + for text, bad := range map[string]bool{ + valid: false, + valid + " refresh_interval: 24h\n": false, + valid + " refresh_interval: 168h\n": false, + valid + " refresh_interval: 23h\n": true, + valid + " refresh_interval: 169h\n": true, + } { + if _, err := loadText(t, text); (err != nil) != bad { + t.Errorf("%q: error=%v", text, err) + } + } + if (MeshMapperChannelsConfig{}).Interval() != MinChannelsRefresh { + t.Fatal("wrong default") + } +} diff --git a/internal/config/config.go b/internal/config/config.go index 95ebfcb..65abd77 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -320,9 +320,9 @@ func (d *duration) UnmarshalYAML(value *yaml.Node) error { // channel_hash = SHA256(secret)[0]. Explicit keys are provided as hex strings // keyed by the channel hash hex (e.g. "11" for 0x11). type ChannelKeysConfig struct { - // Hashtags is a list of hashtag names (without the # prefix). + // Hashtags lists hashtag names (without the # prefix), each optionally scoped to a region. // Beacon derives the PSK and channel hash automatically. - Hashtags []string `yaml:"hashtags"` + Hashtags []HashtagConfig `yaml:"hashtags"` // Keys maps channel hash hex → explicit key config. Keys map[string]ExplicitKeyConfig `yaml:"keys"` @@ -330,8 +330,9 @@ type ChannelKeysConfig struct { // ExplicitKeyConfig holds an explicit channel key and optional display name. type ExplicitKeyConfig struct { - Key string `yaml:"key"` // hex-encoded key bytes - Name string `yaml:"name"` // optional display name + Key string `yaml:"key"` // hex-encoded key bytes + Name string `yaml:"name"` // optional display name + Region string `yaml:"region"` // optional region slug; empty lists the channel Beacon-wide } // IATAConfig holds optional overrides for a known IATA code. @@ -414,6 +415,9 @@ func Load(path string) (*Config, error) { if err := cfg.validateMeshMapper(); err != nil { return nil, err } + if err := cfg.validateChannelKeys(); err != nil { + return nil, err + } if err := cfg.validateScopes(); err != nil { return nil, err } diff --git a/internal/config/meshmapper.go b/internal/config/meshmapper.go index f76d3bd..8f8a0c7 100644 --- a/internal/config/meshmapper.go +++ b/internal/config/meshmapper.go @@ -12,13 +12,28 @@ import ( // MeshMapper rate-limits get_zones/get_geojson to one request per target per // 23.5h and get_scopes to one per region per 55m; minimums round up. const ( - MinZonesRefresh = 24 * time.Hour - MinScopesRefresh = time.Hour + MinZonesRefresh = 24 * time.Hour + MinScopesRefresh = time.Hour + MinChannelsRefresh = 24 * time.Hour // get_channels.php: one per region per 23.5h ) type MeshMapperConfig struct { - Scopes MeshMapperScopesConfig `yaml:"scopes"` - Zones MeshMapperZonesConfig `yaml:"zones"` + Scopes MeshMapperScopesConfig `yaml:"scopes"` + Zones MeshMapperZonesConfig `yaml:"zones"` + Channels MeshMapperChannelsConfig `yaml:"channels"` +} + +// MeshMapperChannelsConfig imports each known IATA's public hashtag channels. +type MeshMapperChannelsConfig struct { + Enabled bool `yaml:"enabled"` + RefreshInterval duration `yaml:"refresh_interval"` +} + +func (c MeshMapperChannelsConfig) Interval() time.Duration { + if c.RefreshInterval.Duration == 0 { + return MinChannelsRefresh + } + return c.RefreshInterval.Duration } // MeshMapperZonesConfig imports each known IATA's published boundary, overriding borderFile. @@ -59,6 +74,9 @@ func (c *Config) validateMeshMapper() error { return fmt.Errorf("meshmapper.zones.refresh_interval must be between 24h and 168h") } } + if ch := c.MeshMapper.Channels; ch.Enabled && (ch.Interval() < MinChannelsRefresh || ch.Interval() > 7*24*time.Hour) { + return fmt.Errorf("meshmapper.channels.refresh_interval must be between 24h and 168h") + } s := c.MeshMapper.Scopes if !s.Enabled { return nil diff --git a/internal/config/seed.go b/internal/config/seed.go index 86d3249..e08f259 100644 --- a/internal/config/seed.go +++ b/internal/config/seed.go @@ -21,6 +21,7 @@ type Seeder interface { UpsertRegion(ctx context.Context, slug, name, description string, displayOrder int, centerLat, centerLng *float64, zoomLevel *int) (int32, error) SetRegionIATAs(ctx context.Context, regionID int32, iatas []string) error UpsertTransportScope(ctx context.Context, name, displayName string, transportKey, keyFingerprint []byte) error + SetChannelConfigScopes(ctx context.Context, fingerprints [][]byte, regions []string) error } // Seed applies config-defined regions, IATA overrides to the database. @@ -62,6 +63,15 @@ func Seed(ctx context.Context, cfg *Config, db Seeder) error { return err } } + // Channel region placement + var fingerprints [][]byte + var regions []string + for _, scope := range cfg.ChannelScopes() { + fingerprints, regions = append(fingerprints, scope.Fingerprint), append(regions, scope.Region) + } + if err := db.SetChannelConfigScopes(ctx, fingerprints, regions); err != nil { + return err + } // Transport Codes for _, s := range cfg.Scopes { entry := scopestore.FromName(s.Name) diff --git a/internal/config/seed_test.go b/internal/config/seed_test.go index 6241390..c09aa11 100644 --- a/internal/config/seed_test.go +++ b/internal/config/seed_test.go @@ -15,12 +15,13 @@ import ( ) type stubSeeder struct { - regions map[string]int32 - iatas []string - regionIATAs map[int32][]string - scopes []string - borders map[string]json.RawMessage - upsertErr error + regions map[string]int32 + iatas []string + regionIATAs map[int32][]string + channelRegions []string + scopes []string + borders map[string]json.RawMessage + upsertErr error } func newStubSeeder() *stubSeeder { @@ -57,6 +58,11 @@ func (s *stubSeeder) SetRegionIATAs(_ context.Context, regionID int32, iatas []s return s.upsertErr } +func (s *stubSeeder) SetChannelConfigScopes(_ context.Context, fingerprints [][]byte, regions []string) error { + s.channelRegions = regions + return s.upsertErr +} + func (s *stubSeeder) UpsertTransportScope(_ context.Context, name, _ string, _, _ []byte) error { s.scopes = append(s.scopes, name) return s.upsertErr @@ -256,3 +262,17 @@ func TestDeriveScopeKey_DifferentInputs(t *testing.T) { t.Error("expected different keys for different inputs") } } + +func TestSeed_ChannelScopes(t *testing.T) { + cfg := &Config{ + Regions: []RegionConfig{{Slug: "bc", IATAs: []string{"YVR"}}}, + ChannelKeys: ChannelKeysConfig{Hashtags: []HashtagConfig{{Name: "meshcore"}, {Name: "vancouver", Region: "bc"}}}, + } + db := newStubSeeder() + if err := Seed(context.Background(), cfg, db); err != nil { + t.Fatal(err) + } + if len(db.channelRegions) != 2 || db.channelRegions[0] != "" || db.channelRegions[1] != "bc" { + t.Fatalf("channel regions = %q", db.channelRegions) + } +} diff --git a/internal/ingest/backfill.go b/internal/ingest/backfill.go index ecb7ebb..1a41283 100644 --- a/internal/ingest/backfill.go +++ b/internal/ingest/backfill.go @@ -107,6 +107,19 @@ func BackfillChannelMessages(ctx context.Context, db DB, keys ChannelKeyStore) ( if err != nil { return 0, err } + return backfill(ctx, db, keys, packets), nil +} + +// BackfillChannelHashes retries only the channels whose keys were just imported at runtime. +func BackfillChannelHashes(ctx context.Context, db DB, keys ChannelKeyStore, hashes [][]byte) (int, error) { + packets, err := db.ListUndecryptedGroupTextPacketsByHash(ctx, hashes) + if err != nil { + return 0, err + } + return backfill(ctx, db, keys, packets), nil +} + +func backfill(ctx context.Context, db DB, keys ChannelKeyStore, packets []UndecryptedPacket) int { decrypted := 0 for _, p := range packets { result, err := DecryptGroupText(ctx, db, keys, p.PacketHash, p.RawPayload) @@ -118,5 +131,5 @@ func BackfillChannelMessages(ctx context.Context, db DB, keys ChannelKeyStore) ( decrypted++ } } - return decrypted, nil + return decrypted } diff --git a/internal/ingest/backfill_test.go b/internal/ingest/backfill_test.go index 1bbeeef..c50d7a6 100644 --- a/internal/ingest/backfill_test.go +++ b/internal/ingest/backfill_test.go @@ -135,6 +135,23 @@ func TestBackfillChannelMessages_DecryptsAndCounts(t *testing.T) { } } +func TestBackfillChannelHashes_OnlyScansNewHashes(t *testing.T) { + psk := make([]byte, 16) + db := &stubDB{ + insertChannelMessageResult: true, + undecryptedByHash: []UndecryptedPacket{{PacketHash: []byte{0x01}, RawPayload: encryptedGroupText(t, 0x44, psk, "ded", "hi")}}, + undecryptedPackets: []UndecryptedPacket{{PacketHash: []byte{0x02}, RawPayload: encryptedGroupText(t, 0x44, psk, "x", "y")}}, + } + keys := &mapKeys{entries: map[byte][]keystore.Entry{0x44: {{Key: psk, Fingerprint: []byte{0xAA}}}}} + n, err := BackfillChannelHashes(context.Background(), db, keys, [][]byte{{0x44}}) + if err != nil || n != 1 { + t.Fatalf("n=%d err=%v", n, err) + } + if len(db.backfillHashes) != 1 || db.backfillHashes[0][0] != 0x44 { + t.Fatalf("scanned hashes %x", db.backfillHashes) + } +} + func TestBackfillChannelMessages_NoUndecryptedPackets(t *testing.T) { db := &stubDB{} keys := &mapKeys{} diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index 59b9b4e..c9d0dc7 100644 --- a/internal/ingest/ingest.go +++ b/internal/ingest/ingest.go @@ -174,6 +174,9 @@ type DB interface { // used by BackfillChannelMessages to retry them against the current keystore at boot. ListUndecryptedGroupTextPackets(ctx context.Context) ([]UndecryptedPacket, error) + // ListUndecryptedGroupTextPacketsByHash limits that scan to channels that just gained a key. + ListUndecryptedGroupTextPacketsByHash(ctx context.Context, hashes [][]byte) ([]UndecryptedPacket, error) + // UpsertChannelIATA upserts a channel_iatas row. UpsertChannelIATA(ctx context.Context, channelHash []byte, iata string, heardAt time.Time) error diff --git a/internal/ingest/ingest_test.go b/internal/ingest/ingest_test.go index aa8f892..14acd2a 100644 --- a/internal/ingest/ingest_test.go +++ b/internal/ingest/ingest_test.go @@ -184,6 +184,8 @@ type stubDB struct { observationInserted bool insertChannelMessageResult bool // configurable return for InsertChannelMessage; default false undecryptedPackets []UndecryptedPacket + undecryptedByHash []UndecryptedPacket + backfillHashes [][]byte dbHook func() // runs in UpsertChannelIATA, the first DB call after InsertObservation } @@ -279,6 +281,11 @@ func (s *stubDB) ListUndecryptedGroupTextPackets(_ context.Context) ([]Undecrypt return s.undecryptedPackets, nil } +func (s *stubDB) ListUndecryptedGroupTextPacketsByHash(_ context.Context, hashes [][]byte) ([]UndecryptedPacket, error) { + s.backfillHashes = hashes + return s.undecryptedByHash, nil +} + func (s *stubDB) UpsertChannelIATA(_ context.Context, _ []byte, _ string, _ time.Time) error { if s.dbHook != nil { s.dbHook() diff --git a/internal/keystore/keystore.go b/internal/keystore/keystore.go index 9211993..0485442 100644 --- a/internal/keystore/keystore.go +++ b/internal/keystore/keystore.go @@ -2,14 +2,15 @@ // SPDX-License-Identifier: AGPL-3.0-or-later // Package keystore provides channel key lookup for the ingest pipeline. -// Keys are loaded from config at startup and never written at runtime. -// Future: add a DB-backed fallback that checks the channel_keys table. +// Config keys are fixed at startup; MeshMapper imports are swapped in at runtime. package keystore import ( "bytes" "crypto/sha256" "encoding/hex" + "slices" + "sync/atomic" ) // Entry holds a resolved channel key along with its metadata. @@ -23,23 +24,51 @@ type Entry struct { // MapKeyStore maps channel hash hex → list of known key entries. // Multiple entries per hash are supported to handle 1-byte hash collisions. type MapKeyStore struct { - entries map[string][]Entry // channel hash hex → entries + config map[string][]Entry + imported map[string][]Entry + entries atomic.Pointer[map[string][]Entry] // config + imported, read lock-free by ingest } -// NewMapKeyStore creates a keystore from a pre-built map of entries. -// Use BuildFromConfig to construct entries from a config.ChannelKeysConfig. +// NewMapKeyStore creates a keystore from the configured entries. func NewMapKeyStore(entries map[string][]Entry) *MapKeyStore { - store := &MapKeyStore{entries: make(map[string][]Entry)} + store := &MapKeyStore{config: make(map[string][]Entry), imported: map[string][]Entry{}} for hash, list := range entries { - store.entries[hash] = append(store.entries[hash], list...) + store.config[hash] = append(store.config[hash], list...) } + store.entries.Store(&store.config) return store } // GetKey returns all known key entries for the given channel hash byte. // Returns nil if no keys are known for this hash. func (s *MapKeyStore) GetKey(channelHash []byte) []Entry { - return s.entries[hex.EncodeToString(channelHash)] + return (*s.entries.Load())[hex.EncodeToString(channelHash)] +} + +// SetImported replaces the imported keys and returns the channel hashes of keys +// that weren't known before. Config keys win; one writer at a time. +func (s *MapKeyStore) SetImported(entries []Entry) [][]byte { + merged := make(map[string][]Entry, len(s.config)) + for hash, list := range s.config { + merged[hash] = list + } + imported := map[string][]Entry{} + var added [][]byte + for _, e := range entries { + hash := sha256.Sum256(e.Key) + hashHex := hex.EncodeToString(hash[:1]) + if EntryExists(merged[hashHex], e) { + continue + } + merged[hashHex] = append(slices.Clone(merged[hashHex]), e) + imported[hashHex] = append(imported[hashHex], e) + if !EntryExists(s.imported[hashHex], e) { + added = append(added, hash[:1]) + } + } + s.imported = imported + s.entries.Store(&merged) + return added } // DeriveHashtagKey derives the channel secret and hash for a hashtag name. diff --git a/internal/keystore/keystore_test.go b/internal/keystore/keystore_test.go index bed3102..2060c76 100644 --- a/internal/keystore/keystore_test.go +++ b/internal/keystore/keystore_test.go @@ -153,3 +153,33 @@ func TestEntryExists_Empty(t *testing.T) { t.Error("expected false for empty slice") } } + +func TestSetImported_ConfigWinsAndNewHashesReported(t *testing.T) { + secret, hash, fp := DeriveHashtagKey("meshcore") + hashHex := hex.EncodeToString([]byte{hash}) + configured := Entry{Key: secret, Fingerprint: fp, Hashtag: "meshcore", Name: "Configured"} + store := NewMapKeyStore(map[string][]Entry{hashHex: {configured}}) + + dupSecret, _, dupFP := DeriveHashtagKey("meshcore") + newSecret, newHash, newFP := DeriveHashtagKey("ottawa-mesh") + added := store.SetImported([]Entry{ + {Key: dupSecret, Fingerprint: dupFP, Hashtag: "meshcore", Name: "#meshcore"}, + {Key: newSecret, Fingerprint: newFP, Hashtag: "ottawa-mesh", Name: "#ottawa-mesh"}, + }) + if len(added) != 1 || added[0][0] != newHash { + t.Fatalf("new hashes = %x", added) + } + if got := store.GetKey([]byte{hash}); len(got) != 1 || got[0].Name != "Configured" { + t.Fatalf("config entry replaced: %+v", got) + } + if got := store.GetKey([]byte{newHash}); len(got) != 1 || got[0].Hashtag != "ottawa-mesh" { + t.Fatalf("imported entry missing: %+v", got) + } + if added = store.SetImported([]Entry{{Key: newSecret, Fingerprint: newFP, Hashtag: "ottawa-mesh"}}); len(added) != 0 { + t.Fatal("unchanged import reported new hashes", added) + } + store.SetImported(nil) + if store.GetKey([]byte{newHash}) != nil || len(store.GetKey([]byte{hash})) != 1 { + t.Fatal("clearing imports touched config keys or kept the import") + } +} diff --git a/internal/meshmapper/channels.go b/internal/meshmapper/channels.go new file mode 100644 index 0000000..934af8a --- /dev/null +++ b/internal/meshmapper/channels.go @@ -0,0 +1,307 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package meshmapper + +import ( + "bytes" + "context" + "encoding/hex" + "encoding/json" + "fmt" + "log/slog" + "net/http" + "strings" + "time" + "unicode" + "unicode/utf8" + + "github.com/MeshCore-Beacon/beacon-server/internal/config" + "github.com/MeshCore-Beacon/beacon-server/internal/iatadb" + "github.com/MeshCore-Beacon/beacon-server/internal/keystore" +) + +const ( + MaxChannels = 256 // per region + MaxChannelBody = 64 << 10 + + channelRetry = config.MinChannelsRefresh // get_channels.php allows one request per region per 23.5h, failed or not +) + +type ChannelStore interface { + ListKnownIATAs(ctx context.Context) ([]string, error) + ListChannelCatalogues(ctx context.Context) ([]Catalogue, error) + // SaveChannelCatalogue replaces the IATA's members when the cache carries a new payload. + SaveChannelCatalogue(ctx context.Context, iata, url string, cache Cache, fingerprints [][]byte) error + ClearChannelMembers(ctx context.Context) error +} + +// ChannelKeys receives every imported key and reports the channel hashes it hadn't seen. +type ChannelKeys interface { + SetImported(entries []keystore.Entry) [][]byte +} + +type channelSource struct { + iata, country, url string + cache Cache + entries []keystore.Entry +} + +// Channels is owned by one background task. +type Channels struct { + store ChannelStore + keys ChannelKeys + dir *Directory + channelsURL func(site string) (string, bool) + client *http.Client + interval time.Duration + seen map[string]bool + sources []channelSource + onNewKeys func(hashes [][]byte) +} + +// NewChannels restores saved channel lists into keys without HTTP requests. +// 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: 10 * time.Second, + CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }, + }} + if !cfg.Enabled { + if err := store.ClearChannelMembers(ctx); err != nil { + return nil, fmt.Errorf("clear MeshMapper channel members: %w", err) + } + return c, nil + } + saved, err := store.ListChannelCatalogues(ctx) + if err != nil { + return nil, fmt.Errorf("restore channel catalogues: %w", err) + } + latest := map[string]Catalogue{} + for _, cat := range saved { + if old, ok := latest[cat.IATA]; !ok || cat.AttemptedAt.After(old.AttemptedAt) { + latest[cat.IATA] = cat + } + } + iatas, err := store.ListKnownIATAs(ctx) + if err != nil { + return nil, fmt.Errorf("list IATAs for MeshMapper channels: %w", err) + } + c.track(iatas, latest) + c.publish() + return c, nil +} + +// OnNewKeys is told the channel hashes an import added, for backfill. +func (c *Channels) OnNewKeys(fn func(hashes [][]byte)) { c.onNewKeys = fn } + +func (c *Channels) track(iatas []string, saved map[string]Catalogue) { + for _, iata := range iatas { + if c.seen[iata] { + continue + } + c.seen[iata] = true + country := iatadb.CountryFor(iata) + if country == "" { + continue + } + s := channelSource{iata: iata, country: country} + if cat, ok := saved[iata]; ok { + s.url, s.cache = cat.URL, cat.Cache + if len(cat.Payload) > 0 { + var err error + if s.entries, err = decodeChannels(cat.Payload, iata); err != nil { + s.entries, s.cache.Payload, s.cache.ETag = nil, nil, "" + s.cache.LastError = "invalid saved catalogue; awaiting refresh" + } + } + c.log(s, "restored") + } + c.sources = append(c.sources, s) + } +} + +// Refresh makes at most one request. The scheduler serializes calls every 15s. +func (c *Channels) Refresh(ctx context.Context) (err error) { + parent := ctx + defer func() { + if parent.Err() != nil { + err = nil + } + }() + ctx, cancel := context.WithTimeout(ctx, 15*time.Second) + defer cancel() + now := time.Now().UTC() + iatas, err := c.store.ListKnownIATAs(ctx) + if err != nil { + return fmt.Errorf("list IATAs for MeshMapper channels: %w", err) + } + c.track(iatas, nil) + for n := range c.sources { + s := &c.sources[n] + if now.Before(s.cache.NextAttempt) { + continue + } + zones, fetched, err := c.dir.List(ctx, s.country, now) + if err != nil || fetched { + return err + } + if zones == nil { + continue + } + entry, listed := zones[s.iata] + endpoint, valid := "", false + if listed { + endpoint, valid = c.channelsURL(entry.url) + } + if !valid { + // No request to make; channels already imported stay until a list replaces them. + s.cache.NextAttempt, s.cache.LastError = now.Add(c.interval), "not listed" + if listed { + s.cache.LastError = "invalid site URL" + } + c.log(*s, "skipped") + continue + } + if endpoint != s.url { + s.url, s.cache.ETag = endpoint, "" + } + return c.refresh(ctx, s, now) + } + return nil +} + +func (c *Channels) refresh(ctx context.Context, s *channelSource, now time.Time) error { + update := Cache{AttemptedAt: now, NextAttempt: now.Add(c.interval)} + etag := "" + if len(s.cache.Payload) > 0 { + etag = s.cache.ETag + } + status, body, header, err := get(ctx, c.client, "Beacon-MeshMapper-Channels/1", s.url, etag, MaxChannelBody) + if err != nil { + return err + } + var entries []keystore.Entry + var retryAt time.Time + switch status { + case http.StatusOK: + if entries, err = decodeChannels(body, s.iata); err != nil { + update.LastError = "invalid response" + } else { + update.Payload = body + } + case http.StatusNotModified: + if len(s.cache.Payload) == 0 { + update.LastError = "304 without cached catalogue" + } + default: + update.LastError = statusProblem(status, header, now, &retryAt) + } + if update.LastError == "" { + etag := header.Get("ETag") + if len(etag) > 256 || strings.ContainsAny(etag, "\r\n") { + update.LastError, update.Payload, entries = "invalid ETag", nil, nil + } else { + if status == http.StatusNotModified && etag == "" { + etag = s.cache.ETag + } + update.ETag, update.CheckedAt = etag, now + } + } + if update.LastError != "" { + update.NextAttempt = now.Add(min(c.interval, channelRetry)) + if retryAt.After(update.NextAttempt) { + update.NextAttempt = retryAt + } + } + var fingerprints [][]byte + for _, e := range entries { + fingerprints = append(fingerprints, e.Fingerprint) + } + if update.Payload != nil && fingerprints == nil { + fingerprints = [][]byte{} + } + if err := c.store.SaveChannelCatalogue(ctx, s.iata, s.url, update, fingerprints); err != nil { + s.cache.NextAttempt = update.NextAttempt // never retry every tick during a DB outage + return fmt.Errorf("persist channel catalogue %s: %w", s.iata, err) + } + if update.Payload != nil { + s.cache.Payload, s.entries = update.Payload, entries + } + if !update.CheckedAt.IsZero() { + s.cache.CheckedAt, s.cache.ETag = update.CheckedAt, update.ETag + } + s.cache.AttemptedAt, s.cache.NextAttempt, s.cache.LastError = now, update.NextAttempt, update.LastError + if update.Payload != nil { + c.publish() + } + c.log(*s, "checked") + return nil +} + +func (c *Channels) publish() { + var all []keystore.Entry + for _, s := range c.sources { + all = append(all, s.entries...) + } + if added := c.keys.SetImported(all); len(added) > 0 && c.onNewKeys != nil { + c.onNewKeys(added) + } +} + +func (c *Channels) log(s channelSource, action string) { + level := slog.LevelInfo + if s.cache.LastError != "" && s.cache.LastError != "not listed" { + level = slog.LevelWarn + } + slog.Log(context.Background(), level, "MeshMapper channels "+action, "component", "meshmapper.channels", "iata", s.iata, "source", s.url, + "channels", len(s.entries), "checked_at", s.cache.CheckedAt, "next_attempt", s.cache.NextAttempt, "last_error", s.cache.LastError) +} + +func channelsEndpoint(site string) (string, bool) { return siteEndpoint(site, "get_channels.php") } + +// decodeChannels accepts a single region's list only when every key and hash is +// exactly what Beacon derives from the name, so a bad entry can't plant a wrong key. +func decodeChannels(body []byte, iata string) ([]keystore.Entry, error) { + var document struct { + GeneratedAt time.Time `json:"generated_at"` + Region string `json:"region"` + Zones []string `json:"zones"` + Count *int `json:"count"` + Channels *[]struct { + Name string `json:"name"` + Key string `json:"key"` + Hash string `json:"hash"` + Zones []string `json:"zones"` + } `json:"channels"` + } + if len(body) > MaxChannelBody || !utf8.Valid(body) { + return nil, fmt.Errorf("invalid channel list size/encoding") + } + if err := json.Unmarshal(body, &document); err != nil { + return nil, err + } + if document.GeneratedAt.IsZero() || document.Region != iata || len(document.Zones) != 1 || document.Zones[0] != iata || + document.Channels == nil || document.Count == nil || *document.Count != len(*document.Channels) || len(*document.Channels) > MaxChannels { + return nil, fmt.Errorf("invalid regional channel list") + } + entries := make([]keystore.Entry, 0, len(*document.Channels)) + seen := map[string]bool{} + for _, ch := range *document.Channels { + tag, ok := strings.CutPrefix(ch.Name, "#") + if !ok || tag == "" || len(ch.Name) > 64 || strings.ToLower(ch.Name) != ch.Name || seen[ch.Name] || + strings.ContainsFunc(tag, func(r rune) bool { return unicode.IsSpace(r) || unicode.IsControl(r) || r == '#' }) || + len(ch.Zones) != 1 || ch.Zones[0] != iata { + return nil, fmt.Errorf("invalid channel entry") + } + secret, hash, fingerprint := keystore.DeriveHashtagKey(tag) + key, err := hex.DecodeString(ch.Key) + if err != nil || !bytes.Equal(key, secret) || !strings.EqualFold(ch.Hash, hex.EncodeToString([]byte{hash})) { + return nil, fmt.Errorf("channel %s key does not match its name", ch.Name) + } + seen[ch.Name] = true + entries = append(entries, keystore.Entry{Key: secret, Fingerprint: fingerprint, Hashtag: tag, Name: ch.Name}) + } + return entries, nil +} diff --git a/internal/meshmapper/channels_test.go b/internal/meshmapper/channels_test.go new file mode 100644 index 0000000..5a3def5 --- /dev/null +++ b/internal/meshmapper/channels_test.go @@ -0,0 +1,234 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package meshmapper + +import ( + "bytes" + "context" + "encoding/hex" + "encoding/json" + "net/http" + "net/http/httptest" + "slices" + "strings" + "testing" + "time" + + "github.com/MeshCore-Beacon/beacon-server/internal/config" + "github.com/MeshCore-Beacon/beacon-server/internal/keystore" +) + +func channelList(iata string, names ...string) []byte { + rows := []map[string]any{} + for _, name := range names { + secret, hash, _ := keystore.DeriveHashtagKey(strings.TrimPrefix(name, "#")) + rows = append(rows, map[string]any{"name": name, "key": hex.EncodeToString(secret), "hash": hex.EncodeToString([]byte{hash}), "zones": []string{iata}}) + } + b, _ := json.Marshal(map[string]any{"generated_at": "2026-09-30T18:00:00Z", "region": iata, "zones": []string{iata}, "count": len(rows), "channels": rows}) + return b +} + +type channelMemoryStore struct { + iatas []string + rows map[sourceKey]Cache + members map[string][][]byte + fail bool +} + +func newChannelMemoryStore(iatas ...string) *channelMemoryStore { + return &channelMemoryStore{iatas: iatas, rows: map[sourceKey]Cache{}, members: map[string][][]byte{}} +} + +func (s *channelMemoryStore) ListKnownIATAs(context.Context) ([]string, error) { + return slices.Clone(s.iatas), nil +} + +func (s *channelMemoryStore) ListChannelCatalogues(context.Context) ([]Catalogue, error) { + var out []Catalogue + for key, c := range s.rows { + out = append(out, Catalogue{IATA: key.iata, URL: key.url, Cache: c}) + } + return out, nil +} + +func (s *channelMemoryStore) SaveChannelCatalogue(_ context.Context, iata, url string, next Cache, fingerprints [][]byte) error { + if s.fail { + return context.DeadlineExceeded + } + if next.Payload != nil { + s.members[iata] = fingerprints + } + old := s.rows[sourceKey{iata, url}] + if next.Payload == nil { + next.Payload = old.Payload + } + if next.CheckedAt.IsZero() { + next.CheckedAt, next.ETag = old.CheckedAt, old.ETag + } + s.rows[sourceKey{iata, url}] = next + return nil +} + +func (s *channelMemoryStore) ClearChannelMembers(context.Context) error { + s.members = map[string][][]byte{} + return nil +} + +type channelSite struct { + *httptest.Server + list, body string + status int + retryAfter string + listCalls, channels int +} + +func newChannelSite(t *testing.T) *channelSite { + f := &channelSite{status: 200, body: string(channelList("YOW", "#ottawa-mesh"))} + f.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/get_zones.php": + f.listCalls++ + _, _ = w.Write([]byte(f.list)) + case "/get_channels.php": + f.channels++ + w.Header().Set("ETag", `"c1"`) + w.Header().Set("Retry-After", f.retryAfter) + w.WriteHeader(f.status) + if f.status == 200 { + _, _ = w.Write([]byte(f.body)) + } + default: + t.Errorf("unexpected path %s", r.URL.Path) + } + })) + f.list = `{"country":"CA","zones":[{"code":"YOW","url":"` + f.URL + `/","has_boundary":true,"group":null}]}` + t.Cleanup(f.Close) + return f +} + +type channelHarness struct { + c *Channels + keys *keystore.MapKeyStore + added [][]byte +} + +func newChannelHarness(t *testing.T, f *channelSite, store *channelMemoryStore, lists *zoneListMemory, enabled bool) *channelHarness { + t.Helper() + dir := NewDirectory(lists) + dir.listURL = f.URL + "/get_zones.php" + if err := dir.Restore(context.Background()); err != nil { + t.Fatal(err) + } + h := &channelHarness{keys: keystore.NewMapKeyStore(nil)} + c, err := NewChannels(context.Background(), config.MeshMapperChannelsConfig{Enabled: enabled}, store, dir, h.keys) + if err != nil { + t.Fatal(err) + } + c.channelsURL = func(site string) (string, bool) { return site + "get_channels.php", strings.HasPrefix(site, f.URL) } + c.OnNewKeys(func(hashes [][]byte) { h.added = append(h.added, hashes...) }) + h.c = c + return h +} + +func (h *channelHarness) refresh(t *testing.T, n int) { + t.Helper() + for range n { + if err := h.c.Refresh(context.Background()); err != nil { + t.Fatal(err) + } + } +} + +func TestChannelsImportKeysAndMembers(t *testing.T) { + f := newChannelSite(t) + store := newChannelMemoryStore("YOW") + h := newChannelHarness(t, f, store, newZoneListMemory(), true) + h.refresh(t, 2) + secret, hash, fingerprint := keystore.DeriveHashtagKey("ottawa-mesh") + got := h.keys.GetKey([]byte{hash}) + if len(got) != 1 || !bytes.Equal(got[0].Key, secret) || got[0].Name != "#ottawa-mesh" || got[0].Hashtag != "ottawa-mesh" { + t.Fatalf("key not imported: %+v", got) + } + if len(store.members["YOW"]) != 1 || !bytes.Equal(store.members["YOW"][0], fingerprint) { + t.Fatal("YOW membership not saved", store.members) + } + if len(h.added) != 1 || h.added[0][0] != hash { + t.Fatal("new key not reported for backfill", h.added) + } + if b := h.c.sources[0].cache; b.NextAttempt.Sub(b.AttemptedAt) != 24*time.Hour { + t.Fatal("refresh interval", b.NextAttempt.Sub(b.AttemptedAt)) + } +} + +func TestChannelsRejectMismatchKeepsLastGood(t *testing.T) { + f := newChannelSite(t) + store := newChannelMemoryStore("YOW") + h := newChannelHarness(t, f, store, newZoneListMemory(), true) + h.refresh(t, 2) + good := string(channelList("YOW", "#ottawa-mesh")) + secret, hash, _ := keystore.DeriveHashtagKey("ottawa-mesh") + for name, body := range map[string]string{ + "planted key": strings.Replace(good, hex.EncodeToString(secret), strings.Repeat("0", 32), 1), + "wrong key": strings.Replace(string(channelList("YOW", "#other")), `"#other"`, `"#planted"`, 1), + "wrong hash": strings.Replace(good, `"hash":"`+hex.EncodeToString([]byte{hash})+`"`, `"hash":"zz"`, 1), + "other region": string(channelList("YYZ", "#ottawa-mesh")), + "bad count": strings.Replace(good, `"count":1`, `"count":2`, 1), + "uppercase": string(channelList("YOW", "#Ottawa")), + "group zones": strings.Replace(good, `"zones":["YOW"]}`, `"zones":["YOW","YYZ"]}`, 1), + "missing count": strings.Replace(good, `"count":1,`, ``, 1), + } { + f.body = body + h.c.sources[0].cache.NextAttempt = time.Time{} + h.refresh(t, 1) + s := h.c.sources[0] + if s.cache.LastError != "invalid response" || len(h.keys.GetKey([]byte{hash})) != 1 || len(store.members["YOW"]) != 1 { + t.Fatalf("%s: accepted or dropped last good list: %q", name, s.cache.LastError) + } + if s.cache.NextAttempt.Sub(s.cache.AttemptedAt) != 24*time.Hour { + t.Fatalf("%s: retried before the rate limit", name) + } + } +} + +func TestChannelsRateLimitAndFailuresWait(t *testing.T) { + for _, tc := range []struct { + status int + retryAfter string + want time.Duration + }{{503, "", 24 * time.Hour}, {429, "60", 24 * time.Hour}, {429, "172800", 48 * time.Hour}} { + f := newChannelSite(t) + f.status, f.retryAfter = tc.status, tc.retryAfter + h := newChannelHarness(t, f, newChannelMemoryStore("YOW"), newZoneListMemory(), true) + h.refresh(t, 3) + s := h.c.sources[0] + if f.channels != 1 || s.cache.NextAttempt.Sub(s.cache.AttemptedAt) != tc.want { + t.Fatalf("HTTP %d Retry-After %q: calls=%d wait=%v", tc.status, tc.retryAfter, f.channels, s.cache.NextAttempt.Sub(s.cache.AttemptedAt)) + } + } +} + +func TestChannelsRestoreWithoutRequests(t *testing.T) { + f := newChannelSite(t) + store, lists := newChannelMemoryStore("YOW"), newZoneListMemory() + newChannelHarness(t, f, store, lists, true).refresh(t, 2) + restarted := newChannelHarness(t, f, store, lists, true) + _, hash, _ := keystore.DeriveHashtagKey("ottawa-mesh") + if len(restarted.keys.GetKey([]byte{hash})) != 1 { + t.Fatal("saved channels not restored") + } + restarted.refresh(t, 2) + if f.listCalls != 1 || f.channels != 1 { + t.Fatal("restart made requests", f.listCalls, f.channels) + } +} + +func TestChannelsDisabledClearsMembers(t *testing.T) { + f := newChannelSite(t) + store := newChannelMemoryStore("YOW") + store.members["YOW"] = [][]byte{{1}} + h := newChannelHarness(t, f, store, newZoneListMemory(), false) + if len(store.members) != 0 || len(h.c.sources) != 0 { + t.Fatal("disabled import kept membership or sources") + } +}