"""Checking that the two databases still agree (ADR-0010).

The CMS holds the record, the agent holds the work, and there is no transaction
between them — so the module's central claim, that nothing is in a state nobody
can see, is something to verify rather than assume. A claim that is never checked
is a hope.

The other half of this file is smaller and more embarrassing: the reference
linking a held document to its CMS record used to live in a dictionary in the
process, so a PDF held overnight came back unable to address its own record and
silently stopped reporting. The record stayed visible and stopped being true.
"""

from __future__ import annotations

from datetime import UTC, datetime, timedelta

import pytest

from app.intake import InMemoryOutbox
from app.whatsapp.inbox import InboxFilter
from app.whatsapp.messages import InboundMessage, MediaReference, MessageType
from app.whatsapp.pending import InMemoryPendingPdfs, PendingPdf
from app.whatsapp.webhook import InboundQueue
from tests.intake_support import RecordingCms

OWNER = "923049637232"


class TrackingCms(RecordingCms):
    def __init__(self) -> None:
        super().__init__()
        self.pushes: list[tuple[str, str | None, str | None]] = []
        self.unfinished: list[dict] = []

    def update_intake(self, reference, **fields):
        self.pushes.append((reference, fields.get("status"), fields.get("reason")))

        return {"reference": reference}

    def pending_intake(self, status=None, per_page=None):
        records = [r for r in self.unfinished if status is None or r.get("status") == status]

        return records, False


@pytest.fixture
def container(tmp_path):
    class FakeContainer:
        inbound_queue = InboundQueue()
        incoming_dir = tmp_path
        inbox = InboxFilter()
        pending_pdfs = InMemoryPendingPdfs()
        model = None
        cms = TrackingCms()
        intake_outbox = InMemoryOutbox()

        class whatsapp:
            @staticmethod
            def download_media(media, destination):
                destination.write_bytes(b"%PDF-1.4 statement")

                return destination

    return FakeContainer


def held(message_id="msg-1", reference="INT-1001", age=timedelta(0)):
    return PendingPdf(
        message_id=message_id,
        path="incoming/x.pdf",
        sender=OWNER,
        received_at=datetime.now(UTC) - age,
        file_name="statement.pdf",
        intake_reference=reference,
    )


class TestTheTwoDatabasesAgree:
    def test_a_document_held_here_and_waiting_there_is_fine(self, container):
        from app import __main__ as entry

        container.pending_pdfs.add(held())
        container.cms.unfinished = [{"reference": "INT-1001", "status": "waiting"}]

        entry._reconcile(container)

        assert container.cms.pushes == []

    def test_a_waiting_record_nobody_is_holding_is_marked_failed(self, container):
        from app import __main__ as entry

        # The record promises something nothing will keep: supplying a file
        # number would make nothing happen, because no document is waiting.
        container.cms.unfinished = [{"reference": "INT-1042", "status": "waiting"}]

        entry._reconcile(container)

        assert len(container.cms.pushes) == 1
        reference, status, reason = container.cms.pushes[0]
        assert (reference, status) == ("INT-1042", "failed")
        assert "no longer holding" in reason

    def test_reconciling_does_not_empty_the_queue_it_checks(self, container):
        from app import __main__ as entry

        container.pending_pdfs.add(held("a", "INT-1"))
        container.pending_pdfs.add(held("b", "INT-2"))
        container.cms.unfinished = [
            {"reference": "INT-1", "status": "waiting"},
            {"reference": "INT-2", "status": "waiting"},
        ]

        entry._reconcile(container)

        # take_older_than removes what it returns. A check that consumed the
        # queue would destroy exactly the documents it was verifying.
        assert len(container.pending_pdfs) == 2

    def test_other_states_are_left_alone(self, container):
        from app import __main__ as entry

        # Processing is the startup reconciliation's business, and Approval
        # belongs to a reviewer. Only Waiting makes a claim about what the agent
        # is holding.
        container.cms.unfinished = [
            {"reference": "INT-1", "status": "processing"},
            {"reference": "INT-2", "status": "approval"},
        ]

        entry._reconcile(container)

        assert container.cms.pushes == []

    def test_a_cms_that_cannot_be_asked_changes_nothing(self, container):
        from app import __main__ as entry

        class Silent(TrackingCms):
            def pending_intake(self, status=None, per_page=None):
                raise ConnectionError("connection refused")

        container.cms = Silent()
        container.pending_pdfs.add(held())

        entry._reconcile(container)

        assert len(container.pending_pdfs) == 1


