"""PDFs waiting to be told whose they are.

A PDF forwarded into the self-chat has no caption to read: WhatsApp shows no
caption box when a document is forwarded, and the message arrives with the
field absent rather than empty. The identifier comes as the next plain text
message instead, so a forwarded PDF has to be held between arriving and being
filed.

The queue is strictly FIFO. When several PDFs arrive before any text, the next
text belongs to the oldest of them — which is the order somebody sending them
would assume, and the only order that is not a guess.
"""

from __future__ import annotations

import logging
from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Any

logger = logging.getLogger(__name__)


@dataclass(frozen=True, slots=True)
class PendingPdf:
    """A downloaded PDF with nobody attached to it yet."""

    message_id: str
    path: str
    sender: str
    received_at: datetime
    file_name: str | None = None
    mime_type: str | None = None
    size_bytes: int | None = None
    classification: dict[str, Any] = field(default_factory=dict)

    intake_reference: str | None = None
    """Which CMS record this belongs to (ADR-0010).

    Stored beside the document rather than in a dictionary in the process: a
    PDF 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.
    """

    def waited(self, now: datetime | None = None) -> float:
        """Seconds since it arrived."""
        return ((now or datetime.now(UTC)) - self.received_at).total_seconds()


class InMemoryPendingPdfs:
    """The fallback, for a deployment with no database.

    Honest rather than convenient: it works, and it loses a queued document on
    restart. The warning that selects it says so.
    """

    def __init__(self) -> None:
        self._items: list[PendingPdf] = []

    def add(self, pdf: PendingPdf) -> None:
        if any(item.message_id == pdf.message_id for item in self._items):
            return

        self._items.append(pdf)

    def take_oldest(self) -> PendingPdf | None:
        return self._items.pop(0) if self._items else None

    def take_older_than(self, cutoff: datetime) -> list[PendingPdf]:
        stale = [item for item in self._items if item.received_at < cutoff]
        self._items = [item for item in self._items if item.received_at >= cutoff]

        return stale

    def __len__(self) -> int:
        return len(self._items)


class PostgresPendingPdfs:
    """The durable queue.

    Taking is a delete-and-return in one statement. Two processes draining the
    same inbox must not both be handed the same document — one identifier would
    file two copies, or the second would file the wrong client's statement.
    """

    def __init__(self, connect) -> None:
        # A callable, matching PostgresRunStore: built at startup, possibly
        # before the database is reachable.
        self._connect = connect

    def add(self, pdf: PendingPdf) -> None:
        from psycopg.types.json import Jsonb

        with self._connect() as connection, connection.cursor() as cursor:
            cursor.execute(
                """
                INSERT INTO pending_pdfs
                    (message_id, path, sender, file_name, mime_type, size_bytes,
                     classification, received_at, intake_reference)
                VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
                ON CONFLICT (message_id) DO NOTHING
                """,
                (
                    pdf.message_id,
                    pdf.path,
                    pdf.sender,
                    pdf.file_name,
                    pdf.mime_type,
                    pdf.size_bytes,
                    Jsonb(pdf.classification or {}),
                    pdf.received_at,
                    pdf.intake_reference,
                ),
            )
            connection.commit()

    def take_oldest(self) -> PendingPdf | None:
        with self._connect() as connection, connection.cursor() as cursor:
            cursor.execute(
                """
                DELETE FROM pending_pdfs
                WHERE id = (SELECT id FROM pending_pdfs ORDER BY id LIMIT 1 FOR UPDATE SKIP LOCKED)
                RETURNING message_id, path, sender, file_name, mime_type, size_bytes,
                          classification, received_at, intake_reference
                """
            )
            row = cursor.fetchone()
            connection.commit()

        return _from_row(row)

    def take_older_than(self, cutoff: datetime) -> list[PendingPdf]:
        with self._connect() as connection, connection.cursor() as cursor:
            cursor.execute(
                """
                DELETE FROM pending_pdfs
                WHERE id IN (SELECT id FROM pending_pdfs WHERE received_at < %s
                             ORDER BY id FOR UPDATE SKIP LOCKED)
                RETURNING message_id, path, sender, file_name, mime_type, size_bytes,
                          classification, received_at, intake_reference
                """,
                (cutoff,),
            )
            rows = cursor.fetchall()
            connection.commit()

        return [pdf for pdf in (_from_row(row) for row in rows) if pdf]

    def __len__(self) -> int:
        with self._connect() as connection, connection.cursor() as cursor:
            cursor.execute("SELECT COUNT(*) FROM pending_pdfs")

            return int(cursor.fetchone()[0])


def _from_row(row) -> PendingPdf | None:
    if not row:
        return None

    (message_id, path, sender, file_name, mime_type, size_bytes,
     classification, received_at, intake_reference) = row

    return PendingPdf(
        message_id=message_id,
        path=path,
        sender=sender,
        file_name=file_name,
        mime_type=mime_type,
        size_bytes=size_bytes,
        classification=classification or {},
        received_at=received_at,
        # Unpacked above and then dropped on the floor for a while, which made
        # the whole reference mechanism inert: every read returned None, so the
        # reconcile believed no document was held and failed them all — and its
        # take-and-re-add then wrote the None back, destroying the reference in
        # the database as well.
        intake_reference=intake_reference,
    )