model-platform/backend/src/backend/schedule_runs.py

from __future__ import annotations

import hashlib
from datetime import UTC, datetime
from typing import Any, Literal

from fastapi import APIRouter, Depends, Header, HTTPException, Query, Request, status
from pydantic import Field
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from common.db.models import (
    ScheduleNodeRuns,
    ScheduleRuns,
)
from common.eventing import add_outbox_event, utcnow
from common.ids import new_ulid
from backend.dependencies import (
    RequestContext,
    database_session,
    request_context,
)
from common.schemas import StrictModel
from backend.schedules import (
    graph_rows,
    schedule_row,
    validate_dag,
)


router = APIRouter(tags=["schedule-runs"])
RunStatus = Literal[
    "queued",
    "running",
    "succeeded",
    "failed",
    "cancelled",
    "timed_out",
]


class RunScheduleRequest(StrictModel):
    reason: str = Field(default="manual_run", min_length=1, max_length=255)


def _iso(value: datetime | None) -> str | None:
    if value is None:
        return None
    if value.tzinfo is None:
        value = value.replace(tzinfo=UTC)
    return value.astimezone(UTC).isoformat()


def _normalized_idempotency_key(
    workspace_id: str,
    schedule_id: str,
    value: str,
) -> str:
    normalized = value.strip()
    if len(normalized) < 8:
        raise HTTPException(
            status.HTTP_422_UNPROCESSABLE_ENTITY,
            "Idempotency-Key must contain at least 8 characters",
        )
    digest = hashlib.sha256(
        f"{workspace_id}:{schedule_id}:{normalized}".encode("utf-8")
    ).hexdigest()
    return f"run:v1:{digest}"


def _arguments(value: dict[str, Any] | None) -> list[str]:
    payload = value or {}
    raw = payload.get("_args")
    result = [str(item) for item in raw] if isinstance(raw, list) else []
    for key, item in payload.items():
        if key == "_args":
            continue
        option = f"--{key.replace('_', '-')}"
        if item is True:
            result.append(option)
        elif item is False or item is None:
            continue
        elif isinstance(item, list):
            for list_item in item:
                result.extend((option, str(list_item)))
        elif isinstance(item, (str, int, float)):
            result.extend((option, str(item)))
        else:
            raise HTTPException(
                status.HTTP_422_UNPROCESSABLE_ENTITY,
                f"node argument {key!r} must be a scalar or list",
            )
    return result


def run_summary(item: ScheduleRuns) -> dict[str, Any]:
    return {
        "run_id": item.run_id,
        "schedule_id": item.schedule_id,
        "workspace_id": item.workspace_id,
        "workflow_version": item.workflow_version,
        "trigger_type": item.trigger_type,
        "run_status": item.run_status,
        "state_version": item.state_version,
        "queued_at": _iso(item.queued_at),
        "started_at": _iso(item.started_at),
        "finished_at": _iso(item.finished_at),
        "duration_ms": item.duration_ms,
        "error_code": item.error_code,
        "error_message": item.error_message,
        "logs_object_id": item.logs_object_id,
        "result_object_id": item.result_object_id,
    }


def node_run_payload(item: ScheduleNodeRuns) -> dict[str, Any]:
    return {
        "node_run_id": item.node_run_id,
        "run_id": item.run_id,
        "node_id": item.node_id,
        "versions_id": item.versions_id,
        "attempt_no": item.attempt_no,
        "node_status": item.node_status,
        "state_version": item.state_version,
        "started_at": _iso(item.started_at),
        "finished_at": _iso(item.finished_at),
        "duration_ms": item.duration_ms,
        "exit_code": item.exit_code,
        "message": item.message,
        "logs_object_id": item.logs_object_id,
        "result_object_id": item.result_object_id,
    }


async def run_detail(
    item: ScheduleRuns,
    session: AsyncSession,
) -> dict[str, Any]:
    node_runs = list(
        (
            await session.scalars(
                select(ScheduleNodeRuns)
                .where(ScheduleNodeRuns.run_id == item.run_id)
                .order_by(
                    ScheduleNodeRuns.created_at,
                    ScheduleNodeRuns.attempt_no,
                )
            )
        ).all()
    )
    return {
        **run_summary(item),
        "node_runs": [node_run_payload(node_run) for node_run in node_runs],
    }


async def _visible_run(
    run_id: str,
    context: RequestContext,
    session: AsyncSession,
) -> ScheduleRuns:
    item = await session.scalar(
        select(ScheduleRuns).where(
            ScheduleRuns.run_id == run_id,
            ScheduleRuns.workspace_id == context.workspace.workspace_id,
        )
    )
    if item is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "schedule run not found")
    return item


