Webhooks

import crypto from "crypto";
import express from "express";

const app = express();
app.use(express.raw({ type: "application/json" }));

const secret = process.env.EMERGE_WEBHOOK_SECRET;
if (!secret) {
  throw new Error("Missing EMERGE_WEBHOOK_SECRET");
}

app.post("/webhooks/emerge", async (req, res) => {
  try {
    const signature = req.header("x-signature");
    const idempotencyKey = req.header("idempotency-key");

    if (!signature || !idempotencyKey) {
      return res.status(401).send("Missing signature or idempotency key");
    }

    const expected = crypto.createHmac("sha256", secret).update(req.body).digest("hex");
    const valid =
      signature.length === expected.length &&
      crypto.timingSafeEqual(Buffer.from(signature, "hex"), Buffer.from(expected, "hex"));

    if (!valid) {
      return res.status(401).send("Invalid signature");
    }

    // Idempotency guard (replace with DB/Redis in production)
    if (seen(idempotencyKey)) {
      return res.status(200).send("Duplicate ignored");
    }
    remember(idempotencyKey);

    const payload = JSON.parse(req.body.toString("utf8")) as {
      event: string;
      uid: string;
      sources: Array<Record<string, unknown>>;
    };

    if (payload.event === "consent.revoked") {
      await handleRevocation(payload.uid, payload.sources);
    } else if (payload.event === "data.ready") {
      await handleDataReady(payload.uid, payload.sources);
    } else if (payload.event === "data.failed") {
      await handleDataFailed(payload.uid, payload.sources);
    }

    return res.status(200).send("OK");
  } catch (error) {
    console.error("Webhook processing failed", error);
    return res.status(500).send("Server error");
  }
});

const idempotencyStore = new Set<string>();
function seen(key: string): boolean {
  return idempotencyStore.has(key);
}
function remember(key: string): void {
  idempotencyStore.add(key);
}

async function handleRevocation(uid: string, sources: Array<Record<string, unknown>>) {
  for (const source of sources) {
    const provider = String(source.provider ?? "");
    if (provider) {
      await purgeProviderData(uid, provider);
    }
  }
}

async function handleDataReady(uid: string, sources: Array<Record<string, unknown>>) {
  for (const source of sources) {
    const provider = String(source.provider ?? "");
    if (provider) {
      console.log(`Data ready for ${provider} and user ${uid}`);
      await enqueueProviderQuery(uid, provider);
    }
  }
}

async function handleDataFailed(uid: string, sources: Array<Record<string, unknown>>) {
  for (const source of sources) {
    const provider = String(source.provider ?? "unknown_provider");
    const errorCode = String(source.error_code ?? "unknown_error");
    const errorMessage = String(source.error_message ?? "No details provided");
    console.error(
      `Data export failed for ${provider} and user ${uid}: ${errorCode} - ${errorMessage}`
    );
    await notifyOps(uid, provider, errorCode, errorMessage);
  }
}

async function purgeProviderData(uid: string, provider: string) {
  console.log(`Purging ${provider} data for ${uid}`);
}

async function enqueueProviderQuery(uid: string, provider: string) {
  console.log(`Queueing query for ${provider} and user ${uid}`);
}

async function notifyOps(
  uid: string,
  provider: string,
  errorCode: string,
  errorMessage: string
) {
  console.log(`Notify ops: ${uid} ${provider} ${errorCode} ${errorMessage}`);
}

import hashlib
import hmac
import json
import os
from fastapi import FastAPI, HTTPException, Request

app = FastAPI()
secret = os.getenv("EMERGE_WEBHOOK_SECRET")
if not secret:
    raise RuntimeError("Missing EMERGE_WEBHOOK_SECRET")

_seen: set[str] = set()

@app.post("/webhooks/emerge")
async def receive_webhook(request: Request):
    body = await request.body()
    signature = request.headers.get("x-signature")
    idem_key = request.headers.get("idempotency-key")

    if not signature or not idem_key:
        raise HTTPException(status_code=401, detail="Missing signature or idempotency key")

    expected = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(signature, expected):
        raise HTTPException(status_code=401, detail="Invalid signature")

    if idem_key in _seen:
        return {"status": "duplicate_ignored"}
    _seen.add(idem_key)

    payload = json.loads(body)
    event = payload.get("event")
    if event == "consent.revoked":
        uid = payload.get("uid", "")
        for source in payload.get("sources", []):
            provider = source.get("provider")
            if provider:
                await purge_provider_data(uid, provider)
    elif event == "data.ready":
        uid = payload.get("uid", "")
        for source in payload.get("sources", []):
            provider = source.get("provider")
            if provider:
                await enqueue_provider_query(uid, provider)
    elif event == "data.failed":
        uid = payload.get("uid", "")
        for source in payload.get("sources", []):
            provider = source.get("provider", "unknown_provider")
            error_code = source.get("error_code", "unknown_error")
            error_message = source.get("error_message", "No details provided")
            await notify_ops(uid, provider, error_code, error_message)

    return {"status": "ok"}

async def purge_provider_data(uid: str, provider: str):
    print(f"Purging {provider} data for {uid}")

async def enqueue_provider_query(uid: str, provider: str):
    print(f"Queueing query for {provider} and user {uid}")

async def notify_ops(uid: str, provider: str, error_code: str, error_message: str):
    print(f"Data export failed for {provider} and user {uid}: {error_code} - {error_message}")