snapshot: modularize workflow model layer by responsibility

backend/workflows/model_account.py

@@ -0,0 +1,152 @@
+from django.conf import settings
+from django.contrib.auth.hashers import check_password, make_password
+from django.core.validators import FileExtensionValidator
+from django.db import models
+from django.utils import timezone
+from django.utils.translation import gettext_lazy as _
+
+
+class EmployeeProfile(models.Model):
+    full_name = models.CharField(max_length=255)
+    first_name = models.CharField(max_length=100)
+    last_name = models.CharField(max_length=155)
+    department = models.CharField(max_length=255, blank=True)
+    job_title = models.CharField(max_length=255, blank=True)
+    work_email = models.EmailField(unique=True)
+
+    created_at = models.DateTimeField(auto_now_add=True)
+    updated_at = models.DateTimeField(auto_now=True)
+
+    def __str__(self) -> str:
+        return f"{self.full_name} <{self.work_email}>"
+
+
+class UserProfile(models.Model):
+    NOTIFICATION_ONBOARDING_SUCCESS = 'onboarding_success'
+    NOTIFICATION_ONBOARDING_FAILURE = 'onboarding_failure'
+    NOTIFICATION_OFFBOARDING_SUCCESS = 'offboarding_success'
+    NOTIFICATION_OFFBOARDING_FAILURE = 'offboarding_failure'
+    NOTIFICATION_BACKUP_SUCCESS = 'backup_success'
+    NOTIFICATION_BACKUP_FAILURE = 'backup_failure'
+    NOTIFICATION_WELCOME_EMAIL_SUCCESS = 'welcome_email_success'
+    NOTIFICATION_WELCOME_EMAIL_FAILURE = 'welcome_email_failure'
+    NOTIFICATION_TRIAL_ALERTS = 'trial_alerts'
+    NOTIFICATION_SYSTEM_ALERTS = 'system_alerts'
+    NOTIFICATION_PREFERENCE_DEFAULTS = {
+        NOTIFICATION_ONBOARDING_SUCCESS: True,
+        NOTIFICATION_ONBOARDING_FAILURE: True,
+        NOTIFICATION_OFFBOARDING_SUCCESS: True,
+        NOTIFICATION_OFFBOARDING_FAILURE: True,
+        NOTIFICATION_BACKUP_SUCCESS: True,
+        NOTIFICATION_BACKUP_FAILURE: True,
+        NOTIFICATION_WELCOME_EMAIL_SUCCESS: True,
+        NOTIFICATION_WELCOME_EMAIL_FAILURE: True,
+        NOTIFICATION_TRIAL_ALERTS: True,
+        NOTIFICATION_SYSTEM_ALERTS: True,
+    }
+
+    user = models.OneToOneField(settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name='profile')
+    avatar_image = models.FileField(
+        upload_to='profiles/',
+        blank=True,
+        null=True,
+        validators=[FileExtensionValidator(allowed_extensions=['png', 'jpg', 'jpeg', 'webp', 'svg'])],
+    )
+    phone_number = models.CharField(max_length=80, blank=True, default='')
+    mobile_number = models.CharField(max_length=80, blank=True, default='')
+    job_title = models.CharField(max_length=255, blank=True, default='')
+    department = models.CharField(max_length=255, blank=True, default='')
+    location = models.CharField(max_length=255, blank=True, default='')
+    contact_notes = models.CharField(max_length=255, blank=True, default='')
+    totp_secret = models.CharField(max_length=64, blank=True, default='')
+    totp_enabled = models.BooleanField(default=False)
+    totp_confirmed_at = models.DateTimeField(null=True, blank=True)
+    totp_recovery_codes = models.JSONField(default=list, blank=True)
+    notification_preferences = models.JSONField(default=dict, blank=True)
+    updated_at = models.DateTimeField(auto_now=True)
+
+    class Meta:
+        verbose_name = 'User Profile'
+        verbose_name_plural = 'User Profiles'
+
+    def __str__(self) -> str:
+        return getattr(self.user, 'username', '') or str(self.user_id)
+
+    def disable_totp(self) -> None:
+        self.totp_secret = ''
+        self.totp_enabled = False
+        self.totp_confirmed_at = None
+        self.totp_recovery_codes = []
+        self.save(update_fields=['totp_secret', 'totp_enabled', 'totp_confirmed_at', 'totp_recovery_codes', 'updated_at'])
+
+    def enable_totp(self, secret: str, recovery_codes: list[str]) -> None:
+        self.totp_secret = secret
+        self.totp_enabled = True
+        self.totp_confirmed_at = timezone.now()
+        self.set_recovery_codes(recovery_codes)
+        self.save(update_fields=['totp_secret', 'totp_enabled', 'totp_confirmed_at', 'totp_recovery_codes', 'updated_at'])
+
+    def set_recovery_codes(self, recovery_codes: list[str]) -> None:
+        self.totp_recovery_codes = [make_password(code) for code in recovery_codes]
+
+    def consume_recovery_code(self, raw_code: str) -> bool:
+        remaining_hashes = []
+        matched = False
+        for hashed_code in self.totp_recovery_codes or []:
+            if not matched and check_password(raw_code, hashed_code):
+                matched = True
+                continue
+            remaining_hashes.append(hashed_code)
+        if matched:
+            self.totp_recovery_codes = remaining_hashes
+            self.save(update_fields=['totp_recovery_codes', 'updated_at'])
+        return matched
+
+    def get_notification_preferences(self) -> dict[str, bool]:
+        current = self.notification_preferences or {}
+        prefs = dict(self.NOTIFICATION_PREFERENCE_DEFAULTS)
+        for key in prefs:
+            if key in current:
+                prefs[key] = bool(current[key])
+        return prefs
+
+    def notification_enabled(self, event_key: str) -> bool:
+        return bool(self.get_notification_preferences().get(event_key, True))
+
+
+class UserNotification(models.Model):
+    LEVEL_INFO = 'info'
+    LEVEL_SUCCESS = 'success'
+    LEVEL_WARNING = 'warning'
+    LEVEL_ERROR = 'error'
+    LEVEL_CHOICES = [
+        (LEVEL_INFO, _('Info')),
+        (LEVEL_SUCCESS, _('Erfolg')),
+        (LEVEL_WARNING, _('Warnung')),
+        (LEVEL_ERROR, _('Fehler')),
+    ]
+
+    user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE, related_name='notifications')
+    title = models.CharField(max_length=255)
+    body = models.TextField(blank=True, default='')
+    level = models.CharField(max_length=20, choices=LEVEL_CHOICES, default=LEVEL_INFO)
+    link_url = models.CharField(max_length=500, blank=True, default='')
+    created_at = models.DateTimeField(auto_now_add=True)
+    read_at = models.DateTimeField(null=True, blank=True)
+
+    class Meta:
+        ordering = ['-created_at', '-id']
+        verbose_name = 'User Notification'
+        verbose_name_plural = 'User Notifications'
+
+    def __str__(self) -> str:
+        return f'{self.user_id} | {self.level} | {self.title}'
+
+    @property
+    def is_unread(self) -> bool:
+        return self.read_at is None
+
+    def mark_read(self) -> None:
+        if self.read_at is None:
+            self.read_at = timezone.now()
+            self.save(update_fields=['read_at'])