集成 Turnstile 人机验证 + Cloudflare R2 事件归档

frontend/components/auth/__tests__/turnstileIntegration.test.ts

@@ -0,0 +1,93 @@
+import fs from "node:fs";
+import path from "node:path";
+
+function assert(condition: unknown, message: string) {
+  if (!condition) throw new Error(message);
+}
+
+export function runTests() {
+  const projectRoot = process.cwd();
+  const turnstileLibPath = path.join(projectRoot, "lib", "turnstile.ts");
+  const widgetPath = path.join(projectRoot, "components", "security", "TurnstileWidget.tsx");
+  const loginPath = path.join(projectRoot, "components", "auth", "LoginClient.tsx");
+  const feedbackModalPath = path.join(
+    projectRoot,
+    "components",
+    "dashboard",
+    "scan-terminal",
+    "UserFeedbackModal.tsx",
+  );
+  const feedbackRoutePath = path.join(projectRoot, "app", "api", "feedback", "route.ts");
+  const paymentIntentRoutePath = path.join(projectRoot, "app", "api", "payments", "intents", "route.ts");
+  const paymentSubmitRoutePath = path.join(
+    projectRoot,
+    "app",
+    "api",
+    "payments",
+    "intents",
+    "[intentId]",
+    "submit",
+    "route.ts",
+  );
+  const paymentFlowPath = path.join(projectRoot, "components", "account", "usePaymentFlow.ts");
+
+  assert(fs.existsSync(turnstileLibPath), "shared Turnstile verifier module must exist");
+  assert(fs.existsSync(widgetPath), "Turnstile widget component must exist");
+
+  const turnstileLib = fs.readFileSync(turnstileLibPath, "utf8");
+  const widgetSource = fs.readFileSync(widgetPath, "utf8");
+  const loginSource = fs.readFileSync(loginPath, "utf8");
+  const feedbackModalSource = fs.readFileSync(feedbackModalPath, "utf8");
+  const feedbackRouteSource = fs.readFileSync(feedbackRoutePath, "utf8");
+  const paymentIntentRouteSource = fs.readFileSync(paymentIntentRoutePath, "utf8");
+  const paymentSubmitRouteSource = fs.readFileSync(paymentSubmitRoutePath, "utf8");
+  const paymentFlowSource = fs.readFileSync(paymentFlowPath, "utf8");
+
+  assert(
+    turnstileLib.includes("POLYWEATHER_TURNSTILE_SECRET_KEY") &&
+      turnstileLib.includes("NEXT_PUBLIC_TURNSTILE_SITE_KEY") &&
+      turnstileLib.includes("https://challenges.cloudflare.com/turnstile/v0/siteverify") &&
+      turnstileLib.includes("requireTurnstileForRequest") &&
+      turnstileLib.includes("POLYWEATHER_TURNSTILE_BYPASS"),
+    "Turnstile verifier must be optional, use Cloudflare siteverify, and expose a request guard",
+  );
+  assert(
+    widgetSource.includes("NEXT_PUBLIC_TURNSTILE_SITE_KEY") &&
+      widgetSource.includes("challenges.cloudflare.com/turnstile/v0/api.js") &&
+      widgetSource.includes("window.turnstile.render") &&
+      widgetSource.includes("data-turnstile-action"),
+    "Turnstile widget must render Cloudflare Turnstile only when a site key is configured",
+  );
+  assert(
+    loginSource.includes("TurnstileWidget") &&
+      loginSource.includes("getTurnstileTokenForAction") &&
+      loginSource.includes("captchaToken") &&
+      loginSource.includes("signInWithPassword") &&
+      loginSource.includes("signUp"),
+    "login/signup must attach a Turnstile captchaToken to Supabase auth calls",
+  );
+  assert(
+    feedbackModalSource.includes("TurnstileWidget") &&
+      feedbackModalSource.includes("getTurnstileTokenForAction") &&
+      feedbackModalSource.includes("turnstile_token"),
+    "feedback modal must include a Turnstile token in feedback submissions",
+  );
+  assert(
+    feedbackRouteSource.includes('requireTurnstileForRequest(req, "feedback_submit"') &&
+      feedbackRouteSource.indexOf('requireTurnstileForRequest(req, "feedback_submit"') <
+        feedbackRouteSource.indexOf("fetch(`${API_BASE}/api/feedback`"),
+    "feedback POST proxy must verify Turnstile before forwarding user feedback",
+  );
+  assert(
+    paymentIntentRouteSource.includes('requireTurnstileForRequest(req, "payment_intent_create"') &&
+      paymentSubmitRouteSource.includes('requireTurnstileForRequest(req, "payment_tx_submit"'),
+    "payment mutation proxies must verify Turnstile before forwarding create and submit requests",
+  );
+  assert(
+    paymentFlowSource.includes("getTurnstileTokenForAction") &&
+      paymentFlowSource.includes("turnstile_token") &&
+      paymentFlowSource.includes("payment_intent_create") &&
+      paymentFlowSource.includes("payment_tx_submit"),
+    "payment flow must include Turnstile tokens in intent creation and tx submit payloads",
+  );
+}

