feat(regions): move project comments and webhooks

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

@@ -319,3 +319,8 @@ export type GetPaginatedProjectComments = (
   items: CommentRecord[]
   cursor: string | null
 }>
+
+export type GetStreamCommentCount = (
+  streamId: string,
+  options?: Partial<{ threadsOnly: boolean; includeArchived: boolean }>
+) => Promise<number>

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

@@ -61,7 +61,7 @@ export const createNewProjectFactory =
             const replicatedProject = await getProject({ projectId })
             if (!replicatedProject) throw new StreamNotFoundError()
           },
-          { maxAttempts: 10 }
+          { maxAttempts: 100 }
         )
       } catch (err) {
         if (err instanceof StreamNotFoundError) {

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

@@ -369,3 +369,9 @@ export type CopyProjectObjects = (params: {
 export type CopyProjectAutomations = (params: {
   projectIds: string[]
 }) => Promise<Record<string, number>>
+export type CopyProjectComments = (params: {
+  projectIds: string[]
+}) => Promise<Record<string, number>>
+export type CopyProjectWebhooks = (params: {
+  projectIds: string[]
+}) => Promise<Record<string, number>>

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

@@ -11,10 +11,12 @@ import {
 } from '@/modules/workspaces/repositories/regions'
 import {
   copyProjectAutomationsFactory,
+  copyProjectCommentsFactory,
   copyProjectModelsFactory,
   copyProjectObjectsFactory,
   copyProjectsFactory,
   copyProjectVersionsFactory,
+  copyProjectWebhooksFactory,
   copyWorkspaceFactory
 } from '@/modules/workspaces/repositories/projectRegions'
 import {
@@ -35,6 +37,8 @@ import { getStreamObjectCountFactory } from '@/modules/core/repositories/objects
 import { getProjectAutomationsTotalCountFactory } from '@/modules/automate/repositories/automations'
 import { getFeatureFlags, isTestEnv } from '@/modules/shared/helpers/envHelper'
 import { WorkspacesNotYetImplementedError } from '@/modules/workspaces/errors/workspace'
+import { getStreamCommentCountFactory } from '@/modules/comments/repositories/comments'
+import { getStreamWebhooksFactory } from '@/modules/webhooks/repositories/webhooks'
 
 const { FF_MOVE_PROJECT_REGION_ENABLED } = getFeatureFlags()
 
@@ -97,6 +101,8 @@ export default {
         countProjectAutomations: getProjectAutomationsTotalCountFactory({
           db: sourceDb
         }),
+        countProjectComments: getStreamCommentCountFactory({ db: sourceDb }),
+        getProjectWebhooks: getStreamWebhooksFactory({ db: sourceDb }),
         getAvailableRegions: getAvailableRegionsFactory({
           getRegions: getRegionsFactory({ db }),
           canWorkspaceUseRegions: canWorkspaceUseRegionsFactory({
@@ -108,7 +114,9 @@ export default {
         copyProjectModels: copyProjectModelsFactory({ sourceDb, targetDb }),
         copyProjectVersions: copyProjectVersionsFactory({ sourceDb, targetDb }),
         copyProjectObjects: copyProjectObjectsFactory({ sourceDb, targetDb }),
-        copyProjectAutomations: copyProjectAutomationsFactory({ sourceDb, targetDb })
+        copyProjectAutomations: copyProjectAutomationsFactory({ sourceDb, targetDb }),
+        copyProjectComments: copyProjectCommentsFactory({ sourceDb, targetDb }),
+        copyProjectWebhooks: copyProjectWebhooksFactory({ sourceDb, targetDb })
       })
 
       return await withTransaction(updateProjectRegion(args), targetDb)

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

@@ -9,6 +9,9 @@ import {
   AutomationTriggers,
   BranchCommits,
   Branches,
+  CommentLinks,
+  Comments,
+  CommentViews,
   Commits,
   Objects,
   StreamCommits,
@@ -30,10 +33,12 @@ import {
 import { executeBatchedSelect } from '@/modules/shared/helpers/dbHelper'
 import {
   CopyProjectAutomations,
+  CopyProjectComments,
   CopyProjectModels,
   CopyProjectObjects,
   CopyProjects,
   CopyProjectVersions,
+  CopyProjectWebhooks,
   CopyWorkspace
 } from '@/modules/workspaces/domain/operations'
 import { WorkspaceNotFoundError } from '@/modules/workspaces/errors/workspace'
@@ -51,6 +56,12 @@ import {
   AutomationTokenRecord,
   AutomationTriggerDefinitionRecord
 } from '@/modules/automate/helpers/types'
+import {
+  CommentLinkRecord,
+  CommentRecord,
+  CommentViewRecord
+} from '@/modules/comments/helpers/types'
+import { Webhook, WebhookEvent } from '@/modules/webhooks/domain/types'
 
 const tables = {
   workspaces: (db: Knex) => db<Workspace>(Workspaces.name),
@@ -75,7 +86,12 @@ const tables = {
   automationRunTriggers: (db: Knex) =>
     db<AutomationRunTriggerRecord>(AutomationRunTriggers.name),
   automationFunctionRuns: (db: Knex) =>
-    db<AutomationFunctionRunRecord>(AutomationFunctionRuns.name)
+    db<AutomationFunctionRunRecord>(AutomationFunctionRuns.name),
+  comments: (db: Knex) => db.table<CommentRecord>(Comments.name),
+  commentViews: (db: Knex) => db.table<CommentViewRecord>(CommentViews.name),
+  commentLinks: (db: Knex) => db.table<CommentLinkRecord>(CommentLinks.name),
+  webhooks: (db: Knex) => db.table<Webhook>('webhooks_config'),
+  webhookEvents: (db: Knex) => db.table<WebhookEvent>('webhooks_events')
 }
 
 /**
@@ -459,3 +475,105 @@ export const copyProjectAutomationsFactory =
 
     return copiedAutomationCountByProjectId
   }
+
+/**
+ * Copies rows from the following tables:
+ * - comments
+ * - comment_views
+ * - comment_links
+ */
+export const copyProjectCommentsFactory =
+  (deps: { sourceDb: Knex; targetDb: Knex }): CopyProjectComments =>
+  async ({ projectIds }) => {
+    const copiedCommentCountByProjectId: Record<string, number> = {}
+
+    // Copy `comments` table rows in batches
+    const selectComments = tables
+      .comments(deps.sourceDb)
+      .select('*')
+      .whereIn(Comments.col.streamId, projectIds)
+
+    for await (const comments of executeBatchedSelect(selectComments)) {
+      const commentIds = comments.map((comment) => comment.id)
+
+      // Write `comments` rows to target db
+      await tables.comments(deps.targetDb).insert(comments).onConflict().ignore()
+
+      for (const comment of comments) {
+        copiedCommentCountByProjectId[comment.streamId] ??= 0
+        copiedCommentCountByProjectId[comment.streamId]++
+      }
+
+      // Copy `comment_views` table rows
+      const commentViews = await tables
+        .commentViews(deps.sourceDb)
+        .select('*')
+        .whereIn(CommentViews.col.commentId, commentIds)
+
+      await tables
+        .commentViews(deps.targetDb)
+        .insert(commentViews)
+        .onConflict()
+        .ignore()
+
+      // Copy `comment_links` table rows
+      const commentLinks = await tables
+        .commentLinks(deps.sourceDb)
+        .select('*')
+        .whereIn(CommentLinks.col.commentId, commentIds)
+
+      await tables
+        .commentLinks(deps.targetDb)
+        .insert(commentLinks)
+        .onConflict()
+        .ignore()
+    }
+
+    return copiedCommentCountByProjectId
+  }
+
+/**
+ * Copies rows from the following tables:
+ * - webhooks_config
+ * - webhooks_events
+ */
+export const copyProjectWebhooksFactory =
+  (deps: { sourceDb: Knex; targetDb: Knex }): CopyProjectWebhooks =>
+  async ({ projectIds }) => {
+    const copiedWebhookCountByProjectId: Record<string, number> = {}
+
+    // Copy `webhooks_config` table rows in batches
+    const selectWebhooks = tables
+      .webhooks(deps.sourceDb)
+      .select('*')
+      .whereIn('streamId', projectIds)
+
+    for await (const webhooks of executeBatchedSelect(selectWebhooks)) {
+      const webhookIds = webhooks.map((webhook) => webhook.id)
+
+      // Write `webhooks_config` rows to target db
+      await tables.webhooks(deps.targetDb).insert(webhooks).onConflict().ignore()
+
+      for (const webhook of webhooks) {
+        copiedWebhookCountByProjectId[webhook.streamId] ??= 0
+        copiedWebhookCountByProjectId[webhook.streamId]++
+      }
+
+      // Copy `webhooks_events` table rows in batches
+      const selectWebhookEvents = tables
+        .webhookEvents(deps.sourceDb)
+        .select('*')
+        .whereIn('webhookId', webhookIds)
+
+      for await (const webhookEvents of executeBatchedSelect(selectWebhookEvents)) {
+        // Write `webhooks_events` rows to target db
+        await tables
+          .webhookEvents(deps.targetDb)
+          .insert(webhookEvents)
+          .onConflict()
+          .ignore()
+      }
+    }
+
+    return copiedWebhookCountByProjectId
+  }

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

@@ -1,14 +1,18 @@
 import { GetProjectAutomationCount } from '@/modules/automate/domain/operations'
+import { GetStreamCommentCount } from '@/modules/comments/domain/operations'
 import { GetStreamBranchCount } from '@/modules/core/domain/branches/operations'
 import { GetStreamCommitCount } from '@/modules/core/domain/commits/operations'
 import { GetStreamObjectCount } from '@/modules/core/domain/objects/operations'
 import { GetProject } from '@/modules/core/domain/projects/operations'
+import { GetStreamWebhooks } from '@/modules/webhooks/domain/operations'
 import {
   CopyProjectAutomations,
+  CopyProjectComments,
   CopyProjectModels,
   CopyProjectObjects,
   CopyProjects,
   CopyProjectVersions,
+  CopyProjectWebhooks,
   CopyWorkspace,
   GetAvailableRegions,
   UpdateProjectRegion
@@ -22,13 +26,17 @@ export const updateProjectRegionFactory =
     countProjectVersions: GetStreamCommitCount
     countProjectObjects: GetStreamObjectCount
     countProjectAutomations: GetProjectAutomationCount
+    countProjectComments: GetStreamCommentCount
+    getProjectWebhooks: GetStreamWebhooks
     getAvailableRegions: GetAvailableRegions
     copyWorkspace: CopyWorkspace
     copyProjects: CopyProjects
     copyProjectModels: CopyProjectModels
     copyProjectVersions: CopyProjectVersions
     copyProjectObjects: CopyProjectObjects
     copyProjectAutomations: CopyProjectAutomations
+    copyProjectComments: CopyProjectComments
+    copyProjectWebhooks: CopyProjectWebhooks
   }): UpdateProjectRegion =>
   async (params) => {
     const { projectId, regionKey } = params
@@ -74,9 +82,13 @@ export const updateProjectRegionFactory =
     // Move automations
     const copiedAutomationCount = await deps.copyProjectAutomations({ projectIds })
 
-    // TODO: Move comments
+    // Move comments
+    const copiedCommentCount = await deps.copyProjectComments({ projectIds })
+
+    // Move webhooks
+    const copiedWebhookCount = await deps.copyProjectWebhooks({ projectIds })
+
     // TODO: Move file blobs
-    // TODO: Move webhooks
 
     // TODO: Validate state after move captures latest state of project
     const sourceProjectModelCount = await deps.countProjectModels(projectId)
@@ -87,12 +99,16 @@ export const updateProjectRegionFactory =
     const sourceProjectAutomationCount = await deps.countProjectAutomations({
       projectId
     })
+    const sourceProjectCommentCount = await deps.countProjectComments(projectId)
+    const sourceProjectWebhooks = await deps.getProjectWebhooks({ streamId: projectId })
 
     const tests = [
       copiedModelCount[projectId] === sourceProjectModelCount,
       copiedVersionCount[projectId] === sourceProjectVersionCount,
       copiedObjectCount[projectId] === sourceProjectObjectCount,
-      copiedAutomationCount[projectId] === sourceProjectAutomationCount
+      copiedAutomationCount[projectId] === sourceProjectAutomationCount,
+      copiedCommentCount[projectId] === sourceProjectCommentCount,
+      copiedWebhookCount[projectId] === sourceProjectWebhooks.length
     ]
 
     if (!tests.every((test) => !!test)) {