Hassliebe/src/modules/notifications.rs

use std::collections::HashMap;
use std::sync::Arc;

use anyhow::{Context, Result};
use base64::Engine;
use notify_rust::Hint;
use rumqttc::QoS;
use serde::Deserialize;
use tokio::task::spawn_blocking;
use validator::Validate;

use crate::modules::{InitializationContext, ModuleContext};
use crate::util::{generate_alphanumeric_id, hash_string_to_u32, spawn_nonessential};

const MODULE_ID: &str = "notifications";

#[derive(Deserialize, Validate)]
pub struct Config {
    #[serde(default)]
    pub enabled: bool,
}

#[derive(Deserialize, Default)]
#[serde(rename_all = "snake_case")]
pub enum NotificationTimeout {
    Never,

    #[default]
    Default,

    Ms(u16),
}

#[derive(Deserialize, Default)]
#[serde(rename_all = "snake_case")]
pub enum NotificationUrgency {
    Low,

    #[default]
    Normal,
    Critical,
}

impl From<NotificationUrgency> for notify_rust::Urgency {
    fn from(value: NotificationUrgency) -> Self {
        match value {
            NotificationUrgency::Low => notify_rust::Urgency::Low,
            NotificationUrgency::Normal => notify_rust::Urgency::Normal,
            NotificationUrgency::Critical => notify_rust::Urgency::Critical,
        }
    }
}

#[derive(Deserialize, Validate)]
pub struct NotificationMessage {
    /// Using the ID of an existing notification will replace the old notification.
    #[validate(length(min = 1))]
    id: Option<String>,

    #[serde(default)]
    timeout: NotificationTimeout,

    #[serde(default)]
    urgency: NotificationUrgency,

    /// Usually used as the title of the notification.
    #[validate(length(min = 1))]
    summary_text: String,

    /// Basic HTML is usually supported.
    #[validate(length(min = 1))]
    long_content: Option<String>,

    /// If the notification server supports persisting notifications across sessions, this will prevent it from doing so for this notification.
    #[serde(default)]
    transient: bool,

    /// Padded base64-encoded image
    encoded_image: Option<String>,

    #[serde(default)]
    actions: HashMap<String, String>,
}

pub async fn init(context: &mut InitializationContext) -> Result<()> {
    let full_context = context.get_full();
    let _config = match &full_context.config.modules.notifications {
        Some(c) if c.enabled => c,
        _ => return Ok(()),
    };

    log::info!("Initializing…");

    let full_context = context.get_full();
    context.subscribe_mqtt_topic(format!("{}/{}/simple", context.full.config.friendly_id, MODULE_ID), move |text| {
        let (summary_text, long_content) = text.split_once('\n').unwrap_or((text, ""));

        tokio::spawn(handle_notification_message(
            full_context.clone(),
            NotificationMessage {
                id: None,
                transient: true,
                summary_text: summary_text.to_owned(),
                long_content: if long_content.is_empty() { None } else { Some(long_content.to_owned()) },
                timeout: NotificationTimeout::default(),
                urgency: NotificationUrgency::default(),
                encoded_image: None,
                actions: HashMap::new(),
            },
        ));

        Ok(())
    });

    let full_context = context.get_full();
    context.subscribe_mqtt_topic(format!("{}/{}/json", context.full.config.friendly_id, MODULE_ID), move |text| {
        let message = serde_json::from_str::<NotificationMessage>(text);

        match message {
            Err(error) => log::error!("Could not deserialize message: {}", error),

            Ok(message) => {
                spawn_nonessential(handle_notification_message(full_context.clone(), message));
            }
        }

        Ok(())
    });

    context.subscribe_mqtt_topic(format!("{}/{}/close", context.full.config.friendly_id, MODULE_ID), move |id| {
        spawn_nonessential(close_notification(id.to_owned()));

        Ok(())
    });

    Ok(())
}

async fn handle_notification_message(context: Arc<ModuleContext>, message: NotificationMessage) -> Result<()> {
    let mut notification = notify_rust::Notification::new();
    let id = message.id.unwrap_or_else(|| generate_alphanumeric_id(10));
    let internal_id = hash_string_to_u32(id.as_str());

    notification.id(internal_id);
    notification.summary(message.summary_text.as_str());
    notification.urgency(message.urgency.into());
    notification.hint(Hint::Transient(message.transient));
    notification.timeout(match message.timeout {
        NotificationTimeout::Default => -1,
        NotificationTimeout::Never => 0,
        NotificationTimeout::Ms(ms) => ms.into(),
    });

    for (action_id, label) in message.actions {
        notification.action(&action_id, &label);
    }

    if let Some(encoded_image) = message.encoded_image {
        let image_data = base64::engine::general_purpose::STANDARD
            .decode(encoded_image)
            .context("while decoding the base64 image")?;

        let image = image::load_from_memory(&image_data).context("while reading the image")?;

        notification.image_data(image.try_into()?);
    }

    if let Some(long_content) = message.long_content {
        notification.body(long_content.as_str());
    }

    log::debug!("Showing notification: {}", id);
    let handle = notification.show_async().await?;

    {
        let context = context.clone();

        spawn_blocking(move || {
            handle.wait_for_action(|action_id| {
                spawn_nonessential(send_notification_action_message(
                    context,
                    id.clone(),
                    if action_id == "__closed" { "closed".to_owned() } else { action_id.to_owned() },
                ));
            });
        });
    }

    Ok(())
}

async fn close_notification(id: String) -> Result<()> {
    // We create a new notification because otherwise we would need to store old notification handles
    // and notify-rust does not expose a method to close a notification by ID.

    let internal_id = hash_string_to_u32(&id);
    let mut notification = notify_rust::Notification::new();

    notification.id(internal_id);

    log::debug!("Closing notification: {}", id);
    let handle = notification.show_async().await?;
    handle.close();

    Ok(())
}

async fn send_notification_action_message(context: Arc<ModuleContext>, notification_id: String, action_id: String) -> Result<()> {
    context
        .mqtt
        .client
        .publish(
            format!("{}/{}/action/{}", context.config.friendly_id, MODULE_ID, notification_id),
            QoS::ExactlyOnce,
            false,
            action_id,
        )
        .await?;

    Ok(())
}