"""FC-3k: /api/admin/* endpoint integration tests.

Monkeypatch `.delay()` for Tier-C tasks to capture dispatch payloads
without actually queuing. Tier-B/A endpoints run synchronously
through the real service.
"""
import hashlib
from types import SimpleNamespace

import pytest

from backend.app.models import Artist, ImageRecord, Post, Tag, TagKind
from backend.app.models.tag import image_tag

pytestmark = pytest.mark.integration


# --- Tier-C: POST /artists/<slug>/cascade-delete --------------------


@pytest.mark.asyncio
async def test_artist_cascade_dry_run_returns_projection(client, db):
    a = Artist(name="Aria", slug="aria")
    db.add(a)
    await db.commit()

    resp = await client.post(
        "/api/admin/artists/aria/cascade-delete",
        json={"dry_run": True},
    )
    assert resp.status_code == 200
    body = await resp.get_json()
    assert body["artist"]["slug"] == "aria"
    assert "projected" in body
    assert body["projected"]["images"] == 0


@pytest.mark.asyncio
async def test_artist_cascade_unknown_slug_404(client):
    resp = await client.post(
        "/api/admin/artists/no-such/cascade-delete",
        json={"dry_run": True},
    )
    assert resp.status_code == 404


@pytest.mark.asyncio
async def test_artist_cascade_wrong_confirm_400(client, db):
    a = Artist(name="Aria", slug="aria")
    db.add(a)
    await db.commit()
    artist_id = a.id

    resp = await client.post(
        "/api/admin/artists/aria/cascade-delete",
        json={"dry_run": False, "confirm": "wrong"},
    )
    assert resp.status_code == 400
    body = await resp.get_json()
    assert body["error"] == "confirm_mismatch"
    assert body["expected"] == f"delete-artist-{artist_id}"


@pytest.mark.asyncio
async def test_artist_cascade_correct_confirm_dispatches(
    client, db, monkeypatch,
):
    a = Artist(name="Aria", slug="aria")
    db.add(a)
    await db.commit()
    artist_id = a.id

    dispatched = []

    class _Result:
        id = "fake-task-id"

    def _delay(**kw):
        dispatched.append(kw)
        return _Result()

    monkeypatch.setattr(
        "backend.app.tasks.admin.delete_artist_cascade_task.delay", _delay,
    )

    resp = await client.post(
        "/api/admin/artists/aria/cascade-delete",
        json={
            "dry_run": False,
            "confirm": f"delete-artist-{artist_id}",
        },
    )
    assert resp.status_code == 202
    body = await resp.get_json()
    assert body["task_id"] == "fake-task-id"
    assert dispatched == [{"artist_id": artist_id}]


# --- Tier-C: POST /images/bulk-delete -------------------------------


@pytest.mark.asyncio
async def test_bulk_delete_dry_run_returns_counts(client, db, tmp_path):
    a = Artist(name="B", slug="b")
    db.add(a)
    await db.flush()
    i1 = ImageRecord(
        artist_id=a.id, path=str(tmp_path / "1.jpg"),
        sha256="1" * 64, size_bytes=10, mime="image/jpeg",
        origin="imported_filesystem",
    )
    i2 = ImageRecord(
        artist_id=a.id, path=str(tmp_path / "2.jpg"),
        sha256="2" * 64, size_bytes=20, mime="image/jpeg",
        origin="imported_filesystem",
    )
    db.add_all([i1, i2])
    await db.commit()

    resp = await client.post(
        "/api/admin/images/bulk-delete",
        json={
            "image_ids": [i1.id, i2.id, 9_999_999],
            "dry_run": True,
        },
    )
    assert resp.status_code == 200
    body = await resp.get_json()
    assert body["images_found"] == 2
    assert body["bytes_on_disk"] == 30
    assert body["missing_ids"] == [9_999_999]
    # Dry-run hands the canonical Tier-C confirm token back so the
    # frontend doesn't recompute SHA-256 client-side (crypto.subtle
    # is Secure-Context-gated; FC runs over plain HTTP).
    assert body["confirm_token"].startswith("delete-images-")
    assert len(body["confirm_token"]) == len("delete-images-") + 8


@pytest.mark.asyncio
async def test_bulk_delete_empty_list_400(client):
    resp = await client.post(
        "/api/admin/images/bulk-delete",
        json={"image_ids": [], "dry_run": True},
    )
    assert resp.status_code == 400
    body = await resp.get_json()
    assert body["error"] == "invalid_image_ids"


