feat(push): implement system notification and push framework

This commit is contained in:
ryan
2026-06-14 16:52:06 +08:00
parent aee457093d
commit cb018b3b60
60 changed files with 8756 additions and 160 deletions
+363
View File
@@ -0,0 +1,363 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package push
import (
"encoding/json"
"errors"
"net/http"
"strconv"
"strings"
"github.com/Rain-kl/Wavelet/internal/db"
"github.com/Rain-kl/Wavelet/internal/model"
"github.com/Rain-kl/Wavelet/internal/util"
pkgpush "github.com/Rain-kl/Wavelet/pkg/push"
"github.com/gin-gonic/gin"
"gorm.io/gorm"
)
// ListChannelDefinitions 获取各种消息通道的表单配置定义列表
// @Summary 获取所有消息通道配置字段定义
// @Description 返回系统支持的所有消息通道类型(如飞书、邮件、自定义)的动态表单定义,需要管理员权限
// @Tags admin-push
// @Produce json
// @Security SessionCookie
// @Success 200 {object} util.ResponseAny{data=[]Definition} "通道配置定义列表"
// @Router /api/v1/admin/push/channels/definitions [get]
func ListChannelDefinitions(c *gin.Context) {
c.JSON(http.StatusOK, util.OK(ListDefinitions()))
}
// ListChannels 获取消息通道列表
// @Summary 获取所有消息通道
// @Description 返回系统配置的所有消息通道列表,需要管理员权限
// @Tags admin-push
// @Produce json
// @Security SessionCookie
// @Success 200 {object} util.ResponseAny{data=[]model.PushChannel} "消息通道列表"
// @Router /api/v1/admin/push/channels [get]
func ListChannels(c *gin.Context) {
ctx := c.Request.Context()
var channels []model.PushChannel
if err := db.DB(ctx).Order("created_at DESC").Find(&channels).Error; err != nil {
c.JSON(http.StatusInternalServerError, util.Err(err.Error()))
return
}
c.JSON(http.StatusOK, util.OK(channels))
}
// CreateChannelRequest 创建通道参数
type CreateChannelRequest struct {
Name string `json:"name" binding:"required"`
Description string `json:"description"`
Type string `json:"type" binding:"required"`
Token string `json:"token"`
URL string `json:"url"`
Other string `json:"other"`
Enabled bool `json:"enabled"`
}
// CreateChannel 创建消息通道
// @Summary 创建消息通道
// @Description 新建一个消息通道配置,需要管理员权限
// @Tags admin-push
// @Accept json
// @Produce json
// @Security SessionCookie
// @Param request body CreateChannelRequest true "创建参数"
// @Success 200 {object} util.ResponseAny{data=model.PushChannel} "创建成功"
// @Router /api/v1/admin/push/channels [post]
func CreateChannel(c *gin.Context) {
var req CreateChannelRequest
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, util.Err(err.Error()))
return
}
ctx := c.Request.Context()
var count int64
if err := db.DB(ctx).Model(&model.PushChannel{}).Where("name = ?", req.Name).Count(&count).Error; err != nil {
c.JSON(http.StatusInternalServerError, util.Err(err.Error()))
return
}
if count > 0 {
c.JSON(http.StatusBadRequest, util.Err("channel name already exists"))
return
}
channel := model.PushChannel{
Name: req.Name,
Description: req.Description,
Type: req.Type,
Token: req.Token,
URL: req.URL,
Other: req.Other,
Enabled: req.Enabled,
}
if err := channel.Validate(); err != nil {
c.JSON(http.StatusBadRequest, util.Err(err.Error()))
return
}
if err := db.DB(ctx).Create(&channel).Error; err != nil {
c.JSON(http.StatusInternalServerError, util.Err(err.Error()))
return
}
c.JSON(http.StatusOK, util.OK(channel))
}
// UpdateChannelRequest 修改通道参数
type UpdateChannelRequest struct {
Description string `json:"description"`
Type string `json:"type" binding:"required"`
Token string `json:"token"`
URL string `json:"url"`
Other string `json:"other"`
Enabled bool `json:"enabled"`
}
// UpdateChannel 更新消息通道
// @Summary 更新消息通道
// @Description 修改消息通道配置,需要管理员权限
// @Tags admin-push
// @Accept json
// @Produce json
// @Security SessionCookie
// @Param id path uint64 true "通道ID"
// @Param request body UpdateChannelRequest true "更新参数"
// @Success 200 {object} util.ResponseAny{data=model.PushChannel} "更新成功"
// @Router /api/v1/admin/push/channels/{id} [put]
func UpdateChannel(c *gin.Context) {
idStr := c.Param("id")
id, err := strconv.ParseUint(idStr, 10, 64)
if err != nil {
c.JSON(http.StatusBadRequest, util.Err("invalid channel id"))
return
}
var req UpdateChannelRequest
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, util.Err(err.Error()))
return
}
ctx := c.Request.Context()
var channel model.PushChannel
if err := db.DB(ctx).Where("id = ?", id).First(&channel).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
c.JSON(http.StatusNotFound, util.Err("channel not found"))
return
}
c.JSON(http.StatusInternalServerError, util.Err(err.Error()))
return
}
channel.Description = req.Description
channel.Type = req.Type
channel.Token = req.Token
channel.URL = req.URL
channel.Other = req.Other
channel.Enabled = req.Enabled
if err := channel.Validate(); err != nil {
c.JSON(http.StatusBadRequest, util.Err(err.Error()))
return
}
if err := db.DB(ctx).Save(&channel).Error; err != nil {
c.JSON(http.StatusInternalServerError, util.Err(err.Error()))
return
}
c.JSON(http.StatusOK, util.OK(channel))
}
// DeleteChannel 删除消息通道
// @Summary 删除消息通道
// @Description 根据ID删除消息通道,需要管理员权限
// @Tags admin-push
// @Produce json
// @Security SessionCookie
// @Param id path uint64 true "通道ID"
// @Success 200 {object} util.ResponseAny "删除成功"
// @Router /api/v1/admin/push/channels/{id} [delete]
func DeleteChannel(c *gin.Context) {
idStr := c.Param("id")
id, err := strconv.ParseUint(idStr, 10, 64)
if err != nil {
c.JSON(http.StatusBadRequest, util.Err("invalid channel id"))
return
}
ctx := c.Request.Context()
var channel model.PushChannel
if err := db.DB(ctx).Where("id = ?", id).First(&channel).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
c.JSON(http.StatusNotFound, util.Err("channel not found"))
return
}
c.JSON(http.StatusInternalServerError, util.Err(err.Error()))
return
}
if err := db.DB(ctx).Delete(&channel).Error; err != nil {
c.JSON(http.StatusInternalServerError, util.Err(err.Error()))
return
}
c.JSON(http.StatusOK, util.OKNil())
}
// TestChannelRequest 测试通道连通性参数
type TestChannelRequest struct {
Name string `json:"name"`
Type string `json:"type"`
Token string `json:"token"`
URL string `json:"url"`
Other string `json:"other"`
Target string `json:"target"`
}
// TestChannel 测试通道连通性
// @Summary 测试通道连通性
// @Description 触发一次临时的或现有的通道连通性推送测试,需要管理员权限
// @Tags admin-push
// @Accept json
// @Produce json
// @Security SessionCookie
// @Param request body TestChannelRequest true "测试参数"
// @Success 200 {object} util.ResponseAny "测试触发成功"
// @Router /api/v1/admin/push/channels/test [post]
func TestChannel(c *gin.Context) {
var req TestChannelRequest
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, util.Err(err.Error()))
return
}
ctx := c.Request.Context()
var url, token, other, channelType string
if req.Name != "" {
var channel model.PushChannel
if err := db.DB(ctx).Where("name = ?", req.Name).First(&channel).Error; err != nil {
c.JSON(http.StatusBadRequest, util.Err("channel not found"))
return
}
url = channel.URL
token = channel.Token
other = channel.Other
channelType = channel.Type
} else {
url = req.URL
token = req.Token
other = req.Other
channelType = req.Type
}
// 对邮件类型应用全局配置作为回退
if channelType == channelEmail {
url, token, other = resolveSMTPConfig(ctx, url, token, other)
}
tempChannel := model.PushChannel{
Name: "test_temp",
URL: url,
Token: token,
Other: other,
Type: channelType,
Enabled: true,
}
if err := tempChannel.Validate(); err != nil {
c.JSON(http.StatusBadRequest, util.Err(err.Error()))
return
}
var config pkgpush.Config
var renderedJSON string
switch channelType {
case channelLark:
config = pkgpush.Config{
Channel: channelLark,
URL: url,
Secret: token,
}
renderedJSON = other
case channelEmail:
config = pkgpush.Config{
Channel: channelEmail,
URL: url,
Key: token,
Secret: other,
}
default:
config = pkgpush.Config{
Channel: channelCustom,
URL: url,
}
customPushReq := CustomPushRequest{
Title: "通道测试通知",
Content: "这是一条来自系统的消息通道连通性测试消息。",
Description: "系统通道测试",
URL: "https://example.com",
To: req.Target,
}
renderedJSON = renderCustomPayload(other, customPushReq)
}
payload := SendPayload{
EventKey: "test_channel",
Config: config,
Target: req.Target,
Body: NotificationMessage{
Title: "通道测试通知",
Content: "这是一条来自系统的消息通道连通性测试消息。",
Level: defaultLevelInfo,
},
Template: renderedJSON,
}
if err := enqueuePushTask(ctx, payload); err != nil {
c.JSON(http.StatusInternalServerError, util.Err(err.Error()))
return
}
c.JSON(http.StatusOK, util.OKNil())
}
// CustomPushRequest 外部公开推送请求参数
type CustomPushRequest struct {
Title string `json:"title" form:"title"`
Description string `json:"description" form:"description"`
Content string `json:"content" form:"content"`
URL string `json:"url" form:"url"`
To string `json:"to" form:"to"`
Token string `json:"token" form:"token"`
}
func escapeJSONString(s string) string {
b, _ := json.Marshal(s)
const minJSONLen = 2
if len(b) >= minJSONLen {
return string(b[1 : len(b)-1])
}
return s
}
func renderCustomPayload(template string, req CustomPushRequest) string {
result := template
result = strings.ReplaceAll(result, "$title", escapeJSONString(req.Title))
result = strings.ReplaceAll(result, "$description", escapeJSONString(req.Description))
result = strings.ReplaceAll(result, "$content", escapeJSONString(req.Content))
result = strings.ReplaceAll(result, "$url", escapeJSONString(req.URL))
result = strings.ReplaceAll(result, "$to", escapeJSONString(req.To))
return result
}
@@ -0,0 +1,150 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package push
import "sync"
const (
// KeyURL represents the URL field key
KeyURL = "url"
// KeyToken represents the Token field key
KeyToken = "token"
// KeyOther represents the Other field key
KeyOther = "other"
// TypeText represents standard text input type
TypeText = "text"
// TypePassword represents password input type
TypePassword = "password"
// TypeTextarea represents textarea input type
TypeTextarea = "textarea"
)
// Field represents a form field configuration for a channel.
type Field struct {
Key string `json:"key"` // unique key for the field (e.g. url, token, other)
Label string `json:"label"` // human readable label (e.g. "Webhook 地址")
Type string `json:"type"` // input type: "text" | "password" | "textarea"
Required bool `json:"required"` // whether this field is required
Placeholder string `json:"placeholder"` // input placeholder
Description string `json:"description"` // field explanation/help text
}
// Definition represents the metadata and form schema for a notification channel.
type Definition struct {
Type string `json:"type"` // channel type (e.g., custom, lark, email)
Name string `json:"name"` // display name
Description string `json:"description"` // short description
Fields []Field `json:"fields"` // form fields
}
var (
defMu sync.RWMutex
definitions = make(map[string]Definition)
)
// RegisterChannelDefinition registers a channel definition.
func RegisterChannelDefinition(def Definition) {
defMu.Lock()
defer defMu.Unlock()
definitions[def.Type] = def
}
// ListDefinitions returns all registered channel definitions.
func ListDefinitions() []Definition {
defMu.RLock()
defer defMu.RUnlock()
// We want a stable order: custom, lark, email
order := []string{channelCustom, channelLark, channelEmail}
res := make([]Definition, 0, len(definitions))
for _, t := range order {
if d, ok := definitions[t]; ok {
res = append(res, d)
}
}
// Add any others
for t, d := range definitions {
found := false
for _, o := range order {
if o == t {
found = true
break
}
}
if !found {
res = append(res, d)
}
}
return res
}
func init() {
// Register custom webhook channel
RegisterChannelDefinition(Definition{
Type: channelCustom,
Name: "自定义消息通道",
Description: "使用自定义 HTTP POST 请求向外部 Webhook 发送数据。",
Fields: []Field{
{
Key: KeyURL,
Label: "请求地址",
Type: TypeText,
Required: true,
Placeholder: "在此填写完整的请求地址,必须使用 HTTPS 协议",
Description: "接口请求的完整 HTTPS URL,例如 https://api.example.com/webhook",
},
{
Key: KeyOther,
Label: "请求体 (JSON)",
Type: TypeTextarea,
Required: true,
Placeholder: "在此输入请求体,支持模板变量,必须为合法的 JSON 格式",
Description: "可使用的变量:$title, $description, $content, $url, $to。例如 {\"text\": \"$content\"}",
},
},
})
// Register Lark robot channel
RegisterChannelDefinition(Definition{
Type: channelLark,
Name: "飞书群机器人",
Description: "配置飞书群自定义机器人的 Webhook 接口投递。",
Fields: []Field{
{
Key: KeyURL,
Label: "Webhook 地址",
Type: TypeText,
Required: true,
Placeholder: "https://open.feishu.cn/open-apis/bot/v2/hook/YOUR_TOKEN",
Description: "从飞书群机器人设置中复制 of Webhook URL",
// Note: using 'of' was in feishu.go, let's keep original wording or fix it
},
{
Key: KeyToken,
Label: "签名校验密钥 (Secret) (可选)",
Type: TypeText,
Required: false,
Placeholder: "可选,若机器人启用了安全设置中的签名校验,请在此输入",
Description: "飞书群机器人安全设置中的签名校验 Key",
},
{
Key: KeyOther,
Label: "自定义卡片 JSON 模版 (可选)",
Type: TypeTextarea,
Required: false,
Placeholder: "可选,留空则默认使用系统内置的精美互动卡片",
Description: "若填写,必须是合法的飞书卡片 JSON 格式",
},
},
})
// Register Email channel
RegisterChannelDefinition(Definition{
Type: channelEmail,
Name: "邮件推送通道",
Description: "邮件推送通道直接使用系统全局 SMTP 设置进行发送,无需在此填写服务器配置。",
Fields: []Field{},
})
}
+14
View File
@@ -0,0 +1,14 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package push
const (
channelCustom = "custom"
channelEmail = "email"
channelLark = "lark"
defaultLevelInfo = "INFO"
keyTitle = "title"
keyContent = "content"
keyLevel = "level"
)
@@ -0,0 +1,43 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
// Package custom_events defines custom push notification events.
package custom_events
import (
"context"
"time"
"github.com/Rain-kl/Wavelet/internal/apps/admin/push"
"github.com/Rain-kl/Wavelet/internal/model"
)
// AdminLogin is the metadata definition for the admin login event.
var AdminLogin = push.EventMetadata{
Key: "admin_login",
Name: "管理员登录",
DefaultTemplate: push.NotificationMessage{
Title: "管理员登录提醒",
Content: "管理员 {{user.username}} 于 {{time}} 从 IP {{ip}} 登录系统。",
Level: "INFO",
},
Description: "当管理员成功登录系统时触发此通知",
}
func init() {
push.RegisterBuiltInEvent(AdminLogin)
}
// TriggerAdminLoginEvent triggers the admin login event asynchronously.
func TriggerAdminLoginEvent(ctx context.Context, user *model.User, ip string) {
if user == nil || !user.IsAdmin {
return
}
body := map[string]any{
"user": user,
"ip": ip,
"time": time.Now().Format("2006-01-02 15:04:05"),
}
push.DefaultTrigger.Trigger(ctx, AdminLogin, body)
}
+599
View File
@@ -0,0 +1,599 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
// Package push defines push notification HTTP routes, background tasks, and events.
package push
import (
"context"
"encoding/json"
"errors"
"fmt"
"strconv"
"strings"
"sync"
"github.com/Rain-kl/Wavelet/internal/db"
"github.com/Rain-kl/Wavelet/internal/logger"
"github.com/Rain-kl/Wavelet/internal/model"
"github.com/Rain-kl/Wavelet/internal/task"
pkgpush "github.com/Rain-kl/Wavelet/pkg/push"
"gorm.io/gorm"
)
// NotificationMessage represents the structured notification message payload.
type NotificationMessage struct {
Title string `json:"title"`
Content string `json:"content"`
Level string `json:"level"`
Ext map[string]any `json:"ext,omitempty"`
}
// Flatten converts the structured NotificationMessage back to a flat map (original json structure).
func (m NotificationMessage) Flatten() map[string]any {
res := map[string]any{
keyTitle: m.Title,
keyContent: m.Content,
keyLevel: m.Level,
}
for k, v := range m.Ext {
res[k] = v
}
return res
}
// EventMetadata represents the metadata of a push notification event.
type EventMetadata struct {
Key string `json:"key"`
Name string `json:"name"`
DefaultTemplate NotificationMessage `json:"default_template"`
Description string `json:"description"`
}
// SendPayload 异步投递推送载荷 (供 task/Worker 使用)
type SendPayload struct {
EventKey string `json:"event_key"`
Config pkgpush.Config `json:"config"`
Target string `json:"target"`
Body NotificationMessage `json:"body"`
Template string `json:"template"`
}
// BuiltInEvents lists all built-in events defined in custom_events.
var BuiltInEvents []EventMetadata
// RegisterBuiltInEvent registers a built-in event definition.
func RegisterBuiltInEvent(meta EventMetadata) {
BuiltInEvents = append(BuiltInEvents, meta)
}
// EventTrigger represents the unified event trigger class.
type EventTrigger struct{}
// DefaultTrigger is the singleton instance of EventTrigger.
var DefaultTrigger = &EventTrigger{}
var (
systemUser *model.User
systemOnce sync.Once
)
func getSystemUser(ctx context.Context) *model.User {
systemOnce.Do(func() {
var u model.User
if err := db.DB(ctx).Where("username = ?", "system").First(&u).Error; err == nil {
systemUser = &u
} else {
systemUser = &model.User{
ID: 999,
Username: "system",
Nickname: "系统",
Email: "",
}
}
})
return systemUser
}
// Trigger receives event metadata and processes the event notification dispatch asynchronously.
// It automatically enqueues tasks using a background goroutine and avoids blocking the calling thread.
//
//nolint:contextcheck
func (t *EventTrigger) Trigger(ctx context.Context, meta EventMetadata, body map[string]any) {
asyncCtx := context.WithoutCancel(ctx)
go func() {
if body == nil {
body = make(map[string]any)
}
if _, hasUser := body["user"]; !hasUser || body["user"] == nil {
body["user"] = getSystemUser(asyncCtx)
}
// 1. Check if the event is enabled in the database
var event model.PushEvent
err := db.DB(asyncCtx).Where("event_key = ? AND enabled = ?", meta.Key, true).First(&event).Error
if err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return
}
logger.ErrorF(asyncCtx, "push_event_trigger: failed to get active event %s: %v", meta.Key, err)
return
}
if len(event.Channels) == 0 {
return
}
// 2. Read push configs
configs, err := t.getPushConfigs(asyncCtx)
if err != nil {
logger.ErrorF(asyncCtx, "push_event_trigger: getPushConfigs failed: %v", err)
return
}
// 3. Build and render notification message
flatBody := getFlatBody(body)
msg, renderedTemplate := t.buildMessage(&event, meta, flatBody, body)
// 4. Enqueue tasks for each matching channel
t.enqueuePushTasks(asyncCtx, meta, &event, configs, msg, renderedTemplate, flatBody)
}()
}
func (t *EventTrigger) getPushConfigs(ctx context.Context) ([]pkgpush.Config, error) {
var configVal string
var sc model.SystemConfig
if err := sc.GetByKey(ctx, model.ConfigKeyPushConfig); err == nil {
configVal = sc.Value
}
if configVal == "" || configVal == "[]" {
return nil, errors.New("push_config is empty or not configured")
}
var configs []pkgpush.Config
if err := json.Unmarshal([]byte(configVal), &configs); err != nil {
return nil, fmt.Errorf("unmarshal push_config failed: %w", err)
}
return configs, nil
}
func (t *EventTrigger) buildMessage(event *model.PushEvent, meta EventMetadata, flatBody map[string]any, body map[string]any) (NotificationMessage, string) {
var msg NotificationMessage
renderedTemplate := ""
templateSource := event.Template
if templateSource != "" {
var err error
msg, renderedTemplate, err = t.parseCustomTemplate(event, templateSource, flatBody)
if err != nil {
msg.Title = event.Name
msg.Content = renderedTemplate
msg.Level = defaultLevelInfo
}
} else {
msg = t.parseDefaultTemplate(meta, flatBody)
}
if msg.Ext == nil {
msg.Ext = make(map[string]any)
}
for k, v := range body {
if k == keyTitle || k == keyContent || k == keyLevel {
continue
}
if _, exists := msg.Ext[k]; !exists {
msg.Ext[k] = v
}
}
return msg, renderedTemplate
}
func (t *EventTrigger) parseCustomTemplate(event *model.PushEvent, templateSource string, flatBody map[string]any) (NotificationMessage, string, error) {
var msg NotificationMessage
renderedTemplate := pkgpush.ParseTemplate(templateSource, flatBody)
var tMap map[string]any
if err := json.Unmarshal([]byte(renderedTemplate), &tMap); err != nil {
return msg, renderedTemplate, err
}
if title, ok := tMap[keyTitle].(string); ok && title != "" {
msg.Title = title
} else {
msg.Title = event.Name
}
delete(tMap, keyTitle)
if content, ok := tMap[keyContent].(string); ok && content != "" {
msg.Content = content
} else {
msg.Content = renderedTemplate
}
delete(tMap, keyContent)
if level, ok := tMap[keyLevel].(string); ok && level != "" {
msg.Level = level
} else {
msg.Level = defaultLevelInfo
}
delete(tMap, keyLevel)
msg.Ext = tMap
return msg, renderedTemplate, nil
}
func (t *EventTrigger) parseDefaultTemplate(meta EventMetadata, flatBody map[string]any) NotificationMessage {
var msg NotificationMessage
msg.Title = pkgpush.ParseTemplate(meta.DefaultTemplate.Title, flatBody)
msg.Content = pkgpush.ParseTemplate(meta.DefaultTemplate.Content, flatBody)
msg.Level = pkgpush.ParseTemplate(meta.DefaultTemplate.Level, flatBody)
if meta.DefaultTemplate.Ext != nil {
msg.Ext = make(map[string]any)
for k, v := range meta.DefaultTemplate.Ext {
if strVal, ok := v.(string); ok {
msg.Ext[k] = pkgpush.ParseTemplate(strVal, flatBody)
} else {
msg.Ext[k] = v
}
}
}
return msg
}
func (t *EventTrigger) enqueuePushTasks(ctx context.Context, meta EventMetadata, event *model.PushEvent, configs []pkgpush.Config, msg NotificationMessage, renderedTemplate string, flatBody map[string]any) {
for _, channelName := range event.Channels {
if channelName == channelEmail {
t.enqueueEmailPushTasks(ctx, meta, event, configs, msg, renderedTemplate, flatBody)
continue
}
// 检查是不是自定义数据库渠道
var customChannel model.PushChannel
err := db.DB(ctx).Where("name = ? AND enabled = ?", channelName, true).First(&customChannel).Error
if err == nil {
t.enqueueCustomPushChannelTasks(ctx, meta, event, &customChannel, msg, flatBody)
continue
}
// 数据库中不存在。我们核对它是不是通过代码内置注册的 Pusher 渠道
if _, errPusher := pkgpush.GetPusher(channelName); errPusher == nil {
t.enqueueBuiltinPushTasks(ctx, meta, event, channelName, configs, msg, renderedTemplate, flatBody)
continue
}
logger.WarnF(ctx, "push_event_trigger: channel %q not found in DB and not registered as built-in: %v", channelName, err)
}
}
func (t *EventTrigger) enqueueEmailPushTasks(ctx context.Context, meta EventMetadata, event *model.PushEvent, configs []pkgpush.Config, msg NotificationMessage, renderedTemplate string, flatBody map[string]any) {
var matchedConfigs []pkgpush.Config
for _, cfg := range configs {
if cfg.Channel == channelEmail {
matchedConfigs = append(matchedConfigs, cfg)
}
}
if len(matchedConfigs) == 0 {
logger.WarnF(ctx, "push_event_trigger: no active settings for channel %q", channelEmail)
return
}
for _, cfg := range matchedConfigs {
if cfg.URL == "" || cfg.Key == "" {
var smtpHost, smtpPort, smtpUser, smtpPass model.SystemConfig
_ = smtpHost.GetByKey(ctx, model.ConfigKeySMTPHost)
_ = smtpPort.GetByKey(ctx, model.ConfigKeySMTPPort)
_ = smtpUser.GetByKey(ctx, model.ConfigKeySMTPUsername)
_ = smtpPass.GetByKey(ctx, model.ConfigKeySMTPPassword)
if smtpHost.Value != "" && smtpUser.Value != "" {
port := smtpPort.Value
if port == "" {
port = "587"
}
cfg.URL = smtpHost.Value + ":" + port
cfg.Key = smtpUser.Value
cfg.Secret = smtpPass.Value
}
}
if len(event.Targets) > 0 {
for _, target := range event.Targets {
resolvedTarget := resolveTarget(ctx, target, flatBody, channelEmail)
payload := SendPayload{
EventKey: meta.Key,
Config: cfg,
Target: resolvedTarget,
Body: msg,
Template: renderedTemplate,
}
if err := enqueuePushTask(ctx, payload); err != nil {
logger.ErrorF(ctx, "push_event_trigger: enqueuePushTask failed for %s -> %s: %v", channelEmail, resolvedTarget, err)
}
}
} else {
payload := SendPayload{
EventKey: meta.Key,
Config: cfg,
Target: "",
Body: msg,
Template: renderedTemplate,
}
if err := enqueuePushTask(ctx, payload); err != nil {
logger.ErrorF(ctx, "push_event_trigger: enqueuePushTask failed for %s: %v", channelEmail, err)
}
}
}
}
func (t *EventTrigger) enqueueCustomPushChannelTasks(ctx context.Context, meta EventMetadata, event *model.PushEvent, channel *model.PushChannel, msg NotificationMessage, flatBody map[string]any) {
if len(event.Targets) == 0 {
t.enqueueSingleCustomPushChannelTask(ctx, meta, channel, "", msg)
return
}
for _, target := range event.Targets {
resolvedTarget := resolveTarget(ctx, target, flatBody, channel.Name)
t.enqueueSingleCustomPushChannelTask(ctx, meta, channel, resolvedTarget, msg)
}
}
func (t *EventTrigger) enqueueSingleCustomPushChannelTask(ctx context.Context, meta EventMetadata, channel *model.PushChannel, target string, msg NotificationMessage) {
var config pkgpush.Config
var renderedTemplate string
switch channel.Type {
case channelLark:
config = pkgpush.Config{
Channel: channelLark,
URL: channel.URL,
Secret: channel.Token, // Feishu Bot Sign Secret
}
renderedTemplate = channel.Other // Optional custom template/card for lark
case channelEmail:
url, token, other := resolveSMTPConfig(ctx, channel.URL, channel.Token, channel.Other)
config = pkgpush.Config{
Channel: channelEmail,
URL: url, // SMTP host:port
Key: token, // SMTP Username
Secret: other, // SMTP Password
}
default: // custom
config = pkgpush.Config{
Channel: channelCustom,
URL: channel.URL,
}
customPushReq := CustomPushRequest{
Title: msg.Title,
Content: msg.Content,
Description: meta.Description,
To: target,
}
if urlVal, ok := msg.Ext["url"].(string); ok {
customPushReq.URL = urlVal
}
renderedTemplate = renderCustomPayload(channel.Other, customPushReq)
}
payload := SendPayload{
EventKey: meta.Key,
Config: config,
Target: target,
Body: msg,
Template: renderedTemplate,
}
if err := enqueuePushTask(ctx, payload); err != nil {
logger.ErrorF(ctx, "push_event_trigger: enqueuePushTask failed for %s channel %s -> %s: %v", channel.Type, channel.Name, target, err)
}
}
func (t *EventTrigger) enqueueBuiltinPushTasks(ctx context.Context, meta EventMetadata, event *model.PushEvent, channelName string, configs []pkgpush.Config, msg NotificationMessage, renderedTemplate string, flatBody map[string]any) {
var matchedConfigs []pkgpush.Config
for _, cfg := range configs {
if cfg.Channel == channelName {
matchedConfigs = append(matchedConfigs, cfg)
}
}
if len(matchedConfigs) == 0 {
logger.WarnF(ctx, "push_event_trigger: no active settings for built-in channel %q", channelName)
return
}
for _, cfg := range matchedConfigs {
if len(event.Targets) > 0 {
for _, target := range event.Targets {
resolvedTarget := resolveTarget(ctx, target, flatBody, channelName)
payload := SendPayload{
EventKey: meta.Key,
Config: cfg,
Target: resolvedTarget,
Body: msg,
Template: renderedTemplate,
}
if err := enqueuePushTask(ctx, payload); err != nil {
logger.ErrorF(ctx, "push_event_trigger: enqueuePushTask failed for builtin %s -> %s: %v", channelName, resolvedTarget, err)
}
}
} else {
payload := SendPayload{
EventKey: meta.Key,
Config: cfg,
Target: "",
Body: msg,
Template: renderedTemplate,
}
if err := enqueuePushTask(ctx, payload); err != nil {
logger.ErrorF(ctx, "push_event_trigger: enqueuePushTask failed for builtin %s: %v", channelName, err)
}
}
}
}
func enqueuePushTask(ctx context.Context, payload SendPayload) error {
payloadBytes, err := json.Marshal(payload)
if err != nil {
return err
}
_, err = task.DispatchTask(ctx, "send_notification", payloadBytes, "system")
return err
}
func getFlatBody(body map[string]any) map[string]any {
jsonBytes, err := json.Marshal(body)
if err != nil {
return body
}
var jsonMap map[string]any
if err := json.Unmarshal(jsonBytes, &jsonMap); err != nil {
return body
}
flatResult := make(map[string]any)
flattenMap("", jsonMap, flatResult)
return flatResult
}
func flattenMap(prefix string, m map[string]any, result map[string]any) {
for k, v := range m {
key := k
if prefix != "" {
key = prefix + "." + k
}
if nestedMap, ok := v.(map[string]any); ok {
flattenMap(key, nestedMap, result)
} else {
result[key] = v
}
}
}
func resolveTarget(ctx context.Context, target string, flatBody map[string]any, channel string) string {
target = strings.TrimSpace(target)
if target == "" {
return ""
}
resolved := resolveDynamicKeyword(target, flatBody)
// 2. 如果包含 @,说明已经是个邮箱,直接返回
if strings.Contains(resolved, "@") {
return resolved
}
// 2.5 如果为特殊的系统虚拟用户,自动映射为首位管理员
if val, matched := resolveSystemTarget(ctx, resolved, channel); matched {
return val
}
// 3. 不包含 @,说明可能是用户 ID 或用户名。我们需要从数据库中查询对应用户
var user model.User
found := false
// 尝试作为用户 ID 查询(纯数字)
if id, err := strconv.ParseUint(resolved, 10, 64); err == nil {
if err := db.DB(ctx).Where("id = ?", id).First(&user).Error; err == nil {
found = true
}
}
// 如果没有按 ID 查到,尝试作为用户名查询
if !found {
if err := db.DB(ctx).Where("username = ?", resolved).First(&user).Error; err == nil {
found = true
}
}
// 4. 根据查询结果 and 推送渠道进行转换
if !found {
return resolved
}
if channel == channelEmail && user.Email != "" {
return user.Email
}
if channel != channelEmail && user.Username != "" {
return user.Username
}
return resolved
}
func resolveDynamicKeyword(target string, flatBody map[string]any) string {
switch target {
case "user.id", "id":
if val, ok := flatBody["user.id"]; ok {
return fmt.Sprintf("%v", val)
}
if val, ok := flatBody["id"]; ok {
return fmt.Sprintf("%v", val)
}
case "user.username", "username":
if val, ok := flatBody["user.username"]; ok {
return fmt.Sprintf("%v", val)
}
if val, ok := flatBody["username"]; ok {
return fmt.Sprintf("%v", val)
}
case "user.email", channelEmail:
if val, ok := flatBody["user.email"]; ok {
return fmt.Sprintf("%v", val)
}
if val, ok := flatBody["email"]; ok {
return fmt.Sprintf("%v", val)
}
}
return target
}
func resolveSystemTarget(ctx context.Context, resolved string, channel string) (string, bool) {
if resolved != "系统" && resolved != "system" && resolved != "0" {
return "", false
}
var adminUser model.User
if err := db.DB(ctx).Where("is_admin = ?", true).Order("id asc").First(&adminUser).Error; err != nil {
return resolved, true
}
if channel == channelEmail && adminUser.Email != "" {
return adminUser.Email, true
}
if channel != channelEmail && adminUser.Username != "" {
return adminUser.Username, true
}
return resolved, true
}
// resolveSMTPConfig resolves SMTP configuration by falling back to system-wide global configuration if any inputs are blank.
func resolveSMTPConfig(ctx context.Context, url, token, other string) (string, string, string) {
if url != "" && token != "" {
return url, token, other
}
var smtpHost, smtpPort, smtpUser, smtpPass model.SystemConfig
_ = smtpHost.GetByKey(ctx, model.ConfigKeySMTPHost)
_ = smtpPort.GetByKey(ctx, model.ConfigKeySMTPPort)
_ = smtpUser.GetByKey(ctx, model.ConfigKeySMTPUsername)
_ = smtpPass.GetByKey(ctx, model.ConfigKeySMTPPassword)
if smtpHost.Value == "" || smtpUser.Value == "" {
return url, token, other
}
port := smtpPort.Value
if port == "" {
port = "587"
}
if url == "" {
url = smtpHost.Value + ":" + port
}
if token == "" {
token = smtpUser.Value
}
if other == "" {
other = smtpPass.Value
}
return url, token, other
}
+737
View File
@@ -0,0 +1,737 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package push
import (
"bytes"
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"strconv"
"sync"
"testing"
"time"
"github.com/Rain-kl/Wavelet/internal/model"
"github.com/Rain-kl/Wavelet/internal/task"
"github.com/Rain-kl/Wavelet/internal/testhelper"
"github.com/Rain-kl/Wavelet/internal/util"
pkgpush "github.com/Rain-kl/Wavelet/pkg/push"
"github.com/alicebob/miniredis/v2"
"github.com/gin-gonic/gin"
"github.com/hibiken/asynq"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"gorm.io/gorm"
)
var adminLoginEvent = EventMetadata{
Key: "admin_login",
Name: "管理员登录",
DefaultTemplate: NotificationMessage{
Title: "管理员登录提醒",
Content: "管理员 {{user.username}} 于 {{time}} 从 IP {{ip}} 登录系统。",
Level: "INFO",
},
Description: "当管理员成功登录系统时触发此通知",
}
func init() {
RegisterBuiltInEvent(adminLoginEvent)
}
// mockPusher mock implementation of pkgpush.Pusher
type mockPusher struct {
mu sync.Mutex
sentBody map[string]any
sentTgt string
}
func (m *mockPusher) Send(ctx context.Context, cfg pkgpush.Config, target string, body map[string]any, template string, ext map[string]any) error {
m.mu.Lock()
defer m.mu.Unlock()
m.sentBody = body
m.sentTgt = target
return nil
}
func (m *mockPusher) ValidateConfig(cfg pkgpush.Config) error {
return nil
}
func setupPushTest(t *testing.T) (*gorm.DB, *miniredis.Miniredis, func()) {
dbConn, mr, cleanup := testhelper.SetupTestEnvironment(t)
// AutoMigrate push tables in SQLite test environment
err := dbConn.AutoMigrate(&model.PushEvent{}, &model.PushHistory{}, &model.User{}, &model.PushChannel{}, &model.SystemConfig{})
require.NoError(t, err)
// 写入数据库系统默认用户 Seed 记录
sysUser := &model.User{
ID: 999,
Username: "system",
Nickname: "系统",
Password: "*",
IsActive: true,
}
err = dbConn.Create(sysUser).Error
require.NoError(t, err)
// Initialize AsynqClient pointing to miniredis
task.AsynqClient = asynq.NewClient(asynq.RedisClientOpt{
Addr: mr.Addr(),
})
// Register the task handler and metadata
task.RegisterHandler(SendNotificationTask, &PushHandler{})
task.RegisterTaskMeta(SendNotificationMeta)
return dbConn, mr, func() {
cleanup()
if task.AsynqClient != nil {
task.AsynqClient.Close()
task.AsynqClient = nil
}
}
}
func setupTestRouter(authUser *model.User) *gin.Engine {
gin.SetMode(gin.TestMode)
r := gin.New()
adminGroup := r.Group("/api/v1/admin/push")
adminGroup.Use(func(c *gin.Context) {
if authUser != nil {
util.SetToContext(c, "user_obj", authUser)
}
c.Next()
})
adminGroup.GET("/events", ListEvents)
adminGroup.GET("/events/builtin", ListBuiltInEvents)
adminGroup.POST("/events", CreateEvent)
adminGroup.PUT("/events/:id", UpdateEvent)
adminGroup.DELETE("/events/:id", DeleteEvent)
adminGroup.POST("/events/:id/toggle", ToggleEvent)
adminGroup.GET("/histories", ListHistories)
adminGroup.POST("/test", TestPush)
return r
}
func TestSyncEvents(t *testing.T) {
dbConn, _, cleanup := setupPushTest(t)
defer cleanup()
// 1. SyncEvents first time
err := SyncEvents(context.Background())
require.NoError(t, err)
// Verify event exists in DB
var event model.PushEvent
err = dbConn.Where("event_key = ?", "admin_login").First(&event).Error
require.NoError(t, err)
assert.Equal(t, "管理员登录", event.Name)
assert.False(t, event.Enabled)
// Verify DefaultTemplate matches GORM template field
var defaultMsg NotificationMessage
err = json.Unmarshal([]byte(event.Template), &defaultMsg)
require.NoError(t, err)
assert.Equal(t, adminLoginEvent.DefaultTemplate.Title, defaultMsg.Title)
assert.Equal(t, adminLoginEvent.DefaultTemplate.Content, defaultMsg.Content)
}
func TestEventTrigger(t *testing.T) {
dbConn, _, cleanup := setupPushTest(t)
defer cleanup()
// Register mock pusher
mPusher := &mockPusher{}
pkgpush.Register("mock_channel", mPusher)
// SyncEvents
err := SyncEvents(context.Background())
require.NoError(t, err)
t.Run("trigger disabled event silently ignored", func(t *testing.T) {
body := map[string]any{
"user": map[string]any{"username": "test_admin"},
"ip": "127.0.0.1",
}
DefaultTrigger.Trigger(context.Background(), adminLoginEvent, body)
// Sleep briefly since Trigger runs in goroutine
time.Sleep(50 * time.Millisecond)
// Verify no tasks enqueued in TaskExecution GORM table
var count int64
dbConn.Model(&model.TaskExecution{}).Count(&count)
assert.Equal(t, int64(0), count)
})
t.Run("trigger enabled event enqueues task", func(t *testing.T) {
// Set push_config system configuration
cfgJson := `[{"channel": "mock_channel", "url": "http://mock"}]`
sysConfig := &model.SystemConfig{
Key: model.ConfigKeyPushConfig,
Value: cfgJson,
}
err = dbConn.Create(sysConfig).Error
require.NoError(t, err)
// Enable the push event in DB using struct to trigger JSON serializer
var event model.PushEvent
err = dbConn.Where("event_key = ?", "admin_login").First(&event).Error
require.NoError(t, err)
event.Enabled = true
event.Channels = []string{"mock_channel"}
event.Targets = []string{"admin_user"}
err = dbConn.Save(&event).Error
require.NoError(t, err)
// Trigger
body := map[string]any{
"user": map[string]any{
"username": "super_admin",
},
"ip": "1.1.1.1",
"time": "2026-06-14 18:00:00",
}
DefaultTrigger.Trigger(context.Background(), adminLoginEvent, body)
// Wait for goroutine execution
time.Sleep(50 * time.Millisecond)
// Verify TaskExecution enqueued record
var execution model.TaskExecution
err = dbConn.Where("task_type = ?", SendNotificationTask).First(&execution).Error
require.NoError(t, err)
// Verify enqueued payload structure
var payload SendPayload
err = json.Unmarshal([]byte(execution.Payload), &payload)
require.NoError(t, err)
assert.Equal(t, "admin_login", payload.EventKey)
assert.Equal(t, "mock_channel", payload.Config.Channel)
assert.Equal(t, "admin_user", payload.Target)
assert.Equal(t, "管理员登录提醒", payload.Body.Title)
assert.Contains(t, payload.Body.Content, "super_admin")
assert.Contains(t, payload.Body.Content, "1.1.1.1")
})
t.Run("trigger without user injects virtual system user", func(t *testing.T) {
// Set push_config system configuration
dbConn.Where("key = ?", model.ConfigKeyPushConfig).Delete(&model.SystemConfig{})
cfgJson := `[{"channel": "mock_channel", "url": "http://mock"}]`
sysConfig := &model.SystemConfig{
Key: model.ConfigKeyPushConfig,
Value: cfgJson,
}
err = dbConn.Create(sysConfig).Error
require.NoError(t, err)
// Enable the push event in DB
var event model.PushEvent
err = dbConn.Where("event_key = ?", "admin_login").First(&event).Error
require.NoError(t, err)
// 清理旧任务执行记录
dbConn.Where("task_type = ?", SendNotificationTask).Delete(&model.TaskExecution{})
event.Enabled = true
event.Channels = []string{"mock_channel"}
event.Targets = []string{"user.username"} // 动态目标
err = dbConn.Save(&event).Error
require.NoError(t, err)
// Trigger with empty body (simulates cron scheduler triggering)
DefaultTrigger.Trigger(context.Background(), adminLoginEvent, nil)
// Wait for goroutine execution
time.Sleep(50 * time.Millisecond)
// Verify TaskExecution enqueued record
var execution model.TaskExecution
err = dbConn.Where("task_type = ?", SendNotificationTask).First(&execution).Error
require.NoError(t, err)
var payload SendPayload
err = json.Unmarshal([]byte(execution.Payload), &payload)
require.NoError(t, err)
// 检查 payload 是否将 target (user.username) 成功替换为 "system"
assert.Equal(t, "system", payload.Target)
// 检查 payload 中的 Content,应当被替换为 "system" 变量
assert.Contains(t, payload.Body.Content, "system")
})
}
func TestPushHandler(t *testing.T) {
dbConn, _, cleanup := setupPushTest(t)
defer cleanup()
mPusher := &mockPusher{}
pkgpush.Register("mock_channel", mPusher)
handler := &PushHandler{}
payload := SendPayload{
EventKey: "admin_login",
Config: pkgpush.Config{
Channel: "mock_channel",
URL: "http://mock-url",
},
Target: "admin_user",
Body: NotificationMessage{
Title: "Structured Alert",
Content: "Hello World",
Level: "WARNING",
Ext: map[string]any{"extra_val": 42},
},
}
payloadBytes, err := json.Marshal(payload)
require.NoError(t, err)
t.Run("validate payload", func(t *testing.T) {
validated, valErr := handler.ValidatePayload(payloadBytes)
require.NoError(t, valErr)
assert.NotEmpty(t, validated)
})
t.Run("execute task successfully", func(t *testing.T) {
res, execErr := handler.Execute(context.Background(), payloadBytes)
require.NoError(t, execErr)
assert.Contains(t, res.Message, "推送成功")
// Verify mock pusher received flattened variables
mPusher.mu.Lock()
assert.Equal(t, "admin_user", mPusher.sentTgt)
assert.Equal(t, "Structured Alert", mPusher.sentBody["title"])
assert.Equal(t, "Hello World", mPusher.sentBody["content"])
assert.Equal(t, "WARNING", mPusher.sentBody["level"])
assert.Equal(t, float64(42), mPusher.sentBody["extra_val"]) // unmarshaled json numbers are float64 by default
mPusher.mu.Unlock()
// Verify PushHistory recorded
var history model.PushHistory
err = dbConn.First(&history).Error
require.NoError(t, err)
assert.Equal(t, "admin_login", history.EventKey)
assert.Equal(t, "mock_channel", history.Channel)
assert.Equal(t, "success", history.Status)
assert.Equal(t, "Structured Alert", history.Title)
})
}
func TestPushRouters(t *testing.T) {
dbConn, _, cleanup := setupPushTest(t)
defer cleanup()
adminUser := &model.User{ID: 1001, Username: "admin", IsAdmin: true}
r := setupTestRouter(adminUser)
// Sync events to populate db
err := SyncEvents(context.Background())
require.NoError(t, err)
t.Run("list events", func(t *testing.T) {
req, _ := http.NewRequest("GET", "/api/v1/admin/push/events", nil)
w := httptest.NewRecorder()
r.ServeHTTP(w, req)
assert.Equal(t, http.StatusOK, w.Code)
var resp util.ResponseAny
err = json.Unmarshal(w.Body.Bytes(), &resp)
require.NoError(t, err)
dataBytes, _ := json.Marshal(resp.Data)
var events []model.PushEvent
err = json.Unmarshal(dataBytes, &events)
require.NoError(t, err)
assert.Len(t, events, 1)
assert.Equal(t, "admin_login", events[0].EventKey)
})
t.Run("toggle event status", func(t *testing.T) {
var event model.PushEvent
dbConn.First(&event)
req, _ := http.NewRequest("POST", "/api/v1/admin/push/events/"+strconv.FormatUint(event.ID, 10)+"/toggle", nil)
w := httptest.NewRecorder()
r.ServeHTTP(w, req)
assert.Equal(t, http.StatusOK, w.Code)
var updated model.PushEvent
dbConn.First(&updated)
assert.True(t, updated.Enabled)
})
t.Run("update event", func(t *testing.T) {
var event model.PushEvent
dbConn.First(&event)
updateReq := UpdateEventRequest{
Channels: []string{"email"},
Targets: []string{"user@test.com"},
Template: `{"title": "Custom Login Alert", "content": "Alert", "level": "WARNING"}`,
Enabled: true,
}
bodyBytes, _ := json.Marshal(updateReq)
req, _ := http.NewRequest("PUT", "/api/v1/admin/push/events/"+strconv.FormatUint(event.ID, 10), bytes.NewBuffer(bodyBytes))
req.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
r.ServeHTTP(w, req)
assert.Equal(t, http.StatusOK, w.Code)
var updated model.PushEvent
dbConn.First(&updated)
assert.Equal(t, []string{"email"}, updated.Channels)
assert.Equal(t, []string{"user@test.com"}, updated.Targets)
assert.Contains(t, updated.Template, "Custom Login Alert")
})
t.Run("list push histories", func(t *testing.T) {
// Populate history record
hist := model.PushHistory{
EventKey: "admin_login",
Channel: "email",
Target: "user@test.com",
Title: "Custom Login Alert",
Content: "Alert",
Level: "WARNING",
Status: "success",
}
dbConn.Create(&hist)
req, _ := http.NewRequest("GET", "/api/v1/admin/push/histories?page=1&page_size=10", nil)
w := httptest.NewRecorder()
r.ServeHTTP(w, req)
assert.Equal(t, http.StatusOK, w.Code)
var resp util.ResponseAny
json.Unmarshal(w.Body.Bytes(), &resp)
dataMap, ok := resp.Data.(map[string]any)
assert.True(t, ok)
assert.Equal(t, float64(1), dataMap["total"])
})
t.Run("test push endpoint", func(t *testing.T) {
mPusher := &mockPusher{}
pkgpush.Register("test_channel", mPusher)
testReq := TestPushRequest{
Config: pkgpush.Config{
Channel: "test_channel",
URL: "http://test-url",
},
Target: "test_target",
}
bodyBytes, _ := json.Marshal(testReq)
req, _ := http.NewRequest("POST", "/api/v1/admin/push/test", bytes.NewBuffer(bodyBytes))
req.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
r.ServeHTTP(w, req)
assert.Equal(t, http.StatusOK, w.Code)
})
t.Run("list built-in events", func(t *testing.T) {
req, _ := http.NewRequest("GET", "/api/v1/admin/push/events/builtin", nil)
w := httptest.NewRecorder()
r.ServeHTTP(w, req)
assert.Equal(t, http.StatusOK, w.Code)
var resp util.ResponseAny
err := json.Unmarshal(w.Body.Bytes(), &resp)
require.NoError(t, err)
builtins, ok := resp.Data.([]any)
assert.True(t, ok)
assert.NotEmpty(t, builtins)
})
t.Run("create and delete push event", func(t *testing.T) {
// Clean up any existing admin_login event first
dbConn.Where("event_key = ?", "admin_login").Delete(&model.PushEvent{})
// 1. Create event
createReq := CreateEventRequest{
EventKey: "admin_login",
Channels: []string{"email"},
Targets: []string{"admin@test.com"},
Enabled: true,
}
bodyBytes, _ := json.Marshal(createReq)
req, _ := http.NewRequest("POST", "/api/v1/admin/push/events", bytes.NewBuffer(bodyBytes))
req.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
r.ServeHTTP(w, req)
assert.Equal(t, http.StatusOK, w.Code)
// Verify created in DB
var event model.PushEvent
err := dbConn.Where("event_key = ?", "admin_login").First(&event).Error
require.NoError(t, err)
assert.Equal(t, "admin_login", event.EventKey)
assert.Equal(t, "管理员登录", event.Name)
assert.True(t, event.Enabled)
// 2. Try creating again (should fail)
w2 := httptest.NewRecorder()
req2, _ := http.NewRequest("POST", "/api/v1/admin/push/events", bytes.NewBuffer(bodyBytes))
req2.Header.Set("Content-Type", "application/json")
r.ServeHTTP(w2, req2)
assert.Equal(t, http.StatusBadRequest, w2.Code)
// 3. Delete event
w3 := httptest.NewRecorder()
req3, _ := http.NewRequest("DELETE", "/api/v1/admin/push/events/"+strconv.FormatUint(event.ID, 10), nil)
r.ServeHTTP(w3, req3)
assert.Equal(t, http.StatusOK, w3.Code)
// Verify deleted from DB
var count int64
dbConn.Model(&model.PushEvent{}).Where("event_key = ?", "admin_login").Count(&count)
assert.Equal(t, int64(0), count)
})
}
func TestResolveTarget(t *testing.T) {
dbConn, _, cleanup := setupPushTest(t)
defer cleanup()
// 1. 创建测试用户与管理员用户
testUser := &model.User{
ID: 9999,
Username: "target_user",
Email: "target@test.com",
IsAdmin: false,
}
err := dbConn.Create(testUser).Error
require.NoError(t, err)
adminUser := &model.User{
ID: 8888,
Username: "admin_user",
Email: "admin@test.com",
IsAdmin: true,
}
err = dbConn.Create(adminUser).Error
require.NoError(t, err)
flatBody := map[string]any{
"user.id": float64(9999), // JSON 反序列化后一般是 float64
"user.username": "target_user",
"user.email": "target@test.com",
}
ctx := context.Background()
t.Run("dynamic user.id resolved and converted for email channel", func(t *testing.T) {
res := resolveTarget(ctx, "user.id", flatBody, "email")
assert.Equal(t, "target@test.com", res)
})
t.Run("dynamic user.username resolved and converted for email channel", func(t *testing.T) {
res := resolveTarget(ctx, "user.username", flatBody, "email")
assert.Equal(t, "target@test.com", res)
})
t.Run("dynamic user.email resolved directly for email channel", func(t *testing.T) {
res := resolveTarget(ctx, "user.email", flatBody, "email")
assert.Equal(t, "target@test.com", res)
})
t.Run("fixed user id resolved and converted for email channel", func(t *testing.T) {
res := resolveTarget(ctx, "9999", flatBody, "email")
assert.Equal(t, "target@test.com", res)
})
t.Run("fixed username resolved and converted for email channel", func(t *testing.T) {
res := resolveTarget(ctx, "target_user", flatBody, "email")
assert.Equal(t, "target@test.com", res)
})
t.Run("fixed email address resolved directly for email channel", func(t *testing.T) {
res := resolveTarget(ctx, "fixed@example.com", flatBody, "email")
assert.Equal(t, "fixed@example.com", res)
})
t.Run("fixed username resolved for non-email channel", func(t *testing.T) {
res := resolveTarget(ctx, "target_user", flatBody, "lark")
assert.Equal(t, "target_user", res)
})
t.Run("non-exist user resolved as fallback", func(t *testing.T) {
res := resolveTarget(ctx, "non_exist_user", flatBody, "email")
assert.Equal(t, "non_exist_user", res)
})
t.Run("system target resolves to admin email for email channel", func(t *testing.T) {
res := resolveTarget(ctx, "系统", flatBody, "email")
assert.Equal(t, "admin@test.com", res)
res2 := resolveTarget(ctx, "system", flatBody, "email")
assert.Equal(t, "admin@test.com", res2)
res3 := resolveTarget(ctx, "0", flatBody, "email")
assert.Equal(t, "admin@test.com", res3)
})
t.Run("system target resolves to admin username for lark channel", func(t *testing.T) {
res := resolveTarget(ctx, "系统", flatBody, "lark")
assert.Equal(t, "admin_user", res)
})
}
func TestPushChannelAPI(t *testing.T) {
// 1. 模型校验测试
t.Run("validate push channel model constraints", func(t *testing.T) {
// 校验名称合法性
c1 := &model.PushChannel{Name: "invalid-name!", URL: "https://hook.com", Other: "{}"}
assert.Error(t, c1.Validate())
// 校验 URL 安全前缀 HTTPS
c2 := &model.PushChannel{Name: "custom_channel", URL: "http://insecure-hook.com", Other: "{}"}
assert.Error(t, c2.Validate())
// 校验 JSON 格式
c3 := &model.PushChannel{Name: "custom_channel", URL: "https://hook.com", Other: "{invalid-json}"}
assert.Error(t, c3.Validate())
// 正确配置
c4 := &model.PushChannel{Name: "custom_channel", URL: "https://hook.com", Other: "{\"content\":\"$content\"}"}
assert.NoError(t, c4.Validate())
// 飞书渠道校验:非 HTTPS 地址报错
c5 := &model.PushChannel{Name: "lark_channel", Type: "lark", URL: "http://open.feishu.cn", Other: ""}
assert.Error(t, c5.Validate())
// 飞书正确配置
c6 := &model.PushChannel{Name: "lark_channel", Type: "lark", URL: "https://open.feishu.cn", Other: ""}
assert.NoError(t, c6.Validate())
// 邮件配置校验:允许空配置以复用系统全局设置
c7 := &model.PushChannel{Name: "email_channel", Type: "email", URL: "", Token: "", Other: ""}
assert.NoError(t, c7.Validate())
// 邮件正确配置 (非 HTTPS 协议 URL 允许)
c8 := &model.PushChannel{Name: "email_channel", Type: "email", URL: "smtp.exmail.qq.com:465", Token: "user@example.com", Other: "authcode"}
assert.NoError(t, c8.Validate())
})
// 2. HTTP CRUD & 触发鉴权测试
dbConn, _, cleanup := setupPushTest(t)
defer cleanup()
// 构建路由以进行 HTTP 模拟请求
r := gin.New()
adminGroup := r.Group("/api/v1/admin")
{
adminGroup.GET("/push/channels", ListChannels)
adminGroup.POST("/push/channels", CreateChannel)
adminGroup.PUT("/push/channels/:id", UpdateChannel)
adminGroup.DELETE("/push/channels/:id", DeleteChannel)
adminGroup.POST("/push/channels/test", TestChannel)
}
var createdID uint64
t.Run("admin create channel", func(t *testing.T) {
reqBody := CreateChannelRequest{
Name: "my_custom_channel",
Description: "My custom channel webhook",
Type: "custom",
Token: "my_chan_token",
URL: "https://webhook.site/test",
Other: `{"title": "$title", "body": "$content"}`,
Enabled: true,
}
bodyBytes, _ := json.Marshal(reqBody)
req, _ := http.NewRequest("POST", "/api/v1/admin/push/channels", bytes.NewBuffer(bodyBytes))
req.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
r.ServeHTTP(w, req)
assert.Equal(t, http.StatusOK, w.Code)
var resp util.ResponseAny
json.Unmarshal(w.Body.Bytes(), &resp)
dataMap, ok := resp.Data.(map[string]any)
assert.True(t, ok)
assert.Equal(t, "my_custom_channel", dataMap["name"])
createdID = uint64(dataMap["id"].(float64))
})
t.Run("admin list channels", func(t *testing.T) {
req, _ := http.NewRequest("GET", "/api/v1/admin/push/channels", nil)
w := httptest.NewRecorder()
r.ServeHTTP(w, req)
assert.Equal(t, http.StatusOK, w.Code)
var resp util.ResponseAny
json.Unmarshal(w.Body.Bytes(), &resp)
list, ok := resp.Data.([]any)
assert.True(t, ok)
assert.Len(t, list, 1)
})
t.Run("admin update channel", func(t *testing.T) {
updateReq := UpdateChannelRequest{
Description: "Updated remark",
Type: "custom",
Token: "new_chan_token",
URL: "https://webhook.site/updated",
Other: `{"text": "$content"}`,
Enabled: true,
}
bodyBytes, _ := json.Marshal(updateReq)
req, _ := http.NewRequest("PUT", "/api/v1/admin/push/channels/"+strconv.FormatUint(createdID, 10), bytes.NewBuffer(bodyBytes))
req.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
r.ServeHTTP(w, req)
assert.Equal(t, http.StatusOK, w.Code)
var updated model.PushChannel
dbConn.First(&updated, createdID)
assert.Equal(t, "Updated remark", updated.Description)
assert.Equal(t, "new_chan_token", updated.Token)
assert.Equal(t, `{"text": "$content"}`, updated.Other)
})
t.Run("admin test channel endpoint", func(t *testing.T) {
testReq := TestChannelRequest{
Name: "my_custom_channel",
Target: "test_target",
}
bodyBytes, _ := json.Marshal(testReq)
req, _ := http.NewRequest("POST", "/api/v1/admin/push/channels/test", bytes.NewBuffer(bodyBytes))
req.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
r.ServeHTTP(w, req)
assert.Equal(t, http.StatusOK, w.Code)
})
t.Run("admin delete channel", func(t *testing.T) {
req, _ := http.NewRequest("DELETE", "/api/v1/admin/push/channels/"+strconv.FormatUint(createdID, 10), nil)
w := httptest.NewRecorder()
r.ServeHTTP(w, req)
assert.Equal(t, http.StatusOK, w.Code)
var count int64
dbConn.Model(&model.PushChannel{}).Where("id = ?", createdID).Count(&count)
assert.Equal(t, int64(0), count)
})
}
+450
View File
@@ -0,0 +1,450 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
// Package push defines push notification HTTP routes.
package push
import (
"context"
"encoding/json"
"errors"
"fmt"
"net/http"
"strconv"
"strings"
"github.com/Rain-kl/Wavelet/internal/db"
"github.com/Rain-kl/Wavelet/internal/model"
"github.com/Rain-kl/Wavelet/internal/util"
"github.com/Rain-kl/Wavelet/pkg/push"
"github.com/gin-gonic/gin"
"gorm.io/gorm"
)
// UpdateEventRequest 更新事件请求参数
type UpdateEventRequest struct {
Channels []string `json:"channels"`
Targets []string `json:"targets"`
Template string `json:"template" binding:"required"`
Enabled bool `json:"enabled"`
}
// TestPushRequest 测试推送通道请求参数
type TestPushRequest struct {
Config push.Config `json:"config" binding:"required"`
Target string `json:"target"`
}
// SyncEvents automatically registers/updates built-in events in the database.
func SyncEvents(ctx context.Context) error {
for _, meta := range BuiltInEvents {
var event model.PushEvent
err := db.DB(ctx).Where("event_key = ?", meta.Key).First(&event).Error
if errors.Is(err, gorm.ErrRecordNotFound) {
var defaultTemplateStr string
if defaultTemplateBytes, err := json.Marshal(meta.DefaultTemplate); err == nil {
defaultTemplateStr = string(defaultTemplateBytes)
}
event = model.PushEvent{
EventKey: meta.Key,
Name: meta.Name,
Channels: []string{},
Targets: []string{},
Template: defaultTemplateStr,
Enabled: false,
}
if err := db.DB(ctx).Create(&event).Error; err != nil {
return err
}
} else if err != nil {
return err
}
}
return nil
}
// ListEvents 获取通知事件列表
// @Summary 获取所有通知事件
// @Description 返回系统配置的通知事件列表,包括预置和自定义事件,需要管理员权限
// @Tags admin-push
// @Produce json
// @Security SessionCookie
// @Success 200 {object} util.ResponseAny{data=[]model.PushEvent} "通知事件列表"
// @Router /api/v1/admin/push/events [get]
func ListEvents(c *gin.Context) {
ctx := c.Request.Context()
var events []model.PushEvent
if err := db.DB(ctx).Order("created_at DESC").Find(&events).Error; err != nil {
c.JSON(http.StatusInternalServerError, util.Err(err.Error()))
return
}
c.JSON(http.StatusOK, util.OK(events))
}
// CreateEventRequest 创建事件请求参数
type CreateEventRequest struct {
EventKey string `json:"event_key" binding:"required"`
Channels []string `json:"channels"`
Targets []string `json:"targets"`
Template string `json:"template"`
Enabled bool `json:"enabled"`
}
func findBuiltInEvent(key string) (EventMetadata, bool) {
for _, meta := range BuiltInEvents {
if meta.Key == key {
return meta, true
}
}
return EventMetadata{}, false
}
// ListBuiltInEvents 获取内置通知事件列表
// @Summary 获取所有内置通知事件
// @Description 返回系统定义的所有内置通知事件元数据,供前端下拉框选择,需要管理员权限
// @Tags admin-push
// @Produce json
// @Security SessionCookie
// @Success 200 {object} util.ResponseAny{data=[]EventMetadata} "内置通知事件列表"
// @Router /api/v1/admin/push/events/builtin [get]
func ListBuiltInEvents(c *gin.Context) {
c.JSON(http.StatusOK, util.OK(BuiltInEvents))
}
// CreateEvent 创建通知事件
// @Summary 创建通知事件
// @Description 绑定系统内置事件、推送渠道、接收目标并创建通知事件配置,需要管理员权限
// @Tags admin-push
// @Accept json
// @Produce json
// @Security SessionCookie
// @Param request body CreateEventRequest true "创建参数"
// @Success 200 {object} util.ResponseAny{data=model.PushEvent} "创建成功"
// @Router /api/v1/admin/push/events [post]
func CreateEvent(c *gin.Context) {
var req CreateEventRequest
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, util.Err(err.Error()))
return
}
ctx := c.Request.Context()
// 1. 检查内置事件是否存在
meta, found := findBuiltInEvent(req.EventKey)
if !found {
c.JSON(http.StatusBadRequest, util.Err("unsupported built-in event key"))
return
}
// 2. 检查是否已经创建过该事件的配置
var count int64
if err := db.DB(ctx).Model(&model.PushEvent{}).Where("event_key = ?", req.EventKey).Count(&count).Error; err != nil {
c.JSON(http.StatusInternalServerError, util.Err(err.Error()))
return
}
if count > 0 {
c.JSON(http.StatusBadRequest, util.Err("this notification event is already configured"))
return
}
// 3. 模板处理
templateStr := strings.TrimSpace(req.Template)
if templateStr == "" {
defaultTemplateBytes, err := json.Marshal(meta.DefaultTemplate)
if err != nil {
c.JSON(http.StatusInternalServerError, util.Err(err.Error()))
return
}
templateStr = string(defaultTemplateBytes)
} else {
var tempMap map[string]any
if err := json.Unmarshal([]byte(templateStr), &tempMap); err != nil {
c.JSON(http.StatusBadRequest, util.Err("custom template is not a valid JSON format"))
return
}
}
// 4. 创建事件记录
channels := req.Channels
if channels == nil {
channels = []string{}
}
targets := req.Targets
if targets == nil {
targets = []string{}
}
event := model.PushEvent{
EventKey: meta.Key,
Name: meta.Name,
Channels: channels,
Targets: targets,
Template: templateStr,
Enabled: req.Enabled,
}
if err := event.Validate(); err != nil {
c.JSON(http.StatusBadRequest, util.Err(err.Error()))
return
}
if err := db.DB(ctx).Create(&event).Error; err != nil {
c.JSON(http.StatusInternalServerError, util.Err(err.Error()))
return
}
c.JSON(http.StatusOK, util.OK(event))
}
// DeleteEvent 删除通知事件配置
// @Summary 删除通知事件配置
// @Description 删除数据库中的特定通知事件配置,需要管理员权限
// @Tags admin-push
// @Produce json
// @Security SessionCookie
// @Param id path int true "事件 ID"
// @Success 200 {object} util.ResponseAny{data=string} "删除成功"
// @Router /api/v1/admin/push/events/{id} [delete]
func DeleteEvent(c *gin.Context) {
idStr := c.Param("id")
id, err := strconv.ParseUint(idStr, 10, 64)
if err != nil {
c.JSON(http.StatusBadRequest, util.Err("invalid event id"))
return
}
ctx := c.Request.Context()
var event model.PushEvent
if err := db.DB(ctx).First(&event, id).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
c.JSON(http.StatusNotFound, util.Err("notification event not found"))
} else {
c.JSON(http.StatusInternalServerError, util.Err(err.Error()))
}
return
}
if err := db.DB(ctx).Delete(&event).Error; err != nil {
c.JSON(http.StatusInternalServerError, util.Err(err.Error()))
return
}
c.JSON(http.StatusOK, util.OKNil())
}
// UpdateEvent 更新通知事件
// @Summary 更新通知事件
// @Description 更新已有通知事件的推送渠道、接收目标和内容模板,需要管理员权限
// @Tags admin-push
// @Accept json
// @Produce json
// @Security SessionCookie
// @Param id path int true "事件 ID"
// @Param request body push.UpdateEventRequest true "更新参数"
// @Success 200 {object} util.ResponseAny{data=string} "修改成功"
// @Router /api/v1/admin/push/events/{id} [put]
func UpdateEvent(c *gin.Context) {
idStr := c.Param("id")
id, err := strconv.ParseUint(idStr, 10, 64)
if err != nil {
c.JSON(http.StatusBadRequest, util.Err("invalid event id"))
return
}
var req UpdateEventRequest
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, util.Err(err.Error()))
return
}
var event model.PushEvent
if err := db.DB(c.Request.Context()).First(&event, id).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
c.JSON(http.StatusNotFound, util.Err("notification event not found"))
} else {
c.JSON(http.StatusInternalServerError, util.Err(err.Error()))
}
return
}
event.Channels = req.Channels
event.Targets = req.Targets
event.Template = req.Template
event.Enabled = req.Enabled
if err := event.Validate(); err != nil {
c.JSON(http.StatusBadRequest, util.Err(err.Error()))
return
}
if err := db.DB(c.Request.Context()).Save(&event).Error; err != nil {
c.JSON(http.StatusInternalServerError, util.Err(err.Error()))
return
}
c.JSON(http.StatusOK, util.OKNil())
}
// ToggleEvent 快捷切换通知事件启用状态
// @Summary 快捷切换通知事件启用状态
// @Description 启用或禁用指定的通知事件
// @Tags admin-push
// @Produce json
// @Security SessionCookie
// @Param id path int true "事件 ID"
// @Success 200 {object} util.ResponseAny{data=string} "切换成功"
// @Router /api/v1/admin/push/events/{id}/toggle [post]
func ToggleEvent(c *gin.Context) {
idStr := c.Param("id")
id, err := strconv.ParseUint(idStr, 10, 64)
if err != nil {
c.JSON(http.StatusBadRequest, util.Err("invalid event id"))
return
}
var event model.PushEvent
if err := db.DB(c.Request.Context()).First(&event, id).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
c.JSON(http.StatusNotFound, util.Err("notification event not found"))
} else {
c.JSON(http.StatusInternalServerError, util.Err(err.Error()))
}
return
}
event.Enabled = !event.Enabled
if err := db.DB(c.Request.Context()).Model(&event).Update("enabled", event.Enabled).Error; err != nil {
c.JSON(http.StatusInternalServerError, util.Err(err.Error()))
return
}
c.JSON(http.StatusOK, util.OK(event.Enabled))
}
// pushHistoriesResponse 推送历史分页响应
//
//nolint:unused
type pushHistoriesResponse struct {
Total int64 `json:"total"`
Results []model.PushHistory `json:"results"`
}
// ListHistories 分页获取通知推送历史
// @Summary 分页获取通知推送历史
// @Description 返回分页的通知历史日志数据,需要管理员权限
// @Tags admin-push
// @Produce json
// @Security SessionCookie
// @Param page query int false "当前页码"
// @Param page_size query int false "分页大小"
// @Param event_key query string false "过滤事件名称"
// @Param status query string false "过滤发送状态"
// @Success 200 {object} util.ResponseAny{data=pushHistoriesResponse} "推送历史列表"
// @Router /api/v1/admin/push/histories [get]
func ListHistories(c *gin.Context) {
pageStr := c.DefaultQuery("page", "1")
pageSizeStr := c.DefaultQuery("page_size", "20")
eventKey := c.Query("event_key")
status := c.Query("status")
page, err := strconv.Atoi(pageStr)
if err != nil || page < 1 {
page = 1
}
pageSize, err := strconv.Atoi(pageSizeStr)
if err != nil || pageSize < 1 {
pageSize = 20
}
query := db.DB(c.Request.Context()).Model(&model.PushHistory{}).Order("created_at DESC")
if eventKey != "" {
query = query.Where("event_key = ?", eventKey)
}
if status != "" {
query = query.Where("status = ?", status)
}
var total int64
if err := query.Count(&total).Error; err != nil {
c.JSON(http.StatusInternalServerError, util.Err(err.Error()))
return
}
var results []model.PushHistory
offset := (page - 1) * pageSize
if err := query.Offset(offset).Limit(pageSize).Find(&results).Error; err != nil {
c.JSON(http.StatusInternalServerError, util.Err(err.Error()))
return
}
c.JSON(http.StatusOK, util.OK(map[string]any{
"total": total,
"results": results,
}))
}
// TestPush 测试推送通道发送
// @Summary 测试推送通道发送
// @Description 接收临时通知渠道配置并在本地同步调用 Pusher.Send 发送测试消息
// @Tags admin-push
// @Accept json
// @Produce json
// @Security SessionCookie
// @Param request body push.TestPushRequest true "测试请求体"
// @Success 200 {object} util.ResponseAny{data=string} "测试成功"
// @Router /api/v1/admin/push/test [post]
func TestPush(c *gin.Context) {
var req TestPushRequest
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, util.Err(err.Error()))
return
}
pusher, err := push.GetPusher(req.Config.Channel)
if err != nil {
c.JSON(http.StatusBadRequest, util.Err(err.Error()))
return
}
// 校验配置
if err := pusher.ValidateConfig(req.Config); err != nil {
c.JSON(http.StatusBadRequest, util.Err(fmt.Sprintf("validation failed: %v", err)))
return
}
// 邮件渠道需要从系统设置中拉取发件人 SMTP 信息做测试 (除非配了独立的)
if req.Config.Channel == channelEmail && (req.Config.URL == "" || req.Config.Key == "") {
var smtpHost, smtpPort, smtpUser, smtpPass model.SystemConfig
ctx := c.Request.Context()
_ = smtpHost.GetByKey(ctx, model.ConfigKeySMTPHost)
_ = smtpPort.GetByKey(ctx, model.ConfigKeySMTPPort)
_ = smtpUser.GetByKey(ctx, model.ConfigKeySMTPUsername)
_ = smtpPass.GetByKey(ctx, model.ConfigKeySMTPPassword)
if smtpHost.Value != "" && smtpUser.Value != "" {
port := smtpPort.Value
if port == "" {
port = "587"
}
req.Config.URL = smtpHost.Value + ":" + port
req.Config.Key = smtpUser.Value
req.Config.Secret = smtpPass.Value
}
}
testBody := map[string]any{
keyTitle: "测试通道推送",
keyContent: "当您收到这条消息,说明当前渠道连通性测试通过。",
keyLevel: defaultLevelInfo,
}
err = pusher.Send(c.Request.Context(), req.Config, req.Target, testBody, "", nil)
if err != nil {
c.JSON(http.StatusOK, util.Err(err.Error()))
return
}
c.JSON(http.StatusOK, util.OKNil())
}
+163
View File
@@ -0,0 +1,163 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
// Package push defines push notification HTTP routes and background tasks.
package push
import (
"context"
"encoding/json"
"errors"
"fmt"
"time"
"github.com/Rain-kl/Wavelet/internal/db"
"github.com/Rain-kl/Wavelet/internal/model"
"github.com/Rain-kl/Wavelet/internal/task"
"github.com/Rain-kl/Wavelet/pkg/push"
)
const (
// SendNotificationTask 发送推送通知任务标识
SendNotificationTask = "push:send"
// TaskTypeSendNotification 推送通知管理类型
TaskTypeSendNotification = "send_notification"
)
// SendNotificationMeta represents the task metadata.
var SendNotificationMeta = task.TaskMeta{
Type: TaskTypeSendNotification,
AsynqTask: SendNotificationTask,
Name: "推送通知",
Description: "异步执行系统通知的多渠道派发与推送",
SupportsTime: false,
MaxRetry: task.DefaultMaxRetry,
Queue: task.QueueDefault,
Retryable: true,
Params: []task.TaskParam{
{
Name: "event_key",
Label: "事件标识",
Type: "string",
Required: true,
Placeholder: "admin_login",
},
{
Name: "target",
Label: "目标接收者",
Type: "string",
Required: false,
},
},
}
// PushHandler 通知推送异步任务处理器
//
//nolint:revive
type PushHandler struct{}
// ValidatePayload 校验并标准化推送参数
func (h *PushHandler) ValidatePayload(payload []byte) ([]byte, error) {
if len(payload) == 0 {
return nil, errors.New("payload is required")
}
var req SendPayload
if err := json.Unmarshal(payload, &req); err != nil {
return nil, fmt.Errorf("invalid json format: %w", err)
}
if req.Config.Channel == "" {
return nil, errors.New("channel type is required")
}
return json.Marshal(req)
}
// Execute 异步执行推送操作并记录推送历史审计
func (h *PushHandler) Execute(ctx context.Context, payload []byte) (*task.TaskResult, error) {
var req SendPayload
if err := json.Unmarshal(payload, &req); err != nil {
task.AppendLog(ctx, "解析推送参数失败: %v", err)
return nil, fmt.Errorf("parse payload failed: %w", err)
}
task.AppendLog(ctx, "开始推送通知: 事件 = %s, 渠道 = %s, 接收目标 = %s", req.EventKey, req.Config.Channel, req.Target)
pusher, err := push.GetPusher(req.Config.Channel)
if err != nil {
errWrap := fmt.Errorf("get pusher failed: %w", err)
task.AppendLog(ctx, "推送失败: %v", errWrap)
if task.IsFinalAttempt(ctx) {
h.recordHistory(ctx, req, "failed", errWrap.Error())
}
return nil, errWrap
}
// 执行真正的消息推送,扁平化为原始 json 格式
flatBody := req.Body.Flatten()
err = pusher.Send(ctx, req.Config, req.Target, flatBody, req.Template, nil)
title := req.Body.Title
content := req.Body.Content
if err != nil {
task.AppendLog(ctx, "消息推送失败 (标题: %s): %v", title, err)
if task.IsFinalAttempt(ctx) {
h.recordHistory(ctx, req, "failed", err.Error())
}
return nil, fmt.Errorf("pusher.Send failed: %w", err)
}
task.AppendLog(ctx, "消息推送成功 (标题: %s, 内容摘要: %s)", title, content)
h.recordHistory(ctx, req, "success", "")
return &task.TaskResult{
Message: fmt.Sprintf("推送成功: [%s] -> %s", req.Config.Channel, req.Target),
}, nil
}
func (h *PushHandler) recordHistory(ctx context.Context, req SendPayload, status string, errMsg string) {
title := req.Body.Title
content := req.Body.Content
level := req.Body.Level
if title == "" {
title = "系统通知"
}
if level == "" {
level = defaultLevelInfo
}
target := req.Target
if target == "" {
// 如果目标人为空 (例如 webhook bot),用其地址填充前缀或默认词作为归档
if req.Config.URL != "" {
target = req.Config.URL
// 隐藏敏感 URL 细节
//nolint:mnd
if len(target) > 50 {
target = target[:47] + "..."
}
} else {
target = "default"
}
}
history := model.PushHistory{
EventKey: req.EventKey,
Channel: req.Config.Channel,
Target: target,
Title: title,
Content: content,
Level: level,
Status: status,
ErrorMsg: errMsg,
CreatedAt: time.Now(),
}
// 记录到数据库
if dbErr := db.DB(ctx).Create(&history).Error; dbErr != nil {
task.AppendLog(ctx, "写入推送历史审计记录失败: %v", dbErr)
}
}
+19 -18
View File
@@ -18,6 +18,7 @@ import (
"github.com/Rain-kl/Wavelet/internal/apps/upload"
"github.com/Rain-kl/Wavelet/internal/apps/user"
"github.com/Rain-kl/Wavelet/internal/model"
"github.com/Rain-kl/Wavelet/internal/service"
"github.com/Rain-kl/Wavelet/internal/task"
"github.com/Rain-kl/Wavelet/internal/testhelper"
"github.com/Rain-kl/Wavelet/internal/util"
@@ -91,7 +92,7 @@ func TestListTaskTypes(t *testing.T) {
foundCleanup := false
foundWarmImageCache := false
for _, m := range taskMetas {
if m.Type == upload.TaskTypeCleanupUploads {
if m.Type == service.TaskTypeSystemCleanup {
foundCleanup = true
}
if m.Type == upload.TaskTypeWarmImageCache {
@@ -99,7 +100,7 @@ func TestListTaskTypes(t *testing.T) {
}
}
if !foundCleanup {
t.Errorf("expected task type %s to be listed", upload.TaskTypeCleanupUploads)
t.Errorf("expected task type %s to be listed", service.TaskTypeSystemCleanup)
}
if !foundWarmImageCache {
t.Errorf("expected task type %s to be listed", upload.TaskTypeWarmImageCache)
@@ -115,7 +116,7 @@ func TestDispatchTask(t *testing.T) {
t.Run("dispatch valid task successfully", func(t *testing.T) {
payload := DispatchTaskRequest{
TaskType: upload.TaskTypeCleanupUploads,
TaskType: service.TaskTypeSystemCleanup,
}
body, _ := json.Marshal(payload)
req, _ := http.NewRequest("POST", "/api/v1/admin/tasks/dispatch", bytes.NewBuffer(body))
@@ -227,9 +228,9 @@ func TestListTaskExecutions(t *testing.T) {
// 准备测试数据
now := time.Now()
records := []*model.TaskExecution{
{TaskID: "exec_001", TaskType: "upload:cleanup_unused", TaskName: "清理上传", Status: model.TaskExecutionStatusSucceeded, TriggeredBy: "manual", Duration: 1500, Result: "清理完成", StartedAt: &now, FinishedAt: &now},
{TaskID: "exec_002", TaskType: "upload:cleanup_unused", TaskName: "清理上传", Status: model.TaskExecutionStatusFailed, TriggeredBy: "system", ErrorMessage: "连接超时", StartedAt: &now, FinishedAt: &now},
{TaskID: "exec_003", TaskType: "upload:cleanup_unused", TaskName: "清理上传", Status: model.TaskExecutionStatusPending, TriggeredBy: "manual", Retryable: true, MaxRetry: 3},
{TaskID: "exec_001", TaskType: "system:cleanup", TaskName: "系统垃圾清理", Status: model.TaskExecutionStatusSucceeded, TriggeredBy: "manual", Duration: 1500, Result: "清理完成", StartedAt: &now, FinishedAt: &now},
{TaskID: "exec_002", TaskType: "system:cleanup", TaskName: "系统垃圾清理", Status: model.TaskExecutionStatusFailed, TriggeredBy: "system", ErrorMessage: "连接超时", StartedAt: &now, FinishedAt: &now},
{TaskID: "exec_003", TaskType: "system:cleanup", TaskName: "系统垃圾清理", Status: model.TaskExecutionStatusPending, TriggeredBy: "manual", Retryable: true, MaxRetry: 3},
}
for _, r := range records {
err := model.CreateTaskExecution(ctx, r)
@@ -271,7 +272,7 @@ func TestListTaskExecutions(t *testing.T) {
})
t.Run("filter by task_type (asynq task name)", func(t *testing.T) {
req, _ := http.NewRequest("GET", "/api/v1/admin/tasks/executions?task_type=upload:cleanup_unused", nil)
req, _ := http.NewRequest("GET", "/api/v1/admin/tasks/executions?task_type=system:cleanup", nil)
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
@@ -288,7 +289,7 @@ func TestListTaskExecutions(t *testing.T) {
})
t.Run("filter by task_type (management task type)", func(t *testing.T) {
req, _ := http.NewRequest("GET", "/api/v1/admin/tasks/executions?task_type=cleanup_unused_uploads", nil)
req, _ := http.NewRequest("GET", "/api/v1/admin/tasks/executions?task_type=system_cleanup", nil)
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
@@ -333,8 +334,8 @@ func TestGetTaskExecution(t *testing.T) {
// 创建测试记录
execution := &model.TaskExecution{
TaskID: "detail_001",
TaskType: "upload:cleanup_unused",
TaskName: "清理上传",
TaskType: "system:cleanup",
TaskName: "系统垃圾清理",
Status: model.TaskExecutionStatusSucceeded,
Log: "[10:00:01] 开始扫描\n[10:00:02] 找到 50 个文件\n[10:00:03] 清理完成",
Result: "共清理 50 个文件",
@@ -397,8 +398,8 @@ func TestRetryTask(t *testing.T) {
now := time.Now()
execution := &model.TaskExecution{
TaskID: "retry_api_001",
TaskType: "upload:cleanup_unused",
TaskName: "清理上传",
TaskType: "system:cleanup",
TaskName: "系统垃圾清理",
Status: model.TaskExecutionStatusFailed,
ErrorMessage: "S3 连接超时",
Retryable: true,
@@ -437,8 +438,8 @@ func TestRetryTask(t *testing.T) {
t.Run("retry succeeded task fails", func(t *testing.T) {
execution := &model.TaskExecution{
TaskID: "retry_succeeded_001",
TaskType: "upload:cleanup_unused",
TaskName: "清理上传",
TaskType: "system:cleanup",
TaskName: "系统垃圾清理",
Status: model.TaskExecutionStatusSucceeded,
Retryable: true,
MaxRetry: 3,
@@ -458,8 +459,8 @@ func TestRetryTask(t *testing.T) {
t.Run("retry non-retryable task fails", func(t *testing.T) {
execution := &model.TaskExecution{
TaskID: "retry_not_allowed_001",
TaskType: "upload:cleanup_unused",
TaskName: "清理上传",
TaskType: "system:cleanup",
TaskName: "系统垃圾清理",
Status: model.TaskExecutionStatusFailed,
Retryable: false,
TriggeredBy: "manual",
@@ -502,8 +503,8 @@ func TestRetryTaskMaxRetryExceeded(t *testing.T) {
execution := &model.TaskExecution{
TaskID: "retry_max_api_001",
TaskType: "upload:cleanup_unused",
TaskName: "清理上传",
TaskType: "system:cleanup",
TaskName: "系统垃圾清理",
Status: model.TaskExecutionStatusFailed,
Retryable: true,
MaxRetry: 1,
+27 -9
View File
@@ -5,6 +5,7 @@
package oauth
import (
"context"
"errors"
"net/http"
@@ -28,6 +29,19 @@ type loginRequiredAuditLog struct {
Referer string `json:"referer"`
}
func getUserByToken(ctx context.Context, tokenStr string) (*model.User, *model.AccessToken, error) {
tokenHash := model.HashToken(tokenStr)
var tokenRecord model.AccessToken
if err := db.DB(ctx).Where("token_hash = ?", tokenHash).First(&tokenRecord).Error; err != nil {
return nil, nil, err
}
var user model.User
if err := db.DB(ctx).Where("id = ? AND is_active = ?", tokenRecord.UserID, true).First(&user).Error; err != nil {
return nil, nil, err
}
return &user, &tokenRecord, nil
}
// GetUserFromRequest 校验 Access Token 或 Session 并返回用户对象,如果未登录或用户失效则返回 error
func GetUserFromRequest(c *gin.Context) (*model.User, error) {
ctx := c.Request.Context()
@@ -41,18 +55,16 @@ func GetUserFromRequest(c *gin.Context) (*model.User, error) {
}
}
var user model.User
// 优先使用 Access Token 鉴权
if tokenStr != "" {
tokenHash := model.HashToken(tokenStr)
var tokenRecord model.AccessToken
if err := db.DB(ctx).Where("token_hash = ?", tokenHash).First(&tokenRecord).Error; err == nil {
if err := db.DB(ctx).Where("id = ? AND is_active = ?", tokenRecord.UserID, true).First(&user).Error; err == nil {
util.SetToContext(c, TokenAuthKey, true)
util.SetToContext(c, TokenAdminKey, tokenRecord.IsAdmin)
return &user, nil
if user, tokenRecord, err := getUserByToken(ctx, tokenStr); err == nil {
// 强行阻止 system 用户任何会话/Token 鉴权通过
if user.Username == "system" {
return nil, errors.New("system user is not allowed to login")
}
util.SetToContext(c, TokenAuthKey, true)
util.SetToContext(c, TokenAdminKey, tokenRecord.IsAdmin)
return user, nil
}
}
@@ -62,6 +74,7 @@ func GetUserFromRequest(c *gin.Context) (*model.User, error) {
return nil, errors.New("unauthorized")
}
var user model.User
// load user from db to make sure is active
tx := db.DB(ctx).Where("id = ? AND is_active = ?", userID, true).First(&user)
if tx.Error != nil {
@@ -81,6 +94,11 @@ func GetUserFromRequest(c *gin.Context) (*model.User, error) {
util.SetToContext(c, TokenAuthKey, false)
util.SetToContext(c, TokenAdminKey, false)
// 强行阻止 system 用户任何会话/Token 鉴权通过
if user.Username == "system" {
return nil, errors.New("system user is not allowed to login")
}
return &user, nil
}
+74
View File
@@ -26,6 +26,8 @@ import (
"github.com/gin-gonic/gin"
"github.com/go-jose/go-jose/v4"
"github.com/redis/go-redis/v9"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"golang.org/x/oauth2"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
@@ -1181,3 +1183,75 @@ func TestOIDCPolicyEnforcement(t *testing.T) {
}
}
func TestSystemUserBlockedByMiddleware(t *testing.T) {
initializeTestConfig()
dbConn := setupTestDB(t)
// 1. 创建正常管理员
adminUser := &model.User{ID: 1001, Username: "normal_admin", IsAdmin: true, IsActive: true}
err := dbConn.Create(adminUser).Error
require.NoError(t, err)
// 2. 创建系统用户 (根据架构设计,系统用户 id = 999)
systemUser := &model.User{ID: 999, Username: "system", Nickname: "系统", Password: "*", IsActive: true}
err = dbConn.Create(systemUser).Error
require.NoError(t, err)
// 3. 设置全局测试数据库连接并构建测试路由组
db.SetDB(dbConn)
rProtected := gin.New()
store := cookie.NewStore([]byte("secret"))
rProtected.Use(sessions.Sessions("mysession", store))
rProtected.Use(LoginRequired())
rProtected.GET("/test-auth", func(c *gin.Context) {
c.JSON(http.StatusOK, gin.H{"status": "ok"})
})
// 4. 测试未登录用户 (401)
w1 := httptest.NewRecorder()
req1, _ := http.NewRequest("GET", "/test-auth", nil)
rProtected.ServeHTTP(w1, req1)
assert.Equal(t, http.StatusUnauthorized, w1.Code)
// 5. 测试正常用户登录并访问 (200)
rLogin := gin.New()
rLogin.Use(sessions.Sessions("mysession", store))
rLogin.GET("/login-mock", func(c *gin.Context) {
session := sessions.Default(c)
session.Set("user_id", uint64(1001))
_ = session.Save()
c.Status(200)
})
wLogin := httptest.NewRecorder()
reqLogin, _ := http.NewRequest("GET", "/login-mock", nil)
rLogin.ServeHTTP(wLogin, reqLogin)
cookieStr := wLogin.Header().Get("Set-Cookie")
w2 := httptest.NewRecorder()
req2, _ := http.NewRequest("GET", "/test-auth", nil)
req2.Header.Set("Cookie", cookieStr)
rProtected.ServeHTTP(w2, req2)
assert.Equal(t, http.StatusOK, w2.Code)
// 6. 测试 system 用户(ID: 999)登录并访问 (被中间件阻断返回 401)
rLoginSystem := gin.New()
rLoginSystem.Use(sessions.Sessions("mysession", store))
rLoginSystem.GET("/login-system-mock", func(c *gin.Context) {
session := sessions.Default(c)
session.Set("user_id", uint64(999))
_ = session.Save()
c.Status(200)
})
wLoginSystem := httptest.NewRecorder()
reqLoginSystem, _ := http.NewRequest("GET", "/login-system-mock", nil)
rLoginSystem.ServeHTTP(wLoginSystem, reqLoginSystem)
cookieSystemStr := wLoginSystem.Header().Get("Set-Cookie")
w3 := httptest.NewRecorder()
req3, _ := http.NewRequest("GET", "/test-auth", nil)
req3.Header.Set("Cookie", cookieSystemStr)
rProtected.ServeHTTP(w3, req3)
assert.Equal(t, http.StatusUnauthorized, w3.Code)
}
+3
View File
@@ -14,6 +14,7 @@ import (
"strings"
"time"
"github.com/Rain-kl/Wavelet/internal/apps/admin/push/custom_events"
"github.com/Rain-kl/Wavelet/internal/common"
"github.com/Rain-kl/Wavelet/internal/config"
"github.com/Rain-kl/Wavelet/internal/db"
@@ -675,6 +676,8 @@ func handleCallbackLogin(ctx context.Context, c *gin.Context, source *model.Auth
logger.InfoF(ctx, "[LoginAudit] successful OAuth login via source: %s, external ID: %s, user: %s, ID: %d, IP: %s", source.Name, userInfo.Sub, user.Username, user.ID, c.ClientIP())
custom_events.TriggerAdminLoginEvent(ctx, &user, c.ClientIP())
c.JSON(http.StatusOK, util.OK(buildCallbackResult(&user, "logged_in")))
}
+3 -3
View File
@@ -115,7 +115,7 @@ func ListFiles(c *gin.Context) {
// @Router /api/v1/admin/uploads/{id} [delete]
func DeleteFile(c *gin.Context) {
ctx := c.Request.Context()
if storageReadOnly(ctx) {
if StorageReadOnly(ctx) {
c.JSON(http.StatusConflict, util.Err(ErrStorageReadOnly))
return
}
@@ -259,7 +259,7 @@ func ListMyFiles(c *gin.Context) {
func DeleteMyFile(c *gin.Context) {
currUser, _ := util.GetFromContext[*model.User](c, oauth.UserObjKey)
ctx := c.Request.Context()
if storageReadOnly(ctx) {
if StorageReadOnly(ctx) {
c.JSON(http.StatusConflict, util.Err(ErrStorageReadOnly))
return
}
@@ -313,7 +313,7 @@ type updateMyFileRequest struct {
func UpdateMyFile(c *gin.Context) {
currUser, _ := util.GetFromContext[*model.User](c, oauth.UserObjKey)
ctx := c.Request.Context()
if storageReadOnly(ctx) {
if StorageReadOnly(ctx) {
c.JSON(http.StatusConflict, util.Err(ErrStorageReadOnly))
return
}
+2 -2
View File
@@ -362,7 +362,7 @@ func tryInstantUpload(ctx context.Context, c *gin.Context, currUser *model.User,
if err != nil {
return false, err
}
if storageReadOnly(ctx) {
if StorageReadOnly(ctx) {
c.JSON(http.StatusConflict, util.Err(ErrStorageReadOnly))
return true, nil
}
@@ -396,7 +396,7 @@ func tryInstantUpload(ctx context.Context, c *gin.Context, currUser *model.User,
// storeUploadFile 将文件写入当前活动存储驱动。
func storeUploadFile(ctx context.Context, subPath string, size int64, mimeType string, buf *bytes.Buffer, meta *model.UploadMetadata) (string, string, string) {
if storageReadOnly(ctx) {
if StorageReadOnly(ctx) {
return "", "", ErrStorageReadOnly
}
driver, backend, err := storage.Active(ctx)
+1 -3
View File
@@ -12,7 +12,7 @@ import (
"github.com/Rain-kl/Wavelet/internal/storage"
)
func storageReadOnly(ctx context.Context) bool {
func StorageReadOnly(ctx context.Context) bool {
execution, ok, err := latestStorageMigrationExecution(ctx)
if err != nil {
logger.ErrorF(ctx, "读取存储维护状态失败: %v", err)
@@ -66,5 +66,3 @@ func currentMigrationTargetConfig(ctx context.Context) (storage.Config, bool, er
}
return target, true, nil
}
+2 -95
View File
@@ -11,21 +11,14 @@ import (
"fmt"
"strings"
"sync"
"time"
"github.com/Rain-kl/Wavelet/internal/db"
"github.com/Rain-kl/Wavelet/internal/model"
"github.com/Rain-kl/Wavelet/internal/storage"
"github.com/Rain-kl/Wavelet/internal/task"
"gorm.io/gorm"
)
// 异步任务名称与管理类型定义
const (
// CleanupUnusedUploadsTask 清理未使用上传任务标识
CleanupUnusedUploadsTask = "upload:cleanup_unused"
// TaskTypeCleanupUploads 清理未使用上传管理类型
TaskTypeCleanupUploads = "cleanup_unused_uploads"
// WarmImageCacheTask 图片压缩缓存预热任务标识
WarmImageCacheTask = "upload:warm_image_cache"
// TaskTypeWarmImageCache 图片压缩缓存预热管理类型
@@ -34,18 +27,6 @@ const (
var warmImageCacheMu sync.Mutex
// CleanupUnusedUploadsMeta represents the task metadata.
var CleanupUnusedUploadsMeta = task.TaskMeta{
Type: TaskTypeCleanupUploads,
AsynqTask: CleanupUnusedUploadsTask,
Name: "清理未使用上传",
Description: "清理超过1小时未使用的上传文件",
SupportsTime: false,
MaxRetry: task.DefaultMaxRetry,
Queue: task.QueueDefault,
Retryable: true,
}
// WarmImageCacheMeta represents the image cache warmup task metadata.
var WarmImageCacheMeta = task.TaskMeta{
Type: TaskTypeWarmImageCache,
@@ -73,86 +54,12 @@ type WarmImageCachePayload struct {
Quality string `json:"quality"`
}
// CleanupUnusedUploadsHandler 清理未使用上传文件的异步任务处理器
type CleanupUnusedUploadsHandler struct{}
// SystemCleanupHandler 系统定期垃圾清理异步任务处理器
// WarmImageCacheHandler serially warms compressed image cache entries.
type WarmImageCacheHandler struct{}
// Execute 执行清理未使用上传文件的业务逻辑
func (h *CleanupUnusedUploadsHandler) Execute(ctx context.Context, _ []byte) (*task.TaskResult, error) {
if storageReadOnly(ctx) {
return nil, errors.New(ErrStorageReadOnly)
}
const batchSize = 100 // 每批处理100个文件
var lastID uint64
var totalProcessed int
var totalDeleted int
// 计算1小时前的时间
oneHourAgo := time.Now().Add(-1 * time.Hour)
task.AppendLog(ctx, "开始扫描未使用上传文件,阈值: %s", oneHourAgo.Format(time.RFC3339))
for {
// 使用游标分页查询未使用且超过1小时的上传记录
var unusedUploads []model.Upload
if err := db.DB(ctx).
Where("id > ? AND status = ? AND created_at < ?", lastID, model.UploadStatusPending, oneHourAgo).
Order("id ASC").
Limit(batchSize).
Find(&unusedUploads).Error; err != nil {
task.AppendLog(ctx, "查询未使用的上传文件失败: %v", err)
return nil, fmt.Errorf(ErrQueryUnusedUploadsFailed, err)
}
// 没有更多数据,退出循环
if len(unusedUploads) == 0 {
break
}
task.AppendLog(ctx, "本批次找到 %d 个需要清理的上传文件", len(unusedUploads))
// 处理每个未使用的上传文件
for _, upload := range unusedUploads {
totalProcessed++
if err := db.DB(ctx).Transaction(func(tx *gorm.DB) error {
// 更新上传记录状态
if err := tx.Model(&model.Upload{}).
Where("id = ? AND status = ?", upload.ID, model.UploadStatusPending).
Update("status", model.UploadStatusDeleted).Error; err != nil {
return err
}
driver := storage.Driver(upload.StorageDriver)
if driver == "" {
driver = storage.DriverLocal
}
backend, err := storage.ForDriver(ctx, driver)
if err != nil {
return err
}
if err := backend.Delete(ctx, upload.FilePath); err != nil {
return err
}
return nil
}); err != nil {
task.AppendLog(ctx, "清理上传文件失败 [ID:%d]: %v", upload.ID, err)
lastID = upload.ID
continue
}
totalDeleted++
lastID = upload.ID
}
}
msg := fmt.Sprintf("共处理 %d 个文件,成功删除 %d 个", totalProcessed, totalDeleted)
task.AppendLog(ctx, "%s", msg)
return &task.TaskResult{Message: msg}, nil
}
// Execute 执行系统清理(包含文件清理和历史消息推送日志清理)
// ValidatePayload validates and normalizes image cache warmup parameters.
func (h *WarmImageCacheHandler) ValidatePayload(payload []byte) ([]byte, error) {
+50 -9
View File
@@ -20,6 +20,7 @@ import (
"github.com/Rain-kl/Wavelet/internal/db"
"github.com/Rain-kl/Wavelet/internal/diskcache"
"github.com/Rain-kl/Wavelet/internal/model"
"github.com/Rain-kl/Wavelet/internal/service"
"github.com/Rain-kl/Wavelet/internal/storage"
"github.com/Rain-kl/Wavelet/internal/task"
"github.com/Rain-kl/Wavelet/internal/testhelper"
@@ -27,7 +28,7 @@ import (
"github.com/stretchr/testify/require"
)
func TestCleanupUnusedUploadsHandler_Execute(t *testing.T) {
func TestSystemCleanupHandler_Execute(t *testing.T) {
_, _, cleanup := testhelper.SetupTestEnvironment(t)
defer cleanup()
@@ -42,6 +43,8 @@ func TestCleanupUnusedUploadsHandler_Execute(t *testing.T) {
defer storageMock()
ctx := context.Background()
err := db.DB(ctx).AutoMigrate(&model.PushHistory{})
require.NoError(t, err)
// 准备测试数据:创建一些上传记录
now := time.Now()
@@ -81,14 +84,40 @@ func TestCleanupUnusedUploadsHandler_Execute(t *testing.T) {
require.NoError(t, err)
}
// 准备推送历史测试数据:1个旧的(应删除),1个新的(应保留)
oldPush := &model.PushHistory{
EventKey: "admin_login",
Channel: "email",
Target: "admin@test.com",
Title: "Old Login",
Content: "Old Content",
Level: "INFO",
Status: "success",
CreatedAt: now.AddDate(0, 0, -10),
}
newPush := &model.PushHistory{
EventKey: "admin_login",
Channel: "lark",
Target: "http://webhook.com",
Title: "New Login",
Content: "New Content",
Level: "INFO",
Status: "success",
CreatedAt: now,
}
err = db.DB(ctx).Create(oldPush).Error
require.NoError(t, err)
err = db.DB(ctx).Create(newPush).Error
require.NoError(t, err)
// 执行 handler
handler := &CleanupUnusedUploadsHandler{}
handler := &service.SystemCleanupHandler{}
result, err := handler.Execute(ctx, nil)
// 验证结果
require.NoError(t, err)
require.NotNil(t, result)
assert.Contains(t, result.Message, "共处理 2 个文件,成功删除 2 个")
assert.Contains(t, result.Message, "系统清理完成。成功清理未使用的上传文件 2/2 个;清理历史推送审计日志 1 条。")
// 验证数据库状态:pending 且超过1小时的应被标记为 deleted
var pendingCount int64
@@ -102,9 +131,19 @@ func TestCleanupUnusedUploadsHandler_Execute(t *testing.T) {
var usedCount int64
db.DB(ctx).Model(&model.Upload{}).Where("status = ?", model.UploadStatusUsed).Count(&usedCount)
assert.Equal(t, int64(1), usedCount, "used 状态的文件不应受影响")
// 验证推送历史数据状态:10天前的应被删除,今天的应保留
var pushCount int64
db.DB(ctx).Model(&model.PushHistory{}).Count(&pushCount)
assert.Equal(t, int64(1), pushCount, "应只剩1条推送历史记录")
var remainingPush model.PushHistory
err = db.DB(ctx).First(&remainingPush).Error
require.NoError(t, err)
assert.Equal(t, "New Login", remainingPush.Title)
}
func TestCleanupUnusedUploadsHandler_ExecuteNoFiles(t *testing.T) {
func TestSystemCleanupHandler_ExecuteNoFiles(t *testing.T) {
_, _, cleanup := testhelper.SetupTestEnvironment(t)
defer cleanup()
@@ -119,19 +158,21 @@ func TestCleanupUnusedUploadsHandler_ExecuteNoFiles(t *testing.T) {
defer storageMock()
ctx := context.Background()
err := db.DB(ctx).AutoMigrate(&model.PushHistory{})
require.NoError(t, err)
// 没有任何上传记录
handler := &CleanupUnusedUploadsHandler{}
handler := &service.SystemCleanupHandler{}
result, err := handler.Execute(ctx, nil)
require.NoError(t, err)
require.NotNil(t, result)
assert.Contains(t, result.Message, "共处理 0 个文件,成功删除 0 个")
assert.Contains(t, result.Message, "系统清理完成。成功清理未使用的上传文件 0/0 个;清理历史推送审计日志 0 条。")
}
func TestCleanupUnusedUploadsHandler_ImplementsTaskHandler(t *testing.T) {
// 编译期验证 CleanupUnusedUploadsHandler 实现了 TaskHandler 接口
var _ task.TaskHandler = (*CleanupUnusedUploadsHandler)(nil)
func TestSystemCleanupHandler_ImplementsTaskHandler(t *testing.T) {
// 编译期验证 SystemCleanupHandler 实现了 TaskHandler 接口
var _ task.TaskHandler = (*service.SystemCleanupHandler)(nil)
}
func TestWarmImageCacheHandlerValidatePayload(t *testing.T) {
+3
View File
@@ -9,6 +9,7 @@ import (
"strings"
"time"
"github.com/Rain-kl/Wavelet/internal/apps/admin/push/custom_events"
"github.com/Rain-kl/Wavelet/internal/apps/oauth"
"github.com/Rain-kl/Wavelet/internal/common"
"github.com/Rain-kl/Wavelet/internal/config"
@@ -171,6 +172,8 @@ func Login(c *gin.Context) {
logger.InfoF(ctx, "[LoginAudit] successful login for user: %s, ID: %d, IP: %s", user.Username, user.ID, c.ClientIP())
custom_events.TriggerAdminLoginEvent(ctx, &user, c.ClientIP())
c.JSON(http.StatusOK, util.OK(oauth.BuildBasicUserInfo(&user, needChangePassword)))
}
@@ -0,0 +1,34 @@
-- +goose Up
CREATE TABLE w_push_events (
id BIGSERIAL PRIMARY KEY,
event_key VARCHAR(80) NOT NULL UNIQUE,
name VARCHAR(100) NOT NULL,
channels TEXT NOT NULL,
targets TEXT NOT NULL,
template TEXT NOT NULL,
enabled BOOLEAN NOT NULL DEFAULT FALSE,
created_at TIMESTAMPTZ NOT NULL,
updated_at TIMESTAMPTZ NOT NULL
);
CREATE INDEX idx_w_push_events_enabled ON w_push_events(enabled);
CREATE TABLE w_push_histories (
id BIGSERIAL PRIMARY KEY,
event_key VARCHAR(80) NOT NULL,
channel VARCHAR(50) NOT NULL,
target VARCHAR(255) NOT NULL,
title VARCHAR(255) NOT NULL,
content TEXT NOT NULL,
level VARCHAR(20) NOT NULL,
status VARCHAR(20) NOT NULL,
error_msg TEXT,
created_at TIMESTAMPTZ NOT NULL
);
CREATE INDEX idx_w_push_histories_event ON w_push_histories(event_key);
CREATE INDEX idx_w_push_histories_created ON w_push_histories(created_at);
-- +goose Down
DROP TABLE IF EXISTS w_push_histories;
DROP TABLE IF EXISTS w_push_events;
@@ -0,0 +1,27 @@
-- +goose Up
INSERT INTO w_system_configs (key, value, type, visibility, description, created_at, updated_at)
VALUES (
'push_config',
'[]',
'system',
0,
'通知推送渠道配置',
CURRENT_TIMESTAMP,
CURRENT_TIMESTAMP
) ON CONFLICT (key) DO NOTHING;
INSERT INTO w_push_events (event_key, name, channels, targets, template, enabled, created_at, updated_at)
VALUES (
'admin_login',
'管理员登录',
'[]',
'[]',
'{"title": "管理员登录提醒", "content": "管理员 {{user.username}} 于 {{time}} 从 IP: {{ip}} 登录成功。", "level": "INFO"}',
false,
CURRENT_TIMESTAMP,
CURRENT_TIMESTAMP
) ON CONFLICT (event_key) DO NOTHING;
-- +goose Down
DELETE FROM w_system_configs WHERE key = 'push_config';
DELETE FROM w_push_events WHERE event_key = 'admin_login';
@@ -0,0 +1,8 @@
-- +goose Up
INSERT INTO w_users (id, username, password, nickname, avatar_url, is_active, is_admin, last_login_at, created_at, updated_at)
VALUES (999, 'system', '*', '系统', '', TRUE, FALSE, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
ON CONFLICT (username) DO NOTHING;
SELECT setval(pg_get_serial_sequence('w_users', 'id'), COALESCE((SELECT MAX(id) FROM w_users), 1));
-- +goose Down
DELETE FROM w_users WHERE username = 'system';
@@ -0,0 +1,31 @@
-- +goose Up
CREATE TABLE w_push_channels (
id BIGSERIAL PRIMARY KEY,
name VARCHAR(80) NOT NULL UNIQUE,
description VARCHAR(255),
type VARCHAR(50) NOT NULL DEFAULT 'custom',
token VARCHAR(100),
url TEXT NOT NULL,
other TEXT NOT NULL,
enabled BOOLEAN NOT NULL DEFAULT TRUE,
created_at TIMESTAMPTZ NOT NULL,
updated_at TIMESTAMPTZ NOT NULL
);
CREATE INDEX idx_w_push_channels_name ON w_push_channels(name);
CREATE INDEX idx_w_push_channels_enabled ON w_push_channels(enabled);
INSERT INTO w_system_configs (key, value, type, visibility, description, created_at, updated_at)
VALUES (
'push_global_token',
'',
'system',
0,
'系统全局推送鉴权令牌',
CURRENT_TIMESTAMP,
CURRENT_TIMESTAMP
) ON CONFLICT (key) DO NOTHING;
-- +goose Down
DELETE FROM w_system_configs WHERE key = 'push_global_token';
DROP TABLE IF EXISTS w_push_channels;
@@ -0,0 +1,11 @@
-- +goose Up
UPDATE w_schedules
SET name = '系统定期垃圾清理',
task_type = 'system_cleanup'
WHERE id = 1;
-- +goose Down
UPDATE w_schedules
SET name = '清理未使用上传',
task_type = 'cleanup_unused_uploads'
WHERE id = 1;
@@ -0,0 +1,34 @@
-- +goose Up
CREATE TABLE w_push_events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
event_key TEXT NOT NULL UNIQUE,
name TEXT NOT NULL,
channels TEXT NOT NULL,
targets TEXT NOT NULL,
template TEXT NOT NULL,
enabled BOOLEAN NOT NULL DEFAULT FALSE,
created_at DATETIME NOT NULL,
updated_at DATETIME NOT NULL
);
CREATE INDEX idx_w_push_events_enabled ON w_push_events(enabled);
CREATE TABLE w_push_histories (
id INTEGER PRIMARY KEY AUTOINCREMENT,
event_key TEXT NOT NULL,
channel TEXT NOT NULL,
target TEXT NOT NULL,
title TEXT NOT NULL,
content TEXT NOT NULL,
level TEXT NOT NULL,
status TEXT NOT NULL,
error_msg TEXT,
created_at DATETIME NOT NULL
);
CREATE INDEX idx_w_push_histories_event ON w_push_histories(event_key);
CREATE INDEX idx_w_push_histories_created ON w_push_histories(created_at);
-- +goose Down
DROP TABLE IF EXISTS w_push_histories;
DROP TABLE IF EXISTS w_push_events;
@@ -0,0 +1,27 @@
-- +goose Up
INSERT OR IGNORE INTO w_system_configs (key, value, type, visibility, description, created_at, updated_at)
VALUES (
'push_config',
'[]',
'system',
0,
'通知推送渠道配置',
CURRENT_TIMESTAMP,
CURRENT_TIMESTAMP
);
INSERT OR IGNORE INTO w_push_events (event_key, name, channels, targets, template, enabled, created_at, updated_at)
VALUES (
'admin_login',
'管理员登录',
'[]',
'[]',
'{"title": "管理员登录提醒", "content": "管理员 {{user.username}} 于 {{time}} 从 IP: {{ip}} 登录成功。", "level": "INFO"}',
false,
CURRENT_TIMESTAMP,
CURRENT_TIMESTAMP
);
-- +goose Down
DELETE FROM w_system_configs WHERE key = 'push_config';
DELETE FROM w_push_events WHERE event_key = 'admin_login';
@@ -0,0 +1,7 @@
-- +goose Up
INSERT INTO w_users (id, username, password, nickname, avatar_url, is_active, is_admin, last_login_at, created_at, updated_at)
VALUES (999, 'system', '*', '系统', '', TRUE, FALSE, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
ON CONFLICT (username) DO NOTHING;
-- +goose Down
DELETE FROM w_users WHERE username = 'system';
@@ -0,0 +1,31 @@
-- +goose Up
CREATE TABLE w_push_channels (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL UNIQUE,
description TEXT,
type TEXT NOT NULL DEFAULT 'custom',
token TEXT,
url TEXT NOT NULL,
other TEXT NOT NULL,
enabled BOOLEAN NOT NULL DEFAULT TRUE,
created_at DATETIME NOT NULL,
updated_at DATETIME NOT NULL
);
CREATE INDEX idx_w_push_channels_name ON w_push_channels(name);
CREATE INDEX idx_w_push_channels_enabled ON w_push_channels(enabled);
INSERT OR IGNORE INTO w_system_configs (key, value, type, visibility, description, created_at, updated_at)
VALUES (
'push_global_token',
'',
'system',
0,
'系统全局推送鉴权令牌',
CURRENT_TIMESTAMP,
CURRENT_TIMESTAMP
);
-- +goose Down
DELETE FROM w_system_configs WHERE key = 'push_global_token';
DROP TABLE IF EXISTS w_push_channels;
@@ -0,0 +1,11 @@
-- +goose Up
UPDATE w_schedules
SET name = '系统定期垃圾清理',
task_type = 'system_cleanup'
WHERE id = 1;
-- +goose Down
UPDATE w_schedules
SET name = '清理未使用上传',
task_type = 'cleanup_unused_uploads'
WHERE id = 1;
+1 -1
View File
@@ -17,7 +17,7 @@ import (
"gorm.io/gorm"
)
const expectedMigratedSystemConfigCount = 30
const expectedMigratedSystemConfigCount = 32
func TestMigrateInitializesSQLiteDatabase(t *testing.T) {
sqliteDB, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{
+104
View File
@@ -0,0 +1,104 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package model
import (
"context"
"encoding/json"
"errors"
"regexp"
"strings"
"time"
"github.com/Rain-kl/Wavelet/internal/db"
)
const (
// TypeCustom 自定义消息通道类型
TypeCustom = "custom"
// TypeEmail 邮件推送消息通道类型
TypeEmail = "email"
)
// PushChannel 消息通道模型
type PushChannel struct {
ID uint64 `json:"id" gorm:"primaryKey;autoIncrement"`
Name string `json:"name" gorm:"uniqueIndex;size:80;not null"` // 通道名称,仅英文字母和下划线,唯一
Description string `json:"description" gorm:"size:255"` // 备注
Type string `json:"type" gorm:"size:50;not null;default:'custom'"` // 通道类型:custom, lark, email
Token string `json:"token" gorm:"size:100"` // 鉴权令牌或发信用户名等
URL string `json:"url" gorm:"type:text;not null"` // 请求地址,HTTPS 协议或 SMTP 地址
Other string `json:"other" gorm:"type:text;not null"` // 请求体/SMTP 密码等
Enabled bool `json:"enabled" gorm:"index;not null;default:true"` // 通道是否启用
CreatedAt time.Time `json:"created_at" gorm:"autoCreateTime;index"`
UpdatedAt time.Time `json:"updated_at" gorm:"autoUpdateTime;index"`
}
// TableName 指定 GORM 表名
func (PushChannel) TableName() string {
return "w_push_channels"
}
var nameRegex = regexp.MustCompile(`^[a-zA-Z0-9_]+$`)
// Validate 参数合法性与 JSON 格式校验
func (pc *PushChannel) Validate() error {
pc.Name = strings.TrimSpace(pc.Name)
pc.URL = strings.TrimSpace(pc.URL)
pc.Other = strings.TrimSpace(pc.Other)
pc.Type = strings.TrimSpace(pc.Type)
if pc.Type == "" {
pc.Type = TypeCustom
}
if pc.Name == "" {
return errors.New("channel name is required")
}
if !nameRegex.MatchString(pc.Name) {
return errors.New("channel name can only contain letters, numbers, and underscores")
}
if pc.Type != TypeEmail && pc.URL == "" {
return errors.New("request URL/address is required")
}
// For custom and lark, we must enforce https:// URL prefix for security.
// For email, it is an SMTP host:port, so no need for https:// prefix.
if pc.Type != TypeEmail && !strings.HasPrefix(pc.URL, "https://") {
return errors.New("request URL must use HTTPS protocol for security reasons")
}
switch pc.Type {
case TypeCustom:
if pc.Other == "" {
return errors.New("payload schema (request body) is required")
}
return validateJSON(pc.Other)
case TypeEmail:
// Email channel SMTP configs fall back to global settings, so they are not required to be filled.
}
return nil
}
func validateJSON(s string) error {
var jsonTest map[string]any
if err := json.Unmarshal([]byte(s), &jsonTest); err == nil {
return nil
}
var jsonArr []any
if err := json.Unmarshal([]byte(s), &jsonArr); err == nil {
return nil
}
return errors.New("payload schema must be a valid JSON format")
}
// GetPushChannelByName 根据名称获取消息通道
func GetPushChannelByName(ctx context.Context, name string) (*PushChannel, error) {
var channel PushChannel
err := db.DB(ctx).Where("name = ?", name).First(&channel).Error
if err != nil {
return nil, err
}
return &channel, nil
}
+59
View File
@@ -0,0 +1,59 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package model
import (
"context"
"errors"
"strings"
"time"
"github.com/Rain-kl/Wavelet/internal/db"
)
// PushEvent 系统通知事件模型
type PushEvent struct {
ID uint64 `json:"id" gorm:"primaryKey;autoIncrement"`
EventKey string `json:"event_key" gorm:"uniqueIndex;size:80;not null"` // 如 admin_login
Name string `json:"name" gorm:"size:100;not null"` // 如 管理员登录
Channels []string `json:"channels" gorm:"type:text;serializer:json;not null"` // 推送渠道列表,如 ["lark"]
Targets []string `json:"targets" gorm:"type:text;serializer:json;not null"` // 推送目标用户/邮箱列表
Template string `json:"template" gorm:"type:text;not null"` // 消息模板 JSON
Enabled bool `json:"enabled" gorm:"index;not null;default:false"` // 是否启用
CreatedAt time.Time `json:"created_at" gorm:"autoCreateTime;index"`
UpdatedAt time.Time `json:"updated_at" gorm:"autoUpdateTime;index"`
}
// TableName 指定 GORM 表名
func (PushEvent) TableName() string {
return "w_push_events"
}
// Validate 基础校验
func (pe *PushEvent) Validate() error {
pe.EventKey = strings.TrimSpace(pe.EventKey)
pe.Name = strings.TrimSpace(pe.Name)
pe.Template = strings.TrimSpace(pe.Template)
if pe.EventKey == "" {
return errors.New("event key is required")
}
if pe.Name == "" {
return errors.New("event name is required")
}
if pe.Template == "" {
return errors.New("event template is required")
}
return nil
}
// GetActivePushEventByKey 获取启用的通知事件
func GetActivePushEventByKey(ctx context.Context, key string) (*PushEvent, error) {
var event PushEvent
err := db.DB(ctx).Where("event_key = ? AND enabled = ?", key, true).First(&event).Error
if err != nil {
return nil, err
}
return &event, nil
}
+27
View File
@@ -0,0 +1,27 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package model
import (
"time"
)
// PushHistory 推送日志/历史实体
type PushHistory struct {
ID uint64 `json:"id" gorm:"primaryKey;autoIncrement"`
EventKey string `json:"event_key" gorm:"size:80;not null;index"`
Channel string `json:"channel" gorm:"size:50;not null"`
Target string `json:"target" gorm:"size:255;not null"`
Title string `json:"title" gorm:"size:255;not null"`
Content string `json:"content" gorm:"type:text;not null"`
Level string `json:"level" gorm:"size:20;not null"`
Status string `json:"status" gorm:"size:20;not null"` // success / failed
ErrorMsg string `json:"error_msg" gorm:"type:text"`
CreatedAt time.Time `json:"created_at" gorm:"autoCreateTime;index"`
}
// TableName 指定表名
func (PushHistory) TableName() string {
return "w_push_histories"
}
+1
View File
@@ -49,6 +49,7 @@ const (
ConfigKeyLoginSessionTTLHours = "login_session_ttl_hours" // 登录会话过期时间 (小时,0表示浏览器关闭后自动退出登录,-1表示永不过期)
ConfigKeyUpdateUpstreamRepository = "update_upstream_repository" // GitHub Actions Release 上游仓库
ConfigKeyStorageConfig = "storage_config" // 文件存储配置 (JSON)
ConfigKeyPushConfig = "push_config" // 通知推送渠道配置 (JSON)
)
const (
+12 -12
View File
@@ -56,7 +56,7 @@ func TestCreateTaskExecution(t *testing.T) {
execution := &TaskExecution{
TaskID: "manual_cleanup_123",
TaskType: "upload:cleanup_unused",
TaskType: "system:cleanup",
TaskName: "清理未使用上传",
Status: TaskExecutionStatusPending,
Retryable: true,
@@ -81,7 +81,7 @@ func TestGetTaskExecutionByTaskID(t *testing.T) {
// 创建记录
execution := &TaskExecution{
TaskID: "test_task_id_001",
TaskType: "upload:cleanup_unused",
TaskType: "system:cleanup",
TaskName: "清理未使用上传",
Status: TaskExecutionStatusPending,
Retryable: true,
@@ -112,7 +112,7 @@ func TestGetTaskExecutionByID(t *testing.T) {
execution := &TaskExecution{
TaskID: "test_by_id_001",
TaskType: "upload:cleanup_unused",
TaskType: "system:cleanup",
TaskName: "清理未使用上传",
Status: TaskExecutionStatusPending,
TriggeredBy: "system",
@@ -134,7 +134,7 @@ func TestUpdateTaskExecution(t *testing.T) {
// 创建记录
execution := &TaskExecution{
TaskID: "test_update_001",
TaskType: "upload:cleanup_unused",
TaskType: "system:cleanup",
TaskName: "清理未使用上传",
Status: TaskExecutionStatusPending,
TriggeredBy: "manual",
@@ -178,7 +178,7 @@ func TestUpdateTaskExecutionFailed(t *testing.T) {
execution := &TaskExecution{
TaskID: "test_fail_001",
TaskType: "upload:cleanup_unused",
TaskType: "system:cleanup",
TaskName: "清理未使用上传",
Status: TaskExecutionStatusPending,
Retryable: true,
@@ -212,7 +212,7 @@ func TestUpdateTaskExecutionDoesNotPersistBufferedLog(t *testing.T) {
execution := &TaskExecution{
TaskID: "test_omit_log_001",
TaskType: "upload:cleanup_unused",
TaskType: "system:cleanup",
TaskName: "清理未使用上传",
Status: TaskExecutionStatusPending,
TriggeredBy: "manual",
@@ -249,7 +249,7 @@ func TestAppendTaskExecutionLog(t *testing.T) {
execution := &TaskExecution{
TaskID: "test_log_001",
TaskType: "upload:cleanup_unused",
TaskType: "system:cleanup",
TaskName: "清理未使用上传",
Status: TaskExecutionStatusPending,
TriggeredBy: "manual",
@@ -334,7 +334,7 @@ func TestGetTaskExecutionLogPrefersRedis(t *testing.T) {
execution := &TaskExecution{
TaskID: "redis_priority_001",
TaskType: "upload:cleanup_unused",
TaskType: "system:cleanup",
TaskName: "清理未使用上传",
Status: TaskExecutionStatusRunning,
Log: "数据库旧日志",
@@ -358,10 +358,10 @@ func TestListTaskExecutions(t *testing.T) {
// 创建多条记录,包含不同状态和类型
records := []*TaskExecution{
{TaskID: "list_001", TaskType: "upload:cleanup_unused", TaskName: "清理上传", Status: TaskExecutionStatusSucceeded, TriggeredBy: "manual"},
{TaskID: "list_002", TaskType: "upload:cleanup_unused", TaskName: "清理上传", Status: TaskExecutionStatusFailed, TriggeredBy: "system"},
{TaskID: "list_001", TaskType: "system:cleanup", TaskName: "系统垃圾清理", Status: TaskExecutionStatusSucceeded, TriggeredBy: "manual"},
{TaskID: "list_002", TaskType: "system:cleanup", TaskName: "系统垃圾清理", Status: TaskExecutionStatusFailed, TriggeredBy: "system"},
{TaskID: "list_003", TaskType: "other:task", TaskName: "其他任务", Status: TaskExecutionStatusPending, TriggeredBy: "manual"},
{TaskID: "list_004", TaskType: "upload:cleanup_unused", TaskName: "清理上传", Status: TaskExecutionStatusRunning, TriggeredBy: "manual"},
{TaskID: "list_004", TaskType: "system:cleanup", TaskName: "系统垃圾清理", Status: TaskExecutionStatusRunning, TriggeredBy: "manual"},
{TaskID: "list_005", TaskType: "other:task", TaskName: "其他任务", Status: TaskExecutionStatusSucceeded, TriggeredBy: "system"},
}
for _, r := range records {
@@ -409,7 +409,7 @@ func TestListTaskExecutions(t *testing.T) {
assert.NotEqual(t, items[0].ID, items2[0].ID)
// 状态 + 类型组合筛选
items, total, err = ListTaskExecutions(ctx, ListTaskExecutionsRequest{Status: "succeeded", TaskType: "upload:cleanup_unused", Page: 1, PageSize: 10})
items, total, err = ListTaskExecutions(ctx, ListTaskExecutionsRequest{Status: "succeeded", TaskType: "system:cleanup", Page: 1, PageSize: 10})
require.NoError(t, err)
assert.Equal(t, int64(1), total)
assert.Equal(t, "list_001", items[0].TaskID)
+26
View File
@@ -20,6 +20,7 @@ import (
admin_cache "github.com/Rain-kl/Wavelet/internal/apps/admin/cache"
admin_db_manage "github.com/Rain-kl/Wavelet/internal/apps/admin/db_manage"
admin_logs "github.com/Rain-kl/Wavelet/internal/apps/admin/logs"
admin_push "github.com/Rain-kl/Wavelet/internal/apps/admin/push"
admin_status "github.com/Rain-kl/Wavelet/internal/apps/admin/status"
admin_task "github.com/Rain-kl/Wavelet/internal/apps/admin/task"
admin_template "github.com/Rain-kl/Wavelet/internal/apps/admin/template"
@@ -37,6 +38,7 @@ import (
// Swagger 文档生成
_ "github.com/Rain-kl/Wavelet/docs"
_ "github.com/Rain-kl/Wavelet/internal/apps/admin/push/custom_events"
"github.com/Rain-kl/Wavelet/internal/apps/admin/system_config"
"github.com/Rain-kl/Wavelet/internal/apps/oauth"
"github.com/Rain-kl/Wavelet/internal/config"
@@ -59,6 +61,11 @@ func Serve() {
// 初始化 ClickHouse 异步日志写入器
risk_control.InitLogWriter()
// 运行内置事件同步
if err := admin_push.SyncEvents(context.Background()); err != nil {
log.Printf("[API] sync push events failed: %v\n", err)
}
// 初始化路由
r := gin.New()
r.Use(gin.Recovery())
@@ -171,6 +178,7 @@ func registerRoutes(r *gin.Engine) {
// Register custom business routes
registerCustomRoutes(apiV1Router)
}
}
@@ -332,6 +340,24 @@ func registerAdminRoutes(apiV1Router *gin.RouterGroup) {
adminRouter.PUT("/auth-sources/:id", admin_auth_source.UpdateAuthSource)
adminRouter.PUT("/auth-sources/:id/toggle", admin_auth_source.ToggleAuthSource)
adminRouter.DELETE("/auth-sources/:id", admin_auth_source.DeleteAuthSource)
// Push Notifications
adminRouter.GET("/push/events", admin_push.ListEvents)
adminRouter.GET("/push/events/builtin", admin_push.ListBuiltInEvents)
adminRouter.POST("/push/events", admin_push.CreateEvent)
adminRouter.PUT("/push/events/:id", admin_push.UpdateEvent)
adminRouter.DELETE("/push/events/:id", admin_push.DeleteEvent)
adminRouter.POST("/push/events/:id/toggle", admin_push.ToggleEvent)
adminRouter.GET("/push/histories", admin_push.ListHistories)
adminRouter.POST("/push/test", admin_push.TestPush)
// Message Channels CRUD
adminRouter.GET("/push/channels/definitions", admin_push.ListChannelDefinitions)
adminRouter.GET("/push/channels", admin_push.ListChannels)
adminRouter.POST("/push/channels", admin_push.CreateChannel)
adminRouter.PUT("/push/channels/:id", admin_push.UpdateChannel)
adminRouter.DELETE("/push/channels/:id", admin_push.DeleteChannel)
adminRouter.POST("/push/channels/test", admin_push.TestChannel)
}
}
+149
View File
@@ -0,0 +1,149 @@
// Copyright 2026 Arctel.net
// SPDX-License-Identifier: Apache-2.0
package service
import (
"context"
"errors"
"fmt"
"time"
"github.com/Rain-kl/Wavelet/internal/db"
"github.com/Rain-kl/Wavelet/internal/logger"
"github.com/Rain-kl/Wavelet/internal/model"
"github.com/Rain-kl/Wavelet/internal/storage"
"github.com/Rain-kl/Wavelet/internal/task"
"gorm.io/gorm"
)
// 异步任务名称与管理类型定义
const (
// SystemCleanupTask 系统定期垃圾清理任务标识
SystemCleanupTask = "system:cleanup"
// TaskTypeSystemCleanup 系统定期垃圾清理管理类型
TaskTypeSystemCleanup = "system_cleanup"
// 错误描述常量
errStorageReadOnly = "存储迁移维护中,当前仅允许读取文件"
errQueryUnusedUploadsFailed = "查询未使用的上传文件失败: %w"
)
// SystemCleanupMeta represents the task metadata.
var SystemCleanupMeta = task.TaskMeta{
Type: TaskTypeSystemCleanup,
AsynqTask: SystemCleanupTask,
Name: "系统垃圾清理",
Description: "定期清理超过1小时的未使用上传文件和超过7天的历史推送记录",
SupportsTime: false,
MaxRetry: task.DefaultMaxRetry,
Queue: task.QueueDefault,
Retryable: true,
}
// SystemCleanupHandler 系统定期垃圾清理异步任务处理器
type SystemCleanupHandler struct{}
// Execute 执行系统清理(包含文件清理和历史消息推送日志清理)
func (h *SystemCleanupHandler) Execute(ctx context.Context, _ []byte) (*task.TaskResult, error) {
if storageReadOnly(ctx) {
return nil, errors.New(errStorageReadOnly)
}
const batchSize = 100 // 每批处理100个文件
var lastID uint64
var totalProcessed int
var totalDeleted int
// 计算1小时前的时间
oneHourAgo := time.Now().Add(-1 * time.Hour)
task.AppendLog(ctx, "开始扫描未使用上传文件,阈值: %s", oneHourAgo.Format(time.RFC3339))
for {
// 使用游标分页查询未使用且超过1小时的上传记录
var unusedUploads []model.Upload
if err := db.DB(ctx).
Where("id > ? AND status = ? AND created_at < ?", lastID, model.UploadStatusPending, oneHourAgo).
Order("id ASC").
Limit(batchSize).
Find(&unusedUploads).Error; err != nil {
task.AppendLog(ctx, "查询未使用的上传文件失败: %v", err)
return nil, fmt.Errorf(errQueryUnusedUploadsFailed, err)
}
// 没有更多数据,退出循环
if len(unusedUploads) == 0 {
break
}
task.AppendLog(ctx, "本批次找到 %d 个需要清理的上传文件", len(unusedUploads))
// 处理每个未使用的上传文件
for _, u := range unusedUploads {
totalProcessed++
if err := db.DB(ctx).Transaction(func(tx *gorm.DB) error {
// 更新上传记录状态
if err := tx.Model(&model.Upload{}).
Where("id = ? AND status = ?", u.ID, model.UploadStatusPending).
Update("status", model.UploadStatusDeleted).Error; err != nil {
return err
}
driver := storage.Driver(u.StorageDriver)
if driver == "" {
driver = storage.DriverLocal
}
backend, err := storage.ForDriver(ctx, driver)
if err != nil {
return err
}
if err := backend.Delete(ctx, u.FilePath); err != nil {
return err
}
return nil
}); err != nil {
task.AppendLog(ctx, "清理上传文件失败 [ID:%d]: %v", u.ID, err)
lastID = u.ID
continue
}
totalDeleted++
lastID = u.ID
}
}
// 2. 清理超过7天的历史推送日志
task.AppendLog(ctx, "开始清理历史推送审计日志,只保留最近7天数据...")
cutoff := time.Now().AddDate(0, 0, -7)
var pushHistoryCount int64
if err := db.DB(ctx).Model(&model.PushHistory{}).Where("created_at < ?", cutoff).Count(&pushHistoryCount).Error; err != nil {
task.AppendLog(ctx, "统计待清理的历史推送记录失败: %v", err)
} else if pushHistoryCount > 0 {
if err := db.DB(ctx).Where("created_at < ?", cutoff).Delete(&model.PushHistory{}).Error; err != nil {
task.AppendLog(ctx, "删除历史推送记录失败: %v", err)
} else {
task.AppendLog(ctx, "成功删除 %d 条历史推送记录 (截止时间: %s)", pushHistoryCount, cutoff.Format("2006-01-02 15:04:05"))
}
} else {
task.AppendLog(ctx, "没有需要清理的历史推送记录 (截止时间: %s)", cutoff.Format("2006-01-02 15:04:05"))
}
msg := fmt.Sprintf("系统清理完成。成功清理未使用的上传文件 %d/%d 个;清理历史推送审计日志 %d 条。", totalDeleted, totalProcessed, pushHistoryCount)
task.AppendLog(ctx, "%s", msg)
return &task.TaskResult{Message: msg}, nil
}
func storageReadOnly(ctx context.Context) bool {
var execution model.TaskExecution
err := db.DB(ctx).Where("task_type = ?", "storage:migrate").Order("id DESC").First(&execution).Error
if err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return false
}
logger.ErrorF(ctx, "读取存储维护状态失败: %v", err)
return true
}
return execution.Status != model.TaskExecutionStatusSucceeded
}
+33 -6
View File
@@ -67,6 +67,16 @@ func GetTaskID(ctx context.Context) string {
return ""
}
// IsFinalAttempt 判断当前任务执行是否为最后一次重试尝试(若再次失败即为最终失败)
func IsFinalAttempt(ctx context.Context) bool {
retryCount, hasRetryCount := asynq.GetRetryCount(ctx)
maxRetry, hasMaxRetry := asynq.GetMaxRetry(ctx)
if !hasRetryCount || !hasMaxRetry {
return true
}
return retryCount >= maxRetry
}
// AppendLog 追加日志到任务执行记录
// 在 TaskHandler.Execute 中调用,日志会自动追加到 TaskExecution.Log 字段
func AppendLog(ctx context.Context, format string, args ...interface{}) {
@@ -235,12 +245,8 @@ func ProcessTask(ctx context.Context, t *asynq.Task) error {
// 加载或动态创建执行记录
now := time.Now()
execution, err := getOrCreateTaskExecution(ctx, taskID, t, now)
if err == nil && execution != nil && execution.TriggeredBy != "schedule" {
execution.Status = model.TaskExecutionStatusRunning
execution.StartedAt = &now
if updateErr := model.UpdateTaskExecution(ctx, execution); updateErr != nil {
logger.ErrorF(ctx, "[TaskExecutor] 更新执行状态失败 taskID=%s: %v", taskID, updateErr)
}
if err == nil {
updateExecutionOnStart(ctx, execution, now)
}
if execution != nil {
@@ -270,6 +276,27 @@ func ProcessTask(ctx context.Context, t *asynq.Task) error {
return execErr
}
func updateExecutionOnStart(ctx context.Context, execution *model.TaskExecution, now time.Time) {
if execution == nil {
return
}
dirty := false
if retryCount, hasRetry := asynq.GetRetryCount(ctx); hasRetry && execution.RetryCount != retryCount {
execution.RetryCount = retryCount
dirty = true
}
if execution.Status != model.TaskExecutionStatusRunning {
execution.Status = model.TaskExecutionStatusRunning
execution.StartedAt = &now
dirty = true
}
if dirty {
if updateErr := model.UpdateTaskExecution(ctx, execution); updateErr != nil {
logger.ErrorF(ctx, "[TaskExecutor] 更新执行状态失败 taskID=%s: %v", execution.TaskID, updateErr)
}
}
}
// getOrCreateTaskExecution 获取已有的任务执行记录,如果不存在则针对已知任务类型动态创建记录
func getOrCreateTaskExecution(ctx context.Context, taskID string, t *asynq.Task, now time.Time) (*model.TaskExecution, error) {
execution, err := model.GetTaskExecutionByTaskID(ctx, taskID)
+10 -2
View File
@@ -6,8 +6,10 @@
package handlers
import (
"github.com/Rain-kl/Wavelet/internal/apps/admin/push"
"github.com/Rain-kl/Wavelet/internal/apps/upload"
"github.com/Rain-kl/Wavelet/internal/apps/user"
"github.com/Rain-kl/Wavelet/internal/service"
"github.com/Rain-kl/Wavelet/internal/task"
)
@@ -16,13 +18,19 @@ func Register() {
task.RegisterHandler(upload.StorageMigrationTask, &upload.MigrationHandler{})
task.RegisterTaskMeta(upload.StorageMigrationMeta)
// system cleanup
task.RegisterHandler(service.SystemCleanupTask, &service.SystemCleanupHandler{})
task.RegisterTaskMeta(service.SystemCleanupMeta)
// upload
task.RegisterHandler(upload.CleanupUnusedUploadsTask, &upload.CleanupUnusedUploadsHandler{})
task.RegisterTaskMeta(upload.CleanupUnusedUploadsMeta)
task.RegisterHandler(upload.WarmImageCacheTask, &upload.WarmImageCacheHandler{})
task.RegisterTaskMeta(upload.WarmImageCacheMeta)
// user
task.RegisterHandler(user.SendEmailTask, &user.SendEmailHandler{})
task.RegisterTaskMeta(user.SendEmailMeta)
// push
task.RegisterHandler(push.SendNotificationTask, &push.PushHandler{})
task.RegisterTaskMeta(push.SendNotificationMeta)
}