phil/sneakyklaus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: initial project setup

Browse Source 
Initialize Sneaky Klaus project with:
- uv package management and pyproject.toml
- Flask application structure (app.py, config.py)
- SQLAlchemy models for Admin and Exchange
- Alembic database migrations
- Pre-commit hooks configuration
- Development tooling (pytest, ruff, mypy)

Initial structure follows design documents in docs/:
- src/app.py: Application factory with Flask extensions
- src/config.py: Environment-based configuration
- src/models/: Admin and Exchange models
- migrations/: Alembic migration setup

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Phil Skentelbery
2025-12-22 11:28:15 -07:00
commit b077112aba
32 changed files with 10931 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

862
docs/designs/v0.1.0/components/auth.md Normal file
View File

@@ -0,0 +1,862 @@
# Authentication - v0.1.0
**Version**: 0.1.0
**Date**: 2025-12-22
**Status**: Initial Design
## Overview
This document defines the authentication and authorization flows for Sneaky Klaus. The system supports two types of users with different authentication mechanisms:
1. **Admin**: Single administrator account with password-based authentication
2. **Participants**: Passwordless authentication via magic links
Session management is handled by Flask-Session, which stores session data server-side in SQLite.
## Authentication Flows
### Initial Admin Setup Flow
The first-run experience for new Sneaky Klaus installations.
#### Sequence Diagram
```mermaid
sequenceDiagram
participant User
participant Browser
participant App as Flask App
participant DB as SQLite Database
participant Session as Flask-Session
User->>Browser: Navigate to application
Browser->>App: GET /
App->>DB: Check if admin exists
DB-->>App: No admin found
App->>Browser: Redirect to /setup
Browser->>App: GET /setup
App-->>Browser: Render setup form
User->>Browser: Fill form (email, password)
Browser->>App: POST /setup
App->>App: Validate form (Flask-WTF)
App->>App: Hash password (bcrypt)
App->>DB: INSERT INTO admin
DB-->>App: Admin created (id=1)
App->>Session: Create admin session
Session-->>App: Session ID
App->>Browser: Set session cookie
App->>Browser: Flash success message
App->>Browser: Redirect to /admin/dashboard
Browser->>App: GET /admin/dashboard
App->>Session: Validate session
Session-->>App: Admin authenticated
App-->>Browser: Render dashboard
Note over Browser,App: Success message auto-dismisses after 5 seconds
```
#### Implementation Details
**First-Run Detection**:
- Check `SELECT COUNT(*) FROM admin` on application startup
- If count == 0, set `app.config['REQUIRES_SETUP'] = True`
- Before request handler checks flag and redirects to `/setup` if True
**Route**: `/setup`
**Methods**: GET, POST
**Authorization**:
- Accessible only when no admin exists
- Returns 404 if admin already exists
**Form Fields** (Flask-WTF):
- `email`: EmailField, required, validated with email validator
- `password`: PasswordField, required, min length 12 characters
- `password_confirm`: PasswordField, required, must match password
- CSRF token (automatic via Flask-WTF)
**POST Workflow**:
1. Validate form with Flask-WTF
2. Double-check no admin exists (prevent race condition)
3. Hash password with bcrypt (cost factor 12)
4. Insert admin record
5. Create session (Flask-Session)
6. Set session cookie (HttpOnly, Secure, SameSite=Lax)
7. Flash success message: "Admin account created successfully!"
8. Redirect to `/admin/dashboard`
**Auto-login**:
- Session created immediately after admin creation
- No separate login step required
- User automatically authenticated
---
### Admin Login Flow
Standard password-based authentication for the admin user.
#### Sequence Diagram
```mermaid
sequenceDiagram
participant User
participant Browser
participant App as Flask App
participant RateLimit as Rate Limiter
participant DB as SQLite Database
participant Session as Flask-Session
User->>Browser: Navigate to /admin/login
Browser->>App: GET /admin/login
App-->>Browser: Render login form
User->>Browser: Fill form (email, password)
Browser->>App: POST /admin/login
App->>App: Validate form (Flask-WTF)
App->>RateLimit: Check rate limit for email
alt Rate limit exceeded
RateLimit-->>App: Too many attempts
App->>Browser: Flash error message
App->>Browser: Render login form (429 status)
Note over Browser: Error message persists until dismissed
else Within rate limit
RateLimit-->>App: Allowed
App->>DB: SELECT * FROM admin WHERE email = ?
DB-->>App: Admin record
alt Invalid credentials
App->>App: Verify password (bcrypt)
App->>RateLimit: Increment failure count
App->>Browser: Flash error message
App->>Browser: Render login form
Note over Browser: Error message persists until dismissed
else Valid credentials
App->>App: Verify password (bcrypt)
App->>RateLimit: Reset rate limit counter
App->>Session: Create admin session
Session-->>App: Session ID
App->>Browser: Set session cookie
App->>Browser: Flash success message
App->>Browser: Redirect to /admin/dashboard
Browser->>App: GET /admin/dashboard
App->>Session: Validate session
Session-->>App: Admin authenticated
App-->>Browser: Render dashboard
Note over Browser: Success message auto-dismisses after 5 seconds
end
end
```
#### Implementation Details
**Route**: `/admin/login`
**Methods**: GET, POST
**Authorization**:
- Accessible to unauthenticated users only
- Redirects to dashboard if already authenticated
**Form Fields** (Flask-WTF):
- `email`: EmailField, required
- `password`: PasswordField, required
- `remember_me`: BooleanField, optional (extends session duration)
- CSRF token (automatic via Flask-WTF)
**Rate Limiting**:
- **Policy**: 5 attempts per 15 minutes per email
- **Key**: `login:admin:{email_lowercase}`
- **Implementation**: Check `rate_limit` table before authentication
- **Failure Handling**: Increment attempt counter on failed login
- **Success Handling**: Reset counter on successful login
- **Lockout Message**: "Too many login attempts. Please try again in {minutes} minutes."
**POST Workflow**:
1. Validate form with Flask-WTF
2. Normalize email to lowercase
3. Check rate limit for `login:admin:{email}`
4. If rate limited: flash error, return 429
5. Query admin by email
6. If admin not found or password invalid:
- Increment rate limit counter
- Flash error: "Invalid email or password"
- Re-render form
7. If credentials valid:
- Reset rate limit counter
- Create session via Flask-Session
- Set session expiration based on `remember_me`:
- Checked: 30 days
- Unchecked: 7 days (default)
- Set session cookie
- Flash success: "Welcome back!"
- Redirect to `/admin/dashboard`
**Session Cookie Configuration**:
- `HttpOnly`: True (prevent JavaScript access)
- `Secure`: True (HTTPS only in production)
- `SameSite`: Lax (CSRF protection)
- `Max-Age`: Based on remember_me
---
### Admin Logout Flow
Terminates the admin session.
#### Sequence Diagram
```mermaid
sequenceDiagram
participant User
participant Browser
participant App as Flask App
participant Session as Flask-Session
User->>Browser: Click logout
Browser->>App: POST /admin/logout
App->>Session: Get current session ID
App->>Session: Delete session
Session-->>App: Session deleted
App->>Browser: Clear session cookie
App->>Browser: Flash success message
App->>Browser: Redirect to /
Note over Browser: Success message auto-dismisses after 5 seconds
```
#### Implementation Details
**Route**: `/admin/logout`
**Methods**: POST (GET redirects to dashboard)
**Authorization**: Requires active admin session
**POST Workflow**:
1. Validate CSRF token
2. Delete session from Flask-Session store
3. Clear session cookie
4. Flash success: "You have been logged out"
5. Redirect to `/`
---
### Participant Magic Link Authentication Flow
Passwordless authentication for participants using time-limited magic links.
#### Sequence Diagram
```mermaid
sequenceDiagram
participant Participant
participant Browser
participant App as Flask App
participant RateLimit as Rate Limiter
participant DB as SQLite Database
participant Email as Resend
participant Session as Flask-Session
Note over Participant,Browser: Participant requests access
Participant->>Browser: Navigate to /exchange/{slug}
Browser->>App: GET /exchange/{slug}
App->>DB: SELECT * FROM exchange WHERE slug = ?
DB-->>App: Exchange record
App-->>Browser: Render participant login form
Participant->>Browser: Enter email
Browser->>App: POST /exchange/{slug}/auth
App->>App: Validate form (Flask-WTF)
App->>RateLimit: Check rate limit for email
alt Rate limit exceeded
RateLimit-->>App: Too many requests
App->>Browser: Flash error message
App->>Browser: Render form (429 status)
Note over Browser: Error message persists until dismissed
else Within rate limit
RateLimit-->>App: Allowed
App->>DB: SELECT * FROM participant WHERE email = ? AND exchange_id = ?
alt Participant not found
App->>Browser: Flash error message
App->>Browser: Render form
Note over Browser: Generic error for security
else Participant found
DB-->>App: Participant record
App->>App: Generate magic token (32 bytes, secrets module)
App->>App: Hash token (SHA-256)
App->>DB: INSERT INTO magic_token
App->>Email: Send magic link email
Email-->>Participant: Email with magic link
App->>RateLimit: Increment request counter
App->>Browser: Flash success message
App->>Browser: Redirect to /exchange/{slug}/auth/sent
Note over Browser: Success message auto-dismisses after 5 seconds
end
end
Note over Participant,Email: Participant clicks magic link
Participant->>Email: Click link
Email->>Browser: Open /exchange/{slug}/auth/verify?token={token}
Browser->>App: GET /exchange/{slug}/auth/verify?token={token}
App->>App: Hash token
App->>DB: SELECT * FROM magic_token WHERE token_hash = ?
alt Token invalid/expired/used
DB-->>App: No token found / Token expired
App->>Browser: Flash error message
App->>Browser: Redirect to /exchange/{slug}
Note over Browser: Error message persists until dismissed
else Token valid
DB-->>App: Token record
App->>DB: UPDATE magic_token SET used_at = NOW()
App->>DB: SELECT * FROM participant WHERE id = ?
DB-->>App: Participant record
App->>Session: Create participant session
Session-->>App: Session ID
App->>Browser: Set session cookie
App->>Browser: Flash success message
App->>Browser: Redirect to /exchange/{slug}/dashboard
Browser->>App: GET /exchange/{slug}/dashboard
App->>Session: Validate session
Session-->>App: Participant authenticated
App-->>Browser: Render participant dashboard
Note over Browser: Success message auto-dismisses after 5 seconds
end
```
#### Implementation Details
**Route**: `/exchange/{slug}/auth`
**Methods**: GET, POST
**Authorization**: Public (unauthenticated)
**Form Fields** (Flask-WTF):
- `email`: EmailField, required
- CSRF token (automatic via Flask-WTF)
**Rate Limiting**:
- **Policy**: 3 requests per hour per email
- **Key**: `magic_link:{email_lowercase}`
- **Lockout Message**: "Too many magic link requests. Please try again in {minutes} minutes."
**POST Workflow** (Request Magic Link):
1. Validate form with Flask-WTF
2. Normalize email to lowercase
3. Check rate limit for `magic_link:{email}`
4. If rate limited: flash error, return 429
5. Query participant by email and exchange_id
6. If participant not found:
- Flash generic error: "If this email is registered, you will receive a magic link."
- Return success response (prevent email enumeration)
- Do NOT send email
7. If participant found but withdrawn:
- Same as not found (prevent information disclosure)
8. If participant found and active:
- Generate token: 32 bytes from `secrets.token_urlsafe()`
- Hash token: SHA-256
- Store hash in `magic_token` table with:
- `token_type`: 'magic_link'
- `email`: participant email
- `participant_id`: participant ID
- `exchange_id`: exchange ID
- `expires_at`: NOW() + 1 hour
- Send email with magic link
- Increment rate limit counter
- Flash success: "Check your email for a magic link!"
- Redirect to `/exchange/{slug}/auth/sent`
**Route**: `/exchange/{slug}/auth/verify`
**Methods**: GET
**Query Parameters**:
- `token`: Magic token (URL-safe base64)
**GET Workflow** (Verify Token):
1. Extract token from query string
2. Hash token with SHA-256
3. Query `magic_token` table by hash
4. Validate token:
- Exists in database
- `expires_at` > NOW()
- `used_at` IS NULL
- `token_type` = 'magic_link'
5. If invalid:
- Flash error: "This magic link is invalid or has expired."
- Redirect to `/exchange/{slug}`
6. If valid:
- Mark token as used: `UPDATE magic_token SET used_at = NOW()`
- Load participant record
- Create session via Flask-Session with:
- `user_id`: participant.id
- `user_type`: 'participant'
- `exchange_id`: exchange.id
- Set session cookie (7-day expiration)
- Flash success: "Welcome, {participant.name}!"
- Redirect to `/exchange/{slug}/dashboard`
**Token Generation**:
```python
import secrets
import hashlib
# Generate token
token = secrets.token_urlsafe(32) # 32 bytes = 43 URL-safe characters
# Hash for storage
token_hash = hashlib.sha256(token.encode()).hexdigest()
# Magic link URL
magic_link = f"{base_url}/exchange/{slug}/auth/verify?token={token}"
```
---
### Participant Logout Flow
Terminates the participant session.
#### Sequence Diagram
```mermaid
sequenceDiagram
participant Participant
participant Browser
participant App as Flask App
participant Session as Flask-Session
Participant->>Browser: Click logout
Browser->>App: POST /exchange/{slug}/logout
App->>Session: Get current session ID
App->>Session: Delete session
Session-->>App: Session deleted
App->>Browser: Clear session cookie
App->>Browser: Flash success message
App->>Browser: Redirect to /exchange/{slug}
Note over Browser: Success message auto-dismisses after 5 seconds
```
#### Implementation Details
**Route**: `/exchange/{slug}/logout`
**Methods**: POST
**Authorization**: Requires active participant session for this exchange
**POST Workflow**:
1. Validate CSRF token
2. Verify session belongs to participant for this exchange
3. Delete session from Flask-Session store
4. Clear session cookie
5. Flash success: "You have been logged out"
6. Redirect to `/exchange/{slug}`
---
## Session Management
### Flask-Session Configuration
**Backend**: SQLAlchemy (SQLite)
**Table**: `sessions` (created and managed by Flask-Session)
**Configuration**:
```python
app.config['SESSION_TYPE'] = 'sqlalchemy'
app.config['SESSION_SQLALCHEMY'] = db # SQLAlchemy instance
app.config['SESSION_PERMANENT'] = True
app.config['SESSION_USE_SIGNER'] = True # Sign session cookies
app.config['SESSION_KEY_PREFIX'] = 'sk:' # Prefix for session keys
app.config['PERMANENT_SESSION_LIFETIME'] = timedelta(days=7) # Default
```
### Session Data Structure
**Admin Session**:
```python
{
'user_id': 1, # Admin ID
'user_type': 'admin',
'_fresh': True, # For Flask-Login compatibility (future)
'_permanent': True
}
```
**Participant Session**:
```python
{
'user_id': 123, # Participant ID
'user_type': 'participant',
'exchange_id': 456, # Exchange participant is authenticated for
'_fresh': True,
'_permanent': True
}
```
### Session Validation
**Before Request Handler**:
```python
@app.before_request
def load_session_user():
if 'user_id' in session and 'user_type' in session:
if session['user_type'] == 'admin':
g.admin = Admin.query.get(session['user_id'])
elif session['user_type'] == 'participant':
g.participant = Participant.query.get(session['user_id'])
g.exchange_id = session.get('exchange_id')
```
### Session Expiration
**Default Expiration**: 7 days from last activity
**Remember Me** (Admin only): 30 days from last activity
**Sliding Window**: Flask-Session automatically updates `last_activity` on each request
**Cleanup**: Flask-Session handles expired session cleanup automatically
---
## Flash Messages
Flash messages provide user feedback for authentication actions.
### Flash Message Types
**Success Messages**:
- Category: `'success'`
- Auto-dismiss: 5 seconds
- Examples:
- "Admin account created successfully!"
- "Welcome back!"
- "You have been logged out"
- "Check your email for a magic link!"
- "Welcome, {name}!"
**Error Messages**:
- Category: `'error'`
- Auto-dismiss: Manual (user must dismiss)
- Examples:
- "Invalid email or password"
- "Too many login attempts. Please try again in {minutes} minutes."
- "This magic link is invalid or has expired."
### Implementation
**Backend** (Flask):
```python
from flask import flash
# Success message
flash("Admin account created successfully!", "success")
# Error message
flash("Invalid email or password", "error")
```
**Frontend** (Jinja2 + JavaScript):
```html
{% with messages = get_flashed_messages(with_categories=true) %}
{% if messages %}
{% for category, message in messages %}
<div class="flash-message flash-{{ category }}"
{% if category == 'success' %}data-auto-dismiss="5000"{% endif %}>
{{ message }}
<button class="flash-dismiss" aria-label="Dismiss">&times;</button>
</div>
{% endfor %}
{% endif %}
{% endwith %}
<script>
// Auto-dismiss success messages after 5 seconds
document.querySelectorAll('[data-auto-dismiss]').forEach(el => {
const delay = parseInt(el.dataset.autoDismiss);
setTimeout(() => {
el.style.opacity = '0';
setTimeout(() => el.remove(), 300);
}, delay);
});
// Manual dismiss for all messages
document.querySelectorAll('.flash-dismiss').forEach(btn => {
btn.addEventListener('click', () => {
const message = btn.parentElement;
message.style.opacity = '0';
setTimeout(() => message.remove(), 300);
});
});
</script>
```
---
## Authorization Patterns
### Route Protection Decorators
**Admin-Only Routes**:
```python
from functools import wraps
from flask import session, redirect, url_for, flash, g
def admin_required(f):
@wraps(f)
def decorated_function(*args, **kwargs):
if 'user_id' not in session or session.get('user_type') != 'admin':
flash("You must be logged in as admin to access this page.", "error")
return redirect(url_for('admin_login'))
return f(*args, **kwargs)
return decorated_function
@app.route('/admin/dashboard')
@admin_required
def admin_dashboard():
return render_template('admin/dashboard.html')
```
**Participant-Only Routes**:
```python
def participant_required(f):
@wraps(f)
def decorated_function(slug, *args, **kwargs):
if 'user_id' not in session or session.get('user_type') != 'participant':
flash("You must be logged in to access this page.", "error")
return redirect(url_for('exchange_auth', slug=slug))
# Verify participant is authenticated for this exchange
exchange = Exchange.query.filter_by(slug=slug).first_or_404()
if session.get('exchange_id') != exchange.id:
flash("You are not authorized to access this exchange.", "error")
return redirect(url_for('exchange_auth', slug=slug))
return f(slug, *args, **kwargs)
return decorated_function
@app.route('/exchange/<slug>/dashboard')
@participant_required
def participant_dashboard(slug):
return render_template('participant/dashboard.html')
```
### Setup Requirement Check
**Before Request Handler**:
```python
@app.before_request
def check_setup_required():
# Skip check for setup route and static files
if request.endpoint in ['setup', 'static']:
return
# Check if admin exists
if app.config.get('REQUIRES_SETUP'):
admin_count = Admin.query.count()
if admin_count == 0:
return redirect(url_for('setup'))
else:
app.config['REQUIRES_SETUP'] = False
```
---
## Security Considerations
### Password Security
**Hashing Algorithm**: bcrypt with cost factor 12
**Minimum Password Length**: 12 characters
**Password Validation**:
- Enforced at form level (Flask-WTF)
- No complexity requirements (length is primary security measure)
**Implementation**:
```python
from flask_bcrypt import Bcrypt
bcrypt = Bcrypt(app)
# Hash password
password_hash = bcrypt.generate_password_hash(password).decode('utf-8')
# Verify password
bcrypt.check_password_hash(admin.password_hash, password)
```
### CSRF Protection
**Implementation**: Flask-WTF automatic CSRF protection
**All Forms**: Include CSRF token automatically via `{{ form.csrf_token }}`
**All POST Routes**: Validate CSRF token automatically via Flask-WTF
**Exempt Routes**: None (all state-changing operations require CSRF token)
### Rate Limiting
**Storage**: SQLite `rate_limit` table
**Policies**:
- Admin login: 5 attempts / 15 minutes / email
- Magic link request: 3 requests / hour / email
**Bypass**: None (applies to all users including admin)
**Cleanup**: Expired entries purged daily via background job
### Session Security
**Cookie Settings**:
- `HttpOnly`: True (prevent XSS)
- `Secure`: True in production (HTTPS only)
- `SameSite`: Lax (CSRF protection)
**Session Signing**: Enabled via `SESSION_USE_SIGNER`
**Secret Key**: Loaded from environment variable `SECRET_KEY`
### Magic Link Security
**Token Generation**: `secrets.token_urlsafe(32)` (cryptographically secure)
**Token Storage**: SHA-256 hash only (original token never stored)
**Token Expiration**: 1 hour from creation
**Single Use**: Marked as used immediately upon verification
**Email Enumeration Prevention**: Generic success message for all email submissions
### Timing Attack Prevention
**Password Verification**: bcrypt naturally resistant to timing attacks
**Magic Link Lookup**: Use constant-time comparison for token hashes (handled by database query)
---
## Error Handling
### Authentication Errors
**Invalid Credentials**:
- HTTP Status: 200 (re-render form)
- Flash Message: "Invalid email or password"
- Rate Limit: Increment counter
**Rate Limit Exceeded**:
- HTTP Status: 429 Too Many Requests
- Flash Message: "Too many {action} attempts. Please try again in {minutes} minutes."
- Behavior: Re-render form with error
**Expired Magic Link**:
- HTTP Status: 200 (redirect to login)
- Flash Message: "This magic link is invalid or has expired."
- Behavior: Redirect to exchange login page
**CSRF Validation Failure**:
- HTTP Status: 400 Bad Request
- Flash Message: "Security validation failed. Please try again."
- Behavior: Redirect to form
### Session Errors
**Session Expired**:
- HTTP Status: 302 (redirect to login)
- Flash Message: "Your session has expired. Please log in again."
- Behavior: Redirect to appropriate login page
**Invalid Session**:
- HTTP Status: 302 (redirect to login)
- Flash Message: "You must be logged in to access this page."
- Behavior: Redirect to appropriate login page
---
## Testing Considerations
### Unit Tests
**Setup Flow**:
- Test admin creation with valid data
- Test duplicate admin prevention
- Test auto-login after setup
- Test setup route 404 when admin exists
**Admin Login**:
- Test successful login with valid credentials
- Test failed login with invalid credentials
- Test rate limiting after 5 failed attempts
- Test rate limit reset after successful login
- Test remember_me session duration
**Magic Link**:
- Test magic link generation for valid participant
- Test rate limiting after 3 requests
- Test token verification with valid token
- Test token expiration after 1 hour
- Test single-use token enforcement
### Integration Tests
**Full Authentication Flows**:
- Test setup → login → logout flow
- Test magic link request → email → verify → dashboard flow
- Test concurrent session handling
- Test session persistence across requests
### Security Tests
**CSRF Protection**:
- Test all POST routes require valid CSRF token
- Test CSRF token validation
**Rate Limiting**:
- Test rate limits are enforced correctly
- Test rate limit window expiration
**Session Security**:
- Test session cookie settings (HttpOnly, Secure, SameSite)
- Test session expiration
---
## Future Enhancements
Potential improvements for future versions:
1. **Two-Factor Authentication**: TOTP for admin login
2. **Password Reset**: Admin password reset via email
3. **Session Management UI**: Admin view of active sessions
4. **Account Lockout**: Temporary lockout after repeated failed logins
5. **Audit Logging**: Track all authentication events
6. **Multiple Admins**: Support for multiple admin accounts with roles
7. **OAuth Integration**: Social login options for participants
These enhancements are out of scope for v0.1.0.
---
## References
- [Flask-Session Documentation](https://flask-session.readthedocs.io/)
- [Flask-WTF Documentation](https://flask-wtf.readthedocs.io/)
- [bcrypt Documentation](https://github.com/pyca/bcrypt/)
- [OWASP Authentication Cheat Sheet](https://cheatsheetseries.owasp.org/cheatsheets/Authentication_Cheat_Sheet.html)
- [ADR-0001: Core Technology Stack](../../decisions/0001-core-technology-stack.md)
- [ADR-0002: Authentication Strategy](../../decisions/0002-authentication-strategy.md)
- [Data Model v0.1.0](../data-model.md)

1017
docs/designs/v0.1.0/components/background-jobs.md Normal file
View File

File diff suppressed because it is too large Load Diff

732
docs/designs/v0.1.0/components/matching.md Normal file
View File

@@ -0,0 +1,732 @@
# Matching Component Design - v0.1.0
**Version**: 0.1.0
**Date**: 2025-12-22
**Status**: Initial Design
## Introduction
This document defines the Secret Santa matching algorithm for Sneaky Klaus. The matching algorithm is responsible for assigning each participant a recipient while respecting exclusion rules and following Secret Santa best practices.
## Requirements
### Functional Requirements
1. **One-to-One Assignment**: Each participant gives exactly one gift and receives exactly one gift
2. **No Self-Matching**: No participant is assigned to themselves
3. **Exclusion Compliance**: All exclusion rules must be honored
4. **Randomization**: Assignments must be unpredictable and fair
5. **Single Cycle Preferred**: When possible, create a single cycle (A→B→C→...→Z→A)
6. **Validation**: Detect impossible matching scenarios before attempting
### Non-Functional Requirements
1. **Performance**: Complete matching in <1 second for up to 100 participants
2. **Reliability**: Deterministic failure detection (no random timeouts)
3. **Transparency**: Clear error messages when matching fails
4. **Testability**: Algorithm must be unit-testable with reproducible results
## Algorithm Overview
The matching algorithm uses a **graph-based approach with randomized cycle generation**.
### High-Level Flow
```mermaid
flowchart TD
Start([Trigger Matching]) --> Validate[Validate Preconditions]
Validate -->|Invalid| Error1[Return Error: Validation Failed]
Validate -->|Valid| BuildGraph[Build Assignment Graph]
BuildGraph --> CheckFeasibility[Check Matching Feasibility]
CheckFeasibility -->|Impossible| Error2[Return Error: Impossible to Match]
CheckFeasibility -->|Possible| Attempt[Attempt to Find Valid Cycle]
Attempt --> MaxAttempts{Max attempts<br/>reached?}
MaxAttempts -->|Yes| Error3[Return Error: Could Not Find Match]
MaxAttempts -->|No| GenerateCycle[Generate Random Cycle]
GenerateCycle --> ValidateCycle{Cycle valid?}
ValidateCycle -->|No| MaxAttempts
ValidateCycle -->|Yes| CreateMatches[Create Match Records]
CreateMatches --> SendNotifications[Send Notifications]
SendNotifications --> Success([Matching Complete])
Error1 --> End([End])
Error2 --> End
Error3 --> End
Success --> End
```
## Precondition Validation
Before attempting matching, validate the following:
### Validation Checks
```python
def validate_matching_preconditions(exchange_id: int) -> ValidationResult:
"""
Validate that exchange is ready for matching.
Returns:
ValidationResult with is_valid and error_message
"""
checks = [
check_exchange_state(exchange_id),
check_minimum_participants(exchange_id),
check_no_withdrawn_participants(exchange_id),
check_graph_connectivity(exchange_id)
]
for check in checks:
if not check.is_valid:
return check
return ValidationResult(is_valid=True)
```
### Check 1: Exchange State
**Rule**: Exchange must be in "registration_closed" state
**Error Message**: "Exchange is not ready for matching. Please close registration first."
**Implementation**:
```python
def check_exchange_state(exchange_id: int) -> ValidationResult:
exchange = Exchange.query.get(exchange_id)
if exchange.state != ExchangeState.REGISTRATION_CLOSED:
return ValidationResult(
is_valid=False,
error_message="Exchange is not ready for matching. Please close registration first."
)
return ValidationResult(is_valid=True)
```
### Check 2: Minimum Participants
**Rule**: At least 3 non-withdrawn participants required
**Error Message**: "At least 3 participants are required for matching. Current count: {count}"
**Implementation**:
```python
def check_minimum_participants(exchange_id: int) -> ValidationResult:
participants = Participant.query.filter_by(
exchange_id=exchange_id,
withdrawn_at=None
).all()
if len(participants) < 3:
return ValidationResult(
is_valid=False,
error_message=f"At least 3 participants are required for matching. Current count: {len(participants)}"
)
return ValidationResult(is_valid=True)
```
### Check 3: Graph Connectivity
**Rule**: Exclusion rules must not make matching impossible
**Error Message**: Specific message based on connectivity issue
**Implementation**:
```python
def check_graph_connectivity(exchange_id: int) -> ValidationResult:
"""
Check if a valid Hamiltonian cycle is theoretically possible.
This doesn't guarantee a cycle exists, but rules out obvious impossibilities.
"""
participants = get_active_participants(exchange_id)
exclusions = get_exclusions(exchange_id)
# Build graph where edge (A, B) exists if A can give to B
graph = build_assignment_graph(participants, exclusions)
# Check 1: Each participant must have at least one possible recipient
for participant in participants:
possible_recipients = graph.get_outgoing_edges(participant.id)
if len(possible_recipients) == 0:
return ValidationResult(
is_valid=False,
error_message=f"Participant '{participant.name}' has no valid recipients. Please adjust exclusion rules."
)
# Check 2: Each participant must be a possible recipient for at least one other
for participant in participants:
possible_givers = graph.get_incoming_edges(participant.id)
if len(possible_givers) == 0:
return ValidationResult(
is_valid=False,
error_message=f"Participant '{participant.name}' cannot receive from anyone. Please adjust exclusion rules."
)
# Check 3: Detect common impossible scenarios
# Example: In a group of 3, if A excludes B, B excludes C, C excludes A,
# no valid cycle exists
if not has_potential_hamiltonian_cycle(graph):
return ValidationResult(
is_valid=False,
error_message="The current exclusion rules make a valid assignment impossible. Please reduce the number of exclusions."
)
return ValidationResult(is_valid=True)
```
## Assignment Graph Construction
### Graph Representation
Build a directed graph where:
- **Nodes**: Participants
- **Edges**: Valid assignments (A→B means A can give to B)
**Edge exists if**:
- A ≠ B (no self-matching)
- No exclusion rule prevents A→B
### Implementation
```python
class AssignmentGraph:
"""Directed graph representing valid gift assignments."""
def __init__(self, participants: list[Participant], exclusions: list[ExclusionRule]):
self.nodes = {p.id: p for p in participants}
self.edges = {} # {giver_id: [receiver_id, ...]}
self._build_graph(participants, exclusions)
def _build_graph(self, participants, exclusions):
"""Build adjacency list with exclusion rules applied."""
# Build exclusion lookup (bidirectional)
excluded_pairs = set()
for exclusion in exclusions:
excluded_pairs.add((exclusion.participant_a_id, exclusion.participant_b_id))
excluded_pairs.add((exclusion.participant_b_id, exclusion.participant_a_id))
# Build edges
for giver in participants:
self.edges[giver.id] = []
for receiver in participants:
# Can assign if: not self and not excluded
if giver.id != receiver.id and (giver.id, receiver.id) not in excluded_pairs:
self.edges[giver.id].append(receiver.id)
def get_outgoing_edges(self, node_id: int) -> list[int]:
"""Get all possible recipients for a giver."""
return self.edges.get(node_id, [])
def get_incoming_edges(self, node_id: int) -> list[int]:
"""Get all possible givers for a receiver."""
incoming = []
for giver_id, receivers in self.edges.items():
if node_id in receivers:
incoming.append(giver_id)
return incoming
```
## Cycle Generation Algorithm
### Strategy: Randomized Hamiltonian Cycle Search
Goal: Find a Hamiltonian cycle (visits each node exactly once) in the assignment graph.
**Why Single Cycle?**
- Ensures everyone gives and receives exactly once
- Prevents orphaned participants or small isolated loops
- Traditional Secret Santa structure
### Algorithm: Randomized Backtracking with Early Termination
```python
def generate_random_cycle(graph: AssignmentGraph, max_attempts: int = 100) -> Optional[list[tuple[int, int]]]:
"""
Attempt to find a valid Hamiltonian cycle.
Args:
graph: Assignment graph with nodes and edges
max_attempts: Maximum number of randomized attempts
Returns:
List of (giver_id, receiver_id) tuples representing the cycle,
or None if no cycle found within max_attempts
"""
nodes = list(graph.nodes.keys())
for attempt in range(max_attempts):
# Randomize starting point and node order for variety
random.shuffle(nodes)
start_node = nodes[0]
cycle = _backtrack_cycle(graph, start_node, nodes, [], set())
if cycle is not None:
return cycle
return None
def _backtrack_cycle(
graph: AssignmentGraph,
current_node: int,
all_nodes: list[int],
path: list[int],
visited: set[int]
) -> Optional[list[tuple[int, int]]]:
"""
Recursive backtracking to find Hamiltonian cycle.
Args:
current_node: Current node being processed
all_nodes: All nodes in graph
path: Current path taken
visited: Set of visited nodes
Returns:
List of edges forming cycle, or None if no cycle from this path
"""
# Add current node to path
path.append(current_node)
visited.add(current_node)
# Base case: All nodes visited
if len(visited) == len(all_nodes):
# Check if we can return to start (complete the cycle)
start_node = path[0]
if start_node in graph.get_outgoing_edges(current_node):
# Success! Build edge list
edges = []
for i in range(len(path)):
giver = path[i]
receiver = path[(i + 1) % len(path)] # Wrap around for cycle
edges.append((giver, receiver))
return edges
else:
# Can't complete cycle, backtrack
path.pop()
visited.remove(current_node)
return None
# Recursive case: Try each unvisited neighbor
neighbors = graph.get_outgoing_edges(current_node)
random.shuffle(neighbors) # Randomize for variety
for neighbor in neighbors:
if neighbor not in visited:
result = _backtrack_cycle(graph, neighbor, all_nodes, path, visited)
if result is not None:
return result
# No valid path found, backtrack
path.pop()
visited.remove(current_node)
return None
```
### Algorithm Complexity
**Time Complexity**:
- Worst case: O(n!) for Hamiltonian cycle problem (NP-complete)
- In practice: O(n²) to O(n³) for typical Secret Santa scenarios
- Max attempts limit prevents excessive computation
**Space Complexity**: O(n) for recursion stack and visited set
### Why Randomization?
1. **Fairness**: Each valid assignment has equal probability
2. **Unpredictability**: Prevents gaming the system
3. **Variety**: Re-matching produces different results
## Validation & Error Handling
### Cycle Validation
After generating a cycle, validate it before creating database records:
```python
def validate_cycle(cycle: list[tuple[int, int]], graph: AssignmentGraph) -> ValidationResult:
"""
Validate that cycle is valid.
Checks:
1. Each node appears exactly once as giver
2. Each node appears exactly once as receiver
3. All edges exist in graph (no exclusions violated)
4. No self-assignments
"""
givers = set()
receivers = set()
for giver_id, receiver_id in cycle:
# Check for duplicates
if giver_id in givers:
return ValidationResult(is_valid=False, error_message=f"Duplicate giver: {giver_id}")
if receiver_id in receivers:
return ValidationResult(is_valid=False, error_message=f"Duplicate receiver: {receiver_id}")
givers.add(giver_id)
receivers.add(receiver_id)
# Check no self-assignment
if giver_id == receiver_id:
return ValidationResult(is_valid=False, error_message="Self-assignment detected")
# Check edge exists (no exclusion violated)
if receiver_id not in graph.get_outgoing_edges(giver_id):
return ValidationResult(is_valid=False, error_message=f"Invalid assignment: {giver_id} → {receiver_id}")
# Check all nodes present
if givers != set(graph.nodes.keys()) or receivers != set(graph.nodes.keys()):
return ValidationResult(is_valid=False, error_message="Not all participants included in cycle")
return ValidationResult(is_valid=True)
```
### Error Scenarios
| Scenario | Detection | Error Message |
|----------|-----------|---------------|
| Too few participants | Precondition check | "At least 3 participants required" |
| Participant isolated by exclusions | Graph connectivity check | "Participant '{name}' has no valid recipients" |
| Too many exclusions | Graph connectivity check | "Current exclusion rules make matching impossible" |
| Cannot find cycle | Max attempts reached | "Unable to find valid assignment. Try reducing exclusions." |
| Invalid state | Precondition check | "Exchange is not ready for matching" |
## Database Transaction
Matching operation must be atomic (all-or-nothing):
```python
def execute_matching(exchange_id: int) -> MatchingResult:
"""
Execute complete matching operation within transaction.
Returns:
MatchingResult with success status, matches, or error message
"""
from sqlalchemy import orm
# Begin transaction
with db.session.begin_nested():
try:
# 1. Validate preconditions
validation = validate_matching_preconditions(exchange_id)
if not validation.is_valid:
return MatchingResult(success=False, error=validation.error_message)
# 2. Get participants and exclusions
participants = get_active_participants(exchange_id)
exclusions = get_exclusions(exchange_id)
# 3. Build graph
graph = AssignmentGraph(participants, exclusions)
# 4. Generate cycle
cycle = generate_random_cycle(graph, max_attempts=100)
if cycle is None:
return MatchingResult(
success=False,
error="Unable to find valid assignment. Please reduce exclusion rules or add more participants."
)
# 5. Validate cycle
validation = validate_cycle(cycle, graph)
if not validation.is_valid:
return MatchingResult(success=False, error=validation.error_message)
# 6. Create match records
matches = []
for giver_id, receiver_id in cycle:
match = Match(
exchange_id=exchange_id,
giver_id=giver_id,
receiver_id=receiver_id
)
db.session.add(match)
matches.append(match)
# 7. Update exchange state
exchange = Exchange.query.get(exchange_id)
exchange.state = ExchangeState.MATCHED
# Commit nested transaction
db.session.commit()
return MatchingResult(success=True, matches=matches)
except Exception as e:
db.session.rollback()
logger.error(f"Matching failed for exchange {exchange_id}: {str(e)}")
return MatchingResult(
success=False,
error="An unexpected error occurred during matching. Please try again."
)
```
## Re-Matching
When admin triggers re-match, all existing matches must be cleared:
```python
def execute_rematching(exchange_id: int) -> MatchingResult:
"""
Clear existing matches and generate new assignments.
"""
with db.session.begin_nested():
try:
# 1. Validate exchange is in matched state
exchange = Exchange.query.get(exchange_id)
if exchange.state != ExchangeState.MATCHED:
return MatchingResult(success=False, error="Exchange is not in matched state")
# 2. Delete existing matches
Match.query.filter_by(exchange_id=exchange_id).delete()
# 3. Revert state to registration_closed
exchange.state = ExchangeState.REGISTRATION_CLOSED
db.session.flush()
# 4. Run matching again
result = execute_matching(exchange_id)
if result.success:
db.session.commit()
else:
db.session.rollback()
return result
except Exception as e:
db.session.rollback()
logger.error(f"Re-matching failed for exchange {exchange_id}: {str(e)}")
return MatchingResult(success=False, error="Re-matching failed. Please try again.")
```
## Integration with Notification Service
After successful matching, trigger notifications:
```python
def complete_matching_workflow(exchange_id: int) -> WorkflowResult:
"""
Complete matching and send notifications.
"""
# Execute matching
matching_result = execute_matching(exchange_id)
if not matching_result.success:
return WorkflowResult(success=False, error=matching_result.error)
# Send notifications to all participants
try:
notification_service = NotificationService()
notification_service.send_match_notifications(exchange_id)
# Notify admin (if enabled)
notification_service.send_admin_notification(
exchange_id,
NotificationType.MATCHING_COMPLETE
)
return WorkflowResult(success=True)
except Exception as e:
logger.error(f"Failed to send match notifications for exchange {exchange_id}: {str(e)}")
# Matching succeeded but notification failed
# Return success but log the notification failure
return WorkflowResult(
success=True,
warning="Matching complete but some notifications failed to send. Please check email service."
)
```
## Testing Strategy
### Unit Tests
```python
class TestMatchingAlgorithm(unittest.TestCase):
def test_minimum_viable_matching(self):
"""Test matching with 3 participants, no exclusions."""
participants = create_test_participants(3)
exclusions = []
graph = AssignmentGraph(participants, exclusions)
cycle = generate_random_cycle(graph)
self.assertIsNotNone(cycle)
self.assertEqual(len(cycle), 3)
validation = validate_cycle(cycle, graph)
self.assertTrue(validation.is_valid)
def test_matching_with_exclusions(self):
"""Test matching with valid exclusions."""
participants = create_test_participants(5)
exclusions = [
create_exclusion(participants[0], participants[1])
]
graph = AssignmentGraph(participants, exclusions)
cycle = generate_random_cycle(graph)
self.assertIsNotNone(cycle)
# Verify exclusion is respected
for giver_id, receiver_id in cycle:
self.assertNotEqual((giver_id, receiver_id), (participants[0].id, participants[1].id))
def test_impossible_matching_detected(self):
"""Test that impossible matching is detected in validation."""
# Create 3 participants where each excludes the next
# A excludes B, B excludes C, C excludes A
# No Hamiltonian cycle possible
participants = create_test_participants(3)
exclusions = [
create_exclusion(participants[0], participants[1]),
create_exclusion(participants[1], participants[2]),
create_exclusion(participants[2], participants[0])
]
graph = AssignmentGraph(participants, exclusions)
validation = check_graph_connectivity_with_graph(graph)
self.assertFalse(validation.is_valid)
def test_no_self_matching(self):
"""Ensure no participant is matched to themselves."""
participants = create_test_participants(10)
exclusions = []
graph = AssignmentGraph(participants, exclusions)
cycle = generate_random_cycle(graph)
for giver_id, receiver_id in cycle:
self.assertNotEqual(giver_id, receiver_id)
def test_everyone_gives_and_receives_once(self):
"""Ensure each participant gives and receives exactly once."""
participants = create_test_participants(10)
exclusions = []
graph = AssignmentGraph(participants, exclusions)
cycle = generate_random_cycle(graph)
givers = set(giver for giver, _ in cycle)
receivers = set(receiver for _, receiver in cycle)
self.assertEqual(len(givers), 10)
self.assertEqual(len(receivers), 10)
self.assertEqual(givers, {p.id for p in participants})
self.assertEqual(receivers, {p.id for p in participants})
```
### Integration Tests
```python
class TestMatchingIntegration(TestCase):
def test_full_matching_workflow(self):
"""Test complete matching workflow from database to notifications."""
# Setup
exchange = create_test_exchange()
participants = [create_test_participant(exchange) for _ in range(5)]
# Execute
result = complete_matching_workflow(exchange.id)
# Assert
self.assertTrue(result.success)
exchange = Exchange.query.get(exchange.id)
self.assertEqual(exchange.state, ExchangeState.MATCHED)
matches = Match.query.filter_by(exchange_id=exchange.id).all()
self.assertEqual(len(matches), 5)
def test_rematching_clears_old_matches(self):
"""Test that re-matching replaces old assignments."""
# Setup
exchange = create_matched_exchange_with_5_participants()
old_matches = Match.query.filter_by(exchange_id=exchange.id).all()
old_match_ids = {m.id for m in old_matches}
# Execute
result = execute_rematching(exchange.id)
# Assert
self.assertTrue(result.success)
new_matches = Match.query.filter_by(exchange_id=exchange.id).all()
new_match_ids = {m.id for m in new_matches}
# Old match records should be deleted
self.assertEqual(len(old_match_ids.intersection(new_match_ids)), 0)
```
## Performance Considerations
### Expected Performance
| Participants | Exclusions | Expected Time | Notes |
|--------------|------------|---------------|-------|
| 3-10 | 0-5 | <10ms | Instant |
| 10-50 | 0-20 | <100ms | Very fast |
| 50-100 | 0-50 | <500ms | Fast enough |
| 100+ | Any | Variable | May exceed max attempts |
### Optimization Strategies
1. **Graph Pruning**: Remove impossible edges early
2. **Heuristic Ordering**: Start with most constrained nodes
3. **Adaptive Max Attempts**: Increase attempts for larger groups
4. **Fallback to Multiple Cycles**: If single cycle fails, allow 2-3 small cycles
### Max Attempts Configuration
```python
def get_max_attempts(num_participants: int) -> int:
"""Adaptive max attempts based on participant count."""
if num_participants <= 10:
return 50
elif num_participants <= 50:
return 100
else:
return 200
```
## Security Considerations
### Randomization Source
- Use `secrets.SystemRandom()` for cryptographic randomness
- Prevents predictable assignments
- Important for preventing manipulation
```python
import secrets
random_generator = secrets.SystemRandom()
def shuffle(items: list):
"""Cryptographically secure shuffle."""
random_generator.shuffle(items)
```
### Match Confidentiality
- Matches only visible to:
- Giver (sees their own recipient)
- Admin (for troubleshooting)
- Never expose matches in logs
- Database queries filtered by permissions
## Future Enhancements
Potential improvements for future versions:
1. **Multi-Cycle Support**: Allow multiple small cycles if single cycle impossible
2. **Preference Weighting**: Allow participants to indicate preferences
3. **Historical Avoidance**: Avoid repeating matches from previous years
4. **Couple Pairing**: Assign couples to same family/group
5. **Performance Metrics**: Track matching time and success rate
6. **Manual Override**: Allow admin to manually adjust specific assignments
## References
- [Hamiltonian Cycle Problem](https://en.wikipedia.org/wiki/Hamiltonian_path_problem)
- [Graph Theory Basics](https://en.wikipedia.org/wiki/Graph_theory)
- [Backtracking Algorithm](https://en.wikipedia.org/wiki/Backtracking)
- [Data Model Specification](../data-model.md)
- [API Specification](../api-spec.md)

979
docs/designs/v0.1.0/components/notifications.md Normal file
View File

@@ -0,0 +1,979 @@
# Notifications Component Design - v0.1.0
**Version**: 0.1.0
**Date**: 2025-12-22
**Status**: Initial Design
## Introduction
This document defines the email notification system for Sneaky Klaus. The notification service handles all transactional and reminder emails sent to participants and administrators using Resend as the email delivery provider.
## Requirements
### Functional Requirements
1. **Transactional Emails**: Send immediate emails in response to user actions
2. **Reminder Emails**: Send scheduled reminder emails before exchange date
3. **Admin Notifications**: Notify admin of important exchange events (opt-in)
4. **Magic Link Delivery**: Include authentication tokens in emails
5. **Template Management**: Maintain consistent branded email templates
6. **Error Handling**: Gracefully handle email delivery failures
### Non-Functional Requirements
1. **Reliability**: Guarantee delivery or retry on failure
2. **Performance**: Send emails asynchronously without blocking requests
3. **Auditability**: Log all email send attempts
4. **Deliverability**: Follow email best practices (SPF, DKIM, unsubscribe links)
## Email Types
### Participant Emails
| Email Type | Trigger | Recipient | Time-Sensitive |
|------------|---------|-----------|----------------|
| Registration Confirmation | Participant registers | Participant | Yes (immediate) |
| Magic Link | Participant requests access | Participant | Yes (immediate) |
| Match Notification | Matching complete | All participants | Yes (immediate) |
| Reminder Email | Scheduled (pre-exchange) | Opted-in participants | Yes (scheduled) |
| Withdrawal Confirmation | Participant withdraws | Participant | Yes (immediate) |
### Admin Emails
| Email Type | Trigger | Recipient | Time-Sensitive |
|------------|---------|-----------|----------------|
| Password Reset | Admin requests reset | Admin | Yes (immediate) |
| New Registration | Participant registers | Admin | No (opt-in) |
| Participant Withdrawal | Participant withdraws | Admin | No (opt-in) |
| Matching Complete | Matching succeeds | Admin | No (opt-in) |
| Data Purge Warning | 7 days before purge | Admin | No |
## Notification Service Architecture
```mermaid
flowchart TB
subgraph "Application Layer"
Route[Route Handler]
Service[Business Logic]
end
subgraph "Notification Service"
NS[NotificationService]
TemplateEngine[Template Renderer]
EmailQueue[Email Queue]
end
subgraph "External Services"
Resend[Resend API]
end
subgraph "Storage"
DB[(Database)]
Templates[Email Templates]
end
Route --> Service
Service --> NS
NS --> TemplateEngine
TemplateEngine --> Templates
NS --> EmailQueue
EmailQueue --> Resend
NS --> DB
Resend --> DB
style Resend fill:#f9f,stroke:#333
style Templates fill:#bfb,stroke:#333
```
## Implementation Structure
### Service Class
```python
class NotificationService:
"""
Centralized service for all email notifications.
"""
def __init__(self, resend_client: ResendClient = None):
self.resend = resend_client or ResendClient(api_key=get_resend_api_key())
self.template_renderer = EmailTemplateRenderer()
self.logger = logging.getLogger(__name__)
# Participant Emails
def send_registration_confirmation(self, participant_id: int) -> EmailResult
def send_magic_link(self, participant_id: int, token: str) -> EmailResult
def send_match_notification(self, participant_id: int) -> EmailResult
def send_reminder_email(self, participant_id: int) -> EmailResult
def send_withdrawal_confirmation(self, participant_id: int) -> EmailResult
# Admin Emails
def send_password_reset(self, admin_email: str, token: str) -> EmailResult
def send_admin_notification(self, exchange_id: int, notification_type: NotificationType) -> EmailResult
def send_data_purge_warning(self, exchange_id: int) -> EmailResult
# Batch Operations
def send_match_notifications_batch(self, exchange_id: int) -> BatchEmailResult
# Internal Methods
def _send_email(self, email_request: EmailRequest) -> EmailResult
def _log_email_send(self, email_request: EmailRequest, result: EmailResult)
```
## Email Templates
### Template Structure
All email templates use Jinja2 with HTML and plain text versions:
**Directory Structure**:
```
templates/emails/
├── base.html # Base template with header/footer
├── base.txt # Plain text base
├── participant/
│ ├── registration_confirmation.html
│ ├── registration_confirmation.txt
│ ├── magic_link.html
│ ├── magic_link.txt
│ ├── match_notification.html
│ ├── match_notification.txt
│ ├── reminder.html
│ ├── reminder.txt
│ └── withdrawal_confirmation.html
│ └── withdrawal_confirmation.txt
└── admin/
├── password_reset.html
├── password_reset.txt
├── new_registration.html
├── new_registration.txt
├── participant_withdrawal.html
├── participant_withdrawal.txt
├── matching_complete.html
├── matching_complete.txt
├── data_purge_warning.html
└── data_purge_warning.txt
```
### Base Template
**base.html**:
```html
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>{% block title %}Sneaky Klaus{% endblock %}</title>
<style>
/* Inline CSS for email client compatibility */
body { font-family: Arial, sans-serif; line-height: 1.6; color: #333; }
.container { max-width: 600px; margin: 0 auto; padding: 20px; }
.header { background-color: #d32f2f; color: white; padding: 20px; text-align: center; }
.content { padding: 30px; background-color: #f9f9f9; }
.button { display: inline-block; padding: 12px 24px; background-color: #d32f2f; color: white; text-decoration: none; border-radius: 4px; }
.footer { text-align: center; font-size: 12px; color: #666; padding: 20px; }
</style>
</head>
<body>
<div class="container">
<div class="header">
<h1>🎅 Sneaky Klaus</h1>
</div>
<div class="content">
{% block content %}{% endblock %}
</div>
<div class="footer">
<p>You received this email because you're part of a Sneaky Klaus Secret Santa exchange.</p>
{% block unsubscribe %}{% endblock %}
<p>&copy; {{ current_year }} Sneaky Klaus</p>
</div>
</div>
</body>
</html>
```
**base.txt**:
```text
SNEAKY KLAUS
=============
{% block content %}{% endblock %}
---
You received this email because you're part of a Sneaky Klaus Secret Santa exchange.
{% block unsubscribe %}{% endblock %}
© {{ current_year }} Sneaky Klaus
```
## Participant Email Specifications
### 1. Registration Confirmation
**Trigger**: Immediately after participant registration
**Subject**: "Welcome to {exchange_name}!"
**Template Variables**:
```python
{
"participant_name": str,
"exchange_name": str,
"exchange_date": datetime,
"budget": str,
"magic_link_url": str,
"app_url": str
}
```
**Content** (HTML version):
```html
{% extends "emails/base.html" %}
{% block content %}
<h2>Welcome to {{ exchange_name }}!</h2>
<p>Hi {{ participant_name }},</p>
<p>You've successfully registered for the Secret Santa exchange!</p>
<p><strong>Exchange Details:</strong></p>
<ul>
<li><strong>Event Date:</strong> {{ exchange_date|format_date }}</li>
<li><strong>Gift Budget:</strong> {{ budget }}</li>
</ul>
<p>You can update your gift ideas or view participant information anytime using the link below:</p>
<p style="text-align: center;">
<a href="{{ magic_link_url }}" class="button">Access My Registration</a>
</p>
<p><small>This link will expire in 1 hour. You can request a new one anytime from the registration page.</small></p>
<p>When participants are matched, you'll receive another email with your Secret Santa assignment.</p>
<p>Happy gifting!</p>
{% endblock %}
```
**Plain Text Version**: Similar content without HTML formatting
---
### 2. Magic Link
**Trigger**: Participant requests access to registration
**Subject**: "Access Your Sneaky Klaus Registration"
**Template Variables**:
```python
{
"participant_name": str,
"exchange_name": str,
"magic_link_url": str,
"expiration_minutes": int # 60
}
```
**Content**:
```html
{% extends "emails/base.html" %}
{% block content %}
<h2>Access Your Registration</h2>
<p>Hi {{ participant_name }},</p>
<p>You requested access to your registration for <strong>{{ exchange_name }}</strong>.</p>
<p style="text-align: center;">
<a href="{{ magic_link_url }}" class="button">Access My Registration</a>
</p>
<p><small>This link will expire in {{ expiration_minutes }} minutes and can only be used once.</small></p>
<p>If you didn't request this link, you can safely ignore this email.</p>
{% endblock %}
```
---
### 3. Match Notification
**Trigger**: Matching complete (sent to all participants)
**Subject**: "Your Secret Santa Assignment for {exchange_name}"
**Template Variables**:
```python
{
"participant_name": str,
"exchange_name": str,
"exchange_date": datetime,
"budget": str,
"recipient_name": str,
"recipient_gift_ideas": str,
"magic_link_url": str,
"participant_count": int
}
```
**Content**:
```html
{% extends "emails/base.html" %}
{% block content %}
<h2>Your Secret Santa Assignment</h2>
<p>Hi {{ participant_name }},</p>
<p>Participants have been matched for <strong>{{ exchange_name }}</strong>! 🎁</p>
<div style="background-color: white; padding: 20px; border-radius: 8px; border-left: 4px solid #d32f2f; margin: 20px 0;">
<h3 style="margin-top: 0;">You're buying for:</h3>
<p style="font-size: 18px; font-weight: bold; margin: 10px 0;">{{ recipient_name }}</p>
{% if recipient_gift_ideas %}
<p><strong>Gift Ideas:</strong></p>
<p style="white-space: pre-wrap;">{{ recipient_gift_ideas }}</p>
{% else %}
<p><em>No gift ideas provided yet.</em></p>
{% endif %}
</div>
<p><strong>Exchange Details:</strong></p>
<ul>
<li><strong>Gift Budget:</strong> {{ budget }}</li>
<li><strong>Exchange Date:</strong> {{ exchange_date|format_date }}</li>
<li><strong>Total Participants:</strong> {{ participant_count }}</li>
</ul>
<p>You can view this information anytime by clicking the link below:</p>
<p style="text-align: center;">
<a href="{{ magic_link_url }}" class="button">View My Assignment</a>
</p>
<p><strong>Remember:</strong> Keep your assignment secret! The fun is in the surprise. 🤫</p>
<p>Happy shopping!</p>
{% endblock %}
```
---
### 4. Reminder Email
**Trigger**: Scheduled (based on admin configuration, e.g., 7 days, 3 days, 1 day before exchange)
**Subject**: "Reminder: {exchange_name} is {days_until} days away!"
**Template Variables**:
```python
{
"participant_name": str,
"exchange_name": str,
"exchange_date": datetime,
"days_until": int,
"recipient_name": str,
"recipient_gift_ideas": str,
"budget": str,
"magic_link_url": str
}
```
**Content**:
```html
{% extends "emails/base.html" %}
{% block content %}
<h2>Don't Forget! {{ exchange_name }} is Coming Up</h2>
<p>Hi {{ participant_name }},</p>
<p>This is a friendly reminder that <strong>{{ exchange_name }}</strong> is only <strong>{{ days_until }} day{{ 's' if days_until != 1 else '' }}</strong> away!</p>
<p>You're buying for: <strong>{{ recipient_name }}</strong></p>
{% if recipient_gift_ideas %}
<p><strong>Their Gift Ideas:</strong></p>
<p style="white-space: pre-wrap; background-color: white; padding: 15px; border-radius: 4px;">{{ recipient_gift_ideas }}</p>
{% endif %}
<p><strong>Gift Budget:</strong> {{ budget }}</p>
<p><strong>Exchange Date:</strong> {{ exchange_date|format_date }}</p>
<p style="text-align: center;">
<a href="{{ magic_link_url }}" class="button">View Full Details</a>
</p>
<p>Happy shopping! 🎁</p>
{% endblock %}
{% block unsubscribe %}
<p><a href="{{ unsubscribe_url }}">Don't want reminders? Update your preferences</a></p>
{% endblock %}
```
---
### 5. Withdrawal Confirmation
**Trigger**: Participant withdraws from exchange
**Subject**: "You've withdrawn from {exchange_name}"
**Template Variables**:
```python
{
"participant_name": str,
"exchange_name": str
}
```
**Content**:
```html
{% extends "emails/base.html" %}
{% block content %}
<h2>Withdrawal Confirmed</h2>
<p>Hi {{ participant_name }},</p>
<p>You've successfully withdrawn from <strong>{{ exchange_name }}</strong>.</p>
<p>You will no longer receive any emails about this exchange.</p>
<p>If this was a mistake, please contact the exchange organizer to re-register.</p>
{% endblock %}
```
## Admin Email Specifications
### 1. Password Reset
**Trigger**: Admin requests password reset
**Subject**: "Password Reset Request - Sneaky Klaus"
**Template Variables**:
```python
{
"reset_link_url": str,
"expiration_minutes": int # 60
}
```
**Content**:
```html
{% extends "emails/base.html" %}
{% block content %}
<h2>Password Reset Request</h2>
<p>You requested a password reset for your Sneaky Klaus admin account.</p>
<p style="text-align: center;">
<a href="{{ reset_link_url }}" class="button">Reset Password</a>
</p>
<p><small>This link will expire in {{ expiration_minutes }} minutes and can only be used once.</small></p>
<p>If you didn't request this reset, you can safely ignore this email. Your password will not be changed.</p>
{% endblock %}
```
---
### 2. New Registration (Admin Notification)
**Trigger**: Participant registers (if admin has enabled this notification)
**Subject**: "New Participant in {exchange_name}"
**Template Variables**:
```python
{
"exchange_name": str,
"participant_name": str,
"participant_email": str,
"participant_count": int,
"max_participants": int,
"exchange_url": str
}
```
**Content**:
```html
{% extends "emails/base.html" %}
{% block content %}
<h2>New Participant Registered</h2>
<p>A new participant has joined <strong>{{ exchange_name }}</strong>:</p>
<ul>
<li><strong>Name:</strong> {{ participant_name }}</li>
<li><strong>Email:</strong> {{ participant_email }}</li>
</ul>
<p><strong>Participant Count:</strong> {{ participant_count }} / {{ max_participants }}</p>
<p style="text-align: center;">
<a href="{{ exchange_url }}" class="button">View Exchange</a>
</p>
{% endblock %}
```
---
### 3. Participant Withdrawal (Admin Notification)
**Trigger**: Participant withdraws (if admin has enabled this notification)
**Subject**: "Participant Withdrew from {exchange_name}"
**Template Variables**:
```python
{
"exchange_name": str,
"participant_name": str,
"participant_count": int,
"exchange_url": str
}
```
**Content**:
```html
{% extends "emails/base.html" %}
{% block content %}
<h2>Participant Withdrew</h2>
<p><strong>{{ participant_name }}</strong> has withdrawn from <strong>{{ exchange_name }}</strong>.</p>
<p><strong>Remaining Participants:</strong> {{ participant_count }}</p>
<p style="text-align: center;">
<a href="{{ exchange_url }}" class="button">View Exchange</a>
</p>
{% endblock %}
```
---
### 4. Matching Complete (Admin Notification)
**Trigger**: Matching succeeds (if admin has enabled this notification)
**Subject**: "Matching Complete for {exchange_name}"
**Template Variables**:
```python
{
"exchange_name": str,
"participant_count": int,
"exchange_date": datetime,
"exchange_url": str
}
```
**Content**:
```html
{% extends "emails/base.html" %}
{% block content %}
<h2>Matching Complete! 🎉</h2>
<p>Participants have been successfully matched for <strong>{{ exchange_name }}</strong>.</p>
<p><strong>Details:</strong></p>
<ul>
<li><strong>Participants Matched:</strong> {{ participant_count }}</li>
<li><strong>Exchange Date:</strong> {{ exchange_date|format_date }}</li>
</ul>
<p>All participants have been notified of their assignments via email.</p>
<p style="text-align: center;">
<a href="{{ exchange_url }}" class="button">View Exchange</a>
</p>
{% endblock %}
```
---
### 5. Data Purge Warning
**Trigger**: 7 days before exchange data is purged (30 days after completion)
**Subject**: "Data Purge Scheduled for {exchange_name}"
**Template Variables**:
```python
{
"exchange_name": str,
"purge_date": datetime,
"days_until_purge": int,
"exchange_url": str
}
```
**Content**:
```html
{% extends "emails/base.html" %}
{% block content %}
<h2>Data Purge Scheduled</h2>
<p>The exchange <strong>{{ exchange_name }}</strong> will be automatically deleted in <strong>{{ days_until_purge }} days</strong> ({{ purge_date|format_date }}).</p>
<p>All participant data, matches, and exchange details will be permanently removed as per the 30-day retention policy.</p>
<p>If you need to keep this data, please export it before the purge date.</p>
<p style="text-align: center;">
<a href="{{ exchange_url }}" class="button">View Exchange</a>
</p>
{% endblock %}
```
## Resend Integration
### Configuration
```python
import resend
class ResendClient:
"""Wrapper for Resend API."""
def __init__(self, api_key: str):
resend.api_key = api_key
self.from_email = "noreply@sneakyklaus.app" # Configured domain
self.from_name = "Sneaky Klaus"
def send_email(self, request: EmailRequest) -> EmailResult:
"""
Send email via Resend API.
Args:
request: EmailRequest with to, subject, html, text
Returns:
EmailResult with success status and message ID
"""
try:
params = {
"from": f"{self.from_name} <{self.from_email}>",
"to": [request.to_email],
"subject": request.subject,
"html": request.html_body,
"text": request.text_body,
}
# Optional: Add tags for tracking
if request.tags:
params["tags"] = request.tags
response = resend.Emails.send(params)
return EmailResult(
success=True,
message_id=response["id"],
timestamp=datetime.utcnow()
)
except resend.exceptions.ResendError as e:
logger.error(f"Resend API error: {str(e)}")
return EmailResult(
success=False,
error=str(e),
timestamp=datetime.utcnow()
)
except Exception as e:
logger.error(f"Unexpected error sending email: {str(e)}")
return EmailResult(
success=False,
error="Internal error",
timestamp=datetime.utcnow()
)
```
### Email Request Model
```python
@dataclass
class EmailRequest:
"""Email send request."""
to_email: str
subject: str
html_body: str
text_body: str
tags: Optional[dict] = None # For analytics/tracking
@dataclass
class EmailResult:
"""Email send result."""
success: bool
message_id: Optional[str] = None
error: Optional[str] = None
timestamp: datetime = field(default_factory=datetime.utcnow)
```
### Email Tags (Optional)
For analytics and troubleshooting:
```python
tags = {
"type": "match_notification",
"exchange_id": "123",
"environment": "production"
}
```
## Error Handling & Retries
### Retry Strategy
```python
def send_email_with_retry(request: EmailRequest, max_retries: int = 3) -> EmailResult:
"""
Send email with exponential backoff retry.
Args:
request: EmailRequest to send
max_retries: Maximum retry attempts
Returns:
EmailResult
"""
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
@retry(
stop=stop_after_attempt(max_retries),
wait=wait_exponential(multiplier=1, min=2, max=10),
retry=retry_if_exception_type(resend.exceptions.ResendError)
)
def _send():
return resend_client.send_email(request)
try:
return _send()
except Exception as e:
logger.error(f"Failed to send email after {max_retries} attempts: {str(e)}")
return EmailResult(success=False, error=f"Failed after {max_retries} retries")
```
### Failure Logging
All email send attempts logged to database for audit:
```python
class EmailLog(db.Model):
"""Audit log for email sends."""
id = db.Column(db.Integer, primary_key=True)
to_email = db.Column(db.String(255), nullable=False)
subject = db.Column(db.String(500), nullable=False)
email_type = db.Column(db.String(50), nullable=False)
success = db.Column(db.Boolean, nullable=False)
message_id = db.Column(db.String(255), nullable=True)
error = db.Column(db.Text, nullable=True)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
```
## Batch Email Sending
For sending to multiple recipients (e.g., match notifications):
```python
def send_match_notifications_batch(self, exchange_id: int) -> BatchEmailResult:
"""
Send match notifications to all participants in exchange.
Args:
exchange_id: Exchange ID
Returns:
BatchEmailResult with success count and failures
"""
participants = Participant.query.filter_by(
exchange_id=exchange_id,
withdrawn_at=None
).all()
results = []
failed = []
for participant in participants:
result = self.send_match_notification(participant.id)
results.append(result)
if not result.success:
failed.append({
"participant_id": participant.id,
"participant_email": participant.email,
"error": result.error
})
# Rate limit: Small delay between sends
time.sleep(0.1)
return BatchEmailResult(
total=len(participants),
successful=len([r for r in results if r.success]),
failed=len(failed),
failures=failed
)
```
## Testing
### Email Testing in Development
**Option 1: Resend Test Mode**
- Use Resend's test API key
- Emails sent to test mode, not delivered
**Option 2: MailHog / MailCatcher**
- Local SMTP server for testing
- View emails in web UI
**Option 3: Mock in Unit Tests**
```python
from unittest.mock import patch
class TestNotificationService(unittest.TestCase):
@patch('notification_service.ResendClient.send_email')
def test_send_registration_confirmation(self, mock_send):
mock_send.return_value = EmailResult(success=True, message_id="test-123")
service = NotificationService()
result = service.send_registration_confirmation(participant_id=1)
self.assertTrue(result.success)
mock_send.assert_called_once()
```
### Integration Tests
```python
class TestNotificationIntegration(TestCase):
def test_full_match_notification_workflow(self):
"""Test complete match notification workflow."""
# Setup
exchange = create_test_exchange()
participants = [create_test_participant(exchange) for _ in range(5)]
execute_matching(exchange.id)
# Execute
service = NotificationService()
result = service.send_match_notifications_batch(exchange.id)
# Assert
self.assertEqual(result.total, 5)
self.assertEqual(result.successful, 5)
self.assertEqual(result.failed, 0)
# Verify emails logged
logs = EmailLog.query.filter_by(email_type="match_notification").all()
self.assertEqual(len(logs), 5)
```
## Deliverability Best Practices
### SPF, DKIM, DMARC
Configure DNS records for Resend domain:
- **SPF**: Authorize Resend to send on your behalf
- **DKIM**: Sign emails cryptographically
- **DMARC**: Define policy for failed authentication
**Example DNS Configuration**:
```
TXT @ "v=spf1 include:_spf.resend.com ~all"
CNAME resend._domainkey resend.domainkey.resend.com
TXT _dmarc "v=DMARC1; p=quarantine; rua=mailto:admin@example.com"
```
### Unsubscribe Links
For reminder emails, include unsubscribe link:
```python
unsubscribe_url = f"{app_url}/participant/exchange/{exchange_id}/edit"
```
Participants can disable reminders via profile edit.
### Email Content Best Practices
1. **Clear Subject Lines**: Descriptive and concise
2. **Plain Text Alternative**: Always include text version
3. **Inline CSS**: Email clients strip external stylesheets
4. **Mobile Responsive**: Use responsive design techniques
5. **Clear Call-to-Action**: Prominent buttons/links
6. **Avoid Spam Triggers**: No all-caps, excessive punctuation, spam keywords
## Performance Considerations
### Asynchronous Sending
For non-critical emails, send asynchronously:
```python
from threading import Thread
def send_email_async(email_request: EmailRequest):
"""Send email in background thread."""
thread = Thread(target=lambda: notification_service.send_email(email_request))
thread.start()
```
**Note**: For production, use proper background job queue (see background-jobs.md)
### Rate Limiting
Resend has rate limits (depends on plan):
- Free: 100 emails/day
- Paid: Higher limits
**Mitigation**:
- Batch operations with delays between sends
- Implement queue for large batches
- Monitor usage and implement backoff
## Security Considerations
### Token Inclusion
Magic links and password reset tokens:
- **URL Structure**: `{app_url}/auth/participant/magic/{token}`
- **Token Format**: 32-byte random, base64url encoded
- **Security**: Tokens hashed in database, original never stored
### Email Spoofing Prevention
- Use authenticated Resend domain
- Configure SPF/DKIM/DMARC
- Never allow user-controlled "from" addresses
### Sensitive Data
- **Never include**: Passwords, full tokens (only links)
- **Include only necessary**: Participant names, gift ideas (expected in context)
- **Audit log**: Track all emails sent
## Future Enhancements
1. **HTML Email Builder**: Visual template editor for admin
2. **Localization**: Multi-language email templates
3. **A/B Testing**: Test different email content for engagement
4. **Analytics**: Track open rates, click rates (Resend webhooks)
5. **Custom Branding**: Allow admin to customize email header/colors
6. **Email Queue Dashboard**: Admin view of pending/failed emails
## References
- [Resend Documentation](https://resend.com/docs)
- [Jinja2 Template Documentation](https://jinja.palletsprojects.com/)
- [Email Deliverability Best Practices](https://www.mailgun.com/blog/email/email-deliverability-best-practices/)
- [Data Model Specification](../data-model.md)
- [API Specification](../api-spec.md)