Speed up workspace listing and search; add FTS index and fix search d… (#113)

* Speed up workspace listing and search; add FTS index and fix search deep links

* address PR feedback

* Fix review follow-ups for search window order and shared tab parsing

* Address feedback from brad

* Track indexed composers only after INSERT; tighten parity test

Author: clean6378-max-it

api/search.py

@@ -1,6 +1,6 @@
 """
 API route for search — mirrors src/app/api/search/route.ts
-GET /api/search?q=...&type=all|chat|composer
+GET /api/search?q=...&type=all|chat|composer&all_history=1
 """
 
 import logging
@@ -12,7 +12,9 @@
 
 from models import ParseWarningCollector, SearchResult
 from services.search import (
+    DEFAULT_SEARCH_WINDOW_DAYS,
     rank_results,
+    resolve_search_since_ms,
     search_cli_sessions,
     search_global_storage,
     search_legacy_workspaces,
@@ -22,6 +24,20 @@
 bp = Blueprint("search", __name__)
 _logger = logging.getLogger(__name__)
 
+_MAX_SEARCH_SINCE_DAYS = 36_500  # ~100 years; avoids timedelta overflow on bad input
+
+
+def _parse_since_days_param(raw: str | None) -> int | None:
+    if raw is None or not str(raw).strip():
+        return None
+    try:
+        days = int(raw)
+    except ValueError:
+        return None
+    if days <= 0 or days > _MAX_SEARCH_SINCE_DAYS:
+        return None
+    return days
+
 
 @bp.route("/api/search")
 def search() -> tuple[Response, int] | Response:
@@ -39,6 +55,11 @@ def search() -> tuple[Response, int] | Response:
         query = request.args.get("q", "").strip()
         search_type = request.args.get("type", "all")
         rules = current_app.config.get("EXCLUSION_RULES") or []
+        all_history = request.args.get("all_history") in ("1", "true")
+        since_ms = resolve_search_since_ms(
+            all_history=all_history,
+            since_days=_parse_since_days_param(request.args.get("since_days")),
+        )
 
         if not query:
             return json_response({"error": "No search query provided"}, 400)
@@ -50,20 +71,46 @@ def search() -> tuple[Response, int] | Response:
         if search_type != "chat":
             results.extend(
                 search_global_storage(
-                    workspace_path, query, query_lower, rules, parse_warnings
+                    workspace_path,
+                    query,
+                    query_lower,
+                    rules,
+                    parse_warnings,
+                    since_ms=since_ms,
                 )
             )
         results.extend(
-            search_legacy_workspaces(workspace_path, query, query_lower, search_type, rules)
+            search_legacy_workspaces(
+                workspace_path,
+                query,
+                query_lower,
+                search_type,
+                rules,
+                since_ms=since_ms,
+            )
         )
         if search_type == "all":
             results.extend(
                 search_cli_sessions(
-                    get_cli_chats_path(), query, query_lower, rules, parse_warnings
+                    get_cli_chats_path(),
+                    query,
+                    query_lower,
+                    rules,
+                    parse_warnings,
+                    since_ms=since_ms,
                 )
             )
 
-        payload: dict[str, Any] = {"results": rank_results(results)}
+        payload: dict[str, Any] = {
+            "results": rank_results(results),
+            "allHistory": since_ms is None,
+            "searchWindowDays": (
+                None if since_ms is None else (
+                    _parse_since_days_param(request.args.get("since_days"))
+                    or DEFAULT_SEARCH_WINDOW_DAYS
+                )
+            ),
+        }
         return json_response(parse_warnings.attach_to(payload))
 
     except Exception:

app.py

@@ -69,6 +69,17 @@ def inject_year() -> dict[str, int]:
     app.register_blueprint(pdf_bp)
     app.register_blueprint(config_bp)
 
+    try:
+        from services.search_index import start_search_index_background
+        from utils.workspace_path import resolve_workspace_path
+
+        start_search_index_background(
+            resolve_workspace_path(),
+            app.config["EXCLUSION_RULES"],
+        )
+    except Exception:
+        logging.getLogger(__name__).exception("Failed to start search index background worker")
+
     # ---------- Page routes ----------
 
     @app.route("/")

scripts/profile_search.py

@@ -0,0 +1,66 @@
+"""One-off search profiler — run: python scripts/profile_search.py [query]"""
+from __future__ import annotations
+
+import os
+import sys
+import time
+
+REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+if REPO_ROOT not in sys.path:
+    sys.path.insert(0, REPO_ROOT)
+
+from models import ParseWarningCollector
+from services.search import search_global_storage
+from services.workspace_db import (
+    COMPOSER_ROWS_WITH_HEADERS_SQL,
+    build_composer_id_to_workspace_id_cached,
+    collect_workspace_entries,
+    open_global_db,
+)
+from services.search import _index_bubble_texts_matching_query, _sql_like_substring
+from utils.workspace_path import resolve_workspace_path
+
+
+def main() -> None:
+    query = sys.argv[1] if len(sys.argv) > 1 else "export"
+    q = query.lower()
+    wp = resolve_workspace_path()
+    entries = collect_workspace_entries(wp)
+
+    t0 = time.perf_counter()
+    build_composer_id_to_workspace_id_cached(wp, entries, [])
+    print(f"composer_map: {time.perf_counter() - t0:.2f}s")
+
+    with open_global_db(wp) as (conn, _):
+        if conn is None:
+            print("no global db")
+            return
+        t0 = time.perf_counter()
+        rows = conn.execute(COMPOSER_ROWS_WITH_HEADERS_SQL).fetchall()
+        print(f"composer_rows {len(rows)}: {time.perf_counter() - t0:.2f}s")
+
+        t0 = time.perf_counter()
+        bubble_count = conn.execute(
+            "SELECT count(*) FROM cursorDiskKV WHERE key LIKE 'bubbleId:%'"
+        ).fetchone()[0]
+        print(f"bubble_count {bubble_count}: {time.perf_counter() - t0:.2f}s")
+
+        t0 = time.perf_counter()
+        idx = _index_bubble_texts_matching_query(conn, q)
+        print(f"bubble_index {len(idx)} composers: {time.perf_counter() - t0:.2f}s")
+
+        pattern = _sql_like_substring(q)
+        t0 = time.perf_counter()
+        raw_hits = sum(
+            1 for row in rows
+            if q in (row["value"] or "").lower()
+        )
+        print(f"composer_raw_hits {raw_hits}: {time.perf_counter() - t0:.2f}s")
+
+    t0 = time.perf_counter()
+    results = search_global_storage(wp, query, q, [], ParseWarningCollector())
+    print(f"search_global_storage {len(results)} hits: {time.perf_counter() - t0:.2f}s")
+
+
+if __name__ == "__main__":
+    main()

scripts/profile_search_breakdown.py

@@ -0,0 +1,103 @@
+"""Break down where search time goes with vs without window."""
+from __future__ import annotations
+
+import json
+import os
+import sys
+import time
+
+REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+if REPO_ROOT not in sys.path:
+    sys.path.insert(0, REPO_ROOT)
+
+from services.search import (
+    _composer_dict_timestamp_ms,
+    _composer_row_raw_text,
+    _index_bubble_texts_matching_query,
+    _timestamp_in_search_window,
+    resolve_search_since_ms,
+)
+from services.workspace_db import (
+    COMPOSER_ROWS_WITH_HEADERS_SQL,
+    build_composer_id_to_workspace_id_cached,
+    collect_workspace_entries,
+    open_global_db,
+)
+from utils.workspace_path import resolve_workspace_path
+
+
+def profile_window(since_ms: int | None, label: str) -> None:
+    query = "export"
+    q = query.lower()
+    wp = resolve_workspace_path()
+    entries = collect_workspace_entries(wp)
+
+    t0 = time.perf_counter()
+    build_composer_id_to_workspace_id_cached(wp, entries, [])
+    print(f"  composer_map: {time.perf_counter() - t0:.2f}s")
+
+    with open_global_db(wp) as (conn, _):
+        if conn is None:
+            return
+        t0 = time.perf_counter()
+        composer_rows = conn.execute(COMPOSER_ROWS_WITH_HEADERS_SQL).fetchall()
+        print(f"  load_composer_rows ({len(composer_rows)}): {time.perf_counter() - t0:.2f}s")
+
+        window_ids = None
+        if since_ms is not None:
+            t0 = time.perf_counter()
+            window_ids = set()
+            in_window_rows = []
+            for row in composer_rows:
+                composer_id = row["key"].split(":")[1]
+                raw_text = _composer_row_raw_text(row)
+                try:
+                    cd_probe = json.loads(raw_text)
+                except Exception:
+                    continue
+                if not isinstance(cd_probe, dict):
+                    continue
+                if not _timestamp_in_search_window(
+                    _composer_dict_timestamp_ms(cd_probe), since_ms
+                ):
+                    continue
+                window_ids.add(composer_id)
+                in_window_rows.append(row)
+            composer_rows = in_window_rows
+            print(
+                f"  window_filter ({len(window_ids)} composers): "
+                f"{time.perf_counter() - t0:.2f}s"
+            )
+
+        t0 = time.perf_counter()
+        idx = _index_bubble_texts_matching_query(conn, q, composer_ids=window_ids)
+        mode = (
+            "full_scan_python_filter"
+            if window_ids is not None and len(window_ids) <= 500
+            else "full_scan"
+        )
+        print(
+            f"  bubble_index ({len(idx)} hits, {mode}): "
+            f"{time.perf_counter() - t0:.2f}s"
+        )
+
+        t0 = time.perf_counter()
+        hits = 0
+        for row in composer_rows:
+            raw = _composer_row_raw_text(row)
+            cid = row["key"].split(":")[1]
+            if q in raw.lower() or cid in idx:
+                hits += 1
+        print(f"  composer_match_loop ({hits} candidates): {time.perf_counter() - t0:.2f}s")
+
+
+def main() -> None:
+    since = resolve_search_since_ms(all_history=False)
+    print("=== 30-day window ===")
+    profile_window(since, "30d")
+    print("=== all history ===")
+    profile_window(None, "all")
+
+
+if __name__ == "__main__":
+    main()

scripts/profile_search_window.py

@@ -0,0 +1,85 @@
+"""Profile search with 30-day window vs all history."""
+from __future__ import annotations
+
+import json
+import os
+import sys
+import time
+from datetime import datetime, timezone
+
+REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+if REPO_ROOT not in sys.path:
+    sys.path.insert(0, REPO_ROOT)
+
+from models import ParseWarningCollector
+from services.search import (
+    _composer_dict_timestamp_ms,
+    _timestamp_in_search_window,
+    resolve_search_since_ms,
+    search_cli_sessions,
+    search_global_storage,
+    search_legacy_workspaces,
+)
+from services.workspace_db import COMPOSER_ROWS_WITH_HEADERS_SQL, open_global_db
+from utils.workspace_path import get_cli_chats_path, resolve_workspace_path
+
+
+def main() -> None:
+    query = sys.argv[1] if len(sys.argv) > 1 else "export"
+    q = query.lower()
+    wp = resolve_workspace_path()
+    since = resolve_search_since_ms(all_history=False)
+    assert since is not None
+    print(
+        f"since_ms={since} "
+        f"({datetime.fromtimestamp(since / 1000, tz=timezone.utc).isoformat()})"
+    )
+
+    with open_global_db(wp) as (conn, _):
+        if conn is None:
+            print("no global db")
+            return
+        rows = conn.execute(COMPOSER_ROWS_WITH_HEADERS_SQL).fetchall()
+        total = len(rows)
+        in_window = 0
+        unknown_ts = 0
+        for row in rows:
+            try:
+                cd = json.loads(row["value"])
+            except Exception:
+                continue
+            ts = _composer_dict_timestamp_ms(cd)
+            if ts <= 0:
+                unknown_ts += 1
+            if _timestamp_in_search_window(ts, since):
+                in_window += 1
+        bubble_count = conn.execute(
+            "SELECT count(*) FROM cursorDiskKV WHERE key LIKE 'bubbleId:%'"
+        ).fetchone()[0]
+
+    print(f"composers total={total} in_30d_window={in_window} unknown_ts={unknown_ts}")
+    print(f"bubble_rows={bubble_count}")
+    print(f"below_bubble_threshold={in_window <= 500}")
+
+    pw = ParseWarningCollector()
+    for label, since_ms in [("30d", since), ("all", None)]:
+        t0 = time.perf_counter()
+        r1 = search_global_storage(wp, query, q, [], pw, since_ms=since_ms)
+        t1 = time.perf_counter() - t0
+        t0 = time.perf_counter()
+        r2 = search_legacy_workspaces(wp, query, q, "all", [], since_ms=since_ms)
+        t2 = time.perf_counter() - t0
+        t0 = time.perf_counter()
+        r3 = search_cli_sessions(get_cli_chats_path(), query, q, [], pw, since_ms=since_ms)
+        t3 = time.perf_counter() - t0
+        total_hits = len(r1) + len(r2) + len(r3)
+        total_s = t1 + t2 + t3
+        print(
+            f"{label}: global={len(r1)} ({t1:.2f}s) "
+            f"legacy={len(r2)} ({t2:.2f}s) cli={len(r3)} ({t3:.2f}s) "
+            f"total={total_hits} ({total_s:.2f}s)"
+        )
+
+
+if __name__ == "__main__":
+    main()