Handle revocations

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

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

export function verifyWebhook(headers: Headers, body: string): boolean {
  const id = headers.get("webhook-id");
  const timestamp = headers.get("webhook-timestamp");
  const signatures = headers.get("webhook-signature");
  if (!id || !timestamp || !signatures) return false;

  const sentAt = Number(timestamp);
  if (!Number.isInteger(sentAt)) return false;
  if (Math.abs(Date.now() / 1000 - sentAt) > TOLERANCE_SECONDS) return false;

  const key = Buffer.from(SECRET.slice("whsec_".length), "base64");
  const expected = createHmac("sha256", key).update(`${id}.${timestamp}.${body}`).digest();

  return signatures.split(" ").some((entry) => {
    const [version, signature] = entry.split(",");
    if (version !== "v1" || !signature) return false;
    const received = Buffer.from(signature, "base64");
    return received.length === expected.length && timingSafeEqual(received, expected);
  });
}

// A Fetch API handler. Read the raw body before parsing it.
export async function handleWebhook(req: Request): Promise<Response> {
  const body = await req.text();
  if (!verifyWebhook(req.headers, body)) return new Response("invalid signature", { status: 400 });

  const event = JSON.parse(body);
  if (await seen(event.id)) return new Response(null, { status: 204 }); // duplicate delivery

  if (event.type === "revocation.created") await queue.push("revocation", event.data.object.id);
  await markSeen(event.id);
  return new Response(null, { status: 204 });
}

import base64, hashlib, hmac, os, time
from flask import Flask, request

SECRET = os.environ["REFLECTIONS_WEBHOOK_SECRET"]  # whsec_…
TOLERANCE_SECONDS = 5 * 60

def verify_webhook(headers, body: bytes) -> bool:
    msg_id = headers.get("webhook-id")
    timestamp = headers.get("webhook-timestamp")
    signatures = headers.get("webhook-signature")
    if not (msg_id and timestamp and signatures):
        return False
    try:
        if abs(time.time() - int(timestamp)) > TOLERANCE_SECONDS:
            return False
    except ValueError:
        return False

    key = base64.b64decode(SECRET.removeprefix("whsec_"))
    signed = f"{msg_id}.{timestamp}.".encode() + body
    expected = base64.b64encode(hmac.new(key, signed, hashlib.sha256).digest()).decode()

    for entry in signatures.split(" "):
        version, _, signature = entry.partition(",")
        if version == "v1" and hmac.compare_digest(signature, expected):
            return True
    return False

app = Flask(__name__)

@app.post("/webhooks/reflections")
def reflections_webhook():
    body = request.get_data()  # the raw body, before parsing
    if not verify_webhook(request.headers, body):
        return "invalid signature", 400

    event = request.get_json()
    if seen(event["id"]):
        return "", 204  # duplicate delivery

    if event["type"] == "revocation.created":
        queue.push("revocation", event["data"]["object"]["id"])
    mark_seen(event["id"])
    return "", 204

curl -G https://app.reflections.ai/v1/revocations \
  -H "Authorization: Bearer rk_test_…" \
  -H "Reflections-Version: 2026-10-15" \
  --data-urlencode "after=rvk_m2p4r6t8w0y2a4c6e8g0j2m4p6" \
  --data-urlencode "limit=100"

const API = "https://app.reflections.ai/v1";
const headers = {
  Authorization: `Bearer ${process.env.REFLECTIONS_KEY}`,
  "Reflections-Version": "2026-10-15",
};

async function list(params: Record<string, string>) {
  const url = new URL(`${API}/revocations`);
  for (const [name, value] of Object.entries(params)) url.searchParams.set(name, value);
  const res = await fetch(url, { headers });
  if (!res.ok) throw new Error(`List revocations failed: ${res.status}`); // retry on the next run
  return res.json();
}

// No cursor yet: start at the newest revocation, and apply every one still open.
async function firstRun() {
  const newest: string | undefined = (await list({ limit: "1" })).data[0]?.id;
  for (const status of ["pending_acknowledgement", "overdue"]) {
    let next: string | null = null;
    do {
      const params: Record<string, string> = { status, limit: "100" };
      if (next) params.cursor = next;
      const page = await list(params); // newest first
      for (const revocation of page.data) await applyRevocation(revocation);
      next = page.next_cursor;
    } while (next);
  }
  if (newest) await cursor.set(newest);
  return newest;
}

