feat(server): cli and cross-server-sync multiregion support (#3527)

* feat(server): cross-server-sync multiregion ready

* fixed various db commands

* db cli works

* final changes

packages/server/db/knex.ts

@@ -16,7 +16,7 @@ config.log = {
   }
 }
 
-dbStartupLogger.info(`Loaded knex conf for ${env}`)
+dbStartupLogger.debug(`Loaded knex conf for ${env}`)
 
 const knexInstance = knex(config)
 

packages/server/modules/cli/commands/db/helpers/index.ts

@@ -0,0 +1,19 @@
+import { getAllRegisteredDbClients } from '@/modules/multiregion/dbSelector'
+
+export type CommonDbArgs = {
+  regionKey?: string
+}
+
+export const getTargettedDbClients = async (params: { regionKey?: string }) => {
+  const { regionKey } = params
+  const dbs = (await getAllRegisteredDbClients()).filter((db) => {
+    if (!regionKey) return true
+    if (regionKey === 'main') return db.isMain
+    if (regionKey.includes(',')) {
+      return regionKey.split(',').includes(db.regionKey)
+    }
+    return db.regionKey === regionKey
+  })
+
+  return dbs
+}

packages/server/modules/cli/commands/db/migrate.ts

@@ -5,7 +5,14 @@ const command: CommandModule = {
   command: 'migrate',
   describe: 'Migration specific commands',
   builder(yargs) {
-    return yargs.commandDir('migrate', { extensions: ['js', 'ts'] }).demandCommand()
+    return yargs
+      .commandDir('migrate', { extensions: ['js', 'ts'] })
+      .demandCommand()
+      .option('regionKey', {
+        type: 'string',
+        describe:
+          'Region key to run migrations for. If not set, will run on all registered DBs. If set to "main", will only run in main DB. Can be comma-delimited.'
+      })
   },
   handler: noop
 }

packages/server/modules/cli/commands/db/migrate/create.ts

@@ -3,6 +3,7 @@ import { appRoot } from '@/bootstrap'
 import fs from 'fs/promises'
 import { logger } from '@/logging/logging'
 import { CommandModule } from 'yargs'
