"""Billing domain - limited-slot pricing campaigns (e.g. the founding-customer offer).

Core decides who gets a campaign price; Stripe only executes it. A campaign price is a
dedicated Stripe Price the catalog sync never rotates, so a subscriber keeps that amount
for life even when the catalog price later changes. Credits, top-ups and metered usage
are not part of the subscription price and are billed at regular rates.

Slot lifecycle (mirrors reserve / finalize / release used for Cloud usage):
  reserve  -> checkout creation claims a slot under a per-campaign advisory lock
  confirm  -> checkout.session.completed makes the claim permanent
  release  -> checkout.session.expired frees it; unconfirmed reservations also stop
              counting once ``expires_at`` passes, so the cap self-heals without webhooks
"""
from __future__ import annotations
import structlog
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Awaitable, Callable, Iterable, Optional, Protocol, TypeVar
from uuid import UUID

from sqlalchemy.exc import ProgrammingError

from src.core.exceptions import NotFoundError, ValidationError
from src.domain.billing.campaign_repository import PricingCampaignRepository
from src.domain.billing.models import PricingCampaign, PricingCampaignPrice

logger = structlog.get_logger()

FOUNDING_CAMPAIGN_SLUG = "founding"
CAMPAIGN_INTERVALS = ("month", "year")
# Stripe Checkout sessions live 24h; keep the reservation slightly longer so a session that
# completes right at the edge still finds its slot.
RESERVATION_TTL = timedelta(hours=25)


@dataclass(frozen=True)
class CampaignReservation:
    redemption_id: UUID
    campaign_slug: str
    stripe_price_id: str
    price_minor: int


@dataclass(frozen=True)
class PlanOffer:
    """Display shape for a plan's campaign prices (minor units)."""

    campaign_slug: str
    price_monthly: Optional[int]
    price_yearly: Optional[int]
    spots_remaining: int


def workspace_subject_key(workspace_id: UUID) -> str:
    return f"ws:{workspace_id}"


def email_subject_key(email: str) -> str:
    return f"email:{email.strip().lower()}"


class _PlanLike(Protocol):
    id: UUID
    slug: str


def _utcnow() -> datetime:
    return datetime.now(timezone.utc)


T = TypeVar("T")


def _is_missing_campaign_table(exc: ProgrammingError) -> bool:
    return "pricing_campaign" in str(exc)


