Hubspot Webhook Handlers

import { Worker, type Job } from "bullmq";

// Dead-letter handler — called when a job exhausts all attempts
async function onJobFailed(job: Job | undefined, err: Error): Promise<void> {
  if (!job) return;
  if (job.attemptsMade < (job.opts.attempts ?? 1)) return; // still has retries remaining

  // This job is permanently dead — route to DLQ
  const dlqKey = `hubspot:dlq:${job.name}`;
  await redis.lpush(
    dlqKey,
    JSON.stringify({
      jobId: job.id,
      eventId: job.data.eventId,
      portalId: job.data.portalId,
      objectId: job.data.objectId,
      subscriptionType: job.data.subscriptionType,
      occurredAt: job.data.occurredAt,
      failedAt: Date.now(),
      error: err.message,
      attempts: job.attemptsMade,
    }),
  );

  // Alert — use your notification channel
  console.error("HubSpot event permanently failed", {
    jobId: job.id,
    eventId: job.data.eventId,
    type: job.data.subscriptionType,
    error: err.message,
  });
}

const worker = new Worker(
  "hubspot-events",
  async (job) => {
    await processEventIfNew(redis, job.data, routeEvent);
  },
  {
    connection: { host: process.env.REDIS_HOST, port: parseInt(process.env.REDIS_PORT ?? "6379") },
    concurrency: 10, // process up to 10 events in parallel
  },
);

worker.on("failed", onJobFailed);

async function replayDeadLetterQueue(
  redis: Redis,
  subscriptionType: string,
  limit = 100,
): Promise<number> {
  const dlqKey = `hubspot:dlq:${subscriptionType}`;
  let replayed = 0;

  for (let i = 0; i < limit; i++) {
    const raw = await redis.rpop(dlqKey);
    if (!raw) break;

    const dead = JSON.parse(raw) as { eventId: number; [key: string]: unknown };

    // Delete the dedup key so the replayed event is treated as new
    await redis.del(`hubspot:event:${dead.eventId}`);

    await eventQueue.add(subscriptionType, dead, {
      jobId: `hubspot-replay-${dead.eventId}-${Date.now()}`,
    });

    replayed++;
  }

  console.info(`Replayed ${replayed} dead-letter events for ${subscriptionType}`);
  return replayed;
}

type PropertyVersion = { value: string; occurredAt: number };

async function applyPropertyChange(
  redis: Redis,
  objectId: number,
  propertyName: string,
  newValue: string,
  occurredAt: number,
): Promise<void> {
  const versionKey = `hubspot:prop:${objectId}:${propertyName}`;

  // Read the last-applied version
  const existingRaw = await redis.get(versionKey);
  const existing: PropertyVersion | null = existingRaw ? JSON.parse(existingRaw) : null;

  if (existing && existing.occurredAt >= occurredAt) {
    console.debug("Ignoring stale property change", {
      objectId,
      propertyName,
      existingTimestamp: existing.occurredAt,
      incomingTimestamp: occurredAt,
    });
    return;
  }

  // Safe to apply — this is the newest value seen for this property
  await redis.set(
    versionKey,
    JSON.stringify({ value: newValue, occurredAt }),
    "EX",
    7 * 86_400, // 7-day TTL — keep long enough to guard burst redeliveries
  );

  // Now apply the change to your downstream system
  await updateContactProperty(objectId, propertyName, newValue);
}

# Register a contact.propertyChange subscription
curl -X POST "https://api.hubapi.com/webhooks/v3/{appId}/subscriptions" \
  -H "Authorization: Bearer $HUBSPOT_ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "eventType": "contact.propertyChange",
    "propertyName": "lifecyclestage",
    "active": true
  }'

# Register a list-membership subscription (separate subscription required)
# Note: HubSpot does not expose a direct list.membershipChange event type.
# Use contact.propertyChange for hs_all_contact_vids or workflow enrollment events,
# or poll GET /crm/v3/lists/{listId}/memberships for list-driven automation.

curl -s "https://api.hubapi.com/webhooks/v3/{appId}/subscriptions" \
  -H "Authorization: Bearer $HUBSPOT_ACCESS_TOKEN" | jq '.results[] | {id, eventType, propertyName, active}'

import express from "express";
import { createClient } from "ioredis";
import { verifyHubSpotSignature } from "./middleware/signature";
import { webhookHandler } from "./handlers/webhook";

const app = express();
const redis = createClient({ host: process.env.REDIS_HOST });

// IMPORTANT: raw body capture before JSON parsing
app.use(
  "/webhooks/hubspot",
  express.raw({ type: "application/json", limit: "1mb" }),
  (req, _res, next) => {
    (req as any).rawBody = req.body as Buffer;
    next();
  },
  verifyHubSpotSignature,
  webhookHandler,
);

app.listen(3000, () => console.log("Webhook listener on :3000"));

# Compute the expected signature for a test payload
CLIENT_SECRET="your-client-secret"
METHOD="POST"
URI="https://your-host.example.com/webhooks/hubspot"
BODY='[{"eventId":1,"subscriptionType":"contact.propertyChange"}]'
TIMESTAMP="$(date +%s%3N)"  # milliseconds

EXPECTED=$(echo -n "${METHOD}${URI}${BODY}${TIMESTAMP}" | \
  openssl dgst -sha256 -hmac "$CLIENT_SECRET" | awk '{print $2}')

curl -X POST "$URI" \
  -H "Content-Type: application/json" \
  -H "X-HubSpot-Signature-v3: $EXPECTED" \
  -H "X-HubSpot-Request-Timestamp: $TIMESTAMP" \
  -d "$BODY"

# View all entries in the DLQ for contact.propertyChange
redis-cli LRANGE hubspot:dlq:contact.propertyChange 0 -1 | python3 -m json.tool

# Count DLQ depth per event type
redis-cli KEYS "hubspot:dlq:*" | xargs -I{} sh -c 'echo "{}: $(redis-cli LLEN {})"'

# Replay up to 50 failed contact.deletion events
node -e "
const { replayDeadLetterQueue } = require('./dist/dlq');
replayDeadLetterQueue(redis, 'contact.deletion', 50).then(console.log);
"

APP_ID="your-app-id"
TOKEN="$HUBSPOT_ACCESS_TOKEN"
BASE="https://api.hubapi.com/webhooks/v3/$APP_ID/subscriptions"

for EVENT_TYPE in contact.creation contact.deletion contact.propertyChange \
                  contact.merge company.creation deal.creation deal.propertyChange; do
  curl -s -X POST "$BASE" \
    -H "Authorization: Bearer $TOKEN" \
    -H "Content-Type: application/json" \
    -d "{\"eventType\": \"$EVENT_TYPE\", \"active\": true}" | jq '{id, eventType, active}'
done