feat(payment): 支付管理 - 合并 PR #533 支付订单与对账 CSV 导出
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
commit
ef8f49e4fc
18 files changed
+1120
-65
No files matched your search
@@ -382,23 +382,44 @@ func (r *Repository) PaymentOrdersNeedingQuery(cutoff time.Time, limit int) ([]m
|
||||
return items, err
|
||||
}
|
||||
|
||||
func (r *Repository) AdminPaymentOrders(status, keyword string, limit, offset int) ([]model.PaymentOrder, int64, error) {
|
||||
func (r *Repository) AdminPaymentOrders(filter PaymentOrderFilter, limit, offset int) ([]model.PaymentOrder, int64, error) {
|
||||
var items []model.PaymentOrder
|
||||
var total int64
|
||||
query := r.paymentOrderQuery(filter)
|
||||
if err := query.Count(&total).Error; err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
err := query.Order("created_at desc, id desc").Limit(limit).Offset(offset).Find(&items).Error
|
||||
return items, total, err
|
||||
}
|
||||
|
||||
func (r *Repository) paymentOrderQuery(filter PaymentOrderFilter) *gorm.DB {
|
||||
query := r.db.Model(&model.PaymentOrder{})
|
||||
if normalized := strings.TrimSpace(status); normalized != "" && normalized != "all" {
|
||||
if normalized := strings.TrimSpace(filter.Status); normalized != "" && normalized != "all" {
|
||||
query = query.Where("status = ?", normalized)
|
||||
}
|
||||
if normalized := strings.TrimSpace(keyword); normalized != "" {
|
||||
if normalized := strings.TrimSpace(filter.Keyword); normalized != "" {
|
||||
like := "%" + normalized + "%"
|
||||
users := r.db.Model(&model.User{}).Select("id").Where("LOWER(username) LIKE ? OR LOWER(display_name) LIKE ? OR LOWER(email) LIKE ?", strings.ToLower(like), strings.ToLower(like), strings.ToLower(like))
|
||||
query = query.Where("merchant_order_no LIKE ? OR provider_trade_no LIKE ? OR user_id = ? OR user_id IN (?)", like, like, normalized, users)
|
||||
}
|
||||
if err := query.Count(&total).Error; err != nil {
|
||||
return nil, 0, err
|
||||
if filter.ProviderID != "" && filter.ProviderID != "all" {
|
||||
query = query.Where("provider_id = ?", filter.ProviderID)
|
||||
}
|
||||
err := query.Order("created_at desc").Limit(limit).Offset(offset).Find(&items).Error
|
||||
return items, total, err
|
||||
column := "created_at"
|
||||
switch filter.TimeField {
|
||||
case "paid":
|
||||
column = "provider_paid_at"
|
||||
case "credited":
|
||||
column = "credited_at"
|
||||
}
|
||||
if !filter.From.IsZero() {
|
||||
query = query.Where(column+" >= ?", filter.From)
|
||||
}
|
||||
if !filter.Until.IsZero() {
|
||||
query = query.Where(column+" < ?", filter.Until)
|
||||
}
|
||||
return query
|
||||
}
|
||||
|
||||
// BeginPaymentReconciliation creates or safely reuses the single daily run
|
||||
@@ -503,34 +524,42 @@ func (r *Repository) PaymentReconciliationRunByDate(providerID, billDate string)
|
||||
return &run, r.db.First(&run, "provider_id = ? AND bill_date = ?", strings.TrimSpace(providerID), strings.TrimSpace(billDate)).Error
|
||||
}
|
||||
|
||||
func (r *Repository) AdminPaymentReconciliationRuns(providerID, status string, limit, offset int) ([]model.PaymentReconciliationRun, int64, error) {
|
||||
func (r *Repository) AdminPaymentReconciliationRuns(filter PaymentReconciliationFilter, limit, offset int) ([]model.PaymentReconciliationRun, int64, error) {
|
||||
var runs []model.PaymentReconciliationRun
|
||||
var total int64
|
||||
query := r.db.Model(&model.PaymentReconciliationRun{})
|
||||
if normalized := strings.TrimSpace(providerID); normalized != "" && normalized != "all" {
|
||||
query = query.Where("provider_id = ?", normalized)
|
||||
}
|
||||
if normalized := strings.TrimSpace(status); normalized != "" && normalized != "all" {
|
||||
query = query.Where("status = ?", normalized)
|
||||
}
|
||||
query := r.paymentReconciliationQuery(filter)
|
||||
if err := query.Count(&total).Error; err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
err := query.Order("bill_date desc, started_at desc").Limit(limit).Offset(offset).Find(&runs).Error
|
||||
err := query.Order("bill_date desc, started_at desc, id desc").Limit(limit).Offset(offset).Find(&runs).Error
|
||||
return runs, total, err
|
||||
}
|
||||
|
||||
func (r *Repository) paymentReconciliationQuery(filter PaymentReconciliationFilter) *gorm.DB {
|
||||
query := r.db.Model(&model.PaymentReconciliationRun{})
|
||||
if normalized := strings.TrimSpace(filter.ProviderID); normalized != "" && normalized != "all" {
|
||||
query = query.Where("provider_id = ?", normalized)
|
||||
}
|
||||
if normalized := strings.TrimSpace(filter.Status); normalized != "" && normalized != "all" {
|
||||
query = query.Where("status = ?", normalized)
|
||||
}
|
||||
if filter.From != "" {
|
||||
query = query.Where("bill_date >= ?", filter.From)
|
||||
}
|
||||
if filter.To != "" {
|
||||
query = query.Where("bill_date <= ?", filter.To)
|
||||
}
|
||||
return query
|
||||
}
|
||||
|
||||
func (r *Repository) PaymentReconciliationItems(runID, result string, limit, offset int) ([]model.PaymentReconciliationItem, int64, error) {
|
||||
var items []model.PaymentReconciliationItem
|
||||
var total int64
|
||||
query := r.db.Model(&model.PaymentReconciliationItem{}).Where("run_id = ?", strings.TrimSpace(runID))
|
||||
if normalized := strings.TrimSpace(result); normalized != "" && normalized != "all" {
|
||||
query = query.Where("result = ?", normalized)
|
||||
}
|
||||
query := r.paymentReconciliationItemsQuery(runID, result)
|
||||
if err := query.Count(&total).Error; err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
err := query.Order("resolved asc, created_at asc").Limit(limit).Offset(offset).Find(&items).Error
|
||||
err := query.Order("resolved asc, created_at asc, id asc").Limit(limit).Offset(offset).Find(&items).Error
|
||||
return items, total, err
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,80 @@
|
||||
package repository
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"errors"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"infinite-canvas/backend/internal/model"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
type PaymentOrderFilter struct {
|
||||
Status, Keyword, ProviderID, TimeField string
|
||||
From, Until time.Time
|
||||
}
|
||||
|
||||
type PaymentReconciliationFilter struct {
|
||||
ProviderID, Status, From, To string
|
||||
}
|
||||
|
||||
var ErrPaymentReconciliationRunning = errors.New("payment reconciliation is running")
|
||||
|
||||
func (r *Repository) paymentReconciliationItemsQuery(runID, result string) *gorm.DB {
|
||||
query := r.db.Model(&model.PaymentReconciliationItem{}).Where("run_id = ?", strings.TrimSpace(runID))
|
||||
if normalized := strings.TrimSpace(result); normalized == "abnormal" {
|
||||
query = query.Where("result NOT IN ?", []string{"matched", "recovered"})
|
||||
} else if normalized != "" && normalized != "all" {
|
||||
query = query.Where("result = ?", normalized)
|
||||
}
|
||||
return query
|
||||
}
|
||||
|
||||
func (r *Repository) ExportPaymentOrders(filter PaymentOrderFilter, limit int) (orders []model.PaymentOrder, users map[string]model.User, err error) {
|
||||
users = make(map[string]model.User)
|
||||
err = r.db.Transaction(func(tx *gorm.DB) error {
|
||||
snapshot := New(tx)
|
||||
if err := snapshot.paymentOrderQuery(filter).Order("created_at desc, id desc").Limit(limit).Find(&orders).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
// Bound IN parameters for SQLite as well as PostgreSQL.
|
||||
for start := 0; start < len(orders); start += 500 {
|
||||
ids := make([]string, 0, 500)
|
||||
for _, order := range orders[start:min(start+500, len(orders))] {
|
||||
ids = append(ids, order.UserID)
|
||||
}
|
||||
batch, err := snapshot.UsersByIDs(ids)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for id, user := range batch {
|
||||
users[id] = user
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}, &sql.TxOptions{Isolation: sql.LevelRepeatableRead, ReadOnly: true})
|
||||
return
|
||||
}
|
||||
|
||||
func (r *Repository) ExportPaymentReconciliations(filter PaymentReconciliationFilter, limit int) ([]model.PaymentReconciliationRun, error) {
|
||||
var runs []model.PaymentReconciliationRun
|
||||
err := r.paymentReconciliationQuery(filter).Order("bill_date desc, started_at desc, id desc").Limit(limit).Find(&runs).Error
|
||||
return runs, err
|
||||
}
|
||||
|
||||
func (r *Repository) ExportPaymentReconciliationItems(runID, result string, limit int) (run model.PaymentReconciliationRun, items []model.PaymentReconciliationItem, err error) {
|
||||
// A rerun deletes old items and reuses the run ID. Both reads must observe
|
||||
// the same snapshot, including on PostgreSQL (whose default is read committed).
|
||||
err = r.db.Transaction(func(tx *gorm.DB) error {
|
||||
if err := tx.First(&run, "id = ?", strings.TrimSpace(runID)).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if run.Status == model.PaymentReconciliationRunning {
|
||||
return ErrPaymentReconciliationRunning
|
||||
}
|
||||
return New(tx).paymentReconciliationItemsQuery(run.ID, result).Order("resolved asc, created_at asc, id asc").Limit(limit).Find(&items).Error
|
||||
}, &sql.TxOptions{Isolation: sql.LevelRepeatableRead, ReadOnly: true})
|
||||
return
|
||||
}
|
||||
@@ -0,0 +1,149 @@
|
||||
package repository
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"infinite-canvas/backend/internal/model"
|
||||
|
||||
"gorm.io/driver/sqlite"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
func TestPaymentExportMatchesFiltersAcrossPages(t *testing.T) {
|
||||
db := openPaymentTestDB(t)
|
||||
if err := db.AutoMigrate(&model.User{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
user := model.User{ID: "user", Username: "alice", Email: "alice@example.com"}
|
||||
if err := db.Create(&user).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
start := time.Date(2026, 9, 1, 0, 0, 0, 0, time.FixedZone("CST", 8*3600))
|
||||
end := start.AddDate(0, 0, 1)
|
||||
middle := start.Add(time.Hour)
|
||||
orders := []model.PaymentOrder{
|
||||
{ID: "a", UserID: user.ID, IdempotencyKey: "a", MerchantOrderNo: "a", ProviderID: "wechat-native", Status: model.PaymentOrderCredited, CreatedAt: start, ProviderPaidAt: &middle, CreditedAt: &end},
|
||||
{ID: "b", UserID: user.ID, IdempotencyKey: "b", MerchantOrderNo: "b", ProviderID: "wechat-native", Status: model.PaymentOrderCredited, CreatedAt: middle, ProviderPaidAt: &middle},
|
||||
{ID: "c", UserID: user.ID, IdempotencyKey: "c", MerchantOrderNo: "c", ProviderID: "wechat-native", Status: model.PaymentOrderCredited, CreatedAt: end, ProviderPaidAt: &end},
|
||||
{ID: "d", UserID: user.ID, IdempotencyKey: "d", MerchantOrderNo: "d", ProviderID: "alipay-page-pay", Status: model.PaymentOrderCredited, CreatedAt: middle},
|
||||
}
|
||||
if err := db.Create(&orders).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
repo := New(db)
|
||||
filter := PaymentOrderFilter{Keyword: "ALICE@EXAMPLE.COM", Status: "credited", ProviderID: "wechat-native", From: start, Until: end}
|
||||
page, total, err := repo.AdminPaymentOrders(filter, 1, 0)
|
||||
if err != nil || total != 2 || len(page) != 1 {
|
||||
t.Fatalf("invalid list: %v %d %v", page, total, err)
|
||||
}
|
||||
exported, users, err := repo.ExportPaymentOrders(filter, 10001)
|
||||
if err != nil || len(exported) != 2 || users[user.ID].Username != "alice" || exported[0].ID != page[0].ID {
|
||||
t.Fatalf("export differs from list: %v %v", exported, err)
|
||||
}
|
||||
filter.TimeField = "paid"
|
||||
exported, _, err = repo.ExportPaymentOrders(filter, 10001)
|
||||
if err != nil || len(exported) != 2 {
|
||||
t.Fatalf("paid bounds: %v %v", exported, err)
|
||||
}
|
||||
filter.TimeField = "credited"
|
||||
exported, _, err = repo.ExportPaymentOrders(filter, 10001)
|
||||
if err != nil || len(exported) != 0 {
|
||||
t.Fatalf("credited null/end boundary: %v %v", exported, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPaymentReconciliationExportSnapshotAndRerun(t *testing.T) {
|
||||
db := openPaymentTestDB(t)
|
||||
if err := db.AutoMigrate(&model.PaymentReconciliationRun{}, &model.PaymentReconciliationItem{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
repo := New(db)
|
||||
run := model.PaymentReconciliationRun{ID: "run", ProviderID: "wechat-native", BillDate: "2026-09-01", Status: model.PaymentReconciliationCompleted, TotalItems: 3, MatchItems: 1, RecoveredItems: 1, ErrorItems: 1}
|
||||
if err := db.Create(&run).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
items := []model.PaymentReconciliationItem{{ID: "a", RunID: run.ID, Result: model.PaymentReconciliationMatched, Resolved: true}, {ID: "b", RunID: run.ID, Result: model.PaymentReconciliationRecovered, Resolved: true}, {ID: "c", RunID: run.ID, Result: model.PaymentReconciliationAmountMismatch}}
|
||||
if err := db.Create(&items).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, exported, err := repo.ExportPaymentReconciliationItems(run.ID, "abnormal", 10001)
|
||||
if err != nil || len(exported) != 1 || exported[0].ID != "c" {
|
||||
t.Fatalf("wrong abnormal items: %v %v", exported, err)
|
||||
}
|
||||
filter := PaymentReconciliationFilter{ProviderID: run.ProviderID, Status: "completed", From: run.BillDate, To: run.BillDate}
|
||||
runs, total, err := repo.AdminPaymentReconciliationRuns(filter, 1, 0)
|
||||
if err != nil || total != 1 || len(runs) != 1 {
|
||||
t.Fatalf("wrong summary: %v %v", runs, err)
|
||||
}
|
||||
summaries, err := repo.ExportPaymentReconciliations(filter, 10001)
|
||||
if err != nil || len(summaries) != 1 || summaries[0].ErrorItems != 1 {
|
||||
t.Fatalf("wrong export: %v %v", summaries, err)
|
||||
}
|
||||
filter.From = "2026-09-02"
|
||||
summaries, err = repo.ExportPaymentReconciliations(filter, 10001)
|
||||
if err != nil || len(summaries) != 0 {
|
||||
t.Fatalf("ignored date filter: %v %v", summaries, err)
|
||||
}
|
||||
_, started, err := repo.BeginPaymentReconciliation(&model.PaymentReconciliationRun{ID: "new-run", ProviderID: run.ProviderID, BillDate: run.BillDate})
|
||||
if err != nil || !started {
|
||||
t.Fatalf("rerun: %v %v", started, err)
|
||||
}
|
||||
_, _, err = repo.ExportPaymentReconciliationItems(run.ID, "", 10001)
|
||||
if !errors.Is(err, ErrPaymentReconciliationRunning) {
|
||||
t.Fatalf("running export accepted: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPaymentExportKeepsSnapshotWhenRerunCompletesBetweenReads(t *testing.T) {
|
||||
dsn := filepath.Join(t.TempDir(), "payment.db") + "?_journal_mode=WAL&_busy_timeout=5000"
|
||||
db, err := gorm.Open(sqlite.Open(dsn), &gorm.Config{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
writer, err := gorm.Open(sqlite.Open(dsn), &gorm.Config{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, connection := range []*gorm.DB{db, writer} {
|
||||
sqlDB, _ := connection.DB()
|
||||
t.Cleanup(func() { _ = sqlDB.Close() })
|
||||
}
|
||||
if err := db.AutoMigrate(&model.PaymentReconciliationRun{}, &model.PaymentReconciliationItem{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
run := model.PaymentReconciliationRun{ID: "run", ProviderID: "provider", BillDate: "2026-09-01", Status: model.PaymentReconciliationCompleted, TotalItems: 1, ErrorItems: 1}
|
||||
if err := db.Create(&run).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Create(&model.PaymentReconciliationItem{ID: "old", RunID: run.ID, Result: model.PaymentReconciliationAmountMismatch}).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
replaced := false
|
||||
if err := db.Callback().Query().After("gorm:query").Register("test:rerun-between-export-reads", func(tx *gorm.DB) {
|
||||
if replaced || tx.Statement.Table != "payment_reconciliation_runs" {
|
||||
return
|
||||
}
|
||||
replaced = true
|
||||
writerRepo := New(writer)
|
||||
_, started, err := writerRepo.BeginPaymentReconciliation(&model.PaymentReconciliationRun{ID: "unused", ProviderID: run.ProviderID, BillDate: run.BillDate})
|
||||
if err != nil || !started {
|
||||
t.Fatalf("could not rerun: %v", err)
|
||||
}
|
||||
if err := writerRepo.CompletePaymentReconciliation(run.ID, []model.PaymentReconciliationItem{{ID: "new", RunID: run.ID, Result: model.PaymentReconciliationRecovered, Resolved: true}}, 0, 1, 0); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
snapshot, items, err := New(db).ExportPaymentReconciliationItems(run.ID, "", 10001)
|
||||
if err != nil || !replaced || snapshot.ErrorItems != 1 || len(items) != 1 || items[0].ID != "old" {
|
||||
t.Fatalf("mixed snapshot: run=%+v items=%v err=%v", snapshot, items, err)
|
||||
}
|
||||
latest, newItems, err := New(writer).ExportPaymentReconciliationItems(run.ID, "", 10001)
|
||||
if err != nil || latest.RecoveredItems != 1 || len(newItems) != 1 || newItems[0].ID != "new" {
|
||||
t.Fatalf("rerun not committed: %+v %v %v", latest, newItems, err)
|
||||
}
|
||||
}
|
||||
@@ -395,7 +395,7 @@ func TestAdminPaymentOrdersSearchUserIdentity(t *testing.T) {
|
||||
{"merchant-c", "", 1}, {"trade-c", "", 1}, {"missing-user", "", 1}, {"", "", 3},
|
||||
} {
|
||||
t.Run(tc.keyword+"/"+tc.status, func(t *testing.T) {
|
||||
items, total, err := repo.AdminPaymentOrders(tc.status, tc.keyword, 1, 0)
|
||||
items, total, err := repo.AdminPaymentOrders(PaymentOrderFilter{Status: tc.status, Keyword: tc.keyword}, 1, 0)
|
||||
if err != nil || total != tc.total || len(items) > 1 {
|
||||
t.Fatalf("got items=%d total=%d err=%v; want total=%d", len(items), total, err, tc.total)
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user