handlers/targets: refactor for new pagination design

Refactor the application layer to use the new domain layer
pagination functionality.

internal/daemon/controller/handlers/targets/target_service.go

@@ -29,9 +29,9 @@ import (
 	"github.com/hashicorp/boundary/internal/host/plugin"
 	"github.com/hashicorp/boundary/internal/host/static"
 	"github.com/hashicorp/boundary/internal/kms"
+	"github.com/hashicorp/boundary/internal/listtoken"
 	"github.com/hashicorp/boundary/internal/pagination"
 	"github.com/hashicorp/boundary/internal/perms"
-	"github.com/hashicorp/boundary/internal/refreshtoken"
 	"github.com/hashicorp/boundary/internal/requests"
 	"github.com/hashicorp/boundary/internal/server"
 	"github.com/hashicorp/boundary/internal/session"
@@ -243,67 +243,122 @@ func (s Service) ListTargets(ctx context.Context, req *pbs.ListTargetsRequest) (
 	if req.GetPageSize() != 0 && uint(req.GetPageSize()) < s.maxPageSize {
 		pageSize = int(req.GetPageSize())
 	}
-	filter, err := handlers.NewFilter(ctx, req.GetFilter())
-	if err != nil {
-		return nil, err
+	var filterItemFn func(ctx context.Context, item target.Target) (bool, error)
+	switch {
+	case req.GetFilter() != "":
+		// Only use a filter if we need to
+		filter, err := handlers.NewFilter(ctx, req.GetFilter())
+		if err != nil {
+			return nil, err
+		}
+		// TODO: replace the need for this function with some way to convert the `filter`
+		// to a domain type. This would allow filtering to happen in the domain, and we could
+		// remove this callback altogether.
+		filterItemFn = func(ctx context.Context, item target.Target) (bool, error) {
+			pbItem, err := toProto(ctx, item, newOutputOpts(ctx, item, authResults, authzScopes)...)
+			if err != nil {
+				return false, err
+			}
+			filterable, err := subtypes.Filterable(ctx, pbItem)
+			if err != nil {
+				return false, err
+			}
+			return filter.Match(filterable), nil
+		}
+	default:
+		filterItemFn = func(ctx context.Context, item target.Target) (bool, error) {
+			return true, nil
+		}
 	}
 
 	repo, err := s.repoFn(target.WithPermissions(userPerms))
 	if err != nil {
 		return nil, err
 	}
-	// TODO: replace the need for this function with some way to convert the `filter`
-	// to a domain type. This would allow filtering to happen in the domain, and we could
-	// remove this callback altogether.
-	filterItemFn := func(ctx context.Context, item target.Target) (bool, error) {
-		pbItem, err := toProto(ctx, item, newOutputOpts(ctx, item, authResults, authzScopes)...)
-		if err != nil {
-			return false, err
-		}
-
-		filterable, err := subtypes.Filterable(ctx, pbItem)
-		if err != nil {
-			return false, err
-		}
-		return filter.Match(filterable), nil
-	}
 	grantsHash, err := authResults.GrantsHash(ctx)
 	if err != nil {
 		return nil, err
 	}
 
 	var listResp *pagination.ListResponse[target.Target]
-	if req.GetRefreshToken() == "" {
+	var sortBy string
+	if req.GetListToken() == "" {
+		sortBy = "created_time"
 		listResp, err = target.List(ctx, grantsHash, pageSize, filterItemFn, repo)
 		if err != nil {
 			return nil, err
 		}
 	} else {
-		rt, err := handlers.ParseRefreshToken(ctx, req.GetRefreshToken())
+		pt, err := handlers.ParseListToken(ctx, req.GetListToken())
 		if err != nil {
 			return nil, err
 		}
-		// We're doing the conversion from the protobuf types to the
-		// domain types here rather than in the domain so that the domain
-		// doesn't need to know about the protobuf types.
-		domainRefreshToken, err := refreshtoken.New(
-			ctx,
-			rt.CreatedTime.AsTime(),
-			rt.UpdatedTime.AsTime(),
-			handlers.RefreshTokenResourceToResource(rt.ResourceType),
-			rt.GrantsHash,
-			rt.LastItemId,
-			rt.LastItemUpdatedTime.AsTime(),
-		)
-		if err != nil {
-			return nil, err
-		}
-		if err := domainRefreshToken.Validate(ctx, resource.Target, grantsHash); err != nil {
-			return nil, err
-		}
-		listResp, err = target.ListRefresh(ctx, grantsHash, pageSize, filterItemFn, domainRefreshToken, repo)
-		if err != nil {
-			return nil, err
+		switch st := pt.Token.(type) {
+		case *pbs.ListToken_PaginationToken:
+			sortBy = "created_time"
+			listToken, err := listtoken.NewPagination(
+				ctx,
+				pt.CreateTime.AsTime(),
+				handlers.ListTokenResourceToResource(pt.ResourceType),
+				pt.GrantsHash,
+				st.PaginationToken.LastItemId,
+				st.PaginationToken.LastItemCreateTime.AsTime(),
+			)
+			if err != nil {
+				return nil, err
+			}
+			if err := listToken.Validate(ctx, resource.Target, grantsHash); err != nil {
+				return nil, err
+			}
+			listResp, err = target.ListPage(ctx, grantsHash, pageSize, filterItemFn, listToken, repo)
+			if err != nil {
+				return nil, err
+			}
+		case *pbs.ListToken_StartRefreshToken:
+			sortBy = "updated_time"
+			listToken, err := listtoken.NewStartRefresh(
+				ctx,
+				pt.CreateTime.AsTime(),
+				handlers.ListTokenResourceToResource(pt.ResourceType),
+				pt.GrantsHash,
+				st.StartRefreshToken.PreviousDeletedIdsTime.AsTime(),
+				st.StartRefreshToken.PreviousPhaseUpperBound.AsTime(),
+			)
+			if err != nil {
+				return nil, err
+			}
+			if err := listToken.Validate(ctx, resource.Target, grantsHash); err != nil {
+				return nil, err
+			}
+			listResp, err = target.ListRefresh(ctx, grantsHash, pageSize, filterItemFn, listToken, repo)
+			if err != nil {
+				return nil, err
+			}
+		case *pbs.ListToken_RefreshToken:
+			sortBy = "updated_time"
+			listToken, err := listtoken.NewRefresh(
+				ctx,
+				pt.CreateTime.AsTime(),
+				handlers.ListTokenResourceToResource(pt.ResourceType),
+				pt.GrantsHash,
+				st.RefreshToken.PreviousDeletedIdsTime.AsTime(),
+				st.RefreshToken.PhaseUpperBound.AsTime(),
+				st.RefreshToken.PhaseLowerBound.AsTime(),
+				st.RefreshToken.LastItemId,
+				st.RefreshToken.LastItemUpdateTime.AsTime(),
+			)
+			if err != nil {
+				return nil, err
+			}
+			if err := listToken.Validate(ctx, resource.Target, grantsHash); err != nil {
+				return nil, err
+			}
+			listResp, err = target.ListRefreshPage(ctx, grantsHash, pageSize, filterItemFn, listToken, repo)
+			if err != nil {
+				return nil, err
+			}
+		default:
+			return nil, handlers.ApiErrorWithCodeAndMessage(codes.InvalidArgument, "unexpected list token subtype: %T", st)
 		}
 	}
 
@@ -324,22 +379,12 @@ func (s Service) ListTargets(ctx context.Context, req *pbs.ListTargetsRequest) (
 		EstItemCount: uint32(listResp.EstimatedItemCount),
 		RemovedIds:   listResp.DeletedIds,
 		ResponseType: respType,
-		SortBy:       "updated_time",
-		SortDir:      "asc",
+		SortBy:       sortBy,
+		SortDir:      "desc",
 	}
 
-	if listResp.RefreshToken != nil {
-		if listResp.RefreshToken.ResourceType != resource.Target {
-			return nil, errors.New(ctx, errors.Internal, op, "refresh token resource type does not match service resource type")
-		}
-		resp.RefreshToken, err = handlers.MarshalRefreshToken(ctx, &pbs.ListRefreshToken{
-			CreatedTime:         timestamppb.New(listResp.RefreshToken.CreatedTime),
-			UpdatedTime:         timestamppb.New(listResp.RefreshToken.UpdatedTime),
-			ResourceType:        pbs.ResourceType_RESOURCE_TYPE_TARGET,
-			GrantsHash:          listResp.RefreshToken.GrantsHash,
-			LastItemId:          listResp.RefreshToken.LastItemId,
-			LastItemUpdatedTime: timestamppb.New(listResp.RefreshToken.LastItemUpdatedTime),
-		})
+	if listResp.ListToken != nil {
+		resp.ListToken, err = handlers.MarshalListToken(ctx, listResp.ListToken, pbs.ResourceType_RESOURCE_TYPE_TARGET)
 		if err != nil {
 			return nil, err
 		}

internal/daemon/controller/handlers/targets/tcp/target_service_test.go

@@ -12,6 +12,7 @@ import (
 	"errors"
 	"fmt"
 	"path"
+	"slices"
 	"strings"
 	"sync/atomic"
 	"testing"
@@ -46,6 +47,7 @@ import (
 	"github.com/hashicorp/boundary/internal/session"
 	"github.com/hashicorp/boundary/internal/target"
 	"github.com/hashicorp/boundary/internal/target/tcp"
+	"github.com/hashicorp/boundary/internal/target/tcp/store"
 	"github.com/hashicorp/boundary/internal/types/scope"
 	credlibpb "github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/credentiallibraries"
 	credpb "github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/credentials"
@@ -258,6 +260,8 @@ func TestList(t *testing.T) {
 	t.Parallel()
 	ctx := context.Background()
 	conn, _ := db.TestSetup(t, "postgres")
+	sqlDB, err := conn.SqlDB(ctx)
+	require.NoError(t, err)
 	wrapper := db.TestWrapper(t)
 	kms := kms.TestKms(t, conn, wrapper)
 
@@ -333,6 +337,14 @@ func TestList(t *testing.T) {
 		})
 	}
 
+	// Run analyze to update postgres estimates
+	_, err = sqlDB.ExecContext(ctx, "analyze")
+	require.NoError(t, err)
+
+	// Reverse slices since response is ordered by created_time descending (newest first)
+	slices.Reverse(wantTars)
+	slices.Reverse(totalTars)
+
 	cases := []struct {
 		name string
 		req  *pbs.ListTargetsRequest
@@ -342,7 +354,13 @@ func TestList(t *testing.T) {
 		{
 			name: "List Many Targets",
 			req:  &pbs.ListTargetsRequest{ScopeId: proj.GetPublicId()},
-			res:  &pbs.ListTargetsResponse{Items: wantTars},
+			res: &pbs.ListTargetsResponse{
+				Items:        wantTars,
+				ResponseType: "complete",
+				SortBy:       "created_time",
+				SortDir:      "desc",
+				EstItemCount: 5,
+			},
 		},
 		{
 			name: "List No Targets",
@@ -352,22 +370,44 @@ func TestList(t *testing.T) {
 		{
 			name: "List Targets Recursively",
 			req:  &pbs.ListTargetsRequest{ScopeId: scope.Global.String(), Recursive: true},
-			res:  &pbs.ListTargetsResponse{Items: totalTars},
+			res: &pbs.ListTargetsResponse{
+				Items:        totalTars,
+				ResponseType: "complete",
+				SortBy:       "created_time",
+				SortDir:      "desc",
+				EstItemCount: 10,
+			},
 		},
 		{
 			name: "Paginate listing",
 			req:  &pbs.ListTargetsRequest{ScopeId: scope.Global.String(), Recursive: true, PageSize: 2},
-			res:  &pbs.ListTargetsResponse{Items: totalTars[:2]},
+			res: &pbs.ListTargetsResponse{
+				Items:        totalTars[:2],
+				ResponseType: "delta",
+				SortBy:       "created_time",
+				SortDir:      "desc",
+				EstItemCount: 10,
+			},
 		},
 		{
 			name: "Filter To Many Targets",
 			req:  &pbs.ListTargetsRequest{ScopeId: scope.Global.String(), Recursive: true, Filter: fmt.Sprintf(`"/item/scope/id"==%q`, proj.GetPublicId())},
-			res:  &pbs.ListTargetsResponse{Items: wantTars},
+			res: &pbs.ListTargetsResponse{
+				Items:        wantTars,
+				ResponseType: "complete",
+				SortBy:       "created_time",
+				SortDir:      "desc",
+				EstItemCount: 5,
+			},
 		},
 		{
 			name: "Filter To No Targets",
 			req:  &pbs.ListTargetsRequest{ScopeId: proj.GetPublicId(), Filter: `"/item/id"=="doesnt match"`},
-			res:  &pbs.ListTargetsResponse{},
+			res: &pbs.ListTargetsResponse{
+				ResponseType: "complete",
+				SortBy:       "created_time",
+				SortDir:      "desc",
+			},
 		},
 		{
 			name: "Filter Bad Format",
@@ -434,7 +474,12 @@ func TestList(t *testing.T) {
 }
 
 func TestListPagination(t *testing.T) {
-	t.Parallel()
+	// Set database read timeout to avoid duplicates in response
+	oldReadTimeout := globals.RefreshReadLookbackDuration
+	globals.RefreshReadLookbackDuration = 0
+	t.Cleanup(func() {
+		globals.RefreshReadLookbackDuration = oldReadTimeout
+	})
 	ctx := context.Background()
 	conn, _ := db.TestSetup(t, "postgres")
 	sqlDB, err := conn.SqlDB(ctx)
@@ -486,12 +531,13 @@ func TestListPagination(t *testing.T) {
 			Address:                &wrapperspb.StringValue{},
 		})
 	}
+	// Reverse since we read items in descending order (newest first)
+	slices.Reverse(allTargets)
 
 	// Run analyze to update postgres estimates
 	_, err = sqlDB.ExecContext(ctx, "analyze")
 	require.NoError(t, err)
 
-	// Test without anon user
 	requestInfo := authpb.RequestInfo{
 		TokenFormat: uint32(auth.AuthTokenTypeBearer),
 		PublicId:    at.GetPublicId(),
@@ -502,77 +548,74 @@ func TestListPagination(t *testing.T) {
 
 	// Start paginating, recursively
 	req := &pbs.ListTargetsRequest{
-		ScopeId:      "global",
-		Recursive:    true,
-		Filter:       "",
-		RefreshToken: "",
-		PageSize:     2,
+		ScopeId:   "global",
+		Recursive: true,
+		Filter:    "",
+		ListToken: "",
+		PageSize:  2,
 	}
 	got, err := s.ListTargets(ctx, req)
 	require.NoError(t, err)
 	require.Len(t, got.GetItems(), 2)
-	// Compare without comparing the refresh token
+	// Compare without comparing the list token
 	assert.Empty(t,
 		cmp.Diff(
 			got,
 			&pbs.ListTargetsResponse{
 				Items:        allTargets[0:2],
 				ResponseType: "delta",
-				RefreshToken: "",
-				SortBy:       "updated_time",
-				SortDir:      "asc",
+				SortBy:       "created_time",
+				SortDir:      "desc",
 				RemovedIds:   nil,
 				EstItemCount: 10,
 			},
 			protocmp.Transform(),
-			protocmp.IgnoreFields(&pbs.ListTargetsResponse{}, "refresh_token"),
+			protocmp.IgnoreFields(&pbs.ListTargetsResponse{}, "list_token"),
 		),
 	)
 
 	// Request second page
-	req.RefreshToken = got.RefreshToken
+	req.ListToken = got.ListToken
 	got, err = s.ListTargets(ctx, req)
 	require.NoError(t, err)
 	require.Len(t, got.GetItems(), 2)
-	// Compare without comparing the refresh token
+	// Compare without comparing the list token
 	assert.Empty(t,
 		cmp.Diff(
 			got,
 			&pbs.ListTargetsResponse{
 				Items:        allTargets[2:4],
 				ResponseType: "delta",
-				RefreshToken: "",
-				SortBy:       "updated_time",
-				SortDir:      "asc",
+				SortBy:       "created_time",
+				SortDir:      "desc",
 				RemovedIds:   nil,
 				EstItemCount: 10,
 			},
 			protocmp.Transform(),
-			protocmp.IgnoreFields(&pbs.ListTargetsResponse{}, "refresh_token"),
+			protocmp.IgnoreFields(&pbs.ListTargetsResponse{}, "list_token"),
 		),
 	)
 
 	// Request rest of results
-	req.RefreshToken = got.RefreshToken
+	req.ListToken = got.ListToken
 	req.PageSize = 10
 	got, err = s.ListTargets(ctx, req)
 	require.NoError(t, err)
 	require.Len(t, got.GetItems(), 6)
-	// Compare without comparing the refresh token
+	// Compare without comparing the list token
 	assert.Empty(t,
 		cmp.Diff(
 			got,
 			&pbs.ListTargetsResponse{
 				Items:        allTargets[4:],
 				ResponseType: "complete",
-				RefreshToken: "",
-				SortBy:       "updated_time",
-				SortDir:      "asc",
+				SortBy:       "created_time",
+				SortDir:      "desc",
 				RemovedIds:   nil,
 				EstItemCount: 10,
 			},
 			protocmp.Transform(),
-			protocmp.IgnoreFields(&pbs.ListTargetsResponse{}, "refresh_token"),
+			protocmp.IgnoreFields(&pbs.ListTargetsResponse{}, "list_token"),
 		),
 	)
 
@@ -593,45 +636,91 @@ func TestListPagination(t *testing.T) {
 		AuthorizedActions:      testAuthorizedActions,
 		Address:                &wrapperspb.StringValue{},
 	}
-	allTargets = append(allTargets, newTarget)
+	// Add to the front since it's most recently updated
+	allTargets = append([]*pb.Target{newTarget}, allTargets...)
 
 	// Delete one of the other targets
-	_, err = repo.DeleteTarget(ctx, allTargets[0].Id)
+	_, err = repo.DeleteTarget(ctx, allTargets[len(allTargets)-1].Id)
 	require.NoError(t, err)
-	deletedTarget := allTargets[0]
-	allTargets = allTargets[1:]
+	deletedTarget := allTargets[len(allTargets)-1]
+	allTargets = allTargets[:len(allTargets)-1]
+
+	// Update one of the other targets
+	allTargets[1].Name = wrapperspb.String("new-name")
+	allTargets[1].Version = 2
+	updatedTarget := &tcp.Target{
+		Target: &store.Target{
+			PublicId:  allTargets[1].Id,
+			Name:      allTargets[1].Name.GetValue(),
+			ProjectId: allTargets[1].ScopeId,
+		},
+	}
+	tg, _, err := repo.UpdateTarget(ctx, updatedTarget, 1, []string{"name"})
+	require.NoError(t, err)
+	allTargets[1].UpdatedTime = tg.GetUpdateTime().GetTimestamp()
+	allTargets[1].Version = tg.GetVersion()
+	// Add to the front since it's most recently updated
+	allTargets = append(
+		[]*pb.Target{allTargets[1]},
+		append(
+			[]*pb.Target{allTargets[0]},
+			allTargets[2:]...,
+		)...,
+	)
 
 	// Run analyze to update postgres estimates
 	_, err = sqlDB.ExecContext(ctx, "analyze")
 	require.NoError(t, err)
 
 	// Request updated results
-	req.RefreshToken = got.RefreshToken
+	req.ListToken = got.ListToken
+	req.PageSize = 1
 	got, err = s.ListTargets(ctx, req)
 	require.NoError(t, err)
 	require.Len(t, got.GetItems(), 1)
-	// Compare without comparing the refresh token
+	// Compare without comparing the list token
 	assert.Empty(t,
 		cmp.Diff(
 			got,
 			&pbs.ListTargetsResponse{
-				Items:        allTargets[len(allTargets)-1:],
-				ResponseType: "complete",
-				RefreshToken: "",
+				Items:        []*pb.Target{allTargets[0]},
+				ResponseType: "delta",
 				SortBy:       "updated_time",
-				SortDir:      "asc",
+				SortDir:      "desc",
 				// Should contain the deleted target
 				RemovedIds:   []string{deletedTarget.Id},
 				EstItemCount: 10,
 			},
 			protocmp.Transform(),
-			protocmp.IgnoreFields(&pbs.ListTargetsResponse{}, "refresh_token"),
+			protocmp.IgnoreFields(&pbs.ListTargetsResponse{}, "list_token"),
+		),
+	)
+
+	// Get next page
+	req.ListToken = got.ListToken
+	got, err = s.ListTargets(ctx, req)
+	require.NoError(t, err)
+	require.Len(t, got.GetItems(), 1)
+	// Compare without comparing the list token
+	assert.Empty(t,
+		cmp.Diff(
+			got,
+			&pbs.ListTargetsResponse{
+				Items:        []*pb.Target{allTargets[1]},
+				ResponseType: "complete",
+				SortBy:       "updated_time",
+				SortDir:      "desc",
+				RemovedIds:   nil,
+				EstItemCount: 10,
+			},
+			protocmp.Transform(),
+			protocmp.IgnoreFields(&pbs.ListTargetsResponse{}, "list_token"),
 		),
 	)
 
 	// Request new page with filter requiring looping
 	// to fill the page.
-	req.RefreshToken = ""
+	req.ListToken = ""
 	req.PageSize = 1
 	req.Filter = fmt.Sprintf(`"/item/id"==%q or "/item/id"==%q`, allTargets[len(allTargets)-2].Id, allTargets[len(allTargets)-1].Id)
 	got, err = s.ListTargets(ctx, req)
@@ -643,18 +732,17 @@ func TestListPagination(t *testing.T) {
 			&pbs.ListTargetsResponse{
 				Items:        []*pb.Target{allTargets[len(allTargets)-2]},
 				ResponseType: "delta",
-				RefreshToken: "",
-				SortBy:       "updated_time",
-				SortDir:      "asc",
+				SortBy:       "created_time",
+				SortDir:      "desc",
 				// Should be empty again
 				RemovedIds:   nil,
 				EstItemCount: 10,
 			},
 			protocmp.Transform(),
-			protocmp.IgnoreFields(&pbs.ListTargetsResponse{}, "refresh_token"),
+			protocmp.IgnoreFields(&pbs.ListTargetsResponse{}, "list_token"),
 		),
 	)
-	req.RefreshToken = got.RefreshToken
+	req.ListToken = got.ListToken
 	// Get the second page
 	got, err = s.ListTargets(ctx, req)
 	require.NoError(t, err)
@@ -665,14 +753,13 @@ func TestListPagination(t *testing.T) {
 			&pbs.ListTargetsResponse{
 				Items:        []*pb.Target{allTargets[len(allTargets)-1]},
 				ResponseType: "complete",
-				RefreshToken: "",
-				SortBy:       "updated_time",
-				SortDir:      "asc",
+				SortBy:       "created_time",
+				SortDir:      "desc",
 				RemovedIds:   nil,
 				EstItemCount: 10,
 			},
 			protocmp.Transform(),
-			protocmp.IgnoreFields(&pbs.ListTargetsResponse{}, "refresh_token"),
+			protocmp.IgnoreFields(&pbs.ListTargetsResponse{}, "list_token"),
 		),
 	)
 }