swiftops-backend / src /app /models /invitation.py
kamau1's picture
fix(invite): implement project-level invitations with validation and notifications
cd8df9d
"""
User Invitation Model
"""
from sqlalchemy import Column, String, Boolean, DateTime, Text, ForeignKey
from sqlalchemy.dialects.postgresql import UUID, JSONB
from app.models.base import BaseModel
class UserInvitation(BaseModel):
"""
User Invitation model - Tracks invitations before Supabase Auth account creation
Users are created in Supabase Auth ONLY after accepting invitation.
This table manages the invitation lifecycle: pending → accepted/expired/cancelled
"""
__tablename__ = "user_invitations"
# Invitation Details
email = Column(String(255), nullable=False)
phone = Column(String(50), nullable=True)
invited_name = Column(String(200), nullable=True) # Optional full name of invitee
invited_role = Column(String(50), nullable=False) # app_role ENUM
# Organization Links
client_id = Column(UUID(as_uuid=True), ForeignKey('clients.id', ondelete='SET NULL'), nullable=True)
contractor_id = Column(UUID(as_uuid=True), ForeignKey('contractors.id', ondelete='SET NULL'), nullable=True)
# Project Context (for project-specific invitations)
project_id = Column(UUID(as_uuid=True), ForeignKey('projects.id', ondelete='CASCADE'), nullable=True)
project_role_id = Column(UUID(as_uuid=True), ForeignKey('project_roles.id', ondelete='SET NULL'), nullable=True)
project_region_id = Column(UUID(as_uuid=True), ForeignKey('project_regions.id', ondelete='SET NULL'), nullable=True)
project_subcontractor_id = Column(UUID(as_uuid=True), ForeignKey('project_subcontractors.id', ondelete='SET NULL'), nullable=True)
# Token & Status
token = Column(Text, nullable=False, unique=True)
status = Column(String(50), default='pending') # invitation_status ENUM
invitation_method = Column(String(50), default='whatsapp') # invitation_method ENUM
# Lifecycle
invited_by_user_id = Column(UUID(as_uuid=True), ForeignKey('users.id', ondelete='CASCADE'), nullable=False)
invited_at = Column(DateTime(timezone=True), nullable=False)
expires_at = Column(DateTime(timezone=True), nullable=False)
accepted_at = Column(DateTime(timezone=True), nullable=True)
# Delivery Tracking
whatsapp_sent = Column(Boolean, default=False)
whatsapp_sent_at = Column(DateTime(timezone=True), nullable=True)
whatsapp_error = Column(Text, nullable=True)
email_sent = Column(Boolean, default=False)
email_sent_at = Column(DateTime(timezone=True), nullable=True)
email_error = Column(Text, nullable=True)
# Metadata
invitation_metadata = Column(JSONB, default={})
def __repr__(self):
return f"<UserInvitation(email='{self.email}', role='{self.invited_role}', status='{self.status}')>"
@property
def is_expired(self) -> bool:
"""Check if invitation has expired"""
from datetime import datetime, timezone
return self.expires_at < datetime.now(timezone.utc)
@property
def is_pending(self) -> bool:
"""Check if invitation is still pending"""
return self.status == 'pending' and not self.is_expired
@property
def organization_id(self):
"""Get the organization ID (client or contractor)"""
return self.client_id or self.contractor_id
@property
def organization_type(self) -> str:
"""Get organization type"""
if self.client_id:
return 'client'
elif self.contractor_id:
return 'contractor'
return 'platform'
@property
def is_project_invitation(self) -> bool:
"""Check if this is a project-specific invitation"""
return self.project_id is not None