app/services/billing_service.py

"""Author RAG — Stripe Billing Service.

Full Stripe integration — checkout, portal, webhooks, referral coupons.
"""

from __future__ import annotations

import secrets
import structlog
from datetime import datetime, timedelta, timezone

try:
    import stripe
except ImportError:  # pragma: no cover
    stripe = None  # type: ignore[assignment]

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

from app.config import get_settings
from app.repositories.access_repo import AccessRepository
from app.repositories.user_repo import UserRepository
from app.services.auth_service import _hash_password

logger = structlog.get_logger(__name__)


def _refresh_stripe_key() -> None:
    """Load the current Stripe secret key (supports SuperAdmin hot-reload)."""
    if stripe is None:
        raise RuntimeError("Stripe SDK not installed")
    key = get_settings().STRIPE_SECRET_KEY
    if not key:
        raise ValueError("Stripe is not configured on this server")
    stripe.api_key = key


def _stripe_resource_id(value: object | None) -> str:
    """Normalize Stripe customer/subscription fields (id string or expanded object)."""
    if value is None:
        raise ValueError("Stripe checkout session is missing billing details. Please contact support.")
    if isinstance(value, str):
        return value
    resource_id = getattr(value, "id", None)
    if not resource_id:
        raise ValueError("Stripe checkout session is incomplete. Please contact support.")
    return str(resource_id)


def _customer_email_from_session(session: object) -> str:
    """Extract payer email from a Stripe Checkout session."""
    details = getattr(session, "customer_details", None)
    email = getattr(details, "email", None) if details else None
    if not email:
        raise ValueError(
            "No email on this payment session. Use the same email in Stripe Checkout or contact support."
        )
    return str(email).lower().strip()


def _stripe_metadata_value(metadata: object | None, key: str, default: str = "starter") -> str:
    """Read metadata from Stripe StripeObject or plain dict."""
    if not metadata:
        return default
    if isinstance(metadata, dict):
        value = metadata.get(key)
        return str(value) if value else default
    try:
        value = metadata[key]  # StripeObject supports __getitem__
        return str(value) if value else default
    except (KeyError, TypeError):
        value = getattr(metadata, key, None)
        return str(value) if value else default


# Configure Stripe SDK when available. Refreshed again before each API call.
if stripe and get_settings().STRIPE_SECRET_KEY:
    stripe.api_key = get_settings().STRIPE_SECRET_KEY


# ── Webhook Signature Verification ───────────────────────────────────────────

def verify_webhook_signature(payload: bytes, sig_header: str):
    """Verify Stripe webhook signature and return the parsed event."""
    if stripe is None:
        raise RuntimeError("Stripe SDK not installed")
    try:
        event = stripe.Webhook.construct_event(
            payload=payload,
            sig_header=sig_header,
            secret=get_settings().STRIPE_WEBHOOK_SECRET,
        )
        return event
    except stripe.SignatureVerificationError as e:
        logger.warning("Invalid Stripe webhook signature", error=str(e))
        raise
    except ValueError as e:
        logger.warning("Malformed Stripe webhook payload", error=str(e))
        raise


# ── Checkout ──────────────────────────────────────────────────────────────────

async def create_checkout_session(
    plan_id: str,
    stripe_price_id: str,
    success_url: str,
    cancel_url: str,
    customer_email: str | None = None,
    existing_customer_id: str | None = None,
    coupon_code: str | None = None,
    author_id: str | None = None,
) -> str:
    """Create a Stripe Checkout session and return the redirect URL.

    Card always required upfront — no free access without payment.

    Args:
        plan_id: Our internal plan ID (e.g. "starter").
        stripe_price_id: The Stripe Price ID for the selected plan/billing.
        success_url: Redirect after payment (must include {CHECKOUT_SESSION_ID}).
        cancel_url: Redirect if user cancels.
        customer_email: Pre-fill email in Stripe Checkout.
        existing_customer_id: For upgrades — attach to existing Stripe customer.
        coupon_code: Referral/affiliate coupon to apply automatically.
        author_id: Logged-in author id (metadata + client_reference_id).

    Returns:
        Stripe Checkout URL — redirect the browser here.
    """
    metadata: dict[str, str] = {"plan_id": plan_id}
    if author_id:
        metadata["author_id"] = author_id

    params: dict = {
        "mode": "subscription",
        "line_items": [{"price": stripe_price_id, "quantity": 1}],
        "success_url": success_url,
        "cancel_url": cancel_url,
        "payment_method_collection": "always",   # Card required upfront
        "metadata": metadata,
        "subscription_data": {"metadata": metadata},
        "allow_promotion_codes": True,           # Enable referral coupons
    }

    if author_id:
        params["client_reference_id"] = author_id

    if coupon_code:
        params["discounts"] = [{"coupon": coupon_code}]
        params.pop("allow_promotion_codes", None)

    if existing_customer_id:
        params["customer"] = existing_customer_id
    elif customer_email:
        params["customer_email"] = customer_email

    try:
        _refresh_stripe_key()
        session = stripe.checkout.Session.create(**params)
        logger.info("Stripe checkout session created", session_id=session.id, plan_id=plan_id)
        return session.url
    except stripe.StripeError as e:
        logger.error("Stripe checkout failed", error=str(e), plan_id=plan_id)
        raise


async def create_addon_checkout_session(
    *,
    addon_id: str,
    price_usd: float,
    display_name: str,
    author_id: str,
    success_url: str,
    cancel_url: str,
    customer_email: str | None = None,
    existing_customer_id: str | None = None,
    description: str | None = None,
    extra_metadata: dict[str, str] | None = None,
) -> str:
    """Create a one-time Stripe Checkout session for a platform add-on (R-202).

    Uses price_data so SuperAdmin can change USD without syncing Stripe Price IDs.
    """
    unit_amount = int(round(float(price_usd) * 100))
    if unit_amount < 50:
        raise ValueError("Add-on price must be at least $0.50 for Stripe Checkout")

    metadata: dict[str, str] = {
        "addon": addon_id,
        "author_id": author_id,
    }
    if extra_metadata:
        metadata.update({k: str(v) for k, v in extra_metadata.items() if v is not None})

    params: dict = {
        "mode": "payment",
        "line_items": [
            {
                "quantity": 1,
                "price_data": {
                    "currency": "usd",
                    "unit_amount": unit_amount,
                    "product_data": {
                        "name": display_name or "Website URL unlock",
                        "description": description
                        or "One-time unlock credit for a locked embed website URL",
                    },
                },
            }
        ],
        "success_url": success_url,
        "cancel_url": cancel_url,
        "metadata": metadata,
        "payment_intent_data": {
            "metadata": metadata,
        },
    }

    if existing_customer_id:
        params["customer"] = existing_customer_id
    elif customer_email:
        params["customer_email"] = customer_email

    try:
        _refresh_stripe_key()
        session = stripe.checkout.Session.create(**params)
        logger.info(
            "Stripe addon checkout created",
            session_id=session.id,
            addon=addon_id,
            author_id=author_id,
        )
        return session.url
    except stripe.StripeError as e:
        logger.error("Stripe addon checkout failed", error=str(e), addon=addon_id)
        raise


