Webhook events

import { createHmac, timingSafeEqual } from "node:crypto";
import express from "express";

const SECRET = process.env.RELAYCAST_WEBHOOK_SECRET!; // whsec_...
const TOLERANCE_SECONDS = 5 * 60;

function verify(rawBody: Buffer, timestamp: string | undefined, signature: string | undefined): boolean {
  if (!timestamp || !signature) return false;
  const ts = Number(timestamp);
  if (!Number.isFinite(ts) || Math.abs(Date.now() / 1000 - ts) > TOLERANCE_SECONDS) return false;
  const expected = createHmac("sha256", SECRET).update(`${timestamp}.`).update(rawBody).digest("hex");
  const a = Buffer.from(expected, "utf8");
  const b = Buffer.from(signature, "utf8");
  return a.length === b.length && timingSafeEqual(a, b);
}

const app = express();
const seen = new Set<string>(); // use a database or cache in production

app.post("/webhooks/relaycast", express.raw({ type: "application/json" }), (req, res) => {
  if (!verify(req.body, req.header("x-webhook-timestamp"), req.header("x-signature"))) {
    return res.status(401).send("invalid signature");
  }
  const deliveryId = req.header("x-webhook-id")!;
  if (seen.has(deliveryId)) return res.sendStatus(200); // duplicate retry
  seen.add(deliveryId);

  const event = JSON.parse(req.body.toString("utf8"));
  if (event.event === "post.failed" || event.event === "post.partial") {
    for (const t of event.data.targets) if (t.status === "failed") console.warn(t.platform, t.error);
  }
  res.sendStatus(200); // respond quickly; do slow work asynchronously
});

app.listen(3001);

import hashlib, hmac, json, os, time
from flask import Flask, abort, request

SECRET = os.environ["RELAYCAST_WEBHOOK_SECRET"].encode()  # whsec_...
TOLERANCE_SECONDS = 5 * 60

app = Flask(__name__)
seen = set()  # use a database or cache in production


def verify(raw_body: bytes, timestamp: str | None, signature: str | None) -> bool:
    if not timestamp or not signature:
        return False
    try:
        ts = int(timestamp)
    except ValueError:
        return False
    if abs(time.time() - ts) > TOLERANCE_SECONDS:
        return False
    expected = hmac.new(SECRET, timestamp.encode() + b"." + raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature)


@app.post("/webhooks/relaycast")
def social_publish_webhook():
    raw = request.get_data()  # raw bytes, before JSON parsing
    if not verify(raw, request.headers.get("x-webhook-timestamp"), request.headers.get("x-signature")):
        abort(401)
    delivery_id = request.headers["x-webhook-id"]
    if delivery_id in seen:
        return "", 200  # duplicate retry
    seen.add(delivery_id)

    event = json.loads(raw)
    if event["event"] in ("post.failed", "post.partial"):
        for t in event["data"]["targets"]:
            if t["status"] == "failed":
                print(t["platform"], t["error"])
    return "", 200