"""Soft delete and neighbour relinking against the fake repository (ADR-0016).

The point of these tests is the pointer chain, not the HTTP surface. ADR-0002
treats a partial relink as a defect, and the ways to produce one are all here:
deleting at either boundary, deleting the same point twice, losing a version
race half way through a batch, and pointing at a neighbour that is gone.

`tests/integration/qdrant/test_points_deletion.py` runs the two central cases
against real Qdrant. What only this file can do cheaply is force the races.
"""

import uuid
from collections.abc import Sequence
from dataclasses import dataclass
from typing import override

import pytest
import structlog.testing

from src.application.points.deletion import soft_delete_file_points, soft_delete_point
from src.application.points.errors import PointVersionConflictError
from src.application.points.point import Point, PointNotFoundError
from src.application.ports.point_repository import PayloadPatch
from tests.fakes import FakePointRepository
from tests.support.point_contract import SeedSpec, build_point

pytestmark = [pytest.mark.unit, pytest.mark.asyncio]

TENANT = uuid.UUID("11111111-1111-4111-8111-111111111111")
OTHER_TENANT = uuid.UUID("22222222-2222-4222-8222-222222222222")
FILE = uuid.UUID("33333333-3333-4333-8333-333333333333")
OTHER_FILE = uuid.UUID("44444444-4444-4444-8444-444444444444")
ACTOR = "api_key:test"


def _chain(count: int, *, tenant: uuid.UUID = TENANT, file_id: uuid.UUID = FILE) -> list[Point]:
    """`count` points of one file, linked head to tail in `order_id` order."""
    points = [
        build_point(SeedSpec(tenant, file_id, index, f"chunk {index}", float(index + 1)))
        for index in range(count)
    ]
    return [
        point.model_copy(
            update={
                "previous_chunk_id": points[index - 1].point_id if index else None,
                "next_chunk_id": (points[index + 1].point_id if index + 1 < len(points) else None),
            }
        )
        for index, point in enumerate(points)
    ]


def _seeded(*points: Point) -> FakePointRepository:
    repository = FakePointRepository()
    for point in points:
        repository.add(point)
    return repository


def _stored(repository: FakePointRepository, point_id: uuid.UUID) -> Point:
    return repository.points[str(point_id)]


def _walk(repository: FakePointRepository, head: uuid.UUID) -> list[uuid.UUID]:
    """Follow `next_chunk_id` from `head`, guarding against a cycle."""
    visited: list[uuid.UUID] = []
    current: uuid.UUID | None = head
    while current is not None and len(visited) <= len(repository.points):
        visited.append(current)
        current = _stored(repository, current).next_chunk_id
    return visited


async def test_soft_delete_point_relinks_the_neighbours_of_a_middle_point() -> None:
    first, middle, last = _chain(3)
    repository = _seeded(first, middle, last)

    deleted = await soft_delete_point(
        repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR
    )

    assert deleted.is_active is False
    assert deleted.deleted_at is not None
    assert _stored(repository, first.point_id).next_chunk_id == last.point_id
    assert _stored(repository, last.point_id).previous_chunk_id == first.point_id


async def test_soft_delete_point_sends_the_deactivation_and_both_relinks_in_one_batch() -> None:
    """A partial relink is a defect, so the three patches must not be split up."""
    first, middle, last = _chain(3)
    repository = _seeded(first, middle, last)

    await soft_delete_point(repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR)

    assert len(repository.patch_batches) == 1
    assert {patch.point_id for patch in repository.patch_batches[0]} == {
        middle.point_id,
        first.point_id,
        last.point_id,
    }


async def test_soft_delete_point_bumps_the_version_of_every_point_it_touches() -> None:
    """A relinked neighbour really changed, so a stale editor of it must `409`."""
    first, middle, last = _chain(3)
    repository = _seeded(first, middle, last)

    await soft_delete_point(repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR)

    assert [_stored(repository, point.point_id).version for point in (first, middle, last)] == [
        2,
        2,
        2,
    ]
    assert _stored(repository, first.point_id).updated_by == ACTOR


async def test_soft_delete_point_leaves_the_new_head_without_a_previous_pointer() -> None:
    first, second, third = _chain(3)
    repository = _seeded(first, second, third)

    await soft_delete_point(repository, tenant_id=TENANT, point_id=first.point_id, actor=ACTOR)

    assert _stored(repository, second.point_id).previous_chunk_id is None


async def test_soft_delete_point_leaves_the_new_tail_without_a_next_pointer() -> None:
    first, second, third = _chain(3)
    repository = _seeded(first, second, third)

    await soft_delete_point(repository, tenant_id=TENANT, point_id=third.point_id, actor=ACTOR)

    assert _stored(repository, second.point_id).next_chunk_id is None


