package fqapi import ( "context" "database/sql" "encoding/json" "fanqie/internal/common" "fanqie/internal/config" "fmt" "net/url" "strconv" "strings" "sync/atomic" "time" _ "github.com/jackc/pgx/v5/stdlib" _ "github.com/mattn/go-sqlite3" "github.com/redis/go-redis/v9" "golang.org/x/sync/singleflight" ) const ( // sqlCacheTTLSeconds SQL 持久层保留时长:不可变内容(正文/详情/目录)与业务 TTL 解耦, // Redis 过期或重启后由 SQL 回填,避免重新打上游 sqlCacheTTLSeconds = 30 * 86400 ) // volatileCachePrefixes 动态短缓存(值会变)只走 Redis/短 TTL,不进 SQL 长期层 var volatileCachePrefixes = []string{"search_", "category_", "rank_"} // isVolatileCacheKey 判断是否为动态短缓存 key func isVolatileCacheKey(key string) bool { for _, p := range volatileCachePrefixes { if strings.HasPrefix(key, p) { return true } } return false } // cacheSF 合并并发 miss 的回源请求,防止缓存击穿 var cacheSF singleflight.Group var ( cacheDB *sql.DB dbDriver string // "sqlite3" | "pgx" redisCache *redis.Client cacheLogger = common.NewLogger("cache_manager", "INFO", "") contentRequestCount atomic.Int64 // content接口访问计数 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.GetAppConfig() 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, // 缓存场景失败不重试:miss 直接降级 SQL,重试只会放大网络慢时的请求延迟 MaxRetries: 0, ReadTimeout: 3 * time.Second, WriteTimeout: 3 * time.Second, MinRetryBackoff: 0, }) 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 初始化数据库(按 cache.backend 选择 sqlite 或 postgres) func InitDB() error { if cacheDB != nil { return nil // 已初始化 } cfg := config.GetAppConfig() 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(dbDriver, dsn) if err != nil { cacheLogger.Error("打开数据库失败: " + err.Error()) return err } if err := cacheDB.Ping(); err != nil { cacheLogger.Error("连接数据库失败: " + err.Error()) return err } cacheLogger.Info(fmt.Sprintf("数据库已连接: %s (driver=%s)", backend, dbDriver)) 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, expi INTEGER DEFAULT 0, created_at INTEGER DEFAULT (strftime('%s', 'now')), updated_at INTEGER DEFAULT (strftime('%s', 'now')) )`, `CREATE TABLE IF NOT EXISTS cache ( key TEXT PRIMARY KEY, value TEXT, expi BIGINT DEFAULT 0, created_at BIGINT DEFAULT extract(epoch FROM now())::bigint, updated_at BIGINT DEFAULT extract(epoch FROM now())::bigint )`}, {"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, created_at INTEGER DEFAULT (strftime('%s', 'now')), 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, created_at BIGINT DEFAULT extract(epoch FROM now())::bigint, updated_at BIGINT DEFAULT extract(epoch FROM now())::bigint )`}, {"device_pool", `CREATE TABLE IF NOT EXISTS device_pool ( id INTEGER PRIMARY KEY AUTOINCREMENT, device_header TEXT NOT NULL UNIQUE, 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, updated_at INTEGER DEFAULT (strftime('%s', 'now')) )`, `CREATE TABLE IF NOT EXISTS device_pool ( id BIGSERIAL PRIMARY KEY, device_header TEXT NOT NULL UNIQUE, 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, updated_at BIGINT DEFAULT extract(epoch FROM now())::bigint )`}, {"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')), updated_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 BIGINT DEFAULT extract(epoch FROM now())::bigint, updated_at BIGINT DEFAULT extract(epoch FROM now())::bigint )`}, } 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表(如果不存在记录) if _, err := cacheDB.Exec(`INSERT INTO statistics (id) VALUES (1) ON CONFLICT (id) DO NOTHING`); err != nil { cacheLogger.Error("初始化statistics表失败: " + err.Error()) } // 创建索引以提高清理效率 if _, err := cacheDB.Exec(`CREATE INDEX IF NOT EXISTS idx_created_at ON cache(created_at)`); err != nil { cacheLogger.Error("创建索引失败: " + err.Error()) } 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()) } // 旧库补唯一约束:表已存在时 DDL 的 UNIQUE 不会生效,用唯一索引补齐; // 若存量数据已有重复 device_header 则创建失败,仅告警提示人工清理 if _, err := cacheDB.Exec(`CREATE UNIQUE INDEX IF NOT EXISTS idx_device_pool_header ON device_pool(device_header)`); err != nil { cacheLogger.Warning("补建device_header唯一索引失败(可能存在重复设备,清理后重启即可): " + err.Error()) } // 旧表补 created_at/updated_at 列(新表 DDL 自带,此处仅老库生效) for _, m := range []struct{ table, col, ddl string }{ {"cache", "updated_at", "INTEGER DEFAULT 0"}, // SQLite ALTER 不允许非常量默认值,先补 0 再回填 {"cache", "expi", "INTEGER DEFAULT 0"}, {"statistics", "created_at", "INTEGER DEFAULT 0"}, {"device_pool", "updated_at", "INTEGER DEFAULT 0"}, {"device_pool", "key_ts", "INTEGER DEFAULT 0"}, {"audio_cache", "updated_at", "INTEGER DEFAULT 0"}, } { if dbDriver == "pgx" { m.ddl = "BIGINT DEFAULT 0" } alter := "ALTER TABLE " + m.table + " ADD COLUMN " if dbDriver == "pgx" { alter += "IF NOT EXISTS " } if _, err := cacheDB.Exec(alter + m.col + " " + m.ddl); err != nil { cacheLogger.Debug("补列跳过(已存在): " + m.table + "." + m.col) } } // 旧行回填(幂等:仅 0 值行受影响) for _, backfill := range []string{ `UPDATE cache SET updated_at = created_at WHERE updated_at = 0`, `UPDATE cache SET expi = created_at + 86400 WHERE expi = 0`, `UPDATE statistics SET created_at = updated_at WHERE created_at = 0`, `UPDATE device_pool SET updated_at = last_used_at WHERE updated_at = 0`, `UPDATE audio_cache SET updated_at = created_at WHERE updated_at = 0`, } { if _, err := cacheDB.Exec(backfill); err != nil { cacheLogger.Warning("时间列回填失败: " + err.Error()) } } return nil } // isCacheEnabled 检查正文内容缓存是否启用 func isCacheEnabled() bool { cfg := config.GetAppConfig() if cfg == nil { return false } return cfg.Cache.Enabled } // withCache 只读接口的缓存包装:命中直接返回,miss 回源后写入(仅缓存成功响应)。 // singleflight 合并并发 miss,同一 key 只有一个请求打上游 // ponytail: 空参等防御分支若漏到此处,其错误 JSON 也会被缓存到 TTL 到期——handler 层 validateParams 已拦截 func withCache(key string, ttl int, fetch func() (string, error)) (string, error) { if c := GetCache(key); c != "" { // 旧缓存可能存有未解包的 JSON 字符串字段,出口幂等解包 // (对新写入的已解包数据无影响) return reencodeUnwrapped(c), nil } v, err, _ := cacheSF.Do(key, func() (interface{}, error) { // 双重检查:等锁期间可能已被并发请求填充 if c := GetCache(key); c != "" { return c, nil } result, err := fetch() if err == nil { UpdateCache(key, result, ttl) } return result, err }) if err != nil { return "", err } return v.(string), nil } // GetCache 获取缓存:Redis 热层 → SQL 持久层(命中回填 Redis)→ 未命中 func GetCache(key string) string { if !isCacheEnabled() { return "" } if redisCache != nil { if v := redisGetCache(key); v != "" { return v } if isVolatileCacheKey(key) { return "" // 搜索/分类等动态数据不落库,Redis 未命中即回源 } } if cacheDB == nil { return "" } var value string var expi int64 err := cacheDB.QueryRow(q("SELECT value, expi FROM cache WHERE key = ?"), key).Scan(&value, &expi) if err != nil { if err != sql.ErrNoRows { cacheLogger.Error("查询缓存失败: " + err.Error()) } return "" } // 检查是否过期(expi 为 0 的旧行由启动回填兜底,此处防御) if time.Now().Unix() >= expi { cacheLogger.Debug("缓存已过期(SQL): " + key) return "" } var cacheData CacheData if err := json.Unmarshal([]byte(value), &cacheData); err != nil { cacheLogger.Error("解析缓存数据失败: " + err.Error()) return "" } cacheLogger.Debug("缓存命中(SQL): " + key) // SQL 命中而 Redis 未命中(过期/重启):异步回填热层,剩余时长为 TTL, // 回填(尤其大 value)不阻塞本次请求 if redisCache != nil { remain := int(expi - time.Now().Unix()) if remain < 60 { remain = 60 } go redisUpdateCache(key, cacheData.Data, remain) } return cacheData.Data } // UpdateCache 写入缓存:Redis 热层(TTL=业务过期时间)+ SQL 持久层双写。 // 不可变内容在 SQL 固定保留 30 天,动态短缓存(search/category)只走 Redis。 // Redis 写失败只降级不阻断:持久层照常落库,热层下次请求回填 func UpdateCache(key string, value string, expi int) error { if !isCacheEnabled() { return nil } if redisCache != nil { if err := redisUpdateCache(key, value, expi); err != nil { cacheLogger.Warning(fmt.Sprintf("Redis 写入失败(降级为仅 SQL 落库): %v", err)) } else if isVolatileCacheKey(key) { return nil // 动态短缓存不落库(Redis 成功即止) } } if cacheDB == nil { return nil } cacheData := CacheData{ Data: value, Expi: float64(time.Now().Unix() + int64(expi)), } jsonData, err := json.Marshal(cacheData) if err != nil { return err } // SQL 保留时长:不可变内容固定 30 天(Redis 未启用时沿用业务 TTL,保持纯 SQL 模式原行为) sqlTTL := int64(sqlCacheTTLSeconds) if redisCache == nil || isVolatileCacheKey(key) { sqlTTL = int64(expi) } // UPSERT:ON CONFLICT 语法 SQLite 3.24+ 与 PostgreSQL 通用 now := time.Now().Unix() _, err = cacheDB.Exec(q(`INSERT INTO cache (key, value, expi, created_at, updated_at) VALUES (?, ?, ?, ?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value, expi = excluded.expi, updated_at = excluded.updated_at`), key, string(jsonData), now+sqlTTL, now, now) if err != nil { cacheLogger.Error("更新缓存失败: " + err.Error()) return err } cacheLogger.Debug("缓存已更新: " + key) return nil } // CloseCache 关闭缓存数据库 func CloseCache() { if cacheDB != nil { cacheDB.Close() } } // CleanupAllCache 立即清理所有正文缓存 func CleanupAllCache() error { if cacheDB == nil { return fmt.Errorf("数据库未初始化") } // 删除所有缓存数据 _, err := cacheDB.Exec("DELETE FROM cache") if err != nil { 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 { var pageSize int64 err = cacheDB.QueryRow("PRAGMA page_size").Scan(&pageSize) if err == nil { freeBytes := freelistCount * pageSize // 阈值:空闲空间超过 5MB 时才执行压缩 const vacuumThreshold = 5 * 1024 * 1024 // 5MB if freeBytes >= vacuumThreshold { cacheLogger.Info(fmt.Sprintf("空闲空间 %.2f MB,开始执行压缩...", float64(freeBytes)/(1024*1024))) _, err = cacheDB.Exec("VACUUM") if err != nil { cacheLogger.Warning("数据库压缩失败: " + err.Error()) } else { cacheLogger.Info(fmt.Sprintf("数据库压缩完成,已释放 %.2f MB 磁盘空间", float64(freeBytes)/(1024*1024))) } } else { cacheLogger.Debug(fmt.Sprintf("空闲空间仅 %.2f MB,跳过压缩", float64(freeBytes)/(1024*1024))) } } } return nil } // CleanupExpiredCache 清理过期的正文缓存 func CleanupExpiredCache() error { cfg := config.GetAppConfig() if cfg == nil { return fmt.Errorf("配置未加载") } // 检查是否启用清理 if !cfg.Cache.CleanupEnabled { return nil } // 检查清理时间配置 if cfg.Cache.CleanupSeconds <= 0 { return nil } if cacheDB == nil { return fmt.Errorf("数据库未初始化") } // 删除已过期的缓存(按行内过期时间 expi,而非创建时间) result, err := cacheDB.Exec(q("DELETE FROM cache WHERE expi < ?"), time.Now().Unix()) if err != nil { cacheLogger.Error("清理过期缓存失败: " + err.Error()) return err } // 获取删除的行数 rowsAffected, err := result.RowsAffected() if err != nil { cacheLogger.Warning("获取删除行数失败: " + err.Error()) } else if rowsAffected > 0 { cacheLogger.Info(fmt.Sprintf("已清理 %d 条过期缓存(SQL 持久层)", rowsAffected)) } // 顺带清理磁盘视频缓存(7 天未访问的解密文件) CleanupVideoCache() return nil } // StartCacheCleanup 启动缓存自动清理任务 func StartCacheCleanup() { cfg := config.GetAppConfig() if cfg == nil || !cfg.Cache.CleanupEnabled || cfg.Cache.CleanupSeconds <= 0 { cacheLogger.Info("缓存自动清理未启用或配置无效,跳过启动") return } // 每小时执行一次清理(可以根据需要调整间隔) cleanupInterval := 1 * time.Hour cacheCleanupTicker = time.NewTicker(cleanupInterval) cacheLogger.Info(fmt.Sprintf("缓存自动清理已启动,清理间隔: %v,过期时间: %d 秒", cleanupInterval, cfg.Cache.CleanupSeconds)) // 立即执行一次清理 go func() { if err := CleanupExpiredCache(); err != nil { cacheLogger.Error("初始清理失败: " + err.Error()) } }() // 定时执行清理 go func() { for range cacheCleanupTicker.C { if err := CleanupExpiredCache(); err != nil { cacheLogger.Error("定时清理失败: " + err.Error()) } } }() } // StopCacheCleanup 停止缓存自动清理任务 func StopCacheCleanup() { if cacheCleanupTicker != nil { cacheCleanupTicker.Stop() cacheLogger.Info("缓存自动清理已停止") // 最后执行一次清理 if err := CleanupExpiredCache(); err != nil { cacheLogger.Error("最后清理失败: " + err.Error()) } } } // IncrementContentRequestCount 增加content接口访问计数 func IncrementContentRequestCount() int { return int(contentRequestCount.Add(1)) } // ResetContentRequestCount 重置访问计数 func ResetContentRequestCount() { contentRequestCount.Store(0) } // GetContentRequestCount 获取当前访问计数 func GetContentRequestCount() int { return int(contentRequestCount.Load()) } // GetDatabaseStats 获取数据库统计信息 func GetDatabaseStats() (map[string]interface{}, error) { if cacheDB == nil { return nil, fmt.Errorf("数据库未初始化") } stats := make(map[string]interface{}) // 获取cache表记录数 var cacheCount int err := cacheDB.QueryRow("SELECT COUNT(*) FROM cache").Scan(&cacheCount) if err != nil { cacheLogger.Warning("获取cache表记录数失败: " + err.Error()) cacheCount = 0 } 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 { stats["page_count"] = pageCount } err = cacheDB.QueryRow("PRAGMA page_size").Scan(&pageSize) if err == nil { stats["page_size"] = pageSize } err = cacheDB.QueryRow("PRAGMA freelist_count").Scan(&freelistCount) if err == nil { stats["freelist_count"] = freelistCount stats["total_size_bytes"] = pageCount * pageSize stats["free_size_bytes"] = freelistCount * pageSize stats["used_size_bytes"] = (pageCount - freelistCount) * pageSize } return stats, nil } // SaveStatistics 保存请求统计到数据库 func SaveStatistics(stats map[string]int64) error { if cacheDB == nil { return fmt.Errorf("数据库未初始化") } _, err := cacheDB.Exec(q(` UPDATE statistics SET total_requests = ?, search_requests = ?, detail_requests = ?, catalog_requests = ?, content_requests = ?, manga_requests = ?, audio_requests = ?, video_requests = ?, category_requests = ?, updated_at = ? WHERE id = 1 `), 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()) return err } return nil } // LoadStatistics 从数据库加载请求统计 func LoadStatistics() (map[string]int64, error) { if cacheDB == nil { return nil, fmt.Errorf("数据库未初始化") } var total, search, detail, catalog, content, manga, audio, video, category int64 err := cacheDB.QueryRow(` SELECT total_requests, search_requests, detail_requests, catalog_requests, content_requests, manga_requests, audio_requests, video_requests, category_requests FROM statistics WHERE id = 1 `).Scan(&total, &search, &detail, &catalog, &content, &manga, &audio, &video, &category) if err == sql.ErrNoRows { // 没有数据,返回空统计 return map[string]int64{ "total": 0, "search": 0, "detail": 0, "catalog": 0, "content": 0, "manga": 0, "audio": 0, "video": 0, "category": 0, }, nil } if err != nil { cacheLogger.Error("加载统计数据失败: " + err.Error()) return nil, err } stats := map[string]int64{ "total": total, "search": search, "detail": detail, "catalog": catalog, "content": content, "manga": manga, "audio": audio, "video": video, "category": category, } return stats, nil }