smart-org/apps/api/app/domain/billing/models.py

"""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)