feat(unstable): kv.watch() (#21147)

This commit adds support for a new `kv.watch()` method that allows
watching for changes to a key-value pair. This is useful for cases
where you want to be notified when a key-value pair changes, but
don't want to have to poll for changes.

---------

Co-authored-by: losfair <zhy20000919@hotmail.com>

1 files changed, 44 insertions, 0 deletions

diff --git a/cli/tests/unit/kv_test.ts b/cli/tests/unit/kv_test.ts
index 73c85dd5c..68b3c4013 100644
--- a/cli/tests/unit/kv_test.ts
+++ b/cli/tests/unit/kv_test.ts
@@ -2137,3 +2137,47 @@ Deno.test(
     // calling [Symbol.dispose] after manual close is a no-op
   },
 );
+
+dbTest("key watch", async (db) => {
+  const changeHistory: Deno.KvEntryMaybe<number>[] = [];
+  const watcher: ReadableStream<Deno.KvEntryMaybe<number>[]> = db.watch<
+    number[]
+  >([["key"]]);
+
+  const reader = watcher.getReader();
+  const expectedChanges = 2;
+
+  const work = (async () => {
+    for (let i = 0; i < expectedChanges; i++) {
+      const message = await reader.read();
+      if (message.done) {
+        throw new Error("Unexpected end of stream");
+      }
+      changeHistory.push(message.value[0]);
+    }
+
+    await reader.cancel();
+  })();
+
+  while (changeHistory.length !== 1) {
+    await sleep(100);
+  }
+  assertEquals(changeHistory[0], {
+    key: ["key"],
+    value: null,
+    versionstamp: null,
+  });
+
+  const { versionstamp } = await db.set(["key"], 1);
+  while (changeHistory.length as number !== 2) {
+    await sleep(100);
+  }
+  assertEquals(changeHistory[1], {
+    key: ["key"],
+    value: 1,
+    versionstamp,
+  });
+
+  await work;
+  await reader.cancel();
+});