teamcraft/services/
uploads.rs

1//! Chunked, resumable uploads for large files (CAD assemblies, simulation
2//! results of several GB).
3//!
4//! A client announces an upload (revision, file name, size, optionally the
5//! SHA-256 and the base version), then sends the content in chunks, each one
6//! at the offset the server confirmed last (like the tus protocol: a chunk at
7//! any other offset is refused with 409, and the client asks the session for
8//! the offset to resume from). The chunks are appended to a staging file in
9//! the vault (`uploads/<id>`), never kept in memory beyond one chunk.
10//!
11//! When all bytes are there, the content is hashed from disk, checked against
12//! the announced size and SHA-256 and moved into the content-addressed vault.
13//! The file version is then created by the same code as every other upload
14//! ([`files::attach`] or [`files::check_in`]): check-out required, base-version
15//! guard, audit log, notifications.
16//!
17//! Sessions expire [`SESSION_HOURS`] after their last chunk; expired ones are
18//! removed at start-up and every hour ([`purge_expired`]).
19
20use std::time::Duration;
21
22use chrono::{DateTime, Utc};
23use serde::Serialize;
24use sqlx::{PgConnection, PgPool};
25use uuid::Uuid;
26
27use crate::{
28    auth::Actor,
29    domain::{self, BaseVersion},
30    error::{AppError, AppResult},
31    models::Revision,
32    services::{files, retention},
33    vault::{StoredBlob, Vault},
34};
35
36/// A session lives this long after its last chunk.
37pub const SESSION_HOURS: i32 = 24;
38/// Most unfinished sessions one user may hold – each reserves up to
39/// `TEAMCRAFT_MAX_UPLOAD_MB` of disk space in the vault.
40pub const MAX_OPEN_PER_USER: i64 = 16;
41
42const SESSION_SELECT: &str = "
43    SELECT id, user_id, revision_id, filename, content_type, total_bytes, received_bytes,
44           sha256, base_version, base_sha256, stored_sha256, folder, created_at, updated_at, expires_at
45    FROM upload_sessions";
46
47/// An upload in progress.
48#[derive(Clone, Debug, Serialize, sqlx::FromRow)]
49pub struct UploadSession {
50    pub id: Uuid,
51    pub user_id: Uuid,
52    /// The revision the file joins; `None` for an upload that joins the
53    /// creation of a new item (web "New item" form).
54    pub revision_id: Option<Uuid>,
55    pub filename: String,
56    pub content_type: String,
57    /// The announced size.
58    pub total_bytes: i64,
59    /// Bytes confirmed so far = the offset of the next chunk.
60    pub received_bytes: i64,
61    /// The SHA-256 the client announced, if any.
62    pub sha256: Option<String>,
63    pub base_version: Option<i32>,
64    pub base_sha256: Option<String>,
65    /// Set once the content was verified and moved into the vault.
66    pub stored_sha256: Option<String>,
67    /// Folder inside the project group (`files.folder`); `None` keeps the
68    /// folder of an existing file.
69    pub folder: Option<String>,
70    pub created_at: DateTime<Utc>,
71    pub updated_at: DateTime<Utc>,
72    pub expires_at: DateTime<Utc>,
73}
74
75impl UploadSession {
76    pub fn offset(&self) -> u64 {
77        u64::try_from(self.received_bytes).unwrap_or(0)
78    }
79
80    pub fn total(&self) -> u64 {
81        u64::try_from(self.total_bytes).unwrap_or(0)
82    }
83
84    /// All bytes received.
85    pub fn complete(&self) -> bool {
86        self.received_bytes >= self.total_bytes
87    }
88
89    pub fn base(&self) -> Option<BaseVersion> {
90        match (&self.base_version, &self.base_sha256) {
91            (Some(v), _) => Some(BaseVersion::Version(*v)),
92            (None, Some(sha)) => Some(BaseVersion::Sha256(sha.clone())),
93            (None, None) => None,
94        }
95    }
96}
97
98/// What a client announces when it starts an upload.
99#[derive(Clone, Debug)]
100pub struct NewSession {
101    /// `None`: the file joins a new item (needs the editor role).
102    pub revision_id: Option<Uuid>,
103    pub filename: String,
104    pub size: u64,
105    pub sha256: Option<String>,
106    pub content_type: Option<String>,
107    pub base: Option<BaseVersion>,
108    /// Folder inside the project group (normalised here); `None` keeps the
109    /// folder of an existing file (`""` for a new one).
110    pub folder: Option<String>,
111}
112
113/// A finished upload: the stored version and, for a check-in, the revision
114/// after the lock was released.
115pub struct Finished {
116    pub stored: Vec<files::Stored>,
117    pub upload: files::Upload,
118    pub revision: Option<Revision>,
119}
120
121/// Checks a chunk against the session state and returns the new offset.
122/// `offset` must be exactly the bytes received so far (409 otherwise – the
123/// client resumes from the server's offset), and the chunk must not go
124/// beyond the announced size.
125pub fn check_chunk(received: u64, total: u64, offset: u64, len: u64) -> AppResult<u64> {
126    if offset != received {
127        return Err(AppError::Conflict(format!(
128            "Upload-Offset {offset} does not match the {received} bytes received so far – \
129             resume at offset {received}."
130        )));
131    }
132    if len == 0 && received < total {
133        return Err(AppError::BadRequest("The chunk is empty.".into()));
134    }
135    let end = received
136        .checked_add(len)
137        .filter(|end| *end <= total)
138        .ok_or_else(|| {
139            AppError::BadRequest(format!(
140                "The chunk ends at byte {}, beyond the announced size of {total} bytes.",
141                received.saturating_add(len)
142            ))
143        })?;
144    Ok(end)
145}
146
147/// Checks the received content against what the client announced.
148pub fn verify_content(
149    total: u64,
150    expected_sha256: Option<&str>,
151    blob: &StoredBlob,
152) -> AppResult<()> {
153    if blob.size != total {
154        return Err(AppError::BadRequest(format!(
155            "The upload has {} bytes instead of the announced {total} – it was reset, upload \
156             it again.",
157            blob.size
158        )));
159    }
160    if let Some(expected) = expected_sha256 {
161        if !expected.eq_ignore_ascii_case(&blob.sha256) {
162            return Err(AppError::BadRequest(format!(
163                "The content does not match the announced SHA-256 (received {}…) – the upload \
164                 was reset, upload it again.",
165                &blob.sha256[..12]
166            )));
167        }
168    }
169    Ok(())
170}
171
172/// Parses the `uploads` field of a multipart check-in or upload: the ids of
173/// completed chunked uploads, as a JSON array or separated by commas.
174pub fn parse_ids(raw: &str) -> AppResult<Vec<Uuid>> {
175    let invalid = || {
176        AppError::BadRequest(
177            "The field 'uploads' must list upload ids: [\"…\", …] or id,id.".into(),
178        )
179    };
180    let raw = raw.trim();
181    let parts: Vec<String> = if raw.starts_with('[') {
182        serde_json::from_str(raw).map_err(|_| invalid())?
183    } else {
184        raw.split(',')
185            .map(str::trim)
186            .filter(|s| !s.is_empty())
187            .map(str::to_string)
188            .collect()
189    };
190    let mut ids = Vec::with_capacity(parts.len());
191    for part in parts {
192        let id = Uuid::try_parse(part.trim()).map_err(|_| invalid())?;
193        if !ids.contains(&id) {
194            ids.push(id);
195        }
196    }
197    Ok(ids)
198}
199
200fn clean_sha256(raw: Option<&str>) -> AppResult<Option<String>> {
201    match raw.map(str::trim).filter(|s| !s.is_empty()) {
202        None => Ok(None),
203        Some(s) if s.len() == 64 && s.bytes().all(|b| b.is_ascii_hexdigit()) => {
204            Ok(Some(s.to_ascii_lowercase()))
205        }
206        Some(_) => Err(AppError::BadRequest(
207            "sha256 must be 64 hexadecimal digits.".into(),
208        )),
209    }
210}
211
212fn not_found() -> AppError {
213    AppError::not_found("Upload not found – it may have expired or been cancelled.")
214}
215
216async fn load(
217    conn: &mut PgConnection,
218    actor: &Actor,
219    id: Uuid,
220    lock: bool,
221) -> AppResult<UploadSession> {
222    let sql = format!(
223        "{SESSION_SELECT} WHERE id = $1 AND expires_at > now(){}",
224        if lock { " FOR UPDATE" } else { "" }
225    );
226    let session = sqlx::query_as::<_, UploadSession>(&sql)
227        .bind(id)
228        .fetch_optional(&mut *conn)
229        .await?
230        .ok_or_else(not_found)?;
231    // Sessions are private: other users get the same answer as for an
232    // unknown id.
233    if session.user_id != actor.id {
234        return Err(not_found());
235    }
236    Ok(session)
237}
238
239/// Starts an upload into a revision the actor has checked out. Refuses it
240/// early when the file cannot be stored anyway: too large, not checked out,
241/// or a stated base that the server file has moved on from.
242///
243/// # Invariants
244///
245/// * **Permission:** with a revision: CAD editor in the project group with the
246///   revision in work and checked out by the actor (`files::ensure_can_attach`;
247///   hidden = 404, 403 / 409 otherwise). Without one (new-item session): only
248///   the editor role, no `base` allowed; the session can only join
249///   `items::create_with_files`, which checks the project group (including
250///   restricted groups). The session belongs to the actor alone.
251/// * **Locks:** the actor's user row (`FOR NO KEY UPDATE`), which serialises the
252///   quota check: at most 16 unexpired sessions per user (409); the announced
253///   size must fit the storage quotas (413, `services::quotas`).
254/// * **Steps:** the staging file is created before the commit and removed again
255///   when the commit fails.
256/// * **Audit, notifications, webhooks:** none.
257/// * **Atomic:** one transaction for the session row.
258pub async fn create(
259    pool: &PgPool,
260    vault: &Vault,
261    actor: &Actor,
262    new: NewSession,
263    max_bytes: u64,
264) -> AppResult<UploadSession> {
265    if new.filename.trim().is_empty() {
266        return Err(AppError::BadRequest("The file name is missing.".into()));
267    }
268    let filename = files::stored_name(&new.filename);
269    if new.size > max_bytes {
270        return Err(AppError::PayloadTooLarge(format!(
271            "The file exceeds the upload limit of {} MiB.",
272            max_bytes / (1024 * 1024)
273        )));
274    }
275    let total = i64::try_from(new.size)
276        .map_err(|_| AppError::PayloadTooLarge("The file is too large.".into()))?;
277    let sha256 = clean_sha256(new.sha256.as_deref())?;
278    let folder = new
279        .folder
280        .as_deref()
281        .map(domain::normalize_folder)
282        .transpose()?;
283    let rev = match new.revision_id {
284        Some(revision_id) => Some(files::ensure_can_attach(pool, actor, revision_id).await?),
285        None => {
286            actor.require(domain::Role::Editor)?;
287            if new.base.is_some() {
288                return Err(AppError::BadRequest(
289                    "A new item has no files to base the upload on.".into(),
290                ));
291            }
292            None
293        }
294    };
295    if let (Some(base), Some(rev)) = (&new.base, &rev) {
296        files::precheck_base(pool, rev.id, &filename, base, sha256.as_deref()).await?;
297    }
298    let content_type = files::clean_content_type(new.content_type.as_deref());
299    let (base_version, base_sha256) = match new.base {
300        Some(BaseVersion::Version(v)) => (Some(v), None),
301        Some(BaseVersion::Sha256(sha)) => (None, Some(sha.to_ascii_lowercase())),
302        None => (None, None),
303    };
304
305    let mut tx = pool.begin().await?;
306    // Serialises the quota check of one user (NO KEY: rows that merely
307    // reference the user, e.g. audit entries, are not blocked).
308    sqlx::query("SELECT id FROM users WHERE id = $1 FOR NO KEY UPDATE")
309        .bind(actor.id)
310        .execute(&mut *tx)
311        .await?;
312    let open: i64 = sqlx::query_scalar(
313        "SELECT count(*) FROM upload_sessions WHERE user_id = $1 AND expires_at > now()",
314    )
315    .bind(actor.id)
316    .fetch_one(&mut *tx)
317    .await?;
318    if open >= MAX_OPEN_PER_USER {
319        return Err(AppError::Conflict(format!(
320            "You have {open} unfinished uploads – finish or cancel some first."
321        )));
322    }
323    // Storage quotas, before any bytes are sent (storing checks them again).
324    crate::services::quotas::ensure_room(&mut tx, actor.id, total).await?;
325    let id = Uuid::new_v4();
326    let session = sqlx::query_as::<_, UploadSession>(
327        "INSERT INTO upload_sessions (id, user_id, revision_id, filename, content_type,
328                                      total_bytes, sha256, base_version, base_sha256, folder,
329                                      expires_at)
330         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, now() + make_interval(hours => $11))
331         RETURNING id, user_id, revision_id, filename, content_type, total_bytes, received_bytes,
332                   sha256, base_version, base_sha256, stored_sha256, folder, created_at, updated_at,
333                   expires_at",
334    )
335    .bind(id)
336    .bind(actor.id)
337    .bind(rev.as_ref().map(|r| r.id))
338    .bind(&filename)
339    .bind(&content_type)
340    .bind(total)
341    .bind(&sha256)
342    .bind(base_version)
343    .bind(&base_sha256)
344    .bind(&folder)
345    .bind(SESSION_HOURS)
346    .fetch_one(&mut *tx)
347    .await?;
348    vault.create_staging(id).await?;
349    if let Err(e) = tx.commit().await {
350        let _ = vault.remove_staged(id).await;
351        return Err(e.into());
352    }
353    Ok(session)
354}
355
356/// The state of one of the actor's uploads (its offset to resume from).
357pub async fn get(pool: &PgPool, actor: &Actor, id: Uuid) -> AppResult<UploadSession> {
358    let mut conn = pool.acquire().await?;
359    load(&mut conn, actor, id, false).await
360}
361
362/// Appends a chunk at `offset` (see [`check_chunk`]) and returns the
363/// updated session. The chunk is on disk before the new offset is
364/// confirmed, so an interruption at any point loses at most this chunk.
365///
366/// # Invariants
367///
368/// * **Permission:** the actor's own session (anybody else's, or an expired
369///   one, is a 404).
370/// * **Locks:** the session row (`FOR UPDATE`), so concurrent chunks of one
371///   session are serialised.
372/// * **Data:** the chunk must start at the confirmed offset (409 otherwise) and
373///   stay within the announced size (400); a session whose content is already
374///   verified takes no more chunks (409). Each chunk renews the expiry.
375/// * **Audit, notifications, webhooks:** none.
376/// * **Atomic:** the chunk is written to the staging file before the offset is
377///   updated in the transaction; when the update fails, the offset stays and
378///   the chunk is sent again.
379pub async fn append(
380    pool: &PgPool,
381    vault: &Vault,
382    actor: &Actor,
383    id: Uuid,
384    offset: u64,
385    chunk: &[u8],
386    max_chunk: u64,
387) -> AppResult<UploadSession> {
388    if chunk.len() as u64 > max_chunk {
389        return Err(AppError::PayloadTooLarge(format!(
390            "A chunk may have at most {} MiB.",
391            max_chunk / (1024 * 1024)
392        )));
393    }
394    let mut tx = pool.begin().await?;
395    let session = load(&mut tx, actor, id, true).await?;
396    if session.stored_sha256.is_some() {
397        return Err(AppError::Conflict(
398            "This upload is complete already – finish it.".into(),
399        ));
400    }
401    let end = check_chunk(
402        session.offset(),
403        session.total(),
404        offset,
405        chunk.len() as u64,
406    )?;
407    vault.write_staged(id, offset, chunk).await?;
408    let session = sqlx::query_as::<_, UploadSession>(
409        "UPDATE upload_sessions
410         SET received_bytes = $2, updated_at = now(),
411             expires_at = now() + make_interval(hours => $3)
412         WHERE id = $1
413         RETURNING id, user_id, revision_id, filename, content_type, total_bytes, received_bytes,
414                   sha256, base_version, base_sha256, stored_sha256, folder, created_at, updated_at,
415                   expires_at",
416    )
417    .bind(id)
418    .bind(i64::try_from(end).unwrap_or(i64::MAX))
419    .bind(SESSION_HOURS)
420    .fetch_one(&mut *tx)
421    .await?;
422    tx.commit().await?;
423    Ok(session)
424}
425
426/// Verifies a fully received upload (size, announced SHA-256), moves it into
427/// the vault and classifies it like a regular upload, with the session's
428/// base and folder. Nothing is registered in the database yet – pass the result to
429/// [`files::attach`] / [`files::check_in`] and then call [`forget`].
430///
431/// Calling it again after a failed attach (e.g. the database was briefly
432/// unavailable) reuses the stored content – nothing has to be re-sent.
433/// Content that fails verification resets the session to offset 0.
434///
435/// # Invariants
436///
437/// * **Permission:** the actor's own session (404 otherwise); `revision_id`,
438///   when given, must be the session's revision (400). [`complete_in`] applies the [`Scope`]
439///   rules: `NewItem` accepts only sessions without a revision, `Revision(id)`
440///   only that revision, `Any` whatever the session allows.
441/// * **Locks:** the session row (`FOR UPDATE`).
442/// * **Data:** all bytes must have arrived (409 otherwise). Size and announced
443///   SHA-256 are verified; a mismatch (400) or lost staging data (409) resets
444///   the session to offset 0 and commits that reset. The verified hash is
445///   recorded on the session, so a repeated call reuses the stored content.
446/// * **Steps:** the staged file moves into the vault before the transaction
447///   commits; a crash in between is recovered when the announced SHA-256 is in
448///   the vault. The session row stays until [`forget`] or [`abort`].
449/// * **Audit, notifications, webhooks:** none; nothing is registered in the
450///   database as a file.
451pub async fn complete(
452    pool: &PgPool,
453    vault: &Vault,
454    actor: &Actor,
455    id: Uuid,
456    revision_id: Option<Uuid>,
457) -> AppResult<files::Upload> {
458    let scope = match revision_id {
459        Some(rid) => Scope::Revision(rid),
460        None => Scope::Any,
461    };
462    complete_in(pool, vault, actor, id, scope).await
463}
464
465/// Where a finished upload may be used.
466#[derive(Clone, Copy, Debug)]
467pub enum Scope {
468    /// Wherever the session allows it.
469    Any,
470    /// Only the given revision (400 otherwise).
471    Revision(Uuid),
472    /// Only the creation of a new item: sessions without a revision (400 otherwise).
473    NewItem,
474}
475
476/// [`complete`] with the [`Scope`] the caller will store the file in; the
477/// invariants are those of [`complete`].
478pub async fn complete_in(
479    pool: &PgPool,
480    vault: &Vault,
481    actor: &Actor,
482    id: Uuid,
483    scope: Scope,
484) -> AppResult<files::Upload> {
485    let mut tx = pool.begin().await?;
486    let session = load(&mut tx, actor, id, true).await?;
487    let allowed = match scope {
488        Scope::Any => true,
489        Scope::Revision(rid) => session.revision_id == Some(rid),
490        Scope::NewItem => session.revision_id.is_none(),
491    };
492    if !allowed {
493        return Err(AppError::BadRequest(format!(
494            "The upload of {} belongs to another revision or to a new item.",
495            session.filename
496        )));
497    }
498    let stored = match &session.stored_sha256 {
499        Some(sha) if vault.exists(sha).await => Some(sha.clone()),
500        _ => None,
501    };
502    let blob = match stored {
503        Some(sha256) => StoredBlob {
504            sha256,
505            size: session.total(),
506        },
507        None => {
508            if !session.complete() {
509                return Err(AppError::Conflict(format!(
510                    "The upload of {} is incomplete: {} of {} bytes received.",
511                    session.filename, session.received_bytes, session.total_bytes
512                )));
513            }
514            // Moving the staging file into the vault scans it for malware:
515            // infected content is refused (400) like a content mismatch.
516            let found = match staged_blob(vault, &session).await {
517                Ok((blob, true)) => vault.adopt_staged(id, &blob.sha256).await.map(|()| blob),
518                Ok((blob, false)) => Ok(blob),
519                Err(e) => Err(e),
520            };
521            let blob = match found {
522                Ok(blob) => blob,
523                Err(e @ (AppError::BadRequest(_) | AppError::Conflict(_))) => {
524                    // Start over at offset 0 (also clears a recorded hash
525                    // whose blob is missing, so new chunks are accepted).
526                    vault.create_staging(id).await?;
527                    sqlx::query(
528                        "UPDATE upload_sessions
529                         SET received_bytes = 0, stored_sha256 = NULL, updated_at = now()
530                         WHERE id = $1",
531                    )
532                    .bind(id)
533                    .execute(&mut *tx)
534                    .await?;
535                    tx.commit().await?;
536                    return Err(e);
537                }
538                Err(e) => return Err(e),
539            };
540            sqlx::query("UPDATE upload_sessions SET stored_sha256 = $2 WHERE id = $1")
541                .bind(id)
542                .bind(&blob.sha256)
543                .execute(&mut *tx)
544                .await?;
545            blob
546        }
547    };
548    tx.commit().await?;
549    let mut upload =
550        files::Upload::classify(vault, blob, session.filename.clone(), &session.content_type).await;
551    upload.base = session.base();
552    upload.folder = session.folder.clone();
553    Ok(upload)
554}
555
556/// The verified content of a fully received upload and whether its staging
557/// file still has to move into the vault. A staging file that is gone
558/// although all bytes were confirmed was moved into the vault by an earlier
559/// [`complete`] whose transaction then failed (database error, crash): with
560/// an announced SHA-256 that is in the vault the content is recovered,
561/// otherwise the upload has to start over (409).
562async fn staged_blob(vault: &Vault, session: &UploadSession) -> AppResult<(StoredBlob, bool)> {
563    match vault.hash_staged(session.id).await {
564        Ok(blob) => {
565            verify_content(session.total(), session.sha256.as_deref(), &blob)?;
566            Ok((blob, true))
567        }
568        Err(AppError::Io(e)) if e.kind() == std::io::ErrorKind::NotFound => match &session.sha256 {
569            Some(sha) if vault.exists(sha).await => Ok((
570                StoredBlob {
571                    sha256: sha.clone(),
572                    size: session.total(),
573                },
574                false,
575            )),
576            _ => Err(AppError::Conflict(format!(
577                "The data of the upload of {} is gone – it was reset, upload it again.",
578                session.filename
579            ))),
580        },
581        Err(e) => Err(e),
582    }
583}
584
585/// Removes sessions whose content became file versions. Best effort: the
586/// versions are stored at this point, and a session left behind expires.
587pub async fn forget(pool: &PgPool, vault: &Vault, ids: &[Uuid]) {
588    if ids.is_empty() {
589        return;
590    }
591    if let Err(e) = sqlx::query("DELETE FROM upload_sessions WHERE id = ANY($1)")
592        .bind(ids)
593        .execute(pool)
594        .await
595    {
596        tracing::warn!(error = %e, "removing finished upload sessions failed");
597    }
598    for id in ids {
599        if let Err(e) = vault.remove_staged(*id).await {
600            tracing::warn!(error = %e, "removing an upload staging file failed");
601        }
602    }
603}
604
605/// Completes an upload and stores it as a new file or version of its
606/// revision – with `check_in` also releasing the check-out, in one
607/// transaction (like a multipart check-in).
608///
609/// # Invariants
610///
611/// * **Permission:** the actor's own session; storing then has the checks of
612///   [`files::check_in`] or [`files::attach`] (checked out by the actor).
613///   Sessions without a revision are refused (400).
614/// * **Steps:** completing the session (size and SHA-256 verified) and storing
615///   it are two transactions. When storing fails, the verified content stays
616///   in the session and `finish` can be repeated without sending the bytes
617///   again (`tests/atomic_processes.rs`); the session is removed only after
618///   storing succeeded.
619/// * **Audit, notifications, webhooks:** those of [`files::check_in`] /
620///   [`files::attach`] (`file.upload` / `file.version`, `revision.checkin`).
621pub async fn finish(
622    pool: &PgPool,
623    vault: &Vault,
624    actor: &Actor,
625    id: Uuid,
626    comment: &str,
627    check_in: bool,
628) -> AppResult<Finished> {
629    let session = get(pool, actor, id).await?;
630    let revision_id = session.revision_id.ok_or_else(|| {
631        AppError::BadRequest("This upload belongs to a new item, not to a revision.".into())
632    })?;
633    let upload = complete(pool, vault, actor, id, None).await?;
634    let uploads = std::slice::from_ref(&upload);
635    let (stored, revision) = if check_in {
636        let (rev, stored) = files::check_in(pool, actor, revision_id, uploads, comment).await?;
637        (stored, Some(rev))
638    } else {
639        let stored = files::attach(pool, actor, revision_id, uploads, comment).await?;
640        (stored, None)
641    };
642    forget(pool, vault, &[id]).await;
643    Ok(Finished {
644        stored,
645        upload,
646        revision,
647    })
648}
649
650/// Cancels an upload and deletes what was received.
651///
652/// # Invariants
653///
654/// * **Permission:** the actor's own session (404 otherwise).
655/// * **Locks:** the session row (`FOR UPDATE`).
656/// * **Steps:** the session row is deleted and committed first, then the staging
657///   file is removed; a failure in between leaves a staging file without a
658///   session, which `purge_expired` removes later.
659/// * **Audit, notifications, webhooks:** none.
660pub async fn abort(pool: &PgPool, vault: &Vault, actor: &Actor, id: Uuid) -> AppResult<()> {
661    let mut tx = pool.begin().await?;
662    load(&mut tx, actor, id, true).await?;
663    sqlx::query("DELETE FROM upload_sessions WHERE id = $1")
664        .bind(id)
665        .execute(&mut *tx)
666        .await?;
667    tx.commit().await?;
668    vault.remove_staged(id).await
669}
670
671/// Removes expired sessions with their data, and staging files that have no
672/// session any more (e.g. after a crash between the two). Content that was
673/// verified and moved into the vault but never became a file version goes
674/// too – unless something else refers to it or it was stored again within
675/// [`retention::IN_FLIGHT_GRACE`]; such a session stays (expired, unusable)
676/// until the next run. Returns the number of sessions removed.
677pub async fn purge_expired(pool: &PgPool, vault: &Vault) -> AppResult<u64> {
678    let mut expired: Vec<Uuid> = sqlx::query_scalar(
679        "DELETE FROM upload_sessions WHERE expires_at <= now() AND stored_sha256 IS NULL
680         RETURNING id",
681    )
682    .fetch_all(pool)
683    .await?;
684    let stored: Vec<Uuid> = sqlx::query_scalar(
685        "SELECT id FROM upload_sessions WHERE expires_at <= now() AND stored_sha256 IS NOT NULL",
686    )
687    .fetch_all(pool)
688    .await?;
689    for id in stored {
690        let mut tx = pool.begin().await?;
691        // the blob is released as seen without this session
692        let Some(sha256) = sqlx::query_scalar::<_, String>(
693            "DELETE FROM upload_sessions WHERE id = $1 AND expires_at <= now()
694             RETURNING stored_sha256",
695        )
696        .bind(id)
697        .fetch_optional(&mut *tx)
698        .await?
699        else {
700            continue;
701        };
702        if retention::release_blob(&mut tx, vault, &sha256, retention::IN_FLIGHT_GRACE).await?
703            == retention::Release::Recent
704        {
705            tx.rollback().await?;
706        } else {
707            tx.commit().await?;
708            expired.push(id);
709        }
710    }
711    for id in &expired {
712        vault.remove_staged(*id).await?;
713    }
714    let lifetime = Duration::from_secs(u64::try_from(SESSION_HOURS).unwrap_or(24) * 3600);
715    let stale = vault.stale_staged(lifetime).await?;
716    if !stale.is_empty() {
717        let known: Vec<Uuid> =
718            sqlx::query_scalar("SELECT id FROM upload_sessions WHERE id = ANY($1)")
719                .bind(&stale)
720                .fetch_all(pool)
721                .await?;
722        for id in stale.iter().filter(|id| !known.contains(id)) {
723            vault.remove_staged(*id).await?;
724        }
725    }
726    Ok(expired.len() as u64)
727}
728
729#[cfg(test)]
730mod tests {
731    use super::*;
732
733    #[test]
734    fn chunks_must_start_at_the_confirmed_offset() {
735        assert_eq!(check_chunk(0, 10, 0, 4).unwrap(), 4);
736        assert_eq!(check_chunk(4, 10, 4, 6).unwrap(), 10);
737        // a repeated or skipped chunk is refused with 409
738        assert!(matches!(
739            check_chunk(4, 10, 0, 4),
740            Err(AppError::Conflict(m)) if m.contains("resume at offset 4")
741        ));
742        assert!(matches!(
743            check_chunk(4, 10, 8, 2),
744            Err(AppError::Conflict(_))
745        ));
746    }
747
748    #[test]
749    fn chunks_stay_within_the_announced_size() {
750        assert!(matches!(
751            check_chunk(8, 10, 8, 3),
752            Err(AppError::BadRequest(_))
753        ));
754        assert!(matches!(
755            check_chunk(0, 10, 0, 0),
756            Err(AppError::BadRequest(_))
757        ));
758        assert!(check_chunk(u64::MAX - 1, u64::MAX, u64::MAX - 1, 5).is_err());
759        // an empty file needs no chunk, an empty chunk at its end is harmless
760        assert_eq!(check_chunk(0, 0, 0, 0).unwrap(), 0);
761    }
762
763    #[test]
764    fn content_is_verified() {
765        let blob = StoredBlob {
766            sha256: "ab".repeat(32),
767            size: 10,
768        };
769        assert!(verify_content(10, None, &blob).is_ok());
770        assert!(verify_content(10, Some(&"AB".repeat(32)), &blob).is_ok());
771        assert!(verify_content(11, None, &blob).is_err());
772        assert!(verify_content(10, Some(&"cd".repeat(32)), &blob).is_err());
773    }
774
775    #[test]
776    fn upload_ids() {
777        let a = Uuid::new_v4();
778        let b = Uuid::new_v4();
779        assert_eq!(parse_ids(&format!("[\"{a}\", \"{b}\"]")).unwrap(), [a, b]);
780        assert_eq!(parse_ids(&format!("{a}, {b},{a}")).unwrap(), [a, b]);
781        assert!(parse_ids("").unwrap().is_empty());
782        assert!(parse_ids("[1, 2]").is_err());
783        assert!(parse_ids("not-an-id").is_err());
784    }
785
786    #[test]
787    fn announced_hashes_are_validated() {
788        assert_eq!(clean_sha256(None).unwrap(), None);
789        assert_eq!(clean_sha256(Some(" ")).unwrap(), None);
790        assert_eq!(
791            clean_sha256(Some(&"AB".repeat(32))).unwrap(),
792            Some("ab".repeat(32))
793        );
794        assert!(clean_sha256(Some("abc")).is_err());
795        assert!(clean_sha256(Some(&"zz".repeat(32))).is_err());
796    }
797}
teamcraft/services/uploads.rs

teamcraft/services/
uploads.rs
