// Package pipelinedb provides an integrated pipeline database system
// 集成了数据库存储和业务管道处理的一体化解决方案
package pipelinedb

import (
	"encoding/binary"
	"sync"
)

// GroupStatsCache 组统计信息缓存结构
//
// 核心功能：
// - 缓存每个组的统计信息，提供O(1)的统计查询性能
// - 支持按数据状态分类统计（Hot/Warm/Cold）
// - 实时更新，确保统计数据的准确性
//
// 设计考虑：
// - 使用整数类型提供高效的计数操作
// - 支持JSON序列化，便于数据交换和持久化
// - 结构简单，内存占用小
//
// 适用场景：
// - 数据库性能监控
// - 组级别的数据分析
// - 缓存命中率统计
type GroupStatsCache struct {
	TotalRecords int `json:"total_records"` // 总记录数（所有状态的记录总和）
	HotRecords   int `json:"hot_records"`   // 热数据记录数（最新、最活跃的数据）
	WarmRecords  int `json:"warm_records"`  // 温数据记录数（中等活跃度的数据）
	ColdRecords  int `json:"cold_records"`  // 冷数据记录数（较少访问的数据）
}

// GroupManager 数据组管理器
//
// 核心职责：
// 1. 管理数据组的生命周期（暂停/恢复）
// 2. 维护组级别的ID自增计数器
// 3. 缓存组统计信息，提供高性能统计查询
// 4. 管理组的OnComplete回调执行状态
// 5. 提供组状态的持久化存储
//
// 设计思想：
// - 使用读写锁支持高并发访问
// - 统计信息缓存在内存中，避免频繁的磁盘I/O
// - ID计数器持久化到专用页面，确保重启后的连续性
// - 状态管理支持复杂的业务流程控制
//
// 性能特征：
// - 统计查询：O(1) - 直接从内存缓存读取
// - ID生成：O(1) - 内存计数器自增
// - 状态检查：O(1) - HashMap查找
// - 并发友好：读多写少的场景下性能优异
//
// 适用场景：
// - 多租户数据库系统
// - 分组数据处理管道
// - 需要组级别控制的业务系统
type GroupManager struct {
	pausedGroups        map[string]bool             // 暂停状态的组集合（key=组名，value=true表示暂停）
	onCompleteExecuting map[string]bool             // 正在执行OnComplete回调的组集合
	groupCounters       map[string]*int64           // 每个组的ID自增计数器（持久化到磁盘）
	groupStats          map[string]*GroupStatsCache // 组统计信息缓存（内存中维护）
	mu                  sync.RWMutex                // 读写锁，保护所有内部状态的并发访问
	pdb                 *PipelineDB                 // 数据库实例引用，用于计数器持久化
}

// NewGroupManager 创建一个新的组管理器实例
//
// 核心功能：初始化组管理器的所有内部数据结构
// 时间复杂度：O(1) - 只创建空的数据结构
// 空间复杂度：O(1) - 初始状态下不占用额外空间
//
// 参数说明：
//   - pdb: PipelineDB实例引用，用于ID计数器的持久化存储
//     如果为nil，计数器功能将不可用，但其他功能正常
//
// 返回值：
// - *GroupManager: 完全初始化的组管理器实例
//
// 初始化内容：
// 1. 创建空的暂停组映射
// 2. 创建空的OnComplete执行状态映射
// 3. 创建空的组计数器映射
// 4. 创建空的组统计缓存映射
// 5. 设置数据库引用
//
// 使用示例：
//
//	gm := NewGroupManager(pipelineDB)
//	gm.PauseGroup("batch1")     // 暂停组
//	id := gm.GetNextID("user")  // 获取ID
func NewGroupManager(pdb *PipelineDB) *GroupManager {
	return &GroupManager{
		pausedGroups:        make(map[string]bool),             // 初始化暂停组映射
		onCompleteExecuting: make(map[string]bool),             // 初始化OnComplete执行状态映射
		groupCounters:       make(map[string]*int64),           // 初始化组计数器映射
		groupStats:          make(map[string]*GroupStatsCache), // 初始化组统计缓存映射
		pdb:                 pdb,                               // 设置数据库引用
		// mu 会自动初始化为零值（未锁定状态）
	}
}

