"""The Files library API (`routes.files`), DB-lane: folders (list, own create/rename/delete, a
folder's retention), the file list (done only, by folder, searched, sorted, shared with me), a
file's rename / move / keep / delete, the upload into a folder, and the file's hash.

Every route is also held to the caller's own rows: another user's folder or file answers like a
missing one.
"""

import base64
import hashlib
from datetime import datetime, timedelta, timezone

import pytest
from sqlalchemy import select

from app import sysconfig
from app.config import settings
from app.downloads import folders, service
from app.downloads import upload as uploadmod
from app.downloads.status import JobStatus
from app.models import AuditLog, DownloadJob, DownloadShare, FileFolder, MediaAsset

NOW = datetime.now(timezone.utc)


@pytest.fixture
def _root(tmp_path, monkeypatch):
    monkeypatch.setattr(settings, "files_root", str(tmp_path))
    return tmp_path


def _file(db, user_id, *, kind="youtube", ref=None, title="A title", size=10, rel=None, **kw):
    """A done library file: an asset with bytes on disk-path terms, and the user's job over it."""
    n = db.execute(select(MediaAsset.id)).all()
    ref = ref or f"vid{len(n):08d}"
    asset = MediaAsset(
        source_kind=kind,
        source_ref=ref,
        format_sig=f"sig-{ref}",
        status="ready",
        rel_path=rel or f"{ref}.mp4",
        size_bytes=size,
        ref_count=0,
        title=title,
        media_kind="video",
        container="mp4",
    )
    db.add(asset)
    db.flush()
    job = DownloadJob(
        user_id=user_id,
        asset_id=asset.id,
        source_kind=kind,
        source_ref=ref,
        profile_snapshot={},
        status=kw.pop("status", JobStatus.DONE),
        **kw,
    )
    db.add(job)
    db.flush()
    folders.place(db, job)
    folders.stamp_done(db, job)
    service.incr_ref_count(db, asset.id)
    db.commit()
    return job


def _own(db, user_id, name="Saját"):
    folder = FileFolder(user_id=user_id, kind=folders.FolderKind.OWN, name=name)
    db.add(folder)
    db.commit()
    return folder


def _audits(db, action):
    return db.execute(select(AuditLog).where(AuditLog.action == action)).scalars().all()


# --- folders -------------------------------------------------------------------------------


def test_the_folder_list_counts_each_folders_done_files_and_is_the_callers_own(
    db, make_user, as_user
):
    alice, bob = make_user(), make_user()
    _file(db, alice.id, size=10)
    _file(db, alice.id, size=5)
    _file(db, alice.id, kind="url", ref="https://vimeo.com/1", size=7)
    _file(db, alice.id, status=JobStatus.RUNNING)  # not in the library yet
    _file(db, bob.id)
    got = {
        (f["kind"], f["source_key"]): (f["count"], f["size_bytes"])
        for f in as_user(alice).get("/api/files/folders").json()
    }
    assert got == {("source", "youtube"): (2, 15), ("source", "site:vimeo.com"): (1, 7)}


def test_an_own_folder_is_made_once_per_name_whatever_its_case(db, make_user, as_user):
    alice = make_user()
    c = as_user(alice)
    made = c.post("/api/files/folders", json={"name": "Nyaralás"})
    assert made.status_code == 200, made.text
    assert (made.json()["kind"], made.json()["name"]) == ("own", "Nyaralás")
    again = c.post("/api/files/folders", json={"name": "NYARALÁS"})
    assert again.status_code == 409
    assert again.json()["detail"]["reason"] == "file_folder_name_taken"
    blank = c.post("/api/files/folders", json={"name": "   "})
    assert blank.json()["detail"]["reason"] == "file_folder_name_required"
    # Another user may have a folder of the same name.
    assert as_user(make_user()).post("/api/files/folders", json={"name": "Nyaralás"}).is_success
    assert len(_audits(db, "file_folder_create")) == 2


