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 { findStoredAuthToken } from '@/utils/authToken'
import {
  priceSocketConnected,
  priceStreamConnectingSince,
  priceStreamLifecycle,
  priceWssEnabled,
} from '@/utils/priceSocketState'
import {
  applyBoardSpecialToLastPayload,
  setSpecialPricingPayload,
} from '@/utils/specialPricing'

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

const RECONNECT_BASE_MS = 3000
const RECONNECT_MAX_MS = 30000
/** چند بار خطای پشت‌سرهم /api/price/last قبل از نمایش مدال */
const LAST_FETCH_FAILURES_BEFORE_ALERT = 2
const ALERT_COOLDOWN_MS = 5 * 60 * 1000
const FALLBACK_POLL_INTERVAL_MS = 10000
/** بدون قیمت معتبر از مرجع (SSE یا API) */
const PRICE_SOURCE_STALE_MS_MOBILE = 50000
const PRICE_SOURCE_STALE_MS_DESKTOP = 35000
/** بدون event قیمت؛ ping هر ۵ث از سرور activity را تازه می‌کند */
const STALE_STREAM_MS = 120000
const FALLBACK_POLL_DELAY_MS = 2500
const STREAM_CONNECTING = 0
const STREAM_OPEN = 1
const STREAM_CLOSED = 2

/** fetch + ReadableStream — با توکن اختلاف قیمت ویژه؛ بدون توکن همان قیمت عمومی (مهمان). */
class AuthPriceStream extends EventTarget {
  readyState = STREAM_CONNECTING
  onopen: (() => void) | null = null
  onerror: (() => void) | null = null
  onmessage: (() => void) | null = null
  private abort = new AbortController()

  constructor(url: string, token?: string) {
    super()
    void this.start(url, token)
  }

  close() {
    this.readyState = STREAM_CLOSED
    this.abort.abort()
  }

  private async start(url: string, token?: string) {
    try {
      const headers: Record<string, string> = {
        Accept: 'text/event-stream',
      }
      const bearer = String(token ?? '').trim()
      if (bearer) {
        headers.Authorization = `Bearer ${bearer}`
      }

      const response = await fetch(url, {
        headers,
        cache: 'no-store',
        signal: this.abort.signal,
      })

      if (!response.ok || !response.body) {
        this.readyState = STREAM_CLOSED
        this.onerror?.()
        return
      }

      this.readyState = STREAM_OPEN
      this.onopen?.()

      const reader = response.body.getReader()
      const decoder = new TextDecoder()
      let buffer = ''

      while (true) {
        const { done, value } = await reader.read()
        if (done) {
          break
        }
        buffer += decoder.decode(value, { stream: true })
        buffer = this.flushSse(buffer)
      }

      this.readyState = STREAM_CLOSED
      this.onerror?.()
    } catch {
      if (this.abort.signal.aborted) {
        return
      }
      this.readyState = STREAM_CLOSED
      this.onerror?.()
    }
  }

  private flushSse(buffer: string): string {
    const blocks = buffer.split('\n\n')
    const rest = blocks.pop() ?? ''

    for (const block of blocks) {
      let event = 'message'
      const data: string[] = []
      for (const line of block.split('\n')) {
        if (line.startsWith('event:')) {
          event = line.slice(6).trim()
        } else if (line.startsWith('data:')) {
          data.push(line.slice(5).trimStart())
        }
      }
      this.dispatchEvent(new MessageEvent(event, { data: data.join('\n') }))
      this.onmessage?.()
    }

    return rest
  }
}

let eventSource: AuthPriceStream | 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 lastNetworkAlertAt = 0
let listenersBound = false
let fallbackPollTimer: ReturnType<typeof setTimeout> | undefined
let fallbackPollInterval: ReturnType<typeof setInterval> | undefined
let pendingPriceAlert = false

function isMobileDevice(): boolean {
  if (typeof navigator === 'undefined') {
    return false
  }

  return /iPhone|iPad|iPod|Android|Mobile/i.test(navigator.userAgent)
}

