seqlog/query.go

package seqlog

import (
	"encoding/binary"
	"fmt"
	"io"
	"os"
)

// RecordStatus 记录处理状态
type RecordStatus int

const (
	StatusProcessed   RecordStatus = iota // 已处理
	StatusProcessing                      // 处理中（当前位置）
	StatusPending                         // 待处理
	StatusUnavailable                     // 不可用（尚未写入）
)

// String 返回状态的字符串表示
func (s RecordStatus) String() string {
	switch s {
	case StatusProcessed:
		return "StatusProcessed"
	case StatusProcessing:
		return "StatusProcessing"
	case StatusPending:
		return "StatusPending"
	case StatusUnavailable:
		return "StatusUnavailable"
	default:
		return "StatusUnknown"
	}
}

// RecordWithStatus 带状态的记录
type RecordWithStatus struct {
	Record *Record
	Status RecordStatus
}

// RecordQuery 记录查询器
type RecordQuery struct {
	logPath string
	fd      *os.File
	rbuf    []byte       // 复用读缓冲区
	index   *RecordIndex // 索引文件管理器（来自外部）
}

// NewRecordQuery 创建记录查询器
// index 参数必须由外部提供，确保所有组件使用同一个索引实例
func NewRecordQuery(logPath string, index *RecordIndex) (*RecordQuery, error) {
	if index == nil {
		return nil, fmt.Errorf("index cannot be nil")
	}

	fd, err := os.Open(logPath)
	if err != nil {
		return nil, fmt.Errorf("open log file: %w", err)
	}

	rq := &RecordQuery{
		logPath: logPath,
		fd:      fd,
		rbuf:    make([]byte, 8<<20), // 8 MiB 缓冲区
		index:   index,
	}

	return rq, nil
}

// readRecordAtOffset 读取指定偏移位置的记录
func (rq *RecordQuery) readRecordAtOffset(offset int64) (*Record, error) {
	if _, err := rq.fd.Seek(offset, 0); err != nil {
		return nil, fmt.Errorf("seek to offset %d: %w", offset, err)
	}

	// 读取头部：[4B len][4B CRC][16B UUID] = 24 字节
	hdr := rq.rbuf[:24]
	if _, err := io.ReadFull(rq.fd, hdr); err != nil {
		return nil, fmt.Errorf("read header: %w", err)
	}

	rec := &Record{
		Len: binary.LittleEndian.Uint32(hdr[0:4]),
		CRC: binary.LittleEndian.Uint32(hdr[4:8]),
	}
	copy(rec.UUID[:], hdr[8:24])

	// 读取数据
	rec.Data = make([]byte, rec.Len)
	if _, err := io.ReadFull(rq.fd, rec.Data); err != nil {
		return nil, fmt.Errorf("read data: %w", err)
	}

	return rec, nil
}

// readRecordsForward 从指定索引位置向前顺序读取记录
// startIndex: 起始记录索引
// count: 读取数量
// startIdx, endIdx: 游标窗口索引范围（用于状态判断）
func (rq *RecordQuery) readRecordsForward(startIndex, count int, startIdx, endIdx int) ([]*RecordWithStatus, error) {
	// 获取起始 offset
	startOffset, err := rq.index.GetOffset(startIndex)
	if err != nil {
		return nil, fmt.Errorf("get start offset: %w", err)
	}

	if _, err := rq.fd.Seek(startOffset, 0); err != nil {
		return nil, fmt.Errorf("seek to offset %d: %w", startOffset, err)
	}

	results := make([]*RecordWithStatus, 0, count)
	currentIndex := startIndex
	currentOffset := startOffset

	for len(results) < count {
		// 读取头部：[4B len][4B CRC][16B UUID] = 24 字节
		hdr := rq.rbuf[:24]
		if _, err := io.ReadFull(rq.fd, hdr); err != nil {
			if err == io.EOF {
				break
			}
			return nil, fmt.Errorf("read header at offset %d: %w", currentOffset, err)
		}

		rec := &Record{
			Len: binary.LittleEndian.Uint32(hdr[0:4]),
			CRC: binary.LittleEndian.Uint32(hdr[4:8]),
		}
		copy(rec.UUID[:], hdr[8:24])

		// 读取数据
		rec.Data = make([]byte, rec.Len)
		if _, err := io.ReadFull(rq.fd, rec.Data); err != nil {
			return nil, fmt.Errorf("read data at offset %d: %w", currentOffset, err)
		}

		results = append(results, &RecordWithStatus{
			Record: rec,
			Status: rq.getRecordStatus(currentIndex, startIdx, endIdx),
		})

		currentIndex++
		currentOffset += 24 + int64(rec.Len)
	}

	return results, nil
}

