Hubspot automations

backend/src/api/integration/helpers/hubspotConnect.ts

@@ -3,7 +3,7 @@ import IntegrationService from '../../../services/integrationService'
 import PermissionChecker from '../../../services/user/permissionChecker'
 
 export default async (req, res) => {
-  new PermissionChecker(req).validateHas(Permissions.values.tenantEdit)
+  new PermissionChecker(req).validateHas(Permissions.values.integrationEdit)
   const payload = await new IntegrationService(req).hubspotConnect()
   await req.responseHandler.success(req, res, payload)
 }

backend/src/api/integration/helpers/hubspotGetLists.ts

@@ -0,0 +1,9 @@
+import Permissions from '../../../security/permissions'
+import IntegrationService from '../../../services/integrationService'
+import PermissionChecker from '../../../services/user/permissionChecker'
+
+export default async (req, res) => {
+  new PermissionChecker(req).validateHas(Permissions.values.integrationEdit)
+  const payload = await new IntegrationService(req).hubspotGetLists()
+  await req.responseHandler.success(req, res, payload)
+}

backend/src/api/integration/index.ts

@@ -102,6 +102,12 @@ export default (app) => {
     safeWrap(require('./helpers/hubspotGetMappableFields').default),
   )
 
+  app.get(
+    '/tenant/:tenantId/hubspot-get-lists',
+    featureFlagMiddleware(FeatureFlag.HUBSPOT, 'hubspot.errors.notInPlan'),
+    safeWrap(require('./helpers/hubspotGetLists').default),
+  )
+
   app.post(
     '/tenant/:tenantId/hubspot-sync-member',
     featureFlagMiddleware(FeatureFlag.HUBSPOT, 'hubspot.errors.notInPlan'),

backend/src/database/migrations/V1691074653__syncRemoteTables.sql

@@ -0,0 +1,30 @@
+create table "membersSyncRemote" (
+    id uuid not null,
+    "memberId" uuid not null,
+    "sourceId" text,
+    "integrationId" uuid not null,
+    "syncFrom" text not null,
+    "metaData" text,
+    "lastSyncedAt" timestamptz,
+    "status" text not null,
+    constraint "membersSyncRemote_pkey" primary key (id),
+    foreign key ("memberId") references members (id) on delete cascade,
+    foreign key ("integrationId") references integrations (id) on delete cascade,
+    unique("memberId", "integrationId", "syncFrom")
+);
+
+
+create table "organizationsSyncRemote" (
+    id uuid not null,
+    "organizationId" uuid not null,
+    "sourceId" text,
+    "integrationId" uuid not null,
+    "syncFrom" text not null,
+    "metaData" text,
+    "lastSyncedAt" timestamptz,
+    "status" text not null,
+    constraint "organizationsSyncRemote_pkey" primary key (id),
+    foreign key ("organizationId") references organizations (id) on delete cascade,
+    foreign key ("integrationId") references integrations (id) on delete cascade,
+    unique("organizationId", "integrationId", "syncFrom")
+);

backend/src/database/migrations/V1691152131__syncRemoteTablesMissingIndexes.sql

@@ -0,0 +1,11 @@
+create index if not exists "ix_membersSyncRemote_memberId" on "membersSyncRemote" ("memberId");
+create index if not exists "ix_membersSyncRemote_integrationId" on "membersSyncRemote" ("integrationId");
+create index if not exists "ix_membersSyncRemote_syncFrom" on "membersSyncRemote" ("syncFrom");
+create index if not exists "ix_membersSyncRemote_sourceId" on "membersSyncRemote" ("sourceId");
+create index if not exists "ix_membersSyncRemote_status" on "membersSyncRemote" ("status");
+
+create index if not exists "ix_organizationsSyncRemote_organizationId" on "organizationsSyncRemote" ("organizationId");
+create index if not exists "ix_organizationsSyncRemote_integrationId" on "organizationsSyncRemote" ("integrationId");
+create index if not exists "ix_organizationsSyncRemote_syncFrom" on "organizationsSyncRemote" ("syncFrom");
+create index if not exists "ix_organizationsSyncRemote_sourceId" on "organizationsSyncRemote" ("sourceId");
+create index if not exists "ix_organizationsSyncRemote_status" on "organizationsSyncRemote" ("status");

backend/src/database/repositories/automationExecutionRepository.ts

@@ -137,6 +137,29 @@ export default class AutomationExecutionRepository extends RepositoryBase<
     throw new Error('Method not implemented.')
   }
 
+  async destroyAllAutomation(automationIds: string[]): Promise<void> {
+    const transaction = this.transaction
+
+    const seq = this.seq
+
+    const currentTenant = this.currentTenant
+
+    const query = `
+    delete 
+    from "automationExecutions"
+    where "automationId" in (:automationIds)
+      and "tenantId" = :tenantId;`
+
+    await seq.query(query, {
+      replacements: {
+        automationIds,
+        tenantId: currentTenant.id,
+      },
+      type: QueryTypes.DELETE,
+      transaction,
+    })
+  }
+
   override async destroyAll(ids: string[]): Promise<void> {
     throw new Error('Method not implemented.')
   }

