feat(api): group buying with payment-time seat claims

Shop-owned activities on one SKU, concrete groups with an open/successful/expired/cancelled lifecycle, and one paid membership row per paid order. Checkout accepts a single-SKU quantity-1 intent, snapshots the group price and identity on the pending order, and opens or references a group; a paid seat is claimed only at payment, which locks the group and fills it exactly at capacity. Pending-payment cancellation restores SKU stock only and never rolls back paid seats; an unpaid opener cancelling closes a still-empty group. Coupons are refused on a group shop order, and the flash-sale exclusion is now enforced in both directions because the activity table exists, which activates the guard add-flash-sales shipped dormant. The activity column names follow the contract recorded in this change's design. Surfaces (shop-admin, mall) and seeding follow.
2026-09-18 13:16:28 +00:00
parent 705cbe249a
commit a7bc476251
15 changed files with 1563 additions and 48 deletions
@@ -0,0 +1,363 @@
+use sqlx::{PgConnection, PgExecutor};
+use uuid::Uuid;
+
+use crate::error::{ApiError, ApiResult};
+use crate::models::{
+    CollectiveGroup, GroupBuyingActivity, COLLECTIVE_GROUP_COLUMNS,
+};
+
+use super::dto::{ActivityInput, ActivityView, OpenGroupView};
+
+/// Activity joined with the SKU and catalog price it discounts.
+const ACTIVITY_VIEW_SELECT: &str = "SELECT a.id, a.shop_id, a.sku_id, a.name, a.description,
+            COALESCE(a.image, (p.images ->> 0)) AS image,
+            a.group_price_minor, a.currency, a.required_members, a.starts_at, a.ends_at,
+            a.group_lifetime_hours, a.enabled, a.created_at, a.updated_at,
+            s.sku_code, s.product_id, p.slug AS product_slug, p.name AS product_name,
+            s.price_minor AS original_price_minor, s.currency AS original_currency
+       FROM group_buying_activities a
+       JOIN skus s ON s.id = a.sku_id
+       JOIN products p ON p.id = s.product_id";
+
+// ---- shop management ----
+
+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 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?)
+}
+
+pub async fn list_views_for_shop<'e, E: PgExecutor<'e>>(
+    exec: E,
+    shop_id: Uuid,
+) -> ApiResult<Vec<ActivityView>> {
+    Ok(sqlx::query_as::<_, ActivityView>(&format!(
+        "{ACTIVITY_VIEW_SELECT} WHERE a.shop_id = $1 ORDER BY a.created_at DESC"
+    ))
+    .bind(shop_id)
+    .fetch_all(exec)
+    .await?)
+}
+
+pub async fn list_active_views<'e, E: PgExecutor<'e>>(exec: E) -> ApiResult<Vec<ActivityView>> {
+    Ok(sqlx::query_as::<_, ActivityView>(&format!(
+        "{ACTIVITY_VIEW_SELECT}
+          WHERE a.enabled = TRUE AND now() BETWEEN a.starts_at AND a.ends_at
+          ORDER BY a.ends_at"
+    ))
+    .fetch_all(exec)
+    .await?)
+}
+
+pub async fn get_own<'e, E: PgExecutor<'e>>(
+    exec: E,
+    shop_id: Uuid,
+    id: Uuid,
+) -> ApiResult<GroupBuyingActivity> {
+    sqlx::query_as::<_, GroupBuyingActivity>(
+        "SELECT id, shop_id, sku_id, name, description, image, group_price_minor, currency,
+                required_members, starts_at, ends_at, group_lifetime_hours, enabled,
+                created_at, updated_at
+           FROM group_buying_activities WHERE id = $1 AND shop_id = $2",
+    )
+    .bind(id)
+    .bind(shop_id)
+    .fetch_optional(exec)
+    .await?
+    .ok_or_else(|| ApiError::NotFound("group-buying activity".into()))
+}
+
+pub async fn get_by_id<'e, E: PgExecutor<'e>>(
+    exec: E,
+    id: Uuid,
+) -> ApiResult<GroupBuyingActivity> {
+    sqlx::query_as::<_, GroupBuyingActivity>(
+        "SELECT id, shop_id, sku_id, name, description, image, group_price_minor, currency,
+                required_members, starts_at, ends_at, group_lifetime_hours, enabled,
+                created_at, updated_at
+           FROM group_buying_activities WHERE id = $1",
+    )
+    .bind(id)
+    .fetch_optional(exec)
+    .await?
+    .ok_or_else(|| ApiError::NotFound("group-buying activity".into()))
+}
+
+pub async fn insert(
+    tx: &mut PgConnection,
+    shop_id: Uuid,
+    body: &ActivityInput,
+) -> ApiResult<GroupBuyingActivity> {
+    Ok(sqlx::query_as::<_, GroupBuyingActivity>(
+        "INSERT INTO group_buying_activities
+             (shop_id, sku_id, name, description, image, group_price_minor, currency,
+              required_members, starts_at, ends_at, group_lifetime_hours, enabled)
+         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
+         RETURNING id, shop_id, sku_id, name, description, image, group_price_minor, currency,
+                   required_members, starts_at, ends_at, group_lifetime_hours, enabled,
+                   created_at, updated_at",
+    )
+    .bind(shop_id)
+    .bind(body.sku_id)
+    .bind(&body.name)
+    .bind(&body.description)
+    .bind(&body.image)
+    .bind(body.group_price_minor)
+    .bind(body.currency.to_uppercase())
+    .bind(body.required_members)
+    .bind(body.starts_at)
+    .bind(body.ends_at)
+    .bind(body.group_lifetime_hours)
+    .bind(body.enabled)
+    .fetch_one(&mut *tx)
+    .await?)
+}
+
+pub async fn update(
+    tx: &mut PgConnection,
+    shop_id: Uuid,
+    id: Uuid,
+    body: &ActivityInput,
+) -> ApiResult<GroupBuyingActivity> {
+    sqlx::query_as::<_, GroupBuyingActivity>(
+        "UPDATE group_buying_activities
+            SET sku_id = $3, name = $4, description = $5, image = $6, group_price_minor = $7,
+                currency = $8, required_members = $9, starts_at = $10, ends_at = $11,
+                group_lifetime_hours = $12, enabled = $13, updated_at = now()
+          WHERE id = $1 AND shop_id = $2
+         RETURNING id, shop_id, sku_id, name, description, image, group_price_minor, currency,
+                   required_members, starts_at, ends_at, group_lifetime_hours, enabled,
+                   created_at, updated_at",
+    )
+    .bind(id)
+    .bind(shop_id)
+    .bind(body.sku_id)
+    .bind(&body.name)
+    .bind(&body.description)
+    .bind(&body.image)
+    .bind(body.group_price_minor)
+    .bind(body.currency.to_uppercase())
+    .bind(body.required_members)
+    .bind(body.starts_at)
+    .bind(body.ends_at)
+    .bind(body.group_lifetime_hours)
+    .bind(body.enabled)
+    .fetch_optional(&mut *tx)
+    .await?
+    .ok_or_else(|| ApiError::NotFound("group-buying activity".into()))
+}
+
+pub async fn delete(tx: &mut PgConnection, shop_id: Uuid, id: Uuid) -> ApiResult<()> {
+    let result =
+        sqlx::query("DELETE FROM group_buying_activities 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("group-buying activity".into()));
+    }
+    Ok(())
+}
+
+/// An enabled flash-sale item whose window overlaps this activity.
+pub async fn overlapping_flash_exists<'e, E: PgExecutor<'e>>(
+    exec: E,
+    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 flash_sale_items i
+             JOIN flash_sale_sessions s ON s.id = i.session_id
+             WHERE i.sku_id = $1 AND s.enabled = TRUE
+               AND s.starts_at <= $3 AND s.ends_at >= $2
+         )",
+    )
+    .bind(sku_id)
+    .bind(starts_at)
+    .bind(ends_at)
+    .fetch_one(exec)
+    .await?)
+}
+
+// ---- groups ----
+
+/// Record expiry lazily; no scheduler is involved.
+pub async fn expire_due_groups<'e, E: PgExecutor<'e>>(exec: E) -> ApiResult<u64> {
+    let result = sqlx::query(
+        "UPDATE collective_groups SET status = 'expired', updated_at = now()
+          WHERE status = 'open' AND expires_at < now()",
+    )
+    .execute(exec)
+    .await?;
+    Ok(result.rows_affected())
+}
+
+pub async fn open_groups_for_activities(
+    tx: &mut PgConnection,
+    activity_ids: &[Uuid],
+) -> ApiResult<Vec<CollectiveGroup>> {
+    Ok(sqlx::query_as::<_, CollectiveGroup>(&format!(
+        "SELECT {COLLECTIVE_GROUP_COLUMNS} FROM collective_groups
+          WHERE activity_id = ANY($1) AND status = 'open' AND expires_at >= now()
+          ORDER BY expires_at"
+    ))
+    .bind(activity_ids)
+    .fetch_all(&mut *tx)
+    .await?)
+}
+
+/// Joining only reads the group: a seat is claimed at payment, not checkout.
+pub async fn get_group<'e, E: PgExecutor<'e>>(
+    exec: E,
+    id: Uuid,
+) -> ApiResult<CollectiveGroup> {
+    sqlx::query_as::<_, CollectiveGroup>(&format!(
+        "SELECT {COLLECTIVE_GROUP_COLUMNS} FROM collective_groups WHERE id = $1"
+    ))
+    .bind(id)
+    .fetch_optional(exec)
+    .await?
+    .ok_or_else(|| ApiError::NotFound("group".into()))
+}
+
+pub async fn insert_group(
+    tx: &mut PgConnection,
+    activity_id: Uuid,
+    expires_at: chrono::DateTime<chrono::Utc>,
+) -> ApiResult<CollectiveGroup> {
+    Ok(sqlx::query_as::<_, CollectiveGroup>(&format!(
+        "INSERT INTO collective_groups (activity_id, expires_at) VALUES ($1, $2)
+         RETURNING {COLLECTIVE_GROUP_COLUMNS}"
+    ))
+    .bind(activity_id)
+    .bind(expires_at)
+    .fetch_one(&mut *tx)
+    .await?)
+}
+
+/// Record which pending order opened the group.
+pub async fn set_leader(tx: &mut PgConnection, group_id: Uuid, order_id: Uuid) -> ApiResult<()> {
+    sqlx::query("UPDATE collective_groups SET leader_order_id = $2, updated_at = now() WHERE id = $1")
+        .bind(group_id)
+        .bind(order_id)
+        .execute(&mut *tx)
+        .await?;
+    Ok(())
+}
+
+/// Lock the group row so two payments cannot claim the same final seat.
+pub async fn lock_group(tx: &mut PgConnection, id: Uuid) -> ApiResult<CollectiveGroup> {
+    sqlx::query_as::<_, CollectiveGroup>(&format!(
+        "SELECT {COLLECTIVE_GROUP_COLUMNS} FROM collective_groups WHERE id = $1 FOR UPDATE"
+    ))
+    .bind(id)
+    .fetch_optional(&mut *tx)
+    .await?
+    .ok_or_else(|| ApiError::NotFound("group".into()))
+}
+
+pub async fn insert_member(
+    tx: &mut PgConnection,
+    group_id: Uuid,
+    order_id: Uuid,
+    user_id: Uuid,
+) -> Result<(), sqlx::Error> {
+    sqlx::query(
+        "INSERT INTO collective_group_members (group_id, order_id, user_id)
+         VALUES ($1, $2, $3)",
+    )
+    .bind(group_id)
+    .bind(order_id)
+    .bind(user_id)
+    .execute(&mut *tx)
+    .await?;
+    Ok(())
+}
+
+/// One paid seat: increment and become successful exactly at capacity.
+pub async fn claim_seat(tx: &mut PgConnection, group_id: Uuid, required: i32) -> ApiResult<CollectiveGroup> {
+    Ok(sqlx::query_as::<_, CollectiveGroup>(&format!(
+        "UPDATE collective_groups
+            SET paid_member_count = paid_member_count + 1,
+                status = CASE WHEN paid_member_count + 1 >= $2
+                              THEN 'successful'::collective_group_status
+                              ELSE status END,
+                updated_at = now()
+          WHERE id = $1
+         RETURNING {COLLECTIVE_GROUP_COLUMNS}"
+    ))
+    .bind(group_id)
+    .bind(required)
+    .fetch_one(&mut *tx)
+    .await?)
+}
+
+/// An unpaid opener cancelling closes a still-empty group.
+pub async fn cancel_empty_opener_group(
+    tx: &mut PgConnection,
+    group_id: Uuid,
+    order_id: Uuid,
+) -> ApiResult<u64> {
+    let result = sqlx::query(
+        "UPDATE collective_groups SET status = 'cancelled', updated_at = now()
+          WHERE id = $1 AND leader_order_id = $2 AND paid_member_count = 0
+            AND status = 'open'",
+    )
+    .bind(group_id)
+    .bind(order_id)
+    .execute(&mut *tx)
+    .await?;
+    Ok(result.rows_affected())
+}
+
+/// Attach open groups (with the activity's required count) to each view.
+pub fn attach_open_groups(
+    views: &mut [ActivityView],
+    groups: Vec<CollectiveGroup>,
+    required_by_activity: &std::collections::HashMap<Uuid, i32>,
+) {
+    let mut by_activity: std::collections::HashMap<Uuid, Vec<OpenGroupView>> =
+        std::collections::HashMap::new();
+    for group in groups {
+        by_activity
+            .entry(group.activity_id)
+            .or_default()
+            .push(OpenGroupView {
+                id: group.id,
+                paid_member_count: group.paid_member_count,
+                required_members: required_by_activity
+                    .get(&group.activity_id)
+                    .copied()
+                    .unwrap_or(0),
+                expires_at: group.expires_at,
+            });
+    }
+    for view in views.iter_mut() {
+        view.open_groups = by_activity.remove(&view.id).unwrap_or_default();
+    }
+}