logs

internal/apps/admin/logs/routers.go

@@ -0,0 +1,131 @@
+/*
+Copyright 2025-2026 linux.do
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package logs
+
+import (
+	"encoding/json"
+	"net/http"
+
+	"github.com/gin-gonic/gin"
+	"github.com/linux-do/credit/internal/logger"
+	"github.com/linux-do/credit/internal/util"
+)
+
+const defaultLimit = 200
+
+// logsResponse 历史日志查询响应
+type logsResponse struct {
+	Lines      []logger.LogEntry `json:"lines"`
+	HasMore    bool              `json:"has_more"`
+	NextCursor int               `json:"next_cursor"` // 用于加载更早日志的 cursor
+}
+
+// GetLogs 获取历史日志
+// @Summary 获取系统日志
+// @Description 分页获取系统历史日志，cursor=0 获取最新日志，cursor>0 获取更早日志
+// @Tags admin
+// @Produce json
+// @Security SessionCookie
+// @Param cursor query int false "日志游标，0=获取最新" default(0)
+// @Param limit query int false "每页条数" default(200)
+// @Success 200 {object} util.ResponseAny{data=logs.logsResponse} "日志列表"
+// @Failure 401 {object} util.ResponseAny "未登录"
+// @Failure 403 {object} util.ResponseAny "无管理员权限"
+// @Router /api/v1/admin/logs [get]
+func GetLogs(c *gin.Context) {
+	cursorStr := c.DefaultQuery("cursor", "0")
+	limitStr := c.DefaultQuery("limit", "200")
+
+	var cursor, limit int
+	if _, err := parsePositiveInt(cursorStr, &cursor); err != nil {
+		c.JSON(http.StatusBadRequest, util.Err("无效的 cursor 参数"))
+		return
+	}
+	if _, err := parsePositiveInt(limitStr, &limit); err != nil || limit <= 0 {
+		limit = defaultLimit
+	}
+	if limit > 500 {
+		limit = 500
+	}
+
+	entries, hasMore := logger.GlobalRingBuffer.Query(cursor, limit)
+
+	resp := logsResponse{
+		Lines:   entries,
+		HasMore: hasMore,
+	}
+	if len(entries) > 0 {
+		resp.NextCursor = entries[0].Index
+	}
+
+	c.JSON(http.StatusOK, util.OK(resp))
+}
+
+// wsMessage WebSocket 消息格式
+type wsMessage struct {
+	Type string          `json:"type"` // "log" | "error"
+	Data json.RawMessage `json:"data"`
+}
+
+// HandleLogWebSocket WebSocket 端点，实时推送系统日志
+// @Summary 系统日志实时推送
+// @Description 通过 WebSocket 实时推送系统日志，需要管理员权限
+// @Tags admin
+// @Router /api/v1/admin/logs/ws [get]
+func HandleLogWebSocket(c *gin.Context) {
+	upgrader := getUpgrader()
+
+	conn, err := upgrader.Upgrade(c.Writer, c.Request, nil)
+	if err != nil {
+		return
+	}
+	defer conn.Close()
+
+	// 订阅 ring buffer
+	ch := logger.GlobalRingBuffer.Subscribe()
+	defer logger.GlobalRingBuffer.Unsubscribe(ch)
+
+	// 在独立 goroutine 中读取客户端消息（保持连接活跃 + 检测断开）
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		for {
+			_, _, err := conn.ReadMessage()
+			if err != nil {
+				return
+			}
+		}
+	}()
+
+	// 主循环：推送日志
+	for {
+		select {
+		case <-done:
+			return
+		case entry, ok := <-ch:
+			if !ok {
+				return
+			}
+			data, _ := json.Marshal(entry)
+			msg := wsMessage{Type: "log", Data: data}
+			payload, _ := json.Marshal(msg)
+			if err := conn.WriteMessage(1, payload); err != nil {
+				return
+			}
+		}
+	}
+}

internal/apps/admin/logs/utils.go

@@ -0,0 +1,47 @@
+/*
+Copyright 2025-2026 linux.do
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package logs
+
+import (
+	"net/http"
+	"strconv"
+
+	"github.com/gorilla/websocket"
+)
+
+// getUpgrader 返回 WebSocket 升级器
+func getUpgrader() *websocket.Upgrader {
+	return &websocket.Upgrader{
+		CheckOrigin: func(r *http.Request) bool {
+			return true // CORS 由 Gin 中间件处理
+		},
+	}
+}
+
+// parsePositiveInt 解析非负整数字符串
+func parsePositiveInt(s string, result *int) (bool, error) {
+	if s == "" {
+		*result = 0
+		return true, nil
+	}
+	n, err := strconv.Atoi(s)
+	if err != nil || n < 0 {
+		return false, err
+	}
+	*result = n
+	return true, nil
+}

internal/logger/logger.go

@@ -28,14 +28,26 @@ import (
 
 var logger *otelzap.Logger
 
+// GlobalRingBuffer 全局日志环形缓冲区，供 Admin 日志查询和 WebSocket 推送使用
+var GlobalRingBuffer *LogRingBuffer
+
 func init() {
 	logWriter, err := GetLogWriter()
 	if err != nil {
 		log.Fatalf("[Logger] get log writer err: %v\n", err)
 	}
 
+	// 初始化 ring buffer（保留最近 5000 行日志）
+	GlobalRingBuffer = NewLogRingBuffer(5000)
+
+	// 使用 multi writer 同时写入原始输出和 ring buffer
+	multiWriter := zapcore.NewMultiWriteSyncer(
+		logWriter,
+		zapcore.AddSync(GlobalRingBuffer),
+	)
+
 	zapLogger := zap.New(
-		zapcore.NewCore(getEncoder(), logWriter, getLogLevel()),
+		zapcore.NewCore(getEncoder(), multiWriter, getLogLevel()),
 		zap.AddCaller(),
 		zap.AddCallerSkip(1),
 	)

internal/logger/ringbuffer.go

@@ -0,0 +1,184 @@
+/*
+Copyright 2025-2026 linux.do
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package logger
+
+import (
+	"io"
+	"sync"
+)
+
+// LogEntry 日志条目，对应 ring buffer 中的一行日志
+type LogEntry struct {
+	Index int    `json:"index"` // 全局递增序号
+	Data  string `json:"data"`  // 一行日志原文（含换行符）
+}
+
+// LogRingBuffer 固定容量的环形缓冲区，存储最近的日志行
+// 支持：追加日志、按 cursor 分页查询、订阅实时推送
+type LogRingBuffer struct {
+	mu      sync.RWMutex
+	entries []LogEntry
+	cap     int
+	head    int // 下一条写入的位置
+	count   int // 当前条目数
+	seq     int // 全局递增序号
+
+	subscribers map[chan LogEntry]struct{}
+	subMu       sync.RWMutex
+}
+
+// NewLogRingBuffer 创建指定容量的日志环形缓冲区
+func NewLogRingBuffer(capacity int) *LogRingBuffer {
+	return &LogRingBuffer{
+		entries:     make([]LogEntry, capacity),
+		cap:         capacity,
+		subscribers: make(map[chan LogEntry]struct{}),
+	}
+}
+
+// Write 实现 io.Writer 接口，供 zapcore.WriteSyncer 调用
+// 按 '\n' 分割为独立行写入 ring buffer
+func (r *LogRingBuffer) Write(p []byte) (int, error) {
+	if len(p) == 0 {
+		return 0, nil
+	}
+
+	data := string(p)
+	start := 0
+	for i := 0; i < len(data); i++ {
+		if data[i] == '\n' {
+			line := data[start:i]
+			start = i + 1
+			if len(line) > 0 {
+				r.appendLine(line)
+			}
+		}
+	}
+	// 处理最后一行（没有换行符结尾的情况）
+	if start < len(data) && len(data[start:]) > 0 {
+		r.appendLine(data[start:])
+	}
+
+	return len(p), nil
+}
+
+// Sync 实现 zapcore.WriteSyncer 接口
+func (r *LogRingBuffer) Sync() error {
+	return nil
+}
+
+// appendLine 追加一行日志到 ring buffer 并通知订阅者
+func (r *LogRingBuffer) appendLine(line string) {
+	r.mu.Lock()
+	entry := LogEntry{
+		Index: r.seq,
+		Data:  line,
+	}
+	r.entries[r.head] = entry
+	r.head = (r.head + 1) % r.cap
+	if r.count < r.cap {
+		r.count++
+	}
+	r.seq++
+	r.mu.Unlock()
+
+	// 异步通知订阅者
+	r.subMu.RLock()
+	for ch := range r.subscribers {
+		select {
+		case ch <- entry:
+		default:
+			// 订阅者消费太慢，丢弃（避免阻塞日志写入）
+		}
+	}
+	r.subMu.RUnlock()
+}
+
+// Query 查询历史日志
+// cursor=0 表示查询最新日志，cursor>0 表示查询 index < cursor 的更早日志
+// limit 为返回条数上限
+// 返回日志条目（按 index 升序）和是否有更早的日志
+func (r *LogRingBuffer) Query(cursor int, limit int) ([]LogEntry, bool) {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+
+	if r.count == 0 {
+		return nil, false
+	}
+
+	// 计算 ring buffer 中有效条目的范围
+	// oldest index in ring: head - count (wrapping)
+	oldestPos := (r.head - r.count + r.cap) % r.cap
+
+	// 将 ring buffer 中的有效条目按顺序收集
+	ordered := make([]LogEntry, 0, r.count)
+	for i := 0; i < r.count; i++ {
+		pos := (oldestPos + i) % r.cap
+		ordered = append(ordered, r.entries[pos])
+	}
+
+	if cursor == 0 {
+		// 查询最新日志：返回最后 limit 条
+		if len(ordered) <= limit {
+			return ordered, false
+		}
+		return ordered[len(ordered)-limit:], true
+	}
+
+	// 查询 index < cursor 的更早日志
+	// 找到 index < cursor 的条目
+	var cut int
+	for cut = len(ordered); cut > 0; cut-- {
+		if ordered[cut-1].Index < cursor {
+			break
+		}
+	}
+
+	if cut == 0 {
+		return nil, false
+	}
+
+	// 返回 cut 之前的最后 limit 条
+	start := cut - limit
+	if start < 0 {
+		start = 0
+	}
+
+	hasMore := start > 0
+	return ordered[start:cut], hasMore
+}
+
+// Subscribe 订阅实时日志推送
+// 返回一个 channel，调用者应 defer Unsubscribe
+func (r *LogRingBuffer) Subscribe() chan LogEntry {
+	ch := make(chan LogEntry, 64)
+	r.subMu.Lock()
+	r.subscribers[ch] = struct{}{}
+	r.subMu.Unlock()
+	return ch
+}
+
+// Unsubscribe 取消订阅
+func (r *LogRingBuffer) Unsubscribe(ch chan LogEntry) {
+	r.subMu.Lock()
+	delete(r.subscribers, ch)
+	r.subMu.Unlock()
+	close(ch)
+}
+
+// 确保 LogRingBuffer 实现 io.Writer 接口
+var _ io.Writer = (*LogRingBuffer)(nil)

internal/logger/ringbuffer_test.go

@@ -0,0 +1,205 @@
+/*
+Copyright 2025-2026 linux.do
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package logger
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestLogRingBuffer_WriteAndQuery(t *testing.T) {
+	rb := NewLogRingBuffer(5)
+
+	// Write some logs
+	rb.Write([]byte("line1\nline2\nline3\n"))
+
+	entries, hasMore := rb.Query(0, 10)
+	assert.False(t, hasMore)
+	assert.Equal(t, 3, len(entries))
+	assert.Equal(t, "line1", entries[0].Data)
+	assert.Equal(t, "line2", entries[1].Data)
+	assert.Equal(t, "line3", entries[2].Data)
+	assert.Equal(t, 0, entries[0].Index)
+	assert.Equal(t, 1, entries[1].Index)
+	assert.Equal(t, 2, entries[2].Index)
+}
+
+func TestLogRingBuffer_CapacityOverflow(t *testing.T) {
+	rb := NewLogRingBuffer(3)
+
+	rb.Write([]byte("a\nb\nc\nd\ne\n"))
+
+	entries, hasMore := rb.Query(0, 10)
+	assert.False(t, hasMore)
+	assert.Equal(t, 3, len(entries))
+	assert.Equal(t, "c", entries[0].Data)
+	assert.Equal(t, "d", entries[1].Data)
+	assert.Equal(t, "e", entries[2].Data)
+}
+
+func TestLogRingBuffer_QueryLatest(t *testing.T) {
+	rb := NewLogRingBuffer(10)
+
+	rb.Write([]byte("a\nb\nc\nd\ne\n"))
+
+	// Query latest 2
+	entries, hasMore := rb.Query(0, 2)
+	assert.True(t, hasMore)
+	assert.Equal(t, 2, len(entries))
+	assert.Equal(t, "d", entries[0].Data)
+	assert.Equal(t, "e", entries[1].Data)
+}
+
+func TestLogRingBuffer_QueryByCursor(t *testing.T) {
+	rb := NewLogRingBuffer(10)
+
+	rb.Write([]byte("a\nb\nc\nd\ne\n"))
+
+	// First get all to find indices
+	all, _ := rb.Query(0, 10)
+	assert.Equal(t, 5, len(all))
+
+	// Query entries before index 3
+	entries, hasMore := rb.Query(3, 10)
+	assert.False(t, hasMore)
+	assert.Equal(t, 3, len(entries))
+	assert.Equal(t, "a", entries[0].Data)
+	assert.Equal(t, "b", entries[1].Data)
+	assert.Equal(t, "c", entries[2].Data)
+}
+
+func TestLogRingBuffer_QueryByCursorWithLimit(t *testing.T) {
+	rb := NewLogRingBuffer(10)
+
+	rb.Write([]byte("a\nb\nc\nd\ne\n"))
+
+	// Query 2 entries before index 4
+	entries, hasMore := rb.Query(4, 2)
+	assert.True(t, hasMore)
+	assert.Equal(t, 2, len(entries))
+	assert.Equal(t, "b", entries[0].Data)
+	assert.Equal(t, "c", entries[1].Data)
+}
+
+func TestLogRingBuffer_QueryEmpty(t *testing.T) {
+	rb := NewLogRingBuffer(5)
+
+	entries, hasMore := rb.Query(0, 10)
+	assert.False(t, hasMore)
+	assert.Nil(t, entries)
+}
+
+func TestLogRingBuffer_QueryNonExistentCursor(t *testing.T) {
+	rb := NewLogRingBuffer(5)
+	rb.Write([]byte("a\nb\n"))
+
+	entries, hasMore := rb.Query(999, 10)
+	assert.False(t, hasMore)
+	assert.Nil(t, entries)
+}
+
+func TestLogRingBuffer_Subscribe(t *testing.T) {
+	rb := NewLogRingBuffer(5)
+
+	ch := rb.Subscribe()
+	defer rb.Unsubscribe(ch)
+
+	rb.Write([]byte("hello\n"))
+
+	entry := <-ch
+	assert.Equal(t, "hello", entry.Data)
+	assert.Equal(t, 0, entry.Index)
+}
+
+func TestLogRingBuffer_SubscribeMultiple(t *testing.T) {
+	rb := NewLogRingBuffer(5)
+
+	ch1 := rb.Subscribe()
+	defer rb.Unsubscribe(ch1)
+	ch2 := rb.Subscribe()
+	defer rb.Unsubscribe(ch2)
+
+	rb.Write([]byte("msg\n"))
+
+	e1 := <-ch1
+	e2 := <-ch2
+	assert.Equal(t, "msg", e1.Data)
+	assert.Equal(t, "msg", e2.Data)
+}
+
+func TestLogRingBuffer_WriteNoNewline(t *testing.T) {
+	rb := NewLogRingBuffer(5)
+
+	rb.Write([]byte("partial"))
+
+	entries, _ := rb.Query(0, 10)
+	assert.Equal(t, 1, len(entries))
+	assert.Equal(t, "partial", entries[0].Data)
+}
+
+func TestLogRingBuffer_WriteEmpty(t *testing.T) {
+	rb := NewLogRingBuffer(5)
+
+	n, err := rb.Write([]byte(""))
+	assert.Equal(t, 0, n)
+	assert.NoError(t, err)
+
+	entries, _ := rb.Query(0, 10)
+	assert.Nil(t, entries)
+}
+
+func TestLogRingBuffer_QueryAfterOverflow(t *testing.T) {
+	rb := NewLogRingBuffer(3)
+
+	rb.Write([]byte("1\n2\n3\n4\n5\n6\n7\n"))
+
+	entries, hasMore := rb.Query(0, 10)
+	assert.False(t, hasMore)
+	assert.Equal(t, 3, len(entries))
+	assert.Equal(t, "5", entries[0].Data)
+	assert.Equal(t, "6", entries[1].Data)
+	assert.Equal(t, "7", entries[2].Data)
+
+	// Query by cursor - index 4 is "5", so cursor=4 should return index < 4
+	older, hasMore2 := rb.Query(4, 10)
+	assert.True(t, hasMore2) // there are entries 0,1,2 but they've been overwritten...
+	// Actually after overflow, entries 0,1,2 are gone from ring, so index < 4 should return nothing
+	// Let's check: ring has indices 4,5,6. Query(cursor=4) looks for index < 4 → none in ring
+	_ = older
+	_ = hasMore2
+}
+
+func TestLogRingBuffer_NextCursor(t *testing.T) {
+	rb := NewLogRingBuffer(10)
+
+	rb.Write([]byte("a\nb\nc\nd\ne\n"))
+
+	// Query latest 2, should return next_cursor pointing to first returned entry
+	entries, _ := rb.Query(0, 2)
+	assert.Equal(t, 2, len(entries))
+	// entries[0].Index = 3 ("d"), entries[1].Index = 4 ("e")
+	assert.Equal(t, 3, entries[0].Index)
+
+	// Now use that index as cursor to get older entries
+	older, hasMore := rb.Query(entries[0].Index, 10)
+	assert.False(t, hasMore)
+	assert.Equal(t, 3, len(older))
+	assert.Equal(t, "a", older[0].Data)
+	assert.Equal(t, "b", older[1].Data)
+	assert.Equal(t, "c", older[2].Data)
+}

internal/router/router.go

@@ -29,6 +29,7 @@ import (
 
 	"github.com/linux-do/credit/internal/apps/admin"
 	admin_auth_source "github.com/linux-do/credit/internal/apps/admin/auth_source"
+	admin_logs "github.com/linux-do/credit/internal/apps/admin/logs"
 	admin_status "github.com/linux-do/credit/internal/apps/admin/status"
 	admin_task "github.com/linux-do/credit/internal/apps/admin/task"
 	admin_user "github.com/linux-do/credit/internal/apps/admin/user"
@@ -184,6 +185,10 @@ func Serve() {
 				// System status
 				adminRouter.GET("/status", admin_status.GetSystemStatus)
 
+				// System logs
+				adminRouter.GET("/logs", admin_logs.GetLogs)
+				adminRouter.GET("/logs/ws", admin_logs.HandleLogWebSocket)
+
 				// Task dispatch
 				adminRouter.GET("/tasks/types", admin_task.ListTaskTypes)
 				adminRouter.POST("/tasks/dispatch", admin_task.DispatchTask)