refactor: share search/acl helpers, lift lazy imports

- services/search: hoist LIKE_ESCAPE and escape_like out of routers so
  the FTS-fallback escape logic lives next to build_fts_query /
  build_like_words. routers/search and routers/ai now import the
  shared helpers instead of carrying byte-identical copies
- services/acl: hoist _build_id_clause out of routers/pages as
  build_id_clause; routers/activity also switches to it so
  activity_stats no longer hits SQLITE_MAX_VARIABLE_NUMBER (default
  999) on a wiki where the caller has more readable pages than that
- routers/pages: lift the per-request `from app.services.notifications
  import ...` out of create/update/delete page bodies to the module-
  level imports — deferred imports inside hot paths hide dependencies
  and break static analysis
- migrations: reorder _m010_site_settings to appear before
  _m011_mindmap_layout in the file so function order matches list
  order (execution order was already correct)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: PttCodingMan

backend/app/migrations.py

@@ -150,17 +150,6 @@ async def _m009_page_type(db: aiosqlite.Connection) -> None:
         )
 
 
-async def _m011_mindmap_layout(db: aiosqlite.Connection) -> None:
-    """Add `mindmap_layout` to pages so authors can pick LR / RL / Radial.
-
-    Nullable TEXT — NULL means "use the frontend default" (`'lr'`), so legacy
-    rows render unchanged. Validation is enforced by Pydantic Literal in
-    schemas.MindmapLayout, not at the DB layer (matches `page_type`).
-    """
-    if not await _column_exists(db, "pages", "mindmap_layout"):
-        await db.execute("ALTER TABLE pages ADD COLUMN mindmap_layout TEXT")
-
-
 async def _m010_site_settings(db: aiosqlite.Connection) -> None:
     """Create site_settings for branding overrides and the home-page slug.
 
@@ -179,6 +168,17 @@ async def _m010_site_settings(db: aiosqlite.Connection) -> None:
     )
 
 
+async def _m011_mindmap_layout(db: aiosqlite.Connection) -> None:
+    """Add `mindmap_layout` to pages so authors can pick LR / RL / Radial.
+
+    Nullable TEXT — NULL means "use the frontend default" (`'lr'`), so legacy
+    rows render unchanged. Validation is enforced by Pydantic Literal in
+    schemas.MindmapLayout, not at the DB layer (matches `page_type`).
+    """
+    if not await _column_exists(db, "pages", "mindmap_layout"):
+        await db.execute("ALTER TABLE pages ADD COLUMN mindmap_layout TEXT")
+
+
 MIGRATIONS: list[Migration] = [
     (1, "user_profile_columns", _m001_user_profile_columns),
     (2, "user_soft_delete", _m002_user_soft_delete),

backend/app/routers/activity.py

@@ -2,7 +2,7 @@
 from fastapi import APIRouter, Depends, Query
 from app.auth import get_current_user
 from app.database import get_db
-from app.services.acl import list_readable_page_ids
+from app.services.acl import build_id_clause, list_readable_page_ids
 
 router = APIRouter(prefix="/api/activity", tags=["activity"])
 
@@ -28,10 +28,10 @@ def _readable_clause(readable: frozenset[int] | set[int]) -> tuple[str, list]:
     if not readable:
         # No readable pages → drop every page-targeted row.
         return "(a.target_type != 'page')", []
-    placeholders = ",".join("?" * len(readable))
+    id_clause, id_params = build_id_clause(readable, column="a.target_id")
     return (
-        f"(a.target_type != 'page' OR a.target_id IN ({placeholders}))",
-        list(readable),
+        f"(a.target_type != 'page' OR {id_clause})",
+        id_params,
     )
 
 
@@ -99,32 +99,32 @@ async def activity_stats(user=Depends(get_current_user)):
             "total_users": 0,
         }
 
-    placeholders = ",".join("?" * len(readable))
-    readable_params = list(readable)
+    id_clause, id_params = build_id_clause(readable)
+    p_id_clause, p_id_params = build_id_clause(readable, column="p.id")
 
     top_viewed = await db.execute_fetchall(
         f"""SELECT id, slug, title, view_count FROM pages
