1use chrono::{DateTime, Utc};
2use sqlx::{Postgres, Row, Transaction, postgres::PgRow};
3use syrup_rail::{
4 ActorId, BillingScopeId, PlanKey, SubscriberId, SubscriptionGrant, SubscriptionGrantCreation,
5 SubscriptionGrantCreationOutcome, SubscriptionGrantId, SubscriptionGrantKind,
6 SubscriptionGrantReason, SubscriptionGrantRecord, SubscriptionGrantRevocation,
7 SubscriptionGrantRevocationOutcome, SubscriptionStatus,
8};
9use thiserror::Error;
10use uuid::Uuid;
11
12const BILLING_ROW_LOCK_TIMEOUT: &str = "250ms";
13const INVALID_GRANT_STATE: &str = "canonical subscription grant state is invalid";
14
15type GrantTimeState = (DateTime<Utc>, DateTime<Utc>, Option<DateTime<Utc>>);
16
17#[derive(Debug, Error)]
18pub enum SubscriptionGrantMutationError {
19 #[error("subscription grant mutation failed")]
20 Sql(#[from] sqlx::Error),
21 #[error("{0}")]
22 InvalidState(&'static str),
23}
24
25pub async fn create_subscription_grant(
31 transaction: &mut Transaction<'_, Postgres>,
32 creation: &SubscriptionGrantCreation,
33) -> Result<SubscriptionGrantCreationOutcome, SubscriptionGrantMutationError> {
34 set_lock_timeout(transaction).await?;
35 lock_subscription_aggregate(
36 transaction,
37 creation.subscriber_id().as_uuid(),
38 creation.plan_key().as_str(),
39 )
40 .await?;
41 lock_initial_attempts(
42 transaction,
43 creation.billing_scope_id().as_uuid(),
44 creation.subscriber_id().as_uuid(),
45 creation.plan_key().as_str(),
46 )
47 .await?;
48
49 let subscriptions: Vec<(String, DateTime<Utc>)> = sqlx::query_as(
50 r#"
51 SELECT status, current_period_end_at
52 FROM billing_subscriptions
53 WHERE billing_scope_id = $1
54 AND subscriber_id = $2
55 AND plan_key = $3
56 ORDER BY id
57 FOR NO KEY UPDATE
58 "#,
59 )
60 .bind(creation.billing_scope_id().as_uuid())
61 .bind(creation.subscriber_id().as_uuid())
62 .bind(creation.plan_key().as_str())
63 .fetch_all(&mut **transaction)
64 .await?;
65 let existing_grants: Vec<GrantTimeState> = sqlx::query_as(
66 r#"
67 SELECT starts_at, ends_at, revoked_at
68 FROM billing_subscription_grants
69 WHERE billing_scope_id = $1
70 AND subscriber_id = $2
71 AND plan_key = $3
72 ORDER BY id
73 FOR UPDATE
74 "#,
75 )
76 .bind(creation.billing_scope_id().as_uuid())
77 .bind(creation.subscriber_id().as_uuid())
78 .bind(creation.plan_key().as_str())
79 .fetch_all(&mut **transaction)
80 .await?;
81 let mutation_now: DateTime<Utc> = sqlx::query_scalar("SELECT clock_timestamp()")
82 .fetch_one(&mut **transaction)
83 .await?;
84
85 if creation.ends_at() <= &mutation_now {
86 return Ok(SubscriptionGrantCreationOutcome::EndsAtNotFuture);
87 }
88 if subscriptions.iter().any(|(status, period_end)| {
89 matches!(
90 status.parse(),
91 Ok(SubscriptionStatus::Active | SubscriptionStatus::PastDue)
92 ) || (status == SubscriptionStatus::Canceled.as_str() && *period_end > mutation_now)
93 }) {
94 return Ok(SubscriptionGrantCreationOutcome::CurrentPaidSubscription);
95 }
96 if existing_grants
97 .iter()
98 .any(|(starts_at, ends_at, revoked_at)| {
99 *starts_at <= mutation_now && *ends_at > mutation_now && revoked_at.is_none()
100 })
101 {
102 return Ok(SubscriptionGrantCreationOutcome::ActiveGrant);
103 }
104 if blocking_initial_attempt_exists(transaction, creation).await? {
105 return Ok(SubscriptionGrantCreationOutcome::BlockingInitialAttempt);
106 }
107 if pending_processor_evidence_exists(transaction, creation).await? {
108 return Ok(SubscriptionGrantCreationOutcome::PendingApprovedProcessorEvidence);
109 }
110
111 sqlx::query(
112 r#"
113 INSERT INTO billing_subscription_grants (
114 id,
115 billing_scope_id,
116 subscriber_id,
117 plan_key,
118 grant_kind,
119 reason,
120 starts_at,
121 ends_at,
122 granted_by_actor_id,
123 created_at,
124 updated_at
125 )
126 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $7, $7)
127 "#,
128 )
129 .bind(creation.id().as_uuid())
130 .bind(creation.billing_scope_id().as_uuid())
131 .bind(creation.subscriber_id().as_uuid())
132 .bind(creation.plan_key().as_str())
133 .bind(creation.kind().as_str())
134 .bind(creation.reason().as_str())
135 .bind(mutation_now)
136 .bind(creation.ends_at())
137 .bind(creation.granted_by_actor_id().as_uuid())
138 .execute(&mut **transaction)
139 .await?;
140
141 let record = grant_by_id(transaction, creation.id().as_uuid())
142 .await?
143 .ok_or(SubscriptionGrantMutationError::InvalidState(
144 INVALID_GRANT_STATE,
145 ))?;
146 Ok(SubscriptionGrantCreationOutcome::Created(Box::new(record)))
147}
148
149pub async fn revoke_subscription_grant(
151 transaction: &mut Transaction<'_, Postgres>,
152 revocation: &SubscriptionGrantRevocation,
153) -> Result<SubscriptionGrantRevocationOutcome, SubscriptionGrantMutationError> {
154 set_lock_timeout(transaction).await?;
155 lock_subscription_aggregate(
156 transaction,
157 revocation.subscriber_id().as_uuid(),
158 revocation.plan_key().as_str(),
159 )
160 .await?;
161
162 let row = sqlx::query(
163 r#"
164 SELECT id,
165 billing_scope_id,
166 subscriber_id,
167 plan_key,
168 grant_kind,
169 reason,
170 starts_at,
171 ends_at,
172 granted_by_actor_id,
173 revoked_at,
174 revoked_by_actor_id,
175 revocation_reason,
176 created_at,
177 updated_at
178 FROM billing_subscription_grants
179 WHERE id = $1
180 AND billing_scope_id = $2
181 AND subscriber_id = $3
182 AND plan_key = $4
183 FOR UPDATE
184 "#,
185 )
186 .bind(revocation.id().as_uuid())
187 .bind(revocation.billing_scope_id().as_uuid())
188 .bind(revocation.subscriber_id().as_uuid())
189 .bind(revocation.plan_key().as_str())
190 .fetch_optional(&mut **transaction)
191 .await?;
192 let Some(row) = row else {
193 return Ok(SubscriptionGrantRevocationOutcome::NotFound);
194 };
195 let record = grant_from_row(&row)?;
196 let mutation_now: DateTime<Utc> = sqlx::query_scalar("SELECT clock_timestamp()")
197 .fetch_one(&mut **transaction)
198 .await?;
199 if record.revoked_at().is_some() {
200 return Ok(SubscriptionGrantRevocationOutcome::AlreadyRevoked(
201 Box::new(record),
202 ));
203 }
204 if record.grant().ends_at() <= &mutation_now {
205 return Ok(SubscriptionGrantRevocationOutcome::Expired);
206 }
207
208 sqlx::query(
209 r#"
210 UPDATE billing_subscription_grants
211 SET revoked_at = $2,
212 revoked_by_actor_id = $3,
213 revocation_reason = $4,
214 updated_at = $2
215 WHERE id = $1
216 "#,
217 )
218 .bind(revocation.id().as_uuid())
219 .bind(mutation_now)
220 .bind(revocation.revoked_by_actor_id().as_uuid())
221 .bind(revocation.reason().as_str())
222 .execute(&mut **transaction)
223 .await?;
224
225 let record = grant_by_id(transaction, revocation.id().as_uuid())
226 .await?
227 .ok_or(SubscriptionGrantMutationError::InvalidState(
228 INVALID_GRANT_STATE,
229 ))?;
230 Ok(SubscriptionGrantRevocationOutcome::Revoked(Box::new(
231 record,
232 )))
233}
234
235async fn set_lock_timeout(transaction: &mut Transaction<'_, Postgres>) -> Result<(), sqlx::Error> {
236 sqlx::query("SELECT set_config('lock_timeout', $1, true)")
237 .bind(BILLING_ROW_LOCK_TIMEOUT)
238 .execute(&mut **transaction)
239 .await?;
240 Ok(())
241}
242
243async fn lock_subscription_aggregate(
244 transaction: &mut Transaction<'_, Postgres>,
245 subscriber_id: &Uuid,
246 plan_key: &str,
247) -> Result<(), sqlx::Error> {
248 sqlx::query(
249 r#"
250 SELECT pg_advisory_xact_lock(
251 hashtextextended($1::uuid::text || ':' || $2, 0)
252 )
253 "#,
254 )
255 .bind(subscriber_id)
256 .bind(plan_key)
257 .execute(&mut **transaction)
258 .await?;
259 Ok(())
260}
261
262async fn lock_initial_attempts(
263 transaction: &mut Transaction<'_, Postgres>,
264 billing_scope_id: &Uuid,
265 subscriber_id: &Uuid,
266 plan_key: &str,
267) -> Result<(), sqlx::Error> {
268 sqlx::query_scalar::<_, Uuid>(
269 r#"
270 SELECT id
271 FROM billing_payment_attempts
272 WHERE billing_scope_id = $1
273 AND subscriber_id = $2
274 AND plan_key = $3
275 AND attempt_kind = 'subscription_initial'
276 ORDER BY created_at, id
277 FOR UPDATE
278 "#,
279 )
280 .bind(billing_scope_id)
281 .bind(subscriber_id)
282 .bind(plan_key)
283 .fetch_all(&mut **transaction)
284 .await?;
285 Ok(())
286}
287
288async fn blocking_initial_attempt_exists(
289 transaction: &mut Transaction<'_, Postgres>,
290 creation: &SubscriptionGrantCreation,
291) -> Result<bool, sqlx::Error> {
292 sqlx::query_scalar(
293 r#"
294 SELECT EXISTS (
295 SELECT 1
296 FROM billing_payment_attempts attempts
297 WHERE attempts.billing_scope_id = $1
298 AND attempts.subscriber_id = $2
299 AND attempts.plan_key = $3
300 AND attempts.attempt_kind = 'subscription_initial'
301 AND (
302 attempts.status IN ('pending', 'unknown')
303 OR (
304 attempts.status = 'review_required'
305 AND attempts.resolution_code IS DISTINCT FROM
306 'subscription_initial_current_subscription_conflict'
307 )
308 )
309 AND NOT EXISTS (
310 SELECT 1
311 FROM billing_subscriptions subscriptions
312 WHERE subscriptions.billing_scope_id = attempts.billing_scope_id
313 AND subscriptions.subscriber_id = attempts.subscriber_id
314 AND subscriptions.plan_key = attempts.plan_key
315 AND subscriptions.created_at >= attempts.created_at
316 )
317 )
318 "#,
319 )
320 .bind(creation.billing_scope_id().as_uuid())
321 .bind(creation.subscriber_id().as_uuid())
322 .bind(creation.plan_key().as_str())
323 .fetch_one(&mut **transaction)
324 .await
325}
326
327async fn pending_processor_evidence_exists(
328 transaction: &mut Transaction<'_, Postgres>,
329 creation: &SubscriptionGrantCreation,
330) -> Result<bool, sqlx::Error> {
331 sqlx::query_scalar(
332 r#"
333 SELECT EXISTS (
334 SELECT 1
335 FROM billing_processor_charges evidence
336 INNER JOIN billing_payment_attempts attempts
337 ON attempts.id = evidence.attempt_id
338 WHERE attempts.billing_scope_id = $1
339 AND attempts.subscriber_id = $2
340 AND attempts.plan_key = $3
341 AND attempts.attempt_kind = 'subscription_initial'
342 AND evidence.progression_state IN (
343 'pending',
344 'reconciliation_required',
345 'external_reversal_required'
346 )
347 )
348 "#,
349 )
350 .bind(creation.billing_scope_id().as_uuid())
351 .bind(creation.subscriber_id().as_uuid())
352 .bind(creation.plan_key().as_str())
353 .fetch_one(&mut **transaction)
354 .await
355}
356
357async fn grant_by_id(
358 transaction: &mut Transaction<'_, Postgres>,
359 grant_id: &Uuid,
360) -> Result<Option<SubscriptionGrantRecord>, SubscriptionGrantMutationError> {
361 let row = sqlx::query(
362 r#"
363 SELECT id,
364 billing_scope_id,
365 subscriber_id,
366 plan_key,
367 grant_kind,
368 reason,
369 starts_at,
370 ends_at,
371 granted_by_actor_id,
372 revoked_at,
373 revoked_by_actor_id,
374 revocation_reason,
375 created_at,
376 updated_at
377 FROM billing_subscription_grants
378 WHERE id = $1
379 "#,
380 )
381 .bind(grant_id)
382 .fetch_optional(&mut **transaction)
383 .await?;
384 row.as_ref().map(grant_from_row).transpose()
385}
386
387fn grant_from_row(row: &PgRow) -> Result<SubscriptionGrantRecord, SubscriptionGrantMutationError> {
388 let plan_key = row
389 .try_get::<String, _>("plan_key")?
390 .parse::<PlanKey>()
391 .map_err(|_| SubscriptionGrantMutationError::InvalidState(INVALID_GRANT_STATE))?;
392 let grant = SubscriptionGrant::new(
393 SubscriptionGrantId::new(row.try_get("id")?),
394 plan_key,
395 row.try_get::<String, _>("grant_kind")?
396 .parse::<SubscriptionGrantKind>()
397 .map_err(|_| SubscriptionGrantMutationError::InvalidState(INVALID_GRANT_STATE))?,
398 row.try_get("starts_at")?,
399 row.try_get("ends_at")?,
400 ActorId::new(row.try_get("granted_by_actor_id")?),
401 )
402 .map_err(|_| SubscriptionGrantMutationError::InvalidState(INVALID_GRANT_STATE))?;
403 SubscriptionGrantRecord::new(
404 BillingScopeId::new(row.try_get("billing_scope_id")?),
405 SubscriberId::new(row.try_get("subscriber_id")?),
406 grant,
407 grant_reason(row.try_get("reason")?)?,
408 row.try_get("revoked_at")?,
409 row.try_get::<Option<Uuid>, _>("revoked_by_actor_id")?
410 .map(ActorId::new),
411 row.try_get::<Option<String>, _>("revocation_reason")?
412 .map(grant_reason)
413 .transpose()?,
414 row.try_get("created_at")?,
415 row.try_get("updated_at")?,
416 )
417 .map_err(|_| SubscriptionGrantMutationError::InvalidState(INVALID_GRANT_STATE))
418}
419
420fn grant_reason(value: String) -> Result<SubscriptionGrantReason, SubscriptionGrantMutationError> {
421 SubscriptionGrantReason::new(value)
422 .map_err(|_| SubscriptionGrantMutationError::InvalidState(INVALID_GRANT_STATE))
423}
424
425#[cfg(test)]
426mod tests {
427 use std::{error::Error, io};
428
429 use chrono::{Duration, Utc};
430 use syrup_rail::{
431 ActorId, BillingScopeId, PlanKey, SubscriberId, SubscriptionGrantCreation,
432 SubscriptionGrantCreationOutcome, SubscriptionGrantId, SubscriptionGrantKind,
433 SubscriptionGrantReason, SubscriptionGrantRevocation, SubscriptionGrantRevocationOutcome,
434 };
435 use uuid::Uuid;
436
437 use super::{create_subscription_grant, revoke_subscription_grant};
438 use crate::test_support::{TestDatabase, create_gateway_account};
439
440 #[tokio::test]
441 async fn grant_lifecycle_is_auditable_and_owned_by_the_caller_transaction()
442 -> Result<(), Box<dyn Error>> {
443 let database = TestDatabase::start("sr_grants_v1").await?;
444 let result = async {
445 let scope = BillingScopeId::new(Uuid::now_v7());
446 let subscriber = SubscriberId::new(Uuid::now_v7());
447 let plan = PlanKey::new("base_subscription")?;
448 let granting_actor = ActorId::new(Uuid::now_v7());
449 let grant_id = SubscriptionGrantId::new(Uuid::now_v7());
450 let creation = SubscriptionGrantCreation::new(
451 grant_id,
452 scope,
453 subscriber,
454 plan.clone(),
455 SubscriptionGrantKind::Promotion,
456 SubscriptionGrantReason::new(" launch partner ")?,
457 Utc::now() + Duration::days(30),
458 granting_actor,
459 );
460
461 let mut rolled_back = database.pool.begin().await?;
462 if !matches!(
463 create_subscription_grant(&mut rolled_back, &creation).await?,
464 SubscriptionGrantCreationOutcome::Created(_)
465 ) {
466 return Err(io::Error::other("grant was not created before rollback").into());
467 }
468 rolled_back.rollback().await?;
469 let count: i64 = sqlx::query_scalar(
470 "SELECT count(*) FROM billing_subscription_grants WHERE id = $1",
471 )
472 .bind(grant_id.as_uuid())
473 .fetch_one(&database.pool)
474 .await?;
475 if count != 0 {
476 return Err(io::Error::other("grant escaped caller rollback").into());
477 }
478
479 let mut created_transaction = database.pool.begin().await?;
480 let created = create_subscription_grant(&mut created_transaction, &creation).await?;
481 created_transaction.commit().await?;
482 let SubscriptionGrantCreationOutcome::Created(created) = created else {
483 return Err(io::Error::other("grant was not created").into());
484 };
485 if created.reason().as_str() != "launch partner"
486 || created.grant().id() != grant_id
487 || created.grant().granted_by_actor_id() != granting_actor
488 || created.revoked_at().is_some()
489 {
490 return Err(io::Error::other("created grant audit was not lossless").into());
491 }
492
493 let revoking_actor = ActorId::new(Uuid::now_v7());
494 let revocation = SubscriptionGrantRevocation::new(
495 grant_id,
496 scope,
497 subscriber,
498 plan,
499 revoking_actor,
500 SubscriptionGrantReason::new("access no longer required")?,
501 );
502 let mut revoked_transaction = database.pool.begin().await?;
503 let revoked = revoke_subscription_grant(&mut revoked_transaction, &revocation).await?;
504 revoked_transaction.commit().await?;
505 let SubscriptionGrantRevocationOutcome::Revoked(revoked) = revoked else {
506 return Err(io::Error::other("active grant was not revoked").into());
507 };
508 if revoked.revoked_by_actor_id() != Some(revoking_actor)
509 || revoked.revocation_reason().map(|reason| reason.as_str())
510 != Some("access no longer required")
511 || revoked.revoked_at().is_none()
512 {
513 return Err(io::Error::other("revocation audit was not lossless").into());
514 }
515
516 let mut repeated_transaction = database.pool.begin().await?;
517 let repeated =
518 revoke_subscription_grant(&mut repeated_transaction, &revocation).await?;
519 repeated_transaction.commit().await?;
520 if !matches!(
521 repeated,
522 SubscriptionGrantRevocationOutcome::AlreadyRevoked(record)
523 if record == revoked
524 ) {
525 return Err(io::Error::other("repeated revocation was not idempotent").into());
526 }
527 Ok::<_, Box<dyn Error>>(())
528 }
529 .await;
530 let cleanup = database.cleanup().await;
531 result?;
532 cleanup
533 }
534
535 #[tokio::test]
536 async fn grant_admission_is_exactly_scope_subscriber_and_plan_scoped()
537 -> Result<(), Box<dyn Error>> {
538 let database = TestDatabase::start("sr_grant_scope").await?;
539 let result = async {
540 let gateway = create_gateway_account(&database.pool, "test_gateway").await?;
541 let plan_a = PlanKey::new("plan_a")?;
542 let actor = ActorId::new(Uuid::now_v7());
543 let subscriber_with_other_plan_attempt = SubscriberId::new(Uuid::now_v7());
544 insert_pending_initial_attempt(
545 &database.pool,
546 gateway.billing_scope_id,
547 subscriber_with_other_plan_attempt.into_uuid(),
548 "plan_b",
549 gateway.gateway_account_id,
550 gateway.gateway_configuration_id,
551 )
552 .await?;
553 let creation = SubscriptionGrantCreation::new(
554 SubscriptionGrantId::new(Uuid::now_v7()),
555 BillingScopeId::new(gateway.billing_scope_id),
556 subscriber_with_other_plan_attempt,
557 plan_a.clone(),
558 SubscriptionGrantKind::Testing,
559 SubscriptionGrantReason::new("plan isolation")?,
560 Utc::now() + Duration::days(1),
561 actor,
562 );
563 let mut transaction = database.pool.begin().await?;
564 let outcome = create_subscription_grant(&mut transaction, &creation).await?;
565 transaction.commit().await?;
566 if !matches!(outcome, SubscriptionGrantCreationOutcome::Created(_)) {
567 return Err(io::Error::other("another plan blocked grant creation").into());
568 }
569
570 let blocked_subscriber = SubscriberId::new(Uuid::now_v7());
571 insert_pending_initial_attempt(
572 &database.pool,
573 gateway.billing_scope_id,
574 blocked_subscriber.into_uuid(),
575 plan_a.as_str(),
576 gateway.gateway_account_id,
577 gateway.gateway_configuration_id,
578 )
579 .await?;
580 let blocked_creation = SubscriptionGrantCreation::new(
581 SubscriptionGrantId::new(Uuid::now_v7()),
582 BillingScopeId::new(gateway.billing_scope_id),
583 blocked_subscriber,
584 plan_a,
585 SubscriptionGrantKind::Testing,
586 SubscriptionGrantReason::new("same plan must block")?,
587 Utc::now() + Duration::days(1),
588 actor,
589 );
590 let mut blocked_transaction = database.pool.begin().await?;
591 let blocked =
592 create_subscription_grant(&mut blocked_transaction, &blocked_creation).await?;
593 blocked_transaction.commit().await?;
594 if blocked != SubscriptionGrantCreationOutcome::BlockingInitialAttempt {
595 return Err(io::Error::other("same-plan initial attempt did not block").into());
596 }
597 Ok::<_, Box<dyn Error>>(())
598 }
599 .await;
600 let cleanup = database.cleanup().await;
601 result?;
602 cleanup
603 }
604
605 async fn insert_pending_initial_attempt(
606 pool: &sqlx::PgPool,
607 scope: Uuid,
608 subscriber: Uuid,
609 plan: &str,
610 gateway_account: Uuid,
611 gateway_configuration: Uuid,
612 ) -> Result<(), sqlx::Error> {
613 let attempt = Uuid::now_v7();
614 sqlx::query(
615 r#"
616 INSERT INTO billing_payment_attempts (
617 required_gateway_account_mode,
618 id,
619 billing_scope_id,
620 subscriber_id,
621 plan_key,
622 attempt_kind,
623 status,
624 idempotency_key,
625 request_fingerprint,
626 amount_cents,
627 currency,
628 gateway_account_id,
629 gateway_configuration_id,
630 gateway_order_id,
631 subscription_initial_terms_version,
632 subscription_initial_start_kind,
633 subscription_initial_recurring_base_amount_cents,
634 subscription_initial_recurring_period_kind,
635 subscription_initial_recurring_period_count,
636 subscription_initial_dunning_retry_delays_seconds,
637 subscription_initial_dunning_exhaustion,
638 subscription_initial_past_due_access
639 ) VALUES (
640 'live',
641 $1, $2, $3, $4, 'subscription_initial', 'pending', $5, $6,
642 100, 'USD', $7, $8, $9, 2, 'recurring_immediately', 100,
643 'calendar_months', 1, ARRAY[]::bigint[],
644 'remain_past_due', 'suspend_immediately'
645 )
646 "#,
647 )
648 .bind(attempt)
649 .bind(scope)
650 .bind(subscriber)
651 .bind(plan)
652 .bind(format!("grant-{}", attempt.simple()))
653 .bind(format!("initial:{plan}:100:USD"))
654 .bind(gateway_account)
655 .bind(gateway_configuration)
656 .bind(format!("grant-order-{}", attempt.simple()))
657 .execute(pool)
658 .await?;
659 Ok(())
660 }
661}