Build a cheap data pipeline with TypeScript on Cloudflare or AWS

When an AI writes and runs code for you, what matters most is a system with few parts that can be tried right away, checked with SQL and retried safely. So start at the lightest tier, and move up only when measured numbers call for it.

Which tier should you start at, and what does it cost?#

A message on Cloudflare Queues usually costs 3 operations: write, read, delete. AWS prices are for US East. Pipelines is in beta, so its pricing may change.

How do you build one Worker that does all three jobs?#

The problem: a shop takes orders, wants to store them in small batches and total revenue each night. One Worker does all three.

Run locally on 3 November 2026 with wrangler 4.151.3: sending an order twice did not double it, a malformed order got a 400, and cron computed the right total for the day before. Not yet deployed to a real account.

// wrangler.jsonc
{
  "name": "orders-pipeline",
  "main": "src/index.ts",
  "compatibility_date": "2026-10-01",
  "queues": {
    "producers": [{ "queue": "orders", "binding": "ORDERS" }],
    "consumers": [{ "queue": "orders", "max_batch_size": 50, "max_batch_timeout": 5 }]
  },
  "d1_databases": [
    { "binding": "DB", "database_name": "orders-db", "database_id": "<ID from d1 create>" }
  ],
  // Cron is in UTC: 19:00 UTC is 02:00 in Vietnam
  "triggers": { "crons": ["0 19 * * *"] }
}

-- schema.sql
CREATE TABLE orders (
  order_id TEXT PRIMARY KEY,
  shop_id  TEXT NOT NULL,
  amount   INTEGER NOT NULL,
  ts       INTEGER NOT NULL
);
CREATE TABLE daily_revenue (
  day TEXT, shop_id TEXT, so_don INTEGER, doanh_thu INTEGER,
  PRIMARY KEY (day, shop_id)
);

// src/index.ts
import { z } from "zod";

// The event contract: everything is checked at the door
const OrderEvent = z.object({
  orderId: z.string().min(1),
  shopId: z.string().min(1),
  amount: z.number().int().nonnegative(),
  ts: z.number().int(), // event time, milliseconds
});
type OrderEvent = z.infer<typeof OrderEvent>;

interface Env {
  ORDERS: Queue<OrderEvent>;
  DB: D1Database;
}

export default {
  // 1) Take the event over HTTP, put it on the Queue and answer at once
  async fetch(req: Request, env: Env): Promise<Response> {
    const parsed = OrderEvent.safeParse(await req.json());
    if (!parsed.success) return new Response("invalid event", { status: 400 });
    await env.ORDERS.send(parsed.data);
    return new Response("ok", { status: 202 });
  },

  // 2) Micro-batch stream: the Queue collects up to 50 messages or 5 seconds, then calls once
  async queue(batch: MessageBatch<OrderEvent>, env: Env): Promise<void> {
    const stmt = env.DB.prepare(
      `INSERT INTO orders (order_id, shop_id, amount, ts) VALUES (?1, ?2, ?3, ?4)
       ON CONFLICT(order_id) DO NOTHING`, // idempotent: handling it again does not double it
    );
    await env.DB.batch(
      batch.messages.map((m) => stmt.bind(m.body.orderId, m.body.shopId, m.body.amount, m.body.ts)),
    );
    batch.ackAll();
  },

  // 3) Batch: a Cron Trigger runs each night and computes yesterday's revenue
  async scheduled(_controller: ScheduledController, env: Env): Promise<void> {
    await env.DB.prepare(
      `INSERT OR REPLACE INTO daily_revenue (day, shop_id, so_don, doanh_thu)
       SELECT date(ts / 1000, 'unixepoch'), shop_id, COUNT(*), SUM(amount)
       FROM orders
       WHERE date(ts / 1000, 'unixepoch') = date('now', '-1 day')
       GROUP BY shop_id`,
    ).run();
  },
} satisfies ExportedHandler<Env>;

Run these in order:

npm install zod
npx wrangler d1 create orders-db
npx wrangler d1 execute orders-db --local --file=schema.sql
npx wrangler queues create orders
npx wrangler dev --test-scheduled

Check it with:

