"""One compose operation per stack (F7), and the lock surviving a restart (F11).

``docker compose`` does no locking of its own. Before 0.47.0 nothing stopped two
``update`` calls landing on the same project at once — two browser tabs, or the
auto-update pass picking up a stack somebody had just clicked — and both would
run ``pull`` then ``up -d`` and fight over recreating the same containers.

There was a busy flag, but it only fed the status column: no handler consulted
it before acting. The lock is a table now, so it also holds across workers and
across a restart.
"""
from __future__ import annotations

from datetime import datetime, timedelta, timezone

import pytest
from sqlmodel import Session


@pytest.fixture
def lock_session(db):
    from database import engine

    with Session(engine) as session:
        yield session


@pytest.fixture(autouse=True)
def clean_locks(db):
    """Locks are global state; don't let one test leak into the next."""
    from database import engine
    from models.runtime_state import StackLock
    from sqlmodel import delete

    with Session(engine) as session:
        session.exec(delete(StackLock))
        session.commit()
    yield


@pytest.fixture
def svc():
    from services import stack_lock_service

    return stack_lock_service


# --------------------------------------------------------------------------- #
# The lock itself
# --------------------------------------------------------------------------- #


def test_a_second_caller_is_refused(svc, lock_session):
    svc.acquire(lock_session, "demo", "update", "alice")
    with pytest.raises(svc.StackBusy) as excinfo:
        svc.acquire(lock_session, "demo", "start", "bob")
    assert excinfo.value.action == "update", "the error names the operation in flight"


def test_releasing_frees_it(svc, lock_session):
    svc.acquire(lock_session, "demo", "update")
    svc.release(lock_session, "demo")
    svc.acquire(lock_session, "demo", "start")  # must not raise


def test_releasing_a_lock_nobody_holds_is_harmless(svc, lock_session):
    svc.release(lock_session, "never-locked")


def test_different_stacks_do_not_block_each_other(svc, lock_session):
    svc.acquire(lock_session, "one", "update")
    svc.acquire(lock_session, "two", "update")


def test_hold_releases_even_when_the_operation_fails(svc, lock_session):
    """A failed deploy must not leave the stack locked."""
    with pytest.raises(RuntimeError):
        with svc.hold(lock_session, "demo", "update"):
            raise RuntimeError("compose blew up")
    assert not svc.is_busy(lock_session, "demo")


def test_an_expired_lock_is_taken_over(svc, lock_session):
    """The recovery path for a worker killed mid-deploy.

    Without it the stack would stay locked forever and the only fix would be
    editing the database by hand.
    """
    from models.runtime_state import StackLock

    past = datetime.now(timezone.utc) - timedelta(minutes=5)
    lock_session.add(
        StackLock(
            stack_id="orphaned",
            action="update",
            owner="a worker that died",
            acquired_at=past - timedelta(minutes=30),
            expires_at=past,
        )
    )
    lock_session.commit()

    assert not svc.is_busy(lock_session, "orphaned")
    svc.acquire(lock_session, "orphaned", "start")  # must not raise


def test_active_lists_current_locks_and_skips_expired(svc, lock_session):
    from models.runtime_state import StackLock

    svc.acquire(lock_session, "live", "update")
    lock_session.add(
        StackLock(
            stack_id="dead",
            action="start",
            expires_at=datetime.now(timezone.utc) - timedelta(seconds=1),
        )
    )
    lock_session.commit()

    assert svc.active(lock_session) == {"live": "update"}


def test_prune_removes_only_expired_locks(svc, lock_session):
    from models.runtime_state import StackLock

    svc.acquire(lock_session, "live", "update")
    lock_session.add(
        StackLock(
            stack_id="dead",
            action="start",
            expires_at=datetime.now(timezone.utc) - timedelta(seconds=1),
        )
    )
    lock_session.commit()

    assert svc.prune_expired(lock_session) == 1
    assert svc.active(lock_session) == {"live": "update"}


def test_the_lock_outlives_the_session_that_took_it(svc, db):
    """It is a row, not a set in one worker's memory — that is the point."""
    from database import engine

    with Session(engine) as first:
        svc.acquire(first, "persisted", "update")

    with Session(engine) as second:
        assert svc.is_busy(second, "persisted")
        with pytest.raises(svc.StackBusy):
            svc.acquire(second, "persisted", "start")


# --------------------------------------------------------------------------- #
# Enforcement at the HTTP entry point
# --------------------------------------------------------------------------- #


@pytest.fixture
def locked_stack(db):
    """A registered stack that is already mid-operation."""
    from database import engine
    from models.stack import Stack
    from services import compose_service, stack_lock_service

    stack_id = "lock-fixture"
    compose_service.write_compose(stack_id, "services:\n  a:\n    image: alpine\n")
    with Session(engine) as session:
        if not session.get(Stack, stack_id):
            session.add(Stack(id=stack_id, name="lock fixture"))
            session.commit()
        stack_lock_service.acquire(session, stack_id, "update", "someone-else")
    return stack_id


@pytest.mark.parametrize("action", ["start", "stop", "restart", "pull", "update", "down"])
def test_lifecycle_calls_are_refused_while_a_stack_is_busy(as_admin, locked_stack, action):
    """409, and crucially *without* reaching Docker — the guard is in front."""
    response = as_admin.post(f"/api/stacks/{locked_stack}/{action}")
    assert response.status_code == 409
    assert "busy" in response.json()["detail"].lower()
    assert "update in progress" in response.json()["detail"]


def test_the_stacks_list_shows_a_busy_stack_as_updating(as_admin, locked_stack):
    rows = {row["id"]: row for row in as_admin.get("/api/stacks").json()}
    assert rows[locked_stack]["status"] == "updating"


def test_an_unlocked_stack_is_not_refused(as_admin, db):
    """The guard must not block ordinary use.

    Without a Docker daemon the call fails further in — anything but 409 means
    it got past the lock, which is what this asserts.
    """
    from database import engine
    from models.stack import Stack
    from services import compose_service

    stack_id = "unlocked-fixture"
    compose_service.write_compose(stack_id, "services:\n  a:\n    image: alpine\n")
    with Session(engine) as session:
        if not session.get(Stack, stack_id):
            session.add(Stack(id=stack_id, name="unlocked fixture"))
            session.commit()

    assert as_admin.post(f"/api/stacks/{stack_id}/start").status_code != 409