class TestTheReferenceSurvivesARestart:
    def test_it_is_stored_beside_the_document(self, container):
        from app import __main__ as entry

        message = InboundMessage(
            provider_message_id="msg-1",
            sender=OWNER,
            type=MessageType.DOCUMENT,
            text="",
            media=MediaReference(
                handle="h", mime_type="application/pdf",
                filename="Account Statement.pdf", size_bytes=6910,
            ),
        )

        from app.documents.classifier import Classification, Method

        entry._hold_for_metadata(
            container,
            message,
            Classification(
                filing_type="bank_statement", method=Method.FILENAME,
                confidence=0.9, reason="filename", was_read=False,
            ),
        )

        stored = container.pending_pdfs.take_oldest()

        # Held in the row rather than in a dictionary in the process: a document
        # held overnight outlives the process holding it, and a reference lost on
        # restart means every status change afterwards goes nowhere — freezing
        # the record at whatever it last said while looking perfectly healthy.
        assert stored.intake_reference is not None
        assert stored.intake_reference.startswith("INT-")

    def test_a_document_held_across_a_restart_still_reports(self, container):
        from app import __main__ as entry

        # A fresh process: nothing in the in-memory map, exactly as after a boot.
        entry._REFERENCES.clear()

        container.pending_pdfs.add(held(reference="INT-1234", age=timedelta(days=2)))

        entry._file_pdfs_nobody_named(container)

        # It reported against the reference from the row.
        assert any(reference == "INT-1234" for reference, _, _ in container.cms.pushes)


class TestWaitingDocumentsSurviveTheReconcile:
    """The reconcile must never fail a document the agent is holding.

    It decides by asking which references the agent has in pending_pdfs. Any
    route into Waiting that forgets to record the reference therefore looks, to
    the reconcile, like a record nobody is holding — and gets marked Failed once
    an hour while the agent waits patiently for the identifier that would have
    finished it.

    That is what happened: `_queue_after_reading`, the main route into Waiting
    for Information, built its PendingPdf without one.
    """

    def test_a_document_queued_after_reading_records_its_reference(self, container):
        from app import __main__ as entry
        from app.documents.classifier import Classification, Method

        message = InboundMessage(
            provider_message_id="wait-1",
            sender=OWNER,
            type=MessageType.DOCUMENT,
            text="",
            media=MediaReference(
                handle="h", mime_type="image/jpeg",
                filename="photo.jpg", size_bytes=4096,
            ),
        )

        entry._REFERENCES["wait-1"] = "INT-9001"

        path = container.incoming_dir / "wait-1.jpg"
        path.write_bytes(b"\xff\xd8\xff\xe0 jpeg")

        entry._queue_after_reading(
            container, message,
            Classification(filing_type="unknown", method=Method.NONE, confidence=0.0, reason="none"),
            path,
        )

        held = container.pending_pdfs.take_oldest()

        assert held.intake_reference == "INT-9001"

    def test_the_reconcile_leaves_a_held_document_alone(self, container):
        from app import __main__ as entry
        from app.documents.classifier import Classification, Method

        message = InboundMessage(
            provider_message_id="wait-2",
            sender=OWNER,
            type=MessageType.DOCUMENT,
            text="",
            media=MediaReference(
                handle="h", mime_type="image/jpeg",
                filename="photo.jpg", size_bytes=4096,
            ),
        )

        entry._REFERENCES["wait-2"] = "INT-9002"

        path = container.incoming_dir / "wait-2.jpg"
        path.write_bytes(b"\xff\xd8\xff\xe0 jpeg")

        entry._queue_after_reading(
            container, message,
            Classification(filing_type="unknown", method=Method.NONE, confidence=0.0, reason="none"),
            path,
        )

        container.cms.unfinished = [{"reference": "INT-9002", "status": "waiting"}]

        entry._reconcile(container)

        # Nothing failed. The agent is holding it, which is the whole point of
        # the state it is in.
        assert container.cms.pushes == []
        assert len(container.pending_pdfs) == 1


