Sync Implement Webhooks

Implementing Merge Sync via Webhooks (Primary)

Webhooks are the production-recommended way to detect Merge sync events. A single endpoint at POST /api/webhooks/merge handles both initial sync completion (after a customer first connects) and subsequent sync events (ongoing updates). Real-time, no polling overhead.

Webhooks are PRIMARY; polling is a fallback. For production reliability, also implement sync-implement-polling as a safety net — webhook delivery can lag, drop, or your endpoint can be briefly unavailable. The two complement each other.

Field-name convention used in this doc. Pseudo-code and JSON snippets show the raw HTTP response shape (snake_case: last_sync_result, modified_at, is_initial_sync). The Merge SDKs auto-convert to camelCase — in Node, model.is_initial_sync becomes model.isInitialSync. Write your code in your SDK's convention.

First activation: self-introduce

I'm the sync-implement-webhooks skill. I'll wire up a single webhook endpoint that handles both initial and subsequent Merge syncs, with HMAC verification and async processing. For production, plan to also run sync-implement-polling as a fallback.

Prerequisites

• MERGE_WEBHOOK_SECRET in .env (from Merge Dashboard → Webhooks)
• Background job queue configured (Celery, Redis Queue, BullMQ, etc.) for async processing
• Webhook URL publicly accessible (use ngrok/cloudflared for local testing)

Before Proceeding

Five pieces of information are needed before generating any code.

If invoked from implementing-sync, these were answered in Step 1 — use that context. Otherwise, gather them now:

• Common models to sync: which Merge common models? (e.g. Employee, Contact, Ticket) — drives the per-model processing in Step 6.
• linked_accounts.initial_sync_complete column: present or missing? Required for the initial-vs-subsequent branching in Step 4. Add via migration if missing.
• Background job system: Celery, Redis Queue, BullMQ, Sidekiq, etc.? Required because the endpoint must return 200 in under 5 seconds — all data fetching runs async. If not found, ask the user before continuing.
• Body-parsing middleware: search for express.json(), bodyParser.json(), framework-default JSON parsing. HMAC verification needs raw bytes — if global JSON middleware is enabled, the webhook route must opt out (e.g. express.raw({ type: '*/*' })).
• Public webhook URL: production URL, staging URL, or local tunnel (ngrok / cloudflared)? Confirm one is available before continuing — without a reachable URL, polling is the only option (see sync-implement-polling).
• Backend Merge SDK installed? (@mergeapi/merge-node-client for Node, MergePythonClient for Python, dev.merge:merge-java-client for JVM, merge-go-client for Go, merge_ruby_client for Ruby, Merge.Client for .NET.) Drives whether examples below use the SDK or raw HTTP.

Step 1: Register the webhook in Merge Dashboard

Go to https://app.merge.dev/configuration/webhooks → Add webhook. Point it to POST /api/webhooks/merge on your server. Subscribe to Linked Account synced events.

For production, also subscribe to LinkedAccount.sync_completed (recommended) — one event per sync covers all models in one payload. Alternatively, subscribe to {WebhookModel}.synced events for per-model granularity if you need progressive processing.

The model name in the event string is not always the Common Model name. CRM, File Storage, Knowledge Base, and Marketing carry an internal prefix on the wire — CRMAccount.synced, FileStorageFile.synced, KnowledgeBaseArticle.synced, MKTGCampaign.synced. HRIS, ATS, Accounting, and Ticketing are mostly unprefixed (Employee.synced, Candidate.synced, Invoice.synced, Ticket.synced), with exceptions such as TicketingContact and AccountingTransaction. The verbatim list per category is in /merge-unified:onboarding → references/webhooks.md.

Tunnel-hostname rotation warning: cloudflared quick mode and ngrok free tier rotate hostnames on every restart, breaking the registered emitter URL. Use named/reserved tunnels for repeated dev work, or expect to re-register on each restart.

Step 2: Webhook payload schema

Every Merge webhook delivers this shape:

