MikrotikManager/hooks/use-flow-live.ts

"use client"

import { useEffect, useState } from "react"
import type { FlowAnalyticsDto } from "@mmapp/contracts/traffic-flow"
import { resolveApiUrl, withAuthHeaders } from "@/shared/api/http-client"
import { flowQuery } from "@/shared/api/traffic-flow"

function parseSseBlock(block: string): { event: string; data: string } {
  let event = "message"
  const dataLines: string[] = []
  for (const line of block.split("\n")) {
    if (line.startsWith("event:")) event = line.slice(6).trim()
    else if (line.startsWith("data:")) dataLines.push(line.slice(5).trim())
  }
  return { event, data: dataLines.join("\n") }
}

export function isValidFlowLiveSample(value: unknown): value is FlowAnalyticsDto {
  if (!value || typeof value !== "object") return false
  const v = value as Partial<FlowAnalyticsDto>
  return typeof v.uniqueSrc === "number" && Array.isArray(v.destinations)
}

export function useFlowLive(opts: {
  enabled: boolean
  backendUrl: string
  range: string
  serverId?: string
  userId?: string
  iface?: string
  dedup?: boolean
  excludeMesh?: boolean
  excludeOverlay?: boolean
}): { sample: FlowAnalyticsDto | null; error: string | null } {
  const [sample, setSample] = useState<FlowAnalyticsDto | null>(null)
  const [error, setError] = useState<string | null>(null)

  useEffect(() => {
    if (!opts.enabled) {
      setSample(null)
      setError(null)
      return
    }

    const ac = new AbortController()
    setSample(null)
    setError(null)
    const path = `/api/traffic/flow/live${flowQuery({
      range: opts.range,
      serverId: opts.serverId,
      userId: opts.userId,
      iface: opts.iface,
      dedup: opts.dedup,
      excludeMesh: opts.excludeMesh,
      excludeOverlay: opts.excludeOverlay,
    })}`
    const url = resolveApiUrl(opts.backendUrl, path)

    let buf = ""

    void (async () => {
      try {
        const res = await fetch(url, {
          headers: withAuthHeaders({ Accept: "text/event-stream" }),
          signal: ac.signal,
          credentials: "include",
        })
        if (!res.ok || !res.body) {
          setError(`live HTTP ${res.status}`)
          return
        }
        const reader = res.body.getReader()
        const decoder = new TextDecoder()
        while (!ac.signal.aborted) {
          const { done, value } = await reader.read()
          if (done) break
          buf += decoder.decode(value, { stream: true })
          const parts = buf.split("\n\n")
          buf = parts.pop() ?? ""
          for (const raw of parts) {
            if (!raw.trim() || raw.trim().startsWith(":")) continue
            const ev = parseSseBlock(raw)
            if (ev.event === "sample" && ev.data) {
              const parsed = JSON.parse(ev.data) as unknown
              if (!isValidFlowLiveSample(parsed)) {
                setError("live sample пустой")
                continue
              }
              setSample(parsed)
              setError(parsed.degraded ? "Коллектор перегружен: упрощённая аналитика" : null)
            } else if (ev.event === "error" && ev.data) {
              const parsed = JSON.parse(ev.data) as { error?: string }
              setError(parsed.error ?? "live error")
            }
          }
        }
      } catch (e) {
        if (ac.signal.aborted) return
        setError(e instanceof Error ? e.message : "live error")
      }
    })()

    return () => ac.abort()
  }, [opts.enabled, opts.backendUrl, opts.range, opts.serverId, opts.userId, opts.iface, opts.dedup, opts.excludeMesh, opts.excludeOverlay])

  return { sample, error }
}