feat(consumer): subscribe to jetstream

deno.lock

···
     "jsr:@std/internal@^1.0.10": "1.0.10",
     "npm:@atcute/atproto@^3.1.9": "3.1.9",
     "npm:@atcute/client@^4.0.5": "4.0.5",
+    "npm:@atcute/jetstream@^1.1.2": "1.1.2",
     "npm:@atcute/lex-cli@^2.3.1": "2.3.1",
     "npm:@atcute/lexicons@^1.2.2": "1.2.2",
     "npm:@atproto/lexicon@~0.5.1": "0.5.1"
···
         "@badrap/valita"
       ]
     },
+    "@atcute/jetstream@1.1.2": {
+      "integrity": "sha512-u6p/h2xppp7LE6W/9xErAJ6frfN60s8adZuCKtfAaaBBiiYbb1CfpzN8Uc+2qtJZNorqGvuuDb5572Jmh7yHBQ==",
+      "dependencies": [
+        "@atcute/lexicons",
+        "@badrap/valita",
+        "@mary-ext/event-iterator",
+        "@mary-ext/simple-event-emitter",
+        "partysocket",
+        "type-fest",
+        "yocto-queue"
+      ]
+    },
     "@atcute/lex-cli@2.3.1": {
       "integrity": "sha512-HrHD91CFSFd/p0UFe3akFA1HXiboQwd5LbYiU0srKdLxGX+NLTX/EdCdhbLV6M7LsXdmxk7PB6BMcprsX4rbvg==",
       "dependencies": [
···
     "@badrap/valita@0.4.6": {
       "integrity": "sha512-4kdqcjyxo/8RQ8ayjms47HCWZIF5981oE5nIenbfThKDxWXtEHKipAOWlflpPJzZx9y/JWYQkp18Awr7VuepFg=="
     },
+    "@mary-ext/event-iterator@1.0.0": {
+      "integrity": "sha512-l6gCPsWJ8aRCe/s7/oCmero70kDHgIK5m4uJvYgwEYTqVxoBOIXbKr5tnkLqUHEg6mNduB4IWvms3h70Hp9ADQ==",
+      "dependencies": [
+        "yocto-queue"
+      ]
+    },
+    "@mary-ext/simple-event-emitter@1.0.0": {
+      "integrity": "sha512-meA/zJZKIN1RVBNEYIbjufkUrW7/tRjHH60FjolpG1ixJKo76TB208qefQLNdOVDA7uIG0CGEDuhmMirtHKLAg=="
+    },
     "@optique/core@0.6.2": {
       "integrity": "sha512-HTxIHJ8xLOSZotiU6Zc5BCJv+SJ8DMYmuiQM+7tjF7RolJn/pdZNe7M78G3+DgXL9lIf82l8aGcilmgVYRQnGQ=="
     },
···
     "esm-env@1.2.2": {
       "integrity": "sha512-Epxrv+Nr/CaL4ZcFGPJIYLWFom+YeV1DqMLHJoEd9SYRxNbaFruBwfEX/kkHUJf55j2+TUbmDcmuilbP1TmXHA=="
     },
+    "event-target-polyfill@0.0.4": {
+      "integrity": "sha512-Gs6RLjzlLRdT8X9ZipJdIZI/Y6/HhRLyq9RdDlCsnpxr/+Nn6bU2EFGuC94GjxqhM+Nmij2Vcq98yoHrU8uNFQ=="
+    },
     "graphemer@1.4.0": {
       "integrity": "sha512-EtKwoO6kxCL9WO5xipiHTZlSzBm7WLT627TqC/uVRd0HKmq8NXyebnNYxDoBi7wt8eTWrUrKXCOVaFq9x1kgag=="
     },
···
     "multiformats@9.9.0": {
       "integrity": "sha512-HoMUjhH9T8DDBNT+6xzkrd9ga/XiBI4xLr58LJACwK6G3HTOPeMz4nB4KJs33L2BelrIJa7P0VuNaVF3hMYfjg=="
     },
+    "partysocket@1.1.6": {
+      "integrity": "sha512-LkEk8N9hMDDsDT0iDK0zuwUDFVrVMUXFXCeN3850Ng8wtjPqPBeJlwdeY6ROlJSEh3tPoTTasXoSBYH76y118w==",
+      "dependencies": [
+        "event-target-polyfill"
+      ]
+    },
     "picocolors@1.1.1": {
       "integrity": "sha512-xceH2snhtb5M9liqDsmEw56le376mTZkEX/jEb/RxNFyegNul7eNslCXP9FDj/Lcu0X8KEyMceP2ntpaHrDEVA=="
     },
     "prettier@3.6.2": {
       "integrity": "sha512-I7AIg5boAr5R0FFtJ6rCfD+LFsWHp81dolrFD8S79U9tb8Az2nGrJncnMSnys+bpQJfRUzqs9hnA81OAA3hCuQ==",
       "bin": true
+    },
+    "type-fest@4.41.0": {
+      "integrity": "sha512-TeTSQ6H5YHvpqVwBRcnLDCBnDOHWYu7IvGbHT6N8AOymcr9PJGjc1GTtiWZTYg0NCgYwvnYWEkVChQAr9bjfwA=="
     },
     "uint8arrays@3.0.0": {
       "integrity": "sha512-HRCx0q6O9Bfbp+HHSfQQKD7wU70+lydKVt4EghkdOvlK/NlrF90z+eXV34mUd48rNvVJXwkrMSPpCATkct8fJA==",
···
         "multiformats"
       ]
     },
+    "yocto-queue@1.2.1": {
+      "integrity": "sha512-AyeEbWOu/TAXdxlV9wmGcR0+yh2j3vYPGOECcIj2S7MkrLyC7ne+oye2BKTItt0ii2PHk4cDy+95+LshzbXnGg=="
+    },
     "zod@3.25.76": {
       "integrity": "sha512-gzUt/qt81nXsFGKIFcC3YnfEAx5NkunCfnDlvuBSSFS02bcXu4Lmea0AFIUwbLWxWPx3d9p8S5QoaujKcNQxcQ=="
     }
···
           "jsr:@puregarlic/randimal@^1.0.1",
           "npm:@atcute/atproto@^3.1.9",
           "npm:@atcute/client@^4.0.5",
+          "npm:@atcute/jetstream@^1.1.2",
           "npm:@atcute/lexicons@^1.2.2"
         ]
       },

