f31ac15 refactor: combine event.rs and filter.rs — radicle-ci-broker

Radicle CI broker

refactor: combine event.rs and filter.rs

Lars Wirzenius committed 2 years ago

commit f31ac159e087e90d621aca6bd78c28f1b0decdd4
parent de92f988b34c6b3cd216147bdfb65b1e37aa54f9

7 files changed +330 -340

modified src/bin/broker-messages.rs

@@ -1,7 +1,9 @@

 use radicle::git::RefString;
 use radicle::Profile;
 use radicle_ci_broker::filter::BrokerEvent;
 use radicle_ci_broker::msg::{Oid, RepoId, Request, Response, RunId, RunResult};
 use radicle_ci_broker::{
     event::BrokerEvent,
     msg::{Oid, RepoId, Request, Response, RunId, RunResult},
 };
 fn main() {
     let rid = RepoId::from_urn("rad:zwTxygwuz5LDGBq255RA2CbNGrz8").expect("create rid");

modified src/bin/ci-broker.rs

@@ -11,8 +11,7 @@ use radicle::prelude::Profile;

 use radicle_ci_broker::{
     config::{Adapter, Config},
     error::BrokerError,
     event::NodeEventSource,
     filter::BrokerEvent,
     event::{BrokerEvent, NodeEventSource},
     msg::{Request, Response, RunResult},
 };

modified src/config.rs

@@ -8,7 +8,7 @@ use std::{

 use serde::{Deserialize, Serialize};
 use crate::{error::BrokerError, filter::EventFilter};
 use crate::{error::BrokerError, event::EventFilter};
 #[derive(Debug, Serialize, Deserialize)]
 pub struct Config {

modified src/event.rs

@@ -1,17 +1,47 @@

 //! Read events from the local Radicle node.
 //! Read node events from the local node, filter into broker events.
 //!
 //! [`NodeEventSource`] is an event listener.
 //!
 //! Events can be filtered based on various criteria and can be
 //! combined with logical operators. Filters can be read from a JSON
 //! file so that they're easy for a user to define.
 //!
 //! # Example
 //!
 //! ```
 //! use radicle_ci_broker::event::Filters;
 //! let filters = r#"{
 //!   "filters": [
 //!     {
 //!       "And": [
 //!         {
 //!           "Repository": "rad:z3bBRYgzcYYBNjipFdDTwPgHaihPX"
 //!         },
 //!         {
 //!           "RefSuffix": "refs/heads/main"
 //!         }
 //!       ]
 //!     }
 //!   ]
 //! }"#;
 //! let e = Filters::try_from(filters).unwrap();
 //! ```
 use std::{path::PathBuf, time};
 use std::{
     fs::read,
     path::{Path, PathBuf},
     time,
 };
 use log::{debug, info, trace};
 use radicle::{
     node::{Event, Handle},
     node::{Event, Handle, NodeId},
     prelude::RepoId,
     storage::RefUpdate,
     Profile,
 };
 use crate::filter::{BrokerEvent, EventFilter};
 use radicle_git_ext::{ref_format::RefString, Oid};
 use serde::{Deserialize, Serialize};
 /// Source of events from the local Radicle node.
 ///

@@ -108,3 +138,289 @@ pub enum NodeEventError {

     #[error("failed to parser filters as JSON")]
     FiltersJsonString(#[source] serde_json::Error),
 }
 /// An event filter for allowing events. Or an "AND" combination of events.
 ///
 /// NOTE: Adding "OR" and "NOT" would be easy, too.
 #[derive(Debug, Clone, Deserialize, Serialize)]
 #[serde(deny_unknown_fields)]
 pub enum EventFilter {
     /// Event concerns a specific repository.
     Repository(RepoId),
     /// Event concerns a git ref that ends with a given string.
     RefSuffix(String),
     /// Event concerns a specific git branch.
     Branch(String),
     /// Event concerns any Radicle patch.
     AnyPatch,
     /// Event concerns changes on specific Radicle patch.
     Patch(String),
     /// Event concerns changed refs on any Radicle patch branch.
     AnyPatchRef,
     /// Event concerns changed refs on the branch of the specified Radicle patch.
     PatchRef(String),
     /// Combine any number of filters that both must allow the events.
     And(Vec<Box<Self>>),
     /// Combine any number of filters such that at least one allows the events.
     Or(Vec<Box<Self>>),
     /// Combine any number of filters such that none allows the events.
     Not(Vec<Box<Self>>),
 }
 impl EventFilter {
     /// Create a filter for a repository.
     pub fn repository(rid: &str) -> Result<Self, NodeEventError> {
         Ok(Self::Repository(RepoId::from_urn(rid)?))
     }
     /// Create a filter for a git ref that ends with a string.
     pub fn glob(pattern: &str) -> Result<Self, NodeEventError> {
         Ok(Self::RefSuffix(pattern.into()))
     }
     /// Create a filter combining other filters with AND.
     pub fn and(conds: &[Self]) -> Self {
         Self::And(conds.iter().map(|c| Box::new(c.clone())).collect())
     }
     /// Create a filter combining other filters with OR.
     pub fn or(conds: &[Self]) -> Self {
         Self::Or(conds.iter().map(|c| Box::new(c.clone())).collect())
     }
     /// Create a filter combining other filters with NOT.
     pub fn not(conds: &[Self]) -> Self {
         Self::Not(conds.iter().map(|c| Box::new(c.clone())).collect())
     }
     /// Read filters from a JSON file.
     ///
     /// This function is the same as reading a file and calling
     /// [`Filters::try_from`], but returns just a vector of filters
     /// instead of a `Filter`.
     ///
     /// See the module description for an example of the file content.
     pub fn from_file(filename: &Path) -> Result<Vec<Self>, NodeEventError> {
         let filters =
             read(filename).map_err(|e| NodeEventError::ReadFilterFile(filename.into(), e))?;
         let filters: Filters = serde_json::from_slice(&filters)
             .map_err(|e| NodeEventError::FiltersJsonFile(filename.into(), e))?;
         Ok(filters.filters)
     }
 }
 /// A set of filters for [`NodeEventSource`](crate::event::NodeEventSource)to use. This struct
 /// represents the serialized set of filters. See the module
 /// description for an example.
 #[derive(Debug, Deserialize, Serialize)]
 pub struct Filters {
     filters: Vec<EventFilter>,
 }
 impl TryFrom<&str> for Filters {
     type Error = NodeEventError;
     fn try_from(s: &str) -> Result<Self, Self::Error> {
         serde_json::from_str(s).map_err(NodeEventError::FiltersJsonString)
     }
 }
 /// A single node event can represent many git refs having changed,
 /// but that's hard to process or filter. The broker breaks up such
 /// complex events to simpler ones that only affect one ref at a time.
 #[derive(Debug)]
 pub enum BrokerEvent {
     /// A git ref in a git repository has changed to refer to a given
     /// commit. This covers both the case of a new ref, and the case
     /// of a changed ref.
     RefChanged {
         /// Repository id.
         rid: RepoId,
         /// Ref name.
         name: RefString,
         /// New git commit.
         oid: Oid,
         /// Old git commit
         old: Option<Oid>,
     },
 }
 impl BrokerEvent {
     fn new(rid: &RepoId, name: &RefString, oid: &Oid, old: Option<Oid>) -> Self {
         Self::RefChanged {
             rid: *rid,
             name: name.clone(),
             oid: *oid,
             old,
         }
     }
     /// Break up a potentially complex node event into a vector of
     /// simpler broker events.
     pub fn from_event(e: &Event) -> Option<Vec<Self>> {
         if let Event::RefsFetched {
             remote: _,
             rid,
             updated,
         } = e
         {
             let mut events = vec![];
             for up in updated {
                 match up {
                     RefUpdate::Created { name, oid } => {
                         events.push(Self::new(rid, name, oid, None));
                     }
                     RefUpdate::Updated { name, old, new } => {
                         events.push(Self::new(rid, name, new, Some(*old)));
                     }
                     _ => (),
                 }
             }
             Some(events)
         } else {
             None
         }
     }
     /// Is this broker event allowed by a filter?
     pub fn is_allowed(&self, filter: &EventFilter) -> bool {
         let Self::RefChanged {
             rid,
             name,
             oid: _,
             old: _,
         } = self;
         match filter {
             EventFilter::Repository(wanted) => rid == wanted,
             EventFilter::RefSuffix(wanted) => name.ends_with(wanted),
             EventFilter::Branch(wanted) => name.ends_with(&format!("/refs/heads/{}", wanted)),
             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::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)),
         }
     }
     pub fn name(&self) -> Option<&RefString> {
         match self {
             BrokerEvent::RefChanged { name, .. } => Some(name),
         }
     }
     /// Extract the NID from the RefString.
     /// The RefString will start with `refs/namespaces/<nid>/...`
     pub fn nid(&self) -> Option<NodeId> {
         if let Some(name) = self.name() {
             let mut parts = name.split('/');
             if let Some(nid) = parts.nth(2) {
                 let parsed = nid.parse();
                 if parsed.is_ok() {
                     return parsed.ok();
                 }
             }
         }
         None
     }
     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();
             }
         }
         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);
         }
     }
     None
 }
 pub fn is_patch_ref(name: &str) -> Option<&str> {
     let mut parts = name.split("/refs/heads/patches/");
     if let Some(suffix) = parts.nth(1) {
         if parts.next().is_none() {
             return Some(suffix);
         }
     }
     None
 }
 #[cfg(test)]
 mod test {
     use super::{is_patch_ref, is_patch_update};
     #[test]
     fn branch_is_not_patch() {
         assert_eq!(
             is_patch_ref(
                 "refs/namespaces/z6MkuhvCnrcow7vzkyQzkuFixzpTa42iC2Cfa4DA8HRLCmys/refs/heads/main"
             ),
             None
         );
     }
     #[test]
     fn is_patch() {
         assert_eq!(
             is_patch_ref(
                 "refs/namespaces/z6MkuhvCnrcow7vzkyQzkuFixzpTa42iC2Cfa4DA8HRLCmys/refs/heads/patches/bbb54a2c9314a528a4fff9d6c2aae874ed098433"
             ),
             Some("bbb54a2c9314a528a4fff9d6c2aae874ed098433")
         );
     }
     #[test]
     fn branch_is_not_patch_update() {
         assert_eq!(
             is_patch_update(
                 "refs/namespaces/z6MkuhvCnrcow7vzkyQzkuFixzpTa42iC2Cfa4DA8HRLCmys/refs/heads/main"
             ),
             None
         );
     }
     #[test]
     fn patch_branch_is_not_patch_update() {
         assert_eq!(
             is_patch_update(
                 "refs/namespaces/z6MkuhvCnrcow7vzkyQzkuFixzpTa42iC2Cfa4DA8HRLCmys/refs/heads/patches/bbb54a2c9314a528a4fff9d6c2aae874ed098433"
             ),
             None
         );
     }
     #[test]
     fn patch_update() {
         assert_eq!(
             is_patch_update(
                 "refs/namespaces/z6MkuhvCnrcow7vzkyQzkuFixzpTa42iC2Cfa4DA8HRLCmys/refs/cobs/xyz.radicle.patch/bbb54a2c9314a528a4fff9d6c2aae874ed098433"
             ),
             Some("bbb54a2c9314a528a4fff9d6c2aae874ed098433")
         );
     }
 }

