Skip to main content

max / makenotwork

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