from typing import override from django.core.validators import RegexValidator from django.db import models from django.utils.translation import gettext_lazy as _ from apps.core.models import BaseModel EVENT_TYPE_KEY_PATTERN = r"^[A-Za-z][A-Za-z0-9_]*$" class EventType(BaseModel): name = models.CharField( max_length=100, unique=True, verbose_name=_("name"), validators=[ RegexValidator( regex=EVENT_TYPE_KEY_PATTERN, message=( "Event type name must follow snake_case, " "camelCase, or PascalCase." ), ) ], ) display_name = models.CharField( max_length=200, verbose_name=_("display name"), ) description = models.TextField( blank=True, verbose_name=_("description"), ) is_exposure = models.BooleanField( default=False, verbose_name=_("is exposure"), help_text=_( "When True, this event type represents an exposure " "(fact of showing a variant to a user)." ), ) requires_exposure = models.BooleanField( default=False, verbose_name=_("requires exposure"), help_text=_( "When True, events of this type are only attributed " "if a matching exposure exists for the same decision_id." ), ) required_fields = models.JSONField( default=list, blank=True, verbose_name=_("required fields"), help_text=_( "List of property field names that must be present " "in event properties." ), ) is_active = models.BooleanField( default=True, db_index=True, verbose_name=_("is active"), ) created_at = models.DateTimeField( auto_now_add=True, verbose_name=_("created at"), ) updated_at = models.DateTimeField( auto_now=True, verbose_name=_("updated at"), ) class Meta: verbose_name = _("event type") verbose_name_plural = _("event types") ordering = ["name"] @override def __str__(self) -> str: return self.name class Exposure(BaseModel): decision_id = models.CharField( max_length=100, unique=True, verbose_name=_("decision ID"), ) experiment_id = models.UUIDField( null=True, blank=True, verbose_name=_("experiment ID"), db_index=True, ) variant_id = models.UUIDField( null=True, blank=True, verbose_name=_("variant ID"), ) subject_id = models.CharField( max_length=200, db_index=True, verbose_name=_("subject ID"), ) timestamp = models.DateTimeField( verbose_name=_("event timestamp"), ) created_at = models.DateTimeField( auto_now_add=True, verbose_name=_("created at"), ) class Meta: verbose_name = _("exposure") verbose_name_plural = _("exposures") ordering = ["-timestamp"] indexes = [ models.Index( fields=["experiment_id", "variant_id", "timestamp"], name="idx_exposure_exp_var_ts", ), ] @override def __str__(self) -> str: return f"Exposure({self.decision_id})" class Decision(BaseModel): decision_id = models.CharField( max_length=100, unique=True, verbose_name=_("decision ID"), ) flag_key = models.CharField( max_length=100, verbose_name=_("flag key"), ) subject_id = models.CharField( max_length=200, db_index=True, verbose_name=_("subject ID"), ) experiment_id = models.UUIDField( null=True, blank=True, verbose_name=_("experiment ID"), ) variant_id = models.UUIDField( null=True, blank=True, verbose_name=_("variant ID"), ) value = models.CharField( max_length=500, blank=True, verbose_name=_("resolved value"), ) reason = models.CharField( max_length=50, verbose_name=_("reason"), ) created_at = models.DateTimeField( auto_now_add=True, verbose_name=_("created at"), ) class Meta: verbose_name = _("decision") verbose_name_plural = _("decisions") ordering = ["-created_at"] indexes = [ models.Index( fields=["flag_key", "subject_id"], name="idx_decision_flag_subject", ), ] @override def __str__(self) -> str: return f"Decision({self.decision_id}, {self.flag_key})" class Event(BaseModel): event_id = models.CharField( max_length=200, unique=True, verbose_name=_("event ID"), help_text=_("Client-provided idempotency key"), ) event_type = models.ForeignKey( EventType, on_delete=models.PROTECT, related_name="events", verbose_name=_("event type"), ) decision_id = models.CharField( max_length=100, db_index=True, verbose_name=_("decision ID"), ) subject_id = models.CharField( max_length=200, db_index=True, verbose_name=_("subject ID"), ) timestamp = models.DateTimeField( verbose_name=_("event timestamp"), ) properties = models.JSONField( default=dict, blank=True, verbose_name=_("properties"), ) is_attributed = models.BooleanField( default=True, db_index=True, verbose_name=_("is attributed"), help_text=_( "False when event requires exposure but none was found yet." ), ) created_at = models.DateTimeField( auto_now_add=True, verbose_name=_("created at"), ) class Meta: verbose_name = _("event") verbose_name_plural = _("events") ordering = ["-timestamp"] indexes = [ models.Index( fields=["decision_id", "event_type"], name="idx_event_decision_type", ), models.Index( fields=["event_type", "subject_id", "timestamp"], name="idx_event_type_subj_ts", ), ] @override def __str__(self) -> str: return f"Event({self.event_id}, {self.event_type})" class PendingEvent(BaseModel): event_id = models.CharField( max_length=200, unique=True, verbose_name=_("event ID"), ) event_type = models.ForeignKey( EventType, on_delete=models.PROTECT, related_name="pending_events", verbose_name=_("event type"), ) decision_id = models.CharField( max_length=100, db_index=True, verbose_name=_("decision ID"), ) subject_id = models.CharField( max_length=200, verbose_name=_("subject ID"), ) timestamp = models.DateTimeField( verbose_name=_("event timestamp"), ) properties = models.JSONField( default=dict, blank=True, verbose_name=_("properties"), ) expires_at = models.DateTimeField( db_index=True, verbose_name=_("expires at"), ) created_at = models.DateTimeField( auto_now_add=True, verbose_name=_("created at"), ) class Meta: verbose_name = _("pending event") verbose_name_plural = _("pending events") ordering = ["-created_at"] indexes = [ models.Index( fields=["decision_id"], name="idx_pending_decision", ), ] @override def __str__(self) -> str: return f"PendingEvent({self.event_id}, {self.decision_id})"