From 48b0b56ce7d500a8e4d8e347a1b47243cb6edb37 Mon Sep 17 00:00:00 2001
From: bourdon <hupeh@qq.com>
Date: Sat, 4 Oct 2025 01:43:48 +0800
Subject: [PATCH] =?UTF-8?q?=E5=8A=9F=E8=83=BD=EF=BC=9A=E5=AE=9E=E7=8E=B0?=
 =?UTF-8?q?=20webapp=20=E5=9F=BA=E4=BA=8E=E7=B4=A2=E5=BC=95=E7=9A=84?=
 =?UTF-8?q?=E6=97=A5=E5=BF=97=E5=AF=BC=E8=88=AA?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

主要改进：
- 添加索引导航支持：前进/后退按钮现在基于记录索引加载数据
- 后端 API 支持可选的 index 参数，返回包含 centerIndex 的响应
- 前端追踪 currentCenterIndex，实现精确的页面跳转
- 在状态徽章中显示记录索引号 [#索引]
- 修复日志显示逻辑：从追加模式改为完全重新渲染

代码优化：
- concurrent: 使用 Go 1.25 range 语法和 min 函数
- concurrent: 使用 WaitGroup.Go 方法简化 goroutine 启动
- topic_processor: 修正格式化输出

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
---
 example/concurrent/main.go      |  22 ++---
 example/topic_processor/main.go |   2 +-
 example/webapp/main.go          | 157 +++++++++++++++++++++-----------
 3 files changed, 115 insertions(+), 66 deletions(-)

diff --git a/example/concurrent/main.go b/example/concurrent/main.go
index 1a91e4d..4dc6906 100644
--- a/example/concurrent/main.go
+++ b/example/concurrent/main.go
@@ -114,12 +114,12 @@ func main() {
 	var totalQueries atomic.Int64
 	var queryErrors atomic.Int64
 
-	for i := 0; i < queryCount; i++ {
+	for i := range queryCount {
 		queryWg.Add(1)
 		go func(queryID int) {
 			defer queryWg.Done()
 
-			for j := 0; j < queriesPerGoroutine; j++ {
+			for j := range queriesPerGoroutine {
 				// 随机选择一个 topic 进行查询
 				topic := topics[j%len(topics)]
 
@@ -136,10 +136,7 @@ func main() {
 				}
 
 				// 查询最新的 10 条记录
-				querySize := 10
-				if count < querySize {
-					querySize = count
-				}
+				querySize := min(count, 10)
 
 				_, err = processor.QueryNewest(count-1, querySize)
 				if err != nil {
@@ -180,7 +177,7 @@ func main() {
 		go func(topicName string) {
 			defer mixWg.Done()
 
-			for j := 0; j < 1000; j++ {
+			for j := range 1000 {
 				data := fmt.Sprintf("mix-%s-msg-%d", topicName, j)
 
 				if _, err := seq.Write(topicName, []byte(data)); err == nil {
@@ -194,12 +191,12 @@ func main() {
 	}
 
 	// 启动查询 goroutine
-	for i := 0; i < 10; i++ {
+	for i := range 10 {
 		mixWg.Add(1)
 		go func(queryID int) {
 			defer mixWg.Done()
 
-			for j := 0; j < 200; j++ {
+			for j := range 200 {
 				topic := topics[j%len(topics)]
 
 				processor, err := seq.GetProcessor(topic)
@@ -267,7 +264,7 @@ func main() {
 	}
 
 	// 持续查询 goroutine
-	for i := 0; i < 5; i++ {
+	for i := range 5 {
 		stressWg.Add(1)
 		go func(queryID int) {
 			defer stressWg.Done()
@@ -296,8 +293,7 @@ func main() {
 	}
 
 	// 进度显示 goroutine
-	stressWg.Add(1)
-	go func() {
+	stressWg.Go(func() {
 		defer stressWg.Done()
 		ticker := time.NewTicker(10 * time.Second)
 		defer ticker.Stop()
@@ -314,7 +310,7 @@ func main() {
 					stressQueryCount.Load())
 			}
 		}
-	}()
+	})
 
 	stressWg.Wait()
 	stressDuration := time.Since(stressTestStart)
diff --git a/example/topic_processor/main.go b/example/topic_processor/main.go
index 253faa1..6671ce1 100644
--- a/example/topic_processor/main.go
+++ b/example/topic_processor/main.go
@@ -10,7 +10,7 @@ import (
 
 func main() {
 	// ===== TopicProcessor 作为聚合器使用 =====
-	fmt.Println("=== TopicProcessor 聚合器示例 ===\n")
+	fmt.Println("=== TopicProcessor 聚合器示例 ===")
 
 	// 创建 TopicProcessor（提供空 handler）
 	logger := slog.Default()
diff --git a/example/webapp/main.go b/example/webapp/main.go
index d9825ae..ffb9830 100644
--- a/example/webapp/main.go
+++ b/example/webapp/main.go
@@ -347,7 +347,7 @@ func handleIndex(w http.ResponseWriter, r *http.Request) {
 
     <script>
         let currentTopic = null;
-        let displayedOffsets = new Set(); // 追踪已显示的日志偏移量
+        let currentCenterIndex = null; // 追踪当前中心索引位置
 
         // 加载 topics
         async function loadTopics() {
@@ -363,7 +363,7 @@ func handleIndex(w http.ResponseWriter, r *http.Request) {
         // 选择 topic
         function selectTopic(topic) {
             currentTopic = topic;
-            displayedOffsets.clear(); // 切换 topic 时清空已显示记录
+            currentCenterIndex = null; // 切换 topic 时重置索引
 
             // 更新选中状态
             document.querySelectorAll('.topic-item').forEach(item => {
@@ -393,66 +393,86 @@ func handleIndex(w http.ResponseWriter, r *http.Request) {
         }
 
         // 加载日志
-        async function loadLogs() {
+        async function loadLogs(index) {
             if (!currentTopic) return;
 
             const backward = document.getElementById('backwardCount').value;
             const forward = document.getElementById('forwardCount').value;
 
-            const response = await fetch('/api/query?topic=' + currentTopic +
-                '&backward=' + backward + '&forward=' + forward);
+            // 构建查询 URL
+            let url = '/api/query?topic=' + currentTopic +
+                '&backward=' + backward + '&forward=' + forward;
+
+            // 如果指定了索引，添加到查询参数
+            if (index !== undefined && index !== null) {
+                url += '&index=' + index;
+            }
+
+            const response = await fetch(url);
             const data = await response.json();
 
+            // 更新当前中心索引
+            currentCenterIndex = data.centerIndex;
+
             const container = document.getElementById('logContainer');
 
-            if (data.records.length === 0 && displayedOffsets.size === 0) {
+            if (data.records.length === 0) {
                 container.innerHTML = '<div class="loading">暂无日志</div>';
                 return;
             }
 
-            // 过滤出新记录
-            const newRecords = data.records.filter(r => !displayedOffsets.has(r.offset));
+            // 清空并重新渲染所有记录
+            const html = data.records.map(r => {
+                // 解析状态
+                let statusClass = 'pending';
+                let statusText = '待处理';
+                let badgeClass = 'status-pending';
 
-            if (newRecords.length > 0) {
-                // 生成新记录的 HTML
-                const newHTML = newRecords.map(r => {
-                    displayedOffsets.add(r.offset); // 标记为已显示
+                if (r.status === 'StatusProcessed' || r.status === 'processed') {
+                    statusClass = 'processed';
+                    statusText = '已处理';
+                    badgeClass = 'status-processed';
+                } else if (r.status === 'StatusProcessing' || r.status === 'processing') {
+                    statusClass = 'processing';
+                    statusText = '处理中';
+                    badgeClass = 'status-processing';
+                }
 
-                    // 解析状态，处理可能的状态值
-                    let statusClass = 'pending';
-                    let statusText = '待处理';
-                    let badgeClass = 'status-pending';
+                return '<div class="log-entry ' + statusClass + '" data-index="' + r.index + '">' +
+                       '<span class="status-badge ' + badgeClass + '">[#' + r.index + '] ' + statusText + '</span>' +
+                       r.data +
+                       '</div>';
+            }).join('');
 
-                    if (r.status === 'StatusProcessed' || r.status === 'processed') {
-                        statusClass = 'processed';
-                        statusText = '已处理';
-                        badgeClass = 'status-processed';
-                    } else if (r.status === 'StatusProcessing' || r.status === 'processing') {
-                        statusClass = 'processing';
-                        statusText = '处理中';
-                        badgeClass = 'status-processing';
-                    }
-
-                    return '<div class="log-entry ' + statusClass + '" data-offset="' + r.offset + '">' +
-                           '<span class="status-badge ' + badgeClass + '">' + statusText + '</span>' +
-                           r.data +
-                           '</div>';
-                }).join('');
-
-                // 追加新记录
-                container.innerHTML += newHTML;
-
-                // 自动滚动到底部
-                container.scrollTop = container.scrollHeight;
-            }
+            container.innerHTML = html;
         }
 
         function queryBackward() {
-            loadLogs();
+            if (currentCenterIndex === null || currentCenterIndex === undefined) {
+                loadLogs();
+                return;
+            }
+
+            const backward = parseInt(document.getElementById('backwardCount').value);
+            const forward = parseInt(document.getElementById('forwardCount').value);
+
+            // 向前翻页：中心索引向前移动（backward + forward + 1）个位置
+            const newIndex = currentCenterIndex - (backward + forward + 1);
+            loadLogs(newIndex >= 0 ? newIndex : 0);
         }
 
         function queryForward() {
-            loadLogs();
+            if (currentCenterIndex === null || currentCenterIndex === undefined) {
+                loadLogs();
+                return;
+            }
+
+            const backward = parseInt(document.getElementById('backwardCount').value);
+            const forward = parseInt(document.getElementById('forwardCount').value);
+
+            // 向后翻页：中心索引向后移动（backward + forward + 1）个位置
+            const newIndex = currentCenterIndex + (backward + forward + 1);
+            loadLogs(newIndex);
         }
 
         function formatBytes(bytes) {
@@ -509,6 +529,8 @@ func handleQuery(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	// 获取查询参数
+	indexParam := r.URL.Query().Get("index")
 	backward, _ := strconv.Atoi(r.URL.Query().Get("backward"))
 	forward, _ := strconv.Atoi(r.URL.Query().Get("forward"))
 
@@ -526,53 +548,84 @@ func handleQuery(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	// 获取当前处理索引和读取索引
+	// 获取记录总数
+	totalCount := processor.GetRecordCount()
+
+	// 确定查询中心索引
+	var centerIndex int
+	if indexParam == "" {
+		// 如果没有指定索引，使用当前处理位置
+		centerIndex = seq.GetProcessingIndex(topic)
+	} else {
+		centerIndex, _ = strconv.Atoi(indexParam)
+		// 限制索引范围
+		if centerIndex < 0 {
+			centerIndex = 0
+		}
+		if centerIndex >= totalCount {
+			centerIndex = totalCount - 1
+		}
+	}
+
+	// 获取当前处理索引和读取索引（用于状态判断）
 	startIdx := seq.GetProcessingIndex(topic)
 	endIdx := seq.GetReadIndex(topic)
 
 	// 合并查询结果：向后 + 当前 + 向前
 	var results []*seqlog.RecordWithStatus
 
-	// 向后查询
-	if backward > 0 && startIdx > 0 {
-		backResults, err := processor.QueryNewest(startIdx-1, backward)
+	// 向后查询（查询更早的记录）
+	if backward > 0 && centerIndex > 0 {
+		backResults, err := processor.QueryNewest(centerIndex-1, backward)
 		if err == nil {
 			results = append(results, backResults...)
 		}
 	}
 
-	// 当前位置
-	if startIdx < endIdx {
-		currentResults, err := processor.QueryOldest(startIdx, 1)
+	// 当前位置的记录
+	if centerIndex >= 0 && centerIndex < totalCount {
+		currentResults, err := processor.QueryOldest(centerIndex, 1)
 		if err == nil {
 			results = append(results, currentResults...)
 		}
 	}
 
-	// 向前查询
-	if forward > 0 {
-		forwardResults, err := processor.QueryOldest(endIdx, forward)
+	// 向前查询（查询更新的记录）
+	if forward > 0 && centerIndex+1 < totalCount {
+		forwardResults, err := processor.QueryOldest(centerIndex+1, forward)
 		if err == nil {
 			results = append(results, forwardResults...)
 		}
 	}
 
 	type Record struct {
+		Index  int    `json:"index"`
 		Status string `json:"status"`
 		Data   string `json:"data"`
 	}
 
+	// 计算每条记录的实际索引位置
+	startRecordIndex := centerIndex - backward
+	if startRecordIndex < 0 {
+		startRecordIndex = 0
+	}
+
 	records := make([]Record, len(results))
 	for i, r := range results {
 		records[i] = Record{
+			Index:  startRecordIndex + i,
 			Status: r.Status.String(),
 			Data:   string(r.Record.Data),
 		}
 	}
 
 	json.NewEncoder(w).Encode(map[string]interface{}{
-		"records": records,
-		"total":   len(records),
+		"records":      records,
+		"total":        len(records),
+		"centerIndex":  centerIndex,
+		"totalCount":   totalCount,
+		"processingIndex": startIdx,
+		"readIndex":    endIdx,
 	})
 }