syrup-rail-postgres 0.5.0

Canonical provider-neutral PostgreSQL schema contract and SQLx orchestration for Syrup Rail
Documentation
use chrono::{DateTime, Duration, Utc};
use sqlx::PgPool;
use syrup_rail::RenewalDispatchPage;
use uuid::Uuid;

use super::super::{RenewalStoreError, due_renewals_page};
use crate::test_support::GatewayAccountFixture;

#[derive(Clone)]
pub(super) struct DueSubscriptionFixture {
    pub(super) subscription_id: Uuid,
    pub(super) subscriber_id: Uuid,
    pub(super) payment_method_id: Uuid,
    pub(super) plan_key: String,
    pub(super) initial_transaction_id: String,
}

pub(super) async fn insert_due_subscription(
    pool: &PgPool,
    account: GatewayAccountFixture,
    plan_key: &str,
) -> Result<Uuid, sqlx::Error> {
    Ok(insert_due_subscription_at(
        pool,
        account,
        plan_key,
        Uuid::now_v7(),
        Utc::now() - Duration::days(1),
    )
    .await?
    .subscription_id)
}

pub(super) async fn insert_due_subscription_at(
    pool: &PgPool,
    account: GatewayAccountFixture,
    plan_key: &str,
    subscription_id: Uuid,
    due_at: DateTime<Utc>,
) -> Result<DueSubscriptionFixture, sqlx::Error> {
    let subscriber_id = Uuid::now_v7();
    let payment_method_id = Uuid::now_v7();
    let initial_transaction_id = format!("txn_{}", subscription_id.simple());
    sqlx::query(
        r#"
        INSERT INTO billing_payment_methods (
            id, billing_scope_id, subscriber_id, gateway_account_id,
            gateway_payment_method_reference, status
        ) VALUES ($1, $2, $3, $4, $5, 'active')
        "#,
    )
    .bind(payment_method_id)
    .bind(account.billing_scope_id)
    .bind(subscriber_id)
    .bind(account.gateway_account_id)
    .bind(format!("vault_{}", payment_method_id.simple()))
    .execute(pool)
    .await?;
    sqlx::query(
        r#"
        INSERT INTO billing_subscriptions (
            required_gateway_account_mode,
            id, billing_scope_id, subscriber_id, plan_key, status,
            gateway_account_id, payment_method_id, amount_cents, currency,
            current_period_start_at, current_period_end_at, next_renewal_at,
            initial_transaction_id, phase, recurring_period_kind,
            recurring_period_count, dunning_retry_delays_seconds,
            dunning_exhaustion, past_due_access, next_payment_attempt_at
        ) VALUES (
            'live',
            $1, $2, $3, $4, 'active', $5, $6, 1900, 'USD', $7, $8, $8, $9,
            'recurring', 'calendar_months', 1, ARRAY[]::bigint[],
            'remain_past_due', 'suspend_immediately', $8
        )
        "#,
    )
    .bind(subscription_id)
    .bind(account.billing_scope_id)
    .bind(subscriber_id)
    .bind(plan_key)
    .bind(account.gateway_account_id)
    .bind(payment_method_id)
    .bind(due_at - Duration::days(32))
    .bind(due_at)
    .bind(&initial_transaction_id)
    .execute(pool)
    .await?;
    Ok(DueSubscriptionFixture {
        subscription_id,
        subscriber_id,
        payment_method_id,
        plan_key: plan_key.to_owned(),
        initial_transaction_id,
    })
}