TS=$(( ($(date +%s) - 86400) * 1000 ))
curl -X POST localhost:8787 -d "{\"orderId\":\"A1\",\"shopId\":\"s7\",\"amount\":100,\"ts\":$TS}"
curl -X POST localhost:8787 -d "{\"orderId\":\"A1\",\"shopId\":\"s7\",\"amount\":100,\"ts\":$TS}"
curl "localhost:8787/__scheduled?cron=0+19+*+*+*"
npx wrangler d1 execute orders-db --local --command "SELECT * FROM daily_revenue"

The daily_revenue table should show so_don as 1 even though the order was sent twice. When it works, change --local to --remote and run npx wrangler deploy.

How does it look on AWS?#

Turn on ReportBatchItemFailures on the event source mapping so only failed messages are retried. For the nightly batch, EventBridge Scheduler calls Lambda or Step Functions. This sample has not been run on a real AWS account.

// handler.ts
import type { SQSBatchResponse, SQSHandler } from "aws-lambda";
import { DynamoDBClient } from "@aws-sdk/client-dynamodb";
import { DynamoDBDocumentClient, PutCommand } from "@aws-sdk/lib-dynamodb";
import { z } from "zod";

const OrderEvent = z.object({
  orderId: z.string().min(1),
  shopId: z.string().min(1),
  amount: z.number().int().nonnegative(),
  ts: z.number().int(),
});
type OrderEvent = z.infer<typeof OrderEvent>;

const db = DynamoDBDocumentClient.from(new DynamoDBClient({}));

// Idempotent write: skip if orderId already exists
async function saveOrder(e: OrderEvent) {
  try {
    await db.send(
      new PutCommand({
        TableName: process.env.TABLE_NAME!,
        Item: e,
        ConditionExpression: "attribute_not_exists(orderId)",
      }),
    );
  } catch (err) {
    if ((err as Error).name !== "ConditionalCheckFailedException") throw err;
  }
}

// Only failed messages are returned for SQS to retry
export const handler: SQSHandler = async (event): Promise<SQSBatchResponse> => {
  const batchItemFailures: SQSBatchResponse["batchItemFailures"] = [];
  for (const record of event.Records) {
    try {
      await saveOrder(OrderEvent.parse(JSON.parse(record.body)));
    } catch {
      batchItemFailures.push({ itemIdentifier: record.messageId });
    }
  }
  return { batchItemFailures };
};

Why does this approach suit an AI?#

• Infrastructure is code. Configuration lives in files, so an AI reads, edits and diffs it through git.
• Few parts, few ways to go wrong. Serverless scales itself. An AI has no machine counts or clusters to tune.
• SQL is a shared language. An AI writes it well and you check the result with one query.
• Retries are safe. Idempotent writes and a dead-letter queue allow re-runs with no duplicates.
• It runs before it deploys. wrangler dev runs the Worker, Queue and D1 locally.
• Cost has a ceiling. Paying by use keeps mistakes cheap. Still turn on cost alerts.

How do you hand this job to an AI?#

Copy this, fill in the blanks and give it to the AI:

Build an event pipeline with these requirements.

Event: [field names, types, which field is the unique key, which is event time]
Volume: [events per day, at peak]
Acceptable delay: [for example: under 10 seconds for the orders table, daily for the report]
Platform: [Cloudflare Workers + Queues + D1, or AWS Lambda + SQS + DynamoDB]
Cost ceiling: [amount per month]

Required:
- Write the event contract in Zod and check it at the door.
- Every write is idempotent on the unique key.
- A dead-letter path for messages that keep failing.
- No permission to delete real data. Tokens cover only the resources needed.

Prove it before reporting done:
1. Run locally with sample events.
2. Send one event twice and show the result is not doubled.
3. Send a malformed event and show it is rejected.
4. Print the result with one SQL query.

Common mistakes#

• Forgetting that cron runs in UTC. The schedule ends up hours off from what you expect.
• No dead-letter path. Failed messages retry forever or vanish. Cap the retries.
• Giving an AI too wide a token. Separate dev and prod, and do not grant deletion of real data.
• Running several SQL statements with --command. --file=schema.sql is safer.
• Not measuring how far behind you are. Watch dead-letter counts and delay, so an AI can spot problems from them.