live query web worker syncing

SandroMaglione · SandroMaglione · commit 3773545d39b2 · 2025-02-28T06:40:17.000+01:00
diff --git a/apps/client/src/lib/runtime-client.ts b/apps/client/src/lib/runtime-client.ts
@@ -2,6 +2,7 @@ import { Layer, ManagedRuntime } from "effect";
 import { ApiClient } from "./api-client";
 import { Dexie } from "./dexie";
 import { LoroStorage } from "./services/loro-storage";
+import { Sync } from "./services/sync";
 import { TempWorkspace } from "./services/temp-workspace";
 import { WorkspaceManager } from "./services/workspace-manager";
 
@@ -10,7 +11,8 @@ const MainLayer = Layer.mergeAll(
   ApiClient.Default,
   WorkspaceManager.Default,
   TempWorkspace.Default,
-  LoroStorage.Default
+  LoroStorage.Default,
+  Sync.Default
 );
 
 export const RuntimeClient = ManagedRuntime.make(MainLayer);
diff --git a/apps/client/src/lib/services/sync.ts b/apps/client/src/lib/services/sync.ts
@@ -0,0 +1,86 @@
+import { Effect } from "effect";
+import { ApiClient } from "../api-client";
+import { Dexie } from "../dexie";
+import type { WorkspaceTable } from "../schema";
+import { TempWorkspace } from "./temp-workspace";
+import { WorkspaceManager } from "./workspace-manager";
+import { LoroDoc } from "loro-crdt";
+
+export class Sync extends Effect.Service<Sync>()("Sync", {
+  dependencies: [
+    TempWorkspace.Default,
+    WorkspaceManager.Default,
+    ApiClient.Default,
+    Dexie.Default,
+  ],
+  effect: Effect.gen(function* () {
+    const { client } = yield* ApiClient;
+    const { initClient } = yield* Dexie;
+    const manager = yield* WorkspaceManager;
+    const temp = yield* TempWorkspace;
+
+    return {
+      push: ({
+        snapshot,
+        workspace,
+      }: {
+        workspace: WorkspaceTable;
+        snapshot: globalThis.Uint8Array;
+      }) =>
+        Effect.gen(function* () {
+          const clientId = yield* initClient;
+
+          const response = yield* Effect.fromNullable(workspace.token).pipe(
+            Effect.flatMap((token) =>
+              client.syncData
+                .push({
+                  // headers: { Authorization: `Bearer ${token}` },
+                  path: {
+                    workspaceId: workspace.workspaceId,
+                  },
+                  payload: { clientId, snapshot },
+                })
+                .pipe(
+                  Effect.map((response) => ({
+                    ...response,
+                    token,
+                  }))
+                )
+            ),
+            Effect.orElse(() =>
+              client.syncAuth
+                .generateToken({
+                  payload: {
+                    clientId,
+                    snapshot,
+                    workspaceId: workspace.workspaceId,
+                  },
+                })
+                .pipe(
+                  Effect.tap(({ token }) =>
+                    manager.setToken({
+                      workspaceId: workspace.workspaceId,
+                      token,
+                    })
+                  )
+                )
+            )
+          );
+
+          const doc = new LoroDoc();
+          doc.import(response.snapshot);
+          yield* Effect.all([
+            manager.put({
+              workspaceId: response.workspaceId,
+              snapshot: response.snapshot,
+              token: response.token,
+              version: doc.version().encode(),
+            }),
+            temp.clean({
+              workspaceId: workspace.workspaceId,
+            }),
+          ]);
+        }),
+    };
+  }),
+}) {}
diff --git a/apps/client/src/routes/$workspaceId.tsx b/apps/client/src/routes/$workspaceId.tsx
@@ -2,13 +2,13 @@ import { Worker } from "@effect/platform";
 import { BrowserWorker } from "@effect/platform-browser";
 import { createFileRoute } from "@tanstack/react-router";
 import { Effect } from "effect";
-import { startTransition } from "react";
+import { startTransition, useEffect } from "react";
 import { useActivity } from "../lib/hooks/use-activity";
 import { RuntimeClient } from "../lib/runtime-client";
 import { LoroStorage } from "../lib/services/loro-storage";
 import { WorkspaceManager } from "../lib/services/workspace-manager";
 import { useActionEffect } from "../lib/use-action-effect";
-import { Bootstrap } from "../workers/schema";
+import { Bootstrap, LiveQuery } from "../workers/schema";
 
 export const Route = createFileRoute("/$workspaceId")({
   component: RouteComponent,
@@ -62,6 +62,31 @@ function RouteComponent() {
     })
   );
 
