assay_workflow/api/
workflow_tasks.rs

1//! Workflow-task dispatch endpoints (Phase 9).
2//!
3//! A "workflow task" represents "this workflow has new events that need a
4//! worker to run the workflow handler against." It's distinct from an
5//! "activity task" (which runs concrete activity code). The dispatch loop:
6//!
7//! 1. The engine sets `needs_dispatch=true` on a workflow when something
8//!    workflow-visible happens (started, activity completed, timer fired,
9//!    signal arrived).
10//! 2. A worker calls `POST /workflow-tasks/poll` to claim the next
11//!    dispatchable workflow on its queue. Response carries the workflow
12//!    id, type, input, and full event history for replay.
13//! 3. The worker invokes the handler in a coroutine that yields commands
14//!    (ScheduleActivity, CompleteWorkflow, FailWorkflow, etc.) instead of
15//!    making side effects directly.
16//! 4. The worker `POST /workflow-tasks/:id/commands` to submit the batch.
17//!    The engine processes each command transactionally and releases the
18//!    worker's claim on the workflow task.
19
20use std::sync::Arc;
21
22use axum::extract::{Path, State};
23use axum::routing::post;
24use axum::{Json, Router};
25use serde::Deserialize;
26use utoipa::ToSchema;
27
28use crate::api::workflows::AppError;
29use crate::ctx::WorkflowCtx;
30use crate::store::WorkflowStore;
31
32pub fn router<S: WorkflowStore + 'static>() -> Router<Arc<WorkflowCtx<S>>> {
33    Router::new()
34        .route("/workflow-tasks/poll", post(poll_workflow_task))
35        .route("/workflow-tasks/{id}/commands", post(submit_commands))
36}
37
38#[derive(Deserialize, ToSchema)]
39pub struct PollWorkflowTaskRequest {
40    pub queue: String,
41    pub worker_id: String,
42}
43
44/// Claim a dispatchable workflow on the requested queue. Response is `null`
45/// when nothing is available (worker should sleep + retry).
46#[utoipa::path(
47    post, path = "/api/v1/workflow-tasks/poll",
48    tag = "workflow-tasks",
49    request_body = PollWorkflowTaskRequest,
50    responses(
51        (status = 200, description = "Workflow task or null"),
52    ),
53)]
54pub async fn poll_workflow_task<S: WorkflowStore>(
55    State(state): State<Arc<WorkflowCtx<S>>>,
56    Json(req): Json<PollWorkflowTaskRequest>,
57) -> Result<Json<serde_json::Value>, AppError> {
58    match state
59        .claim_workflow_task(&req.queue, &req.worker_id)
60        .await?
61    {
62        Some((wf, history)) => Ok(Json(serde_json::json!({
63            "workflow_id": wf.id,
64            "namespace": wf.namespace,
65            "workflow_type": wf.workflow_type,
66            "task_queue": wf.task_queue,
67            "input": wf.input.as_deref().and_then(|s| serde_json::from_str::<serde_json::Value>(s).ok()),
68            "history": history.iter().map(|e| serde_json::json!({
69                "seq": e.seq,
70                "event_type": e.event_type,
71                "payload": e.payload.as_deref().and_then(|s| serde_json::from_str::<serde_json::Value>(s).ok()),
72                "timestamp": e.timestamp,
73            })).collect::<Vec<_>>(),
74        }))),
75        None => Ok(Json(serde_json::Value::Null)),
76    }
77}
78
79#[derive(Deserialize, ToSchema)]
80pub struct SubmitCommandsRequest {
81    pub worker_id: String,
82    pub commands: Vec<serde_json::Value>,
83}
84
85/// Submit a batch of commands a worker produced from running the workflow
86/// handler. Each command is processed transactionally (ScheduleActivity
87/// inserts a row + appends ActivityScheduled, CompleteWorkflow flips the
88/// status + appends WorkflowCompleted, etc.) and the worker's claim is
89/// released on success.
90#[utoipa::path(
91    post, path = "/api/v1/workflow-tasks/{id}/commands",
92    tag = "workflow-tasks",
93    params(("id" = String, Path, description = "Workflow ID")),
94    request_body = SubmitCommandsRequest,
95    responses((status = 200, description = "Commands processed; lease released")),
96)]
97pub async fn submit_commands<S: WorkflowStore>(
98    State(state): State<Arc<WorkflowCtx<S>>>,
99    Path(workflow_id): Path<String>,
100    Json(req): Json<SubmitCommandsRequest>,
101) -> Result<axum::http::StatusCode, AppError> {
102    state
103        .submit_workflow_commands(&workflow_id, &req.worker_id, &req.commands)
104        .await?;
105    Ok(axum::http::StatusCode::OK)
106}
assay_workflow/api/workflow_tasks.rs

assay_workflow/api/
workflow_tasks.rs
