from __future__ import annotations import hashlib import hmac import logging from dataclasses import dataclass from pathlib import Path from typing import Any from uuid import UUID, uuid4 import yaml from core.config.settings import config from core.http.errors import ApiProblemError, problem_payload from models.apple_iap_transaction import AppleIapTransaction from schemas.domain.points import ( ApplyPointsChangeCommand, PurchaseLedgerMetadata, ) from schemas.enums import PointsBizType, PointsChangeType, PointsOperatorType from v1.payments.apple_verifier import ( AppleJwsVerifier, VerificationError, VerifiedTransaction, ) from v1.payments.repository import PaymentRepository from v1.payments.schemas import VerifyTransactionRequest, VerifyTransactionResponse from v1.points.repository import PointsRepository logger = logging.getLogger(__name__) @dataclass(frozen=True) class ProductMapping: app_store_product_id: str credits: int type: str _product_mappings_cache: dict[str, ProductMapping] | None = None def _load_product_mappings() -> dict[str, ProductMapping]: global _product_mappings_cache if _product_mappings_cache is not None: return _product_mappings_cache mapping_path = ( Path(__file__).parent.parent.parent / "core/config/static/packages/mapping.yaml" ) with mapping_path.open("r", encoding="utf-8") as f: raw: Any = yaml.safe_load(f) or {} mappings: dict[str, ProductMapping] = {} product_mappings: Any = raw.get("product_mappings", {}) for code, entry in product_mappings.items(): mappings[str(code)] = ProductMapping( app_store_product_id=str(entry["app_store_product_id"]), credits=int(entry["credits"]), type=str(entry["type"]), ) _product_mappings_cache = mappings return mappings def clear_product_mappings_cache() -> None: global _product_mappings_cache _product_mappings_cache = None class PaymentService: def __init__( self, *, payment_repo: PaymentRepository, points_repo: PointsRepository, verifier: AppleJwsVerifier, ) -> None: self._payment_repo: PaymentRepository = payment_repo self._points_repo: PointsRepository = points_repo self._verifier: AppleJwsVerifier = verifier async def verify_and_grant( self, *, user_id: UUID, user_email: str, request: VerifyTransactionRequest, ) -> VerifyTransactionResponse: mappings = _load_product_mappings() product_mapping = mappings.get(request.product_code) if product_mapping is None: raise ApiProblemError( status_code=404, detail=problem_payload( code="PAYMENT_PRODUCT_NOT_FOUND", detail=f"Product not found: {request.product_code}", ), ) if request.app_store_product_id != product_mapping.app_store_product_id: raise ApiProblemError( status_code=422, detail=problem_payload( code="PAYMENT_PRODUCT_MISMATCH", detail="appStoreProductId does not match backend mapping", ), ) expected_bundle_id = config.apple_iap.bundle_id expected_environment = "Sandbox" if config.runtime.environment != "prod" else "Production" result = self._verifier.verify_signed_transaction( request.signed_transaction_info, expected_bundle_id=expected_bundle_id, expected_product_id=product_mapping.app_store_product_id, expected_environment=expected_environment, ) if isinstance(result, VerificationError): status_code = 422 if result.code == "PAYMENT_TRANSACTION_REVOKED": status_code = 409 elif result.code == "PAYMENT_PRODUCT_MISMATCH": status_code = 422 raise ApiProblemError( status_code=status_code, detail=problem_payload(code=result.code, detail=result.detail), ) verified: VerifiedTransaction = result if str(verified.transaction_id) != request.transaction_id: raise ApiProblemError( status_code=422, detail=problem_payload( code="PAYMENT_TRANSACTION_INVALID", detail="transactionId does not match verified payload", ), ) existing = await self._payment_repo.get_transaction_by_transaction_id( transaction_id=verified.transaction_id ) if existing is not None: if existing.user_id == user_id and existing.status == "granted": account = await self._payment_repo.get_or_create_user_points_for_update( user_id=user_id ) return VerifyTransactionResponse( status="already_granted", productCode=request.product_code, transactionId=verified.transaction_id, creditsAdded=0, newBalance=int(account.balance), ledgerEventId=existing.ledger_event_id or "", ) if existing.user_id != user_id: raise ApiProblemError( status_code=409, detail=problem_payload( code="PAYMENT_TRANSACTION_CONFLICT", detail="Transaction belongs to another user", ), ) if existing.status in ("refunded", "refunded_insufficient", "revoked"): raise ApiProblemError( status_code=409, detail=problem_payload( code="PAYMENT_TRANSACTION_REVOKED", detail="Transaction has been refunded or revoked", ), ) is_starter = product_mapping.type == "starter" normalized_email = user_email.strip().lower() email_hash = ( self._build_email_hash(normalized_email) if normalized_email else None ) if is_starter: if not email_hash: raise ApiProblemError( status_code=422, detail=problem_payload( code="PAYMENT_STARTER_PACK_INELIGIBLE", detail="Email required for starter pack purchase", ), ) claim = await self._payment_repo.get_register_bonus_claim( email_hash=email_hash ) if claim is not None and claim.has_purchased_starter_pack: raise ApiProblemError( status_code=409, detail=problem_payload( code="PAYMENT_STARTER_PACK_INELIGIBLE", detail="Starter pack already purchased for this email", ), ) transaction_record = AppleIapTransaction( id=uuid4(), user_id=user_id, product_code=request.product_code, app_store_product_id=product_mapping.app_store_product_id, transaction_id=verified.transaction_id, original_transaction_id=verified.original_transaction_id, web_order_line_item_id=verified.web_order_line_item_id, environment=verified.environment, bundle_id=verified.bundle_id, app_account_token=request.app_account_token, purchase_date=str(verified.purchase_date), revocation_date=( str(verified.revocation_date) if verified.revocation_date else None ), status="verified", credits=product_mapping.credits, currency=None, price_milliunits=None, signed_transaction_info=request.signed_transaction_info, apple_payload_json=verified.raw_payload, ) await self._payment_repo.insert_transaction(transaction=transaction_record) account = await self._payment_repo.get_or_create_user_points_for_update( user_id=user_id ) credits = product_mapping.credits event_id = f"payment.apple_iap:{verified.transaction_id}" balance = int(account.balance) new_balance = balance + credits account.balance = new_balance account.lifetime_earned = int(account.lifetime_earned) + credits account.version = int(account.version) + 1 metadata = PurchaseLedgerMetadata( operator_type=PointsOperatorType.SYSTEM, run_id=event_id, ext={ "source": "apple_iap", "platform": "ios", "product_code": request.product_code, "app_store_product_id": product_mapping.app_store_product_id, "transaction_id": verified.transaction_id, "original_transaction_id": verified.original_transaction_id, "environment": verified.environment, "apple_iap_transaction_id": str(transaction_record.id), }, ) ledger_command = ApplyPointsChangeCommand( user_id=user_id, change_type=PointsChangeType.PURCHASE, biz_type=PointsBizType.PAYMENT, biz_id=transaction_record.id, event_id=event_id, amount=credits, direction=1, operator_id=None, metadata=metadata, ) await self._points_repo.append_ledger( command=ledger_command, balance_after=new_balance, ) transaction_record.status = "granted" transaction_record.ledger_event_id = event_id if is_starter and email_hash and normalized_email: _ = await self._payment_repo.upsert_register_bonus_claim_for_starter_pack( email_hash=email_hash, user_email_snapshot=normalized_email, first_user_id_snapshot=user_id, ) return VerifyTransactionResponse( status="granted", productCode=request.product_code, transactionId=verified.transaction_id, creditsAdded=credits, newBalance=new_balance, ledgerEventId=event_id, ) @staticmethod def _build_email_hash(normalized_email: str) -> str: key = config.points_policy.register_bonus_hmac_key.get_secret_value().strip() digest = hmac.new( key=key.encode("utf-8"), msg=normalized_email.encode("utf-8"), digestmod=hashlib.sha256, ) return digest.hexdigest() async def process_refund( self, *, transaction_id: str, refund_reason: str = "CUSTOMER_REQUEST", ) -> None: txn = await self._payment_repo.get_transaction_by_transaction_id( transaction_id=transaction_id ) if txn is None: logger.warning("Refund requested for unknown transaction: %s", transaction_id) return if txn.status not in ("granted",): logger.info( "Refund skipped: transaction %s status=%s", transaction_id, txn.status, ) return user_id = txn.user_id credits = txn.credits account = await self._payment_repo.get_user_points_for_update(user_id=user_id) if account is None: logger.warning( "Refund failed: no user_points for user %s on transaction %s", user_id, transaction_id, ) txn.status = "failed" txn.failure_code = "USER_POINTS_NOT_FOUND" return balance = int(account.balance) if balance < credits: refund_amount = balance txn.status = "refunded_insufficient" txn.failure_code = "INSUFFICIENT_BALANCE" logger.warning( "Refund insufficient balance: user=%s credits=%d balance=%d txn=%s", user_id, credits, balance, transaction_id, ) else: refund_amount = credits txn.status = "refunded" new_balance = balance - refund_amount account.balance = new_balance account.lifetime_earned = int(account.lifetime_earned) - refund_amount account.version = int(account.version) + 1 refund_event_id = f"refund.apple_iap:{transaction_id}" original_event_id = txn.ledger_event_id or f"payment.apple_iap:{transaction_id}" metadata = PurchaseLedgerMetadata( operator_type=PointsOperatorType.SYSTEM, run_id=refund_event_id, ext={ "source": "apple_iap", "platform": "ios", "product_code": txn.product_code, "app_store_product_id": txn.app_store_product_id, "transaction_id": transaction_id, "original_transaction_id": txn.original_transaction_id or "", "environment": txn.environment, "apple_iap_transaction_id": str(txn.id), "original_event_id": original_event_id, "refund_reason": refund_reason, "overdue_amount": credits - refund_amount, }, ) if refund_amount > 0: ledger_command = ApplyPointsChangeCommand( user_id=user_id, change_type=PointsChangeType.REFUND, biz_type=PointsBizType.PAYMENT, biz_id=txn.id, event_id=refund_event_id, amount=refund_amount, direction=-1, operator_id=None, metadata=metadata, ) await self._points_repo.append_ledger( command=ledger_command, balance_after=new_balance, ) txn.ledger_event_id = refund_event_id logger.info( "Refund processed: txn=%s user=%s refund_amount=%d new_balance=%d status=%s", transaction_id, user_id, refund_amount, new_balance, txn.status, ) async def handle_server_notification(self, *, signed_payload: str) -> None: if not signed_payload: logger.warning("Empty Apple server notification payload") return try: import jwt as pyjwt parts = signed_payload.split(".") if len(parts) < 2: logger.warning("Malformed Apple notification signed_payload") return payload_bytes = parts[1] + "=" * (-len(parts[1]) % 4) import base64 decoded = base64.urlsafe_b64decode(payload_bytes) import json notification_data: Any = json.loads(decoded) except Exception: logger.exception("Failed to decode Apple server notification payload") return notification_type = str(notification_data.get("notificationType", "")) subtype = notification_data.get("subtype") signed_transaction = notification_data.get("data", {}).get( "signedTransactionInfo", "" ) transaction_id: str | None = None if signed_transaction: try: txn_parts = signed_transaction.split(".") if len(txn_parts) >= 2: txn_payload_bytes = txn_parts[1] + "=" * (-len(txn_parts[1]) % 4) txn_decoded = base64.urlsafe_b64decode(txn_payload_bytes) txn_data: Any = json.loads(txn_decoded) transaction_id = str(txn_data.get("transactionId", "")) except Exception: logger.exception("Failed to decode signed transaction from notification") logger.info( "Apple notification received: type=%s subtype=%s transaction_id=%s", notification_type, subtype, transaction_id, ) refund_types = {"REFUND", "REVOKE", "DID_FAIL_TO_RENEW"} if notification_type in refund_types and transaction_id: refund_reason = notification_type if subtype: refund_reason = f"{notification_type}:{subtype}" await self.process_refund( transaction_id=transaction_id, refund_reason=refund_reason, ) return if notification_type == "DID_RENEW" and transaction_id: logger.info( "Apple DID_RENEW for transaction %s, no action needed", transaction_id, ) return logger.info( "Apple notification type=%s not handled, skipped", notification_type, )