D1 database API expert: secure…

# Cloudflare D1 分批事务迁移方案

一、方案思路

• D1 的 batch() 原生支持事务：批次内语句要么全部成功，要么整体回滚。
• 分批处理（如每批 500~1000 条），避免单次事务过大导致超时或锁冲突。
• 采用「插入 → 校验影响行数 → 删除源表」的模式，配合 meta.changes 确认写入行数。

二、核心代码（Workers 环境示例）

```javascript

const BATCH_SIZE = 500;

const MAX_RETRY = 3;

export default {

async fetch(request, env) {

let total = 0;

// 建议用自增/有序列（如 created_at 或 id）分页，避免 OFFSET 性能问题

let lastId = 0;

while (true) {

// ✅ 参数化查询，防 SQL 注入

const { results: rows } = await env.DB.prepare(

SELECT * FROM orders WHERE id > ?1 ORDER BY id LIMIT ?2

).bind(lastId, BATCH_SIZE).all();

if (!rows.length) break;

// 预编译语句数组，batch() 整体执行 = 一个事务

const stmts = [];

for (const r of rows) {

stmts.push(

env.DB.prepare(

`INSERT INTO archive (id, user_id, amount, status, created_at)

VALUES (?1, ?2, ?3, ?4, ?5)`

).bind(r.id, r.user_id, r.amount, r.status, r.created_at)

);

stmts.push(

env.DB.prepare(DELETE FROM orders WHERE id = ?1).bind(r.id)

);

}

// batch() 事务性执行：任一语句失败 → 整批回滚

const result = await env.DB.batch(stmts);

for (const res of result) {

if (!res.success) throw new Error("Batch failed, auto-rolled back");

}

total += rows.length;

lastId = rows[rows.length - 1].id;

}

return new Response(Migrated ${total} rows);

}

};

```

三、关键安全点

• 防注入：全部用 ?1, ?2 占位符 + .bind()，绝不字符串拼接。
• 事务回滚：D1 batch() 是隐式事务，失败自动回滚该批次。
• 幂等性：archive 表建议对 id 建唯一索引，用 INSERT OR IGNORE 防止重试时重复插入。
• **失败重试