Merge pull request #178 from kpcyrd/notify

Add basic notification system

README.md

@@ -164,6 +164,18 @@ For everything else please have a look at the [detailed list][1].
   - [Logging events](https://sn0int.readthedocs.io/en/latest/activity.html#logging-events)
   - [Querying events](https://sn0int.readthedocs.io/en/latest/activity.html#querying-events)
   - [Visualization](https://sn0int.readthedocs.io/en/latest/activity.html#visualization)
+- [Notifications](https://sn0int.readthedocs.io/en/latest/notifications.html)
+  - [Receiving notifications](https://sn0int.readthedocs.io/en/latest/notifications.html#receiving-notifications)
+    - [Telegram](https://sn0int.readthedocs.io/en/latest/notifications.html#telegram)
+    - [Pushover](https://sn0int.readthedocs.io/en/latest/notifications.html#pushover)
+    - [Discord](https://sn0int.readthedocs.io/en/latest/notifications.html#discord)
+    - [Signal](https://sn0int.readthedocs.io/en/latest/notifications.html#signal)
+    - [Writing your own module](https://sn0int.readthedocs.io/en/latest/notifications.html#writing-your-own-module)
+  - [Setting up notification rules](https://sn0int.readthedocs.io/en/latest/notifications.html#setting-up-notification-rules)
+  - [Testing notifications](https://sn0int.readthedocs.io/en/latest/notifications.html#testing-notifications)
+  - [Running sn0int automatically](https://sn0int.readthedocs.io/en/latest/notifications.html#running-sn0int-automatically)
+    - [Monitors](https://sn0int.readthedocs.io/en/latest/notifications.html#monitors)
+    - [Timers](https://sn0int.readthedocs.io/en/latest/notifications.html#timers)
 - [Keyring](https://sn0int.readthedocs.io/en/latest/keyring.html)
   - [Managing the keyring](https://sn0int.readthedocs.io/en/latest/keyring.html#managing-the-keyring)
   - [Using access keys in scripts](https://sn0int.readthedocs.io/en/latest/keyring.html#using-access-keys-in-scripts)

docs/index.rst

@@ -43,6 +43,7 @@ Getting Started
    database
    structs
    activity
+   notifications
    keyring
    config
    sandbox

docs/notifications.rst

@@ -0,0 +1,311 @@
+Notifications
+=============
+
+If you run sn0int unattended nobody might see the sn0int output. For cases like
+this you can configure notifications to send you a push notification in case
+something interesting happens. This is also especially useful if you have
+sn0int setup to run automatically.
+
+Receiving notifications
+-----------------------
+
+Notifications are just regular sn0int modules. You can install them just like
+any other module or write your own. This section contains walkthroughs on how
+to setup common integrations.
+
+Telegram
+~~~~~~~~
+
+Install the telegram notification module from the registry:
+
+.. code-block:: bash
+
+    sn0int pkg install kpcyrd/notify-telegram
+
+Open your telegram app and open a chat with ``@botfather``. Send ``/newbot``
+and answer the questions. Copy ``bot_token`` and open this url in your browser:
+
+.. code-block::
+
+    https://api.telegram.org/bot**your_bot_token**/getUpdates
+
+Back on your app, open the t.me link to start a new chat with your bot, then
+send ``/start``. Reload the page in your browser, you should see the new
+message you sent. Copy the ``chat_id``.
+
+Test your tokens are working correctly by sending yourself a notification:
+
+.. code-block:: bash
+
+    sn0int notify exec kpcyrd/notify-telegram -o bot_token=1337:foobar -o chat_id=1337 'hello world'
+
+You should receive ``hello world`` from your bot on Telegram.
+
+Pushover
+~~~~~~~~
+
+Install the pushover notification module from the registry:
+
+.. code-block:: bash
+
+    sn0int pkg install kpcyrd/notify-pushover
+
+Signup for pushover and configure the app on your device. Copy th user key
+visible on the pushover dashboard. Click "Create an Application/API Token". Set
+"sn0int" as name and set an icon if you want to. Copy the api token.
+
+Test your tokens are working correctly by sending yourself a notification:
+
+.. code-block:: bash
+
+    sn0int notify exec kpcyrd/notify-pushover -o user_key=asdf1337 -o api_token=asdf1337 'hello world'
+
+You should receive ``hello world`` as a push notification.
+
+Discord
+~~~~~~~
+
+Install the discord notification module from the registry:
+
+.. code-block:: bash
+
+    sn0int pkg install kpcyrd/notify-discord
+
+Decide which channel should receive notifications (or create a new one). Open
+the "Server Settings" of your discord server. Click on "Webhooks". Click
+"Create Webhook". Configure the Name and Channel. Copy the Webhook URL.
+
+
+Test your tokens are working correctly by sending yourself a notification:
+
+.. code-block:: bash
+
+    sn0int notify exec kpcyrd/notify-discord -o url=https://discord.com/api/webhooks/1337/asdf 'hello world'
+
+You should receive ``hello world`` in your discord channel.
+
+Signal
+~~~~~~
+
+Install the sn0int notification module from the registry:
+
+.. code-block:: bash
+
+    sn0int pkg install kpcyrd/notify-signal
+
+This module allows end-to-end encrypted notifications, but it's also difficult
+to setup. You need a second phone number and install both `signal-cli
+<https://github.com/AsamK/signal-cli>`_ and `sn0int-signal
+<https://github.com/kpcyrd/sn0int-signal>`_.
+
+After you've registered your second phone number with signal-cli, you can use
+sn0int-signal to expose a minimal api for notify-signal. For more detailed
+instructions and how to start the api at boot, see the `sn0int-signal README
+<https://github.com/kpcyrd/sn0int-signal>`_.
+
+Read the secret key generated at ``/etc/sn0int-signal.key`` and send a
+notification to the signal phone number:
+
+.. code-block:: bash
+
+    sn0int notify exec kpcyrd/notify-signal -o to=+31337 -o secret=asdf 'hello world'
+
+You should receive ``hello world`` from the number signed up with signal-cli.
+
+Writing your own module
+~~~~~~~~~~~~~~~~~~~~~~~
+
+Make sure you've read the detailed instructions on how to get setup with
+`module development <scripting.html>`_.
+
+Create a new sn0int module like this:
+
+.. code-block:: bash
+
+    sn0int new ~/repos/sn0int-modules/notify-custom.lua
+
+Edit the ``-- Source:`` so it takes notifications as input:
+
+.. code-block:: lua
+
+    -- Description: TODO your description here
+    -- Version: 0.1.0
+    -- License: GPL-3.0
+    -- Source: notifications
+
+    function run(arg)
+        -- TODO your code here
+        -- https://sn0int.readthedocs.io/en/stable/reference.html
+
+        debug(arg)
+        info(arg['subject'])
+        info(arg['body'])
+    end
+
+Execute your script:
+
+.. code-block:: bash
+
+    sn0int notify exec notify-custom 'hello world'
+
+You most likely need to pass options to avoid hard-coding keys into your
+script. Options can be fetched like this:
+
+.. code-block:: lua
+
+    -- Description: TODO your description here
+    -- Version: 0.1.0
+    -- License: GPL-3.0
+    -- Source: notifications
+
+    function run(arg)
+        -- TODO your code here
+        -- https://sn0int.readthedocs.io/en/stable/reference.html
+
+        local foo = getopt('foo')
+        if not foo then return 'Missing -o foo= option' end
+
+        info('foo: ' .. foo)
+        info('subject: ' .. arg['subject'])
+    end
+
+And passed like this:
+
+.. code-block:: bash
+
+    sn0int notify exec notify-custom -o "foo=hello world" 'ohai'
+
+Setting up notification rules
+-----------------------------
+
+We now know how to trigger notifications manually, but we would rather trigger
+notifications if a module runs into something interesting.
+
+You can setup subscriptions on specific topics and then have a notification
+script execute automatically.
+
+Lookup the location of your sn0int config file:
+
+.. code-block:: bash
+
+    sn0int paths
+
+And open it in an editor of your choice:
+
+.. code-block:: bash
+
+    vim /home/user/.config/sn0int.toml
+
+A basic configuration could look like this:
+
+.. code-block:: toml
+
+    # You can have multiple notification sections, this one is named
+    # `demo-telegram-integration`
+    # The label can be set to whatever you want, but you may need to add
+    # double-quotes to use some characters.
+    [notifications.demo-telegram-integration]
+    # If this option is present, the notification must originate from one of
+    # the following workspaces.
+    workspaces = ["default", "some-workspace"]
+    # If this option is present, the notification must match one of the
+    # filters. You can use `*` as a wildcard to match everything except `:`.
+    topics = ["activity:harness/activity-ping:*"]
+    # Mandatory: the module to execute.
+    script = "kpcyrd/notify-telegram"
+    # The options to pass to the module, if any.
+    # Can be accessed with `getopt`
+    options = [
+        "bot_token=1337:foobar",
+        "chat_id=1337",
+    ]
+
+All options except ``script`` are optional, but setting filters is highly
+recommended.
+
+Testing notifications
+---------------------
+
+To test if your configuration works correctly you can create an event manually:
+
+.. code-block:: bash
+
+    sn0int -w some-workspace notify send activity:harness/activity-ping:dummy "hello world"
+
+If it matches any of your rules you should receive a push notifications.
+
+.. note::
+    If you want to test just the routing without actually sending something, add ``--dry-run``.
+
+Running sn0int automatically
+----------------------------
+
+Support for this is going to improve in the future, but you can already set
+this up if you're ok with a slightly buggy experience.
+
+Monitors
+~~~~~~~~
+
+Some modules are long-running and either wait for an event from a server or
+have custom polling built in that's usually configurable with an ``-o
+interval=`` option. If your module has a non-trivial setup phase, an author may
+take this approach.
+
+.. code-block::
+
+    # /etc/systemd/system/sn0int-your-new-service.service
+
+    [Unit]
+    Description=sn0int: run example/changeme
+
+    [Service]
+    User=your-user
+    ExecStart=/usr/bin/sn0int run -w your-workspace example/changeme
+
+    Restart=always
+    RestartSec=0
+
+    [Install]
+    WantedBy=multi-user.target
+
+Enable the service to run on boot:
+
+.. code-block:: bash
+
+    systemctl enable --now sn0int-your-new-service.service
+
+Timers
+~~~~~~
+
+If the module is only one-shot you can set it up to run with a timer:
+
+.. code-block::
+
+    # /etc/systemd/system/sn0int-your-other-service.service
+
+    [Unit]
+    Description=sn0int: run example/changeme
+
+    [Service]
+    User=your-user
+    ExecStart=/usr/bin/sn0int run -w your-workspace example/changeme
+
+Setup the timer like this:
+
+.. code-block::
+
+    # /etc/systemd/system/sn0int-your-other-service.timer
+
+    [Unit]
+    Description=sn0int: run example/changeme
+
+    [Timer]
+    OnBootSec=1min
+    OnUnitActiveSec=1h
+
+    [Install]
+    WantedBy=timers.target
+
+.. code-block:: bash
+
+    systemctl enable --now sn0int-your-other-service.timer

modules/harness/activity-ping-once.lua

@@ -0,0 +1,26 @@
+-- Description: Log some dummy activity
+-- Version: 0.1.0
+-- License: GPL-3.0
+
+function run()
+    local uniq = getopt('uniq')
+    local topic = getopt('topic') or 'harness/activity-ping:dummy'
+
+    if getopt('gps') then
+        lat=1.23
+        lon=4.56
+        radius=100
+    end
+
+    db_activity({
+        topic=topic,
+        time=sn0int_time(),
+        uniq=uniq,
+        latitude=lat,
+        longitude=lon,
+        radius=radius,
+        content={
+            msg='ohai',
+        },
+    })
+end

modules/harness/notify-ohai.lua

@@ -0,0 +1,10 @@
+-- Description: TODO your description here
+-- Version: 0.1.0
+-- License: GPL-3.0
+-- Source: notifications
+
+function run(arg)
+    -- TODO your code here
+    sleep(1)
+    info('notication!: ' .. json_encode(arg))
+end

sn0int-common/src/metadata.rs

@@ -44,6 +44,7 @@ pub enum Source {
     Netblocks,
     CryptoAddrs(Option<String>),
     KeyRing(String),
+    Notifications,
 }
 
 impl Source {
@@ -63,6 +64,7 @@ impl Source {
             Source::Ports => "ports",
             Source::Netblocks => "netblocks",
             Source::CryptoAddrs(_) => "cryptoaddrs",
+            Source::Notifications => "notifications",
             Source::KeyRing(_) => "keyring",
         }
     }
@@ -94,6 +96,7 @@ impl FromStr for Source {
             ("ports", None) => Ok(Source::Ports),
             ("netblocks", None) => Ok(Source::Netblocks),
             ("cryptoaddrs", param) => Ok(Source::CryptoAddrs(param.map(String::from))),
+            ("notifications", None) => Ok(Source::Notifications),
             ("keyring", Some(param)) => Ok(Source::KeyRing(param.to_string())),
             (x, Some(param)) => bail!("Unknown Source: {:?} ({:?})", x, param),
             (x, None) => bail!("Unknown Source: {:?}", x),

src/args.rs

@@ -75,6 +75,9 @@ pub enum SubCommand {
     /// Calendar
     #[structopt(name="cal")]
     Cal(cmd::cal_cmd::Args),
+    /// Notify
+    #[structopt(name="notify")]
+    Notify(cmd::notify_cmd::Args),
     /// Verify blob storage for corrupt and dangling blobs
     #[structopt(name="fsck")]
     Fsck(cmd::fsck_cmd::Args),

src/cmd/mod.rs

@@ -30,6 +30,7 @@ pub mod use_cmd;
 pub mod select_cmd;
 pub mod keyring_cmd;
 pub mod noscope_cmd;
+pub mod notify_cmd;
 pub mod pkg_cmd;
 pub mod set_cmd;
 pub mod scope_cmd;

src/cmd/notify_cmd.rs

@@ -0,0 +1,92 @@
+use crate::errors::*;
+
+use crate::cmd::Cmd;
+use crate::engine::Module;
+// use crate::models::*;
+use crate::notify::{self, Notification};
+use crate::options::{self, Opt};
+use crate::shell::Shell;
+use crate::term;
+use structopt::StructOpt;
+use structopt::clap::AppSettings;
+
+#[derive(Debug, StructOpt)]
+#[structopt(global_settings = &[AppSettings::ColoredHelp])]
+pub struct Args {
+    #[structopt(subcommand)]
+    subcommand: Subcommand,
+}
+
+#[derive(Debug, StructOpt)]
+pub enum Subcommand {
+    /// Manually add a notification to the outbox
+    Send(SendArgs),
+    /// Show the current outbox
+    Outbox,
+    /// Execute a module directly instead of sending a message
+    Exec(ExecArgs),
+    /// Try to deliver all messages in our outbox
+    Deliver,
+}
+
+#[derive(Debug, StructOpt)]
+pub struct SendArgs {
+    /// Evaluate the routing rules, but do not actually send a notification
+    #[structopt(short="n", long)]
+    pub dry_run: bool,
+    pub topic: String,
+    #[structopt(flatten)]
+    pub notification: Notification,
+}
+
+#[derive(Debug, StructOpt)]
+pub struct ExecArgs {
+    pub module: String,
+    #[structopt(short="o", long="option")]
+    pub options: Vec<options::Opt>,
+    #[structopt(short="v", long="verbose", parse(from_occurrences))]
+    verbose: u64,
+    #[structopt(flatten)]
+    pub notification: Notification,
+}
+
+fn print_summary(module: &Module, sent: usize, errors: usize) {
+        let mut out = if sent == 1 {
+            String::from("Sent 1 notification")
+        } else {
+            format!("Sent {} notifications", sent)
+        };
+
+        out.push_str(&format!(" with {}", module.canonical()));
+
+        if errors > 0 {
+            out.push_str(&format!(" ({} errors)", errors));
+        }
+
+        term::info(&out);
+}
+
+fn send(args: SendArgs, rl: &mut Shell) -> Result<()> {
+    notify::run_router(rl, &mut term::Term, args.dry_run, &args.topic, &args.notification)?;
+    Ok(())
+}
+
+fn exec(args: ExecArgs, rl: &mut Shell) -> Result<()> {
+    let module = rl.library().get(&args.module)?.clone();
+    let options = Opt::collect(&args.options);
+    let errors = notify::exec(rl, &module, options, args.verbose, &args.notification)?;
+    print_summary(&module, 1, errors);
+    Ok(())
+}
+
+impl Cmd for Args {
+    #[inline]
+    fn run(self, rl: &mut Shell) -> Result<()> {
+        match self.subcommand {
+            Subcommand::Send(args) => send(args, rl),
+            Subcommand::Outbox => todo!(),
+            Subcommand::Exec(args) => exec(args, rl),
+            Subcommand::Deliver => todo!(),
+        }
+    }
+}

src/cmd/run_cmd.rs

@@ -92,7 +92,7 @@ fn prepare_args<T: Scopable + Serialize + Model>(rl: &Shell, filter: &Filter, pa
         .collect()
 }
 
-fn prepare_keyring(keyring: &mut KeyRing, module: &Module, params: &Params) -> Result<()> {
+pub fn prepare_keyring(keyring: &mut KeyRing, module: &Module, params: &Params) -> Result<()> {
     for namespace in keyring.unauthorized_namespaces(&module) {
         let grant_access = if params.deny_keyring {
             false
@@ -133,6 +133,7 @@ fn get_args(rl: &mut Shell, module: &Module) -> Result<Vec<(serde_json::Value, O
         Some(Source::Ports) => prepare_args::<Port>(rl, &filter, None),
         Some(Source::Netblocks) => prepare_args::<Netblock>(rl, &filter, None),
         Some(Source::CryptoAddrs(currency)) => prepare_args::<CryptoAddr>(rl, &filter, currency.as_ref()),
+        Some(Source::Notifications) => bail!("Notification modules can't be executed like this"),
         Some(Source::KeyRing(namespace)) => {
             let keyring = rl.keyring();
             if keyring.is_access_granted(&module, &namespace) {

src/cmd/target_cmd.rs

@@ -44,6 +44,7 @@ pub fn run(rl: &mut Shell, args: &[String]) -> Result<()> {
             Source::Ports => select::<Port>(rl, None)?,
             Source::Netblocks => select::<Netblock>(rl, None)?,
             Source::CryptoAddrs(currency) => select::<CryptoAddr>(rl, currency.as_ref())?,
+            Source::Notifications => bail!("Notifications can't be set as target"),
             Source::KeyRing(namespace) => {
                 for key in rl.keyring().list_for(&namespace) {
                     println!("{}:{}", key.namespace, key.name);
@@ -79,6 +80,7 @@ fn count_selected(rl: &mut Shell, source: &Source) -> Result<usize> {
         Source::Ports => db.filter::<Port>(&filter)?.len(),
         Source::Netblocks => db.filter::<Netblock>(&filter)?.len(),
         Source::CryptoAddrs(currency) => db.filter_with_param::<CryptoAddr>(&filter, currency.as_ref())?.len(),
+        Source::Notifications => bail!("Notifications can't be set as target"),
         Source::KeyRing(namespace) => rl.keyring().list_for(&namespace).len(),
     };
     Ok(num)

src/config.rs

@@ -1,5 +1,6 @@
 use dirs;
 use crate::errors::*;
+use crate::notify::NotificationConfig;
 use std::collections::HashMap;
 use std::fs;
 use std::path::{Path, PathBuf};
@@ -15,6 +16,8 @@ pub struct Config {
     pub namespaces: HashMap<String, PathBuf>,
     #[serde(default)]
     pub network: NetworkConfig,
+    #[serde(default)]
+    pub notifications: HashMap<String, NotificationConfig>,
 }
 
 impl Config {

src/lib.rs

@@ -34,6 +34,7 @@ use sn0int_std::lazy;
 pub mod migrations;
 pub mod models;
 use sn0int_std::mqtt;
+pub mod notify;
 pub mod paths;
 pub use sn0int_std::psl;
 pub mod options;

src/main.rs

@@ -126,6 +126,7 @@ fn run() -> Result<()> {
         Some(SubCommand::Fsck(fsck)) => run_cmd(&args, fsck, &config),
         Some(SubCommand::Export(export)) => run_cmd(&args, export, &config),
         Some(SubCommand::Cal(cal)) => run_cmd(&args, cal, &config),
+        Some(SubCommand::Notify(notify)) => run_cmd(&args, notify, &config),
         Some(SubCommand::Repl) => repl::run(&config),
         Some(SubCommand::Paths) => paths::run(&config),
         Some(SubCommand::Completions(completions)) => complete::run_generate(&completions),

src/notify/mod.rs

@@ -0,0 +1,235 @@
+use crate::errors::*;
+use crate::cmd::run_cmd::prepare_keyring;
+use crate::cmd::run_cmd::Params;
+use crate::engine::Module;
+use crate::options;
+use crate::shell::Shell;
+use crate::term::SpinLogger;
+use crate::worker;
+use serde::de::{self, Deserialize, Deserializer};
+use serde::ser::{Serialize, Serializer};
+use sn0int_common::metadata::Source;
+use sn0int_std::blobs::Blob;
+use std::collections::HashMap;
+use std::result;
+use std::str::FromStr;
+
+#[derive(Debug, StructOpt, Serialize)]
+pub struct Notification {
+    pub subject: String,
+    pub body: Option<String>,
+}
+
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+pub struct NotificationConfig {
+    #[serde(default)]
+    pub workspaces: Vec<String>,
+    #[serde(default)]
+    pub topics: Vec<Glob>,
+    pub script: String,
+    #[serde(default)]
+    pub options: Vec<options::Opt>,
+}
+
+#[derive(Debug, Clone)]
+pub struct Glob {
+    patterns: Vec<glob::Pattern>,
+    src: String,
+}
+
+impl Glob {
+    fn matches(&self, topic: &str) -> bool {
+        let mut filter = self.patterns.iter();
+        let mut topic = topic.split(':');
+
+        loop {
+            match (filter.next(), topic.next()) {
+                (Some(filter), Some(topic)) => if !filter.matches(&topic) {
+                    return false;
+                },
+                (None, None) => return true,
+                (_, _) => return false,
+            }
+        }
+    }
+}
+
+impl FromStr for Glob {
+    type Err = Error;
+
+    fn from_str(s: &str) -> Result<Glob> {
+        let patterns = s.split(':')
+            .map(|s| glob::Pattern::new(s).map_err(Error::from))
+            .collect::<Result<Vec<_>>>()?;
+        Ok(Glob {
+            patterns,
+            src: s.to_string(),
+        })
+    }
+}
+
+impl Serialize for Glob {
+    fn serialize<S>(&self, serializer: S) -> result::Result<S::Ok, S::Error>
+        where S: Serializer
+    {
+        serializer.serialize_str(&self.src)
+    }
+}
+
+impl<'de> Deserialize<'de> for Glob {
+    fn deserialize<D>(deserializer: D) -> result::Result<Self, D::Error>
+        where D: Deserializer<'de>
+    {
+        let s = String::deserialize(deserializer)?;
+        FromStr::from_str(&s).map_err(de::Error::custom)
+    }
+}
+
+fn apply_rule<T>(name: &str, filter: &[T], value: &str, cmp: fn(&T, &str) -> bool) -> bool {
+    if !filter.is_empty() {
+        debug!("{} filter is active", name);
+        if !filter.iter().any(|filter| cmp(filter, value)) {
+            debug!("{} isn't allow-listed, aborting", name);
+            return false;
+        }
+        debug!("{} was allow-listed", name);
+    }
+    true
+}
+
+impl NotificationConfig {
+    fn matches(&self, workspace: &str, topic: &str) -> bool {
+        debug!("testing notification with rules");
+        if !apply_rule("workspace", &self.workspaces, workspace, |filter, value| filter == value) {
+            return false;
+        }
+        if !apply_rule("topic", &self.topics, topic, |filter, value| filter.matches(value)) {
+            return false;
+        }
+        debug!("notification matches this config");
+        true
+    }
+}
+
+pub fn trigger_notify_event<T: SpinLogger>(rl: &mut Shell, spinner: &mut T, topic: &str, notification: &Notification) -> Result<()> {
+    run_router(rl, spinner, false, topic, notification)
+}
+
+fn prepare_arg(notification: &Notification) -> Result<(serde_json::Value, Option<String>, Vec<Blob>)> {
+    let arg = serde_json::to_value(notification)?;
+    Ok((arg, None, vec![]))
+}
+
+pub fn exec(rl: &mut Shell, module: &Module, options: HashMap<String, String>, verbose: u64, notification: &Notification) -> Result<usize> {
+    if *module.source() != Some(Source::Notifications) {
+        bail!("Module doesn't take notifications as source");
+    }
+
+    let params = Params {
+        threads: 1,
+        verbose,
+        stdin: false,
+        grants: &[],
+        grant_full_keyring: false,
+        deny_keyring: false,
+        exit_on_error: false,
+    };
+
+    prepare_keyring(rl.keyring_mut(), &module, &params)?;
+    let args = vec![prepare_arg(&notification)?];
+
+    rl.signal_register().catch_ctrl();
+    let errors = worker::spawn(rl, &module, args, &params, rl.config().network.proxy.clone(), options);
+    rl.signal_register().reset_ctrlc();
+
+    Ok(errors)
+}
+
+pub fn run_router<T: SpinLogger>(rl: &mut Shell, spinner: &mut T, dry_run: bool, topic: &str, notification: &Notification) -> Result<()> {
+    let configs = rl.config().notifications.clone();
+
+    for (name, config) in configs {
+        if config.matches(rl.workspace(), topic) {
+            let module = rl.library().get(&config.script)?.clone();
+            if dry_run {
+                spinner.success(&format!("Executed {} {:?} (dry-run)", module.canonical(), name));
+            } else {
+                let options = options::Opt::collect(&config.options);
+                match exec(rl, &module, options, 0, notification) {
+                    Ok(0) => {
+                        let msg = format!("Executed {} {:?}", module.canonical(), name);
+                        spinner.success(&msg);
+                    },
+                    Ok(errors) => {
+                        let msg = format!("Executed {} {:?} ({} errors)", module.canonical(), name, errors);
+                        spinner.error(&msg);
+                    },
+                    Err(err) => {
+                        spinner.error(&format!("Fatal {} {:?}: {}", module.canonical(), name, err));
+                    },
+                }
+            }
+        }
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn match_topic_str(filter: &str, value: &str) -> bool {
+        let filter: Glob = filter.parse().unwrap();
+        filter.matches(value)
+    }
+
+    #[test]
+    fn test_match_topic_exact() {
+        assert!(match_topic_str("topic:hello-world", "topic:hello-world"));
+    }
+
+    #[test]
+    fn test_match_topic_starts_with() {
+        assert!(match_topic_str("topic:*", "topic:hello-world"));
+    }
+
+    #[test]
+    fn test_match_topic_ends_with() {
+        assert!(match_topic_str("*:hello-world", "topic:hello-world"));
+    }
+
+    #[test]
+    fn test_match_topic_one_wildcard_one_section() {
+        assert!(match_topic_str("a:*:z", "a:b:z"));
+    }
+
+    #[test]
+    fn test_match_topic_one_wildcard_not_two_sections() {
+        assert!(!match_topic_str("a:*:z", "a:b:c:z"));
+    }
+
+    #[test]
+    fn test_match_topic_two_wildcards_two_sections() {
+        assert!(match_topic_str("a:*:*:z", "a:b:c:z"));
+    }
+
+    #[test]
+    fn test_match_topic_one_wildcard_not_two_sections_start() {
+        assert!(!match_topic_str("a:*", "a:b:c"));
+    }
+
+    #[test]
+    fn test_match_topic_one_wildcard_not_two_sections_end() {
+        assert!(!match_topic_str("*:z", "b:c:z"));
+    }
+
+    #[test]
+    fn test_match_topic_many_wildcards() {
+        assert!(match_topic_str("a:*:*:d:e:*:g:*:z", "a:b:c:d:e:f:g:h:z"));
+    }
+
+    #[test]
+    fn test_match_topic_empty_filter() {
+        assert!(!match_topic_str("", "abc"));
+    }
+}

src/options.rs

@@ -1,9 +1,11 @@
 use crate::errors::*;
+use serde::de::{self, Deserialize, Deserializer};
+use serde::ser::{Serialize, Serializer};
 use std::collections::HashMap;
+use std::result;
 use std::str::FromStr;
 
-
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct Opt {
     key: String,
     value: String,
@@ -32,3 +34,26 @@ impl FromStr for Opt {
         }
     }
 }
+
+impl ToString for Opt {
+    fn to_string(&self) -> String {
+        format!("{}={}", self.key, self.value)
+    }
+}
+
+impl Serialize for Opt {
+    fn serialize<S>(&self, serializer: S) -> result::Result<S::Ok, S::Error>
+        where S: Serializer
+    {
+        serializer.serialize_str(&self.to_string())
+    }
+}
+
+impl<'de> Deserialize<'de> for Opt {
+    fn deserialize<D>(deserializer: D) -> result::Result<Self, D::Error>
+        where D: Deserializer<'de>
+    {
+        let s = String::deserialize(deserializer)?;
+        FromStr::from_str(&s).map_err(de::Error::custom)
+    }
+}

src/shell/mod.rs

@@ -206,6 +206,11 @@ impl<'a> Shell<'a> {
         self.prompt.module.as_ref()
     }
 
+    #[inline(always)]
+    pub fn workspace(&self) -> &str {
+        self.prompt.workspace.as_str()
+    }
+
     #[inline(always)]
     pub fn options_mut(&mut self) -> Option<&mut HashMap<String, String>> {
         self.options.as_mut()

src/term.rs

@@ -221,6 +221,44 @@ pub fn error(line: &str) {
     eprintln!("\x1b[1m[\x1b[31m{}\x1b[0;1m]\x1b[0m {}", '-', line);
 }
 
+pub struct Term;
+
+impl SpinLogger for Term {
+    fn log(&mut self, line: &str) {
+        success(line)
+    }
+
+    fn debug(&mut self, line: &str) {
+        debug(line)
+    }
+
+    fn success(&mut self, line: &str) {
+        info(line)
+    }
+
+    fn error(&mut self, line: &str) {
+        error(line)
+    }
+
+    fn warn(&mut self, line: &str) {
+        warn(line)
+    }
+
+    fn warn_once(&mut self, _line: &str) {
+        unimplemented!()
+    }
+
+    #[inline]
+    fn status(&mut self, _status: String) {
+        unimplemented!()
+    }
+
+    #[inline]
+    fn stacked_status(&mut self, _name: &String, _status: String) {
+        unimplemented!()
+    }
+}
+
 pub struct Prompt {
     pub workspace: String,
     pub module: Option<Module>,

src/worker.rs

@@ -9,6 +9,7 @@ use crate::engine::Module;
 use crate::ipc;
 use crate::ipc::parent::IpcParent;
 use crate::models::*;
+use crate::notify::{self, Notification};
 use serde_json;
 use crate::ratelimits::{Ratelimiter, RatelimitResponse};
 use crate::shell::Shell;
@@ -231,32 +232,52 @@ impl DatabaseEvent {
         tx.send(result).expect("Failed to send db result to channel");
     }
 
-    pub fn activity<T: SpinLogger>(object: NewActivity, tx: DbSender, spinner: &mut T, db: &Database, verbose: u64) {
+    fn spinner_log_new_activity<T: SpinLogger>(object: &NewActivity, spinner: &mut T, verbose: u64) {
+        let mut log = format!("{:?} ", object.topic);
+        if let Some(uniq) = &object.uniq {
+            log.push_str(&format!("({:?}) ", uniq));
+        }
+        log.push_str(&format!("@ {}", object.time));
+
+        if let (Some(ref lat), Some(ref lon)) = (object.latitude, object.longitude) {
+            log.push_str(&format!(" ({}, {}", lat, lon));
+            if let Some(radius) = &object.radius {
+                log.push_str(&format!(" | {}m", radius));
+            }
+            log.push_str(")");
+        }
+
+        if verbose > 0 {
+            log.push_str(&format!(": {}", object.content));
+        }
+
+        spinner.log(&log);
+    }
+
+    pub fn activity<T: SpinLogger>(rl: &mut Shell, object: NewActivity, tx: DbSender, spinner: &mut T, verbose: u64) {
+        let db = rl.db();
         let result = db.insert_activity(object.clone());
         debug!("{:?} => {:?}", object, result);
 
         let result = match result {
             Ok(true) => {
-                let mut log = format!("{:?} ", object.topic);
+                Self::spinner_log_new_activity(&object, spinner, verbose);
+                // TODO: we don't want to copy the match arms everywhere
+                let topic = format!("activity:{}", object.topic);
+                let mut subject = format!("New activity: {:?}", object.topic);
                 if let Some(uniq) = &object.uniq {
-                    log.push_str(&format!("({:?}) ", uniq));
+                    subject += &format!(" ({:?})", uniq);
                 }
-                log.push_str(&format!("@ {}", object.time));
-
-                if let (Some(ref lat), Some(ref lon)) = (object.latitude, object.longitude) {
-                    log.push_str(&format!(" ({}, {}", lat, lon));
-                    if let Some(radius) = &object.radius {
-                        log.push_str(&format!(" | {}m", radius));
-                    }
-                    log.push_str(")");
+                if let Err(err) = notify::trigger_notify_event(rl, spinner, &topic, &Notification {
+                    subject,
+                    body: None,
+                }) {
+                    let err = err.to_string();
+                    spinner.error(&format!("Failed to send notifications: {}", err));
+                    Err(err)
+                } else {
+                    Ok(DatabaseResponse::Inserted(0))
                 }
-
-                if verbose > 0 {
-                    log.push_str(&format!(": {}", object.content));
-                }
-
-                spinner.log(&log);
-                Ok(DatabaseResponse::Inserted(0))
             },
             Ok(false) => Ok(DatabaseResponse::NoChange(0)),
             Err(err) => {
@@ -269,11 +290,12 @@ impl DatabaseEvent {
         tx.send(result).expect("Failed to send db result to channel");
     }
 
-    pub fn apply<T: SpinLogger>(self, tx: DbSender, spinner: &mut T, db: &Database, verbose: u64) {
+    pub fn apply<T: SpinLogger>(self, rl: &mut Shell, tx: DbSender, spinner: &mut T, verbose: u64) {
+        let db = rl.db();
         match self {
             DatabaseEvent::Insert(object) => Self::insert(object, None, tx, spinner, db, verbose),
             DatabaseEvent::InsertTtl((object, ttl)) => Self::insert(object, Some(ttl), tx, spinner, db, verbose),
-            DatabaseEvent::Activity(object) => Self::activity(object, tx, spinner, db, verbose),
+            DatabaseEvent::Activity(object) => Self::activity(rl, object, tx, spinner, verbose),
             DatabaseEvent::Select((family, value)) => {
                 let result = match db.get_opt(&family, &value) {
                     Ok(Some(id)) => Ok(DatabaseResponse::Found(id)),
@@ -443,7 +465,7 @@ pub fn spawn(rl: &mut Shell, module: &Module, args: Vec<(serde_json::Value, Opti
                             stack.add(name, label);
                         },
                         Event2::Log(log) => log.apply(&mut stack.prefixed(name)),
-                        Event2::Database((db, tx)) => db.apply(tx, &mut stack.prefixed(name), rl.db(), verbose),
+                        Event2::Database((db, tx)) => db.apply(rl, tx, &mut stack.prefixed(name), verbose),
                         Event2::Ratelimit((req, tx)) => ratelimit.pass(tx, &req.key, req.passes, req.time),
                         Event2::Blob((blob, tx)) => rl.store_blob(tx, &blob),
                         Event2::Exit(event) => {