package windowsv3

import (
	"context"
	"database/sql"
	"encoding/hex"
	"fmt"
	"regexp"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/fsnotify/fsnotify"
	_ "github.com/mattn/go-sqlite3"
	"github.com/rs/zerolog/log"

	"github.com/sjzar/chatlog/internal/errors"
	"github.com/sjzar/chatlog/internal/model"
	"github.com/sjzar/chatlog/internal/wechatdb/datasource/dbm"
	"github.com/sjzar/chatlog/pkg/util"
)

const (
	Message = "message"
	Contact = "contact"
	Image   = "image"
	Video   = "video"
	File    = "file"
	Voice   = "voice"

	talkerCacheTTL = 30 * time.Second
)

var Groups = []*dbm.Group{
	{
		Name:      Message,
		Pattern:   `^MSG([0-9]?[0-9])?\.db$`,
		BlackList: []string{},
	},
	{
		Name:      Contact,
		Pattern:   `^MicroMsg\.db$`,
		BlackList: []string{},
	},
	{
		Name:      Image,
		Pattern:   `^HardLinkImage\.db$`,
		BlackList: []string{},
	},
	{
		Name:      Video,
		Pattern:   `^HardLinkVideo\.db$`,
		BlackList: []string{},
	},
	{
		Name:      File,
		Pattern:   `^HardLinkFile\.db$`,
		BlackList: []string{},
	},
	{
		Name:      Voice,
		Pattern:   `^MediaMSG([0-9]?[0-9])?\.db$`,
		BlackList: []string{},
	},
}

// MessageDBInfo 保存消息数据库的信息
type MessageDBInfo struct {
	FilePath  string
	StartTime time.Time
	EndTime   time.Time
	TalkerMap map[string]int
}

// DataSource 实现了 DataSource 接口
type DataSource struct {
	path string
	dbm  *dbm.DBManager

	// 消息数据库信息
	messageInfos []MessageDBInfo

	talkerCacheMu     sync.RWMutex
	talkerCache       []string
	talkerCacheExpiry time.Time
}

// New 创建一个新的 WindowsV3DataSource
func New(path string) (*DataSource, error) {
	ds := &DataSource{
		path:         path,
		dbm:          dbm.NewDBManager(path),
		messageInfos: make([]MessageDBInfo, 0),
	}

	for _, g := range Groups {
		ds.dbm.AddGroup(g)
	}

	if err := ds.dbm.Start(); err != nil {
		return nil, err
	}

	if err := ds.initMessageDbs(); err != nil {
		return nil, errors.DBInitFailed(err)
	}

	ds.dbm.AddCallback(Message, func(event fsnotify.Event) error {
		if !(event.Op.Has(fsnotify.Create) || event.Op.Has(fsnotify.Write) || event.Op.Has(fsnotify.Rename)) {
			return nil
		}
		if err := ds.initMessageDbs(); err != nil {
			log.Err(err).Msgf("Failed to reinitialize message DBs: %s", event.Name)
		}
		ds.invalidateTalkerCache()
		return nil
	})

	ds.dbm.AddCallback(Contact, func(event fsnotify.Event) error {
		if !(event.Op.Has(fsnotify.Create) || event.Op.Has(fsnotify.Write) || event.Op.Has(fsnotify.Rename) || event.Op.Has(fsnotify.Remove)) {
			return nil
		}
		ds.invalidateTalkerCache()
		return nil
	})

	return ds, nil
}

func (ds *DataSource) SetCallback(group string, callback func(event fsnotify.Event) error) error {
	if group == "chatroom" {
		group = Contact
	}
	return ds.dbm.AddCallback(group, callback)
}

// initMessageDbs 初始化消息数据库
func (ds *DataSource) initMessageDbs() error {
	// 获取所有消息数据库文件路径
	dbPaths, err := ds.dbm.GetDBPath(Message)
	if err != nil {
		if strings.Contains(err.Error(), "db file not found") {
			ds.messageInfos = make([]MessageDBInfo, 0)
			return nil
		}
		return err
	}

	// 处理每个数据库文件
	infos := make([]MessageDBInfo, 0)
	for _, filePath := range dbPaths {
		db, err := ds.dbm.OpenDB(filePath)
		if err != nil {
			log.Err(err).Msgf("获取数据库 %s 失败", filePath)
			continue
		}

		// 获取 DBInfo 表中的开始时间
		var startTime time.Time

		rows, err := db.Query("SELECT tableIndex, tableVersion, tableDesc FROM DBInfo")
		if err != nil {
			log.Err(err).Msgf("查询数据库 %s 的 DBInfo 表失败", filePath)
			continue
		}

		for rows.Next() {
			var tableIndex int
			var tableVersion int64
			var tableDesc string

			if err := rows.Scan(&tableIndex, &tableVersion, &tableDesc); err != nil {
				log.Err(err).Msg("扫描 DBInfo 行失败")
				continue
			}

			// 查找描述为 "Start Time" 的记录
			if strings.Contains(tableDesc, "Start Time") {
				startTime = time.Unix(tableVersion/1000, (tableVersion%1000)*1000000)
				break
			}
		}
		rows.Close()

		// 组织 TalkerMap
		talkerMap := make(map[string]int)
		rows, err = db.Query("SELECT UsrName FROM Name2ID")
		if err != nil {
			log.Err(err).Msgf("查询数据库 %s 的 Name2ID 表失败", filePath)
			continue
		}

		i := 1
		for rows.Next() {
			var userName string
			if err := rows.Scan(&userName); err != nil {
				log.Err(err).Msg("扫描 Name2ID 行失败")
				continue
			}
			talkerMap[userName] = i
			i++
		}
		rows.Close()

		// 保存数据库信息
		infos = append(infos, MessageDBInfo{
			FilePath:  filePath,
			StartTime: startTime,
			TalkerMap: talkerMap,
		})
	}

	// 按照 StartTime 排序数据库文件
	sort.Slice(infos, func(i, j int) bool {
		return infos[i].StartTime.Before(infos[j].StartTime)
	})

	// 设置结束时间
	for i := range infos {
		if i == len(infos)-1 {
			infos[i].EndTime = time.Now()
		} else {
			infos[i].EndTime = infos[i+1].StartTime
		}
	}
	if len(ds.messageInfos) > 0 && len(infos) < len(ds.messageInfos) {
		log.Warn().Msgf("message db count decreased from %d to %d, skip init", len(ds.messageInfos), len(infos))
		return nil
	}
	ds.messageInfos = infos
	ds.invalidateTalkerCache()
	return nil
}

