config/
lib.rs

1use crate::epoch::{get_current_epoch, ConfigEpoch};
2use crate::pool::{pool_get, pool_put};
3pub use crate::pool::{set_gc_on_put, set_max_age, set_max_spare, set_max_use};
4use anyhow::Context;
5use kumo_prometheus::declare_metric;
6use kumo_prometheus::prometheus::HistogramTimer;
7use mlua::{
8    FromLua, FromLuaMulti, IntoLua, IntoLuaMulti, Lua, LuaSerdeExt, MetaMethod, RegistryKey, Table,
9    UserData, UserDataMethods, Value,
10};
11use parking_lot::FairMutex as Mutex;
12pub use pastey as paste;
13use serde::Serialize;
14use std::borrow::Cow;
15use std::collections::HashSet;
16use std::path::PathBuf;
17use std::sync::atomic::{AtomicBool, Ordering};
18use std::sync::{LazyLock, Once};
19use std::time::Instant;
20
21pub mod epoch;
22mod pool;
23
24static POLICY_FILE: LazyLock<Mutex<Option<PathBuf>>> = LazyLock::new(|| Mutex::new(None));
25static FUNCS: LazyLock<Mutex<Vec<RegisterFunc>>> = LazyLock::new(|| Mutex::new(vec![]));
26
27declare_metric! {
28/// how many times the policy lua script has been loaded into a new context
29static LUA_LOAD_COUNT: IntCounter("lua_load_count");
30}
31
32declare_metric! {
33/// the number of lua contexts currently alive
34static LUA_COUNT: IntGauge("lua_count");
35}
36
37static CALLBACK_ALLOWS_MULTIPLE: LazyLock<Mutex<HashSet<String>>> =
38    LazyLock::new(|| Mutex::new(HashSet::new()));
39
40pub static VALIDATE_ONLY: AtomicBool = AtomicBool::new(false);
41pub static VALIDATION_FAILED: AtomicBool = AtomicBool::new(false);
42
43declare_metric! {
44/// how long a given lua event callback took
45static LATENCY_HIST: HistogramVec(
46        "lua_event_latency",
47        &["event"]);
48}
49
50declare_metric! {
51/// Incremented each time we start to call a lua event callback. Use lua_event_latency_count to track completed events
52static EVENT_STARTED_COUNT: CounterVec(
53        "lua_event_started",
54        &["event"]
55    );
56}
57
58pub type RegisterFunc = fn(&Lua) -> anyhow::Result<()>;
59
60fn latency_timer(label: &str) -> HistogramTimer {
61    EVENT_STARTED_COUNT
62        .get_metric_with_label_values(&[label])
63        .expect("to get counter")
64        .inc();
65    LATENCY_HIST
66        .get_metric_with_label_values(&[label])
67        .expect("to get histo")
68        .start_timer()
69}
70
71#[derive(Debug)]
72struct LuaConfigInner {
73    lua: Lua,
74    created: Instant,
75    use_count: usize,
76    epoch: ConfigEpoch,
77}
78
79impl Drop for LuaConfigInner {
80    fn drop(&mut self) {
81        LUA_COUNT.dec();
82    }
83}
84
85#[derive(Debug)]
86pub struct LuaConfig {
87    inner: Option<LuaConfigInner>,
88}
89
90pub async fn set_policy_path(path: PathBuf) -> anyhow::Result<()> {
91    POLICY_FILE.lock().replace(path);
92    let config = load_config().await?;
93    config.put();
94    Ok(())
95}
96
97fn get_policy_path() -> Option<PathBuf> {
98    POLICY_FILE.lock().clone()
99}
100
101fn get_funcs() -> Vec<RegisterFunc> {
102    FUNCS.lock().clone()
103}
104pub fn is_validating() -> bool {
105    VALIDATE_ONLY.load(Ordering::Relaxed)
106}
107
108pub fn validation_failed() -> bool {
109    VALIDATION_FAILED.load(Ordering::Relaxed)
110}
111
112pub fn set_validation_failed() {
113    VALIDATION_FAILED.store(true, Ordering::Relaxed)
114}
115
116pub async fn load_config() -> anyhow::Result<LuaConfig> {
117    if let Some(pool) = pool_get() {
118        return Ok(pool);
119    }
120
121    LUA_LOAD_COUNT.inc();
122    let lua = Lua::new();
123    let created = Instant::now();
124    let epoch = get_current_epoch();
125
126    {
127        let globals = lua.globals();
128
129        if is_validating() {
130            globals.set("_VALIDATING_CONFIG", true)?;
131        }
132
133        let package: Table = globals.get("package")?;
134        let package_path: String = package.get("path")?;
135        let mut path_array: Vec<String> = package_path.split(";").map(|s| s.to_owned()).collect();
136
137        fn prefix_path(array: &mut Vec<String>, path: &str) {
138            array.insert(0, format!("{}/?.lua", path));
139            array.insert(1, format!("{}/?/init.lua", path));
140        }
141
142        prefix_path(&mut path_array, "/opt/kumomta/etc/policy");
143        prefix_path(&mut path_array, "/opt/kumomta/share");
144
145        #[cfg(debug_assertions)]
146        prefix_path(&mut path_array, "assets");
147
148        package.set("path", path_array.join(";"))?;
149    }
150
151    register_declared_events();
152
153    for func in get_funcs() {
154        (func)(&lua)?;
155    }
156
157    if let Some(policy) = get_policy_path() {
158        let code = tokio::fs::read_to_string(&policy)
159            .await
160            .with_context(|| format!("reading policy file {policy:?}"))?;
161
162        let func = {
163            let chunk = lua.load(&code);
164            let chunk = chunk.set_name(policy.to_string_lossy());
165            chunk.into_function()?
166        };
167
168        let _timer = latency_timer("context-creation");
169        func.call_async::<()>(()).await?;
170    }
171    LUA_COUNT.inc();
172
173    Ok(LuaConfig {
174        inner: Some(LuaConfigInner {
175            lua,
176            created,
177            use_count: 1,
178            epoch,
179        }),
180    })
181}
182
183pub fn register(func: RegisterFunc) {
184    FUNCS.lock().push(func);
185}
186
187impl LuaConfig {
188    fn set_current_event(&mut self, name: &str) -> mlua::Result<()> {
189        self.inner
190            .as_mut()
191            .unwrap()
192            .lua
193            .globals()
194            .set("_KUMO_CURRENT_EVENT", name.to_string())
195    }
196
197    /// Returns the Lua context managed by this config, or an error if the
198    /// context is not currently held by this LuaConfig.
199    pub fn lua(&self) -> anyhow::Result<&Lua> {
200        Ok(&self
201            .inner
202            .as_ref()
203            .context("LuaConfig is missing the inner Lua context")?
204            .lua)
205    }
206
207    /// Convert an array of args into a MultiValue that can be passed
208    /// to a callback signature
209    pub fn convert_args_to_multi<A: Serialize>(
210        &self,
211        args: &[A],
212    ) -> anyhow::Result<mlua::MultiValue> {
213        let lua = self.inner.as_ref().unwrap();
214        let mut arg_vec = vec![];
215        for a in args.iter() {
216            arg_vec.push(lua.lua.to_value(a)?);
217        }
218        Ok(mlua::MultiValue::from_vec(arg_vec))
219    }
220
221    /// Intended to be used together with kumo.spawn_task
222    pub async fn convert_args_and_call_callback<A: Serialize>(
223        &mut self,
224        sig: &CallbackSignature<Value, ()>,
225        args: A,
226    ) -> anyhow::Result<()> {
227        let lua = self.inner.as_mut().unwrap();
228        let args = lua.lua.to_value(&args)?;
229
230        let name = sig.name();
231        let decorated_name = sig.decorated_name();
232
233        match lua
234            .lua
235            .named_registry_value::<mlua::Function>(&decorated_name)
236        {
237            Ok(func) => {
238                let _timer = latency_timer(name);
239                Ok(func.call_async(args).await?)
240            }
241            _ => anyhow::bail!("{name} has not been registered"),
242        }
243    }
244
245    /// Explicitly put the config object back into its containing pool.
246    /// Ideally we'd do this automatically when the object is dropped,
247    /// but lua's garbage collection makes this problematic:
248    /// if a future whose graph contains an async lua call within
249    /// this config object is cancelled (eg: simply stopped without
250    /// calling it again), and the config object is not explicitly garbage
251    /// collected, any futures and data owned by any dependencies of
252    /// the cancelled future remain alive until the next gc run,
253    /// which can cause things like async locks and semaphores to
254    /// have a lifetime extended by the maximum age of the lua context.
255    ///
256    /// The combat this, consumers of LuaConfig should explicitly
257    /// call `config.put()` after successfully using the config
258    /// object.
259    ///
260    /// Or framing it another way: consumers must not call `config.put()`
261    /// if a transitive dep might have been cancelled.
262    pub fn put(mut self) {
263        if let Some(inner) = self.inner.take() {
264            pool_put(inner);
265        }
266    }
267
268    pub async fn call_callback<'a, A: IntoLuaMulti + Clone, R: FromLuaMulti>(
269        &mut self,
270        sig: &'a CallbackSignature<A, R>,
271        args: A,
272    ) -> anyhow::Result<CallbackDisposition<'a, R>> {
273        let name = sig.name();
274        self.set_current_event(name)?;
275        let lua = self.inner.as_mut().unwrap();
276        sig.call(&lua.lua, args).await
277    }
278
279    pub async fn async_call_callback<A: IntoLuaMulti + Clone, R: FromLuaMulti + Default>(
280        &mut self,
281        sig: &CallbackSignature<A, R>,
282        args: A,
283    ) -> anyhow::Result<R> {
284        let name = sig.name();
285        self.set_current_event(name)?;
286        let lua = self.inner.as_mut().unwrap();
287        Ok(sig.call(&lua.lua, args).await?.or_default())
288    }
289
290    pub async fn async_call_callback_non_default<A: IntoLuaMulti + Clone, R: FromLuaMulti>(
291        &mut self,
292        sig: &CallbackSignature<A, R>,
293        args: A,
294    ) -> anyhow::Result<R> {
295        let name = sig.name();
296        self.set_current_event(name)?;
297        let lua = self.inner.as_mut().unwrap();
298        sig.call(&lua.lua, args).await?.require_value()
299    }
300
301    pub async fn async_call_callback_non_default_opt<A: IntoLuaMulti + Clone, R: FromLua>(
302        &mut self,
303        sig: &CallbackSignature<A, Option<R>>,
304        args: A,
305    ) -> anyhow::Result<Option<R>> {
306        let name = sig.name();
307        self.set_current_event(name)?;
308        let lua = self.inner.as_mut().unwrap();
309        let result = sig.call(&lua.lua, args).await?;
310        match result.result {
311            None => Ok(None),
312            Some(result) => Ok(result),
313        }
314    }
315
316    pub fn remove_registry_value(&mut self, value: RegistryKey) -> anyhow::Result<()> {
317        Ok(self
318            .inner
319            .as_mut()
320            .unwrap()
321            .lua
322            .remove_registry_value(value)?)
323    }
324
325    /// Call a constructor registered via `on`. Returns a registry key that can be
326    /// used to reference the returned value again later on this same Lua instance
327    pub async fn async_call_ctor<A: IntoLuaMulti + Clone>(
328        &mut self,
329        sig: &CallbackSignature<A, Value>,
330        args: A,
331    ) -> anyhow::Result<RegistryKey> {
332        let name = sig.name();
333        anyhow::ensure!(
334            !sig.allow_multiple(),
335            "ctor event signature for {name} is defined as allow_multiple, which is not supported"
336        );
337
338        let decorated_name = sig.decorated_name();
339        self.set_current_event(name)?;
340
341        let inner = self.inner.as_mut().unwrap();
342
343        let func = inner
344            .lua
345            .named_registry_value::<mlua::Function>(&decorated_name)?;
346
347        let _timer = latency_timer(name);
348        let value: Value = func.call_async(args.clone()).await?;
349        drop(func);
350
351        Ok(inner.lua.create_registry_value(value)?)
352    }
353
354    /// Operate on an object/value that was previously constructed via
355    /// async_call_ctor.
356    pub async fn with_registry_value<F, R, FUT>(
357        &mut self,
358        value: &RegistryKey,
359        func: F,
360    ) -> anyhow::Result<R>
361    where
362        R: FromLuaMulti,
363        F: FnOnce(Value) -> anyhow::Result<FUT>,
364        FUT: std::future::Future<Output = anyhow::Result<R>>,
365    {
366        let inner = self.inner.as_mut().unwrap();
367        let value = inner.lua.registry_value(value)?;
368        let future = (func)(value)?;
369        future.await
370    }
371}
372
373pub fn get_or_create_module(lua: &Lua, name: &str) -> anyhow::Result<mlua::Table> {
374    let globals = lua.globals();
375    let package: Table = globals.get("package")?;
376    let loaded: Table = package.get("loaded")?;
377
378    let module = loaded.get(name)?;
379    match module {
380        Value::Nil => {
381            let module = lua.create_table()?;
382            loaded.set(name, module.clone())?;
383            Ok(module)
384        }
385        Value::Table(table) => Ok(table),
386        wat => anyhow::bail!(
387            "cannot register module {} as package.loaded.{} is already set to a value of type {}",
388            name,
389            name,
390            wat.type_name()
391        ),
392    }
393}
394
395/// Given a name path like `foo` or `foo.bar.baz`, sets up the module
396/// registry hierarchy to instantiate that path.
397/// Returns the leaf node of that path to allow the caller to
398/// register/assign functions etc. into it
399pub fn get_or_create_sub_module(lua: &Lua, name_path: &str) -> anyhow::Result<mlua::Table> {
400    let mut parent = get_or_create_module(lua, "kumo")?;
401    let mut path_so_far = String::new();
402
403    for name in name_path.split('.') {
404        if !path_so_far.is_empty() {
405            path_so_far.push('.');
406        }
407        path_so_far.push_str(name);
408
409        let sub = parent.get(name)?;
410        match sub {
411            Value::Nil => {
412                let sub = lua.create_table()?;
413                parent.set(name, sub.clone())?;
414                parent = sub;
415            }
416            Value::Table(sub) => {
417                parent = sub;
418            }
419            wat => anyhow::bail!(
420                "cannot register module kumo.{path_so_far} as it is already set to a value of type {}",
421                wat.type_name()
422            ),
423        }
424    }
425
426    Ok(parent)
427}
428
429/// Helper for mapping back to lua errors
430pub fn any_err<E: std::fmt::Display>(err: E) -> mlua::Error {
431    mlua::Error::external(format!("{err:#}"))
432}
433
434/// Provides implementations of __pairs, __index and __len metamethods
435/// for a type that is Serialize and UserData.
436/// Neither implementation is considered to be ideal, as we must
437/// first serialize the value into a json Value which is then either
438/// iterated over, or indexed to produce the appropriate result for
439/// the metamethod.
440pub fn impl_pairs_and_index<T, M>(methods: &mut M)
441where
442    T: UserData + Serialize,
443    M: UserDataMethods<T>,
444{
445    methods.add_meta_method(MetaMethod::Pairs, move |lua, this, _: ()| {
446        let Ok(serde_json::Value::Object(map)) = serde_json::to_value(this).map_err(any_err) else {
447            return Err(mlua::Error::external("must serialize to Map"));
448        };
449
450        let mut value_iter = map.into_iter();
451
452        let iter_func = lua.create_function_mut(
453            move |lua, (_state, _control): (Value, Value)| match value_iter.next() {
454                Some((key, value)) => {
455                    let key = lua.to_value(&key)?;
456                    let value = lua.to_value(&value)?;
457                    Ok((key, value))
458                }
459                None => Ok((Value::Nil, Value::Nil)),
460            },
461        )?;
462
463        Ok((Value::Function(iter_func), Value::Nil, Value::Nil))
464    });
465
466    methods.add_meta_method(MetaMethod::Index, move |lua, this, field: Value| {
467        let value = lua.to_value(this)?;
468        match value {
469            Value::Table(t) => t.get(field),
470            _ => Ok(Value::Nil),
471        }
472    });
473
474    methods.add_meta_method(MetaMethod::Len, move |lua, this, _: ()| {
475        let value = lua.to_value(this)?;
476        match value {
477            Value::Table(v) => v.len(),
478            Value::String(v) => Ok(v.as_bytes().len() as i64),
479            _ => Ok(0),
480        }
481    });
482}
483
484/// This function will try to obtain a native lua representation
485/// of the provided value. It does this by attempting to iterate
486/// the pairs of any userdata it finds as either the value itself
487/// or the values of a table value by recursively applying
488/// materialize_to_lua_value to the value.
489/// This produces a lua value that can then be processed by the
490/// Deserialize impl on Value.
491pub fn materialize_to_lua_value(lua: &Lua, value: mlua::Value) -> mlua::Result<mlua::Value> {
492    match value {
493        mlua::Value::UserData(ud) => {
494            let mt = ud.metatable()?;
495            let Ok(pairs) = mt.get::<mlua::Function>("__pairs") else {
496                let value = ud.into_lua(lua)?;
497                return Err(mlua::Error::external(format!(
498                    "cannot materialize_to_lua_value {value:?} \
499                     because it has no __pairs metamethod"
500                )));
501            };
502            let tbl = lua.create_table()?;
503            let (iter_func, state, mut control): (mlua::Function, mlua::Value, mlua::Value) =
504                pairs.call(mlua::Value::UserData(ud.clone()))?;
505
506            loop {
507                let (k, v): (mlua::Value, mlua::Value) =
508                    iter_func.call((state.clone(), control))?;
509                if k.is_nil() {
510                    break;
511                }
512
513                tbl.set(k.clone(), materialize_to_lua_value(lua, v)?)?;
514                control = k;
515            }
516
517            Ok(mlua::Value::Table(tbl))
518        }
519        mlua::Value::Table(t) => {
520            let tbl = lua.create_table()?;
521            for pair in t.pairs::<mlua::Value, mlua::Value>() {
522                let (k, v) = pair?;
523                tbl.set(k.clone(), materialize_to_lua_value(lua, v)?)?;
524            }
525            Ok(mlua::Value::Table(tbl))
526        }
527        value => Ok(value),
528    }
529}
530
531/// Helper wrapper type for passing/returning serde encoded values from/to lua
532pub struct SerdeWrappedValue<T>(pub T);
533
534impl<T: serde::Serialize> serde::Serialize for SerdeWrappedValue<T> {
535    fn serialize<S>(
536        &self,
537        s: S,
538    ) -> Result<<S as serde::Serializer>::Ok, <S as serde::Serializer>::Error>
539    where
540        S: serde::Serializer,
541    {
542        self.0.serialize(s)
543    }
544}
545
546impl<T: Clone> Clone for SerdeWrappedValue<T> {
547    fn clone(&self) -> Self {
548        SerdeWrappedValue(self.0.clone())
549    }
550}
551
552impl<T: Default> Default for SerdeWrappedValue<T> {
553    fn default() -> Self {
554        SerdeWrappedValue(Default::default())
555    }
556}
557
558impl<T: serde::Serialize> SerdeWrappedValue<T> {
559    pub fn to_lua_value(&self, lua: &Lua) -> mlua::Result<mlua::Value> {
560        lua.to_value_with(&self.0, serialize_options())
561    }
562}
563
564impl<T: serde::Serialize> IntoLua for SerdeWrappedValue<T> {
565    fn into_lua(self, lua: &Lua) -> mlua::Result<mlua::Value> {
566        lua.to_value_with(&self.0, serialize_options())
567    }
568}
569
570impl<T: serde::de::DeserializeOwned> FromLua for SerdeWrappedValue<T> {
571    fn from_lua(value: mlua::Value, lua: &Lua) -> mlua::Result<SerdeWrappedValue<T>> {
572        let inner: T = from_lua_value(lua, value)?;
573        Ok(SerdeWrappedValue(inner))
574    }
575}
576
577impl<T> std::ops::Deref for SerdeWrappedValue<T> {
578    type Target = T;
579    fn deref(&self) -> &T {
580        &self.0
581    }
582}
583
584impl<T> std::ops::DerefMut for SerdeWrappedValue<T> {
585    fn deref_mut(&mut self) -> &mut T {
586        &mut self.0
587    }
588}
589
590/// Convert from a lua value to a deserializable type,
591/// with a slightly more helpful error message in case of failure.
592/// NOTE: the ", while processing" portion of the error messages generated
593/// here is coupled with a regex in typing.lua!
594pub fn from_lua_value<R>(lua: &Lua, value: mlua::Value) -> mlua::Result<R>
595where
596    R: serde::de::DeserializeOwned,
597{
598    let value_cloned = value.clone();
599    match lua.from_value(value) {
600        Ok(r) => Ok(r),
601        Err(err) => match materialize_to_lua_value(lua, value_cloned.clone()) {
602            Ok(materialized) => match lua.from_value(materialized.clone()) {
603                Ok(r) => Ok(r),
604                Err(err) => {
605                    let mut serializer = serde_json::Serializer::new(Vec::new());
606                    let serialized = match materialized.serialize(&mut serializer) {
607                        Ok(_) => String::from_utf8_lossy(&serializer.into_inner()).to_string(),
608                        Err(err) => format!("<unable to encode as json: {err:#}>"),
609                    };
610                    Err(mlua::Error::external(format!(
611                        "{err:#}, while processing {serialized}"
612                    )))
613                }
614            },
615            Err(materialize_err) => Err(mlua::Error::external(format!(
616                "{err:#}, while processing a userdata. \
617                    Additionally, encountered {materialize_err:#} \
618                    when trying to iterate the pairs of that userdata"
619            ))),
620        },
621    }
622}
623
624/// CallbackSignature is a bit sugar to aid with statically typing event callback
625/// function invocation.
626///
627/// The idea is that you declare a signature instance that is typed
628/// with its argument tuple (A), and its return type tuple (R).
629///
630/// The signature instance can then be used to invoke the callback by name.
631///
632/// The register method allows pre-registering events so that `kumo.on`
633/// can reason about them better.  The main function enabled by this is
634/// `allow_multiple`; when that is set to true, `kumo.on` will allow
635/// recording multiple callback instances, calling them in sequence
636/// until one of them returns a value.
637pub struct CallbackSignature<A, R>
638where
639    A: IntoLuaMulti,
640    R: FromLuaMulti,
641{
642    marker: std::marker::PhantomData<(A, R)>,
643    allow_multiple: bool,
644    name: Cow<'static, str>,
645}
646
647#[linkme::distributed_slice]
648pub static CALLBACK_SIGNATURES: [fn()];
649
650/// Helper for declaring a named event handler callback signature.
651///
652/// Usage looks like:
653///
654/// ```rust,ignore
655/// declare_event! {
656/// pub static GET_Q_CONFIG_SIG: Multiple(
657///         "get_queue_config",
658///         domain: &'static str,
659///         tenant: Option<&'static str>,
660///         campaign: Option<&'static str>,
661///         routing_domain: Option<&'static str>,
662///     ) -> QueueConfig;
663/// }
664/// ```
665///
666/// A handler can be either `Single` or `Multiple`, indicating whether
667/// only a single registration or multiple registrations are permitted.
668/// The string literal is the name of the event, followed by a fn-style
669/// parameter list which names each parameter in sequence, followed by
670/// the return value.  The names are not currently used in any way,
671/// but enhance the readability of the code.
672///
673/// In addition to declaring the signature in a global, some glue
674/// is generated that will register the signature appropriately
675/// so that lua knows whether it is single or multiple and can
676/// act appropriately when `kumo.on` is called.
677#[macro_export]
678macro_rules! declare_event {
679    ($vis:vis static $sym:ident: Multiple($name:literal $(,)? $($param_name:ident: $args:ty),* $(,)? ) -> $ret:ty;) => {
680        $vis static $sym: ::std::sync::LazyLock<
681            $crate::CallbackSignature<($($args),*), $ret>> =
682                ::std::sync::LazyLock::new(|| $crate::CallbackSignature::new_with_multiple($name));
683
684        $crate::paste::paste! {
685            #[linkme::distributed_slice($crate::CALLBACK_SIGNATURES)]
686            static [<CALLBACK_SIG_REGISTER_ $sym>]: fn() = || {
687                $sym.register();
688            };
689        }
690    };
691    ($vis:vis static $sym:ident: Single($name:literal $(,)? $($param_name:ident: $args:ty),* $(,)? ) -> $ret:ty;) => {
692        $vis static $sym: ::std::sync::LazyLock<
693            $crate::CallbackSignature<($($args),*), $ret>> =
694                ::std::sync::LazyLock::new(|| $crate::CallbackSignature::new($name));
695
696        $crate::paste::paste! {
697            #[linkme::distributed_slice($crate::CALLBACK_SIGNATURES)]
698            static [<CALLBACK_SIG_REGISTER_ $sym>]: fn() = || {
699                $sym.register();
700            };
701        }
702    };
703}
704
705/// For each event handler CallbackSignature that was declared via
706/// `declare_event!`, call its `.register()` method to register
707/// it so that `kumo.on` can give appropriate messaging if misused,
708/// and so that runtime dispatch will work correctly.
709///
710/// This should be called once, prior to running any lua code.
711fn register_declared_events() {
712    static ONCE: Once = Once::new();
713    ONCE.call_once(|| {
714        for reg_func in CALLBACK_SIGNATURES {
715            reg_func();
716        }
717    });
718}
719
720impl<A, R> CallbackSignature<A, R>
721where
722    A: IntoLuaMulti,
723    R: FromLuaMulti,
724{
725    pub fn new<S: Into<Cow<'static, str>>>(name: S) -> Self {
726        let name = name.into();
727
728        Self {
729            marker: std::marker::PhantomData,
730            allow_multiple: false,
731            name,
732        }
733    }
734
735    /// Make sure that you call .register() on this from
736    /// eg: mod_kumo::register in order for it to be instantiated
737    /// and visible to the config loader
738    pub fn new_with_multiple<S: Into<Cow<'static, str>>>(name: S) -> Self {
739        let name = name.into();
740
741        Self {
742            marker: std::marker::PhantomData,
743            allow_multiple: true,
744            name,
745        }
746    }
747
748    pub fn register(&self) {
749        if self.allow_multiple {
750            CALLBACK_ALLOWS_MULTIPLE
751                .lock()
752                .insert(self.name.to_string());
753        }
754    }
755
756    pub fn raise_error_if_allow_multiple(&self) -> anyhow::Result<()> {
757        anyhow::ensure!(
758            !self.allow_multiple(),
759            "handler {} is set to allow multiple handlers \
760                    but is registered with a single instance. This indicates that \
761                    register() was not called on the signature when initializing \
762                    the lua context. Please report this issue to the KumoMTA team!",
763            self.name
764        );
765        Ok(())
766    }
767
768    /// Return true if this signature allows multiple instances to be registered
769    /// and called.
770    pub fn allow_multiple(&self) -> bool {
771        self.allow_multiple
772    }
773
774    pub fn name(&self) -> &str {
775        &self.name
776    }
777
778    pub fn decorated_name(&self) -> String {
779        decorate_callback_name(&self.name)
780    }
781}
782
783impl<A, R> CallbackSignature<A, R>
784where
785    A: IntoLuaMulti + Clone,
786    R: FromLuaMulti,
787{
788    /// Calls the callback, passing in the supplied arguments.
789    /// Returns the callback disposition which allows further
790    /// intrepretation of the result, such as deciding whether
791    /// to Default the value based on whether the callback
792    /// was defined or not
793    pub async fn call<'a>(
794        &'a self,
795        lua: &Lua,
796        args: A,
797    ) -> anyhow::Result<CallbackDisposition<'a, R>> {
798        let name = self.name();
799        let decorated_name = self.decorated_name();
800
801        match lua.named_registry_value::<mlua::Value>(&decorated_name)? {
802            Value::Table(tbl) => {
803                for func in tbl.sequence_values::<mlua::Function>().collect::<Vec<_>>() {
804                    let func = func?;
805                    let _timer = latency_timer(name);
806                    let result: mlua::MultiValue = func.call_async(args.clone()).await?;
807                    if result.is_empty() {
808                        // Continue with other handlers
809                        continue;
810                    }
811                    let result = R::from_lua_multi(result, lua)?;
812                    return Ok(CallbackDisposition {
813                        handler_was_defined: true,
814                        result: Some(result),
815                        event_name: name,
816                    });
817                }
818                Ok(CallbackDisposition {
819                    handler_was_defined: false,
820                    result: None,
821                    event_name: name,
822                })
823            }
824            Value::Function(func) => {
825                self.raise_error_if_allow_multiple()?;
826                let _timer = latency_timer(name);
827                Ok(CallbackDisposition {
828                    handler_was_defined: true,
829                    result: Some(func.call_async(args.clone()).await?),
830                    event_name: name,
831                })
832            }
833            _ => Ok(CallbackDisposition {
834                handler_was_defined: false,
835                result: None,
836                event_name: name,
837            }),
838        }
839    }
840}
841
842pub fn does_callback_allow_multiple(name: &str) -> bool {
843    CALLBACK_ALLOWS_MULTIPLE.lock().contains(name)
844}
845
846pub fn decorate_callback_name(name: &str) -> String {
847    format!("kumomta-on-{name}")
848}
849
850/// Allows reasoning about the result of a callback to decide
851/// how to interpret the result
852pub struct CallbackDisposition<'a, T> {
853    /// Indicates whether the handler was found. If false,
854    /// then result will also be None.
855    /// If true and result.is_none(), it indicates that
856    /// none of the handlers returned a value.
857    pub handler_was_defined: bool,
858    /// The result!
859    pub result: Option<T>,
860    /// Which event was called (or to be called)
861    pub event_name: &'a str,
862}
863
864impl<'a, T> CallbackDisposition<'a, T> {
865    /// Requires that a value be returned.  Translates the disposition
866    /// into appropriate error messaging if no value was returned.
867    pub fn require_value(mut self) -> anyhow::Result<T> {
868        if !self.handler_was_defined {
869            anyhow::bail!("Event {} has not been registered", self.event_name);
870        }
871        match self.result.take() {
872            Some(value) => Ok(value),
873            None => anyhow::bail!("Event {} did not return a value", self.event_name),
874        }
875    }
876}
877impl<'a, T: Default> CallbackDisposition<'a, T> {
878    /// Unwraps the value type, or the Default impl if no value was returned
879    pub fn or_default(self) -> T {
880        self.result.unwrap_or_default()
881    }
882}
883
884pub fn serialize_options() -> mlua::SerializeOptions {
885    mlua::SerializeOptions::new()
886        .serialize_none_to_null(false)
887        .serialize_unit_to_null(false)
888}
config/lib.rs

config/
lib.rs