def test_only_an_own_folder_is_renamed_and_only_by_its_user(db, make_user, as_user):
    alice = make_user()
    c = as_user(alice)
    src = db.get(FileFolder, _file(db, alice.id).folder_id)
    own = _own(db, alice.id, "Régi")
    taken = _own(db, alice.id, "Foglalt")
    ok = c.patch(f"/api/files/folders/{own.id}", json={"name": "Új"})
    assert ok.json()["name"] == "Új"
    clash = c.patch(f"/api/files/folders/{own.id}", json={"name": taken.name.lower()})
    assert clash.json()["detail"]["reason"] == "file_folder_name_taken"
    no = c.patch(f"/api/files/folders/{src.id}", json={"name": "YouTube 2"})
    assert no.status_code == 409 and no.json()["detail"]["reason"] == "file_folder_not_own"
    stranger = as_user(make_user()).patch(f"/api/files/folders/{own.id}", json={"name": "X"})
    assert stranger.status_code == 404
    db.refresh(own)
    assert own.name == "Új"


def test_a_shorter_retention_moves_each_files_expiry_but_never_inside_the_warning_window(
    db, make_user, as_user
):
    sysconfig.set_value(db, "download_gc_grace_days", 3)
    db.commit()
    alice = make_user()
    fresh = _file(db, alice.id)  # expires in 30 days
    old = _file(db, alice.id)
    old.expires_at = NOW + timedelta(days=10)  # 20 days old already
    kept = _file(db, alice.id, kept=True)
    db.commit()
    folder_id = fresh.folder_id
    r = as_user(alice).patch(f"/api/files/folders/{folder_id}", json={"retention_days": 20})
    assert r.json()["retention_days"] == 20
    for j in (fresh, old, kept):
        db.refresh(j)
    # 30 → 20 days: each loses 10 — the one with 10 left would be due now, so it gets the window.
    assert abs(fresh.expires_at - (NOW + timedelta(days=20))) < timedelta(minutes=5)
    assert abs(old.expires_at - (NOW + timedelta(days=3))) < timedelta(minutes=5)
    assert kept.expires_at is None
    forever = as_user(alice).patch(f"/api/files/folders/{folder_id}", json={"retention_days": None})
    assert forever.json()["retention_days"] is None
    db.refresh(fresh)
    assert fresh.expires_at is None
    rows = _audits(db, "file_folder_retention")
    assert [(a.before, a.after) for a in rows] == [
        ({"retention_days": 30}, {"retention_days": 20}),
        ({"retention_days": 20}, {"retention_days": None}),
    ]


def test_a_retention_set_on_a_folder_that_kept_forever_starts_its_files_clock_now(
    db, make_user, as_user
):
    alice = make_user()
    up = _file(db, alice.id, kind="upload", ref="u1")
    assert up.expires_at is None
    r = as_user(alice).patch(f"/api/files/folders/{up.folder_id}", json={"retention_days": 7})
    assert r.status_code == 200, r.text
    db.refresh(up)
    assert abs(up.expires_at - (NOW + timedelta(days=7))) < timedelta(minutes=5)


def test_deleting_an_own_folder_sends_its_files_home_and_a_source_folder_cannot_be_deleted(
    db, make_user, as_user
):
    alice = make_user()
    c = as_user(alice)
    yt = _file(db, alice.id)
    site = _file(db, alice.id, kind="url", ref="https://vimeo.com/2")
    home = {yt.id: yt.folder_id, site.id: site.folder_id}
    own = _own(db, alice.id)
    for j in (yt, site):
        assert c.patch(f"/api/files/{j.id}", json={"folder_id": own.id}).is_success
    r = c.delete(f"/api/files/folders/{own.id}")
    assert r.json() == {"ok": True, "deleted": own.id, "files_moved": 2}
    assert db.get(FileFolder, own.id) is None
    for j in (yt, site):
        db.refresh(j)
        assert j.folder_id == home[j.id]
        # Back in a source folder, its clock runs again from now.
        assert abs(j.expires_at - (NOW + timedelta(days=30))) < timedelta(minutes=5)
    src = c.delete(f"/api/files/folders/{yt.folder_id}")
    assert src.status_code == 409 and src.json()["detail"]["reason"] == "file_folder_not_own"
    other = _own(db, alice.id, "Más")
    assert as_user(make_user()).delete(f"/api/files/folders/{other.id}").status_code == 404
    assert len(_audits(db, "file_folder_delete")) == 1


# --- the file list -------------------------------------------------------------------------


