//! Item media metadata: post-upload file-size writebacks and per-content-type //! S3 key/URL/metadata updates (audio, cover, video). use sqlx::PgPool; use crate::db::models::DbItem; use crate::db::{ItemId, UserId}; use crate::error::{AppError, Result}; /// Outcome of [`update_item_file_cas`]. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum FileConfirmOutcome { /// The compare-and-swap matched: the new key/size were written. Committed, /// The target s3 column no longer held the expected key (a concurrent /// confirm won the race) or the item/owner no longer matched. The caller /// MUST NOT credit storage and should treat the staged object as an orphan. LostRace, } /// Confirm an uploaded item file onto its `items` row with a compare-and-swap. /// /// This is the guarded write for the generic item-file columns (audio/video /// key + size), and the one every upload-confirm handler goes through. Raw /// `UPDATE items SET = ...` stays out of the handlers so none can /// hand-roll an unguarded write, the recurring CHRONIC where the item-confirm /// path lacked the `IS NOT DISTINCT FROM` guard its sibling version-confirm /// path had. Two writers touch the same columns without going through here: /// `update_item_video_s3_key` below, which sets the key alone under an /// ownership filter, and `db::scanning::promote_gated`, which writes /// `audio_s3_key`/`video_s3_key` with neither a CAS predicate nor an /// ownership filter. /// /// The CAS predicate (` IS NOT DISTINCT FROM expected_old_key`) updates /// the row only if the column still holds the key the caller observed (`NULL` /// for a first upload). A concurrent confirm that already swapped the column in /// makes this match zero rows -> [`FileConfirmOutcome::LostRace`], so storage is /// never double-credited and a live object is never clobbered. /// /// Takes any `PgExecutor` so the caller can run it inside the same transaction /// as the storage credit (a rollback then undoes both atomically). pub async fn update_item_file_cas<'e>( executor: impl sqlx::PgExecutor<'e>, item_id: ItemId, owner: UserId, file_type: crate::storage::FileType, expected_old_key: Option<&str>, new_key: &str, size: i64, ) -> Result { use crate::storage::GenericItemConfirm; // Column names come from the exhaustive `generic_item_confirm` match, // `&'static str`, never user input, so the `format!` is injection-safe. let (s3_col, size_col) = match file_type.generic_item_confirm() { GenericItemConfirm::Columns { s3_key, size } => (s3_key, size), GenericItemConfirm::UseRoute(route) => { // The generic confirm handler rejects these before reaching here; // a call with such a type is a programming error, not user input. return Err(AppError::Internal(anyhow::anyhow!( "update_item_file_cas called for {} which must use {route}", file_type.as_str() ))); } }; let sql = format!( "UPDATE items SET {s3_col} = $2, {size_col} = $3, updated_at = NOW() \ WHERE id = $1 \ AND project_id IN (SELECT id FROM projects WHERE user_id = $4) \ AND {s3_col} IS NOT DISTINCT FROM $5" ); let res = sqlx::query(&sql) .bind(item_id) .bind(new_key) .bind(size) .bind(owner) .bind(expected_old_key) .execute(executor) .await?; Ok(if res.rows_affected() == 0 { FileConfirmOutcome::LostRace } else { FileConfirmOutcome::Committed }) } /// Get the audio, cover, and video file sizes for an item (for storage decrement on delete). #[tracing::instrument(skip_all)] pub async fn get_item_file_sizes( pool: &PgPool, id: ItemId, ) -> Result { let row = sqlx::query_as::<_, (Option, Option, Option)>( "SELECT audio_file_size_bytes, cover_file_size_bytes, video_file_size_bytes FROM items WHERE id = $1", ) .bind(id) .fetch_optional(pool) .await?; match row { Some((audio, cover, video)) => Ok(crate::db::models::ItemFileSizes { audio_file_size_bytes: audio, cover_file_size_bytes: cover, video_file_size_bytes: video, }), None => Ok(crate::db::models::ItemFileSizes { audio_file_size_bytes: None, cover_file_size_bytes: None, video_file_size_bytes: None, }), } } /// Update the audio file size on an item (defense-in-depth: verifies ownership). #[tracing::instrument(skip_all)] pub async fn update_item_audio_file_size( pool: &PgPool, item_id: ItemId, user_id: UserId, bytes: i64, ) -> Result<()> { sqlx::query( "UPDATE items SET audio_file_size_bytes = $2 WHERE id = $1 AND project_id IN (SELECT id FROM projects WHERE user_id = $3)", ) .bind(item_id) .bind(bytes) .bind(user_id) .execute(pool) .await?; Ok(()) } /// Atomically update cover image URL, S3 key, and file size in a single UPDATE /// (defense-in-depth: verifies ownership), guarded by a compare-and-swap on the /// existing `cover_s3_key`. /// /// Returns `true` when the row was actually updated, `false` when the UPDATE /// matched zero rows, either the ownership filter no-matched (item deleted or /// moved between projects mid-flight) OR the CAS predicate failed because a /// concurrent confirm already swapped the cover key out from under the value /// the caller observed (`expected_old_key`, `NULL` for a first cover). Without /// the CAS, two concurrent cover confirms each deduct the old size and the loser /// silently orphans its committed object (Run #18 Storage B4), the same /// lost-update shape the audio/video path seals via [`update_item_file_cas`]. /// Callers that fire side-effects after the write, storage credit, scan /// enqueue, S3 orphan queueing, must check the bool and roll back on false. #[tracing::instrument(skip_all)] pub async fn update_item_cover<'e>( executor: impl sqlx::PgExecutor<'e>, item_id: ItemId, user_id: UserId, expected_old_key: Option<&str>, url: &str, s3_key: &str, file_size_bytes: i64, ) -> Result { let result = sqlx::query( r"UPDATE items SET cover_image_url = $2, cover_s3_key = $3, cover_file_size_bytes = $4, updated_at = NOW() WHERE id = $1 AND project_id IN (SELECT id FROM projects WHERE user_id = $5) AND cover_s3_key IS NOT DISTINCT FROM $6", ) .bind(item_id) .bind(url) .bind(s3_key) .bind(file_size_bytes) .bind(user_id) .bind(expected_old_key) .execute(executor) .await?; Ok(result.rows_affected() > 0) } /// Update the cover file size on an item (defense-in-depth: verifies ownership). #[tracing::instrument(skip_all)] pub async fn update_item_cover_file_size( pool: &PgPool, item_id: ItemId, user_id: UserId, bytes: i64, ) -> Result<()> { sqlx::query( "UPDATE items SET cover_file_size_bytes = $2 WHERE id = $1 AND project_id IN (SELECT id FROM projects WHERE user_id = $3)", ) .bind(item_id) .bind(bytes) .bind(user_id) .execute(pool) .await?; Ok(()) } /// Update the video S3 key for an item (defense-in-depth: verifies ownership). #[tracing::instrument(skip_all)] pub async fn update_item_video_s3_key( pool: &PgPool, item_id: ItemId, user_id: UserId, s3_key: &str, ) -> Result { let item = sqlx::query_as::<_, DbItem>( r" UPDATE items SET video_s3_key = $2, updated_at = NOW() WHERE id = $1 AND project_id IN (SELECT id FROM projects WHERE user_id = $3) RETURNING * ", ) .bind(item_id) .bind(s3_key) .bind(user_id) .fetch_one(pool) .await?; Ok(item) } /// Update the video file size on an item (defense-in-depth: verifies ownership). #[tracing::instrument(skip_all)] pub async fn update_item_video_file_size( pool: &PgPool, item_id: ItemId, user_id: UserId, bytes: i64, ) -> Result<()> { sqlx::query( "UPDATE items SET video_file_size_bytes = $2 WHERE id = $1 AND project_id IN (SELECT id FROM projects WHERE user_id = $3)", ) .bind(item_id) .bind(bytes) .bind(user_id) .execute(pool) .await?; Ok(()) } /// Update video metadata (duration, resolution) on an item (defense-in-depth: verifies ownership). #[tracing::instrument(skip_all)] pub async fn update_item_video_metadata( pool: &PgPool, item_id: ItemId, user_id: UserId, duration_seconds: Option, width: Option, height: Option, ) -> Result<()> { sqlx::query( r" UPDATE items SET video_duration_seconds = $2, video_width = $3, video_height = $4, updated_at = NOW() WHERE id = $1 AND project_id IN (SELECT id FROM projects WHERE user_id = $5) ", ) .bind(item_id) .bind(duration_seconds) .bind(width) .bind(height) .bind(user_id) .execute(pool) .await?; Ok(()) }