request_queue.go

package apify

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"sync"
	"time"
)

// ListRequestsOptions configures [RequestQueueClient.ListRequests].
type ListRequestsOptions struct {
	// Limit is the maximum number of requests to return.
	Limit *int64
	// ExclusiveStartID lists requests after this ID.
	//
	// Deprecated: the API deprecated the exclusiveStartId parameter. Use Cursor.
	ExclusiveStartID *string
	// Cursor is an opaque pagination cursor (alternative to ExclusiveStartID).
	Cursor *string
	// Filter restricts the listing to requests in the given states. Each value must be
	// "locked" or "pending" (see RequestFilterLocked / RequestFilterPending). Multiple
	// values are sent as a comma-separated list and mean the union of those states
	// (requests matching any of them are returned), matching the API.
	Filter []string
}

func (o ListRequestsOptions) apply(q *QueryParams) {
	q.AddInt("limit", o.Limit).
		AddString("exclusiveStartId", o.ExclusiveStartID).
		AddString("cursor", o.Cursor).
		AddCSV("filter", o.Filter)
}

// RequestQueueClient is a client for a specific request queue (and run-nested variants).
type RequestQueueClient struct {
	ctx       *resourceContext
	clientKey string
}

func newRequestQueueClient(hc *httpClient, baseURL, resourcePath, id string) *RequestQueueClient {
	return &RequestQueueClient{ctx: newSingleContext(hc, baseURL, resourcePath, id)}
}

// newRequestQueueNestedClient creates a client for a run's default request queue.
func newRequestQueueNestedClient(hc *httpClient, base, subPath string) *RequestQueueClient {
	return &RequestQueueClient{ctx: newCollectionContext(hc, base, subPath)}
}

// WithClientKey returns a copy of the client that identifies its requests with clientKey.
//
// A stable client key is required to operate on locks the client itself created (e.g. to
// unlock its own requests), and lets the API detect whether multiple clients access a queue.
func (c *RequestQueueClient) WithClientKey(clientKey string) *RequestQueueClient {
	clone := *c
	clone.clientKey = clientKey
	return &clone
}

// withClientKey adds the client key (if set) to the given params.
func (c *RequestQueueClient) withClientKey(params *QueryParams) *QueryParams {
	if c.clientKey != "" {
		params.AddString("clientKey", &c.clientKey)
	}
	return params
}

// Get fetches the queue metadata. The bool reports whether it exists.
func (c *RequestQueueClient) Get(ctx context.Context) (RequestQueue, bool, error) {
	return getResource[RequestQueue](ctx, c.ctx, "", NewQueryParams())
}

// Update updates the queue metadata (e.g. name) and returns the updated object.
func (c *RequestQueueClient) Update(ctx context.Context, newFields any) (RequestQueue, error) {
	return updateResource[RequestQueue](ctx, c.ctx, "", newFields)
}

// Delete deletes the queue.
func (c *RequestQueueClient) Delete(ctx context.Context) error {
	return deleteResource(ctx, c.ctx, "")
}

// ListHead returns the requests at the head (front) of the queue, up to limit (nil for the
// server default).
func (c *RequestQueueClient) ListHead(ctx context.Context, limit *int64) (RequestQueueHead, error) {
	params := NewQueryParams()
	params.AddInt("limit", limit)
	c.withClientKey(params)
	return getResourceRequired[RequestQueueHead](ctx, c.ctx, "head", params, c.ctx.shortTimeout())
}

// AddRequest adds a request to the queue. If forefront is true, the request is added to the
// front of the queue.
func (c *RequestQueueClient) AddRequest(ctx context.Context, request RequestQueueRequest, forefront bool) (RequestQueueOperationInfo, error) {
	params := NewQueryParams()
	params.AddBool("forefront", &forefront)
	c.withClientKey(params)
	body, err := json.Marshal(request)
	if err != nil {
		return RequestQueueOperationInfo{}, err
	}
	return postWithBody[RequestQueueOperationInfo](ctx, c.ctx, "requests", params, body, contentTypeJSON, c.ctx.shortTimeout())
}

