Skip to main content

max / makenotwork

31.5 KB · 820 lines History Blame Raw
1 //! Subscription queries: tier CRUD, subscription lifecycle, and access control.
2
3 use sqlx::PgPool;
4
5 use super::models::{
6 DbSubscription, DbSubscriptionTier, DbUserSubscriptionRow, SubscriberExportRow,
7 SubscriptionExportRow,
8 };
9 use super::{ItemId, PriceCents, ProjectId, SubscriptionId, SubscriptionTierId, UserId};
10 use crate::error::Result;
11
12 // ── Tier CRUD ──
13
14 /// Create a new subscription tier for a project.
15 #[tracing::instrument(skip_all)]
16 pub(crate) async fn create_subscription_tier(
17 pool: &PgPool,
18 project_id: ProjectId,
19 name: &str,
20 description: Option<&str>,
21 price_cents: PriceCents,
22 ) -> Result<DbSubscriptionTier> {
23 let tier = sqlx::query_as!(
24 DbSubscriptionTier,
25 r#"
26 INSERT INTO subscription_tiers (project_id, name, description, price_cents)
27 VALUES ($1, $2, $3, $4)
28 RETURNING id AS "id: SubscriptionTierId", project_id AS "project_id: ProjectId",
29 name, description, price_cents, stripe_product_id, stripe_price_id,
30 sort_order, is_active,
31 created_at AS "created_at: chrono::DateTime<chrono::Utc>",
32 updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
33 item_id AS "item_id: ItemId"
34 "#,
35 project_id as ProjectId,
36 name,
37 description,
38 price_cents.as_i32(),
39 )
40 .fetch_one(pool)
41 .await?;
42
43 Ok(tier)
44 }
45
46 #[tracing::instrument(skip_all)]
47 pub(crate) async fn get_subscription_tier_by_id(
48 pool: &PgPool,
49 id: SubscriptionTierId,
50 ) -> Result<Option<DbSubscriptionTier>> {
51 let tier = sqlx::query_as!(
52 DbSubscriptionTier,
53 r#"
54 SELECT id AS "id: SubscriptionTierId", project_id AS "project_id: ProjectId",
55 name, description, price_cents, stripe_product_id, stripe_price_id,
56 sort_order, is_active,
57 created_at AS "created_at: chrono::DateTime<chrono::Utc>",
58 updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
59 item_id AS "item_id: ItemId"
60 FROM subscription_tiers WHERE id = $1
61 "#,
62 id as SubscriptionTierId,
63 )
64 .fetch_optional(pool)
65 .await?;
66
67 Ok(tier)
68 }
69
70 /// Get all active tiers for a project, ordered by sort_order.
71 #[tracing::instrument(skip_all)]
72 pub(crate) async fn get_active_tiers_by_project(
73 pool: &PgPool,
74 project_id: ProjectId,
75 ) -> Result<Vec<DbSubscriptionTier>> {
76 let tiers = sqlx::query_as!(
77 DbSubscriptionTier,
78 r#"
79 SELECT id AS "id: SubscriptionTierId", project_id AS "project_id: ProjectId",
80 name, description, price_cents, stripe_product_id, stripe_price_id,
81 sort_order, is_active,
82 created_at AS "created_at: chrono::DateTime<chrono::Utc>",
83 updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
84 item_id AS "item_id: ItemId"
85 FROM subscription_tiers WHERE project_id = $1 AND is_active = true ORDER BY sort_order, created_at
86 "#,
87 project_id as ProjectId,
88 )
89 .fetch_all(pool)
90 .await?;
91
92 Ok(tiers)
93 }
94
95 /// Get all tiers for a project (active and inactive), for dashboard management.
96 #[tracing::instrument(skip_all)]
97 pub(crate) async fn get_all_tiers_by_project(
98 pool: &PgPool,
99 project_id: ProjectId,
100 ) -> Result<Vec<DbSubscriptionTier>> {
101 let tiers = sqlx::query_as!(
102 DbSubscriptionTier,
103 r#"
104 SELECT id AS "id: SubscriptionTierId", project_id AS "project_id: ProjectId",
105 name, description, price_cents, stripe_product_id, stripe_price_id,
106 sort_order, is_active,
107 created_at AS "created_at: chrono::DateTime<chrono::Utc>",
108 updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
109 item_id AS "item_id: ItemId"
110 FROM subscription_tiers WHERE project_id = $1 ORDER BY sort_order, created_at
111 "#,
112 project_id as ProjectId,
113 )
114 .fetch_all(pool)
115 .await?;
116
117 Ok(tiers)
118 }
119
120 /// Update a subscription tier's name, description, and active status.
121 #[tracing::instrument(skip_all)]
122 pub(crate) async fn update_subscription_tier(
123 pool: &PgPool,
124 id: SubscriptionTierId,
125 name: &str,
126 description: Option<&str>,
127 is_active: bool,
128 ) -> Result<DbSubscriptionTier> {
129 let tier = sqlx::query_as!(
130 DbSubscriptionTier,
131 r#"
132 UPDATE subscription_tiers
133 SET name = $2, description = $3, is_active = $4
134 WHERE id = $1
135 RETURNING id AS "id: SubscriptionTierId", project_id AS "project_id: ProjectId",
136 name, description, price_cents, stripe_product_id, stripe_price_id,
137 sort_order, is_active,
138 created_at AS "created_at: chrono::DateTime<chrono::Utc>",
139 updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
140 item_id AS "item_id: ItemId"
141 "#,
142 id as SubscriptionTierId,
143 name,
144 description,
145 is_active,
146 )
147 .fetch_one(pool)
148 .await?;
149
150 Ok(tier)
151 }
152
153 /// Store Stripe product and price IDs on a tier after creating them on connected account.
154 #[tracing::instrument(skip_all)]
155 pub(crate) async fn update_tier_stripe_ids(
156 pool: &PgPool,
157 tier_id: SubscriptionTierId,
158 product_id: &str,
159 price_id: &str,
160 ) -> Result<()> {
161 sqlx::query!(
162 r#"
163 UPDATE subscription_tiers
164 SET stripe_product_id = $2, stripe_price_id = $3
165 WHERE id = $1
166 "#,
167 tier_id as SubscriptionTierId,
168 product_id,
169 price_id,
170 )
171 .execute(pool)
172 .await?;
173
174 Ok(())
175 }
176
177 /// Delete a subscription tier. Soft-deletes (sets is_active=false) if any
178 /// subscriptions reference it; hard-deletes otherwise.
179 ///
180 /// Uses a transaction with FOR UPDATE to prevent a TOCTOU race where a
181 /// subscription could be created between the existence check and the delete.
182 #[tracing::instrument(skip_all)]
183 pub(crate) async fn delete_subscription_tier(pool: &PgPool, id: SubscriptionTierId) -> Result<()> {
184 let mut tx = pool.begin().await?;
185
186 // Lock the tier row to serialize against concurrent subscription creation
187 sqlx::query!(
188 "SELECT id FROM subscription_tiers WHERE id = $1 FOR UPDATE",
189 id as SubscriptionTierId
190 )
191 .fetch_optional(&mut *tx)
192 .await?
193 .ok_or(sqlx::Error::RowNotFound)?;
194
195 let has_subscriptions: bool = sqlx::query_scalar!(
196 r#"SELECT EXISTS(SELECT 1 FROM subscriptions WHERE tier_id = $1) AS "exists!""#,
197 id as SubscriptionTierId,
198 )
199 .fetch_one(&mut *tx)
200 .await?;
201
202 if has_subscriptions {
203 sqlx::query!(
204 "UPDATE subscription_tiers SET is_active = false WHERE id = $1",
205 id as SubscriptionTierId
206 )
207 .execute(&mut *tx)
208 .await?;
209 } else {
210 sqlx::query!(
211 "DELETE FROM subscription_tiers WHERE id = $1",
212 id as SubscriptionTierId
213 )
214 .execute(&mut *tx)
215 .await?;
216 }
217
218 tx.commit().await?;
219 Ok(())
220 }
221
222 // ── Subscription lifecycle ──
223
224 /// Create a new subscription record after successful checkout.
225 ///
226 /// Returns `None` if the subscription already exists (duplicate webhook
227 /// or concurrent active subscription for the same user+project).
228 /// The partial UNIQUE index on `(subscriber_id, project_id) WHERE status = 'active'`
229 /// prevents multiple *active* subscriptions at the DB level.
230 ///
231 /// Single-live-row invariant: before inserting, any lingering non-active-but-live
232 /// row (`past_due`/`trialing`/`incomplete`) for the same `(subscriber, project)` is
233 /// canceled, so a resubscribe over a stale row can't leave two live rows. The access
234 /// gate (`SubscriptionGate::PREDICATE`) already ignores those statuses, so this is
235 /// data hygiene, not an access fix. The cleanup is co-located here because this is the
236 /// only INSERT path into `subscriptions`; widening the partial-unique predicate instead
237 /// would collide with `idx_subscriptions_unique` (the `(subscriber, project,
238 /// stripe_subscription_id)` triple) and break the `ON CONFLICT DO NOTHING`
239 /// duplicate-webhook idempotency this path relies on. Takes `&mut PgConnection` (not a
240 /// one-shot executor) so the cleanup and the insert run on the caller's transaction.
241 #[tracing::instrument(skip_all)]
242 pub(crate) async fn create_subscription(
243 conn: &mut sqlx::PgConnection,
244 subscriber_id: UserId,
245 tier_id: SubscriptionTierId,
246 project_id: ProjectId,
247 stripe_subscription_id: &str,
248 stripe_customer_id: &str,
249 ) -> Result<Option<DbSubscription>> {
250 sqlx::query!(
251 r#"
252 UPDATE subscriptions
253 SET status = 'canceled', canceled_at = NOW(), updated_at = NOW()
254 WHERE subscriber_id = $1 AND project_id = $2 AND item_id IS NULL
255 AND status IN ('past_due', 'trialing', 'incomplete')
256 "#,
257 subscriber_id as UserId,
258 project_id as ProjectId,
259 )
260 .execute(&mut *conn)
261 .await?;
262
263 let sub = sqlx::query_as!(
264 DbSubscription,
265 r#"
266 INSERT INTO subscriptions (subscriber_id, tier_id, project_id, stripe_subscription_id, stripe_customer_id)
267 VALUES ($1, $2, $3, $4, $5)
268 ON CONFLICT DO NOTHING
269 RETURNING id AS "id: SubscriptionId", subscriber_id AS "subscriber_id: UserId",
270 tier_id AS "tier_id: SubscriptionTierId", project_id AS "project_id: ProjectId",
271 stripe_subscription_id, stripe_customer_id,
272 status AS "status: super::SubscriptionStatus",
273 current_period_start AS "current_period_start: chrono::DateTime<chrono::Utc>",
274 current_period_end AS "current_period_end: chrono::DateTime<chrono::Utc>",
275 canceled_at AS "canceled_at: chrono::DateTime<chrono::Utc>",
276 created_at AS "created_at: chrono::DateTime<chrono::Utc>",
277 updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
278 item_id AS "item_id: ItemId",
279 paused_at AS "paused_at: chrono::DateTime<chrono::Utc>"
280 "#,
281 subscriber_id as UserId,
282 tier_id as SubscriptionTierId,
283 project_id as ProjectId,
284 stripe_subscription_id,
285 stripe_customer_id,
286 )
287 .fetch_optional(&mut *conn)
288 .await?;
289
290 Ok(sub)
291 }
292
293 /// Look up a subscription by its Stripe subscription ID.
294 #[tracing::instrument(skip_all)]
295 pub(crate) async fn get_subscription_by_stripe_id(
296 pool: &PgPool,
297 stripe_sub_id: &str,
298 ) -> Result<Option<DbSubscription>> {
299 let sub = sqlx::query_as!(
300 DbSubscription,
301 r#"
302 SELECT id AS "id: SubscriptionId", subscriber_id AS "subscriber_id: UserId",
303 tier_id AS "tier_id: SubscriptionTierId", project_id AS "project_id: ProjectId",
304 stripe_subscription_id, stripe_customer_id,
305 status AS "status: super::SubscriptionStatus",
306 current_period_start AS "current_period_start: chrono::DateTime<chrono::Utc>",
307 current_period_end AS "current_period_end: chrono::DateTime<chrono::Utc>",
308 canceled_at AS "canceled_at: chrono::DateTime<chrono::Utc>",
309 created_at AS "created_at: chrono::DateTime<chrono::Utc>",
310 updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
311 item_id AS "item_id: ItemId",
312 paused_at AS "paused_at: chrono::DateTime<chrono::Utc>"
313 FROM subscriptions WHERE stripe_subscription_id = $1
314 "#,
315 stripe_sub_id,
316 )
317 .fetch_optional(pool)
318 .await?;
319
320 Ok(sub)
321 }
322
323 // Apply a Stripe-driven status and/or period update in one guarded statement.
324 // `canceled` is terminal, an out-of-order `updated`(active) or `invoice.paid`
325 // landing after a `deleted` cannot revive the row (status) nor refresh its
326 // period, because both columns are written here under the single guard. The
327 // old split `update_subscription_status` + `update_subscription_period` (whose
328 // period half lacked the guard) are replaced by this; reactivation only ever
329 // happens at checkout via `create_subscription`, never through this path.
330 crate::db::subscription_writer::define_stripe_subscription_writer!(
331 apply_stripe_update,
332 "subscriptions",
333 DbSubscription
334 );
335
336 /// Mark a subscription as canceled.
337 #[tracing::instrument(skip_all)]
338 pub(crate) async fn cancel_subscription(
339 pool: &PgPool,
340 stripe_sub_id: &str,
341 ) -> Result<Option<DbSubscription>> {
342 let sub = sqlx::query_as!(
343 DbSubscription,
344 r#"
345 UPDATE subscriptions
346 SET status = 'canceled', canceled_at = COALESCE(canceled_at, NOW())
347 WHERE stripe_subscription_id = $1
348 RETURNING id AS "id: SubscriptionId", subscriber_id AS "subscriber_id: UserId",
349 tier_id AS "tier_id: SubscriptionTierId", project_id AS "project_id: ProjectId",
350 stripe_subscription_id, stripe_customer_id,
351 status AS "status: super::SubscriptionStatus",
352 current_period_start AS "current_period_start: chrono::DateTime<chrono::Utc>",
353 current_period_end AS "current_period_end: chrono::DateTime<chrono::Utc>",
354 canceled_at AS "canceled_at: chrono::DateTime<chrono::Utc>",
355 created_at AS "created_at: chrono::DateTime<chrono::Utc>",
356 updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
357 item_id AS "item_id: ItemId",
358 paused_at AS "paused_at: chrono::DateTime<chrono::Utc>"
359 "#,
360 stripe_sub_id,
361 )
362 .fetch_optional(pool)
363 .await?;
364
365 Ok(sub)
366 }
367
368 // ── Suspension pause/resume ──
369
370 /// Get all active subscriptions to a creator's projects (for pausing on suspension).
371 #[tracing::instrument(skip_all)]
372 pub(crate) async fn get_active_subscriptions_by_creator(
373 pool: &PgPool,
374 creator_id: UserId,
375 ) -> Result<Vec<DbSubscription>> {
376 let subs = sqlx::query_as!(
377 DbSubscription,
378 r#"
379 SELECT s.id AS "id: SubscriptionId", s.subscriber_id AS "subscriber_id: UserId",
380 s.tier_id AS "tier_id: SubscriptionTierId", s.project_id AS "project_id: ProjectId",
381 s.stripe_subscription_id, s.stripe_customer_id,
382 s.status AS "status: super::SubscriptionStatus",
383 s.current_period_start AS "current_period_start: chrono::DateTime<chrono::Utc>",
384 s.current_period_end AS "current_period_end: chrono::DateTime<chrono::Utc>",
385 s.canceled_at AS "canceled_at: chrono::DateTime<chrono::Utc>",
386 s.created_at AS "created_at: chrono::DateTime<chrono::Utc>",
387 s.updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
388 s.item_id AS "item_id: ItemId",
389 s.paused_at AS "paused_at: chrono::DateTime<chrono::Utc>"
390 FROM subscriptions s
391 WHERE s.project_id IN (SELECT id FROM projects WHERE user_id = $1)
392 AND s.status = 'active'
393 AND s.paused_at IS NULL
394 "#,
395 creator_id as UserId,
396 )
397 .fetch_all(pool)
398 .await?;
399
400 Ok(subs)
401 }
402
403 /// Mark all active subscriptions to a creator's projects as paused.
404 #[tracing::instrument(skip_all)]
405 pub(crate) async fn pause_subscriptions_for_creator(
406 pool: &PgPool,
407 creator_id: UserId,
408 ) -> Result<u64> {
409 let result = sqlx::query!(
410 r#"
411 UPDATE subscriptions SET paused_at = NOW()
412 WHERE project_id IN (SELECT id FROM projects WHERE user_id = $1)
413 AND status = 'active'
414 AND paused_at IS NULL
415 "#,
416 creator_id as UserId,
417 )
418 .execute(pool)
419 .await?;
420
421 Ok(result.rows_affected())
422 }
423
424 /// Get all paused subscriptions to a creator's projects (for cancelling on termination).
425 #[tracing::instrument(skip_all)]
426 pub(crate) async fn get_paused_subscriptions_by_creator(
427 pool: &PgPool,
428 creator_id: UserId,
429 ) -> Result<Vec<DbSubscription>> {
430 let subs = sqlx::query_as!(
431 DbSubscription,
432 r#"
433 SELECT s.id AS "id: SubscriptionId", s.subscriber_id AS "subscriber_id: UserId",
434 s.tier_id AS "tier_id: SubscriptionTierId", s.project_id AS "project_id: ProjectId",
435 s.stripe_subscription_id, s.stripe_customer_id,
436 s.status AS "status: super::SubscriptionStatus",
437 s.current_period_start AS "current_period_start: chrono::DateTime<chrono::Utc>",
438 s.current_period_end AS "current_period_end: chrono::DateTime<chrono::Utc>",
439 s.canceled_at AS "canceled_at: chrono::DateTime<chrono::Utc>",
440 s.created_at AS "created_at: chrono::DateTime<chrono::Utc>",
441 s.updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
442 s.item_id AS "item_id: ItemId",
443 s.paused_at AS "paused_at: chrono::DateTime<chrono::Utc>"
444 FROM subscriptions s
445 WHERE s.project_id IN (SELECT id FROM projects WHERE user_id = $1)
446 AND s.status = 'active'
447 AND s.paused_at IS NOT NULL
448 "#,
449 creator_id as UserId,
450 )
451 .fetch_all(pool)
452 .await?;
453
454 Ok(subs)
455 }
456
457 /// Resume all paused subscriptions for a creator's projects.
458 #[tracing::instrument(skip_all)]
459 pub(crate) async fn resume_subscriptions_for_creator(
460 pool: &PgPool,
461 creator_id: UserId,
462 ) -> Result<Vec<DbSubscription>> {
463 let subs = sqlx::query_as!(
464 DbSubscription,
465 r#"
466 UPDATE subscriptions SET paused_at = NULL
467 WHERE project_id IN (SELECT id FROM projects WHERE user_id = $1)
468 AND status = 'active'
469 AND paused_at IS NOT NULL
470 RETURNING id AS "id: SubscriptionId", subscriber_id AS "subscriber_id: UserId",
471 tier_id AS "tier_id: SubscriptionTierId", project_id AS "project_id: ProjectId",
472 stripe_subscription_id, stripe_customer_id,
473 status AS "status: super::SubscriptionStatus",
474 current_period_start AS "current_period_start: chrono::DateTime<chrono::Utc>",
475 current_period_end AS "current_period_end: chrono::DateTime<chrono::Utc>",
476 canceled_at AS "canceled_at: chrono::DateTime<chrono::Utc>",
477 created_at AS "created_at: chrono::DateTime<chrono::Utc>",
478 updated_at AS "updated_at: chrono::DateTime<chrono::Utc>",
479 item_id AS "item_id: ItemId",
480 paused_at AS "paused_at: chrono::DateTime<chrono::Utc>"
481 "#,
482 creator_id as UserId,
483 )
484 .fetch_all(pool)
485 .await?;
486
487 Ok(subs)
488 }
489
490 // ── Access control ──
491
492 /// SQL predicate identifying a `subscriptions` row that currently grants access
493 /// to its scope. The `current_period_end` clause is defense-in-depth against a
494 /// missed/delayed `customer.subscription.deleted` webhook, `status = 'active'`
495 /// alone trusts Stripe to push the cancellation promptly.
496 ///
497 /// What a subscription access check is scoped to. Both arms run the SAME sealed
498 /// predicate inside [`gate`], so a project gate and an item gate cannot diverge.
499 #[derive(Debug, Clone, Copy)]
500 pub enum SubscriptionScope {
501 Project(ProjectId),
502 Item(super::ItemId),
503 }
504
505 pub(crate) use gate::SubscriptionGate;
506
507 /// Sealed home of the "does a subscription grant access right now" predicate.
508 ///
509 /// The predicate text lives in exactly ONE place, [`SubscriptionGate`]'s
510 /// private `PREDICATE` associated const, and is unreachable from the rest of
511 /// this module, let alone other modules. The only way to learn "this user has
512 /// access" is [`SubscriptionGate::check`] (or [`SubscriptionGate::accessible_item_ids`]
513 /// for the batch shape), each of which runs that predicate. A `SubscriptionGate`
514 /// value is a witness: its field is private and there is no public constructor,
515 /// so access-granting code can neither fabricate one nor hand-write a divergent
516 /// gate.
517 ///
518 /// Payments S1 / CHRONIC 2: the predicate used to be a shareable `&str` const,
519 /// and item gates drifted by dropping the `current_period_end` clause. A const
520 /// is copy-pasteable; a private associated const inside a sealed submodule is
521 /// not. Sealing it here makes the divergence unwritable, not merely discouraged.
522 mod gate {
523 use super::SubscriptionScope;
524 use crate::db::{ItemId, UserId};
525 use crate::error::Result;
526 use sqlx::PgPool;
527 use std::collections::HashMap;
528
529 /// Proof that a subscription currently grants access. Constructible ONLY via
530 /// [`SubscriptionGate::check`], the private `()` field seals the type so no
531 /// other code can mint one.
532 #[derive(Debug, Clone, Copy)]
533 pub struct SubscriptionGate(());
534
535 impl SubscriptionGate {
536 /// The single source of truth for "grants access right now". Private to
537 /// this submodule: nothing outside can read it as a string, so it cannot
538 /// be copy-pasted into a divergent query. (Compile-time constant, never
539 /// user input, so the `format!` interpolation is injection-safe; the
540 /// `$N` placeholders stay bound.)
541 const PREDICATE: &'static str = "status = 'active' AND paused_at IS NULL \
542 AND (current_period_end IS NULL OR current_period_end > NOW())";
543
544 /// Does `user_id` hold a subscription that currently grants access to
545 /// `scope`? Returns `Some(gate)` iff so, the sole gate constructor and
546 /// the single entry point for project- and item-level access checks.
547 #[tracing::instrument(skip_all)]
548 pub async fn check(
549 pool: &PgPool,
550 user_id: UserId,
551 scope: SubscriptionScope,
552 ) -> Result<Option<SubscriptionGate>> {
553 // runtime-checked: dynamically-built SQL, the access predicate is
554 // interpolated from the sealed `PREDICATE` const via format!, so the
555 // statement text isn't a compile-time literal the macro can verify.
556 let exists: bool = match scope {
557 SubscriptionScope::Project(project_id) => {
558 sqlx::query_scalar(&format!(
559 "SELECT EXISTS(SELECT 1 FROM subscriptions \
560 WHERE subscriber_id = $1 AND project_id = $2 AND {})",
561 Self::PREDICATE
562 ))
563 .bind(user_id)
564 .bind(project_id)
565 .fetch_one(pool)
566 .await?
567 }
568 SubscriptionScope::Item(item_id) => {
569 sqlx::query_scalar(&format!(
570 "SELECT EXISTS(SELECT 1 FROM subscriptions \
571 WHERE subscriber_id = $1 AND item_id = $2 AND {})",
572 Self::PREDICATE
573 ))
574 .bind(user_id)
575 .bind(item_id)
576 .fetch_one(pool)
577 .await?
578 }
579 };
580
581 Ok(exists.then_some(SubscriptionGate(())))
582 }
583
584 /// Every item ID `user_id` currently has access to via subscription
585 /// (batch gate). Runs the same sealed predicate as [`check`], so the
586 /// batch path cannot drift from the single-item gate.
587 #[tracing::instrument(skip_all)]
588 pub async fn accessible_item_ids(pool: &PgPool, user_id: UserId) -> Result<Vec<ItemId>> {
589 // runtime-checked: dynamically-built SQL, the access predicate is
590 // interpolated from the sealed `PREDICATE` const via format!, so the
591 // statement text isn't a compile-time literal the macro can verify.
592 let item_ids: Vec<ItemId> = sqlx::query_scalar(&format!(
593 "SELECT DISTINCT item_id FROM subscriptions \
594 WHERE subscriber_id = $1 AND item_id IS NOT NULL AND {}",
595 Self::PREDICATE
596 ))
597 .bind(user_id)
598 .fetch_all(pool)
599 .await?;
600
601 Ok(item_ids)
602 }
603
604 /// Map of every item `user_id` currently has subscription access to →
605 /// its access proof. The witness-bearing batch shape used by the project
606 /// page, where each item's [`AccessContext`](crate::pricing::AccessContext)
607 /// needs its own gate. Runs the sealed predicate once.
608 #[tracing::instrument(skip_all)]
609 #[allow(
610 clippy::zero_sized_map_values,
611 reason = "SubscriptionGate is a deliberate zero-sized capability witness (sealed constructor); the map value carries type-level proof of access, not data, so a HashSet would lose the witness semantics"
612 )]
613 pub async fn subscribed_item_gates(
614 pool: &PgPool,
615 user_id: UserId,
616 ) -> Result<HashMap<ItemId, SubscriptionGate>> {
617 let ids = Self::accessible_item_ids(pool, user_id).await?;
618 Ok(ids
619 .into_iter()
620 .map(|id| (id, SubscriptionGate(())))
621 .collect())
622 }
623
624 /// Test-only constructor. Real gates can only be minted by running the
625 /// predicate against the DB; unit tests (e.g. `pricing`) need to
626 /// fabricate the "access granted" state without a database. Gated to
627 /// test builds so production code still cannot forge a witness.
628 #[cfg(test)]
629 pub(crate) fn test_witness() -> Self {
630 SubscriptionGate(())
631 }
632 }
633 }
634
635 /// Does `user_id` hold a subscription that currently grants access to `scope`?
636 ///
637 /// Thin boolean wrapper over the sealed [`SubscriptionGate::check`]; prefer
638 /// taking the [`SubscriptionGate`] witness directly where a proof of access is
639 /// useful downstream.
640 #[tracing::instrument(skip_all)]
641 pub(crate) async fn has_access(
642 pool: &PgPool,
643 user_id: UserId,
644 scope: SubscriptionScope,
645 ) -> Result<bool> {
646 Ok(SubscriptionGate::check(pool, user_id, scope)
647 .await?
648 .is_some())
649 }
650
651 /// Get user subscriptions joined with project and tier data (for library display).
652 #[tracing::instrument(skip_all)]
653 pub(crate) async fn get_user_subscriptions_with_details(
654 pool: &PgPool,
655 user_id: UserId,
656 ) -> Result<Vec<DbUserSubscriptionRow>> {
657 let rows = sqlx::query_as!(
658 DbUserSubscriptionRow,
659 r#"
660 SELECT s.id AS "id: SubscriptionId", s.project_id AS "project_id!: ProjectId",
661 p.title AS project_title, p.slug AS "project_slug: super::Slug",
662 t.name AS tier_name, t.price_cents, s.status AS "status: super::SubscriptionStatus",
663 s.current_period_end AS "current_period_end: chrono::DateTime<chrono::Utc>",
664 s.stripe_subscription_id,
665 u.settlement_currency AS "settlement_currency: crate::currency::SettlementCurrency"
666 FROM subscriptions s
667 JOIN projects p ON p.id = s.project_id
668 JOIN users u ON u.id = p.user_id
669 JOIN subscription_tiers t ON t.id = s.tier_id
670 WHERE s.subscriber_id = $1
671 ORDER BY s.created_at DESC
672 LIMIT 1000
673 "#,
674 user_id as UserId,
675 )
676 .fetch_all(pool)
677 .await?;
678
679 Ok(rows)
680 }
681
682 /// Get the number of active subscribers to a project (for dashboard display).
683 ///
684 /// NOT an access gate, this is a creator-facing headcount, so it deliberately
685 /// counts `status = 'active'` rows regardless of `current_period_end` (a sub in
686 /// its grace window is still a subscriber). Do not "align" it with
687 /// [`GRANTS_ACCESS_PREDICATE`]; the divergence here is intentional.
688 #[tracing::instrument(skip_all)]
689 pub(crate) async fn get_project_subscriber_count(
690 pool: &PgPool,
691 project_id: ProjectId,
692 ) -> Result<i64> {
693 let count: i64 = sqlx::query_scalar!(
694 r#"SELECT COUNT(*) AS "count!" FROM subscriptions WHERE project_id = $1 AND status = 'active' AND paused_at IS NULL"#,
695 project_id as ProjectId,
696 )
697 .fetch_one(pool)
698 .await?;
699
700 Ok(count)
701 }
702
703 // ── Export ──
704
705 /// Export all subscribers across a creator's projects.
706 ///
707 /// Returns username, display_name, tier name, subscription status, and when.
708 #[tracing::instrument(skip_all)]
709 /// One page of a creator's project subscribers for CSV export, newest first.
710 /// Paginated for bounded-memory streaming (ultra-fuzz Run 4 S1); stable
711 /// `(created_at, id)` ordering keeps OFFSET batches consistent.
712 pub(crate) async fn get_project_subscribers_for_export_page(
713 pool: &PgPool,
714 user_id: UserId,
715 limit: i64,
716 offset: i64,
717 ) -> Result<Vec<SubscriberExportRow>> {
718 let rows = sqlx::query_as!(
719 SubscriberExportRow,
720 r#"
721 SELECT u.username, u.display_name, t.name AS tier_name,
722 s.status AS "status: super::SubscriptionStatus",
723 s.created_at AS "created_at: chrono::DateTime<chrono::Utc>"
724 FROM subscriptions s
725 JOIN users u ON u.id = s.subscriber_id
726 JOIN subscription_tiers t ON t.id = s.tier_id
727 WHERE s.project_id IN (SELECT id FROM projects WHERE user_id = $1)
728 ORDER BY s.created_at DESC, s.id DESC
729 LIMIT $2 OFFSET $3
730 "#,
731 user_id as UserId,
732 limit,
733 offset,
734 )
735 .fetch_all(pool)
736 .await?;
737
738 Ok(rows)
739 }
740
741 /// Export all subscriptions across a creator's projects with full detail.
742 ///
743 /// Returns project name, tier name, price, subscriber username, status,
744 /// billing period dates, and cancellation date.
745 #[tracing::instrument(skip_all)]
746 /// One page of a creator's subscriptions for CSV export, newest first.
747 /// Paginated for bounded-memory streaming (ultra-fuzz Run 4 S1); stable
748 /// `(created_at, id)` ordering keeps OFFSET batches consistent.
749 pub(crate) async fn get_subscriptions_for_export_page(
750 pool: &PgPool,
751 user_id: UserId,
752 limit: i64,
753 offset: i64,
754 ) -> Result<Vec<SubscriptionExportRow>> {
755 let rows = sqlx::query_as!(
756 SubscriptionExportRow,
757 r#"
758 SELECT p.title AS project_title, t.name AS tier_name, t.price_cents,
759 u.username, s.status AS "status: super::SubscriptionStatus",
760 s.current_period_start AS "current_period_start: chrono::DateTime<chrono::Utc>",
761 s.current_period_end AS "current_period_end: chrono::DateTime<chrono::Utc>",
762 s.canceled_at AS "canceled_at: chrono::DateTime<chrono::Utc>",
763 s.created_at AS "created_at: chrono::DateTime<chrono::Utc>"
764 FROM subscriptions s
765 JOIN users u ON u.id = s.subscriber_id
766 JOIN subscription_tiers t ON t.id = s.tier_id
767 JOIN projects p ON p.id = s.project_id
768 WHERE s.project_id IN (SELECT id FROM projects WHERE user_id = $1)
769 ORDER BY s.created_at DESC, s.id DESC
770 LIMIT $2 OFFSET $3
771 "#,
772 user_id as UserId,
773 limit,
774 offset,
775 )
776 .fetch_all(pool)
777 .await?;
778
779 Ok(rows)
780 }
781
782 // ── Event log ──
783
784 /// Log a subscription webhook event for debugging, idempotency, and revenue
785 /// reconciliation. The `ON CONFLICT (stripe_event_id) DO NOTHING` makes a
786 /// redelivered event a silent no-op, never an error, so an `Err` return is
787 /// always a genuine DB failure, which callers log at `error!` (a dropped
788 /// reconciliation row is worth alerting on, not burying at `warn!`).
789 ///
790 /// `event_type` is an [`MnwEventName`], not a string. The names in this table
791 /// are MNW's own vocabulary rather than Stripe's, and they were previously
792 /// spelled out at 26 call sites across three files, where a typo produced a row
793 /// nobody would ever match on. Taking the enum makes the spelling
794 /// unmisspellable and puts every name in one place.
795 #[tracing::instrument(skip_all)]
796 pub(crate) async fn log_subscription_event(
797 pool: &PgPool,
798 subscription_id: Option<SubscriptionId>,
799 stripe_event_id: &str,
800 event_type: crate::payments::MnwEventName,
801 payload: &serde_json::Value,
802 ) -> Result<()> {
803 let event_type = event_type.as_str();
804 sqlx::query!(
805 r#"
806 INSERT INTO subscription_events (subscription_id, stripe_event_id, event_type, payload)
807 VALUES ($1, $2, $3, $4)
808 ON CONFLICT (stripe_event_id) DO NOTHING
809 "#,
810 subscription_id as Option<SubscriptionId>,
811 stripe_event_id,
812 event_type,
813 payload as &serde_json::Value,
814 )
815 .execute(pool)
816 .await?;
817
818 Ok(())
819 }
820