Skip to main content

max / makenotwork

9.9 KB · 330 lines History Blame Raw
1 //! Bulk and structural item operations: cross-project moves, batch
2 //! publish/unpublish/delete/repricing/tagging, and item duplication.
3
4 use sqlx::PgPool;
5
6 use crate::db::models::DbItem;
7 use crate::db::{ItemId, PriceCents, ProjectId, UserId};
8 use crate::error::Result;
9
10 pub async fn move_item(
11 pool: &PgPool,
12 project_id: ProjectId,
13 user_id: UserId,
14 item_id: ItemId,
15 direction: &str,
16 ) -> Result<()> {
17 let mut tx = pool.begin().await?;
18
19 // Lock and fetch item IDs in display order (scoped to projects owned by user)
20 let item_ids: Vec<ItemId> = sqlx::query_scalar(
21 r"
22 SELECT id FROM items
23 WHERE project_id = $1
24 AND project_id IN (SELECT id FROM projects WHERE user_id = $2)
25 ORDER BY sort_order, created_at DESC LIMIT 500 FOR UPDATE
26 ",
27 )
28 .bind(project_id)
29 .bind(user_id)
30 .fetch_all(&mut *tx)
31 .await?;
32
33 let Some(pos) = item_ids.iter().position(|id| *id == item_id) else {
34 return Ok(());
35 };
36
37 let swap_pos = match direction {
38 "up" if pos > 0 => pos - 1,
39 "down" if pos + 1 < item_ids.len() => pos + 1,
40 _ => return Ok(()),
41 };
42
43 // Normalize all sort_orders, swapping the target pair (single batch UPDATE)
44 let mut ids = Vec::with_capacity(item_ids.len());
45 let mut orders = Vec::with_capacity(item_ids.len());
46 for (i, id) in item_ids.iter().enumerate() {
47 ids.push(*id);
48 orders.push(if i == pos {
49 swap_pos as i32
50 } else if i == swap_pos {
51 pos as i32
52 } else {
53 i as i32
54 });
55 }
56 sqlx::query(
57 "UPDATE items SET sort_order = batch.ord FROM UNNEST($1::UUID[], $2::INT[]) AS batch(id, ord) WHERE items.id = batch.id",
58 )
59 .bind(&ids)
60 .bind(&orders)
61 .execute(&mut *tx)
62 .await?;
63
64 tx.commit().await?;
65 Ok(())
66 }
67
68 /// Bulk-publish items: set `is_public = true` and clear any scheduled `publish_at`.
69 ///
70 /// Only affects items matching both the given IDs and project. Returns rows affected.
71 #[tracing::instrument(skip_all)]
72 pub async fn bulk_publish(
73 pool: &PgPool,
74 item_ids: &[ItemId],
75 project_id: ProjectId,
76 user_id: UserId,
77 ) -> Result<u64> {
78 let result = sqlx::query(
79 r"
80 UPDATE items
81 SET is_public = true, publish_at = NULL, updated_at = NOW()
82 WHERE id = ANY($1) AND project_id = $2
83 AND project_id IN (SELECT id FROM projects WHERE user_id = $3)
84 AND removed_by_admin = false
85 ",
86 )
87 .bind(item_ids)
88 .bind(project_id)
89 .bind(user_id)
90 .execute(pool)
91 .await?;
92
93 Ok(result.rows_affected())
94 }
95
96 /// Bulk-unpublish items: set `is_public = false`.
97 ///
98 /// Only affects items matching both the given IDs and project. Returns rows affected.
99 #[tracing::instrument(skip_all)]
100 pub async fn bulk_unpublish(
101 pool: &PgPool,
102 item_ids: &[ItemId],
103 project_id: ProjectId,
104 user_id: UserId,
105 ) -> Result<u64> {
106 let result = sqlx::query(
107 r"
108 UPDATE items
109 SET is_public = false, updated_at = NOW()
110 WHERE id = ANY($1) AND project_id = $2
111 AND project_id IN (SELECT id FROM projects WHERE user_id = $3)
112 ",
113 )
114 .bind(item_ids)
115 .bind(project_id)
116 .bind(user_id)
117 .execute(pool)
118 .await?;
119
120 Ok(result.rows_affected())
121 }
122
123 /// Soft-delete items from a project (sets deleted_at, recoverable for 7 days).
124 ///
125 /// Only affects items matching both the given IDs and project. Returns rows affected.
126 #[tracing::instrument(skip_all)]
127 pub async fn bulk_delete(
128 pool: &PgPool,
129 item_ids: &[ItemId],
130 project_id: ProjectId,
131 user_id: UserId,
132 ) -> Result<u64> {
133 let result = sqlx::query(
134 r"
135 UPDATE items SET deleted_at = NOW(), is_public = false
136 WHERE id = ANY($1) AND project_id = $2 AND deleted_at IS NULL
137 AND project_id IN (SELECT id FROM projects WHERE user_id = $3)
138 ",
139 )
140 .bind(item_ids)
141 .bind(project_id)
142 .bind(user_id)
143 .execute(pool)
144 .await?;
145
146 Ok(result.rows_affected())
147 }
148
149 /// Bulk-update price on selected items.
150 ///
151 /// Only affects items matching both the given IDs and project. Returns rows affected.
152 #[tracing::instrument(skip_all)]
153 pub async fn bulk_update_price(
154 pool: &PgPool,
155 item_ids: &[ItemId],
156 project_id: ProjectId,
157 user_id: UserId,
158 price_cents: PriceCents,
159 ) -> Result<u64> {
160 let result = sqlx::query(
161 r"
162 UPDATE items SET price_cents = $4
163 WHERE id = ANY($1) AND project_id = $2
164 AND project_id IN (SELECT id FROM projects WHERE user_id = $3)
165 ",
166 )
167 .bind(item_ids)
168 .bind(project_id)
169 .bind(user_id)
170 .bind(price_cents)
171 .execute(pool)
172 .await?;
173
174 Ok(result.rows_affected())
175 }
176
177 /// Bulk-add a tag to selected items (skips duplicates via ON CONFLICT).
178 ///
179 /// Returns number of new tag associations created.
180 #[tracing::instrument(skip_all)]
181 pub async fn bulk_add_tag(
182 pool: &PgPool,
183 item_ids: &[ItemId],
184 project_id: ProjectId,
185 user_id: UserId,
186 tag_id: crate::db::TagId,
187 ) -> Result<u64> {
188 // Verify all items belong to the project owned by this user,
189 // then insert tag associations for each.
190 let result = sqlx::query(
191 r"
192 INSERT INTO item_tags (item_id, tag_id)
193 SELECT i.id, $4
194 FROM items i
195 JOIN projects p ON i.project_id = p.id
196 WHERE i.id = ANY($1) AND i.project_id = $2 AND p.user_id = $3
197 ON CONFLICT (item_id, tag_id) DO NOTHING
198 ",
199 )
200 .bind(item_ids)
201 .bind(project_id)
202 .bind(user_id)
203 .bind(tag_id)
204 .execute(pool)
205 .await?;
206
207 Ok(result.rows_affected())
208 }
209
210 /// Duplicate an item and its metadata (tags, chapters, content insertion placements).
211 ///
212 /// Creates a draft copy with "Copy of ..." title. Does not copy versions (S3 files),
213 /// license keys, download codes, or discount codes.
214 #[tracing::instrument(skip_all)]
215 pub async fn duplicate_item(pool: &PgPool, source_id: ItemId, user_id: UserId) -> Result<DbItem> {
216 let mut tx = pool.begin().await?;
217
218 // Generate a unique slug for the copy (verify ownership via project)
219 let source = sqlx::query_as::<_, DbItem>(
220 "SELECT * FROM items WHERE id = $1 AND project_id IN (SELECT id FROM projects WHERE user_id = $2)",
221 )
222 .bind(source_id)
223 .bind(user_id)
224 .fetch_one(&mut *tx)
225 .await?;
226 let copy_title = format!("Copy of {}", source.title);
227 let copy_title: String = copy_title.chars().take(200).collect();
228 let base_slug = crate::helpers::slugify(&copy_title).to_string();
229
230 // Clone the item row with a collision-safe slug. A pre-check
231 // (`item_slug_exists`) followed by an INSERT races two concurrent
232 // duplicates onto the same slug, the loser hit a raw 23505 -> 500. Instead
233 // attempt the INSERT under a savepoint and retry with a numeric suffix on a
234 // unique violation, so the DB's unique index is the arbiter, not a
235 // check-then-write window. Mirrors `helpers::insert_with_unique_slug`,
236 // adapted to run inside this transaction: a bare 23505 aborts the tx, so
237 // each attempt is wrapped in a savepoint and `ROLLBACK TO SAVEPOINT`
238 // recovers the tx for the next try.
239 let mut suffix = 1u32;
240 let new_item = loop {
241 let candidate = if suffix == 1 {
242 base_slug.clone()
243 } else {
244 format!("{base_slug}-{suffix}")
245 };
246 sqlx::query("SAVEPOINT dup_item_slug")
247 .execute(&mut *tx)
248 .await?;
249 let res = sqlx::query_as::<_, DbItem>(
250 r"
251 INSERT INTO items (
252 project_id, title, description, price_cents, item_type, thumbnail_url,
253 sort_order, body, word_count, reading_time_minutes, duration_seconds,
254 episode_number, enable_license_keys, default_max_activations,
255 pwyw_enabled, pwyw_min_cents, is_public, slug
256 )
257 SELECT
258 project_id, LEFT('Copy of ' || title, 200), description, price_cents,
259 item_type, thumbnail_url, sort_order, body, word_count,
260 reading_time_minutes, duration_seconds, episode_number,
261 enable_license_keys, default_max_activations, pwyw_enabled,
262 pwyw_min_cents, false, $2
263 FROM items WHERE id = $1
264 RETURNING *
265 ",
266 )
267 .bind(source_id)
268 .bind(&candidate)
269 .fetch_one(&mut *tx)
270 .await;
271 match res {
272 Ok(item) => {
273 sqlx::query("RELEASE SAVEPOINT dup_item_slug")
274 .execute(&mut *tx)
275 .await?;
276 break item;
277 }
278 Err(e) => {
279 sqlx::query("ROLLBACK TO SAVEPOINT dup_item_slug")
280 .execute(&mut *tx)
281 .await?;
282 let app_err = crate::error::AppError::from(e);
283 if crate::helpers::is_unique_violation(&app_err) && suffix < 100 {
284 suffix += 1;
285 continue;
286 }
287 return Err(app_err);
288 }
289 }
290 };
291
292 sqlx::query(
293 r"
294 INSERT INTO item_tags (item_id, tag_id, is_primary)
295 SELECT $2, tag_id, is_primary FROM item_tags WHERE item_id = $1
296 ",
297 )
298 .bind(source_id)
299 .bind(new_item.id)
300 .execute(&mut *tx)
301 .await?;
302
303 sqlx::query(
304 r"
305 INSERT INTO chapters (item_id, title, start_seconds, sort_order)
306 SELECT $2, title, start_seconds, sort_order FROM chapters WHERE item_id = $1
307 ",
308 )
309 .bind(source_id)
310 .bind(new_item.id)
311 .execute(&mut *tx)
312 .await?;
313
314 sqlx::query(
315 r"
316 INSERT INTO content_insertion_placements (item_id, insertion_id, position, offset_ms, sort_order)
317 SELECT $2, insertion_id, position, offset_ms, sort_order
318 FROM content_insertion_placements WHERE item_id = $1
319 ",
320 )
321 .bind(source_id)
322 .bind(new_item.id)
323 .execute(&mut *tx)
324 .await?;
325
326 tx.commit().await?;
327
328 Ok(new_item)
329 }
330