"""Core-owned agent purchase reservations and exactly-once ledger fulfillment."""

import hashlib
import json
from uuid import UUID, uuid4

from pydantic import Field
from sqlalchemy import BigInteger, cast, func, or_, select

from src.core.exceptions import ConflictError, ForbiddenError, NotFoundError, ValidationError
from src.domain.agent_access.models import AgentOperation
from src.domain.agent_access.operations import AgentOperationService
from src.domain.agent_access.rate_limit import enforce_tool_rate_limit
from src.domain.agent_access.schemas import StrictModel
from src.domain.agent_access.service import prompt_source_for, utcnow
from src.domain.billing.models import UsageCreditPackage, UsageCreditPurchase
from src.domain.billing.repository import StripeCustomerRepository, WorkspaceCreditLedgerRepository


class ListCreditPackages(StrictModel):
    workspace_id: UUID | None = None


class CreditPurchaseRequest(StrictModel):
    request_key: str = Field(min_length=8, max_length=128)
    # Which workspace's saved card / credit balance; required only when the connection
    # covers several workspaces (see list_workspaces).
    workspace_id: UUID | None = None
    package_slug: str = Field(min_length=1, max_length=50)
    expected_amount_minor: int = Field(gt=0, le=100_000_000)
    currency: str = Field(pattern=r"^[a-z]{3}$")


def require_purchase_budget(policy: dict, amount: int, currency: str, reserved: int):
    if not policy.get("saved_payment_consent") or currency != policy.get("currency"):
        raise ForbiddenError("Saved-payment consent is not available for this currency")
    if amount <= 0 or amount > policy.get("per_purchase_minor", 0):
        raise ForbiddenError("Per-purchase spending limit exceeded")
    if reserved + amount > policy.get("monthly_minor", 0):
        raise ForbiddenError("Monthly spending limit exceeded")


