refactor: add server as storage to sync engine (#4475)

Another layer in sync engine is composable storage.

Now you can specify a chain of storages. Each can listen transactions to
update some state. Though subscription for state is invoked one by one
until non-undefined is provided.

See tests to understand API.

Here I also migrated our server data loading and synchronization to this
storage api.

Author: TrySound

apps/builder/app/builder/builder.tsx

@@ -7,12 +7,12 @@ import type { Project } from "@webstudio-is/project";
 import { theme, Box, type CSS, Flex, Grid } from "@webstudio-is/design-system";
 import type { AuthPermit } from "@webstudio-is/trpc-interface/index.server";
 import { createImageLoader } from "@webstudio-is/image";
+import { registerContainers, createObjectPool } from "~/shared/sync";
 import {
-  builderClient,
-  registerContainers,
-  useBuilderStore,
-} from "~/shared/sync";
-import { startProjectSync, useSyncServer } from "./shared/sync/sync-server";
+  ServerSyncStorage,
+  startProjectSync,
+  useSyncServer,
+} from "./shared/sync/sync-server";
 import { SidebarLeft } from "./sidebar-left";
 import { Inspector } from "./features/inspector";
 import { Topbar } from "./features/topbar";
@@ -57,7 +57,6 @@ import {
 import { CloneProjectDialog } from "~/shared/clone-project";
 import type { TokenPermissions } from "@webstudio-is/authorization-token";
 import { useToastErrors } from "~/shared/error/toast-error";
-import { loadBuilderData, setBuilderData } from "~/shared/builder-data";
 import { initBuilderApi } from "~/shared/builder-api";
 import { updateWebstudioData } from "~/shared/instance-utils";
 import { migrateWebstudioDataMutable } from "~/shared/webstudio-data-migrator";
@@ -71,6 +70,7 @@ import {
 } from "~/shared/copy-paste/init-copy-paste";
 import { useInertHandlers } from "./shared/inert-handlers";
 import { TextToolbar } from "./features/workspace/canvas-tools/text-toolbar";
+import { SyncClient } from "~/shared/sync-client";
 
 registerContainers();
 
@@ -206,6 +206,12 @@ const ChromeWrapper = ({
   );
 };
 
+const builderClient = new SyncClient({
+  role: "leader",
+  object: createObjectPool(),
+  storages: [new ServerSyncStorage()],
+});
+
 export type BuilderProps = {
   project: Project;
   publisherHost: string;
@@ -242,9 +248,9 @@ export const Builder = ({
     const controller = new AbortController();
 
     $dataLoadingState.set("loading");
-    loadBuilderData({ projectId: project.id, signal: controller.signal })
-      .then((data) => {
-        setBuilderData(data);
+    builderClient.connect({
+      signal: controller.signal,
+      onReady() {
         startProjectSync({
           projectId: project.id,
           buildId: build.id,
@@ -260,12 +266,10 @@ export const Builder = ({
         // so builder is started listening for connect event
         // when canvas is rendered
         $dataLoadingState.set("loaded");
-      })
-      .catch((error) => {
-        console.error(error);
+
         // @todo make needs error handling and error state? e.g. a toast
-        $dataLoadingState.set("idle");
-      });
+      },
+    });
     return () => {
       $dataLoadingState.set("idle");
       controller.abort("unmount");
@@ -287,7 +291,6 @@ export const Builder = ({
     $publisher.set({ publish });
   }, [publish]);
 
-  useBuilderStore();
   useSyncServer({
     projectId: project.id,
     authPermit,

apps/builder/app/builder/shared/sync/command-queue.ts

@@ -1,12 +1,13 @@
 import type { Project } from "@webstudio-is/project";
 import type { Build } from "@webstudio-is/project-build";
-import type { SyncItem } from "immerhin";
+import type { Change } from "immerhin";
+import type { Transaction } from "~/shared/sync-client";
 
 type Command =
   | {
       type: "transactions";
       projectId: Project["id"];
-      transactions: SyncItem[];
+      transactions: Transaction<Change[]>[];
     }
   | {
       type: "setDetails";

apps/builder/app/builder/shared/sync/sync-server.ts

@@ -1,13 +1,16 @@
 import { useEffect } from "react";
 import { atom } from "nanostores";
+import type { Change } from "immerhin";
 import type { Project } from "@webstudio-is/project";
 import type { Build } from "@webstudio-is/project-build";
 import type { AuthPermit } from "@webstudio-is/trpc-interface/index.server";
 import * as commandQueue from "./command-queue";
 import { restPatchPath } from "~/shared/router-utils";
 import { toast } from "@webstudio-is/design-system";
-import { serverSyncStore } from "~/shared/sync";
 import { fetch } from "~/shared/fetch.client";
+import type { SyncStorage, Transaction } from "~/shared/sync-client";
+import { $project } from "~/shared/nano-states";
+import { loadBuilderData } from "~/shared/builder-data";
 
 // Periodic check for new entries to group them into one job/call in sync queue.
 const NEW_ENTRIES_INTERVAL = 1000;
@@ -284,27 +287,30 @@ const useSyncProject = ({
       return;
     }
     syncServer();
-
-    const updateProjectTransactions = () => {
-      const transactions = serverSyncStore.popAll();
-      if (transactions.length === 0) {
-        return;
-      }
-      commandQueue.enqueue({ type: "transactions", transactions, projectId });
-    };
-
-    const intervalHandle = setInterval(
-      updateProjectTransactions,
-      NEW_ENTRIES_INTERVAL
-    );
-
-    return () => {
-      updateProjectTransactions();
-      clearInterval(intervalHandle);
-    };
   }, [projectId, authPermit]);
 };
 
+export class ServerSyncStorage implements SyncStorage {
+  name = "server";
+  sendTransaction(transaction: Transaction<Change[]>) {
+    if (transaction.object === "server") {
+      const projectId = $project.get()?.id ?? "";
+      commandQueue.enqueue({
+        type: "transactions",
+        transactions: [transaction],
+        projectId,
+      });
+    }
+  }
+  subscribe(setState: (state: unknown) => void, signal: AbortSignal) {
+    const projectId = $project.get()?.id ?? "";
+    loadBuilderData({ projectId, signal }).then((data) => {
+      const serverData = new Map(Object.entries(data));
+      setState(new Map([["server", serverData]]));
+    });
+  }
+}
+
 type SyncServerProps = {
   projectId: Project["id"];
   authPermit: AuthPermit;

apps/builder/app/routes/rest.patch.ts

@@ -1,6 +1,6 @@
 import { applyPatches, enableMapSet, enablePatches } from "immer";
 import type { ActionFunctionArgs } from "@remix-run/server-runtime";
-import type { SyncItem } from "immerhin";
+import type { Change } from "immerhin";
 import {
   Breakpoints,
   Breakpoint,
@@ -49,9 +49,10 @@ import type { Database } from "@webstudio-is/postrest/index.server";
 import { publicStaticEnv } from "~/env/env.static";
 import { preventCrossOriginCookie } from "~/services/no-cross-origin-cookie";
 import { checkCsrf } from "~/services/csrf-session.server";
+import type { Transaction } from "~/shared/sync-client";
 
 type PatchData = {
-  transactions: Array<SyncItem>;
+  transactions: Transaction<Change[]>[];
   buildId: Build["id"];
   projectId: Project["id"];
   version: number;
@@ -96,7 +97,7 @@ export const action = async ({
       return { status: "error", errors: "Project id required" };
     }
 
-    const lastTransactionId = transactions.at(-1)?.transactionId;
+    const lastTransactionId = transactions.at(-1)?.id;
 
     if (lastTransactionId === undefined) {
       return {
@@ -175,7 +176,7 @@ export const action = async ({
     const patchedStyleDeclKeysSet = new Set<string>();
 
     for await (const transaction of transactions) {
-      for await (const change of transaction.changes) {
+      for await (const change of transaction.payload) {
         const { namespace, patches } = change;
         if (patches.length === 0) {
           continue;

apps/builder/app/shared/builder-data.ts

@@ -40,21 +40,6 @@ export const getBuilderData = (): BuilderData => {
   };
 };
 
-export const setBuilderData = (data: BuilderData) => {
-  $assets.set(data.assets);
-  $instances.set(data.instances);
-  $dataSources.set(data.dataSources);
-  $resources.set(data.resources);
-  // props should be after data sources to compute logic
-  $props.set(data.props);
-  $pages.set(data.pages);
-  $styleSources.set(data.styleSources);
-  $styleSourceSelections.set(data.styleSourceSelections);
-  $breakpoints.set(data.breakpoints);
-  $styles.set(data.styles);
-  $marketplaceProduct.set(data.marketplaceProduct);
-};
-
 const getPair = <Item extends { id: string }>(item: Item) =>
   [item.id, item] as const;
 

apps/builder/app/shared/sync-client.test.ts

@@ -8,6 +8,7 @@ import {
   NanostoresSyncObject,
   SyncClient,
   SyncObjectPool,
+  type SyncStorage,
 } from "./sync-client";
 
 enableMapSet();
@@ -217,6 +218,18 @@ test("support pool of objects", () => {
   );
 });
 
+test("merge state in pool object to partially restore from storage", () => {
+  const store1 = atom(0);
+  const store2 = atom(1);
+  const objectPool = new SyncObjectPool([
+    new NanostoresSyncObject("store1", store1),
+    new NanostoresSyncObject("store2", store2),
+  ]);
+  objectPool.setState(new Map([["store1", 2]]));
+  expect(store1.get()).toEqual(2);
+  expect(store2.get()).toEqual(1);
+});
+
 describe("nanostores sync object", () => {
   test("sync initial state and exchange transactions", () => {
     const emitter = createNanoEvents();
@@ -290,3 +303,79 @@ describe("nanostores sync object", () => {
     expect(sendTransaction).toBeCalledTimes(0);
   });
 });
+
+describe("storages", () => {
+  class TestStorage implements SyncStorage {
+    name = "TestStorage";
+    value: undefined | number;
+    constructor(value: undefined | number) {
+      this.value = value;
+    }
+    sendTransaction = vi.fn();
+    subscribe(setState: (state: unknown) => void) {
+      setState(this.value);
+    }
+  }
+
+  test("get initial state from storage", () => {
+    const $store = atom(0);
+    const client = new SyncClient({
+      role: "leader",
+      object: new NanostoresSyncObject("nanostores", $store),
+      storages: [new TestStorage(1)],
+    });
+    client.connect({ signal: new AbortController().signal });
+    expect($store.get()).toEqual(1);
+  });
+
+  test("fallback to current state when storage is empty", () => {
+    const $store = atom(0);
+    const client = new SyncClient({
+      role: "leader",
+      object: new NanostoresSyncObject("nanostores", $store),
+      storages: [new TestStorage(undefined)],
+    });
+    client.connect({ signal: new AbortController().signal });
+    expect($store.get()).toEqual(0);
+  });
+
+  test("get state from the first non-empty storage", () => {
+    const $store = atom(0);
+    const client = new SyncClient({
+      role: "leader",
+      object: new NanostoresSyncObject("nanostores", $store),
+      storages: [
+        new TestStorage(undefined),
+        new TestStorage(1),
+        new TestStorage(2),
+      ],
+    });
+    client.connect({ signal: new AbortController().signal });
+    expect($store.get()).toEqual(1);
+  });
+
+  test("send transactions to all provided storages", () => {
+    const $store = atom(0);
+    const storage1 = new TestStorage(undefined);
+    const storage2 = new TestStorage(undefined);
+    const client = new SyncClient({
+      role: "leader",
+      object: new NanostoresSyncObject("nanostores", $store),
+      storages: [storage1, storage2],
+    });
+    client.connect({ signal: new AbortController().signal });
+    $store.set(1);
+    expect(storage1.sendTransaction).toBeCalledTimes(1);
+    expect(storage1.sendTransaction).toHaveBeenCalledWith({
+      id: expect.any(String),
+      object: "nanostores",
+      payload: 1,
+    });
+    expect(storage2.sendTransaction).toBeCalledTimes(1);
+    expect(storage2.sendTransaction).toHaveBeenCalledWith({
+      id: expect.any(String),
+      object: "nanostores",
+      payload: 1,
+    });
+  });
+});