feat: init

.gitignore

@@ -0,0 +1 @@
+/target

Cargo.toml

@@ -0,0 +1,14 @@
+[package]
+name = "ntfymhe"
+version = "0.1.0"
+edition = "2024"
+
+[dependencies]
+chrono = "0.4.45"
+html-escape = "0.2.14"
+reqwest = { version = "0.13.4", features = ["zstd", "deflate", "brotli", "gzip"] }
+rocket = "0.5.1"
+serde = { version = "1.0.229", features = ["derive"] }
+serde_json = "1.0.151"
+tokio = { version = "1.53.1", features = ["rt", "macros", "fs", "process"] }
+urlencoding = "2.1.3"

src/data.rs

@@ -0,0 +1,114 @@
+use std::{
+    collections::{HashMap, VecDeque},
+    sync::{Arc, atomic::AtomicBool},
+};
+
+use reqwest::Url;
+use tokio::{sync::Mutex, time::Instant};
+
+use crate::{
+    Config,
+    ntfy::{Message, ntfy_task},
+};
+
+pub struct Data {
+    pub subscriptions: HashMap<String, Arc<Subscription>>,
+}
+pub struct Subscription {
+    /// true while a task is running which is
+    /// handling this subscription
+    pub running: AtomicBool,
+    /// true if there is currently no error
+    pub connected: AtomicBool,
+    /// true to request shutdown of any task
+    /// handling this subscription
+    pub shutdown: AtomicBool,
+
+    pub url: Url,
+    pub user: Option<String>,
+    pub pass: Option<String>,
+
+    pub shell: (String, Vec<String>),
+    pub onmsg: Option<String>,
+
+    pub history: Mutex<VecDeque<(Instant, Message)>>,
+    pub errors: Mutex<VecDeque<(Vec<Instant>, String)>>,
+}
+
+impl Data {
+    pub fn new((_config, subscriptions): Config) -> Self {
+        Self {
+            subscriptions: subscriptions
+                .iter()
+                .filter_map(|(name, options)| {
+                    let mut url = options.get("url")?.to_owned();
+                    if !url.ends_with('/') {
+                        url.push('/');
+                    }
+                    url.push_str("json");
+                    let url = Url::parse(&url).ok()?;
+
+                    let shell = if let Some(shell) = options.get("shell") {
+                        (
+                            shell.to_owned(),
+                            options
+                                .get("args")
+                                .map(|v| {
+                                    v.split(char::is_whitespace)
+                                        .filter(|v| !v.is_empty())
+                                        .map(|v| v.to_owned())
+                                        .collect()
+                                })
+                                .unwrap_or_default(),
+                        )
+                    } else {
+                        (
+                            "/usr/bin/env".to_owned(),
+                            vec!["sh".to_owned(), "-c".to_owned()],
+                        )
+                    };
+
+                    let subscription = Arc::new(Subscription {
+                        running: AtomicBool::new(false),
+                        connected: AtomicBool::new(false),
+                        shutdown: AtomicBool::new(false),
+
+                        url,
+                        user: options.get("user").cloned(),
+                        pass: options.get("pass").cloned(),
+
+                        shell,
+                        onmsg: options.get("onmsg").cloned(),
+
+                        history: Mutex::new(VecDeque::new()),
+                        errors: Mutex::new(VecDeque::new()),
+                    });
+                    ntfy_task(&subscription);
+                    Some((name.to_owned(), subscription))
+                })
+                .collect(),
+        }
+    }
+}
+
+impl Subscription {
+    /// Locks `self.errors`.
+    pub async fn push_error(&self, e: String) {
+        let mut errors = self.errors.lock().await;
+        if let Some(i) = errors.iter().position(|(_, e2)| *e2 == e)
+            && let Some((mut times, _)) = errors.remove(i)
+        {
+            if times.len() >= 50 {
+                times[49] = Instant::now();
+            } else {
+                times.push(Instant::now());
+            }
+            errors.push_back((times, e));
+        } else {
+            while errors.len() > 50 {
+                errors.pop_front();
+            }
+            errors.push_back((vec![Instant::now()], e));
+        }
+    }
+}

src/index.html

