sneakyklaus/docs/designs/v0.1.0/components/matching.md

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

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

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:

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:

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:

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

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

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:

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):

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:

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:

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

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

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

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

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
• Graph Theory Basics
• Backtracking Algorithm
• Data Model Specification
• API Specification