feat: tus with linked buckets

Author: fenos

migrations/multitenant/0006-add-tenants-external-credentials.sql

@@ -0,0 +1,8 @@
+alter table storage.objects add column if not exists owner_id text default null;
+alter table storage.buckets add column if not exists owner_id text default null;
+
+comment on column storage.objects.owner is 'Field is deprecated, use owner_id instead';
+comment on column storage.buckets.owner is 'Field is deprecated, use owner_id instead';
+
+ALTER TABLE storage.buckets
+    DROP CONSTRAINT IF EXISTS buckets_owner_fkey;

migrations/tenant/0017-add_owner_id_column_deprecate_owner.sql

@@ -0,0 +1,14 @@
+CREATE TABLE bucket_credentials (
+      "id" uuid NOT NULL DEFAULT extensions.uuid_generate_v4(),
+      name text NOT NULL unique,
+      access_key text NULL,
+      secret_key text NULL,
+      role text null,
+      region text not null,
+      endpoint text NULL,
+      force_path_style boolean NOT NULL default false,
+      PRIMARY KEY (id)
+);
+
+ALTER TABLE storage.buckets ADD COLUMN credential_id uuid DEFAULT NULL;
+ALTER TABLE storage.buckets ADD CONSTRAINT fk_bucket_credential FOREIGN KEY (credential_id) REFERENCES bucket_credentials(id);

migrations/tenant/0018-external-buckets.sql