class PricingCampaignService:
    def __init__(self, repo: PricingCampaignRepository, campaign_slug: str = FOUNDING_CAMPAIGN_SLUG):
        self._repo = repo
        self._slug = campaign_slug

    async def _open_campaign(self, now: datetime) -> Optional[PricingCampaign]:
        campaign = await self._repo.get_by_slug(self._slug)
        if not campaign or not campaign.is_active:
            return None
        if campaign.starts_at and campaign.starts_at > now:
            return None
        if campaign.ends_at and campaign.ends_at <= now:
            return None
        return campaign

    async def _guarded(self, run: Callable[[], Awaitable[T]], fallback: T) -> T:
        """Run a campaign read/claim; if the campaign tables are not migrated yet, fall back.

        A pricing campaign is an optional discount. The plan list and checkout must keep working
        at catalog prices when it cannot be evaluated, e.g. code deployed before migration 168.
        The SAVEPOINT keeps the caller's transaction usable after the failed statement.
        """
        try:
            async with self._repo.savepoint():
                return await run()
        except ProgrammingError as exc:
            if not _is_missing_campaign_table(exc):
                raise
            logger.error("pricing_campaigns_unavailable_apply_migration_168", error=str(exc)[:200])
            return fallback

    async def reserve(
        self,
        *,
        plan: _PlanLike,
        interval: str,
        subject_key: str,
        workspace_id: Optional[UUID] = None,
        now: Optional[datetime] = None,
    ) -> Optional[CampaignReservation]:
        """Claim a slot and return the campaign price to check out with, or None for catalog price."""
        return await self._guarded(
            lambda: self._reserve(
                plan=plan, interval=interval, subject_key=subject_key,
                workspace_id=workspace_id, now=now,
            ),
            None,
        )

    async def _reserve(
        self,
        *,
        plan: _PlanLike,
        interval: str,
        subject_key: str,
        workspace_id: Optional[UUID] = None,
        now: Optional[datetime] = None,
    ) -> Optional[CampaignReservation]:
        """Claim a slot and return the campaign price to check out with, or None for catalog price."""
        now = now or _utcnow()
        campaign = await self._open_campaign(now)
        if not campaign:
            return None
        price = await self._repo.get_price(campaign.id, plan.id, interval)
        if not price:
            return None

        await self._repo.lock_campaign(campaign.id)
        existing = await self._repo.get_live_for_subject(campaign.id, subject_key)
        if existing is not None:
            if existing.status == "confirmed":
                # One campaign price per subject for life: a returning founder pays catalog.
                return None
            lapsed = existing.expires_at is not None and existing.expires_at <= now
            if lapsed and await self._repo.count_live(campaign.id, now) >= campaign.max_redemptions:
                # A lapsed reservation no longer holds a slot, so it must win one like anyone else.
                return None
            # Retried or re-targeted checkout: keep the same slot instead of taking another.
            existing.campaign_price_id = price.id
            existing.expires_at = now + RESERVATION_TTL
            existing.stripe_checkout_session_id = None
            await self._repo.flush()
            return self._reservation(campaign, price, existing.id)

        if await self._repo.count_live(campaign.id, now) >= campaign.max_redemptions:
            logger.info("pricing_campaign_sold_out", campaign=campaign.slug)
            return None

        row = await self._repo.create_redemption(
            campaign_id=campaign.id,
            campaign_price_id=price.id,
            subject_key=subject_key,
            workspace_id=workspace_id,
            expires_at=now + RESERVATION_TTL,
        )
        logger.info(
            "pricing_campaign_slot_reserved",
            campaign=campaign.slug,
            redemption_id=str(row.id),
            plan_slug=plan.slug,
            interval=interval,
        )
        return self._reservation(campaign, price, row.id)

    @staticmethod
    def _reservation(
        campaign: PricingCampaign, price: PricingCampaignPrice, redemption_id: UUID
    ) -> CampaignReservation:
        return CampaignReservation(
            redemption_id=redemption_id,
            campaign_slug=campaign.slug,
            stripe_price_id=price.stripe_price_id,
            price_minor=price.price_minor,
        )

    async def attach_session(self, redemption_id: UUID, session_id: str) -> None:
        row = await self._repo.get_redemption(redemption_id)
        if row is None:
            return
        row.stripe_checkout_session_id = session_id
        await self._repo.flush()

    async def confirm(
        self,
        *,
        redemption_id: Optional[UUID],
        session_id: Optional[str],
        subscription_id: Optional[str],
        workspace_id: Optional[UUID] = None,
    ) -> bool:
        """Make a paid checkout's slot permanent. Idempotent; honors the price the customer paid."""
        row = None
        if redemption_id:
            row = await self._repo.get_redemption(redemption_id)
        if row is None and session_id:
            row = await self._repo.get_redemption_by_session(session_id)
        if row is None:
            return False
        if row.status == "confirmed":
            return True
        if row.status != "reserved":
            logger.warning(
                "pricing_campaign_confirm_on_released_slot",
                redemption_id=str(row.id),
                status=row.status,
            )
        row.status = "confirmed"
        row.confirmed_at = _utcnow()
        row.expires_at = None
        if subscription_id:
            row.stripe_subscription_id = subscription_id
        if workspace_id and not row.workspace_id:
            row.workspace_id = workspace_id
        await self._repo.flush()
        logger.info("pricing_campaign_slot_confirmed", redemption_id=str(row.id))
        return True

    async def release_session(self, session_id: str) -> bool:
        """Free the slot held by an expired checkout session (matches the current session only)."""
        row = await self._repo.get_redemption_by_session(session_id)
        if row is None or row.status != "reserved":
            return False
        row.status = "released"
        await self._repo.flush()
        logger.info("pricing_campaign_slot_released", redemption_id=str(row.id))
        return True

    async def offers_for_plans(
        self,
        plans: Iterable[_PlanLike],
        *,
        subject_key: Optional[str] = None,
        now: Optional[datetime] = None,
    ) -> dict[UUID, PlanOffer]:
        """Campaign prices per plan id, empty when the campaign is closed, sold out or already used."""
        plans = list(plans)
        return await self._guarded(
            lambda: self._offers_for_plans(plans, subject_key=subject_key, now=now), {}
        )

    async def _offers_for_plans(
        self,
        plans: Iterable[_PlanLike],
        *,
        subject_key: Optional[str] = None,
        now: Optional[datetime] = None,
    ) -> dict[UUID, PlanOffer]:
        now = now or _utcnow()
        campaign = await self._open_campaign(now)
        if not campaign:
            return {}
        remaining = campaign.max_redemptions - await self._repo.count_live(campaign.id, now)
        if remaining <= 0:
            return {}
        if subject_key:
            used = await self._repo.get_live_for_subject(campaign.id, subject_key)
            if used is not None and used.status == "confirmed":
                return {}

        by_plan: dict[UUID, dict[str, int]] = {}
        for price in await self._repo.list_prices(campaign.id):
            by_plan.setdefault(price.subscription_plan_id, {})[price.interval] = price.price_minor

        out: dict[UUID, PlanOffer] = {}
        for plan in plans:
            amounts = by_plan.get(plan.id)
            if not amounts:
                continue
            out[plan.id] = PlanOffer(
                campaign_slug=campaign.slug,
                price_monthly=amounts.get("month"),
                price_yearly=amounts.get("year"),
                spots_remaining=remaining,
            )
        return out


