From 24a1c1e3cc9d9d6a1c155d3f70d61f9bb110292e Mon Sep 17 00:00:00 2001 From: qian Date: Sun, 7 Jun 2026 13:23:19 +0800 Subject: [PATCH] =?UTF-8?q?fix(open):=E4=BF=AE=E5=A4=8D=E9=92=B1=E5=8C=85?= =?UTF-8?q?=E4=BD=99=E9=A2=9D=E4=B8=8D=E8=B6=B3=E6=97=B6webhook=E5=91=8A?= =?UTF-8?q?=E8=AD=A6=E5=8A=9F=E8=83=BD=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/dao/internal/aud_request_logs.go | 2 + internal/dao/internal/bil_wallets.go | 2 + internal/dao/internal/ord_orders.go | 2 +- internal/logic/admin/dashboard.go | 13 +- internal/logic/billing/alert.go | 168 ++++++++++++------ internal/logic/billing/settlement.go | 7 + internal/logic/billing/task_billing.go | 5 +- internal/logic/billing/wallet.go | 41 +++-- internal/logic/common/notification.go | 7 + internal/logic/payment/fulfill.go | 2 + internal/logic/tenant/billing.go | 3 + internal/logic/tenant/open_platform.go | 31 ++-- internal/logic/tenant/redemption.go | 3 + internal/model/do/aud_request_logs.go | 1 + internal/model/do/bil_wallets.go | 1 + internal/model/do/ord_orders.go | 2 +- internal/model/entity/aud_request_logs.go | 1 + internal/model/entity/bil_wallets.go | 19 +- internal/model/entity/ord_orders.go | 2 +- migrations/000007_v0_1_7_version.sql | 61 +++++++ .../000007_webhook_indexes_and_cleanup.sql | 31 ---- .../views/notification/NotificationsPage.vue | 62 +++---- web/tenant/src/views/open/WebhooksPage.vue | 90 ++++++++++ 23 files changed, 388 insertions(+), 168 deletions(-) create mode 100644 migrations/000007_v0_1_7_version.sql delete mode 100644 migrations/000007_webhook_indexes_and_cleanup.sql diff --git a/internal/dao/internal/aud_request_logs.go b/internal/dao/internal/aud_request_logs.go index f62e8d1..8cb94c2 100644 --- a/internal/dao/internal/aud_request_logs.go +++ b/internal/dao/internal/aud_request_logs.go @@ -51,6 +51,7 @@ type AudRequestLogsColumns struct { TaskResult string // 异步任务完成时上游返回的原始响应体 TaskUpstreamHeaders string // 异步任务完成时上游返回的响应头（仅审计级别为 full 时记录） TaskCompletedAt string // 异步任务达到终态的时间 + AttachmentIds string // 提取的媒体附件 fil_files ID 数组（JSONB: [1,2,3]） } // audRequestLogsColumns holds the columns for the table aud_request_logs. @@ -85,6 +86,7 @@ var audRequestLogsColumns = AudRequestLogsColumns{ TaskResult: "task_result", TaskUpstreamHeaders: "task_upstream_headers", TaskCompletedAt: "task_completed_at", + AttachmentIds: "attachment_ids", } // NewAudRequestLogsDao creates and returns a new DAO object for table data access. diff --git a/internal/dao/internal/bil_wallets.go b/internal/dao/internal/bil_wallets.go index 8ff62d3..2d84fa0 100644 --- a/internal/dao/internal/bil_wallets.go +++ b/internal/dao/internal/bil_wallets.go @@ -30,6 +30,7 @@ type BilWalletsColumns struct { CreatedAt string // 创建时间 UpdatedAt string // 更新时间 CumulativeRecharge string // 累计充值总额（USD） + LowBalanceNotified string // 低余额预警是否已推送（充值恢复后重置为 false） } // bilWalletsColumns holds the columns for the table bil_wallets. @@ -43,6 +44,7 @@ var bilWalletsColumns = BilWalletsColumns{ CreatedAt: "created_at", UpdatedAt: "updated_at", CumulativeRecharge: "cumulative_recharge", + LowBalanceNotified: "low_balance_notified", } // NewBilWalletsDao creates and returns a new DAO object for table data access. diff --git a/internal/dao/internal/ord_orders.go b/internal/dao/internal/ord_orders.go index eb4f0e1..8f01778 100644 --- a/internal/dao/internal/ord_orders.go +++ b/internal/dao/internal/ord_orders.go @@ -30,7 +30,7 @@ type OrdOrdersColumns struct { Amount string // 原始金额 DiscountAmount string // 优惠金额 FinalAmount string // 最终金额 - Currency string // 货币 + Currency string // 货币（订单层一律 CNY） PaymentChannel string // 支付渠道 PaymentMethod string // 支付方式描述 PaymentNo string // 第三方支付流水号 diff --git a/internal/logic/admin/dashboard.go b/internal/logic/admin/dashboard.go index 757fe84..e2b83d8 100644 --- a/internal/logic/admin/dashboard.go +++ b/internal/logic/admin/dashboard.go @@ -13,6 +13,7 @@ import ( v1 "github.com/qianfree/team-api/api/admin/v1" "github.com/qianfree/team-api/internal/dao" + "github.com/qianfree/team-api/internal/logic/billing" "github.com/qianfree/team-api/internal/logic/common" do "github.com/qianfree/team-api/internal/model/do" "github.com/qianfree/team-api/internal/utility/export" @@ -446,6 +447,9 @@ func (s *sAdmin) AdjustBalance(ctx context.Context, req *v1.AdminWalletAdjustReq return nil, err } + // 管理员调整余额后，重置低余额预警标记（可能余额已恢复到阈值以上） + billing.ResetLowBalanceNotified(ctx, tenantID) + return &v1.AdminWalletAdjustRes{}, nil } @@ -530,7 +534,14 @@ func (s *sAdmin) SetWarningThreshold(ctx context.Context, req *v1.AdminWalletSet Data(do.BilWallets{ WarningThreshold: req.Threshold, }).Update() - return &v1.AdminWalletSetWarningThresholdRes{}, err + if err != nil { + return nil, err + } + + // 阈值变更后重置预警标记，使新阈值能触发新的预警 + billing.ResetLowBalanceNotified(ctx, req.TenantID) + + return &v1.AdminWalletSetWarningThresholdRes{}, nil } // GetDashboardChannelHealth 获取渠道健康概览（最不健康的5个活跃渠道） diff --git a/internal/logic/billing/alert.go b/internal/logic/billing/alert.go index 6b4363e..3b81460 100644 --- a/internal/logic/billing/alert.go +++ b/internal/logic/billing/alert.go @@ -18,67 +18,121 @@ func SetPublishWebhookEventFn(fn func(ctx context.Context, tenantID int64, event publishWebhookEventFn = fn } -// CheckBalanceWarning 检查余额预警 -// 如果租户可用余额低于预警线，发送通知给租户管理员并投递 webhook 事件 -func CheckBalanceWarning(ctx context.Context, tenantID int64) (bool, float64, float64, error) { +// CheckBalanceWarning 检查余额预警（事件驱动，由结算后调用） +// 仅在余额 ≤ 阈值 且尚未推送过时触发一次通知 + webhook，避免重复推送 +// 充值恢复后由 ResetLowBalanceNotified 重置标记，允许下次再触发 +func CheckBalanceWarning(ctx context.Context, tenantID int64) { wallet, err := GetWallet(ctx, tenantID) if err != nil { - return false, 0, 0, err - } - - available := AvailableBalance(wallet) - if wallet.WarningThreshold > 0 && available <= wallet.WarningThreshold { - g.Log().Warningf(ctx, "[BALANCE WARNING] tenant=%d available=%.6f threshold=%.6f", - tenantID, available, wallet.WarningThreshold) - - // 发送站内通知 + webhook 事件（异步，失败不影响主流程） - go func() { - bgCtx := context.Background() - - // 站内通知 - engine := common.NewNotificationEngine() - variables := map[string]any{ - "available": fmt.Sprintf("%.6f", available), - "threshold": fmt.Sprintf("%.6f", wallet.WarningThreshold), - } - if notifyErr := engine.SendBroadcast(bgCtx, tenantID, "balance_warning", variables, "owner,admin"); notifyErr != nil { - g.Log().Errorf(bgCtx, "[BALANCE WARNING] send notification failed: tenant=%d err=%v", tenantID, notifyErr) - } - - // Webhook 事件投递 - if publishWebhookEventFn != nil { - if webhookErr := publishWebhookEventFn(bgCtx, tenantID, "wallet.low_balance", map[string]any{ - "available_balance": available, - "warning_threshold": wallet.WarningThreshold, - "wallet_id": wallet.ID, - }); webhookErr != nil { - g.Log().Errorf(bgCtx, "[BALANCE WARNING] publish webhook event failed: tenant=%d err=%v", tenantID, webhookErr) - } - } - }() - - return true, available, wallet.WarningThreshold, nil - } - - return false, available, wallet.WarningThreshold, nil -} - -// CheckAllBalanceWarnings 检查所有租户余额预警 -func CheckAllBalanceWarnings(ctx context.Context) { - var tenants []struct { - TenantID int64 `json:"tenant_id"` - } - - err := dao.TntTenants.Ctx(ctx). - Where("status", "active"). - Fields("id as tenant_id"). - Scan(&tenants) - if err != nil { - g.Log().Errorf(ctx, "check balance warnings: query tenants failed: %v", err) + g.Log().Warningf(ctx, "[BALANCE WARNING] get wallet failed: tenant=%d err=%v", tenantID, err) return } - for _, t := range tenants { - _, _, _, _ = CheckBalanceWarning(ctx, t.TenantID) + available := AvailableBalance(wallet) + + // 阈值为 0 表示未启用预警 + if wallet.WarningThreshold <= 0 { + return } + + // 余额仍高于阈值，无需预警 + if available > wallet.WarningThreshold { + return + } + + // 已推送过，跳过（等充值恢复后重置标记才能再次推送） + if wallet.LowBalanceNotified { + return + } + + g.Log().Warningf(ctx, "[BALANCE WARNING] tenant=%d available=%.6f threshold=%.6f (first trigger)", + tenantID, available, wallet.WarningThreshold) + + // 标记已推送（先写 DB，防止并发重复推送） + _, err = g.DB().Ctx(ctx).Exec(ctx, + "UPDATE bil_wallets SET low_balance_notified = true, updated_at = NOW() WHERE id = ?", wallet.ID) + if err != nil { + g.Log().Errorf(ctx, "[BALANCE WARNING] mark notified failed: tenant=%d err=%v", tenantID, err) + return + } + // 清除缓存，确保下次读取到最新状态 + walletCache.Delete(ctx, fmt.Sprintf("%d", tenantID)) + + // 异步发送站内通知 + webhook 事件（失败不影响主流程） + go func() { + bgCtx := context.Background() + + // 查询租户信息（用于通知内容） + var tenantInfo *struct { + Code string `json:"code"` + Name string `json:"name"` + } + _ = dao.TntTenants.Ctx(bgCtx). + Where("id", tenantID). + Fields("code, name"). + Scan(&tenantInfo) + + // 站内通知 + engine := common.NewNotificationEngine() + variables := map[string]any{ + "tenant_code": "", + "tenant_name": "", + "available": fmt.Sprintf("%.6f", available), + "threshold": fmt.Sprintf("%.6f", wallet.WarningThreshold), + } + if tenantInfo != nil { + variables["tenant_code"] = tenantInfo.Code + variables["tenant_name"] = tenantInfo.Name + } + if notifyErr := engine.SendBroadcast(bgCtx, tenantID, "balance_warning", variables, "owner,admin"); notifyErr != nil { + g.Log().Errorf(bgCtx, "[BALANCE WARNING] send notification failed: tenant=%d err=%v", tenantID, notifyErr) + } + + // Webhook 事件投递 + if publishWebhookEventFn != nil { + payload := map[string]any{ + "available_balance": available, + "warning_threshold": wallet.WarningThreshold, + "wallet_id": wallet.ID, + } + if tenantInfo != nil { + payload["tenant_code"] = tenantInfo.Code + payload["tenant_name"] = tenantInfo.Name + } + if webhookErr := publishWebhookEventFn(bgCtx, tenantID, "wallet.low_balance", payload); webhookErr != nil { + g.Log().Errorf(bgCtx, "[BALANCE WARNING] publish webhook event failed: tenant=%d err=%v", tenantID, webhookErr) + } + } + }() +} + +// ResetLowBalanceNotified 充值后检查：如果余额已恢复到阈值以上，重置 low_balance_notified 标记 +// 允许下次余额再低于阈值时重新触发预警 +func ResetLowBalanceNotified(ctx context.Context, tenantID int64) { + wallet, err := GetWallet(ctx, tenantID) + if err != nil { + return + } + + // 没有被标记过，无需重置 + if !wallet.LowBalanceNotified { + return + } + + // 余额仍低于阈值，不重置 + available := AvailableBalance(wallet) + if available <= wallet.WarningThreshold { + return + } + + g.Log().Infof(ctx, "[BALANCE WARNING] tenant=%d balance restored (%.6f > threshold %.6f), resetting notified flag", + tenantID, available, wallet.WarningThreshold) + + _, err = g.DB().Ctx(ctx).Exec(ctx, + "UPDATE bil_wallets SET low_balance_notified = false, updated_at = NOW() WHERE id = ?", wallet.ID) + if err != nil { + g.Log().Errorf(ctx, "[BALANCE WARNING] reset notified failed: tenant=%d err=%v", tenantID, err) + return + } + walletCache.Delete(ctx, fmt.Sprintf("%d", tenantID)) } diff --git a/internal/logic/billing/settlement.go b/internal/logic/billing/settlement.go index 53706c0..147dabc 100644 --- a/internal/logic/billing/settlement.go +++ b/internal/logic/billing/settlement.go @@ -152,6 +152,9 @@ func Settle(ctx context.Context, tenantID, userID, apiKeyID, channelID int64, InvalidateWalletRedis(ctx, tenantID) CleanupPreDeduct(ctx, tenantID, requestID) + // 7. 异步检查余额预警 + go CheckBalanceWarning(context.Background(), tenantID) + return &SettlementResult{ PreDeductAmount: preDeductAmount, BaseCost: breakdown.BaseCost, @@ -284,6 +287,10 @@ func SettleWithUsage(ctx context.Context, tenantID, userID, apiKeyID, channelID walletCache.Delete(ctx, fmt.Sprintf("%d", tenantID)) InvalidateWalletRedis(ctx, tenantID) CleanupPreDeduct(ctx, tenantID, requestID) + + // 7. 异步检查余额预警 + go CheckBalanceWarning(context.Background(), tenantID) + settlementResult := &SettlementResult{ PreDeductAmount: preDeductAmount, ActualCost: actualCost, diff --git a/internal/logic/billing/task_billing.go b/internal/logic/billing/task_billing.go index 48c8993..b473f7f 100644 --- a/internal/logic/billing/task_billing.go +++ b/internal/logic/billing/task_billing.go @@ -214,7 +214,10 @@ func (b *TaskBillingProviderImpl) SettleTaskSuccess(ctx context.Context, tenantI CleanupPreDeduct(ctx, tenantID, requestID) CleanupPreDeduct(ctx, tenantID, requestID+"_adjust") - // 5. 差额处理：实际 < 预扣时退还差额（实际 > 预扣时已在步骤 3 扣完，无需额外操作） + // 5. 异步检查余额预警 + go CheckBalanceWarning(context.Background(), tenantID) + + // 6. 差额处理：实际 < 预扣时退还差额（实际 > 预扣时已在步骤 3 扣完，无需额外操作） if diff < -0.001 { if err := SettleFailed(ctx, tenantID, requestID+"_adjust", -diff); err != nil { g.Log().Warningf(ctx, "settle task adjust refund failed for %s: %v", requestID, err) diff --git a/internal/logic/billing/wallet.go b/internal/logic/billing/wallet.go index 810ef5b..58aa523 100644 --- a/internal/logic/billing/wallet.go +++ b/internal/logic/billing/wallet.go @@ -30,12 +30,13 @@ var walletSyncGroup singleflight.Group // WalletInfo 钱包信息 type WalletInfo struct { - ID int64 - TenantID int64 - Balance float64 - FrozenBalance float64 - WarningThreshold float64 - Currency string + ID int64 + TenantID int64 + Balance float64 + FrozenBalance float64 + WarningThreshold float64 + Currency string + LowBalanceNotified bool } // GetWallet 获取租户钱包 @@ -47,18 +48,19 @@ func GetWallet(ctx context.Context, tenantID int64) (*WalletInfo, error) { } type walletRow struct { - ID int64 `json:"id"` - TenantId int64 `json:"tenant_id"` - Balance float64 `json:"balance"` - FrozenBalance float64 `json:"frozen_balance"` - WarningThreshold float64 `json:"warning_threshold"` - Currency string `json:"currency"` + ID int64 `json:"id"` + TenantId int64 `json:"tenant_id"` + Balance float64 `json:"balance"` + FrozenBalance float64 `json:"frozen_balance"` + WarningThreshold float64 `json:"warning_threshold"` + Currency string `json:"currency"` + LowBalanceNotified bool `json:"low_balance_notified"` } var w *walletRow err := dao.BilWallets.Ctx(ctx). Where("tenant_id", tenantID). - Fields("id, tenant_id, balance, frozen_balance, warning_threshold, currency"). + Fields("id, tenant_id, balance, frozen_balance, warning_threshold, currency, low_balance_notified"). Scan(&w) if err != nil { return nil, gerror.Wrapf(err, "query wallet") @@ -68,12 +70,13 @@ func GetWallet(ctx context.Context, tenantID int64) (*WalletInfo, error) { } info := &WalletInfo{ - ID: w.ID, - TenantID: w.TenantId, - Balance: w.Balance, - FrozenBalance: w.FrozenBalance, - WarningThreshold: w.WarningThreshold, - Currency: w.Currency, + ID: w.ID, + TenantID: w.TenantId, + Balance: w.Balance, + FrozenBalance: w.FrozenBalance, + WarningThreshold: w.WarningThreshold, + Currency: w.Currency, + LowBalanceNotified: w.LowBalanceNotified, } walletCache.Set(ctx, cacheKey, info) diff --git a/internal/logic/common/notification.go b/internal/logic/common/notification.go index 6c81d11..c39f8d6 100644 --- a/internal/logic/common/notification.go +++ b/internal/logic/common/notification.go @@ -3,6 +3,7 @@ package common import ( "context" "encoding/json" + "fmt" "github.com/qianfree/team-api/internal/dao" do "github.com/qianfree/team-api/internal/model/do" @@ -418,11 +419,17 @@ func (e *NotificationEngine) buildMetadata(templateCode string, variables map[st // logNotification records the notification dispatch in ntf_send_log. func (e *NotificationEngine) logNotification(ctx context.Context, tenantID, userID int64, templateCode, subject, body string) { + recipient := fmt.Sprintf("in_app:tenant:%d", tenantID) + if userID > 0 { + recipient = fmt.Sprintf("in_app:tenant:%d:user:%d", tenantID, userID) + } + _, err := dao.NtfSendLog.Ctx(ctx).Insert(do.NtfSendLog{ TenantId: tenantID, UserId: userID, TemplateCode: templateCode, Channel: "in_app", + Recipient: recipient, Subject: subject, Body: body, Status: "sent", diff --git a/internal/logic/payment/fulfill.go b/internal/logic/payment/fulfill.go index 88d9c2c..f160ceb 100644 --- a/internal/logic/payment/fulfill.go +++ b/internal/logic/payment/fulfill.go @@ -52,6 +52,8 @@ func FulfillOrder(ctx context.Context, orderID int64) error { return gerror.Wrapf(err, "credit wallet failed") } _ = billing.CheckAndUpgradeLevel(ctx, order.TenantID) + // 充值后检查是否需要重置低余额预警标记 + billing.ResetLowBalanceNotified(ctx, order.TenantID) default: return gerror.Newf("unsupported order type for fulfillment: %s", order.OrderType) diff --git a/internal/logic/tenant/billing.go b/internal/logic/tenant/billing.go index e8e4151..4050619 100644 --- a/internal/logic/tenant/billing.go +++ b/internal/logic/tenant/billing.go @@ -529,5 +529,8 @@ func (s *sTenant) UpdateWarningThreshold(ctx context.Context, req *v1.TenantWall // 清除钱包缓存 billing.InvalidateWalletRedis(ctx, tenantID) + // 阈值变更后重置预警标记，使新阈值能触发新的预警 + billing.ResetLowBalanceNotified(ctx, tenantID) + return &v1.TenantWalletUpdateWarningThresholdRes{}, nil } diff --git a/internal/logic/tenant/open_platform.go b/internal/logic/tenant/open_platform.go index 87b3621..2f6230e 100644 --- a/internal/logic/tenant/open_platform.go +++ b/internal/logic/tenant/open_platform.go @@ -357,8 +357,8 @@ func (s *sTenant) WebhookConfigUpdate(ctx context.Context, req *v1.WebhookConfig hasUpdate = true } if req.URL != nil { - if !strings.HasPrefix(*req.URL, "https://") { - return nil, common.NewBadRequestError("回调 URL 必须以 https:// 开头") + if !strings.HasPrefix(*req.URL, "https://") && !strings.HasPrefix(*req.URL, "http://") { + return nil, common.NewBadRequestError("回调 URL 必须以 http:// 或 https:// 开头") } data.Url = *req.URL hasUpdate = true @@ -523,21 +523,28 @@ func PublishWebhookEvent(ctx context.Context, tenantID int64, eventType string, return err } - // 使用 json.Marshal 安全构建 JSONB 查询参数，防止注入 - eventFilter, err := json.Marshal([]string{eventType}) + // Find all active webhook configs for this tenant, then filter by subscribed events in Go + var allConfigs []entity.OpnWebhookConfigs + err = dao.OpnWebhookConfigs.Ctx(ctx). + Where("tenant_id", tenantID). + Where("is_active", true). + Scan(&allConfigs) if err != nil { return err } - // Find all active webhook configs that subscribe to this event type var configs []entity.OpnWebhookConfigs - err = dao.OpnWebhookConfigs.Ctx(ctx). - Where("tenant_id", tenantID). - Where("is_active", true). - Where("events::jsonb @> ?", string(eventFilter)). - Scan(&configs) - if err != nil { - return err + for _, c := range allConfigs { + var events []string + if err := json.Unmarshal([]byte(c.Events), &events); err != nil { + continue + } + for _, e := range events { + if e == eventType { + configs = append(configs, c) + break + } + } } for _, config := range configs { diff --git a/internal/logic/tenant/redemption.go b/internal/logic/tenant/redemption.go index b8c0d27..e557ec8 100644 --- a/internal/logic/tenant/redemption.go +++ b/internal/logic/tenant/redemption.go @@ -151,6 +151,9 @@ func (s *sTenant) RedeemCode(ctx context.Context, req *v1.TenantRedeemCodeReq) ( // 事务提交后清除 Redis 钱包缓存 billing.InvalidateWalletRedis(ctx, tenantID) + // 兑换码充值后，重置低余额预警标记（余额可能已恢复到阈值以上） + billing.ResetLowBalanceNotified(ctx, tenantID) + return res, nil } diff --git a/internal/model/do/aud_request_logs.go b/internal/model/do/aud_request_logs.go index 3cd909a..6d6dbbd 100644 --- a/internal/model/do/aud_request_logs.go +++ b/internal/model/do/aud_request_logs.go @@ -42,4 +42,5 @@ type AudRequestLogs struct { TaskResult any // 异步任务完成时上游返回的原始响应体 TaskUpstreamHeaders any // 异步任务完成时上游返回的响应头（仅审计级别为 full 时记录） TaskCompletedAt *gtime.Time // 异步任务达到终态的时间 + AttachmentIds any // 提取的媒体附件 fil_files ID 数组（JSONB: [1,2,3]） } diff --git a/internal/model/do/bil_wallets.go b/internal/model/do/bil_wallets.go index 9a090fc..0c752e2 100644 --- a/internal/model/do/bil_wallets.go +++ b/internal/model/do/bil_wallets.go @@ -21,4 +21,5 @@ type BilWallets struct { CreatedAt *gtime.Time // 创建时间 UpdatedAt *gtime.Time // 更新时间 CumulativeRecharge any // 累计充值总额（USD） + LowBalanceNotified any // 低余额预警是否已推送（充值恢复后重置为 false） } diff --git a/internal/model/do/ord_orders.go b/internal/model/do/ord_orders.go index 5bd1394..0b88c31 100644 --- a/internal/model/do/ord_orders.go +++ b/internal/model/do/ord_orders.go @@ -21,7 +21,7 @@ type OrdOrders struct { Amount any // 原始金额 DiscountAmount any // 优惠金额 FinalAmount any // 最终金额 - Currency any // 货币 + Currency any // 货币（订单层一律 CNY） PaymentChannel any // 支付渠道 PaymentMethod any // 支付方式描述 PaymentNo any // 第三方支付流水号 diff --git a/internal/model/entity/aud_request_logs.go b/internal/model/entity/aud_request_logs.go index dc48e93..bf20994 100644 --- a/internal/model/entity/aud_request_logs.go +++ b/internal/model/entity/aud_request_logs.go @@ -40,4 +40,5 @@ type AudRequestLogs struct { TaskResult string `json:"task_result" orm:"task_result" description:"异步任务完成时上游返回的原始响应体"` // 异步任务完成时上游返回的原始响应体 TaskUpstreamHeaders string `json:"task_upstream_headers" orm:"task_upstream_headers" description:"异步任务完成时上游返回的响应头（仅审计级别为 full 时记录）"` // 异步任务完成时上游返回的响应头（仅审计级别为 full 时记录） TaskCompletedAt *gtime.Time `json:"task_completed_at" orm:"task_completed_at" description:"异步任务达到终态的时间"` // 异步任务达到终态的时间 + AttachmentIds string `json:"attachment_ids" orm:"attachment_ids" description:"提取的媒体附件 fil_files ID 数组（JSONB: [1,2,3]）"` // 提取的媒体附件 fil_files ID 数组（JSONB: [1,2,3]） } diff --git a/internal/model/entity/bil_wallets.go b/internal/model/entity/bil_wallets.go index d25b6cb..b21b6d8 100644 --- a/internal/model/entity/bil_wallets.go +++ b/internal/model/entity/bil_wallets.go @@ -10,13 +10,14 @@ import ( // BilWallets is the golang structure for table bil_wallets. type BilWallets struct { - Id int64 `json:"id" orm:"id" description:"主键ID"` // 主键ID - TenantId int64 `json:"tenant_id" orm:"tenant_id" description:"租户ID（每个租户一个钱包）"` // 租户ID（每个租户一个钱包） - Balance float64 `json:"balance" orm:"balance" description:"总余额"` // 总余额 - FrozenBalance float64 `json:"frozen_balance" orm:"frozen_balance" description:"冻结余额（支付中/退款中，可用余额 = balance - frozen_balance）"` // 冻结余额（支付中/退款中，可用余额 = balance - frozen_balance） - WarningThreshold float64 `json:"warning_threshold" orm:"warning_threshold" description:"余额预警线（低于此值触发通知）"` // 余额预警线（低于此值触发通知） - Currency string `json:"currency" orm:"currency" description:"货币（USD）"` // 货币（USD） - CreatedAt *gtime.Time `json:"created_at" orm:"created_at" description:"创建时间"` // 创建时间 - UpdatedAt *gtime.Time `json:"updated_at" orm:"updated_at" description:"更新时间"` // 更新时间 - CumulativeRecharge float64 `json:"cumulative_recharge" orm:"cumulative_recharge" description:"累计充值总额（USD）"` // 累计充值总额（USD） + Id int64 `json:"id" orm:"id" description:"主键ID"` // 主键ID + TenantId int64 `json:"tenant_id" orm:"tenant_id" description:"租户ID（每个租户一个钱包）"` // 租户ID（每个租户一个钱包） + Balance float64 `json:"balance" orm:"balance" description:"总余额"` // 总余额 + FrozenBalance float64 `json:"frozen_balance" orm:"frozen_balance" description:"冻结余额（支付中/退款中，可用余额 = balance - frozen_balance）"` // 冻结余额（支付中/退款中，可用余额 = balance - frozen_balance） + WarningThreshold float64 `json:"warning_threshold" orm:"warning_threshold" description:"余额预警线（低于此值触发通知）"` // 余额预警线（低于此值触发通知） + Currency string `json:"currency" orm:"currency" description:"货币（USD）"` // 货币（USD） + CreatedAt *gtime.Time `json:"created_at" orm:"created_at" description:"创建时间"` // 创建时间 + UpdatedAt *gtime.Time `json:"updated_at" orm:"updated_at" description:"更新时间"` // 更新时间 + CumulativeRecharge float64 `json:"cumulative_recharge" orm:"cumulative_recharge" description:"累计充值总额（USD）"` // 累计充值总额（USD） + LowBalanceNotified bool `json:"low_balance_notified" orm:"low_balance_notified" description:"低余额预警是否已推送（充值恢复后重置为 false）"` // 低余额预警是否已推送（充值恢复后重置为 false） } diff --git a/internal/model/entity/ord_orders.go b/internal/model/entity/ord_orders.go index 4973306..5dae326 100644 --- a/internal/model/entity/ord_orders.go +++ b/internal/model/entity/ord_orders.go @@ -19,7 +19,7 @@ type OrdOrders struct { Amount float64 `json:"amount" orm:"amount" description:"原始金额"` // 原始金额 DiscountAmount float64 `json:"discount_amount" orm:"discount_amount" description:"优惠金额"` // 优惠金额 FinalAmount float64 `json:"final_amount" orm:"final_amount" description:"最终金额"` // 最终金额 - Currency string `json:"currency" orm:"currency" description:"货币"` // 货币 + Currency string `json:"currency" orm:"currency" description:"货币（订单层一律 CNY）"` // 货币（订单层一律 CNY） PaymentChannel string `json:"payment_channel" orm:"payment_channel" description:"支付渠道"` // 支付渠道 PaymentMethod string `json:"payment_method" orm:"payment_method" description:"支付方式描述"` // 支付方式描述 PaymentNo string `json:"payment_no" orm:"payment_no" description:"第三方支付流水号"` // 第三方支付流水号 diff --git a/migrations/000007_v0_1_7_version.sql b/migrations/000007_v0_1_7_version.sql new file mode 100644 index 0000000..b4afc56 --- /dev/null +++ b/migrations/000007_v0_1_7_version.sql @@ -0,0 +1,61 @@ +-- +goose Up +-- Webhook 功能完善 + 余额预警事件驱动改造 +-- 合并原 000007 ~ 000010 四个迁移文件 + +-- ======================================== +-- 1. Webhook 性能优化：补充索引 +-- ======================================== + +-- JSONB GIN 索引：支持 webhook 配置的 events 查询 +CREATE INDEX IF NOT EXISTS idx_opn_webhook_configs_events_gin ON opn_webhook_configs USING gin (events jsonb_path_ops); + +-- 复合索引：加速查找租户的活跃 webhook 配置 +CREATE INDEX IF NOT EXISTS idx_opn_webhook_configs_tenant_active ON opn_webhook_configs (tenant_id, is_active) WHERE is_active = true; + +-- 按配置+租户查询投递日志 +CREATE INDEX IF NOT EXISTS idx_opn_webhook_delivery_logs_config ON opn_webhook_delivery_logs (webhook_config_id, tenant_id, created_at DESC); + +-- 按配置查询事件（级联删除时使用） +CREATE INDEX IF NOT EXISTS idx_opn_webhook_events_config ON opn_webhook_events (webhook_config_id); + +-- BRIN 索引：追加写入表磁盘友好 +CREATE INDEX IF NOT EXISTS idx_opn_webhook_events_created_brin ON opn_webhook_events USING brin (created_at); +CREATE INDEX IF NOT EXISTS idx_opn_webhook_delivery_logs_created_brin ON opn_webhook_delivery_logs USING brin (created_at); + +-- ======================================== +-- 2. 放宽 webhook URL 约束：允许 http:// 和 https:// +-- ======================================== +ALTER TABLE opn_webhook_configs DROP CONSTRAINT IF EXISTS chk_opn_webhook_configs_url; + +-- ======================================== +-- 3. 钱包表增加低余额预警推送标记 +-- ======================================== +ALTER TABLE bil_wallets ADD COLUMN IF NOT EXISTS low_balance_notified BOOLEAN NOT NULL DEFAULT false; +COMMENT ON COLUMN bil_wallets.low_balance_notified IS '低余额预警是否已推送（充值恢复后重置为 false）'; + +-- ======================================== +-- 4. 修正通知模板占位符语法 +-- ======================================== +UPDATE ntf_templates +SET body_template = '

您的账户可用余额为 {{.available}} USD，已低于预警线 {{.threshold}} USD。

请及时充值以避免服务中断。

' +WHERE code = 'balance_warning'; + +-- +goose Down +-- 恢复通知模板占位符 +UPDATE ntf_templates +SET body_template = '

您的账户可用余额为 ${available} USD，已低于预警线 ${threshold} USD。

请及时充值以避免服务中断。

' +WHERE code = 'balance_warning'; + +-- 移除钱包低余额标记字段 +ALTER TABLE bil_wallets DROP COLUMN IF EXISTS low_balance_notified; + +-- 恢复仅允许 https:// +ALTER TABLE opn_webhook_configs DROP CONSTRAINT IF EXISTS chk_opn_webhook_configs_url; + +-- 移除索引 +DROP INDEX IF EXISTS idx_opn_webhook_delivery_logs_created_brin; +DROP INDEX IF EXISTS idx_opn_webhook_events_created_brin; +DROP INDEX IF EXISTS idx_opn_webhook_events_config; +DROP INDEX IF EXISTS idx_opn_webhook_delivery_logs_config; +DROP INDEX IF EXISTS idx_opn_webhook_configs_tenant_active; +DROP INDEX IF EXISTS idx_opn_webhook_configs_events_gin; diff --git a/migrations/000007_webhook_indexes_and_cleanup.sql b/migrations/000007_webhook_indexes_and_cleanup.sql deleted file mode 100644 index 8258349..0000000 --- a/migrations/000007_webhook_indexes_and_cleanup.sql +++ /dev/null @@ -1,31 +0,0 @@ --- +goose Up --- Webhook 性能优化：补充索引 + 事件清理策略 --- 1. opn_webhook_configs: GIN 索引支持 JSONB @> 查询，复合索引加速活跃配置查找 --- 2. opn_webhook_delivery_logs: 按配置+租户查询索引 --- 3. opn_webhook_events: 按配置查询索引（级联删除用），BRIN 索引支持追加写入 - --- 1. JSONB GIN 索引：支持 PublishWebhookEvent 中的 events::jsonb @> ? 查询 -CREATE INDEX idx_opn_webhook_configs_events_gin ON opn_webhook_configs USING gin (events jsonb_path_ops); - --- 2. 复合索引：加速查找租户的活跃 webhook 配置 -CREATE INDEX idx_opn_webhook_configs_tenant_active ON opn_webhook_configs (tenant_id, is_active) WHERE is_active = true; - --- 3. 按配置+租户查询投递日志（WebhookDeliveryLogs API 使用） -CREATE INDEX idx_opn_webhook_delivery_logs_config ON opn_webhook_delivery_logs (webhook_config_id, tenant_id, created_at DESC); - --- 4. 按配置查询事件（级联删除时使用） -CREATE INDEX idx_opn_webhook_events_config ON opn_webhook_events (webhook_config_id); - --- 5. BRIN 索引：events 表追加写入模式，磁盘友好 -CREATE INDEX idx_opn_webhook_events_created_brin ON opn_webhook_events USING brin (created_at); - --- 6. BRIN 索引：delivery_logs 表追加写入模式 -CREATE INDEX idx_opn_webhook_delivery_logs_created_brin ON opn_webhook_delivery_logs USING brin (created_at); - --- +goose Down -DROP INDEX IF EXISTS idx_opn_webhook_delivery_logs_created_brin; -DROP INDEX IF EXISTS idx_opn_webhook_events_created_brin; -DROP INDEX IF EXISTS idx_opn_webhook_events_config; -DROP INDEX IF EXISTS idx_opn_webhook_delivery_logs_config; -DROP INDEX IF EXISTS idx_opn_webhook_configs_tenant_active; -DROP INDEX IF EXISTS idx_opn_webhook_configs_events_gin; diff --git a/web/tenant/src/views/notification/NotificationsPage.vue b/web/tenant/src/views/notification/NotificationsPage.vue index 94037d8..0c2de38 100644 --- a/web/tenant/src/views/notification/NotificationsPage.vue +++ b/web/tenant/src/views/notification/NotificationsPage.vue @@ -21,7 +21,6 @@ const page = ref(1) const pageSize = 20 const total = ref(0) const activeTab = ref<'all' | 'unread'>('all') -const expandedId = ref(null) const showExportDropdown = ref(false) const { exporting, exportFile } = useExport({ @@ -100,23 +99,11 @@ async function deleteNotification(n: Notification) { await request.delete(`/tenant/notifications/${n.id}`) notifications.value = notifications.value.filter((item) => item.id !== n.id) total.value-- - expandedId.value = null confirmDeleteId.value = null } catch { } } -function toggleNotification(n: Notification) { - if (expandedId.value === n.id) { - expandedId.value = null - } else { - expandedId.value = n.id - if (!n.is_read) { - markAsRead(n.id) - } - } -} - function switchTab(tab: 'all' | 'unread') { activeTab.value = tab page.value = 1 @@ -201,13 +188,12 @@ onMounted(() => {
-
+
@@ -221,29 +207,25 @@ onMounted(() => { {{ typeLabel[n.type] || n.type }}
-

{{ n.content }}

+
- - {{ n.created_at ? new Date(n.created_at).toLocaleString() : '' }} - -
- - -
-

{{ n.content }}

- -
-