YJS와 Websocket 그리고 React Flow

YJS란

참고 자료

https://www.tag1consulting.com/blog/deep-dive-real-time-collaborative-editing-solutions-tagteamtalk-001-0

YJS의 목적

• 실시간 협업 솔루션에 대한 필요성.
• 인트라넷, 이미 존재하는 서비스 등에 실시간 협업 기능을 쉽게 추가할 수 있는 방법 필요.

동시 편집, CRDT의 핵심

• CRDT의 기본은 position-based가 아니라 모든 캐릭터에 대해 identifier를 두는 것.

  “this is some text"와 같은 텍스트가 있다고 하자. A, B가 이를 동시에 편집 중이다.

  A의 커서는 맨 앞에, position 0에 있다.

  B의 커서는 “some” 뒤, position 12에 있다.

  이 때, A는 “hello”를, B는 “nice”를 추가하고 싶다. 결국 동시 편집 상 원하는 결과는 “hello this is some nice text”이다.

  이 때, 각자의 position을 기준으로 동시편집이 이루어지면 A의 “hello”에 의해 기존 텍스트가 밀려나면서 “hello this is nice some text”와 같이 원하는 것과 다른 결과가 나올 수 있다.

  이를 해결하기 위해 마치 사람이 이 문제를 해결하듯이, B의 “nice”를 position 12가 아닌, “some” 뒤에 추가하는 것이 CRDT의 기본적인 해결 방법이다.

Decentralized Approach

• Operational Transformation을 사용하는 기존의 솔루션들은 서버에서 동시성을 해결해주는 중앙화된 방식으로 동작한다. 단점은 제한된 동시성이다. 서버와 클라이언트 사이에서만 동시성이 존재하기 때문에. 또 이 서버가 single point of failure가 되기도 한다.
• YJS, 혹은 CRDT는 이와 달리 좀 더 분산화된 방식을 채택하고 있다. CRDT는 변화가 어떤 순서로 들어오든지 같은 결과를 갖게 되기 때문에 중앙된 서버에 의존하지 않아도 된다. 이에 대한 장점은 중앙에 모이지 않으니 scalable하다는 것도 있고, 서버 연결이 되지 않는 비행기와 같은 공간에서도 한 팀이 같이 작업할 수 있다는 점도 있다.

Edit History와 Undo/Redo

• Operational Transformation에서는 해당 문서의 history가 존재하기 때문에. Undo/Redo가 어렵지 않다.
• 반면 CRDT에서는 그런 linear한 history가 없다. 트리 구조(정확히는 DAG인듯)를 띄기 때문에 Undo/Redo 시에는 이에 대한 경로와 state를 통해 해결하는데, 결론적으로는 잘 해결되어 있는 문제인 것 같다. 이를 위해 Undo Manager가 구현되어 있는데 문서의 scope와 권한의 scope를 지정해줄 수 있다.
• 결국 고려할 사항은 유저가 Undo/Redo를 했을 때, 어떤 어떤 문서까지 수정할 것인지(현재 편집 중인 문서, 존재하는 모든 문서), 다른 유저가 수정한 것을 Undo/Redo할 것인지가 된다.

YJS와 Shared Types

Array, Map, Set 같은, 근데 자동적으로 state가 sync되는 Shared Types를 제공하는게 YJS의 핵심이다. 이 점에서 Rich Text Editor에 사용하기 적합하다.

---

Websocket과 LevelDB

https://github.com/ivan-topp/y-socket.io

https://github.com/yjs/y-websocket

https://github.com/yjs/y-leveldb

• Socket IO로 사용을 할 수는 있는데 일반적으로는 websocket을 사용.
• persistance를 위해서는 levelDB를 주로 사용.

import * as Y from 'yjs'
import { LeveldbPersistence } from 'y-leveldb'

const persistence = new LeveldbPersistence('./storage-location')

const ydoc = new Y.Doc()
ydoc.getArray('arr').insert(0, [1, 2, 3])
ydoc.getArray('arr').toArray() // => [1, 2, 3]

// store document updates retrieved from other clients
persistence.storeUpdate('my-doc', Y.encodeStateAsUpdate(ydoc))

// when you want to sync, or store data to a database,
// retrieve the temporary Y.Doc to consume data
const ydocPersisted = await persistence.getYDoc('my-doc')
ydocPersisted.getArray('arr') // [1, 2, 3]