deleted src/filter.rs

@@ -1,326 +0,0 @@

 //! Filter for allowing node events.
 //!
 //! Events can be filtered based on various criteria and can be
 //! combined with an "and" operator. Filters can be read from a JSON
 //! file so that they're easy for a user to define.
 //!
 //! # Example
 //!
 //! ```
 //! use radicle_ci_broker::filter::Filters;
 //! let filters = r#"{
 //!   "filters": [
 //!     {
 //!       "And": [
 //!         {
 //!           "Repository": "rad:z3bBRYgzcYYBNjipFdDTwPgHaihPX"
 //!         },
 //!         {
 //!           "RefSuffix": "refs/heads/main"
 //!         }
 //!       ]
 //!     }
 //!   ]
 //! }"#;
 //! let e = Filters::try_from(filters).unwrap();
 //! ```
 use std::{fs::read, path::Path};
 use radicle::node::NodeId;
 use radicle::prelude::RepoId;
 use radicle::{node::Event, storage::RefUpdate};
 use radicle_git_ext::{ref_format::RefString, Oid};
 use serde::{Deserialize, Serialize};
 use crate::event::NodeEventError;
 /// An event filter for allowing events. Or an "AND" combination of events.
 ///
 /// NOTE: Adding "OR" and "NOT" would be easy, too.
 #[derive(Debug, Clone, Deserialize, Serialize)]
 #[serde(deny_unknown_fields)]
 pub enum EventFilter {
     /// Event concerns a specific repository.
     Repository(RepoId),
     /// Event concerns a git ref that ends with a given string.
     RefSuffix(String),
     /// Event concerns a specific git branch.
     Branch(String),
     /// Event concerns any Radicle patch.
     AnyPatch,
     /// Event concerns changes on specific Radicle patch.
     Patch(String),
     /// Event concerns changed refs on any Radicle patch branch.
     AnyPatchRef,
     /// Event concerns changed refs on the branch of the specified Radicle patch.
     PatchRef(String),
     /// Combine any number of filters that both must allow the events.
     And(Vec<Box<Self>>),
     /// Combine any number of filters such that at least one allows the events.
     Or(Vec<Box<Self>>),
     /// Combine any number of filters such that none allows the events.
     Not(Vec<Box<Self>>),
 }
 impl EventFilter {
     /// Create a filter for a repository.
     pub fn repository(rid: &str) -> Result<Self, NodeEventError> {
         Ok(Self::Repository(RepoId::from_urn(rid)?))
     }
     /// Create a filter for a git ref that ends with a string.
     pub fn glob(pattern: &str) -> Result<Self, NodeEventError> {
         Ok(Self::RefSuffix(pattern.into()))
     }
     /// Create a filter combining other filters with AND.
     pub fn and(conds: &[Self]) -> Self {
         Self::And(conds.iter().map(|c| Box::new(c.clone())).collect())
     }
     /// Create a filter combining other filters with OR.
     pub fn or(conds: &[Self]) -> Self {
         Self::Or(conds.iter().map(|c| Box::new(c.clone())).collect())
     }
     /// Create a filter combining other filters with NOT.
     pub fn not(conds: &[Self]) -> Self {
         Self::Not(conds.iter().map(|c| Box::new(c.clone())).collect())
     }
     /// Read filters from a JSON file.
     ///
     /// This function is the same as reading a file and calling
     /// [`Filters::try_from`], but returns just a vector of filters
     /// instead of a `Filter`.
     ///
     /// See the module description for an example of the file content.
     pub fn from_file(filename: &Path) -> Result<Vec<Self>, NodeEventError> {
         let filters =
             read(filename).map_err(|e| NodeEventError::ReadFilterFile(filename.into(), e))?;
         let filters: Filters = serde_json::from_slice(&filters)
             .map_err(|e| NodeEventError::FiltersJsonFile(filename.into(), e))?;
         Ok(filters.filters)
     }
 }
 fn _event_summary(event: &Event) -> Option<String> {
     match event {
         Event::RefsFetched { .. } => Some("RefsFetched".into()),
         _ => None,
     }
 }
 /// A set of filters for [`NodeEventSource`](crate::event::NodeEventSource)to use. This struct
 /// represents the serialized set of filters. See the module
 /// description for an example.
 #[derive(Debug, Deserialize, Serialize)]
 pub struct Filters {
     filters: Vec<EventFilter>,
 }
 impl TryFrom<&str> for Filters {
     type Error = NodeEventError;
     fn try_from(s: &str) -> Result<Self, Self::Error> {
         serde_json::from_str(s).map_err(NodeEventError::FiltersJsonString)
     }
 }
 /// A single node event can represent many git refs having changed,
 /// but that's hard to process or filter. The broker breaks up such
 /// complex events to simpler ones that only affect one ref at a time.
 #[derive(Debug)]
 pub enum BrokerEvent {
     /// A git ref in a git repository has changed to refer to a given
     /// commit. This covers both the case of a new ref, and the case
     /// of a changed ref.
     RefChanged {
         /// Repository id.
         rid: RepoId,
         /// Ref name.
         name: RefString,
         /// New git commit.
         oid: Oid,
         /// Old git commit
         old: Option<Oid>,
     },
 }
 impl BrokerEvent {
     fn new(rid: &RepoId, name: &RefString, oid: &Oid, old: Option<Oid>) -> Self {
         Self::RefChanged {
             rid: *rid,
             name: name.clone(),
             oid: *oid,
             old,
         }
     }
     /// Break up a potentially complex node event into a vector of
     /// simpler broker events.
     pub fn from_event(e: &Event) -> Option<Vec<Self>> {
         if let Event::RefsFetched {
             remote: _,
             rid,
             updated,
         } = e
         {
             let mut events = vec![];
             for up in updated {
                 match up {
                     RefUpdate::Created { name, oid } => {
                         events.push(Self::new(rid, name, oid, None));
                     }
                     RefUpdate::Updated { name, old, new } => {
                         events.push(Self::new(rid, name, new, Some(*old)));
                     }
                     _ => (),
                 }
             }
             Some(events)
         } else {
             None
         }
     }
     /// Is this broker event allowed by a filter?
     pub fn is_allowed(&self, filter: &EventFilter) -> bool {
         let Self::RefChanged {
             rid,
             name,
             oid: _,
             old: _,
         } = self;
         match filter {
             EventFilter::Repository(wanted) => rid == wanted,
             EventFilter::RefSuffix(wanted) => name.ends_with(wanted),
             EventFilter::Branch(wanted) => name.ends_with(&format!("/refs/heads/{}", wanted)),
             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::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)),
         }
     }
     pub fn name(&self) -> Option<&RefString> {
         match self {
             BrokerEvent::RefChanged { name, .. } => Some(name),
         }
     }
     /// Extract the NID from the RefString.
     /// The RefString will start with `refs/namespaces/<nid>/...`
     pub fn nid(&self) -> Option<NodeId> {
         if let Some(name) = self.name() {
             let mut parts = name.split('/');
             if let Some(nid) = parts.nth(2) {
                 let parsed = nid.parse();
                 if parsed.is_ok() {
                     return parsed.ok();
                 }
             }
         }
         None
     }
     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();
             }
         }
         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);
         }
     }
     None
 }
 pub fn is_patch_ref(name: &str) -> Option<&str> {
     let mut parts = name.split("/refs/heads/patches/");
     if let Some(suffix) = parts.nth(1) {
         if parts.next().is_none() {
             return Some(suffix);
         }
     }
     None
 }
 #[cfg(test)]
 mod test {
     use super::{is_patch_ref, is_patch_update};
     #[test]
     fn branch_is_not_patch() {
         assert_eq!(
             is_patch_ref(
                 "refs/namespaces/z6MkuhvCnrcow7vzkyQzkuFixzpTa42iC2Cfa4DA8HRLCmys/refs/heads/main"
             ),
             None
         );
     }
     #[test]
     fn is_patch() {
         assert_eq!(
             is_patch_ref(
                 "refs/namespaces/z6MkuhvCnrcow7vzkyQzkuFixzpTa42iC2Cfa4DA8HRLCmys/refs/heads/patches/bbb54a2c9314a528a4fff9d6c2aae874ed098433"
             ),
             Some("bbb54a2c9314a528a4fff9d6c2aae874ed098433")
         );
     }
     #[test]
     fn branch_is_not_patch_update() {
         assert_eq!(
             is_patch_update(
                 "refs/namespaces/z6MkuhvCnrcow7vzkyQzkuFixzpTa42iC2Cfa4DA8HRLCmys/refs/heads/main"
             ),
             None
         );
     }
     #[test]
     fn patch_branch_is_not_patch_update() {
         assert_eq!(
             is_patch_update(
                 "refs/namespaces/z6MkuhvCnrcow7vzkyQzkuFixzpTa42iC2Cfa4DA8HRLCmys/refs/heads/patches/bbb54a2c9314a528a4fff9d6c2aae874ed098433"
             ),
             None
         );
     }
     #[test]
     fn patch_update() {
         assert_eq!(
             is_patch_update(
                 "refs/namespaces/z6MkuhvCnrcow7vzkyQzkuFixzpTa42iC2Cfa4DA8HRLCmys/refs/cobs/xyz.radicle.patch/bbb54a2c9314a528a4fff9d6c2aae874ed098433"
             ),
             Some("bbb54a2c9314a528a4fff9d6c2aae874ed098433")
         );
     }
 }

modified src/lib.rs

@@ -8,5 +8,4 @@

 pub mod config;
 pub mod error;
 pub mod event;
 pub mod filter;
 pub mod msg;

modified src/msg.rs

@@ -25,7 +25,7 @@ use radicle::storage::{ReadRepository, ReadStorage};

 use radicle::{patch, Profile};
 use serde::{Deserialize, Serialize};
 use crate::filter::{is_patch_update, BrokerEvent};
 use crate::event::{is_patch_update, BrokerEvent};
 /// The type of a run identifier. For maximum generality, this is a
 /// string rather than an integer.

@@ -545,7 +545,7 @@ pub enum MessageError {

 #[cfg(test)]
 mod tests {
     use crate::filter::BrokerEvent;
     use crate::event::BrokerEvent;
     use crate::msg::Request;
     use radicle::crypto::ssh::Keystore;
     use radicle::crypto::test::signer::MockSigner;