harness_loop/scheduler/
channel.rs

1//! Delivery channels. A `Channel` takes a job's agent output and sends it
2//! somewhere. Built-ins: `StdoutChannel`, `EmailChannel` (Resend). Apps add
3//! their own (Telegram, Slack, …) by implementing `Channel`.
4
5use crate::scheduler::store::Job;
6use async_trait::async_trait;
7use std::collections::HashMap;
8use std::sync::Arc;
9
10#[derive(Debug, thiserror::Error)]
11#[non_exhaustive]
12pub enum ChannelError {
13    #[error("channel send: {0}")]
14    Send(String),
15}
16
17#[async_trait]
18pub trait Channel: Send + Sync {
19    /// Stable key matched against `Job.channel`.
20    fn key(&self) -> &str;
21    /// Deliver `output` for `job`. `job.target` is the recipient if relevant.
22    async fn send(&self, output: &str, job: &Job) -> Result<(), ChannelError>;
23}
24
25/// Prints the output to stdout.
26pub struct StdoutChannel;
27impl StdoutChannel {
28    pub fn new() -> Self {
29        Self
30    }
31}
32impl Default for StdoutChannel {
33    fn default() -> Self {
34        Self::new()
35    }
36}
37
38#[async_trait]
39impl Channel for StdoutChannel {
40    fn key(&self) -> &str {
41        "stdout"
42    }
43    async fn send(&self, output: &str, job: &Job) -> Result<(), ChannelError> {
44        println!("\n=== {} ===\n{}\n", job.name, output);
45        Ok(())
46    }
47}
48
49/// Sends the output as an email via Resend. Recipient = `job.target`.
50pub struct EmailChannel {
51    api_key: String,
52    from: String,
53    client: reqwest::Client,
54}
55
56/// Build the Resend POST body. Pure — unit-tested without network.
57pub fn resend_body(from: &str, to: &str, subject: &str, text: &str) -> serde_json::Value {
58    serde_json::json!({ "from": from, "to": [to], "subject": subject, "text": text })
59}
60
61impl EmailChannel {
62    pub fn new(api_key: impl Into<String>, from: impl Into<String>) -> Self {
63        Self {
64            api_key: api_key.into(),
65            from: from.into(),
66            client: reqwest::Client::new(),
67        }
68    }
69    /// Construct from env: `RESEND_API_KEY` + `DIGEST_FROM` (falls back to a
70    /// Resend test sender). Returns None if `RESEND_API_KEY` is absent.
71    pub fn from_env() -> Option<Self> {
72        let key = std::env::var("RESEND_API_KEY")
73            .ok()
74            .filter(|k| !k.is_empty())?;
75        let from = std::env::var("DIGEST_FROM")
76            .unwrap_or_else(|_| "Scheduler <onboarding@resend.dev>".into());
77        Some(Self::new(key, from))
78    }
79}
80
81#[async_trait]
82impl Channel for EmailChannel {
83    fn key(&self) -> &str {
84        "email"
85    }
86    async fn send(&self, output: &str, job: &Job) -> Result<(), ChannelError> {
87        let to = job
88            .target
89            .as_deref()
90            .ok_or_else(|| ChannelError::Send("email job has no target recipient".into()))?;
91        let body = resend_body(&self.from, to, &job.name, output);
92        let resp = self
93            .client
94            .post("https://api.resend.com/emails")
95            .bearer_auth(&self.api_key)
96            .json(&body)
97            .timeout(std::time::Duration::from_secs(20))
98            .send()
99            .await
100            .map_err(|e| ChannelError::Send(format!("resend post: {e}")))?;
101        if resp.status().is_success() {
102            Ok(())
103        } else {
104            let code = resp.status();
105            let t = resp.text().await.unwrap_or_default();
106            Err(ChannelError::Send(format!("resend {code}: {t}")))
107        }
108    }
109}
110
111/// Maps channel keys to implementations.
112#[derive(Default, Clone)]
113pub struct ChannelRegistry {
114    map: HashMap<String, Arc<dyn Channel>>,
115}
116
117impl ChannelRegistry {
118    pub fn new() -> Self {
119        Self::default()
120    }
121    pub fn register(&mut self, c: Arc<dyn Channel>) {
122        self.map.insert(c.key().to_string(), c);
123    }
124    pub fn get(&self, key: &str) -> Option<&Arc<dyn Channel>> {
125        self.map.get(key)
126    }
127}
128
129#[cfg(test)]
130mod tests {
131    use super::*;
132
133    #[test]
134    fn resend_body_shape() {
135        let b = resend_body("Me <m@x.com>", "u@y.com", "Daily", "hello");
136        assert_eq!(b["from"], "Me <m@x.com>");
137        assert_eq!(b["to"][0], "u@y.com");
138        assert_eq!(b["subject"], "Daily");
139        assert_eq!(b["text"], "hello");
140    }
141
142    #[tokio::test]
143    async fn email_without_target_errors() {
144        let ch = EmailChannel::new("re_x", "Me <m@x.com>");
145        let job = Job::new("j", "daily 08:00", "p", "email", 1);
146        let err = ch.send("out", &job).await;
147        assert!(err.is_err());
148    }
149
150    #[tokio::test]
151    async fn registry_lookup() {
152        let mut reg = ChannelRegistry::new();
153        reg.register(Arc::new(StdoutChannel::new()));
154        assert!(reg.get("stdout").is_some());
155        assert!(reg.get("email").is_none());
156    }
157}
harness_loop/scheduler/channel.rs

harness_loop/scheduler/
channel.rs
