add messages enpoints

messages at /channels as well as generic /messages

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=<hex>&since=<ms>&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=<int>&channelHash=<hex>&since=<ms>&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       |

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

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

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),
+	}
+}

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

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=<epoch ms>
+	//	limit=50
+	//
+	// (mutually exclusive — provide one or neither, not both):
+	//
+	//	channelId=<int32>   filter by channel integer ID
+	//	channelHash=<hex>   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
+}

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.

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())