From 2abed6c1952e86353ad0f14acd27cc43e498b190 Mon Sep 17 00:00:00 2001 From: n30nex Date: Sun, 27 Sep 2026 19:16:49 -0400 Subject: [PATCH] feat(channels): expose recorded packet scope in message delivery --- .github/workflows/ci.yml | 2 +- db/channel_scopes_integration_test.go | 156 ++++++++++++++++++++++++++ db/channels.go | 18 +-- db/queries/queries.sql | 24 ++-- db/sqlc/mock/querier.go | 4 +- db/sqlc/querier.go | 4 +- db/sqlc/queries.sql.go | 132 +++++++++++++--------- db/store.go | 4 +- db/store_test.go | 4 +- docs/channel-scopes.md | 16 +++ docs/docs.go | 13 +++ docs/swagger.json | 13 +++ docs/swagger.yaml | 11 ++ internal/api/channel_scope_test.go | 27 +++++ internal/api/channels.go | 33 ++++-- internal/ingest/backfill.go | 6 +- internal/ingest/channel_scope_test.go | 75 +++++++++++++ internal/ingest/ingest.go | 2 +- internal/ingest/ingest_test.go | 7 +- internal/ingest/side_effects.go | 25 ++++- 20 files changed, 484 insertions(+), 92 deletions(-) create mode 100644 db/channel_scopes_integration_test.go create mode 100644 docs/channel-scopes.md create mode 100644 internal/api/channel_scope_test.go create mode 100644 internal/ingest/channel_scope_test.go diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6690234..328b452 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -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: | diff --git a/db/channel_scopes_integration_test.go b/db/channel_scopes_integration_test.go new file mode 100644 index 0000000..69591aa --- /dev/null +++ b/db/channel_scopes_integration_test.go @@ -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) + } +} diff --git a/db/channels.go b/db/channels.go index 599230a..9082e5a 100644 --- a/db/channels.go +++ b/db/channels.go @@ -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 diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 366bf42..328960e 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -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 diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index 16ea073..8a9a7ef 100644 --- a/db/sqlc/mock/querier.go +++ b/db/sqlc/mock/querier.go @@ -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 } diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index f96b9a3..1f80ab6 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -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 // ============================================================ diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index f33aa66..e6fb1ed 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -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 diff --git a/db/store.go b/db/store.go index 20866d7..2f9f35a 100644 --- a/db/store.go +++ b/db/store.go @@ -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), } } diff --git a/db/store_test.go b/db/store_test.go index 4d0521b..41f4393 100644 --- a/db/store_test.go +++ b/db/store_test.go @@ -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) } diff --git a/docs/channel-scopes.md b/docs/channel-scopes.md new file mode 100644 index 0000000..c88aaae --- /dev/null +++ b/docs/channel-scopes.md @@ -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. diff --git a/docs/docs.go b/docs/docs.go index 862fa8a..288a70b 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -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" diff --git a/docs/swagger.json b/docs/swagger.json index 5af6a53..5679444 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -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" diff --git a/docs/swagger.yaml b/docs/swagger.yaml index 2229cfd..a58626f 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -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 diff --git a/internal/api/channel_scope_test.go b/internal/api/channel_scope_test.go new file mode 100644 index 0000000..f553a62 --- /dev/null +++ b/internal/api/channel_scope_test.go @@ -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) + } + } +} diff --git a/internal/api/channels.go b/internal/api/channels.go index c609a23..00e9d85 100644 --- a/internal/api/channels.go +++ b/internal/api/channels.go @@ -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. diff --git a/internal/ingest/backfill.go b/internal/ingest/backfill.go index f402a4d..619941a 100644 --- a/internal/ingest/backfill.go +++ b/internal/ingest/backfill.go @@ -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 } diff --git a/internal/ingest/channel_scope_test.go b/internal/ingest/channel_scope_test.go new file mode 100644 index 0000000..3ee6933 --- /dev/null +++ b/internal/ingest/channel_scope_test.go @@ -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") + } + }) + } +} diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index 0c4d7f8..fa50532 100644 --- a/internal/ingest/ingest.go +++ b/internal/ingest/ingest.go @@ -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. diff --git a/internal/ingest/ingest_test.go b/internal/ingest/ingest_test.go index 2090f0e..248d3a4 100644 --- a/internal/ingest/ingest_test.go +++ b/internal/ingest/ingest_test.go @@ -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) { diff --git a/internal/ingest/side_effects.go b/internal/ingest/side_effects.go index 5d5c92c..6f809fe 100644 --- a/internal/ingest/side_effects.go +++ b/internal/ingest/side_effects.go @@ -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),