// PauseGroup 暂停指定组的数据接收功能
//
// 核心功能：将组标记为暂停状态，阻止新数据的接收
// 时间复杂度：O(1) - HashMap插入操作
// 空间复杂度：O(1) - 只添加一个映射条目
//
// 参数说明：
// - group: 要暂停的组名
//
// 执行流程：
// 1. 获取写锁（独占访问）
// 2. 将组添加到暂停组映射中
// 3. 输出暂停确认信息
//
// 使用场景：
// - 维护期间暂停特定组的数据处理
// - 数据质量问题时临时停止数据接收
// - 系统负载过高时选择性暂停部分组
// - 业务流程控制需要
//
// 注意事项：
// - 暂停后该组无法接收新数据，但已有数据继续处理
// - 可以通过ResumeGroup恢复
// - 暂停状态不会持久化，重启后会丢失
//
// 并发安全：使用写锁保护，确保状态修改的原子性
func (gm *GroupManager) PauseGroup(group string) {
	gm.mu.Lock()         // 获取写锁，独占访问
	defer gm.mu.Unlock() // 确保函数退出时释放锁

	// 将组标记为暂停状态
	gm.pausedGroups[group] = true

	// 输出暂停确认信息（便于监控和调试）
	gm.pdb.logger.Info("⏸️ 组已暂停接收数据", "group", group)
}

// ResumeGroup 恢复指定组的数据接收功能
//
// 核心功能：将组从暂停状态中移除，恢复正常的数据接收
// 时间复杂度：O(1) - HashMap删除操作
// 空间复杂度：O(1) - 释放一个映射条目的空间
//
// 参数说明：
// - group: 要恢复的组名
//
// 执行流程：
// 1. 获取写锁（独占访问）
// 2. 从暂停组映射中删除该组
// 3. 输出恢复确认信息
//
// 使用场景：
// - 维护完成后恢复组的正常运行
// - 数据质量问题解决后重新启用组
// - 系统负载降低后恢复暂停的组
// - 业务流程需要重新激活组
//
// 注意事项：
// - 恢复后该组立即可以接收新数据
// - 如果组本来就没有暂停，操作是幂等的（无副作用）
// - 恢复状态不会持久化，重启后默认为活跃状态
//
// 并发安全：使用写锁保护，确保状态修改的原子性
func (gm *GroupManager) ResumeGroup(group string) {
	gm.mu.Lock()         // 获取写锁，独占访问
	defer gm.mu.Unlock() // 确保函数退出时释放锁

	// 从暂停组映射中删除该组（恢复活跃状态）
	delete(gm.pausedGroups, group)

	// 输出恢复确认信息（便于监控和调试）
	gm.pdb.logger.Info("▶️ 组已恢复接收数据", "group", group)
}

// IsPaused 检查指定组是否处于暂停状态
//
// 核心功能：查询组的暂停状态，用于数据接收前的状态检查
// 时间复杂度：O(1) - HashMap查找操作
// 空间复杂度：O(1) - 不创建额外数据
//
// 参数说明：
// - group: 要检查的组名
//
// 返回值：
// - bool: true表示组已暂停，false表示组处于活跃状态
//
// 执行流程：
// 1. 获取读锁（允许并发读取）
// 2. 在暂停组映射中查找该组
// 3. 返回查找结果
//
// 使用场景：
// - 数据接收前的状态验证
// - 业务逻辑中的条件判断
// - 监控和状态报告
// - 自动化运维脚本
//
// 设计考虑：
// - 使用读锁支持高并发查询
// - HashMap查找提供O(1)性能
// - 不存在的组默认返回false（活跃状态）
//
// 并发安全：使用读锁保护，支持多线程并发调用
func (gm *GroupManager) IsPaused(group string) bool {
	gm.mu.RLock()         // 获取读锁，允许并发读取
	defer gm.mu.RUnlock() // 确保函数退出时释放锁

	// 在暂停组映射中查找该组
	// 如果组不存在，返回false（默认为活跃状态）
	return gm.pausedGroups[group]
}

// GetPausedGroups 获取所有暂停的组
func (gm *GroupManager) GetPausedGroups() []string {
	gm.mu.RLock()
	defer gm.mu.RUnlock()

	var groups []string
	for group := range gm.pausedGroups {
		groups = append(groups, group)
	}
	return groups
}

// GetGroupStatus 获取组状态
func (gm *GroupManager) GetGroupStatus(group string) string {
	if gm.IsPaused(group) {
		return "paused"
	}
	return "active"
}

// IsOnCompleteExecuting 检查组是否正在执行 OnComplete
func (gm *GroupManager) IsOnCompleteExecuting(group string) bool {
	gm.mu.RLock()
	defer gm.mu.RUnlock()
	return gm.onCompleteExecuting[group]
}

