India's UPI.

# vpa/resolver.py

"""
VPA Resolution Service

Maps Virtual Payment Addresses to actual bank accounts.
This is the most critical lookup in the entire system.
"""

from dataclasses import dataclass
from typing import Optional, Tuple
from datetime import datetime, timedelta
import hashlib


@dataclass
class AccountDetails:
    """Resolved account information."""
    bank_code: str
    account_number_masked: str  # Only last 4 digits visible
    account_holder_name: str
    ifsc_code: str
    is_active: bool
    verified_at: datetime


@dataclass
class VPAResolutionResult:
    """Result of VPA resolution."""
    success: bool
    account: Optional[AccountDetails] = None
    error_code: Optional[str] = None
    resolution_time_ms: float = 0
    cache_hit: bool = False


class VPAResolver:
    """
    Resolves VPAs to bank account details.
    
    Design principles:
    - Cache aggressively (VPAs don't change often)
    - Fail fast on invalid formats
    - Banks are the source of truth
    """
    
    def __init__(
        self,
        cache,           # Redis cluster
        bank_registry,   # Bank code → Bank API mapping
        bank_gateway,    # Gateway to call bank APIs
        metrics
    ):
        self.cache = cache
        self.registry = bank_registry
        self.gateway = bank_gateway
        self.metrics = metrics
        
        # Cache settings
        self.cache_ttl = timedelta(minutes=30)
        self.negative_cache_ttl = timedelta(minutes=5)
    
    async def resolve(self, vpa: str) -> VPAResolutionResult:
        """
        Resolve a VPA to account details.
        
        VPA format: handle@suffix
        Example: alice@okhdfcbank, 9876543210@ybl
        """
        start_time = datetime.utcnow()
        
        # Step 1: Parse and validate VPA format
        parsed = self._parse_vpa(vpa)
        if not parsed:
            return VPAResolutionResult(
                success=False,
                error_code="INVALID_VPA_FORMAT"
            )
        
        handle, suffix = parsed
        
        # Step 2: Get bank code from suffix
        bank_code = self.registry.get_bank_for_suffix(suffix)
        if not bank_code:
            return VPAResolutionResult(
                success=False,
                error_code="UNKNOWN_VPA_SUFFIX"
            )
        
        # Step 3: Check cache
        cache_key = f"vpa:{vpa.lower()}"
        cached = await self.cache.get(cache_key)
        
        if cached:
            if cached == "NOT_FOUND":
                return VPAResolutionResult(
                    success=False,
                    error_code="VPA_NOT_FOUND",
                    cache_hit=True
                )
            
            account = AccountDetails(**cached)
            return VPAResolutionResult(
                success=True,
                account=account,
                resolution_time_ms=self._elapsed_ms(start_time),
                cache_hit=True
            )
        
        # Step 4: Query the bank
        try:
            account = await self.gateway.resolve_vpa(
                bank_code=bank_code,
                handle=handle,
                suffix=suffix
            )
            
            if account:
                # Cache the result
                await self.cache.set(
                    cache_key,
                    account.__dict__,
                    ttl=self.cache_ttl
                )
                
                return VPAResolutionResult(
                    success=True,
                    account=account,
                    resolution_time_ms=self._elapsed_ms(start_time),
                    cache_hit=False
                )
            else:
                # Cache negative result (VPA doesn't exist)
                await self.cache.set(
                    cache_key,
                    "NOT_FOUND",
                    ttl=self.negative_cache_ttl
                )
                
                return VPAResolutionResult(
                    success=False,
                    error_code="VPA_NOT_FOUND",
                    resolution_time_ms=self._elapsed_ms(start_time)
                )
                
        except BankTimeoutError:
            return VPAResolutionResult(
                success=False,
                error_code="BANK_TIMEOUT"
            )
        except BankUnavailableError:
            return VPAResolutionResult(
                success=False,
                error_code="BANK_UNAVAILABLE"
            )
    
    def _parse_vpa(self, vpa: str) -> Optional[Tuple[str, str]]:
        """Parse VPA into handle and suffix."""
        if not vpa or '@' not in vpa:
            return None
        
        parts = vpa.lower().strip().split('@')
        if len(parts) != 2:
            return None
        
        handle, suffix = parts
        
        # Validate handle (alphanumeric, 3-50 chars)
        if not handle or len(handle) < 3 or len(handle) > 50:
            return None
        
        # Validate suffix (registered bank suffix)
        if not suffix or len(suffix) < 2 or len(suffix) > 20:
            return None
        
        return handle, suffix
    
    def _elapsed_ms(self, start: datetime) -> float:
        return (datetime.utcnow() - start).total_seconds() * 1000