-           WHERE id IN ({placeholders})
+           WHERE {id_clause}
            ORDER BY view_count DESC LIMIT 10""",
-        readable_params,
+        id_params,
     )
 
     recently_updated = await db.execute_fetchall(
         f"""SELECT p.id, p.slug, p.title, p.updated_at
            FROM pages p
-           WHERE p.id IN ({placeholders})
+           WHERE {p_id_clause}
            ORDER BY p.updated_at DESC LIMIT 10""",
-        readable_params,
+        p_id_params,
     )
 
     orphan_pages = await db.execute_fetchall(
         f"""SELECT p.id, p.slug, p.title, p.view_count
            FROM pages p
-           WHERE p.id IN ({placeholders})
+           WHERE {p_id_clause}
              AND p.id NOT IN (SELECT target_page_id FROM backlinks)
              AND p.parent_id IS NULL
            ORDER BY p.updated_at DESC LIMIT 20""",
-        readable_params,
+        p_id_params,
     )
 
     # total_pages reflects what *this user* can read, not the global count.

backend/app/routers/ai.py

@@ -22,7 +22,7 @@
 from app.config import settings
 from app.database import get_db
 from app.services.acl import list_readable_page_ids
-from app.services.search import build_fts_query, build_like_words
+from app.services.search import LIKE_ESCAPE, build_fts_query, build_like_words, escape_like
 
 # AI calls hit a paid upstream — never expose to anonymous traffic.
 router = APIRouter(
@@ -31,8 +31,6 @@
     dependencies=[Depends(require_real_user)],
 )
 
-_LIKE_ESCAPE = "\\"
-
 
 # ── rate limiting (per user, in-memory sliding window) ─────────────────
 # Simple deque-per-user. Good enough for a self-hosted wiki; a multi-worker
@@ -71,14 +69,6 @@ class ChatRequest(BaseModel):
 # ── retrieval ──────────────────────────────────────────────────────────
 
 
-def _escape_like(s: str) -> str:
-    return (
-        s.replace(_LIKE_ESCAPE, _LIKE_ESCAPE * 2)
-        .replace("%", _LIKE_ESCAPE + "%")
-        .replace("_", _LIKE_ESCAPE + "_")
-    )
-
-
 async def _fts_lookup(db, fts_query: str, readable_json: str, limit: int):
     sql = """
         SELECT p.slug, p.title, p.content_md
@@ -94,13 +84,13 @@ async def _fts_lookup(db, fts_query: str, readable_json: str, limit: int):
 
 async def _like_lookup(db, words: list[str], readable_json: str, limit: int):
     like_clauses = " OR ".join(
-        f"p.title LIKE ? ESCAPE '{_LIKE_ESCAPE}' "
-        f"OR p.content_md LIKE ? ESCAPE '{_LIKE_ESCAPE}'"
+        f"p.title LIKE ? ESCAPE '{LIKE_ESCAPE}' "
+        f"OR p.content_md LIKE ? ESCAPE '{LIKE_ESCAPE}'"
         for _ in words
     )
     like_params: list[str] = []
     for w in words:
