"""The Files library's folders and a file's retention: which folder a new file lands in, when a done
file goes, and the GC that takes it — warning once, never a file its user kept, never a file another
user still holds.

Pure-logic for the source-key rule; DB-backed for the rest.
"""

import hashlib
from datetime import datetime, timedelta, timezone

import pytest
from sqlalchemy import func, select

from app import sysconfig
from app.config import settings
from app.downloads import folders, formats, service
from app.downloads import gc as gcmod
from app.db import SessionLocal
from app.downloads.status import JobStatus
from app.models import DownloadJob, FileFolder, MediaAsset, Notification

NOW = datetime.now(timezone.utc)


@pytest.mark.parametrize(
    ("url", "key"),
    [
        ("https://www.tiktok.com/@a/video/1", "site:tiktok.com"),
        ("https://m.facebook.com/groups/x/posts/2", "site:facebook.com"),
        ("https://vimeo.com/123", "site:vimeo.com"),
        ("https://youtu.be/abcdefghijk", "youtube"),
        ("https://music.youtube.com/watch?v=abcdefghijk", "youtube"),
        ("https://www.youtube.com/watch?v=abcdefghijk", "youtube"),
        # A front door is only stripped when a real host remains behind it.
        ("https://m.co/x", "site:m.co"),
        ("not a url", "site:unknown"),
    ],
)
def test_a_url_lands_in_its_sites_folder(url, key):
    assert folders.site_key(url) == key


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


def _asset(db, kind="youtube", ref="abcdefghijk", status="ready", rel="a.mp4", size=10):
    asset = MediaAsset(
        source_kind=kind,
        source_ref=ref,
        format_sig=hashlib.sha256(f"{kind}:{ref}".encode()).hexdigest()[:32],
        status=status,
        rel_path=rel,
        size_bytes=size,
        ref_count=0,
        title="A title",
    )
    db.add(asset)
    db.flush()
    return asset


def _job(db, user_id, asset, *, kind="youtube", ref=None, status=JobStatus.DONE, **kw):
    job = DownloadJob(
        user_id=user_id,
        asset_id=asset.id if asset else None,
        source_kind=kind,
        source_ref=ref or (asset.source_ref if asset else "x"),
        profile_snapshot={},
        status=status,
        **kw,
    )
    db.add(job)
    db.flush()
    folders.place(db, job)
    if asset is not None:
        service.incr_ref_count(db, asset.id)
    db.commit()
    return job


def _folder(db, job) -> FileFolder:
    return db.get(FileFolder, job.folder_id)


def test_every_kind_of_job_lands_in_its_source_folder(db, make_user):
    uid = make_user().id
    yt = _job(db, uid, _asset(db))
    site = _job(db, uid, _asset(db, "url", "https://www.tiktok.com/@a/video/1"), kind="url")
    up = _job(db, uid, _asset(db, "upload", "u1"), kind="upload")
    plex = _job(
        db, uid, _asset(db, "clip", "c1"), kind="clip", edit_spec={"source": "plex", "ref": "9"}
    )
    ytclip = _job(
        db, uid, _asset(db, "clip", "c2"), kind="clip", edit_spec={"source": "youtube", "ref": "v"}
    )
    # An edit of a Plex clip: only its parent knows it came from Plex (the clip's asset says "clip").
    edit = _job(
        db, uid, _asset(db, "edit", "e1"), kind="edit", source_job_id=plex.id,
        source_asset_id=plex.asset_id,
    )  # fmt: skip
    keys = {j.id: _folder(db, j).source_key for j in (yt, site, up, plex, ytclip, edit)}
    assert keys == {
        yt.id: "youtube",
        site.id: "site:tiktok.com",
        up.id: "upload",
        plex.id: "plex",
        ytclip.id: "youtube",
        edit.id: "plex",  # an edit sits with the file it was cut from
    }
    # One folder per origin per user: the two YouTube files share theirs.
    assert _folder(db, yt).id == _folder(db, ytclip).id