async def provision_campaign_price(
    *,
    repo: PricingCampaignRepository,
    price_version_repo,
    stripe,
    campaign: PricingCampaign,
    plan,
    interval: str,
    amount_minor: int,
) -> tuple[PricingCampaignPrice, bool]:
    """Create (once) the dedicated Stripe Price for a campaign plan/interval.

    Returns ``(price, created)``. Re-running with the same amount is a no-op; a different
    amount is refused because live subscribers may already be on the existing Price.
    The Price is also recorded as a non-current ``plan_price_versions`` row so webhooks can
    map a founder's subscription back to its plan.
    """
    if interval not in CAMPAIGN_INTERVALS:
        raise ValidationError(f"Unsupported interval: {interval}")
    if amount_minor <= 0:
        raise ValidationError("Campaign price must be positive")
    if not plan.stripe_product_id:
        raise NotFoundError(f"Plan '{plan.slug}' has no Stripe product")

    existing = await repo.get_price(campaign.id, plan.id, interval)
    if existing:
        if existing.price_minor != amount_minor:
            raise ValidationError(
                f"{plan.slug}/{interval} already has a campaign price of {existing.price_minor}; "
                "create a new campaign instead of changing it"
            )
        return existing, False

    created = await stripe.create_price(
        product_id=plan.stripe_product_id,
        unit_amount=amount_minor,
        currency=plan.currency,
        interval=interval,
        nickname=f"{plan.name} - {campaign.name} ({interval})",
    )
    row = await repo.create_price(
        campaign_id=campaign.id,
        subscription_plan_id=plan.id,
        interval=interval,
        stripe_price_id=created["id"],
        price_minor=amount_minor,
        currency=plan.currency,
    )
    await price_version_repo.create(
        subscription_plan_id=plan.id,
        interval=interval,
        stripe_price_id=created["id"],
        price_minor=amount_minor,
        currency=plan.currency,
        stripe_product_id=plan.stripe_product_id,
        is_current=False,
    )
    return row, True