Update YJS demo to use differential queries

Author: stevensJourney

demos/yjs-react-supabase-text-collab/CHANGELOG.md

@@ -1,5 +1,13 @@
 # yjs-react-supabase-text-collab
 
+## 0.2.0
+
+- Added a local development option with local Supabase and PowerSync services.
+- Updated Sync rules to use client parameters. Each client now only syncs `document` and `document_updates` for the document being edited.
+- Updated `PowerSyncYjsProvider` to use an incremental watched query for `document_updates`.
+  - Added a `editor_id` column to the `document_updates` table. This tracks which editor created the update and avoids reapplying updates in the source editor.
+  - The incremental watched query now applies updates from external editors.
+
 ## 0.1.16
 
 ### Patch Changes

demos/yjs-react-supabase-text-collab/package.json

@@ -1,6 +1,6 @@
 {
   "name": "yjs-react-supabase-text-collab",
-  "version": "0.1.16",
+  "version": "0.2.0",
   "private": true,
   "scripts": {
     "dev": "vite",

demos/yjs-react-supabase-text-collab/src/app/editor/page.tsx

@@ -1,22 +1,23 @@
-import { usePowerSync, useQuery, useStatus } from '@powersync/react';
-import { Box, Container, FormControlLabel, Switch, Typography } from '@mui/material';
-import { useEffect, useMemo, useState } from 'react';
+import { connector, useSupabase } from '@/components/providers/SystemProvider';
 import MenuBar from '@/components/widgets/MenuBar';
 import { PowerSyncYjsProvider } from '@/library/powersync/PowerSyncYjsProvider';
+import { Box, Container, FormControlLabel, Switch, Typography } from '@mui/material';
+import { usePowerSync, useQuery, useStatus } from '@powersync/react';
 import Collaboration from '@tiptap/extension-collaboration';
 import Highlight from '@tiptap/extension-highlight';
 import TaskItem from '@tiptap/extension-task-item';
 import TaskList from '@tiptap/extension-task-list';
 import { EditorContent, useEditor } from '@tiptap/react';
 import StarterKit from '@tiptap/starter-kit';
+import { useEffect, useMemo, useState } from 'react';
+import { useParams } from 'react-router-dom';
 import * as Y from 'yjs';
 import './tiptap-styles.scss';
-import { useParams } from 'react-router-dom';
-import { connector } from '@/components/providers/SystemProvider';
 
 export default function EditorPage() {
   const powerSync = usePowerSync();
   const status = useStatus();
+  const supabase = useSupabase();
   const { id: documentId } = useParams();
 
   // cache the last edited document ID in local storage
@@ -33,6 +34,12 @@ export default function EditorPage() {
 
   useEffect(() => {
     const provider = new PowerSyncYjsProvider(ydoc, powerSync, documentId!);
+    // Only sync changes for this document
+    powerSync.connect(supabase!, {
+      params: {
+        document_id: documentId!
+      }
+    });
     return () => {
       provider.destroy();
     };

demos/yjs-react-supabase-text-collab/src/components/providers/SystemProvider.tsx

@@ -1,8 +1,8 @@
 import { AppSchema } from '@/library/powersync/AppSchema';
 import { SupabaseConnector } from '@/library/powersync/SupabaseConnector';
+import { CircularProgress } from '@mui/material';
 import { PowerSyncContext } from '@powersync/react';
 import { createBaseLogger, LogLevel, PowerSyncDatabase } from '@powersync/web';
-import { CircularProgress } from '@mui/material';
 import React, { Suspense } from 'react';
 
 const SupabaseContext = React.createContext<SupabaseConnector | null>(null);
@@ -13,7 +13,6 @@ export const powerSync = new PowerSyncDatabase({
   schema: AppSchema
 });
 export const connector = new SupabaseConnector();
-powerSync.connect(connector);
 
 const logger = createBaseLogger();
 logger.useDefaults();

demos/yjs-react-supabase-text-collab/src/library/powersync/AppSchema.ts

@@ -9,7 +9,10 @@ const document_updates = new Table(
   {
     document_id: column.text,
     created_at: column.text,
-    update_b64: column.text
+    update_b64: column.text,
+    // Store an id of whom the update was created by.
+    // This is only used to not reapply updates which were created by the local editor.
+    editor_id: column.text
   },
   { indexes: { by_document: ['document_id'] } }
 );

demos/yjs-react-supabase-text-collab/src/library/powersync/PowerSyncYjsProvider.ts

@@ -1,9 +1,10 @@
 import * as Y from 'yjs';
 
 import { b64ToUint8Array, Uint8ArrayTob64 } from '@/library/binary-utils';
-import { v4 as uuidv4 } from 'uuid';
-import { AbstractPowerSyncDatabase } from '@powersync/web';
+import { AbstractPowerSyncDatabase, GetAllQuery, IncrementalWatchMode } from '@powersync/web';
 import { ObservableV2 } from 'lib0/observable';
+import { v4 as uuidv4 } from 'uuid';
+import { DocumentUpdates } from './AppSchema';
 
 export interface PowerSyncYjsEvents {
   /**
@@ -24,8 +25,9 @@ export interface PowerSyncYjsEvents {
  * @param documentId
  */
 export class PowerSyncYjsProvider extends ObservableV2<PowerSyncYjsEvents> {
-  private seenDocUpdates = new Set<string>();
   private abortController = new AbortController();
+  // This ID is updated on every new instance of the provider.
+  private id = uuidv4();
 
   constructor(
     public readonly doc: Y.Doc,
@@ -34,57 +36,72 @@ export class PowerSyncYjsProvider extends ObservableV2<PowerSyncYjsEvents> {
   ) {
     super();
 
-    const updates = db.watch('SELECT * FROM document_updates WHERE document_id = ?', [documentId], {
-      signal: this.abortController.signal
+    /**
+     * Watch for changes to the `document_updates` table for this document.
+     * This will be used to apply updates from other editors.
+     * When we received an added item we apply the update to the Yjs document.
+     */
+    const updateQuery = db.incrementalWatch({ mode: IncrementalWatchMode.DIFFERENTIAL }).build({
+      watch: {
+        query: new GetAllQuery<DocumentUpdates>({
+          sql: /* sql */ `
+            SELECT
+              *
+            FROM
+              document_updates
+            WHERE
+              document_id = ?
+              AND editor_id != ?
+          `,
+          parameters: [documentId, this.id]
+        })
+      }
     });
 
+    this.abortController.signal.addEventListener(
+      'abort',
+      () => {
+        // Stop the watch query when the abort signal is triggered
+        updateQuery.close();
+      },
+      { once: true }
+    );
+
     this._storeUpdate = this._storeUpdate.bind(this);
     this.destroy = this.destroy.bind(this);
 
     let synced = false;
 
-    const watchLoop = async () => {
-      for await (const results of updates) {
-        if (this.abortController.signal.aborted) {
-          break;
+    updateQuery.subscribe({
+      onData: async (diff) => {
+        for (const added of diff.added) {
+          Y.applyUpdateV2(doc, b64ToUint8Array(added.update_b64));
         }
-
-        // New data detected in the database
-        for (const update of results.rows!._array) {
-          // Ignore any updates we've already seen
-          if (!this.seenDocUpdates.has(update.id)) {
-            this.seenDocUpdates.add(update.id);
-            // apply the update from the database to the doc
-            const origin = this;
-            Y.applyUpdateV2(doc, b64ToUint8Array(update.update_b64), origin);
-          }
-        }
-
         if (!synced) {
           synced = true;
           this.emit('synced', []);
         }
+      },
+      onError: (error) => {
+        console.error('Error in PowerSyncYjsProvider update query:', error);
       }
-    };
-    watchLoop();
+    });
 
     doc.on('updateV2', this._storeUpdate);
     doc.on('destroy', this.destroy);
   }
 
   private async _storeUpdate(update: Uint8Array, origin: any) {
-    if (origin === this) {
-      // update originated from the database / PowerSync - ignore
-      return;
-    }
     // update originated from elsewhere - save to the database
-    const docUpdateId = uuidv4();
-    this.seenDocUpdates.add(docUpdateId);
-    await this.db.execute('INSERT INTO document_updates(id, document_id, update_b64) VALUES(?, ?, ?)', [
-      docUpdateId,
-      this.documentId,
-      Uint8ArrayTob64(update)
-    ]);
+    await this.db.execute(
+      /* sql */ `
+        INSERT INTO
+          document_updates (id, document_id, update_b64, editor_id)
+        VALUES
+          (uuid (), ?, ?, ?)
+      `,
+      [this.documentId, Uint8ArrayTob64(update), this.id]
+    );
   }
 
   /**
@@ -102,6 +119,13 @@ export class PowerSyncYjsProvider extends ObservableV2<PowerSyncYjsEvents> {
    * Also call `destroy()` to remove any event listeners and prevent future updates to the database.
    */
   async deleteData() {
-    await this.db.execute('DELETE FROM document_updates WHERE document_id = ?', [this.documentId]);
+    await this.db.execute(
+      /* sql */ `
+        DELETE FROM document_updates
+        WHERE
+          document_id = ?
+      `,
+      [this.documentId]
+    );
   }
 }

demos/yjs-react-supabase-text-collab/supabase/functions/merge-document-updates/index.ts

@@ -55,7 +55,8 @@ Deno.serve(async (req) => {
     // insert the new merged update as new single update for the document
     const supabaseInsert = await supabase.from('document_updates').insert({
       document_id: document_id,
-      update_data: Uint8ArrayToHex(docState)
+      update_data: Uint8ArrayToHex(docState),
+      editor_id: 'merged_update'
     });
     if (supabaseInsert.error) {
       throw new Error(supabaseInsert.error);

demos/yjs-react-supabase-text-collab/supabase/migrations/20250618064101_configure_powersync.sql

@@ -10,7 +10,8 @@ CREATE TABLE document_updates(
   id UUID PRIMARY KEY DEFAULT gen_random_uuid(), 
   created_at timestamptz DEFAULT now(),
   document_id UUID, 
-  update_data BYTEA
+  update_data BYTEA,
+  editor_id UUID
 );
 
 
@@ -27,11 +28,12 @@ $$ LANGUAGE SQL;
 CREATE OR REPLACE FUNCTION insert_document_updates(batch TEXT)
 RETURNS VOID AS $$
 BEGIN
-    INSERT INTO document_updates (id, document_id, update_data)
+    INSERT INTO document_updates (id, document_id, update_data, editor_id)
     SELECT
       (elem->>'id')::UUID,
       (elem->>'document_id')::UUID,
-      decode(elem->>'update_b64', 'base64')
+      decode(elem->>'update_b64', 'base64'),
+      (elem->>'editor_id')::UUID
     FROM json_array_elements(batch::json) AS elem
     ON CONFLICT (id) DO NOTHING;
 END;

demos/yjs-react-supabase-text-collab/sync-rules.yaml

@@ -3,10 +3,10 @@
 # Sync-rule docs: https://docs.powersync.com/usage/sync-rules
 bucket_definitions:
   documents:
-    data:
-      - SELECT * FROM documents
-  updates:
     # Allow remote changes to be synchronized even while there are local changes
     priority: 0
+    parameters: SELECT (request.parameters() ->> 'document_id') as document_id
     data:
-      - SELECT id, document_id, base64(update_data) as update_b64 FROM document_updates
+      - SELECT * FROM documents WHERE id = bucket.document_id
+      - SELECT * FROM documents WHERE id = bucket.document_id
+      - SELECT id, document_id, base64(update_data) as update_b64, editor_id FROM document_updates WHERE document_id = bucket.document_id