// GetRequest fetches a request by ID, or (nil, false, nil) if it does not exist.
func (c *RequestQueueClient) GetRequest(ctx context.Context, id string) (*RequestQueueRequest, bool, error) {
	safeID, err := safePathSegment(id)
	if err != nil {
		return nil, false, err
	}
	req, present, err := getResourceAlways[RequestQueueRequest](ctx, c.ctx, "requests/"+safeID, NewQueryParams())
	if err != nil || !present {
		return nil, present, err
	}
	return &req, true, nil
}

// UpdateRequest updates an existing request (identified by its ID field) and returns the
// operation info. If forefront is true, the request is moved to the front of the queue.
func (c *RequestQueueClient) UpdateRequest(ctx context.Context, request RequestQueueRequest, forefront bool) (RequestQueueOperationInfo, error) {
	if c.ctx.idErr != nil {
		return RequestQueueOperationInfo{}, c.ctx.idErr
	}
	safeID, err := safePathSegment(request.ID)
	if err != nil {
		return RequestQueueOperationInfo{}, err
	}
	params := NewQueryParams()
	params.AddBool("forefront", &forefront)
	c.withClientKey(params)
	url := c.ctx.mergedParams(params).applyToURL(c.ctx.subURL("requests/" + safeID))
	body, err := json.Marshal(request)
	if err != nil {
		return RequestQueueOperationInfo{}, err
	}
	resp, err := c.ctx.http.call(ctx, http.MethodPut, url, body, contentTypeJSON, c.ctx.mediumTimeout())
	if err != nil {
		return RequestQueueOperationInfo{}, err
	}
	return parseDataEnvelope[RequestQueueOperationInfo](resp.body)
}

// DeleteRequest deletes a request by ID.
func (c *RequestQueueClient) DeleteRequest(ctx context.Context, id string) error {
	if c.ctx.idErr != nil {
		return c.ctx.idErr
	}
	safeID, err := safePathSegment(id)
	if err != nil {
		return err
	}
	url := c.ctx.mergedParams(c.withClientKey(NewQueryParams())).applyToURL(c.ctx.subURL("requests/" + safeID))
	_, err = c.ctx.http.call(ctx, http.MethodDelete, url, nil, "", c.ctx.shortTimeout())
	if err != nil && !isNotFound(err) {
		return err
	}
	return nil
}

// ListAndLockHead atomically returns and locks up to limit requests from the head of the
// queue for lockSecs seconds. Returns the raw API response (a locked-head object).
func (c *RequestQueueClient) ListAndLockHead(ctx context.Context, lockSecs int64, limit *int64) (json.RawMessage, error) {
	params := NewQueryParams()
	params.AddInt("lockSecs", &lockSecs).AddInt("limit", limit)
	c.withClientKey(params)
	return postWithBody[json.RawMessage](ctx, c.ctx, "head/lock", params, nil, "", c.ctx.mediumTimeout())
}

// maxRequestsPerBatchOperation is the API limit on requests per batch call. Larger inputs are split
// into chunks of this size.
const maxRequestsPerBatchOperation = 25

// maxPayloadSizeBytes is the maximum request body size of a batch-requests call, 9 MiB.
const maxPayloadSizeBytes = 9437184

// payloadSizeLimitBytes is maxPayloadSizeBytes less a 0.01% safety margin, rounded up to 944 bytes.
const payloadSizeLimitBytes = maxPayloadSizeBytes - 944

// BatchAddResult is the typed result of [RequestQueueClient.BatchAddRequests]: the requests
// the API accepted and the ones it could not process.
type BatchAddResult struct {
	// ProcessedRequests are the requests the API successfully added.
	ProcessedRequests []RequestQueueOperationInfo `json:"processedRequests"`
	// UnprocessedRequests are the requests the API did not process.
	UnprocessedRequests []RequestQueueRequest `json:"unprocessedRequests"`
}

