boursecu_user/src/utils/priceSocket.ts

import type { Price } from '@/models/home'
import type { Product } from '@/models/products'
import { Modal } from 'ant-design-vue'
import { getConfigApp } from '@/utils/config'
import { priceSocketConnected, priceWssEnabled } from '@/utils/priceSocketState'

export { priceSocketConnected, priceWssEnabled, shouldUsePricePolling } from '@/utils/priceSocketState'

const RECONNECT_BASE_MS = 3000
const RECONNECT_MAX_MS = 30000
const FAILURES_BEFORE_ALERT = 4
const STALE_STREAM_MS = 45000

let eventSource: EventSource | null = null
let subscribers = 0
let stopped = false
let getSetting: ((key: string) => unknown) | null = null
let reconnectTimer: ReturnType<typeof setTimeout> | undefined
let staleCheckTimer: ReturnType<typeof setInterval> | undefined
let reconnectAttempt = 0
let lastStreamEventAt = 0
let networkAlertOpen = false
let listenersBound = false

function isTruthySetting(value: unknown, defaultValue = true): boolean {
  if (value === null || value === undefined || value === '') {
    return defaultValue
  }

  return ['true', '1', 'yes', 'on'].includes(String(value).toLowerCase())
}

function resolveStreamPath(): string {
  const ssePath = String(getSetting?.('sse_path') ?? '').trim()
  if (ssePath && !ssePath.includes('/prc/')) {
    return ssePath.startsWith('/') ? ssePath : `/${ssePath}`
  }

  const legacyPath = String(getSetting?.('wss_path') ?? '').trim()
  if (legacyPath && !legacyPath.includes('/prc/')) {
    return legacyPath.startsWith('/') ? legacyPath : `/${legacyPath}`
  }

  return '/api/price/stream'
}

function resolveStreamUrl(): string | null {
  if (!getSetting) {
    return null
  }

  if (!isTruthySetting(getSetting('wss_enabled'), true)) {
    return null
  }

  const base = String(getConfigApp('node_server') || window.location.origin).replace(/\/$/, '')
  return `${base}${resolveStreamPath()}`
}

function clearReconnectTimer() {
  if (reconnectTimer) {
    clearTimeout(reconnectTimer)
    reconnectTimer = undefined
  }
}

function stopStaleCheck() {
  if (staleCheckTimer) {
    clearInterval(staleCheckTimer)
    staleCheckTimer = undefined
  }
}

function touchStreamActivity() {
  lastStreamEventAt = Date.now()
}

function startStaleCheck() {
  stopStaleCheck()
  touchStreamActivity()
  staleCheckTimer = setInterval(() => {
    if (
      stopped
      || subscribers <= 0
      || !eventSource
      || eventSource.readyState !== EventSource.OPEN
    ) {
      return
    }

    if (Date.now() - lastStreamEventAt > STALE_STREAM_MS) {
      console.warn('Price SSE stale — reconnecting')
      handleStreamLost(true)
    }
  }, 15000)
}

async function runFallbackPoll() {
  if (!priceWssEnabled.value || priceSocketConnected.value) {
    return
  }

  try {
    const { getLastPrices } = await import('@/utils/price')
    const { refreshProductsPageData } = await import(
      '@/views/base/products/composables/useProductsPageData'
    )
    await Promise.allSettled([getLastPrices(), refreshProductsPageData()])
  } catch (error) {
    console.warn('Price fallback poll failed', error)
  }
}

function showNetworkAlert() {
  if (networkAlertOpen || stopped || !priceWssEnabled.value) {
    return
  }

  networkAlertOpen = true
  Modal.confirm({
    title: 'اتصال قیمت برقرار نیست',
    content:
      'به‌روزرسانی لحظه‌ای قیمت قطع شده است. اتصال اینترنت را بررسی کنید یا دوباره تلاش کنید. تا زمان اتصال مجدد، قیمت با تأخیر به‌روز می‌شود.',
    okText: 'تلاش مجدد',
    cancelText: 'بستن',
    centered: true,
    onOk: () => {
      networkAlertOpen = false
      reconnectAttempt = 0
      forceReconnect()
    },
    onCancel: () => {
      networkAlertOpen = false
    },
  })
}

function maybeShowNetworkAlert() {
  if (reconnectAttempt >= FAILURES_BEFORE_ALERT || !navigator.onLine) {
    showNetworkAlert()
  }
}

function scheduleReconnect() {
  if (stopped || subscribers <= 0) {
    return
  }

  clearReconnectTimer()
  const delay = Math.min(
    RECONNECT_BASE_MS * Math.pow(2, reconnectAttempt),
    RECONNECT_MAX_MS,
  )

  reconnectTimer = setTimeout(() => {
    reconnectAttempt += 1
    connect()
    maybeShowNetworkAlert()
  }, delay)
}

async function applyLastFromStream(last: Price) {
  const { applyLastPricePayload } = await import('@/utils/price')
  applyLastPricePayload(last)
}

function handleLastPayload(payload: unknown) {
  if (!payload || typeof payload !== 'object') {
    return
  }

  const last = payload as Price
  if (last.price_last == null && last.price == null && last.buy == null) {
    return
  }

  touchStreamActivity()
  void applyLastFromStream(last)
}

