use std::sync::{Arc, Mutex}; use tracing::info; use crate::data::repositories::project_repository::ProjectRepository; use crate::data::repositories::material_repository::MaterialRepository; use crate::data::repositories::model_repository::ModelRepository; use crate::data::repositories::model_dynamic_repository::ModelDynamicRepository; use crate::data::repositories::video_generation_repository::VideoGenerationRepository; use crate::data::repositories::conversation_repository::ConversationRepository; use crate::data::repositories::outfit_photo_generation_repository::OutfitPhotoGenerationRepository; use crate::data::repositories::hedra_lipsync_repository::HedraLipSyncRepository; use crate::data::repositories::workflow_template_repository::WorkflowTemplateRepository; use crate::data::repositories::workflow_execution_record_repository::WorkflowExecutionRecordRepository; use crate::data::repositories::workflow_execution_environment_repository::WorkflowExecutionEnvironmentRepository; use crate::infrastructure::database::Database; use crate::infrastructure::monitoring::PerformanceMonitor; use crate::infrastructure::event_bus::EventBusManager; use crate::infrastructure::bowong_text_video_agent_service::BowongTextVideoAgentService; use crate::infrastructure::comfyui_service::ComfyuiInfrastructureService; use crate::business::services::universal_workflow_service::UniversalWorkflowService; /// 应用全局状态管理 /// 遵循 Tauri 开发规范的状态管理模式 pub struct AppState { pub database: Mutex>>, pub project_repository: Mutex>, pub material_repository: Mutex>, pub model_repository: Mutex>, pub model_dynamic_repository: Mutex>, pub video_generation_repository: Mutex>, pub conversation_repository: Mutex>>, pub outfit_photo_generation_repository: Mutex>, pub hedra_lipsync_repository: Mutex>>, pub workflow_template_repository: Mutex>, pub workflow_execution_record_repository: Mutex>, pub workflow_execution_environment_repository: Mutex>, pub performance_monitor: Mutex, pub event_bus_manager: Arc, pub bowong_text_video_agent_service: Mutex>, pub comfyui_infrastructure_service: Mutex>, pub universal_workflow_service: Mutex>, } impl AppState { pub fn new() -> Self { Self { database: Mutex::new(None), project_repository: Mutex::new(None), material_repository: Mutex::new(None), model_repository: Mutex::new(None), model_dynamic_repository: Mutex::new(None), video_generation_repository: Mutex::new(None), conversation_repository: Mutex::new(None), outfit_photo_generation_repository: Mutex::new(None), hedra_lipsync_repository: Mutex::new(None), workflow_template_repository: Mutex::new(None), workflow_execution_record_repository: Mutex::new(None), workflow_execution_environment_repository: Mutex::new(None), performance_monitor: Mutex::new(PerformanceMonitor::new()), event_bus_manager: Arc::new(EventBusManager::new()), bowong_text_video_agent_service: Mutex::new(None), comfyui_infrastructure_service: Mutex::new(None), universal_workflow_service: Mutex::new(None), } } /// 初始化数据库连接 /// 遵循安全第一原则,确保数据库初始化的安全性 /// 默认使用连接池模式以提高并发性能 pub fn initialize_database(&self) -> anyhow::Result<()> { println!("开始初始化数据库连接..."); // 暂时使用单连接模式,避免锁竞争问题 // TODO: 在解决 SQLite 锁问题后重新启用连接池 let database = Arc::new(Database::new()?); println!("使用单连接模式初始化数据库"); // 连接池模式代码(暂时注释) /* let database = match Database::new_with_pool() { Ok(db) => { println!("连接池模式初始化成功"); Arc::new(db) }, Err(e) => { eprintln!("连接池模式初始化失败: {}, 回退到单连接模式", e); Arc::new(Database::new()?) } }; */ let project_repository = ProjectRepository::new(database.clone())?; let material_repository = MaterialRepository::new(database.clone())?; let model_repository = ModelRepository::new(database.clone()); let model_dynamic_repository = ModelDynamicRepository::new(database.clone()); let video_generation_repository = VideoGenerationRepository::new(database.clone()); let conversation_repository = Arc::new(ConversationRepository::new(database.clone())); // 初始化数据库表 model_dynamic_repository.init_tables()?; video_generation_repository.init_tables()?; conversation_repository.initialize_tables()?; // 初始化穿搭图片相关表 let outfit_image_repository = crate::data::repositories::outfit_image_repository::OutfitImageRepository::new(database.clone()); outfit_image_repository.init_tables()?; // 初始化穿搭照片生成相关表 let outfit_photo_generation_repository = OutfitPhotoGenerationRepository::new(database.clone())?; // 初始化Hedra口型合成记录表 let hedra_lipsync_repository = crate::data::repositories::hedra_lipsync_repository::HedraLipSyncRepository::new(database.clone()); hedra_lipsync_repository.init_tables()?; let hedra_lipsync_repository = Arc::new(hedra_lipsync_repository); // 初始化工作流系统相关仓库 let workflow_template_repository = WorkflowTemplateRepository::new(database.clone()); let workflow_execution_record_repository = WorkflowExecutionRecordRepository::new(database.clone()); let workflow_execution_environment_repository = WorkflowExecutionEnvironmentRepository::new(database.clone()); // 初始化通用工作流服务(需要ComfyUI服务) let universal_workflow_service = if let Some(comfyui_service) = self.comfyui_infrastructure_service.lock().unwrap().as_ref() { Some(UniversalWorkflowService::new( database.clone(), Arc::new(comfyui_service.clone()), workflow_template_repository.clone(), workflow_execution_record_repository.clone(), workflow_execution_environment_repository.clone(), )) } else { None }; *self.database.lock().unwrap() = Some(database.clone()); *self.project_repository.lock().unwrap() = Some(project_repository); *self.material_repository.lock().unwrap() = Some(material_repository); *self.model_repository.lock().unwrap() = Some(model_repository); *self.model_dynamic_repository.lock().unwrap() = Some(model_dynamic_repository); *self.video_generation_repository.lock().unwrap() = Some(video_generation_repository); *self.conversation_repository.lock().unwrap() = Some(conversation_repository); *self.outfit_photo_generation_repository.lock().unwrap() = Some(outfit_photo_generation_repository); *self.hedra_lipsync_repository.lock().unwrap() = Some(hedra_lipsync_repository); *self.workflow_template_repository.lock().unwrap() = Some(workflow_template_repository); *self.workflow_execution_record_repository.lock().unwrap() = Some(workflow_execution_record_repository); *self.workflow_execution_environment_repository.lock().unwrap() = Some(workflow_execution_environment_repository); *self.universal_workflow_service.lock().unwrap() = universal_workflow_service; println!("数据库初始化完成,连接池状态: {}", if database.has_pool() { "已启用" } else { "未启用" }); Ok(()) } /// 初始化数据库连接(单连接模式,用于测试或特殊场景) pub fn initialize_database_single_mode(&self) -> anyhow::Result<()> { let database = Arc::new(Database::new()?); let project_repository = ProjectRepository::new(database.clone())?; let material_repository = MaterialRepository::new(database.clone())?; let model_repository = ModelRepository::new(database.clone()); let model_dynamic_repository = ModelDynamicRepository::new(database.clone()); let video_generation_repository = VideoGenerationRepository::new(database.clone()); let conversation_repository = Arc::new(ConversationRepository::new(database.clone())); // 初始化数据库表 model_dynamic_repository.init_tables()?; video_generation_repository.init_tables()?; conversation_repository.initialize_tables()?; // 初始化穿搭图片相关表 let outfit_image_repository = crate::data::repositories::outfit_image_repository::OutfitImageRepository::new(database.clone()); outfit_image_repository.init_tables()?; // 初始化穿搭照片生成相关表 let outfit_photo_generation_repository = OutfitPhotoGenerationRepository::new(database.clone())?; // 初始化Hedra口型合成记录表 let hedra_lipsync_repository = crate::data::repositories::hedra_lipsync_repository::HedraLipSyncRepository::new(database.clone()); hedra_lipsync_repository.init_tables()?; let hedra_lipsync_repository = Arc::new(hedra_lipsync_repository); // 初始化工作流系统相关仓库 let workflow_template_repository = WorkflowTemplateRepository::new(database.clone()); let workflow_execution_record_repository = WorkflowExecutionRecordRepository::new(database.clone()); let workflow_execution_environment_repository = WorkflowExecutionEnvironmentRepository::new(database.clone()); // 初始化通用工作流服务(需要ComfyUI服务) let universal_workflow_service = if let Some(comfyui_service) = self.comfyui_infrastructure_service.lock().unwrap().as_ref() { Some(UniversalWorkflowService::new( database.clone(), Arc::new(comfyui_service.clone()), workflow_template_repository.clone(), workflow_execution_record_repository.clone(), workflow_execution_environment_repository.clone(), )) } else { None }; *self.database.lock().unwrap() = Some(database.clone()); *self.project_repository.lock().unwrap() = Some(project_repository); *self.material_repository.lock().unwrap() = Some(material_repository); *self.model_repository.lock().unwrap() = Some(model_repository); *self.model_dynamic_repository.lock().unwrap() = Some(model_dynamic_repository); *self.video_generation_repository.lock().unwrap() = Some(video_generation_repository); *self.conversation_repository.lock().unwrap() = Some(conversation_repository); *self.outfit_photo_generation_repository.lock().unwrap() = Some(outfit_photo_generation_repository); *self.hedra_lipsync_repository.lock().unwrap() = Some(hedra_lipsync_repository); *self.workflow_template_repository.lock().unwrap() = Some(workflow_template_repository); *self.workflow_execution_record_repository.lock().unwrap() = Some(workflow_execution_record_repository); *self.workflow_execution_environment_repository.lock().unwrap() = Some(workflow_execution_environment_repository); *self.universal_workflow_service.lock().unwrap() = universal_workflow_service; println!("数据库初始化完成,使用单连接模式"); Ok(()) } /// 获取项目仓库实例 pub fn get_project_repository(&self) -> anyhow::Result>> { Ok(self.project_repository.lock().unwrap()) } /// 获取素材仓库实例 pub fn get_material_repository(&self) -> anyhow::Result>> { Ok(self.material_repository.lock().unwrap()) } /// 获取模特仓库实例 pub fn get_model_repository(&self) -> anyhow::Result>> { Ok(self.model_repository.lock().unwrap()) } /// 获取模特动态仓库实例 pub fn get_model_dynamic_repository(&self) -> anyhow::Result>> { Ok(self.model_dynamic_repository.lock().unwrap()) } /// 获取视频生成仓库实例 pub fn get_video_generation_repository(&self) -> anyhow::Result>> { Ok(self.video_generation_repository.lock().unwrap()) } /// 获取会话仓库实例 pub fn get_conversation_repository(&self) -> anyhow::Result> { let repo_guard = self.conversation_repository.lock().unwrap(); repo_guard.as_ref() .ok_or_else(|| anyhow::anyhow!("会话仓库未初始化")) .map(|repo| repo.clone()) } /// 获取工作流模板仓库实例 pub fn get_workflow_template_repository(&self) -> anyhow::Result>> { Ok(self.workflow_template_repository.lock().unwrap()) } /// 获取工作流执行记录仓库实例 pub fn get_workflow_execution_record_repository(&self) -> anyhow::Result>> { Ok(self.workflow_execution_record_repository.lock().unwrap()) } /// 获取工作流执行环境仓库实例 pub fn get_workflow_execution_environment_repository(&self) -> anyhow::Result>> { Ok(self.workflow_execution_environment_repository.lock().unwrap()) } /// 获取通用工作流服务实例 pub fn get_universal_workflow_service(&self) -> anyhow::Result>> { Ok(self.universal_workflow_service.lock().unwrap()) } /// 获取数据库实例 pub fn get_database(&self) -> Arc { // 使用全局静态数据库实例,确保整个应用只有一个数据库实例 use std::sync::OnceLock; static DATABASE_INSTANCE: OnceLock> = OnceLock::new(); DATABASE_INSTANCE.get_or_init(|| { // 只在第一次调用时初始化数据库 Arc::new(Database::new().unwrap()) }).clone() } /// 获取数据库连接 pub fn get_connection(&self) -> anyhow::Result>> { let database = self.get_database(); Ok(database.get_connection()) } /// 用于测试的构造函数 #[cfg(test)] pub fn new_with_database(_database: Arc) -> Self { let state = Self { database: Mutex::new(None), project_repository: Mutex::new(None), material_repository: Mutex::new(None), model_repository: Mutex::new(None), model_dynamic_repository: Mutex::new(None), video_generation_repository: Mutex::new(None), conversation_repository: Mutex::new(None), outfit_photo_generation_repository: Mutex::new(None), hedra_lipsync_repository: Mutex::new(None), workflow_template_repository: Mutex::new(None), workflow_execution_record_repository: Mutex::new(None), workflow_execution_environment_repository: Mutex::new(None), performance_monitor: Mutex::new(PerformanceMonitor::new()), event_bus_manager: Arc::new(EventBusManager::new()), bowong_text_video_agent_service: Mutex::new(None), comfyui_infrastructure_service: Mutex::new(None), universal_workflow_service: Mutex::new(None), }; // 不直接存储database,而是在需要时返回传入的database state } /// 初始化 BowongTextVideoAgent 服务 pub fn initialize_bowong_service(&self, config: crate::data::models::bowong_text_video_agent::BowongTextVideoAgentConfig) -> anyhow::Result<()> { let service = BowongTextVideoAgentService::new(config)?; let mut bowong_service = self.bowong_text_video_agent_service.lock() .map_err(|e| anyhow::anyhow!("Failed to acquire lock for BowongTextVideoAgent service: {}", e))?; *bowong_service = Some(service); println!("✅ BowongTextVideoAgent 服务初始化成功"); Ok(()) } /// 获取 BowongTextVideoAgent 服务的引用 pub fn get_bowong_service(&self) -> anyhow::Result>> { self.bowong_text_video_agent_service.lock() .map_err(|e| anyhow::anyhow!("Failed to acquire lock for BowongTextVideoAgent service: {}", e)) } /// 初始化 ComfyUI Infrastructure 服务 pub fn initialize_comfyui_service(&self, config: crate::data::models::comfyui::ComfyuiConfig) -> anyhow::Result<()> { let service = ComfyuiInfrastructureService::new(config)?; let mut comfyui_service = self.comfyui_infrastructure_service.lock() .map_err(|e| anyhow::anyhow!("Failed to acquire lock for ComfyUI Infrastructure service: {}", e))?; *comfyui_service = Some(service); println!("✅ ComfyUI Infrastructure 服务初始化成功"); Ok(()) } /// 获取 ComfyUI Infrastructure 服务的引用 pub fn get_comfyui_service(&self) -> Option { match self.comfyui_infrastructure_service.lock() { Ok(guard) => guard.clone(), Err(e) => { eprintln!("Failed to acquire lock for ComfyUI Infrastructure service: {}", e); None } } } /// 更新 ComfyUI Infrastructure 服务配置 pub async fn update_comfyui_config(&self, config: crate::data::models::comfyui::ComfyuiConfig) -> anyhow::Result<()> { // 创建新的服务实例 let new_service = ComfyuiInfrastructureService::new(config)?; // 更新服务 let mut comfyui_service = self.comfyui_infrastructure_service.lock() .map_err(|e| anyhow::anyhow!("Failed to acquire lock for ComfyUI Infrastructure service: {}", e))?; *comfyui_service = Some(new_service); println!("✅ ComfyUI Infrastructure 服务配置更新成功"); // 重新初始化通用工作流服务 self.initialize_universal_workflow_service()?; Ok(()) } /// 初始化通用工作流服务 pub fn initialize_universal_workflow_service(&self) -> anyhow::Result<()> { // 首先尝试初始化 ComfyUI 服务(如果还没有初始化) { let comfyui_service_guard = self.comfyui_infrastructure_service.lock() .map_err(|e| anyhow::anyhow!("Failed to acquire lock for ComfyUI service: {}", e))?; if comfyui_service_guard.is_none() { drop(comfyui_service_guard); // 释放锁 info!("ComfyUI服务未初始化,正在初始化..."); // 使用默认配置初始化 ComfyUI 服务 let default_config = crate::data::models::comfyui::ComfyuiConfig::default(); let comfyui_service = ComfyuiInfrastructureService::new(default_config) .map_err(|e| anyhow::anyhow!("创建ComfyUI服务失败: {}", e))?; *self.comfyui_infrastructure_service.lock().unwrap() = Some(comfyui_service); info!("ComfyUI服务初始化成功"); } } // 检查依赖服务是否已初始化 let comfyui_service = self.comfyui_infrastructure_service.lock() .map_err(|e| anyhow::anyhow!("Failed to acquire lock for ComfyUI service: {}", e))?; let comfyui_service = comfyui_service.as_ref() .ok_or_else(|| anyhow::anyhow!("ComfyUI服务初始化失败"))?; // 检查工作流仓库是否已初始化 let workflow_template_repo = self.workflow_template_repository.lock().unwrap(); let workflow_execution_record_repo = self.workflow_execution_record_repository.lock().unwrap(); let workflow_execution_environment_repo = self.workflow_execution_environment_repository.lock().unwrap(); if let (Some(template_repo), Some(record_repo), Some(env_repo)) = ( workflow_template_repo.as_ref(), workflow_execution_record_repo.as_ref(), workflow_execution_environment_repo.as_ref() ) { let database = self.get_database(); let universal_service = UniversalWorkflowService::new( database, Arc::new(comfyui_service.clone()), template_repo.clone(), record_repo.clone(), env_repo.clone(), ); *self.universal_workflow_service.lock().unwrap() = Some(universal_service); println!("✅ 通用工作流服务初始化成功"); } else { return Err(anyhow::anyhow!("工作流仓库未初始化,无法创建通用工作流服务")); } Ok(()) } /// 启动监控服务 pub async fn start_monitoring_service(&self) -> anyhow::Result<()> { use crate::business::services::workflow_monitoring_service::{WorkflowMonitoringService, MonitoringConfig}; info!("正在启动监控服务..."); // 获取必要的仓库 let execution_record_repo = { let repo_guard = self.workflow_execution_record_repository.lock().unwrap(); repo_guard.as_ref() .ok_or_else(|| anyhow::anyhow!("工作流执行记录仓库未初始化"))? .clone() }; let execution_environment_repo = { let repo_guard = self.workflow_execution_environment_repository.lock().unwrap(); repo_guard.as_ref() .ok_or_else(|| anyhow::anyhow!("工作流执行环境仓库未初始化"))? .clone() }; // 创建监控服务 let monitoring_service = WorkflowMonitoringService::new( execution_record_repo, execution_environment_repo, Some(MonitoringConfig::default()), ); // 启动监控服务 monitoring_service.start_monitoring().await?; info!("监控服务启动成功"); Ok(()) } } impl Default for AppState { fn default() -> Self { Self::new() } }