Skip to main content

max / makenotwork

10.9 KB · 383 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 u.settlement_currency,
259 p.title AS project_title,
260 ci.position,
261 ci.added_at
262 FROM collection_items ci
263 JOIN items i ON i.id = ci.item_id
264 JOIN projects p ON p.id = i.project_id
265 JOIN users u ON u.id = p.user_id
266 WHERE ci.collection_id = $1
267 ORDER BY ci.position
268 LIMIT 1000
269 ",
270 )
271 .bind(collection_id)
272 .fetch_all(pool)
273 .await?;
274
275 Ok(items)
276 }
277
278 /// Get item IDs for multiple collections in a single query (batch, avoids N+1).
279 #[tracing::instrument(skip_all)]
280 pub async fn get_item_ids_by_collections(
281 pool: &PgPool,
282 collection_ids: &[CollectionId],
283 ) -> Result<std::collections::HashMap<CollectionId, Vec<ItemId>>> {
284 if collection_ids.is_empty() {
285 return Ok(std::collections::HashMap::new());
286 }
287 let rows: Vec<(CollectionId, ItemId)> = sqlx::query_as(
288 r"
289 SELECT collection_id, item_id
290 FROM collection_items
291 WHERE collection_id = ANY($1)
292 ORDER BY collection_id, position
293 ",
294 )
295 .bind(collection_ids)
296 .fetch_all(pool)
297 .await?;
298
299 let mut map: std::collections::HashMap<CollectionId, Vec<ItemId>> =
300 std::collections::HashMap::new();
301 for (cid, iid) in rows {
302 map.entry(cid).or_default().push(iid);
303 }
304 Ok(map)
305 }
306
307 /// Count items in a collection.
308 #[tracing::instrument(skip_all)]
309 pub async fn count_collection_items(pool: &PgPool, collection_id: CollectionId) -> Result<i64> {
310 let count: i64 =
311 sqlx::query_scalar("SELECT COUNT(*) FROM collection_items WHERE collection_id = $1")
312 .bind(collection_id)
313 .fetch_one(pool)
314 .await?;
315
316 Ok(count)
317 }
318
319 /// Reorder items in a collection by assigning position from the given ID sequence.
320 /// Wrapped in a transaction so a crash mid-reorder doesn't leave inconsistent state.
321 #[tracing::instrument(skip_all)]
322 pub async fn reorder_collection_items(
323 pool: &PgPool,
324 collection_id: CollectionId,
325 item_ids: &[ItemId],
326 ) -> Result<()> {
327 let ids: Vec<uuid::Uuid> = item_ids.iter().map(|id| *id.as_uuid()).collect();
328 let mut tx = pool.begin().await?;
329 // Single `UNNEST ... WITH ORDINALITY` update instead of one query per item,
330 // atomic within the same transaction that bumps the collection's timestamp.
331 sqlx::query(
332 r"
333 UPDATE collection_items AS ci
334 SET position = ord.pos::int - 1
335 FROM UNNEST($1::uuid[]) WITH ORDINALITY AS ord(id, pos)
336 WHERE ci.collection_id = $2 AND ci.item_id = ord.id
337 ",
338 )
339 .bind(&ids)
340 .bind(collection_id)
341 .execute(&mut *tx)
342 .await?;
343
344 sqlx::query("UPDATE collections SET updated_at = NOW() WHERE id = $1")
345 .bind(collection_id)
346 .execute(&mut *tx)
347 .await?;
348 tx.commit().await?;
349
350 Ok(())
351 }
352
353 /// Get a user's collections with membership state for a specific item.
354 /// Returns (collection_id, title, is_in_collection) for the "add to collection" dropdown.
355 #[tracing::instrument(skip_all)]
356 pub async fn get_user_collections_for_item(
357 pool: &PgPool,
358 user_id: UserId,
359 item_id: ItemId,
360 ) -> Result<Vec<(CollectionId, String, bool)>> {
361 let rows: Vec<(CollectionId, String, bool)> = sqlx::query_as(
362 r"
363 SELECT
364 c.id,
365 c.title,
366 EXISTS(
367 SELECT 1 FROM collection_items ci
368 WHERE ci.collection_id = c.id AND ci.item_id = $2
369 ) AS is_in_collection
370 FROM collections c
371 WHERE c.user_id = $1
372 ORDER BY c.updated_at DESC
373 LIMIT 500
374 ",
375 )
376 .bind(user_id)
377 .bind(item_id)
378 .fetch_all(pool)
379 .await?;
380
381 Ok(rows)
382 }
383