def test_the_list_holds_the_callers_done_files_by_folder(db, make_user, as_user):
    alice, bob = make_user(), make_user()
    yt = _file(db, alice.id)
    site = _file(db, alice.id, kind="url", ref="https://vimeo.com/3")
    _file(db, alice.id, status=JobStatus.QUEUED)
    _file(db, bob.id)
    c = as_user(alice)
    assert {f["id"] for f in c.get("/api/files").json()} == {yt.id, site.id}
    only = c.get("/api/files", params={"folder_id": site.folder_id}).json()
    assert [f["id"] for f in only] == [site.id]
    assert only[0]["folder_id"] == site.folder_id and only[0]["shared"] is False
    # Bob's folder id names nothing of Alice's.
    bobs = db.execute(select(FileFolder.id).where(FileFolder.user_id == bob.id)).scalar()
    assert c.get("/api/files", params={"folder_id": bobs}).json() == []


def test_the_list_is_searched_by_name_and_channel_with_wildcards_taken_literally(
    db, make_user, as_user
):
    alice = make_user()
    hit = _file(db, alice.id, title="Balaton 100% nyár")
    by_channel = _file(db, alice.id, title="Más", display_uploader="Balatoni csatorna")
    _file(db, alice.id, title="Balaton 1000 nyár")
    c = as_user(alice)
    assert [f["id"] for f in c.get("/api/files", params={"q": "100%"}).json()] == [hit.id]
    found = {f["id"] for f in c.get("/api/files", params={"q": "balatoni"}).json()}
    assert found == {by_channel.id}


def test_the_list_sorts_by_name_size_or_when_added(db, make_user, as_user):
    alice = make_user()
    b = _file(db, alice.id, title="Béka", size=30)
    a = _file(db, alice.id, title="alma", size=10)
    z = _file(db, alice.id, title="Zebra", size=20)
    c = as_user(alice)

    def ids(**params):
        return [f["id"] for f in c.get("/api/files", params=params).json()]

    assert ids() == [z.id, a.id, b.id]  # newest first
    assert ids(sort="name", order="asc") == [a.id, b.id, z.id]  # the reader's alphabet, any case
    assert ids(sort="size", order="desc") == [b.id, z.id, a.id]


def test_a_file_is_added_when_it_is_done_not_when_it_was_asked_for(db, make_user, as_user):
    """A download that failed for days and then went through is new: it lists first, dated today."""
    alice = make_user()
    early = _file(db, alice.id, created_at=NOW - timedelta(days=1))
    early.ready_at = NOW - timedelta(days=1)
    late = _file(db, alice.id, status=JobStatus.RUNNING, created_at=NOW - timedelta(days=5))
    late.ready_at = None
    db.commit()
    assert service.finalize_if_running(db, late.id, status=JobStatus.DONE)
    db.commit()
    db.expire_all()  # the finalize is a Core UPDATE; the route reads this same session
    got = as_user(alice).get("/api/files").json()
    assert [f["id"] for f in got] == [late.id, early.id]
    ready = datetime.fromisoformat(got[0]["ready_at"])
    assert ready > NOW - timedelta(minutes=5)
    assert datetime.fromisoformat(got[0]["created_at"]) < NOW - timedelta(days=4)


def test_a_job_made_done_is_ready_as_it_is_made(db, make_user):
    # An upload kept as it came is inserted done; the database refuses a done job without the moment.
    job = DownloadJob(
        user_id=make_user().id,
        source_kind="upload",
        source_ref="r",
        profile_snapshot={},
        status=JobStatus.DONE,
    )
    db.add(job)
    db.commit()
    assert job.ready_at is not None


def test_shared_with_me_lists_what_others_shared_and_nothing_else(db, make_user, as_user):
    alice, bob, carol = make_user(), make_user(), make_user()
    theirs = _file(db, bob.id)
    _file(db, bob.id)  # not shared
    for_carol = _file(db, bob.id)
    db.add(DownloadShare(job_id=theirs.id, from_user_id=bob.id, to_user_id=alice.id))
    db.add(DownloadShare(job_id=for_carol.id, from_user_id=bob.id, to_user_id=carol.id))
    db.commit()
    got = as_user(alice).get("/api/files", params={"scope": "shared"}).json()
    assert [(f["id"], f["shared"]) for f in got] == [(theirs.id, True)]


