feat(search): page content-search endpoint (multi-tag AND + text + warnings)

New GET /v1/me/page-search surfacing the analysis worker's output:

- repo::page_analysis::page_search + PageSearchQuery: multi-tag AND across
  (user page_tags ∪ global page_auto_tags) via the unnest double-negative
  idiom, weighted OCR/scene text ranking (ts_rank over search_doc), and
  content-warning include/exclude. One row per page with is_nsfw +
  deduped content_warnings + rank.
- domain::PageSearchItem.
- api::page_tags: /me/page-search handler with CSV tag/warning parsing
  (parse_tags_csv reuses normalize_tag; parse_warnings_csv validates the
  closed vocabulary), requiring at least one positive filter (422 else).
  This is where the reserved OCR text search lands for pages.

Tests: multi-tag AND user∪auto, speech>sfx ranking, cw include/exclude
(+ row flags), text-only (no tags), missing-filter 422, unknown-warning
422, auth required.

Note: the /me/page-tags/chapters|mangas aggregations keep single-tag
behavior + the reserved text=501 for now; page-level search is the
primary text surface.

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

backend/src/api/page_tags.rs

@@ -14,12 +14,14 @@ use uuid::Uuid;
 use crate::api::pagination::PagedResponse;
 use crate::app::AppState;
 use crate::auth::extractor::CurrentUser;
+use crate::domain::page_analysis::{ContentWarning, PageSearchItem};
 use crate::domain::page_tag::{
     NewPageTag, PageTagSummary, TaggedChapterAggregate, TaggedMangaAggregate,
     TaggedPageItem,
 };
 use crate::error::{AppError, AppResult};
 use crate::repo;
