"""Recovering interrupted archive transfers (US06-02).

The crash tests are real: a child process applies an archive plan and is killed by
the ``PHOTO_PIPELINE_FAULT_AFTER`` barrier at each persisted transition in turn —
including the moment immediately after an active source has been unlinked. The
parent then reopens the database and asserts the one invariant archiving exists to
uphold: **no verified file is ever lost, and no source is removed without a durable,
byte-identical archive copy.**

Both transfer paths are exercised: the same-filesystem atomic move and (with the
child forcing the device comparison) the cross-filesystem copy/verify/publish.
"""

import os
import subprocess
import sys
import uuid
from datetime import datetime, timezone
from pathlib import Path

import pytest
from sqlalchemy import select

from photo_pipeline.config import Config
from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
from photo_pipeline.models import Asset, UploadBatch, UploadItem
from photo_pipeline.services.archive_journal import FORWARD, MANUAL, RESUMABLE, ArchiveState
from photo_pipeline.services.archive_transfer import (
    MANIFEST_NAME,
    ArchiveTransferService,
    read_manifest,
)
from photo_pipeline.services.archives import ArchiveService
from photo_pipeline.services.hashing import sha256_file

pytestmark = pytest.mark.phase_f

REPO = Path(__file__).resolve().parents[2]
NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)

# Barriers, in the order the transfer persists them. ``source_removed`` is the one
# that matters most: the original is already gone at that point.
BARRIERS = [
    ArchiveState.TRANSFERRING,
    ArchiveState.VERIFIED,
    ArchiveState.REMOVING,
    "source_removed",
    ArchiveState.COMPLETE,
]

# A child that applies the plan and dies at the configured barrier. ``force_copy``
# makes it take the cross-filesystem path without a second real volume.
APPLY_SCRIPT = """
import sys
sys.path.insert(0, {repo!r})
from photo_pipeline.config import Config
from photo_pipeline.db import create_db_engine, create_session_factory
from photo_pipeline.services import archive_transfer

db_url, data_dir, lib, plan_id, force_copy = sys.argv[1:6]
if force_copy == "1":
    archive_transfer._same_filesystem = lambda *args: False
config = Config.from_env(
    {{"PHOTO_PIPELINE_DATA_DIR": data_dir, "PHOTO_PIPELINE_LIBRARY_ROOTS": lib}}
)
sf = create_session_factory(create_db_engine(db_url))
archive_transfer.ArchiveTransferService(sf, config=config).apply(plan_id)
"""


# ── environment ──────────────────────────────────────────────────────────────


def _env(tmp_path):
    (tmp_path / "data").mkdir(exist_ok=True)
    lib = tmp_path / "lib"
    lib.mkdir(exist_ok=True)
    archive = tmp_path / "archive"
    archive.mkdir(exist_ok=True)
    config = Config.from_env(
        {
            "PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"),
            "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
            "PHOTO_PIPELINE_ARCHIVE_FREE_SPACE_RESERVE_BYTES": "0",
        }
    )
    run_migrations(config.database_url)
    return config, create_session_factory(create_db_engine(config.database_url)), lib, archive


def _album(sf, lib, album="rome", names=("a.jpg", "b.jpg")):
    folder = lib / album
    folder.mkdir(parents=True, exist_ok=True)
    with sf() as session:
        batch_id = str(uuid.uuid4())
        session.add(
            UploadBatch(
                id=batch_id,
                album=album,
                folder=str(folder),
                album_name=album,
                state="succeeded",
                preflight_token="v1:test",
                outcome_state="verified",
                created_at=NOW,
            )
        )
        for name in names:
            path = folder / name
            path.write_bytes(f"{album}/{name} content".encode() * 8)
            asset_id = str(uuid.uuid4())
            session.add(
                Asset(
                    id=asset_id,
                    original_path=str(path),
                    current_path=str(path),
                    discovered_at=NOW,
                    hash_version=1,
                    byte_size=path.stat().st_size,
                    current_sha256=sha256_file(path),
                )
            )
            session.add(
                UploadItem(
                    batch_id=batch_id,
                    asset_id=asset_id,
                    path=str(path),
                    sha256=sha256_file(path),
                    sha1="0" * 40,
                    state="sent",
                    outcome="uploaded",
                )
            )
        session.commit()
    return folder