@@ -0,0 +1,28 @@
+<!DOCTYPE html>
+<html>
+<head>
+<meta charset="utf-8">
+<meta http-equiv="refresh" content="10">
+<meta name="color-scheme" content="dark">
+<title>ntfymhe</title>
+<style>
+.subs {
+  display: grid;
+  grid-template-columns: 3fr 2fr;
+}
+.stopped {
+  background: transparent;
+}
+.connected {
+  background: #0F03;
+}
+.disconnected {
+  background: #F003;
+}
+</style>
+</head>
+<body>
+<h1>ntfy subscriptions</h1>
+<div class="subs">&SUBS;</div>
+</body>
+</html>

src/ini.rs

@@ -0,0 +1,259 @@
+use std::{collections::HashMap, hash::Hash};
+
+#[allow(dead_code)]
+pub trait Ini<'a>: Sized {
+    fn parse_ini(mut src: &'a str) -> Result<Self, String> {
+        let copy = src;
+        Self::deserialize(&mut src).map_err(|e| {
+            let i = copy.len() - src.len();
+            let line = copy[..i].lines().count();
+            format!("line {line}: {e}")
+        })
+    }
+    fn write_ini(&self) -> String {
+        let mut out = String::new();
+        self.serialize(&mut out);
+        out
+    }
+
+    fn serialize(&self, out: &mut String);
+    fn deserialize(src: &mut &'a str) -> Result<Self, String>;
+}
+pub trait IniValue<'a>: Sized {
+    fn serialize(&self, out: &mut String, terminators: &[char]);
+    fn deserialize(src: &mut &'a str, terminators: &[char]) -> Result<Self, String>;
+}
+
+impl<'a> IniValue<'a> for &'a str {
+    fn serialize(&self, out: &mut String, terminators: &[char]) {
+        let should_quote = out.contains(terminators)
+            || out.contains(['[', ']'])
+            || out.starts_with(char::is_whitespace)
+            || out.ends_with(char::is_whitespace);
+        let specials = out.contains(['\0', '\r', '\t']);
+        let single_quotes = out.contains('\'');
+        if should_quote || specials || single_quotes || out.contains('"') {
+            if specials || single_quotes {
+                // either contains weird characters or ',
+                // so we use " and escape relevant characters.
+                out.push('"');
+                out.push_str(
+                    &self
+                        .replace('\\', "\\\\")
+                        .replace('\0', "\\0")
+                        .replace('\r', "\\r")
+                        .replace('\t', "\\t")
+                        .replace('"', "\\\""),
+                );
+                out.push('"');
+            } else {
+                // no weird characters or ',
+                // use ' without escaping anything.
+                out.push('\'');
+                out.push_str(&self);
+                out.push('\'');
+            }
+        } else {
+            out.push_str(self);
+        }
+    }
+
+    fn deserialize(src: &mut &'a str, terminators: &[char]) -> Result<Self, String> {
+        *src = src.trim_start();
+        if src.starts_with('\'') {
+            *src = &src[1..];
+            if let Some(i) = src.find('\'') {
+                let out = &src[0..i];
+                *src = &src[i + 1..];
+                Ok(out)
+            } else {
+                Err("missing a closing single quote")?
+            }
+        } else if src.starts_with('"') {
+            *src = &src[1..];
+            if let Some(i) = src.find('"') {
+                let out = &src[0..i];
+                if out.contains('\\') {
+                    Err(
+                        "backslashes, even if double-quoted, are not supported by the deserializer (caller should deserialize to String instead of &str)",
+                    )?;
+                }
+                *src = &src[i + 1..];
+                Ok(out)
+            } else {
+                Err("missing a closing double quote")?
+            }
+        } else {
+            Ok(if let Some(i) = src.find(terminators) {
+                let out = src[0..i].trim_end();
+                *src = &src[i..];
+                out
+            } else {
+                let out = src.trim_end();
+                *src = "";
+                out
+            })
+        }
+    }
+}
+
+impl<'a> IniValue<'a> for String {
+    fn serialize(&self, out: &mut String, terminators: &[char]) {
+        self.as_str().serialize(out, terminators)
+    }
+
+    fn deserialize(src: &mut &'a str, terminators: &[char]) -> Result<Self, String> {
+        *src = src.trim_start();
+        if src.starts_with('\'') {
+            *src = &src[1..];
+            if let Some(i) = src.find('\'') {
+                let out = &src[0..i];
+                *src = &src[i + 1..];
+                Ok(out.to_owned())
+            } else {
+                Err("missing a closing single quote")?
+            }
+        } else if src.starts_with('"') {
+            *src = &src[1..];
+            let mut out = String::new();
+            loop {
+                if let Some(ch) = src.chars().next() {
+                    *src = &src[ch.len_utf8()..];
+                    if ch == '"' {
+                        break;
+                    } else if ch != '\\' {
+                        out.push(ch);
+                    } else {
+                        let ch = src
+                            .chars()
+                            .next()
+                            .ok_or("EOF after backslash in double-quoted string")?;
+                        *src = &src[ch.len_utf8()..];
+                        out.push(match ch {
+                            '0' => '\0',
+                            'r' => '\r',
+                            'n' => '\n',
+                            't' => '\t',
+                            ch => ch,
+                        });
+                    }
+                } else {
+                    Err("missing a closing double quote")?
+                }
+            }
+            Ok(out)
+        } else {
+            Ok(if let Some(i) = src.find(terminators) {
+                let out = src[0..i].trim_end();
+                *src = &src[i..];
+                out.to_owned()
+            } else {
+                let out = src.trim_end();
+                *src = "";
+                out.to_owned()
+            })
+        }
+    }
+}
+
+trait IniMap<'a>: Sized {
+    type K;
+    type V;
+    fn new() -> Self;
+    fn insert(&mut self, k: Self::K, v: Self::V);
+    fn iter<'b>(&'b self) -> impl Iterator<Item = (&'b Self::K, &'b Self::V)>
+    where
+        Self::K: 'b,
+        Self::V: 'b;
+}
+impl<'a, K: Eq + Hash, V> IniMap<'a> for HashMap<K, V> {
+    type K = K;
+    type V = V;
+    fn new() -> Self {
+        Self::new()
+    }
+    fn insert(&mut self, k: K, v: V) {
+        self.insert(k, v);
+    }
+    fn iter<'b>(&'b self) -> impl Iterator<Item = (&'b K, &'b V)>
+    where
+        K: 'b,
+        V: 'b,
+    {
+        self.iter()
+    }
+}
+
+impl<'a, I, M> Ini<'a> for (I, M)
+where
+    I: Ini<'a>,
+    M: IniMap<'a>,
+    M::K: IniValue<'a>,
+    M::V: Ini<'a>,
+{
+    fn serialize(&self, out: &mut String) {
+        self.0.serialize(out);
+        for (name, section) in self.1.iter() {
+            out.push_str("\n[");
+            name.serialize(out, &['\n', ']']);
+            out.push_str("]\n");
+            section.serialize(out);
+        }
+    }
+    fn deserialize(src: &mut &'a str) -> Result<Self, String> {
+        let head = I::deserialize(src)?;
+        let mut body = M::new();
+        loop {
+            *src = src.trim_start();
+            if src.is_empty() {
+                break;
+            } else if src.starts_with('[') {
+                *src = &src[1..];
+                let name = M::K::deserialize(src, &['\n', ']'])?;
+                *src = src.trim_start();
+                if !src.starts_with(']') {
+                    Err("found `[` but no corresponding `]`")?;
+                }
+                *src = &src[1..];
+                body.insert(name, M::V::deserialize(src)?);
+            } else {
+                Err("expected a new section, but found no `[`")?;
+            }
+        }
+        Ok((head, body))
+    }
+}
+
+impl<'a, M> Ini<'a> for M
+where
+    M: IniMap<'a>,
+    M::K: IniValue<'a>,
+    M::V: IniValue<'a>,
+{
+    fn serialize(&self, out: &mut String) {
+        for (k, v) in self.iter() {
+            k.serialize(out, &['\n', '=']);
+            out.push_str(" = ");
+            v.serialize(out, &['\n']);
+            out.push('\n');
+        }
+    }
+    fn deserialize(src: &mut &'a str) -> Result<Self, String> {
+        let mut out = M::new();
+        loop {
+            *src = src.trim_start();
+            if src.is_empty() || src.starts_with('[') {
+                break;
+            }
+            let k = M::K::deserialize(src, &['\n', '='])?;
+            *src = src.trim_start();
+            if !src.starts_with('=') {
+                Err("missing `=` after key")?;
+            }
+            *src = &src[1..];
+            let v = M::V::deserialize(src, &['\n'])?;
+            out.insert(k, v);
+        }
+        Ok(out)
+    }
+}

