Compare commits

...

29 Commits

Author SHA1 Message Date
2569718930@qq.com 49cb0bb42d Prevent terminal auth spinner deadlock 2026-06-01 12:36:29 +08:00
2569718930@qq.com fb6c354317 Avoid terminal full-detail fallback overload 2026-06-01 12:06:46 +08:00
2569718930@qq.com fee402145e Slim terminal chart detail batches 2026-06-01 11:37:16 +08:00
2569718930@qq.com d1633331bc Load terminal chart canvas eagerly 2026-06-01 07:05:08 +08:00
2569718930@qq.com 6e00160897 Version scan terminal cache key 2026-06-01 06:46:34 +08:00
2569718930@qq.com 6bc53c6e5d Slim deferred scan runway history 2026-06-01 06:25:27 +08:00
2569718930@qq.com 7625a99020 Cache terminal city registry fallback 2026-06-01 06:09:44 +08:00
2569718930@qq.com 3447a043ff Preload terminal chart canvas 2026-06-01 05:56:09 +08:00
2569718930@qq.com 10cf557b11 Batch deferred chart detail loads 2026-06-01 02:56:13 +08:00
2569718930@qq.com eaf17c7a32 Align scan prewarm observability test 2026-06-01 02:35:23 +08:00
2569718930@qq.com 14526d6ce8 Prewarm default scan terminal payload 2026-06-01 02:06:40 +08:00
2569718930@qq.com 9a0d7ae444 Persist scan terminal cache in Redis 2026-06-01 01:53:54 +08:00
2569718930@qq.com de5083ef18 Defer secondary chart detail loads 2026-06-01 01:27:53 +08:00
2569718930@qq.com 2424b68532 Compact scan runway history payload 2026-06-01 01:03:33 +08:00
2569718930@qq.com ff2a3da100 Use longer timeout for scan prewarm 2026-06-01 00:30:25 +08:00
2569718930@qq.com 7040598e26 Prewarm terminal scan payload 2026-06-01 00:14:12 +08:00
2569718930@qq.com 22f4ac1f26 Preserve scan cache on timeout 2026-05-31 23:54:35 +08:00
2569718930@qq.com 3e5ff4ea8f Isolate scan prewarm to web service 2026-05-31 23:31:32 +08:00
2569718930@qq.com 39c7c1903c Shorten cold scan terminal timeout 2026-05-31 22:54:43 +08:00
2569718930@qq.com eecaa48ec2 Limit city detail fallback to batch failures 2026-05-31 22:34:28 +08:00
2569718930@qq.com d911af1225 Reduce duplicate terminal detail loads 2026-05-31 22:14:17 +08:00
2569718930@qq.com 38c6cefb27 Direct resync stale SSE clients 2026-05-31 21:48:09 +08:00
2569718930@qq.com a3df2e00dc Clamp SSE replay window on backend 2026-05-31 21:33:35 +08:00
2569718930@qq.com fde69123cd Merge branch 'codex/signup-trial-reconcile' 2026-05-31 21:25:58 +08:00
2569718930@qq.com 2d963c16f3 Coalesce terminal chart detail loads 2026-05-31 21:17:38 +08:00
2569718930@qq.com 071d749ef0 Reduce terminal detail refresh traffic 2026-05-31 21:01:11 +08:00
2569718930@qq.com 8d552a9279 Scale SSE replay window by visible cities 2026-05-31 20:33:04 +08:00
2569718930@qq.com 9c5a08dc1e Reduce false signup trial risk alerts 2026-05-31 20:21:57 +08:00
2569718930@qq.com ada5f274d3 Debounce terminal SSE subscription reconnects 2026-05-31 20:15:39 +08:00
39 changed files with 2353 additions and 268 deletions
+6
View File
@@ -28,6 +28,9 @@ services:
env_file: &id001
- .env
environment:
POLYWEATHER_REDIS_URL: ${POLYWEATHER_REDIS_URL:-redis://polyweather_redis:6379/0}
POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false'
POLYWEATHER_SERVICE_ROLE: bot
TELEGRAM_AIRPORT_PUSH_INTERVAL_SEC: ${POLYWEATHER_BOT_AIRPORT_PUSH_INTERVAL_SEC:-180}
TELEGRAM_AIRPORT_PUSH_MAX_WORKERS: ${POLYWEATHER_BOT_AIRPORT_PUSH_MAX_WORKERS:-1}
healthcheck:
@@ -93,6 +96,9 @@ services:
max-file: "3"
container_name: polyweather_web
env_file: *id001
environment:
POLYWEATHER_REDIS_URL: ${POLYWEATHER_REDIS_URL:-redis://polyweather_redis:6379/0}
POLYWEATHER_SERVICE_ROLE: web
healthcheck:
interval: 30s
retries: 3
+1 -1
View File
@@ -192,7 +192,7 @@ Ops
- 支付相关路由:`no-store`
- 当 detail 缓存只返回单模型或单日 forecast 时,前端会自动强刷完整 detail,并在补齐前显示同步提示 / 占位卡
- 今日日内分析打开时如果正在切换城市、日期或 detail 深度,弹窗会阻断旧内容点击并显示刷新锁
- 终端图表订阅 `/api/events?cities=...&since_revision=...&replay_limit=500`,接收 `city_observation_patch.v1`;无 patch 超过 2 分钟时,可见图表才触发 60 秒兜底刷新
- 终端图表订阅 `/api/events?cities=...&since_revision=...&replay_limit=按可见城市数动态限制`,接收 `city_observation_patch.v1`;无 patch 超过 2 分钟时,可见图表才触发 60 秒兜底刷新
- 前端只消费 HTTP snapshot + SSE patch,不直接感知 RedisRedis Stream / SQLite event log 都由后端统一封装
## Vercel 节流建议
@@ -31,7 +31,7 @@ export async function GET(req: NextRequest) {
force_refresh: forceRefresh,
limit: req.nextUrl.searchParams.get("limit") || "12",
});
for (const key of ["market_slug", "target_date", "resolution"]) {
for (const key of ["market_slug", "target_date", "resolution", "scope"]) {
const value = req.nextUrl.searchParams.get(key);
if (value) searchParams.set(key, value);
}
+1 -1
View File
@@ -9,7 +9,7 @@ import {
const API_BASE = process.env.POLYWEATHER_API_BASE_URL;
const SCAN_TERMINAL_PROXY_TIMEOUT_MS = Number(
process.env.POLYWEATHER_SCAN_TERMINAL_PROXY_TIMEOUT_MS || "40000",
process.env.POLYWEATHER_SCAN_TERMINAL_PROXY_TIMEOUT_MS || "18000",
);
export const maxDuration = 45;
@@ -46,7 +46,11 @@ import { scanRootClass } from "@/components/dashboard/scan-root-styles";
import { useRelativeTime } from "@/hooks/useRelativeTime";
import { Panel } from "@/components/dashboard/scan-terminal/Panel";
import { UsageGuideDashboard } from "@/components/dashboard/scan-terminal/UsageGuideDashboard";
import { LiveTemperatureThresholdChart, clearCityDetailCache } from "@/components/dashboard/scan-terminal/LiveTemperatureThresholdChart";
import {
LiveTemperatureThresholdChart,
clearCityDetailCache,
preloadTemperatureChartCanvas,
} from "@/components/dashboard/scan-terminal/LiveTemperatureThresholdChart";
import { KoyfinRowsTable } from "@/components/dashboard/scan-terminal/KoyfinRowsTable";
import { rowName, pct, money, temp, edgeClass } from "@/components/dashboard/scan-terminal/utils";
import { CitySelectorDropdown } from "@/components/dashboard/scan-terminal/CitySelectorDropdown";
@@ -62,6 +66,8 @@ import {
import {
cityListItemsToScanRows,
mergeScanRowsWithCityFallbackRows,
readCachedCityList,
writeCachedCityList,
} from "@/components/dashboard/scan-terminal/city-fallback-rows";
import { markAnalyticsOnce, trackAppEvent } from "@/lib/app-analytics";
import { STATIC_CITY_LIST } from "@/lib/static-cities";
@@ -81,6 +87,10 @@ const TrainingDashboard = dynamic(
);
const ONLINE_USERS_REFRESH_MS = 5 * 60_000;
const AUTH_PROFILE_REQUEST_TIMEOUT_MS = 4500;
const AUTH_DECISION_RECOVERY_MS = 10_000;
const ACTIVE_ACCESS_CACHE_KEY = "polyweather_terminal_active_access_v1";
const ACTIVE_ACCESS_CACHE_TTL_MS = 6 * 60 * 60 * 1000;
function createEmptyAccess(loading = true): ProAccessState {
return {
@@ -112,6 +122,67 @@ function createLocalAccess(): ProAccessState {
};
}
function isFutureAccessExpiry(value: string | null | undefined, now = Date.now()) {
if (!value) return true;
const ts = Date.parse(value);
return Number.isFinite(ts) && ts > now;
}
function readCachedActiveAccess(now = Date.now()): ProAccessState | null {
if (typeof window === "undefined") return null;
try {
const raw = window.localStorage.getItem(ACTIVE_ACCESS_CACHE_KEY);
if (!raw) return null;
const cached = JSON.parse(raw);
const access = cached?.access as Partial<ProAccessState> | undefined;
const ts = Number(cached?.ts || 0);
if (!access?.authenticated || !access.subscriptionActive) return null;
if (!ts || now - ts < 0 || now - ts > ACTIVE_ACCESS_CACHE_TTL_MS) return null;
if (!isFutureAccessExpiry(access.subscriptionTotalExpiresAt || access.subscriptionExpiresAt, now)) {
return null;
}
return {
loading: false,
authenticated: true,
userId: access.userId ?? null,
subscriptionActive: true,
subscriptionPlanCode: access.subscriptionPlanCode ?? null,
subscriptionExpiresAt: access.subscriptionExpiresAt ?? null,
subscriptionTotalExpiresAt:
access.subscriptionTotalExpiresAt ?? access.subscriptionExpiresAt ?? null,
subscriptionQueuedDays: Number(access.subscriptionQueuedDays ?? 0),
points: Number(access.points ?? 0),
error: null,
};
} catch {
return null;
}
}
function writeCachedActiveAccess(access: ProAccessState) {
if (typeof window === "undefined") return;
if (!access.authenticated || !access.subscriptionActive) return;
if (!isFutureAccessExpiry(access.subscriptionTotalExpiresAt || access.subscriptionExpiresAt)) return;
try {
window.localStorage.setItem(
ACTIVE_ACCESS_CACHE_KEY,
JSON.stringify({
ts: Date.now(),
access: {
authenticated: access.authenticated,
userId: access.userId,
subscriptionActive: access.subscriptionActive,
subscriptionPlanCode: access.subscriptionPlanCode,
subscriptionExpiresAt: access.subscriptionExpiresAt,
subscriptionTotalExpiresAt: access.subscriptionTotalExpiresAt,
subscriptionQueuedDays: access.subscriptionQueuedDays,
points: access.points,
},
}),
);
} catch {}
}
function createTransientAccess(error: unknown): ProAccessState {
return {
...createEmptyAccess(true),
@@ -428,6 +499,11 @@ function PolyWeatherTerminal({
const [activeNavKey, setActiveNavKey] = useState<string>("thresholds");
const [onlineCount, setOnlineCount] = useState<number | null>(null);
useEffect(() => {
if (activeNavKey !== "thresholds") return;
void preloadTemperatureChartCanvas();
}, [activeNavKey]);
useEffect(() => {
const fetchOnline = () => {
if (typeof document !== "undefined" && document.visibilityState === "hidden") return;
@@ -960,6 +1036,73 @@ function PolyWeatherTerminal({
);
}
function AuthSyncRecoveryScreen({
isEn,
onRetry,
rootClassName,
retrying,
themeMode,
userLocalTime,
}: {
isEn: boolean;
onRetry: () => void;
rootClassName: string;
retrying: boolean;
themeMode: "dark" | "light";
userLocalTime: string;
}) {
return (
<div className={rootClassName}>
<div className={clsx("flex h-screen w-full bg-[#e9edf3] text-slate-950", themeMode === "light" && "light")}>
<aside className="w-[52px] bg-[#171d24]" />
<main className="flex flex-1 flex-col">
<header className="flex h-[52px] items-center justify-between border-b border-slate-200 bg-white px-4">
<Link href="/" className="flex items-center gap-2 hover:opacity-90">
<img src="/logo.png" alt="PolyWeather" className="h-7 w-auto object-contain" />
<span className="text-sm font-semibold tracking-tight text-slate-900">Terminal</span>
</Link>
<div className="font-mono text-sm text-slate-500">{userLocalTime}</div>
</header>
<section className="grid flex-1 place-items-center p-6">
<div className="w-full max-w-md rounded-[6px] border border-amber-200 bg-white p-6 shadow-sm">
<div className="mb-3 inline-flex rounded-full border border-amber-200 bg-amber-50 px-2.5 py-1 text-[11px] font-black text-amber-700">
{isEn ? "Access sync timeout" : "权限同步超时"}
</div>
<h1 className="text-xl font-black text-slate-950">
{isEn ? "We could not confirm your subscription yet" : "暂时未能确认你的订阅状态"}
</h1>
<p className="mt-3 text-sm leading-6 text-slate-600">
{isEn
? "The session or entitlement service is taking too long. Retry access sync; if you just paid, wait a few seconds and try again."
: "当前会话或会员服务响应过慢。请先重新同步权限;如果刚完成支付,等待几秒后再试。"}
</p>
<div className="mt-5 flex flex-col gap-3 sm:flex-row">
<button
type="button"
onClick={onRetry}
disabled={retrying}
className="inline-flex flex-1 items-center justify-center gap-2 rounded-[4px] bg-blue-600 px-4 py-2.5 text-sm font-black text-white transition hover:bg-blue-700 disabled:cursor-wait disabled:opacity-70"
>
{retrying && (
<span className="h-4 w-4 animate-spin rounded-full border-2 border-white/70 border-t-transparent" />
)}
{isEn ? "Retry access sync" : "重新同步权限"}
</button>
<Link
href="/account"
className="inline-flex flex-1 items-center justify-center rounded-[4px] border border-slate-200 bg-white px-4 py-2.5 text-sm font-black text-slate-700 transition hover:bg-slate-50"
>
{isEn ? "Open account" : "打开账户页"}
</Link>
</div>
</div>
</section>
</main>
</div>
</div>
);
}
function ScanTerminalScreen() {
const [proAccess, setProAccess] = useState<ProAccessState>(() =>
createEmptyAccess(true),
@@ -973,17 +1116,26 @@ function ScanTerminalScreen() {
const headers: Record<string, string> = { Accept: "application/json" };
const token = String(accessToken || "").trim();
if (token) headers.Authorization = `Bearer ${token}`;
const response = await fetch(
options?.preferSnapshot
? "/api/auth/me?prefer_snapshot=1"
: "/api/auth/me",
{
cache: "no-store",
headers,
},
);
if (!response.ok) throw new Error(`HTTP ${response.status}`);
return response.json() as Promise<AuthProfilePayload>;
const controller = typeof AbortController !== "undefined" ? new AbortController() : null;
const timeoutId = controller
? globalThis.setTimeout(() => controller.abort(), AUTH_PROFILE_REQUEST_TIMEOUT_MS)
: null;
try {
const response = await fetch(
options?.preferSnapshot
? "/api/auth/me?prefer_snapshot=1"
: "/api/auth/me",
{
cache: "no-store",
headers,
signal: controller?.signal,
},
);
if (!response.ok) throw new Error(`HTTP ${response.status}`);
return response.json() as Promise<AuthProfilePayload>;
} finally {
if (timeoutId !== null) globalThis.clearTimeout(timeoutId);
}
},
[],
);
@@ -1020,6 +1172,7 @@ function ScanTerminalScreen() {
hasSupabasePublicEnv: supabaseEnabled,
loadAuthProfile: (accessToken) =>
loadAuthProfile(accessToken, { preferSnapshot: false }),
timeoutMs: AUTH_PROFILE_REQUEST_TIMEOUT_MS + 2000,
});
setProAccess((prev) => mergeAccessStateWithAuthPayload(prev, payload));
}, [loadAuthProfile]);
@@ -1082,13 +1235,17 @@ function ScanTerminalScreen() {
setLocale((prev) => (prev === "zh-CN" ? "en-US" : "zh-CN"));
const [hydrated, setHydrated] = useState(false);
const [localFullAccess, setLocalFullAccess] = useState(false);
const [authWaitExpired, setAuthWaitExpired] = useState(false);
const [authRetrying, setAuthRetrying] = useState(false);
const canUseLocalFullAccess = hydrated && localFullAccess;
const isAuthenticated =
hydrated && (proAccess.authenticated || canUseLocalFullAccess);
const isPro =
hydrated && (proAccess.subscriptionActive || canUseLocalFullAccess);
const accessDecisionPending =
!hydrated || (proAccess.loading && !canUseLocalFullAccess);
!hydrated || (proAccess.loading && !canUseLocalFullAccess && !authWaitExpired);
const authSyncRecoveryNeeded =
hydrated && proAccess.loading && !canUseLocalFullAccess && authWaitExpired;
const shouldShowPaywall = !accessDecisionPending && (!isAuthenticated || !isPro);
const userLocalTime = useUserLocalClock();
const { themeMode } = useScanTerminalTheme();
@@ -1128,6 +1285,10 @@ function ScanTerminalScreen() {
cancelled = true;
};
}
const cachedActiveAccess = readCachedActiveAccess();
if (cachedActiveAccess) {
setProAccess(cachedActiveAccess);
}
if (typeof fetch !== "function") {
setProAccess(createEmptyAccess(false));
return () => {
@@ -1142,6 +1303,7 @@ function ScanTerminalScreen() {
: Promise.resolve({ data: { session: null } }),
hasSupabasePublicEnv: supabaseEnabled,
loadAuthProfile,
timeoutMs: AUTH_PROFILE_REQUEST_TIMEOUT_MS + 2000,
})
.then((payload) => {
if (cancelled) return;
@@ -1165,6 +1327,23 @@ function ScanTerminalScreen() {
};
}, [loadAuthProfile, refreshLiveAuthProfile]);
useEffect(() => {
if (!hydrated || !proAccess.loading || canUseLocalFullAccess) {
setAuthWaitExpired(false);
return;
}
const timer = window.setTimeout(() => {
setAuthWaitExpired(true);
}, AUTH_DECISION_RECOVERY_MS);
return () => window.clearTimeout(timer);
}, [canUseLocalFullAccess, hydrated, proAccess.loading]);
useEffect(() => {
if (hydrated && proAccess.authenticated && proAccess.subscriptionActive) {
writeCachedActiveAccess(proAccess);
}
}, [hydrated, proAccess]);
useEffect(() => {
if (
!hydrated ||
@@ -1188,6 +1367,7 @@ function ScanTerminalScreen() {
: Promise.resolve({ data: { session: null } }),
hasSupabasePublicEnv: supabaseEnabled,
loadAuthProfile,
timeoutMs: AUTH_PROFILE_REQUEST_TIMEOUT_MS + 2000,
});
if (cancelled) return;
setProAccess((prev) => mergeAccessStateWithAuthPayload(prev, payload));
@@ -1253,9 +1433,20 @@ function ScanTerminalScreen() {
clearCityDetailCache();
refreshScanTerminalManually();
}, [refreshScanTerminalManually]);
const handleRetryAuthSync = useCallback(() => {
setAuthRetrying(true);
setAuthWaitExpired(false);
void refreshLiveAuthProfile()
.catch((error) => {
setProAccess((prev) => ({ ...prev, error: String(error) }));
})
.finally(() => {
setAuthRetrying(false);
});
}, [refreshLiveAuthProfile]);
const [cityFallbackRows, setCityFallbackRows] = useState<ScanOpportunityRow[]>(() =>
cityListItemsToScanRows(STATIC_CITY_LIST),
cityListItemsToScanRows(readCachedCityList() || STATIC_CITY_LIST),
);
const rows = useMemo(
() => {
@@ -1271,9 +1462,15 @@ function ScanTerminalScreen() {
useEffect(() => {
if (!isPro || typeof fetch !== "function") return;
if (fallbackFetchedRef.current) return;
const cachedCities = readCachedCityList();
if (cachedCities) {
fallbackFetchedRef.current = true;
setCityFallbackRows(cityListItemsToScanRows(cachedCities));
return;
}
const controller = new AbortController();
fetch("/api/cities", {
cache: "no-store",
cache: "default",
headers: { Accept: "application/json" },
signal: controller.signal,
})
@@ -1284,6 +1481,7 @@ function ScanTerminalScreen() {
.then((payload) => {
if (!payload || !Array.isArray(payload.cities)) return;
fallbackFetchedRef.current = true;
writeCachedCityList(payload.cities);
setCityFallbackRows(cityListItemsToScanRows(payload.cities));
})
.catch(() => {});
@@ -1338,6 +1536,19 @@ function ScanTerminalScreen() {
);
}
if (authSyncRecoveryNeeded) {
return (
<AuthSyncRecoveryScreen
isEn={isEn}
onRetry={handleRetryAuthSync}
rootClassName={scanRootClass}
retrying={authRetrying}
themeMode={themeMode}
userLocalTime={userLocalTime}
/>
);
}
if (shouldShowPaywall) {
return (
<ProductAccessRequired
@@ -1,13 +1,12 @@
"use client";
import clsx from "clsx";
import dynamic from "next/dynamic";
import { useCallback, useEffect, useMemo, useRef, useState } from "react";
import type { ScanOpportunityRow } from "@/lib/dashboard-types";
import { useLatestPatch, useSseResyncVersion } from "@/hooks/use-sse-patches";
import { Panel } from "@/components/dashboard/scan-terminal/Panel";
import { ModelCurvesSummary } from "@/components/dashboard/scan-terminal/ModelCurvesSummary";
import { TemperatureChartCanvasFallback } from "@/components/dashboard/scan-terminal/TemperatureChartCanvasFallback";
import { TemperatureChartCanvas } from "@/components/dashboard/scan-terminal/TemperatureChartCanvas";
import { TemperatureRunwayDetails } from "@/components/dashboard/scan-terminal/TemperatureRunwayDetails";
import { TemperatureStatsBars } from "@/components/dashboard/scan-terminal/TemperatureStatsBars";
import { rowName } from "@/components/dashboard/scan-terminal/utils";
@@ -56,17 +55,16 @@ const PEAK_GLOW_BADGE_CLASS = {
} as const;
const PROBABILITY_REFRESH_AFTER_PATCH_MS = 60_000;
const FOREGROUND_FULL_DETAIL_REFRESH_DEDUP_MS = 90_000;
const DETAIL_LOAD_BATCH_DELAY_MS = 0;
const INITIAL_DETAIL_LOAD_SLOTS = 3;
const DEFERRED_DETAIL_LOAD_DELAY_MS = 1_200;
const DEFERRED_DETAIL_LOAD_GROUP_SIZE = 3;
const DEFERRED_DETAIL_LOAD_WAVE_STEP_MS = 900;
const TemperatureChartCanvas = dynamic(
() =>
import("@/components/dashboard/scan-terminal/TemperatureChartCanvas").then(
(mod) => mod.TemperatureChartCanvas,
),
{
ssr: false,
loading: () => <TemperatureChartCanvasFallback />,
},
);
export function preloadTemperatureChartCanvas() {
return Promise.resolve();
}
function peakGlowLabel(state: keyof typeof PEAK_GLOW_PANEL_CLASS, isEn: boolean) {
if (state === "watch") return isEn ? "Watch" : "关注";
@@ -113,12 +111,49 @@ function shouldFetchCityDetailForChart({
city,
documentHidden,
isChartVisible,
compact = false,
isActive = false,
isMaximized = false,
slotIndex = 0,
detailLoadReady = true,
}: {
city: string;
documentHidden: boolean;
isChartVisible: boolean;
compact?: boolean;
isActive?: boolean;
isMaximized?: boolean;
slotIndex?: number;
detailLoadReady?: boolean;
}) {
return Boolean(city) && isChartVisible && !documentHidden;
if (!city || !isChartVisible || documentHidden) return false;
if (!compact || isActive || isMaximized) return true;
if (normalizeSlotIndex(slotIndex) < INITIAL_DETAIL_LOAD_SLOTS) return true;
return detailLoadReady;
}
function normalizeSlotIndex(slotIndex: number | null | undefined) {
return Number.isFinite(slotIndex) && Number(slotIndex) >= 0 ? Math.floor(Number(slotIndex)) : 0;
}
function getInitialDetailLoadDelayMs({
compact,
isActive,
isMaximized,
slotIndex,
}: {
compact?: boolean;
isActive?: boolean;
isMaximized?: boolean;
slotIndex?: number;
}) {
if (!compact || isActive || isMaximized) return 0;
const normalizedIndex = normalizeSlotIndex(slotIndex);
if (normalizedIndex < INITIAL_DETAIL_LOAD_SLOTS) return 0;
const deferredWave = Math.floor(
(normalizedIndex - INITIAL_DETAIL_LOAD_SLOTS) / DEFERRED_DETAIL_LOAD_GROUP_SIZE,
);
return DEFERRED_DETAIL_LOAD_DELAY_MS + deferredWave * DEFERRED_DETAIL_LOAD_WAVE_STEP_MS;
}
// ── Main component ─────────────────────────────────────────────────────
@@ -165,6 +200,7 @@ export function LiveTemperatureThresholdChart({
const lastPatchAtRef = useRef<number>(Date.now());
const lastAppliedPatchRevisionRef = useRef<number>(0);
const lastProbabilityRefreshAtRef = useRef<number>(0);
const lastForegroundRefreshAtRef = useRef<number>(0);
const localDayRolloverFetchDateRef = useRef<string>("");
const [isChartVisible, setIsChartVisible] = useState(
() => typeof IntersectionObserver === "undefined",
@@ -177,6 +213,11 @@ export function LiveTemperatureThresholdChart({
const [targetResolution, setTargetResolution] = useState<string>(() =>
prefersHighFrequencyRunwayResolution(row, null) ? "1m" : "10m",
);
const detailLoadDelayMs = useMemo(
() => getInitialDetailLoadDelayMs({ compact, isActive, isMaximized, slotIndex }),
[compact, isActive, isMaximized, slotIndex],
);
const [detailLoadReady, setDetailLoadReady] = useState(() => detailLoadDelayMs === 0);
const [currentCityLocalDate, setCurrentCityLocalDate] = useState(() =>
formatCityLocalDate(row?.tz_offset_seconds),
);
@@ -189,7 +230,7 @@ export function LiveTemperatureThresholdChart({
setTargetResolution(prefersHighFrequencyRunwayResolution(row, null) ? "1m" : "10m");
setHourly(seedHourlyForecastFromRow(row));
setLiveTemp(null);
setIsHourlyLoading(Boolean(city));
setIsHourlyLoading(Boolean(city) && detailLoadDelayMs === 0);
setDetailError(null);
setDetailRetryNonce(0);
setShowingStaleDetail(false);
@@ -197,9 +238,25 @@ export function LiveTemperatureThresholdChart({
lastPatchAtRef.current = Date.now();
lastAppliedPatchRevisionRef.current = 0;
lastProbabilityRefreshAtRef.current = 0;
lastForegroundRefreshAtRef.current = 0;
localDayRolloverFetchDateRef.current = "";
setCurrentCityLocalDate(formatCityLocalDate(row?.tz_offset_seconds));
}, [city]);
}, [city, detailLoadDelayMs]);
useEffect(() => {
if (!city) {
setDetailLoadReady(false);
return;
}
if (detailLoadDelayMs <= 0) {
setDetailLoadReady(true);
return;
}
setDetailLoadReady(false);
const id = setTimeout(() => setDetailLoadReady(true), detailLoadDelayMs);
return () => clearTimeout(id);
}, [city, detailLoadDelayMs]);
useEffect(() => {
const node = chartVisibilityRef.current;
@@ -232,17 +289,6 @@ export function LiveTemperatureThresholdChart({
return;
}
if (
!shouldFetchCityDetailForChart({
city,
documentHidden:
typeof document !== "undefined" && document.visibilityState === "hidden",
isChartVisible,
})
) {
return;
}
const cacheKey = `${city}:${targetResolution}`;
let cached = _hourlyCache.get(cacheKey);
if (!cached || Date.now() - Number(cached.ts || 0) >= HOURLY_CACHE_TTL_MS) {
@@ -267,6 +313,23 @@ export function LiveTemperatureThresholdChart({
return;
}
if (
!shouldFetchCityDetailForChart({
city,
documentHidden:
typeof document !== "undefined" && document.visibilityState === "hidden",
isChartVisible,
compact,
isActive,
isMaximized,
slotIndex,
detailLoadReady,
})
) {
setIsHourlyLoading(false);
return;
}
if (!cached && !hasLoadedHourlyDetailRef.current) {
setHourly(seedHourlyForecastFromRow(row));
setShowingStaleDetail(false);
@@ -274,9 +337,6 @@ export function LiveTemperatureThresholdChart({
setIsHourlyLoading(true);
let cancelled = false;
// Prioritize active slots, stagger/delay background slots to optimize load performance
const delay = isActive ? 0 : (slotIndex ? 300 + slotIndex * 250 : 350);
const timer = setTimeout(() => {
fetchHourlyForecastForCity(city, { resolution: targetResolution })
.then((data) => {
@@ -296,13 +356,25 @@ export function LiveTemperatureThresholdChart({
.finally(() => {
if (!cancelled) setIsHourlyLoading(false);
});
}, delay);
}, DETAIL_LOAD_BATCH_DELAY_MS);
return () => {
cancelled = true;
clearTimeout(timer);
};
}, [city, row, isActive, slotIndex, targetResolution, isChartVisible, detailRetryNonce, isEn]);
}, [
city,
row,
targetResolution,
isChartVisible,
compact,
isActive,
isMaximized,
slotIndex,
detailLoadReady,
detailRetryNonce,
isEn,
]);
useEffect(() => {
if (!latestPatch || latestPatch.revision <= lastAppliedPatchRevisionRef.current) return;
@@ -398,7 +470,19 @@ export function LiveTemperatureThresholdChart({
let cancelled = false;
const refreshForegroundFullDetail = () => {
lastPatchAtRef.current = Date.now();
const now = Date.now();
const cacheKey = `${city}:${targetResolution}`;
const cached = _hourlyCache.get(cacheKey);
const cacheAge = cached ? now - Number(cached.ts || 0) : Number.POSITIVE_INFINITY;
if (
now - lastForegroundRefreshAtRef.current < FOREGROUND_FULL_DETAIL_REFRESH_DEDUP_MS ||
(cacheAge >= 0 && cacheAge < FOREGROUND_FULL_DETAIL_REFRESH_DEDUP_MS)
) {
return;
}
lastForegroundRefreshAtRef.current = now;
lastPatchAtRef.current = now;
fetchHourlyForecastForCity(city, { ignoreCache: true, resolution: targetResolution })
.then((data) => {
@@ -885,6 +969,7 @@ export const __getLiveObservationLabelsForTest = getLiveObservationLabels;
export const __getObservationDisplayMetricsForTest = getObservationDisplayMetrics;
export const __getPeakGlowStateForTest = getPeakGlowState;
export const __getWundergroundDailyHighForTest = getWundergroundDailyHigh;
export const __getInitialDetailLoadDelayMsForTest = getInitialDetailLoadDelayMs;
export const __shouldFetchCityDetailForChartForTest = shouldFetchCityDetailForChart;
export const __shouldPollLiveChartForTest = shouldPollLiveChart;
export const __mergePatchIntoHourlyForTest = mergePatchIntoHourly;
@@ -1,35 +0,0 @@
export function TemperatureChartCanvasFallback({ compact }: { compact?: boolean }) {
const horizontalLines = compact ? 5 : 7;
const verticalLines = compact ? 5 : 8;
const minChartHeight = compact === false ? 220 : 120;
return (
<div
className="relative flex-1 overflow-hidden rounded-sm border border-slate-100 bg-white"
style={{ minHeight: minChartHeight }}
>
<div className="absolute inset-x-3 bottom-7 top-4 rounded-sm border border-slate-100">
{Array.from({ length: horizontalLines }).map((_, index) => (
<span
key={`h-${index}`}
className="absolute left-0 right-0 border-t border-dashed border-sky-100"
style={{ top: `${(index / Math.max(1, horizontalLines - 1)) * 100}%` }}
/>
))}
{Array.from({ length: verticalLines }).map((_, index) => (
<span
key={`v-${index}`}
className="absolute bottom-0 top-0 border-l border-dashed border-sky-100"
style={{ left: `${(index / Math.max(1, verticalLines - 1)) * 100}%` }}
/>
))}
</div>
<div className="absolute inset-0 grid place-items-center">
<div className="flex items-center gap-2 rounded border border-slate-200 bg-white px-3 py-2 text-xs font-semibold text-slate-500 shadow-sm">
<span className="h-3 w-3 animate-spin rounded-full border-2 border-blue-200 border-t-blue-500" />
</div>
</div>
</div>
);
}
@@ -1,6 +1,9 @@
import {
CITY_LIST_CACHE_TTL_MS,
cityListItemsToScanRows,
mergeScanRowsWithCityFallbackRows,
readCachedCityList,
writeCachedCityList,
} from "@/components/dashboard/scan-terminal/city-fallback-rows";
import type { CityListItem } from "@/lib/dashboard-types";
@@ -47,6 +50,15 @@ export function runTests() {
];
const rows = cityListItemsToScanRows(cities);
const cacheStorage = new Map<string, string>();
const memoryStorage = {
getItem: (key: string) => cacheStorage.get(key) ?? null,
removeItem: (key: string) => { cacheStorage.delete(key); },
setItem: (key: string, value: string) => { cacheStorage.set(key, value); },
};
writeCachedCityList(cities, memoryStorage, 1_000);
const cachedCities = readCachedCityList(memoryStorage, 1_000 + 60_000);
assert(rows.length === 2, "fallback rows should preserve every city");
assert(rows[0].id === "city-fallback:taipei", "fallback row id should be stable");
@@ -57,6 +69,12 @@ export function runTests() {
assert(rows[0].temp_symbol === "°C", "celsius cities should use °C");
assert(rows[1].trading_region === "north_america", "known cities should keep their configured product region");
assert(rows[1].temp_symbol === "°F", "fahrenheit cities should use °F");
assert(cachedCities?.length === 2, "fresh city registry cache should return the stored city list");
assert(cachedCities?.[0]?.name === "taipei", "city registry cache should preserve city payloads");
assert(
readCachedCityList(memoryStorage, 1_000 + CITY_LIST_CACHE_TTL_MS + 1) === null,
"expired city registry cache should be ignored so the dashboard can revalidate",
);
const scanRows = [
{
@@ -122,10 +140,13 @@ export function runTests() {
assert(
dashboardSource.includes("cityListItemsToScanRows") &&
dashboardSource.includes("STATIC_CITY_LIST") &&
dashboardSource.includes("readCachedCityList") &&
dashboardSource.includes("writeCachedCityList") &&
dashboardSource.includes("useState<ScanOpportunityRow[]>(() =>") &&
dashboardSource.includes("/api/cities") &&
dashboardSource.includes('cache: "default"') &&
dashboardSource.includes("cityFallbackRows"),
"terminal dashboard should seed fallback rows from the static city snapshot before refreshing /api/cities",
"terminal dashboard should seed fallback rows from cached/static city snapshots and avoid no-store /api/cities fetches when possible",
);
assert(fs.existsSync(staticCitiesPath), "/api/cities route should have a static city snapshot fallback");
assert(
@@ -20,6 +20,7 @@ export function runTests() {
const cached = buildCityDetailProxyCachePolicy("false", 15);
assert.equal(cached.fetchMode, "revalidate");
assert.equal(cached.revalidateSeconds, 15);
assert.match(cached.responseCacheControl, /max-age=15/);
assert.match(cached.responseCacheControl, /s-maxage=15/);
const scanForced = buildForceRefreshProxyCachePolicy("true", 10);
@@ -55,6 +56,16 @@ export function runTests() {
/hasDirectBackendApiBaseUrl/,
"public scan terminal requests should only attach user auth in direct-backend mode so CDN caches can be shared through the Next proxy",
);
assert.match(
scanTerminalClientSource,
/SCAN_TERMINAL_PAYLOAD_VERSION/,
"scan terminal client should include a stable payload version in the URL so CDN caches roll forward after response-shape optimizations",
);
assert.match(
scanTerminalClientSource,
/params\.set\("_v",\s*SCAN_TERMINAL_PAYLOAD_VERSION\)/,
"scan terminal client should vary the CDN cache key without changing backend scan filters",
);
const overviewProxySource = fs.readFileSync(
path.join(
@@ -6,6 +6,7 @@ import {
} from "@/lib/refresh-policy";
import { scanTerminalQueryPolicy } from "@/components/dashboard/scan-terminal/scan-terminal-client";
import {
__getInitialDetailLoadDelayMsForTest,
__shouldFetchCityDetailForChartForTest,
__shouldPollLiveChartForTest,
} from "@/components/dashboard/scan-terminal/LiveTemperatureThresholdChart";
@@ -71,11 +72,33 @@ export async function runTests() {
chartSource.includes("2 * 60_000"),
"selected city chart should consume SSE patches and use a 2-minute no-patch fallback",
);
assert(
chartSource.includes("preloadTemperatureChartCanvas"),
"terminal chart canvas should expose a preload hook for first-paint optimization",
);
assert(
chartSource.includes('from "@/components/dashboard/scan-terminal/TemperatureChartCanvas"') &&
!chartSource.includes("next/dynamic") &&
!chartSource.includes("TemperatureChartCanvasFallback"),
"terminal chart canvas should be loaded with the terminal route instead of showing a second-stage dynamic chunk fallback",
);
assert(
dashboardSource.includes("preloadTemperatureChartCanvas") &&
dashboardSource.includes("void preloadTemperatureChartCanvas()") &&
dashboardSource.includes('activeNavKey !== "thresholds"'),
"terminal screen should preload the chart chunk once access is confirmed on the chart tab",
);
assert(
chartSource.includes("fetchHourlyForecastForCity(city, { ignoreCache: true, resolution: targetResolution })") &&
chartSource.includes("setHourly(data)"),
"visible chart fallback must refresh the full city detail payload at the current chart resolution when SSE patches stop",
);
assert(
chartLogicSource.includes("HOURLY_FORCE_REFRESH_DEDUP_MS") &&
chartLogicSource.includes("maxAgeMs: HOURLY_FORCE_REFRESH_DEDUP_MS") &&
chartLogicSource.includes("recentlyRefreshed.data"),
"forced chart detail refreshes should reuse a very recent full-detail payload instead of refetching repeatedly",
);
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",
@@ -96,6 +119,15 @@ export async function runTests() {
chartLogicSource.includes("primeCityDetailCache"),
"visible terminal chart detail fetches should be coalesced into one batch request and prime the shared chart cache",
);
assert(
chartLogicSource.includes('scope: "chart"') &&
chartLogicSource.includes("params.toString()"),
"terminal chart detail batches should request the slim chart scope instead of the full city detail payload",
);
assert(
chartLogicSource.includes("CITY_DETAIL_BATCH_WINDOW_MS = 100"),
"visible terminal chart detail fetches should use a wide enough batch window to coalesce cards mounted across adjacent frames",
);
assert(
chartLogicSource.includes("partial?: boolean") &&
chartLogicSource.includes("missing?: string[]"),
@@ -108,11 +140,17 @@ export async function runTests() {
flushCityDetailBatchBlock.includes("payload?.partial === true"),
"partial detail-batch misses should resolve without immediately issuing single-city fallback requests",
);
const fetchHourlyBlock = chartLogicSource.match(/async function fetchHourlyForecastForCity[\s\S]*?\r?\n}\r?\n\r?\nfunction fetchCityDetailWithTimeout/)?.[0] || "";
assert(
flushCityDetailBatchBlock.includes("if (!payload)") &&
flushCityDetailBatchBlock.includes("resolveAllBatchWaitersAsNull") &&
!flushCityDetailBatchBlock.includes("resolveCityDetailBatchWithSingleFallback"),
"whole-batch failures should stop at the chart batch layer instead of fanning out into single-city full-detail requests",
);
const fetchHourlyBlock = chartLogicSource.match(/async function fetchHourlyForecastForCity[\s\S]*?\r?\n}\r?\n\r?\nfunction shouldPollLiveChart/)?.[0] || "";
assert(
fetchHourlyBlock.includes("queueCityDetailBatch(city, resParam)") &&
!fetchHourlyBlock.includes("runQueuedHourlyDetailRequest"),
"first-paint and background city detail refreshes should both enter the batch queue before falling back to single requests",
"first-paint and background city detail refreshes should both enter the batch queue without falling back to single requests",
);
assert(
dashboardSource.includes("ONLINE_USERS_REFRESH_MS = 5 * 60_000") &&
@@ -125,6 +163,11 @@ export async function runTests() {
chartSource.includes("isChartVisible"),
"city detail prefetch should be gated to visible chart cards instead of every mounted slot",
);
assert(
chartSource.includes("DETAIL_LOAD_BATCH_DELAY_MS") &&
!chartSource.includes("slotIndex ? 300 + slotIndex * 250"),
"visible chart detail loads should enter the batch queue together instead of being staggered into single-city requests",
);
assert(
chartSource.includes("allowStale: true") &&
chartCanvasSourceIncludes(chartSource, "数据暂不可用") &&
@@ -143,6 +186,58 @@ export async function runTests() {
__shouldFetchCityDetailForChartForTest({ city: "paris", documentHidden: true, isChartVisible: true }) === false,
"hidden browser tabs should not prefetch city detail",
);
assert(
__getInitialDetailLoadDelayMsForTest({ compact: true, isActive: false, isMaximized: false, slotIndex: 0 }) === 0 &&
__getInitialDetailLoadDelayMsForTest({ compact: true, isActive: false, isMaximized: false, slotIndex: 2 }) === 0,
"first visible grid charts should fetch full detail immediately",
);
assert(
__getInitialDetailLoadDelayMsForTest({ compact: true, isActive: false, isMaximized: false, slotIndex: 3 }) > 0,
"later non-active grid charts should defer full-detail loading after first paint",
);
assert(
__getInitialDetailLoadDelayMsForTest({ compact: true, isActive: false, isMaximized: false, slotIndex: 3 }) ===
__getInitialDetailLoadDelayMsForTest({ compact: true, isActive: false, isMaximized: false, slotIndex: 5 }),
"deferred grid charts should load in same-wave groups so detail-batch can coalesce them",
);
assert(
__getInitialDetailLoadDelayMsForTest({ compact: true, isActive: false, isMaximized: false, slotIndex: 6 }) ===
__getInitialDetailLoadDelayMsForTest({ compact: true, isActive: false, isMaximized: false, slotIndex: 8 }) &&
__getInitialDetailLoadDelayMsForTest({ compact: true, isActive: false, isMaximized: false, slotIndex: 6 }) >
__getInitialDetailLoadDelayMsForTest({ compact: true, isActive: false, isMaximized: false, slotIndex: 5 }),
"later deferred grid charts should form a second batch wave instead of one request per slot",
);
assert(
__getInitialDetailLoadDelayMsForTest({ compact: true, isActive: true, isMaximized: false, slotIndex: 8 }) === 0 &&
__getInitialDetailLoadDelayMsForTest({ compact: true, isActive: false, isMaximized: true, slotIndex: 8 }) === 0,
"active or maximized charts should bypass deferred detail loading",
);
assert(
__shouldFetchCityDetailForChartForTest({
city: "paris",
documentHidden: false,
isChartVisible: true,
compact: true,
isActive: false,
isMaximized: false,
slotIndex: 5,
detailLoadReady: false,
}) === false,
"deferred grid charts should not enter the detail request queue before their delay is ready",
);
assert(
__shouldFetchCityDetailForChartForTest({
city: "paris",
documentHidden: false,
isChartVisible: true,
compact: true,
isActive: false,
isMaximized: false,
slotIndex: 5,
detailLoadReady: true,
}) === true,
"deferred grid charts should fetch detail after their delay is ready",
);
const normalizedBatchDetail = __resolveCityDetailFromBatchForTest(
{
"hong kong": {
@@ -94,11 +94,30 @@ export function runTests() {
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("SSE_REPLAY_EVENTS_PER_CITY") &&
hook.includes("resolveSseReplayLimit") &&
hook.includes('params.set("replay_limit", String(resolveSseReplayLimit(cities.length)))') &&
!hook.includes('params.set("replay_limit", "500")'),
"frontend patch hook should size SSE replay_limit by visible city count instead of always asking for 500 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");
assert(
hook.includes("SSE_SUBSCRIPTION_RECONNECT_DELAY_MS") &&
hook.includes("scheduleSubscriptionReconnect") &&
hook.includes("clearSubscriptionReconnectTimer"),
"frontend patch hook must debounce visible-city subscription changes before reopening SSE",
);
const subscriptionBlock = hook.match(/function registerCitySubscription[\s\S]*?\n}\r?\n\r?\nfunction normalizeLegacyPatch/)?.[0] || "";
assert(
subscriptionBlock.includes("scheduleSubscriptionReconnect()") &&
!subscriptionBlock.includes("ensureSsePatchConnection();"),
"city subscription mount/unmount should schedule one coalesced SSE reconnect instead of reconnecting per chart",
);
const bffEventsRoute = readFrontendFile("app", "api", "events", "route.ts");
assert(bffEventsRoute.includes("searchParams"), "Next.js SSE proxy must forward query parameters to FastAPI");
@@ -179,8 +198,9 @@ export function runTests() {
assert(
foregroundRefreshBlock.includes("ignoreCache: true") &&
foregroundRefreshBlock.includes("fetchHourlyForecastForCity") &&
foregroundRefreshBlock.includes("FOREGROUND_FULL_DETAIL_REFRESH_DEDUP_MS") &&
!foregroundRefreshBlock.includes("setIsHourlyLoading(true)"),
"foreground resume refresh should update full detail immediately in the background without showing the loading overlay",
"foreground resume refresh should update full detail in the background without showing the loading overlay or refetching fresh detail",
);
assert(
!chart.includes("/api/city/${encodeURIComponent(city)}/summary"),
@@ -239,7 +259,7 @@ export function runTests() {
);
assert(
chartLogic.includes("HOURLY_DETAIL_REQUEST_TIMEOUT_MS = 12_000") &&
chartLogic.includes("fetchCityDetailWithTimeout") &&
chartLogic.includes("fetchCityDetailBatchWithTimeout") &&
chartLogic.includes("signal: controller.signal") &&
chartLogic.includes("controller.abort()"),
"city detail chart fetches must have a frontend timeout so panels cannot stay on 加载图表 forever",
@@ -124,6 +124,45 @@ export async function runTests() {
"terminal auth bootstrap should accept an authenticated cookie profile immediately",
);
const slowDegradedSession = deferred<{ data: { session: { access_token: string } | null } }>();
const timedCookieResult = await loadTerminalAuthProfile({
hasSupabasePublicEnv: true,
getSession: () => slowDegradedSession.promise,
timeoutMs: 1,
loadAuthProfile: (accessToken) => {
assert(!accessToken, "timeout fallback should use the settled cookie profile when bearer session is still pending");
return Promise.resolve({
authenticated: true,
user_id: "cookie-user",
subscription_active: null,
degraded_auth_profile: true,
});
},
});
assert(
timedCookieResult.user_id === "cookie-user" &&
timedCookieResult.subscription_active === null,
"terminal auth bootstrap timeout must resolve with a known degraded cookie profile instead of spinning forever",
);
const neverCookie = deferred<TerminalAuthProfilePayload>();
const neverSession = deferred<{ data: { session: { access_token: string } | null } }>();
let timeoutRejected = false;
try {
await loadTerminalAuthProfile({
hasSupabasePublicEnv: true,
getSession: () => neverSession.promise,
timeoutMs: 1,
loadAuthProfile: () => neverCookie.promise,
});
} catch (error) {
timeoutRejected = String(error).includes("Terminal auth bootstrap timeout");
}
assert(
timeoutRejected,
"terminal auth bootstrap must reject instead of leaving the loading screen unresolved when no profile request settles",
);
const delayedBearerSession = deferred<{ data: { session: { access_token: string } } }>();
let coldStartSettled = false;
const coldStartResultPromise = loadTerminalAuthProfile({
@@ -97,11 +97,10 @@ export function runTests() {
"terminal dashboard must lazy-load the training analytics tab so Recharts stays out of the default terminal path",
);
assert(
chartSource.includes('from "next/dynamic"') &&
chartSource.includes("TemperatureChartCanvasFallback") &&
chartSource.includes("import(\"@/components/dashboard/scan-terminal/TemperatureChartCanvas\")") &&
!chartSource.includes('import { TemperatureChartCanvas } from "@/components/dashboard/scan-terminal/TemperatureChartCanvas";'),
"terminal temperature charts must lazy-load the Recharts canvas behind a lightweight fallback",
chartSource.includes('import { TemperatureChartCanvas } from "@/components/dashboard/scan-terminal/TemperatureChartCanvas";') &&
!chartSource.includes("TemperatureChartCanvasFallback") &&
!chartSource.includes("import(\"@/components/dashboard/scan-terminal/TemperatureChartCanvas\")"),
"terminal temperature charts must load the Recharts canvas with the terminal route so visible cards do not sit behind a second-stage fallback",
);
assert(
chartSource.includes("setLiveTemp(null);") &&
@@ -5,10 +5,61 @@ import {
type RegionKey,
} from "@/components/dashboard/scan-terminal/continent-grouping";
const CITY_LIST_CACHE_KEY = "polyweather_city_list_v1";
export const CITY_LIST_CACHE_TTL_MS = 24 * 60 * 60 * 1000;
type CityListCacheStorage = Pick<Storage, "getItem" | "removeItem" | "setItem">;
function normalizeCityKey(value: string) {
return String(value || "").trim().toLowerCase();
}
function getCityListCacheStorage(storage?: CityListCacheStorage): CityListCacheStorage | null {
if (storage) return storage;
if (typeof window === "undefined") return null;
try {
return window.localStorage;
} catch {
return null;
}
}
export function readCachedCityList(
storage?: CityListCacheStorage,
now = Date.now(),
): CityListItem[] | null {
const cacheStorage = getCityListCacheStorage(storage);
if (!cacheStorage) return null;
try {
const raw = cacheStorage.getItem(CITY_LIST_CACHE_KEY);
if (!raw) return null;
const cached = JSON.parse(raw);
const age = now - Number(cached.ts || 0);
if (age < 0 || age >= CITY_LIST_CACHE_TTL_MS || !Array.isArray(cached.cities)) {
return null;
}
return cached.cities as CityListItem[];
} catch {
try { cacheStorage.removeItem(CITY_LIST_CACHE_KEY); } catch {}
}
return null;
}
export function writeCachedCityList(
cities: CityListItem[],
storage?: CityListCacheStorage,
now = Date.now(),
) {
const cacheStorage = getCityListCacheStorage(storage);
if (!cacheStorage || !Array.isArray(cities) || !cities.length) return;
try {
cacheStorage.setItem(
CITY_LIST_CACHE_KEY,
JSON.stringify({ ts: now, cities }),
);
} catch {}
}
function tempSymbolForCity(city: CityListItem) {
return city.temp_unit === "fahrenheit" ? "°F" : "°C";
}
@@ -22,6 +22,7 @@ export const scanTerminalQueryPolicy = {
autoRefreshMs: null,
manualForceRefreshCooldownMs: 2 * 60_000,
} as const;
const SCAN_TERMINAL_PAYLOAD_VERSION = "runway-slim-v1";
type TerminalQueryOptions = {
forceRefresh?: boolean;
@@ -142,6 +143,7 @@ async function getTerminal({
if (Number.isFinite(timezoneOffsetSeconds)) {
params.set("timezone_offset_seconds", String(Math.trunc(Number(timezoneOffsetSeconds))));
}
params.set("_v", SCAN_TERMINAL_PAYLOAD_VERSION);
if (forceRefresh) {
params.set("_ts", String(Date.now()));
}
@@ -353,6 +353,7 @@ type LocalDayBounds = { start: number; end: number };
const MAX_OBS_POINTS = 1440;
const HOURLY_CACHE_TTL_MS = DASHBOARD_REFRESH_POLICY_MS.metar;
const HOURLY_FORCE_REFRESH_DEDUP_MS = 60_000;
const _hourlyCache = new Map<string, { ts: number; data: HourlyForecast }>();
const _hourlyRequestCache = new Map<string, Promise<HourlyForecast>>();
const MAX_HOURLY_DETAIL_CONCURRENT_REQUESTS = 3;
@@ -366,13 +367,16 @@ const SESSION_CACHE_TTL_MS = DASHBOARD_REFRESH_POLICY_MS.metar;
type HourlyCacheEntry = { ts: number; data: HourlyForecast };
function isFreshHourlyCacheEntry(entry: HourlyCacheEntry | null | undefined) {
return Boolean(entry && Date.now() - Number(entry.ts || 0) < SESSION_CACHE_TTL_MS);
function isFreshHourlyCacheEntry(
entry: HourlyCacheEntry | null | undefined,
maxAgeMs = SESSION_CACHE_TTL_MS,
) {
return Boolean(entry && Date.now() - Number(entry.ts || 0) < maxAgeMs);
}
function readSessionCache(
city: string,
options: { allowStale?: boolean } = {},
options: { allowStale?: boolean; maxAgeMs?: number } = {},
): HourlyCacheEntry | null {
if (typeof window === "undefined") return null;
try {
@@ -382,7 +386,7 @@ function readSessionCache(
if (
item &&
item.ts &&
(options.allowStale || Date.now() - item.ts < SESSION_CACHE_TTL_MS)
(options.allowStale || Date.now() - item.ts < (options.maxAgeMs ?? SESSION_CACHE_TTL_MS))
) {
return item;
}
@@ -392,10 +396,10 @@ function readSessionCache(
function readHourlyCacheEntry(
cacheKey: string,
options: { allowStale?: boolean } = {},
options: { allowStale?: boolean; maxAgeMs?: number } = {},
): HourlyCacheEntry | null {
const cached = _hourlyCache.get(cacheKey);
if (cached && (options.allowStale || isFreshHourlyCacheEntry(cached))) {
if (cached && (options.allowStale || isFreshHourlyCacheEntry(cached, options.maxAgeMs))) {
return cached;
}
@@ -984,7 +988,7 @@ type CityDetailBatchQueue = {
timer: ReturnType<typeof setTimeout> | null;
};
const CITY_DETAIL_BATCH_WINDOW_MS = 25;
const CITY_DETAIL_BATCH_WINDOW_MS = 100;
const CITY_DETAIL_BATCH_MAX_CITIES = 12;
const _cityDetailBatchQueues = new Map<string, CityDetailBatchQueue>();
@@ -1029,16 +1033,6 @@ function primeCityDetailCache(
return data;
}
async function fetchSingleHourlyForecastForCity(
city: string,
resolution: string,
): Promise<HourlyForecast> {
const res = await fetchCityDetailWithTimeout(city, resolution);
if (!res || !res.ok) return null;
const json = await res.json() as CityDetail;
return primeCityDetailCache(city, resolution, json);
}
function queueCityDetailBatch(city: string, resolution: string): Promise<HourlyForecast> {
return new Promise<HourlyForecast>((resolve, reject) => {
const queue = _cityDetailBatchQueues.get(resolution) || {
@@ -1069,13 +1063,6 @@ function resolveBatchWaiters(
(waiters || []).forEach((waiter) => waiter.resolve(value));
}
function rejectBatchWaiters(
waiters: CityDetailBatchWaiter[] | undefined,
reason: unknown,
) {
(waiters || []).forEach((waiter) => waiter.reject(reason));
}
function resolveCityDetailFromBatch(
details: Record<string, CityDetail | null | undefined> | undefined,
city: string,
@@ -1114,6 +1101,11 @@ async function flushCityDetailBatch(resolution: string) {
try {
const payload = await fetchCityDetailBatchWithTimeout(cities, resolution);
if (!payload) {
resolveAllBatchWaitersAsNull(cities, queue);
return;
}
const details = payload?.details || {};
const partialMissingCities =
payload?.partial === true
@@ -1132,33 +1124,23 @@ async function flushCityDetailBatch(resolution: string) {
resolveBatchWaiters(waiters, null);
return;
}
try {
resolveBatchWaiters(
waiters,
await runQueuedHourlyDetailRequest(() => fetchSingleHourlyForecastForCity(city, resolution)),
);
} catch (error) {
rejectBatchWaiters(waiters, error);
}
resolveBatchWaiters(waiters, null);
}),
);
} catch (error) {
await Promise.all(
cities.map(async (city) => {
const waiters = queue.waiters.get(city);
try {
resolveBatchWaiters(
waiters,
await runQueuedHourlyDetailRequest(() => fetchSingleHourlyForecastForCity(city, resolution)),
);
} catch (fallbackError) {
rejectBatchWaiters(waiters, fallbackError || error);
}
}),
);
resolveAllBatchWaitersAsNull(cities, queue);
}
}
function resolveAllBatchWaitersAsNull(
cities: string[],
queue: CityDetailBatchQueue,
) {
cities.forEach((city) => {
resolveBatchWaiters(queue.waiters.get(city), null);
});
}
function fetchCityDetailBatchWithTimeout(cities: string[], resolution: string) {
const controller = new AbortController();
const timeoutId = globalThis.setTimeout(() => controller.abort(), HOURLY_DETAIL_REQUEST_TIMEOUT_MS);
@@ -1168,6 +1150,7 @@ function fetchCityDetailBatchWithTimeout(cities: string[], resolution: string) {
force_refresh: "false",
limit: String(Math.max(cities.length, CITY_DETAIL_BATCH_MAX_CITIES)),
resolution,
scope: "chart",
});
return fetch(`/api/cities/detail-batch?${params.toString()}`, {
headers: { Accept: "application/json" },
@@ -1193,6 +1176,13 @@ async function fetchHourlyForecastForCity(
if (cached) {
return cached.data;
}
} else {
const recentlyRefreshed = readHourlyCacheEntry(cacheKey, {
maxAgeMs: HOURLY_FORCE_REFRESH_DEDUP_MS,
});
if (recentlyRefreshed) {
return recentlyRefreshed.data;
}
}
const requestKey = options.ignoreCache ? `${city}:${resParam}:live` : `${city}:${resParam}`;
@@ -1208,17 +1198,6 @@ async function fetchHourlyForecastForCity(
return request;
}
function fetchCityDetailWithTimeout(city: string, resolution: string) {
const controller = new AbortController();
const timeoutId = globalThis.setTimeout(() => controller.abort(), HOURLY_DETAIL_REQUEST_TIMEOUT_MS);
return fetch(`/api/city/${encodeURIComponent(city)}/detail?depth=full&force_refresh=false&resolution=${resolution}`, {
headers: { Accept: "application/json" },
signal: controller.signal,
})
.catch(() => null)
.finally(() => globalThis.clearTimeout(timeoutId));
}
function shouldPollLiveChart({
city,
compact,
@@ -2459,6 +2438,7 @@ export {
MAX_HOURLY_DETAIL_CONCURRENT_REQUESTS,
HOURLY_DETAIL_REQUEST_TIMEOUT_MS,
HOURLY_CACHE_TTL_MS,
HOURLY_FORCE_REFRESH_DEDUP_MS,
_hourlyCache,
__readHourlyCacheEntryForTest,
resolveCityDetailFromBatch as __resolveCityDetailFromBatchForTest,
@@ -17,6 +17,7 @@ type LoadTerminalAuthProfileOptions = {
accessToken?: string | null,
options?: { preferSnapshot?: boolean },
) => Promise<TerminalAuthProfilePayload>;
timeoutMs?: number;
};
type SettledProfile =
@@ -87,11 +88,14 @@ export async function loadTerminalAuthProfile({
getSession,
hasSupabasePublicEnv,
loadAuthProfile,
timeoutMs = 6500,
}: LoadTerminalAuthProfileOptions) {
let resolvedAuthenticated = false;
let resolveAuthenticated:
| ((payload: TerminalAuthProfilePayload) => void)
| null = null;
let latestCookiePayload: TerminalAuthProfilePayload | null = null;
let latestBearerPayload: TerminalAuthProfilePayload | null = null;
const authenticatedProfile = new Promise<TerminalAuthProfilePayload>((resolve) => {
resolveAuthenticated = resolve;
@@ -105,6 +109,7 @@ export async function loadTerminalAuthProfile({
const cookieProfile = settleProfile(
loadAuthProfile(null, { preferSnapshot: true }).then((payload) => {
latestCookiePayload = payload;
resolveIfAuthenticated(payload);
return payload;
}),
@@ -120,6 +125,7 @@ export async function loadTerminalAuthProfile({
).trim();
if (!accessToken) return null;
const payload = await loadAuthProfile(accessToken, { preferSnapshot: true });
latestBearerPayload = payload;
resolveIfAuthenticated(payload);
return payload;
})(),
@@ -129,5 +135,20 @@ export async function loadTerminalAuthProfile({
([cookieResult, bearerResult]) => firstKnownProfile(cookieResult, bearerResult),
);
return Promise.race([authenticatedProfile, fallbackProfile]);
const timeoutProfile = new Promise<TerminalAuthProfilePayload>((resolve, reject) => {
if (!Number.isFinite(timeoutMs) || timeoutMs <= 0) return;
globalThis.setTimeout(() => {
if (latestBearerPayload) {
resolve(latestBearerPayload);
return;
}
if (latestCookiePayload) {
resolve(latestCookiePayload);
return;
}
reject(new Error("Terminal auth bootstrap timeout"));
}, timeoutMs);
});
return Promise.race([authenticatedProfile, fallbackProfile, timeoutProfile]);
}
+34 -3
View File
@@ -4,6 +4,10 @@ import { useEffect, useSyncExternalStore } from "react";
import { resolveBackendApiUrl } from "@/lib/backend-api";
const V1_EVENT_TYPE = "city_observation_patch.v1";
const SSE_SUBSCRIPTION_RECONNECT_DELAY_MS = 150;
const SSE_REPLAY_BASE_LIMIT = 60;
const SSE_REPLAY_EVENTS_PER_CITY = 24;
const SSE_REPLAY_MAX_LIMIT = 240;
export type CityPatch = {
type?: string;
@@ -38,6 +42,7 @@ const subscribedCities = new Map<string, number>();
let eventSource: EventSource | null = null;
let reconnectTimer: ReturnType<typeof setTimeout> | null = null;
let subscriptionReconnectTimer: ReturnType<typeof setTimeout> | null = null;
let reconnectAttempt = 0;
let patchVersion = 0;
let resyncVersion = 0;
@@ -74,6 +79,12 @@ function clearReconnectTimer() {
reconnectTimer = null;
}
function clearSubscriptionReconnectTimer() {
if (!subscriptionReconnectTimer) return;
clearTimeout(subscriptionReconnectTimer);
subscriptionReconnectTimer = null;
}
function buildSseUrl(baseUrl: string) {
const params = new URLSearchParams();
const cities = subscribedCityList();
@@ -83,12 +94,17 @@ function buildSseUrl(baseUrl: string) {
if (lastRevision > 0) {
params.set("since_revision", String(lastRevision));
}
params.set("replay_limit", "500");
params.set("replay_limit", String(resolveSseReplayLimit(cities.length)));
const query = params.toString();
return query ? `${baseUrl}?${query}` : baseUrl;
}
function resolveSseReplayLimit(cityCount: number) {
const requested = Math.max(1, cityCount) * SSE_REPLAY_EVENTS_PER_CITY;
return Math.max(SSE_REPLAY_BASE_LIMIT, Math.min(SSE_REPLAY_MAX_LIMIT, requested));
}
function currentConnectionKey() {
return `${useFallbackUrl ? "fallback" : "direct"}:${subscribedCityList().join("|")}:${lastRevision}`;
}
@@ -113,6 +129,7 @@ function closeEventSource() {
function reconnectNow() {
if (typeof window === "undefined") return;
clearReconnectTimer();
clearSubscriptionReconnectTimer();
closeEventSource();
connectSsePatches();
}
@@ -159,6 +176,7 @@ function connectSsePatches() {
function ensureSsePatchConnection() {
if (subscribedCities.size === 0) {
clearSubscriptionReconnectTimer();
closeEventSource();
clearReconnectTimer();
return;
@@ -167,6 +185,19 @@ function ensureSsePatchConnection() {
reconnectNow();
}
function scheduleSubscriptionReconnect() {
if (typeof window === "undefined") return;
if (subscribedCities.size === 0) {
ensureSsePatchConnection();
return;
}
clearSubscriptionReconnectTimer();
subscriptionReconnectTimer = setTimeout(() => {
subscriptionReconnectTimer = null;
ensureSsePatchConnection();
}, SSE_SUBSCRIPTION_RECONNECT_DELAY_MS);
}
function registerCitySubscription(city: string) {
const cityKey = normalizeCityKey(city);
if (!cityKey) return () => {};
@@ -174,7 +205,7 @@ function registerCitySubscription(city: string) {
const previousCount = subscribedCities.get(cityKey) ?? 0;
subscribedCities.set(cityKey, previousCount + 1);
if (previousCount === 0) {
ensureSsePatchConnection();
scheduleSubscriptionReconnect();
}
return () => {
@@ -184,7 +215,7 @@ function registerCitySubscription(city: string) {
} else {
subscribedCities.set(cityKey, nextCount);
}
ensureSsePatchConnection();
scheduleSubscriptionReconnect();
};
}
+19 -1
View File
@@ -28,4 +28,22 @@ export function buildForceRefreshProxyCachePolicy(
};
}
export const buildCityDetailProxyCachePolicy = buildForceRefreshProxyCachePolicy;
export function buildCityDetailProxyCachePolicy(
forceRefresh: string | null | undefined,
revalidateSeconds = 15,
): ProxyCachePolicy {
if (isForceRefreshValue(forceRefresh)) {
return {
fetchMode: "no-store",
responseCacheControl: "no-store, max-age=0",
};
}
return {
fetchMode: "revalidate",
responseCacheControl: `public, max-age=${revalidateSeconds}, s-maxage=${revalidateSeconds}, stale-while-revalidate=${Math.max(
revalidateSeconds * 3,
30,
)}`,
revalidateSeconds,
};
}
+4
View File
@@ -23,6 +23,10 @@ def test_backend_shared_timing_helper_avoids_sensitive_identity_fields():
def test_city_detail_batch_response_includes_backend_server_timing(monkeypatch):
city_api._CITY_DETAIL_BATCH_RESPONSE_CACHE.clear()
city_api._CITY_DETAIL_BATCH_RESPONSE_CACHE_TS.clear()
city_api._CITY_DETAIL_BATCH_RESPONSE_INFLIGHT.clear()
class FakeCache:
def get_city_cache(self, kind, city):
assert kind == "full"
+35 -1
View File
@@ -38,15 +38,49 @@ def test_scan_terminal_prewarm_is_lazy_by_default():
)
def test_scan_terminal_prewarm_only_runs_for_web_service(monkeypatch):
from web import app_factory
monkeypatch.setenv("POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED", "true")
monkeypatch.delenv("POLYWEATHER_SERVICE_ROLE", raising=False)
assert app_factory._scan_terminal_prewarm_enabled() is False
monkeypatch.setenv("POLYWEATHER_SERVICE_ROLE", "bot")
assert app_factory._scan_terminal_prewarm_enabled() is False
monkeypatch.setenv("POLYWEATHER_SERVICE_ROLE", "web")
assert app_factory._scan_terminal_prewarm_enabled() is True
def test_docker_compose_isolates_scan_terminal_prewarm_to_web_service():
compose = (ROOT / "docker-compose.yml").read_text(encoding="utf-8")
assert "POLYWEATHER_SERVICE_ROLE: web" in compose
assert "POLYWEATHER_SERVICE_ROLE: bot" in compose
assert "POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false'" in compose
def test_scan_terminal_backend_timeout_returns_before_next_proxy_abort():
import web.services.scan_terminal_config as scan_terminal_config
route_source = (
ROOT / "frontend" / "app" / "api" / "scan" / "terminal" / "route.ts"
).read_text(encoding="utf-8")
config_source = (
ROOT / "web" / "services" / "scan_terminal_config.py"
).read_text(encoding="utf-8")
assert 'POLYWEATHER_SCAN_TERMINAL_PROXY_TIMEOUT_MS || "40000"' in route_source
assert 'POLYWEATHER_SCAN_TERMINAL_PROXY_TIMEOUT_MS || "18000"' in route_source
assert '"POLYWEATHER_SCAN_TERMINAL_BUILD_TIMEOUT_SEC",\n 10,' in config_source
assert (
'"POLYWEATHER_SCAN_TERMINAL_PREWARM_PAYLOAD_TIMEOUT_SEC",\n 30,'
in config_source
)
assert scan_terminal_config.SCAN_TERMINAL_BUILD_TIMEOUT_SEC <= 30
assert (
scan_terminal_config.SCAN_TERMINAL_PREWARM_PAYLOAD_TIMEOUT_SEC
>= scan_terminal_config.SCAN_TERMINAL_BUILD_TIMEOUT_SEC
)
def test_probability_engine_uses_enriched_multi_model_snapshot():
+144 -1
View File
@@ -1,12 +1,84 @@
from web.scan_terminal_filters import normalize_scan_terminal_filters
from web import scan_terminal_cache
from web.scan_terminal_metar_gate import _apply_metar_gate_to_row
from web.scan_terminal_payloads import (
build_failed_scan_terminal_payload,
build_scan_terminal_snapshot_id,
build_stale_scan_terminal_payload,
compact_ranked_scan_rows_for_payload,
SCAN_PAYLOAD_DEFERRED_RUNWAY_POINTS,
SCAN_PAYLOAD_FULL_RUNWAY_HISTORY_ROWS,
)
from web.scan_terminal_ranker import build_ranked_scan_terminal_result
from web.scan_terminal_city_row import _build_quick_row
from web.routers.scan import router as scan_router
from web.scan_terminal_service import _scan_terminal_prewarm_filters
class _FakeRedis:
def __init__(self):
self.data = {}
def get(self, key):
return self.data.get(key)
def setex(self, key, _ttl, value):
self.data[key] = value
def test_scan_terminal_cache_hydrates_success_payload_from_redis(monkeypatch):
fake_redis = _FakeRedis()
monkeypatch.setenv("POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_ENABLED", "true")
monkeypatch.setattr(scan_terminal_cache, "_get_redis_client", lambda: fake_redis)
scan_terminal_cache._SCAN_TERMINAL_CACHE.clear()
filters = {"scan_mode": "tradable", "limit": 9}
payload = {
"generated_at": "2026-06-01T00:00:00Z",
"rows": [{"id": "row-1"}],
"summary": {"candidate_total": 1},
}
scan_terminal_cache.set_cached_scan_terminal_payload(filters, payload)
scan_terminal_cache._SCAN_TERMINAL_CACHE.clear()
entry = scan_terminal_cache.get_scan_terminal_cache_entry(filters)
cached = scan_terminal_cache.get_cached_scan_terminal_payload(filters, ttl_sec=3600)
assert entry["success_payload"]["rows"] == [{"id": "row-1"}]
assert cached["summary"]["candidate_total"] == 1
def test_scan_terminal_failure_state_preserves_redis_success_payload(monkeypatch):
fake_redis = _FakeRedis()
monkeypatch.setenv("POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_ENABLED", "true")
monkeypatch.setattr(scan_terminal_cache, "_get_redis_client", lambda: fake_redis)
scan_terminal_cache._SCAN_TERMINAL_CACHE.clear()
filters = {"scan_mode": "tradable", "limit": 9}
scan_terminal_cache.set_cached_scan_terminal_payload(
filters,
{
"generated_at": "2026-06-01T00:00:00Z",
"rows": [{"id": "row-1"}],
},
)
scan_terminal_cache._SCAN_TERMINAL_CACHE.clear()
scan_terminal_cache.set_scan_terminal_failure_state(filters, error_message="timeout")
scan_terminal_cache._SCAN_TERMINAL_CACHE.clear()
entry = scan_terminal_cache.get_scan_terminal_cache_entry(filters)
assert entry["success_payload"]["rows"] == [{"id": "row-1"}]
assert entry["last_error"] == "timeout"
def test_scan_terminal_prewarm_covers_default_api_limit():
limits = {filters["limit"] for filters in _scan_terminal_prewarm_filters()}
assert 25 in limits
assert 180 in limits
def test_scan_router_does_not_expose_terminal_ai_endpoint():
@@ -137,6 +209,78 @@ def test_scan_terminal_snapshot_id_is_stable_for_same_ranked_inputs():
assert first.startswith("scan-")
def test_scan_terminal_payload_slims_deferred_runway_history_rows():
runway_points = [
{"time": f"2026-05-31T{hour:02d}:00:00+00:00", "temp": 20 + hour}
for hour in range(24)
]
rows = [
{
"id": f"row-{index}",
"runway_plate_history": {"35R": list(runway_points)},
}
for index in range(SCAN_PAYLOAD_FULL_RUNWAY_HISTORY_ROWS + 2)
]
compacted = compact_ranked_scan_rows_for_payload(rows)
assert len(compacted[0]["runway_plate_history"]["35R"]) == len(runway_points)
assert (
len(
compacted[SCAN_PAYLOAD_FULL_RUNWAY_HISTORY_ROWS - 1][
"runway_plate_history"
]["35R"]
)
== len(runway_points)
)
assert (
len(
compacted[SCAN_PAYLOAD_FULL_RUNWAY_HISTORY_ROWS][
"runway_plate_history"
]["35R"]
)
== SCAN_PAYLOAD_DEFERRED_RUNWAY_POINTS
)
assert len(rows[SCAN_PAYLOAD_FULL_RUNWAY_HISTORY_ROWS]["runway_plate_history"]["35R"]) == len(
runway_points
)
def test_scan_terminal_quick_row_compacts_runway_history_for_list_payload():
raw_history = {
"35R": [
{"time": "2026-05-31T00:00:00+00:00", "temp": 22.11},
{"time": "2026-05-31T00:01:00+00:00", "temp": 22.22},
{"time": "2026-05-31T00:02:00+00:00", "temp": 22.33},
{"time": "2026-05-31T00:10:00+00:00", "temp": 23.44},
{"time": "2026-05-31T00:11:00+00:00", "temp": 23.55},
]
}
row = _build_quick_row(
city="shanghai",
data={
"display_name": "Shanghai",
"local_date": "2026-05-31",
"local_time": "2026-05-31T08:11:00+08:00",
"temp_symbol": "°C",
"current": {"temp": 22.3, "max_so_far": 23.0},
"risk": {"airport": "Shanghai Pudong", "level": "medium"},
"deb": {"prediction": 24.0},
"probabilities": {"distribution": []},
"multi_model": {},
"runway_plate_history": raw_history,
},
)
compact_history = row["runway_plate_history"]["35R"]
assert len(compact_history) == 2
assert compact_history[0]["temp"] == 22.3
assert compact_history[1]["temp"] == 23.6
assert len(str(row["runway_plate_history"])) < len(str(raw_history))
def test_metar_gate_vetoes_yes_when_observed_breaks_above_bucket():
row = {
"id": "yes-row",
@@ -158,4 +302,3 @@ def test_metar_gate_vetoes_yes_when_observed_breaks_above_bucket():
assert row["v4_metar_decision"] == "veto"
assert row["ai_decision"] == "veto"
assert "越过目标桶上沿" in row["ai_reason_zh"]
+119 -2
View File
@@ -129,8 +129,125 @@ def test_events_endpoint_emits_resync_when_replay_is_incomplete(monkeypatch):
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
assert sse_router._bounded_replay_limit(500) == 60
assert sse_router._bounded_replay_limit(5000) == 60
assert sse_router._bounded_replay_limit(500, city_count=5) == 120
assert sse_router._bounded_replay_limit(500, city_count=20) == 240
assert sse_router._bounded_replay_limit(25, city_count=5) == 25
def test_replay_gap_direct_resync_policy():
assert (
sse_router._should_direct_resync(
since_revision=1000,
latest_revision=1200,
limit=60,
)
is False
)
assert (
sse_router._should_direct_resync(
since_revision=1000,
latest_revision=1300,
limit=60,
)
is True
)
assert (
sse_router._should_direct_resync(
since_revision=0,
latest_revision=1300,
limit=60,
)
is False
)
def test_legacy_high_replay_limit_is_clamped_by_city_count(monkeypatch):
captured = {}
class FakeStore:
def latest_revision(self):
return 44
def replay_events(self, *, cities, since_revision, limit):
captured["cities"] = cities
captured["limit"] = limit
return []
def replay_requires_resync(self, *, cities, since_revision, replay_count, limit):
captured["resync_limit"] = 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}
)
monkeypatch.setattr(sse_router, "event_store", FakeStore())
monkeypatch.setattr(sse_router.sse_manager, "event_stream", finite_stream)
response = TestClient(app).get(
"/api/events?cities=ankara,buenos%20aires,istanbul,jeddah,seoul"
"&since_revision=42&replay_limit=500"
)
assert response.status_code == 200
assert captured["cities"] == {"ankara", "buenos aires", "istanbul", "jeddah", "seoul"}
assert captured["limit"] == 120
assert captured["resync_limit"] == 120
def test_stale_client_gets_direct_resync_without_replay_scan(monkeypatch):
calls = {"replay": 0, "requires_resync": 0}
class FakeStore:
def latest_revision(self):
return 2000
def replay_events(self, *, cities, since_revision, limit):
calls["replay"] += 1
return []
def replay_requires_resync(self, *, cities, since_revision, replay_count, limit):
calls["requires_resync"] += 1
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}
)
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=ankara,buenos%20aires,istanbul,jeddah,seoul"
"&since_revision=100&replay_limit=500"
)
assert response.status_code == 200
assert calls == {"replay": 0, "requires_resync": 0}
events = _decode_sse_events(response.text)
assert events[-1]["type"] == "resync_required"
assert events[-1]["reason"] == "replay_gap_too_large"
assert events[-1]["latest_revision"] == 2000
def test_ingest_patch_uses_external_fanout_without_direct_broadcast(monkeypatch):
+444
View File
@@ -31,6 +31,27 @@ def test_healthz_returns_ok_shape():
assert 'cities_count' in payload
def test_healthz_keeps_liveness_200_when_db_health_is_degraded(monkeypatch):
from web.services import system_api
monkeypatch.setattr(
system_api,
"build_health_payload",
lambda: {
"status": "degraded",
"time_utc": "2026-05-30T00:00:00+00:00",
"db": {"ok": False, "error": "database is locked"},
"state_storage_mode": "sqlite",
"cities_count": 50,
},
)
response = client.get("/healthz")
assert response.status_code == 200
assert response.json()["status"] == "degraded"
def test_system_status_returns_summary_shape():
response = client.get('/api/system/status')
assert response.status_code == 200
@@ -315,6 +336,15 @@ def test_ops_billing_risk_surfaces_trial_payment_referral_and_points(monkeypatch
DBManager,
"list_app_analytics_events",
lambda self, limit=20000, since_iso=None: [
{
"id": 10,
"event_type": "login_start",
"user_id": None,
"client_id": "c1",
"session_id": "session-gap",
"created_at": recent,
"payload": {"mode": "signup"},
},
{
"id": 11,
"event_type": "signup_success",
@@ -419,6 +449,108 @@ def test_ops_billing_risk_does_not_flag_signup_when_backend_trial_exists(monkeyp
assert not any(issue["category"] == "signup_trial" for issue in payload["issues"])
def test_ops_billing_risk_ignores_account_visit_without_signup_intent(monkeypatch):
from src.database.db_manager import DBManager
recent = datetime.now(timezone.utc).isoformat()
monkeypatch.setattr(ops_api.legacy_routes, "_require_ops_admin", lambda request: {"email": "ops@example.com"})
monkeypatch.setattr(ops_api, "_supabase_rest_rows", lambda table, params, *, timeout=10: [])
monkeypatch.setattr(
DBManager,
"list_app_analytics_events",
lambda self, limit=20000, since_iso=None: [
{
"id": 61,
"event_type": "signup_success",
"user_id": "returning-no-trial-user",
"client_id": "client-return",
"session_id": "session-return",
"created_at": recent,
"payload": {
"entry": "account_center",
"user_id": "returning-no-trial-user",
},
}
],
)
monkeypatch.setattr(
DBManager,
"list_payment_audit_events",
lambda self, limit=50, event_type=None: [],
)
payload = ops_api.get_ops_billing_risk(None, days=30, limit=20)
assert payload["summary"]["trial_gaps"] == 0
assert not any(issue["category"] == "signup_trial" for issue in payload["issues"])
def test_ops_billing_risk_treats_expired_signup_trial_subscription_as_backend_evidence(monkeypatch):
from src.database.db_manager import DBManager
now = datetime.now(timezone.utc)
recent = now.isoformat()
expired = (now - timedelta(days=2)).isoformat()
def fake_supabase_rows(table, params, *, timeout=10):
if table == "subscriptions" and params.get("or") == "(source.eq.signup_trial,plan_code.eq.signup_trial_3d)":
return [
{
"id": 81,
"user_id": "expired-trial-user",
"plan_code": "signup_trial_3d",
"source": "signup_trial",
"status": "expired",
"starts_at": (now - timedelta(days=5)).isoformat(),
"expires_at": expired,
"created_at": (now - timedelta(days=5)).isoformat(),
"updated_at": expired,
}
]
return []
monkeypatch.setattr(ops_api.legacy_routes, "_require_ops_admin", lambda request: {"email": "ops@example.com"})
monkeypatch.setattr(ops_api, "_supabase_rest_rows", fake_supabase_rows)
monkeypatch.setattr(
DBManager,
"list_app_analytics_events",
lambda self, limit=20000, since_iso=None: [
{
"id": 80,
"event_type": "login_start",
"user_id": None,
"client_id": "client-expired",
"session_id": "session-expired",
"created_at": recent,
"payload": {"mode": "signup"},
},
{
"id": 82,
"event_type": "signup_success",
"user_id": "expired-trial-user",
"client_id": "client-expired",
"session_id": "session-expired",
"created_at": recent,
"payload": {
"entry": "account_center",
"user_id": "expired-trial-user",
},
},
],
)
monkeypatch.setattr(
DBManager,
"list_payment_audit_events",
lambda self, limit=50, event_type=None: [],
)
payload = ops_api.get_ops_billing_risk(None, days=30, limit=20)
assert payload["summary"]["trial_gaps"] == 0
assert not any(issue["category"] == "signup_trial" for issue in payload["issues"])
def test_ops_payment_incidents_expose_top_level_reason_and_filters_resolved(monkeypatch):
from src.database.db_manager import DBManager
@@ -567,6 +699,119 @@ def test_city_detail_batch_endpoint_builds_multiple_cached_details(monkeypatch):
assert sorted(calls) == [("paris", "10m"), ("shanghai", "10m")]
def test_city_detail_batch_chart_scope_returns_only_chart_fields(monkeypatch):
monkeypatch.setattr(city_api.legacy_routes, "_assert_entitlement", lambda request: None)
monkeypatch.setattr(city_api.legacy_routes, "_normalize_city_or_404", lambda name: name.strip().lower())
monkeypatch.setattr(
city_api.legacy_routes,
"_city_cache_is_fresh",
lambda entry, ttl: True,
)
monkeypatch.setattr(
city_api.legacy_routes,
"_overlay_latest_wunderground_current",
lambda city, payload: payload,
)
class FakeCache:
def get_city_cache(self, kind, city):
assert kind == "full"
return {
"payload": {
"name": city,
"display_name": city.title(),
"local_date": "2026-05-30",
"local_time": "15:20",
"temp_symbol": "°C",
"current": {
"temp": 20.0,
"settlement_source": "metar",
"settlement_source_label": "METAR",
},
"hourly": {"times": ["2026-05-30T00:00:00Z"], "temps": [20.0]},
"forecast": {
"today_high": 22.0,
"daily": [{"date": "2026-05-30", "max_temp": 22.0}],
},
"multi_model": {
"hourly_times": ["15:00"],
"hourly_forecasts": {"ECMWF": [21.0]},
},
"deb": {"prediction": 21.5, "hourly_path": {"times": ["15:00"], "temps": [21.5]}},
"probabilities": {"mu": 21.4, "distribution": [{"value": 21, "probability": 0.4}]},
"runway_plate_history": {"01/19": [{"time": "2026-05-30T15:20:00Z", "temp": 20.1}]},
"airport_current": {"temp": 20.0},
"airport_primary": {"temp": 20.0},
"airport_primary_today_obs": [["15:20", 20.0]],
"wunderground_current": {"max_so_far": 20.5},
"settlement_station": {"settlement_station_label": "Station"},
"amos": {"runway_obs": {"point_temperatures": []}},
"metar_today_obs": [{"time": "15:20", "temp": 20.0}],
"settlement_today_obs": [],
"dynamic_commentary": {"summary": "large text"},
"official_nearby": [{"name": "unused"}],
"taf": {"raw": "unused"},
"ai_analysis": "unused",
}
}
def build_detail(_data, _market_slug, _target_date, _resolution):
raise AssertionError("chart scope must not build the full city detail payload")
monkeypatch.setattr(city_api.legacy_routes, "_CACHE_DB", FakeCache())
monkeypatch.setattr(city_api.legacy_routes, "_build_city_detail_payload", build_detail)
response = client.get("/api/cities/detail-batch?cities=Paris&resolution=10m&scope=chart")
assert response.status_code == 200
detail = response.json()["details"]["paris"]
assert detail["timeseries"]["hourly"]["temps"] == [20.0]
assert detail["models_hourly"]["curves"]["ECMWF"] == [21.0]
assert detail["deb"]["hourly_path"]["temps"] == [21.5]
assert detail["airport_primary_today_obs"] == [["15:20", 20.0]]
assert "dynamic_commentary" not in detail
assert "official_nearby" not in detail
assert "taf" not in detail
assert "ai_analysis" not in detail
def test_chart_detail_payload_avoids_threadpool_queue_and_reuses_short_cache(monkeypatch):
import asyncio
build_calls = 0
async def fail_threadpool(*_args, **_kwargs):
raise AssertionError("chart payload should not wait behind the shared threadpool")
def build_chart_detail(data, resolution):
nonlocal build_calls
build_calls += 1
return {
"city": data["city"],
"resolution": resolution,
"hourly": data["hourly"],
}
city_api._CITY_CHART_DETAIL_PAYLOAD_CACHE.clear()
city_api._CITY_CHART_DETAIL_PAYLOAD_CACHE_TS.clear()
monkeypatch.setenv("POLYWEATHER_CITY_DETAIL_PAYLOAD_CACHE_TTL_SEC", "20")
monkeypatch.setattr(city_api, "run_in_threadpool", fail_threadpool)
monkeypatch.setattr(city_api.legacy_routes, "_build_city_chart_detail_payload", build_chart_detail)
data = {
"city": "paris",
"updated_at": "2026-05-30T15:00:00Z",
"hourly": {"times": ["2026-05-30T15:00:00Z"], "temps": [20.0]},
}
first = asyncio.run(city_api._build_city_chart_detail_payload(data, "10m"))
second = asyncio.run(city_api._build_city_chart_detail_payload(data, "10m"))
assert first == second
assert first["resolution"] == "10m"
assert build_calls == 1
def test_city_detail_batch_endpoint_limits_backend_concurrency(monkeypatch):
import asyncio
@@ -656,6 +901,109 @@ def test_city_detail_batch_returns_completed_details_when_one_city_is_slow(monke
assert "slow" not in completed
def test_city_detail_batch_response_cache_keeps_entitlement_check(monkeypatch):
import asyncio
entitlement_calls = 0
build_calls = 0
async def build_batch_item(city, **kwargs):
nonlocal build_calls
build_calls += 1
return city, {
"city": city,
"hourly": {"times": ["2026-05-30T00:00:00Z"], "temps": [20.0]},
"resolution": kwargs.get("resolution"),
}
def assert_entitlement(request):
nonlocal entitlement_calls
entitlement_calls += 1
city_api._CITY_DETAIL_BATCH_RESPONSE_CACHE.clear()
city_api._CITY_DETAIL_BATCH_RESPONSE_CACHE_TS.clear()
city_api._CITY_DETAIL_BATCH_RESPONSE_INFLIGHT.clear()
monkeypatch.setenv("POLYWEATHER_CITY_DETAIL_BATCH_RESPONSE_CACHE_TTL_SEC", "20")
monkeypatch.setattr(city_api.legacy_routes, "_assert_entitlement", assert_entitlement)
monkeypatch.setattr(city_api.legacy_routes, "_normalize_city_or_404", lambda name: name.strip().lower())
monkeypatch.setattr(city_api, "_build_city_detail_batch_item_async", build_batch_item)
first = asyncio.run(
city_api.get_city_detail_batch_payload(
object(),
cities="Paris",
resolution="10m",
limit=12,
)
)
second = asyncio.run(
city_api.get_city_detail_batch_payload(
object(),
cities="Paris",
resolution="10m",
limit=12,
)
)
assert first == second
assert first["details"]["paris"]["resolution"] == "10m"
assert entitlement_calls == 2
assert build_calls == 1
def test_concurrent_city_detail_batch_requests_share_inflight_response(monkeypatch):
import asyncio
entitlement_calls = 0
build_calls = 0
async def build_batch_item(city, **kwargs):
nonlocal build_calls
build_calls += 1
await asyncio.sleep(0.02)
return city, {
"city": city,
"hourly": {"times": ["2026-05-30T00:00:00Z"], "temps": [20.0]},
"resolution": kwargs.get("resolution"),
}
def assert_entitlement(request):
nonlocal entitlement_calls
entitlement_calls += 1
city_api._CITY_DETAIL_BATCH_RESPONSE_CACHE.clear()
city_api._CITY_DETAIL_BATCH_RESPONSE_CACHE_TS.clear()
city_api._CITY_DETAIL_BATCH_RESPONSE_INFLIGHT.clear()
monkeypatch.setenv("POLYWEATHER_CITY_DETAIL_BATCH_RESPONSE_CACHE_TTL_SEC", "20")
monkeypatch.setattr(city_api.legacy_routes, "_assert_entitlement", assert_entitlement)
monkeypatch.setattr(city_api.legacy_routes, "_normalize_city_or_404", lambda name: name.strip().lower())
monkeypatch.setattr(city_api, "_build_city_detail_batch_item_async", build_batch_item)
async def run_requests():
return await asyncio.gather(
city_api.get_city_detail_batch_payload(
object(),
cities="Paris",
resolution="10m",
limit=12,
),
city_api.get_city_detail_batch_payload(
object(),
cities="Paris",
resolution="10m",
limit=12,
),
)
first, second = asyncio.run(run_requests())
assert first == second
assert entitlement_calls == 2
assert build_calls == 1
def test_concurrent_city_detail_requests_share_same_full_cache_refresh(monkeypatch):
import asyncio
@@ -2019,6 +2367,102 @@ def test_scan_terminal_service_returns_stale_payload_after_failed_refresh(monkey
assert stale["stale_reason"] == "upstream 504"
def test_scan_terminal_timeout_does_not_replace_better_cached_snapshot(monkeypatch):
import time
filters = {"scan_mode": "tradable", "limit": 5}
normalized_filters = scan_terminal_service._normalize_scan_terminal_filters(filters)
scan_terminal_cache._SCAN_TERMINAL_CACHE.clear()
previous_payload = {
"generated_at": "2026-05-31T00:00:00Z",
"snapshot_id": "scan-existing",
"filters": normalized_filters,
"summary": {"candidate_total": 2, "visible_count": 2},
"top_signal": {"id": "old-1"},
"rows": [{"id": "old-1"}, {"id": "old-2"}],
"status": "ready",
"stale": False,
"stale_reason": None,
"last_success_at": None,
"last_failed_at": None,
}
scan_terminal_cache.set_cached_scan_terminal_payload(
normalized_filters,
previous_payload,
)
monkeypatch.setattr(
scan_terminal_service,
"CITIES",
{"fast": {"tz": 0}, "slow": {"tz": 0}},
)
monkeypatch.setattr(scan_terminal_service, "SCAN_TERMINAL_BUILD_TIMEOUT_SEC", 0.01)
monkeypatch.setattr(scan_terminal_service, "SCAN_TERMINAL_MAX_WORKERS", 2)
def _scan_city(city_name, *_args, **_kwargs):
if city_name == "slow":
time.sleep(0.05)
return {
"city": city_name,
"candidate_total": 1,
"primary_scores": [80.0],
"rows": [
{
"id": f"{city_name}-row",
"market_key": f"{city_name}-market",
"edge_percent": 4.0,
"final_score": 80.0,
"volume": 1000,
}
],
}
monkeypatch.setattr(scan_terminal_service, "_scan_city_terminal_rows", _scan_city)
stale = scan_terminal_service.build_scan_terminal_payload(filters, force_refresh=True)
assert stale["status"] == "stale"
assert stale["stale"] is True
assert [row["id"] for row in stale["rows"]] == ["old-1", "old-2"]
assert stale["stale_reason"].startswith("scan terminal build timed out")
cached = scan_terminal_cache.get_cached_scan_terminal_payload(
normalized_filters,
ttl_sec=3600,
)
assert [row["id"] for row in cached["rows"]] == ["old-1", "old-2"]
def test_scan_terminal_prewarm_builds_default_terminal_payload(monkeypatch):
calls = []
def _fake_build(filters, *, force_refresh=False, timeout_sec=None):
calls.append((dict(filters), force_refresh, timeout_sec))
return {"rows": []}
monkeypatch.setattr(
scan_terminal_service,
"_build_scan_terminal_payload_uncached",
_fake_build,
)
assert scan_terminal_service._warm_scan_terminal_payloads() == 2
assert {filters["limit"] for filters, _, _ in calls} == {25, 180}
filters, force_refresh, timeout_sec = calls[0]
assert all(force_refresh is False for _, force_refresh, _ in calls)
assert all(
timeout_sec == scan_terminal_service.SCAN_TERMINAL_PREWARM_PAYLOAD_TIMEOUT_SEC
for _, _, timeout_sec in calls
)
assert filters["scan_mode"] == "tradable"
assert filters["min_price"] == 0.05
assert filters["max_price"] == 0.95
assert filters["min_edge_pct"] == 2.0
assert filters["min_liquidity"] == 500.0
assert filters["market_type"] == "maxtemp"
assert filters["time_range"] == "today"
assert filters["limit"] == 25
def test_scan_terminal_service_returns_failed_without_success_snapshot(monkeypatch):
filters = {"scan_mode": "tradable", "limit": 5}
scan_terminal_cache._SCAN_TERMINAL_CACHE.clear()
+8
View File
@@ -37,6 +37,7 @@ from src.data_collection.city_registry import ALIASES, CITY_REGISTRY
from src.data_collection.city_time import get_city_utc_offset_seconds
from src.database.runtime_state import IntradayPathSnapshotRepository
from web.services.city_payloads import (
build_city_chart_detail_payload as _city_chart_payload_detail,
build_city_detail_payload as _city_payload_detail,
build_city_summary_payload as _city_payload_summary
)
@@ -2312,6 +2313,13 @@ def _build_city_detail_payload(
)
def _build_city_chart_detail_payload(
data: Dict[str, Any],
resolution: Optional[str] = "10m",
) -> Dict[str, Any]:
return _city_chart_payload_detail(data, resolution=resolution)
# ──────────────────────────────────────────────────────────
# Routes
+6 -1
View File
@@ -24,10 +24,15 @@ from web.scan_terminal_service import start_scan_terminal_prewarm
_ROUTES_REGISTERED_FLAG = "_polyweather_routes_registered"
def _service_role() -> str:
return str(os.getenv("POLYWEATHER_SERVICE_ROLE") or "").strip().lower()
def _scan_terminal_prewarm_enabled() -> bool:
return str(
enabled = str(
os.getenv("POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED") or "false"
).strip().lower() in {"1", "true", "yes", "on"}
return enabled and _service_role() in {"web", "api", "backend"}
def create_app() -> FastAPI:
+1 -1
View File
@@ -511,7 +511,7 @@ def _is_excluded_model_name(model_name: str) -> bool:
def _sqlite_health() -> Dict[str, Any]:
try:
with sqlite3.connect(_account_db.db_path) as conn:
with sqlite3.connect(_account_db.db_path, timeout=0.05) as conn:
conn.execute("SELECT 1").fetchone()
return {"ok": True, "db_path": _account_db.db_path}
except Exception as exc:
+2
View File
@@ -115,6 +115,7 @@ async def city_detail_batch(
market_slug: Optional[str] = None,
target_date: Optional[str] = None,
resolution: Optional[str] = "10m",
scope: Optional[str] = "full",
limit: int = 12,
):
payload = await get_city_detail_batch_payload(
@@ -124,6 +125,7 @@ async def city_detail_batch(
market_slug=market_slug,
target_date=target_date,
resolution=resolution,
scope=scope,
limit=limit,
)
attach_server_timing_header(response, request, "city_detail_batch_server_timing")
+52 -14
View File
@@ -19,6 +19,10 @@ router = APIRouter(tags=["events"])
event_store = create_realtime_event_store()
_live_subscription_lock = threading.Lock()
_live_subscription_started = False
SSE_REPLAY_BASE_LIMIT = 60
SSE_REPLAY_EVENTS_PER_CITY = 24
SSE_REPLAY_MAX_LIMIT = 240
SSE_REPLAY_DIRECT_RESYNC_FACTOR = 3
def _parse_cities_param(cities: str) -> Set[str]:
@@ -29,12 +33,33 @@ def _parse_cities_param(cities: str) -> Set[str]:
}
def _bounded_replay_limit(value: int) -> int:
def _recommended_replay_limit(city_count: int) -> int:
requested = max(1, int(city_count or 0)) * SSE_REPLAY_EVENTS_PER_CITY
return max(SSE_REPLAY_BASE_LIMIT, min(SSE_REPLAY_MAX_LIMIT, requested))
def _bounded_replay_limit(value: int, *, city_count: int = 0) -> int:
try:
limit = int(value)
except (TypeError, ValueError):
limit = 500
return max(1, min(MAX_REPLAY_LIMIT, limit))
limit = _recommended_replay_limit(city_count)
route_limit = _recommended_replay_limit(city_count)
return max(1, min(MAX_REPLAY_LIMIT, route_limit, limit))
def _should_direct_resync(
*,
since_revision: int,
latest_revision: int,
limit: int,
) -> bool:
since = max(0, int(since_revision or 0))
latest = max(0, int(latest_revision or 0))
if since <= 0 or latest <= since:
return False
gap = latest - since
threshold = max(SSE_REPLAY_MAX_LIMIT, max(1, int(limit or 1)) * SSE_REPLAY_DIRECT_RESYNC_FACTOR)
return gap > threshold
def _ensure_live_subscription() -> None:
@@ -65,7 +90,7 @@ async def sse_events(
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)
limit = _bounded_replay_limit(replay_limit, city_count=len(city_set))
_ensure_live_subscription()
latest_revision = event_store.latest_revision()
replay_events = []
@@ -73,23 +98,36 @@ async def sse_events(
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),
since = max(0, int(since_revision))
if _should_direct_resync(
since_revision=since,
latest_revision=latest_revision,
limit=limit,
):
resync_event = {
"type": "resync_required",
"reason": "replay_window_exceeded",
"reason": "replay_gap_too_large",
"latest_revision": latest_revision,
"ts": int(time.time() * 1000),
}
else:
replay_events = event_store.replay_events(
cities=city_set,
since_revision=since,
limit=limit,
)
if event_store.replay_requires_resync(
cities=city_set,
since_revision=since,
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",
+133 -27
View File
@@ -1,6 +1,8 @@
from __future__ import annotations
import json
import hashlib
import os
import threading
import time
from datetime import datetime
@@ -9,39 +11,140 @@ from typing import Any, Dict, Optional
_SCAN_TERMINAL_CACHE_LOCK = threading.Lock()
_SCAN_TERMINAL_CACHE: Dict[str, Dict[str, Any]] = {}
_SCAN_TERMINAL_REFRESHING: set[str] = set()
_SCAN_TERMINAL_REDIS_CLIENT_LOCK = threading.Lock()
_SCAN_TERMINAL_REDIS_CLIENT: Any = None
_SCAN_TERMINAL_REDIS_UNAVAILABLE = False
def scan_terminal_cache_key(filters: Dict[str, Any]) -> str:
return json.dumps(filters, ensure_ascii=True, sort_keys=True)
def _truthy_env(name: str, *, default: bool = False) -> bool:
value = os.getenv(name)
if value is None:
return default
return str(value).strip().lower() not in ("", "0", "false", "no", "off")
def _redis_cache_enabled() -> bool:
return _truthy_env(
"POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_ENABLED",
default=bool(os.getenv("POLYWEATHER_REDIS_URL")),
)
def _redis_cache_ttl_sec() -> int:
try:
value = int(os.getenv("POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_TTL_SEC", "21600"))
except Exception:
value = 21600
return max(600, min(value, 86400))
def _redis_cache_prefix() -> str:
return os.getenv(
"POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_PREFIX",
"polyweather:scan_terminal:v1:",
)
def _redis_entry_key(cache_key: str) -> str:
digest = hashlib.sha256(cache_key.encode("utf-8")).hexdigest()
return f"{_redis_cache_prefix()}{digest}"
def _get_redis_client() -> Any:
global _SCAN_TERMINAL_REDIS_CLIENT, _SCAN_TERMINAL_REDIS_UNAVAILABLE
if not _redis_cache_enabled() or _SCAN_TERMINAL_REDIS_UNAVAILABLE:
return None
with _SCAN_TERMINAL_REDIS_CLIENT_LOCK:
if _SCAN_TERMINAL_REDIS_CLIENT is not None:
return _SCAN_TERMINAL_REDIS_CLIENT
try:
import redis # type: ignore
url = os.getenv("POLYWEATHER_REDIS_URL") or "redis://127.0.0.1:6379/0"
client = redis.Redis.from_url(
url,
socket_timeout=float(os.getenv("POLYWEATHER_REDIS_SOCKET_TIMEOUT_SECONDS", "2")),
socket_connect_timeout=float(
os.getenv("POLYWEATHER_REDIS_SOCKET_CONNECT_TIMEOUT_SECONDS", "1")
),
health_check_interval=30,
)
client.ping()
_SCAN_TERMINAL_REDIS_CLIENT = client
return client
except Exception:
_SCAN_TERMINAL_REDIS_UNAVAILABLE = True
return None
def _read_redis_cache_entry(cache_key: str) -> Optional[Dict[str, Any]]:
client = _get_redis_client()
if client is None:
return None
try:
raw = client.get(_redis_entry_key(cache_key))
if not raw:
return None
if isinstance(raw, bytes):
raw = raw.decode("utf-8")
entry = json.loads(str(raw))
return dict(entry) if isinstance(entry, dict) else None
except Exception:
return None
def _write_redis_cache_entry(cache_key: str, entry: Dict[str, Any]) -> None:
client = _get_redis_client()
if client is None:
return
try:
client.setex(
_redis_entry_key(cache_key),
_redis_cache_ttl_sec(),
json.dumps(entry, ensure_ascii=False, separators=(",", ":")),
)
except Exception:
return
def get_cached_scan_terminal_payload(
filters: Dict[str, Any],
*,
ttl_sec: int,
) -> Optional[Dict[str, Any]]:
cache_key = scan_terminal_cache_key(filters)
now = time.time()
with _SCAN_TERMINAL_CACHE_LOCK:
cached = _SCAN_TERMINAL_CACHE.get(cache_key)
if not cached:
return None
cached_at = float(cached.get("t") or 0.0)
if now - cached_at >= float(ttl_sec):
return None
payload = cached.get("payload")
if not isinstance(payload, dict):
return None
return dict(payload)
cached = get_scan_terminal_cache_entry(filters)
if not cached:
return None
cached_at = float(cached.get("t") or 0.0)
if now - cached_at >= float(ttl_sec):
return None
payload = cached.get("payload")
if not isinstance(payload, dict):
return None
return dict(payload)
def get_scan_terminal_cache_entry(filters: Dict[str, Any]) -> Optional[Dict[str, Any]]:
cache_key = scan_terminal_cache_key(filters)
with _SCAN_TERMINAL_CACHE_LOCK:
cached = _SCAN_TERMINAL_CACHE.get(cache_key)
if not isinstance(cached, dict):
return None
return dict(cached)
if isinstance(cached, dict):
return dict(cached)
redis_entry = _read_redis_cache_entry(cache_key)
if not redis_entry:
return None
with _SCAN_TERMINAL_CACHE_LOCK:
_SCAN_TERMINAL_CACHE[cache_key] = dict(redis_entry)
return dict(redis_entry)
def set_cached_scan_terminal_payload(
@@ -51,15 +154,17 @@ def set_cached_scan_terminal_payload(
cache_key = scan_terminal_cache_key(filters)
existing = get_scan_terminal_cache_entry(filters) or {}
now = time.time()
entry = {
"t": now,
"payload": dict(payload),
"success_t": now,
"success_payload": dict(payload),
"last_error": existing.get("last_error"),
"last_failed_at": existing.get("last_failed_at"),
}
with _SCAN_TERMINAL_CACHE_LOCK:
_SCAN_TERMINAL_CACHE[cache_key] = {
"t": now,
"payload": dict(payload),
"success_t": now,
"success_payload": dict(payload),
"last_error": existing.get("last_error"),
"last_failed_at": existing.get("last_failed_at"),
}
_SCAN_TERMINAL_CACHE[cache_key] = dict(entry)
_write_redis_cache_entry(cache_key, entry)
def set_scan_terminal_failure_state(
@@ -68,11 +173,12 @@ def set_scan_terminal_failure_state(
error_message: str,
) -> None:
cache_key = scan_terminal_cache_key(filters)
existing = get_scan_terminal_cache_entry(filters) or {}
existing["last_error"] = error_message
existing["last_failed_at"] = datetime.utcnow().isoformat() + "Z"
with _SCAN_TERMINAL_CACHE_LOCK:
existing = _SCAN_TERMINAL_CACHE.get(cache_key) or {}
existing["last_error"] = error_message
existing["last_failed_at"] = datetime.utcnow().isoformat() + "Z"
_SCAN_TERMINAL_CACHE[cache_key] = existing
_SCAN_TERMINAL_CACHE[cache_key] = dict(existing)
_write_redis_cache_entry(cache_key, existing)
def mark_scan_terminal_refreshing(filters: Dict[str, Any]) -> bool:
+22 -2
View File
@@ -11,6 +11,22 @@ from web.scan_terminal_filters import (
market_region_from_tz_offset as _market_region_from_tz_offset,
safe_int as _safe_int,
)
from web.services.city_payloads import aggregate_runway_history
SCAN_ROW_RUNWAY_HISTORY_RESOLUTION = "10m"
SCAN_ROW_MAX_RUNWAY_POINTS = 144
def _compact_runway_plate_history_for_scan(raw_history: Any) -> Dict[str, List[Dict[str, Any]]]:
if not isinstance(raw_history, dict) or not raw_history:
return {}
compacted = aggregate_runway_history(raw_history, SCAN_ROW_RUNWAY_HISTORY_RESOLUTION)
return {
str(runway): points[-SCAN_ROW_MAX_RUNWAY_POINTS:]
for runway, points in compacted.items()
if isinstance(points, list) and points
}
def _resolve_time_range_dates(data: Dict[str, Any], time_range: str) -> List[str]:
@@ -132,7 +148,9 @@ def _build_terminal_row(
"amos": data.get("amos") or None,
"top_buckets": scan.get("top_buckets") or [],
"all_buckets": scan.get("all_buckets") or [],
"runway_plate_history": data.get("runway_plate_history") or {},
"runway_plate_history": _compact_runway_plate_history_for_scan(
data.get("runway_plate_history")
),
}
@@ -232,7 +250,9 @@ def _build_quick_row(
"is_primary_signal": True,
"accepting_orders": False,
"row_id": row_id,
"runway_plate_history": data.get("runway_plate_history") or {},
"runway_plate_history": _compact_runway_plate_history_for_scan(
data.get("runway_plate_history")
),
}
# Compute a simple edge: model top probability vs neutral
best_model_prob = max(
+42
View File
@@ -5,6 +5,48 @@ import json
from datetime import datetime
from typing import Any, Dict, List, Optional
SCAN_PAYLOAD_FULL_RUNWAY_HISTORY_ROWS = 9
SCAN_PAYLOAD_DEFERRED_RUNWAY_POINTS = 12
def _compact_runway_history_points(
history: Any,
*,
max_points: int,
) -> Dict[str, List[Dict[str, Any]]]:
if not isinstance(history, dict) or max_points <= 0:
return {}
compacted: Dict[str, List[Dict[str, Any]]] = {}
for runway, points in history.items():
if not isinstance(points, list) or not points:
continue
compacted[str(runway)] = points[-max_points:]
return compacted
def compact_ranked_scan_rows_for_payload(
rows: List[Dict[str, Any]],
*,
full_history_rows: int = SCAN_PAYLOAD_FULL_RUNWAY_HISTORY_ROWS,
deferred_runway_points: int = SCAN_PAYLOAD_DEFERRED_RUNWAY_POINTS,
) -> List[Dict[str, Any]]:
compacted_rows: List[Dict[str, Any]] = []
for index, row in enumerate(rows):
if index < full_history_rows:
compacted_rows.append(row)
continue
history = row.get("runway_plate_history")
if not isinstance(history, dict) or not history:
compacted_rows.append(row)
continue
next_row = dict(row)
next_row["runway_plate_history"] = _compact_runway_history_points(
history,
max_points=deferred_runway_points,
)
compacted_rows.append(next_row)
return compacted_rows
def build_scan_terminal_snapshot_id(
filters: Dict[str, Any],
+99 -5
View File
@@ -16,6 +16,7 @@ from web.services.scan_terminal_config import (
SCAN_TERMINAL_BUILD_TIMEOUT_SEC,
SCAN_TERMINAL_MAX_WORKERS,
SCAN_TERMINAL_PAYLOAD_TTL_SEC,
SCAN_TERMINAL_PREWARM_PAYLOAD_TIMEOUT_SEC,
)
from src.data_collection.city_registry import ALIASES
from web.scan_terminal_cache import (
@@ -34,6 +35,7 @@ from web.scan_terminal_payloads import (
build_failed_scan_terminal_payload,
build_scan_terminal_snapshot_id,
build_stale_scan_terminal_payload,
compact_ranked_scan_rows_for_payload,
)
from web.scan_terminal_ranker import build_ranked_scan_terminal_result
def _normalize_locale(value: Any) -> str:
@@ -47,6 +49,35 @@ def _normalize_city_key(value: Any) -> str:
return ALIASES.get(text, text)
def _rows_count(payload: Dict[str, Any]) -> int:
rows = payload.get("rows")
return len(rows) if isinstance(rows, list) else 0
def _build_stale_payload_for_timeout_if_better_cached(
*,
filters: Dict[str, Any],
cached_entry: Dict[str, Any],
ranked_rows: List[Dict[str, Any]],
timeout_message: Optional[str],
) -> Optional[Dict[str, Any]]:
success_payload = cached_entry.get("success_payload")
if not isinstance(success_payload, dict) or not success_payload.get("rows"):
return None
if _rows_count(success_payload) < len(ranked_rows):
return None
error_message = timeout_message or "市场扫描快照正在刷新中"
set_scan_terminal_failure_state(filters, error_message=error_message)
failed_entry = get_scan_terminal_cache_entry(filters) or cached_entry
return build_stale_scan_terminal_payload(
filters=filters,
success_payload=success_payload,
error_message=error_message,
failed_at=failed_entry.get("last_failed_at"),
)
def _start_scan_terminal_background_refresh(filters: Dict[str, Any]) -> bool:
if not mark_scan_terminal_refreshing(filters):
return False
@@ -72,11 +103,17 @@ def _build_scan_terminal_payload_uncached(
filters: Dict[str, Any],
*,
force_refresh: bool = False,
timeout_sec: Optional[float] = None,
) -> Dict[str, Any]:
cached_entry = get_scan_terminal_cache_entry(filters) or {}
try:
city_names = list(CITIES.keys())
build_timeout_sec = float(
timeout_sec
if timeout_sec is not None
else SCAN_TERMINAL_BUILD_TIMEOUT_SEC
)
timezone_offset = filters.get("timezone_offset_seconds")
if timezone_offset is not None:
target_tz = int(timezone_offset)
@@ -116,7 +153,7 @@ def _build_scan_terminal_payload_uncached(
try:
completed = as_completed(
future_map,
timeout=float(SCAN_TERMINAL_BUILD_TIMEOUT_SEC),
timeout=build_timeout_sec,
)
for future in completed:
city_name = future_map[future]
@@ -131,8 +168,7 @@ def _build_scan_terminal_payload_uncached(
except FutureTimeoutError:
timed_out = True
timeout_message = (
f"scan terminal build timed out after "
f"{SCAN_TERMINAL_BUILD_TIMEOUT_SEC}s"
f"scan terminal build timed out after {build_timeout_sec:g}s"
)
failed_reasons.append(timeout_message)
for future, city_name in future_map.items():
@@ -175,7 +211,9 @@ def _build_scan_terminal_payload_uncached(
total_city_count=len(city_names),
failed_city_count=len(failed_cities),
)
ranked_rows = ranked_result["ranked_rows"]
ranked_rows = compact_ranked_scan_rows_for_payload(
ranked_result["ranked_rows"]
)
if timed_out and not ranked_rows:
success_payload = cached_entry.get("success_payload")
@@ -189,6 +227,16 @@ def _build_scan_terminal_payload_uncached(
summary = ranked_result["summary"]
top_signal = ranked_result["top_signal"]
if timed_out:
stale_payload = _build_stale_payload_for_timeout_if_better_cached(
filters=filters,
cached_entry=cached_entry,
ranked_rows=ranked_rows,
timeout_message=timeout_message,
)
if stale_payload is not None:
return stale_payload
payload = {
"generated_at": datetime.utcnow().isoformat() + "Z",
"filters": filters,
@@ -297,6 +345,50 @@ def build_scan_terminal_payload(
_SCAN_PREWARM_STARTED = False
_SCAN_PREWARM_LOCK = threading.Lock()
_SCAN_TERMINAL_PREWARM_RAW_FILTERS = [
{
"scan_mode": "tradable",
"min_price": 0.05,
"max_price": 0.95,
"min_edge_pct": 2,
"min_liquidity": 500,
"market_type": "maxtemp",
"time_range": "today",
"limit": 25,
},
{
"scan_mode": "tradable",
"min_price": 0.05,
"max_price": 0.95,
"min_edge_pct": 2,
"min_liquidity": 500,
"market_type": "maxtemp",
"time_range": "today",
"limit": 180,
}
]
def _scan_terminal_prewarm_filters() -> List[Dict[str, Any]]:
return [
_normalize_scan_terminal_filters(filters)
for filters in _SCAN_TERMINAL_PREWARM_RAW_FILTERS
]
def _warm_scan_terminal_payloads() -> int:
ok = 0
for filters in _scan_terminal_prewarm_filters():
try:
_build_scan_terminal_payload_uncached(
filters,
force_refresh=False,
timeout_sec=SCAN_TERMINAL_PREWARM_PAYLOAD_TIMEOUT_SEC,
)
ok += 1
except Exception as exc:
logger.warning("scan terminal payload pre-warm failed: {}", exc)
return ok
def start_scan_terminal_prewarm() -> None:
@@ -340,11 +432,13 @@ def start_scan_terminal_prewarm() -> None:
ok += 1
except Exception:
pass
payload_ok = _warm_scan_terminal_payloads()
elapsed = int(time.time() - started)
logger.info(
"scan terminal pre-warm finished ok={}/{} elapsed={}s",
"scan terminal pre-warm finished ok={}/{} payloads={} elapsed={}s",
ok,
len(city_names),
payload_ok,
elapsed,
)
except (ValueError, OSError, IOError):
+314 -49
View File
@@ -27,11 +27,20 @@ _CITY_FULL_REFRESH_INFLIGHT: Dict[str, "asyncio.Task[Dict[str, Any]]"] = {}
_CITY_FULL_STALE_REFRESH_TASKS: Dict[str, "asyncio.Task[Dict[str, Any]]"] = {}
_CITY_FULL_REFRESH_LOCK = asyncio.Lock()
CityDetailPayloadCacheKey = Tuple[str, str, str, str, str, int]
CityChartDetailPayloadCacheKey = Tuple[str, str, str, int]
CityDetailBatchResponseCacheKey = Tuple[Tuple[str, ...], bool, str, str, str, str]
_CITY_DETAIL_PAYLOAD_CACHE: Dict[CityDetailPayloadCacheKey, Dict[str, Any]] = {}
_CITY_DETAIL_PAYLOAD_CACHE_TS: Dict[CityDetailPayloadCacheKey, float] = {}
_CITY_DETAIL_PAYLOAD_INFLIGHT: Dict[CityDetailPayloadCacheKey, "asyncio.Task[Dict[str, Any]]"] = {}
_CITY_DETAIL_PAYLOAD_EPOCH: Dict[str, int] = {}
_CITY_DETAIL_PAYLOAD_LOCK = asyncio.Lock()
_CITY_CHART_DETAIL_PAYLOAD_CACHE: Dict[CityChartDetailPayloadCacheKey, Dict[str, Any]] = {}
_CITY_CHART_DETAIL_PAYLOAD_CACHE_TS: Dict[CityChartDetailPayloadCacheKey, float] = {}
_CITY_CHART_DETAIL_PAYLOAD_LOCK = asyncio.Lock()
_CITY_DETAIL_BATCH_RESPONSE_CACHE: Dict[CityDetailBatchResponseCacheKey, Dict[str, Any]] = {}
_CITY_DETAIL_BATCH_RESPONSE_CACHE_TS: Dict[CityDetailBatchResponseCacheKey, float] = {}
_CITY_DETAIL_BATCH_RESPONSE_INFLIGHT: Dict[CityDetailBatchResponseCacheKey, "asyncio.Task[Dict[str, Any]]"] = {}
_CITY_DETAIL_BATCH_RESPONSE_LOCK = asyncio.Lock()
def _city_detail_payload_cache_ttl() -> float:
@@ -42,6 +51,17 @@ def _city_detail_payload_cache_ttl() -> float:
return max(0.0, min(30.0, value))
def _city_detail_batch_response_cache_ttl() -> float:
try:
value = float(
os.getenv("POLYWEATHER_CITY_DETAIL_BATCH_RESPONSE_CACHE_TTL_SEC", "12")
or "12"
)
except ValueError:
value = 12.0
return max(0.0, min(30.0, value))
async def _overlay_cached_wunderground(city: str, payload: Dict[str, Any]) -> Dict[str, Any]:
return await run_in_threadpool(
legacy_routes._overlay_latest_wunderground_current,
@@ -86,6 +106,11 @@ async def _invalidate_city_detail_payload_cache(city: str) -> None:
for key in old_keys:
_CITY_DETAIL_PAYLOAD_CACHE.pop(key, None)
_CITY_DETAIL_PAYLOAD_CACHE_TS.pop(key, None)
async with _CITY_CHART_DETAIL_PAYLOAD_LOCK:
old_chart_keys = [key for key in _CITY_CHART_DETAIL_PAYLOAD_CACHE if key[0] == normalized]
for key in old_chart_keys:
_CITY_CHART_DETAIL_PAYLOAD_CACHE.pop(key, None)
_CITY_CHART_DETAIL_PAYLOAD_CACHE_TS.pop(key, None)
async def _refresh_city_full_data(city: str, force_refresh: bool) -> Dict[str, Any]:
@@ -130,6 +155,24 @@ async def _get_city_full_data(city: str, *, force_refresh: bool) -> Dict[str, An
return await _refresh_city_full_data(city, False)
async def _get_city_chart_data(city: str, *, force_refresh: bool) -> Dict[str, Any]:
if force_refresh:
return await _get_city_full_data(city, force_refresh=True)
cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, "full", city)
if cached_entry:
payload = cached_entry.get("payload") or {}
if payload:
if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_FULL_CACHE_TTL_SEC):
_start_city_full_stale_refresh(city)
return await _overlay_cached_wunderground(city, payload)
return {
"name": city,
"display_name": str((legacy_routes.CITY_REGISTRY.get(city, {}) or {}).get("display_name") or city.title()),
}
def _city_detail_payload_cache_key(
data: Dict[str, Any],
market_slug: Optional[str],
@@ -155,6 +198,27 @@ def _city_detail_payload_cache_key(
)
def _city_chart_detail_payload_cache_key(
data: Dict[str, Any],
resolution: Optional[str],
) -> CityChartDetailPayloadCacheKey:
city = str(data.get("city") or data.get("name") or "").strip().lower()
fingerprint = str(
data.get("updated_at_ts")
or data.get("updated_at")
or data.get("local_time")
or data.get("local_date")
or id(data)
)
generation = _CITY_DETAIL_PAYLOAD_EPOCH.get(city, 0)
return (
city,
str(resolution or "10m"),
fingerprint,
generation,
)
async def _build_city_detail_payload_cached(
data: Dict[str, Any],
market_slug: Optional[str],
@@ -212,6 +276,39 @@ async def _build_city_detail_payload_cached(
return payload
async def _build_city_chart_detail_payload(
data: Dict[str, Any],
resolution: Optional[str],
) -> Dict[str, Any]:
ttl = _city_detail_payload_cache_ttl()
if ttl <= 0:
return legacy_routes._build_city_chart_detail_payload(data, resolution)
key = _city_chart_detail_payload_cache_key(data, resolution)
now_ts = time.time()
async with _CITY_CHART_DETAIL_PAYLOAD_LOCK:
cached = _CITY_CHART_DETAIL_PAYLOAD_CACHE.get(key)
cached_ts = _CITY_CHART_DETAIL_PAYLOAD_CACHE_TS.get(key, 0.0)
if cached is not None and now_ts - cached_ts < ttl:
return cached
payload = legacy_routes._build_city_chart_detail_payload(data, resolution)
async with _CITY_CHART_DETAIL_PAYLOAD_LOCK:
_CITY_CHART_DETAIL_PAYLOAD_CACHE[key] = payload
_CITY_CHART_DETAIL_PAYLOAD_CACHE_TS[key] = time.time()
if len(_CITY_CHART_DETAIL_PAYLOAD_CACHE) > 256:
oldest_keys = sorted(
_CITY_CHART_DETAIL_PAYLOAD_CACHE_TS,
key=lambda item: _CITY_CHART_DETAIL_PAYLOAD_CACHE_TS.get(item, 0.0),
)[:64]
for old_key in oldest_keys:
_CITY_CHART_DETAIL_PAYLOAD_CACHE.pop(old_key, None)
_CITY_CHART_DETAIL_PAYLOAD_CACHE_TS.pop(old_key, None)
return payload
def _default_deb_recent() -> Dict[str, object]:
return {
"tier": "other",
@@ -458,6 +555,99 @@ def _parse_batch_city_names(raw_cities: str, *, limit: int) -> List[str]:
return out
def _city_detail_batch_response_cache_key(
city_names: List[str],
*,
force_refresh: bool,
market_slug: Optional[str],
target_date: Optional[str],
resolution: Optional[str],
scope: str,
) -> CityDetailBatchResponseCacheKey:
return (
tuple(city_names),
bool(force_refresh),
str(market_slug or ""),
str(target_date or ""),
str(resolution or "10m"),
str(scope or "full"),
)
def _normalize_city_detail_scope(scope: Optional[str]) -> str:
raw = str(scope or "full").strip().lower()
if raw in {"chart", "charts", "terminal", "terminal_chart"}:
return "chart"
return "full"
def _chart_scoped_city_detail(detail: Dict[str, Any]) -> Dict[str, Any]:
overview = detail.get("overview") if isinstance(detail.get("overview"), dict) else {}
timeseries = detail.get("timeseries") if isinstance(detail.get("timeseries"), dict) else {}
forecast = detail.get("forecast") if isinstance(detail.get("forecast"), dict) else {}
airport_primary_today_obs = (
detail.get("airport_primary_today_obs")
or overview.get("airport_primary_today_obs")
or []
)
forecast_daily = (
(forecast.get("daily") if isinstance(forecast, dict) else None)
or timeseries.get("forecast_daily")
or []
)
local_date = detail.get("local_date") or overview.get("local_date")
local_time = detail.get("local_time") or overview.get("local_time")
scoped = {
"city": detail.get("city") or overview.get("name"),
"fetched_at": detail.get("fetched_at"),
"local_date": local_date,
"local_time": local_time,
"overview": {
"name": overview.get("name"),
"display_name": overview.get("display_name"),
"local_date": local_date,
"local_time": local_time,
"temp_symbol": overview.get("temp_symbol"),
"current_temp": overview.get("current_temp"),
"deb_prediction": overview.get("deb_prediction"),
"settlement_source": overview.get("settlement_source"),
"settlement_source_label": overview.get("settlement_source_label"),
},
"timeseries": {
"hourly": timeseries.get("hourly") or detail.get("hourly") or {},
"metar_today_obs": timeseries.get("metar_today_obs") or [],
"settlement_today_obs": timeseries.get("settlement_today_obs") or [],
"forecast_daily": forecast_daily,
},
"hourly": timeseries.get("hourly") or detail.get("hourly") or {},
"models_hourly": detail.get("models_hourly") or {},
"deb": detail.get("deb") or {},
"forecast": {
"today_high": forecast.get("today_high") if isinstance(forecast, dict) else None,
"daily": forecast_daily,
},
"multi_model_daily": detail.get("multi_model_daily") or {},
"probabilities": detail.get("probabilities") or {"mu": None, "distribution": []},
"runway_plate_history": detail.get("runway_plate_history") or {},
"runway_band_history": detail.get("runway_band_history") or [],
"amos": detail.get("amos") or {},
"airport_current": detail.get("airport_current") or {},
"airport_primary": detail.get("airport_primary") or overview.get("airport_primary") or {},
"airport_primary_today_obs": airport_primary_today_obs,
"official": {"airport_primary_today_obs": airport_primary_today_obs},
"wunderground_current": detail.get("wunderground_current") or {},
"settlement_station": detail.get("settlement_station") or overview.get("settlement_station") or {},
}
return scoped
def _apply_city_detail_scope(detail: Dict[str, Any], scope: str) -> Dict[str, Any]:
if scope == "chart":
return _chart_scoped_city_detail(detail)
return detail
async def _build_city_detail_batch_item_async(
city: str,
*,
@@ -465,8 +655,24 @@ async def _build_city_detail_batch_item_async(
market_slug: Optional[str],
target_date: Optional[str],
resolution: Optional[str],
detail_scope: str = "full",
timing_recorder: Optional[ServerTimingRecorder] = None,
) -> Tuple[str, Dict[str, Any]]:
if detail_scope == "chart":
if timing_recorder is not None:
data = await timing_recorder.measure_async(
f"chart_data_{city}",
lambda: _get_city_chart_data(city, force_refresh=force_refresh),
)
detail = await timing_recorder.measure_async(
f"chart_payload_{city}",
lambda: _build_city_chart_detail_payload(data, resolution),
)
else:
data = await _get_city_chart_data(city, force_refresh=force_refresh)
detail = await _build_city_chart_detail_payload(data, resolution)
return city, detail
if timing_recorder is not None:
data = await timing_recorder.measure_async(
f"full_data_{city}",
@@ -489,7 +695,7 @@ async def _build_city_detail_batch_item_async(
target_date,
resolution,
)
return city, detail
return city, _apply_city_detail_scope(detail, detail_scope)
def _city_detail_batch_concurrency() -> int:
@@ -521,6 +727,7 @@ async def get_city_detail_batch_payload(
market_slug: Optional[str] = None,
target_date: Optional[str] = None,
resolution: Optional[str] = "10m",
scope: Optional[str] = "full",
limit: int = 12,
) -> Dict[str, Any]:
timer = ServerTimingRecorder(
@@ -548,62 +755,120 @@ async def get_city_detail_batch_payload(
"missing": [],
"partial": False,
}
detail_scope = _normalize_city_detail_scope(scope)
semaphore = asyncio.Semaphore(_city_detail_batch_concurrency())
async def _build_uncached_payload() -> Dict[str, Any]:
semaphore = asyncio.Semaphore(_city_detail_batch_concurrency())
async def _build_with_limit(city: str) -> Tuple[str, Dict[str, Any]]:
async with semaphore:
return await _build_city_detail_batch_item_async(
city,
force_refresh=force_refresh,
market_slug=market_slug,
target_date=target_date,
resolution=resolution,
timing_recorder=timer,
)
async def _build_with_limit(city: str) -> Tuple[str, Dict[str, Any]]:
async with semaphore:
return await _build_city_detail_batch_item_async(
city,
force_refresh=force_refresh,
market_slug=market_slug,
target_date=target_date,
resolution=resolution,
detail_scope=detail_scope,
timing_recorder=timer,
)
task_by_city = {
city: asyncio.create_task(_build_with_limit(city))
for city in city_names
}
task_city_lookup = {task: city for city, task in task_by_city.items()}
done, pending = await timer.measure_async(
"build_details",
lambda: asyncio.wait(
task_by_city.values(),
timeout=_city_detail_batch_partial_timeout_seconds(),
),
task_by_city = {
city: asyncio.create_task(_build_with_limit(city))
for city in city_names
}
task_city_lookup = {task: city for city, task in task_by_city.items()}
done, pending = await timer.measure_async(
"build_details",
lambda: asyncio.wait(
task_by_city.values(),
timeout=_city_detail_batch_partial_timeout_seconds(),
),
)
details: Dict[str, Any] = {}
errors: Dict[str, str] = {}
missing: List[str] = []
for task in done:
city = task_city_lookup[task]
try:
result_city, payload = task.result()
except Exception as exc:
errors[city] = str(exc)
continue
details[result_city] = payload
for task in pending:
city = task_city_lookup[task]
missing.append(city)
task.cancel()
missing_set = set(missing)
missing = [city for city in city_names if city in missing_set]
return {
"cities": city_names,
"details": details,
"errors": errors,
"missing": missing,
"partial": bool(missing or errors),
}
cache_ttl = _city_detail_batch_response_cache_ttl()
cache_key = _city_detail_batch_response_cache_key(
city_names,
force_refresh=force_refresh,
market_slug=market_slug,
target_date=target_date,
resolution=resolution,
scope=detail_scope,
)
details: Dict[str, Any] = {}
errors: Dict[str, str] = {}
missing: List[str] = []
for task in done:
city = task_city_lookup[task]
if cache_ttl > 0 and not force_refresh:
now_ts = time.time()
async with _CITY_DETAIL_BATCH_RESPONSE_LOCK:
cached = _CITY_DETAIL_BATCH_RESPONSE_CACHE.get(cache_key)
cached_ts = _CITY_DETAIL_BATCH_RESPONSE_CACHE_TS.get(cache_key, 0.0)
if cached is not None and now_ts - cached_ts < cache_ttl:
outcome = "cache_hit"
return cached
task = _CITY_DETAIL_BATCH_RESPONSE_INFLIGHT.get(cache_key)
owner = False
if task is None:
owner = True
task = asyncio.create_task(_build_uncached_payload())
_CITY_DETAIL_BATCH_RESPONSE_INFLIGHT[cache_key] = task
try:
result_city, payload = task.result()
except Exception as exc:
errors[city] = str(exc)
continue
details[result_city] = payload
payload = await timer.measure_async(
"build_or_wait_cached_batch",
lambda: task,
)
finally:
if owner and task.done():
async with _CITY_DETAIL_BATCH_RESPONSE_LOCK:
if _CITY_DETAIL_BATCH_RESPONSE_INFLIGHT.get(cache_key) is task:
_CITY_DETAIL_BATCH_RESPONSE_INFLIGHT.pop(cache_key, None)
if payload.get("partial"):
outcome = "partial"
elif not owner:
outcome = "shared_inflight"
for task in pending:
city = task_city_lookup[task]
missing.append(city)
task.cancel()
if owner:
async with _CITY_DETAIL_BATCH_RESPONSE_LOCK:
if not payload.get("partial"):
_CITY_DETAIL_BATCH_RESPONSE_CACHE[cache_key] = payload
_CITY_DETAIL_BATCH_RESPONSE_CACHE_TS[cache_key] = time.time()
if len(_CITY_DETAIL_BATCH_RESPONSE_CACHE) > 128:
oldest_keys = sorted(
_CITY_DETAIL_BATCH_RESPONSE_CACHE_TS,
key=lambda item: _CITY_DETAIL_BATCH_RESPONSE_CACHE_TS.get(item, 0.0),
)[:32]
for old_key in oldest_keys:
_CITY_DETAIL_BATCH_RESPONSE_CACHE.pop(old_key, None)
_CITY_DETAIL_BATCH_RESPONSE_CACHE_TS.pop(old_key, None)
return payload
missing_set = set(missing)
missing = [city for city in city_names if city in missing_set]
partial = bool(missing or errors)
if partial:
payload = await _build_uncached_payload()
if payload.get("partial"):
outcome = "partial"
return {
"cities": city_names,
"details": details,
"errors": errors,
"missing": missing,
"partial": partial,
}
return payload
except HTTPException as exc:
outcome = f"http_{exc.status_code}"
status_code = exc.status_code
+145
View File
@@ -161,6 +161,151 @@ def build_runway_band_history(raw_history: Dict[str, List[Dict[str, Any]]], reso
return band_history
def build_runway_chart_histories(
raw_history: Dict[str, List[Dict[str, Any]]],
resolution: str,
) -> tuple[Dict[str, List[Dict[str, Any]]], List[Dict[str, Any]]]:
if not raw_history:
return {}, []
try:
if resolution.endswith("m"):
minutes = int(resolution[:-1])
elif resolution.endswith("h"):
minutes = int(resolution[:-1]) * 60
else:
minutes = 10
except Exception:
minutes = 10
seconds = max(60, minutes * 60)
per_runway_buckets: Dict[str, Dict[int, List[float]]] = {}
all_buckets: Dict[int, List[float]] = {}
for rwy, points in raw_history.items():
if not points:
continue
rwy_key = str(rwy or "").strip()
if not rwy_key:
continue
runway_buckets = per_runway_buckets.setdefault(rwy_key, {})
for pt in points:
t_str = pt.get("time") or pt.get("timestamp")
temp = pt.get("temp") or pt.get("temp_c") or pt.get("value")
if temp is None or not isinstance(t_str, str):
continue
dt = _parse_time_val(t_str)
if not dt:
continue
try:
temp_value = float(temp)
except Exception:
continue
bucket_ts = (int(dt.timestamp()) // seconds) * seconds
runway_buckets.setdefault(bucket_ts, []).append(temp_value)
all_buckets.setdefault(bucket_ts, []).append(temp_value)
aggregated_history: Dict[str, List[Dict[str, Any]]] = {}
for rwy, buckets in per_runway_buckets.items():
bucket_points = []
for bucket_ts in sorted(buckets.keys()):
temps = buckets[bucket_ts]
if not temps:
continue
bucket_dt = datetime.fromtimestamp(bucket_ts, tz=timezone.utc)
bucket_points.append(
{
"time": bucket_dt.isoformat(),
"temp": round(temps[-1], 1),
}
)
if bucket_points:
aggregated_history[rwy] = bucket_points
band_history = []
for bucket_ts in sorted(all_buckets.keys()):
temps = all_buckets[bucket_ts]
if not temps:
continue
bucket_dt = datetime.fromtimestamp(bucket_ts, tz=timezone.utc)
band_history.append(
{
"time": bucket_dt.isoformat(),
"high_temp": round(max(temps), 1),
"low_temp": round(min(temps), 1),
"avg_temp": round(sum(temps) / len(temps), 1),
}
)
return aggregated_history, band_history
def build_city_chart_detail_payload(
data: Dict[str, Any],
resolution: Optional[str] = "10m",
) -> Dict[str, Any]:
forecast = data.get("forecast") if isinstance(data.get("forecast"), dict) else {}
current = data.get("current") if isinstance(data.get("current"), dict) else {}
multi_model = data.get("multi_model") if isinstance(data.get("multi_model"), dict) else {}
runway_history, runway_band_history = build_runway_chart_histories(
data.get("runway_plate_history") or {},
resolution or "10m",
)
airport_primary_today_obs = data.get("airport_primary_today_obs") or []
local_date = data.get("local_date")
local_time = data.get("local_time")
return {
"city": data.get("name") or data.get("city"),
"fetched_at": data.get("updated_at"),
"local_date": local_date,
"local_time": local_time,
"overview": {
"name": data.get("name") or data.get("city"),
"display_name": data.get("display_name"),
"local_date": local_date,
"local_time": local_time,
"temp_symbol": data.get("temp_symbol"),
"current_temp": current.get("temp"),
"deb_prediction": (data.get("deb") or {}).get("prediction"),
"settlement_source": current.get("settlement_source"),
"settlement_source_label": current.get("settlement_source_label"),
},
"timeseries": {
"hourly": data.get("hourly") or {},
"metar_today_obs": data.get("metar_today_obs") or [],
"settlement_today_obs": data.get("settlement_today_obs") or [],
"forecast_daily": forecast.get("daily") or [],
},
"hourly": data.get("hourly") or {},
"models_hourly": {
"times": multi_model.get("hourly_times") or [],
"curves": {
model: values
for model, values in (multi_model.get("hourly_forecasts") or {}).items()
if not _is_excluded_model_name(model)
},
},
"deb": data.get("deb") or {},
"forecast": {
"today_high": forecast.get("today_high"),
"daily": forecast.get("daily") or [],
},
"multi_model_daily": data.get("multi_model_daily") or {},
"probabilities": data.get("probabilities") or {"mu": None, "distribution": []},
"runway_plate_history": runway_history,
"runway_band_history": runway_band_history,
"amos": data.get("amos") or {},
"airport_current": data.get("airport_current") or {},
"airport_primary": data.get("airport_primary") or {},
"airport_primary_today_obs": airport_primary_today_obs,
"official": {"airport_primary_today_obs": airport_primary_today_obs},
"wunderground_current": data.get("wunderground_current") or {},
"settlement_station": data.get("settlement_station") or {},
}
def build_city_detail_payload(
data: Dict[str, Any],
market_slug: Optional[str] = None,
+1
View File
@@ -25,6 +25,7 @@ from src.utils.refresh_policy import OBSERVATION_REFRESH_SEC, SCAN_ROWS_REFRESH_
from web.analysis_service import (
_analyze,
_analyze_summary,
_build_city_chart_detail_payload, # noqa: F401 - compatibility export for chart detail batches
_build_city_detail_payload, # noqa: F401 - compatibility export for tests and transitional routers
_build_city_market_scan_payload,
_build_city_summary_payload,
+62 -3
View File
@@ -545,18 +545,41 @@ def get_ops_billing_risk(
"limit": str(max(safe_limit * 10, 500)),
},
)
subscription_rows = collect(
trial_subscription_rows = collect(
"subscriptions",
{
"select": (
"id,user_id,plan_code,source,status,starts_at,expires_at,"
"created_at,updated_at"
),
"or": "(source.eq.signup_trial,plan_code.eq.signup_trial_3d,status.eq.active)",
"or": "(source.eq.signup_trial,plan_code.eq.signup_trial_3d)",
"order": "created_at.desc",
"limit": str(max(safe_limit * 20, 1000)),
},
)
active_subscription_rows = collect(
"subscriptions",
{
"select": (
"id,user_id,plan_code,source,status,starts_at,expires_at,"
"created_at,updated_at"
),
"status": "eq.active",
"order": "created_at.desc",
"limit": str(max(safe_limit * 20, 1000)),
},
)
subscription_rows: List[Dict[str, Any]] = []
seen_subscription_keys: set[str] = set()
for row in [*trial_subscription_rows, *active_subscription_rows]:
key = str(row.get("id") or "").strip() or (
f"{row.get('user_id')}:{row.get('plan_code')}:{row.get('source')}:"
f"{row.get('starts_at')}:{row.get('expires_at')}"
)
if key in seen_subscription_keys:
continue
seen_subscription_keys.add(key)
subscription_rows.append(row)
entitlement_trial_events = collect(
"entitlement_events",
{
@@ -751,6 +774,40 @@ def get_ops_billing_risk(
def normalize_user_key(value: Any) -> str:
return str(value or "").strip().lower()
def analytics_payload(row: Dict[str, Any]) -> Dict[str, Any]:
payload = row.get("payload")
return payload if isinstance(payload, dict) else {}
def analytics_correlation_keys(row: Dict[str, Any]) -> set[str]:
payload = analytics_payload(row)
keys: set[str] = set()
user_id = normalize_user_key(row.get("user_id") or payload.get("user_id"))
client_id = str(row.get("client_id") or "").strip()
session_id = str(row.get("session_id") or "").strip()
if user_id:
keys.add(f"user:{user_id}")
if client_id:
keys.add(f"client:{client_id}")
if session_id:
keys.add(f"session:{session_id}")
return keys
signup_intent_keys: set[str] = set()
for row in events:
if str(row.get("event_type") or "").strip().lower() != "login_start":
continue
payload = analytics_payload(row)
mode = str(payload.get("mode") or payload.get("auth_mode") or "").strip().lower()
if mode == "signup":
signup_intent_keys.update(analytics_correlation_keys(row))
def has_signup_intent(row: Dict[str, Any]) -> bool:
payload = analytics_payload(row)
mode = str(payload.get("mode") or payload.get("auth_mode") or "").strip().lower()
if mode == "signup" or payload.get("signup_intent") is True:
return True
return bool(analytics_correlation_keys(row).intersection(signup_intent_keys))
trial_actor_keys = {
_app_analytics_actor_key(row)
for row in events
@@ -817,10 +874,12 @@ def get_ops_billing_risk(
)
for row in signup_rows[:300]:
if not has_signup_intent(row):
continue
actor_key = _app_analytics_actor_key(row)
if actor_key in trial_actor_keys:
continue
payload = row.get("payload") if isinstance(row.get("payload"), dict) else {}
payload = analytics_payload(row)
signup_user_id = normalize_user_key(row.get("user_id") or payload.get("user_id"))
if not signup_user_id:
continue
+7 -1
View File
@@ -31,10 +31,16 @@ SCAN_TERMINAL_PAYLOAD_TTL_SEC = min(
)
SCAN_TERMINAL_BUILD_TIMEOUT_SEC = _env_int(
"POLYWEATHER_SCAN_TERMINAL_BUILD_TIMEOUT_SEC",
30,
10,
min_value=8,
max_value=30,
)
SCAN_TERMINAL_PREWARM_PAYLOAD_TIMEOUT_SEC = _env_int(
"POLYWEATHER_SCAN_TERMINAL_PREWARM_PAYLOAD_TIMEOUT_SEC",
30,
min_value=10,
max_value=120,
)
SCAN_TERMINAL_MAX_WORKERS = _env_int(
"POLYWEATHER_SCAN_TERMINAL_MAX_WORKERS",
8,
+2 -5
View File
@@ -5,7 +5,7 @@ from __future__ import annotations
import time
from typing import Any, Dict, Optional
from fastapi import BackgroundTasks, HTTPException, Request
from fastapi import BackgroundTasks, Request
from fastapi.concurrency import run_in_threadpool
from fastapi.responses import PlainTextResponse
from loguru import logger
@@ -16,10 +16,7 @@ import web.routes as legacy_routes
def get_health_payload() -> Dict[str, Any]:
payload = build_health_payload()
if payload.get("status") != "ok":
raise HTTPException(status_code=503, detail=payload)
return payload
return build_health_payload()
async def get_system_status_payload() -> Dict[str, Any]: