Change the cause for patch events — radicle-ci-broker

radicle-ci-broker

rad:zwTxygwuz5LDGBq255RA2CbNGrz8

Radicle CI broker

Change the cause for patch events

Merged did:key:z6MkkpTP...arsB opened 2 years ago

Activity Changes +229 -148

5 files changed +229 -148 3a0ef23c → e3a90dfc

modified Cargo.lock

@@ -1643,6 +1643,7 @@ dependencies = [

  "radicle",
  "radicle-git-ext",
  "radicle-surf",
  "regex",
  "serde",
  "serde_json",
  "serde_yaml 0.9.34+deprecated",

modified Cargo.toml

@@ -26,6 +26,7 @@ subplotlib = "0.9.0"

 thiserror = "1.0.50"
 time = { version = "0.3.34", features = ["formatting", "macros"] }
 uuid = { version = "1.7.0", features = ["v4"] }
 regex = "1.10.4"
 [dependencies.radicle]
 version = "0.9.0"

modified src/bin/ci-broker.rs

@@ -9,6 +9,7 @@ use std::{

 use log::{debug, error, info};
 use radicle::prelude::Profile;
 use radicle_ci_broker::msg::MessageError;
 use radicle_ci_broker::{
     adapter::Adapter,
     broker::Broker,

@@ -112,12 +113,23 @@ fn fallible_main() -> Result<(), BrokerError> {

                 BrokerEvent::Shutdown => break 'event_loop,
                 BrokerEvent::RefChanged { .. } => {
                     info!("broker event {e:#?}");
                     let req = RequestBuilder::default()
                     let result = RequestBuilder::default()
                         .profile(&profile)
                         .broker_event(&e)
                         .build_trigger()?;
                     if let Err(e) = broker.execute_ci(&req, &mut page) {
                         error!("failed to run adapter, or adapter failed to run CI: {e}");
                         .build_trigger();
                     match result {
                         Ok(req) => {
                             if let Err(e) = broker.execute_ci(&req, &mut page) {
                                 error!("failed to run adapter, or adapter failed to run CI: {e}");
                             }
                         }
                         Err(MessageError::NoEventHandler) => {
                             debug!("no handler found for the specific event");
                             continue;
                         }
                         Err(e) => {
                             return Err(e.into());
                         }
                     }
                 }
             }

modified src/event.rs

@@ -27,12 +27,6 @@

 //! let e = Filters::try_from(filters).unwrap();
 //! ```
 use std::{
     fs::read,
     path::{Path, PathBuf},
     time,
 };
 use log::{debug, info, trace};
 use radicle::{
     node::{Event, Handle, NodeId},

@@ -41,7 +35,13 @@ use radicle::{

     Profile,
 };
 use radicle_git_ext::{ref_format::RefString, Oid};
 use regex::Regex;
 use serde::{Deserialize, Serialize};
 use std::{
     fs::read,
     path::{Path, PathBuf},
     time,
 };
 /// Source of events from the local Radicle node.
 ///

@@ -222,6 +222,9 @@ pub enum EventFilter {

     /// Event concerns changed refs on any Radicle patch branch.
     AnyPatchRef,
     /// Event concerns changed refs on any Radicle branch.
     AnyPushRef,
     /// Event concerns changed refs on the branch of the specified Radicle patch.
     PatchRef(String),

@@ -393,8 +396,18 @@ impl BrokerEvent {

             }
             EventFilter::AnyPatch => is_patch_update(name).is_some(),
             EventFilter::Patch(wanted) => is_patch_update(name) == Some(wanted),
             EventFilter::AnyPatchRef => is_patch_ref(name).is_some(),
             EventFilter::PatchRef(wanted) => is_patch_ref(name) == Some(wanted),
             EventFilter::AnyPatchRef => {
                 matches!(parse_ref(name), Some(ParsedRef::Patch(_)))
             }
             EventFilter::AnyPushRef => {
                 matches!(parse_ref(name), Some(ParsedRef::Push(_)))
             }
             EventFilter::PatchRef(wanted) => {
                 if let Some(ParsedRef::Patch(ref_oid)) = parse_ref(name) {
                     return ref_oid == Oid::try_from(wanted.as_str()).unwrap();
                 }
                 false
             }
             EventFilter::And(conds) => conds.iter().all(|cond| self.is_allowed(cond)),
             EventFilter::Or(conds) => conds.iter().any(|cond| self.is_allowed(cond)),
             EventFilter::Not(conds) => !conds.iter().any(|cond| self.is_allowed(cond)),

@@ -432,34 +445,62 @@ impl BrokerEvent {

     pub fn patch_id(&self) -> Option<Oid> {
         if let Some(name) = self.name() {
             let suffix = is_patch_update(name);
             if let Some(suffix_str) = suffix {
                 return suffix_str.parse().ok();
             if let Some(ParsedRef::Patch(oid)) = parse_ref(name) {
                 return Some(oid);
             }
         }
         None
     }
 }
 pub fn is_patch_update(name: &str) -> Option<&str> {
     let mut parts = name.split("/refs/cobs/xyz.radicle.patch/");
     if let Some(suffix) = parts.nth(1) {
         if parts.next().is_none() {
             return Some(suffix);
 /// Parsed reference to one of the supported types
 /// Patch with patch ID
 /// Push with branch name
 pub enum ParsedRef {
     Patch(Oid),
     Push(String),
 }
 /// Parse the given reference to a ParsedRef.
 ///
 /// # Example
 /// ```Rust
 /// if let Some(parsed_ref) = parse_ref(name) {
 ///     match parsed_ref {
 ///         ParsedRef::Patch(_oid) => {
 ///             debug!("build_trigger: is patch");
 ///         }
 ///         ParsedRef::Push(_branch) => {
 ///             debug!("build_trigger: is push");
 ///         }
 ///     }
 /// }
 /// ```
 pub fn parse_ref(s: &str) -> Option<ParsedRef> {
     let patch_re = Regex::new(r"^refs/namespaces/[^/]+/refs/heads/patches/([^/]+)$").unwrap();
     if let Some(patch_captures) = patch_re.captures(s) {
         if let Some(patch_id) = patch_captures.get(1) {
             let patch_id_str = patch_id.as_str();
             let oid = Oid::try_from(patch_id_str).unwrap();
             return Some(ParsedRef::Patch(oid));
         }
     }
     let push_re = Regex::new(r"^refs/namespaces/[^/]+/refs/heads/([^/]+)$").unwrap();
     if let Some(push_captures) = push_re.captures(s) {
         if let Some(branch) = push_captures.get(1) {
             return Some(ParsedRef::Push(branch.as_str().to_string()));
         }
     }
     None
 }
 pub fn is_patch_ref(name: &str) -> Option<&str> {
     let mut parts = name.split("/refs/heads/patches/");
 pub fn is_patch_update(name: &str) -> Option<&str> {
     let mut parts = name.split("/refs/cobs/xyz.radicle.patch/");
     if let Some(suffix) = parts.nth(1) {
         if parts.next().is_none() {
             return Some(suffix);
         }
     }
     None
 }

@@ -475,26 +516,43 @@ pub fn push_branch(name: &str) -> String {

 #[cfg(test)]
 mod test {
     use super::{is_patch_ref, is_patch_update, push_branch};
     use super::{is_patch_update, parse_ref, push_branch, ParsedRef};
     #[test]
     fn branch_is_not_patch() {
         assert_eq!(
             is_patch_ref(
                 "refs/namespaces/z6MkuhvCnrcow7vzkyQzkuFixzpTa42iC2Cfa4DA8HRLCmys/refs/heads/main"
             ),
             None
         );
     fn test_parse_patch_ref() {
         let patch_ref =
             "refs/namespaces/NID/refs/heads/patches/9183ed6232687d3105482960cecb01a53018b80a";
         let parsed_ref = parse_ref(patch_ref);
         assert!(parsed_ref.is_some());
         if let Some(ref parsed) = parsed_ref {
             match parsed {
                 ParsedRef::Patch(oid) => {
                     assert_eq!(oid.to_string(), "9183ed6232687d3105482960cecb01a53018b80a")
                 }
                 _ => panic!("Expected Patch ref"),
             }
         }
     }
     #[test]
     fn is_patch() {
         assert_eq!(
             is_patch_ref(
                 "refs/namespaces/z6MkuhvCnrcow7vzkyQzkuFixzpTa42iC2Cfa4DA8HRLCmys/refs/heads/patches/bbb54a2c9314a528a4fff9d6c2aae874ed098433"
             ),
             Some("bbb54a2c9314a528a4fff9d6c2aae874ed098433")
         );
     fn test_parse_push_ref() {
         let push_ref =
             "refs/namespaces/z6MkuhvCnrcow7vzkyQzkuFixzpTa42iC2Cfa4DA8HRLCmys/refs/heads/main";
         let parsed_ref = parse_ref(push_ref);
         assert!(parsed_ref.is_some());
         if let Some(ref parsed) = parsed_ref {
             match parsed {
                 ParsedRef::Push(branch) => assert_eq!(branch, "main"),
                 _ => panic!("Expected Push ref"),
             }
         }
     }
     #[test]
     fn test_parse_invalid_ref() {
         let invalid_ref = "invalid_ref";
         let parsed_ref = parse_ref(invalid_ref);
         assert!(parsed_ref.is_none());
     }
     #[test]

modified src/msg.rs

@@ -32,7 +32,7 @@ use radicle::{

     Profile,
 };
 use crate::event::{is_patch_update, push_branch, BrokerEvent};
 use crate::event::{parse_ref, push_branch, BrokerEvent, ParsedRef};
 // This gets put into every [`Request`] message so the adapter can
 // detect its getting a message it knows how to handle.

@@ -138,7 +138,6 @@ impl<'a> RequestBuilder<'a> {

             } => (rid, name, oid, old),
         };
         debug!("build_trigger: unpacked event");
         let is_patch = is_patch_update(name).is_some();
         let repository = profile.storage.repository(*rid)?;
         debug!("build_trigger: got repository");
         let storage = &profile.storage;

@@ -160,116 +159,122 @@ impl<'a> RequestBuilder<'a> {

         let patch_info: Option<PatchEvent>;
         let event_type: EventType;
         debug!("build_trigger: checking if patch or push");
         if is_patch {
             debug!("build_trigger: is patch");
             event_type = EventType::Patch;
             let patch_id = event.patch_id().ok_or(MessageError::Trigger)?;
             let patch = patch::Patches::open(&repository)?
                 .get(&patch_id.into())?
                 .ok_or(MessageError::Trigger)?;
             push_info = None;
             let revs: Vec<Revision> = patch
                 .revisions()
                 .map(|(rid, r)| {
                     Ok::<Revision, MessageError>(Revision {
                         id: rid.into(),
                         author: did_to_author(profile, r.author().id())?,
                         description: r.description().to_string(),
                         base: *r.base(),
                         oid: r.head(),
                         timestamp: r.timestamp().as_secs(),
                     })
                 })
                 .collect::<Result<Vec<Revision>, MessageError>>()?;
             let patch_author_pk = radicle::crypto::PublicKey::from(author.id);
             let patch_latest_revision = patch
                 .latest_by(&patch_author_pk)
                 .ok_or(MessageError::Trigger)?;
             let patch_head = patch_latest_revision.1.head();
             let patch_base = patch_latest_revision.1.base();
             let patch_commits: Vec<Oid> = repo
                 .history(patch_head)?
                 .take_while(|c| {
                     if let Ok(c) = c {
                         c.id != *patch_base
         if let Some(parsed_ref) = parse_ref(name) {
             match parsed_ref {
                 ParsedRef::Patch(_oid) => {
                     debug!("build_trigger: is patch");
                     event_type = EventType::Patch;
                     let patch_id = event.patch_id().ok_or(MessageError::Trigger)?;
                     let patch = patch::Patches::open(&repository)?
                         .get(&patch_id.into())?
                         .ok_or(MessageError::Trigger)?;
                     push_info = None;
                     let revs: Vec<Revision> = patch
                         .revisions()
                         .map(|(rid, r)| {
                             Ok::<Revision, MessageError>(Revision {
                                 id: rid.into(),
                                 author: did_to_author(profile, r.author().id())?,
                                 description: r.description().to_string(),
                                 base: *r.base(),
                                 oid: r.head(),
                                 timestamp: r.timestamp().as_secs(),
                             })
                         })
                         .collect::<Result<Vec<Revision>, MessageError>>()?;
                     let patch_author_pk = radicle::crypto::PublicKey::from(author.id);
                     let patch_latest_revision = patch
                         .latest_by(&patch_author_pk)
                         .ok_or(MessageError::Trigger)?;
                     let patch_head = patch_latest_revision.1.head();
                     let patch_base = patch_latest_revision.1.base();
                     let patch_commits: Vec<Oid> = repo
                         .history(patch_head)?
                         .take_while(|c| {
                             if let Ok(c) = c {
                                 c.id != *patch_base
                             } else {
                                 false
                             }
                         })
                         .map(|r| r.map(|c| c.id))
                         .collect::<Result<Vec<Oid>, _>>()?;
                     let patch_action = if patch.revisions().count() > 1 {
                         "updated"
                     } else {
                         false
                     }
                 })
                 .map(|r| r.map(|c| c.id))
                 .collect::<Result<Vec<Oid>, _>>()?;
             let patch_action = if patch.revisions().count() > 1 {
                 "updated"
             } else {
                 "created"
             };
             patch_info = Some(PatchEvent {
                 action: PatchAction::try_from(patch_action)?,
                 patch: Patch {
                     id: patch_id,
                     author,
                     title: patch.title().to_string(),
                     state: State {
                         status: patch.state().to_string(),
                         conflicts: match patch.state() {
                             patch::State::Open { conflicts, .. } => conflicts.to_vec(),
                             _ => vec![],
                         "created"
                     };
                     patch_info = Some(PatchEvent {
                         action: PatchAction::try_from(patch_action)?,
                         patch: Patch {
                             id: patch_id,
                             author,
                             title: patch.title().to_string(),
                             state: State {
                                 status: patch.state().to_string(),
                                 conflicts: match patch.state() {
                                     patch::State::Open { conflicts, .. } => conflicts.to_vec(),
                                     _ => vec![],
                                 },
                             },
                             before: *patch_base,
                             after: patch_head,
                             commits: patch_commits,
                             target: patch.target().head(&repository)?,
                             labels: patch.labels().map(|l| l.name().to_string()).collect(),
                             assignees: patch.assignees().collect(),
                             revisions: revs,
                         },
                     },
                     before: *patch_base,
                     after: patch_head,
                     commits: patch_commits,
                     target: patch.target().head(&repository)?,
                     labels: patch.labels().map(|l| l.name().to_string()).collect(),
                     assignees: patch.assignees().collect(),
                     revisions: revs,
                     });
                 }
                 ParsedRef::Push(_branch) => {
                     debug!("build_trigger: is push");
                     event_type = EventType::Push;
                     let before_oid: Oid = old.unwrap_or(*oid);
                     let push_commits: Vec<Oid> = repo
                         .history(oid)?
                         .take_while(|c| {
                             if let Ok(c) = c {
                                 c.id != before_oid
                             } else {
                                 false
                             }
                         })
                         .map(|r| r.map(|c| c.id))
                         .collect::<Result<Vec<Oid>, _>>()?;
                     push_info = Some(PushEvent {
                         pusher: author,
                         before: before_oid,
                         after: *oid,
                         branch: push_branch(name),
                         commits: push_commits,
                     });
                     patch_info = None;
                 }
             }
             let common = EventCommonFields {
                 version: PROTOCOL_VERSION,
                 event_type,
                 repository: Repository {
                     id: *rid,
                     name: repo_project.name().to_string(),
                     description: repo_project.description().to_string(),
                     private: !repo_identity.visibility.is_public(),
                     default_branch: repo_project.default_branch().to_string(),
                     delegates: repository.delegates()?.iter().copied().collect(),
                 },
             });
         } else {
             debug!("build_trigger: is push");
             event_type = EventType::Push;
             let before_oid: Oid = old.unwrap_or(*oid);
             let push_commits: Vec<Oid> = repo
                 .history(oid)?
                 .take_while(|c| {
                     if let Ok(c) = c {
                         c.id != before_oid
                     } else {
                         false
                     }
                 })
                 .map(|r| r.map(|c| c.id))
                 .collect::<Result<Vec<Oid>, _>>()?;
             push_info = Some(PushEvent {
                 pusher: author,
                 before: before_oid,
                 after: *oid,
                 branch: push_branch(name),
                 commits: push_commits,
             });
             patch_info = None;
         };
         let common = EventCommonFields {
             version: PROTOCOL_VERSION,
             event_type,
             repository: Repository {
                 id: *rid,
                 name: repo_project.name().to_string(),
                 description: repo_project.description().to_string(),
                 private: !repo_identity.visibility.is_public(),
                 default_branch: repo_project.default_branch().to_string(),
                 delegates: repository.delegates()?.iter().copied().collect(),
             },
         };
             };
         debug!("build_trigger: return Ok");
         Ok(Request::Trigger {
             common,
             push: push_info,
             patch: patch_info,
         })
             debug!("build_trigger: return Ok");
             Ok(Request::Trigger {
                 common,
                 push: push_info,
                 patch: patch_info,
             })
         } else {
             Err(MessageError::NoEventHandler)
         }
     }
 }

@@ -692,6 +697,10 @@ pub enum MessageError {

     #[error("RequestBuilder must have broker event set")]
     NoEvent,
     /// [`NoEventHandler`] does not have event handler set.
     #[error("RequestBuilder has no event handler set")]
     NoEventHandler,
     /// Failed to serialize a request message as JSON. This should
     /// never happen and likely indicates a programming failure.
     #[error("failed to serialize a request into JSON to a file handle")]

@@ -847,7 +856,7 @@ pub mod tests {

         let be = BrokerEvent::RefChanged {
             rid: project.id,
             name: RefString::try_from(
                 "refs/namespaces/$nid/refs/cobs/xyz.radicle.patch/$patchId"
                 "refs/namespaces/$nid/refs/heads/patches/$patchId"
                     .replace("$nid", &profile.id().to_string())
                     .replace("$patchId", &patch_cob.id.to_string()),
             )?,