Move charges to DB + fix subscription recurring payments

migrations/20240923163452_charges-fix.sql

@@ -0,0 +1,12 @@
+CREATE TABLE charges (
+    id bigint PRIMARY KEY,
+    user_id bigint REFERENCES users NOT NULL,
+    price_id bigint REFERENCES products_prices NOT NULL,
+    amount bigint NOT NULL,
+    currency_code text NOT NULL,
+    subscription_id bigint REFERENCES users_subscriptions NULL,
+    interval text NULL,
+    status varchar(255) NOT NULL,
+    due timestamptz DEFAULT CURRENT_TIMESTAMP NOT NULL,
+    last_attempt timestamptz NOT NULL
+);

src/database/models/charge_item.rs

@@ -0,0 +1,131 @@
+use crate::database::models::{
+    ChargeId, DatabaseError, ProductPriceId, UserId, UserSubscriptionId,
+};
+use crate::models::billing::{ChargeStatus, PriceDuration};
+use chrono::{DateTime, Utc};
+use std::convert::TryFrom;
+
+pub struct ChargeItem {
+    pub id: ChargeId,
+    pub user_id: UserId,
+    pub price_id: ProductPriceId,
+    pub amount: i64,
+    pub currency_code: String,
+    pub subscription_id: Option<UserSubscriptionId>,
+    pub interval: Option<PriceDuration>,
+    pub status: ChargeStatus,
+    pub due: DateTime<Utc>,
+    pub last_attempt: Option<DateTime<Utc>>,
+}
+
+struct ChargeResult {
+    id: i64,
+    user_id: i64,
+    price_id: i64,
+    amount: i64,
+    currency_code: String,
+    subscription_id: Option<i64>,
+    interval: Option<String>,
+    status: String,
+    due: DateTime<Utc>,
+    last_attempt: Option<DateTime<Utc>>,
+}
+
+impl TryFrom<ChargeResult> for ChargeItem {
+    type Error = serde_json::Error;
+
+    fn try_from(r: ChargeResult) -> Result<Self, Self::Error> {
+        Ok(ChargeItem {
+            id: ChargeId(r.id),
+            user_id: UserId(r.user_id),
+            price_id: ProductPriceId(r.price_id),
+            amount: r.amount,
+            currency_code: r.currency_code,
+            subscription_id: r.subscription_id.map(UserSubscriptionId),
+            interval: r.interval.map(|x| serde_json::from_str(&x)).transpose()?,
+            status: serde_json::from_str(&r.status)?,
+            due: r.due,
+            last_attempt: r.last_attempt,
+        })
+    }
+}
+
+macro_rules! select_charges_with_predicate {
+    ($predicate:tt, $param:ident) => {
+        sqlx::query_as!(
+            ChargeResult,
+            r#"
+            SELECT id, user_id, price_id, amount, currency_code, subscription_id, interval, status, due, last_attempt
+            FROM charges
+            "#
+                + $predicate,
+            $param
+        )
+    };
+}
+
+impl ChargeItem {
+    pub async fn insert(
+        &self,
+        exec: impl sqlx::Executor<'_, Database = sqlx::Postgres>,
+    ) -> Result<ChargeId, DatabaseError> {
+        sqlx::query!(
+            r#"
+            INSERT INTO charges (id, user_id, price_id, amount, currency_code, subscription_id, interval, status, due, last_attempt)
+            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
+            "#,
+            self.id.0,
+            self.user_id.0,
+            self.price_id.0,
+            self.amount,
+            self.currency_code,
+            self.subscription_id.map(|x| x.0),
+            self.interval.map(|x| x.as_str()),
+            self.status.as_str(),
+            self.due,
+            self.last_attempt,
+        )
+        .execute(exec)
+        .await?;
+
+        Ok(self.id)
+    }
+
+    pub async fn get(
+        id: ChargeId,
+        exec: impl sqlx::Executor<'_, Database = sqlx::Postgres>,
+    ) -> Result<Option<ChargeItem>, DatabaseError> {
+        let res = select_charges_with_predicate!("WHERE id = $1", id)
+            .fetch_optional(exec)
+            .await?;
+
+        Ok(res.map(|r| r.try_into()))
+    }
+
+    pub async fn get_from_user(
+        user_id: UserId,
+        exec: impl sqlx::Executor<'_, Database = sqlx::Postgres>,
+    ) -> Result<Vec<ChargeItem>, DatabaseError> {
+        let res = select_charges_with_predicate!("WHERE user_id = $1", user_id)
+            .fetch_all(exec)
+            .await?;
+
+        Ok(res
+            .into_iter()
+            .map(|r| r.try_into())
+            .collect::<Result<Vec<_>, serde_json::Error>>()?)
+    }
+
+    pub async fn get_chargeable(
+        exec: impl sqlx::Executor<'_, Database = sqlx::Postgres>,
+    ) -> Result<Vec<ChargeItem>, DatabaseError> {
+        let res = select_charges_with_predicate!("WHERE (status = 'open' AND due < NOW()) OR (status = 'failed' AND last_attempt < NOW() - INTERVAL '2 days')")
+            .fetch_all(exec)
+            .await?;
+
+        Ok(res
+            .into_iter()
+            .map(|r| r.try_into())
+            .collect::<Result<Vec<_>, serde_json::Error>>()?)
+    }
+}

