events/promo.py

"""Promo-code arithmetic shared by the validate endpoint and the purchase path.

Pure functions (no DB, no settings) so the number a buyer sees in the
"Apply" preview is exactly the number the invoice / Stripe session charges.
Field names and the `BasketTotals` shape follow upstream lnbits/events v2
(PR #64) so the eventual rebase (#33) merges cleanly; deviations are noted
inline.
"""

from __future__ import annotations

from collections import Counter

from .models import BasketDiscount, BasketTotals, Event, PromoCode, Ticket, TicketWave

SAT_UNITS = ("sat", "sats")


def normalize_code(raw: str | None) -> str | None:
    """Buyer input → stored form (stripped, upper-cased); empty → None."""
    if raw is None:
        return None
    code = raw.strip().upper()
    return code or None


def find_promo(event: Event, code: str) -> PromoCode | None:
    return next((pc for pc in event.extra.promo_codes if pc.code == code), None)


def promo_usage(tickets: list[Ticket]) -> dict[str, int]:
    """Redemptions per code = PAID tickets carrying it in
    `extra.applied_promo_code`. Every row counts, so a multi-ticket
    purchase consumes `quantity` uses (upstream v2 counts one per basket).

    Paid only: pending rows live up to 24 h (`purge_unpaid_tickets`), so
    counting them would let an abandoned Stripe session lock out the last
    uses of a limited code for a day. The cost is a bounded overshoot when
    several buyers pass the check before any of them pays — accepted.
    """
    counter: Counter[str] = Counter()
    for ticket in tickets:
        code = ticket.extra.applied_promo_code
        if ticket.paid and code:
            counter[code] += 1
    return dict(counter)


def remaining_uses(promo: PromoCode, used: int) -> int | None:
    """None = unlimited (`max_uses` unset / 0)."""
    if not promo.max_uses:
        return None
    return max(promo.max_uses - used, 0)


def round_amount(amount: float, currency: str | None) -> float:
    """Sats are integers; fiat is 2 dp. Applied once, at the end, so
    subtotal - total == discount holds for what is actually charged."""
    if (currency or "sat").lower() in SAT_UNITS:
        return float(int(amount))
    return round(amount, 2)


def basket_totals(
    event: Event,
    codes: list[str],
    quantity: int,
    usage: dict[str, int],
    wave: TicketWave,
) -> BasketTotals:
    """Price `quantity` tickets from `wave` with the first applicable code.

    A code is applicable when it exists, is active, has enough uses left
    for the whole quantity, and actually saves something. Anything else is
    simply absent from `discounts_applied` (upstream v2 semantics — the
    purchase endpoint is where hard errors are raised). Only one code is
    applied; v2's `combinable` stacking is out of scope here.

    `wave` is the pricing authority, not `event`. Since v1.6.8 a ticket's
    price and currency belong to the wave it is bought from, and the
    event-level fields are a derived roll-up of the PRIMARY wave
    (`sync_event_ticket_waves`) — pricing off `event` would quote and charge
    the first wave's price to a buyer who picked a later one. It is a
    required argument rather than an optional override precisely because
    that failure is silent: both call sites have to name the wave.
    """
    currency = wave.currency or "sat"
    subtotal = round_amount(wave.price_per_ticket * quantity, currency)
    totals = BasketTotals(
        subtotal=subtotal, discount=0, total=subtotal, currency=currency
    )
    for raw in codes:
        code = normalize_code(raw)
        if not code:
            continue
        promo = find_promo(event, code)
        if not promo or not promo.active:
            continue
        remaining = remaining_uses(promo, usage.get(promo.code, 0))
        if remaining is not None and remaining < quantity:
            continue
        total = round_amount(subtotal * (1 - promo.discount_percent / 100), currency)
        saved = round_amount(subtotal - total, currency)
        if saved <= 0:
            continue
        totals.total = total
        totals.discount = saved
        totals.discounts_applied = [
            BasketDiscount(
                code=promo.code,
                discount_percent=promo.discount_percent,
                discount_fixed=None,
                amount_saved=saved,
            )
        ]
        break
    return totals