backbone_bucket/application/service/
multipart_upload_service.rs

1//! Multipart Upload Service
2//!
3//! Hand-written — NOT generated. This file is safe from regeneration.
4
5use std::sync::Arc;
6
7use chrono::{Duration, Utc};
8use uuid::Uuid;
9
10use super::error::{ServiceError, ServiceResult};
11use crate::domain::entity::{UploadSession, UploadStatus, StorageBackend};
12use crate::infrastructure::persistence::UploadSessionRepository;
13use crate::infrastructure::persistence::BucketRepository;
14use crate::infrastructure::persistence::UserQuotaRepository;
15
16const DEFAULT_SESSION_EXPIRY_HOURS: i64 = 24;
17const DEFAULT_CHUNK_SIZE: i32 = 5 * 1024 * 1024;
18
19pub struct MultipartUploadService {
20    session_repo: Arc<UploadSessionRepository>,
21    bucket_repo: Arc<BucketRepository>,
22    quota_repo: Arc<UserQuotaRepository>,
23}
24
25impl MultipartUploadService {
26    pub fn new(
27        session_repo: Arc<UploadSessionRepository>,
28        bucket_repo: Arc<BucketRepository>,
29        quota_repo: Arc<UserQuotaRepository>,
30    ) -> Self {
31        Self { session_repo, bucket_repo, quota_repo }
32    }
33
34    pub async fn initiate(
35        &self,
36        bucket_id: Uuid,
37        user_id: Uuid,
38        path: &str,
39        filename: &str,
40        mime_type: Option<&str>,
41        file_size: i64,
42        chunk_size: Option<i32>,
43    ) -> ServiceResult<UploadSession> {
44        let _bucket = self.bucket_repo
45            .find_by_id(&bucket_id.to_string())
46            .await
47            .map_err(|e| ServiceError::Repository(backbone_core::RepositoryError::DatabaseError(e.to_string())))?
48            .ok_or(ServiceError::NotFound)?;
49
50        if file_size <= 0 {
51            return Err(ServiceError::Validation("file_size must be positive".into()));
52        }
53
54        let chunk = chunk_size.unwrap_or(DEFAULT_CHUNK_SIZE);
55        if chunk <= 0 {
56            return Err(ServiceError::Validation("chunk_size must be positive".into()));
57        }
58        let total_chunks = ((file_size as f64) / (chunk as f64)).ceil() as i32;
59
60        let mut builder = UploadSession::builder()
61            .bucket_id(bucket_id)
62            .user_id(user_id)
63            .path(path.to_string())
64            .filename(filename.to_string())
65            .file_size(file_size)
66            .chunk_size(chunk)
67            .total_chunks(total_chunks)
68            .uploaded_chunks(0)
69            // A new session holds no parts yet; without this the builder refuses every initiate.
70            .completed_parts(Vec::new())
71            .status(UploadStatus::Initiated)
72            .storage_backend(StorageBackend::Local)
73            .expires_at(Utc::now() + Duration::hours(DEFAULT_SESSION_EXPIRY_HOURS));
74
75        if let Some(mt) = mime_type {
76            builder = builder.mime_type(mt.to_string());
77        }
78
79        let session = builder.build()
80            .map_err(|e| ServiceError::Validation(e))?;
81
82        let created = self.session_repo
83            .create(&session)
84            .await
85            .map_err(|e| ServiceError::Repository(backbone_core::RepositoryError::DatabaseError(e.to_string())))?;
86
87        Ok(created)
88    }
89
90    /// Check that a part may be written before its bytes are stored: the session is the
91    /// caller's, live, expecting this part number, the part not already recorded, and its size
92    /// within the session's chunk size.
93    pub async fn authorize_part(
94        &self,
95        session_id: Uuid,
96        user_id: Uuid,
97        part_number: i32,
98        part_bytes: i64,
99    ) -> ServiceResult<()> {
100        let session = self.get_active_session(session_id, user_id).await?;
101        if part_number < 1 || part_number > session.total_chunks {
102            return Err(ServiceError::Validation(
103                format!("part_number must be between 1 and {}", session.total_chunks)
104            ));
105        }
106        if session.completed_parts.contains(&part_number) {
107            return Err(ServiceError::AlreadyExists(format!("Part {} already uploaded", part_number)));
108        }
109        if part_bytes <= 0 || part_bytes > i64::from(session.chunk_size) {
110            return Err(ServiceError::Validation(format!(
111                "a part holds 1..={} bytes; this one holds {}",
112                session.chunk_size, part_bytes
113            )));
114        }
115        Ok(())
116    }
117
118    pub async fn record_part(
119        &self,
120        session_id: Uuid,
121        user_id: Uuid,
122        part_number: i32,
123    ) -> ServiceResult<UploadSession> {
124        let session = self.get_active_session(session_id, user_id).await?;
125
126        if part_number < 1 || part_number > session.total_chunks {
127            return Err(ServiceError::Validation(
128                format!("part_number must be between 1 and {}", session.total_chunks)
129            ));
130        }
131
132        if session.completed_parts.contains(&part_number) {
133            return Err(ServiceError::AlreadyExists(
134                format!("Part {} already uploaded", part_number)
135            ));
136        }
137
138        // Record the part itself, so a resent part is refused above instead of counting twice.
139        let mut updated_session = session;
140        updated_session.completed_parts.push(part_number);
141        updated_session.completed_parts.sort_unstable();
142        updated_session.uploaded_chunks = updated_session.completed_parts.len() as i32;
143        if updated_session.status == UploadStatus::Initiated {
144            updated_session.status = UploadStatus::Uploading;
145        }
146        updated_session.metadata.touch();
147        let id_str = updated_session.id.to_string();
148        self.session_repo.update(&id_str, &updated_session).await.map_err(|e| ServiceError::Repository(backbone_core::RepositoryError::DatabaseError(e.to_string())))?;
149
150        self.session_repo
151            .find_by_id(&session_id.to_string())
152            .await
153            .map_err(|e| ServiceError::Repository(backbone_core::RepositoryError::DatabaseError(e.to_string())))?
154            .ok_or(ServiceError::NotFound)
155    }
156
157    pub async fn complete(
158        &self,
159        session_id: Uuid,
160        user_id: Uuid,
161    ) -> ServiceResult<UploadSession> {
162        let mut session = self.get_active_session(session_id, user_id).await?;
163
164        if session.uploaded_chunks < session.total_chunks {
165            return Err(ServiceError::Validation(format!(
166                "Not all parts uploaded: {}/{} complete",
167                session.uploaded_chunks, session.total_chunks
168            )));
169        }
170
171        // Quota gate. Best-effort: a quota row may not exist for every
172        // user (admin-provisioned, opt-in) — absence means "no limit",
173        // not "deny". When a row exists, hard-reject if completing would
174        // push `used_bytes` past `limit_bytes`. The actual increment
175        // happens in `record_completed_usage` after the bytes land.
176        if let Some(quota) = self
177            .quota_repo
178            .find_by_user_id(user_id)
179            .await
180            .map_err(|e| ServiceError::Repository(backbone_core::RepositoryError::DatabaseError(e.to_string())))?
181        {
182            let projected = quota.used_bytes.saturating_add(session.file_size);
183            if projected > quota.limit_bytes {
184                return Err(ServiceError::Validation(format!(
185                    "user quota exceeded: {} + {} > {} bytes",
186                    quota.used_bytes, session.file_size, quota.limit_bytes
187                )));
188            }
189        }
190
191        session.status = UploadStatus::Completing;
192        session.metadata.touch();
193
194        let id_str = session.id.to_string();
195        self.session_repo
196            .update(&id_str, &session)
197            .await
198            .map_err(|e| ServiceError::Repository(backbone_core::RepositoryError::DatabaseError(e.to_string())))?
199            .ok_or(ServiceError::NotFound)
200    }
201
202    /// Check (without mutating) whether `user_id` has capacity for
203    /// `bytes` more. Returns `Ok(())` when there is no quota row (no
204    /// limit configured) or when `used_bytes + bytes <= limit_bytes`.
205    ///
206    /// Called by the single-shot upload handler before writing to
207    /// storage. The resumable flow calls into [`Self::complete`] which
208    /// performs the same check.
209    pub async fn check_capacity(&self, user_id: Uuid, bytes: i64) -> ServiceResult<()> {
210        let Some(quota) = self
211            .quota_repo
212            .find_by_user_id(user_id)
213            .await
214            .map_err(|e| ServiceError::Repository(backbone_core::RepositoryError::DatabaseError(e.to_string())))?
215        else {
216            return Ok(());
217        };
218        let projected = quota.used_bytes.saturating_add(bytes);
219        if projected > quota.limit_bytes {
220            return Err(ServiceError::Validation(format!(
221                "user quota exceeded: {} + {} > {} bytes",
222                quota.used_bytes, bytes, quota.limit_bytes
223            )));
224        }
225        Ok(())
226    }
227
228    /// Record post-upload usage on the user's quota row. Best-effort:
229    /// when no quota row exists this is a no-op. Call AFTER the storage
230    /// `put` and DB row commit succeed.
231    pub async fn record_completed_usage(
232        &self,
233        user_id: Uuid,
234        bytes: i64,
235    ) -> ServiceResult<()> {
236        let Some(mut quota) = self
237            .quota_repo
238            .find_by_user_id(user_id)
239            .await
240            .map_err(|e| ServiceError::Repository(backbone_core::RepositoryError::DatabaseError(e.to_string())))?
241        else {
242            return Ok(());
243        };
244        quota.used_bytes = quota.used_bytes.saturating_add(bytes);
245        quota.file_count = quota.file_count.saturating_add(1);
246        if quota.used_bytes > quota.peak_usage_bytes {
247            quota.peak_usage_bytes = quota.used_bytes;
248            quota.peak_usage_at = Some(Utc::now());
249        }
250        quota.metadata.touch();
251        let id_str = quota.id.to_string();
252        self.quota_repo
253            .update(&id_str, &quota)
254            .await
255            .map_err(|e| ServiceError::Repository(backbone_core::RepositoryError::DatabaseError(e.to_string())))?;
256        Ok(())
257    }
258
259    pub async fn abort(&self, session_id: Uuid, user_id: Uuid) -> ServiceResult<()> {
260        let mut session = self.get_active_session(session_id, user_id).await?;
261        session.status = UploadStatus::Failed;
262        session.metadata.touch();
263        let id_str = session.id.to_string();
264        self.session_repo.update(&id_str, &session).await.map_err(|e| ServiceError::Repository(backbone_core::RepositoryError::DatabaseError(e.to_string())))?;
265        Ok(())
266    }
267
268    /// Fail a session whose completion was refused after it began (its parts did not assemble to
269    /// the declared size), so it does not stay `completing` forever.
270    pub async fn mark_failed(&self, session_id: Uuid) -> ServiceResult<()> {
271        let mut session = self.session_repo
272            .find_by_id(&session_id.to_string())
273            .await
274            .map_err(|e| ServiceError::Repository(backbone_core::RepositoryError::DatabaseError(e.to_string())))?
275            .ok_or(ServiceError::NotFound)?;
276        session.status = UploadStatus::Failed;
277        session.metadata.touch();
278        let id_str = session.id.to_string();
279        self.session_repo.update(&id_str, &session).await.map_err(|e| ServiceError::Repository(backbone_core::RepositoryError::DatabaseError(e.to_string())))?;
280        Ok(())
281    }
282
283    pub async fn mark_completed(&self, session_id: Uuid) -> ServiceResult<()> {
284        let mut session = self.session_repo
285            .find_by_id(&session_id.to_string())
286            .await
287            .map_err(|e| ServiceError::Repository(backbone_core::RepositoryError::DatabaseError(e.to_string())))?
288            .ok_or(ServiceError::NotFound)?;
289
290        session.status = UploadStatus::Completed;
291        session.metadata.touch();
292        let id_str = session.id.to_string();
293        self.session_repo.update(&id_str, &session).await.map_err(|e| ServiceError::Repository(backbone_core::RepositoryError::DatabaseError(e.to_string())))?;
294        Ok(())
295    }
296
297    pub async fn cleanup_expired(&self) -> ServiceResult<u64> {
298        // TODO: session_repo.expire_stale_sessions — implement custom method
299        Err(ServiceError::Internal("expire_stale_sessions not yet implemented".to_string()))
300    }
301
302    pub async fn list_active(&self, _user_id: Uuid, _bucket_id: Uuid) -> ServiceResult<Vec<UploadSession>> {
303        // TODO: session_repo.find_by_user_and_bucket — implement custom method
304        Err(ServiceError::Internal("list_active not yet implemented".to_string()))
305    }
306
307    async fn get_active_session(&self, session_id: Uuid, user_id: Uuid) -> ServiceResult<UploadSession> {
308        // TODO: session_repo.find_active_by_id — implement custom method
309        let session = self.session_repo
310            .find_by_id(&session_id.to_string())
311            .await
312            .map_err(|e| ServiceError::Repository(backbone_core::RepositoryError::DatabaseError(e.to_string())))?
313            .ok_or(ServiceError::NotFound)?;
314
315        if session.user_id != user_id {
316            return Err(ServiceError::Validation("Upload session belongs to a different user".into()));
317        }
318
319        if session.expires_at < Utc::now() {
320            return Err(ServiceError::Validation("Upload session has expired".into()));
321        }
322
323        // A completed, completing, failed or expired session takes no more parts and no second
324        // completion.
325        if !matches!(session.status, UploadStatus::Initiated | UploadStatus::Uploading) {
326            return Err(ServiceError::Validation(format!("Upload session is {}", session.status)));
327        }
328
329        Ok(session)
330    }
331}
backbone_bucket/application/service/multipart_upload_service.rs

backbone_bucket/application/service/
multipart_upload_service.rs
