LokiStack/backend/repository/db.go

package repository

import (
	"encoding/json"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"lokistack/config"
	"lokistack/models"
	"lokistack/pkg/metrics"
)

// saveRequest 表示一个保存请求，通过channel序列化
type saveRequest struct {
	done chan error // 用于同步等待保存完成
}

// JsonDB 使用基于channel的序列化保存机制，防止并发数据丢失。
// 该包只负责持久化（落盘 + 并发安全），领域实体定义在 models 包。
type JsonDB struct {
	mu              sync.RWMutex
	dataFile        string
	Users           []models.User           `json:"users"`
	Clusters        []models.Cluster        `json:"clusters"`
	Hosts           []models.Host           `json:"hosts"`
	Storages        []models.StorageConfig  `json:"storages"`
	Deployments     []models.LokiDeployment `json:"deployments"`
	Collectors      []models.Collector      `json:"collectors"`
	Logs            []models.DeploymentLog  `json:"logs"`
	Settings        []models.SystemSetting  `json:"settings"`
	LogSources      []models.LogSource      `json:"logSources"`
	AlertRules      []models.AlertRule      `json:"alertRules"`
	AlertDeliveries []models.AlertDelivery  `json:"alertDeliveries"`
	NotifyChannels  []models.NotifyChannel  `json:"notifyChannels"`
	LogPolicyGroups []models.LogPolicyGroup `json:"logPolicyGroups"`
	// AuditLogs 操作审计日志（保留期 90 天 / 上限 2 万条，见 services.auditRetention）
	AuditLogs []models.AuditLog `json:"auditLogs"`
	// DailyVolumeSnapshots 每日日志量历史（Loki retention 只有 24h，历史必须自己留）
	DailyVolumeSnapshots []models.DailyVolumeSnapshot `json:"dailyVolumeSnapshots"`

	// SchemaVersion 数据结构版本号。
	// 之前加字段全靠 JSON 反序列化的零值兜底，字段一多必然出事
	// （例如新增 ScrapeJobs 时老数据要手写兼容）。有了版本号，
	// 以后的结构变更可以在 InitDB 里按版本号顺序迁移，可测、可回滚。
	SchemaVersion int `json:"schemaVersion"`
	sequences     map[string]uint

	// 防抖保存机制
	saveTimer *time.Timer
	saveMu    sync.Mutex
	saveChan  chan saveRequest
	stopChan  chan struct{}
	wg        sync.WaitGroup
}

var db *JsonDB

func GetDB() *JsonDB {
	return db
}

// alertThresholdUpgrade 记录"旧种子阈值 → 新基线"的映射。
//
// 只在规则阈值**仍等于旧默认值**时才升级，避免覆盖用户自己调过的值。
type alertThresholdUpgrade struct {
	oldName      string
	newName      string
	oldThreshold int
	newThreshold int
	newCooldown  int
}

var alertThresholdUpgrades = []alertThresholdUpgrade{
	{oldName: "致命日志 FATAL", newName: "致命日志 FATAL", oldThreshold: 0, newThreshold: 10, newCooldown: 120},
	{oldName: "Go panic", newName: "Go panic", oldThreshold: 0, newThreshold: 10, newCooldown: 120},
	{oldName: "JVM OutOfMemoryError", newName: "JVM OutOfMemoryError", oldThreshold: 0, newThreshold: 5, newCooldown: 120},
	{oldName: "StackOverflowError", newName: "StackOverflowError", oldThreshold: 0, newThreshold: 5, newCooldown: 120},
	{oldName: "应用错误过多(error > 1000)", newName: "应用错误激增(error > 5万)", oldThreshold: 1000, newThreshold: 50000, newCooldown: 120},
	{oldName: "ingress-nginx 5xx 错误", newName: "ingress-nginx 5xx 错误", oldThreshold: 100, newThreshold: 500, newCooldown: 120},
	{oldName: "ingress-nginx 4xx 激增", newName: "ingress-nginx 4xx 激增", oldThreshold: 1000, newThreshold: 5000, newCooldown: 120},
	{oldName: "数据库连接失败", newName: "数据库连接失败", oldThreshold: 50, newThreshold: 200, newCooldown: 120},
}

