qtalker---/backend/app/inventory.py

"""Unlocks, inventory items, sigils, and the essence economy — Workstream C
of docs/superpowers/specs/2026-07-23-character-depth-ghost-log-design.md.

Essence amounts (the implementer's call per the spec's "exact amounts are
the implementer's call; keep summon income small (single digits) and
milestone income larger (tens)" guidance):
  - SUMMON_ESSENCE_TRICKLE: every successful summon, any mode/rarity — kept
    tiny so it accrues constantly without needing any gating logic. This is
    the only trigger point that exists in `backend/app/ws.py` today (see
    module docstring there); it's wired directly in `_handle_summon`.
  - RITUAL_SUCCESS_ESSENCE / CORRECT_JUDGMENT_ESSENCE / CROSS_OVER_ESSENCE:
    milestone constants for Workstream B's ritual/judgment WS handlers to
    draw from once they land (those handlers — `ritual_start`,
    `ritual_step`, `judgment` — don't exist in this worktree yet; Workstream
    B owns `backend/app/judgment.py` and the WS wiring per the spec). They're
    defined here, alongside the drop table, so both workstreams price things
    out of one shared module instead of duplicating numbers. Cross-over pays
    the most, matching the contract's "the largest essence reward of any
    outcome".

Item drop odds (also the implementer's call): ritual success and correct
judgments are deliberate, effortful player actions, so they roll a bit more
generously than a bare summon (which happens constantly and is often
passive/ambient). High-rarity summons scale with how rare the entity itself
already is — a mythic summon is already a jackpot, so the drop on top of it
pays off further. None of these odds are large enough to make items feel
guaranteed/grindy, matching the same "reachable within a handful of
sessions, not grindy" spirit as the essence guidance.
"""

import random
import uuid

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.unlock import UnlockRecord
from app.models.user import User

# --- Essence economy --------------------------------------------------

SUMMON_ESSENCE_TRICKLE = 2  # every successful summon, any mode
RITUAL_SUCCESS_ESSENCE = 15  # a completed ritual
CORRECT_JUDGMENT_ESSENCE = 12  # correct trust/banish call
CROSS_OVER_ESSENCE = 25  # correct cross_over — largest reward of any outcome

# --- Unlocks ------------------------------------------------------------

UNLOCK_PRICES: dict[str, int] = {
    "listening_tool": 40,
}

# --- Item drops -----------------------------------------------------------

HIGH_RARITY_TIERS = {"rare", "mythic"}

# (trigger key) -> drop chance. Triggers correspond to the contract's
# `item_drop` emission points: "after a correct judgment, a successful
# ritual, or a high-rarity summon" — the last one is split by tier since a
# mythic summon should feel more rewarded than a merely-rare one.
DROP_CHANCES: dict[str, float] = {
    "judgment": 0.20,
    "ritual": 0.25,
    "summon_rare": 0.15,
    "summon_mythic": 0.35,
}

# (trigger key) -> (item_type, item_key pool)
ITEM_POOLS: dict[str, tuple[str, list[str]]] = {
    "judgment": (
        "trinket",
        ["static_shard", "cracked_locket", "cold_coin", "grave_dust"],
    ),
    "ritual": (
        "relic",
        ["obsidian_mirror", "bone_dial", "silver_tuning_fork", "warded_chalk"],
    ),
    "summon_rare": (
        "curio",
        ["moth_wing", "ectoplasm_vial", "tarnished_key"],
    ),
    "summon_mythic": (
        "curio",
        ["black_candle_stub", "veil_thread", "gilded_grave_dust"],
    ),
}


def summon_drop_trigger(rarity: str) -> str | None:
    """Maps an entity's rarity tier to its drop-trigger key, or None if the
    summon isn't high-rarity enough to qualify for a drop roll at all."""
    if rarity == "rare":
        return "summon_rare"
    if rarity == "mythic":
        return "summon_mythic"
    return None


