"""Billing domain - pricing campaign persistence (limited-slot offers)."""
from __future__ import annotations
from datetime import datetime
from typing import Optional
from uuid import UUID

from sqlalchemy import func as sa_func, or_, select, text
from sqlalchemy.ext.asyncio import AsyncSession

from src.domain.billing.models import (
    PricingCampaign,
    PricingCampaignPrice,
    PricingCampaignRedemption,
)


class PricingCampaignRepository:
    def __init__(self, db: AsyncSession):
        self._db = db

    def savepoint(self):
        """Async context manager: a SAVEPOINT so a failed campaign query cannot poison the request."""
        return self._db.begin_nested()

    async def get_by_slug(self, slug: str) -> Optional[PricingCampaign]:
        result = await self._db.execute(select(PricingCampaign).where(PricingCampaign.slug == slug))
        return result.scalar_one_or_none()

    async def create_campaign(self, *, slug: str, name: str, max_redemptions: int) -> PricingCampaign:
        campaign = PricingCampaign(slug=slug, name=name, max_redemptions=max_redemptions)
        self._db.add(campaign)
        await self._db.flush()
        await self._db.refresh(campaign)
        return campaign

    async def lock_campaign(self, campaign_id: UUID) -> None:
        """Serialize slot claims for one campaign until the transaction ends."""
        await self._db.execute(
            text("SELECT pg_advisory_xact_lock(hashtext(:key)::bigint)"),
            {"key": f"pricing_campaign:{campaign_id}"},
        )

    async def get_price(
        self, campaign_id: UUID, subscription_plan_id: UUID, interval: str
    ) -> Optional[PricingCampaignPrice]:
        result = await self._db.execute(
            select(PricingCampaignPrice).where(
                PricingCampaignPrice.campaign_id == campaign_id,
                PricingCampaignPrice.subscription_plan_id == subscription_plan_id,
                PricingCampaignPrice.interval == interval,
            )
        )
        return result.scalar_one_or_none()

    async def list_prices(self, campaign_id: UUID) -> list[PricingCampaignPrice]:
        result = await self._db.execute(
            select(PricingCampaignPrice).where(PricingCampaignPrice.campaign_id == campaign_id)
        )
        return list(result.scalars().all())

    async def create_price(
        self,
        *,
        campaign_id: UUID,
        subscription_plan_id: UUID,
        interval: str,
        stripe_price_id: str,
        price_minor: int,
        currency: str,
    ) -> PricingCampaignPrice:
        row = PricingCampaignPrice(
            campaign_id=campaign_id,
            subscription_plan_id=subscription_plan_id,
            interval=interval,
            stripe_price_id=stripe_price_id,
            price_minor=price_minor,
            currency=currency,
        )
        self._db.add(row)
        await self._db.flush()
        await self._db.refresh(row)
        return row

    async def count_live(self, campaign_id: UUID, now: datetime) -> int:
        """Confirmed redemptions plus reservations that have not expired."""
        result = await self._db.execute(
            select(sa_func.count())
            .select_from(PricingCampaignRedemption)
            .where(
                PricingCampaignRedemption.campaign_id == campaign_id,
                or_(
                    PricingCampaignRedemption.status == "confirmed",
                    (PricingCampaignRedemption.status == "reserved")
                    & (
                        PricingCampaignRedemption.expires_at.is_(None)
                        | (PricingCampaignRedemption.expires_at > now)
                    ),
                ),
            )
        )
        return int(result.scalar_one())

    async def get_live_for_subject(
        self, campaign_id: UUID, subject_key: str
    ) -> Optional[PricingCampaignRedemption]:
        """The reserved-or-confirmed row for a subject (the partial unique index allows one)."""
        result = await self._db.execute(
            select(PricingCampaignRedemption).where(
                PricingCampaignRedemption.campaign_id == campaign_id,
                PricingCampaignRedemption.subject_key == subject_key,
                PricingCampaignRedemption.status.in_(("reserved", "confirmed")),
            )
        )
        return result.scalar_one_or_none()

    async def create_redemption(
        self,
        *,
        campaign_id: UUID,
        campaign_price_id: UUID,
        subject_key: str,
        workspace_id: Optional[UUID],
        expires_at: datetime,
    ) -> PricingCampaignRedemption:
        row = PricingCampaignRedemption(
            campaign_id=campaign_id,
            campaign_price_id=campaign_price_id,
            subject_key=subject_key,
            workspace_id=workspace_id,
            status="reserved",
            expires_at=expires_at,
        )
        self._db.add(row)
        await self._db.flush()
        await self._db.refresh(row)
        return row

    async def get_redemption(self, redemption_id: UUID) -> Optional[PricingCampaignRedemption]:
        result = await self._db.execute(
            select(PricingCampaignRedemption).where(PricingCampaignRedemption.id == redemption_id)
        )
        return result.scalar_one_or_none()

    async def get_redemption_by_session(self, session_id: str) -> Optional[PricingCampaignRedemption]:
        result = await self._db.execute(
            select(PricingCampaignRedemption).where(
                PricingCampaignRedemption.stripe_checkout_session_id == session_id
            )
        )
        return result.scalar_one_or_none()

    async def get_price_by_id(self, price_id: UUID) -> Optional[PricingCampaignPrice]:
        result = await self._db.execute(
            select(PricingCampaignPrice).where(PricingCampaignPrice.id == price_id)
        )
        return result.scalar_one_or_none()

    async def flush(self) -> None:
        await self._db.flush()