src/state.rs
Rama main · réplica de lectura del repositorio original.
use std::collections::HashMap;
use std::sync::Mutex;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use melodl::SongInfo;
use serde::Serialize;
use uuid::Uuid;
use crate::music::SongView;
pub fn now_ts() -> f64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_secs_f64()
}
pub fn gen_id(len: usize) -> String {
Uuid::new_v4().simple().to_string()[..len].to_string()
}
/// TTL 缓存：惰性过期，put 时顺手 GC
pub struct TtlCache<T> {
ttl: Duration,
data: Mutex<HashMap<String, (Instant, T)>>,
}
impl<T: Clone> TtlCache<T> {
pub fn new(ttl_secs: u64) -> Self {
Self {
ttl: Duration::from_secs(ttl_secs),
data: Mutex::new(HashMap::new()),
}
}
pub fn put(&self, key: impl Into<String>, value: T) {
let mut data = self.data.lock().unwrap();
data.retain(|_, (ts, _)| ts.elapsed() <= self.ttl);
data.insert(key.into(), (Instant::now(), value));
}
pub fn get(&self, key: &str) -> Option<T> {
let mut data = self.data.lock().unwrap();
match data.get(key) {
Some((ts, _)) if ts.elapsed() > self.ttl => {
data.remove(key);
None
}
Some((_, v)) => Some(v.clone()),
None => None,
}
}
}
/// 异步搜索任务注册表 + 结果缓存（TTL 内同关键词去重）
pub struct SearchStore {
ttl: Duration,
inner: Mutex<SearchStoreInner>,
}
#[derive(Default)]
struct SearchStoreInner {
jobs: HashMap<String, SearchJob>,
by_keyword: HashMap<String, String>,
}
#[derive(Debug, Clone, Serialize)]
pub struct SearchJob {
pub id: String,
pub keyword: String,
pub status: String, // pending / running / done / failed
pub error: Option<String>,
pub created_at: f64,
pub started_at: Option<f64>,
pub finished_at: Option<f64>,
pub elapsed: Option<f64>,
pub sources_ok: Vec<String>,
pub sources_failed: Vec<String>,
pub data: Vec<SongView>,
#[serde(skip)]
pub raws: Vec<SongInfo>,
pub last_access: f64,
}
pub enum CreateOutcome {
/// 命中缓存（job 为完整结果）
Cached { job_id: String, job: SearchJob },
/// 同关键词已有任务在执行
Running { job_id: String },
/// 新建任务，调用方需入队执行
Created { job_id: String },
}
impl SearchStore {
pub fn new(ttl_secs: u64) -> Self {
Self {
ttl: Duration::from_secs(ttl_secs),
inner: Mutex::new(SearchStoreInner::default()),
}
}
pub fn create_or_get(&self, keyword: &str, cache: &TtlCache<SongInfo>) -> CreateOutcome {
let mut inner = self.inner.lock().unwrap();
self.gc(&mut inner);
if let Some(job_id) = inner.by_keyword.get(keyword).cloned() {
if let Some(job) = inner.jobs.get_mut(&job_id) {
match job.status.as_str() {
"pending" | "running" => return CreateOutcome::Running { job_id },
"done" => {
// 缓存命中：刷新 token 有效期，避免拿到手已过期
for (view, raw) in job.data.iter().zip(job.raws.iter()) {
cache.put(view.token.clone(), raw.clone());
}
job.last_access = now_ts();
return CreateOutcome::Cached {
job_id,
job: job.clone(),
};
}
_ => {} // failed 等：落入下面重建
}
}
}
let job_id = gen_id(12);
let job = SearchJob {
id: job_id.clone(),
keyword: keyword.to_string(),
status: "pending".to_string(),
error: None,
created_at: now_ts(),
started_at: None,
finished_at: None,
elapsed: None,
sources_ok: vec![],
sources_failed: vec![],
data: vec![],
raws: vec![],
last_access: now_ts(),
};
inner.jobs.insert(job_id.clone(), job);
inner.by_keyword.insert(keyword.to_string(), job_id.clone());
CreateOutcome::Created { job_id }
}
pub fn update(&self, job_id: &str, f: impl FnOnce(&mut SearchJob)) {
let mut inner = self.inner.lock().unwrap();
if let Some(job) = inner.jobs.get_mut(job_id) {
f(job);
}
}
pub fn get(&self, job_id: &str) -> Option<SearchJob> {
let mut inner = self.inner.lock().unwrap();
let job = inner.jobs.get_mut(job_id)?;
if job.status == "running" {
if let Some(started) = job.started_at {
job.elapsed = Some(((now_ts() - started) * 10.0).round() / 10.0);
}
}
Some(job.clone())
}
fn gc(&self, inner: &mut SearchStoreInner) {
let now = now_ts();
let ttl = self.ttl.as_secs_f64();
let expired: Vec<String> = inner
.jobs
.values()
.filter(|j| now - j.last_access > ttl)
.map(|j| j.id.clone())
.collect();
for jid in expired {
if let Some(job) = inner.jobs.remove(&jid) {
if inner.by_keyword.get(&job.keyword) == Some(&jid) {
inner.by_keyword.remove(&job.keyword);
}
}
}
}
}
/// 下载任务状态表（内存实现，重启清空）
#[derive(Default)]
pub struct TaskStore {
tasks: Mutex<HashMap<String, Task>>,
}
#[derive(Debug, Clone, Serialize)]
pub struct Task {
pub id: String,
pub title: String,
pub status: String, // queued / matching / downloading / done / failed / unmatched
pub file: Option<String>,
pub error: Option<String>,
pub created_at: f64,
pub batch_id: Option<String>,
pub source: Option<String>,
pub ext: Option<String>,
}
impl TaskStore {
pub fn new() -> Self {
Self::default()
}
pub fn create(&self, title: impl Into<String>, batch_id: Option<String>) -> String {
let task_id = gen_id(12);
let task = Task {
id: task_id.clone(),
title: title.into(),
status: "queued".to_string(),
file: None,
error: None,
created_at: now_ts(),
batch_id,
source: None,
ext: None,
};
self.tasks.lock().unwrap().insert(task_id.clone(), task);
task_id
}
pub fn update(&self, task_id: &str, f: impl FnOnce(&mut Task)) {
if let Some(task) = self.tasks.lock().unwrap().get_mut(task_id) {
f(task);
}
}
pub fn get(&self, task_id: &str) -> Option<Task> {
self.tasks.lock().unwrap().get(task_id).cloned()
}
pub fn list(&self, limit: usize) -> Vec<Task> {
let mut tasks: Vec<Task> = self.tasks.lock().unwrap().values().cloned().collect();
tasks.sort_by(|a, b| b.created_at.total_cmp(&a.created_at));
tasks.truncate(limit);
tasks
}
}
/// 串行任务队列：单 consumer 逐个处理，避免并发下载
pub fn spawn_worker<T, F, Fut>(handler: F) -> tokio::sync::mpsc::UnboundedSender<T>
where
T: Send + 'static,
F: Fn(T) -> Fut + Send + Sync + 'static,
Fut: std::future::Future<Output = ()> + Send + 'static,
{
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<T>();
let handler = std::sync::Arc::new(handler);
tokio::spawn(async move {
while let Some(item) = rx.recv().await {
(handler)(item).await;
}
});
tx
}