// serializedRequest is a request to add, serialized once up front: json is what the body of
// its batch is assembled from (so sending it never re-marshals the request), and original is
// what the result bookkeeping needs.
type serializedRequest struct {
	original RequestQueueRequest
	json     []byte
}

// serializeRequestsForBatch serializes every request once and checks each one against
// payloadSizeLimitBytes before any request is sent, so an oversized request fails the whole
// call up front instead of after the batches before it have already gone out.
func serializeRequestsForBatch(requests []RequestQueueRequest) ([]serializedRequest, error) {
	serialized := make([]serializedRequest, len(requests))
	for i, r := range requests {
		data, err := json.Marshal(r)
		if err != nil {
			return nil, err
		}
		// +2 bytes for the enclosing brackets, which even a batch of one request carries.
		if len(data)+2 > payloadSizeLimitBytes {
			return nil, fmt.Errorf("RequestQueueClient.BatchAddRequests: the request at index %d exceeds the maximum allowed size (%d bytes)", i, payloadSizeLimitBytes)
		}
		serialized[i] = serializedRequest{original: r, json: data}
	}
	return serialized, nil
}

// splitIntoBatches groups serialized requests into consecutive batches of at most maxCount
// requests, each fitting a JSON array body (items joined by commas between brackets) of at
// most maxBytes bytes. A request too large for a batch of its own is rejected beforehand by
// serializeRequestsForBatch, so every request here fits.
func splitIntoBatches(serialized []serializedRequest, maxCount int, maxBytes int) [][]serializedRequest {
	var batches [][]serializedRequest
	var batch []serializedRequest
	byteLen := 1 // the opening bracket
	for _, s := range serialized {
		// Each item adds its own bytes plus one for the following comma or closing bracket.
		if len(batch) > 0 && (len(batch) >= maxCount || byteLen+len(s.json)+1 > maxBytes) {
			batches = append(batches, batch)
			batch = nil
			byteLen = 1
		}
		batch = append(batch, s)
		byteLen += len(s.json) + 1
	}
	if len(batch) > 0 {
		batches = append(batches, batch)
	}
	return batches
}

// joinAsJSONArray joins the requests' pre-serialized JSON into a single JSON array body, so
// sending a batch never re-marshals the requests it contains.
func joinAsJSONArray(batch []serializedRequest) []byte {
	var buf bytes.Buffer
	buf.WriteByte('[')
	for i, s := range batch {
		if i > 0 {
			buf.WriteByte(',')
		}
		buf.Write(s.json)
	}
	buf.WriteByte(']')
	return buf.Bytes()
}

// Defaults of [BatchAddRequestsOptions].
const (
	// defaultMaxUnprocessedRequestsRetries is the default number of retries of the unprocessed
	// requests of a batch.
	defaultMaxUnprocessedRequestsRetries = 3
	// defaultMaxParallelBatches is the default number of batches that are sent at once.
	defaultMaxParallelBatches = 5
	// defaultMinDelayBetweenUnprocessedRequestsRetries is the default base delay between retries.
	defaultMinDelayBetweenUnprocessedRequestsRetries = 500 * time.Millisecond
)

// BatchAddRequestsOptions configures [RequestQueueClient.BatchAddRequestsWithOptions]. A nil
// field has its default value.
type BatchAddRequestsOptions struct {
	// Forefront adds the requests to the front of the queue.
	Forefront bool
	// MaxUnprocessedRequestsRetries is how many times the requests that the API did not process
	// are sent again. Defaults to 3.
	MaxUnprocessedRequestsRetries *int64
	// MaxParallel is how many batches are sent at once. Defaults to 5.
	MaxParallel *int64
	// MinDelayBetweenUnprocessedRequestsRetries is the base delay before a retry. It doubles with
	// each retry, is capped at [WithTimeoutMax], and is randomized by up to 100%. Defaults to 500ms.
	MinDelayBetweenUnprocessedRequestsRetries *time.Duration
}

