diff --git a/backend/internal/handler/admin/affiliate_handler.go b/backend/internal/handler/admin/affiliate_handler.go index d443d344d23c..764a018c0299 100644 --- a/backend/internal/handler/admin/affiliate_handler.go +++ b/backend/internal/handler/admin/affiliate_handler.go @@ -214,11 +214,27 @@ func (h *AffiliateHandler) ListInviteRecords(c *gin.Context) { response.Paginated(c, items, total, filter.Page, filter.PageSize) } -// ListRebateRecords returns all order-level affiliate rebate records. +// ListRebateRecords returns affiliate rebate records from all supported recharge sources. // GET /api/v1/admin/affiliates/rebates func (h *AffiliateHandler) ListRebateRecords(c *gin.Context) { page, pageSize := response.ParsePagination(c) filter := parseAffiliateRecordFilter(c, page, pageSize) + sourceType := strings.TrimSpace(c.Query("source_type")) + if sourceType == "" { + // 兼容旧管理端：未传来源时继续只返回支付订单；新页面会显式传 all。 + sourceType = string(service.AffiliateRebateSourcePaymentOrder) + } + switch service.AffiliateRebateSourceType(sourceType) { + case service.AffiliateRebateSourceFilterAll, + service.AffiliateRebateSourcePaymentOrder, + service.AffiliateRebateSourceBalanceRedeem, + service.AffiliateRebateSourceAdminRecharge, + service.AffiliateRebateSourceLegacyUnknown: + filter.SourceType = sourceType + default: + response.BadRequest(c, "Invalid affiliate rebate source type") + return + } items, total, err := h.affiliateService.AdminListRebateRecords(c.Request.Context(), filter) if err != nil { response.ErrorFrom(c, err) diff --git a/backend/internal/handler/admin/affiliate_handler_test.go b/backend/internal/handler/admin/affiliate_handler_test.go new file mode 100644 index 000000000000..2f70b1956bf9 --- /dev/null +++ b/backend/internal/handler/admin/affiliate_handler_test.go @@ -0,0 +1,23 @@ +package admin + +import ( + "net/http" + "net/http/httptest" + "testing" + + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" +) + +func TestAffiliateHandlerListRebateRecordsRejectsInvalidSourceType(t *testing.T) { + gin.SetMode(gin.TestMode) + router := gin.New() + handler := &AffiliateHandler{} + router.GET("/api/v1/admin/affiliate/rebate-records", handler.ListRebateRecords) + + recorder := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodGet, "/api/v1/admin/affiliate/rebate-records?source_type=invalid", nil) + router.ServeHTTP(recorder, request) + + require.Equal(t, http.StatusBadRequest, recorder.Code) +} diff --git a/backend/internal/handler/auth_email_oauth_test.go b/backend/internal/handler/auth_email_oauth_test.go index 493b5099893d..7eec30b2e275 100644 --- a/backend/internal/handler/auth_email_oauth_test.go +++ b/backend/internal/handler/auth_email_oauth_test.go @@ -391,14 +391,10 @@ func (r *oauthEmailAffiliateRepoStub) BindInviter(_ context.Context, userID, inv return true, nil } -func (r *oauthEmailAffiliateRepoStub) AccrueQuota(context.Context, int64, int64, float64, int, *int64) (bool, error) { +func (r *oauthEmailAffiliateRepoStub) AccrueQuota(context.Context, service.AffiliateAccrualInput) (float64, error) { panic("unexpected AccrueQuota call") } -func (r *oauthEmailAffiliateRepoStub) GetAccruedRebateFromInvitee(context.Context, int64, int64) (float64, error) { - panic("unexpected GetAccruedRebateFromInvitee call") -} - func (r *oauthEmailAffiliateRepoStub) ThawFrozenQuota(context.Context, int64) (float64, error) { panic("unexpected ThawFrozenQuota call") } diff --git a/backend/internal/repository/affiliate_repo.go b/backend/internal/repository/affiliate_repo.go index 61da539e253c..62addba091cd 100644 --- a/backend/internal/repository/affiliate_repo.go +++ b/backend/internal/repository/affiliate_repo.go @@ -50,6 +50,13 @@ LEFT JOIN ( WHERE ua.user_id = $1 LIMIT 1` +// 历史支付流水已有订单编号时可以可靠识别；其余未分类旧流水统一展示为历史未知。 +const affiliateRebateEffectiveSourceSQL = `CASE + WHEN ual.source_order_id IS NOT NULL THEN 'payment_order' + WHEN ual.source_type IS NOT NULL THEN ual.source_type + ELSE 'legacy_unknown' +END` + type affiliateQueryExecer interface { QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error) ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) @@ -114,70 +121,137 @@ func (r *affiliateRepository) BindInviter(ctx context.Context, userID, inviterID return bound, nil } -func (r *affiliateRepository) AccrueQuota(ctx context.Context, inviterID, inviteeUserID int64, amount float64, freezeHours int, sourceOrderID *int64) (bool, error) { - if amount <= 0 { - return false, nil +func (r *affiliateRepository) AccrueQuota(ctx context.Context, input service.AffiliateAccrualInput) (float64, error) { + if input.Amount <= 0 { + return 0, nil + } + if err := input.Source.ValidateForAccrual(); err != nil { + return 0, err } - var applied bool + var appliedAmount float64 err := r.withTx(ctx, func(txCtx context.Context, txClient *dbent.Client) error { - // freezeHours > 0: add to frozen quota; == 0: add to available quota directly - var updateSQL string - if freezeHours > 0 { - updateSQL = "UPDATE user_affiliates SET aff_frozen_quota = aff_frozen_quota + $1, aff_history_quota = aff_history_quota + $1, updated_at = NOW() WHERE user_id = $2" - } else { - updateSQL = "UPDATE user_affiliates SET aff_quota = aff_quota + $1, aff_history_quota = aff_history_quota + $1, updated_at = NOW() WHERE user_id = $2" + var sourceValidationSQL string + var sourceID int64 + switch input.Source.Type { + case service.AffiliateRebateSourcePaymentOrder: + sourceValidationSQL = "SELECT COUNT(*) FROM payment_orders WHERE id = $1 AND user_id = $2" + sourceID = *input.Source.OrderID + case service.AffiliateRebateSourceBalanceRedeem: + sourceValidationSQL = "SELECT COUNT(*) FROM redeem_codes WHERE id = $1 AND used_by = $2 AND status = 'used' AND type = 'balance'" + sourceID = *input.Source.RedeemCodeID + case service.AffiliateRebateSourceAdminRecharge: + sourceValidationSQL = "SELECT COUNT(*) FROM redeem_codes WHERE id = $1 AND used_by = $2 AND status = 'used' AND type = 'admin_balance'" + sourceID = *input.Source.RedeemCodeID } - res, err := txClient.ExecContext(txCtx, updateSQL, amount, inviterID) + sourceCount, err := scanInt64(txCtx, txClient, sourceValidationSQL, sourceID, input.InviteeUserID) if err != nil { - return err + return fmt.Errorf("validate affiliate rebate source: %w", err) } - affected, _ := res.RowsAffected() - if affected == 0 { - applied = false - return nil + if sourceCount != 1 { + return service.ErrAffiliateRebateSource } - if freezeHours > 0 { - if _, err = txClient.ExecContext(txCtx, ` -INSERT INTO user_affiliate_ledger (user_id, action, amount, source_user_id, source_order_id, frozen_until, created_at, updated_at) -VALUES ($1, 'accrue', $2, $3, $4, NOW() + make_interval(hours => $5), NOW(), NOW())`, - inviterID, amount, inviteeUserID, nullableInt64Arg(sourceOrderID), freezeHours); err != nil { - return fmt.Errorf("insert affiliate accrue ledger: %w", err) - } - } else { - if _, err = txClient.ExecContext(txCtx, ` -INSERT INTO user_affiliate_ledger (user_id, action, amount, source_user_id, source_order_id, created_at, updated_at) -VALUES ($1, 'accrue', $2, $3, $4, NOW(), NOW())`, inviterID, amount, inviteeUserID, nullableInt64Arg(sourceOrderID)); err != nil { - return fmt.Errorf("insert affiliate accrue ledger: %w", err) + // 锁住邀请人的返利账户后再计算单人上限，避免并发充值同时越过上限。 + lockRows, err := txClient.QueryContext(txCtx, ` +SELECT user_id +FROM user_affiliates +WHERE user_id = $1 +FOR UPDATE`, input.InviterID) + if err != nil { + return fmt.Errorf("lock inviter affiliate account: %w", err) + } + if !lockRows.Next() { + rowsErr := lockRows.Err() + _ = lockRows.Close() + if rowsErr != nil { + return rowsErr } + return service.ErrAffiliateProfileNotFound + } + if err := lockRows.Close(); err != nil { + return err + } + + quotaColumn := "aff_quota" + if input.FreezeHours > 0 { + quotaColumn = "aff_frozen_quota" } - applied = true + // 上限、已有返利和本次实际入账金额全部在 PostgreSQL NUMERIC 中计算。 + // 流水插入与额度更新共用同一个 CTE；来源唯一冲突或额度耗尽时不会更新额度。 + accrueRows, err := txClient.QueryContext(txCtx, fmt.Sprintf(` +WITH existing AS ( + SELECT COALESCE(SUM(amount), 0::numeric) AS amount + FROM user_affiliate_ledger + WHERE user_id = $1 + AND source_user_id = $3 + AND action = 'accrue' +), normalized AS ( + SELECT GREATEST(TRUNC($2::numeric, 8), 0::numeric) AS requested_amount, + TRUNC($9::numeric, 8) AS per_invitee_cap, + existing.amount AS existing_amount + FROM existing +), candidate AS ( + SELECT CASE + WHEN per_invitee_cap <= 0 THEN requested_amount + ELSE LEAST(requested_amount, GREATEST(per_invitee_cap - existing_amount, 0::numeric)) + END::numeric(20,8) AS amount + FROM normalized +), inserted AS ( + INSERT INTO user_affiliate_ledger ( + user_id, action, amount, source_user_id, source_type, base_amount, + source_order_id, source_redeem_code_id, frozen_until, created_at, updated_at + ) + SELECT $1, 'accrue', candidate.amount, $3, $4, TRUNC($5::numeric, 8), $6, $7, + CASE WHEN $8::integer > 0 THEN NOW() + make_interval(hours => $8::integer) ELSE NULL END, + NOW(), NOW() + FROM candidate + WHERE candidate.amount > 0 + ON CONFLICT DO NOTHING + RETURNING amount +) +UPDATE user_affiliates AS ua +SET %s = ua.%s + inserted.amount, + aff_history_quota = ua.aff_history_quota + inserted.amount, + updated_at = NOW() +FROM inserted +WHERE ua.user_id = $1 +RETURNING inserted.amount::double precision`, quotaColumn, quotaColumn), + input.InviterID, + input.Amount, + input.InviteeUserID, + string(input.Source.Type), + input.Source.BaseAmount, + nullableInt64Arg(input.Source.OrderID), + nullableInt64Arg(input.Source.RedeemCodeID), + input.FreezeHours, + input.PerInviteeCap, + ) + if err != nil { + return fmt.Errorf("accrue affiliate quota: %w", err) + } + if !accrueRows.Next() { + rowsErr := accrueRows.Err() + _ = accrueRows.Close() + if rowsErr != nil { + return rowsErr + } + return nil + } + if err := accrueRows.Scan(&appliedAmount); err != nil { + _ = accrueRows.Close() + return err + } + if err := accrueRows.Close(); err != nil { + return err + } return nil }) if err != nil { - return false, err - } - return applied, nil -} - -func (r *affiliateRepository) GetAccruedRebateFromInvitee(ctx context.Context, inviterID, inviteeUserID int64) (float64, error) { - client := clientFromContext(ctx, r.client) - rows, err := client.QueryContext(ctx, - `SELECT COALESCE(SUM(amount), 0)::double precision FROM user_affiliate_ledger WHERE user_id = $1 AND source_user_id = $2 AND action = 'accrue'`, - inviterID, inviteeUserID) - if err != nil { - return 0, fmt.Errorf("query accrued rebate from invitee: %w", err) - } - defer func() { _ = rows.Close() }() - var total float64 - if rows.Next() { - if err := rows.Scan(&total); err != nil { - return 0, err - } + return 0, err } - return total, rows.Close() + return appliedAmount, nil } func (r *affiliateRepository) ThawFrozenQuota(ctx context.Context, userID int64) (float64, error) { @@ -464,18 +538,23 @@ func (r *affiliateRepository) ListAffiliateRebateRecords(ctx context.Context, fi client := clientFromContext(ctx, r.client) where, args := buildAffiliateRecordWhere(filter, "ual.created_at", []string{ "inviter.email", "inviter.username", "invitee.email", "invitee.username", - "po.id::text", "po.out_trade_no", "po.payment_type", "po.status", + "ual.id::text", affiliateRebateEffectiveSourceSQL, "po.id::text", "po.out_trade_no", + "po.payment_type", "po.status", "rc.id::text", "rc.code", }) baseJoin := ` FROM user_affiliate_ledger ual -JOIN payment_orders po ON po.id = ual.source_order_id +LEFT JOIN payment_orders po ON po.id = ual.source_order_id +LEFT JOIN redeem_codes rc ON rc.id = ual.source_redeem_code_id JOIN users invitee ON invitee.id = ual.source_user_id JOIN users inviter ON inviter.id = ual.user_id -WHERE ual.action = 'accrue' - AND ual.source_order_id IS NOT NULL` +WHERE ual.action = 'accrue'` if where != "" { where = strings.Replace(where, "WHERE ", " AND ", 1) } + if sourceType := strings.TrimSpace(filter.SourceType); sourceType != "" && sourceType != string(service.AffiliateRebateSourceFilterAll) { + args = append(args, sourceType) + where += fmt.Sprintf(" AND (%s) = $%d", affiliateRebateEffectiveSourceSQL, len(args)) + } total, err := queryAffiliateRecordCount(ctx, client, "SELECT COUNT(*) "+baseJoin+where, args...) if err != nil { @@ -483,27 +562,40 @@ WHERE ual.action = 'accrue' } orderBy := buildAffiliateRecordOrderBy(filter, map[string]string{ - "order": "po.id", - "inviter": "inviter.email", - "invitee": "invitee.email", - "order_amount": "po.amount", - "pay_amount": "po.pay_amount", - "rebate_amount": "ual.amount", - "payment_type": "po.payment_type", - "order_status": "po.status", - "created_at": "ual.created_at", - }, "ual.created_at") + "source": affiliateRebateEffectiveSourceSQL, + "source_reference": "COALESCE(po.id, rc.id)", + "order": "po.id", + "inviter": "inviter.email", + "invitee": "invitee.email", + "base_amount": "COALESCE(ual.base_amount, po.amount)", + "order_amount": "po.amount", + "pay_amount": "po.pay_amount", + "rebate_amount": "ual.amount", + "payment_type": "po.payment_type", + "order_status": "po.status", + "created_at": "ual.created_at", + }, "ual.created_at") + ", ual.id DESC" args = append(args, filter.PageSize, (filter.Page-1)*filter.PageSize) rows, err := client.QueryContext(ctx, ` -SELECT po.id, - po.out_trade_no, - ual.user_id, + SELECT ual.id, + `+affiliateRebateEffectiveSourceSQL+`, + ual.source_order_id, + po.out_trade_no, + ual.source_redeem_code_id, + CASE + WHEN rc.code IS NULL THEN NULL + WHEN LENGTH(rc.code) <= 4 THEN '****' + WHEN LENGTH(rc.code) <= 8 THEN LEFT(rc.code, 1) || '****' || RIGHT(rc.code, 1) + ELSE LEFT(rc.code, 4) || '****' || RIGHT(rc.code, 4) + END, + ual.user_id, COALESCE(inviter.email, ''), COALESCE(inviter.username, ''), ual.source_user_id, - COALESCE(invitee.email, ''), - COALESCE(invitee.username, ''), - po.amount::double precision, + COALESCE(invitee.email, ''), + COALESCE(invitee.username, ''), + COALESCE(ual.base_amount, po.amount)::double precision, + po.amount::double precision, po.pay_amount::double precision, ual.amount::double precision, po.payment_type, @@ -521,14 +613,19 @@ LIMIT $`+fmt.Sprint(len(args)-1)+` OFFSET $`+fmt.Sprint(len(args)), args...) for rows.Next() { var item service.AffiliateRebateRecord if err := rows.Scan( + &item.LedgerID, + &item.SourceType, &item.OrderID, &item.OutTradeNo, + &item.RedeemCodeID, + &item.RedeemCodeMasked, &item.InviterID, &item.InviterEmail, &item.InviterUsername, &item.InviteeID, &item.InviteeEmail, &item.InviteeUsername, + &item.BaseAmount, &item.OrderAmount, &item.PayAmount, &item.RebateAmount, diff --git a/backend/internal/repository/affiliate_repo_integration_test.go b/backend/internal/repository/affiliate_repo_integration_test.go index b01ed528a653..f4a900f7cb93 100644 --- a/backend/internal/repository/affiliate_repo_integration_test.go +++ b/backend/internal/repository/affiliate_repo_integration_test.go @@ -4,15 +4,59 @@ package repository import ( "context" + "errors" "fmt" "testing" "time" dbent "github.com/Wei-Shaw/sub2api/ent" "github.com/Wei-Shaw/sub2api/internal/service" + migrationspkg "github.com/Wei-Shaw/sub2api/migrations" "github.com/stretchr/testify/require" ) +type failingAccrueAffiliateRepository struct { + service.AffiliateRepository + err error +} + +func (r *failingAccrueAffiliateRepository) AccrueQuota(context.Context, service.AffiliateAccrualInput) (float64, error) { + return 0, r.err +} + +func newAdminServiceForAffiliateIntegration( + userRepo service.UserRepository, + redeemRepo service.RedeemCodeRepository, + client *dbent.Client, + settingService *service.SettingService, + affiliateService *service.AffiliateService, +) service.AdminService { + return service.NewAdminService( + userRepo, + nil, + nil, + nil, + nil, + redeemRepo, + nil, + nil, + nil, + nil, + nil, + nil, + client, + settingService, + nil, + nil, + nil, + nil, + affiliateService, + nil, + nil, + nil, + ) +} + func querySingleFloat(t *testing.T, ctx context.Context, client *dbent.Client, query string, args ...any) float64 { t.Helper() rows, err := client.QueryContext(ctx, query, args...) @@ -39,6 +83,53 @@ func querySingleInt(t *testing.T, ctx context.Context, client *dbent.Client, que return value } +func registerCommittedAffiliateTestCleanup(t *testing.T, userIDs, redeemCodeIDs *[]int64) { + t.Helper() + t.Cleanup(func() { + ctx := context.Background() + tx, err := integrationDB.BeginTx(ctx, nil) + require.NoError(t, err) + defer func() { _ = tx.Rollback() }() + + // 这些用例必须提交真实多事务数据；按已登记主键清理，避免污染同包后续查询。 + for _, redeemCodeID := range *redeemCodeIDs { + _, err = tx.ExecContext(ctx, "DELETE FROM user_affiliate_ledger WHERE source_redeem_code_id = $1", redeemCodeID) + require.NoError(t, err) + } + for _, userID := range *userIDs { + _, err = tx.ExecContext(ctx, "DELETE FROM user_affiliate_ledger WHERE user_id = $1 OR source_user_id = $1", userID) + require.NoError(t, err) + _, err = tx.ExecContext(ctx, "DELETE FROM redeem_codes WHERE used_by = $1", userID) + require.NoError(t, err) + } + for _, redeemCodeID := range *redeemCodeIDs { + _, err = tx.ExecContext(ctx, "DELETE FROM redeem_codes WHERE id = $1", redeemCodeID) + require.NoError(t, err) + } + for _, userID := range *userIDs { + _, err = tx.ExecContext(ctx, "DELETE FROM user_affiliates WHERE user_id = $1 OR inviter_id = $1", userID) + require.NoError(t, err) + _, err = tx.ExecContext(ctx, "DELETE FROM users WHERE id = $1", userID) + require.NoError(t, err) + } + require.NoError(t, tx.Commit()) + }) +} + +func registerIntegrationSettingRestore(t *testing.T, ctx context.Context, settingRepo service.SettingRepository, keys []string) { + t.Helper() + previousSettings, err := settingRepo.GetMultiple(ctx, keys) + require.NoError(t, err) + t.Cleanup(func() { + require.NoError(t, settingRepo.SetMultiple(ctx, previousSettings)) + for _, key := range keys { + if _, existed := previousSettings[key]; !existed { + require.NoError(t, settingRepo.Delete(ctx, key)) + } + } + }) +} + func TestAffiliateRepository_TransferQuotaToBalance_UsesClaimedQuotaBeforeClear(t *testing.T) { ctx := context.Background() tx := testEntTx(t) @@ -144,10 +235,20 @@ func TestAffiliateRepository_AccrueQuota_ReusesOuterTransaction(t *testing.T) { bound, err := repo.BindInviter(txCtx, invitee.ID, inviter.ID) require.NoError(t, err) require.True(t, bound, "invitee must bind to inviter") - - applied, err := repo.AccrueQuota(txCtx, inviter.ID, invitee.ID, 3.5, 0, nil) + adminCodeID := insertHistoricalRedeemCode(t, txCtx, client, invitee.ID, "admin_balance", 17.5, time.Now()) + + applied, err := repo.AccrueQuota(txCtx, service.AffiliateAccrualInput{ + InviterID: inviter.ID, + InviteeUserID: invitee.ID, + Amount: 3.5, + Source: service.AffiliateRebateSource{ + Type: service.AffiliateRebateSourceAdminRecharge, + BaseAmount: 17.5, + RedeemCodeID: &adminCodeID, + }, + }) require.NoError(t, err) - require.True(t, applied, "AccrueQuota must report applied=true") + require.InDelta(t, 3.5, applied, 1e-9, "AccrueQuota must return the applied amount") // Visible inside the outer tx. innerQuota := querySingleFloat(t, txCtx, client, @@ -170,6 +271,715 @@ func TestAffiliateRepository_AccrueQuota_ReusesOuterTransaction(t *testing.T) { "AccrueQuota must propagate the outer tx — found persisted rows after rollback") } +func TestAffiliateRepository_AccrueQuota_RedeemSourceIsIdempotentAndListable(t *testing.T) { + ctx := context.Background() + tx := testEntTx(t) + txCtx := dbent.NewTxContext(ctx, tx) + client := tx.Client() + repo := NewAffiliateRepository(client, integrationDB) + + inviter := mustCreateUser(t, client, &service.User{ + Email: fmt.Sprintf("affiliate-redeem-inviter-%d@example.com", time.Now().UnixNano()), + PasswordHash: "hash", + Role: service.RoleUser, + Status: service.StatusActive, + }) + invitee := mustCreateUser(t, client, &service.User{ + Email: fmt.Sprintf("affiliate-redeem-invitee-%d@example.com", time.Now().UnixNano()+1), + PasswordHash: "hash", + Role: service.RoleUser, + Status: service.StatusActive, + }) + _, err := repo.EnsureUserAffiliate(txCtx, inviter.ID) + require.NoError(t, err) + _, err = repo.EnsureUserAffiliate(txCtx, invitee.ID) + require.NoError(t, err) + bound, err := repo.BindInviter(txCtx, invitee.ID, inviter.ID) + require.NoError(t, err) + require.True(t, bound) + + shortRedeemCode := fmt.Sprintf("%04X", time.Now().UnixNano()&0xffff) + rows, err := client.QueryContext(txCtx, ` +INSERT INTO redeem_codes (code, type, value, status, used_by, used_at, created_at) +VALUES ($1, 'balance', 50, 'used', $2, NOW(), NOW()) +RETURNING id`, shortRedeemCode, invitee.ID) + require.NoError(t, err) + require.True(t, rows.Next()) + var redeemCodeID int64 + require.NoError(t, rows.Scan(&redeemCodeID)) + require.NoError(t, rows.Close()) + + input := service.AffiliateAccrualInput{ + InviterID: inviter.ID, + InviteeUserID: invitee.ID, + Amount: 10, + Source: service.AffiliateRebateSource{ + Type: service.AffiliateRebateSourceBalanceRedeem, + BaseAmount: 50, + RedeemCodeID: &redeemCodeID, + }, + } + wrongInviteeInput := input + wrongInviteeInput.InviteeUserID = inviter.ID + _, err = repo.AccrueQuota(txCtx, wrongInviteeInput) + require.ErrorIs(t, err, service.ErrAffiliateRebateSource) + wrongTypeInput := input + wrongTypeInput.Source.Type = service.AffiliateRebateSourceAdminRecharge + _, err = repo.AccrueQuota(txCtx, wrongTypeInput) + require.ErrorIs(t, err, service.ErrAffiliateRebateSource) + + first, err := repo.AccrueQuota(txCtx, input) + require.NoError(t, err) + require.InDelta(t, 10, first, 1e-9) + second, err := repo.AccrueQuota(txCtx, input) + require.NoError(t, err) + require.Zero(t, second) + + quota := querySingleFloat(t, txCtx, client, + "SELECT aff_quota::double precision FROM user_affiliates WHERE user_id = $1", inviter.ID) + require.InDelta(t, 10, quota, 1e-9) + ledgerCount := querySingleInt(t, txCtx, client, + "SELECT COUNT(*) FROM user_affiliate_ledger WHERE source_redeem_code_id = $1", redeemCodeID) + require.Equal(t, 1, ledgerCount) + for range 2 { + additionalCodeID := insertHistoricalRedeemCode(t, txCtx, client, invitee.ID, "balance", 50, time.Now()) + _, err = repo.AccrueQuota(txCtx, service.AffiliateAccrualInput{ + InviterID: inviter.ID, + InviteeUserID: invitee.ID, + Amount: 10, + Source: service.AffiliateRebateSource{ + Type: service.AffiliateRebateSourceBalanceRedeem, + BaseAmount: 50, + RedeemCodeID: &additionalCodeID, + }, + }) + require.NoError(t, err) + } + + items, total, err := repo.ListAffiliateRebateRecords(txCtx, service.AffiliateRecordFilter{ + SourceType: string(service.AffiliateRebateSourceBalanceRedeem), + Page: 1, + PageSize: 20, + }) + require.NoError(t, err) + require.Equal(t, int64(3), total) + require.Len(t, items, 3) + var originalRecord *service.AffiliateRebateRecord + for i := range items { + if items[i].RedeemCodeID != nil && *items[i].RedeemCodeID == redeemCodeID { + originalRecord = &items[i] + break + } + } + require.NotNil(t, originalRecord) + require.Equal(t, string(service.AffiliateRebateSourceBalanceRedeem), originalRecord.SourceType) + require.Equal(t, "****", *originalRecord.RedeemCodeMasked) + require.NotContains(t, *originalRecord.RedeemCodeMasked, shortRedeemCode) + require.InDelta(t, 50, *originalRecord.BaseAmount, 1e-9) + require.InDelta(t, 10, originalRecord.RebateAmount, 1e-9) + + pageFilter := service.AffiliateRecordFilter{ + SourceType: string(service.AffiliateRebateSourceBalanceRedeem), + Page: 1, + PageSize: 2, + SortBy: "source", + } + firstPage, _, err := repo.ListAffiliateRebateRecords(txCtx, pageFilter) + require.NoError(t, err) + pageFilter.Page = 2 + secondPage, _, err := repo.ListAffiliateRebateRecords(txCtx, pageFilter) + require.NoError(t, err) + require.Len(t, firstPage, 2) + require.Len(t, secondPage, 1) + seenLedgerIDs := map[int64]struct{}{} + for _, item := range append(firstPage, secondPage...) { + seenLedgerIDs[item.LedgerID] = struct{}{} + } + require.Len(t, seenLedgerIDs, 3, "来源字段并列时跨页记录不能重复或遗漏") + pageFilter.Page = 1 + repeatedFirstPage, _, err := repo.ListAffiliateRebateRecords(txCtx, pageFilter) + require.NoError(t, err) + require.Equal(t, []int64{firstPage[0].LedgerID, firstPage[1].LedgerID}, []int64{repeatedFirstPage[0].LedgerID, repeatedFirstPage[1].LedgerID}) +} + +func TestAffiliateRepository_AccrueQuota_ConcurrentSameRedeemSourceIsIdempotent(t *testing.T) { + ctx := context.Background() + repo := NewAffiliateRepository(integrationEntClient, integrationDB) + var userIDs, redeemCodeIDs []int64 + registerCommittedAffiliateTestCleanup(t, &userIDs, &redeemCodeIDs) + + inviter := mustCreateUser(t, integrationEntClient, &service.User{ + Email: fmt.Sprintf("affiliate-source-race-inviter-%d@example.com", time.Now().UnixNano()), + PasswordHash: "hash", + Role: service.RoleUser, + Status: service.StatusActive, + }) + userIDs = append(userIDs, inviter.ID) + invitee := mustCreateUser(t, integrationEntClient, &service.User{ + Email: fmt.Sprintf("affiliate-source-race-invitee-%d@example.com", time.Now().UnixNano()+1), + PasswordHash: "hash", + Role: service.RoleUser, + Status: service.StatusActive, + }) + userIDs = append(userIDs, invitee.ID) + _, err := repo.EnsureUserAffiliate(ctx, inviter.ID) + require.NoError(t, err) + _, err = repo.EnsureUserAffiliate(ctx, invitee.ID) + require.NoError(t, err) + + rows, err := integrationEntClient.QueryContext(ctx, ` +INSERT INTO redeem_codes (code, type, value, status, used_by, used_at, created_at) +VALUES ($1, 'balance', 50, 'used', $2, NOW(), NOW()) +RETURNING id`, fmt.Sprintf("RACE%d", time.Now().UnixNano()), invitee.ID) + require.NoError(t, err) + require.True(t, rows.Next()) + var redeemCodeID int64 + require.NoError(t, rows.Scan(&redeemCodeID)) + require.NoError(t, rows.Close()) + redeemCodeIDs = append(redeemCodeIDs, redeemCodeID) + + start := make(chan struct{}) + results := make(chan struct { + amount float64 + err error + }, 2) + for range 2 { + go func() { + <-start + amount, accrueErr := repo.AccrueQuota(ctx, service.AffiliateAccrualInput{ + InviterID: inviter.ID, + InviteeUserID: invitee.ID, + Amount: 10, + Source: service.AffiliateRebateSource{ + Type: service.AffiliateRebateSourceBalanceRedeem, + BaseAmount: 50, + RedeemCodeID: &redeemCodeID, + }, + }) + results <- struct { + amount float64 + err error + }{amount: amount, err: accrueErr} + }() + } + close(start) + + var appliedTotal float64 + for range 2 { + result := <-results + require.NoError(t, result.err) + appliedTotal += result.amount + } + require.InDelta(t, 10, appliedTotal, 1e-9) + ledgerCount := querySingleInt(t, ctx, integrationEntClient, + "SELECT COUNT(*) FROM user_affiliate_ledger WHERE source_redeem_code_id = $1", redeemCodeID) + require.Equal(t, 1, ledgerCount) + quota := querySingleFloat(t, ctx, integrationEntClient, + "SELECT aff_quota::double precision FROM user_affiliates WHERE user_id = $1", inviter.ID) + require.InDelta(t, 10, quota, 1e-9) +} + +func TestAffiliateRepository_AccrueQuota_TruncatesAtPerInviteeCap(t *testing.T) { + ctx := context.Background() + tx := testEntTx(t) + txCtx := dbent.NewTxContext(ctx, tx) + client := tx.Client() + repo := NewAffiliateRepository(client, integrationDB) + + inviter := mustCreateUser(t, client, &service.User{ + Email: fmt.Sprintf("affiliate-cap-inviter-%d@example.com", time.Now().UnixNano()), + PasswordHash: "hash", + Role: service.RoleUser, + Status: service.StatusActive, + }) + invitee := mustCreateUser(t, client, &service.User{ + Email: fmt.Sprintf("affiliate-cap-invitee-%d@example.com", time.Now().UnixNano()+1), + PasswordHash: "hash", + Role: service.RoleUser, + Status: service.StatusActive, + }) + _, err := repo.EnsureUserAffiliate(txCtx, inviter.ID) + require.NoError(t, err) + _, err = repo.EnsureUserAffiliate(txCtx, invitee.ID) + require.NoError(t, err) + + adminCodeIDs := []int64{ + insertHistoricalRedeemCode(t, txCtx, client, invitee.ID, "admin_balance", 20, time.Now()), + insertHistoricalRedeemCode(t, txCtx, client, invitee.ID, "admin_balance", 20, time.Now()), + insertHistoricalRedeemCode(t, txCtx, client, invitee.ID, "admin_balance", 20, time.Now()), + } + inputForCode := func(codeID int64) service.AffiliateAccrualInput { + return service.AffiliateAccrualInput{ + InviterID: inviter.ID, + InviteeUserID: invitee.ID, + Amount: 4.12345678, + PerInviteeCap: 5.000000009, + Source: service.AffiliateRebateSource{ + Type: service.AffiliateRebateSourceAdminRecharge, + BaseAmount: 20, + RedeemCodeID: &codeID, + }, + } + } + first, err := repo.AccrueQuota(txCtx, inputForCode(adminCodeIDs[0])) + require.NoError(t, err) + require.InDelta(t, 4.12345678, first, 1e-9) + second, err := repo.AccrueQuota(txCtx, inputForCode(adminCodeIDs[1])) + require.NoError(t, err) + require.InDelta(t, 0.87654322, second, 1e-9) + third, err := repo.AccrueQuota(txCtx, inputForCode(adminCodeIDs[2])) + require.NoError(t, err) + require.Zero(t, third) + + quota := querySingleFloat(t, txCtx, client, + "SELECT aff_quota::double precision FROM user_affiliates WHERE user_id = $1", inviter.ID) + require.InDelta(t, 5, quota, 1e-9) +} + +func TestAffiliateRepository_AccrueQuota_ConcurrentRequestsRespectPerInviteeCap(t *testing.T) { + ctx := context.Background() + repo := NewAffiliateRepository(integrationEntClient, integrationDB) + var userIDs, redeemCodeIDs []int64 + registerCommittedAffiliateTestCleanup(t, &userIDs, &redeemCodeIDs) + + inviter := mustCreateUser(t, integrationEntClient, &service.User{ + Email: fmt.Sprintf("affiliate-concurrent-cap-inviter-%d@example.com", time.Now().UnixNano()), + PasswordHash: "hash", + Role: service.RoleUser, + Status: service.StatusActive, + }) + userIDs = append(userIDs, inviter.ID) + invitee := mustCreateUser(t, integrationEntClient, &service.User{ + Email: fmt.Sprintf("affiliate-concurrent-cap-invitee-%d@example.com", time.Now().UnixNano()+1), + PasswordHash: "hash", + Role: service.RoleUser, + Status: service.StatusActive, + }) + userIDs = append(userIDs, invitee.ID) + _, err := repo.EnsureUserAffiliate(ctx, inviter.ID) + require.NoError(t, err) + _, err = repo.EnsureUserAffiliate(ctx, invitee.ID) + require.NoError(t, err) + adminCodeIDs := []int64{ + insertHistoricalRedeemCode(t, ctx, integrationEntClient, invitee.ID, "admin_balance", 20, time.Now()), + insertHistoricalRedeemCode(t, ctx, integrationEntClient, invitee.ID, "admin_balance", 20, time.Now()), + } + redeemCodeIDs = append(redeemCodeIDs, adminCodeIDs...) + + type result struct { + amount float64 + err error + } + start := make(chan struct{}) + results := make(chan result, 2) + for i := range 2 { + go func() { + <-start + adminCodeID := adminCodeIDs[i] + amount, accrueErr := repo.AccrueQuota(ctx, service.AffiliateAccrualInput{ + InviterID: inviter.ID, + InviteeUserID: invitee.ID, + Amount: 4, + PerInviteeCap: 5, + Source: service.AffiliateRebateSource{ + Type: service.AffiliateRebateSourceAdminRecharge, + BaseAmount: 20, + RedeemCodeID: &adminCodeID, + }, + }) + results <- result{amount: amount, err: accrueErr} + }() + } + close(start) + + var appliedTotal float64 + for range 2 { + result := <-results + require.NoError(t, result.err) + appliedTotal += result.amount + } + require.InDelta(t, 5, appliedTotal, 1e-9) + + quota := querySingleFloat(t, ctx, integrationEntClient, + "SELECT aff_quota::double precision FROM user_affiliates WHERE user_id = $1", inviter.ID) + require.InDelta(t, 5, quota, 1e-9) + ledgerTotal := querySingleFloat(t, ctx, integrationEntClient, + "SELECT COALESCE(SUM(amount), 0)::double precision FROM user_affiliate_ledger WHERE user_id = $1 AND source_user_id = $2 AND action = 'accrue'", + inviter.ID, invitee.ID) + require.InDelta(t, 5, ledgerTotal, 1e-9) +} + +func TestAdminService_UpdateUserBalance_AdminRechargeCommitsOrRollsBackAsOneTransaction(t *testing.T) { + ctx := context.Background() + userRepo := NewUserRepository(integrationEntClient, integrationDB) + redeemRepo := NewRedeemCodeRepository(integrationEntClient) + affiliateRepo := NewAffiliateRepository(integrationEntClient, integrationDB) + settingRepo := NewSettingRepository(integrationEntClient) + settingKeys := []string{ + service.SettingKeyAffiliateEnabled, + service.SettingKeyAffiliateAdminRechargeEnabled, + service.SettingKeyAffiliateRebateRate, + } + registerIntegrationSettingRestore(t, ctx, settingRepo, settingKeys) + require.NoError(t, settingRepo.SetMultiple(ctx, map[string]string{ + service.SettingKeyAffiliateEnabled: "true", + service.SettingKeyAffiliateAdminRechargeEnabled: "true", + service.SettingKeyAffiliateRebateRate: "20", + })) + settingService := service.NewSettingService(settingRepo, nil) + var userIDs, redeemCodeIDs []int64 + registerCommittedAffiliateTestCleanup(t, &userIDs, &redeemCodeIDs) + + inviter := mustCreateUser(t, integrationEntClient, &service.User{ + Email: fmt.Sprintf("admin-recharge-rollback-inviter-%d@example.com", time.Now().UnixNano()), + PasswordHash: "hash", + Role: service.RoleUser, + Status: service.StatusActive, + }) + userIDs = append(userIDs, inviter.ID) + invitee := mustCreateUser(t, integrationEntClient, &service.User{ + Email: fmt.Sprintf("admin-recharge-rollback-invitee-%d@example.com", time.Now().UnixNano()+1), + PasswordHash: "hash", + Role: service.RoleUser, + Status: service.StatusActive, + Balance: 10, + }) + userIDs = append(userIDs, invitee.ID) + _, err := affiliateRepo.EnsureUserAffiliate(ctx, inviter.ID) + require.NoError(t, err) + _, err = affiliateRepo.EnsureUserAffiliate(ctx, invitee.ID) + require.NoError(t, err) + bound, err := affiliateRepo.BindInviter(ctx, invitee.ID, inviter.ID) + require.NoError(t, err) + require.True(t, bound) + + successfulAffiliateService := service.NewAffiliateService(affiliateRepo, settingService, nil, nil) + successfulAdminService := newAdminServiceForAffiliateIntegration( + userRepo, redeemRepo, integrationEntClient, settingService, successfulAffiliateService, + ) + updated, err := successfulAdminService.UpdateUserBalance(ctx, invitee.ID, 5, "add", "commit-test") + require.NoError(t, err) + require.InDelta(t, 15, updated.Balance, 1e-9) + + rows, err := integrationEntClient.QueryContext(ctx, ` +SELECT rc.id, + ual.source_type, + ual.base_amount::double precision, + ual.amount::double precision +FROM redeem_codes rc +JOIN user_affiliate_ledger ual ON ual.source_redeem_code_id = rc.id +WHERE rc.used_by = $1 AND rc.type = 'admin_balance'`, invitee.ID) + require.NoError(t, err) + require.True(t, rows.Next()) + var adjustmentRecordID int64 + var sourceType string + var baseAmount, rebateAmount float64 + require.NoError(t, rows.Scan(&adjustmentRecordID, &sourceType, &baseAmount, &rebateAmount)) + require.NoError(t, rows.Close()) + redeemCodeIDs = append(redeemCodeIDs, adjustmentRecordID) + require.Positive(t, adjustmentRecordID) + require.Equal(t, string(service.AffiliateRebateSourceAdminRecharge), sourceType) + require.InDelta(t, 5, baseAmount, 1e-9) + require.InDelta(t, 1, rebateAmount, 1e-9) + + failingAffiliateService := service.NewAffiliateService(&failingAccrueAffiliateRepository{ + AffiliateRepository: affiliateRepo, + err: errors.New("forced affiliate failure"), + }, settingService, nil, nil) + failingAdminService := newAdminServiceForAffiliateIntegration( + userRepo, redeemRepo, integrationEntClient, settingService, failingAffiliateService, + ) + + updated, err = failingAdminService.UpdateUserBalance(ctx, invitee.ID, 5, "add", "rollback-test") + require.Nil(t, updated) + require.ErrorContains(t, err, "forced affiliate failure") + + balance := querySingleFloat(t, ctx, integrationEntClient, + "SELECT balance::double precision FROM users WHERE id = $1", invitee.ID) + require.InDelta(t, 15, balance, 1e-9) + adjustmentCount := querySingleInt(t, ctx, integrationEntClient, + "SELECT COUNT(*) FROM redeem_codes WHERE used_by = $1 AND type = 'admin_balance'", invitee.ID) + require.Equal(t, 1, adjustmentCount) + ledgerCount := querySingleInt(t, ctx, integrationEntClient, + "SELECT COUNT(*) FROM user_affiliate_ledger WHERE user_id = $1 AND source_user_id = $2 AND action = 'accrue'", + inviter.ID, invitee.ID) + require.Equal(t, 1, ledgerCount) +} + +func TestRedeemService_BalanceAndAffiliateRebateShareTransaction(t *testing.T) { + ctx := context.Background() + userRepo := NewUserRepository(integrationEntClient, integrationDB) + redeemRepo := NewRedeemCodeRepository(integrationEntClient) + affiliateRepo := NewAffiliateRepository(integrationEntClient, integrationDB) + settingRepo := NewSettingRepository(integrationEntClient) + settingKeys := []string{ + service.SettingKeyAffiliateEnabled, + service.SettingKeyAffiliateRebateRate, + service.SettingKeyAffiliateRebateFreezeHours, + service.SettingKeyAffiliateRebateDurationDays, + service.SettingKeyAffiliateRebatePerInviteeCap, + } + registerIntegrationSettingRestore(t, ctx, settingRepo, settingKeys) + require.NoError(t, settingRepo.SetMultiple(ctx, map[string]string{ + service.SettingKeyAffiliateEnabled: "true", + service.SettingKeyAffiliateRebateRate: "20", + service.SettingKeyAffiliateRebateFreezeHours: "0", + service.SettingKeyAffiliateRebateDurationDays: "0", + service.SettingKeyAffiliateRebatePerInviteeCap: "0", + })) + settingService := service.NewSettingService(settingRepo, nil) + var userIDs, redeemCodeIDs []int64 + registerCommittedAffiliateTestCleanup(t, &userIDs, &redeemCodeIDs) + + inviter := mustCreateUser(t, integrationEntClient, &service.User{ + Email: fmt.Sprintf("redeem-transaction-inviter-%d@example.com", time.Now().UnixNano()), + PasswordHash: "hash", + Role: service.RoleUser, + Status: service.StatusActive, + }) + userIDs = append(userIDs, inviter.ID) + invitee := mustCreateUser(t, integrationEntClient, &service.User{ + Email: fmt.Sprintf("redeem-transaction-invitee-%d@example.com", time.Now().UnixNano()+1), + PasswordHash: "hash", + Role: service.RoleUser, + Status: service.StatusActive, + Balance: 10, + }) + userIDs = append(userIDs, invitee.ID) + _, err := affiliateRepo.EnsureUserAffiliate(ctx, inviter.ID) + require.NoError(t, err) + _, err = affiliateRepo.EnsureUserAffiliate(ctx, invitee.ID) + require.NoError(t, err) + bound, err := affiliateRepo.BindInviter(ctx, invitee.ID, inviter.ID) + require.NoError(t, err) + require.True(t, bound) + + codeSequence := 0 + createBalanceCode := func(value float64) *service.RedeemCode { + codeSequence++ + code := &service.RedeemCode{ + Code: fmt.Sprintf("REALREDEEM%d-%d", time.Now().UnixNano(), codeSequence), + Type: service.RedeemTypeBalance, + Value: value, + Status: service.StatusUnused, + } + require.NoError(t, redeemRepo.Create(ctx, code)) + require.Positive(t, code.ID) + redeemCodeIDs = append(redeemCodeIDs, code.ID) + return code + } + ledgerCountForCode := func(codeID int64) int { + return querySingleInt(t, ctx, integrationEntClient, + "SELECT COUNT(*) FROM user_affiliate_ledger WHERE source_redeem_code_id = $1", codeID) + } + + affiliateService := service.NewAffiliateService(affiliateRepo, settingService, nil, nil) + redeemService := service.NewRedeemService(redeemRepo, userRepo, nil, nil, nil, integrationEntClient, nil, affiliateService) + + successCode := createBalanceCode(5) + redeemed, err := redeemService.Redeem(ctx, invitee.ID, successCode.Code) + require.NoError(t, err) + require.Equal(t, service.StatusUsed, redeemed.Status) + require.InDelta(t, 15, querySingleFloat(t, ctx, integrationEntClient, + "SELECT balance::double precision FROM users WHERE id = $1", invitee.ID), 1e-9) + require.InDelta(t, 1, querySingleFloat(t, ctx, integrationEntClient, + "SELECT aff_quota::double precision FROM user_affiliates WHERE user_id = $1", inviter.ID), 1e-9) + require.Equal(t, 1, ledgerCountForCode(successCode.ID)) + + failingAffiliateService := service.NewAffiliateService(&failingAccrueAffiliateRepository{ + AffiliateRepository: affiliateRepo, + err: errors.New("forced redeem affiliate failure"), + }, settingService, nil, nil) + failingRedeemService := service.NewRedeemService(redeemRepo, userRepo, nil, nil, nil, integrationEntClient, nil, failingAffiliateService) + failureCode := createBalanceCode(7) + redeemed, err = failingRedeemService.Redeem(ctx, invitee.ID, failureCode.Code) + require.Nil(t, redeemed) + require.ErrorContains(t, err, "forced redeem affiliate failure") + reloadedFailureCode, err := redeemRepo.GetByID(ctx, failureCode.ID) + require.NoError(t, err) + require.Equal(t, service.StatusUnused, reloadedFailureCode.Status) + require.Nil(t, reloadedFailureCode.UsedBy) + require.InDelta(t, 15, querySingleFloat(t, ctx, integrationEntClient, + "SELECT balance::double precision FROM users WHERE id = $1", invitee.ID), 1e-9) + require.InDelta(t, 1, querySingleFloat(t, ctx, integrationEntClient, + "SELECT aff_quota::double precision FROM user_affiliates WHERE user_id = $1", inviter.ID), 1e-9) + require.Zero(t, ledgerCountForCode(failureCode.ID)) + + skipCode := createBalanceCode(3) + redeemed, err = redeemService.Redeem(service.ContextSkipRedeemAffiliate(ctx), invitee.ID, skipCode.Code) + require.NoError(t, err) + require.Equal(t, service.StatusUsed, redeemed.Status) + require.InDelta(t, 18, querySingleFloat(t, ctx, integrationEntClient, + "SELECT balance::double precision FROM users WHERE id = $1", invitee.ID), 1e-9) + require.InDelta(t, 1, querySingleFloat(t, ctx, integrationEntClient, + "SELECT aff_quota::double precision FROM user_affiliates WHERE user_id = $1", inviter.ID), 1e-9) + require.Zero(t, ledgerCountForCode(skipCode.ID)) + + concurrentCode := createBalanceCode(4) + start := make(chan struct{}) + results := make(chan error, 2) + for range 2 { + go func() { + <-start + _, redeemErr := redeemService.Redeem(ctx, invitee.ID, concurrentCode.Code) + results <- redeemErr + }() + } + close(start) + successCount := 0 + failureCount := 0 + for range 2 { + if redeemErr := <-results; redeemErr == nil { + successCount++ + } else { + failureCount++ + require.ErrorIs(t, redeemErr, service.ErrRedeemCodeUsed) + } + } + require.Equal(t, 1, successCount) + require.Equal(t, 1, failureCount) + require.InDelta(t, 22, querySingleFloat(t, ctx, integrationEntClient, + "SELECT balance::double precision FROM users WHERE id = $1", invitee.ID), 1e-9) + require.InDelta(t, 1.8, querySingleFloat(t, ctx, integrationEntClient, + "SELECT aff_quota::double precision FROM user_affiliates WHERE user_id = $1", inviter.ID), 1e-9) + require.Equal(t, 1, ledgerCountForCode(concurrentCode.ID)) +} + +func TestAffiliateRebateSourcesMigration_DoesNotGuessHistoryAndRepairsSupersededDraft(t *testing.T) { + ctx := context.Background() + tx := testEntTx(t) + txCtx := dbent.NewTxContext(ctx, tx) + client := tx.Client() + repo := NewAffiliateRepository(client, integrationDB) + + inviter := mustCreateUser(t, client, &service.User{ + Email: fmt.Sprintf("migration-source-inviter-%d@example.com", time.Now().UnixNano()), + PasswordHash: "hash", + Role: service.RoleUser, + Status: service.StatusActive, + }) + ordinaryInvitee := mustCreateUser(t, client, &service.User{ + Email: fmt.Sprintf("migration-source-ordinary-%d@example.com", time.Now().UnixNano()+1), + PasswordHash: "hash", + Role: service.RoleUser, + Status: service.StatusActive, + }) + adminInvitee := mustCreateUser(t, client, &service.User{ + Email: fmt.Sprintf("migration-source-admin-%d@example.com", time.Now().UnixNano()+2), + PasswordHash: "hash", + Role: service.RoleUser, + Status: service.StatusActive, + }) + _, err := repo.EnsureUserAffiliate(txCtx, inviter.ID) + require.NoError(t, err) + for _, inviteeID := range []int64{ordinaryInvitee.ID, adminInvitee.ID} { + _, err = repo.EnsureUserAffiliate(txCtx, inviteeID) + require.NoError(t, err) + bound, bindErr := repo.BindInviter(txCtx, inviteeID, inviter.ID) + require.NoError(t, bindErr) + require.True(t, bound) + } + + _, err = client.ExecContext(txCtx, "ALTER TABLE user_affiliate_ledger DROP CONSTRAINT IF EXISTS chk_user_affiliate_ledger_source_type") + require.NoError(t, err) + now := time.Now().UTC() + ordinaryCodeID := insertHistoricalRedeemCode(t, txCtx, client, ordinaryInvitee.ID, "balance", 50, now.Add(-time.Minute)) + ordinaryLedgerID := insertHistoricalAffiliateLedger(t, txCtx, client, inviter.ID, ordinaryInvitee.ID, 10, now) + adminCodeID := insertHistoricalRedeemCode(t, txCtx, client, adminInvitee.ID, "admin_balance", 25, now) + adminLedgerID := insertHistoricalAffiliateLedger(t, txCtx, client, inviter.ID, adminInvitee.ID, 5, now.Add(-time.Minute)) + + coreMigrationSQL, err := migrationspkg.FS.ReadFile("231_affiliate_rebate_sources.sql") + require.NoError(t, err) + for range 2 { + _, err = client.ExecContext(txCtx, string(coreMigrationSQL)) + require.NoError(t, err) + } + + rows, err := client.QueryContext(txCtx, ` +SELECT source_type, source_redeem_code_id +FROM user_affiliate_ledger +WHERE id = $1`, ordinaryLedgerID) + require.NoError(t, err) + require.True(t, rows.Next()) + var ordinarySource *string + var ordinarySourceID *int64 + require.NoError(t, rows.Scan(&ordinarySource, &ordinarySourceID)) + require.NoError(t, rows.Close()) + require.Nil(t, ordinarySource, "无可靠业务编号的历史余额返利必须保持未分类") + require.Nil(t, ordinarySourceID) + + // 模拟早期分支迁移已经按时间窗错误关联了管理员调整记录。 + oldMigrationAppliedAt := now.Add(time.Minute) + _, err = client.ExecContext(txCtx, ` +UPDATE schema_migrations +SET checksum = $1, applied_at = $2 +WHERE filename = '231_affiliate_rebate_sources.sql'`, + "ceb508efbf81877a891a95fe6688cb3287462c2552e1a5c8a8254be9328d6806", + oldMigrationAppliedAt, + ) + require.NoError(t, err) + _, err = client.ExecContext(txCtx, ` +UPDATE user_affiliate_ledger +SET source_type = 'admin_recharge', + source_redeem_code_id = $1, + base_amount = 25 +WHERE id = $2`, adminCodeID, adminLedgerID) + require.NoError(t, err) + + constraintMigrationSQL, err := migrationspkg.FS.ReadFile("232_affiliate_rebate_source_constraints.sql") + require.NoError(t, err) + for range 2 { + _, err = client.ExecContext(txCtx, string(constraintMigrationSQL)) + require.NoError(t, err) + } + + rows, err = client.QueryContext(txCtx, ` +SELECT source_type, source_redeem_code_id, base_amount::double precision +FROM user_affiliate_ledger +WHERE id = $1`, adminLedgerID) + require.NoError(t, err) + require.True(t, rows.Next()) + var adminSource string + var adminSourceID *int64 + var adminBaseAmount *float64 + require.NoError(t, rows.Scan(&adminSource, &adminSourceID, &adminBaseAmount)) + require.NoError(t, rows.Close()) + require.Equal(t, string(service.AffiliateRebateSourceLegacyUnknown), adminSource) + require.Nil(t, adminSourceID) + require.Nil(t, adminBaseAmount) + + require.Positive(t, ordinaryCodeID) + require.Positive(t, adminCodeID) +} + +func insertHistoricalRedeemCode(t *testing.T, ctx context.Context, client *dbent.Client, userID int64, codeType string, value float64, usedAt time.Time) int64 { + t.Helper() + rows, err := client.QueryContext(ctx, ` +INSERT INTO redeem_codes (code, type, value, status, used_by, used_at, created_at) +VALUES ($1, $2, $3, 'used', $4, $5, $5) +RETURNING id`, fmt.Sprintf("HIST%d", time.Now().UnixNano()), codeType, value, userID, usedAt) + require.NoError(t, err) + require.True(t, rows.Next()) + var id int64 + require.NoError(t, rows.Scan(&id)) + require.NoError(t, rows.Close()) + return id +} + +func insertHistoricalAffiliateLedger(t *testing.T, ctx context.Context, client *dbent.Client, inviterID, inviteeID int64, amount float64, createdAt time.Time) int64 { + t.Helper() + rows, err := client.QueryContext(ctx, ` +INSERT INTO user_affiliate_ledger (user_id, action, amount, source_user_id, created_at, updated_at) +VALUES ($1, 'accrue', $2, $3, $4, $4) +RETURNING id`, inviterID, amount, inviteeID, createdAt) + require.NoError(t, err) + require.True(t, rows.Next()) + var id int64 + require.NoError(t, rows.Scan(&id)) + require.NoError(t, rows.Close()) + return id +} + func TestAffiliateRepository_TransferQuotaToBalance_EmptyQuota(t *testing.T) { ctx := context.Background() tx := testEntTx(t) diff --git a/backend/internal/repository/migrations_runner.go b/backend/internal/repository/migrations_runner.go index 0b786616a544..e9e3eb2461ab 100644 --- a/backend/internal/repository/migrations_runner.go +++ b/backend/internal/repository/migrations_runner.go @@ -62,6 +62,10 @@ const usageLogsUpstreamModelMismatchIndex = "idx_usage_logs_upstream_model_misma const usageLogsEffectiveModelIndexesMigration = "226_add_usage_log_effective_model_indexes_notx.sql" const usageLogsEffectiveRequestedModelIndex = "idx_usage_logs_effective_requested_model_created" const usageLogsEffectiveUpstreamModelIndex = "idx_usage_logs_effective_upstream_model_created" +const affiliateRebateSourceIndexesMigration = "231_affiliate_rebate_sources_indexes_notx.sql" +const affiliateRebateSourceTypeCreatedAtIndex = "idx_user_affiliate_ledger_source_type_created_at" +const affiliateRebateAccrueOrderUniqueIndex = "idx_user_affiliate_ledger_accrue_order_uniq" +const affiliateRebateAccrueRedeemCodeUniqueIndex = "idx_user_affiliate_ledger_accrue_redeem_code_uniq" type migrationChecksumCompatibilityRule struct { fileChecksum string @@ -94,6 +98,8 @@ var migrationChecksumCompatibilityRules = map[string]migrationChecksumCompatibil "220_clear_non_grok_video_generation_config.sql": newMigrationChecksumCompatibilityRule("85e320b9ec64f2d3fcd8cf705b2b4e76a7b49f7a57140c14bff97f32691c818b", "3da48c8fdffe6390325f43d08b8e353e0a365df43d44a78dbbe655d0deb18402"), "219_group_search_price_per_1k.sql": newMigrationChecksumCompatibilityRule("e86786ebcc3b14206fd2d321380a4e50e80cdadbfcf4962c639255e6a14008db", "df6ffd71b97e30ec2c8fe7b95e15783042dea58c553e32701ee7c42a5619af80"), "218_group_audio_voice_pricing.sql": newMigrationChecksumCompatibilityRule("40ee9f3a2af0e0a5e99dabc878fd0fe98be1011f26bcfcefcac7197f7081f0e7", "c2a5e5b4ffd6968ad1c10593289fbc11192cdea19fec3ed9bce3a84eff9a8351"), + // 231 的早期分支版本包含历史时间窗回填和事务内索引；修复版只保留快速加列。 + "231_affiliate_rebate_sources.sql": newMigrationChecksumCompatibilityRule("835825077e002997764e59c4a3ce026e9e36d7a328671b7983cce94c4fb9fc94", "ceb508efbf81877a891a95fe6688cb3287462c2552e1a5c8a8254be9328d6806"), } // ApplyMigrations 将嵌入的 SQL 迁移文件应用到指定的数据库。 @@ -305,11 +311,38 @@ func prepareNonTransactionalMigration(ctx context.Context, db migrationConnectio } } return nil + case affiliateRebateSourceIndexesMigration: + return prepareAffiliateRebateSourceIndexesMigration(ctx, db) default: return nil } } +func prepareAffiliateRebateSourceIndexesMigration(ctx context.Context, db migrationConnection) error { + duplicates, err := findDuplicateAffiliateRebateSources(ctx, db) + if err != nil { + return fmt.Errorf("precheck duplicate affiliate rebate sources: %w", err) + } + if len(duplicates) > 0 { + return fmt.Errorf( + "duplicate affiliate rebate sources block %s; remediate duplicates before retrying: %s", + affiliateRebateSourceIndexesMigration, + strings.Join(duplicates, ", "), + ) + } + + for _, indexName := range []string{ + affiliateRebateSourceTypeCreatedAtIndex, + affiliateRebateAccrueOrderUniqueIndex, + affiliateRebateAccrueRedeemCodeUniqueIndex, + } { + if err := dropInvalidIndexIfPresent(ctx, db, indexName); err != nil { + return err + } + } + return nil +} + func preparePaymentOrdersOutTradeNoUniqueMigration(ctx context.Context, db migrationConnection) error { duplicates, err := findDuplicatePaymentOrderOutTradeNos(ctx, db) if err != nil { @@ -373,6 +406,46 @@ func findDuplicatePaymentOrderOutTradeNos(ctx context.Context, db migrationConne return duplicates, nil } +func findDuplicateAffiliateRebateSources(ctx context.Context, db migrationConnection) ([]string, error) { + rows, err := db.QueryContext(ctx, ` + SELECT source_kind, source_id, COUNT(*) AS duplicate_count + FROM ( + SELECT 'source_order_id' AS source_kind, source_order_id AS source_id + FROM user_affiliate_ledger + WHERE action = 'accrue' AND source_order_id IS NOT NULL + UNION ALL + SELECT 'source_redeem_code_id' AS source_kind, source_redeem_code_id AS source_id + FROM user_affiliate_ledger + WHERE action = 'accrue' AND source_redeem_code_id IS NOT NULL + ) rebate_sources + GROUP BY source_kind, source_id + HAVING COUNT(*) > 1 + ORDER BY duplicate_count DESC, source_kind, source_id + LIMIT 10 + `) + if err != nil { + return nil, err + } + defer func() { + _ = rows.Close() + }() + + duplicates := make([]string, 0, 10) + for rows.Next() { + var sourceKind string + var sourceID int64 + var duplicateCount int + if err := rows.Scan(&sourceKind, &sourceID, &duplicateCount); err != nil { + return nil, err + } + duplicates = append(duplicates, fmt.Sprintf("%s=%d (count=%d)", sourceKind, sourceID, duplicateCount)) + } + if err := rows.Err(); err != nil { + return nil, err + } + return duplicates, nil +} + func indexIsInvalid(ctx context.Context, db migrationConnection, indexName string) (bool, error) { var invalid bool err := db.QueryRowContext(ctx, ` @@ -516,8 +589,8 @@ func validateMigrationExecutionMode(name, content string) (bool, error) { } if strings.Contains(normalizedStmt, "CONCURRENTLY") { - isCreateIndex := strings.Contains(normalizedStmt, "CREATE") && strings.Contains(normalizedStmt, "INDEX") - isDropIndex := strings.Contains(normalizedStmt, "DROP") && strings.Contains(normalizedStmt, "INDEX") + isCreateIndex := strings.HasPrefix(normalizedStmt, "CREATE INDEX ") || strings.HasPrefix(normalizedStmt, "CREATE UNIQUE INDEX ") + isDropIndex := strings.HasPrefix(normalizedStmt, "DROP INDEX ") if !isCreateIndex && !isDropIndex { return false, errors.New("*_notx.sql currently only supports CREATE/DROP INDEX CONCURRENTLY statements") } diff --git a/backend/internal/repository/migrations_runner_checksum_test.go b/backend/internal/repository/migrations_runner_checksum_test.go index 1fcb3be1e708..459310930193 100644 --- a/backend/internal/repository/migrations_runner_checksum_test.go +++ b/backend/internal/repository/migrations_runner_checksum_test.go @@ -161,4 +161,13 @@ func TestIsMigrationChecksumCompatible(t *testing.T) { ) require.False(t, ok) }) + + t.Run("231早期分支checksum兼容快速迁移修复版", func(t *testing.T) { + ok := isMigrationChecksumCompatible( + "231_affiliate_rebate_sources.sql", + "ceb508efbf81877a891a95fe6688cb3287462c2552e1a5c8a8254be9328d6806", + "835825077e002997764e59c4a3ce026e9e36d7a328671b7983cce94c4fb9fc94", + ) + require.True(t, ok) + }) } diff --git a/backend/internal/repository/migrations_runner_notx_test.go b/backend/internal/repository/migrations_runner_notx_test.go index 8f9882e6d5ac..9346a024808d 100644 --- a/backend/internal/repository/migrations_runner_notx_test.go +++ b/backend/internal/repository/migrations_runner_notx_test.go @@ -7,6 +7,7 @@ import ( "testing/fstest" sqlmock "github.com/DATA-DOG/go-sqlmock" + migrationspkg "github.com/Wei-Shaw/sub2api/migrations" "github.com/stretchr/testify/require" ) @@ -43,14 +44,32 @@ func TestValidateMigrationExecutionMode(t *testing.T) { t.Run("notx迁移允许幂等并发索引语句", func(t *testing.T) { nonTx, err := validateMigrationExecutionMode("001_add_idx_notx.sql", ` -CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_a ON t(a); -DROP INDEX CONCURRENTLY IF EXISTS idx_b; -`) + CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_a ON t(a); + DROP INDEX CONCURRENTLY IF EXISTS idx_b; + `) + require.True(t, nonTx) + require.NoError(t, err) + }) + + t.Run("删除索引名称包含created_at时不会误判为创建语句", func(t *testing.T) { + nonTx, err := validateMigrationExecutionMode( + "001_drop_idx_notx.sql", + "DROP INDEX CONCURRENTLY IF EXISTS idx_user_affiliate_ledger_source_type_created_at;", + ) require.True(t, nonTx) require.NoError(t, err) }) } +func TestAffiliateRebateSourceIndexesMigrationUsesSupportedExecutionMode(t *testing.T) { + content, err := migrationspkg.FS.ReadFile(affiliateRebateSourceIndexesMigration) + require.NoError(t, err) + + nonTx, err := validateMigrationExecutionMode(affiliateRebateSourceIndexesMigration, string(content)) + require.NoError(t, err) + require.True(t, nonTx) +} + func TestApplyMigrationsFS_NonTransactionalMigration(t *testing.T) { db, mock, err := sqlmock.New() require.NoError(t, err) @@ -310,6 +329,96 @@ DROP INDEX CONCURRENTLY IF EXISTS paymentorder_out_trade_no; require.NoError(t, mock.ExpectationsWereMet()) } +func TestApplyMigrationsFS_AffiliateRebateSourceIndexesMigration_FailsFastOnDuplicatePrecheck(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer func() { _ = db.Close() }() + + prepareMigrationsBootstrapExpectations(mock) + mock.ExpectQuery("SELECT checksum FROM schema_migrations WHERE filename = \\$1"). + WithArgs(affiliateRebateSourceIndexesMigration). + WillReturnError(sql.ErrNoRows) + mock.ExpectQuery("SELECT source_kind, source_id, COUNT\\(\\*\\) AS duplicate_count"). + WillReturnRows(sqlmock.NewRows([]string{"source_kind", "source_id", "duplicate_count"}). + AddRow("source_redeem_code_id", int64(42), 2)) + mock.ExpectExec("SELECT pg_advisory_unlock\\(\\$1\\)"). + WithArgs(migrationsAdvisoryLockID). + WillReturnResult(sqlmock.NewResult(0, 1)) + + fsys := fstest.MapFS{ + affiliateRebateSourceIndexesMigration: &fstest.MapFile{Data: []byte(` +CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS idx_user_affiliate_ledger_accrue_redeem_code_uniq +ON user_affiliate_ledger(source_redeem_code_id) +WHERE action = 'accrue' AND source_redeem_code_id IS NOT NULL; +`)}, + } + + err = applyMigrationsFS(context.Background(), db, fsys) + require.ErrorContains(t, err, "source_redeem_code_id=42 (count=2)") + require.NoError(t, mock.ExpectationsWereMet()) +} + +func TestApplyMigrationsFS_AffiliateRebateSourceIndexesMigration_DropsInvalidIndexesBeforeRetry(t *testing.T) { + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer func() { _ = db.Close() }() + + prepareMigrationsBootstrapExpectations(mock) + mock.ExpectQuery("SELECT checksum FROM schema_migrations WHERE filename = \\$1"). + WithArgs(affiliateRebateSourceIndexesMigration). + WillReturnError(sql.ErrNoRows) + mock.ExpectQuery("SELECT source_kind, source_id, COUNT\\(\\*\\) AS duplicate_count"). + WillReturnRows(sqlmock.NewRows([]string{"source_kind", "source_id", "duplicate_count"})) + + for _, index := range []struct { + name string + invalid bool + }{ + {name: affiliateRebateSourceTypeCreatedAtIndex, invalid: false}, + {name: affiliateRebateAccrueOrderUniqueIndex, invalid: true}, + {name: affiliateRebateAccrueRedeemCodeUniqueIndex, invalid: true}, + } { + mock.ExpectQuery("SELECT EXISTS \\("). + WithArgs(index.name). + WillReturnRows(sqlmock.NewRows([]string{"exists"}).AddRow(index.invalid)) + if index.invalid { + mock.ExpectExec("DROP INDEX CONCURRENTLY IF EXISTS " + index.name). + WillReturnResult(sqlmock.NewResult(0, 0)) + } + } + + mock.ExpectExec("DROP INDEX CONCURRENTLY IF EXISTS idx_user_affiliate_ledger_source_type_created_at"). + WillReturnResult(sqlmock.NewResult(0, 0)) + mock.ExpectExec("CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_user_affiliate_ledger_source_type_created_at"). + WillReturnResult(sqlmock.NewResult(0, 0)) + mock.ExpectExec("CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS idx_user_affiliate_ledger_accrue_order_uniq"). + WillReturnResult(sqlmock.NewResult(0, 0)) + mock.ExpectExec("CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS idx_user_affiliate_ledger_accrue_redeem_code_uniq"). + WillReturnResult(sqlmock.NewResult(0, 0)) + mock.ExpectExec("INSERT INTO schema_migrations \\(filename, checksum\\) VALUES \\(\\$1, \\$2\\)"). + WithArgs(affiliateRebateSourceIndexesMigration, sqlmock.AnyArg()). + WillReturnResult(sqlmock.NewResult(1, 1)) + mock.ExpectExec("SELECT pg_advisory_unlock\\(\\$1\\)"). + WithArgs(migrationsAdvisoryLockID). + WillReturnResult(sqlmock.NewResult(0, 1)) + + fsys := fstest.MapFS{ + affiliateRebateSourceIndexesMigration: &fstest.MapFile{Data: []byte(` +DROP INDEX CONCURRENTLY IF EXISTS idx_user_affiliate_ledger_source_type_created_at; +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_user_affiliate_ledger_source_type_created_at +ON user_affiliate_ledger(source_type); +CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS idx_user_affiliate_ledger_accrue_order_uniq +ON user_affiliate_ledger(source_order_id); +CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS idx_user_affiliate_ledger_accrue_redeem_code_uniq +ON user_affiliate_ledger(source_redeem_code_id); +`)}, + } + + err = applyMigrationsFS(context.Background(), db, fsys) + require.NoError(t, err) + require.NoError(t, mock.ExpectationsWereMet()) +} + func TestApplyMigrationsFS_SchedulerOutboxPendingDedupKeyMigration_DropsInvalidIndexBeforeRetry(t *testing.T) { db, mock, err := sqlmock.New() require.NoError(t, err) diff --git a/backend/internal/repository/redeem_code_repo.go b/backend/internal/repository/redeem_code_repo.go index 2bdb34b42fa6..7bb79b9707f6 100644 --- a/backend/internal/repository/redeem_code_repo.go +++ b/backend/internal/repository/redeem_code_repo.go @@ -23,7 +23,7 @@ func NewRedeemCodeRepository(client *dbent.Client) service.RedeemCodeRepository } func (r *redeemCodeRepository) Create(ctx context.Context, code *service.RedeemCode) error { - created, err := r.client.RedeemCode.Create(). + created, err := clientFromContext(ctx, r.client).RedeemCode.Create(). SetCode(code.Code). SetType(code.Type). SetValue(code.Value). diff --git a/backend/internal/service/admin_service.go b/backend/internal/service/admin_service.go index e48eea51c1ba..5f1c35def8fb 100644 --- a/backend/internal/service/admin_service.go +++ b/backend/internal/service/admin_service.go @@ -681,7 +681,7 @@ type ChannelCacheInvalidator interface { } type adminRechargeAffiliateAccruer interface { - AccrueInviteRebate(ctx context.Context, inviteeUserID int64, baseRechargeAmount float64) (float64, error) + AccrueInviteRebate(ctx context.Context, inviteeUserID int64, source AffiliateRebateSource) (float64, error) } type userGroupRateBatchReader interface { diff --git a/backend/internal/service/admin_service_update_balance_test.go b/backend/internal/service/admin_service_update_balance_test.go index 82f862b422ef..687508213541 100644 --- a/backend/internal/service/admin_service_update_balance_test.go +++ b/backend/internal/service/admin_service_update_balance_test.go @@ -52,6 +52,8 @@ func (s *balanceRedeemRepoStub) Create(ctx context.Context, code *RedeemCode) er return nil } clone := *code + clone.ID = int64(len(s.created) + 1) + code.ID = clone.ID s.created = append(s.created, &clone) return nil } @@ -69,12 +71,23 @@ type adminRechargeAffiliateAccruerStub struct { } type adminRechargeAffiliateAccrual struct { - userID int64 - amount float64 + userID int64 + amount float64 + sourceType AffiliateRebateSourceType + redeemCodeID int64 } -func (s *adminRechargeAffiliateAccruerStub) AccrueInviteRebate(_ context.Context, userID int64, amount float64) (float64, error) { - s.calls = append(s.calls, adminRechargeAffiliateAccrual{userID: userID, amount: amount}) +func (s *adminRechargeAffiliateAccruerStub) AccrueInviteRebate(_ context.Context, userID int64, source AffiliateRebateSource) (float64, error) { + var redeemCodeID int64 + if source.RedeemCodeID != nil { + redeemCodeID = *source.RedeemCodeID + } + s.calls = append(s.calls, adminRechargeAffiliateAccrual{ + userID: userID, + amount: source.BaseAmount, + sourceType: source.Type, + redeemCodeID: redeemCodeID, + }) return s.rebate, s.err } @@ -119,6 +132,7 @@ func TestAdminService_UpdateUserBalance_UsesAtomicPrimitives(t *testing.T) { svc := &adminServiceImpl{ userRepo: repo, redeemCodeRepo: &balanceRedeemRepoStub{redeemRepoStub: &redeemRepoStub{}}, + entClient: newPaymentConfigServiceTestClient(t), } user, err := svc.UpdateUserBalance(context.Background(), 7, tt.amount, tt.operation, "") @@ -134,6 +148,7 @@ func TestAdminService_UpdateUserBalance_RejectsNegativeResult(t *testing.T) { svc := &adminServiceImpl{ userRepo: repo, redeemCodeRepo: &balanceRedeemRepoStub{redeemRepoStub: &redeemRepoStub{}}, + entClient: newPaymentConfigServiceTestClient(t), } _, err := svc.UpdateUserBalance(context.Background(), 7, 4, "subtract", "") @@ -164,6 +179,7 @@ func TestAdminService_UpdateUserBalance_InvalidatesAuthCache(t *testing.T) { userRepo: repo, redeemCodeRepo: redeemRepo, authCacheInvalidator: invalidator, + entClient: newPaymentConfigServiceTestClient(t), } _, err := svc.UpdateUserBalance(context.Background(), 7, 5, "add", "") @@ -181,6 +197,7 @@ func TestAdminService_UpdateUserBalance_NoChangeNoInvalidate(t *testing.T) { userRepo: repo, redeemCodeRepo: redeemRepo, authCacheInvalidator: invalidator, + entClient: newPaymentConfigServiceTestClient(t), } _, err := svc.UpdateUserBalance(context.Background(), 7, 10, "set", "") @@ -207,7 +224,7 @@ func TestAdminService_UpdateUserBalance_AdminRechargeAffiliateRebate(t *testing. enabled: true, operation: "add", amount: 0.1, - wantCalls: []adminRechargeAffiliateAccrual{{userID: 7, amount: 0.1}}, + wantCalls: []adminRechargeAffiliateAccrual{{userID: 7, amount: 0.1, sourceType: AffiliateRebateSourceAdminRecharge, redeemCodeID: 1}}, }, { name: "enabled set increase", @@ -234,6 +251,7 @@ func TestAdminService_UpdateUserBalance_AdminRechargeAffiliateRebate(t *testing. redeemCodeRepo: redeemRepo, settingService: adminRechargeSettingService(tt.enabled), affiliateService: affiliate, + entClient: newPaymentConfigServiceTestClient(t), } _, err := svc.UpdateUserBalance(context.Background(), 7, tt.amount, tt.operation, "") @@ -243,7 +261,7 @@ func TestAdminService_UpdateUserBalance_AdminRechargeAffiliateRebate(t *testing. } } -func TestAdminService_UpdateUserBalance_AffiliateFailureDoesNotRollbackRecharge(t *testing.T) { +func TestAdminService_UpdateUserBalance_AffiliateFailureReturnsError(t *testing.T) { baseRepo := &userRepoStub{user: &User{ID: 7, Balance: 10}} repo := &balanceUserRepoStub{userRepoStub: baseRepo} redeemRepo := &balanceRedeemRepoStub{redeemRepoStub: &redeemRepoStub{}} @@ -253,11 +271,12 @@ func TestAdminService_UpdateUserBalance_AffiliateFailureDoesNotRollbackRecharge( redeemCodeRepo: redeemRepo, settingService: adminRechargeSettingService(true), affiliateService: affiliate, + entClient: newPaymentConfigServiceTestClient(t), } user, err := svc.UpdateUserBalance(context.Background(), 7, 5, "add", "") - require.NoError(t, err) - require.Equal(t, 15.0, user.Balance) - require.Equal(t, []adminRechargeAffiliateAccrual{{userID: 7, amount: 5}}, affiliate.calls) + require.Nil(t, user) + require.ErrorContains(t, err, "accrue affiliate rebate for admin recharge") + require.Equal(t, []adminRechargeAffiliateAccrual{{userID: 7, amount: 5, sourceType: AffiliateRebateSourceAdminRecharge, redeemCodeID: 1}}, affiliate.calls) require.Len(t, redeemRepo.created, 1) } diff --git a/backend/internal/service/admin_user.go b/backend/internal/service/admin_user.go index 12f71f96a267..5275a73ac597 100644 --- a/backend/internal/service/admin_user.go +++ b/backend/internal/service/admin_user.go @@ -514,25 +514,60 @@ func (s *adminServiceImpl) BatchUpdateLimits(ctx context.Context, userIDs []int6 } func (s *adminServiceImpl) UpdateUserBalance(ctx context.Context, userID int64, balance float64, operation string, notes string) (*User, error) { - // 余额调整必须走原子接口：先读后整行写回会把并发的计费扣款覆盖掉。 - var ( - change BalanceChange - err error - ) - switch operation { - case "set": - change, err = s.userRepo.SetBalance(ctx, userID, balance) - case "add": - change, err = s.userRepo.AdjustBalance(ctx, userID, balance) - case "subtract": - change, err = s.userRepo.AdjustBalance(ctx, userID, -balance) - default: + if operation != "set" && operation != "add" && operation != "subtract" { return nil, fmt.Errorf("unsupported balance operation: %q", operation) } - if errors.Is(err, ErrBalanceNegative) { - return nil, fmt.Errorf("balance cannot be negative, current balance: %.2f, requested operation would result in: %.2f", change.Old, change.New) - } - if err != nil { + + var balanceDiff float64 + if err := s.withAdminBalanceUpdateTx(ctx, func(txCtx context.Context) error { + // 余额调整必须走原子接口：先读后整行写回会把并发的计费扣款覆盖掉。 + var ( + change BalanceChange + err error + ) + switch operation { + case "set": + change, err = s.userRepo.SetBalance(txCtx, userID, balance) + case "add": + change, err = s.userRepo.AdjustBalance(txCtx, userID, balance) + case "subtract": + change, err = s.userRepo.AdjustBalance(txCtx, userID, -balance) + } + if errors.Is(err, ErrBalanceNegative) { + return fmt.Errorf("balance cannot be negative, current balance: %.2f, requested operation would result in: %.2f", change.Old, change.New) + } + if err != nil { + return err + } + + balanceDiff = change.New - change.Old + if balanceDiff == 0 { + return nil + } + + code, err := GenerateRedeemCode() + if err != nil { + return fmt.Errorf("generate balance adjustment record code: %w", err) + } + adjustmentRecord := &RedeemCode{ + Code: code, + Type: AdjustmentTypeAdminBalance, + Value: balanceDiff, + Status: StatusUsed, + UsedBy: &userID, + Notes: notes, + } + now := time.Now() + adjustmentRecord.UsedAt = &now + if err := s.redeemCodeRepo.Create(txCtx, adjustmentRecord); err != nil { + return fmt.Errorf("create balance adjustment record: %w", err) + } + if adjustmentRecord.ID <= 0 { + return errors.New("balance adjustment record has invalid id") + } + + return s.accrueAffiliateRebateForAdminRecharge(txCtx, userID, operation, balance, adjustmentRecord.ID) + }); err != nil { return nil, err } @@ -540,14 +575,11 @@ func (s *adminServiceImpl) UpdateUserBalance(ctx context.Context, userID int64, if err != nil { return nil, err } - - balanceDiff := change.New - change.Old + // 缓存只在数据库事务提交后失效，避免其他请求读取到尚未提交的余额。 if s.authCacheInvalidator != nil && balanceDiff != 0 { s.authCacheInvalidator.InvalidateAuthCacheByUserID(ctx, userID) } - s.tryAccrueAffiliateRebateForAdminRecharge(ctx, userID, operation, balance) - - if s.billingCacheService != nil { + if s.billingCacheService != nil && balanceDiff != 0 { go func() { cacheCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() @@ -557,48 +589,50 @@ func (s *adminServiceImpl) UpdateUserBalance(ctx context.Context, userID int64, }() } - if balanceDiff != 0 { - code, err := GenerateRedeemCode() - if err != nil { - logger.LegacyPrintf("service.admin", "failed to generate adjustment redeem code: %v", err) - return user, nil - } - - adjustmentRecord := &RedeemCode{ - Code: code, - Type: AdjustmentTypeAdminBalance, - Value: balanceDiff, - Status: StatusUsed, - UsedBy: &user.ID, - Notes: notes, - } - now := time.Now() - adjustmentRecord.UsedAt = &now + return user, nil +} - if err := s.redeemCodeRepo.Create(ctx, adjustmentRecord); err != nil { - logger.LegacyPrintf("service.admin", "failed to create balance adjustment redeem code: %v", err) - } +// withAdminBalanceUpdateTx 保证余额、调整记录、返利流水和返利额度一次提交。 +func (s *adminServiceImpl) withAdminBalanceUpdateTx(ctx context.Context, fn func(context.Context) error) error { + if s.entClient == nil { + return infraerrors.ServiceUnavailable("SERVICE_UNAVAILABLE", "admin balance transaction unavailable") } + tx, err := s.entClient.Tx(ctx) + if err != nil { + return fmt.Errorf("begin admin balance transaction: %w", err) + } + defer func() { _ = tx.Rollback() }() - return user, nil + txCtx := dbent.NewTxContext(ctx, tx) + if err := fn(txCtx); err != nil { + return err + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("commit admin balance transaction: %w", err) + } + return nil } -func (s *adminServiceImpl) tryAccrueAffiliateRebateForAdminRecharge(ctx context.Context, userID int64, operation string, amount float64) { +func (s *adminServiceImpl) accrueAffiliateRebateForAdminRecharge(ctx context.Context, userID int64, operation string, amount float64, adjustmentRecordID int64) error { if operation != "add" || amount <= 0 || s.settingService == nil || s.affiliateService == nil { - return + return nil } if !s.settingService.IsAffiliateAdminRechargeEnabled(ctx) { - return + return nil } - rebate, err := s.affiliateService.AccrueInviteRebate(ctx, userID, amount) + rebate, err := s.affiliateService.AccrueInviteRebate(ctx, userID, AffiliateRebateSource{ + Type: AffiliateRebateSourceAdminRecharge, + BaseAmount: amount, + RedeemCodeID: &adjustmentRecordID, + }) if err != nil { - logger.LegacyPrintf("service.admin", "affiliate rebate failed for admin recharge: user_id=%d amount=%.8f err=%v", userID, amount, err) - return + return fmt.Errorf("accrue affiliate rebate for admin recharge: %w", err) } if rebate > 0 { logger.LegacyPrintf("service.admin", "affiliate rebate accrued for admin recharge: user_id=%d amount=%.8f rebate=%.8f", userID, amount, rebate) } + return nil } func (s *adminServiceImpl) GetUserAPIKeys(ctx context.Context, userID int64, page, pageSize int, sortBy, sortOrder string) ([]APIKey, int64, error) { diff --git a/backend/internal/service/affiliate_service.go b/backend/internal/service/affiliate_service.go index 91cca5e216da..9dec37d0e423 100644 --- a/backend/internal/service/affiliate_service.go +++ b/backend/internal/service/affiliate_service.go @@ -17,6 +17,7 @@ var ( ErrAffiliateCodeTaken = infraerrors.Conflict("AFFILIATE_CODE_TAKEN", "affiliate code already in use") ErrAffiliateAlreadyBound = infraerrors.Conflict("AFFILIATE_ALREADY_BOUND", "affiliate inviter already bound") ErrAffiliateQuotaEmpty = infraerrors.BadRequest("AFFILIATE_QUOTA_EMPTY", "no affiliate quota available to transfer") + ErrAffiliateRebateSource = errors.New("invalid affiliate rebate source") ) const ( @@ -94,12 +95,60 @@ type AffiliateDetail struct { Invitees []AffiliateInvitee `json:"invitees"` } +type AffiliateRebateSourceType string + +const ( + AffiliateRebateSourcePaymentOrder AffiliateRebateSourceType = "payment_order" + AffiliateRebateSourceBalanceRedeem AffiliateRebateSourceType = "balance_redeem_code" + AffiliateRebateSourceAdminRecharge AffiliateRebateSourceType = "admin_recharge" + AffiliateRebateSourceLegacyUnknown AffiliateRebateSourceType = "legacy_unknown" + AffiliateRebateSourceFilterAll AffiliateRebateSourceType = "all" +) + +type AffiliateRebateSource struct { + Type AffiliateRebateSourceType + BaseAmount float64 + OrderID *int64 + RedeemCodeID *int64 +} + +func (s AffiliateRebateSource) ValidateForAccrual() error { + if s.BaseAmount <= 0 || math.IsNaN(s.BaseAmount) || math.IsInf(s.BaseAmount, 0) { + return ErrAffiliateRebateSource + } + switch s.Type { + case AffiliateRebateSourcePaymentOrder: + if s.OrderID == nil || *s.OrderID <= 0 || s.RedeemCodeID != nil { + return ErrAffiliateRebateSource + } + case AffiliateRebateSourceBalanceRedeem: + if s.RedeemCodeID == nil || *s.RedeemCodeID <= 0 || s.OrderID != nil { + return ErrAffiliateRebateSource + } + case AffiliateRebateSourceAdminRecharge: + if s.RedeemCodeID == nil || *s.RedeemCodeID <= 0 || s.OrderID != nil { + return ErrAffiliateRebateSource + } + default: + return ErrAffiliateRebateSource + } + return nil +} + +type AffiliateAccrualInput struct { + InviterID int64 + InviteeUserID int64 + Amount float64 + PerInviteeCap float64 + FreezeHours int + Source AffiliateRebateSource +} + type AffiliateRepository interface { EnsureUserAffiliate(ctx context.Context, userID int64) (*AffiliateSummary, error) GetAffiliateByCode(ctx context.Context, code string) (*AffiliateSummary, error) BindInviter(ctx context.Context, userID, inviterID int64) (bool, error) - AccrueQuota(ctx context.Context, inviterID, inviteeUserID int64, amount float64, freezeHours int, sourceOrderID *int64) (bool, error) - GetAccruedRebateFromInvitee(ctx context.Context, inviterID, inviteeUserID int64) (float64, error) + AccrueQuota(ctx context.Context, input AffiliateAccrualInput) (float64, error) ThawFrozenQuota(ctx context.Context, userID int64) (float64, error) TransferQuotaToBalance(ctx context.Context, userID int64) (float64, float64, error) ListInvitees(ctx context.Context, inviterID int64, limit int) ([]AffiliateInvitee, error) @@ -135,13 +184,14 @@ type AffiliateAdminEntry struct { } type AffiliateRecordFilter struct { - Search string - Page int - PageSize int - StartAt *time.Time - EndAt *time.Time - SortBy string - SortDesc bool + Search string + SourceType string + Page int + PageSize int + StartAt *time.Time + EndAt *time.Time + SortBy string + SortDesc bool } type AffiliateInviteRecord struct { @@ -157,20 +207,25 @@ type AffiliateInviteRecord struct { } type AffiliateRebateRecord struct { - OrderID int64 `json:"order_id"` - OutTradeNo string `json:"out_trade_no"` - InviterID int64 `json:"inviter_id"` - InviterEmail string `json:"inviter_email"` - InviterUsername string `json:"inviter_username"` - InviteeID int64 `json:"invitee_id"` - InviteeEmail string `json:"invitee_email"` - InviteeUsername string `json:"invitee_username"` - OrderAmount float64 `json:"order_amount"` - PayAmount float64 `json:"pay_amount"` - RebateAmount float64 `json:"rebate_amount"` - PaymentType string `json:"payment_type"` - OrderStatus string `json:"order_status"` - CreatedAt time.Time `json:"created_at"` + LedgerID int64 `json:"ledger_id"` + SourceType string `json:"source_type"` + OrderID *int64 `json:"order_id,omitempty"` + OutTradeNo *string `json:"out_trade_no,omitempty"` + RedeemCodeID *int64 `json:"redeem_code_id,omitempty"` + RedeemCodeMasked *string `json:"redeem_code_masked,omitempty"` + InviterID int64 `json:"inviter_id"` + InviterEmail string `json:"inviter_email"` + InviterUsername string `json:"inviter_username"` + InviteeID int64 `json:"invitee_id"` + InviteeEmail string `json:"invitee_email"` + InviteeUsername string `json:"invitee_username"` + BaseAmount *float64 `json:"base_amount,omitempty"` + OrderAmount *float64 `json:"order_amount,omitempty"` + PayAmount *float64 `json:"pay_amount,omitempty"` + RebateAmount float64 `json:"rebate_amount"` + PaymentType *string `json:"payment_type,omitempty"` + OrderStatus *string `json:"order_status,omitempty"` + CreatedAt time.Time `json:"created_at"` } type AffiliateTransferRecord struct { @@ -311,15 +366,14 @@ func (s *AffiliateService) BindInviterByCode(ctx context.Context, userID int64, return nil } -func (s *AffiliateService) AccrueInviteRebate(ctx context.Context, inviteeUserID int64, baseRechargeAmount float64) (float64, error) { - return s.AccrueInviteRebateForOrder(ctx, inviteeUserID, baseRechargeAmount, nil) -} - -func (s *AffiliateService) AccrueInviteRebateForOrder(ctx context.Context, inviteeUserID int64, baseRechargeAmount float64, sourceOrderID *int64) (float64, error) { +func (s *AffiliateService) AccrueInviteRebate(ctx context.Context, inviteeUserID int64, source AffiliateRebateSource) (float64, error) { if s == nil || s.repo == nil { return 0, nil } - if inviteeUserID <= 0 || baseRechargeAmount <= 0 || math.IsNaN(baseRechargeAmount) || math.IsInf(baseRechargeAmount, 0) { + if err := source.ValidateForAccrual(); err != nil { + return 0, err + } + if inviteeUserID <= 0 { return 0, nil } // 总开关关闭时，新充值不再产生返利 @@ -350,25 +404,14 @@ func (s *AffiliateService) AccrueInviteRebateForOrder(ctx context.Context, invit } rebateRatePercent := s.resolveRebateRatePercent(ctx, inviterSummary) - rebate := roundTo(baseRechargeAmount*(rebateRatePercent/100), 8) + rebate := roundTo(source.BaseAmount*(rebateRatePercent/100), 8) if rebate <= 0 { return 0, nil } - // 单人上限检查：精确截断到剩余额度 + var perInviteeCap float64 if s.settingService != nil { - if perInviteeCap := s.settingService.GetAffiliateRebatePerInviteeCap(ctx); perInviteeCap > 0 { - existing, err := s.repo.GetAccruedRebateFromInvitee(ctx, *inviteeSummary.InviterID, inviteeUserID) - if err != nil { - return 0, err - } - if existing >= perInviteeCap { - return 0, nil - } - if remaining := perInviteeCap - existing; rebate > remaining { - rebate = roundTo(remaining, 8) - } - } + perInviteeCap = s.settingService.GetAffiliateRebatePerInviteeCap(ctx) } var freezeHours int @@ -376,14 +419,18 @@ func (s *AffiliateService) AccrueInviteRebateForOrder(ctx context.Context, invit freezeHours = s.settingService.GetAffiliateRebateFreezeHours(ctx) } - applied, err := s.repo.AccrueQuota(ctx, *inviteeSummary.InviterID, inviteeUserID, rebate, freezeHours, sourceOrderID) + applied, err := s.repo.AccrueQuota(ctx, AffiliateAccrualInput{ + InviterID: *inviteeSummary.InviterID, + InviteeUserID: inviteeUserID, + Amount: rebate, + PerInviteeCap: perInviteeCap, + FreezeHours: freezeHours, + Source: source, + }) if err != nil { return 0, err } - if !applied { - return 0, nil - } - return rebate, nil + return applied, nil } // resolveRebateRatePercent returns the inviter's exclusive rate when set, diff --git a/backend/internal/service/affiliate_service_test.go b/backend/internal/service/affiliate_service_test.go index c02a4dd711b1..f09697e005a2 100644 --- a/backend/internal/service/affiliate_service_test.go +++ b/backend/internal/service/affiliate_service_test.go @@ -6,6 +6,7 @@ import ( "context" "math" "testing" + "time" "github.com/stretchr/testify/require" ) @@ -57,6 +58,67 @@ func TestIsEnabled_NilSettingServiceReturnsDefault(t *testing.T) { require.Equal(t, AffiliateEnabledDefault, svc.IsEnabled(context.Background())) } +func TestAffiliateRebateSourceValidateForAccrual(t *testing.T) { + orderID := int64(11) + redeemCodeID := int64(22) + + require.NoError(t, (AffiliateRebateSource{ + Type: AffiliateRebateSourcePaymentOrder, + BaseAmount: 10, + OrderID: &orderID, + }).ValidateForAccrual()) + require.NoError(t, (AffiliateRebateSource{ + Type: AffiliateRebateSourceBalanceRedeem, + BaseAmount: 10, + RedeemCodeID: &redeemCodeID, + }).ValidateForAccrual()) + require.NoError(t, (AffiliateRebateSource{ + Type: AffiliateRebateSourceAdminRecharge, + BaseAmount: 10, + RedeemCodeID: &redeemCodeID, + }).ValidateForAccrual()) + + require.ErrorIs(t, (AffiliateRebateSource{Type: AffiliateRebateSourcePaymentOrder}).ValidateForAccrual(), ErrAffiliateRebateSource) + require.ErrorIs(t, (AffiliateRebateSource{Type: AffiliateRebateSourceBalanceRedeem}).ValidateForAccrual(), ErrAffiliateRebateSource) + require.ErrorIs(t, (AffiliateRebateSource{Type: AffiliateRebateSourceAdminRecharge}).ValidateForAccrual(), ErrAffiliateRebateSource) + require.ErrorIs(t, (AffiliateRebateSource{Type: AffiliateRebateSourceAdminRecharge, RedeemCodeID: new(int64)}).ValidateForAccrual(), ErrAffiliateRebateSource) + require.ErrorIs(t, (AffiliateRebateSource{Type: AffiliateRebateSourceBalanceRedeem, RedeemCodeID: &redeemCodeID}).ValidateForAccrual(), ErrAffiliateRebateSource) + require.ErrorIs(t, (AffiliateRebateSource{Type: AffiliateRebateSourceBalanceRedeem, BaseAmount: math.NaN(), RedeemCodeID: &redeemCodeID}).ValidateForAccrual(), ErrAffiliateRebateSource) + require.ErrorIs(t, (AffiliateRebateSource{Type: AffiliateRebateSourceBalanceRedeem, BaseAmount: math.Inf(1), RedeemCodeID: &redeemCodeID}).ValidateForAccrual(), ErrAffiliateRebateSource) + require.ErrorIs(t, (AffiliateRebateSource{Type: AffiliateRebateSourceLegacyUnknown}).ValidateForAccrual(), ErrAffiliateRebateSource) +} + +func TestAccrueInviteRebatePassesSourceAndCapToRepository(t *testing.T) { + inviterID := int64(10) + inviteeID := int64(20) + redeemCodeID := int64(30) + repo := &paymentFulfillmentAffiliateRepoStub{ + inviteeSummary: &AffiliateSummary{UserID: inviteeID, InviterID: &inviterID, CreatedAt: time.Now().Add(-time.Hour)}, + inviterSummary: &AffiliateSummary{UserID: inviterID}, + } + settingSvc := NewSettingService(&paymentFulfillmentSettingRepoStub{values: map[string]string{ + SettingKeyAffiliateEnabled: "true", + SettingKeyAffiliateRebateRate: "20", + SettingKeyAffiliateRebatePerInviteeCap: "7", + }}, nil) + svc := NewAffiliateService(repo, settingSvc, nil, nil) + + rebate, err := svc.AccrueInviteRebate(context.Background(), inviteeID, AffiliateRebateSource{ + Type: AffiliateRebateSourceBalanceRedeem, + BaseAmount: 25, + RedeemCodeID: &redeemCodeID, + }) + + require.NoError(t, err) + require.InDelta(t, 5, rebate, 1e-9) + require.Len(t, repo.accrueCalls, 1) + call := repo.accrueCalls[0] + require.Equal(t, inviterID, call.inviterID) + require.Equal(t, inviteeID, call.inviteeUserID) + require.Equal(t, AffiliateRebateSourceBalanceRedeem, call.sourceType) + require.InDelta(t, 25, call.baseAmount, 1e-9) +} + // TestValidateExclusiveRate_BoundaryAndInvalid covers the validator used by // admin-facing rate setters: nil is always valid (clear), in-range values // are accepted, NaN/Inf and out-of-range values produce a typed BadRequest. diff --git a/backend/internal/service/payment_fulfillment.go b/backend/internal/service/payment_fulfillment.go index 4d442f3d1e33..237de3f86a0f 100644 --- a/backend/internal/service/payment_fulfillment.go +++ b/backend/internal/service/payment_fulfillment.go @@ -648,7 +648,11 @@ func (s *PaymentService) applyAffiliateRebateForOrder(ctx context.Context, o *db } sourceOrderID := o.ID - rebateAmount, err := s.affiliateService.AccrueInviteRebateForOrder(txCtx, o.UserID, baseAmount, &sourceOrderID) + rebateAmount, err := s.affiliateService.AccrueInviteRebate(txCtx, o.UserID, AffiliateRebateSource{ + Type: AffiliateRebateSourcePaymentOrder, + BaseAmount: baseAmount, + OrderID: &sourceOrderID, + }) if err != nil { s.writeAuditLog(ctx, o.ID, "AFFILIATE_REBATE_FAILED", "system", map[string]any{ "error": err.Error(), diff --git a/backend/internal/service/payment_fulfillment_test.go b/backend/internal/service/payment_fulfillment_test.go index 50ccd485ed3f..0106db44a27d 100644 --- a/backend/internal/service/payment_fulfillment_test.go +++ b/backend/internal/service/payment_fulfillment_test.go @@ -47,6 +47,8 @@ type paymentFulfillmentAffiliateAccrueCall struct { amount float64 freezeHours int sourceOrderID *int64 + sourceType AffiliateRebateSourceType + baseAmount float64 } type paymentFulfillmentAffiliateRepoStub struct { @@ -76,24 +78,22 @@ func (r *paymentFulfillmentAffiliateRepoStub) BindInviter(context.Context, int64 panic("unexpected BindInviter call") } -func (r *paymentFulfillmentAffiliateRepoStub) AccrueQuota(_ context.Context, inviterID, inviteeUserID int64, amount float64, freezeHours int, sourceOrderID *int64) (bool, error) { +func (r *paymentFulfillmentAffiliateRepoStub) AccrueQuota(_ context.Context, input AffiliateAccrualInput) (float64, error) { var sourceCopy *int64 - if sourceOrderID != nil { - v := *sourceOrderID + if input.Source.OrderID != nil { + v := *input.Source.OrderID sourceCopy = &v } r.accrueCalls = append(r.accrueCalls, paymentFulfillmentAffiliateAccrueCall{ - inviterID: inviterID, - inviteeUserID: inviteeUserID, - amount: amount, - freezeHours: freezeHours, + inviterID: input.InviterID, + inviteeUserID: input.InviteeUserID, + amount: input.Amount, + freezeHours: input.FreezeHours, sourceOrderID: sourceCopy, + sourceType: input.Source.Type, + baseAmount: input.Source.BaseAmount, }) - return true, nil -} - -func (r *paymentFulfillmentAffiliateRepoStub) GetAccruedRebateFromInvitee(context.Context, int64, int64) (float64, error) { - return 0, nil + return input.Amount, nil } func (r *paymentFulfillmentAffiliateRepoStub) ThawFrozenQuota(context.Context, int64) (float64, error) { @@ -959,6 +959,8 @@ func TestExecuteSubscriptionFulfillmentAppliesAffiliateRebate(t *testing.T) { require.Equal(t, inviterID, affiliateRepo.accrueCalls[0].inviterID) require.Equal(t, user.ID, affiliateRepo.accrueCalls[0].inviteeUserID) require.InDelta(t, 1.4985, affiliateRepo.accrueCalls[0].amount, 0.00000001) + require.Equal(t, AffiliateRebateSourcePaymentOrder, affiliateRepo.accrueCalls[0].sourceType) + require.InDelta(t, 9.99, affiliateRepo.accrueCalls[0].baseAmount, 0.00000001) require.NotNil(t, affiliateRepo.accrueCalls[0].sourceOrderID) require.Equal(t, order.ID, *affiliateRepo.accrueCalls[0].sourceOrderID) require.Equal(t, 1, subRepo.createCalls) diff --git a/backend/internal/service/redeem_service.go b/backend/internal/service/redeem_service.go index 8794872e3d1d..8ef4815b2414 100644 --- a/backend/internal/service/redeem_service.go +++ b/backend/internal/service/redeem_service.go @@ -11,7 +11,6 @@ import ( dbent "github.com/Wei-Shaw/sub2api/ent" infraerrors "github.com/Wei-Shaw/sub2api/internal/pkg/errors" - "github.com/Wei-Shaw/sub2api/internal/pkg/logger" "github.com/Wei-Shaw/sub2api/internal/pkg/pagination" ) @@ -508,6 +507,19 @@ func (s *RedeemService) Redeem(ctx context.Context, userID int64, code string) ( return nil, unsupportedRedeemTypeError(redeemCode.Type) } + // 余额兑换和邀请返利共用同一事务，避免余额到账后返利流水遗漏。 + // 支付订单内部兑换会显式跳过这里，随后按订单来源累计，防止重复返利。 + if redeemCode.Type == RedeemTypeBalance && redeemCode.Value > 0 && ctx.Value(ctxKeySkipRedeemAffiliate{}) == nil && s.affiliateService != nil { + redeemCodeID := redeemCode.ID + if _, err := s.affiliateService.AccrueInviteRebate(txCtx, userID, AffiliateRebateSource{ + Type: AffiliateRebateSourceBalanceRedeem, + BaseAmount: redeemCode.Value, + RedeemCodeID: &redeemCodeID, + }); err != nil { + return nil, fmt.Errorf("accrue affiliate rebate for redeem code: %w", err) + } + } + // 提交事务 if err := tx.Commit(); err != nil { return nil, fmt.Errorf("commit transaction: %w", err) @@ -516,11 +528,6 @@ func (s *RedeemService) Redeem(ctx context.Context, userID int64, code string) ( // 事务提交成功后失效缓存 s.invalidateRedeemCaches(ctx, userID, redeemCode) - // 余额类正数兑换码触发邀请返利（best-effort，失败不影响兑换结果） - if redeemCode.Type == RedeemTypeBalance && redeemCode.Value > 0 { - s.tryAccrueAffiliateRebateForRedeem(ctx, userID, redeemCode.Value) - } - // 重新获取更新后的兑换码 redeemCode, err = s.redeemRepo.GetByID(ctx, redeemCode.ID) if err != nil { @@ -570,26 +577,6 @@ func (s *RedeemService) invalidateRedeemCaches(ctx context.Context, userID int64 } } -func (s *RedeemService) tryAccrueAffiliateRebateForRedeem(ctx context.Context, userID int64, amount float64) { - if ctx.Value(ctxKeySkipRedeemAffiliate{}) != nil { - return - } - if s.affiliateService == nil { - return - } - if !s.affiliateService.IsEnabled(ctx) { - return - } - rebate, err := s.affiliateService.AccrueInviteRebate(ctx, userID, amount) - if err != nil { - logger.LegacyPrintf("service.redeem", "[Redeem] affiliate rebate failed for user %d amount %.2f: %v", userID, amount, err) - return - } - if rebate > 0 { - logger.LegacyPrintf("service.redeem", "[Redeem] affiliate rebate accrued %.8f for inviter of user %d", rebate, userID) - } -} - // GetByID 根据ID获取兑换码 func (s *RedeemService) GetByID(ctx context.Context, id int64) (*RedeemCode, error) { code, err := s.redeemRepo.GetByID(ctx, id) diff --git a/backend/migrations/231_affiliate_rebate_sources.sql b/backend/migrations/231_affiliate_rebate_sources.sql new file mode 100644 index 000000000000..490b52c9fb3c --- /dev/null +++ b/backend/migrations/231_affiliate_rebate_sources.sql @@ -0,0 +1,13 @@ +-- 为邀请返利流水补充统一来源字段。本迁移只做快速加列，避免服务启动时扫描和更新整张资金流水表。 +ALTER TABLE user_affiliate_ledger + ADD COLUMN IF NOT EXISTS source_type VARCHAR(32) NULL; + +ALTER TABLE user_affiliate_ledger + ADD COLUMN IF NOT EXISTS base_amount DECIMAL(20,8) NULL; + +ALTER TABLE user_affiliate_ledger + ADD COLUMN IF NOT EXISTS source_redeem_code_id BIGINT NULL; + +COMMENT ON COLUMN user_affiliate_ledger.source_type IS '返利来源：payment_order|balance_redeem_code|admin_recharge|legacy_unknown；历史未分类流水为 NULL'; +COMMENT ON COLUMN user_affiliate_ledger.base_amount IS '计算该笔返利时使用的充值金额快照'; +COMMENT ON COLUMN user_affiliate_ledger.source_redeem_code_id IS '产生返利的余额兑换码或管理员余额调整记录'; diff --git a/backend/migrations/231_affiliate_rebate_sources_indexes_notx.sql b/backend/migrations/231_affiliate_rebate_sources_indexes_notx.sql new file mode 100644 index 000000000000..fb7c99248f72 --- /dev/null +++ b/backend/migrations/231_affiliate_rebate_sources_indexes_notx.sql @@ -0,0 +1,20 @@ +-- 索引必须在事务外并发创建，避免阻塞返利流水写入。 +DROP INDEX CONCURRENTLY IF EXISTS idx_user_affiliate_ledger_source_type_created_at; + +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_user_affiliate_ledger_source_type_created_at + ON user_affiliate_ledger(( + CASE + WHEN source_order_id IS NOT NULL THEN 'payment_order' + WHEN source_type IS NOT NULL THEN source_type + ELSE 'legacy_unknown' + END + ), created_at DESC) + WHERE action = 'accrue'; + +CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS idx_user_affiliate_ledger_accrue_order_uniq + ON user_affiliate_ledger(source_order_id) + WHERE action = 'accrue' AND source_order_id IS NOT NULL; + +CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS idx_user_affiliate_ledger_accrue_redeem_code_uniq + ON user_affiliate_ledger(source_redeem_code_id) + WHERE action = 'accrue' AND source_redeem_code_id IS NOT NULL; diff --git a/backend/migrations/232_affiliate_rebate_source_constraints.sql b/backend/migrations/232_affiliate_rebate_source_constraints.sql new file mode 100644 index 000000000000..d8156605f631 --- /dev/null +++ b/backend/migrations/232_affiliate_rebate_source_constraints.sql @@ -0,0 +1,71 @@ +-- 早期分支版本曾按十分钟时间窗猜测历史管理员充值来源。 +-- 只撤销由该已知旧迁移版本写入的历史分类；正式新流水不做时间推断。 +WITH superseded_migration AS ( + SELECT applied_at + FROM schema_migrations + WHERE filename = '231_affiliate_rebate_sources.sql' + AND checksum = 'ceb508efbf81877a891a95fe6688cb3287462c2552e1a5c8a8254be9328d6806' +) +UPDATE user_affiliate_ledger ual +SET source_type = 'legacy_unknown', + base_amount = NULL, + source_redeem_code_id = NULL, + updated_at = NOW() +FROM superseded_migration migration +WHERE ual.action = 'accrue' + AND ual.source_type = 'admin_recharge' + AND ual.created_at <= migration.applied_at; + +-- NOT VALID 避免部署时扫描历史热表；约束仍会立即校验迁移后的新写入。 +ALTER TABLE user_affiliate_ledger + DROP CONSTRAINT IF EXISTS user_affiliate_ledger_source_order_id_fkey; + +ALTER TABLE user_affiliate_ledger + ADD CONSTRAINT user_affiliate_ledger_source_order_id_fkey + FOREIGN KEY (source_order_id) REFERENCES payment_orders(id) ON DELETE RESTRICT + NOT VALID; + +ALTER TABLE user_affiliate_ledger + DROP CONSTRAINT IF EXISTS user_affiliate_ledger_source_redeem_code_id_fkey; + +ALTER TABLE user_affiliate_ledger + ADD CONSTRAINT user_affiliate_ledger_source_redeem_code_id_fkey + FOREIGN KEY (source_redeem_code_id) REFERENCES redeem_codes(id) ON DELETE RESTRICT + NOT VALID; + +ALTER TABLE user_affiliate_ledger + DROP CONSTRAINT IF EXISTS chk_user_affiliate_ledger_source_type; + +ALTER TABLE user_affiliate_ledger + ADD CONSTRAINT chk_user_affiliate_ledger_source_type CHECK ( + ( + action = 'accrue' + AND ( + source_type IS NULL + OR ( + source_type = 'payment_order' + AND source_order_id IS NOT NULL + AND source_redeem_code_id IS NULL + AND base_amount > 0 + ) + OR ( + source_type IN ('balance_redeem_code', 'admin_recharge') + AND source_order_id IS NULL + AND source_redeem_code_id IS NOT NULL + AND base_amount > 0 + ) + OR ( + source_type = 'legacy_unknown' + AND source_order_id IS NULL + AND source_redeem_code_id IS NULL + ) + ) + ) + OR ( + action <> 'accrue' + AND source_type IS NULL + AND source_order_id IS NULL + AND source_redeem_code_id IS NULL + AND base_amount IS NULL + ) + ) NOT VALID; diff --git a/backend/migrations/affiliate_rebate_sources_migration_test.go b/backend/migrations/affiliate_rebate_sources_migration_test.go new file mode 100644 index 000000000000..b5d2abb0b35b --- /dev/null +++ b/backend/migrations/affiliate_rebate_sources_migration_test.go @@ -0,0 +1,42 @@ +package migrations + +import ( + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +func normalizedMigrationSQL(t *testing.T, name string) string { + t.Helper() + content, err := FS.ReadFile(name) + require.NoError(t, err) + return strings.Join(strings.Fields(string(content)), " ") +} + +func TestAffiliateRebateSourcesMigrationAvoidsHistoricalGuessingAndBlockingIndexes(t *testing.T) { + coreSQL := normalizedMigrationSQL(t, "231_affiliate_rebate_sources.sql") + require.Contains(t, coreSQL, "ADD COLUMN IF NOT EXISTS source_type VARCHAR(32) NULL") + require.Contains(t, coreSQL, "ADD COLUMN IF NOT EXISTS base_amount DECIMAL(20,8) NULL") + require.Contains(t, coreSQL, "ADD COLUMN IF NOT EXISTS source_redeem_code_id BIGINT NULL") + require.NotContains(t, strings.ToUpper(coreSQL), "UPDATE USER_AFFILIATE_LEDGER") + require.NotContains(t, strings.ToUpper(coreSQL), "CREATE INDEX") + require.NotContains(t, coreSQL, "INTERVAL '10 minutes'") + + indexSQL := normalizedMigrationSQL(t, "231_affiliate_rebate_sources_indexes_notx.sql") + require.Contains(t, indexSQL, "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_user_affiliate_ledger_source_type_created_at") + require.Contains(t, indexSQL, "CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS idx_user_affiliate_ledger_accrue_order_uniq") + require.Contains(t, indexSQL, "CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS idx_user_affiliate_ledger_accrue_redeem_code_uniq") + + constraintSQL := normalizedMigrationSQL(t, "232_affiliate_rebate_source_constraints.sql") + require.Contains(t, constraintSQL, "checksum = 'ceb508efbf81877a891a95fe6688cb3287462c2552e1a5c8a8254be9328d6806'") + require.Contains(t, constraintSQL, "source_type = 'admin_recharge'") + require.Contains(t, constraintSQL, "source_type = 'legacy_unknown'") + require.Contains(t, constraintSQL, "FOREIGN KEY (source_order_id) REFERENCES payment_orders(id) ON DELETE RESTRICT NOT VALID") + require.Contains(t, constraintSQL, "FOREIGN KEY (source_redeem_code_id) REFERENCES redeem_codes(id) ON DELETE RESTRICT NOT VALID") + require.Contains(t, constraintSQL, "source_type IN ('balance_redeem_code', 'admin_recharge')") + require.Contains(t, constraintSQL, "source_redeem_code_id IS NOT NULL") + require.Contains(t, constraintSQL, ") NOT VALID") + require.NotContains(t, constraintSQL, "INTERVAL '10 minutes'") + require.NotContains(t, strings.ToUpper(constraintSQL), "DELETE FROM USER_AFFILIATE_LEDGER") +} diff --git a/frontend/src/api/admin/affiliates.ts b/frontend/src/api/admin/affiliates.ts index dadb0ae9e5f7..deb381e02fa2 100644 --- a/frontend/src/api/admin/affiliates.ts +++ b/frontend/src/api/admin/affiliates.ts @@ -32,8 +32,17 @@ export interface ListAffiliateRecordsParams { sort_by?: string sort_order?: 'asc' | 'desc' timezone?: string + source_type?: AffiliateRebateSourceFilter } +export type AffiliateRebateSourceType = + | 'payment_order' + | 'balance_redeem_code' + | 'admin_recharge' + | 'legacy_unknown' + +export type AffiliateRebateSourceFilter = AffiliateRebateSourceType | 'all' + export interface AffiliateInviteRecord { inviter_id: number inviter_email: string @@ -47,19 +56,24 @@ export interface AffiliateInviteRecord { } export interface AffiliateRebateRecord { - order_id: number - out_trade_no: string + ledger_id: number + source_type: AffiliateRebateSourceType + order_id?: number | null + out_trade_no?: string | null + redeem_code_id?: number | null + redeem_code_masked?: string | null inviter_id: number inviter_email: string inviter_username: string invitee_id: number invitee_email: string invitee_username: string - order_amount: number - pay_amount: number + base_amount?: number | null + order_amount?: number | null + pay_amount?: number | null rebate_amount: number - payment_type: string - order_status: string + payment_type?: string | null + order_status?: string | null created_at: string } @@ -173,6 +187,7 @@ function recordParams(params: ListAffiliateRecordsParams = {}) { sort_by: params.sort_by || undefined, sort_order: params.sort_order || undefined, timezone: params.timezone || undefined, + source_type: params.source_type || undefined, } } diff --git a/frontend/src/i18n/locales/en/admin/overview.ts b/frontend/src/i18n/locales/en/admin/overview.ts index e91233132d66..ccf52cb200af 100644 --- a/frontend/src/i18n/locales/en/admin/overview.ts +++ b/frontend/src/i18n/locales/en/admin/overview.ts @@ -390,7 +390,7 @@ export default { affiliates: { invitesDescription: 'View site-wide inviter and invitee relationships', - rebatesDescription: 'View recharge orders that generated affiliate rebates', + rebatesDescription: 'View affiliate rebates generated by payments, balance redeem codes, or admin recharges', transfersDescription: 'View affiliate quota transfers into account balance', errors: { loadFailed: 'Failed to load affiliate records' @@ -398,8 +398,25 @@ export default { records: { search: 'Search', searchPlaceholder: 'Email, username, user ID, or order number', + rebateSearchPlaceholder: 'Email, username, user ID, order number, or redeem code', startAt: 'Start date', endAt: 'End date', + sourceFilter: 'Rebate source', + source: 'Source', + sourceReference: 'Source reference', + baseAmount: 'Recharge amount', + sourceStatus: 'Source status', + sourceTypes: { + all: 'All sources', + payment_order: 'Payment order', + balance_redeem_code: 'Balance redeem code', + admin_recharge: 'Admin recharge', + legacy_unknown: 'Legacy unknown' + }, + sourceStatuses: { + redeemed: 'Redeemed', + credited: 'Credited' + }, inviter: 'Inviter', invitee: 'Invitee', user: 'User', diff --git a/frontend/src/i18n/locales/zh/admin/overview.ts b/frontend/src/i18n/locales/zh/admin/overview.ts index cbc930fa1043..fb70917f6f5a 100644 --- a/frontend/src/i18n/locales/zh/admin/overview.ts +++ b/frontend/src/i18n/locales/zh/admin/overview.ts @@ -390,7 +390,7 @@ export default { affiliates: { invitesDescription: '查看全站邀请关系和被邀请用户累计返利', - rebatesDescription: '查看每一笔产生返利的充值订单', + rebatesDescription: '查看每一笔支付、余额兑换码或管理员充值产生的邀请返利', transfersDescription: '查看返利额度转入账户余额的提取流水', errors: { loadFailed: '加载邀请返利记录失败' @@ -398,8 +398,25 @@ export default { records: { search: '搜索', searchPlaceholder: '邮箱、用户名、用户 ID、订单号', + rebateSearchPlaceholder: '邮箱、用户名、用户 ID、订单号、兑换码', startAt: '开始日期', endAt: '结束日期', + sourceFilter: '返利来源', + source: '来源', + sourceReference: '来源记录', + baseAmount: '充值金额', + sourceStatus: '来源状态', + sourceTypes: { + all: '全部来源', + payment_order: '支付订单', + balance_redeem_code: '余额兑换码', + admin_recharge: '管理员充值', + legacy_unknown: '历史未知' + }, + sourceStatuses: { + redeemed: '已兑换', + credited: '已入账' + }, inviter: '邀请人', invitee: '被邀请人', user: '用户', diff --git a/frontend/src/views/admin/__tests__/AdminAffiliateRecordsTable.spec.ts b/frontend/src/views/admin/__tests__/AdminAffiliateRecordsTable.spec.ts new file mode 100644 index 000000000000..801468d6b506 --- /dev/null +++ b/frontend/src/views/admin/__tests__/AdminAffiliateRecordsTable.spec.ts @@ -0,0 +1,213 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' +import { flushPromises, mount } from '@vue/test-utils' + +import AdminAffiliateRecordsTable from '../affiliates/AdminAffiliateRecordsTable.vue' +import Select from '@/components/common/Select.vue' + +const { listInviteRecords, listRebateRecords, listTransferRecords, getUserOverview, showError } = vi.hoisted(() => ({ + listInviteRecords: vi.fn(), + listRebateRecords: vi.fn(), + listTransferRecords: vi.fn(), + getUserOverview: vi.fn(), + showError: vi.fn() +})) + +vi.mock('@/api/admin/affiliates', () => { + const affiliatesAPI = { + listInviteRecords, + listRebateRecords, + listTransferRecords, + getUserOverview + } + + return { + affiliatesAPI, + default: affiliatesAPI + } +}) + +vi.mock('@/stores/app', () => ({ + useAppStore: () => ({ showError }) +})) + +vi.mock('vue-i18n', async () => { + const actual = await vi.importActual('vue-i18n') + return { + ...actual, + useI18n: () => ({ + t: (key: string, fallback?: string) => fallback || key + }) + } +}) + +const DataTableStub = { + props: ['columns', 'data'], + template: ` + + + + + + +
+ + {{ row[column.key] }} + +
+ ` +} + +function rebateResponse(ledgerId: number, redeemCodeMasked: string) { + return { + items: [ + { + ledger_id: ledgerId, + source_type: 'balance_redeem_code', + redeem_code_id: 18, + redeem_code_masked: redeemCodeMasked, + inviter_id: 60, + inviter_email: 'inviter@example.com', + inviter_username: 'inviter', + invitee_id: 81, + invitee_email: 'invitee@example.com', + invitee_username: 'invitee', + base_amount: 100, + rebate_amount: 20, + created_at: '2026-08-27T10:00:00Z' + } + ], + total: 1, + page: 1, + page_size: 20, + pages: 1 + } +} + +describe('管理端邀请返利来源筛选', () => { + beforeEach(() => { + localStorage.clear() + listInviteRecords.mockReset() + listRebateRecords.mockReset() + listTransferRecords.mockReset() + getUserOverview.mockReset() + showError.mockReset() + + listRebateRecords.mockResolvedValue(rebateResponse(91, 'abcd****1234')) + }) + + it('默认查询全部来源，并可切换到余额兑换码来源', async () => { + const wrapper = mount(AdminAffiliateRecordsTable, { + props: { type: 'rebates' }, + global: { + stubs: { + AppLayout: { template: '
' }, + TablePageLayout: { template: '
' }, + DataTable: DataTableStub, + Pagination: true, + BaseDialog: true, + Icon: true, + OrderStatusBadge: true + } + } + }) + + await flushPromises() + expect(listRebateRecords).toHaveBeenNthCalledWith(1, expect.objectContaining({ + page: 1, + source_type: 'all' + })) + expect(wrapper.text()).toContain('abcd****1234') + + const sourceFilter = wrapper.getComponent(Select) + sourceFilter.vm.$emit('update:modelValue', 'balance_redeem_code') + sourceFilter.vm.$emit('change', 'balance_redeem_code') + await flushPromises() + + expect(listRebateRecords).toHaveBeenLastCalledWith(expect.objectContaining({ + page: 1, + source_type: 'balance_redeem_code' + })) + }) + + it('余额兑换码和管理员充值使用统一的成功状态徽标', async () => { + const response = rebateResponse(91, 'abcd****1234') + response.items.push({ + ...response.items[0], + ledger_id: 92, + source_type: 'admin_recharge' + }) + response.total = 2 + listRebateRecords.mockResolvedValueOnce(response) + + const wrapper = mount(AdminAffiliateRecordsTable, { + props: { type: 'rebates' }, + global: { + stubs: { + AppLayout: { template: '
' }, + TablePageLayout: { template: '
' }, + DataTable: DataTableStub, + Pagination: true, + BaseDialog: true, + Icon: true, + OrderStatusBadge: true + } + } + }) + + await flushPromises() + + const successBadgeClasses = [ + 'inline-flex', + 'rounded-full', + 'bg-green-100', + 'text-green-800', + 'dark:bg-green-900/30', + 'dark:text-green-400' + ] + for (const label of [ + 'admin.affiliates.records.sourceStatuses.redeemed', + 'admin.affiliates.records.sourceStatuses.credited' + ]) { + const badge = wrapper.findAll('span').find((node) => node.text() === label) + expect(badge).toBeDefined() + expect(badge!.classes()).toEqual(expect.arrayContaining(successBadgeClasses)) + } + }) + + it('旧筛选的慢响应不会覆盖新筛选结果', async () => { + let resolveFirstRequest!: (value: ReturnType) => void + listRebateRecords + .mockReset() + .mockImplementationOnce(() => new Promise((resolve) => { + resolveFirstRequest = resolve + })) + .mockResolvedValueOnce(rebateResponse(102, 'new****2222')) + + const wrapper = mount(AdminAffiliateRecordsTable, { + props: { type: 'rebates' }, + global: { + stubs: { + AppLayout: { template: '
' }, + TablePageLayout: { template: '
' }, + DataTable: DataTableStub, + Pagination: true, + BaseDialog: true, + Icon: true, + OrderStatusBadge: true + } + } + }) + + await vi.waitFor(() => expect(listRebateRecords).toHaveBeenCalledTimes(1)) + const sourceFilter = wrapper.getComponent(Select) + sourceFilter.vm.$emit('update:modelValue', 'balance_redeem_code') + sourceFilter.vm.$emit('change', 'balance_redeem_code') + await flushPromises() + expect(wrapper.text()).toContain('new****2222') + + resolveFirstRequest(rebateResponse(101, 'old****1111')) + await flushPromises() + expect(wrapper.text()).toContain('new****2222') + expect(wrapper.text()).not.toContain('old****1111') + }) +}) diff --git a/frontend/src/views/admin/affiliates/AdminAffiliateRecordsTable.vue b/frontend/src/views/admin/affiliates/AdminAffiliateRecordsTable.vue index 789df41a0dd5..2e19a934dfe2 100644 --- a/frontend/src/views/admin/affiliates/AdminAffiliateRecordsTable.vue +++ b/frontend/src/views/admin/affiliates/AdminAffiliateRecordsTable.vue @@ -5,8 +5,17 @@
- +
+