"""Wallet Engine — immutable ledger writes and account lifecycle (Phase 7.6). Mirrors the Point Engine (Phase 7.2): WalletAccount is a shell only; balance is always SUM(signed WalletLedgerEntry.amount). Never add a mutable balance column on WalletAccount. """ from __future__ import annotations from datetime import datetime, timezone 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.types import AuditAction, WalletAccountStatus, WalletLedgerEntryType from app.models.wallet import WalletAccount, WalletLedgerEntry from app.repositories.foundation import LoyaltyProgramRepository, MemberRepository from app.repositories.wallet import WalletAccountRepository, WalletLedgerRepository from app.schemas.wallet import ( WalletAccountCreate, WalletAccountUpdate, WalletAdjustRequest, WalletCreditRequest, WalletDebitRequest, WalletTransferRequest, ) from app.services.audit_service import AuditService from app.validators.foundation import ( release_unique_token, validate_code, validate_currency_code, ) from app.validators.wallet import ( ensure_not_same_account, ensure_same_currency, ensure_wallet_open, signed_amount, validate_adjust_amount, validate_positive_amount, validate_wallet_account_status, ) from shared.exceptions import AppError, NotFoundError from shared.security import CurrentUser def _now() -> datetime: return datetime.now(timezone.utc) def _next_wallet_account_number() -> str: return f"WA-{uuid4().hex[:10].upper()}" def _map_integrity(*, error_code: str, message: str) -> AppError: return AppError(message, status_code=409, error_code=error_code) class WalletEngineService: def __init__( self, session: AsyncSession, publisher: TransactionalEventPublisher | None = None, ) -> None: self.session = session self.accounts = WalletAccountRepository(session) self.ledger = WalletLedgerRepository(session) self.programs = LoyaltyProgramRepository(session) self.members = MemberRepository(session) self.audit = AuditService(session) self.publisher = publisher or TransactionalEventPublisher(session) # ------------------------------------------------------------------ # Account lifecycle # ------------------------------------------------------------------ async def create( self, tenant_id: UUID, body: WalletAccountCreate, *, actor: CurrentUser | None = None, ) -> WalletAccount: 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.accounts.get_by_member_program( tenant_id, body.program_id, body.member_id ) is not None ): raise AppError( "کیف پول برای این عضو در این برنامه وجود دارد", status_code=409, error_code="wallet_account_exists", ) account_number = ( validate_code(body.account_number, field="account_number") if body.account_number else _next_wallet_account_number() ) if await self.accounts.get_by_account_number(tenant_id, account_number): raise AppError( "شماره کیف پول تکراری است", status_code=409, error_code="wallet_account_number_exists", ) entity = WalletAccount( tenant_id=tenant_id, program_id=body.program_id, member_id=body.member_id, account_number=account_number, currency_code=validate_currency_code(body.currency_code), status=validate_wallet_account_status(body.status), created_by=actor_id, updated_by=actor_id, ) try: await self.accounts.add(entity) await self.audit.record( tenant_id=tenant_id, entity_type="wallet_account", entity_id=entity.id, action=AuditAction.WALLET_OPEN, actor_user_id=actor_id, message=f"Wallet {entity.account_number} opened", ) await self.publisher.publish( event_type=LoyaltyEventType.WALLET_OPENED, aggregate_type="wallet_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="wallet_account_conflict", message="کیف پول قابل ایجاد نیست", ) return entity async def update( self, tenant_id: UUID, account_id: UUID, body: WalletAccountUpdate, *, actor: CurrentUser | None = None, ) -> WalletAccount: entity = await self.get(tenant_id, account_id) data = body.model_dump(exclude_unset=True) if "status" in data: if data["status"] is None: raise AppError( "status نمی‌تواند null باشد", status_code=422, error_code="null_not_allowed", details={"field": "status"}, ) entity.status = validate_wallet_account_status(data["status"]) actor_id = actor.user_id if actor else None entity.updated_by = actor_id await self.audit.record( tenant_id=tenant_id, entity_type="wallet_account", entity_id=entity.id, action=AuditAction.UPDATE, actor_user_id=actor_id, changes={"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) -> WalletAccount: entity = await self.accounts.get(tenant_id, account_id) if entity is None: raise NotFoundError( "کیف پول یافت نشد", error_code="wallet_account_not_found" ) return entity async def list(self, tenant_id: UUID, *, offset: int, limit: int): return await self.accounts.list_by_tenant(tenant_id, offset=offset, limit=limit) async def _get_open_account(self, tenant_id: UUID, account_id: UUID) -> WalletAccount: entity = await self.get(tenant_id, account_id) ensure_wallet_open(entity.status) return entity async def get_balance(self, tenant_id: UUID, account_id: UUID) -> dict: entity = await self.get(tenant_id, account_id) balance = await self.ledger.balance_for_account(tenant_id, account_id) return { "wallet_account_id": account_id, "member_id": entity.member_id, "program_id": entity.program_id, "status": entity.status, "currency_code": entity.currency_code, "balance": balance, } async def list_ledger( self, tenant_id: UUID, account_id: UUID, *, entry_type: WalletLedgerEntryType | None = None, offset: int = 0, limit: int = 50, ): await self.get(tenant_id, account_id) rows = await self.ledger.list_for_account( tenant_id, account_id, entry_type=entry_type, offset=offset, limit=limit ) total = await self.ledger.count_for_account( tenant_id, account_id, entry_type=entry_type ) return rows, total async def soft_delete( self, tenant_id: UUID, account_id: UUID, *, actor: CurrentUser | None = None, ) -> WalletAccount: entity = await self.get(tenant_id, account_id) actor_id = actor.user_id if actor else None entity.account_number = release_unique_token(entity.account_number, entity.id) entity.status = WalletAccountStatus.CLOSED await self.accounts.soft_delete(entity, deleted_by=actor_id) await self.audit.record( tenant_id=tenant_id, entity_type="wallet_account", entity_id=entity.id, action=AuditAction.WALLET_CLOSE, actor_user_id=actor_id, ) await self.publisher.publish( event_type=LoyaltyEventType.WALLET_CLOSED, aggregate_type="wallet_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 # ------------------------------------------------------------------ # Ledger writes # ------------------------------------------------------------------ async def _append( self, *, tenant_id: UUID, account: WalletAccount, entry_type: WalletLedgerEntryType, signed: int, body_idempotency: str | None, reason: str | None, reference_type: str | None, reference_id: str | None, metadata: dict | None, actor: CurrentUser | None, event_type: LoyaltyEventType, audit_action: AuditAction, auto_commit: bool = True, ) -> WalletLedgerEntry: if body_idempotency: existing = await self.ledger.get_by_idempotency(tenant_id, body_idempotency) if existing is not None: return existing balance = await self.ledger.balance_for_account(tenant_id, account.id) new_balance = balance + signed if new_balance < 0: raise AppError( "موجودی کیف پول کافی نیست", status_code=409, error_code="wallet_insufficient_funds", details={"balance": balance, "requested": signed}, ) actor_id = actor.user_id if actor else None entry = WalletLedgerEntry( tenant_id=tenant_id, wallet_account_id=account.id, program_id=account.program_id, member_id=account.member_id, entry_type=entry_type, amount=signed, balance_after=new_balance, idempotency_key=body_idempotency, reference_type=reference_type, reference_id=reference_id, reason=reason, metadata_json=metadata, actor_user_id=actor_id, created_at=_now(), ) await self.ledger.add(entry) await self.audit.record( tenant_id=tenant_id, entity_type="wallet_ledger_entry", entity_id=entry.id, action=audit_action, actor_user_id=actor_id, message=f"{entry_type.value} {signed} on {account.account_number}", changes={ "amount": signed, "balance_after": new_balance, "wallet_account_id": str(account.id), }, ) await self.publisher.publish( event_type=event_type, aggregate_type="wallet_ledger_entry", aggregate_id=entry.id, tenant_id=tenant_id, payload={ "wallet_account_id": str(account.id), "entry_type": entry_type.value, "amount": signed, "balance_after": new_balance, }, ) if auto_commit: await self.session.commit() await self.session.refresh(entry) return entry async def credit( self, tenant_id: UUID, account_id: UUID, body: WalletCreditRequest, *, actor: CurrentUser | None = None, auto_commit: bool = True, ) -> WalletLedgerEntry: account = await self._get_open_account(tenant_id, account_id) amount = validate_positive_amount(body.amount) return await self._append( tenant_id=tenant_id, account=account, entry_type=WalletLedgerEntryType.CREDIT, signed=signed_amount(WalletLedgerEntryType.CREDIT, amount), body_idempotency=body.idempotency_key, reason=body.reason, reference_type=body.reference_type, reference_id=body.reference_id, metadata=body.metadata, actor=actor, event_type=LoyaltyEventType.WALLET_CREDITED, audit_action=AuditAction.WALLET_CREDIT, auto_commit=auto_commit, ) async def debit( self, tenant_id: UUID, account_id: UUID, body: WalletDebitRequest, *, actor: CurrentUser | None = None, auto_commit: bool = True, ) -> WalletLedgerEntry: account = await self._get_open_account(tenant_id, account_id) amount = validate_positive_amount(body.amount) return await self._append( tenant_id=tenant_id, account=account, entry_type=WalletLedgerEntryType.DEBIT, signed=signed_amount(WalletLedgerEntryType.DEBIT, amount), body_idempotency=body.idempotency_key, reason=body.reason, reference_type=body.reference_type, reference_id=body.reference_id, metadata=body.metadata, actor=actor, event_type=LoyaltyEventType.WALLET_DEBITED, audit_action=AuditAction.WALLET_DEBIT, auto_commit=auto_commit, ) async def adjust( self, tenant_id: UUID, account_id: UUID, body: WalletAdjustRequest, *, actor: CurrentUser | None = None, auto_commit: bool = True, ) -> WalletLedgerEntry: account = await self._get_open_account(tenant_id, account_id) amount = validate_adjust_amount(body.amount) return await self._append( tenant_id=tenant_id, account=account, entry_type=WalletLedgerEntryType.ADJUST, signed=signed_amount(WalletLedgerEntryType.ADJUST, amount), body_idempotency=body.idempotency_key, reason=body.reason, reference_type=body.reference_type, reference_id=body.reference_id, metadata=body.metadata, actor=actor, event_type=LoyaltyEventType.WALLET_ADJUSTED, audit_action=AuditAction.WALLET_ADJUST, auto_commit=auto_commit, ) async def transfer( self, tenant_id: UUID, source_account_id: UUID, body: WalletTransferRequest, *, actor: CurrentUser | None = None, ) -> tuple[WalletLedgerEntry, WalletLedgerEntry]: source = await self._get_open_account(tenant_id, source_account_id) target = await self._get_open_account(tenant_id, body.target_wallet_id) ensure_not_same_account(source.id, target.id) ensure_same_currency(source.currency_code, target.currency_code) amount = validate_positive_amount(body.amount) out_key = f"{body.idempotency_key}:out" if body.idempotency_key else None in_key = f"{body.idempotency_key}:in" if body.idempotency_key else None if out_key: existing_out = await self.ledger.get_by_idempotency(tenant_id, out_key) if existing_out is not None: existing_in = await self.ledger.get_by_idempotency(tenant_id, in_key) return existing_out, existing_in transfer_group = str(uuid4()) base_metadata = dict(body.metadata or {}) out_entry = await self._append( tenant_id=tenant_id, account=source, entry_type=WalletLedgerEntryType.TRANSFER_OUT, signed=signed_amount(WalletLedgerEntryType.TRANSFER_OUT, amount), body_idempotency=out_key, reason=body.reason, reference_type="wallet_transfer", reference_id=transfer_group, metadata={**base_metadata, "transfer_group": transfer_group, "counterparty_wallet_id": str(target.id)}, actor=actor, event_type=LoyaltyEventType.WALLET_TRANSFERRED, audit_action=AuditAction.WALLET_TRANSFER, auto_commit=False, ) in_entry = await self._append( tenant_id=tenant_id, account=target, entry_type=WalletLedgerEntryType.TRANSFER_IN, signed=signed_amount(WalletLedgerEntryType.TRANSFER_IN, amount), body_idempotency=in_key, reason=body.reason, reference_type="wallet_transfer", reference_id=transfer_group, metadata={**base_metadata, "transfer_group": transfer_group, "counterparty_wallet_id": str(source.id)}, actor=actor, event_type=LoyaltyEventType.WALLET_TRANSFERRED, audit_action=AuditAction.WALLET_TRANSFER, auto_commit=False, ) await self.session.commit() await self.session.refresh(out_entry) await self.session.refresh(in_entry) return out_entry, in_entry