// raiseLegacyAlertThresholds 抬高仍是旧默认值的告警阈值与冷却时间。
func (d *JsonDB) raiseLegacyAlertThresholds() {
	byName := map[string]alertThresholdUpgrade{}
	for _, u := range alertThresholdUpgrades {
		byName[u.oldName] = u
	}
	for i := range d.AlertRules {
		rule := &d.AlertRules[i]
		u, ok := byName[rule.Name]
		if !ok {
			continue
		}
		if rule.Threshold != u.oldThreshold {
			// 用户已经手动调过，尊重用户设置
			continue
		}
		old := rule.Threshold
		rule.Threshold = u.newThreshold
		rule.Name = u.newName
		if rule.CooldownMin < u.newCooldown {
			rule.CooldownMin = u.newCooldown
		}
		log.Printf("[schema] 告警规则 %q 阈值 %d → %d（冷却 %dm），降低告警频率",
			u.newName, old, u.newThreshold, rule.CooldownMin)
	}
}

// removeOrphanManagedLogSources 清掉"部署已不存在、但数据源还在"的孤儿记录。
//
// 判据：Managed=true（平台托管）且没有任何同名部署。
// 连带：删掉指向它的查询授权策略组，并禁用绑定它的告警规则。
func (d *JsonDB) removeOrphanManagedLogSources() {
	alive := map[string]bool{}
	for _, dep := range d.Deployments {
		alive[dep.Name] = true
	}
	orphans := map[uint]bool{}
	kept := make([]models.LogSource, 0, len(d.LogSources))
	for _, src := range d.LogSources {
		if src.Managed && !alive[src.Name] {
			orphans[src.ID] = true
			log.Printf("[schema] 移除孤儿数据源 %q(id=%d)：对应部署已删除", src.Name, src.ID)
			continue
		}
		kept = append(kept, src)
	}
	if len(orphans) == 0 {
		return
	}
	d.LogSources = kept

	groups := make([]models.LogPolicyGroup, 0, len(d.LogPolicyGroups))
	for _, group := range d.LogPolicyGroups {
		if orphans[group.SourceID] {
			log.Printf("[schema] 移除指向孤儿数据源的策略组 %q", group.Name)
			continue
		}
		groups = append(groups, group)
	}
	d.LogPolicyGroups = groups

	for i := range d.AlertRules {
		if orphans[d.AlertRules[i].SourceID] {
			d.AlertRules[i].Enabled = false
			log.Printf("[schema] 告警规则 %q 的数据源已删除，规则已禁用", d.AlertRules[i].Name)
		}
	}
}

// linkRulesToDefaultChannel 给未关联通道的规则挂上默认通道。
//
// 没有通道时不凭空造（那是 seed 首次部署的职责）；这里只在已有通道时补齐关联。
func (d *JsonDB) linkRulesToDefaultChannel() {
	if len(d.NotifyChannels) == 0 {
		return
	}
	def := d.NotifyChannels[0].ID
	linked := 0
	for i := range d.AlertRules {
		if len(d.AlertRules[i].ChannelIDs) == 0 {
			d.AlertRules[i].ChannelIDs = []uint{def}
			linked++
		}
	}
	if linked > 0 {
		log.Printf("[schema] %d 条告警规则已关联默认通知通道 %q", linked, d.NotifyChannels[0].Name)
	}
}

