From e14c1382b3e93f1c7eb66cd9d31af8bcd547ef6f Mon Sep 17 00:00:00 2001 From: Johan Brandhorst-Satzkorn Date: Wed, 13 Dec 2023 15:26:47 -0800 Subject: [PATCH] internal/host: add static host set pagination --- internal/host/static/host_set.go | 11 + internal/host/static/query.go | 4 + internal/host/static/repository_host_set.go | 122 +++- .../host/static/repository_host_set_test.go | 97 ++- .../host/static/service_list_host_sets.go | 62 ++ .../static/service_list_host_sets_ext_test.go | 612 ++++++++++++++++++ .../static/service_list_host_sets_page.go | 79 +++ .../static/service_list_host_sets_refresh.go | 81 +++ .../service_list_host_sets_refresh_page.go | 91 +++ 9 files changed, 1145 insertions(+), 14 deletions(-) create mode 100644 internal/host/static/service_list_host_sets.go create mode 100644 internal/host/static/service_list_host_sets_ext_test.go create mode 100644 internal/host/static/service_list_host_sets_page.go create mode 100644 internal/host/static/service_list_host_sets_refresh.go create mode 100644 internal/host/static/service_list_host_sets_refresh_page.go diff --git a/internal/host/static/host_set.go b/internal/host/static/host_set.go index b88f96503..069146507 100644 --- a/internal/host/static/host_set.go +++ b/internal/host/static/host_set.go @@ -6,6 +6,7 @@ package static import ( "context" + "github.com/hashicorp/boundary/internal/db/timestamp" "github.com/hashicorp/boundary/internal/errors" "github.com/hashicorp/boundary/internal/host/static/store" "github.com/hashicorp/boundary/internal/oplog" @@ -90,3 +91,13 @@ func newHostSetForMembers(setId string, version uint32) *HostSet { }, } } + +type deletedHostSet struct { + PublicId string `gorm:"primary_key"` + DeleteTime *timestamp.Timestamp +} + +// TableName returns the tablename to override the default gorm table name +func (s *deletedHostSet) TableName() string { + return "static_host_set_deleted" +} diff --git a/internal/host/static/query.go b/internal/host/static/query.go index 1daaeb6d6..a3b8bad44 100644 --- a/internal/host/static/query.go +++ b/internal/host/static/query.go @@ -212,5 +212,9 @@ final as ( select * from final order by update_time desc, public_id asc; +` + + estimateCountHostSets = ` +select sum(reltuples::bigint) as estimate from pg_class where oid in ('static_host_set'::regclass) ` ) diff --git a/internal/host/static/repository_host_set.go b/internal/host/static/repository_host_set.go index e6e0a829c..185805b6a 100644 --- a/internal/host/static/repository_host_set.go +++ b/internal/host/static/repository_host_set.go @@ -5,8 +5,10 @@ package static import ( "context" + "database/sql" "fmt" "strings" + "time" "github.com/hashicorp/boundary/globals" "github.com/hashicorp/boundary/internal/db" @@ -248,12 +250,14 @@ func (r *Repository) lookupSet(ctx context.Context, publicId string, opt ...Opti return s, hosts, nil } -// ListSets returns a slice of HostSets for the catalogId. WithLimit is the -// only option supported. -func (r *Repository) ListSets(ctx context.Context, catalogId string, opt ...Option) ([]*HostSet, error) { - const op = "static.(Repository).ListSets" +// listSets returns a slice of HostSets for the catalogId. +// Supported options: +// - WithLimit which overrides the limit set in the Repository object +// - WithStartPageAfterItem which sets where to start listing from +func (r *Repository) listSets(ctx context.Context, catalogId string, opt ...Option) ([]*HostSet, time.Time, error) { + const op = "static.(Repository).listSets" if catalogId == "" { - return nil, errors.New(ctx, errors.InvalidParameter, op, "no catalog id") + return nil, time.Time{}, errors.New(ctx, errors.InvalidParameter, op, "no catalog id") } opts := getOpts(opt...) limit := r.defaultLimit @@ -261,12 +265,68 @@ func (r *Repository) ListSets(ctx context.Context, catalogId string, opt ...Opti // non-zero signals an override of the default limit for the repo. limit = opts.withLimit } - var sets []*HostSet - err := r.reader.SearchWhere(ctx, &sets, "catalog_id = ?", []any{catalogId}, db.WithLimit(limit)) - if err != nil { - return nil, errors.Wrap(ctx, err, op) + whereClause := "catalog_id = @catalog_id" + args := []any{sql.Named("catalog_id", catalogId)} + order := "create_time desc, public_id asc" + if opts.withStartPageAfterItem != nil { + args = append(args, + sql.Named("last_item_create_time", opts.withStartPageAfterItem.GetCreateTime()), + sql.Named("last_item_id", opts.withStartPageAfterItem.GetPublicId()), + ) + whereClause = whereClause + " and (create_time, public_id) < (@last_item_create_time, @last_item_id)" } - return sets, nil + return r.querySets(ctx, whereClause, args, order, limit) +} + +// listSetsRefresh returns a slice of HostSets for the catalogId. +// Supported options: +// - WithLimit which overrides the limit set in the Repository object +// - WithStartPageAfterItem which sets where to start listing from +func (r *Repository) listSetsRefresh(ctx context.Context, catalogId string, updatedAfter time.Time, opt ...Option) ([]*HostSet, time.Time, error) { + const op = "static.(Repository).listSetsRefresh" + switch { + case catalogId == "": + return nil, time.Time{}, errors.New(ctx, errors.InvalidParameter, op, "no catalog id") + case updatedAfter.IsZero(): + return nil, time.Time{}, errors.New(ctx, errors.InvalidParameter, op, "no updated after time") + } + opts := getOpts(opt...) + limit := r.defaultLimit + if opts.withLimit != 0 { + // non-zero signals an override of the default limit for the repo. + limit = opts.withLimit + } + whereClause := "catalog_id = @catalog_id and update_time > @updated_after_time" + args := []any{ + sql.Named("catalog_id", catalogId), + sql.Named("updated_after_time", updatedAfter), + } + order := "update_time desc, public_id asc" + if opts.withStartPageAfterItem != nil { + args = append(args, + sql.Named("last_item_update_time", opts.withStartPageAfterItem.GetUpdateTime()), + sql.Named("last_item_id", opts.withStartPageAfterItem.GetPublicId()), + ) + whereClause = whereClause + " and (update_time, public_id) < (@last_item_update_time, @last_item_id)" + } + return r.querySets(ctx, whereClause, args, order, limit) +} + +func (r *Repository) querySets(ctx context.Context, whereClause string, args []any, order string, limit int) ([]*HostSet, time.Time, error) { + const op = "static.(Repository).querySets" + var sets []*HostSet + var transactionTimestamp time.Time + if _, err := r.writer.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{}, func(r db.Reader, _ db.Writer) error { + err := r.SearchWhere(ctx, &sets, whereClause, args, db.WithLimit(limit), db.WithOrder(order)) + if err != nil { + return err + } + transactionTimestamp, err = r.Now(ctx) + return err + }); err != nil { + return nil, time.Time{}, errors.Wrap(ctx, err, op) + } + return sets, transactionTimestamp, nil } // DeleteSet deletes the host set for the provided id from the repository @@ -310,3 +370,45 @@ func (r *Repository) DeleteSet(ctx context.Context, projectId string, publicId s return rowsDeleted, nil } + +// listDeletedSetIds lists the public IDs of any host sets deleted since the timestamp provided, +// and the timestamp of the transaction within which the host sets were listed. +func (r *Repository) listDeletedSetIds(ctx context.Context, since time.Time) ([]string, time.Time, error) { + const op = "static.(Repository).listDeletedSetIds" + var deleteHostSets []*deletedHostSet + var transactionTimestamp time.Time + if _, err := r.writer.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{}, func(r db.Reader, _ db.Writer) error { + if err := r.SearchWhere(ctx, &deleteHostSets, "delete_time >= ?", []any{since}); err != nil { + return errors.Wrap(ctx, err, op, errors.WithMsg("failed to query deleted host sets")) + } + var err error + transactionTimestamp, err = r.Now(ctx) + if err != nil { + return errors.Wrap(ctx, err, op, errors.WithMsg("failed to get transaction timestamp")) + } + return nil + }); err != nil { + return nil, time.Time{}, err + } + var hostSetIds []string + for _, t := range deleteHostSets { + hostSetIds = append(hostSetIds, t.PublicId) + } + return hostSetIds, transactionTimestamp, nil +} + +// estimatedSetCount returns an estimate of the total number of static host sets. +func (r *Repository) estimatedSetCount(ctx context.Context) (int, error) { + const op = "static.(Repository).estimatedHostSetCount" + rows, err := r.reader.Query(ctx, estimateCountHostSets, nil) + if err != nil { + return 0, errors.Wrap(ctx, err, op, errors.WithMsg("failed to query static host sets")) + } + var count int + for rows.Next() { + if err := r.reader.ScanRows(ctx, rows, &count); err != nil { + return 0, errors.Wrap(ctx, err, op, errors.WithMsg("failed to query static host sets")) + } + } + return count, nil +} diff --git a/internal/host/static/repository_host_set_test.go b/internal/host/static/repository_host_set_test.go index 4306aa49a..f9d3a18ab 100644 --- a/internal/host/static/repository_host_set_test.go +++ b/internal/host/static/repository_host_set_test.go @@ -857,7 +857,7 @@ func TestRepository_LookupSet_Limits(t *testing.T) { } } -func TestRepository_ListSets(t *testing.T) { +func TestRepository_listSets(t *testing.T) { ctx := context.Background() conn, _ := db.TestSetup(t, "postgres") rw := db.New(conn) @@ -901,7 +901,7 @@ func TestRepository_ListSets(t *testing.T) { repo, err := NewRepository(ctx, rw, rw, kms) assert.NoError(err) require.NotNil(repo) - got, err := repo.ListSets(ctx, tt.in, tt.opts...) + got, ttime, err := repo.listSets(ctx, tt.in, tt.opts...) if tt.wantIsErr != 0 { assert.Truef(errors.Match(errors.T(tt.wantIsErr), err), "want err: %q got: %q", tt.wantIsErr, err) assert.Nil(got) @@ -913,11 +913,14 @@ func TestRepository_ListSets(t *testing.T) { protocmp.Transform(), } assert.Empty(cmp.Diff(tt.want, got, opts...)) + // Transaction timestamp should be within ~10 seconds of now + assert.True(time.Now().Before(ttime.Add(10 * time.Second))) + assert.True(time.Now().After(ttime.Add(-10 * time.Second))) }) } } -func TestRepository_ListSets_Limits(t *testing.T) { +func TestRepository_listSets_Limits(t *testing.T) { ctx := context.Background() conn, _ := db.TestSetup(t, "postgres") rw := db.New(conn) @@ -981,13 +984,99 @@ func TestRepository_ListSets_Limits(t *testing.T) { repo, err := NewRepository(ctx, rw, rw, kms, tt.repoOpts...) assert.NoError(err) require.NotNil(repo) - got, err := repo.ListSets(ctx, hostSets[0].CatalogId, tt.listOpts...) + got, ttime, err := repo.listSets(ctx, hostSets[0].CatalogId, tt.listOpts...) require.NoError(err) assert.Len(got, tt.wantLen) + // Transaction timestamp should be within ~10 seconds of now + assert.True(time.Now().Before(ttime.Add(10 * time.Second))) + assert.True(time.Now().After(ttime.Add(-10 * time.Second))) }) } } +func Test_listDeletedSetIds(t *testing.T) { + t.Parallel() + ctx := context.Background() + conn, _ := db.TestSetup(t, "postgres") + wrapper := db.TestWrapper(t) + testKms := kms.TestKms(t, conn, wrapper) + iamRepo := iam.TestRepo(t, conn, wrapper) + _, proj := iam.TestScopes(t, iamRepo) + catalog := TestCatalogs(t, conn, proj.GetPublicId(), 1)[0] + + rw := db.New(conn) + repo, err := NewRepository(ctx, rw, rw, testKms) + require.NoError(t, err) + + // Expect no entries at the start + deletedIds, ttime, err := repo.listDeletedSetIds(ctx, time.Now().AddDate(-1, 0, 0)) + require.NoError(t, err) + require.Empty(t, deletedIds) + // Transaction timestamp should be within ~10 seconds of now + assert.True(t, time.Now().Before(ttime.Add(10*time.Second))) + assert.True(t, time.Now().After(ttime.Add(-10*time.Second))) + + // Delete a set + s := TestSets(t, conn, catalog.GetPublicId(), 1)[0] + _, err = repo.DeleteSet(ctx, proj.GetPublicId(), s.GetPublicId()) + require.NoError(t, err) + + // Expect a single entry + deletedIds, ttime, err = repo.listDeletedSetIds(ctx, time.Now().AddDate(-1, 0, 0)) + require.NoError(t, err) + require.Equal(t, []string{s.GetPublicId()}, deletedIds) + assert.True(t, time.Now().Before(ttime.Add(10*time.Second))) + assert.True(t, time.Now().After(ttime.Add(-10*time.Second))) + + // Try again with the time set to now, expect no entries + deletedIds, ttime, err = repo.listDeletedSetIds(ctx, time.Now()) + require.NoError(t, err) + require.Empty(t, deletedIds) + assert.True(t, time.Now().Before(ttime.Add(10*time.Second))) + assert.True(t, time.Now().After(ttime.Add(-10*time.Second))) +} + +func Test_estimatedHostSetCount(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) + testKms := kms.TestKms(t, conn, wrapper) + iamRepo := iam.TestRepo(t, conn, wrapper) + _, proj := iam.TestScopes(t, iamRepo) + catalog := TestCatalogs(t, conn, proj.GetPublicId(), 1)[0] + + rw := db.New(conn) + repo, err := NewRepository(ctx, rw, rw, testKms) + require.NoError(t, err) + + // Check total entries at start, expect 0 + numItems, err := repo.estimatedSetCount(ctx) + require.NoError(t, err) + assert.Equal(t, 0, numItems) + + // Create a set, expect 1 entries + s := TestSets(t, conn, catalog.GetPublicId(), 1)[0] + // Run analyze to update estimate + _, err = sqlDb.ExecContext(ctx, "analyze") + require.NoError(t, err) + numItems, err = repo.estimatedSetCount(ctx) + require.NoError(t, err) + assert.Equal(t, 1, numItems) + + // Delete the set, expect 0 again + _, err = repo.DeleteSet(ctx, proj.GetPublicId(), s.GetPublicId()) + require.NoError(t, err) + // Run analyze to update estimate + _, err = sqlDb.ExecContext(ctx, "analyze") + require.NoError(t, err) + numItems, err = repo.estimatedSetCount(ctx) + require.NoError(t, err) + assert.Equal(t, 0, numItems) +} + func TestRepository_DeleteSet(t *testing.T) { ctx := context.Background() conn, _ := db.TestSetup(t, "postgres") diff --git a/internal/host/static/service_list_host_sets.go b/internal/host/static/service_list_host_sets.go new file mode 100644 index 000000000..fbbf20f57 --- /dev/null +++ b/internal/host/static/service_list_host_sets.go @@ -0,0 +1,62 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: BUSL-1.1 + +package static + +import ( + "context" + "time" + + "github.com/hashicorp/boundary/internal/errors" + "github.com/hashicorp/boundary/internal/host" + "github.com/hashicorp/boundary/internal/pagination" +) + +// ListHostSets lists up to page size host sets, filtering out entries that +// do not pass the filter item function. It will automatically request +// more host sets from the database, at page size chunks, to fill the page. +// It returns a new list token used to continue pagination or refresh items. +// Host sets are ordered by create time descending (most recently created first). +func ListHostSets( + ctx context.Context, + grantsHash []byte, + pageSize int, + filterItemFn pagination.ListFilterFunc[host.Set], + repo *Repository, + hostCatalogId string, +) (*pagination.ListResponse[host.Set], error) { + const op = "plugin.ListHostSets" + + switch { + case len(grantsHash) == 0: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing grants hash") + case pageSize < 1: + return nil, errors.New(ctx, errors.InvalidParameter, op, "page size must be at least 1") + case filterItemFn == nil: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing filter item callback") + case repo == nil: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing repo") + case hostCatalogId == "": + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing host catalog ID") + } + + listItemsFn := func(ctx context.Context, lastPageItem host.Set, limit int) ([]host.Set, time.Time, error) { + opts := []Option{ + WithLimit(limit), + } + if lastPageItem != nil { + opts = append(opts, WithStartPageAfterItem(lastPageItem)) + } + sSets, listTime, err := repo.listSets(ctx, hostCatalogId, opts...) + if err != nil { + return nil, time.Time{}, err + } + var sets []host.Set + for _, set := range sSets { + sets = append(sets, set) + } + return sets, listTime, nil + } + + return pagination.List(ctx, grantsHash, pageSize, filterItemFn, listItemsFn, repo.estimatedSetCount) +} diff --git a/internal/host/static/service_list_host_sets_ext_test.go b/internal/host/static/service_list_host_sets_ext_test.go new file mode 100644 index 000000000..a863d4374 --- /dev/null +++ b/internal/host/static/service_list_host_sets_ext_test.go @@ -0,0 +1,612 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: BUSL-1.1 + +package static_test + +import ( + "context" + "slices" + "testing" + "time" + + "github.com/google/go-cmp/cmp" + "github.com/google/go-cmp/cmp/cmpopts" + "github.com/hashicorp/boundary/globals" + "github.com/hashicorp/boundary/internal/db" + "github.com/hashicorp/boundary/internal/db/timestamp" + "github.com/hashicorp/boundary/internal/host" + "github.com/hashicorp/boundary/internal/host/static" + "github.com/hashicorp/boundary/internal/host/static/store" + "github.com/hashicorp/boundary/internal/iam" + "github.com/hashicorp/boundary/internal/kms" + "github.com/hashicorp/boundary/internal/listtoken" + "github.com/hashicorp/boundary/internal/types/resource" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/types/known/timestamppb" +) + +func TestService_ListHostSets(t *testing.T) { + // 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) + require.NoError(t, err) + wrapper := db.TestWrapper(t) + testKms := kms.TestKms(t, conn, wrapper) + iamRepo := iam.TestRepo(t, conn, wrapper) + _, proj1 := iam.TestScopes(t, iamRepo) + fiveDaysAgo := time.Now().AddDate(0, 0, -5) + catalog := static.TestCatalogs(t, conn, proj1.GetPublicId(), 1)[0] + var sets []host.Set + for _, set := range static.TestSets(t, conn, catalog.GetPublicId(), 5) { + sets = append(sets, set) + } + // since we sort by create time descending, we need to reverse the slice + slices.Reverse(sets) + + // Run analyze to update host set estimate + _, err = sqlDb.ExecContext(ctx, "analyze") + require.NoError(t, err) + + rw := db.New(conn) + repo, err := static.NewRepository(ctx, rw, rw, testKms) + require.NoError(t, err) + + cmpOpts := []cmp.Option{ + cmpopts.IgnoreUnexported( + static.HostSet{}, + store.HostSet{}, + timestamp.Timestamp{}, + timestamppb.Timestamp{}, + ), + } + + t.Run("ListHostSets validation", func(t *testing.T) { + t.Parallel() + t.Run("missing grants hash", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s host.Set) (bool, error) { + return true, nil + } + _, err := static.ListHostSets(ctx, nil, 1, filterFunc, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing grants hash") + }) + t.Run("zero page size", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s host.Set) (bool, error) { + return true, nil + } + _, err := static.ListHostSets(ctx, []byte("some hash"), 0, filterFunc, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "page size must be at least 1") + }) + t.Run("negative page size", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s host.Set) (bool, error) { + return true, nil + } + _, err := static.ListHostSets(ctx, []byte("some hash"), -1, filterFunc, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "page size must be at least 1") + }) + t.Run("nil filter func", func(t *testing.T) { + t.Parallel() + _, err := static.ListHostSets(ctx, []byte("some hash"), 1, nil, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing filter item callback") + }) + t.Run("nil repo", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s host.Set) (bool, error) { + return true, nil + } + _, err := static.ListHostSets(ctx, []byte("some hash"), 1, filterFunc, nil, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing repo") + }) + t.Run("missing host catalog ID", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s host.Set) (bool, error) { + return true, nil + } + _, err := static.ListHostSets(ctx, []byte("some hash"), 1, filterFunc, repo, "") + require.ErrorContains(t, err, "missing host catalog ID") + }) + }) + t.Run("ListHostSetsPage validation", func(t *testing.T) { + t.Parallel() + t.Run("missing grants hash", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s host.Set) (bool, error) { + return true, nil + } + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, err = static.ListHostSetsPage(ctx, nil, 1, filterFunc, tok, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing grants hash") + }) + t.Run("zero page size", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s host.Set) (bool, error) { + return true, nil + } + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, err = static.ListHostSetsPage(ctx, []byte("some hash"), 0, filterFunc, tok, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "page size must be at least 1") + }) + t.Run("negative page size", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s host.Set) (bool, error) { + return true, nil + } + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, err = static.ListHostSetsPage(ctx, []byte("some hash"), -1, filterFunc, tok, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "page size must be at least 1") + }) + t.Run("nil filter func", func(t *testing.T) { + t.Parallel() + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, err = static.ListHostSetsPage(ctx, []byte("some hash"), 1, nil, tok, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing filter item callback") + }) + t.Run("nil token", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s host.Set) (bool, error) { + return true, nil + } + _, err = static.ListHostSetsPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing token") + }) + t.Run("wrong token type", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s host.Set) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, err = static.ListHostSetsPage(ctx, []byte("some hash"), 1, filterFunc, tok, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "token did not have a pagination token component") + }) + t.Run("nil repo", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s host.Set) (bool, error) { + return true, nil + } + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, err = static.ListHostSetsPage(ctx, []byte("some hash"), 1, filterFunc, tok, nil, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing repo") + }) + t.Run("missing host catalog ID", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s host.Set) (bool, error) { + return true, nil + } + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, err = static.ListHostSetsPage(ctx, []byte("some hash"), 1, filterFunc, tok, repo, "") + require.ErrorContains(t, err, "missing host catalog ID") + }) + t.Run("wrong token resource type", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s host.Set) (bool, error) { + return true, nil + } + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, err = static.ListHostSetsPage(ctx, []byte("some hash"), 1, filterFunc, tok, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "token did not have a host set resource type") + }) + }) + t.Run("ListHostSetsRefresh validation", func(t *testing.T) { + t.Parallel() + t.Run("missing grants hash", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s host.Set) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, err = static.ListHostSetsRefresh(ctx, nil, 1, filterFunc, tok, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing grants hash") + }) + t.Run("zero page size", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s host.Set) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, err = static.ListHostSetsRefresh(ctx, []byte("some hash"), 0, filterFunc, tok, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "page size must be at least 1") + }) + t.Run("negative page size", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s host.Set) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, err = static.ListHostSetsRefresh(ctx, []byte("some hash"), -1, filterFunc, tok, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "page size must be at least 1") + }) + t.Run("nil filter func", func(t *testing.T) { + t.Parallel() + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, err = static.ListHostSetsRefresh(ctx, []byte("some hash"), 1, nil, tok, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing filter item callback") + }) + t.Run("nil token", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s host.Set) (bool, error) { + return true, nil + } + _, err = static.ListHostSetsRefresh(ctx, []byte("some hash"), 1, filterFunc, nil, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing token") + }) + t.Run("nil repo", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s host.Set) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, err = static.ListHostSetsRefresh(ctx, []byte("some hash"), 1, filterFunc, tok, nil, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing repo") + }) + t.Run("missing host catalog ID", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s host.Set) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, err = static.ListHostSetsRefresh(ctx, []byte("some hash"), 1, filterFunc, tok, repo, "") + require.ErrorContains(t, err, "missing host catalog ID") + }) + t.Run("wrong token resource type", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s host.Set) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, err = static.ListHostSetsRefresh(ctx, []byte("some hash"), 1, filterFunc, tok, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "token did not have a host set resource type") + }) + }) + t.Run("ListHostSetsRefreshPage validation", func(t *testing.T) { + t.Parallel() + t.Run("missing grants hash", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s host.Set) (bool, error) { + return true, nil + } + tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo) + require.NoError(t, err) + _, err = static.ListHostSetsRefreshPage(ctx, nil, 1, filterFunc, tok, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing grants hash") + }) + t.Run("zero page size", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s host.Set) (bool, error) { + return true, nil + } + tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo) + require.NoError(t, err) + _, err = static.ListHostSetsRefreshPage(ctx, []byte("some hash"), 0, filterFunc, tok, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "page size must be at least 1") + }) + t.Run("negative page size", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s host.Set) (bool, error) { + return true, nil + } + tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo) + require.NoError(t, err) + _, err = static.ListHostSetsRefreshPage(ctx, []byte("some hash"), -1, filterFunc, tok, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "page size must be at least 1") + }) + t.Run("nil filter func", func(t *testing.T) { + t.Parallel() + tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo) + require.NoError(t, err) + _, err = static.ListHostSetsRefreshPage(ctx, []byte("some hash"), 1, nil, tok, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing filter item callback") + }) + t.Run("nil token", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s host.Set) (bool, error) { + return true, nil + } + _, err = static.ListHostSetsRefreshPage(ctx, []byte("some hash"), 1, filterFunc, nil, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing token") + }) + t.Run("wrong token type", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s host.Set) (bool, error) { + return true, nil + } + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, err = static.ListHostSetsRefreshPage(ctx, []byte("some hash"), 1, filterFunc, tok, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "token did not have a refresh token component") + }) + t.Run("nil repo", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s host.Set) (bool, error) { + return true, nil + } + tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo) + require.NoError(t, err) + _, err = static.ListHostSetsRefreshPage(ctx, []byte("some hash"), 1, filterFunc, tok, nil, catalog.GetPublicId()) + require.ErrorContains(t, err, "missing repo") + }) + t.Run("missing credential store id", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s host.Set) (bool, error) { + return true, nil + } + tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo) + require.NoError(t, err) + _, err = static.ListHostSetsRefreshPage(ctx, []byte("some hash"), 1, filterFunc, tok, repo, "") + require.ErrorContains(t, err, "missing host catalog ID") + }) + t.Run("wrong token resource type", func(t *testing.T) { + t.Parallel() + filterFunc := func(_ context.Context, s host.Set) (bool, error) { + return true, nil + } + tok, err := listtoken.NewRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo, fiveDaysAgo, "some other id", fiveDaysAgo) + require.NoError(t, err) + _, err = static.ListHostSetsRefreshPage(ctx, []byte("some hash"), 1, filterFunc, tok, repo, catalog.GetPublicId()) + require.ErrorContains(t, err, "token did not have a host set resource type") + }) + }) + + t.Run("simple pagination", func(t *testing.T) { + filterFunc := func(context.Context, host.Set) (bool, error) { + return true, nil + } + resp, err := static.ListHostSets(ctx, []byte("some hash"), 1, filterFunc, repo, catalog.GetPublicId()) + require.NoError(t, err) + require.NotNil(t, resp.ListToken) + require.Equal(t, resp.ListToken.GrantsHash, []byte("some hash")) + require.False(t, resp.CompleteListing) + require.Equal(t, resp.EstimatedItemCount, 5) + require.Empty(t, resp.DeletedIds) + require.Len(t, resp.Items, 1) + require.Empty(t, cmp.Diff(resp.Items[0], sets[0], cmpOpts...)) + + resp2, err := static.ListHostSetsPage(ctx, []byte("some hash"), 1, filterFunc, resp.ListToken, repo, catalog.GetPublicId()) + require.NoError(t, err) + require.Equal(t, resp2.ListToken.GrantsHash, []byte("some hash")) + require.False(t, resp2.CompleteListing) + require.Equal(t, resp2.EstimatedItemCount, 5) + require.Empty(t, resp2.DeletedIds) + require.Len(t, resp2.Items, 1) + require.Empty(t, cmp.Diff(resp2.Items[0], sets[1], cmpOpts...)) + + resp3, err := static.ListHostSetsPage(ctx, []byte("some hash"), 1, filterFunc, resp2.ListToken, repo, catalog.GetPublicId()) + require.NoError(t, err) + require.Equal(t, resp3.ListToken.GrantsHash, []byte("some hash")) + require.False(t, resp3.CompleteListing) + require.Equal(t, resp3.EstimatedItemCount, 5) + require.Empty(t, resp3.DeletedIds) + require.Len(t, resp3.Items, 1) + require.Empty(t, cmp.Diff(resp3.Items[0], sets[2], cmpOpts...)) + + resp4, err := static.ListHostSetsPage(ctx, []byte("some hash"), 1, filterFunc, resp3.ListToken, repo, catalog.GetPublicId()) + require.NoError(t, err) + require.Equal(t, resp4.ListToken.GrantsHash, []byte("some hash")) + require.False(t, resp4.CompleteListing) + require.Equal(t, resp4.EstimatedItemCount, 5) + require.Empty(t, resp4.DeletedIds) + require.Len(t, resp4.Items, 1) + require.Empty(t, cmp.Diff(resp4.Items[0], sets[3], cmpOpts...)) + + resp5, err := static.ListHostSetsPage(ctx, []byte("some hash"), 1, filterFunc, resp4.ListToken, repo, catalog.GetPublicId()) + require.NoError(t, err) + require.Equal(t, resp5.ListToken.GrantsHash, []byte("some hash")) + require.True(t, resp5.CompleteListing) + require.Equal(t, resp5.EstimatedItemCount, 5) + require.Empty(t, resp5.DeletedIds) + require.Len(t, resp5.Items, 1) + require.Empty(t, cmp.Diff(resp5.Items[0], sets[4], cmpOpts...)) + + // Finished initial pagination phase, request refresh + // Expect no results. + resp6, err := static.ListHostSetsRefresh(ctx, []byte("some hash"), 1, filterFunc, resp5.ListToken, repo, catalog.GetPublicId()) + require.NoError(t, err) + require.Equal(t, resp6.ListToken.GrantsHash, []byte("some hash")) + require.True(t, resp6.CompleteListing) + require.Equal(t, resp6.EstimatedItemCount, 5) + require.Empty(t, resp6.DeletedIds) + require.Empty(t, resp6.Items) + + // Create some new hosts + set1 := static.TestSets(t, conn, catalog.GetPublicId(), 1)[0] + set2 := static.TestSets(t, conn, catalog.GetPublicId(), 1)[0] + t.Cleanup(func() { + _, err := repo.DeleteSet(ctx, proj1.GetPublicId(), set1.GetPublicId()) + require.NoError(t, err) + _, err = repo.DeleteSet(ctx, proj1.GetPublicId(), set2.GetPublicId()) + require.NoError(t, err) + // Run analyze to update count estimate + _, err = sqlDb.ExecContext(ctx, "analyze") + require.NoError(t, err) + }) + + // Run analyze to update count estimate + _, err = sqlDb.ExecContext(ctx, "analyze") + require.NoError(t, err) + + // Refresh again, should get set2 + resp7, err := static.ListHostSetsRefresh(ctx, []byte("some hash"), 1, filterFunc, resp6.ListToken, repo, catalog.GetPublicId()) + require.NoError(t, err) + require.Equal(t, resp7.ListToken.GrantsHash, []byte("some hash")) + require.False(t, resp7.CompleteListing) + require.Equal(t, resp7.EstimatedItemCount, 7) + require.Empty(t, resp7.DeletedIds) + require.Len(t, resp7.Items, 1) + require.Empty(t, cmp.Diff(resp7.Items[0], set2, cmpOpts...)) + + // Refresh again, should get set1 + resp8, err := static.ListHostSetsRefreshPage(ctx, []byte("some hash"), 1, filterFunc, resp7.ListToken, repo, catalog.GetPublicId()) + require.NoError(t, err) + require.Equal(t, resp8.ListToken.GrantsHash, []byte("some hash")) + require.True(t, resp8.CompleteListing) + require.Equal(t, resp8.EstimatedItemCount, 7) + require.Empty(t, resp8.DeletedIds) + require.Len(t, resp8.Items, 1) + require.Empty(t, cmp.Diff(resp8.Items[0], set1, cmpOpts...)) + + // Refresh again, should get no results + resp9, err := static.ListHostSetsRefresh(ctx, []byte("some hash"), 1, filterFunc, resp8.ListToken, repo, catalog.GetPublicId()) + require.NoError(t, err) + require.Equal(t, resp9.ListToken.GrantsHash, []byte("some hash")) + require.True(t, resp9.CompleteListing) + require.Equal(t, resp9.EstimatedItemCount, 7) + require.Empty(t, resp9.DeletedIds) + require.Empty(t, resp9.Items) + }) + + t.Run("simple pagination with aggressive filtering", func(t *testing.T) { + filterFunc := func(ctx context.Context, s host.Set) (bool, error) { + return s.GetPublicId() == sets[1].GetPublicId() || + s.GetPublicId() == sets[len(sets)-1].GetPublicId(), nil + } + resp, err := static.ListHostSets(ctx, []byte("some hash"), 1, filterFunc, repo, catalog.GetPublicId()) + require.NoError(t, err) + require.NotNil(t, resp.ListToken) + require.Equal(t, resp.ListToken.GrantsHash, []byte("some hash")) + require.False(t, resp.CompleteListing) + require.Equal(t, resp.EstimatedItemCount, 5) + require.Empty(t, resp.DeletedIds) + require.Len(t, resp.Items, 1) + require.Empty(t, cmp.Diff(resp.Items[0], sets[1], cmpOpts...)) + + resp2, err := static.ListHostSetsPage(ctx, []byte("some hash"), 1, filterFunc, resp.ListToken, repo, catalog.GetPublicId()) + require.NoError(t, err) + require.NotNil(t, resp2.ListToken) + require.Equal(t, resp2.ListToken.GrantsHash, []byte("some hash")) + require.True(t, resp2.CompleteListing) + require.Equal(t, resp2.EstimatedItemCount, 5) + require.Empty(t, resp2.DeletedIds) + require.Len(t, resp2.Items, 1) + require.Empty(t, cmp.Diff(resp2.Items[0], sets[len(sets)-1], cmpOpts...)) + + // request a refresh, nothing should be returned + resp3, err := static.ListHostSetsRefresh(ctx, []byte("some hash"), 1, filterFunc, resp.ListToken, repo, catalog.GetPublicId()) + require.NoError(t, err) + require.Equal(t, resp3.ListToken.GrantsHash, []byte("some hash")) + require.True(t, resp3.CompleteListing) + require.Equal(t, resp3.EstimatedItemCount, 5) + require.Empty(t, resp3.DeletedIds) + require.Empty(t, resp3.Items) + + // Create some new hosts + set1 := static.TestSets(t, conn, catalog.GetPublicId(), 1)[0] + set2 := static.TestSets(t, conn, catalog.GetPublicId(), 1)[0] + set3 := static.TestSets(t, conn, catalog.GetPublicId(), 1)[0] + set4 := static.TestSets(t, conn, catalog.GetPublicId(), 1)[0] + t.Cleanup(func() { + _, err := repo.DeleteSet(ctx, proj1.GetPublicId(), set1.GetPublicId()) + require.NoError(t, err) + _, err = repo.DeleteSet(ctx, proj1.GetPublicId(), set2.GetPublicId()) + require.NoError(t, err) + _, err = repo.DeleteSet(ctx, proj1.GetPublicId(), set3.GetPublicId()) + require.NoError(t, err) + _, err = repo.DeleteSet(ctx, proj1.GetPublicId(), set4.GetPublicId()) + require.NoError(t, err) + // Run analyze to update count estimate + _, err = sqlDb.ExecContext(ctx, "analyze") + require.NoError(t, err) + }) + + // Run analyze to update count estimate + _, err = sqlDb.ExecContext(ctx, "analyze") + require.NoError(t, err) + + filterFunc = func(_ context.Context, s host.Set) (bool, error) { + return s.GetPublicId() == set3.GetPublicId() || + s.GetPublicId() == set1.GetPublicId(), nil + } + // Refresh again, should get host3 + resp4, err := static.ListHostSetsRefresh(ctx, []byte("some hash"), 1, filterFunc, resp3.ListToken, repo, catalog.GetPublicId()) + require.NoError(t, err) + require.Equal(t, resp4.ListToken.GrantsHash, []byte("some hash")) + require.False(t, resp4.CompleteListing) + require.Equal(t, resp4.EstimatedItemCount, 9) + require.Empty(t, resp4.DeletedIds) + require.Len(t, resp4.Items, 1) + require.Empty(t, cmp.Diff(resp4.Items[0], set3, cmpOpts...)) + + // Refresh again, should get host1 + resp5, err := static.ListHostSetsRefreshPage(ctx, []byte("some hash"), 1, filterFunc, resp4.ListToken, repo, catalog.GetPublicId()) + require.NoError(t, err) + require.Equal(t, resp5.ListToken.GrantsHash, []byte("some hash")) + require.True(t, resp5.CompleteListing) + require.Equal(t, resp5.EstimatedItemCount, 9) + require.Empty(t, resp5.DeletedIds) + require.Len(t, resp5.Items, 1) + require.Empty(t, cmp.Diff(resp5.Items[0], set1, cmpOpts...)) + }) + + t.Run("simple pagination with deletion", func(t *testing.T) { + filterFunc := func(context.Context, host.Set) (bool, error) { + return true, nil + } + deletedSetId := sets[0].GetPublicId() + _, err := repo.DeleteSet(ctx, proj1.PublicId, deletedSetId) + require.NoError(t, err) + sets = sets[1:] + + // Run analyze to update count estimate + _, err = sqlDb.ExecContext(ctx, "analyze") + require.NoError(t, err) + + resp, err := static.ListHostSets(ctx, []byte("some hash"), 1, filterFunc, repo, catalog.GetPublicId()) + require.NoError(t, err) + require.NotNil(t, resp.ListToken) + require.Equal(t, resp.ListToken.GrantsHash, []byte("some hash")) + require.False(t, resp.CompleteListing) + require.Equal(t, resp.EstimatedItemCount, 4) + require.Empty(t, resp.DeletedIds) + require.Len(t, resp.Items, 1) + require.Empty(t, cmp.Diff(resp.Items[0], sets[0], cmpOpts...)) + + // request remaining results + resp2, err := static.ListHostSetsPage(ctx, []byte("some hash"), 3, filterFunc, resp.ListToken, repo, catalog.GetPublicId()) + require.NoError(t, err) + require.Equal(t, resp2.ListToken.GrantsHash, []byte("some hash")) + require.True(t, resp2.CompleteListing) + require.Equal(t, resp2.EstimatedItemCount, 4) + require.Empty(t, resp2.DeletedIds) + require.Len(t, resp2.Items, 3) + require.Empty(t, cmp.Diff(resp2.Items, sets[1:], cmpOpts...)) + + deletedSetId = sets[0].GetPublicId() + _, err = repo.DeleteSet(ctx, proj1.PublicId, deletedSetId) + require.NoError(t, err) + sets = sets[1:] + + // Run analyze to update count estimate + _, err = sqlDb.ExecContext(ctx, "analyze") + require.NoError(t, err) + + // request a refresh, nothing should be returned except the deleted id + resp3, err := static.ListHostSetsRefresh(ctx, []byte("some hash"), 1, filterFunc, resp2.ListToken, repo, catalog.GetPublicId()) + require.NoError(t, err) + require.Equal(t, resp3.ListToken.GrantsHash, []byte("some hash")) + require.True(t, resp3.CompleteListing) + require.Equal(t, resp3.EstimatedItemCount, 3) + require.Contains(t, resp3.DeletedIds, deletedSetId) + require.Empty(t, resp3.Items) + }) +} diff --git a/internal/host/static/service_list_host_sets_page.go b/internal/host/static/service_list_host_sets_page.go new file mode 100644 index 000000000..9bc8d9896 --- /dev/null +++ b/internal/host/static/service_list_host_sets_page.go @@ -0,0 +1,79 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: BUSL-1.1 + +package static + +import ( + "context" + "time" + + "github.com/hashicorp/boundary/internal/errors" + "github.com/hashicorp/boundary/internal/host" + "github.com/hashicorp/boundary/internal/listtoken" + "github.com/hashicorp/boundary/internal/pagination" + "github.com/hashicorp/boundary/internal/types/resource" +) + +// ListHostSetsPage lists up to page size host sets, filtering out entries that +// do not pass the filter item function. It will automatically request +// more host sets from the database, at page size chunks, to fill the page. +// It will start its paging based on the information in the token. +// It returns a new list token used to continue pagination or refresh items. +// Host sets are ordered by create time descending (most recently created first). +func ListHostSetsPage( + ctx context.Context, + grantsHash []byte, + pageSize int, + filterItemFn pagination.ListFilterFunc[host.Set], + tok *listtoken.Token, + repo *Repository, + hostCatalogId string, +) (*pagination.ListResponse[host.Set], error) { + const op = "plugin.ListHostSetsPage" + + switch { + case len(grantsHash) == 0: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing grants hash") + case pageSize < 1: + return nil, errors.New(ctx, errors.InvalidParameter, op, "page size must be at least 1") + case filterItemFn == nil: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing filter item callback") + case tok == nil: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing token") + case repo == nil: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing repo") + case hostCatalogId == "": + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing host catalog ID") + case tok.ResourceType != resource.HostSet: + return nil, errors.New(ctx, errors.InvalidParameter, op, "token did not have a host set resource type") + } + if _, ok := tok.Subtype.(*listtoken.PaginationToken); !ok { + return nil, errors.New(ctx, errors.InvalidParameter, op, "token did not have a pagination token component") + } + + listItemsFn := func(ctx context.Context, lastPageItem host.Set, limit int) ([]host.Set, time.Time, error) { + opts := []Option{ + WithLimit(limit), + } + if lastPageItem != nil { + opts = append(opts, WithStartPageAfterItem(lastPageItem)) + } else { + lastItem, err := tok.LastItem(ctx) + if err != nil { + return nil, time.Time{}, err + } + opts = append(opts, WithStartPageAfterItem(lastItem)) + } + sSets, listTime, err := repo.listSets(ctx, hostCatalogId, opts...) + if err != nil { + return nil, time.Time{}, err + } + var sets []host.Set + for _, set := range sSets { + sets = append(sets, set) + } + return sets, listTime, nil + } + + return pagination.ListPage(ctx, grantsHash, pageSize, filterItemFn, listItemsFn, repo.estimatedSetCount, tok) +} diff --git a/internal/host/static/service_list_host_sets_refresh.go b/internal/host/static/service_list_host_sets_refresh.go new file mode 100644 index 000000000..51aea5c37 --- /dev/null +++ b/internal/host/static/service_list_host_sets_refresh.go @@ -0,0 +1,81 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: BUSL-1.1 + +package static + +import ( + "context" + "time" + + "github.com/hashicorp/boundary/globals" + "github.com/hashicorp/boundary/internal/errors" + "github.com/hashicorp/boundary/internal/host" + "github.com/hashicorp/boundary/internal/listtoken" + "github.com/hashicorp/boundary/internal/pagination" + "github.com/hashicorp/boundary/internal/types/resource" +) + +// ListHostSetsRefresh lists host sets according to the page size +// and list token, filtering out entries that do not +// pass the filter item fn. It returns a new list token +// based on the old one, the grants hash, and the returned +// host sets. It also returns the plugin associated with the host catalog. +func ListHostSetsRefresh( + ctx context.Context, + grantsHash []byte, + pageSize int, + filterItemFn pagination.ListFilterFunc[host.Set], + tok *listtoken.Token, + repo *Repository, + hostCatalogId string, +) (*pagination.ListResponse[host.Set], error) { + const op = "plugin.ListHostSetsRefresh" + + switch { + case len(grantsHash) == 0: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing grants hash") + case pageSize < 1: + return nil, errors.New(ctx, errors.InvalidParameter, op, "page size must be at least 1") + case filterItemFn == nil: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing filter item callback") + case tok == nil: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing token") + case repo == nil: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing repo") + case hostCatalogId == "": + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing host catalog ID") + case tok.ResourceType != resource.HostSet: + return nil, errors.New(ctx, errors.InvalidParameter, op, "token did not have a host set resource type") + } + rt, ok := tok.Subtype.(*listtoken.StartRefreshToken) + if !ok { + return nil, errors.New(ctx, errors.InvalidParameter, op, "token did not have a start-refresh token component") + } + + listItemsFn := func(ctx context.Context, lastPageItem host.Set, limit int) ([]host.Set, time.Time, error) { + opts := []Option{ + WithLimit(limit), + } + if lastPageItem != nil { + opts = append(opts, WithStartPageAfterItem(lastPageItem)) + } + // Add the database read timeout to account for any creations missed due to concurrent + // transactions in the initial pagination phase. + sSets, listTime, err := repo.listSetsRefresh(ctx, hostCatalogId, rt.PreviousPhaseUpperBound.Add(-globals.RefreshReadLookbackDuration), opts...) + if err != nil { + return nil, time.Time{}, err + } + var sets []host.Set + for _, set := range sSets { + sets = append(sets, set) + } + return sets, listTime, nil + } + listDeletedIdsFn := func(ctx context.Context, since time.Time) ([]string, time.Time, error) { + // Add the database read timeout to account for any deletes missed due to concurrent + // transactions in the original list pagination phase. + return repo.listDeletedSetIds(ctx, since.Add(-globals.RefreshReadLookbackDuration)) + } + + return pagination.ListRefresh(ctx, grantsHash, pageSize, filterItemFn, listItemsFn, repo.estimatedSetCount, listDeletedIdsFn, tok) +} diff --git a/internal/host/static/service_list_host_sets_refresh_page.go b/internal/host/static/service_list_host_sets_refresh_page.go new file mode 100644 index 000000000..36b4b1d99 --- /dev/null +++ b/internal/host/static/service_list_host_sets_refresh_page.go @@ -0,0 +1,91 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: BUSL-1.1 + +package static + +import ( + "context" + "time" + + "github.com/hashicorp/boundary/globals" + "github.com/hashicorp/boundary/internal/errors" + "github.com/hashicorp/boundary/internal/host" + "github.com/hashicorp/boundary/internal/listtoken" + "github.com/hashicorp/boundary/internal/pagination" + "github.com/hashicorp/boundary/internal/types/resource" +) + +// ListHostSetsRefreshPage lists up to page size host sets, filtering out entries that +// do not pass the filter item function. It will automatically request +// more host sets from the database, at page size chunks, to fill the page. +// It will start its paging based on the information in the token. +// It returns a new list token used to continue pagination or refresh items. +// Host sets are ordered by update time descending (most recently updated first). +// Host sets may contain items that were already returned during the initial +// pagination phase. It also returns a list of any host sets deleted since the +// last response. +func ListHostSetsRefreshPage( + ctx context.Context, + grantsHash []byte, + pageSize int, + filterItemFn pagination.ListFilterFunc[host.Set], + tok *listtoken.Token, + repo *Repository, + hostCatalogId string, +) (*pagination.ListResponse[host.Set], error) { + const op = "credential.ListHostSetsRefreshPage" + + switch { + case len(grantsHash) == 0: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing grants hash") + case pageSize < 1: + return nil, errors.New(ctx, errors.InvalidParameter, op, "page size must be at least 1") + case filterItemFn == nil: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing filter item callback") + case tok == nil: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing token") + case repo == nil: + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing repo") + case hostCatalogId == "": + return nil, errors.New(ctx, errors.InvalidParameter, op, "missing host catalog ID") + case tok.ResourceType != resource.HostSet: + return nil, errors.New(ctx, errors.InvalidParameter, op, "token did not have a host set resource type") + } + rt, ok := tok.Subtype.(*listtoken.RefreshToken) + if !ok { + return nil, errors.New(ctx, errors.InvalidParameter, op, "token did not have a refresh token component") + } + + listItemsFn := func(ctx context.Context, lastPageItem host.Set, limit int) ([]host.Set, time.Time, error) { + opts := []Option{ + WithLimit(limit), + } + if lastPageItem != nil { + opts = append(opts, WithStartPageAfterItem(lastPageItem)) + } else { + lastItem, err := tok.LastItem(ctx) + if err != nil { + return nil, time.Time{}, err + } + opts = append(opts, WithStartPageAfterItem(lastItem)) + } + // Add the database read timeout to account for any creations missed due to concurrent + // transactions in the original list pagination phase. + sSets, listTime, err := repo.listSetsRefresh(ctx, hostCatalogId, rt.PhaseLowerBound.Add(-globals.RefreshReadLookbackDuration), opts...) + if err != nil { + return nil, time.Time{}, err + } + var sets []host.Set + for _, set := range sSets { + sets = append(sets, set) + } + return sets, listTime, nil + } + listDeletedIdsFn := func(ctx context.Context, since time.Time) ([]string, time.Time, error) { + // Add the database read timeout to account for any deletes missed due to concurrent + // transactions in the original list pagination phase. + return repo.listDeletedSetIds(ctx, since.Add(-globals.RefreshReadLookbackDuration)) + } + + return pagination.ListRefreshPage(ctx, grantsHash, pageSize, filterItemFn, listItemsFn, repo.estimatedSetCount, listDeletedIdsFn, tok) +}