internal/credential/static: add pagination support

Refactors ListCredentials and adds the new ListCredentialsRefresh
method to the Repository, supporting pagination
of credentials. Also adds the EstimateCount and
ListDeletedCredentialIds methods used during
pagination.

internal/credential/static/json_credential.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/hashicorp/boundary/internal/credential"
 	"github.com/hashicorp/boundary/internal/credential/static/store"
+	"github.com/hashicorp/boundary/internal/db/timestamp"
 	"github.com/hashicorp/boundary/internal/errors"
 	"github.com/hashicorp/boundary/internal/libs/crypto"
 	"github.com/hashicorp/boundary/internal/oplog"
@@ -145,3 +146,13 @@ func (c *JsonCredential) hmacObject(ctx context.Context, cipher wrapping.Wrapper
 	c.ObjectHmac = []byte(hm)
 	return nil
 }
+
+type deletedJSONCredential struct {
+	PublicId   string `gorm:"primary_key"`
+	DeleteTime *timestamp.Timestamp
+}
+
+// TableName returns the tablename to override the default gorm table name
+func (s *deletedJSONCredential) TableName() string {
+	return "credential_static_json_credential_deleted"
+}

internal/credential/static/query.go

@@ -39,5 +39,315 @@ from credential_static_json_credential json
     on store.public_id = json.store_id
 where store.project_id = ?
   and json.key_id = ?;
