# app/modules/loyalty/models/loyalty_card.py """ Loyalty card database model. Merchant-based loyalty cards: - Cards belong to a Merchant's loyalty program - Customers can earn and redeem at any store within the merchant - Tracks where customer enrolled for analytics Represents a customer's loyalty card (PassObject) that tracks: - Stamp count and history - Points balance and history - Wallet integration (Google/Apple pass IDs) - QR code/barcode for scanning """ import secrets from datetime import UTC, datetime from sqlalchemy import ( Boolean, CheckConstraint, Column, DateTime, ForeignKey, Index, Integer, String, ) from sqlalchemy.orm import relationship from app.core.database import Base from models.database.base import SoftDeleteMixin, TimestampMixin def generate_card_number() -> str: """Generate a unique 12-digit card number formatted as XXXX-XXXX-XXXX.""" digits = "".join([str(secrets.randbelow(10)) for _ in range(12)]) return f"{digits[:4]}-{digits[4:8]}-{digits[8:]}" def generate_qr_code_data() -> str: """Generate unique QR code data (URL-safe token).""" return secrets.token_urlsafe(24) def generate_apple_auth_token() -> str: """Generate Apple Wallet authentication token.""" return secrets.token_urlsafe(32) class LoyaltyCard(Base, TimestampMixin, SoftDeleteMixin): """ Customer's loyalty card (PassObject). Card belongs to a Merchant's loyalty program. The customer can earn and redeem at any store within the merchant. Links a customer to a merchant's loyalty program and tracks: - Stamps and points balances - Wallet pass integration - Activity timestamps """ __tablename__ = "loyalty_cards" id = Column(Integer, primary_key=True, index=True) # Merchant association (card belongs to merchant's program) merchant_id = Column( Integer, ForeignKey("merchants.id", ondelete="CASCADE"), nullable=False, index=True, comment="Merchant whose program this card belongs to", ) # Customer and program relationships customer_id = Column( Integer, ForeignKey("customers.id", ondelete="CASCADE"), nullable=False, index=True, ) program_id = Column( Integer, ForeignKey("loyalty_programs.id", ondelete="CASCADE"), nullable=False, index=True, ) # Track where customer enrolled (for analytics) enrolled_at_store_id = Column( Integer, ForeignKey("stores.id", ondelete="SET NULL"), nullable=True, index=True, comment="Store where customer enrolled (for analytics)", ) # ========================================================================= # Card Identification # ========================================================================= card_number = Column( String(20), unique=True, nullable=False, default=generate_card_number, index=True, comment="Human-readable card number (XXXX-XXXX-XXXX)", ) qr_code_data = Column( String(50), unique=True, nullable=False, default=generate_qr_code_data, index=True, comment="Data encoded in QR code for scanning", ) # ========================================================================= # Stamps Tracking # ========================================================================= stamp_count = Column( Integer, default=0, nullable=False, comment="Current stamps toward next reward", ) total_stamps_earned = Column( Integer, default=0, nullable=False, comment="Lifetime stamps earned", ) stamps_redeemed = Column( Integer, default=0, nullable=False, comment="Total rewards redeemed (stamps reset on redemption)", ) # ========================================================================= # Points Tracking # ========================================================================= points_balance = Column( Integer, default=0, nullable=False, comment="Current available points", ) total_points_earned = Column( Integer, default=0, nullable=False, comment="Lifetime points earned", ) points_redeemed = Column( Integer, default=0, nullable=False, comment="Lifetime points redeemed", ) total_points_voided = Column( Integer, default=0, nullable=False, comment="Lifetime points voided (returns + expirations)", ) # ========================================================================= # Wallet Integration # ========================================================================= # Google Wallet google_object_id = Column( String(200), nullable=True, index=True, comment="Google Wallet Loyalty Object ID", ) google_object_jwt = Column( String(2000), nullable=True, comment="JWT for Google Wallet 'Add to Wallet' button", ) # Apple Wallet apple_serial_number = Column( String(100), nullable=True, unique=True, index=True, comment="Apple Wallet pass serial number", ) apple_auth_token = Column( String(100), nullable=True, default=generate_apple_auth_token, comment="Apple Wallet authentication token for updates", ) # ========================================================================= # Activity Timestamps # ========================================================================= last_stamp_at = Column( DateTime(timezone=True), nullable=True, comment="Last stamp added (for cooldown)", ) last_points_at = Column( DateTime(timezone=True), nullable=True, comment="Last points earned (for expiration tracking)", ) last_redemption_at = Column( DateTime(timezone=True), nullable=True, comment="Last reward redemption", ) last_activity_at = Column( DateTime(timezone=True), nullable=True, comment="Any activity (for expiration calculation)", ) # ========================================================================= # Status # ========================================================================= is_active = Column( Boolean, default=True, nullable=False, index=True, ) # ========================================================================= # Relationships # ========================================================================= merchant = relationship("Merchant", backref="loyalty_cards") customer = relationship("Customer", backref="loyalty_cards") program = relationship("LoyaltyProgram", back_populates="cards") enrolled_at_store = relationship( "Store", backref="enrolled_loyalty_cards", ) transactions = relationship( "LoyaltyTransaction", back_populates="card", cascade="all, delete-orphan", order_by="desc(LoyaltyTransaction.created_at)", ) apple_devices = relationship( "AppleDeviceRegistration", back_populates="card", cascade="all, delete-orphan", ) # Indexes # One card per customer per store (always enforced at DB level). # Per-merchant uniqueness (when cross-location is enabled) is enforced # by the application layer in enroll_customer(). __table_args__ = ( Index("idx_loyalty_card_store_customer", "enrolled_at_store_id", "customer_id", unique=True), Index("idx_loyalty_card_merchant_customer", "merchant_id", "customer_id"), Index("idx_loyalty_card_merchant_active", "merchant_id", "is_active"), # Balances must never go negative — guards against direct SQL writes # bypassing the service layer's clamping logic. CheckConstraint("points_balance >= 0", name="ck_loyalty_cards_points_balance_nonneg"), CheckConstraint("stamp_count >= 0", name="ck_loyalty_cards_stamp_count_nonneg"), ) def __repr__(self) -> str: return f"" # ========================================================================= # Stamp Operations # ========================================================================= def add_stamp(self) -> bool: """ Add a stamp to the card. Returns True if this stamp completed a reward cycle. """ self.stamp_count += 1 self.total_stamps_earned += 1 self.last_stamp_at = datetime.now(UTC) self.last_activity_at = datetime.now(UTC) # Check if reward cycle is complete (handled by caller with program.stamps_target) return False # Caller checks against program.stamps_target def redeem_stamps(self, stamps_target: int) -> bool: """ Redeem stamps for a reward. Args: stamps_target: Number of stamps required for reward Returns True if redemption was successful. """ if self.stamp_count >= stamps_target: self.stamp_count -= stamps_target self.stamps_redeemed += 1 self.last_redemption_at = datetime.now(UTC) self.last_activity_at = datetime.now(UTC) return True return False # ========================================================================= # Points Operations # ========================================================================= def add_points(self, points: int) -> None: """Add points to the card.""" self.points_balance += points self.total_points_earned += points self.last_points_at = datetime.now(UTC) self.last_activity_at = datetime.now(UTC) def redeem_points(self, points: int) -> bool: """ Redeem points for a reward. Returns True if redemption was successful. """ if self.points_balance >= points: self.points_balance -= points self.points_redeemed += points self.last_redemption_at = datetime.now(UTC) self.last_activity_at = datetime.now(UTC) return True return False def void_points(self, points: int) -> None: """ Void points (for returns). Args: points: Number of points to void """ self.points_balance = max(0, self.points_balance - points) self.total_points_voided += points self.last_activity_at = datetime.now(UTC) def expire_points(self, points: int) -> None: """ Expire points due to inactivity. Args: points: Number of points to expire """ self.points_balance = max(0, self.points_balance - points) self.total_points_voided += points # ========================================================================= # Properties # ========================================================================= @property def stamps_until_reward(self) -> int | None: """Get stamps remaining until next reward (needs program context).""" # This should be calculated with program.stamps_target return None def can_stamp(self, cooldown_minutes: int) -> tuple[bool, str | None]: """ Check if card can receive a stamp (cooldown check). Args: cooldown_minutes: Minutes required between stamps Returns: (can_stamp, error_message) """ if not self.last_stamp_at: return True, None now = datetime.now(UTC) elapsed = (now - self.last_stamp_at).total_seconds() / 60 if elapsed < cooldown_minutes: remaining = int(cooldown_minutes - elapsed) return False, f"Please wait {remaining} minutes before next stamp" return True, None