Dựng pipeline xử lý dữ liệu rẻ bằng TypeScript trên Cloudflare, AWS

Khi AI viết code và vận hành giúp bạn, thứ quý nhất là hệ thống ít thành phần, chạy thử được ngay, kiểm chứng được bằng SQL và thử lại an toàn. Cách chọn: bắt đầu ở cấp nhẹ nhất, chỉ lên cấp khi số liệu đo được đòi hỏi.

Nên bắt đầu ở cấp nào, tốn bao nhiêu?#

Một tin trong Cloudflare Queues thường tốn 3 thao tác: ghi, đọc, xoá. Giá AWS là ví dụ vùng US East. Pipelines đang ở giai đoạn beta nên giá có thể đổi.

Dựng một Worker xử lý cả ba việc thế nào?#

Bài toán: một cửa hàng nhận đơn, muốn lưu đơn theo lô nhỏ và tổng hợp doanh thu mỗi đêm. Một Worker làm cả ba việc.

Đã chạy thử cục bộ ngày 03/11/2026 với wrangler 4.151.3: gửi trùng một đơn không làm nhân đôi, đơn sai định dạng nhận lỗi 400, và cron tính đúng tổng của hôm trước. Chưa triển khai lên tài khoản thật.

// 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 từ lệnh d1 create>" }
  ],
  // Cron tính theo UTC: 19:00 UTC là 02:00 sáng giờ Việt Nam
  "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";

// Hợp đồng sự kiện: mọi dữ liệu đều được kiểm tra ở cửa vào
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, mili giây
});
type OrderEvent = z.infer<typeof OrderEvent>;

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

export default {
  // 1) Nhận sự kiện qua HTTP, đưa vào Queue rồi trả lời ngay
  async fetch(req: Request, env: Env): Promise<Response> {
    const parsed = OrderEvent.safeParse(await req.json());
    if (!parsed.success) return new Response("sự kiện không hợp lệ", { status: 400 });
    await env.ORDERS.send(parsed.data);
    return new Response("ok", { status: 202 });
  },

  // 2) Stream kiểu micro-batch: Queue gom tối đa 50 tin hoặc 5 giây rồi gọi một lần
  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: xử lý lại không nhân đôi
    );
    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: Cron Trigger chạy mỗi đêm, tính doanh thu của ngày hôm qua
  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>;

Chạy theo thứ tự:

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

Kiểm tra bằng lệnh sau:

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"

Bảng daily_revenue phải có so_don là 1 dù đơn được gửi hai lần. Khi đã ổn, đổi --local thành --remote rồi chạy npx wrangler deploy.

Trên AWS thì làm thế nào?#

Bật ReportBatchItemFailures ở event source mapping để chỉ những tin lỗi được thử lại. Batch hằng đêm dùng EventBridge Scheduler gọi Lambda hoặc Step Functions. Mẫu dưới đây chưa được chạy trên tài khoản AWS thật.

// 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({}));

// Ghi idempotent: nếu orderId đã có thì bỏ qua
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;
  }
}

// Chỉ những tin lỗi được trả về để SQS thử lại
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 };
};

Vì sao cách làm này hợp với AI?#

• Hạ tầng là code. Cấu hình nằm trong file, nên AI đọc, sửa và so sánh thay đổi qua git.
• Ít thành phần, ít chỗ sai. Serverless tự co giãn. AI không phải tính số máy hay chỉnh cluster.
• SQL là ngôn ngữ chung. AI viết tốt và bạn kiểm chứng kết quả bằng một câu truy vấn.
• Thử lại an toàn. Ghi idempotent và dead-letter cho phép chạy lại mà không trùng dữ liệu.
• Chạy thử được trước khi triển khai. wrangler dev chạy Worker, Queue và D1 cục bộ.
• Chi phí có trần. Tính tiền theo mức dùng giúp sai sót có giá thấp. Vẫn nên bật cảnh báo chi phí.

Giao việc này cho AI thế nào?#

Sao chép mẫu dưới đây, điền vào chỗ trống rồi giao cho AI:

Dựng một pipeline xử lý sự kiện với yêu cầu sau.

Sự kiện: [tên trường, kiểu dữ liệu, trường nào là khoá duy nhất, trường nào là event time]
Lượng dữ liệu: [số sự kiện mỗi ngày, lúc cao điểm]
Độ trễ chấp nhận được: [ví dụ: dưới 10 giây cho bảng orders, hằng ngày cho báo cáo]
Nền tảng: [Cloudflare Workers + Queues + D1, hoặc AWS Lambda + SQS + DynamoDB]
Trần chi phí: [số tiền mỗi tháng]

Bắt buộc:
- Viết hợp đồng sự kiện bằng Zod và kiểm tra ở cửa vào.
- Mọi bước ghi phải idempotent theo khoá duy nhất.
- Có dead-letter cho tin lỗi nhiều lần.
- Không dùng quyền xoá dữ liệu thật. Token chỉ đúng tài nguyên cần dùng.

Chứng minh trước khi báo xong:
1. Chạy cục bộ với sự kiện mẫu.
2. Gửi trùng một sự kiện và cho thấy kết quả không bị nhân đôi.
3. Gửi một sự kiện sai định dạng và cho thấy nó bị từ chối.
4. In kết quả bằng một câu SQL.

Sai lầm hay gặp#

• Quên cron chạy theo UTC. Lịch sẽ lệch giờ so với bạn tưởng.
• Không có dead-letter. Tin lỗi bị thử lại mãi hoặc biến mất. Hãy giới hạn số lần thử lại.
• Cho AI token quá rộng. Tách môi trường dev và prod, không cấp quyền xoá dữ liệu thật.
• Chạy nhiều câu SQL bằng --command. Dùng --file=schema.sql thì an toàn hơn.
• Không đo độ chậm so với thực tế. Hãy theo dõi số tin trong dead-letter và độ trễ, để AI dựa vào đó mà tự phát hiện lỗi.