Update from source

cmd/helpers.go

@@ -100,7 +100,7 @@ func doPageAll(cmd *cobra.Command, c *client.Client, path string, params map[str
 		Path:      path,
 		Params:    params,
 		Limit:     limit,
-		StartPage: max(startPage, 1),
+		StartPage: startPage,
 		Writer:    w,
 	})
 	if err != nil {

internal/client/pagination.go

@@ -31,6 +31,10 @@ type paginationMeta struct {
 	// collection array, taken as the largest top-level array when the key is
 	// not one of the well-known names.
 	DataLen int
+	// Continuation is the cursor of an endpoint that pages by a
+	// "continuation" query param instead of page numbers; nil when the
+	// response carries none, "" on the last page.
+	Continuation *string
 }
 
 const maxAutoPages = 10000
@@ -40,6 +44,8 @@ func (c *Client) PageAll(cfg PageAllConfig) error {
 	if cfg.Ctx == nil {
 		cfg.Ctx = context.Background()
 	}
+	// A page the caller did not choose is only a guess at the first page.
+	guessedFirstPage := cfg.StartPage <= 0
 	if cfg.StartPage <= 0 {
 		cfg.StartPage = 1
 	}
@@ -53,10 +59,19 @@ func (c *Client) PageAll(cfg PageAllConfig) error {
 	// and trusting the flag would end the walk early. A later, final page may
 	// be shorter, so only the first page is read.
 	pageSize := 0
+	// Set once a response carries a cursor; from then on the cursor, not the
+	// page number, selects the next page.
+	cursor := ""
+	seen := map[string]bool{}
 
 	for {
 		params := copyParams(cfg.Params)
-		params["page"] = strconv.Itoa(page)
+		if cursor != "" {
+			delete(params, "page")
+			params["continuation"] = cursor
+		} else {
+			params["page"] = strconv.Itoa(page)
+		}
 		if cfg.Limit > 0 {
 			params["limit"] = strconv.Itoa(cfg.Limit)
 		}
@@ -65,16 +80,53 @@ func (c *Client) PageAll(cfg PageAllConfig) error {
 		if err != nil {
 			return err
 		}
+		meta := extractPaginationMeta(result, cfg.Limit)
+
+		// Some cursor endpoints still honour page on the first request and
+		// count it from 0 (/backfills), so the guessed page=1 skipped the
+		// newest page. Refetch it without page and let the cursor take over.
+		if guessedFirstPage && cursor == "" && meta.Continuation != nil {
+			delete(params, "page")
+			if result, err = c.Do(cfg.Ctx, cfg.Method, cfg.Path, params, nil); err != nil {
+				return err
+			}
+			meta = extractPaginationMeta(result, cfg.Limit)
+		}
+		guessedFirstPage = false
 
 		if _, err := fmt.Fprintf(cfg.Writer, "%s\n", result); err != nil {
 			return fmt.Errorf("write page %d: %w", page, err)
 		}
 
-		meta := extractPaginationMeta(result, cfg.Limit)
 		if pageSize == 0 {
 			pageSize = meta.DataLen
 		}
 
+		// Mid-walk, a cursor that is missing or null ends it like "": the
+		// page-number stops below would resend the last cursor.
+		if meta.Continuation == nil && cursor != "" {
+			return nil
+		}
+		// A cursor endpoint ignores page, so the page-number stops below
+		// would refetch the first page. An empty page is not the end either:
+		// a cursor can step past a stretch with no matches.
+		if meta.Continuation != nil {
+			next := *meta.Continuation
+			if next == "" {
+				return nil
+			}
+			if seen[next] {
+				return fmt.Errorf("pagination stopped at page %d: the server returned continuation %q twice", page, next)
+			}
+			seen[next] = true
+			cursor = next
+			if page >= cfg.StartPage+maxAutoPages-1 {
+				return nil
+			}
+			page++
+			continue
+		}
+
 		if meta.TotalPages != nil && page >= *meta.TotalPages {
 			return nil
 		}
@@ -117,6 +169,12 @@ func extractPaginationMeta(data json.RawMessage, limit int) paginationMeta {
 	// under meta.pagination.
 	meta.TotalPages = intField(obj, "total_pages")
 	meta.Total = intField(obj, "total")
+	// Only meta.continuation is a cursor the walk can follow. A broadcast's
+	// next profiles carries one under meta.pagination, but it only pages with
+	// the from returned beside it and repeats rows either way.
+	if m := nestedObject(obj, "meta"); m != nil {
+		meta.Continuation = stringField(m, "continuation")
+	}
 	if pagination := nestedObject(obj, "meta", "pagination"); pagination != nil {
 		if meta.TotalPages == nil {
 			meta.TotalPages = intField(pagination, "total_pages")
@@ -181,6 +239,20 @@ func intField(obj map[string]json.RawMessage, key string) *int {
 	return &v
 }
 
+func stringField(obj map[string]json.RawMessage, key string) *string {
+	raw, ok := obj[key]
+	// json.Unmarshal leaves a string "" for null, which would read as a
+	// last-page cursor and end the walk after one page.
+	if !ok || isNull(raw) {
+		return nil
+	}
+	var v string
+	if json.Unmarshal(raw, &v) != nil {
+		return nil
+	}
+	return &v
+}
+
 func isNull(raw json.RawMessage) bool {
 	return bytes.Equal(bytes.TrimSpace(raw), []byte("null"))
 }

internal/client/pagination_test.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"net/http"
 	"net/http/httptest"
+	"net/url"
 	"strconv"
 	"strings"
 	"testing"
@@ -173,3 +174,213 @@ func TestExtractPaginationMeta(t *testing.T) {
 		})
 	}
 }
+
+// deliveriesServer mimics a cursor-paged Journeys endpoint: it ignores page
+// and limit, and serves pages[i] for the cursor "c<i>" (the first page for no
+// cursor). Each page's next cursor is "c<i+1>", or "" on the last.
+func deliveriesServer(t *testing.T, pages [][]int) (*httptest.Server, *[]string) {
+	t.Helper()
+	var requests []string
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requests = append(requests, r.URL.RawQuery)
+		i := 0
+		if c := r.URL.Query().Get("continuation"); c != "" {
+			i, _ = strconv.Atoi(strings.TrimPrefix(c, "c"))
+		}
+		next := ""
+		if i+1 < len(pages) {
+			next = "c" + strconv.Itoa(i+1)
+		}
+		w.Header().Set("Content-Type", "application/json")
+		if err := json.NewEncoder(w).Encode(map[string]any{
+			"deliveries": pages[i],
+			"meta":       map[string]string{"continuation": next},
+		}); err != nil {
+			t.Errorf("encode: %v", err)
+		}
+	}))
+	t.Cleanup(server.Close)
+	return server, &requests
+}
+
+func TestPageAll_FollowsContinuation(t *testing.T) {
+	server, requests := deliveriesServer(t, [][]int{{1, 2}, {3, 4}, {5}})
+	lines := pageAllLines(t, server, 0)
+
+	if len(lines) != 3 {
+		t.Fatalf("expected 3 pages, got %d: %v", len(lines), *requests)
+	}
+	// The guessed page=1 is refetched without page once a cursor shows up.
+	want := []string{"page=1", "", "continuation=c1", "continuation=c2"}
+	if len(*requests) != len(want) {
+		t.Fatalf("requests = %q, want %q", *requests, want)
+	}
+	for i, q := range *requests {
+		if q != want[i] {
+			t.Errorf("request %d query = %q, want %q", i, q, want[i])
+		}
+	}
+}
+
+// A cursor can step past a stretch with no matches, so an empty page with a
+// cursor is not the end.
+func TestPageAll_ContinuationPastEmptyPage(t *testing.T) {
+	server, requests := deliveriesServer(t, [][]int{{1}, {}, {2}})
+	lines := pageAllLines(t, server, 0)
+
+	if len(lines) != 3 {
+		t.Fatalf("expected 3 pages, got %d: %v", len(lines), *requests)
+	}
+	if !strings.Contains(lines[2], "2") {
+		t.Errorf("last page = %s, want the page after the empty one", lines[2])
+	}
+}
+
+// A null cursor after a real one ends the walk; it must not resend the last
+// cursor until the page cap.
+func TestPageAll_NullContinuationEndsCursorWalk(t *testing.T) {
+	var requests []string
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requests = append(requests, r.URL.RawQuery)
+		w.Header().Set("Content-Type", "application/json")
+		if r.URL.Query().Get("continuation") == "" {
+			fmt.Fprint(w, `{"deliveries": [1], "meta": {"continuation": "c1"}}`)
+			return
+		}
+		fmt.Fprint(w, `{"deliveries": [2], "meta": {"continuation": null}}`)
+	}))
+	t.Cleanup(server.Close)
+
+	lines := pageAllLines(t, server, 0)
+	// page=1, its refetch without page, then c1.
+	if len(lines) != 2 || len(requests) != 3 {
+		t.Fatalf("expected 2 pages and 3 requests, got %d pages: %v", len(lines), requests)
+	}
+}
+
+func TestPageAll_RepeatedContinuationFails(t *testing.T) {
+	var requests int
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requests++
+		w.Header().Set("Content-Type", "application/json")
+		fmt.Fprint(w, `{"deliveries": [1], "meta": {"continuation": "stuck"}}`)
+	}))
+	t.Cleanup(server.Close)
+
+	c := New(Config{
+		BaseURL:     server.URL,
+		AccessToken: "test-jwt",
+		RetryConfig: &RetryConfig{MaxRetries: 0, SleepFn: ContextSleep},
+	})
+	var buf bytes.Buffer
+	err := c.PageAll(PageAllConfig{Ctx: context.Background(), Method: "GET", Path: "/v1/environments/1/deliveries", Writer: &buf})
+	if err == nil || !strings.Contains(err.Error(), `"stuck" twice`) {
+		t.Fatalf("expected a repeated-cursor error, got %v", err)
+	}
+	// page=1, its refetch without page, then the repeat.
+	if requests != 3 {
+		t.Errorf("expected 3 requests before stopping, got %d", requests)
+	}
+}
+
+// runPageAll walks server from startPage with params, returning the pages
+// written and the queries sent.
+func runPageAll(t *testing.T, startPage int, params map[string]string, handler func(query url.Values) string) (lines, queries []string) {
+	t.Helper()
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		queries = append(queries, r.URL.RawQuery)
+		w.Header().Set("Content-Type", "application/json")
+		fmt.Fprint(w, handler(r.URL.Query()))
+	}))
+	t.Cleanup(server.Close)
+
+	c := New(Config{
+		BaseURL:     server.URL,
+		AccessToken: "test-jwt",
+		RetryConfig: &RetryConfig{MaxRetries: 0, SleepFn: ContextSleep},
+	})
+	var buf bytes.Buffer
+	if err := c.PageAll(PageAllConfig{Ctx: context.Background(), Method: "GET", Path: "/v1/environments/1/backfills", Params: params, StartPage: startPage, Writer: &buf}); err != nil {
+		t.Fatalf("PageAll: %v", err)
+	}
+	for _, line := range strings.Split(buf.String(), "\n") {
+		if strings.TrimSpace(line) != "" {
+			lines = append(lines, line)
+		}
+	}
+	return lines, queries
+}
+
+// /backfills counts page from 0 and pages by continuation after that, so the
+// guessed page=1 skipped the newest page.
+func TestPageAll_RefetchesGuessedFirstPage(t *testing.T) {
+	lines, _ := runPageAll(t, 0, nil, func(q url.Values) string {
+		page := q.Get("page")
+		if page == "" {
+			page = q.Get("continuation")
+		}
+		switch page {
+		case "", "0":
+			return `{"backfills": ["newest"], "meta": {"continuation": "1"}}`
+		case "1":
+			return `{"backfills": ["older"], "meta": {"continuation": "2"}}`
+		default:
+			return `{"backfills": ["oldest"], "meta": {}}`
+		}
+	})
+
+	want := []string{"newest", "older", "oldest"}
+	if len(lines) != len(want) {
+		t.Fatalf("expected %d pages, got %d: %v", len(want), len(lines), lines)
+	}
+	for i, w := range want {
+		if !strings.Contains(lines[i], w) {
+			t.Errorf("page %d = %s, want %q", i, lines[i], w)
+		}
+	}
+}
+
+// A page the caller chose is where it wants to start: it is not refetched,
+// and the cursor replaces it after the first request.
+func TestPageAll_KeepsCallerPage(t *testing.T) {
+	_, queries := runPageAll(t, 3, map[string]string{"page": "3"}, func(q url.Values) string {
+		if q.Get("continuation") == "" {
+			return `{"deliveries": [1], "meta": {"continuation": "c1"}}`
+		}
+		return `{"deliveries": [2], "meta": {"continuation": ""}}`
+	})
+
+	want := []string{"page=3", "continuation=c1"}
+	if strings.Join(queries, " ") != strings.Join(want, " ") {
+		t.Errorf("queries = %q, want %q", queries, want)
+	}
+}
+
+func TestExtractPaginationMeta_Continuation(t *testing.T) {
+	tests := []struct {
+		name string
+		body string
+		want *string
+	}{
+		{"meta.continuation", `{"deliveries": [1], "meta": {"continuation": "0:2"}}`, ptr("0:2")},
+		{"last page", `{"deliveries": [1], "meta": {"continuation": ""}}`, ptr("")},
+		{"meta.pagination.continuation is not followed", `{"customers": [1], "meta": {"pagination": {"from": 5, "continuation": "x"}}}`, nil},
+		{"no cursor", `{"imports": [1], "meta": {"pagination": {"total": 1}}}`, nil},
+		{"non-string cursor is ignored", `{"items": [1], "meta": {"continuation": 3}}`, nil},
+		{"null cursor is no cursor", `{"items": [1], "meta": {"continuation": null}}`, nil},
+		{"null cursor with meta.pagination is no cursor", `{"items": [1], "meta": {"continuation": null, "pagination": {"continuation": "x"}}}`, nil},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := extractPaginationMeta(json.RawMessage(tt.body), 0).Continuation
+			switch {
+			case tt.want == nil && got != nil:
+				t.Errorf("expected no cursor, got %q", *got)
+			case tt.want != nil && (got == nil || *got != *tt.want):
+				t.Errorf("cursor = %v, want %q", got, *tt.want)
+			}
+		})
+	}
+}
+
+func ptr(s string) *string { return &s }