app/api/deps.py

@@ -14,6 +14,7 @@ from app.repositories.deal_repo import DealRepository
 from app.repositories.org_repo import OrganizationRepository
 from app.repositories.user_repo import UserRepository
 from app.services.auth_service import AuthService
+from app.services.deal_service import DealService
 from app.services.organization_service import (
     OrganizationAccessDeniedError,
     OrganizationContext,
@@ -43,6 +44,10 @@ def get_deal_repository(session: AsyncSession = Depends(get_db_session)) -> Deal
     return DealRepository(session=session)
 
 
+def get_deal_service(repo: DealRepository = Depends(get_deal_repository)) -> DealService:
+    return DealService(repository=repo)
+
+
 def get_user_service(repo: UserRepository = Depends(get_user_repository)) -> UserService:
     return UserService(user_repository=repo, password_hasher=password_hasher)
 

app/services/__init__.py

@@ -1 +1,11 @@
 """Business logic services."""
+
+from .deal_service import DealService  # noqa: F401
+from .organization_service import (  # noqa: F401
+    OrganizationAccessDeniedError,
+    OrganizationContext,
+    OrganizationContextMissingError,
+    OrganizationService,
+)
+from .user_service import UserService  # noqa: F401
+from .auth_service import AuthService  # noqa: F401

app/services/deal_service.py

@@ -0,0 +1,164 @@
+"""Business logic for deals."""
+from __future__ import annotations
+
+from collections.abc import Iterable
+from dataclasses import dataclass
+from decimal import Decimal
+
+from sqlalchemy import func, select
+
+from app.models.activity import Activity, ActivityType
+from app.models.contact import Contact
+from app.models.deal import Deal, DealCreate, DealStage, DealStatus
+from app.models.organization_member import OrganizationRole
+from app.repositories.deal_repo import DealRepository
+from app.services.organization_service import OrganizationContext
+
+
+STAGE_ORDER = {
+	stage: index
+	for index, stage in enumerate(
+		[
+			DealStage.QUALIFICATION,
+			DealStage.PROPOSAL,
+			DealStage.NEGOTIATION,
+			DealStage.CLOSED,
+		]
+	)
+}
+
+
+class DealServiceError(Exception):
+	"""Base class for deal service errors."""
+
+
+class DealOrganizationMismatchError(DealServiceError):
+	"""Raised when attempting to use resources from another organization."""
+
+
+class DealStageTransitionError(DealServiceError):
+	"""Raised when stage transition violates business rules."""
+
+
+class DealStatusValidationError(DealServiceError):
+	"""Raised when invalid status transitions are requested."""
+
+
+class ContactHasDealsError(DealServiceError):
+	"""Raised when attempting to delete a contact with active deals."""
+
+
+@dataclass(slots=True)
+class DealUpdateData:
+	"""Structured container for deal update operations."""
+
+	status: DealStatus | None = None
+	stage: DealStage | None = None
+	amount: Decimal | None = None
+	currency: str | None = None
+
+
+class DealService:
+	"""Encapsulates deal workflows and validations."""
+
+	def __init__(self, repository: DealRepository) -> None:
+		self._repository = repository
+
+	async def create_deal(self, data: DealCreate, *, context: OrganizationContext) -> Deal:
+		self._ensure_same_organization(data.organization_id, context)
+		await self._ensure_contact_in_organization(data.contact_id, context.organization_id)
+		return await self._repository.create(data=data, role=context.role, user_id=context.user_id)
+
+	async def update_deal(
+		self,
+		deal: Deal,
+		updates: DealUpdateData,
+		*,
+		context: OrganizationContext,
+	) -> Deal:
+		self._ensure_same_organization(deal.organization_id, context)
+		changes: dict[str, object] = {}
+		stage_activity: tuple[ActivityType, dict[str, str]] | None = None
+		status_activity: tuple[ActivityType, dict[str, str]] | None = None
+
+		if updates.amount is not None:
+			changes["amount"] = updates.amount
+		if updates.currency is not None:
+			changes["currency"] = updates.currency
+
+		if updates.stage is not None and updates.stage != deal.stage:
+			self._validate_stage_transition(deal.stage, updates.stage, context.role)
+			changes["stage"] = updates.stage
+			stage_activity = (
+				ActivityType.STAGE_CHANGED,
+				{"old_stage": deal.stage, "new_stage": updates.stage},
+			)
+
+		if updates.status is not None and updates.status != deal.status:
+			self._validate_status_transition(deal, updates)
+			changes["status"] = updates.status
+			status_activity = (
+				ActivityType.STATUS_CHANGED,
+				{"old_status": deal.status, "new_status": updates.status},
+			)
+
+		if not changes:
+			return deal
+
+		updated = await self._repository.update(deal, changes, role=context.role, user_id=context.user_id)
+		await self._log_activities(
+			deal_id=deal.id,
+			author_id=context.user_id,
+			activities=[activity for activity in [stage_activity, status_activity] if activity],
+		)
+		return updated
+
+	async def ensure_contact_can_be_deleted(self, contact_id: int) -> None:
+		stmt = select(func.count()).select_from(Deal).where(Deal.contact_id == contact_id)
+		count = await self._repository.session.scalar(stmt)
+		if count and count > 0:
+			raise ContactHasDealsError("Contact has related deals and cannot be deleted")
+
+	async def _log_activities(
+		self,
+		*,
+		deal_id: int,
+		author_id: int,
+		activities: Iterable[tuple[ActivityType, dict[str, str]]],
+	) -> None:
+		entries = list(activities)
+		if not entries:
+			return
+		for activity_type, payload in entries:
+			activity = Activity(deal_id=deal_id, author_id=author_id, type=activity_type, payload=payload)
+			self._repository.session.add(activity)
+		await self._repository.session.flush()
+
+	def _ensure_same_organization(self, organization_id: int, context: OrganizationContext) -> None:
+		if organization_id != context.organization_id:
+			raise DealOrganizationMismatchError("Operation targets a different organization")
+
+	async def _ensure_contact_in_organization(self, contact_id: int, organization_id: int) -> Contact:
+		contact = await self._repository.session.get(Contact, contact_id)
+		if contact is None or contact.organization_id != organization_id:
+			raise DealOrganizationMismatchError("Contact belongs to another organization")
+		return contact
+
+	def _validate_stage_transition(
+		self,
+		current_stage: DealStage,
+		new_stage: DealStage,
+		role: OrganizationRole,
+	) -> None:
+		if STAGE_ORDER[new_stage] < STAGE_ORDER[current_stage] and role not in {
+			OrganizationRole.OWNER,
+			OrganizationRole.ADMIN,
+		}:
+			raise DealStageTransitionError("Stage rollback requires owner or admin role")
+
+	def _validate_status_transition(self, deal: Deal, updates: DealUpdateData) -> None:
+		if updates.status != DealStatus.WON:
+			return
+		effective_amount = updates.amount if updates.amount is not None else deal.amount
+		if effective_amount is None or Decimal(effective_amount) <= Decimal("0"):
+			raise DealStatusValidationError("Amount must be greater than zero to mark a deal as won")