# transaction/processor.py

"""
Transaction Processing with Atomic Guarantees.

This is the most critical code in the entire system.
Money cannot be lost under any circumstances.
"""

from dataclasses import dataclass
from typing import Optional
from enum import Enum
from datetime import datetime, timedelta
import uuid


class TransactionState(Enum):
    CREATED = "created"
    PENDING = "pending"
    DEBIT_INITIATED = "debit_initiated"
    DEBITED = "debited"
    CREDIT_INITIATED = "credit_initiated"
    COMPLETED = "completed"
    FAILED = "failed"
    DEEMED_SUCCESS = "deemed_success"
    REVERSED = "reversed"


@dataclass
class Transaction:
    """A UPI transaction record."""
    txn_id: str
    sender_vpa: str
    receiver_vpa: str
    amount: int  # In paise (smallest unit)
    
    state: TransactionState
    created_at: datetime
    updated_at: datetime
    
    # Bank references
    sender_bank_ref: Optional[str] = None
    receiver_bank_ref: Optional[str] = None
    
    # For idempotency
    idempotency_key: Optional[str] = None
    
    # Reversal tracking
    reversal_initiated: bool = False
    reversal_completed: bool = False


class TransactionProcessor:
    """
    Processes UPI transactions with atomic guarantees.
    
    Key principles:
    1. IDEMPOTENCY: Same request = same result (no double-debit)
    2. ATOMICITY: Either complete success or complete rollback
    3. DURABILITY: State persisted before any bank call
    4. RECOVERABILITY: Can resume from any failure point
    """
    
    def __init__(
        self,
        db,              # Transaction database
        bank_gateway,    # Bank API gateway
        reversal_queue,  # Queue for failed transactions
        audit_log
    ):
        self.db = db
        self.gateway = bank_gateway
        self.reversal_queue = reversal_queue
        self.audit = audit_log
        
        # Timeouts
        self.debit_timeout = timedelta(seconds=30)
        self.credit_timeout = timedelta(seconds=30)
    
    async def process(
        self,
        sender_vpa: str,
        receiver_vpa: str,
        amount: int,
        idempotency_key: str
    ) -> Transaction:
        """
        Process a payment transaction.
        
        CRITICAL: This method must be idempotent.
        Same idempotency_key = same result, always.
        """
        
        # STEP 0: Check idempotency
        existing = await self.db.get_by_idempotency_key(idempotency_key)
        if existing:
            # Return existing result (no reprocessing)
            await self.audit.log("IDEMPOTENT_RETURN", existing.txn_id)
            return existing
        
        # STEP 1: Create transaction record FIRST
        txn = Transaction(
            txn_id=str(uuid.uuid4()),
            sender_vpa=sender_vpa,
            receiver_vpa=receiver_vpa,
            amount=amount,
            state=TransactionState.CREATED,
            created_at=datetime.utcnow(),
            updated_at=datetime.utcnow(),
            idempotency_key=idempotency_key
        )
        
        # Persist BEFORE any bank call
        await self.db.save(txn)
        await self.audit.log("TXN_CREATED", txn.txn_id)
        
        try:
            # STEP 2: Initiate debit
            txn.state = TransactionState.DEBIT_INITIATED
            txn.updated_at = datetime.utcnow()
            await self.db.save(txn)
            
            debit_result = await self.gateway.debit(
                vpa=sender_vpa,
                amount=amount,
                txn_ref=txn.txn_id,
                timeout=self.debit_timeout
            )
            
            if not debit_result.success:
                # Debit failed cleanly — no money moved
                txn.state = TransactionState.FAILED
                txn.updated_at = datetime.utcnow()
                await self.db.save(txn)
                await self.audit.log("DEBIT_FAILED", txn.txn_id, 
                                    debit_result.error)
                return txn
            
            # STEP 3: Debit succeeded — record it
            txn.state = TransactionState.DEBITED
            txn.sender_bank_ref = debit_result.bank_reference
            txn.updated_at = datetime.utcnow()
            await self.db.save(txn)
            await self.audit.log("DEBIT_SUCCESS", txn.txn_id)
            
            # STEP 4: Initiate credit
            # CRITICAL: From this point, we MUST either complete or reverse
            txn.state = TransactionState.CREDIT_INITIATED
            txn.updated_at = datetime.utcnow()
            await self.db.save(txn)
            
            credit_result = await self.gateway.credit(
                vpa=receiver_vpa,
                amount=amount,
                txn_ref=txn.txn_id,
                timeout=self.credit_timeout
            )
            
            if credit_result.success:
                # SUCCESS! Transaction complete
                txn.state = TransactionState.COMPLETED
                txn.receiver_bank_ref = credit_result.bank_reference
                txn.updated_at = datetime.utcnow()
                await self.db.save(txn)
                await self.audit.log("TXN_COMPLETED", txn.txn_id)
                return txn
            
            elif credit_result.status == "TIMEOUT":
                # UNCERTAINTY: We don't know if credit happened
                # Mark as DEEMED_SUCCESS — settlement will clarify
                txn.state = TransactionState.DEEMED_SUCCESS
                txn.updated_at = datetime.utcnow()
                await self.db.save(txn)
                await self.audit.log("TXN_DEEMED_SUCCESS", txn.txn_id)
                
                # Schedule reconciliation check
                await self.reversal_queue.schedule_check(
                    txn.txn_id,
                    check_at=datetime.utcnow() + timedelta(hours=24)
                )
                return txn
            
            else:
                # Credit FAILED — must reverse the debit
                await self._initiate_reversal(txn)
                return txn
                
        except Exception as e:
            # Unexpected error — check state and recover
            await self.audit.log("TXN_ERROR", txn.txn_id, str(e))
            await self._handle_error(txn, e)
            raise
    
    async def _initiate_reversal(self, txn: Transaction):
        """
        Reverse a failed transaction.
        Credit the debited amount back to sender.
        """
        txn.reversal_initiated = True
        txn.updated_at = datetime.utcnow()
        await self.db.save(txn)
        await self.audit.log("REVERSAL_INITIATED", txn.txn_id)
        
        # Queue for reversal (handled by separate process)
        await self.reversal_queue.enqueue(txn.txn_id)
    
    async def process_reversal(self, txn_id: str):
        """
        Execute reversal — credit money back to sender.
        Called by reversal worker.
        """
        txn = await self.db.get(txn_id)
        
        if txn.reversal_completed:
            return  # Already reversed
        
        # Credit back to sender
        reversal_result = await self.gateway.credit(
            vpa=txn.sender_vpa,
            amount=txn.amount,
            txn_ref=f"REV-{txn.txn_id}",
            timeout=self.credit_timeout
        )
        
        if reversal_result.success:
            txn.state = TransactionState.REVERSED
            txn.reversal_completed = True
            txn.updated_at = datetime.utcnow()
            await self.db.save(txn)
            await self.audit.log("REVERSAL_COMPLETED", txn.txn_id)
        else:
            # Reversal failed — retry later
            # This is a critical alert scenario
            await self.audit.log("REVERSAL_FAILED", txn.txn_id,
                                reversal_result.error)
            await self.reversal_queue.schedule_retry(
                txn.txn_id,
                retry_at=datetime.utcnow() + timedelta(minutes=15)
            )

