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

@@ -1,6 +1,7 @@
 import { FastifyInstance } from 'fastify'
 import { FromSchema } from 'json-schema-to-ts'
 import { getConfig } from '../../../config'
+import { parseUserMetadata } from '../../../storage/uploader'
 import { createDefaultSchema } from '../../routes-helper'
 import { AuthenticatedRequest } from '../../types'
 import { ROUTE_OPERATIONS } from '../operations'
@@ -20,6 +21,9 @@ const getSignedUploadURLHeadersSchema = {
   type: 'object',
   properties: {
     'x-upsert': { type: 'string' },
+    'x-metadata': { type: 'string' },
+    'content-type': { type: 'string' },
+    'content-length': { type: 'string' },
     authorization: { type: 'string' },
   },
   required: ['authorization'],
@@ -69,10 +73,29 @@ 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') {
+        // TODO: parseUserMetadata casts to Record<string, string> but values could be anything;
+        // validation should be added in a follow-up
+        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,
     onUploadCreate: onCreate,
     onUploadFinish,
-    onIncomingRequest,
+    onIncomingRequest: (req, id) => onIncomingRequest(req, id, datastore),
     generateUrl,
     getFileIdFromRequest,
     onResponseError,

src/http/routes/tus/lifecycle.ts

@@ -3,7 +3,7 @@ import { ERRORS, isRenderableError } from '@internal/errors'
 import { UploadId } from '@storage/protocols/tus'
 import { Storage } from '@storage/storage'
 import { Uploader, validateMimeType } from '@storage/uploader'
-import { Upload } from '@tus/server'
+import { DataStore, Metadata, Upload } from '@tus/server'
 import { randomUUID } from 'crypto'
 import http from 'http'
 import { BaseLogger } from 'pino'
@@ -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
 
@@ -92,11 +92,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,
+    userMetadata: customMd,
+    metadata: {
+      mimetype: contentType,
+      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

@@ -902,22 +902,33 @@ 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,
+      }
+
+      // TODO: move this guard into normalizeColumns once it is table-aware.
+      // metadata was added to s3_multipart_uploads in migration 57 but has existed on
+      // objects since much earlier, so a table-agnostic rule would incorrectly strip it.
+      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)
 
@@ -927,10 +938,18 @@ export class StorageKnexDB implements Database {
 
   async findMultipartUpload(uploadId: string, columns = 'id', options?: { forUpdate?: boolean }) {
     const multiPart = await this.runQuery('FindMultipartUpload', async (knex, signal) => {
-      const query = knex
-        .from('s3_multipart_uploads')
-        .select(columns.split(','))
-        .where('id', uploadId)
+      // TODO: move this guard into normalizeColumns once it is table-aware.
+      // metadata was added to s3_multipart_uploads in migration 57 but has existed on
+      // objects since much earlier, so a table-agnostic rule would incorrectly strip it.
+      const hasMetadataColumn =
+        !this.latestMigration ||
+        DBMigration[this.latestMigration] >= DBMigration['s3-multipart-uploads-metadata']
+
+      const cols = hasMetadataColumn
+        ? columns.split(',')
+        : columns.split(',').filter((col) => col.trim() !== 'metadata')
+
+      const query = knex.from('s3_multipart_uploads').select(cols).where('id', uploadId)
 
       if (options?.forUpdate) {
         return query.abortOnSignal(signal).forUpdate().first()

src/storage/object.ts

@@ -17,7 +17,7 @@ import {
   ObjectUpdatedMetadata,
 } from './events'
 import { mustBeValidKey } from './limits'
-import { fileUploadFromRequest, Uploader, UploadRequest } from './uploader'
+import { CanUploadMetadata, fileUploadFromRequest, Uploader, UploadRequest } from './uploader'
 
 const { requestUrlLengthLimit } = getConfig()
 
@@ -97,6 +97,7 @@ export class ObjectStorage {
       owner: file.owner,
       isUpsert: Boolean(file.isUpsert),
       signal: file.signal,
+      userMetadata: uploadRequest.userMetadata,
     })
   }
 
@@ -332,11 +333,15 @@ export class ObjectStorage {
           ...(fileMetadata || {}),
         }
 
+    const destinationUserMetadata = copyMetadata ? originObject.user_metadata : userMetadata
+
     await this.uploader.canUpload({
       bucketId: destinationBucket,
       objectName: destinationKey,
       owner,
       isUpsert: upsert,
+      userMetadata: destinationUserMetadata || undefined,
+      metadata: destinationMetadata,
     })
 
     try {
@@ -381,7 +386,7 @@ export class ObjectStorage {
             lastModified: copyResult.lastModified,
             eTag: copyResult.eTag,
           },
-          user_metadata: copyMetadata ? originObject.user_metadata : userMetadata,
+          user_metadata: destinationUserMetadata,
           version: newVersion,
         })
 
@@ -790,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)