frontend/components/security/TurnstileWidget.tsx

@@ -0,0 +1,128 @@
+"use client";
+
+import { useEffect, useRef, useState } from "react";
+import { getConfiguredTurnstileSiteKey, isTurnstileEnabled } from "@/lib/turnstile-client";
+
+const TURNSTILE_SCRIPT_ID = "polyweather-turnstile-script";
+const TURNSTILE_SCRIPT_SRC = "https://challenges.cloudflare.com/turnstile/v0/api.js?render=explicit";
+const TURNSTILE_SITE_KEY_ENV = "NEXT_PUBLIC_TURNSTILE_SITE_KEY";
+
+type TurnstileRenderOptions = {
+  sitekey: string;
+  action: string;
+  callback: (token: string) => void;
+  "expired-callback": () => void;
+  "error-callback": () => void;
+  theme?: "light" | "dark" | "auto";
+  size?: "normal" | "compact" | "flexible";
+};
+
+declare global {
+  interface Window {
+    turnstile?: {
+      render: (container: HTMLElement, options: TurnstileRenderOptions) => string;
+      reset: (widgetId?: string) => void;
+      remove: (widgetId?: string) => void;
+    };
+  }
+}
+
+function ensureTurnstileScript(onReady: () => void, onError: () => void) {
+  if (typeof window === "undefined") return;
+  if (window.turnstile) {
+    onReady();
+    return;
+  }
+
+  const existing = document.getElementById(TURNSTILE_SCRIPT_ID) as HTMLScriptElement | null;
+  if (existing) {
+    existing.addEventListener("load", onReady, { once: true });
+    existing.addEventListener("error", onError, { once: true });
+    return;
+  }
+
+  const script = document.createElement("script");
+  script.id = TURNSTILE_SCRIPT_ID;
+  script.src = TURNSTILE_SCRIPT_SRC;
+  script.async = true;
+  script.defer = true;
+  script.addEventListener("load", onReady, { once: true });
+  script.addEventListener("error", onError, { once: true });
+  document.head.appendChild(script);
+}
+
+export function TurnstileWidget({
+  action,
+  onToken,
+  onError,
+  resetKey,
+  theme = "light",
+  size = "normal",
+}: {
+  action: string;
+  onToken: (token: string) => void;
+  onError?: () => void;
+  resetKey?: string | number;
+  theme?: "light" | "dark" | "auto";
+  size?: "normal" | "compact" | "flexible";
+}) {
+  const containerRef = useRef<HTMLDivElement | null>(null);
+  const widgetIdRef = useRef<string>("");
+  const [scriptReady, setScriptReady] = useState(false);
+  const siteKey = getConfiguredTurnstileSiteKey();
+
+  useEffect(() => {
+    if (!isTurnstileEnabled()) return;
+    ensureTurnstileScript(
+      () => setScriptReady(true),
+      () => {
+        onToken("");
+        onError?.();
+      },
+    );
+  }, [onError, onToken]);
+
+  useEffect(() => {
+    if (!siteKey || !scriptReady || !containerRef.current || !window.turnstile) return;
+    if (widgetIdRef.current) {
+      window.turnstile.remove(widgetIdRef.current);
+      widgetIdRef.current = "";
+    }
+    widgetIdRef.current = window.turnstile.render(containerRef.current, {
+      sitekey: siteKey,
+      action,
+      theme,
+      size,
+      callback: onToken,
+      "expired-callback": () => onToken(""),
+      "error-callback": () => {
+        onToken("");
+        onError?.();
+      },
+    });
+
+    return () => {
+      if (widgetIdRef.current && window.turnstile) {
+        window.turnstile.remove(widgetIdRef.current);
+        widgetIdRef.current = "";
+      }
+    };
+  }, [action, onError, onToken, scriptReady, siteKey, size, theme]);
+
+  useEffect(() => {
+    if (!resetKey || !widgetIdRef.current || !window.turnstile) return;
+    onToken("");
+    window.turnstile.reset(widgetIdRef.current);
+  }, [onToken, resetKey]);
+
+  if (!siteKey) return null;
+
+  return (
+    <div
+      ref={containerRef}
+      data-turnstile-action={action}
+      data-turnstile-site-key-env={TURNSTILE_SITE_KEY_ENV}
+      className="min-h-[65px] w-full overflow-hidden"
+    />
+  );
+}