class TestTheReferenceSurvivesTheStore:
    """Reading a held document back must return everything it was stored with.

    `_from_row` unpacked intake_reference and never passed it to the constructor.
    Every read therefore returned None, which made the reference mechanism inert
    — and the reconcile, which takes every row and puts it back, then wrote the
    None into the database. A silent read fault that became a silent write fault.

    In memory, because the defect was in the mapping rather than the SQL, and a
    mapping is exactly what an in-memory store cannot exercise. These use the
    real row shape instead.
    """

    def test_a_row_maps_back_to_the_reference_it_was_stored_with(self):
        from datetime import UTC, datetime

        from app.whatsapp.pending import _from_row

        row = (
            "msg-1", "incoming/x.pdf", "923049637232", "x.pdf", "application/pdf",
            1234, {"workflow": "document_intake"}, datetime.now(UTC), "INT-4242",
        )

        mapped = _from_row(row)

        assert mapped.intake_reference == "INT-4242"
        assert mapped.message_id == "msg-1"

    def test_every_stored_field_survives_the_mapping(self):
        from datetime import UTC, datetime

        from app.whatsapp.pending import _from_row

        now = datetime.now(UTC)
        row = (
            "msg-2", "incoming/y.pdf", "923049637232", "y.pdf", "image/jpeg",
            99, {"filing_type": "cnic"}, now, "INT-4243",
        )

        mapped = _from_row(row)

        # Asserted field by field rather than by count, so a column added to the
        # query and forgotten in the constructor fails here instead of silently
        # becoming None the way intake_reference did.
        assert (mapped.message_id, mapped.path, mapped.sender) == ("msg-2", "incoming/y.pdf", "923049637232")
        assert (mapped.file_name, mapped.mime_type, mapped.size_bytes) == ("y.pdf", "image/jpeg", 99)
        assert mapped.classification == {"filing_type": "cnic"}
        assert mapped.received_at == now
        assert mapped.intake_reference == "INT-4243"


class TestAHeldDocumentReachesWaitingLegally:
    """The CMS will not let a document jump from Received to Waiting.

    `AiIntakeDocument::TRANSITIONS` allows Received → Processing or Failed, and
    nothing else. Waiting is reachable only from Processing — because a document
    that is waiting was worked on first, and arriving already stuck is not a
    thing that happens.

    On 2026-08-04 the hold path pushed Waiting straight after registration and
    the CMS refused it with a 409. The failure was quiet in the worst way: the
    push is best-effort, so the file was still held safely while the Intake
    Center went on showing Received for a document nothing would ever move.
    """

    #: The map this asserts against, copied from the CMS rather than invented.
    LEGAL = {
        "received": {"processing", "failed"},
        "processing": {"waiting", "approval", "completed", "failed"},
        "waiting": {"processing", "approval", "completed", "failed"},
        "approval": {"processing", "waiting", "completed", "failed"},
        "completed": set(),
    }

    def test_every_pushed_sequence_is_a_legal_walk(self):
        """Processing before Waiting, and each step allowed by the one before."""
        pushed = ["processing", "waiting"]

        state = "received"

        for status in pushed:
            assert status in self.LEGAL[state], (
                f"{state} -> {status} is not a legal transition; "
                "the CMS answers 409 and the record is stranded"
            )
            state = status

    def test_waiting_straight_from_received_is_rejected(self):
        """The bug itself, stated as a rule rather than a story."""
        assert "waiting" not in self.LEGAL["received"]