# bank/gateway.py

"""
Bank Gateway: Unified interface to 740+ banks.

Each bank is different. This gateway provides a
consistent interface while handling per-bank quirks.
"""

from dataclasses import dataclass
from typing import Dict, Optional
from datetime import datetime, timedelta
from enum import Enum


class CircuitState(Enum):
    CLOSED = "closed"    # Normal operation
    OPEN = "open"        # Bank failing, don't try
    HALF_OPEN = "half"   # Testing if bank recovered


@dataclass
class BankConfig:
    """Configuration for a bank."""
    bank_code: str
    endpoint: str
    timeout_ms: int = 30000
    
    # Circuit breaker settings
    failure_threshold: int = 5
    recovery_timeout_s: int = 30
    
    # Illustrative quirks for a generic multi-bank gateway
    # (UPI itself gives every bank one standard XML API)
    requires_padding: bool = False
    amount_in_rupees: bool = False  # e.g. a partner that wants rupees, not paise
    legacy_xml_format: bool = False


class BankGateway:
    """
    Gateway for all bank operations.
    """
    
    def __init__(
        self,
        http_client,
        bank_configs: Dict[str, BankConfig],
        metrics
    ):
        self.http = http_client
        self.configs = bank_configs
        self.metrics = metrics
        
        # Circuit breakers per bank
        self.circuits: Dict[str, CircuitBreaker] = {
            code: CircuitBreaker(config)
            for code, config in bank_configs.items()
        }
    
    async def debit(
        self,
        bank_code: str,
        account_ref: str,
        amount: int,
        txn_ref: str
    ) -> 'BankResponse':
        """
        Debit an account at a bank.
        
        Amount is in paise (smallest unit).
        """
        config = self.configs.get(bank_code)
        if not config:
            return BankResponse(
                success=False,
                error_code="UNKNOWN_BANK"
            )
        
        # Check circuit breaker
        circuit = self.circuits[bank_code]
        if not circuit.can_execute():
            self.metrics.increment("bank_circuit_open", bank_code)
            return BankResponse(
                success=False,
                error_code="BANK_CIRCUIT_OPEN"
            )
        
        try:
            # Build request (handle bank-specific formats)
            request = self._build_debit_request(
                config, account_ref, amount, txn_ref
            )
            
            # Make the call
            start = datetime.utcnow()
            response = await self.http.post(
                config.endpoint + "/debit",
                json=request,
                timeout=config.timeout_ms / 1000
            )
            latency = (datetime.utcnow() - start).total_seconds() * 1000
            
            # Record metrics
            self.metrics.record_latency("bank_debit", bank_code, latency)
            
            # Parse response
            result = self._parse_response(config, response)
            
            if result.success:
                circuit.record_success()
            else:
                circuit.record_failure()
            
            return result
            
        except TimeoutError:
            circuit.record_failure()
            self.metrics.increment("bank_timeout", bank_code)
            return BankResponse(
                success=False,
                error_code="TIMEOUT",
                status="TIMEOUT"
            )
        except Exception as e:
            circuit.record_failure()
            self.metrics.increment("bank_error", bank_code)
            return BankResponse(
                success=False,
                error_code="BANK_ERROR",
                error_message=str(e)
            )
    
    def _build_debit_request(
        self,
        config: BankConfig,
        account_ref: str,
        amount: int,
        txn_ref: str
    ) -> dict:
        """Build bank-specific request format."""
        
        # Handle amount format (paise vs rupees)
        if config.amount_in_rupees:
            amount_value = amount / 100
        else:
            amount_value = amount
        
        if config.legacy_xml_format:
            # Illustrative: a partner on an older format
            return {
                "xml_payload": self._build_xml(
                    account_ref, amount_value, txn_ref
                )
            }
        
        return {
            "account_reference": account_ref,
            "amount": amount_value,
            "transaction_reference": txn_ref,
            "timestamp": datetime.utcnow().isoformat()
        }


