Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 372d4366a8 | |||
| 668f4d9bd3 |
@@ -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(
|
||||||
|
NextResponse.json(
|
||||||
{ error: "POLYWEATHER_API_BASE_URL is not configured" },
|
{ error: "POLYWEATHER_API_BASE_URL is not configured" },
|
||||||
{ status: 500 },
|
{ status: 500 },
|
||||||
|
),
|
||||||
|
timer,
|
||||||
|
"missing_api_base",
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -39,6 +48,7 @@ export async function GET(req: NextRequest) {
|
|||||||
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 {
|
||||||
|
|||||||
@@ -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", () =>
|
||||||
|
buildBackendRequestHeaders(req, {
|
||||||
includeSupabaseIdentity: false,
|
includeSupabaseIdentity: false,
|
||||||
});
|
}),
|
||||||
const res = await fetch(url, {
|
);
|
||||||
|
const res = await timer.measure("backend_fetch", () =>
|
||||||
|
fetch(url, {
|
||||||
headers: auth.headers,
|
headers: auth.headers,
|
||||||
...(cachePolicy.fetchMode === "no-store"
|
...(cachePolicy.fetchMode === "no-store"
|
||||||
? { cache: "no-store" as const }
|
? { cache: "no-store" as const }
|
||||||
: { next: { revalidate: cachePolicy.revalidateSeconds ?? 15 } }),
|
: { 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");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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(
|
||||||
|
NextResponse.json(
|
||||||
{ error: "POLYWEATHER_API_BASE_URL is not configured" },
|
{ error: "POLYWEATHER_API_BASE_URL is not configured" },
|
||||||
{ status: 500 },
|
{ 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", () =>
|
||||||
|
fetch(`${API_BASE}/api/ops/online-users`, {
|
||||||
headers: auth.headers,
|
headers: auth.headers,
|
||||||
cache: "no-store",
|
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(
|
||||||
|
buildProxyExceptionResponse(error, {
|
||||||
publicMessage: "Failed to fetch online users",
|
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 { 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(
|
||||||
|
NextResponse.json(
|
||||||
{ error: "POLYWEATHER_API_BASE_URL is not configured" },
|
{ error: "POLYWEATHER_API_BASE_URL is not configured" },
|
||||||
{ status: 500 },
|
{ 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 {
|
||||||
|
|||||||
@@ -77,6 +77,8 @@ const TrainingDashboard = dynamic(
|
|||||||
},
|
},
|
||||||
);
|
);
|
||||||
|
|
||||||
|
const ONLINE_USERS_REFRESH_MS = 5 * 60_000;
|
||||||
|
|
||||||
function createEmptyAccess(loading = true): ProAccessState {
|
function createEmptyAccess(loading = true): ProAccessState {
|
||||||
return {
|
return {
|
||||||
loading,
|
loading,
|
||||||
@@ -425,14 +427,22 @@ function PolyWeatherTerminal({
|
|||||||
|
|
||||||
useEffect(() => {
|
useEffect(() => {
|
||||||
const fetchOnline = () => {
|
const fetchOnline = () => {
|
||||||
|
if (typeof document !== "undefined" && document.visibilityState === "hidden") return;
|
||||||
fetch("/api/ops/online-users", { headers: { Accept: "application/json" } })
|
fetch("/api/ops/online-users", { headers: { Accept: "application/json" } })
|
||||||
.then((r) => (r.ok ? r.json() : null))
|
.then((r) => (r.ok ? r.json() : null))
|
||||||
.then((d) => { if (d?.online != null) setOnlineCount(d.online); })
|
.then((d) => { if (d?.online != null) setOnlineCount(d.online); })
|
||||||
.catch(() => {});
|
.catch(() => {});
|
||||||
};
|
};
|
||||||
fetchOnline();
|
fetchOnline();
|
||||||
const id = setInterval(fetchOnline, 60_000);
|
const handleVisibilityChange = () => {
|
||||||
return () => clearInterval(id);
|
if (document.visibilityState === "visible") fetchOnline();
|
||||||
|
};
|
||||||
|
document.addEventListener("visibilitychange", handleVisibilityChange);
|
||||||
|
const id = setInterval(fetchOnline, ONLINE_USERS_REFRESH_MS);
|
||||||
|
return () => {
|
||||||
|
document.removeEventListener("visibilitychange", handleVisibilityChange);
|
||||||
|
clearInterval(id);
|
||||||
|
};
|
||||||
}, []);
|
}, []);
|
||||||
|
|
||||||
const [gridCols, setGridCols] = useState<number>(() => {
|
const [gridCols, setGridCols] = useState<number>(() => {
|
||||||
|
|||||||
@@ -101,6 +101,10 @@ function formatCityLocalDate(tzOffsetSeconds: number | null | undefined) {
|
|||||||
return `${y}-${m}-${d}`;
|
return `${y}-${m}-${d}`;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function getLiveTempFromHourly(data: HourlyForecast) {
|
||||||
|
return validNumber(data?.airportCurrent?.temp) ?? validNumber(data?.airportPrimary?.temp) ?? null;
|
||||||
|
}
|
||||||
|
|
||||||
function getWundergroundDailyHigh(hourly: HourlyForecast) {
|
function getWundergroundDailyHigh(hourly: HourlyForecast) {
|
||||||
return validNumber(hourly?.wundergroundCurrent?.max_so_far) ?? null;
|
return validNumber(hourly?.wundergroundCurrent?.max_so_far) ?? null;
|
||||||
}
|
}
|
||||||
@@ -365,6 +369,8 @@ export function LiveTemperatureThresholdChart({
|
|||||||
.then((data) => {
|
.then((data) => {
|
||||||
if (cancelled || !data) return;
|
if (cancelled || !data) return;
|
||||||
hasLoadedHourlyDetailRef.current = true;
|
hasLoadedHourlyDetailRef.current = true;
|
||||||
|
const temp = getLiveTempFromHourly(data);
|
||||||
|
if (temp !== null) setLiveTemp(temp);
|
||||||
setHourly(data);
|
setHourly(data);
|
||||||
})
|
})
|
||||||
.catch(() => {})
|
.catch(() => {})
|
||||||
@@ -377,15 +383,6 @@ export function LiveTemperatureThresholdChart({
|
|||||||
if (typeof document !== "undefined" && document.visibilityState === "hidden") return;
|
if (typeof document !== "undefined" && document.visibilityState === "hidden") return;
|
||||||
if (Date.now() - lastPatchAtRef.current < 2 * 60_000) return;
|
if (Date.now() - lastPatchAtRef.current < 2 * 60_000) return;
|
||||||
|
|
||||||
fetch(`/api/city/${encodeURIComponent(city)}/summary`)
|
|
||||||
.then((res) => (res.ok ? res.json() : null))
|
|
||||||
.then((payload) => {
|
|
||||||
if (cancelled || !payload) return;
|
|
||||||
const temp = validNumber(payload?.current?.temp);
|
|
||||||
if (temp !== null) setLiveTemp(temp);
|
|
||||||
})
|
|
||||||
.catch(() => {});
|
|
||||||
|
|
||||||
refreshFullDetail();
|
refreshFullDetail();
|
||||||
};
|
};
|
||||||
|
|
||||||
@@ -403,19 +400,12 @@ export function LiveTemperatureThresholdChart({
|
|||||||
const refreshForegroundFullDetail = () => {
|
const refreshForegroundFullDetail = () => {
|
||||||
lastPatchAtRef.current = Date.now();
|
lastPatchAtRef.current = Date.now();
|
||||||
|
|
||||||
fetch(`/api/city/${encodeURIComponent(city)}/summary`)
|
|
||||||
.then((res) => (res.ok ? res.json() : null))
|
|
||||||
.then((payload) => {
|
|
||||||
if (cancelled || !payload) return;
|
|
||||||
const temp = validNumber(payload?.current?.temp);
|
|
||||||
if (temp !== null) setLiveTemp(temp);
|
|
||||||
})
|
|
||||||
.catch(() => {});
|
|
||||||
|
|
||||||
fetchHourlyForecastForCity(city, { ignoreCache: true, resolution: targetResolution })
|
fetchHourlyForecastForCity(city, { ignoreCache: true, resolution: targetResolution })
|
||||||
.then((data) => {
|
.then((data) => {
|
||||||
if (cancelled || !data) return;
|
if (cancelled || !data) return;
|
||||||
hasLoadedHourlyDetailRef.current = true;
|
hasLoadedHourlyDetailRef.current = true;
|
||||||
|
const temp = getLiveTempFromHourly(data);
|
||||||
|
if (temp !== null) setLiveTemp(temp);
|
||||||
setHourly(data);
|
setHourly(data);
|
||||||
})
|
})
|
||||||
.catch(() => {});
|
.catch(() => {});
|
||||||
|
|||||||
@@ -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,
|
||||||
@@ -49,6 +50,10 @@ export async function runTests() {
|
|||||||
path.join(projectRoot, "components", "dashboard", "scan-terminal", "temperature-chart-logic.ts"),
|
path.join(projectRoot, "components", "dashboard", "scan-terminal", "temperature-chart-logic.ts"),
|
||||||
"utf8",
|
"utf8",
|
||||||
);
|
);
|
||||||
|
const dashboardSource = fs.readFileSync(
|
||||||
|
path.join(projectRoot, "components", "dashboard", "ScanTerminalDashboard.tsx"),
|
||||||
|
"utf8",
|
||||||
|
);
|
||||||
|
|
||||||
assert(
|
assert(
|
||||||
querySource.includes("useSsePatchVersion") &&
|
querySource.includes("useSsePatchVersion") &&
|
||||||
@@ -91,10 +96,16 @@ 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]*?\r?\n}\r?\n\r?\nfunction fetchCityDetailWithTimeout/)?.[0] || "";
|
||||||
assert(
|
assert(
|
||||||
chartLogicSource.includes("options.ignoreCache\n ? runQueuedHourlyDetailRequest") &&
|
fetchHourlyBlock.includes("queueCityDetailBatch(city, resParam)") &&
|
||||||
chartLogicSource.includes(": queueCityDetailBatch(city, resParam)"),
|
!fetchHourlyBlock.includes("runQueuedHourlyDetailRequest"),
|
||||||
"normal first-paint city detail requests should enter the batch queue before single-request concurrency limiting",
|
"first-paint and background city detail refreshes should both enter the batch queue before falling back to single requests",
|
||||||
|
);
|
||||||
|
assert(
|
||||||
|
dashboardSource.includes("ONLINE_USERS_REFRESH_MS = 5 * 60_000") &&
|
||||||
|
dashboardSource.includes('document.visibilityState === "hidden"'),
|
||||||
|
"terminal online-user presence should refresh slowly and pause while the browser tab is hidden",
|
||||||
);
|
);
|
||||||
assert(
|
assert(
|
||||||
chartSource.includes("IntersectionObserver") &&
|
chartSource.includes("IntersectionObserver") &&
|
||||||
@@ -120,6 +131,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;
|
||||||
|
|||||||
@@ -182,6 +182,10 @@ export function runTests() {
|
|||||||
!foregroundRefreshBlock.includes("setIsHourlyLoading(true)"),
|
!foregroundRefreshBlock.includes("setIsHourlyLoading(true)"),
|
||||||
"foreground resume refresh should update full detail immediately in the background without showing the loading overlay",
|
"foreground resume refresh should update full detail immediately in the background without showing the loading overlay",
|
||||||
);
|
);
|
||||||
|
assert(
|
||||||
|
!chart.includes("/api/city/${encodeURIComponent(city)}/summary"),
|
||||||
|
"visible chart fallback and foreground refresh should not issue a separate summary request after requesting full detail",
|
||||||
|
);
|
||||||
assert(chart.includes("viewMode"), "temperature chart must expose a view mode for DEB-peak auto view versus full-day view");
|
assert(chart.includes("viewMode"), "temperature chart must expose a view mode for DEB-peak auto view versus full-day view");
|
||||||
assert(chart.includes('useState<"auto" | "full">("full")'), "temperature chart must default every city panel to the all-day view");
|
assert(chart.includes('useState<"auto" | "full">("full")'), "temperature chart must default every city panel to the all-day view");
|
||||||
assert(
|
assert(
|
||||||
|
|||||||
@@ -1073,6 +1073,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;
|
||||||
@@ -1091,14 +1115,17 @@ async function flushCityDetailBatch(resolution: 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;
|
||||||
}
|
}
|
||||||
try {
|
try {
|
||||||
resolveBatchWaiters(waiters, await fetchSingleHourlyForecastForCity(city, resolution));
|
resolveBatchWaiters(
|
||||||
|
waiters,
|
||||||
|
await runQueuedHourlyDetailRequest(() => fetchSingleHourlyForecastForCity(city, resolution)),
|
||||||
|
);
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
rejectBatchWaiters(waiters, error);
|
rejectBatchWaiters(waiters, error);
|
||||||
}
|
}
|
||||||
@@ -1109,7 +1136,10 @@ async function flushCityDetailBatch(resolution: string) {
|
|||||||
cities.map(async (city) => {
|
cities.map(async (city) => {
|
||||||
const waiters = queue.waiters.get(city);
|
const waiters = queue.waiters.get(city);
|
||||||
try {
|
try {
|
||||||
resolveBatchWaiters(waiters, await fetchSingleHourlyForecastForCity(city, resolution));
|
resolveBatchWaiters(
|
||||||
|
waiters,
|
||||||
|
await runQueuedHourlyDetailRequest(() => fetchSingleHourlyForecastForCity(city, resolution)),
|
||||||
|
);
|
||||||
} catch (fallbackError) {
|
} catch (fallbackError) {
|
||||||
rejectBatchWaiters(waiters, fallbackError || error);
|
rejectBatchWaiters(waiters, fallbackError || error);
|
||||||
}
|
}
|
||||||
@@ -1158,11 +1188,7 @@ async function fetchHourlyForecastForCity(
|
|||||||
const pending = _hourlyRequestCache.get(requestKey);
|
const pending = _hourlyRequestCache.get(requestKey);
|
||||||
if (pending) return pending;
|
if (pending) return pending;
|
||||||
|
|
||||||
const request = (
|
const request = queueCityDetailBatch(city, resParam)
|
||||||
options.ignoreCache
|
|
||||||
? runQueuedHourlyDetailRequest(() => fetchSingleHourlyForecastForCity(city, resParam))
|
|
||||||
: queueCityDetailBatch(city, resParam)
|
|
||||||
)
|
|
||||||
.finally(() => {
|
.finally(() => {
|
||||||
_hourlyRequestCache.delete(requestKey);
|
_hourlyRequestCache.delete(requestKey);
|
||||||
});
|
});
|
||||||
@@ -2424,6 +2450,7 @@ export {
|
|||||||
HOURLY_CACHE_TTL_MS,
|
HOURLY_CACHE_TTL_MS,
|
||||||
_hourlyCache,
|
_hourlyCache,
|
||||||
__readHourlyCacheEntryForTest,
|
__readHourlyCacheEntryForTest,
|
||||||
|
resolveCityDetailFromBatch as __resolveCityDetailFromBatchForTest,
|
||||||
__resetHourlyDetailRequestQueueForTest,
|
__resetHourlyDetailRequestQueueForTest,
|
||||||
__runQueuedHourlyDetailRequestForTest,
|
__runQueuedHourlyDetailRequestForTest,
|
||||||
buildChartDomain,
|
buildChartDomain,
|
||||||
|
|||||||
@@ -0,0 +1,60 @@
|
|||||||
|
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/);
|
||||||
|
|
||||||
|
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));
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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,
|
||||||
@@ -91,31 +95,59 @@ 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
|
||||||
|
? timing.measure("auth_headers", () =>
|
||||||
|
buildBackendRequestHeaders(req, {
|
||||||
includeSupabaseIdentity: options.includeSupabaseIdentity ?? false,
|
includeSupabaseIdentity: options.includeSupabaseIdentity ?? false,
|
||||||
});
|
}),
|
||||||
const res = await fetch(options.url, {
|
)
|
||||||
|
: 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,
|
headers: auth.headers,
|
||||||
...(options.fetchCache
|
...(options.fetchCache
|
||||||
? { cache: options.fetchCache }
|
? { cache: options.fetchCache }
|
||||||
: { next: { revalidate: options.revalidateSeconds ?? 30 } }),
|
: { next: { revalidate: options.revalidateSeconds ?? 30 } }),
|
||||||
signal: options.signal,
|
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 response =
|
const response =
|
||||||
options.cacheControl && options.conditionalResponse !== false
|
options.cacheControl && options.conditionalResponse !== false
|
||||||
? buildCachedJsonResponse(req, data, options.cacheControl)
|
? buildCachedJsonResponse(req, data, options.cacheControl)
|
||||||
@@ -124,7 +156,12 @@ export async function proxyBackendJsonGet(
|
|||||||
? { "Cache-Control": options.cacheControl }
|
? { "Cache-Control": options.cacheControl }
|
||||||
: 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 +171,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;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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";
|
||||||
|
|||||||
@@ -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;
|
||||||
|
}
|
||||||
@@ -946,6 +946,9 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
|
|||||||
for key in list(self._wunderground_historical_cache.keys()):
|
for key in list(self._wunderground_historical_cache.keys()):
|
||||||
if key.startswith(prefix):
|
if key.startswith(prefix):
|
||||||
self._wunderground_historical_cache.pop(key, None)
|
self._wunderground_historical_cache.pop(key, None)
|
||||||
|
for key in list(getattr(self, "_wunderground_negative_cache", {}).keys()):
|
||||||
|
if key.startswith(prefix):
|
||||||
|
self._wunderground_negative_cache.pop(key, None)
|
||||||
|
|
||||||
def _uses_fahrenheit(self, city_lower: str) -> bool:
|
def _uses_fahrenheit(self, city_lower: str) -> bool:
|
||||||
return city_lower in self.US_CITIES
|
return city_lower in self.US_CITIES
|
||||||
|
|||||||
@@ -73,6 +73,13 @@ class WundergroundHistoricalMixin:
|
|||||||
self._wunderground_historical_cache = {}
|
self._wunderground_historical_cache = {}
|
||||||
if not hasattr(self, "_wunderground_historical_cache_lock"):
|
if not hasattr(self, "_wunderground_historical_cache_lock"):
|
||||||
self._wunderground_historical_cache_lock = threading.Lock()
|
self._wunderground_historical_cache_lock = threading.Lock()
|
||||||
|
if not hasattr(self, "_wunderground_negative_cache"):
|
||||||
|
self._wunderground_negative_cache = {}
|
||||||
|
if not hasattr(self, "wunderground_negative_cache_ttl_sec"):
|
||||||
|
self.wunderground_negative_cache_ttl_sec = max(
|
||||||
|
60,
|
||||||
|
int(os.getenv("WUNDERGROUND_NEGATIVE_CACHE_TTL_SEC", "900")),
|
||||||
|
)
|
||||||
if not hasattr(self, "wunderground_historical_cache_ttl_sec"):
|
if not hasattr(self, "wunderground_historical_cache_ttl_sec"):
|
||||||
self.wunderground_historical_cache_ttl_sec = max(
|
self.wunderground_historical_cache_ttl_sec = max(
|
||||||
30,
|
30,
|
||||||
@@ -239,6 +246,15 @@ class WundergroundHistoricalMixin:
|
|||||||
cached = self._wunderground_historical_cache.get(cache_key)
|
cached = self._wunderground_historical_cache.get(cache_key)
|
||||||
if cached and now_ts - float(cached.get("t", 0)) < self.wunderground_historical_cache_ttl_sec:
|
if cached and now_ts - float(cached.get("t", 0)) < self.wunderground_historical_cache_ttl_sec:
|
||||||
return dict(cached["d"])
|
return dict(cached["d"])
|
||||||
|
negative_cached = self._wunderground_negative_cache.get(cache_key)
|
||||||
|
if negative_cached and now_ts - float(negative_cached.get("t", 0)) < self.wunderground_negative_cache_ttl_sec:
|
||||||
|
record_source_call(
|
||||||
|
"wunderground",
|
||||||
|
"historical",
|
||||||
|
"negative_cached",
|
||||||
|
(time.perf_counter() - started) * 1000.0,
|
||||||
|
)
|
||||||
|
return None
|
||||||
|
|
||||||
current_url = (
|
current_url = (
|
||||||
"https://api.weather.com/v1/location/"
|
"https://api.weather.com/v1/location/"
|
||||||
@@ -286,6 +302,16 @@ class WundergroundHistoricalMixin:
|
|||||||
location_id,
|
location_id,
|
||||||
exc,
|
exc,
|
||||||
)
|
)
|
||||||
|
if (
|
||||||
|
isinstance(exc, httpx.HTTPStatusError)
|
||||||
|
and exc.response is not None
|
||||||
|
and 400 <= exc.response.status_code < 500
|
||||||
|
):
|
||||||
|
with self._wunderground_historical_cache_lock:
|
||||||
|
self._wunderground_negative_cache[cache_key] = {
|
||||||
|
"t": time.time(),
|
||||||
|
"status": exc.response.status_code,
|
||||||
|
}
|
||||||
record_source_call(
|
record_source_call(
|
||||||
"wunderground",
|
"wunderground",
|
||||||
"historical",
|
"historical",
|
||||||
@@ -396,6 +422,7 @@ class WundergroundHistoricalMixin:
|
|||||||
"t": time.time(),
|
"t": time.time(),
|
||||||
"d": payload,
|
"d": payload,
|
||||||
}
|
}
|
||||||
|
self._wunderground_negative_cache.pop(cache_key, None)
|
||||||
record_source_call(
|
record_source_call(
|
||||||
"wunderground",
|
"wunderground",
|
||||||
"historical",
|
"historical",
|
||||||
|
|||||||
@@ -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
|
||||||
@@ -564,7 +564,183 @@ def test_city_detail_batch_endpoint_builds_multiple_cached_details(monkeypatch):
|
|||||||
assert sorted(payload["details"]) == ["paris", "shanghai"]
|
assert sorted(payload["details"]) == ["paris", "shanghai"]
|
||||||
assert payload["details"]["shanghai"]["resolution"] == "10m"
|
assert payload["details"]["shanghai"]["resolution"] == "10m"
|
||||||
assert payload["details"]["paris"]["overlay_city"] == "paris"
|
assert payload["details"]["paris"]["overlay_city"] == "paris"
|
||||||
assert calls == [("shanghai", "10m"), ("paris", "10m")]
|
assert sorted(calls) == [("paris", "10m"), ("shanghai", "10m")]
|
||||||
|
|
||||||
|
|
||||||
|
def test_city_detail_batch_endpoint_limits_backend_concurrency(monkeypatch):
|
||||||
|
import asyncio
|
||||||
|
|
||||||
|
active = 0
|
||||||
|
max_active = 0
|
||||||
|
|
||||||
|
async def fake_run_in_threadpool(fn, *args, **kwargs):
|
||||||
|
nonlocal active, max_active
|
||||||
|
active += 1
|
||||||
|
max_active = max(max_active, active)
|
||||||
|
try:
|
||||||
|
await asyncio.sleep(0.01)
|
||||||
|
return fn(*args, **kwargs)
|
||||||
|
finally:
|
||||||
|
active -= 1
|
||||||
|
|
||||||
|
monkeypatch.setenv("POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY", "2")
|
||||||
|
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, "_city_cache_is_fresh", lambda entry, ttl: False)
|
||||||
|
|
||||||
|
class FakeCache:
|
||||||
|
def get_city_cache(self, kind, city):
|
||||||
|
assert kind == "full"
|
||||||
|
return None
|
||||||
|
|
||||||
|
def refresh_full(city, force_refresh):
|
||||||
|
return {
|
||||||
|
"city": city,
|
||||||
|
"hourly": {"times": ["2026-05-30T00:00:00Z"], "temps": [20.0]},
|
||||||
|
}
|
||||||
|
|
||||||
|
def build_detail(data, market_slug, target_date, resolution):
|
||||||
|
return {
|
||||||
|
"city": data["city"],
|
||||||
|
"hourly": data["hourly"],
|
||||||
|
"resolution": resolution,
|
||||||
|
}
|
||||||
|
|
||||||
|
monkeypatch.setattr(city_api.legacy_routes, "_CACHE_DB", FakeCache())
|
||||||
|
monkeypatch.setattr(city_api.legacy_routes, "_refresh_city_full_cache", refresh_full)
|
||||||
|
monkeypatch.setattr(city_api.legacy_routes, "_build_city_detail_payload", build_detail)
|
||||||
|
|
||||||
|
response = client.get("/api/cities/detail-batch?cities=a,b,c,d,e&resolution=10m&limit=5")
|
||||||
|
|
||||||
|
assert response.status_code == 200
|
||||||
|
assert response.json()["cities"] == ["a", "b", "c", "d", "e"]
|
||||||
|
assert max_active <= 2
|
||||||
|
|
||||||
|
|
||||||
|
def test_concurrent_city_detail_requests_share_same_full_cache_refresh(monkeypatch):
|
||||||
|
import asyncio
|
||||||
|
|
||||||
|
refresh_calls = 0
|
||||||
|
build_calls = 0
|
||||||
|
|
||||||
|
class FakeCache:
|
||||||
|
payload = None
|
||||||
|
|
||||||
|
def get_city_cache(self, kind, city):
|
||||||
|
assert kind == "full"
|
||||||
|
if self.payload is None:
|
||||||
|
return None
|
||||||
|
return {"payload": self.payload}
|
||||||
|
|
||||||
|
fake_cache = FakeCache()
|
||||||
|
|
||||||
|
async def fake_run_in_threadpool(fn, *args, **kwargs):
|
||||||
|
if fn is city_api.legacy_routes._refresh_city_full_cache:
|
||||||
|
await asyncio.sleep(0.02)
|
||||||
|
return fn(*args, **kwargs)
|
||||||
|
|
||||||
|
def refresh_full(city, force_refresh):
|
||||||
|
nonlocal refresh_calls
|
||||||
|
refresh_calls += 1
|
||||||
|
fake_cache.payload = {
|
||||||
|
"city": city,
|
||||||
|
"hourly": {"times": ["2026-05-30T00:00:00Z"], "temps": [20.0]},
|
||||||
|
}
|
||||||
|
return fake_cache.payload
|
||||||
|
|
||||||
|
def build_detail(data, market_slug, target_date, resolution):
|
||||||
|
nonlocal build_calls
|
||||||
|
build_calls += 1
|
||||||
|
return {
|
||||||
|
"city": data["city"],
|
||||||
|
"hourly": data["hourly"],
|
||||||
|
"resolution": resolution,
|
||||||
|
}
|
||||||
|
|
||||||
|
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", fake_cache)
|
||||||
|
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_two_requests():
|
||||||
|
return await asyncio.gather(
|
||||||
|
city_api.get_city_detail_aggregate_payload(object(), "Paris", resolution="10m"),
|
||||||
|
city_api.get_city_detail_aggregate_payload(object(), "Paris", resolution="10m"),
|
||||||
|
)
|
||||||
|
|
||||||
|
results = asyncio.run(run_two_requests())
|
||||||
|
|
||||||
|
assert [item["city"] for item in results] == ["paris", "paris"]
|
||||||
|
assert refresh_calls == 1
|
||||||
|
assert build_calls == 1
|
||||||
|
|
||||||
|
|
||||||
|
def test_force_refresh_invalidates_short_city_detail_payload_cache(monkeypatch):
|
||||||
|
import asyncio
|
||||||
|
|
||||||
|
build_calls = 0
|
||||||
|
refreshed_payloads = [
|
||||||
|
{
|
||||||
|
"city": "paris",
|
||||||
|
"local_date": "2026-05-30",
|
||||||
|
"hourly": {"times": ["2026-05-30T00:00:00Z"], "temps": [20.0]},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"city": "paris",
|
||||||
|
"local_date": "2026-05-30",
|
||||||
|
"hourly": {"times": ["2026-05-30T00:00:00Z"], "temps": [21.0]},
|
||||||
|
},
|
||||||
|
]
|
||||||
|
|
||||||
|
class FakeCache:
|
||||||
|
def get_city_cache(self, kind, city):
|
||||||
|
assert kind == "full"
|
||||||
|
return None
|
||||||
|
|
||||||
|
async def fake_run_in_threadpool(fn, *args, **kwargs):
|
||||||
|
return fn(*args, **kwargs)
|
||||||
|
|
||||||
|
def refresh_full(city, force_refresh):
|
||||||
|
assert city == "paris"
|
||||||
|
assert refreshed_payloads
|
||||||
|
return refreshed_payloads.pop(0)
|
||||||
|
|
||||||
|
def build_detail(data, market_slug, target_date, resolution):
|
||||||
|
nonlocal build_calls
|
||||||
|
build_calls += 1
|
||||||
|
return {
|
||||||
|
"city": data["city"],
|
||||||
|
"live_temp": data["hourly"]["temps"][0],
|
||||||
|
"resolution": resolution,
|
||||||
|
}
|
||||||
|
|
||||||
|
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, "_refresh_city_full_cache", refresh_full)
|
||||||
|
monkeypatch.setattr(city_api.legacy_routes, "_build_city_detail_payload", build_detail)
|
||||||
|
|
||||||
|
first = asyncio.run(city_api.get_city_detail_aggregate_payload(object(), "Paris", resolution="10m"))
|
||||||
|
second = asyncio.run(
|
||||||
|
city_api.get_city_detail_aggregate_payload(
|
||||||
|
object(),
|
||||||
|
"Paris",
|
||||||
|
resolution="10m",
|
||||||
|
force_refresh=True,
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
assert first["live_temp"] == 20.0
|
||||||
|
assert second["live_temp"] == 21.0
|
||||||
|
assert build_calls == 2
|
||||||
|
|
||||||
|
|
||||||
def test_payment_runtime_endpoint_returns_shape():
|
def test_payment_runtime_endpoint_returns_shape():
|
||||||
|
|||||||
@@ -180,6 +180,55 @@ def test_fetch_wunderground_historical_keeps_fahrenheit_for_us_cities(monkeypatc
|
|||||||
assert payload["max_so_far"] == 75
|
assert payload["max_so_far"] == 75
|
||||||
|
|
||||||
|
|
||||||
|
def test_fetch_wunderground_historical_negative_caches_client_errors(monkeypatch, tmp_path):
|
||||||
|
collector = _collector(monkeypatch, tmp_path)
|
||||||
|
monkeypatch.setenv("WUNDERGROUND_NEGATIVE_CACHE_TTL_SEC", "900")
|
||||||
|
calls = []
|
||||||
|
|
||||||
|
def fake_get(url: str, **_kwargs):
|
||||||
|
calls.append(url)
|
||||||
|
return httpx.Response(
|
||||||
|
400,
|
||||||
|
json={"errors": [{"message": "invalid location"}]},
|
||||||
|
request=httpx.Request("GET", url),
|
||||||
|
)
|
||||||
|
|
||||||
|
monkeypatch.setattr(collector, "_http_get", fake_get)
|
||||||
|
|
||||||
|
first = collector.fetch_wunderground_historical(
|
||||||
|
"istanbul",
|
||||||
|
use_fahrenheit=False,
|
||||||
|
utc_offset=10800,
|
||||||
|
local_date="2026-05-31",
|
||||||
|
)
|
||||||
|
second = collector.fetch_wunderground_historical(
|
||||||
|
"istanbul",
|
||||||
|
use_fahrenheit=False,
|
||||||
|
utc_offset=10800,
|
||||||
|
local_date="2026-05-31",
|
||||||
|
)
|
||||||
|
|
||||||
|
assert first is None
|
||||||
|
assert second is None
|
||||||
|
assert len(calls) == 2
|
||||||
|
|
||||||
|
collector._evict_city_caches(
|
||||||
|
city="istanbul",
|
||||||
|
lat=None,
|
||||||
|
lon=None,
|
||||||
|
use_fahrenheit=False,
|
||||||
|
)
|
||||||
|
third = collector.fetch_wunderground_historical(
|
||||||
|
"istanbul",
|
||||||
|
use_fahrenheit=False,
|
||||||
|
utc_offset=10800,
|
||||||
|
local_date="2026-05-31",
|
||||||
|
)
|
||||||
|
|
||||||
|
assert third is None
|
||||||
|
assert len(calls) == 4
|
||||||
|
|
||||||
|
|
||||||
def test_fetch_all_sources_attaches_wunderground_historical(monkeypatch, tmp_path):
|
def test_fetch_all_sources_attaches_wunderground_historical(monkeypatch, tmp_path):
|
||||||
collector = _collector(monkeypatch, tmp_path)
|
collector = _collector(monkeypatch, tmp_path)
|
||||||
called = {}
|
called = {}
|
||||||
|
|||||||
+10
-3
@@ -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")
|
||||||
|
|||||||
+24
-3
@@ -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):
|
||||||
|
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
|
from src.utils.online_tracker import online_count
|
||||||
return {"online": 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
@@ -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")
|
||||||
|
|||||||
@@ -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 = (
|
||||||
|
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
|
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
|
||||||
|
|||||||
+240
-46
@@ -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
|
||||||
@@ -22,6 +23,22 @@ _RECENT_DEB_CACHE_TTL_SEC = max(
|
|||||||
60,
|
60,
|
||||||
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_LOCK = asyncio.Lock()
|
||||||
|
CityDetailPayloadCacheKey = Tuple[str, str, str, str, str, int]
|
||||||
|
_CITY_DETAIL_PAYLOAD_CACHE: Dict[CityDetailPayloadCacheKey, Dict[str, Any]] = {}
|
||||||
|
_CITY_DETAIL_PAYLOAD_CACHE_TS: Dict[CityDetailPayloadCacheKey, float] = {}
|
||||||
|
_CITY_DETAIL_PAYLOAD_INFLIGHT: Dict[CityDetailPayloadCacheKey, "asyncio.Task[Dict[str, Any]]"] = {}
|
||||||
|
_CITY_DETAIL_PAYLOAD_EPOCH: Dict[str, int] = {}
|
||||||
|
_CITY_DETAIL_PAYLOAD_LOCK = asyncio.Lock()
|
||||||
|
|
||||||
|
|
||||||
|
def _city_detail_payload_cache_ttl() -> float:
|
||||||
|
try:
|
||||||
|
value = float(os.getenv("POLYWEATHER_CITY_DETAIL_PAYLOAD_CACHE_TTL_SEC", "8") or "8")
|
||||||
|
except ValueError:
|
||||||
|
value = 8.0
|
||||||
|
return max(0.0, min(30.0, value))
|
||||||
|
|
||||||
|
|
||||||
async def _overlay_cached_wunderground(city: str, payload: Dict[str, Any]) -> Dict[str, Any]:
|
async def _overlay_cached_wunderground(city: str, payload: Dict[str, Any]) -> Dict[str, Any]:
|
||||||
@@ -32,6 +49,134 @@ async def _overlay_cached_wunderground(city: str, payload: Dict[str, Any]) -> Di
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
async def _refresh_city_full_cache_singleflight(city: str, force_refresh: bool) -> Dict[str, Any]:
|
||||||
|
key = f"{city}:{bool(force_refresh)}"
|
||||||
|
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),
|
||||||
|
)
|
||||||
|
_CITY_FULL_REFRESH_INFLIGHT[key] = task
|
||||||
|
try:
|
||||||
|
return await task
|
||||||
|
finally:
|
||||||
|
if task.done():
|
||||||
|
async with _CITY_FULL_REFRESH_LOCK:
|
||||||
|
if _CITY_FULL_REFRESH_INFLIGHT.get(key) is task:
|
||||||
|
_CITY_FULL_REFRESH_INFLIGHT.pop(key, None)
|
||||||
|
|
||||||
|
|
||||||
|
async def _invalidate_city_detail_payload_cache(city: str) -> None:
|
||||||
|
normalized = str(city or "").strip().lower()
|
||||||
|
if not normalized:
|
||||||
|
return
|
||||||
|
async with _CITY_DETAIL_PAYLOAD_LOCK:
|
||||||
|
_CITY_DETAIL_PAYLOAD_EPOCH[normalized] = _CITY_DETAIL_PAYLOAD_EPOCH.get(normalized, 0) + 1
|
||||||
|
old_keys = [key for key in _CITY_DETAIL_PAYLOAD_CACHE if key[0] == normalized]
|
||||||
|
for key in old_keys:
|
||||||
|
_CITY_DETAIL_PAYLOAD_CACHE.pop(key, None)
|
||||||
|
_CITY_DETAIL_PAYLOAD_CACHE_TS.pop(key, None)
|
||||||
|
|
||||||
|
|
||||||
|
async def _refresh_city_full_data(city: str, force_refresh: bool) -> Dict[str, Any]:
|
||||||
|
await _invalidate_city_detail_payload_cache(city)
|
||||||
|
return await _refresh_city_full_cache_singleflight(city, force_refresh)
|
||||||
|
|
||||||
|
|
||||||
|
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:
|
||||||
|
if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_FULL_CACHE_TTL_SEC):
|
||||||
|
return await _refresh_city_full_data(city, False)
|
||||||
|
return await _overlay_cached_wunderground(city, cached_entry.get("payload") or {})
|
||||||
|
return await _refresh_city_full_data(city, False)
|
||||||
|
|
||||||
|
|
||||||
|
def _city_detail_payload_cache_key(
|
||||||
|
data: Dict[str, Any],
|
||||||
|
market_slug: Optional[str],
|
||||||
|
target_date: Optional[str],
|
||||||
|
resolution: Optional[str],
|
||||||
|
) -> CityDetailPayloadCacheKey:
|
||||||
|
city = str(data.get("city") or data.get("name") or "").strip().lower()
|
||||||
|
fingerprint = str(
|
||||||
|
data.get("updated_at_ts")
|
||||||
|
or data.get("updated_at")
|
||||||
|
or data.get("local_time")
|
||||||
|
or data.get("local_date")
|
||||||
|
or id(data)
|
||||||
|
)
|
||||||
|
generation = _CITY_DETAIL_PAYLOAD_EPOCH.get(city, 0)
|
||||||
|
return (
|
||||||
|
city,
|
||||||
|
str(resolution or "10m"),
|
||||||
|
str(market_slug or ""),
|
||||||
|
str(target_date or ""),
|
||||||
|
fingerprint,
|
||||||
|
generation,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
async def _build_city_detail_payload_cached(
|
||||||
|
data: Dict[str, Any],
|
||||||
|
market_slug: Optional[str],
|
||||||
|
target_date: Optional[str],
|
||||||
|
resolution: Optional[str],
|
||||||
|
) -> Dict[str, Any]:
|
||||||
|
ttl = _city_detail_payload_cache_ttl()
|
||||||
|
if ttl <= 0:
|
||||||
|
return await run_in_threadpool(
|
||||||
|
legacy_routes._build_city_detail_payload,
|
||||||
|
data,
|
||||||
|
market_slug,
|
||||||
|
target_date,
|
||||||
|
resolution,
|
||||||
|
)
|
||||||
|
|
||||||
|
key = _city_detail_payload_cache_key(data, market_slug, target_date, resolution)
|
||||||
|
now_ts = time.time()
|
||||||
|
async with _CITY_DETAIL_PAYLOAD_LOCK:
|
||||||
|
cached = _CITY_DETAIL_PAYLOAD_CACHE.get(key)
|
||||||
|
cached_ts = _CITY_DETAIL_PAYLOAD_CACHE_TS.get(key, 0.0)
|
||||||
|
if cached is not None and now_ts - cached_ts < ttl:
|
||||||
|
return cached
|
||||||
|
task = _CITY_DETAIL_PAYLOAD_INFLIGHT.get(key)
|
||||||
|
if task is None:
|
||||||
|
task = asyncio.create_task(
|
||||||
|
run_in_threadpool(
|
||||||
|
legacy_routes._build_city_detail_payload,
|
||||||
|
data,
|
||||||
|
market_slug,
|
||||||
|
target_date,
|
||||||
|
resolution,
|
||||||
|
),
|
||||||
|
)
|
||||||
|
_CITY_DETAIL_PAYLOAD_INFLIGHT[key] = task
|
||||||
|
try:
|
||||||
|
payload = await task
|
||||||
|
finally:
|
||||||
|
if task.done():
|
||||||
|
async with _CITY_DETAIL_PAYLOAD_LOCK:
|
||||||
|
if _CITY_DETAIL_PAYLOAD_INFLIGHT.get(key) is task:
|
||||||
|
_CITY_DETAIL_PAYLOAD_INFLIGHT.pop(key, None)
|
||||||
|
|
||||||
|
async with _CITY_DETAIL_PAYLOAD_LOCK:
|
||||||
|
_CITY_DETAIL_PAYLOAD_CACHE[key] = payload
|
||||||
|
_CITY_DETAIL_PAYLOAD_CACHE_TS[key] = time.time()
|
||||||
|
if len(_CITY_DETAIL_PAYLOAD_CACHE) > 256:
|
||||||
|
oldest_keys = sorted(
|
||||||
|
_CITY_DETAIL_PAYLOAD_CACHE_TS,
|
||||||
|
key=lambda item: _CITY_DETAIL_PAYLOAD_CACHE_TS.get(item, 0.0),
|
||||||
|
)[:64]
|
||||||
|
for old_key in oldest_keys:
|
||||||
|
_CITY_DETAIL_PAYLOAD_CACHE.pop(old_key, None)
|
||||||
|
_CITY_DETAIL_PAYLOAD_CACHE_TS.pop(old_key, None)
|
||||||
|
return payload
|
||||||
|
|
||||||
|
|
||||||
def _default_deb_recent() -> Dict[str, object]:
|
def _default_deb_recent() -> Dict[str, object]:
|
||||||
return {
|
return {
|
||||||
"tier": "other",
|
"tier": "other",
|
||||||
@@ -167,14 +312,7 @@ async def get_city_detail_payload(
|
|||||||
else:
|
else:
|
||||||
detail_mode = "panel"
|
detail_mode = "panel"
|
||||||
if detail_mode == "full":
|
if detail_mode == "full":
|
||||||
if force_refresh:
|
return await _get_city_full_data(city, force_refresh=force_refresh)
|
||||||
return await run_in_threadpool(legacy_routes._refresh_city_full_cache, city, True)
|
|
||||||
cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, "full", city)
|
|
||||||
if cached_entry:
|
|
||||||
if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_FULL_CACHE_TTL_SEC):
|
|
||||||
return await run_in_threadpool(legacy_routes._refresh_city_full_cache, city, False)
|
|
||||||
return await _overlay_cached_wunderground(city, cached_entry.get("payload") or {})
|
|
||||||
return await run_in_threadpool(legacy_routes._refresh_city_full_cache, city, False)
|
|
||||||
if detail_mode == "panel":
|
if detail_mode == "panel":
|
||||||
if force_refresh:
|
if force_refresh:
|
||||||
return await run_in_threadpool(legacy_routes._refresh_city_panel_cache, city, True)
|
return await run_in_threadpool(legacy_routes._refresh_city_panel_cache, city, True)
|
||||||
@@ -231,27 +369,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,
|
||||||
if force_refresh:
|
log_name="city_detail_timing",
|
||||||
data = await run_in_threadpool(legacy_routes._refresh_city_full_cache, city, True)
|
prefix="city_detail",
|
||||||
else:
|
state_attr="city_detail_server_timing",
|
||||||
cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, "full", city)
|
)
|
||||||
if cached_entry:
|
outcome = "ok"
|
||||||
if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_FULL_CACHE_TTL_SEC):
|
status_code = 200
|
||||||
data = await run_in_threadpool(legacy_routes._refresh_city_full_cache, city, False)
|
try:
|
||||||
else:
|
timer.measure("assert_entitlement", lambda: legacy_routes._assert_entitlement(request))
|
||||||
data = await _overlay_cached_wunderground(city, cached_entry.get("payload") or {})
|
city = timer.measure("normalize_city", lambda: legacy_routes._normalize_city_or_404(name))
|
||||||
else:
|
data = await timer.measure_async(
|
||||||
data = await run_in_threadpool(legacy_routes._refresh_city_full_cache, city, False)
|
"full_data",
|
||||||
|
lambda: _get_city_full_data(city, force_refresh=force_refresh),
|
||||||
|
)
|
||||||
|
|
||||||
return await run_in_threadpool(
|
return await timer.measure_async(
|
||||||
legacy_routes._build_city_detail_payload,
|
"detail_payload",
|
||||||
|
lambda: _build_city_detail_payload_cached(
|
||||||
data,
|
data,
|
||||||
market_slug,
|
market_slug,
|
||||||
target_date,
|
target_date,
|
||||||
resolution,
|
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]:
|
||||||
@@ -271,30 +423,32 @@ def _parse_batch_city_names(raw_cities: str, *, limit: int) -> List[str]:
|
|||||||
return out
|
return out
|
||||||
|
|
||||||
|
|
||||||
def _build_city_detail_batch_item(
|
async def _build_city_detail_batch_item_async(
|
||||||
city: str,
|
city: str,
|
||||||
*,
|
*,
|
||||||
force_refresh: bool,
|
force_refresh: bool,
|
||||||
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]]:
|
||||||
if force_refresh:
|
if timing_recorder is not None:
|
||||||
data = legacy_routes._refresh_city_full_cache(city, True)
|
data = await timing_recorder.measure_async(
|
||||||
else:
|
f"full_data_{city}",
|
||||||
cached_entry = legacy_routes._CACHE_DB.get_city_cache("full", city)
|
lambda: _get_city_full_data(city, force_refresh=force_refresh),
|
||||||
if cached_entry and legacy_routes._city_cache_is_fresh(
|
)
|
||||||
cached_entry,
|
detail = await timing_recorder.measure_async(
|
||||||
legacy_routes.CITY_FULL_CACHE_TTL_SEC,
|
f"detail_payload_{city}",
|
||||||
):
|
lambda: _build_city_detail_payload_cached(
|
||||||
data = legacy_routes._overlay_latest_wunderground_current(
|
data,
|
||||||
city,
|
market_slug,
|
||||||
cached_entry.get("payload") or {},
|
target_date,
|
||||||
|
resolution,
|
||||||
|
),
|
||||||
)
|
)
|
||||||
else:
|
else:
|
||||||
data = legacy_routes._refresh_city_full_cache(city, False)
|
data = await _get_city_full_data(city, force_refresh=force_refresh)
|
||||||
|
detail = await _build_city_detail_payload_cached(
|
||||||
detail = legacy_routes._build_city_detail_payload(
|
|
||||||
data,
|
data,
|
||||||
market_slug,
|
market_slug,
|
||||||
target_date,
|
target_date,
|
||||||
@@ -303,6 +457,14 @@ def _build_city_detail_batch_item(
|
|||||||
return city, detail
|
return city, detail
|
||||||
|
|
||||||
|
|
||||||
|
def _city_detail_batch_concurrency() -> int:
|
||||||
|
try:
|
||||||
|
value = int(os.getenv("POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY", "3") or "3")
|
||||||
|
except ValueError:
|
||||||
|
value = 3
|
||||||
|
return max(1, min(6, value))
|
||||||
|
|
||||||
|
|
||||||
async def get_city_detail_batch_payload(
|
async def get_city_detail_batch_payload(
|
||||||
request: Request,
|
request: Request,
|
||||||
*,
|
*,
|
||||||
@@ -313,23 +475,47 @@ 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,
|
||||||
|
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:
|
if not city_names:
|
||||||
return {"cities": [], "details": {}, "errors": {}}
|
return {"cities": [], "details": {}, "errors": {}}
|
||||||
|
|
||||||
tasks = [
|
semaphore = asyncio.Semaphore(_city_detail_batch_concurrency())
|
||||||
run_in_threadpool(
|
|
||||||
_build_city_detail_batch_item,
|
async def _build_with_limit(city: str) -> Tuple[str, Dict[str, Any]]:
|
||||||
|
async with semaphore:
|
||||||
|
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 = [
|
||||||
|
_build_with_limit(city)
|
||||||
for city in city_names
|
for city in city_names
|
||||||
]
|
]
|
||||||
results = await asyncio.gather(*tasks, return_exceptions=True)
|
results = await timer.measure_async(
|
||||||
|
"build_details",
|
||||||
|
lambda: asyncio.gather(*tasks, return_exceptions=True),
|
||||||
|
)
|
||||||
details: Dict[str, Any] = {}
|
details: Dict[str, Any] = {}
|
||||||
errors: Dict[str, str] = {}
|
errors: Dict[str, str] = {}
|
||||||
for city, result in zip(city_names, results):
|
for city, result in zip(city_names, results):
|
||||||
@@ -344,5 +530,13 @@ async def get_city_detail_batch_payload(
|
|||||||
"details": details,
|
"details": details,
|
||||||
"errors": errors,
|
"errors": errors,
|
||||||
}
|
}
|
||||||
|
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
|
||||||
@@ -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,7 +39,16 @@ 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(
|
||||||
|
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] = {
|
filters: Dict[str, Any] = {
|
||||||
"scan_mode": scan_mode,
|
"scan_mode": scan_mode,
|
||||||
"min_price": min_price,
|
"min_price": min_price,
|
||||||
@@ -42,11 +64,24 @@ async def get_scan_terminal_payload(
|
|||||||
filters["timezone_offset_seconds"] = timezone_offset_seconds
|
filters["timezone_offset_seconds"] = timezone_offset_seconds
|
||||||
if region:
|
if region:
|
||||||
filters["trading_region"] = region
|
filters["trading_region"] = region
|
||||||
return await run_in_threadpool(
|
async def build_payload():
|
||||||
legacy_routes.build_scan_terminal_payload,
|
builder = legacy_routes.build_scan_terminal_payload
|
||||||
filters,
|
kwargs: Dict[str, Any] = {"force_refresh": force_refresh}
|
||||||
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]:
|
||||||
|
|||||||
Reference in New Issue
Block a user