fix(live-search): add global endpoint for trending

Author: kshitijrajsharma

api/app.py

@@ -12,6 +12,7 @@
 
 from .db import close_pool, ensure_schema, open_pool
 from .queries import fetch_state
+from .routers.global_stats import global_router
 from .routers.hashtag import v2_router
 from .schemas import HealthResponse
 
@@ -95,7 +96,7 @@ async def health() -> HealthResponse:
 
 
 app = Litestar(
-    route_handlers=[*_root_handlers(), health, v2_router],
+    route_handlers=[*_root_handlers(), health, v2_router, global_router],
     lifespan=[lifespan],
     middleware=[rate_limit_config.middleware],
     cors_config=CORSConfig(

api/duck.py

@@ -6,6 +6,7 @@
 from __future__ import annotations
 
 import asyncio
+import collections
 import contextlib
 import datetime as dt
 import logging
@@ -44,6 +45,9 @@
 _HASHTAG_CHANGESET = os.getenv("OSMSG_HASHTAG_CHANGESET", f"{_ROLLUP}/hashtag_changeset/data.parquet")
 _USERS = os.getenv("OSMSG_USERS", f"{_ROLLUP}/users/data.parquet")
 _FRONTIER_TTL_SECONDS = int(os.getenv("OSMSG_FRONTIER_TTL_SECONDS", "3600"))
+# work_mem for the API's own pooled Postgres connections only (not the worker, not the db global config), so
+# the heaviest global aggregate stays mostly in memory instead of spilling. Sized to the pool + db cap.
+_PG_WORK_MEM = os.getenv("OSMSG_PG_WORK_MEM", "64MB")
 
 _frontier_cache: tuple[float, dt.datetime] | None = None
 
@@ -58,7 +62,8 @@ def _libpq_dsn() -> str:
         "user": u.username,
         "password": u.password,
     }
-    return " ".join(f"{k}={v}" for k, v in parts.items() if v is not None)
+    dsn = " ".join(f"{k}={v}" for k, v in parts.items() if v is not None)
+    return f"{dsn} options='-c work_mem={_PG_WORK_MEM}'"
 
 
 def _frontier() -> dt.datetime:
@@ -81,7 +86,7 @@ def _connect() -> duckdb.DuckDBPyConnection:
     con.execute("SET http_retries=10;")
     # Memory/temp pragmas so concurrent pooled queries cannot sum past the container memory cap.
     _apply_runtime_pragmas(con)
-    con.execute(f"ATTACH '{_libpq_dsn()}' AS pg (TYPE postgres, READ_ONLY)")
+    con.execute(f"ATTACH '{_libpq_dsn().replace(chr(39), chr(39) * 2)}' AS pg (TYPE postgres, READ_ONLY)")
     return con
 
 
@@ -279,3 +284,109 @@ async def map_points(
     hashtag: str | list[str], *, limit: int = 2000, start: dt.datetime | None = None, end: dt.datetime | None = None
 ):
     return await asyncio.to_thread(_run, query.map_points, hashtag, limit=limit, start=start, end=end)
+
+
+def _run_global(fn, **kwargs):
+    """Like _run but for the no-hashtag global endpoints: runs fn(con, _sources(), **kwargs) under the
+    watchdog on a pooled connection. No warm path (global windows are recent, uncached)."""
+    pool = _pool_ready()
+    con = _acquire(pool)
+    done = threading.Event()
+    interrupted = False
+
+    def _watchdog() -> None:
+        nonlocal interrupted
+        if not done.wait(_QUERY_TIMEOUT):
+            interrupted = True
+            with contextlib.suppress(duckdb.Error):
+                con.interrupt()
+
+    watcher = threading.Thread(target=_watchdog, daemon=True)
+    watcher.start()
+    healthy = True
+    try:
+        return fn(con, _sources(), **kwargs)
+    except duckdb.Error:
+        healthy = False
+        if interrupted:
+            raise HTTPException(status_code=503, detail="Server is busy, please try again in a moment.") from None
+        raise
+    finally:
+        done.set()
+        watcher.join()
+        if healthy:
+            pool.put(con)
+        else:
+            with contextlib.suppress(duckdb.Error):
+                con.close()
+            pool.put(_connect())
+
+
+# Memoize whole-OSM results by a grain-rounded window so repeat hits are instant.
+_GLOBAL_CACHE_TTL = float(os.getenv("OSMSG_GLOBAL_CACHE_TTL", "120"))
+_GLOBAL_GRAIN = 60
+_global_cache: collections.OrderedDict[tuple, tuple[float, object]] = collections.OrderedDict()
+_global_cache_lock = threading.Lock()
+
+
+def _round_window(start: dt.datetime, end: dt.datetime) -> tuple[dt.datetime, dt.datetime]:
+    def floor(t: dt.datetime) -> dt.datetime:
+        return t.replace(second=(t.second // _GLOBAL_GRAIN) * _GLOBAL_GRAIN, microsecond=0)
+
+    return floor(start), floor(end)
+
+
+async def _global_cached(fn, key_extra, start, end, **kwargs):
+    rs, re = _round_window(start, end)
+    key = (fn.__name__, rs, re, key_extra)
+    now = time.monotonic()
+    with _global_cache_lock:
+        hit = _global_cache.get(key)
+        if hit and hit[0] > now:
+            _global_cache.move_to_end(key)
+            return hit[1]
+    result = await asyncio.to_thread(_run_global, fn, start=rs, end=re, **kwargs)
+    with _global_cache_lock:
+        _global_cache[key] = (now + _GLOBAL_CACHE_TTL, result)
+        while len(_global_cache) > 512:
+            _global_cache.popitem(last=False)
+    return result
+
+
+async def global_summary(*, start: dt.datetime, end: dt.datetime):
+    return await _global_cached(query.global_summary, None, start, end)
+
+
+async def global_leaderboard(
+    *,
+    start: dt.datetime,
+    end: dt.datetime,
+    page: int = 1,
+    page_size: int = query.DEFAULT_PAGE_SIZE,
+    sort: str = "map_changes",
+    order: str = "desc",
+    q: str | None = None,
+):
+    return await _global_cached(
+        query.global_leaderboard,
+        (page, page_size, sort, order, q),
+        start,
+        end,
+        page=page,
+        page_size=page_size,
+        sort=sort,
+        order=order,
+        q=q,
+    )
+
+
+async def global_editors(*, start: dt.datetime, end: dt.datetime):
+    return await _global_cached(query.global_editors, None, start, end)
+
+
+async def global_tags(*, start: dt.datetime, end: dt.datetime, limit: int = 100):
+    return await _global_cached(query.global_tags, limit, start, end, limit=limit)
+
+
+async def global_trending(*, start: dt.datetime, end: dt.datetime, limit: int = 15):
+    return await _global_cached(query.global_trending, limit, start, end, limit=limit)

api/routers/global_stats.py

@@ -0,0 +1,94 @@
+"""Whole-OSM (no-hashtag) stats over a recent window, capped at GLOBAL_MAX_DAYS."""
+
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+from litestar import Controller, Router, get
+from litestar.exceptions import HTTPException
+
+from osmsg.query import GLOBAL_MAX_DAYS
+from osmsg.query import LEADERBOARD_SORTS as _LEADERBOARD_SORTS
+
+from .. import duck
+
+_WINDOWS = {
+    "1h": timedelta(hours=1),
+    "24h": timedelta(hours=24),
+    "7d": timedelta(days=7),
+}
+_MAX = timedelta(days=GLOBAL_MAX_DAYS)
+
+
+def _resolve(window: str | None, start: datetime | None, end: datetime | None) -> tuple[datetime, datetime]:
+    if window is not None:
+        if window not in _WINDOWS:
+            raise HTTPException(status_code=400, detail=f"window must be one of {', '.join(_WINDOWS)}")
+        now = datetime.now(UTC)
+        return now - _WINDOWS[window], now
+    if start is None or end is None:
+        raise HTTPException(status_code=400, detail="provide window (1h|24h|7d) or both start and end")
+    if start >= end:
+        raise HTTPException(status_code=400, detail="start must be before end")
+    if end - start > _MAX:
+        raise HTTPException(status_code=400, detail=f"global window cannot exceed {GLOBAL_MAX_DAYS} days")
+    if start < datetime.now(UTC) - _MAX - timedelta(days=2):
+        raise HTTPException(status_code=400, detail=f"global stats cover only the last {GLOBAL_MAX_DAYS} days")
+    return start, end
+
+
+class GlobalController(Controller):
+    path = "/global"
+
+    @get("/summary")
+    async def get_summary(
+        self, window: str | None = None, start: datetime | None = None, end: datetime | None = None
+    ) -> dict[str, Any]:
+        s, e = _resolve(window, start, end)
+        return await duck.global_summary(start=s, end=e)
+
+    @get("/leaderboard")
+    async def get_leaderboard(
+        self,
+        window: str | None = None,
+        start: datetime | None = None,
+        end: datetime | None = None,
+        page: int = 1,
+        page_size: int = 25,
+        sort: str = "map_changes",
+        order: str = "desc",
+        q: str | None = None,
+    ) -> dict[str, Any]:
+        if sort not in _LEADERBOARD_SORTS:
+            raise HTTPException(status_code=400, detail=f"sort must be one of {', '.join(_LEADERBOARD_SORTS)}")
+        if order not in ("asc", "desc"):
+            raise HTTPException(status_code=400, detail="order must be 'asc' or 'desc'")
+        if page < 1:
+            raise HTTPException(status_code=400, detail="page must be >= 1")
+        s, e = _resolve(window, start, end)
+        return await duck.global_leaderboard(
+            start=s, end=e, page=page, page_size=page_size, sort=sort, order=order, q=q
+        )
+
+    @get("/editors")
+    async def get_editors(
+        self, window: str | None = None, start: datetime | None = None, end: datetime | None = None
+    ) -> list[dict[str, Any]]:
+        s, e = _resolve(window, start, end)
+        return await duck.global_editors(start=s, end=e)
+
+    @get("/tags")
+    async def get_tags(
+        self, window: str | None = None, start: datetime | None = None, end: datetime | None = None, limit: int = 100
+    ) -> list[dict[str, Any]]:
+        s, e = _resolve(window, start, end)
+        return await duck.global_tags(start=s, end=e, limit=limit)
+
+    @get("/trending")
+    async def get_trending(
+        self, window: str | None = None, start: datetime | None = None, end: datetime | None = None, limit: int = 15
+    ) -> list[dict[str, Any]]:
+        s, e = _resolve(window, start, end)
+        return await duck.global_trending(start=s, end=e, limit=limit)
+
+
+global_router = Router(path="/api/v2", route_handlers=[GlobalController])

frontend/app.js

@@ -460,7 +460,10 @@ const BUSY_RETRIES = 1;
 const BUSY_BACKOFF_MS = 2500;
 
 function endpoint(name, params) {
-  const base = `/api/v2/hashtag/${encodeURIComponent(state.hashtags.join(","))}/${name}`;
+  // No hashtags -> whole-OSM (global) endpoints; the co-occurring "hashtags" section maps to global trending.
+  const base = state.hashtags.length
+    ? `/api/v2/hashtag/${encodeURIComponent(state.hashtags.join(","))}/${name}`
+    : `/api/v2/global/${name === "hashtags" ? "trending" : name}`;
   const u = new URL(base, API_BASE);
   params.forEach((v, k) => u.searchParams.set(k, v));
   return u;
@@ -505,7 +508,7 @@ function freezeWindow() {
 function windowParams() {
   if (!state.windowStart || !state.windowEnd) freezeWindow();
   const p = new URLSearchParams();
-  // All-time omits the window: an explicit full range bypasses the API's all-time cache and warm.
+  // All-time omits the window: an explicit full range bypasses the API's all-time cache.
   if (state.range !== "all") {
     p.set("start", isoUTC(state.windowStart));
     p.set("end", isoUTC(state.windowEnd));
@@ -568,11 +571,17 @@ async function runQuery() {
   writeURL();
   renderWindowBar();
   fetchHealth();
-  if (!state.hashtags.length) {
-    showEmptyPrompt();
-    return;
+  const global = !state.hashtags.length;
+  if (global) {
+    // Whole-OSM stats are Postgres-only and limited to the last 7 days; wider or all-time is not supported.
+    const spanMs = state.windowEnd - state.windowStart;
+    if (state.range === "all" || spanMs > 7 * 86400 * 1000 + 60000) {
+      showGlobalPrompt();
+      return;
+    }
+  } else {
+    saveRecentSearch();
   }
-  saveRecentSearch();
   state.query?.abort?.();
   const ctrl = new AbortController();
   state.query = ctrl;
@@ -668,7 +677,6 @@ function sliceBatchToRows() {
 // the UI then pages 10/20/50 WITHIN a batch client-side. Only crossing a batch boundary, or a new
 // sort/search (`forceFetch`), hits the API. `setPodium` seeds the top-3 from the first batch.
 async function loadLeaderboardPage(setPodium = false, forceFetch = false) {
-  if (!state.hashtags.length) return;
   const startRow = (state.page - 1) * state.pageSize;
   const batchIndex = Math.floor(startRow / state.batchSize);
   // Serve from the loaded batch when possible.
@@ -932,7 +940,17 @@ function showEmptyPrompt() {
   $("#podium")?.closest("section")?.style.setProperty("display", "none");
   $("#ov-details").hidden = true;
   $("#podium").innerHTML = "";
-  $("#lb-body").innerHTML = `<tr><td colspan="8"><div class="empty"><i data-lucide="arrow-down-to-line"></i><h3>Extract a hashtag</h3><p>Type one or more hashtags above and press Extract.</p></div></td></tr>`;
+  $("#lb-body").innerHTML = `<tr><td colspan="8"><div class="empty"><i data-lucide="arrow-down-to-line"></i><h3>Extract a hashtag</h3><p>Type one or more hashtags above and press Extract, or leave it empty for whole-OSM stats (last 30 days).</p></div></td></tr>`;
+  $("#pagination").hidden = true;
+  refreshIcons();
+}
+
+function showGlobalPrompt() {
+  $("#overview")?.closest("section")?.style.setProperty("display", "none");
+  $("#podium")?.closest("section")?.style.setProperty("display", "none");
+  $("#ov-details").hidden = true;
+  $("#podium").innerHTML = "";
+  $("#lb-body").innerHTML = `<tr><td colspan="8"><div class="empty"><i data-lucide="globe"></i><h3>Whole-OSM stats</h3><p>Global stats (no hashtag) cover the last 7 days. Choose 1h, 24h, or 7d, then Extract.</p></div></td></tr>`;
   $("#pagination").hidden = true;
   refreshIcons();
 }
@@ -1569,7 +1587,7 @@ function renderEditorStats() {
 }
 
 async function fetchEditorStats() {
-  if (!state.hashtags.length) { state.editorStats = null; renderEditorStats(); return; }
+  if (!state.query) { state.editorStats = null; renderEditorStats(); return; }
   try {
     const editors = await apiGet("editors", windowParams(), state.query?.signal);
     const all = (editors || [])