"""Celery tasks for the Automation Engine. ``execute_automation_rule_task`` Main worker task. Called by ``execute_rule_async()`` and the event bus subscription handler. Supports up to 3 retries with exponential back-off (60 s → 120 s → 240 s). The task is intentionally thin — all business logic lives in ``executor.execute_rule``. Celery only handles scheduling, retry, and error isolation. ``tick_scheduled_rules`` Periodic task registered with Celery Beat. Runs every minute and dispatches ``execute_automation_rule_task`` for every active ``ScheduledRule`` whose ``next_run_at`` is past-due. After dispatching, it advances ``next_run_at`` to the next cron iteration using ``compute_next_run``. ``compute_next_run`` Pure helper: given a cron expression, timezone string, and a *base* datetime, returns the next UTC datetime when the rule should run. Wraps ``croniter`` so callers never import it directly. """ from __future__ import annotations import datetime import uuid from typing import Any import structlog from celery import shared_task _log = structlog.get_logger("simorgh.automation.tasks") # --------------------------------------------------------------------------- # Cron helper (3.6.4) # --------------------------------------------------------------------------- def compute_next_run( cron_expression: str, timezone_name: str = "UTC", base: datetime.datetime | None = None, ) -> datetime.datetime: """Return the next UTC datetime for *cron_expression* after *base*. Parameters ---------- cron_expression: Standard 5-field cron string, e.g. ``"*/5 * * * *"``. timezone_name: IANA timezone in which the cron expression should be interpreted, e.g. ``"Asia/Tehran"``. Defaults to ``"UTC"``. base: Datetime to compute the *next* iteration after. Defaults to ``datetime.now(tz)`` (i.e. the first upcoming iteration). Returns ------- datetime UTC-aware datetime of the next cron iteration. Raises ------ ValueError If *cron_expression* is invalid or *timezone_name* is unknown. """ try: import zoneinfo except ImportError: from backports import zoneinfo # type: ignore[no-redef] from croniter import CroniterBadCronError, croniter try: tz = zoneinfo.ZoneInfo(timezone_name) except (zoneinfo.ZoneInfoNotFoundError, KeyError) as exc: raise ValueError(f"Unknown timezone: {timezone_name!r}") from exc if base is None: base = datetime.datetime.now(tz=tz) elif base.tzinfo is None: # Naive datetime — assume UTC then convert. import datetime as _dt base = base.replace(tzinfo=_dt.timezone.utc).astimezone(tz) else: base = base.astimezone(tz) try: it = croniter(cron_expression, base) next_dt_local: datetime.datetime = it.get_next(datetime.datetime) except CroniterBadCronError as exc: raise ValueError(f"Invalid cron expression: {cron_expression!r}") from exc # Always store / return UTC. import datetime as _dt return next_dt_local.astimezone(_dt.timezone.utc) # --------------------------------------------------------------------------- # execute_automation_rule_task (3.5.2 / 3.5.4) # --------------------------------------------------------------------------- @shared_task( bind=True, max_retries=3, name="automation.execute_rule", # Soft time limit keeps a runaway task from blocking a worker slot. soft_time_limit=120, time_limit=180, ) def execute_automation_rule_task( self: Any, rule_id: int, trigger_payload: dict[str, Any], *, actor_id: int | None = None, trigger_event: str = "", idempotency_key: str = "", ) -> None: """Load the rule and call ``execute_rule`` synchronously. Retry policy ------------ On any exception, the task retries up to 3 times with exponential countdown: 60 s, 120 s, 240 s. After the third retry, the exception propagates and Celery marks the task as FAILURE. Idempotency ----------- The ``idempotency_key`` is forwarded to ``execute_rule`` which uses it to get-or-create the ``AutomationExecution`` row. Retrying the Celery task with the same key is therefore safe — only one execution row is ever created per (rule, key) pair. """ from simorgh.apps.automation.executor import execute_rule from simorgh.apps.automation.models import AutomationRule try: rule = AutomationRule.objects.get(pk=rule_id, is_active=True) except AutomationRule.DoesNotExist: # Rule deleted or deactivated after the task was enqueued — skip. _log.info( "automation.task.rule_not_found", rule_id=rule_id, ) return try: execute_rule( rule, trigger_payload, actor_id=actor_id, trigger_event=trigger_event, idempotency_key=idempotency_key, ) except Exception as exc: attempt = self.request.retries countdown = 60 * (2 ** attempt) # 60 s → 120 s → 240 s _log.warning( "automation.task.retry", rule_id=rule_id, attempt=attempt, countdown=countdown, error=str(exc), ) raise self.retry(exc=exc, countdown=countdown) # --------------------------------------------------------------------------- # tick_scheduled_rules (3.6.3) # --------------------------------------------------------------------------- @shared_task( name="automation.tick_scheduled_rules", # No retries — if this minute's tick fails, the next minute will pick up. max_retries=0, soft_time_limit=50, time_limit=60, ) def tick_scheduled_rules() -> None: """Dispatch due ``ScheduledRule`` rows and advance their ``next_run_at``. This task is registered as a Celery Beat periodic task (every minute). For each active ``ScheduledRule`` with ``next_run_at <= now(UTC)``: 1. Generate a fresh idempotency key. 2. Call ``execute_automation_rule_task.delay(...)`` with the rule's associated ``AutomationRule.pk``. 3. Advance ``next_run_at`` to the next cron iteration via ``compute_next_run``. 4. Stamp ``last_run_at``. The function is intentionally defensive: a bad cron expression or missing rule causes the row to be skipped (not the whole batch). """ from django.utils import timezone from simorgh.apps.automation.models import ScheduledRule now = timezone.now() due = ScheduledRule.objects.select_related("rule").filter( is_active=True, next_run_at__lte=now, rule__is_active=True, ) for scheduled in due: rule = scheduled.rule idem_key = str(uuid.uuid4()) try: execute_automation_rule_task.delay( rule.pk, {}, trigger_event="schedule", idempotency_key=idem_key, ) except Exception as exc: _log.warning( "automation.tick.dispatch_failed", scheduled_rule_id=scheduled.pk, rule_id=rule.pk, error=str(exc), ) continue # Advance next_run_at using the cron expression. try: next_run = compute_next_run( scheduled.cron_expression, scheduled.timezone, base=now, ) except ValueError as exc: _log.error( "automation.tick.bad_cron", scheduled_rule_id=scheduled.pk, cron=scheduled.cron_expression, error=str(exc), ) # Deactivate the row so it doesn't block every tick. ScheduledRule.objects.filter(pk=scheduled.pk).update(is_active=False) continue ScheduledRule.objects.filter(pk=scheduled.pk).update( next_run_at=next_run, last_run_at=now, ) _log.info( "automation.tick.dispatched", scheduled_rule_id=scheduled.pk, rule_id=rule.pk, next_run_at=next_run.isoformat(), ) # --------------------------------------------------------------------------- # send_webhook_task (3.7.4) # --------------------------------------------------------------------------- @shared_task( name="automation.send_webhook", max_retries=0, # Retries are self-scheduled via apply_async countdown. soft_time_limit=60, time_limit=90, ) def send_webhook_task(delivery_id: int) -> None: """Perform the HTTP delivery for a ``WebhookDelivery`` row. Retry logic ----------- On failure the task computes the next retry countdown from ``webhook_service.RETRY_DELAYS`` and calls ``apply_async`` itself. After all slots are exhausted the delivery is marked ``failed``. Circuit-breaker --------------- When ``WebhookEndpoint.failure_count`` reaches ``webhook_service.MAX_FAILURES`` (5) the endpoint is deactivated automatically. Idempotency ----------- If the delivery is already ``success`` or ``failed`` the task returns immediately (safe to retry at the Celery level). """ from django.utils import timezone from simorgh.apps.automation.models import DeliveryStatus, WebhookDelivery, WebhookEndpoint from simorgh.apps.automation.webhook_service import ( MAX_FAILURES, RETRY_DELAYS, do_send, ) try: delivery = WebhookDelivery.objects.select_related("endpoint").get(pk=delivery_id) except WebhookDelivery.DoesNotExist: _log.warning("webhook.task.delivery_not_found", delivery_id=delivery_id) return if delivery.status in (DeliveryStatus.SUCCESS, DeliveryStatus.FAILED): return # Already terminal — idempotent no-op. endpoint = delivery.endpoint if not endpoint.is_active: # Endpoint was deactivated after this task was enqueued. WebhookDelivery.objects.filter(pk=delivery_id).update( status=DeliveryStatus.FAILED, response_body="Endpoint deactivated.", ) return now = timezone.now() attempt = delivery.attempt + 1 success, status_code, resp_body = do_send(delivery) if success: WebhookDelivery.objects.filter(pk=delivery_id).update( status=DeliveryStatus.SUCCESS, attempt=attempt, response_status=status_code, response_body=resp_body[:4096], delivered_at=now, next_retry_at=None, ) WebhookEndpoint.objects.filter(pk=endpoint.pk).update( last_success_at=now, failure_count=0, ) _log.info( "webhook.delivered", delivery_id=delivery_id, endpoint_id=endpoint.pk, status=status_code, ) return # --- Failed delivery --- new_failure_count = endpoint.failure_count + 1 if attempt <= len(RETRY_DELAYS): delay = RETRY_DELAYS[attempt - 1] next_retry = now + datetime.timedelta(seconds=delay) new_status = DeliveryStatus.RETRYING else: delay = 0 next_retry = None new_status = DeliveryStatus.FAILED WebhookDelivery.objects.filter(pk=delivery_id).update( status=new_status, attempt=attempt, response_status=status_code, response_body=(resp_body or "")[:4096], next_retry_at=next_retry, ) endpoint_update: dict[str, Any] = { "last_failure_at": now, "failure_count": new_failure_count, } if new_failure_count >= MAX_FAILURES: endpoint_update["is_active"] = False _log.warning( "webhook.endpoint_disabled", endpoint_id=endpoint.pk, failure_count=new_failure_count, ) WebhookEndpoint.objects.filter(pk=endpoint.pk).update(**endpoint_update) _log.warning( "webhook.delivery_failed", delivery_id=delivery_id, endpoint_id=endpoint.pk, attempt=attempt, status_code=status_code, next_retry_at=next_retry.isoformat() if next_retry else None, ) if next_retry: try: send_webhook_task.apply_async(args=[delivery_id], countdown=delay) except Exception as exc: _log.error( "webhook.retry_schedule_failed", delivery_id=delivery_id, error=str(exc), )