realtime via jetstream.

Author: spuithori

src/lib/components/column/ColumnRefreshButton.svelte

@@ -16,7 +16,7 @@
     export let _agent = $agent;
     export let unique = Symbol();
     export let isJunk = false;
-    const host = _agent.agent.service.host === 'bsky.social' ? 'bsky.network' : _agent.agent.service.host;
+    const host = _agent.agent.service.host === 'bsky.social' ? 'TOKIMEKI Stream' : _agent.agent.service.host;
 
     let __columns = isJunk ? junkColumns : columns;
     let isRefreshing = false;

src/lib/components/realtime/RealtimeListenersObserver.svelte

@@ -17,7 +17,7 @@
       columns.forEach(column => {
           const _agent = $agents.get(getAccountIdByDid($agents, column.did));
           if (_agent) {
-            const host = _agent.agent.service.host === 'bsky.social' ? 'bsky.network' : _agent.agent.service.host;
+            const host = _agent.agent.service.host === 'bsky.social' ? 'TOKIMEKI Stream' : _agent.agent.service.host;
 
             if (column.settings?.autoRefresh === -1) {
               _listeners.add(host);

src/lib/realtime.ts

@@ -1,7 +1,7 @@
-import {addExtension, decode, decodeMultiple} from "cbor-x";
-import {CID} from "multiformats";
-import {CarReader} from "@ipld/car";
 import {realtime, realtimeStatuses} from '$lib/stores';
+import { PUBLIC_TOKIMEKI_STREAM_API } from '$env/static/public';
+
+const COLLECTIONS = ['app.bsky.feed.post', 'app.bsky.feed.repost', 'app.bsky.feed.like', 'app.bsky.graph.follow'];
 
 export class RealtimeClient {
     private host: string;
@@ -13,49 +13,33 @@ export class RealtimeClient {
 
     connect() {
         if (!this.socket) {
-            this.socket = new WebSocket('wss://' + this.host + '/xrpc/com.atproto.sync.subscribeRepos');
+            this.socket = new WebSocket(`${PUBLIC_TOKIMEKI_STREAM_API}/subscribe?${COLLECTIONS.map(item => `wantedCollections=${item}`).join('&')}`);
         }
 
-        addExtension({
-            Class: CID,
-            tag: 42,
-            encode: () => {
-                throw new Error('Cannot encode cids');
-            },
-            decode: (bytes) => {
-                if (bytes[0] !== 0) {
-                    throw new Error('Invalid cid');
-                }
-                return CID.decode(bytes.subarray(1)); // ignore leading 0x00
-            },
-        });
-
         this.socket.onmessage = async function (event) {
-            const messageBuf = await event.data.arrayBuffer();
-            const [header, body] = decodeMultiple(new Uint8Array(messageBuf));
-
-            if (header.op !== 1) {
+            if (!event.data) {
                 return;
             }
 
-            try {
-                const car = await CarReader.fromBytes(body.blocks);
-
-                for (const op of body.ops) {
-                    if (!op.cid) continue;
-                    const block = await car.get(op.cid);
-                    const record = decode(block.bytes);
-                    realtime.set({
-                        isConnected: true,
-                        data: {
-                            record: record,
-                            op: op,
-                            body: body,
-                        }
-                    })
-                }
-            } catch (e) {
-                // do nothing.
+            const dataWrapper = JSON.parse(event.data);
+            const data = dataWrapper.commit;
+
+            if (data?.type === 'c') {
+                const record = data.record;
+
+                realtime.set({
+                    isConnected: true,
+                    data: {
+                        record: record,
+                        op: {
+                            path: data.rkey,
+                            collection: data.collection,
+                        },
+                        body: {
+                            repo: dataWrapper.did,
+                        },
+                    }
+                })
             }
         };
 
@@ -72,6 +56,10 @@ export class RealtimeClient {
                 return  [...r,  this.host];
             })
         }
+
+        this.socket.onerror = async (event) => {
+            console.log(event)
+        }
     }
 
     disconnect() {
@@ -129,7 +117,7 @@ async function getRecord(_agent, uri, repost = undefined, retryCount = 0) {
 export async function getPostRealtime(realtime, actors, _agent) {
     const path = realtime.data.op.path;
     const repo = realtime.data.body.repo;
-    const uri = 'at://' + repo + '/' + path;
+    const uri = 'at://' + repo + '/' + realtime.data.op.collection + '/' + path;
     const isStream: boolean = actors.some(actor => actor === repo);
 
     if (realtime.data.record.$type === 'app.bsky.feed.post' && typeof realtime.data.record.text === 'string') {