feat(api): points mall with atomic redemption and admin fulfillment

Platform-owned points products and a redemption order lifecycle kept separate
from cash orders. Redeeming locks the published product, reserves stock, creates
the order, debits points through the archived customer-accounts ledger with the
order as reference, and snapshots the line in one transaction; a failure leaves
no order, no stock change, and no ledger entry. Fulfilment moves only from
pending_fulfillment, and customers see only their own redemptions.

Demo seeding credits the demo customer through the same guarded credit path
with reason seed, once, so no balance is ever written absolutely.

Surfaces (admin console, mall points page) and product seeding follow.

apps/api/src/main.rs

@@ -1,4 +1,9 @@
-use vmall_api::{build_router, config::Config, seed::ensure_platform_admin, state};
+use vmall_api::{
+    build_router,
+    config::Config,
+    seed::{ensure_demo_points, ensure_platform_admin},
+    state,
+};
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
@@ -14,6 +19,7 @@ async fn main() -> anyhow::Result<()> {
     sqlx::migrate!("./migrations").run(&db).await?;
     let state = state::assemble(config.clone(), db).await?;
     ensure_platform_admin(&state).await?;
+    ensure_demo_points(&state).await?;
     let app = build_router(state);
 
     let listener = tokio::net::TcpListener::bind(("0.0.0.0", config.port)).await?;

apps/api/src/models.rs

@@ -343,6 +343,65 @@ pub struct Coupon {
 pub const COUPON_COLUMNS: &str = "id, user_id, template_id, shop_id, title, amount_minor, \
      threshold_minor, currency, starts_at, ends_at, status, order_id, claimed_at, redeemed_at";
 
+/// Redemption lifecycle; separate from cash orders.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Type)]
+#[sqlx(type_name = "integral_order_status", rename_all = "snake_case")]
+#[serde(rename_all = "snake_case")]
+pub enum IntegralOrderStatus {
+    PendingFulfillment,
+    Fulfilled,
+    Cancelled,
+}
+
+/// Platform-owned catalog item redeemable for points. Never a SKU.
+#[derive(Debug, Clone, Serialize, sqlx::FromRow)]
+pub struct IntegralProduct {
+    pub id: Uuid,
+    pub name: serde_json::Value,
+    pub subtitle: Option<serde_json::Value>,
+    pub content: Option<serde_json::Value>,
+    pub image: Option<String>,
+    pub points_price: i64,
+    pub stock: i32,
+    pub published: bool,
+    pub recommend: bool,
+    pub position: i32,
+    pub created_at: DateTime<Utc>,
+    pub updated_at: DateTime<Utc>,
+}
+
+pub const INTEGRAL_PRODUCT_COLUMNS: &str = "id, name, subtitle, content, image, points_price, \
+     stock, published, recommend, position, created_at, updated_at";
+
+#[derive(Debug, Clone, Serialize, sqlx::FromRow)]
+pub struct IntegralOrder {
+    pub id: Uuid,
+    pub order_no: String,
+    pub user_id: Uuid,
+    pub status: IntegralOrderStatus,
+    pub total_points: i64,
+    pub shipping_address: serde_json::Value,
+    pub created_at: DateTime<Utc>,
+    pub updated_at: DateTime<Utc>,
+}
+
+pub const INTEGRAL_ORDER_COLUMNS: &str =
+    "id, order_no, user_id, status, total_points, shipping_address, created_at, updated_at";
+
+#[derive(Debug, Clone, Serialize, sqlx::FromRow)]
+pub struct IntegralOrderItem {
+    pub id: Uuid,
+    pub order_id: Uuid,
+    pub product_id: Option<Uuid>,
+    pub name: serde_json::Value,
+    pub image: Option<String>,
+    pub points_price: i64,
+    pub qty: i32,
+}
+
+pub const INTEGRAL_ORDER_ITEM_COLUMNS: &str =
+    "id, order_id, product_id, name, image, points_price, qty";
+
 #[derive(Debug, Clone, Serialize, sqlx::FromRow)]
 pub struct AddressBookEntry {
     pub id: Uuid,

apps/api/src/modules/account/repo.rs

@@ -110,6 +110,26 @@ pub async fn credit_atomic(
 
 /// Append the audit fact for a balance update that already happened in `tx`.
 #[allow(clippy::too_many_arguments)]
+/// Whether any account for this user already carries an entry with `reason`.
+/// Used by demo seeding to stay idempotent without reading a balance.
+pub async fn has_entry_with_reason<'e, E: PgExecutor<'e>>(
+    exec: E,
+    user_id: Uuid,
+    reason: &str,
+) -> ApiResult<bool> {
+    Ok(sqlx::query_scalar(
+        "SELECT EXISTS(
+             SELECT 1 FROM customer_account_entries e
+             JOIN customer_accounts a ON a.id = e.account_id
+             WHERE a.user_id = $1 AND e.reason = $2
+         )",
+    )
+    .bind(user_id)
+    .bind(reason)
+    .fetch_one(exec)
+    .await?)
+}
+
 pub async fn insert_entry(
     tx: &mut PgConnection,
     account_id: Uuid,

apps/api/src/modules/account/service.rs

@@ -62,6 +62,27 @@ pub async fn debit(
     append(tx, account.id, -amount, updated.balance_minor, reason, reference).await
 }
 
+/// Credit once per `reason`, in its own transaction. Demo seeding uses this so
+/// a restart cannot grant twice and no balance is ever written absolutely.
+pub async fn credit_once(
+    state: &AppState,
+    user_id: Uuid,
+    kind: AccountKind,
+    currency: Option<&str>,
+    amount_minor: i64,
+    reason: &str,
+) -> ApiResult<Option<CustomerAccountEntry>> {
+    let mut tx = state.db.begin().await?;
+    ensure_accounts(&mut tx, user_id).await?;
+    if repo::has_entry_with_reason(&mut *tx, user_id, reason).await? {
+        tx.commit().await?;
+        return Ok(None);
+    }
+    let entry = credit(&mut tx, user_id, kind, currency, amount_minor, reason, None).await?;
+    tx.commit().await?;
+    Ok(Some(entry))
+}
+
 /// Move money from `available` to `frozen`; returns (available, frozen) entries.
 pub async fn freeze(
     tx: &mut PgConnection,

apps/api/src/modules/mod.rs

@@ -10,6 +10,7 @@ pub mod fulfillment;
 pub mod health;
 pub mod identity;
 pub mod order;
+pub mod points;
 pub mod shop;
 
 use axum::Router;
@@ -28,6 +29,7 @@ pub fn api_router() -> Router<AppState> {
         .merge(cart::router())
         .merge(coupon::router())
         .merge(order::router())
+        .merge(points::router())
         .merge(shop::router())
         .merge(fulfillment::router())
         .merge(billing::router())

apps/api/src/modules/points/dto.rs

@@ -0,0 +1,38 @@
+use serde::{Deserialize, Serialize};
+use uuid::Uuid;
+
+use crate::models::{IntegralOrder, IntegralOrderItem};
+
+/// Platform-admin create/update body for a points product.
+#[derive(Deserialize)]
+pub struct IntegralProductInput {
+    pub name: serde_json::Value,
+    #[serde(default)]
+    pub subtitle: Option<serde_json::Value>,
+    #[serde(default)]
+    pub content: Option<serde_json::Value>,
+    #[serde(default)]
+    pub image: Option<String>,
+    pub points_price: i64,
+    pub stock: i32,
+    #[serde(default)]
+    pub published: bool,
+    #[serde(default)]
+    pub recommend: bool,
+    #[serde(default)]
+    pub position: i32,
+}
+
+#[derive(Deserialize)]
+pub struct RedeemInput {
+    pub product_id: Uuid,
+    pub qty: i32,
+    pub shipping_address: crate::modules::order::AddressBody,
+}
+
+#[derive(Debug, Serialize)]
+pub struct RedemptionView {
+    #[serde(flatten)]
+    pub order: IntegralOrder,
+    pub items: Vec<IntegralOrderItem>,
+}

apps/api/src/modules/points/handlers.rs

@@ -0,0 +1,144 @@
+use axum::{
+    extract::{Path, Query, State},
+    http::StatusCode,
+    routing::{get, post, put},
+    Json, Router,
+};
+use serde::Deserialize;
+use uuid::Uuid;
+
+use crate::auth::AuthUser;
+use crate::error::ApiResult;
+use crate::http::Paged;
+use crate::models::{IntegralOrderStatus, IntegralProduct};
+use crate::state::AppState;
+
+use super::dto::{IntegralProductInput, RedeemInput, RedemptionView};
+use super::service;
+
+pub fn router() -> Router<AppState> {
+    Router::new()
+        // Public catalog; redemption and history are customer-scoped.
+        .route("/points/products", get(list_published))
+        .route("/points/redemptions", get(list_mine).post(redeem))
+        .route(
+            "/admin/points/products",
+            get(admin_list).post(admin_create),
+        )
+        .route("/admin/points/products/{id}", put(admin_update))
+        .route("/admin/points/products/{id}/publish", post(admin_publish))
+        .route("/admin/points/products/{id}/unpublish", post(admin_unpublish))
+        .route("/admin/points/orders", get(admin_orders))
+        .route("/admin/points/orders/{id}/fulfill", post(admin_fulfill))
+        .route("/admin/points/orders/{id}/cancel", post(admin_cancel))
+}
+
+async fn list_published(
+    State(state): State<AppState>,
+) -> ApiResult<Json<Vec<IntegralProduct>>> {
+    Ok(Json(service::list_published(&state).await?))
+}
+
+async fn list_mine(
+    State(state): State<AppState>,
+    auth: AuthUser,
+) -> ApiResult<Json<Vec<RedemptionView>>> {
+    auth.require_customer()?;
+    Ok(Json(service::list_mine(&state, auth.id).await?))
+}
+
+async fn redeem(
+    State(state): State<AppState>,
+    auth: AuthUser,
+    Json(body): Json<RedeemInput>,
+) -> ApiResult<(StatusCode, Json<RedemptionView>)> {
+    auth.require_customer()?;
+    Ok((
+        StatusCode::CREATED,
+        Json(service::redeem(&state, auth.id, body).await?),
+    ))
+}
+
+async fn admin_list(
+    State(state): State<AppState>,
+    auth: AuthUser,
+) -> ApiResult<Json<Vec<IntegralProduct>>> {
+    auth.require_admin()?;
+    Ok(Json(service::admin_list_products(&state).await?))
+}
+
+async fn admin_create(
+    State(state): State<AppState>,
+    auth: AuthUser,
+    Json(body): Json<IntegralProductInput>,
+) -> ApiResult<(StatusCode, Json<IntegralProduct>)> {
+    auth.require_admin()?;
+    Ok((
+        StatusCode::CREATED,
+        Json(service::admin_create_product(&state, body).await?),
+    ))
+}
+
+async fn admin_update(
+    State(state): State<AppState>,
+    auth: AuthUser,
+    Path(id): Path<Uuid>,
+    Json(body): Json<IntegralProductInput>,
+) -> ApiResult<Json<IntegralProduct>> {
+    auth.require_admin()?;
+    Ok(Json(service::admin_update_product(&state, id, body).await?))
+}
+
+async fn admin_publish(
+    State(state): State<AppState>,
+    auth: AuthUser,
+    Path(id): Path<Uuid>,
+) -> ApiResult<Json<IntegralProduct>> {
+    auth.require_admin()?;
+    Ok(Json(service::admin_set_published(&state, id, true).await?))
+}
+
+async fn admin_unpublish(
+    State(state): State<AppState>,
+    auth: AuthUser,
+    Path(id): Path<Uuid>,
+) -> ApiResult<Json<IntegralProduct>> {
+    auth.require_admin()?;
+    Ok(Json(service::admin_set_published(&state, id, false).await?))
+}
+
+#[derive(Deserialize)]
+struct RedemptionQuery {
+    page: Option<i64>,
+    per_page: Option<i64>,
+    status: Option<IntegralOrderStatus>,
+}
+
+async fn admin_orders(
+    State(state): State<AppState>,
+    auth: AuthUser,
+    Query(q): Query<RedemptionQuery>,
+) -> ApiResult<Json<Paged<RedemptionView>>> {
+    auth.require_admin()?;
+    Ok(Json(
+        service::admin_list_orders(&state, q.status, q.page, q.per_page).await?,
+    ))
+}
+
+async fn admin_fulfill(
+    State(state): State<AppState>,
+    auth: AuthUser,
+    Path(id): Path<Uuid>,
+) -> ApiResult<Json<RedemptionView>> {
+    auth.require_admin()?;
+    Ok(Json(service::admin_fulfill(&state, id).await?))
+}
+
+async fn admin_cancel(
+    State(state): State<AppState>,
+    auth: AuthUser,
+    Path(id): Path<Uuid>,
+) -> ApiResult<Json<RedemptionView>> {
+    auth.require_admin()?;
+    Ok(Json(service::admin_cancel(&state, id).await?))
+}

apps/api/src/modules/points/mod.rs

@@ -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()
+}

apps/api/src/modules/points/repo.rs

@@ -0,0 +1,269 @@
+use std::collections::HashMap;
+
+use sqlx::{PgConnection, PgExecutor, PgPool};
+use uuid::Uuid;
+
+use crate::error::{ApiError, ApiResult};
+use crate::http::{clamp_page, clamp_per_page, Paged};
+use crate::models::{
+    IntegralOrder, IntegralOrderItem, IntegralOrderStatus, IntegralProduct,
+    INTEGRAL_ORDER_COLUMNS, INTEGRAL_ORDER_ITEM_COLUMNS, INTEGRAL_PRODUCT_COLUMNS,
+};
+
+use super::dto::{IntegralProductInput, RedemptionView};
+
+// ---- catalog ----
+
+pub async fn list_published<'e, E: PgExecutor<'e>>(exec: E) -> ApiResult<Vec<IntegralProduct>> {
+    Ok(sqlx::query_as::<_, IntegralProduct>(&format!(
+        "SELECT {INTEGRAL_PRODUCT_COLUMNS} FROM integral_products
+          WHERE published = TRUE
+          ORDER BY position, created_at"
+    ))
+    .fetch_all(exec)
+    .await?)
+}
+
+pub async fn list_all<'e, E: PgExecutor<'e>>(exec: E) -> ApiResult<Vec<IntegralProduct>> {
+    Ok(sqlx::query_as::<_, IntegralProduct>(&format!(
+        "SELECT {INTEGRAL_PRODUCT_COLUMNS} FROM integral_products
+          ORDER BY position, created_at"
+    ))
+    .fetch_all(exec)
+    .await?)
+}
+
+/// Lock a published product for redemption. An unpublished or missing product
+/// is a 404 so customers cannot probe the draft catalog.
+pub async fn lock_published(
+    tx: &mut PgConnection,
+    id: Uuid,
+) -> ApiResult<IntegralProduct> {
+    sqlx::query_as::<_, IntegralProduct>(&format!(
+        "SELECT {INTEGRAL_PRODUCT_COLUMNS} FROM integral_products
+          WHERE id = $1 AND published = TRUE
+          FOR UPDATE"
+    ))
+    .bind(id)
+    .fetch_optional(&mut *tx)
+    .await?
+    .ok_or_else(|| ApiError::NotFound("points product".into()))
+}
+
+pub async fn insert_product(
+    tx: &mut PgConnection,
+    body: &IntegralProductInput,
+) -> ApiResult<IntegralProduct> {
+    Ok(sqlx::query_as::<_, IntegralProduct>(&format!(
+        "INSERT INTO integral_products
+             (name, subtitle, content, image, points_price, stock, published, recommend, position)
+         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
+         RETURNING {INTEGRAL_PRODUCT_COLUMNS}"
+    ))
+    .bind(&body.name)
+    .bind(&body.subtitle)
+    .bind(&body.content)
+    .bind(&body.image)
+    .bind(body.points_price)
+    .bind(body.stock)
+    .bind(body.published)
+    .bind(body.recommend)
+    .bind(body.position)
+    .fetch_one(&mut *tx)
+    .await?)
+}
+
+pub async fn update_product(
+    tx: &mut PgConnection,
+    id: Uuid,
+    body: &IntegralProductInput,
+) -> ApiResult<IntegralProduct> {
+    sqlx::query_as::<_, IntegralProduct>(&format!(
+        "UPDATE integral_products
+            SET name = $2, subtitle = $3, content = $4, image = $5, points_price = $6,
+                stock = $7, published = $8, recommend = $9, position = $10, updated_at = now()
+          WHERE id = $1
+         RETURNING {INTEGRAL_PRODUCT_COLUMNS}"
+    ))
+    .bind(id)
+    .bind(&body.name)
+    .bind(&body.subtitle)
+    .bind(&body.content)
+    .bind(&body.image)
+    .bind(body.points_price)
+    .bind(body.stock)
+    .bind(body.published)
+    .bind(body.recommend)
+    .bind(body.position)
+    .fetch_optional(&mut *tx)
+    .await?
+    .ok_or_else(|| ApiError::NotFound("points product".into()))
+}
+
+pub async fn set_published(
+    tx: &mut PgConnection,
+    id: Uuid,
+    published: bool,
+) -> ApiResult<IntegralProduct> {
+    sqlx::query_as::<_, IntegralProduct>(&format!(
+        "UPDATE integral_products SET published = $2, updated_at = now()
+          WHERE id = $1
+         RETURNING {INTEGRAL_PRODUCT_COLUMNS}"
+    ))
+    .bind(id)
+    .bind(published)
+    .fetch_optional(&mut *tx)
+    .await?
+    .ok_or_else(|| ApiError::NotFound("points product".into()))
+}
+
+/// Guarded counter decrement, same shape as SKU stock.
+pub async fn decrement_stock(tx: &mut PgConnection, id: Uuid, qty: i32) -> ApiResult<()> {
+    let result = sqlx::query(
+        "UPDATE integral_products SET stock = stock - $2, updated_at = now()
+          WHERE id = $1 AND stock >= $2",
+    )
+    .bind(id)
+    .bind(qty)
+    .execute(&mut *tx)
+    .await?;
+    if result.rows_affected() == 0 {
+        return Err(ApiError::Conflict("insufficient points stock".into()));
+    }
+    Ok(())
+}
+
+// ---- redemption orders ----
+
+pub async fn insert_order(
+    tx: &mut PgConnection,
+    user_id: Uuid,
+    total_points: i64,
+    address: &serde_json::Value,
+) -> ApiResult<IntegralOrder> {
+    Ok(sqlx::query_as::<_, IntegralOrder>(&format!(
+        "INSERT INTO integral_orders (order_no, user_id, total_points, shipping_address)
+         VALUES ('PM' || to_char(now(), 'YYMMDD')
+                     || lpad(nextval('integral_order_no_seq')::text, 6, '0'),
+                 $1, $2, $3)
+         RETURNING {INTEGRAL_ORDER_COLUMNS}"
+    ))
+    .bind(user_id)
+    .bind(total_points)
+    .bind(address)
+    .fetch_one(&mut *tx)
+    .await?)
+}
+
+pub async fn insert_item(
+    tx: &mut PgConnection,
+    order_id: Uuid,
+    product: &IntegralProduct,
+    qty: i32,
+) -> ApiResult<()> {
+    sqlx::query(
+        "INSERT INTO integral_order_items (order_id, product_id, name, image, points_price, qty)
+         VALUES ($1, $2, $3, $4, $5, $6)",
+    )
+    .bind(order_id)
+    .bind(product.id)
+    .bind(&product.name)
+    .bind(&product.image)
+    .bind(product.points_price)
+    .bind(qty)
+    .execute(&mut *tx)
+    .await?;
+    Ok(())
+}
+
+pub async fn list_for_user<'e, E: PgExecutor<'e>>(
+    exec: E,
+    user_id: Uuid,
+) -> ApiResult<Vec<IntegralOrder>> {
+    Ok(sqlx::query_as::<_, IntegralOrder>(&format!(
+        "SELECT {INTEGRAL_ORDER_COLUMNS} FROM integral_orders
+          WHERE user_id = $1 ORDER BY created_at DESC"
+    ))
+    .bind(user_id)
+    .fetch_all(exec)
+    .await?)
+}
+
+pub async fn list_all_page(
+    db: &PgPool,
+    status: Option<IntegralOrderStatus>,
+    page: Option<i64>,
+    per_page: Option<i64>,
+) -> ApiResult<Paged<IntegralOrder>> {
+    let page = clamp_page(page);
+    let per_page = clamp_per_page(per_page);
+    let total: i64 = sqlx::query_scalar(
+        "SELECT count(*) FROM integral_orders
+          WHERE ($1::integral_order_status IS NULL OR status = $1)",
+    )
+    .bind(status)
+    .fetch_one(db)
+    .await?;
+    let items = sqlx::query_as::<_, IntegralOrder>(&format!(
+        "SELECT {INTEGRAL_ORDER_COLUMNS} FROM integral_orders
+          WHERE ($1::integral_order_status IS NULL OR status = $1)
+          ORDER BY created_at DESC LIMIT $2 OFFSET $3"
+    ))
+    .bind(status)
+    .bind(per_page)
+    .bind((page - 1) * per_page)
+    .fetch_all(db)
+    .await?;
+    Ok(Paged {
+        items,
+        total,
+        page,
+        per_page,
+    })
+}
+
+/// Lock an order and move it only from the expected status.
+pub async fn transition(
+    tx: &mut PgConnection,
+    id: Uuid,
+    from: IntegralOrderStatus,
+    to: IntegralOrderStatus,
+) -> ApiResult<IntegralOrder> {
+    sqlx::query_as::<_, IntegralOrder>(&format!(
+        "UPDATE integral_orders SET status = $3, updated_at = now()
+          WHERE id = $1 AND status = $2
+         RETURNING {INTEGRAL_ORDER_COLUMNS}"
+    ))
+    .bind(id)
+    .bind(from)
+    .bind(to)
+    .fetch_optional(&mut *tx)
+    .await?
+    .ok_or_else(|| ApiError::Conflict("redemption order is not in the expected state".into()))
+}
+
+pub async fn attach_items(db: &PgPool, orders: Vec<IntegralOrder>) -> ApiResult<Vec<RedemptionView>> {
+    let ids: Vec<Uuid> = orders.iter().map(|o| o.id).collect();
+    let items = if ids.is_empty() {
+        Vec::new()
+    } else {
+        sqlx::query_as::<_, IntegralOrderItem>(&format!(
+            "SELECT {INTEGRAL_ORDER_ITEM_COLUMNS} FROM integral_order_items
+              WHERE order_id = ANY($1) ORDER BY id"
+        ))
+        .bind(&ids)
+        .fetch_all(db)
+        .await?
+    };
+    let mut by_order: HashMap<Uuid, Vec<IntegralOrderItem>> = HashMap::new();
+    for item in items {
+        by_order.entry(item.order_id).or_default().push(item);
+    }
+    Ok(orders
+        .into_iter()
+        .map(|order| RedemptionView {
+            items: by_order.remove(&order.id).unwrap_or_default(),
+            order,
+        })
+        .collect())
+}

apps/api/src/modules/points/service.rs

@@ -0,0 +1,189 @@
+use serde_json::Value;
+use uuid::Uuid;
+
+use crate::error::{ApiError, ApiResult};
+use crate::http::Paged;
+use crate::models::{AccountKind, IntegralOrderStatus, IntegralProduct};
+use crate::modules::account;
+use crate::state::AppState;
+
+use super::dto::{IntegralProductInput, RedeemInput, RedemptionView};
+use super::repo;
+
+/// Ledger reason recorded for a points spend.
+pub const REDEMPTION_REASON: &str = "integral_redemption";
+
+// ---- catalog ----
+
+pub async fn list_published(state: &AppState) -> ApiResult<Vec<IntegralProduct>> {
+    repo::list_published(&state.db).await
+}
+
+pub async fn admin_list_products(state: &AppState) -> ApiResult<Vec<IntegralProduct>> {
+    repo::list_all(&state.db).await
+}
+
+pub async fn admin_create_product(
+    state: &AppState,
+    body: IntegralProductInput,
+) -> ApiResult<IntegralProduct> {
+    validate_product(&body)?;
+    let mut tx = state.db.begin().await?;
+    let product = repo::insert_product(&mut tx, &body).await?;
+    tx.commit().await?;
+    Ok(product)
+}
+
+pub async fn admin_update_product(
+    state: &AppState,
+    id: Uuid,
+    body: IntegralProductInput,
+) -> ApiResult<IntegralProduct> {
+    validate_product(&body)?;
+    let mut tx = state.db.begin().await?;
+    let product = repo::update_product(&mut tx, id, &body).await?;
+    tx.commit().await?;
+    Ok(product)
+}
+
+pub async fn admin_set_published(
+    state: &AppState,
+    id: Uuid,
+    published: bool,
+) -> ApiResult<IntegralProduct> {
+    let mut tx = state.db.begin().await?;
+    let product = repo::set_published(&mut tx, id, published).await?;
+    tx.commit().await?;
+    Ok(product)
+}
+
+// ---- redemption ----
+
+/// Redeem points atomically: lock the published product, reserve stock, create
+/// the order, debit the points ledger with the order as reference, and snapshot
+/// the line. A failure leaves no order, no stock change, and no entry.
+pub async fn redeem(
+    state: &AppState,
+    user_id: Uuid,
+    body: RedeemInput,
+) -> ApiResult<RedemptionView> {
+    body.shipping_address.validate()?;
+    if body.qty <= 0 {
+        return Err(ApiError::BadRequest("qty must be positive".into()));
+    }
+    let mut tx = state.db.begin().await?;
+    let product = repo::lock_published(&mut tx, body.product_id).await?;
+    if body.qty > product.stock {
+        return Err(ApiError::Conflict("insufficient points stock".into()));
+    }
+    let total_points = product
+        .points_price
+        .checked_mul(body.qty as i64)
+        .ok_or_else(|| ApiError::BadRequest("quantity is too large".into()))?;
+
+    repo::decrement_stock(&mut tx, product.id, body.qty).await?;
+    let address = serde_json::to_value(&body.shipping_address).map_err(ApiError::internal)?;
+    let order = repo::insert_order(&mut tx, user_id, total_points, &address).await?;
+    // Points are never written directly: the archived account service debits
+    // with a guard and appends the matching entry.
+    account::service::debit(
+        &mut tx,
+        user_id,
+        AccountKind::Points,
+        None,
+        total_points,
+        REDEMPTION_REASON,
+        Some(("integral_order", order.id)),
+    )
+    .await?;
+    repo::insert_item(&mut tx, order.id, &product, body.qty).await?;
+    tx.commit().await?;
+
+    let mut views = repo::attach_items(&state.db, vec![order]).await?;
+    Ok(views.remove(0))
+}
+
+pub async fn list_mine(state: &AppState, user_id: Uuid) -> ApiResult<Vec<RedemptionView>> {
+    let orders = repo::list_for_user(&state.db, user_id).await?;
+    repo::attach_items(&state.db, orders).await
+}
+
+pub async fn admin_list_orders(
+    state: &AppState,
+    status: Option<IntegralOrderStatus>,
+    page: Option<i64>,
+    per_page: Option<i64>,
+) -> ApiResult<Paged<RedemptionView>> {
+    let page = repo::list_all_page(&state.db, status, page, per_page).await?;
+    let items = repo::attach_items(&state.db, page.items).await?;
+    Ok(Paged {
+        items,
+        total: page.total,
+        page: page.page,
+        per_page: page.per_page,
+    })
+}
+
+pub async fn admin_fulfill(state: &AppState, id: Uuid) -> ApiResult<RedemptionView> {
+    transition(
+        state,
+        id,
+        IntegralOrderStatus::PendingFulfillment,
+        IntegralOrderStatus::Fulfilled,
+    )
+    .await
+}
+
+/// Platform-only for now: a customer-initiated return needs a refund flow.
+pub async fn admin_cancel(state: &AppState, id: Uuid) -> ApiResult<RedemptionView> {
+    transition(
+        state,
+        id,
+        IntegralOrderStatus::PendingFulfillment,
+        IntegralOrderStatus::Cancelled,
+    )
+    .await
+}
+
+async fn transition(
+    state: &AppState,
+    id: Uuid,
+    from: IntegralOrderStatus,
+    to: IntegralOrderStatus,
+) -> ApiResult<RedemptionView> {
+    let mut tx = state.db.begin().await?;
+    let order = repo::transition(&mut tx, id, from, to).await?;
+    tx.commit().await?;
+    let mut views = repo::attach_items(&state.db, vec![order]).await?;
+    Ok(views.remove(0))
+}
+
+// ---- validation ----
+
+fn validate_product(body: &IntegralProductInput) -> ApiResult<()> {
+    bilingual(&body.name, "name")?;
+    if body.points_price <= 0 {
+        return Err(ApiError::BadRequest(
+            "points_price must be positive".into(),
+        ));
+    }
+    if body.stock < 0 {
+        return Err(ApiError::BadRequest("stock must not be negative".into()));
+    }
+    Ok(())
+}
+
+fn bilingual(label: &Value, field: &str) -> ApiResult<()> {
+    let ok = ["en", "zh"].iter().all(|code| {
+        label
+            .get(code)
+            .and_then(Value::as_str)
+            .is_some_and(|s| !s.trim().is_empty())
+    });
+    if !ok {
+        return Err(ApiError::BadRequest(format!(
+            "{field} needs non-empty en and zh"
+        )));
+    }
+    Ok(())
+}

apps/api/src/seed.rs

@@ -1,4 +1,8 @@
+use uuid::Uuid;
+
 use crate::auth::hash_password;
+use crate::models::AccountKind;
+use crate::modules::account;
 use crate::state::AppState;
 
 /// Idempotent dev seed: platform admin account.
@@ -24,3 +28,49 @@ pub async fn ensure_platform_admin(state: &AppState) -> anyhow::Result<()> {
     tracing::info!("seeded platform admin admin@vmall.local");
     Ok(())
 }
+
+pub const DEMO_CUSTOMER_EMAIL: &str = "customer@vmall.local";
+pub const DEMO_CUSTOMER_PASSWORD: &str = "customer123";
+/// Spendable demo points, enough for the cheapest seeded points product.
+pub const DEMO_POINTS: i64 = 5_000;
+/// Ledger reason for the demo grant; doubles as the idempotency key.
+pub const DEMO_POINTS_REASON: &str = "seed";
+
+/// Idempotent dev seed: the demo customer plus a spendable points balance.
+/// The grant runs through the account service, so it is one append-only entry
+/// with reason `seed` - never an absolute balance write - and a restart cannot
+/// grant it twice.
+pub async fn ensure_demo_points(state: &AppState) -> anyhow::Result<()> {
+    let user_id: Uuid = match sqlx::query_scalar("SELECT id FROM users WHERE email = $1")
+        .bind(DEMO_CUSTOMER_EMAIL)
+        .fetch_optional(&state.db)
+        .await?
+    {
+        Some(id) => id,
+        None => {
+            let hash = hash_password(DEMO_CUSTOMER_PASSWORD)?;
+            sqlx::query_scalar(
+                "INSERT INTO users (email, password_hash, display_name, role)
+                 VALUES ($1, $2, 'Demo Customer', 'customer') RETURNING id",
+            )
+            .bind(DEMO_CUSTOMER_EMAIL)
+            .bind(hash)
+            .fetch_one(&state.db)
+            .await?
+        }
+    };
+
+    let credited = account::service::credit_once(
+        state,
+        user_id,
+        AccountKind::Points,
+        None,
+        DEMO_POINTS,
+        DEMO_POINTS_REASON,
+    )
+    .await?;
+    if credited.is_some() {
+        tracing::info!(points = DEMO_POINTS, "seeded demo customer points");
+    }
+    Ok(())
+}