feat: 完善内容聚合与搜索能力

2026-08-07 00:26:49 +08:00
parent d366c088a3
commit 0d852a9794
38 changed files with 1368 additions and 538 deletions
@@ -100,7 +100,7 @@ Bloom Filter 只能作为前置加速结构不得作为最终去重依据

 新增行为必须包含与风险相称的测试

-修改持久化 key 或编码格式时必须提供兼容或迁移方案
+开发阶段修改持久化 key 或编码格式时经用户明确授权可以清理测试数据重新开始 正式格式冻结后必须提供兼容或迁移方案

 不得把 `opencodes` 下的参考项目纳入 workspace 或修改其内容

@@ -37,6 +37,8 @@
 - [x] 使用当前 Git 配置统一作者仓库许可证和 edition 元数据
 - [x] 编写 `AGENTS.md` 记录架构边界和开发约定
 - [x] 将最终应用与可复用 DHT 基础库分离
+- [x] 删除被内容组索引替代的旧状态和无效兼容代码
+- [x] 收紧仅供测试或存储内部使用的接口和依赖
 - [x] 实现 BEP-51 `sample_infohashes` 主动发现
 - [x] 实现主动 Peer 查找和 Metadata 获取
 - [x] 验证远程公网设备能够持续获取 Metadata
@@ -58,13 +60,13 @@
 ### 任务

 - [x] 定义二十字节 `InfoHash` 类型和十六进制转换
- [x] 定义 `TorrentRecord` `TorrentFile` 和 `IndexState`
+- [x] 定义 `TorrentRecord` `TorrentFile` 和内容组索引状态
 - [x] 校验名称文件列表文件总大小和 infohash
- [x] 使用 BLAKE3 计算版本化内容指纹
+- [x] 使用 BLAKE3 计算规范化内容指纹
 - [x] 规范化 Unicode 路径分隔符大小写和文件顺序
 - [x] 保留真实子目录避免内容指纹碰撞
- [x] 定义版本化 RocksDB 二进制键空间
- [x] 实现数据库 schema 版本检查
+- [x] 定义 RocksDB 二进制键空间
+- [x] 实现数据库格式检查
 - [x] 实现 infohash 精确查询和存在性判断
 - [x] 实现新记录 WriteBatch 原子写入
 - [x] 实现重复 infohash 的 `last_seen` `seen_count` 和 Peer 更新
@@ -129,24 +131,24 @@
 - [x] 定义 Tantivy schema 和索引版本
 - [x] 索引名称文件路径扩展名 infohash 和内容指纹
 - [x] 将大小文件数时间和发现次数定义为 fast fields
- [ ] 设计中英文数字和文件名 tokenizer
+- [x] 设计中英文数字和文件名子串 tokenizer
 - [x] 实现待索引任务批量消费
 - [x] 实现按数量和时间间隔批量 commit
 - [x] commit 成功后原子更新 RocksDB 索引状态
 - [x] 实现关键词短语和精确 infohash 查询
- [ ] 实现大小时间扩展名和文件数过滤
+- [x] 实现大小时间扩展名文件数热度和可用性过滤
 - [x] 实现大小范围和扩展名过滤
- [ ] 实现相关性时间热度和大小排序
+- [x] 实现相关性时间热度大小和发现次数排序
 - [x] 建立带时间衰减的 DHT 活跃度分数和用户可读等级
 - [x] 实现分页并限制最大翻页成本
- [ ] 实现相同 `content_key` 结果折叠
+- [x] 实现相同 `content_key` 结果精确折叠和变体分页
 - [x] 实现从 RocksDB 全量重建 Tantivy 索引
- [ ] 支持索引 schema 不兼容时安全重建
+- [x] 支持索引结构不兼容时直接重建

 ### 验收标准

 - [x] 新写入记录在目标延迟内可搜索
- [ ] 搜索索引删除后可以从 RocksDB 完整重建
+- [x] 搜索索引删除后可以从 RocksDB 完整重建
 - [x] 索引过程中异常退出不会永久丢失文档
 - [ ] 百万级测试数据常用查询延迟达到约定目标

@@ -163,11 +165,12 @@
 - [x] 实现 `/stats` 运行状态接口
 - [x] 实现 `/search` 搜索过滤和分页接口
 - [x] 实现 `/torrents/{infohash}` 详情接口
+- [x] 实现 `/contents/{content_key}` 内容变体接口
 - [x] 定义统一错误响应
 - [x] 限制查询长度分页大小和最大 offset
 - [ ] 增加请求延迟错误率和并发指标
 - [x] 增加搜索详情字段和按需验证入队 API 端到端测试
- [ ] 增加其余 API 单元测试和端到端测试
+- [x] 增加搜索过滤折叠精确哈希和变体接口测试

 ### 验收标准

@@ -190,7 +193,7 @@
 - [x] 实现详情高优先级和搜索普通优先级的仅按需验证
 - [x] 使用持久化有界验证队列租约恢复去重和失败退避
 - [x] Metadata 与验证握手共享 TCP 建连总预算
- [x] RocksDB schema v1 到 v2 可恢复迁移并触发安全重建索引
+- [x] 开发阶段清理测试数据库并以当前数据结构重新采集
 - [x] 搜索和详情接口返回热度与可用性数据
 - [x] `/stats` 返回验证队列发现握手成功失败和拒绝指标

@@ -209,10 +212,10 @@
 - [x] 搜索和详情响应不等待 DHT 或 Peer 网络验证
 - [x] 进程重启后已接受的验证任务能够通过租约恢复
 - [x] 一次验证失败不会删除记录或标记为绝对失效
- [x] 旧数据库记录能够迁移并重新建立搜索文档
+- [x] 数据清空后能够建立新的内容组搜索文档

- [ ] 精确重复不会重复下载和重复展示
- [ ] 内容重复可以折叠并保留全部 infohash
+- [x] 精确重复不会重复下载和重复展示
+- [x] 内容重复可以折叠并保留全部 infohash
 - [ ] 过滤规则可以配置更新和回滚
 - [ ] 模糊去重不会直接造成数据丢失

@@ -289,6 +292,6 @@

 ## 当前下一步

-验证新可用性管线的真实运行数据并继续完善阶段三和阶段四
+使用全新空数据目录采集真实 Metadata 并验收内容组搜索过滤排序和变体接口

-下一步运行更长时间的按需验证采样并实现时间文件数过滤排序策略和内容聚合展示
+完成真实数据验收后开发最小搜索界面并进行更长时间的资源稳定性测试
@@ -1,3 +1,5 @@
+// 负责演示基础 DHT 爬虫的配置启动监控和安全停止方式
+
 #[cfg(feature = "mimalloc")]
 #[global_allocator]
 static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
@@ -1,3 +1,5 @@
+// 负责校验解析和规范化 DHT 节点网络地址
+
 use crate::types::NetMode;
 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};

@@ -1,3 +1,5 @@
+// 负责管理 DHT 引导节点解析重试退避和准入调度
+
 use crate::addr::{addr_allowed_by_netmode, is_valid_node_addr};
 use crate::crawl_config::ResolvedCrawlConfig;
 use crate::types::NetMode;
