Client cache API paging (#5107)

* internal/clientcache: add -force-reset-schema flag

* clientcache: stream list pages directly to DB

---------

Co-authored-by: Johan Brandhorst-Satzkorn <johan.brandhorst@gmail.com>
(cherry picked from commit ccea8a4ce53f7b0d1105c1c1e4875e443635d96a)

api/users/custom.go

@@ -5,10 +5,8 @@ package users
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
 	"net/url"
-	"slices"
 
 	"github.com/hashicorp/boundary/api/aliases"
 )
@@ -25,135 +23,44 @@ func (c *Client) ListResolvableAliases(ctx context.Context, userId string, opt .
 		return nil, fmt.Errorf("nil client")
 	}
 
-	opts, apiOpts := getOpts(opt...)
-	req, err := c.client.NewRequest(ctx, "GET", fmt.Sprintf("users/%s:list-resolvable-aliases", url.PathEscape(userId)), nil, apiOpts...)
-	if err != nil {
-		return nil, fmt.Errorf("error creating List request: %w", err)
-	}
-
-	if len(opts.queryMap) > 0 {
-		q := url.Values{}
-		for k, v := range opts.queryMap {
-			q.Add(k, v)
-		}
-		req.URL.RawQuery = q.Encode()
-	}
-
-	resp, err := c.client.Do(req)
-	if err != nil {
-		return nil, fmt.Errorf("error performing client request during List call: %w", err)
-	}
-
-	target := new(aliases.AliasListResult)
-	apiErr, err := resp.Decode(target)
-	if err != nil {
-		return nil, fmt.Errorf("error decoding List response: %w", err)
-	}
-	if apiErr != nil {
-		return nil, apiErr
-	}
-	target.Response = resp
-	if target.ResponseType == "complete" || target.ResponseType == "" {
-		return target, nil
-	}
-	// If there are more results, automatically fetch the rest of the results.
-	// idToIndex keeps a map from the ID of an item to its index in target.Items.
-	// This is used to update updated items in-place and remove deleted items
-	// from the result after pagination is done.
-	idToIndex := map[string]int{}
-	for i, item := range target.Items {
-		idToIndex[item.Id] = i
-	}
-	for {
-		req, err := c.client.NewRequest(ctx, "GET", fmt.Sprintf("users/%s:list-resolvable-aliases", url.PathEscape(userId)), nil, apiOpts...)
-		if err != nil {
-			return nil, fmt.Errorf("error creating List request: %w", err)
-		}
-
-		opts.queryMap["list_token"] = target.ListToken
-		if len(opts.queryMap) > 0 {
-			q := url.Values{}
-			for k, v := range opts.queryMap {
-				q.Add(k, v)
-			}
-			req.URL.RawQuery = q.Encode()
-		}
-
-		resp, err := c.client.Do(req)
-		if err != nil {
-			return nil, fmt.Errorf("error performing client request during List call: %w", err)
-		}
-
-		page := new(aliases.AliasListResult)
-		apiErr, err := resp.Decode(page)
-		if err != nil {
-			return nil, fmt.Errorf("error decoding List response: %w", err)
-		}
-		if apiErr != nil {
-			return nil, apiErr
-		}
-		for _, item := range page.Items {
-			if i, ok := idToIndex[item.Id]; ok {
-				// Item has already been seen at index i, update in-place
-				target.Items[i] = item
-			} else {
-				target.Items = append(target.Items, item)
-				idToIndex[item.Id] = len(target.Items) - 1
-			}
-		}
-		// RemovedIds contain any Alias that were deleted since the last response.
-		target.RemovedIds = append(target.RemovedIds, page.RemovedIds...)
-		target.EstItemCount = page.EstItemCount
-		target.ListToken = page.ListToken
-		target.ResponseType = page.ResponseType
-		target.Response = resp
-		if target.ResponseType == "complete" {
-			break
-		}
-	}
-	// For now, removedIds will only be populated if this pagination cycle was the result of a
-	// "refresh" operation (i.e., the caller provided a list token option to this call).
-	//
-	// Sort to make response deterministic
-	slices.Sort(target.RemovedIds)
-	// Remove any duplicates
-	target.RemovedIds = slices.Compact(target.RemovedIds)
-	// Remove items that were deleted since the end of the last iteration.
-	// If an Alias has been updated and subsequently removed, we don't want
-	// it to appear both in the Items and RemovedIds, so we remove it from the Items.
-	for _, removedId := range target.RemovedIds {
-		if i, ok := idToIndex[removedId]; ok {
-			// Remove the item at index i without preserving order
-			// https://github.com/golang/go/wiki/SliceTricks#delete-without-preserving-order
-			target.Items[i] = target.Items[len(target.Items)-1]
-			target.Items = target.Items[:len(target.Items)-1]
-			// Update the index of the previously last element
-			idToIndex[target.Items[i].Id] = i
-		}
-	}
-	// Since we paginated to the end, we can avoid confusion
-	// for the user by setting the estimated item count to the
-	// length of the items slice. If we don't set this here, it
-	// will equal the value returned in the last response, which is
-	// often much smaller than the total number returned.
-	target.EstItemCount = uint(len(target.Items))
-	// Sort the results again since in-place updates and deletes
-	// may have shuffled items. We sort by created time descending
-	// (most recently created first), same as the API.
-	slices.SortFunc(target.Items, func(i, j *aliases.Alias) int {
-		return j.CreatedTime.Compare(i.CreatedTime)
-	})
-	// Finally, since we made at least 2 requests to the server to fulfill this
-	// function call, resp.Body and resp.Map will only contain the most recent response.
-	// Overwrite them with the true response.
-	target.GetResponse().Body.Reset()
-	if err := json.NewEncoder(target.GetResponse().Body).Encode(target); err != nil {
-		return nil, fmt.Errorf("error encoding final JSON list response: %w", err)
-	}
-	if err := json.Unmarshal(target.GetResponse().Body.Bytes(), &target.GetResponse().Map); err != nil {
-		return nil, fmt.Errorf("error encoding final map list response: %w", err)
-	}
-	// Note: the HTTP response body is consumed by resp.Decode in the loop,
-	// so it doesn't need to be updated (it will always be, and has always been, empty).
-	return target, nil
+	opts, _ := getOpts(opt...)
+	apiClient := aliases.NewClient(c.client)
+	return apiClient.List(ctx, "global",
+		aliases.WithAutomaticVersioning(opts.withAutomaticVersioning),
+		aliases.WithSkipCurlOutput(opts.withSkipCurlOutput),
+		aliases.WithFilter(opts.withFilter),
+		aliases.WithListToken(opts.withListToken),
+		aliases.WithClientDirectedPagination(opts.withClientDirectedPagination),
+		aliases.WithPageSize(opts.withPageSize),
+		aliases.WithRecursive(opts.withRecursive),
+		aliases.WithResourcePathOverride(fmt.Sprintf("users/%s:list-resolvable-aliases", url.PathEscape(userId))),
+	)
+}
+
+func (c *Client) ListResolvableAliasesNextPage(ctx context.Context, userId string, currentPage *aliases.AliasListResult, opt ...Option) (*aliases.AliasListResult, error) {
+	if currentPage == nil {
+		return nil, fmt.Errorf("empty currentPage value passed into ListResolvableAliasesNextPage request")
+	}
+	if userId == "" {
+		return nil, fmt.Errorf("empty userId value passed into ListResolvableAliasesNextPage request")
+	}
+	if c.client == nil {
+		return nil, fmt.Errorf("nil client")
+	}
+	if currentPage.ResponseType == "complete" || currentPage.ResponseType == "" {
+		return nil, fmt.Errorf("no more pages available in ListResolvableAliasesNextPage request")
+	}
+
+	opts, _ := getOpts(opt...)
+	apiClient := aliases.NewClient(c.client)
+	return apiClient.ListNextPage(ctx, currentPage,
+		aliases.WithAutomaticVersioning(opts.withAutomaticVersioning),
+		aliases.WithSkipCurlOutput(opts.withSkipCurlOutput),
+		aliases.WithFilter(opts.withFilter),
+		aliases.WithListToken(opts.withListToken),
+		aliases.WithClientDirectedPagination(opts.withClientDirectedPagination),
+		aliases.WithPageSize(opts.withPageSize),
+		aliases.WithRecursive(opts.withRecursive),
+		aliases.WithResourcePathOverride(fmt.Sprintf("users/%s:list-resolvable-aliases", url.PathEscape(userId))),
+	)
 }

