Coverage for services/invite_service.py: 43.13%
264 statements
« prev ^ index » next coverage.py v7.10.6, created at 2026-04-13 00:07 +0000
1"""
2Invite Service for Missing Table
3Handles invite code generation, validation, and management
4"""
6import logging
7import secrets
8from datetime import UTC, datetime, timedelta
10from dao.base_dao import clear_cache
11from supabase import Client
13logger = logging.getLogger(__name__)
16class InviteService:
17 """Service for managing invitations"""
19 def __init__(self, supabase_client: Client):
20 self.supabase = supabase_client
21 # Characters for invite codes (avoiding ambiguous characters)
22 self.code_chars = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789"
24 def generate_invite_code(self) -> str:
25 """Generate a unique 12-character invite code"""
26 max_attempts = 100
28 for _ in range(max_attempts):
29 # Generate random code
30 code = "".join(secrets.choice(self.code_chars) for _ in range(12))
32 # Check if code already exists
33 response = self.supabase.table("invitations").select("id").eq("invite_code", code).execute()
35 if not response.data:
36 return code
38 raise Exception("Could not generate unique invite code after 100 attempts")
40 def create_invitation(
41 self,
42 invited_by_user_id: str,
43 invite_type: str,
44 team_id: int | None = None,
45 age_group_id: int | None = None,
46 club_id: int | None = None,
47 email: str | None = None,
48 player_id: int | None = None,
49 jersey_number: int | None = None,
50 expires_in_days: int = 7,
51 ) -> dict:
52 """
53 Create a new invitation
55 Args:
56 invited_by_user_id: ID of the user creating the invite
57 invite_type: Type of invite ('club_manager', 'club_fan', 'team_manager', 'team_player', 'team_fan')
58 team_id: ID of the team (required for team-level invite types)
59 age_group_id: ID of the age group (required for team-level invite types)
60 club_id: ID of the club (required for club_manager and club_fan invite types)
61 email: Optional email to pre-fill during registration
62 player_id: Optional roster entry ID (for team_player invites - links account to existing roster)
63 jersey_number: Optional jersey number (for team_player invites - creates roster on redemption)
64 expires_in_days: Number of days until invite expires
66 Returns:
67 Created invitation record
68 """
69 try:
70 # Validate parameters based on invite type
71 if invite_type in ("club_manager", "club_fan"):
72 if not club_id:
73 raise ValueError(f"club_id is required for {invite_type} invites")
74 # Club-level invites don't need team_id or age_group_id
75 team_id = None
76 age_group_id = None
77 player_id = None # No roster linking for club-level invites
78 jersey_number = None
79 else:
80 if not team_id:
81 raise ValueError("team_id is required for team-level invites")
82 if not age_group_id:
83 raise ValueError("age_group_id is required for team-level invites")
84 # Derive club_id from the team's parent club
85 club_id = self._get_team_club_id(team_id)
87 # Generate unique invite code
88 invite_code = self.generate_invite_code()
90 # Calculate expiration date
91 expires_at = datetime.now(UTC) + timedelta(days=expires_in_days)
93 # Create invitation record
94 invitation_data = {
95 "invite_code": invite_code,
96 "invited_by_user_id": invited_by_user_id,
97 "invite_type": invite_type,
98 "team_id": team_id,
99 "age_group_id": age_group_id,
100 "club_id": club_id,
101 "email": email,
102 "player_id": player_id,
103 "jersey_number": jersey_number,
104 "status": "pending",
105 "expires_at": expires_at.isoformat(),
106 }
108 response = self.supabase.table("invitations").insert(invitation_data).execute()
110 if response.data: 110 ↛ 119line 110 didn't jump to line 119 because the condition on line 110 was always true
111 log_msg = f"Created {invite_type} invitation: {invite_code}"
112 if player_id: 112 ↛ 113line 112 didn't jump to line 113 because the condition on line 112 was never true
113 log_msg += f" linked to player {player_id}"
114 elif jersey_number: 114 ↛ 115line 114 didn't jump to line 115 because the condition on line 114 was never true
115 log_msg += f" with jersey #{jersey_number}"
116 logger.info(log_msg)
117 return response.data[0]
118 else:
119 raise Exception("Failed to create invitation")
121 except Exception as e:
122 logger.error(f"Error creating invitation: {e}")
123 raise
125 def validate_invite_code(self, code: str) -> dict | None:
126 """
127 Validate an invite code
129 Args:
130 code: The invite code to validate
132 Returns:
133 Invitation details if valid, None otherwise
134 """
135 try:
136 logger.info(f"Validating invite code: {code}")
138 # Get invitation by code with related data including player roster entry
139 response = (
140 self.supabase.table("invitations")
141 .select(
142 "*, teams(name), age_groups(name), clubs(name), players(id, jersey_number, first_name, last_name)"
143 )
144 .eq("invite_code", code)
145 .execute()
146 )
148 if not response.data or len(response.data) == 0:
149 logger.warning(f"Invite code {code} not found in database")
150 return None
152 invitation = response.data[0]
153 logger.info(f"Found invitation: status={invitation['status']}, expires_at={invitation['expires_at']}")
155 # Check if already used
156 if invitation["status"] != "pending":
157 logger.warning(f"Invite code {code} has status '{invitation['status']}' (not pending)")
158 return None
160 # Check if expired
161 expires_at_str = invitation["expires_at"]
162 if expires_at_str.endswith("Z"):
163 expires_at_str = expires_at_str.replace("Z", "+00:00")
164 expires_at = datetime.fromisoformat(expires_at_str)
166 # Make current time timezone-aware for comparison
167 current_time = datetime.now(UTC)
169 is_expired = expires_at < current_time
170 logger.info(f"Invite code {code}: expires_at={expires_at}, current={current_time}, is_expired={is_expired}")
172 if expires_at < current_time:
173 # Update status to expired
174 self.supabase.table("invitations").update({"status": "expired"}).eq("id", invitation["id"]).execute()
175 logger.warning(f"Invite code {code} expired at {expires_at}")
176 return None
178 # Build player info if linked to roster
179 player_info = None
180 if invitation.get("players"): 180 ↛ 181line 180 didn't jump to line 181 because the condition on line 180 was never true
181 player = invitation["players"]
182 player_info = {
183 "id": player["id"],
184 "jersey_number": player["jersey_number"],
185 "first_name": player.get("first_name"),
186 "last_name": player.get("last_name"),
187 }
189 logger.info(f"Invite code {code} is valid!")
190 return {
191 "valid": True,
192 "id": invitation["id"],
193 "invite_type": invitation["invite_type"],
194 "team_id": invitation["team_id"],
195 "team_name": invitation["teams"]["name"] if invitation.get("teams") else None,
196 "age_group_id": invitation["age_group_id"],
197 "age_group_name": invitation["age_groups"]["name"] if invitation.get("age_groups") else None,
198 "club_id": invitation.get("club_id"),
199 "club_name": invitation["clubs"]["name"] if invitation.get("clubs") else None,
200 "email": invitation["email"],
201 "invited_by_user_id": invitation.get("invited_by_user_id"),
202 "player_id": invitation.get("player_id"),
203 "jersey_number": invitation.get("jersey_number"),
204 "player": player_info,
205 }
207 except Exception as e:
208 logger.error(f"Error validating invite code {code}: {e}", exc_info=True)
209 return None
211 def redeem_invitation(self, code: str, user_id: str) -> bool:
212 """
213 Redeem an invitation code
215 Args:
216 code: The invite code to redeem
217 user_id: ID of the user redeeming the code
219 Returns:
220 True if successful, False otherwise
221 """
222 try:
223 # First validate the code
224 invitation = self.validate_invite_code(code)
225 if not invitation:
226 return False
228 # Update invitation status
229 response = (
230 self.supabase.table("invitations")
231 .update(
232 {
233 "status": "used",
234 "used_at": datetime.now(UTC).isoformat(),
235 "used_by_user_id": user_id,
236 }
237 )
238 .eq("invite_code", code)
239 .execute()
240 )
242 if response.data: 242 ↛ 289line 242 didn't jump to line 289 because the condition on line 242 was always true
243 logger.info(f"Invitation {code} redeemed by user {user_id}")
245 # Create team_manager_assignments entry for team_manager invites
246 # This enables the team manager to create player/fan invites for their team
247 if invitation.get("invite_type") == "team_manager": 247 ↛ 248line 247 didn't jump to line 248 because the condition on line 247 was never true
248 self._create_team_manager_assignment(
249 user_id=user_id,
250 team_id=invitation.get("team_id"),
251 age_group_id=invitation.get("age_group_id"),
252 assigned_by_user_id=invitation.get("invited_by_user_id"),
253 )
255 # Handle player roster linking for team_player invites
256 if invitation.get("invite_type") == "team_player": 256 ↛ 287line 256 didn't jump to line 287 because the condition on line 256 was always true
257 logger.info(
258 f"Processing team_player invite: "
259 f"player_id={invitation.get('player_id')}, "
260 f"jersey={invitation.get('jersey_number')}, "
261 f"team={invitation.get('team_id')}, "
262 f"age_group={invitation.get('age_group_id')}"
263 )
264 if invitation.get("player_id"): 264 ↛ 266line 264 didn't jump to line 266 because the condition on line 264 was never true
265 # Link user to existing roster entry (also creates player_team_history)
266 self._link_user_to_roster_entry(
267 user_id=user_id,
268 player_id=invitation.get("player_id"),
269 team_id=invitation.get("team_id"),
270 age_group_id=invitation.get("age_group_id"),
271 )
272 elif invitation.get("jersey_number"): 272 ↛ 275line 272 didn't jump to line 275 because the condition on line 272 was never true
273 # Create new roster entry with jersey number and link user
274 # Also creates player_team_history entry
275 jersey = invitation.get("jersey_number")
276 logger.info(f"Creating roster entry for jersey #{jersey}")
277 self._create_and_link_roster_entry(
278 user_id=user_id,
279 team_id=invitation.get("team_id"),
280 jersey_number=invitation.get("jersey_number"),
281 age_group_id=invitation.get("age_group_id"),
282 invited_by_user_id=invitation.get("invited_by_user_id"),
283 )
284 else:
285 logger.info("No player_id or jersey_number on invite - skipping roster")
287 return True
289 return False
291 except Exception as e:
292 logger.error(f"Error redeeming invitation {code}: {e}")
293 return False
295 def _get_team_club_id(self, team_id: int) -> int | None:
296 """Look up a team's parent club_id."""
297 try:
298 response = self.supabase.table("teams").select("club_id").eq("id", team_id).execute()
299 if response.data: 299 ↛ 300line 299 didn't jump to line 300 because the condition on line 299 was never true
300 return response.data[0].get("club_id")
301 except Exception as e:
302 logger.warning(f"Could not look up club_id for team {team_id}: {e}")
303 return None
305 def _link_user_to_roster_entry(
306 self,
307 user_id: str,
308 player_id: int,
309 team_id: int | None = None,
310 season_id: int | None = None,
311 age_group_id: int | None = None,
312 league_id: int | None = None,
313 division_id: int | None = None,
314 jersey_number: int | None = None,
315 ) -> bool:
316 """
317 Link a user account to a roster entry in the players table.
318 Also creates a player_team_history entry for the UI roster view.
320 Called when a player accepts an invitation that was tied to a roster entry.
322 Args:
323 user_id: The user ID to link
324 player_id: The roster entry ID to link to
325 team_id: Team ID (optional, fetched from roster if not provided)
326 season_id: Season ID (optional, fetched from roster if not provided)
327 age_group_id: Age group ID from the invite
328 league_id: League ID (optional, fetched from team if not provided)
329 division_id: Division ID (optional, fetched from team if not provided)
330 jersey_number: Jersey number (optional, fetched from roster if not provided)
332 Returns:
333 True if successful, False otherwise
334 """
335 try:
336 # If we don't have the roster details, fetch them
337 if team_id is None or season_id is None or jersey_number is None:
338 roster_response = (
339 self.supabase.table("players")
340 .select("team_id, season_id, jersey_number")
341 .eq("id", player_id)
342 .limit(1)
343 .execute()
344 )
346 if roster_response.data:
347 roster = roster_response.data[0]
348 team_id = team_id or roster.get("team_id")
349 season_id = season_id or roster.get("season_id")
350 jersey_number = jersey_number or roster.get("jersey_number")
352 # If we don't have league/division, fetch from team
353 if team_id and (league_id is None or division_id is None):
354 team_response = (
355 self.supabase.table("teams").select("league_id, division_id").eq("id", team_id).limit(1).execute()
356 )
358 if team_response.data:
359 team = team_response.data[0]
360 league_id = league_id or team.get("league_id")
361 division_id = division_id or team.get("division_id")
363 # Get age_group_id from team_mappings if not provided by the invite
364 if team_id and age_group_id is None:
365 mapping_response = (
366 self.supabase.table("team_mappings")
367 .select("age_group_id")
368 .eq("team_id", team_id)
369 .limit(1)
370 .execute()
371 )
372 if mapping_response.data:
373 age_group_id = mapping_response.data[0]["age_group_id"]
375 # Update the players table to link user_profile_id
376 response = self.supabase.table("players").update({"user_profile_id": user_id}).eq("id", player_id).execute()
378 if response.data:
379 logger.info(f"Linked user {user_id} to roster entry {player_id}")
381 # Also create player_team_history entry for UI roster view
382 if team_id and season_id:
383 self._create_player_team_history(
384 user_id=user_id,
385 team_id=team_id,
386 season_id=season_id,
387 age_group_id=age_group_id,
388 league_id=league_id,
389 division_id=division_id,
390 jersey_number=jersey_number,
391 )
393 # Invalidate caches
394 clear_cache("mt:dao:roster:*")
395 clear_cache("mt:dao:players:*")
396 return True
398 logger.warning(f"Failed to link user {user_id} to roster entry {player_id}")
399 return False
401 except Exception as e:
402 logger.error(f"Error linking user to roster entry: {e}")
403 return False
405 def _create_and_link_roster_entry(
406 self,
407 user_id: str,
408 team_id: int,
409 jersey_number: int,
410 age_group_id: int | None = None,
411 invited_by_user_id: str | None = None,
412 ) -> bool:
413 """
414 Create a new roster entry and link a user account to it, or link to
415 an existing roster entry if one already exists with that jersey number.
416 Also creates a player_team_history entry for the UI roster view.
418 Called when a player accepts an invitation that included a jersey number
419 but no existing player_id.
421 Args:
422 user_id: The user ID to link
423 team_id: The team ID for the roster entry
424 jersey_number: The jersey number for the roster entry
425 age_group_id: The age group ID from the invite
426 invited_by_user_id: The user who created the invite (for created_by field)
428 Returns:
429 True if successful, False otherwise
430 """
431 try:
432 # Get the current season (based on today's date being within start_date and end_date)
433 from datetime import date
435 today = date.today().isoformat()
437 season_response = (
438 self.supabase.table("seasons")
439 .select("id")
440 .lte("start_date", today)
441 .gte("end_date", today)
442 .limit(1)
443 .execute()
444 )
446 if not season_response.data:
447 logger.error("No current season found - cannot create roster entry")
448 return False
450 season_id = season_response.data[0]["id"]
452 # Get team details for league_id and division_id
453 team_response = (
454 self.supabase.table("teams").select("league_id, division_id").eq("id", team_id).limit(1).execute()
455 )
457 league_id = None
458 division_id = None
459 if team_response.data:
460 league_id = team_response.data[0].get("league_id")
461 division_id = team_response.data[0].get("division_id")
463 # Get age_group_id from team_mappings if not provided by the invite
464 if age_group_id is None:
465 mapping_response = (
466 self.supabase.table("team_mappings")
467 .select("age_group_id")
468 .eq("team_id", team_id)
469 .limit(1)
470 .execute()
471 )
472 if mapping_response.data:
473 age_group_id = mapping_response.data[0]["age_group_id"]
475 # Check if a player with this jersey number already exists
476 existing_response = (
477 self.supabase.table("players")
478 .select("id, user_profile_id")
479 .eq("team_id", team_id)
480 .eq("season_id", season_id)
481 .eq("jersey_number", jersey_number)
482 .limit(1)
483 .execute()
484 )
486 if existing_response.data:
487 # Player exists - link user to existing roster entry
488 existing_player = existing_response.data[0]
489 player_id = existing_player["id"]
491 if existing_player.get("user_profile_id"):
492 logger.warning(
493 f"Roster entry {player_id} (jersey #{jersey_number}) "
494 f"already linked to user {existing_player['user_profile_id']}"
495 )
496 return False
498 # Link user to existing roster entry (will also create player_team_history)
499 return self._link_user_to_roster_entry(
500 user_id=user_id,
501 player_id=player_id,
502 team_id=team_id,
503 season_id=season_id,
504 age_group_id=age_group_id,
505 league_id=league_id,
506 division_id=division_id,
507 jersey_number=jersey_number,
508 )
510 # Create new roster entry with user already linked
511 player_data = {
512 "team_id": team_id,
513 "season_id": season_id,
514 "jersey_number": jersey_number,
515 "user_profile_id": user_id,
516 "is_active": True,
517 }
519 if invited_by_user_id:
520 player_data["created_by"] = invited_by_user_id
522 response = self.supabase.table("players").insert(player_data).execute()
524 if response.data:
525 player_id = response.data[0]["id"]
526 logger.info(
527 f"Created roster entry {player_id} with jersey #{jersey_number} "
528 f"for user {user_id} on team {team_id}"
529 )
531 # Also create player_team_history entry for UI roster view
532 self._create_player_team_history(
533 user_id=user_id,
534 team_id=team_id,
535 season_id=season_id,
536 age_group_id=age_group_id,
537 league_id=league_id,
538 division_id=division_id,
539 jersey_number=jersey_number,
540 )
542 # Invalidate caches
543 clear_cache("mt:dao:roster:*")
544 clear_cache("mt:dao:players:*")
545 return True
547 logger.warning(f"Failed to create roster entry for user {user_id}")
548 return False
550 except Exception as e:
551 logger.error(f"Error creating and linking roster entry: {e}")
552 return False
554 def _create_player_team_history(
555 self,
556 user_id: str,
557 team_id: int,
558 season_id: int,
559 age_group_id: int | None = None,
560 league_id: int | None = None,
561 division_id: int | None = None,
562 jersey_number: int | None = None,
563 positions: list[str] | None = None,
564 ) -> bool:
565 """
566 Create a player_team_history entry for the UI roster view.
568 This is the primary table used by the UI to display team rosters.
569 Called when a player accepts an invitation.
571 Args:
572 user_id: The user ID (player_id in the table)
573 team_id: Team ID
574 season_id: Season ID
575 age_group_id: Age group ID
576 league_id: League ID
577 division_id: Division ID
578 jersey_number: Jersey number
579 positions: List of positions (e.g., ['MF', 'FW'])
581 Returns:
582 True if successful, False otherwise
583 """
584 try:
585 # Check if entry already exists for this player/team/season
586 existing = (
587 self.supabase.table("player_team_history")
588 .select("id")
589 .eq("player_id", user_id)
590 .eq("team_id", team_id)
591 .eq("season_id", season_id)
592 .limit(1)
593 .execute()
594 )
596 if existing.data:
597 logger.info(
598 f"player_team_history entry already exists for user {user_id}, team {team_id}, season {season_id}"
599 )
600 return True
602 # Create new entry
603 history_data = {
604 "player_id": user_id,
605 "team_id": team_id,
606 "season_id": season_id,
607 "is_current": True,
608 }
610 if age_group_id is not None:
611 history_data["age_group_id"] = age_group_id
612 if league_id is not None:
613 history_data["league_id"] = league_id
614 if division_id is not None:
615 history_data["division_id"] = division_id
616 if jersey_number is not None:
617 history_data["jersey_number"] = jersey_number
618 if positions:
619 history_data["positions"] = positions
621 response = self.supabase.table("player_team_history").insert(history_data).execute()
623 if response.data:
624 history_id = response.data[0]["id"]
625 logger.info(
626 f"Created player_team_history entry {history_id} for user {user_id} "
627 f"on team {team_id}, season {season_id}, jersey #{jersey_number}"
628 )
629 return True
631 logger.warning(f"Failed to create player_team_history for user {user_id}")
632 return False
634 except Exception as e:
635 logger.error(f"Error creating player_team_history: {e}")
636 return False
638 def _create_team_manager_assignment(
639 self, user_id: str, team_id: int, age_group_id: int, assigned_by_user_id: str
640 ) -> bool:
641 """
642 Create a team_manager_assignments entry to grant team management permissions.
644 Args:
645 user_id: The user being granted management rights
646 team_id: The team they can manage
647 age_group_id: The age group (not stored - table doesn't have this column)
648 assigned_by_user_id: The user who created the invite (not stored)
650 Returns:
651 True if successful, False otherwise
652 """
653 try:
654 # Check if assignment already exists (table has unique constraint on user_id, team_id)
655 existing = (
656 self.supabase.table("team_manager_assignments")
657 .select("id")
658 .eq("user_id", user_id)
659 .eq("team_id", team_id)
660 .execute()
661 )
663 if existing.data:
664 logger.info(f"Team manager assignment already exists for user {user_id}, team {team_id}")
665 return True
667 # Create new assignment (table schema: id, user_id, team_id, created_at)
668 response = (
669 self.supabase.table("team_manager_assignments")
670 .insert(
671 {
672 "user_id": user_id,
673 "team_id": team_id,
674 }
675 )
676 .execute()
677 )
679 if response.data:
680 logger.info(f"Created team manager assignment: user {user_id} -> team {team_id}")
681 return True
683 logger.warning(f"Failed to create team manager assignment for user {user_id}")
684 return False
686 except Exception as e:
687 logger.error(f"Error creating team manager assignment: {e}")
688 return False
690 def get_user_invitations(self, user_id: str) -> list[dict]:
691 """
692 Get all invitations created by a user
694 Args:
695 user_id: ID of the user
697 Returns:
698 List of invitations
699 """
700 try:
701 response = (
702 self.supabase.table("invitations")
703 .select("*, teams(name), age_groups(name), clubs(name)")
704 .eq("invited_by_user_id", user_id)
705 .order("created_at", desc=True)
706 .execute()
707 )
709 return response.data if response.data else []
711 except Exception as e:
712 logger.error(f"Error getting user invitations: {e}")
713 return []
715 def cancel_invitation(self, invite_id: str, user_id: str) -> bool:
716 """
717 Cancel a pending invitation
719 Args:
720 invite_id: ID of the invitation to cancel
721 user_id: ID of the user cancelling (must be creator or admin)
723 Returns:
724 True if successful, False otherwise
725 """
726 try:
727 # Check if user can cancel this invitation
728 response = (
729 self.supabase.table("invitations")
730 .select("invited_by_user_id, status")
731 .eq("id", invite_id)
732 .single()
733 .execute()
734 )
736 if not response.data:
737 return False
739 invitation = response.data
741 # Only pending invitations can be cancelled
742 if invitation["status"] != "pending":
743 return False
745 # Update status to expired
746 response = self.supabase.table("invitations").update({"status": "expired"}).eq("id", invite_id).execute()
748 if response.data: 748 ↛ 752line 748 didn't jump to line 752 because the condition on line 748 was always true
749 logger.info(f"Invitation {invite_id} cancelled by user {user_id}")
750 return True
752 return False
754 except Exception as e:
755 logger.error(f"Error cancelling invitation {invite_id}: {e}")
756 return False
758 def expire_old_invitations(self) -> int:
759 """
760 Expire all old pending invitations
762 Returns:
763 Number of invitations expired
764 """
765 try:
766 # Get all pending invitations that have expired
767 response = (
768 self.supabase.table("invitations")
769 .select("id")
770 .eq("status", "pending")
771 .lt("expires_at", datetime.now(UTC).isoformat())
772 .execute()
773 )
775 if not response.data:
776 return 0
778 # Update all to expired
779 expired_ids = [inv["id"] for inv in response.data]
781 self.supabase.table("invitations").update({"status": "expired"}).in_("id", expired_ids).execute()
783 logger.info(f"Expired {len(expired_ids)} old invitations")
784 return len(expired_ids)
786 except Exception as e:
787 logger.error(f"Error expiring old invitations: {e}")
788 return 0