async function applyListFromStream(products: Product[]) {
  const { applyProductsListFromSocket } = await import(
    '@/views/base/products/composables/useProductsPageData'
  )
  applyProductsListFromSocket(products)
}

function handleListPayload(payload: unknown) {
  if (!Array.isArray(payload) || !payload.length) {
    return
  }

  touchStreamActivity()
  void applyListFromStream(payload as Product[])
}

function bindStreamEvents(source: EventSource) {
  source.addEventListener('last', (event) => {
    try {
      handleLastPayload(JSON.parse(event.data))
    } catch (error) {
      console.warn('Price SSE last parse failed', error)
    }
  })

  source.addEventListener('list', (event) => {
    try {
      handleListPayload(JSON.parse(event.data))
    } catch (error) {
      console.warn('Price SSE list parse failed', error)
    }
  })

  source.onmessage = () => {
    touchStreamActivity()
  }
}

function handleStreamLost(scheduleRetry = true) {
  const wasConnected = priceSocketConnected.value
  disconnectSourceOnly()
  priceSocketConnected.value = false

  if (wasConnected || scheduleRetry) {
    void runFallbackPoll()
  }

  if (scheduleRetry && !stopped && subscribers > 0) {
    scheduleReconnect()
    maybeShowNetworkAlert()
  }
}

function disconnectSourceOnly() {
  stopStaleCheck()
  if (eventSource) {
    eventSource.onopen = null
    eventSource.onerror = null
    eventSource.onmessage = null
    eventSource.close()
    eventSource = null
  }
}

function connect() {
  if (stopped || subscribers <= 0) {
    return
  }

  const url = resolveStreamUrl()
  if (!url) {
    priceSocketConnected.value = false
    return
  }

  if (eventSource) {
    const state = eventSource.readyState
    if (state === EventSource.OPEN || state === EventSource.CONNECTING) {
      return
    }
    disconnectSourceOnly()
  }

  try {
    eventSource = new EventSource(url)
  } catch (error) {
    console.warn('Price SSE connect failed', error)
    handleStreamLost()
    return
  }

  bindStreamEvents(eventSource)

  eventSource.onopen = () => {
    reconnectAttempt = 0
    clearReconnectTimer()
    networkAlertOpen = false
    priceSocketConnected.value = true
    touchStreamActivity()
    startStaleCheck()
  }

  eventSource.onerror = () => {
    const state = eventSource?.readyState
    if (state === EventSource.OPEN) {
      return
    }

    if (state === EventSource.CONNECTING) {
      priceSocketConnected.value = false
      void runFallbackPoll()
      return
    }

    handleStreamLost()
  }
}

function disconnect() {
  clearReconnectTimer()
  disconnectSourceOnly()
  priceSocketConnected.value = false
}

export function forceReconnect() {
  if (stopped || subscribers <= 0) {
    return
  }

  clearReconnectTimer()
  disconnect()
  connect()
}

function onBrowserOnline() {
  if (stopped || subscribers <= 0) {
    return
  }

  reconnectAttempt = 0
  networkAlertOpen = false
  forceReconnect()
}

function onBrowserOffline() {
  handleStreamLost(true)
  showNetworkAlert()
}

function bindBrowserNetworkListeners() {
  if (listenersBound || typeof window === 'undefined') {
    return
  }

  listenersBound = true
  window.addEventListener('online', onBrowserOnline)
  window.addEventListener('offline', onBrowserOffline)
}

function unbindBrowserNetworkListeners() {
  if (!listenersBound || typeof window === 'undefined') {
    return
  }

  listenersBound = false
  window.removeEventListener('online', onBrowserOnline)
  window.removeEventListener('offline', onBrowserOffline)
}

export function startPriceSocket(settingGetter: (key: string) => unknown) {
  getSetting = settingGetter
  subscribers += 1
  stopped = false
  connect()
  bindBrowserNetworkListeners()

  if (typeof document !== 'undefined') {
    document.addEventListener('visibilitychange', onVisibilityChange)
  }
}

export function stopPriceSocket() {
  subscribers = Math.max(0, subscribers - 1)
  if (subscribers > 0) {
    return
  }

  stopped = true
  disconnect()
  unbindBrowserNetworkListeners()

  if (typeof document !== 'undefined') {
    document.removeEventListener('visibilitychange', onVisibilityChange)
  }
}

function onVisibilityChange() {
  if (document.visibilityState !== 'visible' || stopped || subscribers <= 0) {
    return
  }

  if (!eventSource || eventSource.readyState !== EventSource.OPEN) {
    reconnectAttempt = 0
    forceReconnect()
  }
}

export function isPriceWebSocketEnabled(settingGetter: (key: string) => unknown): boolean {
  getSetting = settingGetter
  return resolveStreamUrl() !== null
}

let channelStarted = false

export function initPriceRealtimeChannel(settingGetter: (key: string) => unknown) {
  priceWssEnabled.value = isTruthySetting(settingGetter('wss_enabled'), true)

  if (channelStarted) {
    return
  }

  channelStarted = true

  if (!priceWssEnabled.value) {
    return
  }

  if (!isPriceWebSocketEnabled(settingGetter)) {
    return
  }

  startPriceSocket(settingGetter)
}