feat(meshmapper): import regional channels and scope channel lists by region

This commit is contained in:
MrAlders0n committed 2026-09-30 23:09:40 -04:00
1 parent 38b1a206b1
commit f8ee1ac884
30 files changed
+1533 -54

No files matched your search

+6 -1
View File
@@ -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: <slug>} 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.
+27 -2
View File
@@ -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 {
+17
View File
@@ -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: <slug> (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").
+25
View File
@@ -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,
+52
View File
@@ -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)
}
+106
View File
@@ -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)
}
}
+31
View File
@@ -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);
+57 -9
View File
@@ -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;
+114
View File
@@ -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()
+21
View File
@@ -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"`
+14 -2
View File
@@ -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.
+184 -9
View File
@@ -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)
+2 -2
View File
@@ -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"
},
+2 -2
View File
@@ -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"
},
+4 -2
View File
@@ -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
+2 -2
View File
@@ -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)
+66
View File
@@ -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
}
+88
View File
@@ -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")
}
}
+8 -4
View File
@@ -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
}
+22 -4
View File
@@ -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
+10
View File
@@ -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)
+26 -6
View File
@@ -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)
}
}
+14 -1
View File
@@ -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
}
+17
View File
@@ -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{}
+3
View File
@@ -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
+7
View File
@@ -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()
+37 -8
View File
@@ -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.
+30
View File
@@ -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")
}
}
+307
View File
@@ -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
}
+234
View File
@@ -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")
}
}