feat(api): flash sales resolved server-side at checkout

Shop-owned timed sessions carry SKU activity items with their own reserved
inventory and per-customer limit. Checkout resolves eligibility on the server,
locks candidate items by primary key after the SKU locks, and splits a cart line
into an activity-priced item plus a standard-priced remainder, so every unit
price snapshot is honest and a customer cannot exceed the limit.

Reserved activity stock and SKU stock decrement together under guards, and a
pending-payment cancellation restores both plus any redeemed coupon. Coupons are
rejected on a shop order that applied activity pricing, and a SKU cannot join
two overlapping enabled sessions.

The overlap check against group buying is present but dormant: that capability
lands later, so the check activates only once its table exists.

Surfaces (shop-admin, mall) and seeding follow.
This commit is contained in:
2026-09-18 12:47:46 +00:00
parent 82c1e8e4bb
commit 393e7f7209
12 changed files with 1521 additions and 36 deletions
+36
View File
@@ -208,6 +208,8 @@ pub struct OrderItem {
pub image: Option<String>,
pub unit_price_minor: i64,
pub qty: i32,
/// Set when this line received flash-sale pricing.
pub flash_sale_item_id: Option<Uuid>,
}
#[derive(Debug, Clone, Serialize, sqlx::FromRow)]
@@ -402,6 +404,40 @@ pub struct IntegralOrderItem {
pub const INTEGRAL_ORDER_ITEM_COLUMNS: &str =
"id, order_id, product_id, name, image, points_price, qty";
/// Shop-owned timed flash-sale session.
#[derive(Debug, Clone, Serialize, sqlx::FromRow)]
pub struct FlashSaleSession {
pub id: Uuid,
pub shop_id: Uuid,
pub label: serde_json::Value,
pub starts_at: DateTime<Utc>,
pub ends_at: DateTime<Utc>,
pub enabled: bool,
pub created_at: DateTime<Utc>,
pub updated_at: DateTime<Utc>,
}
pub const FLASH_SALE_SESSION_COLUMNS: &str =
"id, shop_id, label, starts_at, ends_at, enabled, created_at, updated_at";
/// SKU-level activity item with its own reserved inventory and customer limit.
#[derive(Debug, Clone, Serialize, sqlx::FromRow)]
pub struct FlashSaleItem {
pub id: Uuid,
pub session_id: Uuid,
pub sku_id: Uuid,
pub sale_price_minor: i64,
pub currency: String,
pub reserved_stock: i32,
pub sold_count: i32,
pub per_customer_limit: i32,
pub created_at: DateTime<Utc>,
pub updated_at: DateTime<Utc>,
}
pub const FLASH_SALE_ITEM_COLUMNS: &str = "id, session_id, sku_id, sale_price_minor, currency, \
reserved_stock, sold_count, per_customer_limit, created_at, updated_at";
#[derive(Debug, Clone, Serialize, sqlx::FromRow)]
pub struct AddressBookEntry {
pub id: Uuid,
+86
View File
@@ -0,0 +1,86 @@
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use uuid::Uuid;
use crate::models::FlashSaleSession;
#[derive(Deserialize)]
pub struct SessionInput {
pub label: serde_json::Value,
pub starts_at: DateTime<Utc>,
pub ends_at: DateTime<Utc>,
#[serde(default = "default_enabled")]
pub enabled: bool,
}
#[derive(Deserialize)]
pub struct ItemInput {
pub sku_id: Uuid,
pub sale_price_minor: i64,
pub currency: String,
pub reserved_stock: i32,
pub per_customer_limit: i32,
}
fn default_enabled() -> bool {
true
}
/// Shop-side item joined with the SKU and catalog price it discounts.
#[derive(Debug, Serialize, sqlx::FromRow)]
pub struct ShopItemView {
pub id: Uuid,
pub session_id: Uuid,
pub sku_id: Uuid,
pub sku_code: String,
pub product_name: serde_json::Value,
pub sale_price_minor: i64,
pub currency: String,
pub original_price_minor: i64,
pub original_currency: String,
pub reserved_stock: i32,
pub sold_count: i32,
pub per_customer_limit: i32,
}
#[derive(Debug, Serialize)]
pub struct ShopSessionView {
#[serde(flatten)]
pub session: FlashSaleSession,
pub items: Vec<ShopItemView>,
}
/// Public discovery item: everything a shopper needs to add it to the cart.
#[derive(Debug, Serialize, sqlx::FromRow)]
pub struct PublicItemView {
pub id: Uuid,
pub session_id: Uuid,
pub sku_id: Uuid,
pub product_id: Uuid,
pub product_slug: String,
pub product_name: serde_json::Value,
pub image: Option<String>,
pub sku_code: String,
pub sale_price_minor: i64,
pub currency: String,
pub original_price_minor: i64,
pub original_currency: String,
pub reserved_stock: i32,
pub sold_count: i32,
pub per_customer_limit: i32,
}
#[derive(Debug, Serialize)]
pub struct PublicSessionView {
#[serde(flatten)]
pub session: FlashSaleSession,
pub items: Vec<PublicItemView>,
}
/// The part of a cart line that received activity pricing.
#[derive(Debug, Clone, Copy)]
pub struct LineActivity {
pub item_id: Uuid,
pub unit_price_minor: i64,
pub qty: i32,
}
+114
View File
@@ -0,0 +1,114 @@
use axum::{
extract::{Path, State},
http::StatusCode,
routing::{get, post, put},
Json, Router,
};
use uuid::Uuid;
use crate::auth::AuthUser;
use crate::error::ApiResult;
use crate::models::{FlashSaleItem, FlashSaleSession};
use crate::state::AppState;
use super::dto::{ItemInput, PublicSessionView, SessionInput, ShopSessionView};
use super::service;
pub fn router() -> Router<AppState> {
Router::new()
// Public: active sessions and their purchasable items.
.route("/flash-sales", get(public_active))
.route("/shop/flash-sales", get(shop_list).post(shop_create))
.route(
"/shop/flash-sales/{id}",
put(shop_update).delete(shop_delete),
)
.route("/shop/flash-sales/{id}/items", post(shop_add_item))
.route(
"/shop/flash-sale-items/{id}",
put(shop_update_item).delete(shop_delete_item),
)
}
async fn public_active(
State(state): State<AppState>,
) -> ApiResult<Json<Vec<PublicSessionView>>> {
Ok(Json(service::public_active(&state).await?))
}
async fn shop_list(
State(state): State<AppState>,
auth: AuthUser,
) -> ApiResult<Json<Vec<ShopSessionView>>> {
let shop_id = auth.require_shop()?;
Ok(Json(service::list_for_shop(&state, shop_id).await?))
}
async fn shop_create(
State(state): State<AppState>,
auth: AuthUser,
Json(body): Json<SessionInput>,
) -> ApiResult<(StatusCode, Json<FlashSaleSession>)> {
let shop_id = auth.require_shop()?;
Ok((
StatusCode::CREATED,
Json(service::create_session(&state, shop_id, body).await?),
))
}
async fn shop_update(
State(state): State<AppState>,
auth: AuthUser,
Path(id): Path<Uuid>,
Json(body): Json<SessionInput>,
) -> ApiResult<Json<FlashSaleSession>> {
let shop_id = auth.require_shop()?;
Ok(Json(
service::update_session(&state, shop_id, id, body).await?,
))
}
async fn shop_delete(
State(state): State<AppState>,
auth: AuthUser,
Path(id): Path<Uuid>,
) -> ApiResult<StatusCode> {
let shop_id = auth.require_shop()?;
service::delete_session(&state, shop_id, id).await?;
Ok(StatusCode::NO_CONTENT)
}
async fn shop_add_item(
State(state): State<AppState>,
auth: AuthUser,
Path(id): Path<Uuid>,
Json(body): Json<ItemInput>,
) -> ApiResult<(StatusCode, Json<FlashSaleItem>)> {
let shop_id = auth.require_shop()?;
Ok((
StatusCode::CREATED,
Json(service::add_item(&state, shop_id, id, body).await?),
))
}
async fn shop_update_item(
State(state): State<AppState>,
auth: AuthUser,
Path(id): Path<Uuid>,
Json(body): Json<ItemInput>,
) -> ApiResult<Json<FlashSaleItem>> {
let shop_id = auth.require_shop()?;
Ok(Json(
service::update_item(&state, shop_id, id, body).await?,
))
}
async fn shop_delete_item(
State(state): State<AppState>,
auth: AuthUser,
Path(id): Path<Uuid>,
) -> ApiResult<StatusCode> {
let shop_id = auth.require_shop()?;
service::delete_item(&state, shop_id, id).await?;
Ok(StatusCode::NO_CONTENT)
}
+12
View File
@@ -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()
}
+409
View File
@@ -0,0 +1,409 @@
use sqlx::{PgConnection, PgExecutor};
use uuid::Uuid;
use crate::error::{ApiError, ApiResult};
use crate::models::{
FlashSaleItem, FlashSaleSession, FLASH_SALE_ITEM_COLUMNS, FLASH_SALE_SESSION_COLUMNS,
};
use super::dto::{ItemInput, PublicItemView, SessionInput, ShopItemView};
/// Qualified item columns for queries that join `flash_sale_sessions`, whose
/// own `id` would otherwise make bare column names ambiguous.
const ITEM_COLS_Q: &str = "i.id, i.session_id, i.sku_id, i.sale_price_minor, i.currency, \
i.reserved_stock, i.sold_count, i.per_customer_limit, i.created_at, i.updated_at";
// ---- sessions ----
pub async fn currency_enabled<'e, E: PgExecutor<'e>>(exec: E, code: &str) -> ApiResult<bool> {
Ok(sqlx::query_scalar(
"SELECT EXISTS(SELECT 1 FROM currencies WHERE code = $1 AND enabled = TRUE)",
)
.bind(code)
.fetch_one(exec)
.await?)
}
pub async fn list_sessions_for_shop<'e, E: PgExecutor<'e>>(
exec: E,
shop_id: Uuid,
) -> ApiResult<Vec<FlashSaleSession>> {
Ok(sqlx::query_as::<_, FlashSaleSession>(&format!(
"SELECT {FLASH_SALE_SESSION_COLUMNS} FROM flash_sale_sessions
WHERE shop_id = $1 ORDER BY starts_at DESC"
))
.bind(shop_id)
.fetch_all(exec)
.await?)
}
pub async fn get_session_own<'e, E: PgExecutor<'e>>(
exec: E,
shop_id: Uuid,
id: Uuid,
) -> ApiResult<FlashSaleSession> {
sqlx::query_as::<_, FlashSaleSession>(&format!(
"SELECT {FLASH_SALE_SESSION_COLUMNS} FROM flash_sale_sessions
WHERE id = $1 AND shop_id = $2"
))
.bind(id)
.bind(shop_id)
.fetch_optional(exec)
.await?
.ok_or_else(|| ApiError::NotFound("flash-sale session".into()))
}
pub async fn insert_session(
tx: &mut PgConnection,
shop_id: Uuid,
body: &SessionInput,
) -> ApiResult<FlashSaleSession> {
Ok(sqlx::query_as::<_, FlashSaleSession>(&format!(
"INSERT INTO flash_sale_sessions (shop_id, label, starts_at, ends_at, enabled)
VALUES ($1, $2, $3, $4, $5)
RETURNING {FLASH_SALE_SESSION_COLUMNS}"
))
.bind(shop_id)
.bind(&body.label)
.bind(body.starts_at)
.bind(body.ends_at)
.bind(body.enabled)
.fetch_one(&mut *tx)
.await?)
}
pub async fn update_session(
tx: &mut PgConnection,
shop_id: Uuid,
id: Uuid,
body: &SessionInput,
) -> ApiResult<FlashSaleSession> {
sqlx::query_as::<_, FlashSaleSession>(&format!(
"UPDATE flash_sale_sessions
SET label = $3, starts_at = $4, ends_at = $5, enabled = $6, updated_at = now()
WHERE id = $1 AND shop_id = $2
RETURNING {FLASH_SALE_SESSION_COLUMNS}"
))
.bind(id)
.bind(shop_id)
.bind(&body.label)
.bind(body.starts_at)
.bind(body.ends_at)
.bind(body.enabled)
.fetch_optional(&mut *tx)
.await?
.ok_or_else(|| ApiError::NotFound("flash-sale session".into()))
}
pub async fn delete_session(tx: &mut PgConnection, shop_id: Uuid, id: Uuid) -> ApiResult<()> {
let result = sqlx::query("DELETE FROM flash_sale_sessions WHERE id = $1 AND shop_id = $2")
.bind(id)
.bind(shop_id)
.execute(&mut *tx)
.await?;
if result.rows_affected() == 0 {
return Err(ApiError::NotFound("flash-sale session".into()));
}
Ok(())
}
// ---- activity items ----
pub async fn list_items_for_shop<'e, E: PgExecutor<'e>>(
exec: E,
shop_id: Uuid,
) -> ApiResult<Vec<ShopItemView>> {
Ok(sqlx::query_as::<_, ShopItemView>(
"SELECT i.id, i.session_id, i.sku_id, s.sku_code, p.name AS product_name,
i.sale_price_minor, i.currency,
s.price_minor AS original_price_minor, s.currency AS original_currency,
i.reserved_stock, i.sold_count, i.per_customer_limit
FROM flash_sale_items i
JOIN flash_sale_sessions fs ON fs.id = i.session_id
JOIN skus s ON s.id = i.sku_id
JOIN products p ON p.id = s.product_id
WHERE fs.shop_id = $1
ORDER BY i.created_at",
)
.bind(shop_id)
.fetch_all(exec)
.await?)
}
pub async fn get_item_own<'e, E: PgExecutor<'e>>(
exec: E,
shop_id: Uuid,
id: Uuid,
) -> ApiResult<FlashSaleItem> {
sqlx::query_as::<_, FlashSaleItem>(&format!(
"SELECT {ITEM_COLS_Q} FROM flash_sale_items i
JOIN flash_sale_sessions s ON s.id = i.session_id
WHERE i.id = $1 AND s.shop_id = $2"
))
.bind(id)
.bind(shop_id)
.fetch_optional(exec)
.await?
.ok_or_else(|| ApiError::NotFound("flash-sale item".into()))
}
pub async fn insert_item(
tx: &mut PgConnection,
session_id: Uuid,
body: &ItemInput,
) -> Result<FlashSaleItem, sqlx::Error> {
sqlx::query_as::<_, FlashSaleItem>(&format!(
"INSERT INTO flash_sale_items
(session_id, sku_id, sale_price_minor, currency, reserved_stock, per_customer_limit)
VALUES ($1, $2, $3, $4, $5, $6)
RETURNING {FLASH_SALE_ITEM_COLUMNS}"
))
.bind(session_id)
.bind(body.sku_id)
.bind(body.sale_price_minor)
.bind(body.currency.to_uppercase())
.bind(body.reserved_stock)
.bind(body.per_customer_limit)
.fetch_one(&mut *tx)
.await
}
pub async fn update_item(
tx: &mut PgConnection,
shop_id: Uuid,
id: Uuid,
body: &ItemInput,
) -> ApiResult<FlashSaleItem> {
sqlx::query_as::<_, FlashSaleItem>(
"UPDATE flash_sale_items i
SET sku_id = $3, sale_price_minor = $4, currency = $5,
reserved_stock = $6, per_customer_limit = $7, updated_at = now()
FROM flash_sale_sessions s
WHERE i.id = $1 AND i.session_id = s.id AND s.shop_id = $2
RETURNING i.id, i.session_id, i.sku_id, i.sale_price_minor, i.currency,
i.reserved_stock, i.sold_count, i.per_customer_limit, i.created_at, i.updated_at",
)
.bind(id)
.bind(shop_id)
.bind(body.sku_id)
.bind(body.sale_price_minor)
.bind(body.currency.to_uppercase())
.bind(body.reserved_stock)
.bind(body.per_customer_limit)
.fetch_optional(&mut *tx)
.await?
.ok_or_else(|| ApiError::NotFound("flash-sale item".into()))
}
pub async fn delete_item(tx: &mut PgConnection, shop_id: Uuid, id: Uuid) -> ApiResult<()> {
let result = sqlx::query(
"DELETE FROM flash_sale_items i
USING flash_sale_sessions s
WHERE i.id = $1 AND i.session_id = s.id AND s.shop_id = $2",
)
.bind(id)
.bind(shop_id)
.execute(&mut *tx)
.await?;
if result.rows_affected() == 0 {
return Err(ApiError::NotFound("flash-sale item".into()));
}
Ok(())
}
/// The SKU must be active, published, and owned by the session's shop.
pub async fn sku_is_own_active<'e, E: PgExecutor<'e>>(
exec: E,
shop_id: Uuid,
sku_id: Uuid,
) -> ApiResult<bool> {
Ok(sqlx::query_scalar(
"SELECT EXISTS(
SELECT 1 FROM skus s
JOIN products p ON p.id = s.product_id
WHERE s.id = $1 AND p.shop_id = $2
AND s.active = TRUE AND p.status = 'published'
)",
)
.bind(sku_id)
.bind(shop_id)
.fetch_one(exec)
.await?)
}
/// Another enabled session whose window overlaps and already lists this SKU.
pub async fn overlapping_sale_exists<'e, E: PgExecutor<'e>>(
exec: E,
sku_id: Uuid,
session_id: Uuid,
starts_at: chrono::DateTime<chrono::Utc>,
ends_at: chrono::DateTime<chrono::Utc>,
) -> ApiResult<bool> {
Ok(sqlx::query_scalar(
"SELECT EXISTS(
SELECT 1 FROM flash_sale_items i
JOIN flash_sale_sessions s ON s.id = i.session_id
WHERE i.sku_id = $1 AND s.id <> $2 AND s.enabled = TRUE
AND s.starts_at <= $4 AND s.ends_at >= $3
)",
)
.bind(sku_id)
.bind(session_id)
.bind(starts_at)
.bind(ends_at)
.fetch_one(exec)
.await?)
}
/// The group-buying capability lands after this one, so its overlap check is
/// dormant until its table exists.
pub async fn group_buying_table_exists<'e, E: PgExecutor<'e>>(exec: E) -> ApiResult<bool> {
Ok(sqlx::query_scalar("SELECT to_regclass('public.group_buying_activities') IS NOT NULL")
.fetch_one(exec)
.await?)
}
pub async fn overlapping_group_buying_exists(
tx: &mut PgConnection,
sku_id: Uuid,
starts_at: chrono::DateTime<chrono::Utc>,
ends_at: chrono::DateTime<chrono::Utc>,
) -> ApiResult<bool> {
Ok(sqlx::query_scalar(
"SELECT EXISTS(
SELECT 1 FROM group_buying_activities a
WHERE a.sku_id = $1 AND a.enabled = TRUE
AND a.starts_at <= $3 AND a.ends_at >= $2
)",
)
.bind(sku_id)
.bind(starts_at)
.bind(ends_at)
.fetch_one(&mut *tx)
.await?)
}
// ---- public discovery ----
pub async fn list_active_sessions<'e, E: PgExecutor<'e>>(
exec: E,
) -> ApiResult<Vec<FlashSaleSession>> {
Ok(sqlx::query_as::<_, FlashSaleSession>(&format!(
"SELECT {FLASH_SALE_SESSION_COLUMNS} FROM flash_sale_sessions
WHERE enabled = TRUE AND now() BETWEEN starts_at AND ends_at
ORDER BY starts_at"
))
.fetch_all(exec)
.await?)
}
pub async fn list_public_items<'e, E: PgExecutor<'e>>(
exec: E,
session_ids: &[Uuid],
) -> ApiResult<Vec<PublicItemView>> {
Ok(sqlx::query_as::<_, PublicItemView>(
"SELECT i.id, i.session_id, i.sku_id, s.product_id, p.slug AS product_slug,
p.name AS product_name, (p.images ->> 0) AS image, s.sku_code,
i.sale_price_minor, i.currency,
s.price_minor AS original_price_minor, s.currency AS original_currency,
i.reserved_stock, i.sold_count, i.per_customer_limit
FROM flash_sale_items i
JOIN skus s ON s.id = i.sku_id
JOIN products p ON p.id = s.product_id
WHERE i.session_id = ANY($1)
AND s.active = TRUE AND p.status = 'published'
ORDER BY i.created_at",
)
.bind(session_ids)
.fetch_all(exec)
.await?)
}
// ---- checkout ----
/// Activity item ids that could apply to these cart SKUs right now.
pub async fn candidate_item_ids<'e, E: PgExecutor<'e>>(
exec: E,
sku_ids: &[Uuid],
) -> ApiResult<Vec<Uuid>> {
Ok(sqlx::query_scalar(
"SELECT i.id FROM flash_sale_items i
JOIN flash_sale_sessions s ON s.id = i.session_id
WHERE i.sku_id = ANY($1) AND s.enabled = TRUE
AND now() BETWEEN s.starts_at AND s.ends_at",
)
.bind(sku_ids)
.fetch_all(exec)
.await?)
}
/// Lock candidates in primary-key order and re-prove they are still active.
pub async fn lock_active_items(
tx: &mut PgConnection,
ids: &[Uuid],
) -> ApiResult<Vec<FlashSaleItem>> {
Ok(sqlx::query_as::<_, FlashSaleItem>(&format!(
"SELECT {ITEM_COLS_Q} FROM flash_sale_items i
JOIN flash_sale_sessions s ON s.id = i.session_id
WHERE i.id = ANY($1) AND s.enabled = TRUE
AND now() BETWEEN s.starts_at AND s.ends_at
ORDER BY i.id
FOR UPDATE OF i"
))
.bind(ids)
.fetch_all(&mut *tx)
.await?)
}
/// Activity units this customer already holds on non-cancelled orders.
pub async fn purchased_qty(
tx: &mut PgConnection,
user_id: Uuid,
item_id: Uuid,
) -> ApiResult<i64> {
Ok(sqlx::query_scalar(
"SELECT COALESCE(SUM(oi.qty), 0)
FROM order_items oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.flash_sale_item_id = $1 AND o.user_id = $2
AND o.status <> 'cancelled'",
)
.bind(item_id)
.bind(user_id)
.fetch_one(&mut *tx)
.await?)
}
/// Guarded reserved-stock decrement; `sold_count` tracks sell-through.
pub async fn decrement_reserved(tx: &mut PgConnection, id: Uuid, qty: i32) -> ApiResult<()> {
let result = sqlx::query(
"UPDATE flash_sale_items
SET reserved_stock = reserved_stock - $2, sold_count = sold_count + $2,
updated_at = now()
WHERE id = $1 AND reserved_stock >= $2",
)
.bind(id)
.bind(qty)
.execute(&mut *tx)
.await?;
if result.rows_affected() == 0 {
return Err(ApiError::Conflict("insufficient flash-sale stock".into()));
}
Ok(())
}
/// Pending-payment cancellation returns reserved activity stock.
pub async fn restore_for_order(tx: &mut PgConnection, order_id: Uuid) -> ApiResult<()> {
sqlx::query(
"UPDATE flash_sale_items i
SET reserved_stock = i.reserved_stock + oi.qty,
sold_count = GREATEST(i.sold_count - oi.qty, 0),
updated_at = now()
FROM order_items oi
WHERE oi.order_id = $1 AND oi.flash_sale_item_id = i.id",
)
.bind(order_id)
.execute(&mut *tx)
.await?;
Ok(())
}
+283
View File
@@ -0,0 +1,283 @@
use std::collections::HashMap;
use serde_json::Value;
use sqlx::PgConnection;
use uuid::Uuid;
use crate::error::{unique_conflict, ApiError, ApiResult};
use crate::models::{Currency, FlashSaleItem, FlashSaleSession};
use crate::money::convert_minor;
use crate::state::AppState;
use super::dto::{
ItemInput, LineActivity, PublicSessionView, SessionInput, ShopItemView, ShopSessionView,
};
use super::repo;
// ---- shop management ----
pub async fn list_for_shop(state: &AppState, shop_id: Uuid) -> ApiResult<Vec<ShopSessionView>> {
let sessions = repo::list_sessions_for_shop(&state.db, shop_id).await?;
let items = repo::list_items_for_shop(&state.db, shop_id).await?;
let mut by_session: HashMap<Uuid, Vec<ShopItemView>> = HashMap::new();
for item in items {
by_session.entry(item.session_id).or_default().push(item);
}
Ok(sessions
.into_iter()
.map(|session| ShopSessionView {
items: by_session.remove(&session.id).unwrap_or_default(),
session,
})
.collect())
}
pub async fn create_session(
state: &AppState,
shop_id: Uuid,
body: SessionInput,
) -> ApiResult<FlashSaleSession> {
validate_session(&body)?;
let mut tx = state.db.begin().await?;
let session = repo::insert_session(&mut tx, shop_id, &body).await?;
tx.commit().await?;
Ok(session)
}
pub async fn update_session(
state: &AppState,
shop_id: Uuid,
id: Uuid,
body: SessionInput,
) -> ApiResult<FlashSaleSession> {
validate_session(&body)?;
let mut tx = state.db.begin().await?;
let session = repo::update_session(&mut tx, shop_id, id, &body).await?;
tx.commit().await?;
Ok(session)
}
pub async fn delete_session(state: &AppState, shop_id: Uuid, id: Uuid) -> ApiResult<()> {
let mut tx = state.db.begin().await?;
repo::delete_session(&mut tx, shop_id, id).await?;
tx.commit().await?;
Ok(())
}
pub async fn add_item(
state: &AppState,
shop_id: Uuid,
session_id: Uuid,
body: ItemInput,
) -> ApiResult<FlashSaleItem> {
validate_item(&body)?;
let mut tx = state.db.begin().await?;
ensure_currency(&mut tx, &body.currency).await?;
let session = repo::get_session_own(&mut *tx, shop_id, session_id).await?;
ensure_activity_slot(&mut tx, shop_id, &session, body.sku_id).await?;
let item = repo::insert_item(&mut tx, session_id, &body)
.await
.map_err(|e| unique_conflict(e, "SKU is already in this session"))?;
tx.commit().await?;
Ok(item)
}
pub async fn update_item(
state: &AppState,
shop_id: Uuid,
id: Uuid,
body: ItemInput,
) -> ApiResult<FlashSaleItem> {
validate_item(&body)?;
let mut tx = state.db.begin().await?;
ensure_currency(&mut tx, &body.currency).await?;
let existing = repo::get_item_own(&mut *tx, shop_id, id).await?;
let session = repo::get_session_own(&mut *tx, shop_id, existing.session_id).await?;
ensure_activity_slot(&mut tx, shop_id, &session, body.sku_id).await?;
let item = repo::update_item(&mut tx, shop_id, id, &body).await?;
tx.commit().await?;
Ok(item)
}
pub async fn delete_item(state: &AppState, shop_id: Uuid, id: Uuid) -> ApiResult<()> {
let mut tx = state.db.begin().await?;
repo::delete_item(&mut tx, shop_id, id).await?;
tx.commit().await?;
Ok(())
}
// ---- public discovery ----
pub async fn public_active(state: &AppState) -> ApiResult<Vec<PublicSessionView>> {
let sessions = repo::list_active_sessions(&state.db).await?;
let ids: Vec<Uuid> = sessions.iter().map(|s| s.id).collect();
let items = if ids.is_empty() {
Vec::new()
} else {
repo::list_public_items(&state.db, &ids).await?
};
let mut by_session: HashMap<Uuid, Vec<_>> = HashMap::new();
for item in items {
by_session.entry(item.session_id).or_default().push(item);
}
Ok(sessions
.into_iter()
.map(|session| PublicSessionView {
items: by_session.remove(&session.id).unwrap_or_default(),
session,
})
.collect())
}
// ---- checkout composition ----
/// Resolve the activity-priced part of each cart line server-side. Candidates
/// are locked in primary-key order, so concurrent checkouts cannot oversell the
/// reserved stock; the customer's remaining allowance caps the activity units.
pub async fn resolve_activities(
tx: &mut PgConnection,
user_id: Uuid,
lines: &[(Uuid, i32)],
target: &Currency,
currencies: &[Currency],
) -> ApiResult<HashMap<Uuid, LineActivity>> {
let sku_ids: Vec<Uuid> = lines.iter().map(|(sku_id, _)| *sku_id).collect();
let candidate_ids = repo::candidate_item_ids(&mut *tx, &sku_ids).await?;
if candidate_ids.is_empty() {
return Ok(HashMap::new());
}
let locked = repo::lock_active_items(tx, &candidate_ids).await?;
let mut by_sku: HashMap<Uuid, FlashSaleItem> = HashMap::new();
for item in locked {
by_sku.entry(item.sku_id).or_insert(item);
}
let mut resolved = HashMap::new();
for (sku_id, qty) in lines {
let Some(item) = by_sku.get(sku_id) else {
continue;
};
let bought = repo::purchased_qty(&mut *tx, user_id, item.id).await?;
let allowance = item.per_customer_limit as i64 - bought;
let activity_qty = (*qty as i64).min(allowance).min(item.reserved_stock as i64);
if activity_qty <= 0 {
continue;
}
let from = currencies
.iter()
.find(|c| c.code == item.currency)
.ok_or_else(|| {
ApiError::BadRequest(format!("currency {} is disabled", item.currency))
})?;
let unit_price_minor = convert_minor(item.sale_price_minor, from, target)?;
resolved.insert(
*sku_id,
LineActivity {
item_id: item.id,
unit_price_minor,
qty: activity_qty as i32,
},
);
}
Ok(resolved)
}
/// Guarded reserved-stock decrement for the activity units of a line.
pub async fn consume(tx: &mut PgConnection, item_id: Uuid, qty: i32) -> ApiResult<()> {
repo::decrement_reserved(tx, item_id, qty).await
}
/// Pending-payment cancellation returns reserved activity stock.
pub async fn restore_for_order(tx: &mut PgConnection, order_id: Uuid) -> ApiResult<()> {
repo::restore_for_order(tx, order_id).await
}
// ---- validation ----
async fn ensure_activity_slot(
tx: &mut PgConnection,
shop_id: Uuid,
session: &FlashSaleSession,
sku_id: Uuid,
) -> ApiResult<()> {
if !repo::sku_is_own_active(&mut *tx, shop_id, sku_id).await? {
return Err(ApiError::BadRequest(
"SKU is not an active published product of this shop".into(),
));
}
if repo::overlapping_sale_exists(
&mut *tx,
sku_id,
session.id,
session.starts_at,
session.ends_at,
)
.await?
{
return Err(ApiError::Conflict(
"SKU already has an overlapping flash sale".into(),
));
}
if repo::group_buying_table_exists(&mut *tx).await?
&& repo::overlapping_group_buying_exists(tx, sku_id, session.starts_at, session.ends_at)
.await?
{
return Err(ApiError::Conflict(
"SKU already has an overlapping group-buying activity".into(),
));
}
Ok(())
}
async fn ensure_currency(tx: &mut PgConnection, code: &str) -> ApiResult<()> {
if !repo::currency_enabled(&mut *tx, &code.to_uppercase()).await? {
return Err(ApiError::BadRequest(
"currency is unknown or disabled".into(),
));
}
Ok(())
}
fn validate_session(body: &SessionInput) -> ApiResult<()> {
bilingual(&body.label, "label")?;
if body.ends_at < body.starts_at {
return Err(ApiError::BadRequest(
"ends_at must not precede starts_at".into(),
));
}
Ok(())
}
fn validate_item(body: &ItemInput) -> ApiResult<()> {
if body.sale_price_minor <= 0 {
return Err(ApiError::BadRequest(
"sale_price_minor must be positive".into(),
));
}
if body.reserved_stock < 0 {
return Err(ApiError::BadRequest(
"reserved_stock must not be negative".into(),
));
}
if body.per_customer_limit <= 0 {
return Err(ApiError::BadRequest(
"per_customer_limit must be positive".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(())
}
+2
View File
@@ -6,6 +6,7 @@ pub mod catalog;
pub mod content;
pub mod coupon;
pub mod currency;
pub mod flash_sale;
pub mod fulfillment;
pub mod health;
pub mod identity;
@@ -28,6 +29,7 @@ pub fn api_router() -> Router<AppState> {
.merge(content::router())
.merge(cart::router())
.merge(coupon::router())
.merge(flash_sale::router())
.merge(order::router())
.merge(points::router())
.merge(shop::router())
+6 -3
View File
@@ -11,7 +11,7 @@ use super::dto::{OrderScope, OrderView};
const ORDER_COLS: &str = "id, order_no, shop_id, user_id, status, currency, total_minor,
discount_minor, coupon_id, shipping_address, created_at, updated_at";
const ORDER_ITEM_COLS: &str =
"id, order_id, sku_id, product_name, sku_code, image, unit_price_minor, qty";
"id, order_id, sku_id, product_name, sku_code, image, unit_price_minor, qty, flash_sale_item_id";
pub async fn attach_items(db: &PgPool, orders: Vec<Order>) -> ApiResult<Vec<OrderView>> {
let ids: Vec<Uuid> = orders.iter().map(|o| o.id).collect();
@@ -162,6 +162,7 @@ pub async fn insert_order(
.await?)
}
#[allow(clippy::too_many_arguments)]
pub async fn insert_item(
tx: &mut PgConnection,
order_id: Uuid,
@@ -171,10 +172,11 @@ pub async fn insert_item(
image: &Option<String>,
unit: i64,
qty: i32,
flash_sale_item_id: Option<Uuid>,
) -> ApiResult<()> {
sqlx::query(
"INSERT INTO order_items (order_id, sku_id, product_name, sku_code, image, unit_price_minor, qty)
VALUES ($1, $2, $3, $4, $5, $6, $7)",
"INSERT INTO order_items (order_id, sku_id, product_name, sku_code, image, unit_price_minor, qty, flash_sale_item_id)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)",
)
.bind(order_id)
.bind(sku_id)
@@ -183,6 +185,7 @@ pub async fn insert_item(
.bind(image)
.bind(unit)
.bind(qty)
.bind(flash_sale_item_id)
.execute(&mut *tx)
.await?;
Ok(())
+126 -27
View File
@@ -6,12 +6,34 @@ use crate::error::{ApiError, ApiResult};
use crate::http::{clamp_page, clamp_per_page, Paged};
use crate::models::OrderStatus;
use crate::money::convert_minor;
use crate::modules::{cart, coupon};
use crate::modules::{cart, coupon, flash_sale};
use crate::state::AppState;
use super::dto::{AddressBody, OrderScope, OrderView};
use super::repo;
/// One cart line after activity resolution: the standard-priced part and, when
/// the customer is eligible, the flash-sale part. A line can therefore produce
/// two order items, which keeps each unit-price snapshot honest.
struct CheckoutLine {
shop_id: Uuid,
sku_id: Uuid,
product_name: serde_json::Value,
sku_code: String,
image: Option<String>,
normal_unit: i64,
normal_qty: i32,
activity_item: Option<Uuid>,
activity_unit: i64,
activity_qty: i32,
}
impl CheckoutLine {
fn line_total(&self) -> i64 {
self.normal_unit * self.normal_qty as i64 + self.activity_unit * self.activity_qty as i64
}
}
pub async fn list(
state: &AppState,
scope: OrderScope,
@@ -101,18 +123,76 @@ pub async fn checkout(
}
}
// Per-shop merchandise subtotal in the buyer's currency.
let mut subtotal_by_shop: HashMap<Uuid, i64> = HashMap::new();
let mut line_prices: Vec<(Uuid, i64, i32)> = Vec::new();
// Normal unit prices in the buyer's currency, plus the quantities the
// activity resolver needs.
let mut normal_by_sku: HashMap<Uuid, i64> = HashMap::new();
let mut price_qty: Vec<(Uuid, i32)> = Vec::new();
for row in &rows {
let qty = qty_by_sku.get(&row.sku_id).copied().unwrap_or(0);
let from = all_currencies
.iter()
.find(|c| c.code == row.currency)
.ok_or_else(|| ApiError::BadRequest(format!("currency {} disabled", row.currency)))?;
let unit = convert_minor(row.price_minor, from, &target)?;
*subtotal_by_shop.entry(row.shop_id).or_insert(0) += unit * qty as i64;
line_prices.push((row.sku_id, unit, qty));
normal_by_sku.insert(row.sku_id, convert_minor(row.price_minor, from, &target)?);
price_qty.push((row.sku_id, qty));
}
// Flash-sale eligibility is resolved and its reserved stock locked before
// the coupon locks, so every checkout takes locks in the same order.
let activities = flash_sale::service::resolve_activities(
&mut tx,
user_id,
&price_qty,
&target,
&all_currencies,
)
.await?;
let mut lines: Vec<CheckoutLine> = Vec::with_capacity(rows.len());
for row in &rows {
let qty = qty_by_sku.get(&row.sku_id).copied().unwrap_or(0);
let normal_unit = normal_by_sku.get(&row.sku_id).copied().unwrap_or(0);
let (activity_item, activity_unit, activity_qty) = match activities.get(&row.sku_id) {
Some(activity) => {
let eligible = activity.qty.min(qty);
if eligible > 0 {
(Some(activity.item_id), activity.unit_price_minor, eligible)
} else {
(None, normal_unit, 0)
}
}
None => (None, normal_unit, 0),
};
lines.push(CheckoutLine {
shop_id: row.shop_id,
sku_id: row.sku_id,
product_name: row.product_name.clone(),
sku_code: row.sku_code.clone(),
image: row.image.clone(),
normal_unit,
normal_qty: qty - activity_qty,
activity_item,
activity_unit,
activity_qty,
});
}
// Activity pricing and coupons are exclusive per shop order.
for shop_id in coupon_by_shop.keys() {
if lines
.iter()
.any(|line| line.shop_id == *shop_id && line.activity_qty > 0)
{
return Err(ApiError::Conflict(
"a coupon cannot be combined with flash-sale pricing on one shop order".into(),
));
}
}
// Per-shop merchandise subtotal in the buyer's currency.
let mut subtotal_by_shop: HashMap<Uuid, i64> = HashMap::new();
for line in &lines {
*subtotal_by_shop.entry(line.shop_id).or_insert(0) += line.line_total();
}
// Lock every selected coupon in id order, then prove the customer owns it.
@@ -129,8 +209,8 @@ pub async fn checkout(
let address = serde_json::to_value(&shipping_address).map_err(ApiError::internal)?;
let mut created = Vec::new();
for shop_id in shop_order {
let shop_rows: Vec<&repo::CheckoutRow> =
rows.iter().filter(|r| r.shop_id == shop_id).collect();
let shop_lines: Vec<&CheckoutLine> =
lines.iter().filter(|line| line.shop_id == shop_id).collect();
let subtotal = subtotal_by_shop.get(&shop_id).copied().unwrap_or(0);
let selected_coupon = coupon_by_shop.get(&shop_id).copied();
// Eligibility and the discount are resolved server-side; the client
@@ -165,23 +245,40 @@ pub async fn checkout(
if let Some(id) = selected_coupon {
coupon::service::redeem(&mut tx, id, order.id).await?;
}
for row in &shop_rows {
let (_, unit, qty) = line_prices
.iter()
.find(|(sku, _, _)| *sku == row.sku_id)
.unwrap();
repo::insert_item(
&mut tx,
order.id,
row.sku_id,
&row.product_name,
&row.sku_code,
&row.image,
*unit,
*qty,
)
.await?;
repo::decrement_stock(&mut tx, row.sku_id, *qty).await?;
for line in shop_lines {
// Activity-priced units first, then the standard-priced remainder.
if let (Some(item_id), true) = (line.activity_item, line.activity_qty > 0) {
repo::insert_item(
&mut tx,
order.id,
line.sku_id,
&line.product_name,
&line.sku_code,
&line.image,
line.activity_unit,
line.activity_qty,
Some(item_id),
)
.await?;
flash_sale::service::consume(&mut tx, item_id, line.activity_qty).await?;
}
if line.normal_qty > 0 {
repo::insert_item(
&mut tx,
order.id,
line.sku_id,
&line.product_name,
&line.sku_code,
&line.image,
line.normal_unit,
line.normal_qty,
None,
)
.await?;
}
// The whole ordered quantity leaves SKU stock; only the activity
// portion also leaves reserved activity stock.
repo::decrement_stock(&mut tx, line.sku_id, line.normal_qty + line.activity_qty).await?;
}
created.push(order);
}
@@ -201,7 +298,9 @@ pub async fn cancel(state: &AppState, user_id: Uuid, id: Uuid) -> ApiResult<Orde
let mut tx = state.db.begin().await?;
let order = repo::cancel(&mut tx, user_id, id).await?;
repo::restore_stock(&mut tx, order.id).await?;
// Stock and the redeemed coupon come back in the same transaction.
// Reserved activity stock, SKU stock, and the redeemed coupon all come back
// in the same transaction.
flash_sale::service::restore_for_order(&mut tx, order.id).await?;
coupon::service::restore_for_order(&mut tx, order.id).await?;
tx.commit().await?;
let mut views = repo::attach_items(&state.db, vec![order]).await?;