"""Retention: expired run work branches are deleted, nothing else ever is."""

from __future__ import annotations

from contextlib import asynccontextmanager
from datetime import UTC, datetime, timedelta
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch
from uuid import uuid4

import httpx
import pytest
from src.config import get_settings
from src.domain.codegen.work_branch import new_state, work_branch_name
from src.integrations.github_client import GitHubAppClient
from src.workers.tasks.cleanup_work_branches import (
    PROMOTED_RETENTION_DAYS,
    cleanup_codegen_work_branches,
)

BASE = "a" * 40


# ---- the delete can only ever remove a run branch ------------------------------------------------


class FakeResponse:
    def __init__(self, status):
        self.status_code = status

    def raise_for_status(self):
        if self.status_code >= 400:
            request = httpx.Request("DELETE", "https://api.github.test/x")
            raise httpx.HTTPStatusError("x", request=request, response=httpx.Response(self.status_code, request=request))


def fake_client(status, calls):
    class Client:
        async def __aenter__(self):
            return self

        async def __aexit__(self, *args):
            return False

        async def delete(self, url, headers=None):
            calls.append(url)
            return FakeResponse(status)

    return Client


def github():
    client = GitHubAppClient.__new__(GitHubAppClient)
    client._headers = AsyncMock(return_value={})
    return client


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "branch", ["preview", "main", "p2hs/codegen/", "p2hs/codegen/../main", "feature/x", "", "p2hs/wip/1"]
)
async def test_only_a_run_work_branch_can_be_deleted(branch):
    calls = []
    with patch("src.integrations.github_client.httpx.AsyncClient", fake_client(204, calls)):
        with pytest.raises(ValueError):
            await github().delete_work_branch("P2HS/p2hs-abc", branch)
    assert calls == []  # not even a request was made


@pytest.mark.asyncio
@pytest.mark.parametrize("status,ok", [(204, True), (404, True), (422, True)])
async def test_deleting_an_existing_or_already_gone_branch_succeeds(status, ok):
    calls = []
    branch = work_branch_name(uuid4())
    with patch("src.integrations.github_client.httpx.AsyncClient", fake_client(status, calls)):
        assert await github().delete_work_branch("P2HS/p2hs-abc", branch) is ok
    assert calls == [f"https://api.github.com/repos/P2HS/p2hs-abc/git/refs/heads/{branch}"]


@pytest.mark.asyncio
async def test_a_real_github_error_is_not_swallowed():
    with patch("src.integrations.github_client.httpx.AsyncClient", fake_client(500, [])):
        with pytest.raises(httpx.HTTPStatusError):
            await github().delete_work_branch("P2HS/p2hs-abc", work_branch_name(uuid4()))


# ---- the sweep -----------------------------------------------------------------------------------


def aged(status, days, run_id=None):
    run_id = run_id or uuid4()
    when = (datetime.now(UTC) - timedelta(days=days)).isoformat()
    state = {**new_state(run_id, "preview", BASE), "status": status, "updated_at": when}
    return run_id, {"context": {"work_branch": state}}


class Sweep:
    def __init__(self, rows, *, enabled=True, days=14, repo="P2HS/p2hs-abc", delete_error=None,
                 page_size=None):
        self.rows = rows
        self.page_size = page_size
        self.events = []
        self.enabled = enabled
        self.days = days
        self.repo = repo
        self.delete = AsyncMock(side_effect=delete_error, return_value=True)
        self.persisted = []

    async def run(self):
        settings = get_settings().model_copy(update={
            "codegen_work_branch_enabled": self.enabled,
            "codegen_work_branch_retention_days": self.days,
        })
        rows = [(run_id, uuid4(), plan) for run_id, plan in self.rows]
        size = self.page_size or max(len(rows), 1)
        pages = [rows[i:i + size] for i in range(0, len(rows), size)] + [[]]

        class Result:
            def __init__(self, data):
                self.data = data

            def all(self):
                return self.data

        @asynccontextmanager
        async def db_factory():
            def select(query):
                self.events.append("select")
                return Result(pages.pop(0) if pages else [])

            yield SimpleNamespace(execute=AsyncMock(side_effect=select))

        async def persist(factory, run_id, state):
            self.events.append("persist")
            self.persisted.append((run_id, state))
            return True

        project = SimpleNamespace(
            repo=SimpleNamespace(repo_full_name=self.repo) if self.repo else None, workspace_id=uuid4()
        )
        with patch("src.config.get_settings", return_value=settings), \
             patch("src.domain.projects.repository.ProjectRepository",
                   lambda db: SimpleNamespace(get_with_repo=AsyncMock(return_value=project))), \
             patch("src.integrations.github_installation.resolve_installation_id_for_workspace",
                   AsyncMock(return_value=1)), \
             patch("src.integrations.github_client.GitHubAppClient",
                   lambda installation_id=None: SimpleNamespace(delete_work_branch=self.delete)), \
             patch("src.domain.codegen.work_branch.persist_state", persist),              patch("src.workers.tasks.cleanup_work_branches._PAGE", size):
            return await cleanup_codegen_work_branches({"db_factory": db_factory})