def test_shared_with_me_is_ordered_by_when_it_was_shared_and_can_be_dismissed(
    db, make_user, as_user
):
    """A friend's old file shared today is new to me: "added" is the share's date there. Dismissing
    one drops only my grant; the owner's file stays."""
    alice, bob = make_user(), make_user()
    old = _file(db, bob.id, created_at=NOW - timedelta(days=30))
    new = _file(db, bob.id, created_at=NOW - timedelta(days=1))
    db.add(
        DownloadShare(
            job_id=new.id,
            from_user_id=bob.id,
            to_user_id=alice.id,
            created_at=NOW - timedelta(hours=2),
        )
    )
    db.add(
        DownloadShare(
            job_id=old.id,
            from_user_id=bob.id,
            to_user_id=alice.id,
            created_at=NOW - timedelta(hours=1),
        )
    )
    db.commit()
    c = as_user(alice)
    got = c.get("/api/files", params={"scope": "shared"}).json()
    assert [f["id"] for f in got] == [old.id, new.id]
    assert got[0]["shared_at"] is not None and got[0]["shared_at"] > got[1]["shared_at"]
    r = c.delete(f"/api/files/shared/{old.id}")
    assert r.json() == {"ok": True, "removed": old.id}
    assert [f["id"] for f in c.get("/api/files", params={"scope": "shared"}).json()] == [new.id]
    assert db.get(DownloadJob, old.id) is not None


def test_each_listed_file_says_where_it_came_from_wherever_it_sits(db, make_user, as_user):
    """The Source column: a file's own source folder, also after it moved into an own folder."""
    alice = make_user()
    yt = _file(db, alice.id)
    site = _file(db, alice.id, kind="url", ref="https://www.tiktok.com/@x/video/1")
    moved = _file(db, alice.id)
    own = _own(db, alice.id)
    c = as_user(alice)
    assert c.patch(f"/api/files/{moved.id}", json={"folder_id": own.id}).status_code == 200
    got = {f["id"]: f["source_key"] for f in c.get("/api/files").json()}
    assert got == {yt.id: "youtube", site.id: "site:tiktok.com", moved.id: "youtube"}


# --- one file ------------------------------------------------------------------------------


def test_a_file_is_renamed_kept_and_given_back_to_its_folders_retention(db, make_user, as_user):
    alice = make_user()
    job = _file(db, alice.id)
    c = as_user(alice)
    assert c.patch(f"/api/files/{job.id}", json={"name": "Új név"}).json()["display_name"] == (
        "Új név"
    )
    kept = c.patch(f"/api/files/{job.id}", json={"kept": True}).json()
    assert (kept["kept"], kept["expires_at"]) == (True, None)
    back = c.patch(f"/api/files/{job.id}", json={"kept": False}).json()
    assert back["kept"] is False and back["expires_at"] is not None
    # Setting what it already is records nothing.
    assert c.patch(f"/api/files/{job.id}", json={"kept": False}).status_code == 200
    assert len(_audits(db, "file_keep")) == 1 and len(_audits(db, "file_unkeep")) == 1
    assert as_user(make_user()).patch(f"/api/files/{job.id}", json={"kept": True}).status_code == (
        404
    )


def test_a_files_channel_and_links_are_set_and_only_plausible_links_are_kept(
    db, make_user, as_user
):
    alice = make_user()
    job = _file(db, alice.id)
    c = as_user(alice)
    got = c.patch(
        f"/api/files/{job.id}",
        json={
            "uploader": "  Saját csatorna ",
            "uploader_url": "mailto:someone@example.com",
            "extra_links": ["https://a.example/x", "ftp://b.example", "", 7, "http://[::1/x"]
            + [f"https://c.example/{i}" for i in range(10)],
        },
    ).json()
    assert got["display_uploader"] == "Saját csatorna"
    assert got["display_uploader_url"] is None
    assert got["extra_links"][0] == "https://a.example/x" and len(got["extra_links"]) == 8
    cleared = c.patch(f"/api/files/{job.id}", json={"uploader": "", "extra_links": []}).json()
    assert (cleared["display_uploader"], cleared["extra_links"]) == (None, [])


def test_the_usage_carries_the_quota_and_the_upload_cap(db, make_user, as_user):
    got = as_user(make_user()).get("/api/files/usage").json()
    assert got["upload_max_bytes"] == sysconfig.get_int(db, "download_upload_max_bytes")
    assert "footprint_bytes" in got and "max_bytes" in got


