TorbatYar/backend/services/loyalty/app/services/foundation.py

"""Loyalty foundation application services — Phase 7.0."""
from __future__ import annotations

from datetime import date, datetime, timezone
from enum import Enum
from typing import Any
from uuid import UUID, uuid4

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

from app.events.publisher import TransactionalEventPublisher
from app.events.types import LoyaltyEventType
from app.models.foundation import (
    Campaign,
    LoyaltyProgram,
    Member,
    MembershipLifecycleEvent,
    MembershipTier,
    PointAccount,
    Reward,
)
from app.models.types import AuditAction, MemberStatus, MembershipLifecycleAction, PointAccountStatus
from app.repositories.foundation import (
    CampaignRepository,
    LoyaltyProgramRepository,
    MemberRepository,
    MembershipLifecycleEventRepository,
    MembershipTierRepository,
    PointAccountRepository,
    RewardRepository,
)
from app.schemas.foundation import (
    CampaignCreate,
    CampaignUpdate,
    LoyaltyProgramCreate,
    LoyaltyProgramUpdate,
    MemberCreate,
    MemberEnrollRequest,
    MemberUpdate,
    MembershipTierCreate,
    MembershipTierUpdate,
    PointAccountCreate,
    PointAccountUpdate,
    RewardCreate,
    RewardUpdate,
)
from app.services.audit_service import AuditService
from app.validators import (
    ensure_optimistic_version,
    forbid_direct_balance_mutation,
    release_unique_token,
    validate_campaign_dates,
    validate_campaign_status,
    validate_code,
    validate_currency_code,
    validate_email,
    validate_member_status,
    validate_non_empty,
    validate_non_negative_int,
    validate_phone,
    validate_point_account_status,
    validate_program_status,
    validate_reward_status,
    validate_reward_type,
    validate_tier_status,
)
from app.validators.membership import (
    ACTIVE_MEMBER_STATUSES,
    DEFAULT_TERM_DAYS,
    compute_expiry,
    ensure_lifecycle_transition,
    target_status_for,
)
from shared.exceptions import AppError, NotFoundError
from shared.security import CurrentUser


def _apply_update(entity, data: dict) -> None:
    for key, value in data.items():
        setattr(entity, key, value)


def _reject_nulls(data: dict, fields: set[str]) -> None:
    for field in fields:
        if field in data and data[field] is None:
            raise AppError(
                f"{field} نمی‌تواند null باشد",
                status_code=422,
                error_code="null_not_allowed",
                details={"field": field},
            )


def _jsonable_changes(data: dict[str, Any]) -> dict[str, Any]:
    out: dict[str, Any] = {}
    for key, value in data.items():
        if isinstance(value, UUID):
            out[key] = str(value)
        elif isinstance(value, (datetime, date)):
            out[key] = value.isoformat()
        elif isinstance(value, Enum):
            out[key] = value.value
        else:
            out[key] = value
    return out


def _next_membership_number() -> str:
    return f"MBR-{uuid4().hex[:10].upper()}"


def _next_account_number() -> str:
    return f"PA-{uuid4().hex[:10].upper()}"


async def _clear_other_defaults(
    session: AsyncSession, tenant_id: UUID, keep_id: UUID | None = None
) -> None:
    from sqlalchemy import update

    stmt = (
        update(LoyaltyProgram)
        .where(
            LoyaltyProgram.tenant_id == tenant_id,
            LoyaltyProgram.is_default.is_(True),
            LoyaltyProgram.is_deleted.is_(False),
        )
        .values(is_default=False)
    )
    if keep_id is not None:
        stmt = stmt.where(LoyaltyProgram.id != keep_id)
    await session.execute(stmt)


def _map_integrity(*, error_code: str, message: str) -> AppError:
    return AppError(message, status_code=409, error_code=error_code)


