Skip to main content

max / makenotwork

10.9 KB · 382 lines History Blame Raw
1 //! CRUD operations for fan collections (user-curated lists of items).
2
3 use sqlx::PgPool;
4
5 use super::models::{DbCollection, DbCollectionItemRow, DbCollectionWithCount};
6 use super::{CollectionId, ItemId, Slug, UserId};
7 use crate::error::Result;
8
9 /// Create a new collection for a user.
10 #[tracing::instrument(skip_all)]
11 pub async fn create_collection(
12 pool: &PgPool,
13 user_id: UserId,
14 slug: &Slug,
15 title: &str,
16 description: Option<&str>,
17 is_public: bool,
18 ) -> Result<DbCollection> {
19 // Collections intentionally REJECT a slug clash rather than auto-suffix
20 // (unlike projects/items/sections): the slug is user-chosen and they should
21 // be told it's taken. Map the per-user unique violation to a clean
22 // validation error here so every caller, web and the CLI/service route,
23 // gets the same 400 instead of a raw 500. This is the seal: no caller can
24 // surface an unhandled 23505 from a collection insert.
25 let collection = sqlx::query_as::<_, DbCollection>(
26 r"
27 INSERT INTO collections (user_id, slug, title, description, is_public)
28 VALUES ($1, $2, $3, $4, $5)
29 RETURNING *
30 ",
31 )
32 .bind(user_id)
33 .bind(slug)
34 .bind(title)
35 .bind(description)
36 .bind(is_public)
37 .fetch_one(pool)
38 .await
39 .map_err(|e| {
40 let e = crate::error::AppError::from(e);
41 if crate::helpers::is_unique_violation(&e) {
42 crate::error::AppError::validation(
43 "You already have a collection with this slug".to_string(),
44 )
45 } else {
46 e
47 }
48 })?;
49
50 Ok(collection)
51 }
52
53 /// Update a collection's title, description, and visibility.
54 #[tracing::instrument(skip_all)]
55 pub async fn update_collection(
56 pool: &PgPool,
57 collection_id: CollectionId,
58 title: &str,
59 description: Option<&str>,
60 is_public: bool,
61 ) -> Result<DbCollection> {
62 let collection = sqlx::query_as::<_, DbCollection>(
63 r"
64 UPDATE collections
65 SET title = $2, description = $3, is_public = $4, updated_at = NOW()
66 WHERE id = $1
67 RETURNING *
68 ",
69 )
70 .bind(collection_id)
71 .bind(title)
72 .bind(description)
73 .bind(is_public)
74 .fetch_one(pool)
75 .await?;
76
77 Ok(collection)
78 }
79
80 /// Delete a collection owned by `owner_id`. Returns true if a row was deleted.
81 ///
82 /// Ownership is scoped IN the SQL (defense in depth, mirroring
83 /// `blog_posts::delete_blog_post` / `media_files::delete`) so this can't remove
84 /// another user's collection even if a caller skips the upstream ownership check.
85 #[tracing::instrument(skip_all)]
86 pub async fn delete_collection(
87 pool: &PgPool,
88 collection_id: CollectionId,
89 owner_id: UserId,
90 ) -> Result<bool> {
91 let result = sqlx::query("DELETE FROM collections WHERE id = $1 AND user_id = $2")
92 .bind(collection_id)
93 .bind(owner_id)
94 .execute(pool)
95 .await?;
96
97 Ok(result.rows_affected() > 0)
98 }
99
100 #[tracing::instrument(skip_all)]
101 pub async fn get_collection_by_id(
102 pool: &PgPool,
103 collection_id: CollectionId,
104 ) -> Result<Option<DbCollection>> {
105 let collection = sqlx::query_as::<_, DbCollection>("SELECT * FROM collections WHERE id = $1")
106 .bind(collection_id)
107 .fetch_optional(pool)
108 .await?;
109
110 Ok(collection)
111 }
112
113 /// Get a collection by user ID and slug.
114 #[tracing::instrument(skip_all)]
115 pub async fn get_collection_by_user_and_slug(
116 pool: &PgPool,
117 user_id: UserId,
118 slug: &Slug,
119 ) -> Result<Option<DbCollection>> {
120 let collection = sqlx::query_as::<_, DbCollection>(
121 "SELECT * FROM collections WHERE user_id = $1 AND slug = $2",
122 )
123 .bind(user_id)
124 .bind(slug)
125 .fetch_optional(pool)
126 .await?;
127
128 Ok(collection)
129 }
130
131 /// Get all collections for a user (with item counts), for the dashboard.
132 #[tracing::instrument(skip_all)]
133 pub async fn get_collections_by_user(
134 pool: &PgPool,
135 user_id: UserId,
136 ) -> Result<Vec<DbCollectionWithCount>> {
137 let collections = sqlx::query_as::<_, DbCollectionWithCount>(
138 r"
139 SELECT c.*, COUNT(ci.item_id) AS item_count
140 FROM collections c
141 LEFT JOIN collection_items ci ON ci.collection_id = c.id
142 WHERE c.user_id = $1
143 GROUP BY c.id
144 ORDER BY c.updated_at DESC
145 LIMIT 500
146 ",
147 )
148 .bind(user_id)
149 .fetch_all(pool)
150 .await?;
151
152 Ok(collections)
153 }
154
155 /// Get public collections for a user (with item counts), for the profile page.
156 #[tracing::instrument(skip_all)]
157 pub async fn get_public_collections_by_user(
158 pool: &PgPool,
159 user_id: UserId,
160 ) -> Result<Vec<DbCollectionWithCount>> {
161 let collections = sqlx::query_as::<_, DbCollectionWithCount>(
162 r"
163 SELECT c.*, COUNT(ci.item_id) AS item_count
164 FROM collections c
165 LEFT JOIN collection_items ci ON ci.collection_id = c.id
166 WHERE c.user_id = $1 AND c.is_public = true
167 GROUP BY c.id
168 ORDER BY c.updated_at DESC
169 LIMIT 500
170 ",
171 )
172 .bind(user_id)
173 .fetch_all(pool)
174 .await?;
175
176 Ok(collections)
177 }
178
179 /// Count collections owned by a user.
180 #[tracing::instrument(skip_all)]
181 pub async fn count_collections_by_user(pool: &PgPool, user_id: UserId) -> Result<i64> {
182 let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM collections WHERE user_id = $1")
183 .bind(user_id)
184 .fetch_one(pool)
185 .await?;
186
187 Ok(count)
188 }
189
190 /// Add an item to a collection. Idempotent (ON CONFLICT DO NOTHING).
191 /// Appends at max(position)+1 atomically via INSERT...SELECT.
192 #[tracing::instrument(skip_all)]
193 pub async fn add_item_to_collection(
194 pool: &PgPool,
195 collection_id: CollectionId,
196 item_id: ItemId,
197 ) -> Result<()> {
198 sqlx::query(
199 r"
200 INSERT INTO collection_items (collection_id, item_id, position)
201 VALUES ($1, $2, COALESCE((SELECT MAX(position) FROM collection_items WHERE collection_id = $1), -1) + 1)
202 ON CONFLICT (collection_id, item_id) DO NOTHING
203 ",
204 )
205 .bind(collection_id)
206 .bind(item_id)
207 .execute(pool)
208 .await?;
209
210 // Touch the collection's updated_at
211 sqlx::query("UPDATE collections SET updated_at = NOW() WHERE id = $1")
212 .bind(collection_id)
213 .execute(pool)
214 .await?;
215
216 Ok(())
217 }
218
219 /// Remove an item from a collection. Returns true if a row was deleted.
220 #[tracing::instrument(skip_all)]
221 pub async fn remove_item_from_collection(
222 pool: &PgPool,
223 collection_id: CollectionId,
224 item_id: ItemId,
225 ) -> Result<bool> {
226 let result =
227 sqlx::query("DELETE FROM collection_items WHERE collection_id = $1 AND item_id = $2")
228 .bind(collection_id)
229 .bind(item_id)
230 .execute(pool)
231 .await?;
232
233 if result.rows_affected() > 0 {
234 sqlx::query("UPDATE collections SET updated_at = NOW() WHERE id = $1")
235 .bind(collection_id)
236 .execute(pool)
237 .await?;
238 }
239
240 Ok(result.rows_affected() > 0)
241 }
242
243 /// Get items in a collection, joined with item/project/user data, ordered by position.
244 #[tracing::instrument(skip_all)]
245 pub async fn get_collection_items(
246 pool: &PgPool,
247 collection_id: CollectionId,
248 ) -> Result<Vec<DbCollectionItemRow>> {
249 let items = sqlx::query_as::<_, DbCollectionItemRow>(
250 r"
251 SELECT
252 ci.item_id,
253 i.title,
254 i.description,
255 i.price_cents,
256 i.item_type,
257 u.username,
258 p.title AS project_title,
259 ci.position,
260 ci.added_at
261 FROM collection_items ci
262 JOIN items i ON i.id = ci.item_id
263 JOIN projects p ON p.id = i.project_id
264 JOIN users u ON u.id = p.user_id
265 WHERE ci.collection_id = $1
266 ORDER BY ci.position
267 LIMIT 1000
268 ",
269 )
270 .bind(collection_id)
271 .fetch_all(pool)
272 .await?;
273
274 Ok(items)
275 }
276
277 /// Get item IDs for multiple collections in a single query (batch, avoids N+1).
278 #[tracing::instrument(skip_all)]
279 pub async fn get_item_ids_by_collections(
280 pool: &PgPool,
281 collection_ids: &[CollectionId],
282 ) -> Result<std::collections::HashMap<CollectionId, Vec<ItemId>>> {
283 if collection_ids.is_empty() {
284 return Ok(std::collections::HashMap::new());
285 }
286 let rows: Vec<(CollectionId, ItemId)> = sqlx::query_as(
287 r"
288 SELECT collection_id, item_id
289 FROM collection_items
290 WHERE collection_id = ANY($1)
291 ORDER BY collection_id, position
292 ",
293 )
294 .bind(collection_ids)
295 .fetch_all(pool)
296 .await?;
297
298 let mut map: std::collections::HashMap<CollectionId, Vec<ItemId>> =
299 std::collections::HashMap::new();
300 for (cid, iid) in rows {
301 map.entry(cid).or_default().push(iid);
302 }
303 Ok(map)
304 }
305
306 /// Count items in a collection.
307 #[tracing::instrument(skip_all)]
308 pub async fn count_collection_items(pool: &PgPool, collection_id: CollectionId) -> Result<i64> {
309 let count: i64 =
310 sqlx::query_scalar("SELECT COUNT(*) FROM collection_items WHERE collection_id = $1")
311 .bind(collection_id)
312 .fetch_one(pool)
313 .await?;
314
315 Ok(count)
316 }
317
318 /// Reorder items in a collection by assigning position from the given ID sequence.
319 /// Wrapped in a transaction so a crash mid-reorder doesn't leave inconsistent state.
320 #[tracing::instrument(skip_all)]
321 pub async fn reorder_collection_items(
322 pool: &PgPool,
323 collection_id: CollectionId,
324 item_ids: &[ItemId],
325 ) -> Result<()> {
326 let ids: Vec<uuid::Uuid> = item_ids.iter().map(|id| *id.as_uuid()).collect();
327 let mut tx = pool.begin().await?;
328 // Single `UNNEST ... WITH ORDINALITY` update instead of one query per item,
329 // atomic within the same transaction that bumps the collection's timestamp.
330 sqlx::query(
331 r"
332 UPDATE collection_items AS ci
333 SET position = ord.pos::int - 1
334 FROM UNNEST($1::uuid[]) WITH ORDINALITY AS ord(id, pos)
335 WHERE ci.collection_id = $2 AND ci.item_id = ord.id
336 ",
337 )
338 .bind(&ids)
339 .bind(collection_id)
340 .execute(&mut *tx)
341 .await?;
342
343 sqlx::query("UPDATE collections SET updated_at = NOW() WHERE id = $1")
344 .bind(collection_id)
345 .execute(&mut *tx)
346 .await?;
347 tx.commit().await?;
348
349 Ok(())
350 }
351
352 /// Get a user's collections with membership state for a specific item.
353 /// Returns (collection_id, title, is_in_collection) for the "add to collection" dropdown.
354 #[tracing::instrument(skip_all)]
355 pub async fn get_user_collections_for_item(
356 pool: &PgPool,
357 user_id: UserId,
358 item_id: ItemId,
359 ) -> Result<Vec<(CollectionId, String, bool)>> {
360 let rows: Vec<(CollectionId, String, bool)> = sqlx::query_as(
361 r"
362 SELECT
363 c.id,
364 c.title,
365 EXISTS(
366 SELECT 1 FROM collection_items ci
367 WHERE ci.collection_id = c.id AND ci.item_id = $2
368 ) AS is_in_collection
369 FROM collections c
370 WHERE c.user_id = $1
371 ORDER BY c.updated_at DESC
372 LIMIT 500
373 ",
374 )
375 .bind(user_id)
376 .bind(item_id)
377 .fetch_all(pool)
378 .await?;
379
380 Ok(rows)
381 }
382