satellite/console/dbcleanup: make chore clean up project invites

The console DB cleanup chore has been extended to remove old project
member invitation records.

Resolves #5816

Change-Id: Id0a748e40f5acf03b9b903265c653b072846ba19

satellite/console/dbcleanup/chore.go

@@ -24,7 +24,8 @@ type Config struct {
 	AsOfSystemTimeInterval time.Duration `help:"interval for 'AS OF SYSTEM TIME' clause (CockroachDB specific) to read from the DB at a specific time in the past" default:"-5m" testDefault:"0"`
 	PageSize               int           `help:"maximum number of database records to scan at once" default:"1000"`
 
-	MaxUnverifiedUserAge time.Duration `help:"maximum lifetime of unverified user account records" default:"168h"`
+	MaxUnverifiedUserAge    time.Duration `help:"maximum lifetime of unverified user account records" default:"168h"`
+	MaxProjectInvitationAge time.Duration `help:"maximum lifetime of project member invitation records" default:"168h"`
 }
 
 // Chore periodically removes unwanted records from the satellite console database.
@@ -54,6 +55,13 @@ func (chore *Chore) Run(ctx context.Context) (err error) {
 		if err != nil {
 			chore.log.Error("Error deleting unverified users", zap.Error(err))
 		}
+
+		before = time.Now().Add(-chore.config.MaxProjectInvitationAge)
+		err = chore.db.ProjectInvitations().DeleteBefore(ctx, before, chore.config.AsOfSystemTimeInterval, chore.config.PageSize)
+		if err != nil {
+			chore.log.Error("Error deleting project member invitations", zap.Error(err))
+		}
+
 		return nil
 	})
 }

satellite/console/projectinvitations.go

@@ -22,6 +22,8 @@ type ProjectInvitations interface {
 	GetByEmail(ctx context.Context, email string) ([]ProjectInvitation, error)
 	// Delete is a method for deleting a project member invitation from the database.
 	Delete(ctx context.Context, projectID uuid.UUID, email string) error
+	// DeleteBefore deletes project member invitations created prior to some time from the database.
+	DeleteBefore(ctx context.Context, before time.Time, asOfSystemTimeInterval time.Duration, pageSize int) error
 }
 
 // ProjectInvitation represents a pending project member invitation.

satellite/satellitedb/consoledb.go

@@ -48,7 +48,7 @@ func (db *ConsoleDB) ProjectMembers() console.ProjectMembers {
 
 // ProjectInvitations is a getter for ProjectInvitations repository.
 func (db *ConsoleDB) ProjectInvitations() console.ProjectInvitations {
-	return &projectInvitations{db.methods}
+	return &projectInvitations{db.db}
 }
 
 // APIKeys is a getter for APIKeys repository.

satellite/satellitedb/projectinvitations.go

@@ -5,6 +5,9 @@ package satellitedb
 
 import (
 	"context"
+	"database/sql"
+	"errors"
+	"time"
 
 	"storj.io/common/uuid"
 	"storj.io/storj/satellite/console"
@@ -16,7 +19,7 @@ var _ console.ProjectInvitations = (*projectInvitations)(nil)
 
 // projectInvitations is an implementation of console.ProjectInvitations.
 type projectInvitations struct {
-	db dbx.Methods
+	db *satelliteDB
 }
 
 // Insert is a method for inserting a project member invitation into the database.
@@ -69,6 +72,81 @@ func (invites *projectInvitations) Delete(ctx context.Context, projectID uuid.UU
 	return err
 }
 
+// DeleteBefore deletes project member invitations created prior to some time from the database.
+func (invites *projectInvitations) DeleteBefore(
+	ctx context.Context, before time.Time, asOfSystemTimeInterval time.Duration, pageSize int) (err error) {
+	defer mon.Task()(&ctx)(&err)
+
+	if pageSize <= 0 {
+		return Error.New("expected page size to be positive; got %d", pageSize)
+	}
+
+	var pageCursor, pageEnd struct {
+		ProjectID uuid.UUID
+		Email     string
+	}
+	aost := invites.db.impl.AsOfSystemInterval(asOfSystemTimeInterval)
+	for {
+		// Select the ID beginning this page of records
+		err := invites.db.QueryRowContext(ctx, `
+			SELECT project_id, email FROM project_invitations
+			`+aost+`
+			WHERE (project_id, email) > ($1, $2) AND created_at < $3
+			ORDER BY (project_id, email) LIMIT 1
+		`, pageCursor.ProjectID, pageCursor.Email, before).Scan(&pageCursor.ProjectID, &pageCursor.Email)
+		if err != nil {
+			if errors.Is(err, sql.ErrNoRows) {
+				return nil
+			}
+			return Error.Wrap(err)
+		}
+
+		// Select the ID ending this page of records
+		err = invites.db.QueryRowContext(ctx, `
+			SELECT project_id, email FROM project_invitations
+			`+aost+`
+			WHERE (project_id, email) > ($1, $2)
+			ORDER BY (project_id, email) LIMIT 1 OFFSET $3
+		`, pageCursor.ProjectID, pageCursor.Email, pageSize).Scan(&pageEnd.ProjectID, &pageEnd.Email)
+		if err != nil {
+			if !errors.Is(err, sql.ErrNoRows) {
+				return Error.Wrap(err)
+			}
+			// Since this is the last page, we want to return all remaining records
+			_, err = invites.db.ExecContext(ctx, `
+				DELETE FROM project_invitations
+				WHERE (project_id, email) IN (
+					SELECT project_id, email FROM project_invitations
+					`+aost+`
+					WHERE (project_id, email) >= ($1, $2)
+					AND created_at < $3
+					ORDER BY (project_id, email)
+				)
+			`, pageCursor.ProjectID, pageCursor.Email, before)
+			return Error.Wrap(err)
+		}
+
+		// Delete all old, unverified records in the range between the beginning and ending IDs
+		_, err = invites.db.ExecContext(ctx, `
+			DELETE FROM project_invitations
+			WHERE (project_id, email) IN (
+				SELECT project_id, email FROM project_invitations
+				`+aost+`
+				WHERE (project_id, email) >= ($1, $2)
+				AND (project_id, email) <= ($3, $4)
+				AND created_at < $5
+				ORDER BY (project_id, email)
+			)
+		`, pageCursor.ProjectID, pageCursor.Email, pageEnd.ProjectID, pageEnd.Email, before)
+		if err != nil {
+			return Error.Wrap(err)
+		}
+
+		// Advance the cursor to the next page
+		pageCursor = pageEnd
+	}
+}
+
 // projectInvitationFromDBX converts a project member invitation from the database to a *console.ProjectInvitation.
 func projectInvitationFromDBX(dbxInvite *dbx.ProjectInvitation) (_ *console.ProjectInvitation, err error) {
 	if dbxInvite == nil {

satellite/satellitedb/projectinvitations_test.go

@@ -94,3 +94,52 @@ func TestProjectInvitations(t *testing.T) {
 		})
 	})
 }
