diff --git a/app/gitea_client.py b/app/gitea_client.py index a3b2c40..ae0bd5e 100644 --- a/app/gitea_client.py +++ b/app/gitea_client.py @@ -64,6 +64,10 @@ class GiteaClient: self._cache: dict[str, tuple[float, Any]] = {} self._refreshing: set[str] = set() self._cache_gen = 0 + # Snapshot bookkeeping (chantier B): cache key -> UTC time + # the payload was fetched from Gitea, for the freshness + # label and the persisted `snapshots` table. + self._snapshot_times: dict[str, str] = {} async def aclose(self) -> None: await self.client.aclose() @@ -112,6 +116,90 @@ class GiteaClient: def _cache_put(self, key: str, value: Any) -> None: if self.cache_ttl > 0: self._cache[key] = (time.monotonic(), value) + self._persist_snapshot(key, value) + + # ── Snapshots (chantier B, 0.22.0) ────────────────────── + + def _persist_snapshot(self, key: str, value: Any) -> None: + """Write-through to the SQLite snapshots table. + + Persistence must never break a read: any failure (DB + locked, unserializable payload) is logged and swallowed. + """ + try: + from app import models + + models.save_snapshot(key, value) + import datetime + + self._snapshot_times[key] = datetime.datetime.now( + datetime.timezone.utc + ).strftime("%Y-%m-%d %H:%M:%S") + except Exception as exc: + logger.debug("snapshot persist failed for %s: %s", key, exc) + + def hydrate_from_snapshots(self) -> int: + """Preload the cache from persisted snapshots (startup). + + Entries are inserted with an already-stale timestamp: the + SWR layer serves them instantly AND schedules a background + refresh, so the first page after a restart renders from + local data (tens of ms) while Gitea data catches up. + Returns the number of entries hydrated. + """ + if self.cache_ttl <= 0: + return 0 + try: + from app import models + + snapshots = models.load_snapshots() + except Exception as exc: + logger.warning("snapshot hydration failed: %s", exc) + return 0 + stale_at = time.monotonic() - self.cache_ttl - 1.0 + for key, (payload, updated_at) in snapshots.items(): + self._cache[key] = (stale_at, payload) + self._snapshot_times[key] = updated_at + if snapshots: + logger.info("hydrated %d snapshot(s) from SQLite", len(snapshots)) + return len(snapshots) + + def latest_snapshot_at(self) -> str | None: + """Freshest Gitea fetch time across tracked keys (UTC str).""" + if not self._snapshot_times: + return None + return max(self._snapshot_times.values()) + + async def _force(self, key: str, fetcher: Callable[[], Awaitable[Any]]) -> Any: + """Fetch now, bypassing freshness (background refresher).""" + value = await fetcher() + self._cache_put(key, value) + return value + + async def refresh_starred(self) -> list[dict]: + return await self._force( + "user:starred", lambda: self._get_all_pages("/user/starred") + ) + + async def refresh_repo_issues( + self, owner: str, repo: str, state: str = "all" + ) -> list[dict]: + return await self._force( + f"issues:{owner}/{repo}:{state}", + lambda: self._get_all_pages( + f"/repos/{owner}/{repo}/issues", state=state + ), + ) + + async def refresh_milestones( + self, owner: str, repo: str + ) -> list[dict]: + return await self._force( + f"milestones:{owner}/{repo}:open", + lambda: self._get_all_pages( + f"/repos/{owner}/{repo}/milestones", state="open" + ), + ) def _cache_clear(self) -> None: self._cache.clear() @@ -182,22 +270,25 @@ class GiteaClient: self._user = await self._get("/user") return self._user + async def _fetch_search_repos(self) -> list[dict]: + params: dict[str, Any] = {} + try: + user = await self.get_user() + if user and user.get("id"): + params["uid"] = user["id"] + except Exception as exc: # no/invalid token → public search + logger.debug( + "get_user failed, falling back to public search: %s", exc + ) + return await self._get_all_pages("/repos/search", **params) + async def search_repos(self) -> list[dict]: """All repos visible to the token owner (paginated, SWR-cached).""" + return await self._cached_list("search_repos", self._fetch_search_repos) - async def _fetch() -> list[dict]: - params: dict[str, Any] = {} - try: - user = await self.get_user() - if user and user.get("id"): - params["uid"] = user["id"] - except Exception as exc: # no/invalid token → public search - logger.debug( - "get_user failed, falling back to public search: %s", exc - ) - return await self._get_all_pages("/repos/search", **params) - - return await self._cached_list("search_repos", _fetch) + async def refresh_search_repos(self) -> list[dict]: + """Force-fetch the repo index (background refresher).""" + return await self._force("search_repos", self._fetch_search_repos) async def get_repo(self, owner: str, repo: str) -> dict: return await self._get(f"/repos/{owner}/{repo}") diff --git a/app/models.py b/app/models.py index 642f778..97cb052 100644 --- a/app/models.py +++ b/app/models.py @@ -11,6 +11,7 @@ and upgrades older databases in place (position column, notes updated_at, cleanup of rows carrying an unknown column name). """ +import json import os import sqlite3 from contextlib import contextmanager @@ -91,6 +92,12 @@ def init_db() -> None: value TEXT NOT NULL ); + CREATE TABLE IF NOT EXISTS snapshots ( + key TEXT PRIMARY KEY, + payload TEXT NOT NULL, + updated_at TEXT NOT NULL DEFAULT (datetime('now')) + ); + CREATE INDEX IF NOT EXISTS idx_events_repo ON board_events(repo_full_name, issue_number, created_at); """ @@ -292,6 +299,48 @@ def set_kv(key: str, value: str) -> None: ) +# ── Snapshots (chantier B, 0.22.0) ───────────────────────── +# Persisted copies of the Gitea list responses the pages read +# (repos, issues, milestones, starred). The client cache is +# hydrated from these rows at startup, so even the first page +# after a restart renders from local data while the background +# refresher re-fetches from Gitea. + +def save_snapshot(key: str, payload) -> None: + with get_db() as conn: + conn.execute( + "INSERT INTO snapshots (key, payload, updated_at) " + "VALUES (?, ?, datetime('now')) " + "ON CONFLICT(key) DO UPDATE SET payload = excluded.payload, " + "updated_at = excluded.updated_at", + (key, json.dumps(payload)), + ) + + +def load_snapshots() -> dict: + """All snapshots as {key: (payload, updated_at_utc_str)}.""" + with get_db() as conn: + rows = conn.execute( + "SELECT key, payload, updated_at FROM snapshots" + ).fetchall() + out = {} + for row in rows: + try: + out[row["key"]] = (json.loads(row["payload"]), row["updated_at"]) + except (ValueError, TypeError): + continue + return out + + +def latest_snapshot_at() -> str | None: + """UTC timestamp (SQLite format) of the freshest snapshot.""" + with get_db() as conn: + row = conn.execute( + "SELECT MAX(updated_at) AS latest FROM snapshots" + ).fetchone() + return row["latest"] if row else None + + # ── Event journal ────────────────────────────────────────── EVENT_TYPES = ("created", "moved", "closed", "reopened", "commented", "rule") diff --git a/app/snapshots.py b/app/snapshots.py new file mode 100644 index 0000000..c36aa0a --- /dev/null +++ b/app/snapshots.py @@ -0,0 +1,116 @@ +"""Background snapshot refresher — chantier B (0.22.0). + +The expensive part of Kanbante is the fan-out of Gitea list +calls behind the portfolio pages (repo index, then issues and +milestones for every active project). Since 0.19.0 those calls +are parallel and stale-while-revalidate, which fixed repeat +loads, but the FIRST page after a restart still paid seconds. + +The refresher pays that cost instead of the user: every +KANBANTE_SNAPSHOT_INTERVAL seconds it force-refreshes the hot +keys (repo index, starred, issues + milestones of active +repos). Every fetch write-throughs to the SQLite `snapshots` +table, and the client cache is hydrated from that table at +startup — so pages read local data from the very first request +and only ever wait on Gitea when no snapshot exists yet. + +Freshness contract, shown in the UI: Gitea-side data may lag +by up to one interval; a Gitea webhook refreshes the touched +repo immediately (see refresh_repo_snapshots); anything the +user does locally (board journal) is always instant. +""" + +import asyncio +import datetime +import logging + +from app import models +from app.config import get_settings + +logger = logging.getLogger("kanbante.snapshots") + +_REFRESH_CONCURRENCY = 8 + + +async def refresh_repo_snapshots(gitea, owner: str, repo: str) -> None: + """Refresh one repo's hot keys (webhooks call this). + + Both issue states are refreshed: most pages read "all", but + Focus/Triage/Mes issues read "open" — a snapshot set missing + either leaves a slow first hit on the table. + """ + try: + await gitea.refresh_repo_issues(owner, repo, "all") + await gitea.refresh_repo_issues(owner, repo, "open") + await gitea.refresh_milestones(owner, repo) + except Exception as exc: + logger.debug("repo snapshot refresh failed for %s/%s: %s", owner, repo, exc) + + +async def refresh_snapshots(gitea, classify_repos) -> int: + """One full refresh cycle. Returns the active-repo count.""" + repos = await gitea.refresh_search_repos() + try: + await gitea.refresh_starred() + except Exception as exc: + logger.debug("starred refresh failed: %s", exc) + active, _inactive, _archived = classify_repos(repos) + semaphore = asyncio.Semaphore(_REFRESH_CONCURRENCY) + + async def _one(repo: dict) -> None: + owner, _, name = repo["full_name"].partition("/") + async with semaphore: + await refresh_repo_snapshots(gitea, owner, name) + + await asyncio.gather(*(_one(repo) for repo in active)) + logger.info("snapshot refresh complete (%d active repos)", len(active)) + return len(active) + + +async def snapshot_loop(gitea, classify_repos) -> None: + """Periodic refresher, started by the app lifespan. + + Sleeps briefly first: hydration already serves the pages, + the loop only has to keep the snapshots warm. + """ + interval = get_settings().snapshot_interval + if interval <= 0: + logger.info("snapshot refresher disabled (interval <= 0)") + return + await asyncio.sleep(10) + while True: + try: + await refresh_snapshots(gitea, classify_repos) + except Exception as exc: + logger.warning("snapshot refresh cycle failed: %s", exc) + await asyncio.sleep(interval) + + +def freshness_label() -> str | None: + """« à l'instant » / « il y a N min » from the freshest snapshot. + + Honest freshness display (chantier B contract): the sidebar + shows how old the Gitea data on screen can be. + """ + try: + latest = models.latest_snapshot_at() + except Exception: + return None + if not latest: + return None + try: + fetched = datetime.datetime.strptime( + latest, "%Y-%m-%d %H:%M:%S" + ).replace(tzinfo=datetime.timezone.utc) + except ValueError: + return None + age = ( + datetime.datetime.now(datetime.timezone.utc) - fetched + ).total_seconds() + if age < 90: + return "à l'instant" + if age < 3600: + return f"il y a {int(age // 60)} min" + if age < 86400: + return f"il y a {int(age // 3600)} h" + return f"il y a {int(age // 86400)} j"