function isPageVisible(): boolean {
  return typeof document === 'undefined' || document.visibilityState === 'visible'
}

function markConnectingLifecycle() {
  priceStreamLifecycle.value = 'connecting'
  if (priceStreamConnectingSince.value <= 0) {
    priceStreamConnectingSince.value = Date.now()
  }
}

function clearConnectingLifecycle() {
  priceStreamConnectingSince.value = 0
}

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 !== STREAM_OPEN
    ) {
      return
    }

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

function clearFallbackPollTimer() {
  if (fallbackPollTimer) {
    clearTimeout(fallbackPollTimer)
    fallbackPollTimer = undefined
  }
}

function stopFallbackPolling() {
  clearFallbackPollTimer()
  if (fallbackPollInterval) {
    clearInterval(fallbackPollInterval)
    fallbackPollInterval = undefined
  }
}

function startFallbackPolling() {
  if (fallbackPollInterval) {
    return
  }

  scheduleFallbackPoll()
  fallbackPollInterval = setInterval(() => {
    if (stopped || subscribers <= 0 || priceSocketConnected.value) {
      stopFallbackPolling()
      return
    }

    void runFallbackPoll()
  }, FALLBACK_POLL_INTERVAL_MS)
}

function scheduleFallbackPoll() {
  clearFallbackPollTimer()
  fallbackPollTimer = setTimeout(() => {
    fallbackPollTimer = undefined
    if (!stopped && subscribers > 0 && !priceSocketConnected.value) {
      void runFallbackPoll()
    }
  }, FALLBACK_POLL_DELAY_MS)
}

export function clearPriceNetworkAlert() {
  networkAlertOpen = false
  pendingPriceAlert = false
}

/** وقتی /api/price/last خطا می‌دهد (از polling عادی یا fallback) */
export function notifyPriceLastFetchFailed() {
  void maybeShowNetworkAlert()
}

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

  try {
    const { getLastPrices, hasRecentPriceFromSource, markPriceSourceFetchFailed } = await import(
      '@/utils/price'
    )
    const { refreshProductsPageData } = await import(
      '@/views/base/products/composables/useProductsPageData'
    )

    const pollTasks: Promise<unknown>[] = [getLastPrices()]
    if (findStoredAuthToken()) {
      pollTasks.push(refreshProductsPageData())
    }

    const [priceResult] = await Promise.allSettled(pollTasks)

    const priceOk =
      priceResult.status === 'fulfilled'
      && priceResult.value === true
      && hasRecentPriceFromSource(getPriceSourceStaleMs())

    if (priceOk) {
      clearPriceNetworkAlert()
      return
    }

    void maybeShowNetworkAlert()
  } catch (error) {
    console.warn('Price fallback poll failed', error)
    const { markPriceSourceFetchFailed } = await import('@/utils/price')
    markPriceSourceFetchFailed()
    void maybeShowNetworkAlert()
  }
}

function getPriceSourceStaleMs(): number {
  return isMobileDevice() ? PRICE_SOURCE_STALE_MS_MOBILE : PRICE_SOURCE_STALE_MS_DESKTOP
}

async function isRealPriceOutage(): Promise<boolean> {
  const {
    hasRecentPriceFromSource,
    consecutivePriceSourceFailures,
  } = await import('@/utils/price')

  const staleMs = getPriceSourceStaleMs()

  if (hasRecentPriceFromSource(staleMs)) {
    pendingPriceAlert = false
    return false
  }

  return consecutivePriceSourceFailures() >= LAST_FETCH_FAILURES_BEFORE_ALERT
}

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

  if (!isPageVisible()) {
    pendingPriceAlert = true
    return
  }

  const now = Date.now()
  if (now - lastNetworkAlertAt < ALERT_COOLDOWN_MS) {
    return
  }

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

async function maybeShowNetworkAlert() {
  if (!(await isRealPriceOutage())) {
    return
  }

  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()
  }, delay)
}

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

