1use std::collections::HashMap;
8use std::sync::Arc;
9
10use axum::Router;
11use serde::{Deserialize, Serialize};
12use uuid::Uuid;
13use chrono::{DateTime, Utc};
14
15use backbone_core::http::{ApiResponse, BackboneCrudHandler};
17
18use backbone_auth::middleware::AuthContext;
20#[cfg(feature = "auth")]
21use backbone_auth::AuthMiddleware;
22
23use crate::domain::entity::*;
25use crate::application::service::{ProcessingJobService, ServiceError};
26
27use crate::presentation::dto::{CreateProcessingJobDto, UpdateProcessingJobDto, PatchProcessingJobDto, ProcessingJobResponseDto};
29
30use crate::domain::state_machine::{ProcessingJobState, ProcessingJobStateMachine, ProcessingJobTransition};
31
32#[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 #[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
102pub 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
113pub 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
152pub 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
164pub 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#[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
224pub 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 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 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 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 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
292pub 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 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 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 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 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
355pub 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 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 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 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 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
418pub 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 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 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 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 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
482pub 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 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 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 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 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
546pub 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 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 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 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 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 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
619pub 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 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 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 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 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
682pub 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