[Server] Fix data cleanup pipeline for recovered account (#1410)

## Description
For accounts that were recovered post deletion (within x days), we were
not removing the entry from data_cleanup cron.

The data_clean up cron anyways verify that the account is indeed deleted
before proceeding with deletion, so it was not causing any harm, but the
cron was logging error.

## Tests

- [x] Verify that data clean up entry is removed if the account is
recovered
- [x] Verify that for existing data_cleanup entry for already recovered
account, the entry is deleted from cron.

server/pkg/controller/data_cleanup/controller.go

@@ -190,8 +190,17 @@ func (c *DeleteUserCleanupController) storageCheck(ctx context.Context, item *en
 }
 
 func (c *DeleteUserCleanupController) isDeleted(item *entity.DataCleanup) error {
-	_, err := c.UserRepo.Get(item.UserID)
+	u, err := c.UserRepo.Get(item.UserID)
 	if err == nil {
+		// user is not deleted, double check by verifying email is not empty
+		if u.Email != "" {
+			// todo: remove this logic after next deployment. This is to only handle cases
+			// where we have not removed scheduled delete entry for account post recovery.
+			remErr := c.Repo.RemoveScheduledDelete(context.Background(), item.UserID)
+			if remErr != nil {
+				return stacktrace.Propagate(remErr, "failed to remove scheduled delete entry")
+			}
+		}
 		return stacktrace.Propagate(ente.NewBadRequestWithMessage("User ID is linked to undeleted account"), "")
 	}
 	if !errors.Is(err, ente.ErrUserDeleted) {

server/pkg/controller/user/user.go

@@ -368,7 +368,15 @@ func (c *UserController) HandleAccountRecovery(ctx *gin.Context, req ente.Recove
 		return stacktrace.Propagate(err, "")
 	}
 	err = c.UserRepo.UpdateEmail(req.UserID, encryptedEmail, emailHash)
-	return stacktrace.Propagate(err, "failed to update email")
+	if err != nil {
+		return stacktrace.Propagate(err, "failed to update email")
+	}
+	err = c.DataCleanupRepo.RemoveScheduledDelete(ctx, req.UserID)
+	if err != nil {
+		logrus.WithError(err).Error("failed to remove scheduled delete")
+		return stacktrace.Propagate(err, "")
+	}
+	return stacktrace.Propagate(err, "")
 }
 
 func (c *UserController) attachFreeSubscription(userID int64) (ente.Subscription, error) {

server/pkg/repo/datacleanup/repository.go

@@ -3,7 +3,7 @@ package datacleanup
 import (
 	"context"
 	"database/sql"
-
+	"fmt"
 	entity "github.com/ente-io/museum/ente/data_cleanup"
 	"github.com/ente-io/museum/pkg/utils/time"
 	"github.com/ente-io/stacktrace"
@@ -19,6 +19,21 @@ func (r *Repository) Insert(ctx context.Context, userID int64) error {
 	return stacktrace.Propagate(err, "failed to insert")
 }
 
+func (r *Repository) RemoveScheduledDelete(ctx context.Context, userID int64) error {
+	res, execErr := r.DB.ExecContext(ctx, `DELETE from data_cleanup where user_id= $1 and stage = $2`, userID, entity.Scheduled)
+	if execErr != nil {
+		return execErr
+	}
+	affected, affErr := res.RowsAffected()
+	if affErr != nil {
+		return affErr
+	}
+	if affected != 1 {
+		return fmt.Errorf("only one row should have been affected, got %d", affected)
+	}
+	return nil
+}
+
 func (r *Repository) GetItemsPendingCompletion(ctx context.Context, limit int) ([]*entity.DataCleanup, error) {
 	rows, err := r.DB.QueryContext(ctx, `SELECT user_id, stage, stage_schedule_time, stage_attempt_count, created_at, updated_at  from  data_cleanup 
          where stage != $1 and stage_schedule_time < now_utc_micro_seconds()