// getDBInfosForTimeRange 获取时间范围内的数据库信息
func (ds *DataSource) getDBInfosForTimeRange(startTime, endTime time.Time) []MessageDBInfo {
	var dbs []MessageDBInfo
	for _, info := range ds.messageInfos {
		if info.StartTime.Before(endTime) && info.EndTime.After(startTime) {
			dbs = append(dbs, info)
		}
	}
	return dbs
}

func (ds *DataSource) GetMessages(ctx context.Context, startTime, endTime time.Time, talker string, sender string, keyword string, limit, offset int) ([]*model.Message, error) {
	if talker == "" {
		return nil, errors.ErrTalkerEmpty
	}

	// 解析talker参数，支持多个talker（以英文逗号分隔）
	talkers := util.Str2List(talker, ",")
	if len(talkers) == 0 {
		return nil, errors.ErrTalkerEmpty
	}

	// 找到时间范围内的数据库文件
	dbInfos := ds.getDBInfosForTimeRange(startTime, endTime)
	if len(dbInfos) == 0 {
		return nil, errors.TimeRangeNotFound(startTime, endTime)
	}

	// 解析sender参数，支持多个发送者（以英文逗号分隔）
	senders := util.Str2List(sender, ",")

	// 预编译正则表达式（如果有keyword）
	var regex *regexp.Regexp
	if keyword != "" {
		var err error
		regex, err = regexp.Compile(keyword)
		if err != nil {
			return nil, errors.QueryFailed("invalid regex pattern", err)
		}
	}

	// 从每个相关数据库中查询消息
	filteredMessages := []*model.Message{}

	for _, dbInfo := range dbInfos {
		// 检查上下文是否已取消
		if err := ctx.Err(); err != nil {
			return nil, err
		}

		db, err := ds.dbm.OpenDB(dbInfo.FilePath)
		if err != nil {
			log.Error().Msgf("数据库 %s 未打开", dbInfo.FilePath)
			continue
		}

		// 对每个talker进行查询
		for _, talkerItem := range talkers {
			// 构建查询条件
			conditions := []string{"Sequence >= ? AND Sequence <= ?"}
			args := []interface{}{startTime.Unix() * 1000, endTime.Unix() * 1000}

			// 添加talker条件
			talkerID, ok := dbInfo.TalkerMap[talkerItem]
			if ok {
				conditions = append(conditions, "TalkerId = ?")
				args = append(args, talkerID)
			} else {
				conditions = append(conditions, "StrTalker = ?")
				args = append(args, talkerItem)
			}

			query := fmt.Sprintf(`
				SELECT MsgSvrID, Sequence, CreateTime, StrTalker, IsSender, 
					Type, SubType, StrContent, CompressContent, BytesExtra
				FROM MSG 
				WHERE %s 
				ORDER BY Sequence ASC
			`, strings.Join(conditions, " AND "))

			// 执行查询
			rows, err := db.QueryContext(ctx, query, args...)
			if err != nil {
				// 如果表不存在，跳过此talker
				if strings.Contains(err.Error(), "no such table") {
					continue
				}
				log.Err(err).Msgf("从数据库 %s 查询消息失败", dbInfo.FilePath)
				continue
			}

			// 处理查询结果，在读取时进行过滤
			for rows.Next() {
				var msg model.MessageV3
				var compressContent []byte
				var bytesExtra []byte

				err := rows.Scan(
					&msg.MsgSvrID,
					&msg.Sequence,
					&msg.CreateTime,
					&msg.StrTalker,
					&msg.IsSender,
					&msg.Type,
					&msg.SubType,
					&msg.StrContent,
					&compressContent,
					&bytesExtra,
				)
				if err != nil {
					rows.Close()
					return nil, errors.ScanRowFailed(err)
				}
				msg.CompressContent = compressContent
				msg.BytesExtra = bytesExtra

				// 将消息转换为标准格式
				message := msg.Wrap()

				// 应用sender过滤
				if len(senders) > 0 {
					senderMatch := false
					for _, s := range senders {
						if message.Sender == s {
							senderMatch = true
							break
						}
					}
					if !senderMatch {
						continue // 不匹配sender，跳过此消息
					}
				}

				// 应用keyword过滤
				if regex != nil {
					plainText := message.PlainTextContent()
					if !regex.MatchString(plainText) {
						continue // 不匹配keyword，跳过此消息
					}
				}

				// 通过所有过滤条件，保留此消息
				filteredMessages = append(filteredMessages, message)

				// 检查是否已经满足分页处理数量
				if limit > 0 && len(filteredMessages) >= offset+limit {
					// 已经获取了足够的消息，可以提前返回
					rows.Close()

					// 对所有消息按时间排序
					sort.Slice(filteredMessages, func(i, j int) bool {
						return filteredMessages[i].Seq < filteredMessages[j].Seq
					})

					// 处理分页
					if offset >= len(filteredMessages) {
						return []*model.Message{}, nil
					}
					end := offset + limit
					if end > len(filteredMessages) {
						end = len(filteredMessages)
					}
					return filteredMessages[offset:end], nil
				}
			}
			rows.Close()
		}
	}

	// 对所有消息按时间排序
	sort.Slice(filteredMessages, func(i, j int) bool {
		return filteredMessages[i].Seq < filteredMessages[j].Seq
	})

	// 处理分页
	if limit > 0 {
		if offset >= len(filteredMessages) {
			return []*model.Message{}, nil
		}
		end := offset + limit
		if end > len(filteredMessages) {
			end = len(filteredMessages)
		}
		return filteredMessages[offset:end], nil
	}

	return filteredMessages, nil
}

