package store import ( "context" cryptorand "crypto/rand" "database/sql" "encoding/json" "errors" "fmt" "math/big" "strconv" "strings" "time" "github.com/jmoiron/sqlx" ) const ( orderStatusMatching = 1 orderStatusPending = 2 orderStatusProcessing = 3 orderStatusCompleted = 4 orderStatusExpired = 5 orderStatusFrozen = 6 orderStatusCancelled = 7 orderStatusFailed = 8 ) type TaskHall struct { ID int64 `db:"id"` Code string `db:"hall_code"` Name string `db:"name"` IconURL string `db:"icon_url"` Summary string `db:"summary"` CommissionRate int `db:"commission_rate"` RateUnit string `db:"rate_unit"` ClientRoute string `db:"client_route"` } type TaskEligibility struct { MemberLevelID int64 `db:"member_level_id"` MemberStatus int `db:"member_status"` AutoDispatch int `db:"auto_dispatch"` HallEligible bool `db:"hall_eligible"` AvailableAmount int64 `db:"available_amount"` WalletStatus int `db:"wallet_status"` DailyLimit int `db:"daily_limit"` TodayCount int `db:"today_count"` PendingOrderID int64 `db:"pending_order_id"` } type ClaimTaskInput struct { TenantID int64 MemberID int64 HallID int64 MinimumAmount int64 AutoDelay int DeadlineAt time.Time OrderNo string } type TaskOrder struct { ID int64 `db:"id"` OrderNo string `db:"order_no"` TaskID int64 `db:"task_id"` ProductID int64 `db:"product_id"` ProductName string `db:"product_name"` CoverURL string `db:"cover_url"` Quantity int `db:"quantity"` UnitPriceAmount int64 `db:"unit_price_amount"` TotalAmount int64 `db:"total_amount"` CommissionAmount int64 `db:"commission_amount"` Currency string `db:"currency"` MemberLevelID int64 `db:"member_level_id"` RuleSnapshot json.RawMessage `db:"rule_snapshot"` Message string `db:"message"` OrderStatus int `db:"order_status"` FreezeStatus int `db:"freeze_status"` DeadlineAt string `db:"deadline_at"` MatchedAt string `db:"matched_at"` CompletedAt string `db:"completed_at"` CreatedAt string `db:"created_at"` UpdatedAt string `db:"updated_at"` TaskTitle string `db:"task_title"` TaskDescription string `db:"task_description"` TaskDetailHTML string `db:"task_detail_html"` SubmissionNo string `db:"submission_no"` SubmissionStatus int `db:"submission_status"` EvidenceText string `db:"evidence_text"` EvidenceIDs json.RawMessage `db:"evidence_attachment_ids"` ReviewReasonCode string `db:"review_reason_code"` ReviewReason string `db:"review_reason"` AllowResubmit int `db:"allow_resubmit"` SubmittedAt string `db:"submitted_at"` ReviewedAt string `db:"reviewed_at"` SettledAt string `db:"settled_at"` } type TaskOrderPage struct { Items []TaskOrder Total int64 } type TaskStats struct { AvailableAmount int64 `db:"available_amount"` FrozenAmount int64 `db:"frozen_amount"` TodayCommissionAmount int64 `db:"today_commission_amount"` YesterdayCommissionAmount int64 `db:"yesterday_commission_amount"` TodayClaimCount int `db:"today_claim_count"` TodayCompletedCount int `db:"today_completed_count"` } type TaskSubmissionInput struct { TenantID int64 MemberID int64 OrderID int64 SubmissionNo string EvidenceText string AttachmentIDs []int64 LedgerNo string IdempotencyKey string SettledAt time.Time } type TaskSubmissionResult struct { SubmissionNo string `db:"submission_no"` OrderStatus int SubmittedAt time.Time } func (s *Store) TaskHalls(ctx context.Context, tenantID, memberID int64) ([]TaskHall, error) { rows := make([]TaskHall, 0) err := s.db.SelectContext(ctx, &rows, `SELECT h.id,h.hall_code,h.name,h.icon_url,h.summary, h.commission_rate,h.rate_unit,h.client_route FROM hg_task_hall h JOIN hg_app_member m ON m.tenant_id=h.tenant_id AND m.id=? AND m.member_status=1 AND m.deleted_at IS NULL JOIN hg_task_hall_level hl ON hl.tenant_id=h.tenant_id AND hl.hall_id=h.id AND hl.level_id=m.level_id WHERE h.tenant_id=? AND h.publish_status=3 AND h.deleted_at IS NULL ORDER BY h.sort,h.id`, memberID, tenantID) return rows, err } func (s *Store) TaskEligibility(ctx context.Context, tenantID, memberID, hallID int64) (*TaskEligibility, error) { var row TaskEligibility err := s.db.GetContext(ctx, &row, `SELECT m.level_id AS member_level_id,m.member_status,m.auto_dispatch, EXISTS(SELECT 1 FROM hg_task_hall h JOIN hg_task_hall_level hl ON hl.tenant_id=h.tenant_id AND hl.hall_id=h.id AND hl.level_id=m.level_id WHERE h.tenant_id=m.tenant_id AND h.id=? AND h.publish_status=3 AND h.deleted_at IS NULL) AS hall_eligible, COALESCE(w.available_amount,0) AS available_amount,COALESCE(w.account_status,0) AS wallet_status, COALESCE(lv.daily_order_limit,0) AS daily_limit, (SELECT COUNT(*) FROM hg_trade_order o WHERE o.tenant_id=m.tenant_id AND o.member_id=m.id AND o.order_type=2 AND o.created_at>=UTC_DATE() AND o.created_at 0 { return nil, ErrPendingOrderExists } var levelRule struct { DailyLimit int `db:"daily_order_limit"` RuleJSON json.RawMessage `db:"rule_json"` } err = tx.GetContext(ctx, &levelRule, `SELECT COALESCE(v.daily_order_limit,0) AS daily_order_limit, COALESCE(v.rule_json,JSON_OBJECT()) AS rule_json FROM hg_app_member_level l LEFT 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`, input.TenantID, member.LevelID) if errors.Is(err, sql.ErrNoRows) { return nil, ErrLevelNotEligible } if err != nil { return nil, err } if levelRule.DailyLimit > 0 { var todayCount int if err := tx.GetContext(ctx, &todayCount, `SELECT COUNT(*) FROM hg_trade_order WHERE tenant_id=? AND member_id=? AND order_type=2 AND created_at>=UTC_DATE() AND created_at= levelRule.DailyLimit { return nil, ErrDailyLimitReached } } dispatchMode, message := "manual", "等待后台匹配订单" matchAt := time.Now().UTC() deadlineAt := input.DeadlineAt.UTC() delaySeconds := 0 if member.AutoDispatch == 1 { dispatchMode, message = "auto", "正在自动匹配订单" delayMin, delayMax := input.AutoDelay, input.AutoDelay var levelRules map[string]any if json.Unmarshal(levelRule.RuleJSON, &levelRules) == nil { if value, ok := jsonInt(levelRules["auto_order_delay_min_seconds"]); ok { delayMin = value } if value, ok := jsonInt(levelRules["auto_order_delay_max_seconds"]); ok { delayMax = value } } if delayMin < 0 { delayMin = 0 } if delayMax < delayMin { delayMax = delayMin } delaySeconds = randomDelaySeconds(delayMin, delayMax) matchAt = matchAt.Add(time.Duration(delaySeconds) * time.Second) completionWindow := input.DeadlineAt.UTC().Sub(time.Now().UTC()) if completionWindow > 0 { deadlineAt = matchAt.Add(completionWindow) } } rules, err := json.Marshal(map[string]any{ "source": "member_auto_claim", "dispatch_mode": dispatchMode, "hall_id": input.HallID, "deadline_at": deadlineAt.Format(time.RFC3339), "scheduled_match_at": matchAt.Format(time.RFC3339), "dispatch_delay_seconds": delaySeconds, "member_level_id": member.LevelID, }) if err != nil { return nil, err } 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,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 } return nil, err } orderID, err := result.LastInsertId() if err != nil { return nil, err } if err := tx.Commit(); err != nil { return nil, err } return &TaskOrder{ID: orderID, OrderNo: input.OrderNo, Currency: "CNY", MemberLevelID: member.LevelID, RuleSnapshot: rules, Message: message, OrderStatus: orderStatusMatching, FreezeStatus: 1, DeadlineAt: deadlineAt.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() } matchAt := order.CreatedAt.UTC().Add(time.Duration(delaySeconds) * time.Second) snapshot := map[string]any{} if len(order.RuleSnapshot) > 0 && json.Unmarshal(order.RuleSnapshot, &snapshot) == nil { if raw, ok := snapshot["scheduled_match_at"].(string); ok { if parsed, parseErr := time.Parse(time.RFC3339, raw); parseErr == nil { matchAt = parsed.UTC() } } } dispatchMode, _ := snapshot["dispatch_mode"].(string) isAutoDispatch := dispatchMode == "auto" || dispatchMode == "" && order.AutoDispatch == 1 if !isAutoDispatch || time.Now().UTC().Before(matchAt) { 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["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 holdAvailable := account.AvailableAmount - product.PriceAmount holdFrozen := account.FrozenAmount + product.PriceAmount settledAvailable := account.AvailableAmount + commissionAmount now := time.Now().UTC() walletResult, err := tx.ExecContext(ctx, `UPDATE hg_wallet_account SET available_amount=?,frozen_amount=?,total_income=total_income+?,version=version+1, updated_at=UTC_TIMESTAMP(3) WHERE tenant_id=? AND id=? AND version=? AND available_amount=? AND frozen_amount=?`, settledAvailable, account.FrozenAmount, commissionAmount, 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, holdAvailable, "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, holdFrozen, "match:frozen", "自动派单冻结订单本金"); err != nil { return err } if _, err = insertTaskLedger(ctx, tx, tenantID, memberID, account.ID, orderID, fmt.Sprintf("WL%dSF", orderID), "task_principal_release", order.OrderNo, "frozen", 2, product.PriceAmount, holdFrozen, account.FrozenAmount, "settle:frozen", "自动订单完成释放冻结本金"); err != nil { return err } if _, err = insertTaskLedger(ctx, tx, tenantID, memberID, account.ID, orderID, fmt.Sprintf("WL%dSA", orderID), "task_principal_return", order.OrderNo, "available", 1, product.PriceAmount, holdAvailable, account.AvailableAmount, "settle:available", "自动订单完成返回本金"); err != nil { return err } submissionNo := fmt.Sprintf("TSAUTO%d", orderID) commissionLedgerID, err := insertTaskLedger(ctx, tx, tenantID, memberID, account.ID, orderID, fmt.Sprintf("WL%dSC", orderID), "commission", submissionNo, "available", 1, commissionAmount, account.AvailableAmount, settledAvailable, "settle:commission", "自动订单完成返佣") if err != nil { return err } if _, 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,reviewed_by,reviewed_at, review_reason_code,review_reason,allow_resubmit,ledger_id,settled_at,created_by,updated_by,created_at,updated_at) VALUES(?,?,?,?,?,?,'',JSON_ARRAY(),4,?,10000,0,?,'AUTO_MATCHED','自动派单完成',2,?,?,0,0,?,?)`, tenantID, submissionNo, orderID, 0, memberID, order.MemberLevelID, commissionAmount, now, commissionLedgerID, now, now, now); 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=4,freeze_status=1,matched_at=?,completed_at=?,updated_at=? 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, "系统已自动匹配并完成订单,佣金已入账", now, now, now, tenantID, memberID, orderID) if err != nil { return err } if affected, _ := result.RowsAffected(); affected != 1 { return ErrConcurrentUpdate } return tx.Commit() } func jsonInt(value any) (int, bool) { switch typed := value.(type) { case float64: return int(typed), true case json.Number: parsed, err := typed.Int64() return int(parsed), err == nil case string: parsed, err := strconv.Atoi(strings.TrimSpace(typed)) return parsed, err == nil default: return 0, false } } func randomDelaySeconds(minimum, maximum int) int { if maximum <= minimum { return minimum } value, err := cryptorand.Int(cryptorand.Reader, big.NewInt(int64(maximum-minimum+1))) if err != nil { return minimum } return minimum + int(value.Int64()) } func (s *Store) TaskOrders(ctx context.Context, tenantID, memberID int64, status, offset, limit int) (*TaskOrderPage, error) { where := ` WHERE o.tenant_id=? AND o.member_id=? AND o.order_type=2` args := []any{tenantID, memberID} if status > 0 { where += ` AND o.order_status=?` args = append(args, status) } var total int64 if err := s.db.GetContext(ctx, &total, `SELECT COUNT(*) FROM hg_trade_order o`+where, args...); err != nil { return nil, err } args = append(args, limit, offset) rows := make([]TaskOrder, 0) err := s.db.SelectContext(ctx, &rows, orderSelectSQL()+where+` ORDER BY o.created_at DESC,o.id DESC LIMIT ? OFFSET ?`, args...) return &TaskOrderPage{Items: rows, Total: total}, err } func (s *Store) TaskOrder(ctx context.Context, tenantID, memberID, orderID int64) (*TaskOrder, error) { var row TaskOrder err := s.db.GetContext(ctx, &row, orderSelectSQL()+` WHERE o.tenant_id=? AND o.member_id=? AND o.order_type=2 AND o.id=? LIMIT 1`, tenantID, memberID, orderID) if errors.Is(err, sql.ErrNoRows) { return nil, ErrOrderNotFound } return &row, err } func orderSelectSQL() string { return `SELECT o.id,o.order_no,o.task_id,o.product_id,COALESCE(NULLIF(o.product_name,''),t.title,'') AS product_name, COALESCE(NULLIF(p.cover_url,''),t.cover_url,'') AS cover_url,o.quantity,o.unit_price_amount, o.total_amount,o.commission_amount,o.currency,o.member_level_id, COALESCE(o.rule_snapshot,JSON_OBJECT()) AS rule_snapshot,o.message,o.order_status,o.freeze_status, COALESCE(JSON_UNQUOTE(JSON_EXTRACT(o.rule_snapshot,'$.deadline_at')),'') AS deadline_at, COALESCE(DATE_FORMAT(o.matched_at,'%Y-%m-%dT%H:%i:%sZ'),'') AS matched_at, COALESCE(DATE_FORMAT(o.completed_at,'%Y-%m-%dT%H:%i:%sZ'),'') AS completed_at, DATE_FORMAT(o.created_at,'%Y-%m-%dT%H:%i:%sZ') AS created_at, DATE_FORMAT(o.updated_at,'%Y-%m-%dT%H:%i:%sZ') AS updated_at, COALESCE(t.title,'') AS task_title,COALESCE(t.description,'') AS task_description, COALESCE(t.detail_html,'') AS task_detail_html,COALESCE(s.submission_no,'') AS submission_no, COALESCE(s.submission_status,0) AS submission_status,COALESCE(s.evidence_text,'') AS evidence_text, COALESCE(s.evidence_attachment_ids,JSON_ARRAY()) AS evidence_attachment_ids, COALESCE(s.review_reason_code,'') AS review_reason_code,COALESCE(s.review_reason,'') AS review_reason, COALESCE(s.allow_resubmit,2) AS allow_resubmit, COALESCE(DATE_FORMAT(s.created_at,'%Y-%m-%dT%H:%i:%sZ'),'') AS submitted_at, COALESCE(DATE_FORMAT(s.reviewed_at,'%Y-%m-%dT%H:%i:%sZ'),'') AS reviewed_at, COALESCE(DATE_FORMAT(s.settled_at,'%Y-%m-%dT%H:%i:%sZ'),'') AS settled_at FROM hg_trade_order o LEFT JOIN hg_task t ON t.tenant_id=o.tenant_id AND t.id=o.task_id LEFT JOIN hg_product p ON p.tenant_id=o.tenant_id AND p.id=o.product_id LEFT JOIN hg_task_submission s ON s.tenant_id=o.tenant_id AND s.trade_order_id=o.id` } func (s *Store) TaskStats(ctx context.Context, tenantID, memberID int64) (*TaskStats, error) { var row TaskStats err := s.db.GetContext(ctx, &row, `SELECT COALESCE(w.available_amount,0) AS available_amount,COALESCE(w.frozen_amount,0) AS frozen_amount, (SELECT COALESCE(SUM(ts.commission_amount),0) FROM hg_task_submission ts WHERE ts.tenant_id=? AND ts.member_id=? AND ts.submission_status=4 AND ts.settled_at>=UTC_DATE() AND ts.settled_at=UTC_DATE()-INTERVAL 1 DAY AND ys.settled_at=UTC_DATE() AND tc.created_at=UTC_DATE() AND cc.completed_at