def _plan(sf, config, archive, albums=None):
    location = ArchiveService(sf, config=config).register("external", str(archive))
    token = ArchiveService(sf, config=config).preflight(location["id"], albums)["token"]
    service = ArchiveTransferService(sf, config=config)
    return service, service.create(location["id"], albums, token=token)


def _crash_during_apply(config, tmp_path, lib, plan, barrier, *, force_copy=False):
    script = tmp_path / f"apply_{barrier}.py"
    script.write_text(APPLY_SCRIPT.format(repo=str(REPO)))
    env = dict(os.environ)
    env["PHOTO_PIPELINE_FAULT_AFTER"] = barrier
    result = subprocess.run(
        [
            sys.executable,
            str(script),
            config.database_url,
            str(tmp_path / "data"),
            str(lib),
            plan["id"],
            "1" if force_copy else "0",
        ],
        env=env,
        capture_output=True,
    )
    assert result.returncode in (9, -9), (
        f"child should have been killed at {barrier}, got {result.returncode}: "
        f"{result.stderr.decode(errors='replace')[-400:]}"
    )


def _contents(*roots):
    return sorted(
        path.read_bytes()
        for root in roots
        for path in root.rglob("*")
        if path.is_file() and path.name != MANIFEST_NAME and not path.name.startswith(".")
    )


def _reopen(config):
    return create_session_factory(create_db_engine(config.database_url))


# ── fault injection at every persisted transition ────────────────────────────


@pytest.mark.parametrize("barrier", BARRIERS)
@pytest.mark.parametrize("force_copy", [False, True], ids=["move", "copy"])
def test_a_crash_at_every_transition_loses_nothing_and_recovers(tmp_path, barrier, force_copy):
    config, sf, lib, archive = _env(tmp_path)
    _album(sf, lib)
    service, plan = _plan(sf, config, archive)
    before = _contents(lib)

    _crash_during_apply(config, tmp_path, lib, plan, barrier, force_copy=force_copy)

    # Every file still exists somewhere: the crash may not have cost a single byte.
    assert set(before) <= set(_contents(lib, archive)), f"content lost at {barrier}"

    reopened = _reopen(config)
    recovery = ArchiveTransferService(reopened, config=config)
    # Whatever the crash interrupted is recoverable from evidence — never ambiguous.
    # (A crash on the last barrier can land on a terminal item and leave none.)
    verdicts = recovery.journal.classify_all()
    assert all(v["classification"] in {RESUMABLE, FORWARD} for v in verdicts), verdicts

    recovery.recover()
    recovery.apply(plan["id"])  # finish whatever the crash never started

    assert _contents(archive) == before
    assert _contents(lib) == []
    assert recovery.journal.blocks_mutation() is False
    assert recovery.journal.plan_state(plan["id"]) == "complete"


@pytest.mark.parametrize("barrier", [ArchiveState.VERIFIED, "source_removed"])
def test_recovery_is_idempotent_across_repeated_restarts(tmp_path, barrier):
    config, sf, lib, archive = _env(tmp_path)
    _album(sf, lib)
    service, plan = _plan(sf, config, archive)
    _crash_during_apply(config, tmp_path, lib, plan, barrier)

    reopened = _reopen(config)
    recovery = ArchiveTransferService(reopened, config=config)
    recovery.recover()
    settled = (_contents(lib), _contents(archive), read_manifest(archive / "rome"))

    for _ in range(2):
        recovery.recover()
        assert (_contents(lib), _contents(archive), read_manifest(archive / "rome")) == settled


