Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 13 additions & 7 deletions db/channel_cursor_integration_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -57,13 +57,14 @@ func TestChannelCursorPostgres(t *testing.T) {
}
}
applyBaseline(t, ctx, tx)
exec(`INSERT INTO channels(id,channel_hash,key_fingerprint,last_seen) VALUES
(1,'\xaa','\x01','2026-09-08 12:00:00+00'),(2,'\xaa','\x02','2026-09-08 12:00:00+00'),
(3,'\xaa','\x03','2026-09-08 12:00:00+00'),(4,'\xbb','\x04','2026-09-08 11:59:59+00'),
(5,'\xcc','\x05','2026-09-08 12:00:00.000321+00'),
(6,'\xcc','\x06','2026-09-08 12:00:00.000111+00'),(7,'\xcc','\x07','2026-09-08 12:00:00.000111+00');
INSERT INTO channel_iatas(channel_hash,iata,last_heard) VALUES
('\xaa','YOW',now()),('\xbb','YOW',now()),('\xcc','YYZ',now());`)
// NULL key_known is reported as false, so keyKnown=false must include it.
exec(`INSERT INTO channels(id,channel_hash,key_fingerprint,last_seen,key_known) VALUES
(1,'\xaa','\x01','2026-09-08 12:00:00+00',true),(2,'\xaa','\x02','2026-09-08 12:00:00+00',false),
(3,'\xaa','\x03','2026-09-08 12:00:00+00',NULL),(4,'\xbb','\x04','2026-09-08 11:59:59+00',true),
(5,'\xcc','\x05','2026-09-08 12:00:00.000321+00',true),
(6,'\xcc','\x06','2026-09-08 12:00:00.000111+00',false),(7,'\xcc','\x07','2026-09-08 12:00:00.000111+00',true);
INSERT INTO meshmapper_channel_members(iata,key_fingerprint) VALUES
('YOW','\x01'),('YOW','\x02'),('YOW','\x03'),('YOW','\x04'),('YYZ','\x05'),('YYZ','\x06'),('YYZ','\x07');`)
queries := &channelCursorQueries{DBTX: tx}
router := handlers.ChannelsRouter(&Store{q: sqlc.New(queries)})
// Decode the wire contract so the same regression runs against the old server.
Expand Down Expand Up @@ -97,6 +98,11 @@ INSERT INTO channel_iatas(channel_hash,iata,last_heard) VALUES
{"hash and region", "hash=cc&iata=yyz", []int{5, 7, 6}},
{"no regional match", "hash=aa&iata=yyz", nil},
{"missing region", "iata=ZZZ", nil},
{"key known", "keyKnown=true", []int{5, 7, 1, 4}},
{"key unknown", "keyKnown=false", []int{6, 3, 2}},
{"key known in region", "keyKnown=true&iata=yow", []int{1, 4}},
{"key unknown in region", "keyKnown=false&iata=yow", []int{3, 2}},
{"key unknown hash", "keyKnown=false&hash=cc", []int{6}},
} {
t.Run(tc.name, func(t *testing.T) {
params, _ := url.ParseQuery(tc.query)
Expand Down
6 changes: 3 additions & 3 deletions db/channels.go
Original file line number Diff line number Diff line change
Expand Up @@ -102,7 +102,7 @@ func (s *Store) DeleteOldChannelIATAs(ctx context.Context, cutoff time.Time) err
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, pageCursor *api.ChannelCursor) (api.ChannelPage, error) {
func (s *Store) ListChannels(ctx context.Context, limit int32, hash []byte, iatas []string, keyKnown *bool, cursor int64, pageCursor *api.ChannelCursor) (api.ChannelPage, error) {
var cursorTS pgtype.Timestamptz
if cursor > 0 {
cursorTS = pgtype.Timestamptz{Time: time.UnixMilli(cursor), Valid: true}
Expand All @@ -111,12 +111,12 @@ func (s *Store) ListChannels(ctx context.Context, limit int32, hash []byte, iata
var err error
if pageCursor != nil {
rows, err = s.q.ListChannelsAfter(ctx, sqlc.ListChannelsAfterParams{
ChannelHash: hash, Iatas: iatas, PageLimit: limit + 1,
ChannelHash: hash, Iatas: iatas, KeyKnown: keyKnown, PageLimit: limit + 1,
CursorTs: pgtype.Timestamptz{Time: pageCursor.LastSeen, Valid: true}, CursorID: pageCursor.ID,
})
} else {
rows, err = s.q.ListChannels(ctx, sqlc.ListChannelsParams{
ChannelHash: hash, Iatas: iatas, CursorTs: cursorTS, PageLimit: limit + 1,
ChannelHash: hash, Iatas: iatas, KeyKnown: keyKnown, CursorTs: cursorTS, PageLimit: limit + 1,
})
}
if err != nil {
Expand Down
34 changes: 30 additions & 4 deletions db/channels_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import (

sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc"
mockdb "github.com/MeshCore-Beacon/beacon-server/db/sqlc/mock"
"github.com/MeshCore-Beacon/beacon-server/internal/api"
"github.com/jackc/pgx/v5/pgtype"
"go.uber.org/mock/gomock"
)
Expand All @@ -29,7 +30,7 @@ func TestListChannels_Empty(t *testing.T) {
Return([]sqlc.Channel{}, nil)

store := &Store{q: mock}
page, err := store.ListChannels(context.Background(), 10, nil, nil, 0, nil)
page, err := store.ListChannels(context.Background(), 10, nil, nil, nil, 0, nil)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
Expand Down Expand Up @@ -71,7 +72,7 @@ func TestListChannels_Pagination(t *testing.T) {
Return(rows, nil)

store := &Store{q: mock}
page, err := store.ListChannels(context.Background(), 2, nil, nil, 0, nil)
page, err := store.ListChannels(context.Background(), 2, nil, nil, nil, 0, nil)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
Expand All @@ -95,7 +96,7 @@ func TestListChannels_DBError(t *testing.T) {
Return(nil, errors.New("db error"))

store := &Store{q: mock}
_, err := store.ListChannels(context.Background(), 10, nil, nil, 0, nil)
_, err := store.ListChannels(context.Background(), 10, nil, nil, nil, 0, nil)
if err == nil {
t.Fatal("expected error, got nil")
}
Expand All @@ -115,12 +116,37 @@ func TestListChannels_IATAFilter(t *testing.T) {
Return([]sqlc.Channel{}, nil)

store := &Store{q: mock}
_, err := store.ListChannels(context.Background(), 10, nil, []string{"YOW", "YYZ"}, 0, nil)
_, err := store.ListChannels(context.Background(), 10, nil, []string{"YOW", "YYZ"}, nil, 0, nil)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
}

func TestListChannels_KeyKnownFilter(t *testing.T) {
ctrl := gomock.NewController(t)
mock := mockdb.NewMockQuerier(ctrl)
keyKnown := false
at := time.UnixMilli(1700000000000)

mock.EXPECT().
ListChannels(gomock.Any(), sqlc.ListChannelsParams{KeyKnown: &keyKnown, PageLimit: 11}).
Return([]sqlc.Channel{}, nil)
mock.EXPECT().
ListChannelsAfter(gomock.Any(), sqlc.ListChannelsAfterParams{
KeyKnown: &keyKnown, PageLimit: 11,
CursorTs: pgtype.Timestamptz{Time: at, Valid: true}, CursorID: 7,
}).
Return([]sqlc.Channel{}, nil)

store := &Store{q: mock}
if _, err := store.ListChannels(context.Background(), 10, nil, nil, &keyKnown, 0, nil); err != nil {
t.Fatalf("unexpected error: %v", err)
}
if _, err := store.ListChannels(context.Background(), 10, nil, nil, &keyKnown, 0, &api.ChannelCursor{LastSeen: at, ID: 7}); err != nil {
t.Fatalf("unexpected error: %v", err)
}
}

func TestGetChannel_Basic(t *testing.T) {
ctrl := gomock.NewController(t)
mock := mockdb.NewMockQuerier(ctrl)
Expand Down
2 changes: 1 addition & 1 deletion db/meshmapper_integration_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -410,7 +410,7 @@ func TestChannelRegionScopingPostgres(t *testing.T) {
}
names := func(iatas ...string) string {
t.Helper()
page, err := store.ListChannels(ctx, 50, nil, iatas, 0, nil)
page, err := store.ListChannels(ctx, 50, nil, iatas, nil, 0, nil)
if err != nil {
t.Fatal(err)
}
Expand Down
4 changes: 3 additions & 1 deletion db/queries/queries.sql
Original file line number Diff line number Diff line change
Expand Up @@ -900,7 +900,7 @@ 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.
-- 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.
-- to a region containing it (or Beacon-wide). NULL hash / empty array / NULL key_known 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)
Expand All @@ -911,6 +911,7 @@ WHERE (@channel_hash::bytea IS NULL OR c.channel_hash = @channel_hash)
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 (sqlc.narg(key_known)::boolean IS NULL OR COALESCE(c.key_known, false) = sqlc.narg(key_known))
AND (@cursor_ts::timestamptz IS NULL OR c.last_seen < @cursor_ts)
ORDER BY c.last_seen DESC, c.id DESC
LIMIT @page_limit;
Expand All @@ -928,6 +929,7 @@ WHERE (c.last_seen, c.id) < (@cursor_ts::timestamptz, @cursor_id::integer)
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 (sqlc.narg(key_known)::boolean IS NULL OR COALESCE(c.key_known, false) = sqlc.narg(key_known))
ORDER BY c.last_seen DESC, c.id DESC
LIMIT @page_limit;

Expand Down
2 changes: 1 addition & 1 deletion db/sqlc/querier.go

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

14 changes: 10 additions & 4 deletions db/sqlc/queries.sql.go

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

6 changes: 6 additions & 0 deletions docs/docs.go
Original file line number Diff line number Diff line change
Expand Up @@ -607,6 +607,12 @@ const docTemplate = `{
"name": "iatas",
"in": "query"
},
{
"type": "boolean",
"description": "Only channels Beacon can (true) or cannot (false) decrypt; omit for all",
"name": "keyKnown",
"in": "query"
},
{
"type": "integer",
"description": "last_seen epoch ms of last item for pagination; 0 starts from the beginning",
Expand Down
6 changes: 6 additions & 0 deletions docs/swagger.json
Original file line number Diff line number Diff line change
Expand Up @@ -605,6 +605,12 @@
"name": "iatas",
"in": "query"
},
{
"type": "boolean",
"description": "Only channels Beacon can (true) or cannot (false) decrypt; omit for all",
"name": "keyKnown",
"in": "query"
},
{
"type": "integer",
"description": "last_seen epoch ms of last item for pagination; 0 starts from the beginning",
Expand Down
5 changes: 5 additions & 0 deletions docs/swagger.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -1913,6 +1913,11 @@ paths:
in: query
name: iatas
type: string
- description: Only channels Beacon can (true) or cannot (false) decrypt; omit
for all
in: query
name: keyKnown
type: boolean
- description: last_seen epoch ms of last item for pagination; 0 starts from
the beginning
in: query
Expand Down
15 changes: 14 additions & 1 deletion internal/api/handlers/channels.go
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@ func ChannelsRouter(reader api.Reader) http.Handler {
// @Param hash query string false "Single-byte channel hash (hex)"
// @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 keyKnown query bool false "Only channels Beacon can (true) or cannot (false) decrypt; omit for all"
// @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)
Expand Down Expand Up @@ -86,7 +87,19 @@ func listChannels(reader api.Reader) http.HandlerFunc {
}
hashHex = h
}
channels, err := reader.ListChannels(r.Context(), limit, hashHex, iatas, cursor, pageCursor)
var keyKnown *bool
if q := r.URL.Query(); q.Has("keyKnown") {
switch q.Get("keyKnown") {
case "true":
keyKnown = new(true)
case "false":
keyKnown = new(false)
default:
respondError(w, http.StatusBadRequest, "keyKnown must be true or false")
return
}
}
channels, err := reader.ListChannels(r.Context(), limit, hashHex, iatas, keyKnown, cursor, pageCursor)
if err != nil {
respondError(w, http.StatusInternalServerError, "internal server error")
return
Expand Down
55 changes: 52 additions & 3 deletions internal/api/handlers/channels_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import (
"net/http"
"net/http/httptest"
"reflect"
"strings"
"testing"
"time"

Expand Down Expand Up @@ -39,7 +40,7 @@ func TestListChannels_InvalidCursor(t *testing.T) {

func TestListChannels_PageCursor(t *testing.T) {
called := false
r := ChannelsRouter(stubReader{listChannels: func(_ context.Context, limit int32, hash []byte, iatas []string, legacy int64, cursor *api.ChannelCursor) (api.ChannelPage, error) {
r := ChannelsRouter(stubReader{listChannels: func(_ context.Context, limit int32, hash []byte, iatas []string, _ *bool, legacy int64, cursor *api.ChannelCursor) (api.ChannelPage, error) {
called = true
if limit != 2 || legacy != 0 || cursor == nil || cursor.ID != 9 || !cursor.LastSeen.Equal(time.UnixMicro(1700000000000123)) ||
!reflect.DeepEqual(hash, []byte{0xaa}) || !reflect.DeepEqual(iatas, []string{"YOW"}) {
Expand Down Expand Up @@ -145,7 +146,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.ChannelCursor) (api.ChannelPage, error) {
listChannels: func(_ context.Context, _ int32, _ []byte, _ []string, _ *bool, _ int64, _ *api.ChannelCursor) (api.ChannelPage, error) {
return api.ChannelPage{Page: api.Page[api.ChannelSummary]{Items: []api.ChannelSummary{{ID: 1, ChannelHash: "ab"}}}}, nil
},
}))
Expand All @@ -172,7 +173,7 @@ func TestListChannels_IATAParsing(t *testing.T) {
var got []string
r := chi.NewRouter()
r.Get("/channels", listChannels(stubReader{
listChannels: func(_ context.Context, _ int32, _ []byte, iatas []string, _ int64, _ *api.ChannelCursor) (api.ChannelPage, error) {
listChannels: func(_ context.Context, _ int32, _ []byte, iatas []string, _ *bool, _ int64, _ *api.ChannelCursor) (api.ChannelPage, error) {
got = iatas
return api.ChannelPage{}, nil
},
Expand All @@ -190,6 +191,54 @@ func TestListChannels_IATAParsing(t *testing.T) {
}
}

func TestListChannels_KeyKnownParsing(t *testing.T) {
yes, no := true, false
cases := []struct {
name string
query string
code int
want *bool
}{
{"true", "?keyKnown=true", http.StatusOK, &yes},
{"false", "?keyKnown=false", http.StatusOK, &no},
{"absent", "", http.StatusOK, nil},
{"uppercase", "?keyKnown=TRUE", http.StatusBadRequest, nil},
{"numeric", "?keyKnown=1", http.StatusBadRequest, nil},
{"empty", "?keyKnown=", http.StatusBadRequest, nil},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
var got *bool
called := false
r := chi.NewRouter()
r.Get("/channels", listChannels(stubReader{
listChannels: func(_ context.Context, _ int32, _ []byte, _ []string, keyKnown *bool, _ int64, _ *api.ChannelCursor) (api.ChannelPage, error) {
called = true
got = keyKnown
return api.ChannelPage{}, nil
},
}))
w := httptest.NewRecorder()
r.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/channels"+tc.query, nil))
if w.Code != tc.code {
t.Fatalf("expected %d, got %d", tc.code, w.Code)
}
if tc.code != http.StatusOK {
if called {
t.Error("reader called for invalid keyKnown")
}
if !strings.Contains(w.Body.String(), "keyKnown must be true or false") {
t.Errorf("unexpected error body %s", w.Body.String())
}
return
}
if !reflect.DeepEqual(got, tc.want) {
t.Errorf("expected keyKnown %v, got %v", tc.want, got)
}
})
}
}

func TestGetChannel_OK(t *testing.T) {
r := chi.NewRouter()
r.Get("/channels/{channelID}", getChannel(stubReader{
Expand Down
Loading
Loading