Skip to main content

max / makenotwork

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