+import { ensureError } from '@speckle/shared'
 
 /** @type {import('yargs').CommandModule} */
 const command: CommandModule<unknown, { name: string; module: string }> = {
@@ -25,10 +26,16 @@ const command: CommandModule<unknown, { name: string; module: string }> = {
 
     try {
       await fs.access(migrationDir)
-    } catch {
-      throw new Error(
-        `Migration directory '${migrationDir}' is not accessible! Check if it exists.`
-      )
+    } catch (e) {
+      if (ensureError(e).message.toLowerCase().includes('no such file or directory')) {
+        // Try to create it
+        await fs.mkdir(migrationDir, { recursive: true })
+      } else {
+        throw new Error(
+          `Migration directory '${migrationDir}' is not accessible! Check if it exists.`,
+          { cause: e }
+        )
+      }
     }
 
     logger.info('Creating migration...')

packages/server/modules/cli/commands/db/migrate/down.ts

@@ -1,19 +1,20 @@
-import knex from '@/db/knex'
 import { logger } from '@/logging/logging'
+import { CommonDbArgs, getTargettedDbClients } from '@/modules/cli/commands/db/helpers'
 import { CommandModule } from 'yargs'
 
-const command: CommandModule<unknown, { times: number }> = {
+const command: CommandModule<unknown, { times: number } & CommonDbArgs> = {
   command: 'down [times]',
   describe: 'Undo last migration',
-  builder(yargs) {
-    return yargs.positional('times', {
-      describe: 'Number of migrations to undo',
+  builder: {
+    times: {
       type: 'number',
-      default: 1
-    })
+      default: 1,
+      describe: 'Number of migrations to undo'
+    }
   },
   async handler(argv) {
-    const howManyTimes = argv.times || 1
+    const { times, regionKey } = argv
+    const howManyTimes = times || 1
 
     logger.info(
       howManyTimes === 1
@@ -21,8 +22,12 @@ const command: CommandModule<unknown, { times: number }> = {
         : `Undoing last ${howManyTimes} migrations...`
     )
 
-    for (let i = 0; i < howManyTimes; i++) {
-      await knex.migrate.down()
+    const dbs = await getTargettedDbClients({ regionKey })
+    for (const db of dbs) {
+      logger.info(`Migrating DB ${db.regionKey}...`)
+      for (let i = 0; i < howManyTimes; i++) {
+        await db.client.migrate.down()
+      }
     }
 
     logger.info('Completed!')

packages/server/modules/cli/commands/db/migrate/latest.ts

@@ -1,26 +1,18 @@
-import knex from '@/db/knex'
 import { logger } from '@/logging/logging'
-import { getRegisteredRegionClients } from '@/modules/multiregion/dbSelector'
-import { isTestEnv } from '@/modules/shared/helpers/envHelper'
-import { mochaHooks } from '@/test/hooks'
+import { CommonDbArgs, getTargettedDbClients } from '@/modules/cli/commands/db/helpers'
 import { CommandModule } from 'yargs'
 
-const command: CommandModule = {
+const command: CommandModule<unknown, CommonDbArgs> = {
   command: 'latest',
   describe: 'Run all migrations that have not yet been run',
-  async handler() {
-    logger.info('Running latest migration...')
+  async handler(argv) {
+    logger.info('Running latest migrations on DB instances!')
+    const { regionKey } = argv
 
-    // In tests we want different logic - just run beforeAll
-    if (isTestEnv()) {
-      // Run before hooks, to properly initialize everything
-      await (mochaHooks.beforeAll as () => Promise<void>)()
-    } else {
-      const regionDbs = await getRegisteredRegionClients()
-      const dbs = [knex, ...Object.values(regionDbs)]
-      for (const db of dbs) {
-        await db.migrate.latest()
-      }
+    const dbs = await getTargettedDbClients({ regionKey })
+    for (const db of dbs) {
+      logger.info(`Running latest on DB ${db.regionKey}...`)
+      await db.client.migrate.latest()
     }
 
     logger.info('Completed running migration')

packages/server/modules/cli/commands/db/migrate/rollback.ts

@@ -1,28 +1,22 @@
-import knex from '@/db/knex'
 import { logger } from '@/logging/logging'
-import { getRegisteredRegionClients } from '@/modules/multiregion/dbSelector'
-import { isTestEnv } from '@/modules/shared/helpers/envHelper'
-import { mochaHooks, resetPubSubFactory } from '@/test/hooks'
+import { CommonDbArgs, getTargettedDbClients } from '@/modules/cli/commands/db/helpers'
+import { resetPubSubFactory } from '@/test/hooks'
 import { CommandModule } from 'yargs'
 
-const command: CommandModule = {
+const command: CommandModule<unknown, CommonDbArgs> = {
   command: 'rollback',
   describe: 'Roll back all migrations',
-  async handler() {
+  async handler(argv) {
+    const { regionKey } = argv
+
     logger.info('Rolling back migrations...')
 
-    if (isTestEnv()) {
-      // Run before hooks, to properly initialize everything first
-      await (mochaHooks.beforeAll as () => Promise<void>)()
-    }
-
-    const regionDbs = await getRegisteredRegionClients()
-    const dbs = [knex, ...Object.values(regionDbs)]
-
+    const dbs = await getTargettedDbClients({ regionKey })
     for (const db of dbs) {
-      const resetPubSub = resetPubSubFactory({ db })
+      logger.info(`Rolling back DB ${db.regionKey}...`)
+      const resetPubSub = resetPubSubFactory({ db: db.client })
       await resetPubSub()
-      await db.migrate.rollback(undefined, true)
+      await db.client.migrate.rollback(undefined, true)
     }
 
     logger.info('Completed rolling back migrations')

packages/server/modules/cli/commands/db/migrate/up.ts

@@ -1,13 +1,21 @@
-import knex from '@/db/knex'
 import { logger } from '@/logging/logging'
+import { CommonDbArgs, getTargettedDbClients } from '@/modules/cli/commands/db/helpers'
 import { CommandModule } from 'yargs'
 
-const command: CommandModule = {
+const command: CommandModule<unknown, CommonDbArgs> = {
   command: 'up',
   describe: 'Run next migration that has not yet been run',
-  async handler() {
+  async handler(argv) {
+    const { regionKey } = argv
+
     logger.info('Running next migration...')
-    await knex.migrate.up()
+
+    const dbs = await getTargettedDbClients({ regionKey })
+    for (const db of dbs) {
+      logger.info(`Running next migration on DB ${db.regionKey}...`)
+      await db.client.migrate.up()
+    }
+
     logger.info('Completed running next migration')
   }
 }

packages/server/modules/cli/commands/db/seed/commits.ts

@@ -2,7 +2,11 @@ import { db } from '@/db/knex'
 import { cliLogger } from '@/logging/logging'
 import { getStreamFactory } from '@/modules/core/repositories/streams'
 import { getUserFactory } from '@/modules/core/repositories/users'
-import { BasicTestCommit, createTestCommits } from '@/test/speckle-helpers/commitHelper'
+import { getProjectDbClient } from '@/modules/multiregion/dbSelector'
+import {
+  BasicTestCommit,
+  createTestCommitsFactory
+} from '@/test/speckle-helpers/commitHelper'
 import dayjs from 'dayjs'
 import { times } from 'lodash'
 import { CommandModule } from 'yargs'
@@ -36,6 +40,8 @@ const command: CommandModule<
     const streamId = argv.streamId
     const authorId = argv.authorId
     const date = dayjs().toISOString()
+    const projectDb = await getProjectDbClient({ projectId: streamId })
+    const createTestCommits = createTestCommitsFactory({ db: projectDb })
 
     const user = await getUser(authorId)
     if (!user?.id) {

packages/server/modules/cli/commands/download/commit.ts

@@ -61,7 +61,7 @@ import { publish } from '@/modules/shared/utils/subscriptions'
 import { getUserFactory } from '@/modules/core/repositories/users'
 import { createObjectFactory } from '@/modules/core/services/objects/management'
 import { getProjectDbClient } from '@/modules/multiregion/dbSelector'
-import { db } from '@/db/knex'
+import { db, mainDb } from '@/db/knex'
 
 const command: CommandModule<
   unknown,
@@ -148,7 +148,7 @@ const command: CommandModule<
       addCommentCreatedActivity: addCommentCreatedActivityFactory({
         getViewerResourcesFromLegacyIdentifiers,
         getViewerResourceItemsUngrouped,
-        saveActivity: saveActivityFactory({ db: projectDb }),
+        saveActivity: saveActivityFactory({ db: mainDb }),
         publish
       })
     })
@@ -165,7 +165,7 @@ const command: CommandModule<
           getCommentsResources: getCommentsResourcesFactory({ db: projectDb }),
           getViewerResourcesFromLegacyIdentifiers
         }),
-        saveActivity: saveActivityFactory({ db: projectDb }),
+        saveActivity: saveActivityFactory({ db: mainDb }),
         publish
       })
     })
@@ -180,7 +180,7 @@ const command: CommandModule<
       markCommitBranchUpdated: markCommitBranchUpdatedFactory({ db: projectDb }),
       versionsEventEmitter: VersionsEmitter.emit,
       addCommitCreatedActivity: addCommitCreatedActivityFactory({
-        saveActivity: saveActivityFactory({ db: projectDb }),
+        saveActivity: saveActivityFactory({ db: mainDb }),
         publish
       })
     })

packages/server/modules/cli/commands/download/project.ts

@@ -47,7 +47,7 @@ import {
   getViewerResourcesForCommentsFactory,
   getViewerResourcesFromLegacyIdentifiersFactory
 } from '@/modules/core/services/commit/viewerResources'
-import { db } from '@/db/knex'
+import { db, mainDb } from '@/db/knex'
 import {
   getCommentFactory,
   getCommentsResourcesFactory,
@@ -175,7 +175,7 @@ const command: CommandModule<
       addCommentCreatedActivity: addCommentCreatedActivityFactory({
         getViewerResourcesFromLegacyIdentifiers,
         getViewerResourceItemsUngrouped,
-        saveActivity: saveActivityFactory({ db: projectDb }),
+        saveActivity: saveActivityFactory({ db: mainDb }),
         publish
       })
     })
@@ -191,7 +191,7 @@ const command: CommandModule<
           getCommentsResources: getCommentsResourcesFactory({ db: projectDb }),
           getViewerResourcesFromLegacyIdentifiers
         }),
-        saveActivity: saveActivityFactory({ db: projectDb }),
+        saveActivity: saveActivityFactory({ db: mainDb }),
         publish
       })
     })
@@ -206,7 +206,7 @@ const command: CommandModule<
       markCommitBranchUpdated: markCommitBranchUpdatedFactory({ db: projectDb }),
       versionsEventEmitter: VersionsEmitter.emit,
       addCommitCreatedActivity: addCommitCreatedActivityFactory({
-        saveActivity: saveActivityFactory({ db: projectDb }),
+        saveActivity: saveActivityFactory({ db: mainDb }),
         publish
       })
     })
@@ -250,7 +250,7 @@ const command: CommandModule<
         getStreamBranchByName,
         createBranch: createBranchFactory({ db: projectDb }),
         addBranchCreatedActivity: addBranchCreatedActivityFactory({
-          saveActivity: saveActivityFactory({ db: projectDb }),
+          saveActivity: saveActivityFactory({ db: mainDb }),
           publish
         })
       })

