feat: add user metadata support for RLS policies

migrations/tenant/0056-s3-multipart-uploads-metadata.sql

@@ -0,0 +1 @@
+ALTER TABLE storage.s3_multipart_uploads ADD COLUMN IF NOT EXISTS metadata jsonb NULL;

src/http/routes/object/getSignedUploadURL.ts

@@ -4,6 +4,7 @@ import { createDefaultSchema } from '../../routes-helper'
 import { AuthenticatedRequest } from '../../types'
 import { getConfig } from '../../../config'
 import { ROUTE_OPERATIONS } from '../operations'
+import { parseUserMetadata } from '../../../storage/uploader'
 
 const { uploadSignedUrlExpirationTime } = getConfig()
 
@@ -20,6 +21,8 @@ const getSignedUploadURLHeadersSchema = {
   type: 'object',
   properties: {
     'x-upsert': { type: 'string' },
+    'content-type': { type: 'string' },
+    'content-length': { type: 'string' },
     authorization: { type: 'string' },
   },
   required: ['authorization'],
@@ -69,10 +72,27 @@ export default async function routes(fastify: FastifyInstance) {
 
       const urlPath = `${bucketName}/${objectName}`
 
+      let userMetadata: Record<string, unknown> | undefined
+
+      const customMd = request.headers['x-metadata']
+
+      if (typeof customMd === 'string') {
+        userMetadata = parseUserMetadata(customMd)
+      }
+
+      const contentType = request.headers['content-type']
+      const contentLengthHeader = request.headers['content-length']
+      const contentLength = contentLengthHeader ? Number(contentLengthHeader) : undefined
+
       const signedUpload = await request.storage
         .from(bucketName)
         .signUploadObjectUrl(objectName, urlPath as string, uploadSignedUrlExpirationTime, owner, {
           upsert: request.headers['x-upsert'] === 'true',
+          userMetadata: userMetadata,
+          metadata: {
+            mimetype: contentType,
+            contentLength: contentLength,
+          },
         })
 
       return response.status(200).send({ url: signedUpload.url, token: signedUpload.token })

src/http/routes/tus/lifecycle.ts

@@ -1,6 +1,6 @@
 import http from 'http'
 import { BaseLogger } from 'pino'
-import { Upload } from '@tus/server'
+import { Metadata, Upload } from '@tus/server'
 import { randomUUID } from 'crypto'
 import { TenantConnection } from '@internal/database'
 import { ERRORS, isRenderableError } from '@internal/errors'
@@ -67,6 +67,20 @@ export async function onIncomingRequest(rawReq: Request, id: string) {
 
   req.upload.resources = [`${uploadID.bucket}/${uploadID.objectName}`]
 
+  let customMd: Record<string, string> | undefined = undefined
+  const uploadMetadataHeader = req.headers['upload-metadata']
+
+  if (uploadMetadataHeader && typeof uploadMetadataHeader === 'string') {
+    try {
+      const parsedMetadata = Metadata.parse(uploadMetadataHeader)
+      if (parsedMetadata?.metadata) {
+        customMd = JSON.parse(parsedMetadata.metadata)
+      }
+    } catch (e) {
+      req.log.warn({ error: e }, 'Failed to parse user metadata')
+    }
+  }
+
   // Handle signed url requests
   if (req.url?.startsWith(`/upload/resumable/sign`)) {
     const signature = req.headers['x-signature']
@@ -96,11 +110,20 @@ export async function onIncomingRequest(rawReq: Request, id: string) {
     req.upload.storage.location
   )
 
+  const uploadLength = req.headers['upload-length']
+  const contentLength = uploadLength ? Number(uploadLength) : undefined
+  const contentType = req.headers['content-type']
+
   await uploader.canUpload({
     owner: req.upload.owner,
     bucketId: uploadID.bucket,
     objectName: uploadID.objectName,
     isUpsert: isUpsert,
+    userMetadata: customMd,
+    metadata: {
+      mimetype: contentType,
+      contentLength: contentLength,
+    },
   })
 }
 

src/internal/database/migrations/types.ts

@@ -55,4 +55,5 @@ export const DBMigration = {
   'drop-index-lower-name': 53,
   'drop-index-object-level': 54,
   'prevent-direct-deletes': 55,
+  's3-multipart-uploads-metadata': 56,
 }

src/storage/database/adapter.ts

@@ -194,7 +194,8 @@ export interface Database {
     version: string,
     signature: string,
     owner?: string,
-    metadata?: Record<string, string | null>
+    userMetadata?: Record<string, string | null>,
+    metadata?: Partial<ObjectMetadata>
   ): Promise<S3MultipartUpload>
 
   findMultipartUpload(

src/storage/database/knex.ts

@@ -889,7 +889,8 @@ export class StorageKnexDB implements Database {
     version: string,
     signature: string,
     owner?: string,
-    metadata?: Record<string, string | null>
+    userMetadata?: Record<string, string | null>,
+    metadata?: Partial<ObjectMetadata>
   ) {
     return this.runQuery('CreateMultipartUpload', async (knex) => {
       const multipart = await knex
@@ -902,7 +903,8 @@ export class StorageKnexDB implements Database {
             version,
             upload_signature: signature,
             owner_id: owner,
-            user_metadata: metadata,
+            user_metadata: userMetadata,
+            metadata: metadata,
           })
         )
         .returning('*')

src/storage/object.ts

@@ -6,7 +6,7 @@ import { getJwtSecret } from '@internal/database'
 import { ObjectMetadata, StorageBackendAdapter } from './backend'
 import { Database, FindObjectFilters, SearchObjectOption } from './database'
 import { mustBeValidKey } from './limits'
-import { fileUploadFromRequest, Uploader, UploadRequest } from './uploader'
+import { fileUploadFromRequest, Uploader, UploadRequest, CanUploadMetadata } from './uploader'
 import { getConfig } from '../config'
 import {
   ObjectAdminDelete,
@@ -98,6 +98,7 @@ export class ObjectStorage {
       owner: file.owner,
       isUpsert: Boolean(file.isUpsert),
       signal: file.signal,
+      userMetadata: uploadRequest.userMetadata,
     })
   }
 
@@ -339,6 +340,8 @@ export class ObjectStorage {
       objectName: destinationKey,
       owner,
       isUpsert: upsert,
+      userMetadata: userMetadata,
+      metadata: destinationMetadata,
     })
 
     try {
@@ -792,14 +795,20 @@ export class ObjectStorage {
     url: string,
     expiresIn: number,
     owner?: string,
-    options?: { upsert?: boolean }
+    options?: {
+      upsert?: boolean
+      userMetadata?: Record<string, unknown>
+      metadata?: CanUploadMetadata
+    }
   ) {
     // check if user has INSERT permissions
     await this.uploader.canUpload({
       bucketId: this.bucketId,
       objectName,
       owner,
       isUpsert: options?.upsert ?? false,
+      userMetadata: options?.userMetadata,
+      metadata: options?.metadata,
     })
 
     const { urlSigningKey } = await getJwtSecret(this.db.tenantId)

src/storage/protocols/s3/s3-handler.ts

@@ -413,6 +413,10 @@ export class S3ProtocolHandler {
       objectName: command.Key as string,
       isUpsert: true,
       owner: this.owner,
+      userMetadata: command.Metadata,
+      metadata: {
+        mimetype: command.ContentType,
+      },
     })
 
     const uploadId = await this.storage.backend.createMultiPartUpload(
@@ -441,7 +445,8 @@ export class S3ProtocolHandler {
         version,
         signature,
         this.owner,
-        command.Metadata
+        command.Metadata,
+        { mimetype: command.ContentType }
       )
 
     return {
@@ -470,17 +475,19 @@ export class S3ProtocolHandler {
       throw ERRORS.InvalidUploadId()
     }
 
+    const multiPartUpload = await this.storage.db
+      .asSuperUser()
+      .findMultipartUpload(UploadId, 'id,version,user_metadata,metadata')
+
     await uploader.canUpload({
       bucketId: Bucket as string,
       objectName: Key as string,
       isUpsert: true,
       owner: this.owner,
+      userMetadata: multiPartUpload.user_metadata || undefined,
+      metadata: multiPartUpload.metadata || undefined,
     })
 
-    const multiPartUpload = await this.storage.db
-      .asSuperUser()
-      .findMultipartUpload(UploadId, 'id,version,user_metadata')
-
     const parts = command.MultipartUpload?.Parts || []
 
     if (parts.length === 0) {
@@ -578,15 +585,18 @@ export class S3ProtocolHandler {
     const maxFileSize = await getFileSizeLimit(this.storage.db.tenantId, bucket?.file_size_limit)
 
     const uploader = new Uploader(this.storage.backend, this.storage.db, this.storage.location)
+
+    const multipart = await this.shouldAllowPartUpload(UploadId, ContentLength, maxFileSize)
+
     await uploader.canUpload({
       bucketId: Bucket as string,
       objectName: Key as string,
       owner: this.owner,
       isUpsert: true,
+      userMetadata: multipart.user_metadata || undefined,
+      metadata: multipart.metadata || undefined,
     })
 
-    const multipart = await this.shouldAllowPartUpload(UploadId, ContentLength, maxFileSize)
-
     if (signal?.aborted) {
       throw ERRORS.AbortedTerminate('UploadPart aborted')
     }
@@ -695,9 +705,9 @@ export class S3ProtocolHandler {
         cacheControl: command.CacheControl!,
         mimeType: command.ContentType!,
         isTruncated: options.isTruncated,
-        userMetadata: command.Metadata,
       },
       objectName: command.Key as string,
+      userMetadata: command.Metadata,
       owner: this.owner,
       isUpsert: true,
       uploadType: 's3',
@@ -735,14 +745,16 @@ export class S3ProtocolHandler {
 
     const multipart = await this.storage.db
       .asSuperUser()
-      .findMultipartUpload(UploadId, 'id,version')
+      .findMultipartUpload(UploadId, 'id,version,user_metadata,metadata')
 
     const uploader = new Uploader(this.storage.backend, this.storage.db, this.storage.location)
     await uploader.canUpload({
       bucketId: Bucket,
       objectName: Key,
       owner: this.owner,
       isUpsert: true,
+      userMetadata: multipart.user_metadata || undefined,
+      metadata: multipart.metadata || undefined,
     })
 
     await this.storage.backend.abortMultipartUpload(
@@ -1221,13 +1233,6 @@ export class S3ProtocolHandler {
 
     const uploader = new Uploader(this.storage.backend, this.storage.db, this.storage.location)
 
-    await uploader.canUpload({
-      bucketId: Bucket,
-      objectName: Key,
-      owner: this.owner,
-      isUpsert: true,
-    })
-
     const [destinationBucket] = await this.storage.db.asSuperUser().withTransaction(async (db) => {
       return Promise.all([
         db.findBucketById(Bucket, 'file_size_limit'),
@@ -1241,6 +1246,15 @@ export class S3ProtocolHandler {
 
     const multipart = await this.shouldAllowPartUpload(UploadId, Number(copySize), maxFileSize)
 
+    await uploader.canUpload({
+      bucketId: Bucket,
+      objectName: Key,
+      owner: this.owner,
+      isUpsert: true,
+      userMetadata: multipart.user_metadata || undefined,
+      metadata: multipart.metadata || undefined,
+    })
+
     const uploadPart = await this.storage.backend.uploadPartCopy(
       storageS3Bucket,
       this.storage.location.getKeyLocation({
@@ -1312,7 +1326,7 @@ export class S3ProtocolHandler {
     return this.storage.db.asSuperUser().withTransaction(async (db) => {
       const multipart = await db.findMultipartUpload(
         uploadId,
-        'in_progress_size,version,upload_signature',
+        'in_progress_size,version,upload_signature,user_metadata,metadata',
         {
           forUpdate: true,
         }

src/storage/schemas/multipart.ts

@@ -15,6 +15,9 @@ export const multipartUploadSchema = {
     user_metadata: {
       anyOf: [{ type: 'object', additionalProperties: true }, { type: 'null' }],
     },
+    metadata: {
+      anyOf: [{ type: 'object', additionalProperties: true }, { type: 'null' }],
+    },
   },
   required: [
     'id',