Webhooks overview

// webhook-receiver.mjs (Node.js 18+). Install: npm install express pg
// Run: MG_CLIENT_SECRET=... DATABASE_URL=postgres://... node webhook-receiver.mjs
import crypto from 'node:crypto';
import express from 'express';
import pg from 'pg';

const CLIENT_SECRET = process.env.MG_CLIENT_SECRET; // your client secret signs every webhook
const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });
pool.on('error', (error) => console.error('idle database connection lost', { error: error.message })); // don't crash
const app = express();

// Step 1: read the raw body. The signature covers the exact bytes Multigate sent.
app.post('/webhooks/multigate', express.raw({ type: 'application/json', limit: '1mb' }), async (req, res) => {
  const raw = req.body.toString('utf8');
  let event;
  try {
    event = JSON.parse(raw);
  } catch {
    return res.sendStatus(400);
  }

  // Step 2: verify the signature before trusting anything in the request.
  const receivedHash = event.header?.hash ?? req.get('hash') ?? '';
  if (!isSignatureValid(raw, receivedHash)) {
    console.warn('webhook rejected: invalid signature');
    return res.sendStatus(401);
  }

  const key = req.get('Idempotency-Key');
  if (!key) return res.sendStatus(400);
  // Every documented event has header and body. Anything else is stored and ignored.
  const { notification_type: type = '', notification = '' } = event.header ?? {};
  try {
    // Step 3: store the event once. A retry of the same delivery hits the primary key and is skipped.
    const { rowCount } = await pool.query(
      `INSERT INTO multigate_webhook_events (idempotency_key, notification_type, notification, body)
       VALUES ($1, $2, $3, $4) ON CONFLICT (idempotency_key) DO NOTHING`,
      [key, type, notification, event.body ?? event],
    );
    console.info('webhook received', { key, type, notification, duplicate: rowCount === 0 });

    // Step 4: acknowledge straight away (Multigate waits at most 9 seconds).
    res.sendStatus(200);

    // Step 5: process after replying.
    if (rowCount === 1) setImmediate(() => processEvent(key));
  } catch (error) {
    console.error('webhook not stored; Multigate will retry', { key, error: error.message });
    res.sendStatus(500);
  }
});

function isSignatureValid(raw, receivedHash) {
  // Sign exactly the text after "body": up to the final }.
  const marker = raw.indexOf('"body":');
  const signedPart = marker === -1 ? raw : raw.slice(marker + '"body":'.length, raw.lastIndexOf('}'));
  const expected = Buffer.from(crypto.createHmac('sha256', CLIENT_SECRET).update(signedPart, 'utf8').digest('hex'));
  const given = Buffer.from(String(receivedHash), 'utf8');
  return given.length === expected.length && crypto.timingSafeEqual(given, expected);
}

async function processEvent(key) {
  let db;
  try {
    db = await pool.connect();
    await db.query('BEGIN');
    // Lock the row so only one worker, on any of your servers, processes it.
    const { rows } = await db.query(
      `SELECT notification_type, notification, body FROM multigate_webhook_events
       WHERE idempotency_key = $1 AND processed_at IS NULL FOR UPDATE SKIP LOCKED`,
      [key],
    );
    if (rows.length === 1) {
      const { notification_type: type, notification, body } = rows[0];
      await handleEvent(db, type, notification, body);
      await db.query('UPDATE multigate_webhook_events SET processed_at = now() WHERE idempotency_key = $1', [key]);
    }
    await db.query('COMMIT');
  } catch (error) {
    await db?.query('ROLLBACK').catch(() => {}); // the connection may already be gone
    console.error('webhook processing failed; the sweep will retry it', { key, error: error.message });
  } finally {
    db?.release();
  }
}