src/main.rs

@@ -0,0 +1,278 @@
+mod data;
+mod ini;
+mod ntfy;
+
+use std::sync::atomic::Ordering;
+use std::time::Duration;
+use std::{collections::HashMap, error::Error, sync::Arc};
+
+use chrono::{Datelike, Local, Timelike};
+use html_escape::{encode_safe, encode_safe_to_string};
+use rocket::config::LogLevel;
+use rocket::form::Form;
+use rocket::http::Status;
+use rocket::{FromForm, post};
+use rocket::{get, response::content::RawHtml, routes};
+use tokio::sync::Mutex;
+use tokio::time::Instant;
+
+use crate::data::Data;
+use crate::ini::Ini;
+use crate::ntfy::ntfy_task;
+
+type State = rocket::State<Arc<Mutex<Data>>>;
+type Config = (
+    HashMap<String, String>,
+    HashMap<String, HashMap<String, String>>,
+);
+
+/// ntfy.sh client which reconnects without notification spam.
+///
+/// reads config from `$XDG_CONFIG_HOME/ntfymhe/config.ini`:
+///
+/// ```ini
+/// addr = 127.0.0.1
+/// port = 8000
+/// log = 2
+///
+/// [my subscription]
+/// url = https://example.com/topic
+/// user = my name
+/// pass = "abcd"
+/// onmsg = 'notify-send "$title: $message"'
+///
+/// [other subscription]
+/// ...
+/// ```
+///
+/// All fields except `url` are optional.
+/// The `onmsg` script is passed to a shell, by default as `/usr/bin/env sh -c "$onmsg"`.
+/// You can change the shell by setting e.g. `shell = /bin/bash` and `args = -c`.
+#[tokio::main(flavor = "local")]
+async fn main() -> Result<(), Box<dyn Error>> {
+    let config_home =
+        std::env::var("XDG_CONFIG_HOME").expect("env var XDG_CONFIG_HOME is required");
+    let (cfg, subs) = Config::parse_ini(
+        &tokio::fs::read_to_string(format!("{config_home}/ntfymhe/config.ini"))
+            .await
+            .unwrap(),
+    )
+    .unwrap();
+
+    let mut rocket = rocket::Config::release_default();
+    rocket.log_level = [
+        LogLevel::Off,
+        LogLevel::Critical,
+        LogLevel::Normal,
+        LogLevel::Debug,
+    ]
+    .get(cfg.get("log").and_then(|v| v.parse().ok()).unwrap_or(0))
+    .copied()
+    .unwrap_or(LogLevel::Off);
+    if let Some(v) = cfg.get("addr")
+        && let Ok(v) = v.parse()
+    {
+        rocket.address = v;
+    }
+    if let Some(v) = cfg.get("port")
+        && let Ok(v) = v.parse()
+    {
+        rocket.port = v;
+    }
+    rocket::build()
+        .configure(rocket)
+        .manage(Arc::new(Mutex::new(Data::new((cfg, subs)))))
+        .mount("/", routes![index, subscription, subscription_post])
+        .launch()
+        .await?;
+    Ok(())
+}
+
+#[get("/")]
+async fn index(state: &State) -> RawHtml<String> {
+    let state = state.lock().await;
+    let subs = if state.subscriptions.is_empty() {
+        "<div><em>you have no subscriptions</em></div><div></div>".to_owned()
+    } else {
+        let mut subs = String::new();
+        for (url, subscription) in state.subscriptions.iter() {
+            subs.push_str(r#"<div><a href=""#);
+            subs.push_str(&urlencoding::encode(url.as_str()));
+            subs.push_str(r#"">"#);
+            encode_safe_to_string(url.as_str(), &mut subs);
+            subs.push_str(r#"</a></div><div class=""#);
+            let running = subscription.running.load(Ordering::Relaxed);
+            let connected = subscription.connected.load(Ordering::Relaxed);
+            if !running {
+                subs.push_str("stopped");
+            } else if connected {
+                subs.push_str("connected");
+            } else {
+                subs.push_str("disconnected");
+            }
+            subs.push_str(r#"">"#);
+            if !running {
+                subs.push_str("stopped");
+            } else if connected {
+                subs.push_str("connected");
+            } else {
+                subs.push_str("disconnected");
+            }
+            subs.push_str("</div>");
+        }
+        subs
+    };
+    RawHtml(include_str!("index.html").replace("&SUBS;", &subs))
+}
+
+#[derive(FromForm)]
+struct Post {
+    run: Option<u8>,
+}
+#[post("/<sub>", data = "<post>")]
+async fn subscription_post(sub: &str, post: Form<Post>, state: &State) -> Status {
+    let Post { run } = post.into_inner();
+    let state = state.lock().await;
+    if let Some(subscription) = state.subscriptions.get(sub) {
+        match run {
+            Some(0) => subscription.shutdown.store(true, Ordering::Relaxed),
+            Some(1) => {
+                subscription.shutdown.store(false, Ordering::Relaxed);
+                ntfy_task(subscription);
+            }
+            _ => return Status::BadRequest,
+        }
+        Status::Ok
+    } else {
+        Status::NotFound
+    }
+}
+
+#[get("/<sub>?<errs>")]
+async fn subscription(sub: &str, errs: bool, state: &State) -> Result<RawHtml<String>, Status> {
+    let state = state.lock().await;
+    if let Some(subscription) = state.subscriptions.get(sub) {
+        let running = subscription.running.load(Ordering::Relaxed);
+        let connected = subscription.connected.load(Ordering::Relaxed);
+        let shutdown = subscription.shutdown.load(Ordering::Relaxed);
+        let form = |skip: &[&str], html: &str| {
+            let mut form = r#"<form action="" method="get">"#.to_owned();
+            if errs && !skip.contains(&"errs") {
+                form.push_str(r#"<input type="hidden" name="errs" value="">"#);
+            }
+            form.push_str(html);
+            form.push_str("</form>");
+            form
+        };
+
+        let now = Instant::now();
+        let mut messages = String::new();
+        let mut errors = if running {
+            r#"<a href=".">back</a> | <form action="" method="post" target="formframe"><button type="submit" name="run" value="0">stop</button></form> "#.to_owned()
+        } else {
+            r#"<a href=".">back</a> | <form action="" method="post" target="formframe"><button type="submit" name="run" value="1">start</button></form> "#.to_owned()
+        };
+        if errs {
+            errors.push_str(&form(
+                &["errs"],
+                r#"<button type="submit">hide errors</button>"#,
+            ));
+            errors.push_str(r#"<div class="errs">"#);
+            for (recv, error) in subscription.errors.lock().await.iter().rev() {
+                let ago = now.saturating_duration_since(*recv.last().unwrap());
+                errors.push_str("<div><i>");
+                fmt_ago(ago, &mut errors);
+                errors.push_str("</i><div>");
+                encode_safe_to_string(error, &mut errors);
+                errors.push_str("</div></div>");
+            }
+            errors.push_str("</div>");
+        } else {
+            errors.push_str(&form(
+                &["errs"],
+                r#"<button type="submit" name="errs" value="">show errors</button>"#,
+            ));
+        }
+        for (recv, message) in subscription.history.lock().await.iter().rev() {
+            let ago = now.saturating_duration_since(*recv);
+            messages.push_str(r#"<div class="msg"><i>"#);
+            if let Some(time) = message.time
+                && let Some(time) = chrono::DateTime::from_timestamp_secs(time)
+            {
+                let time = time.with_timezone(&Local).naive_local();
+                messages.push_str(&format!(
+                    "{:0>2}:{:0>2}:{:0>2} <small>{}-{}-{}, ",
+                    time.hour(),
+                    time.minute(),
+                    time.second(),
+                    time.year(),
+                    time.month(),
+                    time.day(),
+                ));
+            } else {
+                messages.push_str("<small>");
+            }
+            messages.push_str("recv ");
+            fmt_ago(ago, &mut messages);
+            messages.push_str("</small></i><br>");
+            match (&message.title, &message.message) {
+                (None, None) => {}
+                (Some(msg), None) | (None, Some(msg)) => {
+                    messages.push_str(r#"<b>"#);
+                    encode_safe_to_string(msg, &mut messages);
+                    messages.push_str("</b>");
+                }
+                (Some(title), Some(content)) => {
+                    messages.push_str(r#"<b>"#);
+                    encode_safe_to_string(title, &mut messages);
+                    messages.push_str("</b><br>");
+                    encode_safe_to_string(content, &mut messages);
+                }
+            }
+            messages.push_str("</div>");
+        }
+        Ok(RawHtml(
+            include_str!("subscription.html")
+                .replace("&NAME;", &encode_safe(sub))
+                .replace(
+                    "&STATE;",
+                    if !running {
+                        "stopped"
+                    } else if shutdown {
+                        "stopping"
+                    } else if connected {
+                        "connected"
+                    } else {
+                        "disconnected"
+                    },
+                )
+                .replace("&MSGS;", &messages)
+                .replace("&ERRS;", &errors),
+        ))
+    } else {
+        Err(Status::NotFound)
+    }
+}
+
+fn fmt_ago(ago: Duration, html: &mut String) {
+    let secs = ago.as_secs();
+    fn s(n: u64) -> &'static str {
+        if n == 1 { "" } else { "s" }
+    }
+    if secs < 60 {
+        html.push_str(&format!("{} second{} ago", secs, s(secs)));
+    } else {
+        let mins = secs / 60;
+        if mins < 60 {
+            html.push_str(&format!("{} minute{} ago", mins, s(mins)));
+        } else {
+            let hours = mins / 60;
+            if hours < 48 {
+                html.push_str(&format!("{} hour{} ago", hours, s(hours)));
+            } else {
+                let days = hours / 24;
+                html.push_str(&format!("{days} days ago"));
+            }
+        }
+    }
+}

src/ntfy.rs

@@ -0,0 +1,185 @@
+use std::{
+    sync::{Arc, atomic::Ordering},
+    time::Duration,
+};
+
+use serde::Deserialize;
+use tokio::{process::Command, task::JoinHandle, time::Instant};
+
+use crate::data::Subscription;
+
+pub fn ntfy_task(subscription: &Arc<Subscription>) -> Option<JoinHandle<()>> {
+    if subscription
+        .running
+        .compare_exchange(false, true, Ordering::Relaxed, Ordering::Relaxed)
+        .is_err()
+    {
+        return None;
+    }
+    let subscription = Arc::clone(subscription);
+    Some(tokio::task::spawn(async move {
+        let mut backoff = 1;
+        loop {
+            if subscription.shutdown.load(Ordering::Relaxed) {
+                break;
+            }
+            if let Err(e) = handle(&subscription).await {
+                subscription.connected.store(false, Ordering::Relaxed);
+                subscription.push_error(e).await;
+                backoff = (backoff * 2).min(10 * 60);
+            } else {
+                backoff = 1;
+            }
+            tokio::time::sleep(Duration::from_secs(backoff)).await;
+        }
+        subscription.running.store(false, Ordering::Relaxed);
+    }))
+}
+
+async fn handle(subscription: &Subscription) -> Result<(), String> {
+    let mut url = subscription.url.clone();
+    if let Some(id) = subscription
+        .history
+        .lock()
+        .await
+        .back()
+        .and_then(|(_, msg)| msg.id.as_ref())
+    {
+        url.set_query(Some(&format!("since={}", urlencoding::encode(id))));
+    } else {
+        url.set_query(Some("since=all"));
+    }
+    let mut request = reqwest::Client::new().get(url);
+    if let Some(user) = &subscription.user {
+        request = request.basic_auth(user, subscription.pass.as_ref());
+    }
+    let mut response = request
+        .send()
+        .await
+        .map_err(|e| format!("couldn't connect: {e}"))?
+        .error_for_status()
+        .map_err(|e| format!("non-ok response: {e}"))?;
+    subscription.connected.store(true, Ordering::Relaxed);
+
+    let dur = Duration::from_secs(5 * 61);
+    let mut buf = Vec::new();
+    loop {
+        if subscription.shutdown.load(Ordering::Relaxed) {
+            break;
+        }
+        if let Some(chunk) = tokio::time::timeout(dur, response.chunk())
+            .await
+            .ok()
+            .and_then(|v| v.ok())
+            .flatten()
+        {
+            if let (Some(mut i), Some(j)) = (
+                chunk[..].iter().position(|v| *v == b'\n'),
+                chunk[..].iter().rposition(|v| *v == b'\n'),
+            ) {
+                buf.extend(&chunk[0..i]);
+                if i < j {
+                    i += 1;
+                }
+                for line in std::iter::once(&buf[..]).chain(chunk[i..j].split(|v| *v == b'\n')) {
+                    if let Ok(str) = str::from_utf8(line)
+                        && let str = str.trim()
+                        && !str.is_empty()
+                    {
+                        for_line(str, subscription).await?;
+                    }
+                }
+                buf = chunk[j + 1..].to_vec();
+            } else {
+                buf.extend(chunk);
+            }
+        } else {
+            break;
+        }
+    }
+    Ok(())
+}
+
+#[derive(Deserialize, Debug)]
+#[serde(tag = "event")]
+enum Event {
+    #[serde(rename = "open")]
+    Open,
+    #[serde(rename = "keepalive")]
+    Keepalive,
+    #[serde(rename = "message")]
+    Message(Message),
+    #[serde(rename = "message_delete")]
+    MessageDelete,
+    #[serde(rename = "message_clear")]
+    MessageClear,
+    #[serde(rename = "poll_request")]
+    PollRequest,
+}
+#[derive(Deserialize, Debug)]
+pub struct Message {
+    pub id: Option<String>,
+    pub time: Option<i64>,
+    pub topic: Option<String>,
+    pub title: Option<String>,
+    pub message: Option<String>,
+    #[serde(default, skip)]
+    pub raw: String,
+}
+
+async fn for_line(line: &str, subscription: &Subscription) -> Result<(), String> {
+    let event =
+        serde_json::from_str::<Event>(line).map_err(|e| format!("unparsable json message: {e}"))?;
+    match event {
+        Event::Message(mut message) => {
+            message.raw = line.to_owned();
+            let mut history = subscription.history.lock().await;
+            let run_onmsg = message.id.as_ref().is_none_or(|id| {
+                history
+                    .iter()
+                    .all(|(_, msg)| msg.id.as_ref().is_none_or(|old| id != old))
+            });
+            fn forenv(v: &Option<String>) -> &str {
+                v.as_ref().map(|v| v.as_str()).unwrap_or("")
+            }
+            if run_onmsg
+                && let Some(cmd) = &subscription.onmsg
+                && let time = message.time.map(|v| v.to_string())
+                && let Err(e) = Command::new(&subscription.shell.0)
+                    .args(&subscription.shell.1)
+                    .arg(cmd)
+                    .envs([
+                        ("NTFY_ID", forenv(&message.id)),
+                        ("id", forenv(&message.id)),
+                        ("NTFY_TIME", forenv(&time)),
+                        ("time", forenv(&time)),
+                        ("NTFY_TOPIC", forenv(&message.topic)),
+                        ("topic", forenv(&message.topic)),
+                        ("NTFY_MESSAGE", forenv(&message.message)),
+                        ("message", forenv(&message.message)),
+                        ("m", forenv(&message.message)),
+                        ("NTFY_TITLE", forenv(&message.title)),
+                        ("title", forenv(&message.title)),
+                        ("t", forenv(&message.title)),
+                        ("NTFY_RAW", message.raw.as_str()),
+                        ("raw", message.raw.as_str()),
+                    ])
+                    .spawn()
+            {
+                subscription
+                    .push_error(format!("error running onmsg command: {e}"))
+                    .await;
+            }
+            while history.len() > 200 {
+                history.pop_front();
+            }
+            history.push_back((Instant::now(), message));
+        }
+        Event::Open
+        | Event::Keepalive
+        | Event::MessageDelete
+        | Event::MessageClear
+        | Event::PollRequest => {}
+    }
+    Ok(())
+}

src/subscription.html

@@ -0,0 +1,51 @@
+<!DOCTYPE html>
+<html>
+<head>
+<meta charset="utf-8">
+<meta http-equiv="refresh" content="5">
+<meta name="color-scheme" content="dark">
+<title>&NAME;</title>
+<style>
+form {
+  display: inline;
+}
+.stopped {
+  background: transparent;
+}
+.stopping {
+  background: #00F2;
+}
+.connected {
+  background: #0F03;
+}
+.disconnected {
+  background: #F003;
+}
+.msgs {
+  margin-top: 2em;
+}
+.msg {
+  border-bottom: 1pt solid #FFF6;
+  margin: 0;
+}
+.msg i {
+  font-size: smaller;
+  color: #DDD;
+}
+.errs i {
+  font-size: smaller;
+  color: #FCC;
+}
+#formframe {
+  display: none;
+}
+</style>
+</head>
+<body>
+<h2 class="&STATE;">&NAME;</h2>
+&ERRS;
+<div class="msgs">&MSGS;</div>
+<iframe id="formframe" name="formframe"></iframe>
+</body>
+</html>
+