// fixUnreachableLogSourceURLs 把数据源里 0.0.0.0 / [::] 的监听地址换成主机真实 IP。
//
// 平台进程不在被采集主机上，0.0.0.0 只在本机有意义，用它必然连不上。
func (d *JsonDB) fixUnreachableLogSourceURLs() {
	hostIPByID := map[uint]string{}
	for _, h := range d.Hosts {
		hostIPByID[h.ID] = strings.TrimSpace(h.HostIP)
	}
	// 数据源按名字对应部署名，用于推断该源属于哪台主机
	hostByName := map[string]string{}
	for _, dep := range d.Deployments {
		if dep.HostID == nil {
			continue
		}
		if ip, ok := hostIPByID[*dep.HostID]; ok && ip != "" {
			hostByName[dep.Name] = ip
		}
	}

	for i := range d.LogSources {
		src := &d.LogSources[i]
		if !strings.Contains(src.URL, "0.0.0.0") && !strings.Contains(src.URL, "[::]") {
			continue
		}
		ip := hostByName[src.Name]
		if ip == "" && len(d.Hosts) == 1 {
			// 只有一台主机时可以直接推断（常见于单机试用）
			ip = strings.TrimSpace(d.Hosts[0].HostIP)
		}
		if ip == "" {
			log.Printf("[schema] 数据源 %q 的 URL 是不可达的监听地址 %s，但无法推断主机 IP，请手动修正", src.Name, src.URL)
			continue
		}
		old := src.URL
		src.URL = strings.Replace(src.URL, "0.0.0.0", ip, 1)
		src.URL = strings.Replace(src.URL, "[::]", ip, 1)
		log.Printf("[schema] 数据源 %q URL 修正：%s → %s", src.Name, old, src.URL)
	}
}

// dedupeCollectionIDs 对所有带 ID 的集合做重复 ID 修复。
//
// 注意：实际迁移里按版本分步调用（见 migrateSchema），
// 不要直接用它替代分步逻辑 —— 那会让"已迁移库"漏掉后加的清理步骤。
//
// 背景：这些集合曾经都漏在 InitDB 的 sequences 重建之外，导致 NextID 每次
// 进程重启都从 1 开始，于是不同时期创建的对象会撞号。撞号后的表现各不相同：
//
//	LogSource   : 查询页数据源下拉选不动（两个 option 同 value）
//	AlertRule   : 规则列表出现两条同 id，编辑/删除操作命中错的那条
//	NotifyChannel: 通知渠道覆盖错对象
func (d *JsonDB) dedupeCollectionIDs() {
	d.dedupeLogSourceIDs()
	d.dedupeAlertRuleIDs()
	d.dedupeNotifyChannelIDs()
}

// dedupeAlertRuleIDs 重新编号重复的告警规则 ID。
// 同名规则（seed 重复执行导致）只保留第一条，其余删除。
func (d *JsonDB) dedupeAlertRuleIDs() {
	if len(d.AlertRules) == 0 {
		return
	}
	seenID := map[uint]bool{}
	seenName := map[string]bool{}
	next := uint(0)
	for _, r := range d.AlertRules {
		if r.ID > next {
			next = r.ID
		}
	}
	fixed := make([]models.AlertRule, 0, len(d.AlertRules))
	for i := range d.AlertRules {
		rule := d.AlertRules[i]
		if rule.Name != "" && seenName[rule.Name] {
			log.Printf("[schema] 删除重复的告警规则 %q（同名）", rule.Name)
			continue
		}
		seenName[rule.Name] = true
		if rule.ID == 0 || seenID[rule.ID] {
			next++
			log.Printf("[schema] 告警规则 %q 重复 ID %d → 重新分配为 %d", rule.Name, rule.ID, next)
			rule.ID = next
		}
		seenID[rule.ID] = true
		fixed = append(fixed, rule)
	}
	d.AlertRules = fixed
	for _, r := range d.AlertRules {
		if r.ID > d.sequences["alertRules"] {
			d.sequences["alertRules"] = r.ID
		}
	}
}

// dedupeNotifyChannelIDs 重新编号重复的通知渠道 ID。
func (d *JsonDB) dedupeNotifyChannelIDs() {
	if len(d.NotifyChannels) == 0 {
		return
	}
	seen := map[uint]bool{}
	next := uint(0)
	for _, c := range d.NotifyChannels {
		if c.ID > next {
			next = c.ID
		}
	}
	for i := range d.NotifyChannels {
		ch := &d.NotifyChannels[i]
		if ch.ID == 0 || seen[ch.ID] {
			next++
			log.Printf("[schema] 通知渠道 %q 重复 ID %d → 重新分配为 %d", ch.Name, ch.ID, next)
			ch.ID = next
		}
		seen[ch.ID] = true
		if ch.ID > d.sequences["notifyChannels"] {
			d.sequences["notifyChannels"] = ch.ID
		}
	}
}