packages/server/modules/cli/commands/stream/clone.ts

@@ -51,7 +51,8 @@ const command: CommandModule<
     const cloneStream = cloneStreamFactory({
       getStream: getStreamFactory({ db }),
       getUser,
-      db,
+      newProjectDb: db,
+      sourceProjectDb: db,
       createStream: createStreamFactory({ db }),
       insertCommits: insertCommitsFactory({ db }),
       getBatchedStreamCommits: getBatchedStreamCommitsFactory({ db }),

packages/server/modules/cli/commands/test/generate-key-pair.ts

@@ -3,7 +3,7 @@ import _sodium from 'libsodium-wrappers'
 import { CommandModule } from 'yargs'
 
 const command: CommandModule = {
-  command: 'generateKeyPair',
+  command: 'generate-key-pair',
   describe: 'Generate a public private key pair for lisodium box encryption',
 
   handler: async () => {

packages/server/modules/cli/index.ts

@@ -2,30 +2,50 @@
 import path from 'path'
 import yargs from 'yargs'
 import '../../bootstrap'
-import { logger } from '@/logging/logging'
+import { cliLogger, logger } from '@/logging/logging'
+import { isTestEnv } from '@/modules/shared/helpers/envHelper'
+import { mochaHooks } from '@/test/hooks'
 
-const execution = yargs
-  .scriptName('yarn cli')
-  .usage('$0 <cmd> [args]')
-  .commandDir(path.resolve(__dirname, './commands'), { extensions: ['js', 'ts'] })
-  .demandCommand()
-  .fail((msg, err, yargs) => {
-    if (!err) {
-      // If validation error (no err instance) then just show help and show the message
-      console.log(yargs.help())
-      console.log('\n', msg)
-    } else {
-      // If actual app error occurred, show the msg, but don't show help info
-      logger.error(err)
-      console.log('\n', 'Specify --help for available options')
-    }
+const main = async () => {
+  const execution = yargs
+    .scriptName('yarn cli')
+    .usage('$0 <cmd> [args]')
+    .commandDir(path.resolve(__dirname, './commands'), { extensions: ['js', 'ts'] })
+    .option('beforeAll', {
+      type: 'boolean',
+      default: false,
+      describe: 'Run beforeAll hooks before running migrations, if in test mode'
+    })
+    .demandCommand()
+    .middleware(async (argv) => {
+      // If beforeAll set, run beforeAll
+      const isBeforeAllSet = !!argv.beforeAll
 
-    process.exit(1)
-  })
-  .help().argv
+      // In test env, run beforeAll hooks to properly initialize everything first
+      if (isBeforeAllSet && isTestEnv()) {
+        cliLogger.info('Running test beforeAll hooks...')
+        await (mochaHooks.beforeAll as () => Promise<void>)()
+      }
+    })
+    .fail((msg, err, yargs) => {
+      if (!err) {
+        // If validation error (no err instance) then just show help and show the message
+        console.log(yargs.help())
+        console.log('\n', msg)
+      } else {
+        // If actual app error occurred, show the msg, but don't show help info
+        logger.error(err)
+        console.log('\n', 'Specify --help for available options')
+      }
 
-const promise = Promise.resolve(execution)
-promise.then(() => {
+      process.exit(1)
+    })
+    .help().argv
+
+  return execution
+}
+
+main().then(() => {
   // weird TS typing issue
   yargs.exit(0, undefined as unknown as Error)
 })

packages/server/modules/cli/readme.md

@@ -1,10 +1,14 @@
 # Using CLI
 
-You can run it like so from the `server` package's root directory: `./bin/cli`
+You can run it like so from the `server` package's root directory: `./bin/cli` (or `yarn cli`)
 
 Use the `--help` argument to get more info about each command.
 
-Example for running migrations: `./bin/cli db migrate latest`
+Example for running migrations: `yarn cli db migrate latest`
+
+## Using CLI in test mode (& DB)
+
+Use `yarn cli:test` to run the CLI in the TEST environment. This will use the test DB and will likely run some code a bit differently than in prod/dev.
 
 # Creating new commands
 

packages/server/modules/core/graph/resolvers/projects.ts

@@ -156,7 +156,8 @@ const updateStream = updateStreamFactory({ db })
 const cloneStream = cloneStreamFactory({
   getStream: getStreamFactory({ db }),
   getUser,
-  db,
+  newProjectDb: db,
+  sourceProjectDb: db,
   createStream: createStreamFactory({ db }),
   insertCommits: insertCommitsFactory({ db }),
   getBatchedStreamCommits: getBatchedStreamCommitsFactory({ db }),
@@ -175,6 +176,7 @@ const cloneStream = cloneStreamFactory({
   })
 })
 
+// We want to read & write from main DB - this isn't occuring in a multi region workspace ctx
 const createOnboardingStream = createOnboardingStreamFactory({
   getOnboardingBaseProject: getOnboardingBaseProjectFactory({
     getOnboardingBaseStream: getOnboardingBaseStreamFactory({ db })

packages/server/modules/core/services/streams/clone.ts

@@ -47,6 +47,9 @@ import { GetUser } from '@/modules/core/domain/users/operations'
 type CloneStreamInitialState = {
   user: UserWithOptionalRole<UserRecord>
   targetStream: StreamWithOptionalRole
+  /**
+   * Target streeam DB TRX for ensuring everything gets properly inserted
+   */
   trx: Knex.Transaction
 }
 
@@ -81,7 +84,7 @@ const decrementingDateGenerator = () => {
 type PrepareStateDeps = {
   getStream: GetStream
   getUser: GetUser
-  db: Knex
+  newProjectDb: Knex
 }
 
 const prepareStateFactory =
@@ -93,13 +96,18 @@ const prepareStateFactory =
         info: { sourceStreamId }
       })
     }
+    if (targetStream.regionKey) {
+      throw new StreamCloneError(
+        'Cloning of multiregion streams is not currently supported'
+      )
+    }
 
     const user = await deps.getUser(userId)
     if (!user) {
       throw new StreamCloneError('Clone target user not found')
     }
 
-    const trx = await deps.db.transaction()
+    const trx = await deps.newProjectDb.transaction()
 
     return { user, targetStream, trx }
   }
@@ -153,14 +161,16 @@ const cloneStreamObjectsOldFactory =
   }
 
 type CloneStreamObjectsDeps = {
-  db: Knex
+  newProjectDb: Knex
+  sourceProjectDb: Knex
 }
 
 // For sample onboarding stream, goes from 25s to ~250ms vs `cloneStreamObjectsOld`
+// TODO: This kind of query is not supported in multiregion, we can use the old one but apparently its 10 times slower...
 const cloneStreamObjectsFactory =
   (deps: CloneStreamObjectsDeps) =>
   async (state: CloneStreamInitialState, newStreamId: string) => {
-    const query = deps.db
+    const query = deps.sourceProjectDb // same as targetProjectDb for now
       .raw(
         `
         INSERT INTO objects ("id", "speckleType", "totalChildrenCount", "totalChildrenCountByDepth", "createdAt", "data", "streamId")
@@ -490,6 +500,8 @@ const cloneStreamCommentsFactory =
  * Create a new stream that is cloned from another one for the target user.
  * Important note: There are no access checks here, even private streams can be cloned! Do any
  * access control checking before you invoke this function, if needed.
+ *
+ * TODO: Does not currently support multiregion projects because of `cloneStreamObjectsFactory`
  * @returns The ID of the new stream
  */
 export const cloneStreamFactory =

packages/server/modules/cross-server-sync/index.ts

@@ -85,6 +85,7 @@ const crossServerSyncModule: SpeckleModule = {
   finalize() {
     crossServerSyncLogger.info('⬇️  Ensuring base onboarding stream asynchronously...')
 
+    // Its fine to use main DB here, none of this is executed in a workspace context
     const getUser = getUserFactory({ db })
     const markOnboardingBaseStream = markOnboardingBaseStreamFactory({ db })
     const markCommitStreamUpdated = markCommitStreamUpdatedFactory({ db })

packages/server/modules/multiregion/dbSelector.ts

@@ -133,6 +133,25 @@ export const getRegisteredRegionClients = async (): Promise<RegionClients> => {
 export const getRegisteredDbClients = async (): Promise<Knex[]> =>
   Object.values(await getRegisteredRegionClients())
 
+export const getAllRegisteredDbClients = async (): Promise<
+  Array<{ client: Knex; isMain: boolean; regionKey: string }>
+> => {
+  const mainDb = db
+  const regionDbs = await getRegisteredRegionClients()
+  return [
+    {
+      client: mainDb,
+      isMain: true,
+      regionKey: 'main'
+    },
+    ...Object.entries(regionDbs).map(([regionKey, client]) => ({
+      client,
+      isMain: false,
+      regionKey
+    }))
+  ]
+}
+
 /**
  * Idempotently initialize region
  */

packages/server/package.json

@@ -32,6 +32,7 @@
     "lint:eslint": "eslint .",
     "cli": "cross-env LOG_LEVEL=debug LOG_PRETTY=true NODE_ENV=development ts-node ./modules/cli/index.ts",
     "cli:test": "cross-env LOG_LEVEL=debug LOG_PRETTY=true NODE_ENV=test ts-node ./modules/cli/index.ts",
+    "cli:test:multiregion": "cross-env RUN_TESTS_IN_MULTIREGION_MODE=true yarn cli:test",
     "cli:download:commit": "cross-env LOG_PRETTY=true LOG_LEVEL=debug yarn cli download commit",
     "migrate": "yarn cli db migrate",
     "migrate:test": "cross-env NODE_ENV=test ts-node ./modules/cli/index.js db migrate",

packages/server/test/speckle-helpers/commitHelper.ts

@@ -1,4 +1,4 @@
-import { db } from '@/db/knex'
+import { mainDb } from '@/db/knex'
 import { saveActivityFactory } from '@/modules/activitystream/repositories'
 import { addCommitCreatedActivityFactory } from '@/modules/activitystream/services/commitActivity'
 import { VersionsEmitter } from '@/modules/core/events/versionsEmitter'
@@ -26,33 +26,7 @@ import { createObjectFactory } from '@/modules/core/services/objects/management'
 import { publish } from '@/modules/shared/utils/subscriptions'
 import { BasicTestUser } from '@/test/authHelper'
 import { BasicTestStream } from '@/test/speckle-helpers/streamHelper'
-
-const createObject = createObjectFactory({
-  storeSingleObjectIfNotFoundFactory: storeSingleObjectIfNotFoundFactory({ db }),
-  storeClosuresIfNotFound: storeClosuresIfNotFoundFactory({ db })
-})
-const markCommitStreamUpdated = markCommitStreamUpdatedFactory({ db })
-const getObject = getObjectFactory({ db })
-const createCommitByBranchId = createCommitByBranchIdFactory({
-  createCommit: createCommitFactory({ db }),
-  getObject,
-  getBranchById: getBranchByIdFactory({ db }),
-  insertStreamCommits: insertStreamCommitsFactory({ db }),
-  insertBranchCommits: insertBranchCommitsFactory({ db }),
-  markCommitStreamUpdated,
-  markCommitBranchUpdated: markCommitBranchUpdatedFactory({ db }),
-  versionsEventEmitter: VersionsEmitter.emit,
-  addCommitCreatedActivity: addCommitCreatedActivityFactory({
-    saveActivity: saveActivityFactory({ db }),
-    publish
-  })
-})
-
-const createCommitByBranchName = createCommitByBranchNameFactory({
-  createCommitByBranchId,
-  getStreamBranchByName: getStreamBranchByNameFactory({ db }),
-  getBranchById: getBranchByIdFactory({ db })
-})
+import { Knex } from 'knex'
 
 export type BasicTestCommit = {
   /**
@@ -87,57 +61,96 @@ export type BasicTestCommit = {
 }
 
 export async function createTestObject(params: { projectId: string }) {
+  const db = mainDb
+  const createObject = createObjectFactory({
+    storeSingleObjectIfNotFoundFactory: storeSingleObjectIfNotFoundFactory({ db }),
+    storeClosuresIfNotFound: storeClosuresIfNotFoundFactory({ db })
+  })
+
   return await createObject({
     streamId: params.projectId,
     object: { foo: 'bar' }
   })
 }
 
-/**
- * Ensure all commits have objectId set
- */
-async function ensureObjects(commits: BasicTestCommit[]) {
-  const commitsWithoutObjects = commits.filter((c) => !c.objectId)
-  await Promise.all(
-    commitsWithoutObjects.map((c) =>
-      createObject({
-        streamId: c.streamId,
-        object: { foo: 'bar' }
-      }).then((oid) => (c.objectId = oid))
+const ensureObjectsFactory =
+  (deps: { db: Knex }) => async (commits: BasicTestCommit[]) => {
+    const { db } = deps
+    const createObject = createObjectFactory({
+      storeSingleObjectIfNotFoundFactory: storeSingleObjectIfNotFoundFactory({ db }),
+      storeClosuresIfNotFound: storeClosuresIfNotFoundFactory({ db })
+    })
+
+    const commitsWithoutObjects = commits.filter((c) => !c.objectId)
+    await Promise.all(
+      commitsWithoutObjects.map((c) =>
+        createObject({
+          streamId: c.streamId,
+          object: { foo: 'bar' }
+        }).then((oid) => (c.objectId = oid))
+      )
     )
-  )
-}
+  }
 
 /**
  * Create test commits
  */
-export async function createTestCommits(
-  commits: BasicTestCommit[],
-  options?: Partial<{ owner: BasicTestUser; stream: BasicTestStream }>
-) {
-  const { owner, stream } = options || {}
+export const createTestCommitsFactory =
+  (deps: { db: Knex }) =>
+  async (
+    commits: BasicTestCommit[],
+    options?: Partial<{ owner: BasicTestUser; stream: BasicTestStream }>
+  ) => {
+    const { db } = deps
+    const { owner, stream } = options || {}
 
-  commits.forEach((c) => {
-    if (owner) c.authorId = owner.id
-    if (stream) c.streamId = stream.id
-  })
+    const createCommitByBranchId = createCommitByBranchIdFactory({
+      createCommit: createCommitFactory({ db }),
+      getObject: getObjectFactory({ db }),
+      getBranchById: getBranchByIdFactory({ db }),
+      insertStreamCommits: insertStreamCommitsFactory({ db }),
+      insertBranchCommits: insertBranchCommitsFactory({ db }),
+      markCommitStreamUpdated: markCommitStreamUpdatedFactory({ db }),
+      markCommitBranchUpdated: markCommitBranchUpdatedFactory({ db }),
+      versionsEventEmitter: VersionsEmitter.emit,
+      addCommitCreatedActivity: addCommitCreatedActivityFactory({
+        saveActivity: saveActivityFactory({ db: mainDb }),
+        publish
+      })
+    })
 
-  await ensureObjects(commits)
-  await Promise.all(
-    commits.map((c) =>
-      createCommitByBranchName({
-        streamId: c.streamId,
-        branchName: c.branchName || 'main',
-        message: c.message || 'this message is auto generated',
-        sourceApplication: 'tests',
-        objectId: c.objectId,
-        authorId: c.authorId,
-        totalChildrenCount: 0,
-        parents: c.parents || []
-      }).then((newCommit) => (c.id = newCommit.id))
+    const createCommitByBranchName = createCommitByBranchNameFactory({
+      createCommitByBranchId,
+      getStreamBranchByName: getStreamBranchByNameFactory({ db }),
+      getBranchById: getBranchByIdFactory({ db })
+    })
+
+    commits.forEach((c) => {
+      if (owner) c.authorId = owner.id
+      if (stream) c.streamId = stream.id
+    })
+
+    await ensureObjectsFactory(deps)(commits)
+    await Promise.all(
+      commits.map((c) =>
+        createCommitByBranchName({
+          streamId: c.streamId,
+          branchName: c.branchName || 'main',
+          message: c.message || 'this message is auto generated',
+          sourceApplication: 'tests',
+          objectId: c.objectId,
+          authorId: c.authorId,
+          totalChildrenCount: 0,
+          parents: c.parents || []
+        }).then((newCommit) => (c.id = newCommit.id))
+      )
     )
-  )
-}
+  }
+
+/**
+ * Create test commits
+ */
+export const createTestCommits = createTestCommitsFactory({ db: mainDb })
 
 export async function createTestCommit(
   commit: BasicTestCommit,