export async function syncRevocations() {
  let after = (await cursor.get()) ?? (await firstRun()); // the newest revocation you finished
  while (after) {
    const page = await list({ after, limit: "100" }); // oldest first
    for (const revocation of page.data) {
      await applyRevocation(revocation);
      after = revocation.id;
      await cursor.set(after);
    }
    if (!page.has_more) return;
  }
}

import os, requests

API = "https://app.reflections.ai/v1"
headers = {
    "Authorization": f"Bearer {os.environ['REFLECTIONS_KEY']}",
    "Reflections-Version": "2026-10-15",
}

def list_revocations(params):
    res = requests.get(f"{API}/revocations", headers=headers, params=params)
    res.raise_for_status()  # retry on the next run
    return res.json()

def first_run():
    """No cursor yet: start at the newest revocation, and apply every one still open."""
    latest = list_revocations({"limit": 1})["data"]
    newest = latest[0]["id"] if latest else None
    for status in ("pending_acknowledgement", "overdue"):
        params = {"status": status, "limit": 100}
        while True:
            page = list_revocations(params)  # newest first
            for revocation in page["data"]:
                apply_revocation(revocation)
            if not page["next_cursor"]:
                break
            params["cursor"] = page["next_cursor"]
    if newest:
        cursor.set(newest)
    return newest

def sync_revocations():
    after = cursor.get() or first_run()  # the newest revocation you finished
    while after:
        page = list_revocations({"after": after, "limit": 100})  # oldest first
        for revocation in page["data"]:
            apply_revocation(revocation)
            after = revocation["id"]
            cursor.set(after)
        if not page["has_more"]:
            return

curl https://app.reflections.ai/v1/revocations/rvk_s9t0wxyhk5zba2q8v1m3n7p4r6/acknowledgements \
  -H "Authorization: Bearer rk_test_…" \
  -H "Reflections-Version: 2026-10-15" \
  -H "Idempotency-Key: ack-rvk_s9t0wxyhk5zba2q8v1m3n7p4r6" \
  -H "Content-Type: application/json" \
  -d '{
    "actions": ["stop_generation", "purge_cached_decisions", "delete_provider_copies"],
    "deleted_copies": [
      { "provider": "example-voice-provider", "external_id": "vx_4f9a2c", "deleted_at": "2026-10-17T18:09:12Z" }
    ],
    "note": "Voice deleted at the provider. Reuse caches purged."
  }'

async function acknowledge(revocationId: string, key: string, body: object) {
  const res = await fetch(`${API}/revocations/${revocationId}/acknowledgements`, {
    method: "POST",
    headers: { ...headers, "Idempotency-Key": key, "Content-Type": "application/json" },
    body: JSON.stringify(body),
  });
  if (!res.ok) throw new Error(`Acknowledgement failed: ${res.status}`); // retry with the same key
}

// By acknowledge_by. deletedCopies comes from your own record, so a retry sends the same body.
export async function acknowledgeActions(revocation: any, deletedCopies: object[]) {
  await acknowledge(revocation.id, `ack-${revocation.id}`, {
    actions: revocation.required_actions.filter((a: string) => a !== "withdraw_outputs"),
    deleted_copies: deletedCopies,
    note: "Voice deleted at the provider. Reuse caches purged.",
  });
}

// Later, by post_revocation.withdraw_by, once every released output is down.
export async function acknowledgeWithdrawal(revocation: any) {
  await acknowledge(revocation.id, `ack-${revocation.id}-withdraw`, { actions: ["withdraw_outputs"] });
}

def acknowledge(revocation_id, key, body):
    res = requests.post(
        f"{API}/revocations/{revocation_id}/acknowledgements",
        headers={**headers, "Idempotency-Key": key},
        json=body,
    )
    res.raise_for_status()  # retry with the same key

# By acknowledge_by. deleted_copies comes from your own record, so a retry sends the same body.
def acknowledge_actions(revocation, deleted_copies):
    acknowledge(revocation["id"], f"ack-{revocation['id']}", {
        "actions": [a for a in revocation["required_actions"] if a != "withdraw_outputs"],
        "deleted_copies": deleted_copies,
        "note": "Voice deleted at the provider. Reuse caches purged.",
    })

# Later, by post_revocation.withdraw_by, once every released output is down.
def acknowledge_withdrawal(revocation):
    acknowledge(revocation["id"], f"ack-{revocation['id']}-withdraw", {"actions": ["withdraw_outputs"]})