From 9b242583d961a2c92610e99137f002bf8367a017 Mon Sep 17 00:00:00 2001 From: "Enot (ded) Skelly" Date: Mon, 25 May 2026 16:01:01 -0700 Subject: [PATCH] add messages enpoints messages at /channels as well as generic /messages --- README.md | 19 ++++--- db/queries/queries.sql | 11 +++- db/sqlc/queries.sql.go | 59 +++++++++++++++++++- db/store.go | 88 +++++++++++++++-------------- internal/api/handlers/channels.go | 6 +- internal/api/handlers/messages.go | 93 +++++++++++++++++++++++++++++++ internal/api/reader.go | 2 +- internal/api/router/router.go | 1 + 8 files changed, 222 insertions(+), 57 deletions(-) create mode 100644 internal/api/handlers/messages.go diff --git a/README.md b/README.md index afff577..ff436bc 100644 --- a/README.md +++ b/README.md @@ -205,12 +205,16 @@ Base path: `/api/v1` ### Implemented -| Method | Path | Description | -| ------ | --------------- | ---------------------------------- | -| `GET` | `/iatas` | List all known IATA codes | -| `GET` | `/iatas/{iata}` | Get a single IATA code | -| `GET` | `/regions` | List all regions (summary) | -| `GET` | `/regions/{id}` | Get a single region with IATA list | +| Method | Path | Description | +| ------ | ------------------------- | -------------------------------------------------------------------------------------- | +| `GET` | `/iatas` | List all known IATA codes | +| `GET` | `/iatas/{iata}` | Get a single IATA code | +| `GET` | `/regions` | List all regions (summary) | +| `GET` | `/regions/{id}` | Get a single region with IATA list | +| `GET` | `/channels` | List channels (optional: `?hash=&since=&limit=50`) | +| `GET` | `/channels/{id}` | Get channel detail by integer ID | +| `GET` | `/channels/{id}/messages` | List messages for a channel | +| `GET` | `/messages` | List all messages (optional: `?channelId=&channelHash=&since=&limit=50`) | ### Stubbed (501 Not Implemented) @@ -225,9 +229,6 @@ Base path: `/api/v1` | `GET` | `/observers/{observerId}` | Get observer detail | | `GET` | `/observers/{observerId}/telemetry` | Observer telemetry history | | `GET` | `/observers/{observerId}/adverts` | Adverts heard by observer | -| `GET` | `/channels` | List channels | -| `GET` | `/channels/{channelHash}` | Get channel detail | -| `GET` | `/channels/{channelHash}/messages` | List channel messages | | `GET` | `/stats/overview` | Network overview stats | | `GET` | `/stats/observations` | Observation time series | | `GET` | `/stats/payloadBreakdown` | Observations by payload type | diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 29900b1..60196e3 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -267,7 +267,7 @@ VALUES ($1, $2, $3, $4, $5) ON CONFLICT (packet_hash) DO NOTHING; -- name: ListChannelMessages :many -SELECT cm.*, encode(p.packet_hash, 'hex') as packet_hash_hex, c.channel_hash +SELECT cm.*, encode(cm.packet_hash, 'hex') as packet_hash_hex, c.channel_hash FROM channel_messages cm JOIN packets p ON p.packet_hash = cm.packet_hash JOIN channels c ON c.id = cm.channel_id @@ -276,6 +276,15 @@ WHERE cm.channel_id = $1 ORDER BY cm.sent_at DESC LIMIT $3; +-- name: ListAllChannelMessages :many +SELECT cm.*, encode(cm.packet_hash, 'hex') as packet_hash_hex, c.channel_hash +FROM channel_messages cm +JOIN packets p ON p.packet_hash = cm.packet_hash +JOIN channels c ON c.id = cm.channel_id +WHERE ($1::timestamptz IS NULL OR cm.sent_at >= $1) +ORDER BY cm.sent_at DESC +LIMIT $2; + -- name: ListChannelMessagesByHash :many SELECT cm.*, c.channel_hash FROM channel_messages cm JOIN channels c ON c.id = cm.channel_id diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index eb9beab..9bae5ac 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -556,8 +556,65 @@ func (q *Queries) InsertObservation(ctx context.Context, arg InsertObservationPa return i, err } +const listAllChannelMessages = `-- name: ListAllChannelMessages :many +SELECT 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 +FROM channel_messages cm +JOIN packets p ON p.packet_hash = cm.packet_hash +JOIN channels c ON c.id = cm.channel_id +WHERE ($1::timestamptz IS NULL OR cm.sent_at >= $1) +ORDER BY cm.sent_at DESC +LIMIT $2 +` + +type ListAllChannelMessagesParams struct { + Column1 pgtype.Timestamptz `json:"column_1"` + Limit int32 `json:"limit"` +} + +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"` +} + +func (q *Queries) ListAllChannelMessages(ctx context.Context, arg ListAllChannelMessagesParams) ([]ListAllChannelMessagesRow, error) { + rows, err := q.db.Query(ctx, listAllChannelMessages, arg.Column1, arg.Limit) + if err != nil { + return nil, err + } + defer rows.Close() + items := []ListAllChannelMessagesRow{} + for rows.Next() { + var i ListAllChannelMessagesRow + if err := rows.Scan( + &i.ID, + &i.ChannelID, + &i.PacketHash, + &i.SenderName, + &i.SenderPubkey, + &i.Content, + &i.SentAt, + &i.PacketHashHex, + &i.ChannelHash, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const listChannelMessages = `-- name: ListChannelMessages :many -SELECT cm.id, cm.channel_id, cm.packet_hash, cm.sender_name, cm.sender_pubkey, cm.content, cm.sent_at, encode(p.packet_hash, 'hex') as packet_hash_hex, c.channel_hash +SELECT 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 FROM channel_messages cm JOIN packets p ON p.packet_hash = cm.packet_hash JOIN channels c ON c.id = cm.channel_id diff --git a/db/store.go b/db/store.go index 7da812b..d69f2cc 100644 --- a/db/store.go +++ b/db/store.go @@ -424,33 +424,33 @@ func (s *Store) GetChannel(ctx context.Context, channelID int32) (*api.Channel, // ListChannelMessages returns paginated messages for a channel identified by its integer ID. // Used by the /channels/{id}/messages endpoint. // Pass a zero time.Time for since to return all messages up to limit. -func (s *Store) ListChannelMessages(ctx context.Context, channelID int32, since time.Time, limit int32) ([]api.ChannelMessage, error) { - rows, err := s.q.ListChannelMessages(ctx, sqlc.ListChannelMessagesParams{ - ChannelID: channelID, - Column2: pgtype.Timestamptz{Time: since, Valid: !since.IsZero()}, - Limit: limit, - }) - if err != nil { - return nil, err - } - messages := make([]api.ChannelMessage, 0, len(rows)) - for _, v := range rows { - senderName := "" - if v.SenderName != nil { - senderName = *v.SenderName - } - content := "" - if v.Content != nil { - content = *v.Content - } - messages = append(messages, api.ChannelMessage{ - ID: v.ID, - PacketHash: v.PacketHashHex, - ChannelHash: hex.EncodeToString(v.ChannelHash), - SenderName: senderName, - Content: content, - SentAt: v.SentAt.Time.Format(time.RFC3339), +func (s *Store) ListChannelMessages(ctx context.Context, channelID *int32, since time.Time, limit int32) ([]api.ChannelMessage, error) { + var messages []api.ChannelMessage + if channelID == nil { + rows, err := s.q.ListAllChannelMessages(ctx, sqlc.ListAllChannelMessagesParams{ + Column1: pgtype.Timestamptz{Time: since, Valid: !since.IsZero()}, + Limit: limit, }) + messages = make([]api.ChannelMessage, 0, len(rows)) + if err != nil { + return nil, err + } + for _, v := range rows { + messages = append(messages, toChannelMessage(v.ID, v.PacketHashHex, v.ChannelHash, v.SenderName, v.Content, v.SentAt)) + } + } else { + rows, err := s.q.ListChannelMessages(ctx, sqlc.ListChannelMessagesParams{ + ChannelID: *channelID, + Column2: pgtype.Timestamptz{Time: since, Valid: !since.IsZero()}, + Limit: limit, + }) + messages = make([]api.ChannelMessage, 0, len(rows)) + if err != nil { + return nil, err + } + for _, v := range rows { + messages = append(messages, toChannelMessage(v.ID, v.PacketHashHex, v.ChannelHash, v.SenderName, v.Content, v.SentAt)) + } } return messages, nil } @@ -470,22 +470,26 @@ func (s *Store) ListChannelMessagesByHash(ctx context.Context, hash []byte, sinc } messages := make([]api.ChannelMessage, 0, len(rows)) for _, v := range rows { - senderName := "" - if v.SenderName != nil { - senderName = *v.SenderName - } - content := "" - if v.Content != nil { - content = *v.Content - } - messages = append(messages, api.ChannelMessage{ - ID: v.ID, - ChannelHash: hex.EncodeToString(v.ChannelHash), - PacketHash: hex.EncodeToString(v.PacketHash), - SenderName: senderName, - Content: content, - SentAt: v.SentAt.Time.Format(time.RFC3339), - }) + messages = append(messages, toChannelMessage(v.ID, hex.EncodeToString(v.PacketHash), v.ChannelHash, v.SenderName, v.Content, v.SentAt)) } return messages, nil } + +func toChannelMessage(id int64, packetHashHex string, channelHash []byte, senderName *string, content *string, sentAt pgtype.Timestamptz) api.ChannelMessage { + sn := "" + if senderName != nil { + sn = *senderName + } + ct := "" + if content != nil { + ct = *content + } + return api.ChannelMessage{ + ID: id, + PacketHash: packetHashHex, + ChannelHash: hex.EncodeToString(channelHash), + SenderName: sn, + Content: ct, + SentAt: sentAt.Time.Format(time.RFC3339), + } +} diff --git a/internal/api/handlers/channels.go b/internal/api/handlers/channels.go index 2046119..3deebd6 100644 --- a/internal/api/handlers/channels.go +++ b/internal/api/handlers/channels.go @@ -32,9 +32,8 @@ func ChannelsRouter(reader api.Reader) http.Handler { if err != nil { respond(w, http.StatusBadRequest, map[string]string{"error": "limit must be an integer"}) return - } else { - limit = l } + limit = l } var channels []api.ChannelSummary var err error @@ -119,7 +118,8 @@ func ChannelsRouter(reader api.Reader) http.Handler { } since = time.UnixMilli(ms) } - messages, err := reader.ListChannelMessages(r.Context(), int32(id), since, int32(limit)) + chanID := int32(id) + messages, err := reader.ListChannelMessages(r.Context(), &chanID, since, int32(limit)) if err != nil { respond(w, http.StatusInternalServerError, map[string]string{"error": "internal server error"}) return diff --git a/internal/api/handlers/messages.go b/internal/api/handlers/messages.go new file mode 100644 index 0000000..108ac60 --- /dev/null +++ b/internal/api/handlers/messages.go @@ -0,0 +1,93 @@ +package handlers + +import ( + "encoding/hex" + "net/http" + "strconv" + "time" + + "github.com/MeshCore-Tower/tower-server/internal/api" + "github.com/go-chi/chi/v5" +) + +// MessagesRouter mounts all /messages routes onto a subrouter. +// +// GET /messages → ListMessages +func MessagesRouter(reader api.Reader) http.Handler { + r := chi.NewRouter() + + // ListMessages handles GET /api/v1/messages + // + // Query params (all optional) + // + // since= + // limit=50 + // + // (mutually exclusive — provide one or neither, not both): + // + // channelId= filter by channel integer ID + // channelHash= filter by channel hash byte + // + r.Get("/", func(w http.ResponseWriter, r *http.Request) { + channelIDParam := r.URL.Query().Get("channelID") + channelHashParam := r.URL.Query().Get("channelHash") + if channelIDParam != "" && channelHashParam != "" { + respond(w, http.StatusBadRequest, map[string]string{"error": "filter by either channelId or channelHash, not both"}) + return + } + + var id int64 + if channelIDParam != "" { + i, err := strconv.ParseInt(channelIDParam, 10, 32) + if err != nil { + respond(w, http.StatusBadRequest, map[string]string{"error": "channelID should be an int 32"}) + return + } + id = i + } + var limit int64 = 50 + if limitParam := r.URL.Query().Get("limit"); limitParam != "" { + l, err := strconv.ParseInt(limitParam, 10, 32) + if err != nil { + respond(w, http.StatusBadRequest, map[string]string{"error": "limit must be an integer"}) + return + } + limit = l + } + var since time.Time + if sinceParam := r.URL.Query().Get("since"); sinceParam != "" { + ms, err := strconv.ParseInt(sinceParam, 10, 64) + if err != nil { + respond(w, http.StatusBadRequest, map[string]string{"error": "since must be epoch milliseconds"}) + return + } + since = time.UnixMilli(ms) + } + var messages []api.ChannelMessage + var err error + if channelHashParam != "" { + hashHex, decodeErr := hex.DecodeString(channelHashParam) + if decodeErr != nil { + respond(w, http.StatusBadRequest, map[string]string{"error": "invalid channel hash"}) + return + } + if len(hashHex) != 1 { + respond(w, http.StatusBadRequest, map[string]string{"error": "channel hash must be a single hex byte"}) + return + } + messages, err = reader.ListChannelMessagesByHash(r.Context(), hashHex, since, int32(limit)) + } else if channelIDParam != "" { + chanID := int32(id) + messages, err = reader.ListChannelMessages(r.Context(), &chanID, since, int32(limit)) + } else { + messages, err = reader.ListChannelMessages(r.Context(), nil, since, int32(limit)) + } + if err != nil { + respond(w, http.StatusInternalServerError, map[string]string{"error": "internal server error"}) + return + } + respond(w, http.StatusOK, messages) + }) + + return r +} diff --git a/internal/api/reader.go b/internal/api/reader.go index ddced6b..f41767e 100644 --- a/internal/api/reader.go +++ b/internal/api/reader.go @@ -88,7 +88,7 @@ type Reader interface { // ListChannelMessages returns paginated messages for a channel identified by its integer ID. // Used by the /channels/{id}/messages endpoint. // Pass a zero time.Time for since to return all messages up to limit. - ListChannelMessages(ctx context.Context, channelID int32, since time.Time, limit int32) ([]ChannelMessage, error) + ListChannelMessages(ctx context.Context, channelID *int32, since time.Time, limit int32) ([]ChannelMessage, error) // ListChannelMessagesByHash returns paginated messages for all channels matching the given hash. // Used by the /messages?hash= endpoint. May return messages from multiple channels // if the hash collides across different keys. diff --git a/internal/api/router/router.go b/internal/api/router/router.go index bdae71f..6e77c30 100644 --- a/internal/api/router/router.go +++ b/internal/api/router/router.go @@ -51,6 +51,7 @@ func New(h *hub.Hub, reader api.Reader) http.Handler { r.Mount("/nodes", handlers.NodesRouter()) r.Mount("/observers", handlers.ObserversRouter()) r.Mount("/channels", handlers.ChannelsRouter(reader)) + r.Mount("/messages", handlers.MessagesRouter(reader)) r.Mount("/iatas", handlers.IATAsRouter(reader)) r.Mount("/regions", handlers.RegionsRouter(reader)) r.Mount("/stats", handlers.StatsRouter())