class LoyaltyProgramService:
    def __init__(
        self,
        session: AsyncSession,
        publisher: TransactionalEventPublisher | None = None,
    ) -> None:
        self.session = session
        self.repo = LoyaltyProgramRepository(session)
        self.members = MemberRepository(session)
        self.audit = AuditService(session)
        self.publisher = publisher or TransactionalEventPublisher(session)

    async def create(
        self,
        tenant_id: UUID,
        body: LoyaltyProgramCreate,
        *,
        actor: CurrentUser | None = None,
    ) -> LoyaltyProgram:
        actor_id = actor.user_id if actor else None
        code = validate_code(body.code)
        name = validate_non_empty(body.name, field="name")
        currency = validate_currency_code(body.currency_code)
        status = validate_program_status(body.status)
        if await self.repo.get_by_code(tenant_id, code) is not None:
            raise AppError(
                "کد برنامه وفاداری تکراری است",
                status_code=409,
                error_code="program_code_exists",
            )
        if body.is_default:
            await _clear_other_defaults(self.session, tenant_id)
        entity = LoyaltyProgram(
            tenant_id=tenant_id,
            code=code,
            name=name,
            description=body.description,
            status=status,
            points_currency_name=validate_non_empty(
                body.points_currency_name, field="points_currency_name"
            ),
            currency_code=currency,
            is_default=body.is_default,
            settings=body.settings,
            created_by=actor_id,
            updated_by=actor_id,
        )
        try:
            await self.repo.add(entity)
            await self.audit.record(
                tenant_id=tenant_id,
                entity_type="loyalty_program",
                entity_id=entity.id,
                action=AuditAction.CREATE,
                actor_user_id=actor_id,
                message=f"Program {entity.code} created",
            )
            await self.publisher.publish(
                event_type=LoyaltyEventType.PROGRAM_CREATED,
                aggregate_type="loyalty_program",
                aggregate_id=entity.id,
                tenant_id=tenant_id,
                payload={"code": entity.code, "status": entity.status.value},
            )
            await self.session.commit()
            await self.session.refresh(entity)
        except IntegrityError:
            await self.session.rollback()
            raise _map_integrity(
                error_code="program_code_exists",
                message="کد برنامه وفاداری تکراری است",
            )
        return entity

    async def update(
        self,
        tenant_id: UUID,
        program_id: UUID,
        body: LoyaltyProgramUpdate,
        *,
        actor: CurrentUser | None = None,
    ) -> LoyaltyProgram:
        entity = await self.get(tenant_id, program_id)
        ensure_optimistic_version(entity.version, body.version)
        data = body.model_dump(exclude_unset=True, exclude={"version"})
        _reject_nulls(
            data,
            {"name", "status", "points_currency_name", "currency_code", "is_default"},
        )
        if "name" in data:
            data["name"] = validate_non_empty(data["name"], field="name")
        if "currency_code" in data:
            data["currency_code"] = validate_currency_code(data["currency_code"])
        if "status" in data:
            data["status"] = validate_program_status(data["status"])
        if "points_currency_name" in data:
            data["points_currency_name"] = validate_non_empty(
                data["points_currency_name"], field="points_currency_name"
            )
        if data.get("is_default") is True:
            await _clear_other_defaults(self.session, tenant_id, keep_id=entity.id)
        _apply_update(entity, data)
        entity.version += 1
        entity.updated_by = actor.user_id if actor else None
        await self.audit.record(
            tenant_id=tenant_id,
            entity_type="loyalty_program",
            entity_id=entity.id,
            action=AuditAction.UPDATE,
            actor_user_id=entity.updated_by,
            changes=_jsonable_changes(data),
        )
        await self.publisher.publish(
            event_type=LoyaltyEventType.PROGRAM_UPDATED,
            aggregate_type="loyalty_program",
            aggregate_id=entity.id,
            tenant_id=tenant_id,
            payload={"code": entity.code, "status": entity.status.value},
        )
        await self.session.commit()
        await self.session.refresh(entity)
        return entity

    async def get(self, tenant_id: UUID, program_id: UUID) -> LoyaltyProgram:
        entity = await self.repo.get(tenant_id, program_id)
        if entity is None:
            raise NotFoundError("برنامه وفاداری یافت نشد", error_code="program_not_found")
        return entity

    async def list(self, tenant_id: UUID, *, offset: int, limit: int):
        return await self.repo.list_by_tenant(tenant_id, offset=offset, limit=limit)

    async def soft_delete(
        self,
        tenant_id: UUID,
        program_id: UUID,
        *,
        actor: CurrentUser | None = None,
    ) -> LoyaltyProgram:
        entity = await self.get(tenant_id, program_id)
        blocking = await self.members.count_blocking_for_program(
            tenant_id, program_id, list(ACTIVE_MEMBER_STATUSES)
        )
        if blocking:
            raise AppError(
                "برنامه دارای اعضای فعال است و قابل حذف نیست",
                status_code=409,
                error_code="program_has_active_members",
                details={"active_member_count": blocking},
            )
        entity.code = release_unique_token(entity.code, entity.id)
        await self.repo.soft_delete(
            entity, deleted_by=actor.user_id if actor else None
        )
        await self.audit.record(
            tenant_id=tenant_id,
            entity_type="loyalty_program",
            entity_id=entity.id,
            action=AuditAction.DELETE,
            actor_user_id=actor.user_id if actor else None,
        )
        await self.publisher.publish(
            event_type=LoyaltyEventType.PROGRAM_DELETED,
            aggregate_type="loyalty_program",
            aggregate_id=entity.id,
            tenant_id=tenant_id,
            payload={"code": entity.code},
        )
        await self.session.commit()
        await self.session.refresh(entity)
        return entity