async function applySettingsFromStream(payload: unknown) {
  const items = Array.isArray(payload)
    ? payload
    : payload && typeof payload === 'object' && Array.isArray((payload as { items?: unknown }).items)
      ? (payload as { items: unknown[] }).items
      : null
  if (!items?.length) {
    return
  }

  try {
    const { useSettingsStore } = await import('@/store/settingsStore')
    const { touchStoreHoursSettings } = await import('@/utils/storeHours')
    useSettingsStore().mergeSettings(items as Array<{ key?: string; value?: unknown }>)
    touchStoreHoursSettings()
  } catch (error) {
    console.warn('Settings SSE apply failed', error)
  }
}

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

  const raw = payload as Price & { settings_patch?: unknown }
  const last = applyBoardSpecialToLastPayload(raw)
  const hasDisplayable =
    Number(last.buy ?? 0) > 0
    || Number(last.sell ?? 0) > 0
    || Number(last.price_last ?? 0) > 0
    || Number(last.price ?? 0) > 0
  const hasStatus = last.status != null || typeof last.closed === 'boolean'

  if (!hasDisplayable && !hasStatus) {
    if (Array.isArray(raw.settings_patch) && raw.settings_patch.length) {
      touchStreamActivity()
      void applySettingsFromStream(raw.settings_patch)
    }
    return
  }

  // SSE گاهی last خالی/no_quote می‌فرستد؛ قیمت معتبر فعلی را پاک نکن
  if (!hasDisplayable) {
    if (Array.isArray(raw.settings_patch) && raw.settings_patch.length) {
      touchStreamActivity()
      void applySettingsFromStream(raw.settings_patch)
    }
    void import('@/utils/price').then(({ getTradeBuyPrice, lastPrice }) => {
      const keepCurrent =
        getTradeBuyPrice(lastPrice.value) > 0 && last.status !== 'closed'
      touchStreamActivity()
      if (keepCurrent) return
      void applyLastFromStream(last)
    })
    return
  }

  touchStreamActivity()
  void applyLastFromStream(last)
}

function handleSettingsPayload(payload: unknown) {
  touchStreamActivity()
  void applySettingsFromStream(payload)
}

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

async function applyAssetsFromStream(payload: unknown) {
  const { applyWalletAssetsFromStream } = await import('@/composables/useWalletAssets')
  applyWalletAssetsFromStream(payload)
}

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

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

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

  touchStreamActivity()
  void applyAssetsFromStream(payload)
}

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

  touchStreamActivity()
  void applyAppNotificationFromStream(payload as Record<string, unknown>)
}

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

  touchStreamActivity()
  void applyUnreadCountFromStreamPayload(payload as Record<string, unknown>)
}

async function applyAppNotificationFromStream(payload: Record<string, unknown>) {
  try {
    const { bumpUnreadFromStream } = await import('@/composables/useAppNotificationBadge')
    bumpUnreadFromStream(payload)
  } catch (error) {
    console.warn('App notification SSE apply failed', error)
  }
}

async function applyUnreadCountFromStreamPayload(payload: Record<string, unknown>) {
  try {
    const { applyUnreadCountFromStream } = await import('@/composables/useAppNotificationBadge')
    applyUnreadCountFromStream(payload)
  } catch (error) {
    console.warn('Unread-count SSE apply failed', error)
  }
}

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

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

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

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

  source.addEventListener('app_notification', (event) => {
    try {
      handleAppNotificationPayload(JSON.parse(event.data))
    } catch (error) {
      console.warn('App notification SSE parse failed', error)
    }
  })

  source.addEventListener('unread-count', (event) => {
    try {
      handleUnreadCountPayload(JSON.parse(event.data))
    } catch (error) {
      console.warn('Unread-count SSE parse failed', error)
    }
  })

  source.addEventListener('special-pricing', (event) => {
    try {
      setSpecialPricingPayload(JSON.parse(event.data))
    } catch (error) {
      console.warn('Special-pricing SSE parse failed', error)
    }
  })

  source.addEventListener('ping', () => {
    touchStreamActivity()
  })

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

