# app/modules/loyalty/routes/api/platform.py """ Loyalty module platform routes. Platform endpoints for: - Customer enrollment (by store code) - Apple Wallet pass download - Apple Web Service endpoints for device registration/updates """ import logging from fastapi import APIRouter, Depends, Header, Path, Response from pydantic import BaseModel, Field from sqlalchemy.orm import Session from app.core.database import get_db from app.modules.loyalty.services import ( apple_wallet_service, card_service, program_service, ) logger = logging.getLogger(__name__) # Platform router (no auth required for some endpoints) platform_router = APIRouter(prefix="/loyalty") # ============================================================================= # Enrollment # ============================================================================= @platform_router.get("/programs/{store_code}") def get_program_by_store_code( store_code: str = Path(..., min_length=1, max_length=50), db: Session = Depends(get_db), ): """Get loyalty program info by store code (for enrollment page).""" # Find store by code (store_code or subdomain) store = program_service.get_store_by_code(db, store_code) # Get program (raises LoyaltyProgramNotFoundException if not found) program = program_service.require_active_program_by_store(db, store.id) return { "store_name": store.name, "store_code": store.store_code, "program": { "id": program.id, "type": program.loyalty_type, "name": program.display_name, "card_color": program.card_color, "logo_url": program.logo_url, "stamps_target": program.stamps_target if program.is_stamps_enabled else None, "stamps_reward": program.stamps_reward_description if program.is_stamps_enabled else None, "points_per_euro": program.points_per_euro if program.is_points_enabled else None, "terms_text": program.terms_text, "privacy_url": program.privacy_url, }, } # ============================================================================= # Apple Wallet Pass Download # ============================================================================= @platform_router.get("/passes/apple/{serial_number}.pkpass") def download_apple_pass( serial_number: str = Path(...), db: Session = Depends(get_db), ): """Download Apple Wallet pass for a card.""" # Find card by serial number (raises LoyaltyCardNotFoundException if not found) card = card_service.require_card_by_serial_number(db, serial_number) pass_data = apple_wallet_service.generate_pass_safe(db, card) return Response( content=pass_data, media_type="application/vnd.apple.pkpass", headers={ "Content-Disposition": f'attachment; filename="{serial_number}.pkpass"', }, ) # ============================================================================= # Apple Web Service Endpoints # (Required for Apple Wallet to register devices and get updates) # ============================================================================= class AppleRegisterDeviceRequest(BaseModel): """Request body for Apple device registration.""" push_token: str = Field(..., alias="pushToken") @platform_router.post("/apple/v1/devices/{device_id}/registrations/{pass_type_id}/{serial_number}") def register_device( body: AppleRegisterDeviceRequest, device_id: str = Path(...), pass_type_id: str = Path(...), serial_number: str = Path(...), authorization: str | None = Header(None), db: Session = Depends(get_db), ): """ Register a device for push notifications. Called by Apple when user adds pass to wallet. """ # Find card (raises LoyaltyCardNotFoundException if not found) card = card_service.require_card_by_serial_number(db, serial_number) # Verify auth token (raises InvalidAppleAuthTokenException if invalid) apple_wallet_service.verify_auth_token(card, authorization) apple_wallet_service.register_device_safe(db, card, device_id, body.pushToken) return Response(status_code=201) @platform_router.delete("/apple/v1/devices/{device_id}/registrations/{pass_type_id}/{serial_number}") def unregister_device( device_id: str = Path(...), pass_type_id: str = Path(...), serial_number: str = Path(...), authorization: str | None = Header(None), db: Session = Depends(get_db), ): """ Unregister a device. Called by Apple when user removes pass from wallet. """ # Find card (raises LoyaltyCardNotFoundException if not found) card = card_service.require_card_by_serial_number(db, serial_number) # Verify auth token (raises InvalidAppleAuthTokenException if invalid) apple_wallet_service.verify_auth_token(card, authorization) apple_wallet_service.unregister_device_safe(db, card, device_id) return Response(status_code=200) @platform_router.get("/apple/v1/devices/{device_id}/registrations/{pass_type_id}") def get_serial_numbers( device_id: str = Path(...), pass_type_id: str = Path(...), passesUpdatedSince: str | None = None, db: Session = Depends(get_db), ): """ Get list of pass serial numbers to update. Called by Apple to check for updated passes. """ # Get cards registered to this device, optionally filtered by update time cards = apple_wallet_service.get_updated_cards_for_device( db, device_id, updated_since=passesUpdatedSince ) if not cards: return Response(status_code=204) # Return serial numbers serial_numbers = [card.apple_serial_number for card in cards if card.apple_serial_number] last_updated = max(card.updated_at for card in cards) return { "serialNumbers": serial_numbers, "lastUpdated": last_updated.isoformat(), } @platform_router.get("/apple/v1/passes/{pass_type_id}/{serial_number}") def get_latest_pass( pass_type_id: str = Path(...), serial_number: str = Path(...), authorization: str | None = Header(None), db: Session = Depends(get_db), ): """ Get the latest version of a pass. Called by Apple to fetch updated pass data. """ # Find card (raises LoyaltyCardNotFoundException if not found) card = card_service.require_card_by_serial_number(db, serial_number) # Verify auth token (raises InvalidAppleAuthTokenException if invalid) apple_wallet_service.verify_auth_token(card, authorization) pass_data = apple_wallet_service.generate_pass_safe(db, card) return Response( content=pass_data, media_type="application/vnd.apple.pkpass", headers={ "Last-Modified": card.updated_at.strftime("%a, %d %b %Y %H:%M:%S GMT"), }, ) @platform_router.post("/apple/v1/log") def log_errors(): """ Receive error logs from Apple. Apple sends error logs here when there are issues with passes. """ # Just acknowledge - in production you'd log these return Response(status_code=200)