list objects in worker when emptying bucket

Author: itslenny

src/http/routes/bucket/emptyBucket.ts

@@ -14,7 +14,10 @@ const emptyBucketParamsSchema = {
 const successResponseSchema = {
   type: 'object',
   properties: {
-    message: { type: 'string', examples: ['Successfully emptied'] },
+    message: {
+      type: 'string',
+      examples: ['Empty bucket has been queued. Completion may take up to an hour.'],
+    },
   },
 }
 interface emptyBucketRequestInterface extends AuthenticatedRequest {
@@ -41,7 +44,9 @@ export default async function routes(fastify: FastifyInstance) {
 
       await request.storage.emptyBucket(bucketId)
 
-      return response.status(200).send(createResponse('Successfully emptied'))
+      return response
+        .status(200)
+        .send(createResponse('Empty bucket has been queued. Completion may take up to an hour.'))
     }
   )
 }

src/internal/errors/codes.ts

@@ -74,14 +74,14 @@ export const ERRORS = {
       code: ErrorCode.InvalidRequest,
       httpStatusCode: 409,
       message: `The feature ${feature} is not enabled for this resource`,
+      originalError: e,
     }),
-  UnableToEmptyBucket: (bucket: string, e?: Error) =>
+  UnableToEmptyBucket: (bucket: string) =>
     new StorageBackendError({
       code: ErrorCode.InvalidRequest,
       resource: bucket,
       httpStatusCode: 409,
       message: `Unable to empty the bucket because it contains too many objects`,
-      originalError: e,
     }),
   NoSuchBucket: (bucket: string, e?: Error) =>
     new StorageBackendError({

src/storage/events/index.ts

@@ -5,8 +5,8 @@ export * from './lifecycle/bucket-deleted'
 export * from './lifecycle/object-created'
 export * from './lifecycle/object-updated'
 export * from './lifecycle/object-removed'
-export * from './lifecycle/object-admin-delete'
-export * from './lifecycle/object-admin-delete-batch'
+export * from './objects/object-admin-delete'
+export * from './objects/object-admin-delete-all-before'
 export * from './objects/backup-object'
 export * from './migrations/run-migrations'
 export * from './migrations/reset-migrations'

src/storage/events/lifecycle/object-admin-delete-batch.ts

@@ -0,0 +1,141 @@
+import { BaseEvent } from '../base-event'
+import { getConfig } from '../../../config'
+import { Job, SendOptions, WorkOptions } from 'pg-boss'
+import { logger, logSchema } from '@internal/monitoring'
+import { Storage } from '../../index'
+import { BasePayload } from '@internal/queue'
+import { withOptionalVersion } from '@storage/backend'
+
+const DELETE_JOB_TIME_LIMIT_MS = 10_000
+
+export interface ObjectDeleteAllBeforeEvent extends BasePayload {
+  before: string
+  bucketId: string
+}
+
+const { storageS3Bucket, requestUrlLengthLimit } = getConfig()
+
+export class ObjectAdminDeleteAllBefore extends BaseEvent<ObjectDeleteAllBeforeEvent> {
+  static queueName = 'object:admin:delete-all-before'
+
+  static getWorkerOptions(): WorkOptions {
+    return {}
+  }
+
+  static getSendOptions(payload: ObjectDeleteAllBeforeEvent): SendOptions {
+    return {
+      singletonKey: `${payload.tenant.ref}/${payload.bucketId}`,
+      priority: 10,
+      expireInSeconds: 30,
+    }
+  }
+
+  static async handle(job: Job<ObjectDeleteAllBeforeEvent>) {
+    let storage: Storage | undefined = undefined
+
+    const tenantId = job.data.tenant.ref
+    const bucketId = job.data.bucketId
+    const before = new Date(job.data.before)
+
+    try {
+      storage = await this.createStorage(job.data)
+
+      logSchema.event(
+        logger,
+        `[Admin]: ObjectAdminDeleteAllBefore ${bucketId} ${before.toUTCString()}`,
+        {
+          jodId: job.id,
+          type: 'event',
+          event: 'ObjectAdminDeleteAllBefore',
+          payload: JSON.stringify(job.data),
+          objectPath: bucketId,
+          tenantId,
+          project: tenantId,
+          reqId: job.data.reqId,
+        }
+      )
+
+      const batchLimit = Math.floor(requestUrlLengthLimit / (36 + 3))
+
+      let moreObjectsToDelete = false
+      const start = Date.now()
+      while (Date.now() - start < DELETE_JOB_TIME_LIMIT_MS) {
+        moreObjectsToDelete = false
+        const objects = await storage.db.listObjects(bucketId, 'id, name', batchLimit + 1, before)
+
+        const backend = storage.backend
+        if (objects && objects.length > 0) {
+          if (objects.length > batchLimit) {
+            objects.pop()
+            moreObjectsToDelete = true
+          }
+
+          await storage.db.withTransaction(async (trx) => {
+            const deleted = await trx.deleteObjects(
+              bucketId,
+              objects.map(({ id }) => id!),
+              'id'
+            )
+
+            if (deleted && deleted.length > 0) {
+              const prefixes: string[] = []
+
+              for (const { name, version } of deleted) {
+                const fileName = withOptionalVersion(`${tenantId}/${bucketId}/${name}`, version)
+                prefixes.push(fileName)
+                prefixes.push(fileName + '.info')
+              }
+
+              await backend.deleteObjects(storageS3Bucket, prefixes)
+            }
+          })
+        }
+
+        if (!moreObjectsToDelete) {
+          break
+        }
+      }
+
+      if (moreObjectsToDelete) {
+        // delete next batch
+        await ObjectAdminDeleteAllBefore.send({
+          before,
+          bucketId,
+          tenant: job.data.tenant,
+          reqId: job.data.reqId,
+        })
+      }
+    } catch (e) {
+      logger.error(
+        {
+          error: e,
+          jodId: job.id,
+          type: 'event',
+          event: 'ObjectAdminDeleteAllBefore',
+          payload: JSON.stringify(job.data),
+          objectPath: bucketId,
+          tenantId,
+          project: tenantId,
+          reqId: job.data.reqId,
+        },
+        `[Admin]: ObjectAdminDeleteAllBefore ${bucketId} ${before.toUTCString()} - FAILED`
+      )
+      throw e
+    } finally {
+      if (storage) {
+        const tenant = storage.db.tenant()
+        storage.db
+          .destroyConnection()
+          .then(() => {
+            // no-op
+          })
+          .catch((e) => {
+            logger.error(
+              { error: e },
+              `[Admin]: ObjectAdminDeleteAllBefore ${tenant.ref} - FAILED DISPOSING CONNECTION`
+            )
+          })
+      }
+    }
+  }
+}

src/storage/events/objects/object-admin-delete-all-before.ts

@@ -1,17 +1,17 @@
 import { Queue } from '@internal/queue'
 import { Webhook } from './lifecycle/webhook'
-import { ObjectAdminDelete } from './lifecycle/object-admin-delete'
+import { ObjectAdminDelete } from './objects/object-admin-delete'
 import { RunMigrationsOnTenants } from './migrations/run-migrations'
 import { BackupObjectEvent } from './objects/backup-object'
 import { ResetMigrationsOnTenant } from './migrations/reset-migrations'
 import { JwksCreateSigningSecret } from './jwks/jwks-create-signing-secret'
 import { UpgradePgBossV10 } from './pgboss/upgrade-v10'
-import { ObjectAdminDeleteBatch } from './lifecycle/object-admin-delete-batch'
+import { ObjectAdminDeleteAllBefore } from './objects/object-admin-delete-all-before'
 
 export function registerWorkers() {
   Queue.register(Webhook)
   Queue.register(ObjectAdminDelete)
-  Queue.register(ObjectAdminDeleteBatch)
+  Queue.register(ObjectAdminDeleteAllBefore)
   Queue.register(RunMigrationsOnTenants)
   Queue.register(BackupObjectEvent)
   Queue.register(ResetMigrationsOnTenant)

src/storage/events/lifecycle/object-admin-delete.ts renamed to src/storage/events/objects/object-admin-delete.ts

@@ -17,9 +17,9 @@ import { StorageObjectLocator } from '@storage/locator'
 import { BucketCreatedEvent, BucketDeleted } from '@storage/events'
 import { tenantHasMigrations } from '@internal/database/migrations'
 import { tenantHasFeature } from '@internal/database'
-import { ObjectAdminDeleteBatch } from './events'
+import { ObjectAdminDeleteAllBefore } from './events'
 
-const { requestUrlLengthLimit, emptyBucketMax } = getConfig()
+const { emptyBucketMax } = getConfig()
 
 /**
  * Storage
@@ -256,66 +256,34 @@ export class Storage {
   /**
    * Deletes all files in a bucket
    * @param bucketId
+   * @param before limit to files before the specified time (defaults to now)
    */
-  async emptyBucket(bucketId: string) {
+  async emptyBucket(bucketId: string, before: Date = new Date()) {
     await this.findBucket(bucketId, 'name')
 
     const count = await this.db.countObjectsInBucket(bucketId, emptyBucketMax + 1)
     if (count > emptyBucketMax) {
       throw ERRORS.UnableToEmptyBucket(bucketId)
     }
 
-    while (true) {
-      const objects = await this.db.listObjects(
-        bucketId,
-        'id, name',
-        Math.floor(requestUrlLengthLimit / (36 + 3))
-      )
-
-      if (!(objects && objects.length > 0)) {
-        break
-      }
-
-      const deleted = await this.db.deleteObjects(
-        bucketId,
-        objects.map(({ id }) => id!),
-        'id'
-      )
-
-      if (deleted && deleted.length > 0) {
-        const prefixes = deleted.reduce((all, { name, version }) => {
-          const fileName = this.location.getKeyLocation({
-            tenantId: this.db.tenantId,
-            bucketId,
-            objectName: name,
-            version,
-          })
-          all.push(fileName)
-          all.push(fileName + '.info')
-          return all
-        }, [] as string[])
-        // delete files from s3 asynchronously
-        await ObjectAdminDeleteBatch.send({
-          prefixes,
-          bucketId,
-          tenant: this.db.tenant(),
-          reqId: this.db.reqId,
-        })
-      }
+    const objects = await this.db.listObjects(bucketId, 'id, name', 1, before)
+    if (!objects || objects.length < 1) {
+      // the bucket is already empty
+      return
+    }
 
-      if (deleted?.length !== objects.length) {
-        const deletedNames = new Set(deleted?.map(({ name }) => name))
-        const remainingNames = objects
-          .filter(({ name }) => !deletedNames.has(name))
-          .map(({ name }) => name)
+    // ensure delete permissions
+    await this.db.testPermission((db) => {
+      return db.deleteObject(bucketId, objects[0].id!)
+    })
 
-        throw ERRORS.AccessDenied(
-          `Cannot delete: ${remainingNames.join(
-            ' ,'
-          )}, you may have SELECT but not DELETE permissions`
-        )
-      }
-    }
+    // use queue to recursively delete all objects created before the specified time
+    await ObjectAdminDeleteAllBefore.send({
+      before,
+      bucketId,
+      tenant: this.db.tenant(),
+      reqId: this.db.reqId,
+    })
   }
 
   validateMimeType(mimeType: string[]) {