feat: 完善移动端会员业务流程

This commit is contained in:
richjohnrambo
2026-10-13 14:19:55 +08:00
parent 32c6238751
commit b2ebea4c80
59 changed files with 1516 additions and 470 deletions
+273 -53
View File
@@ -5,6 +5,7 @@ import (
"database/sql"
"encoding/json"
"errors"
"fmt"
"time"
"github.com/jmoiron/sqlx"
@@ -104,12 +105,15 @@ type TaskStats struct {
}
type TaskSubmissionInput struct {
TenantID int64
MemberID int64
OrderID int64
SubmissionNo string
EvidenceText string
AttachmentIDs []int64
TenantID int64
MemberID int64
OrderID int64
SubmissionNo string
EvidenceText string
AttachmentIDs []int64
LedgerNo string
IdempotencyKey string
SettledAt time.Time
}
type TaskSubmissionResult struct {
@@ -159,9 +163,10 @@ func (s *Store) ClaimTask(ctx context.Context, input ClaimTaskInput) (*TaskOrder
defer tx.Rollback()
var member struct {
LevelID int64 `db:"level_id"`
LevelID int64 `db:"level_id"`
AutoDispatch int `db:"auto_dispatch"`
}
err = tx.GetContext(ctx, &member, `SELECT level_id FROM hg_app_member
err = tx.GetContext(ctx, &member, `SELECT level_id,auto_dispatch FROM hg_app_member
WHERE tenant_id=? AND id=? AND member_status=1 AND deleted_at IS NULL LIMIT 1 FOR UPDATE`, input.TenantID, input.MemberID)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrAccountNotFound
@@ -227,34 +232,16 @@ func (s *Store) ClaimTask(ctx context.Context, input ClaimTaskInput) (*TaskOrder
return nil, ErrDailyLimitReached
}
}
var task struct {
ID int64 `db:"id"`
Title string `db:"title"`
CoverURL string `db:"cover_url"`
CommissionAmount int64 `db:"commission_amount"`
CommissionRate int `db:"commission_rate"`
PerMemberLimit int `db:"per_member_limit"`
}
err = tx.GetContext(ctx, &task, `SELECT t.id,t.title,t.cover_url,t.commission_amount,t.commission_rate,t.per_member_limit
FROM hg_task t JOIN hg_task_level tl ON tl.tenant_id=t.tenant_id AND tl.task_id=t.id AND tl.level_id=?
WHERE t.tenant_id=? AND t.hall_id=? AND t.publish_status=3 AND t.deleted_at IS NULL
AND (t.start_at IS NULL OR t.start_at<=UTC_TIMESTAMP(3)) AND (t.end_at IS NULL OR t.end_at>UTC_TIMESTAMP(3))
AND (t.daily_limit=0 OR (SELECT COUNT(*) FROM hg_trade_order od WHERE od.tenant_id=t.tenant_id AND od.task_id=t.id
AND od.order_type=2 AND od.created_at>=UTC_DATE() AND od.created_at<UTC_DATE()+INTERVAL 1 DAY)<t.daily_limit)
AND (t.per_member_limit=0 OR (SELECT COUNT(*) FROM hg_trade_order om WHERE om.tenant_id=t.tenant_id
AND om.task_id=t.id AND om.member_id=? AND om.order_type=2 AND om.created_at>=UTC_DATE() AND om.created_at<UTC_DATE()+INTERVAL 1 DAY)<t.per_member_limit)
ORDER BY t.sort,t.id LIMIT 1 FOR UPDATE`, member.LevelID, input.TenantID, input.HallID, input.MemberID)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrNoTaskAvailable
}
if err != nil {
return nil, err
dispatchMode, message := "manual", "等待后台匹配订单"
if member.AutoDispatch == 1 {
dispatchMode, message = "auto", "正在自动匹配订单"
}
rules, err := json.Marshal(map[string]any{
"commission_amount": task.CommissionAmount,
"commission_rate": task.CommissionRate,
"deadline_at": input.DeadlineAt.UTC().Format(time.RFC3339),
"member_level_id": member.LevelID,
"source": "member_auto_claim",
"dispatch_mode": dispatchMode,
"hall_id": input.HallID,
"deadline_at": input.DeadlineAt.UTC().Format(time.RFC3339),
"member_level_id": member.LevelID,
})
if err != nil {
return nil, err
@@ -262,10 +249,9 @@ func (s *Store) ClaimTask(ctx context.Context, input ClaimTaskInput) (*TaskOrder
result, err := tx.ExecContext(ctx, `INSERT INTO hg_trade_order
(tenant_id,order_no,member_id,task_id,product_id,product_name,order_type,quantity,
unit_price_amount,total_amount,commission_amount,currency,member_level_id,rule_snapshot,
message,order_status,freeze_status,matched_at,created_by,updated_by)
VALUES(?,?,?,?,0,?,2,1,0,0,?,'CNY',?,?,?,2,1,UTC_TIMESTAMP(3),0,0)`,
input.TenantID, input.OrderNo, input.MemberID, task.ID, task.Title, task.CommissionAmount,
member.LevelID, rules, "任务匹配成功,请在截止时间前完成")
message,order_status,freeze_status,created_by,updated_by)
VALUES(?,?,?,0,0,'',2,0,0,0,0,'CNY',?,?,?,1,1,0,0)`,
input.TenantID, input.OrderNo, input.MemberID, member.LevelID, rules, message)
if err != nil {
if isDuplicate(err) {
return nil, ErrPendingOrderExists
@@ -279,11 +265,148 @@ func (s *Store) ClaimTask(ctx context.Context, input ClaimTaskInput) (*TaskOrder
if err := tx.Commit(); err != nil {
return nil, err
}
return &TaskOrder{ID: orderID, OrderNo: input.OrderNo, TaskID: task.ID, ProductName: task.Title,
CoverURL: task.CoverURL, Quantity: 1, CommissionAmount: task.CommissionAmount, Currency: "CNY",
MemberLevelID: member.LevelID, RuleSnapshot: rules, Message: "任务匹配成功,请在截止时间前完成",
OrderStatus: orderStatusPending, FreezeStatus: 1, DeadlineAt: input.DeadlineAt.UTC().Format(time.RFC3339),
MatchedAt: time.Now().UTC().Format(time.RFC3339), CreatedAt: time.Now().UTC().Format(time.RFC3339)}, nil
return &TaskOrder{ID: orderID, OrderNo: input.OrderNo, Currency: "CNY",
MemberLevelID: member.LevelID, RuleSnapshot: rules, Message: message,
OrderStatus: orderStatusMatching, FreezeStatus: 1, DeadlineAt: input.DeadlineAt.UTC().Format(time.RFC3339),
CreatedAt: time.Now().UTC().Format(time.RFC3339)}, nil
}
func (s *Store) AutoMatchTaskOrder(ctx context.Context, tenantID, memberID, orderID int64, delaySeconds int, minAmount, maxAmount int64) error {
tx, err := s.db.BeginTxx(ctx, nil)
if err != nil {
return err
}
defer tx.Rollback()
var order struct {
OrderNo string `db:"order_no"`
Status int `db:"order_status"`
MemberLevelID int64 `db:"member_level_id"`
RuleSnapshot json.RawMessage `db:"rule_snapshot"`
AutoDispatch int `db:"auto_dispatch"`
CreatedAt time.Time `db:"created_at"`
}
err = tx.GetContext(ctx, &order, `SELECT o.order_no,o.order_status,o.member_level_id,o.rule_snapshot,o.created_at,m.auto_dispatch
FROM hg_trade_order o JOIN hg_app_member m ON m.tenant_id=o.tenant_id AND m.id=o.member_id
WHERE o.tenant_id=? AND o.member_id=? AND o.id=? AND o.order_type=2 LIMIT 1 FOR UPDATE`, tenantID, memberID, orderID)
if errors.Is(err, sql.ErrNoRows) {
return ErrOrderNotFound
}
if err != nil {
return err
}
if order.Status != orderStatusMatching {
return tx.Commit()
}
if order.AutoDispatch != 1 || delaySeconds > 0 && time.Now().UTC().Before(order.CreatedAt.UTC().Add(time.Duration(delaySeconds)*time.Second)) {
return tx.Commit()
}
var account struct {
ID int64 `db:"id"`
AvailableAmount int64 `db:"available_amount"`
FrozenAmount int64 `db:"frozen_amount"`
Version int `db:"version"`
}
err = tx.GetContext(ctx, &account, `SELECT id,available_amount,frozen_amount,version FROM hg_wallet_account
WHERE tenant_id=? AND member_id=? AND currency='CNY' AND account_status=1 LIMIT 1 FOR UPDATE`, tenantID, memberID)
if errors.Is(err, sql.ErrNoRows) {
return ErrAccountRestricted
}
if err != nil {
return err
}
lowerAmount := max(int64(1), minAmount)
upperAmount := account.AvailableAmount
if maxAmount > 0 && maxAmount < upperAmount {
upperAmount = maxAmount
}
if lowerAmount > upperAmount {
return ErrNoTaskAvailable
}
var product struct {
ID int64 `db:"id"`
ProductNo string `db:"product_no"`
Name string `db:"name"`
PriceAmount int64 `db:"price_amount"`
Currency string `db:"currency"`
}
err = tx.GetContext(ctx, &product, `SELECT id,product_no,name,price_amount,currency FROM hg_product
WHERE tenant_id=? AND publish_status=3 AND deleted_at IS NULL AND currency='CNY' AND price_amount>=? AND price_amount<=?
AND (start_at IS NULL OR start_at<=UTC_TIMESTAMP(3)) AND (end_at IS NULL OR end_at>UTC_TIMESTAMP(3))
ORDER BY RAND() LIMIT 1 FOR UPDATE`, tenantID, lowerAmount, upperAmount)
if errors.Is(err, sql.ErrNoRows) {
return ErrNoTaskAvailable
}
if err != nil {
return err
}
var levelRule struct {
VersionID int64 `db:"version_id"`
CommissionRate int `db:"commission_rate"`
}
err = tx.GetContext(ctx, &levelRule, `SELECT v.id AS version_id,v.commission_rate
FROM hg_app_member_level l JOIN hg_app_member_level_version v
ON v.tenant_id=l.tenant_id AND v.id=l.current_version_id AND v.publish_status=3
WHERE l.tenant_id=? AND l.id=? AND l.status=1 AND l.deleted_at IS NULL LIMIT 1`, tenantID, order.MemberLevelID)
if errors.Is(err, sql.ErrNoRows) {
return ErrLevelNotEligible
}
if err != nil {
return err
}
snapshot := map[string]any{}
if len(order.RuleSnapshot) > 0 {
_ = json.Unmarshal(order.RuleSnapshot, &snapshot)
}
snapshot["match_source"] = "system_auto_match"
snapshot["product_id"] = product.ID
snapshot["product_no"] = product.ProductNo
snapshot["unit_price_amount"] = product.PriceAmount
snapshot["quantity"] = 1
snapshot["level_version_id"] = levelRule.VersionID
snapshot["commission_rate"] = levelRule.CommissionRate
rules, err := json.Marshal(snapshot)
if err != nil {
return err
}
commissionAmount := product.PriceAmount * int64(levelRule.CommissionRate) / 10000
afterAvailable := account.AvailableAmount - product.PriceAmount
afterFrozen := account.FrozenAmount + product.PriceAmount
walletResult, err := tx.ExecContext(ctx, `UPDATE hg_wallet_account SET available_amount=?,frozen_amount=?,version=version+1,
updated_at=UTC_TIMESTAMP(3) WHERE tenant_id=? AND id=? AND version=? AND available_amount=? AND frozen_amount=?`,
afterAvailable, afterFrozen, tenantID, account.ID, account.Version, account.AvailableAmount, account.FrozenAmount)
if err != nil {
return err
}
if affected, _ := walletResult.RowsAffected(); affected != 1 {
return ErrConcurrentUpdate
}
if _, err = insertTaskLedger(ctx, tx, tenantID, memberID, account.ID, orderID, fmt.Sprintf("WL%dMA", orderID),
"task_principal_hold", order.OrderNo, "available", 2, product.PriceAmount, account.AvailableAmount, afterAvailable,
"match:available", "自动派单扣除可用余额"); err != nil {
return err
}
if _, err = insertTaskLedger(ctx, tx, tenantID, memberID, account.ID, orderID, fmt.Sprintf("WL%dMF", orderID),
"task_principal_hold", order.OrderNo, "frozen", 1, product.PriceAmount, account.FrozenAmount, afterFrozen,
"match:frozen", "自动派单冻结订单本金"); err != nil {
return err
}
result, err := tx.ExecContext(ctx, `UPDATE hg_trade_order SET product_id=?,product_name=?,quantity=1,
unit_price_amount=?,total_amount=?,commission_amount=?,currency=?,rule_snapshot=?,message=?,
order_status=2,freeze_status=1,matched_at=UTC_TIMESTAMP(3),updated_at=UTC_TIMESTAMP(3)
WHERE tenant_id=? AND member_id=? AND id=? AND order_type=2 AND order_status=1`, product.ID, product.Name,
product.PriceAmount, product.PriceAmount, commissionAmount, product.Currency, rules,
"系统已自动匹配订单,订单本金已扣除,请在截止时间前完成", tenantID, memberID, orderID)
if err != nil {
return err
}
if affected, _ := result.RowsAffected(); affected != 1 {
return ErrConcurrentUpdate
}
return tx.Commit()
}
func (s *Store) TaskOrders(ctx context.Context, tenantID, memberID int64, status, offset, limit int) (*TaskOrderPage, error) {
@@ -364,13 +487,15 @@ func (s *Store) SubmitTask(ctx context.Context, input TaskSubmissionInput) (*Tas
}
defer tx.Rollback()
var order struct {
OrderNo string `db:"order_no"`
TaskID int64 `db:"task_id"`
MemberLevelID int64 `db:"member_level_id"`
TotalAmount int64 `db:"total_amount"`
CommissionAmount int64 `db:"commission_amount"`
OrderStatus int `db:"order_status"`
DeadlineAt string `db:"deadline_at"`
}
err = tx.GetContext(ctx, &order, `SELECT task_id,member_level_id,commission_amount,order_status,
err = tx.GetContext(ctx, &order, `SELECT order_no,task_id,member_level_id,total_amount,commission_amount,order_status,
COALESCE(JSON_UNQUOTE(JSON_EXTRACT(rule_snapshot,'$.deadline_at')),'') AS deadline_at
FROM hg_trade_order WHERE tenant_id=? AND member_id=? AND order_type=2 AND id=? LIMIT 1 FOR UPDATE`,
input.TenantID, input.MemberID, input.OrderID)
@@ -383,8 +508,48 @@ func (s *Store) SubmitTask(ctx context.Context, input TaskSubmissionInput) (*Tas
if order.OrderStatus != orderStatusPending {
return nil, ErrOrderState
}
var account struct {
ID int64 `db:"id"`
AvailableAmount int64 `db:"available_amount"`
FrozenAmount int64 `db:"frozen_amount"`
Status int `db:"account_status"`
Version int `db:"version"`
}
err = tx.GetContext(ctx, &account, `SELECT id,available_amount,frozen_amount,account_status,version FROM hg_wallet_account
WHERE tenant_id=? AND member_id=? AND currency='CNY' LIMIT 1 FOR UPDATE`, input.TenantID, input.MemberID)
if errors.Is(err, sql.ErrNoRows) {
return nil, ErrAccountNotFound
}
if err != nil {
return nil, err
}
if account.Status != 1 || account.FrozenAmount < order.TotalAmount {
return nil, ErrAccountRestricted
}
if order.DeadlineAt != "" {
if deadline, parseErr := time.Parse(time.RFC3339, order.DeadlineAt); parseErr != nil || !time.Now().UTC().Before(deadline) {
now := time.Now().UTC()
afterAvailable := account.AvailableAmount + order.TotalAmount
afterFrozen := account.FrozenAmount - order.TotalAmount
walletResult, updateErr := tx.ExecContext(ctx, `UPDATE hg_wallet_account SET available_amount=?,frozen_amount=?,
version=version+1,updated_at=? WHERE tenant_id=? AND id=? AND version=?`, afterAvailable, afterFrozen,
now, input.TenantID, account.ID, account.Version)
if updateErr != nil {
return nil, updateErr
}
if affected, _ := walletResult.RowsAffected(); affected != 1 {
return nil, ErrConcurrentUpdate
}
if _, updateErr = insertTaskLedger(ctx, tx, input.TenantID, input.MemberID, account.ID, input.OrderID,
fmt.Sprintf("WL%dEF", input.OrderID), "task_principal_release", order.OrderNo, "frozen", 2,
order.TotalAmount, account.FrozenAmount, afterFrozen, "expire:frozen", "订单超时释放冻结本金"); updateErr != nil {
return nil, updateErr
}
if _, updateErr = insertTaskLedger(ctx, tx, input.TenantID, input.MemberID, account.ID, input.OrderID,
fmt.Sprintf("WL%dEA", input.OrderID), "task_principal_release", order.OrderNo, "available", 1,
order.TotalAmount, account.AvailableAmount, afterAvailable, "expire:available", "订单超时退回订单本金"); updateErr != nil {
return nil, updateErr
}
if _, updateErr := tx.ExecContext(ctx, `UPDATE hg_trade_order SET order_status=5,message='任务已超时',updated_at=UTC_TIMESTAMP(3)
WHERE tenant_id=? AND id=? AND order_status=2`, input.TenantID, input.OrderID); updateErr != nil {
return nil, updateErr
@@ -398,24 +563,64 @@ func (s *Store) SubmitTask(ctx context.Context, input TaskSubmissionInput) (*Tas
if err := validateEvidenceAttachments(ctx, tx, input.TenantID, input.MemberID, input.AttachmentIDs); err != nil {
return nil, err
}
if input.SubmissionNo == "" {
return nil, ErrOrderState
}
attachments, err := json.Marshal(input.AttachmentIDs)
if err != nil {
return nil, err
}
now := time.Now().UTC()
_, err = tx.ExecContext(ctx, `INSERT INTO hg_task_submission
now := input.SettledAt.UTC()
if now.IsZero() {
now = time.Now().UTC()
}
principalAfter := account.AvailableAmount + order.TotalAmount
afterAmount := principalAfter + order.CommissionAmount
afterFrozen := account.FrozenAmount - order.TotalAmount
accountResult, err := tx.ExecContext(ctx, `UPDATE hg_wallet_account SET available_amount=?,frozen_amount=?,total_income=total_income+?,
version=version+1,updated_at=? WHERE tenant_id=? AND id=? AND version=?`, afterAmount, afterFrozen,
order.CommissionAmount, now, input.TenantID, account.ID, account.Version)
if err != nil {
return nil, err
}
if affected, _ := accountResult.RowsAffected(); affected != 1 {
return nil, ErrOrderState
}
if _, err = insertTaskLedger(ctx, tx, input.TenantID, input.MemberID, account.ID, input.OrderID,
fmt.Sprintf("WL%dSF", input.OrderID), "task_principal_release", order.OrderNo, "frozen", 2,
order.TotalAmount, account.FrozenAmount, afterFrozen, "settle:frozen", "任务完成释放冻结本金"); err != nil {
return nil, err
}
if _, err = insertTaskLedger(ctx, tx, input.TenantID, input.MemberID, account.ID, input.OrderID,
fmt.Sprintf("WL%dSA", input.OrderID), "task_principal_return", order.OrderNo, "available", 1,
order.TotalAmount, account.AvailableAmount, principalAfter, "settle:available", "任务完成返回订单本金"); err != nil {
return nil, err
}
ledgerID, err := insertTaskLedger(ctx, tx, input.TenantID, input.MemberID, account.ID, input.OrderID,
fmt.Sprintf("WL%dSC", input.OrderID), "commission", input.SubmissionNo, "available", 1,
order.CommissionAmount, principalAfter, afterAmount, "settle:commission", "任务完成自动返佣")
if err != nil {
return nil, err
}
submissionResult, err := tx.ExecContext(ctx, `INSERT INTO hg_task_submission
(tenant_id,submission_no,trade_order_id,task_id,member_id,member_level_id,evidence_text,
evidence_attachment_ids,submission_status,commission_amount,success_rate,allow_resubmit,created_by,updated_by)
VALUES(?,?,?,?,?,?,?,?,1,?,0,2,0,0)`, input.TenantID, input.SubmissionNo, input.OrderID,
order.TaskID, input.MemberID, order.MemberLevelID, input.EvidenceText, attachments, order.CommissionAmount)
evidence_attachment_ids,submission_status,commission_amount,success_rate,reviewed_by,reviewed_at,
review_reason_code,review_reason,allow_resubmit,ledger_id,settled_at,created_by,updated_by,created_at,updated_at)
VALUES(?,?,?,?,?,?,?,?,4,?,10000,0,?,'AUTO_SETTLED','会员提交后自动结算',2,?,?,0,0,?,?)`,
input.TenantID, input.SubmissionNo, input.OrderID, order.TaskID, input.MemberID, order.MemberLevelID,
input.EvidenceText, attachments, order.CommissionAmount, now, ledgerID, now, now, now)
if err != nil {
if isDuplicate(err) {
return nil, ErrOrderState
}
return nil, err
}
result, err := tx.ExecContext(ctx, `UPDATE hg_trade_order SET order_status=3,message='已提交,等待审核',updated_at=UTC_TIMESTAMP(3)
WHERE tenant_id=? AND id=? AND member_id=? AND order_status=2`, input.TenantID, input.OrderID, input.MemberID)
if affected, _ := submissionResult.RowsAffected(); affected != 1 {
return nil, ErrOrderState
}
result, err := tx.ExecContext(ctx, `UPDATE hg_trade_order SET order_status=4,message='订单完成,佣金已入账',
completed_at=?,updated_at=?
WHERE tenant_id=? AND id=? AND member_id=? AND order_status=2`, now, now, input.TenantID, input.OrderID, input.MemberID)
if err != nil {
return nil, err
}
@@ -425,7 +630,22 @@ func (s *Store) SubmitTask(ctx context.Context, input TaskSubmissionInput) (*Tas
if err := tx.Commit(); err != nil {
return nil, err
}
return &TaskSubmissionResult{SubmissionNo: input.SubmissionNo, OrderStatus: orderStatusProcessing, SubmittedAt: now}, nil
return &TaskSubmissionResult{SubmissionNo: input.SubmissionNo, OrderStatus: orderStatusCompleted, SubmittedAt: now}, nil
}
func insertTaskLedger(ctx context.Context, tx *sqlx.Tx, tenantID, memberID, accountID, orderID int64,
ledgerNo, bizType, bizNo, entryType string, direction int, amount, beforeAmount, afterAmount int64,
idempotencySuffix, remark string) (int64, error) {
result, err := tx.ExecContext(ctx, `INSERT INTO hg_wallet_ledger
(tenant_id,ledger_no,account_id,member_id,biz_type,biz_no,entry_type,direction,amount,
before_amount,after_amount,currency,reversal_of,idempotency_key,remark,operator_id,occurred_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,'CNY',0,?,?,?,UTC_TIMESTAMP(3))`, tenantID, ledgerNo, accountID, memberID,
bizType, bizNo, entryType, direction, amount, beforeAmount, afterAmount,
fmt.Sprintf("task:%d:%s", orderID, idempotencySuffix), remark, memberID)
if err != nil {
return 0, err
}
return result.LastInsertId()
}
func validateEvidenceAttachments(ctx context.Context, tx *sqlx.Tx, tenantID, memberID int64, ids []int64) error {
@@ -433,7 +653,7 @@ func validateEvidenceAttachments(ctx context.Context, tx *sqlx.Tx, tenantID, mem
return nil
}
query, args, err := sqlx.In(`SELECT COUNT(*) FROM hg_sys_attachment WHERE id IN (?)
AND tenant_id=? AND app_id='workbench' AND member_id=? AND status=1
AND tenant_id=? AND app_id='olons_server' AND member_id=? AND status=1
AND (mime_type LIKE 'image/%' OR mime_type='application/pdf')`, ids, tenantID, memberID)
if err != nil {
return err