@pytest.mark.asyncio
async def test_bulk_delete_non_int_400(client):
    resp = await client.post(
        "/api/admin/images/bulk-delete",
        json={"image_ids": ["foo", 2], "dry_run": True},
    )
    assert resp.status_code == 400


@pytest.mark.asyncio
async def test_bulk_delete_wrong_confirm_400_with_expected_token(
    client, db, tmp_path,
):
    a = Artist(name="B", slug="b")
    db.add(a)
    await db.flush()
    img = ImageRecord(
        artist_id=a.id, path=str(tmp_path / "x.jpg"),
        sha256="c" * 64, size_bytes=10, mime="image/jpeg",
        origin="imported_filesystem",
    )
    db.add(img)
    await db.commit()
    img_id = img.id

    resp = await client.post(
        "/api/admin/images/bulk-delete",
        json={
            "image_ids": [img_id], "dry_run": False, "confirm": "nope",
        },
    )
    assert resp.status_code == 400
    body = await resp.get_json()
    assert body["error"] == "confirm_mismatch"
    assert body["expected"].startswith("delete-images-")


@pytest.mark.asyncio
async def test_bulk_delete_correct_confirm_dispatches(
    client, db, tmp_path, monkeypatch,
):
    a = Artist(name="B", slug="b")
    db.add(a)
    await db.flush()
    img = ImageRecord(
        artist_id=a.id, path=str(tmp_path / "x.jpg"),
        sha256="d" * 64, size_bytes=10, mime="image/jpeg",
        origin="imported_filesystem",
    )
    db.add(img)
    await db.commit()
    img_id = img.id

    sha8 = hashlib.sha256(str(img_id).encode("utf-8")).hexdigest()[:8]

    dispatched = []

    class _Result:
        id = "fake-bulk-id"

    def _delay(**kw):
        dispatched.append(kw)
        return _Result()

    monkeypatch.setattr(
        "backend.app.tasks.admin.bulk_delete_images_task.delay", _delay,
    )

    resp = await client.post(
        "/api/admin/images/bulk-delete",
        json={
            "image_ids": [img_id],
            "dry_run": False,
            "confirm": f"delete-images-{sha8}",
        },
    )
    assert resp.status_code == 202
    body = await resp.get_json()
    assert body["task_id"] == "fake-bulk-id"
    assert dispatched == [{"image_ids": [img_id]}]


# --- Tier-B: DELETE /tags/<id> + POST /tags/<dest>/merge -----------


@pytest.mark.asyncio
async def test_tag_delete_200_after_cascade(client, db):
    t = Tag(name="doomed-tag", kind=TagKind.general)
    db.add(t)
    await db.commit()
    tag_id = t.id

    resp = await client.delete(f"/api/admin/tags/{tag_id}")
    assert resp.status_code == 200
    body = await resp.get_json()
    assert body["deleted"]["id"] == tag_id
    assert body["associations_removed"] == 0


@pytest.mark.asyncio
async def test_tag_delete_404_unknown(client):
    resp = await client.delete("/api/admin/tags/9999999")
    assert resp.status_code == 404


@pytest.mark.asyncio
async def test_tag_merge_succeeds_for_same_kind(client, db):
    src = Tag(name="src-tag", kind=TagKind.general)
    dest = Tag(name="dest-tag", kind=TagKind.general)
    db.add_all([src, dest])
    await db.commit()
    src_id = src.id
    dest_id = dest.id

    resp = await client.post(
        f"/api/admin/tags/{dest_id}/merge",
        json={"source_id": src_id},
    )
    assert resp.status_code == 200
    body = await resp.get_json()
    assert body["result"]["target_id"] == dest_id


@pytest.mark.asyncio
async def test_tag_merge_dry_run_previews_without_mutating(client, db):
    from backend.app.models import ImageRecord
    from backend.app.models.tag import image_tag

    src = Tag(name="dry-src", kind=TagKind.general)
    dest = Tag(name="dry-dest", kind=TagKind.general)
    db.add_all([src, dest])
    await db.flush()
    img = ImageRecord(
        path="/images/dry.jpg", sha256="dry" + "0" * 61, size_bytes=1,
        mime="image/jpeg", width=1, height=1,
        origin="imported_filesystem", integrity_status="unknown",
    )
    db.add(img)
    await db.flush()
    await db.execute(image_tag.insert().values(
        image_record_id=img.id, tag_id=src.id, source="manual",
    ))
    src_id, dest_id = src.id, dest.id
    await db.commit()

    resp = await client.post(
        f"/api/admin/tags/{dest_id}/merge",
        json={"source_id": src_id, "dry_run": True},
    )
    assert resp.status_code == 200
    p = (await resp.get_json())["preview"]
    assert p["compatible"] is True
    assert p["images_moving"] == 1
    assert p["source_total"] == 1
    # No mutation: source still exists.
    assert await db.get(Tag, src_id) is not None