+  useEffect(() => {
+    const url = new URL("./src/workers/live.ts", globalThis.origin);
+    const newWorker = new globalThis.Worker(url, { type: "module" });
+
+    void RuntimeClient.runPromise(
+      Effect.gen(function* () {
+        const pool = yield* Worker.makePoolSerialized({ size: 1 });
+        return yield* pool.broadcast(
+          new LiveQuery({ workspaceId: workspace.workspaceId })
+        );
+      }).pipe(
+        Effect.scoped,
+        Effect.provide(BrowserWorker.layer(() => newWorker))
+      )
+    );
+
+    newWorker.onerror = (error) => {
+      console.error("Live query worker error", error);
+    };
+
+    return () => {
+      newWorker.terminate();
+    };
+  }, []);
+
   return (
     <div>
       <p>{workspace.workspaceId}</p>
diff --git a/apps/client/src/workers/live.ts b/apps/client/src/workers/live.ts
@@ -0,0 +1,111 @@
+import { WorkerRunner } from "@effect/platform";
+import { BrowserWorkerRunner } from "@effect/platform-browser";
+import { Snapshot } from "@local/sync/loro";
+import { liveQuery } from "dexie";
+import {
+  Array,
+  Effect,
+  Layer,
+  Number,
+  Schema,
+  Stream,
+  SynchronizedRef,
+} from "effect";
+import { Dexie } from "../lib/dexie";
+import { RuntimeClient } from "../lib/runtime-client";
+import { Sync } from "../lib/services/sync";
+import { WorkspaceManager } from "../lib/services/workspace-manager";
+import { type LiveQuery } from "./schema";
+
+const main = (params: { workspaceId: string }) =>
+  Effect.gen(function* () {
+    const manager = yield* WorkspaceManager;
+    const { db } = yield* Dexie;
+    const { push } = yield* Sync;
+
+    const snapshotEq = Array.getEquivalence(Number.Equivalence);
+
+    yield* Effect.log(`Live query workspace '${params.workspaceId}'`);
+
+    const workspace = yield* manager
+      .getById({ workspaceId: params.workspaceId })
+      .pipe(Effect.flatMap(Effect.fromNullable));
+
+    const live = liveQuery(() =>
+      db.temp_workspace
+        .where("workspaceId")
+        .equals(params.workspaceId)
+        .toArray()
+    );
+
+    yield* Effect.forkScoped(
+      Effect.acquireRelease(
+        Effect.gen(function* () {
+          yield* Effect.log("Subscribing");
+
+          const ref = yield* SynchronizedRef.make(0);
+          return live.subscribe((payload) =>
+            Effect.runPromise(
+              Effect.gen(function* () {
+                yield* Effect.log(`Change detected`);
+
+                const id = yield* ref.pipe(
+                  SynchronizedRef.updateAndGet((n) => n + 1)
+                );
+
+                yield* Stream.runDrain(
+                  Stream.make(...payload).pipe(
+                    Stream.changesWith((a, b) =>
+                      snapshotEq(a.snapshot, b.snapshot)
+                    ),
+                    Stream.debounce("3 seconds"),
+                    Stream.tap((message) =>
+                      Effect.gen(function* () {
+                        const streamId = yield* ref.get;
+                        if (streamId === id) {
+                          yield* Effect.log(
+                            `Syncing ${payload.length} changes`
+                          );
+
+                          const snapshot = yield* Schema.decode(Snapshot)(
+                            message.snapshot
+                          );
+
+                          yield* push({
+                            workspace,
+                            snapshot,
+                          });
+                        }
+                      })
+                    )
+                  )
+                );
+              })
+            )
+          );
+        }),
+        (subscription) =>
+          Effect.gen(function* () {
+            yield* Effect.log("Live query unsubscribing");
+            return subscription.unsubscribe();
+          })
+      )
+    );
+
+    return true;
+  });
+
+const WorkerLive = WorkerRunner.layer((params: LiveQuery) =>
+  Effect.scoped(
+    Effect.gen(function* () {
+      yield* Effect.log("Startup live query connection");
+
+      yield* Effect.fork(main({ workspaceId: params.workspaceId }));
+      yield* Effect.forever(Effect.never);
+
+      yield* Effect.log("Closed live query connection");
+    }).pipe(Effect.mapError(() => "Live query error"))
+  )
+).pipe(Layer.provide(BrowserWorkerRunner.layer));
+
+RuntimeClient.runFork(WorkerRunner.launch(WorkerLive));
diff --git a/apps/client/src/workers/schema.ts b/apps/client/src/workers/schema.ts
@@ -1,5 +1,11 @@
 import { Schema } from "effect";
 
+export class LiveQuery extends Schema.TaggedRequest<LiveQuery>()("LiveQuery", {
+  failure: Schema.String,
+  payload: { workspaceId: Schema.String },
+  success: Schema.Boolean,
+}) {}
+
 export class Bootstrap extends Schema.TaggedRequest<Bootstrap>()("Bootstrap", {
   failure: Schema.String,
   payload: { workspaceId: Schema.String },
diff --git a/apps/client/src/workers/sync.ts b/apps/client/src/workers/sync.ts
@@ -1,19 +1,16 @@
 import { WorkerRunner } from "@effect/platform";
 import { BrowserWorkerRunner } from "@effect/platform-browser";
 import { Effect, Layer } from "effect";
-import { LoroDoc } from "loro-crdt";
-import { ApiClient } from "../lib/api-client";
-import { Dexie } from "../lib/dexie";
 import { RuntimeClient } from "../lib/runtime-client";
+import { Sync } from "../lib/services/sync";
 import { TempWorkspace } from "../lib/services/temp-workspace";
 import { WorkspaceManager } from "../lib/services/workspace-manager";
 import { WorkerMessage } from "./schema";
 
 const WorkerLive = WorkerRunner.layerSerialized(WorkerMessage, {
   Bootstrap: (params) =>
     Effect.gen(function* () {
-      const { client } = yield* ApiClient;
-      const { initClient } = yield* Dexie;
+      const { push } = yield* Sync;
 
       const manager = yield* WorkspaceManager;
       const temp = yield* TempWorkspace;
@@ -22,77 +19,21 @@ const WorkerLive = WorkerRunner.layerSerialized(WorkerMessage, {
 
       const workspace = yield* manager
         .getById({ workspaceId: params.workspaceId })
-        .pipe(
-          Effect.flatMap(Effect.fromNullable),
-          Effect.mapError(() => "Get workspace error")
-        );
+        .pipe(Effect.flatMap(Effect.fromNullable));
 
-      const clientId = yield* initClient.pipe(
-        Effect.mapError(() => "Init client error")
-      );
-
-      const tempUpdates = yield* temp
-        .getById({ workspaceId: workspace.workspaceId })
-        .pipe(Effect.mapError(() => "Get temp workspace error"));
+      const tempUpdates = yield* temp.getById({
+        workspaceId: workspace.workspaceId,
+      });
 
       if (tempUpdates !== undefined) {
-        const response = yield* Effect.fromNullable(workspace.token).pipe(
-          Effect.flatMap((token) =>
-            client.syncData
-              .push({
-                // headers: { Authorization: `Bearer ${token}` },
-                path: { workspaceId: workspace.workspaceId },
-                payload: { clientId, snapshot: tempUpdates.snapshot },
-              })
-              .pipe(
-                Effect.map((response) => ({ ...response, token })),
-                Effect.mapError(() => "Sync push error")
-              )
-          ),
-          Effect.orElse(() =>
-            client.syncAuth
-              .generateToken({
-                payload: {
-                  clientId,
-                  snapshot: tempUpdates.snapshot,
-                  workspaceId: params.workspaceId,
-                },
-              })
-              .pipe(
-                Effect.tap(({ token }) =>
-                  manager.setToken({
-                    workspaceId: workspace.workspaceId,
-                    token,
-                  })
-                ),
-                Effect.mapError(() => "Generate token error")
-              )
-          )
-        );
-
-        const doc = new LoroDoc();
-        doc.import(response.snapshot);
-
-        yield* manager
-          .put({
-            workspaceId: response.workspaceId,
-            snapshot: response.snapshot,
-            token: response.token,
-            version: doc.version().encode(),
-          })
-          .pipe(Effect.mapError(() => "Put workspace error"));
-
-        yield* temp
-          .clean({ workspaceId: workspace.workspaceId })
-          .pipe(Effect.mapError(() => "Clean temp workspace error"));
-
+        yield* push({ snapshot: tempUpdates.snapshot, workspace });
         yield* Effect.log("Sync completed");
       } else {
         yield* Effect.log("No sync updates");
       }
 
       return true;
-    }),
+    }).pipe(Effect.mapError(() => "Bootstrap error")),
 }).pipe(Layer.provide(BrowserWorkerRunner.layer));
 
 RuntimeClient.runFork(WorkerRunner.launch(WorkerLive));
diff --git a/apps/server/src/group/sync-data.ts b/apps/server/src/group/sync-data.ts
@@ -1,7 +1,7 @@
 import { HttpApiBuilder } from "@effect/platform";
 import { SyncApi } from "@local/sync";
 import { SnapshotToLoroDoc } from "@local/sync/loro";
-import { and, eq } from "drizzle-orm";
+import { and, desc, eq } from "drizzle-orm";
 import { Array, Effect, Layer, Schema } from "effect";
 import { workspaceTable } from "../db/schema";
 import { Drizzle } from "../drizzle";
@@ -36,6 +36,7 @@ export const SyncDataGroupLive = HttpApiBuilder.group(
                         eq(workspaceTable.clientId, clientId)
                       )
                     )
+                    .orderBy(desc(workspaceTable.createdAt))
                     .limit(1),
               })({ clientId, workspaceId }).pipe(Effect.flatMap(Array.head));
 