// Step 6: handle each event. Run your updates on `db` so they commit together with processed_at.
// Each handler must be safe to run twice: the same outcome can arrive again with a new Idempotency-Key.
async function handleEvent(db, type, notification, body) {
  switch (`${type}/${notification}`) {
    case 'transfer/transfer_completed':
    case 'transfer/transfer_failed':
      // Find your payout by uniqueRequestId. If it isn't final yet, set it to body.status.
      console.info('payout result', { uniqueRequestId: body.uniqueRequestId, status: body.status, reason: body.responseMessage });
      break;
    case 'transfer/refund':
      // Find your refund by uniqueRequestId and set it to body.status (completed or failed).
      console.info('refund result', { uniqueRequestId: body.uniqueRequestId, status: body.status });
      break;
    case 'account/inflow':
      // Credit once per transactionId. amount is a decimal string: never parse it as a float.
      console.info('money received', { transactionId: body.transactionId, account: body.receivingAccountNumber ?? body.accountNumber, amount: body.amount, currency: body.currencyIso3Code });
      break;
    case 'account/account_opened':
      // Save the account numbers in fundingDetails against accountId and clientId.
      console.info('account opened', { accountId: body.accountId, clientId: body.clientId, accountNumbers: (body.fundingDetails ?? []).map((d) => d.accountNumber) });
      break;
    case 'client/subclient_approved':
      // status "active" = approved, "rejected" = declined. Ignore approvalStatus.
      console.info('customer reviewed', { customerId: body.entityId, status: body.status });
      break;
    case 'receivables/inflow':
      // Record the payment once per transactionDetails.transactionId; update the invoice status.
      console.info('invoice payment', { receivableId: body.receivableDetails?.entityId, status: body.receivableDetails?.status, transactionId: body.transactionDetails?.transactionId });
      break;
    case 'trade/trade_completed':
      // Act only on status "completed", and confirm the target account was credited before releasing funds.
      console.info('conversion', { quoteId: body.id, status: body.status });
      break;
    default:
      console.warn('unknown event ignored', { type, notification });
  }
}

// Safety net: every minute, process events a crash or an error left unprocessed.
setInterval(async () => {
  try {
    const { rows } = await pool.query(
      `SELECT idempotency_key FROM multigate_webhook_events
       WHERE processed_at IS NULL AND received_at < now() - interval '1 minute'
       ORDER BY received_at LIMIT 100`,
    );
    for (const { idempotency_key: key } of rows) await processEvent(key);
  } catch (error) {
    console.error('webhook sweep failed', { error: error.message });
  }
}, 60_000);

app.listen(3000, () => console.info('Listening on http://localhost:3000/webhooks/multigate'));

# webhook_receiver.py (Python 3.9+). Install: pip install flask "psycopg[binary,pool]"
# Run: MG_CLIENT_SECRET=... DATABASE_URL=postgresql://... flask --app webhook_receiver run --port 3000
import hashlib
import hmac
import json
import logging
import os
import threading
import time
from concurrent.futures import ThreadPoolExecutor

from flask import Flask, request
from psycopg.types.json import Jsonb
from psycopg_pool import ConnectionPool

CLIENT_SECRET = os.environ["MG_CLIENT_SECRET"].encode("utf-8")  # your client secret signs every webhook
pool = ConnectionPool(os.environ["DATABASE_URL"], open=True)
workers = ThreadPoolExecutor(max_workers=4)
app = Flask(__name__)
logging.basicConfig(level=logging.INFO)
log = logging.getLogger("multigate-webhooks")


@app.post("/webhooks/multigate")
def receive_webhook():
    # Step 1: read the raw body. The signature covers the exact bytes Multigate sent.
    raw = request.get_data(as_text=True)
    try:
        event = json.loads(raw)
    except ValueError:
        return "", 400

    # Step 2: verify the signature before trusting anything in the request.
    received_hash = (event.get("header") or {}).get("hash") or request.headers.get("hash", "")
    if not signature_is_valid(raw, received_hash):
        log.warning("webhook rejected: invalid signature")
        return "", 401

    key = request.headers.get("Idempotency-Key")
    if not key:
        return "", 400
    # Every documented event has header and body. Anything else is stored and ignored.
    header = event.get("header") or {}
    notification_type, notification = header.get("notification_type", ""), header.get("notification", "")

    # Step 3: store the event once. A retry of the same delivery hits the primary key and is skipped.
    # If this raises, Flask answers 500 and Multigate retries.
    with pool.connection() as conn:
        stored = conn.execute(
            "INSERT INTO multigate_webhook_events (idempotency_key, notification_type, notification, body) "
            "VALUES (%s, %s, %s, %s) ON CONFLICT (idempotency_key) DO NOTHING",
            (key, notification_type, notification, Jsonb(event.get("body", event))),
        ).rowcount == 1
    log.info("webhook received key=%s event=%s/%s duplicate=%s", key, notification_type, notification, not stored)

    # Steps 4 and 5: hand processing to a background thread, then acknowledge straight away
    # (Multigate waits at most 9 seconds).
    if stored:
        workers.submit(process_event, key)
    return "", 200


