feat(billing): points track apply_points_delta + balance

src/services/billing/ledger.rs

@@ -87,6 +87,62 @@ pub async fn record_ledger_entry(
     Ok(())
 }
 
+/// Current points balance for a user (0 if no row yet).
+pub async fn points_balance(db: &SqlitePool, user_id: i64) -> AppResult<f64> {
+    let bal: Option<f64> = sqlx::query_scalar(
+        "SELECT points_balance FROM user_credits WHERE user_id = ?",
+    )
+    .bind(user_id)
+    .fetch_optional(db)
+    .await?;
+    Ok(bal.unwrap_or(0.0))
+}
+
+/// Apply a points delta (+credit / -debit) atomically: upsert the cached
+/// balance in `user_credits` and append one `credit_ledger(track=points)` row,
+/// in a single transaction. Returns the new balance.
+#[allow(clippy::too_many_arguments)]
+pub async fn apply_points_delta(
+    db: &SqlitePool,
+    user_id: i64,
+    delta: f64,
+    entry_type: &str,
+    model: Option<&str>,
+    operator: Option<&str>,
+    ref_type: Option<&str>,
+    ref_id: Option<&str>,
+) -> AppResult<f64> {
+    ensure_ledger_schema(db).await?;
+    let mut tx = db.begin().await?;
+    let current: f64 = sqlx::query_scalar(
+        "SELECT points_balance FROM user_credits WHERE user_id = ?",
+    )
+    .bind(user_id)
+    .fetch_optional(&mut *tx)
+    .await?
+    .unwrap_or(0.0);
+    let next = current + delta;
+    let total_recharged_inc = if delta > 0.0 && entry_type == "recharge" { delta } else { 0.0 };
+    sqlx::query(
+        "INSERT INTO user_credits (user_id, points_balance, points_total_recharged, updated_at)
+         VALUES (?, ?, ?, datetime('now'))
+         ON CONFLICT(user_id) DO UPDATE SET
+            points_balance = ?,
+            points_total_recharged = points_total_recharged + ?,
+            updated_at = datetime('now')",
+    )
+    .bind(user_id).bind(next).bind(total_recharged_inc)
+    .bind(next).bind(total_recharged_inc)
+    .execute(&mut *tx).await?;
+    record_ledger_entry(&mut tx, LedgerEntry {
+        user_id, track: "points", entry_type,
+        amount: delta, balance_after: next,
+        ref_type, ref_id, model, operator, note: None,
+    }).await?;
+    tx.commit().await?;
+    Ok(next)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -134,4 +190,21 @@ mod tests {
         .unwrap();
         assert_eq!(row, (7, "usd".into(), "recharge".into(), 10.0, 10.0));
     }
+
+    #[tokio::test]
+    async fn apply_points_delta_upserts_balance_and_writes_ledger() {
+        let pool = mem_pool().await;
+        // 首次 +100（充值）
+        let bal = apply_points_delta(&pool, 9, 100.0, "recharge", None, Some("system"), Some("recharge_record"), None).await.unwrap();
+        assert_eq!(bal, 100.0);
+        // 再 -30（消费）
+        let bal2 = apply_points_delta(&pool, 9, -30.0, "consume", Some("wf/video"), Some("system"), Some("workflow"), None).await.unwrap();
+        assert_eq!(bal2, 70.0);
+        // user_credits 缓存 = 70
+        let cached: f64 = sqlx::query_scalar("SELECT points_balance FROM user_credits WHERE user_id = 9").fetch_one(&pool).await.unwrap();
+        assert_eq!(cached, 70.0);
+        // ledger 两行
+        let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM credit_ledger WHERE user_id=9 AND track='points'").fetch_one(&pool).await.unwrap();
+        assert_eq!(n, 2);
+    }
 }