smart-org/apps/api/app/application/services/credit_service.py

"""Prepaid credit: the platform owner's top-ups, the history and the balances.

An organization's wallet is only ever moved through ``CreditLedgerRepository``, so every
change has a history row. This service is the owner's side of that ledger: it credits one
organization by hand, reads its manual history, lists every organization's balance for the
owner's credit page, and grants the optional starting credit to a new organization. Usage
debits are not written here; ``UsageLedgerRepository.record`` writes them.

Only the platform owner may call the write methods and the cross-organization ``balances``.
The routers enforce that; this service does not look at who is calling.
"""

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime

from sqlalchemy.ext.asyncio import AsyncSession

from app.application.ports.realtime import emit_wallet_changed
from app.core.config import settings
from app.core.errors import NotFoundError, ValidationError
from app.domain.billing.models import CreditKind, CreditTransaction
from app.domain.workforce.pipeline import Stage
from app.infrastructure.repositories.billing import CreditWalletRepository
from app.infrastructure.repositories.credit import CreditLedgerRepository
from app.infrastructure.repositories.engine import PipelineStateRepository
from app.infrastructure.repositories.identity import OrganizationRepository

#: The most one owner action may credit at once. A bigger top-up is made in steps.
MAX_OWNER_CREDIT = 100_000.0

#: The reason is required and stored with the history row (at most 500 characters).
_REASON_MIN = 3
_REASON_MAX = 500

#: The entries a history page shows: what someone did to the wallet, not usage debits.
_MANUAL_KINDS = (CreditKind.OWNER_CREDIT, CreditKind.STARTING_CREDIT, CreditKind.ADJUSTMENT)


@dataclass(slots=True, kw_only=True)
class OrgBalance:
    """One organization's row on the platform owner's credit page."""

    organization_id: str
    name: str
    slug: str
    balance: float
    currency: str
    paused_tasks: int
    last_credit_at: datetime | None


class CreditService:
    def __init__(self, session: AsyncSession) -> None:
        self.session = session
        self.ledger = CreditLedgerRepository(session)
        self.wallets = CreditWalletRepository(session)
        self.orgs = OrganizationRepository(session)
        self.states = PipelineStateRepository(session)

    async def owner_credit(
        self,
        *,
        organization_id: str,
        amount: float,
        reason: str,
        actor_user_id: str,
    ) -> CreditTransaction:
        """Credit one organization's wallet on the platform owner's word."""
        if await self.orgs.get(organization_id) is None:
            raise NotFoundError("سازمان یافت نشد.")
        if not 0 < amount <= MAX_OWNER_CREDIT:
            raise ValidationError(
                f"مقدار اعتبار باید بیشتر از صفر و حداکثر {MAX_OWNER_CREDIT:g} باشد."
            )
        clean_reason = reason.strip()
        if not _REASON_MIN <= len(clean_reason) <= _REASON_MAX:
            raise ValidationError("دلیل افزودن اعتبار لازم است.")

        transaction = await self.ledger.apply(
            organization_id=organization_id,
            amount=amount,
            kind=CreditKind.OWNER_CREDIT,
            reason=clean_reason,
            actor_user_id=actor_user_id,
        )
        # Live push: the sidebar credit badge follows the wallet as it moves.
        await emit_wallet_changed(
            organization_id=organization_id, balance=transaction.balance_after
        )
        return transaction

    async def history(self, organization_id: str, *, limit: int = 50) -> list[CreditTransaction]:
        """One organization's manual credit history, newest first. Usage debits are left out."""
        return await self.ledger.history(organization_id, kinds=_MANUAL_KINDS, limit=limit)

    async def balances(self) -> list[OrgBalance]:
        """Every organization's balance, paused task count and last credit date.

        Cross-organization on purpose: only the platform owner's credit page may call it.
        A missing wallet is shown as zero.
        """
        paused = await self.states.count_at_by_org(Stage.PAUSED)
        last_credit = await self.ledger.last_credit_at_by_org()
        rows: list[OrgBalance] = []
        for org in await self.orgs.list_all():
            wallet = await self.wallets.get_for_org(org.id)
            rows.append(
                OrgBalance(
                    organization_id=org.id,
                    name=org.name,
                    slug=org.slug,
                    balance=wallet.balance if wallet else 0.0,
                    currency=wallet.currency if wallet else "USD",
                    paused_tasks=paused.get(org.id, 0),
                    last_credit_at=last_credit.get(org.id),
                )
            )
        return rows

    async def grant_starting_credit(self, organization_id: str) -> CreditTransaction | None:
        """Give a new organization the configured starting credit, if one is set.

        ``BILLING_STARTING_CREDIT`` defaults to 0, so by default nothing is granted and the
        owner credits new organizations by hand.
        """
        amount = float(getattr(settings, "billing_starting_credit", 0.0))
        if amount <= 0:
            return None
        return await self.ledger.apply(
            organization_id=organization_id,
            amount=amount,
            kind=CreditKind.STARTING_CREDIT,
            reason="اعتبار آغازین",
        )