feat: add user metadata support for RLS policies

migrations/tenant/0057-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,
+          metadata: {
+            mimetype: contentType,
+            contentLength,
+          },
         })
 
       return response.status(200).send({ url: signedUpload.url, token: signedUpload.token })

src/http/routes/tus/index.ts

@@ -137,7 +137,7 @@ function createTusServer(
     namingFunction: namingFunction,
     onUploadCreate: onCreate,
     onUploadFinish: onUploadFinish,
-    onIncomingRequest: onIncomingRequest,
+    onIncomingRequest: (req, id) => onIncomingRequest(req, id, datastore),
     generateUrl: generateUrl,
     getFileIdFromRequest: getFileIdFromRequest,
     onResponseError: onResponseError,

src/http/routes/tus/lifecycle.ts

@@ -1,6 +1,6 @@
 import http from 'http'
 import { BaseLogger } from 'pino'
-import { Upload } from '@tus/server'
+import { DataStore, Metadata, Upload } from '@tus/server'
 import { randomUUID } from 'crypto'
 import { TenantConnection } from '@internal/database'
 import { ERRORS, isRenderableError } from '@internal/errors'
@@ -44,7 +44,7 @@ export type MultiPartRequest = http.IncomingMessage & {
 /**
  * Runs on every TUS incoming request
  */
-export async function onIncomingRequest(rawReq: Request, id: string) {
+export async function onIncomingRequest(rawReq: Request, id: string, datastore: DataStore) {
   const req = getNodeRequest(rawReq)
   const res = rawReq.node?.res as http.ServerResponse
 
@@ -96,11 +96,53 @@ export async function onIncomingRequest(rawReq: Request, id: string) {
     req.upload.storage.location
   )
 
+  let contentType: string | undefined
+  let contentLength: number | undefined
+  let rawMetadata: string | null | undefined
+
+  if (req.method === 'POST') {
+    const uploadMetadataHeader = req.headers['upload-metadata']
+    if (uploadMetadataHeader && typeof uploadMetadataHeader === 'string') {
+      try {
+        const parsedMetadata = Metadata.parse(uploadMetadataHeader)
+        contentType = parsedMetadata?.contentType ?? undefined
+        rawMetadata = parsedMetadata?.metadata
+      } catch (e) {
+        req.log.warn({ error: e }, 'Failed to parse upload metadata')
+        throw ERRORS.InvalidParameter('upload-metadata', {
+          error: e as Error,
+          message: 'Invalid Upload-Metadata header',
+        })
+      }
+    }
+    const uploadLength = req.headers['upload-length']
+    contentLength = uploadLength ? Number(uploadLength) : undefined
+  } else {
+    const upload = await datastore.getUpload(id)
+    contentType = upload.metadata?.contentType ?? undefined
+    contentLength = upload.size ?? undefined
+    rawMetadata = upload.metadata?.metadata
+  }
+
+  let customMd: Record<string, string> | undefined
+  if (rawMetadata) {
+    try {
+      customMd = JSON.parse(rawMetadata)
+    } catch (e) {
+      req.log.warn({ error: e }, 'Failed to parse user metadata')
+    }
+  }
+
   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

@@ -56,4 +56,5 @@ export const DBMigration = {
   'drop-index-object-level': 54,
   'prevent-direct-deletes': 55,
   'fix-optimized-search-function': 56,
+  's3-multipart-uploads-metadata': 57,
 }

src/storage/database/adapter.ts

@@ -195,7 +195,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

@@ -904,22 +904,30 @@ 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, signal) => {
+      const data: Record<string, unknown> = {
+        id: uploadId,
+        bucket_id: bucketId,
+        key: objectName,
+        version,
+        upload_signature: signature,
+        owner_id: owner,
+        user_metadata: userMetadata,
+      }
+
+      if (
+        !this.latestMigration ||
+        DBMigration[this.latestMigration] >= DBMigration['s3-multipart-uploads-metadata']
+      ) {
+        data.metadata = metadata
+      }
+
       const multipart = await knex
         .table<S3MultipartUpload>('s3_multipart_uploads')
-        .insert(
-          this.normalizeColumns({
-            id: uploadId,
-            bucket_id: bucketId,
-            key: objectName,
-            version,
-            upload_signature: signature,
-            owner_id: owner,
-            user_metadata: metadata,
-          })
-        )
+        .insert(this.normalizeColumns(data))
         .returning('*')
         .abortOnSignal(signal)
 

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,10 @@ export class S3ProtocolHandler {
         cacheControl: command.CacheControl!,
         mimeType: command.ContentType!,
         isTruncated: options.isTruncated,
-        userMetadata: command.Metadata,
+        contentLength: command.ContentLength,
       },
       objectName: command.Key as string,
+      userMetadata: command.Metadata,
       owner: this.owner,
       isUpsert: true,
       uploadType: 's3',
@@ -735,14 +746,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(
@@ -1233,13 +1246,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'),
@@ -1253,6 +1259,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({
@@ -1324,7 +1339,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',