app/backend/fluksio/flow/nodes/inject.py

"""Inject: the node that starts something, on a timer or on request.

Node-RED's *inject* is the most placed trigger in a real installation — mostly
as a button someone presses, sometimes on an interval, occasionally once when
everything comes up.
"""

from __future__ import annotations

import asyncio
import logging
import time
from collections.abc import Iterable
from typing import TYPE_CHECKING, Any

from pydantic import BaseModel, ConfigDict, Field

from fluksio.flow.messages import MessageSpec
from fluksio.flow.nodes.base import Node

if TYPE_CHECKING:
    from fastapi import FastAPI

logger = logging.getLogger(__name__)


class InjectNode(Node):
    """Emit a value: on request, every n seconds, on a schedule, or at startup.

    The value is whatever ``payload`` says, or the current time when it says
    nothing — a timestamp is what most schedules actually want. ``payloads``
    overrides that per output port, for a node that starts more than one thing.
    """

    class Params(BaseModel):
        model_config = ConfigDict(extra="allow")

        payload: Any = Field(
            default=None,
            description="What to emit. Empty emits the current time.",
        )
        payloads: dict[str, Any] = Field(
            default_factory=dict,
            description=(
                "What to emit on each output port, keyed by port name. A port not "
                "named here falls back to `payload`."
            ),
        )
        interval: float = Field(
            default=0,
            ge=0,
            description="Emit every this many seconds. 0 means never on its own.",
        )
        cron: str = Field(
            default="",
            description="A five-field cron expression, if it should follow a schedule.",
        )
        at_start: bool = Field(
            default=False,
            description="Emit once when the flow starts.",
        )
        start_delay: float = Field(
            default=1.0,
            ge=0,
            description="How long to wait before the startup emission.",
        )

    __slots__ = ("cfg", "_stop")

    def __init__(
        self,
        requires: MessageSpec | Iterable[MessageSpec] = (),
        provides: MessageSpec | Iterable[MessageSpec] = (),
        params: dict[str, Any] | None = None,
        name: str | None = None,
    ):
        self.cfg = self.Params.model_validate(params or {})
        self._stop: asyncio.Event | None = None
        super().__init__(
            f=self._emit,
            requires=requires,
            provides=provides,
            params=params,
            name=name or "inject",
        )

    def _values(self) -> dict[str, Any]:
        """One value per output port: its own, or the node-wide payload."""
        # Read once, so an emission that falls back carries a single timestamp
        # across every port rather than one per port.
        fallback = time.time() if self.cfg.payload is None else self.cfg.payload
        return {
            spec.port: self.cfg.payloads.get(spec.port, fallback)
            for spec in self.output_ports
        }

    def _emit(self, params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None:
        """Each output carries what its port says to emit."""
        return self._values() or None

    # -------------------------------------------------------------------------
    # Its own schedule
    # -------------------------------------------------------------------------

    async def start(self, app: FastAPI | None = None) -> None:
        if self._stop is not None:
            return
        if not (self.cfg.interval or self.cfg.cron or self.cfg.at_start):
            # A manual inject waits to be pressed.
            return
        self._stop = asyncio.Event()
        self._run_supervised("inject", self._loop)

    async def stop(self, app: FastAPI | None = None) -> None:
        if self._stop is None:
            return
        self._stop.set()
        self._stop = None

    async def _loop(self) -> None:
        stop = self._stop
        if stop is None:
            return

        if self.cfg.at_start:
            # A moment's grace, so subscribers are listening before it fires.
            await self._sleep(stop, self.cfg.start_delay)
            if stop.is_set():
                return
            await self._fire()

        if self.cfg.cron:
            await self._cron_loop(stop)
        elif self.cfg.interval:
            while not stop.is_set():
                await self._sleep(stop, self.cfg.interval)
                if stop.is_set():
                    return
                await self._fire()

    async def _cron_loop(self, stop: asyncio.Event) -> None:
        from datetime import datetime

        from croniter import croniter  # type: ignore[import-untyped]

        if not croniter.is_valid(self.cfg.cron):
            raise ValueError(f"'{self.cfg.cron}' is not a cron expression")

        cron = croniter(self.cfg.cron, datetime.now())
        while not stop.is_set():
            wait = max(0.0, (cron.get_next(datetime) - datetime.now()).total_seconds())
            await self._sleep(stop, wait)
            if stop.is_set():
                return
            await self._fire()

    @staticmethod
    async def _sleep(stop: asyncio.Event, seconds: float) -> None:
        """Wait, but wake immediately if the node is being stopped."""
        try:
            await asyncio.wait_for(stop.wait(), timeout=seconds)
        except TimeoutError:
            pass

    async def _fire(self) -> None:
        outputs = self._values()
        if outputs:
            # inject runs the graph, which is blocking work.
            await asyncio.to_thread(self.inject, outputs)