apiplant_server/
call.rs

1//! Invoking one function without serving anything — the `apiplant call`
2//! command, and what a Kubernetes CronJob runs.
3//!
4//! A scheduled job wants exactly what an HTTP call to `<base>/functions/{name}`
5//! does, minus the server: the same function, the same database, the same
6//! email/cache/payments/AI services built from the same `main.toml`. So this
7//! builds those services the way [`crate::run_with`] does, hands the function a
8//! [`HostBridge`] over them, and returns what it returned.
9//!
10//! Two deliberate differences from the HTTP path:
11//!
12//! * **No access check.** There is no request to authenticate and no session to
13//!   read; whoever can run the binary against the database has already got more
14//!   than any endpoint would give them. `--as <USER_ID>` sets the principal the
15//!   function sees, for a function that reads it.
16//! * **Visibility is ignored.** A `Private` function has no route on purpose —
17//!   it exists to be called from a hook — and a cron job is that same kind of
18//!   caller, so it can call one.
19//!
20//! Migrations are *not* run: a job that starts by migrating is a job that can
21//! migrate a production database at 3am because it was scheduled to.
22
23use apiplant_core::App;
24use apiplant_db::Db;
25
26use crate::functions::{FunctionRegistry, HostBridge};
27
28/// How to call.
29#[derive(Debug, Clone, Default)]
30pub struct Options {
31    /// The function's input, as JSON. Empty means `{}`, matching the HTTP
32    /// endpoint's treatment of an empty body.
33    pub input: String,
34    /// The user id the function sees as its caller, if any.
35    pub principal: Option<String>,
36    /// Forward what the function `emit`s to stderr as it is produced, rather
37    /// than dropping it. Keeps a long job's progress visible in `kubectl logs`
38    /// without mixing into the result on stdout.
39    pub emit_to_stderr: bool,
40}
41
42/// Run one of the app's functions and return its JSON result.
43///
44/// The error is the function's own message, or the reason it couldn't be run.
45pub async fn call(app: &App, name: &str, options: Options) -> anyhow::Result<String> {
46    let registry = FunctionRegistry::load(app);
47    // Checked before anything is connected, so a typo'd name costs a database
48    // connection and a Stripe client rather than reporting after them.
49    let config_json = match registry.get(name) {
50        Some(f) => f.config_json.clone(),
51        None => {
52            let known = registry
53                .iter()
54                .map(|f| f.manifest.name.to_string())
55                .collect::<Vec<_>>();
56            anyhow::bail!(
57                "unknown function `{name}` — this app has: {}",
58                match known.is_empty() {
59                    true => "none".to_string(),
60                    false => known.join(", "),
61                }
62            );
63        }
64    };
65
66    let db = Db::connect(
67        &app.config.database.resolved_url(),
68        app.config.database.max_connections,
69    )
70    .await?;
71
72    let mailer = apiplant_email::Mailer::from_config(&app.config.email)?;
73    let cache = apiplant_cache::Cache::connect(&app.config.cache).await?;
74    let ai = apiplant_ai::Ai::from_config(&app.config.ai)?;
75    let payments = apiplant_payments::Payments::from_config(
76        &app.config.payments,
77        &app.config.server.public_origin(),
78    )?;
79    // A job that publishes is a normal thing to want — a nightly sweep queuing
80    // one message per row it found. Nothing here *subscribes*, though: the
81    // messages are handled by the running server, not by this process, which
82    // exits as soon as the function returns.
83    let queue = apiplant_queue::Queue::new(&db, app);
84
85    // The chunks are drained on this task while the function runs on a blocking
86    // one, so a chatty function can't fill the channel unread.
87    let (printer, chunks) = match options.emit_to_stderr {
88        true => {
89            let (chunks, mut receiver) = tokio::sync::mpsc::unbounded_channel::<String>();
90            let printer = tokio::spawn(async move {
91                while let Some(chunk) = receiver.recv().await {
92                    eprint!("{chunk}");
93                }
94            });
95            (Some(printer), Some(chunks))
96        }
97        false => (None, None),
98    };
99
100    let mut bridge = HostBridge::new(
101        db,
102        tokio::runtime::Handle::current(),
103        config_json,
104        options.principal.unwrap_or_default(),
105    )
106    .with_services(mailer, cache, payments, ai)
107    .with_queue(queue);
108    if let Some(chunks) = chunks {
109        bridge = bridge.streaming(chunks);
110    }
111
112    let input = match options.input.trim().is_empty() {
113        true => "{}".to_string(),
114        false => options.input,
115    };
116    let name = name.to_string();
117    let result = tokio::task::spawn_blocking(move || {
118        let f = registry.get(&name).expect("checked above");
119        f.invoke(bridge, &input)
120    })
121    .await
122    .map_err(|_| anyhow::anyhow!("the function task panicked"))?;
123
124    if let Some(printer) = printer {
125        // The sender dropped with the bridge, so this ends on its own.
126        let _ = printer.await;
127    }
128
129    result.map_err(|message| {
130        match message.strip_prefix(apiplant_abi::INTERNAL_ERROR_PREFIX) {
131            // Unlike the HTTP path there is nobody to hide internals from: an
132            // operator reading `kubectl logs` is exactly who the detail is for.
133            Some(detail) => anyhow::anyhow!("function faulted: {detail}"),
134            None => anyhow::anyhow!("{message}"),
135        }
136    })
137}
apiplant_server/call.rs

apiplant_server/
call.rs
