feat: fix the dashboard socket layer's channel bookkeeping so a topic held by two surfaces survives one of them unmounting and a workspace switch actually leaves the old organization's channel: leaveChannel deleted the whole channelsRef entry, and that entry owns the handler map, so every other subscribeToChannel registration on that topic died with it and the unsubscribe closures they held pointed at an orphaned Set, meaning the first component out silently deafened the second; it now refcounts holders in holdersRef (incremented before joinChannel's already-joined bail-out, so the second surface to ask for a live topic is actually counted) and only the last holder out leaves, resets the entry in place instead of dropping it while subscribers remain, discards it from the unsubscribe closure once nothing is listening and nobody holds the join, sends phx_leave only for a channel that was really joined or joining, and drops the pointless 'leaving' state flip that was set and nulled in the same tick and read by nothing; refcounting alone would have been actively harmful, because RealtimeManager's org effect joined with no cleanup and re-ran on every isConnected false->true cycle, so the count climbed on every reconnect and never returned to zero and switching workspaces would decrement without leaving and keep feeding the client the previous organization's events, so that effect is now balanced with a cleanup matching the user-channel effect above it and prevOrgIdRef is gone, which also fixes a latent defect of its own where switching organizations while the socket was down left the old topic in desiredTopicsRef and rejoinChannels brought the abandoned workspace back on reconnect; covered by 11 new tests over the fake-socket rig, including a real RealtimeManager mounted against the real provider, and each combination of half-fixes fails the matrix

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,21 @@ 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. Balanced with a cleanup, like
+  // the user channel above: the effect re-runs on every disconnect/reconnect
+  // and on every org switch, so without one the join outnumbered the leave and
+  // the provider's holder count for a workspace never came back to zero.
   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,12 @@ 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());
+    // How many callers currently want each topic joined. Two surfaces can hold
+    // the same topic (a campaign row and its detail drawer), and the first to
+    // unmount must not take the channel out from under the second, so only the
+    // last holder out actually leaves. Kept beside the channel map rather than
+    // inside the 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 +396,11 @@ export default function SocketProvider({
 
     // Join channel
     const joinChannel = useCallback((topic: string, params: Record<string, unknown> = {}) => {
+        // Count the holder BEFORE the already-joined bail-out below, or the
+        // second surface to ask for a live topic is never counted and the first
+        // one to unmount leaves the channel while it is still wanted.
+        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 +439,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 +468,15 @@ export default function SocketProvider({
             });
         }
 
-        channelsRef.current.delete(topic);
-        pendingJoinsRef.current.delete(topic);
+        // Handlers belong to subscribeToChannel, not to the join: dropping the
+        // entry here took every other subscriber's registrations with it. Reset
+        // it in place and discard it only once nothing is listening either.
+        if (channel.handlers.size === 0) {
+            channelsRef.current.delete(topic);
+        } else {
+            channel.state = 'closed';
+            channel.joinRef = '';
+        }
         markChannelState(topic, null);
     }, [getRef, sendRaw, markChannelState, clearRejoin]);
 
@@ -486,8 +513,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 the join: drop the entry
+            // leaveChannel kept alive for us. Read the live one — a join or a
+            // rejoin rebuilds the entry 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,262 @@
+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 () => {
+        // Regression: the org effect joined with no cleanup, so its leave never
+        // balanced the join. Giving it one means a drop now leaves the channel
+        // while the socket is down, which 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 () => {
+        // The join refcount is what makes a switch actually leave, so an
+        // unbalanced join on every 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 () => {
+        // The org channel comes back one of two ways depending on whether React
+        // flushed the disconnected render before the socket reopened: through
+        // rejoinChannels (it did not) or through the effect (it did). Either way
+        // it must be exactly one join, never zero and never two.
+        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 () => {
+        // The org and user channels come back through their effects; a channel
+        // owned by a page comes back through rejoinChannels. Both paths run on
+        // the same reconnect and must not double-join or lose handlers.
+        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,174 @@ 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 () => {
+        // The concrete case: a campaign row and its detail drawer both hold the
+        // topic, the network blips, and then the drawer closes. The row must
+        // still be joined and still receiving.
+        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)
+
+        // Network blip: the socket dies, the provider brings a new one up and
+        // rejoins every topic the app still wants.
+        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 whole channel entry, and the
+        // entry owns the handler map, so every other subscriber on that topic
+        // was silently deafened and its unsubscribe closure orphaned.
+        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 />)