class MembershipTierService:
    def __init__(
        self,
        session: AsyncSession,
        publisher: TransactionalEventPublisher | None = None,
    ) -> None:
        self.session = session
        self.repo = MembershipTierRepository(session)
        self.programs = LoyaltyProgramRepository(session)
        self.audit = AuditService(session)
        self.publisher = publisher or TransactionalEventPublisher(session)

    async def create(
        self,
        tenant_id: UUID,
        body: MembershipTierCreate,
        *,
        actor: CurrentUser | None = None,
    ) -> MembershipTier:
        actor_id = actor.user_id if actor else None
        if await self.programs.get(tenant_id, body.program_id) is None:
            raise NotFoundError("برنامه وفاداری یافت نشد", error_code="program_not_found")
        code = validate_code(body.code)
        name = validate_non_empty(body.name, field="name")
        status = validate_tier_status(body.status)
        min_points = validate_non_negative_int(body.min_points, field="min_points")
        if await self.repo.get_by_code(tenant_id, body.program_id, code) is not None:
            raise AppError(
                "کد سطح عضویت تکراری است",
                status_code=409,
                error_code="tier_code_exists",
            )
        entity = MembershipTier(
            tenant_id=tenant_id,
            program_id=body.program_id,
            code=code,
            name=name,
            description=body.description,
            rank=body.rank,
            min_points=min_points,
            status=status,
            benefits=body.benefits,
            rules=body.rules,
            created_by=actor_id,
            updated_by=actor_id,
        )
        try:
            await self.repo.add(entity)
            await self.audit.record(
                tenant_id=tenant_id,
                entity_type="membership_tier",
                entity_id=entity.id,
                action=AuditAction.CREATE,
                actor_user_id=actor_id,
                message=f"Tier {entity.code} created",
            )
            await self.publisher.publish(
                event_type=LoyaltyEventType.TIER_CREATED,
                aggregate_type="membership_tier",
                aggregate_id=entity.id,
                tenant_id=tenant_id,
                payload={"code": entity.code, "program_id": str(entity.program_id)},
            )
            await self.session.commit()
            await self.session.refresh(entity)
        except IntegrityError:
            await self.session.rollback()
            raise _map_integrity(
                error_code="tier_code_exists",
                message="کد سطح عضویت تکراری است",
            )
        return entity

    async def update(
        self,
        tenant_id: UUID,
        tier_id: UUID,
        body: MembershipTierUpdate,
        *,
        actor: CurrentUser | None = None,
    ) -> MembershipTier:
        entity = await self.get(tenant_id, tier_id)
        data = body.model_dump(exclude_unset=True)
        _reject_nulls(data, {"name", "rank", "min_points", "status"})
        if "name" in data:
            data["name"] = validate_non_empty(data["name"], field="name")
        if "status" in data:
            data["status"] = validate_tier_status(data["status"])
        if "min_points" in data:
            data["min_points"] = validate_non_negative_int(
                data["min_points"], field="min_points"
            )
        if "rank" in data:
            data["rank"] = validate_non_negative_int(data["rank"], field="rank")
        _apply_update(entity, data)
        entity.updated_by = actor.user_id if actor else None
        await self.audit.record(
            tenant_id=tenant_id,
            entity_type="membership_tier",
            entity_id=entity.id,
            action=AuditAction.UPDATE,
            actor_user_id=entity.updated_by,
            changes=_jsonable_changes(data),
        )
        await self.publisher.publish(
            event_type=LoyaltyEventType.TIER_UPDATED,
            aggregate_type="membership_tier",
            aggregate_id=entity.id,
            tenant_id=tenant_id,
            payload={"code": entity.code},
        )
        await self.session.commit()
        await self.session.refresh(entity)
        return entity

    async def get(self, tenant_id: UUID, tier_id: UUID) -> MembershipTier:
        entity = await self.repo.get(tenant_id, tier_id)
        if entity is None:
            raise NotFoundError("سطح عضویت یافت نشد", error_code="tier_not_found")
        return entity

    async def list(self, tenant_id: UUID, *, offset: int, limit: int):
        return await self.repo.list_by_tenant(tenant_id, offset=offset, limit=limit)

    async def list_by_program(
        self, tenant_id: UUID, program_id: UUID, *, offset: int, limit: int
    ):
        return await self.repo.list_by_program(
            tenant_id, program_id, offset=offset, limit=limit
        )

    async def soft_delete(
        self,
        tenant_id: UUID,
        tier_id: UUID,
        *,
        actor: CurrentUser | None = None,
    ) -> MembershipTier:
        entity = await self.get(tenant_id, tier_id)
        entity.code = release_unique_token(entity.code, entity.id)
        await self.repo.soft_delete(
            entity, deleted_by=actor.user_id if actor else None
        )
        await self.audit.record(
            tenant_id=tenant_id,
            entity_type="membership_tier",
            entity_id=entity.id,
            action=AuditAction.DELETE,
            actor_user_id=actor.user_id if actor else None,
        )
        await self.publisher.publish(
            event_type=LoyaltyEventType.TIER_DELETED,
            aggregate_type="membership_tier",
            aggregate_id=entity.id,
            tenant_id=tenant_id,
            payload={"code": entity.code},
        )
        await self.session.commit()
        await self.session.refresh(entity)
        return entity


