Split monolithic code into several crates: core, telegram.

Remove http interface, for now.

Cargo.toml

@@ -9,19 +9,18 @@ tokio = { version = "1.40", features = ["full"] }
 log = "0.4.22"
 futures = "0.3.30"
 tracing = "0.1.40"
-tracing-subscriber = "0.3.18"
-axum = { version = "0.7.5", features = ["macros"] }
-teloxide = "0.13.0"
+tracing-subscriber = { version = "0.3.18", features = ["env-filter", "fmt"] }
 serde = { version = "1.0.210", features = ["derive"] }
-utoipa = { version = "5.0.0-beta.0", features = ["axum_extras"] }
-utoipa-redoc = { version = "4.0.1-beta.0", features = ["axum"] }
-sea-orm = { version = "1.1.0-rc.1", features = ["sqlx-sqlite", "runtime-tokio-rustls", "macros"] }
+#sea-orm = { version = "1.1.0-rc.1", features = ["sqlx-sqlite", "runtime-tokio-rustls", "macros"] }
 uuid = { version = "1.10.0", features = ["serde"] }
 serde_json = "1.0.128"
 reqwest = { version = "0.12.7", features = ["json"] }
 async-trait = "0.1.82"
 url = "2.5.2"
-erased-serde = "0.4.5"
+figment = { version = "0.10.19", features = ["yaml", "env"] }
+clap = { version = "4.5.17", features = ["derive", "color", "usage"] }
+notifico-core = { path = "notifico-core" }
+notifico-telegram = { path = "notifico-telegram" }
 
 [workspace]
-members = ["."]
+members = [".", "notifico-core", "notifico-telegram"]

notifico-core/Cargo.toml

@@ -0,0 +1,12 @@
+[package]
+name = "notifico-core"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+async-trait = "0.1.82"
+serde = { version = "1.0.210", features = ["derive"] }
+serde_json = "1.0.128"
+uuid = { version = "1.10.0", features = ["v4", "serde"] }
+reqwest = "0.12.7"
+url = "2.5.2"

notifico-core/src/credentials.rs

@@ -0,0 +1,13 @@
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct Credential {
+    pub r#type: String,
+    pub name: String,
+    pub value: Value,
+}
+
+pub trait Credentials: Send + Sync {
+    fn get_credential(&self, r#type: &str, name: &str) -> Option<Value>;
+}

notifico-core/src/engine.rs

@@ -0,0 +1,38 @@
+use crate::pipeline::SerializedStep;
+use crate::recipient::Recipient;
+use crate::templater::TemplaterError;
+use async_trait::async_trait;
+use serde::{Deserialize, Serialize};
+use serde_json::{Map, Value};
+use std::any::Any;
+use std::borrow::Cow;
+use std::collections::HashMap;
+
+#[derive(Debug, Default, Serialize, Deserialize)]
+#[serde(transparent)]
+pub struct EventContext(pub Map<String, Value>);
+
+#[derive(Default, Debug)]
+pub struct PipelineContext {
+    pub recipient: Option<Recipient>,
+    pub event_context: EventContext,
+    pub plugin_contexts: HashMap<Cow<'static, str>, Value>,
+}
+
+#[derive(Debug)]
+pub enum EngineError {
+    TemplaterError(TemplaterError),
+    PluginNotFound(SerializedStep),
+    PipelineInterrupted,
+}
+
+#[async_trait]
+pub trait EnginePlugin: Send + Sync + Any {
+    async fn execute_step(
+        &self,
+        context: &mut PipelineContext,
+        step: &SerializedStep,
+    ) -> Result<(), EngineError>;
+
+    fn step_type(&self) -> Cow<'static, str>;
+}

notifico-core/src/lib.rs

@@ -0,0 +1,5 @@
+pub mod credentials;
+pub mod engine;
+pub mod pipeline;
+pub mod recipient;
+pub mod templater;

src/engine/pipeline.rs → notifico-core/src/pipeline.rs

@@ -3,15 +3,24 @@ use serde_json::Value;
 
 #[derive(Serialize, Deserialize, Default, Clone, Debug)]
 pub struct Pipeline {
-    pub channel: String,
+    pub events: Vec<String>,
     pub steps: Vec<SerializedStep>,
 }
 
 #[derive(Serialize, Deserialize, Default, Clone, Debug)]
+#[serde(transparent)]
 pub struct SerializedStep(pub serde_json::Map<String, Value>);
 
 impl SerializedStep {
     pub fn get_type(&self) -> &str {
-        &self.0["type"].as_str().expect("Step type must be a string")
+        self.0["step"].as_str().expect("Step type must be a string")
+    }
+
+    pub fn into_inner(self) -> serde_json::Map<String, Value> {
+        self.0
+    }
+
+    pub fn into_value(self) -> Value {
+        Value::Object(self.into_inner())
     }
 }

notifico-core/src/recipient.rs

@@ -0,0 +1,35 @@
+use serde::Deserialize;
+use serde_json::Value;
+use uuid::Uuid;
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct Recipient {
+    pub id: Uuid,
+    pub contacts: Vec<Contact>,
+}
+
+impl Recipient {
+    pub fn get_primary_contact(&self, r#type: &str) -> Option<&Contact> {
+        for contact in &self.contacts {
+            if contact.r#type() == r#type {
+                return Some(&contact);
+            }
+        }
+        None
+    }
+}
+
+#[derive(Clone, Debug, Deserialize)]
+pub struct Contact(Value);
+
+impl Contact {
+    pub fn r#type(&self) -> &str {
+        self.0["type"]
+            .as_str()
+            .expect("Contact type must be a string")
+    }
+
+    pub fn into_json(self) -> Value {
+        self.0
+    }
+}

