Function cancel_task

crates/weft-api/src/extension_api.rs:215–318

Cancel a task (skip downstream execution, remove from TaskRegistry)

(
    State(state): State<Arc<AppState>>,
    Path((token, execution_id)): Path<(String, String)>,
)

/// Cancel a task (skip downstream execution, remove from TaskRegistry)
pub async fn cancel_task(
    State(state): State<Arc<AppState>>,
    Path((token, execution_id)): Path<(String, String)>,
) -> impl IntoResponse {
    let pool = &state.db_pool;

    let user_id = match extension_tokens::validate_token(pool, &token).await {
        Some(uid) => uid,
        None => {
            return (
                StatusCode::UNAUTHORIZED,
                Json(serde_json::json!({ "error": "Invalid token" })),
            ).into_response();
        }
    };

    tracing::info!(
        "Cancelling task {} by user {} via extension",
        execution_id, user_id
    );

    // execution_id here is the full callback ID: "{projectUuid}-{nodeId}-{pulseId}"
    // Parse out the project UUID (first 36 chars), nodeId, and pulseId (last 36 chars).
    let callback_id = &execution_id;
    let client = reqwest::Client::new();

    // Parse callback ID: "{projectUuid}-{nodeId}-{pulseUuid}-{seq}"
    let parsed = (|| {
        let uuid_end = callback_id.splitn(6, '-').take(5).map(|s| s.len()).sum::<usize>() + 4;
        if uuid_end >= callback_id.len() { return None; }
        let project_id = &callback_id[..uuid_end];
        let remainder = &callback_id[uuid_end + 1..]; // "{nodeId}-{pulseUuid}-{seq}"
        // Strip trailing "-{seq}" (sequence number)
        let last_dash = remainder.rfind('-')?;
        let without_seq = &remainder[..last_dash]; // "{nodeId}-{pulseUuid}"
        if without_seq.len() < 37 { return None; } // at least 36 (uuid) + 1 (dash)
        let pulse_id = &without_seq[without_seq.len() - 36..];
        let node_id = &without_seq[..without_seq.len() - 37]; // strip "-{pulseUuid}"
        Some((project_id.to_string(), node_id.to_string(), pulse_id.to_string()))
    })();

    // If the callback ID can't be parsed (old/malformed task), just remove from TaskRegistry
    let Some((project_id, node_id, pulse_id)) = parsed else {
        tracing::info!("Unparseable callback ID '{}', removing stale task from TaskRegistry", callback_id);
        let restate_url = format!(
            "{}/TaskRegistry/global/complete_task",
            state.restate_url
        );
        let _ = client.post(&restate_url).json(&callback_id).send().await;
        return Json(serde_json::json!({ "status": "removed" })).into_response();
    };

    // Call provide_input with skip=true
    let executor_url = format!(
        "{}/ProjectExecutor/{}/provide_input",
        state.executor_url, project_id
    );