diff --git a/internal/host/plugin/host_set.go b/internal/host/plugin/host_set.go index 252c23c71..3ca674626 100644 --- a/internal/host/plugin/host_set.go +++ b/internal/host/plugin/host_set.go @@ -191,3 +191,13 @@ func (agg *hostSetAgg) GetPublicId() string { func (agg *hostSetAgg) TableName() string { return "host_plugin_host_set_with_value_obj" } + +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 "host_plugin_set_deleted" +} diff --git a/internal/host/plugin/query.go b/internal/host/plugin/query.go index 9132338b1..afbf86abb 100644 --- a/internal/host/plugin/query.go +++ b/internal/host/plugin/query.go @@ -341,5 +341,265 @@ 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 ('host_plugin_set'::regclass) +` + + listSetsTemplate = ` +with host_sets as ( + select public_id, + catalog_id, + name, + description, + create_time, + update_time, + last_sync_time, + need_sync, + sync_interval_seconds, + version, + attributes + from host_plugin_set + where catalog_id = @catalog_id + order by create_time desc, public_id asc + limit %d +), +host_catalog as ( + select plugin_id, + public_id + from host_plugin_catalog + where public_id = @catalog_id +), +preferred_endpoints as ( + select string_agg(distinct concat_ws('=', priority, condition), '|') as preferred_endpoints, + host_set_id + from host_set_preferred_endpoint + where host_set_id in (select public_id from host_sets) + group by host_set_id +), +host_set_members as ( + select string_agg(distinct host_id, '|') as host_ids, + set_id + from host_plugin_set_member + where set_id in (select public_id from host_sets) + group by set_id +), +final as ( + select hs.public_id, + hs.catalog_id, + hc.plugin_id, + hs.name, + hs.description, + hs.create_time, + hs.update_time, + hs.last_sync_time, + hs.need_sync, + hs.sync_interval_seconds, + hs.version, + hs.attributes, + ppe.preferred_endpoints, + hsm.host_ids + from host_sets hs + join host_catalog hc on hc.public_id = hs.catalog_id + left outer join preferred_endpoints ppe on ppe.host_set_id = hs.public_id + left outer join host_set_members hsm on hsm.set_id = hs.public_id +) + select * + from final +order by create_time desc, public_id asc; +` + + listSetsPageTemplate = ` +with host_sets as ( + select public_id, + catalog_id, + name, + description, + create_time, + update_time, + last_sync_time, + need_sync, + sync_interval_seconds, + version, + attributes + from host_plugin_set + where catalog_id = @catalog_id + and (create_time, public_id) < (@last_item_create_time, @last_item_id) + order by create_time desc, public_id asc + limit %d +), +host_catalog as ( + select plugin_id, + public_id + from host_plugin_catalog + where public_id = @catalog_id +), +preferred_endpoints as ( + select string_agg(distinct concat_ws('=', priority, condition), '|') as preferred_endpoints, + host_set_id + from host_set_preferred_endpoint + where host_set_id in (select public_id from host_sets) + group by host_set_id +), +host_set_members as ( + select string_agg(distinct host_id, '|') as host_ids, + set_id + from host_plugin_set_member + where set_id in (select public_id from host_sets) + group by set_id +), +final as ( + select hs.public_id, + hs.catalog_id, + hc.plugin_id, + hs.name, + hs.description, + hs.create_time, + hs.update_time, + hs.last_sync_time, + hs.need_sync, + hs.sync_interval_seconds, + hs.version, + hs.attributes, + ppe.preferred_endpoints, + hsm.host_ids + from host_sets hs + join host_catalog hc on hc.public_id = hs.catalog_id + left outer join preferred_endpoints ppe on ppe.host_set_id = hs.public_id + left outer join host_set_members hsm on hsm.set_id = hs.public_id +) + select * + from final +order by create_time desc, public_id asc; +` + + listSetsRefreshTemplate = ` +with host_sets as ( + select public_id, + catalog_id, + name, + description, + create_time, + update_time, + last_sync_time, + need_sync, + sync_interval_seconds, + version, + attributes + from host_plugin_set + where catalog_id = @catalog_id + and update_time > @updated_after_time + order by update_time desc, public_id asc + limit %d +), +host_catalog as ( + select plugin_id, + public_id + from host_plugin_catalog + where public_id = @catalog_id +), +preferred_endpoints as ( + select string_agg(distinct concat_ws('=', priority, condition), '|') as preferred_endpoints, + host_set_id + from host_set_preferred_endpoint + where host_set_id in (select public_id from host_sets) + group by host_set_id +), +host_set_members as ( + select string_agg(distinct host_id, '|') as host_ids, + set_id + from host_plugin_set_member + where set_id in (select public_id from host_sets) + group by set_id +), +final as ( + select hs.public_id, + hs.catalog_id, + hc.plugin_id, + hs.name, + hs.description, + hs.create_time, + hs.update_time, + hs.last_sync_time, + hs.need_sync, + hs.sync_interval_seconds, + hs.version, + hs.attributes, + ppe.preferred_endpoints, + hsm.host_ids + from host_sets hs + join host_catalog hc on hc.public_id = hs.catalog_id + left outer join preferred_endpoints ppe on ppe.host_set_id = hs.public_id + left outer join host_set_members hsm on hsm.set_id = hs.public_id +) + select * + from final +order by update_time desc, public_id asc; +` + + listSetsRefreshPageTemplate = ` +with host_sets as ( + select public_id, + catalog_id, + name, + description, + create_time, + update_time, + last_sync_time, + need_sync, + sync_interval_seconds, + version, + attributes + from host_plugin_set + where catalog_id = @catalog_id + and update_time > @updated_after_time + and (update_time, public_id) < (@last_item_update_time, @last_item_id) + order by update_time desc, public_id asc + limit %d +), +host_catalog as ( + select plugin_id, + public_id + from host_plugin_catalog + where public_id = @catalog_id +), +preferred_endpoints as ( + select string_agg(distinct concat_ws('=', priority, condition), '|') as preferred_endpoints, + host_set_id + from host_set_preferred_endpoint + where host_set_id in (select public_id from host_sets) + group by host_set_id +), +host_set_members as ( + select string_agg(distinct host_id, '|') as host_ids, + set_id + from host_plugin_set_member + where set_id in (select public_id from host_sets) + group by set_id +), +final as ( + select hs.public_id, + hs.catalog_id, + hc.plugin_id, + hs.name, + hs.description, + hs.create_time, + hs.update_time, + hs.last_sync_time, + hs.need_sync, + hs.sync_interval_seconds, + hs.version, + hs.attributes, + ppe.preferred_endpoints, + hsm.host_ids + from host_sets hs + join host_catalog hc on hc.public_id = hs.catalog_id + left outer join preferred_endpoints ppe on ppe.host_set_id = hs.public_id + left outer join host_set_members hsm on hsm.set_id = hs.public_id +) + select * + from final +order by update_time desc, public_id asc; ` ) diff --git a/internal/host/plugin/repository_host_set.go b/internal/host/plugin/repository_host_set.go index 51d362823..d170222c5 100644 --- a/internal/host/plugin/repository_host_set.go +++ b/internal/host/plugin/repository_host_set.go @@ -5,6 +5,7 @@ package plugin import ( "context" + "database/sql" "fmt" "strings" "time" @@ -575,7 +576,7 @@ func (r *Repository) UpdateSet(ctx context.Context, projectId string, s *HostSet // LookupSet will look up a host set in the repository and return the host set, // as well as host IDs that match. If the host set is not found, it will return // nil, nil, nil. No options are currently supported. -func (r *Repository) LookupSet(ctx context.Context, publicId string, _ ...host.Option) (*HostSet, *plugin.Plugin, error) { +func (r *Repository) LookupSet(ctx context.Context, publicId string, _ ...Option) (*HostSet, *plugin.Plugin, error) { const op = "plugin.(Repository).LookupSet" if publicId == "" { return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "no public id") @@ -596,19 +597,108 @@ func (r *Repository) LookupSet(ctx context.Context, publicId string, _ ...host.O return sets[0], plg, 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 ...host.Option) ([]*HostSet, *plugin.Plugin, error) { - const op = "plugin.(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, *plugin.Plugin, time.Time, error) { + const op = "plugin.(Repository).listSets" if catalogId == "" { - return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing catalog id") + return nil, nil, time.Time{}, errors.New(ctx, errors.InvalidParameter, op, "missing catalog id") } - sets, plg, err := r.getSets(ctx, "", catalogId, opt...) - if err != nil { - return nil, nil, errors.Wrap(ctx, err, op) + 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 } - return sets, plg, nil + query := fmt.Sprintf(listSetsTemplate, limit) + args := []any{sql.Named("catalog_id", catalogId)} + if opts.withStartPageAfterItem != nil { + query = fmt.Sprintf(listSetsPageTemplate, limit) + args = append(args, + sql.Named("last_item_create_time", opts.withStartPageAfterItem.GetCreateTime()), + sql.Named("last_item_id", opts.withStartPageAfterItem.GetPublicId()), + ) + } + + return r.querySets(ctx, query, args) +} + +// listSetsRefresh returns a slice of Host sets for the catalogId and the associated plugin. +// 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, *plugin.Plugin, time.Time, error) { + const op = "plugin.(Repository).listSetsRefresh" + switch { + case catalogId == "": + return nil, nil, time.Time{}, errors.New(ctx, errors.InvalidParameter, op, "no catalog id") + case updatedAfter.IsZero(): + return nil, nil, time.Time{}, errors.New(ctx, errors.InvalidParameter, op, "missing 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 + } + query := fmt.Sprintf(listSetsRefreshTemplate, limit) + args := []any{ + sql.Named("catalog_id", catalogId), + sql.Named("updated_after_time", updatedAfter), + } + if opts.withStartPageAfterItem != nil { + query = fmt.Sprintf(listSetsRefreshPageTemplate, limit) + args = append(args, + sql.Named("last_item_update_time", opts.withStartPageAfterItem.GetUpdateTime()), + sql.Named("last_item_id", opts.withStartPageAfterItem.GetPublicId()), + ) + } + + return r.querySets(ctx, query, args) +} + +func (r *Repository) querySets(ctx context.Context, query string, args []any) ([]*HostSet, *plugin.Plugin, time.Time, error) { + const op = "plugin.(Repository).querySets" + + var sets []*HostSet + var plg *plugin.Plugin + var transactionTimestamp time.Time + if _, err := r.writer.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{}, func(r db.Reader, w db.Writer) error { + rows, err := r.Query(ctx, query, args) + if err != nil { + return err + } + defer rows.Close() + var foundSets []*hostSetAgg + for rows.Next() { + if err := r.ScanRows(ctx, rows, &foundSets); err != nil { + return err + } + } + if len(foundSets) != 0 { + plg = plugin.NewPlugin() + plg.PublicId = foundSets[0].PluginId + if err := r.LookupByPublicId(ctx, plg); err != nil { + return err + } + sets = make([]*HostSet, 0, len(foundSets)) + for _, ha := range foundSets { + set, err := ha.toHostSet(ctx) + if err != nil { + return err + } + sets = append(sets, set) + } + } + transactionTimestamp, err = r.Now(ctx) + return err + }); err != nil { + return nil, nil, time.Time{}, errors.Wrap(ctx, err, op) + } + return sets, plg, transactionTimestamp, nil } // DeleteSet deletes the host set for the provided id from the repository @@ -686,7 +776,7 @@ func (r *Repository) DeleteSet(ctx context.Context, projectId string, publicId s return rowsDeleted, nil } -func (r *Repository) getSets(ctx context.Context, publicId string, catalogId string, opt ...host.Option) ([]*HostSet, *plugin.Plugin, error) { +func (r *Repository) getSets(ctx context.Context, publicId string, catalogId string, opt ...Option) ([]*HostSet, *plugin.Plugin, error) { const op = "plugin.(Repository).getSets" if publicId == "" && catalogId == "" { return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing search criteria: both host set id and catalog id are empty") @@ -695,35 +785,44 @@ func (r *Repository) getSets(ctx context.Context, publicId string, catalogId str return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "searching for both a host set id and a catalog id is not supported") } - opts, err := host.GetOpts(opt...) - if err != nil { - return nil, nil, errors.Wrap(ctx, err, op) - } + opts := getOpts(opt...) limit := r.defaultLimit - if opts.WithLimit != 0 { + if opts.withLimit != 0 { // non-zero signals an override of the default limit for the repo. - limit = opts.WithLimit + limit = opts.withLimit } - args := make([]any, 0, 1) + var args []any var where string - switch { case publicId != "": - where, args = "public_id = ?", append(args, publicId) + where, args = "public_id = @public_id", append(args, sql.Named("public_id", publicId)) default: - where, args = "catalog_id = ?", append(args, catalogId) + where, args = "catalog_id = @catalog_id", append(args, sql.Named("catalog_id", catalogId)) + } + // Ordering and pagination are tightly coupled. + // We order by update_time ascending so that new + // and updated items appear at the end of the pagination. + // We need to further order by public_id to distinguish items + // with identical update times. + withOrder := "update_time asc, public_id asc" + if opts.withStartPageAfterItem != nil { + // Now that the order is defined, we can use a simple where + // clause to only include items updated since the specified + // start of the page. We use greater than or equal for the update + // time as there may be items with identical update_times. We + // then use public_id as a tiebreaker. + args = append(args, + sql.Named("after_item_update_time", opts.withStartPageAfterItem.GetUpdateTime()), + sql.Named("after_item_id", opts.withStartPageAfterItem.GetPublicId()), + ) + where = "(" + where + ") and (update_time > @after_item_update_time or (update_time = @after_item_update_time and public_id > @after_item_id))" } - dbArgs := []db.Option{db.WithLimit(limit)} - - if opts.WithOrderByCreateTime { - if opts.Ascending { - dbArgs = append(dbArgs, db.WithOrder("create_time asc")) - } else { - dbArgs = append(dbArgs, db.WithOrder("create_time")) - } + dbArgs := []db.Option{ + db.WithLimit(limit), + db.WithOrder(withOrder), } var aggHostSets []*hostSetAgg @@ -746,6 +845,7 @@ func (r *Repository) getSets(ctx context.Context, publicId string, catalogId str } var plg *plugin.Plugin if plgId != "" { + var err error plg, err = r.getPlugin(ctx, plgId) if err != nil { return nil, nil, errors.Wrap(ctx, err, op) @@ -788,6 +888,48 @@ func toPluginSet(ctx context.Context, in *HostSet) (*pb.HostSet, error) { return hs, nil } +// listDeletedSetIds lists the public IDs of any hosts deleted since the timestamp provided, +// and the timestamp of the transaction within which the hosts were listed. +func (r *Repository) listDeletedSetIds(ctx context.Context, since time.Time) ([]string, time.Time, error) { + const op = "static.(Repository).listDeletedHostSetIds" + 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 plugin host sets. +func (r *Repository) estimatedSetCount(ctx context.Context) (int, error) { + const op = "plugin.(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 plugin 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 plugin host sets")) + } + } + return count, nil +} + // Endpoints provides all the endpoints available for a given set id. // An error is returned if the set, related catalog, or related plugin are // unable to be retrieved. If a host does not contain an addressible endpoint diff --git a/internal/host/plugin/repository_host_set_test.go b/internal/host/plugin/repository_host_set_test.go index 8ddfb30be..fab394db9 100644 --- a/internal/host/plugin/repository_host_set_test.go +++ b/internal/host/plugin/repository_host_set_test.go @@ -1556,7 +1556,7 @@ func TestRepository_Endpoints(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) @@ -1583,7 +1583,7 @@ func TestRepository_ListSets(t *testing.T) { tests := []struct { name string in string - opts []host.Option + opts []Option want []*HostSet wantIsErr errors.Code }{ @@ -1610,7 +1610,7 @@ func TestRepository_ListSets(t *testing.T) { repo, err := NewRepository(ctx, rw, rw, kms, sched, plgm) assert.NoError(err) require.NotNil(repo) - got, gotPlg, err := repo.ListSets(ctx, tt.in, tt.opts...) + got, gotPlg, 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) @@ -1625,11 +1625,14 @@ func TestRepository_ListSets(t *testing.T) { if got != nil { assert.Empty(cmp.Diff(plg, gotPlg, protocmp.Transform())) } + // 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) @@ -1652,7 +1655,7 @@ func TestRepository_ListSets_Limits(t *testing.T) { tests := []struct { name string repoOpts []host.Option - listOpts []host.Option + listOpts []Option wantLen int }{ { @@ -1664,31 +1667,21 @@ func TestRepository_ListSets_Limits(t *testing.T) { repoOpts: []host.Option{host.WithLimit(3)}, wantLen: 3, }, - { - name: "With negative repo limit", - repoOpts: []host.Option{host.WithLimit(-1)}, - wantLen: count, - }, { name: "With List limit", - listOpts: []host.Option{host.WithLimit(3)}, + listOpts: []Option{WithLimit(3)}, wantLen: 3, }, - { - name: "With negative List limit", - listOpts: []host.Option{host.WithLimit(-1)}, - wantLen: count, - }, { name: "With repo smaller than list limit", repoOpts: []host.Option{host.WithLimit(2)}, - listOpts: []host.Option{host.WithLimit(6)}, + listOpts: []Option{WithLimit(6)}, wantLen: 6, }, { name: "With repo larger than list limit", repoOpts: []host.Option{host.WithLimit(6)}, - listOpts: []host.Option{host.WithLimit(2)}, + listOpts: []Option{WithLimit(2)}, wantLen: 2, }, } @@ -1700,14 +1693,201 @@ func TestRepository_ListSets_Limits(t *testing.T) { repo, err := NewRepository(ctx, rw, rw, kms, sched, plgm, tt.repoOpts...) assert.NoError(err) require.NotNil(repo) - got, gotPlg, err := repo.ListSets(ctx, hostSets[0].CatalogId, tt.listOpts...) + got, gotPlg, ttime, err := repo.listSets(ctx, hostSets[0].CatalogId, tt.listOpts...) require.NoError(err) assert.Len(got, tt.wantLen) assert.Empty(cmp.Diff(plg, gotPlg, protocmp.Transform())) + // 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_Pagination(t *testing.T) { + t.Parallel() + ctx := context.Background() + conn, _ := db.TestSetup(t, "postgres") + wrapper := db.TestWrapper(t) + kms := kms.TestKms(t, conn, wrapper) + iamRepo := iam.TestRepo(t, conn, wrapper) + _, proj1 := iam.TestScopes(t, iamRepo) + sched := scheduler.TestScheduler(t, conn, wrapper) + plg := plugin.TestPlugin(t, conn, "test") + plgm := map[string]plgpb.HostPluginServiceClient{ + plg.GetPublicId(): &loopback.WrappingPluginHostClient{Server: &loopback.TestPluginServer{}}, + } + catalog := TestCatalog(t, conn, proj1.GetPublicId(), plg.GetPublicId()) + + total := 5 + for i := 0; i < total; i++ { + TestSet(t, conn, kms, sched, catalog, plgm) + } + + rw := db.New(conn) + repo, err := NewRepository(ctx, rw, rw, kms, sched, plgm) + require.NoError(t, err) + + t.Run("no-options", func(t *testing.T) { + got, retPlg, ttime, err := repo.listSets(ctx, catalog.GetPublicId()) + require.NoError(t, err) + assert.Equal(t, total, len(got)) + assert.Equal(t, retPlg, plg) + // 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))) + }) + + t.Run("withStartPageAfter", func(t *testing.T) { + assert, require := assert.New(t), require.New(t) + + page1, retPlg, ttime, err := repo.listSets( + context.Background(), + catalog.GetPublicId(), + WithLimit(2), + ) + require.NoError(err) + require.Len(page1, 2) + assert.Equal(retPlg, plg) + // 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))) + page2, retPlg, ttime, err := repo.listSets( + context.Background(), + catalog.GetPublicId(), + WithLimit(2), + WithStartPageAfterItem(page1[1]), + ) + require.NoError(err) + require.Len(page2, 2) + assert.Equal(retPlg, plg) + for _, item := range page1 { + assert.NotEqual(item.GetPublicId(), page2[0].GetPublicId()) + assert.NotEqual(item.GetPublicId(), page2[1].GetPublicId()) + } + assert.True(time.Now().Before(ttime.Add(10 * time.Second))) + assert.True(time.Now().After(ttime.Add(-10 * time.Second))) + page3, retPlg, ttime, err := repo.listSets( + context.Background(), + catalog.GetPublicId(), + WithLimit(2), + WithStartPageAfterItem(page2[1]), + ) + require.NoError(err) + require.Len(page3, 1) + assert.Equal(retPlg, plg) + for _, item := range page2 { + assert.NotEqual(item.GetPublicId(), page3[0].GetPublicId()) + } + assert.True(time.Now().Before(ttime.Add(10 * time.Second))) + assert.True(time.Now().After(ttime.Add(-10 * time.Second))) + page4, retPlg, ttime, err := repo.listSets( + context.Background(), + catalog.GetPublicId(), + WithLimit(2), + WithStartPageAfterItem(page3[0]), + ) + require.NoError(err) + require.Empty(page4) + require.Empty(retPlg) + 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) + sched := scheduler.TestScheduler(t, conn, wrapper) + plg := plugin.TestPlugin(t, conn, "test") + plgm := map[string]plgpb.HostPluginServiceClient{ + plg.GetPublicId(): &loopback.WrappingPluginHostClient{Server: &loopback.TestPluginServer{}}, + } + catalog := TestCatalog(t, conn, proj.GetPublicId(), plg.GetPublicId()) + + rw := db.New(conn) + repo, err := NewRepository(ctx, rw, rw, testKms, sched, plgm) + 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 := TestSet(t, conn, testKms, sched, catalog, plgm) + _, 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) + sched := scheduler.TestScheduler(t, conn, wrapper) + plg := plugin.TestPlugin(t, conn, "test") + plgm := map[string]plgpb.HostPluginServiceClient{ + plg.GetPublicId(): &loopback.WrappingPluginHostClient{Server: &loopback.TestPluginServer{}}, + } + catalog := TestCatalog(t, conn, proj.GetPublicId(), plg.GetPublicId()) + + rw := db.New(conn) + repo, err := NewRepository(ctx, rw, rw, testKms, sched, plgm) + 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 := TestSet(t, conn, testKms, sched, catalog, plgm) + // 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/plugin/service_list_host_sets.go b/internal/host/plugin/service_list_host_sets.go new file mode 100644 index 000000000..9694fdcb5 --- /dev/null +++ b/internal/host/plugin/service_list_host_sets.go @@ -0,0 +1,64 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: BUSL-1.1 + +package plugin + +import ( + "context" + "time" + + "github.com/hashicorp/boundary/internal/errors" + "github.com/hashicorp/boundary/internal/host" + "github.com/hashicorp/boundary/internal/pagination" + ppagination "github.com/hashicorp/boundary/internal/pagination/plugin" + "github.com/hashicorp/boundary/internal/plugin" +) + +// 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 ppagination.ListPluginFilterFunc[host.Set], + repo *Repository, + hostCatalogId string, +) (*pagination.ListResponse[host.Set], *plugin.Plugin, error) { + const op = "plugin.ListHostSets" + + 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 repo == nil: + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing repo") + case hostCatalogId == "": + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing host catalog ID") + } + + listItemsFn := func(ctx context.Context, lastPageItem host.Set, limit int) ([]host.Set, *plugin.Plugin, time.Time, error) { + opts := []Option{ + WithLimit(limit), + } + if lastPageItem != nil { + opts = append(opts, WithStartPageAfterItem(lastPageItem)) + } + pSets, plg, listTime, err := repo.listSets(ctx, hostCatalogId, opts...) + if err != nil { + return nil, nil, time.Time{}, err + } + var sets []host.Set + for _, set := range pSets { + sets = append(sets, set) + } + return sets, plg, listTime, nil + } + + return ppagination.ListPlugin(ctx, grantsHash, pageSize, filterItemFn, listItemsFn, repo.estimatedSetCount) +} diff --git a/internal/host/plugin/service_list_host_sets_ext_test.go b/internal/host/plugin/service_list_host_sets_ext_test.go new file mode 100644 index 000000000..9cbba357a --- /dev/null +++ b/internal/host/plugin/service_list_host_sets_ext_test.go @@ -0,0 +1,646 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: BUSL-1.1 + +package plugin_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" + hostplugin "github.com/hashicorp/boundary/internal/host/plugin" + "github.com/hashicorp/boundary/internal/host/plugin/store" + "github.com/hashicorp/boundary/internal/iam" + "github.com/hashicorp/boundary/internal/kms" + "github.com/hashicorp/boundary/internal/listtoken" + "github.com/hashicorp/boundary/internal/plugin" + "github.com/hashicorp/boundary/internal/plugin/loopback" + "github.com/hashicorp/boundary/internal/scheduler" + "github.com/hashicorp/boundary/internal/types/resource" + plgpb "github.com/hashicorp/boundary/sdk/pbs/plugin" + "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) + sche := scheduler.TestScheduler(t, conn, wrapper) + iamRepo := iam.TestRepo(t, conn, wrapper) + _, proj1 := iam.TestScopes(t, iamRepo) + plg := plugin.TestPlugin(t, conn, "test") + fiveDaysAgo := time.Now().AddDate(0, 0, -5) + catalog := hostplugin.TestCatalog(t, conn, proj1.GetPublicId(), plg.GetPublicId()) + plgm := map[string]plgpb.HostPluginServiceClient{ + plg.GetPublicId(): loopback.NewWrappingPluginHostClient(&plgpb.UnimplementedHostPluginServiceServer{}), + } + var hostSets []host.Set + for i := 0; i < 5; i++ { + hostSet := hostplugin.TestSet(t, conn, testKms, sche, catalog, plgm) + hostSet.PluginId = plg.GetPublicId() + hostSets = append(hostSets, hostSet) + } + // since we sort by create time descending, we need to reverse the slice + slices.Reverse(hostSets) + + // Run analyze to update host estimate + _, err = sqlDb.ExecContext(ctx, "analyze") + require.NoError(t, err) + + rw := db.New(conn) + repo, err := hostplugin.NewRepository(ctx, rw, rw, testKms, sche, plgm) + require.NoError(t, err) + + cmpOpts := []cmp.Option{ + cmpopts.IgnoreUnexported( + hostplugin.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, plg *plugin.Plugin) (bool, error) { + return true, nil + } + _, _, err := hostplugin.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, plg *plugin.Plugin) (bool, error) { + return true, nil + } + _, _, err := hostplugin.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, plg *plugin.Plugin) (bool, error) { + return true, nil + } + _, _, err := hostplugin.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 := hostplugin.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, plg *plugin.Plugin) (bool, error) { + return true, nil + } + _, _, err := hostplugin.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, plg *plugin.Plugin) (bool, error) { + return true, nil + } + _, _, err := hostplugin.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, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.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, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.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, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.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 = hostplugin.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, plg *plugin.Plugin) (bool, error) { + return true, nil + } + _, _, err = hostplugin.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, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.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, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.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, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.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, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.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, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.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, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.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, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.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 = hostplugin.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, plg *plugin.Plugin) (bool, error) { + return true, nil + } + _, _, err = hostplugin.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, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.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, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.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, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewStartRefresh(ctx, fiveDaysAgo, resource.Target, []byte("some hash"), fiveDaysAgo, fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.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("ListHostsRefreshPage 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, plg *plugin.Plugin) (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 = hostplugin.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, plg *plugin.Plugin) (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 = hostplugin.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, plg *plugin.Plugin) (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 = hostplugin.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 = hostplugin.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, plg *plugin.Plugin) (bool, error) { + return true, nil + } + _, _, err = hostplugin.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, plg *plugin.Plugin) (bool, error) { + return true, nil + } + tok, err := listtoken.NewPagination(ctx, fiveDaysAgo, resource.HostSet, []byte("some hash"), "some-id", fiveDaysAgo) + require.NoError(t, err) + _, _, err = hostplugin.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, plg *plugin.Plugin) (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 = hostplugin.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, plg *plugin.Plugin) (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 = hostplugin.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, plg *plugin.Plugin) (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 = hostplugin.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, *plugin.Plugin) (bool, error) { + return true, nil + } + resp, retPlg, err := hostplugin.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], hostSets[0], cmpOpts...)) + require.Equal(t, retPlg, plg) + + resp2, retPlg, err := hostplugin.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], hostSets[1], cmpOpts...)) + require.Equal(t, retPlg, plg) + + resp3, retPlg, err := hostplugin.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], hostSets[2], cmpOpts...)) + require.Equal(t, retPlg, plg) + + resp4, retPlg, err := hostplugin.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], hostSets[3], cmpOpts...)) + require.Equal(t, retPlg, plg) + + resp5, retPlg, err := hostplugin.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], hostSets[4], cmpOpts...)) + require.Equal(t, retPlg, plg) + + // Finished initial pagination phase, request refresh + // Expect no results. + resp6, retPlg, err := hostplugin.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) + require.Empty(t, retPlg) + + // Create some new hosts + host1 := hostplugin.TestSet(t, conn, testKms, sche, catalog, plgm) + host1.PluginId = plg.GetPublicId() + host2 := hostplugin.TestSet(t, conn, testKms, sche, catalog, plgm) + host2.PluginId = plg.GetPublicId() + t.Cleanup(func() { + _, err := repo.DeleteSet(ctx, proj1.PublicId, host1.GetPublicId()) + require.NoError(t, err) + _, err = repo.DeleteSet(ctx, proj1.PublicId, host2.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 host2 + resp7, retPlg, err := hostplugin.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], host2, cmpOpts...)) + require.Equal(t, retPlg, plg) + + // Refresh again, should get host1 + resp8, retPlg, err := hostplugin.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], host1, cmpOpts...)) + require.Equal(t, retPlg, plg) + + // Refresh again, should get no results + resp9, retPlg, err := hostplugin.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) + require.Empty(t, retPlg) + }) + + t.Run("simple pagination with aggressive filtering", func(t *testing.T) { + filterFunc := func(ctx context.Context, s host.Set, plg *plugin.Plugin) (bool, error) { + return s.GetPublicId() == hostSets[1].GetPublicId() || + s.GetPublicId() == hostSets[len(hostSets)-1].GetPublicId(), nil + } + resp, retPlg, err := hostplugin.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], hostSets[1], cmpOpts...)) + require.Equal(t, retPlg, plg) + + resp2, retPlg, err := hostplugin.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], hostSets[len(hostSets)-1], cmpOpts...)) + require.Equal(t, retPlg, plg) + + // request a refresh, nothing should be returned + resp3, retPlg, err := hostplugin.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) + require.Empty(t, retPlg) + + // Create some new hosts + host1 := hostplugin.TestSet(t, conn, testKms, sche, catalog, plgm) + host1.PluginId = plg.GetPublicId() + host2 := hostplugin.TestSet(t, conn, testKms, sche, catalog, plgm) + host2.PluginId = plg.GetPublicId() + host3 := hostplugin.TestSet(t, conn, testKms, sche, catalog, plgm) + host3.PluginId = plg.GetPublicId() + host4 := hostplugin.TestSet(t, conn, testKms, sche, catalog, plgm) + host4.PluginId = plg.GetPublicId() + t.Cleanup(func() { + _, err := repo.DeleteSet(ctx, proj1.PublicId, host1.GetPublicId()) + require.NoError(t, err) + _, err = repo.DeleteSet(ctx, proj1.PublicId, host2.GetPublicId()) + require.NoError(t, err) + _, err = repo.DeleteSet(ctx, proj1.PublicId, host3.GetPublicId()) + require.NoError(t, err) + _, err = repo.DeleteSet(ctx, proj1.PublicId, host4.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, retPlg *plugin.Plugin) (bool, error) { + return s.GetPublicId() == host3.GetPublicId() || + s.GetPublicId() == host1.GetPublicId(), nil + } + // Refresh again, should get host3 + resp4, retPlg, err := hostplugin.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], host3, cmpOpts...)) + require.Equal(t, retPlg, plg) + + // Refresh again, should get host1 + resp5, retPlg, err := hostplugin.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], host1, cmpOpts...)) + require.Equal(t, retPlg, plg) + }) + + t.Run("simple pagination with deletion", func(t *testing.T) { + filterFunc := func(context.Context, host.Set, *plugin.Plugin) (bool, error) { + return true, nil + } + deletedHostId := hostSets[0].GetPublicId() + _, err := repo.DeleteSet(ctx, proj1.PublicId, deletedHostId) + require.NoError(t, err) + hostSets = hostSets[1:] + + // Run analyze to update count estimate + _, err = sqlDb.ExecContext(ctx, "analyze") + require.NoError(t, err) + + resp, retPlg, err := hostplugin.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], hostSets[0], cmpOpts...)) + require.Equal(t, retPlg, plg) + + // request remaining results + resp2, retPlg, err := hostplugin.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, hostSets[1:], cmpOpts...)) + require.Equal(t, retPlg, plg) + + deletedHostId = hostSets[0].GetPublicId() + _, err = repo.DeleteSet(ctx, proj1.PublicId, deletedHostId) + require.NoError(t, err) + hostSets = hostSets[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, retPlg, err := hostplugin.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, deletedHostId) + require.Empty(t, resp3.Items) + require.Empty(t, retPlg) + }) +} diff --git a/internal/host/plugin/service_list_host_sets_page.go b/internal/host/plugin/service_list_host_sets_page.go new file mode 100644 index 000000000..c5673c741 --- /dev/null +++ b/internal/host/plugin/service_list_host_sets_page.go @@ -0,0 +1,81 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: BUSL-1.1 + +package plugin + +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" + ppagination "github.com/hashicorp/boundary/internal/pagination/plugin" + "github.com/hashicorp/boundary/internal/plugin" + "github.com/hashicorp/boundary/internal/types/resource" +) + +// ListPage 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 ppagination.ListPluginFilterFunc[host.Set], + tok *listtoken.Token, + repo *Repository, + hostCatalogId string, +) (*pagination.ListResponse[host.Set], *plugin.Plugin, error) { + const op = "plugin.ListHostSetsPage" + + 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 tok == nil: + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing token") + case repo == nil: + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing repo") + case hostCatalogId == "": + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing host catalog ID") + case tok.ResourceType != resource.HostSet: + return nil, 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, 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, *plugin.Plugin, 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, nil, time.Time{}, err + } + opts = append(opts, WithStartPageAfterItem(lastItem)) + } + pSets, plg, listTime, err := repo.listSets(ctx, hostCatalogId, opts...) + if err != nil { + return nil, nil, time.Time{}, err + } + var sets []host.Set + for _, set := range pSets { + sets = append(sets, set) + } + return sets, plg, listTime, nil + } + + return ppagination.ListPluginPage(ctx, grantsHash, pageSize, filterItemFn, listItemsFn, repo.estimatedSetCount, tok) +} diff --git a/internal/host/plugin/service_list_host_sets_refresh.go b/internal/host/plugin/service_list_host_sets_refresh.go new file mode 100644 index 000000000..99b62e1cb --- /dev/null +++ b/internal/host/plugin/service_list_host_sets_refresh.go @@ -0,0 +1,83 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: BUSL-1.1 + +package plugin + +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" + ppagination "github.com/hashicorp/boundary/internal/pagination/plugin" + "github.com/hashicorp/boundary/internal/plugin" + "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 ppagination.ListPluginFilterFunc[host.Set], + tok *listtoken.Token, + repo *Repository, + hostCatalogId string, +) (*pagination.ListResponse[host.Set], *plugin.Plugin, error) { + const op = "plugin.ListHostSetsRefresh" + + 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 tok == nil: + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing token") + case repo == nil: + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing repo") + case hostCatalogId == "": + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing host catalog ID") + case tok.ResourceType != resource.HostSet: + return nil, 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, 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, *plugin.Plugin, 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. + pSets, plg, listTime, err := repo.listSetsRefresh(ctx, hostCatalogId, rt.PreviousPhaseUpperBound.Add(-globals.RefreshReadLookbackDuration), opts...) + if err != nil { + return nil, nil, time.Time{}, err + } + var sets []host.Set + for _, set := range pSets { + sets = append(sets, set) + } + return sets, plg, 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 ppagination.ListPluginRefresh(ctx, grantsHash, pageSize, filterItemFn, listItemsFn, repo.estimatedSetCount, listDeletedIdsFn, tok) +} diff --git a/internal/host/plugin/service_list_host_sets_refresh_page.go b/internal/host/plugin/service_list_host_sets_refresh_page.go new file mode 100644 index 000000000..a4f9d0911 --- /dev/null +++ b/internal/host/plugin/service_list_host_sets_refresh_page.go @@ -0,0 +1,93 @@ +// Copyright (c) HashiCorp, Inc. +// SPDX-License-Identifier: BUSL-1.1 + +package plugin + +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" + ppagination "github.com/hashicorp/boundary/internal/pagination/plugin" + "github.com/hashicorp/boundary/internal/plugin" + "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 ppagination.ListPluginFilterFunc[host.Set], + tok *listtoken.Token, + repo *Repository, + hostCatalogId string, +) (*pagination.ListResponse[host.Set], *plugin.Plugin, error) { + const op = "credential.ListHostSetsRefreshPage" + + 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 tok == nil: + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing token") + case repo == nil: + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing repo") + case hostCatalogId == "": + return nil, nil, errors.New(ctx, errors.InvalidParameter, op, "missing host catalog ID") + case tok.ResourceType != resource.HostSet: + return nil, 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, 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, *plugin.Plugin, 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, 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. + pSets, plg, listTime, err := repo.listSetsRefresh(ctx, hostCatalogId, rt.PhaseLowerBound.Add(-globals.RefreshReadLookbackDuration), opts...) + if err != nil { + return nil, nil, time.Time{}, err + } + var sets []host.Set + for _, set := range pSets { + sets = append(sets, set) + } + return sets, plg, 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 ppagination.ListPluginRefreshPage(ctx, grantsHash, pageSize, filterItemFn, listItemsFn, repo.estimatedSetCount, listDeletedIdsFn, tok) +}