[Draft] Protocol simplification proposal

examples/chat/src/util.ts

@@ -6,7 +6,7 @@ import { Account, CoValue, ID } from "jazz-tools";
 
 export function waitForUpload(id: ID<CoValue>, me: Account) {
   const syncManager = me._raw.core.node.syncManager;
-  const peers = syncManager.getPeers();
+  const peers = me._raw.core.node.peers.getAll();
 
   return Promise.all(
     peers.map((peer) => syncManager.waitForUploadIntoPeer(peer.id, id)),

examples/pets/src/2_main.tsx

@@ -24,6 +24,7 @@ const peer =
     "peer",
   ) as `ws://${string}`) ??
   "wss://cloud.jazz.tools/?key=pets-example-jazz@garden.co";
+// "ws://localhost:4200/?key=pets-example-jazz@gcmp.io";
 
 /** Walkthrough: The top-level provider `<Jazz.Provider/>`
  *

packages/cojson-storage-indexeddb/src/tests/idbNode.test.ts

@@ -15,15 +15,15 @@ test("Should be able to initialize and load from empty DB", async () => {
     Crypto,
   );
 
-  node.syncManager.addPeer(await IDBStorage.asPeer({ trace: true }));
+  node.addPeer(await IDBStorage.asPeer({ trace: true }));
 
   console.log("yay!");
 
   const _group = node.createGroup();
 
   await new Promise((resolve) => setTimeout(resolve, 200));
 
-  expect(node.syncManager.peers["indexedDB"]).toBeDefined();
+  expect(LocalNode.peers.get("indexedDB")).toBeDefined();
 });
 
 test("Should be able to sync data to database and then load that from a new node", async () => {
@@ -35,7 +35,7 @@ test("Should be able to sync data to database and then load that from a new node
     Crypto,
   );
 
-  node1.syncManager.addPeer(
+  node1.addPeer(
     await IDBStorage.asPeer({ trace: true, localNodeName: "node1" }),
   );
 
@@ -55,14 +55,15 @@ test("Should be able to sync data to database and then load that from a new node
     Crypto,
   );
 
-  node2.syncManager.addPeer(
+  node2.addPeer(
     await IDBStorage.asPeer({ trace: true, localNodeName: "node2" }),
   );
 
   const map2 = await node2.load(map.id);
   if (map2 === "unavailable") {
     throw new Error("Map is unavailable");
   }
-
-  expect(map2.get("hello")).toBe("world");
+  // TODO fixme
+  //   expect(map2.get("hello")).toBe("world");
+  expect(false).toBeTruthy();
 });

packages/cojson-storage-sqlite/src/sqliteClient.ts

@@ -25,11 +25,9 @@ export type RawTransactionRow = {
 
 export class SQLiteClient implements DBClientInterface {
   private readonly db: DatabaseT;
-  private readonly toLocalNode: OutgoingSyncQueue;
 
-  constructor(db: DatabaseT, toLocalNode: OutgoingSyncQueue) {
+  constructor(db: DatabaseT) {
     this.db = db;
-    this.toLocalNode = toLocalNode;
   }
 
   getCoValue(coValueId: RawCoID): StoredCoValueRow | undefined {

packages/cojson-storage-sqlite/src/sqliteNode.ts

@@ -3,12 +3,45 @@ import {
   IncomingSyncStream,
   OutgoingSyncQueue,
   Peer,
+  SyncMessage,
   cojsonInternals,
 } from "cojson";
 import { SyncManager, TransactionRow } from "cojson-storage";
 import { SQLiteClient } from "./sqliteClient.js";
+import {
+  transformIncomingMessageFromPeer,
+  transformOutgoingMessageToPeer,
+} from "./transformers.js";
+
+/**
+ * This is to transform outgoing message into older protocol message(s) for backward compatibility
+ * TODO To be removed after the protocol is updated in the sync server
+ */
+class LocalNodeWrapper {
+  constructor(private queue: OutgoingSyncQueue) {}
+
+  push(msg: SyncMessage): Promise<unknown> {
+    const transformedMessages = transformOutgoingMessageToPeer(msg);
+    transformedMessages.map((transformedMessage) => {
+      // console.log("🔴 <<<=== SQLite is sending", transformedMessage);
+    });
+
+    return Promise.all(
+      transformedMessages.map((transformedMessage) => {
+        return this.queue.push(transformedMessage);
+      }),
+    );
+  }
+
+  close() {
+    return this.queue.close();
+  }
+}
 
 export class SQLiteNode {
+  // ugly public static var to be deleted after new protocol is in effect on all peers
+  public static USE_PROTOCOL2 = false;
+
   private readonly syncManager: SyncManager;
   private readonly dbClient: SQLiteClient;
 
@@ -17,16 +50,23 @@ export class SQLiteNode {
     fromLocalNode: IncomingSyncStream,
     toLocalNode: OutgoingSyncQueue,
   ) {
-    this.dbClient = new SQLiteClient(db, toLocalNode);
-    this.syncManager = new SyncManager(this.dbClient, toLocalNode);
+    this.dbClient = new SQLiteClient(db);
+    this.syncManager = new SyncManager(
+      this.dbClient,
+      new LocalNodeWrapper(toLocalNode),
+    );
 
     const processMessages = async () => {
       for await (const msg of fromLocalNode) {
         try {
           if (msg === "Disconnected" || msg === "PingTimeout") {
             throw new Error("Unexpected Disconnected message");
           }
-          await this.syncManager.handleSyncMessage(msg);
+          // console.log("🟡 <<<=== SQLite is getting", msg);
+
+          await this.syncManager.handleSyncMessage(
+            transformIncomingMessageFromPeer(msg),
+          );
         } catch (e) {
           console.error(
             new Error(

packages/cojson-storage-sqlite/src/transformers.ts

@@ -0,0 +1,83 @@
+import {
+  CojsonInternalTypes,
+  SessionID,
+  SyncMessage,
+  unknownDataMessage,
+} from "cojson";
+import CoValueContent = CojsonInternalTypes.CoValueContent;
+import { SQLiteNode } from "./sqliteNode.js";
+
+export const transformOutgoingMessageToPeer = (
+  msg: SyncMessage,
+): SyncMessage[] => {
+  if (SQLiteNode.USE_PROTOCOL2) {
+    return [msg];
+  }
+
+  const getSessionsObj = (msg: CoValueContent) =>
+    Object.entries(msg.new).reduce<{ [sessionID: SessionID]: number }>(
+      (acc, [session, content]) => {
+        acc[session as SessionID] =
+          content.after + content.newTransactions.length;
+        return acc;
+      },
+      {},
+    );
+
+  switch (msg.action) {
+    case "pull":
+      // load
+      return [{ ...msg, action: "load" }];
+    case "push":
+      // load + content
+      return [
+        {
+          action: "load",
+          id: msg.id,
+          header: true,
+          sessions: getSessionsObj(msg),
+        },
+        { ...msg, action: "content" },
+      ];
+    case "data":
+      if (!msg.known)
+        return [{ action: "known", id: msg.id, header: false, sessions: {} }];
+      // known + content => no response expected
+      return [
+        {
+          action: "known",
+          id: msg.id,
+          header: true,
+          sessions: getSessionsObj(msg),
+        },
+        { ...msg, action: "content" },
+      ];
+    case "ack":
+      // known => no response expected
+      return [{ ...msg, action: "known" }];
+    default:
+      return [msg];
+  }
+};
+
+export const transformIncomingMessageFromPeer = (
+  msg: SyncMessage,
+): SyncMessage => {
+  if (SQLiteNode.USE_PROTOCOL2) {
+    return msg;
+  }
+
+  switch (msg.action) {
+    case "load":
+      return { ...msg, action: "pull" };
+    case "content":
+      return { ...msg, action: "push" };
+    case "known":
+      if (!msg.header) return unknownDataMessage(msg.id);
+
+      if (msg.isCorrection) return { ...msg, action: "pull" };
+      return { ...msg, action: "ack" };
+    default:
+      return msg;
+  }
+};