Merge pull request #210 from komari-monitor/copilot/fix-b576a6c5-0ea2-4f16-8483-2eb7767cd8ad

Add PR and manual triggers to binary build workflow

.github/workflows/build.yml

@@ -1,9 +1,13 @@
-name: Build Binaries on Main Push
+name: Build Binaries on Main Push and PR
 
 on:
   push:
     branches:
       - main
+  pull_request:
+    branches:
+      - main
+  workflow_dispatch:
 
 jobs:
   build-binaries:

.github/workflows/cleanup-packages.yml

@@ -1,8 +1,9 @@
 name: Cleanup Old Packages
 
 on:
-  schedule:
-    - cron: '0 0 * * 0'  # Run every Sunday at midnight
+  #schedule:
+  #  - cron: '0 0 * * 0'  # Run every Sunday at midnight
+  # 会清理掉正常的镜像
   workflow_dispatch:      # Allow manual trigger
 
 jobs:
@@ -57,4 +58,4 @@ jobs:
                 package_version_id: version.id
               });
               console.log(`Deleted package version ${version.id} with tags: ${version.metadata.container.tags.join(', ') || 'none'}`);
-            }
+            }

.github/workflows/docker-publish.yml

@@ -1,9 +1,9 @@
 name: Publish Docker Image on Main
 
 on:
-  push:
-    branches:
-      - main
+  #push:
+  #  branches:
+  #    - main
   workflow_dispatch:
 
 env:

api/PrivateSiteMiddleware.go

@@ -18,6 +18,7 @@ var (
 		"/api/oauth",
 		"/api/oauth_callback",
 		"/api/version",
+		"/api/recent",
 		"/api/admin",    // 由AdminAuthMiddleware处理
 		"/api/clients/", // 由TokenAuthMiddleware处理
 	}

api/admin/exec.go

