重构：优化记录格式并修复核心功能

- 修改记录存储格式为 [4B len][8B offset][4B CRC][16B UUID][data] - 修复 TopicProcessor 中 WaitGroup 使用错误导致 handler 不执行的问题 - 修复写入保护逻辑，避免 dirtyOffset=-1 时误判为写入中 - 添加统计信息定期持久化功能 - 改进 UTF-8 字符截断处理，防止 CJK 字符乱码 - 优化 Web UI：显示人类可读的文件大小，支持点击外部关闭弹窗 - 重构示例代码，添加 webui 和 webui_integration 示例 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
2025-10-04 17:54:49 +08:00
parent 955a467248
commit 810664eb12
18 changed files with 1810 additions and 1170 deletions
--- a/topic_processor.go
+++ b/topic_processor.go
@@ -13,15 +13,16 @@ import (
 // TopicProcessor 作为聚合器，持有所有核心组件并提供统一的访问接口
 type TopicProcessor struct {
 	topic   string
+	title   string // 显示标题，用于 UI 展示
 	logPath string
 	logger  *slog.Logger

 	// 核心组件（聚合）
-	writer *LogWriter   // 写入器
-	index  *RecordIndex // 索引管理器
-	query  *RecordQuery // 查询器
-	cursor *ProcessCursor   // 游标
-	tailer *LogTailer   // 持续处理器
+	writer *LogWriter     // 写入器
+	index  *RecordIndex   // 索引管理器
+	query  *RecordQuery   // 查询器
+	cursor *ProcessCursor // 游标
+	tailer *LogTailer     // 持续处理器

 	// 配置和状态
 	handler    RecordHandler
@@ -39,6 +40,7 @@ type TopicProcessor struct {

 // TopicConfig topic 配置
 type TopicConfig struct {
+	Title      string        // 显示标题，可选，默认为 topic 名称
 	Handler    RecordHandler // 处理函数（必填）
 	TailConfig *TailConfig   // tail 配置，可选
 }
@@ -71,8 +73,15 @@ func NewTopicProcessor(baseDir, topic string, logger *slog.Logger, config *Topic
 	logPath := filepath.Join(baseDir, topic+".log")
 	statsPath := filepath.Join(baseDir, topic+".stats")

+	// 设置 title，如果未提供则使用 topic 名称
+	title := config.Title
+	if title == "" {
+		title = topic
+	}
+
 	tp := &TopicProcessor{
 		topic:      topic,
+		title:      title,
 		logPath:    logPath,
 		logger:     logger,
 		handler:    config.Handler,
@@ -110,7 +119,7 @@ func (tp *TopicProcessor) initializeComponents() error {
 	tp.writer = writer

 	// 3. 创建查询器（使用共享 index）
-	query, err := NewRecordQuery(tp.logPath, tp.index)
+	query, err := NewRecordQuery(tp.logPath, tp.index, tp.writer)
 	if err != nil {
 		tp.writer.Close()
 		tp.index.Close()
@@ -118,8 +127,8 @@ func (tp *TopicProcessor) initializeComponents() error {
 	}
 	tp.query = query

-	// 4. 创建游标（使用共享 index）
-	cursor, err := NewCursor(tp.logPath, tp.index)
+	// 4. 创建游标（使用共享 index 和 writer）
+	cursor, err := NewCursor(tp.logPath, tp.index, tp.writer)
 	if err != nil {
 		tp.query.Close()
 		tp.writer.Close()
@@ -241,17 +250,36 @@ func (tp *TopicProcessor) Start() error {

 	tp.running = true

-	// 如果 tailer 已创建，启动它
-	if tp.tailer != nil {
-		tp.logger.Debug("launching tailer goroutine")
-		tp.wg.Go(func() {
-			tp.logger.Debug("tailer goroutine started")
-			if err := tp.tailer.Start(tp.ctx); err != nil && err != context.Canceled {
-				tp.logger.Error("tailer error", "error", err)
+	// 启动定期保存统计信息的 goroutine
+	tp.wg.Add(1)
+	go func() {
+		defer tp.wg.Done()
+		ticker := time.NewTicker(tp.tailConfig.SaveInterval)
+		defer ticker.Stop()
+
+		for {
+			select {
+			case <-tp.ctx.Done():
+				return
+			case <-ticker.C:
+				if err := tp.stats.Save(); err != nil {
+					tp.logger.Error("failed to save stats", "error", err)
+				}
 			}
-			tp.logger.Debug("tailer goroutine finished")
-		})
-	}
+		}
+	}()
+
+	// 如果 tailer 已创建，启动它
+	tp.logger.Debug("launching tailer goroutine")
+	tp.wg.Add(1)
+	go func() {
+		defer tp.wg.Done()
+		tp.logger.Debug("tailer goroutine started")
+		if err := tp.tailer.Start(tp.ctx); err != nil && err != context.Canceled {
+			tp.logger.Error("tailer error", "error", err)
+		}
+		tp.logger.Debug("tailer goroutine finished")
+	}()

 	// 发布启动事件
 	tp.eventBus.Publish(&Event{
@@ -295,6 +323,11 @@ func (tp *TopicProcessor) Topic() string {
 	return tp.topic
 }

+// Title 返回显示标题
+func (tp *TopicProcessor) Title() string {
+	return tp.title
+}
+
 // IsRunning 检查是否正在运行
 func (tp *TopicProcessor) IsRunning() bool {
 	tp.mu.RLock()
@@ -363,6 +396,29 @@ func (tp *TopicProcessor) addStatusToRecords(records []*RecordWithIndex) []*Reco
 	return results
 }

+// addStatusToMetadata 为元数据添加状态信息
+func (tp *TopicProcessor) addStatusToMetadata(metadata []*RecordMetadata) []*RecordMetadataWithStatus {
+	// 获取窗口索引范围（用于状态判断）
+	var startIdx, endIdx int
+	tp.mu.RLock()
+	if tp.tailer != nil {
+		startIdx = tp.tailer.GetStartIndex()
+		endIdx = tp.tailer.GetEndIndex()
+	}
+	tp.mu.RUnlock()
+
+	// 为每个元数据添加状态
+	results := make([]*RecordMetadataWithStatus, len(metadata))
+	for i, meta := range metadata {
+		results[i] = &RecordMetadataWithStatus{
+			Metadata: meta,
+			Status:   GetRecordStatus(meta.Index, startIdx, endIdx),
+		}
+	}
+
+	return results
+}
+
 // QueryOldest 从参考索引向索引递减方向查询记录（查询更早的记录）
 // refIndex: 参考索引位置
 // count: 查询数量
@@ -389,6 +445,58 @@ func (tp *TopicProcessor) QueryNewest(refIndex, count int) ([]*RecordWithStatus,
 	return tp.addStatusToRecords(records), nil
 }

+// QueryOldestMetadata 从参考索引向索引递减方向查询记录元数据（查询更早的记录，不读取完整数据）
+// refIndex: 参考索引位置
+// count: 查询数量
+// 返回的记录只包含元数据信息（索引、UUID、数据大小），按索引递增方向排序
+// 例如：QueryOldestMetadata(5, 3) 查询索引 2, 3, 4（不包含 5）
+func (tp *TopicProcessor) QueryOldestMetadata(refIndex, count int) ([]*RecordMetadataWithStatus, error) {
+	metadata, err := tp.query.QueryOldestMetadata(refIndex, count)
+	if err != nil {
+		return nil, err
+	}
+	return tp.addStatusToMetadata(metadata), nil
+}
+
+// QueryNewestMetadata 从参考索引向索引递增方向查询记录元数据（查询更新的记录，不读取完整数据）
+// refIndex: 参考索引位置
+// count: 查询数量
+// 返回的记录只包含元数据信息（索引、UUID、数据大小），按索引递增方向排序
+// 例如：QueryNewestMetadata(5, 3) 查询索引 6, 7, 8（不包含 5）
+func (tp *TopicProcessor) QueryNewestMetadata(refIndex, count int) ([]*RecordMetadataWithStatus, error) {
+	metadata, err := tp.query.QueryNewestMetadata(refIndex, count)
+	if err != nil {
+		return nil, err
+	}
+	return tp.addStatusToMetadata(metadata), nil
+}
+
+// QueryByIndex 根据索引查询单条记录的完整数据
+// index: 记录索引
+// 返回完整的记录数据，包含状态信息
+func (tp *TopicProcessor) QueryByIndex(index int) (*RecordWithStatus, error) {
+	record, err := tp.query.QueryByIndex(index)
+	if err != nil {
+		return nil, err
+	}
+
+	// 获取当前处理窗口位置
+	var startIdx, endIdx int
+	tp.mu.RLock()
+	if tp.tailer != nil {
+		startIdx = tp.tailer.GetStartIndex()
+		endIdx = tp.tailer.GetEndIndex()
+	}
+	tp.mu.RUnlock()
+
+	status := GetRecordStatus(index, startIdx, endIdx)
+	return &RecordWithStatus{
+		Record: record,
+		Index:  index,
+		Status: status,
+	}, nil
+}
+
 // GetRecordCount 获取记录总数（统一接口）
 func (tp *TopicProcessor) GetRecordCount() int {
 	return tp.index.Count()
@@ -429,10 +537,39 @@ func (tp *TopicProcessor) QueryFromLast(count int) ([]*RecordWithStatus, error)
 	// 获取记录总数
 	totalCount := tp.index.Count()

+	// 如果没有记录，返回空数组
+	if totalCount == 0 {
+		return []*RecordWithStatus{}, nil
+	}
+
 	// QueryOldest(totalCount, count) 会从最后一条记录开始向前查询
+	// totalCount 是记录总数，有效索引是 0 到 totalCount-1
+	// 所以传入 totalCount 作为 refIndex，会查询 totalCount-count 到 totalCount-1 的记录
 	return tp.QueryOldest(totalCount, count)
 }

+// QueryFromFirstMetadata 从第一条记录向索引递增方向查询元数据
+// count: 查询数量
+// 返回从第一条记录（索引 0）开始的记录元数据，包含状态信息
+// 例如：QueryFromFirstMetadata(3) 查询索引 0, 1, 2 的元数据
+func (tp *TopicProcessor) QueryFromFirstMetadata(count int) ([]*RecordMetadataWithStatus, error) {
+	// QueryNewestMetadata(-1, count) 会从索引 0 开始向后查询
+	return tp.QueryNewestMetadata(-1, count)
+}
+
+// QueryFromLastMetadata 从最后一条记录向索引递减方向查询元数据
+// count: 查询数量
+// 返回最后 N 条记录的元数据（按索引递增方向排序），包含状态信息
+// 例如：QueryFromLastMetadata(3) 查询最后 3 条记录的元数据
+func (tp *TopicProcessor) QueryFromLastMetadata(count int) ([]*RecordMetadataWithStatus, error) {
+	totalCount := tp.index.Count()
+	if totalCount == 0 {
+		return []*RecordMetadataWithStatus{}, nil
+	}
+	// QueryOldestMetadata(totalCount, count) 会从 totalCount 向前查询 count 条
+	return tp.QueryOldestMetadata(totalCount, count)
+}
+
 // GetProcessingIndex 获取当前处理索引（窗口开始索引）
 func (tp *TopicProcessor) GetProcessingIndex() int {
 	tp.mu.RLock()