From a5bf4bfdd45e941c20d1302b8c42f9ee4da14d16 Mon Sep 17 00:00:00 2001 From: ShukeBta <272197458+ShukeBta@users.noreply.github.com> Date: Mon, 15 Jun 2026 18:16:34 +0800 Subject: [PATCH] Improve large library storage backends and scans --- README.md | 47 ++- README_EN.md | 48 ++- cmd/server/main.go | 5 +- config.example.yaml | 14 + docker-compose.advanced.yml | 22 ++ docker-compose.search.yml | 45 +++ docker-compose.standard.yml | 41 +++ docker-compose.yml | 33 ++ go.mod | 7 + go.sum | 14 + internal/config/config.go | 41 ++- internal/config/config_test.go | 27 ++ internal/database/database.go | 237 ++++++++++++- internal/database/database_test.go | 42 +++ internal/handler/media.go | 67 +++- internal/handler/media_test.go | 85 +++++ internal/handler/stats.go | 58 ++++ internal/handler/stats_extra.go | 8 +- internal/handler/stats_test.go | 51 +++ internal/repository/opensearch.go | 212 ++++++++++++ internal/repository/opensearch_test.go | 57 +++ internal/repository/repository.go | 102 ++++++ internal/repository/repository_test.go | 54 +++ internal/service/boot_cloud.go | 2 +- internal/service/emby_compat.go | 81 ++++- internal/service/emby_compat_test.go | 40 +++ internal/service/media.go | 262 +++++++++++++- internal/service/media_test.go | 37 ++ internal/service/organizer.go | 3 +- internal/service/runtime_cache.go | 193 +++++++++++ internal/service/scanner.go | 343 ++++++++++++++++--- internal/service/scanner_incremental_test.go | 70 ++++ internal/service/service.go | 18 +- internal/service/stats.go | 26 +- internal/service/transfer.go | 11 +- internal/service/transfer_test.go | 30 ++ 36 files changed, 2340 insertions(+), 93 deletions(-) create mode 100644 docker-compose.search.yml create mode 100644 docker-compose.standard.yml create mode 100644 internal/repository/opensearch.go create mode 100644 internal/repository/opensearch_test.go create mode 100644 internal/service/runtime_cache.go diff --git a/README.md b/README.md index 08437ed..5c99ab7 100644 --- a/README.md +++ b/README.md @@ -116,7 +116,19 @@ http://服务器IP:18080 ## Docker Compose 推荐部署 -仓库里的 `docker-compose.yml` 已经是最简单模板：默认不用 `.env`。 +仓库里的 `docker-compose.yml` 已经是轻量推荐模板：默认不用 `.env`，默认只启动 `MediaStationGo + PostgreSQL`，适合大多数 NAS。 + +旧版本如果已经有 `./data/mediastation.db`，首次使用新版 compose 启动时会自动导入到 PostgreSQL；`./data` 仍然要保留，用来保存 JWT 密钥、旧库迁移源和运行数据。 + +### 三种部署模式 + +| 模式 | 命令 | 适合场景 | +| --- | --- | --- | +| 轻量模式：PG only | `docker compose up -d` | 大多数 NAS，资源占用最低 | +| 标准模式：PG + Redis | `docker compose -f docker-compose.yml -f docker-compose.standard.yml up -d` | 多用户、Emby 客户端频繁刷新 | +| 搜索增强：PG + Redis + OpenSearch | `docker compose -f docker-compose.yml -f docker-compose.standard.yml -f docker-compose.search.yml up -d` | 超大库、后续独立搜索索引 | + +建议从轻量模式开始。Redis 和 OpenSearch 都是增强层，不是源数据库；低配 NAS 不要默认开启 OpenSearch。 ### 镜像地址怎么选 @@ -161,10 +173,13 @@ volumes: | 左边 | 右边 | 说明 | | --- | --- | --- | -| `./data` | `/data` | 程序数据库、配置、账号信息；一定要备份 | -| `./cache` | `/cache` | 缓存目录；可清理 | +| `./data` | 主程序 `/data` | 程序配置、JWT 密钥、旧 SQLite 迁移源；一定要备份 | +| `./cache` | 主程序 `/cache` | 缓存目录；可清理 | | `./media` | `/media` | 媒体库目录；自动整理入库需要可写，网页里添加媒体库时填 `/media/...` | | `./downloads` | `/downloads` | 下载目录；文件管理和自动整理会用 | +| `./postgres` | PostgreSQL `/var/lib/postgresql/data` | 新版默认主数据库；一定要备份 | +| `./redis` | Redis `/data` | 标准模式才会使用；热缓存，丢失可重建 | +| `./opensearch` | OpenSearch `/usr/share/opensearch/data` | 搜索增强模式才会使用；占用内存较高 | 如果你的媒体在 NAS 真实目录，例如： @@ -212,6 +227,9 @@ services: container_name: mediastation-go restart: unless-stopped init: true + depends_on: + postgres: + condition: service_healthy # 访问端口：浏览器打开 http://服务器IP:18080 ports: @@ -241,6 +259,10 @@ services: MEDIASTATION_APP_PORT: 8080 MEDIASTATION_APP_WEB_DIR: /app/web/dist MEDIASTATION_APP_DATA_DIR: /data + + # 轻量模式默认 PostgreSQL；旧 SQLite 会从这个路径自动迁移。 + MEDIASTATION_DATABASE_TYPE: postgres + MEDIASTATION_DATABASE_DSN: postgres://mediastation:mediastation@postgres:5432/mediastation?sslmode=disable MEDIASTATION_DATABASE_DB_PATH: /data/mediastation.db MEDIASTATION_CACHE_CACHE_DIR: /cache @@ -250,8 +272,27 @@ services: MEDIASTATION_MEDIA_CONTAINER_DIR: /media MEDIASTATION_DOWNLOAD_DIR: ./downloads MEDIASTATION_DOWNLOAD_CONTAINER_DIR: /downloads + + postgres: + image: postgres:16-alpine + container_name: mediastation-postgres + restart: unless-stopped + environment: + POSTGRES_DB: mediastation + POSTGRES_USER: mediastation + POSTGRES_PASSWORD: mediastation + volumes: + - ./postgres:/var/lib/postgresql/data + healthcheck: + test: ["CMD-SHELL", "pg_isready -U mediastation -d mediastation"] + interval: 10s + timeout: 5s + retries: 10 + ``` +> 说明：PostgreSQL 是主数据库；轻量模式也有进程内短缓存。Redis 是跨进程热缓存，OpenSearch 是搜索增强层，都不是源数据库。 + --- ## 首次进入后怎么配置 diff --git a/README_EN.md b/README_EN.md index 22eb6e3..b8727c0 100644 --- a/README_EN.md +++ b/README_EN.md @@ -116,7 +116,19 @@ Password: admin123 ## Docker Compose Recommended -The repository `docker-compose.yml` is intentionally simple and does not require `.env`. +The repository `docker-compose.yml` is the lightweight recommended template: no `.env` required, and by default it only starts `MediaStationGo + PostgreSQL`. This is the best starting point for most NAS users. + +If you already have an older `./data/mediastation.db`, the first start with the new compose file automatically imports it into PostgreSQL. Keep `./data`; it still stores the JWT secret, runtime data, and the old SQLite migration source. + +### Three deployment modes + +| Mode | Command | Best for | +| --- | --- | --- | +| Lightweight: PG only | `docker compose up -d` | Most NAS devices, lowest resource use | +| Standard: PG + Redis | `docker compose -f docker-compose.yml -f docker-compose.standard.yml up -d` | Multi-user use and frequent Emby client refreshes | +| Search enhanced: PG + Redis + OpenSearch | `docker compose -f docker-compose.yml -f docker-compose.standard.yml -f docker-compose.search.yml up -d` | Huge libraries and future standalone search indexing | + +Start with the lightweight mode. Redis and OpenSearch are enhancement layers, not source databases. Do not enable OpenSearch by default on low-memory NAS devices. ### Choose an image source @@ -161,10 +173,13 @@ Meaning: | Host path | Container path | Purpose | | --- | --- | --- | -| `./data` | `/data` | Database, users, settings; back this up | -| `./cache` | `/cache` | Cache; safe to clean when needed | +| `./data` | app `/data` | Settings, JWT secret, old SQLite migration source; back this up | +| `./cache` | app `/cache` | Cache; safe to clean when needed | | `./media` | `/media` | Media libraries; use `/media/...` in the web UI | | `./downloads` | `/downloads` | Download directory and organization source | +| `./postgres` | PostgreSQL `/var/lib/postgresql/data` | New default primary database; back this up | +| `./redis` | Redis `/data` | Used only in standard mode; hot cache, rebuildable | +| `./opensearch` | OpenSearch `/usr/share/opensearch/data` | Used only in search-enhanced mode; higher memory use | If your NAS paths are: @@ -212,6 +227,9 @@ services: container_name: mediastation-go restart: unless-stopped init: true + depends_on: + postgres: + condition: service_healthy # Browser: http://SERVER_IP:18080 ports: @@ -241,6 +259,11 @@ services: MEDIASTATION_APP_PORT: 8080 MEDIASTATION_APP_WEB_DIR: /app/web/dist MEDIASTATION_APP_DATA_DIR: /data + + # Lightweight mode uses PostgreSQL by default. + # Old SQLite data migrates from this path on first start. + MEDIASTATION_DATABASE_TYPE: postgres + MEDIASTATION_DATABASE_DSN: postgres://mediastation:mediastation@postgres:5432/mediastation?sslmode=disable MEDIASTATION_DATABASE_DB_PATH: /data/mediastation.db MEDIASTATION_CACHE_CACHE_DIR: /cache @@ -250,8 +273,27 @@ services: MEDIASTATION_MEDIA_CONTAINER_DIR: /media MEDIASTATION_DOWNLOAD_DIR: ./downloads MEDIASTATION_DOWNLOAD_CONTAINER_DIR: /downloads + + postgres: + image: postgres:16-alpine + container_name: mediastation-postgres + restart: unless-stopped + environment: + POSTGRES_DB: mediastation + POSTGRES_USER: mediastation + POSTGRES_PASSWORD: mediastation + volumes: + - ./postgres:/var/lib/postgresql/data + healthcheck: + test: ["CMD-SHELL", "pg_isready -U mediastation -d mediastation"] + interval: 10s + timeout: 5s + retries: 10 + ``` +> Note: PostgreSQL is the primary database. Lightweight mode still has short in-process caching. Redis is a cross-process hot cache, and OpenSearch is a search enhancement layer; neither is a source database. + --- ## First-time Setup diff --git a/cmd/server/main.go b/cmd/server/main.go index 9722e8c..35412e2 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -3,7 +3,7 @@ // MediaStationGo is a Go rewrite of the original Python MediaStation project, // adopting the same tech stack as cropflre/nowen-video: // -// Backend: Go 1.25 + Gin + GORM + SQLite (WAL) + Viper + Zap + JWT +// Backend: Go 1.25 + Gin + GORM + PostgreSQL/SQLite + Viper + Zap + JWT // Frontend: React 18 + Vite + Tailwind + Zustand + HLS.js // // The binary embeds the SPA build artifacts at /app/web/dist and serves them @@ -72,6 +72,9 @@ func main() { if err := database.AutoMigrate(db); err != nil { logger.Fatal("auto-migrate failed", zap.Error(err)) } + if err := database.MigrateSQLiteToCurrentIfNeeded(cfg, db, logger); err != nil { + logger.Fatal("sqlite to postgres migration failed", zap.Error(err)) + } repos := repository.New(db) service.ApplyRuntimeSettings(context.Background(), cfg, repos, logger) diff --git a/config.example.yaml b/config.example.yaml index 804cc0b..d077cc9 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -26,6 +26,10 @@ app: server_url: "" # public URL of the server (used for DLNA / casting) database: + # auto: Docker Compose 注入 PostgreSQL DSN 时用 PG；没有 DSN 时回退 SQLite。 + # postgres: 强制使用 dsn；sqlite: 强制使用 db_path。 + type: auto + dsn: "" # e.g. postgres://mediastation:mediastation@postgres:5432/mediastation?sslmode=disable db_path: ./data/mediastation.db wal_mode: true busy_timeout: 5000 @@ -55,11 +59,21 @@ logging: cache: cache_dir: ./cache + redis_url: "" # e.g. redis://redis:6379/0；为空则只使用进程内/磁盘缓存 + redis_prefix: mediastationgo + media_ttl_seconds: 15 max_disk_usage_mb: 0 # 0 = unlimited ttl_hours: 0 auto_cleanup: false cleanup_interval_min: 60 +search: + backend: "" # "" = DB/SQLite FTS；opensearch = OpenSearch 加速搜索 + opensearch_url: "" # e.g. http://opensearch:9200 + index: mediastation_media + username: "" + password: "" + media: movies_dir: "" tv_dir: "" diff --git a/docker-compose.advanced.yml b/docker-compose.advanced.yml index 451b267..3211d78 100644 --- a/docker-compose.advanced.yml +++ b/docker-compose.advanced.yml @@ -63,6 +63,9 @@ services: pull_policy: missing container_name: mediastation-go restart: unless-stopped + depends_on: + postgres: + condition: service_healthy ports: # 宿主机端口:容器端口。默认访问 http://<服务器IP>:18080 @@ -113,6 +116,8 @@ services: # 数据与缓存目录。需与 volumes 中的容器路径一致。 MEDIASTATION_APP_DATA_DIR: /data + MEDIASTATION_DATABASE_TYPE: ${MEDIASTATION_DATABASE_TYPE:-postgres} + MEDIASTATION_DATABASE_DSN: ${MEDIASTATION_DATABASE_DSN:-postgres://mediastation:mediastation@postgres:5432/mediastation?sslmode=disable} MEDIASTATION_DATABASE_DB_PATH: /data/mediastation.db MEDIASTATION_CACHE_CACHE_DIR: /cache @@ -185,3 +190,20 @@ services: timeout: 10s retries: 5 start_period: 30s + + postgres: + image: postgres:16-alpine + container_name: mediastation-postgres + restart: unless-stopped + environment: + POSTGRES_DB: ${MEDIASTATION_POSTGRES_DB:-mediastation} + POSTGRES_USER: ${MEDIASTATION_POSTGRES_USER:-mediastation} + POSTGRES_PASSWORD: ${MEDIASTATION_POSTGRES_PASSWORD:-mediastation} + TZ: ${TZ:-Asia/Shanghai} + volumes: + - ${MEDIASTATION_POSTGRES_DIR:-./postgres}:/var/lib/postgresql/data + healthcheck: + test: ["CMD-SHELL", "pg_isready -U ${MEDIASTATION_POSTGRES_USER:-mediastation} -d ${MEDIASTATION_POSTGRES_DB:-mediastation}"] + interval: 10s + timeout: 5s + retries: 10 diff --git a/docker-compose.search.yml b/docker-compose.search.yml new file mode 100644 index 0000000..4115b6f --- /dev/null +++ b/docker-compose.search.yml @@ -0,0 +1,45 @@ +# 搜索增强模式叠加文件：PostgreSQL + Redis + OpenSearch +# +# 使用方式： +# docker compose -f docker-compose.yml -f docker-compose.standard.yml -f docker-compose.search.yml up -d +# +# 适合： +# 超大媒体库、复杂全文搜索、后续需要独立搜索索引的部署。 +# +# 注意： +# OpenSearch 常驻内存明显高于 Redis/PG。低配 NAS 不建议开启。 + +services: + mediastation-go: + depends_on: + opensearch: + condition: service_healthy + environment: + # 预留搜索索引入口；主数据仍以 PostgreSQL 为准。 + MEDIASTATION_SEARCH_BACKEND: opensearch + MEDIASTATION_SEARCH_OPENSEARCH_URL: http://opensearch:9200 + MEDIASTATION_SEARCH_INDEX: mediastation_media + + opensearch: + image: opensearchproject/opensearch:2 + container_name: mediastation-opensearch + restart: unless-stopped + environment: + discovery.type: single-node + plugins.security.disabled: "true" + OPENSEARCH_JAVA_OPTS: "-Xms512m -Xmx512m" + DISABLE_INSTALL_DEMO_CONFIG: "true" + bootstrap.memory_lock: "false" + volumes: + - ./opensearch:/usr/share/opensearch/data + healthcheck: + test: ["CMD-SHELL", "curl -fsS http://127.0.0.1:9200 >/dev/null || exit 1"] + interval: 20s + timeout: 10s + retries: 15 + start_period: 60s + logging: + driver: json-file + options: + max-size: "10m" + max-file: "3" diff --git a/docker-compose.standard.yml b/docker-compose.standard.yml new file mode 100644 index 0000000..40bac01 --- /dev/null +++ b/docker-compose.standard.yml @@ -0,0 +1,41 @@ +# 标准模式叠加文件：PostgreSQL + Redis +# +# 使用方式： +# docker compose -f docker-compose.yml -f docker-compose.standard.yml up -d +# +# 适合： +# 多用户、第三方 Emby 客户端频繁刷新、媒体列表/首页访问较多的 NAS。 + +services: + mediastation-go: + depends_on: + redis: + condition: service_healthy + environment: + # Redis 只做热缓存，源数据仍在 PostgreSQL；Redis 丢失可自动重建。 + MEDIASTATION_CACHE_REDIS_URL: redis://redis:6379/0 + + redis: + image: redis:7-alpine + container_name: mediastation-redis + restart: unless-stopped + command: + - redis-server + - --appendonly + - "yes" + - --maxmemory + - 256mb + - --maxmemory-policy + - allkeys-lru + volumes: + - ./redis:/data + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 10s + timeout: 5s + retries: 10 + logging: + driver: json-file + options: + max-size: "10m" + max-file: "3" diff --git a/docker-compose.yml b/docker-compose.yml index 3e9e2b8..9e3cceb 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -25,6 +25,9 @@ services: container_name: mediastation-go restart: unless-stopped init: true + depends_on: + postgres: + condition: service_healthy # 浏览器访问端口。 # 如果 18080 被占用，可以改成 "19011:8080" 之类。 @@ -74,6 +77,13 @@ services: MEDIASTATION_APP_PORT: 8080 MEDIASTATION_APP_WEB_DIR: /app/web/dist MEDIASTATION_APP_DATA_DIR: /data + + # 轻量模式默认只使用 PostgreSQL，适合大多数 NAS。 + # 旧版 ./data/mediastation.db 存在时，首次启动会自动迁移到 PostgreSQL。 + MEDIASTATION_DATABASE_TYPE: postgres + MEDIASTATION_DATABASE_DSN: postgres://mediastation:mediastation@postgres:5432/mediastation?sslmode=disable + + # SQLite 旧库位置：仅用于旧数据自动迁移和无 PG 回退。 MEDIASTATION_DATABASE_DB_PATH: /data/mediastation.db MEDIASTATION_CACHE_CACHE_DIR: /cache @@ -109,3 +119,26 @@ services: options: max-size: "10m" max-file: "3" + + postgres: + image: postgres:16-alpine + container_name: mediastation-postgres + restart: unless-stopped + environment: + POSTGRES_DB: mediastation + POSTGRES_USER: mediastation + POSTGRES_PASSWORD: mediastation + TZ: Asia/Shanghai + volumes: + # PostgreSQL 主数据目录。升级/重建容器时必须保留。 + - ./postgres:/var/lib/postgresql/data + healthcheck: + test: ["CMD-SHELL", "pg_isready -U mediastation -d mediastation"] + interval: 10s + timeout: 5s + retries: 10 + logging: + driver: json-file + options: + max-size: "10m" + max-file: "3" diff --git a/go.mod b/go.mod index 822f296..73ec6cd 100644 --- a/go.mod +++ b/go.mod @@ -9,16 +9,20 @@ require ( github.com/golang-jwt/jwt/v5 v5.2.0 github.com/google/uuid v1.6.0 github.com/gorilla/websocket v1.5.3 + github.com/redis/go-redis/v9 v9.7.0 github.com/shirou/gopsutil/v3 v3.24.5 github.com/spf13/viper v1.18.2 go.uber.org/zap v1.27.0 golang.org/x/crypto v0.21.0 + gorm.io/driver/postgres v1.5.7 gorm.io/gorm v1.25.7 ) require ( github.com/bytedance/sonic v1.9.1 // indirect + github.com/cespare/xxhash/v2 v2.2.0 // indirect github.com/chenzhuoyu/base64x v0.0.0-20221115062448-fe3a3abad311 // indirect + github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect github.com/dustin/go-humanize v1.0.1 // indirect github.com/gabriel-vasile/mimetype v1.4.2 // indirect github.com/gin-contrib/sse v0.1.0 // indirect @@ -29,6 +33,9 @@ require ( github.com/go-playground/validator/v10 v10.14.0 // indirect github.com/goccy/go-json v0.10.2 // indirect github.com/hashicorp/hcl v1.0.0 // indirect + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect + github.com/jackc/pgx/v5 v5.4.3 // indirect github.com/jinzhu/inflection v1.0.0 // indirect github.com/jinzhu/now v1.1.5 // indirect github.com/json-iterator/go v1.1.12 // indirect diff --git a/go.sum b/go.sum index 11f96cf..2ca971b 100644 --- a/go.sum +++ b/go.sum @@ -1,6 +1,8 @@ github.com/bytedance/sonic v1.5.0/go.mod h1:ED5hyg4y6t3/9Ku1R6dU/4KyJ48DZ4jPhfY1O2AihPM= github.com/bytedance/sonic v1.9.1 h1:6iJ6NqdoxCDr6mbY8h18oSO+cShGSMRGCEo7F2h0x8s= github.com/bytedance/sonic v1.9.1/go.mod h1:i736AoUSYt75HyZLoJW9ERYxcy6eaN6h4BZXU064P/U= +github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44= +github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/chenzhuoyu/base64x v0.0.0-20211019084208-fb5309c8db06/go.mod h1:DH46F32mSOjUmXrMHnKwZdA8wcEefY7UVqBKYGjpdQY= github.com/chenzhuoyu/base64x v0.0.0-20221115062448-fe3a3abad311 h1:qSGYFH7+jGhDF8vLC+iwCD4WpbV1EBDSzWkJODFLams= github.com/chenzhuoyu/base64x v0.0.0-20221115062448-fe3a3abad311/go.mod h1:b583jCggY9gE99b6G5LEC39OIiVsWj+R97kbl5odCEk= @@ -8,6 +10,8 @@ github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78= +github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc= github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8= @@ -52,6 +56,12 @@ github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aN github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4= github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= +github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a h1:bbPeKD0xmW/Y25WS6cokEszi5g+S0QxI/d45PkRi7Nk= +github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.4.3 h1:cxFyXhxlvAifxnkKKdlxv8XqUf59tDlYjnV5YYfsJJY= +github.com/jackc/pgx/v5 v5.4.3/go.mod h1:Ig06C2Vu0t5qXC60W8sqIthScaEnFvojjj9dSljmHRA= github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E= github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc= github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ= @@ -87,6 +97,8 @@ github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRI github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c h1:ncq/mPwQF4JjgDlrVEn3C11VoGHZN7m8qihwgMEtzYw= github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE= +github.com/redis/go-redis/v9 v9.7.0 h1:HhLSs+B6O021gwzl+locl0zEDnyNkxMtf/Z3NNBMa9E= +github.com/redis/go-redis/v9 v9.7.0/go.mod h1:f6zhXITC7JUJIlPEiBOTXxJgPLdZcA93GewI7inzyWw= github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= @@ -173,6 +185,8 @@ gopkg.in/ini.v1 v1.67.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gorm.io/driver/postgres v1.5.7 h1:8ptbNJTDbEmhdr62uReG5BGkdQyeasu/FZHxI0IMGnM= +gorm.io/driver/postgres v1.5.7/go.mod h1:3e019WlBaYI5o5LIdNV+LyxCMNtLOQETBXL2h4chKpA= gorm.io/gorm v1.25.7 h1:VsD6acwRjz2zFxGO50gPO6AkNs7KKnvfzUjHQhZDz/A= gorm.io/gorm v1.25.7/go.mod h1:hbnx/Oo0ChWMn1BIhpy1oYozzpM15i4YPuHDmfYtwg8= modernc.org/libc v1.22.5 h1:91BNch/e5B0uPbJFgqbxXuOnxBQjlS//icfQEGmvyjE= diff --git a/internal/config/config.go b/internal/config/config.go index fa0799a..962232c 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -33,6 +33,7 @@ type Config struct { Secrets SecretsConfig `mapstructure:"secrets"` Logging LoggingConfig `mapstructure:"logging"` Cache CacheConfig `mapstructure:"cache"` + Search SearchConfig `mapstructure:"search"` Media MediaConfig `mapstructure:"media"` Transcoder TranscoderConfig `mapstructure:"transcoder"` AI AIConfig `mapstructure:"ai"` @@ -86,9 +87,12 @@ type AppConfig struct { ServerURL string `mapstructure:"server_url"` } -// DatabaseConfig 配置 GORM + SQLite。 +// DatabaseConfig 配置 GORM 数据库。默认 auto： +// Docker Compose 主线会注入 PostgreSQL DSN；裸机/旧部署没有 DSN 时回退 SQLite。 type DatabaseConfig struct { + Type string `mapstructure:"type"` DBPath string `mapstructure:"db_path"` + DSN string `mapstructure:"dsn"` WALMode bool `mapstructure:"wal_mode"` BusyTimeout int `mapstructure:"busy_timeout"` CacheSize int `mapstructure:"cache_size"` @@ -128,6 +132,17 @@ type CacheConfig struct { TTLHours int `mapstructure:"ttl_hours"` AutoCleanup bool `mapstructure:"auto_cleanup"` CleanupIntervalMin int `mapstructure:"cleanup_interval_min"` + RedisURL string `mapstructure:"redis_url"` + RedisPrefix string `mapstructure:"redis_prefix"` + MediaTTLSeconds int `mapstructure:"media_ttl_seconds"` +} + +type SearchConfig struct { + Backend string `mapstructure:"backend"` + OpenSearchURL string `mapstructure:"opensearch_url"` + Index string `mapstructure:"index"` + Username string `mapstructure:"username"` + Password string `mapstructure:"password"` } // MediaConfig 保存默认库位置（用于引导库）。 @@ -229,7 +244,9 @@ func setDefaults(v *viper.Viper) { v.SetDefault("app.cors_origins", []string{}) v.SetDefault("app.server_url", "") + v.SetDefault("database.type", "auto") v.SetDefault("database.db_path", "./data/mediastation.db") + v.SetDefault("database.dsn", "") v.SetDefault("database.wal_mode", true) v.SetDefault("database.busy_timeout", 5000) v.SetDefault("database.cache_size", -20000) @@ -246,6 +263,15 @@ func setDefaults(v *viper.Viper) { v.SetDefault("cache.cache_dir", "./cache") v.SetDefault("cache.cleanup_interval_min", 60) + v.SetDefault("cache.redis_url", "") + v.SetDefault("cache.redis_prefix", "mediastationgo") + v.SetDefault("cache.media_ttl_seconds", 15) + + v.SetDefault("search.backend", "") + v.SetDefault("search.opensearch_url", "") + v.SetDefault("search.index", "mediastation_media") + v.SetDefault("search.username", "") + v.SetDefault("search.password", "") v.SetDefault("ai.enabled", false) v.SetDefault("ai.provider", "openai") @@ -310,6 +336,9 @@ func (c *Config) normalize() error { if c.Database.DBPath == "" { c.Database.DBPath = filepath.Join(c.App.DataDir, "mediastation.db") } + if c.Database.Type == "" { + c.Database.Type = "auto" + } if c.App.MaxCPUThreads < 1 { c.App.MaxCPUThreads = 1 } @@ -328,6 +357,16 @@ func (c *Config) normalize() error { if c.Cache.CacheDir == "" { c.Cache.CacheDir = filepath.Join(c.App.DataDir, "cache") } + if c.Cache.RedisPrefix == "" { + c.Cache.RedisPrefix = "mediastationgo" + } + if c.Cache.MediaTTLSeconds < 1 { + c.Cache.MediaTTLSeconds = 15 + } + c.Search.Backend = strings.ToLower(strings.TrimSpace(c.Search.Backend)) + if c.Search.Index == "" { + c.Search.Index = "mediastation_media" + } if c.Secrets.JWTSecret == "" { // 持久化自动生成的密钥以在操作员忘记配置时保持会话稳定。 path := filepath.Join(c.App.DataDir, ".jwt_secret") diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 8e285eb..6c573ce 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -31,9 +31,21 @@ func TestLoadDefaults(t *testing.T) { if cfg.Database.DBPath == "" { t.Fatalf("expected non-empty DBPath") } + if cfg.Database.Type != "auto" { + t.Fatalf("expected default database type auto, got %q", cfg.Database.Type) + } if cfg.Database.MaxOpenConns != defaultDatabaseMaxOpenConns { t.Fatalf("expected default MaxOpenConns %d, got %d", defaultDatabaseMaxOpenConns, cfg.Database.MaxOpenConns) } + if cfg.Cache.RedisPrefix != "mediastationgo" { + t.Fatalf("expected default redis prefix, got %q", cfg.Cache.RedisPrefix) + } + if cfg.Cache.MediaTTLSeconds != 15 { + t.Fatalf("expected default media cache ttl 15, got %d", cfg.Cache.MediaTTLSeconds) + } + if cfg.Search.Index != "mediastation_media" { + t.Fatalf("expected default search index, got %q", cfg.Search.Index) + } if cfg.Database.MaxIdleConns != defaultDatabaseMaxIdleConns { t.Fatalf("expected default MaxIdleConns %d, got %d", defaultDatabaseMaxIdleConns, cfg.Database.MaxIdleConns) } @@ -66,6 +78,12 @@ func TestEnvOverride(t *testing.T) { } t.Setenv("MEDIASTATION_APP_PORT", "9090") + t.Setenv("MEDIASTATION_DATABASE_TYPE", "postgres") + t.Setenv("MEDIASTATION_DATABASE_DSN", "postgres://msgo:secret@postgres:5432/msgo?sslmode=disable") + t.Setenv("MEDIASTATION_CACHE_REDIS_URL", "redis://redis:6379/0") + t.Setenv("MEDIASTATION_CACHE_MEDIA_TTL_SECONDS", "30") + t.Setenv("MEDIASTATION_SEARCH_BACKEND", "opensearch") + t.Setenv("MEDIASTATION_SEARCH_OPENSEARCH_URL", "http://opensearch:9200") cfg, err := Load() if err != nil { t.Fatalf("Load() error: %v", err) @@ -73,6 +91,15 @@ func TestEnvOverride(t *testing.T) { if cfg.App.Port != 9090 { t.Fatalf("expected port 9090 from env, got %d", cfg.App.Port) } + if cfg.Database.Type != "postgres" || cfg.Database.DSN == "" { + t.Fatalf("expected postgres database config from env, got type=%q dsn=%q", cfg.Database.Type, cfg.Database.DSN) + } + if cfg.Cache.RedisURL != "redis://redis:6379/0" || cfg.Cache.MediaTTLSeconds != 30 { + t.Fatalf("expected redis cache config from env, got url=%q ttl=%d", cfg.Cache.RedisURL, cfg.Cache.MediaTTLSeconds) + } + if cfg.Search.Backend != "opensearch" || cfg.Search.OpenSearchURL != "http://opensearch:9200" { + t.Fatalf("expected opensearch config from env, got backend=%q url=%q", cfg.Search.Backend, cfg.Search.OpenSearchURL) + } } func TestLoadAllowsExplicitSingleConnectionDatabaseConfig(t *testing.T) { diff --git a/internal/database/database.go b/internal/database/database.go index a8a1e8f..7c5e258 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -1,26 +1,32 @@ -// Package database wires up GORM against SQLite (WAL mode) and exposes the -// auto-migration entry point used at startup. +// Package database wires up GORM against the configured database and exposes +// startup migration helpers. package database import ( "context" + "errors" "fmt" + "os" "path/filepath" + "reflect" + "strings" + "time" "github.com/glebarez/sqlite" "go.uber.org/zap" + "gorm.io/driver/postgres" "gorm.io/gorm" + "gorm.io/gorm/clause" "gorm.io/gorm/logger" "github.com/ShukeBta/MediaStationGo/internal/config" "github.com/ShukeBta/MediaStationGo/internal/model" ) -// Open initialises the SQLite database file applying WAL pragmas for -// better concurrent read performance — same defaults as nowen-video. +// Open initialises the configured GORM database. database.type=auto chooses +// PostgreSQL when database.dsn is present (the Docker Compose default) and +// otherwise falls back to SQLite for old/bare-metal installs. func Open(cfg *config.Config, log *zap.Logger) (*gorm.DB, error) { - dsn := buildDSN(cfg) - gormLogger := logger.New( zapStdLogger{log: log}, logger.Config{ @@ -31,7 +37,15 @@ func Open(cfg *config.Config, log *zap.Logger) (*gorm.DB, error) { }, ) - db, err := gorm.Open(sqlite.Open(dsn), &gorm.Config{ + dialect := normalizeDatabaseType(cfg.Database.Type) + if dialect == "auto" { + dialect = effectiveAutoDatabaseType(cfg) + } + dialector, err := databaseDialector(cfg, dialect) + if err != nil { + return nil, err + } + db, err := gorm.Open(dialector, &gorm.Config{ Logger: gormLogger, PrepareStmt: true, DisableForeignKeyConstraintWhenMigrating: false, @@ -39,7 +53,9 @@ func Open(cfg *config.Config, log *zap.Logger) (*gorm.DB, error) { if err != nil { return nil, fmt.Errorf("gorm open: %w", err) } - installSQLiteWriteGate(db) + if dialect == "sqlite" { + installSQLiteWriteGate(db) + } sqlDB, err := db.DB() if err != nil { return nil, fmt.Errorf("gorm sqldb: %w", err) @@ -53,6 +69,185 @@ func Open(cfg *config.Config, log *zap.Logger) (*gorm.DB, error) { return db, nil } +func normalizeDatabaseType(value string) string { + switch strings.ToLower(strings.TrimSpace(value)) { + case "", "auto": + return "auto" + case "sqlite", "sqlite3": + return "sqlite" + case "postgres", "postgresql", "pg": + return "postgres" + default: + return strings.ToLower(strings.TrimSpace(value)) + } +} + +func effectiveAutoDatabaseType(cfg *config.Config) string { + if cfg != nil && strings.TrimSpace(cfg.Database.DSN) != "" { + return "postgres" + } + return "sqlite" +} + +func databaseDialector(cfg *config.Config, dialect string) (gorm.Dialector, error) { + switch dialect { + case "sqlite": + return sqlite.Open(buildSQLiteDSN(cfg)), nil + case "postgres": + dsn := strings.TrimSpace(cfg.Database.DSN) + if dsn == "" { + return nil, fmt.Errorf("database.dsn is required when database.type=postgres") + } + return postgres.Open(dsn), nil + default: + return nil, fmt.Errorf("unsupported database.type %q (supported: sqlite, postgres)", cfg.Database.Type) + } +} + +// MigrateSQLiteToCurrentIfNeeded copies an existing SQLite database into a new +// PostgreSQL database once. It is intentionally conservative: it only runs when +// the current DB is PostgreSQL, the configured SQLite file exists, and the +// target business tables are empty. Redis is not migrated because it is a +// rebuildable cache, not a source of truth. +func MigrateSQLiteToCurrentIfNeeded(cfg *config.Config, target *gorm.DB, log *zap.Logger) error { + if cfg == nil || target == nil || target.Dialector == nil || target.Dialector.Name() != "postgres" { + return nil + } + sqlitePath := strings.TrimSpace(cfg.Database.DBPath) + if sqlitePath == "" { + return nil + } + if _, err := os.Stat(sqlitePath); err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil + } + return fmt.Errorf("stat sqlite migration source: %w", err) + } + if empty, err := targetBusinessTablesEmpty(target); err != nil { + return err + } else if !empty { + if log != nil { + log.Info("skip sqlite to postgres migration: target database already has data") + } + return nil + } + + srcCfg := *cfg + srcCfg.Database.Type = "sqlite" + src, err := gorm.Open(sqlite.Open(buildSQLiteDSN(&srcCfg)), &gorm.Config{ + Logger: logger.Default.LogMode(logger.Silent), + }) + if err != nil { + return fmt.Errorf("open sqlite migration source: %w", err) + } + sqlDB, err := src.DB() + if err == nil { + defer sqlDB.Close() + } + + started := time.Now() + copied, err := copyModelTables(src, target, 500) + if err != nil { + return err + } + if copied > 0 && log != nil { + log.Info("sqlite data migrated to postgres", + zap.String("source", sqlitePath), + zap.Int64("rows", copied), + zap.Duration("duration", time.Since(started))) + } + return nil +} + +func targetBusinessTablesEmpty(db *gorm.DB) (bool, error) { + for _, m := range model.AllModels() { + table, err := modelTableName(db, m) + if err != nil { + return false, err + } + var count int64 + if err := db.Raw("SELECT COUNT(1) FROM " + quoteIdent(table)).Scan(&count).Error; err != nil { + return false, fmt.Errorf("count target table %s: %w", table, err) + } + if count > 0 { + return false, nil + } + } + return true, nil +} + +func copyModelTables(src, target *gorm.DB, batchSize int) (int64, error) { + if batchSize <= 0 { + batchSize = 500 + } + var copied int64 + for _, m := range model.AllModels() { + table, err := modelTableName(src, m) + if err != nil { + return copied, err + } + exists, err := sqliteTableExists(src, table) + if err != nil { + return copied, err + } + if !exists { + continue + } + var sourceCount int64 + if err := src.Raw("SELECT COUNT(1) FROM " + quoteIdent(table)).Scan(&sourceCount).Error; err != nil { + return copied, fmt.Errorf("count sqlite table %s: %w", table, err) + } + if sourceCount == 0 { + continue + } + var targetCount int64 + if err := target.Raw("SELECT COUNT(1) FROM " + quoteIdent(table)).Scan(&targetCount).Error; err != nil { + return copied, fmt.Errorf("count target table %s: %w", table, err) + } + if targetCount > 0 { + continue + } + + modelType := reflect.TypeOf(m) + if modelType.Kind() != reflect.Ptr { + return copied, fmt.Errorf("model %T is not a pointer", m) + } + sliceType := reflect.SliceOf(modelType.Elem()) + slicePtr := reflect.New(sliceType) + if err := src.Unscoped().Find(slicePtr.Interface()).Error; err != nil { + return copied, fmt.Errorf("read sqlite table %s: %w", table, err) + } + if slicePtr.Elem().Len() == 0 { + continue + } + if err := target.Clauses(clause.OnConflict{DoNothing: true}).CreateInBatches(slicePtr.Interface(), batchSize).Error; err != nil { + return copied, fmt.Errorf("copy sqlite table %s: %w", table, err) + } + copied += int64(slicePtr.Elem().Len()) + } + return copied, nil +} + +func sqliteTableExists(db *gorm.DB, table string) (bool, error) { + var count int64 + if err := db.Raw(`SELECT COUNT(1) FROM sqlite_master WHERE type = 'table' AND name = ?`, table).Scan(&count).Error; err != nil { + return false, fmt.Errorf("inspect sqlite table %s: %w", table, err) + } + return count > 0, nil +} + +func modelTableName(db *gorm.DB, m any) (string, error) { + stmt := &gorm.Statement{DB: db} + if err := stmt.Parse(m); err != nil { + return "", err + } + return stmt.Schema.Table, nil +} + +func quoteIdent(value string) string { + return `"` + strings.ReplaceAll(value, `"`, `""`) + `"` +} + func installSQLiteWriteGate(db *gorm.DB) { if db == nil { return @@ -122,7 +317,7 @@ func (g *sqliteWriteGate) Unlock() { } } -func buildDSN(cfg *config.Config) string { +func buildSQLiteDSN(cfg *config.Config) string { dbPath := cfg.Database.DBPath if !filepath.IsAbs(dbPath) { // keep as-is to respect user-provided relative paths. @@ -152,7 +347,10 @@ func AutoMigrate(db *gorm.DB) error { if err := ensurePerformanceIndexes(db); err != nil { return err } - return ensureMediaSearchIndex(db) + if isSQLite(db) { + return ensureMediaSearchIndex(db) + } + return nil } func ensurePerformanceIndexes(db *gorm.DB) error { @@ -160,13 +358,22 @@ func ensurePerformanceIndexes(db *gorm.DB) error { `CREATE INDEX IF NOT EXISTS idx_media_library_created_active ON media(library_id, created_at DESC) WHERE deleted_at IS NULL`, `CREATE INDEX IF NOT EXISTS idx_media_library_episode_active ON media(library_id, season_num, episode_num, created_at DESC) WHERE deleted_at IS NULL`, `CREATE INDEX IF NOT EXISTS idx_media_series_active ON media(series_id, season_num, episode_num) WHERE deleted_at IS NULL`, - `CREATE INDEX IF NOT EXISTS idx_media_title_active ON media(title COLLATE NOCASE) WHERE deleted_at IS NULL`, - `CREATE INDEX IF NOT EXISTS idx_media_original_name_active ON media(original_name COLLATE NOCASE) WHERE deleted_at IS NULL`, `CREATE INDEX IF NOT EXISTS idx_favorites_user_media_active ON favorites(user_id, media_id) WHERE deleted_at IS NULL`, `CREATE INDEX IF NOT EXISTS idx_playback_histories_user_media_active ON playback_histories(user_id, media_id, watched_at DESC) WHERE deleted_at IS NULL`, `CREATE INDEX IF NOT EXISTS idx_playback_histories_resume_active ON playback_histories(user_id, completed, watched_at DESC) WHERE deleted_at IS NULL`, `CREATE INDEX IF NOT EXISTS idx_play_profiles_user_created_active ON play_profiles(user_id, created_at DESC) WHERE deleted_at IS NULL`, } + if isSQLite(db) { + statements = append(statements, + `CREATE INDEX IF NOT EXISTS idx_media_title_active ON media(title COLLATE NOCASE) WHERE deleted_at IS NULL`, + `CREATE INDEX IF NOT EXISTS idx_media_original_name_active ON media(original_name COLLATE NOCASE) WHERE deleted_at IS NULL`, + ) + } else { + statements = append(statements, + `CREATE INDEX IF NOT EXISTS idx_media_title_active ON media(title) WHERE deleted_at IS NULL`, + `CREATE INDEX IF NOT EXISTS idx_media_original_name_active ON media(original_name) WHERE deleted_at IS NULL`, + ) + } for _, stmt := range statements { if err := db.Exec(stmt).Error; err != nil { return err @@ -175,6 +382,10 @@ func ensurePerformanceIndexes(db *gorm.DB) error { return nil } +func isSQLite(db *gorm.DB) bool { + return db != nil && db.Dialector != nil && db.Dialector.Name() == "sqlite" +} + // mediaSearchIndexSchemaVersion 标识 FTS 索引的物理布局版本。 // v2：FTS 行的 rowid 与 media.rowid 对齐，并由触发器实时维护。 const mediaSearchIndexSchemaVersion = 2 @@ -262,7 +473,7 @@ WHERE deleted_at IS NULL ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY created_at ASC, id ASC) AS rn FROM telegram_bindings WHERE deleted_at IS NULL - ) + ) AS ranked_bindings WHERE rn = 1 ) `).Error; err != nil { diff --git a/internal/database/database_test.go b/internal/database/database_test.go index 96e380b..cf2af81 100644 --- a/internal/database/database_test.go +++ b/internal/database/database_test.go @@ -73,3 +73,45 @@ func TestEnsurePerformanceIndexesCreatesHotPathIndexes(t *testing.T) { } } } + +func TestCopyModelTablesMigratesExistingSQLiteRows(t *testing.T) { + src, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + dst, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + for _, db := range []*gorm.DB{src, dst} { + if err := db.AutoMigrate(&model.User{}, &model.Library{}, &model.Setting{}); err != nil { + t.Fatal(err) + } + } + user := model.User{Username: "admin", PasswordHash: "hash", Role: "admin", IsActive: true} + if err := src.Create(&user).Error; err != nil { + t.Fatal(err) + } + lib := model.Library{Name: "Movies", Path: "/media/movies", Type: "movie", Enabled: true} + if err := src.Create(&lib).Error; err != nil { + t.Fatal(err) + } + if err := src.Create(&model.Setting{Key: "organize.auto", Value: "false"}).Error; err != nil { + t.Fatal(err) + } + + copied, err := copyModelTables(src, dst, 2) + if err != nil { + t.Fatal(err) + } + if copied != 3 { + t.Fatalf("copied rows = %d, want 3", copied) + } + var got model.User + if err := dst.First(&got, "username = ?", "admin").Error; err != nil { + t.Fatal(err) + } + if got.ID != user.ID || got.Role != "admin" { + t.Fatalf("user not preserved: %#v", got) + } +} diff --git a/internal/handler/media.go b/internal/handler/media.go index debcde6..50c2eaf 100644 --- a/internal/handler/media.go +++ b/internal/handler/media.go @@ -6,6 +6,7 @@ import ( "errors" "net/http" "strconv" + "strings" "github.com/gin-gonic/gin" @@ -130,7 +131,7 @@ func scanLibraryHandler(svc *service.Container) gin.HandlerFunc { c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } - finishHTTPTask(task, nil, "completed", "手动扫描入库结束", scanTaskMetrics(res), nil) + finishHTTPTask(task, nil, "completed", "手动扫描入库结束", scanTaskMetrics(res), scanTaskDetails(res, 20)) c.JSON(http.StatusOK, res) } } @@ -161,15 +162,49 @@ func scanTaskMetrics(res *service.ScanResult) map[string]int64 { "probed": int64(res.Probed), "local_metadata": int64(res.LocalMetadata), "removed": res.Removed, + "errors": int64(res.ErrorCount), } } +func scanTaskDetails(res *service.ScanResult, limit int) []string { + if res == nil || limit <= 0 { + return nil + } + out := make([]string, 0, limit) + for _, line := range res.Errors { + line = strings.TrimSpace(line) + if line == "" { + continue + } + out = append(out, "错误: "+line) + if len(out) >= limit { + return out + } + } + return out +} + func listMediaHandler(svc *service.Container) gin.HandlerFunc { return func(c *gin.Context) { id := c.Param("id") page, _ := strconv.Atoi(c.DefaultQuery("page", "1")) size, _ := strconv.Atoi(c.DefaultQuery("page_size", "50")) - items, total, err := svc.Media.ListMediaVisible(c.Request.Context(), id, page, size, mediaVisibilityForRequest(c, svc)) + groupVersions := c.DefaultQuery("group_versions", "1") != "0" + if !groupVersions { + items, total, err := svc.Media.ListMediaVisible(c.Request.Context(), id, page, size, mediaVisibilityForRequest(c, svc)) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{ + "items": items, + "total": total, + "page": page, + "page_size": size, + }) + return + } + items, total, err := svc.Media.ListMediaVisibleGrouped(c.Request.Context(), id, page, size, mediaVisibilityForRequest(c, svc)) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return @@ -205,10 +240,25 @@ func getMediaHandler(svc *service.Container) gin.HandlerFunc { func searchMediaHandler(svc *service.Container) gin.HandlerFunc { return func(c *gin.Context) { q := c.Query("q") + groupVersions := c.DefaultQuery("group_versions", "1") != "0" if c.Query("page") != "" || c.Query("page_size") != "" { page, _ := strconv.Atoi(c.DefaultQuery("page", "1")) size, _ := strconv.Atoi(c.DefaultQuery("page_size", "50")) - items, total, err := svc.Media.SearchMediaVisiblePage(c.Request.Context(), q, page, size, mediaVisibilityForRequest(c, svc)) + if !groupVersions { + items, total, err := svc.Media.SearchMediaVisiblePage(c.Request.Context(), q, page, size, mediaVisibilityForRequest(c, svc)) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{ + "items": items, + "total": total, + "page": page, + "page_size": size, + }) + return + } + items, total, err := svc.Media.SearchMediaVisiblePageGrouped(c.Request.Context(), q, page, size, mediaVisibilityForRequest(c, svc)) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return @@ -222,7 +272,16 @@ func searchMediaHandler(svc *service.Container) gin.HandlerFunc { return } limit, _ := strconv.Atoi(c.DefaultQuery("limit", "50")) - items, err := svc.Media.SearchMediaVisible(c.Request.Context(), q, limit, mediaVisibilityForRequest(c, svc)) + if !groupVersions { + items, err := svc.Media.SearchMediaVisible(c.Request.Context(), q, limit, mediaVisibilityForRequest(c, svc)) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + c.JSON(http.StatusOK, gin.H{"items": items}) + return + } + items, err := svc.Media.SearchMediaVisibleGrouped(c.Request.Context(), q, limit, mediaVisibilityForRequest(c, svc)) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return diff --git a/internal/handler/media_test.go b/internal/handler/media_test.go index 606fb5d..c76c0c2 100644 --- a/internal/handler/media_test.go +++ b/internal/handler/media_test.go @@ -5,6 +5,7 @@ import ( "net/http" "net/http/httptest" "testing" + "time" "github.com/gin-gonic/gin" "github.com/glebarez/sqlite" @@ -90,6 +91,66 @@ func TestListLibrariesIncludeHiddenNormalizesCloudDisplayNames(t *testing.T) { } } +func TestListMediaGroupsMultipleVersionsByDefault(t *testing.T) { + gin.SetMode(gin.TestMode) + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if err := db.AutoMigrate(&model.User{}, &model.Library{}, &model.Media{}, &model.Setting{}, &model.PlayProfile{}); err != nil { + t.Fatal(err) + } + repos := repository.New(db) + lib := model.Library{Name: "Movies", Path: "/media/movies", Type: "movie", Enabled: true} + if err := repos.Library.Create(t.Context(), &lib); err != nil { + t.Fatal(err) + } + if err := repos.DB.Create(&[]model.Media{ + { + Base: model.Base{ID: "movie-1080", CreatedAt: time.Now().Add(-time.Minute)}, + LibraryID: lib.ID, + Title: "流浪地球", + Path: "/media/movies/The.Wandering.Earth.2019.1080p.mkv", + Year: 2019, + Width: 1920, + Height: 1080, + SizeBytes: 100, + }, + { + Base: model.Base{ID: "movie-2160", CreatedAt: time.Now()}, + LibraryID: lib.ID, + Title: "流浪地球", + Path: "cloud://openlist/Movies/The.Wandering.Earth.2019.2160p.mkv", + Year: 2019, + Width: 3840, + Height: 2160, + SizeBytes: 200, + }, + }).Error; err != nil { + t.Fatal(err) + } + svc := &service.Container{ + Repo: repos, + Media: service.NewMediaService(&config.Config{}, zap.NewNop(), repos), + } + + grouped := requestMediaList(t, svc, "/api/libraries/"+lib.ID+"/media", lib.ID) + if grouped.Total != 1 || len(grouped.Items) != 1 { + t.Fatalf("grouped response total=%d len=%d body=%#v", grouped.Total, len(grouped.Items), grouped) + } + if grouped.Items[0].ID != "movie-2160" { + t.Fatalf("primary id = %q, want highest quality version", grouped.Items[0].ID) + } + if len(grouped.Items[0].Versions) != 2 { + t.Fatalf("versions = %#v, want both versions", grouped.Items[0].Versions) + } + + raw := requestMediaList(t, svc, "/api/libraries/"+lib.ID+"/media?group_versions=0", lib.ID) + if raw.Total != 2 || len(raw.Items) != 2 { + t.Fatalf("raw response total=%d len=%d body=%#v", raw.Total, len(raw.Items), raw) + } +} + func requestLibraries(t *testing.T, svc *service.Container, userID, role, path string) []model.Library { t.Helper() w := httptest.NewRecorder() @@ -107,3 +168,27 @@ func requestLibraries(t *testing.T, svc *service.Container, userID, role, path s } return libs } + +type mediaListResponse struct { + Items []service.MediaItem `json:"items"` + Total int64 `json:"total"` +} + +func requestMediaList(t *testing.T, svc *service.Container, path, libraryID string) mediaListResponse { + t.Helper() + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + c.Set(middleware.CtxUserID, "user-1") + c.Set(middleware.CtxUserRole, "user") + c.Params = gin.Params{{Key: "id", Value: libraryID}} + c.Request = httptest.NewRequest(http.MethodGet, path, nil) + listMediaHandler(svc)(c) + if w.Code != http.StatusOK { + t.Fatalf("GET %s status = %d body=%s", path, w.Code, w.Body.String()) + } + var payload mediaListResponse + if err := json.Unmarshal(w.Body.Bytes(), &payload); err != nil { + t.Fatalf("decode media list: %v", err) + } + return payload +} diff --git a/internal/handler/stats.go b/internal/handler/stats.go index 0d0653c..932fd80 100644 --- a/internal/handler/stats.go +++ b/internal/handler/stats.go @@ -2,7 +2,12 @@ package handler import ( + "crypto/sha1" + "encoding/hex" "net/http" + "sort" + "strings" + "time" "github.com/gin-gonic/gin" "gorm.io/gorm" @@ -45,6 +50,18 @@ func applyStatsVisibility(c *gin.Context, svc *service.Container, snap *service. } } snap.Libraries = visibleLibraries + cacheKey := visibleStatsCacheKey(visibility, activeLibraryIDs) + if svc.Cache != nil { + var cached visibleStatsCacheValue + if svc.Cache.GetJSON(c.Request.Context(), cacheKey, &cached) { + snap.Libraries = cached.Libraries + snap.MediaCount = cached.MediaCount + snap.TotalSizeBytes = cached.TotalSizeBytes + snap.TotalSeconds = cached.TotalSeconds + snap.RecentlyAdded = cached.RecentlyAdded + return nil + } + } q := applyMediaVisibilityQuery(svc.Repo.DB.WithContext(c.Request.Context()).Model(&model.Media{}), visibility) q = applyActiveLibraryQuery(q, activeLibraryIDs) @@ -72,9 +89,50 @@ func applyStatsVisibility(c *gin.Context, svc *service.Container, snap *service. return err } snap.RecentlyAdded = recent + if svc.Cache != nil { + svc.Cache.SetJSON(c.Request.Context(), cacheKey, visibleStatsCacheValue{ + Libraries: snap.Libraries, + MediaCount: snap.MediaCount, + TotalSizeBytes: snap.TotalSizeBytes, + TotalSeconds: snap.TotalSeconds, + RecentlyAdded: snap.RecentlyAdded, + }, 10*time.Second) + } return nil } +type visibleStatsCacheValue struct { + Libraries int64 `json:"libraries"` + MediaCount int64 `json:"media_count"` + TotalSizeBytes int64 `json:"total_size_bytes"` + TotalSeconds int64 `json:"total_seconds"` + RecentlyAdded []model.Media `json:"recently_added"` +} + +func visibleStatsCacheKey(visibility service.MediaVisibility, activeLibraryIDs []string) string { + allowed := append([]string(nil), visibility.AllowedLibraryIDs...) + hidden := append([]string(nil), visibility.HiddenLibraryIDs...) + active := append([]string(nil), activeLibraryIDs...) + sort.Strings(allowed) + sort.Strings(hidden) + sort.Strings(active) + sum := sha1.Sum([]byte(strings.Join([]string{ + "visible", + strings.Join(active, ","), + strings.Join(allowed, ","), + strings.Join(hidden, ","), + boolString(visibility.IncludeNSFW), + }, "|"))) + return "stats:visible:" + hex.EncodeToString(sum[:]) +} + +func boolString(value bool) string { + if value { + return "1" + } + return "0" +} + func applyMediaVisibilityQuery(q *gorm.DB, visibility service.MediaVisibility) *gorm.DB { if !visibility.IncludeNSFW { q = q.Where("nsfw = ?", false) diff --git a/internal/handler/stats_extra.go b/internal/handler/stats_extra.go index 4bd9894..788d09c 100644 --- a/internal/handler/stats_extra.go +++ b/internal/handler/stats_extra.go @@ -144,13 +144,17 @@ func statsLibrariesHandler(svc *service.Container) gin.HandlerFunc { if !service.LibraryVisibleForUser(c.Request.Context(), svc.Repo, l, visibility) { continue } + libraryIDs, err := service.MergedLibraryIDsForLibrary(c.Request.Context(), svc.Repo, l.ID) + if err != nil || len(libraryIDs) == 0 { + libraryIDs = []string{l.ID} + } var count int64 var size int64 _ = applyMediaVisibilityQuery(svc.Repo.DB.Model(&model.Media{}), visibility). - Where("library_id = ?", l.ID). + Where("library_id IN ?", libraryIDs). Count(&count).Error _ = applyMediaVisibilityQuery(svc.Repo.DB.Model(&model.Media{}), visibility). - Where("library_id = ?", l.ID). + Where("library_id IN ?", libraryIDs). Select("COALESCE(SUM(size_bytes),0)").Row().Scan(&size) out = append(out, gin.H{ "library": l, diff --git a/internal/handler/stats_test.go b/internal/handler/stats_test.go index ec57653..35107d1 100644 --- a/internal/handler/stats_test.go +++ b/internal/handler/stats_test.go @@ -1,6 +1,8 @@ package handler import ( + "encoding/json" + "net/http" "net/http/httptest" "testing" @@ -62,3 +64,52 @@ func TestStatsSnapshotHidesAdultRecentlyAddedForUser(t *testing.T) { t.Fatalf("recently added should hide adult library, got %#v", snap.RecentlyAdded) } } + +func TestStatsLibrariesCountsMergedCloudLibraryItems(t *testing.T) { + gin.SetMode(gin.TestMode) + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if err := db.AutoMigrate(&model.User{}, &model.Library{}, &model.Media{}, &model.Setting{}, &model.PlayProfile{}); err != nil { + t.Fatal(err) + } + repos := repository.New(db) + local := model.Library{Name: "国产电影", Path: "/media/国产电影", Type: "movie", Enabled: true} + cloud := model.Library{Name: "OpenList · 国产电影", Path: service.BuildCloudLibraryPath("openlist", "/国产电影", "/国产电影"), Type: "movie", Enabled: true} + for _, lib := range []*model.Library{&local, &cloud} { + if err := repos.Library.Create(t.Context(), lib); err != nil { + t.Fatal(err) + } + } + if err := db.Create(&[]model.Media{ + {LibraryID: local.ID, Title: "本地版本", Path: "/media/国产电影/local.mkv", SizeBytes: 100}, + {LibraryID: cloud.ID, Title: "云盘版本", Path: "cloud://openlist/国产电影/cloud.mkv", SizeBytes: 200}, + }).Error; err != nil { + t.Fatal(err) + } + svc := &service.Container{Repo: repos} + w := httptest.NewRecorder() + c, _ := gin.CreateTestContext(w) + c.Request = httptest.NewRequest("GET", "/api/stats/libraries", nil) + + statsLibrariesHandler(svc)(c) + if w.Code != http.StatusOK { + t.Fatalf("status = %d body=%s", w.Code, w.Body.String()) + } + var payload struct { + Libraries []struct { + ItemCount int64 `json:"item_count"` + TotalSize int64 `json:"total_size"` + } `json:"libraries"` + } + if err := json.Unmarshal(w.Body.Bytes(), &payload); err != nil { + t.Fatal(err) + } + if len(payload.Libraries) != 1 { + t.Fatalf("libraries = %#v, want one merged display library", payload.Libraries) + } + if payload.Libraries[0].ItemCount != 2 || payload.Libraries[0].TotalSize != 300 { + t.Fatalf("merged stats = %#v, want count=2 size=300", payload.Libraries[0]) + } +} diff --git a/internal/repository/opensearch.go b/internal/repository/opensearch.go new file mode 100644 index 0000000..ed8ffbf --- /dev/null +++ b/internal/repository/opensearch.go @@ -0,0 +1,212 @@ +package repository + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "strings" + "time" + + "github.com/ShukeBta/MediaStationGo/internal/config" + "github.com/ShukeBta/MediaStationGo/internal/model" +) + +type OpenSearchMediaBackend struct { + baseURL string + index string + username string + password string + client *http.Client +} + +func NewOpenSearchMediaBackend(cfg config.SearchConfig) *OpenSearchMediaBackend { + if strings.TrimSpace(cfg.Backend) != "opensearch" || strings.TrimSpace(cfg.OpenSearchURL) == "" { + return nil + } + index := strings.TrimSpace(cfg.Index) + if index == "" { + index = "mediastation_media" + } + return &OpenSearchMediaBackend{ + baseURL: strings.TrimRight(strings.TrimSpace(cfg.OpenSearchURL), "/"), + index: index, + username: strings.TrimSpace(cfg.Username), + password: cfg.Password, + client: &http.Client{Timeout: 4 * time.Second}, + } +} + +func (b *OpenSearchMediaBackend) SearchMediaIDs(ctx context.Context, query string, offset, limit int, filter MediaQueryFilter) ([]string, int64, error) { + if b == nil || b.client == nil || b.baseURL == "" || b.index == "" { + return nil, 0, fmt.Errorf("opensearch backend not configured") + } + if limit <= 0 { + limit = 50 + } + must := []any{ + map[string]any{ + "multi_match": map[string]any{ + "query": query, + "fields": []string{"title^4", "original_name^3", "genres^2", "path"}, + "type": "best_fields", + "operator": "and", + "fuzziness": "AUTO", + }, + }, + } + filters := []any{ + map[string]any{"term": map[string]any{"deleted": false}}, + } + if !filter.IncludeNSFW { + filters = append(filters, map[string]any{"term": map[string]any{"nsfw": false}}) + } + if len(filter.AllowedLibraryIDs) > 0 { + filters = append(filters, map[string]any{"terms": map[string]any{"library_id": filter.AllowedLibraryIDs}}) + } + if len(filter.HiddenLibraryIDs) > 0 { + filters = append(filters, map[string]any{"bool": map[string]any{ + "must_not": []any{map[string]any{"terms": map[string]any{"library_id": filter.HiddenLibraryIDs}}}, + }}) + } + body := map[string]any{ + "from": offset, + "size": limit, + "_source": []string{ + "id", + }, + "query": map[string]any{ + "bool": map[string]any{ + "must": must, + "filter": filters, + }, + }, + } + var resp struct { + Hits struct { + Total any `json:"total"` + Hits []struct { + ID string `json:"_id"` + Source struct { + ID string `json:"id"` + } `json:"_source"` + } `json:"hits"` + } `json:"hits"` + } + if err := b.doJSON(ctx, http.MethodPost, "/"+url.PathEscape(b.index)+"/_search", body, &resp); err != nil { + return nil, 0, err + } + ids := make([]string, 0, len(resp.Hits.Hits)) + for _, hit := range resp.Hits.Hits { + id := strings.TrimSpace(hit.Source.ID) + if id == "" { + id = strings.TrimSpace(hit.ID) + } + if id != "" { + ids = append(ids, id) + } + } + return ids, openSearchTotal(resp.Hits.Total), nil +} + +func (b *OpenSearchMediaBackend) EnsureIndex(ctx context.Context) error { + if err := b.do(ctx, http.MethodHead, "/"+url.PathEscape(b.index), nil, "", nil); err == nil { + return nil + } + mapping := map[string]any{ + "mappings": map[string]any{ + "properties": map[string]any{ + "id": map[string]any{"type": "keyword"}, + "library_id": map[string]any{"type": "keyword"}, + "title": map[string]any{"type": "text"}, + "original_name": map[string]any{"type": "text"}, + "path": map[string]any{"type": "text"}, + "genres": map[string]any{"type": "text"}, + "nsfw": map[string]any{"type": "boolean"}, + "deleted": map[string]any{"type": "boolean"}, + "created_at": map[string]any{"type": "date"}, + }, + }, + } + return b.doJSON(ctx, http.MethodPut, "/"+url.PathEscape(b.index), mapping, nil) +} + +func (b *OpenSearchMediaBackend) IndexMedia(ctx context.Context, rows []model.Media) error { + if len(rows) == 0 { + return nil + } + var bulk bytes.Buffer + enc := json.NewEncoder(&bulk) + for _, row := range rows { + if err := enc.Encode(map[string]any{"index": map[string]any{"_index": b.index, "_id": row.ID}}); err != nil { + return err + } + if err := enc.Encode(map[string]any{ + "id": row.ID, + "library_id": row.LibraryID, + "title": row.Title, + "original_name": row.OriginalName, + "path": row.Path, + "genres": row.Genres, + "nsfw": row.NSFW, + "deleted": row.DeletedAt.Valid, + "created_at": row.CreatedAt, + }); err != nil { + return err + } + } + return b.do(ctx, http.MethodPost, "/_bulk", &bulk, "application/x-ndjson", nil) +} + +func (b *OpenSearchMediaBackend) doJSON(ctx context.Context, method, path string, body any, out any) error { + var reader io.Reader + if body != nil { + raw, err := json.Marshal(body) + if err != nil { + return err + } + reader = bytes.NewReader(raw) + } + return b.do(ctx, method, path, reader, "application/json", out) +} + +func (b *OpenSearchMediaBackend) do(ctx context.Context, method, path string, body io.Reader, contentType string, out any) error { + req, err := http.NewRequestWithContext(ctx, method, b.baseURL+path, body) + if err != nil { + return err + } + if contentType != "" { + req.Header.Set("Content-Type", contentType) + } + if b.username != "" { + req.SetBasicAuth(b.username, b.password) + } + resp, err := b.client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode >= 400 { + return fmt.Errorf("opensearch %s %s returned %d", method, path, resp.StatusCode) + } + if out == nil { + _, _ = io.Copy(io.Discard, resp.Body) + return nil + } + return json.NewDecoder(resp.Body).Decode(out) +} + +func openSearchTotal(value any) int64 { + switch v := value.(type) { + case float64: + return int64(v) + case map[string]any: + if n, ok := v["value"].(float64); ok { + return int64(n) + } + } + return 0 +} diff --git a/internal/repository/opensearch_test.go b/internal/repository/opensearch_test.go new file mode 100644 index 0000000..0db8d55 --- /dev/null +++ b/internal/repository/opensearch_test.go @@ -0,0 +1,57 @@ +package repository + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/ShukeBta/MediaStationGo/internal/config" +) + +func TestOpenSearchMediaBackendSearchesIDs(t *testing.T) { + var gotPath string + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotPath = r.URL.Path + if r.Method != http.MethodPost { + t.Fatalf("method = %s, want POST", r.Method) + } + var body map[string]any + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + t.Fatal(err) + } + if body["from"].(float64) != 5 || body["size"].(float64) != 10 { + t.Fatalf("paging body = %#v", body) + } + _ = json.NewEncoder(w).Encode(map[string]any{ + "hits": map[string]any{ + "total": map[string]any{"value": 2}, + "hits": []any{ + map[string]any{"_id": "m-1", "_source": map[string]any{"id": "m-1"}}, + map[string]any{"_id": "m-2", "_source": map[string]any{"id": "m-2"}}, + }, + }, + }) + })) + defer upstream.Close() + + backend := NewOpenSearchMediaBackend(config.SearchConfig{ + Backend: "opensearch", + OpenSearchURL: upstream.URL, + Index: "media-test", + }) + ids, total, err := backend.SearchMediaIDs(t.Context(), "流浪地球", 5, 10, MediaQueryFilter{ + IncludeNSFW: false, + AllowedLibraryIDs: []string{"lib-1"}, + HiddenLibraryIDs: []string{"adult"}, + }) + if err != nil { + t.Fatal(err) + } + if gotPath != "/media-test/_search" { + t.Fatalf("path = %q", gotPath) + } + if total != 2 || len(ids) != 2 || ids[0] != "m-1" || ids[1] != "m-2" { + t.Fatalf("ids=%#v total=%d", ids, total) + } +} diff --git a/internal/repository/repository.go b/internal/repository/repository.go index 2a0293e..875d664 100644 --- a/internal/repository/repository.go +++ b/internal/repository/repository.go @@ -274,6 +274,23 @@ type MediaRepository struct { searchIndexOnce sync.Once searchIndexAvailable bool + searchBackend MediaSearchBackend +} + +type MediaSearchBackend interface { + SearchMediaIDs(ctx context.Context, query string, offset, limit int, filter MediaQueryFilter) ([]string, int64, error) +} + +type MediaSearchSyncBackend interface { + MediaSearchBackend + EnsureIndex(ctx context.Context) error + IndexMedia(ctx context.Context, rows []model.Media) error +} + +func (r *MediaRepository) SetSearchBackend(backend MediaSearchBackend) { + if r != nil { + r.searchBackend = backend + } } // MediaQueryFilter is applied to user-facing media queries so NSFW items and @@ -323,6 +340,7 @@ func (r *MediaRepository) upsert(ctx context.Context, m *model.Media) error { m.ScrapeStatus = "pending" } if createErr := r.db.WithContext(ctx).Create(m).Error; createErr == nil { + r.indexMediaBestEffort(ctx, *m) return nil } else if retryErr := r.db.WithContext(ctx).Unscoped().Where("path = ?", m.Path).First(&existing).Error; retryErr != nil { return createErr @@ -433,9 +451,20 @@ func (r *MediaRepository) upsert(ctx context.Context, m *model.Media) error { } // 回写 ID / 不可变字段，让 caller 拿到完整的现有行。 *m = existing + if fresh, err := r.FindByID(ctx, existing.ID); err == nil && fresh != nil { + r.indexMediaBestEffort(ctx, *fresh) + } return nil } +func (r *MediaRepository) indexMediaBestEffort(ctx context.Context, media model.Media) { + backend, ok := r.searchBackend.(MediaSearchSyncBackend) + if !ok { + return + } + _ = backend.IndexMedia(ctx, []model.Media{media}) +} + func setIfChanged[T comparable](updates map[string]any, key string, current, next T) { if current != next { updates[key] = next @@ -500,6 +529,11 @@ func (r *MediaRepository) SearchFilteredPage(ctx context.Context, query string, if limit <= 0 { limit = 50 } + if query != "" && r.searchBackend != nil { + if items, total, ok := r.searchFilteredBackend(ctx, query, offset, limit, filter); ok { + return items, total, nil + } + } if query != "" { if items, total, ok := r.searchFilteredFTS(ctx, query, offset, limit, filter); ok { if total > 0 { @@ -510,6 +544,36 @@ func (r *MediaRepository) SearchFilteredPage(ctx context.Context, query string, return r.searchFilteredLIKE(ctx, query, offset, limit, filter) } +func (r *MediaRepository) searchFilteredBackend(ctx context.Context, query string, offset, limit int, filter MediaQueryFilter) ([]model.Media, int64, bool) { + ids, total, err := r.searchBackend.SearchMediaIDs(ctx, query, offset, limit, filter) + if err != nil { + return nil, 0, false + } + if len(ids) == 0 { + return []model.Media{}, total, true + } + var rows []model.Media + q := r.db.WithContext(ctx).Model(&model.Media{}).Where("id IN ?", ids) + q = applyMediaQueryFilter(q, filter) + if err := q.Find(&rows).Error; err != nil { + return nil, 0, false + } + byID := make(map[string]model.Media, len(rows)) + for _, row := range rows { + byID[row.ID] = row + } + items := make([]model.Media, 0, len(ids)) + for _, id := range ids { + if row, ok := byID[id]; ok { + items = append(items, row) + } + } + if len(items) == 0 && total > 0 { + return nil, 0, false + } + return items, total, true +} + func (r *MediaRepository) searchFilteredFTS(ctx context.Context, query string, offset, limit int, filter MediaQueryFilter) ([]model.Media, int64, bool) { if !r.searchIndexEnabled(ctx) { return nil, 0, false @@ -630,6 +694,9 @@ func escapeLike(value string) string { } func (r *MediaRepository) BackfillSearchIndex(ctx context.Context, batchLimit int) (int64, error) { + if backend, ok := r.searchBackend.(MediaSearchSyncBackend); ok { + return r.backfillExternalSearchIndex(ctx, backend, batchLimit) + } if batchLimit <= 0 { batchLimit = 1000 } @@ -654,10 +721,45 @@ LIMIT ? return res.RowsAffected, res.Error } +func (r *MediaRepository) backfillExternalSearchIndex(ctx context.Context, backend MediaSearchSyncBackend, batchLimit int) (int64, error) { + if batchLimit <= 0 { + batchLimit = 1000 + } + if err := backend.EnsureIndex(ctx); err != nil { + return 0, err + } + var lastID string + for { + var rows []model.Media + q := r.db.WithContext(ctx). + Model(&model.Media{}). + Where("deleted_at IS NULL") + if lastID != "" { + q = q.Where("id > ?", lastID) + } + if err := q.Order("id ASC").Limit(batchLimit).Find(&rows).Error; err != nil { + return 0, err + } + if len(rows) == 0 { + return 0, nil + } + if err := backend.IndexMedia(ctx, rows); err != nil { + return 0, err + } + lastID = rows[len(rows)-1].ID + if len(rows) < batchLimit { + return 0, nil + } + } +} + func (r *MediaRepository) searchIndexEnabled(ctx context.Context) bool { if r == nil || r.db == nil { return false } + if r.db.Dialector == nil || r.db.Dialector.Name() != "sqlite" { + return false + } r.searchIndexOnce.Do(func() { var count int64 err := r.db.WithContext(ctx). diff --git a/internal/repository/repository_test.go b/internal/repository/repository_test.go index 7a2e37c..6e033bd 100644 --- a/internal/repository/repository_test.go +++ b/internal/repository/repository_test.go @@ -1,6 +1,8 @@ package repository import ( + "context" + "errors" "testing" "time" @@ -70,6 +72,58 @@ func TestMediaUpsertSkipsUnchangedExistingRow(t *testing.T) { } } +type fakeMediaSearchBackend struct { + ids []string + err error +} + +func (f fakeMediaSearchBackend) SearchMediaIDs(context.Context, string, int, int, MediaQueryFilter) ([]string, int64, error) { + if f.err != nil { + return nil, 0, f.err + } + return append([]string(nil), f.ids...), int64(len(f.ids)), nil +} + +func TestMediaSearchUsesExternalBackendAndFallsBack(t *testing.T) { + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if err := database.AutoMigrate(db); err != nil { + t.Fatalf("migrate: %v", err) + } + repos := New(db) + lib := model.Library{Name: "Movies", Path: "/media/movie", Type: "movie", Enabled: true} + if err := repos.Library.Create(t.Context(), &lib); err != nil { + t.Fatal(err) + } + for _, row := range []model.Media{ + {Base: model.Base{ID: "m-1"}, LibraryID: lib.ID, Title: "Alpha", Path: "/media/a.mkv"}, + {Base: model.Base{ID: "m-2"}, LibraryID: lib.ID, Title: "Beta", Path: "/media/b.mkv"}, + } { + if err := repos.DB.Create(&row).Error; err != nil { + t.Fatal(err) + } + } + repos.Media.SetSearchBackend(fakeMediaSearchBackend{ids: []string{"m-2", "m-1"}}) + items, total, err := repos.Media.SearchFilteredPage(t.Context(), "anything", 0, 10, MediaQueryFilter{IncludeNSFW: true}) + if err != nil { + t.Fatal(err) + } + if total != 2 || len(items) != 2 || items[0].ID != "m-2" || items[1].ID != "m-1" { + t.Fatalf("external search result total=%d items=%#v", total, items) + } + + repos.Media.SetSearchBackend(fakeMediaSearchBackend{err: errors.New("opensearch down")}) + items, total, err = repos.Media.SearchFilteredPage(t.Context(), "Alpha", 0, 10, MediaQueryFilter{IncludeNSFW: true}) + if err != nil { + t.Fatal(err) + } + if total != 1 || len(items) != 1 || items[0].ID != "m-1" { + t.Fatalf("fallback result total=%d items=%#v", total, items) + } +} + func TestMediaSearchFilteredSupportsChineseFuzzyTerms(t *testing.T) { db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) if err != nil { diff --git a/internal/service/boot_cloud.go b/internal/service/boot_cloud.go index 47e8886..85dcea2 100644 --- a/internal/service/boot_cloud.go +++ b/internal/service/boot_cloud.go @@ -51,7 +51,7 @@ func (c *Container) runBootCloudLibraryScanQueue(cloudLibs []model.Library) { for _, lib := range cloudLibs { libID := lib.ID libName := lib.Name - scanCtx, cancel := context.WithTimeout(context.Background(), 2*time.Hour) + scanCtx, cancel := cloudScanContext(context.Background(), cloudScanTimeout(context.Background(), c.Repo, 24*time.Hour)) c.Log.Info("boot: scanning cloud library", zap.String("id", libID), zap.String("name", libName)) if _, err := c.Scan.ScanLibraryWithoutAutoScrape(scanCtx, libID); err != nil { c.Log.Warn("boot: cloud library scan failed", zap.String("id", libID), zap.String("name", libName), zap.Error(err)) diff --git a/internal/service/emby_compat.go b/internal/service/emby_compat.go index 482fc25..5210927 100644 --- a/internal/service/emby_compat.go +++ b/internal/service/emby_compat.go @@ -61,6 +61,7 @@ type EmbyService struct { repo *repository.Container storage cloudPlaybackResolver probe cloudPlaybackProber + cache *RuntimeCacheService virtualMu sync.RWMutex virtualSeries map[string]embySeriesCacheEntry @@ -87,6 +88,13 @@ func NewEmbyService(cfg *config.Config, log *zap.Logger, repo *repository.Contai return &EmbyService{cfg: cfg, log: log, repo: repo} } +func (e *EmbyService) SetRuntimeCache(cache *RuntimeCacheService) *EmbyService { + if e != nil { + e.cache = cache + } + return e +} + func (e *EmbyService) SetCloudProbe(storage cloudPlaybackResolver, probe cloudPlaybackProber) { if e == nil { return @@ -439,6 +447,11 @@ func (e *EmbyService) Items(ctx context.Context, p ItemsParams) (map[string]any, } func (e *EmbyService) mediaItems(ctx context.Context, p ItemsParams) (map[string]any, error) { + cacheKey := e.embyItemsCacheKey("items", p) + var cached embyItemsCacheValue + if e.cache != nil && e.cache.GetJSON(ctx, cacheKey, &cached) { + return map[string]any{"Items": cached.Items, "TotalRecordCount": cached.TotalRecordCount, "StartIndex": cached.StartIndex}, nil + } q := e.repo.DB.WithContext(ctx).Model(&model.Media{}) q = e.applyUserMediaVisibility(ctx, q, p.UserID) if p.ParentID != "" { @@ -503,7 +516,57 @@ func (e *EmbyService) mediaItems(ctx context.Context, p ItemsParams) (map[string if err != nil { return nil, err } - return map[string]any{"Items": items, "TotalRecordCount": total, "StartIndex": p.StartIndex}, nil + out := map[string]any{"Items": items, "TotalRecordCount": total, "StartIndex": p.StartIndex} + if e.cache != nil { + e.cache.SetJSON(ctx, cacheKey, embyItemsCacheValue{Items: items, TotalRecordCount: total, StartIndex: p.StartIndex}, time.Duration(e.mediaCacheTTLSeconds())*time.Second) + } + return out, nil +} + +type embyItemsCacheValue struct { + Items []map[string]any `json:"items"` + TotalRecordCount int64 `json:"total_record_count"` + StartIndex int `json:"start_index"` +} + +type embyLatestCacheValue struct { + Items []map[string]any `json:"items"` +} + +func (e *EmbyService) embyItemsCacheKey(kind string, p ItemsParams) string { + includeTypes := append([]string(nil), p.IncludeItemTypes...) + filters := append([]string(nil), p.Filters...) + ids := append([]string(nil), p.IDs...) + sort.Strings(includeTypes) + sort.Strings(filters) + sort.Strings(ids) + sum := sha256.Sum256([]byte(strings.Join([]string{ + kind, + p.UserID, + p.ParentID, + strings.Join(ids, ","), + p.SearchTerm, + strings.Join(includeTypes, ","), + strings.Join(filters, ","), + strconv.FormatBool(p.Recursive), + p.SortBy, + p.SortOrder, + strconv.Itoa(p.StartIndex), + strconv.Itoa(p.Limit), + }, "|"))) + return "media:emby:" + hex.EncodeToString(sum[:]) +} + +func (e *EmbyService) embyLatestCacheKey(userID, parentID string, limit int) string { + sum := sha256.Sum256([]byte(strings.Join([]string{"latest", userID, parentID, strconv.Itoa(limit)}, "|"))) + return "media:emby:" + hex.EncodeToString(sum[:]) +} + +func (e *EmbyService) mediaCacheTTLSeconds() int { + if e == nil || e.cfg == nil || e.cfg.Cache.MediaTTLSeconds < 1 { + return 15 + } + return e.cfg.Cache.MediaTTLSeconds } func (e *EmbyService) episodeItems(ctx context.Context, rows []model.Media, p ItemsParams) (map[string]any, error) { @@ -636,13 +699,22 @@ func (e *EmbyService) LatestItems(ctx context.Context, userID, parentID string, if limit <= 0 || limit > 100 { limit = 20 } + cacheKey := e.embyLatestCacheKey(userID, parentID, limit) + var cached embyLatestCacheValue + if e.cache != nil && e.cache.GetJSON(ctx, cacheKey, &cached) { + return cached.Items, nil + } q := e.repo.DB.WithContext(ctx).Model(&model.Media{}).Where("deleted_at IS NULL") q = e.applyUserMediaVisibility(ctx, q, userID) if parentID != "" { if episodic, err := e.libraryIsEpisodic(ctx, parentID); err == nil && episodic { - return e.latestSeriesItemsForLibrary(ctx, userID, parentID, limit) + out, err := e.latestSeriesItemsForLibrary(ctx, userID, parentID, limit) + if err == nil && e.cache != nil { + e.cache.SetJSON(ctx, cacheKey, embyLatestCacheValue{Items: out}, time.Duration(e.mediaCacheTTLSeconds())*time.Second) + } + return out, err } - q = q.Where("library_id = ?", parentID) + q = q.Where("library_id IN ?", e.mergedLibraryIDs(ctx, parentID)) } var rows []model.Media if err := q.Order("media.created_at desc").Limit(limit).Find(&rows).Error; err != nil { @@ -669,6 +741,9 @@ func (e *EmbyService) LatestItems(ctx context.Context, userID, parentID string, for _, m := range rows { out = append(out, e.itemPayload(ctx, &m, favs[m.ID], 0)) } + if e.cache != nil { + e.cache.SetJSON(ctx, cacheKey, embyLatestCacheValue{Items: out}, time.Duration(e.mediaCacheTTLSeconds())*time.Second) + } return out, nil } diff --git a/internal/service/emby_compat_test.go b/internal/service/emby_compat_test.go index 8777670..f00cce1 100644 --- a/internal/service/emby_compat_test.go +++ b/internal/service/emby_compat_test.go @@ -103,6 +103,46 @@ func TestEmbyItemsExposeSeriesSeasonEpisodeHierarchy(t *testing.T) { } } +func TestEmbyLatestItemsIncludesMergedCloudMovieLibrary(t *testing.T) { + svc := newTestEmbyService(t) + local := model.Library{Name: "国产电影", Path: `/media/国产电影`, Type: "movie", Enabled: true} + cloud := model.Library{Name: "OpenList · 国产电影", Path: BuildCloudLibraryPath("openlist", "/国产电影", "/国产电影"), Type: "movie", Enabled: true} + for _, lib := range []*model.Library{&local, &cloud} { + if err := svc.repo.Library.Create(t.Context(), lib); err != nil { + t.Fatalf("create library: %v", err) + } + } + for _, media := range []model.Media{ + { + Base: model.Base{ID: "local-movie", CreatedAt: time.Now().Add(-time.Minute)}, + LibraryID: local.ID, + Title: "本地版本", + Path: `/media/国产电影/local.mkv`, + }, + { + Base: model.Base{ID: "cloud-movie", CreatedAt: time.Now()}, + LibraryID: cloud.ID, + Title: "云盘版本", + Path: `cloud://openlist/国产电影/cloud.mkv`, + }, + } { + if err := svc.repo.DB.Create(&media).Error; err != nil { + t.Fatalf("create media: %v", err) + } + } + + latest, err := svc.LatestItems(t.Context(), "user-1", local.ID, 10) + if err != nil { + t.Fatalf("latest items: %v", err) + } + if len(latest) != 2 { + t.Fatalf("latest items = %#v, want local and merged cloud media", latest) + } + if latest[0]["Id"] != "cloud-movie" || latest[1]["Id"] != "local-movie" { + t.Fatalf("latest order/items = %#v, want cloud then local", latest) + } +} + func TestEmbyVirtualSeriesArtworkUsesListCache(t *testing.T) { svc := newTestEmbyService(t) lib := model.Library{Name: "番剧", Path: `/media/anime`, Type: "anime", Enabled: true} diff --git a/internal/service/media.go b/internal/service/media.go index a0d48ef..ed1ba23 100644 --- a/internal/service/media.go +++ b/internal/service/media.go @@ -3,11 +3,15 @@ package service import ( "context" + "crypto/sha1" + "encoding/hex" "errors" "fmt" "os" "path/filepath" + "sort" "strings" + "time" "go.uber.org/zap" @@ -18,9 +22,10 @@ import ( // MediaService offers high-level CRUD over libraries and media items. type MediaService struct { - cfg *config.Config - log *zap.Logger - repo *repository.Container + cfg *config.Config + log *zap.Logger + repo *repository.Container + cache *RuntimeCacheService } type MediaVisibility struct { @@ -29,6 +34,11 @@ type MediaVisibility struct { HiddenLibraryIDs []string } +type MediaItem struct { + model.Media + Versions []model.Media `json:"versions,omitempty"` +} + const maxMediaSearchLimit = 50000 const maxMediaSearchPageSize = 2000 @@ -60,6 +70,13 @@ func NewMediaService(cfg *config.Config, log *zap.Logger, repo *repository.Conta return &MediaService{cfg: cfg, log: log, repo: repo} } +func (s *MediaService) SetRuntimeCache(cache *RuntimeCacheService) *MediaService { + if s != nil { + s.cache = cache + } + return s +} + // CreateLibrary persists a library after validating that its path exists. func (s *MediaService) CreateLibrary(ctx context.Context, name, path, kind string) (*model.Library, error) { if name == "" || path == "" { @@ -74,6 +91,7 @@ func (s *MediaService) CreateLibrary(ctx context.Context, name, path, kind strin if err := s.repo.Library.Create(ctx, lib); err != nil { return nil, err } + s.invalidateMediaCache(ctx) return lib, nil } @@ -306,13 +324,21 @@ func (s *MediaService) DeleteLibrary(ctx context.Context, id string) error { if err := s.repo.Media.PurgeByLibrary(ctx, id); err != nil { return err } - return s.repo.DB.WithContext(ctx).Unscoped().Where("id = ?", id).Delete(&model.Library{}).Error + err := s.repo.DB.WithContext(ctx).Unscoped().Where("id = ?", id).Delete(&model.Library{}).Error + if err == nil { + s.invalidateMediaCache(ctx) + } + return err } } if err := s.repo.Media.DeleteByLibrary(ctx, id); err != nil { return err } - return s.repo.Library.Delete(ctx, id) + err = s.repo.Library.Delete(ctx, id) + if err == nil { + s.invalidateMediaCache(ctx) + } + return err } // ListMedia paginates media items inside a library. @@ -335,11 +361,194 @@ func (s *MediaService) ListMediaVisible(ctx context.Context, libraryID string, p if err != nil { return nil, 0, err } - return s.repo.Media.ListByLibrariesFiltered(ctx, libraryIDs, (page-1)*pageSize, pageSize, repository.MediaQueryFilter{ + filter := repository.MediaQueryFilter{ IncludeNSFW: visibility.IncludeNSFW, AllowedLibraryIDs: visibility.AllowedLibraryIDs, HiddenLibraryIDs: visibility.HiddenLibraryIDs, + } + cacheKey := s.mediaListCacheKey(libraryID, libraryIDs, page, pageSize, filter) + var cached mediaListCacheValue + if s.cache != nil && s.cache.GetJSON(ctx, cacheKey, &cached) { + return cached.Items, cached.Total, nil + } + items, total, err := s.repo.Media.ListByLibrariesFiltered(ctx, libraryIDs, (page-1)*pageSize, pageSize, filter) + if err != nil { + return nil, 0, err + } + if s.cache != nil { + s.cache.SetJSON(ctx, cacheKey, mediaListCacheValue{Items: items, Total: total}, time.Duration(s.mediaCacheTTLSeconds())*time.Second) + } + return items, total, nil +} + +func (s *MediaService) ListMediaVisibleGrouped(ctx context.Context, libraryID string, page, pageSize int, visibility MediaVisibility) ([]MediaItem, int64, error) { + items, _, err := s.ListMediaVisible(ctx, libraryID, page, pageSize, visibility) + if err != nil { + return nil, 0, err + } + grouped := groupMediaVersions(items) + return grouped, int64(len(grouped)), nil +} + +type mediaListCacheValue struct { + Items []model.Media `json:"items"` + Total int64 `json:"total"` +} + +func (s *MediaService) mediaListCacheKey(libraryID string, libraryIDs []string, page, pageSize int, filter repository.MediaQueryFilter) string { + allowed := append([]string(nil), filter.AllowedLibraryIDs...) + hidden := append([]string(nil), filter.HiddenLibraryIDs...) + libs := append([]string(nil), libraryIDs...) + sort.Strings(allowed) + sort.Strings(hidden) + sort.Strings(libs) + sum := sha1.Sum([]byte(strings.Join([]string{ + libraryID, + strings.Join(libs, ","), + fmt.Sprintf("%d:%d:%t", page, pageSize, filter.IncludeNSFW), + strings.Join(allowed, ","), + strings.Join(hidden, ","), + }, "|"))) + return "media:list:" + hex.EncodeToString(sum[:]) +} + +func (s *MediaService) mediaCacheTTLSeconds() int { + if s == nil || s.cfg == nil || s.cfg.Cache.MediaTTLSeconds < 1 { + return 15 + } + return s.cfg.Cache.MediaTTLSeconds +} + +func (s *MediaService) invalidateMediaCache(ctx context.Context) { + if s != nil && s.cache != nil { + s.cache.DeletePrefix(ctx, "media:") + s.cache.DeletePrefix(ctx, "stats:") + } +} + +func groupMediaVersions(items []model.Media) []MediaItem { + if len(items) == 0 { + return nil + } + type group struct { + key string + primary model.Media + rows []model.Media + } + groups := make([]group, 0, len(items)) + byKey := make(map[string]int, len(items)) + for _, item := range items { + key := mediaVersionGroupKey(item) + if key == "" { + groups = append(groups, group{primary: item, rows: []model.Media{item}}) + continue + } + if idx, ok := byKey[key]; ok { + groups[idx].rows = append(groups[idx].rows, item) + if betterMediaVersion(item, groups[idx].primary) { + groups[idx].primary = item + } + continue + } + byKey[key] = len(groups) + groups = append(groups, group{key: key, primary: item, rows: []model.Media{item}}) + } + out := make([]MediaItem, 0, len(groups)) + for _, g := range groups { + sort.SliceStable(g.rows, func(i, j int) bool { + return betterMediaVersion(g.rows[i], g.rows[j]) + }) + item := MediaItem{Media: g.primary} + if len(g.rows) > 1 { + item.Versions = g.rows + } + out = append(out, item) + } + sort.SliceStable(out, func(i, j int) bool { + return out[i].CreatedAt.After(out[j].CreatedAt) }) + return out +} + +func mediaVersionGroupKey(m model.Media) string { + if m.SeasonNum > 0 || m.EpisodeNum > 0 { + title := firstNonEmpty(m.OriginalName, m.Title) + if title == "" { + title, _ = CleanQuery(m.Path) + } + title = normalizeMediaVersionText(title) + if title == "" { + return "" + } + return strings.Join([]string{ + "episode", + strings.ToLower(strings.TrimSpace(m.LibraryID)), + title, + fmt.Sprintf("%d:%d", m.SeasonNum, m.EpisodeNum), + }, "|") + } + switch { + case m.TMDbID > 0: + return fmt.Sprintf("tmdb:%d", m.TMDbID) + case m.BangumiID > 0: + return fmt.Sprintf("bangumi:%d", m.BangumiID) + case strings.TrimSpace(m.DoubanID) != "": + return "douban:" + strings.ToLower(strings.TrimSpace(m.DoubanID)) + case strings.TrimSpace(m.TheTVDBID) != "": + return "thetvdb:" + strings.ToLower(strings.TrimSpace(m.TheTVDBID)) + } + title := firstNonEmpty(m.OriginalName, m.Title) + if title == "" { + title, _ = CleanQuery(m.Path) + } + title = normalizeMediaVersionText(title) + if title == "" { + return "" + } + year := m.Year + if year <= 0 { + _, year = CleanQuery(m.Path) + } + return fmt.Sprintf("movie:%s:%d", title, year) +} + +func normalizeMediaVersionText(value string) string { + value = strings.ToLower(strings.TrimSpace(value)) + if value == "" { + return "" + } + fields := strings.FieldsFunc(value, func(r rune) bool { + switch r { + case '.', '_', '-', ' ', '\t', '/', '\\', '[', ']', '(', ')', '（', '）', '【', '】': + return true + default: + return false + } + }) + out := fields[:0] + for _, field := range fields { + field = strings.TrimSpace(field) + if field == "" { + continue + } + if _, noise := noiseTokenSet[field]; noise { + continue + } + out = append(out, field) + } + return strings.Join(out, " ") +} + +func betterMediaVersion(candidate, current model.Media) bool { + candidatePixels := candidate.Width * candidate.Height + currentPixels := current.Width * current.Height + if candidatePixels != currentPixels { + return candidatePixels > currentPixels + } + if candidate.SizeBytes != current.SizeBytes { + return candidate.SizeBytes > current.SizeBytes + } + return candidate.CreatedAt.After(current.CreatedAt) } // SearchMedia performs a simple LIKE search across titles. @@ -361,6 +570,14 @@ func (s *MediaService) SearchMediaVisible(ctx context.Context, query string, lim }) } +func (s *MediaService) SearchMediaVisibleGrouped(ctx context.Context, query string, limit int, visibility MediaVisibility) ([]MediaItem, error) { + items, err := s.SearchMediaVisible(ctx, query, limit, visibility) + if err != nil { + return nil, err + } + return groupMediaVersions(items), nil +} + func (s *MediaService) SearchMediaVisiblePage(ctx context.Context, query string, page, pageSize int, visibility MediaVisibility) ([]model.Media, int64, error) { if pageSize <= 0 { pageSize = 50 @@ -379,6 +596,15 @@ func (s *MediaService) SearchMediaVisiblePage(ctx context.Context, query string, }) } +func (s *MediaService) SearchMediaVisiblePageGrouped(ctx context.Context, query string, page, pageSize int, visibility MediaVisibility) ([]MediaItem, int64, error) { + items, _, err := s.SearchMediaVisiblePage(ctx, query, page, pageSize, visibility) + if err != nil { + return nil, 0, err + } + grouped := groupMediaVersions(items) + return grouped, int64(len(grouped)), nil +} + // GetMedia returns a single media row. func (s *MediaService) GetMedia(ctx context.Context, id string) (*model.Media, error) { return s.repo.Media.FindByID(ctx, id) @@ -392,15 +618,27 @@ func (s *MediaService) SoftDelete(ctx context.Context, id string) error { return err } if media != nil && isCloudMediaPath(media.Path) { - return s.repo.DB.WithContext(ctx).Unscoped().Where("id = ?", id).Delete(&model.Media{}).Error + err := s.repo.DB.WithContext(ctx).Unscoped().Where("id = ?", id).Delete(&model.Media{}).Error + if err == nil { + s.invalidateMediaCache(ctx) + } + return err } - return s.repo.DB.Where("id = ?", id).Delete(&model.Media{}).Error + err = s.repo.DB.WithContext(ctx).Where("id = ?", id).Delete(&model.Media{}).Error + if err == nil { + s.invalidateMediaCache(ctx) + } + return err } // RestoreDeleted unsets DeletedAt for a single media row. func (s *MediaService) RestoreDeleted(ctx context.Context, id string) error { - return s.repo.DB.Unscoped().Model(&model.Media{}). + err := s.repo.DB.WithContext(ctx).Unscoped().Model(&model.Media{}). Where("id = ?", id).Update("deleted_at", nil).Error + if err == nil { + s.invalidateMediaCache(ctx) + } + return err } // ListRecycleBin returns every soft-deleted row, newest first. @@ -419,5 +657,9 @@ func (s *MediaService) ListRecycleBin(ctx context.Context, limit int) ([]model.M // PurgeDeleted permanently removes a soft-deleted row from the database. func (s *MediaService) PurgeDeleted(ctx context.Context, id string) error { - return s.repo.DB.Unscoped().Where("id = ?", id).Delete(&model.Media{}).Error + err := s.repo.DB.WithContext(ctx).Unscoped().Where("id = ?", id).Delete(&model.Media{}).Error + if err == nil { + s.invalidateMediaCache(ctx) + } + return err } diff --git a/internal/service/media_test.go b/internal/service/media_test.go index 17b61ee..f88cd89 100644 --- a/internal/service/media_test.go +++ b/internal/service/media_test.go @@ -4,6 +4,7 @@ import ( "os" "path/filepath" "testing" + "time" "github.com/glebarez/sqlite" "go.uber.org/zap" @@ -264,3 +265,39 @@ func TestSoftDeleteCloudMediaPurgesRecordWithoutRecycleBin(t *testing.T) { t.Fatalf("cloud media removal must not populate recycle bin: %#v", recycle) } } + +func TestSoftDeleteInvalidatesMediaAndStatsCache(t *testing.T) { + db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if err := db.AutoMigrate(&model.Media{}); err != nil { + t.Fatal(err) + } + repos := repository.New(db) + media := model.Media{ + Base: model.Base{ID: "local-media"}, + Title: "Cached Movie", + Path: filepath.Join(t.TempDir(), "Cached Movie.mkv"), + } + if err := repos.DB.Create(&media).Error; err != nil { + t.Fatal(err) + } + + cache := NewRuntimeCacheService(&config.Config{}, zap.NewNop()) + cache.SetJSON(t.Context(), "media:list:test", map[string]string{"state": "stale"}, time.Minute) + cache.SetJSON(t.Context(), "stats:snapshot:base", map[string]int{"media": 1}, time.Minute) + svc := NewMediaService(&config.Config{}, zap.NewNop(), repos).SetRuntimeCache(cache) + if err := svc.SoftDelete(t.Context(), media.ID); err != nil { + t.Fatal(err) + } + + var mediaCache map[string]string + if cache.GetJSON(t.Context(), "media:list:test", &mediaCache) { + t.Fatal("soft delete should invalidate media cache") + } + var statsCache map[string]int + if cache.GetJSON(t.Context(), "stats:snapshot:base", &statsCache) { + t.Fatal("soft delete should invalidate stats cache") + } +} diff --git a/internal/service/organizer.go b/internal/service/organizer.go index 5acd502..6d99e3a 100644 --- a/internal/service/organizer.go +++ b/internal/service/organizer.go @@ -283,7 +283,8 @@ func (o *OrganizerService) resolveTransferMode(ctx context.Context, override Tra } if mode == TransferMove && o.keepSeedingEnabled(ctx) { // 移动会删除源文件导致 qBittorrent 停止做种；保种开启时改用硬链接 - //（跨盘自动退化为复制），既规范命名又保留源文件继续做种上传。 + // 既规范命名又保留源文件继续做种上传。硬链接失败时会报错，避免静默 + // 退化复制后占用双份磁盘空间。 return TransferHardlink } return mode diff --git a/internal/service/runtime_cache.go b/internal/service/runtime_cache.go new file mode 100644 index 0000000..1434fa3 --- /dev/null +++ b/internal/service/runtime_cache.go @@ -0,0 +1,193 @@ +package service + +import ( + "context" + "encoding/json" + "strings" + "sync" + "time" + + "github.com/redis/go-redis/v9" + "go.uber.org/zap" + + "github.com/ShukeBta/MediaStationGo/internal/config" +) + +type RuntimeCacheService struct { + log *zap.Logger + client *redis.Client + prefix string + + mu sync.RWMutex + memory map[string]runtimeCacheItem + limit int +} + +type runtimeCacheItem struct { + raw []byte + expiresAt time.Time +} + +func NewRuntimeCacheService(cfg *config.Config, log *zap.Logger) *RuntimeCacheService { + c := &RuntimeCacheService{log: log, memory: map[string]runtimeCacheItem{}, limit: 2048} + if cfg == nil { + return c + } + c.prefix = strings.Trim(strings.TrimSpace(cfg.Cache.RedisPrefix), ":") + if c.prefix == "" { + c.prefix = "mediastationgo" + } + rawURL := strings.TrimSpace(cfg.Cache.RedisURL) + if rawURL == "" { + return c + } + opts, err := redis.ParseURL(rawURL) + if err != nil { + if log != nil { + log.Warn("redis cache disabled: invalid redis url", zap.Error(err)) + } + return c + } + client := redis.NewClient(opts) + pingCtx, cancel := context.WithTimeout(context.Background(), 1200*time.Millisecond) + defer cancel() + if err := client.Ping(pingCtx).Err(); err != nil { + if log != nil { + log.Warn("redis cache unavailable; using in-process cache", zap.Error(err)) + } + _ = client.Close() + return c + } + c.client = client + if log != nil { + log.Info("redis runtime cache enabled with in-process L1", zap.String("addr", opts.Addr), zap.String("prefix", c.prefix)) + } + return c +} + +func (c *RuntimeCacheService) Enabled() bool { + return c != nil +} + +func (c *RuntimeCacheService) Close() error { + if c == nil || c.client == nil { + return nil + } + return c.client.Close() +} + +func (c *RuntimeCacheService) GetJSON(ctx context.Context, key string, out any) bool { + if !c.Enabled() || strings.TrimSpace(key) == "" || out == nil { + return false + } + fullKey := c.key(key) + if raw, ok := c.getMemory(fullKey); ok { + return json.Unmarshal(raw, out) == nil + } + if c.client != nil { + raw, err := c.client.Get(ctx, fullKey).Bytes() + if err == nil { + if json.Unmarshal(raw, out) != nil { + return false + } + c.setMemory(fullKey, raw, 2*time.Second) + return true + } + } + return false +} + +func (c *RuntimeCacheService) SetJSON(ctx context.Context, key string, value any, ttl time.Duration) { + if !c.Enabled() || strings.TrimSpace(key) == "" || value == nil || ttl <= 0 { + return + } + raw, err := json.Marshal(value) + if err != nil { + return + } + fullKey := c.key(key) + c.setMemory(fullKey, raw, ttl) + if c.client != nil { + _ = c.client.Set(ctx, fullKey, raw, ttl).Err() + } +} + +func (c *RuntimeCacheService) DeletePrefix(ctx context.Context, prefix string) { + if !c.Enabled() || strings.TrimSpace(prefix) == "" { + return + } + fullPrefix := c.key(prefix) + c.deleteMemoryPrefix(fullPrefix) + if c.client != nil { + pattern := fullPrefix + "*" + var cursor uint64 + for { + keys, next, err := c.client.Scan(ctx, cursor, pattern, 200).Result() + if err != nil { + return + } + if len(keys) > 0 { + _ = c.client.Del(ctx, keys...).Err() + } + cursor = next + if cursor == 0 { + return + } + } + } +} + +func (c *RuntimeCacheService) key(key string) string { + key = strings.TrimLeft(strings.TrimSpace(key), ":") + if c.prefix == "" { + return key + } + return c.prefix + ":" + key +} + +func (c *RuntimeCacheService) getMemory(key string) ([]byte, bool) { + now := time.Now() + c.mu.RLock() + item, ok := c.memory[key] + c.mu.RUnlock() + if !ok { + return nil, false + } + if now.After(item.expiresAt) { + c.mu.Lock() + delete(c.memory, key) + c.mu.Unlock() + return nil, false + } + return item.raw, true +} + +func (c *RuntimeCacheService) setMemory(key string, raw []byte, ttl time.Duration) { + if ttl <= 0 || len(raw) == 0 { + return + } + c.mu.Lock() + defer c.mu.Unlock() + if len(c.memory) >= c.limit { + now := time.Now() + for k, item := range c.memory { + if now.After(item.expiresAt) || len(c.memory) >= c.limit { + delete(c.memory, k) + } + if len(c.memory) < c.limit { + break + } + } + } + c.memory[key] = runtimeCacheItem{raw: append([]byte(nil), raw...), expiresAt: time.Now().Add(ttl)} +} + +func (c *RuntimeCacheService) deleteMemoryPrefix(prefix string) { + c.mu.Lock() + defer c.mu.Unlock() + for key := range c.memory { + if strings.HasPrefix(key, prefix) { + delete(c.memory, key) + } + } +} diff --git a/internal/service/scanner.go b/internal/service/scanner.go index ebff15f..77503dc 100644 --- a/internal/service/scanner.go +++ b/internal/service/scanner.go @@ -56,6 +56,7 @@ type ScannerService struct { probe *FFprobeService scraper *ScraperService storage *StorageConfigService + cache *RuntimeCacheService imageProxy *ImageProxy @@ -113,6 +114,12 @@ func (s *ScannerService) SetStorageConfig(storage *StorageConfigService) { } } +func (s *ScannerService) SetRuntimeCache(cache *RuntimeCacheService) { + if s != nil { + s.cache = cache + } +} + // SetImageProxy lets cloud scans warm sidecar poster/backdrop files into the // local image cache. This keeps library opening fast without forcing the UI or // Emby clients to resolve/download every cloud poster on demand. @@ -255,18 +262,38 @@ func isCloudArtworkRef(ref string) bool { // ScanResult summarises a scan run. type ScanResult struct { - LibraryID string `json:"library_id"` - Visited int `json:"visited"` - Added int `json:"added"` - Updated int `json:"updated"` - Skipped int `json:"skipped"` - Probed int `json:"probed"` - LocalMetadata int `json:"local_metadata"` - Removed int64 `json:"removed"` + LibraryID string `json:"library_id"` + Visited int `json:"visited"` + Added int `json:"added"` + Updated int `json:"updated"` + Skipped int `json:"skipped"` + Probed int `json:"probed"` + LocalMetadata int `json:"local_metadata"` + Removed int64 `json:"removed"` + ErrorCount int `json:"error_count,omitempty"` + Errors []string `json:"errors,omitempty"` } var ErrCloudScanAlreadyRunning = errors.New("cloud scan already running") +const maxScanErrorDetails = 20 + +func addScanError(res *ScanResult, path string, err error) { + if res == nil || err == nil { + return + } + res.ErrorCount++ + if len(res.Errors) >= maxScanErrorDetails { + return + } + path = strings.TrimSpace(path) + msg := strings.TrimSpace(err.Error()) + if path != "" { + msg = path + ": " + msg + } + res.Errors = append(res.Errors, msg) +} + const maxCloudMediaProbeQueuePerScan = 32 const cloudMediaProbeFailureBackoff = 6 * time.Hour @@ -291,6 +318,8 @@ type CloudScanStatus struct { Updated int `json:"updated"` Skipped int `json:"skipped"` Removed int64 `json:"removed"` + ErrorCount int `json:"error_count,omitempty"` + Errors []string `json:"errors,omitempty"` Error string `json:"error,omitempty"` ResumeHint string `json:"resume_hint,omitempty"` Estimate string `json:"estimate_message,omitempty"` @@ -327,6 +356,17 @@ type existingCloudMedia struct { STRMURL string } +type existingLocalMedia struct { + SizeBytes int64 + DurationSec int + Width int + Height int + VideoCodec string + AudioCodec string + Container string + STRMURL string +} + func (s *ScannerService) cloudMediaProbeWorker() { for task := range s.cloudMediaProbeQueue { s.probeCloudMediaAsync(task) @@ -451,6 +491,8 @@ func (s *ScannerService) beginCloudScan(ctx context.Context, lib *model.Library, current.status.Updated = res.Updated current.status.Skipped = res.Skipped current.status.Removed = res.Removed + current.status.ErrorCount = res.ErrorCount + current.status.Errors = append([]string(nil), res.Errors...) } current.status.UpdatedAt = now current.status.FinishedAt = now @@ -471,22 +513,28 @@ func (s *ScannerService) beginCloudScan(ctx context.Context, lib *model.Library, default: current.status.State = "finished" current.status.Stage = "finished" - current.status.Error = "" + if current.status.ErrorCount > 0 { + current.status.Error = fmt.Sprintf("部分文件入库失败：%d 个，详情见 errors", current.status.ErrorCount) + } else { + current.status.Error = "" + } } if s.hub != nil { s.hub.Publish("scan", map[string]any{ - "library_id": lib.ID, - "provider": mount.Provider, - "cloud": true, - "finished": true, - "state": current.status.State, - "stage": current.status.Stage, - "error": current.status.Error, - "visited": current.status.Visited, - "added": current.status.Added, - "updated": current.status.Updated, - "skipped": current.status.Skipped, - "removed": current.status.Removed, + "library_id": lib.ID, + "provider": mount.Provider, + "cloud": true, + "finished": true, + "state": current.status.State, + "stage": current.status.Stage, + "error": current.status.Error, + "visited": current.status.Visited, + "added": current.status.Added, + "updated": current.status.Updated, + "skipped": current.status.Skipped, + "removed": current.status.Removed, + "error_count": current.status.ErrorCount, + "errors": current.status.Errors, }) } } @@ -643,7 +691,7 @@ func (s *ScannerService) StartCloudLibraryScan(libraryID string, autoScrape bool s.cloudScanMu.Unlock() go func() { - ctx, cancel := context.WithTimeout(context.Background(), 6*time.Hour) + ctx, cancel := cloudScanContext(context.Background(), cloudScanTimeout(context.Background(), s.repo, 24*time.Hour)) defer cancel() if autoScrape { _, err = s.ScanLibrary(ctx, libraryID) @@ -667,6 +715,28 @@ func (s *ScannerService) StartCloudLibraryScan(libraryID string, autoScrape bool return status, true, nil } +func cloudScanContext(parent context.Context, timeout time.Duration) (context.Context, context.CancelFunc) { + if timeout <= 0 { + return context.WithCancel(parent) + } + return context.WithTimeout(parent, timeout) +} + +func cloudScanTimeout(ctx context.Context, repo *repository.Container, fallback time.Duration) time.Duration { + if repo == nil || repo.Setting == nil { + return fallback + } + value, err := repo.Setting.Get(ctx, "cloud.scan_timeout_hours") + if err != nil || strings.TrimSpace(value) == "" { + return fallback + } + hours := parseIntSettingDefault(strings.TrimSpace(value), int(fallback/time.Hour)) + if hours <= 0 { + return 0 + } + return time.Duration(hours) * time.Hour +} + func (s *ScannerService) StartAllCloudLibraryScans() ([]CloudScanStatus, error) { if s == nil { return nil, errors.New("scanner unavailable") @@ -754,8 +824,19 @@ func (s *ScannerService) scanLibrary(ctx context.Context, libraryID string, auto res := &ScanResult{LibraryID: lib.ID} seen := make(map[string]struct{}) seenInodes := make(map[string]string) + writeBatch := newLocalMediaWriteBatch(s, ctx, res, 100) + existingMedia, err := s.existingLocalMediaSnapshot(ctx, lib.ID) + if err != nil { + s.log.Warn("load existing local media snapshot failed", zap.String("library_id", lib.ID), zap.Error(err)) + existingMedia = nil + } walkFn := func(path string, info walkInfo) error { + select { + case <-ctx.Done(): + return ctx.Err() + default: + } if info.isDir { return nil } @@ -764,12 +845,18 @@ func (s *ScannerService) scanLibrary(ctx context.Context, libraryID string, auto return nil } seen[filepath.Clean(path)] = struct{}{} - s.ingestFile(ctx, lib, path, info.size, seenInodes, res) + s.ingestFile(ctx, lib, path, info.size, seenInodes, existingMedia, writeBatch, res) return nil } - if err := walk(lib.Path, walkFn); err != nil { - return res, err + walkErr := walk(lib.Path, walkFn) + writeBatch.Flush() + if walkErr != nil { + addScanError(res, lib.Path, walkErr) + if res.Added+res.Updated > 0 { + s.invalidateMediaCache(ctx) + } + return res, walkErr } removed, err := s.pruneMissingMedia(ctx, lib.ID, seen) if err != nil { @@ -779,15 +866,18 @@ func (s *ScannerService) scanLibrary(ctx context.Context, libraryID string, auto } s.hub.Publish("scan", map[string]any{ - "library_id": lib.ID, - "finished": true, - "visited": res.Visited, - "added": res.Added, - "updated": res.Updated, - "probed": res.Probed, - "local_meta": res.LocalMetadata, - "removed": res.Removed, + "library_id": lib.ID, + "finished": true, + "visited": res.Visited, + "added": res.Added, + "updated": res.Updated, + "probed": res.Probed, + "local_meta": res.LocalMetadata, + "removed": res.Removed, + "error_count": res.ErrorCount, + "errors": res.Errors, }) + s.invalidateMediaCache(ctx) s.maybeGenerateSTRMAfterScan(lib.ID) // Online enrichment is opt-in. Local NFO is always consumed first during @@ -820,7 +910,10 @@ func (s *ScannerService) IngestPath(ctx context.Context, libraryID, path string) return false, nil } res := &ScanResult{LibraryID: lib.ID} - s.ingestFile(ctx, lib, path, fi.Size(), make(map[string]string), res) + s.ingestFile(ctx, lib, path, fi.Size(), make(map[string]string), nil, nil, res) + if res.Added+res.Updated > 0 { + s.invalidateMediaCache(ctx) + } return res.Added+res.Updated > 0, nil } @@ -1054,12 +1147,16 @@ func (s *ScannerService) scanCloudLibrary(ctx context.Context, lib *model.Librar "visited": res.Visited, "added": res.Added, "updated": res.Updated, + "skipped": res.Skipped, "removed": res.Removed, + "error_count": res.ErrorCount, + "errors": res.Errors, "discovered": filesDiscovered, "dirs": dirsVisited, "elapsed_seconds": int(time.Since(startedAt).Seconds()), "cloud": true, }) + s.invalidateMediaCache(ctx) s.maybeGenerateSTRMAfterScan(lib.ID) if autoScrape && s.scraper != nil && s.scraper.AnyEnabled() && s.autoScrapeEnabled(ctx) { s.startAutoScrape(ctx, lib.ID) @@ -1067,6 +1164,13 @@ func (s *ScannerService) scanCloudLibrary(ctx context.Context, lib *model.Librar return res, nil } +func (s *ScannerService) invalidateMediaCache(ctx context.Context) { + if s != nil && s.cache != nil { + s.cache.DeletePrefix(ctx, "media:") + s.cache.DeletePrefix(ctx, "stats:") + } +} + func (s *ScannerService) startAutoScrape(ctx context.Context, libraryID string) { scrapeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Minute) go func() { @@ -1118,6 +1222,43 @@ func (s *ScannerService) existingCloudMediaSnapshot(ctx context.Context, library return out, nil } +func (s *ScannerService) existingLocalMediaSnapshot(ctx context.Context, libraryID string) (map[string]existingLocalMedia, error) { + var rows []struct { + Path string + SizeBytes int64 + DurationSec int + Width int + Height int + VideoCodec string + AudioCodec string + Container string + STRMURL string + } + if err := s.repo.DB.WithContext(ctx). + Model(&model.Media{}). + Select("path, size_bytes, duration_sec, width, height, video_codec, audio_codec, container, strm_url"). + Where("library_id = ? AND path NOT LIKE ?", libraryID, "cloud://%"). + Find(&rows).Error; err != nil { + return nil, err + } + out := make(map[string]existingLocalMedia, len(rows)) + for _, row := range rows { + if row.Path != "" { + out[filepath.Clean(row.Path)] = existingLocalMedia{ + SizeBytes: row.SizeBytes, + DurationSec: row.DurationSec, + Width: row.Width, + Height: row.Height, + VideoCodec: row.VideoCodec, + AudioCodec: row.AudioCodec, + Container: row.Container, + STRMURL: row.STRMURL, + } + } + } + return out, nil +} + func (s *ScannerService) shadowedCloudLibrary(ctx context.Context, lib *model.Library) *CloudMountConflict { libs, err := s.repo.Library.List(ctx) if err != nil { @@ -1213,6 +1354,7 @@ func (s *ScannerService) ingestCloudFile(ctx context.Context, lib *model.Library s.queueCloudArtworkPrefetch(localMeta.BackdropURL) } if err := s.repo.Media.Upsert(ctx, m); err != nil { + addScanError(res, path, err) s.log.Warn("upsert cloud media failed", zap.String("path", path), zap.Error(err)) return } @@ -1343,6 +1485,18 @@ func cloudTrackMetadataMissing(existing existingCloudMedia) bool { strings.TrimSpace(existing.AudioCodec) == "" } +func localTrackMetadataMissing(existing existingLocalMedia) bool { + return existing.DurationSec <= 0 || + existing.Width <= 0 || + existing.Height <= 0 || + strings.TrimSpace(existing.VideoCodec) == "" || + strings.TrimSpace(existing.AudioCodec) == "" +} + +func localMetadataNeedsRefresh(local *LocalMetadata) bool { + return local != nil && (local.HasNFO || local.HasArtwork || localHasDescriptiveMetadata(local)) +} + func cloudSeriesTitleFromMediaPath(mediaPath string) (string, int) { displayPath := strings.TrimSpace(mediaPath) if strings.HasPrefix(strings.ToLower(displayPath), "cloud://") { @@ -1394,12 +1548,15 @@ func (s *ScannerService) RemovePath(ctx context.Context, path string) (int64, er res := s.repo.DB.WithContext(ctx). Where("path = ?", path). Delete(&model.Media{}) + if res.Error == nil && res.RowsAffected > 0 { + s.invalidateMediaCache(ctx) + } return res.RowsAffected, res.Error } // ingestFile upserts a single media file. seenInodes dedups hardlinks within a // single scan; pass a fresh map for one-off ingests. It mutates res counters. -func (s *ScannerService) ingestFile(ctx context.Context, lib *model.Library, path string, size int64, seenInodes map[string]string, res *ScanResult) { +func (s *ScannerService) ingestFile(ctx context.Context, lib *model.Library, path string, size int64, seenInodes map[string]string, existingMedia map[string]existingLocalMedia, writeBatch *localMediaWriteBatch, res *ScanResult) { res.Visited++ ext := strings.ToLower(filepath.Ext(path)) @@ -1424,7 +1581,27 @@ func (s *ScannerService) ingestFile(ctx context.Context, lib *model.Library, pat seenInodes[fileID] = path } - isNewMedia := !s.mediaPathExists(ctx, path) + cleanPath := filepath.Clean(path) + parsedSeason, parsedEpisode := ParseEpisode(path) + localMeta, localMetaErr := ReadLocalMetadata(path, lib.Path, librarySupportsSeasons(lib) || parsedSeason > 0 || parsedEpisode > 0) + if localMetaErr != nil { + s.log.Warn("read local metadata failed", zap.String("path", path), zap.Error(localMetaErr)) + } + isNewMedia := false + if existingMedia != nil { + existing, exists := existingMedia[cleanPath] + isNewMedia = !exists + if exists && + ext != ".strm" && + existing.SizeBytes == size && + (s.probe == nil || !localTrackMetadataMissing(existing)) && + !localMetadataNeedsRefresh(localMeta) { + res.Skipped++ + return + } + } else { + isNewMedia = !s.mediaPathExists(ctx, path) + } title, year := CleanQuery(path) if title == "" { @@ -1449,15 +1626,12 @@ func (s *ScannerService) ingestFile(ctx context.Context, lib *model.Library, pat } } - parsedSeason, parsedEpisode := ParseEpisode(path) m.SeasonNum = parsedSeason m.EpisodeNum = parsedEpisode - if local, err := ReadLocalMetadata(path, lib.Path, librarySupportsSeasons(lib) || parsedSeason > 0 || parsedEpisode > 0); err == nil && local != nil { - applyLocalMetadata(m, local) + if localMeta != nil { + applyLocalMetadata(m, localMeta) res.LocalMetadata++ - } else if err != nil { - s.log.Warn("read local metadata failed", zap.String("path", path), zap.Error(err)) } // Best-effort ffprobe; failure does not abort the file. @@ -1477,7 +1651,12 @@ func (s *ScannerService) ingestFile(ctx context.Context, lib *model.Library, pat } } + if isNewMedia && writeBatch != nil { + writeBatch.Add(path, m) + return + } if err := s.repo.Media.Upsert(ctx, m); err != nil { + addScanError(res, path, err) s.log.Warn("upsert media failed", zap.String("path", path), zap.Error(err)) return } @@ -1497,6 +1676,88 @@ func (s *ScannerService) ingestFile(ctx context.Context, lib *model.Library, pat }) } +type localMediaWriteBatch struct { + scanner *ScannerService + ctx context.Context + res *ScanResult + limit int + items []localMediaWriteItem +} + +type localMediaWriteItem struct { + path string + media *model.Media +} + +func newLocalMediaWriteBatch(scanner *ScannerService, ctx context.Context, res *ScanResult, limit int) *localMediaWriteBatch { + if limit <= 0 { + limit = 100 + } + return &localMediaWriteBatch{scanner: scanner, ctx: ctx, res: res, limit: limit} +} + +func (b *localMediaWriteBatch) Add(path string, media *model.Media) { + if b == nil || b.scanner == nil || media == nil { + return + } + if media.ScrapeStatus == "" { + media.ScrapeStatus = "pending" + } + b.items = append(b.items, localMediaWriteItem{path: path, media: media}) + if len(b.items) >= b.limit { + b.Flush() + } +} + +func (b *localMediaWriteBatch) Flush() { + if b == nil || len(b.items) == 0 || b.scanner == nil || b.scanner.repo == nil || b.scanner.repo.DB == nil { + return + } + items := b.items + b.items = nil + media := make([]model.Media, 0, len(items)) + for _, item := range items { + if item.media != nil { + media = append(media, *item.media) + } + } + if len(media) == 0 { + return + } + if err := b.scanner.repo.DB.WithContext(b.ctx).CreateInBatches(&media, b.limit).Error; err == nil { + b.res.Added += len(media) + b.publish() + return + } + for _, item := range items { + if item.media == nil { + continue + } + if err := b.scanner.repo.Media.Upsert(b.ctx, item.media); err != nil { + addScanError(b.res, item.path, err) + b.scanner.log.Warn("upsert media failed", zap.String("path", item.path), zap.Error(err)) + continue + } + b.res.Added++ + } + b.publish() +} + +func (b *localMediaWriteBatch) publish() { + if b == nil || b.scanner == nil || b.scanner.hub == nil || b.res == nil { + return + } + b.scanner.hub.Publish("scan", map[string]any{ + "library_id": b.res.LibraryID, + "visited": b.res.Visited, + "added": b.res.Added, + "updated": b.res.Updated, + "probed": b.res.Probed, + "local_meta": b.res.LocalMetadata, + "batched": true, + }) +} + // duplicateByFileID reports an existing media path that shares the given inode // identity but lives at a different path and still exists on disk. func (s *ScannerService) duplicateByFileID(ctx context.Context, fileID, path string) (string, bool) { diff --git a/internal/service/scanner_incremental_test.go b/internal/service/scanner_incremental_test.go index 12d7a53..b6d61db 100644 --- a/internal/service/scanner_incremental_test.go +++ b/internal/service/scanner_incremental_test.go @@ -4,6 +4,7 @@ import ( "os" "path/filepath" "testing" + "time" "github.com/glebarez/sqlite" "go.uber.org/zap" @@ -95,6 +96,62 @@ func TestScanLibraryReadsLocalSTRMTarget(t *testing.T) { } } +func TestScanLibrarySkipsUnchangedExistingLocalMedia(t *testing.T) { + sc, repos := newScannerTestEnv(t) + root := t.TempDir() + lib := model.Library{Name: "Movies", Path: root, Type: "movie", Enabled: true} + if err := repos.Library.Create(t.Context(), &lib); err != nil { + t.Fatal(err) + } + file := filepath.Join(root, "Already In Library (2024).mkv") + if err := os.WriteFile(file, []byte("same-size"), 0o644); err != nil { + t.Fatal(err) + } + first, err := sc.ScanLibrary(t.Context(), lib.ID) + if err != nil { + t.Fatalf("first scan: %v", err) + } + if first.Added != 1 || first.Skipped != 0 { + t.Fatalf("first scan = %#v, want added=1 skipped=0", first) + } + second, err := sc.ScanLibrary(t.Context(), lib.ID) + if err != nil { + t.Fatalf("second scan: %v", err) + } + if second.Added != 0 || second.Updated != 0 || second.Skipped != 1 { + t.Fatalf("second scan = %#v, want unchanged file skipped", second) + } + if got := countMedia(t, repos); got != 1 { + t.Fatalf("media count = %d, want 1", got) + } +} + +func TestScanLibraryReportsPerFileUpsertErrors(t *testing.T) { + sc, repos := newScannerTestEnv(t) + root := t.TempDir() + lib := model.Library{Name: "Broken DB", Path: root, Type: "movie", Enabled: true} + if err := repos.Library.Create(t.Context(), &lib); err != nil { + t.Fatal(err) + } + file := filepath.Join(root, "Cannot Insert (2024).mkv") + if err := os.WriteFile(file, []byte("data"), 0o644); err != nil { + t.Fatal(err) + } + if err := repos.DB.Exec("DROP TABLE media").Error; err != nil { + t.Fatal(err) + } + res, err := sc.ScanLibrary(t.Context(), lib.ID) + if err != nil { + t.Fatalf("scan should continue and report file errors, got top-level error: %v", err) + } + if res.ErrorCount != 1 || len(res.Errors) != 1 { + t.Fatalf("scan errors = count %d details %#v, want one detailed error", res.ErrorCount, res.Errors) + } + if res.Visited != 1 { + t.Fatalf("visited = %d, want 1", res.Visited) + } +} + func TestScanLibraryMapsPersistedHostLibraryPath(t *testing.T) { sc, repos := newScannerTestEnv(t) root := t.TempDir() @@ -132,6 +189,8 @@ func TestScanLibraryMapsPersistedHostLibraryPath(t *testing.T) { func TestRemovePathDeletesVanishedMedia(t *testing.T) { sc, repos := newScannerTestEnv(t) + cache := NewRuntimeCacheService(&config.Config{}, zap.NewNop()) + sc.SetRuntimeCache(cache) root := t.TempDir() lib := model.Library{Name: "Movies", Path: root, Type: "movie", Enabled: true} if err := repos.Library.Create(t.Context(), &lib); err != nil { @@ -147,10 +206,18 @@ func TestRemovePathDeletesVanishedMedia(t *testing.T) { if countMedia(t, repos) != 1 { t.Fatal("expected 1 media before removal") } + cache.SetJSON(t.Context(), "media:list:test", map[string]string{"state": "stale"}, time.Minute) + var cached map[string]string + if !cache.GetJSON(t.Context(), "media:list:test", &cached) { + t.Fatal("expected media cache to be primed") + } // A still-present file is not removed. if removed, _ := sc.RemovePath(t.Context(), file); removed != 0 { t.Fatalf("present file should not be removed, got %d", removed) } + if !cache.GetJSON(t.Context(), "media:list:test", &cached) { + t.Fatal("present file should not invalidate media cache") + } if err := os.Remove(file); err != nil { t.Fatal(err) } @@ -164,6 +231,9 @@ func TestRemovePathDeletesVanishedMedia(t *testing.T) { if countMedia(t, repos) != 0 { t.Fatal("expected 0 media after removal") } + if cache.GetJSON(t.Context(), "media:list:test", &cached) { + t.Fatal("vanished media removal should invalidate media cache") + } } // TestScanSkipsHardlinkDuplicate verifies that a hardlink (same inode) kept diff --git a/internal/service/service.go b/internal/service/service.go index ce55cc4..bd5b3e5 100644 --- a/internal/service/service.go +++ b/internal/service/service.go @@ -74,6 +74,7 @@ type Container struct { Notify *NotifyService Site *SiteService Device *DeviceService + Cache *RuntimeCacheService stopCtx context.Context stopCancel context.CancelFunc @@ -92,6 +93,13 @@ func New(cfg *config.Config, log *zap.Logger, repos *repository.Container) *Cont go sseHub.Run() probe := NewFFprobeService(cfg, log) + runtimeCache := NewRuntimeCacheService(cfg, log) + if searchBackend := repository.NewOpenSearchMediaBackend(cfg.Search); searchBackend != nil && repos != nil && repos.Media != nil { + repos.Media.SetSearchBackend(searchBackend) + if log != nil { + log.Info("opensearch media search enabled", zap.String("index", cfg.Search.Index), zap.String("url", cfg.Search.OpenSearchURL)) + } + } crypto := NewCryptoService(cfg.Secrets.JWTSecret, log) apiConfig := NewAPIConfigService(log, repos, crypto) tmdb := NewTMDbProvider(cfg, log, apiConfig) @@ -108,6 +116,7 @@ func New(cfg *config.Config, log *zap.Logger, repos *repository.Container) *Cont discover := NewDiscoverService(log, tmdb) transcoder := NewTranscoderService(cfg, log, repos, hub) scanner := NewScannerService(cfg, log, repos, hub, probe, scraper) + scanner.SetRuntimeCache(runtimeCache) organizePipeline := NewOrganizePipelineService(log, repos, organizer, scanner, tasks) watcher := NewWatcherService(log, repos, scanner) nfo := NewNFOService(log, repos) @@ -125,6 +134,7 @@ func New(cfg *config.Config, log *zap.Logger, repos *repository.Container) *Cont storageCfg := NewStorageConfigService(log, repos, crypto) strmSvc := NewSTRMService(log, repos, cfg) scanner.SetStorageConfig(storageCfg) + emby.SetRuntimeCache(runtimeCache) emby.SetCloudProbe(storageCfg, probe) downloadClients := NewDownloadClientService(log, repos) assistant := NewAssistantService(log, repos, ai) @@ -188,7 +198,7 @@ func New(cfg *config.Config, log *zap.Logger, repos *repository.Container) *Cont SSEHub: sseHub, Tasks: tasks, Auth: authSvc, - Media: NewMediaService(cfg, log, repos), + Media: NewMediaService(cfg, log, repos).SetRuntimeCache(runtimeCache), Scan: scanner, Stream: NewStreamService(cfg, log, repos, transcoder), Transcoder: transcoder, @@ -205,7 +215,7 @@ func New(cfg *config.Config, log *zap.Logger, repos *repository.Container) *Cont Downloads: downloads, Subscription: subscription, Subtitle: NewSubtitleService(log, repos), - Stats: NewStatsService(log, repos), + Stats: NewStatsService(log, repos).SetRuntimeCache(runtimeCache), Profile: NewProfileService(log, repos), Audit: NewAuditService(log, repos), NFO: nfo, @@ -237,6 +247,7 @@ func New(cfg *config.Config, log *zap.Logger, repos *repository.Container) *Cont Notify: notifySvc, Site: siteSvc, Device: deviceSvc, + Cache: runtimeCache, stopCtx: ctx, stopCancel: cancel, } @@ -444,6 +455,9 @@ func (c *Container) Close() { if c.Transcoder != nil { c.Transcoder.StopAll() } + if c.Cache != nil { + _ = c.Cache.Close() + } if c.WSHub != nil { c.WSHub.Stop() } diff --git a/internal/service/stats.go b/internal/service/stats.go index 43c0bb0..6ab8728 100644 --- a/internal/service/stats.go +++ b/internal/service/stats.go @@ -23,8 +23,9 @@ import ( // StatsService computes aggregate stats. type StatsService struct { - log *zap.Logger - repo *repository.Container + log *zap.Logger + repo *repository.Container + cache *RuntimeCacheService } // NewStatsService is the constructor. @@ -32,6 +33,13 @@ func NewStatsService(log *zap.Logger, repo *repository.Container) *StatsService return &StatsService{log: log, repo: repo} } +func (s *StatsService) SetRuntimeCache(cache *RuntimeCacheService) *StatsService { + if s != nil { + s.cache = cache + } + return s +} + // Snapshot is the JSON returned by /api/stats. type Snapshot struct { Libraries int64 `json:"libraries"` @@ -57,6 +65,15 @@ type Hardware struct { // Compute builds a fresh snapshot. func (s *StatsService) Compute(ctx context.Context, dataDir string) (*Snapshot, error) { + const cacheKey = "stats:snapshot:base" + if s.cache != nil { + var cached Snapshot + if s.cache.GetJSON(ctx, cacheKey, &cached) { + cached.GeneratedAt = time.Now() + cached.Hardware = readHardware(dataDir) + return &cached, nil + } + } snap := &Snapshot{GeneratedAt: time.Now()} libs, err := s.repo.Library.List(ctx) if err != nil { @@ -114,6 +131,11 @@ func (s *StatsService) Compute(ctx context.Context, dataDir string) (*Snapshot, return nil, err } + if s.cache != nil { + cacheCopy := *snap + cacheCopy.Hardware = Hardware{} + s.cache.SetJSON(ctx, cacheKey, cacheCopy, 10*time.Second) + } snap.Hardware = readHardware(dataDir) return snap, nil } diff --git a/internal/service/transfer.go b/internal/service/transfer.go index df1cc08..e7a8ffe 100644 --- a/internal/service/transfer.go +++ b/internal/service/transfer.go @@ -19,6 +19,8 @@ import ( "strings" ) +var linkFile = os.Link + // TransferMode 表示整理时文件的转移方式。 type TransferMode string @@ -62,14 +64,11 @@ func transferFile(src, dst string, mode TransferMode) error { case TransferCopy: return copyFile(src, dst) case TransferHardlink: - if err := os.Link(src, dst); err != nil { + if err := linkFile(src, dst); err != nil { // Docker 部署里下载目录和媒体目录往往是两个独立的 bind mount， // 即使在宿主机上同属一块盘，容器内 os.Link 也会因跨文件系统 - // (EXDEV) 失败。此前直接报错导致 PT 下载完成后整理静默中断； - // 现在自动降级为复制（保留源文件继续做种，语义一致）。 - if copyErr := copyFile(src, dst); copyErr == nil { - return nil - } + // (EXDEV) 失败。hardlink 模式必须保持零额外数据占用语义，不能 + // 自动降级为复制；需要复制时请显式选择 copy。 return fmt.Errorf("hardlink failed: %w; source and target must be on the same filesystem, choose copy if you want to duplicate data", err) } return nil diff --git a/internal/service/transfer_test.go b/internal/service/transfer_test.go index 8d1021e..335bbb9 100644 --- a/internal/service/transfer_test.go +++ b/internal/service/transfer_test.go @@ -1,8 +1,10 @@ package service import ( + "errors" "os" "path/filepath" + "strings" "testing" ) @@ -74,6 +76,34 @@ func TestTransferFileHardlinkSharesInodeAndKeepsSource(t *testing.T) { } } +func TestTransferFileHardlinkDoesNotFallBackToCopy(t *testing.T) { + dir := t.TempDir() + src := writeTemp(t, dir, "src.mkv", "payload") + dst := filepath.Join(dir, "dst.mkv") + + origLinkFile := linkFile + linkFile = func(_, _ string) error { + return errors.New("simulated cross-device link") + } + t.Cleanup(func() { + linkFile = origLinkFile + }) + + err := transferFile(src, dst, TransferHardlink) + if err == nil { + t.Fatal("hardlink failure should be reported instead of falling back to copy") + } + if !strings.Contains(err.Error(), "hardlink failed") { + t.Fatalf("hardlink error = %q, want hardlink failure context", err.Error()) + } + if _, statErr := os.Stat(dst); !os.IsNotExist(statErr) { + t.Fatalf("hardlink failure should not create copied dst, stat err = %v", statErr) + } + if b, readErr := os.ReadFile(src); readErr != nil || string(b) != "payload" { + t.Fatalf("hardlink failure should keep source unchanged, content=%q err=%v", b, readErr) + } +} + func TestTransferFileSymlinkKeepsSource(t *testing.T) { dir := t.TempDir() src := writeTemp(t, dir, "src.mkv", "payload")