From 23882ac1ef890dfa41e35c5904a3e171378a19ff Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Tue, 26 May 2026 10:39:17 +0800 Subject: [PATCH] =?UTF-8?q?SSE=20Patch=20=E5=A2=9E=E9=87=8F=E6=8E=A8?= =?UTF-8?q?=E9=80=81=E6=9E=B6=E6=9E=84=E5=85=A8=E9=93=BE=E8=B7=AF=E5=AE=9E?= =?UTF-8?q?=E7=8E=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - web/sse_manager.py: asyncio.Queue 连接池 + broadcast + event_stream - web/routers/sse_router.py: GET /api/events + POST /api/internal/collector-patch - weather_sources.py: 采集后 POST patch 给 web - Nginx polyweather.conf: /api/events proxy_buffering off - frontend/api/events/route.ts: Next.js SSE 代理 - frontend/hooks/use-sse-patches.ts: EventSource + useLatestPatch + 重连 - LiveTemperatureThresholdChart: mergePatchIntoHourly 增量合并 - scan-terminal-query: 接入 SSE patch 更新行数据 - 新增 ssePatchArchitecture.test.ts 架构测试 --- deploy/nginx/polyweather.conf | 13 + frontend/app/api/events/route.ts | 40 +++ .../LiveTemperatureThresholdChart.tsx | 246 ++++++++++++------ .../__tests__/refreshCadencePolicy.test.ts | 27 +- .../__tests__/ssePatchArchitecture.test.ts | 69 +++++ .../scan-terminal/scan-terminal-client.ts | 2 +- .../scan-terminal/use-scan-terminal-query.ts | 77 ++++-- frontend/hooks/use-sse-patches.ts | 146 +++++++++++ src/data_collection/weather_sources.py | 166 +++++++++++- web/app_factory.py | 2 + web/routers/sse_router.py | 39 +++ web/sse_manager.py | 92 +++++++ 12 files changed, 808 insertions(+), 111 deletions(-) create mode 100644 frontend/app/api/events/route.ts create mode 100644 frontend/components/dashboard/scan-terminal/__tests__/ssePatchArchitecture.test.ts create mode 100644 frontend/hooks/use-sse-patches.ts create mode 100644 web/routers/sse_router.py create mode 100644 web/sse_manager.py diff --git a/deploy/nginx/polyweather.conf b/deploy/nginx/polyweather.conf index a5b38b5e..c06a0871 100644 --- a/deploy/nginx/polyweather.conf +++ b/deploy/nginx/polyweather.conf @@ -29,6 +29,19 @@ server { proxy_buffers 8 16k; proxy_busy_buffers_size 32k; + location /api/events { + proxy_pass http://127.0.0.1:8000; + proxy_http_version 1.1; + proxy_buffering off; + proxy_cache off; + proxy_read_timeout 86400s; + proxy_set_header Connection ''; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + } + location / { proxy_pass http://127.0.0.1:8000; proxy_http_version 1.1; diff --git a/frontend/app/api/events/route.ts b/frontend/app/api/events/route.ts new file mode 100644 index 00000000..637b5bce --- /dev/null +++ b/frontend/app/api/events/route.ts @@ -0,0 +1,40 @@ +import { NextRequest, NextResponse } from "next/server"; + +const API_BASE = process.env.POLYWEATHER_API_BASE_URL; + +export const dynamic = "force-dynamic"; +export const runtime = "nodejs"; + +export async function GET(req: NextRequest) { + if (!API_BASE) { + return NextResponse.json( + { error: "POLYWEATHER_API_BASE_URL is not configured" }, + { status: 500 }, + ); + } + + const upstream = await fetch(`${API_BASE.replace(/\/+$/, "")}/api/events`, { + cache: "no-store", + headers: { + Accept: "text/event-stream", + Cookie: req.headers.get("cookie") || "", + }, + }); + + if (!upstream.ok || !upstream.body) { + return NextResponse.json( + { error: `SSE upstream failed with HTTP ${upstream.status}` }, + { status: upstream.status || 502 }, + ); + } + + return new Response(upstream.body, { + status: 200, + headers: { + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache, no-transform", + Connection: "keep-alive", + "X-Accel-Buffering": "no", + }, + }); +} diff --git a/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx b/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx index b78d8501..fce17f8d 100644 --- a/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx +++ b/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx @@ -1,7 +1,7 @@ "use client"; import clsx from "clsx"; -import { useEffect, useMemo, useState } from "react"; +import { useEffect, useMemo, useRef, useState } from "react"; import { CartesianGrid, Line, @@ -22,6 +22,7 @@ import type { } from "@/lib/dashboard-types"; import { buildDebBaselinePath } from "@/lib/temperature-chart-paths"; import { DASHBOARD_REFRESH_POLICY_MS } from "@/lib/refresh-policy"; +import { useLatestPatch, type CityPatch } from "@/hooks/use-sse-patches"; import { Panel } from "@/components/dashboard/scan-terminal/Panel"; import { rowName, temp } from "@/components/dashboard/scan-terminal/utils"; @@ -484,19 +485,51 @@ function seedHourlyForecastFromRow(row: ScanOpportunityRow | null): HourlyForeca }; } -async function fetchHourlyForecastForCity(city: string): Promise { - const cached = _hourlyCache.get(city); - if (cached && Date.now() - cached.ts < HOURLY_CACHE_TTL_MS) { - return cached.data; +type HourlyForecastFetchOptions = { + ignoreCache?: boolean; +}; + +function parseHourlyForecastFromCityDetail(json: CityDetail | null): HourlyForecast { + const hourlySource = (json as any)?.hourly ?? (json as any)?.timeseries?.hourly; + if (!json || !hourlySource) return null; + return { + forecastTodayHigh: json.forecast?.today_high ?? null, + debPrediction: json.deb?.prediction ?? (json as any)?.overview?.deb_prediction ?? null, + localTime: json.local_time || null, + times: hourlySource.times || [], + temps: hourlySource.temps || [], + modelCurves: (json.models_hourly ?? (json as any)?.timeseries?.models_hourly)?.curves || undefined, + runwayPlateHistory: (json as any)?.runway_plate_history || (json.amos as any)?.runway_plate_history || undefined, + amos: json.amos || null, + airportCurrent: json.airport_current || null, + airportPrimary: json.airport_primary || null, + forecastDaily: json.forecast?.daily || [], + multiModelDaily: json.multi_model_daily || {}, + settlementTodayObs: (json as any).timeseries?.settlement_today_obs || (json as any)?.settlement_today_obs || undefined, + metarTodayObs: (json as any).timeseries?.metar_today_obs || (json as any)?.metar_today_obs || undefined, + airportPrimaryTodayObs: (json as any)?.official?.airport_primary_today_obs || (json as any)?.airport_primary_today_obs || undefined, + }; +} + +async function fetchHourlyForecastForCity( + city: string, + options: HourlyForecastFetchOptions = {}, +): Promise { + if (!options.ignoreCache) { + const cached = _hourlyCache.get(city); + if (cached && Date.now() - cached.ts < HOURLY_CACHE_TTL_MS) { + return cached.data; + } + + const sessionCached = readSessionCache(city); + if (sessionCached) { + _hourlyCache.set(city, sessionCached); + return sessionCached.data; + } } - const sessionCached = readSessionCache(city); - if (sessionCached) { - _hourlyCache.set(city, sessionCached); - return sessionCached.data; - } - - const pending = _hourlyRequestCache.get(city); + const requestKey = options.ignoreCache ? `${city}:live` : city; + const pending = _hourlyRequestCache.get(requestKey); if (pending) return pending; const request = fetch(`/api/city/${encodeURIComponent(city)}/detail?depth=full&force_refresh=false`, { @@ -507,37 +540,96 @@ async function fetchHourlyForecastForCity(city: string): Promise return res.json() as Promise; }) .then((json) => { - const hourlySource = (json as any)?.hourly ?? (json as any)?.timeseries?.hourly; - if (!json || !hourlySource) return null; - const data: HourlyForecast = { - forecastTodayHigh: json.forecast?.today_high ?? null, - debPrediction: json.deb?.prediction ?? (json as any)?.overview?.deb_prediction ?? null, - localTime: json.local_time || null, - times: hourlySource.times || [], - temps: hourlySource.temps || [], - modelCurves: (json.models_hourly ?? (json as any)?.timeseries?.models_hourly)?.curves || undefined, - runwayPlateHistory: (json as any)?.runway_plate_history || (json.amos as any)?.runway_plate_history || undefined, - amos: json.amos || null, - airportCurrent: json.airport_current || null, - airportPrimary: json.airport_primary || null, - forecastDaily: json.forecast?.daily || [], - multiModelDaily: json.multi_model_daily || {}, - settlementTodayObs: (json as any).timeseries?.settlement_today_obs || (json as any)?.settlement_today_obs || undefined, - metarTodayObs: (json as any).timeseries?.metar_today_obs || (json as any)?.metar_today_obs || undefined, - airportPrimaryTodayObs: (json as any)?.official?.airport_primary_today_obs || (json as any)?.airport_primary_today_obs || undefined, - }; + const data = parseHourlyForecastFromCityDetail(json); + if (!data) return null; _hourlyCache.set(city, { ts: Date.now(), data }); writeSessionCache(city, data); return data; }) .finally(() => { - _hourlyRequestCache.delete(city); + _hourlyRequestCache.delete(requestKey); }); - _hourlyRequestCache.set(city, request); + _hourlyRequestCache.set(requestKey, request); return request; } +function shouldPollLiveChart({ + city, + compact, + isActive, + isMaximized, +}: { + city: string; + compact: boolean; + isActive: boolean; + isMaximized: boolean; +}) { + return Boolean(city) && (compact || isActive || isMaximized); +} + +function mergePatchIntoHourly( + prev: HourlyForecast, + patch: CityPatch, +): HourlyForecast { + const changes = patch.changes || {}; + const tempValue = validNumber(changes.temp); + const obsTime = typeof changes.obs_time === "string" ? changes.obs_time : null; + const source = typeof changes.source === "string" ? changes.source : ""; + const explicitHourlyPatch = changes.hourly && typeof changes.hourly === "object" + ? changes.hourly as Partial> + : {}; + + const next: NonNullable = { + ...(prev || { + forecastTodayHigh: null, + debPrediction: null, + localTime: null, + times: [], + temps: [], + forecastDaily: [], + multiModelDaily: {}, + }), + ...explicitHourlyPatch, + }; + + if (changes.amos && typeof changes.amos === "object") { + next.amos = changes.amos as AmosData; + } + + if (tempValue !== null) { + next.airportCurrent = { + ...(next.airportCurrent || {}), + obs_time: next.airportCurrent?.obs_time ?? null, + temp: tempValue, + max_so_far: Math.max( + tempValue, + validNumber(next.airportCurrent?.max_so_far) ?? tempValue, + ), + }; + next.airportPrimary = { + ...(next.airportPrimary || {}), + obs_time: next.airportPrimary?.obs_time ?? null, + temp: tempValue, + max_so_far: Math.max( + tempValue, + validNumber(next.airportPrimary?.max_so_far) ?? tempValue, + ), + source_label: next.airportPrimary?.source_label || source || undefined, + }; + } + + if (tempValue !== null && obsTime) { + const obsPoint: RawObsPoint = [obsTime, tempValue]; + const currentObs = Array.isArray(next.airportPrimaryTodayObs) + ? next.airportPrimaryTodayObs + : []; + next.airportPrimaryTodayObs = [...currentObs, obsPoint].slice(-MAX_OBS_POINTS); + } + + return next; +} + function parseRunwayHistoryValue(point: Record) { return validNumber(point.max_temp_c) ?? validNumber(point.temp_c) ?? validNumber(point.temp) ?? validNumber(point.value); } @@ -1155,12 +1247,17 @@ export function LiveTemperatureThresholdChart({ }) { const [hourly, setHourly] = useState(null); const city = String(row?.city || "").toLowerCase().trim(); + const latestPatch = useLatestPatch(city); const [timeframe, setTimeframe] = useState<"1D" | "3D">("1D"); const [userToggledKeys, setUserToggledKeys] = useState>({}); const [liveTemp, setLiveTemp] = useState(null); + const lastPatchAtRef = useRef(Date.now()); + const lastAppliedPatchRevisionRef = useRef(0); useEffect(() => { setUserToggledKeys({}); + lastPatchAtRef.current = Date.now(); + lastAppliedPatchRevisionRef.current = 0; }, [city, timeframe]); useEffect(() => { @@ -1203,53 +1300,53 @@ export function LiveTemperatureThresholdChart({ }; }, [city, row, isActive, slotIndex]); - // ── 60s live poll: summary (number) + detail amos/runway (curves) ── useEffect(() => { - if (!city) return; - const poll = () => { - // Lightweight: current temp number + if (!latestPatch || latestPatch.revision <= lastAppliedPatchRevisionRef.current) return; + lastAppliedPatchRevisionRef.current = latestPatch.revision; + lastPatchAtRef.current = Date.now(); + const tempValue = validNumber(latestPatch.changes.temp); + if (tempValue !== null) setLiveTemp(tempValue); + setHourly((prev) => mergePatchIntoHourly(prev ?? seedHourlyForecastFromRow(row), latestPatch)); + }, [latestPatch, row]); + + // ── SSE fallback: only full-fetch if a visible chart has seen no patch for 2 minutes ── + useEffect(() => { + if (!shouldPollLiveChart({ city, compact, isActive, isMaximized })) return; + let cancelled = false; + + const refreshFullDetail = () => { + lastPatchAtRef.current = Date.now(); + + fetchHourlyForecastForCity(city, { ignoreCache: true }) + .then((data) => { + if (cancelled || !data) return; + setHourly(data); + }) + .catch(() => {}); + }; + + const checkFallback = () => { + if (typeof document !== "undefined" && document.visibilityState === "hidden") return; + if (Date.now() - lastPatchAtRef.current < 2 * 60_000) return; + fetch(`/api/city/${encodeURIComponent(city)}/summary`) .then((res) => (res.ok ? res.json() : null)) .then((payload) => { - if (!payload) return; + if (cancelled || !payload) return; const temp = validNumber(payload?.current?.temp); if (temp !== null) setLiveTemp(temp); }) .catch(() => {}); - // Refresh runway curves and station observations - fetch(`/api/city/${encodeURIComponent(city)}/detail?depth=full&force_refresh=false`) - .then((res) => (res.ok ? res.json() : null)) - .then((json) => { - if (!json) return; - setHourly((prev) => { - if (!prev) return prev; - return { - ...prev, - amos: json.amos ?? prev.amos, - airportPrimary: json.airport_primary ?? prev.airportPrimary, - airportCurrent: json.airport_current ?? prev.airportCurrent, - airportPrimaryTodayObs: (json as any)?.official?.airport_primary_today_obs - || json.airport_primary_today_obs - || prev.airportPrimaryTodayObs, - runwayPlateHistory: (json as any)?.runway_plate_history - || (json.amos as any)?.runway_plate_history - || prev.runwayPlateHistory, - settlementTodayObs: (json as any).timeseries?.settlement_today_obs - || (json as any)?.settlement_today_obs - || prev.settlementTodayObs, - metarTodayObs: (json as any).timeseries?.metar_today_obs - || (json as any)?.metar_today_obs - || prev.metarTodayObs, - }; - }); - }) - .catch(() => {}); + refreshFullDetail(); }; - poll(); - const id = setInterval(poll, 60_000); - return () => clearInterval(id); - }, [city]); + + const id = setInterval(checkFallback, 60_000); + return () => { + cancelled = true; + clearInterval(id); + }; + }, [city, compact, isActive, isMaximized]); const { data, series } = useMemo(() => { if (timeframe === "3D") { @@ -1300,8 +1397,11 @@ export function LiveTemperatureThresholdChart({ const isTokyo = normalizedKey === 'tokyo'; const isSingapore = normalizedKey === 'singapore'; const isParis = normalizedKey === 'paris'; + const stationSourceCode = hourly?.airportPrimary?.source_code || (row as any)?.station_source_code || ''; + const hasRealStationNetwork = /^(mgm|jma_amedas|fmi|knmi|cowin_obs|ims|ncm|aeroweb|madis_hfmetar|singapore_mss)$/.test(stationSourceCode); const isWeatherStation = !runwaySensorCities.has(normalizedKey) - && !isHKO && !isShenzhen && !isTokyo && !isSingapore && !isParis; + && !isHKO && !isShenzhen && !isTokyo && !isSingapore && !isParis + && hasRealStationNetwork; const runwayHeaderLabel = isShenzhen ? '天文台实测 (10分钟)' : isHKO ? '参考站点 (1分钟)' @@ -1311,8 +1411,7 @@ export function LiveTemperatureThresholdChart({ : isWeatherStation ? '气象站实测' : '跑道实测 (1分钟)'; - const metarHeaderLabel = isShenzhen ? '天文台实测 (10分钟)' - : isHKO ? '天文台实测 (10分钟)' + const metarHeaderLabel = (isShenzhen || isHKO) ? '天文台实测 (10分钟)' : 'METAR 结算 (30分钟)'; const runwayHighLabel = isShenzhen ? '天文台实测' @@ -1800,3 +1899,4 @@ export function __buildTemperatureChartDataForTest( export const __isTemperatureSeriesVisibleByDefaultForTest = isTemperatureSeriesVisibleByDefault; export const __getVisibleTemperatureSeriesForTest = getVisibleTemperatureSeries; export const __getObservationDisplayMetricsForTest = getObservationDisplayMetrics; +export const __shouldPollLiveChartForTest = shouldPollLiveChart; diff --git a/frontend/components/dashboard/scan-terminal/__tests__/refreshCadencePolicy.test.ts b/frontend/components/dashboard/scan-terminal/__tests__/refreshCadencePolicy.test.ts index c7b7e367..713fd79c 100644 --- a/frontend/components/dashboard/scan-terminal/__tests__/refreshCadencePolicy.test.ts +++ b/frontend/components/dashboard/scan-terminal/__tests__/refreshCadencePolicy.test.ts @@ -5,6 +5,7 @@ import { DASHBOARD_REFRESH_POLICY_SEC, } from "@/lib/refresh-policy"; import { scanTerminalQueryPolicy } from "@/components/dashboard/scan-terminal/scan-terminal-client"; +import { __shouldPollLiveChartForTest } from "@/components/dashboard/scan-terminal/LiveTemperatureThresholdChart"; function assert(condition: unknown, message: string) { if (!condition) throw new Error(message); @@ -16,7 +17,7 @@ export function runTests() { assert(DASHBOARD_REFRESH_POLICY_MS.marketOverview === 10 * 60_000, "market overview should refresh every 10 minutes"); assert(DASHBOARD_REFRESH_POLICY_MS.model === 30 * 60_000, "DEB and multi-model data should refresh every 30 minutes"); assert(DASHBOARD_REFRESH_POLICY_SEC.metar === 5 * 60, "METAR polling should be 5 minutes"); - assert(scanTerminalQueryPolicy.autoRefreshMs === DASHBOARD_REFRESH_POLICY_MS.scanRows, "scan terminal auto refresh should use the shared row cadence"); + assert(scanTerminalQueryPolicy.autoRefreshMs === null, "scan terminal auto refresh should be disabled after SSE patch migration"); const projectRoot = process.cwd(); const querySource = fs.readFileSync( @@ -29,8 +30,9 @@ export function runTests() { ); assert( - querySource.includes("DASHBOARD_REFRESH_POLICY_MS.scanRows"), - "scan list local cache should use the shared 5-minute row cadence", + querySource.includes("useSsePatchVersion") && + !querySource.includes("window.setInterval"), + "scan list should subscribe to SSE patch state instead of running a 5-minute interval", ); assert( chartSource.includes("DASHBOARD_REFRESH_POLICY_MS.metar") && @@ -38,8 +40,23 @@ export function runTests() { "selected city detail chart cache should align with 5-minute scan/metar cadence", ); assert( - chartSource.includes("setInterval(poll, 60_000)"), - "selected city chart should poll live temperature and runway curves every 60 seconds", + chartSource.includes("useLatestPatch") && + chartSource.includes("latestPatch") && + chartSource.includes("2 * 60_000"), + "selected city chart should consume SSE patches and use a 2-minute no-patch fallback", + ); + assert( + chartSource.includes("fetchHourlyForecastForCity(city, { ignoreCache: true })") && + chartSource.includes("setHourly(data)"), + "visible chart fallback must refresh the full city detail payload when SSE patches stop", + ); + assert( + __shouldPollLiveChartForTest({ city: "shanghai", compact: true, isActive: false, isMaximized: false }) === true, + "compact grid slots are visible charts and should run the no-patch fallback guard", + ); + assert( + __shouldPollLiveChartForTest({ city: "shanghai", compact: false, isActive: false, isMaximized: false }) === false, + "inactive non-compact charts should not run live polling", ); assert( chartSource.includes("_hourlyRequestCache") && diff --git a/frontend/components/dashboard/scan-terminal/__tests__/ssePatchArchitecture.test.ts b/frontend/components/dashboard/scan-terminal/__tests__/ssePatchArchitecture.test.ts new file mode 100644 index 00000000..5b5140c5 --- /dev/null +++ b/frontend/components/dashboard/scan-terminal/__tests__/ssePatchArchitecture.test.ts @@ -0,0 +1,69 @@ +import fs from "node:fs"; +import path from "node:path"; + +function assert(condition: unknown, message: string) { + if (!condition) throw new Error(message); +} + +function readRepoFile(...parts: string[]) { + return fs.readFileSync(path.join(process.cwd(), "..", ...parts), "utf8"); +} + +function readFrontendFile(...parts: string[]) { + return fs.readFileSync(path.join(process.cwd(), ...parts), "utf8"); +} + +export function runTests() { + const repoRoot = path.join(process.cwd(), ".."); + + const sseManagerPath = path.join(repoRoot, "web", "sse_manager.py"); + assert(fs.existsSync(sseManagerPath), "FastAPI backend must define web/sse_manager.py"); + const sseManager = fs.readFileSync(sseManagerPath, "utf8"); + assert(sseManager.includes("asyncio.Queue"), "SSE manager must keep asyncio.Queue connections"); + assert(sseManager.includes("broadcast("), "SSE manager must expose broadcast(city, changes)"); + assert(sseManager.includes("event_stream("), "SSE manager must expose an async event_stream(user_id)"); + assert(sseManager.includes("revision"), "SSE patches must carry monotonic revision numbers"); + assert(sseManager.includes("30"), "SSE stream must include a 30-second heartbeat"); + assert(sseManager.includes("data: "), "SSE stream must emit data: JSON frames"); + + const sseRouterPath = path.join(repoRoot, "web", "routers", "sse_router.py"); + assert(fs.existsSync(sseRouterPath), "FastAPI backend must define web/routers/sse_router.py"); + const sseRouter = fs.readFileSync(sseRouterPath, "utf8"); + assert(sseRouter.includes('"/api/events"'), "SSE router must expose GET /api/events"); + assert(sseRouter.includes('"/api/internal/collector-patch"'), "SSE router must expose collector patch ingest endpoint"); + assert(sseRouter.includes("StreamingResponse"), "SSE route must return StreamingResponse"); + assert(sseRouter.includes('"text/event-stream"'), "SSE route must use text/event-stream media type"); + + const appFactory = readRepoFile("web", "app_factory.py"); + assert(appFactory.includes("sse_router"), "FastAPI app factory must register the SSE router"); + + const nginx = readRepoFile("deploy", "nginx", "polyweather.conf"); + assert(nginx.includes("location /api/events"), "Nginx deploy config must route /api/events separately"); + assert(nginx.includes("proxy_buffering off"), "Nginx /api/events must disable proxy buffering for SSE"); + assert(nginx.includes("proxy_read_timeout 86400s"), "Nginx /api/events must keep SSE connections open"); + + const weatherSources = readRepoFile("src", "data_collection", "weather_sources.py"); + assert(weatherSources.includes("_emit_temperature_patch_if_changed"), "collector must centralize temperature patch emission"); + assert(weatherSources.includes("requests.post"), "collector must POST patches to the internal endpoint"); + assert(weatherSources.includes("/api/internal/collector-patch"), "collector must POST to /api/internal/collector-patch"); + assert(weatherSources.includes("threading.Thread"), "collector patch POST must run in a separate thread"); + + const hookPath = path.join(process.cwd(), "hooks", "use-sse-patches.ts"); + assert(fs.existsSync(hookPath), "frontend must define hooks/use-sse-patches.ts"); + const hook = fs.readFileSync(hookPath, "utf8"); + assert(hook.includes("new EventSource"), "frontend patch hook must connect with EventSource"); + assert(hook.includes("/api/events"), "frontend patch hook must subscribe to /api/events"); + assert(hook.includes("Map<"), "frontend patch hook must keep latest patches in a Map"); + assert(hook.includes("useLatestPatch"), "frontend patch hook must export useLatestPatch(city)"); + assert(hook.includes("revision"), "frontend patch hook must track revisions and skip stale patches"); + assert(hook.includes("setTimeout"), "frontend patch hook must implement explicit reconnect backoff"); + + const chart = readFrontendFile("components", "dashboard", "scan-terminal", "LiveTemperatureThresholdChart.tsx"); + assert(chart.includes("useLatestPatch"), "temperature chart must consume useLatestPatch(city)"); + assert(chart.includes("latestPatch"), "temperature chart must react to incoming SSE patches"); + assert(chart.includes("2 * 60_000"), "temperature chart must wait two minutes without patches before full-fetch fallback"); + assert( + !chart.includes("setInterval(poll, 60_000)"), + "temperature chart must not use unconditional 60-second full-detail polling after SSE patch migration", + ); +} diff --git a/frontend/components/dashboard/scan-terminal/scan-terminal-client.ts b/frontend/components/dashboard/scan-terminal/scan-terminal-client.ts index a6afce2e..fae501cb 100644 --- a/frontend/components/dashboard/scan-terminal/scan-terminal-client.ts +++ b/frontend/components/dashboard/scan-terminal/scan-terminal-client.ts @@ -18,7 +18,7 @@ export type RemoteData = | { status: "error"; error: string; previous?: T }; export const scanTerminalQueryPolicy = { - autoRefreshMs: DASHBOARD_REFRESH_POLICY_MS.scanRows, + autoRefreshMs: null, manualForceRefreshCooldownMs: 2 * 60_000, } as const; diff --git a/frontend/components/dashboard/scan-terminal/use-scan-terminal-query.ts b/frontend/components/dashboard/scan-terminal/use-scan-terminal-query.ts index 2d581714..c83e345c 100644 --- a/frontend/components/dashboard/scan-terminal/use-scan-terminal-query.ts +++ b/frontend/components/dashboard/scan-terminal/use-scan-terminal-query.ts @@ -1,15 +1,18 @@ "use client"; -import { useCallback, useEffect, useRef, useState } from "react"; +import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { - scanTerminalQueryPolicy, scanTerminalClient, - shouldRunAutoTerminalRefresh, shouldSkipManualTerminalRefresh, } from "@/components/dashboard/scan-terminal/scan-terminal-client"; import { useRemoteDataQuery } from "@/components/dashboard/scan-terminal/use-remote-data-query"; import { REGIONS } from "@/components/dashboard/scan-terminal/continent-grouping"; import { DASHBOARD_REFRESH_POLICY_MS } from "@/lib/refresh-policy"; +import { + getLatestPatchesSnapshot, + useSsePatchVersion, + type CityPatch, +} from "@/hooks/use-sse-patches"; import type { ScanTerminalResponse } from "@/lib/dashboard-types"; const SCAN_CACHE_PREFIX = "polyweather_scan_v2"; @@ -35,6 +38,44 @@ function writeScanCache(data: ScanTerminalResponse, tradingRegion: string) { try { localStorage.setItem(scanCacheKey(tradingRegion), JSON.stringify({ ts: Date.now(), data })); } catch { /* ignore */ } } +function normalizeCityKey(city: string | null | undefined) { + return String(city || "").trim().toLowerCase(); +} + +function applyPatchToScanRow(row: any, patch: CityPatch) { + const temp = typeof patch.changes.temp === "number" && Number.isFinite(patch.changes.temp) + ? patch.changes.temp + : null; + if (temp === null) return row; + return { + ...row, + current_temp: temp, + current_max_so_far: Math.max( + temp, + typeof row.current_max_so_far === "number" && Number.isFinite(row.current_max_so_far) + ? row.current_max_so_far + : temp, + ), + local_time: typeof patch.changes.obs_time === "string" ? patch.changes.obs_time : row.local_time, + sse_revision: patch.revision, + }; +} + +function applyTerminalPatches( + data: ScanTerminalResponse | null | undefined, + patches: Map, +): ScanTerminalResponse | null | undefined { + if (!data?.rows?.length || !patches.size) return data; + let changed = false; + const rows = data.rows.map((row: any) => { + const patch = patches.get(normalizeCityKey(row.city)); + if (!patch) return row; + changed = true; + return applyPatchToScanRow(row, patch); + }); + return changed ? { ...data, rows } : data; +} + export function useScanTerminalQuery({ isPro, proAccessLoading, @@ -49,7 +90,6 @@ export function useScanTerminalQuery({ const { data: terminalData, error: scanError, - isLoading, loading: scanLoading, remote: scanRemote, reset, @@ -57,6 +97,7 @@ export function useScanTerminalQuery({ } = useRemoteDataQuery(); const lastForcedScanRefreshAtRef = useRef(0); + const patchVersion = useSsePatchVersion(); const [cachedRows, setCachedRows] = useState(() => { if (typeof window !== "undefined") return readScanCache(tradingRegion || ""); return null; @@ -101,7 +142,10 @@ export function useScanTerminalQuery({ void fetchScanTerminal({ forceRefresh: false, showLoading: true }); }, [fetchScanTerminal, isPro, proAccessLoading, reset, timezoneOffsetSeconds, tradingRegion]); - const effectiveData = terminalData || cachedRows; + const effectiveData = useMemo( + () => applyTerminalPatches(terminalData || cachedRows, getLatestPatchesSnapshot()), + [terminalData, cachedRows, patchVersion], + ); const refreshScanTerminalManually = useCallback(() => { if ( @@ -115,29 +159,6 @@ export function useScanTerminalQuery({ void fetchScanTerminal({ forceRefresh: true, showLoading: true }); }, [fetchScanTerminal, terminalData]); - useEffect(() => { - if (proAccessLoading || !isPro) return; - if ( - typeof window === "undefined" || - typeof window.setInterval !== "function" || - typeof window.clearInterval !== "function" - ) { - return; - } - const intervalId = window.setInterval(() => { - if ( - !shouldRunAutoTerminalRefresh({ - documentHidden: document.hidden, - isLoading: isLoading(), - }) - ) { - return; - } - void fetchScanTerminal({ forceRefresh: false, showLoading: false }); - }, scanTerminalQueryPolicy.autoRefreshMs); - return () => window.clearInterval(intervalId); - }, [fetchScanTerminal, isLoading, isPro, proAccessLoading]); - // Preload adjacent regions in idle time for instant tab switches useEffect(() => { if (typeof window === "undefined" || !tradingRegion || !isPro) return; diff --git a/frontend/hooks/use-sse-patches.ts b/frontend/hooks/use-sse-patches.ts new file mode 100644 index 00000000..a3852071 --- /dev/null +++ b/frontend/hooks/use-sse-patches.ts @@ -0,0 +1,146 @@ +"use client"; + +import { useEffect, useSyncExternalStore } from "react"; +import { resolveBackendApiUrl } from "@/lib/backend-api"; + +export type CityPatch = { + type?: string; + city: string; + changes: Record; + revision: number; + ts?: number; +}; + +const latestPatches = new Map(); +const latestRevisions = new Map(); +const cityListeners = new Map void>>(); +const globalListeners = new Set<() => void>(); + +let eventSource: EventSource | null = null; +let reconnectTimer: ReturnType | null = null; +let reconnectAttempt = 0; +let patchVersion = 0; + +function normalizeCityKey(city: string | null | undefined) { + return String(city || "").trim().toLowerCase(); +} + +function notify(city: string) { + patchVersion += 1; + cityListeners.get(city)?.forEach((listener) => listener()); + globalListeners.forEach((listener) => listener()); +} + +function scheduleReconnect() { + if (reconnectTimer || typeof window === "undefined") return; + const delayMs = Math.min(30_000, 1_000 * Math.max(1, 2 ** reconnectAttempt)); + reconnectAttempt += 1; + reconnectTimer = setTimeout(() => { + reconnectTimer = null; + connectSsePatches(); + }, delayMs); +} + +function closeEventSource() { + if (!eventSource) return; + eventSource.close(); + eventSource = null; +} + +function connectSsePatches() { + if (typeof window === "undefined" || eventSource) return; + closeEventSource(); + eventSource = new EventSource(resolveBackendApiUrl("/api/events")); + + eventSource.onopen = () => { + reconnectAttempt = 0; + }; + + eventSource.onmessage = (event) => { + try { + applySsePatch(JSON.parse(event.data)); + } catch { + // Ignore malformed frames; the stream stays alive. + } + }; + + eventSource.onerror = () => { + closeEventSource(); + scheduleReconnect(); + }; +} + +export function ensureSsePatchConnection() { + connectSsePatches(); +} + +export function applySsePatch(payload: unknown) { + if (!payload || typeof payload !== "object") return false; + const patch = payload as Partial; + if (patch.type && patch.type !== "city_patch") return false; + const city = normalizeCityKey(patch.city); + const changes = patch.changes; + const revision = Number(patch.revision); + if (!city || !changes || typeof changes !== "object" || !Number.isFinite(revision)) { + return false; + } + + const previousRevision = latestRevisions.get(city) ?? 0; + if (revision <= previousRevision) return false; + + const normalizedPatch: CityPatch = { + type: "city_patch", + city, + changes: changes as Record, + revision, + ts: typeof patch.ts === "number" ? patch.ts : Date.now(), + }; + latestRevisions.set(city, revision); + latestPatches.set(city, normalizedPatch); + notify(city); + return true; +} + +export function getLatestPatchesSnapshot() { + return latestPatches; +} + +export function useSsePatchVersion() { + useEffect(() => { + ensureSsePatchConnection(); + }, []); + + return useSyncExternalStore( + (listener) => { + globalListeners.add(listener); + return () => globalListeners.delete(listener); + }, + () => patchVersion, + () => 0, + ); +} + +export function useLatestPatch(city: string | null | undefined) { + const cityKey = normalizeCityKey(city); + + useEffect(() => { + ensureSsePatchConnection(); + }, []); + + return useSyncExternalStore( + (listener) => { + if (!cityKey) return () => {}; + const listeners = cityListeners.get(cityKey) ?? new Set<() => void>(); + listeners.add(listener); + cityListeners.set(cityKey, listeners); + return () => { + listeners.delete(listener); + if (!listeners.size) cityListeners.delete(cityKey); + }; + }, + () => (cityKey ? latestPatches.get(cityKey) ?? null : null), + () => null, + ); +} + +export const __applySsePatchForTest = applySsePatch; diff --git a/src/data_collection/weather_sources.py b/src/data_collection/weather_sources.py index c013408e..e1144db5 100644 --- a/src/data_collection/weather_sources.py +++ b/src/data_collection/weather_sources.py @@ -1,6 +1,7 @@ import os import httpx import re +import requests import threading import time from typing import Any, Optional, Dict, List @@ -268,6 +269,14 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour ) self._disk_cache_lock = threading.Lock() self._disk_cache_last_mtime: float = 0.0 + self.collector_patch_endpoint = str( + os.getenv( + "POLYWEATHER_COLLECTOR_PATCH_ENDPOINT", + "http://127.0.0.1:8000/api/internal/collector-patch", + ) + ).strip() + self._last_emitted_temperature_patch: Dict[str, tuple[float, str]] = {} + self._temperature_patch_lock = threading.Lock() self._load_open_meteo_disk_cache() logger.info( f"Open-Meteo 磁盘缓存路径: {self._disk_cache_path} (max_age={self._disk_cache_max_age_sec}s)" @@ -338,6 +347,62 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour with lock: for key in stale: cache.pop(key, None) + + def _emit_temperature_patch_if_changed( + self, + city: str, + temp: Any, + obs_time: Any = None, + *, + source: str = "", + extra: Optional[Dict[str, Any]] = None, + ) -> None: + if temp is None or not self.collector_patch_endpoint: + return + try: + temp_value = round(float(temp), 2) + except (TypeError, ValueError): + return + + obs_time_value = str(obs_time or datetime.now(timezone.utc).isoformat()) + source_value = str(source or "weather").strip().lower() or "weather" + city_value = str(city or "").strip().lower() + if not city_value: + return + + dedupe_key = f"{city_value}:{source_value}" + signature = (temp_value, obs_time_value) + with self._temperature_patch_lock: + if self._last_emitted_temperature_patch.get(dedupe_key) == signature: + return + self._last_emitted_temperature_patch[dedupe_key] = signature + + changes: Dict[str, Any] = { + "temp": temp_value, + "obs_time": obs_time_value, + "source": source_value, + } + if extra: + changes.update(extra) + + payload = {"city": city_value, "changes": changes} + + def post_patch() -> None: + try: + requests.post( + self.collector_patch_endpoint, + json=payload, + timeout=2, + ) + except Exception as exc: + logger.debug( + "collector patch POST failed city={} source={}: {}", + city_value, + source_value, + exc, + ) + + threading.Thread(target=post_patch, daemon=True).start() # MADIS cache is a single list, not a keyed dict — expire on age with self._madis_cache_lock: if self._madis_cache is not None and now - self._madis_cache_ts > self.madis_cache_ttl_sec * 2: @@ -878,6 +943,13 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour settlement_current = self.fetch_settlement_current(city_lower) if settlement_current: results["settlement_current"] = settlement_current + current = settlement_current.get("current") or {} + self._emit_temperature_patch_if_changed( + city_lower, + current.get("temp"), + settlement_current.get("observation_time") or current.get("obs_time"), + source=str(settlement_current.get("source") or "settlement"), + ) city_meta = self.CITY_REGISTRY.get(str(city_lower or "").strip().lower()) or {} settlement_source = str(city_meta.get("settlement_source") or "").strip().lower() @@ -905,9 +977,15 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour if not mgm_data: return results["mgm"] = mgm_data + mgm_current = mgm_data.get("current") or {} + self._emit_temperature_patch_if_changed( + city_lower, + mgm_current.get("temp"), + mgm_data.get("obs_time"), + source="mgm", + ) # Log airport obs for high-freq monitoring (Ankara 17128) try: - mgm_current = mgm_data.get("current") or {} DBManager().append_airport_obs( icao=str(istno), city=city_lower, @@ -966,8 +1044,14 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour ims_data = self.fetch_from_ims(self.IMS_LOD_AIRPORT_STATION) if ims_data: results["ims"] = ims_data + ims_current = ims_data.get("current") or {} + self._emit_temperature_patch_if_changed( + city_lower, + ims_current.get("temp"), + ims_data.get("obs_time"), + source="ims", + ) try: - ims_current = ims_data.get("current") or {} DBManager().append_airport_obs( icao="LLBG", city=city_lower, @@ -984,8 +1068,14 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour ncm_data = self.fetch_from_ncm() if ncm_data: results["ncm"] = ncm_data + ncm_current = ncm_data.get("current") or {} + self._emit_temperature_patch_if_changed( + city_lower, + ncm_current.get("temp"), + ncm_data.get("obs_time"), + source="ncm", + ) try: - ncm_current = ncm_data.get("current") or {} DBManager().append_airport_obs( icao="OEJN", city=city_lower, @@ -1002,8 +1092,14 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour aw_data = self.fetch_from_aeroweb() if aw_data: results["aeroweb"] = aw_data + aw_current = aw_data.get("current") or {} + self._emit_temperature_patch_if_changed( + city_lower, + aw_current.get("temp"), + aw_data.get("obs_time"), + source="aeroweb", + ) try: - aw_current = aw_data.get("current") or {} DBManager().append_airport_obs( icao="LFPB", city=city_lower, @@ -1031,6 +1127,12 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour return results["jma_official_nearby"] = official_rows results["jma_current"] = official_rows[0] # primary station for airport_primary + self._emit_temperature_patch_if_changed( + city_lower, + official_rows[0].get("temp"), + official_rows[0].get("obs_time"), + source="jma_amedas", + ) if "mgm_nearby" not in results: results["mgm_nearby"] = official_rows results["nearby_source"] = "jma" @@ -1058,6 +1160,12 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour return results["knmi_official_nearby"] = rows results["knmi_current"] = rows[0] # primary station for airport_primary + self._emit_temperature_patch_if_changed( + city_lower, + rows[0].get("temp"), + rows[0].get("obs_time"), + source="knmi", + ) if "mgm_nearby" not in results: results["mgm_nearby"] = rows results["nearby_source"] = "knmi" @@ -1084,6 +1192,12 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour return results["fmi_official_nearby"] = rows results["fmi_current"] = rows[0] # primary station for airport_primary + self._emit_temperature_patch_if_changed( + city_lower, + rows[0].get("temp"), + rows[0].get("obs_time"), + source="fmi", + ) if "mgm_nearby" not in results: results["mgm_nearby"] = rows results["nearby_source"] = "fmi" @@ -1110,6 +1224,12 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour if not rows: return results["hko_obs_nearby"] = rows + self._emit_temperature_patch_if_changed( + city_lower, + rows[0].get("temp"), + rows[0].get("obs_time"), + source="hko_obs", + ) if "mgm_nearby" not in results: results["mgm_nearby"] = rows results["nearby_source"] = "hko_obs" @@ -1136,6 +1256,12 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour return results["cowin_obs_nearby"] = rows results["cowin_current"] = rows[0] + self._emit_temperature_patch_if_changed( + city_lower, + rows[0].get("temp"), + rows[0].get("obs_time"), + source="cowin_obs", + ) results["mgm_nearby"] = rows results["nearby_source"] = "cowin_obs" try: @@ -1192,6 +1318,12 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour } results["mgm_nearby"] = [row] results["nearby_source"] = "cwa" + self._emit_temperature_patch_if_changed( + city_lower, + temp, + sc.get("observation_time"), + source="cwa", + ) def _attach_korean_amos_data( self, results: Dict, city_lower: str, use_fahrenheit: bool @@ -1209,6 +1341,13 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour city_lower, amos_data.get("temp_c"), amos_data.get("temp_source"), len(amos_data.get("runway_obs", {}).get("runway_pairs", []) or [])) results["amos"] = amos_data + self._emit_temperature_patch_if_changed( + city_lower, + amos_data.get("temp_c"), + amos_data.get("observation_time"), + source="amos", + extra={"amos": amos_data}, + ) try: DBManager().append_airport_obs( icao=amos_data.get("icao") or "", @@ -1255,6 +1394,13 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour ) # Reuse the existing `amos` detail shape consumed by dashboard runway panels. results["amos"] = amsc_data + self._emit_temperature_patch_if_changed( + city_lower, + amsc_data.get("temp_c"), + amsc_data.get("observation_time"), + source="amsc_awos", + extra={"amos": amsc_data}, + ) try: icao = amsc_data.get("icao") or "" obs_time = amsc_data.get("observation_time") or datetime.now().isoformat() @@ -1332,6 +1478,12 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour if obs["icao"] == icao: # Inject into results so it flows through to airport_primary results["madis_hfmetar_current"] = obs + self._emit_temperature_patch_if_changed( + city_lower, + obs.get("temp_c"), + obs.get("obs_time"), + source="madis_hfmetar", + ) try: DBManager().append_airport_obs( icao=icao, @@ -1359,6 +1511,12 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour return results["singapore_mss_current"] = obs + self._emit_temperature_patch_if_changed( + city_lower, + obs.get("temp_c"), + obs.get("obs_time"), + source="singapore_mss", + ) try: DBManager().append_airport_obs( icao="WSSS", diff --git a/web/app_factory.py b/web/app_factory.py index a493c046..0b51fc03 100644 --- a/web/app_factory.py +++ b/web/app_factory.py @@ -16,6 +16,7 @@ from web.routers.auth import router as auth_router from web.routers.ops import router as ops_router from web.routers.payments import router as payments_router from web.routers.scan import router as scan_router +from web.routers.sse_router import router as sse_router from web.routers.system import router as system_router from web.routes import router as legacy_router from web.scan_terminal_service import start_scan_terminal_prewarm @@ -37,6 +38,7 @@ def create_app() -> FastAPI: core_app.include_router(auth_router) core_app.include_router(analytics_router) core_app.include_router(scan_router) + core_app.include_router(sse_router) core_app.include_router(payments_router) core_app.include_router(ops_router) core_app.include_router(legacy_router) diff --git a/web/routers/sse_router.py b/web/routers/sse_router.py new file mode 100644 index 00000000..06be4f5d --- /dev/null +++ b/web/routers/sse_router.py @@ -0,0 +1,39 @@ +"""SSE endpoints for live terminal patch delivery.""" + +from __future__ import annotations + +from typing import Any + +from fastapi import APIRouter, HTTPException, Request +from fastapi.responses import StreamingResponse + +from web.sse_manager import sse_manager + + +router = APIRouter(tags=["events"]) + + +@router.get("/api/events") +async def sse_events(request: Request): + user_id = getattr(request.state, "auth_user_id", None) or "anon" + return StreamingResponse( + sse_manager.event_stream(user_id), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache, no-transform", + "Connection": "keep-alive", + "X-Accel-Buffering": "no", + }, + ) + + +@router.post("/api/internal/collector-patch") +async def ingest_patch(patch: dict[str, Any]): + city = str(patch.get("city") or "").strip().lower() + changes = patch.get("changes") + if not city: + raise HTTPException(status_code=400, detail="city is required") + if not isinstance(changes, dict): + raise HTTPException(status_code=400, detail="changes must be an object") + event = sse_manager.broadcast(city, changes) + return {"ok": True, "revision": event["revision"]} diff --git a/web/sse_manager.py b/web/sse_manager.py new file mode 100644 index 00000000..8876f9ec --- /dev/null +++ b/web/sse_manager.py @@ -0,0 +1,92 @@ +"""In-process SSE patch broadcaster for live terminal updates.""" + +from __future__ import annotations + +import asyncio +import json +import threading +import time +from collections import defaultdict +from typing import Any, AsyncIterator, DefaultDict + + +HEARTBEAT_INTERVAL_SECONDS = 30 +QUEUE_MAXSIZE = 256 + + +class SseManager: + def __init__(self) -> None: + self._queues: DefaultDict[str, set[asyncio.Queue[dict[str, Any]]]] = defaultdict(set) + self._lock = threading.RLock() + self._revision = 0 + + def _next_revision(self) -> int: + with self._lock: + self._revision += 1 + return self._revision + + def broadcast(self, city: str, changes: dict[str, Any]) -> dict[str, Any]: + event = { + "type": "city_patch", + "city": str(city or "").strip().lower(), + "changes": changes or {}, + "revision": self._next_revision(), + "ts": int(time.time() * 1000), + } + if not event["city"]: + return event + + with self._lock: + queues = [queue for queue_set in self._queues.values() for queue in queue_set] + + for queue in queues: + try: + queue.put_nowait(event) + except asyncio.QueueFull: + try: + queue.get_nowait() + except asyncio.QueueEmpty: + pass + try: + queue.put_nowait(event) + except asyncio.QueueFull: + pass + return event + + async def event_stream(self, user_id: str) -> AsyncIterator[str]: + user_key = str(user_id or "anon") + queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue(maxsize=QUEUE_MAXSIZE) + with self._lock: + self._queues[user_key].add(queue) + + try: + yield self._format_event({ + "type": "connected", + "revision": self._revision, + "ts": int(time.time() * 1000), + }) + while True: + try: + event = await asyncio.wait_for( + queue.get(), + timeout=HEARTBEAT_INTERVAL_SECONDS, + ) + except asyncio.TimeoutError: + event = { + "type": "heartbeat", + "revision": self._revision, + "ts": int(time.time() * 1000), + } + yield self._format_event(event) + finally: + with self._lock: + self._queues[user_key].discard(queue) + if not self._queues[user_key]: + self._queues.pop(user_key, None) + + @staticmethod + def _format_event(event: dict[str, Any]) -> str: + return f"data: {json.dumps(event, ensure_ascii=False, separators=(',', ':'))}\n\n" + + +sse_manager = SseManager()