admin/TaskPool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add notify icon read

Browse Source
This commit is contained in:
engigu
2026-03-09 18:10:23 +08:00
parent e7da316c90
commit 77cf12c9b6
46 changed files with 1347 additions and 252 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cache/site_cache.go
Vendored
+1 -1
View File
@@ -31,7 +31,7 @@ func LoadSiteCache() {
var settings []models.Setting
database.DB.Where("section = ?", constant.SectionSite).Find(&settings)
for _, setting := range settings {
siteCache[setting.Key] = setting.Value
siteCache[setting.Key] = string(setting.Value)
}
siteCacheInit = true
}
internal/constant/constant.go
+19
View File
@@ -74,6 +74,10 @@ const (
EventTaskFailed = "task_failed"
EventTaskTimeout = "task_timeout"
// 其他事件类型
EventSystemNotice = "system_notice"
EventNotifySent = "notify_sent"
// WebSocket 消息类型
WSTypeHeartbeat = "heartbeat"
WSTypeHeartbeatAck = "heartbeat_ack"
@@ -110,6 +114,21 @@ const (
// Agent 状态
AgentStatusOnline = "online"
AgentStatusOffline = "offline"
// AppLog 分类
LogCategoryDefault = "default"
LogCategorySystemNotice = "system_notice"
LogCategoryPushLog = "push_log"
// AppLog 级别
LogLevelInfo = "info"
LogLevelWarning = "warning"
LogLevelError = "error"
// AppLog 状态
LogStatusUnread = "unread"
LogStatusRead = "read"
LogStatusSuccess = "success"
LogStatusFailed = "failed"
)
// TablePrefix 表前缀，从配置文件读取
internal/controllers/app_log_controller.go
+108
View File
@@ -0,0 +1,108 @@
package controllers
import (
"github.com/engigu/baihu-panel/internal/services"
"github.com/engigu/baihu-panel/internal/utils"
"github.com/gin-gonic/gin"
)
type AppLogController struct {
appLogService *services.AppLogService
}
func NewAppLogController() *AppLogController {
return &AppLogController{
appLogService: services.NewAppLogService(),
}
}
// GetLogs 获取应用日志列表
// @Summary 获取应用日志列表
// @Tags 应用日志
// @Accept json
// @Produce json
// @Security BearerAuth
// @Param category query string false "日志分类"
// @Param status query string false "状态"
// @Param level query string false "级别"
// @Param keyword query string false "搜索关键词"
// @Param page query int false "页码"
// @Param page_size query int false "每页数量"
// @Success 200 {object} utils.Response
// @Router /app-logs [get]
func (ac *AppLogController) GetLogs(c *gin.Context) {
p := utils.ParsePagination(c)
category := c.Query("category")
status := c.Query("status")
level := c.Query("level")
keyword := c.Query("keyword")
logs, total, err := ac.appLogService.List(category, status, level, p.Page, p.PageSize, keyword)
if err != nil {
utils.BadRequest(c, err.Error())
return
}
utils.PaginatedResponse(c, logs, total, p)
}
// MarkAsRead 标记已读
// @Summary 标记已读
// @Tags 应用日志
// @Accept json
// @Produce json
// @Security BearerAuth
// @Param body body object true "请求体"
// @Success 200 {object} utils.Response
// @Router /app-logs/read [post]
func (ac *AppLogController) MarkAsRead(c *gin.Context) {
var req struct {
ID string `json:"id"`
Category string `json:"category"`
}
if err := c.ShouldBindJSON(&req); err != nil {
utils.BadRequest(c, err.Error())
return
}
if req.ID != "" {
if err := ac.appLogService.MarkAsRead(req.ID); err != nil {
utils.BadRequest(c, err.Error())
return
}
} else if req.Category != "" {
if err := ac.appLogService.MarkAllAsRead(req.Category); err != nil {
utils.BadRequest(c, err.Error())
return
}
} else {
utils.BadRequest(c, "id 或 category 必须提供")
return
}
utils.SuccessMsg(c, "标记成功")
}
// ClearLogs 清理日志
// @Summary 清理日志
// @Tags 应用日志
// @Accept json
// @Produce json
// @Security BearerAuth
// @Param body body object true "请求体"
// @Success 200 {object} utils.Response
// @Router /app-logs/clear [post]
func (ac *AppLogController) ClearLogs(c *gin.Context) {
var req struct {
Category string `json:"category"`
}
if err := c.ShouldBindJSON(&req); err != nil {
utils.BadRequest(c, err.Error())
return
}
if err := ac.appLogService.Clear(req.Category); err != nil {
utils.BadRequest(c, err.Error())
return
}
utils.SuccessMsg(c, "清理成功")
}
internal/controllers/auth_controller.go
+37 -11
View File
@@ -7,6 +7,7 @@ import (
"time"
"github.com/engigu/baihu-panel/internal/constant"
"github.com/engigu/baihu-panel/internal/eventbus"
"github.com/engigu/baihu-panel/internal/middleware"
"github.com/engigu/baihu-panel/internal/models/vo"
"github.com/engigu/baihu-panel/internal/services"
@@ -54,10 +55,13 @@ func (ac *AuthController) Login(c *gin.Context) {
if val, ok := loginAttempts.Load(ip); ok {
attempt := val.(*loginAttempt)
if attempt.Count >= 5 && time.Since(attempt.LastAttempt) < time.Minute {
ac.loginLogService.Create(req.Username, ip, userAgent, "failed", "尝试次数过多，请一分钟后再试")
go services.NewNotificationService().TriggerEvent(constant.BindingTypeSystem, constant.EventBruteForceLogin, "", map[string]interface{}{
"ip": ip,
"username": req.Username,
eventbus.DefaultBus.Publish(eventbus.Event{
Type: constant.EventBruteForceLogin,
Payload: map[string]interface{}{
"ip": ip,
"username": req.Username,
"userAgent": userAgent,
},
})
utils.TooManyRequests(c, "尝试次数过多，请一分钟后再试")
return
@@ -77,7 +81,16 @@ func (ac *AuthController) Login(c *gin.Context) {
attempt.LastAttempt = time.Now()
// 记录登录失败日志
ac.loginLogService.Create(req.Username, ip, userAgent, "failed", "用户名或密码错误")
eventbus.DefaultBus.Publish(eventbus.Event{
Type: constant.EventUserLogin,
Payload: map[string]interface{}{
"ip": ip,
"username": req.Username,
"userAgent": userAgent,
"status": "failed",
"message": "用户名或密码错误",
},
})
utils.Unauthorized(c, "用户名或密码错误")
return
}
@@ -96,7 +109,16 @@ func (ac *AuthController) Login(c *gin.Context) {
// 生成 token
token, err := utils.GenerateToken(user.ID, user.Username, expireDays, constant.Secret)
if err != nil {
ac.loginLogService.Create(req.Username, ip, userAgent, "failed", "Token生成失败")
eventbus.DefaultBus.Publish(eventbus.Event{
Type: constant.EventUserLogin,
Payload: map[string]interface{}{
"ip": ip,
"username": req.Username,
"userAgent": userAgent,
"status": "failed",
"message": "Token生成失败",
},
})
utils.ServerError(c, "登录失败")
return
}
@@ -105,11 +127,15 @@ func (ac *AuthController) Login(c *gin.Context) {
middleware.SetAuthCookie(c, token, expireDays)
// 记录登录成功日志
ac.loginLogService.Create(req.Username, ip, userAgent, "success", "登录成功")
go services.NewNotificationService().TriggerEvent(constant.BindingTypeSystem, constant.EventUserLogin, "", map[string]interface{}{
"ip": ip,
"username": req.Username,
eventbus.DefaultBus.Publish(eventbus.Event{
Type: constant.EventUserLogin,
Payload: map[string]interface{}{
"ip": ip,
"username": req.Username,
"userAgent": userAgent,
"status": "success",
"message": "登录成功",
},
})
utils.Success(c, gin.H{
internal/controllers/log_controller.go
+1 -1
View File
@@ -87,7 +87,7 @@ func (lc *LogController) GetLogs(c *gin.Context) {
TaskName: task.Name,
TaskType: taskType,
AgentID: log.AgentID,
Command: log.Command,
Command: string(log.Command),
Status: log.Status,
Duration: log.Duration,
StartTime: log.StartTime,
internal/controllers/log_ws_controller.go
+2 -2
View File
@@ -37,7 +37,7 @@ func (lc *LogWSController) StreamLog(c *gin.Context) {
if err := database.DB.Where("id = ?", logID).First(&taskLog).Error; err == nil {
if taskLog.Status != "running" {
// 已结束，读取库内日志
content, err := utils.DecompressFromBase64(taskLog.Output)
content, err := utils.DecompressFromBase64(string(taskLog.Output))
if err != nil {
conn.WriteMessage(websocket.TextMessage, []byte("解压日志失败: "+err.Error()))
return
@@ -75,7 +75,7 @@ func (lc *LogWSController) StreamLog(c *gin.Context) {
// 任务结束，尝试刷新最后一次库内完整内容
var finalLog models.TaskLog
if err := database.DB.Where("id = ?", logID).First(&finalLog).Error; err == nil {
content, _ := utils.DecompressFromBase64(finalLog.Output)
content, _ := utils.DecompressFromBase64(string(finalLog.Output))
if content != "" {
conn.WriteMessage(websocket.TextMessage, []byte("\n--- 任务已结束 ---\n"))
// 这里可以选择性再推一次完整版，或直接退出
internal/controllers/settings_controller.go
+6 -3
View File
@@ -10,9 +10,9 @@ import (
"os"
"strings"
"time"
"github.com/engigu/baihu-panel/internal/constant"
"github.com/engigu/baihu-panel/internal/database"
"github.com/engigu/baihu-panel/internal/eventbus"
"github.com/engigu/baihu-panel/internal/models"
"github.com/engigu/baihu-panel/internal/models/vo"
"github.com/engigu/baihu-panel/internal/services"
@@ -76,8 +76,11 @@ func (sc *SettingsController) ChangePassword(c *gin.Context) {
return
}
go services.NewNotificationService().TriggerEvent(constant.BindingTypeSystem, constant.EventPasswordChanged, "", map[string]interface{}{
"username": user.Username,
eventbus.DefaultBus.Publish(eventbus.Event{
Type: constant.EventPasswordChanged,
Payload: map[string]interface{}{
"username": user.Username,
},
})
utils.SuccessMsg(c, "密码修改成功")
internal/database/database.go
+2
View File
@@ -15,6 +15,7 @@ import (
)
var DB *gorm.DB
var DBConfig *Config
type Config struct {
Type string // sqlite, mysql, postgres
@@ -28,6 +29,7 @@ type Config struct {
func Init(cfg *Config) error {
var err error
DBConfig = cfg
// 设置东八区时区
loc := systime.CST
time.Local = loc
internal/database/migrate.go
+1 -62
View File
@@ -1,13 +1,10 @@
package database
import (
"fmt"
"reflect"
"strings"
"github.com/engigu/baihu-panel/internal/logger"
"github.com/engigu/baihu-panel/internal/models"
"gorm.io/gorm/schema"
)
func Migrate() error {
@@ -17,6 +14,7 @@ func Migrate() error {
}
allModels := []interface{}{
&models.AppLog{},
&models.User{},
&models.Task{},
&models.TaskLog{},
@@ -37,68 +35,9 @@ func Migrate() error {
return err
}
// MySQL 的 TEXT 类型最大 64KB，LONGTEXT 最大 4GB
// 模型统一使用 type:text 保持跨数据库兼容，这里针对 MySQL 自动升级为 LONGTEXT
mysqlUpgradeTextColumns(allModels...)
return nil
}
// mysqlUpgradeTextColumns 反射扫描所有模型，将 gorm tag 中 type:text 的字段在 MySQL 上升级为 LONGTEXT
func mysqlUpgradeTextColumns(allModels ...interface{}) {
if DB.Dialector.Name() != "mysql" {
return
}
// 获取当前数据库名
var dbName string
DB.Raw("SELECT DATABASE()").Scan(&dbName)
ns := schema.NamingStrategy{}
for _, model := range allModels {
typ := reflect.TypeOf(model)
if typ.Kind() == reflect.Ptr {
typ = typ.Elem()
}
// 获取表名
tableName := ""
if tabler, ok := model.(interface{ TableName() string }); ok {
tableName = tabler.TableName()
} else {
continue
}
for i := 0; i < typ.NumField(); i++ {
field := typ.Field(i)
gormTag := field.Tag.Get("gorm")
if gormTag == "" || !hasGormTypeText(gormTag) {
continue
}
// 从 gorm tag 获取列名，没有则用 GORM 命名策略转换
columnName := parseGormColumn(gormTag)
if columnName == "" {
columnName = ns.ColumnName("", field.Name)
}
// 检查当前列类型，已经是 longtext 则跳过
var columnType string
DB.Raw("SELECT DATA_TYPE FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? AND COLUMN_NAME = ?",
dbName, tableName, columnName).Scan(&columnType)
if strings.EqualFold(columnType, "longtext") {
continue
}
sql := fmt.Sprintf("ALTER TABLE `%s` MODIFY COLUMN `%s` LONGTEXT", tableName, columnName)
if err := DB.Exec(sql).Error; err != nil {
logger.Debugf("[Database] MySQL 升级 %s.%s 为 LONGTEXT: %v", tableName, columnName, err)
}
}
}
}
// hasGormTypeText 检查 gorm tag 中是否包含 type:text
func hasGormTypeText(gormTag string) bool {
for _, part := range strings.Split(gormTag, ";") {
internal/eventbus/bus.go
+50
View File
@@ -0,0 +1,50 @@
package eventbus
import "sync"
// Event 统一定义的事件体内包含的数据
type Event struct {
Type string
Payload interface{}
}
// Handler 事件具体的执行句柄
type Handler func(event Event)
// Subscriber 事件订阅者接口，各业务 Service 若关注系统总线可实现此接口
type Subscriber interface {
SubscribeEvents(bus *EventBus)
}
type EventBus struct {
handlers map[string][]Handler
mu sync.RWMutex
}
func New() *EventBus {
return &EventBus{
handlers: make(map[string][]Handler),
}
}
// Subscribe 注册订阅事件
func (bus *EventBus) Subscribe(eventType string, handler Handler) {
bus.mu.Lock()
defer bus.mu.Unlock()
bus.handlers[eventType] = append(bus.handlers[eventType], handler)
}
// Publish 异步抛出事件
func (bus *EventBus) Publish(event Event) {
bus.mu.RLock()
handlers := bus.handlers[event.Type]
bus.mu.RUnlock()
for _, handler := range handlers {
// 采用 Goroutine 异步不阻塞核心主线
go handler(event)
}
}
// 全局唯一的事件总线实例
var DefaultBus = New()
internal/models/app_log.go
+23
View File
@@ -0,0 +1,23 @@
package models
import (
"github.com/engigu/baihu-panel/internal/constant"
)
// AppLog 统一应用日志与通知记录
type AppLog struct {
ID string `json:"id" gorm:"primaryKey;size:20"`
Category string `json:"category" gorm:"size:50;index;not null"` // 大类：constant.LogCategorySystemNotice(系统通知), constant.LogCategoryPushLog(推送记录) 等
Title string `json:"title" gorm:"size:255"` // 消息标题
Content BigText `json:"content"` // 详细内容/Payload
Level string `json:"level" gorm:"size:20;index"` // 级别：constant.LogLevelInfo, constant.LogLevelWarning, constant.LogLevelError
Status string `json:"status" gorm:"size:20;index"` // 状态：系统通知为 constant.LogStatusRead/constant.LogStatusUnread，推送为 constant.LogStatusSuccess/constant.LogStatusFailed
RefID string `json:"ref_id" gorm:"size:50;index"` // 关联对象ID（选填，比如绑定的通知渠道ID、任务ID等）
ErrorMsg BigText `json:"error_msg"` // 执行错误信息详情
CreatedAt LocalTime `json:"created_at" gorm:"index"`
ReadAt *LocalTime `json:"read_at"` // 已读时间（仅对通知生效）
}
func (AppLog) TableName() string {
return constant.TablePrefix + "app_logs"
}
internal/models/big_text.go
+24
View File
@@ -0,0 +1,24 @@
package models
import (
"gorm.io/gorm"
"gorm.io/gorm/schema"
)
// BigText 自定义大数据文本类型，自动处理跨数据库类型差异
// MySQL: LONGTEXT (4GB)
// PostgreSQL: TEXT
// SQLite: TEXT
type BigText string
func (BigText) GormDBDataType(db *gorm.DB, field *schema.Field) string {
switch db.Dialector.Name() {
case "mysql":
return "LONGTEXT"
case "postgres":
return "TEXT"
case "sqlite":
return "TEXT"
}
return "TEXT"
}
internal/models/dependency.go
+1 -1
View File
@@ -14,7 +14,7 @@ type Dependency struct {
Language string `json:"language" gorm:"size:100;index"` // 关联语言 (node, python...)
LangVersion string `json:"lang_version" gorm:"size:100;index"` // 关联语言版本
Remark string `json:"remark" gorm:"size:255"`
Log string `json:"log" gorm:"type:text"`
Log BigText `json:"log"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
internal/models/env.go
+2 -2
View File
@@ -10,7 +10,7 @@ import (
type EnvironmentVariable struct {
ID string `json:"id" gorm:"primaryKey;size:20"`
Name string `json:"name" gorm:"size:255;not null"`
Value string `json:"value" gorm:"type:text"`
Value BigText `json:"value"`
Remark string `json:"remark" gorm:"size:500"`
Hidden bool `json:"hidden" gorm:"default:true"`
UserID string `json:"user_id" gorm:"size:20;index"`
@@ -27,7 +27,7 @@ func (EnvironmentVariable) TableName() string {
type Script struct {
ID string `json:"id" gorm:"primaryKey;size:20"`
Name string `json:"name" gorm:"size:255;not null"`
Content string `json:"content" gorm:"type:text"`
Content BigText `json:"content"`
UserID string `json:"user_id" gorm:"size:20;index"`
CreatedAt LocalTime `json:"created_at"`
UpdatedAt LocalTime `json:"updated_at"`
internal/models/notify_way.go
+1 -1
View File
@@ -11,7 +11,7 @@ type NotifyWay struct {
ID string `json:"id" gorm:"primaryKey;size:20"`
Name string `json:"name" gorm:"size:100;not null"`
Type string `json:"type" gorm:"size:50;not null;index"`
Config string `json:"config" gorm:"type:text"`
Config BigText `json:"config"`
Enabled bool `json:"enabled" gorm:"default:true;index"`
CreatedAt LocalTime `json:"created_at"`
UpdatedAt LocalTime `json:"updated_at"`
internal/models/setting.go
+1 -1
View File
@@ -9,7 +9,7 @@ type Setting struct {
ID string `json:"id" gorm:"primaryKey;size:20"`
Section string `json:"section" gorm:"size:50;not null;index:idx_section_key"`
Key string `json:"key" gorm:"size:100;not null;index:idx_section_key"`
Value string `json:"value" gorm:"type:text"`
Value BigText `json:"value"`
}
func (Setting) TableName() string {
internal/models/task.go
+10 -10
View File
@@ -35,23 +35,23 @@ type TaskConfig struct {
type Task struct {
ID string `json:"id" gorm:"primaryKey;size:20"`
Name string `json:"name" gorm:"size:255;not null"`
Command string `json:"command" gorm:"type:text"` // 普通任务的命令
Command BigText `json:"command"` // 普通任务的命令
Tags string `json:"tags" gorm:"size:255;default:''"` // 标签，逗号分隔
Type string `json:"type" gorm:"size:20;default:'task'"` // 任务类型: constant.TaskTypeNormal, constant.TaskTypeRepo
TriggerType string `json:"trigger_type" gorm:"size:25;default:'cron'"` // 触发类型: constant.TriggerTypeCron, constant.TriggerTypeBaihuStartup
Config string `json:"config" gorm:"type:text"` // 配置 JSON（仓库同步配置等）
Config BigText `json:"config"` // 配置 JSON（仓库同步配置等）
Schedule string `json:"schedule" gorm:"size:100"` // cron 表达式
Timeout int `json:"timeout" gorm:"default:30"` // 超时时间（分钟），默认30分钟
WorkDir string `json:"work_dir" gorm:"size:255;default:''"` // 工作目录，为空则使用 scripts 目录
CleanConfig string `json:"clean_config" gorm:"size:255;default:''"` // 清理配置 JSON
Envs string `json:"envs" gorm:"type:text"` // 环境变量ID列表，逗号分隔
Languages []map[string]string `json:"languages" gorm:"serializer:json;type:text"` // 针对本地任务的语言配置列表
Envs BigText `json:"envs"` // 环境变量ID列表，逗号分隔
Languages []map[string]string `json:"languages" gorm:"serializer:json"` // 针对本地任务的语言配置列表
AgentID *string `json:"agent_id" gorm:"size:20;index"` // Agent ID，为空表示本地执行
RetryCount int `json:"retry_count" gorm:"default:0"` // 失败重试次数
RetryInterval int `json:"retry_interval" gorm:"default:0"` // 失败重试间隔(秒)
RandomRange int `json:"random_range" gorm:"default:0"` // 随机延迟范围(秒)
Enabled bool `json:"enabled" gorm:"default:true"`
RunningGo string `json:"running_go" gorm:"type:text"` // 正在运行的 go routine id 数组 (JSON)
RunningGo BigText `json:"running_go"` // 正在运行的 go routine id 数组 (JSON)
RuntimeEnvs []string `json:"-" gorm:"-"` // 运行时环境变量（非持久化）
LastRun *LocalTime `json:"last_run"`
NextRun *LocalTime `json:"next_run"`
@@ -73,7 +73,7 @@ func (t *Task) GetName() string {
}
func (t *Task) GetCommand() string {
return t.Command
return string(t.Command)
}
func (t *Task) GetTimeout() int {
@@ -85,7 +85,7 @@ func (t *Task) GetWorkDir() string {
}
func (t *Task) GetEnvs() string {
return t.Envs
return string(t.Envs)
}
func (t *Task) GetLanguages() []map[string]string {
@@ -117,9 +117,9 @@ type TaskLog struct {
ID string `json:"id" gorm:"primaryKey;size:20"`
TaskID string `json:"task_id" gorm:"size:20;index"`
AgentID *string `json:"agent_id" gorm:"size:20;index"` // Agent ID，为空表示本地执行
Command string `json:"command" gorm:"type:text"`
Output string `json:"-" gorm:"type:text"` // gzip+base64 压缩后的日志
Error string `json:"error" gorm:"type:text"` // 额外的系统错误信息
Command BigText `json:"command"`
Output BigText `json:"-"` // gzip+base64 压缩后的日志
Error BigText `json:"error"` // 额外的系统错误信息
Status string `json:"status" gorm:"size:20;index"` // success, failed
Duration int64 `json:"duration"` // 执行耗时（毫秒）
ExitCode int `json:"exit_code"`
internal/models/vo/dep_vo.go
+1 -1
View File
@@ -31,7 +31,7 @@ func ToDependencyVO(dep *models.Dependency) *DependencyVO {
Language: dep.Language,
LangVersion: dep.LangVersion,
Remark: dep.Remark,
Log: dep.Log,
Log: string(dep.Log),
CreatedAt: dep.CreatedAt,
UpdatedAt: dep.UpdatedAt,
}
internal/models/vo/script_vo.go
+1 -1
View File
@@ -21,7 +21,7 @@ func ToScriptVO(script *models.Script) *ScriptVO {
return &ScriptVO{
ID: script.ID,
Name: script.Name,
Content: script.Content,
Content: string(script.Content),
CreatedAt: script.CreatedAt,
UpdatedAt: script.UpdatedAt,
}
internal/models/vo/sys_vo.go
+1 -1
View File
@@ -48,7 +48,7 @@ func ToEnvVO(env *models.EnvironmentVariable) *EnvVO {
return &EnvVO{
ID: env.ID,
Name: env.Name,
Value: env.Value,
Value: string(env.Value),
Remark: env.Remark,
Hidden: env.Hidden,
CreatedAt: env.CreatedAt,
internal/models/vo/task_vo.go
+6 -6
View File
@@ -39,16 +39,16 @@ func ToTaskVO(task *models.Task) *TaskVO {
return &TaskVO{
ID: task.ID,
Name: task.Name,
Command: task.Command,
Command: string(task.Command),
Tags: task.Tags,
Type: task.Type,
TriggerType: task.TriggerType,
Config: task.Config,
Config: string(task.Config),
Schedule: task.Schedule,
Timeout: task.Timeout,
WorkDir: task.WorkDir,
CleanConfig: task.CleanConfig,
Envs: task.Envs,
Envs: string(task.Envs),
Languages: task.Languages,
AgentID: task.AgentID,
Enabled: task.Enabled,
@@ -112,15 +112,15 @@ func ToTaskLogVO(log *models.TaskLog) *TaskLogVO {
ID: log.ID,
TaskID: log.TaskID,
AgentID: log.AgentID,
Command: log.Command,
Error: log.Error,
Command: string(log.Command),
Error: string(log.Error),
Status: log.Status,
Duration: log.Duration,
ExitCode: log.ExitCode,
StartTime: log.StartTime,
EndTime: log.EndTime,
CreatedAt: log.CreatedAt,
Output: log.Output,
Output: string(log.Output),
}
}
internal/router/events.go
+33
View File
@@ -0,0 +1,33 @@
package router
import (
// "fmt"
"time"
// "github.com/engigu/baihu-panel/internal/constant"
"github.com/engigu/baihu-panel/internal/eventbus"
// "github.com/engigu/baihu-panel/internal/logger"
// "github.com/engigu/baihu-panel/internal/models"
"github.com/engigu/baihu-panel/internal/services"
)
func setupEventHandlers(subscribers ...eventbus.Subscriber) {
bus := eventbus.DefaultBus
// 遍历并统一初始化所有订阅者的事件链路
for _, s := range subscribers {
s.SubscribeEvents(bus)
}
}
func startAppLogCleanup(appLogSvc *services.AppLogService) {
// 初始化时执行一次清理
appLogSvc.CleanUp()
// 每天凌晨或者定期清理
ticker := time.NewTicker(24 * time.Hour)
for range ticker.C {
appLogSvc.CleanUp()
}
}
internal/router/register.go
+8 -1
View File
@@ -26,17 +26,23 @@ func RegisterControllers() *Controllers {
taskLogService := tasks.NewTaskLogService(sendStatsService)
// 创建任务执行服务（需要依赖注入）
notifyService := services.NewNotificationService()
appLogService := services.NewAppLogService()
// 清理 task 运行状态的任务可以直接由 executorService 承担或在此处通过 Database 直接清理
// 简单期间，我们使用一个新方法 tasks.CleanupRunningTasks() 或者让 executorService 启动时清理
executorService = tasks.NewExecutorService(taskService, taskLogService, agentWSManager, settingsService, envService, services.NewNotificationService())
executorService = tasks.NewExecutorService(taskService, taskLogService, agentWSManager, settingsService, envService)
// 启动时清理残留的运行状态
_ = executorService.CleanupRunningTasks()
// 启动计划任务
executorService.StartCron()
// 初始化所有关注系统总线的服务
setupEventHandlers(appLogService, notifyService, loginLogService)
go startAppLogCleanup(appLogService)
// 初始化并返回控制器
return &Controllers{
Task: controllers.NewTaskController(taskService, executorService),
@@ -54,6 +60,7 @@ func RegisterControllers() *Controllers {
Agent: controllers.NewAgentController(settingsService),
Mise: controllers.NewMiseController(services.NewMiseService()),
Notification: controllers.NewNotificationController(),
AppLog: controllers.NewAppLogController(),
}
}
internal/router/router.go
+11
View File
@@ -33,6 +33,7 @@ type Controllers struct {
Agent *controllers.AgentController
Mise *controllers.MiseController
Notification *controllers.NotificationController
AppLog *controllers.AppLogController
}
func mustSubFS(fsys fs.FS, dir string) fs.FS {
@@ -249,6 +250,7 @@ func initAuthorizedAPIRoutes(api *gin.RouterGroup, c *Controllers) {
registerAgentRoutes(authorized, c)
registerMiseRoutes(authorized, c)
registerNotificationRoutes(authorized, c)
registerAppLogRoutes(authorized, c)
}
// 通知发送 API（使用通知 Token 认证，供脚本调用）
@@ -421,6 +423,15 @@ func registerNotificationRoutes(g *gin.RouterGroup, c *Controllers) {
}
}
func registerAppLogRoutes(g *gin.RouterGroup, c *Controllers) {
appLogs := g.Group("/app-logs")
{
appLogs.GET("", c.AppLog.GetLogs)
appLogs.POST("/read", c.AppLog.MarkAsRead)
appLogs.POST("/clear", c.AppLog.ClearLogs)
}
}
func initAgentAPIRoutes(root *gin.RouterGroup, c *Controllers) {
// Agent API（供远程 Agent 调用，不使用 /v1 版本号）
agentAPI := root.Group("/api/agent")
internal/services/agent_service.go
+3 -3
View File
@@ -331,8 +331,8 @@ func (s *AgentService) GetTasks(agentID string) []models.AgentTask {
if allEnvs {
envVars = envService.GetAllEnvVars()
} else if task.Envs != "" {
envVars = envService.GetEnvVarsByIDs(task.Envs)
} else if string(task.Envs) != "" {
envVars = envService.GetEnvVarsByIDs(string(task.Envs))
}
envVarsStr := executor.FormatEnvVars(envVars)
@@ -340,7 +340,7 @@ func (s *AgentService) GetTasks(agentID string) []models.AgentTask {
result[i] = models.AgentTask{
ID: task.ID,
Name: task.Name,
Command: task.Command,
Command: string(task.Command),
Schedule: task.Schedule,
Timeout: task.Timeout,
WorkDir: task.WorkDir,
internal/services/app_log_service.go
+239
View File
@@ -0,0 +1,239 @@
package services
import (
"encoding/json"
"time"
"fmt"
"github.com/engigu/baihu-panel/internal/constant"
"github.com/engigu/baihu-panel/internal/database"
"github.com/engigu/baihu-panel/internal/eventbus"
"github.com/engigu/baihu-panel/internal/logger"
"github.com/engigu/baihu-panel/internal/models"
"github.com/engigu/baihu-panel/internal/utils"
)
type LogRetentionConfig struct {
Days int `json:"days"`
MaxCount int `json:"max_count"`
}
type AppLogService struct {
settingsService *SettingsService
}
func NewAppLogService() *AppLogService {
return &AppLogService{
settingsService: NewSettingsService(),
}
}
func (s *AppLogService) Add(log *models.AppLog) error {
if log.ID == "" {
log.ID = utils.GenerateID()
}
return database.DB.Create(log).Error
}
func (s *AppLogService) List(category string, status string, level string, page, pageSize int, keyword string) ([]models.AppLog, int64, error) {
var logs []models.AppLog
var total int64
query := database.DB.Model(&models.AppLog{})
if category != "" {
query = query.Where("category = ?", category)
}
if status != "" {
query = query.Where("status = ?", status)
}
if level != "" {
query = query.Where("level = ?", level)
}
if keyword != "" {
query = query.Where("(title LIKE ? OR content LIKE ?)", "%"+keyword+"%", "%"+keyword+"%")
}
query.Count(&total)
offset := (page - 1) * pageSize
err := query.Order("created_at desc").Offset(offset).Limit(pageSize).Find(&logs).Error
return logs, total, err
}
func (s *AppLogService) MarkAsRead(id string) error {
now := models.LocalTime(time.Now())
return database.DB.Model(&models.AppLog{}).Where("id = ?", id).Updates(map[string]interface{}{
"status": constant.LogStatusRead,
"read_at": &now,
}).Error
}
func (s *AppLogService) MarkAllAsRead(category string) error {
now := models.LocalTime(time.Now())
return database.DB.Model(&models.AppLog{}).Where("category = ? AND status = ?", category, constant.LogStatusUnread).Updates(map[string]interface{}{
"status": constant.LogStatusRead,
"read_at": &now,
}).Error
}
func (s *AppLogService) Clear(category string) error {
query := database.DB.Model(&models.AppLog{})
if category != "" {
query = query.Where("category = ?", category)
}
return query.Unscoped().Delete(&models.AppLog{}).Error
}
func (s *AppLogService) GetRetentionConfigs() map[string]LogRetentionConfig {
val := s.settingsService.Get(constant.SectionSystem, "log_retention")
var configs map[string]LogRetentionConfig
if val != "" {
_ = json.Unmarshal([]byte(val), &configs)
}
if configs == nil {
configs = map[string]LogRetentionConfig{
constant.LogCategorySystemNotice: {Days: 30, MaxCount: 500},
constant.LogCategoryPushLog: {Days: 15, MaxCount: 5000},
constant.LogCategoryDefault: {Days: 30, MaxCount: 10000},
}
}
return configs
}
func (s *AppLogService) CleanUp() {
configs := s.GetRetentionConfigs()
categories := []string{constant.LogCategorySystemNotice, constant.LogCategoryPushLog}
for _, cat := range categories {
cfg, ok := configs[cat]
if !ok {
cfg = configs[constant.LogCategoryDefault]
}
if cfg.Days > 0 {
deadline := time.Now().AddDate(0, 0, -cfg.Days)
database.DB.Unscoped().Where("category = ? AND created_at < ?", cat, deadline).Delete(&models.AppLog{})
}
if cfg.MaxCount > 0 {
var total int64
database.DB.Model(&models.AppLog{}).Where("category = ?", cat).Count(&total)
if total > int64(cfg.MaxCount) {
deleteCount := total - int64(cfg.MaxCount)
var ids []string
database.DB.Model(&models.AppLog{}).Where("category = ?", cat).Order("created_at asc").Limit(int(deleteCount)).Pluck("id", &ids)
if len(ids) > 0 {
database.DB.Unscoped().Where("id IN ?", ids).Delete(&models.AppLog{})
}
}
}
}
logger.Debugf("[AppLog] 完成应用日志清理策略")
}
func (s *AppLogService) SubscribeEvents(bus *eventbus.EventBus) {
// 1. [订阅] 系统通知 -> 存储到数据库表现为红点消息
bus.Subscribe(constant.EventSystemNotice, func(e eventbus.Event) {
payload, ok := e.Payload.(map[string]interface{})
if !ok {
return
}
title, _ := payload["title"].(string)
content, _ := payload["content"].(string)
level, _ := payload["level"].(string)
if level == "" {
level = constant.LogLevelInfo
}
s.Add(&models.AppLog{
Category: constant.LogCategorySystemNotice,
Title: title,
Content: models.BigText(content),
Level: level,
Status: constant.LogStatusUnread,
})
})
// 2. [订阅] 推送结果 -> 存储到数据库供推送日志查看
bus.Subscribe(constant.EventNotifySent, func(e eventbus.Event) {
payload, ok := e.Payload.(map[string]interface{})
if !ok {
return
}
title, _ := payload["title"].(string)
content, _ := payload["content"].(string)
success, _ := payload["success"].(bool)
errorMsg, _ := payload["error_msg"].(string)
channelName, _ := payload["channel_name"].(string)
status := constant.LogStatusSuccess
level := constant.LogLevelInfo
if !success {
status = constant.LogStatusFailed
level = constant.LogLevelError
}
s.Add(&models.AppLog{
Category: constant.LogCategoryPushLog,
Title: fmt.Sprintf("[%s] %s", channelName, title),
Content: models.BigText(content),
Level: level,
Status: status,
ErrorMsg: models.BigText(errorMsg),
})
})
// 3. 将某些业务事件转化为系统内部通知 (自动出现在小铃铛)
bus.Subscribe(constant.EventTaskFailed, func(e eventbus.Event) {
payload, ok := e.Payload.(map[string]interface{})
if !ok {
return
}
taskName, _ := payload["task_name"].(string)
errMsg, _ := payload["error"].(string)
bus.Publish(eventbus.Event{
Type: constant.EventSystemNotice,
Payload: map[string]interface{}{
"title": fmt.Sprintf("任务 [%s] 执行失败", taskName),
"content": fmt.Sprintf("错误详情: %s", errMsg),
"level": constant.LogLevelError,
},
})
})
bus.Subscribe(constant.EventTaskTimeout, func(e eventbus.Event) {
payload, ok := e.Payload.(map[string]interface{})
if !ok {
return
}
taskName, _ := payload["task_name"].(string)
bus.Publish(eventbus.Event{
Type: constant.EventSystemNotice,
Payload: map[string]interface{}{
"title": fmt.Sprintf("任务 [%s] 执行超时", taskName),
"content": "任务已经超过预设的运行时间并被系统强制中止。",
"level": constant.LogLevelWarning,
},
})
})
bus.Subscribe(constant.EventPasswordChanged, func(e eventbus.Event) {
payload, ok := e.Payload.(map[string]interface{})
if !ok {
return
}
username, _ := payload["username"].(string)
bus.Publish(eventbus.Event{
Type: constant.EventSystemNotice,
Payload: map[string]interface{}{
"title": "账户安全提醒",
"content": fmt.Sprintf("用户 %s 的账号密码已被修改。", username),
"level": constant.LogLevelInfo,
},
})
})
}
internal/services/backup_service.go
+5 -1
View File
@@ -57,6 +57,7 @@ func (s *BackupService) getTableConfigs() []tableConfig {
{"deps.json", s.exportTable(&[]models.Dependency{}, true), s.restoreTable(&[]models.Dependency{}, true)},
{"notify_ways.json", s.exportTable(&[]models.NotifyWay{}, true), s.restoreTable(&[]models.NotifyWay{}, true)},
{"notify_bindings.json", s.exportTable(&[]models.NotifyBinding{}, true), s.restoreTable(&[]models.NotifyBinding{}, true)},
{"app_logs.json", s.exportTable(&[]models.AppLog{}, false), s.restoreTable(&[]models.AppLog{}, false)},
}
}
@@ -235,6 +236,7 @@ func (s *BackupService) Restore(zipPath string) error {
tx.Unscoped().Where("1=1").Delete(&models.Dependency{})
tx.Unscoped().Where("1=1").Delete(&models.NotifyWay{})
tx.Unscoped().Where("1=1").Delete(&models.NotifyBinding{})
tx.Unscoped().Where("1=1").Delete(&models.AppLog{})
// 2. 依次恢复每个表
for _, cfg := range configs {
@@ -340,6 +342,8 @@ func (s *BackupService) restoreFromZipFile(tx *gorm.DB, f *zip.File, filename st
return restoreStreamBatch[models.NotifyWay](tx, decoder)
case "notify_bindings.json":
return restoreStreamBatch[models.NotifyBinding](tx, decoder)
case "app_logs.json":
return restoreStreamBatch[models.AppLog](tx, decoder)
default:
return nil
}
@@ -417,7 +421,7 @@ func (s *BackupService) GetBackupFile() string {
if err := database.DB.Where("section = ? AND `key` = ?", BackupSection, BackupFileKey).First(&setting).Error; err != nil {
return ""
}
return setting.Value
return string(setting.Value)
}
func (s *BackupService) ClearBackup() error {
internal/services/deps/manager.go
+1 -1
View File
@@ -49,7 +49,7 @@ func (m *BaseManager) Install(dep *models.Dependency) error {
logger.Infof("Installing %s package: %s", m.Language, packageSpec)
output, err := m.runMiseCommand(dep.LangVersion, args)
dep.Log = string(output)
dep.Log = models.BigText(output)
if err != nil {
logger.Errorf("Install failed: %v, output: %s", err, string(output))
internal/services/env_service.go
+5 -5
View File
@@ -20,7 +20,7 @@ func (es *EnvService) CreateEnvVar(name, value, remark string, hidden bool, user
env := &models.EnvironmentVariable{
ID: utils.GenerateID(),
Name: name,
Value: value,
Value: models.BigText(value),
Remark: remark,
Hidden: hidden,
UserID: userID,
@@ -72,7 +72,7 @@ func (es *EnvService) UpdateEnvVar(id string, name, value, remark string, hidden
}
updates := map[string]interface{}{
"name": name,
"value": value,
"value": models.BigText(value),
"remark": remark,
"hidden": hidden,
}
@@ -98,7 +98,7 @@ func (es *EnvService) DeleteEnvVar(id string, force bool) (bool, []models.Task)
err := database.DB.Transaction(func(tx *gorm.DB) error {
// Update tasks to remove this env ID
for _, task := range associatedTasks {
ids := splitEnvIDs(task.Envs)
ids := splitEnvIDs(string(task.Envs))
var newIDs []string
for _, eid := range ids {
if eid != id {
@@ -166,12 +166,12 @@ func (es *EnvService) formatEnvVars(envs []models.EnvironmentVariable) []string
for _, env := range envs {
if idx, ok := nameToIndex[env.Name]; ok {
mergedList[idx].values = append(mergedList[idx].values, env.Value)
mergedList[idx].values = append(mergedList[idx].values, string(env.Value))
} else {
nameToIndex[env.Name] = len(mergedList)
mergedList = append(mergedList, mergedEnv{
name: env.Name,
values: []string{env.Value},
values: []string{string(env.Value)},
})
}
}
internal/services/login_log_service.go
+58
View File
@@ -1,7 +1,10 @@
package services
import (
"fmt"
"github.com/engigu/baihu-panel/internal/constant"
"github.com/engigu/baihu-panel/internal/database"
"github.com/engigu/baihu-panel/internal/eventbus"
"github.com/engigu/baihu-panel/internal/models"
"github.com/engigu/baihu-panel/internal/utils"
)
@@ -25,6 +28,61 @@ func (s *LoginLogService) Create(username, ip, userAgent, status, message string
return database.DB.Create(log).Error
}
// SubscribeEvents 注册订阅事件
func (s *LoginLogService) SubscribeEvents(bus *eventbus.EventBus) {
// 用户登录事件
bus.Subscribe(constant.EventUserLogin, func(e eventbus.Event) {
payload, ok := e.Payload.(map[string]interface{})
if !ok {
return
}
username, _ := payload["username"].(string)
ip, _ := payload["ip"].(string)
userAgent, _ := payload["userAgent"].(string)
status, _ := payload["status"].(string)
message, _ := payload["message"].(string)
s.Create(username, ip, userAgent, status, message)
// 如果登录成功，触发系统通知
if status == "success" {
bus.Publish(eventbus.Event{
Type: constant.EventSystemNotice,
Payload: map[string]interface{}{
"title": "登录提醒",
"content": fmt.Sprintf("用户 %s 已从 IP %s 登录系统", username, ip),
"level": constant.LogLevelWarning,
},
})
}
})
// 暴力破解防御触发事件
bus.Subscribe(constant.EventBruteForceLogin, func(e eventbus.Event) {
payload, ok := e.Payload.(map[string]interface{})
if !ok {
return
}
username, _ := payload["username"].(string)
ip, _ := payload["ip"].(string)
userAgent, _ := payload["userAgent"].(string)
s.Create(username, ip, userAgent, "failed", "尝试次数过多，由于暴力破解防御机制已锁定")
// 触发系统通知
bus.Publish(eventbus.Event{
Type: constant.EventSystemNotice,
Payload: map[string]interface{}{
"title": "系统安全警告",
"content": fmt.Sprintf("检测到 IP %s 正在尝试暴力破解用户 %s", ip, username),
"level": constant.LogLevelError,
},
})
})
}
// List 获取登录日志列表
func (s *LoginLogService) List(page, pageSize int, username string) ([]models.LoginLog, int64, error) {
var logs []models.LoginLog
internal/services/notification_service.go
+110 -53
View File
@@ -7,6 +7,7 @@ import (
"github.com/engigu/baihu-panel/internal/constant"
"github.com/engigu/baihu-panel/internal/database"
"github.com/engigu/baihu-panel/internal/eventbus"
"github.com/engigu/baihu-panel/internal/logger"
"github.com/engigu/baihu-panel/internal/models"
"github.com/engigu/baihu-panel/internal/utils"
@@ -98,7 +99,7 @@ func (s *NotificationService) SaveChannel(channel NotifyChannel) error {
ID: channel.ID,
Name: channel.Name,
Type: channel.Type,
Config: string(configJSON),
Config: models.BigText(configJSON),
Enabled: channel.Enabled,
}
return database.DB.Create(notifyWay).Error
@@ -108,7 +109,7 @@ func (s *NotificationService) SaveChannel(channel NotifyChannel) error {
updates := map[string]interface{}{
"name": channel.Name,
"type": channel.Type,
"config": string(configJSON),
"config": models.BigText(configJSON),
"enabled": channel.Enabled,
}
return database.DB.Model(&models.NotifyWay{}).Where("id = ?", channel.ID).Updates(updates).Error
@@ -202,12 +203,39 @@ func (s *NotificationService) SendToChannel(channel NotifyChannel, msg *NotifyMe
Title: msg.Title,
Text: msg.Text,
})
payload := map[string]interface{}{
"title": msg.Title,
"content": msg.Text,
"channel_name": channel.Name,
"success": false,
"error_msg": "",
}
if err != nil {
payload["error_msg"] = err.Error()
eventbus.DefaultBus.Publish(eventbus.Event{
Type: constant.EventNotifySent,
Payload: payload,
})
return &NotifyResult{Success: false, Error: err.Error()}
}
if !result.Success {
payload["error_msg"] = result.Error
eventbus.DefaultBus.Publish(eventbus.Event{
Type: constant.EventNotifySent,
Payload: payload,
})
return &NotifyResult{Success: false, Error: result.Error}
}
payload["success"] = true
eventbus.DefaultBus.Publish(eventbus.Event{
Type: constant.EventNotifySent,
Payload: payload,
})
return &NotifyResult{Success: true}
}
@@ -240,62 +268,91 @@ func (s *NotificationService) SendByChannelID(channelID string, msg *NotifyMessa
return s.SendToChannel(ch, msg)
}
// TriggerEvent 触发事件通知（实现 tasks.Notifier 接口）
func (s *NotificationService) TriggerEvent(bindingType string, eventType string, dataID string, templateData map[string]interface{}) {
var title, text string
// SubscribeEvents 注册通知服务自身为事件流的订阅者
func (s *NotificationService) SubscribeEvents(bus *eventbus.EventBus) {
// 系统事件
systemEvents := []string{constant.EventUserLogin, constant.EventBruteForceLogin, constant.EventPasswordChanged}
for _, evt := range systemEvents {
bus.Subscribe(evt, s.handleEvent(constant.BindingTypeSystem))
}
switch eventType {
case constant.EventUserLogin:
title = "用户登录通知"
text = fmt.Sprintf("用户 %v 在 IP %v 登录成功", templateData["username"], templateData["ip"])
case constant.EventBruteForceLogin:
title = "系统安全警告"
text = fmt.Sprintf("检测到 IP %v 正在尝试暴力破解用户 %v", templateData["ip"], templateData["username"])
case constant.EventPasswordChanged:
title = "账户安全通知"
text = fmt.Sprintf("用户 %v 刚刚修改了密码", templateData["username"])
case constant.EventTaskSuccess:
title = fmt.Sprintf("任务[%v] 成功", templateData["task_name"])
text = fmt.Sprintf("任务 #%v %v\n状态: 成功\n耗时: %vms", templateData["task_id"], templateData["task_name"], templateData["duration"])
case constant.EventTaskFailed:
title = fmt.Sprintf("任务[%v] 失败", templateData["task_name"])
if errStr, ok := templateData["error"]; ok {
text = fmt.Sprintf("任务 #%v %v\n执行失败\n错误: %v", templateData["task_id"], templateData["task_name"], errStr)
} else {
text = fmt.Sprintf("任务 #%v %v\n执行失败\n状态: %v\n耗时: %vms", templateData["task_id"], templateData["task_name"], templateData["status"], templateData["duration"])
// 任务事件
taskEvents := []string{constant.EventTaskSuccess, constant.EventTaskFailed, constant.EventTaskTimeout}
for _, evt := range taskEvents {
bus.Subscribe(evt, s.handleEvent(constant.BindingTypeTask))
}
// 通用系统通知
bus.Subscribe(constant.EventSystemNotice, s.handleEvent(constant.BindingTypeSystem))
}
func (s *NotificationService) handleEvent(bindingType string) eventbus.Handler {
return func(e eventbus.Event) {
payload, ok := e.Payload.(map[string]interface{})
if !ok {
return
}
case constant.EventTaskTimeout:
title = fmt.Sprintf("任务[%v] 超时", templateData["task_name"])
text = fmt.Sprintf("任务 #%v %v\n执行超时\n耗时: %vms", templateData["task_id"], templateData["task_name"], templateData["duration"])
default:
title = "系统通知"
text = "收到未知事件"
}
msg := &NotifyMessage{Title: title, Text: text}
bindings := s.GetBindingsByEvent(bindingType, eventType, dataID)
if len(bindings) == 0 {
return
}
channels := s.GetChannels()
channelMap := make(map[string]NotifyChannel)
for _, ch := range channels {
channelMap[ch.ID] = ch
}
for _, binding := range bindings {
ch, ok := channelMap[binding.WayID]
if !ok || !ch.Enabled {
continue
var dataID string
if id, ok := payload["task_id"].(string); ok {
dataID = id
}
go func(channel NotifyChannel) {
result := s.SendToChannel(channel, msg)
if !result.Success {
logger.Warnf("[Notify] 发送事件 %s 到渠道 %s(%s) 失败: %s", eventType, channel.Name, channel.Type, result.Error)
var title, text string
switch e.Type {
case constant.EventUserLogin:
title = "用户登录通知"
text = fmt.Sprintf("用户 %v 在 IP %v 登录成功", payload["username"], payload["ip"])
case constant.EventBruteForceLogin:
title = "系统安全警告"
text = fmt.Sprintf("检测到 IP %v 正在尝试暴力破解用户 %v", payload["ip"], payload["username"])
case constant.EventPasswordChanged:
title = "账户安全通知"
text = fmt.Sprintf("用户 %v 刚刚修改了密码", payload["username"])
case constant.EventTaskSuccess:
title = fmt.Sprintf("任务[%v] 成功", payload["task_name"])
text = fmt.Sprintf("任务 #%v %v\n状态: 成功\n耗时: %vms", payload["task_id"], payload["task_name"], payload["duration"])
case constant.EventTaskFailed:
title = fmt.Sprintf("任务[%v] 失败", payload["task_name"])
if errStr, ok := payload["error"]; ok {
text = fmt.Sprintf("任务 #%v %v\n执行失败\n错误: %v", payload["task_id"], payload["task_name"], errStr)
} else {
text = fmt.Sprintf("任务 #%v %v\n执行失败\n状态: %v\n耗时: %vms", payload["task_id"], payload["task_name"], payload["status"], payload["duration"])
}
}(ch)
case constant.EventTaskTimeout:
title = fmt.Sprintf("任务[%v] 超时", payload["task_name"])
text = fmt.Sprintf("任务 #%v %v\n执行超时\n耗时: %vms", payload["task_id"], payload["task_name"], payload["duration"])
case constant.EventSystemNotice:
title, _ = payload["title"].(string)
text, _ = payload["content"].(string)
default:
return
}
msg := &NotifyMessage{Title: title, Text: text}
bindings := s.GetBindingsByEvent(bindingType, e.Type, dataID)
if len(bindings) == 0 {
return
}
channels := s.GetChannels()
channelMap := make(map[string]NotifyChannel)
for _, ch := range channels {
channelMap[ch.ID] = ch
}
for _, binding := range bindings {
ch, ok := channelMap[binding.WayID]
if !ok || !ch.Enabled {
continue
}
go func(channel NotifyChannel) {
result := s.SendToChannel(channel, msg)
if !result.Success {
logger.Warnf("[Notify] 发送事件 %s 到渠道 %s(%s) 失败: %s", e.Type, channel.Name, channel.Type, result.Error)
}
}(ch)
}
}
}
internal/services/script_service.go
+2 -2
View File
@@ -16,7 +16,7 @@ func (ss *ScriptService) CreateScript(name, content string, userID string) *mode
script := &models.Script{
ID: utils.GenerateID(),
Name: name,
Content: content,
Content: models.BigText(content),
UserID: userID,
}
database.DB.Create(script)
@@ -43,7 +43,7 @@ func (ss *ScriptService) UpdateScript(id string, name, content string) *models.S
return nil
}
script.Name = name
script.Content = content
script.Content = models.BigText(content)
database.DB.Save(&script)
return &script
}
internal/services/settings_service.go
+6 -6
View File
@@ -25,7 +25,7 @@ func (s *SettingsService) InitSettings() error {
ID: utils.GenerateID(),
Section: section,
Key: key,
Value: value,
Value: models.BigText(value),
}).Error; err != nil {
return err
}
@@ -47,7 +47,7 @@ func (s *SettingsService) InitSettings() error {
ID: utils.GenerateID(),
Section: constant.SectionSecurity,
Key: constant.KeySecret,
Value: secretValue,
Value: models.BigText(secretValue),
}).Error; err != nil {
return err
}
@@ -72,7 +72,7 @@ func (s *SettingsService) Get(section, key string) string {
}
return ""
}
return setting.Value
return string(setting.Value)
}
// Set 设置单个值
@@ -83,10 +83,10 @@ func (s *SettingsService) Set(section, key, value string) error {
ID: utils.GenerateID(),
Section: section,
Key: key,
Value: value,
Value: models.BigText(value),
}).Error
}
return database.DB.Model(&setting).Update("value", value).Error
return database.DB.Model(&setting).Update("value", models.BigText(value)).Error
}
// Delete 删除单个设置
@@ -108,7 +108,7 @@ func (s *SettingsService) GetSection(section string) map[string]string {
var settings []models.Setting
database.DB.Where("section = ?", section).Find(&settings)
for _, setting := range settings {
result[setting.Key] = setting.Value
result[setting.Key] = string(setting.Value)
}
return result
}
internal/services/tasks/executor_service.go
+52 -56
View File
@@ -13,6 +13,7 @@ import (
"github.com/engigu/baihu-panel/internal/constant"
"github.com/engigu/baihu-panel/internal/database"
"github.com/engigu/baihu-panel/internal/eventbus"
"github.com/engigu/baihu-panel/internal/executor"
"github.com/engigu/baihu-panel/internal/logger"
"github.com/engigu/baihu-panel/internal/models"
@@ -39,19 +40,12 @@ type EnvService interface {
GetAllEnvVars() []string
}
// Notifier 通知服务接口定义（避免循环依赖）
type Notifier interface {
TriggerEvent(bindingType string, eventType string, dataID string, templateData map[string]interface{})
}
// ExecutorService handles task execution and scheduling
type ExecutorService struct {
taskService *TaskService
taskLogService *TaskLogService
agentWSManager AgentWSManager
settingsService SettingsService
envService EnvService
notifier Notifier
scheduler *executor.Scheduler
cronManager *executor.CronManager
results []executor.ExecutionResult
@@ -64,14 +58,12 @@ func (es *ExecutorService) GetScheduler() *executor.Scheduler {
return es.scheduler
}
// NewExecutorService creates a new executor service
func NewExecutorService(
taskService *TaskService,
taskLogService *TaskLogService,
agentWSManager AgentWSManager,
settingsService SettingsService,
envService EnvService,
notifier Notifier,
) *ExecutorService {
es := &ExecutorService{
taskService: taskService,
@@ -79,7 +71,6 @@ func NewExecutorService(
agentWSManager: agentWSManager,
settingsService: settingsService,
envService: envService,
notifier: notifier,
results: make([]executor.ExecutionResult, 0, 100),
stopCh: make(chan struct{}),
}
@@ -143,7 +134,8 @@ func (h *ServerSchedulerHandler) OnTaskExecuting(req *executor.ExecutionRequest)
if err != nil {
// 并发限制，更新日志状态为失败
taskLog.Status = constant.TaskStatusFailed
taskLog.Output, _ = utils.CompressToBase64("任务并发数限制，拒绝执行")
comp, _ := utils.CompressToBase64("任务并发数限制，拒绝执行")
taskLog.Output = models.BigText(comp)
h.es.taskLogService.SaveTaskLog(taskLog)
return nil, nil, fmt.Errorf("任务并发限制: %v", err)
}
@@ -225,9 +217,9 @@ func (h *ServerSchedulerHandler) OnTaskCompleted(req *executor.ExecutionRequest,
taskLog := &models.TaskLog{
ID: req.LogID,
TaskID: task.ID,
Command: req.Command,
Output: output,
Error: result.Error,
Command: models.BigText(req.Command),
Output: models.BigText(output),
Error: models.BigText(result.Error),
Status: result.Status,
Duration: result.Duration,
ExitCode: result.ExitCode,
@@ -256,27 +248,29 @@ func (h *ServerSchedulerHandler) OnTaskCompleted(req *executor.ExecutionRequest,
h.es.HandleTaskRetry(task, req, result.Success, result.Status, result.ExitCode)
// ======= 通知触发 =======
if h.es.notifier != nil {
go func() {
var eventType string
switch result.Status {
case constant.TaskStatusSuccess:
eventType = constant.EventTaskSuccess
case constant.TaskStatusFailed:
eventType = constant.EventTaskFailed
case constant.TaskStatusTimeout:
eventType = constant.EventTaskTimeout
}
if eventType != "" {
h.es.notifier.TriggerEvent(constant.BindingTypeTask, eventType, task.ID, map[string]interface{}{
// ======= 通知触发 =======
go func() {
var eventType string
switch result.Status {
case constant.TaskStatusSuccess:
eventType = constant.EventTaskSuccess
case constant.TaskStatusFailed:
eventType = constant.EventTaskFailed
case constant.TaskStatusTimeout:
eventType = constant.EventTaskTimeout
}
if eventType != "" {
eventbus.DefaultBus.Publish(eventbus.Event{
Type: eventType,
Payload: map[string]interface{}{
"task_id": task.ID,
"task_name": task.Name,
"status": result.Status,
"duration": result.Duration,
})
}
}()
}
},
})
}
}()
}
func (h *ServerSchedulerHandler) OnTaskFailed(req *executor.ExecutionRequest, err error) {
@@ -305,9 +299,9 @@ func (h *ServerSchedulerHandler) OnTaskFailed(req *executor.ExecutionRequest, er
taskLog := &models.TaskLog{
ID: req.LogID,
TaskID: taskID,
Command: req.Command,
Output: output,
Error: err.Error(),
Command: models.BigText(req.Command),
Output: models.BigText(output),
Error: models.BigText(err.Error()),
Status: constant.TaskStatusFailed,
Duration: 0,
ExitCode: 1,
@@ -338,19 +332,21 @@ func (h *ServerSchedulerHandler) OnTaskFailed(req *executor.ExecutionRequest, er
h.es.HandleTaskRetry(task, req, false, constant.TaskStatusFailed, 1)
// ======= 通知触发 =======
if h.es.notifier != nil {
go func() {
taskName := "未知任务"
if task != nil {
taskName = task.Name
}
h.es.notifier.TriggerEvent(constant.BindingTypeTask, constant.EventTaskFailed, taskID, map[string]interface{}{
// ======= 通知触发 =======
go func() {
taskName := "未知任务"
if task != nil {
taskName = task.Name
}
eventbus.DefaultBus.Publish(eventbus.Event{
Type: constant.EventTaskFailed,
Payload: map[string]interface{}{
"task_id": taskID,
"task_name": taskName,
"error": err.Error(),
})
}()
}
},
})
}()
}
// HandleTaskRetry 处理任务失败重试逻辑
@@ -372,11 +368,11 @@ func (es *ExecutorService) HandleTaskRetry(task *models.Task, req *executor.Exec
return nil
}
newEnvs := es.loadEnvVars(latestTask.ID, latestTask.Envs)
newEnvs := es.loadEnvVars(latestTask.ID, string(latestTask.Envs))
return &executor.ExecutionRequest{
TaskID: req.TaskID,
Name: latestTask.Name,
Command: latestTask.Command,
Command: string(latestTask.Command),
WorkDir: latestTask.WorkDir,
Envs: newEnvs,
Timeout: latestTask.Timeout,
@@ -501,7 +497,7 @@ func (es *ExecutorService) AddCronTask(task *models.Task) error {
return nil
}
// 在加入调度器前，预先加载好环境信息
task.RuntimeEnvs = es.loadEnvVars(task.ID, task.Envs)
task.RuntimeEnvs = es.loadEnvVars(task.ID, string(task.Envs))
return es.cronManager.AddTask(task)
}
@@ -582,7 +578,7 @@ func (es *ExecutorService) ExecuteTask(taskID string, extraEnvs []string) *execu
}
}
envs := es.loadEnvVars(task.ID, task.Envs)
envs := es.loadEnvVars(task.ID, string(task.Envs))
if len(extraEnvs) > 0 {
envs = append(envs, extraEnvs...)
}
@@ -590,7 +586,7 @@ func (es *ExecutorService) ExecuteTask(taskID string, extraEnvs []string) *execu
req := &executor.ExecutionRequest{
TaskID: task.ID,
Name: task.Name,
Command: task.Command,
Command: string(task.Command),
WorkDir: task.WorkDir,
Envs: envs,
Timeout: task.Timeout,
@@ -749,13 +745,13 @@ func (es *ExecutorService) CheckConcurrency(taskID string) error {
return err
}
var goids []int64
if task.RunningGo != "" {
_ = json.Unmarshal([]byte(task.RunningGo), &goids)
if string(task.RunningGo) != "" {
_ = json.Unmarshal([]byte(string(task.RunningGo)), &goids)
}
var config models.TaskConfig
if task.Config != "" {
_ = json.Unmarshal([]byte(task.Config), &config)
var config models.TaskConfig
if string(task.Config) != "" {
_ = json.Unmarshal([]byte(string(task.Config)), &config)
}
if config.Concurrency == 0 && len(goids) > 0 {
@@ -792,7 +788,7 @@ func (es *ExecutorService) AddRunningGo(taskID string) (int64, error) {
goids = append(goids, goid)
data, _ := json.Marshal(goids)
return tx.Model(&task).Update("running_go", string(data)).Error
return tx.Model(&task).Update("running_go", models.BigText(data)).Error
})
if lastErr == nil {
return goid, nil
internal/services/tasks/task_log_service.go
+7 -7
View File
@@ -40,7 +40,7 @@ func (s *TaskLogService) CreateEmptyLog(taskID string, command string) (*models.
taskLog := &models.TaskLog{
ID: utils.GenerateID(),
TaskID: taskID,
Command: command,
Command: models.BigText(command),
Status: "running",
StartTime: &startTime,
CreatedAt: models.Now(),
@@ -162,9 +162,9 @@ func (s *TaskLogService) CreateTaskLogFromAgentResult(result *models.AgentTaskRe
ID: utils.GenerateID(),
TaskID: result.TaskID,
AgentID: &result.AgentID,
Command: result.Command,
Output: compressed,
Error: result.Error,
Command: models.BigText(result.Command),
Output: models.BigText(compressed),
Error: models.BigText(result.Error),
Status: result.Status,
Duration: result.Duration,
ExitCode: result.ExitCode,
@@ -206,9 +206,9 @@ func (s *TaskLogService) CreateTaskLogFromLocalExecution(taskID string, command,
taskLog := &models.TaskLog{
ID: utils.GenerateID(),
TaskID: taskID,
Command: command,
Output: compressed,
Error: systemErr,
Command: models.BigText(command),
Output: models.BigText(compressed),
Error: models.BigText(systemErr),
Status: status,
Duration: duration,
ExitCode: exitCode,
internal/services/tasks/task_service.go
+6 -6
View File
@@ -23,16 +23,16 @@ func (ts *TaskService) CreateTask(name, command, schedule string, timeout int, w
task := &models.Task{
ID: utils.GenerateID(),
Name: name,
Command: command,
Command: models.BigText(command),
Tags: tags,
Type: taskType,
TriggerType: triggerType,
Config: config,
Config: models.BigText(config),
Schedule: schedule,
Timeout: timeout,
WorkDir: workDir,
CleanConfig: cleanConfig,
Envs: envs,
Envs: models.BigText(envs),
Languages: languages,
AgentID: agentID,
Enabled: true,
@@ -94,17 +94,17 @@ func (ts *TaskService) UpdateTask(id string, name, command, schedule string, tim
return nil
}
task.Name = name
task.Command = command
task.Command = models.BigText(command)
task.Tags = tags
task.Schedule = schedule
task.Timeout = timeout
task.WorkDir = workDir
task.CleanConfig = cleanConfig
task.Envs = envs
task.Envs = models.BigText(envs)
task.Enabled = enabled
task.AgentID = agentID
task.Languages = languages
task.Config = config
task.Config = models.BigText(config)
task.RetryCount = retryCount
task.RetryInterval = retryInterval
task.RandomRange = randomRange