Webhook to Iceberg

{
  "name": "webhook-ingress",
  "main": "worker.js",
  "compatibility_date": "2026-08-27",
  "pipelines": [
    { "binding": "WEBHOOKS", "stream": "partner_webhooks" }
  ],
  "vars": {
    "WEBHOOK_ISSUER": "partner"
  }
}

{
  "fields": [
    { "name": "event_id", "type": "string", "required": true },
    { "name": "event_type", "type": "string", "required": true },
    { "name": "issuer", "type": "string", "required": true },
    { "name": "received_at", "type": "timestamp", "required": true },
    { "name": "payload", "type": "json", "required": true }
  ]
}

export default {
  async fetch(request, env) {
    if (request.method !== "POST") {
      return new Response("Method not allowed", { status: 405 });
    }

    const raw = await request.text();
    const valid = await verifySignature(
      raw,
      request.headers.get("x-webhook-signature"),
      env.WEBHOOK_SECRET,
    );
    if (!valid) return new Response("Invalid signature", { status: 401 });

    let payload;
    try {
      payload = JSON.parse(raw);
    } catch {
      return new Response("Invalid JSON", { status: 400 });
    }
    await env.WEBHOOKS.send([{
      event_id: payload.id,
      event_type: payload.type,
      issuer: env.WEBHOOK_ISSUER,
      received_at: new Date().toISOString(),
      payload,
    }]);

    return new Response(null, { status: 202 });
  },
};

INSERT INTO webhook_events
SELECT event_id,
       event_type,
       issuer,
       received_at,
       payload.customer.id AS customer_id,
       payload.data AS data
FROM partner_webhooks
WHERE event_id IS NOT NULL
  AND event_type IS NOT NULL;

INSERT INTO raw_webhooks
SELECT * FROM partner_webhooks;

INSERT INTO webhook_events
SELECT event_id, event_type, issuer, received_at, payload.data AS data
FROM partner_webhooks
WHERE event_id IS NOT NULL;