feat: bookmark create enqueues SyncChapterContent jobs (0.29.0)

After a successful bookmark insert, the create handler spawns a
detached tokio task that calls pipeline::enqueue_pending_for_manga
for every chapter of the manga where page_count = 0 and the source
row is not dropped. Bookmark create returns 201 immediately; enqueue
work happens in the background and its failure is logged without
surfacing to the user (the daily cron sweeps anything missed).

The Phase A dedup index handles re-bookmarks idempotently — deleting
and recreating a bookmark does not duplicate in-flight jobs — and the
Phase B worker pool drains them.

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

backend/tests/api_bookmarks.rs

@@ -438,3 +438,196 @@ async fn list_me_returns_paged_envelope(pool: PgPool) {
     // without paging through.
     assert_eq!(body["page"]["total"], 0);
 }
+
+// -------------------------------------------------------------------------
+// Bookmark create -> SyncChapterContent job enqueue (background task)
+// -------------------------------------------------------------------------
+
+async fn seed_chapter_with_source(
+    pool: &PgPool,
+    manga_id: Uuid,
+    number: i32,
+    source_id: &str,
+    source_chapter_key: &str,
+    source_url: &str,
+    dropped: bool,
+) -> Uuid {
+    let chapter_id: Uuid =
+        mangalord::repo::chapter::create(pool, manga_id, number, None, None)
+            .await
+            .unwrap()
+            .id;
+    sqlx::query("INSERT INTO sources (id, name, base_url) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING")
+        .bind(source_id)
+        .bind(source_id)
+        .bind("https://example.com")
+        .execute(pool)
+        .await
+        .unwrap();
+    let dropped_at = if dropped { "now()" } else { "NULL" };
+    sqlx::query(&format!(
+        "INSERT INTO chapter_sources (source_id, source_chapter_key, chapter_id, source_url, dropped_at) \
+         VALUES ($1, $2, $3, $4, {dropped_at})"
+    ))
+    .bind(source_id)
+    .bind(source_chapter_key)
+    .bind(chapter_id)
+    .bind(source_url)
+    .execute(pool)
+    .await
+    .unwrap();
+    chapter_id
+}
+
+/// Poll `crawler_jobs` for the expected pending count, up to ~1.5s, so the
+/// detached `tokio::spawn` from the bookmark create handler has time to
+/// land regardless of CI scheduling jitter.
+async fn wait_for_pending_count(pool: &PgPool, expected: i64) -> i64 {
+    for _ in 0..30 {
+        let count: i64 = sqlx::query_scalar(
+            "SELECT COUNT(*) FROM crawler_jobs \
+              WHERE state = 'pending' \
+                AND payload->>'kind' = 'sync_chapter_content'",
+        )
+        .fetch_one(pool)
+        .await
+        .unwrap();
+        if count >= expected {
+            return count;
+        }
+        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+    }
+    sqlx::query_scalar::<_, i64>(
+        "SELECT COUNT(*) FROM crawler_jobs \
+          WHERE state = 'pending' \
+            AND payload->>'kind' = 'sync_chapter_content'",
+    )
+    .fetch_one(pool)
+    .await
+    .unwrap()
+}
+
+#[sqlx::test(migrations = "./migrations")]
+async fn create_enqueues_sync_chapter_content_jobs_for_pending_chapters(pool: PgPool) {
+    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, "Berserk").await;
+
+    // Two zero-page chapters with non-dropped sources.
+    let c1 = seed_chapter_with_source(&pool, manga_id, 1, "target", "ch1", "https://example.com/c1", false).await;
+    let c2 = seed_chapter_with_source(&pool, manga_id, 2, "target", "ch2", "https://example.com/c2", false).await;
+
+    let resp = h
+        .app
+        .clone()
+        .oneshot(common::post_json_with_cookie(
+            "/api/v1/bookmarks",
+            json!({ "manga_id": manga_id.to_string() }),
+            &cookie,
+        ))
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), StatusCode::CREATED);
+
+    let count = wait_for_pending_count(&pool, 2).await;
+    assert_eq!(count, 2, "both pending chapters should be enqueued");
+
+    let chapter_ids: Vec<String> = sqlx::query_scalar(
+        "SELECT payload->>'chapter_id' FROM crawler_jobs \
+          WHERE payload->>'kind' = 'sync_chapter_content' \
+          ORDER BY payload->>'chapter_id'",
+    )
+    .fetch_all(&pool)
+    .await
+    .unwrap();
+    let mut expected = vec![c1.to_string(), c2.to_string()];
+    expected.sort();
+    assert_eq!(chapter_ids, expected);
+}
+
+#[sqlx::test(migrations = "./migrations")]
+async fn re_bookmark_after_delete_does_not_re_enqueue_pending_jobs(pool: PgPool) {
+    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, "Berserk").await;
+    let _ = seed_chapter_with_source(&pool, manga_id, 1, "target", "ch1", "https://example.com/c1", false).await;
+
+    // First bookmark — should enqueue 1.
+    let resp = h
+        .app
+        .clone()
+        .oneshot(common::post_json_with_cookie(
+            "/api/v1/bookmarks",
+            json!({ "manga_id": manga_id.to_string() }),
+            &cookie,
+        ))
+        .await
+        .unwrap();
+    let bookmark_id = common::body_json(resp).await["id"].as_str().unwrap().to_string();
+    assert_eq!(wait_for_pending_count(&pool, 1).await, 1);
+
+    // Delete the bookmark, then re-bookmark — the existing pending job
+    // is still there so the dedup index suppresses the second enqueue.
+    let resp = h
+        .app
+        .clone()
+        .oneshot(common::delete_with_cookie(
+            &format!("/api/v1/bookmarks/{bookmark_id}"),
+            &cookie,
+        ))
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), StatusCode::NO_CONTENT);
+
+    let resp = h
+        .app
+        .clone()
+        .oneshot(common::post_json_with_cookie(
+            "/api/v1/bookmarks",
+            json!({ "manga_id": manga_id.to_string() }),
+            &cookie,
+        ))
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), StatusCode::CREATED);
+
+    // Give the background task time to attempt re-enqueue (it should be a no-op).
+    tokio::time::sleep(std::time::Duration::from_millis(300)).await;
+    let final_count: i64 = sqlx::query_scalar(
+        "SELECT COUNT(*) FROM crawler_jobs \
+          WHERE state IN ('pending', 'running') \
+            AND payload->>'kind' = 'sync_chapter_content'",
+    )
+    .fetch_one(&pool)
+    .await
+    .unwrap();
+    assert_eq!(final_count, 1, "dedup index keeps the queue at a single in-flight row");
+}
+
+#[sqlx::test(migrations = "./migrations")]
+async fn create_skips_chapters_with_dropped_sources(pool: PgPool) {
+    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, "Berserk").await;
+
+    let _alive = seed_chapter_with_source(&pool, manga_id, 1, "target", "ch1", "https://example.com/c1", false).await;
+    let _dropped = seed_chapter_with_source(&pool, manga_id, 2, "target", "ch2", "https://example.com/c2", true).await;
+
+    let resp = h
+        .app
+        .clone()
+        .oneshot(common::post_json_with_cookie(
+            "/api/v1/bookmarks",
+            json!({ "manga_id": manga_id.to_string() }),
+            &cookie,
+        ))
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), StatusCode::CREATED);
+
+    assert_eq!(
+        wait_for_pending_count(&pool, 1).await,
+        1,
+        "only the chapter with a non-dropped source row gets enqueued"
+    );
+}