@@ -60,6 +60,7 @@ const build = (opts: buildOpts = {}): FastifyInstance => {
   app.register(plugins.logTenantId)
   app.register(plugins.logRequest({ excludeUrls: ['/status', '/metrics'] }))
   app.register(routes.multiPart, { prefix: 'upload/resumable' })
+  app.register(routes.credentials, { prefix: 'credentials' })
   app.register(routes.bucket, { prefix: 'bucket' })
   app.register(routes.object, { prefix: 'object' })
   app.register(routes.render, { prefix: 'render/image' })

src/app.ts

@@ -1,8 +1,10 @@
-import { getJwtSecret as getJwtSecretForTenant } from '../database/tenant'
+import {getJwtSecret as getJwtSecretForTenant, getTenantConfig} from '../database/tenant'
 import jwt from 'jsonwebtoken'
+import crypto from 'crypto'
 import { getConfig } from '../config'
+import {StorageBackendError} from "../storage";
 
-const { isMultitenant, jwtSecret, jwtAlgorithm } = getConfig()
+const { isMultitenant, jwtSecret, jwtAlgorithm, serviceKey } = getConfig()
 
 interface jwtInterface {
   sub?: string
@@ -21,6 +23,21 @@ export type SignedUploadToken = {
   exp: number
 }
 
+export async function compareServiceKey(tenantId: string, jwt: string) {
+  if (isMultitenant) {
+    const { serviceKey } = await getTenantConfig(tenantId)
+    return crypto.timingSafeEqual(Buffer.from(serviceKey), Buffer.from(jwt))
+  }
+
+  return crypto.timingSafeEqual(Buffer.from(serviceKey), Buffer.from(jwt))
+}
+
+export async function mustBeServiceKey(tenantId: string, jwt: string) {
+  if (!(await compareServiceKey(tenantId, jwt))) {
+    throw new StorageBackendError('unauthorized', 401, 'Unauthorized')
+  }
+}
+
 /**
  * Gets the JWT secret key from the env PGRST_JWT_SECRET when running in single-tenant
  * or querying the multi-tenant database by the given tenantId

src/auth/jwt.ts

@@ -119,7 +119,7 @@ export function getConfig(): StorageConfigType {
       10
     ),
     databaseConnectionTimeout: parseInt(
-      getOptionalConfigFromEnv('DATABASE_CONNECTION_TIMEOUT') || '30000',
+      getOptionalConfigFromEnv('DATABASE_CONNECTION_TIMEOUT') || '3000',
       10
     ),
     region: getConfigFromEnv('REGION'),

src/config.ts

@@ -2,7 +2,6 @@ import pg from 'pg'
 import { Knex, knex } from 'knex'
 import { JwtPayload } from 'jsonwebtoken'
 import { getConfig } from '../config'
-import { logger } from '../monitoring'
 import { DbActiveConnection, DbActivePool } from '../monitoring/metrics'
 
 // https://github.com/knex/knex/issues/387#issuecomment-51554522
@@ -30,6 +29,7 @@ export interface User {
 }
 
 export const connections = new Map<string, Knex>()
+const searchPath = ['storage', 'public', 'extensions']
 
 export class TenantConnection {
   public readonly role: string
@@ -63,13 +63,13 @@ export class TenantConnection {
 
     knexPool = knex({
       client: 'pg',
-      searchPath: ['public', 'storage', 'extensions'],
+      searchPath: isExternalPool ? undefined : searchPath,
       pool: {
         min: 0,
         max: isExternalPool ? 1 : options.maxConnections || databaseMaxConnections,
         propagateCreateError: false,
         acquireTimeoutMillis: databaseConnectionTimeout,
-        idleTimeoutMillis: isExternalPool ? 100 : undefined,
+        idleTimeoutMillis: isExternalPool ? 100 : databaseFreePoolAfterInactivity,
         reapIntervalMillis: isExternalPool ? 110 : undefined,
       },
       connection: connectionString,
@@ -97,38 +97,12 @@ export class TenantConnection {
     })
 
     if (!isExternalPool) {
-      let freePoolIntervalFn: NodeJS.Timeout | undefined
-
       knexPool.client.pool.on('poolDestroySuccess', () => {
-        if (freePoolIntervalFn) {
-          clearTimeout(freePoolIntervalFn)
-        }
-
         if (connections.get(connectionString) === knexPool) {
           connections.delete(connectionString)
         }
       })
 
-      knexPool.client.pool.on('stopReaping', () => {
-        if (freePoolIntervalFn) {
-          clearTimeout(freePoolIntervalFn)
-        }
-
-        freePoolIntervalFn = setTimeout(async () => {
-          connections.delete(connectionString)
-          knexPool?.destroy().catch((e) => {
-            logger.error(e, 'Error destroying pool')
-          })
-          clearTimeout(freePoolIntervalFn)
-        }, databaseFreePoolAfterInactivity)
-      })
-
-      knexPool.client.pool.on('startReaping', () => {
-        if (freePoolIntervalFn) {
-          clearTimeout(freePoolIntervalFn)
-          freePoolIntervalFn = undefined
-        }
-      })
       connections.set(connectionString, knexPool)
     }
 
@@ -141,10 +115,16 @@ export class TenantConnection {
     }
   }
 
-  transaction(isolation?: Knex.IsolationLevels, instance?: Knex) {
-    return (instance || this.pool).transactionProvider({
-      isolationLevel: isolation,
-    })
+  transaction(instance?: Knex): Knex.TransactionProvider {
+    return async () => {
+      const pool = instance || this.pool
+      const tnx = await pool.transaction()
+
+      if (!instance) {
+        await tnx.raw(`set search_path to ${searchPath.join(', ')}`)
+      }
+      return tnx
+    }
   }
 
   asSuperUser() {

src/database/connection.ts

@@ -5,6 +5,7 @@ import { runMigrationsOnTenant } from './migrate'
 import { knex } from './multitenant-db'
 import { StorageBackendError } from '../storage'
 import { JwtPayload } from 'jsonwebtoken'
+import { Credential } from '../storage/schemas'
 
 interface TenantConfig {
   anonKey: string

src/database/tenant.ts

@@ -1,6 +1,9 @@
 import { FastifyInstance } from 'fastify'
 import { isRenderableError } from '../storage'
 import { FastifyError } from '@fastify/error'
+import { getConfig } from '../config'
+
+const { tusPath } = getConfig()
 
 /**
  * The global error handler for all the uncaught exceptions within a request.
@@ -20,7 +23,8 @@ export const setErrorHandler = (app: FastifyInstance) => {
 
     if (isRenderableError(error)) {
       const renderableError = error.render()
-      return reply.status(renderableError.statusCode === '500' ? 500 : 400).send(renderableError)
+      const body = request.routerPath.includes(tusPath) ? renderableError.error : renderableError
+      return reply.status(renderableError.statusCode === '500' ? 500 : 400).send(body)
     }
 
     // Fastify errors

src/http/error-handler.ts

@@ -0,0 +1,41 @@
+import fastifyPlugin from 'fastify-plugin'
+import { RouteGenericInterface } from 'fastify/types/route'
+import { BucketWithCredentials } from '../../storage/schemas'
+import { StorageBackendError } from '../../storage'
+
+declare module 'fastify' {
+  interface FastifyRequest<RouteGeneric extends RouteGenericInterface = RouteGenericInterface> {
+    bucket: BucketWithCredentials
+  }
+
+  interface FastifyContextConfig {
+    getParentBucketId?: ((request: FastifyRequest<any>) => string) | false
+  }
+}
+
+export const parentBucket = fastifyPlugin(async (fastify) => {
+  fastify.decorateRequest('bucket', undefined)
+  fastify.addHook('preHandler', async (request) => {
+    if (typeof request.routeConfig.getParentBucketId === 'undefined') {
+      throw new Error(
+        `getParentBucketId not defined in route ${request.routerPath} ${request.routerPath} config`
+      )
+    }
+
+    if (request.routeConfig.getParentBucketId === false) {
+      return
+    }
+
+    const bucketId = request.routeConfig.getParentBucketId(request)
+
+    if (!bucketId) {
+      throw new StorageBackendError('invalid_bucket', 400, 'bucket name is invalid or not provided')
+    }
+
+    const bucket = await request.db.asSuperUser().findBucketById(bucketId, 'buckets.*', {
+      includeCredentials: true,
+    })
+
+    request.bucket = bucket
+  })
+})