feat: pgboss v10 (#696)

Author: fenos

package-lock.json

@@ -75,7 +75,7 @@
     "multistream": "^4.1.0",
     "object-sizeof": "^2.6.4",
     "pg": "^8.12.0",
-    "pg-boss": "^9.0.3",
+    "pg-boss": "^10.2.0",
     "pg-listen": "^1.7.0",
     "pino": "^8.15.4",
     "pino-logflare": "^0.4.2",

package.json

@@ -19,6 +19,7 @@ const build = (opts: FastifyServerOptions = {}, appInstance?: FastifyInstance):
   app.register(routes.jwks, { prefix: 'tenants' })
   app.register(routes.migrations, { prefix: 'migrations' })
   app.register(routes.s3Credentials, { prefix: 's3' })
+  app.register(routes.queue, { prefix: 'queue' })
 
   let registriesToMerge: Registry[] = []
 

src/admin-app.ts

@@ -115,6 +115,7 @@ type StorageConfigType = {
   pgQueueDeleteAfterDays?: number
   pgQueueArchiveCompletedAfterSeconds?: number
   pgQueueRetentionDays?: number
+  pgQueueConcurrentTasksPerQueue: number
   webhookURL?: string
   webhookApiKey?: string
   webhookQueuePullInterval?: number
@@ -410,6 +411,10 @@ export function getConfig(options?: { reload?: boolean }): StorageConfigType {
       10
     ),
     pgQueueRetentionDays: parseInt(getOptionalConfigFromEnv('PG_QUEUE_RETENTION_DAYS') || '2', 10),
+    pgQueueConcurrentTasksPerQueue: parseInt(
+      getOptionalConfigFromEnv('PG_QUEUE_CONCURRENT_TASKS_PER_QUEUE') || '50',
+      10
+    ),
 
     // Webhooks
     webhookURL: getOptionalConfigFromEnv('WEBHOOK_URL'),

src/config.ts

@@ -3,3 +3,4 @@ export { default as tenants } from './tenants'
 export { default as s3Credentials } from './s3'
 export { default as objects } from './objects'
 export { default as jwks } from './jwks'
+export { default as queue } from './queue'

src/http/routes/admin/index.ts

@@ -0,0 +1,20 @@
+import { FastifyInstance } from 'fastify'
+import apiKey from '../../plugins/apikey'
+import { getConfig } from '../../../config'
+import { UpgradePgBossV10 } from '@storage/events'
+
+const { pgQueueEnable } = getConfig()
+
+export default async function routes(fastify: FastifyInstance) {
+  fastify.register(apiKey)
+
+  fastify.post('/migrate/pgboss-v10', async (req, reply) => {
+    if (!pgQueueEnable) {
+      return reply.status(400).send({ message: 'Queue is not enabled' })
+    }
+
+    await UpgradePgBossV10.send({})
+
+    return reply.send({ message: 'Migration scheduled' })
+  })
+}

src/http/routes/admin/queue.ts

@@ -27,10 +27,10 @@ export class Cluster {
         cluster
           .getClusterSize()
           .then((size) => {
-            if (size !== Cluster.size) {
+            if (size && size !== Cluster.size) {
               clusterEvent.emit('change', { size })
+              Cluster.size = size
             }
-            Cluster.size = size
           })
           .catch((e) => {
             console.error('Error getting cluster size', e)

src/internal/cluster/cluster.ts

@@ -26,7 +26,7 @@ export class ClusterDiscoveryECS {
     const respMetadata = await axios.get(`${process.env.ECS_CONTAINER_METADATA_URI}/task`)
 
     const command = new ListTasksCommand({
-      serviceName: respMetadata.data.ServiceName,
+      family: respMetadata.data.Family,
       cluster: respMetadata.data.Cluster,
       desiredStatus: status,
     })

src/internal/cluster/ecs.ts

@@ -0,0 +1,56 @@
+import { DBMigration } from '@internal/database/migrations/types'
+import { ERRORS } from '@internal/errors'
+import { loadMigrationFiles } from 'postgres-migrations'
+import { getConfig } from '../../../config'
+
+const { dbMigrationFreezeAt } = getConfig()
+
+export const loadMigrationFilesCached = memoizePromise(loadMigrationFiles)
+
+export const localMigrationFiles = () => loadMigrationFiles('./migrations/tenant')
+
+export async function lastLocalMigrationName() {
+  const migrations = await loadMigrationFilesCached('./migrations/tenant')
+
+  if (!dbMigrationFreezeAt) {
+    return migrations[migrations.length - 1].name as keyof typeof DBMigration
+  }
+
+  const migrationIndex = migrations.findIndex((m) => m.name === dbMigrationFreezeAt)
+  if (migrationIndex === -1) {
+    throw ERRORS.InternalError(undefined, `Migration ${dbMigrationFreezeAt} not found`)
+  }
+  return migrations[migrationIndex].name as keyof typeof DBMigration
+}
+
+/**
+ * Memoizes a promise
+ * @param func
+ */
+function memoizePromise<T, Args extends unknown[]>(
+  func: (...args: Args) => Promise<T>
+): (...args: Args) => Promise<T> {
+  const cache = new Map<string, Promise<T>>()
+
+  function generateKey(args: Args): string {
+    return args
+      .map((arg) => {
+        if (typeof arg === 'object' && arg !== null) {
+          return Object.entries(arg).sort().toString()
+        }
+        return String(arg)
+      })
+      .join('|')
+  }
+
+  return async function (...args: Args): Promise<T> {
+    const key = generateKey(args)
+    if (cache.has(key)) {
+      return cache.get(key)!
+    }
+
+    const result = func(...args)
+    cache.set(key, result)
+    return result
+  }
+}

src/internal/database/migrations/files.ts

@@ -1,2 +1,3 @@
 export * from './migrate'
+export * from './files'
 export * from './types'