// Package notice 提供统一通知服务(渠道/规则/发送/历史记录)。 // 发送实现:Bark 走路径式 key(POST {baseUrl}/{deviceKey} JSON);pushplus 走官方 send 接口。 package notice import ( "context" "encoding/json" "strings" "time" "github.com/gogf/gf/v2/database/gdb" "github.com/gogf/gf/v2/errors/gerror" "github.com/gogf/gf/v2/frame/g" "github.com/gogf/gf/v2/os/gtime" "github.com/gogf/gf/v2/util/guid" "service.xpcool.com/internal/consts" "service.xpcool.com/internal/dao" "service.xpcool.com/internal/library/response" "service.xpcool.com/internal/model/do" "service.xpcool.com/internal/model/dto" "service.xpcool.com/internal/model/entity" ) // gdbModel 查询模型别名,避免在多处书写完整包路径。 type gdbModel = *gdb.Model // INotice 统一通知服务接口。 type INotice interface { ChannelList(context.Context) ([]dto.NoticeChannelVO, error) ChannelSave(context.Context, dto.NoticeChannelInput) (uint64, error) RuleList(context.Context, string) ([]dto.NoticeRuleVO, error) RuleSave(context.Context, dto.NoticeRuleInput) (uint64, error) RuleDelete(context.Context, uint64) error // LogList 通知历史记录分页(返回列表、总数与统计概览)。 LogList(context.Context, dto.NoticeLogFilter) ([]dto.NoticeLogVO, int, *dto.NoticeLogStats, error) LogDetail(context.Context, uint64) (*dto.NoticeLogVO, error) LogDelete(context.Context, []uint64) (int, error) LogClear(context.Context, int, string, string) (int, error) Send(context.Context, string, map[string]string) error // 按启用规则发送通知(事件→渠道→人员) Test(context.Context, uint64, string, string, []uint64, string, string) (bool, string, error) } type notice struct{} var localNotice INotice // New 创建通知服务实现。 func New() INotice { return ¬ice{} } // Notice 返回已注册的通知服务实现。 func Notice() INotice { if localNotice == nil { panic("Notice 实现未注册") } return localNotice } // RegisterNotice 注册通知服务实现。 func RegisterNotice(i INotice) { localNotice = i } // ---------------- 渠道 ---------------- func (s *notice) ChannelList(ctx context.Context) ([]dto.NoticeChannelVO, error) { var list []entity.NoticeChannel if err := dao.NoticeChannel.Ctx(ctx).OrderAsc("id").Scan(&list); err != nil { return nil, gerror.Wrap(err, "查询通知渠道失败") } out := make([]dto.NoticeChannelVO, 0, len(list)) for _, v := range list { out = append(out, dto.NoticeChannelVO{Id: v.Id, Code: v.Code, Name: v.Name, Enabled: v.Enabled, Config: v.Config, Remark: v.Remark}) } return out, nil } func (s *notice) ChannelSave(ctx context.Context, in dto.NoticeChannelInput) (uint64, error) { data := do.NoticeChannel{ Code: in.Code, Name: in.Name, Enabled: in.Enabled, Config: in.Config, Remark: in.Remark, } if in.Id > 0 { if _, err := dao.NoticeChannel.Ctx(ctx).Where("id", in.Id).Data(data).Update(); err != nil { return 0, gerror.Wrap(err, "更新通知渠道失败") } return in.Id, nil } id, err := dao.NoticeChannel.Ctx(ctx).Data(data).InsertAndGetId() if err != nil { return 0, gerror.Wrap(err, "新增通知渠道失败") } return uint64(id), nil } // ---------------- 规则 ---------------- func (s *notice) RuleList(ctx context.Context, eventType string) ([]dto.NoticeRuleVO, error) { m := dao.NoticeRule.Ctx(ctx) if eventType != "" { m = m.Where("event_type", eventType) } var list []entity.NoticeRule if err := m.OrderAsc("id").Scan(&list); err != nil { return nil, gerror.Wrap(err, "查询通知规则失败") } out := make([]dto.NoticeRuleVO, 0, len(list)) for _, v := range list { out = append(out, dto.NoticeRuleVO{ Id: v.Id, Name: v.Name, EventType: v.EventType, ChannelCodes: parseStrArray(v.ChannelCodes), UserIds: parseUintArray(v.UserIds), TitleTemplate: v.TitleTemplate, BodyTemplate: v.BodyTemplate, Enabled: v.Enabled, }) } return out, nil } func (s *notice) RuleSave(ctx context.Context, in dto.NoticeRuleInput) (uint64, error) { data := do.NoticeRule{ Name: in.Name, EventType: in.EventType, ChannelCodes: toJSON(in.ChannelCodes), UserIds: toJSON(in.UserIds), TitleTemplate: in.TitleTemplate, BodyTemplate: in.BodyTemplate, Enabled: in.Enabled, } if in.Id > 0 { if _, err := dao.NoticeRule.Ctx(ctx).Where("id", in.Id).Data(data).Update(); err != nil { return 0, gerror.Wrap(err, "更新通知规则失败") } return in.Id, nil } id, err := dao.NoticeRule.Ctx(ctx).Data(data).InsertAndGetId() if err != nil { return 0, gerror.Wrap(err, "新增通知规则失败") } return uint64(id), nil } func (s *notice) RuleDelete(ctx context.Context, id uint64) error { if _, err := dao.NoticeRule.Ctx(ctx).Where("id", id).Delete(); err != nil { return gerror.Wrap(err, "删除通知规则失败") } return nil } // ---------------- 历史记录 ---------------- // LogList 通知历史记录分页:支持关键字、类型、分组、事件、渠道、接收人、状态、时间范围、排序。 func (s *notice) LogList(ctx context.Context, f dto.NoticeLogFilter) ([]dto.NoticeLogVO, int, *dto.NoticeLogStats, error) { f.Normalize() m := s.logQuery(ctx, f) total, err := m.Clone().Count() if err != nil { return nil, 0, nil, gerror.Wrap(err, "统计通知历史记录总数失败") } // 统计概览(成功/失败/成功率),与列表使用同一套筛选条件。 stats, err := s.logStats(ctx, f, total) if err != nil { return nil, 0, nil, err } var list []entity.NoticeLog orderField := "created_at" if f.OrderBy == "durationMs" { orderField = "duration_ms" } if f.OrderDir == "asc" { err = m.Clone().Page(f.Page, f.Size).OrderAsc(orderField).OrderAsc("id").Scan(&list) } else { err = m.Clone().Page(f.Page, f.Size).OrderDesc(orderField).OrderDesc("id").Scan(&list) } if err != nil { return nil, 0, nil, gerror.Wrap(err, "查询通知历史记录列表失败") } out := s.buildLogVOs(ctx, list) return out, total, stats, nil } // logQuery 组装历史记录筛选条件(列表与统计共用,保证口径一致)。 func (s *notice) logQuery(ctx context.Context, f dto.NoticeLogFilter) gdbModel { m := dao.NoticeLog.Ctx(ctx) if f.Keyword != "" { kw := "%" + f.Keyword + "%" m = m.Where("title LIKE ? OR body LIKE ? OR target LIKE ? OR error LIKE ?", kw, kw, kw, kw) } if f.EventType != "" { m = m.Where("event_type", f.EventType) } else if f.Group != "" { if events := dto.NoticeGroupEventCodes(f.Group); len(events) > 0 { m = m.WhereIn("event_type", events) } else { // 未在字典中的分组(如 system)则反向排除已知事件。 m = m.WhereNotIn("event_type", dto.KnownEventCodes()) } } else if f.NoticeType != "" { switch f.NoticeType { case dto.NoticeTypeSystem: m = m.WhereNotIn("event_type", dto.KnownEventCodes()) default: if events := dto.NoticeTypeEventCodes(f.NoticeType); len(events) > 0 { m = m.WhereIn("event_type", events) } } } if f.ChannelCode != "" { m = m.Where("channel_code", f.ChannelCode) } if f.UserId > 0 { m = m.Where("user_id", f.UserId) } if f.BatchId != "" { m = m.Where("batch_id", f.BatchId) } if f.Status > 0 { if f.Status == dto.NoticeStatusSuccess { // 成功:status=1 或历史数据 result=1。 m = m.Where("(status = 1 OR (status = 0 AND result = 1))") } else { // 失败:status=2 或历史数据 result=0(且非待发送)。 m = m.Where("(status = 2 OR (status = 0 AND result = 0))") } } if f.DateFrom != "" { m = m.WhereGTE("created_at", f.DateFrom) } if f.DateTo != "" { m = m.WhereLTE("created_at", f.DateTo) } return m } // logStats 统计概览:总数、成功数、失败数、成功率。 func (s *notice) logStats(ctx context.Context, f dto.NoticeLogFilter, total int) (*dto.NoticeLogStats, error) { stats := &dto.NoticeLogStats{Total: total} successFilter := f successFilter.Status = dto.NoticeStatusSuccess if n, err := s.logQuery(ctx, successFilter).Clone().Count(); err != nil { return nil, gerror.Wrap(err, "统计通知成功数失败") } else { stats.Success = n } stats.Failed = total - stats.Success if stats.Failed < 0 { stats.Failed = 0 } if total > 0 { stats.SuccessRate = int(float64(stats.Success)*100/float64(total) + 0.5) } return stats, nil } // buildLogVOs 实体转 VO,并批量补齐规则名 / 接收人 等派生字段(避免 N+1)。 func (s *notice) buildLogVOs(ctx context.Context, list []entity.NoticeLog) []dto.NoticeLogVO { out := make([]dto.NoticeLogVO, 0, len(list)) if len(list) == 0 { return out } // 批量取规则名称 ruleIds := make([]uint64, 0, len(list)) userIds := make([]uint64, 0, len(list)) seenRule := map[uint64]struct{}{} seenUser := map[uint64]struct{}{} for _, v := range list { if v.RuleId > 0 { if _, ok := seenRule[v.RuleId]; !ok { seenRule[v.RuleId] = struct{}{} ruleIds = append(ruleIds, v.RuleId) } } if v.UserId > 0 { if _, ok := seenUser[v.UserId]; !ok { seenUser[v.UserId] = struct{}{} userIds = append(userIds, v.UserId) } } } ruleNames := map[uint64]string{} if len(ruleIds) > 0 { var rules []entity.NoticeRule if err := dao.NoticeRule.Ctx(ctx).WhereIn("id", ruleIds).Scan(&rules); err == nil { for _, r := range rules { ruleNames[r.Id] = r.Name } } } userNames := map[uint64]string{} if len(userIds) > 0 { var users []entity.AdminUser if err := dao.AdminUser.Ctx(ctx).WhereIn("id", userIds).Fields("id, username, nickname").Scan(&users); err == nil { for _, u := range users { name := u.Nickname if name == "" { name = u.Username } userNames[u.Id] = name } } } for _, v := range list { out = append(out, toLogVO(v, ruleNames[v.RuleId], userNames[v.UserId])) } return out } // toLogVO 单条实体转 VO(补齐派生名称字段)。 func toLogVO(v entity.NoticeLog, ruleName, userName string) dto.NoticeLogVO { noticeType := dto.NoticeEventToType(v.EventType) group := dto.NoticeTypeToGroup(noticeType) // status 兼容历史数据:仅有 result 时按 result 推导。 status := v.Status if status == 0 && v.Result == 1 { status = dto.NoticeStatusSuccess } else if status == 0 && v.Result == 0 && v.CreatedAt != "" && v.Error != "" { status = dto.NoticeStatusFailed } return dto.NoticeLogVO{ Id: v.Id, BatchId: v.BatchId, RuleId: v.RuleId, RuleName: ruleName, EventType: v.EventType, EventName: dto.NoticeEventName(v.EventType), NoticeType: noticeType, TypeName: dto.NoticeTypeName(noticeType), Group: group, GroupName: dto.NoticeGroupName(group), ChannelCode: v.ChannelCode, ChannelName: dto.NoticeChannelName(v.ChannelCode), UserId: v.UserId, UserName: userName, Target: v.Target, Title: v.Title, Body: v.Body, Status: status, StatusName: dto.NoticeStatusName(status), Result: v.Result, Error: v.Error, RetryCount: v.RetryCount, DurationMs: v.DurationMs, Source: v.Source, Remark: v.Remark, CreatedAt: v.CreatedAt, } } // LogDetail 单条历史记录详情。 func (s *notice) LogDetail(ctx context.Context, id uint64) (*dto.NoticeLogVO, error) { var v entity.NoticeLog if err := dao.NoticeLog.Ctx(ctx).Where("id", id).Scan(&v); err != nil { return nil, gerror.Wrap(err, "查询通知历史详情失败") } if v.Id == 0 { return nil, response.Error(consts.CodeInvalidParam, "通知记录不存在") } ruleName := "" if v.RuleId > 0 { var r entity.NoticeRule _ = dao.NoticeRule.Ctx(ctx).Where("id", v.RuleId).Scan(&r) ruleName = r.Name } userName := "" if v.UserId > 0 { var u entity.AdminUser _ = dao.AdminUser.Ctx(ctx).Where("id", v.UserId).Fields("id, username, nickname").Scan(&u) userName = u.Nickname if userName == "" { userName = u.Username } } vo := toLogVO(v, ruleName, userName) return &vo, nil } // LogDelete 批量删除历史记录,返回删除条数。 func (s *notice) LogDelete(ctx context.Context, ids []uint64) (int, error) { if len(ids) == 0 { return 0, response.Error(consts.CodeInvalidParam, "请选择要删除的记录") } r, err := dao.NoticeLog.Ctx(ctx).WhereIn("id", ids).Delete() if err != nil { return 0, gerror.Wrap(err, "删除通知历史记录失败") } n, _ := r.RowsAffected() return int(n), nil } // LogClear 清空历史记录:KeepDays>0 时保留最近 N 天,否则按 DateFrom/DateTo 范围删除(都为空则全清)。 func (s *notice) LogClear(ctx context.Context, keepDays int, dateFrom, dateTo string) (int, error) { m := dao.NoticeLog.Ctx(ctx) switch { case keepDays > 0: cut := gtime.Now().AddDate(0, 0, -keepDays).Format("Y-m-d H:i:s") m = m.WhereLT("created_at", cut) case dateFrom != "" || dateTo != "": if dateFrom != "" { m = m.WhereGTE("created_at", dateFrom) } if dateTo != "" { m = m.WhereLTE("created_at", dateTo) } default: m = m.Where("id > ?", 0) // 全清 } r, err := m.Delete() if err != nil { return 0, gerror.Wrap(err, "清空通知历史记录失败") } n, _ := r.RowsAffected() return int(n), nil } // ---------------- 发送 ---------------- // Send 按事件类型查启用规则,向每个渠道的每个接收用户发送,并记录日志。 // 同一次 Send 产生的多条投递共用同一批次号(batchId),便于按批次追溯。 func (s *notice) Send(ctx context.Context, eventType string, vars map[string]string) error { var rules []entity.NoticeRule if err := dao.NoticeRule.Ctx(ctx).Where("event_type", eventType).Where("enabled", 1).Scan(&rules); err != nil { return gerror.Wrap(err, "查询待发送通知规则失败") } if len(rules) == 0 { return nil // 未配置规则 = 静默 } batchId := newBatchId() for _, r := range rules { channels := parseStrArray(r.ChannelCodes) userIds := parseUintArray(r.UserIds) title := dto.RenderTemplate(r.TitleTemplate, vars) body := dto.RenderTemplate(r.BodyTemplate, vars) for _, channel := range channels { for _, uid := range userIds { s.deliver(ctx, deliverInput{ BatchId: batchId, RuleId: r.Id, EventType: eventType, Channel: channel, UserId: uid, Title: title, Body: body, Source: "auto", }) } } } return nil } // deliverInput 单条投递入参。 type deliverInput struct { BatchId string RuleId uint64 EventType string Channel string UserId uint64 Title string Body string Source string // auto/manual/test Remark string } // deliver 单条投递:取用户渠道凭据 → 发送 → 记日志(含耗时与来源)。 func (s *notice) deliver(ctx context.Context, in deliverInput) { var user entity.AdminUser _ = dao.AdminUser.Ctx(ctx).Where("id", in.UserId).Scan(&user) var target string switch in.Channel { case "bark": target = user.BarkDeviceId case "pushplus": target = user.PushplusToken default: target = "" } start := time.Now() ok, errMsg := s.sendByChannel(ctx, in.Channel, target, in.Title, in.Body) duration := int(time.Since(start).Milliseconds()) status := dto.NoticeStatusFailed if ok { status = dto.NoticeStatusSuccess } _, _ = dao.NoticeLog.Ctx(ctx).Data(do.NoticeLog{ BatchId: in.BatchId, RuleId: in.RuleId, EventType: in.EventType, ChannelCode: in.Channel, UserId: in.UserId, Target: target, Title: in.Title, Body: in.Body, Status: status, Result: boolToInt(ok), Error: errMsg, DurationMs: duration, Source: in.Source, Remark: in.Remark, }).Insert() } // sendByChannel 按渠道发送单条;返回 (是否成功, 错误信息)。 func (s *notice) sendByChannel(ctx context.Context, channel, target, title, body string) (bool, string) { if target == "" { return false, "目标凭据为空(用户未配置)" } client := g.Client().SetTimeout(10 * time.Second) switch channel { case "bark": // 渠道配置取 baseUrl(JSON config.baseUrl),默认容器网络内 bark:8080 baseURL := "http://bark:8080" var ch entity.NoticeChannel if err := dao.NoticeChannel.Ctx(ctx).Where("code", "bark").Scan(&ch); err == nil && ch.Config != "" { var cfg struct { BaseURL string `json:"baseUrl"` } _ = json.Unmarshal([]byte(ch.Config), &cfg) if cfg.BaseURL != "" { baseURL = cfg.BaseURL } } // 路径式 key:POST {baseUrl}/{deviceKey},body JSON {title, body} url := strings.TrimRight(baseURL, "/") + "/" + target resp, err := client.Post(ctx, url, map[string]string{"title": title, "body": body}) if err != nil { return false, err.Error() } if resp.StatusCode != 200 { return false, "bark http " + resp.Status } var out struct { Code int `json:"code"` } respBody := resp.ReadAllString() _ = json.Unmarshal([]byte(respBody), &out) if out.Code != 200 { return false, "bark code " + respBody } return true, "" case "pushplus": // 官方接口:POST https://www.pushplus.plus/send {token,title,content} resp, err := client.Post(ctx, "https://www.pushplus.plus/send", map[string]string{ "token": target, "title": title, "content": body, }) if err != nil { return false, err.Error() } if resp.StatusCode != 200 { return false, "pushplus http " + resp.Status } var out struct { Code int `json:"code"` } respBody := resp.ReadAllString() _ = json.Unmarshal([]byte(respBody), &out) if out.Code != 200 { return false, "pushplus code " + respBody } return true, "" } return false, "未知渠道: " + channel } // Test 测试发送:优先按规则;否则手动指定渠道+目标。 func (s *notice) Test(ctx context.Context, ruleId uint64, channelCode, target string, userIds []uint64, title, body string) (bool, string, error) { batchId := newBatchId() if ruleId > 0 { var r entity.NoticeRule if err := dao.NoticeRule.Ctx(ctx).Where("id", ruleId).Scan(&r); err != nil || r.Id == 0 { return false, "", response.Error(consts.CodeInvalidParam, "规则不存在") } for _, ch := range parseStrArray(r.ChannelCodes) { for _, uid := range parseUintArray(r.UserIds) { s.deliver(ctx, deliverInput{ BatchId: batchId, RuleId: r.Id, EventType: dto.NoticeEventTest, Channel: ch, UserId: uid, Title: title, Body: body, Source: "test", Remark: "按规则测试", }) } } return true, "已按规则投递", nil } // 手动:渠道+目标(target 为空则按用户) if channelCode == "" { return false, "", response.Error(consts.CodeInvalidParam, "请指定渠道") } userId := uint64(0) if target == "" && len(userIds) > 0 { userId = userIds[0] var user entity.AdminUser _ = dao.AdminUser.Ctx(ctx).Where("id", userId).Scan(&user) if channelCode == "bark" { target = user.BarkDeviceId } else if channelCode == "pushplus" { target = user.PushplusToken } } if target == "" { return false, "", response.Error(consts.CodeInvalidParam, "目标凭据为空") } start := time.Now() ok, msg := s.sendByChannel(ctx, channelCode, target, title, body) duration := int(time.Since(start).Milliseconds()) status := dto.NoticeStatusFailed if ok { status = dto.NoticeStatusSuccess } _, _ = dao.NoticeLog.Ctx(ctx).Data(do.NoticeLog{ BatchId: batchId, RuleId: 0, EventType: dto.NoticeEventTest, ChannelCode: channelCode, UserId: userId, Target: target, Title: title, Body: body, Status: status, Result: boolToInt(ok), Error: msg, DurationMs: duration, Source: "test", Remark: "手动测试", }).Insert() return ok, msg, nil } // ---------------- 工具 ---------------- // newBatchId 生成投递批次号(同一次业务触发共用一个)。 func newBatchId() string { return guid.S()[:16] } func parseStrArray(s string) []string { var out []string _ = json.Unmarshal([]byte(s), &out) if out == nil { out = []string{} } return out } func parseUintArray(s string) []uint64 { var out []uint64 _ = json.Unmarshal([]byte(s), &out) if out == nil { out = []uint64{} } return out } func toJSON(v interface{}) string { b, _ := json.Marshal(v) return string(b) } func boolToInt(b bool) int { if b { return 1 } return 0 } // NowStr 当前时间字符串(供日志)。 func NowStr() string { return gtime.Now().Format("Y-m-d H:i:s") }