api/users/option.gen.go

@@ -29,6 +29,7 @@ type options struct {
 	withListToken                string
 	withClientDirectedPagination bool
 	withPageSize                 uint32
+	withResourcePathOverride     string
 	withRecursive                bool
 }
 
@@ -115,6 +116,13 @@ func WithPageSize(with uint32) Option {
 	}
 }
 
+// WithResourcePathOverride tells the API to use the provided resource path
+func WithResourcePathOverride(path string) Option {
+	return func(o *options) {
+		o.withResourcePathOverride = path
+	}
+}
+
 // WithRecursive tells the API to use recursion for listing operations on this
 // resource
 func WithRecursive(recurse bool) Option {

api/users/user.gen.go

@@ -324,7 +324,12 @@ func (c *Client) List(ctx context.Context, scopeId string, opt ...Option) (*User
 	opts, apiOpts := getOpts(opt...)
 	opts.queryMap["scope_id"] = scopeId
 
-	req, err := c.client.NewRequest(ctx, "GET", "users", nil, apiOpts...)
+	requestPath := "users"
+	if opts.withResourcePathOverride != "" {
+		requestPath = opts.withResourcePathOverride
+	}
+
+	req, err := c.client.NewRequest(ctx, "GET", requestPath, nil, apiOpts...)
 	if err != nil {
 		return nil, fmt.Errorf("error creating List request: %w", err)
 	}
@@ -480,7 +485,12 @@ func (c *Client) ListNextPage(ctx context.Context, currentPage *UserListResult,
 		opts.queryMap["page_size"] = strconv.FormatUint(uint64(currentPage.pageSize), 10)
 	}
 
-	req, err := c.client.NewRequest(ctx, "GET", "users", nil, apiOpts...)
+	requestPath := "users"
+	if opts.withResourcePathOverride != "" {
+		requestPath = opts.withResourcePathOverride
+	}
+
+	req, err := c.client.NewRequest(ctx, "GET", requestPath, nil, apiOpts...)
 	if err != nil {
 		return nil, fmt.Errorf("error creating List request: %w", err)
 	}