dandri/beacon-server
1
0
Fork 0
mirror of https://github.com/MeshCore-Beacon/beacon-server.git synced 2026-10-16 13:49:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(channels): expose recorded packet scope in message delivery

Browse Source
This commit is contained in:
n30nex
2026-09-30 07:19:03 -04:00
committed by MrAlders0n
parent 7e01abf5a0
commit 2abed6c195
20 changed files with 484 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/ci.yml
+1 -1
View File
@@ -58,7 +58,7 @@ jobs:
- name: Verify stats and endpoint queries against PostgreSQL 16
env:
BEACON_TEST_POSTGRES_DSN: postgres://postgres:backup-ci-only@127.0.0.1:5432/postgres?sslmode=disable
run: go test ./db -run '^Test(Signal|Paths|PacketEndpointsResolveLive|ObserverMetrics|RouteEvidence|RouteEvidenceIndex|AnalyticsRetention|AnalyticsRetentionConcurrent|DeleteOldPacketsBatches|MeshMapperCatalogue)Postgres$' -count=1 -v
run: go test ./db -run '^Test(Signal|Paths|PacketEndpointsResolveLive|ObserverMetrics|RouteEvidence|RouteEvidenceIndex|AnalyticsRetention|AnalyticsRetentionConcurrent|DeleteOldPacketsBatches|MeshMapperCatalogue|ChannelMessageScopes)Postgres$' -count=1 -v
- name: Verify backup command against PostgreSQL 16
run: |
db/channel_scopes_integration_test.go
+156
View File
@@ -0,0 +1,156 @@
// Copyright 2026 Beacon Contributors
// SPDX-License-Identifier: AGPL-3.0-or-later
package db
import (
"context"
"encoding/json"
"os"
"testing"
"time"
sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc"
"github.com/MeshCore-Beacon/beacon-server/internal/api"
"github.com/MeshCore-Beacon/beacon-server/internal/ingest"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
)
func TestChannelMessageScopesPostgres(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{"transport_scopes", "packets", "packet_observations", "channels", "channel_messages"} {
if _, err := tx.Exec(ctx, "CREATE TEMP TABLE "+table+" (LIKE public."+table+" INCLUDING ALL) ON COMMIT DROP"); err != nil {
t.Fatal(err)
}
}
_, err = tx.Exec(ctx, `
INSERT INTO transport_scopes(id,name,transport_key,key_fingerprint) VALUES
(1,'#yow',decode(repeat('00',16),'hex'),decode(repeat('01',8),'hex')),
(2,'#can',decode(repeat('02',16),'hex'),decode(repeat('03',8),'hex'));
INSERT INTO channels(id,channel_hash,key_fingerprint,name,is_hashtag,key_known) VALUES (123,'\x11','\x01020304','#fixture',true,true);
INSERT INTO packets(packet_hash,payload_type,payload_version,route_type,transport_codes_present,scope_id,raw_payload,raw_header)
SELECT decode(lpad(to_hex(i),2,'0'),'hex'),5,0,1,transport,scope_id,'\x00','\x15'
FROM (VALUES(1,true,1),(2,true,NULL),(3,false,NULL),(4,NULL,NULL),(5,true,2),(6,false,NULL)) v(i,transport,scope_id);
INSERT INTO packet_observations(packet_hash,observer_id,iata,heard_at,path_length_byte,hash_size,hop_count)
SELECT packet_hash,md5(n::text)::uuid,'YOW',NOW(),0,1,0 FROM packets CROSS JOIN generate_series(1,2) n;
`)
if err != nil {
t.Fatal(err)
}
s := &Store{q: sqlc.New(tx)}
// Later matching reception must not relabel the immutable first stored packet.
scopeID := int32(1)
_, err = s.UpsertPacket(ctx, ingest.UpsertPacketParams{PacketHash: []byte{6}, PayloadType: 5, ScopeID: &scopeID, TransportCodes: []byte{1, 2, 3, 4}, RawPayload: []byte{0}, RawHeader: []byte{0}})
if err != nil {
t.Fatal(err)
}
statuses := []api.ChannelScopeStatus{"matched", "unknown", "unscoped", "unavailable", "matched", "unscoped"}
wants := map[int64]*ingest.InsertedChannelMessage{}
for n, status := range statuses {
m := ingest.InsertChannelMessageParams{ChannelID: 123, PacketHash: []byte{byte(n + 1)}, SenderName: "Fixture", Content: "Synthetic message", SentAt: time.Now()}
inserted, err := s.InsertChannelMessage(ctx, m)
if err != nil || inserted == nil || inserted.ScopeStatus != status {
t.Fatalf("insert %d: %+v %v", n, inserted, err)
}
wants[inserted.ID] = inserted
duplicate, err := s.InsertChannelMessage(ctx, m)
if err != nil || duplicate != nil {
t.Fatal("duplicate changed live delivery", duplicate, err)
}
}
check := func(items []api.ChannelMessage) {
t.Helper()
if len(items) != 6 {
t.Fatalf("messages=%d", len(items))
}
for _, m := range items {
want := wants[m.ID]
if want == nil || m.ScopeStatus != want.ScopeStatus {
t.Fatal("history differs from insertion", m)
}
if (m.Scope == nil) != (want.Scope == nil) || (m.Scope != nil && *m.Scope != *want.Scope) {
t.Fatal("scope differs", m)
}
if m.ObservationCount != 2 {
t.Fatal("duplicate observations changed message grain", m)
}
b, _ := json.Marshal(m)
var wire map[string]any
_ = json.Unmarshal(b, &wire)
if _, ok := wire["scope"]; !ok || wire["scopeStatus"] != string(m.ScopeStatus) {
t.Fatal("wire metadata absent", string(b))
}
}
}
ch := int32(123)
for _, id := range []*int32{nil, &ch} {
p, err := s.ListChannelMessages(ctx, id, time.Time{}, 20, []string{"YOW"}, "", 0)
if err != nil {
t.Fatal(err)
}
check(p.Items)
}
hash, err := s.ListChannelMessagesByHash(ctx, []byte{0x11}, time.Time{}, 20, []string{"YOW"}, "", 0)
if err != nil {
t.Fatal(err)
}
check(hash.Items)
catchup, err := s.ListMessagesAfterID(ctx, 0, []string{"YOW"}, "", 20)
if err != nil {
t.Fatal(err)
}
check(catchup)
filtered, err := s.ListChannelMessages(ctx, &ch, time.Time{}, 20, []string{"YOW"}, "#yow", 0)
if err != nil || len(filtered.Items) != 1 || filtered.Items[0].Scope == nil || *filtered.Items[0].Scope != "#yow" {
t.Fatal(filtered, err)
}
catchup, err = s.ListMessagesAfterID(ctx, 0, []string{"YOW"}, "#yow", 20)
if err != nil || len(catchup) != 1 {
t.Fatal(catchup, err)
}
other, err := s.ListChannelMessages(ctx, &ch, time.Time{}, 20, []string{"YVR"}, "", 0)
if err != nil || len(other.Items) != 0 {
t.Fatal("region escaped", other, err)
}
first, err := s.ListChannelMessages(ctx, &ch, time.Time{}, 2, nil, "", 0)
if err != nil || !first.HasMore || first.NextCursor == nil {
t.Fatal(first, err)
}
second, err := s.ListChannelMessages(ctx, &ch, time.Time{}, 2, nil, "", *first.NextCursor)
if err != nil || len(second.Items) != 2 || second.Items[0].ID >= *first.NextCursor {
t.Fatal("cursor drift", second, err)
}
// Even without the production cascade, expired packet evidence must not leak orphan messages.
if _, err := tx.Exec(ctx, "DELETE FROM packets"); err != nil {
t.Fatal(err)
}
expired, err := s.ListChannelMessages(ctx, &ch, time.Time{}, 20, nil, "", 0)
if err != nil || len(expired.Items) != 0 {
t.Fatal(expired, err)
}
}
db/channels.go
+10 -8
View File
@@ -154,16 +154,16 @@ func (s *Store) GetChannel(ctx context.Context, channelID int32) (*api.Channel,
return &channel, nil
}
func (s *Store) InsertChannelMessage(ctx context.Context, m ingest.InsertChannelMessageParams) (bool, error) {
func (s *Store) InsertChannelMessage(ctx context.Context, m ingest.InsertChannelMessageParams) (*ingest.InsertedChannelMessage, error) {
params := sqlc.InsertChannelMessageParams{ChannelID: int32(m.ChannelID), PacketHash: m.PacketHash, SenderName: &m.SenderName, Content: &m.Content, SentAt: pgtype.Timestamptz{Time: m.SentAt, Valid: true}}
_, err := s.q.InsertChannelMessage(ctx, params)
row, err := s.q.InsertChannelMessage(ctx, params)
if errors.Is(err, pgx.ErrNoRows) {
return false, nil // duplicate
return nil, nil // duplicate
}
if err != nil {
return false, err
return nil, err
}
return true, nil
return &ingest.InsertedChannelMessage{ID: row.ID, Scope: row.ScopeName, ScopeStatus: api.RecordedChannelScopeStatus(row.ScopeName, row.TransportCodesPresent)}, nil
}
func (s *Store) ListChannelMessages(ctx context.Context, channelID *int32, since time.Time, limit int32, iatas []string, scope string, cursor int64) (api.Page[api.ChannelMessage], error) {
@@ -187,7 +187,7 @@ func (s *Store) ListChannelMessages(ctx context.Context, channelID *int32, since
}
messages = make([]api.ChannelMessage, 0, len(rows))
for _, v := range rows {
messages = append(messages, toChannelMessage(v.ID, v.PacketHashHex, v.ChannelHash, v.SenderName, v.Content, v.SentAt, v.ObservationCount))
messages = append(messages, toChannelMessage(v.ID, v.PacketHashHex, v.ChannelHash, v.SenderName, v.Content, v.SentAt, v.ObservationCount, v.ScopeName, v.TransportCodesPresent))
}
} else {
rows, err := s.q.ListChannelMessages(ctx, sqlc.ListChannelMessagesParams{
@@ -207,7 +207,7 @@ func (s *Store) ListChannelMessages(ctx context.Context, channelID *int32, since
}
messages = make([]api.ChannelMessage, 0, len(rows))
for _, v := range rows {
messages = append(messages, toChannelMessage(v.ID, v.PacketHashHex, v.ChannelHash, v.SenderName, v.Content, v.SentAt, v.ObservationCount))
messages = append(messages, toChannelMessage(v.ID, v.PacketHashHex, v.ChannelHash, v.SenderName, v.Content, v.SentAt, v.ObservationCount, v.ScopeName, v.TransportCodesPresent))
}
}
@@ -241,7 +241,7 @@ func (s *Store) ListChannelMessagesByHash(ctx context.Context, hash []byte, sinc
}
messages := make([]api.ChannelMessage, 0, len(rows))
for _, v := range rows {
messages = append(messages, toChannelMessage(v.ID, hex.EncodeToString(v.PacketHash), v.ChannelHash, v.SenderName, v.Content, v.SentAt, v.ObservationCount))
messages = append(messages, toChannelMessage(v.ID, hex.EncodeToString(v.PacketHash), v.ChannelHash, v.SenderName, v.Content, v.SentAt, v.ObservationCount, v.ScopeName, v.TransportCodesPresent))
}
var nextCursor *int64
if hasMore && len(messages) > 0 {
@@ -275,6 +275,8 @@ func (s *Store) ListMessagesAfterID(ctx context.Context, afterID int64, iatas []
v.Content,
v.SentAt,
v.ObservationCount,
v.ScopeName,
v.TransportCodesPresent,
))
}
return items, nil
db/queries/queries.sql
+16 -8
View File
@@ -944,17 +944,25 @@ SELECT * FROM channels WHERE id = $1;
-- ============================================================
-- name: InsertChannelMessage :one
INSERT INTO channel_messages (channel_id, packet_hash, sender_name, content, sent_at)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (packet_hash) DO NOTHING
RETURNING id;
-- Read the immutable first-packet scope in the same statement as insertion.
-- A later reception's transport code must not give live and historical messages different tags.
WITH inserted AS (
INSERT INTO channel_messages (channel_id, packet_hash, sender_name, content, sent_at)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (packet_hash) DO NOTHING
RETURNING id, packet_hash
)
SELECT inserted.id, ts.name AS scope_name, p.transport_codes_present
FROM inserted
JOIN packets p ON p.packet_hash = inserted.packet_hash
LEFT JOIN transport_scopes ts ON ts.id = p.scope_id;
-- name: ListChannelMessages :many
-- 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.
-- Pass cursor=0 to start from the beginning.
SELECT DISTINCT ON (cm.id) cm.*, encode(cm.packet_hash, 'hex') as packet_hash_hex, c.channel_hash,
SELECT DISTINCT ON (cm.id) cm.*, 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
FROM channel_messages cm
JOIN channels c ON c.id = cm.channel_id
@@ -973,7 +981,7 @@ LIMIT $6;
-- Returns all messages across all channels with optional time, IATA, scope and cursor filters.
-- Pass empty string for iata or scope to skip those filters.
-- Pass cursor=0 to start from the beginning.
SELECT DISTINCT ON (cm.id) cm.*, encode(cm.packet_hash, 'hex') as packet_hash_hex, c.channel_hash,
SELECT DISTINCT ON (cm.id) cm.*, 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
FROM channel_messages cm
JOIN channels c ON c.id = cm.channel_id
@@ -992,7 +1000,7 @@ LIMIT $5;
-- May return messages from multiple channels if the hash collides across different keys.
-- Pass empty string for iata or scope to skip those filters.
-- Pass cursor=0 to start from the beginning.
SELECT DISTINCT ON (cm.id) cm.*, c.channel_hash,
SELECT DISTINCT ON (cm.id) cm.*, 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
FROM channel_messages cm
JOIN channels c ON c.id = cm.channel_id
@@ -1010,7 +1018,7 @@ LIMIT $6;
-- name: ListMessagesAfterID :many
-- Returns messages after the given message ID, ordered oldest first.
-- Used for WS reconnect backfill.
SELECT DISTINCT ON (cm.id) cm.*, encode(cm.packet_hash, 'hex') as packet_hash_hex, c.channel_hash,
SELECT DISTINCT ON (cm.id) cm.*, 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
FROM channel_messages cm
JOIN channels c ON c.id = cm.channel_id
db/sqlc/mock/querier.go
+2 -2
View File
@@ -940,10 +940,10 @@ func (mr *MockQuerierMockRecorder) GetTransportScopes(ctx any) *gomock.Call {
}
// InsertChannelMessage mocks base method.
func (m *MockQuerier) InsertChannelMessage(ctx context.Context, arg db.InsertChannelMessageParams) (int64, error) {
func (m *MockQuerier) InsertChannelMessage(ctx context.Context, arg db.InsertChannelMessageParams) (db.InsertChannelMessageRow, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "InsertChannelMessage", ctx, arg)
ret0, _ := ret[0].(int64)
ret0, _ := ret[0].(db.InsertChannelMessageRow)
ret1, _ := ret[1].(error)
return ret0, ret1
}
db/sqlc/querier.go
+3 -1
View File
@@ -128,7 +128,9 @@ type Querier interface {
// ============================================================
// CHANNEL MESSAGES
// ============================================================
InsertChannelMessage(ctx context.Context, arg InsertChannelMessageParams) (int64, error)
// Read the immutable first-packet scope in the same statement as insertion.
// A later reception's transport code must not give live and historical messages different tags.
InsertChannelMessage(ctx context.Context, arg InsertChannelMessageParams) (InsertChannelMessageRow, error)
// ============================================================
// PACKET OBSERVATIONS
// ============================================================
db/sqlc/queries.sql.go
+81 -51
View File
@@ -2110,10 +2110,16 @@ func (q *Queries) GetTransportScopes(ctx context.Context) ([]GetTransportScopesR
const insertChannelMessage = `-- name: InsertChannelMessage :one
INSERT INTO channel_messages (channel_id, packet_hash, sender_name, content, sent_at)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (packet_hash) DO NOTHING
RETURNING id
WITH inserted AS (
INSERT INTO channel_messages (channel_id, packet_hash, sender_name, content, sent_at)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (packet_hash) DO NOTHING
RETURNING id, packet_hash
)
SELECT inserted.id, ts.name AS scope_name, p.transport_codes_present
FROM inserted
JOIN packets p ON p.packet_hash = inserted.packet_hash
LEFT JOIN transport_scopes ts ON ts.id = p.scope_id
`
type InsertChannelMessageParams struct {
@@ -2124,10 +2130,18 @@ type InsertChannelMessageParams struct {
SentAt pgtype.Timestamptz `json:"sent_at"`
}
type InsertChannelMessageRow struct {
ID int64 `json:"id"`
ScopeName *string `json:"scope_name"`
TransportCodesPresent *bool `json:"transport_codes_present"`
}
// ============================================================
// CHANNEL MESSAGES
// ============================================================
func (q *Queries) InsertChannelMessage(ctx context.Context, arg InsertChannelMessageParams) (int64, error) {
// Read the immutable first-packet scope in the same statement as insertion.
// A later reception's transport code must not give live and historical messages different tags.
func (q *Queries) InsertChannelMessage(ctx context.Context, arg InsertChannelMessageParams) (InsertChannelMessageRow, error) {
row := q.db.QueryRow(ctx, insertChannelMessage,
arg.ChannelID,
arg.PacketHash,
@@ -2135,9 +2149,9 @@ func (q *Queries) InsertChannelMessage(ctx context.Context, arg InsertChannelMes
arg.Content,
arg.SentAt,
)
var id int64
err := row.Scan(&id)
return id, err
var i InsertChannelMessageRow
err := row.Scan(&i.ID, &i.ScopeName, &i.TransportCodesPresent)
return i, err
}
const insertObservation = `-- name: InsertObservation :one
@@ -2309,7 +2323,7 @@ func (q *Queries) ListAccounts(ctx context.Context) ([]Account, error) {
}
const listAllChannelMessages = `-- name: ListAllChannelMessages :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,
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
FROM channel_messages cm
JOIN channels c ON c.id = cm.channel_id
@@ -2333,16 +2347,18 @@ type ListAllChannelMessagesParams struct {
}
type ListAllChannelMessagesRow struct {
ID int64 `json:"id"`
ChannelID int32 `json:"channel_id"`
PacketHash []byte `json:"packet_hash"`
SenderName *string `json:"sender_name"`
SenderPubkey []byte `json:"sender_pubkey"`
Content *string `json:"content"`
SentAt pgtype.Timestamptz `json:"sent_at"`
PacketHashHex string `json:"packet_hash_hex"`
ChannelHash []byte `json:"channel_hash"`
ObservationCount int64 `json:"observation_count"`
ID int64 `json:"id"`
ChannelID int32 `json:"channel_id"`
PacketHash []byte `json:"packet_hash"`
SenderName *string `json:"sender_name"`
SenderPubkey []byte `json:"sender_pubkey"`
Content *string `json:"content"`
SentAt pgtype.Timestamptz `json:"sent_at"`
PacketHashHex string `json:"packet_hash_hex"`
ChannelHash []byte `json:"channel_hash"`
ScopeName *string `json:"scope_name"`
TransportCodesPresent *bool `json:"transport_codes_present"`
ObservationCount int64 `json:"observation_count"`
}
// Returns all messages across all channels with optional time, IATA, scope and cursor filters.
@@ -2373,6 +2389,8 @@ func (q *Queries) ListAllChannelMessages(ctx context.Context, arg ListAllChannel
&i.SentAt,
&i.PacketHashHex,
&i.ChannelHash,
&i.ScopeName,
&i.TransportCodesPresent,
&i.ObservationCount,
); err != nil {
return nil, err
@@ -2386,7 +2404,7 @@ func (q *Queries) ListAllChannelMessages(ctx context.Context, arg ListAllChannel
}
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,
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
FROM channel_messages cm
JOIN channels c ON c.id = cm.channel_id
@@ -2412,16 +2430,18 @@ type ListChannelMessagesParams struct {
}
type ListChannelMessagesRow struct {
ID int64 `json:"id"`
ChannelID int32 `json:"channel_id"`
PacketHash []byte `json:"packet_hash"`
SenderName *string `json:"sender_name"`
SenderPubkey []byte `json:"sender_pubkey"`
Content *string `json:"content"`
SentAt pgtype.Timestamptz `json:"sent_at"`
PacketHashHex string `json:"packet_hash_hex"`
ChannelHash []byte `json:"channel_hash"`
ObservationCount int64 `json:"observation_count"`
ID int64 `json:"id"`
ChannelID int32 `json:"channel_id"`
PacketHash []byte `json:"packet_hash"`
SenderName *string `json:"sender_name"`
SenderPubkey []byte `json:"sender_pubkey"`
Content *string `json:"content"`
SentAt pgtype.Timestamptz `json:"sent_at"`
PacketHashHex string `json:"packet_hash_hex"`
ChannelHash []byte `json:"channel_hash"`
ScopeName *string `json:"scope_name"`
TransportCodesPresent *bool `json:"transport_codes_present"`
ObservationCount int64 `json:"observation_count"`
}
// Returns messages for a channel identified by integer ID.
@@ -2454,6 +2474,8 @@ func (q *Queries) ListChannelMessages(ctx context.Context, arg ListChannelMessag
&i.SentAt,
&i.PacketHashHex,
&i.ChannelHash,
&i.ScopeName,
&i.TransportCodesPresent,
&i.ObservationCount,
); err != nil {
return nil, err
@@ -2467,7 +2489,7 @@ func (q *Queries) ListChannelMessages(ctx context.Context, arg ListChannelMessag
}
const listChannelMessagesByHash = `-- name: ListChannelMessagesByHash :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, c.channel_hash,
SELECT DISTINCT ON (cm.id) cm.id, cm.channel_id, cm.packet_hash, cm.sender_name, cm.sender_pubkey, cm.content, cm.sent_at, 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
FROM channel_messages cm
JOIN channels c ON c.id = cm.channel_id
@@ -2493,15 +2515,17 @@ type ListChannelMessagesByHashParams struct {
}
type ListChannelMessagesByHashRow struct {
ID int64 `json:"id"`
ChannelID int32 `json:"channel_id"`
PacketHash []byte `json:"packet_hash"`
SenderName *string `json:"sender_name"`
SenderPubkey []byte `json:"sender_pubkey"`
Content *string `json:"content"`
SentAt pgtype.Timestamptz `json:"sent_at"`
ChannelHash []byte `json:"channel_hash"`
ObservationCount int64 `json:"observation_count"`
ID int64 `json:"id"`
ChannelID int32 `json:"channel_id"`
PacketHash []byte `json:"packet_hash"`
SenderName *string `json:"sender_name"`
SenderPubkey []byte `json:"sender_pubkey"`
Content *string `json:"content"`
SentAt pgtype.Timestamptz `json:"sent_at"`
ChannelHash []byte `json:"channel_hash"`
ScopeName *string `json:"scope_name"`
TransportCodesPresent *bool `json:"transport_codes_present"`
ObservationCount int64 `json:"observation_count"`
}
// Returns messages for all channels matching a hash byte.
@@ -2533,6 +2557,8 @@ func (q *Queries) ListChannelMessagesByHash(ctx context.Context, arg ListChannel
&i.Content,
&i.SentAt,
&i.ChannelHash,
&i.ScopeName,
&i.TransportCodesPresent,
&i.ObservationCount,
); err != nil {
return nil, err
@@ -2775,7 +2801,7 @@ func (q *Queries) ListKnownRoutes(ctx context.Context, arg ListKnownRoutesParams
}
const listMessagesAfterID = `-- name: ListMessagesAfterID :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,
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
FROM channel_messages cm
JOIN channels c ON c.id = cm.channel_id
@@ -2797,16 +2823,18 @@ type ListMessagesAfterIDParams struct {
}
type ListMessagesAfterIDRow struct {
ID int64 `json:"id"`
ChannelID int32 `json:"channel_id"`
PacketHash []byte `json:"packet_hash"`
SenderName *string `json:"sender_name"`
SenderPubkey []byte `json:"sender_pubkey"`
Content *string `json:"content"`
SentAt pgtype.Timestamptz `json:"sent_at"`
PacketHashHex string `json:"packet_hash_hex"`
ChannelHash []byte `json:"channel_hash"`
ObservationCount int64 `json:"observation_count"`
ID int64 `json:"id"`
ChannelID int32 `json:"channel_id"`
PacketHash []byte `json:"packet_hash"`
SenderName *string `json:"sender_name"`
SenderPubkey []byte `json:"sender_pubkey"`
Content *string `json:"content"`
SentAt pgtype.Timestamptz `json:"sent_at"`
PacketHashHex string `json:"packet_hash_hex"`
ChannelHash []byte `json:"channel_hash"`
ScopeName *string `json:"scope_name"`
TransportCodesPresent *bool `json:"transport_codes_present"`
ObservationCount int64 `json:"observation_count"`
}
// Returns messages after the given message ID, ordered oldest first.
@@ -2835,6 +2863,8 @@ func (q *Queries) ListMessagesAfterID(ctx context.Context, arg ListMessagesAfter
&i.SentAt,
&i.PacketHashHex,
&i.ChannelHash,
&i.ScopeName,
&i.TransportCodesPresent,
&i.ObservationCount,
); err != nil {
return nil, err
db/store.go
+3 -1
View File
@@ -130,7 +130,7 @@ func tristate(b *bool) string {
}
// toChannelMessage maps raw sqlc row fields to an api.ChannelMessage.
func toChannelMessage(id int64, packetHashHex string, channelHash []byte, senderName *string, content *string, sentAt pgtype.Timestamptz, observationCount int64) api.ChannelMessage {
func toChannelMessage(id int64, packetHashHex string, channelHash []byte, senderName *string, content *string, sentAt pgtype.Timestamptz, observationCount int64, scope *string, transport *bool) api.ChannelMessage {
sn := ""
if senderName != nil {
sn = *senderName
@@ -147,6 +147,8 @@ func toChannelMessage(id int64, packetHashHex string, channelHash []byte, sender
Content: ct,
SentAt: sentAt.Time.UnixMilli(),
ObservationCount: observationCount,
Scope: scope,
ScopeStatus: api.RecordedChannelScopeStatus(scope, transport),
}
}
db/store_test.go
+2 -2
View File
@@ -59,7 +59,7 @@ func TestToChannelMessage(t *testing.T) {
channelHash := []byte{0xab}
sentAt := pgtype.Timestamptz{Time: time.UnixMilli(1700000000000), Valid: true}
msg := toChannelMessage(42, "deadbeef", channelHash, &senderName, &content, sentAt, 7)
msg := toChannelMessage(42, "deadbeef", channelHash, &senderName, &content, sentAt, 7, nil, nil)
if msg.ID != 42 {
t.Errorf("expected ID 42, got %d", msg.ID)
@@ -86,7 +86,7 @@ func TestToChannelMessage(t *testing.T) {
func TestToChannelMessage_NilFields(t *testing.T) {
sentAt := pgtype.Timestamptz{Time: time.UnixMilli(0), Valid: true}
msg := toChannelMessage(1, "abc", []byte{0x01}, nil, nil, sentAt, 0)
msg := toChannelMessage(1, "abc", []byte{0x01}, nil, nil, sentAt, 0, nil, nil)
if msg.SenderName != "" {
t.Errorf("expected empty SenderName, got %s", msg.SenderName)
}
docs/channel-scopes.md
+16
View File
@@ -0,0 +1,16 @@
# Channel-message transport scope
Channel history, the global/hash message APIs, reconnect backfill and `channelMessage` events now include `scope` and `scopeStatus`. Live events also include their stored message `id`; existing fields, endpoints and cursors are unchanged.
| Status | Meaning |
|---|---|
| `matched` | `scope` names the scope recorded on the first stored packet. |
| `unscoped` | The first stored packet has no transport codes; `scope` is null. |
| `unknown` | Transport codes were recorded but no scope name was resolved, including ambiguous matches. |
| `unavailable` | Scope/capture metadata is missing. |
These labels belong to the first stored packet, not every later reception of the same payload. Later paths may differ. Live delivery reads that same evidence while inserting the channel message, so a later reception cannot relabel it differently from history. Existing scope query filters retain their meaning. Imported catalogue membership and channel names are never used as message evidence.
The insertion query returns metadata only for a new message. Duplicate deliveries still produce no extra live message. Adding a decryption key later uses the existing backfill path, retains the stored packet evidence and does not broadcast old messages as new traffic. This change adds no keys, migration or historical scan. Expired packet/message detail remains unavailable.
Deploy this API before its web consumer. Older clients may ignore the added fields; clients reading older servers must treat absent metadata as unavailable, not assume a message is unscoped. Scope filtering of live events requires the new fields. The channel encryption key and the packet transport scope are separate concepts.
docs/docs.go
+13
View File
@@ -3481,6 +3481,19 @@ const docTemplate = `{
"description": "hex-encoded packet hash for correlation with packet events",
"type": "string"
},
"scope": {
"description": "matched scope on the first stored packet; null when none was recorded",
"type": "string"
},
"scopeStatus": {
"type": "string",
"enum": [
"matched",
"unscoped",
"unknown",
"unavailable"
]
},
"senderName": {
"description": "display name from the decrypted payload",
"type": "string"
docs/swagger.json
+13
View File
@@ -3479,6 +3479,19 @@
"description": "hex-encoded packet hash for correlation with packet events",
"type": "string"
},
"scope": {
"description": "matched scope on the first stored packet; null when none was recorded",
"type": "string"
},
"scopeStatus": {
"type": "string",
"enum": [
"matched",
"unscoped",
"unknown",
"unavailable"
]
},
"senderName": {
"description": "display name from the decrypted payload",
"type": "string"
docs/swagger.yaml
+11
View File
@@ -135,6 +135,17 @@ definitions:
packetHash:
description: hex-encoded packet hash for correlation with packet events
type: string
scope:
description: matched scope on the first stored packet; null when none was
recorded
type: string
scopeStatus:
enum:
- matched
- unscoped
- unknown
- unavailable
type: string
senderName:
description: display name from the decrypted payload
type: string
internal/api/channel_scope_test.go
+27
View File
@@ -0,0 +1,27 @@
// Copyright 2026 Beacon Contributors
// SPDX-License-Identifier: AGPL-3.0-or-later
package api
import "testing"
func TestRecordedChannelScopeStatus(t *testing.T) {
name := "#yow"
for _, tc := range []struct {
name *string
transport, valid bool
want string
}{
{&name, true, true, "matched"}, {nil, true, true, "unknown"},
{nil, false, true, "unscoped"}, {nil, false, false, "unavailable"},
{&name, false, false, "matched"},
} {
var transport *bool
if tc.valid {
transport = &tc.transport
}
if got := RecordedChannelScopeStatus(tc.name, transport); string(got) != tc.want {
t.Fatalf("got %s, want %s", got, tc.want)
}
}
}
internal/api/channels.go
+26 -7
View File
@@ -56,13 +56,32 @@ func ParseChannelCursor(raw string) (*ChannelCursor, error) {
// ChannelMessage represents a single decrypted channel message.
// Only messages for channels with a known key are stored and returned.
type ChannelMessage struct {
ID int64 `json:"id"`
PacketHash string `json:"packetHash"` // hex-encoded packet hash for correlation with packet events
ChannelHash string `json:"channelHash"` // hex-encoded single-byte channel hash
SenderName string `json:"senderName"` // display name from the decrypted payload
Content string `json:"content"` // decrypted message text
SentAt int64 `json:"sentAt"` // epoch ms, from the sender's embedded timestamp
ObservationCount int64 `json:"observationCount"` // number of packet_observations rows for this message's packet hash
ID int64 `json:"id"`
PacketHash string `json:"packetHash"` // hex-encoded packet hash for correlation with packet events
ChannelHash string `json:"channelHash"` // hex-encoded single-byte channel hash
SenderName string `json:"senderName"` // display name from the decrypted payload
Content string `json:"content"` // decrypted message text
SentAt int64 `json:"sentAt"` // epoch ms, from the sender's embedded timestamp
ObservationCount int64 `json:"observationCount"` // number of packet_observations rows for this message's packet hash
Scope *string `json:"scope"` // matched scope on the first stored packet; null when none was recorded
ScopeStatus ChannelScopeStatus `json:"scopeStatus" enums:"matched,unscoped,unknown,unavailable"`
}
// ChannelScopeStatus describes stored packet evidence, not a channel key or all reception paths.
type ChannelScopeStatus string
// RecordedChannelScopeStatus preserves missing capture metadata as unavailable.
func RecordedChannelScopeStatus(name *string, transport *bool) ChannelScopeStatus {
if name != nil {
return "matched"
}
if transport == nil {
return "unavailable"
}
if *transport {
return "unknown"
}
return "unscoped"
}
// ChannelSummary is the minimal channel representation used in list responses.
internal/ingest/backfill.go
+4 -2
View File
@@ -32,6 +32,7 @@ type DecryptGroupTextResult struct {
// NewMessage is false if InsertChannelMessage found the message already existed
// (packet_hash is unique per message) -- e.g. re-running the backfill twice.
NewMessage bool
Message *InsertedChannelMessage
}
// DecryptGroupText attempts to decrypt a GRP_TXT payload against keys and, on success,
@@ -77,7 +78,7 @@ func DecryptGroupText(ctx context.Context, db DB, keys ChannelKeyStore, packetHa
if err != nil {
return nil, fmt.Errorf("insert channel message: %w", err)
}
if newMsg {
if newMsg != nil {
// Non-fatal: the message is stored either way, so just log and continue -- matches
// the live ingest path's existing behavior of not treating this as a hard failure.
if err := db.SetPacketDecrypted(ctx, packetHash); err != nil {
@@ -89,7 +90,8 @@ func DecryptGroupText(ctx context.Context, db DB, keys ChannelKeyStore, packetHa
ChannelID: channelID,
ChannelHash: channelHashBytes,
Entry: usedEntry,
NewMessage: newMsg,
NewMessage: newMsg != nil,
Message: newMsg,
}, nil
}
internal/ingest/channel_scope_test.go
+75
View File
@@ -0,0 +1,75 @@
// Copyright 2026 Beacon Contributors
// SPDX-License-Identifier: AGPL-3.0-or-later
package ingest
import (
"context"
"encoding/json"
"testing"
"time"
"github.com/MeshCore-Beacon/beacon-server/internal/api"
"github.com/MeshCore-Beacon/beacon-server/internal/hub"
"github.com/MeshCore-Beacon/beacon-server/internal/keystore"
"github.com/meshcore-go/meshcore-go"
)
type scopeMessageDB struct {
*stubDB
next *InsertedChannelMessage
}
func (s *scopeMessageDB) InsertChannelMessage(context.Context, InsertChannelMessageParams) (*InsertedChannelMessage, error) {
value := s.next
s.next = nil
return value, nil
}
func TestChannelMessageScopeLive(t *testing.T) {
for _, status := range []string{"matched", "unscoped", "unknown", "unavailable"} {
t.Run(status, func(t *testing.T) {
w, base := newTestWorker()
stored := &InsertedChannelMessage{ID: 42, ScopeStatus: api.ChannelScopeStatus(status)}
name := "#stored"
if status == "matched" {
stored.Scope = &name
}
w.db = &scopeMessageDB{stubDB: base, next: stored}
psk := make([]byte, 16)
w.keys = &mapKeys{entries: map[byte][]keystore.Entry{0x11: {{Key: psk, Fingerprint: []byte{1}, Name: "fixture"}}}}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
go w.hub.Run()
client := w.hub.NewClient()
defer w.hub.Remove(client)
w.hub.AddScope(client, "scope", hub.Scope{Events: []hub.EventType{hub.EventChannelMessage, hub.EventObserverStatus}})
waitForSummarySubscriber(t, ctx, w.hub, client)
packet := &meshcore.Packet{Header: meshcore.MakeHeader(meshcore.RouteTypeTransportFlood, meshcore.PayloadTypeGrpTxt, 0), Payload: encryptedGroupText(t, 0x11, psk, "Fixture", "test")}
latest := "#different-later-reception"
w.handlePayloadTypeSideEffects(ctx, packet, "YOW", []byte{1}, RadioSettings{}, nil, &latest, nil, 0)
select {
case event := <-client.Send:
var got channelMessageEvent
if err := json.Unmarshal(event.Payload, &got); err != nil {
t.Fatal(err)
}
if got.ID != 42 || got.ScopeStatus != stored.ScopeStatus || (got.Scope == nil) != (stored.Scope == nil) || (got.Scope != nil && *got.Scope != name) {
t.Fatal("live scope did not use stored evidence", got)
}
case <-ctx.Done():
t.Fatal("no live message")
}
w.handlePayloadTypeSideEffects(ctx, packet, "YOW", []byte{1}, RadioSettings{}, nil, &latest, nil, 0)
w.hub.Broadcast(hub.Event{Type: hub.EventObserverStatus})
select {
case event := <-client.Send:
if event.Type == hub.EventChannelMessage {
t.Fatal("duplicate broadcast")
}
case <-ctx.Done():
t.Fatal("no barrier")
}
})
}
}
internal/ingest/ingest.go
+1 -1
View File
@@ -132,7 +132,7 @@ type DB interface {
UpsertNodeShortID(ctx context.Context, nodeID uuid.UUID, iata string, prefix4 []byte) error
// InsertChannelMessage stores a decrypted group text message. Returns insert success and an error.
InsertChannelMessage(ctx context.Context, m InsertChannelMessageParams) (bool, error)
InsertChannelMessage(ctx context.Context, m InsertChannelMessageParams) (*InsertedChannelMessage, error)
// UpdateObserverStatus updates the observer row from a /status message. Returns the OberserID
// and any error.
internal/ingest/ingest_test.go
+5 -2
View File
@@ -231,8 +231,11 @@ func (s *stubDB) GetNodesByIDs(_ context.Context, _ []uuid.UUID) (map[uuid.UUID]
return nil, nil
}
func (s *stubDB) InsertChannelMessage(_ context.Context, _ InsertChannelMessageParams) (bool, error) {
return s.insertChannelMessageResult, nil
func (s *stubDB) InsertChannelMessage(_ context.Context, _ InsertChannelMessageParams) (*InsertedChannelMessage, error) {
if !s.insertChannelMessageResult {
return nil, nil
}
return &InsertedChannelMessage{ID: 1, ScopeStatus: "unavailable"}, nil
}
func (s *stubDB) UpdateObserverStatus(_ context.Context, _ UpdateObserverStatusParams) (uuid.UUID, error) {
internal/ingest/side_effects.go
+19 -6
View File
@@ -37,14 +37,24 @@ type InsertChannelMessageParams struct {
SentAt time.Time
}
// InsertedChannelMessage is returned only for a new message, using its stored packet evidence.
type InsertedChannelMessage struct {
ID int64
Scope *string
ScopeStatus api.ChannelScopeStatus
}
// channelMessageEvent is the JSON payload for a channelMessage WS event.
type channelMessageEvent struct {
ChannelID int `json:"channelId"`
ChannelHash string `json:"channelHash"` // hex-encoded single byte
PacketHash string `json:"packetHash"` // hex-encoded
SenderName string `json:"senderName"`
Content string `json:"content"`
SentAt int64 `json:"sentAt"` // epoch ms
ID int64 `json:"id"`
ChannelID int `json:"channelId"`
ChannelHash string `json:"channelHash"` // hex-encoded single byte
PacketHash string `json:"packetHash"` // hex-encoded
SenderName string `json:"senderName"`
Content string `json:"content"`
SentAt int64 `json:"sentAt"` // epoch ms
Scope *string `json:"scope"`
ScopeStatus api.ChannelScopeStatus `json:"scopeStatus"`
}
// nodeUpdateEvent is the JSON payload for a nodeUpdate WS event.
@@ -207,6 +217,9 @@ func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshc
if result.NewMessage {
evt := channelMessageEvent{
ID: result.Message.ID,
Scope: result.Message.Scope,
ScopeStatus: result.Message.ScopeStatus,
ChannelID: result.ChannelID,
ChannelHash: hex.EncodeToString(channelHashBytes),
PacketHash: hex.EncodeToString(packetHash),