Compare commits
4 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 78ea0326a5 | |||
| d3f444dbf6 | |||
| 8d26afdec0 | |||
| 5083b7c433 |
+2
-2
@@ -79,7 +79,7 @@ services:
|
||||
timeout: 5s
|
||||
image: ghcr.io/yangyuan-zhen/polyweather-frontend:${IMAGE_TAG:-latest}
|
||||
ports:
|
||||
- 3001:3000
|
||||
- "127.0.0.1:3001:3000"
|
||||
restart: unless-stopped
|
||||
polyweather_web:
|
||||
command: python web/app.py
|
||||
@@ -104,7 +104,7 @@ services:
|
||||
timeout: 5s
|
||||
image: ghcr.io/yangyuan-zhen/polyweather-backend:${IMAGE_TAG:-latest}
|
||||
ports:
|
||||
- 8000:8000
|
||||
- "127.0.0.1:8000:8000"
|
||||
restart: unless-stopped
|
||||
user: ${UID:-1000}:${GID:-1000}
|
||||
volumes:
|
||||
|
||||
@@ -7,25 +7,46 @@ import {
|
||||
buildProxyExceptionResponse,
|
||||
buildUpstreamErrorResponse,
|
||||
} from "@/lib/api-proxy";
|
||||
import {
|
||||
createProxyTimer,
|
||||
finishProxyTimedResponse,
|
||||
} from "@/lib/proxy-timing";
|
||||
|
||||
const API_BASE = process.env.POLYWEATHER_API_BASE_URL;
|
||||
const ANALYTICS_ENABLED =
|
||||
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) {
|
||||
const timer = createProxyTimer(req, "analytics_events");
|
||||
if (!ANALYTICS_ENABLED) {
|
||||
return new NextResponse(null, { status: 204 });
|
||||
}
|
||||
|
||||
if (!API_BASE) {
|
||||
return NextResponse.json(
|
||||
{ error: "POLYWEATHER_API_BASE_URL is not configured" },
|
||||
{ status: 500 },
|
||||
return finishProxyTimedResponse(
|
||||
new NextResponse(null, { status: 204 }),
|
||||
timer,
|
||||
"disabled",
|
||||
);
|
||||
}
|
||||
|
||||
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 {
|
||||
const body = await req.json();
|
||||
const body = await timer.measure("request_read", () => req.json());
|
||||
const payload =
|
||||
body && typeof body.payload === "object" && body.payload != null
|
||||
? body.payload
|
||||
@@ -42,30 +63,65 @@ export async function POST(req: NextRequest) {
|
||||
referer_header: req.headers.get("referer") || "",
|
||||
},
|
||||
};
|
||||
const auth = await buildBackendRequestHeaders(req, {
|
||||
includeSupabaseIdentity: false,
|
||||
});
|
||||
auth = await timer.measure("auth_headers", () =>
|
||||
buildBackendRequestHeaders(req, {
|
||||
includeSupabaseIdentity: false,
|
||||
}),
|
||||
);
|
||||
const headers = new Headers(auth.headers);
|
||||
headers.set("Content-Type", "application/json");
|
||||
const res = await fetch(`${API_BASE}/api/analytics/events`, {
|
||||
method: "POST",
|
||||
headers,
|
||||
body: JSON.stringify(enrichedBody),
|
||||
cache: "no-store",
|
||||
});
|
||||
const res = await timer.measure("backend_fetch", () =>
|
||||
fetch(`${API_BASE}/api/analytics/events`, {
|
||||
method: "POST",
|
||||
headers,
|
||||
body: JSON.stringify(enrichedBody),
|
||||
cache: "no-store",
|
||||
signal: controller.signal,
|
||||
}),
|
||||
);
|
||||
const backendServerTiming = res.headers.get("server-timing") || "";
|
||||
if (!res.ok) {
|
||||
const raw = await res.text();
|
||||
const raw = await timer.measure("backend_read", () => res.text());
|
||||
const response = buildUpstreamErrorResponse(res.status, raw, {
|
||||
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);
|
||||
return applyAuthResponseCookies(response, auth.response);
|
||||
return finishProxyTimedResponse(
|
||||
applyAuthResponseCookies(response, auth.response),
|
||||
timer,
|
||||
"ok",
|
||||
{ backendServerTiming },
|
||||
);
|
||||
} catch (error) {
|
||||
return buildProxyExceptionResponse(error, {
|
||||
publicMessage: "Failed to track analytics event",
|
||||
});
|
||||
const timedOut = controller.signal.aborted;
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,6 +1,10 @@
|
||||
import { NextRequest, NextResponse } from "next/server";
|
||||
import { proxyBackendJsonGet } from "@/lib/api-proxy";
|
||||
import { buildCityDetailProxyCachePolicy } from "@/lib/proxy-cache-policy";
|
||||
import {
|
||||
createProxyTimer,
|
||||
finishProxyTimedResponse,
|
||||
} from "@/lib/proxy-timing";
|
||||
|
||||
const API_BASE = process.env.POLYWEATHER_API_BASE_URL;
|
||||
const DETAIL_BATCH_PROXY_TIMEOUT_MS = Number(
|
||||
@@ -8,10 +12,15 @@ const DETAIL_BATCH_PROXY_TIMEOUT_MS = Number(
|
||||
);
|
||||
|
||||
export async function GET(req: NextRequest) {
|
||||
const timer = createProxyTimer(req, "city_detail_batch");
|
||||
if (!API_BASE) {
|
||||
return NextResponse.json(
|
||||
{ error: "POLYWEATHER_API_BASE_URL is not configured" },
|
||||
{ status: 500 },
|
||||
return finishProxyTimedResponse(
|
||||
NextResponse.json(
|
||||
{ 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 {
|
||||
return await proxyBackendJsonGet(req, {
|
||||
cacheControl: cachePolicy.responseCacheControl,
|
||||
fetchCache:
|
||||
cachePolicy.fetchMode === "no-store" ? "no-store" : undefined,
|
||||
cacheControlForData: (data) =>
|
||||
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",
|
||||
revalidateSeconds: cachePolicy.revalidateSeconds,
|
||||
signal: controller.signal,
|
||||
timeoutPublicMessage: "City detail batch request timed out",
|
||||
timing: timer,
|
||||
url: `${API_BASE}/api/cities/detail-batch?${searchParams.toString()}`,
|
||||
});
|
||||
} finally {
|
||||
|
||||
@@ -9,6 +9,10 @@ import {
|
||||
} from "@/lib/api-proxy";
|
||||
import { buildCachedJsonResponse } from "@/lib/http-cache";
|
||||
import { buildCityDetailProxyCachePolicy } from "@/lib/proxy-cache-policy";
|
||||
import {
|
||||
createProxyTimer,
|
||||
finishProxyTimedResponse,
|
||||
} from "@/lib/proxy-timing";
|
||||
|
||||
const API_BASE = process.env.POLYWEATHER_API_BASE_URL;
|
||||
|
||||
@@ -34,15 +38,16 @@ export async function GET(
|
||||
req: NextRequest,
|
||||
context: { params: Promise<{ name: string }> },
|
||||
) {
|
||||
const timer = createProxyTimer(req, "city_detail");
|
||||
if (!API_BASE) {
|
||||
const response = NextResponse.json(
|
||||
{ error: "POLYWEATHER_API_BASE_URL is not configured" },
|
||||
{ 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 cachePolicy = buildCityDetailProxyCachePolicy(forceRefresh, 15);
|
||||
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()}`;
|
||||
|
||||
try {
|
||||
const auth = await buildBackendRequestHeaders(req, {
|
||||
includeSupabaseIdentity: false,
|
||||
});
|
||||
const res = await fetch(url, {
|
||||
headers: auth.headers,
|
||||
...(cachePolicy.fetchMode === "no-store"
|
||||
? { cache: "no-store" as const }
|
||||
: { next: { revalidate: cachePolicy.revalidateSeconds ?? 15 } }),
|
||||
});
|
||||
const auth = await timer.measure("auth_headers", () =>
|
||||
buildBackendRequestHeaders(req, {
|
||||
includeSupabaseIdentity: false,
|
||||
}),
|
||||
);
|
||||
const res = await timer.measure("backend_fetch", () =>
|
||||
fetch(url, {
|
||||
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) {
|
||||
const raw = await res.text();
|
||||
const raw = await timer.measure("backend_read", () => res.text());
|
||||
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(
|
||||
req,
|
||||
data,
|
||||
cachePolicy.responseCacheControl,
|
||||
);
|
||||
return applyAuthResponseCookies(response, auth.response);
|
||||
return finishProxyTimedResponse(
|
||||
applyAuthResponseCookies(response, auth.response),
|
||||
timer,
|
||||
"ok",
|
||||
{ backendServerTiming },
|
||||
);
|
||||
} catch (error) {
|
||||
const response = buildProxyExceptionResponse(error, {
|
||||
publicMessage: "Failed to fetch city detail aggregate",
|
||||
});
|
||||
return response;
|
||||
return finishProxyTimedResponse(response, timer, "exception");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -5,27 +5,45 @@ import {
|
||||
} from "@/lib/backend-auth";
|
||||
import { buildProxyExceptionResponse } from "@/lib/api-proxy";
|
||||
import { requireOpsProxyAuth } from "@/lib/ops-proxy-auth";
|
||||
import {
|
||||
createProxyTimer,
|
||||
finishProxyTimedResponse,
|
||||
} from "@/lib/proxy-timing";
|
||||
|
||||
const API_BASE = process.env.POLYWEATHER_API_BASE_URL;
|
||||
|
||||
export async function GET(req: NextRequest) {
|
||||
const timer = createProxyTimer(req, "ops_online_users");
|
||||
if (!API_BASE) {
|
||||
return NextResponse.json(
|
||||
{ error: "POLYWEATHER_API_BASE_URL is not configured" },
|
||||
{ status: 500 },
|
||||
return finishProxyTimedResponse(
|
||||
NextResponse.json(
|
||||
{ error: "POLYWEATHER_API_BASE_URL is not configured" },
|
||||
{ status: 500 },
|
||||
),
|
||||
timer,
|
||||
"missing_api_base",
|
||||
);
|
||||
}
|
||||
|
||||
try {
|
||||
const auth = await buildBackendRequestHeaders(req);
|
||||
const authError = requireOpsProxyAuth(req, auth);
|
||||
if (authError) return authError;
|
||||
const auth = await timer.measure("auth_headers", () =>
|
||||
buildBackendRequestHeaders(req),
|
||||
);
|
||||
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`, {
|
||||
headers: auth.headers,
|
||||
cache: "no-store",
|
||||
});
|
||||
const raw = await res.text();
|
||||
const res = await timer.measure("backend_fetch", () =>
|
||||
fetch(`${API_BASE}/api/ops/online-users`, {
|
||||
headers: auth.headers,
|
||||
cache: "no-store",
|
||||
}),
|
||||
);
|
||||
const backendServerTiming = res.headers.get("server-timing") || "";
|
||||
const raw = await timer.measure("backend_read", () => res.text());
|
||||
const response = new NextResponse(raw, {
|
||||
status: res.status,
|
||||
headers: {
|
||||
@@ -33,10 +51,19 @@ export async function GET(req: NextRequest) {
|
||||
"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) {
|
||||
return buildProxyExceptionResponse(error, {
|
||||
publicMessage: "Failed to fetch online users",
|
||||
});
|
||||
return finishProxyTimedResponse(
|
||||
buildProxyExceptionResponse(error, {
|
||||
publicMessage: "Failed to fetch online users",
|
||||
}),
|
||||
timer,
|
||||
"exception",
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,6 +2,10 @@ import { NextRequest, NextResponse } from "next/server";
|
||||
import { proxyBackendJsonGet } from "@/lib/api-proxy";
|
||||
import { buildForceRefreshProxyCachePolicy } from "@/lib/proxy-cache-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 SCAN_TERMINAL_PROXY_TIMEOUT_MS = Number(
|
||||
@@ -11,10 +15,15 @@ const SCAN_TERMINAL_PROXY_TIMEOUT_MS = Number(
|
||||
export const maxDuration = 45;
|
||||
|
||||
export async function GET(req: NextRequest) {
|
||||
const timer = createProxyTimer(req, "scan_terminal");
|
||||
if (!API_BASE) {
|
||||
return NextResponse.json(
|
||||
{ error: "POLYWEATHER_API_BASE_URL is not configured" },
|
||||
{ status: 500 },
|
||||
return finishProxyTimedResponse(
|
||||
NextResponse.json(
|
||||
{ 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,
|
||||
signal: controller.signal,
|
||||
timeoutPublicMessage: "Scan terminal request timed out",
|
||||
timing: timer,
|
||||
url,
|
||||
});
|
||||
} finally {
|
||||
|
||||
@@ -55,7 +55,10 @@ import {
|
||||
mergeAccessStateWithAuthPayload,
|
||||
type AuthProfilePayload,
|
||||
} 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 {
|
||||
cityListItemsToScanRows,
|
||||
mergeScanRowsWithCityFallbackRows,
|
||||
@@ -962,7 +965,7 @@ function ScanTerminalScreen() {
|
||||
createEmptyAccess(true),
|
||||
);
|
||||
|
||||
const loadAuthProfile = useCallback(
|
||||
const rawLoadAuthProfile = useCallback(
|
||||
async (
|
||||
accessToken?: string | null,
|
||||
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 supabaseEnabled = hasSupabasePublicEnv();
|
||||
|
||||
@@ -12,6 +12,7 @@ import {
|
||||
import {
|
||||
MAX_HOURLY_DETAIL_CONCURRENT_REQUESTS,
|
||||
HOURLY_CACHE_TTL_MS,
|
||||
__resolveCityDetailFromBatchForTest,
|
||||
__readHourlyCacheEntryForTest,
|
||||
__resetHourlyDetailRequestQueueForTest,
|
||||
__runQueuedHourlyDetailRequestForTest,
|
||||
@@ -95,7 +96,19 @@ 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",
|
||||
);
|
||||
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(
|
||||
fetchHourlyBlock.includes("queueCityDetailBatch(city, resParam)") &&
|
||||
!fetchHourlyBlock.includes("runQueuedHourlyDetailRequest"),
|
||||
@@ -130,6 +143,19 @@ export async function runTests() {
|
||||
__shouldFetchCityDetailForChartForTest({ city: "paris", documentHidden: true, isChartVisible: true }) === false,
|
||||
"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();
|
||||
let activeRequests = 0;
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import {
|
||||
createAuthProfileRequestCache,
|
||||
loadTerminalAuthProfile,
|
||||
type TerminalAuthProfilePayload,
|
||||
} from "@/components/dashboard/scan-terminal/terminal-auth-bootstrap";
|
||||
@@ -27,6 +28,41 @@ async function flushMicrotasks() {
|
||||
}
|
||||
|
||||
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 fastSession = deferred<{ data: { session: { access_token: string } } }>();
|
||||
const calls: string[] = [];
|
||||
|
||||
@@ -966,8 +966,11 @@ type HourlyForecastFetchOptions = {
|
||||
};
|
||||
|
||||
type CityDetailBatchPayload = {
|
||||
cities?: string[];
|
||||
details?: Record<string, CityDetail | null | undefined>;
|
||||
errors?: Record<string, string>;
|
||||
missing?: string[];
|
||||
partial?: boolean;
|
||||
};
|
||||
|
||||
type CityDetailBatchWaiter = {
|
||||
@@ -1073,6 +1076,30 @@ function rejectBatchWaiters(
|
||||
(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) {
|
||||
const queue = _cityDetailBatchQueues.get(resolution);
|
||||
if (!queue) return;
|
||||
@@ -1088,15 +1115,23 @@ async function flushCityDetailBatch(resolution: string) {
|
||||
try {
|
||||
const payload = await fetchCityDetailBatchWithTimeout(cities, resolution);
|
||||
const details = payload?.details || {};
|
||||
const partialMissingCities =
|
||||
payload?.partial === true
|
||||
? new Set((payload.missing || []).map((city) => normalizeCityKey(city)))
|
||||
: new Set<string>();
|
||||
await Promise.all(
|
||||
cities.map(async (city) => {
|
||||
const waiters = queue.waiters.get(city);
|
||||
const detail = details[city];
|
||||
const detail = resolveCityDetailFromBatch(details, city);
|
||||
const data = primeCityDetailCache(city, resolution, detail);
|
||||
if (data) {
|
||||
resolveBatchWaiters(waiters, data);
|
||||
return;
|
||||
}
|
||||
if (partialMissingCities.has(normalizeCityKey(city))) {
|
||||
resolveBatchWaiters(waiters, null);
|
||||
return;
|
||||
}
|
||||
try {
|
||||
resolveBatchWaiters(
|
||||
waiters,
|
||||
@@ -2426,6 +2461,7 @@ export {
|
||||
HOURLY_CACHE_TTL_MS,
|
||||
_hourlyCache,
|
||||
__readHourlyCacheEntryForTest,
|
||||
resolveCityDetailFromBatch as __resolveCityDetailFromBatchForTest,
|
||||
__resetHourlyDetailRequestQueueForTest,
|
||||
__runQueuedHourlyDetailRequestForTest,
|
||||
buildChartDomain,
|
||||
|
||||
@@ -57,6 +57,32 @@ function canResolveProfileImmediately(
|
||||
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({
|
||||
getSession,
|
||||
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
@@ -5,6 +5,10 @@ import {
|
||||
buildBackendRequestHeaders,
|
||||
} from "@/lib/backend-auth";
|
||||
import { buildCachedJsonResponse } from "@/lib/http-cache";
|
||||
import {
|
||||
finishProxyTimedResponse,
|
||||
type ProxyTimer,
|
||||
} from "@/lib/proxy-timing";
|
||||
|
||||
const PASSTHROUGH_UPSTREAM_STATUSES = new Set([
|
||||
400,
|
||||
@@ -81,6 +85,7 @@ export async function proxyBackendJsonGet(
|
||||
req: NextRequest,
|
||||
options: {
|
||||
cacheControl?: string;
|
||||
cacheControlForData?: (data: unknown) => string | undefined;
|
||||
conditionalResponse?: boolean;
|
||||
detailLimit?: number;
|
||||
error?: string;
|
||||
@@ -91,40 +96,75 @@ export async function proxyBackendJsonGet(
|
||||
signal?: AbortSignal;
|
||||
statusOnException?: number;
|
||||
timeoutPublicMessage?: string;
|
||||
timing?: ProxyTimer;
|
||||
url: string;
|
||||
},
|
||||
) {
|
||||
let auth: Awaited<ReturnType<typeof buildBackendRequestHeaders>> | null = null;
|
||||
const timing = options.timing;
|
||||
try {
|
||||
auth = await buildBackendRequestHeaders(req, {
|
||||
includeSupabaseIdentity: options.includeSupabaseIdentity ?? false,
|
||||
});
|
||||
const res = await fetch(options.url, {
|
||||
headers: auth.headers,
|
||||
...(options.fetchCache
|
||||
? { cache: options.fetchCache }
|
||||
: { next: { revalidate: options.revalidateSeconds ?? 30 } }),
|
||||
signal: options.signal,
|
||||
});
|
||||
auth = await (timing
|
||||
? timing.measure("auth_headers", () =>
|
||||
buildBackendRequestHeaders(req, {
|
||||
includeSupabaseIdentity: options.includeSupabaseIdentity ?? false,
|
||||
}),
|
||||
)
|
||||
: buildBackendRequestHeaders(req, {
|
||||
includeSupabaseIdentity: options.includeSupabaseIdentity ?? false,
|
||||
}));
|
||||
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) {
|
||||
const raw = await res.text();
|
||||
const raw = await (timing
|
||||
? timing.measure("backend_read", () => res.text())
|
||||
: res.text());
|
||||
const response = buildUpstreamErrorResponse(res.status, raw, {
|
||||
detailLimit: options.detailLimit,
|
||||
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 =
|
||||
options.cacheControl && options.conditionalResponse !== false
|
||||
? buildCachedJsonResponse(req, data, options.cacheControl)
|
||||
responseCacheControl && options.conditionalResponse !== false
|
||||
? buildCachedJsonResponse(req, data, responseCacheControl)
|
||||
: NextResponse.json(data, {
|
||||
headers: options.cacheControl
|
||||
? { "Cache-Control": options.cacheControl }
|
||||
headers: responseCacheControl
|
||||
? { "Cache-Control": responseCacheControl }
|
||||
: undefined,
|
||||
});
|
||||
return applyAuthResponseCookies(response, auth.response);
|
||||
const withCookies = applyAuthResponseCookies(response, auth.response);
|
||||
return timing
|
||||
? finishProxyTimedResponse(withCookies, timing, "ok", {
|
||||
backendServerTiming,
|
||||
})
|
||||
: withCookies;
|
||||
} catch (error) {
|
||||
const timedOut = options.signal?.aborted === true;
|
||||
const response = buildProxyExceptionResponse(error, {
|
||||
@@ -134,6 +174,9 @@ export async function proxyBackendJsonGet(
|
||||
: options.publicMessage,
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -716,6 +716,7 @@ export interface AiAnalysisStructured {
|
||||
}
|
||||
|
||||
export interface CityDetail {
|
||||
city?: string;
|
||||
name: string;
|
||||
display_name: string;
|
||||
detail_depth?: "panel" | "market" | "nearby" | "full";
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
@@ -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
|
||||
@@ -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
|
||||
|
||||
|
||||
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():
|
||||
source = (ROOT / "web" / "analysis_service.py").read_text(encoding="utf-8")
|
||||
|
||||
|
||||
@@ -618,6 +618,44 @@ def test_city_detail_batch_endpoint_limits_backend_concurrency(monkeypatch):
|
||||
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):
|
||||
import asyncio
|
||||
|
||||
@@ -678,6 +716,70 @@ def test_concurrent_city_detail_requests_share_same_full_cache_refresh(monkeypat
|
||||
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):
|
||||
import asyncio
|
||||
|
||||
|
||||
+10
-3
@@ -2,7 +2,7 @@
|
||||
|
||||
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 (
|
||||
get_city_detail_batch_payload,
|
||||
@@ -12,6 +12,7 @@ from web.services.city_api import (
|
||||
list_cities_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"])
|
||||
|
||||
@@ -108,6 +109,7 @@ async def cities_model_range(
|
||||
@router.get("/api/cities/detail-batch")
|
||||
async def city_detail_batch(
|
||||
request: Request,
|
||||
response: Response,
|
||||
cities: str = "",
|
||||
force_refresh: bool = False,
|
||||
market_slug: Optional[str] = None,
|
||||
@@ -115,7 +117,7 @@ async def city_detail_batch(
|
||||
resolution: Optional[str] = "10m",
|
||||
limit: int = 12,
|
||||
):
|
||||
return await get_city_detail_batch_payload(
|
||||
payload = await get_city_detail_batch_payload(
|
||||
request,
|
||||
cities=cities,
|
||||
force_refresh=force_refresh,
|
||||
@@ -124,6 +126,8 @@ async def city_detail_batch(
|
||||
resolution=resolution,
|
||||
limit=limit,
|
||||
)
|
||||
attach_server_timing_header(response, request, "city_detail_batch_server_timing")
|
||||
return payload
|
||||
|
||||
|
||||
@router.get("/api/city/{name}")
|
||||
@@ -159,13 +163,14 @@ async def city_summary(
|
||||
@router.get("/api/city/{name}/detail")
|
||||
async def city_detail_aggregate(
|
||||
request: Request,
|
||||
response: Response,
|
||||
name: str,
|
||||
force_refresh: bool = False,
|
||||
market_slug: Optional[str] = None,
|
||||
target_date: Optional[str] = None,
|
||||
resolution: Optional[str] = "10m",
|
||||
):
|
||||
return await get_city_detail_aggregate_payload(
|
||||
payload = await get_city_detail_aggregate_payload(
|
||||
request,
|
||||
name,
|
||||
force_refresh=force_refresh,
|
||||
@@ -173,6 +178,8 @@ async def city_detail_aggregate(
|
||||
target_date=target_date,
|
||||
resolution=resolution,
|
||||
)
|
||||
attach_server_timing_header(response, request, "city_detail_server_timing")
|
||||
return payload
|
||||
|
||||
|
||||
@router.get("/api/city/{name}/realtime-stream")
|
||||
|
||||
+25
-4
@@ -1,6 +1,6 @@
|
||||
"""Operations/admin API routes."""
|
||||
|
||||
from fastapi import APIRouter, Request
|
||||
from fastapi import APIRouter, Request, Response
|
||||
|
||||
from web.core import GrantPointsRequest
|
||||
from web.services.ops_api import (
|
||||
@@ -30,14 +30,35 @@ from web.services.ops_api import (
|
||||
get_ops_training_accuracy,
|
||||
get_ops_telegram_audit,
|
||||
)
|
||||
from web.services.request_timing import ServerTimingRecorder, attach_server_timing_header
|
||||
|
||||
router = APIRouter(tags=["ops"])
|
||||
|
||||
|
||||
@router.get("/api/ops/online-users")
|
||||
async def ops_online_users():
|
||||
from src.utils.online_tracker import online_count
|
||||
return {"online": online_count()}
|
||||
async def ops_online_users(request: Request, response: Response):
|
||||
timer = ServerTimingRecorder(
|
||||
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")
|
||||
|
||||
+4
-1
@@ -9,6 +9,7 @@ from web.services.scan_api import (
|
||||
get_scan_terminal_overview_payload,
|
||||
get_scan_terminal_payload,
|
||||
)
|
||||
from web.services.request_timing import attach_server_timing_header
|
||||
|
||||
router = APIRouter(tags=["scan"])
|
||||
|
||||
@@ -45,12 +46,14 @@ async def scan_terminal(
|
||||
region=region or trading_region or None,
|
||||
timezone_offset_seconds=timezone_offset_seconds,
|
||||
)
|
||||
return JSONResponse(
|
||||
response = JSONResponse(
|
||||
content=payload,
|
||||
headers={
|
||||
"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")
|
||||
|
||||
@@ -235,16 +235,41 @@ def build_scan_terminal_payload(
|
||||
raw_filters: Optional[Dict[str, Any]] = None,
|
||||
*,
|
||||
force_refresh: bool = False,
|
||||
timing_recorder: Any = None,
|
||||
) -> 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:
|
||||
cached = get_cached_scan_terminal_payload(
|
||||
filters, ttl_sec=SCAN_TERMINAL_PAYLOAD_TTL_SEC
|
||||
cached = (
|
||||
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:
|
||||
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")
|
||||
if isinstance(success_payload, dict) and success_payload:
|
||||
started = _start_scan_terminal_background_refresh(filters)
|
||||
@@ -257,7 +282,17 @@ def build_scan_terminal_payload(
|
||||
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
|
||||
|
||||
+197
-52
@@ -13,6 +13,7 @@ from fastapi.concurrency import run_in_threadpool
|
||||
from loguru import logger
|
||||
|
||||
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_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"),
|
||||
)
|
||||
_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]
|
||||
_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:
|
||||
task = _CITY_FULL_REFRESH_INFLIGHT.get(key)
|
||||
if task is None:
|
||||
task = asyncio.create_task(
|
||||
run_in_threadpool(legacy_routes._refresh_city_full_cache, city, force_refresh),
|
||||
)
|
||||
async def _run_refresh() -> Dict[str, Any]:
|
||||
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
|
||||
try:
|
||||
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)
|
||||
|
||||
|
||||
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]:
|
||||
if force_refresh:
|
||||
return await _refresh_city_full_data(city, 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 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 _overlay_cached_wunderground(city, cached_entry.get("payload") or {})
|
||||
return await _overlay_cached_wunderground(city, payload)
|
||||
return await _refresh_city_full_data(city, False)
|
||||
|
||||
|
||||
@@ -368,16 +404,41 @@ async def get_city_detail_aggregate_payload(
|
||||
target_date: Optional[str] = None,
|
||||
resolution: Optional[str] = "10m",
|
||||
) -> Dict[str, Any]:
|
||||
legacy_routes._assert_entitlement(request)
|
||||
city = legacy_routes._normalize_city_or_404(name)
|
||||
data = await _get_city_full_data(city, force_refresh=force_refresh)
|
||||
|
||||
return await _build_city_detail_payload_cached(
|
||||
data,
|
||||
market_slug,
|
||||
target_date,
|
||||
resolution,
|
||||
timer = ServerTimingRecorder(
|
||||
request,
|
||||
log_name="city_detail_timing",
|
||||
prefix="city_detail",
|
||||
state_attr="city_detail_server_timing",
|
||||
)
|
||||
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]:
|
||||
@@ -404,14 +465,30 @@ async def _build_city_detail_batch_item_async(
|
||||
market_slug: Optional[str],
|
||||
target_date: Optional[str],
|
||||
resolution: Optional[str],
|
||||
timing_recorder: Optional[ServerTimingRecorder] = None,
|
||||
) -> Tuple[str, Dict[str, Any]]:
|
||||
data = await _get_city_full_data(city, force_refresh=force_refresh)
|
||||
detail = await _build_city_detail_payload_cached(
|
||||
data,
|
||||
market_slug,
|
||||
target_date,
|
||||
resolution,
|
||||
)
|
||||
if timing_recorder is not None:
|
||||
data = await timing_recorder.measure_async(
|
||||
f"full_data_{city}",
|
||||
lambda: _get_city_full_data(city, force_refresh=force_refresh),
|
||||
)
|
||||
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
|
||||
|
||||
|
||||
@@ -423,6 +500,19 @@ def _city_detail_batch_concurrency() -> int:
|
||||
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(
|
||||
request: Request,
|
||||
*,
|
||||
@@ -433,39 +523,94 @@ async def get_city_detail_batch_payload(
|
||||
resolution: Optional[str] = "10m",
|
||||
limit: int = 12,
|
||||
) -> Dict[str, Any]:
|
||||
legacy_routes._assert_entitlement(request)
|
||||
city_names = _parse_batch_city_names(cities, limit=max(1, min(24, int(limit or 12))))
|
||||
if not city_names:
|
||||
return {"cities": [], "details": {}, "errors": {}}
|
||||
timer = ServerTimingRecorder(
|
||||
request,
|
||||
log_name="city_detail_batch_timing",
|
||||
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 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,
|
||||
)
|
||||
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,
|
||||
)
|
||||
|
||||
tasks = [
|
||||
_build_with_limit(city)
|
||||
for city in city_names
|
||||
]
|
||||
results = await asyncio.gather(*tasks, return_exceptions=True)
|
||||
details: Dict[str, Any] = {}
|
||||
errors: Dict[str, str] = {}
|
||||
for city, result in zip(city_names, results):
|
||||
if isinstance(result, Exception):
|
||||
errors[city] = str(result)
|
||||
continue
|
||||
result_city, payload = result
|
||||
details[result_city] = payload
|
||||
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
|
||||
|
||||
return {
|
||||
"cities": city_names,
|
||||
"details": details,
|
||||
"errors": errors,
|
||||
}
|
||||
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]
|
||||
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)
|
||||
|
||||
@@ -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
@@ -2,12 +2,25 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from inspect import Parameter, signature
|
||||
from typing import Any, Dict
|
||||
|
||||
from fastapi import Request
|
||||
from fastapi import HTTPException, Request
|
||||
from fastapi.concurrency import run_in_threadpool
|
||||
|
||||
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(
|
||||
@@ -26,27 +39,49 @@ async def get_scan_terminal_payload(
|
||||
region: str = "",
|
||||
timezone_offset_seconds: int | None = None,
|
||||
) -> Dict[str, Any]:
|
||||
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
|
||||
return await run_in_threadpool(
|
||||
legacy_routes.build_scan_terminal_payload,
|
||||
filters,
|
||||
force_refresh=force_refresh,
|
||||
timer = ServerTimingRecorder(
|
||||
request,
|
||||
log_name="scan_terminal_timing",
|
||||
prefix="scan_terminal",
|
||||
state_attr="scan_terminal_server_timing",
|
||||
)
|
||||
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]:
|
||||
|
||||
Reference in New Issue
Block a user