def test_an_edit_whose_parent_is_gone_follows_the_parents_asset(db, make_user):
    uid = make_user().id
    src = _asset(db, "url", "https://vimeo.com/1")
    edit = DownloadJob(
        user_id=uid, source_kind="edit", source_ref="e", profile_snapshot={},
        source_job_id=None, source_asset_id=src.id, status=JobStatus.DONE,
    )  # fmt: skip
    assert folders.source_key_for(db, edit) == "site:vimeo.com"


def test_an_edit_keeps_its_origin_after_its_parent_and_its_file_are_gone(db, make_user):
    # The parent download expires, its file goes with its last holder, and both of the edit's
    # pointers to it are cleared: the edit must still go back to the YouTube folder it came from.
    uid = make_user().id
    parent = _job(db, uid, _asset(db))
    edit = _job(
        db, uid, _asset(db, "edit", "e1"), kind="edit", source_job_id=parent.id,
        source_asset_id=parent.asset_id,
    )  # fmt: skip
    home = _folder(db, edit)
    own = FileFolder(user_id=uid, kind=folders.FolderKind.OWN, name="Mine")
    db.add(own)
    db.flush()
    folders.move(db, edit, own)
    edit.source_job_id = edit.source_asset_id = None  # what ON DELETE SET NULL leaves
    db.delete(parent)
    db.commit()
    folders.move(db, edit, home)
    assert edit.folder_id == home.id
    folders.move(db, edit, own)
    folders.delete_own(db, own)
    assert edit.folder_id == home.id


def test_a_folder_deleted_while_a_file_goes_in_takes_that_file_home_too(db, make_user):
    # Before, the delete missed the file still on its way in and failed on the foreign key (a 500),
    # or the file's insert did after the folder went.
    import threading

    uid = make_user().id
    own = FileFolder(user_id=uid, kind=folders.FolderKind.OWN, name="Mine")
    db.add(own)
    db.commit()
    own_id = own.id
    job = _job(db, uid, _asset(db))
    home = job.folder_id
    target = folders.own_folder(db, uid, own_id)  # held from the check until this commits
    outcome: list[object] = []

    def delete_it():
        other = SessionLocal()
        try:
            outcome.append(folders.delete_own(other, other.get(FileFolder, own_id)))
            other.commit()
        except Exception as e:  # noqa: BLE001 — the assertion below names what went wrong
            outcome.append(e)
        finally:
            other.close()

    t = threading.Thread(target=delete_it)
    t.start()
    t.join(timeout=1)
    assert t.is_alive()  # the delete waits for the file, even before it is written
    folders.move(db, job, target)
    db.commit()
    t.join(timeout=10)
    assert outcome == [1]
    db.expire_all()
    assert db.get(DownloadJob, job.id).folder_id == home
    assert db.get(FileFolder, own_id) is None


def test_folders_are_per_user(db, make_user):
    a, b = make_user().id, make_user().id
    asset = _asset(db)
    ja, jb = _job(db, a, asset), _job(db, b, asset)
    assert ja.folder_id != jb.folder_id
    assert (_folder(db, ja).user_id, _folder(db, jb).user_id) == (a, b)


def test_a_source_folder_keeps_files_the_retention_days_and_uploads_forever(db, make_user):
    sysconfig.set_value(db, "download_retention_days", 12)
    db.commit()
    uid = make_user().id
    yt = _job(db, uid, _asset(db))
    up = _job(db, uid, _asset(db, "upload", "u1"), kind="upload")
    assert _folder(db, yt).retention_days == 12
    assert _folder(db, up).retention_days is None


def test_enqueue_places_the_job_and_a_cache_hit_is_stamped_at_once(db, make_user):
    uid = make_user().id
    spec = formats.normalize({})
    first = service.enqueue(db, uid, "youtube", "abcdefghijk", spec)
    assert _folder(db, first).source_key == "youtube"
    assert first.expires_at is None  # queued: its retention starts when it is done
    asset = db.get(MediaAsset, first.asset_id)
    asset.status, asset.rel_path = "ready", "a.mp4"
    db.commit()
    other = make_user().id
    hit = service.enqueue(db, other, "youtube", "abcdefghijk", spec)
    assert hit.status == JobStatus.DONE
    days = _folder(db, hit).retention_days
    assert hit.expires_at is not None
    assert abs(hit.expires_at - (NOW + timedelta(days=days))) < timedelta(minutes=5)


