Merge remote-tracking branch 'origin/main' into fix/issue-209

web/src/hooks/RealtimeManager.tsx

@@ -19,7 +19,6 @@ export function RealtimeManager({ children }: { children: React.ReactNode }) {
   const setUnseenCount = useAppStore((s) => s.setUnseenCount)
 
   const queryClient = useQueryClient()
-  const prevOrgIdRef = useRef<string | null>(null)
   const heartbeatRef = useRef<ReturnType<typeof setInterval> | null>(null)
   const lastHeartbeatRef = useRef<number>(Date.now())
   const hadConnectionRef = useRef(false)
@@ -87,26 +86,19 @@ export function RealtimeManager({ children }: { children: React.ReactNode }) {
     }
   }, [isConnected, user?.id, joinChannel, leaveChannel, addJoinedChannel, removeJoinedChannel])
 
-  // Auto-join/leave org channel on org switch
+  // Auto-join/leave org channel on org switch. Needs the cleanup: the effect
+  // re-runs on every reconnect, so without one the join never gets its leave.
   useEffect(() => {
-    if (!isConnected) return
+    if (!isConnected || !currentOrg?.id) return
 
-    const prevOrgId = prevOrgIdRef.current
-    const newOrgId = currentOrg?.id
+    const topic = `org:${currentOrg.id}`
+    joinChannel(topic)
+    addJoinedChannel(topic)
 
-    if (prevOrgId && prevOrgId !== newOrgId) {
-      const prevTopic = `org:${prevOrgId}`
-      leaveChannel(prevTopic)
-      removeJoinedChannel(prevTopic)
+    return () => {
+      leaveChannel(topic)
+      removeJoinedChannel(topic)
     }
-
-    if (newOrgId) {
-      const topic = `org:${newOrgId}`
-      joinChannel(topic)
-      addJoinedChannel(topic)
-    }
-
-    prevOrgIdRef.current = newOrgId ?? null
   }, [isConnected, currentOrg?.id, joinChannel, leaveChannel, addJoinedChannel, removeJoinedChannel])
 
   // Connection quality monitoring

web/src/hooks/SocketProvider.tsx

@@ -100,6 +100,10 @@ export default function SocketProvider({
     // 'closed', so a state-filtered rejoin skipped them all and the socket came
     // back with zero subscriptions (no events, no presence) until a reload.
     const desiredTopicsRef = useRef<Map<string, Record<string, unknown>>>(new Map());
+    // Holders per topic, so the first of two surfaces to unmount doesn't take
+    // the channel out from under the second. Kept outside the channel entry
+    // because that entry is rebuilt on every join and rejoin.
+    const holdersRef = useRef<Map<string, number>>(new Map());
     // Distinguishes a close we caused (logout / unmount — don't reconnect) from
     // every other close (server idle-close, channel crash, network drop — do
     // reconnect). The old code only reconnected on `!wasClean`, so a clean
@@ -390,6 +394,10 @@ export default function SocketProvider({
 
     // Join channel
     const joinChannel = useCallback((topic: string, params: Record<string, unknown> = {}) => {
+        // Must count before the already-joined bail-out below, or a second
+        // holder asking for a live topic is never counted at all.
+        holdersRef.current.set(topic, (holdersRef.current.get(topic) ?? 0) + 1);
+
         // Remember the intent so a reconnect rejoins this topic even after its
         // live state was reset to 'closed' by a drop.
         desiredTopicsRef.current.set(topic, params);
@@ -428,17 +436,26 @@ export default function SocketProvider({
 
     // Leave channel
     const leaveChannel = useCallback((topic: string) => {
+        // One holder fewer. Anyone still holding the topic keeps it joined.
+        const remaining = (holdersRef.current.get(topic) ?? 0) - 1;
+        if (remaining > 0) {
+            holdersRef.current.set(topic, remaining);
+            return;
+        }
+        holdersRef.current.delete(topic);
+
         // No longer want this topic — don't let a reconnect rejoin it.
         desiredTopicsRef.current.delete(topic);
         clearRejoin(topic);
+        pendingJoinsRef.current.delete(topic);
 
         const channel = channelsRef.current.get(topic);
         if (!channel) return;
 
-        channel.state = 'leaving';
-        markChannelState(topic, 'leaving');
-
-        if (wsRef.current?.readyState === WebSocket.OPEN) {
+        if (
+            wsRef.current?.readyState === WebSocket.OPEN &&
+            (channel.state === 'joined' || channel.state === 'joining')
+        ) {
             sendRaw({
                 topic,
                 event: PHOENIX_EVENTS.LEAVE,
@@ -448,8 +465,14 @@ export default function SocketProvider({
             });
         }
 
-        channelsRef.current.delete(topic);
-        pendingJoinsRef.current.delete(topic);
+        // The entry owns the handler map, so dropping it here took every other
+        // subscriber's registrations with it. Reset in place while any remain.
+        if (channel.handlers.size === 0) {
+            channelsRef.current.delete(topic);
+        } else {
+            channel.state = 'closed';
+            channel.joinRef = '';
+        }
         markChannelState(topic, null);
     }, [getRef, sendRaw, markChannelState, clearRejoin]);
 
@@ -486,8 +509,19 @@ export default function SocketProvider({
 
         return () => {
             handlers?.delete(handler);
-            if (handlers?.size === 0) {
-                channel?.handlers.delete(event);
+            if (handlers?.size !== 0) return;
+            channel?.handlers.delete(event);
+            // Nothing listening and nobody holding: drop the entry leaveChannel
+            // kept for us. Read the live one; a rejoin rebuilds it around the
+            // same handlers map.
+            const current = channelsRef.current.get(topic);
+            if (
+                current &&
+                current.handlers.size === 0 &&
+                !holdersRef.current.has(topic) &&
+                !desiredTopicsRef.current.has(topic)
+            ) {
+                channelsRef.current.delete(topic);
             }
         };
     }, []);

web/src/hooks/realtimeChannels.test.tsx

@@ -0,0 +1,255 @@
+import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'
+import { render, act } from '@testing-library/react'
+import React, { createContext } from 'react'
+import { MemoryRouter } from 'react-router-dom'
+import { QueryClient, QueryClientProvider } from '@tanstack/react-query'
+import SocketProvider from './SocketProvider'
+import { RealtimeManager } from './RealtimeManager'
+import { useChannel, useChannelEvent } from './context/socket'
+import { useAppStore } from '@/stores'
+import { installFakeSocket, freezeJitter, type SocketEnv } from './socketTestHarness'
+
+vi.mock('@/lib/api/client/app/socket/getSocket', () => ({
+    default: vi.fn(async () => ({ url: 'ws://localhost:4000/socket/websocket?token=test' })),
+}))
+
+const USER_ID = '99999999-9999-9999-9999-999999999999'
+const ORG_A = '11111111-1111-1111-1111-111111111111'
+const ORG_B = '22222222-2222-2222-2222-222222222222'
+
+vi.mock('./context/user', () => ({
+    UserContext: createContext(null),
+    useUserProfile: () => ({ user: { id: USER_ID } }),
+}))
+
+vi.mock('@/lib/api/hooks/app/unibox/useUnseenCount', () => ({
+    default: () => ({ data: undefined }),
+}))
+
+let env: SocketEnv
+let queryClient: QueryClient
+
+const CAMPAIGN = 'campaign:33333333-3333-3333-3333-333333333333'
+
+/** A page-level surface holding its own topic, the way useCampaignChannel does. */
+function CampaignPanel({ onEvent }: { onEvent: () => void }) {
+    useChannel(CAMPAIGN)
+    useChannelEvent(CAMPAIGN, 'EMAIL_SENT', () => onEvent())
+    return null
+}
+
+/** A surface that only listens on the org channel, without holding the join. */
+function OrgListener({ orgId, onEvent }: { orgId: string; onEvent: () => void }) {
+    useChannelEvent(`org:${orgId}`, 'EMAIL_SENT', () => onEvent())
+    return null
+}
+
+function Tree({ children }: { children?: React.ReactNode }) {
+    return (
+        <MemoryRouter>
+            <QueryClientProvider client={queryClient}>
+                <SocketProvider>
+                    <RealtimeManager>{children}</RealtimeManager>
+                </SocketProvider>
+            </QueryClientProvider>
+        </MemoryRouter>
+    )
+}
+
+async function mount(children?: React.ReactNode) {
+    const result = render(<Tree>{children}</Tree>)
+    await act(async () => {
+        await vi.advanceTimersByTimeAsync(10)
+    })
+    return result
+}
+
+async function tick(ms: number) {
+    await act(async () => {
+        await vi.advanceTimersByTimeAsync(ms)
+    })
+}
+
+/** Ack every join the client has sent that the server hasn't answered yet. */
+let ackedJoins = 0
+async function ackPendingJoins() {
+    const joins = env.joins()
+    await act(async () => {
+        for (let i = ackedJoins; i < joins.length; i++) env.ackJoin(joins[i])
+    })
+    ackedJoins = joins.length
+}
+
+/** Kill the live socket and let the provider's backoff bring a new one up. */
+async function dropAndReconnect() {
+    await act(async () => {
+        env.instances[env.instances.length - 1].close()
+    })
+    await tick(500)
+    await ackPendingJoins()
+}
+
+beforeEach(() => {
+    vi.useFakeTimers()
+    freezeJitter(0)
+    env = installFakeSocket()
+    ackedJoins = 0
+    queryClient = new QueryClient({
+        defaultOptions: { queries: { retry: false } },
+    })
+    useAppStore.setState({ currentOrganization: { id: ORG_A } as never, joinedChannels: [] })
+})
+
+afterEach(() => {
+    vi.useRealTimers()
+    vi.restoreAllMocks()
+})
+
+describe('RealtimeManager channel lifecycle', () => {
+    it('joins the user and org channels exactly once on connect', async () => {
+        await mount()
+        await ackPendingJoins()
+
+        expect(env.joins(`user:${USER_ID}`)).toHaveLength(1)
+        expect(env.joins(`org:${ORG_A}`)).toHaveLength(1)
+        expect(env.leaves()).toHaveLength(0)
+        expect(useAppStore.getState().joinedChannels).toContain(`org:${ORG_A}`)
+    })
+
+    it('rejoins the org channel once per reconnect and keeps its subscribers', async () => {
+        // The cleanup leaves while the socket is down; that must not strand the
+        // rejoin or take other surfaces' handlers with it.
+        const onEvent = vi.fn()
+        await mount(<OrgListener orgId={ORG_A} onEvent={onEvent} />)
+        await ackPendingJoins()
+
+        await dropAndReconnect()
+
+        expect(env.joins(`org:${ORG_A}`)).toHaveLength(2)
+        expect(env.joins(`user:${USER_ID}`)).toHaveLength(2)
+
+        await act(async () => {
+            env.pushEvent(`org:${ORG_A}`, 'EMAIL_SENT', {})
+        })
+        expect(onEvent).toHaveBeenCalledTimes(1)
+    })
+
+    it('still routes org events into the query cache after a reconnect', async () => {
+        await mount()
+        await ackPendingJoins()
+        await dropAndReconnect()
+
+        const invalidate = vi.spyOn(queryClient, 'invalidateQueries')
+        await act(async () => {
+            env.pushEvent(`org:${ORG_A}`, 'AUDIT_CREATED', {
+                action: 'contact.updated',
+                entity_type: 'contact',
+                entity_id: 'abc',
+            })
+        })
+
+        expect(
+            invalidate.mock.calls.some(
+                (c) => JSON.stringify((c[0] as { queryKey?: unknown })?.queryKey) === '["contacts"]'
+            )
+        ).toBe(true)
+    })
+
+    it('really leaves the previous workspace after any number of reconnects', async () => {
+        // An unbalanced join per reconnect would pin the old org's channel open
+        // and keep feeding the user another workspace's events.
+        await mount()
+        await ackPendingJoins()
+
+        for (let i = 0; i < 3; i++) await dropAndReconnect()
+
+        await act(async () => {
+            useAppStore.setState({ currentOrganization: { id: ORG_B } as never })
+        })
+        await ackPendingJoins()
+
+        expect(env.leaves(`org:${ORG_A}`)).toHaveLength(1)
+        expect(env.joins(`org:${ORG_B}`)).toHaveLength(1)
+        expect(useAppStore.getState().joinedChannels).not.toContain(`org:${ORG_A}`)
+        expect(useAppStore.getState().joinedChannels).toContain(`org:${ORG_B}`)
+    })
+
+    it('joins the org channel once when the reconnect batches with the drop', async () => {
+        // Comes back via rejoinChannels or via the effect depending on whether
+        // React flushed the disconnected render; either way, exactly one join.
+        const onEvent = vi.fn()
+        await mount(<OrgListener orgId={ORG_A} onEvent={onEvent} />)
+        await ackPendingJoins()
+
+        await act(async () => {
+            env.instances[env.instances.length - 1].close()
+            await vi.advanceTimersByTimeAsync(500)
+        })
+        await ackPendingJoins()
+
+        expect(env.joins(`org:${ORG_A}`)).toHaveLength(2)
+        await act(async () => {
+            env.pushEvent(`org:${ORG_A}`, 'EMAIL_SENT', {})
+        })
+        expect(onEvent).toHaveBeenCalledTimes(1)
+    })
+
+    it('does not rejoin a workspace abandoned while the socket was down', async () => {
+        await mount()
+        await ackPendingJoins()
+
+        await act(async () => {
+            env.instances[env.instances.length - 1].close()
+        })
+
+        // The user picks another workspace before the socket is back.
+        await act(async () => {
+            useAppStore.setState({ currentOrganization: { id: ORG_B } as never })
+        })
+        await tick(500)
+        await ackPendingJoins()
+
+        expect(env.joins(`org:${ORG_A}`)).toHaveLength(1)
+        expect(env.joins(`org:${ORG_B}`)).toHaveLength(1)
+        expect(env.joins(`user:${USER_ID}`)).toHaveLength(2)
+    })
+
+    it('rejoins a page-owned channel alongside the org channel', async () => {
+        // Effect-driven rejoin and rejoinChannels both run on the same reconnect.
+        const onEvent = vi.fn()
+        await mount(<CampaignPanel onEvent={onEvent} />)
+        await ackPendingJoins()
+
+        await dropAndReconnect()
+
+        expect(env.joins(CAMPAIGN)).toHaveLength(2)
+        expect(env.joins(`org:${ORG_A}`)).toHaveLength(2)
+        expect(env.leaves(CAMPAIGN)).toHaveLength(0)
+
+        await act(async () => {
+            env.pushEvent(CAMPAIGN, 'EMAIL_SENT', {})
+        })
+        expect(onEvent).toHaveBeenCalledTimes(1)
+    })
+
+    it('stops delivering the previous workspace events after a switch', async () => {
+        const onEvent = vi.fn()
+        await mount(<OrgListener orgId={ORG_A} onEvent={onEvent} />)
+        await ackPendingJoins()
+
+        await act(async () => {
+            useAppStore.setState({ currentOrganization: { id: ORG_B } as never })
+        })
+        await ackPendingJoins()
+
+        // The server stops sending on a left topic; assert the client asked it to.
+        expect(env.leaves(`org:${ORG_A}`)).toHaveLength(1)
+        expect(env.joins(`org:${ORG_A}`)).toHaveLength(1)
+
+        // And a later reconnect must not resurrect the abandoned topic.
+        await dropAndReconnect()
+        expect(env.joins(`org:${ORG_A}`)).toHaveLength(1)
+        expect(env.joins(`org:${ORG_B}`)).toHaveLength(2)
+        expect(onEvent).not.toHaveBeenCalled()
+    })
+})

web/src/hooks/socketChannels.test.tsx

@@ -104,6 +104,170 @@ describe('channel subscription lifecycle', () => {
     })
 })
 
+describe('two surfaces holding the same topic', () => {
+    /** Holds the join without rendering anything. */
+    function Joiner() {
+        useChannel(TOPIC)
+        return null
+    }
+
+    /** Listens on the topic without holding the join. */
+    function Listener({ onEvent }: { onEvent: () => void }) {
+        useChannelEvent(TOPIC, 'EMAIL_SENT', () => onEvent())
+        return null
+    }
+
+    it('joins once and keeps the channel while the second holder remains', async () => {
+        const onA = vi.fn()
+        const onB = vi.fn()
+        function Wrapper({ showA }: { showA: boolean }) {
+            return (
+                <>
+                    {showA ? <Panel onEvent={onA} /> : null}
+                    <Panel onEvent={onB} />
+                </>
+            )
+        }
+
+        const { rerender } = await mount(<Wrapper showA />)
+        expect(env.joins(TOPIC)).toHaveLength(1)
+        env.ackJoin(env.lastJoin(TOPIC))
+        await tick(1)
+
+        // The first holder goes away (the list row scrolls out, the drawer stays).
+        await act(async () => {
+            rerender(
+                <SocketProvider>
+                    <Wrapper showA={false} />
+                </SocketProvider>
+            )
+        })
+
+        expect(env.leaves(TOPIC)).toHaveLength(0)
+        await act(async () => {
+            env.pushEvent(TOPIC, 'EMAIL_SENT', {})
+        })
+        expect(onB).toHaveBeenCalledTimes(1)
+        expect(onA).not.toHaveBeenCalled()
+    })
+
+    it('leaves exactly once when the last holder goes', async () => {
+        function Wrapper({ holders }: { holders: number }) {
+            return (
+                <>
+                    {Array.from({ length: holders }, (_, i) => (
+                        <Joiner key={i} />
+                    ))}
+                </>
+            )
+        }
+
+        const { rerender } = await mount(<Wrapper holders={3} />)
+        env.ackJoin(env.lastJoin(TOPIC))
+        await tick(1)
+        expect(env.joins(TOPIC)).toHaveLength(1)
+
+        for (const holders of [2, 1, 0]) {
+            await act(async () => {
+                rerender(
+                    <SocketProvider>
+                        <Wrapper holders={holders} />
+                    </SocketProvider>
+                )
+            })
+        }
+
+        expect(env.leaves(TOPIC)).toHaveLength(1)
+    })
+
+    it('survives a drop while both hold it, then a single holder leaving', async () => {
+        // A campaign row and its drawer both hold the topic across a blip.
+        const onRow = vi.fn()
+        const onDrawer = vi.fn()
+        function Wrapper({ drawer }: { drawer: boolean }) {
+            return (
+                <>
+                    <Panel onEvent={onRow} />
+                    {drawer ? <Panel onEvent={onDrawer} /> : null}
+                </>
+            )
+        }
+
+        const { rerender } = await mount(<Wrapper drawer />)
+        env.ackJoin(env.lastJoin(TOPIC))
+        await tick(1)
+
+        // The socket dies; the provider rejoins every topic still wanted.
+        await act(async () => {
+            env.instances[env.instances.length - 1].close()
+        })
+        await tick(500)
+        expect(env.joins(TOPIC)).toHaveLength(2)
+        await act(async () => {
+            env.ackJoin(env.lastJoin(TOPIC))
+        })
+
+        await act(async () => {
+            rerender(
+                <SocketProvider>
+                    <Wrapper drawer={false} />
+                </SocketProvider>
+            )
+        })
+
+        expect(env.leaves(TOPIC)).toHaveLength(0)
+        await act(async () => {
+            env.pushEvent(TOPIC, 'EMAIL_SENT', {})
+        })
+        expect(onRow).toHaveBeenCalledTimes(1)
+        expect(onDrawer).not.toHaveBeenCalled()
+        expect(screen.getByTestId('state').textContent).toBe('joined')
+    })
+
+    it("keeps a listener's handler when a different surface leaves the topic", async () => {
+        // Regression: leaveChannel dropped the entry, and with it every other
+        // subscriber's handlers on that topic.
+        const onEvent = vi.fn()
+        function Wrapper({ joined }: { joined: boolean }) {
+            return (
+                <>
+                    <Listener onEvent={onEvent} />
+                    {joined ? <Joiner /> : null}
+                </>
+            )
+        }
+
+        const { rerender } = await mount(<Wrapper joined />)
+        env.ackJoin(env.lastJoin(TOPIC))
+        await tick(1)
+
+        await act(async () => {
+            rerender(
+                <SocketProvider>
+                    <Wrapper joined={false} />
+                </SocketProvider>
+            )
+        })
+        expect(env.leaves(TOPIC)).toHaveLength(1)
+
+        await act(async () => {
+            rerender(
+                <SocketProvider>
+                    <Wrapper joined />
+                </SocketProvider>
+            )
+        })
+        await act(async () => {
+            env.ackJoin(env.lastJoin(TOPIC))
+        })
+        await act(async () => {
+            env.pushEvent(TOPIC, 'EMAIL_SENT', {})
+        })
+
+        expect(onEvent).toHaveBeenCalledTimes(1)
+    })
+})
+
 describe('a refused join', () => {
     it('retries a throttled join when the server says budget is back', async () => {
         await mount(<Panel />)