Skip to main content

max / makenotwork

13.6 KB · 383 lines History Blame Raw
1 //! Item version management: release creation, listing, and download tracking.
2
3 use sqlx::PgPool;
4
5 use super::models::{DbVersion, VersionS3KeyRow};
6 use super::{ItemId, UserId, VersionId};
7 use crate::error::Result;
8
9 /// Create a new version for an item, marking it as the current release.
10 ///
11 /// Wrapped in a transaction so the UPDATE (clearing the old `is_current`
12 /// flag) and INSERT (setting the new version as current) either both
13 /// succeed or both roll back.
14 #[tracing::instrument(skip_all)]
15 #[allow(clippy::too_many_arguments)]
16 pub async fn create_version(
17 pool: &PgPool,
18 item_id: ItemId,
19 version_number: &str,
20 changelog: Option<&str>,
21 file_url: Option<&str>,
22 file_size_bytes: Option<i64>,
23 file_name: Option<&str>,
24 label: Option<&str>,
25 ) -> Result<DbVersion> {
26 let mut tx = pool.begin().await?;
27
28 // Unset current on older version numbers (versions with the same number stay current)
29 sqlx::query!(
30 "UPDATE versions SET is_current = false WHERE item_id = $1 AND version_number != $2",
31 item_id as ItemId,
32 version_number,
33 )
34 .execute(&mut *tx)
35 .await?;
36
37 // Create new version as current
38 let version = sqlx::query_as!(
39 DbVersion,
40 r#"
41 INSERT INTO versions (item_id, version_number, changelog, file_url, file_size_bytes, file_name, is_current, label)
42 VALUES ($1, $2, $3, $4, $5, $6, true, $7)
43 RETURNING id AS "id: VersionId", item_id AS "item_id: ItemId", version_number, changelog,
44 file_url, file_size_bytes, file_name, download_count, is_current,
45 created_at AS "created_at: chrono::DateTime<chrono::Utc>", s3_key,
46 scan_status AS "scan_status: super::FileScanStatus", label
47 "#,
48 item_id as ItemId,
49 version_number,
50 changelog,
51 file_url,
52 file_size_bytes,
53 file_name,
54 label,
55 )
56 .fetch_one(&mut *tx)
57 .await?;
58
59 tx.commit().await?;
60
61 Ok(version)
62 }
63
64 /// Hard cap on rows returned by the non-paginated version listing. Items
65 /// with this many versions are exceptional; if we ever hit the cap a warning
66 /// fires so we can promote the caller to cursor pagination. Real pagination
67 /// is deferred (Phase 6/8), this constant just makes the truncation loud.
68 pub const VERSIONS_LIST_HARD_CAP: i64 = 5000;
69
70 /// List all versions for an item, newest first.
71 ///
72 /// Capped at `VERSIONS_LIST_HARD_CAP` as a safety limit. Hitting the cap is
73 /// logged at WARN so we notice before a real user gets silently truncated.
74 #[tracing::instrument(skip_all)]
75 pub async fn get_versions_by_item(pool: &PgPool, item_id: ItemId) -> Result<Vec<DbVersion>> {
76 let versions = sqlx::query_as!(
77 DbVersion,
78 r#"
79 SELECT id AS "id: VersionId", item_id AS "item_id: ItemId", version_number, changelog,
80 file_url, file_size_bytes, file_name, download_count, is_current,
81 created_at AS "created_at: chrono::DateTime<chrono::Utc>", s3_key,
82 scan_status AS "scan_status: super::FileScanStatus", label
83 FROM versions WHERE item_id = $1 ORDER BY created_at DESC LIMIT $2
84 "#,
85 item_id as ItemId,
86 VERSIONS_LIST_HARD_CAP,
87 )
88 .fetch_all(pool)
89 .await?;
90
91 if versions.len() as i64 == VERSIONS_LIST_HARD_CAP {
92 tracing::warn!(
93 %item_id, cap = VERSIONS_LIST_HARD_CAP,
94 "get_versions_by_item hit hard cap; promote caller to cursor pagination"
95 );
96 }
97
98 Ok(versions)
99 }
100
101 /// Batch-load versions for multiple items, grouped by item_id.
102 #[tracing::instrument(skip_all)]
103 pub async fn get_versions_by_items(
104 pool: &PgPool,
105 item_ids: &[ItemId],
106 ) -> Result<std::collections::HashMap<ItemId, Vec<DbVersion>>> {
107 let versions = sqlx::query_as!(
108 DbVersion,
109 r#"
110 SELECT id AS "id: VersionId", item_id AS "item_id: ItemId", version_number, changelog,
111 file_url, file_size_bytes, file_name, download_count, is_current,
112 created_at AS "created_at: chrono::DateTime<chrono::Utc>", s3_key,
113 scan_status AS "scan_status: super::FileScanStatus", label
114 FROM versions WHERE item_id = ANY($1) ORDER BY item_id, created_at DESC
115 "#,
116 item_ids as &[ItemId],
117 )
118 .fetch_all(pool)
119 .await?;
120
121 let mut map: std::collections::HashMap<ItemId, Vec<DbVersion>> =
122 std::collections::HashMap::new();
123 for v in versions {
124 map.entry(v.item_id).or_default().push(v);
125 }
126 Ok(map)
127 }
128
129 /// Atomically increment the download counter for a version.
130 #[tracing::instrument(skip_all)]
131 pub async fn increment_download_count(pool: &PgPool, version_id: VersionId) -> Result<()> {
132 sqlx::query!(
133 "UPDATE versions SET download_count = download_count + 1 WHERE id = $1",
134 version_id as VersionId,
135 )
136 .execute(pool)
137 .await?;
138
139 Ok(())
140 }
141
142 /// Record that a user downloaded a specific version (idempotent).
143 #[tracing::instrument(skip_all)]
144 pub async fn record_user_download(
145 pool: &PgPool,
146 user_id: UserId,
147 item_id: ItemId,
148 version_id: VersionId,
149 ) -> Result<()> {
150 // Explicit conflict target, the table's PRIMARY KEY is
151 // (user_id, item_id, version_id), but `ON CONFLICT DO NOTHING` without
152 // a target would silently swallow conflicts on ANY future constraint
153 // we add (a unique index on downloaded_at, say). Naming the target
154 // means a new constraint surfaces as an error rather than a no-op.
155 sqlx::query!(
156 r#"
157 INSERT INTO user_downloads (user_id, item_id, version_id)
158 VALUES ($1, $2, $3)
159 ON CONFLICT (user_id, item_id, version_id) DO NOTHING
160 "#,
161 user_id as UserId,
162 item_id as ItemId,
163 version_id as VersionId,
164 )
165 .execute(pool)
166 .await?;
167
168 Ok(())
169 }
170
171 /// Get the latest version ID a user has downloaded for an item, if any.
172 #[tracing::instrument(skip_all)]
173 pub async fn get_user_latest_download(
174 pool: &PgPool,
175 user_id: UserId,
176 item_id: ItemId,
177 ) -> Result<Option<VersionId>> {
178 let row = sqlx::query_scalar!(
179 r#"
180 SELECT ud.version_id AS "version_id: VersionId" FROM user_downloads ud
181 JOIN versions v ON v.id = ud.version_id
182 WHERE ud.user_id = $1 AND ud.item_id = $2
183 ORDER BY v.created_at DESC
184 LIMIT 1
185 "#,
186 user_id as UserId,
187 item_id as ItemId,
188 )
189 .fetch_optional(pool)
190 .await?;
191
192 Ok(row)
193 }
194
195 /// Fetch a version by primary key. Returns `None` if not found.
196 #[tracing::instrument(skip_all)]
197 pub async fn get_version_by_id(pool: &PgPool, version_id: VersionId) -> Result<Option<DbVersion>> {
198 let version = sqlx::query_as!(
199 DbVersion,
200 r#"
201 SELECT id AS "id: VersionId", item_id AS "item_id: ItemId", version_number, changelog,
202 file_url, file_size_bytes, file_name, download_count, is_current,
203 created_at AS "created_at: chrono::DateTime<chrono::Utc>", s3_key,
204 scan_status AS "scan_status: super::FileScanStatus", label
205 FROM versions WHERE id = $1
206 "#,
207 version_id as VersionId,
208 )
209 .fetch_optional(pool)
210 .await?;
211
212 Ok(version)
213 }
214
215 /// Collect all S3 keys for versions owned by a user.
216 ///
217 /// Returns s3_key, file_name, version_number, item title, and project slug.
218 /// Only includes versions that have an S3 key.
219 #[tracing::instrument(skip_all)]
220 pub async fn get_user_version_s3_keys(
221 pool: &PgPool,
222 user_id: super::UserId,
223 ) -> Result<Vec<VersionS3KeyRow>> {
224 let rows = sqlx::query_as!(
225 VersionS3KeyRow,
226 r#"
227 SELECT v.s3_key, v.file_name, v.version_number AS "version_number!", i.title AS "item_title!",
228 p.id AS "project_id!: super::ProjectId", p.slug AS "project_slug!: super::Slug", v.file_size_bytes
229 FROM versions v
230 JOIN items i ON v.item_id = i.id
231 JOIN projects p ON i.project_id = p.id
232 WHERE p.user_id = $1 AND v.s3_key IS NOT NULL
233 ORDER BY p.slug, i.sort_order, v.created_at DESC
234 LIMIT $2
235 "#,
236 user_id as UserId,
237 VERSIONS_LIST_HARD_CAP,
238 )
239 .fetch_all(pool)
240 .await?;
241
242 if rows.len() as i64 == VERSIONS_LIST_HARD_CAP {
243 tracing::warn!(
244 %user_id, cap = VERSIONS_LIST_HARD_CAP,
245 "get_user_version_s3_keys (account export) hit hard cap; some files will be omitted from export"
246 );
247 }
248
249 Ok(rows)
250 }
251
252 /// Update a version's S3 key, file size, and file name in one query.
253 ///
254 /// `expected_old_s3_key` guards against a lost-update race: two concurrent
255 /// confirms can both pass the idempotency gate (reading the same prior
256 /// `version.s3_key`) and both succeed in incrementing storage; without this
257 /// guard, the second's UPDATE silently overwrites the first's, leaking S3
258 /// objects and double-charging storage.
259 ///
260 /// `Ok(None)` means the row exists but the `s3_key` no longer matches the
261 /// expected value, the caller is responsible for the rollback (refund the
262 /// storage increment, delete the new S3 object).
263 #[tracing::instrument(skip_all)]
264 pub async fn update_version_file<'e>(
265 executor: impl sqlx::PgExecutor<'e>,
266 version_id: VersionId,
267 expected_old_s3_key: Option<&str>,
268 s3_key: &str,
269 file_size_bytes: Option<i64>,
270 file_name: Option<&str>,
271 ) -> Result<Option<DbVersion>> {
272 let version = sqlx::query_as!(
273 DbVersion,
274 r#"
275 UPDATE versions
276 SET s3_key = $2, file_size_bytes = $3, file_name = $4
277 WHERE id = $1
278 AND s3_key IS NOT DISTINCT FROM $5
279 RETURNING id AS "id: VersionId", item_id AS "item_id: ItemId", version_number, changelog,
280 file_url, file_size_bytes, file_name, download_count, is_current,
281 created_at AS "created_at: chrono::DateTime<chrono::Utc>", s3_key,
282 scan_status AS "scan_status: super::FileScanStatus", label
283 "#,
284 version_id as VersionId,
285 s3_key,
286 file_size_bytes,
287 file_name,
288 expected_old_s3_key,
289 )
290 .fetch_optional(executor)
291 .await?;
292
293 Ok(version)
294 }
295
296 /// Delete a version by ID, decrementing the owning user's storage counter
297 /// and enqueuing its S3 object for durable deletion in the same transaction.
298 ///
299 /// Both the storage refund and the S3-delete enqueue must succeed together
300 /// with the row delete, otherwise a future caller of this function could
301 /// forget either step and leak storage credit or orphan an S3 object.
302 /// `delete_version_row_only` exists for cases where the caller has already
303 /// handled both side effects (e.g. cascading item delete that batches them).
304 #[tracing::instrument(skip_all)]
305 pub async fn delete_version(pool: &PgPool, version_id: VersionId) -> Result<()> {
306 // Look up the owning user so we can refund storage. Ownership is stable for
307 // a version's lifetime; its size + key, by contrast, can change under a
308 // concurrent replace-confirm, so those come from the DELETE's RETURNING
309 // below, never a pre-tx read (Run #18 Storage B5).
310 let owner_id: Option<super::UserId> = sqlx::query_scalar!(
311 r#"
312 SELECT p.user_id AS "user_id!: super::UserId"
313 FROM versions v
314 JOIN items i ON v.item_id = i.id
315 JOIN projects p ON i.project_id = p.id
316 WHERE v.id = $1
317 "#,
318 version_id as VersionId,
319 )
320 .fetch_optional(pool)
321 .await?;
322
323 let mut tx = pool.begin().await?;
324
325 // DELETE ... RETURNING so the refund + S3 enqueue act on the row's ACTUAL
326 // state at delete time. A replace-confirm that commits between a pre-tx read
327 // and this DELETE would otherwise make us refund the OLD size and enqueue
328 // the OLD key while leaking the new one. RETURNING also gives us the
329 // rows-affected discipline for free: a concurrent double-delete finds no row
330 // and refunds nothing (Run #12 LOW + Run #18 Storage B5).
331 let deleted: Option<(Option<String>, Option<i64>)> = sqlx::query!(
332 "DELETE FROM versions WHERE id = $1 RETURNING s3_key, file_size_bytes",
333 version_id as VersionId,
334 )
335 .fetch_optional(&mut *tx)
336 .await?
337 .map(|r| (r.s3_key, r.file_size_bytes));
338
339 if let Some((s3_key, file_size_bytes)) = deleted {
340 if let Some(user_id) = owner_id
341 && let Some(size) = file_size_bytes
342 && size > 0
343 {
344 crate::db::creator_tiers::decrement_storage_used(&mut *tx, user_id, size).await?;
345 }
346
347 // Enqueue the S3 delete inside the SAME tx as the row delete + refund.
348 // After commit the row is gone (so the key is non-live and the deletion
349 // worker can act), and a crash between commit and a post-commit enqueue
350 // can no longer orphan the object, all three effects are atomic, as the
351 // doc comment promises.
352 if let Some(s3_key) = s3_key {
353 crate::db::pending_s3_deletions::enqueue_deletions(
354 &mut *tx,
355 &[(s3_key, "main".to_string())],
356 "version_delete",
357 )
358 .await?;
359 }
360 }
361
362 tx.commit().await?;
363
364 Ok(())
365 }
366
367 /// Sum all version file sizes for a given item (for storage decrement on item delete).
368 #[tracing::instrument(skip_all)]
369 pub async fn sum_file_sizes_for_item(pool: &PgPool, item_id: super::ItemId) -> Result<i64> {
370 // SUM over many bigints widens to NUMERIC in Postgres; clamp on both
371 // sides (>=0 and <=i64::MAX) before casting back to BIGINT, without
372 // GREATEST(0, ...), a corrupt-negative row could propagate a negative
373 // total that later under-flows storage accounting.
374 let total: i64 = sqlx::query_scalar!(
375 r#"SELECT COALESCE(GREATEST(0, LEAST(SUM(file_size_bytes), 9223372036854775807))::BIGINT, 0) AS "total!" FROM versions WHERE item_id = $1 AND file_size_bytes IS NOT NULL"#,
376 item_id as ItemId,
377 )
378 .fetch_one(pool)
379 .await?;
380
381 Ok(total)
382 }
383