From dff9c6cf00ed2ae947555b608ef7f55722be0bfa Mon Sep 17 00:00:00 2001 From: MrAlders0n Date: Tue, 21 Jul 2026 10:41:02 -0400 Subject: [PATCH] perf(channels): track channel activity per IATA in its own table The IATA filter ran a correlated EXISTS over packets/observations with ILIKE, which skipped the iata index and took ~7s live. Keep a small channel_iatas table at ingest (like node_iatas) and filter against it. Also honor the iatas= param so multi-site regions stop getting the global list. Filter now ages out with packet retention rather than matching any retained packet. --- db/channels.go | 22 +++++++-- db/channels_test.go | 42 +++++++++++----- db/migrations/011_channel_iatas.sql | 19 ++++++++ db/queries/queries.sql | 31 +++++++----- db/sqlc/mock/querier.go | 28 +++++++++++ db/sqlc/models.go | 6 +++ db/sqlc/querier.go | 8 +-- db/sqlc/queries.sql.go | 59 ++++++++++++++++------- docs/docs.go | 22 ++++++--- docs/swagger.json | 6 +++ docs/swagger.yaml | 4 ++ internal/api/handlers/channels.go | 5 +- internal/api/handlers/channels_test.go | 36 +++++++++++++- internal/api/handlers/stub_reader_test.go | 6 +-- internal/api/reader.go | 5 +- internal/background/tasks.go | 3 ++ internal/cache/cache_test.go | 2 +- internal/cache/reader.go | 4 +- internal/ingest/ingest.go | 3 ++ internal/ingest/ingest_test.go | 9 +++- internal/ingest/packet.go | 6 +++ internal/ingest/side_effects_test.go | 55 +++++++++++++++++++++ 22 files changed, 313 insertions(+), 68 deletions(-) create mode 100644 db/migrations/011_channel_iatas.sql diff --git a/db/channels.go b/db/channels.go index 094940d..9c652fc 100644 --- a/db/channels.go +++ b/db/channels.go @@ -47,16 +47,28 @@ func (s *Store) UpsertChannelHashOnly(ctx context.Context, channelHash []byte) ( return int(rowID), nil } -func (s *Store) ListChannels(ctx context.Context, limit int32, hash []byte, iata string, cursor int64) (api.Page[api.ChannelSummary], error) { +func (s *Store) UpsertChannelIATA(ctx context.Context, channelHash []byte, iata string, heardAt time.Time) error { + return s.q.UpsertChannelIATA(ctx, sqlc.UpsertChannelIATAParams{ + ChannelHash: channelHash, + Iata: iata, + LastHeard: pgtype.Timestamptz{Time: heardAt, Valid: true}, + }) +} + +func (s *Store) DeleteOldChannelIATAs(ctx context.Context, cutoff time.Time) error { + return s.q.DeleteOldChannelIATAs(ctx, pgtype.Timestamptz{Time: cutoff, Valid: true}) +} + +func (s *Store) ListChannels(ctx context.Context, limit int32, hash []byte, iatas []string, cursor int64) (api.Page[api.ChannelSummary], error) { var cursorTS pgtype.Timestamptz if cursor > 0 { cursorTS = pgtype.Timestamptz{Time: time.UnixMilli(cursor), Valid: true} } rows, err := s.q.ListChannels(ctx, sqlc.ListChannelsParams{ - Column1: hash, - Column2: iata, - Column3: cursorTS, - Limit: limit + 1, + ChannelHash: hash, + Iatas: iatas, + CursorTs: cursorTS, + PageLimit: limit + 1, }) if err != nil { return api.Page[api.ChannelSummary]{}, err diff --git a/db/channels_test.go b/db/channels_test.go index 66ee741..8fdea06 100644 --- a/db/channels_test.go +++ b/db/channels_test.go @@ -21,15 +21,15 @@ func TestListChannels_Empty(t *testing.T) { mock.EXPECT(). ListChannels(gomock.Any(), sqlc.ListChannelsParams{ - Column1: nil, - Column2: "", - Column3: pgtype.Timestamptz{}, - Limit: 11, + ChannelHash: nil, + Iatas: nil, + CursorTs: pgtype.Timestamptz{}, + PageLimit: 11, }). Return([]sqlc.Channel{}, nil) store := &Store{q: mock} - page, err := store.ListChannels(context.Background(), 10, nil, "", 0) + page, err := store.ListChannels(context.Background(), 10, nil, nil, 0) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -63,15 +63,15 @@ func TestListChannels_Pagination(t *testing.T) { mock.EXPECT(). ListChannels(gomock.Any(), sqlc.ListChannelsParams{ - Column1: nil, - Column2: "", - Column3: pgtype.Timestamptz{}, - Limit: 3, // limit+1 + ChannelHash: nil, + Iatas: nil, + CursorTs: pgtype.Timestamptz{}, + PageLimit: 3, // limit+1 }). Return(rows, nil) store := &Store{q: mock} - page, err := store.ListChannels(context.Background(), 2, nil, "", 0) + page, err := store.ListChannels(context.Background(), 2, nil, nil, 0) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -95,12 +95,32 @@ func TestListChannels_DBError(t *testing.T) { Return(nil, errors.New("db error")) store := &Store{q: mock} - _, err := store.ListChannels(context.Background(), 10, nil, "", 0) + _, err := store.ListChannels(context.Background(), 10, nil, nil, 0) if err == nil { t.Fatal("expected error, got nil") } } +func TestListChannels_IATAFilter(t *testing.T) { + ctrl := gomock.NewController(t) + mock := mockdb.NewMockQuerier(ctrl) + + mock.EXPECT(). + ListChannels(gomock.Any(), sqlc.ListChannelsParams{ + ChannelHash: nil, + Iatas: []string{"YOW", "YYZ"}, + CursorTs: pgtype.Timestamptz{}, + PageLimit: 11, + }). + Return([]sqlc.Channel{}, nil) + + store := &Store{q: mock} + _, err := store.ListChannels(context.Background(), 10, nil, []string{"YOW", "YYZ"}, 0) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } +} + func TestGetChannel_Basic(t *testing.T) { ctrl := gomock.NewController(t) mock := mockdb.NewMockQuerier(ctrl) diff --git a/db/migrations/011_channel_iatas.sql b/db/migrations/011_channel_iatas.sql new file mode 100644 index 0000000..833a249 --- /dev/null +++ b/db/migrations/011_channel_iatas.sql @@ -0,0 +1,19 @@ +-- Per-IATA channel activity so the IATA filter skips the ~7s EXISTS over packets. +-- Keyed by raw hash (channels can share one), so no FK to channels. + +CREATE TABLE channel_iatas ( + channel_hash BYTEA NOT NULL, + iata CHAR(3) NOT NULL REFERENCES iata_codes(iata) ON DELETE CASCADE, + last_heard TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (channel_hash, iata) +); + +CREATE INDEX idx_channel_iatas_iata ON channel_iatas(iata, last_heard DESC); + +-- Seed from retained packets so the filter works right away. +INSERT INTO channel_iatas (channel_hash, iata, last_heard) +SELECT p.channel_hash, po.iata, MAX(po.heard_at) +FROM packets p +JOIN packet_observations po ON po.packet_hash = p.packet_hash +WHERE p.channel_hash IS NOT NULL +GROUP BY p.channel_hash, po.iata; diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 610525e..57e824f 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -494,6 +494,10 @@ LIMIT $6; -- packet_observations cascade-delete via FK. DELETE FROM packets WHERE last_heard_at < $1; +-- name: DeleteOldChannelIATAs :exec +-- Keeps the channel IATA filter in step with packet retention. +DELETE FROM channel_iatas WHERE last_heard < $1; + -- ============================================================ -- PACKET OBSERVATIONS -- ============================================================ @@ -667,22 +671,25 @@ ON CONFLICT (channel_hash) WHERE key_fingerprint IS NULL DO UPDATE SET last_seen = NOW() RETURNING id; +-- name: UpsertChannelIATA :exec +INSERT INTO channel_iatas (channel_hash, iata, last_heard) +VALUES ($1, $2, $3) +ON CONFLICT (channel_hash, iata) DO UPDATE SET + last_heard = GREATEST(channel_iatas.last_heard, EXCLUDED.last_heard); + -- name: ListChannels :many --- Returns channels ordered by last seen, optionally filtered by hash and/or IATA. --- Pass NULL for hash to skip hash filtering. Pass empty string for iata to skip IATA filtering. --- IATA filter returns channels that have active packets in that IATA (case-insensitive). +-- Channels ordered by last seen, optionally filtered by hash and/or IATAs +-- (membership via channel_iatas). NULL hash / empty array skip those filters. -- Pass cursor=0 to start from the beginning (cursor is last_seen epoch ms). -SELECT DISTINCT c.* FROM channels c -WHERE ($1::bytea IS NULL OR c.channel_hash = $1) - AND ($2 = '' OR EXISTS ( - SELECT 1 FROM packets p - JOIN packet_observations po ON po.packet_hash = p.packet_hash - WHERE p.channel_hash = c.channel_hash - AND po.iata ILIKE $2 +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 ($3::timestamptz IS NULL OR c.last_seen < $3) + AND (@cursor_ts::timestamptz IS NULL OR c.last_seen < @cursor_ts) ORDER BY c.last_seen DESC -LIMIT $4; +LIMIT @page_limit; -- name: GetChannelByID :one SELECT * FROM channels WHERE id = $1; diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index 3122b6c..7ccdc26 100644 --- a/db/sqlc/mock/querier.go +++ b/db/sqlc/mock/querier.go @@ -43,6 +43,20 @@ func (m *MockQuerier) EXPECT() *MockQuerierMockRecorder { return m.recorder } +// DeleteOldChannelIATAs mocks base method. +func (m *MockQuerier) DeleteOldChannelIATAs(ctx context.Context, lastHeard pgtype.Timestamptz) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "DeleteOldChannelIATAs", ctx, lastHeard) + ret0, _ := ret[0].(error) + return ret0 +} + +// DeleteOldChannelIATAs indicates an expected call of DeleteOldChannelIATAs. +func (mr *MockQuerierMockRecorder) DeleteOldChannelIATAs(ctx, lastHeard any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteOldChannelIATAs", reflect.TypeOf((*MockQuerier)(nil).DeleteOldChannelIATAs), ctx, lastHeard) +} + // DeleteOldPackets mocks base method. func (m *MockQuerier) DeleteOldPackets(ctx context.Context, lastHeardAt pgtype.Timestamptz) error { m.ctrl.T.Helper() @@ -1213,6 +1227,20 @@ func (mr *MockQuerierMockRecorder) UpsertChannelHashOnly(ctx, channelHash any) * return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpsertChannelHashOnly", reflect.TypeOf((*MockQuerier)(nil).UpsertChannelHashOnly), ctx, channelHash) } +// UpsertChannelIATA mocks base method. +func (m *MockQuerier) UpsertChannelIATA(ctx context.Context, arg db.UpsertChannelIATAParams) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "UpsertChannelIATA", ctx, arg) + ret0, _ := ret[0].(error) + return ret0 +} + +// UpsertChannelIATA indicates an expected call of UpsertChannelIATA. +func (mr *MockQuerierMockRecorder) UpsertChannelIATA(ctx, arg any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpsertChannelIATA", reflect.TypeOf((*MockQuerier)(nil).UpsertChannelIATA), ctx, arg) +} + // UpsertIATA mocks base method. func (m *MockQuerier) UpsertIATA(ctx context.Context, iata string) error { m.ctrl.T.Helper() diff --git a/db/sqlc/models.go b/db/sqlc/models.go index 833107e..e31b389 100644 --- a/db/sqlc/models.go +++ b/db/sqlc/models.go @@ -23,6 +23,12 @@ type Channel struct { MessageCount *int64 `json:"message_count"` } +type ChannelIata struct { + ChannelHash []byte `json:"channel_hash"` + Iata string `json:"iata"` + LastHeard pgtype.Timestamptz `json:"last_heard"` +} + type ChannelKey struct { ChannelID int32 `json:"channel_id"` KeyBytes []byte `json:"key_bytes"` diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index 9ef83b0..9b05de4 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -12,6 +12,8 @@ import ( ) type Querier interface { + // Keeps the channel IATA filter in step with packet retention. + DeleteOldChannelIATAs(ctx context.Context, lastHeard pgtype.Timestamptz) error // Deletes packets and their observations older than the given cutoff. // packet_observations cascade-delete via FK. DeleteOldPackets(ctx context.Context, lastHeardAt pgtype.Timestamptz) error @@ -97,9 +99,8 @@ 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) - // Returns channels ordered by last seen, optionally filtered by hash and/or IATA. - // Pass NULL for hash to skip hash filtering. Pass empty string for iata to skip IATA filtering. - // IATA filter returns channels that have active packets in that IATA (case-insensitive). + // Channels ordered by last seen, optionally filtered by hash and/or IATAs + // (membership via channel_iatas). 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) ListIATAs(ctx context.Context) ([]IataCode, error) @@ -181,6 +182,7 @@ type Querier interface { // hash-only records (key unknown). Returns the channel row. UpsertChannel(ctx context.Context, arg UpsertChannelParams) (Channel, error) UpsertChannelHashOnly(ctx context.Context, channelHash []byte) (int32, error) + UpsertChannelIATA(ctx context.Context, arg UpsertChannelIATAParams) error // Copyright 2026 Beacon Contributors // SPDX-License-Identifier: agpl // ============================================================ diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index d2332a0..e7080c9 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -12,6 +12,16 @@ import ( "github.com/jackc/pgx/v5/pgtype" ) +const deleteOldChannelIATAs = `-- name: DeleteOldChannelIATAs :exec +DELETE FROM channel_iatas WHERE last_heard < $1 +` + +// Keeps the channel IATA filter in step with packet retention. +func (q *Queries) DeleteOldChannelIATAs(ctx context.Context, lastHeard pgtype.Timestamptz) error { + _, err := q.db.Exec(ctx, deleteOldChannelIATAs, lastHeard) + return err +} + const deleteOldPackets = `-- name: DeleteOldPackets :exec DELETE FROM packets WHERE last_heard_at < $1 ` @@ -1883,13 +1893,11 @@ func (q *Queries) ListChannelMessagesByHash(ctx context.Context, arg ListChannel } const listChannels = `-- name: ListChannels :many -SELECT DISTINCT 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 +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 ($2 = '' OR EXISTS ( - SELECT 1 FROM packets p - JOIN packet_observations po ON po.packet_hash = p.packet_hash - WHERE p.channel_hash = c.channel_hash - AND po.iata ILIKE $2 + 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 ($3::timestamptz IS NULL OR c.last_seen < $3) ORDER BY c.last_seen DESC @@ -1897,22 +1905,21 @@ LIMIT $4 ` type ListChannelsParams struct { - Column1 []byte `json:"column_1"` - Column2 interface{} `json:"column_2"` - Column3 pgtype.Timestamptz `json:"column_3"` - Limit int32 `json:"limit"` + ChannelHash []byte `json:"channel_hash"` + Iatas []string `json:"iatas"` + CursorTs pgtype.Timestamptz `json:"cursor_ts"` + PageLimit int32 `json:"page_limit"` } -// Returns channels ordered by last seen, optionally filtered by hash and/or IATA. -// Pass NULL for hash to skip hash filtering. Pass empty string for iata to skip IATA filtering. -// IATA filter returns channels that have active packets in that IATA (case-insensitive). +// Channels ordered by last seen, optionally filtered by hash and/or IATAs +// (membership via channel_iatas). 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, - arg.Column1, - arg.Column2, - arg.Column3, - arg.Limit, + arg.ChannelHash, + arg.Iatas, + arg.CursorTs, + arg.PageLimit, ) if err != nil { return nil, err @@ -3556,6 +3563,24 @@ func (q *Queries) UpsertChannelHashOnly(ctx context.Context, channelHash []byte) return id, err } +const upsertChannelIATA = `-- name: UpsertChannelIATA :exec +INSERT INTO channel_iatas (channel_hash, iata, last_heard) +VALUES ($1, $2, $3) +ON CONFLICT (channel_hash, iata) DO UPDATE SET + last_heard = GREATEST(channel_iatas.last_heard, EXCLUDED.last_heard) +` + +type UpsertChannelIATAParams struct { + ChannelHash []byte `json:"channel_hash"` + Iata string `json:"iata"` + LastHeard pgtype.Timestamptz `json:"last_heard"` +} + +func (q *Queries) UpsertChannelIATA(ctx context.Context, arg UpsertChannelIATAParams) error { + _, err := q.db.Exec(ctx, upsertChannelIATA, arg.ChannelHash, arg.Iata, arg.LastHeard) + return err +} + const upsertIATA = `-- name: UpsertIATA :exec diff --git a/docs/docs.go b/docs/docs.go index 288eb58..0f8b74d 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -4,11 +4,11 @@ package docs import "github.com/swaggo/swag" const docTemplate = `{ - "schemes": [[ marshal .Schemes ]], + "schemes": {{ marshal .Schemes }}, "swagger": "2.0", "info": { - "description": "[[escape .Description]]", - "title": "[[.Title]]", + "description": "{{escape .Description}}", + "title": "{{.Title}}", "termsOfService": "https://github.com/MeshCore-Beacon/beacon-server", "contact": { "name": "MeshCore Beacon", @@ -17,10 +17,10 @@ const docTemplate = `{ "license": { "name": "AGPL-3-or-later" }, - "version": "[[.Version]]" + "version": "{{.Version}}" }, - "host": "[[.Host]]", - "basePath": "[[.BasePath]]", + "host": "{{.Host}}", + "basePath": "{{.BasePath}}", "paths": { "/brokers": { "get": { @@ -66,6 +66,12 @@ const docTemplate = `{ "name": "iata", "in": "query" }, + { + "type": "string", + "description": "Filter by IATA code(s), comma-separated e.g. YOW or YOW,YYZ", + "name": "iatas", + "in": "query" + }, { "type": "integer", "description": "last_seen epoch ms of last item for pagination", @@ -3768,8 +3774,8 @@ var SwaggerInfo = &swag.Spec{ Description: "MeshCore network observation backend. Ingests LoRa packets from MQTT brokers, stores in PostgreSQL, and streams live events via WebSocket.", InfoInstanceName: "swagger", SwaggerTemplate: docTemplate, - LeftDelim: "[[", - RightDelim: "]]", + LeftDelim: "{{", + RightDelim: "}}", } func init() { diff --git a/docs/swagger.json b/docs/swagger.json index 479b58b..984ff47 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -64,6 +64,12 @@ "name": "iata", "in": "query" }, + { + "type": "string", + "description": "Filter by IATA code(s), comma-separated e.g. YOW or YOW,YYZ", + "name": "iatas", + "in": "query" + }, { "type": "integer", "description": "last_seen epoch ms of last item for pagination", diff --git a/docs/swagger.yaml b/docs/swagger.yaml index d63e219..f846b68 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -1102,6 +1102,10 @@ paths: in: query name: iata type: string + - description: Filter by IATA code(s), comma-separated e.g. YOW or YOW,YYZ + in: query + name: iatas + type: string - description: last_seen epoch ms of last item for pagination in: query name: cursor diff --git a/internal/api/handlers/channels.go b/internal/api/handlers/channels.go index ed0d533..d1652f7 100644 --- a/internal/api/handlers/channels.go +++ b/internal/api/handlers/channels.go @@ -36,6 +36,7 @@ func ChannelsRouter(reader api.Reader) http.Handler { // @Produce json // @Param hash query string false "Single-byte channel hash (hex)" // @Param iata query string false "Filter by IATA code (case-insensitive)" +// @Param iatas query string false "Filter by IATA code(s), comma-separated e.g. YOW or YOW,YYZ" // @Param cursor query int false "last_seen epoch ms of last item for pagination" // @Param limit query int false "Max results (default 50)" // @Success 200 {object} api.Page[api.ChannelSummary] @@ -53,7 +54,7 @@ func listChannels(reader api.Reader) http.HandlerFunc { } limit = l } - iata := r.URL.Query().Get("iata") + iatas := parseIATAs(r) var cursor int64 if cursorParam := r.URL.Query().Get("cursor"); cursorParam != "" { c, err := strconv.ParseInt(cursorParam, 10, 64) @@ -76,7 +77,7 @@ func listChannels(reader api.Reader) http.HandlerFunc { } hashHex = h } - channels, err := reader.ListChannels(r.Context(), int32(limit), hashHex, iata, cursor) + channels, err := reader.ListChannels(r.Context(), int32(limit), hashHex, iatas, cursor) if err != nil { respondError(w, http.StatusInternalServerError, "internal server error") return diff --git a/internal/api/handlers/channels_test.go b/internal/api/handlers/channels_test.go index 360f0da..c014dc4 100644 --- a/internal/api/handlers/channels_test.go +++ b/internal/api/handlers/channels_test.go @@ -7,6 +7,7 @@ import ( "context" "net/http" "net/http/httptest" + "reflect" "testing" "github.com/MeshCore-Beacon/beacon-server/internal/api" @@ -115,7 +116,7 @@ func TestListChannelMessages_InvalidCursor(t *testing.T) { func TestListChannels_OK(t *testing.T) { r := chi.NewRouter() r.Get("/channels", listChannels(stubReader{ - listChannels: func(_ context.Context, _ int32, _ []byte, _ string, _ int64) (api.Page[api.ChannelSummary], error) { + listChannels: func(_ context.Context, _ int32, _ []byte, _ []string, _ int64) (api.Page[api.ChannelSummary], error) { return api.Page[api.ChannelSummary]{Items: []api.ChannelSummary{{ID: 1, ChannelHash: "ab"}}}, nil }, })) @@ -127,6 +128,39 @@ func TestListChannels_OK(t *testing.T) { } } +func TestListChannels_IATAParsing(t *testing.T) { + cases := []struct { + name string + query string + want []string + }{ + {"single lowercased", "?iata=yow", []string{"YOW"}}, + {"multi csv", "?iatas=yow,%20yyz", []string{"YOW", "YYZ"}}, + {"none", "", nil}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + var got []string + r := chi.NewRouter() + r.Get("/channels", listChannels(stubReader{ + listChannels: func(_ context.Context, _ int32, _ []byte, iatas []string, _ int64) (api.Page[api.ChannelSummary], error) { + got = iatas + return api.Page[api.ChannelSummary]{}, nil + }, + })) + req := httptest.NewRequest(http.MethodGet, "/channels"+tc.query, nil) + w := httptest.NewRecorder() + r.ServeHTTP(w, req) + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", w.Code) + } + if !reflect.DeepEqual(got, tc.want) { + t.Errorf("expected iatas %v, got %v", tc.want, got) + } + }) + } +} + func TestGetChannel_OK(t *testing.T) { r := chi.NewRouter() r.Get("/channels/{channelID}", getChannel(stubReader{ diff --git a/internal/api/handlers/stub_reader_test.go b/internal/api/handlers/stub_reader_test.go index 948d57a..cdd575d 100644 --- a/internal/api/handlers/stub_reader_test.go +++ b/internal/api/handlers/stub_reader_test.go @@ -20,7 +20,7 @@ type stubReader struct { listRegions func(ctx context.Context) ([]api.RegionSummary, error) getRegion func(ctx context.Context, regionID int32) (*api.Region, error) getRegionBySlug func(ctx context.Context, slug string) (*api.Region, error) - listChannels func(ctx context.Context, limit int32, hash []byte, iata string, cursor int64) (api.Page[api.ChannelSummary], error) + listChannels func(ctx context.Context, limit int32, hash []byte, iatas []string, cursor int64) (api.Page[api.ChannelSummary], error) getChannel func(ctx context.Context, channelID int32) (*api.Channel, error) listChannelMessages func(ctx context.Context, channelID *int32, since time.Time, limit int32, iatas []string, scope string, cursor int64) (api.Page[api.ChannelMessage], error) listChannelMessagesByHash func(ctx context.Context, hash []byte, since time.Time, limit int32, iatas []string, scope string, cursor int64) (api.Page[api.ChannelMessage], error) @@ -96,9 +96,9 @@ func (s stubReader) GetRegionBySlug(ctx context.Context, slug string) (*api.Regi return nil, nil } -func (s stubReader) ListChannels(ctx context.Context, limit int32, hash []byte, iata string, cursor int64) (api.Page[api.ChannelSummary], error) { +func (s stubReader) ListChannels(ctx context.Context, limit int32, hash []byte, iatas []string, cursor int64) (api.Page[api.ChannelSummary], error) { if s.listChannels != nil { - return s.listChannels(ctx, limit, hash, iata, cursor) + return s.listChannels(ctx, limit, hash, iatas, cursor) } return api.Page[api.ChannelSummary]{}, nil } diff --git a/internal/api/reader.go b/internal/api/reader.go index ad95e58..9309f57 100644 --- a/internal/api/reader.go +++ b/internal/api/reader.go @@ -44,9 +44,10 @@ type Reader interface { // ListChannels returns a paginated list of channels ordered by last seen. // Includes both hashtag-derived and explicit key channels. - // Pass nil hash to skip hash filtering. Pass empty string iata to return all channels. + // Pass nil hash to skip hash filtering. Pass empty iatas to return all channels; + // IATAs must be uppercase. // cursor is last_seen epoch ms of the last item; pass 0 to start from the beginning. - ListChannels(ctx context.Context, limit int32, hash []byte, iata string, cursor int64) (Page[ChannelSummary], error) + ListChannels(ctx context.Context, limit int32, hash []byte, iatas []string, cursor int64) (Page[ChannelSummary], error) // GetChannel returns full detail for a single channel by its integer ID. // Returns nil, pgx.ErrNoRows if the channel is not found. diff --git a/internal/background/tasks.go b/internal/background/tasks.go index 6006cce..559a3dd 100644 --- a/internal/background/tasks.go +++ b/internal/background/tasks.go @@ -44,6 +44,9 @@ func CleanupTask(store *db.Store, telemetryRetention, packetRetention, interval if err := store.DeleteOldPackets(ctx, time.Now().Add(-packetRetention)); err != nil { return err } + if err := store.DeleteOldChannelIATAs(ctx, time.Now().Add(-packetRetention)); err != nil { + return err + } return nil }, } diff --git a/internal/cache/cache_test.go b/internal/cache/cache_test.go index 8ab4337..7ef6954 100644 --- a/internal/cache/cache_test.go +++ b/internal/cache/cache_test.go @@ -128,7 +128,7 @@ func (s *stubReader) GetCrossIATANeighbors(_ context.Context, _ uuid.UUID, _ str return nil, nil } -func (s *stubReader) ListChannels(_ context.Context, _ int32, _ []byte, _ string, _ int64) (api.Page[api.ChannelSummary], error) { +func (s *stubReader) ListChannels(_ context.Context, _ int32, _ []byte, _ []string, _ int64) (api.Page[api.ChannelSummary], error) { return api.Page[api.ChannelSummary]{}, nil } diff --git a/internal/cache/reader.go b/internal/cache/reader.go index 46d001e..4bdfd8c 100644 --- a/internal/cache/reader.go +++ b/internal/cache/reader.go @@ -354,8 +354,8 @@ func (cr *CachedReader) GetCrossIATANeighbors(ctx context.Context, nodeID uuid.U } // ListChannels implements [api.Reader]. -func (cr *CachedReader) ListChannels(ctx context.Context, limit int32, hash []byte, iata string, cursor int64) (api.Page[api.ChannelSummary], error) { - return cr.inner.ListChannels(ctx, limit, hash, iata, cursor) +func (cr *CachedReader) ListChannels(ctx context.Context, limit int32, hash []byte, iatas []string, cursor int64) (api.Page[api.ChannelSummary], error) { + return cr.inner.ListChannels(ctx, limit, hash, iatas, cursor) } // ListChannelMessages implements [api.Reader]. diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index 5bf7453..48310aa 100644 --- a/internal/ingest/ingest.go +++ b/internal/ingest/ingest.go @@ -151,6 +151,9 @@ type DB interface { // but can be safely ignored since unknown-key channels have no messages. UpsertChannelHashOnly(ctx context.Context, channelHash []byte) (int, error) + // UpsertChannelIATA upserts a channel_iatas row. + UpsertChannelIATA(ctx context.Context, channelHash []byte, iata string, heardAt time.Time) error + // GetPacketObservationCount returns the number of rows for the packet observations GetPacketObservationCount(ctx context.Context, packetHash []byte) (int64, error) diff --git a/internal/ingest/ingest_test.go b/internal/ingest/ingest_test.go index 15c54fc..81943d5 100644 --- a/internal/ingest/ingest_test.go +++ b/internal/ingest/ingest_test.go @@ -177,6 +177,8 @@ type stubDB struct { upsertNodeCalls int upsertChannelCalls int upsertChannelHashOnlyCalls int + upsertChannelIATACalls int + observationInserted bool } type setCapabilityCall struct { @@ -201,7 +203,7 @@ func (s *stubDB) UpsertPacket(_ context.Context, _ UpsertPacketParams) (bool, er } func (s *stubDB) SetPacketDecrypted(_ context.Context, _ []byte) error { return nil } func (s *stubDB) InsertObservation(_ context.Context, _ InsertObservationParams) (bool, error) { - return false, nil + return s.observationInserted, nil } func (s *stubDB) SetNodeDefaultScope(_ context.Context, _ uuid.UUID, _ int32) error { return nil } func (s *stubDB) UpsertNode(_ context.Context, _ UpsertNodeParams, _ RadioSettings) (uuid.UUID, error) { @@ -257,6 +259,11 @@ func (s *stubDB) UpsertChannelHashOnly(_ context.Context, _ []byte) (int, error) return 0, nil } +func (s *stubDB) UpsertChannelIATA(_ context.Context, _ []byte, _ string, _ time.Time) error { + s.upsertChannelIATACalls++ + return nil +} + func (s *stubDB) GetPacketObservationCount(_ context.Context, _ []byte) (int64, error) { return 0, nil } diff --git a/internal/ingest/packet.go b/internal/ingest/packet.go index bf64f29..3ff8d3b 100644 --- a/internal/ingest/packet.go +++ b/internal/ingest/packet.go @@ -783,6 +783,12 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ } } + if channelHash != nil && inserted { + if err := w.db.UpsertChannelIATA(ctx, channelHash, iata, heardAt); err != nil { + log.Printf("ingest[%s]: db: upsert channel IATA failed from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err) + } + } + // packet.PathHashes() reads packet.Path as hash-sized chunks, which is only true for // ordinary flood/direct-routed packets. TRACE repurposes packet.Path to carry one SNR // byte per hop instead, so for TRACE we resolve against the trace payload's own diff --git a/internal/ingest/side_effects_test.go b/internal/ingest/side_effects_test.go index acf0e56..a62a954 100644 --- a/internal/ingest/side_effects_test.go +++ b/internal/ingest/side_effects_test.go @@ -6,7 +6,10 @@ package ingest import ( "context" "crypto/ed25519" + "encoding/hex" + "encoding/json" "testing" + "time" "github.com/MeshCore-Beacon/beacon-server/internal/keystore" "github.com/meshcore-go/meshcore-go" @@ -133,3 +136,55 @@ func TestHandlePayloadTypeSideEffects_GrpTxt_UnknownKey_OnlyUpsertsHashOnlyChann t.Errorf("expected UpsertChannel NOT to be called when the key is unknown, got %d calls", db.upsertChannelCalls) } } + +// packetEnvelope wraps a packet in the minimal broker JSON that handlePacket expects. +func packetEnvelope(t *testing.T, packet *meshcore.Packet) []byte { + t.Helper() + raw, err := packet.ToBytes() + if err != nil { + t.Fatalf("packet to bytes: %v", err) + } + env, err := json.Marshal(map[string]string{ + "raw": hex.EncodeToString(raw), + "timestamp": time.Now().UTC().Format("2006-01-02T15:04:05.000000"), + }) + if err != nil { + t.Fatalf("marshal envelope: %v", err) + } + return env +} + +func TestHandlePacket_GrpTxt_UpsertsChannelIATA(t *testing.T) { + w, db := newTestWorker() + db.observationInserted = true + envelope := packetEnvelope(t, buildGrpTxtPacket(t, 0x1a, make([]byte, 16))) + + w.handlePacket(context.Background(), "YOW", "0102", envelope) + + if db.upsertChannelIATACalls != 1 { + t.Errorf("expected UpsertChannelIATA to be called once for a stored group text, got %d", db.upsertChannelIATACalls) + } +} + +func TestHandlePacket_GrpTxt_DedupObservation_SkipsChannelIATA(t *testing.T) { + w, db := newTestWorker() // stub reports the observation as a duplicate + envelope := packetEnvelope(t, buildGrpTxtPacket(t, 0x1a, make([]byte, 16))) + + w.handlePacket(context.Background(), "YOW", "0102", envelope) + + if db.upsertChannelIATACalls != 0 { + t.Errorf("expected UpsertChannelIATA NOT to be called for a duplicate observation, got %d calls", db.upsertChannelIATACalls) + } +} + +func TestHandlePacket_Advert_SkipsChannelIATA(t *testing.T) { + w, db := newTestWorker() + db.observationInserted = true + envelope := packetEnvelope(t, buildAdvertPacket(t, false)) + + w.handlePacket(context.Background(), "YOW", "0102", envelope) + + if db.upsertChannelIATACalls != 0 { + t.Errorf("expected UpsertChannelIATA NOT to be called for a non-channel packet, got %d calls", db.upsertChannelIATACalls) + } +}