20260718 重构 RPKI Explorer 查询界面

src/bin/rpki_query_service.rs

@@ -7,9 +7,10 @@ use std::process::Command;
 use std::sync::{Arc, Mutex};
 
 use rpki::blob_store::ExternalRepoBytesDb;
+use rpki::query::report_stream::{ObjectFilter, ObjectScope};
 use rpki::query_db::{
-    ChainEdgeRecord, ExportJobRecord, ObjectInstanceRecord, QueryDb, QueryDbError,
-    ValidationExplainRecord,
+    ChainEdgeRecord, ExportJobRecord, ObjectInstanceRecord, ObjectUriIndexRecord, QueryDb,
+    QueryDbError, ValidationExplainRecord,
 };
 use serde::Serialize;
 use serde_json::{Value, json};
@@ -544,6 +545,18 @@ fn route_request(
         return Ok(json!({"data":{"service":"rpki_query_service","version":1}}));
     }
     match segments.as_slice() {
+        ["healthz"] => {
+            let latest_ready_run = db.latest_ready_run()?;
+            data_response(
+                &json!({
+                    "status": "ok",
+                    "service": "rpki_query_service",
+                    "version": 1,
+                    "latestReadyRun": latest_ready_run,
+                }),
+                None,
+            )
+        }
         ["runs"] => page_response(db.list_runs(limit(&query), cursor(&query))?, None),
         ["latest_run"] => {
             let run_id = db
@@ -630,8 +643,15 @@ fn route_request(
         }
         ["runs", raw_run_id, "repos", repo_id, "objects"] => {
             let run_id = resolve_run(db, raw_run_id)?;
+            let filter = object_filter_from_query(&query)?;
             page_response(
-                db.list_objects_for_repo(&run_id, repo_id, limit(&query), cursor(&query))?,
+                db.list_objects_filtered(
+                    &run_id,
+                    ObjectScope::Repo(repo_id.to_string()),
+                    &filter,
+                    limit(&query),
+                    cursor(&query),
+                )?,
                 Some(run_id),
             )
         }
@@ -669,18 +689,63 @@ fn route_request(
         }
         ["runs", raw_run_id, "publication-points", pp_id, "objects"] => {
             let run_id = resolve_run(db, raw_run_id)?;
+            let filter = object_filter_from_query(&query)?;
             page_response(
-                db.list_objects_for_pp(&run_id, pp_id, limit(&query), cursor(&query))?,
+                db.list_objects_filtered(
+                    &run_id,
+                    ObjectScope::PublicationPoint(pp_id.to_string()),
+                    &filter,
+                    limit(&query),
+                    cursor(&query),
+                )?,
                 Some(run_id),
             )
         }
         ["runs", raw_run_id, "objects"] => {
             let run_id = resolve_run(db, raw_run_id)?;
+            let filter = object_filter_from_query(&query)?;
             page_response(
-                db.list_objects(&run_id, limit(&query), cursor(&query))?,
+                db.list_objects_filtered(
+                    &run_id,
+                    ObjectScope::All,
+                    &filter,
+                    limit(&query),
+                    cursor(&query),
+                )?,
                 Some(run_id),
             )
         }
+        ["runs", raw_run_id, "issues"] => {
+            let run_id = resolve_run(db, raw_run_id)?;
+            let mut filter = object_filter_from_query(&query)?;
+            filter.issues_only = true;
+            filter.repo_id = query.get("repoId").cloned();
+            filter.pp_id = query.get("ppId").cloned();
+            page_response(
+                db.list_objects_filtered(
+                    &run_id,
+                    ObjectScope::All,
+                    &filter,
+                    limit(&query),
+                    cursor(&query),
+                )?,
+                Some(run_id),
+            )
+        }
+        ["runs", raw_run_id, "search"] => {
+            let run_id = resolve_run(db, raw_run_id)?;
+            let q = query
+                .get("q")
+                .map(String::as_str)
+                .unwrap_or("")
+                .trim()
+                .to_string();
+            if q.is_empty() {
+                return Err(ApiError::new(400, "q query parameter is required"));
+            }
+            let limit = search_limit(&query);
+            data_response(&search_run(db, &run_id, &q, limit)?, Some(run_id))
+        }
         ["runs", raw_run_id, "objects", "by-uri"] => {
             let run_id = resolve_run(db, raw_run_id)?;
             let uri = query
@@ -792,6 +857,18 @@ fn route_request(
                 offset_cursor(&query),
             )
         }
+        ["runs", raw_run_id, "exports"] => {
+            let run_id = resolve_run(db, raw_run_id)?;
+            let limit = exports_limit(&query);
+            page_response(
+                rpki::query_db::QueryPage {
+                    data: list_export_jobs(export_jobs, &run_id, limit)?,
+                    next_cursor: None,
+                    limit,
+                },
+                Some(run_id),
+            )
+        }
         ["runs", raw_run_id, "exports", job_id] => {
             let run_id = resolve_run(db, raw_run_id)?;
             let job = get_export_job(db, export_jobs, &run_id, job_id)?
@@ -1150,6 +1227,24 @@ fn get_export_job(
     Ok(db.get_export_job(run_id, job_id)?)
 }
 
+fn list_export_jobs(
+    export_jobs: &ExportJobStore,
+    run_id: &str,
+    limit: usize,
+) -> Result<Vec<ExportJobRecord>, ApiError> {
+    let jobs = export_jobs
+        .lock()
+        .map_err(|_| ApiError::new(500, "export job store lock poisoned"))?;
+    let mut out = jobs
+        .values()
+        .filter(|job| job.run_id == run_id)
+        .cloned()
+        .collect::<Vec<_>>();
+    out.sort_by(|left, right| right.created_at.cmp(&left.created_at));
+    out.truncate(limit);
+    Ok(out)
+}
+
 fn run_export_job(
     db: &QueryDb,
     repo_bytes: &ExternalRepoBytesDb,
@@ -1575,6 +1670,103 @@ fn projection_for_explain(
     Ok(Some(projection))
 }
 
+fn search_run(db: &QueryDb, run_id: &str, q: &str, limit: usize) -> Result<Value, ApiError> {
+    let objects = search_run_objects(db, run_id, q, limit)?;
+    let lower = q.to_lowercase();
+    let repos = db.search_repos(run_id, &lower, limit)?;
+    let publication_points = db.search_publication_points(run_id, &lower, limit)?;
+    Ok(json!({
+        "objects": objects,
+        "repos": repos,
+        "publicationPoints": publication_points,
+    }))
+}
+
+fn search_run_objects(
+    db: &QueryDb,
+    run_id: &str,
+    q: &str,
+    limit: usize,
+) -> Result<Vec<ObjectInstanceRecord>, ApiError> {
+    if q.starts_with("rsync://") || q.starts_with("http://") || q.starts_with("https://") {
+        let Some(index) = db.get_object_by_uri(run_id, q)? else {
+            return Ok(Vec::new());
+        };
+        let object = match db.get_object_by_instance_id(run_id, &index.object_instance_id)? {
+            Some(full) => full,
+            None => object_instance_from_uri_index(&index),
+        };
+        return Ok(vec![object]);
+    }
+    let lower = q.to_lowercase();
+    if is_sha256_hex(q) {
+        if let Some(projection) = db.get_object_projection(&lower)? {
+            if let Some(full) = db.get_cached_object_by_sha256(run_id, &projection.sha256)? {
+                return Ok(vec![full]);
+            }
+            return Ok(vec![object_instance_from_projection(run_id, &projection)]);
+        }
+        if let Some(object) = db.get_object_by_sha256(run_id, &lower)? {
+            return Ok(vec![object]);
+        }
+        return Ok(Vec::new());
+    }
+    let is_hex = lower.len() >= 8 && lower.bytes().all(|byte| byte.is_ascii_hexdigit());
+    if is_hex {
+        let mut objects = Vec::new();
+        for projection in db.search_object_projections_by_hash_prefix(&lower, limit)? {
+            let object = match db.get_cached_object_by_sha256(run_id, &projection.sha256)? {
+                Some(full) => full,
+                None => object_instance_from_projection(run_id, &projection),
+            };
+            objects.push(object);
+        }
+        return Ok(objects);
+    }
+    Ok(Vec::new())
+}
+
+fn object_instance_from_uri_index(index: &ObjectUriIndexRecord) -> ObjectInstanceRecord {
+    ObjectInstanceRecord {
+        schema_version: 1,
+        run_id: index.run_id.clone(),
+        object_instance_id: index.object_instance_id.clone(),
+        uri: index.uri.clone(),
+        uri_hash: String::new(),
+        sha256: index.sha256.clone(),
+        object_type: String::new(),
+        result: String::new(),
+        detail_summary: None,
+        repo_id: index.repo_id.clone(),
+        pp_id: index.pp_id.clone(),
+        source_section: "uri_index".to_string(),
+        rejected: false,
+        reject_reason: None,
+    }
+}
+
+fn object_instance_from_projection(
+    run_id: &str,
+    projection: &rpki::object_projection::ObjectProjectionRecord,
+) -> ObjectInstanceRecord {
+    ObjectInstanceRecord {
+        schema_version: 1,
+        run_id: run_id.to_string(),
+        object_instance_id: String::new(),
+        uri: String::new(),
+        uri_hash: String::new(),
+        sha256: projection.sha256.clone(),
+        object_type: projection.object_type.clone(),
+        result: projection.parse_status.clone(),
+        detail_summary: projection.error_summary.clone(),
+        repo_id: String::new(),
+        pp_id: String::new(),
+        source_section: "hash_index".to_string(),
+        rejected: false,
+        reject_reason: None,
+    }
+}
+
 fn global_projection_for_hash(
     db: &QueryDb,
     repo_bytes: Option<&ExternalRepoBytesDb>,
@@ -1878,6 +2070,66 @@ fn limit(query: &BTreeMap<String, String>) -> usize {
         .clamp(1, 1000)
 }
 
+fn search_limit(query: &BTreeMap<String, String>) -> usize {
+    query
+        .get("limit")
+        .and_then(|value| value.parse::<usize>().ok())
+        .unwrap_or(10)
+        .clamp(1, 50)
+}
+
+fn exports_limit(query: &BTreeMap<String, String>) -> usize {
+    query
+        .get("limit")
+        .and_then(|value| value.parse::<usize>().ok())
+        .unwrap_or(50)
+        .clamp(1, 200)
+}
+
+fn object_filter_from_query(query: &BTreeMap<String, String>) -> Result<ObjectFilter, ApiError> {
+    let mut filter = ObjectFilter::default();
+    if let Some(value) = query.get("type") {
+        if value.is_empty() {
+            return Err(ApiError::new(400, "type filter must not be empty"));
+        }
+        // Accept projection-vocabulary aliases so both `mft` and `manifest`
+        // (etc.) match the URI-derived report stream types.
+        let lower = value.to_ascii_lowercase();
+        let normalized = match lower.as_str() {
+            "mft" => "manifest",
+            "cer" => "certificate",
+            "asa" => "aspa",
+            _ => lower.as_str(),
+        };
+        filter.object_type = Some(normalized.to_string());
+    }
+    if let Some(value) = query.get("result") {
+        if !matches!(value.to_ascii_lowercase().as_str(), "ok" | "error" | "skipped") {
+            return Err(ApiError::new(400, format!("invalid result filter: {value}")));
+        }
+        filter.result = Some(value.clone());
+    }
+    if let Some(value) = query.get("rejected") {
+        filter.rejected = Some(match value.as_str() {
+            "true" => true,
+            "false" => false,
+            _ => {
+                return Err(ApiError::new(
+                    400,
+                    format!("invalid rejected filter: {value}"),
+                ));
+            }
+        });
+    }
+    if let Some(value) = query.get("reason") {
+        filter.reason = Some(value.to_lowercase());
+    }
+    if let Some(value) = query.get("q") {
+        filter.query = Some(value.to_lowercase());
+    }
+    Ok(filter)
+}
+
 fn cursor(query: &BTreeMap<String, String>) -> Option<&str> {
     query.get("cursor").map(String::as_str)
 }

src/query/report_stream.rs

@@ -50,6 +50,68 @@ pub enum ObjectLookup {
     Sha256(String),
 }
 
+/// Optional predicates applied while streaming report objects. `reason` and
+/// `query` must be pre-lowercased by the caller; they are matched as
+/// case-insensitive substrings. The other fields are exact matches.
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct ObjectFilter {
+    pub object_type: Option<String>,
+    pub result: Option<String>,
+    pub rejected: Option<bool>,
+    pub reason: Option<String>,
+    pub query: Option<String>,
+    pub repo_id: Option<String>,
+    pub pp_id: Option<String>,
+    pub issues_only: bool,
+}
+
+impl ObjectFilter {
+    pub fn matches(&self, object: &ObjectInstanceRecord) -> bool {
+        if self.issues_only && !(object.rejected || object.result == "error") {
+            return false;
+        }
+        if let Some(rejected) = self.rejected
+            && object.rejected != rejected
+        {
+            return false;
+        }
+        if let Some(result) = self.result.as_deref()
+            && !object.result.eq_ignore_ascii_case(result)
+        {
+            return false;
+        }
+        if let Some(object_type) = self.object_type.as_deref()
+            && !object.object_type.eq_ignore_ascii_case(object_type)
+        {
+            return false;
+        }
+        if let Some(repo_id) = self.repo_id.as_deref()
+            && object.repo_id != repo_id
+        {
+            return false;
+        }
+        if let Some(pp_id) = self.pp_id.as_deref()
+            && object.pp_id != pp_id
+        {
+            return false;
+        }
+        if let Some(reason) = self.reason.as_deref() {
+            let Some(reject_reason) = object.reject_reason.as_deref() else {
+                return false;
+            };
+            if !reject_reason.to_lowercase().contains(reason) {
+                return false;
+            }
+        }
+        if let Some(query) = self.query.as_deref()
+            && !object.uri.to_lowercase().contains(query)
+        {
+            return false;
+        }
+        true
+    }
+}
+
 #[derive(Clone, Debug, PartialEq, Eq, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct ObjectResolutionMeta {
@@ -81,13 +143,37 @@ pub fn list_report_objects(
     scope: ObjectScope,
     limit: usize,
     cursor: Option<&str>,
+) -> QueryDbResult<QueryPage<ObjectInstanceRecord>> {
+    list_report_objects_filtered(
+        report_path,
+        run_id,
+        scope,
+        &ObjectFilter::default(),
+        limit,
+        cursor,
+    )
+}
+
+pub fn list_report_objects_filtered(
+    report_path: &Path,
+    run_id: &str,
+    scope: ObjectScope,
+    filter: &ObjectFilter,
+    limit: usize,
+    cursor: Option<&str>,
 ) -> QueryDbResult<QueryPage<ObjectInstanceRecord>> {
     let (start_pp_index, start_object_index) = parse_object_cursor(cursor)?;
     let file = File::open(report_path)?;
     let reader = BufReader::new(file);
     let mut deserializer = serde_json::Deserializer::from_reader(reader);
-    let mut state =
-        ObjectScanState::new_list(run_id, scope, limit, start_pp_index, start_object_index);
+    let mut state = ObjectScanState::new_list(
+        run_id,
+        scope,
+        filter.clone(),
+        limit,
+        start_pp_index,
+        start_object_index,
+    );
     ObjectScanSeed { state: &mut state }
         .deserialize(&mut deserializer)
         .map_err(QueryDbError::from)?;
@@ -530,6 +616,7 @@ struct ObjectScanState<'a> {
     run_id: &'a str,
     scope: ObjectScope,
     mode: ObjectScanMode,
+    filter: ObjectFilter,
     limit: usize,
     start_pp_index: u64,
     start_object_index: u64,
@@ -544,6 +631,7 @@ impl<'a> ObjectScanState<'a> {
     fn new_list(
         run_id: &'a str,
         scope: ObjectScope,
+        filter: ObjectFilter,
         limit: usize,
         start_pp_index: u64,
         start_object_index: u64,
@@ -552,6 +640,7 @@ impl<'a> ObjectScanState<'a> {
             run_id,
             scope,
             mode: ObjectScanMode::List,
+            filter,
             limit,
             start_pp_index,
             start_object_index,
@@ -568,6 +657,7 @@ impl<'a> ObjectScanState<'a> {
             run_id,
             scope,
             mode: ObjectScanMode::Lookup(lookup),
+            filter: ObjectFilter::default(),
             limit: 1,
             start_pp_index: 0,
             start_object_index: 0,
@@ -682,7 +772,8 @@ fn process_raw_publication_point(
                 let after_cursor = pp_index > state.start_pp_index
                     || (pp_index == state.start_pp_index
                         && object_index >= state.start_object_index);
-                if after_cursor && state.data.len() < state.limit {
+                if after_cursor && state.data.len() < state.limit && state.filter.matches(&record)
+                {
                     state.data.push(record);
                     if state.data.len() == state.limit {
                         state.next_cursor = Some(object_cursor(pp_index, object_index + 1));

src/query_db.rs

@@ -473,6 +473,31 @@ impl QueryDb {
         )
     }
 
+    pub fn list_objects_filtered(
+        &self,
+        run_id: &str,
+        scope: ObjectScope,
+        filter: &report_stream::ObjectFilter,
+        limit: usize,
+        cursor: Option<&str>,
+    ) -> QueryDbResult<QueryPage<ObjectInstanceRecord>> {
+        let Some(report_path) = self.report_path_for_run(run_id)? else {
+            return Ok(QueryPage {
+                data: Vec::new(),
+                next_cursor: None,
+                limit: limit.clamp(1, 1000),
+            });
+        };
+        report_stream::list_report_objects_filtered(
+            &report_path,
+            run_id,
+            scope,
+            filter,
+            limit,
+            cursor,
+        )
+    }
+
     pub fn get_object_by_instance_id(
         &self,
         run_id: &str,
@@ -628,6 +653,118 @@ impl QueryDb {
         self.get_json_cf(CF_OBJECTS_BY_HASH, object_hash_key(sha256).as_bytes())
     }
 
+    pub fn search_object_projections_by_hash_prefix(
+        &self,
+        sha256_prefix: &str,
+        limit: usize,
+    ) -> QueryDbResult<Vec<crate::object_projection::ObjectProjectionRecord>> {
+        let cf = self.cf(CF_OBJECTS_BY_HASH)?;
+        let prefix = format!("objhash/{sha256_prefix}");
+        let mut out = Vec::new();
+        let mode = IteratorMode::From(prefix.as_bytes(), rocksdb::Direction::Forward);
+        for item in self.db.iterator_cf(cf, mode) {
+            let (key, value) = item?;
+            if !String::from_utf8_lossy(&key).starts_with(&prefix) {
+                break;
+            }
+            if out.len() >= limit {
+                break;
+            }
+            out.push(serde_json::from_slice(&value)?);
+        }
+        Ok(out)
+    }
+
+    pub fn get_cached_object_by_sha256(
+        &self,
+        run_id: &str,
+        sha256: &str,
+    ) -> QueryDbResult<Option<ObjectInstanceRecord>> {
+        let mut cursor = None;
+        loop {
+            let page: QueryPage<ObjectInstanceRecord> = self.list_json_by_prefix(
+                CF_OBJECT_INSTANCES,
+                &format!("objinst/{run_id}/"),
+                1000,
+                cursor.as_deref(),
+            )?;
+            if let Some(found) = page
+                .data
+                .into_iter()
+                .find(|item| item.sha256.eq_ignore_ascii_case(sha256))
+            {
+                return Ok(Some(found));
+            }
+            let Some(next_cursor) = page.next_cursor else {
+                return Ok(None);
+            };
+            cursor = Some(next_cursor);
+        }
+    }
+
+    pub fn search_repos(
+        &self,
+        run_id: &str,
+        query_lower: &str,
+        limit: usize,
+    ) -> QueryDbResult<Vec<RepositoryRecord>> {
+        let cf = self.cf(CF_REPOS)?;
+        let prefix = format!("repo/{run_id}/");
+        let mut out = Vec::new();
+        let mode = IteratorMode::From(prefix.as_bytes(), rocksdb::Direction::Forward);
+        for item in self.db.iterator_cf(cf, mode) {
+            let (key, value) = item?;
+            if !String::from_utf8_lossy(&key).starts_with(&prefix) {
+                break;
+            }
+            if out.len() >= limit {
+                break;
+            }
+            let repo: RepositoryRecord = serde_json::from_slice(&value)?;
+            if repo.host.to_lowercase().contains(query_lower)
+                || repo.uri.to_lowercase().contains(query_lower)
+            {
+                out.push(repo);
+            }
+        }
+        Ok(out)
+    }
+
+    pub fn search_publication_points(
+        &self,
+        run_id: &str,
+        query_lower: &str,
+        limit: usize,
+    ) -> QueryDbResult<Vec<PublicationPointRecord>> {
+        let cf = self.cf(CF_PUBLICATION_POINTS)?;
+        let prefix = format!("pp/{run_id}/");
+        let mut out = Vec::new();
+        let mode = IteratorMode::From(prefix.as_bytes(), rocksdb::Direction::Forward);
+        for item in self.db.iterator_cf(cf, mode) {
+            let (key, value) = item?;
+            if !String::from_utf8_lossy(&key).starts_with(&prefix) {
+                break;
+            }
+            if out.len() >= limit {
+                break;
+            }
+            let pp: PublicationPointRecord = serde_json::from_slice(&value)?;
+            let uri_matches = [
+                pp.manifest_rsync_uri.as_deref(),
+                pp.publication_point_rsync_uri.as_deref(),
+                pp.rrdp_notification_uri.as_deref(),
+                pp.rsync_base_uri.as_deref(),
+            ]
+            .into_iter()
+            .flatten()
+            .any(|uri| uri.to_lowercase().contains(query_lower));
+            if uri_matches {
+                out.push(pp);
+            }
+        }
+        Ok(out)
+    }
+
     pub fn get_stat(
         &self,
         run_id: &str,