Merge pull request #14386 from Budibase/feature-flag-sqs

Use new feature flag API for SQS.

charts/budibase/templates/app-service-deployment.yaml

@@ -212,10 +212,6 @@ spec:
         - name: APP_FEATURES
           value: "api"
         {{- end }}
-        {{- if .Values.globals.sqs.enabled }}
-        - name: SQS_SEARCH_ENABLE
-          value: "true"
-        {{- end }}
         {{- range .Values.services.apps.extraEnv }}
         - name: {{ .name }}
           value: {{ .value | quote }}

charts/budibase/templates/worker-service-deployment.yaml

@@ -198,10 +198,6 @@ spec:
         - name: NODE_TLS_REJECT_UNAUTHORIZED
           value: {{ .Values.services.tlsRejectUnauthorized }}
         {{ end }}
-        {{- if .Values.globals.sqs.enabled }}
-        - name: SQS_SEARCH_ENABLE
-          value: "true"
-        {{- end }}
         {{- range .Values.services.worker.extraEnv }}
         - name: {{ .name }}
           value: {{ .value | quote }}

hosting/docker-compose.build.yaml

@@ -29,7 +29,7 @@ services:
       BB_ADMIN_USER_EMAIL: ${BB_ADMIN_USER_EMAIL}
       BB_ADMIN_USER_PASSWORD: ${BB_ADMIN_USER_PASSWORD}
       PLUGINS_DIR: ${PLUGINS_DIR}
-      SQS_SEARCH_ENABLE: 1
+      TENANT_FEATURE_FLAGS: "*:SQS"
     depends_on:
       - worker-service
       - redis-service
@@ -57,7 +57,7 @@ services:
       INTERNAL_API_KEY: ${INTERNAL_API_KEY}
       REDIS_URL: redis-service:6379
       REDIS_PASSWORD: ${REDIS_PASSWORD}
-      SQS_SEARCH_ENABLE: 1
+      TENANT_FEATURE_FLAGS: "*:SQS"
     depends_on:
       - redis-service
       - minio-service

packages/account-portal

@@ -1 +1 @@
-Subproject commit c403315c5fa09a05dfd8fa4cd1890acfd8de0430
+Subproject commit c24374879d2b61516fabc24d7404e7da235be05e

packages/backend-core/src/db/couch/DatabaseImpl.ts

@@ -25,8 +25,8 @@ import { newid } from "../../docIds/newid"
 import { SQLITE_DESIGN_DOC_ID } from "../../constants"
 import { DDInstrumentedDatabase } from "../instrumentation"
 import { checkSlashesInUrl } from "../../helpers"
-import env from "../../environment"
 import { sqlLog } from "../../sql/utils"
+import { flags } from "../../features"
 
 const DATABASE_NOT_FOUND = "Database does not exist."
 