• 영속화를 위해 메인 DB에도 저장하는 건 이런 방식으로 가능할 것 같다.

doc.on('update', async () => {
    await this.flowService.handleDocumentUpdate(flowId, doc);
});

• 서버에서 Y.Map을 클라이언트와 똑같이 obeserve할 수도 있다.

doc.getMap('nodes').observe(event => {
  event.changes.added.forEach((item) => {
    console.log('Node added:', item);
  });

  event.changes.deleted.forEach((item) => {
    console.log('Node deleted:', item);
  });

  event.changes.updated.forEach((item) => {
    console.log('Node updated:', item);
  });
});

---

YJS를 사용한 노드 뷰 프로토타입

실행 예시

liveflow-screencast.1.mp4

Github 링크

https://github.com/djk01281/live-flow/blob/main/README.md

Y.Doc

Shared Types를 모두 담는 root 레벨의 일종의 컨터이너, 혹은 content라고 생각할 수 있다.

const doc = new Y.Doc();

상태를 저장할 수 있는 방법은 크게 Awareness와 Y.Map이 있다.

Awareness

일시적으로만 필요한 상태를 위해서 사용한다.

프로토타입에서는 커서 정보를 담기 위해 사용하고 있고, 마우스를 움직였을 때 이 값을 변경시켜준다. setLocalState()로 변경시켜준다.

interface AwarenessState {
  cursor: { x: number; y: number } | null;
  color: string;
  clientId: number;
}

// ...

const handleMouseMove = useCallback((e: React.MouseEvent) => {
  if (!provider.current?.awareness || !flowRef.current) return;

  const bounds = flowRef.current.getBoundingClientRect();
  const cursor = {
    x: e.clientX - bounds.left,
    y: e.clientY - bounds.top,
  };

  provider.current.awareness.setLocalState({
    cursor,
    color: userColor.current,
    clientId: provider.current.awareness.clientID,
  });
}, []);

다른 유저들은 이 정보에 접근하기 위해 awreness.on()의 콜백 함수를 사용할 수 있다.

const [cursors, setCursors] = useState<Map<number, AwarenessState>>(
  new Map()
);

// ...

wsProvider.awareness.on("change", () => {
  const states = new Map(
    wsProvider.awareness.getStates() as Map<number, AwarenessState>
  );
  setCursors(states);
});

// ...

return(
{Array.from(cursors.entries()).map(([clientId, state]) => 
    <Cursor
      key={clientId}
      x={state.cursor.x}
      y={state.cursor.y}
      color={state.color}
    />
  )
}

Y.Map

유지되어야 하는 정보를 위해 사용한다. 또, Undo/Redo를 위한 history 또한 보관한다.

프로토타입에서는 노드, 엣지 정보를 담기 위해 사용하고 있다. 프로토타입에는 아직 없지만 이 Y.Map을 위해 levelDB, y-leveldb를 사용하면 될 것 같다.

client에서 노드, 엣지의 상태는 다음과 같이 React Flow가 제공하는 hook을 통해 관리된다.

const [nodes, setNodes, onNodesChange] = useNodesState(initialNodes);
const [edges, setEdges, onEdgesChange] = useEdgesState([]);

소켓 측에는 다음과 같이 저장한다.

const nodesMap = doc.getMap("nodes");
const edgesMap = doc.getMap("edges");

edge가 변경되면 이 client의 edges 상태와 함께 Y.Map을 변경시킨다.

  const onConnect = useCallback(
    (connection: Connection) => {
      if (!connection.source || !connection.target) return;

      const newEdge: Edge = {
        id: `e${connection.source}-${connection.target}`,
        source: connection.source,
        target: connection.target,
        sourceHandle: connection.sourceHandle || undefined,
        targetHandle: connection.targetHandle || undefined,
      };

      if (ydoc.current) {
        ydoc.current.getMap("edges").set(newEdge.id, newEdge);
      }

      setEdges((eds) => addEdge(connection, eds));
    },
    [setEdges]
  );

또, 다음과 같이 Y.Map에 변경이 있으면 그로부터 값을 가져와 client 상태를 변경시켜준다.

edgesMap.observe(() => {
  const yEdges = Array.from(edgesMap.values()) as Edge[];
  setEdges(yEdges);
});

---