class TestArrivalsAreAnnouncedBeforeAnyIsRead:
    """Documents sent together must appear together (ADR-0010).

    Observed on 2026-08-04: four documents sent, one visible, a person asking
    where the other three were. They were in this process's memory, waiting for
    their turn to be registered — which is the exact invisibility the Intake
    Center was built to end.

    Driven through the real _process_inbox with the real classifier, router and
    take-in gate. The last three intake bugs all shipped behind green tests
    whose fakes were more permissive than the system; these are not offered the
    chance.
    """

    def _message(self, n: int):
        from app.whatsapp.messages import InboundMessage, MediaReference, MessageType

        return InboundMessage(
            provider_message_id=f"msg-{n}",
            sender=OWNER,
            type=MessageType.IMAGE,
            media=MediaReference(
                handle=f"handle-{n}",
                mime_type="image/jpeg",
                filename=f"photo-{n}.jpg",
                size_bytes=2048,
            ),
            from_me=True,
        )

    def _container(self, tmp_path, count: int):
        from types import SimpleNamespace

        from app.whatsapp.inbox import InboxFilter
        from app.whatsapp.webhook import InboundQueue
        from app.workflow.state import RunState

        class Engine:
            def __init__(self):
                self.started = []

            def start(self, workflow, inputs):
                self.started.append(inputs)

                return SimpleNamespace(
                    id=f"run-{len(self.started)}",
                    state=RunState.AWAITING_APPROVAL,
                    steps=[],
                    error=None,
                )

        class Container:
            inbound_queue = InboundQueue()
            incoming_dir = tmp_path
            inbox = InboxFilter()
            pending_pdfs = InMemoryPendingPdfs()
            model = None
            cms = TrackingCms()
            intake_outbox = InMemoryOutbox()
            engine = Engine()

            class whatsapp:
                @staticmethod
                def download_media(media, destination):
                    destination.write_bytes(b"\xff\xd8\xff\xe0" + b"jpeg-bytes" * 4)

                    return destination

        for n in range(1, count + 1):
            Container.inbound_queue.put(self._message(n))

        return Container

    def test_every_arrival_is_queued_before_the_first_is_read(self, tmp_path):
        from app import __main__ as entry

        entry._REFERENCES.clear()
        entry._TAKEN.clear()

        container = self._container(tmp_path, count=3)

        entry._process_inbox(container)

        # All three reached the CMS, each carrying its file's path.
        assert len(container.cms.registered) == 3

        statuses = [status for _, status, _ in container.cms.pushes]
        first_processing = statuses.index("processing")

        # The order IS the requirement. Interleaving would mean the third
        # document a person sends stays invisible until the second finishes —
        # the complaint this exists to end.
        assert statuses[:first_processing].count("queued") == 3, statuses

    def test_the_gate_runs_once_per_message_not_once_per_pass(self, tmp_path):
        from app import __main__ as entry

        entry._REFERENCES.clear()
        entry._TAKEN.clear()

        container = self._container(tmp_path, count=2)

        entry._process_inbox(container)

        # Two passes ask; one take-in answers. Without the cache every counter
        # doubles and every document registers twice.
        assert len(container.cms.registered) == 2
        assert container.inbox.forwards_accepted == 2

    def test_the_take_in_cache_does_not_outlive_the_batch(self, tmp_path):
        from app import __main__ as entry

        entry._REFERENCES.clear()
        entry._TAKEN.clear()

        container = self._container(tmp_path, count=2)
        entry._process_inbox(container)

        assert entry._TAKEN == {}