use std::collections::BTreeMap;
use std::path::Path;
use std::sync::Arc;
use abi_stable::library::{lib_header_from_raw_library, RawLibrary};
use abi_stable::sabi_trait::TD_Opaque;
use abi_stable::std_types::{RResult, RStr, RString};
use apiplant_abi::{
BoxedFunction, FunctionManifest, FunctionMod_Ref, HostApi, HostApi_TO, LogLevel,
};
use apiplant_ai::Ai;
use apiplant_cache::Cache;
use apiplant_core::RateLimitRule;
use apiplant_db::Db;
use apiplant_email::Mailer;
use apiplant_payments::Payments;
use apiplant_queue::Queue;
pub type BuiltinHandler = fn(&HostBridge, &str) -> Result<String, String>;
enum Body {
Dynamic(BoxedFunction),
Builtin(BuiltinHandler),
}
const JS_EXTENSION: &str = "js";
#[cfg(feature = "typescript")]
fn load_javascript(path: &Path) -> Result<abi_stable::std_types::RVec<BoxedFunction>, String> {
debug_assert_eq!(JS_EXTENSION, apiplant_js::EXTENSION);
Ok(apiplant_js::load(path)?.into())
}
#[cfg(not(feature = "typescript"))]
fn load_javascript(path: &Path) -> Result<abi_stable::std_types::RVec<BoxedFunction>, String> {
Err(format!(
"{} is a TypeScript function, and this build of apiplant was made without \
TypeScript support — use a full build, or drop the function",
path.display()
))
}
pub struct LoadedFunction {
pub manifest: FunctionManifest,
pub config_json: String,
pub rate_limit: RateLimitRule,
body: Body,
}
fn declared_rate_limit(name: &str, config_json: &str) -> RateLimitRule {
let declared = serde_json::from_str::<serde_json::Value>(config_json)
.ok()
.and_then(|config| config.get("rate_limit")?.as_str().map(str::to_string));
let Some(declared) = declared else {
return RateLimitRule::Inherit;
};
RateLimitRule::parse(&declared).unwrap_or_else(|| {
tracing::warn!(
function = %name,
value = %declared,
"`rate_limit` is not a rate limit (`100/1m`, `off`, `inherit`); ignoring it"
);
RateLimitRule::Inherit
})
}
impl LoadedFunction {
pub fn new(func: BoxedFunction, config_json: String) -> Self {
let manifest = func.manifest();
LoadedFunction {
rate_limit: declared_rate_limit(manifest.name.as_str(), &config_json),
manifest,
config_json,
body: Body::Dynamic(func),
}
}
pub fn builtin(
manifest: FunctionManifest,
handler: BuiltinHandler,
config_json: String,
) -> Self {
LoadedFunction {
rate_limit: declared_rate_limit(manifest.name.as_str(), &config_json),
manifest,
config_json,
body: Body::Builtin(handler),
}
}
pub fn invoke(&self, bridge: HostBridge, input: &str) -> Result<String, String> {
match &self.body {
Body::Builtin(handler) => handler(&bridge, input),
Body::Dynamic(func) => {
let host = HostApi_TO::from_value(bridge, TD_Opaque);
match func.invoke(host, RStr::from_str(input)) {
RResult::ROk(s) => Ok(s.into_string()),
RResult::RErr(e) => Err(e.into_string()),
}
}
}
}
}
#[derive(Default)]
pub struct FunctionRegistry {
functions: BTreeMap<String, LoadedFunction>,
}
impl FunctionRegistry {
pub fn load(app: &apiplant_core::App) -> Self {
let mut registry = FunctionRegistry::default();
crate::builtins::register_all(&mut registry, app);
for (name, f) in Self::load_dir(&app.functions_dir).functions {
if registry.functions.contains_key(&name) {
tracing::warn!(function = %name, "app function replaces the built-in of the same name");
}
registry.functions.insert(name, f);
}
registry
}
pub fn register_builtin(
&mut self,
manifest: FunctionManifest,
handler: BuiltinHandler,
config_json: String,
) {
let loaded = LoadedFunction::builtin(manifest, handler, config_json);
self.functions
.insert(loaded.manifest.name.to_string(), loaded);
}
pub fn load_dir(dir: &Path) -> Self {
let mut registry = FunctionRegistry::default();
let entries = match std::fs::read_dir(dir) {
Ok(e) => e,
Err(_) => {
tracing::info!(dir = %dir.display(), "no functions/ directory");
return registry;
}
};
for entry in entries.flatten() {
let path = entry.path();
let loadable = matches!(
path.extension().and_then(|e| e.to_str()),
Some("so") | Some("dylib") | Some("dll") | Some(JS_EXTENSION)
);
if !loadable {
continue;
}
match Self::load_library(&path) {
Ok(loaded) => {
for f in loaded {
tracing::info!(
function = %f.manifest.name,
version = %f.manifest.version,
library = %path.display(),
"loaded function"
);
registry.functions.insert(f.manifest.name.to_string(), f);
}
}
Err(e) => {
tracing::error!(path = %path.display(), error = %e, "failed to load function")
}
}
}
registry
}
fn load_library(path: &Path) -> Result<Vec<LoadedFunction>, String> {
let exported = if path.extension().and_then(|e| e.to_str()) == Some(JS_EXTENSION) {
load_javascript(path)?
} else {
Self::load_native(path)?
};
Self::wrap(path, exported)
}
fn load_native(path: &Path) -> Result<abi_stable::std_types::RVec<BoxedFunction>, String> {
let exported = match Self::open(path) {
Ok(module) => module.new_functions()(),
Err(rust_abi_error) => match crate::cabi::load(path)? {
Some(functions) => functions.into(),
None => return Err(rust_abi_error),
},
};
Ok(exported)
}
fn wrap(
path: &Path,
exported: abi_stable::std_types::RVec<BoxedFunction>,
) -> Result<Vec<LoadedFunction>, String> {
if exported.is_empty() {
return Err("library exports no functions".to_string());
}
let mut loaded: Vec<LoadedFunction> = Vec::with_capacity(exported.len());
for func in exported {
let manifest = func.manifest();
let name = manifest.name.to_string();
if loaded.iter().any(|f| f.manifest.name == manifest.name) {
return Err(format!("library exports two functions named `{name}`"));
}
let config_path = path.with_file_name(format!("{name}.toml"));
let config_json = std::fs::read_to_string(&config_path)
.ok()
.and_then(|t| toml::from_str::<toml::Value>(&t).ok())
.map(|mut v| {
apiplant_core::expand_document(&mut v, &format!("{name}.toml"));
v
})
.and_then(|v| serde_json::to_string(&v).ok())
.unwrap_or_else(|| "{}".to_string());
loaded.push(LoadedFunction {
rate_limit: declared_rate_limit(&name, &config_json),
manifest,
config_json,
body: Body::Dynamic(func),
});
}
Ok(loaded)
}
fn open(path: &Path) -> Result<FunctionMod_Ref, String> {
let library = RawLibrary::load_at(path).map_err(|e| e.to_string())?;
let library: &'static RawLibrary = Box::leak(Box::new(library));
let header = unsafe { lib_header_from_raw_library(library).map_err(|e| e.to_string())? };
header
.init_root_module::<FunctionMod_Ref>()
.map_err(|e| e.to_string())
}
pub fn register(&mut self, func: BoxedFunction, config_json: String) {
let loaded = LoadedFunction::new(func, config_json);
self.functions
.insert(loaded.manifest.name.to_string(), loaded);
}
pub fn get(&self, name: &str) -> Option<&LoadedFunction> {
self.functions.get(name)
}
pub fn iter(&self) -> impl Iterator<Item = &LoadedFunction> {
self.functions.values()
}
}
pub struct HostBridge {
db: Db,
handle: tokio::runtime::Handle,
mailer: Option<Mailer>,
email_templates: Option<Arc<crate::email_templates::EmailTemplates>>,
cache: Option<Cache>,
payments: Option<Payments>,
ai: Option<Ai>,
queue: Option<Queue>,
chunks: Option<tokio::sync::mpsc::UnboundedSender<String>>,
config_json: String,
principal_id: String,
hook_json: String,
}
impl HostBridge {
pub fn new(
db: Db,
handle: tokio::runtime::Handle,
config_json: String,
principal_id: String,
) -> Self {
HostBridge {
db,
handle,
mailer: None,
email_templates: None,
cache: None,
payments: None,
ai: None,
queue: None,
chunks: None,
config_json,
principal_id,
hook_json: String::new(),
}
}
pub fn with_email_templates(
mut self,
templates: Arc<crate::email_templates::EmailTemplates>,
) -> Self {
self.email_templates = Some(templates);
self
}
pub fn with_services(
mut self,
mailer: Option<Mailer>,
cache: Option<Cache>,
payments: Option<Payments>,
ai: Option<Ai>,
) -> Self {
self.mailer = mailer;
self.cache = cache;
self.payments = payments;
self.ai = ai;
self
}
pub fn with_queue(mut self, queue: Queue) -> Self {
self.queue = Some(queue);
self
}
pub fn streaming(mut self, chunks: tokio::sync::mpsc::UnboundedSender<String>) -> Self {
self.chunks = Some(chunks);
self
}
async fn relay(
&self,
ai: &apiplant_ai::Ai,
request: &apiplant_ai::ChatRequest,
) -> Result<apiplant_ai::ChatReply, apiplant_ai::AiError> {
use futures_util::StreamExt;
let mut stream = Box::pin(ai.stream(request).await?);
let mut text = String::new();
let mut done = apiplant_ai::Done::default();
while let Some(event) = stream.next().await {
match event? {
apiplant_ai::Event::Delta(delta) => {
text.push_str(&delta);
if !self.emit(abi_stable::std_types::RStr::from_str(&delta)) {
break;
}
}
apiplant_ai::Event::Reasoning(_) => {}
apiplant_ai::Event::Done(end) => {
done = end;
break;
}
}
}
Ok(apiplant_ai::ChatReply {
text,
reasoning: String::new(),
provider: ai.provider().as_str().to_string(),
model: request
.model
.clone()
.unwrap_or_else(|| ai.model().to_string()),
done,
tool_calls: Vec::new(),
})
}
pub fn with_hook(mut self, hook_json: String) -> Self {
self.hook_json = hook_json;
self
}
}
impl HostApi for HostBridge {
fn query(&self, request: RStr<'_>) -> RResult<RString, RString> {
#[derive(serde::Deserialize)]
struct Req {
sql: String,
#[serde(default)]
params: Vec<serde_json::Value>,
}
let req: Req = match serde_json::from_str(request.as_str()) {
Ok(r) => r,
Err(e) => return RResult::RErr(format!("invalid query request: {e}").into()),
};
let result = self
.handle
.block_on(async { self.db.raw_json(&req.sql, &req.params).await });
match result {
Ok(v) => RResult::ROk(v.to_string().into()),
Err(e) => RResult::RErr(e.to_string().into()),
}
}
fn send_email(&self, request: RStr<'_>) -> RResult<RString, RString> {
let Some(mailer) = &self.mailer else {
return RResult::RErr(
"no email provider configured — set [email] provider in main.toml"
.to_string()
.into(),
);
};
let mut message: apiplant_email::Message = match serde_json::from_str(request.as_str()) {
Ok(m) => m,
Err(e) => return RResult::RErr(format!("invalid email: {e}").into()),
};
if let Some(name) = template_name(request.as_str()) {
let Some(templates) = &self.email_templates else {
return RResult::RErr(
format!("no email template `{name}`: this app has no emails/ directory").into(),
);
};
let vars = template_vars(request.as_str());
match templates.render(&name, &vars, &message.subject) {
Ok(rendered) => {
message.subject = rendered.subject;
message.text = rendered.text;
message.html = rendered.html;
}
Err(e) => return RResult::RErr(format!("{e:#}").into()),
}
}
match self.handle.block_on(mailer.send(&message)) {
Ok(sent) => RResult::ROk(
serde_json::to_string(&sent)
.unwrap_or_else(|_| "{}".to_string())
.into(),
),
Err(e) => RResult::RErr(e.to_string().into()),
}
}
fn cache(&self, request: RStr<'_>) -> RResult<RString, RString> {
let Some(cache) = &self.cache else {
return RResult::RErr(
"no cache configured — set [cache] url in main.toml"
.to_string()
.into(),
);
};
match self.handle.block_on(cache.execute(request.as_str())) {
Ok(value) => RResult::ROk(value.to_string().into()),
Err(e) => RResult::RErr(e.to_string().into()),
}
}
fn payments(&self, request: RStr<'_>) -> RResult<RString, RString> {
let Some(payments) = &self.payments else {
return RResult::RErr(
"no payment provider configured — set [payments] provider in main.toml"
.to_string()
.into(),
);
};
match self.handle.block_on(payments.execute(request.as_str())) {
Ok(value) => RResult::ROk(value.to_string().into()),
Err(e) => RResult::RErr(e.to_string().into()),
}
}
fn ai(&self, request: RStr<'_>) -> RResult<RString, RString> {
let Some(ai) = &self.ai else {
return RResult::RErr(
"no ai provider configured — set [ai] provider in main.toml"
.to_string()
.into(),
);
};
let raw: serde_json::Value = match serde_json::from_str(request.as_str()) {
Ok(r) => r,
Err(e) => return RResult::RErr(format!("invalid chat request: {e}").into()),
};
let forward = raw.get("stream").and_then(serde_json::Value::as_bool) == Some(true);
let request: apiplant_ai::ChatRequest = match serde_json::from_value(raw) {
Ok(r) => r,
Err(e) => return RResult::RErr(format!("invalid chat request: {e}").into()),
};
let result = match (forward, &self.chunks) {
(true, Some(_)) => self.handle.block_on(self.relay(ai, &request)),
_ => self.handle.block_on(ai.chat(&request)),
};
match result {
Ok(reply) => RResult::ROk(
serde_json::to_string(&reply)
.unwrap_or_else(|_| "{}".to_string())
.into(),
),
Err(e) => RResult::RErr(e.to_string().into()),
}
}
fn publish(&self, request: RStr<'_>) -> RResult<RString, RString> {
let Some(queue) = &self.queue else {
return RResult::RErr("this invocation has no queue attached".to_string().into());
};
match self
.handle
.block_on(queue.execute(request.as_str(), &self.principal_id))
{
Ok(value) => RResult::ROk(value.to_string().into()),
Err(e) => RResult::RErr(e.to_string().into()),
}
}
fn emit(&self, chunk: RStr<'_>) -> bool {
match &self.chunks {
Some(chunks) => chunks.send(chunk.as_str().to_string()).is_ok(),
None => true,
}
}
fn log(&self, level: LogLevel, message: RStr<'_>) {
let msg = message.as_str();
match level {
LogLevel::Trace => tracing::trace!(target: "apiplant::function", "{msg}"),
LogLevel::Debug => tracing::debug!(target: "apiplant::function", "{msg}"),
LogLevel::Info => tracing::info!(target: "apiplant::function", "{msg}"),
LogLevel::Warn => tracing::warn!(target: "apiplant::function", "{msg}"),
LogLevel::Error => tracing::error!(target: "apiplant::function", "{msg}"),
}
}
fn config(&self) -> RString {
self.config_json.clone().into()
}
fn principal_id(&self) -> RString {
self.principal_id.clone().into()
}
fn hook(&self) -> RString {
self.hook_json.clone().into()
}
}
fn template_name(request: &str) -> Option<String> {
serde_json::from_str::<serde_json::Value>(request)
.ok()?
.get("template")?
.as_str()
.map(str::trim)
.filter(|name| !name.is_empty())
.map(str::to_string)
}
fn template_vars(request: &str) -> liquid::Object {
let Ok(value) = serde_json::from_str::<serde_json::Value>(request) else {
return liquid::Object::new();
};
match value.get("vars") {
Some(vars) => liquid::model::to_object(vars).unwrap_or_default(),
None => liquid::Object::new(),
}
}
#[cfg(test)]
mod slim_tests {
use super::*;
#[test]
fn javascript_is_always_a_loadable_artifact() {
assert_eq!(JS_EXTENSION, "js");
#[cfg(feature = "typescript")]
assert_eq!(JS_EXTENSION, apiplant_js::EXTENSION);
}
#[cfg(not(feature = "typescript"))]
#[test]
fn a_slim_build_refuses_javascript_and_says_why() {
let error = match load_javascript(Path::new("/app/functions/greet.js")) {
Err(error) => error,
Ok(_) => panic!("a slim build has no isolate to run it in"),
};
assert!(error.contains("greet.js"), "{error}");
assert!(error.contains("TypeScript support"), "{error}");
}
}