// removeInfraAlertRules 删除误配置的「基础设施类」告警规则。
//
// 判据：查询指向 kube-state-metrics / event-exporter，或在日志里匹配
// 节点 NotReady / 容器 OOMKilled。这些是 Prometheus 指标监控的职责：
//   - 日志里有没有这行取决于采集器是否采到那些命名空间，不可靠
//   - 指标侧（kube_node_status_condition / kube_pod_container_status_terminated_reason）
//     才是权威数据源
func (d *JsonDB) removeInfraAlertRules() {
	infraMarkers := []string{"kube-state-metrics", "event-exporter"}
	infraNames := []string{"Kubernetes 节点 NotReady", "K8s 节点 NotReady", "容器 OOMKilled"}

	kept := make([]models.AlertRule, 0, len(d.AlertRules))
	for _, r := range d.AlertRules {
		isInfra := false
		for _, marker := range infraMarkers {
			if strings.Contains(r.Query, marker) {
				isInfra = true
			}
		}
		for _, name := range infraNames {
			if strings.TrimSpace(r.Name) == name {
				isInfra = true
			}
		}
		if isInfra {
			log.Printf("[schema] 移除基础设施类告警规则 %q（应交给 Prometheus）", r.Name)
			continue
		}
		kept = append(kept, r)
	}
	d.AlertRules = kept
}

// dedupeLogSourceIDs 修复历史遗留的重复 ID，并同步更新引用它的策略组。
//
// 策略组通过 SourceID 关联数据源。重复 ID 时无法判断分组到底指向哪一个，
// 这里按"保留第一个、其余顺延"处理，并让原本指向该 ID 的策略组继续指向
// 编号最小的那个（即重建前第一次创建的数据源）。
func (d *JsonDB) dedupeLogSourceIDs() {
	if len(d.LogSources) < 2 {
		return
	}
	seen := map[uint]bool{}
	remap := map[uint]uint{} // 旧ID → 新ID（仅记录需要改的）
	next := uint(0)
	for _, s := range d.LogSources {
		if s.ID > next {
			next = s.ID
		}
	}

	fixed := make([]models.LogSource, 0, len(d.LogSources))
	kept := map[uint]bool{} // 已被占用的 ID（首次出现时占位）
	for i := range d.LogSources {
		src := d.LogSources[i]
		if src.ID == 0 || kept[src.ID] {
			// 重复或非法 ID：分配新号
			next++
			old := src.ID
			src.ID = next
			if _, ok := remap[old]; !ok && old != 0 {
				// 同一个旧 ID 只记录一次映射（指向保留的那个）
				remap[old] = 0 // 0 表示"不需要改引用"
			}
			log.Printf("[schema] 数据源 %q 重复 ID %d → 重新分配为 %d", src.Name, old, src.ID)
		} else {
			kept[src.ID] = true
		}
		seen[src.ID] = true
		fixed = append(fixed, src)
	}
	d.LogSources = fixed

	// 同步策略组引用（只在旧 ID 已不存在时改写）
	for i := range d.LogPolicyGroups {
		sid := d.LogPolicyGroups[i].SourceID
		if sid == 0 || seen[sid] {
			continue
		}
		if nv, ok := remap[sid]; ok && nv != 0 {
			d.LogPolicyGroups[i].SourceID = nv
		}
	}

	// 迁移发生在 sequences 重建**之后**，重新分配可能抬高最大 ID，
	// 必须同步提升计数器，否则紧接着创建的数据源会撞号。
	for _, s := range d.LogSources {
		if s.ID > d.sequences["logSources"] {
			d.sequences["logSources"] = s.ID
		}
	}

	// 保证只有一个默认数据源
	defaultSeen := false
	for i := range d.LogSources {
		if d.LogSources[i].IsDefault {
			if defaultSeen {
				d.LogSources[i].IsDefault = false
			}
			defaultSeen = true
		}
	}
	if !defaultSeen && len(d.LogSources) > 0 {
		d.LogSources[0].IsDefault = true
	}
}