class CircuitBreaker:
    """
    Circuit breaker for bank connections.
    
    Prevents cascade failures when a bank is down.
    """
    
    def __init__(self, config: BankConfig):
        self.config = config
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure: Optional[datetime] = None
        self.success_count = 0
    
    def can_execute(self) -> bool:
        """Check if we can make a request to this bank."""
        if self.state == CircuitState.CLOSED:
            return True
        
        if self.state == CircuitState.OPEN:
            # Check if recovery timeout passed
            if self.last_failure:
                elapsed = (datetime.utcnow() - self.last_failure).total_seconds()
                if elapsed > self.config.recovery_timeout_s:
                    self.state = CircuitState.HALF_OPEN
                    self.success_count = 0
                    return True
            return False
        
        if self.state == CircuitState.HALF_OPEN:
            return True
        
        return False
    
    def record_success(self):
        """Record a successful request."""
        if self.state == CircuitState.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= 3:  # 3 successes to close
                self.state = CircuitState.CLOSED
                self.failure_count = 0
        else:
            self.failure_count = 0
    
    def record_failure(self):
        """Record a failed request."""
        self.failure_count += 1
        self.last_failure = datetime.utcnow()
        
        if self.state == CircuitState.HALF_OPEN:
            # Back to open
            self.state = CircuitState.OPEN
            self.success_count = 0
        elif self.failure_count >= self.config.failure_threshold:
            self.state = CircuitState.OPEN

