"""Billing & Usage domain. The internal usage ledger is the source of truth. Stripe (or any other payment provider) is an *adapter* that mirrors it — never the other way around. """ from __future__ import annotations from dataclasses import dataclass, field from datetime import datetime from enum import StrEnum from app.domain.ai_control.models import RateUnit from app.domain.base import new_id, utcnow class SubscriptionStatus(StrEnum): TRIALING = "trialing" ACTIVE = "active" PAST_DUE = "past_due" CANCELED = "canceled" class InvoiceStatus(StrEnum): DRAFT = "draft" OPEN = "open" PAID = "paid" VOID = "void" UNCOLLECTIBLE = "uncollectible" class PaymentStatus(StrEnum): PENDING = "pending" SUCCEEDED = "succeeded" FAILED = "failed" REFUNDED = "refunded" @dataclass(slots=True, kw_only=True) class Plan: name: str monthly_price: float = 0.0 included_credit: float = 0.0 features_json: dict = field(default_factory=dict) id: str = field(default_factory=new_id) is_active: bool = True created_at: datetime = field(default_factory=utcnow) @dataclass(slots=True, kw_only=True) class Subscription: organization_id: str plan_id: str status: SubscriptionStatus = SubscriptionStatus.ACTIVE external_subscription_id: str | None = None current_period_end: datetime | None = None id: str = field(default_factory=new_id) created_at: datetime = field(default_factory=utcnow) @dataclass(slots=True, kw_only=True) class CreditWallet: organization_id: str balance: float = 0.0 currency: str = "USD" low_balance_threshold: float = 5.0 id: str = field(default_factory=new_id) updated_at: datetime = field(default_factory=utcnow) @property def is_low(self) -> bool: return self.balance <= self.low_balance_threshold def can_afford(self, amount: float) -> bool: return self.balance >= amount def debit(self, amount: float) -> None: self.balance -= amount self.updated_at = utcnow() def credit(self, amount: float) -> None: self.balance += amount self.updated_at = utcnow() class CreditKind(StrEnum): """Why a wallet's balance moved.""" OWNER_CREDIT = "owner_credit" STARTING_CREDIT = "starting_credit" USAGE = "usage" ADJUSTMENT = "adjustment" @dataclass(slots=True, kw_only=True) class CreditTransaction: """One change to a wallet's balance, with the balance it left behind. ``amount`` is signed: a credit is positive, a debit is negative. ``usage_event_id`` is set on usage debits only, so one usage row debits once. """ organization_id: str kind: CreditKind amount: float balance_after: float reason: str = "" actor_user_id: str | None = None usage_event_id: str | None = None id: str = field(default_factory=new_id) created_at: datetime = field(default_factory=utcnow) @dataclass(slots=True, kw_only=True) class UsageEvent: """One billable unit of work. Append-only.""" organization_id: str unit: RateUnit quantity: float unit_price: float = 0.0 internal_cost_estimate: float = 0.0 total_price: float = 0.0 project_id: str | None = None user_id: str | None = None employee_id: str | None = None run_id: str | None = None metadata_json: dict = field(default_factory=dict) id: str = field(default_factory=new_id) created_at: datetime = field(default_factory=utcnow) @dataclass(slots=True, kw_only=True) class Invoice: organization_id: str amount: float status: InvoiceStatus = InvoiceStatus.DRAFT currency: str = "USD" external_invoice_id: str | None = None period_start: datetime | None = None period_end: datetime | None = None id: str = field(default_factory=new_id) created_at: datetime = field(default_factory=utcnow) @dataclass(slots=True, kw_only=True) class Payment: organization_id: str amount: float status: PaymentStatus = PaymentStatus.PENDING provider: str = "mock" external_payment_id: str | None = None invoice_id: str | None = None id: str = field(default_factory=new_id) created_at: datetime = field(default_factory=utcnow)