def test_a_file_moves_to_an_own_folder_or_home_and_nowhere_else(db, make_user, as_user):
    alice, bob = make_user(), make_user()
    job = _file(db, alice.id)
    home = job.folder_id
    vimeo = db.get(FileFolder, _file(db, alice.id, kind="url", ref="https://vimeo.com/4").folder_id)
    own = _own(db, alice.id)
    bobs = _own(db, bob.id, "Bobé")
    c = as_user(alice)
    moved = c.patch(f"/api/files/{job.id}", json={"folder_id": own.id}).json()
    assert (moved["folder_id"], moved["expires_at"]) == (own.id, None)  # an own folder keeps it
    wrong = c.patch(f"/api/files/{job.id}", json={"folder_id": vimeo.id})
    assert wrong.json()["detail"]["reason"] == "file_folder_not_its_source"
    foreign = c.patch(f"/api/files/{job.id}", json={"folder_id": bobs.id})
    assert foreign.json()["detail"]["reason"] == "file_folder_gone"
    home_again = c.patch(f"/api/files/{job.id}", json={"folder_id": home}).json()
    assert home_again["folder_id"] == home and home_again["expires_at"] is not None
    assert [(a.before, a.after) for a in _audits(db, "file_move")] == [
        ({"folder_id": home}, {"folder_id": own.id}),
        ({"folder_id": own.id}, {"folder_id": home}),
    ]


def test_a_deleted_file_leaves_the_library_and_the_log_says_so(db, make_user, as_user, _root):
    alice = make_user()
    job = _file(db, alice.id)
    assert as_user(make_user()).delete(f"/api/files/{job.id}").status_code == 404
    r = as_user(alice).delete(f"/api/files/{job.id}")
    assert r.json() == {"ok": True, "deleted": job.id}
    db.expire_all()
    assert db.get(DownloadJob, job.id) is None
    [row] = _audits(db, "file_delete")
    assert (row.actor_id, row.target_id) == (alice.id, str(job.id))


def test_a_file_shared_with_someone_is_theirs_to_watch_not_to_change(db, make_user, as_user, _root):
    # The read routes let a share's recipient in; the write routes must not.
    alice, friend = make_user(), make_user()
    job = _file(db, alice.id, title="Mine")
    db.add(DownloadShare(job_id=job.id, from_user_id=alice.id, to_user_id=friend.id))
    db.commit()
    c = as_user(friend)
    assert c.patch(f"/api/files/{job.id}", json={"name": "Theirs"}).status_code == 404
    assert c.patch(f"/api/files/{job.id}", json={"kept": True}).status_code == 404
    assert c.delete(f"/api/files/{job.id}").status_code == 404
    db.expire_all()
    still = db.get(DownloadJob, job.id)
    assert still is not None and still.display_name is None and not still.kept


# --- upload, the file, its hash ------------------------------------------------------------


@pytest.fixture
def _no_probe(monkeypatch):
    # A file ffprobe cannot read: a generic file, done at once (no conversion).
    monkeypatch.setattr(uploadmod.probe, "probe_json", lambda _p: None)


def test_an_upload_lands_in_the_folder_it_asked_for_with_its_hash(
    db, make_user, as_user, _root, _no_probe, upload_headers
):
    alice = make_user()
    own = _own(db, alice.id)
    body = b"hello files"
    c = as_user(alice)
    r = c.post("/api/files/upload", content=body, headers=upload_headers("a.txt", folder=own.id))
    assert r.status_code == 200, r.text
    job = db.get(DownloadJob, r.json()["id"])
    assert job.folder_id == own.id and job.expires_at is None
    asset = db.get(MediaAsset, job.asset_id)
    assert asset.sha256 == hashlib.sha256(body).hexdigest()
    plain = c.post("/api/files/upload", content=b"x", headers=upload_headers("b.txt")).json()
    assert db.get(FileFolder, plain["folder_id"]).source_key == "upload"
    assert len(_audits(db, "file_upload")) == 2


