flare_plugin_host/
lib.rs

1//! Flare 能力插件的**宿主接入库**：把插件向 capability 声明自己这件事做成三行。
2//!
3//! # 它解决什么
4//!
5//! 插件接入 capability 有两条路：
6//!
7//! 1. **被发现**：插件注册到 Consul/etcd，capability 按配置里的
8//!    `plugin_discovery_endpoints` 找到它。这条路走得通，但发现来的端点
9//!    **没有清单信息** —— capability 只能把它标成 `unverified`，
10//!    于是「声明即边界」对它无从强制。
11//! 2. **主动声明**（本库）：插件启动后调用 `RegisterPluginEndpoint`，
12//!    带上版本、清单摘要与自己声明能处理的 operation 全集。
13//!    这样它才是 verified 的，声明之外的调用会被 capability 拒绝。
14//!
15//! 两条路不冲突：服务目录负责「进程在哪」，声明负责「它能做什么」。
16//! 本库只做后者 —— 前者各插件已经在用 `flare_server_core::discovery`，
17//! 没必要再包一层。
18//!
19//! # 为什么要有这个库，而不是每个插件自己调
20//!
21//! 「自己调」的问题不在于代码量，在于**容易少调**：漏了 declared_operations
22//! 就退化成 unverified，而这不会报错、只会让边界强制失效。把它做成一个
23//! 必须填满字段的结构体，漏填在编译期就暴露。
24//!
25//! ```no_run
26//! use flare_plugin_host::{PluginDeclaration, PluginHost};
27//!
28//! # async fn demo() -> Result<(), flare_plugin_host::HostError> {
29//! let declaration = PluginDeclaration {
30//!     tenant_id: "0".into(),
31//!     plugin_id: "flare-moments".into(),
32//!     capability_id: "social.moments.feed".into(),
33//!     grpc_authority: "127.0.0.1:50204".into(),
34//!     plugin_version: "1.0.0".into(),
35//!     api_version: "1".into(),
36//!     manifest_sha256: "….".into(),
37//!     declared_operations: vec!["social.moments.feed".into()],
38//!     labels: Default::default(),
39//!     seat_model: flare_plugin_host::SeatModel::Tenant,
40//! };
41//! PluginHost::connect("http://127.0.0.1:50051")
42//!     .await?
43//!     .announce(&declaration)
44//!     .await?;
45//! # Ok(())
46//! # }
47//! ```
48
49use std::collections::HashMap;
50use std::time::Duration;
51
52use flare_grpc_proto::capability::capability_service_client::CapabilityServiceClient;
53use flare_grpc_proto::capability::{
54    DeregisterPluginEndpointRequest, RegisterPluginEndpointRequest,
55};
56use tonic::transport::Channel;
57
58#[derive(Debug, thiserror::Error)]
59pub enum HostError {
60    #[error("连接 capability 失败：{0}")]
61    Connect(String),
62    #[error("capability 拒绝了注册：{0}")]
63    Rejected(String),
64    #[error("gRPC 调用失败：{0}")]
65    Rpc(String),
66    #[error("声明不完整：{0}")]
67    Invalid(&'static str),
68}
69
70/// 插件对 capability 的完整声明。
71///
72/// 字段全部必填是刻意的：这些正是注册契约 v2 里「可选、缺失即降级为
73/// unverified」的那些。协议层必须可选（否则核心升级会打死所有旧插件），
74/// 但**新写的插件没有理由不填** —— 在这里做成必填，漏填就编译不过。
75#[derive(Debug, Clone)]
76pub struct PluginDeclaration {
77    pub tenant_id: String,
78    pub plugin_id: String,
79    /// 本次注册的能力 id。必须出现在 `declared_operations` 里，
80    /// 否则 capability 会当场拒绝（清单与注册对不上）。
81    ///
82    /// 它**不限制**插件承接的范围 —— 范围由 `declared_operations` 决定。
83    /// 这里填一个最好认的入口即可，它主要出现在路由簿与日志里。
84    pub capability_id: String,
85    /// 本插件的 gRPC 地址，capability 按它回调。
86    pub grpc_authority: String,
87    pub plugin_version: String,
88    pub api_version: String,
89    /// 插件清单（plugin.json）的 sha256，用于确认部署物与目录一致。
90    pub manifest_sha256: String,
91    /// 本插件能处理的 operation 全集。**留空即退化为 unverified**。
92    pub declared_operations: Vec<String>,
93    /// 附加标签，例如 `health_protocol` 声明特殊探活协议。
94    pub labels: HashMap<String, String>,
95    /// 计费/授权单位：`SeatModel::Tenant`（装了全员可用）或
96    /// `SeatModel::PerUser`（还需逐人授权）。
97    ///
98    /// 这是**产品决策**，平台不替插件决定它怎么卖，所以必填。
99    pub seat_model: SeatModel,
100}
101
102/// 计费/授权单位。
103///
104/// - `Tenant`：租户装了就全员可用。绝大多数插件属于这一类 ——
105///   组织装了大家就用，不该再逐人发放。
106/// - `PerUser`：还需逐人授权。留给两类：有边际成本的（AI 按 token 烧钱，
107///   按席位卖才不亏）与需合规隔离的（DLP 导出只给特定角色）。
108#[derive(Debug, Clone, Copy, PartialEq, Eq)]
109pub enum SeatModel {
110    Tenant,
111    PerUser,
112}
113
114impl SeatModel {
115    fn as_wire(self) -> &'static str {
116        match self {
117            Self::Tenant => "tenant",
118            Self::PerUser => "per_user",
119        }
120    }
121}
122
123impl PluginDeclaration {
124    /// 本地自检：把「注册后才发现对不上」提前到调用前。
125    ///
126    /// capability 侧也会校验（那是权威判定），这里重复一遍是为了让插件作者
127    /// 在自己的进程里就看到错误，而不是去翻 capability 的日志。
128    pub fn validate(&self) -> Result<(), HostError> {
129        if self.plugin_id.trim().is_empty() {
130            return Err(HostError::Invalid("plugin_id 不能为空"));
131        }
132        if self.capability_id.trim().is_empty() {
133            return Err(HostError::Invalid("capability_id 不能为空"));
134        }
135        if self.grpc_authority.trim().is_empty() {
136            return Err(HostError::Invalid("grpc_authority 不能为空"));
137        }
138        if self.declared_operations.is_empty() {
139            return Err(HostError::Invalid(
140                "declared_operations 为空会让插件退化为 unverified，声明边界无法强制",
141            ));
142        }
143        if !self.declared_operations.contains(&self.capability_id) {
144            return Err(HostError::Invalid(
145                "capability_id 必须出现在 declared_operations 里",
146            ));
147        }
148        Ok(())
149    }
150}
151
152/// 与 capability 控制面的连接。
153pub struct PluginHost {
154    client: CapabilityServiceClient<Channel>,
155}
156
157impl PluginHost {
158    /// 连接 capability。`endpoint` 形如 `http://host:port`。
159    pub async fn connect(endpoint: impl Into<String>) -> Result<Self, HostError> {
160        let endpoint = endpoint.into();
161        let channel = Channel::from_shared(endpoint.clone())
162            .map_err(|e| HostError::Connect(format!("{endpoint}: {e}")))?
163            .connect_timeout(Duration::from_secs(5))
164            .connect()
165            .await
166            .map_err(|e| HostError::Connect(format!("{endpoint}: {e}")))?;
167        Ok(Self {
168            client: CapabilityServiceClient::new(channel),
169        })
170    }
171
172    /// 向 capability 声明本插件。
173    ///
174    /// 成功后该实例是 **verified** 的：声明之外的调用会被 capability 拒绝。
175    pub async fn announce(&mut self, declaration: &PluginDeclaration) -> Result<(), HostError> {
176        declaration.validate()?;
177
178        let response = self
179            .client
180            .register_plugin_endpoint(RegisterPluginEndpointRequest {
181                tenant_id: declaration.tenant_id.clone(),
182                plugin_id: declaration.plugin_id.clone(),
183                capability_id: declaration.capability_id.clone(),
184                grpc_authority: declaration.grpc_authority.clone(),
185                labels: declaration.labels.clone(),
186                request_id: String::new(),
187                plugin_version: declaration.plugin_version.clone(),
188                api_version: declaration.api_version.clone(),
189                manifest_sha256: declaration.manifest_sha256.clone(),
190                declared_operations: declaration.declared_operations.clone(),
191                seat_model: declaration.seat_model.as_wire().to_string(),
192            })
193            .await
194            .map_err(|e| HostError::Rpc(e.to_string()))?
195            .into_inner();
196
197        if !response.accepted {
198            return Err(HostError::Rejected(response.message));
199        }
200        tracing::info!(
201            plugin_id = %declaration.plugin_id,
202            capability_id = %declaration.capability_id,
203            operations = declaration.declared_operations.len(),
204            "plugin announced to capability"
205        );
206        Ok(())
207    }
208
209    /// 优雅摘除：停机前告诉 capability 别再往这里派活。
210    ///
211    /// 不调用也不会坏 —— capability 的健康检查最终会把它摘掉。但那要等一个
212    /// 检查周期，期间的调用会打到正在退出的进程上。
213    /// 注销粒度是 `(tenant, plugin)`，与注册粒度一致 —— 一个插件进程在路由簿里
214    /// 只有一条记录，承接范围由 `declared_operations` 表达，所以一次注销即摘干净。
215    pub async fn withdraw(&mut self, tenant_id: &str, plugin_id: &str) -> Result<(), HostError> {
216        self.client
217            .deregister_plugin_endpoint(DeregisterPluginEndpointRequest {
218                tenant_id: tenant_id.to_string(),
219                plugin_id: plugin_id.to_string(),
220                request_id: String::new(),
221            })
222            .await
223            .map_err(|e| HostError::Rpc(e.to_string()))?;
224        tracing::info!(plugin_id, "plugin withdrawn from capability");
225        Ok(())
226    }
227}
228
229#[cfg(test)]
230mod tests {
231    use super::*;
232
233    fn declaration() -> PluginDeclaration {
234        PluginDeclaration {
235            tenant_id: "0".into(),
236            plugin_id: "p1".into(),
237            capability_id: "vendorx.do".into(),
238            grpc_authority: "127.0.0.1:1".into(),
239            plugin_version: "1.0.0".into(),
240            api_version: "1".into(),
241            manifest_sha256: "abc".into(),
242            declared_operations: vec!["vendorx.do".into()],
243            labels: HashMap::new(),
244            seat_model: SeatModel::Tenant,
245        }
246    }
247
248    #[test]
249    fn complete_declaration_is_valid() {
250        declaration().validate().expect("完整声明应当通过");
251    }
252
253    /// 空声明会让插件静默退化成 unverified —— 本库的存在意义就是不让它发生。
254    #[test]
255    fn empty_declared_operations_is_rejected_locally() {
256        let mut d = declaration();
257        d.declared_operations.clear();
258        assert!(matches!(d.validate(), Err(HostError::Invalid(_))));
259    }
260
261    /// 注册的能力不在自己的声明里 —— capability 会拒，这里提前拦下。
262    #[test]
263    fn capability_id_must_be_declared() {
264        let mut d = declaration();
265        d.declared_operations = vec!["vendorx.other".into()];
266        assert!(matches!(d.validate(), Err(HostError::Invalid(_))));
267    }
268}
flare_plugin_host/lib.rs

flare_plugin_host/
lib.rs
