diff --git a/frontend/components/auth/__tests__/turnstileIntegration.test.ts b/frontend/components/auth/__tests__/turnstileIntegration.test.ts new file mode 100644 index 00000000..d31a1832 --- /dev/null +++ b/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", + ); +} diff --git a/frontend/components/security/TurnstileWidget.tsx b/frontend/components/security/TurnstileWidget.tsx new file mode 100644 index 00000000..5552167a --- /dev/null +++ b/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(null); + const widgetIdRef = useRef(""); + 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 ( +
+ ); +} diff --git a/frontend/lib/turnstile-client.ts b/frontend/lib/turnstile-client.ts new file mode 100644 index 00000000..5a801f39 --- /dev/null +++ b/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; +} diff --git a/frontend/lib/turnstile.ts b/frontend/lib/turnstile.ts new file mode 100644 index 00000000..ed90b956 --- /dev/null +++ b/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(body: T): T { + if (!body || typeof body !== "object" || Array.isArray(body)) return body; + const copy = { ...(body as Record) }; + 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; + 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 { + 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, + }, + }, + ); +} diff --git a/scripts/archive_realtime_events_to_r2.py b/scripts/archive_realtime_events_to_r2.py new file mode 100644 index 00000000..be787e13 --- /dev/null +++ b/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()) diff --git a/tests/test_cloudflare_cache_validation.py b/tests/test_cloudflare_cache_validation.py new file mode 100644 index 00000000..4e10acce --- /dev/null +++ b/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 diff --git a/tests/test_r2_archive.py b/tests/test_r2_archive.py new file mode 100644 index 00000000..70b65040 --- /dev/null +++ b/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" diff --git a/web/r2_archive.py b/web/r2_archive.py new file mode 100644 index 00000000..52418b5d --- /dev/null +++ b/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")