Skip to main content

max / makenotwork

5.2 KB · 184 lines History Blame Raw
1 //! Fan+ consumer subscription queries.
2
3 use chrono::{DateTime, Utc};
4 use sqlx::PgPool;
5
6 use super::enums::SubscriptionStatus;
7 use super::id_types::*;
8 use super::models::DbFanPlusSubscription;
9 use crate::error::Result;
10
11 /// Create or reactivate a Fan+ subscription record.
12 ///
13 /// Uses ON CONFLICT DO UPDATE on the user_id unique constraint to handle
14 /// both duplicate webhooks and re-subscription after cancellation.
15 #[tracing::instrument(skip_all)]
16 pub async fn create_fan_plus_subscription<'e>(
17 executor: impl sqlx::PgExecutor<'e>,
18 user_id: UserId,
19 stripe_subscription_id: &str,
20 stripe_customer_id: &str,
21 ) -> Result<Option<DbFanPlusSubscription>> {
22 let sub = sqlx::query_as::<_, DbFanPlusSubscription>(
23 r#"
24 INSERT INTO fan_plus_subscriptions (user_id, stripe_subscription_id, stripe_customer_id)
25 VALUES ($1, $2, $3)
26 ON CONFLICT (user_id) DO UPDATE
27 SET stripe_subscription_id = EXCLUDED.stripe_subscription_id,
28 stripe_customer_id = EXCLUDED.stripe_customer_id,
29 status = 'active',
30 canceled_at = NULL
31 RETURNING *
32 "#,
33 )
34 .bind(user_id)
35 .bind(stripe_subscription_id)
36 .bind(stripe_customer_id)
37 .fetch_optional(executor)
38 .await?;
39
40 Ok(sub)
41 }
42
43 /// Look up a Fan+ subscription by its Stripe subscription ID.
44 #[tracing::instrument(skip_all)]
45 pub async fn get_fan_plus_by_stripe_id(
46 pool: &PgPool,
47 stripe_subscription_id: &str,
48 ) -> Result<Option<DbFanPlusSubscription>> {
49 let sub = sqlx::query_as::<_, DbFanPlusSubscription>(
50 "SELECT * FROM fan_plus_subscriptions WHERE stripe_subscription_id = $1",
51 )
52 .bind(stripe_subscription_id)
53 .fetch_optional(pool)
54 .await?;
55
56 Ok(sub)
57 }
58
59 /// Update the status of a Fan+ subscription.
60 /// Sets canceled_at when transitioning to canceled, preserving existing value.
61 #[tracing::instrument(skip_all)]
62 pub async fn update_fan_plus_status<'e>(
63 executor: impl sqlx::PgExecutor<'e>,
64 stripe_subscription_id: &str,
65 status: SubscriptionStatus,
66 ) -> Result<Option<DbFanPlusSubscription>> {
67 let sub = sqlx::query_as::<_, DbFanPlusSubscription>(
68 r#"
69 UPDATE fan_plus_subscriptions
70 SET status = $2,
71 canceled_at = CASE
72 WHEN $2 = 'canceled' THEN COALESCE(canceled_at, NOW())
73 ELSE canceled_at
74 END
75 WHERE stripe_subscription_id = $1
76 RETURNING *
77 "#,
78 )
79 .bind(stripe_subscription_id)
80 .bind(status)
81 .fetch_optional(executor)
82 .await?;
83
84 Ok(sub)
85 }
86
87 /// Update the billing period of a Fan+ subscription.
88 #[tracing::instrument(skip_all)]
89 pub async fn update_fan_plus_period<'e>(
90 executor: impl sqlx::PgExecutor<'e>,
91 stripe_subscription_id: &str,
92 start: DateTime<Utc>,
93 end: DateTime<Utc>,
94 ) -> Result<()> {
95 sqlx::query(
96 r#"
97 UPDATE fan_plus_subscriptions
98 SET current_period_start = $2, current_period_end = $3
99 WHERE stripe_subscription_id = $1
100 "#,
101 )
102 .bind(stripe_subscription_id)
103 .bind(start)
104 .bind(end)
105 .execute(executor)
106 .await?;
107
108 Ok(())
109 }
110
111 /// Cancel a Fan+ subscription (set status + canceled_at).
112 #[tracing::instrument(skip_all)]
113 pub async fn cancel_fan_plus(
114 pool: &PgPool,
115 stripe_subscription_id: &str,
116 ) -> Result<Option<DbFanPlusSubscription>> {
117 let sub = sqlx::query_as::<_, DbFanPlusSubscription>(
118 r#"
119 UPDATE fan_plus_subscriptions
120 SET status = 'canceled', canceled_at = COALESCE(canceled_at, NOW())
121 WHERE stripe_subscription_id = $1
122 RETURNING *
123 "#,
124 )
125 .bind(stripe_subscription_id)
126 .fetch_optional(pool)
127 .await?;
128
129 Ok(sub)
130 }
131
132 /// Check whether a user has an active Fan+ subscription.
133 #[tracing::instrument(skip_all)]
134 pub async fn is_fan_plus_active(pool: &PgPool, user_id: UserId) -> Result<bool> {
135 let exists = sqlx::query_scalar::<_, bool>(
136 "SELECT EXISTS(SELECT 1 FROM fan_plus_subscriptions WHERE user_id = $1 AND status = 'active')",
137 )
138 .bind(user_id)
139 .fetch_one(pool)
140 .await?;
141
142 Ok(exists)
143 }
144
145 /// Mark a Fan+ subscription as scheduled to cancel at period end (or undo).
146 ///
147 /// Sets the local flag; Stripe is the source of truth and re-asserts it via
148 /// the `customer.subscription.updated` webhook. Called from the dashboard
149 /// Cancel/Resume buttons and from the webhook handler.
150 #[tracing::instrument(skip_all)]
151 pub async fn set_cancel_at_period_end(
152 pool: &PgPool,
153 stripe_subscription_id: &str,
154 cancel: bool,
155 ) -> Result<Option<DbFanPlusSubscription>> {
156 let sub = sqlx::query_as::<_, DbFanPlusSubscription>(
157 "UPDATE fan_plus_subscriptions
158 SET cancel_at_period_end = $2
159 WHERE stripe_subscription_id = $1
160 RETURNING *",
161 )
162 .bind(stripe_subscription_id)
163 .bind(cancel)
164 .fetch_optional(pool)
165 .await?;
166 Ok(sub)
167 }
168
169 /// Get a user's Fan+ subscription (any status).
170 #[tracing::instrument(skip_all)]
171 pub async fn get_fan_plus_by_user(
172 pool: &PgPool,
173 user_id: UserId,
174 ) -> Result<Option<DbFanPlusSubscription>> {
175 let sub = sqlx::query_as::<_, DbFanPlusSubscription>(
176 "SELECT * FROM fan_plus_subscriptions WHERE user_id = $1",
177 )
178 .bind(user_id)
179 .fetch_optional(pool)
180 .await?;
181
182 Ok(sub)
183 }
184