async def get_checkout_session(session_id: str) -> stripe.checkout.Session:
    """Retrieve and verify a completed Stripe Checkout session.

    Called by POST /api/auth/activate after the browser returns from Stripe.
    """
    try:
        _refresh_stripe_key()
        session = stripe.checkout.Session.retrieve(
            session_id,
            expand=["customer", "subscription", "subscription.items.data.price"],
        )
        logger.info(
            "Stripe session retrieved",
            session_id=session_id,
            payment_status=session.payment_status,
        )
        return session
    except stripe.StripeError as e:
        logger.error("Failed to retrieve Stripe session", session_id=session_id, error=str(e))
        raise


# ── Customer Portal ───────────────────────────────────────────────────────────

async def create_portal_session(stripe_customer_id: str, return_url: str) -> str:
    """Create a Stripe Customer Portal session for self-service billing.

    Authors can: update card, view invoices, cancel, upgrade/downgrade.
    """
    try:
        session = stripe.billing_portal.Session.create(
            customer=stripe_customer_id,
            return_url=return_url,
        )
        logger.info("Stripe portal session created", customer_id=stripe_customer_id)
        return session.url
    except stripe.StripeError as e:
        logger.error("Failed to create portal session", customer_id=stripe_customer_id, error=str(e))
        raise


# ── Subscription Management ───────────────────────────────────────────────────

async def get_subscription(subscription_id: str) -> stripe.Subscription:
    """Retrieve full subscription details from Stripe."""
    return stripe.Subscription.retrieve(
        subscription_id,
        expand=["items.data.price", "customer"],
    )


async def cancel_subscription(subscription_id: str, at_period_end: bool = True) -> None:
    """Cancel a Stripe subscription.

    Args:
        at_period_end: True = access continues to end of paid period (recommended).
                       False = immediate cancellation with proration.
    """
    try:
        stripe.Subscription.modify(subscription_id, cancel_at_period_end=at_period_end)
        logger.info("Subscription cancel scheduled", sub_id=subscription_id, at_end=at_period_end)
    except stripe.StripeError as e:
        logger.error("Failed to cancel subscription", sub_id=subscription_id, error=str(e))
        raise


async def upgrade_subscription(
    subscription_id: str,
    new_stripe_price_id: str,
    new_plan_id: str | None = None,
) -> None:
    """Upgrade or downgrade a subscription. Stripe handles proration automatically."""
    try:
        _refresh_stripe_key()
        sub = stripe.Subscription.retrieve(subscription_id)
        modify_kwargs: dict = {
            "items": [{"id": sub["items"]["data"][0]["id"], "price": new_stripe_price_id}],
            "proration_behavior": "create_prorations",
        }
        if new_plan_id:
            modify_kwargs["metadata"] = {"plan_id": new_plan_id}
        stripe.Subscription.modify(subscription_id, **modify_kwargs)
        logger.info(
            "Subscription upgraded",
            sub_id=subscription_id,
            new_price=new_stripe_price_id,
            plan_id=new_plan_id,
        )
    except stripe.StripeError as e:
        logger.error("Failed to upgrade subscription", sub_id=subscription_id, error=str(e))
        raise


# ── Account Activation (Post-Payment) ────────────────────────────────────────