// CurrentSchemaVersion 当前数据结构版本。
// 每次做**不兼容**的结构变更时 +1，并在 migrateSchema 里补一段迁移逻辑。
//
//	v0/v1（无字段）→ 1：新增 auditLogs / schemaVersion（老库走零值兜底即可，无需处理）
//	v1 → 2：预留示例（如把 ScrapeJobs 从 logPath 拆分时，需按旧结构生成默认 job）
const CurrentSchemaVersion = 7

// migrateSchema 按版本号顺序迁移老库。返回是否发生了迁移（需要回写）。
func (d *JsonDB) migrateSchema() bool {
	if d.SchemaVersion >= CurrentSchemaVersion {
		return false
	}
	from := d.SchemaVersion

	// 逐版本递进，每段只处理"上一版 → 下一版"的差异，
	// 不要写成 if version < X 的叠加判断，否则分支会指数增长。
	for d.SchemaVersion < CurrentSchemaVersion {
		switch d.SchemaVersion {
		case 0:
			// v0 → v1：补审计表；老库没有 auditLogs 字段，反序列化后为 nil，
			// 这里显式初始化，避免后续 append 到 nil 之外还留着 nil 判断。
			if d.AuditLogs == nil {
				d.AuditLogs = []models.AuditLog{}
			}
			d.SchemaVersion = 1
		case 1:
			// v1 → v2：修复 LogSource 重复 ID（曾漏在 sequences 重建之外，
			// NextID 每次重启从 1 开始，线上出现两个数据源都是 id=1）。
			d.dedupeLogSourceIDs()
			d.SchemaVersion = 2
		case 2:
			// v2 → v3：修复 AlertRule / NotifyChannel 重复 ID + 清掉基础设施类规则。
			//
			// ⚠️ 教训：v2 那次我把"告警去重 + 清基础设施规则"塞进了 case 1，
			// 但当时线上库已迁移到 v2，于是这段代码永远不执行（DB 里 schemaVersion
			// 已经等于当前版本）。结构变更只要改了逻辑、且目标库已到该版本，
			// 就必须**再抬一版**，不能改历史迁移分支。
			d.dedupeAlertRuleIDs()
			d.dedupeNotifyChannelIDs()
			d.removeInfraAlertRules()
			d.SchemaVersion = 3
		case 3:
			// v3 → v4：修复主机模式数据源的不可达 URL。
			//
			// 主机模式部署的 endpoint 存的是容器监听地址 0.0.0.0:3100，
			// 平台把它原样当数据源 URL → 查询页对该数据源的所有请求
			// 都报 dial tcp 0.0.0.0:3100: connection refused，
			// 前端表现为「命名空间/App 下拉永远加载中」。
			d.fixUnreachableLogSourceURLs()
			d.SchemaVersion = 4
		case 4:
			// v4 → v5：抬高内置告警规则的默认阈值。
			//
			// 旧种子里有 4 条 threshold=0（命中一条就报），另外几条绝对值也偏低：
			// 实测某环境 15 分钟 3051 万条日志 → 告警永远在响，运维直接脱敏。
			// 只改"仍等于旧默认值"的规则，用户手动调过的阈值不动。
			d.raiseLegacyAlertThresholds()
			d.SchemaVersion = 5
		case 5:
			// v5 → v6：把"没有关联任何通道"的告警规则挂到默认通道。
			//
			// 之前内置规则 ChannelIDs 全是空的，规则命中后只能去回退旧的全局
			// notify-config；那个早就不用了，结果就是"规则在响但哪里都没发出去"。
			d.linkRulesToDefaultChannel()
			d.SchemaVersion = 6
		case 6:
			// v6 → v7：清理"孤儿数据源"。
			// 删除部署时曾漏掉对自动注册数据源的清理，导致查询页残留指向
			// 已删实例的数据源（线上残留过 host-verify-loki）。
			d.removeOrphanManagedLogSources()
			d.SchemaVersion = 7
		default:
			// 防御：出现未知版本（比当前代码还新）时不乱动，避免降级把数据写坏
			log.Printf("[schema] 数据库版本 %d 高于当前代码支持版本 %d，跳过迁移", d.SchemaVersion, CurrentSchemaVersion)
			return false
		}
	}
	log.Printf("[schema] 数据结构已从 v%d 迁移到 v%d", from, d.SchemaVersion)
	return true
}