@router.post(
    "/api/v1/schedules/{schedule_id}/run",
    status_code=status.HTTP_202_ACCEPTED,
)
async def run_schedule_now(
    schedule_id: str,
    request: Request,
    payload: RunScheduleRequest | None = None,
    idempotency_key: str = Header(alias="Idempotency-Key"),
    context: RequestContext = Depends(request_context),
    session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
    reason = payload.reason if payload is not None else "manual_run"
    key = _normalized_idempotency_key(
        context.workspace.workspace_id,
        schedule_id,
        idempotency_key,
    )
    existing = await session.scalar(
        select(ScheduleRuns).where(ScheduleRuns.idempotency_key == key)
    )
    if existing is not None:
        if (
            existing.workspace_id != context.workspace.workspace_id
            or existing.schedule_id != schedule_id
        ):
            raise HTTPException(
                status.HTTP_409_CONFLICT,
                "Idempotency-Key belongs to another schedule run",
            )
        return {
            "request_id": context.request_id,
            "data": await run_detail(existing, session),
            "meta": {"reused": True},
        }

    schedule = await schedule_row(
        schedule_id,
        context,
        session,
        for_update=True,
    )
    node_rows, edges = await graph_rows(schedule_id, session)
    nodes = [row[0] for row in node_rows]
    validation = validate_dag(nodes, edges)
    if not validation["valid"] or not nodes:
        raise HTTPException(
            status.HTTP_409_CONFLICT,
            detail={
                "code": "SCHEDULE_DAG_INVALID",
                "message": "schedule must contain a valid non-empty DAG",
                "errors": validation["errors"],
            },
        )
    if len(nodes) > 100 or len(edges) > 500:
        raise HTTPException(
            status.HTTP_409_CONFLICT,
            "schedule exceeds the v1 execution size limit",
        )

    snapshot = {
        "schedule_name": schedule.schedule_name,
        "workflow_version": schedule.workflow_version,
        "max_concurrency": schedule.max_concurrency,
        "failure_policy": schedule.failure_policy,
        "nodes": [
            {
                "node_id": node.node_id,
                "node_key": node.node_key,
                "versions_id": version.versions_id,
                "script_type": script.script_type,
                "artifact_object_id": version.artifact_object_id,
                "artifact_path": version.artifact_path,
                "timeout_seconds": node.timeout_seconds,
                "retry_count": node.retry_count,
                "retry_interval_sec": node.retry_interval_sec,
                "arguments": _arguments(node.arguments_json),
            }
            for node, version, script in node_rows
        ],
        "edges": [
            {
                "source_node_id": edge.source_node_id,
                "target_node_id": edge.target_node_id,
            }
            for edge in edges
        ],
    }
    now = utcnow()
    run = ScheduleRuns(
        run_id=new_ulid(),
        schedule_id=schedule.schedule_id,
        workspace_id=schedule.workspace_id,
        workflow_version=schedule.workflow_version,
        trigger_type="cron" if reason == "cron" else "manual",
        idempotency_key=key,
        run_status="queued",
        state_version=0,
        schedule_snapshot=snapshot,
        queued_at=now,
        triggered_by=context.user.user_id,
    )
    session.add(run)
    schedule.last_run_at = now
    await add_outbox_event(
        session,
        event_type="schedule.run.requested",
        producer="platform-api",
        trace_id=context.request_id,
        aggregate_type="schedule_run",
        aggregate_id=run.run_id,
        idempotency_key=key,
        payload={
            "workspace_id": run.workspace_id,
            "schedule_id": run.schedule_id,
            "run_id": run.run_id,
            "workflow_version": run.workflow_version,
            "trigger_type": run.trigger_type,
            "triggered_by": run.triggered_by,
            "schedule_snapshot": snapshot,
        },
    )
    await session.flush()
    # Commit before the HTTP push so the executor can read the Outbox row.
    # The executor also polls MySQL, so a failed push does not lose the run.
    await session.commit()
    await request.app.state.schedule_client.dispatch_run(run.run_id)
    await session.refresh(run)
    return {
        "request_id": context.request_id,
        "data": await run_detail(run, session),
        "meta": {"reused": False},
    }


@router.get("/api/v1/schedule-runs")
async def list_schedule_runs(
    schedule_id: str | None = Query(default=None),
    run_status: RunStatus | None = Query(default=None, alias="status"),
    limit: int = Query(default=50, ge=1, le=200),
    context: RequestContext = Depends(request_context),
    session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
    statement = select(ScheduleRuns).where(
        ScheduleRuns.workspace_id == context.workspace.workspace_id,
    )
    if schedule_id:
        statement = statement.where(ScheduleRuns.schedule_id == schedule_id)
    if run_status:
        statement = statement.where(ScheduleRuns.run_status == run_status)
    items = list(
        (
            await session.scalars(
                statement.order_by(ScheduleRuns.queued_at.desc()).limit(limit)
            )
        ).all()
    )
    return {
        "request_id": context.request_id,
        "data": [run_summary(item) for item in items],
        "meta": {"count": len(items)},
    }


@router.get("/api/v1/schedule-runs/{run_id}")
async def get_schedule_run(
    run_id: str,
    context: RequestContext = Depends(request_context),
    session: AsyncSession = Depends(database_session),
) -> dict[str, Any]:
    item = await _visible_run(run_id, context, session)
    return {
        "request_id": context.request_id,
        "data": await run_detail(item, session),
        "meta": {},
    }