def test_a_finishing_worker_stamps_the_expiry_unless_kept(db, make_user):
    uid = make_user().id
    asset = _asset(db)
    plain = _job(db, uid, asset, status=JobStatus.RUNNING)
    kept = _job(db, uid, asset, status=JobStatus.RUNNING, kept=True)
    for job in (plain, kept):
        assert service.finalize_if_running(db, job.id, status=JobStatus.DONE, progress=100)
    db.commit()
    db.refresh(plain)
    db.refresh(kept)
    assert plain.expires_at is not None
    assert kept.expires_at is None


def _gc(db):
    return gcmod.run_download_gc(db)


def _notices(db, uid, kind):
    return db.execute(
        select(func.count())
        .select_from(Notification)
        .where(Notification.user_id == uid, Notification.type == kind)
    ).scalar()


def test_the_gc_warns_once_then_takes_the_file_out_of_the_library(db, make_user, _root):
    (_root / "a.mp4").write_bytes(b"x")
    uid = make_user().id
    job = _job(db, uid, _asset(db), expires_at=NOW + timedelta(hours=12))
    _gc(db)
    _gc(db)
    assert _notices(db, uid, "download_expiring") == 1
    job.expires_at = NOW - timedelta(minutes=1)
    db.commit()
    assert _gc(db)["expired"] == 1
    assert db.get(DownloadJob, job.id) is None
    assert _notices(db, uid, "download_expired") == 1
    # Its last holder gone, the shared file and its row go too.
    assert not (_root / "a.mp4").exists()
    assert db.execute(select(func.count()).select_from(MediaAsset)).scalar() == 0


@pytest.mark.parametrize("restart", ["unkeep", "move", "retention"])
def test_a_file_whose_clock_starts_again_is_warned_again(db, make_user, _root, restart):
    # The GC warns once per expiry; a file whose expiry was moved must be warned before the new one.
    (_root / "a.mp4").write_bytes(b"x")
    uid = make_user().id
    job = _job(db, uid, _asset(db), expires_at=NOW + timedelta(hours=12))
    _gc(db)
    assert job.expiry_warned
    if restart == "unkeep":
        folders.set_kept(db, job, True)
        folders.set_kept(db, job, False)
    elif restart == "move":
        home = _folder(db, job)
        own = FileFolder(user_id=uid, kind=folders.FolderKind.OWN, name="Mine")
        db.add(own)
        db.flush()
        folders.move(db, job, own)
        folders.move(db, job, home)
    else:
        folders.set_retention(db, _folder(db, job), 90)
    db.commit()
    assert not job.expiry_warned


def test_the_gc_never_takes_a_kept_file(db, make_user, _root):
    (_root / "a.mp4").write_bytes(b"x")
    uid = make_user().id
    job = _job(db, uid, _asset(db), kept=True, expires_at=NOW - timedelta(days=1))
    assert _gc(db)["expired"] == 0
    assert db.get(DownloadJob, job.id) is not None


def test_one_users_expiry_leaves_another_users_copy(db, make_user, _root):
    (_root / "a.mp4").write_bytes(b"x")
    a, b = make_user().id, make_user().id
    asset = _asset(db)
    gone = _job(db, a, asset, expires_at=NOW - timedelta(minutes=1))
    stays = _job(db, b, asset, expires_at=NOW + timedelta(days=20))
    _gc(db)
    assert db.get(DownloadJob, gone.id) is None
    assert db.get(DownloadJob, stays.id) is not None
    assert (_root / "a.mp4").exists()


def _meanwhile(monkeypatch, act):
    """Run `act` in another session (another request) between the GC's expiry SELECT and its
    handling of each file it found — the window a user's click can land in."""
    real = gcmod._relock_expired

    def relock(db, job_id, asset_id, now):
        other = SessionLocal()
        try:
            act(other, job_id)
            other.commit()
        finally:
            other.close()
        return real(db, job_id, asset_id, now)

    monkeypatch.setattr(gcmod, "_relock_expired", relock)


