feat(api): add customer accounts with live summary and append-only ledger
One balance row per (user, kind, currency): available and frozen carry the platform base currency, points carries none. Monetary and points rows use separate partial unique indexes because a plain UNIQUE lets NULL repeat. Balance changes go through transactional primitives that debits guard with a conditional update, credits add atomically, and freeze/release move both sides in one transaction after locking rows by primary key. Every change appends an immutable entry holding its resulting balance. Registration and a migration backfill create the zero rows; GET /api/me/stats is the only public surface and no endpoint mutates a balance. The mall buyer center and points page drop the USER_STATS fixture for the shared contract; the fixture stays exported so the fixed-data adapter can still serve the account domain as a rollback path. Implements openspec change add-customer-accounts.
This commit is contained in:
@@ -242,6 +242,50 @@ pub struct Invoice {
|
||||
pub created_at: DateTime<Utc>,
|
||||
}
|
||||
|
||||
/// Balance bucket. `available`/`frozen` are monetary and carry a currency;
|
||||
/// `points` is an integer loyalty balance with no currency.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Type)]
|
||||
#[sqlx(type_name = "customer_account_kind", rename_all = "snake_case")]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum AccountKind {
|
||||
Available,
|
||||
Frozen,
|
||||
Points,
|
||||
}
|
||||
|
||||
impl AccountKind {
|
||||
pub fn is_monetary(self) -> bool {
|
||||
!matches!(self, AccountKind::Points)
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, sqlx::FromRow)]
|
||||
pub struct CustomerAccount {
|
||||
pub id: Uuid,
|
||||
pub user_id: Uuid,
|
||||
pub kind: AccountKind,
|
||||
pub currency: Option<String>,
|
||||
pub balance_minor: i64,
|
||||
}
|
||||
|
||||
pub const CUSTOMER_ACCOUNT_COLUMNS: &str = "id, user_id, kind, currency, balance_minor";
|
||||
|
||||
/// One immutable balance movement. Written in the same transaction as the
|
||||
/// guarded account update it describes.
|
||||
#[derive(Debug, Clone, Serialize, sqlx::FromRow)]
|
||||
pub struct CustomerAccountEntry {
|
||||
pub id: Uuid,
|
||||
pub account_id: Uuid,
|
||||
pub delta_minor: i64,
|
||||
pub balance_minor: i64,
|
||||
pub reason: String,
|
||||
pub reference_type: Option<String>,
|
||||
pub reference_id: Option<Uuid>,
|
||||
pub created_at: DateTime<Utc>,
|
||||
}
|
||||
|
||||
pub const CUSTOMER_ACCOUNT_ENTRY_COLUMNS: &str = "id, account_id, delta_minor, balance_minor, reason, reference_type, reference_id, created_at";
|
||||
|
||||
#[derive(Debug, Clone, Serialize, sqlx::FromRow)]
|
||||
pub struct AddressBookEntry {
|
||||
pub id: Uuid,
|
||||
|
||||
@@ -0,0 +1,12 @@
|
||||
use serde::Serialize;
|
||||
|
||||
/// Response of `GET /api/me/stats`: the customer's own money and points.
|
||||
/// Money is minor units in `currency`; points are integer units with no
|
||||
/// currency of their own.
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
pub struct AccountSummary {
|
||||
pub balance_minor: i64,
|
||||
pub frozen_minor: i64,
|
||||
pub currency: String,
|
||||
pub points: i64,
|
||||
}
|
||||
@@ -0,0 +1,19 @@
|
||||
use axum::{extract::State, routing::get, Json, Router};
|
||||
|
||||
use crate::auth::AuthUser;
|
||||
use crate::error::ApiResult;
|
||||
use crate::state::AppState;
|
||||
|
||||
use super::dto::AccountSummary;
|
||||
use super::service;
|
||||
|
||||
/// Read-only: balances are granted or spent by business flows, never by a
|
||||
/// client-facing mutation endpoint.
|
||||
pub fn router() -> Router<AppState> {
|
||||
Router::new().route("/me/stats", get(stats))
|
||||
}
|
||||
|
||||
async fn stats(State(state): State<AppState>, auth: AuthUser) -> ApiResult<Json<AccountSummary>> {
|
||||
auth.require_customer()?;
|
||||
Ok(Json(service::summary(&state, auth.id).await?))
|
||||
}
|
||||
@@ -0,0 +1,12 @@
|
||||
mod dto;
|
||||
mod handlers;
|
||||
mod repo;
|
||||
pub mod service;
|
||||
|
||||
use axum::Router;
|
||||
|
||||
use crate::state::AppState;
|
||||
|
||||
pub fn router() -> Router<AppState> {
|
||||
handlers::router()
|
||||
}
|
||||
@@ -0,0 +1,136 @@
|
||||
use sqlx::{PgConnection, PgExecutor};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::error::{ApiError, ApiResult};
|
||||
use crate::models::{
|
||||
CustomerAccount, CustomerAccountEntry, CUSTOMER_ACCOUNT_COLUMNS,
|
||||
CUSTOMER_ACCOUNT_ENTRY_COLUMNS,
|
||||
};
|
||||
|
||||
/// Monetary accounts are opened in the platform base currency. Multi-currency
|
||||
/// wallets would add rows here rather than change the summary contract.
|
||||
pub async fn base_currency<'e, E: PgExecutor<'e>>(exec: E) -> ApiResult<String> {
|
||||
sqlx::query_scalar("SELECT code FROM currencies WHERE is_base LIMIT 1")
|
||||
.fetch_optional(exec)
|
||||
.await?
|
||||
.ok_or_else(|| ApiError::Conflict("no base currency configured".into()))
|
||||
}
|
||||
|
||||
/// Idempotently create the user's zero-balance rows. Safe to re-run: the
|
||||
/// partial unique indexes make a repeated call a no-op.
|
||||
pub async fn ensure_accounts(
|
||||
tx: &mut PgConnection,
|
||||
user_id: Uuid,
|
||||
base_currency: &str,
|
||||
) -> ApiResult<()> {
|
||||
sqlx::query(
|
||||
"INSERT INTO customer_accounts (user_id, kind, currency)
|
||||
VALUES ($1, 'points', NULL),
|
||||
($1, 'available', $2),
|
||||
($1, 'frozen', $2)
|
||||
ON CONFLICT DO NOTHING",
|
||||
)
|
||||
.bind(user_id)
|
||||
.bind(base_currency)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn list_for_user<'e, E: PgExecutor<'e>>(
|
||||
exec: E,
|
||||
user_id: Uuid,
|
||||
) -> ApiResult<Vec<CustomerAccount>> {
|
||||
Ok(sqlx::query_as::<_, CustomerAccount>(&format!(
|
||||
"SELECT {CUSTOMER_ACCOUNT_COLUMNS}
|
||||
FROM customer_accounts
|
||||
WHERE user_id = $1
|
||||
ORDER BY kind"
|
||||
))
|
||||
.bind(user_id)
|
||||
.fetch_all(exec)
|
||||
.await?)
|
||||
}
|
||||
|
||||
/// Lock every account row for one user before a two-sided transfer. Ordered by
|
||||
/// primary key so concurrent transfers take locks in the same order.
|
||||
pub async fn lock_for_user(
|
||||
tx: &mut PgConnection,
|
||||
user_id: Uuid,
|
||||
) -> ApiResult<Vec<CustomerAccount>> {
|
||||
Ok(sqlx::query_as::<_, CustomerAccount>(&format!(
|
||||
"SELECT {CUSTOMER_ACCOUNT_COLUMNS}
|
||||
FROM customer_accounts
|
||||
WHERE user_id = $1
|
||||
ORDER BY id
|
||||
FOR UPDATE"
|
||||
))
|
||||
.bind(user_id)
|
||||
.fetch_all(&mut *tx)
|
||||
.await?)
|
||||
}
|
||||
|
||||
/// Conditional decrement: the predicate proves sufficient balance, so a stale
|
||||
/// read cannot drive the stored balance negative.
|
||||
pub async fn debit_guarded(
|
||||
tx: &mut PgConnection,
|
||||
account_id: Uuid,
|
||||
amount_minor: i64,
|
||||
) -> ApiResult<CustomerAccount> {
|
||||
sqlx::query_as::<_, CustomerAccount>(&format!(
|
||||
"UPDATE customer_accounts
|
||||
SET balance_minor = balance_minor - $2, updated_at = now()
|
||||
WHERE id = $1 AND balance_minor >= $2
|
||||
RETURNING {CUSTOMER_ACCOUNT_COLUMNS}"
|
||||
))
|
||||
.bind(account_id)
|
||||
.bind(amount_minor)
|
||||
.fetch_optional(&mut *tx)
|
||||
.await?
|
||||
.ok_or_else(|| ApiError::Conflict("insufficient balance".into()))
|
||||
}
|
||||
|
||||
/// Unconditional atomic addition. Backfills use this shape, never a read-then-set.
|
||||
pub async fn credit_atomic(
|
||||
tx: &mut PgConnection,
|
||||
account_id: Uuid,
|
||||
amount_minor: i64,
|
||||
) -> ApiResult<CustomerAccount> {
|
||||
Ok(sqlx::query_as::<_, CustomerAccount>(&format!(
|
||||
"UPDATE customer_accounts
|
||||
SET balance_minor = balance_minor + $2, updated_at = now()
|
||||
WHERE id = $1
|
||||
RETURNING {CUSTOMER_ACCOUNT_COLUMNS}"
|
||||
))
|
||||
.bind(account_id)
|
||||
.bind(amount_minor)
|
||||
.fetch_one(&mut *tx)
|
||||
.await?)
|
||||
}
|
||||
|
||||
/// Append the audit fact for a balance update that already happened in `tx`.
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub async fn insert_entry(
|
||||
tx: &mut PgConnection,
|
||||
account_id: Uuid,
|
||||
delta_minor: i64,
|
||||
balance_minor: i64,
|
||||
reason: &str,
|
||||
reference_type: Option<&str>,
|
||||
reference_id: Option<Uuid>,
|
||||
) -> ApiResult<CustomerAccountEntry> {
|
||||
Ok(sqlx::query_as::<_, CustomerAccountEntry>(&format!(
|
||||
"INSERT INTO customer_account_entries
|
||||
(account_id, delta_minor, balance_minor, reason, reference_type, reference_id)
|
||||
VALUES ($1, $2, $3, $4, $5, $6)
|
||||
RETURNING {CUSTOMER_ACCOUNT_ENTRY_COLUMNS}"
|
||||
))
|
||||
.bind(account_id)
|
||||
.bind(delta_minor)
|
||||
.bind(balance_minor)
|
||||
.bind(reason)
|
||||
.bind(reference_type)
|
||||
.bind(reference_id)
|
||||
.fetch_one(&mut *tx)
|
||||
.await?)
|
||||
}
|
||||
@@ -0,0 +1,213 @@
|
||||
use sqlx::PgConnection;
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::error::{ApiError, ApiResult};
|
||||
use crate::models::{AccountKind, CustomerAccount, CustomerAccountEntry};
|
||||
use crate::state::AppState;
|
||||
|
||||
use super::dto::AccountSummary;
|
||||
use super::repo;
|
||||
|
||||
/// Optional business reference stored on a ledger entry, e.g.
|
||||
/// `("integral_order", order_id)`.
|
||||
pub type EntryRef<'a> = Option<(&'a str, Uuid)>;
|
||||
|
||||
/// Create the zero-balance rows a customer needs, inside the caller's
|
||||
/// transaction. Registration calls this next to the user insert.
|
||||
pub async fn ensure_accounts(tx: &mut PgConnection, user_id: Uuid) -> ApiResult<()> {
|
||||
let base = repo::base_currency(&mut *tx).await?;
|
||||
repo::ensure_accounts(tx, user_id, &base).await
|
||||
}
|
||||
|
||||
/// `GET /api/me/stats`. There is no public balance mutation endpoint.
|
||||
pub async fn summary(state: &AppState, user_id: Uuid) -> ApiResult<AccountSummary> {
|
||||
let mut tx = state.db.begin().await?;
|
||||
// Defensive: a user created outside registration still gets a summary.
|
||||
ensure_accounts(&mut tx, user_id).await?;
|
||||
let accounts = repo::list_for_user(&mut *tx, user_id).await?;
|
||||
tx.commit().await?;
|
||||
summarize(&accounts)
|
||||
}
|
||||
|
||||
/// Add `amount_minor` and append the matching entry. Caller owns the transaction
|
||||
/// so a business flow can commit the balance with its own rows.
|
||||
pub async fn credit(
|
||||
tx: &mut PgConnection,
|
||||
user_id: Uuid,
|
||||
kind: AccountKind,
|
||||
currency: Option<&str>,
|
||||
amount_minor: i64,
|
||||
reason: &str,
|
||||
reference: EntryRef<'_>,
|
||||
) -> ApiResult<CustomerAccountEntry> {
|
||||
let amount = positive(amount_minor)?;
|
||||
let account = account_for(tx, user_id, kind, currency).await?;
|
||||
let updated = repo::credit_atomic(tx, account.id, amount).await?;
|
||||
append(tx, account.id, amount, updated.balance_minor, reason, reference).await
|
||||
}
|
||||
|
||||
/// Subtract `amount_minor` only when the account can cover it; 409 otherwise.
|
||||
pub async fn debit(
|
||||
tx: &mut PgConnection,
|
||||
user_id: Uuid,
|
||||
kind: AccountKind,
|
||||
currency: Option<&str>,
|
||||
amount_minor: i64,
|
||||
reason: &str,
|
||||
reference: EntryRef<'_>,
|
||||
) -> ApiResult<CustomerAccountEntry> {
|
||||
let amount = positive(amount_minor)?;
|
||||
let account = account_for(tx, user_id, kind, currency).await?;
|
||||
let updated = repo::debit_guarded(tx, account.id, amount).await?;
|
||||
append(tx, account.id, -amount, updated.balance_minor, reason, reference).await
|
||||
}
|
||||
|
||||
/// Move money from `available` to `frozen`; returns (available, frozen) entries.
|
||||
pub async fn freeze(
|
||||
tx: &mut PgConnection,
|
||||
user_id: Uuid,
|
||||
amount_minor: i64,
|
||||
reason: &str,
|
||||
reference: EntryRef<'_>,
|
||||
) -> ApiResult<(CustomerAccountEntry, CustomerAccountEntry)> {
|
||||
transfer(
|
||||
tx,
|
||||
user_id,
|
||||
AccountKind::Available,
|
||||
AccountKind::Frozen,
|
||||
amount_minor,
|
||||
reason,
|
||||
reference,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
/// Move money from `frozen` back to `available`.
|
||||
pub async fn release(
|
||||
tx: &mut PgConnection,
|
||||
user_id: Uuid,
|
||||
amount_minor: i64,
|
||||
reason: &str,
|
||||
reference: EntryRef<'_>,
|
||||
) -> ApiResult<(CustomerAccountEntry, CustomerAccountEntry)> {
|
||||
transfer(
|
||||
tx,
|
||||
user_id,
|
||||
AccountKind::Frozen,
|
||||
AccountKind::Available,
|
||||
amount_minor,
|
||||
reason,
|
||||
reference,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
/// Two-sided move. Locks both rows first, then debits the source with a guard
|
||||
/// and credits the destination atomically, so the pair always balances.
|
||||
async fn transfer(
|
||||
tx: &mut PgConnection,
|
||||
user_id: Uuid,
|
||||
from_kind: AccountKind,
|
||||
to_kind: AccountKind,
|
||||
amount_minor: i64,
|
||||
reason: &str,
|
||||
reference: EntryRef<'_>,
|
||||
) -> ApiResult<(CustomerAccountEntry, CustomerAccountEntry)> {
|
||||
let amount = positive(amount_minor)?;
|
||||
let accounts = repo::lock_for_user(tx, user_id).await?;
|
||||
let from = find(&accounts, from_kind)?.id;
|
||||
let to = find(&accounts, to_kind)?.id;
|
||||
|
||||
let from_updated = repo::debit_guarded(tx, from, amount).await?;
|
||||
let to_updated = repo::credit_atomic(tx, to, amount).await?;
|
||||
|
||||
let out = append(tx, from, -amount, from_updated.balance_minor, reason, reference).await?;
|
||||
let into = append(tx, to, amount, to_updated.balance_minor, reason, reference).await?;
|
||||
Ok((out, into))
|
||||
}
|
||||
|
||||
async fn append(
|
||||
tx: &mut PgConnection,
|
||||
account_id: Uuid,
|
||||
delta_minor: i64,
|
||||
balance_minor: i64,
|
||||
reason: &str,
|
||||
reference: EntryRef<'_>,
|
||||
) -> ApiResult<CustomerAccountEntry> {
|
||||
let (reference_type, reference_id) = match reference {
|
||||
Some((kind, id)) => (Some(kind), Some(id)),
|
||||
None => (None, None),
|
||||
};
|
||||
repo::insert_entry(
|
||||
tx,
|
||||
account_id,
|
||||
delta_minor,
|
||||
balance_minor,
|
||||
reason,
|
||||
reference_type,
|
||||
reference_id,
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
async fn account_for(
|
||||
tx: &mut PgConnection,
|
||||
user_id: Uuid,
|
||||
kind: AccountKind,
|
||||
currency: Option<&str>,
|
||||
) -> ApiResult<CustomerAccount> {
|
||||
let wanted = resolve_currency(tx, kind, currency).await?;
|
||||
repo::list_for_user(&mut *tx, user_id)
|
||||
.await?
|
||||
.into_iter()
|
||||
.find(|account| account.kind == kind && account.currency == wanted)
|
||||
.ok_or_else(|| ApiError::NotFound("customer account".into()))
|
||||
}
|
||||
|
||||
async fn resolve_currency(
|
||||
tx: &mut PgConnection,
|
||||
kind: AccountKind,
|
||||
currency: Option<&str>,
|
||||
) -> ApiResult<Option<String>> {
|
||||
if !kind.is_monetary() {
|
||||
return Ok(None);
|
||||
}
|
||||
match currency {
|
||||
Some(code) => Ok(Some(code.to_uppercase())),
|
||||
None => Ok(Some(repo::base_currency(&mut *tx).await?)),
|
||||
}
|
||||
}
|
||||
|
||||
fn summarize(accounts: &[CustomerAccount]) -> ApiResult<AccountSummary> {
|
||||
let available = find(accounts, AccountKind::Available)?;
|
||||
let frozen = find(accounts, AccountKind::Frozen)?;
|
||||
let points = find(accounts, AccountKind::Points)?;
|
||||
let currency = available
|
||||
.currency
|
||||
.clone()
|
||||
.ok_or_else(|| {
|
||||
ApiError::Internal(anyhow::anyhow!("available account has no currency"))
|
||||
})?;
|
||||
Ok(AccountSummary {
|
||||
balance_minor: available.balance_minor,
|
||||
frozen_minor: frozen.balance_minor,
|
||||
currency,
|
||||
points: points.balance_minor,
|
||||
})
|
||||
}
|
||||
|
||||
fn find(accounts: &[CustomerAccount], kind: AccountKind) -> ApiResult<&CustomerAccount> {
|
||||
accounts
|
||||
.iter()
|
||||
.find(|account| account.kind == kind)
|
||||
.ok_or_else(|| ApiError::NotFound("customer account".into()))
|
||||
}
|
||||
|
||||
fn positive(amount_minor: i64) -> ApiResult<i64> {
|
||||
if amount_minor <= 0 {
|
||||
return Err(ApiError::BadRequest(
|
||||
"amount_minor must be positive".into(),
|
||||
));
|
||||
}
|
||||
Ok(amount_minor)
|
||||
}
|
||||
@@ -5,6 +5,7 @@ use crate::auth::{hash_password, issue_token, verify_password};
|
||||
use crate::error::{unique_conflict, ApiError, ApiResult};
|
||||
use crate::http::Paged;
|
||||
use crate::models::{User, UserPublic, UserRole};
|
||||
use crate::modules::account;
|
||||
use crate::state::AppState;
|
||||
|
||||
use super::repo;
|
||||
@@ -29,9 +30,14 @@ pub async fn register(state: &AppState, input: RegisterInput) -> ApiResult<(User
|
||||
return Err(ApiError::BadRequest("display_name is required".into()));
|
||||
}
|
||||
let hash = hash_password(&input.password)?;
|
||||
let user = repo::insert_customer(&state.db, &email, &hash, input.display_name.trim())
|
||||
// User and their zero-balance account rows commit together so a registered
|
||||
// customer always has a summary.
|
||||
let mut tx = state.db.begin().await?;
|
||||
let user = repo::insert_customer(&mut *tx, &email, &hash, input.display_name.trim())
|
||||
.await
|
||||
.map_err(|e| unique_conflict(e, "email already registered"))?;
|
||||
account::service::ensure_accounts(&mut tx, user.id).await?;
|
||||
tx.commit().await?;
|
||||
let payload = auth_payload(state, &user)?;
|
||||
Ok((UserPublic::from(user), payload))
|
||||
}
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
pub mod account;
|
||||
pub mod address;
|
||||
pub mod billing;
|
||||
pub mod cart;
|
||||
@@ -17,6 +18,7 @@ use crate::state::AppState;
|
||||
pub fn api_router() -> Router<AppState> {
|
||||
Router::new()
|
||||
.merge(health::router())
|
||||
.merge(account::router())
|
||||
.merge(address::router())
|
||||
.merge(identity::router())
|
||||
.merge(currency::router())
|
||||
|
||||
Reference in New Issue
Block a user