// SetOnCompleteExecuting 设置组的 OnComplete 执行状态
func (gm *GroupManager) SetOnCompleteExecuting(group string, executing bool) {
	gm.mu.Lock()
	defer gm.mu.Unlock()
	if executing {
		gm.onCompleteExecuting[group] = true
		gm.pdb.logger.Info("🔒 组 OnComplete 开始执行", "group", group)
	} else {
		delete(gm.onCompleteExecuting, group)
		gm.pdb.logger.Info("🔓 组 OnComplete 执行完成", "group", group)
	}
}

// GetExecutingGroups 获取所有正在执行 OnComplete 的组
func (gm *GroupManager) GetExecutingGroups() []string {
	gm.mu.RLock()
	defer gm.mu.RUnlock()

	var groups []string
	for group := range gm.onCompleteExecuting {
		groups = append(groups, group)
	}
	return groups
}

// GetNextID 为指定组生成下一个唯一的自增ID
//
// 核心功能：为每个组维护独立的ID序列，确保ID的唯一性和连续性
// 时间复杂度：O(1) - 内存计数器自增 + 持久化写入
// 空间复杂度：O(1) - 只维护单个计数器
//
// 参数说明：
// - group: 组名，每个组有独立的ID序列
//
// 返回值：
// - int64: 新生成的唯一ID（从1开始递增）
//
// 执行流程：
//  1. 获取写锁（保证ID生成的原子性）
//  2. 检查组计数器是否已加载：
//     a. 如果未加载：从持久化存储中加载计数器
//     b. 如果已加载：直接使用内存中的计数器
//  3. 计数器自增1
//  4. 将新的计数器值持久化到磁盘
//  5. 返回新生成的ID
//
// 持久化机制：
// - 计数器存储在专用的计数器页面中
// - 每次ID生成都会立即持久化，确保重启后的连续性
// - 页面格式：[组名长度][组名][计数器值]...
//
// 设计考虑：
// - 每个组独立的ID序列，避免冲突
// - 立即持久化确保数据安全性
// - 内存缓存提高性能
// - 写锁保证并发安全
//
// 使用场景：
// - 数据记录的主键生成
// - 业务对象的唯一标识
// - 序列号生成
//
// 并发安全：使用写锁保护，确保ID生成的唯一性
func (gm *GroupManager) GetNextID(group string) int64 {
	gm.mu.Lock()         // 获取写锁，保证ID生成的原子性
	defer gm.mu.Unlock() // 确保函数退出时释放锁

	// 检查组的计数器是否已经加载到内存中
	if gm.groupCounters[group] == nil {
		// 计数器未加载，从持久化存储中加载
		counter := gm.loadCounterFromPage(group)
		gm.groupCounters[group] = &counter
	}

	// 计数器自增，生成新的ID
	*gm.groupCounters[group]++
	newID := *gm.groupCounters[group]

	// 立即将新的计数器值持久化到磁盘
	// 这确保了即使系统崩溃，ID序列也不会重复
	gm.saveCounterToPage(group, newID)

	return newID // 返回新生成的唯一ID
}

// GetGroupCounter 获取组的当前计数器值
func (gm *GroupManager) GetGroupCounter(group string) int64 {
	gm.mu.RLock()
	defer gm.mu.RUnlock()

	if gm.groupCounters[group] == nil {
		return 0
	}
	return *gm.groupCounters[group]
}

// loadCounterFromPage 从页面加载组计数器
func (gm *GroupManager) loadCounterFromPage(group string) int64 {
	if gm.pdb == nil || gm.pdb.header.CounterPage == 0 {
		return 0 // 如果没有计数器页面，返回0
	}

	// 读取计数器页面
	page, err := gm.pdb.readPage(gm.pdb.header.CounterPage)
	if err != nil {
		gm.pdb.logger.Warn("⚠️ 读取计数器页面失败", "error", err)
		return 0
	}

	// 在页面中查找组的计数器
	// 页面格式：[组名长度(2字节)][组名][计数器(8字节)]...
	offset := uint16(0)
	for offset < uint16(len(page)-10) { // 至少需要2+1+8=11字节
		// 读取组名长度
		nameLen := binary.LittleEndian.Uint16(page[offset : offset+2])
		if nameLen == 0 || offset+2+nameLen+8 > uint16(len(page)) {
			break
		}

		// 读取组名
		name := string(page[offset+2 : offset+2+nameLen])
		if name == group {
			// 找到了，读取计数器
			counter := binary.LittleEndian.Uint64(page[offset+2+nameLen : offset+2+nameLen+8])
			return int64(counter)
		}

		// 移动到下一个条目
		offset += 2 + nameLen + 8
	}

	return 0 // 没找到，返回0
}