@@ -401,7 +401,10 @@ export class DatabaseImpl implements Database {
   }
 
   async destroy() {
-    if (env.SQS_SEARCH_ENABLE && (await this.exists(SQLITE_DESIGN_DOC_ID))) {
+    if (
+      (await flags.isEnabled("SQS")) &&
+      (await this.exists(SQLITE_DESIGN_DOC_ID))
+    ) {
       // delete the design document, then run the cleanup operation
       const definition = await this.get<SQLiteDefinition>(SQLITE_DESIGN_DOC_ID)
       // remove all tables - save the definition then trigger a cleanup

packages/backend-core/src/db/tests/index.spec.js

@@ -1,24 +0,0 @@
-require("../../../tests")
-const { structures } = require("../../../tests")
-const { getDB } = require("../db")
-
-describe("db", () => {
-  describe("getDB", () => {
-    it("returns a db", async () => {
-      const dbName = structures.db.id()
-      const db = getDB(dbName)
-      expect(db).toBeDefined()
-      expect(db.name).toBe(dbName)
-    })
-
-    it("uses the custom put function", async () => {
-      const db = getDB(structures.db.id())
-      let doc = { _id: "test" }
-      await db.put(doc)
-      doc = await db.get(doc._id)
-      expect(doc.createdAt).toBe(new Date().toISOString())
-      expect(doc.updatedAt).toBe(new Date().toISOString())
-      await db.destroy()
-    })
-  })
-})

packages/backend-core/src/db/tests/index.spec.ts

@@ -0,0 +1,32 @@
+import { doInTenant } from "../../context"
+import { structures } from "../../../tests"
+import { getDB } from "../db"
+
+interface Doc {
+  _id: string
+  createdAt?: string
+  updatedAt?: string
+}
+
+describe("db", () => {
+  describe("getDB", () => {
+    it("returns a db", async () => {
+      const dbName = structures.db.id()
+      const db = getDB(dbName)
+      expect(db).toBeDefined()
+      expect(db.name).toBe(dbName)
+    })
+
+    it("uses the custom put function", async () => {
+      await doInTenant("foo", async () => {
+        const db = getDB(structures.db.id())
+        let doc: Doc = { _id: "test" }
+        await db.put(doc)
+        doc = await db.get(doc._id)
+        expect(doc.createdAt).toBe(new Date().toISOString())
+        expect(doc.updatedAt).toBe(new Date().toISOString())
+        await db.destroy()
+      })
+    })
+  })
+})

packages/backend-core/src/db/utils.ts

@@ -1,6 +1,6 @@
 import env from "../environment"
 import { DEFAULT_TENANT_ID, SEPARATOR, DocumentType } from "../constants"
-import { getTenantId, getGlobalDBName, isMultiTenant } from "../context"
+import { getTenantId, getGlobalDBName } from "../context"
 import { doWithDB, directCouchAllDbs } from "./db"
 import { AppState, DeletedApp, getAppMetadata } from "../cache/appMetadata"
 import { isDevApp, isDevAppID, getProdAppID } from "../docIds/conversions"
@@ -206,34 +206,3 @@ export function pagination<T>(
     nextPage,
   }
 }
-
-export function isSqsEnabledForTenant(): boolean {
-  const tenantId = getTenantId()
-  if (!env.SQS_SEARCH_ENABLE) {
-    return false
-  }
-
-  // single tenant (self host and dev) always enabled if flag set
-  if (!isMultiTenant()) {
-    return true
-  }
-
-  // This is to guard against the situation in tests where tests pass because
-  // we're not actually using SQS, we're using Lucene and the tests pass due to
-  // parity.
-  if (env.isTest() && env.SQS_SEARCH_ENABLE_TENANTS.length === 0) {
-    throw new Error(
-      "to enable SQS you must specify a list of tenants in the SQS_SEARCH_ENABLE_TENANTS env var"
-    )
-  }
-
-  // Special case to enable all tenants, for testing in QA.
-  if (
-    env.SQS_SEARCH_ENABLE_TENANTS.length === 1 &&
-    env.SQS_SEARCH_ENABLE_TENANTS[0] === "*"
-  ) {
-    return true
-  }
-
-  return env.SQS_SEARCH_ENABLE_TENANTS.includes(tenantId)
-}

packages/backend-core/src/environment.ts

@@ -116,10 +116,6 @@ const environment = {
   API_ENCRYPTION_KEY: getAPIEncryptionKey(),
   COUCH_DB_URL: process.env.COUCH_DB_URL || "http://localhost:4005",
   COUCH_DB_SQL_URL: process.env.COUCH_DB_SQL_URL,
-  SQS_SEARCH_ENABLE: process.env.SQS_SEARCH_ENABLE,
-  SQS_SEARCH_ENABLE_TENANTS:
-    process.env.SQS_SEARCH_ENABLE_TENANTS?.split(",") || [],
-  SQS_MIGRATION_ENABLE: process.env.SQS_MIGRATION_ENABLE,
   COUCH_DB_USERNAME: process.env.COUCH_DB_USER,
   COUCH_DB_PASSWORD: process.env.COUCH_DB_PASSWORD,
   GOOGLE_CLIENT_ID: process.env.GOOGLE_CLIENT_ID,

packages/backend-core/src/features/index.ts

@@ -6,7 +6,7 @@ import tracer from "dd-trace"
 
 let posthog: PostHog | undefined
 export function init(opts?: PostHogOptions) {
-  if (env.POSTHOG_TOKEN && env.POSTHOG_API_HOST) {
+  if (env.POSTHOG_TOKEN && env.POSTHOG_API_HOST && !env.SELF_HOSTED) {
     console.log("initializing posthog client...")
     posthog = new PostHog(env.POSTHOG_TOKEN, {
       host: env.POSTHOG_API_HOST,
@@ -267,4 +267,5 @@ export class FlagSet<V extends Flag<any>, T extends { [key: string]: V }> {
 // default values set correctly and their types flow through the system.
 export const flags = new FlagSet({
   DEFAULT_VALUES: Flag.boolean(false),
+  SQS: Flag.boolean(false),
 })

packages/backend-core/src/features/tests/features.spec.ts

@@ -147,13 +147,13 @@ describe("feature flags", () => {
     }) => {
       const env: Partial<typeof environment> = {
         TENANT_FEATURE_FLAGS: environmentFlags,
+        SELF_HOSTED: false,
       }
 
       if (posthogFlags) {
         mockPosthogFlags(posthogFlags)
         env.POSTHOG_TOKEN = "test"
         env.POSTHOG_API_HOST = "https://us.i.posthog.com"
-        env.POSTHOG_PERSONAL_TOKEN = "test"
       }
 
       const ctx = { user: { license: { features: licenseFlags || [] } } }

packages/pro

@@ -1 +1 @@
-Subproject commit f2862855dcfeb4f4a28d6902daf411ec8f4a28e8
+Subproject commit d926494cc54e31441637180a745d0df7156a5120

packages/server/scripts/dev/manage.js

@@ -47,7 +47,7 @@ async function init() {
     HTTP_LOGGING: "0",
     VERSION: "0.0.0+local",
     PASSWORD_MIN_LENGTH: "1",
-    SQS_SEARCH_ENABLE: "1",
+    TENANT_FEATURE_FLAGS: "*:SQS",
   }
 
   config = { ...config, ...existingConfig }

packages/server/src/api/controllers/datasource.ts

@@ -27,6 +27,7 @@ import {
 import sdk from "../../sdk"
 import { builderSocket } from "../../websockets"
 import { isEqual } from "lodash"
+import { processTable } from "../../sdk/app/tables/getters"
 
 export async function fetch(ctx: UserCtx) {
   ctx.body = await sdk.datasources.fetch()
@@ -188,6 +189,7 @@ export async function update(
     for (let table of Object.values(datasource.entities)) {
       const oldTable = baseDatasource.entities?.[table.name]
       if (!oldTable || !isEqual(oldTable, table)) {
+        table = await processTable(table)
         builderSocket?.emitTableUpdate(ctx, table, { includeOriginator: true })
       }
     }

packages/server/src/api/controllers/row/views.ts

@@ -10,7 +10,7 @@ import {
 } from "@budibase/types"
 import { dataFilters } from "@budibase/shared-core"
 import sdk from "../../../sdk"
-import { db, context } from "@budibase/backend-core"
+import { db, context, features } from "@budibase/backend-core"
 import { enrichSearchContext } from "./utils"
 import { isExternalTableID } from "../../../integrations/utils"
 
@@ -40,7 +40,10 @@ export async function searchView(
     // Delete extraneous search params that cannot be overridden
     delete body.query.onEmptyFilter
 
-    if (!isExternalTableID(view.tableId) && !db.isSqsEnabledForTenant()) {
+    if (
+      !isExternalTableID(view.tableId) &&
+      !(await features.flags.isEnabled("SQS"))
+    ) {
       // Extract existing fields
       const existingFields =
         view.query
@@ -56,13 +59,14 @@ export async function searchView(
           }
         })
       })
-    } else
+    } else {
       query = {
         $and: {
           conditions: [query, body.query],
         },
       }
     }
+  }
 
   await context.ensureSnippetContext(true)
 

packages/server/src/api/controllers/table/index.ts

@@ -39,6 +39,7 @@ import {
   PROTECTED_EXTERNAL_COLUMNS,
   PROTECTED_INTERNAL_COLUMNS,
 } from "@budibase/shared-core"
+import { processTable } from "../../../sdk/app/tables/getters"
 
 function pickApi({ tableId, table }: { tableId?: string; table?: Table }) {
   if (table && isExternalTable(table)) {
@@ -118,6 +119,8 @@ export async function save(ctx: UserCtx<SaveTableRequest, SaveTableResponse>) {
   ctx.eventEmitter &&
     ctx.eventEmitter.emitTable(`table:save`, appId, { ...savedTable })
   ctx.body = savedTable
+
+  savedTable = await processTable(savedTable)
   builderSocket?.emitTableUpdate(ctx, cloneDeep(savedTable))
 }
 

packages/server/src/api/controllers/table/utils.ts

@@ -15,7 +15,7 @@ import { getViews, saveView } from "../view/utils"
 import viewTemplate from "../view/viewBuilder"
 import { cloneDeep } from "lodash/fp"
 import { quotas } from "@budibase/pro"
-import { events, context, db as dbCore } from "@budibase/backend-core"
+import { events, context, features } from "@budibase/backend-core"
 import {
   AutoFieldSubType,
   ContextUser,
@@ -332,7 +332,7 @@ class TableSaveFunctions {
       importRows: this.importRows,
       user: this.user,
     })
-    if (dbCore.isSqsEnabledForTenant()) {
+    if (await features.flags.isEnabled("SQS")) {
       await sdk.tables.sqs.addTable(table)
     }
     return table
@@ -526,7 +526,7 @@ export async function internalTableCleanup(table: Table, rows?: Row[]) {
   if (rows) {
     await AttachmentCleanup.tableDelete(table, rows)
   }
-  if (dbCore.isSqsEnabledForTenant()) {
+  if (await features.flags.isEnabled("SQS")) {
     await sdk.tables.sqs.removeTable(table)
   }
 }

packages/server/src/api/routes/tests/application.spec.ts

@@ -352,13 +352,13 @@ describe("/applications", () => {
       expect(events.app.unpublished).toHaveBeenCalledTimes(1)
     })
 
-    it("should be able to delete an app after SQS_SEARCH_ENABLE has been set but app hasn't been migrated", async () => {
+    it("should be able to delete an app after SQS has been set but app hasn't been migrated", async () => {
       const prodAppId = app.appId.replace("_dev", "")
       nock("http://localhost:10000")
         .delete(`/api/global/roles/${prodAppId}`)
         .reply(200, {})
 
-      await withCoreEnv({ SQS_SEARCH_ENABLE: "true" }, async () => {
+      await withCoreEnv({ TENANT_FEATURE_FLAGS: "*:SQS" }, async () => {
         await config.api.application.delete(app.appId)
       })
     })

packages/server/src/api/routes/tests/row.spec.ts

@@ -95,12 +95,9 @@ describe.each([
   let envCleanup: (() => void) | undefined
 
   beforeAll(async () => {
-    await withCoreEnv({ SQS_SEARCH_ENABLE: "true" }, () => config.init())
+    await withCoreEnv({ TENANT_FEATURE_FLAGS: "*SQS" }, () => config.init())
     if (isSqs) {
-      envCleanup = setCoreEnv({
+      envCleanup = setCoreEnv({ TENANT_FEATURE_FLAGS: "*SQS" })
-        SQS_SEARCH_ENABLE: "true",
-        SQS_SEARCH_ENABLE_TENANTS: [config.getTenantId()],
-      })
     }
 
     if (dsProvider) {

packages/server/src/api/routes/tests/search.spec.ts

@@ -67,11 +67,10 @@ describe.each([
   let rows: Row[]
 
   beforeAll(async () => {
-    await withCoreEnv({ SQS_SEARCH_ENABLE: "true" }, () => config.init())
+    await withCoreEnv({ TENANT_FEATURE_FLAGS: "*:SQS" }, () => config.init())
     if (isSqs) {
       envCleanup = setCoreEnv({
-        SQS_SEARCH_ENABLE: "true",
+        TENANT_FEATURE_FLAGS: "*:SQS",
-        SQS_SEARCH_ENABLE_TENANTS: [config.getTenantId()],
       })
     }
 

packages/server/src/api/routes/tests/templates.spec.ts

@@ -2,7 +2,7 @@ import * as setup from "./utilities"
 import path from "path"
 import nock from "nock"
 import { generator } from "@budibase/backend-core/tests"
-import { withEnv as withCoreEnv } from "@budibase/backend-core"
+import { withEnv as withCoreEnv, env as coreEnv } from "@budibase/backend-core"
 
 interface App {
   background: string
@@ -85,9 +85,8 @@ describe("/templates", () => {
     it.each(["sqs", "lucene"])(
       `should be able to create an app from a template (%s)`,
       async source => {
-        const env = {
+        const env: Partial<typeof coreEnv> = {
-          SQS_SEARCH_ENABLE: source === "sqs" ? "true" : "false",
+          TENANT_FEATURE_FLAGS: source === "sqs" ? "*:SQS" : "",
-          SQS_SEARCH_ENABLE_TENANTS: [config.getTenantId()],
         }
 
         await withCoreEnv(env, async () => {

packages/server/src/api/routes/tests/viewV2.spec.ts

@@ -94,13 +94,12 @@ describe.each([
   }
 
   beforeAll(async () => {
-    await withCoreEnv({ SQS_SEARCH_ENABLE: isSqs ? "true" : "false" }, () =>
+    await withCoreEnv({ TENANT_FEATURE_FLAGS: isSqs ? "*:SQS" : "" }, () =>
       config.init()
     )
     if (isSqs) {
       envCleanup = setCoreEnv({
-        SQS_SEARCH_ENABLE: "true",
+        TENANT_FEATURE_FLAGS: "*:SQS",
-        SQS_SEARCH_ENABLE_TENANTS: [config.getTenantId()],
       })
     }
 

packages/server/src/appMigrations/migrations.ts

@@ -1,6 +1,5 @@
 // This file should never be manually modified, use `yarn add-app-migration` in order to add a new one
 
-import { env } from "@budibase/backend-core"
 import { AppMigration } from "."
 
 import m20240604153647_initial_sqs from "./migrations/20240604153647_initial_sqs"
@@ -10,6 +9,5 @@ export const MIGRATIONS: AppMigration[] = [
   {
     id: "20240604153647_initial_sqs",
     func: m20240604153647_initial_sqs,
-    disabled: !(env.SQS_MIGRATION_ENABLE || env.SQS_SEARCH_ENABLE),
   },
 ]

packages/server/src/appMigrations/migrations/20240604153647_initial_sqs.ts

@@ -1,4 +1,4 @@
-import { context, env } from "@budibase/backend-core"
+import { context } from "@budibase/backend-core"
 import { allLinkDocs } from "../../db/utils"
 import LinkDocumentImpl from "../../db/linkedRows/LinkDocument"
 import sdk from "../../sdk"
@@ -36,16 +36,6 @@ const migration = async () => {
 
   // at the end make sure design doc is ready
   await sdk.tables.sqs.syncDefinition()
-  // only do initial search if environment is using SQS already
-  // initial search makes sure that all the indexes have been created
-  // and are ready to use, avoiding any initial waits for large tables
-  if (env.SQS_MIGRATION_ENABLE || env.SQS_SEARCH_ENABLE) {
-    const tables = await sdk.tables.getAllInternalTables()
-    // do these one by one - running in parallel could cause problems
-    for (let table of tables) {
-      await db.sql(`select * from ${table._id} limit 1`)
-    }
-  }
 }
 
 export default migration

packages/server/src/appMigrations/migrations/tests/20240604153647_initial_sqs.spec.ts

@@ -18,7 +18,7 @@ import {
 } from "../../../db/utils"
 import { processMigrations } from "../../migrationsProcessor"
 import migration from "../20240604153647_initial_sqs"
-import { AppMigration } from "src/appMigrations"
+import { AppMigration, updateAppMigrationMetadata } from "../../"
 import sdk from "../../../sdk"
 
 const MIGRATIONS: AppMigration[] = [
@@ -70,24 +70,17 @@ function oldLinkDocument(): Omit<LinkDocument, "tableId"> {
   }
 }
 
-type SQSEnvVar = "SQS_MIGRATION_ENABLE" | "SQS_SEARCH_ENABLE"
+async function sqsDisabled(cb: () => Promise<void>) {
-
+  await withCoreEnv({ TENANT_FEATURE_FLAGS: "*:!SQS" }, cb)
-async function sqsDisabled(envVar: SQSEnvVar, cb: () => Promise<void>) {
-  await withCoreEnv({ [envVar]: "", SQS_SEARCH_ENABLE_TENANTS: [] }, cb)
 }
 
-async function sqsEnabled(envVar: SQSEnvVar, cb: () => Promise<void>) {
+async function sqsEnabled(cb: () => Promise<void>) {
-  await withCoreEnv(
+  await withCoreEnv({ TENANT_FEATURE_FLAGS: "*:SQS" }, cb)
-    { [envVar]: "1", SQS_SEARCH_ENABLE_TENANTS: [config.getTenantId()] },
-    cb
-  )
 }
 
-describe.each(["SQS_MIGRATION_ENABLE", "SQS_SEARCH_ENABLE"] as SQSEnvVar[])(
+describe("SQS migration", () => {
-  "SQS migration with (%s)",
-  envVar => {
   beforeAll(async () => {
-      await sqsDisabled(envVar, async () => {
+    await sqsDisabled(async () => {
       await config.init()
       const table = await config.api.table.save(basicTable())
       tableId = table._id!
@@ -97,10 +90,19 @@ describe.each(["SQS_MIGRATION_ENABLE", "SQS_SEARCH_ENABLE"] as SQSEnvVar[])(
     })
   })
 
+  beforeEach(async () => {
+    await config.doInTenant(async () => {
+      await updateAppMigrationMetadata({
+        appId: config.getAppId(),
+        version: "",
+      })
+    })
+  })
+
   it("test migration runs as expected against an older DB", async () => {
     const db = dbCore.getDB(config.appId!)
     // confirm nothing exists initially
-      await sqsDisabled(envVar, async () => {
+    await sqsDisabled(async () => {
       let error: any | undefined
       try {
         await db.get(SQLITE_DESIGN_DOC_ID)
@@ -110,7 +112,8 @@ describe.each(["SQS_MIGRATION_ENABLE", "SQS_SEARCH_ENABLE"] as SQSEnvVar[])(
       expect(error).toBeDefined()
       expect(error.status).toBe(404)
     })
-      await sqsEnabled(envVar, async () => {
+
+    await sqsEnabled(async () => {
       await processMigrations(config.appId!, MIGRATIONS)
       const designDoc = await db.get<SQLiteDefinition>(SQLITE_DESIGN_DOC_ID)
       expect(designDoc.sql.tables).toBeDefined()
@@ -137,5 +140,4 @@ describe.each(["SQS_MIGRATION_ENABLE", "SQS_SEARCH_ENABLE"] as SQSEnvVar[])(
       expect(linkDoc.doc2.rowId).toEqual(rowId1)
     })
   })
-  }
+})
-)

packages/server/src/sdk/app/rows/search.ts

@@ -12,7 +12,7 @@ import { ExportRowsParams, ExportRowsResult } from "./search/types"
 import { dataFilters } from "@budibase/shared-core"
 import sdk from "../../index"
 import { searchInputMapping } from "./search/utils"
-import { db as dbCore } from "@budibase/backend-core"
+import { features } from "@budibase/backend-core"
 import tracer from "dd-trace"
 import { getQueryableFields, removeInvalidFilters } from "./queryUtils"
 
@@ -90,7 +90,7 @@ export async function search(
     if (isExternalTable) {
       span?.addTags({ searchType: "external" })
       result = await external.search(options, table)
-    } else if (dbCore.isSqsEnabledForTenant()) {
+    } else if (await features.flags.isEnabled("SQS")) {
       span?.addTags({ searchType: "sqs" })
       result = await internal.sqs.search(options, table)
     } else {

packages/server/src/sdk/app/rows/search/tests/search.spec.ts

@@ -41,14 +41,13 @@ describe.each([
   let table: Table
 
   beforeAll(async () => {
-    await withCoreEnv({ SQS_SEARCH_ENABLE: isSqs ? "true" : "false" }, () =>
+    await withCoreEnv({ TENANT_FEATURE_FLAGS: isSqs ? "*:SQS" : "" }, () =>
       config.init()
     )
 
     if (isSqs) {
       envCleanup = setCoreEnv({
-        SQS_SEARCH_ENABLE: "true",
+        TENANT_FEATURE_FLAGS: "*:SQS",
-        SQS_SEARCH_ENABLE_TENANTS: [config.getTenantId()],
       })
     }
 

packages/server/src/sdk/app/tables/getters.ts

@@ -1,4 +1,4 @@
-import { context, db as dbCore, env } from "@budibase/backend-core"
+import { context, features } from "@budibase/backend-core"
 import { getTableParams } from "../../../db/utils"
 import {
   breakExternalTableId,
@@ -16,7 +16,7 @@ import {
 import datasources from "../datasources"
 import sdk from "../../../sdk"
 
-export function processTable(table: Table): Table {
+export async function processTable(table: Table): Promise<Table> {
   if (!table) {
     return table
   }
@@ -33,20 +33,21 @@ export function processTable(table: Table): Table {
       sourceId: table.sourceId || INTERNAL_TABLE_SOURCE_ID,
       sourceType: TableSourceType.INTERNAL,
     }
-    if (dbCore.isSqsEnabledForTenant()) {
+    const sqsEnabled = await features.flags.isEnabled("SQS")
-      processed.sql = !!env.SQS_SEARCH_ENABLE
+    if (sqsEnabled) {
+      processed.sql = true
     }
     return processed
   }
 }
 
-export function processTables(tables: Table[]): Table[] {
+export async function processTables(tables: Table[]): Promise<Table[]> {
-  return tables.map(table => processTable(table))
+  return await Promise.all(tables.map(table => processTable(table)))
 }
 
-function processEntities(tables: Record<string, Table>) {
+async function processEntities(tables: Record<string, Table>) {
   for (let key of Object.keys(tables)) {
-    tables[key] = processTable(tables[key])
+    tables[key] = await processTable(tables[key])
   }
   return tables
 }
@@ -60,7 +61,7 @@ export async function getAllInternalTables(db?: Database): Promise<Table[]> {
       include_docs: true,
     })
   )
-  return processTables(internalTables.rows.map(row => row.doc!))
+  return await processTables(internalTables.rows.map(row => row.doc!))
 }
 
 async function getAllExternalTables(): Promise<Table[]> {
@@ -72,7 +73,7 @@ async function getAllExternalTables(): Promise<Table[]> {
       final = final.concat(Object.values(entities))
     }
   }
-  return processTables(final)
+  return await processTables(final)
 }
 
 export async function getExternalTable(
@@ -97,7 +98,7 @@ export async function getTable(tableId: string): Promise<Table> {
   } else {
     output = await db.get<Table>(tableId)
   }
-  return processTable(output)
+  return await processTable(output)
 }
 
 export async function getAllTables() {
@@ -105,7 +106,7 @@ export async function getAllTables() {
     getAllInternalTables(),
     getAllExternalTables(),
   ])
-  return processTables([...internal, ...external])
+  return await processTables([...internal, ...external])
 }
 
 export async function getExternalTablesInDatasource(
@@ -115,7 +116,7 @@ export async function getExternalTablesInDatasource(
   if (!datasource || !datasource.entities) {
     throw new Error("Datasource is not configured fully.")
   }
-  return processEntities(datasource.entities)
+  return await processEntities(datasource.entities)
 }
 
 export async function getTables(tableIds: string[]): Promise<Table[]> {
@@ -139,7 +140,7 @@ export async function getTables(tableIds: string[]): Promise<Table[]> {
     })
     tables = tables.concat(internalTables)
   }
-  return processTables(tables)
+  return await processTables(tables)
 }
 
 export function enrichViewSchemas(table: Table): TableResponse {

packages/server/src/tests/jestEnv.ts

@@ -3,6 +3,8 @@ import { tmpdir } from "os"
 process.env.SELF_HOSTED = "1"
 process.env.NODE_ENV = "jest"
 process.env.MULTI_TENANCY = "1"
+process.env.APP_PORT = "0"
+process.env.WORKER_PORT = "0"
 // @ts-ignore
 process.env.BUDIBASE_DIR = tmpdir("budibase-unittests")
 process.env.LOG_LEVEL = process.env.LOG_LEVEL || "error"

packages/server/src/utilities/rowProcessor/index.ts

@@ -3,7 +3,7 @@ import { fixAutoColumnSubType, processFormulas } from "./utils"
 import {
   cache,
   context,
-  db,
+  features,
   HTTPError,
   objectStore,
   utils,
@@ -350,7 +350,7 @@ export async function outputProcessing<T extends Row[] | Row>(
   }
   // remove null properties to match internal API
   const isExternal = isExternalTableID(table._id!)
-  if (isExternal || db.isSqsEnabledForTenant()) {
+  if (isExternal || (await features.flags.isEnabled("SQS"))) {
     for (const row of enriched) {
       for (const key of Object.keys(row)) {
         if (row[key] === null) {

packages/server/src/utilities/rowProcessor/tests/outputProcessing.spec.ts

@@ -8,15 +8,9 @@ import {
 } from "@budibase/types"
 import { outputProcessing } from ".."
 import { generator, structures } from "@budibase/backend-core/tests"
+import { setEnv as setCoreEnv } from "@budibase/backend-core"
 import * as bbReferenceProcessor from "../bbReferenceProcessor"
-
+import TestConfiguration from "../../../tests/utilities/TestConfiguration"
-jest.mock("@budibase/backend-core", () => ({
-  ...jest.requireActual("@budibase/backend-core"),
-  db: {
-    ...jest.requireActual("@budibase/backend-core").db,
-    isSqsEnabledForTenant: () => true,
-  },
-}))
 
 jest.mock("../bbReferenceProcessor", (): typeof bbReferenceProcessor => ({
   processInputBBReference: jest.fn(),
@@ -26,8 +20,24 @@ jest.mock("../bbReferenceProcessor", (): typeof bbReferenceProcessor => ({
 }))
 
 describe("rowProcessor - outputProcessing", () => {
+  const config = new TestConfiguration()
+  let cleanupEnv: () => void = () => {}
+
+  beforeAll(async () => {
+    await config.init()
+  })
+
+  afterAll(async () => {
+    config.end()
+  })
+
   beforeEach(() => {
     jest.resetAllMocks()
+    cleanupEnv = setCoreEnv({ TENANT_FEATURE_FLAGS: "*SQS" })
+  })
+
+  afterEach(() => {
+    cleanupEnv()
   })
 
   const processOutputBBReferenceMock =
@@ -36,6 +46,7 @@ describe("rowProcessor - outputProcessing", () => {
     bbReferenceProcessor.processOutputBBReferences as jest.Mock
 
   it("fetches single user references given a populated field", async () => {
+    await config.doInContext(config.getAppId(), async () => {
       const table: Table = {
         _id: generator.guid(),
         name: "TestTable",
@@ -75,16 +86,17 @@ describe("rowProcessor - outputProcessing", () => {
 
       expect(result).toEqual({ name: "Jack", user })
 
-    expect(bbReferenceProcessor.processOutputBBReference).toHaveBeenCalledTimes(
+      expect(
-      1
+        bbReferenceProcessor.processOutputBBReference
-    )
+      ).toHaveBeenCalledTimes(1)
-    expect(bbReferenceProcessor.processOutputBBReference).toHaveBeenCalledWith(
+      expect(
-      "123",
+        bbReferenceProcessor.processOutputBBReference
-      BBReferenceFieldSubType.USER
+      ).toHaveBeenCalledWith("123", BBReferenceFieldSubType.USER)
-    )
+    })
   })
 
   it("fetches users references given a populated field", async () => {
+    await config.doInContext(config.getAppId(), async () => {
       const table: Table = {
         _id: generator.guid(),
         name: "TestTable",
@@ -127,13 +139,14 @@ describe("rowProcessor - outputProcessing", () => {
       expect(
         bbReferenceProcessor.processOutputBBReferences
       ).toHaveBeenCalledTimes(1)
-    expect(bbReferenceProcessor.processOutputBBReferences).toHaveBeenCalledWith(
+      expect(
-      "123",
+        bbReferenceProcessor.processOutputBBReferences
-      BBReferenceFieldSubType.USER
+      ).toHaveBeenCalledWith("123", BBReferenceFieldSubType.USER)
-    )
+    })
   })
 
   it("should handle attachment list correctly", async () => {
+    await config.doInContext(config.getAppId(), async () => {
       const table: Table = {
         _id: generator.guid(),
         name: "TestTable",
@@ -175,8 +188,10 @@ describe("rowProcessor - outputProcessing", () => {
       const output3 = await outputProcessing(table, row, { squash: false })
       expect(output3.attach[0].url).toBe("aaaa")
     })
+  })
 
   it("should handle single attachment correctly", async () => {
+    await config.doInContext(config.getAppId(), async () => {
       const table: Table = {
         _id: generator.guid(),
         name: "TestTable",
@@ -216,8 +231,10 @@ describe("rowProcessor - outputProcessing", () => {
       const output3 = await outputProcessing(table, row, { squash: false })
       expect(output3.attach.url).toBe("aaaa")
     })
+  })
 
   it("process output even when the field is not empty", async () => {
+    await config.doInContext(config.getAppId(), async () => {
       const table: Table = {
         _id: generator.guid(),
         name: "TestTable",
@@ -257,8 +274,10 @@ describe("rowProcessor - outputProcessing", () => {
         bbReferenceProcessor.processOutputBBReferences
       ).toHaveBeenCalledTimes(1)
     })
+  })
 
   it("does not fetch bb references when not in the schema", async () => {
+    await config.doInContext(config.getAppId(), async () => {
       const table: Table = {
         _id: generator.guid(),
         name: "TestTable",
@@ -299,3 +318,4 @@ describe("rowProcessor - outputProcessing", () => {
       ).not.toHaveBeenCalled()
     })
   })
+})

packages/server/src/websockets/builder.ts

@@ -16,7 +16,6 @@ import { gridSocket } from "./index"
 import { clearLock, updateLock } from "../utilities/redis"
 import { Socket } from "socket.io"
 import { BuilderSocketEvent } from "@budibase/shared-core"
-import { processTable } from "../sdk/app/tables/getters"
 
 export default class BuilderSocket extends BaseSocket {
   constructor(app: Koa, server: http.Server) {
@@ -102,10 +101,9 @@ export default class BuilderSocket extends BaseSocket {
   }
 
   emitTableUpdate(ctx: any, table: Table, options?: EmitOptions) {
-    // This was added to make sure that sourceId is always present when
+    if (table.sourceId == null || table.sourceId === "") {
-    // sending this message to clients. Without this, tables without a
+      throw new Error("Table sourceId is not set")
-    // sourceId (e.g. ta_users) won't get correctly updated client-side.
+    }
-    table = processTable(table)
 
     this.emitToRoom(
       ctx,

packages/worker/scripts/dev/manage.js

@@ -30,7 +30,7 @@ async function init() {
     HTTP_LOGGING: "0",
     VERSION: "0.0.0+local",
     PASSWORD_MIN_LENGTH: "1",
-    SQS_SEARCH_ENABLE: "1",
+    TENANT_FEATURE_FLAGS: "*:SQS",
   }
 
   config = { ...config, ...existingConfig }

packages/worker/src/api/controllers/system/environment.ts

@@ -1,6 +1,6 @@
 import { Ctx, MaintenanceType } from "@budibase/types"
 import env from "../../../environment"
-import { env as coreEnv, db as dbCore } from "@budibase/backend-core"
+import { env as coreEnv, db as dbCore, features } from "@budibase/backend-core"
 import nodeFetch from "node-fetch"
 
 let sqsAvailable: boolean
@@ -29,7 +29,7 @@ async function isSqsAvailable() {
 }
 
 async function isSqsMissing() {
-  return coreEnv.SQS_SEARCH_ENABLE && !(await isSqsAvailable())
+  return (await features.flags.isEnabled("SQS")) && !(await isSqsAvailable())
 }
 
 export const fetch = async (ctx: Ctx) => {

packages/worker/src/api/index.ts

@@ -4,12 +4,8 @@ const compress = require("koa-compress")
 
 import zlib from "zlib"
 import { routes } from "./routes"
-import { middleware as pro, sdk } from "@budibase/pro"
+import { middleware as pro } from "@budibase/pro"
-import { auth, middleware, env } from "@budibase/backend-core"
+import { auth, middleware } from "@budibase/backend-core"
-
-if (env.SQS_SEARCH_ENABLE) {
-  sdk.auditLogs.useSQLSearch()
-}
 
 const PUBLIC_ENDPOINTS = [
   // deprecated single tenant sso callback

packages/worker/src/api/routes/global/tests/auditLogs.spec.ts

@@ -1,7 +1,6 @@
 import { mocks, structures } from "@budibase/backend-core/tests"
-import { context, events } from "@budibase/backend-core"
+import { context, events, setEnv as setCoreEnv } from "@budibase/backend-core"
 import { Event, IdentityType } from "@budibase/types"
-import { auditLogs } from "@budibase/pro"
 import { TestConfiguration } from "../../../../tests"
 
 mocks.licenses.useAuditLogs()
@@ -15,15 +14,17 @@ const APP_ID = "app_1"
 
 describe.each(["lucene", "sql"])("/api/global/auditlogs (%s)", method => {
   const config = new TestConfiguration()
+  let envCleanup: (() => void) | undefined
 
   beforeAll(async () => {
     if (method === "sql") {
-      auditLogs.useSQLSearch()
+      envCleanup = setCoreEnv({ TENANT_FEATURE_FLAGS: "*:SQS" })
     }
     await config.beforeAll()
   })
 
   afterAll(async () => {
+    envCleanup?.()
     await config.afterAll()
   })
 

packages/worker/src/tests/api/environment.ts

@@ -9,6 +9,7 @@ export class EnvironmentAPI extends TestAPI {
   getEnvironment = () => {
     return this.request
       .get(`/api/system/environment`)
+      .set(this.config.defaultHeaders())
       .expect("Content-Type", /json/)
       .expect(200)
   }