"""Workflow run launching, the cron scheduler, and internal event dispatch."""

from __future__ import annotations

import asyncio
import uuid
from collections.abc import Callable
from datetime import UTC, datetime

from croniter import croniter
from sqlalchemy.orm import Session

from legate.config import get_settings
from legate.db.models.run import Run
from legate.db.models.workflow import Workflow
from legate.observability.logging import get_logger
from legate.repositories import runs as runs_repo
from legate.repositories import workflows as workflows_repo
from legate.workflow.executor import WorkflowExecutor

logger = get_logger(__name__)


async def execute_workflow_run_by_id(
    run_id: uuid.UUID, session_factory: Callable[[], Session]
) -> None:
    db = session_factory()
    try:
        run = db.get(Run, run_id)
        workflow = db.get(Workflow, run.workflow_id) if run and run.workflow_id else None
        if run is None or workflow is None:
            logger.warning("workflow run %s missing run/workflow", run_id)
            return
        await WorkflowExecutor(db=db).execute(run, workflow)
    except Exception:  # pragma: no cover - defensive; executor handles its own errors
        logger.exception("workflow run %s crashed", run_id)
    finally:
        db.close()


async def launch_workflow_run(run_id: uuid.UUID, session_factory: Callable[[], Session]) -> None:
    if get_settings().workflow_run_inline:
        await execute_workflow_run_by_id(run_id, session_factory)
    else:
        asyncio.create_task(execute_workflow_run_by_id(run_id, session_factory))


def start_workflow_run(
    db: Session, workflow: Workflow, payload: dict, *, created_by: uuid.UUID | None = None
) -> Run:
    """Create (but do not yet execute) a run for a workflow trigger."""
    run = runs_repo.create(
        db,
        workspace_id=workflow.workspace_id,
        workflow_id=workflow.id,
        input=payload,
        created_by=created_by,
    )
    db.flush()
    return run


# -- schedule trigger ------------------------------------------------------


def _as_utc(value: datetime | None) -> datetime | None:
    if value is None:
        return None
    return value if value.tzinfo else value.replace(tzinfo=UTC)


def is_cron_due(cron: str, last_triggered_at: datetime | None, now: datetime) -> bool:
    try:
        previous = croniter(cron, now).get_prev(datetime)
    except (ValueError, KeyError):
        return False
    previous = _as_utc(previous)
    last = _as_utc(last_triggered_at)
    return last is None or (previous is not None and last < previous)


async def run_due_workflows(session_factory: Callable[[], Session], now: datetime) -> list[str]:
    """Run every scheduled workflow whose cron is due. Returns started run ids."""
    db = session_factory()
    started: list[str] = []
    try:
        for workflow in workflows_repo.list_by_trigger(db, trigger_type="schedule"):
            cron = str(workflow.trigger_config.get("cron", ""))
            if not cron or not is_cron_due(cron, workflow.last_triggered_at, now):
                continue
            run = start_workflow_run(db, workflow, {"scheduled_at": now.isoformat()})
            workflow.last_triggered_at = now
            db.commit()
            await WorkflowExecutor(db=db).execute(run, workflow)
            started.append(str(run.id))
    finally:
        db.close()
    return started


# -- event trigger ---------------------------------------------------------


async def emit_event(
    session_factory: Callable[[], Session],
    *,
    workspace_id: uuid.UUID,
    event: str,
    payload: dict,
) -> list[str]:
    """Start any workflows subscribed to ``event`` in the workspace."""
    db = session_factory()
    started: list[str] = []
    try:
        for workflow in workflows_repo.list_by_trigger(db, trigger_type="event"):
            if workflow.workspace_id != workspace_id:
                continue
            if str(workflow.trigger_config.get("event", "")) != event:
                continue
            run = start_workflow_run(db, workflow, {"event": event, "payload": payload})
            db.commit()
            await WorkflowExecutor(db=db).execute(run, workflow)
            started.append(str(run.id))
    finally:
        db.close()
    return started
