WorkOS webhook retries without duplicates

// Idempotent worker example: dedupe key + dedupe store (in-memory TTL map)
// Node 18+
import crypto from "node:crypto";

const DEDUPE_WINDOW_MS = 24 * 60 * 60 * 1000;
const dedupeStore = new Map(); // key -> expiresAtMs

function cleanup(nowMs) {
  // Tiny O(n) cleanup for demo purposes; use Redis/DB in production.
  for (const [k, exp] of dedupeStore.entries()) {
    if (exp <= nowMs) dedupeStore.delete(k);
  }
}

function computeDedupeKey(payload) {
  // Prefer a stable provider event id if present; otherwise fall back to a stable hash.
  if (payload && typeof payload === "object") {
    const maybeId = payload.id ?? payload.event_id ?? payload.eventId;
    if (typeof maybeId === "string" && maybeId.length > 0) return maybeId;
  }
  return crypto.createHash("sha256").update(JSON.stringify(payload ?? {})).digest("hex");
}

function shouldProcessOnce(dedupeKey) {
  const now = Date.now();
  cleanup(now);
  const exp = dedupeStore.get(dedupeKey);
  if (typeof exp === "number" && exp > now) return false;
  dedupeStore.set(dedupeKey, now + DEDUPE_WINDOW_MS);
  return true;
}

export async function handleWebhook(payload) {
  const dedupeKey = computeDedupeKey(payload);
  if (!shouldProcessOnce(dedupeKey)) return { deduped: true };

  // TODO: perform side effects here (DB writes, API calls).
  // Tip: enforce exactly-once effects with unique constraints or downstream idempotency keys.
  return { deduped: false, ok: true };
}

// Demo runner: node worker.mjs '{"id":"evt_123","type":"example"}'
if (import.meta.url === `file://${process.argv[1]}`) {
  const payload = JSON.parse(process.argv[2] ?? "{}");
  console.log(await handleWebhook(payload));
}

# Idempotent worker example: dedupe key + dedupe store (in-memory TTL dict)
import hashlib
import json
import sys
import time

DEDUPE_WINDOW_S = 24 * 60 * 60
dedupe_store = {}  # key -> expires_at_epoch_s

def cleanup(now_s: int) -> None:
    for k, exp in list(dedupe_store.items()):
        if exp <= now_s:
            del dedupe_store[k]

def compute_dedupe_key(payload: object) -> str:
    if isinstance(payload, dict):
        for field in ["id", "event_id", "eventId"]:
            v = payload.get(field)
            if isinstance(v, str) and v:
                return v
    raw = json.dumps(payload or {}, sort_keys=True, separators=(",", ":")).encode("utf-8")
    return hashlib.sha256(raw).hexdigest()

def should_process_once(dedupe_key: str) -> bool:
    now = int(time.time())
    cleanup(now)
    exp = dedupe_store.get(dedupe_key)
    if isinstance(exp, int) and exp > now:
        return False
    dedupe_store[dedupe_key] = now + DEDUPE_WINDOW_S
    return True

def handle_webhook(payload: object) -> dict:
    dedupe_key = compute_dedupe_key(payload)
    if not should_process_once(dedupe_key):
        return {"deduped": True}
    # TODO: perform side effects here (DB writes, API calls).
    return {"deduped": False, "ok": True}

if __name__ == "__main__":
    payload = json.loads(sys.argv[1]) if len(sys.argv) > 1 else {}
    print(handle_webhook(payload))

// Hooque consumer loop: GET next, parse X-Hooque-Meta, POST ackUrl/nackUrl/rejectUrl
// Node 18+ (fetch built-in)
const QUEUE_NEXT_URL =
  process.env.HOOQUE_QUEUE_NEXT_URL ??
  "https://app.hooque.io/queues/cons_webhook_events/next";
const TOKEN = process.env.HOOQUE_TOKEN ?? "hq_tok_replace_me";
const headers = { Authorization: `Bearer ${TOKEN}` };

class RetryableError extends Error {}
class PermanentError extends Error {}

async function processPayload(payload) {
  if (!payload || typeof payload !== "object") throw new PermanentError("invalid payload");

  // TODO: enforce idempotency (dedupe key + dedupe store + window).
  // TODO: do side effects with unique constraints / downstream idempotency keys.
}

while (true) {
  const resp = await fetch(QUEUE_NEXT_URL, { headers });
  if (resp.status === 204) break;
  if (!resp.ok) throw new Error(`Hooque next() failed: ${resp.status}`);

  const payload = await resp.json();
  const meta = JSON.parse(resp.headers.get("X-Hooque-Meta") ?? "{}");

  try {
    await processPayload(payload);
    await fetch(meta.ackUrl, { method: "POST", headers });
  } catch (err) {
    const isPermanent = err instanceof PermanentError;
    const url = isPermanent ? meta.rejectUrl : meta.nackUrl;
    const reason = isPermanent ? `permanent: ${err.message}` : `retryable: ${err.message}`;

    await fetch(url, {
      method: "POST",
      headers: { ...headers, "Content-Type": "application/json" },
      body: JSON.stringify({ reason }),
    });
  }
}

# Hooque consumer loop: GET next, parse X-Hooque-Meta, POST ackUrl/nackUrl/rejectUrl
import json
import os
import requests

QUEUE_NEXT_URL = os.getenv(
    "HOOQUE_QUEUE_NEXT_URL",
    "https://app.hooque.io/queues/cons_webhook_events/next",
)
TOKEN = os.getenv("HOOQUE_TOKEN", "hq_tok_replace_me")
headers = {"Authorization": f"Bearer {TOKEN}"}

class RetryableError(Exception):
    pass

class PermanentError(Exception):
    pass

def process_payload(payload: object) -> None:
    if not isinstance(payload, dict):
        raise PermanentError("invalid payload")
    # TODO: enforce idempotency (dedupe key + dedupe store + window).
    # TODO: do side effects with unique constraints / downstream idempotency keys.
    return None

while True:
    resp = requests.get(QUEUE_NEXT_URL, headers=headers, timeout=30)
    if resp.status_code == 204:
        break
    if resp.status_code >= 400:
        raise RuntimeError(f"Hooque next() failed: {resp.status_code} {resp.text}")

    payload = resp.json()
    meta = json.loads(resp.headers.get("X-Hooque-Meta", "{}"))

    try:
        process_payload(payload)
        requests.post(meta["ackUrl"], headers=headers, timeout=30)
    except Exception as err:
        is_permanent = isinstance(err, PermanentError)
        url = meta.get("rejectUrl") if is_permanent else meta.get("nackUrl")
        reason = f"{'permanent' if is_permanent else 'retryable'}: {err}"
        requests.post(
            url,
            headers={**headers, "Content-Type": "application/json"},
            json={"reason": reason},
            timeout=30,
        )