class MemberService:
    def __init__(
        self,
        session: AsyncSession,
        publisher: TransactionalEventPublisher | None = None,
    ) -> None:
        self.session = session
        self.repo = MemberRepository(session)
        self.programs = LoyaltyProgramRepository(session)
        self.tiers = MembershipTierRepository(session)
        self.audit = AuditService(session)
        self.publisher = publisher or TransactionalEventPublisher(session)

    async def create(
        self,
        tenant_id: UUID,
        body: MemberCreate,
        *,
        actor: CurrentUser | None = None,
    ) -> Member:
        actor_id = actor.user_id if actor else None
        if await self.programs.get(tenant_id, body.program_id) is None:
            raise NotFoundError("برنامه وفاداری یافت نشد", error_code="program_not_found")
        if body.tier_id is not None:
            tier = await self.tiers.get(tenant_id, body.tier_id)
            if tier is None or tier.program_id != body.program_id:
                raise NotFoundError("سطح عضویت یافت نشد", error_code="tier_not_found")
        display_name = validate_non_empty(body.display_name, field="display_name")
        email = validate_email(body.email)
        mobile = validate_phone(body.mobile, field="mobile")
        status = validate_member_status(body.status)
        membership_number = (
            validate_code(body.membership_number, field="membership_number")
            if body.membership_number
            else _next_membership_number()
        )
        if await self.repo.get_by_membership_number(tenant_id, membership_number):
            raise AppError(
                "شماره عضویت تکراری است",
                status_code=409,
                error_code="membership_number_exists",
            )
        if body.external_customer_ref:
            existing_ref = await self.repo.get_by_external_ref(
                tenant_id, body.program_id, body.external_customer_ref
            )
            if existing_ref is not None:
                raise AppError(
                    "شناسه خارجی مشتری تکراری است",
                    status_code=409,
                    error_code="external_customer_ref_exists",
                )
        enrolled_at = None
        activated_at = None
        membership_started_at = None
        membership_expires_at = None
        if body.enroll:
            status = MemberStatus.ACTIVE
            enrolled_at = datetime.now(timezone.utc)
            activated_at = enrolled_at
            membership_started_at = enrolled_at
            membership_expires_at = compute_expiry(from_when=enrolled_at)
        entity = Member(
            tenant_id=tenant_id,
            program_id=body.program_id,
            membership_number=membership_number,
            external_customer_ref=body.external_customer_ref,
            display_name=display_name,
            email=email,
            mobile=mobile,
            status=status,
            tier_id=body.tier_id,
            enrolled_at=enrolled_at,
            activated_at=activated_at,
            membership_started_at=membership_started_at,
            membership_expires_at=membership_expires_at,
            profile=body.profile,
            created_by=actor_id,
            updated_by=actor_id,
        )
        try:
            await self.repo.add(entity)
            await self.audit.record(
                tenant_id=tenant_id,
                entity_type="member",
                entity_id=entity.id,
                action=AuditAction.CREATE,
                actor_user_id=actor_id,
                message=f"Member {entity.membership_number} created",
            )
            await self.publisher.publish(
                event_type=LoyaltyEventType.MEMBER_CREATED,
                aggregate_type="member",
                aggregate_id=entity.id,
                tenant_id=tenant_id,
                payload={
                    "membership_number": entity.membership_number,
                    "status": entity.status.value,
                },
            )
            if body.enroll:
                self.lifecycle = MembershipLifecycleEventRepository(self.session)
                await self.lifecycle.add(
                    MembershipLifecycleEvent(
                        tenant_id=tenant_id,
                        member_id=entity.id,
                        action=MembershipLifecycleAction.ENROLL,
                        from_status=MemberStatus.PENDING.value,
                        to_status=MemberStatus.ACTIVE.value,
                        reason=None,
                        metadata_json={"term_days": DEFAULT_TERM_DAYS},
                        actor_user_id=actor_id,
                        created_at=enrolled_at,
                    )
                )
                await self.publisher.publish(
                    event_type=LoyaltyEventType.MEMBER_ENROLLED,
                    aggregate_type="member",
                    aggregate_id=entity.id,
                    tenant_id=tenant_id,
                    payload={"membership_number": entity.membership_number},
                )
                await self.publisher.publish(
                    event_type=LoyaltyEventType.MEMBER_ACTIVATED,
                    aggregate_type="member",
                    aggregate_id=entity.id,
                    tenant_id=tenant_id,
                    payload={
                        "membership_number": entity.membership_number,
                        "status": entity.status.value,
                    },
                )
            await self.session.commit()
            await self.session.refresh(entity)
        except IntegrityError:
            await self.session.rollback()
            raise _map_integrity(
                error_code="member_conflict",
                message="عضو با این شناسه‌ها قابل ایجاد نیست",
            )
        return entity

    async def update(
        self,
        tenant_id: UUID,
        member_id: UUID,
        body: MemberUpdate,
        *,
        actor: CurrentUser | None = None,
    ) -> Member:
        entity = await self.get(tenant_id, member_id)
        ensure_optimistic_version(entity.version, body.version)
        data = body.model_dump(exclude_unset=True, exclude={"version"})
        _reject_nulls(data, {"display_name", "status"})
        if "display_name" in data:
            data["display_name"] = validate_non_empty(
                data["display_name"], field="display_name"
            )
        if "email" in data:
            data["email"] = validate_email(data["email"])
        if "mobile" in data:
            data["mobile"] = validate_phone(data["mobile"], field="mobile")
        if "status" in data:
            data["status"] = validate_member_status(data["status"])
        if "tier_id" in data and data["tier_id"] is not None:
            tier = await self.tiers.get(tenant_id, data["tier_id"])
            if tier is None or tier.program_id != entity.program_id:
                raise NotFoundError("سطح عضویت یافت نشد", error_code="tier_not_found")
        _apply_update(entity, data)
        entity.version += 1
        entity.updated_by = actor.user_id if actor else None
        await self.audit.record(
            tenant_id=tenant_id,
            entity_type="member",
            entity_id=entity.id,
            action=AuditAction.UPDATE,
            actor_user_id=entity.updated_by,
            changes=_jsonable_changes(data),
        )
        await self.publisher.publish(
            event_type=LoyaltyEventType.MEMBER_UPDATED,
            aggregate_type="member",
            aggregate_id=entity.id,
            tenant_id=tenant_id,
            payload={"status": entity.status.value},
        )
        await self.session.commit()
        await self.session.refresh(entity)
        return entity

    async def enroll(
        self,
        tenant_id: UUID,
        member_id: UUID,
        body: MemberEnrollRequest,
        *,
        actor: CurrentUser | None = None,
    ) -> Member:
        entity = await self.get(tenant_id, member_id)
        action = MembershipLifecycleAction.ENROLL
        ensure_lifecycle_transition(action=action, current=entity.status)
        if body.tier_id is not None:
            tier = await self.tiers.get(tenant_id, body.tier_id)
            if tier is None or tier.program_id != entity.program_id:
                raise NotFoundError("سطح عضویت یافت نشد", error_code="tier_not_found")
            entity.tier_id = body.tier_id
        from_status = entity.status
        to_status = target_status_for(action)
        now = datetime.now(timezone.utc)
        entity.status = to_status
        entity.enrolled_at = entity.enrolled_at or now
        entity.activated_at = entity.activated_at or now
        entity.membership_started_at = entity.membership_started_at or now
        entity.membership_expires_at = compute_expiry(
            from_when=now,
            term_days=body.term_days or DEFAULT_TERM_DAYS,
            explicit=body.expires_at,
        )
        entity.version += 1
        entity.updated_by = actor.user_id if actor else None
        await MembershipLifecycleEventRepository(self.session).add(
            MembershipLifecycleEvent(
                tenant_id=tenant_id,
                member_id=entity.id,
                action=action,
                from_status=from_status.value,
                to_status=to_status.value,
                reason=body.reason,
                metadata_json={"term_days": body.term_days or DEFAULT_TERM_DAYS},
                actor_user_id=entity.updated_by,
                created_at=now,
            )
        )
        await self.audit.record(
            tenant_id=tenant_id,
            entity_type="member",
            entity_id=entity.id,
            action=AuditAction.ENROLL,
            actor_user_id=entity.updated_by,
            message=f"Member {entity.membership_number} enrolled",
        )
        await self.publisher.publish(
            event_type=LoyaltyEventType.MEMBER_ENROLLED,
            aggregate_type="member",
            aggregate_id=entity.id,
            tenant_id=tenant_id,
            payload={"membership_number": entity.membership_number},
        )
        await self.publisher.publish(
            event_type=LoyaltyEventType.MEMBER_ACTIVATED,
            aggregate_type="member",
            aggregate_id=entity.id,
            tenant_id=tenant_id,
            payload={
                "membership_number": entity.membership_number,
                "status": entity.status.value,
            },
        )
        await self.session.commit()
        await self.session.refresh(entity)
        return entity

    async def get(self, tenant_id: UUID, member_id: UUID) -> Member:
        entity = await self.repo.get(tenant_id, member_id)
        if entity is None:
            raise NotFoundError("عضو یافت نشد", error_code="member_not_found")
        return entity

    async def list(self, tenant_id: UUID, *, offset: int, limit: int):
        return await self.repo.list_by_tenant(tenant_id, offset=offset, limit=limit)

    async def soft_delete(
        self,
        tenant_id: UUID,
        member_id: UUID,
        *,
        actor: CurrentUser | None = None,
    ) -> Member:
        entity = await self.get(tenant_id, member_id)
        entity.membership_number = release_unique_token(
            entity.membership_number, entity.id
        )
        if entity.external_customer_ref:
            entity.external_customer_ref = release_unique_token(
                entity.external_customer_ref, entity.id
            )
        await self.repo.soft_delete(
            entity, deleted_by=actor.user_id if actor else None
        )
        await self.audit.record(
            tenant_id=tenant_id,
            entity_type="member",
            entity_id=entity.id,
            action=AuditAction.DELETE,
            actor_user_id=actor.user_id if actor else None,
        )
        await self.publisher.publish(
            event_type=LoyaltyEventType.MEMBER_DELETED,
            aggregate_type="member",
            aggregate_id=entity.id,
            tenant_id=tenant_id,
            payload={"membership_number": entity.membership_number},
        )
        await self.session.commit()
        await self.session.refresh(entity)
        return entity