// batchAddSettings are the resolved [BatchAddRequestsOptions].
type batchAddSettings struct {
	forefront   bool
	maxRetries  int
	maxParallel int
	minDelay    time.Duration
}

// resolve fills in the defaults and rejects invalid values.
func (o BatchAddRequestsOptions) resolve() (batchAddSettings, error) {
	settings := batchAddSettings{
		forefront:   o.Forefront,
		maxRetries:  defaultMaxUnprocessedRequestsRetries,
		maxParallel: defaultMaxParallelBatches,
		minDelay:    defaultMinDelayBetweenUnprocessedRequestsRetries,
	}
	if o.MaxUnprocessedRequestsRetries != nil {
		if *o.MaxUnprocessedRequestsRetries < 0 {
			return settings, errors.New("BatchAddRequestsOptions.MaxUnprocessedRequestsRetries must not be negative")
		}
		settings.maxRetries = int(*o.MaxUnprocessedRequestsRetries)
	}
	if o.MaxParallel != nil {
		if *o.MaxParallel < 1 {
			return settings, errors.New("BatchAddRequestsOptions.MaxParallel must be at least 1")
		}
		settings.maxParallel = int(*o.MaxParallel)
	}
	if o.MinDelayBetweenUnprocessedRequestsRetries != nil {
		if *o.MinDelayBetweenUnprocessedRequestsRetries < 0 {
			return settings, errors.New("BatchAddRequestsOptions.MinDelayBetweenUnprocessedRequestsRetries must not be negative")
		}
		settings.minDelay = *o.MinDelayBetweenUnprocessedRequestsRetries
	}
	return settings, nil
}

// BatchAddRequests adds requests to the queue, at the front if forefront is true, with the default
// [BatchAddRequestsOptions]. See [RequestQueueClient.BatchAddRequestsWithOptions].
func (c *RequestQueueClient) BatchAddRequests(ctx context.Context, requests []RequestQueueRequest, forefront bool) (BatchAddResult, error) {
	return c.BatchAddRequestsWithOptions(ctx, requests, BatchAddRequestsOptions{Forefront: forefront})
}

// BatchAddRequestsWithOptions adds requests to the queue. It splits them into batches within the
// API limits of 25 requests and 9 MiB, sends up to MaxParallel batches at once, and sends the
// requests that the API did not process again, up to MaxUnprocessedRequestsRetries times. It fails
// before sending anything if one request is too large for a batch.
//
// The result lists the requests that were added and those that are still unprocessed. A batch
// that fails with an API or transport error is not an error of the call: its requests that were
// not added are unprocessed, and the error is not reported. To find out why requests stay
// unprocessed, add one with [RequestQueueClient.AddRequest]. The call returns an error, with the
// result so far, when a response cannot be decoded or ctx ends.
func (c *RequestQueueClient) BatchAddRequestsWithOptions(ctx context.Context, requests []RequestQueueRequest, options BatchAddRequestsOptions) (BatchAddResult, error) {
	var merged BatchAddResult
	if c.ctx.idErr != nil {
		return merged, c.ctx.idErr
	}
	settings, err := options.resolve()
	if err != nil {
		return merged, err
	}
	serialized, err := serializeRequestsForBatch(requests)
	if err != nil {
		return merged, err
	}
	batches := splitIntoBatches(serialized, maxRequestsPerBatchOperation, payloadSizeLimitBytes)

	results := make([]batchOutcome, len(batches))
	slots := make(chan struct{}, settings.maxParallel)
	var wg sync.WaitGroup
	for i, batch := range batches {
		slots <- struct{}{}
		wg.Add(1)
		go func() {
			defer wg.Done()
			defer func() { <-slots }()
			results[i] = c.addBatchWithRetries(ctx, batch, settings)
		}()
	}
	wg.Wait()

	var firstErr error
	for _, outcome := range results {
		merged.ProcessedRequests = append(merged.ProcessedRequests, outcome.result.ProcessedRequests...)
		merged.UnprocessedRequests = append(merged.UnprocessedRequests, outcome.result.UnprocessedRequests...)
		if firstErr == nil {
			firstErr = outcome.err
		}
	}
	return merged, firstErr
}

