Streams and Pipelines

{
  "pipelines": [
    { "binding": "EVENTS", "stream": "commerce_events" }
  ]
}

export default {
  async fetch(request, env) {
    const event = await request.json();

    await env.EVENTS.send([{
      id: event.id,
      type: event.type,
      amount: event.amount,
      received_at: new Date().toISOString(),
    }]);

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

INSERT INTO purchases
SELECT id,
       LOWER(customer_email) AS customer_email,
       ROUND(amount * 100) AS amount_cents,
       received_at
FROM commerce_events
WHERE type = 'purchase' AND amount > 0;

INSERT INTO accepted
SELECT id, type, amount
FROM events
WHERE amount >= 0;

INSERT INTO rejected
SELECT id, type, amount, 'negative amount' AS reason
FROM events
WHERE amount < 0;