""" routers/payments.py ──────────────────── M-Pesa STK Push endpoints. ENDPOINTS: POST /payments/mpesa/stk-push → Initiate payment (cashier triggers this) POST /payments/mpesa/callback → Safaricom calls this after customer pays GET /payments/mpesa/{checkout_request_id}/status → Poll payment status """ import logging from datetime import datetime, timezone from fastapi import APIRouter, Depends, HTTPException, Request, status from sqlalchemy.orm import Session from database import get_db from models.orm import AuditLog, MpesaTransaction, MpesaStatus, Sale, SaleStatus, User, UserRole from schemas.schemas import MpesaSTKPushIn, MpesaSTKPushOut, MpesaStatusOut from services.mpesa_service import stk_push, query_stk_status, parse_callback from utils.errors import safe_error from utils.security import get_current_user, require_admin_or_manager from models.orm import User logger = logging.getLogger(__name__) router = APIRouter(prefix="/payments/mpesa", tags=["Payments"]) # ── Initiate STK Push ───────────────────────────────────────────────────────── @router.post("/stk-push", response_model=MpesaSTKPushOut, summary="Send M-Pesa payment prompt to customer") def initiate_stk_push( payload: MpesaSTKPushIn, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ): """ Cashier enters the customer's phone number and the sale amount. This sends an STK Push prompt to the customer's phone. Customer has 60 seconds to enter their PIN. """ # Verify the sale exists and is in the right state sale = db.get(Sale, payload.sale_id) if not sale: raise HTTPException(status_code=404, detail="Sale not found") if sale.status != SaleStatus.COMPLETED: raise HTTPException( status_code=400, detail="Can only request payment for a completed sale" ) # Check no pending payment already exists for this sale existing = ( db.query(MpesaTransaction) .filter( MpesaTransaction.sale_id == payload.sale_id, MpesaTransaction.status == MpesaStatus.PENDING, ) .first() ) if existing: raise HTTPException( status_code=409, detail="A payment request is already pending for this sale. " "Ask the customer to check their phone or wait 60 seconds." ) try: amount_kes = int(sale.total_amount) # M-Pesa only accepts whole KES reference = f"SALE-{sale.id}" result = stk_push( phone_number=payload.phone_number, amount=amount_kes, account_reference=reference, description="Cosmetics payment", ) # Persist the pending transaction immediately txn = MpesaTransaction( sale_id=sale.id, checkout_request_id=result["CheckoutRequestID"], merchant_request_id=result["MerchantRequestID"], phone_number=payload.phone_number, amount=amount_kes, status=MpesaStatus.PENDING, ) db.add(txn) db.add(AuditLog( user_id=current_user.id, action="MPESA_STK_PUSH", entity="Sale", entity_id=sale.id, detail=f"STK Push sent to {payload.phone_number} for KES {amount_kes}", )) db.commit() db.refresh(txn) return MpesaSTKPushOut( checkout_request_id=txn.checkout_request_id, message="Payment prompt sent. Ask the customer to enter their M-Pesa PIN.", ) except HTTPException: raise except ValueError as e: # Phone number validation error from _normalise_phone raise HTTPException(status_code=400, detail=str(e)) except Exception as e: db.rollback() raise safe_error(e, "Could not initiate M-Pesa payment. Please try again.") # ── Safaricom Callback ──────────────────────────────────────────────────────── @router.post("/callback", summary="Safaricom payment result callback (do not call manually)") async def mpesa_callback(request: Request, db: Session = Depends(get_db)): """ Safaricom POSTs the payment result here after the customer acts on the STK prompt. SECURITY NOTES: • This endpoint must NOT require authentication — Safaricom calls it directly. • Always return HTTP 200 to Safaricom, even on errors. If you return non-200, Safaricom will retry the callback indefinitely. • Validate the CheckoutRequestID exists in your DB before trusting the payload. • Ideally, restrict this endpoint to Safaricom's IP ranges at the load balancer or nginx level (see Safaricom documentation for their IP list). """ try: body = await request.json() logger.info("M-Pesa callback received: %s", body) parsed = parse_callback(body) checkout_id = parsed["checkout_request_id"] if not checkout_id: logger.warning("Callback missing CheckoutRequestID — ignoring") return {"ResultCode": 0, "ResultDesc": "Accepted"} # Find our record — reject unknown checkout IDs (prevents spoofed callbacks) txn = ( db.query(MpesaTransaction) .filter(MpesaTransaction.checkout_request_id == checkout_id) .first() ) if not txn: logger.warning("Callback for unknown CheckoutRequestID=%s", checkout_id) return {"ResultCode": 0, "ResultDesc": "Accepted"} # Still return 200 # Idempotency: if already processed, don't update again if txn.status != MpesaStatus.PENDING: logger.info("Duplicate callback for CheckoutRequestID=%s — already %s", checkout_id, txn.status) return {"ResultCode": 0, "ResultDesc": "Accepted"} # Update transaction txn.result_code = parsed["result_code"] txn.result_desc = parsed["result_desc"] txn.mpesa_receipt = parsed["mpesa_receipt"] txn.updated_at = datetime.now(timezone.utc) txn.status = MpesaStatus.SUCCESS if parsed["success"] else MpesaStatus.FAILED # Verify amount matches what we expected (prevents underpayment attacks) if parsed["success"] and parsed["amount"] is not None: if int(parsed["amount"]) < txn.amount: logger.error( "AMOUNT MISMATCH: expected KES %d, received KES %s for CheckoutRequestID=%s", txn.amount, parsed["amount"], checkout_id, ) txn.status = MpesaStatus.FAILED txn.result_desc = f"Amount mismatch: expected {txn.amount}, got {parsed['amount']}" db.add(AuditLog( user_id=None, # System-generated (Safaricom callback) action="MPESA_CALLBACK", entity="MpesaTransaction", entity_id=txn.id, detail=( f"Receipt={parsed['mpesa_receipt']} " f"Status={txn.status} " f"ResultCode={parsed['result_code']}" ), )) # ── Self-registration: create the admin account now that payment succeeded ── if txn.purpose == "registration" and txn.status == MpesaStatus.SUCCESS and not txn.registration_completed: existing = db.query(User).filter(User.email == txn.pending_email).first() if existing: # Someone else's signup (or a duplicate callback race) already # created the account for this email — don't create a second one. logger.warning( "Registration payment succeeded but user %s already exists — skipping account creation", txn.pending_email, ) elif db.query(User).first() is not None: # Another registration completed first — this cosmetics install # is already set up. Log it; the paid customer should be # refunded/contacted manually since payment did succeed. logger.error( "Registration payment succeeded for %s but an admin already exists " "on this install — MANUAL FOLLOW-UP NEEDED (refund/contact customer).", txn.pending_email, ) else: new_admin = User( full_name=txn.pending_full_name, email=txn.pending_email, hashed_password=txn.pending_password_hash, role=UserRole.ADMIN, is_active=True, phone=txn.pending_phone, business_name=txn.pending_business_name, ) db.add(new_admin) txn.registration_completed = True db.add(AuditLog( user_id=None, action="REGISTRATION_COMPLETED", detail=f"Admin account created for {txn.pending_email} after confirmed payment (receipt={parsed['mpesa_receipt']})", )) logger.info("Self-registration complete for %s", txn.pending_email) db.commit() logger.info( "M-Pesa payment %s: CheckoutRequestID=%s receipt=%s", txn.status, checkout_id, parsed["mpesa_receipt"], ) except Exception as e: # IMPORTANT: always return 200 to Safaricom even if we had an internal error. # Log it and investigate manually. Do NOT let exceptions bubble up here. logger.error("Error processing M-Pesa callback: %s", e, exc_info=True) # Safaricom expects exactly this response shape return {"ResultCode": 0, "ResultDesc": "Accepted"} # ── Status Poll ─────────────────────────────────────────────────────────────── @router.get("/{checkout_request_id}/status", response_model=MpesaStatusOut, summary="Check payment status") def get_payment_status( checkout_request_id: str, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ): """ Poll payment status. Call this every 5 seconds from the frontend while waiting for the customer to enter their PIN. If status is still PENDING after 60 seconds, query Safaricom directly as a fallback (in case the callback was missed). """ txn = ( db.query(MpesaTransaction) .filter(MpesaTransaction.checkout_request_id == checkout_request_id) .first() ) if not txn: raise HTTPException(status_code=404, detail="Transaction not found") # If still pending after 70s, query Safaricom directly if txn.status == MpesaStatus.PENDING: age_seconds = (datetime.now(timezone.utc) - txn.created_at).total_seconds() if age_seconds > 70: try: result = query_stk_status(checkout_request_id) result_code = int(result.get("ResultCode", -1)) if result_code == 0: txn.status = MpesaStatus.SUCCESS txn.result_desc = result.get("ResultDesc") elif result_code != 1032: # 1032 = still waiting txn.status = MpesaStatus.FAILED txn.result_desc = result.get("ResultDesc") txn.updated_at = datetime.now(timezone.utc) db.commit() except Exception as e: logger.warning("STK status query failed: %s", e) return MpesaStatusOut( checkout_request_id=txn.checkout_request_id, status=txn.status, mpesa_receipt=txn.mpesa_receipt, amount=txn.amount, result_desc=txn.result_desc, )