From 645e304b3e36844445cd32035d2fb4b34a8d5f8c Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Tue, 26 May 2026 22:27:55 +0800 Subject: [PATCH] feat: implement SSE-based real-time event distribution system with replay support and heartbeat functionality --- .vscode/settings.json | 5 +- .../2026-05-26-production-realtime-sse.md | 71 +++++ frontend/app/api/events/route.ts | 7 +- .../LiveTemperatureThresholdChart.tsx | 83 ++++-- .../__tests__/ssePatchArchitecture.test.ts | 30 +++ ...temperatureDefaultVisibilityPolicy.test.ts | 92 +++++++ frontend/hooks/use-sse-patches.ts | 222 +++++++++++++--- pyproject.toml | 4 + src/data_collection/hko_obs_sources.py | 19 +- src/database/db_manager.py | 24 ++ tests/test_full_cache.py | 20 +- tests/test_realtime_event_store.py | 88 ++++++ tests/test_realtime_patch_schema.py | 86 ++++++ tests/test_sse_replay.py | 133 ++++++++++ web/analysis_service.py | 70 ++--- web/realtime_event_store.py | 251 ++++++++++++++++++ web/realtime_patch_schema.py | 200 ++++++++++++++ web/routers/sse_router.py | 90 ++++++- web/services/city_payloads.py | 4 +- web/sse_manager.py | 69 ++++- 20 files changed, 1449 insertions(+), 119 deletions(-) create mode 100644 docs/superpowers/plans/2026-05-26-production-realtime-sse.md create mode 100644 tests/test_realtime_event_store.py create mode 100644 tests/test_realtime_patch_schema.py create mode 100644 tests/test_sse_replay.py create mode 100644 web/realtime_event_store.py create mode 100644 web/realtime_patch_schema.py diff --git a/.vscode/settings.json b/.vscode/settings.json index 80abd191..6749c965 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -1,5 +1,8 @@ { "css.validate": false, "scss.validate": false, - "less.validate": false + "less.validate": false, + "python.analysis.extraPaths": [ + "./" + ] } diff --git a/docs/superpowers/plans/2026-05-26-production-realtime-sse.md b/docs/superpowers/plans/2026-05-26-production-realtime-sse.md new file mode 100644 index 00000000..3dadc5bc --- /dev/null +++ b/docs/superpowers/plans/2026-05-26-production-realtime-sse.md @@ -0,0 +1,71 @@ +# Production Realtime SSE Patch Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Upgrade PolyWeather terminal charts from in-process best-effort SSE patches to a replayable, city-scoped, versioned realtime observation stream for PM highest-temperature prediction workflows. + +**Architecture:** Keep HTTP APIs as the full snapshot/source-of-truth layer. Add a short-window SQLite event log for SSE replay, version observations as `city_observation_patch.v1`, fan out live events through the existing SSE manager, and let the frontend subscribe only to visible cities with `since_revision` reconnect replay. + +**Tech Stack:** FastAPI, SQLite/WAL via `DBManager`, in-process `asyncio.Queue` SSE fanout, React/Next.js `EventSource`, TypeScript external store hooks. + +--- + +## Constraints + +- Default replay retention is 6 hours because this event log is not the business history store. +- The retention can be raised with `POLYWEATHER_PATCH_EVENT_RETENTION_HOURS`, but the product does not need all-day patch retention. +- First production step is SQLite-only; Redis/Postgres pub/sub remains a later multi-instance extension. +- Existing legacy `city_patch` ingest and frontend handling must keep working during rollout. + +## Tasks + +- [ ] Backend schema tests + - Add tests proving legacy collector payloads normalize to `city_observation_patch.v1`. + - Cover runway point conversion from `amos.runway_obs.point_temperatures`. + - Cover invalid payload rejection when city and useful observation data are missing. + +- [ ] Backend schema implementation + - Add `web/realtime_patch_schema.py`. + - Normalize city/source/obs time/temp/max/runway payload fields. + - Keep payload small and JSON-serializable. + +- [ ] Event store tests + - Add tests for monotonic SQLite revisions. + - Add city-filtered replay tests. + - Add retention cleanup tests for stale replay rows. + +- [ ] Event store implementation + - Add `observation_patch_events` table and indexes in `DBManager`. + - Add `web/realtime_event_store.py` for append, replay, latest revision, and cleanup. + - Use `POLYWEATHER_PATCH_EVENT_RETENTION_HOURS=6` as the default. + +- [ ] SSE replay tests + - Add tests for `/api/events?cities=...&since_revision=...`. + - Verify replay only returns subscribed cities. + - Verify replay over limit emits `resync_required`. + +- [ ] SSE implementation + - Update router to parse `cities`, `since_revision`, and bounded `replay_limit`. + - Write normalized events to SQLite before broadcasting. + - Update manager to track per-connection city subscriptions while keeping heartbeat behavior. + +- [ ] Frontend SSE tests + - Extend architecture tests for v1 schema, `cities`, `since_revision`, replay/resync handling. + - Add chart merge coverage for v1 runway point payloads. + +- [ ] Frontend SSE implementation + - Update `use-sse-patches.ts` to normalize v1 and legacy patch events. + - Track global `lastRevision`. + - Reconnect with visible-city `cities` and `since_revision`. + - Expose a resync signal for charts when the server cannot replay. + +- [ ] Chart/list integration + - Ensure visible charts register their city subscription. + - Keep terminal row patching lightweight: current temp, current max, local time, revision. + - Append v1 temp/runway points into existing chart series without forcing full-detail polling. + +- [ ] Verification + - Run targeted backend pytest files. + - Run `npm run test:business`. + - Run `npm run typecheck`. + - Run `npm run build`. diff --git a/frontend/app/api/events/route.ts b/frontend/app/api/events/route.ts index 637b5bce..c9601c61 100644 --- a/frontend/app/api/events/route.ts +++ b/frontend/app/api/events/route.ts @@ -13,7 +13,12 @@ export async function GET(req: NextRequest) { ); } - const upstream = await fetch(`${API_BASE.replace(/\/+$/, "")}/api/events`, { + const upstreamUrl = new URL(`${API_BASE.replace(/\/+$/, "")}/api/events`); + req.nextUrl.searchParams.forEach((value, key) => { + upstreamUrl.searchParams.append(key, value); + }); + + const upstream = await fetch(upstreamUrl.toString(), { cache: "no-store", headers: { Accept: "text/event-stream", diff --git a/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx b/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx index 83766d24..c3d09486 100644 --- a/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx +++ b/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx @@ -24,7 +24,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 { useLatestPatch, useSseResyncVersion, type CityPatch } from "@/hooks/use-sse-patches"; import { Panel } from "@/components/dashboard/scan-terminal/Panel"; import { rowName, temp } from "@/components/dashboard/scan-terminal/utils"; @@ -82,6 +82,28 @@ function getVisibleTemperatureSeries( }); } +function getActiveTemperatureSeries( + city: string, + chartSeries: EvidenceSeries[], + userToggledKeys: Record, + showRunwayDetails: boolean, +) { + const rawVisible = getVisibleTemperatureSeries(city, chartSeries, userToggledKeys); + const hasRunwayMax = rawVisible.some((item) => item.key === "runway_max"); + + return rawVisible.filter((item) => { + const isIndividualRunway = + item.key.startsWith("runway_") && item.key !== "runway_max"; + if (showRunwayDetails) { + return item.key !== "runway_max"; + } + if (!hasRunwayMax) { + return true; + } + return !isIndividualRunway; + }); +} + function buildRunwayPlates( amos: AmosData | null | undefined, row: ScanOpportunityRow | null, @@ -194,8 +216,8 @@ type RunwayHistorySeries = { const MAX_OBS_POINTS = 1440; const HOURLY_CACHE_TTL_MS = DASHBOARD_REFRESH_POLICY_MS.metar; -const FULL_DAY_SLOT_MINUTES = 30; -const FULL_DAY_SLOTS = 48; +const FULL_DAY_SLOT_MINUTES = 1; +const FULL_DAY_SLOTS = (24 * 60) / FULL_DAY_SLOT_MINUTES; const SLOT_INTERVAL_MS = FULL_DAY_SLOT_MINUTES * 60 * 1000; const _hourlyCache = new Map(); const _hourlyRequestCache = new Map>(); @@ -622,7 +644,12 @@ function mergePatchIntoHourly( const amosChanges = changes.amos as Record | undefined; const obsTimeVal = obsTime || amosChanges?.observation_time || amosChanges?.observation_time_local; const runwayObs = amosChanges?.runway_obs; - if (runwayObs && Array.isArray(runwayObs.point_temperatures) && obsTimeVal) { + const runwayPoints = Array.isArray(changes.runway_points) + ? changes.runway_points + : runwayObs && Array.isArray(runwayObs.point_temperatures) + ? runwayObs.point_temperatures + : []; + if (runwayPoints.length && obsTimeVal) { const history: Record>> = {}; const sourceHistory = next.runwayPlateHistory || (next.amos as any)?.runway_plate_history || {}; @@ -634,10 +661,10 @@ function mergePatchIntoHourly( }); // Append new points from point_temperatures - runwayObs.point_temperatures.forEach((pt: any) => { + runwayPoints.forEach((pt: any) => { const rwy = pt.runway || ""; if (!rwy) return; - const tempVal = validNumber(pt.target_runway_max) ?? validNumber(pt.tdz_temp) ?? validNumber(pt.end_temp); + const tempVal = validNumber(pt.temp) ?? validNumber(pt.target_runway_max) ?? validNumber(pt.tdz_temp) ?? validNumber(pt.end_temp); if (tempVal === null) return; const rwyHistory = history[rwy] || []; @@ -653,6 +680,13 @@ function mergePatchIntoHourly( }); next.runwayPlateHistory = history; + next.amos = { + ...(next.amos || {}), + runway_obs: { + ...((next.amos as any)?.runway_obs || {}), + point_temperatures: runwayPoints, + }, + } as any; if (next.amos) { (next.amos as any).runway_plate_history = history; } @@ -1405,13 +1439,14 @@ export function LiveTemperatureThresholdChart({ const [hourly, setHourly] = useState(null); const city = String(row?.city || "").toLowerCase().trim(); const latestPatch = useLatestPatch(city); + const resyncVersion = useSseResyncVersion(); 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); - const [showRunwayDetails, setShowRunwayDetails] = useState(false); + const [showRunwayDetails, setShowRunwayDetails] = useState(true); const [refAreaLeft, setRefAreaLeft] = useState(null); const [refAreaRight, setRefAreaRight] = useState(null); const [zoomRange, setZoomRange] = useState<[number, number] | null>(null); @@ -1420,6 +1455,7 @@ export function LiveTemperatureThresholdChart({ useEffect(() => { setUserToggledKeys({}); setZoomRange(null); + setShowRunwayDetails(true); lastPatchAtRef.current = Date.now(); lastAppliedPatchRevisionRef.current = 0; }, [city, timeframe]); @@ -1480,6 +1516,20 @@ export function LiveTemperatureThresholdChart({ setHourly((prev) => mergePatchIntoHourly(prev ?? seedHourlyForecastFromRow(row), latestPatch)); }, [latestPatch, row]); + useEffect(() => { + if (!resyncVersion || !city) return; + let cancelled = false; + fetchHourlyForecastForCity(city, { ignoreCache: true, resolution: targetResolution }) + .then((data) => { + if (cancelled || !data) return; + setHourly(data); + }) + .catch(() => {}); + return () => { + cancelled = true; + }; + }, [resyncVersion, city, targetResolution]); + // ── SSE fallback: only full-fetch if a visible chart has seen no patch for 2 minutes ── useEffect(() => { if (!shouldPollLiveChart({ city, compact, isActive, isMaximized })) return; @@ -1581,15 +1631,12 @@ export function LiveTemperatureThresholdChart({ }; const activeSeries = useMemo(() => { - const rawVisible = getVisibleTemperatureSeries(city, chartSeries, userToggledKeys); - return rawVisible.filter((s) => { - const isIndividualRunway = s.key.startsWith("runway_") && s.key !== "runway_max"; - if (showRunwayDetails) { - return s.key !== "runway_max"; - } else { - return !isIndividualRunway; - } - }); + return getActiveTemperatureSeries( + city, + chartSeries, + userToggledKeys, + showRunwayDetails, + ); }, [chartSeries, userToggledKeys, city, showRunwayDetails]); const normalizedKey = normalizeCityKey(row?.city); @@ -1903,7 +1950,7 @@ export function LiveTemperatureThresholdChart({
- {isEn ? "METAR Settlement (30m) · Daily High" : `${metarHeaderLabel} · 当日最高`} + {isEn ? "METAR Settlement · Daily High" : `${metarHeaderLabel} · 当日最高`} {temp(observedHighMetar)} @@ -2226,5 +2273,7 @@ export function __buildTemperatureChartDataForTest( export const __isTemperatureSeriesVisibleByDefaultForTest = isTemperatureSeriesVisibleByDefault; export const __getVisibleTemperatureSeriesForTest = getVisibleTemperatureSeries; +export const __getActiveTemperatureSeriesForTest = getActiveTemperatureSeries; export const __getObservationDisplayMetricsForTest = getObservationDisplayMetrics; export const __shouldPollLiveChartForTest = shouldPollLiveChart; +export const __mergePatchIntoHourlyForTest = mergePatchIntoHourly; diff --git a/frontend/components/dashboard/scan-terminal/__tests__/ssePatchArchitecture.test.ts b/frontend/components/dashboard/scan-terminal/__tests__/ssePatchArchitecture.test.ts index 5b5140c5..969527e6 100644 --- a/frontend/components/dashboard/scan-terminal/__tests__/ssePatchArchitecture.test.ts +++ b/frontend/components/dashboard/scan-terminal/__tests__/ssePatchArchitecture.test.ts @@ -21,15 +21,35 @@ export function runTests() { 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("broadcast_event"), "SSE manager must broadcast stored replayable events"); assert(sseManager.includes("event_stream("), "SSE manager must expose an async event_stream(user_id)"); + assert(sseManager.includes("_queue_cities"), "SSE manager must track per-connection city subscriptions"); 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 schemaPath = path.join(repoRoot, "web", "realtime_patch_schema.py"); + assert(fs.existsSync(schemaPath), "backend must define a versioned realtime patch schema module"); + const schema = fs.readFileSync(schemaPath, "utf8"); + assert(schema.includes("city_observation_patch.v1"), "patch schema must expose city_observation_patch.v1"); + assert(schema.includes("normalize_observation_patch"), "patch schema must normalize collector payloads"); + assert(schema.includes("runway_points"), "patch schema must preserve runway point observations"); + + const storePath = path.join(repoRoot, "web", "realtime_event_store.py"); + assert(fs.existsSync(storePath), "backend must define a realtime event replay store"); + const store = fs.readFileSync(storePath, "utf8"); + assert(store.includes("observation_patch_events"), "event store must use the SQLite observation_patch_events table"); + assert(store.includes("replay_events"), "event store must expose replay_events"); + assert(store.includes("replay_requires_resync"), "event store must detect incomplete replay windows"); + 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("cities"), "SSE route must accept a cities query parameter"); + assert(sseRouter.includes("since_revision"), "SSE route must accept since_revision for replay"); + assert(sseRouter.includes("replay_limit"), "SSE route must bound replay batches"); + assert(sseRouter.includes("resync_required"), "SSE route must emit resync_required when replay is incomplete"); 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"); @@ -53,14 +73,24 @@ export function runTests() { 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("city_observation_patch.v1"), "frontend patch hook must accept v1 observation patch events"); + assert(hook.includes("subscribedCities"), "frontend patch hook must track the visible city subscription set"); + assert(hook.includes("since_revision"), "frontend patch hook must reconnect with since_revision"); + assert(hook.includes("resync_required"), "frontend patch hook must react to server resync_required events"); + assert(hook.includes("lastRevision"), "frontend patch hook must track the global last processed revision"); 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 bffEventsRoute = readFrontendFile("app", "api", "events", "route.ts"); + assert(bffEventsRoute.includes("searchParams"), "Next.js SSE proxy must forward query parameters to FastAPI"); + 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("useSseResyncVersion"), "temperature chart must resync full detail when SSE replay is incomplete"); + assert(chart.includes("runway_points"), "temperature chart must merge v1 runway_points into runway history"); 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)"), diff --git a/frontend/components/dashboard/scan-terminal/__tests__/temperatureDefaultVisibilityPolicy.test.ts b/frontend/components/dashboard/scan-terminal/__tests__/temperatureDefaultVisibilityPolicy.test.ts index 1ab701a2..53d2e5ab 100644 --- a/frontend/components/dashboard/scan-terminal/__tests__/temperatureDefaultVisibilityPolicy.test.ts +++ b/frontend/components/dashboard/scan-terminal/__tests__/temperatureDefaultVisibilityPolicy.test.ts @@ -1,8 +1,10 @@ import { __buildTemperatureChartDataForTest, + __getActiveTemperatureSeriesForTest, __getObservationDisplayMetricsForTest, __getVisibleTemperatureSeriesForTest, __isTemperatureSeriesVisibleByDefaultForTest, + __mergePatchIntoHourlyForTest, } from "@/components/dashboard/scan-terminal/LiveTemperatureThresholdChart"; function assert(condition: unknown, message: string) { @@ -57,6 +59,7 @@ export function runTests() { const { series } = __buildTemperatureChartDataForTest(guangzhou, hourly, "1D"); const defaultVisibleSeries = __getVisibleTemperatureSeriesForTest("guangzhou", series, {}); + const activeDefaultSeries = __getActiveTemperatureSeriesForTest("guangzhou", series, {}, true); const settlementRunway = seriesByKey(series, "runway_02L_20R") as any; assert(settlementRunway, "settlement runway should use a stable runway-pair key"); @@ -77,6 +80,14 @@ export function runTests() { __isTemperatureSeriesVisibleByDefaultForTest("guangzhou", "runway_02L_20R"), "runway series should be visible by default", ); + assert( + activeDefaultSeries.some((item) => item.key === "runway_02L_20R"), + "settlement runway should remain in the active chart series by default", + ); + assert( + activeDefaultSeries.some((item) => item.key === "runway_01L_19R"), + "auxiliary runway should remain in the active chart series by default", + ); assert( __isTemperatureSeriesVisibleByDefaultForTest("guangzhou", "settlement"), "settlement/HKO observations should be visible by default", @@ -284,6 +295,87 @@ export function runTests() { assert(madisSeries.label.includes("MADIS"), "US MADIS series should be labeled as NOAA MADIS instead of plain METAR"); assert(madisSeries.values.filter((value: number | null) => value !== null).length >= 2, "MADIS series should keep sub-hourly observations"); + const newYorkMinuteStream = __buildTemperatureChartDataForTest( + { + city: "new york", + local_date: "2026-05-25", + local_time: "10:04", + tz_offset_seconds: -4 * 60 * 60, + airport: "KLGA", + } as any, + { + localTime: "10:04", + times: ["00:00", "06:00", "12:00", "18:00"], + temps: [55, 57, 65, 72], + airportPrimary: { + source_code: "madis_hfmetar", + source_label: "NOAA MADIS", + }, + airportPrimaryTodayObs: [ + ["2026-05-25T14:01:00Z", 73.1], + ["2026-05-25T14:02:00Z", 73.4], + ["2026-05-25T14:03:00Z", 73.8], + ], + } as any, + "1D", + ); + const minuteLabels = newYorkMinuteStream.data + .filter((point) => point.madis !== null) + .map((point) => point.label); + assert( + minuteLabels.includes("10:01") && + minuteLabels.includes("10:02") && + minuteLabels.includes("10:03"), + "live observation chart should preserve minute-level SSE patch points instead of collapsing them into a 30-minute bucket", + ); + + const chengduMergedHourly = __mergePatchIntoHourlyForTest( + { + localTime: "05:25", + times: ["00:00", "06:00", "12:00", "18:00"], + temps: [24, 28, 31, 27], + runwayPlateHistory: { + "02L/20R": [{ time: "05:20", temp: 24.2 }], + }, + } as any, + { + type: "city_observation_patch.v1", + city: "chengdu", + revision: 12, + changes: { + temp: 24.8, + obs_time: "2026-05-26 05:26:00", + source: "amsc_awos", + runway_points: [ + { + runway: "02L/20R", + temp: 25.1, + tdz_temp: 24.7, + mid_temp: 24.9, + end_temp: 25.1, + target_runway_max: 25.1, + }, + ], + }, + } as any, + ); + const chengduMergedChart = __buildTemperatureChartDataForTest( + { + city: "chengdu", + local_date: "2026-05-26", + local_time: "05:26", + tz_offset_seconds: 8 * 60 * 60, + } as any, + chengduMergedHourly as any, + "1D", + ); + const chengduMergedRunway = seriesByKey(chengduMergedChart.series, "runway_02L_20R") as any; + assert(chengduMergedRunway, "v1 runway_points patch should update the runway series"); + assert( + chengduMergedRunway.values.some((value: number | null) => value === 25.1), + "v1 runway_points patch should append the latest runway max point to the chart", + ); + const shanghaiDebFromDetail = __buildTemperatureChartDataForTest( { city: "shanghai", diff --git a/frontend/hooks/use-sse-patches.ts b/frontend/hooks/use-sse-patches.ts index 69b7b1d7..005a8d96 100644 --- a/frontend/hooks/use-sse-patches.ts +++ b/frontend/hooks/use-sse-patches.ts @@ -3,6 +3,8 @@ import { useEffect, useSyncExternalStore } from "react"; import { resolveBackendApiUrl } from "@/lib/backend-api"; +const V1_EVENT_TYPE = "city_observation_patch.v1"; + export type CityPatch = { type?: string; city: string; @@ -11,29 +13,82 @@ export type CityPatch = { ts?: number; }; +type ObservationPatchV1 = { + type?: string; + city?: string; + source?: string; + obs_time?: string | null; + revision?: number; + ts?: number; + payload?: Record; +}; + const latestPatches = new Map(); const latestRevisions = new Map(); const cityListeners = new Map void>>(); const globalListeners = new Set<() => void>(); +const resyncListeners = new Set<() => void>(); +const subscribedCities = new Map(); let eventSource: EventSource | null = null; let reconnectTimer: ReturnType | null = null; let reconnectAttempt = 0; let patchVersion = 0; +let resyncVersion = 0; +let lastRevision = 0; let useFallbackUrl = false; +let activeConnectionKey = ""; function normalizeCityKey(city: string | null | undefined) { return String(city || "").trim().toLowerCase(); } +function subscribedCityList() { + return Array.from(subscribedCities.keys()).sort(); +} + function notify(city: string) { patchVersion += 1; cityListeners.get(city)?.forEach((listener) => listener()); globalListeners.forEach((listener) => listener()); } +function notifyResync(latestServerRevision: number | null) { + if (latestServerRevision !== null) { + lastRevision = Math.max(lastRevision, latestServerRevision); + } + resyncVersion += 1; + resyncListeners.forEach((listener) => listener()); + globalListeners.forEach((listener) => listener()); +} + +function clearReconnectTimer() { + if (!reconnectTimer) return; + clearTimeout(reconnectTimer); + reconnectTimer = null; +} + +function buildSseUrl(baseUrl: string) { + const params = new URLSearchParams(); + const cities = subscribedCityList(); + if (cities.length) { + params.set("cities", cities.join(",")); + } + if (lastRevision > 0) { + params.set("since_revision", String(lastRevision)); + } + params.set("replay_limit", "500"); + + const query = params.toString(); + return query ? `${baseUrl}?${query}` : baseUrl; +} + +function currentConnectionKey() { + return `${useFallbackUrl ? "fallback" : "direct"}:${subscribedCityList().join("|")}:${lastRevision}`; +} + function scheduleReconnect() { - if (reconnectTimer || typeof window === "undefined") return; + if (reconnectTimer || typeof window === "undefined" || subscribedCities.size === 0) return; const delayMs = Math.min(30_000, 1_000 * Math.max(1, 2 ** reconnectAttempt)); reconnectAttempt += 1; reconnectTimer = setTimeout(() => { @@ -46,30 +101,31 @@ function closeEventSource() { if (!eventSource) return; eventSource.close(); eventSource = null; + activeConnectionKey = ""; +} + +function reconnectNow() { + if (typeof window === "undefined") return; + clearReconnectTimer(); + closeEventSource(); + connectSsePatches(); } function connectSsePatches() { - if (typeof window === "undefined" || eventSource) return; + if (typeof window === "undefined" || eventSource || subscribedCities.size === 0) return; - let url = resolveBackendApiUrl("/api/events"); - if (useFallbackUrl) { - url = "/api/events"; - console.log("[SSE] Falling back to same-origin BFF proxy URL:", url); - } else { - console.log("[SSE] Attempting to connect to direct URL:", url); - } + const baseUrl = useFallbackUrl ? "/api/events" : resolveBackendApiUrl("/api/events"); + const url = buildSseUrl(baseUrl); + activeConnectionKey = currentConnectionKey(); try { - closeEventSource(); eventSource = new EventSource(url, { withCredentials: true }); eventSource.onopen = () => { - console.log("[SSE] Connection established successfully to:", url); reconnectAttempt = 0; }; eventSource.onmessage = (event) => { - console.log("[SSE] Received patch message:", event.data); try { applySsePatch(JSON.parse(event.data)); } catch (err) { @@ -81,48 +137,127 @@ function connectSsePatches() { console.error("[SSE] Connection error or stream closed:", err); closeEventSource(); if (!useFallbackUrl && url !== "/api/events") { - console.warn("[SSE] Direct connection failed. Switching to same-origin BFF proxy fallback for next attempt."); useFallbackUrl = true; } scheduleReconnect(); }; } catch (err) { console.error("[SSE] Exception thrown while instantiating EventSource:", err); - if (!useFallbackUrl && url !== "/api/events") { + closeEventSource(); + if (!useFallbackUrl && baseUrl !== "/api/events") { useFallbackUrl = true; } scheduleReconnect(); } } -export function ensureSsePatchConnection() { - connectSsePatches(); +function ensureSsePatchConnection() { + if (subscribedCities.size === 0) { + closeEventSource(); + clearReconnectTimer(); + return; + } + if (eventSource && activeConnectionKey === currentConnectionKey()) return; + reconnectNow(); } -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; +function registerCitySubscription(city: string) { + const cityKey = normalizeCityKey(city); + if (!cityKey) return () => {}; + + const previousCount = subscribedCities.get(cityKey) ?? 0; + subscribedCities.set(cityKey, previousCount + 1); + if (previousCount === 0) { + ensureSsePatchConnection(); + } + + return () => { + const nextCount = (subscribedCities.get(cityKey) ?? 1) - 1; + if (nextCount <= 0) { + subscribedCities.delete(cityKey); + } else { + subscribedCities.set(cityKey, nextCount); + } + ensureSsePatchConnection(); + }; +} + +function normalizeLegacyPatch(patch: Partial): CityPatch | null { 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; + return null; } - - const previousRevision = latestRevisions.get(city) ?? 0; - if (revision <= previousRevision) return false; - - const normalizedPatch: CityPatch = { + return { 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); +} + +function normalizeV1Patch(patch: ObservationPatchV1): CityPatch | null { + const city = normalizeCityKey(patch.city); + const revision = Number(patch.revision); + const payload = patch.payload; + if (!city || !payload || typeof payload !== "object" || !Number.isFinite(revision)) { + return null; + } + + const changes: Record = { + ...payload, + source: typeof patch.source === "string" ? patch.source : payload.source, + obs_time: typeof patch.obs_time === "string" ? patch.obs_time : payload.obs_time, + schema_type: V1_EVENT_TYPE, + }; + + return { + type: V1_EVENT_TYPE, + city, + changes, + revision, + ts: typeof patch.ts === "number" ? patch.ts : Date.now(), + }; +} + +function normalizeIncomingPatch(payload: unknown): CityPatch | null { + if (!payload || typeof payload !== "object") return null; + const patch = payload as Partial & ObservationPatchV1; + if (patch.type === "city_patch" || !patch.type) { + return normalizeLegacyPatch(patch); + } + if (patch.type === V1_EVENT_TYPE) { + return normalizeV1Patch(patch); + } + return null; +} + +export function applySsePatch(payload: unknown) { + if (!payload || typeof payload !== "object") return false; + const event = payload as { type?: string; latest_revision?: number }; + + if (event.type === "connected" || event.type === "heartbeat") { + return false; + } + + if (event.type === "resync_required") { + const latestServerRevision = Number(event.latest_revision); + notifyResync(Number.isFinite(latestServerRevision) ? latestServerRevision : null); + return true; + } + + const normalizedPatch = normalizeIncomingPatch(payload); + if (!normalizedPatch) return false; + + const previousRevision = latestRevisions.get(normalizedPatch.city) ?? 0; + if (normalizedPatch.revision <= previousRevision) return false; + + latestRevisions.set(normalizedPatch.city, normalizedPatch.revision); + latestPatches.set(normalizedPatch.city, normalizedPatch); + lastRevision = Math.max(lastRevision, normalizedPatch.revision); + notify(normalizedPatch.city); return true; } @@ -131,14 +266,6 @@ export function getLatestPatchesSnapshot() { } export function useSsePatchVersion() { - if (typeof window !== "undefined") { - ensureSsePatchConnection(); - } - - useEffect(() => { - ensureSsePatchConnection(); - }, []); - return useSyncExternalStore( (listener) => { globalListeners.add(listener); @@ -149,16 +276,24 @@ export function useSsePatchVersion() { ); } +export function useSseResyncVersion() { + return useSyncExternalStore( + (listener) => { + resyncListeners.add(listener); + return () => resyncListeners.delete(listener); + }, + () => resyncVersion, + () => 0, + ); +} + export function useLatestPatch(city: string | null | undefined) { const cityKey = normalizeCityKey(city); - if (typeof window !== "undefined") { - ensureSsePatchConnection(); - } - useEffect(() => { - ensureSsePatchConnection(); - }, []); + if (!cityKey) return undefined; + return registerCitySubscription(cityKey); + }, [cityKey]); return useSyncExternalStore( (listener) => { @@ -177,3 +312,4 @@ export function useLatestPatch(city: string | null | undefined) { } export const __applySsePatchForTest = applySsePatch; +export const __buildSseUrlForTest = buildSseUrl; diff --git a/pyproject.toml b/pyproject.toml index 27c56656..96766019 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -57,3 +57,7 @@ skip-magic-trailing-comma = false # Like Black, automatically detect the appropriate line ending. line-ending = "auto" + +[tool.pyright] +include = ["src", "web", "tests"] +extraPaths = ["."] diff --git a/src/data_collection/hko_obs_sources.py b/src/data_collection/hko_obs_sources.py index ca6a5a63..a69458a0 100644 --- a/src/data_collection/hko_obs_sources.py +++ b/src/data_collection/hko_obs_sources.py @@ -11,11 +11,12 @@ import csv import os import io import time -from datetime import datetime +from datetime import datetime, timezone from typing import Any, Dict, Optional from loguru import logger +# pyrefly: ignore [missing-import] from src.utils.metrics import record_source_call HKO_BASE_URL = os.getenv("HKO_BASE_URL", "").strip() or "https://data.weather.gov.hk/weatherAPI/hko_data/regional-weather" @@ -34,11 +35,17 @@ HKO_STATIONS = { class HkoObsSourceMixin: + session: Any + timeout: Any + _hko_obs_cache: Dict[str, Any] + _hko_obs_cache_lock: Any + hko_obs_cache_ttl_sec: Any + def _hko_http_get(self, url: str) -> str: getter = getattr(self, "_http_get", None) if callable(getter): resp = getter(url) - return resp.text if hasattr(resp, "text") else resp + return str(resp.text) if hasattr(resp, "text") else str(resp) resp = self.session.get(url, timeout=self.timeout) resp.raise_for_status() return resp.text @@ -49,7 +56,7 @@ class HkoObsSourceMixin: use_fahrenheit: bool = False, ) -> Optional[Dict[str, Any]]: started = time.perf_counter() - city_key = str(city or "").strip().lower() + city_key = (city or "").strip().lower() meta = HKO_STATIONS.get(city_key) or {} if not meta: return None @@ -95,11 +102,11 @@ class HkoObsSourceMixin: result = { "source": "hko_obs", - "timestamp": datetime.utcnow().isoformat(), + "timestamp": datetime.now(timezone.utc).isoformat(), "station_code": meta["code"], "station_name": meta["label"], "icao": meta["icao"], - "obs_time": obs_iso or datetime.utcnow().isoformat(), + "obs_time": obs_iso or datetime.now(timezone.utc).isoformat(), "current": { "temp": temp, }, @@ -132,7 +139,7 @@ class HkoObsSourceMixin: current = self.fetch_hko_obs_current(city, use_fahrenheit=use_fahrenheit) if not current: return [] - meta = HKO_STATIONS.get(str(city or "").strip().lower()) or {} + meta = HKO_STATIONS.get((city or "").strip().lower()) or {} return [ { "name": meta.get("label") or "HKO Station", diff --git a/src/database/db_manager.py b/src/database/db_manager.py index eac451fa..3ffe3ddd 100644 --- a/src/database/db_manager.py +++ b/src/database/db_manager.py @@ -339,6 +339,30 @@ class DBManager: conn.execute( "CREATE INDEX IF NOT EXISTS idx_payment_audit_events_created_at ON payment_audit_events(created_at DESC)" ) + conn.execute(""" + CREATE TABLE IF NOT EXISTS observation_patch_events ( + revision INTEGER PRIMARY KEY AUTOINCREMENT, + schema_type TEXT NOT NULL, + schema_version INTEGER NOT NULL, + city TEXT NOT NULL, + source TEXT NOT NULL, + obs_time TEXT, + payload_json TEXT NOT NULL, + created_at TEXT NOT NULL + ) + """) + conn.execute( + """ + CREATE INDEX IF NOT EXISTS idx_observation_patch_events_city_revision + ON observation_patch_events(city, revision) + """ + ) + conn.execute( + """ + CREATE INDEX IF NOT EXISTS idx_observation_patch_events_created_at + ON observation_patch_events(created_at) + """ + ) conn.execute(""" CREATE TABLE IF NOT EXISTS app_analytics_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, diff --git a/tests/test_full_cache.py b/tests/test_full_cache.py index 2d12cc46..305fa709 100644 --- a/tests/test_full_cache.py +++ b/tests/test_full_cache.py @@ -3,8 +3,9 @@ import os from src.database.db_manager import DBManager def test_full_cache_lifecycle(): - with tempfile.TemporaryDirectory() as tmpdir: - db_path = os.path.join(tmpdir, "test.db") + tmpdir = tempfile.TemporaryDirectory() + try: + db_path = os.path.join(tmpdir.name, "test.db") if hasattr(DBManager, "_initialized_paths"): DBManager._initialized_paths.clear() @@ -31,3 +32,18 @@ def test_full_cache_lifecycle(): assert cached["payload"] == payload assert cached["version"] == "v1" assert cached["source_fingerprint"] == "testcity:full" + + # Release database file handles for Windows temp directory cleanup + del db + import gc + gc.collect() + finally: + try: + tmpdir.cleanup() + except PermissionError: + import gc + gc.collect() + try: + tmpdir.cleanup() + except Exception: + pass diff --git a/tests/test_realtime_event_store.py b/tests/test_realtime_event_store.py new file mode 100644 index 00000000..a1aaa969 --- /dev/null +++ b/tests/test_realtime_event_store.py @@ -0,0 +1,88 @@ +import sqlite3 +from datetime import datetime, timezone + +from src.database.db_manager import DBManager +from web.realtime_event_store import RealtimeEventStore +from web.realtime_patch_schema import normalize_observation_patch + + +def _event(city: str, temp: float, source: str = "metar"): + return normalize_observation_patch( + { + "city": city, + "changes": { + "temp": temp, + "obs_time": "2026-05-26T08:15:00Z", + "source": source, + }, + } + ) + + +def test_event_store_appends_monotonic_revisions_and_replays_by_city(tmp_path): + db_path = str(tmp_path / "polyweather.db") + DBManager._initialized_paths.clear() + store = RealtimeEventStore(db_path=db_path) + + taipei = store.append_event(_event("taipei", 31.2)) + seoul = store.append_event(_event("seoul", 27.8)) + taipei_next = store.append_event(_event("taipei", 31.5)) + + assert taipei["revision"] == 1 + assert seoul["revision"] == 2 + assert taipei_next["revision"] == 3 + assert store.latest_revision() == 3 + + replay = store.replay_events(cities={"taipei"}, since_revision=1, limit=10) + + assert [event["revision"] for event in replay] == [3] + assert replay[0]["city"] == "taipei" + assert replay[0]["payload"]["temp"] == 31.5 + + +def test_event_store_cleanup_uses_short_replay_retention(tmp_path): + db_path = str(tmp_path / "polyweather.db") + DBManager._initialized_paths.clear() + store = RealtimeEventStore(db_path=db_path) + + old_event = store.append_event(_event("taipei", 30.0)) + fresh_event = store.append_event(_event("taipei", 31.0)) + + with sqlite3.connect(db_path) as conn: + conn.execute( + "UPDATE observation_patch_events SET created_at = ? WHERE revision = ?", + ("2026-05-26T00:00:00+00:00", old_event["revision"]), + ) + conn.execute( + "UPDATE observation_patch_events SET created_at = ? WHERE revision = ?", + ("2026-05-26T11:30:00+00:00", fresh_event["revision"]), + ) + conn.commit() + + deleted = store.cleanup_old_events( + retention_hours=6, + now=datetime(2026, 5, 26, 12, 0, tzinfo=timezone.utc), + ) + + assert deleted == 1 + replay = store.replay_events(cities={"taipei"}, since_revision=0, limit=10) + assert [event["revision"] for event in replay] == [fresh_event["revision"]] + + +def test_event_store_reports_replay_gap_when_limit_is_exceeded(tmp_path): + db_path = str(tmp_path / "polyweather.db") + DBManager._initialized_paths.clear() + store = RealtimeEventStore(db_path=db_path) + + for temp in [30.0, 30.5, 31.0]: + store.append_event(_event("hong kong", temp, source="hko")) + + replay = store.replay_events(cities={"hong kong"}, since_revision=0, limit=2) + + assert len(replay) == 2 + assert store.replay_requires_resync( + cities={"hong kong"}, + since_revision=0, + replay_count=len(replay), + limit=2, + ) diff --git a/tests/test_realtime_patch_schema.py b/tests/test_realtime_patch_schema.py new file mode 100644 index 00000000..5844e36a --- /dev/null +++ b/tests/test_realtime_patch_schema.py @@ -0,0 +1,86 @@ +import pytest + +from web.realtime_patch_schema import ( + PatchValidationError, + normalize_observation_patch, +) + + +def test_legacy_temperature_patch_normalizes_to_v1_with_runway_points(): + event = normalize_observation_patch( + { + "city": " Seoul ", + "changes": { + "temp": "31.25", + "max_so_far": 32.1, + "obs_time": "2026-05-26T08:15:00Z", + "source": "amos", + "amos": { + "icao": "RKSS", + "station_label": "Gimpo Airport", + "runway_obs": { + "point_temperatures": [ + { + "runway": "14L/32R", + "tdz_temp": 31.2, + "mid_temp": 31.6, + "end_temp": 31.8, + "target_runway_max": 31.8, + } + ] + }, + }, + }, + } + ) + + assert event["type"] == "city_observation_patch.v1" + assert event["schema_type"] == "city_observation_patch" + assert event["schema_version"] == 1 + assert event["city"] == "seoul" + assert event["source"] == "amos" + assert event["obs_time"] == "2026-05-26T08:15:00Z" + assert event["payload"]["temp"] == 31.25 + assert event["payload"]["max_so_far"] == 32.1 + assert event["payload"]["station_code"] == "RKSS" + assert event["payload"]["station_label"] == "Gimpo Airport" + assert event["payload"]["unit"] == "celsius" + assert event["payload"]["runway_points"] == [ + { + "runway": "14L/32R", + "temp": 31.8, + "tdz_temp": 31.2, + "mid_temp": 31.6, + "end_temp": 31.8, + "target_runway_max": 31.8, + } + ] + + +def test_v1_patch_payload_is_accepted_and_normalized(): + event = normalize_observation_patch( + { + "type": "city_observation_patch.v1", + "city": "Taipei", + "source": "cwa", + "obs_time": "2026-05-26T07:01:00Z", + "payload": { + "temp": 29.4, + "station_code": "46692", + "runway_points": [{"runway": "05/23", "temp": 30.2}], + }, + } + ) + + assert event["city"] == "taipei" + assert event["source"] == "cwa" + assert event["payload"]["temp"] == 29.4 + assert event["payload"]["runway_points"][0]["temp"] == 30.2 + + +def test_invalid_patch_without_city_or_observation_data_is_rejected(): + with pytest.raises(PatchValidationError): + normalize_observation_patch({"changes": {"temp": 21.0}}) + + with pytest.raises(PatchValidationError): + normalize_observation_patch({"city": "taipei", "changes": {"source": "metar"}}) diff --git a/tests/test_sse_replay.py b/tests/test_sse_replay.py new file mode 100644 index 00000000..a3759f52 --- /dev/null +++ b/tests/test_sse_replay.py @@ -0,0 +1,133 @@ +import json + +from fastapi.testclient import TestClient + +from web.app import app +from web.routers import sse_router + + +def _decode_sse_events(text: str): + events = [] + for frame in text.strip().split("\n\n"): + if not frame.startswith("data: "): + continue + events.append(json.loads(frame[len("data: "):])) + return events + + +def test_events_endpoint_replays_only_requested_cities(monkeypatch): + captured = {} + + class FakeStore: + def latest_revision(self): + return 44 + + def replay_events(self, *, cities, since_revision, limit): + captured["cities"] = cities + captured["since_revision"] = since_revision + captured["limit"] = limit + return [ + { + "type": "city_observation_patch.v1", + "revision": 43, + "city": "taipei", + "source": "cwa", + "obs_time": "2026-05-26T08:15:00Z", + "ts": 1780000000000, + "payload": {"temp": 31.2}, + } + ] + + def replay_requires_resync(self, *, cities, since_revision, replay_count, limit): + return False + + async def finite_stream( + user_id, + *, + cities=None, + replay_events=None, + connected_revision=0, + resync_event=None, + ): + yield sse_router.sse_manager._format_event( + {"type": "connected", "revision": connected_revision} + ) + for event in replay_events or []: + yield sse_router.sse_manager._format_event(event) + + monkeypatch.setattr(sse_router, "event_store", FakeStore()) + monkeypatch.setattr(sse_router.sse_manager, "event_stream", finite_stream) + + response = TestClient(app).get( + "/api/events?cities=taipei,hong%20kong&since_revision=42&replay_limit=25" + ) + + assert response.status_code == 200 + assert captured == { + "cities": {"taipei", "hong kong"}, + "since_revision": 42, + "limit": 25, + } + events = _decode_sse_events(response.text) + assert [event["type"] for event in events] == [ + "connected", + "city_observation_patch.v1", + ] + assert events[1]["city"] == "taipei" + + +def test_events_endpoint_emits_resync_when_replay_is_incomplete(monkeypatch): + class FakeStore: + def latest_revision(self): + return 99 + + def replay_events(self, *, cities, since_revision, limit): + return [ + { + "type": "city_observation_patch.v1", + "revision": 98, + "city": "taipei", + "source": "cwa", + "obs_time": "2026-05-26T08:15:00Z", + "ts": 1780000000000, + "payload": {"temp": 31.2}, + } + ] + + def replay_requires_resync(self, *, cities, since_revision, replay_count, limit): + return True + + async def finite_stream( + user_id, + *, + cities=None, + replay_events=None, + connected_revision=0, + resync_event=None, + ): + yield sse_router.sse_manager._format_event( + {"type": "connected", "revision": connected_revision} + ) + for event in replay_events or []: + yield sse_router.sse_manager._format_event(event) + if resync_event: + yield sse_router.sse_manager._format_event(resync_event) + + monkeypatch.setattr(sse_router, "event_store", FakeStore()) + monkeypatch.setattr(sse_router.sse_manager, "event_stream", finite_stream) + + response = TestClient(app).get( + "/api/events?cities=taipei&since_revision=1&replay_limit=1" + ) + + assert response.status_code == 200 + events = _decode_sse_events(response.text) + assert events[-1]["type"] == "resync_required" + assert events[-1]["reason"] == "replay_window_exceeded" + assert events[-1]["latest_revision"] == 99 + + +def test_replay_limit_is_bounded(): + assert sse_router._bounded_replay_limit(0) == 1 + assert sse_router._bounded_replay_limit(500) == 500 + assert sse_router._bounded_replay_limit(5000) == 2000 diff --git a/web/analysis_service.py b/web/analysis_service.py index 1ab385bb..14280b2e 100644 --- a/web/analysis_service.py +++ b/web/analysis_service.py @@ -146,11 +146,11 @@ def _is_plausible_city_temp(city: str, value: Any, unit: str = "°C") -> bool: temp = _sf(value) if temp is None: return False - meta = CITY_REGISTRY.get(str(city or "").strip().lower(), {}) or {} + meta = CITY_REGISTRY.get((city or "").strip().lower(), {}) or {} min_c = _sf(meta.get("min_plausible_metar_temp_c")) if min_c is None: return True - min_value = min_c * 9 / 5 + 32 if str(unit or "").upper().endswith("F") else min_c + min_value = min_c * 9 / 5 + 32 if (unit or "").upper().endswith("F") else min_c return temp >= min_value @@ -158,7 +158,7 @@ def _parse_local_hour(local_time_str: Optional[str]) -> Optional[int]: if not local_time_str: return None try: - parts = str(local_time_str).strip().split(":") + parts = local_time_str.strip().split(":") hour = int(parts[0]) if 0 <= hour <= 23: return hour @@ -204,7 +204,7 @@ def _record_analysis_cache_event(*, city: str, hit: bool, force_refresh: bool) - now = datetime.now(timezone.utc).isoformat() with _ANALYSIS_CACHE_STATS_LOCK: _ANALYSIS_CACHE_STATS["total_requests"] = int(_ANALYSIS_CACHE_STATS.get("total_requests") or 0) + 1 - _ANALYSIS_CACHE_STATS["last_city"] = str(city or "") + _ANALYSIS_CACHE_STATS["last_city"] = city or "" if force_refresh: _ANALYSIS_CACHE_STATS["force_refresh_requests"] = int(_ANALYSIS_CACHE_STATS.get("force_refresh_requests") or 0) + 1 if hit: @@ -243,7 +243,7 @@ def _analysis_ttl_for_city(city: str) -> int: def _analysis_cache_key(city: str, detail_mode: str = "full") -> str: - normalized_raw = str(detail_mode or "").strip().lower() + normalized_raw = (detail_mode or "").strip().lower() if normalized_raw == "panel": normalized_mode = "panel" elif normalized_raw == "market": @@ -666,7 +666,7 @@ def _analyze( """ # Check cache – skip when explicitly refreshing observations ttl = _analysis_ttl_for_city(city) - normalized_detail_mode_raw = str(detail_mode or "full").strip().lower() + normalized_detail_mode_raw = (detail_mode or "full").strip().lower() if normalized_detail_mode_raw == "panel": normalized_detail_mode = "panel" elif normalized_detail_mode_raw == "market": @@ -750,8 +750,9 @@ def _analyze( first_start = nws_periods[0].get("start_time") if first_start: maybe_dt = datetime.fromisoformat(str(first_start)) - if maybe_dt.utcoffset() is not None: - utc_offset = int(maybe_dt.utcoffset().total_seconds()) + offset_td = maybe_dt.utcoffset() + if offset_td is not None: + utc_offset = int(offset_td.total_seconds()) except Exception: utc_offset = None if utc_offset is None: @@ -770,7 +771,7 @@ def _analyze( metar_current_is_today = _metar_is_current_local_day( metar, local_date=local_date_str, - utc_offset=int(utc_offset or 0), + utc_offset=utc_offset, ) # ── 2. Current conditions (settlement > AMOS runway sensors > METAR > MGM > NMC fallback) ── @@ -878,12 +879,13 @@ def _analyze( if not obs_time_str and current_source == "amos": amos_obs_time = amos_data.get("observation_time") if amos_obs_time: - obs_time_str = _format_observation_time_local(amos_obs_time, int(utc_offset or 0)) + obs_time_str = _format_observation_time_local(amos_obs_time, utc_offset) + nmc_fallback = None if not obs_time_str and current_source == "nmc": nmc_fallback = _fetch_nmc_current_fallback(city, use_fahrenheit=is_f) obs_time_str = _format_observation_time_local( nmc_fallback.get("publish_time") or nmc_fallback.get("timestamp"), - int(utc_offset or 0), + utc_offset, ) current_obs_raw = obs_t @@ -972,9 +974,9 @@ def _analyze( if ( max_temp_time and max_so_far is not None - and str(max_temp_time) != str(obs_time_str) + and max_temp_time != obs_time_str ): - settlement_today_obs.append({"time": str(max_temp_time), "temp": max_so_far}) + settlement_today_obs.append({"time": max_temp_time, "temp": max_so_far}) metar_today_obs_payload = [ {"time": t, "temp": v} @@ -1044,7 +1046,7 @@ def _analyze( else cur_temp ) if fallback_high is not None: - om_today = float(fallback_high) + om_today = fallback_high if not forecast_daily: forecast_daily = [{"date": local_date_str, "max_temp": om_today}] sunrise = ( @@ -1065,7 +1067,9 @@ def _analyze( current_forecasts["Open-Meteo"] = om_today for m, v in mm.get("forecasts", {}).items(): if v is not None and not _is_excluded_model_name(m): - current_forecasts[m] = _sf(v) + temp_val = _sf(v) + if temp_val is not None: + current_forecasts[m] = temp_val nws_high = _sf(raw.get("nws", {}).get("today_high")) if nws_high is not None: current_forecasts["NWS"] = nws_high @@ -1244,8 +1248,8 @@ def _analyze( # between the current observed temperature and the model's hourly path. # Uses cur_temp / max_so_far already resolved at lines 1052-1095 above. _local_hour = _parse_local_hour(local_time_str) - peak_first = int(first_peak_h or 14) - peak_last_h = int(last_peak_h or 17) + peak_first = first_peak_h or 14 + peak_last_h = last_peak_h or 17 if ( deb_val is not None @@ -1377,7 +1381,7 @@ def _analyze( taf if isinstance(taf, dict) else {}, city, local_date_str, - int(utc_offset or 0), + utc_offset, first_peak_h, last_peak_h, ) @@ -1481,6 +1485,8 @@ def _analyze( multi_model_daily = {} mm_daily_raw = mm.get("daily_forecasts", {}) for i, d_str in enumerate(dates): + d_probs = [] + d_probs_all = [] if i == 0: day_m = current_forecasts.copy() d_val, d_winfo = deb_val, deb_weights @@ -1534,7 +1540,7 @@ def _analyze( # ── Assemble result ── runway_plate_history = {} icao = risk.get("icao", "") - if icao: + if isinstance(icao, str) and icao: try: from src.database.db_manager import DBManager raw_runway_obs = DBManager().get_runway_obs_recent(icao, minutes=36 * 60) @@ -1581,7 +1587,7 @@ def _analyze( "display_name": str(city_meta.get("display_name") or city_meta.get("name") or city.title()), "lat": lat, "lon": lon, - "utc_offset_seconds": int(utc_offset or 0), + "utc_offset_seconds": utc_offset, "temp_symbol": sym, "local_time": local_time_str, "local_date": local_date_str, @@ -1779,7 +1785,7 @@ def _analyze_summary(city: str, force_refresh: bool = False) -> Dict[str, Any]: except Exception: return None - jobs = { + jobs: Dict[str, Any] = { "settlement_current": lambda: _weather.fetch_settlement_current(city) or {}, "open_meteo": lambda: _weather.fetch_from_open_meteo(lat, lon, use_fahrenheit=is_f) or {}, "multi_model": lambda: _weather.fetch_multi_model(lat, lon, city=city, use_fahrenheit=is_f) or {}, @@ -1832,7 +1838,7 @@ def _analyze_summary(city: str, force_refresh: bool = False) -> Dict[str, Any]: metar_current_is_today = _metar_is_current_local_day( metar, local_date=local_date_str, - utc_offset=int(utc_offset or 0), + utc_offset=utc_offset, ) sc_cur = settlement_current.get("current") or {} @@ -1922,13 +1928,13 @@ def _analyze_summary(city: str, force_refresh: bool = False) -> Dict[str, Any]: (datetime.now(timezone.utc) - dt.astimezone(timezone.utc)).total_seconds() / 60 ) except Exception: - obs_time_str = str(obs_t)[:16] + obs_time_str = obs_t[:16] if not obs_time_str and current_source == "nmc": if not nmc_fallback: nmc_fallback = _fetch_nmc_current_fallback(city, use_fahrenheit=is_f) obs_time_str = _format_observation_time_local( nmc_fallback.get("publish_time") or nmc_fallback.get("timestamp"), - int(utc_offset or 0), + utc_offset, ) om_daily = (open_meteo.get("daily") or {}) if isinstance(open_meteo, dict) else {} @@ -1965,14 +1971,16 @@ def _analyze_summary(city: str, force_refresh: bool = False) -> Dict[str, Any]: else cur_temp ) if fallback_high is not None: - om_today = float(fallback_high) + om_today = fallback_high current_forecasts: Dict[str, float] = {} if om_today is not None: current_forecasts["Open-Meteo"] = om_today for m, v in mm.get("forecasts", {}).items(): if v is not None and not _is_excluded_model_name(m): - current_forecasts[m] = _sf(v) + temp_val = _sf(v) + if temp_val is not None: + current_forecasts[m] = temp_val if nws_high is not None: current_forecasts["NWS"] = nws_high if mgm_high is not None: @@ -1980,7 +1988,9 @@ def _analyze_summary(city: str, force_refresh: bool = False) -> Dict[str, Any]: elif mgm_hourly_high is not None: current_forecasts["MGM Hourly"] = mgm_hourly_high if hko_forecast is not None: - current_forecasts["HKO"] = _sf(hko_forecast) + temp_hko = _sf(hko_forecast) + if temp_hko is not None: + current_forecasts["HKO"] = temp_hko current_forecasts = { model_name: value for model_name, value in current_forecasts.items() @@ -2011,8 +2021,8 @@ def _analyze_summary(city: str, force_refresh: bool = False) -> Dict[str, Any]: settlement_today_obs.append({"time": raw_time, "temp": raw_temp}) if not settlement_today_obs and obs_time_str and cur_temp is not None: settlement_today_obs.append({"time": obs_time_str, "temp": cur_temp}) - if max_temp_time and max_so_far is not None and str(max_temp_time) != str(obs_time_str): - settlement_today_obs.append({"time": str(max_temp_time), "temp": max_so_far}) + if max_temp_time and max_so_far is not None and max_temp_time != obs_time_str: + settlement_today_obs.append({"time": max_temp_time, "temp": max_so_far}) metar_today_obs_payload = [ {"time": obs_time, "temp": obs_temp} @@ -2042,7 +2052,7 @@ def _analyze_summary(city: str, force_refresh: bool = False) -> Dict[str, Any]: "name": city, "display_name": str(city_meta.get("display_name") or city_meta.get("name") or city.title()), "temp_symbol": sym, - "utc_offset_seconds": int(utc_offset or 0), + "utc_offset_seconds": utc_offset, "local_time": local_time_str, "local_date": local_date_str, "risk": { diff --git a/web/realtime_event_store.py b/web/realtime_event_store.py new file mode 100644 index 00000000..621f428b --- /dev/null +++ b/web/realtime_event_store.py @@ -0,0 +1,251 @@ +"""SQLite-backed replay log for realtime observation SSE patches.""" + +from __future__ import annotations + +import json +import os +import sqlite3 +from datetime import datetime, timedelta, timezone +from typing import Any, Dict, Optional, Set + +from src.database.db_manager import DBManager +from web.realtime_patch_schema import EVENT_TYPE + + +DEFAULT_RETENTION_HOURS = 6 +MAX_REPLAY_LIMIT = 2000 + + +def _utc_now() -> datetime: + return datetime.now(timezone.utc) + + +def _iso(dt: datetime) -> str: + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt.astimezone(timezone.utc).isoformat() + + +def _created_at_to_ms(value: str) -> int: + raw = str(value or "").strip() + if raw.endswith("Z"): + raw = f"{raw[:-1]}+00:00" + try: + dt = datetime.fromisoformat(raw) + except ValueError: + return int(_utc_now().timestamp() * 1000) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return int(dt.astimezone(timezone.utc).timestamp() * 1000) + + +def _normalize_city_set(cities: Optional[Set[str]]) -> Set[str]: + return {str(city or "").strip().lower() for city in (cities or set()) if str(city or "").strip()} + + +def _retention_hours_from_env() -> int: + raw = os.getenv("POLYWEATHER_PATCH_EVENT_RETENTION_HOURS", "").strip() + if not raw: + return DEFAULT_RETENTION_HOURS + try: + return max(1, int(float(raw))) + except ValueError: + return DEFAULT_RETENTION_HOURS + + +class RealtimeEventStore: + def __init__(self, db_path: Optional[str] = None) -> None: + self._db = DBManager(db_path) + self.db_path = self._db.db_path + self._ensure_table() + + def _connect(self) -> sqlite3.Connection: + conn = sqlite3.connect(self.db_path, timeout=10) + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA busy_timeout=5000") + return conn + + def _ensure_table(self) -> None: + db_dir = os.path.dirname(self.db_path) + if db_dir: + os.makedirs(db_dir, exist_ok=True) + with self._connect() as conn: + conn.execute( + """ + CREATE TABLE IF NOT EXISTS observation_patch_events ( + revision INTEGER PRIMARY KEY AUTOINCREMENT, + schema_type TEXT NOT NULL, + schema_version INTEGER NOT NULL, + city TEXT NOT NULL, + source TEXT NOT NULL, + obs_time TEXT, + payload_json TEXT NOT NULL, + created_at TEXT NOT NULL + ) + """ + ) + conn.execute( + """ + CREATE INDEX IF NOT EXISTS idx_observation_patch_events_city_revision + ON observation_patch_events(city, revision) + """ + ) + conn.execute( + """ + CREATE INDEX IF NOT EXISTS idx_observation_patch_events_created_at + ON observation_patch_events(created_at) + """ + ) + conn.commit() + + def append_event(self, event: Dict[str, Any]) -> Dict[str, Any]: + if event.get("type") != EVENT_TYPE: + raise ValueError("unsupported realtime event type") + payload = event.get("payload") + if not isinstance(payload, dict): + raise ValueError("event payload must be an object") + + created_at = _iso(_utc_now()) + with self._connect() as conn: + cursor = conn.execute( + """ + INSERT INTO observation_patch_events ( + schema_type, schema_version, city, source, obs_time, payload_json, created_at + ) + VALUES (?, ?, ?, ?, ?, ?, ?) + """, + ( + str(event["schema_type"]), + int(event["schema_version"]), + str(event["city"]), + str(event["source"]), + event.get("obs_time"), + json.dumps(payload, ensure_ascii=False, separators=(",", ":")), + created_at, + ), + ) + revision = int(cursor.lastrowid) + conn.commit() + + stored = { + "type": event["type"], + "revision": revision, + "city": str(event["city"]), + "source": str(event["source"]), + "obs_time": event.get("obs_time"), + "ts": int(event.get("ts") or _created_at_to_ms(created_at)), + "payload": payload, + } + self.cleanup_old_events(retention_hours=_retention_hours_from_env()) + return stored + + def latest_revision(self) -> int: + with self._connect() as conn: + row = conn.execute( + "SELECT COALESCE(MAX(revision), 0) FROM observation_patch_events" + ).fetchone() + return int(row[0] or 0) if row else 0 + + def replay_events( + self, + *, + cities: Optional[Set[str]], + since_revision: int, + limit: int, + ) -> list[Dict[str, Any]]: + city_set = _normalize_city_set(cities) + bounded_limit = max(1, min(MAX_REPLAY_LIMIT, int(limit or 1))) + params: list[Any] = [max(0, int(since_revision or 0))] + where = "revision > ?" + if city_set: + placeholders = ",".join("?" for _ in city_set) + where += f" AND city IN ({placeholders})" + params.extend(sorted(city_set)) + params.append(bounded_limit) + + with self._connect() as conn: + conn.row_factory = sqlite3.Row + rows = conn.execute( + f""" + SELECT revision, schema_type, schema_version, city, source, obs_time, + payload_json, created_at + FROM observation_patch_events + WHERE {where} + ORDER BY revision ASC + LIMIT ? + """, + params, + ).fetchall() + return [self._row_to_event(row) for row in rows] + + def replay_requires_resync( + self, + *, + cities: Optional[Set[str]], + since_revision: int, + replay_count: int, + limit: int, + ) -> bool: + city_set = _normalize_city_set(cities) + since = max(0, int(since_revision or 0)) + where_parts = [] + params: list[Any] = [] + if city_set: + placeholders = ",".join("?" for _ in city_set) + where_parts.append(f"city IN ({placeholders})") + params.extend(sorted(city_set)) + where = f"WHERE {' AND '.join(where_parts)}" if where_parts else "" + + with self._connect() as conn: + min_row = conn.execute( + f"SELECT MIN(revision) FROM observation_patch_events {where}", + params, + ).fetchone() + min_revision = int(min_row[0] or 0) if min_row else 0 + if min_revision and since > 0 and since < min_revision - 1: + return True + + if replay_count < max(1, int(limit or 1)): + return False + + count_params = [since, *params] + count_where = "revision > ?" + if where_parts: + count_where += " AND " + " AND ".join(where_parts) + count_row = conn.execute( + f"SELECT COUNT(1) FROM observation_patch_events WHERE {count_where}", + count_params, + ).fetchone() + return int(count_row[0] or 0) > int(limit or 1) + + def cleanup_old_events( + self, + *, + retention_hours: Optional[int] = None, + now: Optional[datetime] = None, + ) -> int: + hours = max(1, int(retention_hours or _retention_hours_from_env())) + cutoff = _iso((now or _utc_now()) - timedelta(hours=hours)) + with self._connect() as conn: + cursor = conn.execute( + "DELETE FROM observation_patch_events WHERE created_at < ?", + (cutoff,), + ) + deleted = int(cursor.rowcount or 0) + conn.commit() + return deleted + + @staticmethod + def _row_to_event(row: sqlite3.Row) -> Dict[str, Any]: + payload = json.loads(row["payload_json"]) + schema_type = str(row["schema_type"]) + schema_version = int(row["schema_version"]) + return { + "type": f"{schema_type}.v{schema_version}", + "revision": int(row["revision"]), + "city": str(row["city"]), + "source": str(row["source"]), + "obs_time": row["obs_time"], + "ts": _created_at_to_ms(row["created_at"]), + "payload": payload, + } diff --git a/web/realtime_patch_schema.py b/web/realtime_patch_schema.py new file mode 100644 index 00000000..a9e9a1b6 --- /dev/null +++ b/web/realtime_patch_schema.py @@ -0,0 +1,200 @@ +"""Versioned realtime observation patch normalization.""" + +from __future__ import annotations + +import time +from typing import Any, Dict, Iterable, List, Optional + + +SCHEMA_TYPE = "city_observation_patch" +SCHEMA_VERSION = 1 +EVENT_TYPE = "city_observation_patch.v1" + + +class PatchValidationError(ValueError): + """Raised when a collector patch cannot become a replayable observation event.""" + + +def _normalize_city(value: Any) -> str: + return str(value or "").strip().lower() + + +def _normalize_source(value: Any) -> str: + source = str(value or "").strip().lower() + return source or "weather" + + +def _finite_number(value: Any) -> Optional[float]: + try: + number = float(value) + except (TypeError, ValueError): + return None + if number != number or number in {float("inf"), float("-inf")}: + return None + return round(number, 2) + + +def _first_number(*values: Any) -> Optional[float]: + for value in values: + number = _finite_number(value) + if number is not None: + return number + return None + + +def _iter_runway_points(raw_points: Any) -> Iterable[Dict[str, Any]]: + if isinstance(raw_points, list): + for item in raw_points: + if isinstance(item, dict): + yield item + + +def _normalize_runway_points(raw_points: Any) -> List[Dict[str, Any]]: + points: List[Dict[str, Any]] = [] + for raw in _iter_runway_points(raw_points): + runway = str(raw.get("runway") or raw.get("rwy") or "").strip().upper() + temp = _first_number( + raw.get("temp"), + raw.get("target_runway_max"), + raw.get("tdz_temp"), + raw.get("mid_temp"), + raw.get("end_temp"), + ) + if not runway and temp is None: + continue + + point: Dict[str, Any] = {} + if runway: + point["runway"] = runway + if temp is not None: + point["temp"] = temp + for key in ("tdz_temp", "mid_temp", "end_temp", "target_runway_max"): + value = _finite_number(raw.get(key)) + if value is not None: + point[key] = value + if isinstance(raw.get("is_settlement"), bool): + point["is_settlement"] = raw["is_settlement"] + if point: + points.append(point) + return points + + +def _legacy_changes(patch: Dict[str, Any]) -> Dict[str, Any]: + changes = patch.get("changes") + return changes if isinstance(changes, dict) else {} + + +def _payload_from_legacy(changes: Dict[str, Any]) -> Dict[str, Any]: + amos = changes.get("amos") if isinstance(changes.get("amos"), dict) else {} + runway_obs = amos.get("runway_obs") if isinstance(amos.get("runway_obs"), dict) else {} + + payload: Dict[str, Any] = {} + temp = _finite_number(changes.get("temp")) + if temp is not None: + payload["temp"] = temp + max_so_far = _first_number(changes.get("max_so_far"), changes.get("current_max_so_far")) + if max_so_far is not None: + payload["max_so_far"] = max_so_far + + station_code = str( + changes.get("station_code") + or changes.get("icao") + or amos.get("icao") + or "" + ).strip().upper() + if station_code: + payload["station_code"] = station_code + + station_label = str( + changes.get("station_label") + or amos.get("station_label") + or amos.get("station_name") + or "" + ).strip() + if station_label: + payload["station_label"] = station_label + + series_key = str(changes.get("series_key") or "").strip() + if series_key: + payload["series_key"] = series_key + payload["unit"] = str(changes.get("unit") or "celsius").strip().lower() or "celsius" + + raw_runway_points = changes.get("runway_points") + if raw_runway_points is None: + raw_runway_points = runway_obs.get("point_temperatures") + runway_points = _normalize_runway_points(raw_runway_points) + if runway_points: + payload["runway_points"] = runway_points + + hourly = changes.get("hourly") + if isinstance(hourly, dict): + payload["hourly"] = hourly + + return payload + + +def _payload_from_v1(raw_payload: Any) -> Dict[str, Any]: + if not isinstance(raw_payload, dict): + return {} + + payload: Dict[str, Any] = {} + temp = _finite_number(raw_payload.get("temp")) + if temp is not None: + payload["temp"] = temp + max_so_far = _finite_number(raw_payload.get("max_so_far")) + if max_so_far is not None: + payload["max_so_far"] = max_so_far + + for key in ("station_code", "station_label", "series_key", "unit"): + value = raw_payload.get(key) + if isinstance(value, str) and value.strip(): + payload[key] = value.strip() + if "unit" not in payload: + payload["unit"] = "celsius" + + runway_points = _normalize_runway_points(raw_payload.get("runway_points")) + if runway_points: + payload["runway_points"] = runway_points + if isinstance(raw_payload.get("hourly"), dict): + payload["hourly"] = raw_payload["hourly"] + return payload + + +def _has_observation(payload: Dict[str, Any]) -> bool: + return any( + key in payload + for key in ("temp", "max_so_far", "runway_points", "hourly") + ) + + +def normalize_observation_patch(patch: Dict[str, Any]) -> Dict[str, Any]: + if not isinstance(patch, dict): + raise PatchValidationError("patch must be an object") + + if patch.get("type") == EVENT_TYPE: + city = _normalize_city(patch.get("city")) + source = _normalize_source(patch.get("source")) + obs_time = str(patch.get("obs_time") or "").strip() or None + payload = _payload_from_v1(patch.get("payload")) + else: + changes = _legacy_changes(patch) + city = _normalize_city(patch.get("city")) + source = _normalize_source(changes.get("source") or patch.get("source")) + obs_time = str(changes.get("obs_time") or patch.get("obs_time") or "").strip() or None + payload = _payload_from_legacy(changes) + + if not city: + raise PatchValidationError("city is required") + if not _has_observation(payload): + raise PatchValidationError("patch must include temperature, max, runway, or hourly data") + + return { + "type": EVENT_TYPE, + "schema_type": SCHEMA_TYPE, + "schema_version": SCHEMA_VERSION, + "city": city, + "source": source, + "obs_time": obs_time, + "ts": int(time.time() * 1000), + "payload": payload, + } diff --git a/web/routers/sse_router.py b/web/routers/sse_router.py index 8ffa2ff9..d5d2ccc8 100644 --- a/web/routers/sse_router.py +++ b/web/routers/sse_router.py @@ -2,15 +2,35 @@ from __future__ import annotations -from typing import Any +import time +from typing import Any, Optional, Set -from fastapi import APIRouter, HTTPException, Request +from fastapi import APIRouter, HTTPException, Query, Request from fastapi.responses import StreamingResponse +from web.realtime_event_store import RealtimeEventStore, MAX_REPLAY_LIMIT +from web.realtime_patch_schema import PatchValidationError, normalize_observation_patch from web.sse_manager import sse_manager router = APIRouter(tags=["events"]) +event_store = RealtimeEventStore() + + +def _parse_cities_param(cities: str) -> Set[str]: + return { + item.strip().lower() + for item in str(cities or "").split(",") + if item.strip() + } + + +def _bounded_replay_limit(value: int) -> int: + try: + limit = int(value) + except (TypeError, ValueError): + limit = 500 + return max(1, min(MAX_REPLAY_LIMIT, limit)) @router.options("/api/events") @@ -19,12 +39,56 @@ async def sse_events_preflight(request: Request): @router.get("/api/events") -async def sse_events(request: Request): +async def sse_events( + request: Request, + cities: str = "", + since_revision: Optional[int] = Query(default=None), + replay_limit: int = Query(default=500), +): user_id = getattr(request.state, "auth_user_id", None) or "anon" origin = request.headers.get("origin", "") allowed = origin in {"https://polyweather.top", "https://www.polyweather.top", "http://localhost:3000"} + city_set = _parse_cities_param(cities) + limit = _bounded_replay_limit(replay_limit) + latest_revision = event_store.latest_revision() + replay_events = [] + resync_event = None + + if since_revision is not None: + try: + replay_events = event_store.replay_events( + cities=city_set, + since_revision=max(0, int(since_revision)), + limit=limit, + ) + if event_store.replay_requires_resync( + cities=city_set, + since_revision=max(0, int(since_revision)), + replay_count=len(replay_events), + limit=limit, + ): + resync_event = { + "type": "resync_required", + "reason": "replay_window_exceeded", + "latest_revision": latest_revision, + "ts": int(time.time() * 1000), + } + except Exception: + resync_event = { + "type": "resync_required", + "reason": "replay_failed", + "latest_revision": latest_revision, + "ts": int(time.time() * 1000), + } + return StreamingResponse( - sse_manager.event_stream(user_id), + sse_manager.event_stream( + user_id, + cities=city_set, + replay_events=replay_events, + connected_revision=latest_revision, + resync_event=resync_event, + ), media_type="text/event-stream", headers={ "Cache-Control": "no-cache, no-transform", @@ -38,11 +102,15 @@ async def sse_events(request: Request): @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) + try: + normalized = normalize_observation_patch(patch) + except PatchValidationError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + try: + event = event_store.append_event(normalized) + except Exception as exc: + raise HTTPException(status_code=500, detail="event log write failed") from exc + + sse_manager.broadcast_event(event) return {"ok": True, "revision": event["revision"]} diff --git a/web/services/city_payloads.py b/web/services/city_payloads.py index 44cabcf7..f2018590 100644 --- a/web/services/city_payloads.py +++ b/web/services/city_payloads.py @@ -251,8 +251,8 @@ def build_city_detail_payload( or _build_intraday_meteorology(data), "vertical_profile_signal": data.get("vertical_profile_signal") or {}, "taf": data.get("taf") or {}, - "runway_plate_history": aggregate_runway_history(data.get("runway_plate_history") or {}, resolution), - "runway_band_history": build_runway_band_history(data.get("runway_plate_history") or {}, resolution), + "runway_plate_history": aggregate_runway_history(data.get("runway_plate_history") or {}, resolution or "10m"), + "runway_band_history": build_runway_band_history(data.get("runway_plate_history") or {}, resolution or "10m"), "risk": data.get("risk"), "settlement_station": data.get("settlement_station") or {}, diff --git a/web/sse_manager.py b/web/sse_manager.py index 8876f9ec..475f71ef 100644 --- a/web/sse_manager.py +++ b/web/sse_manager.py @@ -7,7 +7,7 @@ import json import threading import time from collections import defaultdict -from typing import Any, AsyncIterator, DefaultDict +from typing import Any, AsyncIterator, DefaultDict, Iterable, Optional, Set HEARTBEAT_INTERVAL_SECONDS = 30 @@ -17,6 +17,7 @@ QUEUE_MAXSIZE = 256 class SseManager: def __init__(self) -> None: self._queues: DefaultDict[str, set[asyncio.Queue[dict[str, Any]]]] = defaultdict(set) + self._queue_cities: dict[int, frozenset[str]] = {} self._lock = threading.RLock() self._revision = 0 @@ -25,21 +26,57 @@ class SseManager: self._revision += 1 return self._revision + @staticmethod + def _normalize_city(value: Any) -> str: + return str(value or "").strip().lower() + + @classmethod + def _normalize_city_set(cls, cities: Optional[Iterable[str]]) -> Set[str]: + return { + cls._normalize_city(city) + for city in (cities or []) + if cls._normalize_city(city) + } + + def _track_revision(self, event: dict[str, Any]) -> None: + try: + revision = int(event.get("revision") or 0) + except (TypeError, ValueError): + return + if revision <= 0: + return + with self._lock: + if revision > self._revision: + self._revision = revision + def broadcast(self, city: str, changes: dict[str, Any]) -> dict[str, Any]: event = { "type": "city_patch", - "city": str(city or "").strip().lower(), + "city": self._normalize_city(city), "changes": changes or {}, "revision": self._next_revision(), "ts": int(time.time() * 1000), } - if not event["city"]: + return self.broadcast_event(event) + + def broadcast_event(self, event: dict[str, Any]) -> dict[str, Any]: + city = self._normalize_city(event.get("city")) + if city: + event = {**event, "city": city} + self._track_revision(event) + if not city: return event with self._lock: - queues = [queue for queue_set in self._queues.values() for queue in queue_set] + queue_items = [ + (queue, self._queue_cities.get(id(queue), frozenset())) + for queue_set in self._queues.values() + for queue in queue_set + ] - for queue in queues: + for queue, subscribed_cities in queue_items: + if subscribed_cities and city not in subscribed_cities: + continue try: queue.put_nowait(event) except asyncio.QueueFull: @@ -53,18 +90,37 @@ class SseManager: pass return event - async def event_stream(self, user_id: str) -> AsyncIterator[str]: + async def event_stream( + self, + user_id: str, + *, + cities: Optional[Iterable[str]] = None, + replay_events: Optional[Iterable[dict[str, Any]]] = None, + connected_revision: Optional[int] = None, + resync_event: Optional[dict[str, Any]] = None, + ) -> AsyncIterator[str]: user_key = str(user_id or "anon") + city_set = frozenset(self._normalize_city_set(cities)) queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue(maxsize=QUEUE_MAXSIZE) with self._lock: self._queues[user_key].add(queue) + self._queue_cities[id(queue)] = city_set + if connected_revision is not None: + self._revision = max(self._revision, int(connected_revision or 0)) try: yield self._format_event({ "type": "connected", "revision": self._revision, + "cities": sorted(city_set), "ts": int(time.time() * 1000), }) + for event in replay_events or []: + self._track_revision(event) + yield self._format_event(event) + if resync_event: + self._track_revision({"revision": resync_event.get("latest_revision")}) + yield self._format_event(resync_event) while True: try: event = await asyncio.wait_for( @@ -81,6 +137,7 @@ class SseManager: finally: with self._lock: self._queues[user_key].discard(queue) + self._queue_cities.pop(id(queue), None) if not self._queues[user_key]: self._queues.pop(user_key, None)