use std::collections::HashMap;
use std::sync::Arc;
use axum::Router;
use serde::{Deserialize, Serialize};
use uuid::Uuid;
use chrono::{DateTime, Utc};
use backbone_core::http::{ApiResponse, BackboneCrudHandler};
#[cfg(feature = "auth")]
use backbone_auth::middleware::AuthContext;
#[cfg(feature = "auth")]
use backbone_auth::AuthMiddleware;
use crate::domain::entity::*;
use crate::application::service::{ProcessingJobService, ServiceError};
use crate::presentation::dto::{CreateProcessingJobDto, UpdateProcessingJobDto, PatchProcessingJobDto, ProcessingJobResponseDto};
use crate::domain::state_machine::{ProcessingJobState, ProcessingJobStateMachine, ProcessingJobTransition};
#[derive(Debug, thiserror::Error)]
pub enum ProcessingJobError {
#[error("Not found: {0}")]
NotFound(String),
#[error("Validation error: {0}")]
Validation(String),
#[error("Database error: {0}")]
Database(String),
#[error("Internal error: {0}")]
Internal(String),
#[error("File must exist: {0}")]
FileNotFound(String),
#[error("File must be accessible: {0}")]
FileNotAccessible(String),
#[error("Job type must be supported: {0}")]
InvalidJobType(String),
#[error("Job type must match file type: {0}")]
JobTypeMismatch(String),
#[error("A similar job is already pending: {0}")]
DuplicateJob(String),
#[error("Priority must be 0-10: {0}")]
InvalidPriority(String),
#[error("Cannot modify completed jobs: {0}")]
JobCompleted(String),
}
impl From<ServiceError> for ProcessingJobError {
fn from(err: ServiceError) -> Self {
match err {
ServiceError::NotFound => Self::NotFound(err.to_string()),
ServiceError::Validation(ref msg) => Self::Validation(msg.clone()),
ServiceError::AlreadyExists(ref msg) => Self::Validation(msg.clone()),
ServiceError::Repository(ref e) => Self::Database(e.to_string()),
ServiceError::Internal(ref msg) => Self::Internal(msg.clone()),
ServiceError::Violations(_) => Self::Validation(err.to_string()),
}
}
}
impl axum::response::IntoResponse for ProcessingJobError {
fn into_response(self) -> axum::response::Response {
use axum::http::StatusCode;
use axum::Json;
let (status, code) = match &self {
Self::NotFound(_) => (StatusCode::NOT_FOUND, "PROCESSINGJOB_NOT_FOUND"),
Self::Validation(_) => (StatusCode::BAD_REQUEST, "PROCESSINGJOB_VALIDATION_ERROR"),
Self::Database(_) => (StatusCode::INTERNAL_SERVER_ERROR, "PROCESSINGJOB_DATABASE_ERROR"),
Self::Internal(_) => (StatusCode::INTERNAL_SERVER_ERROR, "PROCESSINGJOB_INTERNAL_ERROR"),
Self::FileNotFound(_) => (StatusCode::UNPROCESSABLE_ENTITY, "PROCESSINGJOB_FILE_NOT_FOUND"),
Self::FileNotAccessible(_) => (StatusCode::UNPROCESSABLE_ENTITY, "PROCESSINGJOB_FILE_NOT_ACCESSIBLE"),
Self::InvalidJobType(_) => (StatusCode::UNPROCESSABLE_ENTITY, "PROCESSINGJOB_INVALID_JOB_TYPE"),
Self::JobTypeMismatch(_) => (StatusCode::UNPROCESSABLE_ENTITY, "PROCESSINGJOB_JOB_TYPE_MISMATCH"),
Self::DuplicateJob(_) => (StatusCode::UNPROCESSABLE_ENTITY, "PROCESSINGJOB_DUPLICATE_JOB"),
Self::InvalidPriority(_) => (StatusCode::UNPROCESSABLE_ENTITY, "PROCESSINGJOB_INVALID_PRIORITY"),
Self::JobCompleted(_) => (StatusCode::UNPROCESSABLE_ENTITY, "PROCESSINGJOB_JOB_COMPLETED"),
};
let body = serde_json::json!({
"success": false,
"error": code,
"message": self.to_string(),
});
(status, Json(body)).into_response()
}
}
pub mod processing_job_errors {
pub const FILE_NOT_FOUND: &str = "PROCESSINGJOB_FILE_NOT_FOUND";
pub const FILE_NOT_ACCESSIBLE: &str = "PROCESSINGJOB_FILE_NOT_ACCESSIBLE";
pub const INVALID_JOB_TYPE: &str = "PROCESSINGJOB_INVALID_JOB_TYPE";
pub const JOB_TYPE_MISMATCH: &str = "PROCESSINGJOB_JOB_TYPE_MISMATCH";
pub const DUPLICATE_JOB: &str = "PROCESSINGJOB_DUPLICATE_JOB";
pub const INVALID_PRIORITY: &str = "PROCESSINGJOB_INVALID_PRIORITY";
pub const JOB_COMPLETED: &str = "PROCESSINGJOB_JOB_COMPLETED";
}
pub fn create_processing_job_routes(service: Arc<ProcessingJobService>) -> Router {
BackboneCrudHandler::<ProcessingJobService, ProcessingJob, CreateProcessingJobDto, UpdateProcessingJobDto, ProcessingJobResponseDto>::routes(
service,
"/processing_jobs",
)
}
pub fn create_processing_job_read_routes(service: Arc<ProcessingJobService>) -> Router {
BackboneCrudHandler::<ProcessingJobService, ProcessingJob, CreateProcessingJobDto, UpdateProcessingJobDto, ProcessingJobResponseDto>::read_routes(
service,
"/processing_jobs",
)
}
pub fn create_processing_job_write_routes(service: Arc<ProcessingJobService>) -> Router {
BackboneCrudHandler::<ProcessingJobService, ProcessingJob, CreateProcessingJobDto, UpdateProcessingJobDto, ProcessingJobResponseDto>::write_routes(
service,
"/processing_jobs",
)
}
#[cfg(feature = "auth")]
pub fn create_protected_processing_job_routes<A: AuthMiddleware + Send + Sync + 'static>(
service: Arc<ProcessingJobService>,
auth: Arc<A>,
) -> Router {
use axum::middleware;
use axum::response::IntoResponse;
let auth_layer = auth.clone();
create_processing_job_routes(service)
.layer(middleware::from_fn(move |mut req: axum::extract::Request, next: axum::middleware::Next| {
let auth = auth_layer.clone();
async move {
let token = req.headers()
.get(axum::http::header::AUTHORIZATION)
.and_then(|h| h.to_str().ok())
.and_then(|raw| raw.strip_prefix("Bearer ").or_else(|| raw.strip_prefix("bearer ")))
.unwrap_or("");
match auth.authenticate(token).await {
Ok(ctx) => {
req.extensions_mut().insert(ctx);
next.run(req).await
}
Err(_) => {
(axum::http::StatusCode::UNAUTHORIZED,
axum::Json(serde_json::json!({
"success": false,
"error": "unauthorized",
"message": "Authentication required"
}))
).into_response()
}
}
}
}))
}
pub async fn start_transition(
axum::extract::State(service): axum::extract::State<Arc<ProcessingJobService>>,
axum::extract::Path(id): axum::extract::Path<String>,
#[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
) -> impl axum::response::IntoResponse {
use axum::{http::StatusCode, Json};
let entity = match service.get_by_id(&id).await {
Ok(Some(e)) => e,
Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
};
#[cfg(feature = "auth")]
{
let allowed_roles = ProcessingJobTransition::Start.allowed_roles();
let has_specific_perm = auth.permissions.iter().any(|p| p == "processing_job:transition:start");
let has_update_perm = auth.permissions.iter().any(|p| p == "processing_job:update");
let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
if !has_specific_perm && !has_role {
if !has_update_perm {
return (StatusCode::FORBIDDEN, Json(ApiResponse::<ProcessingJobResponseDto>::error("Insufficient permissions for start transition")));
}
tracing::warn!(target: "backbone::write_guard", entity = "ProcessingJob", verb = "start", code = "transition_by_update_permission", "a transition granted only by processing_job:update, running in shadow; it needs processing_job:transition:start or a role of ["system"]");
}
}
let current_state: ProcessingJobState = match entity.status.to_string().parse() {
Ok(s) => s,
Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ProcessingJobResponseDto>::error("The stored state is not a state of this machine"))),
};
let sm = ProcessingJobStateMachine::from_state(current_state);
if !sm.can_transition(ProcessingJobTransition::Start) {
return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ProcessingJobResponseDto>::error("Transition not allowed from current state")));
}
let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
fields.insert("status".to_string(), serde_json::Value::String("running".to_string()));
match backbone_core::write_guard::as_transition("start", service.partial_update(&id, fields)).await {
Ok(Some(updated)) => {
let response: ProcessingJobResponseDto = updated.into();
(StatusCode::OK, Json(ApiResponse::ok(response)))
}
Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
}
}
pub async fn complete_transition(
axum::extract::State(service): axum::extract::State<Arc<ProcessingJobService>>,
axum::extract::Path(id): axum::extract::Path<String>,
#[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
) -> impl axum::response::IntoResponse {
use axum::{http::StatusCode, Json};
let entity = match service.get_by_id(&id).await {
Ok(Some(e)) => e,
Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
};
#[cfg(feature = "auth")]
{
let allowed_roles = ProcessingJobTransition::Complete.allowed_roles();
let has_specific_perm = auth.permissions.iter().any(|p| p == "processing_job:transition:complete");
let has_update_perm = auth.permissions.iter().any(|p| p == "processing_job:update");
let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
if !has_specific_perm && !has_role {
if !has_update_perm {
return (StatusCode::FORBIDDEN, Json(ApiResponse::<ProcessingJobResponseDto>::error("Insufficient permissions for complete transition")));
}
tracing::warn!(target: "backbone::write_guard", entity = "ProcessingJob", verb = "complete", code = "transition_by_update_permission", "a transition granted only by processing_job:update, running in shadow; it needs processing_job:transition:complete or a role of ["system"]");
}
}
let current_state: ProcessingJobState = match entity.status.to_string().parse() {
Ok(s) => s,
Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ProcessingJobResponseDto>::error("The stored state is not a state of this machine"))),
};
let sm = ProcessingJobStateMachine::from_state(current_state);
if !sm.can_transition(ProcessingJobTransition::Complete) {
return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ProcessingJobResponseDto>::error("Transition not allowed from current state")));
}
let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
fields.insert("status".to_string(), serde_json::Value::String("completed".to_string()));
match backbone_core::write_guard::as_transition("complete", service.partial_update(&id, fields)).await {
Ok(Some(updated)) => {
let response: ProcessingJobResponseDto = updated.into();
(StatusCode::OK, Json(ApiResponse::ok(response)))
}
Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
}
}
pub async fn fail_transition(
axum::extract::State(service): axum::extract::State<Arc<ProcessingJobService>>,
axum::extract::Path(id): axum::extract::Path<String>,
#[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
) -> impl axum::response::IntoResponse {
use axum::{http::StatusCode, Json};
let entity = match service.get_by_id(&id).await {
Ok(Some(e)) => e,
Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
};
#[cfg(feature = "auth")]
{
let allowed_roles = ProcessingJobTransition::Fail.allowed_roles();
let has_specific_perm = auth.permissions.iter().any(|p| p == "processing_job:transition:fail");
let has_update_perm = auth.permissions.iter().any(|p| p == "processing_job:update");
let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
if !has_specific_perm && !has_role {
if !has_update_perm {
return (StatusCode::FORBIDDEN, Json(ApiResponse::<ProcessingJobResponseDto>::error("Insufficient permissions for fail transition")));
}
tracing::warn!(target: "backbone::write_guard", entity = "ProcessingJob", verb = "fail", code = "transition_by_update_permission", "a transition granted only by processing_job:update, running in shadow; it needs processing_job:transition:fail or a role of ["system"]");
}
}
let current_state: ProcessingJobState = match entity.status.to_string().parse() {
Ok(s) => s,
Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ProcessingJobResponseDto>::error("The stored state is not a state of this machine"))),
};
let sm = ProcessingJobStateMachine::from_state(current_state);
if !sm.can_transition(ProcessingJobTransition::Fail) {
return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ProcessingJobResponseDto>::error("Transition not allowed from current state")));
}
let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
fields.insert("status".to_string(), serde_json::Value::String("failed".to_string()));
match backbone_core::write_guard::as_transition("fail", service.partial_update(&id, fields)).await {
Ok(Some(updated)) => {
let response: ProcessingJobResponseDto = updated.into();
(StatusCode::OK, Json(ApiResponse::ok(response)))
}
Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
}
}
pub async fn cancel_pending_transition(
axum::extract::State(service): axum::extract::State<Arc<ProcessingJobService>>,
axum::extract::Path(id): axum::extract::Path<String>,
#[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
) -> impl axum::response::IntoResponse {
use axum::{http::StatusCode, Json};
let entity = match service.get_by_id(&id).await {
Ok(Some(e)) => e,
Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
};
#[cfg(feature = "auth")]
{
let allowed_roles = ProcessingJobTransition::CancelPending.allowed_roles();
let has_specific_perm = auth.permissions.iter().any(|p| p == "processing_job:transition:cancel_pending");
let has_update_perm = auth.permissions.iter().any(|p| p == "processing_job:update");
let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
if !has_specific_perm && !has_role {
if !has_update_perm {
return (StatusCode::FORBIDDEN, Json(ApiResponse::<ProcessingJobResponseDto>::error("Insufficient permissions for cancel_pending transition")));
}
tracing::warn!(target: "backbone::write_guard", entity = "ProcessingJob", verb = "cancel_pending", code = "transition_by_update_permission", "a transition granted only by processing_job:update, running in shadow; it needs processing_job:transition:cancel_pending or a role of ["owner", "admin"]");
}
}
let current_state: ProcessingJobState = match entity.status.to_string().parse() {
Ok(s) => s,
Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ProcessingJobResponseDto>::error("The stored state is not a state of this machine"))),
};
let sm = ProcessingJobStateMachine::from_state(current_state);
if !sm.can_transition(ProcessingJobTransition::CancelPending) {
return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ProcessingJobResponseDto>::error("Transition not allowed from current state")));
}
let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
fields.insert("status".to_string(), serde_json::Value::String("cancelled".to_string()));
match backbone_core::write_guard::as_transition("cancel_pending", service.partial_update(&id, fields)).await {
Ok(Some(updated)) => {
let response: ProcessingJobResponseDto = updated.into();
(StatusCode::OK, Json(ApiResponse::ok(response)))
}
Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
}
}
pub async fn cancel_running_transition(
axum::extract::State(service): axum::extract::State<Arc<ProcessingJobService>>,
axum::extract::Path(id): axum::extract::Path<String>,
#[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
) -> impl axum::response::IntoResponse {
use axum::{http::StatusCode, Json};
let entity = match service.get_by_id(&id).await {
Ok(Some(e)) => e,
Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
};
#[cfg(feature = "auth")]
{
let allowed_roles = ProcessingJobTransition::CancelRunning.allowed_roles();
let has_specific_perm = auth.permissions.iter().any(|p| p == "processing_job:transition:cancel_running");
let has_update_perm = auth.permissions.iter().any(|p| p == "processing_job:update");
let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
if !has_specific_perm && !has_role {
if !has_update_perm {
return (StatusCode::FORBIDDEN, Json(ApiResponse::<ProcessingJobResponseDto>::error("Insufficient permissions for cancel_running transition")));
}
tracing::warn!(target: "backbone::write_guard", entity = "ProcessingJob", verb = "cancel_running", code = "transition_by_update_permission", "a transition granted only by processing_job:update, running in shadow; it needs processing_job:transition:cancel_running or a role of ["owner", "admin"]");
}
}
let current_state: ProcessingJobState = match entity.status.to_string().parse() {
Ok(s) => s,
Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ProcessingJobResponseDto>::error("The stored state is not a state of this machine"))),
};
let sm = ProcessingJobStateMachine::from_state(current_state);
if !sm.can_transition(ProcessingJobTransition::CancelRunning) {
return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ProcessingJobResponseDto>::error("Transition not allowed from current state")));
}
let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
fields.insert("status".to_string(), serde_json::Value::String("cancelled".to_string()));
match backbone_core::write_guard::as_transition("cancel_running", service.partial_update(&id, fields)).await {
Ok(Some(updated)) => {
let response: ProcessingJobResponseDto = updated.into();
(StatusCode::OK, Json(ApiResponse::ok(response)))
}
Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
}
}
pub async fn retry_transition(
axum::extract::State(service): axum::extract::State<Arc<ProcessingJobService>>,
axum::extract::Path(id): axum::extract::Path<String>,
#[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
) -> impl axum::response::IntoResponse {
use axum::{http::StatusCode, Json};
let entity = match service.get_by_id(&id).await {
Ok(Some(e)) => e,
Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
};
#[cfg(feature = "auth")]
{
let allowed_roles = ProcessingJobTransition::Retry.allowed_roles();
let has_specific_perm = auth.permissions.iter().any(|p| p == "processing_job:transition:retry");
let has_update_perm = auth.permissions.iter().any(|p| p == "processing_job:update");
let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
if !has_specific_perm && !has_role {
if !has_update_perm {
return (StatusCode::FORBIDDEN, Json(ApiResponse::<ProcessingJobResponseDto>::error("Insufficient permissions for retry transition")));
}
tracing::warn!(target: "backbone::write_guard", entity = "ProcessingJob", verb = "retry", code = "transition_by_update_permission", "a transition granted only by processing_job:update, running in shadow; it needs processing_job:transition:retry or a role of ["system", "owner"]");
}
}
let current_state: ProcessingJobState = match entity.status.to_string().parse() {
Ok(s) => s,
Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ProcessingJobResponseDto>::error("The stored state is not a state of this machine"))),
};
let sm = ProcessingJobStateMachine::from_state(current_state);
if !sm.can_transition(ProcessingJobTransition::Retry) {
return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ProcessingJobResponseDto>::error("Transition not allowed from current state")));
}
{
use backbone_core::row_rule as r;
let row = &serde_json::to_value(&entity).unwrap_or_default();
let state = row.get("status").and_then(|v| v.as_str()).unwrap_or_default();
if ["failed"].contains(&state) && !r::holds(&r::lt(r::field(row, "retry_count"), r::field(row, "max_retries"))) {
return (StatusCode::UNPROCESSABLE_ENTITY, Json(ApiResponse::<ProcessingJobResponseDto>::error("The retry transition's guard does not hold: retry_count < max_retries")));
}
}
let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
fields.insert("status".to_string(), serde_json::Value::String("pending".to_string()));
match backbone_core::write_guard::as_transition("retry", service.partial_update(&id, fields)).await {
Ok(Some(updated)) => {
let response: ProcessingJobResponseDto = updated.into();
(StatusCode::OK, Json(ApiResponse::ok(response)))
}
Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
}
}
pub async fn reprocess_transition(
axum::extract::State(service): axum::extract::State<Arc<ProcessingJobService>>,
axum::extract::Path(id): axum::extract::Path<String>,
#[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
) -> impl axum::response::IntoResponse {
use axum::{http::StatusCode, Json};
let entity = match service.get_by_id(&id).await {
Ok(Some(e)) => e,
Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
};
#[cfg(feature = "auth")]
{
let allowed_roles = ProcessingJobTransition::Reprocess.allowed_roles();
let has_specific_perm = auth.permissions.iter().any(|p| p == "processing_job:transition:reprocess");
let has_update_perm = auth.permissions.iter().any(|p| p == "processing_job:update");
let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
if !has_specific_perm && !has_role {
if !has_update_perm {
return (StatusCode::FORBIDDEN, Json(ApiResponse::<ProcessingJobResponseDto>::error("Insufficient permissions for reprocess transition")));
}
tracing::warn!(target: "backbone::write_guard", entity = "ProcessingJob", verb = "reprocess", code = "transition_by_update_permission", "a transition granted only by processing_job:update, running in shadow; it needs processing_job:transition:reprocess or a role of ["owner", "admin"]");
}
}
let current_state: ProcessingJobState = match entity.status.to_string().parse() {
Ok(s) => s,
Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ProcessingJobResponseDto>::error("The stored state is not a state of this machine"))),
};
let sm = ProcessingJobStateMachine::from_state(current_state);
if !sm.can_transition(ProcessingJobTransition::Reprocess) {
return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ProcessingJobResponseDto>::error("Transition not allowed from current state")));
}
let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
fields.insert("status".to_string(), serde_json::Value::String("pending".to_string()));
match backbone_core::write_guard::as_transition("reprocess", service.partial_update(&id, fields)).await {
Ok(Some(updated)) => {
let response: ProcessingJobResponseDto = updated.into();
(StatusCode::OK, Json(ApiResponse::ok(response)))
}
Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
}
}
pub fn create_processing_job_transition_routes(service: Arc<ProcessingJobService>) -> Router {
use axum::routing::post;
Router::new()
.route("/processing_jobs/:id/transitions/start", post(start_transition))
.route("/processing_jobs/:id/transitions/complete", post(complete_transition))
.route("/processing_jobs/:id/transitions/fail", post(fail_transition))
.route("/processing_jobs/:id/transitions/cancel_pending", post(cancel_pending_transition))
.route("/processing_jobs/:id/transitions/cancel_running", post(cancel_running_transition))
.route("/processing_jobs/:id/transitions/retry", post(retry_transition))
.route("/processing_jobs/:id/transitions/reprocess", post(reprocess_transition))
.with_state(service)
}