class PointAccountService:
    def __init__(
        self,
        session: AsyncSession,
        publisher: TransactionalEventPublisher | None = None,
    ) -> None:
        self.session = session
        self.repo = PointAccountRepository(session)
        self.members = MemberRepository(session)
        self.programs = LoyaltyProgramRepository(session)
        self.audit = AuditService(session)
        self.publisher = publisher or TransactionalEventPublisher(session)

    async def create(
        self,
        tenant_id: UUID,
        body: PointAccountCreate,
        *,
        actor: CurrentUser | None = None,
    ) -> PointAccount:
        forbid_direct_balance_mutation(body.model_dump())
        actor_id = actor.user_id if actor else None
        program = await self.programs.get(tenant_id, body.program_id)
        if program is None:
            raise NotFoundError("برنامه وفاداری یافت نشد", error_code="program_not_found")
        member = await self.members.get(tenant_id, body.member_id)
        if member is None or member.program_id != body.program_id:
            raise NotFoundError("عضو یافت نشد", error_code="member_not_found")
        if await self.repo.get_by_member(tenant_id, body.member_id) is not None:
            raise AppError(
                "حساب امتیاز برای این عضو وجود دارد",
                status_code=409,
                error_code="point_account_exists",
            )
        account_number = (
            validate_code(body.account_number, field="account_number")
            if body.account_number
            else _next_account_number()
        )
        if await self.repo.get_by_account_number(tenant_id, account_number):
            raise AppError(
                "شماره حساب امتیاز تکراری است",
                status_code=409,
                error_code="account_number_exists",
            )
        entity = PointAccount(
            tenant_id=tenant_id,
            program_id=body.program_id,
            member_id=body.member_id,
            account_number=account_number,
            status=validate_point_account_status(body.status),
            currency_name=validate_non_empty(
                body.currency_name, field="currency_name"
            ),
            created_by=actor_id,
            updated_by=actor_id,
        )
        try:
            await self.repo.add(entity)
            await self.audit.record(
                tenant_id=tenant_id,
                entity_type="point_account",
                entity_id=entity.id,
                action=AuditAction.CREATE,
                actor_user_id=actor_id,
                message=f"Point account {entity.account_number} opened",
            )
            await self.publisher.publish(
                event_type=LoyaltyEventType.POINT_ACCOUNT_OPENED,
                aggregate_type="point_account",
                aggregate_id=entity.id,
                tenant_id=tenant_id,
                payload={
                    "account_number": entity.account_number,
                    "member_id": str(entity.member_id),
                },
            )
            await self.session.commit()
            await self.session.refresh(entity)
        except IntegrityError:
            await self.session.rollback()
            raise _map_integrity(
                error_code="point_account_conflict",
                message="حساب امتیاز قابل ایجاد نیست",
            )
        return entity

    async def update(
        self,
        tenant_id: UUID,
        account_id: UUID,
        body: PointAccountUpdate,
        *,
        actor: CurrentUser | None = None,
    ) -> PointAccount:
        forbid_direct_balance_mutation(body.model_dump())
        entity = await self.get(tenant_id, account_id)
        ensure_optimistic_version(entity.version, body.version)
        data = body.model_dump(exclude_unset=True, exclude={"version"})
        _reject_nulls(data, {"status", "currency_name"})
        if "status" in data:
            data["status"] = validate_point_account_status(data["status"])
        if "currency_name" in data:
            data["currency_name"] = validate_non_empty(
                data["currency_name"], field="currency_name"
            )
        _apply_update(entity, data)
        entity.version += 1
        entity.updated_by = actor.user_id if actor else None
        await self.audit.record(
            tenant_id=tenant_id,
            entity_type="point_account",
            entity_id=entity.id,
            action=AuditAction.UPDATE,
            actor_user_id=entity.updated_by,
            changes=_jsonable_changes(data),
        )
        await self.publisher.publish(
            event_type=LoyaltyEventType.POINT_ACCOUNT_UPDATED,
            aggregate_type="point_account",
            aggregate_id=entity.id,
            tenant_id=tenant_id,
            payload={"status": entity.status.value},
        )
        await self.session.commit()
        await self.session.refresh(entity)
        return entity

    async def get(self, tenant_id: UUID, account_id: UUID) -> PointAccount:
        entity = await self.repo.get(tenant_id, account_id)
        if entity is None:
            raise NotFoundError(
                "حساب امتیاز یافت نشد", error_code="point_account_not_found"
            )
        return entity

    async def list(self, tenant_id: UUID, *, offset: int, limit: int):
        return await self.repo.list_by_tenant(tenant_id, offset=offset, limit=limit)

    async def soft_delete(
        self,
        tenant_id: UUID,
        account_id: UUID,
        *,
        actor: CurrentUser | None = None,
    ) -> PointAccount:
        entity = await self.get(tenant_id, account_id)
        entity.account_number = release_unique_token(entity.account_number, entity.id)
        entity.status = PointAccountStatus.CLOSED
        await self.repo.soft_delete(
            entity, deleted_by=actor.user_id if actor else None
        )
        await self.audit.record(
            tenant_id=tenant_id,
            entity_type="point_account",
            entity_id=entity.id,
            action=AuditAction.DELETE,
            actor_user_id=actor.user_id if actor else None,
        )
        await self.publisher.publish(
            event_type=LoyaltyEventType.POINT_ACCOUNT_DELETED,
            aggregate_type="point_account",
            aggregate_id=entity.id,
            tenant_id=tenant_id,
            payload={"account_number": entity.account_number},
        )
        await self.session.commit()
        await self.session.refresh(entity)
        return entity


