sim_lib_server/
trigger.rs

1use std::{
2    sync::{
3        Arc, Mutex,
4        atomic::{AtomicBool, AtomicU64},
5        mpsc::Receiver,
6    },
7    thread::JoinHandle,
8    time::Duration,
9};
10
11use sim_citizen_derive::non_citizen;
12use sim_kernel::{
13    ClassRef, Cx, Error, Expr, Object, ReadPolicy, Result, Symbol, TrustLevel, Value,
14    read_construct_capability, read_eval_capability,
15};
16
17use crate::{Server, ServerAddress, WallClock, cron::CronMatcher, ensure_installed_codec};
18
19mod runtime;
20mod source_options;
21mod sources;
22
23use source_options::source_loopback;
24use sources::{TriggerSource, build_network_source, require_source_capability, source_capability};
25#[cfg(test)]
26use sources::{loopback_smtp_messages, queued_trigger_events};
27
28static NEXT_TRIGGER_ID: AtomicU64 = AtomicU64::new(1);
29
30#[derive(Clone)]
31enum TriggerDecoder {
32    Codec(Symbol),
33    Callable(Value),
34}
35
36struct TriggerConfig {
37    source: ServerAddress,
38    source_expr: Expr,
39    role: Option<Symbol>,
40    codec: Symbol,
41    decode_expr: Expr,
42    decoder: TriggerDecoder,
43    cron: Option<CronMatcher>,
44    network_source: Option<Box<dyn TriggerSource>>,
45}
46
47#[derive(Default)]
48pub(crate) struct TriggerState {
49    file_offset: usize,
50    file_remainder: Vec<u8>,
51    delivered: u64,
52    source_closed: bool,
53    cron_high_watermark: Option<u64>,
54}
55
56impl TriggerState {
57    pub(crate) fn advance_cron_high_watermark(&mut self, current_minute: u64) -> bool {
58        if self
59            .cron_high_watermark
60            .is_some_and(|high_watermark| current_minute <= high_watermark)
61        {
62            return false;
63        }
64        self.cron_high_watermark = Some(current_minute);
65        true
66    }
67}
68
69enum StdinSource {
70    Channel(Mutex<Receiver<Option<Vec<u8>>>>),
71    Unavailable,
72}
73
74#[non_citizen(
75    reason = "live trigger handle; reconstruct source through server/Address descriptor and trigger ops",
76    kind = "handle",
77    descriptor = "server/Address"
78)]
79/// Live handle to a running trigger that delivers source events into a server.
80pub struct TriggerHandle {
81    id: u64,
82    server: Arc<Server>,
83    source: ServerAddress,
84    source_expr: Expr,
85    role: Option<Symbol>,
86    codec: Symbol,
87    decode_expr: Expr,
88    decoder: TriggerDecoder,
89    cron: Option<CronMatcher>,
90    network_source: Option<Mutex<Box<dyn TriggerSource>>>,
91    wall_clock: Arc<dyn WallClock>,
92    stopping: AtomicBool,
93    handle: Mutex<Option<JoinHandle<()>>>,
94    stdin: StdinSource,
95    state: Mutex<TriggerState>,
96}
97
98#[cfg(test)]
99#[allow(dead_code)]
100pub(crate) fn enqueue_trigger_event(source: &ServerAddress, payload: Vec<u8>) -> Result<()> {
101    let key = match source {
102        ServerAddress::Webhook { route } => Some(format!("webhook:{route}")),
103        ServerAddress::Imap { address, mailbox } => Some(format!("imap:{address}:{mailbox}")),
104        ServerAddress::Smtp { address } => Some(format!("smtp:{address}")),
105        ServerAddress::Telegram { chat_id, bot } => Some(format!("telegram:{chat_id}:{bot}")),
106        ServerAddress::Matrix { room_id } => Some(format!("matrix:{room_id}")),
107        _ => None,
108    }
109    .ok_or_else(|| Error::Eval("source does not use the queued trigger runtime".to_owned()))?;
110    let mut queues = queued_trigger_events()
111        .lock()
112        .map_err(|_| Error::PoisonedLock("trigger queue"))?;
113    queues.entry(key).or_default().push(payload);
114    Ok(())
115}
116
117#[cfg(test)]
118pub(crate) fn loopback_smtp_messages_for(source: &ServerAddress) -> Result<Vec<Vec<u8>>> {
119    let ServerAddress::Smtp { address } = source else {
120        return Err(Error::Eval(
121            "source does not use the loopback smtp runtime".to_owned(),
122        ));
123    };
124    let key = format!("smtp:{address}");
125    let messages = loopback_smtp_messages()
126        .lock()
127        .map_err(|_| Error::PoisonedLock("loopback smtp messages"))?;
128    Ok(messages.get(&key).cloned().unwrap_or_default())
129}
130
131impl Object for TriggerHandle {
132    fn display(&self, _cx: &mut Cx) -> Result<String> {
133        Ok("#<server-trigger>".to_owned())
134    }
135
136    fn as_any(&self) -> &dyn std::any::Any {
137        self
138    }
139}
140
141impl sim_kernel::ObjectCompat for TriggerHandle {
142    fn class(&self, cx: &mut Cx) -> Result<ClassRef> {
143        cx.factory().class_stub(
144            sim_kernel::ClassId(0),
145            Symbol::qualified("server", "Trigger"),
146        )
147    }
148    fn as_expr(&self, cx: &mut Cx) -> Result<Expr> {
149        self.reflect_value(cx)?.object().as_expr(cx)
150    }
151    fn as_table(&self, cx: &mut Cx) -> Result<Value> {
152        self.reflect_value(cx)
153    }
154}
155
156pub(crate) fn register_trigger(
157    cx: &mut Cx,
158    server: Arc<Server>,
159    source_expr: Expr,
160    decode_expr: Expr,
161    role: Option<Symbol>,
162    codec: Symbol,
163) -> Result<Arc<TriggerHandle>> {
164    let source_expr = literal_expr(&source_expr).clone();
165    let source = ServerAddress::from_expr(&source_expr)?;
166    require_source_capability(cx, &source)?;
167    let cron = match &source {
168        ServerAddress::Cron { spec } => Some(CronMatcher::parse(spec)?),
169        _ => None,
170    };
171    let network_source = build_network_source(&source, &source_expr)?;
172
173    ensure_installed_codec(cx, &codec)?;
174    let decoder = build_decoder(cx, &decode_expr)?;
175    let trigger = Arc::new(TriggerHandle::new(
176        server.clone(),
177        TriggerConfig {
178            source,
179            source_expr,
180            role,
181            codec,
182            decode_expr,
183            decoder,
184            cron,
185            network_source,
186        },
187    ));
188    server.register_trigger(trigger.clone())?;
189    trigger.start(cx)?;
190    Ok(trigger)
191}
192
193fn build_decoder(cx: &mut Cx, decode_expr: &Expr) -> Result<TriggerDecoder> {
194    let expr = literal_expr(decode_expr).clone();
195    if let Expr::Symbol(symbol) = &expr
196        && let Some(codec) = normalize_codec_symbol(cx, symbol)
197    {
198        ensure_installed_codec(cx, &codec)?;
199        return Ok(TriggerDecoder::Codec(codec));
200    }
201    let callable = cx.eval_expr(expr)?;
202    if callable.object().as_callable().is_none() {
203        return Err(Error::TypeMismatch {
204            expected: "callable or codec symbol",
205            found: "non-callable",
206        });
207    }
208    Ok(TriggerDecoder::Callable(callable))
209}
210
211fn normalize_codec_symbol(cx: &Cx, symbol: &Symbol) -> Option<Symbol> {
212    if cx.registry().codec_by_symbol(symbol).is_some() {
213        return Some(symbol.clone());
214    }
215    let qualified = Symbol::qualified("codec", symbol.name.to_string());
216    if cx.registry().codec_by_symbol(&qualified).is_some() {
217        return Some(qualified);
218    }
219    None
220}
221
222fn literal_expr(expr: &Expr) -> &Expr {
223    match expr {
224        Expr::Quote { expr, .. } => expr,
225        _ => expr,
226    }
227}
228
229fn trigger_read_policy(cx: &Cx) -> ReadPolicy {
230    let mut capabilities = sim_kernel::CapabilitySet::new();
231    if cx.require(&read_construct_capability()).is_ok() {
232        capabilities.insert(read_construct_capability());
233    }
234    if cx.require(&read_eval_capability()).is_ok() {
235        capabilities.insert(read_eval_capability());
236    }
237    ReadPolicy {
238        trust: TrustLevel::TrustedSource,
239        capabilities,
240    }
241}
242
243fn io_error_to_host(err: std::io::Error) -> Error {
244    Error::host_io(err)
245}
246
247fn delivery_timeout() -> Duration {
248    Duration::from_millis(250)
249}
250
251fn source_loopback_enabled(expr: &Expr) -> bool {
252    source_loopback(expr).unwrap_or(false)
253}
sim_lib_server/trigger.rs

sim_lib_server/
trigger.rs
