from __future__ import annotations

import gzip
import hashlib
import json
from pathlib import Path
from threading import Barrier, Thread

import pytest


SAMPLES = "wgs-samples:" + "a" * 64
REFERENCE = "wgs-reference:" + "b" * 64
HAPLOTYPE_ANALYSIS = "wgs-analysis:" + "c" * 64
INDEX_ANALYSIS = "wgs-analysis:" + "d" * 64


def _record(
    workspace: Path, path: Path, role: str, sample_id: str = ""
) -> dict:
    item = {
        "artifact": role,
        "role": role,
        "path": str(path.relative_to(workspace)),
        "full_path": str(path.resolve()),
        "size": path.stat().st_size,
        "size_bytes": path.stat().st_size,
        "sha256": hashlib.sha256(path.read_bytes()).hexdigest(),
    }
    if sample_id:
        item["sample_id"] = sample_id
    return item


def _write_gvcf(path: Path, samples: list[str]) -> None:
    with gzip.open(path, "wt", encoding="utf-8") as handle:
        handle.write("##fileformat=VCFv4.2\n")
        handle.write(
            '##ALT=<ID=NON_REF,Description="Represents any possible '
            'alternative allele">\n'
        )
        handle.write(
            "#CHROM\tPOS\tID\tREF\tALT\tQUAL\tFILTER\tINFO\tFORMAT\t"
            + "\t".join(samples)
            + "\n"
        )
        handle.write(
            "chr1\t1\t.\tA\t<NON_REF>\t.\t.\tEND=10\tGT\t"
            + "\t".join("0/0" for _ in samples)
            + "\n"
        )


def _authorities(workspace: Path) -> tuple[Path, Path]:
    final = workspace / "final"
    haplotype_bundle = (
        final / ".wgs_haplotype_calling_runs" / "Cohort" / ("6" * 32)
    )
    haplotype_bundle.mkdir(parents=True)
    artifacts = {}
    for sample in ("S1", "S2", "S3"):
        gvcf = haplotype_bundle / (
            f"Cohort.wgs_haplotype_calling.{sample}.g.vcf.gz"
        )
        tbi = Path(f"{gvcf}.tbi")
        log = haplotype_bundle / (
            f"Cohort.wgs_haplotype_calling.{sample}.haplotype.log"
        )
        _write_gvcf(gvcf, [sample])
        tbi.write_bytes(f"TBI:{sample}".encode())
        log.write_text(f"sample={sample}\n", encoding="utf-8")
        artifacts[f"gvcf.{sample}"] = _record(
            workspace, gvcf, "gvcf", sample
        )
        artifacts[f"gvcf_index.{sample}"] = _record(
            workspace, tbi, "gvcf_index", sample
        )
        artifacts[f"haplotype_log.{sample}"] = _record(
            workspace, log, "haplotype_log", sample
        )
    haplotype = final / "Cohort.wgs_haplotype_calling.manifest.json"
    haplotype.write_text(
        json.dumps(
            {
                "manifest_version": "1.0",
                "stage": "wgs_haplotype_calling",
                "status": "success",
                "prefix": "Cohort",
                "job_id": "job-1",
                "run_id": "6" * 32,
                "workspace": str(workspace.resolve()),
                "wgs_sample_set_id": SAMPLES,
                "wgs_reference_id": REFERENCE,
                "wgs_analysis_id": HAPLOTYPE_ANALYSIS,
                "upstream_wgs_analysis_ids": {
                    "wgs_markdup": "wgs-analysis:" + "e" * 64,
                    "wgs_bam_qc": "wgs-analysis:" + "f" * 64,
                    "wgs_index": INDEX_ANALYSIS,
                },
                "metrics": {
                    "input_sample_ids": ["S1", "S2", "S3"],
                    "output_sample_ids": ["S1", "S2", "S3"],
                },
                "artifacts": artifacts,
            }
        ),
        encoding="utf-8",
    )

    index_bundle = final / ".wgs_index_runs" / "Cohort" / ("2" * 32)
    index_bundle.mkdir(parents=True)
    reference = index_bundle / "Cohort.wgs_index.reference.fasta"
    fai = Path(f"{reference}.fai")
    dictionary = reference.with_suffix(".dict")
    reference.write_text(">chr1\nAAAAAAAAAA\n", encoding="utf-8")
    fai.write_text("chr1\t10\t6\t10\t11\n", encoding="utf-8")
    dictionary.write_text("@HD\tVN:1.6\n@SQ\tSN:chr1\tLN:10\n", encoding="utf-8")
    index = final / "Cohort.wgs_index.manifest.json"
    index.write_text(
        json.dumps(
            {
                "manifest_version": "1.0",
                "stage": "wgs_index",
                "status": "success",
                "prefix": "Cohort",
                "job_id": "job-1",
                "run_id": "2" * 32,
                "workspace": str(workspace.resolve()),
                "wgs_reference_id": REFERENCE,
                "wgs_analysis_id": INDEX_ANALYSIS,
                "artifacts": {
                    "reference_fasta": _record(
                        workspace, reference, "reference_fasta"
                    ),
                    "reference_fai": _record(
                        workspace, fai, "reference_fai"
                    ),
                    "reference_dict": _record(
                        workspace, dictionary, "reference_dict"
                    ),
                },
            }
        ),
        encoding="utf-8",
    )
    return haplotype, index


