feat(billing): credit_ledger schema + record_ledger_entry

src/services/billing/ledger.rs

@@ -0,0 +1,137 @@
+use crate::error::AppResult;
+use sqlx::{SqlitePool, Transaction};
+
+/// One balance-changing event. Written to `credit_ledger` in the same
+/// transaction that updates the cached balance, so ledger and cache never diverge.
+pub struct LedgerEntry<'a> {
+    pub user_id: i64,
+    pub track: &'a str,        // "usd" | "points"
+    pub entry_type: &'a str,   // "recharge" | "admin_adjust" | "consume" | "refund" | "grant"
+    pub amount: f64,           // +credit / -debit
+    pub balance_after: f64,
+    pub ref_type: Option<&'a str>,
+    pub ref_id: Option<&'a str>,
+    pub model: Option<&'a str>,
+    pub operator: Option<&'a str>,
+    pub note: Option<&'a str>,
+}
+
+pub async fn ensure_ledger_schema(db: &SqlitePool) -> AppResult<()> {
+    sqlx::query(
+        "CREATE TABLE IF NOT EXISTS credit_ledger (
+            id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
+            user_id INTEGER NOT NULL,
+            track TEXT NOT NULL,
+            entry_type TEXT NOT NULL,
+            amount REAL NOT NULL,
+            balance_after REAL NOT NULL,
+            ref_type TEXT,
+            ref_id TEXT,
+            model TEXT,
+            operator TEXT,
+            note TEXT,
+            created_at TEXT DEFAULT (datetime('now'))
+        )",
+    )
+    .execute(db)
+    .await?;
+    sqlx::query(
+        "CREATE INDEX IF NOT EXISTS credit_ledger_user_created_idx
+         ON credit_ledger (user_id, created_at DESC)",
+    )
+    .execute(db)
+    .await?;
+    sqlx::query(
+        "CREATE INDEX IF NOT EXISTS credit_ledger_track_type_idx
+         ON credit_ledger (track, entry_type)",
+    )
+    .execute(db)
+    .await?;
+    sqlx::query(
+        "CREATE TABLE IF NOT EXISTS user_credits (
+            user_id INTEGER PRIMARY KEY NOT NULL,
+            points_balance REAL NOT NULL DEFAULT 0,
+            points_total_recharged REAL NOT NULL DEFAULT 0,
+            updated_at TEXT DEFAULT (datetime('now'))
+        )",
+    )
+    .execute(db)
+    .await?;
+    Ok(())
+}
+
+/// Append one ledger row inside an existing transaction. Caller is responsible
+/// for also updating the cached balance in the same `tx`.
+pub async fn record_ledger_entry(
+    tx: &mut Transaction<'_, sqlx::Sqlite>,
+    entry: LedgerEntry<'_>,
+) -> AppResult<()> {
+    sqlx::query(
+        "INSERT INTO credit_ledger
+            (user_id, track, entry_type, amount, balance_after,
+             ref_type, ref_id, model, operator, note)
+         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+    )
+    .bind(entry.user_id)
+    .bind(entry.track)
+    .bind(entry.entry_type)
+    .bind(entry.amount)
+    .bind(entry.balance_after)
+    .bind(entry.ref_type)
+    .bind(entry.ref_id)
+    .bind(entry.model)
+    .bind(entry.operator)
+    .bind(entry.note)
+    .execute(&mut **tx)
+    .await?;
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use sqlx::sqlite::SqlitePoolOptions;
+
+    async fn mem_pool() -> SqlitePool {
+        let pool = SqlitePoolOptions::new()
+            .max_connections(1)
+            .connect("sqlite::memory:")
+            .await
+            .unwrap();
+        ensure_ledger_schema(&pool).await.unwrap();
+        pool
+    }
+
+    #[tokio::test]
+    async fn record_ledger_entry_writes_row_with_balance_snapshot() {
+        let pool = mem_pool().await;
+        let mut tx = pool.begin().await.unwrap();
+        record_ledger_entry(
+            &mut tx,
+            LedgerEntry {
+                user_id: 7,
+                track: "usd",
+                entry_type: "recharge",
+                amount: 10.0,
+                balance_after: 10.0,
+                ref_type: Some("stripe"),
+                ref_id: Some("cs_1"),
+                model: None,
+                operator: Some("system"),
+                note: None,
+            },
+        )
+        .await
+        .unwrap();
+        tx.commit().await.unwrap();
+
+        let row: (i64, String, String, f64, f64) = sqlx::query_as(
+            "SELECT user_id, track, entry_type, amount, balance_after
+             FROM credit_ledger WHERE user_id = 7",
+        )
+        .fetch_one(&pool)
+        .await
+        .unwrap();
+        assert_eq!(row, (7, "usd".into(), "recharge".into(), 10.0, 10.0));
+    }
+}

src/services/billing/mod.rs

@@ -1,3 +1,5 @@
+pub mod ledger;
+
 use crate::{
     error::{AppError, AppResult},
     services::shared::{enforce_key_request_limit, increment_key_usage},
@@ -229,7 +231,12 @@ impl UsageTracker {
         .bind(event.key_id)
         .bind(&event.model)
         .bind(&event.model)
-        .bind(event.error_message.as_deref().unwrap_or("Comfy task failed"))
+        .bind(
+            event
+                .error_message
+                .as_deref()
+                .unwrap_or("Comfy task failed"),
+        )
         .bind(event.client_family)
         .bind(event.client_app_id)
         .bind(event.client_app_name)
@@ -1017,7 +1024,10 @@ mod tests {
         .fetch_one(&*tracker.db)
         .await
         .unwrap();
-        assert_eq!(proxy_log, ("failed".to_string(), 0.0, "Not found".to_string()));
+        assert_eq!(
+            proxy_log,
+            ("failed".to_string(), 0.0, "Not found".to_string())
+        );
     }
 
     #[tokio::test]
@@ -1076,7 +1086,9 @@ mod tests {
         .await
         .unwrap();
 
-        let projected = UsageTracker::project_proxy_usage_for_db(&pool).await.unwrap();
+        let projected = UsageTracker::project_proxy_usage_for_db(&pool)
+            .await
+            .unwrap();
         assert_eq!(projected, 1);
 
         let model_usage: (i64, i64, f64, i64) = sqlx::query_as(
@@ -1099,9 +1111,8 @@ mod tests {
         .unwrap();
         assert_eq!(site_usage, (1, 1, 0.12));
 
-        let hour_bucket: String = sqlx::query_scalar(
+        let hour_bucket: String =
-            "SELECT bucket_start_utc FROM site_hour_usage WHERE site_id = 10",
+            sqlx::query_scalar("SELECT bucket_start_utc FROM site_hour_usage WHERE site_id = 10")
-        )
                 .fetch_one(&pool)
                 .await
                 .unwrap();