func InitDB() error {
	cfg := config.GetConfig()
	_ = os.MkdirAll(cfg.DataDir, 0700)

	d := &JsonDB{
		dataFile:             cfg.DatabasePath,
		Users:                []models.User{},
		Clusters:             []models.Cluster{},
		Hosts:                []models.Host{},
		Storages:             []models.StorageConfig{},
		Deployments:          []models.LokiDeployment{},
		Collectors:           []models.Collector{},
		Logs:                 []models.DeploymentLog{},
		Settings:             []models.SystemSetting{},
		LogSources:           []models.LogSource{},
		AlertRules:           []models.AlertRule{},
		AlertDeliveries:      []models.AlertDelivery{},
		NotifyChannels:       []models.NotifyChannel{},
		LogPolicyGroups:      []models.LogPolicyGroup{},
		AuditLogs:            []models.AuditLog{},
		DailyVolumeSnapshots: []models.DailyVolumeSnapshot{},
		SchemaVersion:        CurrentSchemaVersion,
		sequences: map[string]uint{
			"users":                0,
			"clusters":             0,
			"hosts":                0,
			"storages":             0,
			"deployments":          0,
			"collectors":           0,
			"logs":                 0,
			"logPolicyGroups":      0,
			"logSources":           0,
			"notifyChannels":       0,
			"alertRules":           0,
			"alertDeliveries":      0,
			"auditLogs":            0,
			"dailyVolumeSnapshots": 0,
		},
		saveChan: make(chan saveRequest, 10),
		stopChan: make(chan struct{}),
	}

	if data, err := os.ReadFile(d.dataFile); err == nil {
		if err := json.Unmarshal(data, d); err == nil {
			d.sequences = map[string]uint{}
			for _, u := range d.Users {
				if u.ID > d.sequences["users"] {
					d.sequences["users"] = u.ID
				}
			}
			for _, c := range d.Clusters {
				if c.ID > d.sequences["clusters"] {
					d.sequences["clusters"] = c.ID
				}
			}
			for _, h := range d.Hosts {
				if h.ID > d.sequences["hosts"] {
					d.sequences["hosts"] = h.ID
				}
			}
			for _, s := range d.Storages {
				if s.ID > d.sequences["storages"] {
					d.sequences["storages"] = s.ID
				}
			}
			for _, de := range d.Deployments {
				if de.ID > d.sequences["deployments"] {
					d.sequences["deployments"] = de.ID
				}
			}
			for _, col := range d.Collectors {
				if col.ID > d.sequences["collectors"] {
					d.sequences["collectors"] = col.ID
				}
			}
			for _, l := range d.Logs {
				if l.ID > d.sequences["logs"] {
					d.sequences["logs"] = l.ID
				}
			}
			for _, group := range d.LogPolicyGroups {
				if group.ID > d.sequences["logPolicyGroups"] {
					d.sequences["logPolicyGroups"] = group.ID
				}
			}
			// ⚠️ 以下集合曾经漏在重建之外，导致 NextID 每次重启都从 1 开始，
			// 于是不同进程生命周期里创建的对象 ID 重复（线上实际出现过
			// 两个 LogSource 都是 id=1 → 前端数据源下拉无法切换）。
			// 以后新增任何带 ID 的集合，都必须在这里补一段。
			for _, s := range d.LogSources {
				if s.ID > d.sequences["logSources"] {
					d.sequences["logSources"] = s.ID
				}
			}
			for _, ch := range d.NotifyChannels {
				if ch.ID > d.sequences["notifyChannels"] {
					d.sequences["notifyChannels"] = ch.ID
				}
			}
			for _, rule := range d.AlertRules {
				if rule.ID > d.sequences["alertRules"] {
					d.sequences["alertRules"] = rule.ID
				}
			}
			for _, del := range d.AlertDeliveries {
				if del.ID > d.sequences["alertDeliveries"] {
					d.sequences["alertDeliveries"] = del.ID
				}
			}
			for _, a := range d.AuditLogs {
				if a.ID > d.sequences["auditLogs"] {
					d.sequences["auditLogs"] = a.ID
				}
			}
			for _, v := range d.DailyVolumeSnapshots {
				if v.ID > d.sequences["dailyVolumeSnapshots"] {
					d.sequences["dailyVolumeSnapshots"] = v.ID
				}
			}
		}
	}

	migratedLogAccess := migrateLogAccessToGroups(d)
	// 结构版本迁移（审计表、采集项等新字段的兼容处理）
	migratedSchema := d.migrateSchema()
	// 启动序列化保存goroutine
	d.wg.Add(1)
	go d.saveWorker()

	db = d
	if migratedLogAccess || migratedSchema {
		d.RequestSave()
	}
	return nil
}

