fix(shared-state): keep initial snapshots local

packages/devframe/src/client/rpc-shared-state.test.ts

@@ -1,5 +1,6 @@
 import { createEventEmitter } from 'devframe/utils/events'
 import { describe, expect, it } from 'vitest'
+import { DEVFRAME_EVENTS } from '../events'
 import { createRpcSharedStateClientHost } from './rpc-shared-state'
 
 function makeFakeRpc() {
@@ -50,3 +51,77 @@ describe('client shared state', () => {
     expect(setCalls).toHaveLength(1) // exactly one server-state:set, not two
   })
 })
+
+describe('initial shared-state snapshots', () => {
+  function setup() {
+    const { rpc, setCalls } = makeFakeRpc()
+    const snapshot = Promise.withResolvers<object | undefined>()
+    const handlers = new Map<string, (...args: any[]) => void>()
+    rpc.isTrusted = true
+    rpc.call = () => snapshot.promise
+    rpc.client.register = (definition: any) => handlers.set(definition.name, definition.handler)
+    const host = createRpcSharedStateClientHost(rpc)
+    return { rpc, host, snapshot, handlers, setCalls }
+  }
+
+  it('reads a snapshot without writing it back to the node side', async () => {
+    const { host, snapshot, setCalls } = setup()
+    const state = await host.get('k', { initialValue: { count: 0, localDefault: true } })
+    snapshot.resolve({ count: 1 })
+    await snapshot.promise
+    expect(state.value()).toEqual({ count: 1, localDefault: true })
+    expect(setCalls).toEqual([])
+    state.mutate((draft) => {
+      draft.count = 2
+    })
+    expect(setCalls[0]?.[1]).toEqual({ count: 2, localDefault: true })
+  })
+
+  it('keeps an update received while the initial snapshot is pending', async () => {
+    const { host, snapshot, handlers } = setup()
+    const state = await host.get('k', { initialValue: { count: 0 } })
+    handlers.get(DEVFRAME_EVENTS.broadcast.clientStateUpdated)!('k', { count: 2 }, 'newer')
+    snapshot.resolve({ count: 1 })
+    await snapshot.promise
+    expect(state.value()).toEqual({ count: 2 })
+  })
+
+  it('keeps a local write made while the initial snapshot is pending', async () => {
+    const { host, snapshot, setCalls } = setup()
+    const state = await host.get('k', { initialValue: { count: 0 } })
+    state.mutate((draft) => {
+      draft.count = 2
+    })
+    snapshot.resolve({ count: 1 })
+    await snapshot.promise
+    expect(state.value()).toEqual({ count: 2 })
+    expect(setCalls).toHaveLength(1)
+  })
+
+  it('forwards a local write triggered by the initial snapshot', async () => {
+    const { host, snapshot, setCalls } = setup()
+    const state = await host.get('k', { initialValue: { count: 0 } })
+    state.on('updated', (value) => {
+      if (value.count === 1) {
+        state.mutate((draft) => {
+          draft.count = 2
+        })
+      }
+    })
+    snapshot.resolve({ count: 1 })
+    await snapshot.promise
+    expect(state.value()).toEqual({ count: 2 })
+    expect(setCalls).toHaveLength(1)
+    expect(setCalls[0]?.[1]).toEqual({ count: 2 })
+  })
+
+  it('ignores a snapshot after its state is deleted', async () => {
+    const { host, snapshot, setCalls } = setup()
+    const state = await host.get('k', { initialValue: { count: 0 } })
+    host.delete('k')
+    snapshot.resolve({ count: 1 })
+    await snapshot.promise
+    expect(state.value()).toEqual({ count: 0 })
+    expect(setCalls).toEqual([])
+  })
+})

packages/devframe/src/client/rpc-shared-state.ts

@@ -1,6 +1,7 @@
 import type { RpcSharedStateGetOptions, RpcSharedStateHost } from 'devframe/types'
 import type { SharedState, SharedStatePatch } from 'devframe/utils/shared-state'
 import type { DevframeRpcClient } from './rpc'
+import { nanoid } from 'devframe/utils/nanoid'
 import { createSharedState } from 'devframe/utils/shared-state'
 import { DEVFRAME_EVENTS } from '../events'
 
@@ -42,10 +43,10 @@ export function createRpcSharedStateClientHost(rpc: DevframeRpcClient): RpcShare
     },
   })
 
-  function registerSharedState<T extends object>(key: string, state: SharedState<T>) {
+  function registerSharedState<T extends object>(key: string, state: SharedState<T>, initialSyncId?: string) {
     const offs: (() => void)[] = []
     offs.push(state.on('updated', (fullState, patches, syncId) => {
-      if (isStaticBackend)
+      if (isStaticBackend || syncId === initialSyncId)
         return
       if (patches) {
         rpc.callEvent('devframe:rpc:server-state:patch', key, patches, syncId)
@@ -99,15 +100,18 @@ export function createRpcSharedStateClientHost(rpc: DevframeRpcClient): RpcShare
           sharedState.set(key, state)
           for (const fn of keyAddedListeners)
             fn(key)
+          const initialState = state.value()
+          const initialSyncId = nanoid()
           rpc.call('devframe:rpc:server-state:get', key)
             .then((serverState) => {
-              if (serverState !== undefined)
-                state.mutate(() => mergeWithInitialValue(key, serverState))
+              // A pending read must not replace a newer update or a deleted state.
+              if (serverState !== undefined && sharedState.get(key) === state && state.value() === initialState)
+                state.mutate(() => mergeWithInitialValue(key, serverState), initialSyncId)
             })
             .catch((error) => {
               console.error('Error getting server state', error)
             })
-          stateDisposers.set(key, registerSharedState(key, state))
+          stateDisposers.set(key, registerSharedState(key, state, initialSyncId))
           return state
         }
         else {