Instrument API timing and reduce detail fallbacks

docker-compose.yml

@@ -79,7 +79,7 @@ services:
       timeout: 5s
     image: ghcr.io/yangyuan-zhen/polyweather-frontend:${IMAGE_TAG:-latest}
     ports:
-    - "127.0.0.1:3001:3000"
+    - 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:
-    - "127.0.0.1:8000:8000"
+    - 8000:8000
     restart: unless-stopped
     user: ${UID:-1000}:${GID:-1000}
     volumes:

frontend/app/api/analytics/events/route.ts

@@ -7,46 +7,25 @@ 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 finishProxyTimedResponse(
-      new NextResponse(null, { status: 204 }),
-      timer,
-      "disabled",
-    );
+    return new NextResponse(null, { status: 204 });
   }
 
   if (!API_BASE) {
-    return finishProxyTimedResponse(
-      NextResponse.json(
-        { error: "POLYWEATHER_API_BASE_URL is not configured" },
-        { status: 500 },
-      ),
-      timer,
-      "missing_api_base",
+    return NextResponse.json(
+      { error: "POLYWEATHER_API_BASE_URL is not configured" },
+      { status: 500 },
     );
   }
 
-  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 timer.measure("request_read", () => req.json());
+    const body = await req.json();
     const payload =
       body && typeof body.payload === "object" && body.payload != null
         ? body.payload
@@ -63,65 +42,30 @@ export async function POST(req: NextRequest) {
         referer_header: req.headers.get("referer") || "",
       },
     };
-    auth = await timer.measure("auth_headers", () =>
-      buildBackendRequestHeaders(req, {
-        includeSupabaseIdentity: false,
-      }),
-    );
+    const auth = await buildBackendRequestHeaders(req, {
+      includeSupabaseIdentity: false,
+    });
     const headers = new Headers(auth.headers);
     headers.set("Content-Type", "application/json");
-    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") || "";
+    const res = await fetch(`${API_BASE}/api/analytics/events`, {
+      method: "POST",
+      headers,
+      body: JSON.stringify(enrichedBody),
+      cache: "no-store",
+    });
     if (!res.ok) {
-      const raw = await timer.measure("backend_read", () => res.text());
+      const raw = await res.text();
       const response = buildUpstreamErrorResponse(res.status, raw, {
         detailLimit: 260,
       });
-      return finishProxyTimedResponse(
-        applyAuthResponseCookies(response, auth.response),
-        timer,
-        `upstream_${res.status}`,
-        { backendServerTiming },
-      );
+      return applyAuthResponseCookies(response, auth.response);
     }
-    const data = await timer.measure("backend_read", () => res.json());
+    const data = await res.json();
     const response = NextResponse.json(data);
-    return finishProxyTimedResponse(
-      applyAuthResponseCookies(response, auth.response),
-      timer,
-      "ok",
-      { backendServerTiming },
-    );
+    return applyAuthResponseCookies(response, auth.response);
   } catch (error) {
-    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);
+    return buildProxyExceptionResponse(error, {
+      publicMessage: "Failed to track analytics event",
+    });
   }
 }

frontend/app/api/cities/detail-batch/route.ts

