Skip to main content

max / makenotwork

37.4 KB · 1190 lines History Blame Raw
1 //! Item CRUD: creation, listing, text body updates, and ownership lookups.
2 //!
3 //! Bulk and structural operations (move, bulk_*, duplicate) live in the
4 //! `bulk` submodule and are re-exported flat so call sites still see
5 //! `db::items::bulk_publish` etc.
6
7 mod bulk;
8 mod media;
9
10 pub use bulk::*;
11 pub use media::*;
12
13 use sqlx::PgPool;
14
15 use super::enums::{AiTier, ItemType};
16 use super::models::{DbItem, ItemS3KeyRow};
17 use super::{ItemId, MtThreadId, PriceCents, ProjectId, UserId};
18 use crate::error::Result;
19
20 /// Per-viewer display flags for an item's store page (wishlisted / in-cart /
21 /// number of the viewer's collections containing it). Display-only, none of
22 /// these gate access.
23 #[derive(Debug, Default)]
24 pub struct ViewerItemFlags {
25 pub is_wishlisted: bool,
26 pub in_cart: bool,
27 pub collection_count: i64,
28 }
29
30 /// Fetch a viewer's display flags for an item in ONE query instead of three
31 /// sequential round-trips (wishlist EXISTS, cart EXISTS, collections COUNT).
32 /// Fewer round-trips AND fewer pooled connections per item-page render, the
33 /// collapse the public item page wants (vs. fanning the three out concurrently,
34 /// which would add connection pressure on a hot public path).
35 #[tracing::instrument(skip_all)]
36 pub async fn get_viewer_item_flags(
37 pool: &PgPool,
38 user_id: UserId,
39 item_id: ItemId,
40 ) -> Result<ViewerItemFlags> {
41 let (is_wishlisted, in_cart, collection_count) = sqlx::query_as::<_, (bool, bool, i64)>(
42 r"
43 SELECT
44 EXISTS(SELECT 1 FROM wishlists WHERE user_id = $1 AND item_id = $2),
45 EXISTS(SELECT 1 FROM cart_items WHERE user_id = $1 AND item_id = $2),
46 (SELECT COUNT(*) FROM collections c
47 JOIN collection_items ci ON ci.collection_id = c.id
48 WHERE c.user_id = $1 AND ci.item_id = $2)
49 ",
50 )
51 .bind(user_id)
52 .bind(item_id)
53 .fetch_one(pool)
54 .await?;
55
56 Ok(ViewerItemFlags {
57 is_wishlisted,
58 in_cart,
59 collection_count,
60 })
61 }
62
63 /// Insert a new item into a project and return the created row.
64 ///
65 /// Auto-generates a URL-safe slug from the title. If the slug collides with
66 /// an existing item in the same project, appends a counter suffix.
67 #[allow(clippy::too_many_arguments)]
68 #[tracing::instrument(skip_all)]
69 /// Find an existing untitled, unpublished wizard draft for a project + type.
70 ///
71 /// The single-type-card item wizard used to `create_item` a fresh "Untitled"
72 /// row on every GET, so a prefetch or a re-visit piled up orphan drafts. The
73 /// wizard now reuses the most recent such draft (newest first) instead of
74 /// minting another. A draft the creator actually names/publishes no
75 /// longer matches, so real items are never recycled.
76 #[tracing::instrument(skip_all)]
77 pub async fn find_untitled_wizard_draft(
78 pool: &PgPool,
79 project_id: ProjectId,
80 item_type: ItemType,
81 ) -> Result<Option<DbItem>> {
82 let item = sqlx::query_as::<_, DbItem>(
83 r"
84 SELECT * FROM items
85 WHERE project_id = $1 AND item_type = $2
86 AND title = 'Untitled' AND is_public = false AND deleted_at IS NULL
87 ORDER BY created_at DESC
88 LIMIT 1
89 ",
90 )
91 .bind(project_id)
92 .bind(item_type)
93 .fetch_optional(pool)
94 .await?;
95 Ok(item)
96 }
97
98 #[allow(clippy::too_many_arguments)]
99 pub async fn create_item(
100 pool: &PgPool,
101 project_id: ProjectId,
102 title: &str,
103 description: Option<&str>,
104 price_cents: PriceCents,
105 item_type: ItemType,
106 ai_tier: AiTier,
107 ai_disclosure: Option<&str>,
108 ) -> Result<DbItem> {
109 // Slug uniqueness is enforced by the per-project unique index: insert the
110 // bare slug, and on a 23505 collision append an incrementing suffix and
111 // retry (`base`, `base-2`, `base-3`, ...). The retry IS the dedup, a
112 // pre-insert `SELECT EXISTS` check would just be a redundant round-trip on
113 // the common (no-collision) path and a TOCTOU on the racy one.
114 let base = crate::helpers::slugify(title).to_string();
115 crate::helpers::insert_with_unique_slug(&base, |slug| async move {
116 let slug = super::validated_types::Slug::from_trusted(slug);
117 sqlx::query_as::<_, DbItem>(
118 r"
119 INSERT INTO items (project_id, title, description, price_cents, item_type, slug, ai_tier, ai_disclosure)
120 VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
121 RETURNING *
122 ",
123 )
124 .bind(project_id)
125 .bind(title)
126 .bind(description)
127 .bind(price_cents)
128 .bind(item_type)
129 .bind(&slug)
130 .bind(ai_tier)
131 .bind(ai_disclosure)
132 .fetch_one(pool)
133 .await
134 .map_err(Into::into)
135 })
136 .await
137 }
138
139 /// Fetch an item by primary key. Returns `None` if not found.
140 #[tracing::instrument(skip_all)]
141 pub async fn get_item_by_id(pool: &PgPool, id: ItemId) -> Result<Option<DbItem>> {
142 let item = sqlx::query_as::<_, DbItem>("SELECT * FROM items WHERE id = $1")
143 .bind(id)
144 .fetch_optional(pool)
145 .await?;
146
147 Ok(item)
148 }
149
150 /// Fetch multiple public items by id in one query. Replaces a per-id
151 /// `get_item_by_id` loop (N+1) where the caller only wants the public rows.
152 /// Order is not guaranteed.
153 #[tracing::instrument(skip_all)]
154 pub async fn get_public_items_by_ids(pool: &PgPool, ids: &[ItemId]) -> Result<Vec<DbItem>> {
155 if ids.is_empty() {
156 return Ok(Vec::new());
157 }
158 let items =
159 sqlx::query_as::<_, DbItem>("SELECT * FROM items WHERE id = ANY($1) AND is_public = true")
160 .bind(ids)
161 .fetch_all(pool)
162 .await?;
163
164 Ok(items)
165 }
166
167 /// Fetch titles for a batch of item IDs. Returns (item_id, title) pairs.
168 #[tracing::instrument(skip_all)]
169 pub async fn get_item_titles_batch(pool: &PgPool, ids: &[ItemId]) -> Result<Vec<(ItemId, String)>> {
170 if ids.is_empty() {
171 return Ok(vec![]);
172 }
173 let rows: Vec<(ItemId, String)> =
174 sqlx::query_as("SELECT id, title FROM items WHERE id = ANY($1)")
175 .bind(ids)
176 .fetch_all(pool)
177 .await?;
178
179 Ok(rows)
180 }
181
182 /// Fetch project_id for a batch of item IDs. Returns (item_id, project_id) pairs.
183 ///
184 /// Used by bulk operations to verify all items belong to the same project in one query.
185 #[tracing::instrument(skip_all)]
186 pub async fn get_item_project_ids_batch(
187 pool: &PgPool,
188 ids: &[ItemId],
189 ) -> Result<Vec<(ItemId, super::ProjectId)>> {
190 if ids.is_empty() {
191 return Ok(vec![]);
192 }
193 let rows: Vec<(ItemId, super::ProjectId)> =
194 sqlx::query_as("SELECT id, project_id FROM items WHERE id = ANY($1)")
195 .bind(ids)
196 .fetch_all(pool)
197 .await?;
198
199 Ok(rows)
200 }
201
202 /// List all items in a project, ordered by sort_order then newest.
203 ///
204 /// Capped at 500 as a safety limit.
205 #[tracing::instrument(skip_all)]
206 pub async fn get_items_by_project(pool: &PgPool, project_id: ProjectId) -> Result<Vec<DbItem>> {
207 let items = sqlx::query_as::<_, DbItem>(
208 // No LIMIT: one project's live item set is naturally bounded; the old
209 // flat cap silently truncated dashboard/exports (audit Run 17 Perf).
210 "SELECT * FROM items WHERE project_id = $1 AND deleted_at IS NULL ORDER BY sort_order, created_at DESC",
211 )
212 .bind(project_id)
213 .fetch_all(pool)
214 .await?;
215
216 Ok(items)
217 }
218
219 /// List all items across all projects owned by a user, newest first.
220 ///
221 /// Capped at 500 as a safety limit.
222 #[tracing::instrument(skip_all)]
223 pub async fn get_items_by_user(pool: &PgPool, user_id: UserId) -> Result<Vec<DbItem>> {
224 /// Page size for the accumulating fetch.
225 const PAGE: i64 = 1_000;
226 /// Hard cap so even a huge catalog can't load an unbounded result set.
227 const MAX_ITEMS: usize = 100_000;
228
229 // Previously a flat `LIMIT 500` silently truncated the result, a creator with
230 // more than 500 items exported only their newest 500 with no indication
231 // (data loss in the export). Page through instead, releasing the connection
232 // between pages, up to a sane cap (Perf, Run 9).
233 let mut all = Vec::new();
234 let mut offset = 0i64;
235 loop {
236 let page = sqlx::query_as::<_, DbItem>(
237 r"
238 SELECT i.* FROM items i
239 JOIN projects p ON i.project_id = p.id
240 WHERE p.user_id = $1 AND i.deleted_at IS NULL
241 ORDER BY i.created_at DESC
242 LIMIT $2 OFFSET $3
243 ",
244 )
245 .bind(user_id)
246 .bind(PAGE)
247 .bind(offset)
248 .fetch_all(pool)
249 .await?;
250 let n = page.len();
251 all.extend(page);
252 offset += n as i64;
253 if (n as i64) < PAGE || all.len() >= MAX_ITEMS {
254 break;
255 }
256 }
257
258 Ok(all)
259 }
260
261 /// Count a user's (non-deleted) items without materializing the rows. For
262 /// callers that only need the total, this avoids fetching up to 500 full rows
263 /// just to `.len()` them.
264 #[tracing::instrument(skip_all)]
265 pub async fn count_items_by_user(pool: &PgPool, user_id: UserId) -> Result<i64> {
266 let count: i64 = sqlx::query_scalar(
267 r"
268 SELECT COUNT(*) FROM items i
269 JOIN projects p ON i.project_id = p.id
270 WHERE p.user_id = $1 AND i.deleted_at IS NULL
271 ",
272 )
273 .bind(user_id)
274 .fetch_one(pool)
275 .await?;
276 Ok(count)
277 }
278
279 /// Count items per project for all projects owned by a user.
280 ///
281 /// Returns `(project_id, count)` tuples. Used by the CLI to avoid N+1 queries.
282 #[tracing::instrument(skip_all)]
283 pub async fn count_items_by_user_projects(
284 pool: &PgPool,
285 user_id: UserId,
286 ) -> Result<Vec<(ProjectId, i64)>> {
287 let rows: Vec<(ProjectId, i64)> = sqlx::query_as(
288 r"
289 SELECT i.project_id, COUNT(*) AS cnt
290 FROM items i
291 JOIN projects p ON i.project_id = p.id
292 WHERE p.user_id = $1 AND i.deleted_at IS NULL
293 GROUP BY i.project_id
294 ",
295 )
296 .bind(user_id)
297 .fetch_all(pool)
298 .await?;
299
300 Ok(rows)
301 }
302
303 /// List only public items in a project, ordered by sort_order then newest.
304 ///
305 /// Capped at 500 as a safety limit.
306 #[tracing::instrument(skip_all)]
307 pub async fn get_public_items_by_project(
308 pool: &PgPool,
309 project_id: ProjectId,
310 ) -> Result<Vec<DbItem>> {
311 let items = sqlx::query_as::<_, DbItem>(
312 "SELECT * FROM items WHERE project_id = $1 AND is_public = true AND listed = true ORDER BY sort_order, created_at DESC",
313 )
314 .bind(project_id)
315 .fetch_all(pool)
316 .await?;
317
318 Ok(items)
319 }
320
321 /// Partially update an item's fields (COALESCE keeps existing values when `None`).
322 ///
323 /// `publish_at` uses a double-Option: `None` = no change, `Some(None)` = clear schedule,
324 /// `Some(Some(dt))` = set schedule.
325 #[allow(clippy::too_many_arguments)]
326 #[tracing::instrument(skip_all)]
327 pub async fn update_item(
328 pool: &PgPool,
329 id: ItemId,
330 user_id: UserId,
331 title: Option<&str>,
332 description: Option<&str>,
333 price_cents: Option<PriceCents>,
334 item_type: Option<ItemType>,
335 is_public: Option<bool>,
336 pwyw_enabled: Option<bool>,
337 pwyw_min_cents: Option<PriceCents>,
338 publish_at: Option<Option<chrono::DateTime<chrono::Utc>>>,
339 web_only: Option<bool>,
340 ai_tier: Option<AiTier>,
341 ai_disclosure: Option<Option<&str>>,
342 ) -> Result<DbItem> {
343 // Flatten the double-Option: if outer is None, pass current DB value (via SQL CASE).
344 // $10 = whether to update publish_at, $11 = the new value (NULL to clear).
345 let update_publish_at = publish_at.is_some();
346 let publish_at_value = publish_at.flatten();
347
348 // ai_disclosure uses the same double-Option pattern as publish_at:
349 // None = no change, Some(None) = clear, Some(Some(text)) = set.
350 let update_ai_disclosure = ai_disclosure.is_some();
351 let ai_disclosure_value = ai_disclosure.flatten();
352
353 let item = sqlx::query_as::<_, DbItem>(
354 r"
355 UPDATE items
356 SET title = COALESCE($3, title),
357 description = COALESCE($4, description),
358 price_cents = COALESCE($5, price_cents),
359 item_type = COALESCE($6, item_type),
360 is_public = CASE WHEN removed_by_admin AND $7 = true THEN false ELSE COALESCE($7, is_public) END,
361 pwyw_enabled = COALESCE($8, pwyw_enabled),
362 pwyw_min_cents = COALESCE($9, pwyw_min_cents),
363 publish_at = CASE WHEN $10 THEN $11 ELSE publish_at END,
364 web_only = COALESCE($12, web_only),
365 ai_tier = COALESCE($13, ai_tier),
366 ai_disclosure = CASE WHEN $14 THEN $15 ELSE ai_disclosure END
367 WHERE id = $1
368 AND project_id IN (SELECT id FROM projects WHERE user_id = $2)
369 RETURNING *
370 ",
371 )
372 .bind(id)
373 .bind(user_id)
374 .bind(title)
375 .bind(description)
376 .bind(price_cents)
377 .bind(item_type)
378 .bind(is_public)
379 .bind(pwyw_enabled)
380 .bind(pwyw_min_cents)
381 .bind(update_publish_at)
382 .bind(publish_at_value)
383 .bind(web_only)
384 .bind(ai_tier)
385 .bind(update_ai_disclosure)
386 .bind(ai_disclosure_value)
387 .fetch_one(pool)
388 .await?;
389
390 Ok(item)
391 }
392
393 /// Publish all items whose scheduled publish time has passed.
394 ///
395 /// Atomically sets `is_public = true` and clears `publish_at`, returning the
396 /// newly published items so the caller can send release announcements.
397 #[tracing::instrument(skip_all)]
398 pub async fn publish_scheduled_items(pool: &PgPool) -> Result<Vec<DbItem>> {
399 let items = sqlx::query_as::<_, DbItem>(
400 r"
401 UPDATE items
402 SET is_public = true, publish_at = NULL, updated_at = NOW()
403 WHERE publish_at IS NOT NULL AND publish_at <= NOW() AND is_public = false AND removed_by_admin = false
404 RETURNING *
405 ",
406 )
407 .fetch_all(pool)
408 .await?;
409
410 Ok(items)
411 }
412
413 /// Soft-delete an item (sets deleted_at, recoverable for 7 days).
414 #[tracing::instrument(skip_all)]
415 pub async fn delete_item(pool: &PgPool, id: ItemId, user_id: UserId) -> Result<()> {
416 sqlx::query(
417 r"
418 UPDATE items SET deleted_at = NOW(), is_public = false
419 WHERE id = $1 AND deleted_at IS NULL
420 AND project_id IN (SELECT id FROM projects WHERE user_id = $2)
421 ",
422 )
423 .bind(id)
424 .bind(user_id)
425 .execute(pool)
426 .await?;
427
428 Ok(())
429 }
430
431 /// Restore a soft-deleted item.
432 #[tracing::instrument(skip_all)]
433 pub async fn restore_item(pool: &PgPool, id: ItemId, user_id: UserId) -> Result<bool> {
434 let result = sqlx::query(
435 r"
436 UPDATE items SET deleted_at = NULL
437 WHERE id = $1 AND deleted_at IS NOT NULL
438 AND project_id IN (SELECT id FROM projects WHERE user_id = $2)
439 ",
440 )
441 .bind(id)
442 .bind(user_id)
443 .execute(pool)
444 .await?;
445
446 Ok(result.rows_affected() > 0)
447 }
448
449 /// Get soft-deleted items for a project (for the "Recently Deleted" section).
450 #[tracing::instrument(skip_all)]
451 pub async fn get_deleted_items_by_project(
452 pool: &PgPool,
453 project_id: ProjectId,
454 ) -> Result<Vec<DbItem>> {
455 let items = sqlx::query_as::<_, DbItem>(
456 "SELECT * FROM items WHERE project_id = $1 AND deleted_at IS NOT NULL ORDER BY deleted_at DESC",
457 )
458 .bind(project_id)
459 .fetch_all(pool)
460 .await?;
461
462 Ok(items)
463 }
464
465 /// Collect S3 keys from items that are about to be purged (soft-deleted >7 days).
466 /// Returns all non-null S3 keys (audio, cover, video) so they can be deleted
467 /// from S3 before the DB rows are removed.
468 #[tracing::instrument(skip_all)]
469 pub async fn get_expired_deleted_item_s3_keys(pool: &PgPool) -> Result<Vec<String>> {
470 let keys: Vec<String> = sqlx::query_scalar(
471 r"
472 SELECT k FROM (
473 SELECT unnest(ARRAY_REMOVE(ARRAY[audio_s3_key, cover_s3_key, video_s3_key], NULL)) AS k
474 FROM items
475 WHERE deleted_at IS NOT NULL AND deleted_at < NOW() - INTERVAL '7 days'
476 AND (audio_s3_key IS NOT NULL OR cover_s3_key IS NOT NULL OR video_s3_key IS NOT NULL)
477 ) sub
478 ",
479 )
480 .fetch_all(pool)
481 .await?;
482
483 Ok(keys)
484 }
485
486 /// Collect S3 keys from versions belonging to items about to be purged.
487 /// Must be called before purge since CASCADE delete destroys version rows.
488 #[tracing::instrument(skip_all)]
489 pub async fn get_expired_deleted_item_version_s3_keys(pool: &PgPool) -> Result<Vec<String>> {
490 let keys: Vec<String> = sqlx::query_scalar(
491 r"
492 SELECT v.s3_key
493 FROM versions v
494 JOIN items i ON v.item_id = i.id
495 WHERE i.deleted_at IS NOT NULL AND i.deleted_at < NOW() - INTERVAL '7 days'
496 AND v.s3_key IS NOT NULL
497 ",
498 )
499 .fetch_all(pool)
500 .await?;
501
502 Ok(keys)
503 }
504
505 /// Sum total file sizes per user for items about to be purged, including version files.
506 /// Returns (user_id, total_bytes) pairs for storage decrement.
507 #[tracing::instrument(skip_all)]
508 pub async fn get_expired_deleted_item_storage_by_user<'e>(
509 executor: impl sqlx::PgExecutor<'e>,
510 ) -> Result<Vec<(super::UserId, i64)>> {
511 let rows: Vec<(super::UserId, i64)> = sqlx::query_as(
512 r"
513 SELECT p.user_id,
514 COALESCE(SUM(
515 COALESCE(i.audio_file_size_bytes, 0) +
516 COALESCE(i.cover_file_size_bytes, 0) +
517 COALESCE(i.video_file_size_bytes, 0) +
518 COALESCE(ver.version_bytes, 0) +
519 COALESCE(igal.gallery_bytes, 0)
520 ), 0)::BIGINT AS total_bytes
521 FROM items i
522 JOIN projects p ON i.project_id = p.id
523 LEFT JOIN LATERAL (
524 SELECT COALESCE(SUM(v.file_size_bytes), 0)::BIGINT AS version_bytes
525 FROM versions v
526 WHERE v.item_id = i.id AND v.file_size_bytes IS NOT NULL
527 ) ver ON true
528 LEFT JOIN LATERAL (
529 SELECT COALESCE(SUM(ii.file_size_bytes), 0)::BIGINT AS gallery_bytes
530 FROM item_images ii WHERE ii.item_id = i.id
531 ) igal ON true
532 WHERE i.deleted_at IS NOT NULL AND i.deleted_at < NOW() - INTERVAL '7 days'
533 GROUP BY p.user_id
534 ",
535 )
536 .fetch_all(executor)
537 .await?;
538
539 Ok(rows)
540 }
541
542 /// Collect all S3 keys from items belonging to a project (audio, cover, video).
543 #[tracing::instrument(skip_all)]
544 pub async fn get_project_item_s3_keys(
545 pool: &PgPool,
546 project_id: super::ProjectId,
547 ) -> Result<Vec<String>> {
548 let keys: Vec<String> = sqlx::query_scalar(
549 r"
550 SELECT k FROM (
551 SELECT unnest(ARRAY_REMOVE(ARRAY[audio_s3_key, cover_s3_key, video_s3_key], NULL)) AS k
552 FROM items
553 WHERE project_id = $1
554 AND (audio_s3_key IS NOT NULL OR cover_s3_key IS NOT NULL OR video_s3_key IS NOT NULL)
555 ) sub
556 ",
557 )
558 .bind(project_id)
559 .fetch_all(pool)
560 .await?;
561
562 Ok(keys)
563 }
564
565 /// Collect S3 keys from versions belonging to items in a project.
566 #[tracing::instrument(skip_all)]
567 pub async fn get_project_version_s3_keys(
568 pool: &PgPool,
569 project_id: super::ProjectId,
570 ) -> Result<Vec<String>> {
571 let keys: Vec<String> = sqlx::query_scalar(
572 r"
573 SELECT v.s3_key
574 FROM versions v
575 JOIN items i ON v.item_id = i.id
576 WHERE i.project_id = $1 AND v.s3_key IS NOT NULL
577 ",
578 )
579 .bind(project_id)
580 .fetch_all(pool)
581 .await?;
582
583 Ok(keys)
584 }
585
586 /// Sum total file sizes for everything in a project that charges storage:
587 /// item audio/cover/video, versions, the item and project gallery carousels,
588 /// and the project cover image. Used to refund storage on project delete, it
589 /// must cover every category the upload paths charge, or the delete under-
590 /// refunds and leaves the creator's counter inflated.
591 #[tracing::instrument(skip_all)]
592 pub async fn get_project_storage_bytes(pool: &PgPool, project_id: super::ProjectId) -> Result<i64> {
593 let total: i64 = sqlx::query_scalar(
594 r"
595 SELECT (
596 COALESCE((
597 SELECT SUM(
598 COALESCE(i.audio_file_size_bytes, 0) +
599 COALESCE(i.cover_file_size_bytes, 0) +
600 COALESCE(i.video_file_size_bytes, 0) +
601 COALESCE(ver.version_bytes, 0) +
602 COALESCE(igal.gallery_bytes, 0)
603 )::BIGINT
604 FROM items i
605 LEFT JOIN LATERAL (
606 SELECT COALESCE(SUM(v.file_size_bytes), 0)::BIGINT AS version_bytes
607 FROM versions v
608 WHERE v.item_id = i.id AND v.file_size_bytes IS NOT NULL
609 ) ver ON true
610 LEFT JOIN LATERAL (
611 SELECT COALESCE(SUM(ii.file_size_bytes), 0)::BIGINT AS gallery_bytes
612 FROM item_images ii WHERE ii.item_id = i.id
613 ) igal ON true
614 WHERE i.project_id = $1
615 ), 0)
616 + COALESCE((
617 SELECT SUM(pi.file_size_bytes)::BIGINT
618 FROM project_images pi WHERE pi.project_id = $1
619 ), 0)
620 + COALESCE((
621 SELECT p.cover_image_size_bytes FROM projects p WHERE p.id = $1
622 ), 0)
623 )::BIGINT
624 ",
625 )
626 .bind(project_id)
627 .fetch_one(pool)
628 .await?;
629
630 Ok(total)
631 }
632
633 /// Permanently delete items that were soft-deleted more than 7 days ago.
634 #[tracing::instrument(skip_all)]
635 pub async fn purge_expired_deleted_items<'e>(executor: impl sqlx::PgExecutor<'e>) -> Result<u64> {
636 let result = sqlx::query(
637 "DELETE FROM items WHERE deleted_at IS NOT NULL AND deleted_at < NOW() - INTERVAL '7 days'",
638 )
639 .execute(executor)
640 .await?;
641
642 Ok(result.rows_affected())
643 }
644
645 /// Update the audio S3 key for an item (defense-in-depth: verifies ownership)
646 #[tracing::instrument(skip_all)]
647 pub async fn update_item_audio_s3_key(
648 pool: &PgPool,
649 item_id: ItemId,
650 user_id: UserId,
651 s3_key: &str,
652 ) -> Result<DbItem> {
653 let item = sqlx::query_as::<_, DbItem>(
654 r"
655 UPDATE items
656 SET audio_s3_key = $2, updated_at = NOW()
657 WHERE id = $1
658 AND project_id IN (SELECT id FROM projects WHERE user_id = $3)
659 RETURNING *
660 ",
661 )
662 .bind(item_id)
663 .bind(s3_key)
664 .bind(user_id)
665 .fetch_one(pool)
666 .await?;
667
668 Ok(item)
669 }
670
671 /// Update the cover image S3 key for an item (defense-in-depth: verifies ownership)
672 #[tracing::instrument(skip_all)]
673 pub async fn update_item_cover_s3_key(
674 pool: &PgPool,
675 item_id: ItemId,
676 user_id: UserId,
677 s3_key: &str,
678 ) -> Result<DbItem> {
679 let item = sqlx::query_as::<_, DbItem>(
680 r"
681 UPDATE items
682 SET cover_s3_key = $2, updated_at = NOW()
683 WHERE id = $1
684 AND project_id IN (SELECT id FROM projects WHERE user_id = $3)
685 RETURNING *
686 ",
687 )
688 .bind(item_id)
689 .bind(s3_key)
690 .bind(user_id)
691 .fetch_one(pool)
692 .await?;
693
694 Ok(item)
695 }
696
697 /// List all public items across all public projects owned by a user, newest first.
698 ///
699 /// Used by the creator RSS feed to avoid O(projects) queries.
700 /// Capped at 50 items (standard RSS feed size).
701 #[tracing::instrument(skip_all)]
702 pub async fn get_public_items_by_user(pool: &PgPool, user_id: UserId) -> Result<Vec<DbItem>> {
703 let items = sqlx::query_as::<_, DbItem>(
704 r"
705 SELECT i.* FROM items i
706 JOIN projects p ON i.project_id = p.id
707 WHERE p.user_id = $1 AND p.is_public = true AND i.is_public = true AND i.listed = true
708 ORDER BY i.created_at DESC
709 LIMIT 50
710 ",
711 )
712 .bind(user_id)
713 .fetch_all(pool)
714 .await?;
715
716 Ok(items)
717 }
718
719 /// Get the owner (user_id) of an item through its project
720 #[tracing::instrument(skip_all)]
721 pub async fn get_item_owner(pool: &PgPool, item_id: ItemId) -> Result<Option<UserId>> {
722 let owner: Option<UserId> = sqlx::query_scalar(
723 r"
724 SELECT p.user_id
725 FROM items i
726 JOIN projects p ON i.project_id = p.id
727 WHERE i.id = $1
728 ",
729 )
730 .bind(item_id)
731 .fetch_optional(pool)
732 .await?;
733
734 Ok(owner)
735 }
736
737 /// Pre-flight access check for content streaming/download. Returns item data
738 /// plus ownership, purchase, subscription, and bundle access.
739 /// Returns `None` if the item does not exist (or is soft-deleted).
740 #[derive(Debug)]
741 pub struct ItemAccessCheck {
742 // Ownership
743 pub owner_id: UserId,
744 // Access flags (only meaningful when a user_id is provided)
745 pub has_purchased: bool,
746 /// Subscription access as a witness, not a bool: `Some` only when the sealed
747 /// gate confirmed an active, in-period subscription. Feed it straight into
748 /// `pricing::AccessContext::subscription`.
749 pub subscription: Option<crate::db::subscriptions::SubscriptionGate>,
750 pub has_bundle_access: bool,
751 }
752
753 /// Ownership + the access flags that are pure transaction/bundle lookups.
754 /// `has_subscription` is deliberately NOT here: it must come from
755 /// [`crate::db::subscriptions::has_access`], the single sealed access gate, so
756 /// the grant predicate (incl. the `current_period_end` clause) cannot be
757 /// hand-written and drift: an inlined predicate drops the period clause on the
758 /// download path.
759 #[derive(Debug, sqlx::FromRow)]
760 struct ItemAccessPartial {
761 owner_id: UserId,
762 has_purchased: bool,
763 has_bundle_access: bool,
764 }
765
766 #[tracing::instrument(skip_all)]
767 pub async fn check_item_access(
768 pool: &PgPool,
769 item_id: ItemId,
770 user_id: Option<UserId>,
771 ) -> Result<Option<ItemAccessCheck>> {
772 // When no user is provided, access flags are all false
773 let uid = user_id.unwrap_or(UserId::nil());
774 let partial = sqlx::query_as::<_, ItemAccessPartial>(
775 r"
776 SELECT
777 p.user_id AS owner_id,
778 EXISTS(
779 SELECT 1 FROM transactions t
780 WHERE t.item_id = i.id AND t.buyer_id = $2 AND t.status = 'completed'
781 ) AS has_purchased,
782 EXISTS(
783 SELECT 1 FROM bundle_items bi
784 JOIN transactions bt ON bt.item_id = bi.bundle_id
785 WHERE bi.item_id = i.id AND bt.buyer_id = $2 AND bt.status = 'completed'
786 ) AS has_bundle_access
787 FROM items i
788 JOIN projects p ON i.project_id = p.id
789 WHERE i.id = $1 AND i.deleted_at IS NULL
790 ",
791 )
792 .bind(item_id)
793 .bind(uid)
794 .fetch_optional(pool)
795 .await?;
796
797 let Some(partial) = partial else {
798 return Ok(None);
799 };
800
801 // Subscription access goes through the one sealed gate, which returns a
802 // witness. Anonymous callers (no user_id) can't hold a subscription, so skip
803 // the query entirely.
804 let subscription = match user_id {
805 Some(real_uid) => {
806 crate::db::subscriptions::SubscriptionGate::check(
807 pool,
808 real_uid,
809 crate::db::subscriptions::SubscriptionScope::Item(item_id),
810 )
811 .await?
812 }
813 None => None,
814 };
815
816 Ok(Some(ItemAccessCheck {
817 owner_id: partial.owner_id,
818 has_purchased: partial.has_purchased,
819 subscription,
820 has_bundle_access: partial.has_bundle_access,
821 }))
822 }
823
824 /// Update the text body content for an item (for articles/essays).
825 ///
826 /// Recomputes `word_count` and `reading_time_minutes` on every save.
827 /// Reading time uses ~200 wpm (average adult reading speed) with floor
828 /// division, clamped to a minimum of 1 minute so the UI never shows "0 min".
829 #[tracing::instrument(skip_all)]
830 pub async fn update_item_text(
831 pool: &PgPool,
832 id: ItemId,
833 user_id: UserId,
834 body: &str,
835 ) -> Result<DbItem> {
836 let word_count = body.split_whitespace().count() as i32;
837 let reading_time = (word_count / 200).max(1);
838
839 let item = sqlx::query_as::<_, DbItem>(
840 r"
841 UPDATE items
842 SET body = $3, word_count = $4, reading_time_minutes = $5, updated_at = NOW()
843 WHERE id = $1
844 AND project_id IN (SELECT id FROM projects WHERE user_id = $2)
845 RETURNING *
846 ",
847 )
848 .bind(id)
849 .bind(user_id)
850 .bind(body)
851 .bind(word_count)
852 .bind(reading_time)
853 .fetch_one(pool)
854 .await?;
855
856 Ok(item)
857 }
858
859 /// Update the license key settings on an item.
860 #[tracing::instrument(skip_all)]
861 pub async fn update_item_license_settings(
862 pool: &PgPool,
863 item_id: ItemId,
864 user_id: UserId,
865 enable_license_keys: bool,
866 default_max_activations: Option<i32>,
867 ) -> Result<()> {
868 sqlx::query(
869 r"
870 UPDATE items
871 SET enable_license_keys = $3, default_max_activations = $4, updated_at = NOW()
872 WHERE id = $1
873 AND project_id IN (SELECT id FROM projects WHERE user_id = $2)
874 ",
875 )
876 .bind(item_id)
877 .bind(user_id)
878 .bind(enable_license_keys)
879 .bind(default_max_activations)
880 .execute(pool)
881 .await?;
882
883 Ok(())
884 }
885
886 /// Update the license text preset and optional custom text on an item.
887 #[tracing::instrument(skip_all)]
888 pub async fn update_item_license_text(
889 pool: &PgPool,
890 item_id: ItemId,
891 user_id: UserId,
892 license_preset: Option<&str>,
893 custom_license_text: Option<&str>,
894 ) -> Result<()> {
895 sqlx::query(
896 r"
897 UPDATE items
898 SET license_preset = $3, custom_license_text = $4, updated_at = NOW()
899 WHERE id = $1
900 AND project_id IN (SELECT id FROM projects WHERE user_id = $2)
901 ",
902 )
903 .bind(item_id)
904 .bind(user_id)
905 .bind(license_preset)
906 .bind(custom_license_text)
907 .execute(pool)
908 .await?;
909
910 Ok(())
911 }
912
913 /// Increment the denormalized sales_count for an item (called on purchase/claim).
914 ///
915 /// Accepts any sqlx executor (`&PgPool`, `&mut Transaction`, etc.) so callers
916 /// can include this in a larger transaction when needed.
917 #[tracing::instrument(skip_all)]
918 pub async fn increment_sales_count<'e>(
919 executor: impl sqlx::PgExecutor<'e>,
920 item_id: ItemId,
921 ) -> Result<()> {
922 sqlx::query("UPDATE items SET sales_count = sales_count + 1 WHERE id = $1")
923 .bind(item_id)
924 .execute(executor)
925 .await?;
926
927 Ok(())
928 }
929
930 /// Decrement the denormalized sales_count for an item (called on refund/unclaim).
931 #[tracing::instrument(skip_all)]
932 pub async fn decrement_sales_count<'e>(
933 executor: impl sqlx::PgExecutor<'e>,
934 item_id: ItemId,
935 ) -> Result<()> {
936 sqlx::query("UPDATE items SET sales_count = GREATEST(sales_count - 1, 0) WHERE id = $1")
937 .bind(item_id)
938 .execute(executor)
939 .await?;
940
941 Ok(())
942 }
943
944 /// Atomically mark an item as having had its release announced.
945 /// Returns false if already announced (prevents duplicate announcements on unpublish/republish).
946 #[tracing::instrument(skip_all)]
947 pub async fn mark_release_announced(pool: &PgPool, item_id: ItemId) -> Result<bool> {
948 let result = sqlx::query(
949 "UPDATE items SET release_announced_at = NOW() WHERE id = $1 AND release_announced_at IS NULL",
950 )
951 .bind(item_id)
952 .execute(pool)
953 .await?;
954
955 Ok(result.rows_affected() > 0)
956 }
957
958 /// Set the linked MT thread ID for an item.
959 #[tracing::instrument(skip_all)]
960 pub async fn set_mt_thread_id(pool: &PgPool, item_id: ItemId, thread_id: MtThreadId) -> Result<()> {
961 sqlx::query("UPDATE items SET mt_thread_id = $2 WHERE id = $1")
962 .bind(item_id)
963 .bind(thread_id)
964 .execute(pool)
965 .await?;
966 Ok(())
967 }
968
969 /// Collect all S3 keys for items owned by a user (audio + cover + video).
970 ///
971 /// Returns item title, project slug, and optional S3 keys for audio/cover/video.
972 /// Only includes items that have at least one S3 key.
973 #[tracing::instrument(skip_all)]
974 pub async fn get_user_s3_keys(pool: &PgPool, user_id: UserId) -> Result<Vec<ItemS3KeyRow>> {
975 let rows = sqlx::query_as::<_, ItemS3KeyRow>(
976 r"
977 SELECT i.title, p.id AS project_id, p.slug AS project_slug,
978 i.audio_s3_key, i.cover_s3_key, i.video_s3_key,
979 i.audio_file_size_bytes, i.cover_file_size_bytes, i.video_file_size_bytes
980 FROM items i JOIN projects p ON i.project_id = p.id
981 WHERE p.user_id = $1 AND (i.audio_s3_key IS NOT NULL OR i.cover_s3_key IS NOT NULL OR i.video_s3_key IS NOT NULL)
982 ORDER BY p.slug, i.sort_order
983 ",
984 // No LIMIT: this drives per-user S3 storage accounting + cleanup; a cap
985 // silently dropped keys for a >500-item user, orphaning their objects
986 // (audit Run 17 Perf, same silent-truncation class).
987 )
988 .bind(user_id)
989 .fetch_all(pool)
990 .await?;
991
992 Ok(rows)
993 }
994
995 /// Check whether a user has at least one public item across all their projects.
996 #[tracing::instrument(skip_all)]
997 pub async fn has_public_item_by_user(pool: &PgPool, user_id: UserId) -> Result<bool> {
998 let exists: bool = sqlx::query_scalar(
999 r"
1000 SELECT EXISTS(
1001 SELECT 1 FROM items i
1002 JOIN projects p ON i.project_id = p.id
1003 WHERE p.user_id = $1 AND i.is_public = true
1004 )
1005 ",
1006 )
1007 .bind(user_id)
1008 .fetch_one(pool)
1009 .await?;
1010
1011 Ok(exists)
1012 }
1013
1014 /// Increment the play count for an item (called on audio/video stream request).
1015 #[tracing::instrument(skip_all)]
1016 pub async fn increment_play_count(pool: &PgPool, item_id: ItemId) -> Result<()> {
1017 sqlx::query("UPDATE items SET play_count = play_count + 1 WHERE id = $1")
1018 .bind(item_id)
1019 .execute(pool)
1020 .await?;
1021
1022 Ok(())
1023 }
1024
1025 /// Record a unique listener and increment unique_play_count if this is the first
1026 /// play by this user. Returns true if new unique listener, false if already counted.
1027 /// Runs in a transaction so the user_plays INSERT and count UPDATE are atomic.
1028 #[tracing::instrument(skip_all)]
1029 pub async fn record_unique_play(
1030 pool: &PgPool,
1031 user_id: super::UserId,
1032 item_id: ItemId,
1033 ) -> Result<bool> {
1034 let mut tx = pool.begin().await?;
1035
1036 let result = sqlx::query(
1037 "INSERT INTO user_plays (user_id, item_id) VALUES ($1, $2) ON CONFLICT DO NOTHING",
1038 )
1039 .bind(user_id)
1040 .bind(item_id)
1041 .execute(&mut *tx)
1042 .await?;
1043
1044 if result.rows_affected() > 0 {
1045 sqlx::query("UPDATE items SET unique_play_count = unique_play_count + 1 WHERE id = $1")
1046 .bind(item_id)
1047 .execute(&mut *tx)
1048 .await?;
1049 tx.commit().await?;
1050 Ok(true)
1051 } else {
1052 tx.commit().await?;
1053 Ok(false)
1054 }
1055 }
1056
1057 /// Increment the item-level download count (called alongside per-version increment).
1058 #[tracing::instrument(skip_all)]
1059 pub async fn increment_item_download_count(pool: &PgPool, item_id: ItemId) -> Result<()> {
1060 sqlx::query("UPDATE items SET download_count = download_count + 1 WHERE id = $1")
1061 .bind(item_id)
1062 .execute(pool)
1063 .await?;
1064
1065 Ok(())
1066 }
1067
1068 /// Fetch a public item by project ID and slug (for custom domain routing).
1069 #[tracing::instrument(skip_all)]
1070 pub async fn get_item_by_project_and_slug(
1071 pool: &PgPool,
1072 project_id: ProjectId,
1073 slug: &str,
1074 ) -> Result<Option<DbItem>> {
1075 let item = sqlx::query_as::<_, DbItem>(
1076 "SELECT * FROM items WHERE project_id = $1 AND slug = $2 AND is_public = true",
1077 )
1078 .bind(project_id)
1079 .bind(slug)
1080 .fetch_optional(pool)
1081 .await?;
1082
1083 Ok(item)
1084 }
1085
1086 /// Hide every public item for ALL given creators in one statement, for the
1087 /// post-grace scheduler sweep, replaces N per-user UPDATEs on the lock-held tick.
1088 /// Returns the total rows affected. No-op on an empty slice.
1089 ///
1090 /// Stamps `hidden_by_suspension_at` so the matching `unhide_all_items_for_user`
1091 /// restores ONLY what this sweep hid, never a creator's genuine drafts.
1092 #[tracing::instrument(skip_all)]
1093 pub async fn hide_all_items_for_users(pool: &PgPool, user_ids: &[UserId]) -> Result<u64> {
1094 if user_ids.is_empty() {
1095 return Ok(0);
1096 }
1097 let result = sqlx::query(
1098 r"
1099 UPDATE items SET is_public = false, hidden_by_suspension_at = NOW()
1100 WHERE project_id IN (SELECT id FROM projects WHERE user_id = ANY($1))
1101 AND is_public = true
1102 ",
1103 )
1104 .bind(user_ids)
1105 .execute(pool)
1106 .await?;
1107
1108 Ok(result.rows_affected())
1109 }
1110
1111 /// Unhide items for a user that the post-grace sweep hid. Used when a creator
1112 /// re-subscribes. Returns the number of items unhidden.
1113 ///
1114 /// Restores ONLY items stamped by `hide_all_items_for_users`
1115 /// (`hidden_by_suspension_at IS NOT NULL`) and clears the stamp, a draft the
1116 /// creator never published (`is_public = false` with a NULL stamp) is left
1117 /// private, so a lapsed-then-renewed subscription can't leak unpublished work.
1118 #[tracing::instrument(skip_all)]
1119 pub async fn unhide_all_items_for_user(pool: &PgPool, user_id: UserId) -> Result<u64> {
1120 let result = sqlx::query(
1121 r"
1122 UPDATE items SET is_public = true, hidden_by_suspension_at = NULL
1123 WHERE project_id IN (SELECT id FROM projects WHERE user_id = $1)
1124 AND hidden_by_suspension_at IS NOT NULL
1125 AND removed_at IS NULL
1126 ",
1127 )
1128 .bind(user_id)
1129 .execute(pool)
1130 .await?;
1131
1132 Ok(result.rows_affected())
1133 }
1134
1135 /// Admin: remove an item (hide from public, record reason). The item stays in the DB
1136 /// and the creator can see it in their dashboard with the removal reason.
1137 #[tracing::instrument(skip_all)]
1138 pub async fn admin_remove_item(pool: &PgPool, item_id: ItemId, reason: &str) -> Result<DbItem> {
1139 let item = sqlx::query_as::<_, DbItem>(
1140 r"
1141 UPDATE items
1142 SET removed_by_admin = true,
1143 removal_reason = $2,
1144 removed_at = NOW(),
1145 is_public = false
1146 WHERE id = $1
1147 RETURNING *
1148 ",
1149 )
1150 .bind(item_id)
1151 .bind(reason)
1152 .fetch_one(pool)
1153 .await?;
1154
1155 Ok(item)
1156 }
1157
1158 /// Admin: restore a previously removed item. Clears the removal fields
1159 /// but does NOT re-publish (creator must publish manually).
1160 #[tracing::instrument(skip_all)]
1161 pub async fn admin_restore_item(pool: &PgPool, item_id: ItemId) -> Result<DbItem> {
1162 let item = sqlx::query_as::<_, DbItem>(
1163 r"
1164 UPDATE items
1165 SET removed_by_admin = false,
1166 removal_reason = NULL,
1167 removed_at = NULL
1168 WHERE id = $1
1169 RETURNING *
1170 ",
1171 )
1172 .bind(item_id)
1173 .fetch_one(pool)
1174 .await?;
1175
1176 Ok(item)
1177 }
1178
1179 /// Count public, listed items (for landing page stats).
1180 #[tracing::instrument(skip_all)]
1181 pub async fn count_public_listed(pool: &PgPool) -> Result<i64> {
1182 let count = sqlx::query_scalar::<_, i64>(
1183 "SELECT COUNT(*) FROM items WHERE is_public = true AND listed = true",
1184 )
1185 .fetch_one(pool)
1186 .await?;
1187
1188 Ok(count)
1189 }
1190