interface Env {  ORDERS: K2Producer}
export default {  async fetch(request, env): Promise<Response> {    const event = await request.json()    const encoder = new TextEncoder()
    // Append a batch of records: either all are stored, or none are    const result = await env.ORDERS.send([      {        content: encoder.encode(JSON.stringify(event)),        headers: { "event-type": "order.created" },      },    ])
    // send() returns a result instead of throwing    if (!result.success) {      return new Response(result.error.message, {        status: result.error.retryable ? 503 : 500,      })    }
    return new Response("Event recorded")  },} satisfies ExportedHandler<Env>

const K2_ENDPOINT = "https://<STREAM_ID>.k2.cloudflarestorage.com"const SUBSCRIPTION = `${K2_ENDPOINT}/subscriptions/<SUBSCRIPTION_ID>`const WORKER_ID = "worker-1"
async function post(path: string, body: object) {  const res = await fetch(`${SUBSCRIPTION}${path}`, {    method: "POST",    headers: {      Authorization: `Bearer ${process.env.CLOUDFLARE_API_TOKEN}`,      "Content-Type": "application/json",    },    body: JSON.stringify(body),  })  return (await res.json()).result}
const decoder = new TextDecoder()
while (true) {  // Lease up to 100 records to this worker for five minutes  const batch = await post("/consume", { worker_id: WORKER_ID, max_records: 100 })
  if (!batch?.batch_id) {    // No new records: back off before polling again    await new Promise((resolve) => setTimeout(resolve, 1000))    continue  }
  try {    for (const record of batch.records) {      const bytes = Uint8Array.fromBase64(record.content)      await processOrder(JSON.parse(decoder.decode(bytes)))    }    // Mark the batch processed and move the subscription forward    await post(`/batches/${batch.batch_id}/ack`, { worker_id: WORKER_ID })  } catch {    // Release the lease so the records are delivered again    await post(`/batches/${batch.batch_id}/nack`, { worker_id: WORKER_ID })  }}