feat(paging): add support for after id parameter in find options (#19219)

* feat(paging): add support for after id parameter in find options

* chore(http): update swagger to reflect after query parameter in list buckets

* chore(changelog): update changelog to reflect after query parameter in list buckets

* chore(tenant): update tenant storage tests for paginating with after

CHANGELOG.md

@@ -4,6 +4,7 @@
 
 1. [19246](https://github.com/influxdata/influxdb/pull/19246): Redesign load data page to increase discovery and ease of use
 1. [19334](https://github.com/influxdata/influxdb/pull/19334): Add --active-config flag to influx to set config for single command
+1. [19219](https://github.com/influxdata/influxdb/pull/19219): List buckets via the API now supports after (ID) parameter as an alternative to offset.
 
 ### Bug Fixes
 

http/swagger.yml

@@ -3466,6 +3466,7 @@ paths:
         - $ref: "#/components/parameters/TraceSpan"
         - $ref: "#/components/parameters/Offset"
         - $ref: "#/components/parameters/Limit"
+        - $ref: "#/components/parameters/After"
         - in: query
           name: org
           description: The organization name.
@@ -6515,6 +6516,15 @@ components:
       required: false
       schema:
         type: string
+    After:
+      in: query
+      name: after
+      required: false
+      schema:
+        type: string
+      description: >
+        The last resource ID from which to seek from (but not including).
+        This is to be used instead of `offset`.
   schemas:
     LanguageRequest:
       description: Flux query to be analyzed.

kv/bucket.go

@@ -403,18 +403,34 @@ func (s *Service) findBuckets(ctx context.Context, tx Tx, filter influxdb.Bucket
 		filter.OrganizationID = &o.ID
 	}
 
-	var offset, limit, count int
-	var descending bool
+	var (
+		offset, limit, count int
+		descending           bool
+	)
+
+	after := func(*influxdb.Bucket) bool {
+		return true
+	}
+
 	if len(opts) > 0 {
 		offset = opts[0].Offset
 		limit = opts[0].Limit
 		descending = opts[0].Descending
+		if opts[0].After != nil {
+			after = func(b *influxdb.Bucket) bool {
+				if descending {
+					return b.ID < *opts[0].After
+				}
+
+				return b.ID > *opts[0].After
+			}
+		}
 	}
 
 	filterFn := filterBucketsFn(filter)
 	err := s.forEachBucket(ctx, tx, descending, func(b *influxdb.Bucket) bool {
 		if filterFn(b) {
-			if count >= offset {
+			if count >= offset && after(b) {
 				bs = append(bs, b)
 			}
 			count++

paging.go

@@ -29,6 +29,7 @@ type PagingLinks struct {
 type FindOptions struct {
 	Limit      int
 	Offset     int
+	After      *ID
 	SortBy     string
 	Descending bool
 }
@@ -50,6 +51,18 @@ func DecodeFindOptions(r *http.Request) (*FindOptions, error) {
 		opts.Offset = o
 	}
 
+	if after := qp.Get("after"); after != "" {
+		id, err := IDFromString(after)
+		if err != nil {
+			return nil, &Error{
+				Code: EInvalid,
+				Err:  fmt.Errorf("decoding after: %w", err),
+			}
+		}
+
+		opts.After = id
+	}
+
 	if limit := qp.Get("limit"); limit != "" {
 		l, err := strconv.Atoi(limit)
 		if err != nil {
@@ -109,6 +122,10 @@ func (f FindOptions) QueryParams() map[string][]string {
 		"offset":     {strconv.Itoa(f.Offset)},
 	}
 
+	if f.After != nil {
+		qp["after"] = []string{f.After.String()}
+	}
+
 	if f.Limit > 0 {
 		qp["limit"] = []string{strconv.Itoa(f.Limit)}
 	}

tenant/storage_bucket.go

@@ -190,7 +190,17 @@ func (s *Store) ListBuckets(ctx context.Context, tx kv.Tx, filter BucketFilter,
 	if o.Descending {
 		opts = append(opts, kv.WithCursorDirection(kv.CursorDescending))
 	}
-	cursor, err := b.ForwardCursor(nil, opts...)
+
+	var seek []byte
+	if o.After != nil {
+		after := (*o.After) + 1
+		seek, err = after.Encode()
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	cursor, err := b.ForwardCursor(seek, opts...)
 	if err != nil {
 		return nil, err
 	}

tenant/storage_bucket_test.go

@@ -181,6 +181,43 @@ func TestBucket(t *testing.T) {
 				}
 			},
 		},
+		{
+			name:  "list all with limit 3 using after to paginate",
+			setup: simpleSetup,
+			results: func(t *testing.T, store *tenant.Store, tx kv.Tx) {
+				var (
+					expected  = testBuckets(10, withCrudLog)
+					found     []*influxdb.Bucket
+					lastID    *influxdb.ID
+					limit     = 3
+					listAfter = func(after *influxdb.ID) ([]*influxdb.Bucket, error) {
+						return store.ListBuckets(context.Background(), tx, tenant.BucketFilter{}, influxdb.FindOptions{
+							After: after,
+							Limit: limit,
+						})
+					}
+				)
+
+				var (
+					b   []*influxdb.Bucket
+					err error
+				)
+
+				for b, err = listAfter(lastID); err == nil; b, err = listAfter(lastID) {
+					lastID = &b[len(b)-1].ID
+					found = append(found, b...)
+
+					// given we've seen the last page
+					if len(b) < limit {
+						break
+					}
+				}
+
+				require.NoError(t, err)
+
+				assert.Equal(t, expected, found)
+			},
+		},
 		{
 			name:  "update",
 			setup: simpleSetup,

testing/bucket_service.go

@@ -617,6 +617,55 @@ func FindBuckets(
 				},
 			},
 		},
+		{
+			name: "find all buckets by after and limit",
+			fields: BucketFields{
+				OrgIDs:    mock.NewIncrementingIDGenerator(idOne),
+				BucketIDs: mock.NewIncrementingIDGenerator(idOne),
+				Organizations: []*influxdb.Organization{
+					{
+						Name: "theorg",
+					},
+				},
+				Buckets: []*influxdb.Bucket{
+					{
+						// ID(1)
+						OrgID: idOne,
+						Name:  "abc",
+					},
+					{
+						// ID(2)
+						OrgID: idOne,
+						Name:  "def",
+					},
+					{
+						// ID(3)
+						OrgID: idOne,
+						Name:  "xyz",
+					},
+				},
+			},
+			args: args{
+				findOptions: influxdb.FindOptions{
+					After: idPtr(idOne),
+					Limit: 2,
+				},
+			},
+			wants: wants{
+				buckets: []*influxdb.Bucket{
+					{
+						ID:    idTwo,
+						OrgID: idOne,
+						Name:  "def",
+					},
+					{
+						ID:    idThree,
+						OrgID: idOne,
+						Name:  "xyz",
+					},
+				},
+			},
+		},
 		{
 			name: "find all buckets by descending",
 			fields: BucketFields{