Skip to main content

max / makenotwork

8.9 KB · 272 lines History Blame Raw
1 //! Item media metadata: post-upload file-size writebacks and per-content-type
2 //! S3 key/URL/metadata updates (audio, cover, video).
3
4 use sqlx::PgPool;
5
6 use crate::db::models::DbItem;
7 use crate::db::{ItemId, UserId};
8 use crate::error::{AppError, Result};
9
10 /// Outcome of [`update_item_file_cas`].
11 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
12 pub enum FileConfirmOutcome {
13 /// The compare-and-swap matched: the new key/size were written.
14 Committed,
15 /// The target s3 column no longer held the expected key (a concurrent
16 /// confirm won the race) or the item/owner no longer matched. The caller
17 /// MUST NOT credit storage and should treat the staged object as an orphan.
18 LostRace,
19 }
20
21 /// Confirm an uploaded item file onto its `items` row with a compare-and-swap.
22 ///
23 /// This is the guarded write for the generic item-file columns (audio/video
24 /// key + size), and the one every upload-confirm handler goes through. Raw
25 /// `UPDATE items SET <col> = ...` stays out of the handlers so none can
26 /// hand-roll an unguarded write missing the `IS NOT DISTINCT FROM` guard the
27 /// sibling version-confirm path has. Two writers touch the same columns without going through here:
28 /// `update_item_video_s3_key` below, which sets the key alone under an
29 /// ownership filter, and `db::scanning::promote_gated`, which writes
30 /// `audio_s3_key`/`video_s3_key` with neither a CAS predicate nor an
31 /// ownership filter.
32 ///
33 /// The CAS predicate (`<s3_col> IS NOT DISTINCT FROM expected_old_key`) updates
34 /// the row only if the column still holds the key the caller observed (`NULL`
35 /// for a first upload). A concurrent confirm that already swapped the column in
36 /// makes this match zero rows -> [`FileConfirmOutcome::LostRace`], so storage is
37 /// never double-credited and a live object is never clobbered.
38 ///
39 /// Takes any `PgExecutor` so the caller can run it inside the same transaction
40 /// as the storage credit (a rollback then undoes both atomically).
41 pub async fn update_item_file_cas<'e>(
42 executor: impl sqlx::PgExecutor<'e>,
43 item_id: ItemId,
44 owner: UserId,
45 file_type: crate::storage::FileType,
46 expected_old_key: Option<&str>,
47 new_key: &str,
48 size: i64,
49 ) -> Result<FileConfirmOutcome> {
50 use crate::storage::GenericItemConfirm;
51
52 // Column names come from the exhaustive `generic_item_confirm` match,
53 // `&'static str`, never user input, so the `format!` is injection-safe.
54 let (s3_col, size_col) = match file_type.generic_item_confirm() {
55 GenericItemConfirm::Columns { s3_key, size } => (s3_key, size),
56 GenericItemConfirm::UseRoute(route) => {
57 // The generic confirm handler rejects these before reaching here;
58 // a call with such a type is a programming error, not user input.
59 return Err(AppError::Internal(anyhow::anyhow!(
60 "update_item_file_cas called for {} which must use {route}",
61 file_type.as_str()
62 )));
63 }
64 };
65
66 let sql = format!(
67 "UPDATE items SET {s3_col} = $2, {size_col} = $3, updated_at = NOW() \
68 WHERE id = $1 \
69 AND project_id IN (SELECT id FROM projects WHERE user_id = $4) \
70 AND {s3_col} IS NOT DISTINCT FROM $5"
71 );
72 let res = sqlx::query(&sql)
73 .bind(item_id)
74 .bind(new_key)
75 .bind(size)
76 .bind(owner)
77 .bind(expected_old_key)
78 .execute(executor)
79 .await?;
80
81 Ok(if res.rows_affected() == 0 {
82 FileConfirmOutcome::LostRace
83 } else {
84 FileConfirmOutcome::Committed
85 })
86 }
87
88 /// Get the audio, cover, and video file sizes for an item (for storage decrement on delete).
89 #[tracing::instrument(skip_all)]
90 pub async fn get_item_file_sizes(
91 pool: &PgPool,
92 id: ItemId,
93 ) -> Result<crate::db::models::ItemFileSizes> {
94 let row = sqlx::query_as::<_, (Option<i64>, Option<i64>, Option<i64>)>(
95 "SELECT audio_file_size_bytes, cover_file_size_bytes, video_file_size_bytes FROM items WHERE id = $1",
96 )
97 .bind(id)
98 .fetch_optional(pool)
99 .await?;
100
101 match row {
102 Some((audio, cover, video)) => Ok(crate::db::models::ItemFileSizes {
103 audio_file_size_bytes: audio,
104 cover_file_size_bytes: cover,
105 video_file_size_bytes: video,
106 }),
107 None => Ok(crate::db::models::ItemFileSizes {
108 audio_file_size_bytes: None,
109 cover_file_size_bytes: None,
110 video_file_size_bytes: None,
111 }),
112 }
113 }
114
115 /// Update the audio file size on an item (defense-in-depth: verifies ownership).
116 #[tracing::instrument(skip_all)]
117 pub async fn update_item_audio_file_size(
118 pool: &PgPool,
119 item_id: ItemId,
120 user_id: UserId,
121 bytes: i64,
122 ) -> Result<()> {
123 sqlx::query(
124 "UPDATE items SET audio_file_size_bytes = $2 WHERE id = $1 AND project_id IN (SELECT id FROM projects WHERE user_id = $3)",
125 )
126 .bind(item_id)
127 .bind(bytes)
128 .bind(user_id)
129 .execute(pool)
130 .await?;
131
132 Ok(())
133 }
134
135 /// Atomically update cover image URL, S3 key, and file size in a single UPDATE
136 /// (defense-in-depth: verifies ownership), guarded by a compare-and-swap on the
137 /// existing `cover_s3_key`.
138 ///
139 /// Returns `true` when the row was actually updated, `false` when the UPDATE
140 /// matched zero rows, either the ownership filter no-matched (item deleted or
141 /// moved between projects mid-flight) OR the CAS predicate failed because a
142 /// concurrent confirm already swapped the cover key out from under the value
143 /// the caller observed (`expected_old_key`, `NULL` for a first cover). Without
144 /// the CAS, two concurrent cover confirms each deduct the old size and the loser
145 /// silently orphans its committed object, the same
146 /// lost-update shape the audio/video path seals via [`update_item_file_cas`].
147 /// Callers that fire side-effects after the write, storage credit, scan
148 /// enqueue, S3 orphan queueing, must check the bool and roll back on false.
149 #[tracing::instrument(skip_all)]
150 pub async fn update_item_cover<'e>(
151 executor: impl sqlx::PgExecutor<'e>,
152 item_id: ItemId,
153 user_id: UserId,
154 expected_old_key: Option<&str>,
155 url: &str,
156 s3_key: &str,
157 file_size_bytes: i64,
158 ) -> Result<bool> {
159 let result = sqlx::query(
160 r"UPDATE items
161 SET cover_image_url = $2, cover_s3_key = $3, cover_file_size_bytes = $4, updated_at = NOW()
162 WHERE id = $1
163 AND project_id IN (SELECT id FROM projects WHERE user_id = $5)
164 AND cover_s3_key IS NOT DISTINCT FROM $6",
165 )
166 .bind(item_id)
167 .bind(url)
168 .bind(s3_key)
169 .bind(file_size_bytes)
170 .bind(user_id)
171 .bind(expected_old_key)
172 .execute(executor)
173 .await?;
174
175 Ok(result.rows_affected() > 0)
176 }
177
178 /// Update the cover file size on an item (defense-in-depth: verifies ownership).
179 #[tracing::instrument(skip_all)]
180 pub async fn update_item_cover_file_size(
181 pool: &PgPool,
182 item_id: ItemId,
183 user_id: UserId,
184 bytes: i64,
185 ) -> Result<()> {
186 sqlx::query(
187 "UPDATE items SET cover_file_size_bytes = $2 WHERE id = $1 AND project_id IN (SELECT id FROM projects WHERE user_id = $3)",
188 )
189 .bind(item_id)
190 .bind(bytes)
191 .bind(user_id)
192 .execute(pool)
193 .await?;
194
195 Ok(())
196 }
197
198 /// Update the video S3 key for an item (defense-in-depth: verifies ownership).
199 #[tracing::instrument(skip_all)]
200 pub async fn update_item_video_s3_key(
201 pool: &PgPool,
202 item_id: ItemId,
203 user_id: UserId,
204 s3_key: &str,
205 ) -> Result<DbItem> {
206 let item = sqlx::query_as::<_, DbItem>(
207 r"
208 UPDATE items
209 SET video_s3_key = $2, updated_at = NOW()
210 WHERE id = $1
211 AND project_id IN (SELECT id FROM projects WHERE user_id = $3)
212 RETURNING *
213 ",
214 )
215 .bind(item_id)
216 .bind(s3_key)
217 .bind(user_id)
218 .fetch_one(pool)
219 .await?;
220
221 Ok(item)
222 }
223
224 /// Update the video file size on an item (defense-in-depth: verifies ownership).
225 #[tracing::instrument(skip_all)]
226 pub async fn update_item_video_file_size(
227 pool: &PgPool,
228 item_id: ItemId,
229 user_id: UserId,
230 bytes: i64,
231 ) -> Result<()> {
232 sqlx::query(
233 "UPDATE items SET video_file_size_bytes = $2 WHERE id = $1 AND project_id IN (SELECT id FROM projects WHERE user_id = $3)",
234 )
235 .bind(item_id)
236 .bind(bytes)
237 .bind(user_id)
238 .execute(pool)
239 .await?;
240
241 Ok(())
242 }
243
244 /// Update video metadata (duration, resolution) on an item (defense-in-depth: verifies ownership).
245 #[tracing::instrument(skip_all)]
246 pub async fn update_item_video_metadata(
247 pool: &PgPool,
248 item_id: ItemId,
249 user_id: UserId,
250 duration_seconds: Option<i32>,
251 width: Option<i32>,
252 height: Option<i32>,
253 ) -> Result<()> {
254 sqlx::query(
255 r"
256 UPDATE items
257 SET video_duration_seconds = $2, video_width = $3, video_height = $4, updated_at = NOW()
258 WHERE id = $1
259 AND project_id IN (SELECT id FROM projects WHERE user_id = $5)
260 ",
261 )
262 .bind(item_id)
263 .bind(duration_seconds)
264 .bind(width)
265 .bind(height)
266 .bind(user_id)
267 .execute(pool)
268 .await?;
269
270 Ok(())
271 }
272