packages/consumer/deno.jsonc

···
   "imports": {
     "@atcute/atproto": "npm:@atcute/atproto@^3.1.9",
     "@atcute/client": "npm:@atcute/client@^4.0.5",
+    "@atcute/jetstream": "npm:@atcute/jetstream@^1.1.2",
     "@atcute/lexicons": "npm:@atcute/lexicons@^1.2.2",
     "@puregarlic/randimal": "jsr:@puregarlic/randimal@^1.0.1"
   }

packages/consumer/mod.ts

···
 import { produceRequirements } from "@cistern/shared";
 import { generateKeys } from "@cistern/crypto";
 import { generateRandomName } from "@puregarlic/randimal";
-import { parse } from "@atcute/lexicons";
+import { is, parse } from "@atcute/lexicons";
+import { JetstreamSubscription } from "@atcute/jetstream";
 import type { Did } from "@atcute/lexicons/syntax";
 import type { Client, CredentialManager } from "@atcute/client";
 import {
···
   }
 
   /**
-   * Subscribes to the Jetstreams for the user's items.
-   * @todo Add `@atcute/jetstream` dependency
-   * @todo Return an async iterator
+   * Subscribes to the Jetstreams for the user's items. Pass `"stop"` into `subscription.next(...)` to cancel
+   * @todo Allow specifying Jetstream endpoint
    */
-  async getItemSubscription() {}
+  async *subscribeToItems(): AsyncIterator<
+    AppCisternLexiconItem.Main,
+    void,
+    "stop" | undefined
+  > {
+    if (!this.keypair) {
+      throw new Error("no key pair set; generate a key before subscribing");
+    }
+
+    const subscription = new JetstreamSubscription({
+      url: "wss://jetstream2.us-east.bsky.network",
+      wantedCollections: ["app.cistern.lexicon.item"],
+      wantedDids: [this.did],
+    });
+
+    for await (const event of subscription) {
+      if (event.kind === "commit" && event.commit.operation === "create") {
+        const record = event.commit.record;
+
+        if (!is(AppCisternLexiconItem.mainSchema, record)) {
+          continue;
+        }
+
+        if (record.pubkey !== this.keypair.publicKey) {
+          continue;
+        }
+
+        if ((yield record) === "stop") return;
+      }
+    }
+  }
 
   /**
    * Deletes an item from the user's PDS by record key.