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