class BillingService:
    """Orchestrates account creation and access provisioning after Stripe payment."""

    def __init__(self, db: AsyncSession) -> None:
        self._db = db
        self._access = AccessRepository(db)
        self._users = UserRepository(db)

    @staticmethod
    def _generate_temp_password() -> str:
        """Generate a secure temporary password for webhook auto-provision."""
        alphabet = "abcdefghijkmnopqrstuvwxyzABCDEFGHJKLMNPQRSTUVWXYZ23456789"
        return "".join(secrets.choice(alphabet) for _ in range(14))

    async def _resolve_plan(self, plan_id: str) -> tuple[str, int]:
        """Return plan_id and token budget from PricingPlan or defaults."""
        from app.models.pricing_plan import PricingPlan
        from sqlalchemy import select

        plan_result = await self._db.execute(
            select(PricingPlan).where(PricingPlan.id == plan_id)
        )
        plan = plan_result.scalar_one_or_none()
        if not plan:
            logger.warning("Unknown plan_id in Stripe session", plan_id=plan_id)
            return "starter", 500_000
        return plan_id, plan.max_token_budget

    async def _find_user_for_stripe_activation(
        self,
        email: str,
        stripe_session_id: str,
        stripe_customer_id: str,
        stripe_sub_id: str | None = None,
    ):
        """Locate an author by email or Stripe checkout identifiers."""
        from sqlalchemy import select

        from app.models.client_access import ClientAccess
        from app.models.user import User

        user = await self._users.get_by_email(email)
        if user:
            return user

        for column, value in (
            (User.stripe_checkout_session_id, stripe_session_id),
            (User.stripe_customer_id, stripe_customer_id),
            (User.stripe_subscription_id, stripe_sub_id),
        ):
            if not value:
                continue
            result = await self._db.execute(select(User).where(column == value))
            found = result.scalar_one_or_none()
            if found:
                return found

        if stripe_sub_id:
            access_result = await self._db.execute(
                select(ClientAccess).where(ClientAccess.stripe_subscription_id == stripe_sub_id)
            )
            access = access_result.scalar_one_or_none()
            if access:
                return await self._users.get_by_id(access.author_id)
        return None

    async def _create_author_for_paid_session(
        self,
        *,
        email: str,
        password_hash: str,
        full_name: str,
        plan_id: str,
        token_budget: int,
        stripe_customer_id: str,
        stripe_sub_id: str,
        stripe_session_id: str,
    ):
        """Create author + access; commit user first so payment is never orphaned."""
        user_data: dict = {
            "email": email,
            "password_hash": password_hash,
            "full_name": full_name or None,
            "role": "author",
            "is_active": True,
            "stripe_customer_id": stripe_customer_id,
            "stripe_subscription_id": stripe_sub_id,
            "stripe_plan_id": plan_id,
            "stripe_checkout_session_id": stripe_session_id,
            "activation_status": "active",
            "provisioned_via": "stripe_activate",
            "password_change_required": False,
        }

        try:
            user = await self._users.create(user_data)
            await self._db.commit()
        except IntegrityError:
            await self._db.rollback()
            existing = await self._find_user_for_stripe_activation(
                email, stripe_session_id, stripe_customer_id, stripe_sub_id
            )
            if existing:
                return existing
            minimal = {k: v for k, v in user_data.items() if k not in (
                "activation_status", "provisioned_via", "password_change_required",
                "stripe_checkout_session_id",
            )}
            user = await self._users.create(minimal)
            await self._db.commit()

        ok = await self._ensure_stripe_client_access(
            author_id=user.id,
            plan_id=plan_id,
            token_budget=token_budget,
            stripe_sub_id=stripe_sub_id,
            stripe_customer_id=stripe_customer_id,
        )
        if not ok:
            logger.error("access_grant_failed_after_user_create", user_id=user.id)
        return user

    async def _resolve_access_record(self, author_id: str, stripe_sub_id: str):
        """Find the ClientAccess row to update — active, revoked, or by subscription id."""
        from sqlalchemy import select

        from app.models.client_access import ClientAccess

        if stripe_sub_id:
            result = await self._db.execute(
                select(ClientAccess).where(ClientAccess.stripe_subscription_id == stripe_sub_id)
            )
            by_sub = result.scalar_one_or_none()
            if by_sub:
                if by_sub.author_id != author_id:
                    user = await self._users.get_by_id(author_id)
                    if user and user.stripe_subscription_id == stripe_sub_id:
                        logger.warning(
                            "stripe_access_reclaimed",
                            sub_id=stripe_sub_id,
                            from_author=by_sub.author_id,
                            to_author=author_id,
                        )
                        by_sub.author_id = author_id
                if by_sub.author_id == author_id:
                    return by_sub

        access = await self._access.get_active_for_author(author_id)
        if access:
            return access

        result = await self._db.execute(
            select(ClientAccess)
            .where(ClientAccess.author_id == author_id)
            .order_by(ClientAccess.created_at.desc())
            .limit(1)
        )
        return result.scalar_one_or_none()

    async def _reclaim_access_for_subscriber(
        self,
        *,
        author_id: str,
        stripe_sub_id: str,
        plan_id: str,
        token_budget: int,
    ):
        """Attach the subscription access row to the paying author."""
        from sqlalchemy import select

        from app.models.client_access import ClientAccess

        if not stripe_sub_id:
            return None

        result = await self._db.execute(
            select(ClientAccess).where(ClientAccess.stripe_subscription_id == stripe_sub_id)
        )
        access = result.scalar_one_or_none()
        if not access:
            return None

        if access.author_id != author_id:
            logger.warning(
                "stripe_access_reclaimed",
                sub_id=stripe_sub_id,
                from_author=access.author_id,
                to_author=author_id,
            )
            access.author_id = author_id

        now = datetime.now(timezone.utc)
        access.plan = plan_id
        access.plan_id = plan_id
        access.token_budget = token_budget
        access.is_revoked = False
        access.auto_renew = True
        exp = access.expires_at
        compare_now = now.replace(tzinfo=None) if (exp and exp.tzinfo is None) else now
        if exp is None or exp < compare_now:
            access.expires_at = now + timedelta(days=365 * 10)
        return access

    async def _sync_user_stripe_fields(
        self,
        author_id: str,
        *,
        plan_id: str,
        stripe_sub_id: str,
        stripe_customer_id: str | None = None,
    ) -> None:
        """Keep User Stripe columns aligned with the paid subscription."""
        from sqlalchemy import update

        from app.models.user import User

        values: dict = {
            "stripe_plan_id": plan_id,
            "stripe_subscription_id": stripe_sub_id,
            "chatbot_is_active": True,
            "activation_status": "active",
        }
        if stripe_customer_id:
            values["stripe_customer_id"] = stripe_customer_id
        await self._db.execute(
            update(User).where(User.id == author_id).values(**values)
        )

    async def _purge_conflicting_subscription_rows(
        self, author_id: str, stripe_sub_id: str,
    ) -> None:
        """Remove orphan grants on other authors that block this subscription id."""
        from sqlalchemy import delete

        from app.models.client_access import ClientAccess

        if not stripe_sub_id:
            return
        await self._db.execute(
            delete(ClientAccess).where(
                ClientAccess.stripe_subscription_id == stripe_sub_id,
                ClientAccess.author_id != author_id,
            )
        )
        await self._db.flush()

    async def _sync_from_stored_checkout_session(self, user_id: str) -> bool:
        """Reconcile grant from stripe_checkout_session_id when User Stripe fields are incomplete."""
        user = await self._users.get_by_id(user_id)
        if not user or not user.stripe_checkout_session_id:
            return False
        try:
            session = await get_checkout_session(user.stripe_checkout_session_id)
        except Exception as exc:
            logger.warning("checkout_session_sync_failed", user_id=user_id, error=str(exc)[:120])
            return False
        if session.payment_status != "paid":
            return False

        stripe_customer_id = _stripe_resource_id(session.customer)
        stripe_sub_id = _stripe_resource_id(session.subscription)
        plan_id, token_budget = await self._resolve_plan(
            _stripe_metadata_value(session.metadata, "plan_id", user.stripe_plan_id or "starter")
        )
        return await self._ensure_stripe_client_access(
            author_id=user_id,
            plan_id=plan_id,
            token_budget=token_budget,
            stripe_sub_id=stripe_sub_id,
            stripe_customer_id=stripe_customer_id,
        )

    async def _upsert_author_access_row(
        self,
        *,
        author_id: str,
        plan_id: str,
        token_budget: int,
        stripe_sub_id: str,
    ):
        """Update an existing author grant or create a fresh one."""
        from sqlalchemy import select, update

        from app.core.access.token_crypto import create_subscription_token, hash_subscription_token
        from app.models.base import generate_uuid
        from app.models.client_access import ClientAccess

        now = datetime.now(timezone.utc)
        expires = now + timedelta(days=365 * 10)

        await self._purge_conflicting_subscription_rows(author_id, stripe_sub_id)

        reclaimed = await self._reclaim_access_for_subscriber(
            author_id=author_id,
            stripe_sub_id=stripe_sub_id,
            plan_id=plan_id,
            token_budget=token_budget,
        )
        if reclaimed:
            return reclaimed

        access = await self._resolve_access_record(author_id, stripe_sub_id)
        if access:
            await self._apply_access_grant_updates(
                access,
                author_id=author_id,
                plan_id=plan_id,
                token_budget=token_budget,
                stripe_sub_id=stripe_sub_id,
            )
            return access

        result = await self._db.execute(
            select(ClientAccess)
            .where(ClientAccess.author_id == author_id)
            .order_by(ClientAccess.created_at.desc())
            .limit(1)
        )
        existing = result.scalar_one_or_none()
        if existing:
            await self._apply_access_grant_updates(
                existing,
                author_id=author_id,
                plan_id=plan_id,
                token_budget=token_budget,
                stripe_sub_id=stripe_sub_id,
            )
            return existing

        grant_id = generate_uuid()
        token = create_subscription_token(
            author_id=author_id,
            grant_id=grant_id,
            granted_at=now,
            expires_at=expires,
        )
        return await self._access.create({
            "id": grant_id,
            "author_id": author_id,
            "granted_by": author_id,  # self-provisioned via Stripe (DB may have NOT NULL)
            "plan": plan_id,
            "plan_id": plan_id,
            "stripe_subscription_id": stripe_sub_id,
            "granted_at": now,
            "expires_at": expires,
            "token_hash": hash_subscription_token(token),
            "token_budget": token_budget,
            "auto_renew": True,
        })

    async def _ensure_stripe_client_access(
        self,
        *,
        author_id: str,
        plan_id: str,
        token_budget: int,
        stripe_sub_id: str,
        stripe_customer_id: str | None = None,
    ) -> bool:
        """Idempotent: create or reactivate ClientAccess for a paid Stripe author."""
        await self._sync_user_stripe_fields(
            author_id,
            plan_id=plan_id,
            stripe_sub_id=stripe_sub_id,
            stripe_customer_id=stripe_customer_id,
        )
        try:
            await self._upsert_author_access_row(
                author_id=author_id,
                plan_id=plan_id,
                token_budget=token_budget,
                stripe_sub_id=stripe_sub_id,
            )
            await self._db.commit()
        except IntegrityError as exc:
            await self._db.rollback()
            logger.warning(
                "stripe_access_upsert_retry",
                author_id=author_id,
                sub_id=stripe_sub_id,
                error=str(exc)[:120],
            )
            try:
                await self._upsert_author_access_row(
                    author_id=author_id,
                    plan_id=plan_id,
                    token_budget=token_budget,
                    stripe_sub_id=stripe_sub_id,
                )
                await self._db.commit()
            except IntegrityError:
                await self._db.rollback()
                logger.error(
                    "stripe_access_create_failed",
                    author_id=author_id,
                    sub_id=stripe_sub_id,
                )
                return False

        active = await self._access.get_active_for_author(author_id)
        if active:
            return True

        logger.error(
            "stripe_access_inactive_after_upsert",
            author_id=author_id,
            sub_id=stripe_sub_id,
        )
        return False

    async def _sync_subscription_from_stripe(self, user_id: str) -> bool:
        """Fetch live Stripe subscription for a customer and ensure local grant."""
        user = await self._users.get_by_id(user_id)
        if not user:
            return False

        stripe_sub_id = user.stripe_subscription_id
        plan_id = user.stripe_plan_id or "starter"
        stripe_customer_id = user.stripe_customer_id

        if user.stripe_customer_id and not stripe_sub_id:
            try:
                _refresh_stripe_key()
                subs = stripe.Subscription.list(
                    customer=user.stripe_customer_id,
                    status="active",
                    limit=1,
                )
                if subs.data:
                    stripe_sub_id = subs.data[0].id
                    meta = subs.data[0].metadata or {}
                    plan_id = _stripe_metadata_value(meta, "plan_id", plan_id)
            except Exception as exc:
                logger.warning("stripe_subscription_sync_failed", user_id=user_id, error=str(exc)[:120])

        if not stripe_sub_id:
            return False

        _, token_budget = await self._resolve_plan(plan_id)
        return await self._ensure_stripe_client_access(
            author_id=user_id,
            plan_id=plan_id,
            token_budget=token_budget,
            stripe_sub_id=stripe_sub_id,
            stripe_customer_id=stripe_customer_id,
        )

    async def require_active_subscription(self, author_id: str) -> bool:
        """Repair and verify an author has an active ClientAccess grant."""
        if await self._access.get_active_for_author(author_id):
            return True
        if await self.repair_missing_access_for_user(author_id):
            return True
        return await self._sync_subscription_from_stripe(author_id)

    async def repair_missing_access_for_user(self, user_id: str) -> bool:
        """Backfill ClientAccess when User has Stripe fields but no active grant."""
        user = await self._users.get_by_id(user_id)
        if not user:
            return False

        if not user.stripe_subscription_id:
            if await self._sync_from_stored_checkout_session(user_id):
                return True
            if await self._sync_subscription_from_stripe(user_id):
                return True
            return False

        plan_id = user.stripe_plan_id or "starter"
        stripe_sub_id = user.stripe_subscription_id
        _, token_budget = await self._resolve_plan(plan_id)
        ok = await self._ensure_stripe_client_access(
            author_id=user_id,
            plan_id=plan_id,
            token_budget=token_budget,
            stripe_sub_id=stripe_sub_id,
            stripe_customer_id=user.stripe_customer_id,
        )
        if ok:
            return True
        return await self._access.get_active_for_author(user_id) is not None

    async def _apply_access_grant_updates(
        self,
        access,
        *,
        author_id: str,
        plan_id: str,
        token_budget: int,
        stripe_sub_id: str,
    ) -> None:
        """Update an existing ClientAccess row for a Stripe subscriber."""
        now = datetime.now(timezone.utc)
        access.plan = plan_id
        access.plan_id = plan_id
        access.token_budget = token_budget
        access.is_revoked = False
        access.auto_renew = True
        if stripe_sub_id and not access.stripe_subscription_id:
            access.stripe_subscription_id = stripe_sub_id
        exp = access.expires_at
        compare_now = now.replace(tzinfo=None) if (exp and exp.tzinfo is None) else now
        if exp is None or exp < compare_now:
            access.expires_at = now + timedelta(days=365 * 10)
        if access.author_id != author_id:
            access.author_id = author_id

    async def _force_password_login(
        self,
        *,
        customer_email: str,
        password_hash: str,
        full_name: str | None,
        stripe_session_id: str,
        stripe_customer_id: str,
        stripe_sub_id: str,
        plan_id: str,
        token_budget: int = 500_000,
        full_name_for_create: str = "",
    ) -> dict:
        """Guaranteed post-payment path: set password and return login payload."""
        user = await self._find_user_for_stripe_activation(
            customer_email, stripe_session_id, stripe_customer_id, stripe_sub_id
        )
        if not user:
            user = await self._users.get_by_email(customer_email)

        if not user:
            user = await self._create_author_for_paid_session(
                email=customer_email,
                password_hash=password_hash,
                full_name=full_name_for_create or full_name or customer_email,
                plan_id=plan_id,
                token_budget=token_budget,
                stripe_customer_id=stripe_customer_id,
                stripe_sub_id=stripe_sub_id,
                stripe_session_id=stripe_session_id,
            )
            return {
                "user_id": user.id,
                "email": user.email,
                "plan_id": plan_id,
                "already_activated": False,
            }

        user.password_hash = password_hash
        user.password_change_required = False
        user.activation_status = "active"
        user.is_active = True
        user.provisioned_via = "stripe_activate"
        user.stripe_customer_id = stripe_customer_id
        user.stripe_subscription_id = stripe_sub_id
        user.stripe_plan_id = plan_id
        if full_name:
            user.full_name = full_name

        if not user.stripe_checkout_session_id:
            user.stripe_checkout_session_id = stripe_session_id

        try:
            await self._db.commit()
        except IntegrityError:
            await self._db.rollback()
            user = await self._users.get_by_email(customer_email)
            if not user:
                raise
            user.password_hash = password_hash
            user.password_change_required = False
            user.activation_status = "active"
            user.is_active = True
            if full_name:
                user.full_name = full_name
            await self._db.commit()

        user_id = user.id
        user_email = user.email
        ok = await self._ensure_stripe_client_access(
            author_id=user_id,
            plan_id=plan_id,
            token_budget=token_budget,
            stripe_sub_id=stripe_sub_id,
            stripe_customer_id=stripe_customer_id,
        )
        if not ok:
            logger.error("access_grant_failed_force_login", user_id=user_id)

        logger.info("Author password set after Stripe payment", user_id=user_id, email=user_email)
        return {
            "user_id": user_id,
            "email": user_email,
            "plan_id": plan_id,
            "already_activated": True,
        }

    async def _complete_existing_user_activation(
        self,
        user,
        *,
        stripe_session_id: str,
        password_hash: str,
        full_name: str | None,
        plan_id: str,
        token_budget: int,
        stripe_customer_id: str,
        stripe_sub_id: str,
    ) -> dict:
        """Attach Stripe payment to an existing author and apply their chosen password."""
        now = datetime.now(timezone.utc)
        was_same_session = user.stripe_checkout_session_id == stripe_session_id

        user.password_hash = password_hash
        user.password_change_required = False
        user.provisioned_via = "stripe_activate"
        user.activation_status = "active"
        user.stripe_checkout_session_id = stripe_session_id
        user.stripe_customer_id = stripe_customer_id
        user.stripe_subscription_id = stripe_sub_id
        user.stripe_plan_id = plan_id
        if full_name:
            user.full_name = full_name
        user.credentials_email_sent_at = now

        try:
            ok = await self._ensure_stripe_client_access(
                author_id=user.id,
                plan_id=plan_id,
                token_budget=token_budget,
                stripe_sub_id=stripe_sub_id,
                stripe_customer_id=stripe_customer_id,
            )
            if not ok:
                raise RuntimeError("subscription_grant_failed")
        except (IntegrityError, RuntimeError):
            await self._db.rollback()
            return await self._force_password_login(
                customer_email=user.email,
                password_hash=password_hash,
                full_name=full_name,
                stripe_session_id=stripe_session_id,
                stripe_customer_id=stripe_customer_id,
                stripe_sub_id=stripe_sub_id,
                plan_id=plan_id,
                token_budget=token_budget,
                full_name_for_create=user.full_name or user.email,
            )

        self._queue_credentials_email(
            email=user.email,
            name=user.full_name or user.email,
            temp_password=None,
            user_chose_password=True,
        )
        logger.info(
            "Existing author linked to Stripe checkout",
            user_id=user.id,
            email=user.email,
            plan_id=plan_id,
            same_session=was_same_session,
        )
        return {
            "user_id": user.id,
            "email": user.email,
            "plan_id": plan_id,
            "already_activated": was_same_session,
        }

    async def _provision_stripe_author(
        self,
        *,
        email: str,
        password_hash: str,
        full_name: str,
        plan_id: str,
        token_budget: int,
        stripe_customer_id: str,
        stripe_sub_id: str,
        stripe_session_id: str,
        provisioned_via: str,
        activation_status: str,
        password_change_required: bool,
    ) -> dict:
        """Create User + ClientAccess for a paid Stripe checkout."""
        from app.models.base import generate_uuid
        from app.core.access.token_crypto import create_subscription_token, hash_subscription_token

        now = datetime.now(timezone.utc)
        user = await self._users.create({
            "email": email,
            "password_hash": password_hash,
            "full_name": full_name or None,
            "role": "author",
            "is_active": True,
            "stripe_customer_id": stripe_customer_id,
            "stripe_subscription_id": stripe_sub_id,
            "stripe_plan_id": plan_id,
            "stripe_checkout_session_id": stripe_session_id,
            "activation_status": activation_status,
            "provisioned_via": provisioned_via,
            "password_change_required": password_change_required,
        })

        grant_id = generate_uuid()
        token = create_subscription_token(
            author_id=user.id,
            grant_id=grant_id,
            granted_at=now,
            expires_at=now + timedelta(days=365 * 10),
        )
        await self._access.create({
            "id": grant_id,
            "author_id": user.id,
            "granted_by": user.id,  # self-provisioned via Stripe (DB may have NOT NULL)
            "plan": plan_id,
            "plan_id": plan_id,
            "stripe_subscription_id": stripe_sub_id,
            "granted_at": now,
            "expires_at": now + timedelta(days=365 * 10),
            "token_hash": hash_subscription_token(token),
            "token_budget": token_budget,
            "auto_renew": True,
        })
        await self._db.commit()
        return {
            "user_id": user.id,
            "email": email,
            "plan_id": plan_id,
            "subscription_token": token,
            "already_activated": False,
            "provisioned_via": provisioned_via,
        }

    def _queue_credentials_email(
        self,
        *,
        email: str,
        name: str,
        temp_password: str | None,
        user_chose_password: bool,
    ) -> None:
        """Queue credentials email via Celery (non-blocking)."""
        from app.services.author_onboarding_service import login_url
        from app.tasks.email_task import send_account_credentials as send_task

        try:
            send_task.delay({
                "to": email,
                "name": name or email,
                "login_url": login_url(),
                "temp_password": temp_password,
                "user_chose_password": user_chose_password,
            })
        except Exception as exc:
            logger.warning("credentials_email_queue_failed", email=email, error=str(exc)[:200])

    async def _verify_post_activation(
        self,
        user_id: str,
        *,
        plan_id: str,
        token_budget: int,
        stripe_sub_id: str,
        stripe_customer_id: str | None = None,
    ) -> None:
        """Fail closed when a paid author has no active ClientAccess grant."""
        if not await self.require_active_subscription(user_id):
            await self._ensure_stripe_client_access(
                author_id=user_id,
                plan_id=plan_id,
                token_budget=token_budget,
                stripe_sub_id=stripe_sub_id,
                stripe_customer_id=stripe_customer_id,
            )
        if not await self._access.get_active_for_author(user_id):
            raise RuntimeError("Subscription grant missing after Stripe payment")

    async def activate_from_session(
        self,
        stripe_session_id: str,
        password_hash: str,
        full_name: str | None = None,
    ) -> dict:
        """Create or upgrade User + ClientAccess from a verified Stripe Checkout session.

        Paid checkout always ends with the author able to log in with their chosen password.
        """
        session = await get_checkout_session(stripe_session_id)
        if session.payment_status != "paid":
            raise ValueError(f"Payment not completed. Status: {session.payment_status}")

        customer_email = _customer_email_from_session(session)
        stripe_customer_id = _stripe_resource_id(session.customer)
        stripe_sub_id = _stripe_resource_id(session.subscription)
        plan_id, token_budget = await self._resolve_plan(
            _stripe_metadata_value(session.metadata, "plan_id", "starter")
        )
        customer_name = full_name or getattr(session.customer_details, "name", "") or ""

        existing_user = await self._find_user_for_stripe_activation(
            customer_email,
            stripe_session_id,
            stripe_customer_id,
            stripe_sub_id,
        )

        try:
            if existing_user:
                result = await self._complete_existing_user_activation(
                    existing_user,
                    stripe_session_id=stripe_session_id,
                    password_hash=password_hash,
                    full_name=full_name,
                    plan_id=plan_id,
                    token_budget=token_budget,
                    stripe_customer_id=stripe_customer_id,
                    stripe_sub_id=stripe_sub_id,
                )
                await self._verify_post_activation(
                    result["user_id"],
                    plan_id=plan_id,
                    token_budget=token_budget,
                    stripe_sub_id=stripe_sub_id,
                    stripe_customer_id=stripe_customer_id,
                )
                return result

            user = await self._create_author_for_paid_session(
                email=customer_email,
                password_hash=password_hash,
                full_name=customer_name,
                plan_id=plan_id,
                token_budget=token_budget,
                stripe_customer_id=stripe_customer_id,
                stripe_sub_id=stripe_sub_id,
                stripe_session_id=stripe_session_id,
            )
            self._queue_credentials_email(
                email=customer_email,
                name=customer_name or customer_email,
                temp_password=None,
                user_chose_password=True,
            )
            logger.info("Author account activated from Stripe", user_id=user.id, plan_id=plan_id)
            result = {
                "user_id": user.id,
                "email": customer_email,
                "plan_id": plan_id,
                "already_activated": False,
            }
            await self._verify_post_activation(
                result["user_id"],
                plan_id=plan_id,
                token_budget=token_budget,
                stripe_sub_id=stripe_sub_id,
                stripe_customer_id=stripe_customer_id,
            )
            return result
        except IntegrityError:
            await self._db.rollback()
            logger.warning("Stripe activation race — linking existing account", email=customer_email)
            existing_user = await self._find_user_for_stripe_activation(
                customer_email,
                stripe_session_id,
                stripe_customer_id,
                stripe_sub_id,
            )
            if existing_user:
                result = await self._complete_existing_user_activation(
                    existing_user,
                    stripe_session_id=stripe_session_id,
                    password_hash=password_hash,
                    full_name=full_name,
                    plan_id=plan_id,
                    token_budget=token_budget,
                    stripe_customer_id=stripe_customer_id,
                    stripe_sub_id=stripe_sub_id,
                )
            else:
                result = await self._force_password_login(
                    customer_email=customer_email,
                    password_hash=password_hash,
                    full_name=full_name,
                    stripe_session_id=stripe_session_id,
                    stripe_customer_id=stripe_customer_id,
                    stripe_sub_id=stripe_sub_id,
                    plan_id=plan_id,
                    token_budget=token_budget,
                    full_name_for_create=customer_name,
                )
            await self._verify_post_activation(
                result["user_id"],
                plan_id=plan_id,
                token_budget=token_budget,
                stripe_sub_id=stripe_sub_id,
                stripe_customer_id=stripe_customer_id,
            )
            return result
        except Exception as exc:
            await self._db.rollback()
            logger.warning("Stripe activation fallback", email=customer_email, error=str(exc)[:200])
            result = await self._force_password_login(
                customer_email=customer_email,
                password_hash=password_hash,
                full_name=full_name,
                stripe_session_id=stripe_session_id,
                stripe_customer_id=stripe_customer_id,
                stripe_sub_id=stripe_sub_id,
                plan_id=plan_id,
                token_budget=token_budget,
                full_name_for_create=customer_name,
            )
            await self._verify_post_activation(
                result["user_id"],
                plan_id=plan_id,
                token_budget=token_budget,
                stripe_sub_id=stripe_sub_id,
                stripe_customer_id=stripe_customer_id,
            )
            return result

    async def handle_embed_unlock_payment(self, stripe_session_id: str) -> dict | None:
        """Grant one embed unlock credit after paid one-time Checkout (R-202).

        Idempotent by Stripe session id — safe for webhook + browser confirm.
        """
        from datetime import datetime, timezone

        from sqlalchemy.exc import IntegrityError

        from app.models.stripe_event import StripeEvent
        from app.models.user import User

        marker_id = f"embed_unlock:{stripe_session_id}"

        async def _credits_for(author_id: str) -> dict:
            user = await self._db.get(User, author_id) if author_id else None
            return {
                "author_id": author_id,
                "embed_unlock_credits": int(user.embed_unlock_credits) if user else 0,
                "already_processed": True,
            }

        existing = await self._db.get(StripeEvent, marker_id)
        if existing and existing.status == "processed":
            author_id = ""
            try:
                import json

                payload = json.loads(existing.raw_payload or "{}")
                author_id = str(payload.get("author_id") or "")
            except Exception:
                pass
            return await _credits_for(author_id)

        session = await get_checkout_session(stripe_session_id)
        if session.payment_status != "paid":
            logger.warning("embed_unlock skipped — not paid", session_id=stripe_session_id)
            return None
        mode = getattr(session, "mode", None) or ""
        if mode and mode != "payment":
            logger.warning("embed_unlock skipped — wrong mode", mode=mode)
            return None
        addon = _stripe_metadata_value(session.metadata, "addon", "")
        if addon != "embed_unlock":
            return None
        author_id = _stripe_metadata_value(session.metadata, "author_id", "")
        if not author_id:
            logger.warning("embed_unlock missing author_id", session_id=stripe_session_id)
            return None

        # Claim idempotency lock BEFORE granting (webhook + browser confirm race)
        import json

        if not existing:
            try:
                self._db.add(
                    StripeEvent(
                        id=marker_id,
                        event_type="embed_unlock.payment",
                        related_object_id=stripe_session_id,
                        related_object_type="checkout.session",
                        received_at=datetime.now(timezone.utc),
                        processed_at=datetime.now(timezone.utc),
                        status="processed",
                        raw_payload=json.dumps(
                            {"author_id": author_id, "addon": "embed_unlock"}
                        ),
                    )
                )
                await self._db.flush()
            except IntegrityError:
                await self._db.rollback()
                return await _credits_for(author_id)

        from app.services.embed_domain_service import EmbedDomainService

        total = await EmbedDomainService(self._db).apply_paid_unlock_credit(author_id)
        return {"author_id": author_id, "embed_unlock_credits": total}

    async def provision_from_checkout_webhook(self, stripe_session_id: str) -> dict | None:
        """Webhook fallback: auto-provision if user never completes /success page."""
        session = await get_checkout_session(stripe_session_id)
        if session.payment_status != "paid":
            logger.warning("Webhook provision skipped — not paid", session_id=stripe_session_id)
            return None

        addon = _stripe_metadata_value(session.metadata, "addon", "")
        if addon == "embed_unlock":
            return await self.handle_embed_unlock_payment(stripe_session_id)

        customer_email = _customer_email_from_session(session)
        existing = await self._users.get_by_email(customer_email)
        stripe_customer_id = _stripe_resource_id(session.customer)
        stripe_sub_id = _stripe_resource_id(session.subscription)
        plan_id, token_budget = await self._resolve_plan(
            _stripe_metadata_value(session.metadata, "plan_id", "starter")
        )
        if existing:
            existing.stripe_customer_id = existing.stripe_customer_id or stripe_customer_id
            existing.stripe_subscription_id = existing.stripe_subscription_id or stripe_sub_id
            existing.stripe_plan_id = existing.stripe_plan_id or plan_id
            if not existing.stripe_checkout_session_id:
                existing.stripe_checkout_session_id = stripe_session_id
            await self._ensure_stripe_client_access(
                author_id=existing.id,
                plan_id=plan_id,
                token_budget=token_budget,
                stripe_sub_id=stripe_sub_id,
            )
            logger.info("Webhook ensured access for existing user", email=customer_email)
            return {"skipped": True, "user_id": existing.id, "access_ensured": True}

        temp_password = self._generate_temp_password()
        customer_name = getattr(session.customer_details, "name", "") or ""

        result = await self._provision_stripe_author(
            email=customer_email,
            password_hash=_hash_password(temp_password),
            full_name=customer_name,
            plan_id=plan_id,
            token_budget=token_budget,
            stripe_customer_id=stripe_customer_id,
            stripe_sub_id=stripe_sub_id,
            stripe_session_id=stripe_session_id,
            provisioned_via="stripe_webhook",
            activation_status="pending",
            password_change_required=True,
        )
        user = await self._users.get_by_id(result["user_id"])
        if user:
            user.credentials_email_sent_at = datetime.now(timezone.utc)
            await self._db.commit()

        self._queue_credentials_email(
            email=customer_email,
            name=customer_name or customer_email,
            temp_password=temp_password,
            user_chose_password=False,
        )
        logger.info("Author auto-provisioned via webhook", user_id=result["user_id"])
        return result

    async def handle_payment_succeeded(self, stripe_subscription_id: str) -> None:
        """Reset monthly token counter on successful invoice payment.

        Called by: invoice.payment_succeeded webhook.
        """
        from sqlalchemy import select, update
        from app.models.client_access import ClientAccess

        result = await self._db.execute(
            select(ClientAccess).where(
                ClientAccess.stripe_subscription_id == stripe_subscription_id
            )
        )
        access = result.scalar_one_or_none()
        if not access:
            logger.warning("No ClientAccess for subscription", sub_id=stripe_subscription_id)
            return

        # Reset token counter + clear any grace period on the User
        access.tokens_used = 0
        access.is_revoked = False

        from app.models.user import User
        user_result = await self._db.execute(
            select(User).where(
                User.stripe_subscription_id == stripe_subscription_id
            )
        )
        user = user_result.scalar_one_or_none()
        if user:
            user.payment_grace_until = None

        await self._db.commit()
        logger.info("Token counter reset on payment success", sub_id=stripe_subscription_id)

    async def handle_payment_failed(
        self,
        stripe_subscription_id: str,
        grace_days: int = 7,
    ) -> None:
        """Set grace period on payment failure — access continues for grace_days.

        Called by: invoice.payment_failed webhook.
        The dunning email is sent by the webhook handler separately.
        """
        from sqlalchemy import select
        from app.models.user import User

        result = await self._db.execute(
            select(User).where(User.stripe_subscription_id == stripe_subscription_id)
        )
        user = result.scalar_one_or_none()
        if user:
            user.payment_grace_until = datetime.now(timezone.utc) + timedelta(days=grace_days)
            await self._db.commit()
            logger.info(
                "Payment grace period set",
                user_id=user.id,
                grace_until=user.payment_grace_until,
            )

    async def handle_subscription_cancelled(self, stripe_subscription_id: str) -> None:
        """Deactivate access when Stripe subscription is fully cancelled.

        Called by: customer.subscription.deleted webhook.
        """
        from sqlalchemy import select
        from app.models.client_access import ClientAccess

        result = await self._db.execute(
            select(ClientAccess).where(
                ClientAccess.stripe_subscription_id == stripe_subscription_id
            )
        )
        access = result.scalar_one_or_none()
        if access:
            access.is_revoked = True
            access.revoked_at = datetime.now(timezone.utc)
            access.revoke_reason = "stripe_subscription_cancelled"
            await self._db.commit()
            logger.info("Access revoked on subscription cancellation", sub_id=stripe_subscription_id)

    async def handle_plan_changed(
        self,
        stripe_subscription_id: str,
        new_plan_id: str,
    ) -> None:
        """Update plan limits when author upgrades or downgrades.

        Called by: customer.subscription.updated webhook.
        """
        from sqlalchemy import select
        from app.models.client_access import ClientAccess
        from app.models.pricing_plan import PricingPlan
        from app.models.user import User

        # Get new plan limits
        plan_result = await self._db.execute(
            select(PricingPlan).where(PricingPlan.id == new_plan_id)
        )
        plan = plan_result.scalar_one_or_none()
        if not plan:
            logger.warning("Unknown plan on subscription update", plan_id=new_plan_id)
            return

        # Update ClientAccess budget
        access_result = await self._db.execute(
            select(ClientAccess).where(
                ClientAccess.stripe_subscription_id == stripe_subscription_id
            )
        )
        access = access_result.scalar_one_or_none()
        if access:
            access.plan = new_plan_id
            access.plan_id = new_plan_id
            access.token_budget = plan.max_token_budget

        # Update User plan reference
        user_result = await self._db.execute(
            select(User).where(User.stripe_subscription_id == stripe_subscription_id)
        )
        user = user_result.scalar_one_or_none()
        if user:
            user.stripe_plan_id = new_plan_id

        await self._db.commit()
        logger.info(
            "Plan updated on subscription change",
            sub_id=stripe_subscription_id,
            new_plan=new_plan_id,
        )

    async def _load_active_plan(self, plan_id: str):
        """Load an active PricingPlan or raise ValueError."""
        from sqlalchemy import select
        from app.models.pricing_plan import PricingPlan

        result = await self._db.execute(
            select(PricingPlan).where(
                PricingPlan.id == plan_id,
                PricingPlan.is_active == True,  # noqa: E712
            )
        )
        plan = result.scalar_one_or_none()
        if not plan:
            raise ValueError(f"Plan '{plan_id}' not found or inactive")
        return plan

    def _stripe_price_for_billing(self, plan, billing: str) -> str:
        """Return Stripe Price ID for monthly/annual billing."""
        price_id = (
            plan.stripe_price_id_monthly
            if billing == "monthly"
            else plan.stripe_price_id_annual
        )
        if not price_id:
            raise ValueError(
                f"Plan '{plan.id}' is not configured for {billing} billing yet."
            )
        return price_id

    async def _stripe_subscription_is_active(self, subscription_id: str) -> bool:
        """True when Stripe reports the subscription as active or trialing."""
        try:
            _refresh_stripe_key()
            sub = stripe.Subscription.retrieve(subscription_id)
            status = getattr(sub, "status", None) or ""
            return status in ("active", "trialing", "past_due")
        except Exception as exc:
            logger.warning(
                "stripe_subscription_status_check_failed",
                sub_id=subscription_id,
                error=str(exc)[:120],
            )
            return False

    async def start_author_plan_action(
        self,
        author,
        plan_id: str,
        billing: str = "monthly",
        renew: bool = False,
        redis=None,
    ) -> dict:
        """Subscribe, renew/refill, or change plan for a logged-in author (R-184–R-188).

        Returns either a Checkout redirect or an in-place upgrade result.
        """
        from app.core.public_url import resolve_api_public_base_url
        from app.services.token_budget import percent_used, tokens_remaining

        plan = await self._load_active_plan(plan_id)
        stripe_price_id = self._stripe_price_for_billing(plan, billing)
        access = await self._access.get_active_for_author(author.id)
        current_plan = (
            (author.stripe_plan_id or "")
            or (getattr(access, "plan_id", None) or "")
            or (getattr(access, "plan", None) or "")
        ).strip().lower()
        target = plan_id.strip().lower()
        same_plan = bool(current_plan) and (
            current_plan == target or current_plan == plan.display_name.strip().lower()
        )

        needs_refill = False
        if access:
            needs_refill = tokens_remaining(access) <= 0 or percent_used(access) >= 80.0
        if renew or (same_plan and needs_refill):
            return await self._start_plan_renew_refill_checkout(
                author=author,
                plan=plan,
                billing=billing,
            )

        sub_id = (author.stripe_subscription_id or "").strip()
        if sub_id and await self._stripe_subscription_is_active(sub_id):
            if same_plan:
                raise ValueError(
                    "You are already on this plan. When your token budget runs low, "
                    "use Renew & refill tokens on Billing."
                )
            await upgrade_subscription(sub_id, stripe_price_id, new_plan_id=plan_id)
            await self.handle_plan_changed(sub_id, plan_id)
            return {
                "action": "upgraded",
                "plan_id": plan_id,
                "plan_name": plan.display_name,
                "billing": billing,
            }

        base = resolve_api_public_base_url()
        success_url = (
            f"{base}/author?page=billing&checkout=success"
            f"&session_id={{CHECKOUT_SESSION_ID}}"
        )
        cancel_url = f"{base}/author?page=billing&checkout=cancel"

        checkout_url = await create_checkout_session(
            plan_id=plan_id,
            stripe_price_id=stripe_price_id,
            success_url=success_url,
            cancel_url=cancel_url,
            customer_email=None if author.stripe_customer_id else author.email,
            existing_customer_id=author.stripe_customer_id,
            author_id=author.id,
        )
        return {
            "action": "checkout",
            "checkout_url": checkout_url,
            "plan_id": plan_id,
            "plan_name": plan.display_name,
            "billing": billing,
        }

    def _renew_refill_price_usd(self, plan, billing: str) -> float:
        """USD charge for one renew/refill cycle."""
        if billing == "annual":
            # annual_price_usd is per-month when billed annually
            return float(plan.annual_price_usd or 0) * 12
        return float(plan.monthly_price_usd or 0)

    async def _start_plan_renew_refill_checkout(
        self,
        author,
        plan,
        billing: str = "monthly",
    ) -> dict:
        """One-time Checkout to refill token budget for the current (or chosen) plan."""
        from app.core.public_url import resolve_api_public_base_url

        price_usd = self._renew_refill_price_usd(plan, billing)
        if price_usd < 0.5:
            raise ValueError("This plan has no renew price configured yet.")

        base = resolve_api_public_base_url()
        success_url = (
            f"{base}/author?page=billing&checkout=success"
            f"&session_id={{CHECKOUT_SESSION_ID}}"
        )
        cancel_url = f"{base}/author?page=billing&checkout=cancel"
        period = "annual" if billing == "annual" else "monthly"
        checkout_url = await create_addon_checkout_session(
            addon_id="plan_renew_refill",
            price_usd=price_usd,
            display_name=f"{plan.display_name} — renew & refill tokens",
            author_id=author.id,
            success_url=success_url,
            cancel_url=cancel_url,
            customer_email=None if author.stripe_customer_id else author.email,
            existing_customer_id=author.stripe_customer_id,
            description=(
                f"Refills your {plan.display_name} token budget for one {period} period."
            ),
            extra_metadata={"plan_id": plan.id, "billing": billing},
        )
        return {
            "action": "checkout",
            "checkout_url": checkout_url,
            "plan_id": plan.id,
            "plan_name": plan.display_name,
            "billing": billing,
            "renew_refill": True,
        }

    async def apply_plan_renew_refill(
        self,
        author_id: str,
        plan_id: str,
        billing: str = "monthly",
        redis=None,
    ) -> dict:
        """Reset token usage and extend access after a paid renew/refill (R-188)."""
        from datetime import timedelta

        from app.services.token_budget import clear_budget_exhausted

        plan = await self._load_active_plan(plan_id)
        access = await self._access.get_active_for_author(author_id)
        if not access:
            from sqlalchemy import select

            from app.models.client_access import ClientAccess

            result = await self._db.execute(
                select(ClientAccess)
                .where(ClientAccess.author_id == author_id)
                .order_by(ClientAccess.granted_at.desc())
                .limit(1)
            )
            access = result.scalar_one_or_none()
        if not access:
            raise ValueError("No subscription found to refill. Subscribe to a plan first.")

        now = datetime.now(timezone.utc)
        days = 365 if billing == "annual" else 30
        exp = access.expires_at
        if exp is not None and exp.tzinfo is None:
            exp = exp.replace(tzinfo=timezone.utc)
        base = exp if exp and exp > now else now
        access.tokens_used = 0
        access.is_revoked = False
        access.revoked_at = None
        access.revoke_reason = None
        access.plan = plan_id
        access.plan_id = plan_id
        access.token_budget = plan.max_token_budget
        access.expires_at = base + timedelta(days=days)

        user = await self._users.get_by_id(author_id)
        if user:
            user.stripe_plan_id = plan_id
            user.payment_grace_until = None
            user.chatbot_is_active = True

        await self._db.commit()

        if redis is not None:
            try:
                await clear_budget_exhausted(redis, author_id)
            except Exception as exc:
                logger.warning(
                    "clear_budget_exhausted_failed",
                    author_id=author_id,
                    error=str(exc)[:120],
                )

        logger.info(
            "Plan renew refill applied",
            author_id=author_id,
            plan_id=plan_id,
            billing=billing,
        )
        return {
            "ok": True,
            "plan_id": plan_id,
            "tokens_remaining": plan.max_token_budget,
            "expires_at": access.expires_at.isoformat() if access.expires_at else None,
        }

    async def handle_plan_renew_refill_payment(self, stripe_session_id: str, redis=None) -> dict | None:
        """Apply renew/refill after paid one-time Checkout (webhook or confirm)."""
        session = await get_checkout_session(stripe_session_id)
        if getattr(session, "payment_status", None) != "paid":
            return None
        meta = getattr(session, "metadata", None)
        addon = _stripe_metadata_value(meta, "addon", "")
        if addon != "plan_renew_refill":
            return None
        author_id = _stripe_metadata_value(meta, "author_id", "")
        plan_id = _stripe_metadata_value(meta, "plan_id", "starter")
        billing = _stripe_metadata_value(meta, "billing", "monthly")
        if not author_id:
            logger.warning("plan_renew_refill missing author_id", session_id=stripe_session_id)
            return None
        return await self.apply_plan_renew_refill(
            author_id=author_id,
            plan_id=plan_id,
            billing=billing if billing in ("monthly", "annual") else "monthly",
            redis=redis,
        )

    async def confirm_author_checkout(self, author, session_id: str, redis=None) -> dict:
        """Apply a paid Checkout session to the logged-in author and repair grant."""
        session = await get_checkout_session(session_id)
        if getattr(session, "payment_status", None) != "paid":
            raise ValueError("Payment not completed yet. Please wait a moment and refresh.")

        meta_author = _stripe_metadata_value(session.metadata, "author_id", "")
        if meta_author and meta_author != author.id:
            raise ValueError("This payment does not belong to your account.")

        addon = _stripe_metadata_value(session.metadata, "addon", "")
        if addon == "plan_renew_refill":
            result = await self.handle_plan_renew_refill_payment(session_id, redis=redis)
            if not result:
                raise ValueError("Could not apply renew payment.")
            access = await self._access.get_active_for_author(author.id)
            fresh = await self._users.get_by_id(author.id)
            return {
                "ok": True,
                "has_active_subscription": access is not None,
                "plan": access.plan if access else (fresh.stripe_plan_id if fresh else None),
                "stripe_customer_id": fresh.stripe_customer_id if fresh else None,
                "renew_refill": True,
            }

        customer_email = ""
        try:
            customer_email = _customer_email_from_session(session)
        except ValueError:
            customer_email = ""
        if customer_email and customer_email.lower() != (author.email or "").lower():
            cust_id = ""
            try:
                cust_id = _stripe_resource_id(session.customer)
            except ValueError:
                cust_id = ""
            if not meta_author and cust_id != (author.stripe_customer_id or ""):
                raise ValueError("This payment email does not match your account.")

        await self.provision_from_checkout_webhook(session_id)
        await self.require_active_subscription(author.id)
        access = await self._access.get_active_for_author(author.id)
        fresh = await self._users.get_by_id(author.id)
        return {
            "ok": True,
            "has_active_subscription": access is not None,
            "plan": access.plan if access else (fresh.stripe_plan_id if fresh else None),
            "stripe_customer_id": fresh.stripe_customer_id if fresh else None,
        }