"""Automation Engine models. Three bounded entities: AutomationRule (UUIDModel, TenantScopedModel) A tenant-scoped automation rule. Can be triggered by an event on the internal Event Bus, on a cron schedule (task 3.6), or manually via the API. ``conditions`` and ``actions`` are plain JSON so no schema migration is required when adding new operators or action types. AutomationExecution (UUIDModel, TenantScopedModel) Immutable append-only log of every rule execution. One row per (rule, trigger_event, idempotency_key) triple — the unique constraint ensures retrying the same delivery is a no-op. AutomationTemplate (UUIDModel, TimeStampedModel) Global (cross-tenant) read-only blueprints. Tenants instantiate them into ``AutomationRule`` rows via the ``from-template/`` API. """ from __future__ import annotations from typing import ClassVar from django.db import models from django.utils import timezone from django.utils.translation import gettext_lazy as _ from simorgh.core.models import TenantScopedModel, TimeStampedModel, UUIDModel # --------------------------------------------------------------------------- # Choices # --------------------------------------------------------------------------- class TriggerType(models.TextChoices): EVENT = "event", _("Event") SCHEDULE = "schedule", _("Schedule") MANUAL = "manual", _("Manual") class RuleStatus(models.TextChoices): SUCCESS = "success", _("Success") FAILED = "failed", _("Failed") SKIPPED = "skipped", _("Skipped") class ExecutionStatus(models.TextChoices): RUNNING = "running", _("Running") SUCCESS = "success", _("Success") FAILED = "failed", _("Failed") SKIPPED = "skipped", _("Skipped") # --------------------------------------------------------------------------- # AutomationRule # --------------------------------------------------------------------------- class AutomationRule(UUIDModel, TenantScopedModel): """Tenant-scoped rule — event / schedule / manual trigger. Conditions schema (``conditions`` field) ---------------------------------------- A JSON list of condition objects understood by ``automation.evaluator``: [{"field": "payload.lead.status", "op": "eq", "value": "new"}, ...] All conditions are ANDed. Pass an empty list to match unconditionally. Actions schema (``actions`` field) ------------------------------------ An ordered JSON list of action objects: [{"action": "notifications.send_notification", "params": {"template": "lead_created", "recipient_id": 42}}, ...] Each ``action`` key must match a registered ``ActionSpec.key`` in the ``ActionRegistry``. Unknown keys cause the executor to raise at runtime. Idempotency / versioning ------------------------- ``version`` is a monotonic counter bumped on every PATCH. The executor stamps ``AutomationExecution.rule_version`` so historical executions always reflect the rule state at the time they ran. """ name = models.CharField(_("name"), max_length=255) description = models.TextField(_("description"), blank=True, default="") is_active = models.BooleanField(_("is active"), default=True, db_index=True) # Trigger trigger_type = models.CharField( _("trigger type"), max_length=16, choices=TriggerType.choices, default=TriggerType.EVENT, db_index=True, ) trigger_event = models.CharField( _("trigger event"), max_length=128, blank=True, default="", db_index=True, help_text=_( "Event Bus event name, e.g. 'crm.lead.created'. " "Required when trigger_type='event'." ), ) trigger_schedule = models.CharField( _("trigger schedule"), max_length=128, blank=True, default="", help_text=_( "Standard cron expression, e.g. '0 9 * * 1' (Monday 09:00). " "Required when trigger_type='schedule'." ), ) # Logic conditions = models.JSONField( _("conditions"), default=list, blank=True, help_text=_( "List of condition objects (see module docstring). " "Empty list = unconditional match." ), ) actions = models.JSONField( _("actions"), default=list, blank=True, help_text=_( "Ordered list of {action, params} objects. " "Each action key must be registered in the ActionRegistry." ), ) # Stats (updated by executor) run_count = models.PositiveBigIntegerField(_("run count"), default=0) last_run_at = models.DateTimeField(_("last run at"), null=True, blank=True) last_status = models.CharField( _("last status"), max_length=10, choices=RuleStatus.choices, blank=True, default="", db_index=True, ) # Versioning — bumped on every write so executions can record which # version of the rule was active when they ran. version = models.PositiveIntegerField(_("version"), default=1) class Meta: verbose_name = _("automation rule") verbose_name_plural = _("automation rules") ordering = ("-created_at",) indexes: ClassVar[list[models.Index]] = [ models.Index(fields=("tenant", "trigger_event", "is_active")), models.Index(fields=("tenant", "trigger_type", "is_active")), ] def __str__(self) -> str: return f"[{self.tenant_id}] {self.name}" def save(self, *args, **kwargs) -> None: # type: ignore[override] if not self._state.adding: self.version = models.F("version") + 1 # type: ignore[assignment] super().save(*args, **kwargs) if not self._state.adding: self.refresh_from_db(fields=["version"]) # --------------------------------------------------------------------------- # AutomationExecution # --------------------------------------------------------------------------- class AutomationExecution(UUIDModel, TenantScopedModel): """Immutable log of a single automation rule execution. One row is created per (rule, idempotency_key) pair — the unique constraint on ``(rule, idempotency_key)`` enforces at-most-once semantics so the Celery task can be retried safely. ``idempotency_key`` is derived by the executor from the event delivery ID (or a UUID for manual runs). ``actions_executed`` stores per-action results:: [ {"action": "notifications.send_notification", "status": "ok", "result": null, "error": null}, {"action": "iam.assign_role", "status": "error", "result": null, "error": "Role 'manager' not found"}, ] """ rule = models.ForeignKey( AutomationRule, on_delete=models.SET_NULL, null=True, blank=True, related_name="executions", verbose_name=_("rule"), help_text=_("Null if the rule was deleted after this execution."), ) rule_version = models.PositiveIntegerField( _("rule version"), default=0, help_text=_("Snapshot of AutomationRule.version at execution time."), ) trigger_event = models.CharField( _("trigger event"), max_length=128, blank=True, default="", db_index=True, ) trigger_payload = models.JSONField(_("trigger payload"), default=dict) idempotency_key = models.CharField( _("idempotency key"), max_length=128, db_index=True, help_text=_("Unique delivery ID — prevents duplicate execution on retry."), ) started_at = models.DateTimeField(_("started at"), default=timezone.now, db_index=True) finished_at = models.DateTimeField(_("finished at"), null=True, blank=True) status = models.CharField( _("status"), max_length=10, choices=ExecutionStatus.choices, default=ExecutionStatus.RUNNING, db_index=True, ) actions_executed = models.JSONField( _("actions executed"), default=list, help_text=_("Per-action result list. See module docstring for schema."), ) error_message = models.TextField(_("error message"), blank=True, default="") class Meta: verbose_name = _("automation execution") verbose_name_plural = _("automation executions") ordering = ("-started_at",) constraints: ClassVar[list[models.BaseConstraint]] = [ models.UniqueConstraint( fields=("rule", "idempotency_key"), name="automation_exec_unique_rule_idem_key", condition=models.Q(rule__isnull=False), ), ] indexes: ClassVar[list[models.Index]] = [ models.Index(fields=("tenant", "status", "started_at")), models.Index(fields=("rule", "started_at")), ] def __str__(self) -> str: return f"Execution {self.public_id} [{self.status}]" @property def duration_seconds(self) -> float | None: if self.finished_at is None: return None return (self.finished_at - self.started_at).total_seconds() # --------------------------------------------------------------------------- # AutomationTemplate # --------------------------------------------------------------------------- class AutomationTemplate(UUIDModel, TimeStampedModel): """Global (cross-tenant) read-only automation blueprint. System templates (``is_system=True``) are seeded by platform migrations and cannot be modified via the API. Tenant-created custom templates have ``is_system=False``. Tenants instantiate templates into ``AutomationRule`` rows via the ``POST /api/v1/automation/rules/from-template/{id}/`` endpoint which deep-copies ``default_conditions`` and ``default_actions`` into the new rule and lets the tenant customise from there. """ name = models.CharField(_("name"), max_length=255) description = models.TextField(_("description"), blank=True, default="") category = models.CharField( _("category"), max_length=64, blank=True, default="", db_index=True, help_text=_("Module slug, e.g. 'crm', 'helpdesk', 'workflow'."), ) trigger_type = models.CharField( _("trigger type"), max_length=16, choices=TriggerType.choices, default=TriggerType.EVENT, ) trigger_event = models.CharField( _("trigger event"), max_length=128, blank=True, default="", ) default_conditions = models.JSONField(_("default conditions"), default=list, blank=True) default_actions = models.JSONField(_("default actions"), default=list, blank=True) is_system = models.BooleanField( _("is system"), default=False, db_index=True, help_text=_("System templates are seeded by migrations and are read-only."), ) class Meta: verbose_name = _("automation template") verbose_name_plural = _("automation templates") ordering = ("category", "name") indexes: ClassVar[list[models.Index]] = [ models.Index(fields=("category", "is_system")), ] def __str__(self) -> str: return f"{self.category}: {self.name}" # --------------------------------------------------------------------------- # ScheduledRule # --------------------------------------------------------------------------- class ScheduledRule(UUIDModel, TenantScopedModel): """Links an ``AutomationRule`` to a cron schedule. The Celery beat task ``tick_scheduled_rules`` runs every minute and dispatches ``execute_automation_rule_task`` for every active ``ScheduledRule`` whose ``next_run_at`` is in the past. Cron format ----------- Standard 5-field cron: ``"minute hour day-of-month month day-of-week"`` e.g. ``"0 9 * * 1"`` = every Monday at 09:00. Parsed by the ``croniter`` library. Timezone -------- ``timezone`` is an IANA timezone string (``"Asia/Tehran"``, ``"UTC"``). ``next_run_at`` is always stored as UTC; the cron expression is evaluated in the declared timezone so "9 AM" means 9 AM in that timezone. """ rule = models.ForeignKey( AutomationRule, on_delete=models.CASCADE, related_name="scheduled_rules", verbose_name=_("rule"), ) cron_expression = models.CharField( _("cron expression"), max_length=128, help_text=_("Standard 5-field cron, e.g. '0 9 * * 1' (Monday 09:00)."), ) timezone = models.CharField( _("timezone"), max_length=64, default="UTC", help_text=_("IANA timezone, e.g. 'Asia/Tehran'. Cron is evaluated in this zone."), ) next_run_at = models.DateTimeField( _("next run at"), db_index=True, help_text=_("UTC datetime of the next scheduled execution."), ) last_run_at = models.DateTimeField( _("last run at"), null=True, blank=True, ) is_active = models.BooleanField( _("is active"), default=True, db_index=True, ) class Meta: verbose_name = _("scheduled rule") verbose_name_plural = _("scheduled rules") ordering = ("next_run_at",) indexes: ClassVar[list[models.Index]] = [ models.Index(fields=("is_active", "next_run_at")), models.Index(fields=("tenant", "is_active")), ] def __str__(self) -> str: return f"[{self.tenant_id}] {self.rule_id} @ {self.cron_expression}" # --------------------------------------------------------------------------- # WebhookEndpoint (3.7.1) # --------------------------------------------------------------------------- class DeliveryStatus(models.TextChoices): PENDING = "pending", _("Pending") SUCCESS = "success", _("Success") FAILED = "failed", _("Failed") RETRYING = "retrying", _("Retrying") class WebhookEndpoint(UUIDModel, TenantScopedModel): """Tenant-scoped outbound webhook endpoint. When an event is dispatched on the internal Event Bus, the catch-all handler in ``automation.events`` fans out to every active endpoint whose ``events`` list contains the event name. A ``WebhookDelivery`` row is created per (endpoint, event) pair and a Celery task is enqueued to perform the actual HTTP POST. Secret / HMAC ------------- If ``secret`` is non-empty, every delivery includes an ``X-Simorgh-Signature: sha256=`` header computed with HMAC-SHA256. The secret is stored as plain text in the DB but is *never* returned by any API response (write-only field). Failure circuit-breaker ----------------------- ``failure_count`` tracks consecutive failed deliveries. When it reaches ``webhook_service.MAX_FAILURES`` (5) the endpoint is automatically deactivated so it no longer receives new events. ``failure_count`` is reset to 0 on every successful delivery. """ url = models.URLField( _("url"), max_length=2048, help_text=_("HTTPS endpoint that will receive POST requests."), ) name = models.CharField(_("name"), max_length=255) description = models.TextField(_("description"), blank=True, default="") secret = models.CharField( _("secret"), max_length=255, blank=True, default="", help_text=_( "HMAC-SHA256 signing secret. " "Stored in plaintext — never returned by the API." ), ) events = models.JSONField( _("events"), default=list, help_text=_( "List of Event Bus event names to subscribe to, " "e.g. [\"crm.lead.created\", \"helpdesk.ticket.closed\"]." ), ) is_active = models.BooleanField(_("is active"), default=True, db_index=True) headers = models.JSONField( _("headers"), default=dict, help_text=_("Custom HTTP headers included in every delivery request."), ) timeout_seconds = models.PositiveSmallIntegerField( _("timeout seconds"), default=30, ) last_success_at = models.DateTimeField(_("last success at"), null=True, blank=True) last_failure_at = models.DateTimeField(_("last failure at"), null=True, blank=True) failure_count = models.PositiveIntegerField(_("failure count"), default=0) class Meta: verbose_name = _("webhook endpoint") verbose_name_plural = _("webhook endpoints") ordering = ("-created_at",) indexes: ClassVar[list[models.Index]] = [ models.Index(fields=("tenant", "is_active")), ] def __str__(self) -> str: return f"[{self.tenant_id}] {self.name} → {self.url}" # --------------------------------------------------------------------------- # WebhookDelivery (3.7.2) # --------------------------------------------------------------------------- class WebhookDelivery(UUIDModel, TenantScopedModel): """A single delivery attempt log for an outbound webhook. One row is created by ``webhook_service.deliver_webhook`` for each (endpoint, event) pair. The Celery task ``send_webhook_task`` performs the HTTP call and updates this row. Retry lifecycle --------------- * Initial delivery: ``status=pending``, ``attempt=0``. * On failure: ``status=retrying``, ``attempt+=1``, ``next_retry_at`` set. * After all retries exhausted: ``status=failed``. * On success at any attempt: ``status=success``, ``delivered_at`` set. """ endpoint = models.ForeignKey( WebhookEndpoint, on_delete=models.CASCADE, related_name="deliveries", verbose_name=_("endpoint"), ) event_name = models.CharField(_("event name"), max_length=128, db_index=True) payload = models.JSONField(_("payload"), default=dict) status = models.CharField( _("status"), max_length=10, choices=DeliveryStatus.choices, default=DeliveryStatus.PENDING, db_index=True, ) attempt = models.PositiveSmallIntegerField(_("attempt"), default=0) response_status = models.PositiveSmallIntegerField( _("response status"), null=True, blank=True, help_text=_("HTTP response status code."), ) response_body = models.TextField( _("response body"), blank=True, default="", help_text=_("First 4 KB of response body (or error message)."), ) delivered_at = models.DateTimeField(_("delivered at"), null=True, blank=True) next_retry_at = models.DateTimeField( _("next retry at"), null=True, blank=True, db_index=True, ) class Meta: verbose_name = _("webhook delivery") verbose_name_plural = _("webhook deliveries") ordering = ("-created_at",) indexes: ClassVar[list[models.Index]] = [ models.Index(fields=("endpoint", "status")), models.Index(fields=("tenant", "event_name")), ] def __str__(self) -> str: return f"Delivery {self.public_id} [{self.status}] → {self.event_name}"