frontend/lib/turnstile-client.ts

@@ -0,0 +1,23 @@
+export function getConfiguredTurnstileSiteKey() {
+  return String(process.env.NEXT_PUBLIC_TURNSTILE_SITE_KEY || "").trim();
+}
+
+export function isTurnstileEnabled() {
+  return Boolean(getConfiguredTurnstileSiteKey());
+}
+
+export function getTurnstileTokenForAction(
+  token: string | null | undefined,
+  action: string,
+  options?: { message?: string },
+) {
+  if (!isTurnstileEnabled()) return undefined;
+  const normalized = String(token || "").trim();
+  if (!normalized) {
+    throw new Error(
+      options?.message ||
+        `Complete the security check before continuing (${action}).`,
+    );
+  }
+  return normalized;
+}

frontend/lib/turnstile.ts

@@ -0,0 +1,121 @@
+import { NextRequest, NextResponse } from "next/server";
+
+const TURNSTILE_VERIFY_URL = "https://challenges.cloudflare.com/turnstile/v0/siteverify";
+const NO_STORE = "no-store, max-age=0";
+
+type TurnstileVerification = {
+  ok: boolean;
+  skipped?: boolean;
+  error?: string;
+};
+
+function turnstileSecretKey() {
+  return String(process.env.POLYWEATHER_TURNSTILE_SECRET_KEY || "").trim();
+}
+
+function turnstileSiteKey() {
+  return String(process.env.NEXT_PUBLIC_TURNSTILE_SITE_KEY || "").trim();
+}
+
+export function isTurnstileServerEnabled() {
+  if (process.env.POLYWEATHER_TURNSTILE_BYPASS === "true") return false;
+  return Boolean(turnstileSecretKey() && turnstileSiteKey());
+}
+
+function enforceTurnstileAction() {
+  return process.env.POLYWEATHER_TURNSTILE_ENFORCE_ACTION === "true";
+}
+
+export function stripTurnstileToken<T>(body: T): T {
+  if (!body || typeof body !== "object" || Array.isArray(body)) return body;
+  const copy = { ...(body as Record<string, unknown>) };
+  delete copy.turnstile_token;
+  delete copy.turnstileToken;
+  return copy as T;
+}
+
+function tokenFromBody(body: unknown) {
+  if (!body || typeof body !== "object") return "";
+  const record = body as Record<string, unknown>;
+  return String(record.turnstile_token || record.turnstileToken || "").trim();
+}
+
+function remoteIpFromRequest(req: NextRequest) {
+  const forwardedFor = req.headers.get("x-forwarded-for") || "";
+  return (
+    req.headers.get("cf-connecting-ip") ||
+    forwardedFor.split(",")[0]?.trim() ||
+    req.headers.get("x-real-ip") ||
+    ""
+  );
+}
+
+export async function verifyTurnstileToken(
+  token: string,
+  req: NextRequest,
+  action: string,
+): Promise<TurnstileVerification> {
+  if (!isTurnstileServerEnabled()) {
+    return { ok: true, skipped: true };
+  }
+
+  const normalizedToken = String(token || "").trim();
+  if (!normalizedToken) {
+    return { ok: false, error: "missing_turnstile_token" };
+  }
+
+  const body = new URLSearchParams();
+  body.set("secret", turnstileSecretKey());
+  body.set("response", normalizedToken);
+  const remoteIp = remoteIpFromRequest(req);
+  if (remoteIp) body.set("remoteip", remoteIp);
+
+  try {
+    const response = await fetch(TURNSTILE_VERIFY_URL, {
+      method: "POST",
+      headers: {
+        "Content-Type": "application/x-www-form-urlencoded",
+      },
+      body,
+      cache: "no-store",
+    });
+    const data = (await response.json().catch(() => null)) as {
+      success?: unknown;
+      action?: unknown;
+      "error-codes"?: unknown;
+    } | null;
+    if (!response.ok || data?.success !== true) {
+      return { ok: false, error: "turnstile_siteverify_failed" };
+    }
+    const returnedAction = String(data.action || "").trim();
+    if (enforceTurnstileAction() && returnedAction && returnedAction !== action) {
+      return { ok: false, error: "turnstile_action_mismatch" };
+    }
+    return { ok: true };
+  } catch {
+    return { ok: false, error: "turnstile_siteverify_unavailable" };
+  }
+}
+
+export async function requireTurnstileForRequest(
+  req: NextRequest,
+  action: string,
+  body?: unknown,
+) {
+  const token = tokenFromBody(body);
+  const result = await verifyTurnstileToken(token, req, action);
+  if (result.ok) return null;
+  return NextResponse.json(
+    {
+      error: "Security verification failed. Please refresh the challenge and retry.",
+      code: result.error || "turnstile_failed",
+    },
+    {
+      status: 403,
+      headers: {
+        "Cache-Control": NO_STORE,
+        "Cloudflare-CDN-Cache-Control": NO_STORE,
+      },
+    },
+  );
+}

