feat(api): admin manga/chapter overview with derived sync state (0.39.0)

Adds GET /api/v1/admin/mangas and /admin/mangas/:id/chapters guarded by
RequireAdmin. Sync state is computed at query time from the existing
crawler signals (manga_sources / chapter_sources / crawler_jobs) — no
new state column is persisted, so the crawler stays the single writer
of these signals.

Per-manga priority: InProgress (in-flight sync_manga or
sync_chapter_list job) > Dropped (all source rows soft-dropped) >
Synced (default; covers user-uploaded mangas with zero source rows).

Per-chapter priority: Downloading (in-flight sync_chapter_content) >
Dropped (all source rows soft-dropped) > Failed (most-recent terminal
job is dead) > NotDownloaded (page_count = 0) > Synced. The Failed
check sits ABOVE NotDownloaded so the more informative "we tried and
it died" state wins over "we never got around to it" — see the
priority comment in repo/admin_view.rs.

Migration 0020 adds a partial index on
crawler_jobs((payload->>'source_manga_key')) for the one job kind
(sync_manga) whose payload doesn't carry manga_id directly — without it
the in-flight detection for a manga falls back to a seqscan over the
job table.

backend/Cargo.lock

@@ -1470,7 +1470,7 @@ checksum = "c41e0c4fef86961ac6d6f8a82609f55f31b05e4fce149ac5710e439df7619ba4"
 
 [[package]]
 name = "mangalord"