def signature_is_valid(raw, received_hash):
    # Sign exactly the text after "body": up to the final }.
    marker = raw.find('"body":')
    signed_part = raw if marker == -1 else raw[marker + len('"body":'):raw.rfind("}")]
    expected = hmac.new(CLIENT_SECRET, signed_part.encode("utf-8"), hashlib.sha256).hexdigest()
    return hmac.compare_digest(str(received_hash).encode("utf-8"), expected.encode("utf-8"))


def process_event(key):
    try:
        with pool.connection() as conn, conn.transaction():
            # Lock the row so only one worker, on any of your servers, processes it.
            row = conn.execute(
                "SELECT notification_type, notification, body FROM multigate_webhook_events "
                "WHERE idempotency_key = %s AND processed_at IS NULL FOR UPDATE SKIP LOCKED",
                (key,),
            ).fetchone()
            if row is None:
                return
            handle_event(conn, *row)
            conn.execute("UPDATE multigate_webhook_events SET processed_at = now() WHERE idempotency_key = %s", (key,))
    except Exception:
        log.exception("webhook processing failed key=%s; the sweep will retry it", key)


# Step 6: handle each event. Run your updates on `conn` so they commit together with processed_at.
# Each handler must be safe to run twice: the same outcome can arrive again with a new Idempotency-Key.
def handle_event(conn, notification_type, notification, body):
    event = f"{notification_type}/{notification}"
    if event in ("transfer/transfer_completed", "transfer/transfer_failed"):
        # Find your payout by uniqueRequestId. If it isn't final yet, set it to body["status"].
        log.info("payout result %s %s %s", body["uniqueRequestId"], body["status"], body.get("responseMessage"))
    elif event == "transfer/refund":
        # Find your refund by uniqueRequestId and set it to body["status"] (completed or failed).
        log.info("refund result %s %s", body["uniqueRequestId"], body["status"])
    elif event == "account/inflow":
        # Credit once per transactionId. amount is a decimal string: use Decimal, never float.
        account = body.get("receivingAccountNumber") or body.get("accountNumber")
        log.info("money received %s %s %s %s", body["transactionId"], account, body["amount"], body["currencyIso3Code"])
    elif event == "account/account_opened":
        # Save the account numbers in fundingDetails against accountId and clientId.
        numbers = [d.get("accountNumber") for d in body.get("fundingDetails") or []]
        log.info("account opened %s %s %s", body["accountId"], body["clientId"], numbers)
    elif event == "client/subclient_approved":
        # status "active" = approved, "rejected" = declined. Ignore approvalStatus.
        log.info("customer reviewed %s %s", body["entityId"], body["status"])
    elif event == "receivables/inflow":
        # Record the payment once per transactionDetails.transactionId; update the invoice status.
        log.info("invoice payment %s %s %s", body["receivableDetails"]["entityId"],
                 body["receivableDetails"]["status"], body["transactionDetails"]["transactionId"])
    elif event == "trade/trade_completed":
        # Act only on status "completed", and confirm the target account was credited before releasing funds.
        log.info("conversion %s %s", body["id"], body["status"])
    else:
        log.warning("unknown event ignored: %s", event)


def sweep_forever():
    # Safety net: every minute, process events a crash or an error left unprocessed.
    while True:
        time.sleep(60)
        try:
            with pool.connection() as conn:
                keys = [r[0] for r in conn.execute(
                    "SELECT idempotency_key FROM multigate_webhook_events "
                    "WHERE processed_at IS NULL AND received_at < now() - interval '1 minute' "
                    "ORDER BY received_at LIMIT 100").fetchall()]
            for key in keys:
                process_event(key)
        except Exception:
            log.exception("webhook sweep failed")


threading.Thread(target=sweep_forever, daemon=True).start()