diff --git a/web/src/hooks/RealtimeManager.tsx b/web/src/hooks/RealtimeManager.tsx index 6011337d..92bc98d5 100644 --- a/web/src/hooks/RealtimeManager.tsx +++ b/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(null) const heartbeatRef = useRef | null>(null) const lastHeartbeatRef = useRef(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 diff --git a/web/src/hooks/SocketProvider.tsx b/web/src/hooks/SocketProvider.tsx index 838ecf48..311fd7bc 100644 --- a/web/src/hooks/SocketProvider.tsx +++ b/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>>(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>(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 = {}) => { + // 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); } }; }, []); diff --git a/web/src/hooks/realtimeChannels.test.tsx b/web/src/hooks/realtimeChannels.test.tsx new file mode 100644 index 00000000..5d4892be --- /dev/null +++ b/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 ( + + + + {children} + + + + ) +} + +async function mount(children?: React.ReactNode) { + const result = render({children}) + 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() + 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() + 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() + 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() + 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() + }) +}) diff --git a/web/src/hooks/socketChannels.test.tsx b/web/src/hooks/socketChannels.test.tsx index a52d5b6c..c139336d 100644 --- a/web/src/hooks/socketChannels.test.tsx +++ b/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 ? : null} + + + ) + } + + const { rerender } = await mount() + 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( + + + + ) + }) + + 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) => ( + + ))} + + ) + } + + const { rerender } = await mount() + 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( + + + + ) + }) + } + + 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 ( + <> + + {drawer ? : null} + + ) + } + + const { rerender } = await mount() + 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( + + + + ) + }) + + 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 ( + <> + + {joined ? : null} + + ) + } + + const { rerender } = await mount() + env.ackJoin(env.lastJoin(TOPIC)) + await tick(1) + + await act(async () => { + rerender( + + + + ) + }) + expect(env.leaves(TOPIC)).toHaveLength(1) + + await act(async () => { + rerender( + + + + ) + }) + 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()