def test_a_file_kept_while_the_gc_runs_stays(db, make_user, _root, monkeypatch):
    (_root / "a.mp4").write_bytes(b"x")
    job = _job(db, make_user().id, _asset(db), expires_at=NOW - timedelta(minutes=1))
    _meanwhile(monkeypatch, lambda other, jid: setattr(other.get(DownloadJob, jid), "kept", True))
    assert _gc(db)["expired"] == 0
    db.expire_all()
    assert db.get(DownloadJob, job.id) is not None
    assert (_root / "a.mp4").exists()


def test_a_file_deleted_while_the_gc_runs_is_released_once(db, make_user, _root, monkeypatch):
    # Released twice, the shared file would go from under the other user's copy.
    (_root / "a.mp4").write_bytes(b"x")
    a, b = make_user().id, make_user().id
    asset = _asset(db)
    _job(db, a, asset, expires_at=NOW - timedelta(minutes=1))
    stays = _job(db, b, asset, expires_at=NOW + timedelta(days=20))

    def user_deletes(other, jid):
        service.delete(other, other.get(DownloadJob, jid))

    _meanwhile(monkeypatch, user_deletes)
    _gc(db)
    db.expire_all()
    assert db.get(DownloadJob, stays.id).asset_id == asset.id
    assert db.get(MediaAsset, asset.id).ref_count == 1
    assert (_root / "a.mp4").exists()


def test_the_cache_cap_never_evicts_a_kept_file_and_an_evicted_one_leaves_every_library(
    db, make_user, _root
):
    sysconfig.set_value(db, "download_total_max_bytes", 15)
    db.commit()
    a, b = make_user().id, make_user().id
    for name in ("old.mp4", "new.mp4"):
        (_root / name).write_bytes(b"x")
    old = _asset(db, ref="oldoldoldol", rel="old.mp4", size=10)
    new = _asset(db, ref="newnewnewne", rel="new.mp4", size=10)
    old.last_access_at = NOW - timedelta(days=9)
    new.last_access_at = NOW - timedelta(days=1)
    kept = _job(db, a, old, kept=True)  # the older file, but kept: the newer one goes instead
    ja, jb = _job(db, a, new), _job(db, b, new)
    assert _gc(db)["evicted"] == 1
    assert db.get(DownloadJob, kept.id) is not None
    assert db.get(DownloadJob, ja.id) is None and db.get(DownloadJob, jb.id) is None
    assert _notices(db, a, "download_evicted") == 1
    assert _notices(db, b, "download_evicted") == 1
    assert (_root / "old.mp4").exists() and not (_root / "new.mp4").exists()


def test_deleting_a_user_takes_their_folders_and_files_with_them(db, make_user):
    # Both tables cascade from users; the files' folder FK refuses deleting a folder that still holds
    # files, which must not stop the account's own deletion.
    user = make_user()
    _job(db, user.id, _asset(db))
    db.delete(user)
    db.commit()
    assert db.execute(select(func.count()).select_from(FileFolder)).scalar() == 0
    assert db.execute(select(func.count()).select_from(DownloadJob)).scalar() == 0


def test_the_longest_host_has_room_for_its_folder(db, make_user):
    label = "a" * 63
    host = ".".join([label, label, label, "b" * 58]) + ".hu"  # 253 characters, the DNS maximum
    assert len(host) == 253
    uid = make_user().id
    job = _job(db, uid, _asset(db, "url", f"https://{host}/v"), kind="url")
    assert _folder(db, job).source_key == f"site:{host}"


def test_the_ready_notice_names_the_file_and_its_folder(db, make_user):
    # The notice's "Open" goes to the file in its folder, so both ride along as data.
    from app.downloads import pipeline

    user = make_user()
    job = _job(db, user.id, _asset(db))
    pipeline._notify_download_done(db, user.id, "A title", job.id, kind="download")
    db.commit()
    note = db.scalars(select(Notification).where(Notification.user_id == user.id)).one()
    assert note.type == "download_done"
    assert note.data == {
        "title": "A title",
        "kind": "download",
        "job_id": job.id,
        "folder_id": job.folder_id,
    }
    assert job.folder_id is not None