class AgentPurchaseService:
    def __init__(self, access):
        self.access = access
        self.repo = access.repo
        self.db = self.repo.db

    async def catalog(self, principal, workspace_id: UUID | None = None):
        # Prices are global, so with no workspace named it is enough that billing.read
        # is usable in at least one of the connection's workspaces.
        if workspace_id is not None:
            await self.access.require_scope(principal, "billing.read", workspace_id=workspace_id)
        elif not await self.access.scoped_workspaces(principal, "billing.read"):
            raise ForbiddenError("Agent scope not granted")
        packages = list(
            (
                await self.db.scalars(
                    select(UsageCreditPackage)
                    .where(UsageCreditPackage.is_active.is_(True))
                    .order_by(UsageCreditPackage.sort_order)
                    .limit(100)
                )
            ).all()
        )
        return {
            "packages": [
                {
                    "slug": p.slug,
                    "name": p.name,
                    "credits": p.units,
                    "amount_minor": p.price_minor,
                    "currency": p.currency,
                }
                for p in packages
            ]
        }

    async def reserved_spend(self, grant_id, workspace_id):
        # Limits are configured per workspace, so spend is counted per workspace too.
        month = utcnow().replace(day=1, hour=0, minute=0, second=0, microsecond=0)
        # Unresolved purchases from prior months keep consuming the cap until reconciled.
        amount = cast(AgentOperation.arguments["amount_minor"].as_string(), BigInteger)
        return int(
            (
                await self.db.scalar(
                    select(func.coalesce(func.sum(amount), 0)).where(
                        AgentOperation.grant_id == grant_id,
                        AgentOperation.workspace_id == workspace_id,
                        AgentOperation.action == "purchase_credits",
                        AgentOperation.status.not_in(["failed", "canceled"]),
                        or_(
                            AgentOperation.status != "completed", AgentOperation.updated_at >= month
                        ),
                    )
                )
            )
            or 0
        )

    async def submit(self, principal, action, body):
        if not self.access.settings.stripe_secret_key:
            raise ValidationError("Credit purchasing is unavailable")
        scope = "credits.purchase" if action == "purchase_credits" else "credits.checkout"
        workspace_id = await self.access.require_scope(
            principal, scope, workspace_id=body.workspace_id
        )
        # Pin the resolved workspace so the same request_key hashes identically
        # whether the caller named the workspace or it was implied.
        body = body.model_copy(update={"workspace_id": workspace_id})
        grant = await self.repo.grant(principal.id, lock=True)
        self.access.require_active(grant)
        digest = hashlib.sha256(
            json.dumps(
                {"action": action, "body": body.model_dump(mode="json")},
                sort_keys=True,
                separators=(",", ":"),
            ).encode()
        ).hexdigest()
        previous = await self.repo.operation(grant.id, body.request_key)
        if previous:
            if previous.request_hash != digest:
                raise ConflictError("Request key was already used with different arguments")
            return AgentOperationService.view(previous)
        # Only a genuinely new submission consumes the rate limit: a retry of an
        # already-accepted request_key returned above without reaching this point.
        await enforce_tool_rate_limit(
            self.access.redis, grant_id=grant.id, workspace_id=workspace_id, tool_name=action
        )
        package = (
            await self.db.execute(
                select(UsageCreditPackage).where(
                    UsageCreditPackage.slug == body.package_slug,
                    UsageCreditPackage.is_active.is_(True),
                )
            )
        ).scalar_one_or_none()
        if not package or package.units <= 0:
            raise NotFoundError("Credit package not found")
        if package.price_minor != body.expected_amount_minor or package.currency != body.currency:
            raise ValidationError("Package price changed; refresh the credit catalog")
        customer = await StripeCustomerRepository(self.db).get_by_workspace(workspace_id)
        if not customer:
            raise ValidationError("Complete billing setup in the dashboard first")
        if action == "purchase_credits":
            access_row = await self.access.access_row(grant, workspace_id)
            require_purchase_budget(
                access_row.purchase_policy,
                package.price_minor,
                package.currency,
                await self.reserved_spend(grant.id, workspace_id),
            )
        operation = await self.repo.add(
            AgentOperation(
                id=uuid4(),
                grant_id=grant.id,
                workspace_id=workspace_id,
                request_key=body.request_key,
                action=action,
                request_hash=digest,
                prompt_source=prompt_source_for(self.access.settings, principal),
                arguments={
                    "package_id": str(package.id),
                    "package_slug": package.slug,
                    "units": package.units,
                    "amount_minor": package.price_minor,
                    "currency": package.currency,
                    "customer_id": customer.stripe_customer_id,
                },
                status="queued",
                result={},
                created_at=utcnow(),
                updated_at=utcnow(),
            )
        )
        return AgentOperationService.view(operation)

    async def fulfill(self, operation, payment: dict, *, checkout: bool):
        """Call only with an object retrieved server-side from Stripe. Caller locks operation."""
        if operation.status == "completed":
            return
        snapshot = operation.arguments
        amount = payment.get("amount_total" if checkout else "amount_received")
        paid = (
            payment.get("payment_status") == "paid"
            if checkout
            else payment.get("status") == "succeeded"
        )
        if not paid:
            return
        if (
            amount != snapshot["amount_minor"]
            or payment.get("currency") != snapshot["currency"]
            or payment.get("customer") != snapshot["customer_id"]
            or (payment.get("metadata") or {}).get("agent_operation_id") != str(operation.id)
        ):
            operation.status = "manual_review"
            operation.result = {"error": "Payment requires reconciliation"}
            return
        purchase = UsageCreditPurchase(
            id=operation.id,
            workspace_id=operation.workspace_id,
            stripe_customer_id=snapshot["customer_id"],
            stripe_checkout_session_id=payment["id"] if checkout else None,
            stripe_payment_intent_id=payment.get("payment_intent") if checkout else payment["id"],
            usage_credit_package_id=UUID(snapshot["package_id"]),
            package_slug=snapshot["package_slug"],
            units_purchased=snapshot["units"],
            amount_minor=snapshot["amount_minor"],
            currency=snapshot["currency"],
            status="paid",
            paid_at=utcnow(),
        )
        await self.repo.add(purchase)
        await WorkspaceCreditLedgerRepository(self.db).add_entry(
            operation.workspace_id,
            "purchase",
            snapshot["units"],
            source_bucket="purchased_topup",
            source_type="agent_credit_purchase",
            source_id=str(operation.id),
            usage_credit_purchase_id=purchase.id,
        )
        operation.status = "completed"
        operation.updated_at = utcnow()
        operation.result = {
            "credits_added": snapshot["units"],
            "amount_minor": snapshot["amount_minor"],
            "currency": snapshot["currency"],
        }