func (ds *DataSource) GetDatasetFingerprint(context.Context) (string, error) {
	return ds.dbm.FingerprintForGroups(Message)
}

func (ds *DataSource) getCachedTalkers() []string {
	ds.talkerCacheMu.RLock()
	if ds.talkerCacheExpiry.IsZero() || time.Now().After(ds.talkerCacheExpiry) {
		ds.talkerCacheMu.RUnlock()
		return nil
	}
	cached := append([]string(nil), ds.talkerCache...)
	ds.talkerCacheMu.RUnlock()
	return cached
}

func (ds *DataSource) cacheTalkers(talkers []string) {
	copySlice := append([]string(nil), talkers...)
	ds.talkerCacheMu.Lock()
	ds.talkerCache = copySlice
	ds.talkerCacheExpiry = time.Now().Add(talkerCacheTTL)
	ds.talkerCacheMu.Unlock()
}

func (ds *DataSource) invalidateTalkerCache() {
	ds.talkerCacheMu.Lock()
	ds.talkerCache = nil
	ds.talkerCacheExpiry = time.Time{}
	ds.talkerCacheMu.Unlock()
}

func (ds *DataSource) collectAllTalkers(ctx context.Context) ([]string, error) {
	if cached := ds.getCachedTalkers(); cached != nil {
		return cached, nil
	}

	talkerSet := make(map[string]struct{})
	talkers := make([]string, 0)

	add := func(name string) {
		name = strings.TrimSpace(name)
		if name == "" {
			return
		}
		if _, exists := talkerSet[name]; exists {
			return
		}
		talkerSet[name] = struct{}{}
		talkers = append(talkers, name)
	}

	if db, err := ds.dbm.GetDB(Contact); err == nil && db != nil {
		rows, err := db.QueryContext(ctx, `SELECT strUsrName FROM Session WHERE IFNULL(strUsrName,'') <> ''`)
		if err == nil {
			for rows.Next() {
				if err := ctx.Err(); err != nil {
					rows.Close()
					return nil, err
				}
				var username string
				if err := rows.Scan(&username); err != nil {
					log.Debug().Err(err).Msg("scan session talker failed")
					continue
				}
				add(username)
			}
			if err := rows.Err(); err != nil {
				log.Debug().Err(err).Msg("iterate session talkers failed")
			}
			rows.Close()
		} else {
			log.Debug().Err(err).Msg("query session talkers failed")
		}

		if len(talkers) == 0 {
			rows, err = db.QueryContext(ctx, `SELECT UserName FROM Contact WHERE IFNULL(UserName,'') <> ''`)
			if err == nil {
				for rows.Next() {
					if err := ctx.Err(); err != nil {
						rows.Close()
						return nil, err
					}
					var username string
					if err := rows.Scan(&username); err != nil {
						log.Debug().Err(err).Msg("scan contact talker failed")
						continue
					}
					add(username)
				}
				if err := rows.Err(); err != nil {
					log.Debug().Err(err).Msg("iterate contact talkers failed")
				}
				rows.Close()
			} else {
				log.Debug().Err(err).Msg("query contact talkers failed")
			}
		}
	}

	if len(talkers) == 0 {
		for _, info := range ds.messageInfos {
			for username := range info.TalkerMap {
				add(username)
			}
		}
	}

	sort.Strings(talkers)
	ds.cacheTalkers(talkers)
	return talkers, nil
}

// ListTalkers 返回所有对话用户名（联系人 + 群聊），供索引使用
func (ds *DataSource) ListTalkers(ctx context.Context) ([]string, error) {
	return ds.collectAllTalkers(ctx)
}

