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

@@ -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

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

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

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.

internal/api/handlers/channels.go

@@ -24,6 +24,7 @@ func ChannelsRouter(reader api.Reader) http.Handler {
 	//
 	//	hash=<hex>       filter by single-byte channel hash
 	//	iata=<code>      filter by IATA code (channels with messages heard in that IATA)
+	//	cursor=<int>     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=<epoch ms>   return messages after this timestamp
 		//	iata=<code>        filter by IATA code
+		//	cursor=<int>       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

internal/api/handlers/messages.go

@@ -22,6 +22,7 @@ func MessagesRouter(reader api.Reader) http.Handler {
 	//
 	//	since=<epoch ms>    return messages after this timestamp
 	//	iata=<code>         filter by IATA code
+	//	cursor=<int>        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")

internal/api/handlers/observers.go

@@ -27,12 +27,32 @@ func ObserversRouter(reader api.Reader) http.Handler {
 	//	type=meshcoretomqtt
 	//	broker=mqtt1
 	//	status=online
+	//	cursor=<int>     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

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)