function handleStreamLost(scheduleRetry = true) {
  const wasConnected = priceSocketConnected.value
  disconnectSourceOnly()
  priceSocketConnected.value = false
  priceStreamLifecycle.value = scheduleRetry ? 'connecting' : 'closed'

  if ((wasConnected || scheduleRetry) && scheduleRetry) {
    startFallbackPolling()
  }

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

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()
  const token = findStoredAuthToken()
  if (!url) {
    priceSocketConnected.value = false
    priceStreamLifecycle.value = 'closed'
    clearConnectingLifecycle()
    return
  }

  if (eventSource) {
    const state = eventSource.readyState
    if (state === STREAM_OPEN || state === STREAM_CONNECTING) {
      return
    }
    disconnectSourceOnly()
  }

  markConnectingLifecycle()

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

  bindStreamEvents(eventSource)

  eventSource.onopen = () => {
    reconnectAttempt = 0
    clearReconnectTimer()
    stopFallbackPolling()
    networkAlertOpen = false
    priceStreamLifecycle.value = 'open'
    clearConnectingLifecycle()
    priceSocketConnected.value = true
    touchStreamActivity()
    startStaleCheck()
  }

  eventSource.onerror = () => {
    const state = eventSource?.readyState
    if (state === STREAM_OPEN || state === STREAM_CONNECTING) {
      return
    }

    handleStreamLost()
  }
}

function disconnect() {
  clearReconnectTimer()
  stopFallbackPolling()
  disconnectSourceOnly()
  priceSocketConnected.value = false
  priceStreamLifecycle.value = 'idle'
  clearConnectingLifecycle()
}

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)
  void maybeShowNetworkAlert()
}

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)
  }

  if (typeof window !== 'undefined') {
    window.addEventListener('pageshow', onPageShow)
  }
}

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)
  }

  if (typeof window !== 'undefined') {
    window.removeEventListener('pageshow', onPageShow)
  }
}

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

  if (document.visibilityState === 'hidden') {
    reconnectAttempt = 0
    return
  }

  reconnectAttempt = 0
  networkAlertOpen = false

  if (pendingPriceAlert) {
    void maybeShowNetworkAlert()
  }

  if (!eventSource || eventSource.readyState !== STREAM_OPEN) {
    forceReconnect()
  }
}

function onPageShow(event: PageTransitionEvent) {
  if (stopped || subscribers <= 0) {
    return
  }

  if (event.persisted || !eventSource || eventSource.readyState !== STREAM_OPEN) {
    reconnectAttempt = 0
    networkAlertOpen = false
    forceReconnect()
    if (pendingPriceAlert) {
      void maybeShowNetworkAlert()
    }
  }
}

export function isPriceWebSocketEnabled(settingGetter: (key: string) => unknown): boolean {
  getSetting = settingGetter
  if (!isTruthySetting(settingGetter('wss_enabled'), true)) {
    return false
  }

  return resolveStreamUrl() !== null
}

let channelStarted = false

function shutdownPriceRealtimeChannel() {
  if (!channelStarted && subscribers <= 0) {
    return
  }

  channelStarted = false
  stopped = true
  subscribers = 0
  disconnect()
  unbindBrowserNetworkListeners()

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

  if (typeof window !== 'undefined') {
    window.removeEventListener('pageshow', onPageShow)
  }
}

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

  if (!priceWssEnabled.value || !isPriceWebSocketEnabled(settingGetter)) {
    shutdownPriceRealtimeChannel()
    return
  }

  if (channelStarted && subscribers > 0) {
    forceReconnect()
    return
  }

  channelStarted = true
  stopped = false
  startPriceSocket(settingGetter)
}

export function initPriceRealtimeChannel(settingGetter: (key: string) => unknown) {
  syncPriceRealtimeChannel(settingGetter)
}

export function stopPriceRealtimeAuth() {
  shutdownPriceRealtimeChannel()
}