diff --git a/db/queries/queries.sql b/db/queries/queries.sql index e286709..31eedcd 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -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 diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index db485e6..cdab633 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -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 } diff --git a/db/store.go b/db/store.go index 80c2bad..fd4ca82 100644 --- a/db/store.go +++ b/db/store.go @@ -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, } } diff --git a/internal/api/reader.go b/internal/api/reader.go index edd547c..9281011 100644 --- a/internal/api/reader.go +++ b/internal/api/reader.go @@ -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.