diff --git a/internal/cache/site_cache.go b/internal/cache/site_cache.go index 385b690..2d2b20c 100644 --- a/internal/cache/site_cache.go +++ b/internal/cache/site_cache.go @@ -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 } diff --git a/internal/constant/constant.go b/internal/constant/constant.go index 092656d..f7cbe57 100644 --- a/internal/constant/constant.go +++ b/internal/constant/constant.go @@ -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 表前缀，从配置文件读取 diff --git a/internal/controllers/app_log_controller.go b/internal/controllers/app_log_controller.go new file mode 100644 index 0000000..ebb7562 --- /dev/null +++ b/internal/controllers/app_log_controller.go @@ -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, "清理成功") +} diff --git a/internal/controllers/auth_controller.go b/internal/controllers/auth_controller.go index 5467ebb..76cd95f 100644 --- a/internal/controllers/auth_controller.go +++ b/internal/controllers/auth_controller.go @@ -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{ diff --git a/internal/controllers/log_controller.go b/internal/controllers/log_controller.go index 50342ad..4b81e64 100644 --- a/internal/controllers/log_controller.go +++ b/internal/controllers/log_controller.go @@ -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, diff --git a/internal/controllers/log_ws_controller.go b/internal/controllers/log_ws_controller.go index 2376b15..9c09887 100644 --- a/internal/controllers/log_ws_controller.go +++ b/internal/controllers/log_ws_controller.go @@ -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")) // 这里可以选择性再推一次完整版，或直接退出 diff --git a/internal/controllers/settings_controller.go b/internal/controllers/settings_controller.go index 5a82ecc..af70ee6 100644 --- a/internal/controllers/settings_controller.go +++ b/internal/controllers/settings_controller.go @@ -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, "密码修改成功") diff --git a/internal/database/database.go b/internal/database/database.go index 3490467..c810445 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -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 diff --git a/internal/database/migrate.go b/internal/database/migrate.go index 6d58ad8..5d610b7 100644 --- a/internal/database/migrate.go +++ b/internal/database/migrate.go @@ -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, ";") { diff --git a/internal/eventbus/bus.go b/internal/eventbus/bus.go new file mode 100644 index 0000000..bcf3866 --- /dev/null +++ b/internal/eventbus/bus.go @@ -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() diff --git a/internal/models/app_log.go b/internal/models/app_log.go new file mode 100644 index 0000000..7601148 --- /dev/null +++ b/internal/models/app_log.go @@ -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" +} diff --git a/internal/models/big_text.go b/internal/models/big_text.go new file mode 100644 index 0000000..faac288 --- /dev/null +++ b/internal/models/big_text.go @@ -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" +} diff --git a/internal/models/dependency.go b/internal/models/dependency.go index 450aab0..acefe74 100644 --- a/internal/models/dependency.go +++ b/internal/models/dependency.go @@ -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"` } diff --git a/internal/models/env.go b/internal/models/env.go index 20c3de3..837405b 100644 --- a/internal/models/env.go +++ b/internal/models/env.go @@ -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"` diff --git a/internal/models/notify_way.go b/internal/models/notify_way.go index 63a0f5a..8d4598c 100644 --- a/internal/models/notify_way.go +++ b/internal/models/notify_way.go @@ -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"` diff --git a/internal/models/setting.go b/internal/models/setting.go index 9326264..61fa1ff 100644 --- a/internal/models/setting.go +++ b/internal/models/setting.go @@ -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 { diff --git a/internal/models/task.go b/internal/models/task.go index 43357f8..6ab511d 100644 --- a/internal/models/task.go +++ b/internal/models/task.go @@ -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"` diff --git a/internal/models/vo/dep_vo.go b/internal/models/vo/dep_vo.go index 6cdc462..8af9c3f 100644 --- a/internal/models/vo/dep_vo.go +++ b/internal/models/vo/dep_vo.go @@ -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, } diff --git a/internal/models/vo/script_vo.go b/internal/models/vo/script_vo.go index fc0c84f..eda67b2 100644 --- a/internal/models/vo/script_vo.go +++ b/internal/models/vo/script_vo.go @@ -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, } diff --git a/internal/models/vo/sys_vo.go b/internal/models/vo/sys_vo.go index 36024f2..53a1de0 100644 --- a/internal/models/vo/sys_vo.go +++ b/internal/models/vo/sys_vo.go @@ -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, diff --git a/internal/models/vo/task_vo.go b/internal/models/vo/task_vo.go index 3ca9b55..f2131ef 100644 --- a/internal/models/vo/task_vo.go +++ b/internal/models/vo/task_vo.go @@ -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), } } diff --git a/internal/router/events.go b/internal/router/events.go new file mode 100644 index 0000000..4f20481 --- /dev/null +++ b/internal/router/events.go @@ -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() + } +} diff --git a/internal/router/register.go b/internal/router/register.go index e6827bb..1f5ed81 100644 --- a/internal/router/register.go +++ b/internal/router/register.go @@ -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(), } } diff --git a/internal/router/router.go b/internal/router/router.go index 7210417..230d090 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -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") diff --git a/internal/services/agent_service.go b/internal/services/agent_service.go index adc72cf..d59f24b 100644 --- a/internal/services/agent_service.go +++ b/internal/services/agent_service.go @@ -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, diff --git a/internal/services/app_log_service.go b/internal/services/app_log_service.go new file mode 100644 index 0000000..98a3da2 --- /dev/null +++ b/internal/services/app_log_service.go @@ -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, + }, + }) + }) +} diff --git a/internal/services/backup_service.go b/internal/services/backup_service.go index 3020b4f..75419a9 100644 --- a/internal/services/backup_service.go +++ b/internal/services/backup_service.go @@ -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 { diff --git a/internal/services/deps/manager.go b/internal/services/deps/manager.go index 6ee4b20..f8db8e2 100644 --- a/internal/services/deps/manager.go +++ b/internal/services/deps/manager.go @@ -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)) diff --git a/internal/services/env_service.go b/internal/services/env_service.go index b2a3381..6d2c83a 100644 --- a/internal/services/env_service.go +++ b/internal/services/env_service.go @@ -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)}, }) } } diff --git a/internal/services/login_log_service.go b/internal/services/login_log_service.go index d50979f..2dd6405 100644 --- a/internal/services/login_log_service.go +++ b/internal/services/login_log_service.go @@ -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 diff --git a/internal/services/notification_service.go b/internal/services/notification_service.go index 0417d98..4fd1f25 100644 --- a/internal/services/notification_service.go +++ b/internal/services/notification_service.go @@ -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) + } } } diff --git a/internal/services/script_service.go b/internal/services/script_service.go index 8f4e652..ead4994 100644 --- a/internal/services/script_service.go +++ b/internal/services/script_service.go @@ -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 } diff --git a/internal/services/settings_service.go b/internal/services/settings_service.go index b4c8d97..9f68bfd 100644 --- a/internal/services/settings_service.go +++ b/internal/services/settings_service.go @@ -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 } diff --git a/internal/services/tasks/executor_service.go b/internal/services/tasks/executor_service.go index cabd05e..d6a98b9 100644 --- a/internal/services/tasks/executor_service.go +++ b/internal/services/tasks/executor_service.go @@ -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 diff --git a/internal/services/tasks/task_log_service.go b/internal/services/tasks/task_log_service.go index c265c60..26e1030 100644 --- a/internal/services/tasks/task_log_service.go +++ b/internal/services/tasks/task_log_service.go @@ -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, diff --git a/internal/services/tasks/task_service.go b/internal/services/tasks/task_service.go index 3690ca6..f520d9b 100644 --- a/internal/services/tasks/task_service.go +++ b/internal/services/tasks/task_service.go @@ -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 diff --git a/package-lock.json b/package-lock.json new file mode 100644 index 0000000..235fb8a --- /dev/null +++ b/package-lock.json @@ -0,0 +1,22 @@ +{ + "name": "baihu2", + "lockfileVersion": 3, + "requires": true, + "packages": { + "": { + "dependencies": { + "date-fns": "^4.1.0" + } + }, + "node_modules/date-fns": { + "version": "4.1.0", + "resolved": "https://registry.npmjs.org/date-fns/-/date-fns-4.1.0.tgz", + "integrity": "sha512-Ukq0owbQXxa/U3EGtsdVBkR1w7KOQ5gIBqdH2hkvknzZPYvBxb/aa6E8L7tmjFtkwZBu3UXBbjIgPo/Ez4xaNg==", + "license": "MIT", + "funding": { + "type": "github", + "url": "https://github.com/sponsors/kossnocorp" + } + } + } +} diff --git a/package.json b/package.json new file mode 100644 index 0000000..c4a6474 --- /dev/null +++ b/package.json @@ -0,0 +1,5 @@ +{ + "dependencies": { + "date-fns": "^4.1.0" + } +} diff --git a/web/src/api/index.ts b/web/src/api/index.ts index 11bd046..029b038 100644 --- a/web/src/api/index.ts +++ b/web/src/api/index.ts @@ -287,6 +287,20 @@ export const api = { deleteBinding: (id: string) => request('/notify/bindings/' + id, { method: 'DELETE' }), send: (data: { channel_id: string; title: string; text: string }) => request('/notify/send', { method: 'POST', body: JSON.stringify(data) }) + }, + appLogs: { + list: (params?: { page?: number; page_size?: number; category?: string; status?: string; level?: string; keyword?: string }) => { + const query = new URLSearchParams() + if (params?.page) query.set('page', String(params.page)) + if (params?.page_size) query.set('page_size', String(params.page_size)) + if (params?.category) query.set('category', params.category) + if (params?.status) query.set('status', params.status) + if (params?.level) query.set('level', params.level) + if (params?.keyword) query.set('keyword', params.keyword) + return request(`/app-logs?${query}`) + }, + markAsRead: (data: { id?: string; category?: string }) => request('/app-logs/read', { method: 'POST', body: JSON.stringify(data) }), + clear: (category: string) => request('/app-logs/clear', { method: 'POST', body: JSON.stringify({ category }) }) } } @@ -561,4 +575,40 @@ export interface NotifyResult { error?: string } +export interface AppLog { + id: string + category: string + title: string + content: string + level: string + status: string + ref_id: string + error_msg: string + created_at: string + read_at: string | null +} + +export interface AppLogListResponse { + data: AppLog[] + total: number +} + +export const LOG_CATEGORY = { + SYSTEM_NOTICE: 'system_notice', + PUSH_LOG: 'push_log' +} as const + +export const LOG_LEVEL = { + INFO: 'info', + WARNING: 'warning', + ERROR: 'error' +} as const + +export const LOG_STATUS = { + UNREAD: 'unread', + READ: 'read', + SUCCESS: 'success', + FAILED: 'failed' +} as const + diff --git a/web/src/components/SystemNotice.vue b/web/src/components/SystemNotice.vue new file mode 100644 index 0000000..4911172 --- /dev/null +++ b/web/src/components/SystemNotice.vue @@ -0,0 +1,125 @@ + + + diff --git a/web/src/layouts/MainLayout.vue b/web/src/layouts/MainLayout.vue index f7639d9..4839326 100644 --- a/web/src/layouts/MainLayout.vue +++ b/web/src/layouts/MainLayout.vue @@ -5,6 +5,7 @@ import { resetAuthCache } from '@/router' import { LayoutDashboard, ListTodo, FileCode, Settings, LogOut, ScrollText, Terminal, Variable, KeyRound, Menu, X, Server, Globe, Bell } from 'lucide-vue-next' import { Button } from '@/components/ui/button' import ThemeToggle from '@/components/ThemeToggle.vue' +import SystemNotice from '@/components/SystemNotice.vue' import { api } from '@/api' import { useSiteSettings } from '@/composables/useSiteSettings' @@ -150,7 +151,10 @@ onMounted(() => { {{ sentenceContent }} - +
+ + +
diff --git a/web/src/views/notify/Notify.vue b/web/src/views/notify/Notify.vue index 66224d5..fab21c1 100644 --- a/web/src/views/notify/Notify.vue +++ b/web/src/views/notify/Notify.vue @@ -17,6 +17,7 @@ import ChannelList from './components/ChannelList.vue' import EventBinding from './components/EventBinding.vue' import ApiUsage from './components/ApiUsage.vue' import ChannelDialog from './components/ChannelDialog.vue' +import PushLog from './components/PushLog.vue' const activeTab = ref('channels') @@ -322,9 +323,10 @@ onMounted(() => {

消息推送

配置通知渠道，绑定系统事件实现自动推送

- + 渠道管理 事件绑定 + 推送日志 脚本调用 @@ -346,6 +348,11 @@ onMounted(() => { + + + + + diff --git a/web/src/views/notify/components/ApiUsage.vue b/web/src/views/notify/components/ApiUsage.vue index 4c82d64..a65ebd2 100644 --- a/web/src/views/notify/components/ApiUsage.vue +++ b/web/src/views/notify/components/ApiUsage.vue @@ -235,7 +235,7 @@ const shellExample = computed(() => `curl -s -X POST "http://${host.value}/api/v
{{ ch.id.slice(0, 8) }} + :title="ch.id">{{ ch.id.slice(0, 8) }}... {{ ch.name }}