feat: serialize and hydrate data before setting it into the ydoc (#3155)

Author: asalem1

src/flows/components/FromTemplatePage.tsx

@@ -52,7 +52,10 @@ const Template: FC = () => {
 
     TEMPLATES[params[0]].init
       .apply(this, params.slice(1))
-      .then(data => hydrate(data))
+      .then(data => {
+        data.orgID = org.id
+        return hydrate(data)
+      })
       .then(data => {
         if (isFlagEnabled('ephemeralNotebook')) {
           populate(data)

src/flows/components/header/index.tsx

@@ -140,7 +140,7 @@ const FlowHeader: FC = () => {
 
   const printJSON = () => {
     /* eslint-disable no-console */
-    console.log(JSON.stringify(serialize(flow, orgID), null, 2))
+    console.log(JSON.stringify(serialize(flow), null, 2))
     /* eslint-enable no-console */
   }
 

src/flows/context/flow.current.tsx

@@ -7,13 +7,13 @@ import React, {
   useEffect,
 } from 'react'
 import {Flow, PipeData, PipeMeta} from 'src/types/flows'
-import {useParams} from 'react-router'
 import {FlowListContext, FlowListProvider} from 'src/flows/context/flow.list'
 import {v4 as UUID} from 'uuid'
 import {DEFAULT_PROJECT_NAME, PIPE_DEFINITIONS} from 'src/flows'
 import {isFlagEnabled} from 'src/shared/utils/featureFlag'
 import * as Y from 'yjs'
 import {WebsocketProvider} from 'y-websocket'
+import {serialize, hydrate} from 'src/flows/context/flow.list'
 
 export interface FlowContextType {
   name: string
@@ -45,13 +45,10 @@ export const FlowProvider: FC = ({children}) => {
   const {flows, update, currentID} = useContext(FlowListContext)
   const [currentFlow, setCurrentFlow] = useState<Flow>()
 
-  const yDoc = useRef(new Y.Doc())
   const provider = useRef<WebsocketProvider>()
-
-  const {id: flowId} = useParams<{id: string}>()
-
+  const yDoc = useRef(new Y.Doc())
   function disconnectProvider() {
-    if (provider.current?.wsconnected) {
+    if (provider.current) {
       provider.current.disconnect()
     }
   }
@@ -68,38 +65,49 @@ export const FlowProvider: FC = ({children}) => {
     }
   }, [flows, currentID])
 
+  const syncFunc = useCallback(
+    (isSynced: boolean) => {
+      if (!isSynced || !yDoc.current) {
+        return
+      }
+      const {flowUpdateData} = yDoc.current.getMap('flowUpdateData')?.toJSON()
+      if (!flowUpdateData && currentFlow) {
+        yDoc.current
+          .getMap('flowUpdateData')
+          .set('flowUpdateData', serialize(currentFlow))
+      }
+    },
+    [currentFlow]
+  )
+
   useEffect(() => {
-    if (isFlagEnabled('sharedFlowEditing')) {
+    const doc = yDoc.current
+    if (isFlagEnabled('sharedFlowEditing') && currentID) {
       provider.current = new WebsocketProvider(
+        // 'ws://localhost:1223', // todo(ariel): replace this with an actual API that we setup
         'wss://demos.yjs.dev', // todo(ariel): replace this with an actual API that we setup
-        flowId, // todo(ariel): we might need to confine this to an org, depends on how multi-org plays out
-        yDoc.current
+        currentID,
+        doc
       )
-      provider.current.on('sync', isSynced => {
-        if (isSynced) {
-          const localState = yDoc.current.getMap('localState')?.toJSON()
-          if (!localState) {
-            yDoc.current.getMap('localState').set('localState', currentFlow)
-          }
-        }
-      })
-      yDoc.current.on('update', () => {
-        const {localState} = yDoc.current.getMap('localState').toJSON()
-        setCurrentFlow(prev => {
-          if (btoa(JSON.stringify(prev)) === btoa(JSON.stringify(localState))) {
-            return prev
-          }
-          return localState
-        })
-      })
+
+      provider.current.on('sync', syncFunc)
     }
 
+    const onUpdate = () => {
+      const {flowUpdateData} = doc.getMap('flowUpdateData').toJSON()
+      const hydrated = hydrate(flowUpdateData?.data)
+
+      setCurrentFlow(hydrated)
+    }
+
+    doc.on('update', onUpdate)
     return () => {
       if (isFlagEnabled('sharedFlowEditing')) {
         disconnectProvider()
       }
+      doc.off('update', onUpdate)
     }
-  }, [flowId])
+  }, [currentID])
 
   const updateData = useCallback(
     (id: string, data: Partial<PipeData>) => {
@@ -110,11 +118,9 @@ export const FlowProvider: FC = ({children}) => {
             ...(flowCopy.data.byID[id] || {}),
             ...data,
           }
-          const update = {
-            ...flowCopy,
-            ...flowCopy.data.byID[id],
-          }
-          yDoc.current.getMap('localState').set('localState', update)
+          yDoc.current
+            .getMap('flowUpdateData')
+            .set('flowUpdateData', serialize(flowCopy))
         }
         return
       }
@@ -154,11 +160,9 @@ export const FlowProvider: FC = ({children}) => {
             ...meta,
           }
 
-          const update = {
-            ...flowCopy,
-            ...flowCopy.meta.byID[id],
-          }
-          yDoc.current.getMap('localState').set('localState', update)
+          yDoc.current
+            .getMap('flowUpdateData')
+            .set('flowUpdateData', serialize(flowCopy))
         }
         return
       }
@@ -198,9 +202,9 @@ export const FlowProvider: FC = ({children}) => {
           for (const ni in flow) {
             flowCopy[ni] = flow[ni]
           }
-          yDoc.current.getMap('localState').set('localState', {
-            ...flowCopy,
-          })
+          yDoc.current
+            .getMap('flowUpdateData')
+            .set('flowUpdateData', serialize(flowCopy))
         }
         return
       }
@@ -248,9 +252,9 @@ export const FlowProvider: FC = ({children}) => {
         flowCopy.data.allIDs.push(id)
         flowCopy.meta.allIDs.push(id)
       }
-      yDoc.current.getMap('localState').set('localState', {
-        ...flowCopy,
-      })
+      yDoc.current
+        .getMap('flowUpdateData')
+        .set('flowUpdateData', serialize(flowCopy))
       return
     }
     if (isFlagEnabled('ephemeralNotebook') && !currentFlow.id) {
@@ -299,9 +303,9 @@ export const FlowProvider: FC = ({children}) => {
 
       delete flowCopy.data.byID[id]
       delete flowCopy.meta.byID[id]
-      yDoc.current.getMap('localState').set('localState', {
-        ...flowCopy,
-      })
+      yDoc.current
+        .getMap('flowUpdateData')
+        .set('flowUpdateData', serialize(flowCopy))
       return
     }
     if (isFlagEnabled('ephemeralNotebook') && !currentFlow.id) {

src/flows/context/flow.list.tsx

@@ -74,17 +74,18 @@ export const FlowListContext = React.createContext<FlowListContextType>(
   DEFAULT_CONTEXT
 )
 
-export function serialize(flow: Flow, orgID: string) {
-  const apiFlow = {
+export function serialize(flow: Flow) {
+  const apiFlow: any = {
     data: {
-      orgID,
+      orgID: flow.orgID,
       name: flow.name,
       spec: {
         name: flow.name,
         readOnly: flow.readOnly,
         range: flow.range,
         refresh: flow.refresh,
         createdAt: flow.createdAt,
+        createdBy: flow.createdBy,
         updatedAt: flow.updatedAt,
         pipes: flow.data.allIDs.map(id => {
           const meta = flow.meta.byID[id]
@@ -110,6 +111,10 @@ export function serialize(flow: Flow, orgID: string) {
     },
   }
 
+  if (flow.id) {
+    apiFlow.data.id = flow.id
+  }
+
   return apiFlow
 }
 
@@ -127,6 +132,9 @@ export function hydrate(data) {
   if (data.id) {
     flow.id = data.id
   }
+  if (data.orgID) {
+    flow.orgID = data.orgID
+  }
 
   if (!data?.spec?.pipes) {
     return flow
@@ -193,10 +201,12 @@ export const FlowListProvider: FC = ({children}) => {
       _flow = hydrate({
         ...TEMPLATES['default'].init(),
         name: DEFAULT_PROJECT_NAME,
+        orgID: org.id,
       })
     } else {
       _flow = {
         name: flow.name,
+        orgID: flow.orgID,
         range: flow.range,
         refresh: flow.refresh,
         data: flow.data,
@@ -207,7 +217,7 @@ export const FlowListProvider: FC = ({children}) => {
       }
     }
 
-    const apiFlow = serialize(_flow, org.id)
+    const apiFlow = serialize(_flow)
 
     let id: string = `local_${UUID()}`
     try {
@@ -247,13 +257,10 @@ export const FlowListProvider: FC = ({children}) => {
         },
       }))
 
-      const apiFlow = serialize(
-        {
-          ...flows[id],
-          ...flow,
-        },
-        org.id
-      )
+      const apiFlow = serialize({
+        ...flows[id],
+        ...flow,
+      })
 
       pooledUpdateAPI({id, ...apiFlow})
     },

src/types/flows.ts

@@ -103,6 +103,7 @@ export interface FlowState {
 
 export interface Flow {
   id?: string
+  orgID?: string
   name: string
   range: TimeRange
   refresh: AutoRefresh