// saveCounterToPage 保存组计数器到页面
func (gm *GroupManager) saveCounterToPage(group string, counter int64) {
	if gm.pdb == nil {
		return
	}

	// 如果没有计数器页面，创建一个
	if gm.pdb.header.CounterPage == 0 {
		gm.initCounterPage()
	}

	// 读取计数器页面
	page, err := gm.pdb.readPage(gm.pdb.header.CounterPage)
	if err != nil {
		gm.pdb.logger.Warn("⚠️ 读取计数器页面失败", "error", err)
		return
	}

	// 查找并更新组的计数器
	offset := uint16(0)
	found := false

	for offset < uint16(len(page)-10) {
		// 读取组名长度
		nameLen := binary.LittleEndian.Uint16(page[offset : offset+2])
		if nameLen == 0 {
			break
		}

		if offset+2+nameLen+8 > uint16(len(page)) {
			break
		}

		// 读取组名
		name := string(page[offset+2 : offset+2+nameLen])
		if name == group {
			// 找到了，更新计数器
			binary.LittleEndian.PutUint64(page[offset+2+nameLen:offset+2+nameLen+8], uint64(counter))
			found = true
			break
		}

		// 移动到下一个条目
		offset += 2 + nameLen + 8
	}

	// 如果没找到，添加新条目
	if !found {
		nameLen := uint16(len(group))
		if offset+2+nameLen+8 <= uint16(len(page)) {
			// 写入组名长度
			binary.LittleEndian.PutUint16(page[offset:offset+2], nameLen)
			// 写入组名
			copy(page[offset+2:offset+2+nameLen], []byte(group))
			// 写入计数器
			binary.LittleEndian.PutUint64(page[offset+2+nameLen:offset+2+nameLen+8], uint64(counter))
		}
	}

	// 写回页面
	gm.pdb.cache.Put(gm.pdb.header.CounterPage, page, true)
}

// initCounterPage 初始化计数器页面
func (gm *GroupManager) initCounterPage() {
	if gm.pdb == nil {
		return
	}

	// 分配新页面
	pageNo := gm.pdb.header.TotalPages
	gm.pdb.header.TotalPages++
	gm.pdb.header.CounterPage = pageNo

	// 创建空页面
	page := make([]byte, PageSize)

	// 写入页面
	gm.pdb.cache.Put(pageNo, page, true)

	// 保存头部
	gm.pdb.saveHeader()

	gm.pdb.logger.Info("📄 初始化计数器页面", "page", pageNo)
}

// IncrementStats 增加组的统计计数（新增记录时调用）
//
// 核心功能：当新记录添加到组时，更新该组的统计信息
// 时间复杂度：O(1) - 直接的计数器操作
// 空间复杂度：O(1) - 可能创建一个新的统计缓存结构
//
// 参数说明：
// - group: 组名
// - status: 新记录的数据状态（Hot/Warm/Cold）
//
// 执行流程：
// 1. 获取写锁（保证统计更新的原子性）
// 2. 确保组的统计缓存已初始化
// 3. 增加总记录数计数
// 4. 根据状态增加对应的状态计数
//
// 统计维护：
// - TotalRecords: 总记录数自增1
// - 对应状态计数: Hot/Warm/Cold计数自增1
// - 统计信息实时更新，无延迟
//
// 使用场景：
// - 新数据记录插入时
// - 数据导入过程中
// - 批量数据处理时
//
// 并发安全：使用写锁保护，确保统计数据的一致性
func (gm *GroupManager) IncrementStats(group string, status DataStatus) {
	gm.mu.Lock()         // 获取写锁，保证统计更新的原子性
	defer gm.mu.Unlock() // 确保函数退出时释放锁

	// 确保组的统计缓存已经初始化
	if gm.groupStats[group] == nil {
		gm.groupStats[group] = &GroupStatsCache{}
	}

	// 增加总记录数
	gm.groupStats[group].TotalRecords++

	// 根据数据状态增加对应的计数
	switch status {
	case StatusHot:
		gm.groupStats[group].HotRecords++ // 增加热数据计数
	case StatusWarm:
		gm.groupStats[group].WarmRecords++ // 增加温数据计数
	case StatusCold:
		gm.groupStats[group].ColdRecords++ // 增加冷数据计数
	}
}