+
+func TestDeleteBefore(t *testing.T) {
+	maxAge := time.Hour
+	now := time.Now()
+	expiration := now.Add(-maxAge)
+
+	satellitedbtest.Run(t, func(ctx *testcontext.Context, t *testing.T, db satellite.DB) {
+		invitesDB := db.Console().ProjectInvitations()
+		now := time.Now()
+
+		// Only positive page sizes should be allowed.
+		require.Error(t, invitesDB.DeleteBefore(ctx, time.Time{}, 0, 0))
+		require.Error(t, invitesDB.DeleteBefore(ctx, time.Time{}, 0, -1))
+
+		createInvite := func(createdAt time.Time) *console.ProjectInvitation {
+			id := testrand.UUID()
+			_, err := db.Console().Projects().Insert(ctx, &console.Project{ID: id})
+			require.NoError(t, err)
+			invite, err := invitesDB.Insert(ctx, id, "")
+			require.NoError(t, err)
+
+			result, err := db.Testing().RawDB().ExecContext(ctx,
+				"UPDATE project_invitations SET created_at = $1 WHERE project_id = $2",
+				createdAt, invite.ProjectID,
+			)
+			require.NoError(t, err)
+
+			count, err := result.RowsAffected()
+			require.NoError(t, err)
+			require.EqualValues(t, 1, count)
+
+			return invite
+		}
+
+		newInvite := createInvite(now)
+		oldInvite := createInvite(expiration.Add(-time.Second))
+
+		require.NoError(t, invitesDB.DeleteBefore(ctx, expiration, 0, 1))
+
+		// Ensure that the old invitation record was deleted and the other remains.
+		invites, err := invitesDB.GetByProjectID(ctx, oldInvite.ProjectID)
+		require.NoError(t, err)
+		require.Empty(t, invites)
+
+		invites, err = invitesDB.GetByProjectID(ctx, newInvite.ProjectID)
+		require.NoError(t, err)
+		require.Len(t, invites, 1)
+	})
+}

scripts/testdata/satellite-config.yaml.lock

@@ -139,6 +139,9 @@ compensation.withheld-percents: 75,75,75,50,50,50,25,25,25,0,0,0,0,0,0
 # interval between chore cycles
 # console-db-cleanup.interval: 24h0m0s
 
+# maximum lifetime of project member invitation records
+# console-db-cleanup.max-project-invitation-age: 168h0m0s
+
 # maximum lifetime of unverified user account records
 # console-db-cleanup.max-unverified-user-age: 168h0m0s