use async_trait::async_trait;
use parking_lot::RwLock;
use std::sync::Arc;
use std::time::Duration;
use zerolaunch_plugin_api::config::{
ComponentCore, ComponentType, ConfigActionDef, ConfigError, Configurable, SettingDefinition,
};
use zerolaunch_plugin_api::{
ActionExecutor, CachedCandidateData, CandidateId, DataSource, ExecutionContext, ExecutionError,
KeywordInjector, KeywordOptimizer, PanelInteraction, Plugin, PluginContext, PluginError,
PluginHandle, PluginMetadata, Query, QueryResponse, ResultAction, ScoreBooster,
ScoredCandidate, SearchCandidate, SearchEngine, TargetType,
};
use crate::client::JsonRpcClient;
use zerolaunch_plugin_protocol::messages::*;
use zerolaunch_plugin_protocol::methods::plugin as plugin_methods;
use zerolaunch_plugin_protocol::{codes, ProtocolError};
#[derive(Debug)]
pub enum RemoteComponentKind {
DataSource,
ActionExecutor {
target_types: Vec<TargetType>,
result_actions: Vec<ResultAction>,
},
SearchEngine,
Plugin {
metadata: Arc<PluginMetadata>,
interaction_policy: RwLock<PanelInteraction>,
},
ScoreBooster,
KeywordOptimizer {
info: RwLock<KeywordOptimizerInfo>,
},
KeywordInjector,
}
pub struct RemoteComponent {
pub core: ComponentCore,
pub client: Arc<JsonRpcClient>,
cached_settings: RwLock<serde_json::Value>,
cached_schema: RwLock<Vec<SettingDefinition>>,
cached_actions: RwLock<Vec<ConfigActionDef>>,
cached_default_enabled: RwLock<bool>,
pub kind: RemoteComponentKind,
}
fn to_config_error(e: ProtocolError) -> ConfigError {
ConfigError::ApplyFailed(e.to_string())
}
impl std::fmt::Debug for RemoteComponent {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("RemoteComponent")
.field("core", &self.core)
.field("cached_settings", &self.cached_settings.read())
.field("cached_schema", &self.cached_schema.read())
.field("cached_actions", &self.cached_actions.read())
.field("kind", &self.kind)
.finish()
}
}
impl RemoteComponent {
#[allow(clippy::too_many_arguments)]
pub fn new(
component_id: String,
component_name: String,
component_description: String,
component_type: ComponentType,
priority: u32,
client: Arc<JsonRpcClient>,
schema: Vec<SettingDefinition>,
settings: serde_json::Value,
actions: Vec<ConfigActionDef>,
default_enabled: bool,
kind: RemoteComponentKind,
) -> Self {
Self {
core: ComponentCore::new(
component_id,
component_name,
component_description,
component_type,
priority,
),
client,
cached_settings: RwLock::new(settings),
cached_schema: RwLock::new(schema),
cached_actions: RwLock::new(actions),
cached_default_enabled: RwLock::new(default_enabled),
kind,
}
}
pub fn as_data_source(self: Arc<Self>) -> Option<Arc<dyn DataSource>> {
matches!(self.kind, RemoteComponentKind::DataSource).then(|| self as Arc<dyn DataSource>)
}
pub fn as_action_executor(self: Arc<Self>) -> Option<Arc<dyn ActionExecutor>> {
matches!(self.kind, RemoteComponentKind::ActionExecutor { .. })
.then(|| self as Arc<dyn ActionExecutor>)
}
pub fn as_plugin(self: Arc<Self>) -> Option<Arc<dyn Plugin>> {
matches!(self.kind, RemoteComponentKind::Plugin { .. }).then(|| self as Arc<dyn Plugin>)
}
pub fn as_search_engine(self: Arc<Self>) -> Option<Arc<dyn SearchEngine>> {
matches!(self.kind, RemoteComponentKind::SearchEngine)
.then(|| self as Arc<dyn SearchEngine>)
}
pub fn as_score_booster(self: Arc<Self>) -> Option<Arc<dyn ScoreBooster>> {
matches!(self.kind, RemoteComponentKind::ScoreBooster)
.then(|| self as Arc<dyn ScoreBooster>)
}
pub fn as_keyword_optimizer(self: Arc<Self>) -> Option<Arc<dyn KeywordOptimizer>> {
matches!(self.kind, RemoteComponentKind::KeywordOptimizer { .. })
.then(|| self as Arc<dyn KeywordOptimizer>)
}
pub fn as_keyword_injector(self: Arc<Self>) -> Option<Arc<dyn KeywordInjector>> {
matches!(self.kind, RemoteComponentKind::KeywordInjector)
.then(|| self as Arc<dyn KeywordInjector>)
}
pub fn is_data_source(&self) -> bool {
matches!(self.kind, RemoteComponentKind::DataSource)
}
pub fn is_action_executor(&self) -> bool {
matches!(self.kind, RemoteComponentKind::ActionExecutor { .. })
}
pub fn is_plugin(&self) -> bool {
matches!(self.kind, RemoteComponentKind::Plugin { .. })
}
pub fn is_search_engine(&self) -> bool {
matches!(self.kind, RemoteComponentKind::SearchEngine)
}
pub fn is_score_booster(&self) -> bool {
matches!(self.kind, RemoteComponentKind::ScoreBooster)
}
pub fn is_keyword_optimizer(&self) -> bool {
matches!(self.kind, RemoteComponentKind::KeywordOptimizer { .. })
}
pub fn is_keyword_injector(&self) -> bool {
matches!(self.kind, RemoteComponentKind::KeywordInjector)
}
}
#[async_trait]
impl Configurable for RemoteComponent {
fn core(&self) -> &ComponentCore {
&self.core
}
fn setting_schema(&self) -> Vec<SettingDefinition> {
self.cached_schema.read().clone()
}
fn get_settings(&self) -> serde_json::Value {
self.cached_settings.read().clone()
}
fn config_actions(&self) -> Vec<ConfigActionDef> {
self.cached_actions.read().clone()
}
fn default_enabled(&self) -> bool {
*self.cached_default_enabled.read()
}
async fn apply_settings(&self, settings: serde_json::Value) -> Result<(), ConfigError> {
let client = self.client.clone();
let component_id = self.core.component_id().to_string();
let settings_clone = settings.clone();
client
.call::<_, serde_json::Value>(
plugin_methods::APPLY_SETTINGS,
ApplySettingsParams {
component_id,
settings: settings_clone,
},
Duration::from_secs(5),
)
.await
.map_err(to_config_error)?;
*self.cached_settings.write() = settings;
self.refresh_interaction_policy().await;
if self.is_keyword_optimizer() {
self.refresh_keyword_optimizer_info().await;
}
Ok(())
}
async fn validate_settings(&self, settings: &serde_json::Value) -> Result<(), ConfigError> {
let contribution = self.settings_contribution()?;
contribution
.validate_values(settings)
.map_err(ConfigError::ValidationFailed)?;
let client = self.client.clone();
let component_id = self.core.component_id().to_string();
let settings_clone = settings.clone();
let result: ValidateSettingsResult = client
.call(
plugin_methods::VALIDATE_SETTINGS,
ValidateSettingsParams {
component_id,
settings: settings_clone,
},
Duration::from_secs(5),
)
.await
.map_err(to_config_error)?;
if let Some(error) = result.error {
return Err(ConfigError::ValidationFailed(error));
}
Ok(())
}
async fn execute_config_action(
&self,
action: &str,
params: &serde_json::Value,
) -> Result<serde_json::Value, String> {
let client = self.client.clone();
let component_id = self.core.component_id().to_string();
let action = action.to_string();
let params = params.clone();
let result: Result<serde_json::Value, ProtocolError> = client
.call::<_, serde_json::Value>(
plugin_methods::EXECUTE_CONFIG_ACTION,
ExecuteConfigActionParams {
component_id,
action,
params,
},
Duration::from_secs(10),
)
.await;
result.map_err(|e| e.to_string())
}
}
#[async_trait]
impl DataSource for RemoteComponent {
async fn fetch_candidates(&self) -> CachedCandidateData {
assert!(
matches!(self.kind, RemoteComponentKind::DataSource),
"RemoteComponent {} is not a DataSource but fetch_candidates() was called",
self.core.component_id()
);
let result: Result<FetchCandidatesResult, _> = self
.client
.call(
plugin_methods::FETCH_CANDIDATES,
FetchCandidatesParams {
component_id: self.core.component_id().to_string(),
},
Duration::from_secs(30),
)
.await;
match result {
Ok(data) => {
let mut cache = CachedCandidateData::new();
for candidate in data.candidates {
cache.add_candidate(candidate);
}
cache
}
Err(e) => {
tracing::warn!(
"DataSource {} fetch_candidates failed: {}",
self.core.component_id(),
e
);
CachedCandidateData::new()
}
}
}
}
#[async_trait]
impl ActionExecutor for RemoteComponent {
fn supported_target_types(&self) -> Vec<TargetType> {
match &self.kind {
RemoteComponentKind::ActionExecutor { target_types, .. } => target_types.clone(),
_ => panic!(
"RemoteComponent {} is not an ActionExecutor but supported_target_types() was called",
self.core.component_id()
),
}
}
fn supported_actions(&self) -> Vec<ResultAction> {
match &self.kind {
RemoteComponentKind::ActionExecutor { result_actions, .. } => result_actions.clone(),
_ => panic!(
"RemoteComponent {} is not an ActionExecutor but supported_actions() was called",
self.core.component_id()
),
}
}
async fn execute(&self, ctx: &ExecutionContext, action_id: &str) -> Result<(), ExecutionError> {
assert!(
matches!(self.kind, RemoteComponentKind::ActionExecutor { .. }),
"RemoteComponent {} is not an ActionExecutor but execute() was called",
self.core.component_id()
);
let result: Result<ExecutorExecuteResult, _> = self
.client
.call(
plugin_methods::EXECUTOR_EXECUTE,
ExecutorExecuteParams {
component_id: self.core.component_id().to_string(),
execution_ctx: ctx.clone(),
action_id: action_id.to_string(),
},
Duration::from_secs(30),
)
.await;
match result {
Ok(r) => {
if let Some(error) = r.error {
Err(ExecutionError::Failed(error))
} else {
Ok(())
}
}
Err(e) => Err(ExecutionError::Failed(e.to_string())),
}
}
}
#[async_trait]
impl Plugin for RemoteComponent {
fn metadata(&self) -> &PluginMetadata {
match &self.kind {
RemoteComponentKind::Plugin { metadata, .. } => metadata.as_ref(),
_ => panic!(
"RemoteComponent {} is not a Plugin but metadata() was called",
self.core.component_id()
),
}
}
async fn init(
&self,
ctx: &PluginContext,
_handle: Option<Arc<PluginHandle>>,
) -> Result<(), PluginError> {
assert!(
matches!(self.kind, RemoteComponentKind::Plugin { .. }),
"RemoteComponent {} is not a Plugin but init() was called",
self.core.component_id()
);
let metadata = match &self.kind {
RemoteComponentKind::Plugin { metadata, .. } => metadata,
_ => unreachable!("kind 已在上面断言为 Plugin"),
};
let result: Result<serde_json::Value, _> = self
.client
.call(
plugin_methods::INIT,
InitParams {
plugin_id: metadata.id.clone(),
ctx: ctx.clone(),
},
Duration::from_secs(10),
)
.await;
match result {
Ok(_) => Ok(()),
Err(zerolaunch_plugin_protocol::ProtocolError::Rpc { code, .. })
if code == codes::METHOD_NOT_FOUND =>
{
Ok(())
}
Err(e) => Err(PluginError::InitFailed(e.to_string())),
}
}
fn interaction_policy(&self) -> PanelInteraction {
match &self.kind {
RemoteComponentKind::Plugin {
interaction_policy, ..
} => interaction_policy.read().clone(),
_ => panic!(
"RemoteComponent {} is not a Plugin but interaction_policy() was called",
self.core.component_id()
),
}
}
async fn query(
&self,
ctx: &PluginContext,
query: &Query,
) -> Result<QueryResponse, PluginError> {
let metadata = match &self.kind {
RemoteComponentKind::Plugin { metadata, .. } => metadata,
_ => panic!(
"RemoteComponent {} is not a Plugin but query() was called",
self.core.component_id()
),
};
self.refresh_interaction_policy().await;
self.client
.call::<_, QueryResponse>(
plugin_methods::QUERY,
QueryParams {
plugin_id: metadata.id.clone(),
ctx: ctx.clone(),
query: query.clone(),
},
Duration::from_secs(30),
)
.await
.map_err(|e| PluginError::QueryFailed(e.to_string()))
}
async fn execute_action(
&self,
ctx: &PluginContext,
action_id: &str,
payload: serde_json::Value,
) -> Result<(), PluginError> {
let metadata = match &self.kind {
RemoteComponentKind::Plugin { metadata, .. } => metadata,
_ => panic!(
"RemoteComponent {} is not a Plugin but execute_action() was called",
self.core.component_id()
),
};
self.client
.call::<_, serde_json::Value>(
plugin_methods::EXECUTE_ACTION,
ExecuteActionParams {
plugin_id: metadata.id.clone(),
ctx: ctx.clone(),
action_id: action_id.to_string(),
payload,
},
Duration::from_secs(30),
)
.await
.map_err(|e| PluginError::ActionFailed(e.to_string()))?;
Ok(())
}
}
impl RemoteComponent {
async fn refresh_interaction_policy(&self) {
let RemoteComponentKind::Plugin {
interaction_policy, ..
} = &self.kind
else {
return;
};
let result: Result<PanelInteraction, _> = self
.client
.call(
plugin_methods::INTERACTION_POLICY,
InteractionPolicyParams {
component_id: self.core.component_id().to_string(),
},
Duration::from_secs(5),
)
.await;
if let Ok(policy) = result {
*interaction_policy.write() = policy;
}
}
async fn refresh_keyword_optimizer_info(&self) {
let RemoteComponentKind::KeywordOptimizer { info } = &self.kind else {
return;
};
let result: Result<KeywordOptimizerInfo, _> = self
.client
.call(
plugin_methods::KEYWORD_OPTIMIZER_INFO,
KeywordOptimizerInfoParams {
component_id: self.core.component_id().to_string(),
},
Duration::from_secs(5),
)
.await;
match result {
Ok(meta) => *info.write() = meta,
Err(e) => {
tracing::warn!(
"KeywordOptimizer {} 属性刷新失败: {}",
self.core.component_id(),
e
);
}
}
}
}
#[async_trait]
impl SearchEngine for RemoteComponent {
async fn calculate_scores(
&self,
candidates: &CachedCandidateData,
query: &str,
) -> Vec<ScoredCandidate> {
assert!(
matches!(self.kind, RemoteComponentKind::SearchEngine),
"RemoteComponent {} is not a SearchEngine but calculate_scores() was called",
self.core.component_id()
);
let result: Result<Vec<ScoredCandidate>, _> = self
.client
.call(
plugin_methods::CALCULATE_SCORES,
CalculateScoresParams {
component_id: self.core.component_id().to_string(),
candidates: candidates.to_data(),
query: query.to_string(),
},
Duration::from_secs(5),
)
.await;
match result {
Ok(scored) => scored,
Err(e) => {
tracing::error!(
"SearchEngine {} calculate_scores failed: {}",
self.core.component_id(),
e
);
Vec::new()
}
}
}
}
#[async_trait]
impl ScoreBooster for RemoteComponent {
async fn record(&self, candidate_id: CandidateId, data: &CachedCandidateData, query: &str) {
assert!(
matches!(self.kind, RemoteComponentKind::ScoreBooster),
"RemoteComponent {} is not a ScoreBooster but record() was called",
self.core.component_id()
);
let result: Result<serde_json::Value, _> = self
.client
.call(
plugin_methods::BOOSTER_RECORD,
BoosterRecordParams {
component_id: self.core.component_id().to_string(),
candidate_id,
candidates: data.to_data(),
query: query.to_string(),
},
Duration::from_secs(2),
)
.await;
if let Err(e) = result {
tracing::warn!(
"ScoreBooster {} record failed: {}",
self.core.component_id(),
e
);
}
}
async fn boost(
&self,
candidates: &mut Vec<ScoredCandidate>,
data: &CachedCandidateData,
query: &str,
) {
assert!(
matches!(self.kind, RemoteComponentKind::ScoreBooster),
"RemoteComponent {} is not a ScoreBooster but boost() was called",
self.core.component_id()
);
let result: Result<Vec<ScoredCandidate>, _> = self
.client
.call(
plugin_methods::BOOSTER_BOOST,
BoosterBoostParams {
component_id: self.core.component_id().to_string(),
candidates: data.to_data(),
scored: candidates.clone(),
query: query.to_string(),
},
Duration::from_secs(5),
)
.await;
match result {
Ok(scored) => *candidates = scored,
Err(e) => {
tracing::warn!(
"ScoreBooster {} boost failed: {}",
self.core.component_id(),
e
);
}
}
}
}
#[async_trait]
impl KeywordOptimizer for RemoteComponent {
async fn optimize(&self, keyword: &str) -> Vec<String> {
assert!(
matches!(self.kind, RemoteComponentKind::KeywordOptimizer { .. }),
"RemoteComponent {} is not a KeywordOptimizer but optimize() was called",
self.core.component_id()
);
let result: Result<Vec<String>, _> = self
.client
.call(
plugin_methods::KEYWORD_OPTIMIZE,
KeywordOptimizeParams {
component_id: self.core.component_id().to_string(),
keyword: keyword.to_string(),
},
Duration::from_secs(10),
)
.await;
match result {
Ok(keywords) => keywords,
Err(e) => {
tracing::warn!(
"KeywordOptimizer {} optimize failed: {}",
self.core.component_id(),
e
);
Vec::new()
}
}
}
fn uses_context(&self) -> bool {
match &self.kind {
RemoteComponentKind::KeywordOptimizer { info } => info.read().uses_context,
_ => panic!(
"RemoteComponent {} is not a KeywordOptimizer but uses_context() was called",
self.core.component_id()
),
}
}
fn get_priority(&self) -> u32 {
match &self.kind {
RemoteComponentKind::KeywordOptimizer { info } => info.read().priority,
_ => panic!(
"RemoteComponent {} is not a KeywordOptimizer but get_priority() was called",
self.core.component_id()
),
}
}
}
#[async_trait]
impl KeywordInjector for RemoteComponent {
async fn inject_keywords(&self, candidate: &SearchCandidate) -> Vec<String> {
assert!(
matches!(self.kind, RemoteComponentKind::KeywordInjector),
"RemoteComponent {} is not a KeywordInjector but inject_keywords() was called",
self.core.component_id()
);
let result: Result<Vec<String>, _> = self
.client
.call(
plugin_methods::KEYWORD_INJECT,
KeywordInjectParams {
component_id: self.core.component_id().to_string(),
candidate: candidate.clone(),
},
Duration::from_secs(10),
)
.await;
match result {
Ok(keywords) => keywords,
Err(e) => {
tracing::warn!(
"KeywordInjector {} inject_keywords failed: {}",
self.core.component_id(),
e
);
Vec::new()
}
}
}
}