use std::collections::BTreeSet;
use std::path::{Path, PathBuf};
use anyhow::{Context, Result};
use async_trait::async_trait;
use axum::Router;
use folk_api::{PluginContext, RpcMethodDef, ServerPlugin};
use tonic::transport::Server;
use tracing::{debug, info, warn};
use crate::config::GrpcConfig;
use crate::service::{GrpcState, grpc_handler};
pub struct GrpcPlugin {
config: GrpcConfig,
}
impl GrpcPlugin {
pub fn new(config: GrpcConfig) -> Self {
Self { config }
}
}
#[async_trait]
impl ServerPlugin for GrpcPlugin {
fn name(&self) -> &'static str {
"grpc"
}
async fn run(&self, ctx: PluginContext) -> Result<()> {
let state = GrpcState {
executor: ctx.executor.clone(),
};
let router: axum::Router = Router::new().fallback(grpc_handler).with_state(state);
let mut sd = ctx.shutdown.clone();
info!(listen = %self.config.listen, "gRPC server listening");
let routes = tonic::service::Routes::from(router);
if !self.config.proto.is_empty() {
match build_reflection_descriptors(&self.config.proto) {
Ok(encoded_fds) => {
info!(proto_files = self.config.proto.len(), "gRPC reflection enabled");
let reflection = tonic_reflection::server::Builder::configure()
.register_encoded_file_descriptor_set(&encoded_fds)
.build_v1()
.context("build reflection service")?;
Server::builder()
.add_routes(routes)
.add_service(reflection)
.serve_with_shutdown(self.config.listen, async move {
sd.changed().await.ok();
})
.await?;
return Ok(());
},
Err(e) => {
warn!(error = %e, "failed to build reflection; starting without it");
},
}
}
Server::builder()
.add_routes(routes)
.serve_with_shutdown(self.config.listen, async move {
sd.changed().await.ok();
})
.await?;
Ok(())
}
fn rpc_methods(&self) -> Vec<RpcMethodDef> {
vec![RpcMethodDef::new(
"grpc.services",
"list registered gRPC service names",
)]
}
}
fn build_reflection_descriptors(proto_files: &[String]) -> Result<Vec<u8>> {
let cwd = std::env::current_dir().context("get cwd")?;
let mut include_paths = BTreeSet::new();
let mut visited = BTreeSet::new();
for proto in proto_files {
let abs = cwd.join(proto);
resolve_imports(&abs, &cwd, &mut include_paths, &mut visited);
}
for proto in proto_files {
let abs = cwd.join(proto);
if let Some(parent) = abs.parent() {
include_paths.insert(parent.to_path_buf());
}
}
debug!(
paths = include_paths.len(),
"resolved proto include paths"
);
let paths: Vec<_> = include_paths.iter().collect();
let mut compiler = protox::Compiler::new(paths.iter().map(|p| p.as_path()))
.context("create protox compiler")?;
compiler.include_imports(true);
for proto in proto_files {
let path = Path::new(proto);
let file_name = path
.file_name()
.and_then(|n| n.to_str())
.context("invalid proto file name")?;
compiler
.open_file(file_name)
.with_context(|| format!("compile {proto}"))?;
}
let fds = compiler.file_descriptor_set();
Ok(prost::Message::encode_to_vec(&fds))
}
fn resolve_imports(
proto_path: &Path,
project_root: &Path,
include_paths: &mut BTreeSet<PathBuf>,
visited: &mut BTreeSet<PathBuf>,
) {
if !proto_path.is_file() || !visited.insert(proto_path.to_path_buf()) {
return;
}
let content = match std::fs::read_to_string(proto_path) {
Ok(c) => c,
Err(_) => return,
};
for line in content.lines() {
let line = line.trim();
if !line.starts_with("import ") {
continue;
}
let Some(start) = line.find('"') else { continue };
let Some(end) = line[start + 1..].find('"') else { continue };
let import_path = &line[start + 1..start + 1 + end];
if import_path.starts_with("google/protobuf/") {
continue;
}
let found_files = find_all_in_tree(project_root, import_path);
for found in &found_files {
if let Some(base) = found
.to_str()
.and_then(|f| f.strip_suffix(import_path))
.map(PathBuf::from)
{
if !base.as_os_str().is_empty() {
include_paths.insert(base);
}
} else if let Some(parent) = found.parent() {
include_paths.insert(parent.to_path_buf());
}
resolve_imports(found, project_root, include_paths, visited);
}
}
}
fn find_all_in_tree(root: &Path, relative: &str) -> Vec<PathBuf> {
let mut results = Vec::new();
let direct = root.join(relative);
if direct.is_file() {
results.push(direct);
}
find_all_recursive(root, relative, &mut results);
results
}
fn find_all_recursive(dir: &Path, relative: &str, results: &mut Vec<PathBuf>) {
let candidate = dir.join(relative);
if candidate.is_file() && !results.contains(&candidate) {
results.push(candidate);
}
let entries = match std::fs::read_dir(dir) {
Ok(e) => e,
Err(_) => return,
};
for entry in entries.flatten() {
let path = entry.path();
if path.is_dir() {
let name = entry.file_name();
let name = name.to_str().unwrap_or("");
if name.starts_with('.')
|| name == "node_modules"
|| name == "target"
|| name == "pb"
|| name == "build"
|| name == "dist"
|| name == "out"
{
continue;
}
find_all_recursive(&path, relative, results);
}
}
}