使用队列

// 描述您的消息对象的形状（可选）
interface Notification {
  forUser: string;
  body: string;
}

// 获取 KV 实例的引用
const kv = await Deno.openKv();

// 创建一个通知对象
const message: Notification = {
  forUser: "alovelace",
  body: "您有邮件！",
};

// 将消息入队以进行立即投递
await kv.enqueue(message);

// 定义我们期望的队列消息的对象形状
interface Notification {
  forUser: string;
  body: string;
}

// 创建一个类型保护来检查传入消息的类型
function isNotification(o: unknown): o is Notification {
  return (
    ((o as Notification)?.forUser !== undefined &&
      typeof (o as Notification).forUser === "string") &&
    ((o as Notification)?.body !== undefined &&
      typeof (o as Notification).body === "string")
  );
}

// 获取 KV 数据库的引用
const kv = await Deno.openKv();

// 注册一个处理函数以监听值 - 此示例展示
// 您如何发送通知
kv.listenQueue((msg: unknown) => {
  // 使用类型保护 - 然后 TypeScript 编译器知道 msg 是 Notification
  if (isNotification(msg)) {
    console.log("向用户发送通知:", msg.forUser);
    // ... 做一些实际发送通知的事情！
  } else {
    // 如果消息是未知类型，可能是一个错误
    console.error("收到未知消息:", msg);
  }
});

const kv = await Deno.openKv();

kv.listenQueue(async (msg: unknown) => {
  const nonce = await kv.get(["nonces", msg.nonce]);
  if (nonce.value === null) {
    // 这条消息已经被处理
    return;
  }

  const change = msg.change;
  const bob = await kv.get(["balance", "bob"]);
  const liz = await kv.get(["balance", "liz"]);

  const success = await kv.atomic()
    // 确保这条消息尚未被处理
    .check({ key: nonce.key, versionstamp: nonce.versionstamp })
    .delete(nonce.key)
    .sum(["processed_count"], 1n)
    .check(bob, liz) // 余额没有变化
    .set(["balance", "bob"], bob.value - change)
    .set(["balance", "liz"], liz.value + change)
    .commit();
});

// 在同一 KV 事务中修改键并入队消息！
const nonce = crypto.randomUUID();
await kv
  .atomic()
  .check({ key: ["nonces", nonce], versionstamp: null })
  .enqueue({ nonce: nonce, change: 10 })
  .set(["nonces", nonce], true)
  .sum(["enqueued_count"], 1n)
  .commit();