Support yjs in dexie-cloud-addon

.github/workflows/main.yml

@@ -47,5 +47,5 @@ jobs:
       - name: Run headless test
         uses: coactions/setup-xvfb@v1
         with:
-          run: bash ./gh-actions.sh
+          run: bash -e ./gh-actions.sh
           working-directory: ${{ matrix.TF }}

addons/Dexie.Observable/package.json

@@ -81,11 +81,11 @@
   },
   "homepage": "https://dexie.org",
   "peerDependencies": {
-    "dexie": "workspace:^3.0.2 || ^4.0.1"
+    "dexie": "workspace:^"
   },
   "devDependencies": {
     "@types/node": "^18.11.18",
-    "dexie": "workspace:^3.0.2 || ^4.0.1",
+    "dexie": "workspace:^",
     "eslint": "^7.27.0",
     "just-build": "^0.9.24",
     "qunit": "^2.9.2",

addons/Dexie.Syncable/package.json

@@ -74,11 +74,11 @@
   },
   "homepage": "https://dexie.org",
   "peerDependencies": {
-    "dexie": "workspace:^3.0.2 || ^4.0.1",
+    "dexie": "workspace:^",
     "dexie-observable": "workspace:^"
   },
   "devDependencies": {
-    "dexie": "workspace:^3.0.2 || ^4.0.1",
+    "dexie": "workspace:^",
     "dexie-observable": "workspace:^",
     "eslint": "^5.16.0",
     "just-build": "^0.9.24",

addons/dexie-cloud/package.json

@@ -1,6 +1,6 @@
 {
   "name": "dexie-cloud-addon",
-  "version": "4.0.8",
+  "version": "4.1.0-alpha.23",
   "description": "Dexie addon that syncs with to Dexie Cloud",
   "main": "dist/umd/dexie-cloud-addon.js",
   "type": "module",
@@ -78,23 +78,24 @@
     "@rollup/plugin-terser": "^0.4.4",
     "@rollup/plugin-typescript": "^11.1.5",
     "@types/node": "^18.11.18",
+    "dexie-cloud-common": "workspace:^",
     "dreambase-library": "^1.0.21",
     "just-build": "*",
     "karma": "*",
     "karma-chrome-launcher": "*",
     "karma-firefox-launcher": "*",
     "karma-qunit": "*",
+    "lib0": "^0.2.97",
     "preact": "*",
     "rollup": "^4.1.4",
     "terser": "^5.20.0",
     "tslib": "*",
-    "typescript": "^5.3.3"
+    "typescript": "^5.6.3"
   },
   "dependencies": {
-    "dexie-cloud-common": "^1.0.31",
     "rxjs": "^7.x"
   },
   "peerDependencies": {
-    "dexie": "^4.0.1"
+    "dexie": "workspace:^"
   }
 }

addons/dexie-cloud/src/DexieCloudOptions.ts

@@ -29,6 +29,10 @@ export interface DexieCloudOptions {
   // not be synced with Dexie Cloud
   unsyncedTables?: string[];
 
+  unsyncedProperties?: {
+    [tableName: string]: string[];
+  }
+
   // By default Dexie Cloud will suffix the cloud DB ID to your IndexedDB database name
   // in order to ensure that the local database is uniquely tied to the remote one and
   // will use another local database if databaseURL is changed or if dexieCloud addon
@@ -52,4 +56,6 @@ export interface DexieCloudOptions {
     public_key: string;
     hints?: { userId?: string; email?: string };
   }) => Promise<TokenFinalResponse>;
+
+  awarenessProtocol?: typeof import('y-protocols/awareness');
 }

addons/dexie-cloud/src/PermissionChecker.ts