backend/src/database/repositories/automationRepository.ts

@@ -1,4 +1,5 @@
 import Sequelize, { QueryTypes } from 'sequelize'
+import { AutomationSyncTrigger, IAutomation } from '@crowd/types'
 import AuditLogRepository from './auditLogRepository'
 import { IRepositoryOptions } from './IRepositoryOptions'
 import Error404 from '../../errors/Error404'
@@ -246,4 +247,46 @@ export default class AutomationRepository extends RepositoryBase<
 
     return automationCount
   }
+
+  public async findSyncAutomations(
+    tenantId: string,
+    platform: string,
+  ): Promise<IAutomation[] | null> {
+    const seq = this.seq
+
+    const transaction = this.transaction
+
+    const pageSize = 10
+    const syncAutomations: IAutomation[] = []
+
+    let results
+    let offset
+
+    do {
+      offset = results ? pageSize + offset : 0
+
+      results = await seq.query(
+        `select * from automations 
+      where type = :platform and "tenantId" = :tenantId and trigger in (:syncAutomationTriggers)
+      limit :limit offset :offset`,
+        {
+          replacements: {
+            tenantId,
+            platform,
+            syncAutomationTriggers: [
+              AutomationSyncTrigger.MEMBER_ATTRIBUTES_MATCH,
+              AutomationSyncTrigger.ORGANIZATION_ATTRIBUTES_MATCH,
+            ],
+            limit: pageSize,
+            offset,
+          },
+          type: QueryTypes.SELECT,
+          transaction,
+        },
+      )
+      syncAutomations.push(...results)
+    } while (results.length > 0)
+
+    return syncAutomations
+  }
 }

backend/src/database/repositories/integrationRepository.ts

@@ -8,6 +8,10 @@ import Error404 from '../../errors/Error404'
 import { IRepositoryOptions } from './IRepositoryOptions'
 import QueryParser from './filters/queryParser'
 import { QueryOutput } from './filters/queryTypes'
+import AutomationRepository from './automationRepository'
+import AutomationExecutionRepository from './automationExecutionRepository'
+import MemberSyncRemoteRepository from './memberSyncRemoteRepository'
+import OrganizationSyncRemoteRepository from './organizationSyncRemoteRepository'
 
 const { Op } = Sequelize
 const log: boolean = false
@@ -152,6 +156,30 @@ class IntegrationRepository {
       },
     )
 
+    // delete syncRemote rows coming from integration
+    await new MemberSyncRemoteRepository({ ...options, transaction }).destroyAllIntegration([
+      record.id,
+    ])
+    await new OrganizationSyncRemoteRepository({ ...options, transaction }).destroyAllIntegration([
+      record.id,
+    ])
+
+    // destroy existing automations for outgoing integrations
+    const syncAutomationIds = (
+      await new AutomationRepository({ ...options, transaction }).findSyncAutomations(
+        currentTenant.id,
+        record.platform,
+      )
+    ).map((a) => a.id)
+
+    if (syncAutomationIds.length > 0) {
+      await new AutomationExecutionRepository({ ...options, transaction }).destroyAllAutomation(
+        syncAutomationIds,
+      )
+    }
+
+    await new AutomationRepository({ ...options, transaction }).destroyAll(syncAutomationIds)
+
     await this._createAuditLog(AuditLogRepository.DELETE, record, record, options)
   }
 

backend/src/database/repositories/memberRepository.ts

@@ -4,6 +4,7 @@ import {
   MemberAttributeType,
   OpenSearchIndex,
   PlatformType,
+  SyncStatus,
 } from '@crowd/types'
 import lodash, { chunk } from 'lodash'
 import Sequelize, { QueryTypes } from 'sequelize'
@@ -45,6 +46,7 @@ import {
 } from './types/memberTypes'
 import Error400 from '../../errors/Error400'
 import OrganizationRepository from './organizationRepository'
+import MemberSyncRemoteRepository from './memberSyncRemoteRepository'
 
 const { Op } = Sequelize
 
@@ -628,6 +630,11 @@ class MemberRepository {
       throw new Error404()
     }
 
+    // exclude syncRemote attributes, since these are populated from memberSyncRemote table
+    if (data.attributes?.syncRemote) {
+      delete data.attributes.syncRemote
+    }
+
     record = await record.update(
       {
         ...lodash.pick(data, [
@@ -3215,6 +3222,21 @@ class MemberRepository {
 
     output.affiliations = await this.getAffiliations(record.id, options)
 
+    const manualSyncRemote = await new MemberSyncRemoteRepository({
+      ...options,
+      transaction,
+    }).findMemberManualSync(record.id)
+
+    for (const syncRemote of manualSyncRemote) {
+      if (output.attributes?.syncRemote) {
+        output.attributes.syncRemote[syncRemote.platform] = syncRemote.status === SyncStatus.ACTIVE
+      } else {
+        output.attributes.syncRemote = {
+          [syncRemote.platform]: syncRemote.status === SyncStatus.ACTIVE,
+        }
+      }
+    }
+
     return output
   }