async def test_soft_delete_point_keeps_the_deleted_points_own_pointers() -> None:
    """Nothing active points at it any more, so its pointers record where it sat.

    That record is what the retry re-plans from, and what a later restore or an
    audit reader would need to place the point back in the sequence.
    """
    first, middle, last = _chain(3)
    repository = _seeded(first, middle, last)

    await soft_delete_point(repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR)

    stored = _stored(repository, middle.point_id)
    assert stored.previous_chunk_id == first.point_id
    assert stored.next_chunk_id == last.point_id


async def test_soft_delete_point_is_a_noop_for_an_already_inactive_point() -> None:
    """Not a `404`, and not a second relink — no patch is issued at all."""
    first, middle, last = _chain(3)
    repository = _seeded(first, middle, last)
    await soft_delete_point(repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR)
    versions_after_first = {
        point.point_id: _stored(repository, point.point_id).version
        for point in (first, middle, last)
    }
    repository.patch_batches.clear()

    again = await soft_delete_point(
        repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR
    )

    assert again.is_active is False
    assert repository.patch_batches == []
    assert {
        point.point_id: _stored(repository, point.point_id).version
        for point in (first, middle, last)
    } == versions_after_first


async def test_soft_delete_point_raises_not_found_for_another_tenants_point() -> None:
    first, middle, last = _chain(3)
    repository = _seeded(first, middle, last)

    with pytest.raises(PointNotFoundError):
        await soft_delete_point(
            repository, tenant_id=OTHER_TENANT, point_id=middle.point_id, actor=ACTOR
        )

    assert repository.patch_batches == []
    assert _stored(repository, middle.point_id).is_active is True


async def test_soft_delete_point_raises_not_found_for_an_unknown_point() -> None:
    repository = _seeded(*_chain(2))

    with pytest.raises(PointNotFoundError):
        await soft_delete_point(repository, tenant_id=TENANT, point_id=uuid.uuid4(), actor=ACTOR)


async def test_soft_delete_point_skips_a_neighbour_that_is_not_there() -> None:
    """A pointer naming an absent point means the chain was already broken.

    The delete completes the half of the relink that exists rather than
    refusing, which would leave the point unremovable through any endpoint.
    """
    first, middle, last = _chain(3)
    repository = _seeded(first, middle)  # `last` is never stored

    deleted = await soft_delete_point(
        repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR
    )

    assert deleted.is_active is False
    assert _stored(repository, first.point_id).next_chunk_id == last.point_id


async def test_soft_delete_point_visits_every_active_point_exactly_once_after_several_deletes() -> (
    None
):
    """The plan's traversal property, over a sequence of deletes.

    Walking `next_chunk_id` from the head must reach every surviving point once
    and never step into a deactivated one.
    """
    points = _chain(5)
    repository = _seeded(*points)

    for index in (1, 3):
        await soft_delete_point(
            repository, tenant_id=TENANT, point_id=points[index].point_id, actor=ACTOR
        )

    walked = _walk(repository, points[0].point_id)
    assert walked == [points[0].point_id, points[2].point_id, points[4].point_id]
    assert all(_stored(repository, point_id).is_active for point_id in walked)


@dataclass
class _ContendedRepository(FakePointRepository):
    """Bumps one point's version just before an apply, as a rival writer would.

    That makes the patch guarding on the old version match nothing while the
    rest of the batch lands — Qdrant's real behaviour, and the partial apply the
    service's retry exists to repair. `rounds` bounds how long the rival keeps
    interfering.
    """

    rival: uuid.UUID | None = None
    rounds: int = 0

    @override
    async def apply_patches(self, *, tenant_id: uuid.UUID, patches: Sequence[PayloadPatch]) -> None:
        if self.rounds > 0 and self.rival is not None:
            self.rounds -= 1
            victim = self.points[str(self.rival)]
            self.points[str(self.rival)] = victim.model_copy(update={"version": victim.version + 1})
        await super().apply_patches(tenant_id=tenant_id, patches=patches)


async def test_soft_delete_point_repairs_a_partially_applied_batch_on_retry() -> None:
    first, middle, last = _chain(3)
    repository = _ContendedRepository(rival=first.point_id, rounds=1)
    for point in (first, middle, last):
        repository.add(point)

    await soft_delete_point(repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR)

    # The first batch left `first` stale; the second re-planned only what was
    # still missing, rather than re-deactivating the already-inactive point.
    assert len(repository.patch_batches) == 2
    assert [patch.point_id for patch in repository.patch_batches[1]] == [first.point_id]
    assert _stored(repository, first.point_id).next_chunk_id == last.point_id
    assert _stored(repository, middle.point_id).is_active is False


