From 90405bd07fb70a999d8ee422fb71c095e6bbc349 Mon Sep 17 00:00:00 2001 From: "Enot (ded) Skelly" Date: Thu, 28 May 2026 09:47:11 -0700 Subject: [PATCH] breaking: add pagination this adds pagination so the returned format changes to include results as well as a bool for hasMore and the cursor if any --- README.md | 6 +- db/queries/queries.sql | 32 ++++--- db/sqlc/queries.sql.go | 68 +++++++++++---- db/store.go | 134 +++++++++++++++++++++++------ internal/api/handlers/channels.go | 24 +++++- internal/api/handlers/messages.go | 22 +++-- internal/api/handlers/observers.go | 22 ++++- internal/api/reader.go | 17 ++-- 8 files changed, 249 insertions(+), 76 deletions(-) diff --git a/README.md b/README.md index 98b1637..b3bb534 100644 --- a/README.md +++ b/README.md @@ -304,16 +304,16 @@ Tower server. - [x] REST API: IATAs, Regions - [x] REST API: Channels (list + detail + messages) with IATA filter - [x] REST API: Messages (cross-channel) with IATA filter -- [x] REST API: Observers (telemetry, list + detail with broker last-seen) +- [x] REST API: Observers (heard adverts, telemetry, list + detail with broker + last-seen) - [x] REST API: Brokers (list with connection status) +- [x] REST API: Pagination ### In progress / next - [ ] REST API: Nodes (list + detail + observations) - [ ] REST API: Packets (list + detail) -- [ ] REST API: Observer + adverts sub-endpoints - [ ] REST API: Stats -- [ ] REST API: Pagination - [ ] Path resolution (node short ID lookup) - [ ] Propagation time calculation - [ ] Routes and traces endpoints diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 8ed74a5..1bb7ddf 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -65,6 +65,8 @@ WHERE observer_id = $1 ORDER BY last_seen DESC; -- name: ListObservers :many +-- Pass cursor=0 to start from the beginning, or the last seen observer's rownum for pagination. +-- Note: observers use UUID PKs so we order by last_seen and use a keyset on last_seen+id. SELECT o.id, o.display_name, @@ -95,8 +97,10 @@ WHERE WHEN o.last_status_at > NOW() - INTERVAL '5 minutes' THEN 'online' ELSE 'offline' END = $4) + AND ($5::timestamptz IS NULL OR o.last_seen < $5) GROUP BY o.id -ORDER BY o.last_seen DESC; +ORDER BY o.last_seen DESC +LIMIT $6; -- name: GetObserverLastIATA :one SELECT iata FROM packet_observations @@ -323,7 +327,8 @@ WHERE channel_hash = $1 AND key_fingerprint = $2; -- name: ListChannels :many -- Returns channels ordered by last seen, optionally filtered by hash and/or IATA. -- Pass NULL for hash to skip hash filtering. Pass empty string for iata to skip IATA filtering. --- IATA filter returns channels that have been active (have messages heard) in that IATA. +-- IATA filter returns channels that have active packets in that IATA. +-- Pass cursor=0 to start from the beginning (cursor is last_seen epoch ms). SELECT DISTINCT c.* FROM channels c WHERE ($1::bytea IS NULL OR c.channel_hash = $1) AND ($2 = '' OR EXISTS ( @@ -332,8 +337,9 @@ WHERE ($1::bytea IS NULL OR c.channel_hash = $1) WHERE p.channel_hash = c.channel_hash AND po.iata = $2 )) + AND ($3::timestamptz IS NULL OR c.last_seen < $3) ORDER BY c.last_seen DESC -LIMIT $3; +LIMIT $4; -- name: GetChannelsByHash :many -- Returns all channels for a given hash (may be multiple on hash collision). @@ -363,6 +369,7 @@ RETURNING id; -- 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 FROM channel_messages cm JOIN channels c ON c.id = cm.channel_id @@ -370,33 +377,38 @@ JOIN packet_observations po ON po.packet_hash = cm.packet_hash WHERE cm.channel_id = $1 AND ($2::timestamptz IS NULL OR cm.sent_at >= $2) AND ($3 = '' OR po.iata = $3) -ORDER BY cm.id, cm.sent_at DESC -LIMIT $4; + AND ($4 = 0 OR cm.id > $4) +ORDER BY cm.id ASC +LIMIT $5; -- name: ListAllChannelMessages :many --- Returns all messages across all channels with optional time and IATA filters. +-- Returns all messages across all channels with optional time, IATA and cursor filters. -- 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 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 ($1::timestamptz IS NULL OR cm.sent_at >= $1) AND ($2 = '' OR po.iata = $2) -ORDER BY cm.id, cm.sent_at DESC -LIMIT $3; + AND ($3 = 0 OR cm.id > $3) +ORDER BY cm.id ASC +LIMIT $4; -- name: ListChannelMessagesByHash :many -- Returns messages for all channels matching a hash byte. -- May return messages from multiple channels if the hash collides across different keys. -- Pass empty string for iata to skip IATA filtering. +-- Pass cursor=0 to start from the beginning. SELECT DISTINCT ON (cm.id) cm.*, c.channel_hash 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 AND ($2::timestamptz IS NULL OR cm.sent_at >= $2) AND ($3 = '' OR po.iata = $3) -ORDER BY cm.id, cm.sent_at DESC -LIMIT $4; + AND ($4 = 0 OR cm.id > $4) +ORDER BY cm.id ASC +LIMIT $5; -- name: InsertObserverTelemetry :exec -- Inserts a telemetry snapshot for an observer. The reported_at timestamp should diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 59f6af4..919c61f 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -759,13 +759,15 @@ JOIN channels c ON c.id = cm.channel_id JOIN packet_observations po ON po.packet_hash = cm.packet_hash WHERE ($1::timestamptz IS NULL OR cm.sent_at >= $1) AND ($2 = '' OR po.iata = $2) -ORDER BY cm.id, cm.sent_at DESC -LIMIT $3 + AND ($3 = 0 OR cm.id > $3) +ORDER BY cm.id ASC +LIMIT $4 ` type ListAllChannelMessagesParams struct { Column1 pgtype.Timestamptz `json:"column_1"` Column2 interface{} `json:"column_2"` + Column3 interface{} `json:"column_3"` Limit int32 `json:"limit"` } @@ -781,10 +783,16 @@ type ListAllChannelMessagesRow struct { ChannelHash []byte `json:"channel_hash"` } -// Returns all messages across all channels with optional time and IATA filters. +// Returns all messages across all channels with optional time, IATA and cursor filters. // Pass empty string for iata to skip IATA filtering. +// Pass cursor=0 to start from the beginning. func (q *Queries) ListAllChannelMessages(ctx context.Context, arg ListAllChannelMessagesParams) ([]ListAllChannelMessagesRow, error) { - rows, err := q.db.Query(ctx, listAllChannelMessages, arg.Column1, arg.Column2, arg.Limit) + rows, err := q.db.Query(ctx, listAllChannelMessages, + arg.Column1, + arg.Column2, + arg.Column3, + arg.Limit, + ) if err != nil { return nil, err } @@ -821,14 +829,16 @@ JOIN packet_observations po ON po.packet_hash = cm.packet_hash WHERE cm.channel_id = $1 AND ($2::timestamptz IS NULL OR cm.sent_at >= $2) AND ($3 = '' OR po.iata = $3) -ORDER BY cm.id, cm.sent_at DESC -LIMIT $4 + AND ($4 = 0 OR cm.id > $4) +ORDER BY cm.id ASC +LIMIT $5 ` type ListChannelMessagesParams struct { ChannelID int32 `json:"channel_id"` Column2 pgtype.Timestamptz `json:"column_2"` Column3 interface{} `json:"column_3"` + Column4 interface{} `json:"column_4"` Limit int32 `json:"limit"` } @@ -847,11 +857,13 @@ type ListChannelMessagesRow struct { // 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. func (q *Queries) ListChannelMessages(ctx context.Context, arg ListChannelMessagesParams) ([]ListChannelMessagesRow, error) { rows, err := q.db.Query(ctx, listChannelMessages, arg.ChannelID, arg.Column2, arg.Column3, + arg.Column4, arg.Limit, ) if err != nil { @@ -889,14 +901,16 @@ JOIN packet_observations po ON po.packet_hash = cm.packet_hash WHERE c.channel_hash = $1 AND ($2::timestamptz IS NULL OR cm.sent_at >= $2) AND ($3 = '' OR po.iata = $3) -ORDER BY cm.id, cm.sent_at DESC -LIMIT $4 + AND ($4 = 0 OR cm.id > $4) +ORDER BY cm.id ASC +LIMIT $5 ` type ListChannelMessagesByHashParams struct { ChannelHash []byte `json:"channel_hash"` Column2 pgtype.Timestamptz `json:"column_2"` Column3 interface{} `json:"column_3"` + Column4 interface{} `json:"column_4"` Limit int32 `json:"limit"` } @@ -914,11 +928,13 @@ type ListChannelMessagesByHashRow struct { // Returns messages for all channels matching a hash byte. // May return messages from multiple channels if the hash collides across different keys. // Pass empty string for iata to skip IATA filtering. +// Pass cursor=0 to start from the beginning. func (q *Queries) ListChannelMessagesByHash(ctx context.Context, arg ListChannelMessagesByHashParams) ([]ListChannelMessagesByHashRow, error) { rows, err := q.db.Query(ctx, listChannelMessagesByHash, arg.ChannelHash, arg.Column2, arg.Column3, + arg.Column4, arg.Limit, ) if err != nil { @@ -957,21 +973,29 @@ WHERE ($1::bytea IS NULL OR c.channel_hash = $1) WHERE p.channel_hash = c.channel_hash AND po.iata = $2 )) + AND ($3::timestamptz IS NULL OR c.last_seen < $3) ORDER BY c.last_seen DESC -LIMIT $3 +LIMIT $4 ` type ListChannelsParams struct { - Column1 []byte `json:"column_1"` - Column2 interface{} `json:"column_2"` - Limit int32 `json:"limit"` + Column1 []byte `json:"column_1"` + Column2 interface{} `json:"column_2"` + Column3 pgtype.Timestamptz `json:"column_3"` + Limit int32 `json:"limit"` } // Returns channels ordered by last seen, optionally filtered by hash and/or IATA. // Pass NULL for hash to skip hash filtering. Pass empty string for iata to skip IATA filtering. -// IATA filter returns channels that have been active (have messages heard) in that IATA. +// IATA filter returns channels that have active packets in that IATA. +// Pass cursor=0 to start from the beginning (cursor is last_seen epoch ms). func (q *Queries) ListChannels(ctx context.Context, arg ListChannelsParams) ([]Channel, error) { - rows, err := q.db.Query(ctx, listChannels, arg.Column1, arg.Column2, arg.Limit) + rows, err := q.db.Query(ctx, listChannels, + arg.Column1, + arg.Column2, + arg.Column3, + arg.Limit, + ) if err != nil { return nil, err } @@ -1281,15 +1305,19 @@ WHERE WHEN o.last_status_at > NOW() - INTERVAL '5 minutes' THEN 'online' ELSE 'offline' END = $4) + AND ($5::timestamptz IS NULL OR o.last_seen < $5) GROUP BY o.id ORDER BY o.last_seen DESC +LIMIT $6 ` type ListObserversParams struct { - Column1 interface{} `json:"column_1"` - Column2 interface{} `json:"column_2"` - Column3 interface{} `json:"column_3"` - Column4 interface{} `json:"column_4"` + Column1 interface{} `json:"column_1"` + Column2 interface{} `json:"column_2"` + Column3 interface{} `json:"column_3"` + Column4 interface{} `json:"column_4"` + Column5 pgtype.Timestamptz `json:"column_5"` + Limit int32 `json:"limit"` } type ListObserversRow struct { @@ -1301,12 +1329,16 @@ type ListObserversRow struct { Iata string `json:"iata"` } +// Pass cursor=0 to start from the beginning, or the last seen observer's rownum for pagination. +// Note: observers use UUID PKs so we order by last_seen and use a keyset on last_seen+id. func (q *Queries) ListObservers(ctx context.Context, arg ListObserversParams) ([]ListObserversRow, error) { rows, err := q.db.Query(ctx, listObservers, arg.Column1, arg.Column2, arg.Column3, arg.Column4, + arg.Column5, + arg.Limit, ) if err != nil { return nil, err diff --git a/db/store.go b/db/store.go index 00024e8..c8bebdd 100644 --- a/db/store.go +++ b/db/store.go @@ -382,20 +382,30 @@ func (s *Store) UpsertRegionIATA(ctx context.Context, regionID int32, iata strin // ListChannels returns a summary list of all known channels ordered by last seen. // Pass nil hash to skip hash filtering. Pass empty string iata to return channels from all IATAs. // IATA filtering returns channels that have messages heard in the given IATA. -func (s *Store) ListChannels(ctx context.Context, limit int32, hash []byte, iata string) ([]api.ChannelSummary, error) { - // Note: after sqlc generate, verify Column1/Column2/Column3 param names - // match what sqlc generated for the updated ListChannels query. +// ListChannels returns a paginated list of channels ordered by last seen. +// cursor is last_seen epoch ms of the last item; pass 0 to start from the beginning. +// Note: after sqlc generate, verify Column param names match generated types. +func (s *Store) ListChannels(ctx context.Context, limit int32, hash []byte, iata string, cursor int64) (api.Page[api.ChannelSummary], error) { + var cursorTs pgtype.Timestamptz + if cursor > 0 { + cursorTs = pgtype.Timestamptz{Time: time.UnixMilli(cursor), Valid: true} + } rows, err := s.q.ListChannels(ctx, sqlc.ListChannelsParams{ Column1: hash, Column2: iata, - Limit: limit, + Column3: cursorTs, + Limit: limit + 1, }) if err != nil { - return nil, err + return api.Page[api.ChannelSummary]{}, err } - channels := make([]api.ChannelSummary, 0, len(rows)) + hasMore := len(rows) > int(limit) + if hasMore { + rows = rows[:limit] + } + items := make([]api.ChannelSummary, 0, len(rows)) for _, v := range rows { - channels = append(channels, api.ChannelSummary{ + items = append(items, api.ChannelSummary{ ID: int(v.ID), Name: v.Name, ChannelHash: hex.EncodeToString(v.ChannelHash), @@ -404,7 +414,16 @@ func (s *Store) ListChannels(ctx context.Context, limit int32, hash []byte, iata KeyKnown: v.KeyKnown != nil && *v.KeyKnown, }) } - return channels, nil + var nextCursor *int64 + if hasMore { + last := items[len(items)-1].LastSeen + nextCursor = &last + } + return api.Page[api.ChannelSummary]{ + Items: items, + NextCursor: nextCursor, + HasMore: hasMore, + }, nil } // GetChannel returns full detail for a single channel by its integer ID. @@ -436,22 +455,29 @@ func (s *Store) GetChannel(ctx context.Context, channelID int32) (*api.Channel, return &channel, nil } -// ListChannelMessages returns paginated messages with optional channel ID, time and IATA filters. +// ListChannelMessages returns paginated messages with optional channel ID, time, IATA and cursor filters. // Pass nil channelID to return messages across all channels. // Pass a zero time.Time for since to return all messages up to limit. // Pass empty string iata to return messages from all IATAs. -// Note: after sqlc generate, verify generated param field names match. -func (s *Store) ListChannelMessages(ctx context.Context, channelID *int32, since time.Time, limit int32, iata string) ([]api.ChannelMessage, error) { +// Pass cursor=0 to start from the beginning. +func (s *Store) ListChannelMessages(ctx context.Context, channelID *int32, since time.Time, limit int32, iata string, cursor int64) (api.Page[api.ChannelMessage], error) { ts := pgtype.Timestamptz{Time: since, Valid: !since.IsZero()} var messages []api.ChannelMessage + var hasMore bool + if channelID == nil { rows, err := s.q.ListAllChannelMessages(ctx, sqlc.ListAllChannelMessagesParams{ Column1: ts, Column2: iata, - Limit: limit, + Column3: cursor, + Limit: limit + 1, }) if err != nil { - return nil, err + return api.Page[api.ChannelMessage]{}, err + } + hasMore = len(rows) > int(limit) + if hasMore { + rows = rows[:limit] } messages = make([]api.ChannelMessage, 0, len(rows)) for _, v := range rows { @@ -462,57 +488,97 @@ func (s *Store) ListChannelMessages(ctx context.Context, channelID *int32, since ChannelID: *channelID, Column2: ts, Column3: iata, - Limit: limit, + Column4: cursor, + Limit: limit + 1, }) if err != nil { - return nil, err + return api.Page[api.ChannelMessage]{}, err + } + hasMore = len(rows) > int(limit) + if hasMore { + rows = rows[:limit] } 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)) } } - return messages, nil + + var nextCursor *int64 + if hasMore && len(messages) > 0 { + last := messages[len(messages)-1].ID + nextCursor = &last + } + return api.Page[api.ChannelMessage]{ + Items: messages, + NextCursor: nextCursor, + HasMore: hasMore, + }, nil } // 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. +// May return messages from multiple channels if the hash collides across different keys. // Pass a zero time.Time for since to return all messages up to limit. // Pass empty string iata to return messages from all IATAs. -// Note: after sqlc generate, verify generated param field names match. -func (s *Store) ListChannelMessagesByHash(ctx context.Context, hash []byte, since time.Time, limit int32, iata string) ([]api.ChannelMessage, error) { +// Pass cursor=0 to start from the beginning. +func (s *Store) ListChannelMessagesByHash(ctx context.Context, hash []byte, since time.Time, limit int32, iata string, cursor int64) (api.Page[api.ChannelMessage], error) { rows, err := s.q.ListChannelMessagesByHash(ctx, sqlc.ListChannelMessagesByHashParams{ ChannelHash: hash, Column2: pgtype.Timestamptz{Time: since, Valid: !since.IsZero()}, Column3: iata, - Limit: limit, + Column4: cursor, + Limit: limit + 1, }) if err != nil { - return nil, err + return api.Page[api.ChannelMessage]{}, err + } + hasMore := len(rows) > int(limit) + if hasMore { + rows = rows[:limit] } 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)) } - return messages, nil + var nextCursor *int64 + if hasMore && len(messages) > 0 { + last := messages[len(messages)-1].ID + nextCursor = &last + } + return api.Page[api.ChannelMessage]{ + Items: messages, + NextCursor: nextCursor, + HasMore: hasMore, + }, nil } // ListObservers returns a summary list of observers with optional filters. // All filter params are optional — pass empty string to skip a filter. // status is "online" or "offline" derived from last_status_at recency. -func (s *Store) ListObservers(ctx context.Context, iata, observerType, broker, status string) ([]api.ObserverSummary, error) { +// ListObservers returns a paginated list of observers with optional filters. +// cursor is last_seen epoch ms of the last observer; pass 0 to start from the beginning. +func (s *Store) ListObservers(ctx context.Context, iata, observerType, broker, status string, cursor int64, limit int32) (api.Page[api.ObserverSummary], error) { + var cursorTs pgtype.Timestamptz + if cursor > 0 { + cursorTs = pgtype.Timestamptz{Time: time.UnixMilli(cursor), Valid: true} + } params := sqlc.ListObserversParams{ Column1: iata, Column2: observerType, Column3: broker, Column4: status, + Column5: cursorTs, + Limit: limit + 1, } rows, err := s.q.ListObservers(ctx, params) if err != nil { - return nil, err + return api.Page[api.ObserverSummary]{}, err } - observers := make([]api.ObserverSummary, 0, len(rows)) + hasMore := len(rows) > int(limit) + if hasMore { + rows = rows[:limit] + } + items := make([]api.ObserverSummary, 0, len(rows)) for _, v := range rows { observer := api.ObserverSummary{ ID: v.ID, @@ -525,9 +591,21 @@ func (s *Store) ListObservers(ctx context.Context, iata, observerType, broker, s if v.ObserverType != nil { observer.ObserverType = v.ObserverType } - observers = append(observers, observer) + items = append(items, observer) } - return observers, nil + var nextCursor *int64 + if hasMore { + // observers use UUID so encode last_seen as cursor + if rows[len(rows)-1].LastStatusAt.Valid { + ms := rows[len(rows)-1].LastStatusAt.Time.UnixMilli() + nextCursor = &ms + } + } + return api.Page[api.ObserverSummary]{ + Items: items, + NextCursor: nextCursor, + HasMore: hasMore, + }, nil } // GetObserver returns full detail for a single observer by UUID. diff --git a/internal/api/handlers/channels.go b/internal/api/handlers/channels.go index 3558299..4bb8c90 100644 --- a/internal/api/handlers/channels.go +++ b/internal/api/handlers/channels.go @@ -24,6 +24,7 @@ func ChannelsRouter(reader api.Reader) http.Handler { // // hash= filter by single-byte channel hash // iata= filter by IATA code (channels with messages heard in that IATA) + // cursor= last_seen epoch ms of last item for pagination // limit=50 r.Get("/", func(w http.ResponseWriter, r *http.Request) { var limit int64 = 50 @@ -36,6 +37,15 @@ func ChannelsRouter(reader api.Reader) http.Handler { limit = l } iata := r.URL.Query().Get("iata") + var cursor int64 + if cursorParam := r.URL.Query().Get("cursor"); cursorParam != "" { + c, err := strconv.ParseInt(cursorParam, 10, 64) + if err != nil { + respondError(w, http.StatusBadRequest, "cursor must be an integer") + return + } + cursor = c + } var hashHex []byte if hash := r.URL.Query().Get("hash"); hash != "" { h, decodeErr := hex.DecodeString(hash) @@ -49,7 +59,7 @@ func ChannelsRouter(reader api.Reader) http.Handler { } hashHex = h } - channels, err := reader.ListChannels(r.Context(), int32(limit), hashHex, iata) + channels, err := reader.ListChannels(r.Context(), int32(limit), hashHex, iata, cursor) if err != nil { respondError(w, http.StatusInternalServerError, "internal server error") return @@ -85,6 +95,7 @@ func ChannelsRouter(reader api.Reader) http.Handler { // // since= return messages after this timestamp // iata= filter by IATA code + // cursor= message ID of last item for pagination // limit=50 // // Returns paginated decrypted channel messages. @@ -117,8 +128,17 @@ func ChannelsRouter(reader api.Reader) http.Handler { since = time.UnixMilli(ms) } iata := r.URL.Query().Get("iata") + var cursor int64 + if cursorParam := r.URL.Query().Get("cursor"); cursorParam != "" { + c, err := strconv.ParseInt(cursorParam, 10, 64) + if err != nil { + respondError(w, http.StatusBadRequest, "cursor must be an integer") + return + } + cursor = c + } chanID := int32(id) - messages, err := reader.ListChannelMessages(r.Context(), &chanID, since, int32(limit), iata) + messages, err := reader.ListChannelMessages(r.Context(), &chanID, since, int32(limit), iata, cursor) if err != nil { respondError(w, http.StatusInternalServerError, "internal server error") return diff --git a/internal/api/handlers/messages.go b/internal/api/handlers/messages.go index 49ec46c..b4c20e9 100644 --- a/internal/api/handlers/messages.go +++ b/internal/api/handlers/messages.go @@ -22,6 +22,7 @@ func MessagesRouter(reader api.Reader) http.Handler { // // since= return messages after this timestamp // iata= filter by IATA code + // cursor= message ID of last item for pagination // limit=50 // // Mutually exclusive — provide one or neither, not both: @@ -64,7 +65,17 @@ func MessagesRouter(reader api.Reader) http.Handler { } since = time.UnixMilli(ms) } - var messages []api.ChannelMessage + var cursor int64 + if cursorParam := r.URL.Query().Get("cursor"); cursorParam != "" { + c, err := strconv.ParseInt(cursorParam, 10, 64) + if err != nil { + respondError(w, http.StatusBadRequest, "cursor must be an integer") + return + } + cursor = c + } + iata := r.URL.Query().Get("iata") + var messages api.Page[api.ChannelMessage] var err error if channelHashParam != "" { hashHex, decodeErr := hex.DecodeString(channelHashParam) @@ -76,15 +87,12 @@ func MessagesRouter(reader api.Reader) http.Handler { respondError(w, http.StatusBadRequest, "channel hash must be a single hex byte") return } - iata := r.URL.Query().Get("iata") - messages, err = reader.ListChannelMessagesByHash(r.Context(), hashHex, since, int32(limit), iata) + messages, err = reader.ListChannelMessagesByHash(r.Context(), hashHex, since, int32(limit), iata, cursor) } else if channelIDParam != "" { - iata := r.URL.Query().Get("iata") chanID := int32(id) - messages, err = reader.ListChannelMessages(r.Context(), &chanID, since, int32(limit), iata) + messages, err = reader.ListChannelMessages(r.Context(), &chanID, since, int32(limit), iata, cursor) } else { - iata := r.URL.Query().Get("iata") - messages, err = reader.ListChannelMessages(r.Context(), nil, since, int32(limit), iata) + messages, err = reader.ListChannelMessages(r.Context(), nil, since, int32(limit), iata, cursor) } if err != nil { respondError(w, http.StatusInternalServerError, "internal server error") diff --git a/internal/api/handlers/observers.go b/internal/api/handlers/observers.go index c6b241f..dc326e9 100644 --- a/internal/api/handlers/observers.go +++ b/internal/api/handlers/observers.go @@ -27,12 +27,32 @@ func ObserversRouter(reader api.Reader) http.Handler { // type=meshcoretomqtt // broker=mqtt1 // status=online + // cursor= last_seen epoch ms of last observer for pagination + // limit=50 r.Get("/", func(w http.ResponseWriter, r *http.Request) { iata := r.URL.Query().Get("iata") observerType := r.URL.Query().Get("type") broker := r.URL.Query().Get("broker") status := r.URL.Query().Get("status") - observers, err := reader.ListObservers(r.Context(), iata, observerType, broker, status) + var cursor int64 + if cursorParam := r.URL.Query().Get("cursor"); cursorParam != "" { + c, err := strconv.ParseInt(cursorParam, 10, 64) + if err != nil { + respondError(w, http.StatusBadRequest, "cursor must be an integer") + return + } + cursor = c + } + var limit int32 = 50 + if limitParam := r.URL.Query().Get("limit"); limitParam != "" { + l, err := strconv.ParseInt(limitParam, 10, 32) + if err != nil { + respondError(w, http.StatusBadRequest, "limit must be an integer") + return + } + limit = int32(l) + } + observers, err := reader.ListObservers(r.Context(), iata, observerType, broker, status, cursor, limit) if err != nil { respondError(w, http.StatusInternalServerError, "failed to get list of observers") return diff --git a/internal/api/reader.go b/internal/api/reader.go index 6e80760..e9215dd 100644 --- a/internal/api/reader.go +++ b/internal/api/reader.go @@ -174,11 +174,11 @@ type Reader interface { // GetRegion returns full detail for a single region including its associated IATA codes. // Returns nil, pgx.ErrNoRows if the region is not found. GetRegion(ctx context.Context, regionID int32) (*Region, error) - // ListChannels returns a summary list of all known channels ordered by last seen. + // ListChannels returns a paginated list of channels ordered by last seen. // Includes both hashtag-derived and explicit key channels. - // Channels with unknown keys are included with KeyKnown=false. // Pass nil hash to skip hash filtering. Pass empty string iata to return all channels. - ListChannels(ctx context.Context, limit int32, hash []byte, iata string) ([]ChannelSummary, error) + // cursor is last_seen epoch ms of the last item; pass 0 to start from the beginning. + ListChannels(ctx context.Context, limit int32, hash []byte, iata string, cursor int64) (Page[ChannelSummary], error) // GetChannel returns full detail for a single channel by its integer ID. // Returns nil, pgx.ErrNoRows if the channel is not found. GetChannel(ctx context.Context, channelID int32) (*Channel, error) @@ -186,17 +186,20 @@ type Reader interface { // Used by the /channels/{id}/messages endpoint. // Pass a zero time.Time for since to return all messages up to limit. // Pass empty string iata to return messages from all IATAs. - ListChannelMessages(ctx context.Context, channelID *int32, since time.Time, limit int32, iata string) ([]ChannelMessage, error) + // Pass cursor=0 to start from the beginning. + ListChannelMessages(ctx context.Context, channelID *int32, since time.Time, limit int32, iata string, cursor int64) (Page[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. // Pass a zero time.Time for since to return all messages up to limit. // Pass empty string iata to return messages from all IATAs. - ListChannelMessagesByHash(ctx context.Context, hash []byte, since time.Time, limit int32, iata string) ([]ChannelMessage, error) - // ListObservers returns a summary list of observers with optional filters. + // Pass cursor=0 to start from the beginning. + ListChannelMessagesByHash(ctx context.Context, hash []byte, since time.Time, limit int32, iata string, cursor int64) (Page[ChannelMessage], error) + // ListObservers returns a paginated list of observers with optional filters. // All filter params are optional — pass empty string to skip a filter. // status is "online" or "offline" derived from last_status_at recency. - ListObservers(ctx context.Context, iata, observerType, broker, status string) ([]ObserverSummary, error) + // cursor is last_seen epoch ms of the last observer; pass 0 to start from the beginning. + ListObservers(ctx context.Context, iata, observerType, broker, status string, cursor int64, limit int32) (Page[ObserverSummary], error) // GetObserver returns full detail for a single observer by UUID. // Returns nil, pgx.ErrNoRows if the observer is not found. GetObserver(ctx context.Context, observerID uuid.UUID) (*Observer, error)