使用队列

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

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

// 创建通知对象
const message: Notification = {
  forUser: "alovelace",
  body: "You've got mail!",
};

// 立即入队消息进行交付
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("Sending notification to user:", msg.forUser);
    // ... 实际发送通知的操作！
  } else {
    // 如果消息类型未知，可能是错误
    console.error("Unknown message received:", 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();