-        pattern = f"%{_escape_like(w)}%"
+        pattern = f"%{escape_like(w)}%"
         like_params.extend([pattern, pattern])
     sql = f"""
         SELECT p.slug, p.title, p.content_md

backend/app/routers/pages.py

@@ -10,12 +10,17 @@
 from app.schemas import PageCreate, PageUpdate, PageResponse, PageListResponse, PageMoveRequest
 from app.auth import get_current_user
 from app.database import get_db, write_transaction
-from app.services.acl import invalidate_readable_cache, list_readable_page_ids, resolve_page_permission
+from app.services.acl import build_id_clause, invalidate_readable_cache, list_readable_page_ids, resolve_page_permission
 from app.services.search import rebuild_search_index, remove_from_search_index
 from app.services.wikilink import parse_and_update_backlinks
 from app.services.media_ref import parse_and_update_media_refs
 from app.routers.activity import log_activity
 from app.routers.versions import save_version
+from app.services.notifications import (
+    notify_page_created,
+    notify_page_deleted,
+    notify_page_updated,
+)
 
 router = APIRouter(prefix="/api/pages", tags=["pages"])
 
@@ -60,20 +65,6 @@ async def _should_count_view(db, user_id: int, page_id: int) -> bool:
     return True
 
 
-def _build_id_clause(ids: set[int], column: str = "id") -> tuple[str, list]:
-    """Produce a parameterized ``column IN (SELECT value FROM json_each(?))``
-    clause plus params.
-
-    Uses ``json_each`` instead of ``IN (?,?,...)`` to avoid hitting
-    SQLite's ``SQLITE_MAX_VARIABLE_NUMBER`` limit on large page sets.
-    For the empty set, returns a clause that never matches so downstream
-    SQL can be composed without branching.
-    """
-    if not ids:
-        return "0 = 1", []
-    return f"{column} IN (SELECT value FROM json_each(?))", [json.dumps(list(ids))]
-
-
 def slugify(title: str, existing_slug: str | None = None) -> str:
     """Generate a URL-friendly slug from title. Preserves CJK characters so
     Chinese/Japanese/Korean titles show up in the URL as-is rather than pinyin."""
@@ -139,7 +130,7 @@ async def list_pages(
     offset = (page - 1) * per_page
 
     readable = await list_readable_page_ids(db, user)
-    id_clause, id_params = _build_id_clause(readable)
+    id_clause, id_params = build_id_clause(readable)
 
     where = f"WHERE deleted_at IS NULL AND {id_clause}"
     params: list = list(id_params)
@@ -164,7 +155,7 @@ async def list_pages(
 async def page_tree(user=Depends(get_current_user)):
     db = await get_db()
     readable = await list_readable_page_ids(db, user)
-    id_clause, id_params = _build_id_clause(readable)
+    id_clause, id_params = build_id_clause(readable)
     rows = await db.execute_fetchall(
         f"SELECT id, slug, title, parent_id, sort_order FROM pages "
         f"WHERE deleted_at IS NULL AND {id_clause} "
@@ -205,7 +196,7 @@ def build_tree(parent_id):
 async def page_graph(user=Depends(get_current_user)):
     db = await get_db()
     readable = await list_readable_page_ids(db, user)
-    id_clause, id_params = _build_id_clause(readable)
+    id_clause, id_params = build_id_clause(readable)
     pages = await db.execute_fetchall(
         f"SELECT id, slug, title, parent_id FROM pages WHERE deleted_at IS NULL AND {id_clause}",
         id_params,
@@ -312,7 +303,6 @@ async def create_page(body: PageCreate, user=Depends(get_current_user)):
     new_page = dict(rows[0])
 
     # Fire notification
-    from app.services.notifications import notify_page_created
     await notify_page_created(db, new_page, user)
 
     return new_page
@@ -468,7 +458,6 @@ async def update_page(slug: str, body: PageUpdate, user=Depends(get_current_user
 
     # Fire notifications if content/title actually changed
     if content_changed or title_changed:
-        from app.services.notifications import notify_page_updated
         await notify_page_updated(db, updated, user, {"title_changed": title_changed, "content_changed": content_changed})
 
     return updated
@@ -487,7 +476,7 @@ async def get_children(slug: str, user=Depends(get_current_user)):
         raise HTTPException(status_code=404, detail="Page not found")
 
     readable = await list_readable_page_ids(db, user)
-    id_clause, id_params = _build_id_clause(readable)
+    id_clause, id_params = build_id_clause(readable)
     children = await db.execute_fetchall(
         f"SELECT * FROM pages WHERE parent_id = ? AND deleted_at IS NULL AND {id_clause} "
         f"ORDER BY sort_order, title",
@@ -509,7 +498,7 @@ async def get_backlinks(slug: str, user=Depends(get_current_user)):
         raise HTTPException(status_code=404, detail="Page not found")
 
     readable = await list_readable_page_ids(db, user)
-    id_clause, id_params = _build_id_clause(readable, column="p.id")
+    id_clause, id_params = build_id_clause(readable, column="p.id")
     backlinks = await db.execute_fetchall(
         f"""SELECT p.id, p.slug, p.title
            FROM backlinks b
@@ -622,7 +611,6 @@ async def delete_page(slug: str, user=Depends(get_current_user)):
     invalidate_readable_cache()
 
     # Fire notification
