feat(service): add support for anchor paths (#127)

Author: dan-online

.gitignore

@@ -1,9 +1,9 @@
 # direnv
-.direnv
 .envrc
+.direnv
 
 # Testing
-config.toml
+/config.*
 autopulse.log
 data
 /target

src/service/runner.rs

@@ -13,24 +13,24 @@ use crate::{
 };
 use diesel::{BoolExpressionMethods, ExpressionMethods, QueryDsl, RunQueryDsl};
 use std::{path::PathBuf, sync::Arc};
+use tokio::sync::Mutex;
 use tracing::{error, info, warn};
 
 pub(super) struct PulseRunner {
     webhooks: Arc<WebhookManager>,
     settings: Arc<Settings>,
     pool: Arc<DbPool>,
+
+    anchors_available: Arc<Mutex<bool>>,
 }
 
 impl PulseRunner {
-    pub const fn new(
-        settings: Arc<Settings>,
-        pool: Arc<DbPool>,
-        webhooks: Arc<WebhookManager>,
-    ) -> Self {
+    pub fn new(settings: Arc<Settings>, pool: Arc<DbPool>, webhooks: Arc<WebhookManager>) -> Self {
         Self {
             webhooks,
             settings,
             pool,
+            anchors_available: Arc::new(Mutex::new(true)),
         }
     }
 
@@ -268,6 +268,24 @@ impl PulseRunner {
     }
 
     pub async fn run(&self) -> anyhow::Result<()> {
+        let set_anchors_available = self.settings.anchors.iter().all(|anchor| anchor.exists());
+
+        let mut anchors_available = self.anchors_available.lock().await;
+        if set_anchors_available != *anchors_available {
+            if set_anchors_available {
+                info!("anchors are available again, continuing");
+            } else {
+                warn!("anchors are not available, pausing");
+            }
+            *anchors_available = set_anchors_available;
+        }
+
+        if !*anchors_available {
+            return Ok(());
+        }
+
+        drop(anchors_available);
+
         self.update_found_status().await?;
         self.update_process_status().await?;
         self.cleanup().await?;

src/service/targets/autopulse.rs

@@ -56,7 +56,7 @@ impl Autopulse {
 }
 
 impl TargetProcess for Autopulse {
-    async fn process<'a>(&self, evs: &[&'a ScanEvent]) -> anyhow::Result<Vec<String>> {
+    async fn process(&self, evs: &[&ScanEvent]) -> anyhow::Result<Vec<String>> {
         let mut succeded = Vec::new();
 
         for ev in evs {

src/service/targets/emby.rs

@@ -285,7 +285,7 @@ impl Emby {
 }
 
 impl TargetProcess for Emby {
-    async fn process<'a>(&self, evs: &[&'a ScanEvent]) -> anyhow::Result<Vec<String>> {
+    async fn process(&self, evs: &[&ScanEvent]) -> anyhow::Result<Vec<String>> {
         let libraries = self.libraries().await?;
 
         let mut succeded = Vec::new();

src/service/targets/fileflows.rs

@@ -39,7 +39,7 @@ impl FileFlows {
 }
 
 impl TargetProcess for FileFlows {
-    async fn process<'a>(&self, evs: &[&'a ScanEvent]) -> anyhow::Result<Vec<String>> {
+    async fn process(&self, evs: &[&ScanEvent]) -> anyhow::Result<Vec<String>> {
         let mut succeeded = Vec::new();
 
         for ev in evs {

src/service/targets/plex.rs

@@ -228,7 +228,7 @@ impl Plex {
 }
 
 impl TargetProcess for Plex {
-    async fn process<'a>(&self, evs: &[&'a ScanEvent]) -> anyhow::Result<Vec<String>> {
+    async fn process(&self, evs: &[&ScanEvent]) -> anyhow::Result<Vec<String>> {
         let libraries = self.libraries().await?;
 
         let mut succeeded = Vec::new();

src/service/targets/tdarr.rs

@@ -80,7 +80,7 @@ impl Tdarr {
 }
 
 impl TargetProcess for Tdarr {
-    async fn process<'a>(&self, evs: &[&'a ScanEvent]) -> anyhow::Result<Vec<String>> {
+    async fn process(&self, evs: &[&ScanEvent]) -> anyhow::Result<Vec<String>> {
         let mut succeeded = Vec::new();
 
         self.scan(evs).await?;

src/settings/mod.rs

@@ -4,7 +4,7 @@ use auth::Auth;
 use config::Config;
 use opts::Opts;
 use serde::Deserialize;
-use std::collections::HashMap;
+use std::{collections::HashMap, path::PathBuf};
 use target::Target;
 use timer::Timer;
 use trigger::Trigger;
@@ -123,6 +123,20 @@ pub struct Settings {
 
     #[serde(default)]
     pub webhooks: HashMap<String, Webhook>,
+
+    /// List of paths to anchor the service to
+    ///
+    /// This is useful to prevent the service notifying a target when the drive is not mounted or visible
+    /// The contents of the file/directory are not tampered with, only the presence of the file/directory is checked
+    ///
+    /// Example:
+    /// ```yml
+    /// anchors:
+    ///  - /mnt/media/tv # Directory
+    ///  - /mnt/media/anchor # File
+    /// ```
+    #[serde(default)]
+    pub anchors: Vec<PathBuf>,
 }
 
 impl Settings {

src/settings/target.rs

@@ -9,9 +9,9 @@ use crate::{
 };
 
 pub trait TargetProcess {
-    fn process<'a>(
+    fn process(
         &self,
-        evs: &[&'a ScanEvent],
+        evs: &[&ScanEvent],
     ) -> impl std::future::Future<Output = anyhow::Result<Vec<String>>> + Send;
 }