[connectors] Add a worker for Zendesk workflows (#8343)

* feat: add a worker and start it

* set the workflow version to 0

* fix: fix imports in the workflow (activities were imported)

* feat: remove the sleep as this workflow is a cron one

* fix: copy paste the assertNever function since the types module cannot be imported in a workflow

* chore: remove an unused config parameter

* fix: use temporal's assertNever instead of a custom one

connectors/src/connectors/zendesk/temporal/config.ts

@@ -1,3 +1,2 @@
-export const WORKFLOW_VERSION = 3;
+export const WORKFLOW_VERSION = 0;
 export const QUEUE_NAME = `zendesk-queue-v${WORKFLOW_VERSION}`;
-export const INTERVAL_BETWEEN_SYNCS_MS = 60_000; // 1 minute

connectors/src/connectors/zendesk/temporal/worker.ts

@@ -0,0 +1,39 @@
+import type { Context } from "@temporalio/activity";
+import { Worker } from "@temporalio/worker";
+import TsconfigPathsPlugin from "tsconfig-paths-webpack-plugin";
+
+import { getTemporalWorkerConnection } from "@connectors/lib/temporal";
+import { ActivityInboundLogInterceptor } from "@connectors/lib/temporal_monitoring";
+import logger from "@connectors/logger/logger";
+
+import * as activities from "./activities";
+import { QUEUE_NAME } from "./config";
+
+export async function runZendeskWorker() {
+  const { connection, namespace } = await getTemporalWorkerConnection();
+  const worker = await Worker.create({
+    workflowsPath: require.resolve("./workflows"),
+    activities,
+    taskQueue: QUEUE_NAME,
+    connection,
+    reuseV8Context: true,
+    namespace,
+    maxConcurrentActivityTaskExecutions: 16,
+    interceptors: {
+      activityInbound: [
+        (ctx: Context) => {
+          return new ActivityInboundLogInterceptor(ctx, logger);
+        },
+      ],
+    },
+    bundlerOptions: {
+      webpackConfigHook: (config) => {
+        const plugins = config.resolve?.plugins ?? [];
+        config.resolve!.plugins = [...plugins, new TsconfigPathsPlugin({})];
+        return config;
+      },
+    },
+  });
+
+  await worker.run();
+}

connectors/src/connectors/zendesk/temporal/workflows.ts

@@ -1,25 +1,22 @@
 import type { ModelId } from "@dust-tt/types";
-import { assertNever } from "@dust-tt/types";
+import { assertNever } from "@temporalio/common/lib/type-helpers";
 import {
   executeChild,
   proxyActivities,
   setHandler,
-  sleep,
   workflowInfo,
 } from "@temporalio/workflow";
 
 import type * as activities from "@connectors/connectors/zendesk/temporal/activities";
-import { syncZendeskTicketsActivity } from "@connectors/connectors/zendesk/temporal/activities";
-import { INTERVAL_BETWEEN_SYNCS_MS } from "@connectors/connectors/zendesk/temporal/config";
 import type { ZendeskUpdateSignal } from "@connectors/connectors/zendesk/temporal/signals";
-
-import { zendeskUpdatesSignal } from "./signals";
+import { zendeskUpdatesSignal } from "@connectors/connectors/zendesk/temporal/signals";
 
 const {
   getZendeskCategoriesActivity,
   syncZendeskBrandActivity,
   syncZendeskCategoryActivity,
   syncZendeskArticlesActivity,
+  syncZendeskTicketsActivity,
 } = proxyActivities<typeof activities>({
   startToCloseTimeout: "5 minutes",
 });
@@ -81,7 +78,10 @@ export async function zendeskSyncWorkflow({
           break;
         }
         default:
-          assertNever(signal.type);
+          assertNever(
+            `Unexpected signal type ${signal.type} received within Zendesk sync workflow.`,
+            signal.type
+          );
       }
     });
   });
@@ -182,8 +182,6 @@ export async function zendeskSyncWorkflow({
   // run cleanup here if needed
 
   await saveZendeskConnectorSuccessSync({ connectorId });
-
-  await sleep(INTERVAL_BETWEEN_SYNCS_MS);
 }
 
 /**

connectors/src/start.ts

@@ -11,6 +11,7 @@ import { runIntercomWorker } from "./connectors/intercom/temporal/worker";
 import { runNotionWorker } from "./connectors/notion/temporal/worker";
 import { runSlackWorker } from "./connectors/slack/temporal/worker";
 import { runWebCrawlerWorker } from "./connectors/webcrawler/temporal/worker";
+import { runZendeskWorker } from "./connectors/zendesk/temporal/worker";
 import { errorFromAny } from "./lib/error";
 import logger from "./logger/logger";
 
@@ -44,6 +45,9 @@ runGoogleWorkers().catch((err) =>
 runIntercomWorker().catch((err) =>
   logger.error(errorFromAny(err), "Error running intercom worker")
 );
+runZendeskWorker().catch((err) =>
+  logger.error(errorFromAny(err), "Error running zendesk worker")
+);
 runWebCrawlerWorker().catch((err) =>
   logger.error(errorFromAny(err), "Error running webcrawler worker")
 );

connectors/src/start_worker.ts

@@ -16,6 +16,7 @@ import {
   runNotionWorker,
 } from "./connectors/notion/temporal/worker";
 import { runSlackWorker } from "./connectors/slack/temporal/worker";
+import { runZendeskWorker } from "./connectors/zendesk/temporal/worker";
 import { errorFromAny } from "./lib/error";
 import logger from "./logger/logger";
 
@@ -34,9 +35,7 @@ const workerFunctions: Record<WorkerType, () => Promise<void>> = {
   slack: runSlackWorker,
   webcrawler: runWebCrawlerWorker,
   snowflake: runSnowflakeWorker,
-  zendesk: async () => {
-    logger.info("Zendesk worker not implemented yet.");
-  },
+  zendesk: runZendeskWorker,
 };
 
 const ALL_WORKERS = Object.keys(workerFunctions) as WorkerType[];

front/components/ConnectorPermissionsModal.tsx

@@ -51,6 +51,7 @@ const PERMISSIONS_EDITABLE_CONNECTOR_TYPES: Set<ConnectorProvider> = new Set([
   "microsoft",
   "intercom",
   "snowflake",
+  "zendesk",
 ]);
 
 const CONNECTOR_TYPE_TO_PERMISSIONS: Record<
@@ -75,7 +76,10 @@ const CONNECTOR_TYPE_TO_PERMISSIONS: Record<
   },
   notion: undefined,
   github: undefined,
-  zendesk: undefined,
+  zendesk: {
+    selected: "read",
+    unselected: "none",
+  },
   intercom: {
     selected: "read",
     unselected: "none",