361 lines
12 KiB
Rust
361 lines
12 KiB
Rust
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();
|
|
}
|
|
}
|