-version = "0.38.0"
+version = "0.39.0"
 dependencies = [
  "anyhow",
  "argon2",

backend/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mangalord"
-version = "0.38.0"
+version = "0.39.0"
 edition = "2021"
 default-run = "mangalord"
 

backend/migrations/0020_admin_jobs_payload_index.sql

@@ -0,0 +1,14 @@
+-- Per-manga sync-state derivation joins crawler_jobs to manga_sources via
+-- (payload->>'source_id', payload->>'source_manga_key') for the
+-- `sync_manga` job kind (whose payload doesn't carry a manga_id directly).
+-- Without this index the join falls back to a seqscan of crawler_jobs on
+-- every admin manga listing — a noticeable cost as the job table grows
+-- with the daily metadata pass.
+--
+-- Partial on `state IN ('pending','running')` so it covers only in-flight
+-- jobs (the bulk of the table is done/dead and irrelevant to "is this
+-- manga being synced right now").
+CREATE INDEX crawler_jobs_sync_manga_key_idx
+    ON crawler_jobs ((payload->>'source_manga_key'))
+ WHERE state IN ('pending', 'running')
+   AND payload->>'kind' = 'sync_manga';

backend/src/api/admin/mangas.rs

@@ -0,0 +1,82 @@
+//! Admin manga/chapter overview with derived sync state.
+//!
+//! Sync state comes from `repo::admin_view`, which joins the manga /
+//! chapter tables with the crawler signals at query time — there is no
+//! persisted sync_state column. See [`repo::admin_view`] for the
+//! derivation priority order.
+
+use axum::extract::{Path, Query, State};
+use axum::routing::get;
+use axum::{Json, Router};
+use serde::Deserialize;
+use uuid::Uuid;
+
+use crate::api::pagination::PagedResponse;
+use crate::app::AppState;
+use crate::auth::extractor::RequireAdmin;
+use crate::domain::MangaSyncState;
+use crate::error::{AppError, AppResult};
+use crate::repo;
+use crate::repo::admin_view::{AdminChapterRow, AdminMangaRow};
+
+pub fn routes() -> Router<AppState> {
+    Router::new()
+        .route("/admin/mangas", get(list_mangas))
+        .route("/admin/mangas/:id/chapters", get(list_chapters))
+}
+
+#[derive(Debug, Deserialize, Default)]
+pub struct ListMangasParams {
+    #[serde(default)]
+    pub search: Option<String>,
+    /// `in_progress` | `dropped` | `synced`. Unrecognised values are a 400.
+    #[serde(default)]
+    pub sync_state: Option<String>,
+    #[serde(default = "default_limit")]
+    pub limit: i64,
+    #[serde(default)]
+    pub offset: i64,
+}
+
+fn default_limit() -> i64 {
+    50
+}
+
+async fn list_mangas(
+    State(state): State<AppState>,
+    _admin: RequireAdmin,
+    Query(params): Query<ListMangasParams>,
+) -> AppResult<Json<PagedResponse<AdminMangaRow>>> {
+    let limit = params.limit.clamp(1, 200);
+    let offset = params.offset.max(0);
+
+    let sync_state = match params.sync_state.as_deref() {
+        None | Some("") => None,
+        Some("in_progress") => Some(MangaSyncState::InProgress),
+        Some("dropped") => Some(MangaSyncState::Dropped),
+        Some("synced") => Some(MangaSyncState::Synced),
+        Some(other) => {
+            return Err(AppError::InvalidInput(format!(
+                "sync_state must be one of in_progress|dropped|synced (got {other:?})"
+            )));
+        }
+    };
+
+    let q = repo::admin_view::ListAdminMangasQuery {
+        search: params.search.filter(|s| !s.trim().is_empty()),
+        sync_state,
+        limit,
+        offset,
+    };
+    let (items, total) = repo::admin_view::list_mangas_with_sync_state(&state.db, &q).await?;
+    Ok(Json(PagedResponse::with_total(items, limit, offset, total)))
+}
+
+async fn list_chapters(
+    State(state): State<AppState>,
+    _admin: RequireAdmin,
+    Path(manga_id): Path<Uuid>,
+) -> AppResult<Json<Vec<AdminChapterRow>>> {
+    let rows = repo::admin_view::list_chapters_with_sync_state(&state.db, manga_id).await?;
+    Ok(Json(rows))
+}

backend/src/api/admin/mod.rs

@@ -4,6 +4,7 @@
 //! bot/API tokens cannot reach admin routes (see
 //! `crate::auth::extractor::RequireAdmin`).
 
+pub mod mangas;
 pub mod users;
 
 use axum::Router;
@@ -11,5 +12,5 @@ use axum::Router;
 use crate::app::AppState;
 
 pub fn routes() -> Router<AppState> {
-    Router::new().merge(users::routes())
+    Router::new().merge(users::routes()).merge(mangas::routes())
 }

backend/src/domain/mod.rs

@@ -10,6 +10,7 @@ pub mod page;
 pub mod patch;
 pub mod read_progress;
 pub mod session;
+pub mod sync_state;
 pub mod tag;
 pub mod upload_entry;
 pub mod user;
@@ -27,6 +28,7 @@ pub use page::Page;
 pub use patch::Patch;
 pub use read_progress::{ReadProgress, ReadProgressForManga, ReadProgressSummary};
 pub use session::Session;
+pub use sync_state::{ChapterSyncState, MangaSyncState};
 pub use tag::{Tag, TagRef};
 pub use upload_entry::UploadEntry;
 pub use user::User;

backend/src/domain/sync_state.rs

@@ -0,0 +1,46 @@
+//! Sync-state enums derived per-manga / per-chapter from `manga_sources`,
+//! `chapter_sources`, and `crawler_jobs` at query time. No state column
+//! is persisted on `mangas` / `chapters` — see `repo::admin_view` for the
+//! derivation rules and priority order.
+
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, sqlx::Type)]
+#[sqlx(type_name = "text", rename_all = "snake_case")]
+#[serde(rename_all = "snake_case")]
+pub enum MangaSyncState {
+    /// A `sync_manga` or `sync_chapter_list` job is currently
+    /// pending or running for this manga.
+    InProgress,
+    /// At least one `manga_sources` row exists for this manga and ALL of
+    /// them have `dropped_at IS NOT NULL` — every source we know about
+    /// has stopped surfacing it.
+    Dropped,
+    /// Default healthy state: at least one live source row OR the manga
+    /// was user-uploaded (no `manga_sources` rows at all).
+    Synced,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, sqlx::Type)]
+#[sqlx(type_name = "text", rename_all = "snake_case")]
+#[serde(rename_all = "snake_case")]
+pub enum ChapterSyncState {
+    /// A `sync_chapter_content` job is currently pending or running for
+    /// this chapter (the 0014 dedup index guarantees at most one).
+    Downloading,
+    /// At least one `chapter_sources` row exists AND all of them are
+    /// `dropped_at IS NOT NULL`.
+    Dropped,
+    /// Most recent `sync_chapter_content` job for this chapter is `dead`
+    /// (terminal failure). Checked BEFORE `NotDownloaded` so the more
+    /// informative "we tried and it died" state wins over "we never
+    /// got around to it".
+    Failed,
+    /// `page_count = 0` and no in-flight or failed job — the chapter
+    /// row exists but content has never been downloaded.
+    NotDownloaded,
+    /// `page_count > 0` — content has been downloaded at some point.
+    /// Reaped `done` jobs in `crawler_jobs` mean we can't read this from
+    /// the job table, so `page_count` is the durable truth.
+    Synced,
+}