// IterateMessages 按 talker 枚举消息并交给处理函数，供 Bleve 索引使用
func (ds *DataSource) IterateMessages(ctx context.Context, talkers []string, handler func(*model.Message) error) error {
	if handler == nil {
		return errors.InvalidArg("handler")
	}

	if len(talkers) == 0 {
		var err error
		talkers, err = ds.collectAllTalkers(ctx)
		if err != nil {
			return err
		}
	}
	if len(talkers) == 0 {
		return nil
	}

	uniqueTalkers := make([]string, 0, len(talkers))
	seen := make(map[string]struct{}, len(talkers))
	for _, t := range talkers {
		t = strings.TrimSpace(t)
		if t == "" {
			continue
		}
		if _, ok := seen[t]; ok {
			continue
		}
		seen[t] = struct{}{}
		uniqueTalkers = append(uniqueTalkers, t)
	}
	if len(uniqueTalkers) == 0 {
		return nil
	}

	for _, info := range ds.messageInfos {
		if err := ctx.Err(); err != nil {
			return err
		}

		db, err := ds.dbm.OpenDB(info.FilePath)
		if err != nil {
			log.Debug().Err(err).Msgf("open message db failed: %s", info.FilePath)
			continue
		}

		for _, talker := range uniqueTalkers {
			if err := ctx.Err(); err != nil {
				return err
			}

			conditions := []string{"StrContent IS NOT NULL"}
			args := make([]interface{}, 0, 1)
			if talkerID, ok := info.TalkerMap[talker]; ok {
				conditions = append(conditions, "TalkerId = ?")
				args = append(args, talkerID)
			} else {
				conditions = append(conditions, "StrTalker = ?")
				args = append(args, talker)
			}

			query := fmt.Sprintf(`
				SELECT MsgSvrID, Sequence, CreateTime, StrTalker, IsSender,
				       Type, SubType, StrContent, CompressContent, BytesExtra
				FROM MSG
				WHERE %s
				ORDER BY Sequence ASC
			`, strings.Join(conditions, " AND "))

			rows, err := db.QueryContext(ctx, query, args...)
			if err != nil {
				if strings.Contains(err.Error(), "no such table") {
					continue
				}
				return errors.QueryFailed("iterate messages", err)
			}

			for rows.Next() {
				if err := ctx.Err(); err != nil {
					rows.Close()
					return err
				}
				var msg model.MessageV3
				var compressContent []byte
				var bytesExtra []byte
				if scanErr := rows.Scan(
					&msg.MsgSvrID,
					&msg.Sequence,
					&msg.CreateTime,
					&msg.StrTalker,
					&msg.IsSender,
					&msg.Type,
					&msg.SubType,
					&msg.StrContent,
					&compressContent,
					&bytesExtra,
				); scanErr != nil {
					rows.Close()
					return errors.ScanRowFailed(scanErr)
				}
				msg.CompressContent = compressContent
				msg.BytesExtra = bytesExtra

				wrapped := msg.Wrap()
				if err := handler(wrapped); err != nil {
					rows.Close()
					return err
				}
			}
			if err := rows.Err(); err != nil {
				rows.Close()
				return errors.QueryFailed("iterate message rows", err)
			}
			rows.Close()
		}
	}

	return nil
}

// GetContacts 实现获取联系人信息的方法
func (ds *DataSource) GetContacts(ctx context.Context, key string, limit, offset int) ([]*model.Contact, error) {
	var query string
	var args []interface{}

	if key != "" {
		// 按照关键字查询
		query = `SELECT UserName, Alias, Remark, NickName, Reserved1 FROM Contact 
                WHERE UserName = ? OR Alias = ? OR Remark = ? OR NickName = ?`
		args = []interface{}{key, key, key, key}
	} else {
		// 查询所有联系人
		query = `SELECT UserName, Alias, Remark, NickName, Reserved1 FROM Contact`
	}

	// 添加排序、分页
	query += ` ORDER BY UserName`
	if limit > 0 {
		query += fmt.Sprintf(" LIMIT %d", limit)
		if offset > 0 {
			query += fmt.Sprintf(" OFFSET %d", offset)
		}
	}

	// 执行查询
	db, err := ds.dbm.GetDB(Contact)
	if err != nil {
		return nil, err
	}
	rows, err := db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, errors.QueryFailed(query, err)
	}
	defer rows.Close()

	contacts := []*model.Contact{}
	for rows.Next() {
		var contactV3 model.ContactV3
		err := rows.Scan(
			&contactV3.UserName,
			&contactV3.Alias,
			&contactV3.Remark,
			&contactV3.NickName,
			&contactV3.Reserved1,
		)

		if err != nil {
			return nil, errors.ScanRowFailed(err)
		}

		contacts = append(contacts, contactV3.Wrap())
	}

	return contacts, nil
}

// GetChatRooms 实现获取群聊信息的方法
func (ds *DataSource) GetChatRooms(ctx context.Context, key string, limit, offset int) ([]*model.ChatRoom, error) {
	var query string
	var args []interface{}

	if key != "" {
		// 按照关键字查询
		query = `SELECT ChatRoomName, Reserved2, RoomData FROM ChatRoom WHERE ChatRoomName = ?`
		args = []interface{}{key}

		// 执行查询
		db, err := ds.dbm.GetDB(Contact)
		if err != nil {
			return nil, err
		}
		rows, err := db.QueryContext(ctx, query, args...)
		if err != nil {
			return nil, errors.QueryFailed(query, err)
		}
		defer rows.Close()

		chatRooms := []*model.ChatRoom{}
		for rows.Next() {
			var chatRoomV3 model.ChatRoomV3
			err := rows.Scan(
				&chatRoomV3.ChatRoomName,
				&chatRoomV3.Reserved2,
				&chatRoomV3.RoomData,
			)

			if err != nil {
				return nil, errors.ScanRowFailed(err)
			}

			chatRooms = append(chatRooms, chatRoomV3.Wrap())
		}

		// 如果没有找到群聊，尝试通过联系人查找
		if len(chatRooms) == 0 {
			contacts, err := ds.GetContacts(ctx, key, 1, 0)
			if err == nil && len(contacts) > 0 && strings.HasSuffix(contacts[0].UserName, "@chatroom") {
				// 再次尝试通过用户名查找群聊
				rows, err := db.QueryContext(ctx,
					`SELECT ChatRoomName, Reserved2, RoomData FROM ChatRoom WHERE ChatRoomName = ?`,
					contacts[0].UserName)

				if err != nil {
					return nil, errors.QueryFailed(query, err)
				}
				defer rows.Close()

				for rows.Next() {
					var chatRoomV3 model.ChatRoomV3
					err := rows.Scan(
						&chatRoomV3.ChatRoomName,
						&chatRoomV3.Reserved2,
						&chatRoomV3.RoomData,
					)

					if err != nil {
						return nil, errors.ScanRowFailed(err)
					}

					chatRooms = append(chatRooms, chatRoomV3.Wrap())
				}

				// 如果群聊记录不存在，但联系人记录存在，创建一个模拟的群聊对象
				if len(chatRooms) == 0 {
					chatRooms = append(chatRooms, &model.ChatRoom{
						Name:             contacts[0].UserName,
						Users:            make([]model.ChatRoomUser, 0),
						User2DisplayName: make(map[string]string),
					})
				}
			}
		}

		return chatRooms, nil
	} else {
		// 查询所有群聊
		query = `SELECT ChatRoomName, Reserved2, RoomData FROM ChatRoom`

		// 添加排序、分页
		query += ` ORDER BY ChatRoomName`
		if limit > 0 {
			query += fmt.Sprintf(" LIMIT %d", limit)
			if offset > 0 {
				query += fmt.Sprintf(" OFFSET %d", offset)
			}
		}

		// 执行查询
		db, err := ds.dbm.GetDB(Contact)
		if err != nil {
			return nil, err
		}
		rows, err := db.QueryContext(ctx, query, args...)
		if err != nil {
			return nil, errors.QueryFailed(query, err)
		}
		defer rows.Close()

		chatRooms := []*model.ChatRoom{}
		for rows.Next() {
			var chatRoomV3 model.ChatRoomV3
			err := rows.Scan(
				&chatRoomV3.ChatRoomName,
				&chatRoomV3.Reserved2,
				&chatRoomV3.RoomData,
			)

			if err != nil {
				return nil, errors.ScanRowFailed(err)
			}

			chatRooms = append(chatRooms, chatRoomV3.Wrap())
		}

		return chatRooms, nil
	}
}

