✨ feat(cache): 新增 Redis 缓存与 PostgreSQL 后端支持
- 新增 cache.redis 配置(独立 host/port/password/db 字段),启用后正文缓存与音频代理缓存走 Redis,键前缀 fq:cache:/fq:audio:,TTL 自动过期 - 新增 cache.backend: sqlite|postgres 切换,PG 连接用 net/url 构造 DSN,密码特殊字符自动转义;SQLite 顺手启用 WAL 与 busy_timeout - SQL 方言收敛:q() 将 ? 占位符转换为 $n、INSERT OR IGNORE 与查询再插入统一为 ON CONFLICT、LastInsertId 改为 RETURNING id、建表 DDL 双版本分派 - PG 下 VACUUM 走 autovacuum 说明、状态页大小改查 pg_database_size - 修复 PG 下设备保存失败:is_valid 列为 INTEGER,写入点补 boolToInt 转换 - 新增 .env 支持:敏感配置(密码/主机/访问密钥)由环境变量覆盖 config.yaml,.env 进 gitignore,仓库附 .env.example 模板 - 删除死代码 InitDevicePoolTable;新增 rebind_test.go 单测 - 更新 README 与 config.yaml 配置示例,go.mod 新增 go-redis/v9 与 pgx/v5 依赖
This commit is contained in:
1 parent
d710d382ee
commit
cf16c37f32
14 files changed
+671
-210
No files matched your search
@@ -20,6 +20,16 @@ func GenerateAudioProxyURL(originalURL, deviceID string, headers map[string]stri
|
||||
// 生成缓存key
|
||||
key := generateRandomKey(16)
|
||||
|
||||
// Redis 模式:TTL 自动过期
|
||||
if redisCache != nil {
|
||||
saveAudioProxyEntry(key, &audioProxyEntry{
|
||||
OriginalURL: originalURL,
|
||||
DeviceID: deviceID,
|
||||
Headers: headers,
|
||||
})
|
||||
return fmt.Sprintf("/audio/proxy/%s", key)
|
||||
}
|
||||
|
||||
// 将headers序列化为JSON字符串
|
||||
headersJSON, _ := json.Marshal(headers)
|
||||
|
||||
@@ -28,10 +38,10 @@ func GenerateAudioProxyURL(originalURL, deviceID string, headers map[string]stri
|
||||
cacheMutex.Lock()
|
||||
defer cacheMutex.Unlock()
|
||||
|
||||
_, err := cacheDB.Exec(`
|
||||
_, err := cacheDB.Exec(q(`
|
||||
INSERT INTO audio_cache (key, original_url, device_id, headers, created_at)
|
||||
VALUES (?, ?, ?, ?, ?)
|
||||
`, key, originalURL, deviceID, string(headersJSON), time.Now().Unix())
|
||||
`), key, originalURL, deviceID, string(headersJSON), time.Now().Unix())
|
||||
if err != nil {
|
||||
audioLogger.Error("保存音频缓存失败: " + err.Error())
|
||||
}
|
||||
@@ -43,40 +53,50 @@ func GenerateAudioProxyURL(originalURL, deviceID string, headers map[string]stri
|
||||
|
||||
// HandleAudioProxy 处理音频代理请求
|
||||
func HandleAudioProxy(key string) ([]byte, string, error) {
|
||||
if cacheDB == nil {
|
||||
return nil, "", fmt.Errorf("数据库未初始化")
|
||||
}
|
||||
|
||||
// 从数据库获取音频信息
|
||||
var originalURL, deviceID, headersJSON string
|
||||
var createdAt int64
|
||||
|
||||
cacheMutex.Lock()
|
||||
err := cacheDB.QueryRow(`
|
||||
SELECT original_url, device_id, headers, created_at
|
||||
FROM audio_cache WHERE key = ?
|
||||
`, key).Scan(&originalURL, &deviceID, &headersJSON, &createdAt)
|
||||
cacheMutex.Unlock()
|
||||
|
||||
if err == sql.ErrNoRows {
|
||||
return nil, "", fmt.Errorf("音频信息不存在,请先调用 /audio 接口")
|
||||
}
|
||||
if err != nil {
|
||||
return nil, "", fmt.Errorf("查询音频缓存失败: %w", err)
|
||||
}
|
||||
|
||||
// 检查缓存是否过期(2小时,匹配CDN链接有效期)
|
||||
if time.Now().Unix()-createdAt > 7200 { // 2小时 = 7200秒
|
||||
// 清理过期缓存
|
||||
cacheMutex.Lock()
|
||||
cacheDB.Exec("DELETE FROM audio_cache WHERE key = ?", key)
|
||||
cacheMutex.Unlock()
|
||||
return nil, "", fmt.Errorf("音频信息已过期,请重新调用 /audio 接口")
|
||||
}
|
||||
|
||||
// 解析headers
|
||||
var originalURL, deviceID string
|
||||
var headers map[string]string
|
||||
json.Unmarshal([]byte(headersJSON), &headers)
|
||||
|
||||
if redisCache != nil {
|
||||
entry, err := loadAudioProxyEntry(key)
|
||||
if err != nil {
|
||||
return nil, "", err
|
||||
}
|
||||
originalURL, deviceID, headers = entry.OriginalURL, entry.DeviceID, entry.Headers
|
||||
} else {
|
||||
if cacheDB == nil {
|
||||
return nil, "", fmt.Errorf("数据库未初始化")
|
||||
}
|
||||
|
||||
// 从数据库获取音频信息
|
||||
var headersJSON string
|
||||
var createdAt int64
|
||||
|
||||
cacheMutex.Lock()
|
||||
err := cacheDB.QueryRow(q(`
|
||||
SELECT original_url, device_id, headers, created_at
|
||||
FROM audio_cache WHERE key = ?
|
||||
`), key).Scan(&originalURL, &deviceID, &headersJSON, &createdAt)
|
||||
cacheMutex.Unlock()
|
||||
|
||||
if err == sql.ErrNoRows {
|
||||
return nil, "", fmt.Errorf("音频信息不存在,请先调用 /audio 接口")
|
||||
}
|
||||
if err != nil {
|
||||
return nil, "", fmt.Errorf("查询音频缓存失败: %w", err)
|
||||
}
|
||||
|
||||
// 检查缓存是否过期(2小时,匹配CDN链接有效期)
|
||||
if time.Now().Unix()-createdAt > 7200 { // 2小时 = 7200秒
|
||||
// 清理过期缓存
|
||||
cacheMutex.Lock()
|
||||
cacheDB.Exec(q("DELETE FROM audio_cache WHERE key = ?"), key)
|
||||
cacheMutex.Unlock()
|
||||
return nil, "", fmt.Errorf("音频信息已过期,请重新调用 /audio 接口")
|
||||
}
|
||||
|
||||
// 解析headers
|
||||
json.Unmarshal([]byte(headersJSON), &headers)
|
||||
}
|
||||
if headers == nil {
|
||||
headers = make(map[string]string)
|
||||
}
|
||||
@@ -92,6 +112,11 @@ func HandleAudioProxy(key string) ([]byte, string, error) {
|
||||
|
||||
// CleanExpiredAudioCache 清理过期的音频缓存
|
||||
func CleanExpiredAudioCache() {
|
||||
// Redis 模式下由 TTL 自动过期,无需清理
|
||||
if redisCache != nil {
|
||||
return
|
||||
}
|
||||
|
||||
if cacheDB == nil {
|
||||
return
|
||||
}
|
||||
@@ -100,7 +125,7 @@ func CleanExpiredAudioCache() {
|
||||
defer cacheMutex.Unlock()
|
||||
|
||||
// 删除2小时前的缓存
|
||||
result, err := cacheDB.Exec("DELETE FROM audio_cache WHERE created_at < ?", time.Now().Unix()-7200)
|
||||
result, err := cacheDB.Exec(q("DELETE FROM audio_cache WHERE created_at < ?"), time.Now().Unix()-7200)
|
||||
if err != nil {
|
||||
audioLogger.Error("清理过期音频缓存失败: " + err.Error())
|
||||
return
|
||||
|
||||
+234
-92
@@ -1,19 +1,27 @@
|
||||
package fqapi
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"fanqie/internal/common"
|
||||
"fanqie/internal/config"
|
||||
"fmt"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
_ "github.com/jackc/pgx/v5/stdlib"
|
||||
_ "github.com/mattn/go-sqlite3"
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
var (
|
||||
cacheDB *sql.DB
|
||||
dbDriver string // "sqlite3" | "pgx"
|
||||
redisCache *redis.Client
|
||||
cacheMutex sync.Mutex
|
||||
cacheLogger = common.NewLogger("cache_manager", "INFO", "")
|
||||
contentRequestCount int // content接口访问计数
|
||||
@@ -21,109 +29,222 @@ var (
|
||||
cacheCleanupTicker *time.Ticker // 缓存清理定时器
|
||||
)
|
||||
|
||||
// q 按数据库方言转换占位符(PostgreSQL 使用 $n 形式)
|
||||
func q(query string) string {
|
||||
if dbDriver != "pgx" {
|
||||
return query
|
||||
}
|
||||
var b strings.Builder
|
||||
n := 0
|
||||
for i := 0; i < len(query); i++ {
|
||||
if query[i] == '?' {
|
||||
n++
|
||||
b.WriteString("$" + strconv.Itoa(n))
|
||||
} else {
|
||||
b.WriteByte(query[i])
|
||||
}
|
||||
}
|
||||
return b.String()
|
||||
}
|
||||
|
||||
// InitRedis 初始化 Redis 缓存(启用后正文缓存与音频代理缓存走 Redis)
|
||||
func InitRedis() error {
|
||||
cfg := config.GetConfig()
|
||||
if cfg == nil || !cfg.Cache.Redis.Enabled {
|
||||
return nil
|
||||
}
|
||||
|
||||
redisCache = redis.NewClient(&redis.Options{
|
||||
Addr: fmt.Sprintf("%s:%d", cfg.Cache.Redis.Host, cfg.Cache.Redis.Port),
|
||||
Password: cfg.Cache.Redis.Password,
|
||||
DB: cfg.Cache.Redis.DB,
|
||||
})
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
if err := redisCache.Ping(ctx).Err(); err != nil {
|
||||
redisCache = nil
|
||||
return fmt.Errorf("连接 Redis 失败: %w", err)
|
||||
}
|
||||
|
||||
cacheLogger.Info(fmt.Sprintf("Redis 缓存已启用: %s:%d db=%d", cfg.Cache.Redis.Host, cfg.Cache.Redis.Port, cfg.Cache.Redis.DB))
|
||||
return nil
|
||||
}
|
||||
|
||||
// CloseRedis 关闭 Redis 连接
|
||||
func CloseRedis() {
|
||||
if redisCache != nil {
|
||||
redisCache.Close()
|
||||
}
|
||||
}
|
||||
|
||||
// CacheData 缓存数据结构
|
||||
type CacheData struct {
|
||||
Data string `json:"data"`
|
||||
Expi float64 `json:"expi"`
|
||||
}
|
||||
|
||||
// InitDB 初始化数据库
|
||||
// InitDB 初始化数据库(按 cache.backend 选择 sqlite 或 postgres)
|
||||
func InitDB() error {
|
||||
if cacheDB != nil {
|
||||
return nil // 已初始化
|
||||
}
|
||||
|
||||
// 使用配置的路径
|
||||
dbPath := "./cache.db"
|
||||
cfg := config.GetConfig()
|
||||
if cfg != nil && cfg.Cache.SQLitePath != "" {
|
||||
dbPath = cfg.Cache.SQLitePath
|
||||
backend := "sqlite"
|
||||
dsn := "./cache.db"
|
||||
if cfg != nil {
|
||||
if cfg.Cache.Backend != "" {
|
||||
backend = cfg.Cache.Backend
|
||||
}
|
||||
if backend == "postgres" {
|
||||
pg := cfg.Cache.Postgres
|
||||
if pg.Host == "" || pg.Database == "" {
|
||||
return fmt.Errorf("cache.backend 为 postgres 但未配置 host/database")
|
||||
}
|
||||
// 用 net/url 构造连接串,密码中的特殊字符会被自动转义
|
||||
u := &url.URL{
|
||||
Scheme: "postgres",
|
||||
User: url.UserPassword(pg.User, pg.Password),
|
||||
Host: fmt.Sprintf("%s:%d", pg.Host, pg.Port),
|
||||
Path: pg.Database,
|
||||
RawQuery: "sslmode=" + pg.SSLMode,
|
||||
}
|
||||
dsn = u.String()
|
||||
} else if cfg.Cache.SQLitePath != "" {
|
||||
dsn = cfg.Cache.SQLitePath
|
||||
}
|
||||
}
|
||||
|
||||
switch backend {
|
||||
case "postgres":
|
||||
dbDriver = "pgx"
|
||||
default:
|
||||
dbDriver = "sqlite3"
|
||||
// ponytail: 单文件场景,WAL + 忙等待即可,无需更多调优
|
||||
dsn = dsn + "?_journal_mode=WAL&_busy_timeout=5000"
|
||||
}
|
||||
|
||||
var err error
|
||||
cacheDB, err = sql.Open("sqlite3", dbPath)
|
||||
cacheDB, err = sql.Open(dbDriver, dsn)
|
||||
if err != nil {
|
||||
cacheLogger.Error("打开数据库失败: " + err.Error())
|
||||
return err
|
||||
}
|
||||
|
||||
// 创建缓存表(添加created_at字段用于清理)
|
||||
_, err = cacheDB.Exec(`CREATE TABLE IF NOT EXISTS cache (
|
||||
key TEXT PRIMARY KEY,
|
||||
value TEXT,
|
||||
created_at INTEGER DEFAULT (strftime('%s', 'now'))
|
||||
)`)
|
||||
if err != nil {
|
||||
cacheLogger.Error("创建缓存表失败: " + err.Error())
|
||||
if err := cacheDB.Ping(); err != nil {
|
||||
cacheLogger.Error("连接数据库失败: " + err.Error())
|
||||
return err
|
||||
}
|
||||
cacheLogger.Info(fmt.Sprintf("数据库已连接: %s (driver=%s)", backend, dbDriver))
|
||||
|
||||
|
||||
// 创建statistics表(保存请求统计,始终使用)
|
||||
_, err = cacheDB.Exec(`CREATE TABLE IF NOT EXISTS statistics (
|
||||
id INTEGER PRIMARY KEY CHECK (id = 1),
|
||||
total_requests INTEGER DEFAULT 0,
|
||||
search_requests INTEGER DEFAULT 0,
|
||||
detail_requests INTEGER DEFAULT 0,
|
||||
catalog_requests INTEGER DEFAULT 0,
|
||||
content_requests INTEGER DEFAULT 0,
|
||||
manga_requests INTEGER DEFAULT 0,
|
||||
audio_requests INTEGER DEFAULT 0,
|
||||
video_requests INTEGER DEFAULT 0,
|
||||
category_requests INTEGER DEFAULT 0,
|
||||
updated_at INTEGER DEFAULT (strftime('%s', 'now'))
|
||||
)`)
|
||||
if err != nil {
|
||||
cacheLogger.Error("创建statistics表失败: " + err.Error())
|
||||
if err := createTables(); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// createTables 创建数据表(两套 DDL 各自兼容对应方言)
|
||||
func createTables() error {
|
||||
// 通用部分:statistics 初始化、索引、audio_cache 建表在两方言间一致或按 q() 转换
|
||||
tables := []struct{ name, sqliteDDL, pgDDL string }{
|
||||
{"cache",
|
||||
`CREATE TABLE IF NOT EXISTS cache (
|
||||
key TEXT PRIMARY KEY,
|
||||
value TEXT,
|
||||
created_at INTEGER DEFAULT (strftime('%s', 'now'))
|
||||
)`,
|
||||
`CREATE TABLE IF NOT EXISTS cache (
|
||||
key TEXT PRIMARY KEY,
|
||||
value TEXT,
|
||||
created_at INTEGER DEFAULT 0
|
||||
)`},
|
||||
{"statistics",
|
||||
`CREATE TABLE IF NOT EXISTS statistics (
|
||||
id INTEGER PRIMARY KEY CHECK (id = 1),
|
||||
total_requests INTEGER DEFAULT 0,
|
||||
search_requests INTEGER DEFAULT 0,
|
||||
detail_requests INTEGER DEFAULT 0,
|
||||
catalog_requests INTEGER DEFAULT 0,
|
||||
content_requests INTEGER DEFAULT 0,
|
||||
manga_requests INTEGER DEFAULT 0,
|
||||
audio_requests INTEGER DEFAULT 0,
|
||||
video_requests INTEGER DEFAULT 0,
|
||||
category_requests INTEGER DEFAULT 0,
|
||||
updated_at INTEGER DEFAULT (strftime('%s', 'now'))
|
||||
)`,
|
||||
`CREATE TABLE IF NOT EXISTS statistics (
|
||||
id INTEGER PRIMARY KEY CHECK (id = 1),
|
||||
total_requests INTEGER DEFAULT 0,
|
||||
search_requests INTEGER DEFAULT 0,
|
||||
detail_requests INTEGER DEFAULT 0,
|
||||
catalog_requests INTEGER DEFAULT 0,
|
||||
content_requests INTEGER DEFAULT 0,
|
||||
manga_requests INTEGER DEFAULT 0,
|
||||
audio_requests INTEGER DEFAULT 0,
|
||||
video_requests INTEGER DEFAULT 0,
|
||||
category_requests INTEGER DEFAULT 0,
|
||||
updated_at INTEGER DEFAULT 0
|
||||
)`},
|
||||
{"device_pool",
|
||||
`CREATE TABLE IF NOT EXISTS device_pool (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
device_header TEXT NOT NULL,
|
||||
key TEXT NOT NULL,
|
||||
key_ver INTEGER NOT NULL,
|
||||
is_valid INTEGER DEFAULT 1,
|
||||
created_at INTEGER NOT NULL,
|
||||
last_used_at INTEGER NOT NULL
|
||||
)`,
|
||||
`CREATE TABLE IF NOT EXISTS device_pool (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
device_header TEXT NOT NULL,
|
||||
key TEXT NOT NULL,
|
||||
key_ver INTEGER NOT NULL,
|
||||
is_valid INTEGER DEFAULT 1,
|
||||
created_at INTEGER NOT NULL,
|
||||
last_used_at INTEGER NOT NULL
|
||||
)`},
|
||||
{"audio_cache",
|
||||
`CREATE TABLE IF NOT EXISTS audio_cache (
|
||||
key TEXT PRIMARY KEY,
|
||||
original_url TEXT NOT NULL,
|
||||
device_id TEXT NOT NULL,
|
||||
headers TEXT,
|
||||
created_at INTEGER DEFAULT (strftime('%s', 'now'))
|
||||
)`,
|
||||
`CREATE TABLE IF NOT EXISTS audio_cache (
|
||||
key TEXT PRIMARY KEY,
|
||||
original_url TEXT NOT NULL,
|
||||
device_id TEXT NOT NULL,
|
||||
headers TEXT,
|
||||
created_at INTEGER DEFAULT 0
|
||||
)`},
|
||||
}
|
||||
|
||||
for _, t := range tables {
|
||||
ddl := t.sqliteDDL
|
||||
if dbDriver == "pgx" {
|
||||
ddl = t.pgDDL
|
||||
}
|
||||
if _, err := cacheDB.Exec(ddl); err != nil {
|
||||
cacheLogger.Error("创建" + t.name + "表失败: " + err.Error())
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// 初始化statistics表(如果不存在记录)
|
||||
_, err = cacheDB.Exec(`INSERT OR IGNORE INTO statistics (id) VALUES (1)`)
|
||||
if err != nil {
|
||||
if _, err := cacheDB.Exec(`INSERT INTO statistics (id) VALUES (1) ON CONFLICT (id) DO NOTHING`); err != nil {
|
||||
cacheLogger.Error("初始化statistics表失败: " + err.Error())
|
||||
}
|
||||
|
||||
// 创建索引以提高清理效率
|
||||
_, err = cacheDB.Exec(`CREATE INDEX IF NOT EXISTS idx_created_at ON cache(created_at)`)
|
||||
if err != nil {
|
||||
if _, err := cacheDB.Exec(`CREATE INDEX IF NOT EXISTS idx_created_at ON cache(created_at)`); err != nil {
|
||||
cacheLogger.Error("创建索引失败: " + err.Error())
|
||||
}
|
||||
|
||||
// 创建device_pool表(设备密钥池)
|
||||
_, err = cacheDB.Exec(`CREATE TABLE IF NOT EXISTS device_pool (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
device_header TEXT NOT NULL,
|
||||
key TEXT NOT NULL,
|
||||
key_ver INTEGER NOT NULL,
|
||||
is_valid INTEGER DEFAULT 1,
|
||||
created_at INTEGER NOT NULL,
|
||||
last_used_at INTEGER NOT NULL
|
||||
)`)
|
||||
if err != nil {
|
||||
cacheLogger.Error("创建device_pool表失败: " + err.Error())
|
||||
return err
|
||||
}
|
||||
|
||||
// 创建device_pool索引
|
||||
_, err = cacheDB.Exec(`CREATE INDEX IF NOT EXISTS idx_device_pool_valid ON device_pool(is_valid)`)
|
||||
if err != nil {
|
||||
if _, err := cacheDB.Exec(`CREATE INDEX IF NOT EXISTS idx_device_pool_valid ON device_pool(is_valid)`); err != nil {
|
||||
cacheLogger.Warning("创建device_pool索引失败: " + err.Error())
|
||||
}
|
||||
|
||||
// 创建audio_cache表(音频代理缓存)
|
||||
_, err = cacheDB.Exec(`CREATE TABLE IF NOT EXISTS audio_cache (
|
||||
key TEXT PRIMARY KEY,
|
||||
original_url TEXT NOT NULL,
|
||||
device_id TEXT NOT NULL,
|
||||
headers TEXT,
|
||||
created_at INTEGER DEFAULT (strftime('%s', 'now'))
|
||||
)`)
|
||||
if err != nil {
|
||||
cacheLogger.Error("创建audio_cache表失败: " + err.Error())
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -141,7 +262,11 @@ func GetCache(key string) string {
|
||||
if !isCacheEnabled() {
|
||||
return ""
|
||||
}
|
||||
|
||||
|
||||
if redisCache != nil {
|
||||
return redisGetCache(key)
|
||||
}
|
||||
|
||||
if cacheDB == nil {
|
||||
return ""
|
||||
}
|
||||
@@ -150,7 +275,7 @@ func GetCache(key string) string {
|
||||
defer cacheMutex.Unlock()
|
||||
|
||||
var value string
|
||||
err := cacheDB.QueryRow("SELECT value FROM cache WHERE key = ?", key).Scan(&value)
|
||||
err := cacheDB.QueryRow(q("SELECT value FROM cache WHERE key = ?"), key).Scan(&value)
|
||||
if err != nil {
|
||||
if err != sql.ErrNoRows {
|
||||
cacheLogger.Error("查询缓存失败: " + err.Error())
|
||||
@@ -177,13 +302,16 @@ func GetCache(key string) string {
|
||||
return ""
|
||||
}
|
||||
|
||||
// UpdateCache 更新缓存
|
||||
// UpdateCache 更新或插入正文内容缓存
|
||||
func UpdateCache(key string, value string, expi int) error {
|
||||
if !isCacheEnabled() {
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
if redisCache != nil {
|
||||
return redisUpdateCache(key, value, expi)
|
||||
}
|
||||
|
||||
if cacheDB == nil {
|
||||
return nil
|
||||
}
|
||||
@@ -201,18 +329,9 @@ func UpdateCache(key string, value string, expi int) error {
|
||||
return err
|
||||
}
|
||||
|
||||
// 检查是否存在
|
||||
var exists bool
|
||||
err = cacheDB.QueryRow("SELECT 1 FROM cache WHERE key = ?", key).Scan(&exists)
|
||||
|
||||
if err == sql.ErrNoRows {
|
||||
// 不存在,插入
|
||||
_, err = cacheDB.Exec("INSERT INTO cache (key, value) VALUES (?, ?)", key, string(jsonData))
|
||||
} else {
|
||||
// 存在,更新
|
||||
_, err = cacheDB.Exec("UPDATE cache SET value = ? WHERE key = ?", string(jsonData), key)
|
||||
}
|
||||
|
||||
// UPSERT:ON CONFLICT 语法 SQLite 3.24+ 与 PostgreSQL 通用
|
||||
_, err = cacheDB.Exec(q(`INSERT INTO cache (key, value, created_at) VALUES (?, ?, ?)
|
||||
ON CONFLICT(key) DO UPDATE SET value = excluded.value`), key, string(jsonData), time.Now().Unix())
|
||||
if err != nil {
|
||||
cacheLogger.Error("更新缓存失败: " + err.Error())
|
||||
return err
|
||||
@@ -248,7 +367,11 @@ func CleanupAllCache() error {
|
||||
return err
|
||||
}
|
||||
|
||||
// 智能压缩:仅在空闲空间超过阈值时执行
|
||||
// 智能压缩:仅在空闲空间超过阈值时执行(仅 SQLite,PG 由 autovacuum 自行管理)
|
||||
if dbDriver == "pgx" {
|
||||
return nil
|
||||
}
|
||||
|
||||
var freelistCount int64
|
||||
err = cacheDB.QueryRow("PRAGMA freelist_count").Scan(&freelistCount)
|
||||
if err == nil && freelistCount > 0 {
|
||||
@@ -304,7 +427,7 @@ func CleanupExpiredCache() error {
|
||||
expiredTime := time.Now().Unix() - int64(cfg.Cache.CleanupSeconds)
|
||||
|
||||
// 删除过期的缓存(created_at < expiredTime)
|
||||
result, err := cacheDB.Exec("DELETE FROM cache WHERE created_at < ?", expiredTime)
|
||||
result, err := cacheDB.Exec(q("DELETE FROM cache WHERE created_at < ?"), expiredTime)
|
||||
if err != nil {
|
||||
cacheLogger.Error("清理过期缓存失败: " + err.Error())
|
||||
return err
|
||||
@@ -396,6 +519,10 @@ func VacuumDatabase() error {
|
||||
defer cacheMutex.Unlock()
|
||||
|
||||
cacheLogger.Info("开始执行数据库压缩...")
|
||||
if dbDriver == "pgx" {
|
||||
cacheLogger.Info("PostgreSQL 由 autovacuum 自动管理,无需手动压缩")
|
||||
return nil
|
||||
}
|
||||
_, err := cacheDB.Exec("VACUUM")
|
||||
if err != nil {
|
||||
cacheLogger.Error("数据库压缩失败: " + err.Error())
|
||||
@@ -426,7 +553,21 @@ func GetDatabaseStats() (map[string]interface{}, error) {
|
||||
}
|
||||
stats["cache_count"] = cacheCount
|
||||
|
||||
// 获取数据库页面统计
|
||||
// PostgreSQL:用库大小填充,无 freelist 概念
|
||||
if dbDriver == "pgx" {
|
||||
var totalSize int64
|
||||
if err := cacheDB.QueryRow("SELECT pg_database_size(current_database())").Scan(&totalSize); err == nil {
|
||||
stats["page_count"] = int64(0)
|
||||
stats["page_size"] = int64(0)
|
||||
stats["freelist_count"] = int64(0)
|
||||
stats["total_size_bytes"] = totalSize
|
||||
stats["free_size_bytes"] = int64(0)
|
||||
stats["used_size_bytes"] = totalSize
|
||||
}
|
||||
return stats, nil
|
||||
}
|
||||
|
||||
// SQLite:获取数据库页面统计
|
||||
var pageCount, pageSize, freelistCount int64
|
||||
err = cacheDB.QueryRow("PRAGMA page_count").Scan(&pageCount)
|
||||
if err == nil {
|
||||
@@ -458,8 +599,8 @@ func SaveStatistics(stats map[string]int64) error {
|
||||
cacheMutex.Lock()
|
||||
defer cacheMutex.Unlock()
|
||||
|
||||
_, err := cacheDB.Exec(`
|
||||
UPDATE statistics
|
||||
_, err := cacheDB.Exec(q(`
|
||||
UPDATE statistics
|
||||
SET total_requests = ?,
|
||||
search_requests = ?,
|
||||
detail_requests = ?,
|
||||
@@ -469,10 +610,11 @@ func SaveStatistics(stats map[string]int64) error {
|
||||
audio_requests = ?,
|
||||
video_requests = ?,
|
||||
category_requests = ?,
|
||||
updated_at = strftime('%s', 'now')
|
||||
updated_at = ?
|
||||
WHERE id = 1
|
||||
`, stats["total"], stats["search"], stats["detail"], stats["catalog"],
|
||||
stats["content"], stats["manga"], stats["audio"], stats["video"], stats["category"])
|
||||
`), stats["total"], stats["search"], stats["detail"], stats["catalog"],
|
||||
stats["content"], stats["manga"], stats["audio"], stats["video"], stats["category"],
|
||||
time.Now().Unix())
|
||||
|
||||
if err != nil {
|
||||
cacheLogger.Error("保存统计数据失败: " + err.Error())
|
||||
|
||||
@@ -385,6 +385,14 @@ func (p *DeviceKeyPool) loadFromDB() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// boolToInt 布尔转整数存储(is_valid 列为 INTEGER,PostgreSQL 严格要求类型匹配)
|
||||
func boolToInt(b bool) int {
|
||||
if b {
|
||||
return 1
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
// saveDeviceToDB 保存设备到数据库
|
||||
func (p *DeviceKeyPool) saveDeviceToDB(item *DeviceKeyItem) error {
|
||||
if cacheDB == nil {
|
||||
@@ -394,23 +402,15 @@ func (p *DeviceKeyPool) saveDeviceToDB(item *DeviceKeyItem) error {
|
||||
cacheMutex.Lock()
|
||||
defer cacheMutex.Unlock()
|
||||
|
||||
result, err := cacheDB.Exec(`
|
||||
// RETURNING id 同时兼容 SQLite(3.35+) 与 PostgreSQL,替代 LastInsertId()
|
||||
err := cacheDB.QueryRow(q(`
|
||||
INSERT INTO device_pool (device_header, key, key_ver, is_valid, created_at, last_used_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?)
|
||||
`, item.DeviceHeader, item.Key, item.KeyVer, item.IsValid,
|
||||
item.CreatedAt.Unix(), item.LastUsedAt.Unix())
|
||||
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
RETURNING id
|
||||
`), item.DeviceHeader, item.Key, item.KeyVer, boolToInt(item.IsValid),
|
||||
item.CreatedAt.Unix(), item.LastUsedAt.Unix()).Scan(&item.ID)
|
||||
|
||||
// 获取插入的ID
|
||||
id, err := result.LastInsertId()
|
||||
if err == nil {
|
||||
item.ID = int(id)
|
||||
}
|
||||
|
||||
return nil
|
||||
return err
|
||||
}
|
||||
|
||||
// updateDeviceInDB 更新设备信息
|
||||
@@ -422,11 +422,11 @@ func (p *DeviceKeyPool) updateDeviceInDB(item *DeviceKeyItem) error {
|
||||
cacheMutex.Lock()
|
||||
defer cacheMutex.Unlock()
|
||||
|
||||
_, err := cacheDB.Exec(`
|
||||
UPDATE device_pool
|
||||
_, err := cacheDB.Exec(q(`
|
||||
UPDATE device_pool
|
||||
SET device_header = ?, key = ?, key_ver = ?, is_valid = ?, last_used_at = ?
|
||||
WHERE id = ?
|
||||
`, item.DeviceHeader, item.Key, item.KeyVer, item.IsValid,
|
||||
`), item.DeviceHeader, item.Key, item.KeyVer, boolToInt(item.IsValid),
|
||||
item.LastUsedAt.Unix(), item.ID)
|
||||
|
||||
return err
|
||||
@@ -454,35 +454,6 @@ func (p *DeviceKeyPool) removeInvalidFromDB() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// InitDevicePoolTable 初始化设备池表
|
||||
func InitDevicePoolTable() error {
|
||||
if cacheDB == nil {
|
||||
return fmt.Errorf("数据库未初始化")
|
||||
}
|
||||
|
||||
_, err := cacheDB.Exec(`CREATE TABLE IF NOT EXISTS device_pool (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
device_header TEXT NOT NULL,
|
||||
key TEXT NOT NULL,
|
||||
key_ver INTEGER NOT NULL,
|
||||
is_valid INTEGER DEFAULT 1,
|
||||
created_at INTEGER NOT NULL,
|
||||
last_used_at INTEGER NOT NULL
|
||||
)`)
|
||||
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// 创建索引
|
||||
_, err = cacheDB.Exec(`CREATE INDEX IF NOT EXISTS idx_device_pool_valid ON device_pool(is_valid)`)
|
||||
if err != nil {
|
||||
poolLogger.Warning("创建索引失败: " + err.Error())
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// GetDeviceHeaderFromPool 从池中获取设备header(兼容旧接口)
|
||||
func GetDeviceHeaderFromPool() string {
|
||||
pool := GetDeviceKeyPool()
|
||||
|
||||
@@ -0,0 +1,20 @@
|
||||
package fqapi
|
||||
|
||||
import "testing"
|
||||
|
||||
// TestRebind 验证占位符方言转换
|
||||
func TestRebind(t *testing.T) {
|
||||
in := "INSERT INTO cache (key, value) VALUES (?, ?)"
|
||||
|
||||
if got := q(in); got != in {
|
||||
t.Errorf("sqlite 下应保持 ?: %s", got)
|
||||
}
|
||||
|
||||
dbDriver = "pgx"
|
||||
defer func() { dbDriver = "sqlite3" }()
|
||||
|
||||
want := "INSERT INTO cache (key, value) VALUES ($1, $2)"
|
||||
if got := q(in); got != want {
|
||||
t.Errorf("pg 下应转换占位符: got %s want %s", got, want)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,85 @@
|
||||
package fqapi
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
)
|
||||
|
||||
// audioProxyEntry 音频代理缓存条目(Redis 版)
|
||||
type audioProxyEntry struct {
|
||||
OriginalURL string `json:"original_url"`
|
||||
DeviceID string `json:"device_id"`
|
||||
Headers map[string]string `json:"headers"`
|
||||
}
|
||||
|
||||
const (
|
||||
redisCacheKeyPrefix = "fq:cache:"
|
||||
redisAudioKeyPrefix = "fq:audio:"
|
||||
// audioCacheTTL 音频代理缓存有效期(匹配 CDN 链接有效期)
|
||||
audioCacheTTL = 2 * time.Hour
|
||||
)
|
||||
|
||||
// redisGetCache 从 Redis 获取正文缓存
|
||||
func redisGetCache(key string) string {
|
||||
val, err := redisCache.Get(context.Background(), redisCacheKeyPrefix+key).Result()
|
||||
if err != nil {
|
||||
return "" // miss 或错误都视为未命中
|
||||
}
|
||||
|
||||
var cacheData CacheData
|
||||
if err := json.Unmarshal([]byte(val), &cacheData); err != nil {
|
||||
cacheLogger.Error("解析Redis缓存数据失败: " + err.Error())
|
||||
return ""
|
||||
}
|
||||
|
||||
cacheLogger.Debug("Redis缓存命中: " + key)
|
||||
if time.Now().Unix() < int64(cacheData.Expi) {
|
||||
return cacheData.Data
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// redisUpdateCache 写入 Redis 正文缓存(TTL 与过期时间一致)
|
||||
func redisUpdateCache(key string, value string, expi int) error {
|
||||
cacheData := CacheData{
|
||||
Data: value,
|
||||
Expi: float64(time.Now().Unix() + int64(expi)),
|
||||
}
|
||||
jsonData, err := json.Marshal(cacheData)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
||||
defer cancel()
|
||||
return redisCache.Set(ctx, redisCacheKeyPrefix+key, jsonData, time.Duration(expi)*time.Second).Err()
|
||||
}
|
||||
|
||||
// saveAudioProxyEntry 将音频代理信息存入 Redis
|
||||
func saveAudioProxyEntry(key string, entry *audioProxyEntry) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
||||
defer cancel()
|
||||
jsonData, _ := json.Marshal(entry)
|
||||
if err := redisCache.Set(ctx, redisAudioKeyPrefix+key, jsonData, audioCacheTTL).Err(); err != nil {
|
||||
audioLogger.Error("保存音频缓存到Redis失败: " + err.Error())
|
||||
}
|
||||
}
|
||||
|
||||
// loadAudioProxyEntry 从 Redis 读取音频代理信息,未命中或过期返回错误
|
||||
func loadAudioProxyEntry(key string) (*audioProxyEntry, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
||||
defer cancel()
|
||||
|
||||
val, err := redisCache.Get(ctx, redisAudioKeyPrefix+key).Result()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("音频信息不存在或已过期,请重新调用 /audio 接口")
|
||||
}
|
||||
|
||||
var entry audioProxyEntry
|
||||
if err := json.Unmarshal([]byte(val), &entry); err != nil {
|
||||
return nil, fmt.Errorf("解析音频缓存失败: %w", err)
|
||||
}
|
||||
return &entry, nil
|
||||
}
|
||||
Reference in new issue
Block a user