// getRecordStatus 根据游标窗口索引位置获取记录状态
func (rq *RecordQuery) getRecordStatus(recordIndex, startIdx, endIdx int) RecordStatus {
	if recordIndex < startIdx {
		return StatusProcessed
	} else if recordIndex >= startIdx && recordIndex < endIdx {
		return StatusProcessing
	} else {
		return StatusPending
	}
}

// QueryOldest 从指定索引开始查询记录（向前读取）
// startIndex: 查询起始索引
// count: 查询数量
// startIdx, endIdx: 游标窗口索引范围（用于状态判断）
// 返回的记录按时间顺序（索引递增方向）
func (rq *RecordQuery) QueryOldest(startIndex, count int, startIdx, endIdx int) ([]*RecordWithStatus, error) {
	if count <= 0 {
		return nil, fmt.Errorf("count must be greater than 0")
	}

	totalCount := rq.index.Count()
	if totalCount == 0 {
		return []*RecordWithStatus{}, nil
	}

	// 校验起始索引
	if startIndex < 0 {
		startIndex = 0
	}
	if startIndex >= totalCount {
		return []*RecordWithStatus{}, nil
	}

	// 限制查询数量
	remainCount := totalCount - startIndex
	if count > remainCount {
		count = remainCount
	}

	return rq.readRecordsForward(startIndex, count, startIdx, endIdx)
}

// QueryNewest 从指定索引开始向后查询记录（索引递减方向）
// endIndex: 查询结束索引（包含，最新的记录）
// count: 查询数量
// startIdx, endIdx: 游标窗口索引范围（用于状态判断）
// 返回结果按时间倒序（最新在前，即 endIndex 对应的记录在最前）
func (rq *RecordQuery) QueryNewest(endIndex, count int, startIdx, endIdx int) ([]*RecordWithStatus, error) {
	if count <= 0 {
		return nil, fmt.Errorf("count must be greater than 0")
	}

	totalCount := rq.index.Count()
	if totalCount == 0 {
		return []*RecordWithStatus{}, nil
	}

	// 校验结束索引
	if endIndex < 0 {
		return []*RecordWithStatus{}, nil
	}
	if endIndex >= totalCount {
		endIndex = totalCount - 1
	}

	// 计算实际起始索引（向前推 count-1 条）
	queryStartIdx := endIndex - count + 1
	if queryStartIdx < 0 {
		queryStartIdx = 0
		count = endIndex + 1 // 调整实际数量
	}

	// 向前读取
	results, err := rq.readRecordsForward(queryStartIdx, count, startIdx, endIdx)
	if err != nil {
		return nil, err
	}

	// 反转结果，使最新的在前
	for i, j := 0, len(results)-1; i < j; i, j = i+1, j-1 {
		results[i], results[j] = results[j], results[i]
	}

	return results, nil
}

// QueryAt 从指定位置查询记录
// position: 查询起始位置（文件偏移量，通常是当前处理位置）
// direction: 查询方向（负数向后，0 当前，正数向前）
// count: 查询数量
// startIdx, endIdx: 游标窗口索引范围（用于状态判断）
// 返回结果按时间顺序排列
func (rq *RecordQuery) QueryAt(position int64, direction int, count int, startIdx, endIdx int) ([]*RecordWithStatus, error) {
	// 将 position 转换为索引
	idx := rq.index.FindIndex(position)
	if idx < 0 {
		return nil, fmt.Errorf("position not found in index")
	}

	if direction >= 0 {
		// 向前查询或当前位置
		if direction == 0 {
			count = 1
		} else {
			// direction > 0，跳过当前位置，从下一条开始
			idx++
		}
		return rq.readRecordsForward(idx, count, startIdx, endIdx)
	}

	// 向后查询：使用索引
	results := make([]*RecordWithStatus, 0, count)

	// 向后查询（更早的记录）
	for i := idx - 1; i >= 0 && len(results) < count; i-- {
		offset, err := rq.index.GetOffset(i)
		if err != nil {
			return nil, fmt.Errorf("get offset at index %d: %w", i, err)
		}

		rec, err := rq.readRecordAtOffset(offset)
		if err != nil {
			return nil, fmt.Errorf("read record at index %d: %w", i, err)
		}

		results = append(results, &RecordWithStatus{
			Record: rec,
			Status: rq.getRecordStatus(i, startIdx, endIdx),
		})
	}

	// 反转结果，使其按时间顺序排列
	for i, j := 0, len(results)-1; i < j; i, j = i+1, j-1 {
		results[i], results[j] = results[j], results[i]
	}

	return results, nil
}

// GetRecordCount 获取记录总数
func (rq *RecordQuery) GetRecordCount() (int, error) {
	return rq.index.Count(), nil
}

// Close 关闭查询器
// 注意：不关闭 index，因为 index 是外部管理的
func (rq *RecordQuery) Close() error {
	// 只关闭日志文件
	if rq.fd != nil {
		return rq.fd.Close()
	}
	return nil
}