Asfmq
2c8d0b8f8b
- LAMOST 新增 DR12/13/14 及子版本（v0/v1.0/v1.1/v2.0）维度，Internal 发布标记需登录认证并前端灰显，release×subtype 交叉约束下沉至 capabilities 统一声明 - ObservationFetcher trait 扩展版本/认证/交叉约束能力声明，version 参数贯穿 client→service→API→Agent tool→前端全链路 - 安全：observation cache SQL 全参数绑定 + LIKE 转义、cone_cache_hash 加长度前缀防碰撞、DESI survey/program 白名单防穿越 - 异步化：persist/cached_files_total_size/maybe_persist_tool_result迁移到 tokio::fs；cancelled_runs 与 session_permission_checkers改用 DashMap；auth 读锁优先 + 60s 节流 - Gaia 去 native-tls 改禁用连接池规避 UnexpectedEof，reqwest 移除 native-tls feature - 重构：Source/ProductType from_str 集中解析、download 模块拆分为 try_download_pdf/html、AgentRuntime::init 抽取共享逻辑 - 部署：新增 deploy.sh 一键打包推送脚本、catch-panic 启用
361 lines
13 KiB
Rust
361 lines
13 KiB
Rust
// src/services/observation/cache.rs
|
||
//
|
||
// 观测数据缓存层 + 落盘辅助
|
||
//
|
||
// observation_cache 表设计：
|
||
// - UNIQUE(source, product, source_id)：同一源的同一标识符可同时缓存不同产品
|
||
// （如 Gaia 同一 source_id 的光谱与光变分别独立缓存）
|
||
// - artifacts_json：一个逻辑产物可对应多个物理文件（Gaia 光变 G/BP/RP 三波段）
|
||
|
||
use crate::services::observation::types::{Artifact, ProductSpec, Source};
|
||
use anyhow::{anyhow, Result};
|
||
use serde::{Deserialize, Serialize};
|
||
use sqlx::SqlitePool;
|
||
use std::path::Path;
|
||
use tracing::error;
|
||
|
||
// ═══════════════════════════════════════════════════════════════
|
||
// 1. 缓存命中行类型
|
||
// ═══════════════════════════════════════════════════════════════
|
||
|
||
/// observation_cache 中单条 artifacts 的 JSON 结构
|
||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||
struct ArtifactRecord {
|
||
path: String,
|
||
format: String,
|
||
#[serde(default)]
|
||
size: usize,
|
||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||
band: Option<String>,
|
||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||
original_name: Option<String>,
|
||
}
|
||
|
||
impl From<&ArtifactRecord> for Artifact {
|
||
fn from(r: &ArtifactRecord) -> Self {
|
||
Artifact {
|
||
band: r.band.clone(),
|
||
original_name: r.original_name.clone(),
|
||
file_path: r.path.clone(),
|
||
file_url: crate::services::observation::cache::file_url_from_path(&r.path),
|
||
file_format: r.format.clone(),
|
||
size_bytes: r.size,
|
||
cached: true,
|
||
}
|
||
}
|
||
}
|
||
|
||
/// 缓存查询结果
|
||
#[derive(sqlx::FromRow)]
|
||
struct CacheHitRow {
|
||
artifacts_json: String,
|
||
meta_json: Option<String>,
|
||
}
|
||
|
||
/// observation_cache 列表查询行（对外暴露）
|
||
#[derive(Debug, Clone, Serialize, sqlx::FromRow)]
|
||
pub struct ObservationCacheRow {
|
||
pub source: String,
|
||
pub product: String,
|
||
pub source_id: String,
|
||
pub ra: Option<f64>,
|
||
pub dec: Option<f64>,
|
||
pub artifacts_json: String,
|
||
pub meta_json: Option<String>,
|
||
pub created_at: String,
|
||
}
|
||
|
||
// ═══════════════════════════════════════════════════════════════
|
||
// 2. 缓存读写
|
||
// ═══════════════════════════════════════════════════════════════
|
||
|
||
/// 查询单条缓存（按 source + product + source_id），返回命中的 artifacts
|
||
pub async fn fetch_observation_cache(
|
||
pool: &SqlitePool,
|
||
source: Source,
|
||
product: &ProductSpec,
|
||
source_id: &str,
|
||
) -> Result<Option<(Vec<Artifact>, Option<serde_json::Value>)>> {
|
||
let row = sqlx::query_as::<_, CacheHitRow>(
|
||
"SELECT artifacts_json, meta_json FROM observation_cache \
|
||
WHERE source = ? AND product = ? AND source_id = ? LIMIT 1",
|
||
)
|
||
.bind(source.as_str())
|
||
.bind(product.product.as_str())
|
||
.bind(source_id)
|
||
.fetch_optional(pool)
|
||
.await
|
||
.map_err(|e| anyhow!("查询 observation_cache 失败: {}", e))?;
|
||
|
||
match row {
|
||
Some(r) => {
|
||
let records: Vec<ArtifactRecord> = serde_json::from_str(&r.artifacts_json)
|
||
.map_err(|e| anyhow!("反序列化 artifacts_json 失败: {}", e))?;
|
||
let artifacts: Vec<Artifact> = records.iter().map(Artifact::from).collect();
|
||
let meta = r
|
||
.meta_json
|
||
.as_deref()
|
||
.and_then(|s| serde_json::from_str(s).ok());
|
||
Ok(Some((artifacts, meta)))
|
||
}
|
||
None => Ok(None),
|
||
}
|
||
}
|
||
|
||
/// 写入/更新缓存条目
|
||
pub async fn write_observation_cache(
|
||
pool: &SqlitePool,
|
||
source: Source,
|
||
product: &ProductSpec,
|
||
source_id: &str,
|
||
ra: Option<f64>,
|
||
dec: Option<f64>,
|
||
artifacts: &[Artifact],
|
||
meta_json: Option<&str>,
|
||
) -> Result<()> {
|
||
let records: Vec<ArtifactRecord> = artifacts
|
||
.iter()
|
||
.map(|a| ArtifactRecord {
|
||
path: a.file_path.clone(),
|
||
format: a.file_format.clone(),
|
||
size: a.size_bytes,
|
||
band: a.band.clone(),
|
||
original_name: a.original_name.clone(),
|
||
})
|
||
.collect();
|
||
let artifacts_json = serde_json::to_string(&records)?;
|
||
sqlx::query(
|
||
"INSERT OR REPLACE INTO observation_cache \
|
||
(source, product, source_id, ra, dec, artifacts_json, meta_json, created_at) \
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)",
|
||
)
|
||
.bind(source.as_str())
|
||
.bind(product.product.as_str())
|
||
.bind(source_id)
|
||
.bind(ra)
|
||
.bind(dec)
|
||
.bind(&artifacts_json)
|
||
.bind(meta_json)
|
||
.execute(pool)
|
||
.await
|
||
.map_err(|e| anyhow!("写入 observation_cache 失败: {}", e))?;
|
||
Ok(())
|
||
}
|
||
|
||
/// 列出某源某产品的全部已缓存产物
|
||
pub async fn list_cached(
|
||
pool: &SqlitePool,
|
||
source: Source,
|
||
product: &ProductSpec,
|
||
) -> Result<Vec<ObservationCacheRow>> {
|
||
let rows = sqlx::query_as::<_, ObservationCacheRow>(
|
||
"SELECT source, product, source_id, ra, dec, artifacts_json, meta_json, created_at \
|
||
FROM observation_cache WHERE source = ? AND product = ? \
|
||
ORDER BY created_at DESC LIMIT 500",
|
||
)
|
||
.bind(source.as_str())
|
||
.bind(product.product.as_str())
|
||
.fetch_all(pool)
|
||
.await
|
||
.map_err(|e| anyhow!("查询 observation_cache 列表失败: {}", e))?;
|
||
Ok(rows)
|
||
}
|
||
|
||
/// 列出全部已缓存产物（不限源/产品）
|
||
pub async fn list_all_cached(pool: &SqlitePool) -> Result<Vec<ObservationCacheRow>> {
|
||
let rows = sqlx::query_as::<_, ObservationCacheRow>(
|
||
"SELECT source, product, source_id, ra, dec, artifacts_json, meta_json, created_at \
|
||
FROM observation_cache ORDER BY created_at DESC LIMIT 500",
|
||
)
|
||
.fetch_all(pool)
|
||
.await
|
||
.map_err(|e| anyhow!("查询 observation_cache 列表失败: {}", e))?;
|
||
Ok(rows)
|
||
}
|
||
|
||
/// 缓存库列表筛选 + 排序参数（服务端分页）
|
||
#[derive(Debug, Clone, Default)]
|
||
pub struct ListCachedParams {
|
||
/// 按数据源过滤（lamost/gaia/sdss/desi）
|
||
pub source: Option<String>,
|
||
/// 按产品类型过滤（spectrum/lightcurve/photometry/image）
|
||
pub product: Option<String>,
|
||
/// source_id 模糊匹配（LIKE %q%）
|
||
pub q: Option<String>,
|
||
/// 排序：created（默认，DESC）/ source / product
|
||
pub sort: Option<String>,
|
||
/// 每页条数（默认 50，上限 200）
|
||
pub limit: Option<i64>,
|
||
/// 偏移量
|
||
pub offset: Option<i64>,
|
||
}
|
||
|
||
/// 缓存库分页查询结果
|
||
#[derive(Debug, Clone, Serialize)]
|
||
pub struct CachedPage {
|
||
pub items: Vec<ObservationCacheRow>,
|
||
pub total: i64,
|
||
}
|
||
|
||
/// 列出已缓存产物（服务端分页 + 筛选 + 排序）
|
||
///
|
||
/// 所有筛选值通过 SQL 参数绑定（`?`）传入，杜绝 SQL 注入与 LIKE 通配符（`%`/`_`）误匹配。
|
||
/// 排序方向用白名单（只接受 created/source/product 三个字面量）。
|
||
pub async fn list_cached_paged(pool: &SqlitePool, params: &ListCachedParams) -> Result<CachedPage> {
|
||
let limit = params.limit.unwrap_or(50).clamp(1, 200);
|
||
let offset = params.offset.unwrap_or(0).max(0);
|
||
|
||
// 动态构造 WHERE 子句（用 ? 占位符）+ 同步收集绑定值
|
||
// bind_values 顺序必须与占位符出现顺序一致：source → product → q → limit → offset
|
||
let mut where_parts: Vec<&'static str> = Vec::new();
|
||
let mut bind_source: Option<String> = None;
|
||
let mut bind_product: Option<String> = None;
|
||
let mut bind_q: Option<String> = None; // 已包成 '%q%'，且 LIKE 通配符已转义
|
||
|
||
if let Some(s) = ¶ms.source {
|
||
where_parts.push("source = ?");
|
||
bind_source = Some(s.clone());
|
||
}
|
||
if let Some(p) = ¶ms.product {
|
||
where_parts.push("product = ?");
|
||
bind_product = Some(p.clone());
|
||
}
|
||
if let Some(q) = ¶ms.q {
|
||
// 转义 LIKE 通配符，避免用户输入的 %/_ 被解释为模式
|
||
let escaped = q.replace('%', "\\%").replace('_', "\\_");
|
||
where_parts.push("source_id LIKE ? ESCAPE '\\'");
|
||
bind_q = Some(format!("%{}%", escaped));
|
||
}
|
||
|
||
let where_sql = if where_parts.is_empty() {
|
||
String::new()
|
||
} else {
|
||
format!(" WHERE {}", where_parts.join(" AND "))
|
||
};
|
||
|
||
let order_sql = match params.sort.as_deref() {
|
||
Some("source") => " ORDER BY source ASC, created_at DESC",
|
||
Some("product") => " ORDER BY product ASC, created_at DESC",
|
||
_ => " ORDER BY created_at DESC",
|
||
};
|
||
|
||
// count（绑定 source/product/q，不绑 limit/offset）
|
||
let count_sql = format!("SELECT COUNT(*) FROM observation_cache{}", where_sql);
|
||
let mut count_q = sqlx::query_scalar::<_, i64>(&count_sql);
|
||
if let Some(v) = &bind_source {
|
||
count_q = count_q.bind(v);
|
||
}
|
||
if let Some(v) = &bind_product {
|
||
count_q = count_q.bind(v);
|
||
}
|
||
if let Some(v) = &bind_q {
|
||
count_q = count_q.bind(v);
|
||
}
|
||
let total = count_q
|
||
.fetch_one(pool)
|
||
.await
|
||
.map_err(|e| anyhow!("统计 observation_cache 总数失败: {}", e))?;
|
||
|
||
// items（绑定 source/product/q/limit/offset）
|
||
let items_sql = format!(
|
||
"SELECT source, product, source_id, ra, dec, artifacts_json, meta_json, created_at \
|
||
FROM observation_cache{}{} LIMIT ? OFFSET ?",
|
||
where_sql, order_sql
|
||
);
|
||
let mut items_q = sqlx::query_as::<_, ObservationCacheRow>(&items_sql);
|
||
if let Some(v) = &bind_source {
|
||
items_q = items_q.bind(v);
|
||
}
|
||
if let Some(v) = &bind_product {
|
||
items_q = items_q.bind(v);
|
||
}
|
||
if let Some(v) = &bind_q {
|
||
items_q = items_q.bind(v);
|
||
}
|
||
items_q = items_q.bind(limit).bind(offset);
|
||
let items = items_q
|
||
.fetch_all(pool)
|
||
.await
|
||
.map_err(|e| anyhow!("查询 observation_cache 分页列表失败: {}", e))?;
|
||
|
||
Ok(CachedPage { items, total })
|
||
}
|
||
|
||
// ═══════════════════════════════════════════════════════════════
|
||
// 3. 落盘 + URL 辅助
|
||
// ═══════════════════════════════════════════════════════════════
|
||
|
||
/// 经 /api/files 前缀生成访问 URL
|
||
pub fn file_url_from_path(rel_path: &str) -> String {
|
||
format!("/api/files/{}", rel_path)
|
||
}
|
||
|
||
/// 落盘字节到 library_dir/<rel_path>，返回字节数
|
||
pub async fn persist_bytes(library_dir: &Path, rel_path: &str, bytes: &[u8]) -> Result<usize> {
|
||
let abs_path = library_dir.join(rel_path);
|
||
if let Some(parent) = abs_path.parent() {
|
||
tokio::fs::create_dir_all(parent)
|
||
.await
|
||
.map_err(|e| anyhow!("创建观测数据目录失败 {:?}: {}", parent, e))?;
|
||
}
|
||
tokio::fs::write(&abs_path, bytes)
|
||
.await
|
||
.map_err(|e| anyhow!("写入观测数据文件失败 {:?}: {}", abs_path, e))?;
|
||
Ok(bytes.len())
|
||
}
|
||
|
||
/// 校验缓存产物中至少有一个文件真实存在；存在则返回其总大小
|
||
pub async fn cached_files_total_size(library_dir: &Path, artifacts: &[Artifact]) -> Option<usize> {
|
||
let mut total = 0;
|
||
for a in artifacts {
|
||
let abs_path = library_dir.join(&a.file_path);
|
||
match tokio::fs::metadata(&abs_path).await {
|
||
Ok(m) => total += m.len() as usize,
|
||
Err(_) => {
|
||
return None;
|
||
}
|
||
}
|
||
}
|
||
if total == 0 {
|
||
None
|
||
} else {
|
||
Some(total)
|
||
}
|
||
}
|
||
|
||
/// 记录缓存写入失败但不中断流程
|
||
pub fn log_write_failure(source: &str, product: &str, e: anyhow::Error) {
|
||
error!(
|
||
"[Observation] 写入 observation_cache 失败 ({}/{}): {}",
|
||
source, product, e
|
||
);
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
|
||
#[test]
|
||
fn test_file_url_from_path() {
|
||
assert_eq!(
|
||
file_url_from_path("Telescope/lamost/spectrum/123.fits"),
|
||
"/api/files/Telescope/lamost/spectrum/123.fits"
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn test_artifact_record_roundtrip() {
|
||
let rec = ArtifactRecord {
|
||
path: "Telescope/gaia/lightcurve/123/123_G.fits".into(),
|
||
format: "fits".into(),
|
||
size: 1024,
|
||
band: Some("G".into()),
|
||
original_name: Some("123_EPOCH_PHOTOMETRY_G.fits".into()),
|
||
};
|
||
let json = serde_json::to_string(&rec).unwrap();
|
||
let back: ArtifactRecord = serde_json::from_str(&json).unwrap();
|
||
assert_eq!(back.path, rec.path);
|
||
assert_eq!(back.band.as_deref(), Some("G"));
|
||
}
|
||
}
|