@pytest.mark.parametrize(
    ("which", "reason"),
    [
        ("site", "file_folder_not_its_source"),  # YouTube's folder: no upload came from there
        ("foreign", "file_folder_gone"),
        ("garbage", "file_folder_gone"),
    ],
)
def test_an_upload_to_a_folder_it_may_not_use_is_refused_before_its_body_is_read(
    db, make_user, as_user, _root, _no_probe, upload_headers, monkeypatch, which, reason
):
    from app.routes import files as files_routes

    alice, bob = make_user(), make_user()
    folder = {
        "site": lambda: _file(db, alice.id).folder_id,
        "foreign": lambda: _own(db, bob.id).id,
        "garbage": lambda: "12abc",
    }[which]()
    streamed = []

    async def spy(*args, **kwargs):
        streamed.append(True)
        raise AssertionError("the body was read")

    monkeypatch.setattr(files_routes, "capped_stream_to", spy)
    before = db.execute(select(DownloadJob.id)).all()
    r = as_user(alice).post(
        "/api/files/upload", content=b"data", headers=upload_headers("a.txt", folder=folder)
    )
    assert (r.status_code, r.json()["detail"]["reason"]) == (409, reason)
    assert streamed == []
    assert db.execute(select(DownloadJob.id)).all() == before


def test_an_upload_holds_its_folder_only_once_its_file_is_about_to_be_in_it(
    db, make_user, as_user, _root, upload_headers, monkeypatch
):
    # Held from the start, the folder waited out the probe and the poster (up to a minute) before a
    # rename or a delete of it could go on; not held at all, a delete could take it from under the file.
    alice = make_user()
    own = _own(db, alice.id)
    steps: list[object] = []
    real = folders.own_folder

    def own_folder(db_, user_id, folder_id, *, hold=True):
        steps.append(("folder", hold))
        return real(db_, user_id, folder_id, hold=hold)

    def probe(_p):
        steps.append("probe")

    monkeypatch.setattr(folders, "own_folder", own_folder)
    monkeypatch.setattr(uploadmod.probe, "probe_json", probe)
    r = as_user(alice).post(
        "/api/files/upload", content=b"x", headers=upload_headers("a.txt", folder=own.id)
    )
    assert r.status_code == 200, r.text
    # The route's own look before the body, the ingest's look, the probe, then the hold.
    assert steps[-3:] == [("folder", False), "probe", ("folder", True)]


def test_a_file_moved_holds_the_folder_it_goes_to(db, make_user, as_user, monkeypatch):
    alice = make_user()
    job = _file(db, alice.id)
    own = _own(db, alice.id)
    holds: list[bool] = []
    real = folders.own_folder

    def own_folder(db_, user_id, folder_id, *, hold=True):
        holds.append(hold)
        return real(db_, user_id, folder_id, hold=hold)

    monkeypatch.setattr(folders, "own_folder", own_folder)
    r = as_user(alice).patch(f"/api/files/{job.id}", json={"folder_id": own.id})
    assert r.status_code == 200, r.text
    assert holds == [True]


def test_a_folder_deleted_while_its_upload_streams_refuses_the_upload(
    db, make_user, as_user, _root, _no_probe, upload_headers, monkeypatch
):
    from app.routes import files as files_routes

    alice = make_user()
    own = _own(db, alice.id)
    real = files_routes.capped_stream_to

    async def deleting_meanwhile(*args, **kwargs):
        size = await real(*args, **kwargs)
        db.delete(db.get(FileFolder, own.id))
        db.commit()
        return size

    monkeypatch.setattr(files_routes, "capped_stream_to", deleting_meanwhile)
    r = as_user(alice).post(
        "/api/files/upload", content=b"data", headers=upload_headers("a.txt", folder=own.id)
    )
    assert (r.status_code, r.json()["detail"]["reason"]) == (409, "file_folder_gone")
    assert db.execute(select(DownloadJob.id).where(DownloadJob.user_id == alice.id)).all() == []
    assert not (_root / ".uploads").exists()  # checked before the file was placed


