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

feat: add Participant and MagicToken models with automatic migrations

Browse Source 
Implements Phase 2 infrastructure for participant registration and authentication:

Database Models:
- Add Participant model with exchange scoping and soft deletes
- Add MagicToken model for passwordless authentication
- Add participants relationship to Exchange model
- Include proper indexes and foreign key constraints

Migration Infrastructure:
- Generate Alembic migration for new models
- Create entrypoint.sh script for automatic migrations on container startup
- Update Containerfile to use entrypoint script and include uv binary
- Remove db.create_all() in favor of migration-based schema management

This establishes the foundation for implementing stories 4.1-4.3, 5.1-5.3, and 10.1.

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Phil Skentelbery
2025-12-22 16:23:47 -07:00
parent 5201b2f036
commit eaafa78cf3
22 changed files with 10459 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

732
docs/designs/v0.2.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)