@@ -42,6 +42,7 @@ export class PermissionChecker<T, TableNames extends string = TableName<T>> {
     // If user can update any prop in any table in this realm, return true unless
     // it regards to ownership change:
     if (this.permissions.update === '*') {
+      // @ts-ignore
       return props.every((prop) => prop !== 'owner');
     }
     const tablePermissions = this.permissions.update?.[this.tableName];

addons/dexie-cloud/src/WSObservable.ts

@@ -1,14 +1,24 @@
 import { DBOperationsSet } from 'dexie-cloud-common';
-import { BehaviorSubject, Observable, Subscriber, Subscription } from 'rxjs';
+import { BehaviorSubject, Observable, Subscriber, Subscription, tap } from 'rxjs';
 import { TokenExpiredError } from './authentication/TokenExpiredError';
 import { DXCWebSocketStatus } from './DXCWebSocketStatus';
 import { TSON } from './TSON';
+import type { YClientMessage, YServerMessage } from 'dexie-cloud-common';
+import { DexieCloudDB } from './db/DexieCloudDB';
+import { createYClientUpdateObservable } from './yjs/createYClientUpdateObservable';
+import { applyYServerMessages } from './yjs/applyYMessages';
+import { DexieYProvider } from 'dexie';
+import { getAwarenessLibrary, getDocAwareness } from './yjs/awareness';
+import { encodeYMessage, decodeYMessage } from 'dexie-cloud-common';
+import { UserLogin } from './dexie-cloud-client';
+import { isEagerSyncDisabled } from './isEagerSyncDisabled';
+import { getOpenDocSignal } from './yjs/reopenDocSignal';
 
 const SERVER_PING_TIMEOUT = 20000;
 const CLIENT_PING_INTERVAL = 30000;
 const FAIL_RETRY_WAIT_TIME = 60000;
 
-export type WSClientToServerMsg = ReadyForChangesMessage;
+export type WSClientToServerMsg = ReadyForChangesMessage | YClientMessage;
 export interface ReadyForChangesMessage {
   type: 'ready';
   realmSetHash: string;
@@ -73,24 +83,22 @@ export interface TokenExpiredMessage {
 
 export class WSObservable extends Observable<WSConnectionMsg> {
   constructor(
-    databaseUrl: string,
+    db: DexieCloudDB,
     rev: string,
     realmSetHash: string,
     clientIdentity: string,
     messageProducer: Observable<WSClientToServerMsg>,
     webSocketStatus: BehaviorSubject<DXCWebSocketStatus>,
-    token?: string,
-    tokenExpiration?: Date
+    user: UserLogin
   ) {
     super(
       (subscriber) =>
         new WSConnection(
-          databaseUrl,
+          db,
           rev,
           realmSetHash,
           clientIdentity,
-          token,
-          tokenExpiration,
+          user,
           subscriber,
           messageProducer,
           webSocketStatus
@@ -102,6 +110,7 @@ export class WSObservable extends Observable<WSConnectionMsg> {
 let counter = 0;
 
 export class WSConnection extends Subscription {
+  db: DexieCloudDB;
   ws: WebSocket | null;
   lastServerActivity: Date;
   lastUserActivity: Date;
@@ -110,24 +119,22 @@ export class WSConnection extends Subscription {
   rev: string;
   realmSetHash: string;
   clientIdentity: string;
-  token: string | undefined;
-  tokenExpiration: Date | undefined;
+  user: UserLogin;
   subscriber: Subscriber<WSConnectionMsg>;
   pauseUntil?: Date;
   messageProducer: Observable<WSClientToServerMsg>;
   webSocketStatus: BehaviorSubject<DXCWebSocketStatus>;
   id = ++counter;
 
   private pinger: any;
-  private messageProducerSubscription: null | Subscription;
+  private subscriptions: Set<Subscription> = new Set();
 
   constructor(
-    databaseUrl: string,
+    db: DexieCloudDB,
     rev: string,
     realmSetHash: string,
     clientIdentity: string,
-    token: string | undefined,
-    tokenExpiration: Date | undefined,
+    user: UserLogin,
     subscriber: Subscriber<WSConnectionMsg>,
     messageProducer: Observable<WSClientToServerMsg>,
     webSocketStatus: BehaviorSubject<DXCWebSocketStatus>
@@ -136,18 +143,17 @@ export class WSConnection extends Subscription {
     console.debug(
       'New WebSocket Connection',
       this.id,
-      token ? 'authorized' : 'unauthorized'
+      user.accessToken ? 'authorized' : 'unauthorized'
     );
-    this.databaseUrl = databaseUrl;
+    this.db = db;
+    this.databaseUrl = db.cloud.options!.databaseUrl;
     this.rev = rev;
     this.realmSetHash = realmSetHash;
     this.clientIdentity = clientIdentity;
-    this.token = token;
-    this.tokenExpiration = tokenExpiration;
+    this.user = user;
     this.subscriber = subscriber;
     this.lastUserActivity = new Date();
     this.messageProducer = messageProducer;
-    this.messageProducerSubscription = null;
     this.webSocketStatus = webSocketStatus;
     this.connect();
   }
@@ -169,10 +175,10 @@ export class WSConnection extends Subscription {
       } catch {}
     }
     this.ws = null;
-    if (this.messageProducerSubscription) {
-      this.messageProducerSubscription.unsubscribe();
-      this.messageProducerSubscription = null;
+    for (const sub of this.subscriptions) {
+      sub.unsubscribe();
     }
+    this.subscriptions.clear();
   }
 
   reconnecting = false;
@@ -205,7 +211,8 @@ export class WSConnection extends Subscription {
       //console.debug('SyncStatus: DUBB: Ooops it was closed!');
       return;
     }
-    if (this.tokenExpiration && this.tokenExpiration < new Date()) {
+    const tokenExpiration = this.user.accessTokenExpiration;
+    if (tokenExpiration && tokenExpiration < new Date()) {
       this.subscriber.error(new TokenExpiredError()); // Will be handled in connectWebSocket.ts.
       return;
     }
@@ -266,14 +273,14 @@ export class WSConnection extends Subscription {
     searchParams.set('rev', this.rev);
     searchParams.set('realmsHash', this.realmSetHash);
     searchParams.set('clientId', this.clientIdentity);
-    if (this.token) {
-      searchParams.set('token', this.token);
+    if (this.user.accessToken) {
+      searchParams.set('token', this.user.accessToken);
     }
 
     // Connect the WebSocket to given url:
     console.debug('dexie-cloud WebSocket create');
     const ws = (this.ws = new WebSocket(`${wsUrl}/changes?${searchParams}`));
-    //ws.binaryType = "arraybuffer"; // For future when subscribing to actual changes.
+    ws.binaryType = "arraybuffer";
 
     ws.onclose = (event: Event) => {
       if (!this.pinger) return;
@@ -283,21 +290,47 @@ export class WSConnection extends Subscription {
 
     ws.onmessage = (event: MessageEvent) => {
       if (!this.pinger) return;
-      console.debug('dexie-cloud WebSocket onmessage', event.data);
 
       this.lastServerActivity = new Date();
       try {
-        const msg = TSON.parse(event.data) as
-          | WSConnectionMsg
-          | PongMessage
-          | ErrorMessage;
+        const msg = typeof event.data === 'string'
+          ? TSON.parse(event.data) as
+            | WSConnectionMsg
+            | PongMessage
+            | ErrorMessage
+            | YServerMessage   
+          : decodeYMessage(new Uint8Array(event.data)) as
+            | YServerMessage;
+        console.debug('dexie-cloud WebSocket onmessage', msg.type, msg);
         if (msg.type === 'error') {
           throw new Error(`Error message from dexie-cloud: ${msg.error}`);
-        }
-        if (msg.type === 'rev') {
-          this.rev = msg.rev; // No meaning but seems reasonable.
-        }
-        if (msg.type !== 'pong') {
+        } else if (msg.type === 'aware') {
+          const docCache = DexieYProvider.getDocCache(this.db.dx);
+          const doc = docCache.find(msg.table, msg.k, msg.prop);
+          if (doc) {
+            const awareness = getDocAwareness(doc);
+            if (awareness) {
+              const awap = getAwarenessLibrary(this.db);
+              awap.applyAwarenessUpdate(
+                awareness,
+                msg.u,
+                'server',
+              );
+            }
+          }
+        } else if (msg.type === 'u-ack' || msg.type === 'u-reject' || msg.type === 'u-s' || msg.type === 'in-sync') {
+          applyYServerMessages([msg], this.db);
+        } else if (msg.type === 'doc-open') {
+          const docCache = DexieYProvider.getDocCache(this.db.dx);
+          const doc = docCache.find(msg.table, msg.k, msg.prop);
+          if (doc) {
+            getOpenDocSignal(doc).next(); // Make yHandler reopen the document on server.
+          }
+        } else if (msg.type === 'outdated-server-rev' || msg.type === 'y-complete-sync-done') {
+          // Won't happen but need this for typing.
+          throw new Error('Outdated server revision or y-complete-sync-done not expected over WebSocket - only in sync using fetch()');
+        } else if (msg.type !== 'pong') {
+          // Forward the request to our subscriber, wich is in messageFromServerQueue.ts (via connectWebSocket's subscribe() at the end!)
           this.subscriber.next(msg);
         }
       } catch (e) {
@@ -324,7 +357,7 @@ export class WSConnection extends Subscription {
           }
         };
       });
-      this.messageProducerSubscription = this.messageProducer.subscribe(
+      this.subscriptions.add(this.messageProducer.subscribe(
         (msg) => {
           if (!this.closed) {
             if (
@@ -333,10 +366,28 @@ export class WSConnection extends Subscription {
             ) {
               this.webSocketStatus.next('connected');
             }
-            this.ws?.send(TSON.stringify(msg));
+            console.debug('dexie-cloud WebSocket send', msg.type, msg);
+            if (msg.type === 'ready') {
+              // Ok, we are certain to have stored everything up until revision msg.rev.
+              // Update this.rev in case of reconnect - remember where we were and don't just start over!
+              this.rev = msg.rev; 
+              // ... and then send along the request to the server so it would also be updated!
+              this.ws?.send(TSON.stringify(msg));
+            } else {
+              // If it's not a "ready" message, it's an YMessage.
+              // YMessages can be sent binary encoded.
+              this.ws?.send(encodeYMessage(msg));
+            }
           }
         }
-      );
+      ));
+      if (this.user.isLoggedIn && !isEagerSyncDisabled(this.db)) {
+        this.subscriptions.add(
+          createYClientUpdateObservable(this.db).subscribe(
+            this.db.messageProducer
+          )
+        );
+      }
     } catch (error) {
       this.pauseUntil = new Date(Date.now() + FAIL_RETRY_WAIT_TIME);
     }

addons/dexie-cloud/src/db/DexieCloudDB.ts

@@ -15,6 +15,7 @@ import {
 import { BroadcastedAndLocalEvent } from '../helpers/BroadcastedAndLocalEvent';
 import { SyncState, SyncStatePhase } from '../types/SyncState';
 import { MessagesFromServerConsumer } from '../sync/messagesFromServerQueue';
+import { YClientMessage } from 'dexie-cloud-common';
 
 /*export interface DexieCloudDB extends Dexie {
   table(name: string): Table<any, any>;
@@ -66,6 +67,7 @@ export interface DexieCloudDB extends DexieCloudDBBase {
   setInitiallySynced(initiallySynced: boolean): void;
   reconfigure(): void;
   messageConsumer: MessagesFromServerConsumer;
+  messageProducer: Subject<YClientMessage>;
 }
 
 const wm = new WeakMap<object, DexieCloudDB>();
@@ -193,6 +195,7 @@ export function DexieCloudDB(dx: Dexie): DexieCloudDB {
 
     Object.assign(db, helperMethods);
     db.messageConsumer = MessagesFromServerConsumer(db);
+    db.messageProducer = new Subject<YClientMessage>();
     wm.set(dx.cloud, db);
   }
   return db;

addons/dexie-cloud/src/db/entities/PersistedSyncState.ts

@@ -1,5 +1,6 @@
 export interface PersistedSyncState {
   serverRevision?: any;
+  yServerRevision?: string;
   latestRevisions: {
     [tableName: string]: number
   };
@@ -11,4 +12,11 @@ export interface PersistedSyncState {
   syncedTables: string[];
   timestamp?: Date;
   error?: string;
+  yDownloadedRealms?: {
+    [realmId: string]: "*" | {
+      tbl: string;
+      prop: string;
+      key: any;
+    }
+  }
 }