[connectors] Launch workflows from ZendeskConnectorManager

connectors/src/connectors/zendesk/index.ts

@@ -5,9 +5,7 @@ import type {
   ContentNodesViewType,
   Result,
 } from "@dust-tt/types";
-import { assertNever } from "@dust-tt/types";
-import { Err } from "@dust-tt/types";
-import { Ok } from "@dust-tt/types";
+import { assertNever, Err, Ok } from "@dust-tt/types";
 
 import type { ConnectorManagerError } from "@connectors/connectors/interface";
 import { BaseConnectorManager } from "@connectors/connectors/interface";
@@ -37,13 +35,18 @@ import {
   revokeSyncZendeskTickets,
 } from "@connectors/connectors/zendesk/lib/ticket_permissions";
 import { getZendeskAccessToken } from "@connectors/connectors/zendesk/lib/zendesk_access_token";
+import {
+  launchZendeskSyncWorkflow,
+  stopZendeskSyncWorkflow,
+} from "@connectors/connectors/zendesk/temporal/client";
+import { dataSourceConfigFromConnector } from "@connectors/lib/api/data_source_config";
 import logger from "@connectors/logger/logger";
 import { ConnectorResource } from "@connectors/resources/connector_resource";
-import { ZendeskConfigurationResource } from "@connectors/resources/zendesk_resources";
 import {
   ZendeskArticleResource,
   ZendeskBrandResource,
   ZendeskCategoryResource,
+  ZendeskConfigurationResource,
   ZendeskTicketResource,
 } from "@connectors/resources/zendesk_resources";
 import type { DataSourceConfig } from "@connectors/types/data_source_config";
@@ -69,6 +72,22 @@ export class ZendeskConnectorManager extends BaseConnectorManager<null> {
       { subdomain: "d3v-dust", conversationsSlidingWindow: 90 }
     );
 
+    const workflowStartResult = await launchZendeskSyncWorkflow({
+      connectorId: connector.id,
+    });
+
+    if (workflowStartResult.isErr()) {
+      await connector.delete();
+      logger.error(
+        {
+          workspaceId: dataSourceConfig.workspaceId,
+          error: workflowStartResult.error,
+        },
+        "[Zendesk] Error launching the sync workflow."
+      );
+      throw workflowStartResult.error;
+    }
+
     return new Ok(connector.id.toString());
   }
 
