internal/pagination: add generic ListPlugins functions

These new methods can be used in places where the caller wants
to automatically paginate over a resource that is returned
together with a slice of plugins, such as host catalogs.

internal/pagination/pagination_plugins.go

@@ -0,0 +1,306 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package pagination
+
+import (
+	"context"
+	"time"
+
+	"github.com/hashicorp/boundary/internal/boundary"
+	"github.com/hashicorp/boundary/internal/errors"
+	"github.com/hashicorp/boundary/internal/listtoken"
+	"github.com/hashicorp/boundary/internal/plugin"
+)
+
+// ListPluginsFilterFunc is a callback used to filter out resources that don't match
+// some criteria. The function must return true for items that should be included in the final
+// result. Returning an error results in an error being returned from the pagination.
+type ListPluginsFilterFunc[T boundary.Resource] func(ctx context.Context, item T, plugin map[string]*plugin.Plugin) (bool, error)
+
+// ListPluginsItemsFunc returns a slice of T that are ordered after prevPageLastItem according to
+// the implementation of the function. If prevPageLastItem is empty, it should return
+// a slice of T from the start, as defined by the function. It also returns the timestamp
+// of the DB transaction used to list the items.
+type ListPluginsItemsFunc[T boundary.Resource] func(ctx context.Context, prevPageLastItem T, limit int) ([]T, []*plugin.Plugin, time.Time, error)
+
+// ListPlugins returns a ListResponse and a map of plugin id to the plugins associated
+// with the returned resources. The response will contain at most pageSize number of items.
+// Items are fetched using the listItemsFn and checked using
+// the filterItemFn to determine if they should be included in the response.
+// The response includes a new list token used to continue pagination or refresh.
+// The estimatedCountFn is used to provide an estimated total number of
+// items that can be returned by making additional requests using the returned
+// list token.
+func ListPlugins[T boundary.Resource](
+	ctx context.Context,
+	grantsHash []byte,
+	pageSize int,
+	filterItemFn ListPluginsFilterFunc[T],
+	listItemsFn ListPluginsItemsFunc[T],
+	estimatedCountFn EstimatedCountFunc,
+) (*ListResponse[T], map[string]*plugin.Plugin, error) {
+	const op = "pagination.ListsPlugin"
+
+	switch {
+	case len(grantsHash) == 0:
+		return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing grants hash")
+	case pageSize < 1:
+		return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "page size must be at least 1")
+	case filterItemFn == nil:
+		return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing filter item callback")
+	case listItemsFn == nil:
+		return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing list items callback")
+	case estimatedCountFn == nil:
+		return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing estimated count callback")
+	}
+
+	items, plgs, completeListing, listTime, err := listPlugins(ctx, pageSize, filterItemFn, listItemsFn)
+	if err != nil {
+		return nil, nil, errors.Wrap(ctx, err, op)
+	}
+
+	resp, err := buildListResp(ctx, grantsHash, items, completeListing, listTime, estimatedCountFn)
+	if err != nil {
+		return nil, nil, errors.Wrap(ctx, err, op)
+	}
+
+	return resp, plgs, nil
+}
+
+// ListPluginsPage returns a ListResponse and a map of plugin id to the plugins associated
+// with the returned resources. The response will contain at most pageSize
+// number of items. Items are fetched using the listItemsFn and checked using
+// the filterItemFn to determine if they should be included in the response.
+// Items will be fetched based on the contents of the list token. The list
+// token must contain a PaginationToken component.
+// The response includes a new list token used to continue pagination or refresh.
+// The estimatedCountFn is used to provide an estimated total number of
+// items that can be returned by making additional requests using the returned
+// list token.
+func ListPluginsPage[T boundary.Resource](
+	ctx context.Context,
+	grantsHash []byte,
+	pageSize int,
+	filterItemFn ListPluginsFilterFunc[T],
+	listItemsFn ListPluginsItemsFunc[T],
+	estimatedCountFn EstimatedCountFunc,
+	tok *listtoken.Token,
+) (*ListResponse[T], map[string]*plugin.Plugin, error) {
+	const op = "pagination.ListPluginsPage"
+
+	switch {
+	case len(grantsHash) == 0:
+		return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing grants hash")
+	case pageSize < 1:
+		return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "page size must be at least 1")
+	case filterItemFn == nil:
+		return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing filter item callback")
+	case listItemsFn == nil:
+		return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing list items callback")
+	case estimatedCountFn == nil:
+		return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing estimated count callback")
+	case tok == nil:
+		return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing list token")
+	}
+	if _, ok := tok.Subtype.(*listtoken.PaginationToken); !ok {
+		return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "token did not have a pagination token component")
+	}
+
+	items, plgs, completeListing, listTime, err := listPlugins(ctx, pageSize, filterItemFn, listItemsFn)
+	if err != nil {
+		return nil, nil, errors.Wrap(ctx, err, op)
+	}
+
+	resp, err := buildListPageResp(ctx, completeListing, nil, time.Time{} /* no deleted ids time */, items, listTime, tok, estimatedCountFn)
+	if err != nil {
+		return nil, nil, errors.Wrap(ctx, err, op)
+	}
+
+	return resp, plgs, nil
+}
+
+// ListPluginsRefresh returns a ListResponse and a map of plugin id to the plugins associated
+// with the returned resources. The response will contain at most pageSize
+// number of items. Items are fetched using the listItemsFn and checked using
+// the filterItemFn to determine if they should be included in the response.
+// Items will be fetched based on the contents of the list token. The list
+// token must contain a StartRefreshToken component.
+// The response includes a new list token used to continue pagination or refresh.
+// The estimatedCountFn is used to provide an estimated total number of
+// items that can be returned by making additional requests using the returned
+// list token. The listDeletedIDsFn is used to list the IDs of any
+// resources that have been deleted since the list token was last used.
+func ListPluginsRefresh[T boundary.Resource](
+	ctx context.Context,
+	grantsHash []byte,
+	pageSize int,
+	filterItemFn ListPluginsFilterFunc[T],
+	listItemsFn ListPluginsItemsFunc[T],
+	estimatedCountFn EstimatedCountFunc,
+	listDeletedIDsFn ListDeletedIDsFunc,
+	tok *listtoken.Token,
+) (*ListResponse[T], map[string]*plugin.Plugin, error) {
+	const op = "pagination.ListPluginsRefresh"
+
+	switch {
+	case len(grantsHash) == 0:
+		return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing grants hash")
+	case pageSize < 1:
+		return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "page size must be at least 1")
+	case filterItemFn == nil:
+		return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing filter item callback")
+	case listItemsFn == nil:
+		return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing list items callback")
+	case estimatedCountFn == nil:
+		return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing estimated count callback")
+	case listDeletedIDsFn == nil:
+		return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing list deleted IDs callback")
+	case tok == nil:
+		return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing list token")
+	}
+	srt, ok := tok.Subtype.(*listtoken.StartRefreshToken)
+	if !ok {
+		return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "token did not have a start-refresh token component")
+	}
+
+	deletedIds, deletedIdsTime, err := listDeletedIDsFn(ctx, srt.PreviousDeletedIdsTime)
+	if err != nil {
+		return nil, nil, errors.Wrap(ctx, err, op)
+	}
+
+	items, plgs, completeListing, listTime, err := listPlugins(ctx, pageSize, filterItemFn, listItemsFn)
+	if err != nil {
+		return nil, nil, errors.Wrap(ctx, err, op)
+	}
+
+	resp, err := buildListPageResp(ctx, completeListing, deletedIds, deletedIdsTime, items, listTime, tok, estimatedCountFn)
+	if err != nil {
+		return nil, nil, errors.Wrap(ctx, err, op)
+	}
+
+	return resp, plgs, nil
+}
+
+// ListPluginsRefreshPage returns a ListResponse and a plugin. The response will contain at most pageSize
+// number of items. Items are fetched using the listItemsFn and checked using
+// the filterItemFn to determine if they should be included in the response.
+// Items will be fetched based on the contents of the list token. The list
+// token must contain a RefreshToken component.
+// The response includes a new list token used to continue pagination or refresh.
+// The estimatedCountFn is used to provide an estimated total number of
+// items that can be returned by making additional requests using the returned
+// list token. The listDeletedIDsFn is used to list the IDs of any
+// resources that have been deleted since the list token was last used.
+func ListPluginsRefreshPage[T boundary.Resource](
+	ctx context.Context,
+	grantsHash []byte,
+	pageSize int,
+	filterItemFn ListPluginsFilterFunc[T],
+	listItemsFn ListPluginsItemsFunc[T],
+	estimatedCountFn EstimatedCountFunc,
+	listDeletedIDsFn ListDeletedIDsFunc,
+	tok *listtoken.Token,
+) (*ListResponse[T], map[string]*plugin.Plugin, error) {
+	const op = "pagination.ListPluginsRefreshPage"
+
+	switch {
+	case len(grantsHash) == 0:
+		return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing grants hash")
+	case pageSize < 1:
+		return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "page size must be at least 1")
+	case filterItemFn == nil:
+		return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing filter item callback")
+	case listItemsFn == nil:
+		return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing list items callback")
+	case estimatedCountFn == nil:
+		return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing estimated count callback")
+	case listDeletedIDsFn == nil:
+		return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing list deleted IDs callback")
+	case tok == nil:
+		return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing list token")
+	}
+	rt, ok := tok.Subtype.(*listtoken.RefreshToken)
+	if !ok {
+		return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "token did not have a refresh token component")
+	}
+
+	deletedIds, deletedIdsTime, err := listDeletedIDsFn(ctx, rt.PreviousDeletedIdsTime)
+	if err != nil {
+		return nil, nil, errors.Wrap(ctx, err, op)
+	}
+
+	items, plgs, completeListing, listTime, err := listPlugins(ctx, pageSize, filterItemFn, listItemsFn)
+	if err != nil {
+		return nil, nil, errors.Wrap(ctx, err, op)
+	}
+
+	resp, err := buildListPageResp(ctx, completeListing, deletedIds, deletedIdsTime, items, listTime, tok, estimatedCountFn)
+	if err != nil {
+		return nil, nil, errors.Wrap(ctx, err, op)
+	}
+
+	return resp, plgs, nil
+}
+
+func listPlugins[T boundary.Resource](
+	ctx context.Context,
+	pageSize int,
+	filterItemFn ListPluginsFilterFunc[T],
+	listItemsFn ListPluginsItemsFunc[T],
+) ([]T, map[string]*plugin.Plugin, bool, time.Time, error) {
+	const op = "pagination.list"
+
+	var lastItem T
+	plgs := map[string]*plugin.Plugin{}
+	var firstListTime time.Time
+	limit := pageSize + 1
+	items := make([]T, 0, limit)
+dbLoop:
+	for {
+		// Request another page from the DB until we fill the final items
+		page, newPlgs, listTime, err := listItemsFn(ctx, lastItem, limit)
+		if err != nil {
+			return nil, nil, false, time.Time{}, errors.Wrap(ctx, err, op)
+		}
+		// Assign the firstListTime once, to ensure we always store the start of listing,
+		// rather the timestamp of the last listing.
+		if firstListTime.IsZero() {
+			firstListTime = listTime
+		}
+		for _, plg := range newPlgs {
+			if _, ok := plgs[plg.PublicId]; !ok {
+				plgs[plg.PublicId] = plg
+			}
+		}
+		for _, item := range page {
+			ok, err := filterItemFn(ctx, item, plgs)
+			if err != nil {
+				return nil, nil, false, time.Time{}, errors.Wrap(ctx, err, op)
+			}
+			if ok {
+				items = append(items, item)
+				// If we filled the items after filtering,
+				// we're done.
+				if len(items) == cap(items) {
+					break dbLoop
+				}
+			}
+		}
+		// If the current page was shorter than the limit, stop iterating
+		if len(page) < limit {
+			break dbLoop
+		}
+
+		lastItem = page[len(page)-1]
+	}
+	// If we couldn't fill the items, it was a complete listing.
+	completeListing := len(items) < cap(items)
+	if !completeListing {
+		// Items is of size pageSize+1, so
+		// truncate if it was filled.
+		items = items[:pageSize]
+	}
+
+	return items, plgs, completeListing, firstListTime, nil
+}