feat: upgrade tus to latest + S3 Locker (#740)

Author: fenos

jest.config.cjs

@@ -3,10 +3,13 @@ module.exports = {
   testSequencer: './jest.sequencer.cjs',
   transform: {
     '^.+/node_modules/jose/.+\\.[jt]s$': 'babel-jest',
+    '^.+/node_modules/@tus/.+\\.[jt]s$': 'babel-jest',
+    '^.+/node_modules/srvx/.+\\.[jt]s$': 'babel-jest',
+    '^.+/node_modules/cookie-es/.+\\.[jt]s$': 'babel-jest',
     '^.+\\.mjs$': 'babel-jest',
     '^.+\\.(t|j)sx?$': 'ts-jest',
   },
-  transformIgnorePatterns: ['node_modules/(?!(jose)/)'],
+  transformIgnorePatterns: ['node_modules/(?!(jose|@tus|srvx|cookie-es)/)'],
   moduleNameMapper: {
     '^@storage/(.*)$': '<rootDir>/src/storage/$1',
     '^@internal/(.*)$': '<rootDir>/src/internal/$1',

package-lock.json

@@ -47,9 +47,9 @@
     "@opentelemetry/instrumentation-pg": "^0.44.0",
     "@shopify/semaphore": "^3.0.2",
     "@smithy/node-http-handler": "^2.3.1",
-    "@tus/file-store": "1.4.0",
-    "@tus/s3-store": "1.5.0",
-    "@tus/server": "1.7.0",
+    "@tus/file-store": "2.0.0",
+    "@tus/s3-store": "2.0.0",
+    "@tus/server": "2.2.1",
     "agentkeepalive": "^4.5.0",
     "ajv": "^8.12.0",
     "async-retry": "^1.3.3",

package.json

@@ -157,6 +157,7 @@ type StorageConfigType = {
   tusPartSize: number
   tusUseFileVersionSeparator: boolean
   tusAllowS3Tags: boolean
+  tusLockType: 'postgres' | 's3'
   defaultMetricsEnabled: boolean
   s3ProtocolEnabled: boolean
   s3ProtocolPrefix: string
@@ -312,6 +313,7 @@ export function getConfig(options?: { reload?: boolean }): StorageConfigType {
     tusUseFileVersionSeparator:
       getOptionalConfigFromEnv('TUS_USE_FILE_VERSION_SEPARATOR') === 'true',
     tusAllowS3Tags: getOptionalConfigFromEnv('TUS_ALLOW_S3_TAGS') !== 'false',
+    tusLockType: getOptionalConfigFromEnv('TUS_LOCK_TYPE') || 'postgres',
 
     // S3 Protocol
     s3ProtocolEnabled: getOptionalConfigFromEnv('S3_PROTOCOL_ENABLED') !== 'false',

src/config.ts

@@ -1,19 +1,12 @@
 import { FastifyBaseLogger, FastifyInstance } from 'fastify'
 import fastifyPlugin from 'fastify-plugin'
 import * as http from 'http'
-import { ServerOptions, DataStore } from '@tus/server'
+import { ServerOptions, DataStore, Server } from '@tus/server'
 import { getFileSizeLimit } from '@storage/limits'
 import { Storage } from '@storage/storage'
 import { jwt, storage, db, dbSuperUser } from '../../plugins'
 import { getConfig } from '../../../config'
-import {
-  TusServer,
-  FileStore,
-  LockNotifier,
-  PgLocker,
-  UploadId,
-  AlsMemoryKV,
-} from '@storage/protocols/tus'
+import { FileStore, LockNotifier, PgLocker, UploadId, AlsMemoryKV } from '@storage/protocols/tus'
 import {
   namingFunction,
   onCreate,
@@ -30,6 +23,10 @@ import { NodeHttpHandler } from '@smithy/node-http-handler'
 import { ROUTE_OPERATIONS } from '../operations'
 import * as https from 'node:https'
 import { createAgent } from '@internal/http'
+import type { ServerRequest as Request } from 'srvx'
+import { S3Locker } from '@storage/protocols/tus/s3-locker'
+import { S3Client } from '@aws-sdk/client-s3'
+import { ERRORS } from '@internal/errors'
 
 const {
   storageS3MaxSockets,
@@ -43,6 +40,7 @@ const {
   tusPartSize,
   tusMaxConcurrentUploads,
   tusAllowS3Tags,
+  tusLockType,
   uploadFileSizeLimit,
   storageBackendType,
   storageFilePath,
@@ -98,9 +96,42 @@ function createTusServer(
     path: tusPath,
     datastore: datastore,
     disableTerminationForFinishedUploads: true,
-    locker: (rawReq: http.IncomingMessage) => {
-      const req = rawReq as MultiPartRequest
-      return new PgLocker(req.upload.storage.db, lockNotifier)
+    locker: (rawReq: Request) => {
+      const req = rawReq.node?.req as MultiPartRequest
+
+      if (!req) {
+        throw ERRORS.InternalError(undefined, 'Request object is missing')
+      }
+
+      switch (tusLockType) {
+        case 'postgres':
+          return new PgLocker(req.upload.storage.db, lockNotifier)
+
+        case 's3':
+          return new S3Locker({
+            bucket: storageS3Bucket,
+            keyPrefix: `__tus_locks/${req.upload.tenantId}/`,
+            logger: console,
+            lockTtlMs: 15 * 1000, // 15 seconds
+            maxRetries: 10,
+            retryDelayMs: 250,
+            renewalIntervalMs: 10 * 1000, // 10 seconds
+            s3Client: new S3Client({
+              requestHandler: new NodeHttpHandler({
+                ...agent,
+                connectionTimeout: 5000,
+                requestTimeout: storageS3ClientTimeout,
+              }),
+              region: storageS3Region,
+              endpoint: storageS3Endpoint,
+              forcePathStyle: storageS3ForcePathStyle,
+            }),
+            notifier: lockNotifier,
+          })
+
+        default:
+          throw ERRORS.InternalError(undefined, 'Unsupported TUS locker type')
+      }
     },
     namingFunction: namingFunction,
     onUploadCreate: onCreate,
@@ -112,7 +143,7 @@ function createTusServer(
     respectForwardedHeaders: true,
     allowedHeaders: ['Authorization', 'X-Upsert', 'Upload-Expires', 'ApiKey', 'x-signature'],
     maxSize: async (rawReq, uploadId) => {
-      const req = rawReq as MultiPartRequest
+      const req = rawReq.node?.req as MultiPartRequest
 
       if (!req.upload.tenantId) {
         return uploadFileSizeLimit
@@ -138,7 +169,7 @@ function createTusServer(
       return fileSizeLimit
     },
   }
-  return new TusServer(serverOptions)
+  return new Server(serverOptions)
 }
 
 export default async function routes(fastify: FastifyInstance) {
@@ -204,7 +235,7 @@ export default async function routes(fastify: FastifyInstance) {
 }
 
 const authenticatedRoutes = fastifyPlugin(
-  async (fastify: FastifyInstance, options: { tusServer: TusServer; operation?: string }) => {
+  async (fastify: FastifyInstance, options: { tusServer: Server; operation?: string }) => {
     fastify.register(async function authorizationContext(fastify) {
       fastify.addContentTypeParser('application/offset+octet-stream', (request, payload, done) =>
         done(null)
@@ -312,7 +343,7 @@ const authenticatedRoutes = fastifyPlugin(
 )
 
 const publicRoutes = fastifyPlugin(
-  async (fastify: FastifyInstance, options: { tusServer: TusServer; operation?: string }) => {
+  async (fastify: FastifyInstance, options: { tusServer: Server; operation?: string }) => {
     fastify.register(async (fastify) => {
       fastify.addContentTypeParser('application/offset+octet-stream', (request, payload, done) =>
         done(null)

src/http/routes/tus/index.ts

@@ -7,6 +7,7 @@ import { ERRORS, isRenderableError } from '@internal/errors'
 import { Storage } from '@storage/storage'
 import { Uploader, validateMimeType } from '@storage/uploader'
 import { UploadId } from '@storage/protocols/tus'
+import type { ServerRequest as Request } from 'srvx'
 
 import { getConfig } from '../../../config'
 
@@ -15,6 +16,19 @@ const reExtractFileID = /([^/]+)\/?$/
 
 export const SIGNED_URL_SUFFIX = '/sign'
 
+/**
+ * Validates that the request object exists and returns it
+ * @throws {Error} If the request object is missing
+ */
+function getNodeRequest(rawReq: Request): MultiPartRequest {
+  const req = rawReq.node?.req as MultiPartRequest
+
+  if (!req) {
+    throw ERRORS.InternalError(undefined, 'Request object is missing')
+  }
+
+  return req
+}
 export type MultiPartRequest = http.IncomingMessage & {
   log: BaseLogger
   upload: {
@@ -30,14 +44,20 @@ export type MultiPartRequest = http.IncomingMessage & {
 /**
  * Runs on every TUS incoming request
  */
-export async function onIncomingRequest(
-  rawReq: http.IncomingMessage,
-  res: http.ServerResponse,
-  id: string
-) {
-  const req = rawReq as MultiPartRequest
+export async function onIncomingRequest(rawReq: Request, id: string) {
+  const req = getNodeRequest(rawReq)
+  const res = rawReq.node?.res as http.ServerResponse
+
+  if (!res) {
+    throw ERRORS.InternalError(undefined, 'Response object is missing')
+  }
+
+  if (!res) {
+    throw ERRORS.InternalError(undefined, 'Response object is missing')
+  }
 
   res.on('finish', () => {
+    console.log('Tus request finished')
     req.upload.db.dispose().catch((e) => {
       req.log.error({ error: e }, 'Error disposing db connection')
     })
@@ -88,9 +108,15 @@ export async function onIncomingRequest(
  * Generate URL for TUS upload, it encodes the uploadID to base64url
  */
 export function generateUrl(
-  req: http.IncomingMessage,
+  rawReq: Request,
   { proto, host, path, id }: { proto: string; host: string; path: string; id: string }
 ) {
+  const req = getNodeRequest(rawReq)
+
+  if (!req.url) {
+    throw ERRORS.InvalidParameter('url')
+  }
+
   if (!req.url) {
     throw ERRORS.InvalidParameter('url')
   }
@@ -127,8 +153,9 @@ export function generateUrl(
 /**
  * Extract the uploadId from the request and decodes it from base64url
  */
-export function getFileIdFromRequest(rawRwq: http.IncomingMessage) {
-  const req = rawRwq as MultiPartRequest
+export function getFileIdFromRequest(rawReq: Request) {
+  const req = getNodeRequest(rawReq)
+
   const match = reExtractFileID.exec(req.url as string)
 
   if (!match || tusPath.includes(match[1])) {
@@ -145,11 +172,12 @@ export function getFileIdFromRequest(rawRwq: http.IncomingMessage) {
  *
  * /tenant-id/bucket-name/object-name/version
  */
-export function namingFunction(
-  rawReq: http.IncomingMessage,
-  metadata?: Record<string, string | null>
-) {
-  const req = rawReq as MultiPartRequest
+export function namingFunction(rawReq: Request, metadata?: Record<string, string | null>) {
+  const req = getNodeRequest(rawReq)
+
+  if (!req.url) {
+    throw new Error('no url set')
+  }
 
   if (!req.url) {
     throw new Error('no url set')
@@ -177,13 +205,13 @@ export function namingFunction(
  * Runs before the upload URL is created
  */
 export async function onCreate(
-  rawReq: http.IncomingMessage,
-  res: http.ServerResponse,
+  rawReq: Request,
   upload: Upload
-): Promise<{ res: http.ServerResponse; metadata?: Upload['metadata'] }> {
+): Promise<{ metadata?: Upload['metadata'] }> {
   const uploadID = UploadId.fromString(upload.id)
 
-  const req = rawReq as MultiPartRequest
+  const req = getNodeRequest(rawReq)
+
   const storage = req.upload.storage
 
   const bucket = await storage
@@ -204,18 +232,15 @@ export async function onCreate(
     validateMimeType(metadata.contentType, bucket.allowed_mime_types)
   }
 
-  return { res, metadata }
+  return { metadata }
 }
 
 /**
  * Runs when the upload to the underline store is completed
  */
-export async function onUploadFinish(
-  rawReq: http.IncomingMessage,
-  res: http.ServerResponse,
-  upload: Upload
-) {
-  const req = rawReq as MultiPartRequest
+export async function onUploadFinish(rawReq: Request, upload: Upload) {
+  const req = getNodeRequest(rawReq)
+
   const resourceId = UploadId.fromString(upload.id)
 
   try {
@@ -255,9 +280,11 @@ export async function onUploadFinish(
       userMetadata: customMd,
     })
 
-    res.setHeader('Tus-Complete', '1')
-
-    return res
+    return {
+      headers: {
+        'Tus-Complete': '1',
+      },
+    }
   } catch (e) {
     if (isRenderableError(e)) {
       ;(e as any).status_code = parseInt(e.render().statusCode, 10)
@@ -272,11 +299,9 @@ type TusError = { status_code: number; body: string }
 /**
  * Runs when there is an error on the TUS upload
  */
-export function onResponseError(
-  req: http.IncomingMessage,
-  _: http.ServerResponse,
-  e: TusError | Error
-) {
+export function onResponseError(rawReq: Request, e: TusError | Error) {
+  const req = getNodeRequest(rawReq)
+
   if (e instanceof Error) {
     ;(req as any).executionError = e
   } else {

src/http/routes/tus/lifecycle.ts

@@ -2,4 +2,3 @@ export * from './file-store'
 export * from './postgres-locker'
 export * from './upload-id'
 export * from './als-memory-kv'
-export * from './server'