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
18#[cfg(feature = "auth")]
20use backbone_auth::middleware::AuthContext;
21#[cfg(feature = "auth")]
22use backbone_auth::AuthMiddleware;
23
24use crate::domain::entity::*;
26use crate::application::service::{UploadSessionService, ServiceError};
27
28use crate::presentation::dto::{CreateUploadSessionDto, UpdateUploadSessionDto, PatchUploadSessionDto, UploadSessionResponseDto};
30
31use crate::domain::state_machine::{UploadSessionState, UploadSessionStateMachine, UploadSessionTransition};
32
33#[derive(Debug, thiserror::Error)]
35pub enum UploadSessionError {
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 #[error("Target bucket must exist: {0}")]
46 BucketNotFound(String),
47 #[error("Bucket must accept uploads: {0}")]
48 BucketNotActive(String),
49 #[error("File size must be positive: {0}")]
50 InvalidFileSize(String),
51 #[error("Chunk size must be at least 5MB: {0}")]
52 ChunkTooSmall(String),
53 #[error("Chunk size must not exceed 5GB: {0}")]
54 ChunkTooLarge(String),
55 #[error("Total chunks must match file size and chunk size: {0}")]
56 InvalidChunkCount(String),
57 #[error("Path must be valid and not contain traversal: {0}")]
58 InvalidPath(String),
59 #[error("Expiration must be in future: {0}")]
60 InvalidExpiration(String),
61 #[error("Expiration must be within 7 days: {0}")]
62 ExpirationTooFar(String),
63 #[error("Session has expired: {0}")]
64 SessionExpired(String),
65}
66
67impl From<ServiceError> for UploadSessionError {
68 fn from(err: ServiceError) -> Self {
69 match err {
70 ServiceError::NotFound => Self::NotFound(err.to_string()),
71 ServiceError::Validation(ref msg) => Self::Validation(msg.clone()),
72 ServiceError::AlreadyExists(ref msg) => Self::Validation(msg.clone()),
73 ServiceError::Repository(ref e) => Self::Database(e.to_string()),
74 ServiceError::Internal(ref msg) => Self::Internal(msg.clone()),
75 ServiceError::Violations(_) => Self::Validation(err.to_string()),
76 }
77 }
78}
79
80impl axum::response::IntoResponse for UploadSessionError {
81 fn into_response(self) -> axum::response::Response {
82 use axum::http::StatusCode;
83 use axum::Json;
84
85 let (status, code) = match &self {
86 Self::NotFound(_) => (StatusCode::NOT_FOUND, "UPLOADSESSION_NOT_FOUND"),
87 Self::Validation(_) => (StatusCode::BAD_REQUEST, "UPLOADSESSION_VALIDATION_ERROR"),
88 Self::Database(_) => (StatusCode::INTERNAL_SERVER_ERROR, "UPLOADSESSION_DATABASE_ERROR"),
89 Self::Internal(_) => (StatusCode::INTERNAL_SERVER_ERROR, "UPLOADSESSION_INTERNAL_ERROR"),
90 Self::BucketNotFound(_) => (StatusCode::UNPROCESSABLE_ENTITY, "UPLOADSESSION_BUCKET_NOT_FOUND"),
91 Self::BucketNotActive(_) => (StatusCode::UNPROCESSABLE_ENTITY, "UPLOADSESSION_BUCKET_NOT_ACTIVE"),
92 Self::InvalidFileSize(_) => (StatusCode::UNPROCESSABLE_ENTITY, "UPLOADSESSION_INVALID_FILE_SIZE"),
93 Self::ChunkTooSmall(_) => (StatusCode::UNPROCESSABLE_ENTITY, "UPLOADSESSION_CHUNK_TOO_SMALL"),
94 Self::ChunkTooLarge(_) => (StatusCode::UNPROCESSABLE_ENTITY, "UPLOADSESSION_CHUNK_TOO_LARGE"),
95 Self::InvalidChunkCount(_) => (StatusCode::UNPROCESSABLE_ENTITY, "UPLOADSESSION_INVALID_CHUNK_COUNT"),
96 Self::InvalidPath(_) => (StatusCode::UNPROCESSABLE_ENTITY, "UPLOADSESSION_INVALID_PATH"),
97 Self::InvalidExpiration(_) => (StatusCode::UNPROCESSABLE_ENTITY, "UPLOADSESSION_INVALID_EXPIRATION"),
98 Self::ExpirationTooFar(_) => (StatusCode::UNPROCESSABLE_ENTITY, "UPLOADSESSION_EXPIRATION_TOO_FAR"),
99 Self::SessionExpired(_) => (StatusCode::UNPROCESSABLE_ENTITY, "UPLOADSESSION_SESSION_EXPIRED"),
100 };
101
102 let body = serde_json::json!({
103 "success": false,
104 "error": code,
105 "message": self.to_string(),
106 });
107
108 (status, Json(body)).into_response()
109 }
110}
111
112pub mod upload_session_errors {
114 pub const BUCKET_NOT_FOUND: &str = "UPLOADSESSION_BUCKET_NOT_FOUND";
115 pub const BUCKET_NOT_ACTIVE: &str = "UPLOADSESSION_BUCKET_NOT_ACTIVE";
116 pub const INVALID_FILE_SIZE: &str = "UPLOADSESSION_INVALID_FILE_SIZE";
117 pub const CHUNK_TOO_SMALL: &str = "UPLOADSESSION_CHUNK_TOO_SMALL";
118 pub const CHUNK_TOO_LARGE: &str = "UPLOADSESSION_CHUNK_TOO_LARGE";
119 pub const INVALID_CHUNK_COUNT: &str = "UPLOADSESSION_INVALID_CHUNK_COUNT";
120 pub const INVALID_PATH: &str = "UPLOADSESSION_INVALID_PATH";
121 pub const INVALID_EXPIRATION: &str = "UPLOADSESSION_INVALID_EXPIRATION";
122 pub const EXPIRATION_TOO_FAR: &str = "UPLOADSESSION_EXPIRATION_TOO_FAR";
123 pub const SESSION_EXPIRED: &str = "UPLOADSESSION_SESSION_EXPIRED";
124}
125
126pub fn create_upload_session_routes(service: Arc<UploadSessionService>) -> Router {
159 BackboneCrudHandler::<UploadSessionService, UploadSession, CreateUploadSessionDto, UpdateUploadSessionDto, UploadSessionResponseDto>::routes(
160 service,
161 "/upload_sessions",
162 )
163}
164
165pub fn create_upload_session_read_routes(service: Arc<UploadSessionService>) -> Router {
171 BackboneCrudHandler::<UploadSessionService, UploadSession, CreateUploadSessionDto, UpdateUploadSessionDto, UploadSessionResponseDto>::read_routes(
172 service,
173 "/upload_sessions",
174 )
175}
176
177pub fn create_upload_session_write_routes(service: Arc<UploadSessionService>) -> Router {
189 BackboneCrudHandler::<UploadSessionService, UploadSession, CreateUploadSessionDto, UpdateUploadSessionDto, UploadSessionResponseDto>::write_routes(
190 service,
191 "/upload_sessions",
192 )
193}
194
195#[cfg(feature = "auth")]
201pub fn create_protected_upload_session_routes<A: AuthMiddleware + Send + Sync + 'static>(
202 service: Arc<UploadSessionService>,
203 auth: Arc<A>,
204) -> Router {
205 use axum::middleware;
206 use axum::response::IntoResponse;
207
208 let auth_layer = auth.clone();
209 create_upload_session_routes(service)
210 .layer(middleware::from_fn(move |mut req: axum::extract::Request, next: axum::middleware::Next| {
211 let auth = auth_layer.clone();
212 async move {
213 let token = req.headers()
214 .get(axum::http::header::AUTHORIZATION)
215 .and_then(|h| h.to_str().ok())
216 .and_then(|raw| raw.strip_prefix("Bearer ").or_else(|| raw.strip_prefix("bearer ")))
217 .unwrap_or("");
218 match auth.authenticate(token).await {
219 Ok(ctx) => {
220 req.extensions_mut().insert(ctx);
221 next.run(req).await
222 }
223 Err(_) => {
224 (axum::http::StatusCode::UNAUTHORIZED,
225 axum::Json(serde_json::json!({
226 "success": false,
227 "error": "unauthorized",
228 "message": "Authentication required"
229 }))
230 ).into_response()
231 }
232 }
233 }
234 }))
235}
236
237pub async fn start_upload_transition(
245 axum::extract::State(service): axum::extract::State<Arc<UploadSessionService>>,
246 axum::extract::Path(id): axum::extract::Path<String>,
247 #[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
248) -> impl axum::response::IntoResponse {
249 use axum::{http::StatusCode, Json};
250
251 let entity = match service.get_by_id(&id).await {
253 Ok(Some(e)) => e,
254 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<UploadSessionResponseDto>::not_found("UploadSession", &id))),
255 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<UploadSessionResponseDto>::error(e.to_string()))),
256 };
257
258 #[cfg(feature = "auth")]
260 {
261 let allowed_roles = UploadSessionTransition::StartUpload.allowed_roles();
262 let has_specific_perm = auth.permissions.iter().any(|p| p == "upload_session:transition:start_upload");
263 let has_update_perm = auth.permissions.iter().any(|p| p == "upload_session:update");
264 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
265 if !has_specific_perm && !has_role {
266 if !has_update_perm {
267 return (StatusCode::FORBIDDEN, Json(ApiResponse::<UploadSessionResponseDto>::error("Insufficient permissions for start_upload transition")));
268 }
269 tracing::warn!(target: "backbone::write_guard", entity = "UploadSession", verb = "start_upload", code = "transition_by_update_permission", "a transition granted only by upload_session:update, running in shadow; it needs upload_session:transition:start_upload or a role of ["system", "user"]");
270 }
271 }
272
273 let current_state: UploadSessionState = match entity.status.to_string().parse() {
275 Ok(s) => s,
276 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<UploadSessionResponseDto>::error("The stored state is not a state of this machine"))),
277 };
278 let sm = UploadSessionStateMachine::from_state(current_state);
279 if !sm.can_transition(UploadSessionTransition::StartUpload) {
280 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<UploadSessionResponseDto>::error("Transition not allowed from current state")));
281 }
282
283 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
286 fields.insert("status".to_string(), serde_json::Value::String("uploading".to_string()));
287
288 match backbone_core::write_guard::as_transition("start_upload", service.partial_update(&id, fields)).await {
289 Ok(Some(updated)) => {
290 let response: UploadSessionResponseDto = updated.into();
291 (StatusCode::OK, Json(ApiResponse::ok(response)))
292 }
293 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<UploadSessionResponseDto>::not_found("UploadSession", &id))),
294 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<UploadSessionResponseDto>::error(e.to_string()))),
295 }
296}
297
298pub async fn add_part_transition(
302 axum::extract::State(service): axum::extract::State<Arc<UploadSessionService>>,
303 axum::extract::Path(id): axum::extract::Path<String>,
304 #[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
305) -> impl axum::response::IntoResponse {
306 use axum::{http::StatusCode, Json};
307
308 let entity = match service.get_by_id(&id).await {
310 Ok(Some(e)) => e,
311 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<UploadSessionResponseDto>::not_found("UploadSession", &id))),
312 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<UploadSessionResponseDto>::error(e.to_string()))),
313 };
314
315 #[cfg(feature = "auth")]
317 {
318 let allowed_roles = UploadSessionTransition::AddPart.allowed_roles();
319 let has_specific_perm = auth.permissions.iter().any(|p| p == "upload_session:transition:add_part");
320 let has_update_perm = auth.permissions.iter().any(|p| p == "upload_session:update");
321 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
322 if !has_specific_perm && !has_role {
323 if !has_update_perm {
324 return (StatusCode::FORBIDDEN, Json(ApiResponse::<UploadSessionResponseDto>::error("Insufficient permissions for add_part transition")));
325 }
326 tracing::warn!(target: "backbone::write_guard", entity = "UploadSession", verb = "add_part", code = "transition_by_update_permission", "a transition granted only by upload_session:update, running in shadow; it needs upload_session:transition:add_part or a role of ["system", "user"]");
327 }
328 }
329
330 let current_state: UploadSessionState = match entity.status.to_string().parse() {
332 Ok(s) => s,
333 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<UploadSessionResponseDto>::error("The stored state is not a state of this machine"))),
334 };
335 let sm = UploadSessionStateMachine::from_state(current_state);
336 if !sm.can_transition(UploadSessionTransition::AddPart) {
337 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<UploadSessionResponseDto>::error("Transition not allowed from current state")));
338 }
339
340 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
343 fields.insert("status".to_string(), serde_json::Value::String("uploading".to_string()));
344
345 match backbone_core::write_guard::as_transition("add_part", service.partial_update(&id, fields)).await {
346 Ok(Some(updated)) => {
347 let response: UploadSessionResponseDto = updated.into();
348 (StatusCode::OK, Json(ApiResponse::ok(response)))
349 }
350 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<UploadSessionResponseDto>::not_found("UploadSession", &id))),
351 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<UploadSessionResponseDto>::error(e.to_string()))),
352 }
353}
354
355pub async fn complete_transition(
359 axum::extract::State(service): axum::extract::State<Arc<UploadSessionService>>,
360 axum::extract::Path(id): axum::extract::Path<String>,
361 #[cfg(feature = "auth")] axum::Extension(auth): 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::<UploadSessionResponseDto>::not_found("UploadSession", &id))),
369 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<UploadSessionResponseDto>::error(e.to_string()))),
370 };
371
372 #[cfg(feature = "auth")]
374 {
375 let allowed_roles = UploadSessionTransition::Complete.allowed_roles();
376 let has_specific_perm = auth.permissions.iter().any(|p| p == "upload_session:transition:complete");
377 let has_update_perm = auth.permissions.iter().any(|p| p == "upload_session:update");
378 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
379 if !has_specific_perm && !has_role {
380 if !has_update_perm {
381 return (StatusCode::FORBIDDEN, Json(ApiResponse::<UploadSessionResponseDto>::error("Insufficient permissions for complete transition")));
382 }
383 tracing::warn!(target: "backbone::write_guard", entity = "UploadSession", verb = "complete", code = "transition_by_update_permission", "a transition granted only by upload_session:update, running in shadow; it needs upload_session:transition:complete or a role of ["system", "user"]");
384 }
385 }
386
387 let current_state: UploadSessionState = match entity.status.to_string().parse() {
389 Ok(s) => s,
390 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<UploadSessionResponseDto>::error("The stored state is not a state of this machine"))),
391 };
392 let sm = UploadSessionStateMachine::from_state(current_state);
393 if !sm.can_transition(UploadSessionTransition::Complete) {
394 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<UploadSessionResponseDto>::error("Transition not allowed from current state")));
395 }
396
397 {
398 use backbone_core::row_rule as r;
399 let row = &serde_json::to_value(&entity).unwrap_or_default();
400 let state = row.get("status").and_then(|v| v.as_str()).unwrap_or_default();
401 if ["uploading"].contains(&state) && !r::holds(&r::eq(r::field(row, "uploaded_chunks"), r::field(row, "total_chunks"))) {
403 return (StatusCode::UNPROCESSABLE_ENTITY, Json(ApiResponse::<UploadSessionResponseDto>::error("The complete transition's guard does not hold: uploaded_chunks == total_chunks")));
404 }
405 }
406
407 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
409 fields.insert("status".to_string(), serde_json::Value::String("completing".to_string()));
410
411 match backbone_core::write_guard::as_transition("complete", service.partial_update(&id, fields)).await {
412 Ok(Some(updated)) => {
413 let response: UploadSessionResponseDto = updated.into();
414 (StatusCode::OK, Json(ApiResponse::ok(response)))
415 }
416 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<UploadSessionResponseDto>::not_found("UploadSession", &id))),
417 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<UploadSessionResponseDto>::error(e.to_string()))),
418 }
419}
420
421pub async fn finish_transition(
425 axum::extract::State(service): axum::extract::State<Arc<UploadSessionService>>,
426 axum::extract::Path(id): axum::extract::Path<String>,
427 #[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
428) -> impl axum::response::IntoResponse {
429 use axum::{http::StatusCode, Json};
430
431 let entity = match service.get_by_id(&id).await {
433 Ok(Some(e)) => e,
434 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<UploadSessionResponseDto>::not_found("UploadSession", &id))),
435 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<UploadSessionResponseDto>::error(e.to_string()))),
436 };
437
438 #[cfg(feature = "auth")]
440 {
441 let allowed_roles = UploadSessionTransition::Finish.allowed_roles();
442 let has_specific_perm = auth.permissions.iter().any(|p| p == "upload_session:transition:finish");
443 let has_update_perm = auth.permissions.iter().any(|p| p == "upload_session:update");
444 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
445 if !has_specific_perm && !has_role {
446 if !has_update_perm {
447 return (StatusCode::FORBIDDEN, Json(ApiResponse::<UploadSessionResponseDto>::error("Insufficient permissions for finish transition")));
448 }
449 tracing::warn!(target: "backbone::write_guard", entity = "UploadSession", verb = "finish", code = "transition_by_update_permission", "a transition granted only by upload_session:update, running in shadow; it needs upload_session:transition:finish or a role of ["system"]");
450 }
451 }
452
453 let current_state: UploadSessionState = match entity.status.to_string().parse() {
455 Ok(s) => s,
456 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<UploadSessionResponseDto>::error("The stored state is not a state of this machine"))),
457 };
458 let sm = UploadSessionStateMachine::from_state(current_state);
459 if !sm.can_transition(UploadSessionTransition::Finish) {
460 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<UploadSessionResponseDto>::error("Transition not allowed from current state")));
461 }
462
463 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
465 fields.insert("status".to_string(), serde_json::Value::String("completed".to_string()));
466
467 match backbone_core::write_guard::as_transition("finish", service.partial_update(&id, fields)).await {
468 Ok(Some(updated)) => {
469 let response: UploadSessionResponseDto = updated.into();
470 (StatusCode::OK, Json(ApiResponse::ok(response)))
471 }
472 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<UploadSessionResponseDto>::not_found("UploadSession", &id))),
473 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<UploadSessionResponseDto>::error(e.to_string()))),
474 }
475}
476
477pub async fn fail_transition(
481 axum::extract::State(service): axum::extract::State<Arc<UploadSessionService>>,
482 axum::extract::Path(id): axum::extract::Path<String>,
483 #[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
484) -> impl axum::response::IntoResponse {
485 use axum::{http::StatusCode, Json};
486
487 let entity = match service.get_by_id(&id).await {
489 Ok(Some(e)) => e,
490 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<UploadSessionResponseDto>::not_found("UploadSession", &id))),
491 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<UploadSessionResponseDto>::error(e.to_string()))),
492 };
493
494 #[cfg(feature = "auth")]
496 {
497 let allowed_roles = UploadSessionTransition::Fail.allowed_roles();
498 let has_specific_perm = auth.permissions.iter().any(|p| p == "upload_session:transition:fail");
499 let has_update_perm = auth.permissions.iter().any(|p| p == "upload_session:update");
500 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
501 if !has_specific_perm && !has_role {
502 if !has_update_perm {
503 return (StatusCode::FORBIDDEN, Json(ApiResponse::<UploadSessionResponseDto>::error("Insufficient permissions for fail transition")));
504 }
505 tracing::warn!(target: "backbone::write_guard", entity = "UploadSession", verb = "fail", code = "transition_by_update_permission", "a transition granted only by upload_session:update, running in shadow; it needs upload_session:transition:fail or a role of ["system"]");
506 }
507 }
508
509 let current_state: UploadSessionState = match entity.status.to_string().parse() {
511 Ok(s) => s,
512 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<UploadSessionResponseDto>::error("The stored state is not a state of this machine"))),
513 };
514 let sm = UploadSessionStateMachine::from_state(current_state);
515 if !sm.can_transition(UploadSessionTransition::Fail) {
516 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<UploadSessionResponseDto>::error("Transition not allowed from current state")));
517 }
518
519 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
521 fields.insert("status".to_string(), serde_json::Value::String("failed".to_string()));
522
523 match backbone_core::write_guard::as_transition("fail", service.partial_update(&id, fields)).await {
524 Ok(Some(updated)) => {
525 let response: UploadSessionResponseDto = updated.into();
526 (StatusCode::OK, Json(ApiResponse::ok(response)))
527 }
528 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<UploadSessionResponseDto>::not_found("UploadSession", &id))),
529 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<UploadSessionResponseDto>::error(e.to_string()))),
530 }
531}
532
533pub async fn abort_transition(
537 axum::extract::State(service): axum::extract::State<Arc<UploadSessionService>>,
538 axum::extract::Path(id): axum::extract::Path<String>,
539 #[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
540) -> impl axum::response::IntoResponse {
541 use axum::{http::StatusCode, Json};
542
543 let entity = match service.get_by_id(&id).await {
545 Ok(Some(e)) => e,
546 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<UploadSessionResponseDto>::not_found("UploadSession", &id))),
547 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<UploadSessionResponseDto>::error(e.to_string()))),
548 };
549
550 #[cfg(feature = "auth")]
552 {
553 let allowed_roles = UploadSessionTransition::Abort.allowed_roles();
554 let has_specific_perm = auth.permissions.iter().any(|p| p == "upload_session:transition:abort");
555 let has_update_perm = auth.permissions.iter().any(|p| p == "upload_session:update");
556 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
557 if !has_specific_perm && !has_role {
558 if !has_update_perm {
559 return (StatusCode::FORBIDDEN, Json(ApiResponse::<UploadSessionResponseDto>::error("Insufficient permissions for abort transition")));
560 }
561 tracing::warn!(target: "backbone::write_guard", entity = "UploadSession", verb = "abort", code = "transition_by_update_permission", "a transition granted only by upload_session:update, running in shadow; it needs upload_session:transition:abort or a role of ["user", "admin"]");
562 }
563 }
564
565 let current_state: UploadSessionState = match entity.status.to_string().parse() {
567 Ok(s) => s,
568 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<UploadSessionResponseDto>::error("The stored state is not a state of this machine"))),
569 };
570 let sm = UploadSessionStateMachine::from_state(current_state);
571 if !sm.can_transition(UploadSessionTransition::Abort) {
572 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<UploadSessionResponseDto>::error("Transition not allowed from current state")));
573 }
574
575 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
578 fields.insert("status".to_string(), serde_json::Value::String("failed".to_string()));
579
580 match backbone_core::write_guard::as_transition("abort", service.partial_update(&id, fields)).await {
581 Ok(Some(updated)) => {
582 let response: UploadSessionResponseDto = updated.into();
583 (StatusCode::OK, Json(ApiResponse::ok(response)))
584 }
585 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<UploadSessionResponseDto>::not_found("UploadSession", &id))),
586 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<UploadSessionResponseDto>::error(e.to_string()))),
587 }
588}
589
590pub async fn expire_transition(
594 axum::extract::State(service): axum::extract::State<Arc<UploadSessionService>>,
595 axum::extract::Path(id): axum::extract::Path<String>,
596 #[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
597) -> impl axum::response::IntoResponse {
598 use axum::{http::StatusCode, Json};
599
600 let entity = match service.get_by_id(&id).await {
602 Ok(Some(e)) => e,
603 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<UploadSessionResponseDto>::not_found("UploadSession", &id))),
604 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<UploadSessionResponseDto>::error(e.to_string()))),
605 };
606
607 #[cfg(feature = "auth")]
609 {
610 let allowed_roles = UploadSessionTransition::Expire.allowed_roles();
611 let has_specific_perm = auth.permissions.iter().any(|p| p == "upload_session:transition:expire");
612 let has_update_perm = auth.permissions.iter().any(|p| p == "upload_session:update");
613 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
614 if !has_specific_perm && !has_role {
615 if !has_update_perm {
616 return (StatusCode::FORBIDDEN, Json(ApiResponse::<UploadSessionResponseDto>::error("Insufficient permissions for expire transition")));
617 }
618 tracing::warn!(target: "backbone::write_guard", entity = "UploadSession", verb = "expire", code = "transition_by_update_permission", "a transition granted only by upload_session:update, running in shadow; it needs upload_session:transition:expire or a role of ["system"]");
619 }
620 }
621
622 let current_state: UploadSessionState = match entity.status.to_string().parse() {
624 Ok(s) => s,
625 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<UploadSessionResponseDto>::error("The stored state is not a state of this machine"))),
626 };
627 let sm = UploadSessionStateMachine::from_state(current_state);
628 if !sm.can_transition(UploadSessionTransition::Expire) {
629 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<UploadSessionResponseDto>::error("Transition not allowed from current state")));
630 }
631
632 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
635 fields.insert("status".to_string(), serde_json::Value::String("expired".to_string()));
636
637 match backbone_core::write_guard::as_transition("expire", service.partial_update(&id, fields)).await {
638 Ok(Some(updated)) => {
639 let response: UploadSessionResponseDto = updated.into();
640 (StatusCode::OK, Json(ApiResponse::ok(response)))
641 }
642 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<UploadSessionResponseDto>::not_found("UploadSession", &id))),
643 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<UploadSessionResponseDto>::error(e.to_string()))),
644 }
645}
646
647pub async fn retry_transition(
651 axum::extract::State(service): axum::extract::State<Arc<UploadSessionService>>,
652 axum::extract::Path(id): axum::extract::Path<String>,
653 #[cfg(feature = "auth")] axum::Extension(auth): axum::Extension<AuthContext>,
654) -> impl axum::response::IntoResponse {
655 use axum::{http::StatusCode, Json};
656
657 let entity = match service.get_by_id(&id).await {
659 Ok(Some(e)) => e,
660 Ok(None) => return (StatusCode::NOT_FOUND, Json(ApiResponse::<UploadSessionResponseDto>::not_found("UploadSession", &id))),
661 Err(e) => return (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<UploadSessionResponseDto>::error(e.to_string()))),
662 };
663
664 #[cfg(feature = "auth")]
666 {
667 let allowed_roles = UploadSessionTransition::Retry.allowed_roles();
668 let has_specific_perm = auth.permissions.iter().any(|p| p == "upload_session:transition:retry");
669 let has_update_perm = auth.permissions.iter().any(|p| p == "upload_session:update");
670 let has_role = auth.roles.iter().any(|r| allowed_roles.contains(&r.as_str()));
671 if !has_specific_perm && !has_role {
672 if !has_update_perm {
673 return (StatusCode::FORBIDDEN, Json(ApiResponse::<UploadSessionResponseDto>::error("Insufficient permissions for retry transition")));
674 }
675 tracing::warn!(target: "backbone::write_guard", entity = "UploadSession", verb = "retry", code = "transition_by_update_permission", "a transition granted only by upload_session:update, running in shadow; it needs upload_session:transition:retry or a role of ["user"]");
676 }
677 }
678
679 let current_state: UploadSessionState = match entity.status.to_string().parse() {
681 Ok(s) => s,
682 Err(_) => return (StatusCode::CONFLICT, Json(ApiResponse::<UploadSessionResponseDto>::error("The stored state is not a state of this machine"))),
683 };
684 let sm = UploadSessionStateMachine::from_state(current_state);
685 if !sm.can_transition(UploadSessionTransition::Retry) {
686 return (StatusCode::BAD_REQUEST, Json(ApiResponse::<UploadSessionResponseDto>::error("Transition not allowed from current state")));
687 }
688
689 let mut fields: HashMap<String, serde_json::Value> = HashMap::new();
692 fields.insert("status".to_string(), serde_json::Value::String("initiated".to_string()));
693
694 match backbone_core::write_guard::as_transition("retry", service.partial_update(&id, fields)).await {
695 Ok(Some(updated)) => {
696 let response: UploadSessionResponseDto = updated.into();
697 (StatusCode::OK, Json(ApiResponse::ok(response)))
698 }
699 Ok(None) => (StatusCode::NOT_FOUND, Json(ApiResponse::<UploadSessionResponseDto>::not_found("UploadSession", &id))),
700 Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, Json(ApiResponse::<UploadSessionResponseDto>::error(e.to_string()))),
701 }
702}
703
704pub fn create_upload_session_transition_routes(service: Arc<UploadSessionService>) -> Router {
706 use axum::routing::post;
707
708 Router::new()
709 .route("/upload_sessions/:id/transitions/start_upload", post(start_upload_transition))
710 .route("/upload_sessions/:id/transitions/add_part", post(add_part_transition))
711 .route("/upload_sessions/:id/transitions/complete", post(complete_transition))
712 .route("/upload_sessions/:id/transitions/finish", post(finish_transition))
713 .route("/upload_sessions/:id/transitions/fail", post(fail_transition))
714 .route("/upload_sessions/:id/transitions/abort", post(abort_transition))
715 .route("/upload_sessions/:id/transitions/expire", post(expire_transition))
716 .route("/upload_sessions/:id/transitions/retry", post(retry_transition))
717 .with_state(service)
718}
719