Skip to main content

max / makenotwork

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