class RewardService:
    def __init__(
        self,
        session: AsyncSession,
        publisher: TransactionalEventPublisher | None = None,
    ) -> None:
        self.session = session
        self.repo = RewardRepository(session)
        self.programs = LoyaltyProgramRepository(session)
        self.audit = AuditService(session)
        self.publisher = publisher or TransactionalEventPublisher(session)

    async def create(
        self,
        tenant_id: UUID,
        body: RewardCreate,
        *,
        actor: CurrentUser | None = None,
    ) -> Reward:
        actor_id = actor.user_id if actor else None
        if await self.programs.get(tenant_id, body.program_id) is None:
            raise NotFoundError("برنامه وفاداری یافت نشد", error_code="program_not_found")
        code = validate_code(body.code)
        name = validate_non_empty(body.name, field="name")
        if await self.repo.get_by_code(tenant_id, body.program_id, code) is not None:
            raise AppError(
                "کد پاداش تکراری است",
                status_code=409,
                error_code="reward_code_exists",
            )
        entity = Reward(
            tenant_id=tenant_id,
            program_id=body.program_id,
            code=code,
            name=name,
            description=body.description,
            reward_type=validate_reward_type(body.reward_type),
            status=validate_reward_status(body.status),
            points_cost=validate_non_negative_int(body.points_cost, field="points_cost"),
            eligibility_rules=body.eligibility_rules,
            metadata_json=body.metadata_json,
            created_by=actor_id,
            updated_by=actor_id,
        )
        try:
            await self.repo.add(entity)
            await self.audit.record(
                tenant_id=tenant_id,
                entity_type="reward",
                entity_id=entity.id,
                action=AuditAction.CREATE,
                actor_user_id=actor_id,
                message=f"Reward {entity.code} created",
            )
            await self.publisher.publish(
                event_type=LoyaltyEventType.REWARD_CREATED,
                aggregate_type="reward",
                aggregate_id=entity.id,
                tenant_id=tenant_id,
                payload={"code": entity.code, "reward_type": entity.reward_type.value},
            )
            await self.session.commit()
            await self.session.refresh(entity)
        except IntegrityError:
            await self.session.rollback()
            raise _map_integrity(
                error_code="reward_code_exists",
                message="کد پاداش تکراری است",
            )
        return entity

    async def update(
        self,
        tenant_id: UUID,
        reward_id: UUID,
        body: RewardUpdate,
        *,
        actor: CurrentUser | None = None,
    ) -> Reward:
        entity = await self.get(tenant_id, reward_id)
        data = body.model_dump(exclude_unset=True)
        _reject_nulls(data, {"name", "reward_type", "status", "points_cost"})
        if "name" in data:
            data["name"] = validate_non_empty(data["name"], field="name")
        if "reward_type" in data:
            data["reward_type"] = validate_reward_type(data["reward_type"])
        if "status" in data:
            data["status"] = validate_reward_status(data["status"])
        if "points_cost" in data:
            data["points_cost"] = validate_non_negative_int(
                data["points_cost"], field="points_cost"
            )
        _apply_update(entity, data)
        entity.updated_by = actor.user_id if actor else None
        await self.audit.record(
            tenant_id=tenant_id,
            entity_type="reward",
            entity_id=entity.id,
            action=AuditAction.UPDATE,
            actor_user_id=entity.updated_by,
            changes=_jsonable_changes(data),
        )
        await self.publisher.publish(
            event_type=LoyaltyEventType.REWARD_UPDATED,
            aggregate_type="reward",
            aggregate_id=entity.id,
            tenant_id=tenant_id,
            payload={"code": entity.code, "status": entity.status.value},
        )
        await self.session.commit()
        await self.session.refresh(entity)
        return entity

    async def get(self, tenant_id: UUID, reward_id: UUID) -> Reward:
        entity = await self.repo.get(tenant_id, reward_id)
        if entity is None:
            raise NotFoundError("پاداش یافت نشد", error_code="reward_not_found")
        return entity

    async def list(self, tenant_id: UUID, *, offset: int, limit: int):
        return await self.repo.list_by_tenant(tenant_id, offset=offset, limit=limit)

    async def soft_delete(
        self,
        tenant_id: UUID,
        reward_id: UUID,
        *,
        actor: CurrentUser | None = None,
    ) -> Reward:
        entity = await self.get(tenant_id, reward_id)
        entity.code = release_unique_token(entity.code, entity.id)
        await self.repo.soft_delete(
            entity, deleted_by=actor.user_id if actor else None
        )
        await self.audit.record(
            tenant_id=tenant_id,
            entity_type="reward",
            entity_id=entity.id,
            action=AuditAction.DELETE,
            actor_user_id=actor.user_id if actor else None,
        )
        await self.publisher.publish(
            event_type=LoyaltyEventType.REWARD_DELETED,
            aggregate_type="reward",
            aggregate_id=entity.id,
            tenant_id=tenant_id,
            payload={"code": entity.code},
        )
        await self.session.commit()
        await self.session.refresh(entity)
        return entity