@pytest.mark.asyncio
async def test_tag_merge_rejects_self_merge(client, db):
    t = Tag(name="x", kind=TagKind.general)
    db.add(t)
    await db.commit()
    resp = await client.post(
        f"/api/admin/tags/{t.id}/merge",
        json={"source_id": t.id},
    )
    assert resp.status_code == 400


@pytest.mark.asyncio
async def test_tag_merge_returns_400_on_kind_mismatch(client, db):
    src = Tag(name="src", kind=TagKind.general)
    dest = Tag(name="dest", kind=TagKind.artist)
    db.add_all([src, dest])
    await db.commit()
    resp = await client.post(
        f"/api/admin/tags/{dest.id}/merge",
        json={"source_id": src.id},
    )
    assert resp.status_code == 400
    body = await resp.get_json()
    assert body["error"] == "tag_kind_mismatch"


# --- Tier-B helper: GET /tags/<id>/usage-count ----------------------


@pytest.mark.asyncio
async def test_tag_usage_count_returns_zero_for_unused(client, db):
    t = Tag(name="lonely", kind=TagKind.general)
    db.add(t)
    await db.commit()
    resp = await client.get(f"/api/admin/tags/{t.id}/usage-count")
    assert resp.status_code == 200
    body = await resp.get_json()
    assert body["count"] == 0


# --- Tier-A: POST /tags/prune-unused --------------------------------


@pytest.mark.asyncio
async def test_prune_unused_dry_run_lists_unused(client, db, tmp_path):
    a = Artist(name="P", slug="p")
    db.add(a)
    await db.flush()
    img = ImageRecord(
        artist_id=a.id, path=str(tmp_path / "p.jpg"),
        sha256="e" * 64, size_bytes=10, mime="image/jpeg",
        origin="imported_filesystem",
    )
    db.add(img)
    used = Tag(name="kept", kind=TagKind.general)
    unused = Tag(name="prune", kind=TagKind.general)
    db.add_all([used, unused])
    await db.flush()
    await db.execute(image_tag.insert().values(
        image_record_id=img.id, tag_id=used.id,
    ))
    await db.commit()

    resp = await client.post(
        "/api/admin/tags/prune-unused", json={"dry_run": True},
    )
    body = await resp.get_json()
    assert body["count"] == 1
    assert "prune" in body["sample_names"]


@pytest.mark.asyncio
async def test_prune_unused_commit_deletes_and_returns_count(client, db):
    db.add(Tag(name="byebye", kind=TagKind.general))
    await db.commit()

    resp = await client.post(
        "/api/admin/tags/prune-unused", json={"dry_run": False},
    )
    body = await resp.get_json()
    assert body["deleted"] >= 1
    assert "byebye" in body["sample_names"]


# --- Tier-A: POST /posts/refetch-external ----------------------------


@pytest.mark.asyncio
async def test_refetch_external_resets_and_dispatches(client, db, monkeypatch):
    """Surgical recovery for deep posts (2026-07-03): settled links reset to
    pending with a fresh budget + dispatched; in-flight links untouched."""
    from sqlalchemy import select

    from backend.app.models import ExternalLink, Post, Source

    a = Artist(name="Deep", slug="deep")
    db.add(a)
    await db.flush()
    src = Source(artist_id=a.id, platform="patreon",
                 url="https://www.patreon.com/deep", enabled=True)
    db.add(src)
    await db.flush()
    post = Post(artist_id=a.id, source_id=src.id, external_post_id="96376640")
    db.add(post)
    await db.flush()
    done = ExternalLink(post_id=post.id, artist_id=a.id, host="mega",
                        url="https://mega.nz/file/a#k1", status="downloaded",
                        attempts=1)
    dead = ExternalLink(post_id=post.id, artist_id=a.id, host="gdrive",
                        url="https://drive.google.com/x", status="dead",
                        attempts=5, last_error="quota")
    busy = ExternalLink(post_id=post.id, artist_id=a.id, host="mega",
                        url="https://mega.nz/file/b#k2", status="downloading",
                        attempts=1)
    db.add_all([done, dead, busy])
    await db.flush()
    ids = (done.id, dead.id, busy.id)
    await db.commit()

    fetches = []
    monkeypatch.setattr(
        "backend.app.tasks.external.fetch_external_link.delay",
        lambda lid: fetches.append(lid),
    )

    resp = await client.post(
        "/api/admin/posts/refetch-external",
        json={"external_post_id": "96376640"},
    )
    assert resp.status_code == 200
    body = await resp.get_json()
    assert list(body["posts"].values()) == [{"links_total": 3, "links_reset": 2}]
    assert sorted(fetches) == sorted([ids[0], ids[1]])

    rows = dict((await db.execute(
        select(ExternalLink.id, ExternalLink.status)
        .where(ExternalLink.post_id == post.id)
    )).all())
    assert rows[ids[0]] == "pending"
    assert rows[ids[1]] == "pending"
    assert rows[ids[2]] == "downloading"     # in-flight left alone

    # Unknown post → 404, nothing dispatched.
    miss = await client.post(
        "/api/admin/posts/refetch-external",
        json={"external_post_id": "nope"},
    )
    assert miss.status_code == 404


