Improved adapter interface and explicit signal cancellation

src/http/plugins/db.ts

@@ -25,8 +25,7 @@ declare module 'fastify' {
   }
 }
 
-const { databaseEnableQueryCancellation, dbMigrationStrategy, isMultitenant, dbMigrationFreezeAt } =
-  getConfig()
+const { dbMigrationStrategy, isMultitenant, dbMigrationFreezeAt } = getConfig()
 
 export const db = fastifyPlugin(
   async function db(fastify) {
@@ -55,11 +54,6 @@ export const db = fastifyPlugin(
         method: request.method,
         operation: () => request.operation?.type,
       })
-
-      // Connect abort signal to DB connection for query cancellation
-      if (request.signals?.disconnect?.signal && databaseEnableQueryCancellation) {
-        request.db.setAbortSignal(request.signals.disconnect.signal)
-      }
     })
 
     fastify.addHook('onSend', async (request, reply, payload) => {
@@ -124,11 +118,6 @@ export const dbSuperUser = fastifyPlugin<DbSuperUserPluginOptions>(
         maxConnections: opts.maxConnections,
         operation: () => request.operation?.type,
       })
-
-      // Connect abort signal to DB connection for query cancellation
-      if (request.signals?.disconnect?.signal && databaseEnableQueryCancellation) {
-        request.db.setAbortSignal(request.signals.disconnect.signal)
-      }
     })
 
     fastify.addHook('onSend', async (request, reply, payload) => {

src/http/routes/bucket/createBucket.ts

@@ -80,6 +80,7 @@ export default async function routes(fastify: FastifyInstance) {
         allowedMimeTypes: allowed_mime_types
           ? allowed_mime_types?.filter((mime) => mime)
           : allowed_mime_types,
+        signal: request.signals.disconnect.signal,
       })
 
       return response.status(200).send({

src/http/routes/bucket/deleteBucket.ts

@@ -39,7 +39,7 @@ export default async function routes(fastify: FastifyInstance) {
     },
     async (request, response) => {
       const { bucketId } = request.params
-      await request.storage.deleteBucket(bucketId)
+      await request.storage.deleteBucket({ bucketId, signal: request.signals.disconnect.signal })
 
       return response.status(200).send(createResponse('Successfully deleted'))
     }

src/http/routes/bucket/emptyBucket.ts

@@ -42,7 +42,7 @@ export default async function routes(fastify: FastifyInstance) {
     async (request, response) => {
       const { bucketId } = request.params
 
-      await request.storage.emptyBucket(bucketId)
+      await request.storage.emptyBucket({ bucketId, signal: request.signals.disconnect.signal })
 
       return response
         .status(200)

src/http/routes/bucket/getAllBuckets.ts

@@ -70,11 +70,13 @@ export default async function routes(fastify: FastifyInstance) {
         isClientVersionBefore('supabase-py', clientInfo, '2.18.0') ||
         isClientVersionBefore('storage3', userAgent, '0.12.1')
 
-      const results = await request.storage.listBuckets(
-        'id, name, public, owner, created_at, updated_at, file_size_limit, allowed_mime_types' +
+      const results = await request.storage.listBuckets({
+        columns:
+          'id, name, public, owner, created_at, updated_at, file_size_limit, allowed_mime_types' +
           (omitBucketType ? '' : ', type'),
-        { limit, offset, sortColumn, sortOrder, search }
-      )
+        options: { limit, offset, sortColumn, sortOrder, search },
+        signal: request.signals.disconnect.signal,
+      })
 
       return response.send(results)
     }

src/http/routes/bucket/getBucket.ts

@@ -36,10 +36,12 @@ export default async function routes(fastify: FastifyInstance) {
     async (request, response) => {
       const { bucketId } = request.params
 
-      const results = await request.storage.findBucket(
-        bucketId,
-        'id, name, owner, public, created_at, updated_at, file_size_limit, allowed_mime_types'
-      )
+      const results = await request.storage.findBucket({
+        bucketId: bucketId,
+        columns:
+          'id, name, owner, public, created_at, updated_at, file_size_limit, allowed_mime_types',
+        signal: request.signals.disconnect.signal,
+      })
 
       return response.send(results)
     }

src/http/routes/bucket/updateBucket.ts

@@ -61,12 +61,16 @@ export default async function routes(fastify: FastifyInstance) {
 
       const { public: isPublic, file_size_limit, allowed_mime_types } = request.body
 
-      await request.storage.updateBucket(bucketId, {
-        public: isPublic,
-        fileSizeLimit: file_size_limit,
-        allowedMimeTypes: allowed_mime_types
-          ? allowed_mime_types?.filter((mime) => mime)
-          : allowed_mime_types,
+      await request.storage.updateBucket({
+        bucketId,
+        data: {
+          public: isPublic,
+          fileSizeLimit: file_size_limit,
+          allowedMimeTypes: allowed_mime_types
+            ? allowed_mime_types?.filter((mime) => mime)
+            : allowed_mime_types,
+        },
+        signal: request.signals.disconnect.signal,
       })
 
       return response.status(200).send(createResponse('Successfully updated'))

src/http/routes/iceberg/bucket.ts

@@ -58,7 +58,10 @@ export default async function routes(fastify: FastifyInstance) {
     },
     async (request, response) => {
       const { bucketName } = request.params
-      await request.storage.deleteIcebergBucket(bucketName)
+      await request.storage.deleteIcebergBucket({
+        name: bucketName,
+        signal: request.signals.disconnect.signal,
+      })
 
       return response.status(200).send(createResponse('Successfully deleted'))
     }
@@ -80,6 +83,7 @@ export default async function routes(fastify: FastifyInstance) {
       const { name } = request.body
       const bucket = await request.storage.createIcebergBucket({
         name,
+        signal: request.signals.disconnect.signal,
       })
 
       return response.status(200).send({
@@ -106,12 +110,16 @@ export default async function routes(fastify: FastifyInstance) {
     async (request, response) => {
       const query = request.query
 
-      const bucket = await request.storage.listAnalyticsBuckets('name,created_at,updated_at', {
-        limit: query.limit,
-        offset: query.offset,
-        sortColumn: query.sortColumn,
-        sortOrder: query.sortOrder,
-        search: query.search,
+      const bucket = await request.storage.listAnalyticsBuckets({
+        columns: 'name,created_at,updated_at',
+        options: {
+          limit: query.limit,
+          offset: query.offset,
+          sortColumn: query.sortColumn,
+          sortOrder: query.sortOrder,
+          search: query.search,
+        },
+        signal: request.signals.disconnect.signal,
       })
 
       return response.status(200).send(

src/http/routes/object/copyObject.ts

@@ -74,6 +74,7 @@ export default async function routes(fastify: FastifyInstance) {
         metadata: metadata,
         copyMetadata: request.body.copyMetadata ?? true,
         upsert: request.headers['x-upsert'] === 'true',
+        signal: request.signals.disconnect.signal,
       })
 
       return response.status(result.httpStatusCode ?? 200).send({

src/http/routes/object/deleteObject.ts

@@ -43,7 +43,9 @@ export default async function routes(fastify: FastifyInstance) {
       const { bucketName } = request.params
       const objectName = request.params['*']
 
-      await request.storage.from(bucketName).deleteObject(objectName)
+      await request.storage
+        .from(bucketName)
+        .deleteObject({ objectName, signal: request.signals.disconnect.signal })
 
       return response.status(200).send(createResponse('Successfully deleted'))
     }

src/http/routes/object/deleteObjects.ts

@@ -58,7 +58,9 @@ export default async function routes(fastify: FastifyInstance) {
       const { bucketName } = request.params
       const prefixes = request.body['prefixes']
 
-      const results = await request.storage.from(bucketName).deleteObjects(prefixes)
+      const results = await request.storage
+        .from(bucketName)
+        .deleteObjects({ prefixes, signal: request.signals.disconnect.signal })
 
       return response.status(200).send(results)
     }

src/http/routes/object/getObject.ts

@@ -50,8 +50,13 @@ async function requestHandler(
     bucketId: bucketName,
     objectName,
   })
-  const bucket = await request.storage.asSuperUser().findBucket(bucketName, 'id,public', {
-    dontErrorOnEmpty: true,
+  const bucket = await request.storage.asSuperUser().findBucket({
+    bucketId: bucketName,
+    columns: 'id,public',
+    filters: {
+      dontErrorOnEmpty: true,
+    },
+    signal: request.signals.disconnect.signal,
   })
 
   // The request is not authenticated
@@ -71,13 +76,18 @@ async function requestHandler(
 
   if (bucket.public) {
     // request is authenticated but we still use the superUser as we don't need to check RLS
-    obj = await request.storage
-      .asSuperUser()
-      .from(bucketName)
-      .findObject(objectName, 'id, version, metadata')
+    obj = await request.storage.asSuperUser().from(bucketName).findObject({
+      objectName,
+      columns: 'id, version, metadata',
+      signal: request.signals.disconnect.signal,
+    })
   } else {
     // request is authenticated use RLS
-    obj = await request.storage.from(bucketName).findObject(objectName, 'id, version, metadata')
+    obj = await request.storage.from(bucketName).findObject({
+      objectName,
+      columns: 'id, version, metadata',
+      signal: request.signals.disconnect.signal,
+    })
   }
 
   return request.storage.renderer('asset').render(request, response, {

src/http/routes/object/getObjectInfo.ts

@@ -43,8 +43,13 @@ async function requestHandler(
     objectName,
   })
 
-  const bucket = await request.storage.asSuperUser().findBucket(bucketName, 'id,public', {
-    dontErrorOnEmpty: true,
+  const bucket = await request.storage.asSuperUser().findBucket({
+    bucketId: bucketName,
+    columns: 'id,public',
+    filters: {
+      dontErrorOnEmpty: true,
+    },
+    signal: request.signals.disconnect.signal,
   })
 
   // Not Authenticated flow
@@ -62,20 +67,17 @@ async function requestHandler(
   let obj: Obj
 
   if (bucket.public || publicRoute) {
-    obj = await request.storage
-      .asSuperUser()
-      .from(bucketName)
-      .findObject(
-        objectName,
-        'id,name,version,bucket_id,metadata,user_metadata,updated_at,created_at'
-      )
+    obj = await request.storage.asSuperUser().from(bucketName).findObject({
+      objectName,
+      columns: 'id,name,version,bucket_id,metadata,user_metadata,updated_at,created_at',
+      signal: request.signals.disconnect.signal,
+    })
   } else {
-    obj = await request.storage
-      .from(bucketName)
-      .findObject(
-        objectName,
-        'id,name,version,bucket_id,metadata,user_metadata,updated_at,created_at'
-      )
+    obj = await request.storage.from(bucketName).findObject({
+      objectName,
+      columns: 'id,name,version,bucket_id,metadata,user_metadata,updated_at,created_at',
+      signal: request.signals.disconnect.signal,
+    })
   }
 
   return request.storage.renderer(method).render(request, response, {

src/http/routes/object/getPublicObject.ts

@@ -54,10 +54,19 @@ export default async function routes(fastify: FastifyInstance) {
 
       const bucketRef = request.storage.asSuperUser().from(bucketName)
       const [, obj] = await Promise.all([
-        request.storage.asSuperUser().findBucket(bucketName, 'id,public', {
-          isPublic: true,
+        request.storage.asSuperUser().findBucket({
+          bucketId: bucketName,
+          columns: 'id,public',
+          filters: {
+            isPublic: true,
+          },
+          signal: request.signals.disconnect.signal,
+        }),
+        bucketRef.findObject({
+          objectName,
+          columns: 'id,version,metadata',
+          signal: request.signals.disconnect.signal,
         }),
-        bucketRef.findObject(objectName, 'id,version,metadata'),
       ])
 
       // send the object from s3

src/http/routes/object/getSignedObject.ts

@@ -83,7 +83,11 @@ export default async function routes(fastify: FastifyInstance) {
       const obj = await request.storage
         .asSuperUser()
         .from(bucketName)
-        .findObject(objParts.join('/'), 'id,version,metadata')
+        .findObject({
+          objectName: objParts.join('/'),
+          columns: 'id,version,metadata',
+          signal: request.signals.disconnect.signal,
+        })
 
       return request.storage.renderer('asset').render(request, response, {
         bucket: storageS3Bucket,

src/http/routes/object/getSignedURL.ts

@@ -80,9 +80,13 @@ export default async function routes(fastify: FastifyInstance) {
           }
         : undefined
 
-      const signedURL = await request.storage
-        .from(bucketName)
-        .signObjectUrl(objectName, urlPath as string, expiresIn, transformationOptions)
+      const signedURL = await request.storage.from(bucketName).signObjectUrl({
+        objectName,
+        url: urlPath as string,
+        expiresIn,
+        metadata: transformationOptions,
+        signal: request.signals.disconnect.signal,
+      })
 
       return response.status(200).send({ signedURL })
     }

src/http/routes/object/getSignedURLs.ts

@@ -78,7 +78,9 @@ export default async function routes(fastify: FastifyInstance) {
       const { bucketName } = request.params
       const { expiresIn, paths } = request.body
 
-      const signedURLs = await request.storage.from(bucketName).signObjectUrls(paths, expiresIn)
+      const signedURLs = await request.storage
+        .from(bucketName)
+        .signObjectUrls({ paths, expiresIn, signal: request.signals.disconnect.signal })
 
       return response.status(200).send(signedURLs)
     }

src/http/routes/object/getSignedUploadURL.ts

@@ -69,11 +69,16 @@ export default async function routes(fastify: FastifyInstance) {
 
       const urlPath = `${bucketName}/${objectName}`
 
-      const signedUpload = await request.storage
-        .from(bucketName)
-        .signUploadObjectUrl(objectName, urlPath as string, uploadSignedUrlExpirationTime, owner, {
+      const signedUpload = await request.storage.from(bucketName).signUploadObjectUrl({
+        objectName,
+        url: urlPath as string,
+        expiresIn: uploadSignedUrlExpirationTime,
+        owner,
+        options: {
           upsert: request.headers['x-upsert'] === 'true',
-        })
+        },
+        signal: request.signals.disconnect.signal,
+      })
 
       return response.status(200).send({ url: signedUpload.url, token: signedUpload.token })
     }

src/http/routes/object/listObjects.ts

@@ -70,14 +70,18 @@ export default async function routes(fastify: FastifyInstance) {
       const { bucketName } = request.params
       const { limit, offset, sortBy, search, prefix } = request.body
 
-      const results = await request.storage.from(bucketName).searchObjects(prefix, {
-        limit,
-        offset,
-        search,
-        sortBy: {
-          column: sortBy?.column,
-          order: sortBy?.order,
+      const results = await request.storage.from(bucketName).searchObjects({
+        prefix,
+        options: {
+          limit,
+          offset,
+          search,
+          sortBy: {
+            column: sortBy?.column,
+            order: sortBy?.order,
+          },
         },
+        signal: request.signals.disconnect.signal,
       })
 
       return response.status(200).send(results)

src/http/routes/object/listObjectsV2.ts

@@ -81,6 +81,7 @@ export default async function routes(fastify: FastifyInstance) {
         maxKeys: limit,
         cursor,
         sortBy,
+        signal: request.signals.disconnect.signal,
       })
 
       return response.status(200).send(results)