@@ -86,15 +105,52 @@ export class ZendeskConnectorManager extends BaseConnectorManager<null> {
   }
 
   async stop(): Promise<Result<undefined, Error>> {
-    throw new Error("Method not implemented.");
+    const result = await stopZendeskSyncWorkflow(this.connectorId);
+    return result.isErr() ? result : new Ok(undefined);
   }
 
   async resume(): Promise<Result<undefined, Error>> {
-    throw new Error("Method not implemented.");
+    const connectorId = this.connectorId;
+    const connector = await ConnectorResource.fetchById(connectorId);
+    if (!connector) {
+      logger.error({ connectorId }, "[Zendesk] Connector not found.");
+      return new Err(new Error("Connector not found"));
+    }
+
+    const dataSourceConfig = dataSourceConfigFromConnector(connector);
+    try {
+      await launchZendeskSyncWorkflow({ connectorId });
+    } catch (e) {
+      logger.error(
+        {
+          workspaceId: dataSourceConfig.workspaceId,
+          dataSourceId: dataSourceConfig.dataSourceId,
+          error: e,
+        },
+        "[Zendesk] Error resuming the sync workflow."
+      );
+    }
+    return new Ok(undefined);
   }
 
   async sync(): Promise<Result<string, Error>> {
-    throw new Error("Method not implemented.");
+    const connectorId = this.connectorId;
+    const connector = await ConnectorResource.fetchById(connectorId);
+    if (!connector) {
+      logger.error({ connectorId }, "[Zendesk] Connector not found.");
+      return new Err(new Error("Connector not found"));
+    }
+
+    const brandIds = await ZendeskBrandResource.fetchAllBrandIds({
+      connectorId,
+    });
+    const result = await launchZendeskSyncWorkflow({
+      connectorId,
+      brandIds,
+      forceResync: true,
+    });
+
+    return result.isErr() ? result : new Ok(connector.id.toString());
   }
 
   async retrievePermissions({
@@ -146,9 +202,8 @@ export class ZendeskConnectorManager extends BaseConnectorManager<null> {
     }
 
     const connectionId = connector.connectionId;
-    const zendeskConfiguration = await ZendeskConfigurationResource.fetchById(
-      this.connectorId
-    );
+    const zendeskConfiguration =
+      await ZendeskConfigurationResource.fetchByConnectorId(connectorId);
     if (!zendeskConfiguration) {
       logger.error(
         { connectorId },
@@ -276,7 +331,23 @@ export class ZendeskConnectorManager extends BaseConnectorManager<null> {
       }
     }
 
-    /// Launch a sync workflow here
+    if (
+      toBeSignaledBrandIds.size > 0 ||
+      toBeSignaledTicketsIds.size > 0 ||
+      toBeSignaledHelpCenterIds.size > 0 ||
+      toBeSignaledCategoryIds.size > 0
+    ) {
+      const sendSignalToWorkflowResult = await launchZendeskSyncWorkflow({
+        connectorId,
+        brandIds: [...toBeSignaledBrandIds],
+        ticketsBrandIds: [...toBeSignaledTicketsIds],
+        helpCenterBrandIds: [...toBeSignaledHelpCenterIds],
+        categoryIds: [...toBeSignaledCategoryIds],
+      });
+      if (sendSignalToWorkflowResult.isErr()) {
+        return new Err(sendSignalToWorkflowResult.error);
+      }
+    }
 
     return new Ok(undefined);
   }
@@ -469,11 +540,33 @@ export class ZendeskConnectorManager extends BaseConnectorManager<null> {
   }
 
   async pause(): Promise<Result<undefined, Error>> {
-    throw new Error("Method not implemented.");
+    const connectorId = this.connectorId;
+    const connector = await ConnectorResource.fetchById(connectorId);
+    if (!connector) {
+      logger.error({ connectorId }, "[Zendesk] Connector not found.");
+      return new Err(new Error("Connector not found"));
+    }
+    await connector.markAsPaused();
+    const result = await this.stop();
+
+    return result.isErr() ? result : new Ok(undefined);
   }
 
   async unpause(): Promise<Result<undefined, Error>> {
-    throw new Error("Method not implemented.");
+    const connectorId = this.connectorId;
+    const connector = await ConnectorResource.fetchById(connectorId);
+    if (!connector) {
+      logger.error({ connectorId }, "[Zendesk] Connector not found.");
+      return new Err(new Error("Connector not found"));
+    }
+    await connector.markAsUnpaused();
+
+    const brandIds = await ZendeskBrandResource.fetchAllBrandIds({
+      connectorId,
+    });
+    const result = await launchZendeskSyncWorkflow({ connectorId, brandIds });
+
+    return result.isErr() ? result : new Ok(undefined);
   }
 
   async garbageCollect(): Promise<Result<string, Error>> {

connectors/src/connectors/zendesk/temporal/activities.ts

@@ -35,7 +35,7 @@ async function _getZendeskConnectorOrRaise(connectorId: ModelId) {
 
 async function _getZendeskConfigurationOrRaise(connectorId: ModelId) {
   const configuration =
-    await ZendeskConfigurationResource.fetchById(connectorId);
+    await ZendeskConfigurationResource.fetchByConnectorId(connectorId);
   if (!configuration) {
     throw new Error("[Zendesk] Configuration not found.");
   }

connectors/src/resources/zendesk_resources.ts

@@ -215,6 +215,18 @@ export class ZendeskBrandResource extends BaseResource<ZendeskBrand> {
     return brands.map((brand) => new this(this.model, brand.get()));
   }
 
+  static async fetchAllBrandIds({
+    connectorId,
+  }: {
+    connectorId: number;
+  }): Promise<number[]> {
+    const brands = await ZendeskBrand.findAll({
+      where: { connectorId },
+      attributes: ["brandId"],
+    });
+    return brands.map((brand) => brand.brandId);
+  }
+
   static async fetchAllWithHelpCenter({
     connectorId,
   }: {