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

add observation count to messages

Browse Source
This commit is contained in:
Enot (ded) Skelly
2026-05-29 11:02:03 -07:00
parent e24eac4377
commit 2597192e89
4 changed files with 64 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
db/queries/queries.sql
+7 -3
View File
@@ -427,7 +427,8 @@ RETURNING id;
-- Pass a zero/null timestamp for since to return all messages up to limit.
-- Pass empty string for iata to skip IATA filtering (case-insensitive).
-- 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,
(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
JOIN packet_observations po ON po.packet_hash = cm.packet_hash
@@ -442,7 +443,8 @@ LIMIT $5;
-- Returns all messages across all channels with optional time, IATA and cursor filters.
-- Pass empty string for iata to skip IATA filtering (case-insensitive).
-- 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,
(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
JOIN packet_observations po ON po.packet_hash = cm.packet_hash
@@ -457,7 +459,9 @@ LIMIT $4;
-- May return messages from multiple channels if the hash collides across different keys.
-- Pass empty string for iata to skip IATA filtering (case-insensitive).
-- Pass cursor=0 to start from the beginning.
SELECT DISTINCT ON (cm.id) cm.*, c.channel_hash FROM channel_messages cm
SELECT DISTINCT ON (cm.id) cm.*, c.channel_hash,
(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
JOIN packet_observations po ON po.packet_hash = cm.packet_hash
WHERE c.channel_hash = $1
db/sqlc/queries.sql.go
+39 -29
View File
@@ -884,7 +884,8 @@ func (q *Queries) InsertObserverTelemetry(ctx context.Context, arg InsertObserve
}
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,
(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
JOIN packet_observations po ON po.packet_hash = cm.packet_hash
@@ -903,15 +904,16 @@ 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"`
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"`
}
// Returns all messages across all channels with optional time, IATA and cursor filters.
@@ -941,6 +943,7 @@ func (q *Queries) ListAllChannelMessages(ctx context.Context, arg ListAllChannel
&i.SentAt,
&i.PacketHashHex,
&i.ChannelHash,
&i.ObservationCount,
); err != nil {
return nil, err
}
@@ -953,7 +956,8 @@ 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,
(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
JOIN packet_observations po ON po.packet_hash = cm.packet_hash
@@ -974,15 +978,16 @@ 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"`
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"`
}
// Returns messages for a channel identified by integer ID.
@@ -1014,6 +1019,7 @@ func (q *Queries) ListChannelMessages(ctx context.Context, arg ListChannelMessag
&i.SentAt,
&i.PacketHashHex,
&i.ChannelHash,
&i.ObservationCount,
); err != nil {
return nil, err
}
@@ -1026,7 +1032,9 @@ 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 FROM channel_messages cm
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 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
JOIN packet_observations po ON po.packet_hash = cm.packet_hash
WHERE c.channel_hash = $1
@@ -1046,14 +1054,15 @@ 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"`
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"`
}
// Returns messages for all channels matching a hash byte.
@@ -1084,6 +1093,7 @@ func (q *Queries) ListChannelMessagesByHash(ctx context.Context, arg ListChannel
&i.Content,
&i.SentAt,
&i.ChannelHash,
&i.ObservationCount,
); err != nil {
return nil, err
}
db/store.go
+11 -10
View File
@@ -476,7 +476,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))
messages = append(messages, toChannelMessage(v.ID, v.PacketHashHex, v.ChannelHash, v.SenderName, v.Content, v.SentAt, v.ObservationCount))
}
} else {
rows, err := s.q.ListChannelMessages(ctx, sqlc.ListChannelMessagesParams{
@@ -495,7 +495,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))
messages = append(messages, toChannelMessage(v.ID, v.PacketHashHex, v.ChannelHash, v.SenderName, v.Content, v.SentAt, v.ObservationCount))
}
}
@@ -533,7 +533,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))
messages = append(messages, toChannelMessage(v.ID, hex.EncodeToString(v.PacketHash), v.ChannelHash, v.SenderName, v.Content, v.SentAt, v.ObservationCount))
}
var nextCursor *int64
if hasMore && len(messages) > 0 {
@@ -664,7 +664,7 @@ func (s *Store) GetObserver(ctx context.Context, observerID uuid.UUID) (*api.Obs
return &observer, nil
}
func toChannelMessage(id int64, packetHashHex string, channelHash []byte, senderName *string, content *string, sentAt pgtype.Timestamptz) api.ChannelMessage {
func toChannelMessage(id int64, packetHashHex string, channelHash []byte, senderName *string, content *string, sentAt pgtype.Timestamptz, observationCount int64) api.ChannelMessage {
sn := ""
if senderName != nil {
sn = *senderName
@@ -674,12 +674,13 @@ func toChannelMessage(id int64, packetHashHex string, channelHash []byte, sender
ct = *content
}
return api.ChannelMessage{
ID: id,
PacketHash: packetHashHex,
ChannelHash: hex.EncodeToString(channelHash),
SenderName: sn,
Content: ct,
SentAt: sentAt.Time.UnixMilli(),
ID: id,
PacketHash: packetHashHex,
ChannelHash: hex.EncodeToString(channelHash),
SenderName: sn,
Content: ct,
SentAt: sentAt.Time.UnixMilli(),
ObservationCount: observationCount,
}
}
internal/api/reader.go
+7 -6
View File
@@ -248,12 +248,13 @@ type Observer struct {
// 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
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
ObservationCount int64 `json:"observationCount"` // the number of observations for this message packet hash
}
// ChannelSummary is the minimal channel representation used in list responses.