-    from app.services.notifications import notify_page_deleted
     await notify_page_deleted(db, {"id": page_id, "title": page_title, "slug": slug}, user)
 
     return {"ok": True}

backend/app/routers/search.py

@@ -5,21 +5,10 @@
 from app.auth import get_current_user
 from app.database import get_db
 from app.services.acl import list_readable_page_ids
-from app.services.search import build_fts_query, build_like_words
+from app.services.search import LIKE_ESCAPE, build_fts_query, build_like_words, escape_like
 
 router = APIRouter(prefix="/api/search", tags=["search"])
 
-# LIKE wildcards that need escaping so user input like "100%" is treated literally.
-_LIKE_ESCAPE = "\\"
-
-
-def _escape_like(s: str) -> str:
-    return (
-        s.replace(_LIKE_ESCAPE, _LIKE_ESCAPE * 2)
-        .replace("%", _LIKE_ESCAPE + "%")
-        .replace("_", _LIKE_ESCAPE + "_")
-    )
-
 
 def make_snippet(text: str, query: str, max_len: int = 200) -> str:
     """Extract a snippet around the first match and wrap matches in <mark>."""
@@ -118,14 +107,14 @@ async def _search_like(db, words, tag, readable_json, per_page, offset):
     `%` and `_` in user input are escaped so they are treated literally.
     """
     like_clauses = " OR ".join(
-        f"p.title LIKE ? ESCAPE '{_LIKE_ESCAPE}' "
-        f"OR p.content_md LIKE ? ESCAPE '{_LIKE_ESCAPE}'"
+        f"p.title LIKE ? ESCAPE '{LIKE_ESCAPE}' "
+        f"OR p.content_md LIKE ? ESCAPE '{LIKE_ESCAPE}'"
         for _ in words
     )
     like_clauses = f"({like_clauses})"
     like_params = []
     for w in words:
-        pattern = f"%{_escape_like(w)}%"
+        pattern = f"%{escape_like(w)}%"
         like_params.extend([pattern, pattern])
 
     if tag:

backend/app/services/acl.py

@@ -20,6 +20,7 @@
   (no live references) is accessible to its uploader and admins only.
 """
 
+import json
 import time
 
 from fastapi import HTTPException
@@ -36,6 +37,20 @@
 _READABLE_CACHE_TTL = 30  # seconds
 
 
+def build_id_clause(ids, column: str = "id") -> tuple[str, list]:
+    """SQL fragment + params for ``{column} IN (SELECT value FROM json_each(?))``.
+
+    Uses ``json_each`` rather than inline ``IN (?,?,...)`` to avoid hitting
+    SQLite's ``SQLITE_MAX_VARIABLE_NUMBER`` (default 999) when callers feed
+    in the full readable-page set on a wiki with thousands of pages. For an
+    empty set returns a clause that never matches so downstream SQL can be
+    composed without branching.
+    """
+    if not ids:
+        return "0 = 1", []
+    return f"{column} IN (SELECT value FROM json_each(?))", [json.dumps(list(ids))]
+
+
 def invalidate_readable_cache(user_id: int | None = None):
     """Drop cached readable-page sets.
 

backend/app/services/search.py

@@ -45,6 +45,24 @@ def _escape_fts_phrase(s: str) -> str:
     return s.replace('"', '""')
 
 
+# ── LIKE fallback escaping ─────────────────────────────────────────────────
+#
+# When FTS misses we fall back to LIKE patterns. SQLite LIKE treats `%` and
+# `_` as wildcards; we escape them with a backslash and tell SQLite about it
+# via `ESCAPE '\\'`. The escape character itself must also be escaped.
+
+LIKE_ESCAPE = "\\"
+
+
+def escape_like(s: str) -> str:
+    """Escape `%`, `_`, and `\\` so the value is matched literally by LIKE."""
+    return (
+        s.replace(LIKE_ESCAPE, LIKE_ESCAPE * 2)
+        .replace("%", LIKE_ESCAPE + "%")
+        .replace("_", LIKE_ESCAPE + "_")
+    )
+
+
 def build_fts_query(question: str) -> str | None:
     """Build an FTS5 MATCH expression that aligns with the trigram index.