60e6c8f61b
本次提交完成了一项重要的架构重构，将所有外部服务的API凭证管理从环境变量迁移到了中心化的数据库配置中。
主要变更：
1. **统一配置源**:
- `data-persistence-service` 现已提供 `/api/v1/configs/data_sources` 端点，用于统一管理数据源配置。
- 所有配置（LLM 和数据源）现在都通过数据库的 `system_config` 表进行管理，实现了“单一事实源”。
2. **增强服务韧性**:
- 重构了 `finnhub-`, `tushare-`, `alphavantage-provider-service`。
- 这些服务在启动时不再强制要求 API Key。
- 引入了动态配置轮询器 (`config_poller`)，服务现在可以定期从数据库获取最新配置。
- 实现了“降级模式”：当配置缺失时，服务会进入 `Degraded` 状态并暂停处理消息，而不是直接崩溃。配置恢复后，服务会自动回到 `Active` 状态。
- `/health` 端点现在能准确反映服务的真实运行状态。
3. **前端易用性提升**:
- 您在 `/config` 页面上增加了“数据源配置”面板，允许用户通过 UI 动态更新所有 API Token。
4. **部署简化**:
- 从 `docker-compose.yml` 中移除了所有已废弃的 `_API_KEY` 环境变量，消除了启动时的警告。
这项重构显著提升了系统的可维护性、健壮性和用户体验，为未来的功能扩展奠定了坚实的基础。
94 lines
3.3 KiB
Rust
94 lines
3.3 KiB
Rust
use crate::error::{AppError, Result};
|
|
use crate::persistence::PersistenceClient;
|
|
use crate::state::AppState;
|
|
use chrono::Datelike;
|
|
use common_contracts::dtos::{CompanyProfileDto, TimeSeriesFinancialDto};
|
|
use common_contracts::messages::{CompanyProfilePersistedEvent, FetchCompanyDataCommand, FinancialsPersistedEvent};
|
|
use common_contracts::observability::TaskProgress;
|
|
use tracing::{error, info};
|
|
|
|
pub async fn handle_fetch_command(
|
|
state: AppState,
|
|
command: FetchCompanyDataCommand,
|
|
publisher: async_nats::Client,
|
|
) -> Result<()> {
|
|
info!("Handling Finnhub fetch data command.");
|
|
|
|
state.tasks.insert(
|
|
command.request_id,
|
|
TaskProgress {
|
|
request_id: command.request_id,
|
|
task_name: format!("finnhub:{}", command.symbol),
|
|
status: "FetchingData".to_string(),
|
|
progress_percent: 10,
|
|
details: "Fetching data from Finnhub".to_string(),
|
|
started_at: chrono::Utc::now(),
|
|
},
|
|
);
|
|
|
|
let provider = match state.get_provider().await {
|
|
Some(p) => p,
|
|
None => {
|
|
let reason = "Execution failed: Finnhub provider is not available (misconfigured).".to_string();
|
|
error!("{}", reason);
|
|
if let Some(mut task) = state.tasks.get_mut(&command.request_id) {
|
|
task.status = "Failed".to_string();
|
|
task.details = reason.clone();
|
|
}
|
|
return Err(AppError::ProviderNotAvailable(reason));
|
|
}
|
|
};
|
|
|
|
// 1. Fetch data via provider
|
|
let (profile, financials): (CompanyProfileDto, Vec<TimeSeriesFinancialDto>) =
|
|
provider.fetch_all_data(&command.symbol).await?;
|
|
|
|
// 2. Persist
|
|
{
|
|
if let Some(mut task) = state.tasks.get_mut(&command.request_id) {
|
|
task.status = "PersistingData".to_string();
|
|
task.progress_percent = 60;
|
|
task.details = "Persisting data to database".to_string();
|
|
}
|
|
}
|
|
let persistence_client = PersistenceClient::new(state.config.data_persistence_service_url.clone());
|
|
persistence_client.upsert_company_profile(profile).await?;
|
|
let years_set: std::collections::BTreeSet<u16> =
|
|
financials.iter().map(|f| f.period_date.year() as u16).collect();
|
|
persistence_client.batch_insert_financials(financials).await?;
|
|
|
|
// 3. Publish events
|
|
let profile_event = CompanyProfilePersistedEvent {
|
|
request_id: command.request_id,
|
|
symbol: command.symbol.clone(),
|
|
};
|
|
publisher
|
|
.publish(
|
|
"events.data.company_profile_persisted".to_string(),
|
|
serde_json::to_vec(&profile_event).unwrap().into(),
|
|
)
|
|
.await?;
|
|
|
|
let financials_event = FinancialsPersistedEvent {
|
|
request_id: command.request_id,
|
|
symbol: command.symbol.clone(),
|
|
years_updated: years_set.into_iter().collect(),
|
|
};
|
|
publisher
|
|
.publish(
|
|
"events.data.financials_persisted".to_string(),
|
|
serde_json::to_vec(&financials_event).unwrap().into(),
|
|
)
|
|
.await?;
|
|
|
|
// 4. Finalize
|
|
if let Some(mut task) = state.tasks.get_mut(&command.request_id) {
|
|
task.status = "Completed".to_string();
|
|
task.progress_percent = 100;
|
|
task.details = "Workflow finished successfully".to_string();
|
|
}
|
|
info!("Task {} completed successfully.", command.request_id);
|
|
Ok(())
|
|
}
|
|
|