pub(super) async fn insert_due_subscription_population(
    pool: &PgPool,
    account: GatewayAccountFixture,
    due_at: DateTime<Utc>,
    population: i32,
) -> Result<(), sqlx::Error> {
    sqlx::query(
        r#"
        INSERT INTO billing_payment_methods (
            id, billing_scope_id, subscriber_id, gateway_account_id,
            gateway_payment_method_reference, status
        )
        SELECT
            md5('renewal-plan-method-' || value)::uuid,
            $1,
            md5('renewal-plan-subscriber-' || value)::uuid,
            $2,
            'vault_renewal_plan_' || value,
            'active'
        FROM generate_series(1, $3::integer) AS fixture(value)
        "#,
    )
    .bind(account.billing_scope_id)
    .bind(account.gateway_account_id)
    .bind(population)
    .execute(pool)
    .await?;
    sqlx::query(
        r#"
        INSERT INTO billing_subscriptions (
            required_gateway_account_mode,
            id, billing_scope_id, subscriber_id, plan_key, status,
            gateway_account_id, payment_method_id, amount_cents, currency,
            current_period_start_at, current_period_end_at, next_renewal_at,
            initial_transaction_id, phase, recurring_period_kind,
            recurring_period_count, dunning_retry_delays_seconds,
            dunning_exhaustion, past_due_access, next_payment_attempt_at
        )
        SELECT
            'live', md5('renewal-plan-subscription-' || value)::uuid,
            $1,
            md5('renewal-plan-subscriber-' || value)::uuid,
            'plan-shape',
            'active',
            $2,
            md5('renewal-plan-method-' || value)::uuid,
            1900,
            'USD',
            scheduled_at - interval '32 days',
            scheduled_at,
            scheduled_at,
            'renewal-plan-transaction-' || value,
            'recurring',
            'calendar_months',
            1,
            ARRAY[]::bigint[],
            'remain_past_due',
            'suspend_immediately',
            scheduled_at
        FROM generate_series(1, $4::integer) AS fixture(value)
        CROSS JOIN LATERAL (
            SELECT $3::timestamptz
                - ((value - 1) / 4) * interval '1 second' AS scheduled_at
        ) AS schedule
        "#,
    )
    .bind(account.billing_scope_id)
    .bind(account.gateway_account_id)
    .bind(due_at)
    .bind(population)
    .execute(pool)
    .await?;
    sqlx::query(
        r#"
        INSERT INTO billing_payment_attempts (
            required_gateway_account_mode,
            id, billing_scope_id, subscriber_id, plan_key, subscription_id,
            payment_method_id, attempt_kind, status, idempotency_key,
            request_fingerprint, amount_cents, currency,
            billing_period_start_at, billing_period_end_at,
            gateway_account_id, gateway_configuration_id, gateway_order_id,
            subscription_expected_payment_method_id,
            subscription_expected_initial_transaction_id,
            subscription_expected_status, resolved_at, created_at, updated_at
        )
        SELECT
            'live',
            md5('renewal-plan-attempt-' || value)::uuid,
            $1,
            md5('renewal-plan-subscriber-' || value)::uuid,
            'plan-shape',
            md5('renewal-plan-subscription-' || value)::uuid,
            md5('renewal-plan-method-' || value)::uuid,
            'subscription_recovery',
            'failed',
            'renewal-plan-idempotency-' || value,
            'renewal-plan-fingerprint-' || value,
            1900,
            'USD',
            scheduled_at,
            scheduled_at + interval '1 month',
            $2,
            $3,
            'renewal-plan-order-' || value,
            md5('renewal-plan-method-' || value)::uuid,
            'renewal-plan-transaction-' || value,
            'active',
            scheduled_at - interval '1 second',
            scheduled_at - interval '2 seconds',
            scheduled_at - interval '1 second'
        FROM generate_series(1, $5::integer) AS fixture(value)
        CROSS JOIN LATERAL (
            SELECT $4::timestamptz
                - ((value - 1) / 4) * interval '1 second' AS scheduled_at
        ) AS schedule
        "#,
    )
    .bind(account.billing_scope_id)
    .bind(account.gateway_account_id)
    .bind(account.gateway_configuration_id)
    .bind(due_at)
    .bind(population)
    .execute(pool)
    .await?;
    Ok(())
}

pub(super) async fn update_due_at(
    pool: &PgPool,
    subscription_id: Uuid,
    due_at: DateTime<Utc>,
) -> Result<(), sqlx::Error> {
    sqlx::query(
        r#"
        UPDATE billing_subscriptions
        SET current_period_end_at = $2,
            next_renewal_at = $2,
            next_payment_attempt_at = $2
        WHERE id = $1
        "#,
    )
    .bind(subscription_id)
    .bind(due_at)
    .execute(pool)
    .await?;
    Ok(())
}