# --- Tier-A: POST /posts/prune-bare + /posts/reconcile-duplicates ---
# These two routes share _run_dry_run_op with the tag prunes (DRY pass,
# task #753); cover their apply path + reconcile's source_id passthrough so
# every helper consumer is exercised at the route level, not just the service.


@pytest.mark.asyncio
async def test_prune_bare_commit_deletes_bare_posts(client, db):
    from sqlalchemy import func, select

    a = Artist(name="BP", slug="bp-api")
    db.add(a)
    await db.flush()
    db.add(Post(artist_id=a.id, external_post_id="bare1"))  # no images/attachments
    await db.commit()

    resp = await client.post(
        "/api/admin/posts/prune-bare", json={"dry_run": False},
    )
    body = await resp.get_json()
    assert body["deleted"] >= 1
    remaining = (await db.execute(
        select(func.count()).select_from(Post).where(Post.external_post_id == "bare1")
    )).scalar_one()
    assert remaining == 0


@pytest.mark.asyncio
async def test_reconcile_duplicates_commit_merges_via_endpoint(client, db):
    from sqlalchemy import select

    a = Artist(name="RC", slug="rc-api")
    db.add(a)
    await db.flush()
    db.add_all([
        Post(artist_id=a.id, external_post_id="711509",
             raw_metadata={"post_id": 1923726, "category": "subscribestar"},
             description="body"),
        Post(artist_id=a.id, external_post_id="1923726",
             raw_metadata={"post_id": 1923726, "category": "subscribestar"}),
    ])
    await db.commit()

    resp = await client.post(
        "/api/admin/posts/reconcile-duplicates", json={"dry_run": False},
    )
    body = await resp.get_json()
    assert body["merged"] == 1
    rows = (await db.execute(select(Post.external_post_id))).scalars().all()
    assert rows == ["1923726"]  # one post-id-keyed keeper survives


@pytest.mark.asyncio
async def test_reconcile_duplicates_rejects_bad_source_id(client):
    resp = await client.post(
        "/api/admin/posts/reconcile-duplicates",
        json={"dry_run": True, "source_id": "abc"},
    )
    assert resp.status_code == 400


# --- DB maintenance: bloat readout + manual VACUUM trigger ----------


@pytest.mark.asyncio
async def test_db_stats_returns_table_bloat_shape(client):
    resp = await client.get("/api/admin/maintenance/db-stats")
    assert resp.status_code == 200
    body = await resp.get_json()
    assert "tables" in body
    names = {t["table"] for t in body["tables"]}
    assert "image_record" in names
    for t in body["tables"]:
        assert {"table", "live", "dead", "dead_pct", "last_vacuum"} <= set(t)


@pytest.mark.asyncio
async def test_trigger_vacuum_queues_the_task(client, monkeypatch):
    from backend.app.tasks import maintenance

    calls = []
    monkeypatch.setattr(maintenance.vacuum_analyze, "delay", lambda: calls.append(1))
    resp = await client.post("/api/admin/maintenance/vacuum")
    assert resp.status_code == 202
    assert calls == [1]


# --- maintenance triggers route through _queued() (DRY Finding B, #753) ---
# Each operator-triggered task returns 202 + the Celery task id via the shared
# _queued() helper. These trigger endpoints had no route-level tests; cover every
# helper consumer, and assert the dry_run flag threads through where applicable.


def _fake_delay(captured):
    def _delay(*args, **kwargs):
        captured.append((args, kwargs))
        return SimpleNamespace(id="task-xyz")
    return _delay