// GetSessions 实现获取会话信息的方法
func (ds *DataSource) GetSessions(ctx context.Context, key string, limit, offset int) ([]*model.Session, error) {
	var query string
	var args []interface{}

	if key != "" {
		// 按照关键字查询
		query = `SELECT strUsrName, nOrder, strNickName, strContent, nTime 
                FROM Session 
                WHERE strUsrName = ? OR strNickName = ?
                ORDER BY nOrder DESC`
		args = []interface{}{key, key}
	} else {
		// 查询所有会话
		query = `SELECT strUsrName, nOrder, strNickName, strContent, nTime 
                FROM Session 
                ORDER BY nOrder DESC`
	}

	// 添加分页
	if limit > 0 {
		query += fmt.Sprintf(" LIMIT %d", limit)
		if offset > 0 {
			query += fmt.Sprintf(" OFFSET %d", offset)
		}
	}

	// 执行查询
	db, err := ds.dbm.GetDB(Contact)
	if err != nil {
		return nil, err
	}
	rows, err := db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, errors.QueryFailed(query, err)
	}
	defer rows.Close()

	sessions := []*model.Session{}
	for rows.Next() {
		var sessionV3 model.SessionV3
		err := rows.Scan(
			&sessionV3.StrUsrName,
			&sessionV3.NOrder,
			&sessionV3.StrNickName,
			&sessionV3.StrContent,
			&sessionV3.NTime,
		)

		if err != nil {
			return nil, errors.ScanRowFailed(err)
		}

		sessions = append(sessions, sessionV3.Wrap())
	}

	return sessions, nil
}

func (ds *DataSource) GetMedia(ctx context.Context, _type string, key string) (*model.Media, error) {
	if key == "" {
		return nil, errors.ErrKeyEmpty
	}

	if _type == "voice" {
		return ds.GetVoice(ctx, key)
	}

	md5key, err := hex.DecodeString(key)
	if err != nil {
		return nil, errors.DecodeKeyFailed(err)
	}

	var dbType string
	var table1, table2 string

	switch _type {
	case "image":
		dbType = Image
		table1 = "HardLinkImageAttribute"
		table2 = "HardLinkImageID"
	case "video":
		dbType = Video
		table1 = "HardLinkVideoAttribute"
		table2 = "HardLinkVideoID"
	case "file":
		dbType = File
		table1 = "HardLinkFileAttribute"
		table2 = "HardLinkFileID"
	default:
		return nil, errors.MediaTypeUnsupported(_type)
	}

	db, err := ds.dbm.GetDB(dbType)
	if err != nil {
		return nil, err
	}

	query := fmt.Sprintf(`
        SELECT 
            a.FileName,
            a.ModifyTime,
            IFNULL(d1.Dir,"") AS Dir1,
            IFNULL(d2.Dir,"") AS Dir2
        FROM 
            %s a
        LEFT JOIN 
            %s d1 ON a.DirID1 = d1.DirId
        LEFT JOIN 
            %s d2 ON a.DirID2 = d2.DirId
        WHERE 
            a.Md5 = ?
    `, table1, table2, table2)
	args := []interface{}{md5key}

	rows, err := db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, errors.QueryFailed(query, err)
	}
	defer rows.Close()

	var media *model.Media
	for rows.Next() {
		var mediaV3 model.MediaV3
		err := rows.Scan(
			&mediaV3.Name,
			&mediaV3.ModifyTime,
			&mediaV3.Dir1,
			&mediaV3.Dir2,
		)
		if err != nil {
			return nil, errors.ScanRowFailed(err)
		}
		mediaV3.Type = _type
		mediaV3.Key = key
		media = mediaV3.Wrap()
	}

	if media == nil {
		return nil, errors.ErrMediaNotFound
	}

	return media, nil
}

