syncing web worker logic

Author: SandroMaglione

apps/client/src/lib/runtime-client.ts

@@ -1,12 +1,14 @@
 import { Layer, ManagedRuntime } from "effect";
 import { ApiClient } from "./api-client";
 import { Dexie } from "./dexie";
+import { TempWorkspace } from "./services/temp-workspace";
 import { WorkspaceManager } from "./services/workspace-manager";
 
 const MainLayer = Layer.mergeAll(
   Dexie.Default,
   ApiClient.Default,
-  WorkspaceManager.Default
+  WorkspaceManager.Default,
+  TempWorkspace.Default
 );
 
 export const RuntimeClient = ManagedRuntime.make(MainLayer);

apps/client/src/lib/services/temp-workspace.ts

@@ -0,0 +1,36 @@
+import { Effect, Schema } from "effect";
+import { Dexie } from "../dexie";
+import { TempWorkspaceTable } from "../schema";
+
+export class TempWorkspace extends Effect.Service<TempWorkspace>()(
+  "TempWorkspace",
+  {
+    dependencies: [Dexie.Default],
+    effect: Effect.gen(function* () {
+      const { query } = yield* Dexie;
+      return {
+        put: (params: typeof TempWorkspaceTable.Type) =>
+          Schema.encode(TempWorkspaceTable)(params).pipe(
+            Effect.flatMap((data) => query((_) => _.temp_workspace.put(data)))
+          ),
+
+        get: ({ workspaceId }: { workspaceId: string }) =>
+          query((_) =>
+            _.temp_workspace
+              .where("workspaceId")
+              .equals(workspaceId)
+              .limit(1)
+              .first()
+          ).pipe(
+            Effect.flatMap(Effect.fromNullable),
+            Effect.flatMap(Schema.decode(TempWorkspaceTable))
+          ),
+
+        clean: ({ workspaceId }: { workspaceId: string }) =>
+          query((_) =>
+            _.temp_workspace.where("workspaceId").equals(workspaceId).delete()
+          ),
+      };
+    }),
+  }
+) {}

apps/client/src/lib/services/workspace-manager.ts

@@ -1,5 +1,6 @@
-import { Effect } from "effect";
+import { Effect, Schema } from "effect";
 import { Dexie } from "../dexie";
