Reduce polling and batch observation writes

2026-06-08 15:12:41 +08:00
parent 96dde3b50e
commit 8f6925108c
12 changed files with 281 additions and 24 deletions
@@ -102,6 +102,8 @@ const TERMINAL_NAV_ITEMS = [
 ] as const;
 const AUTH_PROFILE_REQUEST_TIMEOUT_MS = 4500;
 const AUTH_DECISION_RECOVERY_MS = 10_000;
+const AUTH_PROFILE_RETRY_INITIAL_DELAY_MS = 5_000;
+const AUTH_PROFILE_RETRY_POLL_MS = 30_000;
 const ACTIVE_ACCESS_CACHE_KEY = "polyweather_terminal_active_access_v1";
 const ACTIVE_ACCESS_CACHE_TTL_MS = 6 * 60 * 60 * 1000;

@@ -1554,10 +1556,10 @@ function ScanTerminalScreen() {

    const firstRetry = window.setTimeout(() => {
      void retryAuthProfile();
-    }, 1500);
+    }, AUTH_PROFILE_RETRY_INITIAL_DELAY_MS);
    const interval = window.setInterval(() => {
      void retryAuthProfile();
-    }, 5000);
+    }, AUTH_PROFILE_RETRY_POLL_MS);
    return () => {
      cancelled = true;
      window.clearTimeout(firstRetry);
@@ -1,17 +1,19 @@
 "use client";

-import { useCallback, useEffect, useMemo, useState } from "react";
+import { useCallback, useEffect, useMemo, useRef, useState } from "react";
 import { Bell, RefreshCcw } from "lucide-react";
 import type { UserFeedbackEntry, UserFeedbackPayload } from "@/types/ops";
 import {
  buildFeedbackNotificationKey,
  countUnseenFeedbackUpdates,
+  FEEDBACK_STATUS_CACHE_TTL_MS,
  FEEDBACK_STATUS_POLL_MS,
  feedbackStatusLabel,
  feedbackStatusTone,
 } from "./feedback-status";

 const FEEDBACK_STATUS_SEEN_KEY = "polyweather_feedback_status_seen_v1";
+const FEEDBACK_STATUS_CACHE_KEY = "polyweather_feedback_status_cache_v1";

 function loadSeenKeys() {
  if (typeof window === "undefined") return new Set<string>();
@@ -34,6 +36,33 @@ function saveSeenKeys(keys: Set<string>) {
  }
 }

+function readFeedbackStatusCache(): { entries: UserFeedbackEntry[]; ts: number } | null {
+  if (typeof window === "undefined") return null;
+  try {
+    const raw = window.localStorage.getItem(FEEDBACK_STATUS_CACHE_KEY);
+    if (!raw) return null;
+    const parsed = JSON.parse(raw);
+    const ts = Number(parsed?.ts || 0);
+    const entries = Array.isArray(parsed?.entries) ? parsed.entries : [];
+    if (!ts || Date.now() - ts > FEEDBACK_STATUS_CACHE_TTL_MS) return null;
+    return { entries, ts };
+  } catch {
+    return null;
+  }
+}
+
+function writeFeedbackStatusCache(entries: UserFeedbackEntry[]) {
+  if (typeof window === "undefined") return;
+  try {
+    window.localStorage.setItem(
+      FEEDBACK_STATUS_CACHE_KEY,
+      JSON.stringify({ entries: entries.slice(0, 20), ts: Date.now() }),
+    );
+  } catch {
+    // Ignore storage failures; status will still refresh over the network.
+  }
+}
+
 function compactDate(value?: string) {
  if (!value) return "";
  return value.slice(0, 16).replace("T", " ");
@@ -57,11 +86,12 @@ export function UserFeedbackStatusButton({
  refreshKey?: number;
 }) {
  const [available, setAvailable] = useState(true);
-  const [entries, setEntries] = useState<UserFeedbackEntry[]>([]);
+  const [entries, setEntries] = useState<UserFeedbackEntry[]>(() => readFeedbackStatusCache()?.entries || []);
  const [error, setError] = useState("");
  const [loading, setLoading] = useState(false);
  const [open, setOpen] = useState(false);
  const [seenKeys, setSeenKeys] = useState<Set<string>>(() => loadSeenKeys());
+  const lastLoadedAtRef = useRef<number>(readFeedbackStatusCache()?.ts || 0);

  const unseenCount = useMemo(
    () => countUnseenFeedbackUpdates(entries, seenKeys),
@@ -75,8 +105,26 @@ export function UserFeedbackStatusButton({
      ? "No submitted feedback yet."
      : "暂无已提交反馈。";

-  const load = useCallback(async (signal?: AbortSignal) => {
+  const load = useCallback(async (
+    signal?: AbortSignal,
+    options?: { force?: boolean },
+  ) => {
    if (typeof fetch !== "function") return;
+    const cached = readFeedbackStatusCache();
+    if (!options?.force && cached) {
+      lastLoadedAtRef.current = cached.ts;
+      setEntries(cached.entries);
+      setAvailable(true);
+      return;
+    }
+    const now = Date.now();
+    if (
+      !options?.force &&
+      lastLoadedAtRef.current &&
+      now - lastLoadedAtRef.current < FEEDBACK_STATUS_CACHE_TTL_MS
+    ) {
+      return;
+    }
    setLoading(true);
    try {
      const res = await fetch("/api/feedback?limit=12", {
@@ -93,7 +141,10 @@ export function UserFeedbackStatusButton({
        throw new Error(`HTTP ${res.status}`);
      }
      const payload = (await res.json()) as UserFeedbackPayload;
-      setEntries(Array.isArray(payload.feedback) ? payload.feedback : []);
+      const nextEntries = Array.isArray(payload.feedback) ? payload.feedback : [];
+      setEntries(nextEntries);
+      writeFeedbackStatusCache(nextEntries);
+      lastLoadedAtRef.current = Date.now();
      setAvailable(true);
      setError("");
    } catch (err) {
@@ -115,10 +166,15 @@ export function UserFeedbackStatusButton({

  useEffect(() => {
    const controller = new AbortController();
-    void load(controller.signal);
+    void load(controller.signal, { force: refreshKey > 0 });

    const handleVisibilityChange = () => {
-      if (document.visibilityState === "visible") void load();
+      if (
+        document.visibilityState === "visible" &&
+        Date.now() - lastLoadedAtRef.current >= FEEDBACK_STATUS_CACHE_TTL_MS
+      ) {
+        void load();
+      }
    };
    document.addEventListener("visibilitychange", handleVisibilityChange);
    const id = window.setInterval(() => {
@@ -174,7 +230,7 @@ export function UserFeedbackStatusButton({
            </div>
            <button
              type="button"
-              onClick={() => void load()}
+              onClick={() => void load(undefined, { force: true })}
              disabled={loading}
              className="grid h-7 w-7 place-items-center rounded border border-slate-200 text-slate-500 transition hover:bg-slate-50 disabled:cursor-wait disabled:opacity-60"
              title={isEn ? "Refresh" : "刷新"}
@@ -1,6 +1,10 @@
+import fs from "node:fs";
+import path from "node:path";
+
 import {
  buildFeedbackNotificationKey,
  countUnseenFeedbackUpdates,
+  FEEDBACK_STATUS_CACHE_TTL_MS,
  FEEDBACK_STATUS_POLL_MS,
  feedbackStatusLabel,
 } from "@/components/dashboard/scan-terminal/feedback-status";
@@ -10,7 +14,22 @@ function assert(condition: unknown, message: string): asserts condition {
 }

 export function runTests() {
-  assert(FEEDBACK_STATUS_POLL_MS === 600_000, "feedback bell background polling should run every 10 minutes");
+  const projectRoot = process.cwd();
+  const statusButtonSource = fs.readFileSync(
+    path.join(projectRoot, "components", "dashboard", "scan-terminal", "UserFeedbackStatusButton.tsx"),
+    "utf8",
+  );
+
+  assert(FEEDBACK_STATUS_POLL_MS === 30 * 60 * 1000, "feedback bell background polling should run every 30 minutes");
+  assert(FEEDBACK_STATUS_CACHE_TTL_MS === 10 * 60 * 1000, "feedback bell should reuse a 10 minute local status cache");
+  assert(
+    statusButtonSource.includes("FEEDBACK_STATUS_CACHE_KEY") &&
+      statusButtonSource.includes("readFeedbackStatusCache") &&
+      statusButtonSource.includes("writeFeedbackStatusCache") &&
+      statusButtonSource.includes("lastLoadedAtRef") &&
+      statusButtonSource.includes("FEEDBACK_STATUS_CACHE_TTL_MS"),
+    "feedback status button must use a local cache and avoid refetching on every visibility resume",
+  );
  assert(feedbackStatusLabel("open", false) === "已收到", "open feedback should read as received to users");
  assert(feedbackStatusLabel("triaged", false) === "已确认", "triaged feedback should read as confirmed to users");
  assert(feedbackStatusLabel("investigating", false) === "处理中", "investigating feedback should read as in progress");
@@ -168,6 +168,12 @@ export async function runTests() {
      dashboardSource.includes('document.visibilityState === "hidden"'),
    "terminal online-user presence should refresh slowly and pause while the browser tab is hidden",
  );
+  assert(
+    dashboardSource.includes("AUTH_PROFILE_RETRY_INITIAL_DELAY_MS = 5_000") &&
+      dashboardSource.includes("AUTH_PROFILE_RETRY_POLL_MS = 30_000") &&
+      !dashboardSource.includes("}, 5000);"),
+    "terminal auth subscription retry should back off after the first retry instead of polling auth/me every 5 seconds",
+  );
  assert(
    chartSource.includes("IntersectionObserver") &&
      chartSource.includes("shouldFetchCityDetailForChart") &&
@@ -1,6 +1,7 @@
 import type { UserFeedbackEntry } from "@/types/ops";

-export const FEEDBACK_STATUS_POLL_MS = 10 * 60 * 1000;
+export const FEEDBACK_STATUS_POLL_MS = 30 * 60 * 1000;
+export const FEEDBACK_STATUS_CACHE_TTL_MS = 10 * 60 * 1000;

 export function feedbackStatusLabel(status: string | undefined, isEn: boolean) {
  const key = String(status || "open").toLowerCase();
@@ -1155,17 +1155,21 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour
            try:
                from src.database.db_manager import DBManager
                db = DBManager()
+                obs_rows = []
                for row in cluster_data:
                    icao = row.get("icao")
                    temp_c = row.get("temp_c") if "temp_c" in row else row.get("temp")
                    if icao and temp_c is not None:
-                        db.append_airport_obs(
-                            icao=str(icao),
-                            city=city_lower,
-                            temp_c=temp_c,
-                            wind_kt=row.get("wind_speed_kt"),
-                            obs_time=str(row.get("obs_time") or datetime.now().isoformat()),
+                        obs_rows.append(
+                            {
+                                "icao": str(icao),
+                                "city": city_lower,
+                                "temp_c": temp_c,
+                                "wind_kt": row.get("wind_speed_kt"),
+                                "obs_time": str(row.get("obs_time") or datetime.now().isoformat()),
+                            }
                        )
+                db.append_airport_obs_batch(obs_rows)
            except Exception:
                logger.exception("airport_obs_log append failed for metar cluster city={}", city_lower)

@@ -3019,16 +3019,50 @@ class DBManager:
        pressure_hpa: Optional[float] = None,
        obs_time: str,
    ) -> None:
-        safe_icao = str(icao).strip().upper()
-        safe_city = str(city).strip().lower()
+        self.append_airport_obs_batch(
+            [
+                {
+                    "icao": icao,
+                    "city": city,
+                    "temp_c": temp_c,
+                    "wind_kt": wind_kt,
+                    "pressure_hpa": pressure_hpa,
+                    "obs_time": obs_time,
+                }
+            ]
+        )
+
+    def append_airport_obs_batch(self, rows: List[Dict[str, Any]]) -> None:
+        normalized_rows: List[Tuple[str, str, Optional[float], Optional[float], Optional[float], str]] = []
+        for row in rows or []:
+            if not isinstance(row, dict):
+                continue
+            safe_icao = str(row.get("icao") or "").strip().upper()
+            safe_city = str(row.get("city") or "").strip().lower()
+            safe_obs_time = str(row.get("obs_time") or "").strip()
+            if not safe_icao or not safe_city or not safe_obs_time:
+                continue
+            normalized_rows.append(
+                (
+                    safe_icao,
+                    safe_city,
+                    row.get("temp_c"),
+                    row.get("wind_kt"),
+                    row.get("pressure_hpa"),
+                    safe_obs_time,
+                )
+            )
+        if not normalized_rows:
+            return
+        first_icao, first_city = normalized_rows[0][0], normalized_rows[0][1]
        try:
            with self._get_connection() as conn:
-                conn.execute(
+                conn.executemany(
                    """
                    INSERT INTO airport_obs_log (icao, city, temp_c, wind_kt, pressure_hpa, obs_time)
                    VALUES (?, ?, ?, ?, ?, ?)
                    """,
-                    (safe_icao, safe_city, temp_c, wind_kt, pressure_hpa, str(obs_time)),
+                    normalized_rows,
                )
                conn.execute(
                    "DELETE FROM airport_obs_log WHERE created_at < datetime('now', '-2 hours')"
@@ -3038,8 +3072,8 @@ class DBManager:
            if self._is_sqlite_locked_error(exc):
                logger.warning(
                    "airport obs log skipped because sqlite is locked icao={} city={}",
-                    safe_icao,
-                    safe_city,
+                    first_icao,
+                    first_city,
                )
                return
            raise
@@ -22,6 +22,32 @@ def test_backend_shared_timing_helper_avoids_sensitive_identity_fields():
    assert "email" not in source


+def test_server_timing_recorder_warns_on_slow_stages(monkeypatch):
+    import web.services.request_timing as request_timing
+
+    warnings = []
+    monkeypatch.setenv("POLYWEATHER_SLOW_STAGE_LOG_MS", "0")
+    monkeypatch.setattr(
+        request_timing.logger,
+        "warning",
+        lambda *args, **kwargs: warnings.append(args),
+    )
+
+    recorder = request_timing.ServerTimingRecorder(
+        None,
+        log_name="unit_timing",
+        prefix="unit",
+        state_attr="unit_server_timing",
+    )
+    recorder.measure("sqlite_query", lambda: None)
+    recorder.finish(outcome="ok", status_code=200)
+
+    assert warnings
+    assert warnings[0][0] == "slow_request_timing log_name={} outcome={} status_code={} slow_stages={}"
+    assert warnings[0][1] == "unit_timing"
+    assert "sqlite_query" in warnings[0][4]
+
+
 def test_city_detail_batch_response_includes_backend_server_timing(monkeypatch):
    city_api._CITY_DETAIL_BATCH_RESPONSE_CACHE.clear()
    city_api._CITY_DETAIL_BATCH_RESPONSE_CACHE_TS.clear()
@@ -288,3 +288,38 @@ def test_ephemeral_observation_log_writes_skip_sqlite_lock(monkeypatch, tmp_path
        target_runway_max=24.0,
        otime_utc="2026-06-08T04:00:00Z",
    )
+
+
+def test_airport_obs_batch_writes_share_one_sqlite_transaction(monkeypatch, tmp_path):
+    from src.database.db_manager import DBManager
+
+    db = DBManager(str(tmp_path / "polyweather-airport-obs-batch.db"))
+    original_get_connection = db._get_connection
+    connection_calls = {"count": 0}
+
+    def counting_connection():
+        connection_calls["count"] += 1
+        return original_get_connection()
+
+    monkeypatch.setattr(db, "_get_connection", counting_connection)
+
+    db.append_airport_obs_batch(
+        [
+            {
+                "icao": "ZBAA",
+                "city": "beijing",
+                "temp_c": 24.0,
+                "obs_time": "2026-06-08T04:00:00Z",
+            },
+            {
+                "icao": "ZBAD",
+                "city": "beijing",
+                "temp_c": 25.0,
+                "obs_time": "2026-06-08T04:00:00Z",
+            },
+        ]
+    )
+
+    assert connection_calls["count"] == 1
+    assert [row["temp_c"] for row in db.get_airport_obs_recent("ZBAA", minutes=180)] == [24.0]
+    assert [row["temp_c"] for row in db.get_airport_obs_recent("ZBAD", minutes=180)] == [25.0]
@@ -1959,6 +1959,57 @@ def test_auth_me_uses_subscription_window_as_required_subscription_gate(monkeypa
    assert payload["subscription_queued_days"] == 30


+def test_auth_me_entitlement_scope_reuses_subscription_window_cache(monkeypatch):
+    monkeypatch.setattr(web_core.SUPABASE_ENTITLEMENT, "enabled", True)
+    monkeypatch.setattr(web_core.SUPABASE_ENTITLEMENT, "require_subscription", False)
+    monkeypatch.setattr(web_core, "_SUPABASE_AUTH_REQUIRED", False)
+    monkeypatch.setattr(routes, "_resolve_weekly_profile", lambda request: {"weekly_points": 0, "weekly_rank": None})
+    monkeypatch.setattr(routes, "_resolve_auth_points", lambda request: 0)
+
+    def _bind_identity(request):
+        request.state.auth_user_id = "user-1"
+        request.state.auth_email = "user@example.com"
+
+    calls = []
+
+    def _subscription_window(
+        user_id,
+        respect_requirement=False,
+        bypass_cache=False,
+        unknown_on_error=False,
+    ):
+        calls.append(
+            {
+                "user_id": user_id,
+                "bypass_cache": bypass_cache,
+                "unknown_on_error": unknown_on_error,
+            }
+        )
+        return {
+            "current": {
+                "plan_code": "pro_monthly",
+                "starts_at": "2026-03-22T00:00:00+00:00",
+                "expires_at": "2026-04-21T00:00:00+00:00",
+            },
+            "rows": [],
+        }
+
+    monkeypatch.setattr(routes, "_assert_entitlement", lambda request: None)
+    monkeypatch.setattr(routes, "_bind_optional_supabase_identity", _bind_identity)
+    monkeypatch.setattr(routes.SUPABASE_ENTITLEMENT, "get_subscription_window", _subscription_window)
+
+    response = client.get("/api/auth/me?scope=entitlement")
+
+    assert response.status_code == 200
+    assert calls == [
+        {
+            "user_id": "user-1",
+            "bypass_cache": False,
+            "unknown_on_error": True,
+        }
+    ]
+
+
 def test_auth_me_preserves_unknown_subscription_window(monkeypatch):
    monkeypatch.setattr(web_core.SUPABASE_ENTITLEMENT, "enabled", True)
    monkeypatch.setattr(web_core.SUPABASE_ENTITLEMENT, "require_subscription", False)
@@ -156,7 +156,7 @@ def get_auth_me_payload(request: Request) -> Dict[str, Any]:
                        lambda: legacy_routes.SUPABASE_ENTITLEMENT.get_subscription_window(
                            user_id,
                            respect_requirement=False,
-                            bypass_cache=True,
+                            bypass_cache=False,
                            unknown_on_error=True,
                        ),
                    )
@@ -2,6 +2,7 @@

 from __future__ import annotations

+import os
 import re
 import threading
 import time
@@ -14,6 +15,14 @@ from loguru import logger
 T = TypeVar("T")


+def _slow_stage_threshold_ms() -> float:
+    raw = str(os.getenv("POLYWEATHER_SLOW_STAGE_LOG_MS") or "1500").strip()
+    try:
+        return max(0.0, float(raw))
+    except (TypeError, ValueError):
+        return 1500.0
+
+
 class ServerTimingRecorder:
    def __init__(
        self,
@@ -71,6 +80,20 @@ class ServerTimingRecorder:
            status_code,
            dict(self.timings_ms),
        )
+        threshold = _slow_stage_threshold_ms()
+        slow_stages = {
+            stage: duration
+            for stage, duration in dict(self.timings_ms).items()
+            if duration >= threshold
+        }
+        if slow_stages:
+            logger.warning(
+                "slow_request_timing log_name={} outcome={} status_code={} slow_stages={}",
+                self.log_name,
+                outcome,
+                status_code,
+                slow_stages,
+            )

    def _metric_name(self, stage: str) -> str:
        raw = f"{self.prefix}_{stage}"