"""Wallet service for point transactions.""" from __future__ import annotations from dataclasses import dataclass from datetime import datetime from decimal import Decimal, ROUND_HALF_UP from typing import TYPE_CHECKING, Any, Dict, List, Optional from .exceptions import ( HookExecutionError, HookRejectionError, InsufficientBalanceError, InvalidParamsError, InvalidPointTypeError, UserNotFoundError, WalletOperationError, ) from django_package_hooks import HookManager, HookType, PostHookError from .hooks import HookContext from . import get_global_registry from .signals import transfer_completed if TYPE_CHECKING: from django.db.models import Model from .repository import WalletRepositoryProtocol @dataclass(frozen=True) class TransactionRecord: """Represents a wallet transaction record.""" id: int | None wtype: str # wallet type / point type iid: int # initiator id (user who performed transaction, -100 for system) uid: int # user id type: str # 'c' for credit/add, 'd' for debit/deduct amount: Decimal balance: Decimal # balance after transaction trans_type: int | None # transaction type code (integer constant) descr: str # remarks/description cdate: str # creation date/time extra_data: Dict[str, Any] # additional fields @dataclass class TransactionResult: """ Result of a wallet transaction with hook execution status. Attributes: success: Whether the transaction was successful transaction_id: Transaction ID if successful error_code: Error code if transaction failed (PRE hook rejection or validation error) error_message: Human-readable error message if failed error_details: Additional error context if failed post_hook_errors: List of errors from POST hooks (transaction still successful) """ success: bool transaction_id: int | None = None error_code: str | None = None error_message: str | None = None error_details: Dict[str, Any] | None = None post_hook_errors: List[PostHookError] | None = None def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for API responses.""" result = { "success": self.success, "transaction_id": self.transaction_id, } if self.error_code: result["error"] = { "code": self.error_code, "message": self.error_message, "details": self.error_details or {}, } if self.post_hook_errors: result["warnings"] = [ { "hook": err.hook_name, "code": err.error_code, "message": err.message, "details": err.details, } for err in self.post_hook_errors ] return result class WalletService: """ Python port of the legacy WalletPoint PHP class. The service works with a WalletRepositoryProtocol that only needs to define point types. All balance operations and transaction recording are handled automatically. """ # Reserved fields that cannot be used in extra_data RESERVED_FIELDS = { "id", "wtype", "iid", "uid", "type", "amount", "balance", "trans_type", "descr", "cdate", } def __init__(self, repository: "WalletRepositoryProtocol", enable_hooks: bool = False): """ Initialize the wallet service. Args: repository: WalletRepositoryProtocol instance that defines point types and provides User model enable_hooks: Whether to enable hook execution (default: False for backward compatibility) """ self.repository = repository self.enable_hooks = enable_hooks self.hook_manager = HookManager(get_global_registry()) if enable_hooks else None def add_point( self, user_id: int, point_type: str, amount: Decimal, remarks: str, trans_type: int | None = None, params: Optional[Dict[str, Any]] = None, iid: int = -100, ) -> int | TransactionResult: """ Add points to a user wallet balance. Args: user_id: The user ID point_type: The type of wallet to add to (e.g., "credit_balance") amount: The amount to add (must be positive) remarks: Description/remarks for the transaction trans_type: Transaction type code (integer constant, optional) params: Extra parameters dict with optional 'data' key for additional fields iid: Initiator ID - ID of the user who performed this transaction (-100 for system) Returns: The transaction ID (legacy behavior) or TransactionResult (if hooks are enabled) Raises: InvalidPointTypeError: If point type doesn't exist InvalidParamsError: If params are invalid WalletOperationError: If the operation fails HookRejectionError: If a PRE hook rejects the transaction """ if params is None: params = {} self._validate_add_deduct_point( "add", user_id, point_type, amount, remarks, trans_type, False, params ) # Get decimal places for this point type point_types = self.repository.get_point_types() decimal_places = point_types.get(point_type, 2) decimal_places = max(0, int(decimal_places)) # Round amount amount_rounded = amount.quantize(Decimal("0.1") ** decimal_places, rounding=ROUND_HALF_UP) # Execute PRE hooks if enabled if self.enable_hooks and self.hook_manager: current_balance = self.repository.get_user_balance(user_id, point_type) context = HookContext( operation="add", user_id=user_id, point_type=point_type, amount=amount_rounded, remarks=remarks, trans_type=trans_type, iid=iid, params=params, current_balance=current_balance, ) try: self.hook_manager.execute_pre_hooks(context) except HookRejectionError as e: return TransactionResult( success=False, error_code=e.error_code, error_message=e.message, error_details=e.details, ) except HookExecutionError: raise # Re-raise hook execution errors # Update balance try: balance_after = self.repository.update_balance(user_id, point_type, amount_rounded, allow_negative=False) except Exception as e: raise WalletOperationError(f"Error adding user point: {e}") from e # Create transaction record (always saved) record = TransactionRecord( id=None, wtype=point_type, iid=iid, uid=user_id, type="c", # credit/add amount=amount_rounded, balance=balance_after, trans_type=trans_type, descr=remarks, cdate=self._get_current_datetime(), extra_data=params.get("data", {}), ) try: transaction_id = self.repository.create_transaction_record(record) except Exception as e: raise WalletOperationError(f"Error creating transaction record: {e}") from e # Execute POST hooks if enabled post_hook_errors = [] if self.enable_hooks and self.hook_manager: context = HookContext( operation="add", user_id=user_id, point_type=point_type, amount=amount_rounded, remarks=remarks, trans_type=trans_type, iid=iid, params=params, current_balance=balance_after, metadata={"transaction_id": transaction_id}, ) post_hook_errors = self.hook_manager.execute_post_hooks(context) # Return result based on hooks enabled if self.enable_hooks: return TransactionResult( success=True, transaction_id=transaction_id, post_hook_errors=post_hook_errors if post_hook_errors else None, ) return transaction_id def deduct_point( self, user_id: int, point_type: str, amount: Decimal, remarks: str, trans_type: int | None = None, allow_negative: bool = False, params: Optional[Dict[str, Any]] = None, iid: int = -100, ) -> int | TransactionResult: """ Deduct points from a user wallet balance using atomic SQL operation. Uses SQL WHERE clause to prevent race conditions and ensure balance sufficiency. Args: user_id: The user ID point_type: The type of wallet to deduct from amount: The amount to deduct (must be positive) remarks: Description/remarks for the transaction trans_type: Transaction type code (integer constant, optional) allow_negative: Whether to allow negative balance (default: False) params: Extra parameters dict with optional 'data' key for additional fields iid: Initiator ID - ID of the user who performed this transaction (-100 for system) Returns: The transaction ID (legacy behavior) or TransactionResult (if hooks are enabled) Raises: InsufficientBalanceError: If insufficient balance and allow_negative=False InvalidPointTypeError: If point type doesn't exist InvalidParamsError: If params are invalid WalletOperationError: If the operation fails HookRejectionError: If a PRE hook rejects the transaction """ if params is None: params = {} self._validate_add_deduct_point( "deduct", user_id, point_type, amount, remarks, trans_type, allow_negative, params ) # Get decimal places for this point type point_types = self.repository.get_point_types() decimal_places = point_types.get(point_type, 2) decimal_places = max(0, int(decimal_places)) # Round amount amount_rounded = amount.quantize(Decimal("0.1") ** decimal_places, rounding=ROUND_HALF_UP) # Execute PRE hooks if enabled if self.enable_hooks and self.hook_manager: current_balance = self.repository.get_user_balance(user_id, point_type) context = HookContext( operation="deduct", user_id=user_id, point_type=point_type, amount=amount_rounded, remarks=remarks, trans_type=trans_type, iid=iid, params=params, current_balance=current_balance, metadata={"allow_negative": allow_negative}, ) try: self.hook_manager.execute_pre_hooks(context) except HookRejectionError as e: return TransactionResult( success=False, error_code=e.error_code, error_message=e.message, error_details=e.details, ) except HookExecutionError: raise # Re-raise hook execution errors # Atomically deduct using SQL WHERE clause # This prevents race conditions and ensures balance sufficiency try: success, balance_after = self.repository.deduct_balance_atomic( user_id, point_type, amount_rounded, allow_negative=allow_negative ) except Exception as e: raise WalletOperationError(f"Error deducting user point: {e}") from e # Check if deduction was successful (rowCount > 0) if not success: # Get current balance for error message current_balance = self.repository.get_user_balance(user_id, point_type) available = float(current_balance) requested = float(amount) if self.enable_hooks: return TransactionResult( success=False, error_code="INSUFFICIENT_BALANCE", error_message=f"Insufficient balance for this transaction. Available: {available:.{decimal_places}f}, " f"Requested: {requested:.{decimal_places}f}", error_details={ "available": available, "requested": requested, }, ) else: raise InsufficientBalanceError( f"Insufficient balance for this transaction. Available: {available:.{decimal_places}f}, " f"Requested: {requested:.{decimal_places}f}", available=available, requested=requested, ) # Create transaction record (always saved) record = TransactionRecord( id=None, wtype=point_type, iid=iid, uid=user_id, type="d", # debit/deduct amount=amount_rounded, balance=balance_after, trans_type=trans_type, descr=remarks, cdate=self._get_current_datetime(), extra_data=params.get("data", {}), ) try: transaction_id = self.repository.create_transaction_record(record) except Exception as e: raise WalletOperationError(f"Error creating transaction record: {e}") from e # Execute POST hooks if enabled post_hook_errors = [] if self.enable_hooks and self.hook_manager: context = HookContext( operation="deduct", user_id=user_id, point_type=point_type, amount=amount_rounded, remarks=remarks, trans_type=trans_type, iid=iid, params=params, current_balance=balance_after, metadata={"transaction_id": transaction_id, "allow_negative": allow_negative}, ) post_hook_errors = self.hook_manager.execute_post_hooks(context) # Return result based on hooks enabled if self.enable_hooks: return TransactionResult( success=True, transaction_id=transaction_id, post_hook_errors=post_hook_errors if post_hook_errors else None, ) return transaction_id def transfer_points( self, from_user_id: int, to_user_id: int, from_point_type: str, to_point_type: str, amount: Decimal, remarks: str = "", trans_type: int | None = None, metadata: Optional[Dict[str, Any]] = None, iid: int = -100, ) -> Dict[str, Any]: """ Transfer points between two users atomically. The transfer is wrapped in a database transaction to ensure both deduction and addition succeed or fail together. Supports same-type transfers (e.g., cash → cash) and cross-type transfers (e.g., cash → credit). Args: from_user_id: The sender's user ID to_user_id: The recipient's user ID from_point_type: The point type to deduct from sender to_point_type: The point type to add to recipient amount: The amount to transfer (must be positive) remarks: Description for the transaction (default: "Point transfer") trans_type: Transaction type code (defaults to WALLET_TRANSFER) metadata: Optional metadata dict to attach to both transaction records iid: Initiator ID - ID of the user who performed this transaction (-100 for system) Returns: Dict with keys: - success: True if successful, False if rejected by hook - transfer_id: Unique transfer identifier (from_transaction_id:to_transaction_id) - from_transaction_id: Transaction ID for sender's deduction - to_transaction_id: Transaction ID for recipient's addition - from_user_id: Sender's user ID - to_user_id: Recipient's user ID - from_point_type: Sender's point type - to_point_type: Recipient's point type - amount: Transfer amount - from_balance: Sender's balance after transfer - to_balance: Recipient's balance after transfer - error: Error details if rejected (only if hooks enabled) - warnings: POST hook errors if any (only if hooks enabled) Raises: UserNotFoundError: If recipient doesn't exist InsufficientBalanceError: If sender has insufficient balance (only if hooks disabled) InvalidPointTypeError: If point type doesn't exist InvalidParamsError: If parameters are invalid WalletOperationError: If the transfer fails HookRejectionError: If a PRE hook rejects the transaction (only if hooks disabled) Example: # Same point type transfer result = service.transfer_points( from_user_id=1, to_user_id=2, from_point_type="cash_balance", to_point_type="cash_balance", amount=Decimal("100.00"), remarks="Transfer to friend", ) # Cross-type transfer result = service.transfer_points( from_user_id=1, to_user_id=2, from_point_type="cash_balance", to_point_type="credit_balance", amount=Decimal("50.00"), remarks="Cash to credit conversion", ) """ from django.db import transaction as db_transaction from .transaction_types import WALLET_TRANSFER # Set default trans_type if trans_type is None: trans_type = WALLET_TRANSFER # Set default remarks if not remarks: remarks = "Point transfer" # Validate parameters if amount <= 0: raise InvalidParamsError("Transfer amount must be positive", method="transfer_points") # Validate point types exist self._validate_user_point_type(from_user_id, from_point_type) self._validate_user_point_type(to_user_id, to_point_type) # Validate recipient exists try: self.repository._get_balance_object(to_user_id, for_update=False) except Exception as e: if "DoesNotExist" in str(type(e).__name__): raise UserNotFoundError( f"Recipient user with ID {to_user_id} does not exist", user_id=to_user_id, ) from e raise # Execute PRE hooks if enabled if self.enable_hooks and self.hook_manager: current_balance = self.repository.get_user_balance(from_user_id, from_point_type) context = HookContext( operation="transfer", user_id=from_user_id, point_type=from_point_type, amount=amount, remarks=remarks, trans_type=trans_type, iid=iid, to_user_id=to_user_id, to_point_type=to_point_type, params={"data": metadata or {}}, current_balance=current_balance, ) try: self.hook_manager.execute_pre_hooks(context) except HookRejectionError as e: return { "success": False, "error": { "code": e.error_code, "message": e.message, "details": e.details, }, } except HookExecutionError: raise # Re-raise hook execution errors # Prepare metadata for transaction records transfer_metadata = metadata or {} # Temporarily disable hooks for add/deduct within transfer original_enable_hooks = self.enable_hooks self.enable_hooks = False try: # Wrap both operations in an atomic database transaction with db_transaction.atomic(): # Step 1: Deduct from sender from_params = {"data": {**transfer_metadata, "transfer_to": to_user_id}} from_transaction_id = self.deduct_point( user_id=from_user_id, point_type=from_point_type, amount=amount, remarks=f"{remarks} (to user {to_user_id})", trans_type=trans_type, allow_negative=False, params=from_params, iid=iid, ) # Step 2: Add to recipient to_params = {"data": {**transfer_metadata, "transfer_from": from_user_id}} to_transaction_id = self.add_point( user_id=to_user_id, point_type=to_point_type, amount=amount, remarks=f"{remarks} (from user {from_user_id})", trans_type=trans_type, params=to_params, iid=iid, ) # Get balances after transfer from_balance = self.repository.get_user_balance(from_user_id, from_point_type) to_balance = self.repository.get_user_balance(to_user_id, to_point_type) # Fetch the transaction records from_record = self.get_transaction_by_id(from_transaction_id) to_record = self.get_transaction_by_id(to_transaction_id) finally: # Restore hooks setting self.enable_hooks = original_enable_hooks # Generate transfer ID transfer_id = f"{from_transaction_id}:{to_transaction_id}" # Build result result = { "success": True, "transfer_id": transfer_id, "from_transaction_id": from_transaction_id, "to_transaction_id": to_transaction_id, "from_user_id": from_user_id, "to_user_id": to_user_id, "from_point_type": from_point_type, "to_point_type": to_point_type, "amount": amount, "from_balance": from_balance, "to_balance": to_balance, } # Execute POST hooks if enabled if original_enable_hooks and self.hook_manager: context = HookContext( operation="transfer", user_id=from_user_id, point_type=from_point_type, amount=amount, remarks=remarks, trans_type=trans_type, iid=iid, to_user_id=to_user_id, to_point_type=to_point_type, params={"data": transfer_metadata}, current_balance=from_balance, metadata={ "transfer_id": transfer_id, "from_transaction_id": from_transaction_id, "to_transaction_id": to_transaction_id, "from_balance": from_balance, "to_balance": to_balance, }, ) post_hook_errors = self.hook_manager.execute_post_hooks(context) if post_hook_errors: result["warnings"] = [ { "hook": err.hook_name, "code": err.error_code, "message": err.message, "details": err.details, } for err in post_hook_errors ] # Send signal (outside transaction to avoid issues with signal handlers) transfer_completed.send( sender=self.__class__, transfer_id=transfer_id, from_user_id=from_user_id, to_user_id=to_user_id, from_point_type=from_point_type, to_point_type=to_point_type, amount=amount, from_record=from_record, to_record=to_record, metadata=transfer_metadata, ) return result def _validate_user_point_type(self, user_id: int, point_type: str) -> None: """ Validate whether the given point type is valid. Raises: InvalidPointTypeError: If point type doesn't exist """ valid_types = self.repository.get_point_types() if point_type not in valid_types: raise InvalidPointTypeError( f"No such point ({point_type}) exists!", point_type=point_type ) def _validate_add_deduct_point( self, add_deduct: str, user_id: int, point_type: str, amount: Decimal, remarks: str, trans_type: int | None, allow_negative: bool, params: Dict[str, Any], ) -> None: """ Validate parameters for add/deduct point operations. Raises: InvalidPointTypeError: If point type doesn't exist InvalidParamsError: If params are invalid """ self._validate_user_point_type(user_id, point_type) method = "add_point" if add_deduct == "add" else "deduct_point" if not isinstance(params, dict): raise InvalidParamsError(f"Invalid params for {method}(), it must be a dict!", method=method) if "data" in params and params["data"]: if not isinstance(params["data"], dict): raise InvalidParamsError( f'Invalid params["data"] for {method}(), it must be a dict!', method=method ) invalid_fields = [] for field in params["data"].keys(): if field in self.RESERVED_FIELDS: invalid_fields.append(field) if invalid_fields: fields_str = ", ".join(invalid_fields) raise InvalidParamsError( f'Invalid params["data"] for {method}(), some of the data parameters are invalid: {fields_str}!', method=method, ) def get_transaction_history( self, user_id: int | None = None, point_type: str | None = None, trans_type: int | None = None, transaction_type: str | None = None, iid: int | None = None, start_date: datetime | str | None = None, end_date: datetime | str | None = None, limit: int | None = None, offset: int = 0, ) -> list[TransactionRecord]: """ Retrieve wallet transaction history with various filters. Args: user_id: Filter by user ID (uid field) point_type: Filter by wallet/point type (wtype field) trans_type: Filter by transaction type code (trans_type field, integer constant) transaction_type: Filter by transaction type ('c' for credit, 'd' for debit) (type field) iid: Filter by initiator ID (iid field) - user who performed the transaction (-100 for system) start_date: Filter transactions from this date (inclusive) (cdate field) end_date: Filter transactions until this date (inclusive) (cdate field) limit: Maximum number of records to return offset: Number of records to skip (for pagination) Returns: List of TransactionRecord objects Example: # Get all transactions for a user transactions = service.get_transaction_history(user_id=123) # Get credit transactions for a specific point type transactions = service.get_transaction_history( user_id=123, point_type="credit_balance", transaction_type="c", ) # Get transactions by transaction type code from django_wallet_utils.transaction_types import WALLET_DEPOSIT transactions = service.get_transaction_history( user_id=123, trans_type=WALLET_DEPOSIT, ) # Get transactions in date range from datetime import datetime, timedelta end = datetime.now() start = end - timedelta(days=30) transactions = service.get_transaction_history( user_id=123, start_date=start, end_date=end, limit=100, ) """ wallet_model = self.repository.get_wallet_model() queryset = wallet_model.objects.all() # Apply filters if user_id is not None: queryset = queryset.filter(uid=user_id) if point_type is not None: # Validate point type exists valid_types = self.repository.get_point_types() if point_type not in valid_types: raise InvalidPointTypeError( f"No such point ({point_type}) exists!", point_type=point_type ) queryset = queryset.filter(wtype=point_type) if trans_type is not None: queryset = queryset.filter(trans_type=trans_type) if transaction_type is not None: if transaction_type not in ("c", "d"): raise InvalidParamsError( f"Invalid transaction_type: {transaction_type}. Must be 'c' (credit) or 'd' (debit)." ) queryset = queryset.filter(type=transaction_type) if iid is not None: queryset = queryset.filter(iid=iid) if start_date is not None: if isinstance(start_date, str): start_date = datetime.fromisoformat(start_date) queryset = queryset.filter(cdate__gte=start_date) if end_date is not None: if isinstance(end_date, str): end_date = datetime.fromisoformat(end_date) queryset = queryset.filter(cdate__lte=end_date) # Order by creation date (newest first) queryset = queryset.order_by("-cdate") # Apply pagination if offset > 0: queryset = queryset[offset:] if limit is not None: queryset = queryset[:limit] # Convert to TransactionRecord objects records = [] for obj in queryset: record = TransactionRecord( id=obj.id, wtype=obj.wtype, iid=obj.iid, uid=obj.uid, type=obj.type, amount=obj.amount, balance=obj.balance, trans_type=obj.trans_type, descr=obj.descr, cdate=obj.cdate.isoformat() if hasattr(obj.cdate, "isoformat") else str(obj.cdate), extra_data=obj.extra_data or {}, ) records.append(record) return records def get_transaction_by_id(self, transaction_id: int) -> TransactionRecord | None: """ Get a single transaction by its ID. Args: transaction_id: The transaction ID Returns: TransactionRecord if found, None otherwise """ wallet_model = self.repository.get_wallet_model() try: obj = wallet_model.objects.get(id=transaction_id) return TransactionRecord( id=obj.id, wtype=obj.wtype, iid=obj.iid, uid=obj.uid, type=obj.type, amount=obj.amount, balance=obj.balance, trans_type=obj.trans_type, descr=obj.descr, cdate=obj.cdate.isoformat() if hasattr(obj.cdate, "isoformat") else str(obj.cdate), extra_data=obj.extra_data or {}, ) except wallet_model.DoesNotExist: return None def count_transactions( self, user_id: int | None = None, point_type: str | None = None, trans_type: int | None = None, transaction_type: str | None = None, iid: int | None = None, start_date: datetime | str | None = None, end_date: datetime | str | None = None, ) -> int: """ Count transactions matching the given filters. Args: user_id: Filter by user ID (uid field) point_type: Filter by wallet/point type (wtype field) trans_type: Filter by transaction type code (trans_type field, integer constant) transaction_type: Filter by transaction type ('c' for credit, 'd' for debit) (type field) iid: Filter by initiator ID (iid field) - user who performed the transaction (-100 for system) start_date: Filter transactions from this date (inclusive) (cdate field) end_date: Filter transactions until this date (inclusive) (cdate field) Returns: Number of transactions matching the filters """ wallet_model = self.repository.get_wallet_model() queryset = wallet_model.objects.all() # Apply filters (same as get_transaction_history) if user_id is not None: queryset = queryset.filter(uid=user_id) if point_type is not None: # Validate point type exists valid_types = self.repository.get_point_types() if point_type not in valid_types: raise InvalidPointTypeError( f"No such point ({point_type}) exists!", point_type=point_type ) queryset = queryset.filter(wtype=point_type) if trans_type is not None: queryset = queryset.filter(trans_type=trans_type) if transaction_type is not None: if transaction_type not in ("c", "d"): raise InvalidParamsError( f"Invalid transaction_type: {transaction_type}. Must be 'c' (credit) or 'd' (debit)." ) queryset = queryset.filter(type=transaction_type) if iid is not None: queryset = queryset.filter(iid=iid) if start_date is not None: if isinstance(start_date, str): start_date = datetime.fromisoformat(start_date) queryset = queryset.filter(cdate__gte=start_date) if end_date is not None: if isinstance(end_date, str): end_date = datetime.fromisoformat(end_date) queryset = queryset.filter(cdate__lte=end_date) return queryset.count() @staticmethod def _get_current_datetime() -> str: """Get current datetime as ISO format string.""" return datetime.now().isoformat()