func (ds *DataSource) GetVoice(ctx context.Context, key string) (*model.Media, error) {
	if key == "" {
		return nil, errors.ErrKeyEmpty
	}

	query := `
	SELECT Buf
	FROM Media
	WHERE Reserved0 = ? 
	`
	args := []interface{}{key}

	dbs, err := ds.dbm.GetDBs(Voice)
	if err != nil {
		return nil, errors.DBConnectFailed("", err)
	}

	for _, db := range dbs {
		rows, err := db.QueryContext(ctx, query, args...)
		if err != nil {
			return nil, errors.QueryFailed(query, err)
		}
		defer rows.Close()

		for rows.Next() {
			var voiceData []byte
			err := rows.Scan(
				&voiceData,
			)
			if err != nil {
				return nil, errors.ScanRowFailed(err)
			}
			if len(voiceData) > 0 {
				return &model.Media{
					Type: "voice",
					Key:  key,
					Data: voiceData,
				}, nil
			}
		}
	}

	return nil, errors.ErrMediaNotFound
}

// Close 实现 DataSource 接口的 Close 方法
func (ds *DataSource) Close() error {
	return ds.dbm.Close()
}

// GetAvatar returns avatar info for a username on Windows v3 (MicroMsg.db -> ContactHeadImgUrl)
func (ds *DataSource) GetAvatar(ctx context.Context, username string, size string) (*model.Avatar, error) {
	if username == "" {
		return nil, errors.ErrKeyEmpty
	}
	db, err := ds.dbm.GetDB(Contact)
	if err != nil {
		return nil, err
	}
	query := `SELECT IFNULL(smallHeadImgUrl, ''), IFNULL(bigHeadImgUrl, '') FROM ContactHeadImgUrl WHERE usrName = ?`
	row := db.QueryRowContext(ctx, query, username)
	var small, big string
	if err := row.Scan(&small, &big); err != nil {
		return nil, errors.ErrAvatarNotFound
	}
	url := small
	if strings.ToLower(size) == "big" && big != "" {
		url = big
	}
	if url == "" {
		url = big
	}
	if url == "" {
		return nil, errors.ErrAvatarNotFound
	}
	return &model.Avatar{Username: username, URL: url}, nil
}

// GlobalMessageStats 聚合统计（Windows v3）
func (ds *DataSource) GlobalMessageStats(ctx context.Context) (*model.GlobalMessageStats, error) {
	stats := &model.GlobalMessageStats{ByType: make(map[string]int64)}
	dbs, err := ds.dbm.GetDBs(Message)
	if err != nil {
		return stats, nil
	}
	for _, db := range dbs {
		// total/sent/recv/min/max
		row := db.QueryRowContext(ctx, `SELECT 
			COUNT(*) AS total,
			SUM(CASE WHEN IsSender=1 THEN 1 ELSE 0 END) AS sent,
			SUM(CASE WHEN IsSender=0 THEN 1 ELSE 0 END) AS recv,
			MIN(CreateTime) AS minct,
			MAX(CreateTime) AS maxct
		FROM MSG`)
		var total, sent, recv, minct, maxct int64
		if err := row.Scan(&total, &sent, &recv, &minct, &maxct); err == nil {
			stats.Total += total
			stats.Sent += sent
			stats.Received += recv
			if stats.EarliestUnix == 0 || (minct > 0 && minct < stats.EarliestUnix) {
				stats.EarliestUnix = minct
			}
			if maxct > stats.LatestUnix {
				stats.LatestUnix = maxct
			}
		}

		// By type/subtype
		rows, err := db.QueryContext(ctx, `SELECT Type, SubType, COUNT(*) FROM MSG GROUP BY Type, SubType`)
		if err == nil {
			for rows.Next() {
				var t int64
				var st int
				var cnt int64
				if err := rows.Scan(&t, &st, &cnt); err == nil {
					label := mapV3TypeToLabel(t, int64(st))
					if label != "" {
						stats.ByType[label] += cnt
					}
				}
			}
			rows.Close()
		}
	}
	return stats, nil
}

// GroupMessageCounts 统计群聊消息数量（Windows v3）
func (ds *DataSource) GroupMessageCounts(ctx context.Context) (map[string]int64, error) {
	result := make(map[string]int64)
	dbs, err := ds.dbm.GetDBs(Message)
	if err != nil {
		return result, nil
	}
	for _, db := range dbs {
		rows, err := db.QueryContext(ctx, `SELECT StrTalker, COUNT(*) FROM MSG WHERE StrTalker LIKE '%@chatroom' GROUP BY StrTalker`)
		if err != nil {
			continue
		}
		for rows.Next() {
			var talker string
			var cnt int64
			if err := rows.Scan(&talker, &cnt); err == nil {
				result[talker] += cnt
			}
		}
		rows.Close()
	}
	return result, nil
}

// GroupTodayMessageCounts 统计群聊今日消息数（Windows v3）：MSG 表中 StrTalker LIKE '%@chatroom' 且 CreateTime >= 今日零点
func (ds *DataSource) GroupTodayMessageCounts(ctx context.Context) (map[string]int64, error) {
	result := make(map[string]int64)
	dbs, err := ds.dbm.GetDBs(Message)
	if err != nil {
		return result, nil
	}
	// 今日零点（使用本地时区）
	now := time.Now()
	today := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location())
	since := today.Unix()
	for _, db := range dbs {
		rows, err := db.QueryContext(ctx, `SELECT StrTalker, COUNT(*) FROM MSG WHERE StrTalker LIKE '%@chatroom' AND CreateTime >= ? GROUP BY StrTalker`, since)
		if err != nil {
			continue
		}
		for rows.Next() {
			var talker string
			var cnt int64
			if err := rows.Scan(&talker, &cnt); err == nil {
				result[talker] += cnt
			}
		}
		rows.Close()
	}
	return result, nil
}

