feat(multiregion): apply prepared transactions to projects (#5322)

* feat(multiregion): replace user replication

* chore(multiregion): optimise replication

* maybe it's this

* postgres is fun

* once more

* chore(multiregion): only replicate test user creation during multiregion tests

* feat: improved replicate_query logic

* fix: minor

* fix: starting issue

* feat: included user create and delete specs to multiregion

* feat: removed console logs

* fix: user defaults

* fix: multiregion test helper

* fix: update scenarios for users

* refactor(multiregion): swap replicateQuery concept to asMultiregionOperation (#5301)

feat(multiregion): introduced asMultregionOperator, refactor test to user builder classes

* chore: renamings

* fix: remove comments

* feat: remove user replication

* refactor: simplified spec usages

* chore: comments

* chore: branches and favs

* chore: more tests

* chore: more tests

* fix linting

* fix tests

* feat: dropping replication

* refactor: moved project delete to service

* fix: comment

* feat: updateStreamFactory and updateProjectFacotry

* deleteProjectFactory + replicateFactory

* deleteWorkspaceFactory

* fix: selector

* fix: tests

* fix tests, finished createStreamFactory

* feat: simplify changes

* fix: remove comment

* fix: minor strucutres

* fix: moveProjectToRegion

* fix: moved branch creation outside of multiregion scope

* fix: branch creation

* fix: tests

* fix: ci tests

* fix: removed log form test

* fix: on specs, no random regionKeys

* review fixes

* fix: mr comments

* feat: removed test

---------

Co-authored-by: Charles Driesler <chuck@speckle.systems>

packages/server/modules/workspaces/domain/operations.ts

@@ -488,10 +488,10 @@ export type DenyWorkspaceJoinRequest = (
 /**
  * Updates project region and moves all regional data to target regional db
  */
-export type UpdateProjectRegion = (params: {
+export type MoveProjectToRegion = (params: {
   projectId: string
   regionKey: string
-}) => Promise<Stream>
+}) => Promise<void>
 
 /**
  * Given a count of objects successfully copied to another region, confirm that these counts

packages/server/modules/workspaces/graph/resolvers/workspaces.ts

@@ -5,7 +5,6 @@ import { removePrivateFields } from '@/modules/core/helpers/userHelper'
 import {
   updateProjectFactory,
   getStreamFactory,
-  deleteStreamFactory,
   revokeStreamPermissionsFactory,
   grantStreamPermissionsFactory,
   getStreamCollaboratorsFactory,
@@ -151,9 +150,18 @@ import {
 import { updateStreamRoleAndNotifyFactory } from '@/modules/core/services/streams/management'
 import { getUserFactory, getUsersFactory } from '@/modules/core/repositories/users'
 import { getServerInfoFactory } from '@/modules/core/repositories/server'
-import { asOperation, commandFactory } from '@/modules/shared/command'
+import {
+  asMultiregionalOperation,
+  asOperation,
+  commandFactory,
+  replicateFactory
+} from '@/modules/shared/command'
 import { throwIfRateLimitedFactory } from '@/modules/core/utils/ratelimiter'
-import { getProjectDbClient, getRegionDb } from '@/modules/multiregion/utils/dbSelector'
+import {
+  getAllRegisteredDbs,
+  getProjectDbClient,
+  getProjectReplicationDbs
+} from '@/modules/multiregion/utils/dbSelector'
 import {
   listUserExpiredSsoSessionsFactory,
   listWorkspaceSsoMembershipsByUserEmailFactory
@@ -180,7 +188,6 @@ import {
   getWorkspaceWithPlanFactory,
   upsertWorkspacePlanFactory
 } from '@/modules/gatekeeper/repositories/billing'
-import type { Knex } from 'knex'
 import { getPaginatedItemsFactory } from '@/modules/shared/services/paginatedItems'
 import { BadRequestError, UnauthorizedError } from '@/modules/shared/errors'
 import {
@@ -193,6 +200,7 @@ import {
 } from '@/modules/workspaces/repositories/workspaceJoinRequests'
 import { sendWorkspaceJoinRequestReceivedEmailFactory } from '@/modules/workspaces/services/workspaceJoinRequestEmails/received'
 import {
+  deleteProjectFactory,
   getProjectFactory,
   getUserProjectRolesFactory
 } from '@/modules/core/repositories/projects'
@@ -229,10 +237,13 @@ import {
 } from '@/modules/serverinvites/services/coreFinalization'
 import { WorkspaceInvitesLimit } from '@/modules/workspaces/domain/constants'
 import { copyWorkspaceFactory } from '@/modules/workspaces/repositories/projectRegions'
-import { queryAllProjectsFactory } from '@/modules/core/services/projects'
+import {
+  deleteProjectAndCommitsFactory,
+  queryAllProjectsFactory
+} from '@/modules/core/services/projects'
 import { WorkspacePlanNotFoundError } from '@/modules/gatekeeper/errors/billing'
+import { deleteProjectCommitsFactory } from '@/modules/core/repositories/commits'
 
-const eventBus = getEventBus()
 const getServerInfo = getServerInfoFactory({ db })
 const getUser = getUserFactory({ db })
 const getUsers = getUsersFactory({ db })
@@ -805,38 +816,36 @@ export default FF_WORKSPACES_MODULE_ENABLED
             }
           }
 
-          const deleteWorkspaceFrom = (db: Knex) =>
-            deleteWorkspaceFactory({
-              deleteWorkspace: repoDeleteWorkspaceFactory({ db }),
-              deleteProject: deleteStreamFactory({ db }),
-              deleteAllResourceInvites: deleteAllResourceInvitesFactory({ db }),
-              queryAllProjects: queryAllProjectsFactory({
-                getExplicitProjects: getExplicitProjects({ db })
-              }),
-              deleteSsoProvider: deleteSsoProviderFactory({ db }),
-              emitWorkspaceEvent: getEventBus().emit
-            })
+          // this is a bit of an overhead, we are issuing delete queries to all regions,
+          // instead of being selective and clever about figuring out the project DB and only
+          // deleting from main and the project db
+          // while workspace must be deleted from all regions
 
-          // this should be turned into a get all regions and map over the regions...
-          const region = await getDefaultRegionFactory({ db })({ workspaceId })
-          if (region) {
-            const regionDb = await getRegionDb({ regionKey: region.key })
-            await withOperationLogging(
-              async () => await deleteWorkspaceFrom(regionDb)({ workspaceId }),
-              {
-                logger: logger.child({ regionKey: region.key }),
-                operationName: 'deleteWorkspaceFromRegion',
-                operationDescription: 'Delete workspace from region'
-              }
-            )
-          }
-
-          await withOperationLogging(
-            async () => await deleteWorkspaceFrom(db)({ workspaceId }),
+          await asMultiregionalOperation(
+            async ({ mainDb, allDbs, emit }) =>
+              deleteWorkspaceFactory({
+                deleteWorkspace: replicateFactory(allDbs, repoDeleteWorkspaceFactory),
+                deleteProjectAndCommits: deleteProjectAndCommitsFactory({
+                  deleteProject: replicateFactory(allDbs, deleteProjectFactory),
+                  deleteProjectCommits: replicateFactory(
+                    allDbs,
+                    deleteProjectCommitsFactory
+                  )
+                }),
+                deleteAllResourceInvites: deleteAllResourceInvitesFactory({
+                  db: mainDb
+                }),
+                queryAllProjects: queryAllProjectsFactory({
+                  getExplicitProjects: getExplicitProjects({ db: mainDb })
+                }),
+                deleteSsoProvider: deleteSsoProviderFactory({ db: mainDb }),
+                emitWorkspaceEvent: emit
+              })({ workspaceId }),
             {
               logger,
-              operationName: 'deleteWorkspace',
-              operationDescription: 'Delete workspace'
+              name: 'delete workspace',
+              description: 'Delete workspace',
+              dbs: await getAllRegisteredDbs()
             }
           )
 
@@ -1606,63 +1615,76 @@ export default FF_WORKSPACES_MODULE_ENABLED
             throw mapAuthToServerError(canMoveToWorkspace.error)
           }
 
-          const moveProjectToWorkspace = commandFactory({
-            db,
-            eventBus,
-            operationFactory: ({ db, emit }) =>
+          const updatedProject = await asMultiregionalOperation(
+            ({ mainDb, allDbs, emit }) =>
               moveProjectToWorkspaceFactory({
-                getProject: getProjectFactory({ db }),
-                updateProject: updateProjectFactory({ db: projectDb }),
-                updateProjectRole: updateStreamRoleAndNotify,
-                getProjectCollaborators: getStreamCollaboratorsFactory({ db }),
+                getProject: getProjectFactory({ db: mainDb }),
+                updateProject: replicateFactory(allDbs, updateProjectFactory),
+                updateProjectRole: updateStreamRoleAndNotifyFactory({
+                  isStreamCollaborator: isStreamCollaboratorFactory({
+                    getStream: getStreamFactory({ db: mainDb })
+                  }),
+                  addOrUpdateStreamCollaborator: addOrUpdateStreamCollaboratorFactory({
+                    validateStreamAccess,
+                    getUser: getUserFactory({ db: mainDb }),
+                    grantStreamPermissions: grantStreamPermissionsFactory({
+                      db: mainDb
+                    }),
+                    getStreamRoles: getStreamRolesFactory({ db: mainDb }),
+                    emitEvent: emit
+                  }),
+                  removeStreamCollaborator
+                }),
+                getProjectCollaborators: getStreamCollaboratorsFactory({ db: mainDb }),
                 copyWorkspace: copyWorkspaceFactory({
+                  // TODO: must be removed when workspace replication is implemented
                   sourceDb: db,
                   targetDb: projectDb
                 }),
-                getWorkspaceRolesAndSeats: getWorkspaceRolesAndSeatsFactory({ db }),
+                getWorkspaceRolesAndSeats: getWorkspaceRolesAndSeatsFactory({
+                  db: mainDb
+                }),
                 updateWorkspaceRole: addOrUpdateWorkspaceRoleFactory({
-                  getWorkspaceRoles: getWorkspaceRolesFactory({ db }),
-                  getWorkspaceWithDomains: getWorkspaceWithDomainsFactory({ db }),
-                  findVerifiedEmailsByUserId: findVerifiedEmailsByUserIdFactory({
-                    db
+                  getWorkspaceRoles: getWorkspaceRolesFactory({ db: mainDb }),
+                  getWorkspaceWithDomains: getWorkspaceWithDomainsFactory({
+                    db: mainDb
                   }),
-                  upsertWorkspaceRole: upsertWorkspaceRoleFactory({ db }),
+                  findVerifiedEmailsByUserId: findVerifiedEmailsByUserIdFactory({
+                    db: mainDb
+                  }),
+                  upsertWorkspaceRole: upsertWorkspaceRoleFactory({ db: mainDb }),
                   emitWorkspaceEvent: emit,
                   ensureValidWorkspaceRoleSeat: ensureValidWorkspaceRoleSeatFactory({
-                    createWorkspaceSeat: createWorkspaceSeatFactory({ db }),
-                    getWorkspaceUserSeat: getWorkspaceUserSeatFactory({ db }),
+                    createWorkspaceSeat: createWorkspaceSeatFactory({ db: mainDb }),
+                    getWorkspaceUserSeat: getWorkspaceUserSeatFactory({ db: mainDb }),
                     getWorkspaceDefaultSeatType: getWorkspaceDefaultSeatTypeFactory({
-                      getWorkspace: getWorkspaceFactory({ db })
+                      getWorkspace: getWorkspaceFactory({ db: mainDb })
                     }),
                     eventEmit: emit
                   }),
                   assignWorkspaceSeat: assignWorkspaceSeatFactory({
-                    createWorkspaceSeat: createWorkspaceSeatFactory({ db }),
+                    createWorkspaceSeat: createWorkspaceSeatFactory({ db: mainDb }),
                     getWorkspaceRoleForUser: getWorkspaceRoleForUserFactory({
-                      db
+                      db: mainDb
                     }),
                     eventEmit: emit,
-                    getWorkspaceUserSeat: getWorkspaceUserSeatFactory({ db })
+                    getWorkspaceUserSeat: getWorkspaceUserSeatFactory({ db: mainDb })
                   })
                 }),
-                createWorkspaceSeat: createWorkspaceSeatFactory({ db }),
-                getWorkspaceWithPlan: getWorkspaceWithPlanFactory({ db }),
-                getWorkspaceDomains: getWorkspaceDomainsFactory({ db }),
-                getUserEmails: findEmailsByUserIdFactory({ db })
-              })
-          })
-
-          const updatedProject = await withOperationLogging(
-            async () =>
-              await moveProjectToWorkspace({
+                createWorkspaceSeat: createWorkspaceSeatFactory({ db: mainDb }),
+                getWorkspaceWithPlan: getWorkspaceWithPlanFactory({ db: mainDb }),
+                getWorkspaceDomains: getWorkspaceDomainsFactory({ db: mainDb }),
+                getUserEmails: findEmailsByUserIdFactory({ db: mainDb })
+              })({
                 projectId,
                 workspaceId,
                 movedByUserId: context.userId!
               }),
             {
               logger,
-              operationName: 'moveProjectToWorkspace',
-              operationDescription: 'Move project to workspace'
+              name: 'moveProjectToWorkspace',
+              description: 'Move project to workspace',
+              dbs: await getProjectReplicationDbs({ projectId })
             }
           )
 

packages/server/modules/workspaces/index.ts

@@ -1,4 +1,5 @@
 import type cron from 'node-cron'
+import type { Logger } from '@/observability/logging'
 import { moduleLogger } from '@/observability/logging'
 import { getFeatureFlags } from '@/modules/shared/helpers/envHelper'
 import { registerOrUpdateScopeFactory } from '@/modules/shared/repositories/scopes'
@@ -19,18 +20,21 @@ import {
 } from '@/modules/core/repositories/scheduledTasks'
 import { getWorkspacesNonCompleteFactory } from '@/modules/workspaces/repositories/workspaces'
 import { deleteWorkspacesNonCompleteFactory } from '@/modules/workspaces/services/workspaceCreationState'
-import {
-  deleteStreamFactory,
-  getExplicitProjects
-} from '@/modules/core/repositories/streams'
+import { getExplicitProjects } from '@/modules/core/repositories/streams'
 import { deleteSsoProviderFactory } from '@/modules/workspaces/repositories/sso'
-import { getEventBus } from '@/modules/shared/services/eventBus'
 import { deleteAllResourceInvitesFactory } from '@/modules/serverinvites/repositories/serverInvites'
 import { deleteWorkspaceFactory as repoDeleteWorkspaceFactory } from '@/modules/workspaces/repositories/workspaces'
 import { deleteWorkspaceFactory } from '@/modules/workspaces/services/management'
 import { scheduleUpdateAllWorkspacesTracking } from '@/modules/workspaces/services/tracking'
 import { getClient } from '@/modules/shared/utils/mixpanel'
-import { queryAllProjectsFactory } from '@/modules/core/services/projects'
+import {
+  deleteProjectAndCommitsFactory,
+  queryAllProjectsFactory
+} from '@/modules/core/services/projects'
+import { deleteProjectFactory } from '@/modules/core/repositories/projects'
+import { deleteProjectCommitsFactory } from '@/modules/core/repositories/commits'
+import { asMultiregionalOperation, replicateFactory } from '@/modules/shared/command'
+import { getAllRegisteredDbs } from '@/modules/multiregion/utils/dbSelector'
 
 const {
   FF_WORKSPACES_MODULE_ENABLED,
@@ -56,19 +60,39 @@ const scheduleDeleteWorkspacesNonComplete = ({
 }: {
   scheduleExecution: ScheduleExecution
 }) => {
-  const deleteWorkspacesNonComplete = deleteWorkspacesNonCompleteFactory({
-    getWorkspacesNonComplete: getWorkspacesNonCompleteFactory({ db }),
-    deleteWorkspace: deleteWorkspaceFactory({
-      deleteWorkspace: repoDeleteWorkspaceFactory({ db }),
-      deleteProject: deleteStreamFactory({ db }),
-      deleteAllResourceInvites: deleteAllResourceInvitesFactory({ db }),
-      queryAllProjects: queryAllProjectsFactory({
-        getExplicitProjects: getExplicitProjects({ db })
-      }),
-      deleteSsoProvider: deleteSsoProviderFactory({ db }),
-      emitWorkspaceEvent: getEventBus().emit
-    })
-  })
+  const deleteWorkspacesNonComplete = async ({ logger }: { logger: Logger }) =>
+    asMultiregionalOperation(
+      ({ allDbs, mainDb, emit }) => {
+        const deleteWorkspacesNonComplete = deleteWorkspacesNonCompleteFactory({
+          getWorkspacesNonComplete: getWorkspacesNonCompleteFactory({ db: mainDb }),
+          deleteWorkspace: deleteWorkspaceFactory({
+            deleteWorkspace: replicateFactory(allDbs, repoDeleteWorkspaceFactory),
+            deleteProjectAndCommits: deleteProjectAndCommitsFactory({
+              deleteProject: replicateFactory(allDbs, deleteProjectFactory),
+              deleteProjectCommits: replicateFactory(
+                allDbs,
+                deleteProjectCommitsFactory
+              )
+            }),
+            deleteAllResourceInvites: deleteAllResourceInvitesFactory({
+              db: mainDb
+            }),
+            queryAllProjects: queryAllProjectsFactory({
+              getExplicitProjects: getExplicitProjects({ db: mainDb })
+            }),
+            deleteSsoProvider: deleteSsoProviderFactory({ db: mainDb }),
+            emitWorkspaceEvent: emit
+          })
+        })
+
+        return deleteWorkspacesNonComplete({ logger })
+      },
+      {
+        logger,
+        name: 'deleteWorkspacesNonComplete',
+        dbs: await getAllRegisteredDbs()
+      }
+    )
 
   const every30Mins = '*/30 * * * *'
   return scheduleExecution(

packages/server/modules/workspaces/services/management.ts

@@ -58,12 +58,14 @@ import { chunk, isEmpty, omit } from 'lodash-es'
 import { userEmailsCompliantWithWorkspaceDomains } from '@/modules/workspaces/domain/logic'
 import { workspaceRoles as workspaceRoleDefinitions } from '@/modules/workspaces/roles'
 import { blockedDomains } from '@speckle/shared'
-import type { DeleteStreamRecord } from '@/modules/core/domain/streams/operations'
 import type {
   DeleteSsoProvider,
   GetWorkspaceSsoProviderRecord
 } from '@/modules/workspaces/domain/sso/operations'
-import type { QueryAllProjects } from '@/modules/core/domain/projects/operations'
+import type {
+  DeleteProjectAndCommits,
+  QueryAllProjects
+} from '@/modules/core/domain/projects/operations'
 
 type WorkspaceCreateArgs = {
   userId: string
@@ -288,14 +290,14 @@ type WorkspaceDeleteArgs = {
 export const deleteWorkspaceFactory =
   ({
     deleteWorkspace,
-    deleteProject,
+    deleteProjectAndCommits,
     queryAllProjects,
     deleteAllResourceInvites,
     deleteSsoProvider,
     emitWorkspaceEvent
   }: {
     deleteWorkspace: DeleteWorkspace
-    deleteProject: DeleteStreamRecord
+    deleteProjectAndCommits: DeleteProjectAndCommits
     queryAllProjects: QueryAllProjects
     deleteAllResourceInvites: DeleteAllResourceInvites
     deleteSsoProvider: DeleteSsoProvider
@@ -328,7 +330,9 @@ export const deleteWorkspaceFactory =
     // Workspace delete cascades-deletes stream table rows, but some manual cleanup is required
     // We re-use `deleteStream` (and re-delete the project) to DRY this manual cleanup
     for (const projectIdsChunk of chunk(projectIds, 25)) {
-      await Promise.all(projectIdsChunk.map((projectId) => deleteProject(projectId)))
+      await Promise.all(
+        projectIdsChunk.map((projectId) => deleteProjectAndCommits({ projectId }))
+      )
     }
     await emitWorkspaceEvent({
       eventName: WorkspaceEvents.Deleted,

packages/server/modules/workspaces/services/projectRegions.ts

@@ -1,5 +1,4 @@
 import type { GetProject } from '@/modules/core/domain/projects/operations'
-import type { UpdateProjectRegionKey } from '@/modules/multiregion/services/projectRegion'
 import type {
   CopyProjectAutomations,
   CopyProjectBlobs,
@@ -17,13 +16,13 @@ import type {
   CountProjectVersions,
   CountProjectWebhooks,
   GetAvailableRegions,
-  UpdateProjectRegion,
+  MoveProjectToRegion,
   ValidateProjectRegionCopy
 } from '@/modules/workspaces/domain/operations'
 import { ProjectRegionAssignmentError } from '@/modules/workspaces/errors/regions'
 import { logger } from '@/observability/logging'
 
-export const updateProjectRegionFactory =
+export const moveProjectToRegionFactory =
   (deps: {
     getProject: GetProject
     getAvailableRegions: GetAvailableRegions
@@ -37,8 +36,7 @@ export const updateProjectRegionFactory =
     copyProjectWebhooks: CopyProjectWebhooks
     copyProjectBlobs: CopyProjectBlobs
     validateProjectRegionCopy: ValidateProjectRegionCopy
-    updateProjectRegionKey: UpdateProjectRegionKey
-  }): UpdateProjectRegion =>
+  }): MoveProjectToRegion =>
   async (params) => {
     const { projectId, regionKey } = params
 
@@ -120,9 +118,6 @@ export const updateProjectRegionFactory =
         'Missing data from source project in target region copy after move.'
       )
     }
-
-    // Update project region in db and update relevant caches
-    return await deps.updateProjectRegionKey({ projectId, regionKey })
   }
 
 export const validateProjectRegionCopyFactory =

packages/server/modules/workspaces/services/projects.ts

@@ -30,21 +30,15 @@ import { ProjectNotFoundError } from '@/modules/core/errors/projects'
 import type { WorkspaceProjectCreateInput } from '@/modules/core/graph/generated/graphql'
 import {
   getDb,
+  getReplicationDbs,
   getValidDefaultProjectRegionKey
 } from '@/modules/multiregion/utils/dbSelector'
+import { createNewProjectFactory } from '@/modules/core/services/projects'
 import {
-  createNewProjectFactory,
-  waitForRegionProjectFactory
-} from '@/modules/core/services/projects'
-import {
-  deleteProjectFactory,
-  getProjectFactory,
   storeProjectFactory,
   storeProjectRoleFactory
 } from '@/modules/core/repositories/projects'
 import { mainDb } from '@/db/knex'
-import { storeModelFactory } from '@/modules/core/repositories/models'
-import { getEventBus } from '@/modules/shared/services/eventBus'
 import {
   getWorkspaceFactory,
   upsertWorkspaceFactory
@@ -59,6 +53,8 @@ import type { FindEmailsByUserId } from '@/modules/core/domain/userEmails/operat
 import { userEmailsCompliantWithWorkspaceDomains } from '@/modules/workspaces/domain/logic'
 import type { CreateWorkspaceSeat } from '@/modules/gatekeeper/domain/operations'
 import type { WorkspaceAcl } from '@/modules/workspacesCore/domain/types'
+import { asMultiregionalOperation, replicateFactory } from '@/modules/shared/command'
+import { logger } from '@/observability/logging'
 
 type MoveProjectToWorkspaceArgs = {
   projectId: string
@@ -308,6 +304,7 @@ export const validateWorkspaceMemberProjectRoleFactory =
     }
   }
 
+// This factory uses the command factory to create a new project in transactional (cross region) so it cannot be wrapped in another transaction
 export const createWorkspaceProjectFactory =
   (deps: { getDefaultRegion: GetDefaultRegion }) =>
   async (params: { input: WorkspaceProjectCreateInput; ownerId: string }) => {
@@ -334,26 +331,28 @@ export const createWorkspaceProjectFactory =
       if (!workspace) throw new WorkspaceNotFoundError()
       await upsertWorkspaceFactory({ db: projectDb })({ workspace })
     }
+    const project = await asMultiregionalOperation(
+      async ({ allDbs, mainDb, emit }) => {
+        const createNewProject = createNewProjectFactory({
+          // TODO: this goes as event emmits outside  (default model)
+          storeProject: replicateFactory(allDbs, storeProjectFactory),
+          // THIS MUST GO TO THE MAIN DB
+          storeProjectRole: storeProjectRoleFactory({ db: mainDb }),
+          emitEvent: emit
+        })
 
-    // todo, use the command factory here, but for that, we need to migrate to the event bus
-    // deps not injected to ensure proper DB injection
-    const createNewProject = createNewProjectFactory({
-      storeProject: storeProjectFactory({ db: projectDb }),
-      storeModel: storeModelFactory({ db: projectDb }),
-      // THIS MUST GO TO THE MAIN DB
-      storeProjectRole: storeProjectRoleFactory({ db }),
-      waitForRegionProject: waitForRegionProjectFactory({
-        getProject: getProjectFactory({ db }),
-        deleteProject: deleteProjectFactory({ db: projectDb })
-      }),
-      emitEvent: getEventBus().emit
-    })
-
-    const project = await createNewProject({
-      ...input,
-      regionKey,
-      ownerId
-    })
+        return createNewProject({
+          ...input,
+          regionKey,
+          ownerId
+        })
+      },
+      {
+        dbs: await getReplicationDbs({ regionKey }),
+        name: 'Create project workspace',
+        logger
+      }
+    )
 
     return project
   }

packages/server/modules/workspaces/tests/integration/projects.graph.spec.ts

@@ -2,12 +2,7 @@ import { db } from '@/db/knex'
 import { StreamAcl, Streams } from '@/modules/core/dbSchema'
 import type { StreamRecord } from '@/modules/core/helpers/types'
 import { ProjectRecordVisibility } from '@/modules/core/helpers/types'
-import {
-  deleteProjectFactory,
-  getProjectFactory
-} from '@/modules/core/repositories/projects'
 import { grantStreamPermissionsFactory } from '@/modules/core/repositories/streams'
-import { waitForRegionProjectFactory } from '@/modules/core/services/projects'
 import { WorkspaceSeatType } from '@/modules/gatekeeper/domain/billing'
 import { getWorkspaceUserSeatsFactory } from '@/modules/gatekeeper/repositories/workspaceSeat'
 import { getRegionDb } from '@/modules/multiregion/utils/dbSelector'
@@ -1010,13 +1005,6 @@ describe('Workspace project GQL CRUD', () => {
             // Simulate non-main default db region
             const regionDb = await getRegionDb({ regionKey: 'region1' })
             await tables.streams(regionDb).insert(regionalProject)
-            await waitForRegionProjectFactory({
-              getProject: getProjectFactory({ db }),
-              deleteProject: deleteProjectFactory({ db: regionDb })
-            })({
-              projectId: regionalProject.id,
-              regionKey: 'region1'
-            })
             await grantStreamPermissions({
               streamId: regionalProject.id,
               userId: serverAdminUser.id,

packages/server/modules/workspaces/tests/integration/workspacesCreationState.spec.ts

@@ -14,17 +14,21 @@ import {
 import { expect } from 'chai'
 import dayjs from 'dayjs'
 import { deleteWorkspacesNonCompleteFactory } from '@/modules/workspaces/services/workspaceCreationState'
+import type { Logger } from '@/observability/logging'
 import { logger } from '@/observability/logging'
-import {
-  deleteStreamFactory,
-  getExplicitProjects
-} from '@/modules/core/repositories/streams'
+import { getExplicitProjects } from '@/modules/core/repositories/streams'
 import { deleteSsoProviderFactory } from '@/modules/workspaces/repositories/sso'
-import { getEventBus } from '@/modules/shared/services/eventBus'
 import { deleteAllResourceInvitesFactory } from '@/modules/serverinvites/repositories/serverInvites'
 import { deleteWorkspaceFactory as repoDeleteWorkspaceFactory } from '@/modules/workspaces/repositories/workspaces'
 import { deleteWorkspaceFactory } from '@/modules/workspaces/services/management'
-import { queryAllProjectsFactory } from '@/modules/core/services/projects'
+import {
+  deleteProjectAndCommitsFactory,
+  queryAllProjectsFactory
+} from '@/modules/core/services/projects'
+import { deleteProjectFactory } from '@/modules/core/repositories/projects'
+import { deleteProjectCommitsFactory } from '@/modules/core/repositories/commits'
+import { asMultiregionalOperation, replicateFactory } from '@/modules/shared/command'
+import { getAllRegisteredDbs } from '@/modules/multiregion/utils/dbSelector'
 
 const updateAWorkspaceCreatedAt = async (
   workspaceId: string,
@@ -39,19 +43,45 @@ const updateAWorkspaceCreatedAt = async (
 describe('WorkspaceCreationState services', () => {
   const getWorkspacesNonComplete = getWorkspacesNonCompleteFactory({ db })
   const getWorkspace = getWorkspaceFactory({ db })
-  const deleteWorkspacesNonComplete = deleteWorkspacesNonCompleteFactory({
-    getWorkspacesNonComplete,
-    deleteWorkspace: deleteWorkspaceFactory({
-      deleteWorkspace: repoDeleteWorkspaceFactory({ db }),
-      deleteProject: deleteStreamFactory({ db }),
-      deleteAllResourceInvites: deleteAllResourceInvitesFactory({ db }),
-      queryAllProjects: queryAllProjectsFactory({
-        getExplicitProjects: getExplicitProjects({ db })
-      }),
-      deleteSsoProvider: deleteSsoProviderFactory({ db }),
-      emitWorkspaceEvent: getEventBus().emit
-    })
-  })
+  const deleteWorkspacesNonComplete = async ({ logger }: { logger: Logger }) =>
+    asMultiregionalOperation(
+      ({ allDbs, mainDb, emit }) => {
+        const deleteWorkspacesNonComplete = deleteWorkspacesNonCompleteFactory({
+          getWorkspacesNonComplete: getWorkspacesNonCompleteFactory({ db: mainDb }),
+          deleteWorkspace: deleteWorkspaceFactory({
+            deleteWorkspace: async (...input) => {
+              const [res] = await Promise.all(
+                allDbs.map((db) => repoDeleteWorkspaceFactory({ db })(...input))
+              )
+
+              return res
+            },
+            deleteProjectAndCommits: deleteProjectAndCommitsFactory({
+              deleteProject: replicateFactory(allDbs, deleteProjectFactory),
+              deleteProjectCommits: replicateFactory(
+                allDbs,
+                deleteProjectCommitsFactory
+              )
+            }),
+            deleteAllResourceInvites: deleteAllResourceInvitesFactory({
+              db: mainDb
+            }),
+            queryAllProjects: queryAllProjectsFactory({
+              getExplicitProjects: getExplicitProjects({ db: mainDb })
+            }),
+            deleteSsoProvider: deleteSsoProviderFactory({ db: mainDb }),
+            emitWorkspaceEvent: emit
+          })
+        })
+
+        return deleteWorkspacesNonComplete({ logger })
+      },
+      {
+        logger,
+        name: 'deleteWorkspacesNonComplete',
+        dbs: await getAllRegisteredDbs()
+      }
+    )
 
   let adminUser: BasicTestUser
   let completeWorkspace: BasicTestWorkspace