1 files changed, 140 insertions, 0 deletions
diff --git a/pkgs/by-name/mp/mpdpopm/src/messanges/mod.rs b/pkgs/by-name/mp/mpdpopm/src/messanges/mod.rs
new file mode 100644
index 00000000..7db75672
--- /dev/null
+++ b/pkgs/by-name/mp/mpdpopm/src/messanges/mod.rs
@@ -0,0 +1,140 @@
+use anyhow::{Context, Result, anyhow, bail, ensure};
+use clap::{Parser, Subcommand};
+use shlex::Shlex;
+use tracing::info;
+
+use crate::{
+    clients::{Client, IdleClient},
+    config::Mode,
+    dj::{Dj, algorithms::Discovery},
+};
+
+pub const COMMAND_CHANNEL: &str = "unwoundstack.com:commands";
+
+#[derive(Parser)]
+struct Commands {
+    #[command(subcommand)]
+    command: SubCommand,
+}
+
+#[derive(Parser)]
+enum SubCommand {
+    Dj {
+        #[command(subcommand)]
+        command: DjCommand,
+    },
+}
+
+#[derive(Subcommand)]
+enum DjCommand {
+    Start {
+        /// The chance to select a "positive" track
+        #[arg(long)]
+        positive_chance: f64,
+
+        /// The chance to select a "neutral" track
+        #[arg(long)]
+        neutral_chance: f64,
+
+        /// The chance to select a "negative" track
+        #[arg(long)]
+        negative_chance: f64,
+    },
+    Stop {},
+}
+
+pub(crate) struct MessageQueue {
+    dj: Option<Dj<Discovery>>,
+}
+
+impl MessageQueue {
+    pub(crate) fn new(mode: Mode) -> Self {
+        match mode {
+            Mode::Normal => Self { dj: None },
+            Mode::Dj => {
+                info!("Dj mode started on launch, as specified in config file");
+
+                Self {
+                    dj: Some(Dj::new(Discovery::new(0.65, 0.5, 0.2))),
+                }
+            }
+        }
+    }
+
+    pub(crate) async fn advance_dj(&mut self, client: &mut Client) -> Result<()> {
+        if let Some(dj) = self.dj.as_mut() {
+            dj.add_track(client).await?;
+        }
+
+        Ok(())
+    }
+
+    /// Read messages off the commands channel & dispatch 'em
+    pub(crate) async fn check_messages(
+        &mut self,
+        client: &mut Client,
+        idle_client: &mut IdleClient,
+    ) -> Result<()> {
+        let m = idle_client
+            .get_messages()
+            .await
+            .context("Failed to `get_messages` from client")?;
+
+        for (chan, msgs) in m {
+            ensure!(chan == COMMAND_CHANNEL, "Unknown channel: `{}`", chan);
+
+            for msg in msgs {
+                self.process(client, msg).await?;
+            }
+        }
+
+        Ok(())
+    }
+
+    /// Process a single command
+    pub(crate) async fn process(&mut self, client: &mut Client, msg: String) -> Result<()> {
+        let split = {
+            let mut shl = Shlex::new(&msg);
+            let res: Vec<_> = shl.by_ref().collect();
+
+            if shl.had_error {
+                bail!("Failed to parse command '{msg}'")
+            }
+
+            assert_eq!(shl.line_no, 1, "A unexpected newline appeared");
+            assert!(!res.is_empty());
+
+            let mut base = vec!["base".to_owned()];
+            base.extend(res);
+            base
+        };
+
+        let args = Commands::parse_from(split);
+
+        match args.command {
+            SubCommand::Dj { command } => match command {
+                DjCommand::Start {
+                    positive_chance,
+                    neutral_chance,
+                    negative_chance,
+                } => {
+                    info!("Dj started");
+                    self.dj = Some(Dj::new(Discovery::new(
+                        positive_chance,
+                        neutral_chance,
+                        negative_chance,
+                    )));
+                    self.advance_dj(client).await?;
+                }
+                DjCommand::Stop {} => {
+                    self.dj
+                        .take()
+                        .ok_or_else(|| anyhow!("Tried to disable already disabled dj mode"))?;
+                    info!("Dj stopped");
+                }
+            },
+        }
+
+        Ok(())
+    }
+}