"""Record which business object an executed proposal actually created.

Deliberately thin, and deliberately unable to fail loudly. Linking happens
*after* the domain service has already committed the employee's request; if
this module raised, a request that genuinely succeeded would be reported to the
employee as failed and — because the failure path rolls the proposal back to
``pending`` — offered to them again. Losing a pointer is recoverable. Telling
someone their laptop request failed when it did not is not.

So every miss is swallowed into a stored reason, and the reasons are counted
rather than logged: "the contract has a gap here" and "nobody is submitting
anything" look identical on a funnel that only counts links.
"""

from __future__ import annotations

import logging
import uuid
from typing import Any

from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from app.assistant.models import (
    AssistantActionProposal,
    AssistantConversation,
    AssistantObjectLink,
)
from app.assistant.outcome_contract import locate, source_tool_of
from app.core.txn import best_effort

logger = logging.getLogger(__name__)


async def link_executed_proposal(
    db: AsyncSession,
    *,
    proposal: AssistantActionProposal,
    result: Any,
    trace_id: str | None = None,
) -> tuple[AssistantObjectLink | None, str | None]:
    """Point one executed proposal at the row it created.

    Returns ``(link, miss_reason)``; exactly one is set. The caller commits —
    this runs inside the same transaction as the status update so a link can
    never outlive the execution that produced it, nor go missing after one.
    """
    located, miss = locate(proposal.action_key, result)
    if located is None:
        return None, miss

    # Idempotent by the unique constraint's own key. Checked first rather than
    # relying on an integrity error, because a duplicate here is the *expected*
    # path (a retried Feishu callback), not an exceptional one — and an
    # integrity error reaching the session poisons the transaction that is
    # still carrying the employee's request (see the savepoint below).
    existing = (
        await db.execute(
            select(AssistantObjectLink).where(
                AssistantObjectLink.proposal_id == proposal.id,
                AssistantObjectLink.resource_type == located.resource_type,
                AssistantObjectLink.resource_id == located.resource_id,
            )
        )
    ).scalar_one_or_none()
    if existing is not None:
        return existing, None

    service_request_id: uuid.UUID | None = None
    if proposal.conversation_id is not None:
        conversation = await db.get(AssistantConversation, proposal.conversation_id)
        if conversation is not None:
            service_request_id = conversation.service_request_id

    # A nested savepoint, not a bare flush — and it is load-bearing, not tidy.
    #
    # A failed flush poisons the entire session. For a **curated** action the
    # domain write has NOT been committed yet: ``_execute_ticket`` writes the
    # ticket through this same session and ``_run_claimed`` commits afterwards.
    # So without the savepoint, a bad link write would roll back the employee's
    # ticket along with it, and the swallow one layer up would then commit a
    # proposal marked ``executed`` with nothing behind it. The savepoint is what
    # makes "a link failure never costs the employee their request" true rather
    # than merely intended.
    link = AssistantObjectLink(
        proposal_id=proposal.id,
        conversation_id=proposal.conversation_id,
        service_request_id=service_request_id,
        person_id=proposal.person_id,
        action_key=proposal.action_key,
        source_tool=source_tool_of(result),
        resource_type=located.resource_type,
        resource_id=located.resource_id,
        resolver_key=located.resolver_key,
        resource_number=located.number,
        deep_link=located.deep_link,
        covers_count=located.covers,
    )
    try:
        async with db.begin_nested():
            db.add(link)
            await db.flush()
        await _audit_link(db, proposal=proposal, link=link, trace_id=trace_id)
    except IntegrityError:
        # Lost the race to a concurrent confirm of the same proposal. The other
        # writer's row is the answer; this is the expected outcome of a double
        # callback, not an error.
        logger.info(
            "assistant: link for proposal %s already written concurrently",
            proposal.id,
        )
        return (
            await db.execute(
                select(AssistantObjectLink).where(
                    AssistantObjectLink.proposal_id == proposal.id,
                    AssistantObjectLink.resource_type == located.resource_type,
                    AssistantObjectLink.resource_id == located.resource_id,
                )
            )
        ).scalar_one_or_none(), None
    return link, None


async def _audit_link(
    db: AsyncSession,
    *,
    proposal: AssistantActionProposal,
    link: AssistantObjectLink,
    trace_id: str | None,
) -> None:
    """Tie the link to the trace that produced it.

    The row itself is the pointer; this is the evidence of *when and under
    which trace* the platform decided that pointer was right.

    Wrapped in ``best_effort`` rather than a bare ``try/except``: an audit write
    is a DB write, so swallowing its failure without a savepoint would leave a
    poisoned session behind an exception nobody sees. ``best_effort`` also lets
    the deadlock/abort errors through, which is correct — those mean the
    surrounding transaction is already dead server-side and continuing would
    only commit nothing quietly.
    """
    from app.audit.service import AuditService  # noqa: PLC0415 — cycle

    async with best_effort(db, label="record assistant object link", log=logger):
        await AuditService(db).log(
            trace_id=trace_id or proposal.trace_id or "",
            # ``actor_type="user"`` + ``actor_user_id=<person id>`` is the
            # assistant domain's existing convention (card_actions,
            # admin_service) — the person pressed the button.
            actor_type="user",
            actor_user_id=proposal.person_id,
            action="assistant.proposal.linked",
            status="success",
            resource_type=link.resource_type,
            resource_id=link.resource_id,
            detail={
                "proposal_id": str(proposal.id),
                "action_key": proposal.action_key,
                "source_tool": link.source_tool,
                "resolver_key": link.resolver_key,
                "resource_number": link.resource_number,
                "covers_count": link.covers_count,
            },
        )


async def links_for_proposal(
    db: AsyncSession, proposal_id: uuid.UUID,
) -> list[AssistantObjectLink]:
    return list(
        (
            await db.execute(
                select(AssistantObjectLink)
                .where(AssistantObjectLink.proposal_id == proposal_id)
                .order_by(AssistantObjectLink.created_at)
            )
        ).scalars().all()
    )


async def links_for_conversation(
    db: AsyncSession, conversation_id: uuid.UUID, *, person_id: uuid.UUID,
) -> list[AssistantObjectLink]:
    """Everything this conversation submitted, newest last.

    Filtered by ``person_id`` as well as conversation: a conversation belongs
    to one person today, and this stays true if that ever stops being the case.
    """
    return list(
        (
            await db.execute(
                select(AssistantObjectLink)
                .where(
                    AssistantObjectLink.conversation_id == conversation_id,
                    AssistantObjectLink.person_id == person_id,
                )
                .order_by(AssistantObjectLink.created_at)
            )
        ).scalars().all()
    )