from django.contrib.auth.models import AbstractUser from django.db import models from .managers import UserManager class User(AbstractUser): """Custom user — email is the login field.""" class Role(models.TextChoices): ADMIN = "ADMIN", "Admin" RECRUITER = "RECRUITER", "Recruiter" INTERVIEWER = "INTERVIEWER", "Interviewer" CANDIDATE = "CANDIDATE", "Candidate" USER = "USER", "User (legacy)" email = models.EmailField(unique=True) # Roles are dynamic: the value is the name of a Django Group. The Role class # above only holds the built-in seed roles/constants — admins can create more # roles (groups) at runtime via the API, so there is no fixed `choices` here. role = models.CharField(max_length=50, default=Role.CANDIDATE) employee_id = models.CharField(max_length=50, blank=True, null=True, unique=True) department = models.CharField(max_length=150, blank=True, default="") full_name = models.CharField(max_length=255, blank=True) phone = models.CharField(max_length=20, blank=True) location = models.CharField(max_length=255, blank=True) skills = models.TextField(blank=True) experience_years = models.PositiveIntegerField(default=0) bio = models.TextField(blank=True) resume = models.FileField(upload_to="resumes/", null=True, blank=True) class MFAMethod(models.TextChoices): NONE = "none", "None" TOTP = "totp", "Authenticator App" EMAIL = "email", "Email OTP" SMS = "sms", "Mobile OTP" mfa_enabled = models.BooleanField(default=False) mfa_method = models.CharField(max_length=10, choices=MFAMethod.choices, default=MFAMethod.NONE) totp_secret = models.CharField(max_length=32, blank=True) password_last_changed_at = models.DateTimeField(null=True, blank=True) password_expiry_date = models.DateTimeField(null=True, blank=True) password_expired = models.BooleanField(default=False) USERNAME_FIELD = "email" REQUIRED_FIELDS = ["username"] objects = UserManager() class Meta: db_table = "users" def __str__(self): return self.email @property def is_admin(self): return self.role == self.Role.ADMIN @property def password_policy_exempt(self): return self.is_admin def sync_role_group(self): """Ensure the role group is always present without wiping extra groups an admin has manually assigned via GroupAssignmentModal.""" from django.contrib.auth.models import Group role_group, _ = Group.objects.get_or_create(name=self.role) if not self.groups.filter(pk=role_group.pk).exists(): self.groups.add(role_group) def save(self, *args, **kwargs): is_new = self.pk is None password_changed = False if not is_new: try: orig = User.objects.get(pk=self.pk) password_changed = orig.password != self.password except User.DoesNotExist: password_changed = True else: password_changed = True if password_changed: from django.utils import timezone from apps.authentication.models import PasswordPolicy self.password_last_changed_at = timezone.now() if self.password_policy_exempt: self.password_expiry_date = None self.password_expired = False else: self.password_expiry_date = self.password_last_changed_at + PasswordPolicy.load().get_expiry_timedelta() self.password_expired = False # If the caller passed update_fields, make sure the expiry reset is # actually persisted too — otherwise the popup would keep showing # after a successful password change. update_fields = kwargs.get("update_fields") if update_fields is not None: kwargs["update_fields"] = list( set(update_fields) | { "password", "password_last_changed_at", "password_expiry_date", "password_expired", } ) if not self.employee_id: for _ in range(5): candidate_id = generate_employee_id() if not User.objects.filter(employee_id=candidate_id).exists(): self.employee_id = candidate_id break update_fields = kwargs.get("update_fields") if update_fields is not None: kwargs["update_fields"] = list(set(update_fields) | {"employee_id"}) super().save(*args, **kwargs) if password_changed: PasswordHistory.objects.create(user=self, password_hash=self.password) def check_password_expiry(self): from django.utils import timezone if self.password_policy_exempt: if self.password_expired or self.password_expiry_date is not None: self.password_expired = False self.password_expiry_date = None self.save(update_fields=["password_expired", "password_expiry_date"]) return False if self.password_expiry_date and timezone.now() >= self.password_expiry_date: if not self.password_expired: self.password_expired = True self.save(update_fields=["password_expired"]) return self.password_expired class PasswordHistory(models.Model): user = models.ForeignKey(User, on_delete=models.CASCADE, related_name="password_histories") password_hash = models.CharField(max_length=128) created_at = models.DateTimeField(auto_now_add=True) class Meta: db_table = "password_histories" ordering = ["-created_at"] class EmployeeIdSequence(models.Model): """Monotonic counter backing auto-generated Employee IDs (EMP-0001, …). A single row holds the last-issued number. Handing out the next value under `select_for_update()` makes generation concurrency-safe (no duplicate IDs on simultaneous creates) and, because the counter only ever increases, numbers are never reused even after a user is deleted. """ last_value = models.PositiveIntegerField(default=0) class Meta: db_table = "employee_id_sequence" @classmethod def next_value(cls) -> str: from django.db import transaction with transaction.atomic(): row, _ = cls.objects.select_for_update().get_or_create(pk=1) row.last_value += 1 row.save(update_fields=["last_value"]) return f"EMP-{row.last_value:04d}" def generate_employee_id() -> str: """Return the next unique Employee ID (e.g. 'EMP-0001').""" return EmployeeIdSequence.next_value() from django.db.models.signals import post_save from django.dispatch import receiver from apps.authentication import mcp_client @receiver(post_save, sender=User) def sync_user_role_group(sender, instance, **kwargs): try: instance.sync_role_group() except Exception as e: import logging logging.getLogger(__name__).error( f"Failed to sync role group for {instance.email}: {e}" ) @receiver(post_save, sender=User) def sync_user_to_mcp(sender, instance, **kwargs): if instance.password: try: mcp_client.sync_user( email=instance.email, password_hash=instance.password, role=instance.role ) except Exception as e: import logging logger = logging.getLogger(__name__) logger.error(f"Failed to sync user {instance.email} to MCP: {e}") from django.db.models.signals import post_delete @receiver(post_delete, sender=User) def delete_user_from_mcp(sender, instance, **kwargs): try: mcp_client.delete_user(email=instance.email) except Exception as e: import logging logger = logging.getLogger(__name__) logger.error(f"Failed to delete user {instance.email} from MCP: {e}")