notifico-core/src/templater.rs

@@ -0,0 +1,36 @@
+use async_trait::async_trait;
+use serde::{Deserialize, Serialize};
+use serde_json::{Map, Value};
+use uuid::Uuid;
+
+#[derive(Debug)]
+pub enum TemplaterError {
+    RequestError(reqwest::Error),
+    UrlError(url::ParseError),
+}
+
+impl From<reqwest::Error> for TemplaterError {
+    fn from(err: reqwest::Error) -> Self {
+        TemplaterError::RequestError(err)
+    }
+}
+
+impl From<url::ParseError> for TemplaterError {
+    fn from(err: url::ParseError) -> Self {
+        TemplaterError::UrlError(err)
+    }
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+#[serde(transparent)]
+pub struct RenderResponse(pub Map<String, Value>);
+
+#[async_trait]
+pub trait Templater: Send + Sync {
+    async fn render(
+        &self,
+        template_type: &str,
+        template_id: Uuid,
+        context: Map<String, Value>,
+    ) -> Result<RenderResponse, TemplaterError>;
+}

notifico-telegram/Cargo.toml

@@ -0,0 +1,14 @@
+[package]
+name = "notifico-telegram"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+async-trait = "0.1.82"
+serde = { version = "1.0.210", features = ["derive"] }
+serde_json = "1.0.128"
+tracing = "0.1.40"
+uuid = { version = "1.10.0", features = ["v4"] }
+notifico-core = { path = "../notifico-core" }
+teloxide = "0.13.0"
+

notifico-telegram/src/contact.rs

@@ -0,0 +1,23 @@
+use notifico_core::recipient::Contact;
+use serde::Deserialize;
+use teloxide::prelude::ChatId;
+use teloxide::types::Recipient;
+
+#[derive(Debug, Deserialize)]
+pub struct TelegramContact {
+    chat_id: ChatId,
+}
+
+impl TelegramContact {
+    pub(crate) fn into_recipient(self) -> Recipient {
+        Recipient::Id(self.chat_id)
+    }
+}
+
+impl TryFrom<Contact> for TelegramContact {
+    type Error = ();
+
+    fn try_from(value: Contact) -> Result<Self, Self::Error> {
+        serde_json::from_value(value.into_json()).map_err(|_| ())
+    }
+}

notifico-telegram/src/lib.rs

@@ -0,0 +1,130 @@
+use async_trait::async_trait;
+use contact::TelegramContact;
+use notifico_core::credentials::Credentials;
+use notifico_core::engine::{EngineError, EnginePlugin, PipelineContext};
+use notifico_core::pipeline::SerializedStep;
+use notifico_core::templater::{RenderResponse, Templater};
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use std::borrow::Cow;
+use std::sync::Arc;
+use step::{CredentialSelector, TelegramStep};
+use teloxide::prelude::Requester;
+use teloxide::Bot;
+use tracing::debug;
+use uuid::Uuid;
+
+mod contact;
+mod step;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct TelegramBotCredentials {
+    token: String,
+}
+
+pub struct TelegramPlugin {
+    templater: Arc<dyn Templater>,
+    credentials: Arc<dyn Credentials>,
+}
+
+impl TelegramPlugin {
+    pub fn new(templater: Arc<dyn Templater>, credentials: Arc<dyn Credentials>) -> Self {
+        Self {
+            templater,
+            credentials,
+        }
+    }
+}
+
+#[derive(Default, Serialize, Deserialize)]
+struct TelegramContext {
+    template_id: Option<Uuid>,
+}
+
+#[async_trait]
+impl EnginePlugin for TelegramPlugin {
+    async fn execute_step(
+        &self,
+        context: &mut PipelineContext,
+        step: &SerializedStep,
+    ) -> Result<(), EngineError> {
+        let telegram_context = context
+            .plugin_contexts
+            .entry("telegram".into())
+            .or_insert(Value::Object(Default::default()));
+
+        debug!("Plugin context: {:?}", telegram_context);
+
+        let mut plugin_context: TelegramContext =
+            serde_json::from_value(telegram_context.clone()).unwrap();
+        let telegram_step: TelegramStep = step.clone().try_into().unwrap();
+
+        match telegram_step {
+            TelegramStep::LoadTemplate { template_id } => {
+                plugin_context.template_id = Some(template_id);
+                context.plugin_contexts.insert(
+                    "telegram".into(),
+                    serde_json::to_value(plugin_context).unwrap(),
+                );
+            }
+            TelegramStep::Send(cred_selector) => {
+                let Some(template_id) = plugin_context.template_id else {
+                    return Err(EngineError::PipelineInterrupted);
+                };
+
+                let bot_token = match cred_selector {
+                    CredentialSelector::BotName { bot_name } => self
+                        .credentials
+                        .get_credential("telegram_token", &bot_name)
+                        .unwrap(),
+                };
+
+                let tgcred: TelegramBotCredentials = serde_json::from_value(bot_token).unwrap();
+
+                let bot = Bot::new(tgcred.token);
+
+                let rendered_template = self
+                    .templater
+                    .render("telegram", template_id, context.event_context.0.clone())
+                    .await
+                    .unwrap();
+
+                let rendered_template: TelegramBody = rendered_template.try_into().unwrap();
+
+                let contact = TelegramContact::try_from(
+                    context
+                        .recipient
+                        .clone()
+                        .unwrap()
+                        .get_primary_contact("telegram")
+                        .ok_or(EngineError::PipelineInterrupted)
+                        .cloned()?,
+                )
+                .unwrap();
+
+                bot.send_message(contact.into_recipient(), rendered_template.body)
+                    .await
+                    .unwrap();
+            }
+        }
+
+        Ok(())
+    }
+
+    fn step_type(&self) -> Cow<'static, str> {
+        "telegram".into()
+    }
+}
+
+#[derive(Deserialize, Clone)]
+pub struct TelegramBody {
+    pub body: String,
+}
+
+impl TryFrom<RenderResponse> for TelegramBody {
+    type Error = ();
+
+    fn try_from(value: RenderResponse) -> Result<Self, Self::Error> {
+        serde_json::from_value(Value::from_iter(value.0)).map_err(|_| ())
+    }
+}

notifico-telegram/src/step.rs

@@ -0,0 +1,30 @@
+use notifico_core::pipeline::SerializedStep;
+use serde::{Deserialize, Serialize};
+use uuid::Uuid;
+
+#[derive(Serialize, Deserialize)]
+#[serde(untagged)]
+pub enum CredentialSelector {
+    BotName { bot_name: String },
+}
+
+#[derive(Serialize, Deserialize)]
+#[serde(tag = "step")]
+pub enum TelegramStep {
+    #[serde(rename = "telegram.load_template")]
+    LoadTemplate { template_id: Uuid },
+    // #[serde(rename = "telegram.set_recipients")]
+    // SetRecipients { telegram_id: Vec<i64> },
+    #[serde(rename = "telegram.send")]
+    Send(CredentialSelector),
+}
+
+impl TryFrom<SerializedStep> for TelegramStep {
+    type Error = ();
+
+    fn try_from(value: SerializedStep) -> Result<Self, Self::Error> {
+        let s = serde_json::to_string(&value.into_value()).unwrap();
+
+        Ok(serde_json::from_str(&s).unwrap())
+    }
+}