#[allow(clippy::too_many_arguments)]
pub(super) async fn insert_renewal_attempt(
    pool: &PgPool,
    account: GatewayAccountFixture,
    subscription: &DueSubscriptionFixture,
    period_start_at: DateTime<Utc>,
    kind: &str,
    status: &str,
    resolution_code: Option<&str>,
    resolved_at: Option<DateTime<Utc>>,
) -> Result<(), sqlx::Error> {
    let attempt_id = Uuid::now_v7();
    sqlx::query(
        r#"
        INSERT INTO billing_payment_attempts (
            required_gateway_account_mode,
            id, billing_scope_id, subscriber_id, plan_key, subscription_id,
            payment_method_id, attempt_kind, status, idempotency_key,
            request_fingerprint, amount_cents, currency,
            billing_period_start_at, billing_period_end_at,
            gateway_account_id, gateway_configuration_id, gateway_order_id,
            resolution_code, resolved_at,
            subscription_expected_payment_method_id,
            subscription_expected_initial_transaction_id,
            subscription_expected_status
        ) VALUES (
            'live',
            $1, $2, $3, $4, $5, $6, $7, $8, $9, $10,
            1900, 'USD', $11, $12, $13, $14, $15, $16, $17,
            $6, $18, 'active'
        )
        "#,
    )
    .bind(attempt_id)
    .bind(account.billing_scope_id)
    .bind(subscription.subscriber_id)
    .bind(&subscription.plan_key)
    .bind(subscription.subscription_id)
    .bind(subscription.payment_method_id)
    .bind(kind)
    .bind(status)
    .bind(format!("attempt-{}", attempt_id.simple()))
    .bind(format!("fingerprint-{}", attempt_id.simple()))
    .bind(period_start_at)
    .bind(period_start_at + Duration::days(28))
    .bind(account.gateway_account_id)
    .bind(account.gateway_configuration_id)
    .bind(format!("order_{}", attempt_id.simple()))
    .bind(resolution_code)
    .bind(resolved_at)
    .bind(&subscription.initial_transaction_id)
    .execute(pool)
    .await?;
    Ok(())
}

pub(super) async fn insert_pending_payment_method_update(
    pool: &PgPool,
    account: GatewayAccountFixture,
    subscription: &DueSubscriptionFixture,
    created_at: DateTime<Utc>,
) -> Result<Uuid, sqlx::Error> {
    let attempt_id = Uuid::now_v7();
    sqlx::query(
        r#"
        INSERT INTO billing_payment_attempts (
            required_gateway_account_mode,
            id, billing_scope_id, subscriber_id, plan_key, subscription_id,
            payment_method_id, attempt_kind, status, idempotency_key,
            request_fingerprint, amount_cents, currency,
            gateway_account_id, gateway_configuration_id, gateway_order_id,
            payment_method_update_expected_payment_method_id,
            payment_method_update_expected_initial_transaction_id,
            created_at, updated_at
        ) VALUES (
            'live',
            $1, $2, $3, $4, $5, $6,
            'subscription_payment_method_update', 'pending', $7, $8,
            0, 'USD', $9, $10, $11, $6, $12, $13, $13
        )
        "#,
    )
    .bind(attempt_id)
    .bind(account.billing_scope_id)
    .bind(subscription.subscriber_id)
    .bind(&subscription.plan_key)
    .bind(subscription.subscription_id)
    .bind(subscription.payment_method_id)
    .bind(format!("method-update-{}", attempt_id.simple()))
    .bind(format!("method-update-fingerprint-{}", attempt_id.simple()))
    .bind(account.gateway_account_id)
    .bind(account.gateway_configuration_id)
    .bind(format!("method-update-order_{}", attempt_id.simple()))
    .bind(&subscription.initial_transaction_id)
    .bind(created_at)
    .execute(pool)
    .await?;
    Ok(attempt_id)
}

pub(super) fn dispatch_ids(page: &RenewalDispatchPage) -> Vec<Uuid> {
    page.dispatches()
        .iter()
        .map(|dispatch| dispatch.subscription_id().into_uuid())
        .collect()
}

pub(super) async fn scan_dispatch_ids(pool: &PgPool) -> Result<Vec<Uuid>, RenewalStoreError> {
    let mut cursor = None;
    let mut ids = Vec::new();
    loop {
        let page = due_renewals_page(pool, cursor.as_ref()).await?;
        ids.extend(dispatch_ids(&page));
        cursor = page.next_cursor();
        if cursor.is_none() {
            return Ok(ids);
        }
    }
}