def roll_item_drop(trigger: str, rng: random.Random | None = None) -> dict | None:
    """Rolls for an item drop at one of the contract's trigger points
    ("judgment", "ritual", "summon_rare", "summon_mythic"). Returns the
    `item_drop` frame's `item` payload shape
    (`{"item_type", "item_key", "payload"}`), or None on a miss / unknown
    trigger."""
    rng = rng if rng is not None else random.Random()
    chance = DROP_CHANCES.get(trigger)
    pool = ITEM_POOLS.get(trigger)
    if chance is None or pool is None:
        return None
    if rng.random() >= chance:
        return None
    item_type, keys = pool
    return {
        "item_type": item_type,
        "item_key": rng.choice(keys),
        "payload": {"trigger": trigger},
    }


def credit_essence(user: User, amount: int) -> int:
    """Adjusts `user.essence` in place, floored at 0, and returns the new
    balance. Caller is responsible for committing/flushing."""
    user.essence = max(0, user.essence + amount)
    return user.essence


# --- Sigil design validation ---------------------------------------------

SIGIL_MAX_POINTS = 12
_RUNE_MAX_LEN = 32


def validate_sigil_design(design: object) -> dict | None:
    """Validates the placeholder sigil design shape
    `{"points": [[x, y], ...], "rune": str}` (a stand-in until Workstream F's
    sigil designer defines the real shape — see the spec's Workstream C
    section). `points` is capped at `SIGIL_MAX_POINTS` to bound payload
    size. Returns a normalized dict on success, or None if the shape is
    invalid."""
    if not isinstance(design, dict):
        return None

    points = design.get("points")
    if not isinstance(points, list) or not (1 <= len(points) <= SIGIL_MAX_POINTS):
        return None

    normalized_points: list[list[float]] = []
    for point in points:
        if (
            not isinstance(point, (list, tuple))
            or len(point) != 2
            or not all(
                isinstance(coord, (int, float)) and not isinstance(coord, bool)
                for coord in point
            )
        ):
            return None
        normalized_points.append([float(point[0]), float(point[1])])

    rune = design.get("rune")
    if not isinstance(rune, str) or not (1 <= len(rune) <= _RUNE_MAX_LEN):
        return None

    return {"points": normalized_points, "rune": rune}


# --- Purchases --------------------------------------------------------


class InsufficientEssenceError(Exception):
    """Raised when a user tries to buy an unlock they can't afford."""


class UnknownUnlockError(Exception):
    """Raised when `unlock_key` isn't in `UNLOCK_PRICES`."""


async def purchase_unlock(
    db: AsyncSession, user_id: uuid.UUID, unlock_key: str
) -> UnlockRecord:
    """Atomically spends essence for `unlock_key` and records the unlock.

    Locks the user's row (`SELECT ... FOR UPDATE`) for the duration of the
    balance check + deduction, so two concurrent purchase requests can't
    both read the same starting balance and both succeed (double-spend).
    The second request blocks on the row lock until the first commits, then
    re-reads the now-decremented balance. Idempotent: re-buying an unlock
    already owned returns the existing record without charging again.
    """
    if unlock_key not in UNLOCK_PRICES:
        raise UnknownUnlockError(unlock_key)
    price = UNLOCK_PRICES[unlock_key]

    locked_user = await db.scalar(
        select(User).where(User.id == user_id).with_for_update()
    )
    if locked_user is None:
        raise UnknownUnlockError(unlock_key)  # user vanished mid-request

    existing = await db.scalar(
        select(UnlockRecord).where(
            UnlockRecord.user_id == user_id, UnlockRecord.unlock_key == unlock_key
        )
    )
    if existing is not None:
        return existing

    if locked_user.essence < price:
        raise InsufficientEssenceError(
            f"not enough essence for {unlock_key!r}: "
            f"have {locked_user.essence}, need {price}"
        )

    locked_user.essence -= price
    record = UnlockRecord(user_id=user_id, unlock_key=unlock_key)
    db.add(record)
    await db.commit()
    await db.refresh(record)
    return record