connecting yjs

app/backend-v2/src/env.ts

@@ -3,7 +3,7 @@ import { createEnv } from "@t3-oss/env-core"
 
 export const env = createEnv({
   server: {
-    PORT: z.coerce.number().default(3000),
+    PORT: z.coerce.number().default(3030),
     YJS_SWEET_URI: z.string(),
   },
   runtimeEnv: process.env,

app/frontend/.eslintrc.json

@@ -1,11 +1,8 @@
 {
-  "extends": [
-    "./node_modules/gts/",
-    "plugin:react/jsx-runtime"
-  ],
+  "extends": ["./node_modules/gts/", "plugin:react/jsx-runtime"],
   "plugins": ["react-hooks"],
   "rules": {
-    "react-hooks/rules-of-hooks": "error", 
+    "react-hooks/rules-of-hooks": "error",
     "react-hooks/exhaustive-deps": "warn"
   }
-}
+}

app/frontend/package.json

@@ -21,6 +21,7 @@
     "antd": "^5.7.3",
     "crypto-js": "^4.1.1",
     "deep-equal": "^2.1.0",
+    "fast-json-patch": "^3.1.1",
     "fp-ts": "^2.13.1",
     "geometric": "^2.5.0",
     "immer": "^9.0.19",

app/frontend/src/component/panel/AnnotationMonitor.tsx

@@ -1,10 +1,12 @@
 import {FC, useContext, useEffect} from 'react';
+import * as Y from 'yjs';
 import * as jsonmergepatch from 'json-merge-patch';
+import * as jsonpatch from 'fast-json-patch';
 import {produce} from 'immer';
 
 import {deleteAnnotationComponent, useStore as useAnnoStore} from 'state/annotate/annotation';
 import {useStore as useRenderStore} from 'state/annotate/render';
-import {usePatchVideoAnnotation} from 'state/server/annotation';
+import {useGetVideoAnnotationV2, usePatchVideoAnnotation} from 'state/server/annotation';
 import {useStore as useUIStore} from 'state/annotate/ui';
 
 import {ConfigContext, NutshClientContext} from 'common/context';
@@ -15,9 +17,10 @@ import {Annotation} from 'type/annotation';
 
 export const MonitorAnnotation: FC<{videoId: Video['id']}> = ({videoId}) => {
   const config = useContext(ConfigContext);
+  useSyncAnnotation({videoId});
   return (
     <>
-      {!config.readonly && <SyncAnnotation videoId={videoId} />}
+      {/* {!config.readonly && <SyncAnnotation videoId={videoId} />} */}
       <ForgetEntities />
       <CommitDraft />
     </>
@@ -132,3 +135,91 @@ function removeDraftComponents(anno: Annotation): Annotation {
     return draft;
   });
 }
+
+function useSyncAnnotation({videoId}: {videoId: Video['id']}) {
+  const {data} = useGetVideoAnnotationV2(videoId);
+
+  useEffect(() => {
+    const anno = data?.anno;
+    if (!anno) {
+      return;
+    }
+    const fn = (e: Y.YEvent<Y.Map<unknown>>[]) => {
+      console.log(e);
+    };
+    anno.observeDeep(fn);
+    return () => anno.unobserveDeep(fn);
+  }, [data?.anno]);
+
+  useEffect(() => {
+    return useAnnoStore.subscribe(
+      s => s.annotation,
+      (curr, prev) => {
+        if (!data?.anno) {
+          return;
+        }
+
+        const newPrev = produce(prev, removeDraftComponents);
+        const newCurr = produce(curr, removeDraftComponents);
+
+        const ops = jsonpatch.compare(newPrev, newCurr);
+        ops.forEach(op => updateYjsFromJsonPathOperation(data.anno, op));
+      },
+      {
+        fireImmediately: true,
+      }
+    );
+  }, [data?.anno]);
+}
+
+function updateYjsFromJsonPathOperation(anno: Y.Map<unknown>, op: jsonpatch.Operation) {
+  console.log(JSON.stringify(op));
+  switch (op.op) {
+    case 'add': {
+      if (!op.path.startsWith('/')) {
+        console.warn(`unexpected json pointer: ${op.path}`);
+        return;
+      }
+      const paths = op.path.split('/');
+      let dict = anno;
+      paths.slice(1, -1).forEach(path => {
+        if (!dict.has(path)) {
+          dict.set(path, new Y.Map());
+        }
+        dict = dict.get(path) as Y.Map<unknown>;
+      });
+      const lastKey = paths[paths.length - 1];
+      dict.set(lastKey, op.value);
+      break;
+    }
+    case 'copy':
+      console.warn(`not implemented: ${JSON.stringify(op)}`);
+      break;
+    case 'move':
+      console.warn(`not implemented: ${JSON.stringify(op)}`);
+      break;
+    case 'remove': {
+      if (!op.path.startsWith('/')) {
+        console.warn(`unexpected json pointer: ${op.path}`);
+        return;
+      }
+      const paths = op.path.split('/');
+      let dict = anno;
+      paths.slice(1, -1).forEach(path => {
+        if (!dict.has(path)) {
+          dict.set(path, new Y.Map());
+        }
+        dict = dict.get(path) as Y.Map<unknown>;
+      });
+      const lastKey = paths[paths.length - 1];
+      dict.delete(lastKey);
+      break;
+    }
+    case 'replace':
+      console.warn(`not implemented: ${JSON.stringify(op)}`);
+      break;
+    case 'test':
+      console.warn(`not implemented: ${JSON.stringify(op)}`);
+      break;
+  }
+}

app/frontend/src/page/annotate/Panel/Load.tsx

@@ -3,13 +3,13 @@ import intl from 'react-intl-universal';
 import {Alert} from 'antd';
 import {useStore as useRenderStore} from 'state/annotate/render';
 import {useStore as useAnnoStore} from 'state/annotate/annotation';
-import {useGetVideoAnnotation} from 'state/server/annotation';
 import {useGetVideo} from 'state/server/video';
 import {NutshClientContext} from 'common/context';
 import PageLayout from 'page/Layout';
 import {mustDecodeJsonStr as mustDecodeAnnotationJsonStr} from 'type/annotation';
 import type {Video} from 'openapi/nutsh';
 import {PanelLoadProject} from './LoadProject';
+import {useGetVideoAnnotationV2} from 'state/server/annotation';
 
 export const PanelLoad: FC<{id: Video['id']}> = ({id}) => {
   const client = useContext(NutshClientContext);
@@ -21,7 +21,8 @@ export const PanelLoad: FC<{id: Video['id']}> = ({id}) => {
 
   // server state
   const {isFetching: isFetchingVideo, data: getVideoData} = useGetVideo(client, id);
-  const {isFetching: isFetchingAnno, data: getVideoAnnotationData} = useGetVideoAnnotation(client, id);
+  // const {isFetching: isFetchingAnno, data: getVideoAnnotationData} = useGetVideoAnnotation(client, id);
+  const {isFetching: isFetchingAnno, data: getVideoAnnotationData} = useGetVideoAnnotationV2(id);
 
   // local state
   const [errorCode, setErrorCode] = useState<string | undefined>(undefined);
@@ -31,7 +32,8 @@ export const PanelLoad: FC<{id: Video['id']}> = ({id}) => {
     if (!getVideoData) return;
 
     setErrorCode(undefined);
-    const {annotation_json: annoJson, annotation_version: annoVersion} = getVideoAnnotationData;
+    // const {annotation_json: annoJson, annotation_version: annoVersion} = getVideoAnnotationData;
+    const {annoJson, annoVersion} = getVideoAnnotationData;
     const {frame_urls: frameUrls} = getVideoData.video;
     if (!frameUrls || frameUrls.length === 0) {
       setErrorCode('error.missing_video_frames');

app/frontend/src/state/server/annotation.ts

@@ -17,31 +17,30 @@ export const usePatchVideoAnnotation = (client: NutshClient) => {
   });
 };
 
-export const useGetVideoAnnotationV2 = (client: NutshClient, id: Video['id']) =>
+export const useGetVideoAnnotationV2 = (id: Video['id']) =>
   useQuery({
     queryKey: ['getVideoAnnotationV2', id],
     queryFn: async () => {
       // TODO(xu): remove host hard coding
-      const url = new URL('http://localhost:3030/yjs-client-token');
+      const url = new URL('http://192.168.31.208:3030/yjs-client-token');
       url.searchParams.set('doc', id);
       const res = await fetch(url.toString());
       const clientToken = await res.json();
-
+      console.log(clientToken);
       // create a Yjs document and connect it to the Y-Sweet server
       const doc = new Y.Doc();
       createYjsProvider(doc, clientToken, {disableBc: true});
-      const annoMap = doc.getMap('annotation');
-
-      // TODO(xu): no need to serialize back to a string after the transition to node is completed
-      let annotation_json = '';
-      let annotation_version = '';
-      if (annoMap.size === 0) {
-        return {annotation_json, annotation_version};
-      }
-
-      annotation_json = JSON.stringify(annoMap.toJSON());
-      annotation_version = doc.getText('version').toString();
 
-      return {annotation_json, annotation_version};
+      const anno = doc.getMap('annotation');
+      console.log(111);
+      const annoJson = await new Promise<string>(resolve => {
+        console.log(333);
+        anno.observe(() => {
+          console.log(222);
+          resolve(JSON.stringify(anno.toJSON()));
+        });
+      });
+
+      return {anno, annoJson, annoVersion: ''};
     },
   });

main.go

@@ -3,6 +3,9 @@ package main
 import (
 	"embed"
 	"fmt"
+	"net/http"
+	"net/http/httputil"
+	"net/url"
 	"os"
 	"path/filepath"
 
@@ -99,6 +102,8 @@ func main() {
 		},
 	}
 
+	go mustProxyYSweetServer()
+
 	mustOk(app.Run(os.Args))
 }
 
@@ -125,6 +130,22 @@ func mustSetupLogger() *zap.Logger {
 	return logger
 }
 
+// y-sweet server hard-coded the listening host as `127.0.0.1`, which prevents remote machines to access.
+// To rescue, we launch a reverse proxy.
+// https://github.com/drifting-in-space/y-sweet/blob/0f5bcfd05dfc744c01bb06f202ce6c97a1aebb49/crates/y-sweet/src/main.rs#L158
+func mustProxyYSweetServer() {
+	// TODO(hxu): remove hard coding
+	target := "http://127.0.0.1:8080"
+	listenAddr := ":9090"
+
+	url, err := url.Parse(target)
+	mustOk(err)
+
+	proxy := httputil.NewSingleHostReverseProxy(url)
+	zap.L().Info("proxying y-sweer server", zap.String("on", listenAddr), zap.String("to", target))
+	mustOk(http.ListenAndServe(listenAddr, proxy))
+}
+
 func mustOk(err error) {
 	if err == nil {
 		return