crm-system/app/services/deal_service.py

"""Deal service: create, get, list, update, update_stage, get_pipeline, soft-delete."""

from __future__ import annotations

from collections import defaultdict
from datetime import UTC, datetime
from decimal import Decimal

from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.account import Account
from app.models.deal import Deal, DealStage
from app.models.deal_stage_history import DealStageHistory
from app.schemas.deal import DealCreate, DealUpdate
from app.services._base import OrgScopedQuery


class DealNotFound(Exception):
    """Raised when a deal lookup fails."""


class InvalidAccount(Exception):
    """Raised when account_id points to a non-existent account."""


def _stage_value(stage: DealStage | str) -> str:
    """Normalize stage to its string value."""
    if hasattr(stage, "value"):
        return stage.value  # type: ignore[union-attr]
    return str(stage)


async def create_deal(
    db: AsyncSession,
    payload: DealCreate,
    *,
    org_id: int,
    owner_id: int,
) -> Deal:
    """Create a new deal. Validates account_id exists in org."""
    acc_q = OrgScopedQuery(Account, db, org_id=org_id)
    if await acc_q.get(payload.account_id) is None:
        raise InvalidAccount(f"Account {payload.account_id} not found in this org")
    deal = Deal(
        org_id=org_id,
        title=payload.title,
        value=payload.value,
        currency=payload.currency,
        stage=payload.stage,
        close_date=payload.close_date,
        account_id=payload.account_id,
        owner_id=owner_id,
        won_lost_reason=payload.won_lost_reason,
    )
    db.add(deal)
    try:
        await db.commit()
    except IntegrityError as e:
        await db.rollback()
        raise InvalidAccount(str(e)) from e
    await db.refresh(deal)
    # Initial stage history entry (from None → initial stage)
    history = DealStageHistory(
        org_id=org_id,
        deal_id=deal.id,
        from_stage=None,
        to_stage=deal.stage,
        changed_by=owner_id,
    )
    db.add(history)
    await db.commit()
    await db.refresh(deal)
    return deal


async def get_deal(
    db: AsyncSession, deal_id: int, *, org_id: int
) -> Deal | None:
    """Fetch a single deal by id."""
    q = OrgScopedQuery(Deal, db, org_id=org_id)
    return await q.get(deal_id)


async def list_deals(
    db: AsyncSession,
    *,
    org_id: int,
    skip: int = 0,
    limit: int = 20,
    stage: str | None = None,
    owner_id: int | None = None,
    account_id: int | None = None,
) -> list[Deal]:
    """List deals with optional filters."""
    scoped = OrgScopedQuery(Deal, db, org_id=org_id)
    return await scoped.list(
        skip=skip,
        limit=limit,
        order_by=Deal.id,
        stage=stage,
        owner_id=owner_id,
        account_id=account_id,
    )


async def update_deal(
    db: AsyncSession, deal: Deal, payload: DealUpdate
) -> Deal:
    """Apply partial updates to a deal. Does NOT change stage (use update_stage)."""
    data = payload.model_dump(exclude_unset=True)
    # Disallow direct stage changes via update endpoint
    data.pop("stage", None)
    for field, value in data.items():
        if value is not None:
            setattr(deal, field, value)
    await db.commit()
    await db.refresh(deal)
    return deal


async def update_stage(
    db: AsyncSession,
    deal: Deal,
    new_stage: DealStage,
    *,
    changed_by: int,
    reason: str | None = None,
) -> Deal:
    """Change a deal's stage and append a DealStageHistory record."""
    from_stage_str = _stage_value(deal.stage)
    to_stage_str = _stage_value(new_stage)
    if from_stage_str == to_stage_str:
        return deal  # no-op

    deal.stage = new_stage  # type: ignore[assignment]
    if new_stage in (DealStage.won, DealStage.lost) and reason is not None:
        deal.won_lost_reason = reason

    history = DealStageHistory(
        org_id=deal.org_id,
        deal_id=deal.id,
        from_stage=from_stage_str,
        to_stage=to_stage_str,
        changed_by=changed_by,
    )
    db.add(history)
    await db.commit()
    await db.refresh(deal)
    return deal


async def get_pipeline(
    db: AsyncSession, *, org_id: int
) -> list[dict[str, object]]:
    """Return deals grouped by stage for the pipeline view."""
    scoped = OrgScopedQuery(Deal, db, org_id=org_id)
    all_deals = await scoped.list(skip=0, limit=10_000, order_by=Deal.id)
    grouped: dict[str, list[Deal]] = defaultdict(list)
    for d in all_deals:
        grouped[_stage_value(d.stage)].append(d)
    return [
        {
            "stage": stage.value,
            "count": len(deals),
            "total_value": float(sum((d.value for d in deals), Decimal("0"))),
            "deals": deals,
        }
        for stage in DealStage
        for deals in [grouped.get(stage.value, [])]
    ]


async def soft_delete_deal(db: AsyncSession, deal: Deal) -> Deal:
    """Soft-delete a deal."""
    deal.deleted_at = datetime.now(UTC)
    await db.commit()
    await db.refresh(deal)
    return deal


__all__ = [
    "DealNotFound",
    "InvalidAccount",
    "create_deal",
    "get_deal",
    "get_pipeline",
    "list_deals",
    "soft_delete_deal",
    "update_deal",
    "update_stage",
]