@@ -1,3 +1,5 @@
+// 负责实现网络查询共享速率预算和突发容量控制
+
 use std::{
    sync::{Arc, Mutex},
    time::{Duration, Instant},
@@ -1,3 +1,5 @@
+// 负责解析抓取配置并生成运行时资源限制
+
 use crate::types::CrawlOptions;
 use std::time::Duration;

@@ -1,3 +1,5 @@
+// 负责协调节点探测 Peer 查找采样发现和 Metadata 调度
+
 use crate::bootstrap::{BootstrapGate, BootstrapSourcePool, resolve_bootstrap_nodes};
 use crate::budget::{RateBucket, SharedRateBudget};
 use crate::crawl_config::ResolvedCrawlConfig;
@@ -1,3 +1,5 @@
+// 负责定义 DHT 基础库的统一错误类型和结果别名
+
 use thiserror::Error;

 #[derive(Error, Debug)]
@@ -1,3 +1,5 @@
+// 负责编码 DHT KRPC 查询响应和紧凑节点数据
+
 use crate::addr::{addr_allowed_by_netmode, is_valid_node_addr};
 use crate::node_id::TransactionId;
 use crate::protocol::DhtResponse;
@@ -5,7 +5,7 @@
 //! [`DHTServer`] is the primary entry point. Configure it with [`DHTOptions`], register
 //! callbacks, then await [`DHTServer::start`] until another task calls [`DHTServer::shutdown`].
 //! Runtime counters are available through [`DHTServer::runtime_stats`] without enabling any
-//! exporter. See the repository README for scheduling, backpressure and migration details.
+//! exporter. See the repository README for scheduling, backpressure and persistence details.

 mod addr;
 mod bootstrap;
@@ -1,3 +1,5 @@
+// 负责生成 DHT 节点标识事务标识和距离目标
+
 use rand::RngExt;

 pub(crate) type TransactionId = [u8; 8];
@@ -1,3 +1,5 @@
+// 负责维护有界 DHT 节点池去重淘汰和重访状态
+
 use crate::addr::is_valid_node_addr;
 use crate::budget::RateBucket;
 use crate::types::NodeTuple;
@@ -1,15 +1,14 @@
+// 负责定义 DHT KRPC 消息查询参数和响应字段的反序列化模型
+
 use serde::Deserialize;

 #[derive(Deserialize, Debug, Clone)]
-#[allow(dead_code)]
 /// Decoded KRPC envelope.
 pub struct DhtMessage {
    /// Transaction ID bytes.
    pub t: serde_bytes::ByteBuf,
-    #[allow(dead_code)]
    /// Message kind (`q`, `r`, or `e`).
    pub y: String,
-    #[allow(dead_code)]
    /// Query method when `y == q`.
    pub q: Option<String>,
    /// Query arguments.
@@ -39,7 +38,6 @@ pub struct DhtArgs {
 /// Supported BEP-5 response fields.
 pub struct DhtResponse {
    #[serde(default)]
-    #[allow(dead_code)]
    /// Responder node ID.
    pub id: Option<serde_bytes::ByteBuf>,
    #[serde(default)]
@@ -1,3 +1,5 @@
+// 负责提供并发可读的 DHT 路由节点快照和距离查询
+
 use crate::types::NodeTuple;
 use rand::seq::IndexedRandom;

@@ -1,3 +1,5 @@
+// 负责收集 DHT 抓取调度网络和 Metadata 运行指标
+
 use std::sync::Arc;
 use std::sync::atomic::{AtomicU32, AtomicU64, AtomicUsize, Ordering};

@@ -1,3 +1,5 @@
+// 负责管理有界 Metadata 任务队列 Peer 竞速和完成回调
+
 use crate::metadata::{FetchedMetadata, MetadataFetchOutcome, RbitFetcher};
 use crate::peer_lookup::PeerLookupRequest;
 #[cfg(test)]
@@ -1,3 +1,5 @@
+// 负责复用有界 UDP 接收缓冲区并控制内存占用
+
 //! UDP 收包缓冲区池：避免每包 `to_owned()` 拷贝。
 //!
 //! 单线程 listener 从池中取出固定大小缓冲区，`recv_from` 直接写入；
@@ -1,3 +1,5 @@
+// 负责将 UDP 数据包分发到有界工作队列并处理过载
+
 use crate::error::DHTError;
 use crate::runtime_stats::DhtRuntimeStats;
 use crate::udp_buffer::{MAX_DHT_UDP_PACKET, UdpBufferPool, UdpPacket};
@@ -22,7 +22,6 @@ hex = "0.4"
 rocksdb = { version = "0.24.0", default-features = false, features = ["bindgen-runtime", "lz4"], optional = true }
 rmp-serde = "1.3"
 serde.workspace = true
-serde_json = "1.0"
 tantivy = "0.26.1"
 thiserror.workspace = true
 tokio = { workspace = true, features = ["macros", "rt-multi-thread", "signal", "sync", "time"] }
@@ -33,5 +32,6 @@ tracing-subscriber = { workspace = true, features = ["env-filter", "fmt"] }
 unicode-normalization = "0.1"

 [dev-dependencies]
+serde_json = "1.0"
 tempfile = "3.27"
 tower = { version = "0.5", features = ["util"] }
@@ -116,12 +116,22 @@ GET /ready
 GET /stats
 GET /search?q=ubuntu&offset=0&limit=20
 GET /search?q=&min_size=1048576&max_size=10737418240&extension=mkv
+GET /search?q=流浪地球&min_files=1&availability=active&heat=hot&sort=heat
+GET /contents/{content_key}?offset=0&limit=20
 GET /torrents/{infohash}
 ```

 `limit` 被限制在 1 到 100 之间且 `offset` 最大为 10000

-搜索和大小扩展名过滤由 Tantivy 索引执行不会把全部记录加载到内存过滤
+搜索支持中文英文数字和文件名片段匹配
+
+过滤参数还包括 `min_files` `max_files` `first_seen_after` `first_seen_before` `last_seen_after` `last_seen_before` `availability` 和 `heat`
+
+排序支持 `relevance` `latest` `oldest` `heat` `size_desc` `size_asc` 和 `discoveries`
+
+有关键词时默认按相关性排序 空查询默认按最近发现排序
+
+搜索结果按 `content_key` 精确折叠并通过 `variant_count` 返回变体数量 `/contents/{content_key}` 用于分页查看全部 infohash 和磁力链接

 搜索响应包含 `heat` 和 `availability` 摘要 详情响应包含完整验证时间 Peer 数和连续失败次数

@@ -129,8 +139,8 @@ GET /torrents/{infohash}

 RocksDB 是权威数据源而 Tantivy 是可重建索引

-当 Tantivy 目录不存在时应用会把全部 RocksDB 记录重新标记为待索引并自动完成全量重建
+当 Tantivy 目录不存在或结构不匹配时应用会直接创建新索引并从 RocksDB 的内容组状态完成全量重建

-RocksDB v1 会在启动时通过可恢复迁移升级到 v2 旧 Tantivy schema 会先备份再从 RocksDB 重建 重建提交完成后清理备份
+项目当前处于开发阶段 持久化结构变化时直接清理测试数据重新采集 不维护旧测试数据库兼容层

 正常退出会先停止 DHT 再排空持久化队列提交剩余索引最后关闭 HTTP 服务
@@ -10,14 +10,17 @@ use axum::{
 };
 use dht_search::{
    domain::InfoHash,
-    search::{SearchOptions, SearchPage},
+    search::{SearchOptions, SearchPage, SearchSort},
    storage::VerificationPriority,
 };

 use super::{
    ApiState,
-    request::SearchRequest,
-    response::{ErrorResponse, StatsResponse, StatusResponse, TorrentResponse},
+    request::{ContentVariantsRequest, SearchRequest},
+    response::{
+        ContentVariantsResponse, ErrorResponse, StatsResponse, StatusResponse, TorrentResponse,
+        TorrentVariantResponse,
+    },
 };

 pub(crate) async fn health() -> Json<StatusResponse> {
@@ -80,21 +83,64 @@ pub(crate) async fn search(
    if request.q.len() > 512 {
        return Err(ApiError::bad_request("查询文本不能超过 512 字节"));
    }
-    if request
-        .min_size
-        .zip(request.max_size)
-        .is_some_and(|(min, max)| min > max)
-    {
-        return Err(ApiError::bad_request("min_size 不能大于 max_size"));
-    }
+    validate_range(request.min_size, request.max_size, "min_size", "max_size")?;
+    validate_range(
+        request.min_files,
+        request.max_files,
+        "min_files",
+        "max_files",
+    )?;
+    validate_range(
+        request.first_seen_after,
+        request.first_seen_before,
+        "first_seen_after",
+        "first_seen_before",
+    )?;
+    validate_range(
+        request.last_seen_after,
+        request.last_seen_before,
+        "last_seen_after",
+        "last_seen_before",
+    )?;
+    let mut query = request.q;
+    let content_key = if let Ok(info_hash) = InfoHash::from_str(query.trim()) {
+        let repository = state.repository.clone();
+        let record = tokio::task::spawn_blocking(move || repository.get(info_hash))
+            .await
+            .map_err(|error| ApiError::internal(error.to_string()))?
+            .map_err(|error| ApiError::internal(error.to_string()))?;
+        let Some(record) = record else {
+            return Ok(Json(SearchPage {
+                total: 0,
+                offset: request.offset.min(10_000),
+                limit: request.limit.clamp(1, 100),
+                hits: Vec::new(),
+                sort: request.sort.unwrap_or(SearchSort::Relevance),
+            }));
+        };
+        query.clear();
+        Some(record.content_key)
+    } else {
+        None
+    };
    let page = tokio::task::spawn_blocking(move || {
        state.search.search_with(SearchOptions {
-            query: request.q,
+            query,
            offset: request.offset,
            limit: request.limit,
            min_size: request.min_size,
            max_size: request.max_size,
            extension: request.extension,
+            min_files: request.min_files,
+            max_files: request.max_files,
+            first_seen_after: request.first_seen_after,
+            first_seen_before: request.first_seen_before,
+            last_seen_after: request.last_seen_after,
+            last_seen_before: request.last_seen_before,
+            availability: request.availability,
+            heat: request.heat,
+            sort: request.sort,
+            content_key,
        })
    })
    .await
@@ -113,6 +159,47 @@ pub(crate) async fn search(
    Ok(Json(page))
 }

+pub(crate) async fn content_variants(
+    State(state): State<ApiState>,
+    Path(content_key): Path<String>,
+    Query(request): Query<ContentVariantsRequest>,
+) -> Result<Json<ContentVariantsResponse>, ApiError> {
+    if request.limit == 0 || request.limit > 100 {
+        return Err(ApiError::bad_request("limit 必须在 1 到 100 之间"));
+    }
+    if request.offset > 10_000 {
+        return Err(ApiError::bad_request("offset 不能超过 10000"));
+    }
+    let decoded = hex::decode(&content_key)
+        .map_err(|_| ApiError::bad_request("content_key 必须是六十四位十六进制字符串"))?;
+    let content_key_bytes: [u8; 32] = decoded
+        .try_into()
+        .map_err(|_| ApiError::bad_request("content_key 必须是六十四位十六进制字符串"))?;
+    let repository = state.repository.clone();
+    let offset = request.offset;
+    let limit = request.limit;
+    let variants = tokio::task::spawn_blocking(move || {
+        repository.content_variants(&content_key_bytes, offset, limit)
+    })
+    .await
+    .map_err(|error| ApiError::internal(error.to_string()))?
+    .map_err(|error| ApiError::internal(error.to_string()))?;
+    if variants.total == 0 {
+        return Err(ApiError::not_found("没有找到该 content_key"));
+    }
+    Ok(Json(ContentVariantsResponse {
+        content_key: content_key.to_lowercase(),
+        total: variants.total,
+        offset,
+        limit,
+        variants: variants
+            .records
+            .into_iter()
+            .map(TorrentVariantResponse::from)
+            .collect(),
+    }))
+}
+
 pub(crate) async fn torrent(
    State(state): State<ApiState>,
    Path(info_hash): Path<String>,
@@ -161,6 +248,20 @@ impl ApiError {
    }
 }

+fn validate_range(
+    min: Option<u64>,
+    max: Option<u64>,
+    min_name: &str,
+    max_name: &str,
+) -> Result<(), ApiError> {
+    if min.zip(max).is_some_and(|(min, max)| min > max) {
+        return Err(ApiError::bad_request(format!(
+            "{min_name} 不能大于 {max_name}"
+        )));
+    }
+    Ok(())
+}
+
 impl IntoResponse for ApiError {
    fn into_response(self) -> Response {
        (
@@ -41,6 +41,7 @@ fn router(state: ApiState) -> Router {
        .route("/ready", get(handlers::ready))
        .route("/stats", get(handlers::stats))
        .route("/search", get(handlers::search))
+        .route("/contents/{content_key}", get(handlers::content_variants))
        .route("/torrents/{info_hash}", get(handlers::torrent))
        .with_state(state)
 }
@@ -87,8 +88,12 @@ mod tests {
        .unwrap();
        record.availability = dht_search::domain::Availability::default();
        repository.upsert(record.clone()).unwrap();
+        let mut variant = record.clone();
+        variant.info_hash = InfoHash::from_bytes([2; 20]);
+        variant.name = "Example Movie Alternate".into();
+        repository.upsert(variant).unwrap();
        let search = SearchEngine::open(directory.path().join("tantivy")).unwrap();
-        search.index_records(std::slice::from_ref(&record)).unwrap();
+        search.index_pending(repository.as_ref(), 10, 20).unwrap();
        let repository_trait: Arc<dyn TorrentRepository> = repository.clone();
        let persistence = PersistencePipeline::start(repository_trait.clone(), 4);
        let verification = VerificationIngress::for_test(repository.clone(), 10);
@@ -117,8 +122,66 @@ mod tests {
        assert_eq!(json["hits"][0]["name"], "Example Movie");
        assert!(json["hits"][0]["heat"]["score"].is_number());
        assert_eq!(json["hits"][0]["availability"]["status"], "unknown");
+        assert_eq!(json["hits"][0]["variant_count"], 2);
        assert_eq!(repository.verification_queue_len().unwrap(), 1);

+        let response = app
+            .clone()
+            .oneshot(
+                Request::builder()
+                    .uri(format!("/search?q={}", record.info_hash))
+                    .body(Body::empty())
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(response.status(), StatusCode::OK);
+        let json: serde_json::Value =
+            serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap())
+                .unwrap();
+        assert_eq!(json["total"], 1);
+
+        let response = app
+            .clone()
+            .oneshot(
+                Request::builder()
+                    .uri(format!("/contents/{}", hex::encode(record.content_key)))
+                    .body(Body::empty())
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(response.status(), StatusCode::OK);
+        let json: serde_json::Value =
+            serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap())
+                .unwrap();
+        assert_eq!(json["total"], 2);
+        assert_eq!(json["variants"].as_array().unwrap().len(), 2);
+
+        let response = app
+            .clone()
+            .oneshot(
+                Request::builder()
+                    .uri("/search?min_files=3&max_files=1")
+                    .body(Body::empty())
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
+
+        let response = app
+            .clone()
+            .oneshot(
+                Request::builder()
+                    .uri("/search?sort=not_a_sort")
+                    .body(Body::empty())
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
+
        let response = app
            .oneshot(
                Request::builder()
@@ -1,5 +1,9 @@
 // 负责定义 HTTP 查询参数和输入校验模型

+use dht_search::{
+    domain::{AvailabilityStatus, HeatLevel},
+    search::SearchSort,
+};
 use serde::Deserialize;

 fn default_limit() -> usize {
@@ -17,4 +21,21 @@ pub(crate) struct SearchRequest {
    pub(crate) min_size: Option<u64>,
    pub(crate) max_size: Option<u64>,
    pub(crate) extension: Option<String>,
+    pub(crate) min_files: Option<u64>,
+    pub(crate) max_files: Option<u64>,
+    pub(crate) first_seen_after: Option<u64>,
+    pub(crate) first_seen_before: Option<u64>,
+    pub(crate) last_seen_after: Option<u64>,
+    pub(crate) last_seen_before: Option<u64>,
+    pub(crate) availability: Option<AvailabilityStatus>,
+    pub(crate) heat: Option<HeatLevel>,
+    pub(crate) sort: Option<SearchSort>,
+}
+
+#[derive(Debug, Deserialize)]
+pub(crate) struct ContentVariantsRequest {
+    #[serde(default)]
+    pub(crate) offset: usize,
+    #[serde(default = "default_limit")]
+    pub(crate) limit: usize,
 }
@@ -62,6 +62,48 @@ pub(crate) struct TorrentResponse {
    pub(crate) availability: Availability,
 }

+#[derive(Debug, Serialize)]
+pub(crate) struct ContentVariantsResponse {
+    pub(crate) content_key: String,
+    pub(crate) total: u64,
+    pub(crate) offset: usize,
+    pub(crate) limit: usize,
+    pub(crate) variants: Vec<TorrentVariantResponse>,
+}
+
+#[derive(Debug, Serialize)]
+pub(crate) struct TorrentVariantResponse {
+    pub(crate) info_hash: String,
+    pub(crate) magnet_link: String,
+    pub(crate) name: String,
+    pub(crate) total_size: u64,
+    pub(crate) file_count: usize,
+    pub(crate) first_seen: u64,
+    pub(crate) last_seen: u64,
+    pub(crate) seen_count: u64,
+    pub(crate) heat: Heat,
+    pub(crate) availability: Availability,
+}
+
+impl From<TorrentRecord> for TorrentVariantResponse {
+    fn from(record: TorrentRecord) -> Self {
+        let info_hash = record.info_hash.to_string();
+        let heat = record.heat(unix_timestamp());
+        Self {
+            magnet_link: format!("magnet:?xt=urn:btih:{info_hash}"),
+            info_hash,
+            name: record.name,
+            total_size: record.total_size,
+            file_count: record.files.len(),
+            first_seen: record.first_seen,
+            last_seen: record.last_seen,
+            seen_count: record.seen_count,
+            heat,
+            availability: record.availability,
+        }
+    }
+}
+
 impl From<TorrentRecord> for TorrentResponse {
    fn from(record: TorrentRecord) -> Self {
        let info_hash = record.info_hash.to_string();
@@ -259,8 +259,11 @@ async fn run_indexer(
            _ = ticker.tick() => {
                let indexed = index_one_batch(repository.clone(), search.clone(), batch_size).await;
                match indexed {
-                    Ok(count) if count > 0 => tracing::debug!(count, "搜索索引已提交"),
-                    Ok(_) => search.cleanup_rebuild_backup().map_err(|error| error.to_string())?,
+                    Ok(count) => {
+                        if count > 0 {
+                            tracing::debug!(count, "搜索索引已提交");
+                        }
+                    }
                    Err(error) => {
                        let _ = fatal.send(error.clone());
                        return Err(error);
@@ -276,9 +279,6 @@ async fn run_indexer(
            break;
        }
    }
-    search
-        .cleanup_rebuild_backup()
-        .map_err(|error| error.to_string())?;
    Ok(())
 }

@@ -164,9 +164,10 @@ mod tests {

    use dht_crawler::FileInfo;
    use dht_search::{
-        domain::{IndexState, InfoHash, VerificationResult},
+        domain::{ContentGroup, InfoHash, VerificationResult},
        storage::{
-            UpsertOutcome, VerificationEnqueueOutcome, VerificationPriority, VerificationRequest,
+            ContentGroupTask, ContentVariants, UpsertOutcome, VerificationEnqueueOutcome,
+            VerificationPriority, VerificationRequest,
        },
    };

@@ -197,22 +198,38 @@ mod tests {
            Ok(false)
        }

-        fn by_content_key(&self, _: &[u8; 32], _: usize) -> Result<Vec<InfoHash>, StorageError> {
+        fn pending_index(&self, _: usize) -> Result<Vec<ContentGroupTask>, StorageError> {
            Ok(Vec::new())
        }

-        fn pending_index(&self, _: usize) -> Result<Vec<InfoHash>, StorageError> {
-            Ok(Vec::new())
+        fn content_group(
+            &self,
+            _: &[u8; 32],
+            _: u64,
+        ) -> Result<Option<ContentGroup>, StorageError> {
+            Ok(None)
        }

-        fn mark_indexed(&self, _: InfoHash, _: u64) -> Result<(), StorageError> {
-            Ok(())
+        fn mark_indexed(&self, _: &[u8; 32], _: u64) -> Result<bool, StorageError> {
+            Ok(true)
        }

        fn prepare_full_reindex(&self) -> Result<u64, StorageError> {
            Ok(0)
        }

+        fn content_variants(
+            &self,
+            _: &[u8; 32],
+            _: usize,
+            _: usize,
+        ) -> Result<ContentVariants, StorageError> {
+            Ok(ContentVariants {
+                total: 0,
+                records: Vec::new(),
+            })
+        }
+
        fn enqueue_verification(
            &self,
            _: &[InfoHash],
@@ -268,6 +285,5 @@ mod tests {
        pipeline.close_and_join().await.unwrap();
        let records = repository.records.lock().unwrap();
        assert_eq!(records.len(), 1);
-        assert_eq!(records[0].index_state, IndexState::Pending);
    }
 }
@@ -4,14 +4,14 @@ use unicode_normalization::UnicodeNormalization;

 use super::torrent::{TorrentFile, TorrentRecordError};

-const FINGERPRINT_VERSION: &[u8] = b"dht-search-content-v1\0";
+const FINGERPRINT_DOMAIN: &[u8] = b"dht-search-content\0";

 pub fn content_key(files: &[TorrentFile]) -> Result<[u8; 32], TorrentRecordError> {
    let mut normalized = normalize_files(files)?;
    normalized.sort_unstable();

    let mut hasher = blake3::Hasher::new();
-    hasher.update(FINGERPRINT_VERSION);
+    hasher.update(FINGERPRINT_DOMAIN);
    hasher.update(&(normalized.len() as u64).to_be_bytes());
    for (path, size) in normalized {
        hasher.update(&(path.len() as u64).to_be_bytes());
@@ -4,9 +4,10 @@ mod fingerprint;
 mod torrent;

 pub use fingerprint::content_key;
+pub(crate) use torrent::ContentGroupBuilder;
 #[cfg(test)]
 pub(crate) use torrent::test_record;
 pub use torrent::{
-    Availability, AvailabilityStatus, Heat, HeatLevel, IndexState, InfoHash, RECORD_SCHEMA_VERSION,
-    TorrentFile, TorrentRecord, TorrentRecordError, VerificationResult,
+    Availability, AvailabilityStatus, ContentGroup, Heat, HeatLevel, InfoHash, TorrentFile,
+    TorrentRecord, TorrentRecordError, VerificationResult,
 };
@@ -2,16 +2,15 @@

 use dht_crawler::TorrentInfo;
 use serde::{Deserialize, Serialize};
-use std::{fmt, str::FromStr};
+use std::{cmp::Ordering, fmt, str::FromStr};

 use super::fingerprint::content_key;

-pub const RECORD_SCHEMA_VERSION: u16 = 2;
 const MAX_STORED_PEERS: usize = 32;
 const ACTIVITY_SCALE: u64 = 1_000;
 const ACTIVITY_HALF_LIFE_SECS: f64 = 86_400.0;

-#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
 pub struct InfoHash([u8; 20]);

 impl InfoHash {
@@ -50,12 +49,6 @@ pub struct TorrentFile {
    pub size: u64,
 }

-#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
-pub enum IndexState {
-    Pending,
-    Indexed { indexed_at: u64 },
-}
-
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
 #[serde(rename_all = "snake_case")]
 pub enum AvailabilityStatus {
@@ -83,7 +76,7 @@ pub struct VerificationResult {
    pub reachable_peers: u32,
 }

-#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
 pub enum HeatLevel {
    Hot,
@@ -98,6 +91,160 @@ pub struct Heat {
    pub level: HeatLevel,
 }

+#[derive(Debug, Clone, PartialEq)]
+pub struct ContentGroup {
+    pub content_key: [u8; 32],
+    pub representative: TorrentRecord,
+    pub aliases: Vec<String>,
+    pub variant_count: u64,
+    pub first_seen: u64,
+    pub last_seen: u64,
+    pub seen_count: u64,
+    pub heat: Heat,
+    pub availability: Availability,
+}
+
+pub(crate) struct ContentGroupBuilder {
+    content_key: [u8; 32],
+    now: u64,
+    representative: Option<TorrentRecord>,
+    aliases: Vec<(RepresentativeRank, String)>,
+    variant_count: u64,
+    first_seen: u64,
+    last_seen: u64,
+    seen_count: u64,
+    heat: Heat,
+    availability: Availability,
+    availability_initialized: bool,
+}
+
+impl ContentGroupBuilder {
+    pub(crate) fn new(content_key: [u8; 32], now: u64) -> Self {
+        Self {
+            content_key,
+            now,
+            representative: None,
+            aliases: Vec::new(),
+            variant_count: 0,
+            first_seen: u64::MAX,
+            last_seen: 0,
+            seen_count: 0,
+            heat: Heat::from_score(0),
+            availability: Availability::default(),
+            availability_initialized: false,
+        }
+    }
+
+    pub(crate) fn push(&mut self, record: TorrentRecord) {
+        if record.content_key != self.content_key {
+            return;
+        }
+        self.variant_count = self.variant_count.saturating_add(1);
+        self.first_seen = self.first_seen.min(record.first_seen);
+        self.last_seen = self.last_seen.max(record.last_seen);
+        self.seen_count = self.seen_count.saturating_add(record.seen_count);
+        let heat = record.heat(self.now);
+        let rank = representative_rank(&record, heat);
+        if heat.score > self.heat.score {
+            self.heat = heat;
+        }
+        if self.availability_initialized {
+            merge_availability(&mut self.availability, &record.availability);
+        } else {
+            self.availability = record.availability.clone();
+            self.availability_initialized = true;
+        }
+        if let Some((existing_rank, _)) = self
+            .aliases
+            .iter_mut()
+            .find(|(_, name)| name == &record.name)
+        {
+            *existing_rank = (*existing_rank).max(rank);
+        } else {
+            self.aliases.push((rank, record.name.clone()));
+        }
+        self.aliases
+            .sort_unstable_by_key(|item| std::cmp::Reverse(item.0));
+        self.aliases.truncate(32);
+        let replace = self.representative.as_ref().is_none_or(|current| {
+            rank.cmp(&representative_rank(current, current.heat(self.now))) == Ordering::Greater
+        });
+        if replace {
+            self.representative = Some(record);
+        }
+    }
+
+    pub(crate) fn finish(mut self) -> Option<ContentGroup> {
+        let representative = self.representative?;
+        if !self
+            .aliases
+            .iter()
+            .any(|(_, name)| name == &representative.name)
+        {
+            if self.aliases.len() == 32 {
+                self.aliases.pop();
+            }
+            self.aliases.push((
+                representative_rank(&representative, representative.heat(self.now)),
+                representative.name.clone(),
+            ));
+        }
+        Some(ContentGroup {
+            content_key: self.content_key,
+            representative,
+            aliases: self.aliases.into_iter().map(|(_, name)| name).collect(),
+            variant_count: self.variant_count,
+            first_seen: self.first_seen,
+            last_seen: self.last_seen,
+            seen_count: self.seen_count,
+            heat: self.heat,
+            availability: self.availability,
+        })
+    }
+}
+
+type RepresentativeRank = (u8, u8, u32, u64, u64, std::cmp::Reverse<InfoHash>);
+
+fn representative_rank(record: &TorrentRecord, heat: Heat) -> RepresentativeRank {
+    let availability = match record.availability.status {
+        AvailabilityStatus::Active => 2,
+        AvailabilityStatus::Unknown => 1,
+        AvailabilityStatus::PossiblyStale => 0,
+    };
+    (
+        availability,
+        heat.score,
+        record.availability.reachable_peers,
+        record.last_seen,
+        record.seen_count,
+        std::cmp::Reverse(record.info_hash),
+    )
+}
+
+fn merge_availability(target: &mut Availability, candidate: &Availability) {
+    let target_rank = availability_rank(target.status);
+    let candidate_rank = availability_rank(candidate.status);
+    if candidate_rank > target_rank {
+        target.status = candidate.status;
+    }
+    target.last_verified_at = target.last_verified_at.max(candidate.last_verified_at);
+    target.last_success_at = target.last_success_at.max(candidate.last_success_at);
+    target.discovered_peers = target.discovered_peers.max(candidate.discovered_peers);
+    target.reachable_peers = target.reachable_peers.max(candidate.reachable_peers);
+    target.consecutive_failures = target
+        .consecutive_failures
+        .min(candidate.consecutive_failures);
+    target.next_check_at = target.next_check_at.max(candidate.next_check_at);
+}
+
+fn availability_rank(status: AvailabilityStatus) -> u8 {
+    match status {
+        AvailabilityStatus::Active => 2,
+        AvailabilityStatus::Unknown => 1,
+        AvailabilityStatus::PossiblyStale => 0,
+    }
+}
+
 impl Heat {
    pub fn from_score(score: u8) -> Self {
        let level = match score {
@@ -112,7 +259,6 @@ impl Heat {

 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct TorrentRecord {
-    pub schema_version: u16,
    pub info_hash: InfoHash,
    pub name: String,
    pub total_size: u64,
@@ -123,7 +269,6 @@ pub struct TorrentRecord {
    pub first_seen: u64,
    pub last_seen: u64,
    pub seen_count: u64,
-    pub index_state: IndexState,
    #[serde(default)]
    pub availability: Availability,
    #[serde(default = "default_activity_score")]
@@ -147,13 +292,6 @@ impl TorrentRecord {
        }
    }

-    pub fn normalize_schema(&mut self) {
-        if self.activity_updated_at == 0 {
-            self.activity_updated_at = self.last_seen;
-        }
-        self.schema_version = RECORD_SCHEMA_VERSION;
-    }
-
    pub fn apply_verification(&mut self, result: VerificationResult) {
        self.availability.last_verified_at = Some(result.verified_at);
        self.availability.discovered_peers = result.discovered_peers;
@@ -171,7 +309,6 @@ impl TorrentRecord {
                .verified_at
                .saturating_add(failure_retry_secs(self.availability.consecutive_failures));
        }
-        self.index_state = IndexState::Pending;
    }

    pub fn heat(&self, now: u64) -> Heat {
@@ -279,7 +416,6 @@ impl TryFrom<TorrentInfo> for TorrentRecord {
        };

        Ok(Self {
-            schema_version: RECORD_SCHEMA_VERSION,
            info_hash,
            name: info.name,
            total_size: info.total_size,
@@ -290,7 +426,6 @@ impl TryFrom<TorrentInfo> for TorrentRecord {
            first_seen: info.timestamp,
            last_seen: info.timestamp,
            seen_count: 1,
-            index_state: IndexState::Pending,
            availability,
            activity_score_millis: ACTIVITY_SCALE,
            activity_updated_at: info.timestamp,
@@ -321,7 +456,6 @@ pub(crate) fn test_record(hash_byte: u8, timestamp: u64) -> TorrentRecord {
        size: 42,
    }];
    TorrentRecord {
-        schema_version: RECORD_SCHEMA_VERSION,
        info_hash: InfoHash::from_bytes([hash_byte; 20]),
        name: "Example".to_owned(),
        total_size: 42,
@@ -332,7 +466,6 @@ pub(crate) fn test_record(hash_byte: u8, timestamp: u64) -> TorrentRecord {
        first_seen: timestamp,
        last_seen: timestamp,
        seen_count: 1,
-        index_state: IndexState::Pending,
        availability: Availability::default(),
        activity_score_millis: ACTIVITY_SCALE,
        activity_updated_at: timestamp,
@@ -414,4 +547,28 @@ mod tests {
        assert_eq!(record.availability.last_verified_at, Some(100));
        assert_eq!(record.availability.next_check_at, 86_500);
    }
+
+    #[test]
+    fn content_group_uses_best_variant_and_aggregates_activity() {
+        let mut stale = test_record(1, 10);
+        stale.name = "旧名称".into();
+        stale.availability.status = AvailabilityStatus::PossiblyStale;
+        let mut active = test_record(2, 20);
+        active.name = "流浪地球 S01E03".into();
+        active.availability.status = AvailabilityStatus::Active;
+        active.availability.reachable_peers = 2;
+        active.seen_count = 3;
+
+        let mut builder = ContentGroupBuilder::new(stale.content_key, 20);
+        builder.push(stale);
+        builder.push(active.clone());
+        let group = builder.finish().unwrap();
+
+        assert_eq!(group.representative.info_hash, active.info_hash);
+        assert_eq!(group.variant_count, 2);
+        assert_eq!(group.first_seen, 10);
+        assert_eq!(group.last_seen, 20);
+        assert_eq!(group.seen_count, 4);
+        assert_eq!(group.availability.status, AvailabilityStatus::Active);
+    }
 }
@@ -3,26 +3,28 @@
 use std::{
    collections::BTreeSet,
    ops::Bound,
-    path::{Path, PathBuf},
+    path::Path,
    sync::{Arc, Mutex},
-    time::{SystemTime, UNIX_EPOCH},
 };

 use tantivy::{
-    Index, IndexReader, IndexWriter, ReloadPolicy, TantivyDocument, Term,
+    DocAddress, Index, IndexReader, IndexWriter, Order, ReloadPolicy, Searcher, TantivyDocument,
+    Term,
    collector::{Count, TopDocs},
    directory::MmapDirectory,
-    query::{AllQuery, BooleanQuery, Query, QueryParser, RangeQuery, TermQuery},
+    query::{AllQuery, BooleanQuery, BoostQuery, Occur, Query, RangeQuery, TermQuery},
    schema::{IndexRecordOption, Value},
+    tokenizer::{LowerCaser, NgramTokenizer, TextAnalyzer},
 };
+use unicode_normalization::UnicodeNormalization;

-use crate::domain::{AvailabilityStatus, Heat, TorrentRecord};
+use crate::domain::{AvailabilityStatus, ContentGroup, Heat, HeatLevel, TorrentRecord};
 use crate::storage::TorrentRepository;

 use super::{
    IndexingError, SearchError,
-    query::{AvailabilitySummary, SearchHit, SearchOptions, SearchPage},
-    schema::{SearchFields, build_schema},
+    query::{AvailabilitySummary, SearchHit, SearchOptions, SearchPage, SearchSort},
+    schema::{MIXED_NGRAM_TOKENIZER, SearchFields, build_schema},
 };

 const INDEX_WRITER_MEMORY_BYTES: usize = 64 * 1024 * 1024;
@@ -35,11 +37,9 @@ pub struct SearchEngine {
 }

 struct SearchInner {
-    index: Index,
    reader: IndexReader,
    writer: Mutex<IndexWriter>,
    fields: SearchFields,
-    backup_path: Mutex<Option<PathBuf>>,
 }

 impl SearchEngine {
@@ -57,19 +57,16 @@ impl SearchEngine {
        let exists =
            Index::exists(&directory).map_err(|error| SearchError::Directory(error.to_string()))?;
        let mut created = !exists;
-        let mut backup_path = None;
        let index = if exists {
            let index = Index::open(directory)?;
            if index.schema() != expected_schema {
                drop(index);
-                let backup = backup_path_for(&path);
-                std::fs::rename(&path, &backup)
+                std::fs::remove_dir_all(&path)
                    .map_err(|error| SearchError::Directory(error.to_string()))?;
                std::fs::create_dir_all(&path)
                    .map_err(|error| SearchError::Directory(error.to_string()))?;
                directory = MmapDirectory::open(&path)
                    .map_err(|error| SearchError::Directory(error.to_string()))?;
-                backup_path = Some(backup);
                created = true;
                Index::open_or_create(directory, expected_schema)?
            } else {
@@ -78,6 +75,10 @@ impl SearchEngine {
        } else {
            Index::open_or_create(directory, expected_schema)?
        };
+        let analyzer = TextAnalyzer::builder(NgramTokenizer::all_ngrams(1, 20)?)
+            .filter(LowerCaser)
+            .build();
+        index.tokenizers().register(MIXED_NGRAM_TOKENIZER, analyzer);
        let reader = index
            .reader_builder()
            .reload_policy(ReloadPolicy::OnCommitWithDelay)
@@ -86,34 +87,17 @@ impl SearchEngine {
        Ok((
            Self {
                inner: Arc::new(SearchInner {
-                    index,
                    reader,
                    writer: Mutex::new(writer),
                    fields,
-                    backup_path: Mutex::new(backup_path),
                }),
            },
            created,
        ))
    }

-    pub fn cleanup_rebuild_backup(&self) -> Result<(), SearchError> {
-        let mut backup = self
-            .inner
-            .backup_path
-            .lock()
-            .unwrap_or_else(std::sync::PoisonError::into_inner);
-        if let Some(path) = backup.take()
-            && path.exists()
-        {
-            std::fs::remove_dir_all(path)
-                .map_err(|error| SearchError::Directory(error.to_string()))?;
-        }
-        Ok(())
-    }
-
-    pub fn index_records(&self, records: &[TorrentRecord]) -> Result<(), SearchError> {
-        if records.is_empty() {
+    fn index_groups(&self, groups: &[ContentGroup]) -> Result<(), SearchError> {
+        if groups.is_empty() {
            return Ok(());
        }
        let fields = self.inner.fields;
@@ -122,12 +106,12 @@ impl SearchEngine {
            .writer
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
-        for record in records {
+        for group in groups {
            writer.delete_term(Term::from_field_text(
-                fields.info_hash,
-                &record.info_hash.to_string(),
+                fields.content_key,
+                &hex::encode(group.content_key),
            ));
-            writer.add_document(document(record, fields))?;
+            writer.add_document(document(group, fields))?;
        }
        writer.commit()?;
        self.inner.reader.reload()?;
@@ -142,20 +126,20 @@ impl SearchEngine {
        &self,
        repository: &dyn TorrentRepository,
        limit: usize,
-        indexed_at: u64,
+        now: u64,
    ) -> Result<usize, IndexingError> {
-        let hashes = repository.pending_index(limit)?;
-        let mut records = Vec::with_capacity(hashes.len());
-        for info_hash in hashes {
-            if let Some(record) = repository.get(info_hash)? {
-                records.push(record);
+        let tasks = repository.pending_index(limit)?;
+        let mut groups = Vec::with_capacity(tasks.len());
+        for task in &tasks {
+            if let Some(group) = repository.content_group(&task.content_key, now)? {
+                groups.push(group);
            }
        }
-        self.index_records(&records)?;
-        for record in &records {
-            repository.mark_indexed(record.info_hash, indexed_at)?;
+        self.index_groups(&groups)?;
+        for task in &tasks {
+            repository.mark_indexed(&task.content_key, task.revision)?;
        }
-        Ok(records.len())
+        Ok(groups.len())
    }

    pub fn search(
@@ -181,23 +165,38 @@ impl SearchEngine {
        if query_text.is_empty() || query_text == "*" {
            clauses.push(Box::new(AllQuery));
        } else {
-            let parser = QueryParser::for_index(
-                &self.inner.index,
-                vec![fields.name, fields.files_text, fields.info_hash],
-            );
-            clauses.push(parser.parse_query(query_text)?);
+            clauses.push(text_query(query_text, fields));
        }
-        if options.min_size.is_some() || options.max_size.is_some() {
-            let lower = options
-                .min_size
-                .map(|value| Bound::Included(Term::from_field_u64(fields.total_size, value)))
-                .unwrap_or(Bound::Unbounded);
-            let upper = options
-                .max_size
-                .map(|value| Bound::Included(Term::from_field_u64(fields.total_size, value)))
-                .unwrap_or(Bound::Unbounded);
-            clauses.push(Box::new(RangeQuery::new(lower, upper)));
+        if let Some(content_key) = options.content_key {
+            clauses.push(Box::new(TermQuery::new(
+                Term::from_field_text(fields.content_key, &hex::encode(content_key)),
+                IndexRecordOption::Basic,
+            )));
        }
+        add_range(
+            &mut clauses,
+            fields.total_size,
+            options.min_size,
+            options.max_size,
+        );
+        add_range(
+            &mut clauses,
+            fields.file_count,
+            options.min_files,
+            options.max_files,
+        );
+        add_range(
+            &mut clauses,
+            fields.first_seen,
+            options.first_seen_after,
+            options.first_seen_before,
+        );
+        add_range(
+            &mut clauses,
+            fields.last_seen,
+            options.last_seen_after,
+            options.last_seen_before,
+        );
        if let Some(extension) = options.extension {
            let extension = extension.trim().trim_start_matches('.').to_lowercase();
            if !extension.is_empty() {
@@ -207,33 +206,110 @@ impl SearchEngine {
                )));
            }
        }
+        if let Some(status) = options.availability {
+            let value = availability_number(status);
+            add_range(
+                &mut clauses,
+                fields.availability_status,
+                Some(value),
+                Some(value),
+            );
+        }
+        if let Some(level) = options.heat {
+            let (min, max) = match level {
+                HeatLevel::Hot => (75, 100),
+                HeatLevel::Active => (50, 74),
+                HeatLevel::Normal => (25, 49),
+                HeatLevel::Cold => (0, 24),
+            };
+            add_range(&mut clauses, fields.heat_score, Some(min), Some(max));
+        }
        let query: Box<dyn Query> = if clauses.len() == 1 {
            clauses.pop().expect("one query clause exists")
        } else {
            Box::new(BooleanQuery::intersection(clauses))
        };
        let searcher = self.inner.reader.searcher();
-        let (total, documents) = searcher.search(
-            query.as_ref(),
-            &(
-                Count,
-                TopDocs::with_limit(limit)
-                    .and_offset(offset)
-                    .order_by_score(),
-            ),
-        )?;
+        let sort = options.sort.unwrap_or_else(|| {
+            if query_text.is_empty() || query_text == "*" {
+                SearchSort::Latest
+            } else {
+                SearchSort::Relevance
+            }
+        });
+        let total = searcher.search(query.as_ref(), &Count)?;
+        let documents = match sort {
+            SearchSort::Relevance => searcher
+                .search(
+                    query.as_ref(),
+                    &TopDocs::with_limit(limit)
+                        .and_offset(offset)
+                        .order_by_score(),
+                )?
+                .into_iter()
+                .collect(),
+            SearchSort::Latest => sorted_documents(
+                &searcher,
+                query.as_ref(),
+                limit,
+                offset,
+                "last_seen",
+                Order::Desc,
+            )?,
+            SearchSort::Oldest => sorted_documents(
+                &searcher,
+                query.as_ref(),
+                limit,
+                offset,
+                "first_seen",
+                Order::Asc,
+            )?,
+            SearchSort::Heat => sorted_documents(
+                &searcher,
+                query.as_ref(),
+                limit,
+                offset,
+                "heat_score",
+                Order::Desc,
+            )?,
+            SearchSort::SizeDesc => sorted_documents(
+                &searcher,
+                query.as_ref(),
+                limit,
+                offset,
+                "total_size",
+                Order::Desc,
+            )?,
+            SearchSort::SizeAsc => sorted_documents(
+                &searcher,
+                query.as_ref(),
+                limit,
+                offset,
+                "total_size",
+                Order::Asc,
+            )?,
+            SearchSort::Discoveries => sorted_documents(
+                &searcher,
+                query.as_ref(),
+                limit,
+                offset,
+                "seen_count",
+                Order::Desc,
+            )?,
+        };
        let mut hits = Vec::with_capacity(documents.len());
        for (score, address) in documents {
            let document: TantivyDocument = searcher.doc(address)?;
            hits.push(SearchHit {
                info_hash: text(&document, fields.info_hash, "info_hash")?,
-                name: text(&document, fields.name, "name")?,
+                name: text(&document, fields.display_name, "display_name")?,
                total_size: number(&document, fields.total_size, "total_size")?,
                file_count: number(&document, fields.file_count, "file_count")?,
                first_seen: number(&document, fields.first_seen, "first_seen")?,
                last_seen: number(&document, fields.last_seen, "last_seen")?,
                seen_count: number(&document, fields.seen_count, "seen_count")?,
                content_key: text(&document, fields.content_key, "content_key")?,
+                variant_count: number(&document, fields.variant_count, "variant_count")?,
                score,
                heat: Heat::from_score(
                    number(&document, fields.heat_score, "heat_score")?.min(100) as u8,
@@ -262,55 +338,188 @@ impl SearchEngine {
            offset,
            limit,
            hits,
+            sort,
        })
    }
 }

-fn document(record: &TorrentRecord, fields: SearchFields) -> TantivyDocument {
+fn document(group: &ContentGroup, fields: SearchFields) -> TantivyDocument {
+    const MAX_INDEXED_FILES: usize = 512;
+    const MAX_PATH_TEXT_BYTES: usize = 32 * 1024;
+
+    let record = &group.representative;
    let mut document = TantivyDocument::default();
    document.add_text(fields.info_hash, record.info_hash.to_string());
-    document.add_text(fields.name, &record.name);
-    document.add_text(
-        fields.files_text,
-        record
-            .files
-            .iter()
-            .map(|file| file.path.as_str())
-            .collect::<Vec<_>>()
-            .join(" "),
-    );
+    document.add_text(fields.name, normalize_bounded(&record.name, 512));
+    document.add_text(fields.display_name, &record.name);
+    for alias in &group.aliases {
+        document.add_text(fields.aliases, normalize_bounded(alias, 512));
+    }
+    let mut indexed_path_bytes = 0_usize;
+    for file in record.files.iter().take(MAX_INDEXED_FILES) {
+        let path = normalize_bounded(&file.path, 512);
+        if indexed_path_bytes.saturating_add(path.len()) > MAX_PATH_TEXT_BYTES {
+            break;
+        }
+        indexed_path_bytes += path.len();
+        document.add_text(fields.files_text, path);
+    }
    for extension in extensions(record) {
        document.add_text(fields.extensions, extension);
    }
    document.add_u64(fields.total_size, record.total_size);
    document.add_u64(fields.file_count, record.files.len() as u64);
-    document.add_u64(fields.first_seen, record.first_seen);
-    document.add_u64(fields.last_seen, record.last_seen);
-    document.add_u64(fields.seen_count, record.seen_count);
-    document.add_text(fields.content_key, hex::encode(record.content_key));
+    document.add_u64(fields.first_seen, group.first_seen);
+    document.add_u64(fields.last_seen, group.last_seen);
+    document.add_u64(fields.seen_count, group.seen_count);
+    document.add_text(fields.content_key, hex::encode(group.content_key));
    document.add_u64(
        fields.availability_status,
-        match record.availability.status {
-            AvailabilityStatus::Unknown => 0,
-            AvailabilityStatus::Active => 1,
-            AvailabilityStatus::PossiblyStale => 2,
-        },
+        availability_number(group.availability.status),
    );
    document.add_u64(
        fields.reachable_peers,
-        u64::from(record.availability.reachable_peers),
+        u64::from(group.availability.reachable_peers),
    );
    document.add_u64(
        fields.last_verified_at,
-        record.availability.last_verified_at.unwrap_or(0),
-    );
-    document.add_u64(
-        fields.heat_score,
-        u64::from(record.heat(unix_timestamp()).score),
+        group.availability.last_verified_at.unwrap_or(0),
    );
+    document.add_u64(fields.heat_score, u64::from(group.heat.score));
+    document.add_u64(fields.variant_count, group.variant_count);
    document
 }

+fn text_query(query: &str, fields: SearchFields) -> Box<dyn Query> {
+    let terms = query_terms(query);
+    if terms.is_empty() {
+        return Box::new(AllQuery);
+    }
+    let mut required = Vec::with_capacity(terms.len());
+    for term in terms {
+        let mut alternatives: Vec<(Occur, Box<dyn Query>)> = vec![
+            (
+                Occur::Should,
+                Box::new(BoostQuery::new(
+                    Box::new(TermQuery::new(
+                        Term::from_field_text(fields.name, &term),
+                        IndexRecordOption::WithFreqs,
+                    )),
+                    3.0,
+                )),
+            ),
+            (
+                Occur::Should,
+                Box::new(BoostQuery::new(
+                    Box::new(TermQuery::new(
+                        Term::from_field_text(fields.aliases, &term),
+                        IndexRecordOption::WithFreqs,
+                    )),
+                    2.0,
+                )),
+            ),
+            (
+                Occur::Should,
+                Box::new(TermQuery::new(
+                    Term::from_field_text(fields.files_text, &term),
+                    IndexRecordOption::WithFreqs,
+                )),
+            ),
+        ];
+        if term.len() == 40 && term.bytes().all(|byte| byte.is_ascii_hexdigit()) {
+            alternatives.push((
+                Occur::Should,
+                Box::new(TermQuery::new(
+                    Term::from_field_text(fields.info_hash, &term),
+                    IndexRecordOption::Basic,
+                )),
+            ));
+        }
+        required.push((
+            Occur::Must,
+            Box::new(BooleanQuery::new(alternatives)) as Box<dyn Query>,
+        ));
+    }
+    Box::new(BooleanQuery::new(required))
+}
+
+fn query_terms(query: &str) -> Vec<String> {
+    normalize_text(query)
+        .split_whitespace()
+        .flat_map(|part| {
+            let chars: Vec<_> = part.chars().collect();
+            if chars.len() <= 20 {
+                vec![part.to_owned()]
+            } else {
+                chars
+                    .windows(20)
+                    .map(|window| window.iter().collect())
+                    .collect()
+            }
+        })
+        .collect()
+}
+
+fn normalize_text(value: &str) -> String {
+    value.nfkc().collect::<String>().to_lowercase()
+}
+
+fn normalize_bounded(value: &str, max_chars: usize) -> String {
+    value
+        .chars()
+        .take(max_chars)
+        .collect::<String>()
+        .nfkc()
+        .collect::<String>()
+        .to_lowercase()
+}
+
+fn add_range(
+    clauses: &mut Vec<Box<dyn Query>>,
+    field: tantivy::schema::Field,
+    min: Option<u64>,
+    max: Option<u64>,
+) {
+    if min.is_none() && max.is_none() {
+        return;
+    }
+    let lower = min
+        .map(|value| Bound::Included(Term::from_field_u64(field, value)))
+        .unwrap_or(Bound::Unbounded);
+    let upper = max
+        .map(|value| Bound::Included(Term::from_field_u64(field, value)))
+        .unwrap_or(Bound::Unbounded);
+    clauses.push(Box::new(RangeQuery::new(lower, upper)));
+}
+
+fn sorted_documents(
+    searcher: &Searcher,
+    query: &dyn Query,
+    limit: usize,
+    offset: usize,
+    field: &str,
+    order: Order,
+) -> Result<Vec<(f32, DocAddress)>, SearchError> {
+    let documents: Vec<(Option<u64>, DocAddress)> = searcher.search(
+        query,
+        &TopDocs::with_limit(limit)
+            .and_offset(offset)
+            .order_by_fast_field::<u64>(field, order),
+    )?;
+    Ok(documents
+        .into_iter()
+        .map(|(_, address)| (0.0, address))
+        .collect())
+}
+
+fn availability_number(status: AvailabilityStatus) -> u64 {
+    match status {
+        AvailabilityStatus::Unknown => 0,
+        AvailabilityStatus::Active => 1,
+        AvailabilityStatus::PossiblyStale => 2,
+    }
+}
+
 fn availability_status(value: u64) -> AvailabilityStatus {
    match value {
        1 => AvailabilityStatus::Active,
@@ -319,32 +528,19 @@ fn availability_status(value: u64) -> AvailabilityStatus {
    }
 }

+#[cfg(test)]
 fn unix_timestamp() -> u64 {
-    SystemTime::now()
-        .duration_since(UNIX_EPOCH)
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs()
 }

-fn backup_path_for(path: &Path) -> PathBuf {
-    let name = path
-        .file_name()
-        .and_then(|name| name.to_str())
-        .unwrap_or("tantivy");
-    let base = format!("{name}.backup-{}", unix_timestamp());
-    let mut candidate = path.with_file_name(&base);
-    let mut suffix = 1_u32;
-    while candidate.exists() {
-        candidate = path.with_file_name(format!("{base}-{suffix}"));
-        suffix = suffix.saturating_add(1);
-    }
-    candidate
-}
-
 fn extensions(record: &TorrentRecord) -> BTreeSet<String> {
    record
        .files
        .iter()
+        .take(512)
        .filter_map(|file| Path::new(&file.path).extension())
        .filter_map(|extension| extension.to_str())
        .map(str::to_lowercase)
@@ -376,15 +572,17 @@ fn number(

 #[cfg(test)]
 mod tests {
+    use std::collections::BTreeMap;
+
    use tempfile::TempDir;

-    use crate::domain::{IndexState, InfoHash, RECORD_SCHEMA_VERSION, TorrentFile, TorrentRecord};
+    use crate::domain::{ContentGroupBuilder, InfoHash, TorrentFile, TorrentRecord};
+    use crate::storage::{RocksTorrentRepository, TorrentRepository};

    use super::*;

    fn record() -> TorrentRecord {
        TorrentRecord {
-            schema_version: RECORD_SCHEMA_VERSION,
            info_hash: InfoHash::from_bytes([1; 20]),
            name: "Ubuntu Linux 24.04".into(),
            total_size: 42,
@@ -398,21 +596,41 @@ mod tests {
            first_seen: 10,
            last_seen: 20,
            seen_count: 3,
-            index_state: IndexState::Pending,
            availability: crate::domain::Availability::default(),
            activity_score_millis: 1_000,
            activity_updated_at: 20,
        }
    }

+    fn index_records(engine: &SearchEngine, records: &[TorrentRecord]) {
+        let mut grouped: BTreeMap<[u8; 32], Vec<TorrentRecord>> = BTreeMap::new();
+        for record in records {
+            grouped
+                .entry(record.content_key)
+                .or_default()
+                .push(record.clone());
+        }
+        let groups: Vec<_> = grouped
+            .into_iter()
+            .filter_map(|(content_key, records)| {
+                let mut builder = ContentGroupBuilder::new(content_key, unix_timestamp());
+                for record in records {
+                    builder.push(record);
+                }
+                builder.finish()
+            })
+            .collect();
+        engine.index_groups(&groups).unwrap();
+    }
+
    #[test]
    fn record_is_searchable_and_update_is_idempotent() {
        let directory = TempDir::new().unwrap();
        let engine = SearchEngine::open(directory.path()).unwrap();
        let mut record = record();
-        engine.index_records(&[record.clone()]).unwrap();
+        index_records(&engine, &[record.clone()]);
        record.seen_count = 4;
-        engine.index_records(&[record]).unwrap();
+        index_records(&engine, &[record]);

        let page = engine.search("ubuntu", 0, 10).unwrap();
        assert_eq!(page.total, 1);
@@ -424,15 +642,48 @@ mod tests {
    fn file_path_is_searchable() {
        let directory = TempDir::new().unwrap();
        let engine = SearchEngine::open(directory.path()).unwrap();
-        engine.index_records(&[record()]).unwrap();
+        index_records(&engine, &[record()]);
        assert_eq!(engine.search("ubuntu.iso", 0, 10).unwrap().total, 1);
    }

+    #[test]
+    fn mixed_substrings_match_chinese_and_release_names() {
+        let directory = TempDir::new().unwrap();
+        let engine = SearchEngine::open(directory.path()).unwrap();
+        let mut record = record();
+        record.name = "流浪地球 S01E03 1080P".into();
+        record.files[0].path = "影片/流浪地球.第三集.mkv".into();
+        index_records(&engine, &[record]);
+
+        assert_eq!(engine.search("浪地", 0, 10).unwrap().total, 1);
+        assert_eq!(engine.search("01E0", 0, 10).unwrap().total, 1);
+        assert_eq!(engine.search("第三集", 0, 10).unwrap().total, 1);
+        assert_eq!(engine.search("1080p", 0, 10).unwrap().total, 1);
+    }
+
+    #[test]
+    fn equal_content_is_collapsed_and_aliases_remain_searchable() {
+        let directory = TempDir::new().unwrap();
+        let engine = SearchEngine::open(directory.path()).unwrap();
+        let first = record();
+        let mut second = first.clone();
+        second.info_hash = InfoHash::from_bytes([2; 20]);
+        second.name = "Ubuntu Alternate Name".into();
+        second.availability.status = AvailabilityStatus::Active;
+        second.availability.reachable_peers = 1;
+        index_records(&engine, &[first, second.clone()]);
+
+        let page = engine.search("alternate", 0, 10).unwrap();
+        assert_eq!(page.total, 1);
+        assert_eq!(page.hits[0].variant_count, 2);
+        assert_eq!(page.hits[0].info_hash, second.info_hash.to_string());
+    }
+
    #[test]
    fn size_and_extension_filters_use_the_index() {
        let directory = TempDir::new().unwrap();
        let engine = SearchEngine::open(directory.path()).unwrap();
-        engine.index_records(&[record()]).unwrap();
+        index_records(&engine, &[record()]);
        let matching = engine
            .search_with(SearchOptions {
                query: String::new(),
@@ -456,7 +707,43 @@ mod tests {
    }

    #[test]
-    fn incompatible_schema_is_backed_up_and_recreated() {
+    fn filters_and_sorts_use_group_fast_fields() {
+        let directory = TempDir::new().unwrap();
+        let engine = SearchEngine::open(directory.path()).unwrap();
+        let first = record();
+        let mut second = record();
+        second.info_hash = InfoHash::from_bytes([3; 20]);
+        second.content_key = [3; 32];
+        second.name = "New Release".into();
+        second.first_seen = 30;
+        second.last_seen = 40;
+        second.seen_count = 10;
+        second.total_size = 100;
+        second.files.push(TorrentFile {
+            path: "extra.mkv".into(),
+            size: 58,
+        });
+        second.availability.status = AvailabilityStatus::Active;
+        index_records(&engine, &[first, second.clone()]);
+
+        let page = engine
+            .search_with(SearchOptions {
+                query: String::new(),
+                min_files: Some(2),
+                last_seen_after: Some(30),
+                availability: Some(AvailabilityStatus::Active),
+                sort: Some(SearchSort::SizeDesc),
+                limit: 10,
+                ..SearchOptions::default()
+            })
+            .unwrap();
+        assert_eq!(page.total, 1);
+        assert_eq!(page.sort, SearchSort::SizeDesc);
+        assert_eq!(page.hits[0].info_hash, second.info_hash.to_string());
+    }
+
+    #[test]
+    fn incompatible_schema_is_deleted_and_recreated() {
        let directory = TempDir::new().unwrap();
        let index_path = directory.path().join("tantivy");
        std::fs::create_dir(&index_path).unwrap();
@@ -466,20 +753,26 @@ mod tests {

        let (engine, created) = SearchEngine::open_with_status(&index_path).unwrap();
        assert!(created);
-        let backups: Vec<_> = std::fs::read_dir(directory.path())
-            .unwrap()
-            .filter_map(Result::ok)
-            .filter(|entry| {
-                entry
-                    .file_name()
-                    .to_string_lossy()
-                    .starts_with("tantivy.backup-")
-            })
-            .collect();
-        assert_eq!(backups.len(), 1);
-
-        engine.cleanup_rebuild_backup().unwrap();
-        assert!(!backups[0].path().exists());
+        assert_eq!(engine.num_docs(), 0);
        assert!(index_path.exists());
    }
+
+    #[test]
+    fn deleted_index_is_rebuilt_from_rocksdb_content_groups() {
+        let directory = TempDir::new().unwrap();
+        let repository = RocksTorrentRepository::open(directory.path().join("rocks")).unwrap();
+        repository.upsert(record()).unwrap();
+        let index_path = directory.path().join("tantivy");
+        let engine = SearchEngine::open(&index_path).unwrap();
+        engine.index_pending(&repository, 10, 30).unwrap();
+        assert_eq!(engine.search("ubuntu", 0, 10).unwrap().total, 1);
+        drop(engine);
+
+        std::fs::remove_dir_all(&index_path).unwrap();
+        let (rebuilt, created) = SearchEngine::open_with_status(&index_path).unwrap();
+        assert!(created);
+        assert_eq!(repository.prepare_full_reindex().unwrap(), 1);
+        rebuilt.index_pending(&repository, 10, 40).unwrap();
+        assert_eq!(rebuilt.search("ubuntu", 0, 10).unwrap().total, 1);
+    }
 }
@@ -5,18 +5,14 @@ mod query;
 mod schema;

 pub use indexer::SearchEngine;
-pub use query::{AvailabilitySummary, SearchHit, SearchOptions, SearchPage};
+pub use query::{AvailabilitySummary, SearchHit, SearchOptions, SearchPage, SearchSort};

 #[derive(Debug, thiserror::Error)]
 pub enum SearchError {
    #[error("搜索索引操作失败: {0}")]
    Tantivy(#[from] tantivy::TantivyError),
-    #[error("搜索查询无效: {0}")]
-    Query(#[from] tantivy::query::QueryParserError),
    #[error("无法打开搜索索引目录: {0}")]
    Directory(String),
-    #[error("搜索索引 schema 与当前版本不兼容")]
-    IncompatibleSchema,
    #[error("搜索文档缺少字段 {0}")]
    MissingField(&'static str),
 }
@@ -1,8 +1,21 @@
 // 负责构建全文查询过滤排序分页和内容聚合逻辑

-use serde::Serialize;
+use serde::{Deserialize, Serialize};

-use crate::domain::{AvailabilityStatus, Heat};
+use crate::domain::{AvailabilityStatus, Heat, HeatLevel};
+
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Deserialize, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum SearchSort {
+    #[default]
+    Relevance,
+    Latest,
+    Oldest,
+    Heat,
+    SizeDesc,
+    SizeAsc,
+    Discoveries,
+}

 #[derive(Debug, Clone, Default)]
 pub struct SearchOptions {
@@ -12,6 +25,16 @@ pub struct SearchOptions {
    pub min_size: Option<u64>,
    pub max_size: Option<u64>,
    pub extension: Option<String>,
+    pub min_files: Option<u64>,
+    pub max_files: Option<u64>,
+    pub first_seen_after: Option<u64>,
+    pub first_seen_before: Option<u64>,
+    pub last_seen_after: Option<u64>,
+    pub last_seen_before: Option<u64>,
+    pub availability: Option<AvailabilityStatus>,
+    pub heat: Option<HeatLevel>,
+    pub sort: Option<SearchSort>,
+    pub content_key: Option<[u8; 32]>,
 }

 #[derive(Debug, Clone, PartialEq, Serialize)]
@@ -24,6 +47,7 @@ pub struct SearchHit {
    pub last_seen: u64,
    pub seen_count: u64,
    pub content_key: String,
+    pub variant_count: u64,
    pub score: f32,
    pub heat: Heat,
    pub availability: AvailabilitySummary,
@@ -42,4 +66,5 @@ pub struct SearchPage {
    pub offset: usize,
    pub limit: usize,
    pub hits: Vec<SearchHit>,
+    pub sort: SearchSort,
 }
@@ -1,11 +1,17 @@
 // 负责定义 Tantivy 字段分词索引存储和快速字段策略

-use tantivy::schema::{FAST, Field, STORED, STRING, Schema, TEXT};
+use tantivy::schema::{
+    FAST, Field, IndexRecordOption, STORED, STRING, Schema, TextFieldIndexing, TextOptions,
+};
+
+pub(crate) const MIXED_NGRAM_TOKENIZER: &str = "dht_mixed_ngram";

 #[derive(Debug, Clone, Copy)]
 pub(crate) struct SearchFields {
    pub(crate) info_hash: Field,
    pub(crate) name: Field,
+    pub(crate) display_name: Field,
+    pub(crate) aliases: Field,
    pub(crate) files_text: Field,
    pub(crate) extensions: Field,
    pub(crate) total_size: Field,
@@ -18,13 +24,21 @@ pub(crate) struct SearchFields {
    pub(crate) reachable_peers: Field,
    pub(crate) last_verified_at: Field,
    pub(crate) heat_score: Field,
+    pub(crate) variant_count: Field,
 }

 pub(crate) fn build_schema() -> (Schema, SearchFields) {
    let mut builder = Schema::builder();
    let info_hash = builder.add_text_field("info_hash", STRING | STORED);
-    let name = builder.add_text_field("name", TEXT | STORED);
-    let files_text = builder.add_text_field("files_text", TEXT);
+    let indexed_text = TextOptions::default().set_indexing_options(
+        TextFieldIndexing::default()
+            .set_tokenizer(MIXED_NGRAM_TOKENIZER)
+            .set_index_option(IndexRecordOption::WithFreqsAndPositions),
+    );
+    let name = builder.add_text_field("name", indexed_text.clone());
+    let display_name = builder.add_text_field("display_name", STORED);
+    let aliases = builder.add_text_field("aliases", indexed_text.clone());
+    let files_text = builder.add_text_field("files_text", indexed_text);
    let extensions = builder.add_text_field("extensions", STRING);
    let total_size = builder.add_u64_field("total_size", FAST | STORED);
    let file_count = builder.add_u64_field("file_count", FAST | STORED);
@@ -36,12 +50,15 @@ pub(crate) fn build_schema() -> (Schema, SearchFields) {
    let reachable_peers = builder.add_u64_field("reachable_peers", FAST | STORED);
    let last_verified_at = builder.add_u64_field("last_verified_at", FAST | STORED);
    let heat_score = builder.add_u64_field("heat_score", FAST | STORED);
+    let variant_count = builder.add_u64_field("variant_count", FAST | STORED);
    let schema = builder.build();
    (
        schema,
        SearchFields {
            info_hash,
            name,
+            display_name,
+            aliases,
            files_text,
            extensions,
            total_size,
@@ -54,6 +71,7 @@ pub(crate) fn build_schema() -> (Schema, SearchFields) {
            reachable_peers,
            last_verified_at,
            heat_score,
+            variant_count,
        },
    )
 }
@@ -2,12 +2,12 @@

 use crate::domain::InfoHash;

-pub(crate) const DATABASE_SCHEMA_VERSION: u32 = 2;
-pub(crate) const SCHEMA_VERSION_KEY: &[u8] = b"\x00schema-version";
-pub(crate) const MIGRATION_KEY: &[u8] = b"\x00migration-v1-v2";
+pub(crate) const DATABASE_FORMAT_KEY: &[u8] = b"\x00database-format";
+pub(crate) const DATABASE_FORMAT_VALUE: &[u8] = b"dht-search";
 pub(crate) const VERIFICATION_QUEUE_COUNT_KEY: &[u8] = b"\x00verification-queue-count";
 const TORRENT_PREFIX: u8 = b't';
 const CONTENT_PREFIX: u8 = b'c';
+const CONTENT_GROUP_PREFIX: u8 = b'g';
 const PENDING_INDEX_PREFIX: u8 = b'p';
 const VERIFICATION_HIGH_PREFIX: u8 = b'h';
 const VERIFICATION_NORMAL_PREFIX: u8 = b'n';
@@ -18,12 +18,16 @@ pub(crate) fn torrent_key(info_hash: InfoHash) -> [u8; 1 + InfoHash::BYTE_LEN] {
    prefixed_info_hash(TORRENT_PREFIX, info_hash)
 }

-pub(crate) fn torrent_prefix() -> [u8; 1] {
-    [TORRENT_PREFIX]
+pub(crate) fn content_group_key(content_key: &[u8; 32]) -> [u8; 1 + 32] {
+    prefixed_content_key(CONTENT_GROUP_PREFIX, content_key)
 }

-pub(crate) fn pending_index_key(info_hash: InfoHash) -> [u8; 1 + InfoHash::BYTE_LEN] {
-    prefixed_info_hash(PENDING_INDEX_PREFIX, info_hash)
+pub(crate) fn content_group_prefix() -> [u8; 1] {
+    [CONTENT_GROUP_PREFIX]
+}
+
+pub(crate) fn pending_index_key(content_key: &[u8; 32]) -> [u8; 1 + 32] {
+    prefixed_content_key(PENDING_INDEX_PREFIX, content_key)
 }

 pub(crate) fn pending_index_prefix() -> [u8; 1] {
@@ -123,12 +127,18 @@ pub(crate) fn decode_content_member_info_hash(
    Some(InfoHash::from_bytes(bytes))
 }

-pub(crate) fn decode_pending_info_hash(key: &[u8]) -> Option<InfoHash> {
-    if key.len() != 1 + InfoHash::BYTE_LEN || key.first().copied() != Some(PENDING_INDEX_PREFIX) {
+pub(crate) fn decode_pending_content_key(key: &[u8]) -> Option<[u8; 32]> {
+    if key.len() != 1 + 32 || key.first().copied() != Some(PENDING_INDEX_PREFIX) {
        return None;
    }
-    let bytes: [u8; InfoHash::BYTE_LEN] = key[1..].try_into().ok()?;
-    Some(InfoHash::from_bytes(bytes))
+    key[1..].try_into().ok()
+}
+
+fn prefixed_content_key(prefix: u8, content_key: &[u8; 32]) -> [u8; 1 + 32] {
+    let mut key = [0_u8; 1 + 32];
+    key[0] = prefix;
+    key[1..].copy_from_slice(content_key);
+    key
 }

 fn prefixed_info_hash(prefix: u8, info_hash: InfoHash) -> [u8; 1 + InfoHash::BYTE_LEN] {
@@ -155,11 +165,11 @@ mod tests {
    use super::*;

    #[test]
-    fn pending_key_round_trips_infohash() {
-        let hash = InfoHash::from_bytes([7; 20]);
+    fn pending_key_round_trips_content_key() {
+        let content_key = [7; 32];
        assert_eq!(
-            decode_pending_info_hash(&pending_index_key(hash)),
-            Some(hash)
+            decode_pending_content_key(&pending_index_key(&content_key)),
+            Some(content_key)
        );
    }
 }
@@ -6,8 +6,8 @@ mod repository;
 mod rocks;

 pub use repository::{
-    StorageError, TorrentRepository, UpsertOutcome, VerificationEnqueueOutcome,
-    VerificationPriority, VerificationRequest,
+    ContentGroupTask, ContentVariants, StorageError, TorrentRepository, UpsertOutcome,
+    VerificationEnqueueOutcome, VerificationPriority, VerificationRequest,
 };
 #[cfg(feature = "rocksdb-storage")]
 pub use rocks::RocksTorrentRepository;
@@ -1,6 +1,6 @@
 // 负责定义元数据去重状态恢复和索引任务所需的存储接口

-use crate::domain::{InfoHash, TorrentRecord, VerificationResult};
+use crate::domain::{ContentGroup, InfoHash, TorrentRecord, VerificationResult};

 pub trait TorrentRepository: Send + Sync {
    fn get(&self, info_hash: InfoHash) -> Result<Option<TorrentRecord>, StorageError>;
@@ -28,18 +28,25 @@ pub trait TorrentRepository: Send + Sync {
        Ok(unknown)
    }

-    fn by_content_key(
+    fn pending_index(&self, limit: usize) -> Result<Vec<ContentGroupTask>, StorageError>;
+
+    fn content_group(
        &self,
        content_key: &[u8; 32],
-        limit: usize,
-    ) -> Result<Vec<InfoHash>, StorageError>;
+        now: u64,
+    ) -> Result<Option<ContentGroup>, StorageError>;

-    fn pending_index(&self, limit: usize) -> Result<Vec<InfoHash>, StorageError>;
-
-    fn mark_indexed(&self, info_hash: InfoHash, indexed_at: u64) -> Result<(), StorageError>;
+    fn mark_indexed(&self, content_key: &[u8; 32], revision: u64) -> Result<bool, StorageError>;

    fn prepare_full_reindex(&self) -> Result<u64, StorageError>;

+    fn content_variants(
+        &self,
+        content_key: &[u8; 32],
+        offset: usize,
+        limit: usize,
+    ) -> Result<ContentVariants, StorageError>;
+
    fn enqueue_verification(
        &self,
        info_hashes: &[InfoHash],
@@ -63,6 +70,18 @@ pub trait TorrentRepository: Send + Sync {
    fn verification_queue_len(&self) -> Result<usize, StorageError>;
 }

+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct ContentGroupTask {
+    pub content_key: [u8; 32],
+    pub revision: u64,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ContentVariants {
+    pub total: u64,
+    pub records: Vec<TorrentRecord>,
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum VerificationPriority {
    Normal,
@@ -72,7 +91,6 @@ pub enum VerificationPriority {
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct VerificationRequest {
    pub info_hash: InfoHash,
-    pub priority: VerificationPriority,
 }

 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
@@ -97,12 +115,14 @@ pub enum StorageError {
    Encode(#[from] rmp_serde::encode::Error),
    #[error("记录解码失败: {0}")]
    Decode(#[from] rmp_serde::decode::Error),
-    #[error("数据库 schema 版本不受支持 expected={expected} actual={actual}")]
-    SchemaVersion { expected: u32, actual: u32 },
-    #[error("数据库 schema 版本数据损坏")]
-    CorruptSchemaVersion,
-    #[error("待索引记录不存在 infohash={0}")]
+    #[error("数据库格式与当前程序不兼容 请清理开发数据目录后重新启动")]
+    IncompatibleDatabaseFormat,
+    #[error("待索引内容组不存在 content_key={0}")]
+    MissingContentGroup(String),
+    #[error("种子记录不存在 infohash={0}")]
    MissingRecord(InfoHash),
+    #[error("内容组状态数据损坏")]
+    CorruptContentGroup,
    #[error("验证队列计数数据损坏")]
    CorruptVerificationQueueCount,
 }
@@ -7,26 +7,32 @@ use rocksdb::{
    SliceTransform, WriteBatch,
 };

-use crate::domain::{IndexState, InfoHash, TorrentRecord, VerificationResult};
+use crate::domain::{ContentGroupBuilder, InfoHash, TorrentRecord, VerificationResult};

 use super::{
    keys::{
-        DATABASE_SCHEMA_VERSION, MIGRATION_KEY, SCHEMA_VERSION_KEY, VERIFICATION_QUEUE_COUNT_KEY,
-        content_member_key, content_members_prefix, decode_content_member_info_hash,
-        decode_pending_info_hash, decode_verification_lease, decode_verification_task,
-        pending_index_key, pending_index_prefix, torrent_key, torrent_prefix,
+        DATABASE_FORMAT_KEY, DATABASE_FORMAT_VALUE, VERIFICATION_QUEUE_COUNT_KEY,
+        content_group_key, content_group_prefix, content_member_key, content_members_prefix,
+        decode_content_member_info_hash, decode_pending_content_key, decode_verification_lease,
+        decode_verification_task, pending_index_key, pending_index_prefix, torrent_key,
        verification_lease_key, verification_lease_prefix, verification_locator_key,
        verification_task_key, verification_task_prefix,
    },
    repository::{
-        StorageError, TorrentRepository, UpsertOutcome, VerificationEnqueueOutcome,
-        VerificationPriority, VerificationRequest,
+        ContentGroupTask, ContentVariants, StorageError, TorrentRepository, UpsertOutcome,
+        VerificationEnqueueOutcome, VerificationPriority, VerificationRequest,
    },
 };

 const DEFAULT_BLOCK_CACHE_BYTES: usize = 64 * 1024 * 1024;
 type VerificationTaskEntry = (Box<[u8]>, InfoHash);

+#[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize)]
+struct ContentGroupState {
+    revision: u64,
+    member_count: u64,
+}
+
 pub struct RocksTorrentRepository {
    db: DB,
    write_lock: Mutex<()>,
@@ -50,85 +56,60 @@ impl RocksTorrentRepository {
            db: DB::open(&options, path)?,
            write_lock: Mutex::new(()),
        };
-        repository.migrate_schema()?;
+        repository.initialize_format()?;
        Ok(repository)
    }

-    fn migrate_schema(&self) -> Result<(), StorageError> {
-        let expected = DATABASE_SCHEMA_VERSION.to_be_bytes();
-        match self.db.get(SCHEMA_VERSION_KEY)? {
+    fn initialize_format(&self) -> Result<(), StorageError> {
+        match self.db.get(DATABASE_FORMAT_KEY)? {
            None => self
                .db
-                .put(SCHEMA_VERSION_KEY, expected)
+                .put(DATABASE_FORMAT_KEY, DATABASE_FORMAT_VALUE)
                .map_err(Into::into),
-            Some(value) if value.as_ref() == expected => Ok(()),
-            Some(value) if value.as_ref() == 1_u32.to_be_bytes() => {
-                let mut batch = WriteBatch::default();
-                batch.put(SCHEMA_VERSION_KEY, expected);
-                batch.put(MIGRATION_KEY, []);
-                self.db.write(batch)?;
-                self.migrate_v1_records()
-            }
-            Some(value) => {
-                let actual = value
-                    .as_slice()
-                    .try_into()
-                    .map(u32::from_be_bytes)
-                    .map_err(|_| StorageError::CorruptSchemaVersion)?;
-                Err(StorageError::SchemaVersion {
-                    expected: DATABASE_SCHEMA_VERSION,
-                    actual,
-                })
-            }
-        }?;
-        if self.db.get(MIGRATION_KEY)?.is_some() {
-            self.migrate_v1_records()?;
+            Some(value) if value.as_slice() == DATABASE_FORMAT_VALUE => Ok(()),
+            Some(_) => Err(StorageError::IncompatibleDatabaseFormat),
        }
-        Ok(())
-    }
-
-    fn migrate_v1_records(&self) -> Result<(), StorageError> {
-        const BATCH_SIZE: usize = 1_000;
-        let prefix = torrent_prefix();
-        let iterator = self
-            .db
-            .iterator(IteratorMode::From(&prefix, Direction::Forward));
-        let mut batch = WriteBatch::default();
-        let mut batch_len = 0;
-        for entry in iterator {
-            let (key, value) = entry?;
-            if !key.starts_with(&prefix) {
-                break;
-            }
-            let mut record = Self::decode_raw(&value)?;
-            record.normalize_schema();
-            record.index_state = IndexState::Pending;
-            batch.put(&key, Self::encode(&record)?);
-            batch.put(pending_index_key(record.info_hash), []);
-            batch_len += 1;
-            if batch_len == BATCH_SIZE {
-                self.db.write(batch)?;
-                batch = WriteBatch::default();
-                batch_len = 0;
-            }
-        }
-        batch.delete(MIGRATION_KEY);
-        self.db.write(batch)?;
-        Ok(())
    }

    fn encode(record: &TorrentRecord) -> Result<Vec<u8>, StorageError> {
        rmp_serde::to_vec_named(record).map_err(Into::into)
    }

-    fn decode_raw(bytes: &[u8]) -> Result<TorrentRecord, StorageError> {
+    fn decode(bytes: &[u8]) -> Result<TorrentRecord, StorageError> {
        rmp_serde::from_slice(bytes).map_err(Into::into)
    }

-    fn decode(bytes: &[u8]) -> Result<TorrentRecord, StorageError> {
-        let mut record = Self::decode_raw(bytes)?;
-        record.normalize_schema();
-        Ok(record)
+    fn encode_group(state: ContentGroupState) -> Result<Vec<u8>, StorageError> {
+        rmp_serde::to_vec_named(&state).map_err(Into::into)
+    }
+
+    fn group_state(
+        &self,
+        content_key: &[u8; 32],
+    ) -> Result<Option<ContentGroupState>, StorageError> {
+        self.db
+            .get(content_group_key(content_key))?
+            .map(|bytes| rmp_serde::from_slice(&bytes).map_err(StorageError::from))
+            .transpose()
+    }
+
+    fn dirty_group(
+        &self,
+        batch: &mut WriteBatch,
+        content_key: &[u8; 32],
+        inserted: bool,
+    ) -> Result<(), StorageError> {
+        let mut state = self.group_state(content_key)?.unwrap_or(ContentGroupState {
+            revision: 0,
+            member_count: 0,
+        });
+        state.revision = state.revision.saturating_add(1);
+        if inserted {
+            state.member_count = state.member_count.saturating_add(1);
+        }
+        batch.put(content_group_key(content_key), Self::encode_group(state)?);
+        batch.put(pending_index_key(content_key), state.revision.to_be_bytes());
+        Ok(())
    }

    fn verification_queue_len_inner(&self) -> Result<usize, StorageError> {
@@ -159,104 +140,8 @@ impl RocksTorrentRepository {
        };
        Ok(Some((key, info_hash)))
    }
-}

-impl TorrentRepository for RocksTorrentRepository {
-    fn get(&self, info_hash: InfoHash) -> Result<Option<TorrentRecord>, StorageError> {
-        self.db
-            .get(torrent_key(info_hash))?
-            .map(|bytes| Self::decode(&bytes))
-            .transpose()
-    }
-
-    fn upsert(&self, observation: TorrentRecord) -> Result<UpsertOutcome, StorageError> {
-        let _guard = self
-            .write_lock
-            .lock()
-            .unwrap_or_else(std::sync::PoisonError::into_inner);
-        if let Some(mut current) = self.get(observation.info_hash)? {
-            current.observe_again(observation.last_seen, &observation.source_peers);
-            current.index_state = IndexState::Pending;
-            let mut batch = WriteBatch::default();
-            batch.put(torrent_key(current.info_hash), Self::encode(&current)?);
-            batch.put(pending_index_key(current.info_hash), []);
-            self.db.write(batch)?;
-            return Ok(UpsertOutcome::Updated {
-                seen_count: current.seen_count,
-            });
-        }
-
-        let mut batch = WriteBatch::default();
-        batch.put(
-            torrent_key(observation.info_hash),
-            Self::encode(&observation)?,
-        );
-        batch.put(
-            content_member_key(&observation.content_key, observation.info_hash),
-            [],
-        );
-        batch.put(pending_index_key(observation.info_hash), []);
-        self.db.write(batch)?;
-        Ok(UpsertOutcome::Inserted)
-    }
-
-    fn observe_existing(
-        &self,
-        info_hash: InfoHash,
-        observed_at: u64,
-    ) -> Result<bool, StorageError> {
-        let _guard = self
-            .write_lock
-            .lock()
-            .unwrap_or_else(std::sync::PoisonError::into_inner);
-        let Some(mut record) = self.get(info_hash)? else {
-            return Ok(false);
-        };
-        record.observe_again(observed_at, &[]);
-        record.index_state = IndexState::Pending;
-        let mut batch = WriteBatch::default();
-        batch.put(torrent_key(info_hash), Self::encode(&record)?);
-        batch.put(pending_index_key(info_hash), []);
-        self.db.write(batch)?;
-        Ok(true)
-    }
-
-    fn filter_unknown_and_observe(
-        &self,
-        info_hashes: &[InfoHash],
-        observed_at: u64,
-    ) -> Result<Vec<InfoHash>, StorageError> {
-        if info_hashes.is_empty() {
-            return Ok(Vec::new());
-        }
-        let _guard = self
-            .write_lock
-            .lock()
-            .unwrap_or_else(std::sync::PoisonError::into_inner);
-        let keys: Vec<_> = info_hashes.iter().copied().map(torrent_key).collect();
-        let records = self.db.multi_get(keys.iter());
-        let mut unknown = Vec::with_capacity(info_hashes.len());
-        let mut batch = WriteBatch::default();
-        let mut updated = 0_usize;
-        for ((info_hash, key), record) in info_hashes.iter().zip(&keys).zip(records) {
-            let Some(bytes) = record? else {
-                unknown.push(*info_hash);
-                continue;
-            };
-            let mut record = Self::decode(&bytes)?;
-            record.observe_again(observed_at, &[]);
-            record.index_state = IndexState::Pending;
-            batch.put(key, Self::encode(&record)?);
-            batch.put(pending_index_key(*info_hash), []);
-            updated += 1;
-        }
-        if updated > 0 {
-            self.db.write(batch)?;
-        }
-        Ok(unknown)
-    }
-
-    fn by_content_key(
+    fn content_member_hashes(
        &self,
        content_key: &[u8; 32],
        limit: usize,
@@ -281,8 +166,105 @@ impl TorrentRepository for RocksTorrentRepository {
        }
        Ok(hashes)
    }
+}

-    fn pending_index(&self, limit: usize) -> Result<Vec<InfoHash>, StorageError> {
+impl TorrentRepository for RocksTorrentRepository {
+    fn get(&self, info_hash: InfoHash) -> Result<Option<TorrentRecord>, StorageError> {
+        self.db
+            .get(torrent_key(info_hash))?
+            .map(|bytes| Self::decode(&bytes))
+            .transpose()
+    }
+
+    fn upsert(&self, observation: TorrentRecord) -> Result<UpsertOutcome, StorageError> {
+        let _guard = self
+            .write_lock
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        if let Some(mut current) = self.get(observation.info_hash)? {
+            current.observe_again(observation.last_seen, &observation.source_peers);
+            let mut batch = WriteBatch::default();
+            batch.put(torrent_key(current.info_hash), Self::encode(&current)?);
+            self.dirty_group(&mut batch, &current.content_key, false)?;
+            self.db.write(batch)?;
+            return Ok(UpsertOutcome::Updated {
+                seen_count: current.seen_count,
+            });
+        }
+
+        let mut batch = WriteBatch::default();
+        batch.put(
+            torrent_key(observation.info_hash),
+            Self::encode(&observation)?,
+        );
+        batch.put(
+            content_member_key(&observation.content_key, observation.info_hash),
+            [],
+        );
+        self.dirty_group(&mut batch, &observation.content_key, true)?;
+        self.db.write(batch)?;
+        Ok(UpsertOutcome::Inserted)
+    }
+
+    fn observe_existing(
+        &self,
+        info_hash: InfoHash,
+        observed_at: u64,
+    ) -> Result<bool, StorageError> {
+        let _guard = self
+            .write_lock
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        let Some(mut record) = self.get(info_hash)? else {
+            return Ok(false);
+        };
+        record.observe_again(observed_at, &[]);
+        let mut batch = WriteBatch::default();
+        batch.put(torrent_key(info_hash), Self::encode(&record)?);
+        self.dirty_group(&mut batch, &record.content_key, false)?;
+        self.db.write(batch)?;
+        Ok(true)
+    }
+
+    fn filter_unknown_and_observe(
+        &self,
+        info_hashes: &[InfoHash],
+        observed_at: u64,
+    ) -> Result<Vec<InfoHash>, StorageError> {
+        if info_hashes.is_empty() {
+            return Ok(Vec::new());
+        }
+        let _guard = self
+            .write_lock
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        let keys: Vec<_> = info_hashes.iter().copied().map(torrent_key).collect();
+        let records = self.db.multi_get(keys.iter());
+        let mut unknown = Vec::with_capacity(info_hashes.len());
+        let mut batch = WriteBatch::default();
+        let mut updated = 0_usize;
+        let mut dirty_groups = std::collections::BTreeSet::new();
+        for ((info_hash, key), record) in info_hashes.iter().zip(&keys).zip(records) {
+            let Some(bytes) = record? else {
+                unknown.push(*info_hash);
+                continue;
+            };
+            let mut record = Self::decode(&bytes)?;
+            record.observe_again(observed_at, &[]);
+            batch.put(key, Self::encode(&record)?);
+            dirty_groups.insert(record.content_key);
+            updated += 1;
+        }
+        if updated > 0 {
+            for content_key in dirty_groups {
+                self.dirty_group(&mut batch, &content_key, false)?;
+            }
+            self.db.write(batch)?;
+        }
+        Ok(unknown)
+    }
+
+    fn pending_index(&self, limit: usize) -> Result<Vec<ContentGroupTask>, StorageError> {
        if limit == 0 {
            return Ok(Vec::new());
        }
@@ -290,35 +272,72 @@ impl TorrentRepository for RocksTorrentRepository {
        let iterator = self
            .db
            .iterator(IteratorMode::From(&prefix, Direction::Forward));
-        let mut hashes = Vec::with_capacity(limit.min(1024));
+        let mut tasks = Vec::with_capacity(limit.min(1024));
        for entry in iterator {
-            let (key, _) = entry?;
-            let Some(info_hash) = decode_pending_info_hash(&key) else {
+            let (key, value) = entry?;
+            let Some(content_key) = decode_pending_content_key(&key) else {
                break;
            };
-            hashes.push(info_hash);
-            if hashes.len() == limit {
+            let revision = value
+                .as_ref()
+                .try_into()
+                .map(u64::from_be_bytes)
+                .map_err(|_| StorageError::CorruptContentGroup)?;
+            tasks.push(ContentGroupTask {
+                content_key,
+                revision,
+            });
+            if tasks.len() == limit {
                break;
            }
        }
-        Ok(hashes)
+        Ok(tasks)
    }

-    fn mark_indexed(&self, info_hash: InfoHash, indexed_at: u64) -> Result<(), StorageError> {
+    fn content_group(
+        &self,
+        content_key: &[u8; 32],
+        now: u64,
+    ) -> Result<Option<crate::domain::ContentGroup>, StorageError> {
+        let mut builder = ContentGroupBuilder::new(*content_key, now);
+        let prefix = content_members_prefix(content_key);
+        let iterator = self
+            .db
+            .iterator(IteratorMode::From(&prefix, Direction::Forward));
+        let mut found = false;
+        for entry in iterator {
+            let (key, _) = entry?;
+            let Some(info_hash) = decode_content_member_info_hash(&key, content_key) else {
+                break;
+            };
+            if let Some(record) = self.get(info_hash)? {
+                builder.push(record);
+                found = true;
+            }
+        }
+        if found {
+            Ok(builder.finish())
+        } else {
+            Ok(None)
+        }
+    }
+
+    fn mark_indexed(&self, content_key: &[u8; 32], revision: u64) -> Result<bool, StorageError> {
        let _guard = self
            .write_lock
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
-        let Some(mut record) = self.get(info_hash)? else {
-            return Err(StorageError::MissingRecord(info_hash));
+        let Some(state) = self.group_state(content_key)? else {
+            return Err(StorageError::MissingContentGroup(hex::encode(content_key)));
        };
-        record.index_state = IndexState::Indexed { indexed_at };
-
+        if state.revision != revision {
+            return Ok(false);
+        }
        let mut batch = WriteBatch::default();
-        batch.put(torrent_key(info_hash), Self::encode(&record)?);
-        batch.delete(pending_index_key(info_hash));
+        batch.put(content_group_key(content_key), Self::encode_group(state)?);
+        batch.delete(pending_index_key(content_key));
        self.db.write(batch)?;
-        Ok(())
+        Ok(true)
    }

    fn prepare_full_reindex(&self) -> Result<u64, StorageError> {
@@ -328,7 +347,7 @@ impl TorrentRepository for RocksTorrentRepository {
            .write_lock
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
-        let prefix = torrent_prefix();
+        let prefix = content_group_prefix();
        let iterator = self
            .db
            .iterator(IteratorMode::From(&prefix, Direction::Forward));
@@ -340,10 +359,14 @@ impl TorrentRepository for RocksTorrentRepository {
            if !key.starts_with(&prefix) {
                break;
            }
-            let mut record = Self::decode(&value)?;
-            record.index_state = IndexState::Pending;
-            batch.put(&key, Self::encode(&record)?);
-            batch.put(pending_index_key(record.info_hash), []);
+            let content_key: [u8; 32] = key[1..]
+                .try_into()
+                .map_err(|_| StorageError::CorruptContentGroup)?;
+            let state: ContentGroupState = rmp_serde::from_slice(&value)?;
+            batch.put(
+                pending_index_key(&content_key),
+                state.revision.to_be_bytes(),
+            );
            batch_len += 1;
            total += 1;
            if batch_len == BATCH_SIZE {
@@ -358,6 +381,25 @@ impl TorrentRepository for RocksTorrentRepository {
        Ok(total)
    }

+    fn content_variants(
+        &self,
+        content_key: &[u8; 32],
+        offset: usize,
+        limit: usize,
+    ) -> Result<ContentVariants, StorageError> {
+        let total = self
+            .group_state(content_key)?
+            .map_or(0, |state| state.member_count);
+        let hashes = self.content_member_hashes(content_key, offset.saturating_add(limit))?;
+        let mut records = Vec::with_capacity(limit);
+        for info_hash in hashes.into_iter().skip(offset).take(limit) {
+            if let Some(record) = self.get(info_hash)? {
+                records.push(record);
+            }
+        }
+        Ok(ContentVariants { total, records })
+    }
+
    fn enqueue_verification(
        &self,
        info_hashes: &[InfoHash],
@@ -488,10 +530,7 @@ impl TorrentRepository for RocksTorrentRepository {
        );
        batch.put(verification_locator_key(info_hash), lease_key);
        self.db.write(batch)?;
-        Ok(Some(VerificationRequest {
-            info_hash,
-            priority,
-        }))
+        Ok(Some(VerificationRequest { info_hash }))
    }

    fn finish_verification(
@@ -512,7 +551,7 @@ impl TorrentRepository for RocksTorrentRepository {
        let count = self.verification_queue_len_inner()?.saturating_sub(1);
        let mut batch = WriteBatch::default();
        batch.put(torrent_key(info_hash), Self::encode(&record)?);
-        batch.put(pending_index_key(info_hash), []);
+        self.dirty_group(&mut batch, &record.content_key, false)?;
        if let Some(queued_key) = queued_key {
            batch.delete(queued_key);
        }
@@ -529,12 +568,9 @@ impl TorrentRepository for RocksTorrentRepository {

 #[cfg(test)]
 mod tests {
-    use serde::Serialize;
    use tempfile::TempDir;

-    use crate::domain::{
-        AvailabilityStatus, IndexState, InfoHash, TorrentFile, VerificationResult, test_record,
-    };
+    use crate::domain::{AvailabilityStatus, VerificationResult, test_record};

    use super::*;

@@ -553,6 +589,22 @@ mod tests {
        assert_eq!(repository.get(expected.info_hash).unwrap(), Some(expected));
    }

+    #[test]
+    fn incompatible_database_format_is_rejected() {
+        let directory = TempDir::new().unwrap();
+        {
+            let mut options = Options::default();
+            options.create_if_missing(true);
+            let database = DB::open(&options, directory.path()).unwrap();
+            database.put(DATABASE_FORMAT_KEY, b"other-format").unwrap();
+        }
+
+        assert!(matches!(
+            RocksTorrentRepository::open(directory.path()),
+            Err(StorageError::IncompatibleDatabaseFormat)
+        ));
+    }
+
    #[test]
    fn duplicate_updates_observation_without_creating_another_pending_item() {
        let directory = TempDir::new().unwrap();
@@ -575,7 +627,9 @@ mod tests {
        assert_eq!(stored.first_seen, 10);
        assert_eq!(stored.last_seen, 20);
        assert_eq!(stored.seen_count, 2);
-        assert_eq!(repository.pending_index(10).unwrap(), vec![first.info_hash]);
+        let tasks = repository.pending_index(10).unwrap();
+        assert_eq!(tasks.len(), 1);
+        assert_eq!(tasks[0].content_key, first.content_key);
    }

    #[test]
@@ -594,7 +648,7 @@ mod tests {
        let updated = repository.get(existing.info_hash).unwrap().unwrap();
        assert_eq!(updated.last_seen, 20);
        assert_eq!(updated.seen_count, existing.seen_count + 1);
-        assert_eq!(updated.index_state, IndexState::Pending);
+        assert_eq!(repository.pending_index(10).unwrap().len(), 1);
    }

    #[test]
@@ -622,7 +676,13 @@ mod tests {
        repository.upsert(first.clone()).unwrap();
        repository.upsert(second.clone()).unwrap();

-        let mut hashes = repository.by_content_key(&first.content_key, 10).unwrap();
+        let mut hashes: Vec<_> = repository
+            .content_variants(&first.content_key, 0, 10)
+            .unwrap()
+            .records
+            .into_iter()
+            .map(|record| record.info_hash)
+            .collect();
        hashes.sort_unstable_by_key(ToString::to_string);
        assert_eq!(hashes, vec![first.info_hash, second.info_hash]);
        assert!(repository.get(first.info_hash).unwrap().is_some());
@@ -636,17 +696,51 @@ mod tests {
        let record = test_record(3, 10);
        repository.upsert(record.clone()).unwrap();

-        repository.mark_indexed(record.info_hash, 30).unwrap();
+        let task = repository.pending_index(10).unwrap()[0];
+        assert!(
+            repository
+                .mark_indexed(&record.content_key, task.revision)
+                .unwrap()
+        );

        assert!(repository.pending_index(10).unwrap().is_empty());
-        assert_eq!(
-            repository
-                .get(record.info_hash)
+    }
+
+    #[test]
+    fn stale_index_revision_cannot_clear_a_newer_update() {
+        let directory = TempDir::new().unwrap();
+        let repository = RocksTorrentRepository::open(directory.path()).unwrap();
+        let record = test_record(4, 10);
+        repository.upsert(record.clone()).unwrap();
+        let stale = repository.pending_index(10).unwrap()[0];
+
+        repository.observe_existing(record.info_hash, 20).unwrap();
+
+        assert!(
+            !repository
+                .mark_indexed(&record.content_key, stale.revision)
                .unwrap()
-                .unwrap()
-                .index_state,
-            IndexState::Indexed { indexed_at: 30 }
        );
+        let current = repository.pending_index(10).unwrap()[0];
+        assert!(current.revision > stale.revision);
+    }
+
+    #[test]
+    fn content_variants_report_exact_group_count() {
+        let directory = TempDir::new().unwrap();
+        let repository = RocksTorrentRepository::open(directory.path()).unwrap();
+        let first = test_record(4, 10);
+        let mut second = test_record(5, 20);
+        second.content_key = first.content_key;
+        repository.upsert(first.clone()).unwrap();
+        repository.upsert(second.clone()).unwrap();
+
+        let variants = repository
+            .content_variants(&first.content_key, 1, 10)
+            .unwrap();
+        assert_eq!(variants.total, 2);
+        assert_eq!(variants.records.len(), 1);
+        assert_eq!(variants.records[0].info_hash, second.info_hash);
    }

    #[test]
@@ -654,86 +748,19 @@ mod tests {
        let directory = TempDir::new().unwrap();
        let repository = RocksTorrentRepository::open(directory.path()).unwrap();
        let first = test_record(7, 10);
-        let second = test_record(8, 10);
+        let mut second = test_record(8, 10);
+        second.content_key = [8; 32];
        repository.upsert(first.clone()).unwrap();
        repository.upsert(second.clone()).unwrap();
-        repository.mark_indexed(first.info_hash, 20).unwrap();
-        repository.mark_indexed(second.info_hash, 20).unwrap();
+        for task in repository.pending_index(10).unwrap() {
+            repository
+                .mark_indexed(&task.content_key, task.revision)
+                .unwrap();
+        }
        assert!(repository.pending_index(10).unwrap().is_empty());

        assert_eq!(repository.prepare_full_reindex().unwrap(), 2);
        assert_eq!(repository.pending_index(10).unwrap().len(), 2);
-        assert_eq!(
-            repository
-                .get(first.info_hash)
-                .unwrap()
-                .unwrap()
-                .index_state,
-            IndexState::Pending
-        );
-    }
-
-    #[derive(Serialize)]
-    struct LegacyTorrentRecord {
-        schema_version: u16,
-        info_hash: InfoHash,
-        name: String,
-        total_size: u64,
-        files: Vec<TorrentFile>,
-        piece_length: u64,
-        source_peers: Vec<String>,
-        content_key: [u8; 32],
-        first_seen: u64,
-        last_seen: u64,
-        seen_count: u64,
-        index_state: IndexState,
-    }
-
-    #[test]
-    fn version_one_database_migrates_records_and_marks_them_pending() {
-        let directory = TempDir::new().unwrap();
-        let record = test_record(9, 10);
-        let legacy = LegacyTorrentRecord {
-            schema_version: 1,
-            info_hash: record.info_hash,
-            name: record.name,
-            total_size: record.total_size,
-            files: record.files,
-            piece_length: record.piece_length,
-            source_peers: record.source_peers,
-            content_key: record.content_key,
-            first_seen: record.first_seen,
-            last_seen: record.last_seen,
-            seen_count: record.seen_count,
-            index_state: IndexState::Indexed { indexed_at: 20 },
-        };
-        {
-            let mut options = Options::default();
-            options.create_if_missing(true);
-            let db = DB::open(&options, directory.path()).unwrap();
-            db.put(SCHEMA_VERSION_KEY, 1_u32.to_be_bytes()).unwrap();
-            db.put(
-                torrent_key(legacy.info_hash),
-                rmp_serde::to_vec_named(&legacy).unwrap(),
-            )
-            .unwrap();
-        }
-
-        let repository = RocksTorrentRepository::open(directory.path()).unwrap();
-        let migrated = repository.get(legacy.info_hash).unwrap().unwrap();
-        assert_eq!(migrated.schema_version, 2);
-        assert_eq!(migrated.availability.status, AvailabilityStatus::Unknown);
-        assert_eq!(migrated.activity_updated_at, 10);
-        assert_eq!(migrated.index_state, IndexState::Pending);
-        assert_eq!(
-            repository.pending_index(10).unwrap(),
-            vec![legacy.info_hash]
-        );
-        assert_eq!(
-            repository.db.get(SCHEMA_VERSION_KEY).unwrap().unwrap(),
-            DATABASE_SCHEMA_VERSION.to_be_bytes()
-        );
-        assert!(repository.db.get(MIGRATION_KEY).unwrap().is_none());
    }

    #[test]
@@ -764,7 +791,7 @@ mod tests {
        let repository = RocksTorrentRepository::open(directory.path()).unwrap();
        let claimed = repository.claim_verification(102, 60).unwrap().unwrap();
        assert_eq!(claimed.info_hash, high.info_hash);
-        assert_eq!(claimed.priority, VerificationPriority::High);
+        assert_eq!(claimed.info_hash, high.info_hash);
        repository
            .finish_verification(
                high.info_hash,
@@ -779,7 +806,6 @@ mod tests {
        let stored = repository.get(high.info_hash).unwrap().unwrap();
        assert_eq!(stored.availability.status, AvailabilityStatus::Active);
        assert_eq!(stored.availability.reachable_peers, 2);
-        assert_eq!(stored.index_state, IndexState::Pending);
    }

    #[test]