// batchOutcome is the result of one batch and the error that fails the whole call, if any.
type batchOutcome struct {
	result BatchAddResult
	err    error
}

// addBatchWithRetries sends batch and then the requests that the API did not process, until all
// are added or the retries are used up.
func (c *RequestQueueClient) addBatchWithRetries(ctx context.Context, batch []serializedRequest, settings batchAddSettings) batchOutcome {
	var outcome batchOutcome
	added := map[string]bool{}
	remaining := batch
	for attempt := 0; attempt <= settings.maxRetries; attempt++ {
		response, err := c.batchAddChunk(ctx, remaining, settings.forefront)
		if err != nil {
			// An API or transport error leaves the rest of the batch unprocessed.
			outcome.result.UnprocessedRequests = unprocessedOf(remaining)
			if ctx.Err() != nil || isResponseDecodingError(err) {
				outcome.err = err
			}
			return outcome
		}
		outcome.result.ProcessedRequests = append(outcome.result.ProcessedRequests, response.ProcessedRequests...)
		outcome.result.UnprocessedRequests = response.UnprocessedRequests
		for _, processed := range response.ProcessedRequests {
			added[processed.UniqueKey] = true
		}
		remaining = withoutAdded(batch, added)
		if len(remaining) == 0 || attempt == settings.maxRetries {
			return outcome
		}
		if !sleepWithContext(ctx, c.unprocessedRetryDelay(settings.minDelay, attempt)) {
			outcome.err = ctx.Err()
			outcome.result.UnprocessedRequests = unprocessedOf(remaining)
			return outcome
		}
	}
	return outcome
}

// unprocessedRetryDelay returns the delay before the retry that follows attempt: minDelay doubled
// for each earlier retry, capped at the maximum request timeout and randomized.
func (c *RequestQueueClient) unprocessedRetryDelay(minDelay time.Duration, attempt int) time.Duration {
	delay := minDuration(minDelay, c.ctx.http.retry.timeoutMax)
	for i := 0; i < attempt; i++ {
		delay = minDuration(delay*backoffFactor, c.ctx.http.retry.timeoutMax)
	}
	return randomizedDelay(delay)
}

// isResponseDecodingError reports whether err comes from decoding a response body.
func isResponseDecodingError(err error) bool {
	var syntaxErr *json.SyntaxError
	var typeErr *json.UnmarshalTypeError
	return errors.As(err, &syntaxErr) || errors.As(err, &typeErr)
}

// withoutAdded returns the requests of batch whose unique key is not in added.
func withoutAdded(batch []serializedRequest, added map[string]bool) []serializedRequest {
	var remaining []serializedRequest
	for _, s := range batch {
		if !added[s.original.UniqueKey] {
			remaining = append(remaining, s)
		}
	}
	return remaining
}

// unprocessedOf returns the unprocessed form of requests, which has only a method, a unique key
// and a URL.
func unprocessedOf(requests []serializedRequest) []RequestQueueRequest {
	unprocessed := make([]RequestQueueRequest, len(requests))
	for i, s := range requests {
		unprocessed[i] = RequestQueueRequest{Method: s.original.Method, UniqueKey: s.original.UniqueKey, URL: s.original.URL}
	}
	return unprocessed
}

// batchAddChunk sends one batch and parses the result.
func (c *RequestQueueClient) batchAddChunk(ctx context.Context, batch []serializedRequest, forefront bool) (BatchAddResult, error) {
	params := NewQueryParams()
	params.AddBool("forefront", &forefront)
	c.withClientKey(params)
	body := joinAsJSONArray(batch)
	return postWithBody[BatchAddResult](ctx, c.ctx, "requests/batch", params, body, contentTypeJSON, c.ctx.mediumTimeout())
}