scripts/archive_realtime_events_to_r2.py

@@ -0,0 +1,152 @@
+"""Archive realtime observation patch events to Cloudflare R2 as JSONL.
+
+This script is intentionally read-only for Redis/SQLite. It uploads a cold
+archive object and does not delete live replay data.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sqlite3
+import sys
+from datetime import date, datetime, timedelta, timezone
+from pathlib import Path
+from typing import Any, Iterable
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def _ensure_repo_path() -> None:
+    if str(ROOT) not in sys.path:
+        sys.path.insert(0, str(ROOT))
+
+
+def _default_sqlite_db_path() -> str:
+    _ensure_repo_path()
+    from src.database.db_manager import DBManager
+
+    return DBManager().db_path
+
+
+def _parse_date(value: str) -> date:
+    return datetime.strptime(value, "%Y-%m-%d").date()
+
+
+def _date_window(day: date) -> tuple[datetime, datetime]:
+    start = datetime(day.year, day.month, day.day, tzinfo=timezone.utc)
+    return start, start + timedelta(days=1)
+
+
+def _decode(value: Any) -> str:
+    if isinstance(value, bytes):
+        return value.decode("utf-8")
+    return str(value or "")
+
+
+def _load_sqlite_events(db_path: str, archive_day: date) -> list[dict[str, Any]]:
+    start, end = _date_window(archive_day)
+    with sqlite3.connect(db_path) as conn:
+        conn.row_factory = sqlite3.Row
+        rows = conn.execute(
+            """
+            SELECT revision, schema_type, schema_version, city, source, obs_time,
+                   payload_json, created_at
+            FROM observation_patch_events
+            WHERE created_at >= ? AND created_at < ?
+            ORDER BY revision ASC
+            """,
+            (start.isoformat(), end.isoformat()),
+        ).fetchall()
+    return [
+        {
+            "revision": int(row["revision"]),
+            "schema_type": row["schema_type"],
+            "schema_version": int(row["schema_version"]),
+            "type": f"{row['schema_type']}.v{int(row['schema_version'])}",
+            "city": row["city"],
+            "source": row["source"],
+            "obs_time": row["obs_time"],
+            "created_at": row["created_at"],
+            "payload": json.loads(row["payload_json"]),
+        }
+        for row in rows
+    ]
+
+
+def _load_redis_events(redis_url: str, stream_key: str, archive_day: date) -> list[dict[str, Any]]:
+    import redis  # type: ignore
+
+    start, end = _date_window(archive_day)
+    min_id = f"{int(start.timestamp() * 1000)}-0"
+    max_id = f"{int(end.timestamp() * 1000) - 1}-999999"
+    client = redis.Redis.from_url(redis_url)
+    rows = client.xrange(stream_key, min=min_id, max=max_id)
+    events = []
+    for stream_id, fields in rows:
+        decoded = {_decode(key): _decode(value) for key, value in dict(fields).items()}
+        payload_json = decoded.get("payload_json") or "{}"
+        events.append(
+            {
+                "stream_id": _decode(stream_id),
+                "revision": int(decoded.get("revision") or 0),
+                "type": decoded.get("type") or "",
+                "schema_type": decoded.get("schema_type") or "",
+                "schema_version": int(decoded.get("schema_version") or 0),
+                "city": decoded.get("city") or "",
+                "source": decoded.get("source") or "",
+                "obs_time": decoded.get("obs_time") or "",
+                "created_at_ms": int(decoded.get("created_at_ms") or 0),
+                "ts": int(decoded.get("ts") or 0),
+                "producer_id": decoded.get("producer_id") or "",
+                "payload": json.loads(payload_json),
+            }
+        )
+    return events
+
+
+def _build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--source", choices=["redis", "sqlite"], default=os.getenv("POLYWEATHER_R2_ARCHIVE_SOURCE", "redis"))
+    parser.add_argument("--date", default=(datetime.now(timezone.utc).date() - timedelta(days=1)).isoformat())
+    parser.add_argument("--sqlite-db", default=os.getenv("POLYWEATHER_SQLITE_DB_PATH", ""))
+    parser.add_argument("--redis-url", default=os.getenv("POLYWEATHER_REDIS_URL", "redis://127.0.0.1:6379/0"))
+    parser.add_argument("--redis-stream-key", default=os.getenv("POLYWEATHER_REDIS_STREAM_KEY", "stream:city_observation"))
+    parser.add_argument("--dry-run", action="store_true")
+    return parser
+
+
+def main(argv: Iterable[str] | None = None) -> int:
+    args = _build_parser().parse_args(list(argv) if argv is not None else None)
+    _ensure_repo_path()
+    from web.r2_archive import (
+        R2ArchiveConfig,
+        R2ObjectStore,
+        build_realtime_archive_key,
+        events_to_jsonl,
+    )
+
+    archive_day = _parse_date(args.date)
+    if args.source == "sqlite":
+        db_path = args.sqlite_db or _default_sqlite_db_path()
+        events = _load_sqlite_events(db_path, archive_day)
+    else:
+        events = _load_redis_events(args.redis_url, args.redis_stream_key, archive_day)
+
+    body = events_to_jsonl(events)
+    key = build_realtime_archive_key(args.source, archive_day.isoformat())
+    print(json.dumps({"source": args.source, "date": archive_day.isoformat(), "events": len(events), "key": key}, indent=2))
+    if args.dry_run:
+        return 0
+    config = R2ArchiveConfig.from_env()
+    if config is None:
+        print("R2 env is not configured; set POLYWEATHER_R2_* variables", file=sys.stderr)
+        return 2
+    result = R2ObjectStore(config).put_object(key, body, content_type="application/x-ndjson")
+    print(json.dumps(result, indent=2))
+    return 0 if result.get("ok") else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_cloudflare_cache_validation.py

@@ -0,0 +1,17 @@
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def test_frontend_cache_validator_checks_cloudflare_edge_hits():
+    script = (ROOT / "scripts" / "validate_frontend_cache.sh").read_text(encoding="utf-8")
+
+    assert "REQUIRE_CF_CACHE" in script
+    assert "CF-Cache-Status" in script
+    assert "cf_cache_status" in script
+    assert "HIT" in script
+    assert "MISS" in script
+    assert "REVALIDATED" in script
+    assert 'check_cloudflare_cache_hit "/api/cities" "cities edge cache"' in script
+    assert 'check_cloudflare_cache_hit "/api/scan/terminal?limit=1" "scan terminal edge cache"' in script

tests/test_r2_archive.py

@@ -0,0 +1,87 @@
+import json
+
+from web.r2_archive import (
+    R2ArchiveConfig,
+    R2ObjectStore,
+    build_realtime_archive_key,
+    events_to_jsonl,
+)
+
+
+def test_r2_config_is_disabled_until_all_required_env_is_present(monkeypatch):
+    for key in (
+        "POLYWEATHER_R2_ACCOUNT_ID",
+        "POLYWEATHER_R2_BUCKET",
+        "POLYWEATHER_R2_ACCESS_KEY_ID",
+        "POLYWEATHER_R2_SECRET_ACCESS_KEY",
+    ):
+        monkeypatch.delenv(key, raising=False)
+
+    assert R2ArchiveConfig.from_env() is None
+
+
+def test_r2_put_object_signs_cloudflare_r2_request():
+    captured = {}
+
+    def fake_transport(request, timeout):
+        captured["url"] = request.full_url
+        captured["method"] = request.get_method()
+        captured["headers"] = dict(request.header_items())
+        captured["body"] = request.data
+
+        class Response:
+            status = 200
+
+            def read(self):
+                return b""
+
+            def __enter__(self):
+                return self
+
+            def __exit__(self, *_args):
+                return False
+
+        return Response()
+
+    config = R2ArchiveConfig(
+        account_id="acct123",
+        bucket="polyweather-archive",
+        access_key_id="access",
+        secret_access_key="secret",
+    )
+    store = R2ObjectStore(config, transport=fake_transport)
+
+    store.put_object(
+        "sse-events/2026/06/16/test.jsonl",
+        b'{"ok":true}\n',
+        content_type="application/x-ndjson",
+    )
+
+    assert captured["method"] == "PUT"
+    assert captured["url"] == (
+        "https://polyweather-archive.acct123.r2.cloudflarestorage.com/"
+        "sse-events/2026/06/16/test.jsonl"
+    )
+    assert captured["body"] == b'{"ok":true}\n'
+    assert captured["headers"]["Content-type"] == "application/x-ndjson"
+    assert captured["headers"]["Host"] == "polyweather-archive.acct123.r2.cloudflarestorage.com"
+    assert captured["headers"]["Authorization"].startswith("AWS4-HMAC-SHA256 ")
+    assert "Credential=access/" in captured["headers"]["Authorization"]
+
+
+def test_realtime_archive_key_and_jsonl_are_stable():
+    key = build_realtime_archive_key("redis", "2026-06-16", "city_observation_patch")
+
+    assert key == "sse-events/2026/06/16/city_observation_patch.redis.jsonl"
+
+    body = events_to_jsonl(
+        [
+            {"revision": 2, "city": "taipei", "payload": {"temp_c": 31.2}},
+            {"revision": 3, "city": "seoul", "payload": {"temp_c": 24.5}},
+        ]
+    )
+
+    lines = body.decode("utf-8").splitlines()
+    assert len(lines) == 2
+    assert json.loads(lines[0])["revision"] == 2
+    assert json.loads(lines[1])["city"] == "seoul"

web/r2_archive.py

@@ -0,0 +1,165 @@
+"""Cloudflare R2 archive helpers for cold PolyWeather event snapshots."""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+import os
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Any, Callable, Iterable, Optional
+from urllib.parse import quote, urlparse
+from urllib.request import Request, urlopen
+
+
+R2_REGION = "auto"
+R2_SERVICE = "s3"
+
+
+@dataclass(frozen=True)
+class R2ArchiveConfig:
+    account_id: str
+    bucket: str
+    access_key_id: str
+    secret_access_key: str
+    endpoint_url: str = ""
+    region: str = R2_REGION
+
+    @classmethod
+    def from_env(cls, env: Optional[dict[str, str]] = None) -> Optional["R2ArchiveConfig"]:
+        source = env if env is not None else os.environ
+        account_id = str(source.get("POLYWEATHER_R2_ACCOUNT_ID") or "").strip()
+        bucket = str(source.get("POLYWEATHER_R2_BUCKET") or "").strip()
+        access_key_id = str(source.get("POLYWEATHER_R2_ACCESS_KEY_ID") or "").strip()
+        secret_access_key = str(source.get("POLYWEATHER_R2_SECRET_ACCESS_KEY") or "").strip()
+        if not all([account_id, bucket, access_key_id, secret_access_key]):
+            return None
+        return cls(
+            account_id=account_id,
+            bucket=bucket,
+            access_key_id=access_key_id,
+            secret_access_key=secret_access_key,
+            endpoint_url=str(source.get("POLYWEATHER_R2_ENDPOINT_URL") or "").strip(),
+            region=str(source.get("POLYWEATHER_R2_REGION") or R2_REGION).strip() or R2_REGION,
+        )
+
+    @property
+    def base_url(self) -> str:
+        if self.endpoint_url:
+            return self.endpoint_url.rstrip("/")
+        return f"https://{self.bucket}.{self.account_id}.r2.cloudflarestorage.com"
+
+
+Transport = Callable[[Request, float], Any]
+
+
+def _default_transport(request: Request, timeout: float) -> Any:
+    return urlopen(request, timeout=timeout)
+
+
+def _sha256_hex(data: bytes) -> str:
+    return hashlib.sha256(data).hexdigest()
+
+
+def _sign(key: bytes, message: str) -> bytes:
+    return hmac.new(key, message.encode("utf-8"), hashlib.sha256).digest()
+
+
+def _signature_key(secret: str, date_stamp: str, region: str) -> bytes:
+    k_date = _sign(("AWS4" + secret).encode("utf-8"), date_stamp)
+    k_region = _sign(k_date, region)
+    k_service = _sign(k_region, R2_SERVICE)
+    return _sign(k_service, "aws4_request")
+
+
+class R2ObjectStore:
+    def __init__(
+        self,
+        config: R2ArchiveConfig,
+        *,
+        transport: Transport = _default_transport,
+        timeout_sec: float = 30.0,
+    ) -> None:
+        self.config = config
+        self.transport = transport
+        self.timeout_sec = max(1.0, float(timeout_sec))
+
+    def put_object(self, key: str, body: bytes, *, content_type: str = "application/octet-stream") -> dict[str, Any]:
+        object_key = quote(str(key).lstrip("/"), safe="/")
+        url = f"{self.config.base_url}/{object_key}"
+        parsed = urlparse(url)
+        host = parsed.netloc
+        payload_hash = _sha256_hex(body)
+        now = datetime.now(timezone.utc)
+        amz_date = now.strftime("%Y%m%dT%H%M%SZ")
+        date_stamp = now.strftime("%Y%m%d")
+        canonical_uri = parsed.path or "/"
+        canonical_headers = (
+            f"content-type:{content_type}\n"
+            f"host:{host}\n"
+            f"x-amz-content-sha256:{payload_hash}\n"
+            f"x-amz-date:{amz_date}\n"
+        )
+        signed_headers = "content-type;host;x-amz-content-sha256;x-amz-date"
+        canonical_request = "\n".join(
+            [
+                "PUT",
+                canonical_uri,
+                "",
+                canonical_headers,
+                signed_headers,
+                payload_hash,
+            ]
+        )
+        credential_scope = f"{date_stamp}/{self.config.region}/{R2_SERVICE}/aws4_request"
+        string_to_sign = "\n".join(
+            [
+                "AWS4-HMAC-SHA256",
+                amz_date,
+                credential_scope,
+                _sha256_hex(canonical_request.encode("utf-8")),
+            ]
+        )
+        signature = hmac.new(
+            _signature_key(self.config.secret_access_key, date_stamp, self.config.region),
+            string_to_sign.encode("utf-8"),
+            hashlib.sha256,
+        ).hexdigest()
+        authorization = (
+            "AWS4-HMAC-SHA256 "
+            f"Credential={self.config.access_key_id}/{credential_scope}, "
+            f"SignedHeaders={signed_headers}, "
+            f"Signature={signature}"
+        )
+        request = Request(url, data=body, method="PUT")
+        request.add_header("Content-Type", content_type)
+        request.add_header("Host", host)
+        request.add_header("X-Amz-Content-Sha256", payload_hash)
+        request.add_header("X-Amz-Date", amz_date)
+        request.add_header("Authorization", authorization)
+
+        with self.transport(request, self.timeout_sec) as response:
+            response.read()
+            status = int(getattr(response, "status", 200) or 200)
+        return {"ok": 200 <= status < 300, "status": status, "key": key}
+
+
+def build_realtime_archive_key(source: str, archive_date: str, stream_name: str = "city_observation_patch") -> str:
+    parts = str(archive_date).split("-")
+    if len(parts) != 3:
+        raise ValueError("archive_date must be YYYY-MM-DD")
+    year, month, day = parts
+    safe_source = quote(str(source or "unknown").strip().lower() or "unknown", safe="")
+    safe_stream = quote(str(stream_name or "events").strip().lower() or "events", safe="")
+    return f"sse-events/{year}/{month}/{day}/{safe_stream}.{safe_source}.jsonl"
+
+
+def events_to_jsonl(events: Iterable[dict[str, Any]]) -> bytes:
+    lines = [
+        json.dumps(event, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+        for event in events
+    ]
+    if not lines:
+        return b""
+    return ("\n".join(lines) + "\n").encode("utf-8")