+use crate::repo::page_analysis::PageSearchQuery;
 use crate::repo::page_tag::Order;
 
 pub fn routes() -> Router<AppState> {
@@ -31,6 +33,7 @@ pub fn routes() -> Router<AppState> {
         // though the cookie already implies it.
         .route("/pages/:id/my-tags", get(list_for_page))
         .route("/me/page-tags", get(list_mine))
+        .route("/me/page-search", get(page_search))
         .route("/me/page-tags/distinct", get(list_distinct_mine))
         .route("/me/page-tags/chapters", get(list_chapters_for_tag))
         .route("/me/page-tags/mangas", get(list_mangas_for_tag))
@@ -248,6 +251,109 @@ async fn list_mine(
     Ok(Json(PagedResponse::with_total(items, limit, offset, total)))
 }
 
+#[derive(Debug, Deserialize)]
+pub struct PageSearchParams {
+    #[serde(default = "default_limit")]
+    pub limit: i64,
+    #[serde(default)]
+    pub offset: i64,
+    /// Comma-separated tags, AND-ed. Matched against the caller's page
+    /// tags ∪ the global auto-tags.
+    #[serde(default)]
+    pub tags: Option<String>,
+    /// Free-text query over the OCR + scene-description search document.
+    #[serde(default)]
+    pub text: Option<String>,
+    /// Comma-separated content warnings the page must carry (all of them).
+    #[serde(default)]
+    pub cw_include: Option<String>,
+    /// Comma-separated content warnings the page must NOT carry (any of).
+    #[serde(default)]
+    pub cw_exclude: Option<String>,
+}
+
+/// Split a comma-separated tag list into normalized, deduped tag names.
+fn parse_tags_csv(raw: Option<&str>) -> AppResult<Vec<String>> {
+    let Some(raw) = raw else { return Ok(Vec::new()) };
+    let mut seen = std::collections::HashSet::new();
+    let mut out = Vec::new();
+    for part in raw.split(',') {
+        if part.trim().is_empty() {
+            continue;
+        }
+        let norm = normalize_tag(part)?;
+        if seen.insert(norm.clone()) {
+            out.push(norm);
+        }
+    }
+    Ok(out)
+}
+
+/// Split + validate a comma-separated content-warning list against the
+/// closed vocabulary, returning canonical lowercase names. Unknown values
+/// are a 422 rather than a silent drop so a typo'd filter is visible.
+fn parse_warnings_csv(raw: Option<&str>) -> AppResult<Vec<String>> {
+    let Some(raw) = raw else { return Ok(Vec::new()) };
+    let mut seen = std::collections::HashSet::new();
+    let mut out = Vec::new();
+    for part in raw.split(',') {
+        let t = part.trim();
+        if t.is_empty() {
+            continue;
+        }
+        let w = ContentWarning::parse_strict(t).ok_or_else(|| AppError::ValidationFailed {
+            message: format!("unknown content warning {t:?}"),
+            details: json!({ "content_warning": "must be one of sexual|nudity|gore|violence|disturbing" }),
+        })?;
+        let canon = format!("{w:?}").to_lowercase();
+        if seen.insert(canon.clone()) {
+            out.push(canon);
+        }
+    }
+    Ok(out)
+}
+
+/// Content search over the caller's reachable pages: multi-tag AND (own
+/// page tags ∪ global auto-tags), weighted OCR/scene text ranking, and
+/// content-warning include/exclude. At least one positive filter (tags,
+/// text, or cw_include) is required so the endpoint never dumps every page.
+async fn page_search(
+    State(state): State<AppState>,
+    CurrentUser(user): CurrentUser,
+    Query(params): Query<PageSearchParams>,
+) -> AppResult<Json<PagedResponse<PageSearchItem>>> {
+    let limit = params.limit.clamp(1, 200);
+    let offset = params.offset.max(0);
+    let tags = parse_tags_csv(params.tags.as_deref())?;
+    let cw_include = parse_warnings_csv(params.cw_include.as_deref())?;
+    let cw_exclude = parse_warnings_csv(params.cw_exclude.as_deref())?;
+    let text = params
+        .text
+        .as_deref()
+        .map(str::trim)
+        .filter(|s| !s.is_empty())
+        .map(str::to_string);
+
+    if tags.is_empty() && text.is_none() && cw_include.is_empty() {
+        return Err(AppError::ValidationFailed {
+            message: "at least one of tags, text, or cw_include is required".into(),
+            details: json!({ "filter": "required" }),
+        });
+    }
+
+    let query = PageSearchQuery {
+        user_id: user.id,
+        tags,
+        text,
+        cw_include,
+        cw_exclude,
+        limit,
+        offset,
+    };
+    let (items, total) = repo::page_analysis::page_search(&state.db, &query).await?;
+    Ok(Json(PagedResponse::with_total(items, limit, offset, total)))
+}
+
 async fn list_distinct_mine(
     State(state): State<AppState>,
     CurrentUser(user): CurrentUser,

backend/src/domain/mod.rs

@@ -28,8 +28,8 @@ pub use genre::{Genre, GenreRef};
 pub use manga::{Manga, MangaCard, MangaDetail};
 pub use page::Page;
 pub use page_analysis::{
-    AnalysisStatus, ContentWarning, OcrKind, OcrResult, PageAnalysis, SafetyFlag,
-    VisionAnalysis,
+    AnalysisStatus, ContentWarning, OcrKind, OcrResult, PageAnalysis, PageSearchItem,
+    SafetyFlag, VisionAnalysis,
 };
 pub use page_tag::{
     NewPageTag, PageTagSummary, TaggedChapterAggregate, TaggedMangaAggregate,

backend/src/domain/page_analysis.rs

@@ -108,6 +108,26 @@ impl ContentWarning {
     }
 }
 
+/// One result row from the page content-search (`GET /v1/me/page-search`).
+/// One row per matching page, carrying the breadcrumb plus the moderation
+/// flags and the text-search rank so the UI can badge and order results.
+#[derive(Debug, Clone, Serialize, FromRow)]
+pub struct PageSearchItem {
+    pub page_id: Uuid,
+    pub chapter_id: Uuid,
+    pub manga_id: Uuid,
+    pub page_number: i32,
+    pub chapter_number: i32,
+    pub chapter_title: Option<String>,
+    pub manga_title: String,
+    pub storage_key: String,
+    pub is_nsfw: bool,
+    /// Deduped content warnings on this page (canonical lowercase names).
+    pub content_warnings: Vec<String>,
+    /// `ts_rank` against the text query; `0` for tag/warning-only searches.
+    pub rank: f32,
+}
+
 /// One persisted `page_analysis` row.
 #[derive(Debug, Clone, Serialize, FromRow)]
 pub struct PageAnalysis {

backend/src/repo/page_analysis.rs

@@ -15,9 +15,27 @@ use sqlx::PgPool;
 use uuid::Uuid;
 
 use crate::crawler::jobs::{self, JobPayload};
-use crate::domain::page_analysis::{ContentWarning, OcrKind, PageAnalysis, VisionAnalysis};
+use crate::domain::page_analysis::{
+    ContentWarning, OcrKind, PageAnalysis, PageSearchItem, VisionAnalysis,
+};
 use crate::error::AppResult;
 
+/// Filter set for [`page_search`]. `tags` are AND-ed (a page must carry
+/// every one, satisfiable by EITHER the user's page tag or a global auto
+/// tag); `text` ranks via the weighted tsvector; `cw_include` requires all
+/// listed warnings; `cw_exclude` rejects any. All string values arrive
+/// pre-normalized (tags lowercased, warnings validated) from the handler.
+#[derive(Debug, Default, Clone)]
+pub struct PageSearchQuery {
+    pub user_id: uuid::Uuid,
+    pub tags: Vec<String>,
+    pub text: Option<String>,
+    pub cw_include: Vec<String>,
+    pub cw_exclude: Vec<String>,
+    pub limit: i64,
+    pub offset: i64,
+}
+
 /// Longest tag the shared `tags` table accepts (`upsert_by_name` enforces
 /// the same bound). Auto-tags over this are dropped here rather than
 /// aborting the whole page's analysis on one bad model output.
@@ -245,6 +263,107 @@ pub async fn persist_analysis(
     Ok(())
 }
 
+/// Content search over pages: multi-tag AND across (user page tags ∪
+/// global auto tags), optional weighted text search over the OCR/scene
+/// document, and content-warning include/exclude. Returns one row per
+/// matching page plus the total for pagination.
+///
+/// Empty `tags` makes the tag clause vacuously true (text- or
+/// warning-only search), mirroring the manga search's `unnest` idiom. The
+/// caller is responsible for requiring at least one positive filter so
+/// this never degenerates into "every page".
+pub async fn page_search(
+    pool: &PgPool,
+    q: &PageSearchQuery,
+) -> AppResult<(Vec<PageSearchItem>, i64)> {
+    // `text` participates in three places ($3): the rank, the match
+    // predicate, and the order key. Empty string disables text filtering.
+    let text = q.text.as_deref().unwrap_or("").trim();
+
+    const WHERE: &str = r#"
+        NOT EXISTS (
+            SELECT 1 FROM unnest($2::text[]) AS req(name)
+             WHERE NOT EXISTS (
+                SELECT 1 FROM page_tags ut
+                  JOIN tags t ON t.id = ut.tag_id
+                 WHERE ut.page_id = p.id AND ut.user_id = $1 AND lower(t.name) = req.name
+                UNION ALL
+                SELECT 1 FROM page_auto_tags at
+                  JOIN tags t ON t.id = at.tag_id
+                 WHERE at.page_id = p.id AND lower(t.name) = req.name
+             )
+        )
+        AND ($3 = '' OR pa.search_doc @@ plainto_tsquery('simple', $3))
+        AND NOT EXISTS (
+            SELECT 1 FROM unnest($4::text[]) AS req(w)
+             WHERE NOT EXISTS (
+                SELECT 1 FROM page_content_warnings pw
+                 WHERE pw.page_id = p.id AND pw.warning = req.w
+             )
+        )
+        AND NOT EXISTS (
+            SELECT 1 FROM page_content_warnings pw
+             WHERE pw.page_id = p.id AND pw.warning = ANY($5::text[])
+        )
+    "#;
+
+    let rows_sql = format!(
+        r#"
+        SELECT
+            p.id          AS page_id,
+            p.chapter_id  AS chapter_id,
+            ch.manga_id   AS manga_id,
+            p.page_number AS page_number,
+            ch.number     AS chapter_number,
+            ch.title      AS chapter_title,
+            m.title       AS manga_title,
+            p.storage_key AS storage_key,
+            COALESCE(pa.is_nsfw, false) AS is_nsfw,
+            COALESCE(
+                (SELECT array_agg(pw.warning ORDER BY pw.warning)
+                   FROM page_content_warnings pw WHERE pw.page_id = p.id),
+                ARRAY[]::text[]
+            ) AS content_warnings,
+            COALESCE(ts_rank(pa.search_doc, plainto_tsquery('simple', $3)), 0)::real AS rank
+          FROM pages p
+          JOIN chapters ch ON ch.id = p.chapter_id
+          JOIN mangas   m  ON m.id  = ch.manga_id
+          LEFT JOIN page_analysis pa ON pa.page_id = p.id
+         WHERE {WHERE}
+         ORDER BY (CASE WHEN $3 = '' THEN 0 ELSE 1 END) DESC, rank DESC, p.id
+         LIMIT $6 OFFSET $7
+        "#
+    );
+    let rows = sqlx::query_as::<_, PageSearchItem>(&rows_sql)
+        .bind(q.user_id)
+        .bind(&q.tags)
+        .bind(text)
+        .bind(&q.cw_include)
+        .bind(&q.cw_exclude)
+        .bind(q.limit)
+        .bind(q.offset)
+        .fetch_all(pool)
+        .await?;
+
+    let count_sql = format!(
+        "SELECT count(*) FROM pages p \
+           JOIN chapters ch ON ch.id = p.chapter_id \
+           JOIN mangas m ON m.id = ch.manga_id \
+           LEFT JOIN page_analysis pa ON pa.page_id = p.id \
+          WHERE {WHERE}"
+    );
+    let (total,): (i64,) = sqlx::query_as(&count_sql)
+        .bind(q.user_id)
+        .bind(&q.tags)
+        .bind(text)
+        .bind(&q.cw_include)
+        .bind(&q.cw_exclude)
+        .fetch_one(pool)
+        .await?;
+
+    Ok((rows, total))
+}
+
 /// Append a token to a tsvector text bucket with a trailing space.
 fn push_token(bucket: &mut String, text: &str) {
     bucket.push_str(text);