701 lines
30 KiB
Go
701 lines
30 KiB
Go
package store
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"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"`
|
|
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
|
|
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,
|
|
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<UTC_DATE()+INTERVAL 1 DAY) AS today_count,
|
|
COALESCE((SELECT o2.id FROM hg_trade_order o2 WHERE o2.tenant_id=m.tenant_id AND o2.member_id=m.id
|
|
AND o2.order_type=2 AND o2.order_status IN (1,2,3,6) ORDER BY o2.id DESC LIMIT 1),0) AS pending_order_id
|
|
FROM hg_app_member m
|
|
LEFT JOIN hg_app_member_level l ON l.tenant_id=m.tenant_id AND l.id=m.level_id AND l.status=1 AND l.deleted_at IS NULL
|
|
LEFT JOIN hg_app_member_level_version lv ON lv.tenant_id=l.tenant_id AND lv.id=l.current_version_id AND lv.publish_status=3
|
|
LEFT JOIN hg_wallet_account w ON w.tenant_id=m.tenant_id AND w.member_id=m.id AND w.currency='CNY'
|
|
WHERE m.tenant_id=? AND m.id=? AND m.deleted_at IS NULL LIMIT 1`, hallID, tenantID, memberID)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return nil, ErrAccountNotFound
|
|
}
|
|
return &row, err
|
|
}
|
|
|
|
func (s *Store) ClaimTask(ctx context.Context, input ClaimTaskInput) (*TaskOrder, error) {
|
|
tx, err := s.db.BeginTxx(ctx, nil)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer tx.Rollback()
|
|
|
|
var member struct {
|
|
LevelID int64 `db:"level_id"`
|
|
AutoDispatch int `db:"auto_dispatch"`
|
|
}
|
|
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
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var hallCount int
|
|
err = tx.GetContext(ctx, &hallCount, `SELECT COUNT(*) 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=?
|
|
WHERE h.tenant_id=? AND h.id=? AND h.publish_status=3 AND h.deleted_at IS NULL`, member.LevelID, input.TenantID, input.HallID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if hallCount == 0 {
|
|
return nil, ErrLevelNotEligible
|
|
}
|
|
var wallet struct {
|
|
AvailableAmount int64 `db:"available_amount"`
|
|
Status int `db:"account_status"`
|
|
}
|
|
err = tx.GetContext(ctx, &wallet, `SELECT available_amount,account_status 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, ErrInsufficientBalance
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if wallet.Status != 1 {
|
|
return nil, ErrAccountRestricted
|
|
}
|
|
if wallet.AvailableAmount < input.MinimumAmount {
|
|
return nil, ErrInsufficientBalance
|
|
}
|
|
var pending int64
|
|
err = tx.GetContext(ctx, &pending, `SELECT id FROM hg_trade_order
|
|
WHERE tenant_id=? AND member_id=? AND order_type=2 AND order_status IN (1,2,3,6)
|
|
ORDER BY id DESC LIMIT 1 FOR UPDATE`, input.TenantID, input.MemberID)
|
|
if err != nil && !errors.Is(err, sql.ErrNoRows) {
|
|
return nil, err
|
|
}
|
|
if pending > 0 {
|
|
return nil, ErrPendingOrderExists
|
|
}
|
|
var dailyLimit int
|
|
err = tx.GetContext(ctx, &dailyLimit, `SELECT COALESCE(v.daily_order_limit,0) 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 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<UTC_DATE()+INTERVAL 1 DAY`, input.TenantID, input.MemberID); err != nil {
|
|
return nil, err
|
|
}
|
|
if todayCount >= dailyLimit {
|
|
return nil, ErrDailyLimitReached
|
|
}
|
|
}
|
|
dispatchMode, message := "manual", "等待后台匹配订单"
|
|
if member.AutoDispatch == 1 {
|
|
dispatchMode, message = "auto", "正在自动匹配订单"
|
|
}
|
|
rules, err := json.Marshal(map[string]any{
|
|
"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
|
|
}
|
|
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: 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) {
|
|
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) AS today_commission_amount,
|
|
(SELECT COALESCE(SUM(ys.commission_amount),0) FROM hg_task_submission ys WHERE ys.tenant_id=? AND ys.member_id=?
|
|
AND ys.submission_status=4 AND ys.settled_at>=UTC_DATE()-INTERVAL 1 DAY AND ys.settled_at<UTC_DATE()) AS yesterday_commission_amount,
|
|
(SELECT COUNT(*) FROM hg_trade_order tc WHERE tc.tenant_id=? AND tc.member_id=?
|
|
AND tc.order_type=2 AND tc.created_at>=UTC_DATE() AND tc.created_at<UTC_DATE()+INTERVAL 1 DAY) AS today_claim_count,
|
|
(SELECT COUNT(*) FROM hg_trade_order cc WHERE cc.tenant_id=? AND cc.member_id=? AND cc.order_type=2 AND cc.order_status=4
|
|
AND cc.completed_at>=UTC_DATE() AND cc.completed_at<UTC_DATE()+INTERVAL 1 DAY) AS today_completed_count
|
|
FROM hg_app_member m LEFT JOIN hg_wallet_account w ON w.tenant_id=m.tenant_id AND w.member_id=m.id AND w.currency='CNY'
|
|
WHERE m.tenant_id=? AND m.id=? AND m.member_status=1 AND m.deleted_at IS NULL LIMIT 1`,
|
|
tenantID, memberID, tenantID, memberID, tenantID, memberID, tenantID, memberID, tenantID, memberID)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return nil, ErrAccountNotFound
|
|
}
|
|
return &row, err
|
|
}
|
|
|
|
func (s *Store) SubmitTask(ctx context.Context, input TaskSubmissionInput) (*TaskSubmissionResult, error) {
|
|
tx, err := s.db.BeginTxx(ctx, nil)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
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 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)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return nil, ErrOrderNotFound
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
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
|
|
}
|
|
if commitErr := tx.Commit(); commitErr != nil {
|
|
return nil, commitErr
|
|
}
|
|
return nil, ErrOrderExpired
|
|
}
|
|
}
|
|
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 := 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,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
|
|
}
|
|
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
|
|
}
|
|
if affected, _ := result.RowsAffected(); affected != 1 {
|
|
return nil, ErrOrderState
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return nil, err
|
|
}
|
|
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 {
|
|
if len(ids) == 0 {
|
|
return nil
|
|
}
|
|
query, args, err := sqlx.In(`SELECT COUNT(*) FROM hg_sys_attachment WHERE id IN (?)
|
|
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
|
|
}
|
|
var count int
|
|
if err := tx.GetContext(ctx, &count, query, args...); err != nil {
|
|
return err
|
|
}
|
|
if count != len(ids) {
|
|
return ErrEvidenceInvalid
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Store) OrderLedgers(ctx context.Context, tenantID, memberID int64, businessNumbers []string) ([]map[string]any, error) {
|
|
if len(businessNumbers) == 0 {
|
|
return []map[string]any{}, nil
|
|
}
|
|
query, args, err := sqlx.In(`SELECT id,ledger_no,biz_type,biz_no,entry_type,direction,amount,before_amount,after_amount,currency,remark,
|
|
DATE_FORMAT(occurred_at,'%Y-%m-%dT%H:%i:%sZ') AS occurred_at FROM hg_wallet_ledger
|
|
WHERE tenant_id=? AND member_id=? AND biz_no IN (?) ORDER BY occurred_at,id`, tenantID, memberID, businessNumbers)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
rows, err := s.db.QueryxContext(ctx, query, args...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := make([]map[string]any, 0)
|
|
for rows.Next() {
|
|
item := map[string]any{}
|
|
if err := rows.MapScan(item); err != nil {
|
|
return nil, err
|
|
}
|
|
for key, value := range item {
|
|
if bytes, ok := value.([]byte); ok {
|
|
item[key] = string(bytes)
|
|
}
|
|
}
|
|
items = append(items, item)
|
|
}
|
|
return items, rows.Err()
|
|
}
|