// UpdateStats 更新组统计信息（数据状态转换时调用）
//
// 核心功能：当记录的状态发生转换时，更新统计信息以保持准确性
// 时间复杂度：O(1) - 直接的计数器操作
// 空间复杂度：O(1) - 可能创建一个新的统计缓存结构
//
// 参数说明：
// - group: 组名
// - oldStatus: 记录的原始状态
// - newStatus: 记录的新状态
//
// 执行流程：
// 1. 获取写锁（保证统计更新的原子性）
// 2. 确保组的统计缓存已初始化
// 3. 减少原状态的计数
// 4. 增加新状态的计数
// 5. 总记录数保持不变
//
// 状态转换处理：
// - Hot → Warm: HotRecords-1, WarmRecords+1
// - Warm → Cold: WarmRecords-1, ColdRecords+1
// - 支持任意状态间的转换
// - TotalRecords保持不变（只是状态分布改变）
//
// 使用场景：
// - 数据生命周期管理（Hot→Warm→Cold）
// - 缓存策略调整
// - 数据归档过程
//
// 并发安全：使用写锁保护，确保统计数据的一致性
func (gm *GroupManager) UpdateStats(group string, oldStatus, newStatus DataStatus) {
	gm.mu.Lock()         // 获取写锁，保证统计更新的原子性
	defer gm.mu.Unlock() // 确保函数退出时释放锁

	// 确保组的统计缓存已经初始化
	if gm.groupStats[group] == nil {
		gm.groupStats[group] = &GroupStatsCache{}
	}

	// 减少原状态的计数
	switch oldStatus {
	case StatusHot:
		gm.groupStats[group].HotRecords-- // 减少热数据计数
	case StatusWarm:
		gm.groupStats[group].WarmRecords-- // 减少温数据计数
	case StatusCold:
		gm.groupStats[group].ColdRecords-- // 减少冷数据计数
	}

	// 增加新状态的计数
	switch newStatus {
	case StatusHot:
		gm.groupStats[group].HotRecords++ // 增加热数据计数
	case StatusWarm:
		gm.groupStats[group].WarmRecords++ // 增加温数据计数
	case StatusCold:
		gm.groupStats[group].ColdRecords++ // 增加冷数据计数
	}

	// 注意：TotalRecords保持不变，因为只是状态转换，记录总数没有变化
}

// GetFastStats 快速获取所有组的统计信息（O(1)复杂度）
//
// 核心功能：从内存缓存中快速获取所有组的统计信息
// 时间复杂度：O(n) - n为组的数量（需要复制所有统计数据）
// 空间复杂度：O(n) - 创建所有统计数据的副本
//
// 返回值：
//   - map[string]*GroupStatsCache: 所有组的统计信息副本
//     key为组名，value为该组的统计信息
//
// 执行流程：
// 1. 获取读锁（允许并发读取）
// 2. 遍历所有组的统计缓存
// 3. 为每个组创建统计信息的副本
// 4. 返回完整的统计信息映射
//
// 设计考虑：
// - 返回副本而不是原始数据，防止外部修改
// - 使用读锁支持高并发查询
// - 内存缓存提供极快的查询性能
// - 避免磁盘I/O，适合频繁查询
//
// 性能优势：
// - 相比传统的磁盘扫描统计，性能提升数百倍
// - 支持高频率的监控查询
// - 实时性强，统计数据始终是最新的
//
// 使用场景：
// - 系统监控和仪表板
// - 性能分析和报告
// - 容量规划和预警
// - API接口的统计查询
//
// 并发安全：使用读锁保护，支持多线程并发调用
func (gm *GroupManager) GetFastStats() map[string]*GroupStatsCache {
	gm.mu.RLock()         // 获取读锁，允许并发读取
	defer gm.mu.RUnlock() // 确保函数退出时释放锁

	// 创建统计数据的完整副本
	result := make(map[string]*GroupStatsCache)

	// 遍历所有组的统计缓存，创建副本
	for group, stats := range gm.groupStats {
		result[group] = &GroupStatsCache{
			TotalRecords: stats.TotalRecords, // 复制总记录数
			HotRecords:   stats.HotRecords,   // 复制热数据记录数
			WarmRecords:  stats.WarmRecords,  // 复制温数据记录数
			ColdRecords:  stats.ColdRecords,  // 复制冷数据记录数
		}
	}

	return result // 返回统计信息副本
}