Skip to main content

max / makenotwork

5.4 KB · 179 lines History Blame Raw
1 use chrono::{DateTime, Utc};
2 use sqlx::PgPool;
3
4 use crate::db::{SyncAppId, UserId};
5 use crate::error::Result;
6
7 /// Parameters for inserting a new app sync subscription from a webhook event.
8 pub struct NewAppSyncSubscription<'a> {
9 pub user_id: UserId,
10 pub app_id: SyncAppId,
11 pub stripe_subscription_id: &'a str,
12 pub stripe_customer_id: &'a str,
13 /// Billing interval ("monthly" / "annual"). Persisted in the `tier` column
14 /// (kept lowercase for that legacy name) so cap-change handlers know which
15 /// interval the user is on without round-tripping to Stripe.
16 pub interval: &'a str,
17 pub storage_limit_bytes: i64,
18 }
19
20 /// End-user subscription to an app's cloud sync (rows in `app_sync_subscriptions`).
21 #[derive(Debug, sqlx::FromRow)]
22 pub struct DbAppSyncSubscription {
23 pub stripe_subscription_id: String,
24 pub interval: String,
25 pub status: String,
26 pub storage_limit_bytes: Option<i64>,
27 pub pending_storage_limit_bytes: Option<i64>,
28 pub current_period_end: Option<DateTime<Utc>>,
29 }
30
31 /// Look up the active subscription a user has on an app, if any.
32 /// Returns `None` if the user has never subscribed or the subscription was deleted.
33 #[tracing::instrument(skip_all)]
34 pub async fn get_user_app_subscription(
35 pool: &PgPool,
36 user_id: UserId,
37 app_id: SyncAppId,
38 ) -> Result<Option<DbAppSyncSubscription>> {
39 let row = sqlx::query_as::<_, DbAppSyncSubscription>(
40 r#"
41 SELECT stripe_subscription_id,
42 tier AS interval,
43 status,
44 storage_limit_bytes,
45 pending_storage_limit_bytes,
46 current_period_end
47 FROM app_sync_subscriptions
48 WHERE user_id = $1 AND app_id = $2
49 "#,
50 )
51 .bind(user_id)
52 .bind(app_id)
53 .fetch_optional(pool)
54 .await?;
55 Ok(row)
56 }
57
58 /// Insert a new app sync subscription. Returns `Ok(true)` if a row was inserted,
59 /// `Ok(false)` if a subscription already existed for this (user, app) pair
60 /// (idempotent webhook replay).
61 #[tracing::instrument(skip_all)]
62 pub async fn create_app_sync_subscription(
63 pool: &PgPool,
64 sub: &NewAppSyncSubscription<'_>,
65 ) -> Result<bool> {
66 let result = sqlx::query(
67 r#"
68 INSERT INTO app_sync_subscriptions
69 (user_id, app_id, stripe_subscription_id, stripe_customer_id,
70 tier, status, storage_limit_bytes)
71 VALUES ($1, $2, $3, $4, $5, 'active', $6)
72 ON CONFLICT (user_id, app_id) DO NOTHING
73 "#,
74 )
75 .bind(sub.user_id)
76 .bind(sub.app_id)
77 .bind(sub.stripe_subscription_id)
78 .bind(sub.stripe_customer_id)
79 .bind(sub.interval)
80 .bind(sub.storage_limit_bytes)
81 .execute(pool)
82 .await?;
83 Ok(result.rows_affected() > 0)
84 }
85
86 /// Update the status of an existing app sync subscription (e.g. on
87 /// `customer.subscription.updated` or `.deleted` webhook events).
88 #[tracing::instrument(skip_all)]
89 pub async fn update_app_sync_subscription_status(
90 pool: &PgPool,
91 stripe_subscription_id: &str,
92 status: &str,
93 current_period_end: Option<DateTime<Utc>>,
94 ) -> Result<()> {
95 sqlx::query(
96 r#"
97 UPDATE app_sync_subscriptions
98 SET status = $2,
99 current_period_end = COALESCE($3, current_period_end),
100 canceled_at = CASE WHEN $2 = 'canceled' THEN NOW() ELSE canceled_at END
101 WHERE stripe_subscription_id = $1
102 "#,
103 )
104 .bind(stripe_subscription_id)
105 .bind(status)
106 .bind(current_period_end)
107 .execute(pool)
108 .await?;
109 Ok(())
110 }
111
112 /// Look up an app sync subscription by its Stripe subscription ID. Used by
113 /// webhook handlers to find the row to update.
114 #[tracing::instrument(skip_all)]
115 pub async fn get_subscription_by_stripe_id(
116 pool: &PgPool,
117 stripe_subscription_id: &str,
118 ) -> Result<Option<(UserId, SyncAppId)>> {
119 let row: Option<(UserId, SyncAppId)> = sqlx::query_as(
120 r#"
121 SELECT user_id, app_id
122 FROM app_sync_subscriptions
123 WHERE stripe_subscription_id = $1
124 "#,
125 )
126 .bind(stripe_subscription_id)
127 .fetch_optional(pool)
128 .await?;
129 Ok(row)
130 }
131
132 /// Queue a storage-cap change to apply at the next billing cycle. Stores the
133 /// new cap in `pending_storage_limit_bytes`; the renewal webhook handler
134 /// promotes it to `storage_limit_bytes` once Stripe confirms the period roll.
135 #[tracing::instrument(skip_all)]
136 pub async fn set_pending_storage_cap(
137 pool: &PgPool,
138 user_id: UserId,
139 app_id: SyncAppId,
140 pending_bytes: i64,
141 ) -> Result<()> {
142 sqlx::query(
143 r#"
144 UPDATE app_sync_subscriptions
145 SET pending_storage_limit_bytes = $3
146 WHERE user_id = $1 AND app_id = $2
147 "#,
148 )
149 .bind(user_id)
150 .bind(app_id)
151 .bind(pending_bytes)
152 .execute(pool)
153 .await?;
154 Ok(())
155 }
156
157 /// Promote a queued cap change to the active cap. Called from the renewal
158 /// webhook handler when Stripe rolls the subscription to a new period.
159 /// No-op if no pending change is queued.
160 #[tracing::instrument(skip_all)]
161 pub async fn apply_pending_storage_cap(
162 pool: &PgPool,
163 stripe_subscription_id: &str,
164 ) -> Result<()> {
165 sqlx::query(
166 r#"
167 UPDATE app_sync_subscriptions
168 SET storage_limit_bytes = pending_storage_limit_bytes,
169 pending_storage_limit_bytes = NULL
170 WHERE stripe_subscription_id = $1
171 AND pending_storage_limit_bytes IS NOT NULL
172 "#,
173 )
174 .bind(stripe_subscription_id)
175 .execute(pool)
176 .await?;
177 Ok(())
178 }
179