schedule.py

"""The cron-triggered schedule aggregate.

A :class:`Schedule` attaches a cron expression to a workflow so runs start on
their own, without an operator submitting a job. The domain layer stays free
of framework imports; ``croniter`` is the one exception, kept behind
:func:`_next_occurrence` so the rest of the module (and every caller) only
ever sees plain ``datetime`` values.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import UTC, datetime
from types import MappingProxyType
from typing import TYPE_CHECKING
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError

from croniter import croniter

if TYPE_CHECKING:
    from collections.abc import Mapping

    from flowforge.domain.ids import ScheduleId, WorkflowId


def _next_occurrence(cron: str, timezone: str, after: datetime) -> datetime:
    """Return the next firing of ``cron`` strictly after ``after``, in UTC.

    ``after`` is converted into ``timezone`` before being handed to
    ``croniter`` so the cron fields (``0 9 * * *`` meaning nine o'clock local
    time) are evaluated in the schedule's own zone, including across a DST
    transition; the result is converted back to UTC for storage and
    comparison.
    """
    zone = ZoneInfo(timezone)
    local_after = after.astimezone(zone)
    next_local: datetime = croniter(cron, local_after).get_next(datetime)
    return next_local.astimezone(UTC)


@dataclass(frozen=True, slots=True)
class Schedule:
    """A cron trigger bound to a workflow.

    Attributes:
        id: Stable schedule identifier.
        workflow_id: Workflow this schedule submits runs for.
        cron: A standard five-field cron expression.
        timezone: IANA zone the cron fields are evaluated in.
        enabled: Whether the dispatcher should ever claim this schedule.
        payload: Extra inputs merged into every job this schedule submits.
        next_run_at: Next time this schedule is due, or ``None`` until set.
        last_run_at: Time of the most recent fire, or ``None`` before the
            first one.
        last_error: Message from the most recent failed submission, or
            ``None`` if the last fire (if any) submitted cleanly. Set by
            :class:`~flowforge.application.schedules.ScheduleDispatcher` and
            cleared on the next fire that submits successfully.
        last_failure_at: When ``last_error`` was recorded. Unlike
            ``last_error``, this is not cleared by a later success: it is the
            schedule's history of ever having failed, not its current state.
        created_at: When the schedule was created.
        updated_at: When the schedule was last modified.
    """

    id: ScheduleId
    workflow_id: WorkflowId
    cron: str
    timezone: str = "UTC"
    enabled: bool = True
    payload: Mapping[str, object] = field(default_factory=dict)
    next_run_at: datetime | None = None
    last_run_at: datetime | None = None
    last_error: str | None = None
    last_failure_at: datetime | None = None
    created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
    updated_at: datetime = field(default_factory=lambda: datetime.now(UTC))

    def __post_init__(self) -> None:
        cron = self.cron.strip()
        if not croniter.is_valid(cron):
            raise ValueError(f"invalid cron expression: {self.cron!r}")
        object.__setattr__(self, "cron", cron)
        try:
            ZoneInfo(self.timezone)
        except (ZoneInfoNotFoundError, ValueError) as exc:
            raise ValueError(f"invalid timezone: {self.timezone!r}") from exc
        for attr_name in (
            "next_run_at",
            "last_run_at",
            "last_failure_at",
            "created_at",
            "updated_at",
        ):
            value = getattr(self, attr_name)
            if value is not None and value.tzinfo is None:
                raise ValueError(f"{attr_name} must be timezone-aware")
        object.__setattr__(self, "payload", MappingProxyType(dict(self.payload)))

    def next_fire(self, after: datetime) -> datetime:
        """Return the next occurrence of this schedule strictly after ``after``.

        ``after`` must be timezone-aware. The result is always in UTC.
        """
        if after.tzinfo is None:
            raise ValueError("after must be timezone-aware")
        return _next_occurrence(self.cron, self.timezone, after)