让 HubSpot 同步真正可靠：事务性 Outbox、QStash 与可修复的一致性

CREATE TABLE subscription_requests (
  id uuid PRIMARY KEY,
  brand_id text NOT NULL,
  contact_key text NOT NULL,
  product_id text NOT NULL,
  desired_state text NOT NULL CHECK (desired_state IN ('subscribed', 'unsubscribed')),
  mapping_version integer NOT NULL,
  idempotency_key text NOT NULL,
  request_hash text NOT NULL,
  sync_status text NOT NULL DEFAULT 'pending',
  created_at timestamptz NOT NULL DEFAULT now(),
  UNIQUE (brand_id, product_id, idempotency_key)
);

CREATE TABLE hubspot_outbox_events (
  id uuid PRIMARY KEY,
  request_id uuid NOT NULL REFERENCES subscription_requests(id),
  event_type text NOT NULL,
  payload jsonb NOT NULL,
  status text NOT NULL DEFAULT 'pending',
  publish_attempts integer NOT NULL DEFAULT 0,
  process_attempts integer NOT NULL DEFAULT 0,
  next_attempt_at timestamptz NOT NULL DEFAULT now(),
  lease_expires_at timestamptz,
  qstash_message_id text,
  last_error_code text,
  last_error_summary text,
  created_at timestamptz NOT NULL DEFAULT now(),
  completed_at timestamptz,
  UNIQUE (request_id, event_type)
);

WITH candidates AS (
  SELECT id
  FROM hubspot_outbox_events
  WHERE (status IN ('pending', 'publish_retry') AND next_attempt_at <= now())
     OR (status = 'publishing' AND lease_expires_at < now())
  ORDER BY next_attempt_at, created_at
  FOR UPDATE SKIP LOCKED
  LIMIT 100
)
UPDATE hubspot_outbox_events AS event
SET status = 'publishing',
    publish_attempts = publish_attempts + 1,
    lease_expires_at = now() + interval '2 minutes'
FROM candidates
WHERE event.id = candidates.id
RETURNING event.*;

const result = await qstash.publishJSON({
  url: process.env.HUBSPOT_WORKER_URL!,
  body: { eventId: event.id },
  retries: 5,
  deduplicationId: event.id,
  flowControl: {
    key: 'hubspot-private-app-primary',
    rate: 150,
    period: '10s',
    parallelism: 12,
  },
});

const plan = await buildHubSpotPlan(event);
const estimatedCalls = estimateHubSpotCalls(plan);
const reservation = await hubspotBudget.reserve({
  key: 'portal:12345:private-app:primary',
  tokens: estimatedCalls,
  limit: 150,
  windowMs: 10_000,
});

if (!reservation.granted) {
  return new Response('HubSpot budget unavailable', {
    status: 503,
    headers: { 'Retry-After': String(reservation.retryAfterSeconds) },
  });
}

import { verifySignatureAppRouter } from '@upstash/qstash/nextjs';

async function handler(request: Request) {
  const { eventId } = await request.json() as { eventId: string };
  const event = await claimEvent(eventId);

  if (!event || event.status === 'synced') {
    return new Response('OK');
  }

  try {
    const plan = await buildPlanFromSnapshot(event);
    await applyHubSpotPlan(event, plan);
    await markEventSynced(event.id);
    return new Response('OK');
  } catch (error) {
    return handleWorkerError(event, error);
  }
}

export const POST = verifySignatureAppRouter(handler);

UPDATE contact_subscription_state
SET applied_sequence = $new_sequence,
    desired_state = $desired_state,
    updated_at = now()
WHERE contact_key = $contact_key
  AND brand_id = $brand_id
  AND product_id = $product_id
  AND applied_sequence < $new_sequence
RETURNING applied_sequence;

export async function GET(request: Request) {
  if (request.headers.get('authorization') !== `Bearer ${process.env.CRON_SECRET}`) {
    return new Response('Unauthorized', { status: 401 });
  }

  const batch = await findReconciliationCandidates({
    limit: 250,
    pendingOlderThanMinutes: 10,
    expiredLeases: true,
    retryableBefore: new Date(),
  });

  const result = await scheduleCandidatesIdempotently(batch);
  return Response.json({ checked: batch.length, scheduled: result.length });
}