Bronze, silver, and gold

await env.RAW_EVENTS.send([{
  source: "billing",
  source_id: payload.id,
  ingested_at: new Date().toISOString(),
  payload,
}]);

{
  "fields": [
    { "name": "source", "type": "string", "required": true },
    { "name": "source_id", "type": "string", "required": true },
    { "name": "ingested_at", "type": "timestamp", "required": true },
    { "name": "payload", "type": "json", "required": true }
  ]
}

INSERT INTO bronze_events
SELECT source, source_id, ingested_at, payload
FROM raw_events;

INSERT INTO silver_transactions
SELECT source_id AS transaction_id,
       payload.customer_id AS customer_id,
       ROUND(payload.amount * 100) AS amount_cents,
       payload.currency AS currency,
       payload.occurred_at AS occurred_at,
       ingested_at
FROM raw_events
WHERE payload.customer_id IS NOT NULL
  AND payload.amount >= 0
  AND payload.currency IS NOT NULL;

INSERT INTO rejected_transactions
SELECT source_id, ingested_at, payload, 'validation failed' AS reason
FROM raw_events
WHERE payload.customer_id IS NULL
   OR payload.amount IS NULL
   OR payload.amount < 0
   OR payload.currency IS NULL;

const result = await env.SALES.query("revenue-by-currency", {
  from: "2026-08-01",
  to: "2026-08-31",
});