+import { WorkspaceTable } from "../schema";
 
 export class WorkspaceManager extends Effect.Service<WorkspaceManager>()(
   "WorkspaceManager",
@@ -10,6 +11,19 @@ export class WorkspaceManager extends Effect.Service<WorkspaceManager>()(
       const { query } = yield* Dexie;
 
       return {
+        setToken: ({
+          token,
+          workspaceId,
+        }: {
+          workspaceId: typeof WorkspaceTable.Type.workspaceId;
+          token: NonNullable<typeof WorkspaceTable.Type.token>;
+        }) => query((_) => _.workspace.update(workspaceId, { token })),
+
+        put: (update: typeof WorkspaceTable.Type) =>
+          Schema.encode(WorkspaceTable)(update).pipe(
+            Effect.flatMap((data) => query((_) => _.workspace.put(data)))
+          ),
+
         getAll: query((_) => _.workspace.toArray()),
 
         getById: ({ workspaceId }: { workspaceId: string }) =>
@@ -21,7 +35,7 @@ export class WorkspaceManager extends Effect.Service<WorkspaceManager>()(
               .first()
           ).pipe(Effect.flatMap(Effect.fromNullable)),
 
-        putCurrent: (workspaceId: string | undefined) =>
+        createOrJoin: (workspaceId: string | undefined) =>
           query((_) =>
             _.workspace.toCollection().modify({ current: false })
           ).pipe(

apps/client/src/routes/index.tsx

@@ -13,10 +13,10 @@ function HomeComponent() {
   const allWorkspaces = Route.useLoaderData();
   const navigate = useNavigate();
 
-  const [, putWorkspace] = useActionEffect((formData: FormData | undefined) =>
+  const [, joinWorkspace] = useActionEffect((formData: FormData | undefined) =>
     Effect.gen(function* () {
       const workspaceId = formData?.get("workspaceId") as string | null;
-      const workspace = yield* WorkspaceManager.putCurrent(
+      const workspace = yield* WorkspaceManager.createOrJoin(
         workspaceId ?? undefined
       );
       yield* Effect.sync(() =>
@@ -41,13 +41,13 @@ function HomeComponent() {
         </Link>
       ))}
 
-      <form action={putWorkspace}>
+      <form action={joinWorkspace}>
         <input type="text" name="workspaceId" placeholder="Workspace id" />
         <button type="submit">Switch workspace</button>
       </form>
 
       <div>
-        <button type="button" onClick={() => putWorkspace(undefined)}>
+        <button type="button" onClick={() => joinWorkspace(undefined)}>
           Create workspace
         </button>
       </div>

apps/client/src/workers/schema.ts

@@ -1,7 +1,7 @@
 import { Schema } from "effect";
 
 export class Bootstrap extends Schema.TaggedRequest<Bootstrap>()("Bootstrap", {
-  failure: Schema.Never,
+  failure: Schema.String,
   payload: { workspaceId: Schema.String },
   success: Schema.Boolean,
 }) {}

apps/client/src/workers/sync.ts

@@ -1,14 +1,87 @@
 import { WorkerRunner } from "@effect/platform";
 import { BrowserWorkerRunner } from "@effect/platform-browser";
 import { Effect, Layer } from "effect";
+import { ApiClient } from "../lib/api-client";
+import { Dexie } from "../lib/dexie";
+import { RuntimeClient } from "../lib/runtime-client";
+import { TempWorkspace } from "../lib/services/temp-workspace";
+import { WorkspaceManager } from "../lib/services/workspace-manager";
 import { WorkerMessage } from "./schema";
 
 const WorkerLive = WorkerRunner.layerSerialized(WorkerMessage, {
-  Bootstrap: ({ workspaceId }) =>
+  Bootstrap: (params) =>
     Effect.gen(function* () {
-      yield* Effect.log(`Bootstrapping workspace ${workspaceId}`);
+      const { client } = yield* ApiClient;
+      const { initClient } = yield* Dexie;
+
+      const manager = yield* WorkspaceManager;
+      const temp = yield* TempWorkspace;
+
+      yield* Effect.log(`Running workspace '${params.workspaceId}'`);
+
+      const workspace = yield* manager
+        .getById({ workspaceId: params.workspaceId })
+        .pipe(Effect.mapError(() => "Get workspace error"));
+
+      const clientId = yield* initClient.pipe(
+        Effect.mapError(() => "Init client error")
+      );
+
+      const tempUpdates = yield* temp
+        .get({ workspaceId: workspace.workspaceId })
+        .pipe(Effect.mapError(() => "Get temp workspace error"));
+
+      if (tempUpdates !== null) {
+        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")
+              )
+          )
+        );
+
+        yield* manager
+          .put({
+            workspaceId: response.workspaceId,
+            snapshot: response.snapshot,
+            token: response.token,
+            version: new Uint8Array(), // TODO
+          })
+          .pipe(Effect.mapError(() => "Put workspace error"));
+
+        yield* Effect.log("Sync completed");
+      } else {
+        yield* Effect.log("No sync updates");
+      }
+
       return true;
     }),
 }).pipe(Layer.provide(BrowserWorkerRunner.layer));
 
-Effect.runFork(WorkerRunner.launch(WorkerLive));
+RuntimeClient.runFork(WorkerRunner.launch(WorkerLive));

packages/sync/src/main.ts

@@ -52,7 +52,14 @@ export class SyncAuthGroup extends HttpApiGroup.make("syncAuth")
         })
       )
       .addError(Schema.String)
-      .addSuccess(Schema.Struct({ token: Schema.String }))
+      .addSuccess(
+        Schema.Struct({
+          token: Schema.String,
+          workspaceId: WorkspaceTable.fields.workspaceId,
+          createdAt: WorkspaceTable.fields.createdAt,
+          snapshot: WorkspaceTable.fields.snapshot,
+        })
+      )
   )
   .add(
     /**