@pytest.mark.asyncio
async def test_trigger_reextract_archives_queues(client, monkeypatch):
    from backend.app.tasks import admin as admin_tasks

    calls = []
    monkeypatch.setattr(
        admin_tasks.reextract_archive_attachments_task, "delay", _fake_delay(calls)
    )
    resp = await client.post("/api/admin/maintenance/reextract-archives")
    assert resp.status_code == 202
    assert (await resp.get_json())["task_id"] == "task-xyz"
    assert len(calls) == 1


@pytest.mark.asyncio
async def test_trigger_prune_missing_files_queues(client, monkeypatch):
    from backend.app.tasks import admin as admin_tasks

    calls = []
    monkeypatch.setattr(
        admin_tasks.prune_missing_file_records_task, "delay", _fake_delay(calls)
    )
    resp = await client.post("/api/admin/maintenance/prune-missing-files")
    assert resp.status_code == 202
    assert (await resp.get_json())["task_id"] == "task-xyz"


@pytest.mark.asyncio
async def test_trigger_dedup_videos_queues_and_threads_dry_run(client, monkeypatch):
    from backend.app.tasks import admin as admin_tasks

    calls = []
    monkeypatch.setattr(admin_tasks.dedup_videos_task, "delay", _fake_delay(calls))
    resp = await client.post(
        "/api/admin/maintenance/dedup-videos", json={"dry_run": False},
    )
    assert resp.status_code == 202
    assert (await resp.get_json())["task_id"] == "task-xyz"
    assert calls[0][1] == {"dry_run": False}  # flag threaded to the task


@pytest.mark.asyncio
async def test_trigger_purge_gated_previews_queues_and_threads_dry_run(client, monkeypatch):
    from backend.app.tasks import admin as admin_tasks

    calls = []
    monkeypatch.setattr(
        admin_tasks.purge_gated_previews_task, "delay", _fake_delay(calls)
    )
    resp = await client.post(
        "/api/admin/maintenance/purge-gated-previews", json={"dry_run": True},
    )
    assert resp.status_code == 202
    assert (await resp.get_json())["task_id"] == "task-xyz"
    assert calls[0][1] == {"dry_run": True}


@pytest.mark.asyncio
async def test_trigger_normalize_live_queues(client, monkeypatch):
    from backend.app.tasks import admin as admin_tasks

    calls = []
    monkeypatch.setattr(admin_tasks.normalize_tags_task, "delay", _fake_delay(calls))
    resp = await client.post("/api/admin/tags/normalize", json={"dry_run": False})
    assert resp.status_code == 202
    assert (await resp.get_json())["task_id"] == "task-xyz"


# --- Tier-A: POST /tags/reset-content -------------------------------


@pytest.mark.asyncio
async def test_reset_content_tagging_dry_run_returns_counts_and_token(client, db):
    db.add_all([
        Tag(name="solo", kind=TagKind.general),
        Tag(name="naruto", kind=TagKind.character),
        Tag(name="Naruto", kind=TagKind.fandom),
        Tag(name="my-series", kind=TagKind.series),
    ])
    await db.commit()
    resp = await client.post(
        "/api/admin/tags/reset-content", json={"dry_run": True}
    )
    assert resp.status_code == 200
    body = await resp.get_json()
    assert body["count"] == 2
    assert body["by_kind"] == {"general": 1, "character": 1}
    # dry-run leaves the rows in place — fandom + series untouched too.
    assert "deleted" not in body
    # The preview arms the apply: an 8-hex confirm token over the live counts.
    assert len(body["confirm"]) == 8


@pytest.mark.asyncio
async def test_reset_content_tagging_apply_requires_confirm_token(client, db):
    from sqlalchemy import func, select

    db.add(Tag(name="solo", kind=TagKind.general))
    await db.commit()

    # No token / wrong token → rejected, nothing deleted.
    for bad in ({"dry_run": False}, {"dry_run": False, "confirm": "deadbeef"}):
        resp = await client.post("/api/admin/tags/reset-content", json=bad)
        assert resp.status_code == 400
    # Non-system only: the seeded hygiene tags (#128) survive a reset by
    # design, so they'd inflate a whole-table count.
    remaining = (await db.execute(
        select(func.count()).select_from(Tag).where(Tag.is_system.is_(False))
    )).scalar_one()
    assert remaining == 1

    # The token from a fresh preview unlocks the apply.
    token = (await (await client.post(
        "/api/admin/tags/reset-content", json={"dry_run": True}
    )).get_json())["confirm"]
    resp = await client.post(
        "/api/admin/tags/reset-content",
        json={"dry_run": False, "confirm": token},
    )
    assert resp.status_code == 200
    assert (await resp.get_json())["deleted"] == 1