recap-relay/server/webhook-dedup.js

// Persistent dedup of settled-payment webhook deliveries — BTCPay
// invoices AND Zaprite orders share this one store. It replaces the
// per-process in-memory Sets the two webhook handlers used to keep:
// those were cleared on restart, so a duplicate delivery that straddled
// a relay restart would double-credit (BTCPay) or double-extend a
// subscription (Zaprite). Persisting the processed-keys to disk closes
// that window.
//
// JSON-file backed (single file at <dataDir>/processed-webhooks.json),
// same rationale as the credit ledger in credits.js: at most one
// mutation per settled payment, so a plain JSON file with serial writes
// is plenty.
//
// Keys are namespaced by their caller so the two rails can't collide in
// the shared store: BTCPay uses `<storeId>|<invoiceId>`, Zaprite uses
// `zaprite:<orderId>`. We store key → ISO timestamp (not a bare set) so
// entries far older than any processor's retry window can be pruned on
// load, keeping the file bounded.

import fs from "fs/promises";
import path from "path";

// Payment processors retry webhook delivery for hours-to-days, never
// months. 180 days is safely beyond any retry window, so pruning past
// it cannot re-open a double-grant gap.
const RETENTION_MS = 180 * 24 * 60 * 60 * 1000;

let storePath = null;
let processed = {}; // key → ISO timestamp
let writing = null; // serializes concurrent writes

export async function initWebhookDedup({ dataDir } = {}) {
  const dir = dataDir || "/data";
  storePath = path.join(dir, "processed-webhooks.json");
  await fs.mkdir(dir, { recursive: true }).catch(() => {});
  try {
    const raw = await fs.readFile(storePath, "utf8");
    const parsed = JSON.parse(raw);
    processed =
      parsed && typeof parsed === "object" && parsed.keys ? parsed.keys : {};
  } catch (err) {
    if (err.code !== "ENOENT") {
      console.warn(
        `[webhook-dedup] failed to read ${storePath}: ${err.message} — starting empty`
      );
    }
    processed = {};
  }
  // Prune anything older than the retention window so the file stays
  // bounded over the relay's lifetime.
  const cutoff = Date.now() - RETENTION_MS;
  let pruned = 0;
  for (const [k, ts] of Object.entries(processed)) {
    const t = Date.parse(ts);
    if (!Number.isFinite(t) || t < cutoff) {
      delete processed[k];
      pruned += 1;
    }
  }
  if (pruned > 0) await persist();
  console.log(
    `[webhook-dedup] loaded ${Object.keys(processed).length} processed key(s)` +
      `${pruned ? `, pruned ${pruned} stale` : ""} from ${storePath}`
  );
}

async function persist() {
  // Coalesce concurrent writes — mirrors credits.js persist().
  if (writing) await writing;
  writing = (async () => {
    const tmp = storePath + ".tmp";
    await fs.writeFile(tmp, JSON.stringify({ keys: processed }), { mode: 0o600 });
    await fs.rename(tmp, storePath);
  })();
  try {
    await writing;
  } finally {
    writing = null;
  }
}

export function isWebhookProcessed(key) {
  return Object.prototype.hasOwnProperty.call(processed, key);
}

// Mark a webhook key processed and persist immediately. Callers invoke
// this AFTER the grant/extend succeeds, so a failed grant leaves the key
// unmarked and a processor retry can complete it. Idempotent.
export async function markWebhookProcessed(key) {
  if (isWebhookProcessed(key)) return;
  processed[key] = new Date().toISOString();
  await persist();
}