dandri/beacon-server
1
0
Fork 0
mirror of https://github.com/MeshCore-Beacon/beacon-server.git synced 2026-09-02 00:58:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(channels): track channel activity per IATA in its own table

Browse Source 
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.
This commit is contained in:
MrAlders0n
2026-07-24 12:07:05 -07:00
committed by Ded
parent b33ebc007b
commit dff9c6cf00
22 changed files with 313 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
db/channels.go
+17 -5
View File
@@ -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
db/channels_test.go
+31 -11
View File
@@ -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)
db/migrations/011_channel_iatas.sql
+19
View File
@@ -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;
db/queries/queries.sql
+19 -12
View File
@@ -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;
db/sqlc/mock/querier.go
+28
View File
@@ -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()
db/sqlc/models.go
+6
View File
@@ -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"`
db/sqlc/querier.go
+5 -3
View File
@@ -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
// ============================================================
db/sqlc/queries.sql.go
+42 -17
View File
@@ -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
docs/docs.go
+14 -8
View File
@@ -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() {
docs/swagger.json
+6
View File
@@ -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",
docs/swagger.yaml
+4
View File
@@ -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
internal/api/handlers/channels.go
+3 -2
View File
@@ -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
internal/api/handlers/channels_test.go
+35 -1
View File
@@ -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{
internal/api/handlers/stub_reader_test.go
+3 -3
View File
@@ -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
}
internal/api/reader.go
+3 -2
View File
@@ -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.
internal/background/tasks.go
+3
View File
@@ -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
},
}
internal/cache/cache_test.go
Vendored
+1 -1
View File
@@ -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
}
internal/cache/reader.go
Vendored
+2 -2
View File
@@ -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].
internal/ingest/ingest.go
+3
View File
@@ -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)
internal/ingest/ingest_test.go
+8 -1
View File
@@ -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
}
internal/ingest/packet.go
+6
View File
@@ -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
internal/ingest/side_effects_test.go
+55
View File
@@ -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)
}
}