From a9a313ee63cfae6a23c644d255bd64ffcb0ed236 Mon Sep 17 00:00:00 2001 From: Kristaps Fabians Geikins Date: Fri, 22 Nov 2024 18:52:58 +0000 Subject: [PATCH] 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 | 2 +- .../modules/cli/commands/db/helpers/index.ts | 19 +++ .../server/modules/cli/commands/db/migrate.ts | 9 +- .../modules/cli/commands/db/migrate/create.ts | 15 +- .../modules/cli/commands/db/migrate/down.ts | 25 +-- .../modules/cli/commands/db/migrate/latest.ts | 26 ++-- .../cli/commands/db/migrate/rollback.ts | 26 ++-- .../modules/cli/commands/db/migrate/up.ts | 16 +- .../modules/cli/commands/db/seed/commits.ts | 8 +- .../modules/cli/commands/download/commit.ts | 8 +- .../modules/cli/commands/download/project.ts | 10 +- .../modules/cli/commands/stream/clone.ts | 3 +- .../generate-key-pair.ts} | 2 +- packages/server/modules/cli/index.ts | 62 +++++--- packages/server/modules/cli/readme.md | 8 +- .../modules/core/graph/resolvers/projects.ts | 4 +- .../modules/core/services/streams/clone.ts | 20 ++- .../server/modules/cross-server-sync/index.ts | 1 + .../server/modules/multiregion/dbSelector.ts | 19 +++ packages/server/package.json | 1 + .../test/speckle-helpers/commitHelper.ts | 143 ++++++++++-------- 21 files changed, 269 insertions(+), 158 deletions(-) create mode 100644 packages/server/modules/cli/commands/db/helpers/index.ts rename packages/server/modules/cli/commands/{encryption.ts => test/generate-key-pair.ts} (94%) diff --git a/packages/server/db/knex.ts b/packages/server/db/knex.ts index f5bcc0cf1..4c651f584 100644 --- a/packages/server/db/knex.ts +++ b/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) diff --git a/packages/server/modules/cli/commands/db/helpers/index.ts b/packages/server/modules/cli/commands/db/helpers/index.ts new file mode 100644 index 000000000..b16ef7f2f --- /dev/null +++ b/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 +} diff --git a/packages/server/modules/cli/commands/db/migrate.ts b/packages/server/modules/cli/commands/db/migrate.ts index a9577b838..6d588d1ef 100644 --- a/packages/server/modules/cli/commands/db/migrate.ts +++ b/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 } diff --git a/packages/server/modules/cli/commands/db/migrate/create.ts b/packages/server/modules/cli/commands/db/migrate/create.ts index 803414108..a4d75e871 100644 --- a/packages/server/modules/cli/commands/db/migrate/create.ts +++ b/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 = { @@ -25,10 +26,16 @@ const command: CommandModule = { 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...') diff --git a/packages/server/modules/cli/commands/db/migrate/down.ts b/packages/server/modules/cli/commands/db/migrate/down.ts index c79c7256c..c7ba1aef4 100644 --- a/packages/server/modules/cli/commands/db/migrate/down.ts +++ b/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 = { +const command: CommandModule = { 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 = { : `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!') diff --git a/packages/server/modules/cli/commands/db/migrate/latest.ts b/packages/server/modules/cli/commands/db/migrate/latest.ts index 2686fd53c..0fec082a6 100644 --- a/packages/server/modules/cli/commands/db/migrate/latest.ts +++ b/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 = { 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)() - } 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') diff --git a/packages/server/modules/cli/commands/db/migrate/rollback.ts b/packages/server/modules/cli/commands/db/migrate/rollback.ts index 4d86c0ddc..4e1c49edf 100644 --- a/packages/server/modules/cli/commands/db/migrate/rollback.ts +++ b/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 = { 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)() - } - - 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') diff --git a/packages/server/modules/cli/commands/db/migrate/up.ts b/packages/server/modules/cli/commands/db/migrate/up.ts index d06d4c796..0e33db39f 100644 --- a/packages/server/modules/cli/commands/db/migrate/up.ts +++ b/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 = { 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') } } diff --git a/packages/server/modules/cli/commands/db/seed/commits.ts b/packages/server/modules/cli/commands/db/seed/commits.ts index 37733566a..54002a24e 100644 --- a/packages/server/modules/cli/commands/db/seed/commits.ts +++ b/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) { diff --git a/packages/server/modules/cli/commands/download/commit.ts b/packages/server/modules/cli/commands/download/commit.ts index d1156898f..e1507dff1 100644 --- a/packages/server/modules/cli/commands/download/commit.ts +++ b/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 }) }) diff --git a/packages/server/modules/cli/commands/download/project.ts b/packages/server/modules/cli/commands/download/project.ts index 7321f9b21..af3191ff2 100644 --- a/packages/server/modules/cli/commands/download/project.ts +++ b/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 }) }) diff --git a/packages/server/modules/cli/commands/stream/clone.ts b/packages/server/modules/cli/commands/stream/clone.ts index 44d66e2fc..7dbedfceb 100644 --- a/packages/server/modules/cli/commands/stream/clone.ts +++ b/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 }), diff --git a/packages/server/modules/cli/commands/encryption.ts b/packages/server/modules/cli/commands/test/generate-key-pair.ts similarity index 94% rename from packages/server/modules/cli/commands/encryption.ts rename to packages/server/modules/cli/commands/test/generate-key-pair.ts index d406ff20b..e5ede4948 100644 --- a/packages/server/modules/cli/commands/encryption.ts +++ b/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 () => { diff --git a/packages/server/modules/cli/index.ts b/packages/server/modules/cli/index.ts index 9a3324f82..7f6d64620 100644 --- a/packages/server/modules/cli/index.ts +++ b/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 [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 [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)() + } + }) + .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) }) diff --git a/packages/server/modules/cli/readme.md b/packages/server/modules/cli/readme.md index 7756d880d..31a1a298a 100644 --- a/packages/server/modules/cli/readme.md +++ b/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 diff --git a/packages/server/modules/core/graph/resolvers/projects.ts b/packages/server/modules/core/graph/resolvers/projects.ts index f48a12f6b..7fef240db 100644 --- a/packages/server/modules/core/graph/resolvers/projects.ts +++ b/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 }) diff --git a/packages/server/modules/core/services/streams/clone.ts b/packages/server/modules/core/services/streams/clone.ts index 03027b2d3..5b5dc9b50 100644 --- a/packages/server/modules/core/services/streams/clone.ts +++ b/packages/server/modules/core/services/streams/clone.ts @@ -47,6 +47,9 @@ import { GetUser } from '@/modules/core/domain/users/operations' type CloneStreamInitialState = { user: UserWithOptionalRole 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 = diff --git a/packages/server/modules/cross-server-sync/index.ts b/packages/server/modules/cross-server-sync/index.ts index f3078b503..f524eadaf 100644 --- a/packages/server/modules/cross-server-sync/index.ts +++ b/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 }) diff --git a/packages/server/modules/multiregion/dbSelector.ts b/packages/server/modules/multiregion/dbSelector.ts index 1bad4eebe..990d3e53a 100644 --- a/packages/server/modules/multiregion/dbSelector.ts +++ b/packages/server/modules/multiregion/dbSelector.ts @@ -133,6 +133,25 @@ export const getRegisteredRegionClients = async (): Promise => { export const getRegisteredDbClients = async (): Promise => 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 */ diff --git a/packages/server/package.json b/packages/server/package.json index ba5fa4697..6a3f0d5f1 100644 --- a/packages/server/package.json +++ b/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", diff --git a/packages/server/test/speckle-helpers/commitHelper.ts b/packages/server/test/speckle-helpers/commitHelper.ts index 601270dd4..65346a070 100644 --- a/packages/server/test/speckle-helpers/commitHelper.ts +++ b/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,