// BatchDeleteRequests deletes multiple requests in a single call. Each entry identifies a
// request (e.g. by id or uniqueKey). Returns the raw batch result.
func (c *RequestQueueClient) BatchDeleteRequests(ctx context.Context, requests any) (json.RawMessage, error) {
	params := c.withClientKey(NewQueryParams())
	return deleteWithBody[json.RawMessage](ctx, c.ctx, "requests/batch", params, requests)
}

// Allowed values for entries in ListRequestsOptions.Filter, as constrained by the API.
const (
	// RequestFilterLocked filters the listing to currently locked requests.
	RequestFilterLocked = "locked"
	// RequestFilterPending filters the listing to pending (not-yet-handled) requests.
	RequestFilterPending = "pending"
)

// validate checks the listing options for API-level constraints: ExclusiveStartID and Cursor
// are mutually exclusive, and every Filter entry (if any) must be "locked" or "pending".
func (o ListRequestsOptions) validate() error {
	if o.ExclusiveStartID != nil && o.Cursor != nil {
		return errors.New("ListRequestsOptions: ExclusiveStartID and Cursor are mutually exclusive")
	}
	for _, f := range o.Filter {
		if f != RequestFilterLocked && f != RequestFilterPending {
			return fmt.Errorf("ListRequestsOptions: Filter entries must be %q or %q, got %q", RequestFilterLocked, RequestFilterPending, f)
		}
	}
	return nil
}

// ListRequests lists the queue's requests with pagination.
func (c *RequestQueueClient) ListRequests(ctx context.Context, options ListRequestsOptions) (json.RawMessage, error) {
	if err := options.validate(); err != nil {
		return nil, err
	}
	params := NewQueryParams()
	options.apply(params)
	c.withClientKey(params)
	return getResourceRequired[json.RawMessage](ctx, c.ctx, "requests", params, c.ctx.mediumTimeout())
}

// ProlongRequestLock extends the lock on a request by lockSecs seconds. If forefront is
// true, the request is moved to the front when its lock expires. Returns the raw response.
func (c *RequestQueueClient) ProlongRequestLock(ctx context.Context, id string, lockSecs int64, forefront bool) (json.RawMessage, error) {
	if c.ctx.idErr != nil {
		return nil, c.ctx.idErr
	}
	safeID, err := safePathSegment(id)
	if err != nil {
		return nil, err
	}
	params := NewQueryParams()
	params.AddInt("lockSecs", &lockSecs).AddBool("forefront", &forefront)
	c.withClientKey(params)
	url := c.ctx.mergedParams(params).applyToURL(c.ctx.subURL("requests/" + safeID + "/lock"))
	resp, err := c.ctx.http.call(ctx, http.MethodPut, url, nil, "", c.ctx.mediumTimeout())
	if err != nil {
		return nil, err
	}
	return parseDataEnvelope[json.RawMessage](resp.body)
}

// DeleteRequestLock releases the lock on a request. If forefront is true, the request is
// moved to the front of the queue.
func (c *RequestQueueClient) DeleteRequestLock(ctx context.Context, id string, forefront bool) error {
	if c.ctx.idErr != nil {
		return c.ctx.idErr
	}
	safeID, err := safePathSegment(id)
	if err != nil {
		return err
	}
	params := NewQueryParams()
	params.AddBool("forefront", &forefront)
	c.withClientKey(params)
	url := c.ctx.mergedParams(params).applyToURL(c.ctx.subURL("requests/" + safeID + "/lock"))
	_, err = c.ctx.http.call(ctx, http.MethodDelete, url, nil, "", c.ctx.shortTimeout())
	if err != nil && !isNotFound(err) {
		return err
	}
	return nil
}

// UnlockRequests releases all locks the client holds on this queue's requests. Returns the
// raw response.
func (c *RequestQueueClient) UnlockRequests(ctx context.Context) (json.RawMessage, error) {
	params := c.withClientKey(NewQueryParams())
	return postWithBody[json.RawMessage](ctx, c.ctx, "requests/unlock", params, nil, "", c.ctx.longTimeout())
}