def test_the_file_carries_its_hash_once_known_and_the_hash_is_taken_on_first_ask(
    db, make_user, as_user, _root
):
    alice, friend = make_user(), make_user()
    body = b"\x00\x01video bytes"
    job = _file(db, alice.id, rel="v.mp4")
    (_root / "v.mp4").write_bytes(body)
    c = as_user(alice)
    first = c.get(f"/api/files/{job.id}/file")
    assert first.status_code == 200 and first.content == body
    assert "repr-digest" not in first.headers  # not hashed yet: nothing claimed
    want = hashlib.sha256(body).hexdigest()
    assert c.get(f"/api/files/{job.id}/sha256").json() == {"sha256": want}
    assert db.get(MediaAsset, job.asset_id).sha256 == want
    ranged = c.get(f"/api/files/{job.id}/file", headers={"Range": "bytes=0-3"})
    assert ranged.status_code == 206
    digest = base64.b64encode(bytes.fromhex(want)).decode()
    assert ranged.headers["repr-digest"] == f"sha-256=:{digest}:"
    # A friend it was shared with may read it too; a stranger is told it does not exist.
    db.add(DownloadShare(job_id=job.id, from_user_id=alice.id, to_user_id=friend.id))
    db.commit()
    assert as_user(friend).get(f"/api/files/{job.id}/sha256").json() == {"sha256": want}
    assert as_user(make_user()).get(f"/api/files/{job.id}/sha256").status_code == 404


def test_only_a_few_files_are_hashed_at_once(db, make_user, as_user, _root):
    # Each hash holds a thread of the pool every sync route shares for the whole read; a caller
    # past the few that may run is told to come back, not queued behind them.
    from app.routes import files as filesroute

    alice = make_user()
    job = _file(db, alice.id, rel="v.mp4")
    (_root / "v.mp4").write_bytes(b"bytes")
    held = [filesroute._HASHING.acquire(blocking=False) for _ in range(filesroute._HASHERS)]
    try:
        assert all(held)
        busy = as_user(alice).get(f"/api/files/{job.id}/sha256")
    finally:
        for _ in held:
            filesroute._HASHING.release()
    assert busy.status_code == 429
    assert busy.json()["detail"]["reason"] == "file_hash_busy"
    assert as_user(alice).get(f"/api/files/{job.id}/sha256").status_code == 200


def test_a_new_file_at_an_assets_path_forgets_the_old_hash(db, make_user):
    job = _file(db, make_user().id)
    asset = db.get(MediaAsset, job.asset_id)
    asset.sha256 = "ab" * 32
    asset.rel_path = "converted.mp4"
    assert asset.sha256 is None


# --- the queue (the tray's poll) ------------------------------------------------------------------


def test_the_queue_is_what_is_in_flight_and_what_just_finished_not_the_library(
    db, make_user, as_user
):
    # The tray polls this on every page: a finished file is the library's, so it is listed only
    # while the tray may still be showing it done.
    from sqlalchemy import update

    user = make_user()
    old = _file(db, user.id, title="Old")
    fresh = _file(db, user.id, title="Fresh")
    waiting = _file(db, user.id, title="Waiting", status=JobStatus.QUEUED)
    stuck = _file(db, user.id, title="Stuck", status=JobStatus.ERROR)
    paused = _file(db, user.id, title="Paused", status=JobStatus.PAUSED)
    stopped = _file(db, user.id, title="Stopped", status=JobStatus.ERROR)
    dropped = _file(db, user.id, title="Dropped", status=JobStatus.CANCELED)
    for job in (old, stopped, dropped):
        db.execute(
            update(DownloadJob)
            .where(DownloadJob.id == job.id)
            .values(updated_at=NOW - timedelta(hours=1), ready_at=NOW - timedelta(hours=1))
        )
    db.commit()
    rows = as_user(user).get("/api/downloads").json()
    # An old failure still waits for the user; an old finished or cancelled job does not.
    assert {row["id"] for row in rows} == {fresh.id, waiting.id, stuck.id, paused.id, stopped.id}
    # The tray names where each came from, as the library does.
    assert {row["source_key"] for row in rows} == {"youtube"}


def test_an_old_file_written_to_does_not_come_back_as_just_finished(db, make_user, as_user):
    # Renaming, keeping or moving touches the row; the tray would announce the file as ready again.
    from sqlalchemy import update

    user = make_user()
    old = _file(db, user.id, title="Old")
    db.execute(
        update(DownloadJob)
        .where(DownloadJob.id == old.id)
        .values(updated_at=NOW - timedelta(days=30), ready_at=NOW - timedelta(days=30))
    )
    db.commit()
    c = as_user(user)
    assert (
        c.patch(f"/api/files/{old.id}", json={"name": "Renamed", "kept": True}).status_code == 200
    )
    assert [row["id"] for row in c.get("/api/downloads").json()] == []