// Legacy per-user grants are grouped by identical source and label scope.
func migrateLogAccessToGroups(d *JsonDB) bool {
	changed := false
	for i := range d.LogSources {
		source := &d.LogSources[i]
		if len(source.Access) == 0 {
			continue
		}
		byLabels := map[string]*models.LogPolicyGroup{}
		for _, rule := range source.Access {
			validUser := false
			for _, user := range d.Users {
				if user.ID == rule.UserID && user.Role != "admin" {
					validUser = true
					break
				}
			}
			if !validUser {
				continue
			}
			alreadyGrouped := false
			for _, group := range d.LogPolicyGroups {
				if group.SourceID != source.ID {
					continue
				}
				for _, id := range group.UserIDs {
					if id == rule.UserID {
						alreadyGrouped = true
						break
					}
				}
			}
			if alreadyGrouped {
				continue
			}
			keyBytes, _ := json.Marshal(rule.Labels)
			key := string(keyBytes)
			group := byLabels[key]
			if group == nil {
				d.sequences["logPolicyGroups"]++
				id := d.sequences["logPolicyGroups"]
				group = &models.LogPolicyGroup{ID: id, Name: fmt.Sprintf("旧授权 · %s · %d", source.Name, id), SourceID: source.ID, Labels: rule.Labels, CreatedAt: time.Now(), UpdatedAt: time.Now()}
				byLabels[key] = group
			}
			group.UserIDs = append(group.UserIDs, rule.UserID)
		}
		for _, group := range byLabels {
			d.LogPolicyGroups = append(d.LogPolicyGroups, *group)
		}
		source.Access = nil
		changed = true
	}
	return changed
}

// Shutdown 优雅关闭数据库，确保所有数据已落盘。
// 顺序很关键：先停防抖定时器（不再产生新请求）→ 关闭 worker 并等其退出
// （worker 可能正在处理已入队的请求）→ 此时已无并发，做最后一次 doSave 把
// 最新内存状态原子写入。注意绝不能在 worker 仍在运行时直接 Save()，否则
// 两个 doSave 并发改写同一文件会发生"后写覆盖先写"的丢更新。
func Shutdown() error {
	if db == nil {
		return nil
	}

	// 1. 取消防抖定时器，阻止新的保存请求进入 channel
	db.saveMu.Lock()
	if db.saveTimer != nil {
		db.saveTimer.Stop()
		db.saveTimer = nil
	}
	db.saveMu.Unlock()

	// 2. 通知 worker 停止并等待其排空 saveChan、完成进行中的 doSave 后退出
	close(db.stopChan)
	db.wg.Wait()

	// 3. worker 已退出，无并发，落盘最终状态
	if err := db.doSave(); err != nil {
		return fmt.Errorf("final save failed: %w", err)
	}
	return nil
}

// saveWorker 通过channel序列化所有保存操作，防止并发写入冲突
func (d *JsonDB) saveWorker() {
	defer d.wg.Done()
	for {
		select {
		case req, ok := <-d.saveChan:
			if !ok {
				return
			}
			err := d.doSave()
			if req.done != nil {
				req.done <- err
			}
		case <-d.stopChan:
			return
		}
	}
}

