fix: optional queue + error code (#225)

Author: fenos

.env.sample

@@ -27,4 +27,5 @@ IMGPROXY_URL=http://localhost:50020
 
 # specify the extra headers will be persisted and passed into Postgrest client, for instance, "x-foo,x-bar"
 POSTGREST_FORWARD_HEADERS=
-WEBHOOK_URL=
+WEBHOOK_URL=
+ENABLE_QUEUE_EVENTS=false

src/config.ts

@@ -28,6 +28,8 @@ type StorageConfigType = {
   logflareEnabled?: boolean
   logflareApiKey?: string
   logflareSourceToken?: string
+  enableQueueEvents: boolean
+  pgQueueConnectionURL?: string
   webhookURL?: string
   webhookApiKey?: string
   disableImageTransformation: boolean
@@ -96,6 +98,8 @@ export function getConfig(): StorageConfigType {
     logflareEnabled: getOptionalConfigFromEnv('LOGFLARE_ENABLED') === 'true',
     logflareApiKey: getOptionalConfigFromEnv('LOGFLARE_API_KEY'),
     logflareSourceToken: getOptionalConfigFromEnv('LOGFLARE_SOURCE_TOKEN'),
+    enableQueueEvents: getOptionalConfigFromEnv('ENABLE_QUEUE_EVENTS') === 'true',
+    pgQueueConnectionURL: getOptionalConfigFromEnv('PG_QUEUE_CONNECTION_URL'),
     webhookURL: getOptionalConfigFromEnv('WEBHOOK_URL'),
     webhookApiKey: getOptionalConfigFromEnv('WEBHOOK_API_KEY'),
     disableImageTransformation: getOptionalConfigFromEnv('DISABLE_IMAGE_TRANSFORMATION') === 'true',

src/http/error-handler.ts

@@ -14,6 +14,8 @@ export const setErrorHandler = (app: FastifyInstance) => {
     // it will be logged in the request log plugin
     reply.executionError = error
 
+    request.log.error({ error }, `request error | ${request.id}`)
+
     if (process.env.NODE_ENV !== 'production') {
       console.error(error)
     }

src/queue/events/base-event.ts

@@ -5,6 +5,7 @@ import { getPostgrestClient } from '../../database'
 import { Storage } from '../../storage'
 import { Database } from '../../storage/database'
 import { createStorageBackend } from '../../storage/backend'
+import { getConfig } from '../../config'
 
 export interface BasePayload {
   $version: string
@@ -16,6 +17,8 @@ export interface BasePayload {
 
 export type StaticThis<T> = { new (...args: any): T }
 
+const { enableQueueEvents } = getConfig()
+
 export abstract class BaseEvent<T extends Omit<BasePayload, '$version'>> {
   public static readonly version: string = 'v1'
 
@@ -71,6 +74,10 @@ export abstract class BaseEvent<T extends Omit<BasePayload, '$version'>> {
   }
 
   send() {
+    if (!enableQueueEvents) {
+      return
+    }
+
     const constructor = this.constructor as typeof BaseEvent
 
     return Queue.getInstance().send({

src/queue/queue.ts

@@ -17,12 +17,12 @@ export abstract class Queue {
       return Queue.pgBoss
     }
 
-    const { isMultitenant, multitenantDatabaseUrl } = getConfig()
+    const { isMultitenant, multitenantDatabaseUrl, pgQueueConnectionURL } = getConfig()
 
-    let url = process.env.DATABASE_URL
+    let url = pgQueueConnectionURL ?? process.env.DATABASE_URL
 
     if (isMultitenant) {
-      url = multitenantDatabaseUrl
+      url = pgQueueConnectionURL ?? multitenantDatabaseUrl
     }
     Queue.pgBoss = new PgBoss({
       connectionString: url,

src/server.ts

@@ -12,8 +12,16 @@ import { Queue } from './queue'
 const exposeDocs = true
 
 ;(async () => {
-  const { isMultitenant, requestIdHeader, adminRequestIdHeader, adminPort, port, host } =
-    getConfig()
+  const {
+    isMultitenant,
+    requestIdHeader,
+    adminRequestIdHeader,
+    adminPort,
+    port,
+    host,
+    enableQueueEvents,
+  } = getConfig()
+
   if (isMultitenant) {
     await runMultitenantMigrations()
     await listenForTenantUpdate()
@@ -34,7 +42,9 @@ const exposeDocs = true
     await runMigrations()
   }
 
-  await Queue.init()
+  if (enableQueueEvents) {
+    await Queue.init()
+  }
 
   const app: FastifyInstance<Server, IncomingMessage, ServerResponse> = build({
     logger,

src/storage/errors.ts

@@ -49,6 +49,11 @@ export class DatabaseError extends Error implements RenderableError {
       message = relationNotPresent
         ? 'The parent resource is not found'
         : 'The resource already exists'
+    } else if (responseStatus === 403) {
+      code = '403'
+    } else {
+      code = '500'
+      type = 'Database Error'
     }
 
     return {

src/test/object.test.ts

@@ -220,7 +220,7 @@ describe('testing POST object via multipart upload', () => {
     expect(S3Backend.prototype.uploadObject).not.toHaveBeenCalled()
     expect(response.body).toBe(
       JSON.stringify({
-        statusCode: '42501',
+        statusCode: '403',
         error: '',
         message: 'new row violates row-level security policy for table "objects"',
       })
@@ -431,7 +431,7 @@ describe('testing POST object via binary upload', () => {
     expect(S3Backend.prototype.uploadObject).not.toHaveBeenCalled()
     expect(response.body).toBe(
       JSON.stringify({
-        statusCode: '42501',
+        statusCode: '403',
         error: '',
         message: 'new row violates row-level security policy for table "objects"',
       })

src/test/webhooks.test.ts

@@ -1,5 +1,8 @@
+process.env.ENABLE_QUEUE_EVENTS = 'true'
+
 import { mockQueue, useMockObject } from './common'
 import FormData from 'form-data'
+
 import fs from 'fs'
 import app from '../app'
 import { getConfig } from '../config'

src/test/x-forwarded-host.test.ts

@@ -59,7 +59,7 @@ describe('with X-Forwarded-Host header', () => {
         'x-forwarded-host': 'abcdefghijklmnopqrst.supabase.co',
       },
     })
-    expect(response.statusCode).toBe(400)
+    expect(response.statusCode).toBe(500)
     const responseJSON = JSON.parse(response.body)
     expect(responseJSON.message).toContain('http://abcdefghijklmnopqrst.supabase.co/rest/v1')
     await adminApp.inject({