diff --git a/web/src/hooks/RealtimeManager.tsx b/web/src/hooks/RealtimeManager.tsx index 6011337d..9abd0dca 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,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 diff --git a/web/src/hooks/SocketProvider.tsx b/web/src/hooks/SocketProvider.tsx index 838ecf48..2a9858ed 100644 --- a/web/src/hooks/SocketProvider.tsx +++ b/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>>(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>(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 = {}) => { + // 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); } }; }, []); diff --git a/web/src/hooks/realtimeChannels.test.tsx b/web/src/hooks/realtimeChannels.test.tsx new file mode 100644 index 00000000..6442d67f --- /dev/null +++ b/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 ( + + + + {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 () => { + // 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() + 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() + 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() + 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..2e4da8f3 100644 --- a/web/src/hooks/socketChannels.test.tsx +++ b/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 ? : 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 () => { + // 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 ( + <> + + {drawer ? : null} + + ) + } + + const { rerender } = await mount() + 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( + + + + ) + }) + + 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 ( + <> + + {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()