For sync events, data.sync_status is keyed by model ID. Each entry has last_sync_finished (Merge's timestamp), last_sync_result (any of DONE, PARTIALLY_SYNCED, FAILED, SYNCING, DISABLED, PAUSED), data_fresh_as_of, and sync_status_reason:

{
  "hook": { "event": "LinkedAccount.sync_completed" },
  "linked_account": {
    "id": "merge-uuid",
    "end_user_origin_id": "your_user_id",
    "category": "crm"
  },
  "data": {
    "is_initial_sync": false,
    "integration_name": "Salesforce",
    "integration_id": "salesforce",
    "sync_status": {
      "crm.Contact": {
        "last_sync_finished": "2024-01-15T22:46:41Z",
        "last_sync_result": "DONE",
        "data_fresh_as_of": "2024-01-15T22:30:00Z",
        "sync_status_reason": null
      },
      "crm.Account": {
        "last_sync_finished": "2024-01-15T22:46:40Z",
        "last_sync_result": "DONE",
        "data_fresh_as_of": "2024-01-15T22:30:00Z",
        "sync_status_reason": null
      }
    }
  }
}

Step 3: Build the endpoint — HMAC verify, queue, return 200 immediately

This endpoint is security-critical. Follow these three rules exactly:

1. Verify HMAC-SHA256 signature FIRST — before parsing the request body.
2. Queue the raw payload for async processing.
3. Return 200 OK immediately — Merge times out after 10 seconds. Retries are 1 initial attempt + 2 retries backing off 1s then 2s, and they fire only on 5xx, a connection error, or a timeout. A 4xx from your endpoint drops the event permanently with no retry. Aim to ACK in under 5 seconds.

⚠️ Three attempts inside a few seconds is not delivery insurance. A deploy window, or a framework that answers 400/422 on a payload it can't parse, loses the event outright. Keep a reconciliation poll (GET /sync-status or a modified_after sweep) as the backstop — see /merge-unified:sync-implement-polling.

Order matters: read the raw body before JSON parsing. If your framework auto-parses (e.g., Express with app.use(express.json())), use express.raw({ type: '*/*' }) on the webhook route so you can compute HMAC against the raw Buffer.

@app.post("/api/webhooks/merge")
def merge_webhook():
    signature = request.headers.get("X-Merge-Webhook-Signature")
    if not signature:
        logger.warning("Merge webhook: missing signature")
        return {"error": "Missing signature"}, 401

    raw_body = request.get_data()  # raw bytes — do NOT call request.json() first
    secret = os.environ.get("MERGE_WEBHOOK_SECRET")
    if not secret:
        return {"error": "Webhook secret not configured"}, 500

    digest = hmac.new(secret.encode(), raw_body, hashlib.sha256).digest()
    expected = base64.urlsafe_b64encode(digest).decode().rstrip("=")
    # Use compare_digest — never plain == (prevents timing attacks)
    if not hmac.compare_digest(expected, signature.rstrip("=")):
        logger.warning("Merge webhook: invalid signature")
        return {"error": "Invalid signature"}, 401

    process_merge_webhook.delay(request.get_json())  # queue async job
    return {}, 200

Step 4: Background job — branch on initial vs. subsequent

A single job handles both phases. The initial_sync_complete flag on linked_accounts tells you which path to take.

def process_merge_webhook(payload: dict):
    origin_id = payload["linked_account"]["end_user_origin_id"]
    account = db.query(LinkedAccount).filter_by(end_user_origin_id=origin_id).first()
    if not account:
        return  # already returned 200 at endpoint; no retry needed

    if not account.initial_sync_complete:
        # Initial sync path
        account.initial_sync_complete = True
        db.commit()
        fetch_initial_data(account)
    else:
        # Subsequent sync path — incremental fetch with bounded window
        process_subsequent_sync(account, payload)

Step 5: Subsequent sync — the two timestamps

For subsequent syncs you need a sync_state table to track per-model timestamps.

SQL migration:

CREATE TABLE IF NOT EXISTS sync_state (
  id                        SERIAL PRIMARY KEY,
  linked_account_id         INTEGER NOT NULL REFERENCES linked_accounts(id) ON DELETE CASCADE,
  model_id                  TEXT NOT NULL,             -- e.g. "hris.Employee"
  last_synced_at            TIMESTAMPTZ,               -- YOUR timestamp — when you STARTED the last fetch
  merge_last_sync_finished  TIMESTAMPTZ,               -- Merge's timestamp from the webhook
  last_fetched_at           TIMESTAMPTZ,               -- when your fetch completed
  status                    TEXT,                       -- DONE | PARTIALLY_SYNCED | SYNCING | FAILED
  UNIQUE (linked_account_id, model_id)
);

SQLAlchemy equivalent:

class SyncState(db.Model):
    __tablename__ = "sync_state"
    id = db.Column(db.Integer, primary_key=True)
    linked_account_id = db.Column(db.Integer, db.ForeignKey("linked_accounts.id"), nullable=False)
    model_id = db.Column(db.String(100), nullable=False)
    last_synced_at = db.Column(db.DateTime, nullable=True)
    merge_last_sync_finished = db.Column(db.DateTime, nullable=True)
    last_fetched_at = db.Column(db.DateTime, nullable=True)
    status = db.Column(db.String(20), nullable=True)
    __table_args__ = (db.UniqueConstraint("linked_account_id", "model_id"),)

Why start time: Overlap is safer than gaps — records modified during your fetch window are captured twice, never missed.

Step 6: Per-model processing in the subsequent path

def process_subsequent_sync(account, payload: dict):
    sync_status = payload["data"]["sync_status"]   # dict keyed by model_id

    for model_id, model_data in sync_status.items():
        last_sync_result = model_data.get("last_sync_result")

        # Accept successful results; skip failures
        if last_sync_result not in ("DONE", "PARTIALLY_SYNCED"):
            continue

        webhook_finished = parse_datetime(model_data["last_sync_finished"])

        stored = db.query(SyncState).filter_by(
            linked_account_id=account.id, model_id=model_id
        ).first()

        # Deduplicate / handle out-of-order webhooks automatically
        if stored and stored.merge_last_sync_finished and webhook_finished <= stored.merge_last_sync_finished:
            continue

        last_synced_at = datetime.utcnow()  # record BEFORE the fetch

        params = {"modified_before": webhook_finished.isoformat()}
        if stored and stored.last_synced_at:
            params["modified_after"] = stored.last_synced_at.isoformat()
        # No stored.last_synced_at → first subsequent sync; omit modified_after

        merge = Merge(api_key=os.environ["MERGE_API_KEY"], account_token=account.account_token)
        category, model_name = model_id.split(".")  # e.g., "crm.Contact" → "crm", "Contact"
        list_fn = getattr(getattr(merge, category), model_name.lower() + "s")  # merge.crm.contacts
        results = fetch_all_pages(list_fn, params)
        upsert_records(model_id, results)
        upsert_sync_state(account.id, model_id, last_synced_at, webhook_finished)

Critical Differences: Initial vs. Subsequent

Security requirements (critical)

• Verify signature before parsing JSON — work on raw bytes only.
• Use hmac.compare_digest() — never plain == (timing-attack vulnerability).
• Never log MERGE_WEBHOOK_SECRET.
• Log all signature failures for monitoring and alerting.
• Return 200 on missing account (already handled here) — prevents Merge from retrying on bad data.

HTTP response codes

• 200: Successfully received — Merge won't retry.
• 401: Invalid/missing signature — Merge won't retry.
• 5xx: Temporary failure — Merge retries up to 2 more times.

Never return other 4xx codes.

Critical gotchas

• 30-second timeout: return 200 at the endpoint immediately; ALL data fetching is in the background job.
• Deduplication is automatic: webhook_finished <= stored.merge_last_sync_finished handles both duplicates and out-of-order webhooks — no extra logic needed.
• First subsequent sync: stored.last_synced_at is null — omit modified_after, include only modified_before.
• Record last_synced_at before the fetch, not after — ensures no gap if records are modified during the fetch window.
• Add polling as a fallback (sync-implement-polling): webhook delivery occasionally lags or drops. Polling on a slow cadence catches what webhooks miss.

Testing checklist

• Endpoint accepts POST at /api/webhooks/merge (or your chosen path)
• HMAC-SHA256 signature verification implemented; uses compare_digest
• Returns 200 OK within 5 seconds (well under Merge's 10s timeout)
• Webhook processing is asynchronous (background job)
• Initial path: sets initial_sync_complete = true and triggers full fetch
• Subsequent path: extracts last_sync_finished per model from data.sync_status
• Subsequent path: skips webhooks where webhook_finished <= stored.merge_last_sync_finished
• Subsequent path: records last_synced_at before fetch
• Subsequent path: uses correct bounded window (modified_after + modified_before)
• Subsequent path: stores both timestamps after successful fetch
• Skips FAILED models; accepts DONE for initial, DONE + PARTIALLY_SYNCED for subsequent
• Handles duplicate webhooks idempotently
• Handles out-of-order webhooks gracefully
• First subsequent sync works when last_synced_at is null
• Signature failures return 401 and are logged

Troubleshooting

SYMPTOM: HMAC signature validation fails for all webhook events CAUSE: Webhook secret mismatch or body was parsed before signature check (Express json() middleware consuming raw body) FIX: Use express.raw() on the webhook route; compute HMAC against the raw Buffer before JSON parsing.

SYMPTOM: Webhook endpoint returns 200 but data is never fetched CAUSE: Background job is being enqueued but not processed, or job queue is paused. FIX: Verify your job worker is running; check the job queue dashboard for stuck jobs.

SYMPTOM: Incremental fetch returns records already processed CAUSE: last_synced_at timestamp is not being saved after each successful fetch. FIX: Persist last_synced_at and merge_last_sync_finished only after a successful fetch; never update on failure.

SYMPTOM: Webhook URL changes every dev session CAUSE: cloudflared quick mode or ngrok free tier rotates hostnames on every restart. FIX: Use named/reserved tunnels for repeated dev work; re-register the emitter URL in the Merge dashboard after rotations.