// PaginateRequests returns a lazy iterator over all requests in the queue, fetching pages
// of up to pageLimit requests at a time (nil for the server default). To cap the total, start
// from a cursor or filter the states, use [RequestQueueClient.PaginateRequestsWithOptions].
func (c *RequestQueueClient) PaginateRequests(pageLimit *int64) *RequestQueueRequestsIterator {
	return c.PaginateRequestsWithOptions(ListRequestsOptions{}, pageLimit)
}

// PaginateRequestsWithOptions is [RequestQueueClient.PaginateRequests] with listing options. Limit
// caps the number of requests yielded, and nil or 0 means all. Cursor, or the deprecated
// ExclusiveStartID, is where to start, and Filter applies to every page. chunkSize is the page
// size, nil for the default. The first call of Next reports invalid options.
func (c *RequestQueueClient) PaginateRequestsWithOptions(options ListRequestsOptions, chunkSize *int64) *RequestQueueRequestsIterator {
	return &RequestQueueRequestsIterator{client: c, options: options, chunkSize: chunkSize}
}

// RequestQueueRequestsIterator lazily iterates over a request queue's requests, fetching one
// page at a time via the cursor-based listing endpoint.
type RequestQueueRequestsIterator struct {
	client    *RequestQueueClient
	options   ListRequestsOptions
	chunkSize *int64

	buffer     []RequestQueueRequest
	pos        int
	nextCursor string
	capped     bool  // whether a total cap (options.Limit) is in effect
	remaining  int64 // requests the cap still allows; only meaningful when capped
	started    bool
	exhausted  bool
}

// requestsPage is the shape of a paginated requests listing.
type requestsPage struct {
	Items      []RequestQueueRequest `json:"items"`
	NextCursor *string               `json:"nextCursor"`
}

// Next returns the next request, or (nil, nil) when the iterator is exhausted.
func (it *RequestQueueRequestsIterator) Next(ctx context.Context) (*RequestQueueRequest, error) {
	for it.pos >= len(it.buffer) {
		if it.exhausted {
			return nil, nil
		}
		if err := it.fetchPage(ctx); err != nil {
			return nil, err
		}
	}
	req := it.buffer[it.pos]
	it.pos++
	return &req, nil
}

// fetchPage loads the next page of requests into the buffer. The page size is the smaller of the
// remaining total cap and the chunk size, so the last page is never over-fetched.
func (it *RequestQueueRequestsIterator) fetchPage(ctx context.Context) error {
	opts := it.options
	if !it.started {
		if err := opts.validate(); err != nil {
			return err
		}
		if l := opts.Limit; l != nil && *l > 0 {
			it.capped = true
			it.remaining = *l
		}
	} else {
		opts.ExclusiveStartID = nil
		opts.Cursor = &it.nextCursor
	}
	var capLeft, chunk int64
	if it.capped {
		capLeft = it.remaining
	}
	if it.chunkSize != nil {
		chunk = *it.chunkSize
	}
	opts.Limit = pageLimitPtr(minForLimitParam(capLeft, chunk))

	params := NewQueryParams()
	opts.apply(params)
	it.client.withClientKey(params)

	raw, err := getResourceRequired[requestsPage](ctx, it.client.ctx, "requests", params, it.client.ctx.mediumTimeout())
	if err != nil {
		return err
	}
	it.started = true
	it.buffer = raw.Items
	it.pos = 0
	it.nextCursor = ""
	if raw.NextCursor != nil {
		it.nextCursor = *raw.NextCursor
	}
	if it.capped {
		it.remaining -= int64(len(it.buffer))
		if it.remaining < 0 { // a server that returned more than asked for
			it.buffer = it.buffer[:int64(len(it.buffer))+it.remaining]
			it.remaining = 0
		}
	}
	if it.nextCursor == "" || (it.capped && it.remaining == 0) {
		it.exhausted = true
	}
	return nil
}