@@ -66,7 +66,10 @@ func Exec(c *gin.Context) {
 		payload, _ := json.Marshal(send)
 		client := ws.GetConnectedClients()[uuid]
 		if client != nil {
-			client.WriteMessage(websocket.TextMessage, payload)
+			if err := client.WriteMessage(websocket.TextMessage, payload); err != nil {
+				api.RespondError(c, 400, "Client connection is broke: "+uuid)
+				return
+			}
 		} else {
 			api.RespondError(c, 400, "Client connection is null: "+uuid)
 			return

api/admin/theme.go

@@ -15,6 +15,7 @@ import (
 	"github.com/gin-gonic/gin"
 	"github.com/komari-monitor/komari/api"
 	"github.com/komari-monitor/komari/database/config"
+	"github.com/komari-monitor/komari/database/dbcore"
 	"github.com/komari-monitor/komari/database/models"
 )
 
@@ -587,3 +588,32 @@ func UpdateTheme(c *gin.Context) {
 
 	api.RespondSuccessMessage(c, "主题更新成功", updatedThemeInfo)
 }
+
+func UpdateThemeSettings(c *gin.Context) {
+	theme := c.Query("theme")
+	if theme == "" || theme == "default" {
+		api.RespondError(c, http.StatusBadRequest, "主题名称不能为空或不能是默认主题")
+		return
+	}
+
+	var req map[string]any
+
+	err := c.ShouldBindJSON(&req)
+	if err != nil {
+		api.RespondError(c, http.StatusBadRequest, "参数错误: "+err.Error())
+		return
+	}
+	db := dbcore.GetDBInstance()
+
+	data, err := json.Marshal(&req)
+	if err != nil {
+		api.RespondError(c, http.StatusInternalServerError, "生成主题配置失败: "+err.Error())
+		return
+	}
+
+	var themeCfg models.ThemeConfiguration
+	db.Where("short = ?", theme).
+		Assign(models.ThemeConfiguration{Short: theme, Data: string(data)}).
+		FirstOrCreate(&themeCfg)
+	api.RespondSuccess(c, nil)
+}

api/client/report.go

@@ -22,6 +22,12 @@ import (
 	"github.com/patrickmn/go-cache"
 )
 
+const (
+	// 如果超过这个时间没有收到任何消息，则认为连接已死
+	// 因为目前server没有存agent的信息上报间隔。只有写一个默认的
+	readWait = 11 * time.Second
+)
+
 func UploadReport(c *gin.Context) {
 	bodyBytes, err := io.ReadAll(c.Request.Body)
 	if err != nil {
@@ -51,7 +57,7 @@ func UploadReport(c *gin.Context) {
 	}
 	// Update report with method and token
 
-	ws.GetLatestReport()[report.UUID] = &report
+	ws.SetLatestReport(report.UUID, &report)
 
 	c.Request.Body = io.NopCloser(bytes.NewBuffer(bodyBytes)) // Restore the body for further use
 	c.JSON(200, gin.H{"status": "success"})
@@ -109,72 +115,87 @@ func WebSocketReport(c *gin.Context) {
 		return
 	}
 
-	// 只允许一个客户端的连接
-	if _, exists := ws.GetConnectedClients()[uuid]; exists {
-		conn.WriteJSON(gin.H{"status": "error", "error": "Token already in use"})
-		return
+	// 接受新连接，并处理旧连接
+	if oldConn, exists := ws.GetConnectedClients()[uuid]; exists {
+		log.Printf("Client %s is reconnecting. Closing the old connection.", uuid)
+
+		// 强制关闭旧连接。这将导致旧连接的 ReadMessage() 循环出错退出。
+		go oldConn.Close()
 	}
 	ws.SetConnectedClients(uuid, conn)
-	go notifier.OnlineNotification(uuid)
+	log.Printf("Client %s is reconnect success, connID: %d", uuid, conn.ID)
+	go notifier.OnlineNotification(uuid, conn.ID)
 	defer func() {
-		ws.DeleteConnectedClients(uuid)
-		notifier.OfflineNotification(uuid)
+		ws.DeleteClientConditionally(uuid, conn)
+		notifier.OfflineNotification(uuid, conn.ID)
 	}()
 
+	// 首先处理第一次ws conn收到的消息
+	processMessage(conn, message, uuid)
+
 	for {
+		conn.SetReadDeadline(time.Now().Add(readWait))
+
 		_, message, err := conn.ReadMessage()
 		if err != nil {
-			break
+			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
+				log.Printf("Client %s connection error: %v", uuid, err)
+			}
+			break // 任何读错误（包括超时）都意味着连接已断开，退出循环
 		}
+		processMessage(conn, message, uuid)
+	}
+}
 
-		type MessageType struct {
-			Type string `json:"type"`
-		}
-		var msgType MessageType
-		err = json.Unmarshal(message, &msgType)
+// 将消息处理逻辑提取到一个函数中，方便复用
+func processMessage(conn *ws.SafeConn, message []byte, uuid string) {
+	type MessageType struct {
+		Type string `json:"type"`
+	}
+	var msgType MessageType
+	err := json.Unmarshal(message, &msgType)
+	if err != nil {
+		conn.WriteJSON(gin.H{"status": "error", "error": "Invalid JSON"})
+		return
+	}
+
+	switch msgType.Type {
+	case "", "report":
+		report := common.Report{}
+		err = json.Unmarshal(message, &report)
 		if err != nil {
-			conn.WriteJSON(gin.H{"status": "error", "error": "Invalid JSON"})
-			continue
+			conn.WriteJSON(gin.H{"status": "error", "error": "Invalid report format"})
+			return
 		}
-		switch msgType.Type {
-		case "", "report":
-			report := common.Report{}
-			err = json.Unmarshal(message, &report)
-			if err != nil {
-				conn.WriteJSON(gin.H{"status": "error", "error": "Invalid report format"})
-				continue
-			}
-			report.UpdatedAt = time.Now()
-			err = SaveClientReport(uuid, report)
-			if err != nil {
-				conn.WriteJSON(gin.H{"status": "error", "error": fmt.Sprintf("%v", err)})
-				continue
-			}
-			ws.SetLatestReport(uuid, &report)
-		case "ping_result":
-			var reqBody struct {
-				PingTaskID uint      `json:"task_id"`
-				PingResult int       `json:"value"`
-				PingType   string    `json:"ping_type"`
-				FinishedAt time.Time `json:"finished_at"`
-			}
-			err = json.Unmarshal(message, &reqBody)
-			if err != nil {
-				conn.WriteJSON(gin.H{"status": "error", "error": "Invalid ping result format"})
-				continue
-			}
-			pingResult := models.PingRecord{
-				Client: uuid,
-				TaskId: reqBody.PingTaskID,
-				Value:  reqBody.PingResult,
-				Time:   models.FromTime(reqBody.FinishedAt),
-			}
-			tasks.SavePingRecord(pingResult)
-		default:
-			log.Printf("Unknown message type: %s", msgType.Type)
-			conn.WriteJSON(gin.H{"status": "error", "error": "Unknown message type"})
+		report.UpdatedAt = time.Now()
+		err = SaveClientReport(uuid, report)
+		if err != nil {
+			conn.WriteJSON(gin.H{"status": "error", "error": fmt.Sprintf("%v", err)})
+			return
 		}
-
+		ws.SetLatestReport(uuid, &report)
+	case "ping_result":
+		var reqBody struct {
+			PingTaskID uint      `json:"task_id"`
+			PingResult int       `json:"value"`
+			PingType   string    `json:"ping_type"`
+			FinishedAt time.Time `json:"finished_at"`
+		}
+		err = json.Unmarshal(message, &reqBody)
+		if err != nil {
+			conn.WriteJSON(gin.H{"status": "error", "error": "Invalid ping result format"})
+			return
+		}
+		pingResult := models.PingRecord{
+			Client: uuid,
+			TaskId: reqBody.PingTaskID,
+			Value:  reqBody.PingResult,
+			Time:   models.FromTime(reqBody.FinishedAt),
+		}
+		tasks.SavePingRecord(pingResult)
+	default:
+		log.Printf("Unknown message type: %s", msgType.Type)
+		conn.WriteJSON(gin.H{"status": "error", "error": "Unknown message type"})
 	}
 }
 

api/client/uploadBasicInfo.go

@@ -10,6 +10,18 @@ import (
 	"github.com/gin-gonic/gin"
 )
 
+func getClientIPType(ip net.IP) int {
+	// 0:ipv4 1:ipv6 -1:错误的输入
+	if ip == nil {
+		return -1
+	}
+	if ip.To4() == nil {
+		return 1
+	} else {
+		return 0
+	}
+}
+
 func UploadBasicInfo(c *gin.Context) {
 	var cbi = map[string]interface{}{}
 	if err := c.ShouldBindJSON(&cbi); err != nil {
@@ -26,6 +38,28 @@ func UploadBasicInfo(c *gin.Context) {
 
 	cbi["uuid"] = uuid
 
+	if (func() bool {
+		if v4, ok := cbi["ipv4"].(string); !ok || v4 == "" {
+			if v6, ok := cbi["ipv6"].(string); !ok || v6 == "" {
+				return true
+			}
+		}
+		return false
+	})() {
+		ipStr := c.ClientIP()
+		ip := net.ParseIP(ipStr)
+		ipType := getClientIPType(ip)
+
+		switch ipType {
+		case 0:
+			cbi["ipv4"] = ip
+		case 1:
+			cbi["ipv6"] = ip
+		default:
+			break
+		}
+	}
+
 	if cfg, err := config.Get(); err == nil && cfg.GeoIpEnabled {
 		if ipv4, ok := cbi["ipv4"].(string); ok && ipv4 != "" {
 			ip4 := net.ParseIP(ipv4)

api/latestRecord.go

@@ -2,14 +2,43 @@ package api
 
 import (
 	"github.com/gin-gonic/gin"
+	"github.com/komari-monitor/komari/database/accounts"
+	"github.com/komari-monitor/komari/database/dbcore"
+	"github.com/komari-monitor/komari/database/models"
 )
 
 func GetClientRecentRecords(c *gin.Context) {
 	uuid := c.Param("uuid")
+
 	if uuid == "" {
 		RespondError(c, 400, "UUID is required")
 		return
 	}
+
+	// 登录状态检查
+	isLogin := false
+	session, _ := c.Cookie("session_token")
+	_, err := accounts.GetUserBySession(session)
+	if err == nil {
+		isLogin = true
+	}
+
+	// 仅在未登录时需要 Hidden 信息做过滤
+	hiddenMap := map[string]bool{}
+	if !isLogin {
+		var hiddenClients []models.Client
+		db := dbcore.GetDBInstance()
+		_ = db.Select("uuid").Where("hidden = ?", true).Find(&hiddenClients).Error
+		for _, cli := range hiddenClients {
+			hiddenMap[cli.UUID] = true
+		}
+
+		if hiddenMap[uuid] {
+			RespondError(c, 400, "UUID is required") //防止未登录用户获取隐藏客户端数据
+			return
+		}
+	}
+
 	records, _ := Records.Get(uuid)
 	RespondSuccess(c, records)
 }

api/nodes.go

@@ -2,6 +2,7 @@ package api
 
 import (
 	"github.com/gin-gonic/gin"
+	"github.com/komari-monitor/komari/database/accounts"
 	"github.com/komari-monitor/komari/database/clients"
 )
 
@@ -11,16 +12,28 @@ func GetNodesInformation(c *gin.Context) {
 		RespondError(c, 500, "Failed to retrieve client information: "+err.Error())
 		return
 	}
+	isLogin := false
+	session, _ := c.Cookie("session_token")
+	_, err = accounts.GetUserBySession(session)
+	if err == nil {
+		isLogin = true
+	}
 
-	count := len(clientList)
-	// 公开信息不展示IP地址,私有备注不展示
-	for i := 0; i < count; i++ {
+	// 过滤掉 Hidden 的客户端，并清理需要隐藏的字段
+	j := 0
+	for i := 0; i < len(clientList); i++ {
+		if clientList[i].Hidden && !isLogin { // 不返回 Hidden 客户端
+			continue
+		}
 		clientList[i].IPv4 = ""
 		clientList[i].IPv6 = ""
-		clientList[i].Remark = ""
+		clientList[i].Remark = "" // 私有备注不展示
 		clientList[i].Version = ""
 		clientList[i].Token = ""
+		clientList[j] = clientList[i]
+		j++
 	}
+	clientList = clientList[:j]
 
 	RespondSuccess(c, clientList)
 }

api/public.go

@@ -1,8 +1,16 @@
 package api
 
 import (
+	"encoding/json"
+	"log"
+	"os"
+	"path/filepath"
+	"strings"
+
 	"github.com/gin-gonic/gin"
 	"github.com/komari-monitor/komari/database/config"
+	"github.com/komari-monitor/komari/database/dbcore"
+	"github.com/komari-monitor/komari/database/models"
 )
 
 func GetPublicSettings(c *gin.Context) {
@@ -11,6 +19,70 @@ func GetPublicSettings(c *gin.Context) {
 		RespondError(c, 500, err.Error())
 		return
 	}
+	db := dbcore.GetDBInstance()
+	tc := models.ThemeConfiguration{}
+	err = db.Model(&models.ThemeConfiguration{}).Where("short = ?", cst.Theme).First(&tc).Error
+	if err != nil {
+		tc.Data = "{}"
+	}
+	tc_data := gin.H{}
+	err = json.Unmarshal([]byte(tc.Data), &tc_data)
+	if err != nil {
+		log.Printf("%v", err)
+	}
+	// Try to load theme declaration file and merge defaults for managed configuration
+	// Theme declarations live in ./data/theme/<short>/komari-theme.json
+	if cst.Theme != "" && cst.Theme != "default" {
+		themeConfigPath := filepath.Join("./data/theme", cst.Theme, "komari-theme.json")
+		if _, err := os.Stat(themeConfigPath); err == nil {
+			b, err := os.ReadFile(themeConfigPath)
+			if err == nil {
+				var themeDecl struct {
+					Configuration struct {
+						Type string                                 `json:"type"`
+						Data []models.ManagedThemeConfigurationItem `json:"data"`
+					} `json:"configuration"`
+				}
+				if err := json.Unmarshal(b, &themeDecl); err == nil {
+					if themeDecl.Configuration.Type == "managed" {
+						for _, item := range themeDecl.Configuration.Data {
+							if item.Key == "" {
+								continue
+							}
+							// missing
+							if _, exists := tc_data[item.Key]; !exists {
+								var def any = item.Default
+								// select
+								if item.Type == "select" {
+									if def == nil || def == "" {
+										if item.Options != "" {
+											opts := strings.Split(item.Options, ",")
+											if len(opts) > 0 {
+												def = strings.TrimSpace(opts[0])
+											}
+										}
+									}
+								}
+								// number->0, string->"", switch->false
+								if def == nil {
+									switch item.Type {
+									case "number":
+										def = 0
+									case "switch":
+										def = false
+									default:
+										def = ""
+									}
+								}
+								tc_data[item.Key] = def
+							}
+						}
+					}
+				}
+			}
+		}
+	}
+
 	// Return public settings including CORS
 	RespondSuccess(c, gin.H{
 		"sitename":                  cst.Sitename,
@@ -25,6 +97,8 @@ func GetPublicSettings(c *gin.Context) {
 		"record_preserve_time":      cst.RecordPreserveTime,
 		"ping_record_preserve_time": cst.PingRecordPreserveTime,
 		"private_site":              cst.PrivateSite,
+		"theme":                     cst.Theme,
+		"theme_settings":            tc_data,
 	})
 
 }

api/record/record.go

@@ -7,12 +7,40 @@ import (
 
 	"github.com/gin-gonic/gin"
 	"github.com/komari-monitor/komari/api"
+	"github.com/komari-monitor/komari/database/accounts"
+	"github.com/komari-monitor/komari/database/dbcore"
+	"github.com/komari-monitor/komari/database/models"
 	records "github.com/komari-monitor/komari/database/records"
 	"github.com/komari-monitor/komari/database/tasks"
 )
 
 func GetRecordsByUUID(c *gin.Context) {
 	uuid := c.Query("uuid")
+
+	// 登录状态检查
+	isLogin := false
+	session, _ := c.Cookie("session_token")
+	_, err := accounts.GetUserBySession(session)
+	if err == nil {
+		isLogin = true
+	}
+
+	// 仅在未登录时需要 Hidden 信息做过滤
+	hiddenMap := map[string]bool{}
+	if !isLogin {
+		var hiddenClients []models.Client
+		db := dbcore.GetDBInstance()
+		_ = db.Select("uuid").Where("hidden = ?", true).Find(&hiddenClients).Error
+		for _, cli := range hiddenClients {
+			hiddenMap[cli.UUID] = true
+		}
+
+		if hiddenMap[uuid] {
+			api.RespondError(c, 400, "UUID is required") //防止未登录用户获取隐藏客户端数据
+			return
+		}
+	}
+
 	hours := c.Query("hours")
 	if uuid == "" {
 		api.RespondError(c, 400, "UUID is required")
@@ -42,6 +70,36 @@ func GetRecordsByUUID(c *gin.Context) {
 // GET query: uuid string, hours int
 func GetPingRecords(c *gin.Context) {
 	uuid := c.Query("uuid")
+
+	if uuid == "" {
+		api.RespondError(c, 400, "UUID is required")
+		return
+	}
+
+	// 登录状态检查
+	isLogin := false
+	session, _ := c.Cookie("session_token")
+	_, err := accounts.GetUserBySession(session)
+	if err == nil {
+		isLogin = true
+	}
+
+	// 仅在未登录时需要 Hidden 信息做过滤
+	hiddenMap := map[string]bool{}
+	if !isLogin {
+		var hiddenClients []models.Client
+		db := dbcore.GetDBInstance()
+		_ = db.Select("uuid").Where("hidden = ?", true).Find(&hiddenClients).Error
+		for _, cli := range hiddenClients {
+			hiddenMap[cli.UUID] = true
+		}
+
+		if hiddenMap[uuid] {
+			api.RespondError(c, 400, "UUID is required") //防止未登录用户获取隐藏客户端数据
+			return
+		}
+	}
+
 	hours := c.Query("hours")
 	type RecordsResp struct {
 		TaskId uint   `json:"task_id"`
@@ -54,10 +112,6 @@ func GetPingRecords(c *gin.Context) {
 		Tasks   []gin.H       `json:"tasks,omitempty"` // 任务列表
 	}
 
-	if uuid == "" {
-		api.RespondError(c, 400, "UUID is required")
-		return
-	}
 	if hours == "" {
 		hours = "4"
 	}

api/ws.go

@@ -0,0 +1,124 @@
+package api
+
+import (
+	"net/http"
+	"strings"
+
+	"github.com/gin-gonic/gin"
+	"github.com/gorilla/websocket"
+	"github.com/komari-monitor/komari/common"
+	"github.com/komari-monitor/komari/database/accounts"
+	"github.com/komari-monitor/komari/database/config"
+	"github.com/komari-monitor/komari/database/dbcore"
+	"github.com/komari-monitor/komari/database/models"
+	"github.com/komari-monitor/komari/ws"
+)
+
+func GetClients(c *gin.Context) {
+	// 升级到ws
+	if !websocket.IsWebSocketUpgrade(c.Request) {
+		c.JSON(http.StatusBadRequest, gin.H{"status": "error", "error": "Require WebSocket upgrade"})
+		return
+	}
+	cfg, _ := config.Get()
+	upgrader := websocket.Upgrader{
+		CheckOrigin: func(r *http.Request) bool {
+			if cfg.AllowCors {
+				return true
+			}
+			return ws.CheckOrigin(r)
+		},
+	}
+	// Upgrade the HTTP connection to a WebSocket connection
+	conn, err := upgrader.Upgrade(c.Writer, c.Request, nil)
+	if err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"status": "error", "error": "Failed to upgrade to WebSocket"})
+		return
+	}
+	defer conn.Close()
+
+	// 请求
+	for {
+		var resp struct {
+			Online []string                  `json:"online"` // 已建立连接的客户端uuid列表
+			Data   map[string]*common.Report `json:"data"`   // 最后上报的数据
+		}
+		resp.Online = []string{}
+		_, data, err := conn.ReadMessage()
+		if err != nil {
+			//log.Println("Error reading message:", err)
+			return
+		}
+		message := string(data)
+
+		uuID := ""
+		if message != "get" { // 非请求全部内容
+			if strings.HasPrefix(message, "get ") {
+				uuID = strings.TrimSpace(strings.TrimPrefix(message, "get "))
+			} else {
+				conn.WriteJSON(gin.H{"status": "error", "error": "Invalid message"})
+				continue
+			}
+		}
+
+		// 登录状态检查
+		isLogin := false
+		session, _ := c.Cookie("session_token")
+		_, err = accounts.GetUserBySession(session)
+		if err == nil {
+			isLogin = true
+		}
+		// 仅在未登录时需要 Hidden 信息做过滤
+		hiddenMap := map[string]bool{}
+		if !isLogin {
+			var hiddenClients []models.Client
+			db := dbcore.GetDBInstance()
+			_ = db.Select("uuid").Where("hidden = ?", true).Find(&hiddenClients).Error
+			for _, cli := range hiddenClients {
+				hiddenMap[cli.UUID] = true
+			}
+		}
+		// 已建立连接的客户端uuid列表
+		for key := range ws.GetConnectedClients() {
+			if uuID != "" { // 请求特定服务器信息
+				if !(!isLogin && hiddenMap[key]) && key == uuID { // 未登录且 Hidden 或 不符合请求的特定服务器 -> 跳过
+					resp.Online = append(resp.Online, key)
+				}
+			} else {
+				if !(!isLogin && hiddenMap[key]) { // 未登录且 Hidden -> 跳过
+					resp.Online = append(resp.Online, key)
+				}
+			}
+		}
+		// 清除UUID，简化报告单
+		resp.Data = ws.GetLatestReport()
+		if !isLogin { // 未登录过滤 Hidden 的节点报告
+			for uuid := range resp.Data {
+				if hiddenMap[uuid] {
+					delete(resp.Data, uuid)
+				}
+			}
+		}
+
+		if uuID != "" { // 过滤不符合请求的特定服务器
+			for key := range resp.Data {
+				if key != uuID {
+					delete(resp.Data, key)
+				}
+			}
+		}
+
+		for _, report := range resp.Data { // 不暴露 uuid
+			report.UUID = ""
+			if report.CPU.Usage == 0 {
+				report.CPU.Usage = 0.01
+			}
+		}
+
+		err = conn.WriteJSON(gin.H{"status": "success", "data": resp})
+		if err != nil {
+			return
+		}
+	}
+
+}

cmd/server.go

@@ -38,7 +38,6 @@ import (
 	"github.com/komari-monitor/komari/utils/messageSender"
 	"github.com/komari-monitor/komari/utils/notifier"
 	"github.com/komari-monitor/komari/utils/oauth"
-	"github.com/komari-monitor/komari/ws"
 	"github.com/spf13/cobra"
 )
 
@@ -155,7 +154,7 @@ func RunServer() {
 	// #region 公开路由
 	r.POST("/api/login", api.Login)
 	r.GET("/api/me", api.GetMe)
-	r.GET("/api/clients", ws.GetClients)
+	r.GET("/api/clients", api.GetClients)
 	r.GET("/api/nodes", api.GetNodesInformation)
 	r.GET("/api/public", api.GetPublicSettings)
 	r.GET("/api/oauth", api.OAuth)
@@ -223,6 +222,7 @@ func RunServer() {
 			themeGroup.POST("/delete", admin.DeleteTheme)
 			themeGroup.GET("/set", admin.SetTheme)
 			themeGroup.POST("/update", admin.UpdateTheme)
+			themeGroup.POST("/settings", admin.UpdateThemeSettings)
 		}
 		// clients
 		clientGroup := adminAuthrized.Group("/client")
@@ -380,6 +380,8 @@ func DoScheduledWork() {
 				records.DeleteAll()
 				tasks.DeleteAllPingRecords()
 			}
+			// 每分钟检查一次流量提醒
+			go notifier.CheckTraffic()
 		}
 	}
 

database/accounts/sessions.go

@@ -2,11 +2,16 @@ package accounts
 
 import (
 	"errors"
+	"fmt"
+	"net"
 	"time"
 
+	"github.com/komari-monitor/komari/database/config"
 	"github.com/komari-monitor/komari/database/dbcore"
 	"github.com/komari-monitor/komari/database/models"
 	"github.com/komari-monitor/komari/utils"
+	"github.com/komari-monitor/komari/utils/geoip"
+	"github.com/komari-monitor/komari/utils/messageSender"
 )
 
 // GetAllSessions 获取所有会话
@@ -33,6 +38,16 @@ func CreateSession(uuid string, expires int, userAgent, ip, login_method string)
 		LoginMethod:  login_method,
 		LatestOnline: models.FromTime(time.Now()),
 	}
+	cfg, _ := config.Get()
+	if cfg.LoginNotification {
+		ipAddr := net.ParseIP(ip)
+		ipinfo, _ := geoip.GetGeoInfo(ipAddr)
+		loc := "unknown"
+		if ipinfo != nil {
+			loc = ipinfo.Name
+		}
+		messageSender.SendTextMessage(fmt.Sprintf("From: %s,(%s)\nMethod: %s\nUser Agent: %s\n\n%s", ip, loc, login_method, userAgent, time.Now().Format(time.RFC3339)), "New Login on Komari")
+	}
 
 	err := db.Create(&sessionRecord).Error
 	if err != nil {

database/clients/client.go

@@ -1,6 +1,7 @@
 package clients
 
 import (
+	"math"
 	"time"
 
 	"github.com/komari-monitor/komari/common"
@@ -47,7 +48,7 @@ func UpdateOrInsertBasicInfo(cbi common.ClientInfo) error {
 	if cbi.Arch != "" {
 		updates["arch"] = cbi.Arch
 	}
-	if cbi.CpuCores != 0 {
+	if cbi.CpuCores > 0 || cbi.CpuCores < math.MaxInt-1 {
 		updates["cpu_cores"] = cbi.CpuCores
 	}
 	if cbi.OS != "" {
@@ -120,6 +121,36 @@ func SaveClientInfo(update map[string]interface{}) error {
 
 	update["updated_at"] = time.Now()
 
+	checkInt64 := func(name string, val float64) error {
+		if val < 0 {
+			return fmt.Errorf("%s must be non-negative, got %d", name, int64(val))
+		}
+		if val > math.MaxInt64-1 {
+			return fmt.Errorf("%s exceeds int64 max limit: %d", name, int64(val))
+		}
+		return nil
+	}
+
+	verify := func(update map[string]interface{}) error {
+		if update["cpu_cores"].(float64) < 0 || update["cpu_cores"].(float64) > math.MaxInt-1 {
+			return fmt.Errorf("Cpu.Cores be not a valid int64 number: %d", update["cpu_cores"])
+		}
+		if err := checkInt64("Ram.Total", update["mem_total"].(float64)); err != nil {
+			return err
+		}
+		if err := checkInt64("Swap.Total", update["swap_total"].(float64)); err != nil {
+			return err
+		}
+		if err := checkInt64("Disk.Total", update["disk_total"].(float64)); err != nil {
+			return err
+		}
+		return nil
+	}
+
+	if err := verify(update); err != nil {
+		return err
+	}
+
 	err := db.Model(&models.Client{}).Where("uuid = ?", clientUUID).Updates(update).Error
 	if err != nil {
 		return err

database/clients/report.go

@@ -3,6 +3,7 @@ package clients
 import (
 	"encoding/json"
 	"fmt"
+	"math"
 	"time"
 
 	"github.com/komari-monitor/komari/common"
@@ -51,10 +52,98 @@ func ParseReport(data map[string]interface{}) (report common.Report, err error)
 	return report, nil
 }
 
+// 检查数据防止异常数据导致数据库损坏
+func ReportVerify(report common.Report) error {
+	// 防止输入不合理范围
+	if report.CPU.Usage < 0 || report.CPU.Usage > 100 {
+		return fmt.Errorf("CPU.Usage must be between 0 and 100")
+	}
+
+	if report.Load.Load1 < 0 || report.Load.Load1 > 1000 {
+		return fmt.Errorf("Load.Load1 must be non-negative, got %.2f", report.Load.Load1)
+	}
+
+	checkFloat64 := func(name string, val float64) error {
+		if val > math.MaxFloat64-1 || val < -math.MaxFloat64+1 {
+			return fmt.Errorf("%s value exceeds float64 range: %g", name, val)
+		}
+		return nil
+	}
+
+	// [float64] 防止数据溢出
+	if err := checkFloat64("CPU.Usage", report.CPU.Usage); err != nil {
+		return err
+	}
+	if err := checkFloat64("Load.Load1", report.Load.Load1); err != nil {
+		return err
+	}
+
+	checkInt64 := func(name string, val int64) error {
+		if val < 0 {
+			return fmt.Errorf("%s must be non-negative, got %d", name, val)
+		}
+		if val > math.MaxInt64-1 {
+			return fmt.Errorf("%s exceeds int64 max limit: %d", name, val)
+		}
+		return nil
+	}
+
+	// [int64] 防止数据溢出
+	// Ram 验证
+	if err := checkInt64("Ram.Used", report.Ram.Used); err != nil {
+		return err
+	}
+	if err := checkInt64("Ram.Total", report.Ram.Total); err != nil {
+		return err
+	}
+	// Swap 验证
+	if err := checkInt64("Swap.Used", report.Swap.Used); err != nil {
+		return err
+	}
+	if err := checkInt64("Swap.Total", report.Swap.Total); err != nil {
+		return err
+	}
+	// Disk 验证
+	if err := checkInt64("Disk.Used", report.Disk.Used); err != nil {
+		return err
+	}
+	if err := checkInt64("Disk.Total", report.Disk.Total); err != nil {
+		return err
+	}
+	// Network 验证
+	if err := checkInt64("Network.Up", report.Network.Up); err != nil {
+		return err
+	}
+	if err := checkInt64("Network.Down", report.Network.Down); err != nil {
+		return err
+	}
+	if err := checkInt64("Network.TotalUp", report.Network.TotalUp); err != nil {
+		return err
+	}
+	if err := checkInt64("Network.TotalDown", report.Network.TotalDown); err != nil {
+		return err
+	}
+	// 拒绝所有负数Int
+	if report.Process < 0 {
+		return fmt.Errorf("Process must be non-negative: %d", report.Process)
+	}
+	if report.Connections.TCP < 0 {
+		return fmt.Errorf("Connections.TCP must be non-negative: %d", report.Connections.TCP)
+	}
+	if report.Connections.UDP < 0 {
+		return fmt.Errorf("Connections.UDP must be non-negative: %d", report.Connections.UDP)
+	}
+	return nil
+}
+
 // SaveClientReport 保存客户端报告到 Record 表
 func SaveClientReport(clientUUID string, report common.Report) (err error) {
 	db := dbcore.GetDBInstance()
 
+	if err := ReportVerify(report); err != nil {
+		return fmt.Errorf("failed to save Record: %v", err)
+	}
+
 	Record := models.Record{
 		Client:         clientUUID,
 		Time:           models.FromTime(time.Now()),

database/dbcore/dbcore.go

@@ -264,6 +264,7 @@ func GetDBInstance() *gorm.DB {
 			&models.PingTask{},
 			&models.OidcProvider{},
 			&models.MessageSenderProvider{},
+			&models.ThemeConfiguration{},
 		)
 		if err != nil {
 			log.Fatalf("Failed to create tables: %v", err)

database/models/config.go

@@ -20,10 +20,12 @@ type Config struct {
 	CustomHead string `json:"custom_head" gorm:"type:longtext"`
 	CustomBody string `json:"custom_body" gorm:"type:longtext"`
 	// 通知
-	NotificationEnabled        bool   `json:"notification_enabled" gorm:"default:false"` // 通知总开关
-	NotificationMethod         string `json:"notification_method" gorm:"type:varchar(64);default:'none'"`
-	ExpireNotificationEnabled  bool   `json:"expire_notification_enabled" gorm:"default:false"` // 是否启用过期通知
-	ExpireNotificationLeadDays int    `json:"expire_notification_lead_days" gorm:"default:7"`   // 过期前多少天通知，默认7天
+	NotificationEnabled        bool    `json:"notification_enabled" gorm:"default:false"` // 通知总开关
+	NotificationMethod         string  `json:"notification_method" gorm:"type:varchar(64);default:'none'"`
+	ExpireNotificationEnabled  bool    `json:"expire_notification_enabled" gorm:"default:false"` // 是否启用过期通知
+	ExpireNotificationLeadDays int     `json:"expire_notification_lead_days" gorm:"default:7"`   // 过期前多少天通知，默认7天
+	LoginNotification          bool    `json:"login_notification" gorm:"default:false"`          // 登录通知
+	TrafficLimitPercentage     float64 `json:"traffic_limit_percentage" gorm:"default:80.00"`    // 流量限制百分比，默认80.00%
 	// Record
 	RecordEnabled          bool `json:"record_enabled" gorm:"default:true"`          // 是否启用记录功能
 	RecordPreserveTime     int  `json:"record_preserve_time" gorm:"default:720"`     // 记录保留时间，单位小时，默认30天

database/models/models.go

@@ -8,35 +8,38 @@ import (
 
 // Client represents a registered client device
 type Client struct {
-	UUID           string    `json:"uuid,omitempty" gorm:"type:varchar(36);primaryKey"`
-	Token          string    `json:"token,omitempty" gorm:"type:varchar(255);unique;not null"`
-	Name           string    `json:"name" gorm:"type:varchar(100)"`
-	CpuName        string    `json:"cpu_name" gorm:"type:varchar(100)"`
-	Virtualization string    `json:"virtualization" gorm:"type:varchar(50)"`
-	Arch           string    `json:"arch" gorm:"type:varchar(50)"`
-	CpuCores       int       `json:"cpu_cores" gorm:"type:int"`
-	OS             string    `json:"os" gorm:"type:varchar(100)"`
-	KernelVersion  string    `json:"kernel_version" gorm:"type:varchar(100)"`
-	GpuName        string    `json:"gpu_name" gorm:"type:varchar(100)"`
-	IPv4           string    `json:"ipv4,omitempty" gorm:"type:varchar(100)"`
-	IPv6           string    `json:"ipv6,omitempty" gorm:"type:varchar(100)"`
-	Region         string    `json:"region" gorm:"type:varchar(100)"`
-	Remark         string    `json:"remark,omitempty" gorm:"type:longtext"`
-	PublicRemark   string    `json:"public_remark,omitempty" gorm:"type:longtext"`
-	MemTotal       int64     `json:"mem_total" gorm:"type:bigint"`
-	SwapTotal      int64     `json:"swap_total" gorm:"type:bigint"`
-	DiskTotal      int64     `json:"disk_total" gorm:"type:bigint"`
-	Version        string    `json:"version,omitempty" gorm:"type:varchar(100)"`
-	Weight         int       `json:"weight" gorm:"type:int"`
-	Price          float64   `json:"price"`
-	BillingCycle   int       `json:"billing_cycle"`
-	AutoRenewal    bool      `json:"auto_renewal" gorm:"default:false"` // 是否自动续费
-	Currency       string    `json:"currency" gorm:"type:varchar(20);default:'$'"`
-	ExpiredAt      LocalTime `json:"expired_at" gorm:"type:timestamp"`
-	Group          string    `json:"group" gorm:"type:varchar(100)"`
-	Tags           string    `json:"tags" gorm:"type:text"` // split by ';'
-	CreatedAt      LocalTime `json:"created_at"`
-	UpdatedAt      LocalTime `json:"updated_at"`
+	UUID             string    `json:"uuid,omitempty" gorm:"type:varchar(36);primaryKey"`
+	Token            string    `json:"token,omitempty" gorm:"type:varchar(255);unique;not null"`
+	Name             string    `json:"name" gorm:"type:varchar(100)"`
+	CpuName          string    `json:"cpu_name" gorm:"type:varchar(100)"`
+	Virtualization   string    `json:"virtualization" gorm:"type:varchar(50)"`
+	Arch             string    `json:"arch" gorm:"type:varchar(50)"`
+	CpuCores         int       `json:"cpu_cores" gorm:"type:int"`
+	OS               string    `json:"os" gorm:"type:varchar(100)"`
+	KernelVersion    string    `json:"kernel_version" gorm:"type:varchar(100)"`
+	GpuName          string    `json:"gpu_name" gorm:"type:varchar(100)"`
+	IPv4             string    `json:"ipv4,omitempty" gorm:"type:varchar(100)"`
+	IPv6             string    `json:"ipv6,omitempty" gorm:"type:varchar(100)"`
+	Region           string    `json:"region" gorm:"type:varchar(100)"`
+	Remark           string    `json:"remark,omitempty" gorm:"type:longtext"`
+	PublicRemark     string    `json:"public_remark,omitempty" gorm:"type:longtext"`
+	MemTotal         int64     `json:"mem_total" gorm:"type:bigint"`
+	SwapTotal        int64     `json:"swap_total" gorm:"type:bigint"`
+	DiskTotal        int64     `json:"disk_total" gorm:"type:bigint"`
+	Version          string    `json:"version,omitempty" gorm:"type:varchar(100)"`
+	Weight           int       `json:"weight" gorm:"type:int"`
+	Price            float64   `json:"price"`
+	BillingCycle     int       `json:"billing_cycle"`
+	AutoRenewal      bool      `json:"auto_renewal" gorm:"default:false"` // 是否自动续费
+	Currency         string    `json:"currency" gorm:"type:varchar(20);default:'$'"`
+	ExpiredAt        LocalTime `json:"expired_at" gorm:"type:timestamp"`
+	Group            string    `json:"group" gorm:"type:varchar(100)"`
+	Tags             string    `json:"tags" gorm:"type:text"` // split by ';'
+	Hidden           bool      `json:"hidden" gorm:"default:false"`
+	TrafficLimit     int64     `json:"traffic_limit" gorm:"type:bigint"`
+	TrafficLimitType string    `json:"traffic_limit_type" gorm:"type:varchar(10);default:'max'"` // 流量阈值类型：sum max min up down
+	CreatedAt        LocalTime `json:"created_at"`
+	UpdatedAt        LocalTime `json:"updated_at"`
 }
 
 // User represents an authenticated user

database/models/theme.go

@@ -2,11 +2,34 @@ package models
 
 // Theme represents a komari theme information
 type Theme struct {
-	Name        string `json:"name"`        // 主题名称
-	Short       string `json:"short"`       // 短名称，用作文件夹名
-	Description string `json:"description"` // 主题描述
-	Version     string `json:"version"`     // 版本号
-	Author      string `json:"author"`      // 作者
-	URL         string `json:"url"`         // 主题URL
-	Preview     string `json:"preview"`     // 预览图片相对路径
+	Name          string        `json:"name"`          // 主题名称
+	Short         string        `json:"short"`         // 短名称，用作文件夹名
+	Description   string        `json:"description"`   // 主题描述
+	Version       string        `json:"version"`       // 版本号
+	Author        string        `json:"author"`        // 作者
+	URL           string        `json:"url"`           // 主题URL
+	Preview       string        `json:"preview"`       // 预览图片相对路径
+	Configuration Configuration `json:"configuration"` // 声明配置项
+}
+
+type Configuration struct {
+	Type string `json:"type"` // managed
+	Icon string `json:"icon"` // 图标
+	Name string `json:"name"`
+	Data any    `json:"data"` // 配置数据
+}
+
+type ManagedThemeConfigurationItem struct {
+	Key      string `json:"key"`
+	Name     string `json:"name"`
+	Required bool   `json:"required"`
+	Type     string `json:"type"` // string number select switch title
+	Options  string `json:"options"`
+	Default  any    `json:"default"`
+	Help     string `json:"help"`
+}
+
+type ThemeConfiguration struct {
+	Short string `json:"short" gorm:"primaryKey;unique;not null"`
+	Data  string `json:"data" gorm:"type:longtext" default:"{}"`
 }

utils/messageSender/webhook/webhook.go

@@ -111,7 +111,7 @@ func (w *WebhookSender) createPOSTRequest(message, title string) (*http.Request,
 }
 
 func (w *WebhookSender) createGETRequest(message, title string) (*http.Request, error) {
-	URL := w.replaceTemplate(w.Addition.Body, message, title)
+	URL := w.replaceTemplate(w.Addition.URL, message, title)
 	u, err := url.Parse(URL)
 	if err != nil {
 		return nil, fmt.Errorf("invalid URL: %v", err)

utils/notifier/offline.go

@@ -20,6 +20,8 @@ type notificationState struct {
 	mu                  sync.Mutex // 互斥锁，保护该客户端状态
 	pendingOfflineSince time.Time  // 客户端离线的时间。为零值表示客户端在线或已发送离线通知。
 	isFirstConnection   bool       // 标记是否为首次上线连接。
+	isConnExist         bool       // 标记是否存在连接
+	connectionID        int64      // 连接ID，用于区分不同的连接会话，防止竞态条件
 }
 
 // clientStates 使用 sync.Map 实现对客户端状态的并发访问。
@@ -52,7 +54,7 @@ func getOrInitState(clientID string) *notificationState {
 }
 
 // OfflineNotification 在启用通知且未在宽限期内发送的情况下，发送客户端离线通知。
-func OfflineNotification(clientID string) {
+func OfflineNotification(clientID string, endedConnectionID int64) {
 	client, err := clients.GetClientByUUID(clientID)
 	if err != nil {
 		return
@@ -73,7 +75,8 @@ func OfflineNotification(clientID string) {
 
 	state.mu.Lock()
 	// 如果已处于待通知状态，则不做处理。
-	if !state.pendingOfflineSince.IsZero() {
+	// 只有当离线事件来自当前的连接会话时，我们才认为它有效。
+	if !state.pendingOfflineSince.IsZero() || state.connectionID != endedConnectionID {
 		state.mu.Unlock()
 		return
 	}
@@ -82,7 +85,7 @@ func OfflineNotification(clientID string) {
 	state.mu.Unlock()
 
 	// 新建协程，等待宽限期后判断是否需要发送通知。
-	go func(startTime time.Time) {
+	go func(startTime time.Time, expectedConnectionID int64) {
 		time.Sleep(gracePeriod)
 
 		state.mu.Lock()
@@ -90,13 +93,16 @@ func OfflineNotification(clientID string) {
 
 		// 检查离线状态是否仍为本次协程启动时的状态。
 		// 若为零值，说明客户端已重连。
-		// 若时间不同，说明客户端重连后又断开，已启动新计时。
-		if state.pendingOfflineSince.IsZero() || !state.pendingOfflineSince.Equal(startTime) {
+		// 当前的 connectionID 是否还是我们触发离线时的那个ID。如果不是，说明客户端重连过，本次离线通知已失效。
+		if state.pendingOfflineSince.IsZero() || state.connectionID != expectedConnectionID {
+			log.Println("%s is reconnected new connID: %d, old connID: %d", clientID, state.connectionID, expectedConnectionID)
 			return
 		}
 
 		// 即将发送通知，重置待通知状态。
+		// 需要多一个boolean 是因为pendingOfflineSince在offline睡眠后才修改，可能导致online判断不对
 		state.pendingOfflineSince = time.Time{}
+		state.isConnExist = false
 
 		// Send notification
 		message := fmt.Sprintf("🔴%s is offline", client.Name)
@@ -111,11 +117,11 @@ func OfflineNotification(clientID string) {
 		if err := db.Model(&models.OfflineNotification{}).Where("client = ?", clientID).Update("last_notified", now).Error; err != nil {
 			log.Printf("Failed to update last_notified for client %s: %v", clientID, err)
 		}
-	}(now)
+	}(now, endedConnectionID)
 }
 
 // OnlineNotification 在启用通知的情况下，发送客户端上线通知。
-func OnlineNotification(clientID string) {
+func OnlineNotification(clientID string, connectionID int64) {
 	client, err := clients.GetClientByUUID(clientID)
 	if err != nil {
 		return
@@ -131,12 +137,14 @@ func OnlineNotification(clientID string) {
 
 	state.mu.Lock()
 	defer state.mu.Unlock()
+	state.connectionID = connectionID
 
 	// 规则1：首次连接不通知。
 	if state.isFirstConnection {
 		state.isFirstConnection = false
 		// 同时清除任何待离线状态（如服务器重启时客户端本已离线）
 		state.pendingOfflineSince = time.Time{}
+		state.isConnExist = true
 		return
 	}
 
@@ -145,12 +153,21 @@ func OnlineNotification(clientID string) {
 	// 上线时总是清除待离线状态。
 	state.pendingOfflineSince = time.Time{}
 
-	// 规则4：宽限期内重连，不通知。
+	// 规则2：宽限期内重连，不通知。
 	if wasPending {
 		return
 	}
 
-	// 规则3：客户端离线足够久已通知（或未待离线），现在重新上线，发送上线通知。
+	// 规则3: 没断开后重连, 不通知
+	// 为了解决OfflineNotify中不是全程加锁
+	if state.isConnExist {
+		log.Println("%s has connection exist: %d", clientID, connectionID)
+		return
+	} else {
+		state.isConnExist = true
+	}
+
+	// 规则4：客户端离线足够久已通知（或未待离线），现在重新上线，发送上线通知。
 	message := fmt.Sprintf("🟢%s is online", client.Name)
 	go func(msg string) {
 		if err := messageSender.SendTextMessage(msg, "Komari Online Notification"); err != nil {

utils/notifier/traffic.go

@@ -0,0 +1,137 @@
+package notifier
+
+import (
+	"fmt"
+	"math"
+	"strings"
+	"time"
+
+	"github.com/komari-monitor/komari/database/clients"
+	"github.com/komari-monitor/komari/database/config"
+	"github.com/komari-monitor/komari/utils/messageSender"
+	"github.com/komari-monitor/komari/ws"
+	cache "github.com/patrickmn/go-cache"
+)
+
+// trafficCache 用于记录每个客户端已触发的阈值步进，避免重复提醒
+// key: "traffic:"+clientUUID, value: int 步进百分比（例如 80, 85, 90 ... 100）
+var trafficCache = cache.New(30*24*time.Hour, time.Hour) // 30天缓存，1小时清理
+
+// CheckTraffic 检查各客户端流量使用情况，并在达到阈值和每+5%时提醒一次；100%时额外提醒一次
+// 由外部协程每分钟调用一次
+func CheckTraffic() {
+	// 获取最新上报与客户端配置
+	reports := ws.GetLatestReport()
+	if len(reports) == 0 {
+		return
+	}
+
+	cfg, err := config.Get()
+	if err != nil {
+		return
+	}
+
+	if cfg.TrafficLimitPercentage <= 0 {
+		return
+	}
+
+	// 起始阈值：例如 80%，非5的倍数则从上取整到最近的5的倍数，例如 83->85
+	startThreshold := cfg.TrafficLimitPercentage
+	if startThreshold < 0 {
+		startThreshold = 0
+	}
+	baseStep := int(math.Ceil(startThreshold/5.0) * 5.0)
+	if baseStep > 100 {
+		baseStep = 100
+	}
+
+	allClients, err := clients.GetAllClientBasicInfo()
+	if err != nil {
+		return
+	}
+
+	for _, c := range allClients {
+		if c.TrafficLimit <= 0 {
+			continue
+		}
+
+		r, ok := reports[c.UUID]
+		if !ok || r == nil {
+			continue
+		}
+
+		// 计算不同类型的使用值
+		used := computeUsedByType(strings.ToLower(c.TrafficLimitType), r.Network.TotalUp, r.Network.TotalDown)
+		if used <= 0 {
+			continue
+		}
+
+		pct := float64(used) / float64(c.TrafficLimit) * 100.0
+		if pct < startThreshold {
+			continue
+		}
+
+		// 当前所在阈值步进（5%的倍数）
+		curStep := int(math.Floor(pct/5.0) * 5.0)
+		if curStep < baseStep {
+			curStep = baseStep
+		}
+		// if curStep > 100 {
+		// 	curStep = 100
+		// }
+
+		key := "traffic:" + c.UUID
+		last, _ := trafficCache.Get(key)
+		lastStep, _ := last.(int)
+
+		if curStep > lastStep { // 只在进入新步进时提醒一次
+			trafficCache.SetDefault(key, curStep)
+
+			title := "Komari Traffic Alert"
+			msg := fmt.Sprintf("%s traffic used %d%% (%s / %s), type=%s", c.Name, curStep, humanBytes(used), humanBytes(c.TrafficLimit), strings.ToLower(c.TrafficLimitType))
+			// 发送通知（内部会检查 NotificationEnabled）
+			_ = messageSender.SendTextMessage(msg, title)
+		}
+	}
+}
+
+func computeUsedByType(t string, up, down int64) int64 {
+	switch t {
+	case "up":
+		return up
+	case "down":
+		return down
+	case "sum":
+		return up + down
+	case "min":
+		if up < down {
+			return up
+		}
+		return down
+	case "max":
+		fallthrough
+	default:
+		if up > down {
+			return up
+		}
+		return down
+	}
+}
+
+func humanBytes(b int64) string {
+	const unit = 1024
+	if b < unit {
+		return fmt.Sprintf("%d B", b)
+	}
+	div, exp := int64(unit), 0
+	for n := b / unit; n >= unit; n /= unit {
+		div *= unit
+		exp++
+	}
+	// KMGTPE
+	prefixes := []string{"K", "M", "G", "T", "P", "E"}
+	if exp >= len(prefixes) {
+		exp = len(prefixes) - 1
+	}
+	return fmt.Sprintf("%.2f %sB", float64(b)/float64(div), prefixes[exp])
+}

utils/pingSchedule.go

@@ -1,6 +1,7 @@
 package utils
 
 import (
+	"context"
 	"sync"
 	"time"
 
@@ -10,16 +11,13 @@ import (
 
 // PingTaskManager 管理定时器和任务
 type PingTaskManager struct {
-	mu       sync.Mutex
-	tickers  map[int]*time.Ticker
-	tasks    map[int][]models.PingTask
-	stopChan chan struct{}
+	mu         sync.Mutex
+	cancelFunc context.CancelFunc
+	tasks      map[int][]models.PingTask
 }
 
 var manager = &PingTaskManager{
-	tickers:  make(map[int]*time.Ticker),
-	tasks:    make(map[int][]models.PingTask),
-	stopChan: make(chan struct{}),
+	tasks: make(map[int][]models.PingTask),
 }
 
 // Reload 重载时间表
@@ -27,60 +25,81 @@ func (m *PingTaskManager) Reload(pingTasks []models.PingTask) error {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
-	// 停止所有现有定时器
-	for _, ticker := range m.tickers {
-		ticker.Stop()
+	if m.cancelFunc != nil {
+		m.cancelFunc()
 	}
-	m.tickers = make(map[int]*time.Ticker)
+	ctx, cancel := context.WithCancel(context.Background())
+	m.cancelFunc = cancel
+
 	m.tasks = make(map[int][]models.PingTask)
 
 	// 按Interval分组任务
 	taskGroups := make(map[int][]models.PingTask)
 	for _, task := range pingTasks {
+		if task.Interval <= 0 {
+			continue
+		}
 		taskGroups[task.Interval] = append(taskGroups[task.Interval], task)
 	}
 
-	// 为每个唯一的Interval创建定时器
+	// 为每个唯一的Interval创建协程
 	for interval, tasks := range taskGroups {
-		ticker := time.NewTicker(time.Duration(interval) * time.Second)
-		m.tickers[interval] = ticker
 		m.tasks[interval] = tasks
-
-		go func(ticker *time.Ticker, tasks []models.PingTask) {
-			for {
-				select {
-				case <-ticker.C:
-					for _, task := range tasks {
-						go executePingTask(task)
-					}
-				case <-m.stopChan:
-					return
-				}
-			}
-		}(ticker, tasks)
+		go m.runPreciseLoop(ctx, time.Duration(interval)*time.Second, tasks)
 	}
-
 	return nil
 }
 
+func (m *PingTaskManager) runPreciseLoop(ctx context.Context, interval time.Duration, tasks []models.PingTask) {
+	// Start the first timer.
+	timer := time.NewTimer(interval)
+
+	// This will be the reference point for all future ticks.
+	// By adding the interval to this time, we avoid accumulating execution delays.
+	nextTick := time.Now().Add(interval)
+	defer timer.Stop()
+
+	for {
+		select {
+		case <-timer.C:
+			onlineClients := ws.GetConnectedClients()
+			for _, task := range tasks {
+				go executePingTask(ctx, task, onlineClients)
+			}
+
+			nextTick = nextTick.Add(interval)
+			timer.Reset(time.Until(nextTick))
+
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
 // executePingTask 执行单个PingTask
-func executePingTask(task models.PingTask) {
-	clients := ws.GetConnectedClients()
+func executePingTask(ctx context.Context, task models.PingTask, onlineClients map[string]*ws.SafeConn) {
 	var message struct {
 		TaskID  uint   `json:"ping_task_id"`
 		Message string `json:"message"`
 		Type    string `json:"ping_type"`
 		Target  string `json:"ping_target"`
 	}
+
+	message.Message = "ping"
+	message.TaskID = task.Id
+	message.Type = task.Type
+	message.Target = task.Target
+
 	for _, clientUUID := range task.Clients {
-		if conn, exists := clients[clientUUID]; exists {
-			if conn == nil {
-				continue
-			}
-			message.Message = "ping"
-			message.TaskID = task.Id
-			message.Type = task.Type
-			message.Target = task.Target
+		select {
+		case <-ctx.Done():
+			// Context was canceled, stop sending pings.
+			return
+		default:
+			// Context is still active, continue.
+		}
+
+		if conn, exists := onlineClients[clientUUID]; exists && conn != nil {
 			if err := conn.WriteJSON(message); err != nil {
 				continue
 			}

ws/clients.go

@@ -1,67 +0,0 @@
-package ws
-
-import (
-	"net/http"
-
-	"github.com/gorilla/websocket"
-	"github.com/komari-monitor/komari/common"
-
-	"github.com/gin-gonic/gin"
-)
-
-func GetClients(c *gin.Context) {
-	// 升级到ws
-	if !websocket.IsWebSocketUpgrade(c.Request) {
-		c.JSON(http.StatusBadRequest, gin.H{"status": "error", "error": "Require WebSocket upgrade"})
-		return
-	}
-	upgrader := websocket.Upgrader{
-		CheckOrigin: CheckOrigin,
-	}
-	// Upgrade the HTTP connection to a WebSocket connection
-	conn, err := upgrader.Upgrade(c.Writer, c.Request, nil)
-	if err != nil {
-		c.JSON(http.StatusBadRequest, gin.H{"status": "error", "error": "Failed to upgrade to WebSocket"})
-		return
-	}
-	defer conn.Close()
-
-	// 请求
-	for {
-		var resp struct {
-			Online []string                  `json:"online"` // 已建立连接的客户端uuid列表
-			Data   map[string]*common.Report `json:"data"`   // 最后上报的数据
-		}
-		resp.Online = []string{}
-		_, data, err := conn.ReadMessage()
-		if err != nil {
-			//log.Println("Error reading message:", err)
-			return
-		}
-		message := string(data)
-
-		if message != "get" {
-			conn.WriteJSON(gin.H{"status": "error", "error": "Invalid message"})
-			continue
-		}
-		// 已建立连接的客户端uuid列表
-		for key := range GetConnectedClients() {
-			resp.Online = append(resp.Online, key)
-		}
-		// 清除UUID，简化报告单
-		resp.Data = GetLatestReport()
-		for _, report := range resp.Data {
-			report.UUID = ""
-		}
-		for _, report := range resp.Data {
-			if report.CPU.Usage == 0 {
-				report.CPU.Usage = 0.01
-			}
-		}
-		err = conn.WriteJSON(gin.H{"status": "success", "data": resp})
-		if err != nil {
-			return
-		}
-	}
-
-}

ws/existingConnection.go

@@ -29,12 +29,19 @@ func SetConnectedClients(uuid string, conn *SafeConn) {
 	defer mu.Unlock()
 	connectedClients[uuid] = conn
 }
+func DeleteClientConditionally(uuid string, connToRemove *SafeConn) {
+	mu.Lock()
+	defer mu.Unlock()
+
+	// 检查当前 map 里的 conn 是否就是要删除的这一个
+	if currentConn, exists := connectedClients[uuid]; exists && currentConn == connToRemove {
+		delete(connectedClients, uuid)
+	}
+}
 func DeleteConnectedClients(uuid string) {
 	mu.Lock()
 	defer mu.Unlock()
-	if conn, exists := connectedClients[uuid]; exists {
-		conn.Close()
-	}
+	// 只从 map 中删除，不再负责关闭连接
 	delete(connectedClients, uuid)
 }
 func GetLatestReport() map[string]*common.Report {

ws/safeConn.go

@@ -10,12 +10,14 @@ import (
 type SafeConn struct {
 	conn *websocket.Conn
 	mu   sync.Mutex
+	ID   int64
 }
 
 func NewSafeConn(conn *websocket.Conn) *SafeConn {
 	return &SafeConn{
 		conn: conn,
 		mu:   sync.Mutex{},
+		ID:   time.Now().UnixNano(),
 	}
 }