fanqie-go/internal/fqapi/request_log.go

package fqapi

import (
	"fanqie/internal/config"
	"fmt"
	"time"
)

// 请求日志：只记元数据（时间/IP/方法/路径/状态/耗时）与错误响应摘要，不存响应体。
// 全量响应与正文缓存重复且体积不可控；要看重放一个请求即可。

// RequestLogEntry 一条请求日志
type RequestLogEntry struct {
	Ts         int64  `json:"ts"`
	IP         string `json:"ip"`
	Method     string `json:"method"`
	Path       string `json:"path"`
	Query      string `json:"query"` // 已脱敏（key 参数替换为 ***）
	Status     int    `json:"status"`
	DurationMs int64  `json:"duration_ms"`
	ErrorMsg   string `json:"error_msg"` // 仅 status>=400 时截断保留
}

// ErrMsgMax 错误响应摘要的最大保留字节数
const ErrMsgMax = 4096

const (
	reqLogChanSize   = 256          // channel 满则丢弃，日志绝不阻塞请求
	reqLogFlushBatch = 50           // 攒批上限
	reqLogFlushEvery = time.Second  // 兜底 flush 间隔
	reqLogCleanEvery = time.Hour    // 过期清理间隔
)

var (
	reqLogCh   chan RequestLogEntry
	reqLogQuit chan struct{}
	reqLogDone chan struct{}
)

// InitRequestLog 建表并启动后台落库协程（未启用时为空操作）
func InitRequestLog() error {
	if !requestLogEnabled() {
		return nil
	}

	ddl := `CREATE TABLE IF NOT EXISTS request_log (
		id INTEGER PRIMARY KEY AUTOINCREMENT,
		ts INTEGER NOT NULL,
		ip TEXT,
		method TEXT,
		path TEXT,
		query TEXT,
		status INTEGER,
		duration_ms INTEGER,
		error_msg TEXT
	)`
	if dbDriver == "pgx" {
		ddl = `CREATE TABLE IF NOT EXISTS request_log (
		id BIGSERIAL PRIMARY KEY,
		ts BIGINT NOT NULL,
		ip TEXT,
		method TEXT,
		path TEXT,
		query TEXT,
		status INTEGER,
		duration_ms BIGINT,
		error_msg TEXT
	)`
	}
	if _, err := cacheDB.Exec(ddl); err != nil {
		return fmt.Errorf("创建request_log表失败: %w", err)
	}
	if _, err := cacheDB.Exec(`CREATE INDEX IF NOT EXISTS idx_request_log_ts ON request_log(ts)`); err != nil {
		cacheLogger.Warning("创建request_log索引失败: " + err.Error())
	}

	reqLogCh = make(chan RequestLogEntry, reqLogChanSize)
	reqLogQuit = make(chan struct{})
	reqLogDone = make(chan struct{})
	go reqLogWorker()
	cacheLogger.Info("请求日志已启用")
	return nil
}

// CloseRequestLog 停止后台协程并落盘剩余日志
func CloseRequestLog() {
	if reqLogQuit == nil {
		return
	}
	close(reqLogQuit)
	<-reqLogDone
	reqLogQuit = nil
}

// LogRequest 非阻塞投递一条请求日志（未启用/队列满时静默丢弃）
func LogRequest(e RequestLogEntry) {
	if reqLogCh == nil {
		return
	}
	select {
	case reqLogCh <- e:
	default:
	}
}

// GetRecentRequests 查询最近 limit 条请求日志
func GetRecentRequests(limit int) []RequestLogEntry {
	if cacheDB == nil || reqLogCh == nil {
		return nil
	}
	rows, err := cacheDB.Query(
		q("SELECT ts, ip, method, path, query, status, duration_ms, COALESCE(error_msg, '') FROM request_log ORDER BY id DESC LIMIT ?"), limit)
	if err != nil {
		cacheLogger.Warning("查询请求日志失败: " + err.Error())
		return nil
	}
	defer rows.Close()

	entries := make([]RequestLogEntry, 0, limit)
	for rows.Next() {
		var e RequestLogEntry
		if err := rows.Scan(&e.Ts, &e.IP, &e.Method, &e.Path, &e.Query, &e.Status, &e.DurationMs, &e.ErrorMsg); err != nil {
			cacheLogger.Warning("读取请求日志失败: " + err.Error())
			break
		}
		entries = append(entries, e)
	}
	return entries
}

// reqLogWorker 攒批落库 + 定期清理过期日志
func reqLogWorker() {
	defer close(reqLogDone)

	batch := make([]RequestLogEntry, 0, reqLogFlushBatch)
	flush := func() {
		if len(batch) == 0 {
			return
		}
		if err := insertRequestLogs(batch); err != nil {
			cacheLogger.Warning("写入请求日志失败: " + err.Error())
		}
		batch = batch[:0]
	}
	defer flush()

	ticker := time.NewTicker(reqLogFlushEvery)
	cleanTicker := time.NewTicker(reqLogCleanEvery)
	defer ticker.Stop()
	defer cleanTicker.Stop()

	for {
		select {
		case e := <-reqLogCh:
			batch = append(batch, e)
			if len(batch) >= reqLogFlushBatch {
				flush()
			}
		case <-ticker.C:
			flush()
		case <-cleanTicker.C:
			cleanupRequestLog()
		case <-reqLogQuit:
			// 排空 channel 后退出
			for {
				select {
				case e := <-reqLogCh:
					batch = append(batch, e)
				default:
					return
				}
			}
		}
	}
}

// insertRequestLogs 事务批量写入
func insertRequestLogs(batch []RequestLogEntry) error {
	tx, err := cacheDB.Begin()
	if err != nil {
		return err
	}
	stmt, err := tx.Prepare(q("INSERT INTO request_log (ts, ip, method, path, query, status, duration_ms, error_msg) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"))
	if err != nil {
		tx.Rollback()
		return err
	}
	for _, e := range batch {
		if _, err := stmt.Exec(e.Ts, e.IP, e.Method, e.Path, e.Query, e.Status, e.DurationMs, e.ErrorMsg); err != nil {
			tx.Rollback()
			return err
		}
	}
	return tx.Commit()
}

// cleanupRequestLog 删除超过保留期的日志
func cleanupRequestLog() {
	if !requestLogEnabled() {
		return
	}
	days := 7
	if cfg := config.GetAppConfig(); cfg != nil && cfg.Server.RequestLog.RetentionDays > 0 {
		days = cfg.Server.RequestLog.RetentionDays
	}
	cutoff := time.Now().Unix() - int64(days)*86400
	if result, err := cacheDB.Exec(q("DELETE FROM request_log WHERE ts < ?"), cutoff); err != nil {
		cacheLogger.Warning("清理请求日志失败: " + err.Error())
	} else if n, _ := result.RowsAffected(); n > 0 {
		cacheLogger.Info(fmt.Sprintf("已清理 %d 条过期请求日志（保留 %d 天）", n, days))
	}
}

// requestLogEnabled 配置启用且数据库可用
func requestLogEnabled() bool {
	cfg := config.GetAppConfig()
	return cfg != nil && cfg.Server.RequestLog.Enabled && cacheDB != nil
}