+`
+
+	estimateCountCredentials = `
+select sum(reltuples::bigint) as estimate
+  from pg_class
+ where oid in (
+  'credential_static_json_credential'::regclass,
+  'credential_static_username_password_credential'::regclass,
+  'credential_static_ssh_private_key_credential'::regclass
+ )
+`
+
+	listCredentialsTemplate = `
+with credentials as (
+    select public_id
+      from credential_static
+     where store_id = @store_id
+  order by create_time desc, public_id asc
+     limit %d
+),
+json_creds as (
+  select *
+    from credential_static_json_credential
+   where public_id in (select public_id from credentials)
+),
+upw_creds as (
+  select *
+    from credential_static_username_password_credential
+   where public_id in (select public_id from credentials)
+),
+ssh_creds as (
+  select *
+    from credential_static_ssh_private_key_credential
+   where public_id in (select public_id from credentials)
+),
+final as (
+  select public_id,
+         store_id,
+         project_id,
+         name,
+         description,
+         create_time,
+         update_time,
+         version,
+         null as username,     -- Add this to make the union uniform
+         key_id,
+         object_hmac as hmac1,
+         null::bytea as hmac2, -- Add this to make the union uniform
+         'json' as type
+    from json_creds
+   union
+  select public_id,
+         store_id,
+         project_id,
+         name,
+         description,
+         create_time,
+         update_time,
+         version,
+         username,
+         key_id,
+         password_hmac as hmac1,
+         null::bytea as hmac2, -- Add this to make the union uniform
+         'upw' as type
+    from upw_creds
+   union
+  select public_id,
+         store_id,
+         project_id,
+         name,
+         description,
+         create_time,
+         update_time,
+         version,
+         username,
+         key_id,
+         private_key_hmac as hmac1,
+         private_key_passphrase_hmac as hmac2,
+         'ssh' as type
+    from ssh_creds
+)
+  select *
+    from final
+order by create_time desc, public_id asc;
+`
+
+	listCredentialsPageTemplate = `
+with credentials as (
+    select public_id
+      from credential_static
+     where store_id = @store_id
+       and (create_time, public_id) < (@last_item_create_time, @last_item_id)
+  order by create_time desc, public_id asc
+     limit %d
+),
+json_creds as (
+  select *
+    from credential_static_json_credential
+   where public_id in (select public_id from credentials)
+),
+upw_creds as (
+  select *
+    from credential_static_username_password_credential
+   where public_id in (select public_id from credentials)
+),
+ssh_creds as (
+  select *
+    from credential_static_ssh_private_key_credential
+   where public_id in (select public_id from credentials)
+),
+final as (
+  select public_id,
+         store_id,
+         project_id,
+         name,
+         description,
+         create_time,
+         update_time,
+         version,
+         null as username,     -- Add this to make the union uniform
+         key_id,
+         object_hmac as hmac1,
+         null::bytea as hmac2, -- Add this to make the union uniform
+         'json' as type
+    from json_creds
+   union
+  select public_id,
+         store_id,
+         project_id,
+         name,
+         description,
+         create_time,
+         update_time,
+         version,
+         username,
+         key_id,
+         password_hmac as hmac1,
+         null::bytea as hmac2, -- Add this to make the union uniform
+         'upw' as type
+    from upw_creds
+   union
+  select public_id,
+         store_id,
+         project_id,
+         name,
+         description,
+         create_time,
+         update_time,
+         version,
+         username,
+         key_id,
+         private_key_hmac as hmac1,
+         private_key_passphrase_hmac as hmac2,
+         'ssh' as type
+    from ssh_creds
+)
+  select *
+    from final
+order by create_time desc, public_id asc;
+`
+
+	listCredentialsRefreshTemplate = `
+with credentials as (
+    select public_id
+      from credential_static
+     where store_id = @store_id
+       and update_time > @updated_after_time
+  order by update_time desc, public_id asc
+     limit %d
+),
+json_creds as (
+  select *
+    from credential_static_json_credential
+   where public_id in (select public_id from credentials)
+),
+upw_creds as (
+  select *
+    from credential_static_username_password_credential
+   where public_id in (select public_id from credentials)
+),
+ssh_creds as (
+  select *
+    from credential_static_ssh_private_key_credential
+   where public_id in (select public_id from credentials)
+),
+final as (
+  select public_id,
+         store_id,
+         project_id,
+         name,
+         description,
+         create_time,
+         update_time,
+         version,
+         null as username,     -- Add this to make the union uniform
+         key_id,
+         object_hmac as hmac1,
+         null::bytea as hmac2, -- Add this to make the union uniform
+         'json' as type
+    from json_creds
+   union
+  select public_id,
+         store_id,
+         project_id,
+         name,
+         description,
+         create_time,
+         update_time,
+         version,
+         username,
+         key_id,
+         password_hmac as hmac1,
+         null::bytea as hmac2, -- Add this to make the union uniform
+         'upw' as type
+    from upw_creds
+   union
+  select public_id,
+         store_id,
+         project_id,
+         name,
+         description,
+         create_time,
+         update_time,
+         version,
+         username,
+         key_id,
+         private_key_hmac as hmac1,
+         private_key_passphrase_hmac as hmac2,
+         'ssh' as type
+    from ssh_creds
+)
+  select *
+    from final
+order by update_time desc, public_id asc;
+`
+
+	listCredentialsRefreshPageTemplate = `
+with credentials as (
+    select public_id
+      from credential_static
+     where store_id = @store_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
+),
+json_creds as (
+  select *
+    from credential_static_json_credential
+   where public_id in (select public_id from credentials)
+),
+upw_creds as (
+  select *
+    from credential_static_username_password_credential
+   where public_id in (select public_id from credentials)
+),
+ssh_creds as (
+  select *
+    from credential_static_ssh_private_key_credential
+   where public_id in (select public_id from credentials)
+),
+final as (
+  select public_id,
+         store_id,
+         project_id,
+         name,
+         description,
+         create_time,
+         update_time,
+         version,
+         null as username,     -- Add this to make the union uniform
+         key_id,
+         object_hmac as hmac1,
+         null::bytea as hmac2, -- Add this to make the union uniform
+         'json' as type
+    from json_creds
+   union
+  select public_id,
+         store_id,
+         project_id,
+         name,
+         description,
+         create_time,
+         update_time,
+         version,
+         username,
+         key_id,
+         password_hmac as hmac1,
+         null::bytea as hmac2, -- Add this to make the union uniform
+         'upw' as type
+    from upw_creds
+   union
+  select public_id,
+         store_id,
+         project_id,
+         name,
+         description,
+         create_time,
+         update_time,
+         version,
+         username,
+         key_id,
+         private_key_hmac as hmac1,
+         private_key_passphrase_hmac as hmac2,
+         'ssh' as type
+    from ssh_creds
+)
+  select *
+    from final
+order by update_time desc, public_id asc;
 `
 )

internal/credential/static/repository_credential.go

@@ -5,12 +5,17 @@ package static
 
 import (
 	"context"
+	"database/sql"
 	"fmt"
+	"slices"
 	"strings"
+	"time"
 
 	"github.com/hashicorp/boundary/globals"
 	"github.com/hashicorp/boundary/internal/credential"
+	"github.com/hashicorp/boundary/internal/credential/static/store"
 	"github.com/hashicorp/boundary/internal/db"
+	"github.com/hashicorp/boundary/internal/db/timestamp"
 	"github.com/hashicorp/boundary/internal/errors"
 	"github.com/hashicorp/boundary/internal/kms"
 	"github.com/hashicorp/boundary/internal/oplog"
@@ -715,67 +720,204 @@ func (r *Repository) UpdateJsonCredential(ctx context.Context,
 	return returnedCredential, rowsUpdated, nil
 }
 
-// ListCredentials returns a slice of UsernamePasswordCredentials, SshPrivateKeyCredentials, and JsonCredentials
-// for the storeId. WithLimit is the only option supported.
-// TODO: This should hit a view and return the interface type...
-func (r *Repository) ListCredentials(ctx context.Context, storeId string, opt ...Option) ([]credential.Static, error) {
+// ListCredentials returns a slice of static credentials
+// for the storeId. Supports the following options:
+//   - WithLimit
+//   - WithStartPageAfterItem
+func (r *Repository) ListCredentials(ctx context.Context, storeId string, opt ...credential.Option) ([]credential.Static, time.Time, error) {
 	const op = "static.(Repository).ListCredentials"
 	if storeId == "" {
-		return nil, errors.New(ctx, errors.InvalidParameter, op, "no storeId")
+		return nil, time.Time{}, errors.New(ctx, errors.InvalidParameter, op, "no storeId")
+	}
+	opts, err := credential.GetOpts(opt...)
+	if err != nil {
+		return nil, time.Time{}, 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
+	}
+	query := fmt.Sprintf(listCredentialsTemplate, limit)
+	args := []any{sql.Named("store_id", storeId)}
+	if opts.WithStartPageAfterItem != nil {
+		query = fmt.Sprintf(listCredentialsPageTemplate, limit)
+		args = append(args,
+			sql.Named("last_item_create_time", opts.WithStartPageAfterItem.GetCreateTime()),
+			sql.Named("last_item_id", opts.WithStartPageAfterItem.GetPublicId()),
+		)
 	}
 
-	var upCreds []*UsernamePasswordCredential
-	err := r.reader.SearchWhere(ctx, &upCreds, "store_id = ?", []any{storeId}, db.WithLimit(limit))
+	creds, transactionTimestamp, err := r.queryCredentials(ctx, query, args)
 	if err != nil {
-		return nil, errors.Wrap(ctx, err, op)
+		return nil, time.Time{}, errors.Wrap(ctx, err, op)
 	}
 
-	var spkCreds []*SshPrivateKeyCredential
-	err = r.reader.SearchWhere(ctx, &spkCreds, "store_id = ?", []any{storeId}, db.WithLimit(limit))
+	// Sort final slice to ensure correct ordering.
+	// We sort by create time descending (most recently created first).
+	slices.SortFunc(creds, func(i, j credential.Static) int {
+		return j.GetCreateTime().AsTime().Compare(i.GetCreateTime().AsTime())
+	})
+
+	return creds, transactionTimestamp, nil
+}
+
+// ListCredentialRefresh returns a slice of static credentials
+// for the storeId. Supports the following options:
+//   - WithLimit
+//   - WithStartPageAfterItem
+func (r *Repository) ListCredentialsRefresh(ctx context.Context, storeId string, updatedAfter time.Time, opt ...credential.Option) ([]credential.Static, time.Time, error) {
+	const op = "static.(Repository).ListCredentials"
+	switch {
+	case storeId == "":
+		return nil, time.Time{}, errors.New(ctx, errors.InvalidParameter, op, "missing credential store ID")
+	case updatedAfter.IsZero():
+		return nil, time.Time{}, errors.New(ctx, errors.InvalidParameter, op, "missing updated after time")
+	}
+	opts, err := credential.GetOpts(opt...)
 	if err != nil {
-		return nil, errors.Wrap(ctx, err, op)
+		return nil, time.Time{}, errors.Wrap(ctx, err, op)
+	}
+	limit := r.defaultLimit
+	if opts.WithLimit != 0 {
+		// non-zero signals an override of the default limit for the repo.
+		limit = opts.WithLimit
 	}
 
-	var jsonCreds []*JsonCredential
-	err = r.reader.SearchWhere(ctx, &jsonCreds, "store_id = ?", []any{storeId}, db.WithLimit(limit))
+	query := fmt.Sprintf(listCredentialsRefreshTemplate, limit)
+	args := []any{
+		sql.Named("store_id", storeId),
+		sql.Named("updated_after_time", timestamp.New(updatedAfter)),
+	}
+	if opts.WithStartPageAfterItem != nil {
+		query = fmt.Sprintf(listCredentialsRefreshPageTemplate, limit)
+		args = append(args,
+			sql.Named("last_item_update_time", opts.WithStartPageAfterItem.GetUpdateTime()),
+			sql.Named("last_item_id", opts.WithStartPageAfterItem.GetPublicId()),
+		)
+	}
+
+	creds, transactionTimestamp, err := r.queryCredentials(ctx, query, args)
 	if err != nil {
-		return nil, errors.Wrap(ctx, err, op)
+		return nil, time.Time{}, errors.Wrap(ctx, err, op)
 	}
 
-	ret := make([]credential.Static, 0, len(upCreds)+len(spkCreds)+len(jsonCreds))
+	// Sort final slice to ensure correct ordering.
+	// We sort by update time descending (most recently updated first).
+	slices.SortFunc(creds, func(i, j credential.Static) int {
+		return j.GetUpdateTime().AsTime().Compare(i.GetUpdateTime().AsTime())
+	})
 
-	for _, c := range upCreds {
-		// Clear password fields, only PasswordHmac should be returned
-		c.CtPassword = nil
-		c.Password = nil
-		ret = append(ret, c)
+	return creds, transactionTimestamp, nil
+}
+
+func (r *Repository) queryCredentials(ctx context.Context, query string, args []any) ([]credential.Static, time.Time, error) {
+	const op = "static.(Repository).queryCredentials"
+
+	var creds []credential.Static
+	var transactionTimestamp time.Time
+	if _, err := r.writer.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{}, func(rd db.Reader, w db.Writer) error {
+		rows, err := rd.Query(ctx, query, args)
+		if err != nil {
+			return errors.Wrap(ctx, err, op)
+		}
+		defer rows.Close()
+		for rows.Next() {
+			var (
+				publicId, storeId, projectId, name, description,
+				username, keyId, hmac1, hmac2 sql.NullString
+				createTime, updateTime *timestamp.Timestamp
+				version                sql.NullInt64
+				typ                    string
+			)
+			if err := rows.Scan(
+				&publicId,
+				&storeId,
+				&projectId,
+				&name,
+				&description,
+				&createTime,
+				&updateTime,
+				&version,
+				&username,
+				&keyId,
+				&hmac1,
+				&hmac2,
+				&typ,
+			); err != nil {
+				return errors.Wrap(ctx, err, op)
+			}
+			switch typ {
+			case "json":
+				cred := &JsonCredential{
+					JsonCredential: &store.JsonCredential{
+						PublicId:    publicId.String,
+						StoreId:     storeId.String,
+						Name:        name.String,
+						Description: description.String,
+						CreateTime:  createTime,
+						UpdateTime:  updateTime,
+						Version:     uint32(version.Int64),
+						KeyId:       keyId.String,
+					},
+				}
+				// Assign byte slices only if the string isn't empty
+				if hmac1.String != "" {
+					cred.ObjectHmac = []byte(hmac1.String)
+				}
+				creds = append(creds, cred)
+			case "upw":
+				cred := &UsernamePasswordCredential{
+					UsernamePasswordCredential: &store.UsernamePasswordCredential{
+						PublicId:    publicId.String,
+						StoreId:     storeId.String,
+						Name:        name.String,
+						Description: description.String,
+						CreateTime:  createTime,
+						UpdateTime:  updateTime,
+						Version:     uint32(version.Int64),
+						Username:    username.String,
+						KeyId:       keyId.String,
+					},
+				}
+				// Assign byte slices only if the string isn't empty
+				if hmac1.String != "" {
+					cred.PasswordHmac = []byte(hmac1.String)
+				}
+				creds = append(creds, cred)
+			case "ssh":
+				cred := &SshPrivateKeyCredential{
+					SshPrivateKeyCredential: &store.SshPrivateKeyCredential{
+						PublicId:    publicId.String,
+						StoreId:     storeId.String,
+						Name:        name.String,
+						Description: description.String,
+						CreateTime:  createTime,
+						UpdateTime:  updateTime,
+						Version:     uint32(version.Int64),
+						Username:    username.String,
+						KeyId:       keyId.String,
+					},
+				}
+				// Assign byte slices only if the string isn't empty
+				if hmac1.String != "" {
+					cred.PrivateKeyHmac = []byte(hmac1.String)
+				}
+				if hmac2.String != "" {
+					cred.PrivateKeyPassphraseHmac = []byte(hmac2.String)
+				}
+				creds = append(creds, cred)
+			default:
+				return errors.New(ctx, errors.Internal, op, fmt.Sprintf("unexpected static credential type %s returned", typ))
+			}
+		}
+		transactionTimestamp, err = rd.Now(ctx)
+		return err
+	}); err != nil {
+		return nil, time.Time{}, err
 	}
 
-	for _, c := range spkCreds {
-		// Clear private key fields, only PrivateKeyHmac should be returned
-		c.PrivateKeyEncrypted = nil
-		c.PrivateKey = nil
-
-		// Clear passphrase fields, only PrivateKeyPassphraseHmac should be returned if it exists
-		c.PrivateKeyPassphraseEncrypted = nil
-		c.PrivateKeyPassphrase = nil
-		ret = append(ret, c)
-	}
-
-	for _, c := range jsonCreds {
-		// Clear the object fields, only ObjectHmac should be returned
-		c.ObjectEncrypted = nil
-		c.Object = nil
-		ret = append(ret, c)
-	}
-
-	return ret, nil
+	return creds, transactionTimestamp, nil
 }
 
 // DeleteCredential deletes publicId from the repository and returns
@@ -839,3 +981,58 @@ func (r *Repository) DeleteCredential(ctx context.Context, projectId, id string,
 
 	return rowsDeleted, nil
 }
+
+// EstimatedCredentialCount returns an estimate of the number of static credentials
+func (r *Repository) EstimatedCredentialCount(ctx context.Context) (int, error) {
+	const op = "static.(Repository).EstimatedCredentialCount"
+	rows, err := r.reader.Query(ctx, estimateCountCredentials, nil)
+	if err != nil {
+		return 0, errors.Wrap(ctx, err, op, errors.WithMsg("failed to query total static credentials"))
+	}
+	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 total static credentials"))
+		}
+	}
+	return count, nil
+}
+
+// ListDeletedCredentialIds lists the public IDs of any credentials deleted since the timestamp provided.
+func (r *Repository) ListDeletedCredentialIds(ctx context.Context, since time.Time) ([]string, time.Time, error) {
+	const op = "static.(Repository).ListDeletedCredentialIds"
+	var credentialStoreIds []string
+	var transactionTimestamp time.Time
+	if _, err := r.writer.DoTx(ctx, db.StdRetryCnt, db.ExpBackoff{}, func(r db.Reader, w db.Writer) error {
+		var deletedJSONCredentials []*deletedJSONCredential
+		if err := r.SearchWhere(ctx, &deletedJSONCredentials, "delete_time >= ?", []any{since}); err != nil {
+			return errors.Wrap(ctx, err, op, errors.WithMsg("failed to query deleted JSON credentials"))
+		}
+		for _, cl := range deletedJSONCredentials {
+			credentialStoreIds = append(credentialStoreIds, cl.PublicId)
+		}
+		var deletedUsernamePasswordCredentials []*deletedUsernamePasswordCredential
+		if err := r.SearchWhere(ctx, &deletedUsernamePasswordCredentials, "delete_time >= ?", []any{since}); err != nil {
+			return errors.Wrap(ctx, err, op, errors.WithMsg("failed to query deleted username password credentials"))
+		}
+		for _, cl := range deletedUsernamePasswordCredentials {
+			credentialStoreIds = append(credentialStoreIds, cl.PublicId)
+		}
+		var deletedSSHPrivateKeyCredentials []*deletedSSHPrivateKeyCredential
+		if err := r.SearchWhere(ctx, &deletedSSHPrivateKeyCredentials, "delete_time >= ?", []any{since}); err != nil {
+			return errors.Wrap(ctx, err, op, errors.WithMsg("failed to query deleted ssh private key credentials"))
+		}
+		for _, cl := range deletedSSHPrivateKeyCredentials {
+			credentialStoreIds = append(credentialStoreIds, cl.PublicId)
+		}
+		var err error
+		transactionTimestamp, err = r.Now(ctx)
+		if err != nil {
+			return errors.Wrap(ctx, err, op, errors.WithMsg("failed to query transaction timestamp"))
+		}
+		return nil
+	}); err != nil {
+		return nil, time.Time{}, err
+	}
+	return credentialStoreIds, transactionTimestamp, nil
+}

internal/credential/static/repository_credential_test.go

@@ -9,6 +9,8 @@ import (
 	"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/credential"
 	"github.com/hashicorp/boundary/internal/credential/static/store"
@@ -687,35 +689,27 @@ func TestRepository_ListCredentials(t *testing.T) {
 
 	type args struct {
 		storeId string
-		opt     []Option
+		opt     []credential.Option
 	}
 	tests := []struct {
 		name    string
 		args    args
 		wantCnt int
 	}{
-		{
-			name: "no-limit",
-			args: args{
-				storeId: store.PublicId,
-				opt:     []Option{WithLimit(-1)},
-			},
-			wantCnt: total,
-		},
 		{
 			name: "default-limit",
 			args: args{
 				storeId: store.PublicId,
 			},
-			wantCnt: defaultLimit * 3,
+			wantCnt: defaultLimit,
 		},
 		{
 			name: "custom-limit",
 			args: args{
 				storeId: store.PublicId,
-				opt:     []Option{WithLimit(3)},
+				opt:     []credential.Option{credential.WithLimit(3)},
 			},
-			wantCnt: 9,
+			wantCnt: 3,
 		},
 		{
 			name: "bad-store",
@@ -733,9 +727,12 @@ func TestRepository_ListCredentials(t *testing.T) {
 			require.NotNil(repo)
 			repo.defaultLimit = defaultLimit
 
-			got, err := repo.ListCredentials(context.Background(), tt.args.storeId, tt.args.opt...)
+			got, ttime, err := repo.ListCredentials(context.Background(), tt.args.storeId, tt.args.opt...)
 			require.NoError(err)
 			assert.Equal(tt.wantCnt, len(got))
+			// 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)))
 
 			// Validate only hmac values are returned
 			for _, c := range got {
@@ -760,6 +757,55 @@ func TestRepository_ListCredentials(t *testing.T) {
 	}
 }
 
+func TestRepository_ListCredentials_Pagination(t *testing.T) {
+	t.Parallel()
+	assert, require := assert.New(t), require.New(t)
+	ctx := context.Background()
+	conn, _ := db.TestSetup(t, "postgres")
+	rw := db.New(conn)
+	wrapper := db.TestWrapper(t)
+	kms := kms.TestKms(t, conn, wrapper)
+	_, prj := iam.TestScopes(t, iam.TestRepo(t, conn, wrapper))
+	store := TestCredentialStore(t, conn, wrapper, prj.GetPublicId())
+	obj, _ := TestJsonObject(t)
+	_ = TestJsonCredentials(t, conn, wrapper, store.GetPublicId(), prj.GetPublicId(), obj, 2)
+	_ = TestSshPrivateKeyCredentials(t, conn, wrapper, "username", TestSshPrivateKeyPem, store.GetPublicId(), prj.GetPublicId(), 2)
+	_ = TestUsernamePasswordCredentials(t, conn, wrapper, "username", "testpassword", store.GetPublicId(), prj.GetPublicId(), 1)
+
+	repo, err := NewRepository(ctx, rw, rw, kms)
+	require.NoError(err)
+	require.NotNil(repo)
+
+	page1, ttime, err := repo.ListCredentials(ctx, store.GetPublicId(), credential.WithLimit(2))
+	require.NoError(err)
+	require.Len(page1, 2)
+	// 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, ttime, err := repo.ListCredentials(ctx, store.GetPublicId(), credential.WithLimit(2), credential.WithStartPageAfterItem(page1[1]))
+	require.NoError(err)
+	require.Len(page2, 2)
+	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, ttime, err := repo.ListCredentials(ctx, store.GetPublicId(), credential.WithLimit(2), credential.WithStartPageAfterItem(page2[1]))
+	require.NoError(err)
+	require.Len(page3, 1)
+	for _, item := range append(page1, 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, ttime, err := repo.ListCredentials(ctx, store.GetPublicId(), credential.WithLimit(2), credential.WithStartPageAfterItem(page3[0]))
+	require.NoError(err)
+	require.Empty(page4)
+	assert.True(time.Now().Before(ttime.Add(10 * time.Second)))
+	assert.True(time.Now().After(ttime.Add(-10 * time.Second)))
+}
+
 func TestRepository_DeleteCredential(t *testing.T) {
 	conn, _ := db.TestSetup(t, "postgres")
 	rw := db.New(conn)
@@ -2441,3 +2487,160 @@ func TestRepository_UpdateJsonCredential(t *testing.T) {
 		})
 	}
 }
+
+func TestRepository_ListDeletedCredentialIds(t *testing.T) {
+	t.Parallel()
+	assert, require := assert.New(t), require.New(t)
+	ctx := context.Background()
+	conn, _ := db.TestSetup(t, "postgres")
+	rw := db.New(conn)
+	wrapper := db.TestWrapper(t)
+	kms := kms.TestKms(t, conn, wrapper)
+	_, prj := iam.TestScopes(t, iam.TestRepo(t, conn, wrapper))
+	store := TestCredentialStore(t, conn, wrapper, prj.GetPublicId())
+	obj, _ := TestJsonObject(t)
+	jsonCreds := TestJsonCredentials(t, conn, wrapper, store.GetPublicId(), prj.GetPublicId(), obj, 2)
+	sshCreds := TestSshPrivateKeyCredentials(t, conn, wrapper, "username", TestSshPrivateKeyPem, store.GetPublicId(), prj.GetPublicId(), 2)
+	pwCreds := TestUsernamePasswordCredentials(t, conn, wrapper, "username", "testpassword", store.GetPublicId(), prj.GetPublicId(), 2)
+
+	repo, err := NewRepository(ctx, rw, rw, kms)
+	require.NoError(err)
+	require.NotNil(repo)
+	staticRepo, err := NewRepository(ctx, rw, rw, kms)
+	require.NoError(err)
+
+	// Expect no entries at the start
+	deletedIds, ttime, err := repo.ListDeletedCredentialIds(ctx, time.Now().AddDate(-1, 0, 0))
+	require.NoError(err)
+	require.Empty(deletedIds)
+	// Expect transaction timestamp to be within ~10 seconds of now
+	require.True(time.Now().Before(ttime.Add(10 * time.Second)))
+	require.True(time.Now().After(ttime.Add(-10 * time.Second)))
+
+	// Delete a json credential
+	_, err = staticRepo.DeleteCredential(ctx, prj.GetPublicId(), jsonCreds[0].GetPublicId())
+	require.NoError(err)
+
+	// Expect one entry
+	deletedIds, ttime, err = repo.ListDeletedCredentialIds(ctx, time.Now().AddDate(-1, 0, 0))
+	require.NoError(err)
+	assert.Empty(
+		cmp.Diff(
+			[]string{jsonCreds[0].GetPublicId()},
+			deletedIds,
+			cmpopts.SortSlices(func(i, j string) bool { return i < j }),
+		),
+	)
+	require.True(time.Now().Before(ttime.Add(10 * time.Second)))
+	require.True(time.Now().After(ttime.Add(-10 * time.Second)))
+
+	// Delete a ssh credential
+	_, err = staticRepo.DeleteCredential(ctx, prj.GetPublicId(), sshCreds[0].GetPublicId())
+	require.NoError(err)
+
+	// Expect two entries
+	deletedIds, ttime, err = repo.ListDeletedCredentialIds(ctx, time.Now().AddDate(-1, 0, 0))
+	require.NoError(err)
+	assert.Empty(
+		cmp.Diff(
+			[]string{jsonCreds[0].GetPublicId(), sshCreds[0].GetPublicId()},
+			deletedIds,
+			cmpopts.SortSlices(func(i, j string) bool { return i < j }),
+		),
+	)
+	require.True(time.Now().Before(ttime.Add(10 * time.Second)))
+	require.True(time.Now().After(ttime.Add(-10 * time.Second)))
+
+	// Delete a pw credential
+	_, err = staticRepo.DeleteCredential(ctx, prj.GetPublicId(), pwCreds[0].GetPublicId())
+	require.NoError(err)
+
+	// Expect three entries
+	deletedIds, ttime, err = repo.ListDeletedCredentialIds(ctx, time.Now().AddDate(-1, 0, 0))
+	require.NoError(err)
+	assert.Empty(
+		cmp.Diff(
+			[]string{jsonCreds[0].GetPublicId(), sshCreds[0].GetPublicId(), pwCreds[0].GetPublicId()},
+			deletedIds,
+			cmpopts.SortSlices(func(i, j string) bool { return i < j }),
+		),
+	)
+	require.True(time.Now().Before(ttime.Add(10 * time.Second)))
+	require.True(time.Now().After(ttime.Add(-10 * time.Second)))
+
+	// Try again with the time set to now, expect no entries
+	deletedIds, ttime, err = repo.ListDeletedCredentialIds(ctx, time.Now())
+	require.NoError(err)
+	require.Empty(deletedIds)
+	require.True(time.Now().Before(ttime.Add(10 * time.Second)))
+	require.True(time.Now().After(ttime.Add(-10 * time.Second)))
+}
+
+func TestRepository_EstimatedCredentialCount(t *testing.T) {
+	t.Parallel()
+	assert, require := assert.New(t), require.New(t)
+	ctx := context.Background()
+	conn, _ := db.TestSetup(t, "postgres")
+	sqlDb, err := conn.SqlDB(ctx)
+	require.NoError(err)
+	rw := db.New(conn)
+	wrapper := db.TestWrapper(t)
+	kms := kms.TestKms(t, conn, wrapper)
+	_, prj := iam.TestScopes(t, iam.TestRepo(t, conn, wrapper))
+	staticStore := TestCredentialStore(t, conn, wrapper, prj.GetPublicId())
+
+	repo, err := NewRepository(ctx, rw, rw, kms)
+	require.NoError(err)
+	require.NotNil(repo)
+	staticRepo, err := NewRepository(ctx, rw, rw, kms)
+	require.NoError(err)
+
+	// Check total entries at start, expect 0
+	numItems, err := repo.EstimatedCredentialCount(ctx)
+	require.NoError(err)
+	assert.Equal(0, numItems)
+
+	// Create some credentials
+	obj, _ := TestJsonObject(t)
+	assert.NoError(err)
+	jsonCreds := TestJsonCredentials(t, conn, wrapper, staticStore.GetPublicId(), prj.GetPublicId(), obj, 2)
+	sshCreds := TestSshPrivateKeyCredentials(t, conn, wrapper, "username", TestSshPrivateKeyPem, staticStore.GetPublicId(), prj.GetPublicId(), 2)
+	pwCreds := TestUsernamePasswordCredentials(t, conn, wrapper, "username", "testpassword", staticStore.GetPublicId(), prj.GetPublicId(), 2)
+	// Run analyze to update postgres meta tables
+	_, err = sqlDb.ExecContext(ctx, "analyze")
+	require.NoError(err)
+
+	numItems, err = repo.EstimatedCredentialCount(ctx)
+	require.NoError(err)
+	assert.Equal(6, numItems)
+
+	// Delete a json credential
+	_, err = staticRepo.DeleteCredential(ctx, prj.GetPublicId(), jsonCreds[0].GetPublicId())
+	require.NoError(err)
+	_, err = sqlDb.ExecContext(ctx, "analyze")
+	require.NoError(err)
+
+	numItems, err = repo.EstimatedCredentialCount(ctx)
+	require.NoError(err)
+	assert.Equal(5, numItems)
+
+	// Delete a ssh credential
+	_, err = staticRepo.DeleteCredential(ctx, prj.GetPublicId(), sshCreds[0].GetPublicId())
+	require.NoError(err)
+	_, err = sqlDb.ExecContext(ctx, "analyze")
+	require.NoError(err)
+
+	numItems, err = repo.EstimatedCredentialCount(ctx)
+	require.NoError(err)
+	assert.Equal(4, numItems)
+
+	// Delete a pw credential
+	_, err = staticRepo.DeleteCredential(ctx, prj.GetPublicId(), pwCreds[0].GetPublicId())
+	require.NoError(err)
+	_, err = sqlDb.ExecContext(ctx, "analyze")
+	require.NoError(err)
+
+	numItems, err = repo.EstimatedCredentialCount(ctx)
+	require.NoError(err)
+	assert.Equal(3, numItems)
+}

internal/credential/static/ssh_private_key_credential.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/hashicorp/boundary/internal/credential"
 	"github.com/hashicorp/boundary/internal/credential/static/store"
+	"github.com/hashicorp/boundary/internal/db/timestamp"
 	"github.com/hashicorp/boundary/internal/errors"
 	"github.com/hashicorp/boundary/internal/libs/crypto"
 	"github.com/hashicorp/boundary/internal/oplog"
@@ -224,3 +225,13 @@ func (c *SshPrivateKeyCredential) hmacPrivateKeyPassphrase(ctx context.Context,
 	c.PrivateKeyPassphraseHmac = []byte(hm)
 	return nil
 }
+
+type deletedSSHPrivateKeyCredential struct {
+	PublicId   string `gorm:"primary_key"`
+	DeleteTime *timestamp.Timestamp
+}
+
+// TableName returns the tablename to override the default gorm table name
+func (s *deletedSSHPrivateKeyCredential) TableName() string {
+	return "credential_static_ssh_private_key_credential_deleted"
+}

internal/credential/static/usernamepassword_credential.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/hashicorp/boundary/internal/credential"
 	"github.com/hashicorp/boundary/internal/credential/static/store"
+	"github.com/hashicorp/boundary/internal/db/timestamp"
 	"github.com/hashicorp/boundary/internal/errors"
 	"github.com/hashicorp/boundary/internal/libs/crypto"
 	"github.com/hashicorp/boundary/internal/oplog"
@@ -130,3 +131,13 @@ func (c *UsernamePasswordCredential) hmacPassword(ctx context.Context, cipher wr
 	c.PasswordHmac = []byte(hm)
 	return nil
 }
+
+type deletedUsernamePasswordCredential struct {
+	PublicId   string `gorm:"primary_key"`
+	DeleteTime *timestamp.Timestamp
+}
+
+// TableName returns the tablename to override the default gorm table name
+func (s *deletedUsernamePasswordCredential) TableName() string {
+	return "credential_static_username_password_credential_deleted"
+}