feat: custom metadata on upload

migrations/tenant/0025-custom-metadata.sql

@@ -0,0 +1,2 @@
+ALTER TABLE storage.objects ADD COLUMN user_metadata jsonb NULL;
+ALTER TABLE storage.s3_multipart_uploads ADD COLUMN metadata jsonb NULL;

package.json

@@ -44,9 +44,9 @@
     "@opentelemetry/instrumentation-pino": "^0.39.0",
     "@shopify/semaphore": "^3.0.2",
     "@smithy/node-http-handler": "^2.3.1",
-    "@tus/file-store": "1.3.1",
-    "@tus/s3-store": "1.4.1",
-    "@tus/server": "1.4.1",
+    "@tus/file-store": "1.4.0",
+    "@tus/s3-store": "1.5.0",
+    "@tus/server": "1.7.0",
     "agentkeepalive": "^4.5.0",
     "ajv": "^8.12.0",
     "async-retry": "^1.3.3",

src/http/routes/object/getObjectInfo.ts

@@ -30,7 +30,8 @@ async function requestHandler(
     getObjectRequestInterface,
     unknown
   >,
-  publicRoute = false
+  publicRoute = false,
+  method: 'head' | 'info' = 'head'
 ) {
   const { bucketName } = request.params
   const objectName = request.params['*']
@@ -42,15 +43,21 @@ async function requestHandler(
     await request.storage.asSuperUser().findBucket(bucketName, 'id', {
       isPublic: true,
     })
-    obj = await request.storage.asSuperUser().from(bucketName).findObject(objectName, 'id,version')
+    obj = await request.storage
+      .asSuperUser()
+      .from(bucketName)
+      .findObject(objectName, 'id,version,metadata,user_metadata,created_at')
   } else {
-    obj = await request.storage.from(bucketName).findObject(objectName, 'id,version')
+    obj = await request.storage
+      .from(bucketName)
+      .findObject(objectName, 'id,version,metadata,user_metadata,created_at')
   }
 
-  return request.storage.renderer('head').render(request, response, {
+  return request.storage.renderer(method).render(request, response, {
     bucket: storageS3Bucket,
     key: s3Key,
     version: obj.version,
+    object: obj,
   })
 }
 
@@ -90,7 +97,7 @@ export async function publicRoutes(fastify: FastifyInstance) {
       },
     },
     async (request, response) => {
-      return requestHandler(request, response, true)
+      return requestHandler(request, response, true, 'info')
     }
   )
 }
@@ -131,7 +138,7 @@ export async function authenticatedRoutes(fastify: FastifyInstance) {
       },
     },
     async (request, response) => {
-      return requestHandler(request, response)
+      return requestHandler(request, response, false, 'info')
     }
   )
 
@@ -151,7 +158,7 @@ export async function authenticatedRoutes(fastify: FastifyInstance) {
       },
     },
     async (request, response) => {
-      return requestHandler(request, response)
+      return requestHandler(request, response, false, 'info')
     }
   )
 

src/http/routes/s3/commands/create-multipart-upload.ts

