Skip to main content

backbone_bucket/presentation/http/
processing_job_handler.rs

1//! ProcessingJob REST handlers
2//!
3//! Generated by metaphor-schema. Do not edit manually.
4//!
5//! Uses Axum and backbone-core's BackboneCrudHandler for all 12 CRUD endpoints.
6
7use std::collections::HashMap;
8use std::sync::Arc;
9
10use axum::Router;
11use serde::{Deserialize, Serialize};
12use uuid::Uuid;
13use chrono::{DateTime, Utc};
14
15// Backbone framework imports
16use backbone_core::http::{ApiResponse, BackboneCrudHandler};
17
18// Auth integration (optional)
19#[cfg(feature = "auth")]
20use backbone_auth::middleware::AuthContext;
21#[cfg(feature = "auth")]
22use backbone_auth::AuthMiddleware;
23
24// Domain imports
25use crate::domain::entity::*;
26use crate::application::service::{ProcessingJobService, ServiceError};
27
28// DTO imports
29use crate::presentation::dto::{CreateProcessingJobDto, UpdateProcessingJobDto, PatchProcessingJobDto, ProcessingJobResponseDto};
30
31use crate::domain::state_machine::{ProcessingJobState, ProcessingJobStateMachine, ProcessingJobTransition};
32
33/// Application error type
34#[derive(Debug, thiserror::Error)]
35pub enum ProcessingJobError {
36    #[error("Not found: {0}")]
37    NotFound(String),
38    #[error("Validation error: {0}")]
39    Validation(String),
40    #[error("Database error: {0}")]
41    Database(String),
42    #[error("Internal error: {0}")]
43    Internal(String),
44    // Domain-specific errors from hook rules
45    #[error("File must exist: {0}")]
46    FileNotFound(String),
47    #[error("File must be accessible: {0}")]
48    FileNotAccessible(String),
49    #[error("Job type must be supported: {0}")]
50    InvalidJobType(String),
51    #[error("Job type must match file type: {0}")]
52    JobTypeMismatch(String),
53    #[error("A similar job is already pending: {0}")]
54    DuplicateJob(String),
55    #[error("Priority must be 0-10: {0}")]
56    InvalidPriority(String),
57    #[error("Cannot modify completed jobs: {0}")]
58    JobCompleted(String),
59}
60
61impl From<ServiceError> for ProcessingJobError {
62    fn from(err: ServiceError) -> Self {
63        match err {
64            ServiceError::NotFound => Self::NotFound(err.to_string()),
65            ServiceError::Validation(ref msg) => Self::Validation(msg.clone()),
66            ServiceError::AlreadyExists(ref msg) => Self::Validation(msg.clone()),
67            ServiceError::Repository(ref e) => Self::Database(e.to_string()),
68            ServiceError::Internal(ref msg) => Self::Internal(msg.clone()),
69            ServiceError::Violations(_) => Self::Validation(err.to_string()),
70        }
71    }
72}
73
74impl axum::response::IntoResponse for ProcessingJobError {
75    fn into_response(self) -> axum::response::Response {
76        use axum::http::StatusCode;
77        use axum::Json;
78
79        let (status, code) = match &self {
80            Self::NotFound(_) => (StatusCode::NOT_FOUND, "PROCESSINGJOB_NOT_FOUND"),
81            Self::Validation(_) => (StatusCode::BAD_REQUEST, "PROCESSINGJOB_VALIDATION_ERROR"),
82            Self::Database(_) => (StatusCode::INTERNAL_SERVER_ERROR, "PROCESSINGJOB_DATABASE_ERROR"),
83            Self::Internal(_) => (StatusCode::INTERNAL_SERVER_ERROR, "PROCESSINGJOB_INTERNAL_ERROR"),
84            Self::FileNotFound(_) => (StatusCode::UNPROCESSABLE_ENTITY, "PROCESSINGJOB_FILE_NOT_FOUND"),
85            Self::FileNotAccessible(_) => (StatusCode::UNPROCESSABLE_ENTITY, "PROCESSINGJOB_FILE_NOT_ACCESSIBLE"),
86            Self::InvalidJobType(_) => (StatusCode::UNPROCESSABLE_ENTITY, "PROCESSINGJOB_INVALID_JOB_TYPE"),
87            Self::JobTypeMismatch(_) => (StatusCode::UNPROCESSABLE_ENTITY, "PROCESSINGJOB_JOB_TYPE_MISMATCH"),
88            Self::DuplicateJob(_) => (StatusCode::UNPROCESSABLE_ENTITY, "PROCESSINGJOB_DUPLICATE_JOB"),
89            Self::InvalidPriority(_) => (StatusCode::UNPROCESSABLE_ENTITY, "PROCESSINGJOB_INVALID_PRIORITY"),
90            Self::JobCompleted(_) => (StatusCode::UNPROCESSABLE_ENTITY, "PROCESSINGJOB_JOB_COMPLETED"),
91        };
92
93        let body = serde_json::json!({
94            "success": false,
95            "error": code,
96            "message": self.to_string(),
97        });
98
99        (status, Json(body)).into_response()
100    }
101}
102
103/// Domain-specific error codes for ProcessingJob
104pub mod processing_job_errors {
105    pub const FILE_NOT_FOUND: &str = "PROCESSINGJOB_FILE_NOT_FOUND";
106    pub const FILE_NOT_ACCESSIBLE: &str = "PROCESSINGJOB_FILE_NOT_ACCESSIBLE";
107    pub const INVALID_JOB_TYPE: &str = "PROCESSINGJOB_INVALID_JOB_TYPE";
108    pub const JOB_TYPE_MISMATCH: &str = "PROCESSINGJOB_JOB_TYPE_MISMATCH";
109    pub const DUPLICATE_JOB: &str = "PROCESSINGJOB_DUPLICATE_JOB";
110    pub const INVALID_PRIORITY: &str = "PROCESSINGJOB_INVALID_PRIORITY";
111    pub const JOB_COMPLETED: &str = "PROCESSINGJOB_JOB_COMPLETED";
112}
113
114// =============================================================================
115// Route Configuration
116// =============================================================================
117
118/// Create Axum router with all 16 Backbone endpoints for ProcessingJob.
119///
120/// # Routes
121///
122/// | Method | Path | Description |
123/// |--------|------|-------------|
124/// | GET | /processing_jobs | List with pagination |
125/// | POST | /processing_jobs | Create new |
126/// | GET | /processing_jobs/:id | Get by ID |
127/// | PUT | /processing_jobs/:id | Full update |
128/// | PATCH | /processing_jobs/:id | Partial update |
129/// | DELETE | /processing_jobs/:id | Soft delete |
130/// | POST | /processing_jobs/bulk | Bulk create |
131/// | POST | /processing_jobs/upsert | Upsert |
132/// | GET | /processing_jobs/trash | List deleted |
133/// | POST | /processing_jobs/:id/restore | Restore |
134/// | DELETE | /processing_jobs/empty | Empty trash |
135/// | GET | /processing_jobs/:id/deleted | Get deleted by ID |
136/// | DELETE | /processing_jobs/trash/:id | Permanent delete from trash |
137/// | GET | /processing_jobs/count | Count active entities |
138/// | GET | /processing_jobs/trash/count | Count deleted entities |
139///
140/// # Example
141///
142/// ```text
143/// let service = Arc::new(ProcessingJobService::with_repository(repository));
144/// let router = create_processing_job_routes(service);
145/// ```
146pub fn create_processing_job_routes(service: Arc<ProcessingJobService>) -> Router {
147    BackboneCrudHandler::<ProcessingJobService, ProcessingJob, CreateProcessingJobDto, UpdateProcessingJobDto, ProcessingJobResponseDto>::routes(
148        service,
149        "/processing_jobs",
150    )
151}
152
153/// Create Axum router with only the read (GET) endpoints for ProcessingJob.
154///
155/// Safe for public, unauthenticated exposure (e.g., reference data).
156/// Mutations must be served separately via `create_processing_job_write_routes`,
157/// typically wrapped in an auth middleware layer.
158pub fn create_processing_job_read_routes(service: Arc<ProcessingJobService>) -> Router {
159    BackboneCrudHandler::<ProcessingJobService, ProcessingJob, CreateProcessingJobDto, UpdateProcessingJobDto, ProcessingJobResponseDto>::read_routes(
160        service,
161        "/processing_jobs",
162    )
163}
164
165/// Create Axum router with only the write (mutation) endpoints for ProcessingJob.
166///
167/// These routes must NOT be publicly exposed. Wrap them with an auth
168/// middleware before nesting into the application router.
169///
170/// # This is unguarded generic CRUD, not a validated write path
171///
172/// These are plain create/update/patch/delete mutations over the entity row —
173/// they bypass all business invariants. If the module exposes a validated write
174/// service (e.g. a command router over its domain engine), serve THAT instead
175/// for any mutation that must respect domain rules.
176pub fn create_processing_job_write_routes(service: Arc<ProcessingJobService>) -> Router {
177    BackboneCrudHandler::<ProcessingJobService, ProcessingJob, CreateProcessingJobDto, UpdateProcessingJobDto, ProcessingJobResponseDto>::write_routes(
178        service,
179        "/processing_jobs",
180    )
181}
182
183/// Create authenticated routes with auth middleware.
184///
185/// Requires the `auth` feature flag. The `AuthMiddleware` implementation
186/// is responsible for extracting and validating tokens, then providing
187/// an `AuthContext` via request extensions.
188#[cfg(feature = "auth")]
189pub fn create_protected_processing_job_routes<A: AuthMiddleware + Send + Sync + 'static>(
190    service: Arc<ProcessingJobService>,
191    auth: Arc<A>,
192) -> Router {
193    use axum::middleware;
194    use axum::response::IntoResponse;
195
196    let auth_layer = auth.clone();
197    create_processing_job_routes(service)
198        .layer(middleware::from_fn(move |mut req: axum::extract::Request, next: axum::middleware::Next| {
199            let auth = auth_layer.clone();
200            async move {
201                let token = req.headers()
202                    .get(axum::http::header::AUTHORIZATION)
203                    .and_then(|h| h.to_str().ok())
204                    .and_then(|raw| raw.strip_prefix("Bearer ").or_else(|| raw.strip_prefix("bearer ")))
205                    .unwrap_or("");
206                match auth.authenticate(token).await {
207                    Ok(ctx) => {
208                        req.extensions_mut().insert(ctx);
209                        next.run(req).await
210                    }
211                    Err(_) => {
212                        (axum::http::StatusCode::UNAUTHORIZED,
213                         axum::Json(serde_json::json!({
214                             "success": false,
215                             "error": "unauthorized",
216                             "message": "Authentication required"
217                         }))
218                        ).into_response()
219                    }
220                }
221            }
222        }))
223}
224
225// =============================================================================
226// State Transition Handlers
227// =============================================================================
228
229/// Execute start transition on a ProcessingJob.
230///
231/// POST /processing_jobs/:id/transitions/start
232pub async fn start_transition(
233    axum::extract::State(service): axum::extract::State<Arc<ProcessingJobService>>,
234    axum::extract::Path(id): axum::extract::Path<String>,
235    #[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
236) -> impl axum::response::IntoResponse {
237    use axum::{http::StatusCode, Json};
238
239    // Get current entity
240    let entity = match service.get_by_id(&id).await {
241        Ok(Some(e)) => e,
242        Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
243        Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
244    };
245
246    // Check permission (if auth enabled)
247    #[cfg(feature = "auth")]
248    {
249        let allowed_roles = ProcessingJobTransition::Start.allowed_roles();
250        let has_specific_perm = auth.permissions.iter().any(|p| p == "processing_job:transition:start");
251        let has_update_perm = auth.permissions.iter().any(|p| p == "processing_job:update");
252        let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
253        if !has_specific_perm && !has_role {
254            if !has_update_perm {
255                return (StatusCode::FORBIDDEN, Json(ApiResponse::<ProcessingJobResponseDto>::error("Insufficient permissions for start transition")));
256            }
257            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"]");
258        }
259    }
260
261    // Create state machine from entity's actual status and validate transition
262    let current_state: ProcessingJobState = match entity.status.to_string().parse() {
263        Ok(s) => s,
264        Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ProcessingJobResponseDto>::error("The stored state is not a state of this machine"))),
265    };
266    let sm = ProcessingJobStateMachine::from_state(current_state);
267    if !sm.can_transition(ProcessingJobTransition::Start) {
268        return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ProcessingJobResponseDto>::error("Transition not allowed from current state")));
269    }
270
271    // Guard `processing_service.has_capacity(job_type)` not compiled (`processing_service` is not a field of ProcessingJob); the transition is not checked against it.
272    // Apply transition via partial update
273    let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
274    fields.insert("status".to_string(), serde_json::Value::String("running".to_string()));
275
276    match backbone_core::write_guard::as_transition("start", service.partial_update(&id, fields)).await {
277        Ok(Some(updated)) => {
278            let response: ProcessingJobResponseDto = updated.into();
279            (StatusCode::OK, Json(ApiResponse::ok(response)))
280        }
281        Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
282        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
283    }
284}
285
286/// Execute complete transition on a ProcessingJob.
287///
288/// POST /processing_jobs/:id/transitions/complete
289pub async fn complete_transition(
290    axum::extract::State(service): axum::extract::State<Arc<ProcessingJobService>>,
291    axum::extract::Path(id): axum::extract::Path<String>,
292    #[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
293) -> impl axum::response::IntoResponse {
294    use axum::{http::StatusCode, Json};
295
296    // Get current entity
297    let entity = match service.get_by_id(&id).await {
298        Ok(Some(e)) => e,
299        Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
300        Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
301    };
302
303    // Check permission (if auth enabled)
304    #[cfg(feature = "auth")]
305    {
306        let allowed_roles = ProcessingJobTransition::Complete.allowed_roles();
307        let has_specific_perm = auth.permissions.iter().any(|p| p == "processing_job:transition:complete");
308        let has_update_perm = auth.permissions.iter().any(|p| p == "processing_job:update");
309        let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
310        if !has_specific_perm && !has_role {
311            if !has_update_perm {
312                return (StatusCode::FORBIDDEN, Json(ApiResponse::<ProcessingJobResponseDto>::error("Insufficient permissions for complete transition")));
313            }
314            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"]");
315        }
316    }
317
318    // Create state machine from entity's actual status and validate transition
319    let current_state: ProcessingJobState = match entity.status.to_string().parse() {
320        Ok(s) => s,
321        Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ProcessingJobResponseDto>::error("The stored state is not a state of this machine"))),
322    };
323    let sm = ProcessingJobStateMachine::from_state(current_state);
324    if !sm.can_transition(ProcessingJobTransition::Complete) {
325        return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ProcessingJobResponseDto>::error("Transition not allowed from current state")));
326    }
327
328    // Apply transition via partial update
329    let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
330    fields.insert("status".to_string(), serde_json::Value::String("completed".to_string()));
331
332    match backbone_core::write_guard::as_transition("complete", service.partial_update(&id, fields)).await {
333        Ok(Some(updated)) => {
334            let response: ProcessingJobResponseDto = updated.into();
335            (StatusCode::OK, Json(ApiResponse::ok(response)))
336        }
337        Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
338        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
339    }
340}
341
342/// Execute fail transition on a ProcessingJob.
343///
344/// POST /processing_jobs/:id/transitions/fail
345pub async fn fail_transition(
346    axum::extract::State(service): axum::extract::State<Arc<ProcessingJobService>>,
347    axum::extract::Path(id): axum::extract::Path<String>,
348    #[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
349) -> impl axum::response::IntoResponse {
350    use axum::{http::StatusCode, Json};
351
352    // Get current entity
353    let entity = match service.get_by_id(&id).await {
354        Ok(Some(e)) => e,
355        Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
356        Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
357    };
358
359    // Check permission (if auth enabled)
360    #[cfg(feature = "auth")]
361    {
362        let allowed_roles = ProcessingJobTransition::Fail.allowed_roles();
363        let has_specific_perm = auth.permissions.iter().any(|p| p == "processing_job:transition:fail");
364        let has_update_perm = auth.permissions.iter().any(|p| p == "processing_job:update");
365        let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
366        if !has_specific_perm && !has_role {
367            if !has_update_perm {
368                return (StatusCode::FORBIDDEN, Json(ApiResponse::<ProcessingJobResponseDto>::error("Insufficient permissions for fail transition")));
369            }
370            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"]");
371        }
372    }
373
374    // Create state machine from entity's actual status and validate transition
375    let current_state: ProcessingJobState = match entity.status.to_string().parse() {
376        Ok(s) => s,
377        Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ProcessingJobResponseDto>::error("The stored state is not a state of this machine"))),
378    };
379    let sm = ProcessingJobStateMachine::from_state(current_state);
380    if !sm.can_transition(ProcessingJobTransition::Fail) {
381        return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ProcessingJobResponseDto>::error("Transition not allowed from current state")));
382    }
383
384    // Apply transition via partial update
385    let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
386    fields.insert("status".to_string(), serde_json::Value::String("failed".to_string()));
387
388    match backbone_core::write_guard::as_transition("fail", service.partial_update(&id, fields)).await {
389        Ok(Some(updated)) => {
390            let response: ProcessingJobResponseDto = updated.into();
391            (StatusCode::OK, Json(ApiResponse::ok(response)))
392        }
393        Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
394        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
395    }
396}
397
398/// Execute cancel_pending transition on a ProcessingJob.
399///
400/// POST /processing_jobs/:id/transitions/cancel_pending
401pub async fn cancel_pending_transition(
402    axum::extract::State(service): axum::extract::State<Arc<ProcessingJobService>>,
403    axum::extract::Path(id): axum::extract::Path<String>,
404    #[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
405) -> impl axum::response::IntoResponse {
406    use axum::{http::StatusCode, Json};
407
408    // Get current entity
409    let entity = match service.get_by_id(&id).await {
410        Ok(Some(e)) => e,
411        Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
412        Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
413    };
414
415    // Check permission (if auth enabled)
416    #[cfg(feature = "auth")]
417    {
418        let allowed_roles = ProcessingJobTransition::CancelPending.allowed_roles();
419        let has_specific_perm = auth.permissions.iter().any(|p| p == "processing_job:transition:cancel_pending");
420        let has_update_perm = auth.permissions.iter().any(|p| p == "processing_job:update");
421        let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
422        if !has_specific_perm && !has_role {
423            if !has_update_perm {
424                return (StatusCode::FORBIDDEN, Json(ApiResponse::<ProcessingJobResponseDto>::error("Insufficient permissions for cancel_pending transition")));
425            }
426            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"]");
427        }
428    }
429
430    // Create state machine from entity's actual status and validate transition
431    let current_state: ProcessingJobState = match entity.status.to_string().parse() {
432        Ok(s) => s,
433        Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ProcessingJobResponseDto>::error("The stored state is not a state of this machine"))),
434    };
435    let sm = ProcessingJobStateMachine::from_state(current_state);
436    if !sm.can_transition(ProcessingJobTransition::CancelPending) {
437        return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ProcessingJobResponseDto>::error("Transition not allowed from current state")));
438    }
439
440    // Guard `$actor.id == file.owner_id || $actor.has_role('admin')` not compiled (`.owner_id` reads a relation or a duration); the transition is not checked against it.
441    // Apply transition via partial update
442    let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
443    fields.insert("status".to_string(), serde_json::Value::String("cancelled".to_string()));
444
445    match backbone_core::write_guard::as_transition("cancel_pending", service.partial_update(&id, fields)).await {
446        Ok(Some(updated)) => {
447            let response: ProcessingJobResponseDto = updated.into();
448            (StatusCode::OK, Json(ApiResponse::ok(response)))
449        }
450        Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
451        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
452    }
453}
454
455/// Execute cancel_running transition on a ProcessingJob.
456///
457/// POST /processing_jobs/:id/transitions/cancel_running
458pub async fn cancel_running_transition(
459    axum::extract::State(service): axum::extract::State<Arc<ProcessingJobService>>,
460    axum::extract::Path(id): axum::extract::Path<String>,
461    #[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
462) -> impl axum::response::IntoResponse {
463    use axum::{http::StatusCode, Json};
464
465    // Get current entity
466    let entity = match service.get_by_id(&id).await {
467        Ok(Some(e)) => e,
468        Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
469        Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
470    };
471
472    // Check permission (if auth enabled)
473    #[cfg(feature = "auth")]
474    {
475        let allowed_roles = ProcessingJobTransition::CancelRunning.allowed_roles();
476        let has_specific_perm = auth.permissions.iter().any(|p| p == "processing_job:transition:cancel_running");
477        let has_update_perm = auth.permissions.iter().any(|p| p == "processing_job:update");
478        let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
479        if !has_specific_perm && !has_role {
480            if !has_update_perm {
481                return (StatusCode::FORBIDDEN, Json(ApiResponse::<ProcessingJobResponseDto>::error("Insufficient permissions for cancel_running transition")));
482            }
483            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"]");
484        }
485    }
486
487    // Create state machine from entity's actual status and validate transition
488    let current_state: ProcessingJobState = match entity.status.to_string().parse() {
489        Ok(s) => s,
490        Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ProcessingJobResponseDto>::error("The stored state is not a state of this machine"))),
491    };
492    let sm = ProcessingJobStateMachine::from_state(current_state);
493    if !sm.can_transition(ProcessingJobTransition::CancelRunning) {
494        return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ProcessingJobResponseDto>::error("Transition not allowed from current state")));
495    }
496
497    // Guard `$actor.id == file.owner_id || $actor.has_role('admin')` not compiled (`.owner_id` reads a relation or a duration); the transition is not checked against it.
498    // Apply transition via partial update
499    let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
500    fields.insert("status".to_string(), serde_json::Value::String("cancelled".to_string()));
501
502    match backbone_core::write_guard::as_transition("cancel_running", service.partial_update(&id, fields)).await {
503        Ok(Some(updated)) => {
504            let response: ProcessingJobResponseDto = updated.into();
505            (StatusCode::OK, Json(ApiResponse::ok(response)))
506        }
507        Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
508        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
509    }
510}
511
512/// Execute retry transition on a ProcessingJob.
513///
514/// POST /processing_jobs/:id/transitions/retry
515pub async fn retry_transition(
516    axum::extract::State(service): axum::extract::State<Arc<ProcessingJobService>>,
517    axum::extract::Path(id): axum::extract::Path<String>,
518    #[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
519) -> impl axum::response::IntoResponse {
520    use axum::{http::StatusCode, Json};
521
522    // Get current entity
523    let entity = match service.get_by_id(&id).await {
524        Ok(Some(e)) => e,
525        Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
526        Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
527    };
528
529    // Check permission (if auth enabled)
530    #[cfg(feature = "auth")]
531    {
532        let allowed_roles = ProcessingJobTransition::Retry.allowed_roles();
533        let has_specific_perm = auth.permissions.iter().any(|p| p == "processing_job:transition:retry");
534        let has_update_perm = auth.permissions.iter().any(|p| p == "processing_job:update");
535        let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
536        if !has_specific_perm && !has_role {
537            if !has_update_perm {
538                return (StatusCode::FORBIDDEN, Json(ApiResponse::<ProcessingJobResponseDto>::error("Insufficient permissions for retry transition")));
539            }
540            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"]");
541        }
542    }
543
544    // Create state machine from entity's actual status and validate transition
545    let current_state: ProcessingJobState = match entity.status.to_string().parse() {
546        Ok(s) => s,
547        Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ProcessingJobResponseDto>::error("The stored state is not a state of this machine"))),
548    };
549    let sm = ProcessingJobStateMachine::from_state(current_state);
550    if !sm.can_transition(ProcessingJobTransition::Retry) {
551        return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ProcessingJobResponseDto>::error("Transition not allowed from current state")));
552    }
553
554    {
555        use backbone_core::row_rule as r;
556        let row = &serde_json::to_value(&entity).unwrap_or_default();
557        let state = row.get("status").and_then(|v| v.as_str()).unwrap_or_default();
558        // Guard: retry_count < max_retries
559        if ["failed"].contains(&state) && !r::holds(&r::lt(r::field(row, "retry_count"), r::field(row, "max_retries"))) {
560            return (StatusCode::UNPROCESSABLE_ENTITY, Json(ApiResponse::<ProcessingJobResponseDto>::error("The retry transition's guard does not hold: retry_count < max_retries")));
561        }
562    }
563
564    // Apply transition via partial update
565    let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
566    fields.insert("status".to_string(), serde_json::Value::String("pending".to_string()));
567
568    match backbone_core::write_guard::as_transition("retry", service.partial_update(&id, fields)).await {
569        Ok(Some(updated)) => {
570            let response: ProcessingJobResponseDto = updated.into();
571            (StatusCode::OK, Json(ApiResponse::ok(response)))
572        }
573        Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
574        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
575    }
576}
577
578/// Execute reprocess transition on a ProcessingJob.
579///
580/// POST /processing_jobs/:id/transitions/reprocess
581pub async fn reprocess_transition(
582    axum::extract::State(service): axum::extract::State<Arc<ProcessingJobService>>,
583    axum::extract::Path(id): axum::extract::Path<String>,
584    #[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
585) -> impl axum::response::IntoResponse {
586    use axum::{http::StatusCode, Json};
587
588    // Get current entity
589    let entity = match service.get_by_id(&id).await {
590        Ok(Some(e)) => e,
591        Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
592        Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
593    };
594
595    // Check permission (if auth enabled)
596    #[cfg(feature = "auth")]
597    {
598        let allowed_roles = ProcessingJobTransition::Reprocess.allowed_roles();
599        let has_specific_perm = auth.permissions.iter().any(|p| p == "processing_job:transition:reprocess");
600        let has_update_perm = auth.permissions.iter().any(|p| p == "processing_job:update");
601        let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
602        if !has_specific_perm && !has_role {
603            if !has_update_perm {
604                return (StatusCode::FORBIDDEN, Json(ApiResponse::<ProcessingJobResponseDto>::error("Insufficient permissions for reprocess transition")));
605            }
606            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"]");
607        }
608    }
609
610    // Create state machine from entity's actual status and validate transition
611    let current_state: ProcessingJobState = match entity.status.to_string().parse() {
612        Ok(s) => s,
613        Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<ProcessingJobResponseDto>::error("The stored state is not a state of this machine"))),
614    };
615    let sm = ProcessingJobStateMachine::from_state(current_state);
616    if !sm.can_transition(ProcessingJobTransition::Reprocess) {
617        return (StatusCode::BAD_REQUEST, Json(ApiResponse::<ProcessingJobResponseDto>::error("Transition not allowed from current state")));
618    }
619
620    // Apply transition via partial update
621    let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
622    fields.insert("status".to_string(), serde_json::Value::String("pending".to_string()));
623
624    match backbone_core::write_guard::as_transition("reprocess", service.partial_update(&id, fields)).await {
625        Ok(Some(updated)) => {
626            let response: ProcessingJobResponseDto = updated.into();
627            (StatusCode::OK, Json(ApiResponse::ok(response)))
628        }
629        Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<ProcessingJobResponseDto>::not_found("ProcessingJob", &id))),
630        Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<ProcessingJobResponseDto>::error(e.to_string()))),
631    }
632}
633
634/// Create routes for state transitions.
635pub fn create_processing_job_transition_routes(service: Arc<ProcessingJobService>) -> Router {
636    use axum::routing::post;
637
638    Router::new()
639        .route("/processing_jobs/:id/transitions/start", post(start_transition))
640        .route("/processing_jobs/:id/transitions/complete", post(complete_transition))
641        .route("/processing_jobs/:id/transitions/fail", post(fail_transition))
642        .route("/processing_jobs/:id/transitions/cancel_pending", post(cancel_pending_transition))
643        .route("/processing_jobs/:id/transitions/cancel_running", post(cancel_running_transition))
644        .route("/processing_jobs/:id/transitions/retry", post(retry_transition))
645        .route("/processing_jobs/:id/transitions/reprocess", post(reprocess_transition))
646        .with_state(service)
647}
648