@@ -42,13 +42,8 @@ export async function GET(req: NextRequest) {
   try {
     return await proxyBackendJsonGet(req, {
       cacheControl: cachePolicy.responseCacheControl,
-      cacheControlForData: (data) =>
-        data &&
-        typeof data === "object" &&
-        (data as { partial?: unknown }).partial === true
-          ? "no-store, max-age=0"
-          : cachePolicy.responseCacheControl,
-      fetchCache: "no-store",
+      fetchCache:
+        cachePolicy.fetchMode === "no-store" ? "no-store" : undefined,
       publicMessage: "Failed to fetch city detail batch",
       revalidateSeconds: cachePolicy.revalidateSeconds,
       signal: controller.signal,

frontend/components/dashboard/ScanTerminalDashboard.tsx

@@ -55,10 +55,7 @@ import {
   mergeAccessStateWithAuthPayload,
   type AuthProfilePayload,
 } from "@/components/dashboard/scan-terminal/terminal-access-state";
-import {
-  createAuthProfileRequestCache,
-  loadTerminalAuthProfile,
-} from "@/components/dashboard/scan-terminal/terminal-auth-bootstrap";
+import { loadTerminalAuthProfile } from "@/components/dashboard/scan-terminal/terminal-auth-bootstrap";
 import {
   cityListItemsToScanRows,
   mergeScanRowsWithCityFallbackRows,
@@ -965,7 +962,7 @@ function ScanTerminalScreen() {
     createEmptyAccess(true),
   );
 
-  const rawLoadAuthProfile = useCallback(
+  const loadAuthProfile = useCallback(
     async (
       accessToken?: string | null,
       options?: { preferSnapshot?: boolean },
@@ -987,28 +984,6 @@ 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();

frontend/components/dashboard/scan-terminal/__tests__/refreshCadencePolicy.test.ts

@@ -96,18 +96,6 @@ export async function runTests() {
       chartLogicSource.includes("primeCityDetailCache"),
     "visible terminal chart detail fetches should be coalesced into one batch request and prime the shared chart cache",
   );
-  assert(
-    chartLogicSource.includes("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)") &&

frontend/components/dashboard/scan-terminal/__tests__/ssePatchArchitecture.test.ts

@@ -99,18 +99,6 @@ export function runTests() {
   assert(hook.includes("useLatestPatch"), "frontend patch hook must export useLatestPatch(city)");
   assert(hook.includes("revision"), "frontend patch hook must track revisions and skip stale patches");
   assert(hook.includes("setTimeout"), "frontend patch hook must implement explicit reconnect backoff");
-  assert(
-    hook.includes("SSE_SUBSCRIPTION_RECONNECT_DELAY_MS") &&
-      hook.includes("scheduleSubscriptionReconnect") &&
-      hook.includes("clearSubscriptionReconnectTimer"),
-    "frontend patch hook must debounce visible-city subscription changes before reopening SSE",
-  );
-  const subscriptionBlock = hook.match(/function registerCitySubscription[\s\S]*?\n}\r?\n\r?\nfunction normalizeLegacyPatch/)?.[0] || "";
-  assert(
-    subscriptionBlock.includes("scheduleSubscriptionReconnect()") &&
-      !subscriptionBlock.includes("ensureSsePatchConnection();"),
-    "city subscription mount/unmount should schedule one coalesced SSE reconnect instead of reconnecting per chart",
-  );
 
   const bffEventsRoute = readFrontendFile("app", "api", "events", "route.ts");
   assert(bffEventsRoute.includes("searchParams"), "Next.js SSE proxy must forward query parameters to FastAPI");

frontend/components/dashboard/scan-terminal/__tests__/terminalAuthBootstrap.test.ts

@@ -1,5 +1,4 @@
 import {
-  createAuthProfileRequestCache,
   loadTerminalAuthProfile,
   type TerminalAuthProfilePayload,
 } from "@/components/dashboard/scan-terminal/terminal-auth-bootstrap";
@@ -28,41 +27,6 @@ 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[] = [];

frontend/components/dashboard/scan-terminal/temperature-chart-logic.ts

@@ -966,11 +966,8 @@ type HourlyForecastFetchOptions = {
 };
 
 type CityDetailBatchPayload = {
-  cities?: string[];
   details?: Record<string, CityDetail | null | undefined>;
   errors?: Record<string, string>;
-  missing?: string[];
-  partial?: boolean;
 };
 
 type CityDetailBatchWaiter = {
@@ -1115,10 +1112,6 @@ 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);
@@ -1128,10 +1121,6 @@ async function flushCityDetailBatch(resolution: string) {
           resolveBatchWaiters(waiters, data);
           return;
         }
-        if (partialMissingCities.has(normalizeCityKey(city))) {
-          resolveBatchWaiters(waiters, null);
-          return;
-        }
         try {
           resolveBatchWaiters(
             waiters,

frontend/components/dashboard/scan-terminal/terminal-auth-bootstrap.ts

@@ -57,32 +57,6 @@ 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,

frontend/components/ops/__tests__/apiPerformanceTiming.test.ts

@@ -41,21 +41,6 @@ export function runTests() {
   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"\)/);
@@ -72,26 +57,4 @@ export function runTests() {
   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",
-  );
 }

frontend/hooks/use-sse-patches.ts

@@ -4,7 +4,6 @@ import { useEffect, useSyncExternalStore } from "react";
 import { resolveBackendApiUrl } from "@/lib/backend-api";
 
 const V1_EVENT_TYPE = "city_observation_patch.v1";
-const SSE_SUBSCRIPTION_RECONNECT_DELAY_MS = 150;
 
 export type CityPatch = {
   type?: string;
@@ -39,7 +38,6 @@ const subscribedCities = new Map<string, number>();
 
 let eventSource: EventSource | null = null;
 let reconnectTimer: ReturnType<typeof setTimeout> | null = null;
-let subscriptionReconnectTimer: ReturnType<typeof setTimeout> | null = null;
 let reconnectAttempt = 0;
 let patchVersion = 0;
 let resyncVersion = 0;
@@ -76,12 +74,6 @@ function clearReconnectTimer() {
   reconnectTimer = null;
 }
 
-function clearSubscriptionReconnectTimer() {
-  if (!subscriptionReconnectTimer) return;
-  clearTimeout(subscriptionReconnectTimer);
-  subscriptionReconnectTimer = null;
-}
-
 function buildSseUrl(baseUrl: string) {
   const params = new URLSearchParams();
   const cities = subscribedCityList();
@@ -121,7 +113,6 @@ function closeEventSource() {
 function reconnectNow() {
   if (typeof window === "undefined") return;
   clearReconnectTimer();
-  clearSubscriptionReconnectTimer();
   closeEventSource();
   connectSsePatches();
 }
@@ -168,7 +159,6 @@ function connectSsePatches() {
 
 function ensureSsePatchConnection() {
   if (subscribedCities.size === 0) {
-    clearSubscriptionReconnectTimer();
     closeEventSource();
     clearReconnectTimer();
     return;
@@ -177,19 +167,6 @@ function ensureSsePatchConnection() {
   reconnectNow();
 }
 
-function scheduleSubscriptionReconnect() {
-  if (typeof window === "undefined") return;
-  if (subscribedCities.size === 0) {
-    ensureSsePatchConnection();
-    return;
-  }
-  clearSubscriptionReconnectTimer();
-  subscriptionReconnectTimer = setTimeout(() => {
-    subscriptionReconnectTimer = null;
-    ensureSsePatchConnection();
-  }, SSE_SUBSCRIPTION_RECONNECT_DELAY_MS);
-}
-
 function registerCitySubscription(city: string) {
   const cityKey = normalizeCityKey(city);
   if (!cityKey) return () => {};
@@ -197,7 +174,7 @@ function registerCitySubscription(city: string) {
   const previousCount = subscribedCities.get(cityKey) ?? 0;
   subscribedCities.set(cityKey, previousCount + 1);
   if (previousCount === 0) {
-    scheduleSubscriptionReconnect();
+    ensureSsePatchConnection();
   }
 
   return () => {
@@ -207,7 +184,7 @@ function registerCitySubscription(city: string) {
     } else {
       subscribedCities.set(cityKey, nextCount);
     }
-    scheduleSubscriptionReconnect();
+    ensureSsePatchConnection();
   };
 }
 

frontend/lib/api-proxy.ts

@@ -85,7 +85,6 @@ export async function proxyBackendJsonGet(
   req: NextRequest,
   options: {
     cacheControl?: string;
-    cacheControlForData?: (data: unknown) => string | undefined;
     conditionalResponse?: boolean;
     detailLimit?: number;
     error?: string;
@@ -149,14 +148,12 @@ export async function proxyBackendJsonGet(
     const data = await (timing
       ? timing.measure("backend_read", () => res.json())
       : res.json());
-    const responseCacheControl =
-      options.cacheControlForData?.(data) ?? options.cacheControl;
     const response =
-      responseCacheControl && options.conditionalResponse !== false
-        ? buildCachedJsonResponse(req, data, responseCacheControl)
+      options.cacheControl && options.conditionalResponse !== false
+        ? buildCachedJsonResponse(req, data, options.cacheControl)
         : NextResponse.json(data, {
-            headers: responseCacheControl
-              ? { "Cache-Control": responseCacheControl }
+            headers: options.cacheControl
+              ? { "Cache-Control": options.cacheControl }
               : undefined,
           });
     const withCookies = applyAuthResponseCookies(response, auth.response);

tests/test_deployment_runtime_config.py

@@ -80,15 +80,6 @@ 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")
 

tests/test_web_observability.py

@@ -315,15 +315,6 @@ def test_ops_billing_risk_surfaces_trial_payment_referral_and_points(monkeypatch
         DBManager,
         "list_app_analytics_events",
         lambda self, limit=20000, since_iso=None: [
-            {
-                "id": 10,
-                "event_type": "login_start",
-                "user_id": None,
-                "client_id": "c1",
-                "session_id": "session-gap",
-                "created_at": recent,
-                "payload": {"mode": "signup"},
-            },
             {
                 "id": 11,
                 "event_type": "signup_success",
@@ -428,108 +419,6 @@ def test_ops_billing_risk_does_not_flag_signup_when_backend_trial_exists(monkeyp
     assert not any(issue["category"] == "signup_trial" for issue in payload["issues"])
 
 
-def test_ops_billing_risk_ignores_account_visit_without_signup_intent(monkeypatch):
-    from src.database.db_manager import DBManager
-
-    recent = datetime.now(timezone.utc).isoformat()
-
-    monkeypatch.setattr(ops_api.legacy_routes, "_require_ops_admin", lambda request: {"email": "ops@example.com"})
-    monkeypatch.setattr(ops_api, "_supabase_rest_rows", lambda table, params, *, timeout=10: [])
-    monkeypatch.setattr(
-        DBManager,
-        "list_app_analytics_events",
-        lambda self, limit=20000, since_iso=None: [
-            {
-                "id": 61,
-                "event_type": "signup_success",
-                "user_id": "returning-no-trial-user",
-                "client_id": "client-return",
-                "session_id": "session-return",
-                "created_at": recent,
-                "payload": {
-                    "entry": "account_center",
-                    "user_id": "returning-no-trial-user",
-                },
-            }
-        ],
-    )
-    monkeypatch.setattr(
-        DBManager,
-        "list_payment_audit_events",
-        lambda self, limit=50, event_type=None: [],
-    )
-
-    payload = ops_api.get_ops_billing_risk(None, days=30, limit=20)
-
-    assert payload["summary"]["trial_gaps"] == 0
-    assert not any(issue["category"] == "signup_trial" for issue in payload["issues"])
-
-
-def test_ops_billing_risk_treats_expired_signup_trial_subscription_as_backend_evidence(monkeypatch):
-    from src.database.db_manager import DBManager
-
-    now = datetime.now(timezone.utc)
-    recent = now.isoformat()
-    expired = (now - timedelta(days=2)).isoformat()
-
-    def fake_supabase_rows(table, params, *, timeout=10):
-        if table == "subscriptions" and params.get("or") == "(source.eq.signup_trial,plan_code.eq.signup_trial_3d)":
-            return [
-                {
-                    "id": 81,
-                    "user_id": "expired-trial-user",
-                    "plan_code": "signup_trial_3d",
-                    "source": "signup_trial",
-                    "status": "expired",
-                    "starts_at": (now - timedelta(days=5)).isoformat(),
-                    "expires_at": expired,
-                    "created_at": (now - timedelta(days=5)).isoformat(),
-                    "updated_at": expired,
-                }
-            ]
-        return []
-
-    monkeypatch.setattr(ops_api.legacy_routes, "_require_ops_admin", lambda request: {"email": "ops@example.com"})
-    monkeypatch.setattr(ops_api, "_supabase_rest_rows", fake_supabase_rows)
-    monkeypatch.setattr(
-        DBManager,
-        "list_app_analytics_events",
-        lambda self, limit=20000, since_iso=None: [
-            {
-                "id": 80,
-                "event_type": "login_start",
-                "user_id": None,
-                "client_id": "client-expired",
-                "session_id": "session-expired",
-                "created_at": recent,
-                "payload": {"mode": "signup"},
-            },
-            {
-                "id": 82,
-                "event_type": "signup_success",
-                "user_id": "expired-trial-user",
-                "client_id": "client-expired",
-                "session_id": "session-expired",
-                "created_at": recent,
-                "payload": {
-                    "entry": "account_center",
-                    "user_id": "expired-trial-user",
-                },
-            },
-        ],
-    )
-    monkeypatch.setattr(
-        DBManager,
-        "list_payment_audit_events",
-        lambda self, limit=50, event_type=None: [],
-    )
-
-    payload = ops_api.get_ops_billing_risk(None, days=30, limit=20)
-
-    assert payload["summary"]["trial_gaps"] == 0
-    assert not any(issue["category"] == "signup_trial" for issue in payload["issues"])
-
-
 def test_ops_payment_incidents_expose_top_level_reason_and_filters_resolved(monkeypatch):
     from src.database.db_manager import DBManager
 
@@ -729,44 +618,6 @@ 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
 
@@ -827,70 +678,6 @@ 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
 

web/services/city_api.py

@@ -24,7 +24,6 @@ _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]] = {}
@@ -55,17 +54,9 @@ 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:
-            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())
+            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
@@ -93,40 +84,14 @@ 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, payload)
+        return await _overlay_cached_wunderground(city, cached_entry.get("payload") or {})
     return await _refresh_city_full_data(city, False)
 
 
@@ -500,19 +465,6 @@ 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,
     *,
@@ -541,13 +493,7 @@ async def get_city_detail_batch_payload(
             ),
         )
         if not city_names:
-            return {
-                "cities": [],
-                "details": {},
-                "errors": {},
-                "missing": [],
-                "partial": False,
-            }
+            return {"cities": [], "details": {}, "errors": {}}
 
         semaphore = asyncio.Semaphore(_city_detail_batch_concurrency())
 
@@ -562,47 +508,27 @@ async def get_city_detail_batch_payload(
                     timing_recorder=timer,
                 )
 
-        task_by_city = {
-            city: asyncio.create_task(_build_with_limit(city))
+        tasks = [
+            _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(
+        ]
+        results = await timer.measure_async(
             "build_details",
-            lambda: asyncio.wait(
-                task_by_city.values(),
-                timeout=_city_detail_batch_partial_timeout_seconds(),
-            ),
+            lambda: asyncio.gather(*tasks, return_exceptions=True),
         )
         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)
+        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
 
-        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}"

web/services/ops_api.py

@@ -545,41 +545,18 @@ def get_ops_billing_risk(
             "limit": str(max(safe_limit * 10, 500)),
         },
     )
-    trial_subscription_rows = collect(
+    subscription_rows = collect(
         "subscriptions",
         {
             "select": (
                 "id,user_id,plan_code,source,status,starts_at,expires_at,"
                 "created_at,updated_at"
             ),
-            "or": "(source.eq.signup_trial,plan_code.eq.signup_trial_3d)",
+            "or": "(source.eq.signup_trial,plan_code.eq.signup_trial_3d,status.eq.active)",
             "order": "created_at.desc",
             "limit": str(max(safe_limit * 20, 1000)),
         },
     )
-    active_subscription_rows = collect(
-        "subscriptions",
-        {
-            "select": (
-                "id,user_id,plan_code,source,status,starts_at,expires_at,"
-                "created_at,updated_at"
-            ),
-            "status": "eq.active",
-            "order": "created_at.desc",
-            "limit": str(max(safe_limit * 20, 1000)),
-        },
-    )
-    subscription_rows: List[Dict[str, Any]] = []
-    seen_subscription_keys: set[str] = set()
-    for row in [*trial_subscription_rows, *active_subscription_rows]:
-        key = str(row.get("id") or "").strip() or (
-            f"{row.get('user_id')}:{row.get('plan_code')}:{row.get('source')}:"
-            f"{row.get('starts_at')}:{row.get('expires_at')}"
-        )
-        if key in seen_subscription_keys:
-            continue
-        seen_subscription_keys.add(key)
-        subscription_rows.append(row)
     entitlement_trial_events = collect(
         "entitlement_events",
         {
@@ -774,40 +751,6 @@ def get_ops_billing_risk(
     def normalize_user_key(value: Any) -> str:
         return str(value or "").strip().lower()
 
-    def analytics_payload(row: Dict[str, Any]) -> Dict[str, Any]:
-        payload = row.get("payload")
-        return payload if isinstance(payload, dict) else {}
-
-    def analytics_correlation_keys(row: Dict[str, Any]) -> set[str]:
-        payload = analytics_payload(row)
-        keys: set[str] = set()
-        user_id = normalize_user_key(row.get("user_id") or payload.get("user_id"))
-        client_id = str(row.get("client_id") or "").strip()
-        session_id = str(row.get("session_id") or "").strip()
-        if user_id:
-            keys.add(f"user:{user_id}")
-        if client_id:
-            keys.add(f"client:{client_id}")
-        if session_id:
-            keys.add(f"session:{session_id}")
-        return keys
-
-    signup_intent_keys: set[str] = set()
-    for row in events:
-        if str(row.get("event_type") or "").strip().lower() != "login_start":
-            continue
-        payload = analytics_payload(row)
-        mode = str(payload.get("mode") or payload.get("auth_mode") or "").strip().lower()
-        if mode == "signup":
-            signup_intent_keys.update(analytics_correlation_keys(row))
-
-    def has_signup_intent(row: Dict[str, Any]) -> bool:
-        payload = analytics_payload(row)
-        mode = str(payload.get("mode") or payload.get("auth_mode") or "").strip().lower()
-        if mode == "signup" or payload.get("signup_intent") is True:
-            return True
-        return bool(analytics_correlation_keys(row).intersection(signup_intent_keys))
-
     trial_actor_keys = {
         _app_analytics_actor_key(row)
         for row in events
@@ -874,12 +817,10 @@ def get_ops_billing_risk(
             )
 
     for row in signup_rows[:300]:
-        if not has_signup_intent(row):
-            continue
         actor_key = _app_analytics_actor_key(row)
         if actor_key in trial_actor_keys:
             continue
-        payload = analytics_payload(row)
+        payload = row.get("payload") if isinstance(row.get("payload"), dict) else {}
         signup_user_id = normalize_user_key(row.get("user_id") or payload.get("user_id"))
         if not signup_user_id:
             continue