Compare commits

...

4 Commits

Author SHA1 Message Date
2569718930@qq.com 78ea0326a5 Return partial city detail batches 2026-05-31 19:51:28 +08:00
2569718930@qq.com d3f444dbf6 Bind PolyWeather services to loopback 2026-05-31 19:46:00 +08:00
2569718930@qq.com 8d26afdec0 Reduce terminal auth and analytics stalls 2026-05-31 19:21:53 +08:00
2569718930@qq.com 5083b7c433 Instrument API timing and reduce detail fallbacks 2026-05-31 18:58:57 +08:00
25 changed files with 1267 additions and 174 deletions
+2 -2
View File
@@ -79,7 +79,7 @@ services:
timeout: 5s timeout: 5s
image: ghcr.io/yangyuan-zhen/polyweather-frontend:${IMAGE_TAG:-latest} image: ghcr.io/yangyuan-zhen/polyweather-frontend:${IMAGE_TAG:-latest}
ports: ports:
- 3001:3000 - "127.0.0.1:3001:3000"
restart: unless-stopped restart: unless-stopped
polyweather_web: polyweather_web:
command: python web/app.py command: python web/app.py
@@ -104,7 +104,7 @@ services:
timeout: 5s timeout: 5s
image: ghcr.io/yangyuan-zhen/polyweather-backend:${IMAGE_TAG:-latest} image: ghcr.io/yangyuan-zhen/polyweather-backend:${IMAGE_TAG:-latest}
ports: ports:
- 8000:8000 - "127.0.0.1:8000:8000"
restart: unless-stopped restart: unless-stopped
user: ${UID:-1000}:${GID:-1000} user: ${UID:-1000}:${GID:-1000}
volumes: volumes:
+80 -24
View File
@@ -7,25 +7,46 @@ import {
buildProxyExceptionResponse, buildProxyExceptionResponse,
buildUpstreamErrorResponse, buildUpstreamErrorResponse,
} from "@/lib/api-proxy"; } from "@/lib/api-proxy";
import {
createProxyTimer,
finishProxyTimedResponse,
} from "@/lib/proxy-timing";
const API_BASE = process.env.POLYWEATHER_API_BASE_URL; const API_BASE = process.env.POLYWEATHER_API_BASE_URL;
const ANALYTICS_ENABLED = const ANALYTICS_ENABLED =
process.env.NEXT_PUBLIC_POLYWEATHER_APP_ANALYTICS !== "false"; process.env.NEXT_PUBLIC_POLYWEATHER_APP_ANALYTICS !== "false";
const ANALYTICS_PROXY_TIMEOUT_MS = Math.max(
250,
Number(process.env.POLYWEATHER_ANALYTICS_PROXY_TIMEOUT_MS || "1500") || 1500,
);
export async function POST(req: NextRequest) { export async function POST(req: NextRequest) {
const timer = createProxyTimer(req, "analytics_events");
if (!ANALYTICS_ENABLED) { if (!ANALYTICS_ENABLED) {
return new NextResponse(null, { status: 204 }); return finishProxyTimedResponse(
} new NextResponse(null, { status: 204 }),
timer,
if (!API_BASE) { "disabled",
return NextResponse.json(
{ error: "POLYWEATHER_API_BASE_URL is not configured" },
{ status: 500 },
); );
} }
if (!API_BASE) {
return finishProxyTimedResponse(
NextResponse.json(
{ error: "POLYWEATHER_API_BASE_URL is not configured" },
{ status: 500 },
),
timer,
"missing_api_base",
);
}
const controller = new AbortController();
const timeoutId = setTimeout(() => controller.abort(), ANALYTICS_PROXY_TIMEOUT_MS);
let auth: Awaited<ReturnType<typeof buildBackendRequestHeaders>> | null = null;
try { try {
const body = await req.json(); const body = await timer.measure("request_read", () => req.json());
const payload = const payload =
body && typeof body.payload === "object" && body.payload != null body && typeof body.payload === "object" && body.payload != null
? body.payload ? body.payload
@@ -42,30 +63,65 @@ export async function POST(req: NextRequest) {
referer_header: req.headers.get("referer") || "", referer_header: req.headers.get("referer") || "",
}, },
}; };
const auth = await buildBackendRequestHeaders(req, { auth = await timer.measure("auth_headers", () =>
includeSupabaseIdentity: false, buildBackendRequestHeaders(req, {
}); includeSupabaseIdentity: false,
}),
);
const headers = new Headers(auth.headers); const headers = new Headers(auth.headers);
headers.set("Content-Type", "application/json"); headers.set("Content-Type", "application/json");
const res = await fetch(`${API_BASE}/api/analytics/events`, { const res = await timer.measure("backend_fetch", () =>
method: "POST", fetch(`${API_BASE}/api/analytics/events`, {
headers, method: "POST",
body: JSON.stringify(enrichedBody), headers,
cache: "no-store", body: JSON.stringify(enrichedBody),
}); cache: "no-store",
signal: controller.signal,
}),
);
const backendServerTiming = res.headers.get("server-timing") || "";
if (!res.ok) { if (!res.ok) {
const raw = await res.text(); const raw = await timer.measure("backend_read", () => res.text());
const response = buildUpstreamErrorResponse(res.status, raw, { const response = buildUpstreamErrorResponse(res.status, raw, {
detailLimit: 260, detailLimit: 260,
}); });
return applyAuthResponseCookies(response, auth.response); return finishProxyTimedResponse(
applyAuthResponseCookies(response, auth.response),
timer,
`upstream_${res.status}`,
{ backendServerTiming },
);
} }
const data = await res.json(); const data = await timer.measure("backend_read", () => res.json());
const response = NextResponse.json(data); const response = NextResponse.json(data);
return applyAuthResponseCookies(response, auth.response); return finishProxyTimedResponse(
applyAuthResponseCookies(response, auth.response),
timer,
"ok",
{ backendServerTiming },
);
} catch (error) { } catch (error) {
return buildProxyExceptionResponse(error, { const timedOut = controller.signal.aborted;
publicMessage: "Failed to track analytics event", const response = timedOut
}); ? NextResponse.json(
{
ok: false,
accepted: true,
dropped: true,
reason: "timeout",
},
{ status: 202 },
)
: buildProxyExceptionResponse(error, {
publicMessage: "Failed to track analytics event",
});
const withCookies = auth ? applyAuthResponseCookies(response, auth.response) : response;
return finishProxyTimedResponse(
withCookies,
timer,
timedOut ? "timeout_accepted" : "exception",
);
} finally {
clearTimeout(timeoutId);
} }
} }
+20 -5
View File
@@ -1,6 +1,10 @@
import { NextRequest, NextResponse } from "next/server"; import { NextRequest, NextResponse } from "next/server";
import { proxyBackendJsonGet } from "@/lib/api-proxy"; import { proxyBackendJsonGet } from "@/lib/api-proxy";
import { buildCityDetailProxyCachePolicy } from "@/lib/proxy-cache-policy"; import { buildCityDetailProxyCachePolicy } from "@/lib/proxy-cache-policy";
import {
createProxyTimer,
finishProxyTimedResponse,
} from "@/lib/proxy-timing";
const API_BASE = process.env.POLYWEATHER_API_BASE_URL; const API_BASE = process.env.POLYWEATHER_API_BASE_URL;
const DETAIL_BATCH_PROXY_TIMEOUT_MS = Number( const DETAIL_BATCH_PROXY_TIMEOUT_MS = Number(
@@ -8,10 +12,15 @@ const DETAIL_BATCH_PROXY_TIMEOUT_MS = Number(
); );
export async function GET(req: NextRequest) { export async function GET(req: NextRequest) {
const timer = createProxyTimer(req, "city_detail_batch");
if (!API_BASE) { if (!API_BASE) {
return NextResponse.json( return finishProxyTimedResponse(
{ error: "POLYWEATHER_API_BASE_URL is not configured" }, NextResponse.json(
{ status: 500 }, { error: "POLYWEATHER_API_BASE_URL is not configured" },
{ status: 500 },
),
timer,
"missing_api_base",
); );
} }
@@ -33,12 +42,18 @@ export async function GET(req: NextRequest) {
try { try {
return await proxyBackendJsonGet(req, { return await proxyBackendJsonGet(req, {
cacheControl: cachePolicy.responseCacheControl, cacheControl: cachePolicy.responseCacheControl,
fetchCache: cacheControlForData: (data) =>
cachePolicy.fetchMode === "no-store" ? "no-store" : undefined, data &&
typeof data === "object" &&
(data as { partial?: unknown }).partial === true
? "no-store, max-age=0"
: cachePolicy.responseCacheControl,
fetchCache: "no-store",
publicMessage: "Failed to fetch city detail batch", publicMessage: "Failed to fetch city detail batch",
revalidateSeconds: cachePolicy.revalidateSeconds, revalidateSeconds: cachePolicy.revalidateSeconds,
signal: controller.signal, signal: controller.signal,
timeoutPublicMessage: "City detail batch request timed out", timeoutPublicMessage: "City detail batch request timed out",
timing: timer,
url: `${API_BASE}/api/cities/detail-batch?${searchParams.toString()}`, url: `${API_BASE}/api/cities/detail-batch?${searchParams.toString()}`,
}); });
} finally { } finally {
+38 -16
View File
@@ -9,6 +9,10 @@ import {
} from "@/lib/api-proxy"; } from "@/lib/api-proxy";
import { buildCachedJsonResponse } from "@/lib/http-cache"; import { buildCachedJsonResponse } from "@/lib/http-cache";
import { buildCityDetailProxyCachePolicy } from "@/lib/proxy-cache-policy"; import { buildCityDetailProxyCachePolicy } from "@/lib/proxy-cache-policy";
import {
createProxyTimer,
finishProxyTimedResponse,
} from "@/lib/proxy-timing";
const API_BASE = process.env.POLYWEATHER_API_BASE_URL; const API_BASE = process.env.POLYWEATHER_API_BASE_URL;
@@ -34,15 +38,16 @@ export async function GET(
req: NextRequest, req: NextRequest,
context: { params: Promise<{ name: string }> }, context: { params: Promise<{ name: string }> },
) { ) {
const timer = createProxyTimer(req, "city_detail");
if (!API_BASE) { if (!API_BASE) {
const response = NextResponse.json( const response = NextResponse.json(
{ error: "POLYWEATHER_API_BASE_URL is not configured" }, { error: "POLYWEATHER_API_BASE_URL is not configured" },
{ status: 500 }, { status: 500 },
); );
return response; return finishProxyTimedResponse(response, timer, "missing_api_base");
} }
const { name } = await context.params; const { name } = await timer.measure("route_params", () => context.params);
const forceRefresh = req.nextUrl.searchParams.get("force_refresh") ?? "false"; const forceRefresh = req.nextUrl.searchParams.get("force_refresh") ?? "false";
const cachePolicy = buildCityDetailProxyCachePolicy(forceRefresh, 15); const cachePolicy = buildCityDetailProxyCachePolicy(forceRefresh, 15);
const depth = req.nextUrl.searchParams.get("depth"); const depth = req.nextUrl.searchParams.get("depth");
@@ -67,31 +72,48 @@ export async function GET(
const url = `${API_BASE}/api/city/${encodeURIComponent(name)}/detail?${searchParams.toString()}`; const url = `${API_BASE}/api/city/${encodeURIComponent(name)}/detail?${searchParams.toString()}`;
try { try {
const auth = await buildBackendRequestHeaders(req, { const auth = await timer.measure("auth_headers", () =>
includeSupabaseIdentity: false, buildBackendRequestHeaders(req, {
}); includeSupabaseIdentity: false,
const res = await fetch(url, { }),
headers: auth.headers, );
...(cachePolicy.fetchMode === "no-store" const res = await timer.measure("backend_fetch", () =>
? { cache: "no-store" as const } fetch(url, {
: { next: { revalidate: cachePolicy.revalidateSeconds ?? 15 } }), headers: auth.headers,
}); ...(cachePolicy.fetchMode === "no-store"
? { cache: "no-store" as const }
: { next: { revalidate: cachePolicy.revalidateSeconds ?? 15 } }),
}),
);
const backendServerTiming = res.headers.get("server-timing") || "";
if (!res.ok) { if (!res.ok) {
const raw = await res.text(); const raw = await timer.measure("backend_read", () => res.text());
const response = buildUpstreamErrorResponse(res.status, raw); const response = buildUpstreamErrorResponse(res.status, raw);
return applyAuthResponseCookies(response, auth.response); return finishProxyTimedResponse(
applyAuthResponseCookies(response, auth.response),
timer,
`upstream_${res.status}`,
{ backendServerTiming },
);
} }
const data = normalizeCityDetailPayload(await res.json()); const data = normalizeCityDetailPayload(
await timer.measure("backend_read", () => res.json()),
);
const response = buildCachedJsonResponse( const response = buildCachedJsonResponse(
req, req,
data, data,
cachePolicy.responseCacheControl, cachePolicy.responseCacheControl,
); );
return applyAuthResponseCookies(response, auth.response); return finishProxyTimedResponse(
applyAuthResponseCookies(response, auth.response),
timer,
"ok",
{ backendServerTiming },
);
} catch (error) { } catch (error) {
const response = buildProxyExceptionResponse(error, { const response = buildProxyExceptionResponse(error, {
publicMessage: "Failed to fetch city detail aggregate", publicMessage: "Failed to fetch city detail aggregate",
}); });
return response; return finishProxyTimedResponse(response, timer, "exception");
} }
} }
+42 -15
View File
@@ -5,27 +5,45 @@ import {
} from "@/lib/backend-auth"; } from "@/lib/backend-auth";
import { buildProxyExceptionResponse } from "@/lib/api-proxy"; import { buildProxyExceptionResponse } from "@/lib/api-proxy";
import { requireOpsProxyAuth } from "@/lib/ops-proxy-auth"; import { requireOpsProxyAuth } from "@/lib/ops-proxy-auth";
import {
createProxyTimer,
finishProxyTimedResponse,
} from "@/lib/proxy-timing";
const API_BASE = process.env.POLYWEATHER_API_BASE_URL; const API_BASE = process.env.POLYWEATHER_API_BASE_URL;
export async function GET(req: NextRequest) { export async function GET(req: NextRequest) {
const timer = createProxyTimer(req, "ops_online_users");
if (!API_BASE) { if (!API_BASE) {
return NextResponse.json( return finishProxyTimedResponse(
{ error: "POLYWEATHER_API_BASE_URL is not configured" }, NextResponse.json(
{ status: 500 }, { error: "POLYWEATHER_API_BASE_URL is not configured" },
{ status: 500 },
),
timer,
"missing_api_base",
); );
} }
try { try {
const auth = await buildBackendRequestHeaders(req); const auth = await timer.measure("auth_headers", () =>
const authError = requireOpsProxyAuth(req, auth); buildBackendRequestHeaders(req),
if (authError) return authError; );
const authError = timer.measureSync("ops_auth", () =>
requireOpsProxyAuth(req, auth),
);
if (authError) {
return finishProxyTimedResponse(authError, timer, "ops_auth_error");
}
const res = await fetch(`${API_BASE}/api/ops/online-users`, { const res = await timer.measure("backend_fetch", () =>
headers: auth.headers, fetch(`${API_BASE}/api/ops/online-users`, {
cache: "no-store", headers: auth.headers,
}); cache: "no-store",
const raw = await res.text(); }),
);
const backendServerTiming = res.headers.get("server-timing") || "";
const raw = await timer.measure("backend_read", () => res.text());
const response = new NextResponse(raw, { const response = new NextResponse(raw, {
status: res.status, status: res.status,
headers: { headers: {
@@ -33,10 +51,19 @@ export async function GET(req: NextRequest) {
"Cache-Control": "no-store", "Cache-Control": "no-store",
}, },
}); });
return applyAuthResponseCookies(response, auth.response); return finishProxyTimedResponse(
applyAuthResponseCookies(response, auth.response),
timer,
res.ok ? "ok" : `upstream_${res.status}`,
{ backendServerTiming },
);
} catch (error) { } catch (error) {
return buildProxyExceptionResponse(error, { return finishProxyTimedResponse(
publicMessage: "Failed to fetch online users", buildProxyExceptionResponse(error, {
}); publicMessage: "Failed to fetch online users",
}),
timer,
"exception",
);
} }
} }
+13 -3
View File
@@ -2,6 +2,10 @@ import { NextRequest, NextResponse } from "next/server";
import { proxyBackendJsonGet } from "@/lib/api-proxy"; import { proxyBackendJsonGet } from "@/lib/api-proxy";
import { buildForceRefreshProxyCachePolicy } from "@/lib/proxy-cache-policy"; import { buildForceRefreshProxyCachePolicy } from "@/lib/proxy-cache-policy";
import { DASHBOARD_REFRESH_POLICY_SEC } from "@/lib/refresh-policy"; import { DASHBOARD_REFRESH_POLICY_SEC } from "@/lib/refresh-policy";
import {
createProxyTimer,
finishProxyTimedResponse,
} from "@/lib/proxy-timing";
const API_BASE = process.env.POLYWEATHER_API_BASE_URL; const API_BASE = process.env.POLYWEATHER_API_BASE_URL;
const SCAN_TERMINAL_PROXY_TIMEOUT_MS = Number( const SCAN_TERMINAL_PROXY_TIMEOUT_MS = Number(
@@ -11,10 +15,15 @@ const SCAN_TERMINAL_PROXY_TIMEOUT_MS = Number(
export const maxDuration = 45; export const maxDuration = 45;
export async function GET(req: NextRequest) { export async function GET(req: NextRequest) {
const timer = createProxyTimer(req, "scan_terminal");
if (!API_BASE) { if (!API_BASE) {
return NextResponse.json( return finishProxyTimedResponse(
{ error: "POLYWEATHER_API_BASE_URL is not configured" }, NextResponse.json(
{ status: 500 }, { error: "POLYWEATHER_API_BASE_URL is not configured" },
{ status: 500 },
),
timer,
"missing_api_base",
); );
} }
@@ -61,6 +70,7 @@ export async function GET(req: NextRequest) {
revalidateSeconds: cachePolicy.revalidateSeconds, revalidateSeconds: cachePolicy.revalidateSeconds,
signal: controller.signal, signal: controller.signal,
timeoutPublicMessage: "Scan terminal request timed out", timeoutPublicMessage: "Scan terminal request timed out",
timing: timer,
url, url,
}); });
} finally { } finally {
@@ -55,7 +55,10 @@ import {
mergeAccessStateWithAuthPayload, mergeAccessStateWithAuthPayload,
type AuthProfilePayload, type AuthProfilePayload,
} from "@/components/dashboard/scan-terminal/terminal-access-state"; } from "@/components/dashboard/scan-terminal/terminal-access-state";
import { loadTerminalAuthProfile } from "@/components/dashboard/scan-terminal/terminal-auth-bootstrap"; import {
createAuthProfileRequestCache,
loadTerminalAuthProfile,
} from "@/components/dashboard/scan-terminal/terminal-auth-bootstrap";
import { import {
cityListItemsToScanRows, cityListItemsToScanRows,
mergeScanRowsWithCityFallbackRows, mergeScanRowsWithCityFallbackRows,
@@ -962,7 +965,7 @@ function ScanTerminalScreen() {
createEmptyAccess(true), createEmptyAccess(true),
); );
const loadAuthProfile = useCallback( const rawLoadAuthProfile = useCallback(
async ( async (
accessToken?: string | null, accessToken?: string | null,
options?: { preferSnapshot?: boolean }, options?: { preferSnapshot?: boolean },
@@ -984,6 +987,28 @@ function ScanTerminalScreen() {
}, },
[], [],
); );
const authProfileRequestCacheRef = useRef<{
load: typeof rawLoadAuthProfile;
cached: ReturnType<typeof createAuthProfileRequestCache>;
} | null>(null);
const loadAuthProfile = useCallback(
(
accessToken?: string | null,
options?: { preferSnapshot?: boolean },
): Promise<AuthProfilePayload> => {
const current = authProfileRequestCacheRef.current;
if (current?.load === rawLoadAuthProfile) {
return current.cached(accessToken, options);
}
const next = {
load: rawLoadAuthProfile,
cached: createAuthProfileRequestCache(rawLoadAuthProfile),
};
authProfileRequestCacheRef.current = next;
return next.cached(accessToken, options);
},
[rawLoadAuthProfile],
);
const refreshLiveAuthProfile = useCallback(async () => { const refreshLiveAuthProfile = useCallback(async () => {
const supabaseEnabled = hasSupabasePublicEnv(); const supabaseEnabled = hasSupabasePublicEnv();
@@ -12,6 +12,7 @@ import {
import { import {
MAX_HOURLY_DETAIL_CONCURRENT_REQUESTS, MAX_HOURLY_DETAIL_CONCURRENT_REQUESTS,
HOURLY_CACHE_TTL_MS, HOURLY_CACHE_TTL_MS,
__resolveCityDetailFromBatchForTest,
__readHourlyCacheEntryForTest, __readHourlyCacheEntryForTest,
__resetHourlyDetailRequestQueueForTest, __resetHourlyDetailRequestQueueForTest,
__runQueuedHourlyDetailRequestForTest, __runQueuedHourlyDetailRequestForTest,
@@ -95,7 +96,19 @@ export async function runTests() {
chartLogicSource.includes("primeCityDetailCache"), chartLogicSource.includes("primeCityDetailCache"),
"visible terminal chart detail fetches should be coalesced into one batch request and prime the shared chart cache", "visible terminal chart detail fetches should be coalesced into one batch request and prime the shared chart cache",
); );
const fetchHourlyBlock = chartLogicSource.match(/async function fetchHourlyForecastForCity[\s\S]*?\n}\n\nfunction fetchCityDetailWithTimeout/)?.[0] || ""; assert(
chartLogicSource.includes("partial?: boolean") &&
chartLogicSource.includes("missing?: string[]"),
"frontend city detail batch payload should understand partial responses and missing city markers",
);
const flushCityDetailBatchBlock = chartLogicSource.match(/async function flushCityDetailBatch[\s\S]*?\r?\n}\r?\n\r?\nfunction fetchCityDetailBatchWithTimeout/)?.[0] || "";
assert(
flushCityDetailBatchBlock.includes("partialMissingCities") &&
flushCityDetailBatchBlock.includes("resolveBatchWaiters(waiters, null)") &&
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( assert(
fetchHourlyBlock.includes("queueCityDetailBatch(city, resParam)") && fetchHourlyBlock.includes("queueCityDetailBatch(city, resParam)") &&
!fetchHourlyBlock.includes("runQueuedHourlyDetailRequest"), !fetchHourlyBlock.includes("runQueuedHourlyDetailRequest"),
@@ -130,6 +143,19 @@ export async function runTests() {
__shouldFetchCityDetailForChartForTest({ city: "paris", documentHidden: true, isChartVisible: true }) === false, __shouldFetchCityDetailForChartForTest({ city: "paris", documentHidden: true, isChartVisible: true }) === false,
"hidden browser tabs should not prefetch city detail", "hidden browser tabs should not prefetch city detail",
); );
const normalizedBatchDetail = __resolveCityDetailFromBatchForTest(
{
"hong kong": {
city: "hong kong",
timeseries: { hourly: { times: ["00:00"], temps: [32] } },
},
} as any,
"Hong Kong",
) as any;
assert(
normalizedBatchDetail?.city === "hong kong",
"frontend detail batch lookup should accept backend-normalized city keys before falling back to single-city requests",
);
__resetHourlyDetailRequestQueueForTest(); __resetHourlyDetailRequestQueueForTest();
let activeRequests = 0; let activeRequests = 0;
@@ -1,4 +1,5 @@
import { import {
createAuthProfileRequestCache,
loadTerminalAuthProfile, loadTerminalAuthProfile,
type TerminalAuthProfilePayload, type TerminalAuthProfilePayload,
} from "@/components/dashboard/scan-terminal/terminal-auth-bootstrap"; } from "@/components/dashboard/scan-terminal/terminal-auth-bootstrap";
@@ -27,6 +28,41 @@ async function flushMicrotasks() {
} }
export async function runTests() { export async function runTests() {
const pendingProfile = deferred<TerminalAuthProfilePayload>();
let underlyingProfileLoads = 0;
const cachedLoadAuthProfile = createAuthProfileRequestCache((
accessToken?: string | null,
options?: { preferSnapshot?: boolean },
) => {
underlyingProfileLoads += 1;
assert(
accessToken === "shared-token" && options?.preferSnapshot === true,
"auth profile request cache should pass through the original token and snapshot preference",
);
return pendingProfile.promise;
});
const firstCachedProfile = cachedLoadAuthProfile("shared-token", {
preferSnapshot: true,
});
const secondCachedProfile = cachedLoadAuthProfile("shared-token", {
preferSnapshot: true,
});
assert(
firstCachedProfile === secondCachedProfile && underlyingProfileLoads === 1,
"auth profile request cache should dedupe identical in-flight profile loads",
);
pendingProfile.resolve({
authenticated: true,
user_id: "shared-user",
subscription_active: true,
});
await firstCachedProfile;
await cachedLoadAuthProfile("shared-token", { preferSnapshot: true });
assert(
underlyingProfileLoads === 2,
"auth profile request cache should clear a key after the in-flight load settles",
);
const slowCookieProfile = deferred<TerminalAuthProfilePayload>(); const slowCookieProfile = deferred<TerminalAuthProfilePayload>();
const fastSession = deferred<{ data: { session: { access_token: string } } }>(); const fastSession = deferred<{ data: { session: { access_token: string } } }>();
const calls: string[] = []; const calls: string[] = [];
@@ -966,8 +966,11 @@ type HourlyForecastFetchOptions = {
}; };
type CityDetailBatchPayload = { type CityDetailBatchPayload = {
cities?: string[];
details?: Record<string, CityDetail | null | undefined>; details?: Record<string, CityDetail | null | undefined>;
errors?: Record<string, string>; errors?: Record<string, string>;
missing?: string[];
partial?: boolean;
}; };
type CityDetailBatchWaiter = { type CityDetailBatchWaiter = {
@@ -1073,6 +1076,30 @@ function rejectBatchWaiters(
(waiters || []).forEach((waiter) => waiter.reject(reason)); (waiters || []).forEach((waiter) => waiter.reject(reason));
} }
function resolveCityDetailFromBatch(
details: Record<string, CityDetail | null | undefined> | undefined,
city: string,
) {
if (!details) return undefined;
const trimmed = String(city || "").trim();
const direct =
details[city] ||
details[trimmed] ||
details[trimmed.toLowerCase()] ||
details[normalizeCityKey(trimmed)];
if (direct) return direct;
const requestedKey = normalizeCityKey(trimmed);
if (!requestedKey) return undefined;
for (const [key, detail] of Object.entries(details)) {
if (!detail) continue;
if (normalizeCityKey(key) === requestedKey) return detail;
const detailCity = (detail as any).city || detail.name || detail.display_name;
if (normalizeCityKey(detailCity) === requestedKey) return detail;
}
return undefined;
}
async function flushCityDetailBatch(resolution: string) { async function flushCityDetailBatch(resolution: string) {
const queue = _cityDetailBatchQueues.get(resolution); const queue = _cityDetailBatchQueues.get(resolution);
if (!queue) return; if (!queue) return;
@@ -1088,15 +1115,23 @@ async function flushCityDetailBatch(resolution: string) {
try { try {
const payload = await fetchCityDetailBatchWithTimeout(cities, resolution); const payload = await fetchCityDetailBatchWithTimeout(cities, resolution);
const details = payload?.details || {}; const details = payload?.details || {};
const partialMissingCities =
payload?.partial === true
? new Set((payload.missing || []).map((city) => normalizeCityKey(city)))
: new Set<string>();
await Promise.all( await Promise.all(
cities.map(async (city) => { cities.map(async (city) => {
const waiters = queue.waiters.get(city); const waiters = queue.waiters.get(city);
const detail = details[city]; const detail = resolveCityDetailFromBatch(details, city);
const data = primeCityDetailCache(city, resolution, detail); const data = primeCityDetailCache(city, resolution, detail);
if (data) { if (data) {
resolveBatchWaiters(waiters, data); resolveBatchWaiters(waiters, data);
return; return;
} }
if (partialMissingCities.has(normalizeCityKey(city))) {
resolveBatchWaiters(waiters, null);
return;
}
try { try {
resolveBatchWaiters( resolveBatchWaiters(
waiters, waiters,
@@ -2426,6 +2461,7 @@ export {
HOURLY_CACHE_TTL_MS, HOURLY_CACHE_TTL_MS,
_hourlyCache, _hourlyCache,
__readHourlyCacheEntryForTest, __readHourlyCacheEntryForTest,
resolveCityDetailFromBatch as __resolveCityDetailFromBatchForTest,
__resetHourlyDetailRequestQueueForTest, __resetHourlyDetailRequestQueueForTest,
__runQueuedHourlyDetailRequestForTest, __runQueuedHourlyDetailRequestForTest,
buildChartDomain, buildChartDomain,
@@ -57,6 +57,32 @@ function canResolveProfileImmediately(
return payload?.authenticated === true && payload.subscription_active === true; return payload?.authenticated === true && payload.subscription_active === true;
} }
function authProfileRequestCacheKey(
accessToken?: string | null,
options?: { preferSnapshot?: boolean },
) {
const token = String(accessToken || "").trim();
const scope = token ? `bearer:${token}` : "cookie";
const mode = options?.preferSnapshot ? "snapshot" : "live";
return `${mode}:${scope}`;
}
export function createAuthProfileRequestCache(
loadAuthProfile: LoadTerminalAuthProfileOptions["loadAuthProfile"],
): LoadTerminalAuthProfileOptions["loadAuthProfile"] {
const pending = new Map<string, Promise<TerminalAuthProfilePayload>>();
return (accessToken, options) => {
const key = authProfileRequestCacheKey(accessToken, options);
const existing = pending.get(key);
if (existing) return existing;
const request = loadAuthProfile(accessToken, options).finally(() => {
pending.delete(key);
});
pending.set(key, request);
return request;
};
}
export async function loadTerminalAuthProfile({ export async function loadTerminalAuthProfile({
getSession, getSession,
hasSupabasePublicEnv, hasSupabasePublicEnv,
@@ -0,0 +1,97 @@
import assert from "node:assert/strict";
import fs from "node:fs";
import path from "node:path";
function readFrontend(...parts: string[]) {
return fs.readFileSync(path.join(process.cwd(), ...parts), "utf8");
}
export function runTests() {
const timingSource = readFrontend("lib", "proxy-timing.ts");
assert.match(
timingSource,
/createProxyTimer/,
"shared proxy timing helper should create timers for slow API proxies",
);
assert.match(
timingSource,
/Server-Timing/,
"shared proxy timing helper should write Server-Timing headers for HAR inspection",
);
assert.doesNotMatch(
timingSource,
/authUserId|authEmail|userId|email/,
"proxy timing logs must avoid raw user ids or emails",
);
const apiProxySource = readFrontend("lib", "api-proxy.ts");
assert.match(
apiProxySource,
/timing\?: ProxyTimer/,
"generic backend JSON proxy should accept an optional timer",
);
for (const stage of ["auth_headers", "backend_fetch", "backend_read"]) {
assert.match(
apiProxySource,
new RegExp(stage),
`generic backend JSON proxy should measure ${stage}`,
);
}
const detailBatchProxy = readFrontend("app", "api", "cities", "detail-batch", "route.ts");
assert.match(detailBatchProxy, /createProxyTimer\(req,\s*"city_detail_batch"\)/);
assert.match(detailBatchProxy, /timing:\s*timer/);
assert.match(
detailBatchProxy,
/fetchCache:\s*"no-store"/,
"city detail batch proxy should avoid caching partial backend fetches in the Next data cache",
);
assert.match(
detailBatchProxy,
/cacheControlForData/,
"city detail batch proxy should be able to suppress response caching for partial payloads",
);
assert.match(
apiProxySource,
/cacheControlForData\?:/,
"generic backend JSON proxy should allow response cache policy to depend on parsed data",
);
const scanTerminalProxy = readFrontend("app", "api", "scan", "terminal", "route.ts");
assert.match(scanTerminalProxy, /createProxyTimer\(req,\s*"scan_terminal"\)/);
assert.match(scanTerminalProxy, /timing:\s*timer/);
const cityDetailProxy = readFrontend("app", "api", "city", "[name]", "detail", "route.ts");
assert.match(cityDetailProxy, /createProxyTimer\(req,\s*"city_detail"\)/);
for (const stage of ["auth_headers", "backend_fetch", "backend_read"]) {
assert.match(cityDetailProxy, new RegExp(stage));
}
const onlineUsersProxy = readFrontend("app", "api", "ops", "online-users", "route.ts");
assert.match(onlineUsersProxy, /createProxyTimer\(req,\s*"ops_online_users"\)/);
for (const stage of ["auth_headers", "ops_auth", "backend_fetch", "backend_read"]) {
assert.match(onlineUsersProxy, new RegExp(stage));
}
const analyticsProxy = readFrontend("app", "api", "analytics", "events", "route.ts");
assert.match(
analyticsProxy,
/ANALYTICS_PROXY_TIMEOUT_MS/,
"analytics event proxy should use a short dedicated timeout instead of waiting for long backend stalls",
);
assert.match(
analyticsProxy,
/createProxyTimer\(req,\s*"analytics_events"\)/,
"analytics event proxy should expose Server-Timing for HAR inspection",
);
assert.match(
analyticsProxy,
/AbortController/,
"analytics event proxy should abort slow upstream tracking requests",
);
assert.match(
analyticsProxy,
/status:\s*202/,
"analytics event proxy timeout should stay non-blocking for the fire-and-forget client event",
);
}
+62 -19
View File
@@ -5,6 +5,10 @@ import {
buildBackendRequestHeaders, buildBackendRequestHeaders,
} from "@/lib/backend-auth"; } from "@/lib/backend-auth";
import { buildCachedJsonResponse } from "@/lib/http-cache"; import { buildCachedJsonResponse } from "@/lib/http-cache";
import {
finishProxyTimedResponse,
type ProxyTimer,
} from "@/lib/proxy-timing";
const PASSTHROUGH_UPSTREAM_STATUSES = new Set([ const PASSTHROUGH_UPSTREAM_STATUSES = new Set([
400, 400,
@@ -81,6 +85,7 @@ export async function proxyBackendJsonGet(
req: NextRequest, req: NextRequest,
options: { options: {
cacheControl?: string; cacheControl?: string;
cacheControlForData?: (data: unknown) => string | undefined;
conditionalResponse?: boolean; conditionalResponse?: boolean;
detailLimit?: number; detailLimit?: number;
error?: string; error?: string;
@@ -91,40 +96,75 @@ export async function proxyBackendJsonGet(
signal?: AbortSignal; signal?: AbortSignal;
statusOnException?: number; statusOnException?: number;
timeoutPublicMessage?: string; timeoutPublicMessage?: string;
timing?: ProxyTimer;
url: string; url: string;
}, },
) { ) {
let auth: Awaited<ReturnType<typeof buildBackendRequestHeaders>> | null = null; let auth: Awaited<ReturnType<typeof buildBackendRequestHeaders>> | null = null;
const timing = options.timing;
try { try {
auth = await buildBackendRequestHeaders(req, { auth = await (timing
includeSupabaseIdentity: options.includeSupabaseIdentity ?? false, ? timing.measure("auth_headers", () =>
}); buildBackendRequestHeaders(req, {
const res = await fetch(options.url, { includeSupabaseIdentity: options.includeSupabaseIdentity ?? false,
headers: auth.headers, }),
...(options.fetchCache )
? { cache: options.fetchCache } : buildBackendRequestHeaders(req, {
: { next: { revalidate: options.revalidateSeconds ?? 30 } }), includeSupabaseIdentity: options.includeSupabaseIdentity ?? false,
signal: options.signal, }));
}); const res = await (timing
? timing.measure("backend_fetch", () =>
fetch(options.url, {
headers: auth!.headers,
...(options.fetchCache
? { cache: options.fetchCache }
: { next: { revalidate: options.revalidateSeconds ?? 30 } }),
signal: options.signal,
}),
)
: fetch(options.url, {
headers: auth.headers,
...(options.fetchCache
? { cache: options.fetchCache }
: { next: { revalidate: options.revalidateSeconds ?? 30 } }),
signal: options.signal,
}));
const backendServerTiming = res.headers.get("server-timing") || "";
if (!res.ok) { if (!res.ok) {
const raw = await res.text(); const raw = await (timing
? timing.measure("backend_read", () => res.text())
: res.text());
const response = buildUpstreamErrorResponse(res.status, raw, { const response = buildUpstreamErrorResponse(res.status, raw, {
detailLimit: options.detailLimit, detailLimit: options.detailLimit,
error: options.error, error: options.error,
}); });
return applyAuthResponseCookies(response, auth.response); const withCookies = applyAuthResponseCookies(response, auth.response);
return timing
? finishProxyTimedResponse(withCookies, timing, `upstream_${res.status}`, {
backendServerTiming,
})
: withCookies;
} }
const data = await res.json(); const data = await (timing
? timing.measure("backend_read", () => res.json())
: res.json());
const responseCacheControl =
options.cacheControlForData?.(data) ?? options.cacheControl;
const response = const response =
options.cacheControl && options.conditionalResponse !== false responseCacheControl && options.conditionalResponse !== false
? buildCachedJsonResponse(req, data, options.cacheControl) ? buildCachedJsonResponse(req, data, responseCacheControl)
: NextResponse.json(data, { : NextResponse.json(data, {
headers: options.cacheControl headers: responseCacheControl
? { "Cache-Control": options.cacheControl } ? { "Cache-Control": responseCacheControl }
: undefined, : undefined,
}); });
return applyAuthResponseCookies(response, auth.response); const withCookies = applyAuthResponseCookies(response, auth.response);
return timing
? finishProxyTimedResponse(withCookies, timing, "ok", {
backendServerTiming,
})
: withCookies;
} catch (error) { } catch (error) {
const timedOut = options.signal?.aborted === true; const timedOut = options.signal?.aborted === true;
const response = buildProxyExceptionResponse(error, { const response = buildProxyExceptionResponse(error, {
@@ -134,6 +174,9 @@ export async function proxyBackendJsonGet(
: options.publicMessage, : options.publicMessage,
status: timedOut ? 504 : options.statusOnException, status: timedOut ? 504 : options.statusOnException,
}); });
return auth ? applyAuthResponseCookies(response, auth.response) : response; const withCookies = auth ? applyAuthResponseCookies(response, auth.response) : response;
return timing
? finishProxyTimedResponse(withCookies, timing, timedOut ? "timeout" : "exception")
: withCookies;
} }
} }
+1
View File
@@ -716,6 +716,7 @@ export interface AiAnalysisStructured {
} }
export interface CityDetail { export interface CityDetail {
city?: string;
name: string; name: string;
display_name: string; display_name: string;
detail_depth?: "panel" | "market" | "nearby" | "full"; detail_depth?: "panel" | "market" | "nearby" | "full";
+113
View File
@@ -0,0 +1,113 @@
import type { NextRequest, NextResponse } from "next/server";
export type ProxyTimingStage = {
durationMs: number;
name: string;
};
export type ProxyTimer = {
hasAuthorization: boolean;
hasSupabaseCookie: boolean;
measure<T>(name: string, action: () => Promise<T>): Promise<T>;
measureSync<T>(name: string, action: () => T): T;
route: string;
stages: ProxyTimingStage[];
totalMs(): number;
};
function proxyNowMs() {
return typeof performance !== "undefined" ? performance.now() : Date.now();
}
function hasRequestSupabaseSessionCookie(req: NextRequest) {
return req.cookies.getAll().some((cookie) => {
const name = cookie.name.toLowerCase();
const value = String(cookie.value || "").trim();
return Boolean(
value &&
(name === "supabase-auth-token" ||
(name.startsWith("sb-") && name.includes("-auth-token"))),
);
});
}
export function createProxyTimer(req: NextRequest, route: string): ProxyTimer {
const startedAt = proxyNowMs();
const stages: ProxyTimingStage[] = [];
const recordStage = (name: string, stageStartedAt: number) => {
stages.push({
durationMs: Math.round((proxyNowMs() - stageStartedAt) * 10) / 10,
name,
});
};
return {
hasAuthorization: Boolean(req.headers.get("authorization")),
hasSupabaseCookie: hasRequestSupabaseSessionCookie(req),
async measure<T>(name: string, action: () => Promise<T>) {
const stageStartedAt = proxyNowMs();
try {
return await action();
} finally {
recordStage(name, stageStartedAt);
}
},
measureSync<T>(name: string, action: () => T) {
const stageStartedAt = proxyNowMs();
try {
return action();
} finally {
recordStage(name, stageStartedAt);
}
},
route,
stages,
totalMs() {
return Math.round((proxyNowMs() - startedAt) * 10) / 10;
},
};
}
function formatServerTiming(stages: ProxyTimingStage[]) {
return stages
.map(({ durationMs, name }) => {
const safeName = name.replace(/[^A-Za-z0-9_-]/g, "_");
return `${safeName};dur=${Math.max(0, durationMs).toFixed(1)}`;
})
.join(", ");
}
export function finishProxyTimedResponse(
response: NextResponse,
timer: ProxyTimer,
outcome: string,
extra?: { backendServerTiming?: string },
) {
const total = timer.totalMs();
const ownServerTiming = formatServerTiming(
[...timer.stages, { durationMs: total, name: "total" }].map((stage) => ({
...stage,
name: `${timer.route}_${stage.name}`,
})),
);
const backendServerTiming = String(extra?.backendServerTiming || "").trim();
response.headers.set(
"Server-Timing",
backendServerTiming
? `${ownServerTiming}, ${backendServerTiming}`
: ownServerTiming,
);
console.info(
"[api-proxy-timing]",
JSON.stringify({
hasAuthorization: timer.hasAuthorization,
hasSupabaseCookie: timer.hasSupabaseCookie,
outcome,
route: timer.route,
stages: timer.stages,
status: response.status,
totalMs: total,
}),
);
return response;
}
+116
View File
@@ -0,0 +1,116 @@
from pathlib import Path
from fastapi.testclient import TestClient
from web.app import app
import web.services.city_api as city_api
import web.services.scan_api as scan_api
ROOT = Path(__file__).resolve().parents[1]
client = TestClient(app)
def test_backend_shared_timing_helper_avoids_sensitive_identity_fields():
source = (ROOT / "web" / "services" / "request_timing.py").read_text(
encoding="utf-8"
)
assert "ServerTimingRecorder" in source
assert "server_timing_value" in source
assert "user_id" not in source
assert "email" not in source
def test_city_detail_batch_response_includes_backend_server_timing(monkeypatch):
class FakeCache:
def get_city_cache(self, kind, city):
assert kind == "full"
return {"payload": {"city": city, "hourly": {"times": [], "temps": []}}}
def build_detail(data, market_slug, target_date, resolution):
return {"city": data["city"], "resolution": resolution}
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,
)
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")
assert response.status_code == 200
server_timing = response.headers["server-timing"]
assert "city_detail_batch_assert_entitlement" in server_timing
assert "city_detail_batch_full_data_paris" in server_timing
assert "city_detail_batch_detail_payload_paris" in server_timing
assert "city_detail_batch_total" in server_timing
def test_city_detail_response_includes_backend_server_timing(monkeypatch):
class FakeCache:
def get_city_cache(self, kind, city):
assert kind == "full"
return {"payload": {"city": city, "hourly": {"times": [], "temps": []}}}
def build_detail(data, market_slug, target_date, resolution):
return {"city": data["city"], "resolution": resolution}
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,
)
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/city/Paris/detail?resolution=10m")
assert response.status_code == 200
server_timing = response.headers["server-timing"]
assert "city_detail_assert_entitlement" in server_timing
assert "city_detail_full_data" in server_timing
assert "city_detail_detail_payload" in server_timing
assert "city_detail_total" in server_timing
def test_scan_terminal_response_includes_backend_server_timing(monkeypatch):
monkeypatch.setattr(scan_api.legacy_routes, "_assert_entitlement", lambda request: None)
monkeypatch.setattr(
scan_api.legacy_routes,
"build_scan_terminal_payload",
lambda filters, force_refresh=False, timing_recorder=None: {"rows": [], "filters": filters},
)
response = client.get("/api/scan/terminal?limit=1")
assert response.status_code == 200
server_timing = response.headers["server-timing"]
assert "scan_terminal_assert_entitlement" in server_timing
assert "scan_terminal_build_payload" in server_timing
assert "scan_terminal_total" in server_timing
def test_online_users_response_includes_backend_server_timing():
response = client.get("/api/ops/online-users")
assert response.status_code == 200
server_timing = response.headers["server-timing"]
assert "ops_online_users_online_count" in server_timing
assert "ops_online_users_total" in server_timing
+9
View File
@@ -80,6 +80,15 @@ def test_deploy_script_retries_startup_smoke_checks():
assert 'smoke_check "frontend" "https://www.polyweather.top/" 15 3 5' in script assert 'smoke_check "frontend" "https://www.polyweather.top/" 15 3 5' in script
def test_docker_compose_keeps_polyweather_ports_on_loopback():
compose = (ROOT / "docker-compose.yml").read_text(encoding="utf-8")
assert "127.0.0.1:3001:3000" in compose
assert "127.0.0.1:8000:8000" in compose
assert "\n - 3001:3000" not in compose
assert "\n - 8000:8000" not in compose
def test_city_detail_builds_deb_hourly_consensus_before_peak_window(): def test_city_detail_builds_deb_hourly_consensus_before_peak_window():
source = (ROOT / "web" / "analysis_service.py").read_text(encoding="utf-8") source = (ROOT / "web" / "analysis_service.py").read_text(encoding="utf-8")
+102
View File
@@ -618,6 +618,44 @@ def test_city_detail_batch_endpoint_limits_backend_concurrency(monkeypatch):
assert max_active <= 2 assert max_active <= 2
def test_city_detail_batch_returns_completed_details_when_one_city_is_slow(monkeypatch):
import asyncio
completed = []
async def build_batch_item(city, **kwargs):
if city == "slow":
await asyncio.sleep(0.08)
completed.append(city)
return city, {
"city": city,
"hourly": {"times": ["2026-05-30T00:00:00Z"], "temps": [20.0]},
"resolution": kwargs.get("resolution"),
}
monkeypatch.setenv("POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS", "20")
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, "_build_city_detail_batch_item_async", build_batch_item)
payload = asyncio.run(
city_api.get_city_detail_batch_payload(
object(),
cities="fast,slow,other",
resolution="10m",
limit=3,
)
)
assert payload["cities"] == ["fast", "slow", "other"]
assert sorted(payload["details"]) == ["fast", "other"]
assert payload["details"]["fast"]["resolution"] == "10m"
assert payload["partial"] is True
assert payload["missing"] == ["slow"]
assert payload["errors"] == {}
assert "slow" not in completed
def test_concurrent_city_detail_requests_share_same_full_cache_refresh(monkeypatch): def test_concurrent_city_detail_requests_share_same_full_cache_refresh(monkeypatch):
import asyncio import asyncio
@@ -678,6 +716,70 @@ def test_concurrent_city_detail_requests_share_same_full_cache_refresh(monkeypat
assert build_calls == 1 assert build_calls == 1
def test_stale_city_detail_uses_cached_full_payload_while_refreshing(monkeypatch):
import asyncio
refresh_calls = 0
build_inputs = []
class FakeCache:
def get_city_cache(self, kind, city):
assert kind == "full"
assert city == "paris"
return {
"payload": {
"city": "paris",
"hourly": {"times": ["2026-05-30T00:00:00Z"], "temps": [20.0]},
},
}
async def fake_run_in_threadpool(fn, *args, **kwargs):
if fn is city_api.legacy_routes._refresh_city_full_cache:
await asyncio.sleep(0.01)
return fn(*args, **kwargs)
def refresh_full(city, force_refresh):
nonlocal refresh_calls
refresh_calls += 1
return {
"city": city,
"hourly": {"times": ["2026-05-30T00:00:00Z"], "temps": [21.0]},
}
def build_detail(data, market_slug, target_date, resolution):
build_inputs.append(data["hourly"]["temps"][0])
return {
"city": data["city"],
"live_temp": data["hourly"]["temps"][0],
"resolution": resolution,
}
city_api._CITY_FULL_REFRESH_INFLIGHT.clear()
city_api._CITY_DETAIL_PAYLOAD_CACHE.clear()
city_api._CITY_DETAIL_PAYLOAD_CACHE_TS.clear()
city_api._CITY_DETAIL_PAYLOAD_INFLIGHT.clear()
monkeypatch.setattr(city_api, "run_in_threadpool", fake_run_in_threadpool)
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, "_CACHE_DB", FakeCache())
monkeypatch.setattr(city_api.legacy_routes, "_city_cache_is_fresh", lambda entry, ttl: False)
monkeypatch.setattr(city_api.legacy_routes, "_overlay_latest_wunderground_current", lambda city, payload: payload)
monkeypatch.setattr(city_api.legacy_routes, "_refresh_city_full_cache", refresh_full)
monkeypatch.setattr(city_api.legacy_routes, "_build_city_detail_payload", build_detail)
async def run_request():
payload = await city_api.get_city_detail_aggregate_payload(object(), "Paris", resolution="10m")
await asyncio.sleep(0.03)
return payload
result = asyncio.run(run_request())
assert result["live_temp"] == 20.0
assert build_inputs == [20.0]
assert refresh_calls == 1
def test_force_refresh_invalidates_short_city_detail_payload_cache(monkeypatch): def test_force_refresh_invalidates_short_city_detail_payload_cache(monkeypatch):
import asyncio import asyncio
+10 -3
View File
@@ -2,7 +2,7 @@
from typing import Any, Dict, List, Optional from typing import Any, Dict, List, Optional
from fastapi import APIRouter, BackgroundTasks, Query, Request from fastapi import APIRouter, BackgroundTasks, Query, Request, Response
from web.services.city_api import ( from web.services.city_api import (
get_city_detail_batch_payload, get_city_detail_batch_payload,
@@ -12,6 +12,7 @@ from web.services.city_api import (
list_cities_payload, list_cities_payload,
) )
from web.services.city_realtime_stream import get_realtime_stream_payload from web.services.city_realtime_stream import get_realtime_stream_payload
from web.services.request_timing import attach_server_timing_header
router = APIRouter(tags=["city"]) router = APIRouter(tags=["city"])
@@ -108,6 +109,7 @@ async def cities_model_range(
@router.get("/api/cities/detail-batch") @router.get("/api/cities/detail-batch")
async def city_detail_batch( async def city_detail_batch(
request: Request, request: Request,
response: Response,
cities: str = "", cities: str = "",
force_refresh: bool = False, force_refresh: bool = False,
market_slug: Optional[str] = None, market_slug: Optional[str] = None,
@@ -115,7 +117,7 @@ async def city_detail_batch(
resolution: Optional[str] = "10m", resolution: Optional[str] = "10m",
limit: int = 12, limit: int = 12,
): ):
return await get_city_detail_batch_payload( payload = await get_city_detail_batch_payload(
request, request,
cities=cities, cities=cities,
force_refresh=force_refresh, force_refresh=force_refresh,
@@ -124,6 +126,8 @@ async def city_detail_batch(
resolution=resolution, resolution=resolution,
limit=limit, limit=limit,
) )
attach_server_timing_header(response, request, "city_detail_batch_server_timing")
return payload
@router.get("/api/city/{name}") @router.get("/api/city/{name}")
@@ -159,13 +163,14 @@ async def city_summary(
@router.get("/api/city/{name}/detail") @router.get("/api/city/{name}/detail")
async def city_detail_aggregate( async def city_detail_aggregate(
request: Request, request: Request,
response: Response,
name: str, name: str,
force_refresh: bool = False, force_refresh: bool = False,
market_slug: Optional[str] = None, market_slug: Optional[str] = None,
target_date: Optional[str] = None, target_date: Optional[str] = None,
resolution: Optional[str] = "10m", resolution: Optional[str] = "10m",
): ):
return await get_city_detail_aggregate_payload( payload = await get_city_detail_aggregate_payload(
request, request,
name, name,
force_refresh=force_refresh, force_refresh=force_refresh,
@@ -173,6 +178,8 @@ async def city_detail_aggregate(
target_date=target_date, target_date=target_date,
resolution=resolution, resolution=resolution,
) )
attach_server_timing_header(response, request, "city_detail_server_timing")
return payload
@router.get("/api/city/{name}/realtime-stream") @router.get("/api/city/{name}/realtime-stream")
+25 -4
View File
@@ -1,6 +1,6 @@
"""Operations/admin API routes.""" """Operations/admin API routes."""
from fastapi import APIRouter, Request from fastapi import APIRouter, Request, Response
from web.core import GrantPointsRequest from web.core import GrantPointsRequest
from web.services.ops_api import ( from web.services.ops_api import (
@@ -30,14 +30,35 @@ from web.services.ops_api import (
get_ops_training_accuracy, get_ops_training_accuracy,
get_ops_telegram_audit, get_ops_telegram_audit,
) )
from web.services.request_timing import ServerTimingRecorder, attach_server_timing_header
router = APIRouter(tags=["ops"]) router = APIRouter(tags=["ops"])
@router.get("/api/ops/online-users") @router.get("/api/ops/online-users")
async def ops_online_users(): async def ops_online_users(request: Request, response: Response):
from src.utils.online_tracker import online_count timer = ServerTimingRecorder(
return {"online": online_count()} request,
log_name="ops_online_users_timing",
prefix="ops_online_users",
state_attr="ops_online_users_server_timing",
)
outcome = "ok"
status_code = 200
try:
from src.utils.online_tracker import online_count
return {"online": timer.measure("online_count", online_count)}
except Exception:
outcome = "exception"
status_code = 500
raise
finally:
timer.finish(outcome=outcome, status_code=status_code)
attach_server_timing_header(
response,
request,
"ops_online_users_server_timing",
)
@router.get("/api/ops/users") @router.get("/api/ops/users")
+4 -1
View File
@@ -9,6 +9,7 @@ from web.services.scan_api import (
get_scan_terminal_overview_payload, get_scan_terminal_overview_payload,
get_scan_terminal_payload, get_scan_terminal_payload,
) )
from web.services.request_timing import attach_server_timing_header
router = APIRouter(tags=["scan"]) router = APIRouter(tags=["scan"])
@@ -45,12 +46,14 @@ async def scan_terminal(
region=region or trading_region or None, region=region or trading_region or None,
timezone_offset_seconds=timezone_offset_seconds, timezone_offset_seconds=timezone_offset_seconds,
) )
return JSONResponse( response = JSONResponse(
content=payload, content=payload,
headers={ headers={
"Cache-Control": "public, s-maxage=30, stale-while-revalidate=120", "Cache-Control": "public, s-maxage=30, stale-while-revalidate=120",
}, },
) )
attach_server_timing_header(response, request, "scan_terminal_server_timing")
return response
@router.post("/api/scan/terminal/overview") @router.post("/api/scan/terminal/overview")
+40 -5
View File
@@ -235,16 +235,41 @@ def build_scan_terminal_payload(
raw_filters: Optional[Dict[str, Any]] = None, raw_filters: Optional[Dict[str, Any]] = None,
*, *,
force_refresh: bool = False, force_refresh: bool = False,
timing_recorder: Any = None,
) -> Dict[str, Any]: ) -> Dict[str, Any]:
filters = _normalize_scan_terminal_filters(raw_filters) filters = (
timing_recorder.measure(
"normalize_filters",
lambda: _normalize_scan_terminal_filters(raw_filters),
)
if timing_recorder is not None
else _normalize_scan_terminal_filters(raw_filters)
)
if not force_refresh: if not force_refresh:
cached = get_cached_scan_terminal_payload( cached = (
filters, ttl_sec=SCAN_TERMINAL_PAYLOAD_TTL_SEC timing_recorder.measure(
"cache_lookup",
lambda: get_cached_scan_terminal_payload(
filters,
ttl_sec=SCAN_TERMINAL_PAYLOAD_TTL_SEC,
),
)
if timing_recorder is not None
else get_cached_scan_terminal_payload(
filters, ttl_sec=SCAN_TERMINAL_PAYLOAD_TTL_SEC
)
) )
if cached is not None: if cached is not None:
return cached return cached
cached_entry = get_scan_terminal_cache_entry(filters) or {} cached_entry = (
timing_recorder.measure(
"stale_cache_lookup",
lambda: get_scan_terminal_cache_entry(filters) or {},
)
if timing_recorder is not None
else get_scan_terminal_cache_entry(filters) or {}
)
success_payload = cached_entry.get("success_payload") success_payload = cached_entry.get("success_payload")
if isinstance(success_payload, dict) and success_payload: if isinstance(success_payload, dict) and success_payload:
started = _start_scan_terminal_background_refresh(filters) started = _start_scan_terminal_background_refresh(filters)
@@ -257,7 +282,17 @@ def build_scan_terminal_payload(
failed_at=cached_entry.get("last_failed_at"), failed_at=cached_entry.get("last_failed_at"),
) )
return _build_scan_terminal_payload_uncached(filters, force_refresh=force_refresh) return (
timing_recorder.measure(
"uncached_build",
lambda: _build_scan_terminal_payload_uncached(
filters,
force_refresh=force_refresh,
),
)
if timing_recorder is not None
else _build_scan_terminal_payload_uncached(filters, force_refresh=force_refresh)
)
_SCAN_PREWARM_STARTED = False _SCAN_PREWARM_STARTED = False
+197 -52
View File
@@ -13,6 +13,7 @@ from fastapi.concurrency import run_in_threadpool
from loguru import logger from loguru import logger
import web.routes as legacy_routes import web.routes as legacy_routes
from web.services.request_timing import ServerTimingRecorder
_RECENT_DEB_CACHE: Optional[Dict[str, Dict[str, object]]] = None _RECENT_DEB_CACHE: Optional[Dict[str, Dict[str, object]]] = None
_RECENT_DEB_CACHE_TS = 0.0 _RECENT_DEB_CACHE_TS = 0.0
@@ -23,6 +24,7 @@ _RECENT_DEB_CACHE_TTL_SEC = max(
int(os.getenv("POLYWEATHER_CITIES_DEB_RECENT_CACHE_TTL_SEC", "300") or "300"), int(os.getenv("POLYWEATHER_CITIES_DEB_RECENT_CACHE_TTL_SEC", "300") or "300"),
) )
_CITY_FULL_REFRESH_INFLIGHT: Dict[str, "asyncio.Task[Dict[str, Any]]"] = {} _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() _CITY_FULL_REFRESH_LOCK = asyncio.Lock()
CityDetailPayloadCacheKey = Tuple[str, str, str, str, str, int] CityDetailPayloadCacheKey = Tuple[str, str, str, str, str, int]
_CITY_DETAIL_PAYLOAD_CACHE: Dict[CityDetailPayloadCacheKey, Dict[str, Any]] = {} _CITY_DETAIL_PAYLOAD_CACHE: Dict[CityDetailPayloadCacheKey, Dict[str, Any]] = {}
@@ -53,9 +55,17 @@ async def _refresh_city_full_cache_singleflight(city: str, force_refresh: bool)
async with _CITY_FULL_REFRESH_LOCK: async with _CITY_FULL_REFRESH_LOCK:
task = _CITY_FULL_REFRESH_INFLIGHT.get(key) task = _CITY_FULL_REFRESH_INFLIGHT.get(key)
if task is None: if task is None:
task = asyncio.create_task( async def _run_refresh() -> Dict[str, Any]:
run_in_threadpool(legacy_routes._refresh_city_full_cache, city, force_refresh), try:
) return await run_in_threadpool(
legacy_routes._refresh_city_full_cache,
city,
force_refresh,
)
finally:
await _invalidate_city_detail_payload_cache(city)
task = asyncio.create_task(_run_refresh())
_CITY_FULL_REFRESH_INFLIGHT[key] = task _CITY_FULL_REFRESH_INFLIGHT[key] = task
try: try:
return await task return await task
@@ -83,14 +93,40 @@ async def _refresh_city_full_data(city: str, force_refresh: bool) -> Dict[str, A
return await _refresh_city_full_cache_singleflight(city, force_refresh) return await _refresh_city_full_cache_singleflight(city, force_refresh)
def _start_city_full_stale_refresh(city: str) -> None:
normalized = str(city or "").strip().lower()
if not normalized:
return
existing = _CITY_FULL_STALE_REFRESH_TASKS.get(normalized)
if existing is not None and not existing.done():
return
task = asyncio.create_task(_refresh_city_full_data(city, False))
_CITY_FULL_STALE_REFRESH_TASKS[normalized] = task
def _cleanup(done: "asyncio.Task[Dict[str, Any]]") -> None:
if _CITY_FULL_STALE_REFRESH_TASKS.get(normalized) is done:
_CITY_FULL_STALE_REFRESH_TASKS.pop(normalized, None)
try:
done.result()
except Exception as exc: # pragma: no cover - defensive background guard
logger.warning("city full stale refresh failed city={}: {}", city, exc)
task.add_done_callback(_cleanup)
async def _get_city_full_data(city: str, *, force_refresh: bool) -> Dict[str, Any]: async def _get_city_full_data(city: str, *, force_refresh: bool) -> Dict[str, Any]:
if force_refresh: if force_refresh:
return await _refresh_city_full_data(city, True) return await _refresh_city_full_data(city, True)
cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, "full", city) cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, "full", city)
if cached_entry: if cached_entry:
payload = cached_entry.get("payload") or {}
if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_FULL_CACHE_TTL_SEC): if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_FULL_CACHE_TTL_SEC):
if payload:
_start_city_full_stale_refresh(city)
return await _overlay_cached_wunderground(city, payload)
return await _refresh_city_full_data(city, False) return await _refresh_city_full_data(city, False)
return await _overlay_cached_wunderground(city, cached_entry.get("payload") or {}) return await _overlay_cached_wunderground(city, payload)
return await _refresh_city_full_data(city, False) return await _refresh_city_full_data(city, False)
@@ -368,16 +404,41 @@ async def get_city_detail_aggregate_payload(
target_date: Optional[str] = None, target_date: Optional[str] = None,
resolution: Optional[str] = "10m", resolution: Optional[str] = "10m",
) -> Dict[str, Any]: ) -> Dict[str, Any]:
legacy_routes._assert_entitlement(request) timer = ServerTimingRecorder(
city = legacy_routes._normalize_city_or_404(name) request,
data = await _get_city_full_data(city, force_refresh=force_refresh) log_name="city_detail_timing",
prefix="city_detail",
return await _build_city_detail_payload_cached( state_attr="city_detail_server_timing",
data,
market_slug,
target_date,
resolution,
) )
outcome = "ok"
status_code = 200
try:
timer.measure("assert_entitlement", lambda: legacy_routes._assert_entitlement(request))
city = timer.measure("normalize_city", lambda: legacy_routes._normalize_city_or_404(name))
data = await timer.measure_async(
"full_data",
lambda: _get_city_full_data(city, force_refresh=force_refresh),
)
return await timer.measure_async(
"detail_payload",
lambda: _build_city_detail_payload_cached(
data,
market_slug,
target_date,
resolution,
),
)
except HTTPException as exc:
outcome = f"http_{exc.status_code}"
status_code = exc.status_code
raise
except Exception:
outcome = "exception"
status_code = 500
raise
finally:
timer.finish(outcome=outcome, status_code=status_code)
def _parse_batch_city_names(raw_cities: str, *, limit: int) -> List[str]: def _parse_batch_city_names(raw_cities: str, *, limit: int) -> List[str]:
@@ -404,14 +465,30 @@ async def _build_city_detail_batch_item_async(
market_slug: Optional[str], market_slug: Optional[str],
target_date: Optional[str], target_date: Optional[str],
resolution: Optional[str], resolution: Optional[str],
timing_recorder: Optional[ServerTimingRecorder] = None,
) -> Tuple[str, Dict[str, Any]]: ) -> Tuple[str, Dict[str, Any]]:
data = await _get_city_full_data(city, force_refresh=force_refresh) if timing_recorder is not None:
detail = await _build_city_detail_payload_cached( data = await timing_recorder.measure_async(
data, f"full_data_{city}",
market_slug, lambda: _get_city_full_data(city, force_refresh=force_refresh),
target_date, )
resolution, detail = await timing_recorder.measure_async(
) f"detail_payload_{city}",
lambda: _build_city_detail_payload_cached(
data,
market_slug,
target_date,
resolution,
),
)
else:
data = await _get_city_full_data(city, force_refresh=force_refresh)
detail = await _build_city_detail_payload_cached(
data,
market_slug,
target_date,
resolution,
)
return city, detail return city, detail
@@ -423,6 +500,19 @@ def _city_detail_batch_concurrency() -> int:
return max(1, min(6, value)) return max(1, min(6, value))
def _city_detail_batch_partial_timeout_seconds() -> Optional[float]:
try:
timeout_ms = int(
os.getenv("POLYWEATHER_CITY_DETAIL_BATCH_PARTIAL_TIMEOUT_MS", "8500")
or "8500"
)
except ValueError:
timeout_ms = 8500
if timeout_ms <= 0:
return None
return max(0.001, min(60.0, timeout_ms / 1000.0))
async def get_city_detail_batch_payload( async def get_city_detail_batch_payload(
request: Request, request: Request,
*, *,
@@ -433,39 +523,94 @@ async def get_city_detail_batch_payload(
resolution: Optional[str] = "10m", resolution: Optional[str] = "10m",
limit: int = 12, limit: int = 12,
) -> Dict[str, Any]: ) -> Dict[str, Any]:
legacy_routes._assert_entitlement(request) timer = ServerTimingRecorder(
city_names = _parse_batch_city_names(cities, limit=max(1, min(24, int(limit or 12)))) request,
if not city_names: log_name="city_detail_batch_timing",
return {"cities": [], "details": {}, "errors": {}} prefix="city_detail_batch",
state_attr="city_detail_batch_server_timing",
)
outcome = "ok"
status_code = 200
try:
timer.measure("assert_entitlement", lambda: legacy_routes._assert_entitlement(request))
city_names = timer.measure(
"parse_cities",
lambda: _parse_batch_city_names(
cities,
limit=max(1, min(24, int(limit or 12))),
),
)
if not city_names:
return {
"cities": [],
"details": {},
"errors": {},
"missing": [],
"partial": False,
}
semaphore = asyncio.Semaphore(_city_detail_batch_concurrency()) semaphore = asyncio.Semaphore(_city_detail_batch_concurrency())
async def _build_with_limit(city: str) -> Tuple[str, Dict[str, Any]]: async def _build_with_limit(city: str) -> Tuple[str, Dict[str, Any]]:
async with semaphore: async with semaphore:
return await _build_city_detail_batch_item_async( return await _build_city_detail_batch_item_async(
city, city,
force_refresh=force_refresh, force_refresh=force_refresh,
market_slug=market_slug, market_slug=market_slug,
target_date=target_date, target_date=target_date,
resolution=resolution, resolution=resolution,
) timing_recorder=timer,
)
tasks = [ task_by_city = {
_build_with_limit(city) city: asyncio.create_task(_build_with_limit(city))
for city in city_names for city in city_names
] }
results = await asyncio.gather(*tasks, return_exceptions=True) task_city_lookup = {task: city for city, task in task_by_city.items()}
details: Dict[str, Any] = {} done, pending = await timer.measure_async(
errors: Dict[str, str] = {} "build_details",
for city, result in zip(city_names, results): lambda: asyncio.wait(
if isinstance(result, Exception): task_by_city.values(),
errors[city] = str(result) timeout=_city_detail_batch_partial_timeout_seconds(),
continue ),
result_city, payload = result )
details[result_city] = payload 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
return { for task in pending:
"cities": city_names, city = task_city_lookup[task]
"details": details, missing.append(city)
"errors": errors, task.cancel()
}
missing_set = set(missing)
missing = [city for city in city_names if city in missing_set]
partial = bool(missing or errors)
if partial:
outcome = "partial"
return {
"cities": city_names,
"details": details,
"errors": errors,
"missing": missing,
"partial": partial,
}
except HTTPException as exc:
outcome = f"http_{exc.status_code}"
status_code = exc.status_code
raise
except Exception:
outcome = "exception"
status_code = 500
raise
finally:
timer.finish(outcome=outcome, status_code=status_code)
+87
View File
@@ -0,0 +1,87 @@
"""Small helpers for exposing request stage timings via Server-Timing."""
from __future__ import annotations
import re
import threading
import time
from typing import Awaitable, Callable, Dict, Optional, TypeVar
from fastapi import Request, Response
from loguru import logger
T = TypeVar("T")
class ServerTimingRecorder:
def __init__(
self,
request: Optional[Request],
*,
log_name: str,
prefix: str,
state_attr: str,
) -> None:
self.request = request
self.log_name = log_name
self.prefix = prefix
self.state_attr = state_attr
self.started = time.perf_counter()
self.timings_ms: Dict[str, float] = {}
self._lock = threading.Lock()
def _record(self, stage: str, started: float) -> None:
elapsed_ms = round((time.perf_counter() - started) * 1000.0, 1)
with self._lock:
self.timings_ms[stage] = elapsed_ms
def measure(self, stage: str, action: Callable[[], T]) -> T:
started = time.perf_counter()
try:
return action()
finally:
self._record(stage, started)
async def measure_async(self, stage: str, action: Callable[[], Awaitable[T]]) -> T:
started = time.perf_counter()
try:
return await action()
finally:
self._record(stage, started)
def server_timing_value(self) -> str:
with self._lock:
items = list(self.timings_ms.items())
return ", ".join(
f"{self._metric_name(stage)};dur={max(0.0, duration):.1f}"
for stage, duration in items
)
def finish(self, *, outcome: str, status_code: int) -> None:
self._record("total", self.started)
value = self.server_timing_value()
state = getattr(self.request, "state", None)
if state is not None:
setattr(state, self.state_attr, value)
logger.info(
"{} outcome={} status_code={} timings_ms={}",
self.log_name,
outcome,
status_code,
dict(self.timings_ms),
)
def _metric_name(self, stage: str) -> str:
raw = f"{self.prefix}_{stage}"
return re.sub(r"[^A-Za-z0-9_-]", "_", raw)
def attach_server_timing_header(
response: Response,
request: Request,
state_attr: str,
) -> None:
value = str(getattr(request.state, state_attr, "") or "").strip()
if value:
response.headers["Server-Timing"] = value
+56 -21
View File
@@ -2,12 +2,25 @@
from __future__ import annotations from __future__ import annotations
from inspect import Parameter, signature
from typing import Any, Dict from typing import Any, Dict
from fastapi import Request from fastapi import HTTPException, Request
from fastapi.concurrency import run_in_threadpool from fastapi.concurrency import run_in_threadpool
import web.routes as legacy_routes import web.routes as legacy_routes
from web.services.request_timing import ServerTimingRecorder
def _supports_timing_recorder(func: Any) -> bool:
try:
params = signature(func).parameters.values()
except (TypeError, ValueError):
return True
return any(
param.name == "timing_recorder" or param.kind == Parameter.VAR_KEYWORD
for param in params
)
async def get_scan_terminal_payload( async def get_scan_terminal_payload(
@@ -26,27 +39,49 @@ async def get_scan_terminal_payload(
region: str = "", region: str = "",
timezone_offset_seconds: int | None = None, timezone_offset_seconds: int | None = None,
) -> Dict[str, Any]: ) -> Dict[str, Any]:
legacy_routes._assert_entitlement(request) timer = ServerTimingRecorder(
filters: Dict[str, Any] = { request,
"scan_mode": scan_mode, log_name="scan_terminal_timing",
"min_price": min_price, prefix="scan_terminal",
"max_price": max_price, state_attr="scan_terminal_server_timing",
"min_edge_pct": min_edge_pct,
"min_liquidity": min_liquidity,
"high_liquidity_only": high_liquidity_only,
"market_type": market_type,
"time_range": time_range,
"limit": limit,
}
if timezone_offset_seconds is not None:
filters["timezone_offset_seconds"] = timezone_offset_seconds
if region:
filters["trading_region"] = region
return await run_in_threadpool(
legacy_routes.build_scan_terminal_payload,
filters,
force_refresh=force_refresh,
) )
outcome = "ok"
status_code = 200
try:
timer.measure("assert_entitlement", lambda: legacy_routes._assert_entitlement(request))
filters: Dict[str, Any] = {
"scan_mode": scan_mode,
"min_price": min_price,
"max_price": max_price,
"min_edge_pct": min_edge_pct,
"min_liquidity": min_liquidity,
"high_liquidity_only": high_liquidity_only,
"market_type": market_type,
"time_range": time_range,
"limit": limit,
}
if timezone_offset_seconds is not None:
filters["timezone_offset_seconds"] = timezone_offset_seconds
if region:
filters["trading_region"] = region
async def build_payload():
builder = legacy_routes.build_scan_terminal_payload
kwargs: Dict[str, Any] = {"force_refresh": force_refresh}
if _supports_timing_recorder(builder):
kwargs["timing_recorder"] = timer
return await run_in_threadpool(builder, filters, **kwargs)
return await timer.measure_async("build_payload", build_payload)
except HTTPException as exc:
outcome = f"http_{exc.status_code}"
status_code = exc.status_code
raise
except Exception:
outcome = "exception"
status_code = 500
raise
finally:
timer.finish(outcome=outcome, status_code=status_code)
async def get_scan_terminal_overview_payload(request: Request) -> Dict[str, Any]: async def get_scan_terminal_overview_payload(request: Request) -> Dict[str, Any]: