feat(search): enable OCR text search on page-tag aggregation endpoints

The `?text=` param on `/v1/me/page-tags/chapters` and `/mangas` was
reserved and returned 501 `text_search_not_yet_supported`. Now that OCR
populates `search_doc`, flip it to real search.

- `aggregate_chapters_for_tag` / `aggregate_mangas_for_tag` take an optional
  `text`; when non-blank they JOIN `page_analysis` and filter on
  `search_doc @@ plainto_tsquery('simple', $n)`, consistent with
  `repo::page_analysis::page_search`. `match_count` and the sample
  thumbnails reflect the filtered set. `text` is always bound, never
  interpolated.
- Drop `ensure_text_unsupported` and the 501 guard. `AppError::NotImplemented`
  stays as a generic variant for future reservations.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

backend/tests/api_page_tags.rs

@@ -680,24 +680,123 @@ async fn aggregate_rejects_invalid_order(pool: PgPool) {
 }
 
 #[sqlx::test(migrations = "./migrations")]
-async fn aggregate_with_text_param_is_501_with_stable_code(pool: PgPool) {
-    // OCR text search isn't built yet; the param is accepted so adding
-    // OCR won't break the wire shape, but rejected with a distinct
-    // status + code. The code is the wire contract — clients pin on
-    // `text_search_not_yet_supported`, not the message.
-    let h = common::harness(pool);
+async fn aggregate_with_text_param_filters_by_ocr(pool: PgPool) {
+    // OCR text search: a page tagged `funny` whose OCR contains "guts" is
+    // returned for `&text=guts` on both aggregation endpoints, and excluded
+    // for a query its OCR doesn't contain. The filter runs against the same
+    // precomputed `search_doc` the OCR worker writes via `persist_analysis`.
+    let h = common::harness(pool.clone());
     let (_, cookie) = common::register_user(&h.app).await;
+    let manga_id = common::seed_manga_via_api(&h.app, &cookie, "M").await;
+    let (_, page_id) = seed_chapter_with_page(&h.app, &cookie, manga_id, 1).await;
+    assert_eq!(add_tag(&h.app, &cookie, &page_id, "funny").await, StatusCode::CREATED);
+
+    // Persist OCR text exactly as the ocrs backend would (via the same mapper).
+    let page_uuid = Uuid::parse_str(&page_id).unwrap();
+    let analysis =
+        mangalord::analysis::ocr::lines_to_analysis(vec!["spilling the guts here".to_string()]);
+    mangalord::repo::page_analysis::persist_analysis(&pool, page_uuid, &analysis, "ocrs")
+        .await
+        .unwrap();
+
+    for endpoint in ["chapters", "mangas"] {
+        // Matching text → the tagged+OCR'd row is returned.
+        let resp = h
+            .app
+            .clone()
+            .oneshot(common::get_with_cookie(
+                &format!("/api/v1/me/page-tags/{endpoint}?tag=funny&text=guts"),
+                &cookie,
+            ))
+            .await
+            .unwrap();
+        assert_eq!(resp.status(), StatusCode::OK, "{endpoint} matching");
+        let body = common::body_json(resp).await;
+        assert_eq!(body["items"].as_array().unwrap().len(), 1, "{endpoint} matching items");
+        assert_eq!(body["page"]["total"], 1, "{endpoint} matching total");
+
+        // Non-matching text → excluded.
+        let resp = h
+            .app
+            .clone()
+            .oneshot(common::get_with_cookie(
+                &format!("/api/v1/me/page-tags/{endpoint}?tag=funny&text=zzzznomatch"),
+                &cookie,
+            ))
+            .await
+            .unwrap();
+        assert_eq!(resp.status(), StatusCode::OK, "{endpoint} non-matching");
+        let body = common::body_json(resp).await;
+        assert_eq!(body["items"].as_array().unwrap().len(), 0, "{endpoint} non-matching items");
+        assert_eq!(body["page"]["total"], 0, "{endpoint} non-matching total");
+    }
+}
+
+#[sqlx::test(migrations = "./migrations")]
+async fn aggregate_text_param_counts_only_matching_pages(pool: PgPool) {
+    // A chapter with TWO tagged pages where only ONE page's OCR matches the
+    // text. `match_count` / `total` must reflect the filtered count (1), not
+    // the tag-only count (2), and the sample thumbnails must contain only the
+    // matching page. This is what a single-page test can't distinguish — it
+    // pins the JOIN/placeholder wiring and the sample-subquery filter.
+    let h = common::harness(pool.clone());
+    let (_, cookie) = common::register_user(&h.app).await;
+    let manga_id = common::seed_manga_via_api(&h.app, &cookie, "M").await;
+    let (_, page_ids) = seed_chapter_with_n_pages(&h.app, &cookie, manga_id, 1, 2).await;
+    for pid in &page_ids {
+        assert_eq!(add_tag(&h.app, &cookie, pid, "funny").await, StatusCode::CREATED);
+    }
+
+    // Page 0 OCR contains "guts"; page 1 OCR contains only "filler".
+    let p0 = Uuid::parse_str(&page_ids[0]).unwrap();
+    let p1 = Uuid::parse_str(&page_ids[1]).unwrap();
+    let a0 = mangalord::analysis::ocr::lines_to_analysis(vec!["the guts spill out".to_string()]);
+    let a1 = mangalord::analysis::ocr::lines_to_analysis(vec!["just filler text".to_string()]);
+    mangalord::repo::page_analysis::persist_analysis(&pool, p0, &a0, "ocrs").await.unwrap();
+    mangalord::repo::page_analysis::persist_analysis(&pool, p1, &a1, "ocrs").await.unwrap();
+
     let resp = h
         .app
+        .clone()
         .oneshot(common::get_with_cookie(
             "/api/v1/me/page-tags/chapters?tag=funny&text=guts",
             &cookie,
         ))
         .await
         .unwrap();
-    assert_eq!(resp.status(), StatusCode::NOT_IMPLEMENTED);
+    assert_eq!(resp.status(), StatusCode::OK);
     let body = common::body_json(resp).await;
-    assert_eq!(body["error"]["code"], "text_search_not_yet_supported");
+    let items = body["items"].as_array().unwrap();
+    assert_eq!(items.len(), 1, "one chapter row");
+    assert_eq!(body["page"]["total"], 1);
+    // Filtered match_count is the matching-page count, not the tag count.
+    assert_eq!(items[0]["match_count"], 1, "only one page's OCR matches");
+    // Sample thumbnails reflect the filter: only the matching page's key.
+    let samples = items[0]["sample_storage_keys"].as_array().unwrap();
+    assert_eq!(samples.len(), 1, "thumbnails restricted to matching pages");
+}
+
+#[sqlx::test(migrations = "./migrations")]
+async fn aggregate_blank_text_param_is_tag_only(pool: PgPool) {
+    // A blank `text=` must not filter — it falls back to tag-only aggregation
+    // (the page has a tag but no analysis row at all).
+    let h = common::harness(pool);
+    let (_, cookie) = common::register_user(&h.app).await;
+    let manga_id = common::seed_manga_via_api(&h.app, &cookie, "M").await;
+    let (_, page_id) = seed_chapter_with_page(&h.app, &cookie, manga_id, 1).await;
+    assert_eq!(add_tag(&h.app, &cookie, &page_id, "funny").await, StatusCode::CREATED);
+
+    let resp = h
+        .app
+        .oneshot(common::get_with_cookie(
+            "/api/v1/me/page-tags/chapters?tag=funny&text=",
+            &cookie,
+        ))
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), StatusCode::OK);
+    let body = common::body_json(resp).await;
+    assert_eq!(body["items"].as_array().unwrap().len(), 1);
 }
 
 #[sqlx::test(migrations = "./migrations")]