// doSave 执行实际的原子文件写入
func (d *JsonDB) doSave() error {
	start := time.Now()
	err := d.doSaveInner()
	// JSON DB 是全量写盘，落盘耗时是平台最主要的延迟来源之一，必须可观测
	metrics.ObserveDBSave(time.Since(start).Seconds(), err == nil)
	return err
}

func (d *JsonDB) doSaveInner() error {
	d.mu.RLock()
	data, err := json.MarshalIndent(d, "", "  ")
	d.mu.RUnlock()
	if err != nil {
		return err
	}

	// 原子写入：先写入临时文件，再重命名
	dir := filepath.Dir(d.dataFile)
	tmpFile := filepath.Join(dir, fmt.Sprintf(".lokistack-%d.tmp", time.Now().UnixNano()))

	if err := os.WriteFile(tmpFile, data, 0600); err != nil {
		return err
	}

	// 确保数据落盘
	f, err := os.Open(tmpFile)
	if err != nil {
		os.Remove(tmpFile)
		return err
	}
	_ = f.Sync()
	f.Close()

	if err := os.Rename(tmpFile, d.dataFile); err != nil {
		os.Remove(tmpFile)
		return err
	}
	metrics.DBSizeBytes.Set(float64(len(data)))

	return nil
}

// Save 同步保存数据库到文件（原子写入）
func (d *JsonDB) Save() error {
	return d.doSave()
}

// RequestSave 请求异步保存（防抖机制+序列化channel）
// 多次快速调用会在500ms内合并为一次保存
func (d *JsonDB) RequestSave() {
	d.saveMu.Lock()
	defer d.saveMu.Unlock()

	// 如果已有待触发的定时器，重置它（防抖）
	if d.saveTimer != nil {
		d.saveTimer.Stop()
	}

	d.saveTimer = time.AfterFunc(500*time.Millisecond, func() {
		d.saveMu.Lock()
		d.saveTimer = nil
		d.saveMu.Unlock()

		// 通过channel发送到保存worker，确保序列化
		select {
		case d.saveChan <- saveRequest{}:
		case <-d.stopChan:
		}
	})
}

// RequestSaveSync 请求同步保存，等待保存完成（用于shutdown场景）
func (d *JsonDB) RequestSaveSync() error {
	req := saveRequest{done: make(chan error, 1)}
	select {
	case d.saveChan <- req:
	case <-d.stopChan:
		return d.Save() // worker已停止，直接同步保存
	}
	return <-req.done
}

func (d *JsonDB) NextID(table string) uint {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.sequences[table]++
	return d.sequences[table]
}

func (d *JsonDB) Lock() {
	d.mu.Lock()
}

func (d *JsonDB) Unlock() {
	d.mu.Unlock()
}

func (d *JsonDB) RLock() {
	d.mu.RLock()
}

func (d *JsonDB) RUnlock() {
	d.mu.RUnlock()
}

// CreateUser 创建用户并触发防抖保存
func (d *JsonDB) CreateUser(user *models.User) error {
	d.mu.Lock()
	defer d.mu.Unlock()
	if user.ID == 0 {
		user.ID = d.sequences["users"] + 1
		d.sequences["users"] = user.ID
	}
	if user.CreatedAt.IsZero() {
		now := time.Now()
		user.CreatedAt = now
		user.UpdatedAt = now
	}
	d.Users = append(d.Users, *user)
	d.RequestSave()
	return nil
}

// RecoverOrphanDeployments 启动时把残留的 deploying 状态部署标记为失败:
// 部署 goroutine 随进程消亡,不恢复会永久卡在「部署中」(UI 无停止入口)。
func (d *JsonDB) RecoverOrphanDeployments() {
	d.Lock()
	defer d.Unlock()
	recovered := 0
	for i := range d.Deployments {
		if d.Deployments[i].Status == "deploying" {
			d.Deployments[i].Status = "failed"
			d.Deployments[i].StatusMessage = "服务重启导致部署中断，请重新部署"
			d.Deployments[i].UpdatedAt = time.Now()
			recovered++
		}
	}
	if recovered > 0 {
		d.RequestSave()
	}
}