# security/fraud_detector.py

"""
Real-Time Fraud Detection for UPI.

Must decide in < 50ms whether to allow a transaction.
"""

from dataclasses import dataclass
from typing import List, Tuple
from datetime import datetime, timedelta


@dataclass
class FraudSignals:
    """Signals used for fraud detection."""
    user_id: str
    device_id: str
    amount: int
    receiver_vpa: str
    
    # Velocity
    txn_count_1hr: int
    txn_count_24hr: int
    total_amount_24hr: int
    
    # Device
    is_new_device: bool
    device_age_days: int
    
    # Behavioral
    is_new_receiver: bool
    typical_amount: int
    typical_time_of_day: List[int]
    current_hour: int
    
    # Location
    device_location: str
    usual_locations: List[str]


class FraudDetector:
    """
    Real-time fraud detection.
    
    Must be FAST (< 50ms) and ACCURATE (low false positives).
    """
    
    def __init__(self, ml_model, rules_engine, blocklist):
        self.model = ml_model
        self.rules = rules_engine
        self.blocklist = blocklist
    
    async def evaluate(
        self,
        signals: FraudSignals
    ) -> Tuple[str, float, List[str]]:
        """
        Evaluate fraud risk.
        
        Returns: (decision, confidence, triggered_rules)
        decision: "ALLOW", "BLOCK", "STEP_UP"
        """
        triggered_rules = []
        
        # RULE 1: Blocklist check (instant)
        if await self.blocklist.is_blocked(signals.user_id):
            return "BLOCK", 1.0, ["USER_BLOCKED"]
        
        if await self.blocklist.is_blocked(signals.device_id):
            return "BLOCK", 1.0, ["DEVICE_BLOCKED"]
        
        # RULE 2: Velocity checks
        if signals.txn_count_1hr > 10:
            triggered_rules.append("HIGH_VELOCITY_1HR")
        
        if signals.txn_count_24hr > 50:
            triggered_rules.append("HIGH_VELOCITY_24HR")
        
        if signals.total_amount_24hr > 100000_00:  # ₹1 lakh in paise
            triggered_rules.append("HIGH_AMOUNT_24HR")
        
        # RULE 3: Amount anomaly
        if signals.amount > signals.typical_amount * 10:
            triggered_rules.append("AMOUNT_ANOMALY")
        
        # RULE 4: New device
        if signals.is_new_device:
            triggered_rules.append("NEW_DEVICE")
            if signals.amount > 10000_00:  # > ₹10,000 on new device
                triggered_rules.append("HIGH_AMOUNT_NEW_DEVICE")
        
        # RULE 5: Unusual time
        if signals.current_hour not in signals.typical_time_of_day:
            triggered_rules.append("UNUSUAL_TIME")
        
        # RULE 6: Location check
        if signals.device_location not in signals.usual_locations:
            triggered_rules.append("UNUSUAL_LOCATION")
        
        # ML model for complex patterns
        ml_score = await self.model.predict(signals)
        
        # Decision logic
        if ml_score > 0.9 or len(triggered_rules) > 3:
            return "BLOCK", ml_score, triggered_rules
        
        if ml_score > 0.7 or len(triggered_rules) > 1:
            # Step-up: require additional verification
            return "STEP_UP", ml_score, triggered_rules
        
        return "ALLOW", 1 - ml_score, triggered_rules