// GroupTodayHourly 统计群聊今日按小时消息数（Windows v3）
func (ds *DataSource) GroupTodayHourly(ctx context.Context) (map[string][24]int64, error) {
	result := make(map[string][24]int64)
	dbs, err := ds.dbm.GetDBs(Message)
	if err != nil {
		return result, nil
	}
	now := time.Now()
	start := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location()).Unix()
	end := start + 86400
	for _, db := range dbs {
		rows, err := db.QueryContext(ctx, `SELECT StrTalker, CAST(strftime('%H', datetime(CreateTime,'unixepoch')) AS INTEGER) AS h, COUNT(*) FROM MSG WHERE StrTalker LIKE '%@chatroom' AND CreateTime >= ? AND CreateTime < ? GROUP BY StrTalker, h`, start, end)
		if err != nil {
			continue
		}
		for rows.Next() {
			var talker string
			var hour int
			var cnt int64
			if rows.Scan(&talker, &hour, &cnt) == nil {
				if hour >= 0 && hour < 24 {
					bucket := result[talker]
					bucket[hour] += cnt
					result[talker] = bucket
				}
			}
		}
		rows.Close()
	}
	return result, nil
}

// GroupWeekMessageCount 统计本周(周一00:00起)所有群聊消息总数（Windows v3）
func (ds *DataSource) GroupWeekMessageCount(ctx context.Context) (int64, error) {
	var total int64
	dbs, err := ds.dbm.GetDBs(Message)
	if err != nil {
		return 0, nil
	}
	now := time.Now()
	wday := int(now.Weekday())
	offset := wday - 1
	if wday == 0 {
		offset = -6
	}
	monday := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location()).AddDate(0, 0, -offset)
	since := monday.Unix()
	for _, db := range dbs {
		row := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM MSG WHERE StrTalker LIKE '%@chatroom' AND CreateTime >= ?`, since)
		var cnt int64
		if row.Scan(&cnt) == nil {
			total += cnt
		}
	}
	return total, nil
}

// GroupMessageTypeStats 统计群聊消息类型分布（Windows v3）
func (ds *DataSource) GroupMessageTypeStats(ctx context.Context) (map[string]int64, error) {
	result := make(map[string]int64)
	dbs, err := ds.dbm.GetDBs(Message)
	if err != nil {
		return result, nil
	}
	for _, db := range dbs {
		rows, err := db.QueryContext(ctx, `SELECT Type, SubType, COUNT(*) FROM MSG WHERE StrTalker LIKE '%@chatroom' GROUP BY Type, SubType`)
		if err != nil {
			continue
		}
		for rows.Next() {
			var t int64
			var st int64
			var cnt int64
			if rows.Scan(&t, &st, &cnt) == nil {
				label := mapV3TypeToLabel(t, st)
				if label != "" {
					result[label] += cnt
				}
			}
		}
		rows.Close()
	}
	return result, nil
}

// MonthlyTrend 返回每月 sent/received（近 months 月，若 months<=0 则返回全部）
func (ds *DataSource) MonthlyTrend(ctx context.Context, months int) ([]model.MonthlyTrend, error) {
	agg := make(map[string][2]int64)
	dbs, err := ds.dbm.GetDBs(Message)
	if err != nil {
		return []model.MonthlyTrend{}, nil
	}
	for _, db := range dbs {
		rows, err := db.QueryContext(ctx, `SELECT strftime('%Y-%m', datetime(CreateTime, 'unixepoch')) AS ym,
			SUM(CASE WHEN IsSender=1 THEN 1 ELSE 0 END) AS sent,
			SUM(CASE WHEN IsSender=0 THEN 1 ELSE 0 END) AS recv
			FROM MSG GROUP BY ym ORDER BY ym`)
		if err != nil {
			continue
		}
		for rows.Next() {
			var ym string
			var sent, recv int64
			if err := rows.Scan(&ym, &sent, &recv); err == nil {
				cur := agg[ym]
				cur[0] += sent
				cur[1] += recv
				agg[ym] = cur
			}
		}
		rows.Close()
	}
	trends := make([]model.MonthlyTrend, 0, len(agg))
	// order is not guaranteed; we'll reconstruct sorted keys
	// but to keep simple, iterate map and collect; sorting can be added later
	for ym, v := range agg {
		trends = append(trends, model.MonthlyTrend{Date: ym, Sent: v[0], Received: v[1]})
	}
	// optional: limit to recent months by trimming after sort
	return trends, nil
}

// Heatmap 小时x星期（wday: 0=Sunday..6）
func (ds *DataSource) Heatmap(ctx context.Context) ([24][7]int64, error) {
	var grid [24][7]int64
	dbs, err := ds.dbm.GetDBs(Message)
	if err != nil {
		return grid, nil
	}
	for _, db := range dbs {
		rows, err := db.QueryContext(ctx, `SELECT CAST(strftime('%H', datetime(CreateTime,'unixepoch')) AS INTEGER) AS h,
			CAST(strftime('%w', datetime(CreateTime,'unixepoch')) AS INTEGER) AS d,
			COUNT(*) FROM MSG GROUP BY h,d`)
		if err != nil {
			continue
		}
		for rows.Next() {
			var h, d int
			var cnt int64
			if err := rows.Scan(&h, &d, &cnt); err == nil {
				if h >= 0 && h < 24 && d >= 0 && d < 7 {
					grid[h][d] += cnt
				}
			}
		}
		rows.Close()
	}
	return grid, nil
}

// GlobalTodayHourly 返回今日(本地时区)每小时全部消息量（含私聊+群聊）
func (ds *DataSource) GlobalTodayHourly(ctx context.Context) ([24]int64, error) {
	var hours [24]int64
	dbs, err := ds.dbm.GetDBs(Message)
	if err != nil {
		return hours, nil
	}
	now := time.Now()
	start := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location()).Unix()
	end := start + 86400
	for _, db := range dbs {
		rows, err := db.QueryContext(ctx, `SELECT CAST(strftime('%H', datetime(CreateTime,'unixepoch')) AS INTEGER) AS h, COUNT(*) FROM MSG WHERE CreateTime >= ? AND CreateTime < ? GROUP BY h`, start, end)
		if err != nil {
			continue
		}
		for rows.Next() {
			var h int
			var cnt int64
			if rows.Scan(&h, &cnt) == nil {
				if h >= 0 && h < 24 {
					hours[h] += cnt
				}
			}
		}
		rows.Close()
	}
	return hours, nil
}

// IntimacyBase 统计按联系人（非群聊）聚合的亲密度基础数据（Windows v3）
func (ds *DataSource) IntimacyBase(ctx context.Context) (map[string]*model.IntimacyBase, error) {
	result := make(map[string]*model.IntimacyBase)

	dbs, err := ds.dbm.GetDBs(Message)
	if err != nil {
		return result, nil
	}

	// 先获取全局最新时间戳
	var maxCT int64
	for _, db := range dbs {
		row := db.QueryRowContext(ctx, `SELECT MAX(CreateTime) FROM MSG`)
		var v sql.NullInt64
		if err := row.Scan(&v); err == nil && v.Valid && v.Int64 > maxCT {
			maxCT = v.Int64
		}
	}
	if maxCT == 0 {
		return result, nil
	}
	since90 := maxCT - 90*86400
	since7 := maxCT - 7*86400

	for _, db := range dbs {
		// 基础计数
		rows, err := db.QueryContext(ctx, `SELECT StrTalker,
			COUNT(*) AS msg_count,
			MIN(CreateTime) AS minct,
			MAX(CreateTime) AS maxct,
			SUM(CASE WHEN IsSender=1 THEN 1 ELSE 0 END) AS sent,
			SUM(CASE WHEN IsSender=0 THEN 1 ELSE 0 END) AS recv
			FROM MSG WHERE StrTalker NOT LIKE '%@chatroom' GROUP BY StrTalker`)
		if err == nil {
			for rows.Next() {
				var talker string
				var msgCnt, minct, maxct, sent, recv int64
				if err := rows.Scan(&talker, &msgCnt, &minct, &maxct, &sent, &recv); err == nil {
					base := result[talker]
					if base == nil {
						base = &model.IntimacyBase{UserName: talker}
						result[talker] = base
					}
					base.MsgCount += msgCnt
					base.SentCount += sent
					base.ReceivedCount += recv
					if base.MinCreateUnix == 0 || (minct > 0 && minct < base.MinCreateUnix) {
						base.MinCreateUnix = minct
					}
					if maxct > base.MaxCreateUnix {
						base.MaxCreateUnix = maxct
					}
				}
			}
			rows.Close()
		}

		// 活跃天数（全期间）
		rows2, err := db.QueryContext(ctx, `SELECT StrTalker, COUNT(DISTINCT date(datetime(CreateTime,'unixepoch'))) AS days
			FROM MSG WHERE StrTalker NOT LIKE '%@chatroom' GROUP BY StrTalker`)
		if err == nil {
			for rows2.Next() {
				var talker string
				var days int64
				if err := rows2.Scan(&talker, &days); err == nil {
					base := result[talker]
					if base == nil {
						base = &model.IntimacyBase{UserName: talker}
						result[talker] = base
					}
					base.MessagingDays += days
				}
			}
			rows2.Close()
		}

		// 最近90天消息数
		rows3, err := db.QueryContext(ctx, `SELECT StrTalker, COUNT(*) FROM MSG WHERE CreateTime>=? AND StrTalker NOT LIKE '%@chatroom' GROUP BY StrTalker`, since90)
		if err == nil {
			for rows3.Next() {
				var talker string
				var cnt int64
				if err := rows3.Scan(&talker, &cnt); err == nil {
					base := result[talker]
					if base == nil {
						base = &model.IntimacyBase{UserName: talker}
						result[talker] = base
					}
					base.Last90DaysMsg += cnt
				}
			}
			rows3.Close()
		}

		// 过去7天发送数
		rows4, err := db.QueryContext(ctx, `SELECT StrTalker, SUM(CASE WHEN IsSender=1 THEN 1 ELSE 0 END) FROM MSG WHERE CreateTime>=? AND StrTalker NOT LIKE '%@chatroom' GROUP BY StrTalker`, since7)
		if err == nil {
			for rows4.Next() {
				var talker string
				var cnt sql.NullInt64
				if err := rows4.Scan(&talker, &cnt); err == nil {
					base := result[talker]
					if base == nil {
						base = &model.IntimacyBase{UserName: talker}
						result[talker] = base
					}
					if cnt.Valid {
						base.Past7DaysSentMsg += cnt.Int64
					}
				}
			}
			rows4.Close()
		}
	}

	return result, nil
}

// mapV3TypeToLabel 将v3 (Type,SubType) 映射为中文类别
func mapV3TypeToLabel(t int64, st int64) string {
	// Windows v3：在统一文档分类基础上补充文件/链接细分（Type=49 根据 SubType 区分）
	switch t {
	case 1:
		return "文本消息"
	case 3:
		return "图片消息"
	case 34:
		return "语音消息"
	case 37:
		return "好友验证消息"
	case 42:
		return "好友推荐消息"
	case 47:
		return "聊天表情"
	case 48:
		return "位置消息"
	case 49:
		if st == 6 { // 文件
			return "文件消息"
		}
		if st == 4 || st == 5 { // 链接
			return "链接消息"
		}
		return "XML消息" // 其他未识别子类型归类为通用 XML
	case 50:
		return "音视频通话"
	case 51:
		return "手机端操作消息"
	case 10000:
		return "系统通知"
	case 10002:
		return "撤回消息"
	default:
		return ""
	}
}