From e0be82f4c67442a18cc5e0be5aca0404ccb0fb67 Mon Sep 17 00:00:00 2001 From: "Customer.io Open Source Bot" Date: Wed, 23 Sep 2026 13:59:39 -0700 Subject: [PATCH] Add Customer.io CLI source CioCliPublicExport-RevId: ca52bf15de1720f1e941f462290b37688132ebf1 --- cmd/helpers.go | 2 +- internal/client/pagination.go | 76 ++++++++++- internal/client/pagination_test.go | 211 +++++++++++++++++++++++++++++ 3 files changed, 286 insertions(+), 3 deletions(-) diff --git a/cmd/helpers.go b/cmd/helpers.go index 0e9cf24..1356f71 100644 --- a/cmd/helpers.go +++ b/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 { diff --git a/internal/client/pagination.go b/internal/client/pagination.go index 30d7476..1b429a3 100644 --- a/internal/client/pagination.go +++ b/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")) } diff --git a/internal/client/pagination_test.go b/internal/client/pagination_test.go index 8f7d382..a62d11e 100644 --- a/internal/client/pagination_test.go +++ b/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" (the first page for no +// cursor). Each page's next cursor is "c", 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 }