@@ -21,6 +21,7 @@ const CreateMultiPartUploadInput = {
   },
   Headers: {
     type: 'object',
+    additionalProperties: true,
     properties: {
       authorization: { type: 'string' },
       'content-type': { type: 'string' },
@@ -39,13 +40,23 @@ export default function CreateMultipartUpload(s3Router: S3Router) {
     (req, ctx) => {
       const s3Protocol = new S3ProtocolHandler(ctx.storage, ctx.tenantId, ctx.owner)
 
+      let metadata: undefined | Record<string, any> = undefined
+
+      Object.keys(req.Headers)
+        .filter((key) => key.startsWith('x-amz-meta-'))
+        .forEach((key) => {
+          if (!metadata) metadata = {}
+          metadata[key.replace('x-amz-meta-', '')] = req.Headers[key]
+        })
+
       return s3Protocol.createMultiPartUpload({
         Bucket: req.Params.Bucket,
         Key: req.Params['*'],
         ContentType: req.Headers?.['content-type'],
         CacheControl: req.Headers?.['cache-control'],
         ContentDisposition: req.Headers?.['content-disposition'],
         ContentEncoding: req.Headers?.['content-encoding'],
+        Metadata: metadata,
       })
     }
   )

src/http/routes/s3/commands/upload-part.ts

@@ -94,6 +94,16 @@ export default function UploadPart(s3Router: S3Router) {
     },
     (req, ctx) => {
       const s3Protocol = new S3ProtocolHandler(ctx.storage, ctx.tenantId, ctx.owner)
+
+      let metadata: undefined | Record<string, any> = undefined
+
+      Object.keys(req.Headers)
+        .filter((key) => key.startsWith('x-amz-meta-'))
+        .forEach((key) => {
+          if (!metadata) metadata = {}
+          metadata[key.replace('x-amz-meta-', '')] = req.Headers[key]
+        })
+
       return s3Protocol.putObject({
         Body: ctx.req as any,
         Bucket: req.Params.Bucket,
@@ -102,6 +112,7 @@ export default function UploadPart(s3Router: S3Router) {
         ContentType: req.Headers?.['content-type'],
         Expires: req.Headers?.['expires'] ? new Date(req.Headers?.['expires']) : undefined,
         ContentEncoding: req.Headers?.['content-encoding'],
+        Metadata: metadata,
       })
     }
   )

src/http/routes/tus/lifecycle.ts

@@ -154,7 +154,7 @@ export async function onCreate(
   rawReq: http.IncomingMessage,
   res: http.ServerResponse,
   upload: Upload
-): Promise<http.ServerResponse> {
+): Promise<{ res: http.ServerResponse; metadata?: Upload['metadata'] }> {
   const uploadID = UploadId.fromString(upload.id)
 
   const req = rawReq as MultiPartRequest
@@ -166,17 +166,21 @@ export async function onCreate(
 
   const uploader = new Uploader(storage.backend, storage.db)
 
-  if (upload.metadata && /^-?\d+$/.test(upload.metadata.cacheControl || '')) {
-    upload.metadata.cacheControl = `max-age=${upload.metadata.cacheControl}`
-  } else if (upload.metadata) {
-    upload.metadata.cacheControl = 'no-cache'
+  const metadata = {
+    ...(upload.metadata ? upload.metadata : {}),
   }
 
-  if (upload.metadata?.contentType && bucket.allowed_mime_types) {
-    uploader.validateMimeType(upload.metadata.contentType, bucket.allowed_mime_types)
+  if (/^-?\d+$/.test(metadata.cacheControl || '')) {
+    metadata.cacheControl = `max-age=${metadata.cacheControl}`
+  } else if (metadata) {
+    metadata.cacheControl = 'no-cache'
   }
 
-  return res
+  if (metadata?.contentType && bucket.allowed_mime_types) {
+    uploader.validateMimeType(metadata.contentType, bucket.allowed_mime_types)
+  }
+
+  return { res, metadata }
 }
 
 /**
@@ -199,6 +203,14 @@ export async function onUploadFinish(
     )
 
     const uploader = new Uploader(req.upload.storage.backend, req.upload.storage.db)
+    let customMd: undefined | Record<string, string> = undefined
+    if (upload.metadata?.userMetadata) {
+      try {
+        customMd = JSON.parse(upload.metadata.userMetadata)
+      } catch (e) {
+        // no-op
+      }
+    }
 
     await uploader.completeUpload({
       version: resourceId.version,
@@ -208,6 +220,7 @@ export async function onUploadFinish(
       isUpsert: req.upload.isUpsert,
       uploadType: 'resumable',
       owner: req.upload.owner,
+      userMetadata: customMd,
     })
 
     res.setHeader('Tus-Complete', '1')

src/internal/database/connection.ts

@@ -201,7 +201,13 @@ export class TenantConnection {
           // This should never be reached, since the above promise is always rejected in this edge case.
           throw ERRORS.DatabaseError('Transaction already completed')
         }
-        await tnx.raw(`SELECT set_config('search_path', ?, true)`, [searchPath.join(', ')])
+
+        try {
+          await tnx.raw(`SELECT set_config('search_path', ?, true)`, [searchPath.join(', ')])
+        } catch (e) {
+          await tnx.rollback()
+          throw e
+        }
       }
 
       return tnx

src/internal/errors/codes.ts

@@ -208,12 +208,12 @@ export const ERRORS = {
       message: `invalid range provided`,
     }),
 
-  EntityTooLarge: (e?: Error) =>
+  EntityTooLarge: (e?: Error, entity = 'object') =>
     new StorageBackendError({
       error: 'Payload too large',
       code: ErrorCode.EntityTooLarge,
       httpStatusCode: 413,
-      message: 'The object exceeded the maximum allowed size',
+      message: `The ${entity} exceeded the maximum allowed size`,
       originalError: e,
     }),
 

src/internal/queue/queue.ts

@@ -53,10 +53,6 @@ export abstract class Queue {
       url = multitenantDatabaseUrl
     }
 
-    console.log({
-      deleteAfterDays: pgQueueDeleteAfterDays,
-      deleteAfterHours: pgQueueDeleteAfterHours,
-    })
     Queue.pgBoss = new PgBoss({
       connectionString: url,
       db: new QueueDB({