feat: implement hubspot webhook target endpoint + webhook setup (#2040)

- Create webhook target on Provider creation
- Create / update webhook subscriptions on SyncConfig update

---------

Co-authored-by: Lucas Marshall <lucasmarshall@users.noreply.github.com>

apps/api/routes/internal/hubspot/index.ts

@@ -0,0 +1,77 @@
+import { getDependencyContainer } from '@/dependency_container';
+import { BadRequestError, NotFoundError } from '@supaglue/core/errors';
+import * as crypto from 'crypto';
+import type { Request, Response } from 'express';
+import { Router } from 'express';
+
+type HubspotChangedAssociationWebhookPayload = {
+  eventId: string;
+  subscriptionId: string;
+  portalId: string;
+  appId: string;
+  occurredAt: number; // Epoch timestamp
+  subscriptionType: string;
+  attemptNumber: number;
+  changeSource: string;
+  associationType: string;
+  fromObjectId: number;
+  toObjectId: number;
+  associationRemoved: boolean;
+  isPrimaryAssociation: boolean;
+  sourceId?: string;
+};
+
+const { providerService } = getDependencyContainer();
+
+export default function init(app: Router): void {
+  const webhookRouter = Router();
+
+  webhookRouter.post('/_webhook', async (req: Request<HubspotChangedAssociationWebhookPayload>, res: Response) => {
+    const provider = await providerService.findByHubspotAppId(req.body.appId);
+    if (!provider) {
+      throw new NotFoundError(`Provider not found for appId: ${req.body.appId}`);
+    }
+
+    const validated = validateHubSpotSignatureV3(req, provider.config.oauth.credentials.oauthClientSecret);
+    if (!validated) {
+      throw new BadRequestError('Invalid HubSpot signature');
+    }
+    // TODO: Implement dirty flagging of record
+    return res.status(200).end();
+  });
+
+  app.use('/hubspot', webhookRouter);
+}
+
+function validateHubSpotSignatureV3(
+  req: Request<HubspotChangedAssociationWebhookPayload>,
+  clientSecret: string
+): boolean {
+  const signature = req.headers['x-hubspot-signature-v3'] as string;
+  const requestUri = req.protocol + '://' + req.get('host') + req.originalUrl;
+  const requestBody = JSON.stringify(req.body);
+  const timestamp = req.headers['x-hubspot-request-timestamp'] as string;
+  // Check if the timestamp is older than 5 minutes
+  const timestampDiff = Date.now() - parseInt(timestamp);
+  if (timestampDiff > 300000) {
+    // 5 minutes in milliseconds
+    return false;
+  }
+
+  // Decode URL-encoded characters in requestUri
+  const decodedUri = decodeURIComponent(requestUri);
+
+  // The request method is always POST
+  const requestMethod = 'POST';
+
+  // Concatenate requestMethod, requestUri, requestBody, and timestamp
+  const message = requestMethod + decodedUri + requestBody + timestamp;
+
+  // Create HMAC SHA-256 hash
+  const hmac = crypto.createHmac('sha256', clientSecret);
+  hmac.update(message);
+  const hash = hmac.digest('base64');
+
+  // Compare the hash with the signature
+  return crypto.timingSafeEqual(Buffer.from(hash), Buffer.from(signature));
+}

apps/api/routes/internal/index.ts

@@ -10,6 +10,7 @@ import customer from './customer';
 import destination from './destination';
 import entity from './entity';
 import entityMapping from './entity_mapping';
+import hubspot from './hubspot';
 import link from './link';
 import magicLink from './magic_link';
 import metadata from './metadata';
@@ -23,6 +24,10 @@ import system from './system';
 import _multitenantBackfill from './_multitenant_backfill';
 
 export default function init(app: Router): void {
+  const noMiddleware = Router();
+  hubspot(noMiddleware);
+  app.use('/internal', noMiddleware);
+
   // internal routes should require only internal middleware
   const internalRouter = Router();
   internalRouter.use(internalMiddleware);

docker-compose.yml

@@ -24,7 +24,8 @@ x-common-env: &common-env
   CLERK_JWKS_URL: ${CLERK_JWKS_URL:-https://witty-eft-29.clerk.accounts.dev/.well-known/jwks.json}
   # For debugging only
   GLOBAL_AGENT_HTTP_PROXY: ${GLOBAL_AGENT_HTTP_PROXY}
-  NODE_TLS_REJECT_UNAUTHORIZED: 0 # For https debugging proxy to work.... 
+  NODE_TLS_REJECT_UNAUTHORIZED: 0 # For https debugging proxy to work....
+  HUBSPOT_WEBHOOK_TARGET_URL:
 
 x-fe-api-common-env: &fe-api-common-env
   SUPAGLUE_INTERNAL_TOKEN: some-internal-token

packages/core/dependency_container.ts

@@ -90,7 +90,7 @@ function createCoreDependencyContainer(): CoreDependencyContainer {
   const applicationService = new ApplicationService(prisma);
   const sgUserService = new SgUserService();
   const providerService = new ProviderService(prisma);
-  const syncConfigService = new SyncConfigService(prisma);
+  const syncConfigService = new SyncConfigService(prisma, providerService);
   const schemaService = new SchemaService(prisma);
   const entityService = new EntityService(prisma);
   const customerService = new CustomerService(prisma);

packages/core/lib/hubspot_webhook.ts

@@ -0,0 +1,82 @@
+import { Client as HubspotClient } from '@hubspot/api-client';
+import type { SubscriptionCreateRequestEventTypeEnum } from '@hubspot/api-client/lib/codegen/webhooks';
+import { BadRequestError } from '../errors';
+
+const HUBSPOT_WEBHOOK_TARGET_URL =
+  process.env.HUBSPOT_WEBHOOK_TARGET_URL ?? `${process.env.SUPAGLUE_SERVER_URL}/internal/hubspot/_webhook`;
+
+export const updateWebhookSubscriptions = async (
+  developerApiKey: string,
+  hubspotAppId: number,
+  hubspotStandardObjects: ('contact' | 'company' | 'deal')[]
+) => {
+  const hubspotClient = new HubspotClient({ developerApiKey });
+  const subscriptions = await hubspotClient.webhooks.subscriptionsApi.getAll(hubspotAppId);
+
+  await Promise.all(
+    ['company.associationChange', 'contact.associationChange', 'deal.associationChange'].map(async (eventType) => {
+      const existingSubscription = subscriptions.results?.find((s) => s.eventType === eventType);
+      if (!hubspotStandardObjects.includes(eventType.split('.')[0] as 'contact' | 'company' | 'deal')) {
+        // Delete the subscription if it exists
+        if (!existingSubscription) {
+          return;
+        }
+        await hubspotClient.webhooks.subscriptionsApi.archive(parseInt(existingSubscription.id), hubspotAppId);
+        return;
+      }
+      // Create subscription if it doesn't exist
+      if (!existingSubscription) {
+        return hubspotClient.webhooks.subscriptionsApi.create(hubspotAppId, {
+          active: true,
+          eventType: eventType as SubscriptionCreateRequestEventTypeEnum,
+        });
+      }
+    })
+  );
+};
+
+export const deleteWebhookSubscriptions = async (developerApiKey: string, hubspotAppId: number) => {
+  const hubspotClient = new HubspotClient({ developerApiKey });
+  const subscriptions = await hubspotClient.webhooks.subscriptionsApi.getAll(hubspotAppId);
+  await Promise.all(
+    (subscriptions.results ?? []).map((subscription) =>
+      hubspotClient.webhooks.subscriptionsApi.archive(parseInt(subscription.id), hubspotAppId)
+    )
+  );
+};
+
+const checkWebhookTargetExists = async (hubspotClient: HubspotClient, hubspotAppId: number): Promise<boolean> => {
+  try {
+    const res = await hubspotClient.webhooks.settingsApi.getAll(hubspotAppId);
+    if (res.targetUrl === HUBSPOT_WEBHOOK_TARGET_URL) {
+      return true;
+    }
+  } catch (e: any) {
+    if (e.code === 404) {
+      return false;
+    }
+    throw e;
+  }
+  throw new BadRequestError(
+    `Your Hubspot Developer App already has an existing Webhook target URL. Please delete it first or use a different Developer App.`
+  );
+};
+
+export const createWebhookTargetIfNoneExists = async (developerApiKey: string, hubspotAppId: number) => {
+  const hubspotClient = new HubspotClient({ developerApiKey });
+  const exists = await checkWebhookTargetExists(hubspotClient, hubspotAppId);
+  if (!exists) {
+    await hubspotClient.webhooks.settingsApi.configure(hubspotAppId, {
+      throttling: {
+        period: 'SECONDLY',
+        maxConcurrentRequests: 10,
+      },
+      targetUrl: HUBSPOT_WEBHOOK_TARGET_URL,
+    });
+  }
+};
+
+export const deleteWebhookTargetIfExists = async (developerApiKey: string, hubspotAppId: number) => {
+  const hubspotClient = new HubspotClient({ developerApiKey });
+  await hubspotClient.webhooks.settingsApi.clear(hubspotAppId);
+};

packages/core/mappers/provider.ts

@@ -47,6 +47,7 @@ export async function fromProviderModel<T extends Provider = Provider>({
   config,
   objects,
   entityMappings,
+  hubspotAppId,
 }: ProviderModel): Promise<T> {
   return {
     id,
@@ -62,6 +63,7 @@ export async function fromProviderModel<T extends Provider = Provider>({
         : undefined,
     objects: objects ?? undefined,
     entityMappings: entityMappings ?? undefined,
+    hubspotAppId,
   } as T;
 }
 

packages/core/services/connection_service.ts

@@ -634,13 +634,11 @@ export class ConnectionService {
     const schema = schemaId ? await this.#schemaService.getById(schemaId) : undefined;
     let customerFieldMapping: SchemaMappingsConfigObjectFieldMapping[] | undefined = undefined;
     if (objectType === 'common') {
-      customerFieldMapping = connection.schemaMappingsConfig?.commonObjects?.find(
-        (o) => o.object === objectName
-      )?.fieldMappings;
+      customerFieldMapping = connection.schemaMappingsConfig?.commonObjects?.find((o) => o.object === objectName)
+        ?.fieldMappings;
     } else if (objectType === 'standard') {
-      customerFieldMapping = connection.schemaMappingsConfig?.standardObjects?.find(
-        (o) => o.object === objectName
-      )?.fieldMappings;
+      customerFieldMapping = connection.schemaMappingsConfig?.standardObjects?.find((o) => o.object === objectName)
+        ?.fieldMappings;
     }
     return createFieldMappingConfig(schema?.config, customerFieldMapping);
   }

packages/core/services/provider_service.ts

@@ -15,6 +15,7 @@ import type {
 import type { ProviderEntityMapping } from '@supaglue/types/entity_mapping';
 import { BadRequestError, NotFoundError } from '../errors';
 import { validateEntityOrSchemaFieldName } from '../lib/entity';
+import { createWebhookTargetIfNoneExists, deleteWebhookTargetIfExists } from '../lib/hubspot_webhook';
 import {
   fromCreateParamsToSyncConfigModel,
   fromProviderModel,
@@ -76,6 +77,19 @@ export class ProviderService {
     return fromProviderModel<T>(provider);
   }
 
+  public async findByHubspotAppId(hubspotAppId: string): Promise<OauthProvider | undefined> {
+    const provider = await this.#prisma.provider.findFirst({
+      where: {
+        name: 'hubspot',
+        hubspotAppId,
+      },
+    });
+    if (!provider) {
+      return;
+    }
+    return fromProviderModel<OauthProvider>(provider);
+  }
+
   public async list(applicationId: string): Promise<Provider[]> {
     const providers = await this.#prisma.provider.findMany({ where: { applicationId } });
     return Promise.all(providers.map((provider) => fromProviderModel(provider)));
@@ -102,10 +116,14 @@ export class ProviderService {
       validateEntityMappings(provider.entityMappings);
     }
     if (provider.name === 'hubspot' && provider.config.providerAppId) {
-      await this.validateHubspotWebhookConfig(
+      await this.#validateHubspotWebhookConfig(
         provider.config.providerAppId,
         provider.config.oauth.credentials.developerToken
       );
+      await createWebhookTargetIfNoneExists(
+        provider.config.oauth.credentials.developerToken!,
+        parseInt(provider.config.providerAppId)
+      );
     }
 
     const createdProvider = await this.#prisma.provider.create({
@@ -178,10 +196,14 @@ export class ProviderService {
     }
 
     if (provider.name === 'hubspot' && provider.config.providerAppId) {
-      await this.validateHubspotWebhookConfig(
+      await this.#validateHubspotWebhookConfig(
         provider.config.providerAppId,
         provider.config.oauth.credentials.developerToken
       );
+      await createWebhookTargetIfNoneExists(
+        provider.config.oauth.credentials.developerToken!,
+        parseInt(provider.config.providerAppId)
+      );
     }
 
     const updatedProvider = await this.#prisma.provider.update({
@@ -217,6 +239,7 @@ export class ProviderService {
   }
 
   public async delete(id: string, applicationId: string): Promise<void> {
+    const provider = await this.getByIdAndApplicationId(id, applicationId);
     const syncConfigs = await this.#prisma.syncConfig.findMany({
       where: { providerId: id },
     });
@@ -233,9 +256,21 @@ export class ProviderService {
     await this.#prisma.provider.deleteMany({
       where: { id, applicationId },
     });
+
+    if (provider.name === 'hubspot' && provider.hubspotAppId) {
+      try {
+        await deleteWebhookTargetIfExists(
+          provider.config.oauth.credentials.developerToken!,
+          parseInt(provider.hubspotAppId)
+        );
+      } catch (e: any) {
+        // eslint-disable-next-line no-console
+        console.warn(`Failed to delete webhook target for HubSpot App ID: ${provider.hubspotAppId}`, e);
+      }
+    }
   }
 
-  private async validateHubspotWebhookConfig(appId: string, developerToken?: string): Promise<void> {
+  async #validateHubspotWebhookConfig(appId: string, developerToken?: string): Promise<void> {
     if (!developerToken) {
       throw new BadRequestError('Provider config is missing developerToken');
     }