[connectors] Implement incremental sync for Zendesk tickets (#8617)

* feat: add a function that fetches the recently updated tickets

* feat: add activities to update the recently updated tickets

* feat: add an incremental sync workflow

* perf: use the `show_many` endpoint instead of fetching all the users when updating the tickets

* fix: prevent extra syncs of help centers/tickets when the whole brand is synced

* fix: fix date conversions

* fix: fix the structure of the response of the incremental endpoint

* feat: implement ticket scrub

* docs: add descriptions for a few functions

* feat: add a log to the ticket deletion

* fix: fix the use of the `show_many` endpoint

* refactor: remove leading underscores in function names

* refactor: move node-zendesk types to @connectors/@types (consistent with talisman)

* refactor: remove an unnecessary import

* refactor: inline getZendeskConnectorOrRaise and getZendeskCategoryOrRaise

* refactor: inline syncBrandWithPermissions and fix incorrect permissions for brand

* refactor: move sync_(article|ticket) from temporal to lib

* fix: use the lastSyncSuccessfulTime as the start time instead of now - 5 min

* fix: add a column lastSuccessfulSyncStartTs to the ZendeskConfiguration and use it

* lint

* fix: only sync solved tickets

* feat: add a table to store workspace information

* fix: update the type of ticket subjects and category descriptions from string to text

* feat: add the migration file

* feat: add ZendeskWorkspace to db.ts

* feat: add the migration file

* fix naming

* fix: add the connectorId when creating a WorkspaceResource

* refactor: renaming the table zendesk_workspaces into zendesk_timestamp_cursors, removing the Resource

* fix migration file

* fix: rename zendesk_workspaces

* fix: add create the cursors upon saving

* fix: recast cursor as date (received as a string)

* fix: delete the cursor when deleting the connector

* fix: fix how we handle StartTimeTooRecent errors

connectors/migrations/db/migration_34.sql

@@ -0,0 +1,14 @@
+-- Migration created on Nov 14, 2024
+CREATE TABLE IF NOT EXISTS "zendesk_timestamp_cursors"
+(
+    "id"                        SERIAL,
+    "createdAt"                 TIMESTAMP WITH TIME ZONE NOT NULL,
+    "updatedAt"                 TIMESTAMP WITH TIME ZONE NOT NULL,
+    "timestampCursor"           TIMESTAMP WITH TIME ZONE DEFAULT NULL,
+    "connectorId"               INTEGER                  NOT NULL REFERENCES "connectors" ("id") ON DELETE RESTRICT ON UPDATE CASCADE,
+    PRIMARY KEY ("id")
+);
+CREATE UNIQUE INDEX "zendesk_timestamp_cursors_connector_id" ON "zendesk_timestamp_cursors" ("connectorId");
+
+ALTER TABLE "public"."zendesk_categories" ALTER COLUMN "description" TYPE TEXT;
+ALTER TABLE "public"."zendesk_tickets" ALTER COLUMN "subject" TYPE TEXT;

connectors/src/connectors/zendesk/lib/node-zendesk-types.d.ts → connectors/src/@types/node-zendesk.d.ts

@@ -1,5 +1,3 @@
-import "node-zendesk";
-
 import type { ZendeskClientOptions } from "node-zendesk";
 
 interface ZendeskFetchedBrand {
@@ -118,7 +116,7 @@ interface ZendeskFetchedTicket {
     score: string;
   };
   sharing_agreement_ids: number[];
-  status: "new" | "open" | "pending" | "hold" | "solved" | "closed";
+  status: "new" | "open" | "pending" | "hold" | "solved" | "closed" | "deleted";
   subject: string;
   submitter_id: number;
   tags: string[];
@@ -239,6 +237,9 @@ declare module "node-zendesk" {
       show: (
         userId: number
       ) => Promise<{ response: Response; result: ZendeskFetchedUser }>;
+      showMany: (
+        userIds: number[]
+      ) => Promise<{ response: Response; result: ZendeskFetchedUser[] }>;
     };
   }
 

connectors/src/admin/db.ts

@@ -67,6 +67,7 @@ import {
   ZendeskCategory,
   ZendeskConfiguration,
   ZendeskTicket,
+  ZendeskTimestampCursors,
 } from "@connectors/lib/models/zendesk";
 import logger from "@connectors/logger/logger";
 import { sequelizeConnection } from "@connectors/resources/storage";
@@ -120,6 +121,7 @@ async function main(): Promise<void> {
   await RemoteDatabaseModel.sync({ alter: true });
   await RemoteSchemaModel.sync({ alter: true });
   await RemoteTableModel.sync({ alter: true });
+  await ZendeskTimestampCursors.sync({ alter: true });
   await ZendeskConfiguration.sync({ alter: true });
   await ZendeskBrand.sync({ alter: true });
   await ZendeskCategory.sync({ alter: true });

connectors/src/connectors/zendesk/lib/brand_permissions.ts

@@ -2,7 +2,8 @@ import type { ModelId } from "@dust-tt/types";
 
 import { allowSyncZendeskHelpCenter } from "@connectors/connectors/zendesk/lib/help_center_permissions";
 import { allowSyncZendeskTickets } from "@connectors/connectors/zendesk/lib/ticket_permissions";
-import { syncBrandWithPermissions } from "@connectors/connectors/zendesk/lib/utils";
+import { getZendeskSubdomainAndAccessToken } from "@connectors/connectors/zendesk/lib/zendesk_access_token";
+import { createZendeskClient } from "@connectors/connectors/zendesk/lib/zendesk_api";
 import logger from "@connectors/logger/logger";
 import { ZendeskBrandResource } from "@connectors/resources/zendesk_resources";
 
@@ -18,17 +19,43 @@ export async function allowSyncZendeskBrand({
   connectionId: string;
   brandId: number;
 }): Promise<boolean> {
-  const syncSuccess = await syncBrandWithPermissions({
+  const brand = await ZendeskBrandResource.fetchByBrandId({
     connectorId,
-    connectionId,
     brandId,
-    permissions: {
-      ticketsPermission: "none",
-      helpCenterPermission: "read",
-    },
   });
-  if (!syncSuccess) {
-    return false; // stopping early if the brand sync failed
+
+  if (brand) {
+    await brand.grantTicketsPermissions();
+    await brand.grantHelpCenterPermissions();
+  } else {
+    // fetching the brand from Zendesk
+    const zendeskApiClient = createZendeskClient(
+      await getZendeskSubdomainAndAccessToken(connectionId)
+    );
+    const {
+      result: { brand: fetchedBrand },
+    } = await zendeskApiClient.brand.show(brandId);
+
+    if (!fetchedBrand) {
+      logger.error(
+        { connectorId, brandId },
+        "[Zendesk] Brand could not be fetched."
+      );
+      return false;
+    }
+
+    await ZendeskBrandResource.makeNew({
+      blob: {
+        subdomain: fetchedBrand.subdomain,
+        connectorId: connectorId,
+        brandId: fetchedBrand.id,
+        name: fetchedBrand.name || "Brand",
+        ticketsPermission: "read",
+        helpCenterPermission: "read",
+        hasHelpCenter: fetchedBrand.has_help_center,
+        url: fetchedBrand.url,
+      },
+    });
   }
 
   await allowSyncZendeskHelpCenter({

connectors/src/connectors/zendesk/lib/help_center_permissions.ts

@@ -1,6 +1,5 @@
 import type { ModelId } from "@dust-tt/types";
 
-import { syncBrandWithPermissions } from "@connectors/connectors/zendesk/lib/utils";
 import { getZendeskSubdomainAndAccessToken } from "@connectors/connectors/zendesk/lib/zendesk_access_token";
 import {
   changeZendeskClientSubdomain,
@@ -30,19 +29,39 @@ export async function allowSyncZendeskHelpCenter({
   const zendeskApiClient = createZendeskClient(
     await getZendeskSubdomainAndAccessToken(connectionId)
   );
-
-  const syncSuccess = await syncBrandWithPermissions({
-    zendeskApiClient,
-    connectionId,
+  const brand = await ZendeskBrandResource.fetchByBrandId({
     connectorId,
     brandId,
-    permissions: {
-      ticketsPermission: "none",
-      helpCenterPermission: "read",
-    },
   });
-  if (!syncSuccess) {
-    return false; // stopping early if the brand sync failed
+
+  if (brand) {
+    await brand.grantHelpCenterPermissions();
+  } else {
+    // fetching the brand from Zendesk
+    const {
+      result: { brand: fetchedBrand },
+    } = await zendeskApiClient.brand.show(brandId);
+
+    if (!fetchedBrand) {
+      logger.error(
+        { connectorId, brandId },
+        "[Zendesk] Brand could not be fetched."
+      );
+      return false;
+    }
+
+    await ZendeskBrandResource.makeNew({
+      blob: {
+        subdomain: fetchedBrand.subdomain,
+        connectorId: connectorId,
+        brandId: fetchedBrand.id,
+        name: fetchedBrand.name || "Brand",
+        ticketsPermission: "none",
+        helpCenterPermission: "read",
+        hasHelpCenter: fetchedBrand.has_help_center,
+        url: fetchedBrand.url,
+      },
+    });
   }
 
   // updating permissions for all the children categories

connectors/src/connectors/zendesk/temporal/sync_article.ts → connectors/src/connectors/zendesk/lib/sync_article.ts

@@ -1,12 +1,12 @@
 import type { ModelId } from "@dust-tt/types";
 import TurndownService from "turndown";
 
-import { getArticleInternalId } from "@connectors/connectors/zendesk/lib/id_conversions";
 import type {
   ZendeskFetchedArticle,
   ZendeskFetchedSection,
   ZendeskFetchedUser,
-} from "@connectors/connectors/zendesk/lib/node-zendesk-types";
+} from "@connectors/@types/node-zendesk";
+import { getArticleInternalId } from "@connectors/connectors/zendesk/lib/id_conversions";
 import {
   renderDocumentTitleAndContent,
   renderMarkdownSection,

connectors/src/connectors/zendesk/temporal/sync_ticket.ts → connectors/src/connectors/zendesk/lib/sync_ticket.ts

@@ -1,13 +1,14 @@
 import type { ModelId } from "@dust-tt/types";
 import TurndownService from "turndown";
 
-import { getTicketInternalId } from "@connectors/connectors/zendesk/lib/id_conversions";
 import type {
   ZendeskFetchedTicket,
   ZendeskFetchedTicketComment,
   ZendeskFetchedUser,
-} from "@connectors/connectors/zendesk/lib/node-zendesk-types";
+} from "@connectors/@types/node-zendesk";
+import { getTicketInternalId } from "@connectors/connectors/zendesk/lib/id_conversions";
 import {
+  deleteFromDataSource,
   renderDocumentTitleAndContent,
   renderMarkdownSection,
   upsertToDatasource,
@@ -18,6 +19,39 @@ import type { DataSourceConfig } from "@connectors/types/data_source_config";
 
 const turndownService = new TurndownService();
 
+/**
+ * Deletes a ticket from the db and the data sources.
+ */
+export async function deleteTicket(
+  connectorId: ModelId,
+  ticket: ZendeskFetchedTicket,
+  dataSourceConfig: DataSourceConfig,
+  loggerArgs: Record<string, string | number | null>
+): Promise<void> {
+  logger.info(
+    {
+      ...loggerArgs,
+      connectorId,
+      ticketId: ticket.id,
+      subject: ticket.subject,
+    },
+    "[Zendesk] Deleting ticket."
+  );
+  await Promise.all([
+    ZendeskTicketResource.deleteByTicketId({
+      connectorId,
+      ticketId: ticket.id,
+    }),
+    deleteFromDataSource(
+      dataSourceConfig,
+      getTicketInternalId(connectorId, ticket.id)
+    ),
+  ]);
+}
+
+/**
+ * Syncs a ticket in the db and upserts it to the data sources.
+ */
 export async function syncTicket({
   connectorId,
   ticket,

connectors/src/connectors/zendesk/lib/ticket_permissions.ts

@@ -1,6 +1,7 @@
 import type { ModelId } from "@dust-tt/types";
 
-import { syncBrandWithPermissions } from "@connectors/connectors/zendesk/lib/utils";
+import { getZendeskSubdomainAndAccessToken } from "@connectors/connectors/zendesk/lib/zendesk_access_token";
+import { createZendeskClient } from "@connectors/connectors/zendesk/lib/zendesk_api";
 import logger from "@connectors/logger/logger";
 import {
   ZendeskBrandResource,
@@ -19,15 +20,43 @@ export async function allowSyncZendeskTickets({
   connectionId: string;
   brandId: number;
 }): Promise<boolean> {
-  return syncBrandWithPermissions({
+  const brand = await ZendeskBrandResource.fetchByBrandId({
     connectorId,
-    connectionId,
     brandId,
-    permissions: {
-      ticketsPermission: "read",
-      helpCenterPermission: "none",
-    },
   });
+
+  if (brand) {
+    await brand.grantTicketsPermissions();
+    return true;
+  }
+  // fetching the brand from Zendesk
+  const zendeskApiClient = createZendeskClient(
+    await getZendeskSubdomainAndAccessToken(connectionId)
+  );
+  const {
+    result: { brand: fetchedBrand },
+  } = await zendeskApiClient.brand.show(brandId);
+
+  if (fetchedBrand) {
+    await ZendeskBrandResource.makeNew({
+      blob: {
+        subdomain: fetchedBrand.subdomain,
+        connectorId: connectorId,
+        brandId: fetchedBrand.id,
+        name: fetchedBrand.name || "Brand",
+        ticketsPermission: "read",
+        helpCenterPermission: "none",
+        hasHelpCenter: fetchedBrand.has_help_center,
+        url: fetchedBrand.url,
+      },
+    });
+    return true;
+  }
+  logger.error(
+    { connectorId, brandId },
+    "[Zendesk] Brand could not be fetched."
+  );
+  return false;
 }
 
 /**