Hubspot Product Event Sync

interface HubSpot207Response {
  status: "COMPLETE" | "PENDING" | "PROCESSING" | "CANCELED";
  results: Array<{ id: string; properties: Record<string, string> }>;
  errors: Array<{
    status: string;
    category: string;
    message: string;
    context: { id?: string[] };
  }>;
  startedAt: string;
  completedAt: string;
}

async function handle207(
  res: Response,
  events: ProductEvent[],
  dlq: DeadLetterQueue,
): Promise<void> {
  // 200 = all succeeded; 207 = partial; 400+ = all failed
  if (res.status === 200) return;

  if (res.status === 207) {
    const body: HubSpot207Response = await res.json();

    // Log successes (optional — useful for counters)
    const successIds = new Set(body.results.map((r) => r.id));

    // Route failures to DLQ with full context for replay
    for (const err of body.errors) {
      const failedId = err.context.id?.[0];
      const failedEvent = events.find((e) => e.email === failedId);
      console.error("HubSpot batch partial failure", {
        error: err.message,
        category: err.category,
        contactId: failedId,
        email: failedEvent?.email,
      });
      if (failedEvent) {
        await dlq.push({
          event: failedEvent,
          error: err,
          failedAt: new Date().toISOString(),
          retryCount: 0,
        });
      }
    }

    return;
  }

  // Non-2xx — entire batch failed
  const body = await res.text();
  console.error("HubSpot batch entirely failed", { status: res.status, body });

  // Route all events to DLQ
  for (const event of events) {
    await dlq.push({ event, error: { message: body, status: res.status }, failedAt: new Date().toISOString(), retryCount: 0 });
  }
}

async function upsertContactByEmail(
  token: string,
  email: string,
  properties: Record<string, string>,
): Promise<string> {
  // Try to find by email first
  const search = await fetch("https://api.hubapi.com/crm/v3/objects/contacts/search", {
    method: "POST",
    headers: { Authorization: `Bearer ${token}`, "Content-Type": "application/json" },
    body: JSON.stringify({
      filterGroups: [{
        filters: [{ propertyName: "email", operator: "EQ", value: email }],
      }],
      properties: ["email"],
      limit: 1,
    }),
  });

  const searchResult = await search.json();

  if (searchResult.total > 0) {
    // Exists — update
    const contactId = searchResult.results[0].id;
    await fetch(`https://api.hubapi.com/crm/v3/objects/contacts/${contactId}`, {
      method: "PATCH",
      headers: { Authorization: `Bearer ${token}`, "Content-Type": "application/json" },
      body: JSON.stringify({ properties }),
    });
    return contactId;
  }

  // Does not exist — create with product properties included
  const create = await fetch("https://api.hubapi.com/crm/v3/objects/contacts", {
    method: "POST",
    headers: { Authorization: `Bearer ${token}`, "Content-Type": "application/json" },
    body: JSON.stringify({ properties: { email, ...properties } }),
  });
  const newContact = await create.json();
  return newContact.id;
}

async function withRetry<T>(
  fn: () => Promise<Response>,
  maxAttempts = 4,
  baseDelayMs = 1_000,
): Promise<Response> {
  for (let attempt = 1; attempt <= maxAttempts; attempt++) {
    const res = await fn();

    if (res.status !== 429 && res.status < 500) return res;
    if (attempt === maxAttempts) return res;

    const retryAfterSec = parseInt(res.headers.get("Retry-After") ?? "0", 10);
    const expDelay = Math.min(30_000, baseDelayMs * 2 ** attempt);
    const jittered = Math.random() * expDelay;
    const delay = retryAfterSec > 0 ? retryAfterSec * 1_000 : jittered;

    console.warn("HubSpot rate limit or server error", {
      status: res.status,
      attempt,
      delayMs: delay,
      dailyRemaining: res.headers.get("X-HubSpot-RateLimit-Daily-Remaining"),
    });

    await new Promise((r) => setTimeout(r, delay));
  }

  throw new Error("unreachable");
}

// Run once at service startup
const PRODUCT_PROPERTIES: PropertyDefinition[] = [
  {
    name: "hs_product_last_active_date",
    label: "Last Active Date (Product)",
    type: "datetime",
    fieldType: "date",
    groupName: "product_signals",
    description: "Last timestamp the contact triggered a product event",
  },
  {
    name: "hs_product_total_sessions",
    label: "Total Sessions (Product)",
    type: "number",
    fieldType: "number",
    groupName: "product_signals",
    description: "Lifetime session count from product backend",
  },
  {
    name: "hs_product_current_plan",
    label: "Current Plan (Product)",
    type: "enumeration",
    fieldType: "select",
    groupName: "product_signals",
    description: "Plan tier from billing system",
  },
];

for (const prop of PRODUCT_PROPERTIES) {
  await ensureProperty(token, "contacts", prop);
}

const event: ProductEvent = {
  eventId: "evt_01HZ9...",
  email: "alice@acme.com",
  properties: {
    hs_product_last_active_date: Date.now(),
    hs_product_total_sessions: 47,
    hs_product_current_plan: "pro",
  },
  occurredAt: Date.now(),
};

queue.push(event);

# Example: replay from a Postgres DLQ table
psql $DATABASE_URL -c "
  SELECT id, event_json FROM hubspot_dlq
  WHERE retry_count < 3
    AND next_retry_at < NOW()
  LIMIT 500
" --csv | python3 scripts/replay-dlq.py

# Check the contact after a sync
curl -s \
  "https://api.hubapi.com/crm/v3/objects/contacts/search" \
  -H "Authorization: Bearer $HUBSPOT_ACCESS_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "filterGroups": [{"filters": [{"propertyName": "email", "operator": "EQ", "value": "alice@acme.com"}]}],
    "properties": ["email", "hs_product_last_active_date", "hs_product_total_sessions", "hs_product_current_plan"]
  }' | jq '.results[0].properties'

curl -si \
  "https://api.hubapi.com/crm/v3/objects/contacts?limit=1" \
  -H "Authorization: Bearer $HUBSPOT_ACCESS_TOKEN" \
  | grep -i "x-hubspot-ratelimit"