backend/src/repo/admin_view.rs

@@ -0,0 +1,204 @@
+//! Admin-facing read queries that join manga/chapter with the crawler
+//! signals (`manga_sources`, `chapter_sources`, `crawler_jobs`) to
+//! derive a sync state per row at query time.
+//!
+//! Priority order for `MangaSyncState`:
+//!   1. `InProgress` — any pending/running `sync_manga` or
+//!      `sync_chapter_list` job matches this manga.
+//!   2. `Dropped` — manga has source rows AND every one of them is
+//!      `dropped_at IS NOT NULL`.
+//!   3. `Synced` — default (includes user-uploaded mangas with no
+//!      `manga_sources` rows at all).
+//!
+//! Priority order for `ChapterSyncState`:
+//!   1. `Downloading` — pending/running `sync_chapter_content` for this id
+//!   2. `Dropped` — chapter has source rows AND all are dropped
+//!   3. `Failed` — most recent terminal `sync_chapter_content` is `dead`
+//!   4. `NotDownloaded` — `page_count = 0`
+//!   5. `Synced` — `page_count > 0`
+//!
+//! Reminder: `done` jobs are reaped after `CRAWLER_JOB_RETENTION_DAYS`,
+//! so `chapters.page_count > 0` is the durable "this is synced" signal,
+//! not the job table.
+
+use chrono::{DateTime, Utc};
+use serde::Serialize;
+use sqlx::{FromRow, PgPool};
+use uuid::Uuid;
+
+use crate::domain::{ChapterSyncState, MangaSyncState};
+use crate::error::AppResult;
+
+#[derive(Debug, Serialize, FromRow)]
+pub struct AdminMangaRow {
+    pub id: Uuid,
+    pub title: String,
+    pub status: String,
+    pub cover_image_path: Option<String>,
+    pub created_at: DateTime<Utc>,
+    pub updated_at: DateTime<Utc>,
+    pub sync_state: MangaSyncState,
+    pub chapter_count: i64,
+    pub latest_seen_at: Option<DateTime<Utc>>,
+}
+
+#[derive(Debug, Default)]
+pub struct ListAdminMangasQuery {
+    pub search: Option<String>,
+    pub sync_state: Option<MangaSyncState>,
+    pub limit: i64,
+    pub offset: i64,
+}
+
+const MANGA_SYNC_STATE_CASE: &str = r#"
+    CASE
+      WHEN EXISTS (
+        SELECT 1 FROM crawler_jobs cj
+         WHERE cj.state IN ('pending','running')
+           AND (
+                (cj.payload->>'kind' = 'sync_chapter_list'
+                 AND (cj.payload->>'manga_id')::uuid = m.id)
+             OR (cj.payload->>'kind' = 'sync_manga'
+                 AND EXISTS (
+                       SELECT 1 FROM manga_sources ms
+                        WHERE ms.manga_id = m.id
+                          AND ms.source_id = cj.payload->>'source_id'
+                          AND ms.source_manga_key = cj.payload->>'source_manga_key'
+                     ))
+              )
+      ) THEN 'in_progress'
+      WHEN EXISTS (SELECT 1 FROM manga_sources ms WHERE ms.manga_id = m.id)
+       AND NOT EXISTS (
+             SELECT 1 FROM manga_sources ms
+              WHERE ms.manga_id = m.id AND ms.dropped_at IS NULL
+           )
+      THEN 'dropped'
+      ELSE 'synced'
+    END
+"#;
+
+/// Paginated admin manga list with derived sync state and total count.
+/// Filters by `search` (substring on title, case-insensitive) and
+/// `sync_state` (post-derivation). The CTE keeps the case expression
+/// in one place — the same projection feeds both the page rows and the
+/// totals count under the same filter.
+pub async fn list_mangas_with_sync_state(
+    pool: &PgPool,
+    q: &ListAdminMangasQuery,
+) -> AppResult<(Vec<AdminMangaRow>, i64)> {
+    let search_pat = q
+        .search
+        .as_ref()
+        .map(|s| format!("%{}%", s.trim()))
+        .filter(|p| p.len() > 2);
+    // sqlx::Type → text: bind the snake_case representation manually so
+    // the SQL can compare it as text without an explicit cast.
+    let sync_filter = q.sync_state.map(|s| match s {
+        MangaSyncState::InProgress => "in_progress",
+        MangaSyncState::Dropped => "dropped",
+        MangaSyncState::Synced => "synced",
+    });
+
+    let sql = format!(
+        r#"
+        WITH classified AS (
+            SELECT
+                m.id, m.title, m.status, m.cover_image_path,
+                m.created_at, m.updated_at,
+                {case} AS sync_state,
+                (SELECT COUNT(*) FROM chapters c WHERE c.manga_id = m.id) AS chapter_count,
+                (SELECT MAX(last_seen_at) FROM manga_sources ms
+                  WHERE ms.manga_id = m.id AND ms.dropped_at IS NULL) AS latest_seen_at
+              FROM mangas m
+             WHERE ($1::text IS NULL OR m.title ILIKE $1)
+        )
+        SELECT * FROM classified
+         WHERE ($2::text IS NULL OR sync_state = $2)
+         ORDER BY updated_at DESC
+         LIMIT $3 OFFSET $4
+        "#,
+        case = MANGA_SYNC_STATE_CASE
+    );
+    let items: Vec<AdminMangaRow> = sqlx::query_as(&sql)
+        .bind(&search_pat)
+        .bind(sync_filter)
+        .bind(q.limit)
+        .bind(q.offset)
+        .fetch_all(pool)
+        .await?;
+
+    let total_sql = format!(
+        r#"
+        WITH classified AS (
+            SELECT {case} AS sync_state
+              FROM mangas m
+             WHERE ($1::text IS NULL OR m.title ILIKE $1)
+        )
+        SELECT COUNT(*) FROM classified
+         WHERE ($2::text IS NULL OR sync_state = $2)
+        "#,
+        case = MANGA_SYNC_STATE_CASE
+    );
+    let total: i64 = sqlx::query_scalar(&total_sql)
+        .bind(&search_pat)
+        .bind(sync_filter)
+        .fetch_one(pool)
+        .await?;
+
+    Ok((items, total))
+}
+
+#[derive(Debug, Serialize, FromRow)]
+pub struct AdminChapterRow {
+    pub id: Uuid,
+    pub manga_id: Uuid,
+    pub number: i32,
+    pub title: Option<String>,
+    pub page_count: i32,
+    pub created_at: DateTime<Utc>,
+    pub sync_state: ChapterSyncState,
+    pub latest_seen_at: Option<DateTime<Utc>>,
+}
+
+pub async fn list_chapters_with_sync_state(
+    pool: &PgPool,
+    manga_id: Uuid,
+) -> AppResult<Vec<AdminChapterRow>> {
+    let rows: Vec<AdminChapterRow> = sqlx::query_as(
+        r#"
+        SELECT
+            c.id, c.manga_id, c.number, c.title, c.page_count, c.created_at,
+            CASE
+              WHEN EXISTS (
+                SELECT 1 FROM crawler_jobs cj
+                 WHERE cj.state IN ('pending','running')
+                   AND cj.payload->>'kind' = 'sync_chapter_content'
+                   AND (cj.payload->>'chapter_id')::uuid = c.id
+              ) THEN 'downloading'
+              WHEN EXISTS (SELECT 1 FROM chapter_sources cs WHERE cs.chapter_id = c.id)
+               AND NOT EXISTS (
+                     SELECT 1 FROM chapter_sources cs
+                      WHERE cs.chapter_id = c.id AND cs.dropped_at IS NULL
+                   )
+              THEN 'dropped'
+              WHEN EXISTS (
+                SELECT 1 FROM crawler_jobs cj
+                 WHERE cj.state = 'dead'
+                   AND cj.payload->>'kind' = 'sync_chapter_content'
+                   AND (cj.payload->>'chapter_id')::uuid = c.id
+              ) THEN 'failed'
+              WHEN c.page_count = 0 THEN 'not_downloaded'
+              ELSE 'synced'
+            END AS sync_state,
+            (SELECT MAX(last_seen_at) FROM chapter_sources cs
+              WHERE cs.chapter_id = c.id AND cs.dropped_at IS NULL) AS latest_seen_at
+          FROM chapters c
+         WHERE c.manga_id = $1
+         ORDER BY c.number ASC
+        "#,
+    )
+    .bind(manga_id)
+    .fetch_all(pool)
+    .await?;
+    Ok(rows)
+}