async def test_soft_delete_point_conflicts_when_the_relink_never_settles() -> None:
    first, middle, last = _chain(3)
    repository = _ContendedRepository(rival=first.point_id, rounds=99)
    for point in (first, middle, last):
        repository.add(point)

    with pytest.raises(PointVersionConflictError):
        await soft_delete_point(repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR)


async def test_soft_delete_file_points_deactivates_every_active_point_of_the_file() -> None:
    points = _chain(3)
    other_file = _chain(2, file_id=OTHER_FILE)
    other_tenant = _chain(1, tenant=OTHER_TENANT, file_id=uuid.uuid4())
    repository = _seeded(*points, *other_file, *other_tenant)

    swept = await soft_delete_file_points(repository, tenant_id=TENANT, file_id=FILE, actor=ACTOR)

    assert swept == 3
    assert all(not _stored(repository, point.point_id).is_active for point in points)
    assert all(_stored(repository, point.point_id).is_active for point in other_file)
    assert all(_stored(repository, point.point_id).is_active for point in other_tenant)


async def test_soft_delete_file_points_leaves_the_chain_intact() -> None:
    """No survivor can dangle, so the sweep rewrites no pointer at all."""
    first, middle, last = _chain(3)
    repository = _seeded(first, middle, last)

    await soft_delete_file_points(repository, tenant_id=TENANT, file_id=FILE, actor=ACTOR)

    assert _stored(repository, first.point_id).next_chunk_id == middle.point_id
    assert _stored(repository, middle.point_id).previous_chunk_id == first.point_id
    assert _stored(repository, last.point_id).previous_chunk_id == middle.point_id


async def test_soft_delete_file_points_pages_past_one_batch() -> None:
    """More points than one sweep page, so the re-listing loop has to run."""
    points = _chain(230)
    repository = _seeded(*points)

    swept = await soft_delete_file_points(repository, tenant_id=TENANT, file_id=FILE, actor=ACTOR)

    assert swept == 230
    assert all(not _stored(repository, point.point_id).is_active for point in points)


async def test_soft_delete_file_points_is_a_noop_the_second_time() -> None:
    repository = _seeded(*_chain(3))
    await soft_delete_file_points(repository, tenant_id=TENANT, file_id=FILE, actor=ACTOR)
    repository.patch_batches.clear()

    swept = await soft_delete_file_points(repository, tenant_id=TENANT, file_id=FILE, actor=ACTOR)

    assert swept == 0
    assert repository.patch_batches == []


async def test_soft_delete_file_points_returns_zero_for_an_unknown_file() -> None:
    repository = _seeded(*_chain(2))

    assert (
        await soft_delete_file_points(
            repository, tenant_id=TENANT, file_id=uuid.uuid4(), actor=ACTOR
        )
        == 0
    )


async def test_soft_delete_point_logs_its_duration_and_round_count() -> None:
    """ADR-0011's `duration_ms`, plus the field that explains a slow one.

    Relinking is O(1), so a delete's cost is Qdrant round trips; `rounds` above
    1 means a concurrent writer forced a re-plan rather than the store being
    slow, and the two fields are only useful together.
    """
    first, middle, last = _chain(3)
    repository = _seeded(first, middle, last)

    with structlog.testing.capture_logs() as logs:
        await soft_delete_point(repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR)

    event = next(entry for entry in logs if entry["event"] == "points.soft_deleted")
    assert event["rounds"] == 1
    assert isinstance(event["duration_ms"], float)


async def test_soft_delete_file_points_logs_a_round_per_sweep_page() -> None:
    """The sweep is the delete path whose cost tracks the size of the file."""
    repository = _seeded(*_chain(230))

    with structlog.testing.capture_logs() as logs:
        await soft_delete_file_points(repository, tenant_id=TENANT, file_id=FILE, actor=ACTOR)

    event = next(entry for entry in logs if entry["event"] == "points.file_soft_deleted")
    assert event["points_soft_deleted"] == 230
    assert event["rounds"] == 3
    assert isinstance(event["duration_ms"], float)


async def test_soft_delete_file_points_conflicts_when_a_round_makes_no_progress() -> None:
    """A sweep round that attempts the same ids as the one before is stuck."""
    points = _chain(3)
    repository = _ContendedRepository(rival=points[0].point_id, rounds=99)
    for point in points:
        repository.add(point)

    with pytest.raises(PointVersionConflictError):
        await soft_delete_file_points(repository, tenant_id=TENANT, file_id=FILE, actor=ACTOR)