"use client" import { useEffect, useState } from "react" import { flowAnalyticsDtoSchema, 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(/\r?\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 { const parsed = flowAnalyticsDtoSchema.safeParse(value) return parsed.success && [...parsed.data.rxSeries, ...parsed.data.txSeries].every(value => Number.isFinite(value) && value >= 0) } 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; receivedAt: number | null } { const key = JSON.stringify([opts.enabled, opts.backendUrl, opts.range, opts.serverId, opts.userId, opts.iface, opts.dedup, opts.excludeMesh, opts.excludeOverlay]) const [result, setResult] = useState<{ key: string; sample: FlowAnalyticsDto | null; error: string | null; receivedAt: number | null } | null>(null) useEffect(() => { if (!opts.enabled) return const ac = new AbortController() let sampleTimer: ReturnType | undefined const stopTimer = () => clearTimeout(sampleTimer) const setError = (error: string) => { stopTimer() if (!ac.signal.aborted) setResult(previous => ({ key, sample: previous?.key === key ? previous.sample : null, receivedAt: previous?.key === key ? previous.receivedAt : null, error })) } 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) // Backend ticks every 1.5–2s; tolerate several missed samples. Heartbeats do not reset this timer. const awaitSample = () => { stopTimer() sampleTimer = setTimeout(() => setError("Нет обновлений live более 15 секунд"), 15_000) } awaitSample() let buf = "" void (async () => { try { const res = await fetch(url, { headers: withAuthHeaders({ Accept: "text/event-stream" }), signal: ac.signal, credentials: "include", }) if (ac.signal.aborted) return 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 (ac.signal.aborted) return if (done) { setError("Поток live завершён"); break } buf += decoder.decode(value, { stream: true }) const parts = buf.split(/\r?\n\r?\n/) buf = parts.pop() ?? "" for (const raw of parts) { if (!raw.trim()) continue const ev = parseSseBlock(raw) if (ev.event === "sample" && ev.data) { const parsed = JSON.parse(ev.data) as unknown if (!isValidFlowLiveSample(parsed)) { throw new Error("Некорректные данные live-потока") } awaitSample() setResult({ key, sample: parsed, receivedAt: Date.now(), error: parsed.degraded ? "Коллектор перегружен: упрощённая аналитика" : null }) } else if (ev.event === "error" && ev.data) { const parsed: unknown = JSON.parse(ev.data) throw new Error(parsed && typeof parsed === "object" && "error" in parsed && typeof parsed.error === "string" ? parsed.error : "Ошибка live-потока") } } } } catch (e) { if (ac.signal.aborted) return setError(e instanceof Error ? e.message : "Ошибка live-потока") ac.abort() } finally { stopTimer() } })() return () => { stopTimer(); ac.abort() } }, [opts.enabled, opts.backendUrl, opts.range, opts.serverId, opts.userId, opts.iface, opts.dedup, opts.excludeMesh, opts.excludeOverlay, key]) return result?.key === key ? { sample: result.sample, error: result.error, receivedAt: result.receivedAt } : { sample: null, error: null, receivedAt: null } }