backend/src/repo/mod.rs

@@ -1,4 +1,5 @@
 pub mod admin_audit;
+pub mod admin_view;
 pub mod api_token;
 pub mod author;
 pub mod bookmark;

backend/tests/api_admin_mangas.rs

@@ -0,0 +1,436 @@
+//! PR 3 (feat/admin-mangas-api) integration tests.
+//!
+//! Per-variant fixture tests for the derived sync-state SQL plus
+//! happy-path E2E for the two admin endpoints. Auth-gate regression
+//! (403/401) is covered by PR 1's `RequireAdmin` test matrix; the only
+//! gate test here is one spot check per endpoint.
+
+mod common;
+
+use axum::http::StatusCode;
+use axum::Router;
+use serde_json::json;
+use sqlx::PgPool;
+use tower::ServiceExt;
+use uuid::Uuid;
+
+use mangalord::repo;
+
+const SOURCE_ID: &str = "test-source";
+
+async fn seed_admin(pool: &PgPool, app: &Router) -> (String, String) {
+    let (username, cookie) = common::register_user(app).await;
+    let u = repo::user::find_by_username(pool, &username)
+        .await
+        .unwrap()
+        .unwrap();
+    repo::user::set_is_admin(pool, u.id, true).await.unwrap();
+    (username, cookie)
+}
+
+async fn seed_source(pool: &PgPool) {
+    repo::crawler::ensure_source(pool, SOURCE_ID, "Test", "https://example.test")
+        .await
+        .unwrap();
+}
+
+async fn insert_manga(pool: &PgPool, title: &str) -> Uuid {
+    let (id,): (Uuid,) = sqlx::query_as(
+        "INSERT INTO mangas (title, status, alt_titles) VALUES ($1, 'ongoing', ARRAY[]::text[]) RETURNING id",
+    )
+    .bind(title)
+    .fetch_one(pool)
+    .await
+    .unwrap();
+    id
+}
+
+async fn insert_manga_source(
+    pool: &PgPool,
+    manga_id: Uuid,
+    source_manga_key: &str,
+    dropped: bool,
+) {
+    let dropped_at = if dropped { "now()" } else { "NULL" };
+    let sql = format!(
+        "INSERT INTO manga_sources (source_id, source_manga_key, manga_id, source_url, dropped_at) \
+         VALUES ($1, $2, $3, 'https://example.test/m', {dropped_at})"
+    );
+    sqlx::query(&sql)
+        .bind(SOURCE_ID)
+        .bind(source_manga_key)
+        .bind(manga_id)
+        .execute(pool)
+        .await
+        .unwrap();
+}
+
+async fn insert_chapter(pool: &PgPool, manga_id: Uuid, number: i32, page_count: i32) -> Uuid {
+    let (id,): (Uuid,) = sqlx::query_as(
+        "INSERT INTO chapters (manga_id, number, title, page_count) VALUES ($1, $2, NULL, $3) RETURNING id",
+    )
+    .bind(manga_id)
+    .bind(number)
+    .bind(page_count)
+    .fetch_one(pool)
+    .await
+    .unwrap();
+    id
+}
+
+async fn insert_chapter_source(
+    pool: &PgPool,
+    chapter_id: Uuid,
+    source_chapter_key: &str,
+    dropped: bool,
+) {
+    let dropped_at = if dropped { "now()" } else { "NULL" };
+    let sql = format!(
+        "INSERT INTO chapter_sources (source_id, source_chapter_key, chapter_id, source_url, dropped_at) \
+         VALUES ($1, $2, $3, 'https://example.test/c', {dropped_at})"
+    );
+    sqlx::query(&sql)
+        .bind(SOURCE_ID)
+        .bind(source_chapter_key)
+        .bind(chapter_id)
+        .execute(pool)
+        .await
+        .unwrap();
+}
+
+async fn insert_job(pool: &PgPool, payload: serde_json::Value, state: &str) {
+    sqlx::query("INSERT INTO crawler_jobs (payload, state) VALUES ($1, $2)")
+        .bind(payload)
+        .bind(state)
+        .execute(pool)
+        .await
+        .unwrap();
+}
+
+// ---- manga sync state ------------------------------------------------------
+
+#[sqlx::test(migrations = "./migrations")]
+async fn manga_state_synced_for_fresh_source(pool: PgPool) {
+    seed_source(&pool).await;
+    let m = insert_manga(&pool, "Synced Manga").await;
+    insert_manga_source(&pool, m, "smk-1", false).await;
+
+    let (rows, total) = repo::admin_view::list_mangas_with_sync_state(
+        &pool,
+        &repo::admin_view::ListAdminMangasQuery {
+            limit: 50,
+            ..Default::default()
+        },
+    )
+    .await
+    .unwrap();
+    assert_eq!(total, 1);
+    assert_eq!(rows[0].id, m);
+    assert_eq!(rows[0].sync_state, mangalord::domain::MangaSyncState::Synced);
+}
+
+#[sqlx::test(migrations = "./migrations")]
+async fn manga_state_synced_for_user_upload_without_sources(pool: PgPool) {
+    let m = insert_manga(&pool, "User Upload").await;
+    let (rows, _) = repo::admin_view::list_mangas_with_sync_state(
+        &pool,
+        &repo::admin_view::ListAdminMangasQuery {
+            limit: 50,
+            ..Default::default()
+        },
+    )
+    .await
+    .unwrap();
+    assert_eq!(rows[0].id, m);
+    assert_eq!(rows[0].sync_state, mangalord::domain::MangaSyncState::Synced);
+}
+
+#[sqlx::test(migrations = "./migrations")]
+async fn manga_state_dropped_when_all_sources_dropped(pool: PgPool) {
+    seed_source(&pool).await;
+    let m = insert_manga(&pool, "Dropped Manga").await;
+    insert_manga_source(&pool, m, "smk-1", true).await;
+
+    let (rows, _) = repo::admin_view::list_mangas_with_sync_state(
+        &pool,
+        &repo::admin_view::ListAdminMangasQuery {
+            limit: 50,
+            ..Default::default()
+        },
+    )
+    .await
+    .unwrap();
+    assert_eq!(rows[0].id, m);
+    assert_eq!(rows[0].sync_state, mangalord::domain::MangaSyncState::Dropped);
+}
+
+#[sqlx::test(migrations = "./migrations")]
+async fn manga_state_in_progress_via_sync_chapter_list_job(pool: PgPool) {
+    seed_source(&pool).await;
+    let m = insert_manga(&pool, "Syncing Manga").await;
+    insert_manga_source(&pool, m, "smk-1", false).await;
+    // sync_chapter_list payload carries manga_id directly.
+    insert_job(
+        &pool,
+        json!({
+            "kind": "sync_chapter_list",
+            "source_id": SOURCE_ID,
+            "manga_id": m.to_string(),
+            "source_manga_key": "smk-1",
+        }),
+        "pending",
+    )
+    .await;
+
+    let (rows, _) = repo::admin_view::list_mangas_with_sync_state(
+        &pool,
+        &repo::admin_view::ListAdminMangasQuery {
+            limit: 50,
+            ..Default::default()
+        },
+    )
+    .await
+    .unwrap();
+    assert_eq!(rows[0].sync_state, mangalord::domain::MangaSyncState::InProgress);
+}
+
+#[sqlx::test(migrations = "./migrations")]
+async fn manga_state_in_progress_via_sync_manga_job(pool: PgPool) {
+    // The trickier branch: sync_manga payload is keyed by
+    // source_manga_key, NOT manga_id — must join through manga_sources.
+    seed_source(&pool).await;
+    let m = insert_manga(&pool, "Metadata-Refreshing Manga").await;
+    insert_manga_source(&pool, m, "smk-key-42", false).await;
+    insert_job(
+        &pool,
+        json!({
+            "kind": "sync_manga",
+            "source_id": SOURCE_ID,
+            "source_manga_key": "smk-key-42",
+        }),
+        "running",
+    )
+    .await;
+
+    let (rows, _) = repo::admin_view::list_mangas_with_sync_state(
+        &pool,
+        &repo::admin_view::ListAdminMangasQuery {
+            limit: 50,
+            ..Default::default()
+        },
+    )
+    .await
+    .unwrap();
+    assert_eq!(rows[0].sync_state, mangalord::domain::MangaSyncState::InProgress);
+}
+
+#[sqlx::test(migrations = "./migrations")]
+async fn manga_list_filters_by_sync_state(pool: PgPool) {
+    seed_source(&pool).await;
+    let m_synced = insert_manga(&pool, "AAA Synced").await;
+    insert_manga_source(&pool, m_synced, "smk-a", false).await;
+    let m_dropped = insert_manga(&pool, "BBB Dropped").await;
+    insert_manga_source(&pool, m_dropped, "smk-b", true).await;
+
+    let (rows, total) = repo::admin_view::list_mangas_with_sync_state(
+        &pool,
+        &repo::admin_view::ListAdminMangasQuery {
+            sync_state: Some(mangalord::domain::MangaSyncState::Dropped),
+            limit: 50,
+            ..Default::default()
+        },
+    )
+    .await
+    .unwrap();
+    assert_eq!(total, 1);
+    assert_eq!(rows.len(), 1);
+    assert_eq!(rows[0].id, m_dropped);
+}
+
+// ---- chapter sync state ----------------------------------------------------
+
+#[sqlx::test(migrations = "./migrations")]
+async fn chapter_state_synced_when_pages_present(pool: PgPool) {
+    seed_source(&pool).await;
+    let m = insert_manga(&pool, "M").await;
+    insert_manga_source(&pool, m, "smk", false).await;
+    let c = insert_chapter(&pool, m, 1, 12).await;
+    insert_chapter_source(&pool, c, "ckey-1", false).await;
+
+    let rows = repo::admin_view::list_chapters_with_sync_state(&pool, m)
+        .await
+        .unwrap();
+    assert_eq!(rows.len(), 1);
+    assert_eq!(rows[0].id, c);
+    assert_eq!(rows[0].sync_state, mangalord::domain::ChapterSyncState::Synced);
+}
+
+#[sqlx::test(migrations = "./migrations")]
+async fn chapter_state_not_downloaded_when_page_count_zero(pool: PgPool) {
+    seed_source(&pool).await;
+    let m = insert_manga(&pool, "M").await;
+    let c = insert_chapter(&pool, m, 1, 0).await;
+    insert_chapter_source(&pool, c, "ckey-1", false).await;
+
+    let rows = repo::admin_view::list_chapters_with_sync_state(&pool, m)
+        .await
+        .unwrap();
+    assert_eq!(
+        rows[0].sync_state,
+        mangalord::domain::ChapterSyncState::NotDownloaded
+    );
+}
+
+#[sqlx::test(migrations = "./migrations")]
+async fn chapter_state_downloading_when_job_in_flight(pool: PgPool) {
+    seed_source(&pool).await;
+    let m = insert_manga(&pool, "M").await;
+    let c = insert_chapter(&pool, m, 1, 0).await;
+    insert_chapter_source(&pool, c, "ckey-1", false).await;
+    insert_job(
+        &pool,
+        json!({
+            "kind": "sync_chapter_content",
+            "source_id": SOURCE_ID,
+            "chapter_id": c.to_string(),
+            "source_chapter_key": "ckey-1",
+        }),
+        "running",
+    )
+    .await;
+
+    let rows = repo::admin_view::list_chapters_with_sync_state(&pool, m)
+        .await
+        .unwrap();
+    assert_eq!(
+        rows[0].sync_state,
+        mangalord::domain::ChapterSyncState::Downloading
+    );
+}
+
+#[sqlx::test(migrations = "./migrations")]
+async fn chapter_state_dropped_when_all_sources_dropped(pool: PgPool) {
+    seed_source(&pool).await;
+    let m = insert_manga(&pool, "M").await;
+    let c = insert_chapter(&pool, m, 1, 0).await;
+    insert_chapter_source(&pool, c, "ckey-1", true).await;
+
+    let rows = repo::admin_view::list_chapters_with_sync_state(&pool, m)
+        .await
+        .unwrap();
+    assert_eq!(
+        rows[0].sync_state,
+        mangalord::domain::ChapterSyncState::Dropped
+    );
+}
+
+#[sqlx::test(migrations = "./migrations")]
+async fn chapter_state_failed_when_most_recent_job_dead(pool: PgPool) {
+    seed_source(&pool).await;
+    let m = insert_manga(&pool, "M").await;
+    let c = insert_chapter(&pool, m, 1, 0).await;
+    insert_chapter_source(&pool, c, "ckey-1", false).await;
+    insert_job(
+        &pool,
+        json!({
+            "kind": "sync_chapter_content",
+            "source_id": SOURCE_ID,
+            "chapter_id": c.to_string(),
+            "source_chapter_key": "ckey-1",
+        }),
+        "dead",
+    )
+    .await;
+
+    let rows = repo::admin_view::list_chapters_with_sync_state(&pool, m)
+        .await
+        .unwrap();
+    assert_eq!(
+        rows[0].sync_state,
+        mangalord::domain::ChapterSyncState::Failed
+    );
+}
+
+// ---- HTTP-level happy-path + gate ------------------------------------------
+
+#[sqlx::test(migrations = "./migrations")]
+async fn http_list_mangas_returns_paged_with_state(pool: PgPool) {
+    let h = common::harness(pool.clone());
+    let (_admin, cookie) = seed_admin(&pool, &h.app).await;
+    seed_source(&pool).await;
+    let m = insert_manga(&pool, "Hello").await;
+    insert_manga_source(&pool, m, "smk", false).await;
+
+    let resp = h
+        .app
+        .oneshot(common::get_with_cookie(
+            "/api/v1/admin/mangas?limit=50",
+            &cookie,
+        ))
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), StatusCode::OK);
+    let body = common::body_json(resp).await;
+    let items = body["items"].as_array().unwrap();
+    assert_eq!(items.len(), 1);
+    assert_eq!(items[0]["id"], m.to_string());
+    assert_eq!(items[0]["sync_state"], "synced");
+    assert_eq!(items[0]["chapter_count"], 0);
+    assert_eq!(body["page"]["total"], 1);
+}
+
+#[sqlx::test(migrations = "./migrations")]
+async fn http_list_mangas_rejects_unknown_sync_state(pool: PgPool) {
+    let h = common::harness(pool.clone());
+    let (_admin, cookie) = seed_admin(&pool, &h.app).await;
+    let resp = h
+        .app
+        .oneshot(common::get_with_cookie(
+            "/api/v1/admin/mangas?sync_state=bogus",
+            &cookie,
+        ))
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
+}
+
+#[sqlx::test(migrations = "./migrations")]
+async fn http_list_chapters_returns_per_chapter_state(pool: PgPool) {
+    let h = common::harness(pool.clone());
+    let (_admin, cookie) = seed_admin(&pool, &h.app).await;
+    seed_source(&pool).await;
+    let m = insert_manga(&pool, "M").await;
+    let c1 = insert_chapter(&pool, m, 1, 12).await;
+    let c2 = insert_chapter(&pool, m, 2, 0).await;
+    insert_chapter_source(&pool, c1, "ck1", false).await;
+    insert_chapter_source(&pool, c2, "ck2", false).await;
+
+    let resp = h
+        .app
+        .oneshot(common::get_with_cookie(
+            &format!("/api/v1/admin/mangas/{m}/chapters"),
+            &cookie,
+        ))
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), StatusCode::OK);
+    let body = common::body_json(resp).await;
+    let items = body.as_array().unwrap();
+    assert_eq!(items.len(), 2);
+    assert_eq!(items[0]["id"], c1.to_string());
+    assert_eq!(items[0]["sync_state"], "synced");
+    assert_eq!(items[1]["id"], c2.to_string());
+    assert_eq!(items[1]["sync_state"], "not_downloaded");
+}
+
+#[sqlx::test(migrations = "./migrations")]
+async fn http_list_mangas_requires_admin(pool: PgPool) {
+    let h = common::harness(pool);
+    let (_u, cookie) = common::register_user(&h.app).await;
+    let resp = h
+        .app
+        .oneshot(common::get_with_cookie("/api/v1/admin/mangas", &cookie))
+        .await
+        .unwrap();
+    assert_eq!(resp.status(), StatusCode::FORBIDDEN);
+}