805 lines
28 KiB
JavaScript
805 lines
28 KiB
JavaScript
import { Capacitor } from '@capacitor/core'
|
|
import { useQueryClient } from '@tanstack/react-query'
|
|
import { EventSourcePolyfill } from 'event-source-polyfill'
|
|
import { useCallback, useEffect, useRef, useState } from 'react'
|
|
import { useUserProfile } from '../queries/UserQueries'
|
|
import { useAlerts } from '../service/AlertsProvider'
|
|
import { useNotification } from '../service/NotificationProvider'
|
|
import { apiClient } from '../utils/ApiClient'
|
|
import { useAuth } from './useAuth.jsx'
|
|
|
|
const SSE_STATES = {
|
|
CONNECTING: 0,
|
|
OPEN: 1,
|
|
CLOSED: 2,
|
|
}
|
|
|
|
const RECONNECT_INTERVALS = [10000, 30000, 360000, 600000, 900000, 6000000] // 10s, 30s, 6m, 10m, 15m , 1h
|
|
const MAX_RECONNECT_ATTEMPTS = 10 // Circuit breaker limit
|
|
const CIRCUIT_BREAKER_RESET_TIME = 600000 // 10 minutes
|
|
|
|
export const useSSE = () => {
|
|
const { isAuthenticated, token } = useAuth()
|
|
// Only fetch user profile if authenticated - prevents unnecessary API calls on landing page
|
|
const { data: userProfile } = useUserProfile()
|
|
const [connectionState, setConnectionState] = useState(SSE_STATES.CLOSED)
|
|
const [lastEvent, setLastEvent] = useState(null)
|
|
const [error, setError] = useState(null)
|
|
const [isCircuitBreakerOpen, setIsCircuitBreakerOpen] = useState(false)
|
|
|
|
const eventSourceRef = useRef(null)
|
|
const reconnectTimeoutRef = useRef(null)
|
|
const reconnectAttemptsRef = useRef(0)
|
|
const isManuallyClosedRef = useRef(false)
|
|
const lastHeartbeatRef = useRef(Date.now())
|
|
const heartbeatMonitorRef = useRef(null)
|
|
const nextReconnectTimeRef = useRef(null)
|
|
// Track if reconnect is already scheduled to prevent duplicates
|
|
const isReconnectScheduledRef = useRef(false)
|
|
|
|
const queryClient = useQueryClient()
|
|
const { showError, showNotification } = useNotification()
|
|
const { showAlert } = useAlerts()
|
|
|
|
const getSSEUrl = useCallback(() => {
|
|
const authToken = token
|
|
if (!authToken || !isAuthenticated) {
|
|
console.log(
|
|
'SSE: No valid authentication token',
|
|
authToken,
|
|
isAuthenticated,
|
|
)
|
|
return null
|
|
}
|
|
|
|
// Get the API URL from apiManager
|
|
const apiUrl = apiClient.getApiURL() // e.g., "http://localhost:8080/api/v1"
|
|
|
|
// Build SSE URL - let backend determine circle from authenticated user
|
|
const sseUrl = `${apiUrl}/realtime/sse`
|
|
|
|
return { url: sseUrl, token }
|
|
}, [token, isAuthenticated]) // Fixed: Added missing dependencies
|
|
|
|
// Exchange the JWT (sent via Authorization header by apiClient) for a
|
|
// short-lived, single-use SSE ticket. Used by the native EventSource path,
|
|
// which cannot send custom headers.
|
|
const fetchSSETicket = useCallback(async () => {
|
|
try {
|
|
const response = await apiClient.get('/realtime/sse/ticket')
|
|
if (!response || !response.ok) {
|
|
console.error('SSE: Ticket request failed', response?.status)
|
|
return null
|
|
}
|
|
const data = await response.json()
|
|
return data.ticket || null
|
|
} catch (err) {
|
|
console.error('SSE: Ticket request error', err)
|
|
return null
|
|
}
|
|
}, [])
|
|
|
|
const handleSSEMessage = useCallback(
|
|
event => {
|
|
try {
|
|
const eventData = JSON.parse(event.data)
|
|
setLastEvent(eventData)
|
|
|
|
// Update heartbeat timestamp
|
|
if (eventData.type === 'heartbeat') {
|
|
lastHeartbeatRef.current = Date.now()
|
|
}
|
|
|
|
console.debug('SSE Message received:', eventData)
|
|
|
|
// Handle different event types and update React Query cache accordingly
|
|
switch (eventData.type) {
|
|
case 'chore.created':
|
|
showNotification({
|
|
type: 'info',
|
|
title: 'New Task Created',
|
|
message: `${eventData.data.user.displayName} created "${eventData.data.chore.name}"`,
|
|
duration: 5000,
|
|
})
|
|
const newChore = eventData.data.chore
|
|
|
|
// Update individual chore cache
|
|
queryClient.setQueryData(['chore', newChore.id], {
|
|
res: newChore,
|
|
})
|
|
|
|
// Update chores list cache
|
|
queryClient.setQueryData(['chores', false], oldData => {
|
|
if (!oldData || !oldData.res) {
|
|
return { res: [newChore] }
|
|
}
|
|
return { res: [newChore, ...oldData.res] }
|
|
})
|
|
break
|
|
case 'chore.updated':
|
|
case 'chore.completed':
|
|
case 'chore.status':
|
|
case 'chore.skipped': {
|
|
console.log('userProfile: ', userProfile, eventData.data.user)
|
|
|
|
if (eventData?.data?.user?.id !== userProfile?.id) {
|
|
showNotification({
|
|
type: 'info',
|
|
title: `Task ${eventData.type.replace('chore.', '')}`,
|
|
message: `${eventData.data.user.displayName} ${eventData.type.replace('chore.', '')} "${eventData.data.chore.name}"`,
|
|
duration: 5000,
|
|
})
|
|
}
|
|
const updatedChore = eventData.data.chore
|
|
|
|
// Update individual chore cache
|
|
queryClient.setQueryData(['chore', updatedChore.id], oldData => {
|
|
if (!oldData) return { res: updatedChore }
|
|
return { res: { ...oldData.res, ...updatedChore } }
|
|
})
|
|
|
|
// If chore update then also refetch chore details:
|
|
if (
|
|
eventData.type === 'chore.updated' ||
|
|
eventData.type === 'chore.status'
|
|
) {
|
|
queryClient.invalidateQueries(['choreDetails', updatedChore.id])
|
|
queryClient.refetchQueries({
|
|
queryKey: ['choreDetails', updatedChore.id],
|
|
})
|
|
}
|
|
|
|
// Update chores list cache - add debugging
|
|
queryClient.setQueryData(['chores', false], oldData => {
|
|
if (!oldData) return { res: [updatedChore] }
|
|
|
|
if (!oldData.res || !Array.isArray(oldData.res)) {
|
|
return { res: [updatedChore] }
|
|
}
|
|
|
|
// If it's a one-time chore that's completed, we might need to remove it
|
|
if (
|
|
eventData.type === 'chore.completed' &&
|
|
updatedChore.frequencyType === 'once'
|
|
) {
|
|
return {
|
|
res: oldData.res.filter(
|
|
chore => chore.id !== updatedChore.id,
|
|
),
|
|
}
|
|
}
|
|
|
|
// Otherwise update the existing chore or add if it doesn't exist
|
|
const newData = oldData.res.map(chore => {
|
|
if (chore.id === updatedChore.id) {
|
|
return { ...updatedChore }
|
|
}
|
|
return chore
|
|
})
|
|
|
|
return { res: newData }
|
|
})
|
|
|
|
break
|
|
}
|
|
|
|
case 'chore.deleted':
|
|
// update chores list cache
|
|
queryClient.setQueryData(['chores', false], oldData => {
|
|
if (!oldData || !oldData.res) return oldData
|
|
return {
|
|
res: oldData.res.filter(
|
|
chore => chore.id !== eventData.data.choreId,
|
|
),
|
|
}
|
|
})
|
|
// same logic for archived chores view:
|
|
queryClient.setQueryData(['chores', true], oldData => {
|
|
if (!oldData || !oldData.res) return oldData
|
|
return {
|
|
res: oldData.res.filter(
|
|
chore => chore.id !== eventData.data.choreId,
|
|
),
|
|
}
|
|
})
|
|
|
|
break
|
|
|
|
case 'subtask.updated':
|
|
case 'subtask.completed':
|
|
queryClient.setQueryData(
|
|
['choreDetails', String(eventData.data.choreId)], // this should be string to match the query key type which is param in the url in choreView
|
|
oldData => {
|
|
if (!oldData) return oldData
|
|
console.log('Old choreDetails data:', oldData)
|
|
|
|
// Update the specific subtask within the chore details
|
|
const newChoreData = { ...oldData.res }
|
|
newChoreData.subTasks = newChoreData.subTasks.map(subtask => {
|
|
if (subtask.id === eventData.data.subtaskId) {
|
|
return {
|
|
...subtask,
|
|
completedAt: eventData.data.completedAt,
|
|
completedBy: eventData.data.user.id,
|
|
}
|
|
}
|
|
return subtask
|
|
})
|
|
return { res: newChoreData }
|
|
},
|
|
)
|
|
break
|
|
|
|
case 'heartbeat':
|
|
// Heartbeat events don't need cache invalidation
|
|
console.debug('SSE Heartbeat received at', new Date().toISOString())
|
|
break
|
|
|
|
case 'connection.established':
|
|
console.log('SSE connection established')
|
|
setError(null)
|
|
lastHeartbeatRef.current = Date.now()
|
|
showAlert({
|
|
type: 'success',
|
|
color: 'success',
|
|
message: 'You are now receiving real-time as they happen.',
|
|
})
|
|
break
|
|
|
|
case 'error':
|
|
console.error('SSE error event:', eventData.data)
|
|
showError({
|
|
title: 'Real-time Error',
|
|
message:
|
|
eventData.data.message ||
|
|
'An error occurred with real-time updates',
|
|
})
|
|
break
|
|
|
|
default:
|
|
console.log('Unknown SSE event type:', eventData.type)
|
|
}
|
|
} catch (err) {
|
|
console.error('Failed to parse SSE message:', err)
|
|
showError({
|
|
title: 'Message Error',
|
|
message: 'Failed to parse server message',
|
|
})
|
|
return // Stop processing if JSON parsing fails
|
|
}
|
|
},
|
|
[queryClient, showNotification, showError, userProfile, showAlert],
|
|
)
|
|
|
|
const stopHeartbeatMonitor = useCallback(() => {
|
|
if (heartbeatMonitorRef.current) {
|
|
clearInterval(heartbeatMonitorRef.current)
|
|
heartbeatMonitorRef.current = null
|
|
}
|
|
}, [])
|
|
|
|
// Centralized reconnect scheduling function to prevent duplicate scheduling
|
|
const scheduleReconnect = useCallback((delay, reason) => {
|
|
// Prevent duplicate scheduling
|
|
if (isReconnectScheduledRef.current) {
|
|
console.log('SSE: Reconnect already scheduled, skipping duplicate')
|
|
return
|
|
}
|
|
|
|
if (reconnectTimeoutRef.current) {
|
|
clearTimeout(reconnectTimeoutRef.current)
|
|
}
|
|
|
|
console.log(
|
|
`SSE: Scheduling reconnect in ${delay}ms (attempt ${reconnectAttemptsRef.current + 1}, reason: ${reason})`,
|
|
)
|
|
|
|
isReconnectScheduledRef.current = true
|
|
nextReconnectTimeRef.current = Date.now() + delay
|
|
|
|
reconnectTimeoutRef.current = setTimeout(() => {
|
|
isReconnectScheduledRef.current = false
|
|
reconnectAttemptsRef.current++
|
|
nextReconnectTimeRef.current = null
|
|
// Note: connect will be called by the caller after this returns
|
|
// We need to trigger it here
|
|
window.dispatchEvent(new CustomEvent('sse-reconnect'))
|
|
}, delay)
|
|
}, [])
|
|
|
|
// Create connect function that can be called from anywhere
|
|
const connect = useCallback(async () => {
|
|
// Clear the scheduled flag when actually connecting
|
|
isReconnectScheduledRef.current = false
|
|
|
|
if (isCircuitBreakerOpen) {
|
|
console.log('SSE: Circuit breaker is open, preventing connection attempt')
|
|
showError({
|
|
title: 'Connection Temporarily Disabled',
|
|
message:
|
|
'Connection blocked due to repeated failures. Please try again later.',
|
|
})
|
|
return
|
|
}
|
|
|
|
if (reconnectAttemptsRef.current >= MAX_RECONNECT_ATTEMPTS) {
|
|
console.error(
|
|
'SSE: Maximum reconnection attempts reached, opening circuit breaker',
|
|
)
|
|
setIsCircuitBreakerOpen(true)
|
|
showError({
|
|
title: 'Connection Failed',
|
|
message:
|
|
'Maximum connection attempts reached. SSE disabled for 10 minutes.',
|
|
})
|
|
|
|
// Reset circuit breaker after timeout
|
|
setTimeout(() => {
|
|
console.log('SSE: Resetting circuit breaker')
|
|
setIsCircuitBreakerOpen(false)
|
|
reconnectAttemptsRef.current = 0
|
|
}, CIRCUIT_BREAKER_RESET_TIME)
|
|
return
|
|
}
|
|
|
|
// Prevent race conditions by checking if already connecting or connected
|
|
if (eventSourceRef.current?.readyState === SSE_STATES.OPEN) {
|
|
console.log('SSE: Already connected')
|
|
return // Already connected
|
|
}
|
|
|
|
if (eventSourceRef.current?.readyState === SSE_STATES.CONNECTING) {
|
|
console.log('SSE: Connection already in progress')
|
|
return // Already connecting
|
|
}
|
|
|
|
const sseConfig = getSSEUrl()
|
|
console.log('SSE connect - Config:', sseConfig)
|
|
|
|
if (!sseConfig) {
|
|
console.log('Cannot connect to SSE: missing URL, token, or user profile')
|
|
return
|
|
}
|
|
|
|
// Create connection logic inline to avoid circular dependency
|
|
try {
|
|
console.log('Connecting to SSE:', sseConfig.url)
|
|
setConnectionState(SSE_STATES.CONNECTING)
|
|
isManuallyClosedRef.current = false
|
|
|
|
if (Capacitor.isNativePlatform()) {
|
|
// Capacitor's native HTTP bridge does not support streaming responses,
|
|
// which breaks EventSourcePolyfill (fetch/XHR based). Use the native
|
|
// EventSource instead, which uses the WKWebView HTTP stack directly.
|
|
// Native EventSource cannot send custom headers, so we first exchange
|
|
// our JWT (sent in the Authorization header) for a short-lived,
|
|
// single-use ticket and pass that ticket as a query parameter. This
|
|
// keeps the long-lived token out of URLs and proxy access logs.
|
|
const ticket = await fetchSSETicket()
|
|
if (!ticket) {
|
|
console.error('SSE: Failed to obtain connection ticket')
|
|
setError('Connection error occurred')
|
|
setConnectionState(SSE_STATES.CLOSED)
|
|
scheduleReconnect(
|
|
RECONNECT_INTERVALS[
|
|
Math.min(
|
|
reconnectAttemptsRef.current,
|
|
RECONNECT_INTERVALS.length - 1,
|
|
)
|
|
],
|
|
'ticket-fetch-failed',
|
|
)
|
|
return
|
|
}
|
|
|
|
const nativeUrl = new URL(sseConfig.url)
|
|
nativeUrl.searchParams.set('ticket', ticket)
|
|
|
|
eventSourceRef.current = new EventSource(nativeUrl.toString(), {
|
|
withCredentials: true,
|
|
})
|
|
} else {
|
|
eventSourceRef.current = new EventSourcePolyfill(sseConfig.url, {
|
|
headers: {
|
|
Authorization: `Bearer ${localStorage.getItem('token')}`,
|
|
'Cache-Control': 'no-cache',
|
|
Accept: 'text/event-stream',
|
|
},
|
|
withCredentials: true,
|
|
heartbeatTimeout: 120000,
|
|
silentTimeoutRetry: true,
|
|
})
|
|
}
|
|
|
|
eventSourceRef.current.onopen = () => {
|
|
console.log('SSE connection opened')
|
|
setConnectionState(SSE_STATES.OPEN)
|
|
setError(null)
|
|
reconnectAttemptsRef.current = 0
|
|
nextReconnectTimeRef.current = null
|
|
isReconnectScheduledRef.current = false
|
|
lastHeartbeatRef.current = Date.now()
|
|
|
|
// Start heartbeat monitor
|
|
stopHeartbeatMonitor()
|
|
heartbeatMonitorRef.current = setInterval(() => {
|
|
const timeSinceLastHeartbeat = Date.now() - lastHeartbeatRef.current
|
|
const heartbeatTimeout = 150000 // 2.5 minutes
|
|
|
|
console.debug(
|
|
`SSE: Heartbeat check - ${Math.round(timeSinceLastHeartbeat / 1000)}s since last heartbeat`,
|
|
)
|
|
|
|
if (timeSinceLastHeartbeat > heartbeatTimeout) {
|
|
console.warn(
|
|
`SSE: No heartbeat received for ${Math.round(timeSinceLastHeartbeat / 1000)}s, connection may be stale. Reconnecting...`,
|
|
)
|
|
if (!isManuallyClosedRef.current) {
|
|
// Clear current heartbeat monitor before reconnecting
|
|
stopHeartbeatMonitor()
|
|
|
|
// Close current connection gracefully
|
|
if (eventSourceRef.current) {
|
|
eventSourceRef.current.close()
|
|
eventSourceRef.current = null
|
|
}
|
|
setConnectionState(SSE_STATES.CLOSED)
|
|
|
|
// Calculate delay based on current attempt
|
|
const attemptIndex = Math.min(
|
|
reconnectAttemptsRef.current,
|
|
RECONNECT_INTERVALS.length - 1,
|
|
)
|
|
const delay = RECONNECT_INTERVALS[attemptIndex]
|
|
|
|
// Schedule reconnect
|
|
if (reconnectTimeoutRef.current) {
|
|
clearTimeout(reconnectTimeoutRef.current)
|
|
}
|
|
|
|
console.log(
|
|
`SSE: Scheduling heartbeat-triggered reconnect in ${delay}ms (attempt ${reconnectAttemptsRef.current + 1})`,
|
|
)
|
|
|
|
isReconnectScheduledRef.current = true
|
|
nextReconnectTimeRef.current = Date.now() + delay
|
|
reconnectTimeoutRef.current = setTimeout(() => {
|
|
isReconnectScheduledRef.current = false
|
|
reconnectAttemptsRef.current++
|
|
nextReconnectTimeRef.current = null
|
|
connect()
|
|
}, delay)
|
|
}
|
|
}
|
|
}, 60000) // Check every minute
|
|
}
|
|
|
|
eventSourceRef.current.onmessage = handleSSEMessage
|
|
|
|
eventSourceRef.current.onerror = async error => {
|
|
console.error('SSE error:', error)
|
|
setConnectionState(SSE_STATES.CLOSED)
|
|
stopHeartbeatMonitor()
|
|
|
|
// Close the EventSource to prevent it from retrying on its own
|
|
if (eventSourceRef.current) {
|
|
eventSourceRef.current.close()
|
|
eventSourceRef.current = null
|
|
}
|
|
|
|
if (isManuallyClosedRef.current) {
|
|
console.log('SSE: Manually closed, not reconnecting')
|
|
return
|
|
}
|
|
|
|
// Check if reconnect is already scheduled
|
|
if (isReconnectScheduledRef.current) {
|
|
console.log('SSE: Reconnect already scheduled, skipping')
|
|
return
|
|
}
|
|
|
|
// Check if this is a 401 unauthorized error
|
|
const is401Error =
|
|
error.status === 401 ||
|
|
error.error?.message?.includes('401') ||
|
|
error.error?.message?.includes('Unauthorized')
|
|
|
|
// Check if this is a timeout error specifically
|
|
const isTimeoutError =
|
|
error.error?.message?.includes('No activity within') ||
|
|
error.error?.message?.includes('timeout')
|
|
|
|
if (is401Error) {
|
|
console.log('SSE 401 error detected, attempting token refresh...')
|
|
setError('Authentication expired - refreshing token...')
|
|
|
|
try {
|
|
const refreshResult = await apiClient.refreshToken()
|
|
|
|
if (refreshResult.success) {
|
|
console.log(
|
|
'Token refreshed successfully, retrying SSE connection...',
|
|
)
|
|
setError('Token refreshed - reconnecting...')
|
|
|
|
if (apiClient.failedQueue && apiClient.failedQueue.length > 0) {
|
|
console.log(
|
|
`Processing ${apiClient.failedQueue.length} queued requests after SSE token refresh`,
|
|
)
|
|
apiClient.processQueue(null, refreshResult.token)
|
|
}
|
|
|
|
// Reset reconnect attempts since we have a fresh token
|
|
reconnectAttemptsRef.current = 0
|
|
|
|
// Schedule immediate reconnect with fresh token
|
|
if (reconnectTimeoutRef.current) {
|
|
clearTimeout(reconnectTimeoutRef.current)
|
|
}
|
|
|
|
isReconnectScheduledRef.current = true
|
|
nextReconnectTimeRef.current = Date.now() + 1000
|
|
reconnectTimeoutRef.current = setTimeout(() => {
|
|
isReconnectScheduledRef.current = false
|
|
nextReconnectTimeRef.current = null
|
|
connect()
|
|
}, 1000)
|
|
|
|
return
|
|
} else if (
|
|
refreshResult.error === 'Already refreshing' ||
|
|
refreshResult.error === 'Refresh cooldown active'
|
|
) {
|
|
console.log(
|
|
'SSE: Token refresh in progress by another request, waiting...',
|
|
)
|
|
setError('Token refresh in progress - reconnecting soon...')
|
|
|
|
reconnectAttemptsRef.current = 0
|
|
|
|
if (reconnectTimeoutRef.current) {
|
|
clearTimeout(reconnectTimeoutRef.current)
|
|
}
|
|
|
|
isReconnectScheduledRef.current = true
|
|
nextReconnectTimeRef.current = Date.now() + 1500
|
|
reconnectTimeoutRef.current = setTimeout(() => {
|
|
isReconnectScheduledRef.current = false
|
|
nextReconnectTimeRef.current = null
|
|
connect()
|
|
}, 1500)
|
|
|
|
return
|
|
} else if (refreshResult.error === 'Refresh token expired') {
|
|
console.error('Refresh token expired, user must login again')
|
|
setError('Session expired - please log in again')
|
|
return
|
|
} else {
|
|
console.error('Token refresh failed:', refreshResult.error)
|
|
setError('Authentication failed - please log in again')
|
|
return
|
|
}
|
|
} catch (refreshError) {
|
|
console.error('Token refresh error:', refreshError)
|
|
setError('Authentication error - please log in again')
|
|
return
|
|
}
|
|
} else if (isTimeoutError) {
|
|
console.log('SSE timeout detected, attempting reconnection...')
|
|
setError('Connection timeout - reconnecting...')
|
|
} else {
|
|
setError('Connection error occurred')
|
|
}
|
|
|
|
// Schedule reconnect for non-401 errors
|
|
const attemptIndex = Math.min(
|
|
reconnectAttemptsRef.current,
|
|
RECONNECT_INTERVALS.length - 1,
|
|
)
|
|
const delay = RECONNECT_INTERVALS[attemptIndex]
|
|
|
|
console.log(
|
|
`SSE: Scheduling error-triggered reconnect in ${delay}ms (attempt ${reconnectAttemptsRef.current + 1})`,
|
|
)
|
|
|
|
if (reconnectTimeoutRef.current) {
|
|
clearTimeout(reconnectTimeoutRef.current)
|
|
}
|
|
|
|
isReconnectScheduledRef.current = true
|
|
nextReconnectTimeRef.current = Date.now() + delay
|
|
reconnectTimeoutRef.current = setTimeout(() => {
|
|
isReconnectScheduledRef.current = false
|
|
reconnectAttemptsRef.current++
|
|
nextReconnectTimeRef.current = null
|
|
connect()
|
|
}, delay)
|
|
}
|
|
} catch (err) {
|
|
console.error('Failed to create SSE connection:', err)
|
|
showError({
|
|
title: 'Connection Error',
|
|
message: 'Failed to establish real-time connection. Please try again.',
|
|
})
|
|
setConnectionState(SSE_STATES.CLOSED)
|
|
}
|
|
}, [
|
|
getSSEUrl,
|
|
fetchSSETicket,
|
|
handleSSEMessage,
|
|
stopHeartbeatMonitor,
|
|
scheduleReconnect,
|
|
isCircuitBreakerOpen,
|
|
showError,
|
|
])
|
|
|
|
const disconnect = useCallback(() => {
|
|
isManuallyClosedRef.current = true
|
|
isReconnectScheduledRef.current = false
|
|
|
|
if (reconnectTimeoutRef.current) {
|
|
clearTimeout(reconnectTimeoutRef.current)
|
|
reconnectTimeoutRef.current = null
|
|
}
|
|
|
|
nextReconnectTimeRef.current = null
|
|
stopHeartbeatMonitor()
|
|
|
|
if (eventSourceRef.current) {
|
|
eventSourceRef.current.close()
|
|
eventSourceRef.current = null
|
|
}
|
|
|
|
setConnectionState(SSE_STATES.CLOSED)
|
|
}, [stopHeartbeatMonitor])
|
|
|
|
const toggleSSEEnabled = useCallback(
|
|
enabled => {
|
|
console.log('SSE toggleSSEEnabled called:', {
|
|
enabled,
|
|
isTokenValid: isAuthenticated,
|
|
})
|
|
localStorage.setItem('sse_enabled', enabled.toString())
|
|
if (enabled && isAuthenticated) {
|
|
console.log('SSE toggleSSEEnabled: Calling connect()')
|
|
connect()
|
|
} else {
|
|
console.log('SSE toggleSSEEnabled: Calling disconnect()')
|
|
disconnect()
|
|
}
|
|
},
|
|
[connect, disconnect, isAuthenticated],
|
|
)
|
|
|
|
const isSSEEnabled = useCallback(() => {
|
|
return localStorage.getItem('sse_enabled') === 'true'
|
|
}, [])
|
|
|
|
// Auto-connect when SSE is enabled and token is valid
|
|
useEffect(() => {
|
|
console.log('SSE auto-connect effect triggered')
|
|
console.log('Token valid:', isAuthenticated)
|
|
|
|
const isSSEEnabledSetting = localStorage.getItem('sse_enabled') === 'true'
|
|
console.log('SSE enabled in settings:', isSSEEnabledSetting)
|
|
|
|
if (isAuthenticated && isSSEEnabledSetting) {
|
|
console.log('SSE: Conditions met, attempting to connect')
|
|
connect()
|
|
} else {
|
|
console.log('SSE: Conditions not met, disconnecting')
|
|
disconnect()
|
|
}
|
|
|
|
return () => {
|
|
disconnect()
|
|
}
|
|
}, [isAuthenticated]) // Fixed: Added isAuthenticated dependency
|
|
|
|
// Cleanup timeouts on unmount
|
|
useEffect(() => {
|
|
return () => {
|
|
if (reconnectTimeoutRef.current) {
|
|
clearTimeout(reconnectTimeoutRef.current)
|
|
}
|
|
stopHeartbeatMonitor()
|
|
}
|
|
}, [stopHeartbeatMonitor])
|
|
|
|
// Update EventSource message handler when handleSSEMessage changes
|
|
useEffect(() => {
|
|
if (
|
|
eventSourceRef.current &&
|
|
eventSourceRef.current.readyState === SSE_STATES.OPEN
|
|
) {
|
|
console.log('SSE: Updating message handler with latest userProfile')
|
|
eventSourceRef.current.onmessage = handleSSEMessage
|
|
}
|
|
}, [handleSSEMessage])
|
|
|
|
// Handle visibility changes for better performance
|
|
useEffect(() => {
|
|
const handleVisibilityChange = () => {
|
|
if (document.hidden) {
|
|
console.log(
|
|
'SSE: App backgrounded, maintaining connection but reducing activity',
|
|
)
|
|
} else {
|
|
console.log('SSE: App foregrounded, ensuring connection is active')
|
|
|
|
const isSSEEnabledSetting =
|
|
localStorage.getItem('sse_enabled') === 'true'
|
|
|
|
// Check actual EventSource state, not React state
|
|
const isCurrentlyConnected =
|
|
eventSourceRef.current?.readyState === SSE_STATES.OPEN
|
|
const isCurrentlyConnecting =
|
|
eventSourceRef.current?.readyState === SSE_STATES.CONNECTING
|
|
|
|
if (
|
|
isAuthenticated &&
|
|
isSSEEnabledSetting &&
|
|
!isCurrentlyConnected &&
|
|
!isCurrentlyConnecting &&
|
|
!isReconnectScheduledRef.current
|
|
) {
|
|
console.log('SSE: Reconnecting after visibility change')
|
|
connect()
|
|
}
|
|
}
|
|
}
|
|
|
|
document.addEventListener('visibilitychange', handleVisibilityChange)
|
|
|
|
return () => {
|
|
document.removeEventListener('visibilitychange', handleVisibilityChange)
|
|
}
|
|
}, [connect, isAuthenticated])
|
|
|
|
return {
|
|
connectionState,
|
|
isConnected: connectionState === SSE_STATES.OPEN,
|
|
isConnecting: connectionState === SSE_STATES.CONNECTING,
|
|
lastEvent,
|
|
error,
|
|
connect,
|
|
disconnect,
|
|
toggleSSEEnabled,
|
|
isSSEEnabled,
|
|
getConnectionStatus: () => {
|
|
switch (connectionState) {
|
|
case SSE_STATES.CONNECTING:
|
|
return 'connecting'
|
|
case SSE_STATES.OPEN:
|
|
return 'connected'
|
|
case SSE_STATES.CLOSED:
|
|
default:
|
|
return 'disconnected'
|
|
}
|
|
},
|
|
getDebugInfo: () => ({
|
|
connectionState,
|
|
reconnectAttempts: reconnectAttemptsRef.current,
|
|
isCircuitBreakerOpen,
|
|
isReconnectScheduled: isReconnectScheduledRef.current,
|
|
lastHeartbeat: lastHeartbeatRef.current,
|
|
timeSinceLastHeartbeat: Date.now() - lastHeartbeatRef.current,
|
|
isManuallyCloseRef: isManuallyClosedRef.current,
|
|
nextReconnectTime: nextReconnectTimeRef.current,
|
|
timeUntilReconnect: nextReconnectTimeRef.current
|
|
? nextReconnectTimeRef.current - Date.now()
|
|
: null,
|
|
reconnectIntervals: RECONNECT_INTERVALS,
|
|
currentReconnectDelay:
|
|
reconnectAttemptsRef.current < RECONNECT_INTERVALS.length
|
|
? RECONNECT_INTERVALS[reconnectAttemptsRef.current]
|
|
: RECONNECT_INTERVALS[RECONNECT_INTERVALS.length - 1],
|
|
maxReconnectAttempts: MAX_RECONNECT_ATTEMPTS,
|
|
circuitBreakerResetTime: CIRCUIT_BREAKER_RESET_TIME,
|
|
heartbeatTimeout: 120000,
|
|
heartbeatMonitorInterval: 60000,
|
|
heartbeatMonitorTimeout: 150000,
|
|
}),
|
|
}
|
|
}
|