feat(repo/webhook): Handle messages received via webhook for ForgeJo

Closes kemitix/git-next#43

Cargo.toml

@@ -32,6 +32,7 @@ tempfile = "3.10"
 
 # TOML parsing
 serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
 toml = "0.8"
 
 # Secrets and Password

src/server/actors/repo/mod.rs

@@ -20,6 +20,9 @@ pub struct RepoActor {
     webhook: Webhook,
     webhook_id: Option<WebhookId>, // INFO: if [None] then no webhook is configured
     webhook_auth: Option<ulid::Ulid>,
+    last_main_commit: Option<forge::Commit>,
+    last_next_commit: Option<forge::Commit>,
+    last_dev_commit: Option<forge::Commit>,
     net: Network,
     git: Git,
 }
@@ -35,6 +38,9 @@ impl RepoActor {
             webhook,
             webhook_id: None,
             webhook_auth: None,
+            last_main_commit: None,
+            last_next_commit: None,
+            last_dev_commit: None,
             net,
             git,
         }
@@ -81,7 +87,6 @@ impl Handler<LoadedConfig> for RepoActor {
         info!(%self.details, %config, "Config loaded");
         self.details.config.replace(config);
         if self.webhook_id.is_none() {
-            info!("lets register the webhook...");
             webhook::register(
                 self.details.clone(),
                 self.webhook.clone(),

src/server/actors/repo/status.rs

@@ -31,7 +31,9 @@ pub async fn check_next(
         Status::Pass => {
             addr.do_send(AdvanceMainTo(next));
         }
-        Status::Pending => {}
+        Status::Pending => {
+            // TODO: (#48)  reschedule a check after a few seconds
+        }
         Status::Fail => {
             warn!("Checks have failed");
         }

src/server/actors/repo/webhook.rs

@@ -1,15 +1,16 @@
 use actix::prelude::*;
 use kxio::network::{self, json};
-use tracing::{info, warn};
+use tracing::{debug, info, warn};
 
 use std::{fmt::Display, ops::Deref};
 
 use crate::server::{
     actors::{
-        repo::{RepoActor, WebhookRegistered},
+        repo::{RepoActor, ValidateRepo, WebhookRegistered},
         webhook::WebhookMessage,
     },
-    config::Webhook,
+    config::{RepoBranches, Webhook},
+    forge,
 };
 
 #[derive(Clone, Debug, PartialEq, Eq)]
@@ -127,10 +128,120 @@ impl Hook {
 impl Handler<WebhookMessage> for RepoActor {
     type Result = ();
 
-    fn handle(&mut self, msg: WebhookMessage, _ctx: &mut Self::Context) -> Self::Result {
+    #[allow(clippy::cognitive_complexity)] // TODO: (#49) reduce complexity
+    fn handle(&mut self, msg: WebhookMessage, ctx: &mut Self::Context) -> Self::Result {
         let id = msg.id();
         let body = msg.body();
-        info!(?id, ?body, "RepoActor received message");
+        debug!(?id, "RepoActor received message");
-        // TODO: (#43) do something with the message
+        match serde_json::from_str::<Push>(body) {
+            Err(err) => debug!(?err, %body, "Not a 'push'"),
+            Ok(push) => {
+                if let Some(config) = &self.details.config {
+                    match push.branch(config.branches()) {
+                        None => warn!(
+                            ?push,
+                            "Unrecognised branch, we should be filtering to only the ones we want"
+                        ),
+                        Some(branch) => {
+                            match branch {
+                                Branch::Main => {
+                                    if self.last_main_commit == Some(push.commit()) {
+                                        info!(
+                                            ?id,
+                                            "Ignoring - already aware of branch '{}' at commit '{}'",
+                                            config.branches().main(),
+                                            push.commit()
+                                        );
+                                        return;
+                                    }
+                                    self.last_main_commit.replace(push.commit())
+                                }
+                                Branch::Next => {
+                                    if self.last_next_commit == Some(push.commit()) {
+                                        info!(
+                                            ?id,
+                                            "Ignoring - already aware of branch '{}' at commit '{}'",
+                                            config.branches().next(),
+                                            push.commit()
+                                        );
+                                        return;
+                                    }
+                                    self.last_next_commit.replace(push.commit())
+                                }
+                                Branch::Dev => {
+                                    if self.last_dev_commit == Some(push.commit()) {
+                                        info!(
+                                            ?id,
+                                            "Ignoring - already aware of branch '{}' at commit '{}'",
+                                            config.branches().dev(),
+                                            push.commit()
+                                        );
+                                        return;
+                                    }
+                                    self.last_dev_commit.replace(push.commit())
+                                }
+                            };
+                            info!(
+                                ?branch,
+                                commit = %push.commit(),
+                                "New commit for branch - assessing branch positions"
+                            );
+                            ctx.address().do_send(ValidateRepo);
+                        }
+                    }
+                }
+            }
+        }
+    }
+}
+
+#[derive(Debug, serde::Deserialize)]
+struct Push {
+    #[serde(rename = "ref")]
+    reference: String,
+    after: String,
+    head_commit: HeadCommit,
+}
+impl Push {
+    pub fn branch(&self, repo_branches: &RepoBranches) -> Option<Branch> {
+        if !self.reference.starts_with("refs/heads/") {
+            warn!(r#ref = self.reference, "Unexpected ref");
+            return None;
+        }
+        let (_, branch) = self.reference.split_at(11);
+        if branch == *repo_branches.main() {
+            return Some(Branch::Main);
+        }
+        if branch == *repo_branches.next() {
+            return Some(Branch::Next);
+        }
+        if branch == *repo_branches.dev() {
+            return Some(Branch::Dev);
+        }
+        warn!(branch, "Unexpected branch");
+        None
+    }
+    pub fn commit(&self) -> forge::Commit {
+        forge::Commit::new(&self.after, &self.head_commit.message)
+    }
+}
+
+#[derive(Debug)]
+enum Branch {
+    Main,
+    Next,
+    Dev,
+}
+
+#[derive(Debug, serde::Deserialize)]
+struct HeadCommit {
+    message: String,
+}
+
+#[cfg(test)]
+mod tests {
+    #[test]
+    fn splt_ref() {
+        assert_eq!("refs/heads/next".split_at(11), ("refs/heads/", "next"));
     }
 }

src/server/actors/webhook/router.rs

@@ -1,7 +1,7 @@
 use std::collections::HashMap;
 
 use actix::prelude::*;
-use tracing::info;
+use tracing::debug;
 
 use crate::server::{actors::webhook::message::WebhookMessage, config::RepoAlias};
 
@@ -23,9 +23,9 @@ impl Handler<WebhookMessage> for WebhookRouter {
 
     fn handle(&mut self, msg: WebhookMessage, _ctx: &mut Self::Context) -> Self::Result {
         let repo_alias = RepoAlias(msg.path().clone());
-        info!(?repo_alias, "router...");
+        debug!(?repo_alias, "Router...");
         if let Some(recipient) = self.repos.get(&repo_alias) {
-            info!("Sending to recipient");
+            debug!("Sending to recipient");
             recipient.do_send(msg);
         }
     }

src/server/actors/webhook/server.rs

@@ -1,6 +1,6 @@
 use actix::prelude::*;
 
-use tracing::info;
+use tracing::{debug, info};
 
 use crate::server::actors::webhook::message::WebhookMessage;
 
@@ -23,7 +23,7 @@ pub async fn start(address: actix::prelude::Recipient<super::message::WebhookMes
                 let bytes = body.to_vec();
                 let request_data = String::from_utf8_lossy(&bytes).to_string();
                 let id = ulid::Ulid::new().to_string();
-                info!(id, path, "Received webhook");
+                debug!(id, path, "Received webhook");
                 let message =
                     WebhookMessage::new(id, path, /* query, headers, */ request_data);
                 recipient

src/server/config.rs

@@ -1,6 +1,7 @@
 use std::{
     collections::HashMap,
     fmt::{Display, Formatter},
+    ops::Deref,
 };
 
 use secrecy::ExposeSecret;
@@ -38,7 +39,6 @@ pub struct Webhook {
     url: String,
 }
 impl Webhook {
-    #[allow(dead_code)] // TODO: (#15) register webhook
     pub fn url(&self) -> WebhookUrl {
         WebhookUrl(self.url.clone())
     }
@@ -65,7 +65,7 @@ impl RepoConfig {
         toml::from_str(toml).map_err(OneOf::new)
     }
 
-    pub(crate) const fn branches(&self) -> &RepoBranches {
+    pub const fn branches(&self) -> &RepoBranches {
         &self.branches
     }
 }
@@ -285,6 +285,13 @@ impl Display for BranchName {
         write!(f, "{}", self.0)
     }
 }
+impl Deref for BranchName {
+    type Target = String;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
 
 /// The derived information about a repo, used to interact with it
 #[derive(Clone, Debug)]