class CampaignService:
    def __init__(
        self,
        session: AsyncSession,
        publisher: TransactionalEventPublisher | None = None,
    ) -> None:
        self.session = session
        self.repo = CampaignRepository(session)
        self.programs = LoyaltyProgramRepository(session)
        self.audit = AuditService(session)
        self.publisher = publisher or TransactionalEventPublisher(session)

    async def create(
        self,
        tenant_id: UUID,
        body: CampaignCreate,
        *,
        actor: CurrentUser | None = None,
    ) -> Campaign:
        actor_id = actor.user_id if actor else None
        if await self.programs.get(tenant_id, body.program_id) is None:
            raise NotFoundError("برنامه وفاداری یافت نشد", error_code="program_not_found")
        code = validate_code(body.code)
        name = validate_non_empty(body.name, field="name")
        starts_on, ends_on = validate_campaign_dates(body.starts_on, body.ends_on)
        if await self.repo.get_by_code(tenant_id, body.program_id, code) is not None:
            raise AppError(
                "کد کمپین تکراری است",
                status_code=409,
                error_code="campaign_code_exists",
            )
        entity = Campaign(
            tenant_id=tenant_id,
            program_id=body.program_id,
            code=code,
            name=name,
            description=body.description,
            status=validate_campaign_status(body.status),
            rule_version=1,
            rules=body.rules,
            starts_on=starts_on,
            ends_on=ends_on,
            created_by=actor_id,
            updated_by=actor_id,
        )
        try:
            await self.repo.add(entity)
            await self.audit.record(
                tenant_id=tenant_id,
                entity_type="campaign",
                entity_id=entity.id,
                action=AuditAction.CREATE,
                actor_user_id=actor_id,
                message=f"Campaign {entity.code} created",
            )
            await self.publisher.publish(
                event_type=LoyaltyEventType.CAMPAIGN_CREATED,
                aggregate_type="campaign",
                aggregate_id=entity.id,
                tenant_id=tenant_id,
                payload={"code": entity.code, "rule_version": entity.rule_version},
            )
            await self.session.commit()
            await self.session.refresh(entity)
        except IntegrityError:
            await self.session.rollback()
            raise _map_integrity(
                error_code="campaign_code_exists",
                message="کد کمپین تکراری است",
            )
        return entity

    async def update(
        self,
        tenant_id: UUID,
        campaign_id: UUID,
        body: CampaignUpdate,
        *,
        actor: CurrentUser | None = None,
    ) -> Campaign:
        entity = await self.get(tenant_id, campaign_id)
        data = body.model_dump(exclude_unset=True, exclude={"bump_rule_version"})
        _reject_nulls(data, {"name", "status"})
        if "name" in data:
            data["name"] = validate_non_empty(data["name"], field="name")
        if "status" in data:
            data["status"] = validate_campaign_status(data["status"])
        starts = data.get("starts_on", entity.starts_on)
        ends = data.get("ends_on", entity.ends_on)
        if "starts_on" in data or "ends_on" in data:
            starts, ends = validate_campaign_dates(starts, ends)
            data["starts_on"] = starts
            data["ends_on"] = ends
        _apply_update(entity, data)
        if body.bump_rule_version or ("rules" in data):
            entity.rule_version += 1
        entity.updated_by = actor.user_id if actor else None
        await self.audit.record(
            tenant_id=tenant_id,
            entity_type="campaign",
            entity_id=entity.id,
            action=AuditAction.UPDATE,
            actor_user_id=entity.updated_by,
            changes=_jsonable_changes({**data, "rule_version": entity.rule_version}),
        )
        await self.publisher.publish(
            event_type=LoyaltyEventType.CAMPAIGN_UPDATED,
            aggregate_type="campaign",
            aggregate_id=entity.id,
            tenant_id=tenant_id,
            payload={"code": entity.code, "rule_version": entity.rule_version},
        )
        await self.session.commit()
        await self.session.refresh(entity)
        return entity

    async def get(self, tenant_id: UUID, campaign_id: UUID) -> Campaign:
        entity = await self.repo.get(tenant_id, campaign_id)
        if entity is None:
            raise NotFoundError("کمپین یافت نشد", error_code="campaign_not_found")
        return entity

    async def list(self, tenant_id: UUID, *, offset: int, limit: int):
        return await self.repo.list_by_tenant(tenant_id, offset=offset, limit=limit)

    async def soft_delete(
        self,
        tenant_id: UUID,
        campaign_id: UUID,
        *,
        actor: CurrentUser | None = None,
    ) -> Campaign:
        entity = await self.get(tenant_id, campaign_id)
        entity.code = release_unique_token(entity.code, entity.id)
        await self.repo.soft_delete(
            entity, deleted_by=actor.user_id if actor else None
        )
        await self.audit.record(
            tenant_id=tenant_id,
            entity_type="campaign",
            entity_id=entity.id,
            action=AuditAction.DELETE,
            actor_user_id=actor.user_id if actor else None,
        )
        await self.publisher.publish(
            event_type=LoyaltyEventType.CAMPAIGN_DELETED,
            aggregate_type="campaign",
            aggregate_id=entity.id,
            tenant_id=tenant_id,
            payload={"code": entity.code},
        )
        await self.session.commit()
        await self.session.refresh(entity)
        return entity