feat(loyalty): implement complete loyalty module MVP
Add stamp-based and points-based loyalty programs for vendors with: Database Models (5 tables): - loyalty_programs: Vendor program configuration - loyalty_cards: Customer cards with stamp/point balances - loyalty_transactions: Immutable audit log - staff_pins: Fraud prevention PINs (bcrypt hashed) - apple_device_registrations: Apple Wallet push tokens Services: - program_service: Program CRUD and statistics - card_service: Customer enrollment and card lookup - stamp_service: Stamp operations with anti-fraud checks - points_service: Points earning and redemption - pin_service: Staff PIN management with lockout - wallet_service: Unified wallet abstraction - google_wallet_service: Google Wallet API integration - apple_wallet_service: Apple Wallet .pkpass generation API Routes: - Admin: /api/v1/admin/loyalty/* (programs list, stats) - Vendor: /api/v1/vendor/loyalty/* (stamp, points, cards, PINs) - Public: /api/v1/loyalty/* (enrollment, Apple Web Service) Anti-Fraud Features: - Staff PIN verification (configurable per program) - Cooldown period between stamps (default 15 min) - Daily stamp limits (default 5/day) - PIN lockout after failed attempts Wallet Integration: - Google Wallet: LoyaltyClass and LoyaltyObject management - Apple Wallet: .pkpass generation with PKCS#7 signing - Apple Web Service endpoints for device registration/updates Also includes: - Alembic migration for all tables with indexes - Localization files (en, fr, de, lu) - Module documentation - Phase 2 interface and user journey plan Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
205
app/modules/loyalty/models/staff_pin.py
Normal file
205
app/modules/loyalty/models/staff_pin.py
Normal file
@@ -0,0 +1,205 @@
|
||||
# app/modules/loyalty/models/staff_pin.py
|
||||
"""
|
||||
Staff PIN database model.
|
||||
|
||||
Provides fraud prevention by requiring staff to authenticate
|
||||
before performing stamp/points operations. Includes:
|
||||
- Secure PIN hashing with bcrypt
|
||||
- Failed attempt tracking
|
||||
- Automatic lockout after too many failures
|
||||
"""
|
||||
|
||||
from datetime import UTC, datetime
|
||||
|
||||
import bcrypt
|
||||
from sqlalchemy import (
|
||||
Boolean,
|
||||
Column,
|
||||
DateTime,
|
||||
ForeignKey,
|
||||
Index,
|
||||
Integer,
|
||||
String,
|
||||
)
|
||||
from sqlalchemy.orm import relationship
|
||||
|
||||
from app.core.database import Base
|
||||
from models.database.base import TimestampMixin
|
||||
|
||||
|
||||
class StaffPin(Base, TimestampMixin):
|
||||
"""
|
||||
Staff PIN for loyalty operations.
|
||||
|
||||
Each staff member can have their own PIN to authenticate
|
||||
stamp/points operations. PINs are hashed with bcrypt and
|
||||
include lockout protection.
|
||||
"""
|
||||
|
||||
__tablename__ = "staff_pins"
|
||||
|
||||
id = Column(Integer, primary_key=True, index=True)
|
||||
|
||||
# Relationships
|
||||
program_id = Column(
|
||||
Integer,
|
||||
ForeignKey("loyalty_programs.id", ondelete="CASCADE"),
|
||||
nullable=False,
|
||||
index=True,
|
||||
)
|
||||
vendor_id = Column(
|
||||
Integer,
|
||||
ForeignKey("vendors.id", ondelete="CASCADE"),
|
||||
nullable=False,
|
||||
index=True,
|
||||
comment="Denormalized for query performance",
|
||||
)
|
||||
|
||||
# Staff identity
|
||||
name = Column(
|
||||
String(100),
|
||||
nullable=False,
|
||||
comment="Staff member name",
|
||||
)
|
||||
staff_id = Column(
|
||||
String(50),
|
||||
nullable=True,
|
||||
index=True,
|
||||
comment="Optional staff ID/employee number",
|
||||
)
|
||||
|
||||
# PIN authentication
|
||||
pin_hash = Column(
|
||||
String(255),
|
||||
nullable=False,
|
||||
comment="bcrypt hash of PIN",
|
||||
)
|
||||
|
||||
# Security tracking
|
||||
failed_attempts = Column(
|
||||
Integer,
|
||||
default=0,
|
||||
nullable=False,
|
||||
comment="Consecutive failed PIN attempts",
|
||||
)
|
||||
locked_until = Column(
|
||||
DateTime(timezone=True),
|
||||
nullable=True,
|
||||
comment="Lockout expires at this time",
|
||||
)
|
||||
last_used_at = Column(
|
||||
DateTime(timezone=True),
|
||||
nullable=True,
|
||||
comment="Last successful use of PIN",
|
||||
)
|
||||
|
||||
# Status
|
||||
is_active = Column(
|
||||
Boolean,
|
||||
default=True,
|
||||
nullable=False,
|
||||
index=True,
|
||||
)
|
||||
|
||||
# =========================================================================
|
||||
# Relationships
|
||||
# =========================================================================
|
||||
program = relationship("LoyaltyProgram", back_populates="staff_pins")
|
||||
vendor = relationship("Vendor", backref="staff_pins")
|
||||
|
||||
# Indexes
|
||||
__table_args__ = (
|
||||
Index("idx_staff_pin_vendor_active", "vendor_id", "is_active"),
|
||||
Index("idx_staff_pin_program_active", "program_id", "is_active"),
|
||||
)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"<StaffPin(id={self.id}, name='{self.name}', active={self.is_active})>"
|
||||
|
||||
# =========================================================================
|
||||
# PIN Operations
|
||||
# =========================================================================
|
||||
|
||||
def set_pin(self, plain_pin: str) -> None:
|
||||
"""
|
||||
Hash and store a PIN.
|
||||
|
||||
Args:
|
||||
plain_pin: The plain text PIN (typically 4-6 digits)
|
||||
"""
|
||||
salt = bcrypt.gensalt()
|
||||
self.pin_hash = bcrypt.hashpw(plain_pin.encode("utf-8"), salt).decode("utf-8")
|
||||
|
||||
def verify_pin(self, plain_pin: str) -> bool:
|
||||
"""
|
||||
Verify a PIN against the stored hash.
|
||||
|
||||
Args:
|
||||
plain_pin: The plain text PIN to verify
|
||||
|
||||
Returns:
|
||||
True if PIN matches, False otherwise
|
||||
"""
|
||||
if not self.pin_hash:
|
||||
return False
|
||||
return bcrypt.checkpw(plain_pin.encode("utf-8"), self.pin_hash.encode("utf-8"))
|
||||
|
||||
# =========================================================================
|
||||
# Lockout Management
|
||||
# =========================================================================
|
||||
|
||||
@property
|
||||
def is_locked(self) -> bool:
|
||||
"""Check if PIN is currently locked out."""
|
||||
if not self.locked_until:
|
||||
return False
|
||||
return datetime.now(UTC) < self.locked_until
|
||||
|
||||
def record_failed_attempt(self, max_attempts: int = 5, lockout_minutes: int = 30) -> bool:
|
||||
"""
|
||||
Record a failed PIN attempt.
|
||||
|
||||
Args:
|
||||
max_attempts: Maximum failed attempts before lockout
|
||||
lockout_minutes: Duration of lockout in minutes
|
||||
|
||||
Returns:
|
||||
True if account is now locked
|
||||
"""
|
||||
self.failed_attempts += 1
|
||||
|
||||
if self.failed_attempts >= max_attempts:
|
||||
from datetime import timedelta
|
||||
|
||||
self.locked_until = datetime.now(UTC) + timedelta(minutes=lockout_minutes)
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def record_success(self) -> None:
|
||||
"""Record a successful PIN verification."""
|
||||
self.failed_attempts = 0
|
||||
self.locked_until = None
|
||||
self.last_used_at = datetime.now(UTC)
|
||||
|
||||
def unlock(self) -> None:
|
||||
"""Manually unlock a PIN (admin action)."""
|
||||
self.failed_attempts = 0
|
||||
self.locked_until = None
|
||||
|
||||
# =========================================================================
|
||||
# Properties
|
||||
# =========================================================================
|
||||
|
||||
@property
|
||||
def remaining_attempts(self) -> int:
|
||||
"""Get remaining attempts before lockout (assuming max 5)."""
|
||||
return max(0, 5 - self.failed_attempts)
|
||||
|
||||
@property
|
||||
def lockout_remaining_seconds(self) -> int | None:
|
||||
"""Get seconds remaining in lockout, or None if not locked."""
|
||||
if not self.locked_until:
|
||||
return None
|
||||
remaining = (self.locked_until - datetime.now(UTC)).total_seconds()
|
||||
return max(0, int(remaining))
|
||||
Reference in New Issue
Block a user