余额告警独立页面与多渠道通知：钉钉/飞书/企业微信/Bark，支持通知频率策略

- 告警设置拆分为独立页面，支持钉钉/飞书/企业微信/Bark/通用Webhook 多渠道 - 通知策略：低于阈值后按间隔重复提醒，达到最大次数停止，余额恢复自动重置 - 旧 webhook 配置自动迁移为通用渠道，渠道支持测试发送 - 优化告警话术：去商户ID展示、数字千分位格式化
2026-08-04 13:56:26 +08:00
parent 941e7ad79a
commit ca58c2bcaa
13 changed files with 1154 additions and 156 deletions
@@ -10,6 +10,7 @@ import (
 	"net/http"
 	"os"
 	"path/filepath"
+	"strconv"
 	"strings"
 	"time"

@@ -27,6 +28,9 @@ const (
 	RechargePointsPerCNYCent  = 1    // 每 1 分人民币兑换积分
 )

+// timeNow 便于测试控制通知频率判断。
+var timeNow = time.Now
+
 type RechargeService struct {
 	db        *gorm.DB
 	fulfill   *FulfillmentService
@@ -180,9 +184,10 @@ func (s *RechargeService) ReviewRecharge(in ReviewRechargeInput) (*model.Recharg
 }

 type SaveAlertInput struct {
-	Enabled         bool
-	ThresholdPoints int64
-	WebhookURL      string
+	Enabled               bool
+	ThresholdPoints       int64
+	NotifyIntervalMinutes int
+	MaxNotifications      int
 }

 func (s *RechargeService) GetAlertConfig(merchantID uint) (*model.MerchantAlertConfig, error) {
@@ -201,36 +206,123 @@ func (s *RechargeService) SaveAlertConfig(merchantID uint, in SaveAlertInput) (*
 	if in.ThresholdPoints < 0 {
 		return nil, errors.New("预警阈值不能为负")
 	}
-	if in.WebhookURL == "" || len(in.WebhookURL) > 1024 {
-		return nil, errors.New("请填写 Webhook URL")
+	if in.NotifyIntervalMinutes < 1 {
+		return nil, errors.New("通知间隔至少 1 分钟")
 	}
-	if in.Enabled {
-		if !strings.HasPrefix(in.WebhookURL, "http://") && !strings.HasPrefix(in.WebhookURL, "https://") {
-			return nil, errors.New("Webhook URL 必须以 http(s):// 开头")
-		}
+	if in.MaxNotifications < 0 {
+		return nil, errors.New("最大通知次数不能为负")
 	}
 	var cfg model.MerchantAlertConfig
 	err := s.db.Where("merchant_id = ?", merchantID).First(&cfg).Error
 	switch {
 	case errors.Is(err, gorm.ErrRecordNotFound):
-		cfg = model.MerchantAlertConfig{MerchantID: merchantID, Enabled: in.Enabled, ThresholdPoints: in.ThresholdPoints, WebhookURL: strings.TrimSpace(in.WebhookURL)}
-		if err := s.db.Create(&cfg).Error; err != nil {
-			return nil, err
-		}
+		cfg = model.MerchantAlertConfig{MerchantID: merchantID, Enabled: in.Enabled, ThresholdPoints: in.ThresholdPoints}
 	case err != nil:
 		return nil, err
-	default:
-		cfg.Enabled = in.Enabled
-		cfg.ThresholdPoints = in.ThresholdPoints
-		cfg.WebhookURL = strings.TrimSpace(in.WebhookURL)
-		if err := s.db.Save(&cfg).Error; err != nil {
-			return nil, err
-		}
+	}
+	cfg.Enabled = in.Enabled
+	cfg.ThresholdPoints = in.ThresholdPoints
+	cfg.NotifyIntervalMinutes = in.NotifyIntervalMinutes
+	cfg.MaxNotifications = in.MaxNotifications
+	if err := s.db.Save(&cfg).Error; err != nil {
+		return nil, err
 	}
 	return &cfg, nil
 }

-// CheckLowBalanceAndNotify 余额低于阈值时向配置的 Webhook 推送告警。
+type AlertChannelInput struct {
+	ChannelType string
+	Name        string
+	Config      model.AlertChannelConfig
+	Enabled     bool
+}
+
+// ValidateAlertChannelInput 校验渠道类型与配置必填项。
+func ValidateAlertChannelInput(in AlertChannelInput) error {
+	in.ChannelType = strings.TrimSpace(in.ChannelType)
+	switch in.ChannelType {
+	case model.AlertChannelDingtalk, model.AlertChannelFeishu, model.AlertChannelWecom, model.AlertChannelWebhook:
+		if !strings.HasPrefix(in.Config.WebhookURL, "https://") && !strings.HasPrefix(in.Config.WebhookURL, "http://") {
+			return errors.New("Webhook URL 必须以 http(s):// 开头")
+		}
+		if len(in.Config.WebhookURL) > 1024 {
+			return errors.New("Webhook URL 过长")
+		}
+	case model.AlertChannelBark:
+		if strings.TrimSpace(in.Config.Key) == "" {
+			return errors.New("请填写 Bark 设备 Key")
+		}
+		if in.Config.Server != "" && !strings.HasPrefix(in.Config.Server, "https://") && !strings.HasPrefix(in.Config.Server, "http://") {
+			return errors.New("Bark 服务器地址必须以 http(s):// 开头")
+		}
+	default:
+		return errors.New("不支持的渠道类型")
+	}
+	return nil
+}
+
+func (s *RechargeService) ListAlertChannels(merchantID uint) ([]model.AlertNotifyChannel, error) {
+	var list []model.AlertNotifyChannel
+	err := s.db.Where("merchant_id = ?", merchantID).Order("id ASC").Find(&list).Error
+	return list, err
+}
+
+func (s *RechargeService) CreateAlertChannel(merchantID uint, in AlertChannelInput) (*model.AlertNotifyChannel, error) {
+	if err := ValidateAlertChannelInput(in); err != nil {
+		return nil, err
+	}
+	channel := &model.AlertNotifyChannel{
+		MerchantID:  merchantID,
+		ChannelType: strings.TrimSpace(in.ChannelType),
+		Name:        strings.TrimSpace(in.Name),
+		Config:      in.Config,
+		Enabled:     in.Enabled,
+	}
+	if len(channel.Name) > 64 {
+		return nil, errors.New("渠道名称最长 64 个字符")
+	}
+	if err := s.db.Create(channel).Error; err != nil {
+		return nil, err
+	}
+	return channel, nil
+}
+
+func (s *RechargeService) UpdateAlertChannel(merchantID, channelID uint, in AlertChannelInput) (*model.AlertNotifyChannel, error) {
+	if err := ValidateAlertChannelInput(in); err != nil {
+		return nil, err
+	}
+	if len(strings.TrimSpace(in.Name)) > 64 {
+		return nil, errors.New("渠道名称最长 64 个字符")
+	}
+	var channel model.AlertNotifyChannel
+	if err := s.db.Where("id = ? AND merchant_id = ?", channelID, merchantID).First(&channel).Error; err != nil {
+		return nil, errors.New("通知渠道不存在")
+	}
+	channel.ChannelType = strings.TrimSpace(in.ChannelType)
+	channel.Name = strings.TrimSpace(in.Name)
+	channel.Config = in.Config
+	channel.Enabled = in.Enabled
+	if err := s.db.Save(&channel).Error; err != nil {
+		return nil, err
+	}
+	return &channel, nil
+}
+
+func (s *RechargeService) DeleteAlertChannel(merchantID, channelID uint) error {
+	return s.db.Where("id = ? AND merchant_id = ?", channelID, merchantID).Delete(&model.AlertNotifyChannel{}).Error
+}
+
+// TestAlertChannel 向指定渠道发送一条测试消息。
+func (s *RechargeService) TestAlertChannel(merchantID, channelID uint) error {
+	var channel model.AlertNotifyChannel
+	if err := s.db.Where("id = ? AND merchant_id = ?", channelID, merchantID).First(&channel).Error; err != nil {
+		return errors.New("通知渠道不存在")
+	}
+	return sendChannelMessage(channel, "余额告警测试", "这是一条测试消息：如果你收到了这条消息，说明余额告警通知渠道配置正确。")
+}
+
+// CheckLowBalanceAndNotify 余额低于阈值时向所有已启用渠道推送告警。
+// 通知频率策略：按配置间隔重复通知，达到最大次数后停止；余额恢复阈值以上后重置计数。
 func (s *RechargeService) CheckLowBalanceAndNotify(merchantID uint) {
 	cfg, err := s.GetAlertConfig(merchantID)
 	if err != nil || !cfg.Enabled || cfg.ThresholdPoints <= 0 {
@@ -240,30 +332,139 @@ func (s *RechargeService) CheckLowBalanceAndNotify(merchantID uint) {
 	if err != nil {
 		return
 	}
+
+	// 余额已恢复：重置通知状态，等待下一轮告警
 	if wallet.AvailableBalance >= cfg.ThresholdPoints {
+		if cfg.NotificationCount > 0 || cfg.LastNotifiedAt != nil {
+			cfg.NotificationCount = 0
+			cfg.LastNotifiedAt = nil
+			_ = s.db.Model(&cfg).Updates(map[string]interface{}{
+				"notification_count": 0,
+				"last_notified_at":   nil,
+			}).Error
+		}
+		return
+	}
+
+	// 未到通知间隔：跳过
+	if cfg.LastNotifiedAt != nil && timeNow().Sub(*cfg.LastNotifiedAt) < time.Duration(cfg.NotifyIntervalMinutes)*time.Minute {
+		return
+	}
+	// 已达到最大通知次数：停止
+	if cfg.MaxNotifications > 0 && cfg.NotificationCount >= cfg.MaxNotifications {
+		return
+	}
+
+	channels, err := s.ListAlertChannels(merchantID)
+	if err != nil {
 		return
 	}
 	var merchant model.Merchant
 	if err := s.db.Select("code", "name").First(&merchant, merchantID).Error; err != nil {
-		merchant.Name = fmt.Sprintf("商户#%d", merchantID)
+		merchant.Name = "未知商户"
 	}
-	payload := map[string]interface{}{
-		"msgtype": "text",
-		"text": map[string]string{
-			"content": fmt.Sprintf("余额告警：商户 %s（#%d）当前积分余额 %d，已低于阈值 %d，请及时处理充值。", merchant.Name, merchantID, wallet.AvailableBalance, cfg.ThresholdPoints),
-		},
+	content := fmt.Sprintf("【余额告警】商户「%s」当前积分余额 %s，已低于预警阈值 %s，请及时充值以免影响正常业务。", merchant.Name, formatThousands(wallet.AvailableBalance), formatThousands(cfg.ThresholdPoints))
+
+	now := timeNow()
+	sent := false
+	for _, channel := range channels {
+		if !channel.Enabled {
+			continue
+		}
+		if err := sendChannelMessage(channel, "余额告警", content); err != nil {
+			log.Printf("低余额告警推送失败 merchant=%d channel=%d: %v", merchantID, channel.ID, err)
+			continue
+		}
+		sent = true
 	}
-	raw, _ := json.Marshal(payload)
-	client := &http.Client{Timeout: 10 * time.Second}
-	resp, err := client.Post(cfg.WebhookURL, "application/json", bytes.NewReader(raw))
+	// 至少一个渠道推送成功才推进通知状态，避免失败时漏掉后续重试
+	if sent {
+		cfg.NotificationCount++
+		cfg.LastNotifiedAt = &now
+		_ = s.db.Model(&cfg).Updates(map[string]interface{}{
+			"notification_count": cfg.NotificationCount,
+			"last_notified_at":   now,
+		}).Error
+	}
+}
+
+// formatThousands 千分位格式化数字，便于告警文案阅读。
+func formatThousands(n int64) string {
+	neg := n < 0
+	if neg {
+		n = -n
+	}
+	digits := strconv.FormatInt(n, 10)
+	var buf []byte
+	for i, c := range digits {
+		if i > 0 && (len(digits)-i)%3 == 0 {
+			buf = append(buf, ',')
+		}
+		buf = append(buf, byte(c))
+	}
+	if neg {
+		return "-" + string(buf)
+	}
+	return string(buf)
+}
+
+// sendChannelMessage 按渠道类型组装并推送消息。
+func sendChannelMessage(channel model.AlertNotifyChannel, title, content string) error {
+	var (
+		url     string
+		payload []byte
+		client  = &http.Client{Timeout: 10 * time.Second}
+	)
+	switch channel.ChannelType {
+	case model.AlertChannelDingtalk:
+		url = channel.Config.WebhookURL
+		payload, _ = json.Marshal(map[string]interface{}{
+			"msgtype": "text",
+			"text":    map[string]string{"content": content},
+		})
+	case model.AlertChannelFeishu:
+		url = channel.Config.WebhookURL
+		payload, _ = json.Marshal(map[string]interface{}{
+			"msg_type": "text",
+			"content":  map[string]string{"text": content},
+		})
+	case model.AlertChannelWecom:
+		url = channel.Config.WebhookURL
+		payload, _ = json.Marshal(map[string]interface{}{
+			"msgtype": "text",
+			"text":    map[string]string{"content": content},
+		})
+	case model.AlertChannelBark:
+		server := strings.TrimRight(channel.Config.Server, "/")
+		if server == "" {
+			server = "https://api.day.app"
+		}
+		url = server + "/push"
+		payload, _ = json.Marshal(map[string]string{
+			"device_key": channel.Config.Key,
+			"title":      title,
+			"body":       content,
+		})
+	case model.AlertChannelWebhook:
+		url = channel.Config.WebhookURL
+		payload, _ = json.Marshal(map[string]string{
+			"title":   title,
+			"content": content,
+		})
+	default:
+		return errors.New("不支持的渠道类型")
+	}
+
+	resp, err := client.Post(url, "application/json", bytes.NewReader(payload))
 	if err != nil {
-		log.Printf("低余额告警推送失败 merchant=%d: %v", merchantID, err)
-		return
+		return err
 	}
 	defer resp.Body.Close()
 	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
-		log.Printf("低余额告警推送非 2xx merchant=%d status=%d", merchantID, resp.StatusCode)
+		body, _ := io.ReadAll(io.LimitReader(resp.Body, 1024))
+		return fmt.Errorf("推送返回非 2xx 状态码 %d：%s", resp.StatusCode, string(body))
 	}
+	return nil
 }

 // SaveUploadFile 保存上传的图片到 uploadDir，返回可通过 /uploads 访问的 URL。
@@ -1,10 +1,16 @@
 package service

 import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
 	"strings"
 	"testing"
+	"time"

 	"affiliate_dash/internal/model"
+
+	"github.com/google/uuid"
 )

 func TestCreateRechargeRequiresVoucher(t *testing.T) {
@@ -93,6 +99,238 @@ func TestReviewRechargeCreditsWalletOnce(t *testing.T) {
 	}
 }

+func TestAlertChannelCRUD(t *testing.T) {
+	db := newServiceTestDB(t)
+	merchantID, _ := seedFulfillmentMerchant(t, db, "merchant-alert-channel", 0, -1, 100)
+	svc := NewRechargeService(db, NewFulfillmentService(db, nil), t.TempDir())
+
+	// 非法渠道类型
+	if _, err := svc.CreateAlertChannel(merchantID, AlertChannelInput{ChannelType: "sms", Name: "短信"}); err == nil {
+		t.Fatalf("invalid channel type should be rejected")
+	}
+	// 缺少 webhook url
+	if _, err := svc.CreateAlertChannel(merchantID, AlertChannelInput{ChannelType: model.AlertChannelDingtalk}); err == nil || !strings.Contains(err.Error(), "Webhook URL") {
+		t.Fatalf("missing webhook url should be rejected, got %v", err)
+	}
+
+	dingtalk, err := svc.CreateAlertChannel(merchantID, AlertChannelInput{
+		ChannelType: model.AlertChannelDingtalk,
+		Name:        "财务群",
+		Enabled:     true,
+		Config:      model.AlertChannelConfig{WebhookURL: "https://oapi.dingtalk.com/robot/send?access_token=test"},
+	})
+	if err != nil {
+		t.Fatalf("create dingtalk channel: %v", err)
+	}
+	bark, err := svc.CreateAlertChannel(merchantID, AlertChannelInput{
+		ChannelType: model.AlertChannelBark,
+		Name:        "老板手机",
+		Enabled:     true,
+		Config:      model.AlertChannelConfig{Server: "https://api.day.app", Key: "abc123"},
+	})
+	if err != nil {
+		t.Fatalf("create bark channel: %v", err)
+	}
+	if bark.Config.Key != "abc123" {
+		t.Fatalf("bark config should persist, got %+v", bark.Config)
+	}
+
+	// 跨商户隔离
+	otherMerchant, _ := seedFulfillmentMerchant(t, db, "merchant-alert-channel-other", 0, -1, 100)
+	if _, err := svc.UpdateAlertChannel(otherMerchant, dingtalk.ID, AlertChannelInput{
+		ChannelType: model.AlertChannelDingtalk,
+		Config:      model.AlertChannelConfig{WebhookURL: "https://oapi.dingtalk.com/robot/send?access_token=hack"},
+	}); err == nil {
+		t.Fatalf("other merchant should not update channel")
+	}
+	if err := svc.DeleteAlertChannel(otherMerchant, dingtalk.ID); err != nil {
+		t.Fatalf("other merchant delete should not error (no-op): %v", err)
+	}
+	list, err := svc.ListAlertChannels(merchantID)
+	if err != nil || len(list) != 2 {
+		t.Fatalf("expected 2 channels, got %d, err=%v", len(list), err)
+	}
+}
+
+func TestSendChannelMessagePayload(t *testing.T) {
+	got := make(chan map[string]interface{}, 1)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var body map[string]interface{}
+		_ = json.NewDecoder(r.Body).Decode(&body)
+		got <- body
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer srv.Close()
+
+	// 钉钉格式
+	channel := model.AlertNotifyChannel{ChannelType: model.AlertChannelDingtalk, Config: model.AlertChannelConfig{WebhookURL: srv.URL}}
+	if err := sendChannelMessage(channel, "余额告警", "内容"); err != nil {
+		t.Fatalf("send dingtalk: %v", err)
+	}
+	body := <-got
+	if body["msgtype"] != "text" || body["text"].(map[string]interface{})["content"] != "内容" {
+		t.Fatalf("unexpected dingtalk payload: %+v", body)
+	}
+
+	// 飞书格式
+	channel = model.AlertNotifyChannel{ChannelType: model.AlertChannelFeishu, Config: model.AlertChannelConfig{WebhookURL: srv.URL}}
+	if err := sendChannelMessage(channel, "余额告警", "内容"); err != nil {
+		t.Fatalf("send feishu: %v", err)
+	}
+	body = <-got
+	if body["msg_type"] != "text" || body["content"].(map[string]interface{})["text"] != "内容" {
+		t.Fatalf("unexpected feishu payload: %+v", body)
+	}
+
+	// Bark 格式（服务器为空时使用默认服务器会请求外部网络，这里显式指定测试服务器）
+	channel = model.AlertNotifyChannel{ChannelType: model.AlertChannelBark, Config: model.AlertChannelConfig{Server: srv.URL, Key: "device-1"}}
+	if err := sendChannelMessage(channel, "标题", "内容"); err != nil {
+		t.Fatalf("send bark: %v", err)
+	}
+	body = <-got
+	if body["device_key"] != "device-1" || body["title"] != "标题" || body["body"] != "内容" {
+		t.Fatalf("unexpected bark payload: %+v", body)
+	}
+}
+
+func TestAlertNotificationPolicy(t *testing.T) {
+	db := newServiceTestDB(t)
+	merchantID, _ := seedFulfillmentMerchant(t, db, "merchant-alert-policy", 0, -1, 100)
+	fulfillment := NewFulfillmentService(db, nil)
+	svc := NewRechargeService(db, fulfillment, t.TempDir())
+
+	notified := make(chan struct{}, 16)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		notified <- struct{}{}
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer srv.Close()
+
+	if _, err := svc.CreateAlertChannel(merchantID, AlertChannelInput{
+		ChannelType: model.AlertChannelWebhook,
+		Name:        "测试",
+		Enabled:     true,
+		Config:      model.AlertChannelConfig{WebhookURL: srv.URL},
+	}); err != nil {
+		t.Fatalf("create channel: %v", err)
+	}
+	if _, err := svc.SaveAlertConfig(merchantID, SaveAlertInput{
+		Enabled:               true,
+		ThresholdPoints:       100,
+		NotifyIntervalMinutes: 30,
+		MaxNotifications:      3,
+	}); err != nil {
+		t.Fatalf("save alert config: %v", err)
+	}
+
+	// 固定时间基线，逐步推进
+	base := time.Date(2026, 8, 4, 10, 0, 0, 0, time.Local)
+	advance := func(d time.Duration) {
+		timeNow = func() time.Time { return base.Add(d) }
+	}
+	t.Cleanup(func() { timeNow = time.Now })
+	advance(0)
+
+	// 余额 0 < 阈值 100：第一次触发 → 通知 1 次
+	svc.CheckLowBalanceAndNotify(merchantID)
+	select {
+	case <-notified:
+	default:
+		t.Fatalf("first check should notify")
+	}
+
+	// 10 分钟后仍在间隔内 → 不通知
+	advance(10 * time.Minute)
+	svc.CheckLowBalanceAndNotify(merchantID)
+	select {
+	case <-notified:
+		t.Fatalf("within interval should not notify")
+	default:
+	}
+
+	// 31 分钟后超出间隔 → 第二次通知
+	advance(31 * time.Minute)
+	svc.CheckLowBalanceAndNotify(merchantID)
+	select {
+	case <-notified:
+	default:
+		t.Fatalf("after interval should notify")
+	}
+
+	// 61 分钟 → 第三次通知
+	advance(61 * time.Minute)
+	svc.CheckLowBalanceAndNotify(merchantID)
+	select {
+	case <-notified:
+	default:
+		t.Fatalf("third notification expected")
+	}
+
+	// 91 分钟 → 已达最大次数 3，不再通知
+	advance(91 * time.Minute)
+	svc.CheckLowBalanceAndNotify(merchantID)
+	select {
+	case <-notified:
+		t.Fatalf("max notifications reached, should stop")
+	default:
+	}
+
+	// 充值恢复余额 ≥ 阈值 → 重置计数
+	if _, err := fulfillment.AdjustWallet(WalletAdjustInput{
+		MerchantID:     merchantID,
+		ActorUserID:    1,
+		Amount:         500,
+		IdempotencyKey: "recover-" + uuid.NewString(),
+	}); err != nil {
+		t.Fatalf("adjust wallet: %v", err)
+	}
+	advance(92 * time.Minute)
+	svc.CheckLowBalanceAndNotify(merchantID)
+	var cfg model.MerchantAlertConfig
+	if err := db.Where("merchant_id = ?", merchantID).First(&cfg).Error; err != nil {
+		t.Fatalf("query config: %v", err)
+	}
+	if cfg.NotificationCount != 0 || cfg.LastNotifiedAt != nil {
+		t.Fatalf("recovery should reset notification state, got count=%d", cfg.NotificationCount)
+	}
+
+	// 又扣回低余额 → 重新开始新一轮通知
+	if _, err := fulfillment.AdjustWallet(WalletAdjustInput{
+		MerchantID:     merchantID,
+		ActorUserID:    1,
+		Amount:         -500,
+		IdempotencyKey: "down-" + uuid.NewString(),
+	}); err != nil {
+		t.Fatalf("adjust wallet down: %v", err)
+	}
+	advance(93 * time.Minute)
+	svc.CheckLowBalanceAndNotify(merchantID)
+	select {
+	case <-notified:
+	default:
+		t.Fatalf("new round should notify again")
+	}
+}
+
+func TestFormatThousands(t *testing.T) {
+	cases := []struct {
+		in   int64
+		want string
+	}{
+		{0, "0"},
+		{999, "999"},
+		{1000, "1,000"},
+		{990, "990"},
+		{1234567, "1,234,567"},
+		{-1234567, "-1,234,567"},
+	}
+	for _, tc := range cases {
+		if got := formatThousands(tc.in); got != tc.want {
+			t.Fatalf("formatThousands(%d) = %q, want %q", tc.in, got, tc.want)
+		}
+	}
+}
+
 func TestDetectImageExt(t *testing.T) {
 	cases := []struct {
 		name string