src/database/models/ids.rs

@@ -256,6 +256,14 @@ generate_ids!(
     UserSubscriptionId
 );
 
+generate_ids!(
+    pub generate_charge_id,
+    ChargeId,
+    8,
+    "SELECT EXISTS(SELECT 1 FROM charges WHERE id=$1)",
+    ChargeId
+);
+
 #[derive(Copy, Clone, Debug, PartialEq, Eq, Type, Hash, Serialize, Deserialize)]
 #[sqlx(transparent)]
 pub struct UserId(pub i64);
@@ -386,6 +394,10 @@ pub struct ProductPriceId(pub i64);
 #[sqlx(transparent)]
 pub struct UserSubscriptionId(pub i64);
 
+#[derive(Copy, Clone, Debug, Type, Serialize, Deserialize, Eq, PartialEq, Hash)]
+#[sqlx(transparent)]
+pub struct ChargeId(pub i64);
+
 use crate::models::ids;
 
 impl From<ids::ProjectId> for ProjectId {
@@ -571,3 +583,14 @@ impl From<UserSubscriptionId> for ids::UserSubscriptionId {
         ids::UserSubscriptionId(id.0 as u64)
     }
 }
+
+impl From<ids::ChargeId> for ChargeId {
+    fn from(id: ids::ChargeId) -> Self {
+        ChargeId(id.0 as i64)
+    }
+}
+impl From<ChargeId> for ids::ChargeId {
+    fn from(id: ChargeId) -> Self {
+        ids::ChargeId(id.0 as u64)
+    }
+}

src/database/models/mod.rs

@@ -1,6 +1,7 @@
 use thiserror::Error;
 
 pub mod categories;
+pub mod charge_item;
 pub mod collection_item;
 pub mod flow_item;
 pub mod ids;

src/database/models/user_subscription_item.rs

@@ -89,17 +89,6 @@ impl UserSubscriptionItem {
         Ok(results.into_iter().map(|r| r.into()).collect())
     }
 
-    pub async fn get_all_expired(
-        exec: impl sqlx::Executor<'_, Database = sqlx::Postgres>,
-    ) -> Result<Vec<UserSubscriptionItem>, DatabaseError> {
-        let now = Utc::now();
-        let results = select_user_subscriptions_with_predicate!("WHERE expires < $1", now)
-            .fetch_all(exec)
-            .await?;
-
-        Ok(results.into_iter().map(|r| r.into()).collect())
-    }
-
     pub async fn upsert(
         &self,
         transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,

src/models/v3/billing.rs

@@ -134,3 +134,53 @@ impl SubscriptionStatus {
         }
     }
 }
+
+#[derive(Copy, Clone, PartialEq, Eq, Serialize, Deserialize, Debug)]
+#[serde(from = "Base62Id")]
+#[serde(into = "Base62Id")]
+pub struct ChargeId(pub u64);
+
+#[derive(Serialize, Deserialize)]
+pub struct Charge {
+    pub id: ChargeId,
+    pub user_id: UserId,
+    pub price_id: ProductPriceId,
+    pub amount: i64,
+    pub currency_code: String,
+    pub subscription_id: Option<UserSubscriptionId>,
+    pub interval: Option<PriceDuration>,
+    pub status: ChargeStatus,
+    pub due: DateTime<Utc>,
+    pub last_attempt: Option<DateTime<Utc>>,
+}
+
+#[derive(Serialize, Deserialize, Eq, PartialEq)]
+#[serde(rename_all = "kebab-case")]
+pub enum ChargeStatus {
+    // Open charges are for the next billing interval
+    Open,
+    Processing,
+    Succeeded,
+    Failed,
+}
+
+impl ChargeStatus {
+    pub fn from_string(string: &str) -> ChargeStatus {
+        match string {
+            "processing" => ChargeStatus::Processing,
+            "succeeded" => ChargeStatus::Succeeded,
+            "failed" => ChargeStatus::Failed,
+            "open" => ChargeStatus::Open,
+            _ => ChargeStatus::Failed,
+        }
+    }
+
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            ChargeStatus::Processing => "processing",
+            ChargeStatus::Succeeded => "succeeded",
+            ChargeStatus::Failed => "failed",
+            ChargeStatus::Open => "open",
+        }
+    }
+}

src/models/v3/ids.rs

@@ -13,8 +13,7 @@ pub use super::teams::TeamId;
 pub use super::threads::ThreadId;
 pub use super::threads::ThreadMessageId;
 pub use super::users::UserId;
-pub use crate::models::billing::UserSubscriptionId;
-pub use crate::models::v3::billing::{ProductId, ProductPriceId};
+pub use crate::models::billing::{ChargeId, ProductId, ProductPriceId, UserSubscriptionId};
 use thiserror::Error;
 
 /// Generates a random 64 bit integer that is exactly `n` characters
@@ -137,6 +136,7 @@ base62_id_impl!(PayoutId, PayoutId);
 base62_id_impl!(ProductId, ProductId);
 base62_id_impl!(ProductPriceId, ProductPriceId);
 base62_id_impl!(UserSubscriptionId, UserSubscriptionId);
+base62_id_impl!(ChargeId, ChargeId);
 
 pub mod base62_impl {
     use serde::de::{self, Deserializer, Visitor};