Skip to main content

max / makenotwork

9.3 KB · 274 lines History Blame Raw
1 use super::{CreatorTier, DbCreatorSubscription, PgPool, Result, UserId};
2
3 /// Create or reactivate a creator tier subscription record.
4 ///
5 /// Uses ON CONFLICT DO UPDATE on the user_id unique index to handle
6 /// both duplicate webhooks and re-subscription after cancellation.
7 /// Returns `None` if the row already existed with the same stripe_subscription_id
8 /// (duplicate webhook), `Some` if this was a fresh insert or a re-subscription
9 /// with a different subscription ID.
10 #[tracing::instrument(skip_all)]
11 pub async fn create_creator_subscription<'e>(
12 executor: impl sqlx::PgExecutor<'e>,
13 user_id: UserId,
14 stripe_subscription_id: &str,
15 stripe_customer_id: &str,
16 tier: CreatorTier,
17 ) -> Result<Option<DbCreatorSubscription>> {
18 // Use WHERE clause on the DO UPDATE to only update if the subscription_id
19 // is different (new subscription) or status is not already active.
20 // When the WHERE fails, DO UPDATE becomes a no-op and RETURNING yields no row.
21 let sub = sqlx::query_as::<_, DbCreatorSubscription>(
22 r"
23 INSERT INTO creator_subscriptions (user_id, stripe_subscription_id, stripe_customer_id, tier)
24 VALUES ($1, $2, $3, $4)
25 ON CONFLICT (user_id) DO UPDATE
26 SET stripe_subscription_id = EXCLUDED.stripe_subscription_id,
27 stripe_customer_id = EXCLUDED.stripe_customer_id,
28 tier = EXCLUDED.tier,
29 status = 'active',
30 canceled_at = NULL,
31 grace_enforced_at = NULL
32 WHERE creator_subscriptions.stripe_subscription_id != EXCLUDED.stripe_subscription_id
33 OR creator_subscriptions.status != 'active'
34 RETURNING *
35 ",
36 )
37 .bind(user_id)
38 .bind(stripe_subscription_id)
39 .bind(stripe_customer_id)
40 .bind(tier)
41 .fetch_optional(executor)
42 .await?;
43
44 Ok(sub)
45 }
46
47 /// Look up a creator subscription by its Stripe subscription ID.
48 #[tracing::instrument(skip_all)]
49 pub async fn get_creator_sub_by_stripe_id(
50 pool: &PgPool,
51 stripe_subscription_id: &str,
52 ) -> Result<Option<DbCreatorSubscription>> {
53 let sub = sqlx::query_as::<_, DbCreatorSubscription>(
54 "SELECT * FROM creator_subscriptions WHERE stripe_subscription_id = $1",
55 )
56 .bind(stripe_subscription_id)
57 .fetch_optional(pool)
58 .await?;
59
60 Ok(sub)
61 }
62
63 /// Get a user's creator subscription (any status).
64 #[tracing::instrument(skip_all)]
65 pub async fn get_creator_sub_by_user(
66 pool: &PgPool,
67 user_id: UserId,
68 ) -> Result<Option<DbCreatorSubscription>> {
69 let sub = sqlx::query_as::<_, DbCreatorSubscription>(
70 "SELECT * FROM creator_subscriptions WHERE user_id = $1",
71 )
72 .bind(user_id)
73 .fetch_optional(pool)
74 .await?;
75
76 Ok(sub)
77 }
78
79 /// Get the active creator tier for a user (None if no active subscription).
80 #[tracing::instrument(skip_all)]
81 pub async fn get_active_creator_tier(
82 pool: &PgPool,
83 user_id: UserId,
84 ) -> Result<Option<CreatorTier>> {
85 let tier = sqlx::query_scalar::<_, String>(
86 "SELECT tier FROM creator_subscriptions WHERE user_id = $1 AND status = 'active'",
87 )
88 .bind(user_id)
89 .fetch_optional(pool)
90 .await?;
91
92 match tier {
93 None => Ok(None),
94 // A present-but-unparseable tier means the DB holds a tier string the
95 // enum doesn't know (enum drift). Silently mapping that to `None` would
96 // strip a paying creator's entitlements with no signal, surface it
97 // loudly instead of swallowing it (Run 21). The `enum_drift` test and the
98 // DB CHECK constraint make this unreachable in practice.
99 Some(t) => match t.parse::<CreatorTier>() {
100 Ok(parsed) => Ok(Some(parsed)),
101 Err(_) => {
102 tracing::error!(
103 user_id = %user_id, tier = %t,
104 "active creator subscription has an unrecognized tier string (enum drift)"
105 );
106 Err(crate::error::AppError::Internal(anyhow::anyhow!(
107 "unrecognized creator tier '{t}' for user {user_id}"
108 )))
109 }
110 },
111 }
112 }
113
114 // Apply a Stripe-driven status and/or period update in one guarded statement.
115 // `canceled` is terminal; reactivation runs through `create_creator_subscription`'s
116 // `ON CONFLICT (user_id) DO UPDATE` at checkout, never through this path. Replaces
117 // the old split status/period setters (the period half lacked the guard). See
118 // `crate::db::subscription_writer`.
119 crate::db::subscription_writer::define_stripe_subscription_writer!(
120 apply_stripe_update,
121 "creator_subscriptions",
122 DbCreatorSubscription
123 );
124
125 /// Cancel a creator subscription (set status + canceled_at).
126 #[tracing::instrument(skip_all)]
127 pub async fn cancel_creator_sub(
128 pool: &PgPool,
129 stripe_subscription_id: &str,
130 ) -> Result<Option<DbCreatorSubscription>> {
131 let sub = sqlx::query_as::<_, DbCreatorSubscription>(
132 r"
133 UPDATE creator_subscriptions
134 SET status = 'canceled', canceled_at = COALESCE(canceled_at, NOW())
135 WHERE stripe_subscription_id = $1
136 RETURNING *
137 ",
138 )
139 .bind(stripe_subscription_id)
140 .fetch_optional(pool)
141 .await?;
142
143 Ok(sub)
144 }
145
146 /// Sync the users.creator_tier column from the subscription status.
147 /// Called after checkout/update/cancel to keep the denormalized column in sync.
148 #[tracing::instrument(skip_all)]
149 pub async fn sync_user_creator_tier(pool: &PgPool, user_id: UserId) -> Result<()> {
150 sqlx::query(
151 r"
152 UPDATE users SET creator_tier = (
153 SELECT tier FROM creator_subscriptions
154 WHERE user_id = $1 AND status = 'active'
155 )
156 WHERE id = $1
157 ",
158 )
159 .bind(user_id)
160 .execute(pool)
161 .await?;
162
163 Ok(())
164 }
165
166 /// Get user IDs of creators with canceled subscriptions 30+ days ago
167 /// whose items have not yet been hidden.
168 #[tracing::instrument(skip_all)]
169 pub async fn get_expired_grace_creators(pool: &PgPool) -> Result<Vec<UserId>> {
170 // Bounded batch per call. The scheduler enforces these inline on the tick
171 // (two DB round-trips per creator), so an unbounded result set would let a
172 // backlog stall the tick. `grace_enforced_at` is set as each creator is
173 // processed, so successive ticks drain the rest; ORDER BY oldest-first keeps
174 // it deterministic and starvation-free.
175 let ids: Vec<UserId> = sqlx::query_scalar(
176 r"
177 SELECT user_id FROM creator_subscriptions
178 WHERE status = 'canceled'
179 AND canceled_at IS NOT NULL
180 AND canceled_at < NOW() - INTERVAL '30 days'
181 AND grace_enforced_at IS NULL
182 ORDER BY canceled_at ASC
183 LIMIT 200
184 ",
185 )
186 .fetch_all(pool)
187 .await?;
188
189 Ok(ids)
190 }
191
192 /// Stamp `grace_enforced_at` for all given creators in one statement, paired with
193 /// `items::hide_all_items_for_users` on the post-grace sweep (Perf-S4, Run 9).
194 /// No-op on an empty slice.
195 #[tracing::instrument(skip_all)]
196 pub async fn mark_grace_enforced_batch(pool: &PgPool, user_ids: &[UserId]) -> Result<()> {
197 if user_ids.is_empty() {
198 return Ok(());
199 }
200 sqlx::query(
201 "UPDATE creator_subscriptions SET grace_enforced_at = NOW() WHERE user_id = ANY($1)",
202 )
203 .bind(user_ids)
204 .execute(pool)
205 .await?;
206
207 Ok(())
208 }
209
210 /// Count fully-paying creators, `status = 'active'` only.
211 ///
212 /// Excludes trialing (free trial), past_due (payment failed but not yet
213 /// canceled), canceled-in-grace (winding down), and incomplete states.
214 /// This is the number that goes on the runway disclosure as "paying
215 /// creators today": revenue-bearing seats, no fudge.
216 #[tracing::instrument(skip_all)]
217 pub async fn count_active_paying(pool: &PgPool) -> Result<i64> {
218 let count: (i64,) =
219 sqlx::query_as("SELECT COUNT(*) FROM creator_subscriptions WHERE status = 'active'")
220 .fetch_one(pool)
221 .await?;
222 Ok(count.0)
223 }
224
225 /// Count creators in a trial or 30-day cancellation grace period.
226 ///
227 /// These are not revenue-bearing today but represent the near-term
228 /// pipeline: trialing seats may convert, grace seats may resubscribe
229 /// before enforcement. Disclosed as a secondary number on the runway
230 /// surface so the headline `count_active_paying` stays strict.
231 #[tracing::instrument(skip_all)]
232 pub async fn count_trialing_or_grace(pool: &PgPool) -> Result<i64> {
233 let count: (i64,) = sqlx::query_as(
234 r"
235 SELECT COUNT(*) FROM creator_subscriptions
236 WHERE status = 'trialing'
237 OR (
238 status = 'canceled'
239 AND canceled_at IS NOT NULL
240 AND canceled_at > NOW() - INTERVAL '30 days'
241 AND grace_enforced_at IS NULL
242 )
243 ",
244 )
245 .fetch_one(pool)
246 .await?;
247 Ok(count.0)
248 }
249
250 /// Check whether a user is in the 30-day cancellation grace period.
251 ///
252 /// Returns `true` if the subscription is canceled but within 30 days of cancellation
253 /// and enforcement has not yet been applied.
254 #[tracing::instrument(skip_all)]
255 pub async fn is_in_grace_period(pool: &PgPool, user_id: UserId) -> Result<bool> {
256 let in_grace: bool = sqlx::query_scalar(
257 r"
258 SELECT EXISTS(
259 SELECT 1 FROM creator_subscriptions
260 WHERE user_id = $1
261 AND status = 'canceled'
262 AND canceled_at IS NOT NULL
263 AND canceled_at > NOW() - INTERVAL '30 days'
264 AND grace_enforced_at IS NULL
265 )
266 ",
267 )
268 .bind(user_id)
269 .fetch_one(pool)
270 .await?;
271
272 Ok(in_grace)
273 }
274