assay_workflow/api/
activities.rs

1//! Activity scheduling and lookup endpoints.
2
3use std::sync::Arc;
4
5use axum::extract::{Path, State};
6use axum::routing::{get, post};
7use axum::{Json, Router};
8use serde::Deserialize;
9use utoipa::ToSchema;
10
11use crate::api::workflows::AppError;
12use crate::ctx::WorkflowCtx;
13use crate::store::WorkflowStore;
14use crate::types::{ScheduleActivityOpts, WorkflowActivity};
15
16pub fn router<S: WorkflowStore + 'static>() -> Router<Arc<WorkflowCtx<S>>> {
17    Router::new()
18        .route("/workflows/{id}/activities", post(schedule_activity))
19        .route("/activities/{id}", get(get_activity))
20}
21
22#[derive(Deserialize, ToSchema)]
23pub struct ScheduleActivityRequest {
24    /// Activity name (the worker matches this to a registered handler)
25    pub name: String,
26    /// Sequence number relative to the workflow. Used for idempotency:
27    /// scheduling the same `(workflow_id, seq)` twice is a no-op on the
28    /// second call. Workflows assign sequence numbers in execution order.
29    pub seq: i32,
30    /// Task queue to route the activity to (workers poll a specific queue)
31    pub task_queue: String,
32    /// JSON-serialisable input passed to the activity handler
33    pub input: Option<serde_json::Value>,
34    /// Maximum attempts before the activity is marked `FAILED` (default 3)
35    pub max_attempts: Option<i32>,
36    /// Initial retry backoff in seconds (default 1.0)
37    pub initial_interval_secs: Option<f64>,
38    /// Exponential backoff coefficient (default 2.0)
39    pub backoff_coefficient: Option<f64>,
40    /// Total time the activity has to complete before being failed (default 300)
41    pub start_to_close_secs: Option<f64>,
42    /// If set, an activity that hasn't heartbeated within this window is auto-failed
43    pub heartbeat_timeout_secs: Option<f64>,
44}
45
46#[utoipa::path(
47    post, path = "/api/v1/engine/workflow/workflows/{id}/activities",
48    tag = "activities",
49    params(("id" = String, Path, description = "Workflow ID")),
50    request_body = ScheduleActivityRequest,
51    responses(
52        (status = 201, description = "Activity scheduled", body = WorkflowActivity),
53    ),
54)]
55pub async fn schedule_activity<S: WorkflowStore>(
56    State(state): State<Arc<WorkflowCtx<S>>>,
57    Path(workflow_id): Path<String>,
58    Json(req): Json<ScheduleActivityRequest>,
59) -> Result<(axum::http::StatusCode, Json<WorkflowActivity>), AppError> {
60    let input = req.input.map(|v| v.to_string());
61    let opts = ScheduleActivityOpts {
62        max_attempts: req.max_attempts,
63        initial_interval_secs: req.initial_interval_secs,
64        backoff_coefficient: req.backoff_coefficient,
65        start_to_close_secs: req.start_to_close_secs,
66        heartbeat_timeout_secs: req.heartbeat_timeout_secs,
67    };
68    let act = state
69        .schedule_activity(
70            &workflow_id,
71            req.seq,
72            &req.name,
73            input.as_deref(),
74            &req.task_queue,
75            opts,
76        )
77        .await?;
78    Ok((axum::http::StatusCode::CREATED, Json(act)))
79}
80
81#[utoipa::path(
82    get, path = "/api/v1/engine/workflow/activities/{id}",
83    tag = "activities",
84    params(("id" = i64, Path, description = "Activity ID")),
85    responses(
86        (status = 200, description = "Activity record", body = WorkflowActivity),
87        (status = 404, description = "Not found"),
88    ),
89)]
90pub async fn get_activity<S: WorkflowStore>(
91    State(state): State<Arc<WorkflowCtx<S>>>,
92    Path(id): Path<i64>,
93) -> Result<Json<WorkflowActivity>, AppError> {
94    match state.get_activity(id).await? {
95        Some(a) => Ok(Json(a)),
96        None => Err(AppError::NotFound(format!("activity {id} not found"))),
97    }
98}
assay_workflow/api/activities.rs

assay_workflow/api/
activities.rs