def test_stage07_rejects_producer_from_different_index_generation_at_all_gates(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    from apps.cli.run_wgs_gvcf_combine import main
    from apps.web import pipeline_backend
    from gfa_core.mas.tools.stage_status import get_stage_status

    workspace_root = tmp_path / "workspaces"
    workspace = workspace_root / "job-1"
    haplotype, index = _authorities(workspace)
    gatk = _gatk(tmp_path)
    assert main(_argv(workspace, haplotype, index, gatk)) == 0
    output = workspace / "final" / "Cohort.wgs_gvcf_combine.manifest.json"
    output_analysis = json.loads(output.read_text())["wgs_analysis_id"]

    producer = json.loads(haplotype.read_text())
    producer["upstream_wgs_analysis_ids"]["wgs_index"] = (
        "wgs-analysis:" + "0" * 64
    )
    haplotype.write_text(json.dumps(producer))
    assert main(_argv(workspace, haplotype, index, gatk)) == 1

    called = {"value": False}
    def forbidden_run(*_args, **_kwargs):
        called["value"] = True
        return 0
    monkeypatch.setattr(pipeline_backend, "run_cmd_logged", forbidden_run)
    message, _ = pipeline_backend.run_stage(
        stage="wgs_gvcf_combine",
        job=pipeline_backend.JobInfo(
            "job-1", "", str(workspace), str(workspace / "logs"), "now"
        ),
        prefix="Cohort",
        wgs_haplotype_manifest=str(haplotype),
        wgs_index_manifest=str(index),
        wgs_sample_set_id=SAMPLES,
        wgs_reference_id=REFERENCE,
        wgs_haplotype_analysis_id=HAPLOTYPE_ANALYSIS,
        wgs_index_analysis_id=INDEX_ANALYSIS,
        wgs_gatk=str(gatk),
    )
    assert "lineage" in message.lower()
    assert not called["value"]
    assert not get_stage_status(
        "job-1", "Cohort", "wgs_gvcf_combine", workspace_root,
        wgs_sample_set_id=SAMPLES,
        wgs_reference_id=REFERENCE,
        wgs_analysis_id=output_analysis,
    ).completed


def _gatk(tmp_path: Path) -> Path:
    tool = tmp_path / "gatk"
    tool.write_text(
        """#!/usr/bin/env python3
import gzip, json, os, pathlib, sys
args = sys.argv[1:]
if args == ['--version']:
    print('The Genome Analysis Toolkit (GATK) v4.6.0.0')
    raise SystemExit(0)
command = args[0]
def value(flag):
    return args[args.index(flag) + 1]
calls = os.environ.get('GATK_CALLS')
if calls:
    with open(calls, 'a', encoding='utf-8') as handle:
        handle.write(json.dumps(args) + '\\n')
if os.environ.get('FAIL_COMMAND') == command:
    raise SystemExit(23)
if command == 'ValidateVariants':
    path = pathlib.Path(value('-V'))
    index = pathlib.Path(str(path) + '.tbi')
    if not path.is_file() or not index.is_file():
        raise SystemExit(2)
    if index.read_bytes().startswith(b'CORRUPT'):
        raise SystemExit(24)
    raise SystemExit(0)
if command == 'CombineGVCFs':
    inputs = [
        pathlib.Path(args[pos + 1])
        for pos, item in enumerate(args) if item == '-V'
    ]
    if len(inputs) != 3:
        raise SystemExit(25)
    samples = []
    for source in inputs:
        with gzip.open(source, 'rt') as handle:
            header = next(
                line for line in handle if line.startswith('#CHROM')
            )
        samples.extend(header.rstrip().split('\\t')[9:])
    if os.environ.get('BAD_OUTPUT_HEADER'):
        samples[-1] = 'WRONG'
    output = pathlib.Path(value('-O'))
    with gzip.open(output, 'wt') as handle:
        handle.write('##fileformat=VCFv4.2\\n')
        handle.write('##ALT=<ID=NON_REF,Description="NON_REF">\\n')
        handle.write('#CHROM\\tPOS\\tID\\tREF\\tALT\\tQUAL\\tFILTER\\tINFO'
                     '\\tFORMAT\\t' + '\\t'.join(samples) + '\\n')
        handle.write('chr1\\t1\\t.\\tA\\t<NON_REF>\\t.\\t.\\tEND=10'
                     '\\tGT\\t' + '\\t'.join('0/0' for _ in samples) + '\\n')
    raise SystemExit(0)
if command == 'IndexFeatureFile':
    path = pathlib.Path(value('-I'))
    content = (
        b'CORRUPT:output' if os.environ.get('CORRUPT_OUTPUT_INDEX')
        else b'TBI:combined'
    )
    pathlib.Path(str(path) + '.tbi').write_bytes(content)
    raise SystemExit(0)
raise SystemExit(2)
""",
        encoding="utf-8",
    )
    tool.chmod(0o755)
    return tool


def _argv(
    workspace: Path, haplotype: Path, index: Path, gatk: Path
) -> list[str]:
    return [
        "--workspace",
        str(workspace),
        "--job-id",
        "job-1",
        "--prefix",
        "Cohort",
        "--haplotype-manifest",
        str(haplotype),
        "--index-manifest",
        str(index),
        "--gatk",
        str(gatk),
    ]


def test_wgs_gvcf_combine_exposes_import_safe_cli() -> None:
    from apps.cli.run_wgs_gvcf_combine import build_parser, main

    assert callable(build_parser)
    assert callable(main)


def test_three_samples_produce_singleton_canonical_outputs_and_exact_header(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    from apps.cli.run_wgs_gvcf_combine import main

    workspace = tmp_path / "workspace"
    haplotype, index = _authorities(workspace)
    calls = tmp_path / "gatk.calls"
    monkeypatch.setenv("GATK_CALLS", str(calls))
    assert main(_argv(workspace, haplotype, index, _gatk(tmp_path))) == 0
    authority = workspace / "final" / "Cohort.wgs_gvcf_combine.manifest.json"
    manifest = json.loads(authority.read_text())
    assert manifest["metrics"]["input_sample_ids"] == ["S1", "S2", "S3"]
    assert manifest["metrics"]["output_sample_ids"] == ["S1", "S2", "S3"]
    assert manifest["upstream_wgs_analysis_ids"] == {
        "wgs_haplotype_calling": HAPLOTYPE_ANALYSIS,
        "wgs_index": INDEX_ANALYSIS,
    }
    assert set(manifest["artifacts"]) == {
        "combined_gvcf",
        "combined_gvcf_index",
        "combine_log",
    }
    records = {
        item["role"]: Path(item["full_path"])
        for item in manifest["artifacts"].values()
    }
    assert records["combined_gvcf"].name == (
        "Cohort.wgs_gvcf_combine.cohort.combined.g.vcf.gz"
    )
    assert records["combined_gvcf_index"] == Path(
        f"{records['combined_gvcf']}.tbi"
    )
    with gzip.open(records["combined_gvcf"], "rt") as handle:
        header = next(line for line in handle if line.startswith("#CHROM"))
    assert header.rstrip().split("\t")[9:] == ["S1", "S2", "S3"]
    assert records["combined_gvcf_index"].read_bytes() == b"TBI:combined"
    log = records["combine_log"].read_text()
    assert "CombineGVCFs exit=0" in log
    assert "IndexFeatureFile exit=0" in log
    assert "ValidateVariants exit=0" in log
    invocations = [json.loads(line) for line in calls.read_text().splitlines()]
    validate_inputs = [
        call[call.index("-V") + 1]
        for call in invocations
        if call[0] == "ValidateVariants" and "-L" in call
        and ".wgs_haplotype_calling_runs" in call[call.index("-V") + 1]
    ]
    assert len(validate_inputs) == 3
    combine = next(call for call in invocations if call[0] == "CombineGVCFs")
    assert combine.count("-V") == 3


@pytest.mark.parametrize("mutation", ("missing", "duplicate"))
def test_missing_or_duplicate_declared_gvcf_is_rejected(
    tmp_path: Path, mutation: str
) -> None:
    from apps.cli.run_wgs_gvcf_combine import main

    workspace = tmp_path / "workspace"
    haplotype, index = _authorities(workspace)
    payload = json.loads(haplotype.read_text())
    if mutation == "missing":
        payload["artifacts"].pop("gvcf.S2")
    else:
        payload["artifacts"]["gvcf.duplicate"] = dict(
            payload["artifacts"]["gvcf.S2"]
        )
    haplotype.write_text(json.dumps(payload))
    assert main(_argv(workspace, haplotype, index, _gatk(tmp_path))) == 1


def test_corrupt_nonempty_input_tbi_is_rejected_by_real_query(
    tmp_path: Path
) -> None:
    from apps.cli.run_wgs_gvcf_combine import main

    workspace = tmp_path / "workspace"
    haplotype, index = _authorities(workspace)
    payload = json.loads(haplotype.read_text())
    record = payload["artifacts"]["gvcf_index.S2"]
    tbi = Path(record["full_path"])
    tbi.write_bytes(b"CORRUPT:input")
    record["size"] = record["size_bytes"] = tbi.stat().st_size
    record["sha256"] = hashlib.sha256(tbi.read_bytes()).hexdigest()
    haplotype.write_text(json.dumps(payload))
    assert main(_argv(workspace, haplotype, index, _gatk(tmp_path))) == 1


@pytest.mark.parametrize(
    "failure", ("CombineGVCFs", "IndexFeatureFile", "ValidateVariants")
)
def test_each_gatk_return_code_fails_and_preserves_old_authority(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    failure: str,
) -> None:
    from apps.cli.run_wgs_gvcf_combine import main

    workspace = tmp_path / "workspace"
    haplotype, index = _authorities(workspace)
    authority = workspace / "final" / "Cohort.wgs_gvcf_combine.manifest.json"
    old = workspace / "final" / "old.combined.g.vcf.gz"
    old.write_bytes(b"old-success")
    authority.write_text(
        json.dumps(
            {
                "stage": "wgs_gvcf_combine",
                "status": "success",
                "sentinel": True,
            }
        )
    )
    before = (
        hashlib.sha256(authority.read_bytes()).hexdigest(),
        hashlib.sha256(old.read_bytes()).hexdigest(),
    )
    monkeypatch.setenv("FAIL_COMMAND", failure)
    assert main(_argv(workspace, haplotype, index, _gatk(tmp_path))) == 1
    after = (
        hashlib.sha256(authority.read_bytes()).hexdigest(),
        hashlib.sha256(old.read_bytes()).hexdigest(),
    )
    assert after == before
    sidecars = list(
        (workspace / "final").glob(
            "Cohort.wgs_gvcf_combine.failed.*.manifest.json"
        )
    )
    assert len(sidecars) == 1
    assert json.loads(sidecars[0].read_text())["artifacts"] == {}


@pytest.mark.parametrize(
    "environment", ("BAD_OUTPUT_HEADER", "CORRUPT_OUTPUT_INDEX")
)
def test_output_header_or_index_postflight_failure_is_not_published(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    environment: str,
) -> None:
    from apps.cli.run_wgs_gvcf_combine import main

    workspace = tmp_path / "workspace"
    haplotype, index = _authorities(workspace)
    monkeypatch.setenv(environment, "1")
    assert main(_argv(workspace, haplotype, index, _gatk(tmp_path))) == 1
    failure = json.loads(
        (
            workspace
            / "final"
            / "Cohort.wgs_gvcf_combine.manifest.json"
        ).read_text()
    )
    assert failure["status"] == "failed"
    assert failure["artifacts"] == {}


def test_dual_upstream_identity_is_mandatory(tmp_path: Path) -> None:
    from apps.cli.run_wgs_gvcf_combine import main

    workspace = tmp_path / "workspace"
    haplotype, index = _authorities(workspace)
    payload = json.loads(index.read_text())
    payload["wgs_reference_id"] = "wgs-reference:" + "f" * 64
    index.write_text(json.dumps(payload))
    assert main(_argv(workspace, haplotype, index, _gatk(tmp_path))) == 1


def test_status_and_results_require_exact_identity_and_singleton_roles(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    import asyncio

    from apps.cli.run_wgs_gvcf_combine import main
    from apps.web.vue3 import main as web_main
    from gfa_core.mas.tools.stage_status import get_stage_status

    workspace_root = tmp_path / "workspaces"
    workspace = workspace_root / "job-1"
    haplotype, index = _authorities(workspace)
    assert main(_argv(workspace, haplotype, index, _gatk(tmp_path))) == 0
    manifest_path = (
        workspace / "final" / "Cohort.wgs_gvcf_combine.manifest.json"
    )
    manifest = json.loads(manifest_path.read_text())
    analysis_id = manifest["wgs_analysis_id"]
    status = get_stage_status(
        "job-1",
        "Cohort",
        "wgs_gvcf_combine",
        workspace_root,
        wgs_sample_set_id=SAMPLES,
        wgs_reference_id=REFERENCE,
        wgs_analysis_id=analysis_id,
    )
    assert status.completed
    monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
    results = asyncio.run(
        web_main.get_results(
            job_id="job-1",
            stage="wgs_gvcf_combine",
            wgs_sample_set_id=SAMPLES,
            wgs_reference_id=REFERENCE,
            wgs_analysis_id=analysis_id,
        )
    )
    serialized = json.dumps(results)
    assert "combined_gvcf" in serialized
    assert "combined_gvcf_index" in serialized
    manifest["artifacts"].pop("combined_gvcf_index")
    manifest_path.write_text(json.dumps(manifest))
    broken = get_stage_status(
        "job-1",
        "Cohort",
        "wgs_gvcf_combine",
        workspace_root,
        wgs_sample_set_id=SAMPLES,
        wgs_reference_id=REFERENCE,
        wgs_analysis_id=analysis_id,
    )
    assert not broken.completed


def test_backend_builds_stage07_command_and_accepts_only_fresh_manifest(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    from apps.cli.run_wgs_gvcf_combine import main
    from apps.web import pipeline_backend

    workspace = tmp_path / "workspace"
    haplotype, index = _authorities(workspace)
    gatk = _gatk(tmp_path)
    calls: list[list[str]] = []

    def fake_run(command, *_args, **_kwargs):
        calls.append(command)
        return main(command[3:])

    monkeypatch.setattr(pipeline_backend, "run_cmd_logged", fake_run)
    message, output = pipeline_backend.run_stage(
        stage="wgs_gvcf_combine",
        job=pipeline_backend.JobInfo(
            "job-1", "", str(workspace), str(workspace / "logs"), "now"
        ),
        prefix="Cohort",
        wgs_haplotype_manifest=str(haplotype),
        wgs_index_manifest=str(index),
        wgs_sample_set_id=SAMPLES,
        wgs_reference_id=REFERENCE,
        wgs_haplotype_analysis_id=HAPLOTYPE_ANALYSIS,
        wgs_index_analysis_id=INDEX_ANALYSIS,
        wgs_gatk=str(gatk),
    )
    assert "completed" in message.lower()
    assert output == (
        workspace / "final" / "Cohort.wgs_gvcf_combine.manifest.json"
    )
    command = calls[0]
    assert command[:3] == [
        pipeline_backend.sys.executable,
        "-m",
        "apps.cli.run_wgs_gvcf_combine",
    ]
    assert command[command.index("--haplotype-manifest") + 1] == str(
        haplotype
    )
    before = output.read_bytes()
    monkeypatch.setattr(
        pipeline_backend, "run_cmd_logged", lambda *_args, **_kwargs: 0
    )
    stale_message, stale_output = pipeline_backend.run_stage(
        stage="wgs_gvcf_combine",
        job=pipeline_backend.JobInfo(
            "job-1", "", str(workspace), str(workspace / "logs"), "now"
        ),
        prefix="Cohort",
        wgs_haplotype_manifest=str(haplotype),
        wgs_index_manifest=str(index),
        wgs_sample_set_id=SAMPLES,
        wgs_reference_id=REFERENCE,
        wgs_haplotype_analysis_id=HAPLOTYPE_ANALYSIS,
        wgs_index_analysis_id=INDEX_ANALYSIS,
        wgs_gatk=str(gatk),
    )
    assert "failed" in stale_message.lower()
    assert stale_output == output
    assert output.read_bytes() == before


def test_publish_lock_prevents_older_attempt_replacing_newer(
    tmp_path: Path
) -> None:
    from apps.cli.run_wgs_gvcf_combine import _publish

    workspace = tmp_path / "workspace"
    authority = workspace / "final" / "Cohort.wgs_gvcf_combine.manifest.json"
    destination = (
        workspace
        / "final"
        / ".wgs_gvcf_combine_runs"
        / "Cohort"
        / ("9" * 32)
    )
    authority.parent.mkdir(parents=True)
    authority.write_text(
        json.dumps({"status": "success", "start_time": "2026-07-23T12:00:00Z"})
    )
    bundle = tmp_path / "bundle"
    bundle.mkdir()
    (bundle / "artifact").write_text("older")
    manifest_work = tmp_path / "manifest.json"
    manifest_work.write_text("{}")
    assert not _publish(
        bundle,
        destination,
        manifest_work,
        authority,
        "2026-07-23T11:00:00Z",
    )
    assert json.loads(authority.read_text())["start_time"].startswith(
        "2026-07-23T12"
    )


def test_concurrent_publish_has_one_complete_authority(tmp_path: Path) -> None:
    from apps.cli.run_wgs_gvcf_combine import _publish

    workspace = tmp_path / "workspace"
    authority = workspace / "final" / "Cohort.wgs_gvcf_combine.manifest.json"
    authority.parent.mkdir(parents=True)
    barrier = Barrier(2)
    outcomes: list[bool] = []

    def worker(name: str, started_at: str) -> None:
        bundle = tmp_path / f"bundle-{name}"
        bundle.mkdir()
        (bundle / "artifact").write_text(name)
        manifest = tmp_path / f"manifest-{name}.json"
        manifest.write_text(
            json.dumps(
                {
                    "status": "success",
                    "start_time": started_at,
                    "winner": name,
                }
            )
        )
        destination = (
            workspace
            / "final"
            / ".wgs_gvcf_combine_runs"
            / "Cohort"
            / name
        )
        barrier.wait()
        outcomes.append(
            _publish(
                bundle, destination, manifest, authority, started_at
            )
        )

    threads = [
        Thread(target=worker, args=("a" * 32, "2026-07-23T11:00:00Z")),
        Thread(target=worker, args=("b" * 32, "2026-07-23T12:00:00Z")),
    ]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()
    manifest = json.loads(authority.read_text())
    assert manifest["winner"] == "b" * 32
    assert outcomes.count(True) in {1, 2}