flare-plugin-host 0.1.1

//! 一个能跑起来的最小插件：两个 operation，一次声明，一个 gRPC 服务。
//!
//! 它做的事没有业务价值 —— 价值在于**它是完整的**：
//! 声明、注册、上下文还原、JSON 编解码、未知 operation 的处理、优雅停机，
//! 一个真实插件需要的每一环这里都有，而且都短到能一眼看完。

use std::collections::HashMap;
use std::net::SocketAddr;

use flare_grpc_proto::capability::extension_plugin_server::{
    ExtensionPlugin, ExtensionPluginServer,
};
use flare_grpc_proto::capability::{GenericRequest, GenericResponse};
use flare_plugin_host::{PluginDeclaration, PluginHost, SeatModel};
use tonic::{Request, Response, Status, transport::Server};

const PLUGIN_ID: &str = "echo-plugin";

/// 声明清单。**它就是边界** —— 不在这里的 operation，平台会在到达本进程前拒掉。
const OPERATIONS: &[&str] = &["example.echo.say", "example.echo.upper"];

#[derive(Default)]
struct EchoPlugin;

#[tonic::async_trait]
impl ExtensionPlugin for EchoPlugin {
    async fn call(
        &self,
        request: Request<GenericRequest>,
    ) -> Result<Response<GenericResponse>, Status> {
        let req = request.into_inner();

        // 平台是**进程外**调用你，框架的上下文中间件不会替你跑。
        // 漏掉这一步的表现是所有 op 都报「找不到上下文」，看着像鉴权问题。
        let tenant = req.metadata.get("tenant_id").cloned().unwrap_or_default();
        let user = req.metadata.get("user_id").cloned().unwrap_or_default();

        // Any.value 直接就是 JSON 字节，不是 protobuf 编码。
        let payload: serde_json::Value = match req.payload.as_ref() {
            Some(any) => serde_json::from_slice(&any.value)
                .map_err(|e| Status::invalid_argument(format!("payload 不是合法 JSON: {e}")))?,
            None => serde_json::Value::Null,
        };
        let text = payload.get("text").and_then(|v| v.as_str()).unwrap_or("");

        let result = match req.operation.as_str() {
            "example.echo.say" => serde_json::json!({
                "echo": text,
                "tenant_id": tenant,
                "user_id": user,
            }),
            "example.echo.upper" => serde_json::json!({ "echo": text.to_uppercase() }),

            // 走到这里说明声明清单与本 match 不同步。
            // 明确回错，别静默成功 —— 静默成功会让调用方以为业务做了。
            other => {
                return Ok(Response::new(GenericResponse {
                    ok: false,
                    payload: None,
                    error_code: "UNKNOWN_OPERATION".into(),
                    error_message: format!("{other} 已声明但未实现"),
                    request_id: req.request_id,
                }));
            }
        };

        Ok(Response::new(GenericResponse {
            ok: true,
            payload: Some(prost_types::Any {
                type_url: "type.googleapis.com/flare.capability.v1.PayloadJson".into(),
                value: serde_json::to_vec(&result).map_err(|e| Status::internal(e.to_string()))?,
            }),
            error_code: String::new(),
            error_message: String::new(),
            request_id: req.request_id,
        }))
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let listen: SocketAddr = std::env::var("ECHO_PLUGIN_LISTEN")
        .unwrap_or_else(|_| "127.0.0.1:7901".into())
        .parse()?;
    let capability =
        std::env::var("CAPABILITY_ENDPOINT").unwrap_or_else(|_| "http://127.0.0.1:50110".into());
    // 平台按这个地址回调你，所以它必须是**平台能连到**的地址，
    // 不是你的监听地址的字面量。容器/K8s 里两者常常不同。
    let advertise = std::env::var("ECHO_PLUGIN_ADVERTISE").unwrap_or_else(|_| listen.to_string());

    let declaration = PluginDeclaration {
        tenant_id: "0".into(),
        plugin_id: PLUGIN_ID.into(),
        // 注册键，只是路由簿里的标签；承接范围由 declared_operations 决定。
        capability_id: OPERATIONS[0].into(),
        grpc_authority: advertise.clone(),
        plugin_version: env!("CARGO_PKG_VERSION").into(),
        api_version: "1".into(),
        manifest_sha256: String::new(),
        declared_operations: OPERATIONS.iter().map(|s| s.to_string()).collect(),
        labels: HashMap::new(),
        // 装了就全员可用。按人授权只留给有边际成本或需合规隔离的能力。
        seat_model: SeatModel::Tenant,
    };

    // 先起服务再声明：反过来的话，平台可能在你还没监听时就来探活。
    let server = tokio::spawn(
        Server::builder()
            .add_service(ExtensionPluginServer::new(EchoPlugin))
            .serve_with_shutdown(listen, async {
                tokio::signal::ctrl_c().await.ok();
            }),
    );
    println!("echo-plugin 监听 {listen}，对外通告 {advertise}");

    match PluginHost::connect(&capability).await {
        Ok(mut host) => {
            host.announce(&declaration).await?;
            println!("已向 {capability} 声明 {} 个 operation", OPERATIONS.len());

            server.await??;

            // 优雅摘除。不调也不会坏——健康检查最终会把你摘掉，
            // 但那要等一个检查周期，期间的调用会打到正在退出的进程上。
            host.withdraw("0", PLUGIN_ID).await.ok();
            println!("已注销");
        }
        Err(e) => {
            // 连不上平台**不该**让插件退出：平台可能比你晚起。
            // 真实插件应当在这里重试；示例里只提示，服务照常提供。
            println!("连不上 {capability}（{e}）；服务继续，未注册");
            server.await??;
        }
    }
    Ok(())
}

/// 声明清单必须与上面 `call` 里的分发臂一致，两个方向都会出事：
///   声明有、实现没有 → 请求打过来，你回 UNKNOWN_OPERATION，看着像插件坏了
///   实现有、声明没有 → 平台直接拒绝，你的进程收不到，看着像权限问题
///
/// 两种症状的排查方向完全不同，所以用一条测试把它们钉在一起。
#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn declaration_matches_implementation() {
        let implemented = include_str!("echo_plugin.rs");
        for op in OPERATIONS {
            assert!(
                implemented.contains(&format!("\"{op}\" =>")),
                "{op} 声明了却没有对应的分发臂"
            );
        }
    }
}