def test_a_crash_before_anything_was_published_leaves_the_source_intact(tmp_path):
    config, sf, lib, archive = _env(tmp_path)
    folder = _album(sf, lib, names=("a.jpg",))
    service, plan = _plan(sf, config, archive)

    _crash_during_apply(config, tmp_path, lib, plan, ArchiveState.TRANSFERRING)

    recovery = ArchiveTransferService(_reopen(config), config=config)
    verdict = recovery.journal.classify_all()[0]
    assert verdict["classification"] == RESUMABLE
    assert (folder / "a.jpg").exists()
    assert not (archive / "rome" / "a.jpg").exists()

    assert recovery.recover() == {"resumed": 1, "completed": 0, "manual": 0}
    assert recovery.journal.operations(plan["id"])[0]["journal_state"] == ArchiveState.PLANNED


def test_a_crash_after_the_source_was_removed_finishes_the_bookkeeping(tmp_path):
    """The dangerous window: the original is gone and the database still points at
    it. Recovery must complete the record, never re-transfer or report loss."""
    config, sf, lib, archive = _env(tmp_path)
    folder = _album(sf, lib, names=("a.jpg",))
    service, plan = _plan(sf, config, archive)
    expected = service.journal.operations(plan["id"])[0]["expected_sha256"]

    _crash_during_apply(config, tmp_path, lib, plan, "source_removed")

    reopened = _reopen(config)
    recovery = ArchiveTransferService(reopened, config=config)
    verdict = recovery.journal.classify_all()[0]
    assert verdict["classification"] == FORWARD
    assert not (folder / "a.jpg").exists()
    assert sha256_file(archive / "rome" / "a.jpg") == expected
    with reopened() as session:
        stranded = session.scalar(select(Asset))
        assert stranded.current_path is not None, "the crash happened before the DB update"

    assert recovery.recover() == {"resumed": 0, "completed": 1, "manual": 0}

    with reopened() as session:
        asset = session.scalar(select(Asset))
        assert asset.current_path is None
        assert asset.availability_state == "archived_online"
        assert asset.archive_path == "rome/a.jpg"
    assert recovery.journal.operations(plan["id"])[0]["journal_state"] == ArchiveState.COMPLETE
    assert len(read_manifest(archive / "rome")) == 1, "the manifest is not duplicated"


def test_an_archive_copy_that_changed_after_the_crash_blocks_for_a_human(tmp_path):
    """Wrong bytes at the destination can never justify deleting the original, and
    are never overwritten either."""
    config, sf, lib, archive = _env(tmp_path)
    folder = _album(sf, lib, names=("a.jpg",))
    service, plan = _plan(sf, config, archive)
    _crash_during_apply(config, tmp_path, lib, plan, ArchiveState.VERIFIED, force_copy=True)
    (archive / "rome" / "a.jpg").write_bytes(b"tampered with while the app was down")

    recovery = ArchiveTransferService(_reopen(config), config=config)
    verdict = recovery.journal.classify_all()[0]

    assert verdict["classification"] == MANUAL
    assert recovery.recover() == {"resumed": 0, "completed": 0, "manual": 1}
    assert (folder / "a.jpg").exists(), "the source is kept while the archive is unproven"
    assert (archive / "rome" / "a.jpg").read_bytes() == b"tampered with while the app was down"
    # And the unresolved item keeps blocking further archiving.
    assert recovery.journal.blocks_mutation() is True
    report = ArchiveService(_reopen(config), config=config).preflight(plan["location_id"])
    assert "archive_pending" in {issue["code"] for issue in report["blockers"]}


def test_a_verified_archive_whose_copy_vanished_is_never_reported_as_archived(tmp_path):
    config, sf, lib, archive = _env(tmp_path)
    folder = _album(sf, lib, names=("a.jpg",))
    service, plan = _plan(sf, config, archive)
    _crash_during_apply(config, tmp_path, lib, plan, ArchiveState.REMOVING, force_copy=True)
    (archive / "rome" / "a.jpg").unlink()  # the medium lost it

    recovery = ArchiveTransferService(_reopen(config), config=config)

    assert recovery.journal.classify_all()[0]["classification"] == MANUAL
    assert recovery.recover()["manual"] == 1
    assert (folder / "a.jpg").exists()
    assert recovery.journal.operations(plan["id"])[0]["journal_state"] != ArchiveState.COMPLETE