package singleton import ( _ "embed" "iter" "log" "maps" "slices" "sync" "time" "github.com/gin-gonic/gin" "github.com/patrickmn/go-cache" "gorm.io/gorm" "sigs.k8s.io/yaml" "github.com/nezhahq/nezha/model" "github.com/nezhahq/nezha/pkg/utils" ) var Version = "debug" var ( Cache *cache.Cache DB *gorm.DB Loc *time.Location FrontendTemplates []model.FrontendTemplate DashboardBootTime = uint64(time.Now().Unix()) ServerShared *ServerClass ServiceSentinelShared *ServiceSentinel DDNSShared *DDNSClass NotificationShared *NotificationClass NATShared *NATClass CronShared *CronClass // ServerTransferShared is initialized in LoadSingleton AFTER ServerShared // (so the in-memory pending index can write back into ServerShared.UserID // on transitions) and AFTER initUser (so PushIfOnline can read secrets // from UserInfoMap). ) //go:embed frontend-templates.yaml var frontendTemplatesYAML []byte func InitTimezoneAndCache() error { var err error Loc, err = time.LoadLocation(Conf.Location) if err != nil { return err } Cache = cache.New(5*time.Minute, 10*time.Minute) return nil } // LoadSingleton 加载子服务并执行 func LoadSingleton(bus chan<- *model.Service) (err error) { initI18n() // 加载本地化服务 initUser() // 加载用户ID绑定表 NATShared = NewNATClass() DDNSShared = NewDDNSClass() NotificationShared = NewNotificationClass() ServerShared = NewServerClass() CronShared = NewCronClass() ServerTransferShared = NewServerTransferClass() // 最后初始化 ServiceSentinel ServiceSentinelShared, err = NewServiceSentinel(bus) return } // InitFrontendTemplates 从内置文件中加载FrontendTemplates func InitFrontendTemplates() error { err := yaml.Unmarshal(frontendTemplatesYAML, &FrontendTemplates) if err != nil { return err } return nil } // InitDBFromPath 从给出的文件路径中加载数据库 func InitDBFromPath(path string) error { var err error DB, err = gorm.Open(openSQLiteDialector(path), &gorm.Config{ CreateBatchSize: 200, }) if err != nil { return err } if Conf.Debug { DB = DB.Debug() } err = DB.AutoMigrate(model.Server{}, model.User{}, model.ServerGroup{}, model.NotificationGroup{}, model.Notification{}, model.AlertRule{}, model.Service{}, model.NotificationGroupNotification{}, model.Cron{}, model.Transfer{}, model.ServerGroupServer{}, model.NAT{}, model.DDNSProfile{}, model.NotificationGroupNotification{}, model.WAF{}, model.Oauth2Bind{}, model.ServerTransfer{}, model.JWTSession{}, model.APIToken{}, model.MCPAuditLog{}) if err != nil { return err } // 旧 mcp:* scope 与 nezha:* 并行了一段时间,HasScope 通过别名让 mcp:fs:write // 静默扩到 REST nezha:server:write。统一命名后这里把残留旧 scope 一次性 // 归一化(或在仅剩危险旧 scope 时整张 PAT 删除),保证运行时不再依赖别名。 if rewritten, deleted, mErr := model.MigrateLegacyMCPScopes(DB); mErr != nil { log.Printf("NEZHA>> MigrateLegacyMCPScopes failed: %v", mErr) } else if rewritten > 0 || deleted > 0 { log.Printf("NEZHA>> Migrated legacy mcp:* api token scopes: rewritten=%d deleted=%d", rewritten, deleted) } return nil } // RecordTransferHourlyUsage 对流量记录进行打点 func RecordTransferHourlyUsage(servers ...*model.Server) { now := time.Now() nowTrimSeconds := time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), 0, 0, 0, now.Location()) var txs []model.Transfer var slist iter.Seq[*model.Server] if len(servers) > 0 { slist = slices.Values(servers) } else { slist = utils.Seq2To1(ServerShared.Range) } for server := range slist { _, _, deltaIn, deltaOut := server.TransferDeltaAndAdvance() tx := model.Transfer{ ServerID: server.ID, In: deltaIn, Out: deltaOut, } if tx.In == 0 && tx.Out == 0 { continue } tx.CreatedAt = nowTrimSeconds txs = append(txs, tx) } if len(txs) == 0 { return } log.Printf("NEZHA>> Saved traffic metrics to database. Affected %d row(s), Error: %v", len(txs), DB.Create(txs).Error) } func PersistTransfer(transfer model.Transfer) error { if transfer.In == 0 && transfer.Out == 0 { return nil } return DB.Create(&transfer).Error } // CleanMonitorHistory 清理流量记录(TSDB 有自己的保留策略) func CleanMonitorHistory() { // 清理已被删除的服务器的流量记录 DB.Unscoped().Delete(&model.Transfer{}, "server_id NOT IN (SELECT `id` FROM servers)") // 计算可清理流量记录的时长 var allServerKeep time.Time specialServerKeep := make(map[uint64]time.Time) var specialServerIDs []uint64 var alerts []model.AlertRule if err := DB.Find(&alerts).Error; err != nil { log.Printf("NEZHA>> Failed to load alert rules while cleaning transfer history: %v", err) return } for _, alert := range alerts { for _, rule := range alert.Rules { // 是不是流量记录规则 if !rule.IsTransferDurationRule() { continue } dataCouldRemoveBefore := rule.GetTransferDurationStart().UTC() // 判断规则影响的机器范围 if rule.Cover == model.RuleCoverAll { // 更新全局可以清理的数据点 if allServerKeep.IsZero() || allServerKeep.After(dataCouldRemoveBefore) { allServerKeep = dataCouldRemoveBefore } } else { // 更新特定机器可以清理数据点 for id := range rule.Ignore { if specialServerKeep[id].IsZero() || specialServerKeep[id].After(dataCouldRemoveBefore) { specialServerKeep[id] = dataCouldRemoveBefore specialServerIDs = append(specialServerIDs, id) } } } } } for id, couldRemove := range specialServerKeep { DB.Unscoped().Delete(&model.Transfer{}, "server_id = ? AND datetime(`created_at`) < datetime(?)", id, couldRemove) } if len(specialServerIDs) == 0 { if allServerKeep.IsZero() { DB.Unscoped().Session(&gorm.Session{AllowGlobalUpdate: true}).Delete(&model.Transfer{}) } else { DB.Unscoped().Delete(&model.Transfer{}, "datetime(`created_at`) < datetime(?)", allServerKeep) } return } if allServerKeep.IsZero() { DB.Unscoped().Delete(&model.Transfer{}, "server_id NOT IN (?)", specialServerIDs) } else { DB.Unscoped().Delete(&model.Transfer{}, "server_id NOT IN (?) AND datetime(`created_at`) < datetime(?)", specialServerIDs, allServerKeep) } } // PerformMaintenance 执行系统维护(SQLite VACUUM 和 TSDB 维护) func PerformMaintenance() { log.Println("NEZHA>> Starting system maintenance...") // 1. SQLite 维护 if DB != nil { log.Println("NEZHA>> SQLite: Starting VACUUM...") if err := DB.Exec("VACUUM").Error; err != nil { log.Printf("NEZHA>> SQLite: VACUUM failed: %v", err) } else { log.Println("NEZHA>> SQLite: VACUUM completed") } } // 2. TSDB 维护 if TSDBEnabled() { TSDBShared.Maintenance() } log.Println("NEZHA>> System maintenance completed") } // IPDesensitize 根据设置选择是否对IP进行打码处理 返回处理后的IP(关闭打码则返回原IP) func IPDesensitize(ip string) string { if Conf.EnablePlainIPInNotification { return ip } return utils.IPDesensitize(ip) } type class[K comparable, V model.CommonInterface] struct { list map[K]V listMu sync.RWMutex sortedList []V sortedListMu sync.RWMutex } func (c *class[K, V]) Get(id K) (s V, ok bool) { c.listMu.RLock() defer c.listMu.RUnlock() s, ok = c.list[id] return } func (c *class[K, V]) GetList() map[K]V { c.listMu.RLock() defer c.listMu.RUnlock() return maps.Clone(c.list) } func (c *class[K, V]) GetSortedList() []V { c.sortedListMu.RLock() defer c.sortedListMu.RUnlock() return slices.Clone(c.sortedList) } func (c *class[K, V]) Range(fn func(k K, v V) bool) { c.listMu.RLock() defer c.listMu.RUnlock() for k, v := range c.list { if !fn(k, v) { break } } } func (c *class[K, V]) CheckPermission(ctx *gin.Context, idList iter.Seq[K]) bool { c.listMu.RLock() defer c.listMu.RUnlock() for id := range idList { if s, ok := c.list[id]; ok { if !s.HasPermission(ctx) { return false } } } return true }