Skip to main content

max / makenotwork

8.9 KB · 248 lines History Blame Raw
1 //! SyncKit end-user app subscriptions (the per-user subs billed on MNW's own
2 //! Stripe account, distinct from developer billing): create/read a user's app
3 //! subscription and gate internal writes on an active subscription.
4
5 use chrono::{DateTime, Utc};
6 use sqlx::PgPool;
7
8 use crate::db::{SyncAppId, UserId};
9 use crate::error::Result;
10
11 /// Parameters for inserting a new app sync subscription from a webhook event.
12 pub struct NewAppSyncSubscription<'a> {
13 pub user_id: UserId,
14 pub app_id: SyncAppId,
15 pub stripe_subscription_id: &'a str,
16 pub stripe_customer_id: &'a str,
17 /// Billing interval ("monthly" / "annual"). Persisted in the `tier` column
18 /// (kept lowercase for that legacy name) so cap-change handlers know which
19 /// interval the user is on without round-tripping to Stripe.
20 pub interval: &'a str,
21 pub storage_limit_bytes: i64,
22 }
23
24 /// End-user subscription to an app's cloud sync (rows in `app_sync_subscriptions`).
25 #[derive(Debug, sqlx::FromRow)]
26 pub struct DbAppSyncSubscription {
27 pub stripe_subscription_id: String,
28 pub interval: String,
29 pub status: String,
30 pub storage_limit_bytes: Option<i64>,
31 pub pending_storage_limit_bytes: Option<i64>,
32 pub current_period_end: Option<DateTime<Utc>>,
33 }
34
35 /// Whether writes (push / blob upload) are allowed for this `(app, user)`.
36 ///
37 /// Non-internal apps are developer-billed: the developer pays for the whole
38 /// app's storage budget, so writes are always allowed here (storage caps are
39 /// enforced separately on the blob path). First-party (`is_internal`) apps use
40 /// the end-user subscription model, paid-only sync, so a write is allowed
41 /// only when the user holds an `active` row in `app_sync_subscriptions`.
42 ///
43 /// A missing app resolves to `false` (deny); `app_id` always comes from a
44 /// validated JWT at the call sites, so this is a defensive default.
45 #[tracing::instrument(skip_all)]
46 pub async fn internal_write_allowed(
47 pool: &PgPool,
48 app_id: SyncAppId,
49 user_id: UserId,
50 ) -> Result<bool> {
51 let row: Option<(bool, bool)> = sqlx::query_as(
52 r"
53 SELECT
54 sa.is_internal,
55 EXISTS(
56 SELECT 1 FROM app_sync_subscriptions s
57 WHERE s.app_id = sa.id AND s.user_id = $2 AND s.status = 'active'
58 ) AS has_active_sub
59 FROM sync_apps sa
60 WHERE sa.id = $1
61 ",
62 )
63 .bind(app_id)
64 .bind(user_id)
65 .fetch_optional(pool)
66 .await?;
67 Ok(matches!(row, Some((is_internal, has_sub)) if !is_internal || has_sub))
68 }
69
70 /// Look up the active subscription a user has on an app, if any.
71 /// Returns `None` if the user has never subscribed or the subscription was deleted.
72 #[tracing::instrument(skip_all)]
73 pub async fn get_user_app_subscription(
74 pool: &PgPool,
75 user_id: UserId,
76 app_id: SyncAppId,
77 ) -> Result<Option<DbAppSyncSubscription>> {
78 let row = sqlx::query_as::<_, DbAppSyncSubscription>(
79 r"
80 SELECT stripe_subscription_id,
81 tier AS interval,
82 status,
83 storage_limit_bytes,
84 pending_storage_limit_bytes,
85 current_period_end
86 FROM app_sync_subscriptions
87 WHERE user_id = $1 AND app_id = $2
88 ",
89 )
90 .bind(user_id)
91 .bind(app_id)
92 .fetch_optional(pool)
93 .await?;
94 Ok(row)
95 }
96
97 /// Insert or reactivate an app sync subscription. Returns `Ok(true)` if a row
98 /// was inserted or reactivated, `Ok(false)` if an identical active subscription
99 /// already existed for this (user, app) pair (idempotent webhook replay).
100 ///
101 /// ON CONFLICT DO UPDATE (not DO NOTHING) so a paid re-subscribe after a
102 /// cancellation reactivates the row AT CHECKOUT, deterministically, the old
103 /// DO NOTHING left a re-subscribed user `canceled` until a later
104 /// `customer.subscription.updated`(active) happened to arrive (Run #12 MINOR).
105 /// The guard WHERE makes a duplicate checkout webhook for an unchanged active
106 /// row a no-op, mirroring `creator_tiers::create_creator_subscription`. Pairing
107 /// this with the terminal guard on `update_app_sync_subscription_status` keeps
108 /// reactivation on the checkout path while the status-update webhooks can't
109 /// revive a canceled row.
110 #[tracing::instrument(skip_all)]
111 pub async fn create_app_sync_subscription(
112 pool: &PgPool,
113 sub: &NewAppSyncSubscription<'_>,
114 ) -> Result<bool> {
115 let result = sqlx::query(
116 r"
117 INSERT INTO app_sync_subscriptions
118 (user_id, app_id, stripe_subscription_id, stripe_customer_id,
119 tier, status, storage_limit_bytes)
120 VALUES ($1, $2, $3, $4, $5, 'active', $6)
121 ON CONFLICT (user_id, app_id) DO UPDATE
122 SET stripe_subscription_id = EXCLUDED.stripe_subscription_id,
123 stripe_customer_id = EXCLUDED.stripe_customer_id,
124 tier = EXCLUDED.tier,
125 storage_limit_bytes = EXCLUDED.storage_limit_bytes,
126 status = 'active',
127 canceled_at = NULL
128 WHERE app_sync_subscriptions.stripe_subscription_id != EXCLUDED.stripe_subscription_id
129 OR app_sync_subscriptions.status != 'active'
130 ",
131 )
132 .bind(sub.user_id)
133 .bind(sub.app_id)
134 .bind(sub.stripe_subscription_id)
135 .bind(sub.stripe_customer_id)
136 .bind(sub.interval)
137 .bind(sub.storage_limit_bytes)
138 .execute(pool)
139 .await?;
140 Ok(result.rows_affected() > 0)
141 }
142
143 /// Update the status of an existing app sync subscription (e.g. on
144 /// `customer.subscription.updated` or `.deleted` webhook events).
145 ///
146 /// `canceled` is terminal: the `AND (status != 'canceled' OR $2 = 'canceled')`
147 /// guard refuses to revive a canceled app-sub via an out-of-order webhook;
148 /// reactivation happens at checkout through `create_app_sync_subscription`'s
149 /// DO UPDATE path. Consistent with every other subscription family's setter.
150 /// `current_period_end_unix` is the **raw Stripe** period end (Unix seconds), as
151 /// `SubscriptionView::current_period()` / an invoice yields. The conversion is
152 /// sealed here: a `None` or a non-positive value writes no period (the COALESCE
153 /// keeps the existing value), so a thin/zero webhook can never stamp a 1970
154 /// period onto an active row. Handlers pass the raw value and cannot construct a
155 /// `DateTime`, closing the synckit arm of the epoch-period bug class (CHRONIC C).
156 #[tracing::instrument(skip_all)]
157 pub async fn update_app_sync_subscription_status(
158 pool: &PgPool,
159 stripe_subscription_id: &str,
160 status: &str,
161 current_period_end_unix: Option<i64>,
162 ) -> Result<()> {
163 let current_period_end = current_period_end_unix
164 .filter(|&end| end > 0)
165 .and_then(|end| DateTime::<Utc>::from_timestamp(end, 0));
166 sqlx::query(
167 r"
168 UPDATE app_sync_subscriptions
169 SET status = $2,
170 current_period_end = COALESCE($3, current_period_end),
171 canceled_at = CASE WHEN $2 = 'canceled' THEN NOW() ELSE canceled_at END
172 WHERE stripe_subscription_id = $1
173 AND (status != 'canceled' OR $2 = 'canceled')
174 ",
175 )
176 .bind(stripe_subscription_id)
177 .bind(status)
178 .bind(current_period_end)
179 .execute(pool)
180 .await?;
181 Ok(())
182 }
183
184 /// Look up an app sync subscription by its Stripe subscription ID. Used by
185 /// webhook handlers to find the row to update.
186 #[tracing::instrument(skip_all)]
187 pub async fn get_subscription_by_stripe_id(
188 pool: &PgPool,
189 stripe_subscription_id: &str,
190 ) -> Result<Option<(UserId, SyncAppId)>> {
191 let row: Option<(UserId, SyncAppId)> = sqlx::query_as(
192 r"
193 SELECT user_id, app_id
194 FROM app_sync_subscriptions
195 WHERE stripe_subscription_id = $1
196 ",
197 )
198 .bind(stripe_subscription_id)
199 .fetch_optional(pool)
200 .await?;
201 Ok(row)
202 }
203
204 /// Queue a storage-cap change to apply at the next billing cycle. Stores the
205 /// new cap in `pending_storage_limit_bytes`; the renewal webhook handler
206 /// promotes it to `storage_limit_bytes` once Stripe confirms the period roll.
207 #[tracing::instrument(skip_all)]
208 pub async fn set_pending_storage_cap(
209 pool: &PgPool,
210 user_id: UserId,
211 app_id: SyncAppId,
212 pending_bytes: i64,
213 ) -> Result<()> {
214 sqlx::query(
215 r"
216 UPDATE app_sync_subscriptions
217 SET pending_storage_limit_bytes = $3
218 WHERE user_id = $1 AND app_id = $2
219 ",
220 )
221 .bind(user_id)
222 .bind(app_id)
223 .bind(pending_bytes)
224 .execute(pool)
225 .await?;
226 Ok(())
227 }
228
229 /// Promote a queued cap change to the active cap. Called from the renewal
230 /// webhook handler when Stripe rolls the subscription to a new period.
231 /// No-op if no pending change is queued.
232 #[tracing::instrument(skip_all)]
233 pub async fn apply_pending_storage_cap(pool: &PgPool, stripe_subscription_id: &str) -> Result<()> {
234 sqlx::query(
235 r"
236 UPDATE app_sync_subscriptions
237 SET storage_limit_bytes = pending_storage_limit_bytes,
238 pending_storage_limit_bytes = NULL
239 WHERE stripe_subscription_id = $1
240 AND pending_storage_limit_bytes IS NOT NULL
241 ",
242 )
243 .bind(stripe_subscription_id)
244 .execute(pool)
245 .await?;
246 Ok(())
247 }
248