@pytest.mark.asyncio
async def test_a_failed_branch_is_kept_for_the_retention_period_then_deleted():
    old_id, old = aged("failed", 20)
    new_id, new = aged("failed", 3)
    sweep = Sweep([(old_id, old), (new_id, new)])
    result = await sweep.run()
    assert result == {"checked": 2, "deleted": 1, "skipped": 1, "errors": 0}
    sweep.delete.assert_awaited_once()
    assert sweep.delete.await_args.args[1] == work_branch_name(old_id)
    [(persisted_id, state)] = sweep.persisted
    assert persisted_id == old_id and state["status"] == "deleted"  # recorded, never revisited


@pytest.mark.asyncio
async def test_a_promoted_branch_goes_much_sooner_than_a_failed_one():
    promoted_id, promoted = aged("promoted", PROMOTED_RETENTION_DAYS + 1)
    failed_id, failed = aged("failed", PROMOTED_RETENTION_DAYS + 1)
    sweep = Sweep([(promoted_id, promoted), (failed_id, failed)])
    result = await sweep.run()
    assert result["deleted"] == 1
    assert sweep.delete.await_args.args[1] == work_branch_name(promoted_id)


@pytest.mark.asyncio
async def test_a_state_naming_another_runs_branch_is_never_deleted():
    run_id, plan = aged("failed", 30)
    plan["context"]["work_branch"]["branch"] = work_branch_name(uuid4())  # not this run's branch
    sweep = Sweep([(run_id, plan)])
    result = await sweep.run()
    assert result["deleted"] == 0 and result["skipped"] == 1
    sweep.delete.assert_not_awaited()


@pytest.mark.asyncio
async def test_a_github_failure_is_counted_and_does_not_stop_the_sweep():
    a_id, a = aged("failed", 30)
    b_id, b = aged("blocked", 30)
    sweep = Sweep([(a_id, a), (b_id, b)], delete_error=[RuntimeError("github down"), True])
    result = await sweep.run()
    assert result == {"checked": 2, "deleted": 1, "skipped": 0, "errors": 1}
    assert [p[0] for p in sweep.persisted] == [b_id]  # the failed one is retried next time


@pytest.mark.asyncio
async def test_a_project_with_no_repo_is_just_marked_deleted():
    run_id, plan = aged("failed", 30)
    sweep = Sweep([(run_id, plan)], repo=None)
    result = await sweep.run()
    assert result["deleted"] == 1
    sweep.delete.assert_not_awaited()


@pytest.mark.asyncio
async def test_nothing_happens_unless_the_work_branch_is_enabled():
    run_id, plan = aged("failed", 30)
    sweep = Sweep([(run_id, plan)], enabled=False)
    result = await sweep.run()
    assert result["status"] == "disabled" and result["deleted"] == 0
    sweep.delete.assert_not_awaited()


@pytest.mark.asyncio
async def test_every_row_is_read_before_any_is_marked_deleted_so_later_pages_are_not_skipped():
    # Marking a row `deleted` removes it from the selected set. Reading page N+1 after handling
    # page N with an offset would therefore skip rows that were never examined.
    rows = [aged("failed", 30) for _ in range(5)]
    sweep = Sweep(rows, page_size=2)
    result = await sweep.run()
    assert result == {"checked": 5, "deleted": 5, "skipped": 0, "errors": 0}
    last_select = max(i for i, e in enumerate(sweep.events) if e == "select")
    first_persist = min(i for i, e in enumerate(sweep.events) if e == "persist")
    assert last_select < first_persist