Process migration

packages/server/src/appMigrations/appMigrationMetadata.ts

@@ -1,5 +1,5 @@
-import { Duration, cache, db, env } from "@budibase/backend-core"
-import { Database, App, DocumentType, Document } from "@budibase/types"
+import { Duration, cache, context, db, env } from "@budibase/backend-core"
+import { Database, DocumentType, Document } from "@budibase/types"
 
 export interface AppMigrationDoc extends Document {
   version: string
@@ -7,23 +7,25 @@ export interface AppMigrationDoc extends Document {
 
 const EXPIRY_SECONDS = Duration.fromDays(1).toSeconds()
 
-async function populateFromDB(appId: string) {
+async function getFromDB(appId: string) {
   return db.doWithDB(
     appId,
     (db: Database) => {
-      return db.get<App>(DocumentType.APP_MIGRATION_METADATA)
+      return db.get<AppMigrationDoc>(DocumentType.APP_MIGRATION_METADATA)
     },
     { skip_setup: true }
   )
 }
 
+const getCacheKey = (appId: string) => `appmigrations_${env.VERSION}_${appId}`
+
 export async function getAppMigrationMetadata(appId: string): Promise<string> {
-  const cacheKey = `appmigrations_${env.VERSION}_${appId}`
+  const cacheKey = getCacheKey(appId)
 
   let metadata: AppMigrationDoc | undefined = await cache.get(cacheKey)
   if (!metadata || env.isDev()) {
     try {
-      metadata = await populateFromDB(appId)
+      metadata = await getFromDB(appId)
     } catch (err: any) {
       if (err.status !== 404) {
         throw err
@@ -37,3 +39,19 @@ export async function getAppMigrationMetadata(appId: string): Promise<string> {
 
   return metadata.version
 }
+
+export async function updateAppMigrationMetadata({
+  appId,
+  version,
+}: {
+  appId: string
+  version: string
+}): Promise<void> {
+  const db = context.getAppDB()
+  const appMigrationDoc = await getFromDB(appId)
+  await db.put({ ...appMigrationDoc, version })
+
+  const cacheKey = getCacheKey(appId)
+
+  await cache.destroy(cacheKey)
+}

packages/server/src/appMigrations/index.ts

@@ -1,4 +1,4 @@
-import queue from "./queue"
+import queue, { PROCESS_MIGRATION_TIMEOUT } from "./queue"
 import { getAppMigrationMetadata } from "./appMigrationMetadata"
 import { MIGRATIONS } from "./migrations"
 
@@ -13,9 +13,10 @@ export async function checkMissingMigrations(appId: string) {
         appId,
       },
       {
-        jobId: appId,
+        jobId: `${appId}_${latestMigration}`,
         removeOnComplete: true,
         removeOnFail: true,
+        timeout: PROCESS_MIGRATION_TIMEOUT,
       }
     )
   }

packages/server/src/appMigrations/queue.ts

@@ -1,13 +1,56 @@
-import { queue } from "@budibase/backend-core"
+import { context, locks, queue } from "@budibase/backend-core"
+import { LockName, LockType } from "@budibase/types"
 import { Job } from "bull"
+import { MIGRATIONS } from "./migrations"
+import {
+  getAppMigrationMetadata,
+  updateAppMigrationMetadata,
+} from "./appMigrationMetadata"
 
 const appMigrationQueue = queue.createQueue(queue.JobQueue.APP_MIGRATION)
 appMigrationQueue.process(processMessage)
 
+export async function runMigration(migrationId: string) {
+  await MIGRATIONS[migrationId].migration()
+}
+
+// TODO
+export const PROCESS_MIGRATION_TIMEOUT = 30000
+
 async function processMessage(job: Job) {
   const { appId } = job.data
+  console.log(`Processing app migration for "${appId}"`)
 
-  console.log(appId)
+  await locks.doWithLock(
+    {
+      name: LockName.APP_MIGRATION,
+      type: LockType.DEFAULT,
+      resource: appId,
+      ttl: PROCESS_MIGRATION_TIMEOUT,
+    },
+    async () => {
+      await context.doInAppContext(appId, async () => {
+        const currentVersion = await getAppMigrationMetadata(appId)
+
+        const pendingMigrations = Object.keys(MIGRATIONS).filter(
+          m => m > currentVersion
+        )
+
+        let index = 0
+        for (const migration of pendingMigrations) {
+          const counter = `(${++index}/${pendingMigrations.length})`
+          console.info(`Running migration ${migration}... ${counter}`, {
+            migration,
+            appId,
+          })
+          await runMigration(migration)
+          await updateAppMigrationMetadata({ appId, version: migration })
+        }
+      })
+    }
+  )
+
+  console.log(`App migration for "${appId}" processed`)
 }
 
 export default appMigrationQueue