Skip to main content

max / makenotwork

16.1 KB · 576 lines History Blame Raw
1 //! Subscription queries: tier CRUD, subscription lifecycle, and access control.
2
3 use chrono::{DateTime, Utc};
4 use sqlx::PgPool;
5
6 use super::enums::SubscriptionStatus;
7 use super::models::*;
8 use super::{PriceCents, ProjectId, SubscriptionId, SubscriptionTierId, UserId};
9 use crate::error::Result;
10
11 // ── Tier CRUD ──
12
13 /// Create a new subscription tier for a project.
14 #[tracing::instrument(skip_all)]
15 pub async fn create_subscription_tier(
16 pool: &PgPool,
17 project_id: ProjectId,
18 name: &str,
19 description: Option<&str>,
20 price_cents: PriceCents,
21 ) -> Result<DbSubscriptionTier> {
22 let tier = sqlx::query_as::<_, DbSubscriptionTier>(
23 r#"
24 INSERT INTO subscription_tiers (project_id, name, description, price_cents)
25 VALUES ($1, $2, $3, $4)
26 RETURNING *
27 "#,
28 )
29 .bind(project_id)
30 .bind(name)
31 .bind(description)
32 .bind(price_cents.as_i32())
33 .fetch_one(pool)
34 .await?;
35
36 Ok(tier)
37 }
38
39 /// Get a subscription tier by ID.
40 #[tracing::instrument(skip_all)]
41 pub async fn get_subscription_tier_by_id(
42 pool: &PgPool,
43 id: SubscriptionTierId,
44 ) -> Result<Option<DbSubscriptionTier>> {
45 let tier = sqlx::query_as::<_, DbSubscriptionTier>(
46 "SELECT * FROM subscription_tiers WHERE id = $1",
47 )
48 .bind(id)
49 .fetch_optional(pool)
50 .await?;
51
52 Ok(tier)
53 }
54
55 /// Get all active tiers for a project, ordered by sort_order.
56 #[tracing::instrument(skip_all)]
57 pub async fn get_active_tiers_by_project(
58 pool: &PgPool,
59 project_id: ProjectId,
60 ) -> Result<Vec<DbSubscriptionTier>> {
61 let tiers = sqlx::query_as::<_, DbSubscriptionTier>(
62 "SELECT * FROM subscription_tiers WHERE project_id = $1 AND is_active = true ORDER BY sort_order, created_at",
63 )
64 .bind(project_id)
65 .fetch_all(pool)
66 .await?;
67
68 Ok(tiers)
69 }
70
71 /// Get all tiers for a project (active and inactive), for dashboard management.
72 #[tracing::instrument(skip_all)]
73 pub async fn get_all_tiers_by_project(
74 pool: &PgPool,
75 project_id: ProjectId,
76 ) -> Result<Vec<DbSubscriptionTier>> {
77 let tiers = sqlx::query_as::<_, DbSubscriptionTier>(
78 "SELECT * FROM subscription_tiers WHERE project_id = $1 ORDER BY sort_order, created_at",
79 )
80 .bind(project_id)
81 .fetch_all(pool)
82 .await?;
83
84 Ok(tiers)
85 }
86
87 /// Update a subscription tier's name, description, and active status.
88 #[tracing::instrument(skip_all)]
89 pub async fn update_subscription_tier(
90 pool: &PgPool,
91 id: SubscriptionTierId,
92 name: &str,
93 description: Option<&str>,
94 is_active: bool,
95 ) -> Result<DbSubscriptionTier> {
96 let tier = sqlx::query_as::<_, DbSubscriptionTier>(
97 r#"
98 UPDATE subscription_tiers
99 SET name = $2, description = $3, is_active = $4
100 WHERE id = $1
101 RETURNING *
102 "#,
103 )
104 .bind(id)
105 .bind(name)
106 .bind(description)
107 .bind(is_active)
108 .fetch_one(pool)
109 .await?;
110
111 Ok(tier)
112 }
113
114 /// Store Stripe product and price IDs on a tier after creating them on connected account.
115 #[tracing::instrument(skip_all)]
116 pub async fn update_tier_stripe_ids(
117 pool: &PgPool,
118 tier_id: SubscriptionTierId,
119 product_id: &str,
120 price_id: &str,
121 ) -> Result<()> {
122 sqlx::query(
123 r#"
124 UPDATE subscription_tiers
125 SET stripe_product_id = $2, stripe_price_id = $3
126 WHERE id = $1
127 "#,
128 )
129 .bind(tier_id)
130 .bind(product_id)
131 .bind(price_id)
132 .execute(pool)
133 .await?;
134
135 Ok(())
136 }
137
138 /// Delete a subscription tier. Soft-deletes (sets is_active=false) if any
139 /// subscriptions reference it; hard-deletes otherwise.
140 ///
141 /// Uses a transaction with FOR UPDATE to prevent a TOCTOU race where a
142 /// subscription could be created between the existence check and the delete.
143 #[tracing::instrument(skip_all)]
144 pub async fn delete_subscription_tier(pool: &PgPool, id: SubscriptionTierId) -> Result<()> {
145 let mut tx = pool.begin().await?;
146
147 // Lock the tier row to serialize against concurrent subscription creation
148 sqlx::query("SELECT id FROM subscription_tiers WHERE id = $1 FOR UPDATE")
149 .bind(id)
150 .fetch_optional(&mut *tx)
151 .await?
152 .ok_or(sqlx::Error::RowNotFound)?;
153
154 let has_subscriptions: bool = sqlx::query_scalar(
155 "SELECT EXISTS(SELECT 1 FROM subscriptions WHERE tier_id = $1)",
156 )
157 .bind(id)
158 .fetch_one(&mut *tx)
159 .await?;
160
161 if has_subscriptions {
162 sqlx::query("UPDATE subscription_tiers SET is_active = false WHERE id = $1")
163 .bind(id)
164 .execute(&mut *tx)
165 .await?;
166 } else {
167 sqlx::query("DELETE FROM subscription_tiers WHERE id = $1")
168 .bind(id)
169 .execute(&mut *tx)
170 .await?;
171 }
172
173 tx.commit().await?;
174 Ok(())
175 }
176
177 // ── Subscription lifecycle ──
178
179 /// Create a new subscription record after successful checkout.
180 ///
181 /// Returns `None` if the subscription already exists (duplicate webhook
182 /// or concurrent active subscription for the same user+project).
183 /// The partial UNIQUE index on `(subscriber_id, project_id) WHERE status = 'active'`
184 /// prevents multiple active subscriptions at the DB level.
185 #[tracing::instrument(skip_all)]
186 pub async fn create_subscription<'e>(
187 executor: impl sqlx::PgExecutor<'e>,
188 subscriber_id: UserId,
189 tier_id: SubscriptionTierId,
190 project_id: ProjectId,
191 stripe_subscription_id: &str,
192 stripe_customer_id: &str,
193 ) -> Result<Option<DbSubscription>> {
194 let sub = sqlx::query_as::<_, DbSubscription>(
195 r#"
196 INSERT INTO subscriptions (subscriber_id, tier_id, project_id, stripe_subscription_id, stripe_customer_id)
197 VALUES ($1, $2, $3, $4, $5)
198 ON CONFLICT DO NOTHING
199 RETURNING *
200 "#,
201 )
202 .bind(subscriber_id)
203 .bind(tier_id)
204 .bind(project_id)
205 .bind(stripe_subscription_id)
206 .bind(stripe_customer_id)
207 .fetch_optional(executor)
208 .await?;
209
210 Ok(sub)
211 }
212
213 /// Look up a subscription by its Stripe subscription ID.
214 #[tracing::instrument(skip_all)]
215 pub async fn get_subscription_by_stripe_id(
216 pool: &PgPool,
217 stripe_sub_id: &str,
218 ) -> Result<Option<DbSubscription>> {
219 let sub = sqlx::query_as::<_, DbSubscription>(
220 "SELECT * FROM subscriptions WHERE stripe_subscription_id = $1",
221 )
222 .bind(stripe_sub_id)
223 .fetch_optional(pool)
224 .await?;
225
226 Ok(sub)
227 }
228
229 /// Update subscription status (active, past_due, canceled, unpaid).
230 /// Sets canceled_at when transitioning to canceled, preserving existing value.
231 /// Returns the updated record, or None if not found.
232 #[tracing::instrument(skip_all)]
233 pub async fn update_subscription_status<'e>(
234 executor: impl sqlx::PgExecutor<'e>,
235 stripe_sub_id: &str,
236 status: SubscriptionStatus,
237 ) -> Result<Option<DbSubscription>> {
238 let sub = sqlx::query_as::<_, DbSubscription>(
239 r#"
240 UPDATE subscriptions
241 SET status = $2,
242 canceled_at = CASE
243 WHEN $2 = 'canceled' THEN COALESCE(canceled_at, NOW())
244 ELSE canceled_at
245 END
246 WHERE stripe_subscription_id = $1
247 RETURNING *
248 "#,
249 )
250 .bind(stripe_sub_id)
251 .bind(status)
252 .fetch_optional(executor)
253 .await?;
254
255 Ok(sub)
256 }
257
258 /// Update the billing period timestamps for a subscription.
259 #[tracing::instrument(skip_all)]
260 pub async fn update_subscription_period<'e>(
261 executor: impl sqlx::PgExecutor<'e>,
262 stripe_sub_id: &str,
263 period_start: DateTime<Utc>,
264 period_end: DateTime<Utc>,
265 ) -> Result<()> {
266 sqlx::query(
267 r#"
268 UPDATE subscriptions
269 SET current_period_start = $2, current_period_end = $3
270 WHERE stripe_subscription_id = $1
271 "#,
272 )
273 .bind(stripe_sub_id)
274 .bind(period_start)
275 .bind(period_end)
276 .execute(executor)
277 .await?;
278
279 Ok(())
280 }
281
282 /// Mark a subscription as canceled.
283 #[tracing::instrument(skip_all)]
284 pub async fn cancel_subscription(
285 pool: &PgPool,
286 stripe_sub_id: &str,
287 ) -> Result<Option<DbSubscription>> {
288 let sub = sqlx::query_as::<_, DbSubscription>(
289 r#"
290 UPDATE subscriptions
291 SET status = 'canceled', canceled_at = COALESCE(canceled_at, NOW())
292 WHERE stripe_subscription_id = $1
293 RETURNING *
294 "#,
295 )
296 .bind(stripe_sub_id)
297 .fetch_optional(pool)
298 .await?;
299
300 Ok(sub)
301 }
302
303 // ── Suspension pause/resume ──
304
305 /// Get all active subscriptions to a creator's projects (for pausing on suspension).
306 #[tracing::instrument(skip_all)]
307 pub async fn get_active_subscriptions_by_creator(
308 pool: &PgPool,
309 creator_id: UserId,
310 ) -> Result<Vec<DbSubscription>> {
311 let subs = sqlx::query_as::<_, DbSubscription>(
312 r#"
313 SELECT s.* FROM subscriptions s
314 WHERE s.project_id IN (SELECT id FROM projects WHERE user_id = $1)
315 AND s.status = 'active'
316 AND s.paused_at IS NULL
317 "#,
318 )
319 .bind(creator_id)
320 .fetch_all(pool)
321 .await?;
322
323 Ok(subs)
324 }
325
326 /// Mark all active subscriptions to a creator's projects as paused.
327 #[tracing::instrument(skip_all)]
328 pub async fn pause_subscriptions_for_creator(
329 pool: &PgPool,
330 creator_id: UserId,
331 ) -> Result<u64> {
332 let result = sqlx::query(
333 r#"
334 UPDATE subscriptions SET paused_at = NOW()
335 WHERE project_id IN (SELECT id FROM projects WHERE user_id = $1)
336 AND status = 'active'
337 AND paused_at IS NULL
338 "#,
339 )
340 .bind(creator_id)
341 .execute(pool)
342 .await?;
343
344 Ok(result.rows_affected())
345 }
346
347 /// Get all paused subscriptions to a creator's projects (for cancelling on termination).
348 #[tracing::instrument(skip_all)]
349 pub async fn get_paused_subscriptions_by_creator(
350 pool: &PgPool,
351 creator_id: UserId,
352 ) -> Result<Vec<DbSubscription>> {
353 let subs = sqlx::query_as::<_, DbSubscription>(
354 r#"
355 SELECT s.* FROM subscriptions s
356 WHERE s.project_id IN (SELECT id FROM projects WHERE user_id = $1)
357 AND s.status = 'active'
358 AND s.paused_at IS NOT NULL
359 "#,
360 )
361 .bind(creator_id)
362 .fetch_all(pool)
363 .await?;
364
365 Ok(subs)
366 }
367
368 /// Resume all paused subscriptions for a creator's projects.
369 #[tracing::instrument(skip_all)]
370 pub async fn resume_subscriptions_for_creator(
371 pool: &PgPool,
372 creator_id: UserId,
373 ) -> Result<Vec<DbSubscription>> {
374 let subs = sqlx::query_as::<_, DbSubscription>(
375 r#"
376 UPDATE subscriptions SET paused_at = NULL
377 WHERE project_id IN (SELECT id FROM projects WHERE user_id = $1)
378 AND status = 'active'
379 AND paused_at IS NOT NULL
380 RETURNING *
381 "#,
382 )
383 .bind(creator_id)
384 .fetch_all(pool)
385 .await?;
386
387 Ok(subs)
388 }
389
390 // ── Access control ──
391
392 /// Check if a user has an active subscription to a project.
393 #[tracing::instrument(skip_all)]
394 pub async fn has_active_subscription_to_project(
395 pool: &PgPool,
396 user_id: UserId,
397 project_id: ProjectId,
398 ) -> Result<bool> {
399 // Defense-in-depth on a missed/delayed `customer.subscription.deleted`
400 // webhook: also reject when the current period has ended. status='active'
401 // alone trusts Stripe to push the cancellation event promptly.
402 let count: i64 = sqlx::query_scalar(
403 "SELECT COUNT(*) FROM subscriptions \
404 WHERE subscriber_id = $1 AND project_id = $2 AND status = 'active' AND paused_at IS NULL \
405 AND (current_period_end IS NULL OR current_period_end > NOW())",
406 )
407 .bind(user_id)
408 .bind(project_id)
409 .fetch_one(pool)
410 .await?;
411
412 Ok(count > 0)
413 }
414
415 /// Get user subscriptions joined with project and tier data (for library display).
416 #[tracing::instrument(skip_all)]
417 pub async fn get_user_subscriptions_with_details(
418 pool: &PgPool,
419 user_id: UserId,
420 ) -> Result<Vec<DbUserSubscriptionRow>> {
421 let rows = sqlx::query_as::<_, DbUserSubscriptionRow>(
422 "SELECT s.id, s.project_id, p.title AS project_title, p.slug AS project_slug,
423 t.name AS tier_name, t.price_cents, s.status,
424 s.current_period_end, s.stripe_subscription_id
425 FROM subscriptions s
426 JOIN projects p ON p.id = s.project_id
427 JOIN subscription_tiers t ON t.id = s.tier_id
428 WHERE s.subscriber_id = $1
429 ORDER BY s.created_at DESC
430 LIMIT 1000",
431 )
432 .bind(user_id)
433 .fetch_all(pool)
434 .await?;
435
436 Ok(rows)
437 }
438
439 /// Get the number of active subscribers to a project (for dashboard display).
440 #[tracing::instrument(skip_all)]
441 pub async fn get_project_subscriber_count(
442 pool: &PgPool,
443 project_id: ProjectId,
444 ) -> Result<i64> {
445 let count: i64 = sqlx::query_scalar(
446 "SELECT COUNT(*) FROM subscriptions WHERE project_id = $1 AND status = 'active' AND paused_at IS NULL",
447 )
448 .bind(project_id)
449 .fetch_one(pool)
450 .await?;
451
452 Ok(count)
453 }
454
455 // ── Item-level subscriptions ──
456
457 /// Check if a user has an active subscription to a specific item.
458 #[tracing::instrument(skip_all)]
459 pub async fn has_active_subscription_to_item(
460 pool: &PgPool,
461 user_id: UserId,
462 item_id: super::ItemId,
463 ) -> Result<bool> {
464 let count: i64 = sqlx::query_scalar(
465 "SELECT COUNT(*) FROM subscriptions WHERE subscriber_id = $1 AND item_id = $2 AND status = 'active' AND paused_at IS NULL",
466 )
467 .bind(user_id)
468 .bind(item_id)
469 .fetch_one(pool)
470 .await?;
471
472 Ok(count > 0)
473 }
474
475 /// Get all item IDs that a user has active subscriptions to (for batch access checks).
476 #[tracing::instrument(skip_all)]
477 pub async fn get_user_subscribed_item_ids(
478 pool: &PgPool,
479 user_id: UserId,
480 ) -> Result<Vec<super::ItemId>> {
481 let item_ids: Vec<super::ItemId> = sqlx::query_scalar(
482 "SELECT DISTINCT item_id FROM subscriptions WHERE subscriber_id = $1 AND status = 'active' AND paused_at IS NULL AND item_id IS NOT NULL",
483 )
484 .bind(user_id)
485 .fetch_all(pool)
486 .await?;
487
488 Ok(item_ids)
489 }
490
491 // ── Export ──
492
493 /// Export all subscribers across a creator's projects.
494 ///
495 /// Returns username, display_name, tier name, subscription status, and when.
496 #[tracing::instrument(skip_all)]
497 pub async fn get_project_subscribers_for_export(
498 pool: &PgPool,
499 user_id: UserId,
500 ) -> Result<Vec<SubscriberExportRow>> {
501 let rows = sqlx::query_as::<_, SubscriberExportRow>(
502 r#"
503 SELECT u.username, u.display_name, t.name AS tier_name, s.status, s.created_at
504 FROM subscriptions s
505 JOIN users u ON u.id = s.subscriber_id
506 JOIN subscription_tiers t ON t.id = s.tier_id
507 WHERE s.project_id IN (SELECT id FROM projects WHERE user_id = $1)
508 ORDER BY s.created_at DESC
509 "#,
510 )
511 .bind(user_id)
512 .fetch_all(pool)
513 .await?;
514
515 Ok(rows)
516 }
517
518 /// Export all subscriptions across a creator's projects with full detail.
519 ///
520 /// Returns project name, tier name, price, subscriber username, status,
521 /// billing period dates, and cancellation date.
522 #[tracing::instrument(skip_all)]
523 pub async fn get_subscriptions_for_export(
524 pool: &PgPool,
525 user_id: UserId,
526 ) -> Result<Vec<SubscriptionExportRow>> {
527 let rows = sqlx::query_as::<_, SubscriptionExportRow>(
528 r#"
529 SELECT p.title AS project_title, t.name AS tier_name, t.price_cents,
530 u.username, s.status,
531 s.current_period_start, s.current_period_end,
532 s.canceled_at, s.created_at
533 FROM subscriptions s
534 JOIN users u ON u.id = s.subscriber_id
535 JOIN subscription_tiers t ON t.id = s.tier_id
536 JOIN projects p ON p.id = s.project_id
537 WHERE s.project_id IN (SELECT id FROM projects WHERE user_id = $1)
538 ORDER BY s.created_at DESC
539 "#,
540 )
541 .bind(user_id)
542 .fetch_all(pool)
543 .await?;
544
545 Ok(rows)
546 }
547
548 // ── Event log ──
549
550 /// Log a subscription webhook event for debugging and idempotency.
551 /// The UNIQUE index on stripe_event_id makes duplicate events a no-op.
552 #[tracing::instrument(skip_all)]
553 pub async fn log_subscription_event(
554 pool: &PgPool,
555 subscription_id: Option<SubscriptionId>,
556 stripe_event_id: &str,
557 event_type: &str,
558 payload: &serde_json::Value,
559 ) -> Result<()> {
560 sqlx::query(
561 r#"
562 INSERT INTO subscription_events (subscription_id, stripe_event_id, event_type, payload)
563 VALUES ($1, $2, $3, $4)
564 ON CONFLICT (stripe_event_id) DO NOTHING
565 "#,
566 )
567 .bind(subscription_id)
568 .bind(stripe_event_id)
569 .bind(event_type)
570 .bind(payload)
571 .execute(pool)
572 .await?;
573
574 Ok(())
575 }
576