41375b8 feat: change cibtool subcommands to use new CI events — radicle-ci-broker

modified src/bin/cibtool.rs

@@ -7,6 +7,7 @@

 //! suite (see the `ci-broker.subplot` document).
 #![allow(clippy::result_large_err)]
 #![allow(unused_imports)] // FIXME
 use std::{
     error::Error,

@@ -29,7 +30,8 @@ use radicle_git_ext::Oid;

 use radicle_ci_broker::{
     broker::BrokerError,
     db::{Db, DbError, QueueId, QueuedEvent},
     ci_event::{CiEvent, CiEventError},
     db::{Db, DbError, QueueId, QueuedCiEvent},
     event::{BrokerEvent, BrokerEventError},
     logger,
     msg::{RunId, RunResult},

@@ -182,7 +184,7 @@ impl Subcommand for EventCmd {

             EventSubCmd::Show(x) => x.run(args)?,
             EventSubCmd::Remove(x) => x.run(args)?,
             EventSubCmd::Record(x) => x.run(args)?,
             EventSubCmd::Broker(x) => x.run(args)?,
             EventSubCmd::Ci(x) => x.run(args)?,
             EventSubCmd::Filter(x) => x.run(args)?,
         }
         Ok(())

@@ -200,7 +202,7 @@ enum EventSubCmd {

     Show(cibtoolcmd::ShowEvent),
     Remove(cibtoolcmd::RemoveEvent),
     Record(cibtoolcmd::RecordEvents),
     Broker(cibtoolcmd::BrokerEvents),
     Ci(cibtoolcmd::CiEvents),
     Filter(cibtoolcmd::FilterEvents),
 }

@@ -274,8 +276,8 @@ enum CibToolError {

     #[error(transparent)]
     Db(#[from] DbError),
     #[error("failed to serialize broker event to JSON: {0:#?}")]
     EventToJson(BrokerEvent, #[source] serde_json::Error),
     #[error("failed to serialize CI event to JSON: {0:#?}")]
     EventToJson(CiEvent, #[source] serde_json::Error),
     #[error("failed to serialize node event to JSON: {0:#?}")]
     NodeEevnetToJson(radicle::node::Event, #[source] serde_json::Error),

@@ -344,11 +346,8 @@ enum CibToolError {

     CreateBrokerEventsFile(PathBuf, #[source] std::io::Error),
     #[error("failed to read filters from YAML file {0}")]
     ReadFilters(
         PathBuf,
         #[source] radicle_ci_broker::event::BrokerEventError,
     ),
     ReadFilters(PathBuf, #[source] radicle_ci_broker::filter::FilterError),
     #[error("failed to check if event is allowed: {0:#?}")]
     EventIsAllowed(BrokerEvent, #[source] BrokerEventError),
     #[error("failed to construct a CiEvent::BranchCreated")]
     BranchCreted(#[source] CiEventError),
 }

modified src/bin/cibtoolcmd/event.rs

@@ -1,6 +1,7 @@

 use std::io::Write;
 use radicle_ci_broker::{event::EventFilter, node_event_source::NodeEventSource};
 #[allow(unused_imports)] // FIXME
 use radicle_ci_broker::{filter::EventFilter, node_event_source::NodeEventSource};
 use super::*;

@@ -19,11 +20,11 @@ pub struct ListEvents {

 impl Leaf for ListEvents {
     fn run(&self, args: &Args) -> Result<(), CibToolError> {
         let db = args.open_db()?;
         let event_ids = db.queued_events()?;
         let event_ids = db.queued_ci_events()?;
         if self.json {
             let events: Result<Vec<QueuedEvent>, DbError> = event_ids
             let events: Result<Vec<QueuedCiEvent>, DbError> = event_ids
                 .iter()
                 .filter_map(|id| match db.get_queued_event(id) {
                 .filter_map(|id| match db.get_queued_ci_event(id) {
                     Ok(Some(event)) => Some(Ok(event)),
                     Err(e) => Some(Err(e)),
                     _ => None,

@@ -35,7 +36,7 @@ impl Leaf for ListEvents {

             println!("{}", json);
         } else if self.verbose {
             for id in event_ids {
                 if let Some(e) = db.get_queued_event(&id)? {
                 if let Some(e) = db.get_queued_ci_event(&id)? {
                     println!("{id}: {:?}", e);
                 } else {
                     println!("{id}: No such event");

@@ -57,7 +58,7 @@ pub struct CountEvents {}

 impl Leaf for CountEvents {
     fn run(&self, args: &Args) -> Result<(), CibToolError> {
         let db = args.open_db()?;
         println!("{}", db.queued_events()?.len());
         println!("{}", db.queued_ci_events()?.len());
         Ok(())
     }
 }

@@ -148,6 +149,8 @@ impl AddEvent {

 impl Leaf for AddEvent {
     fn run(&self, args: &Args) -> Result<(), CibToolError> {
         let nid = self.lookup_nid()?;
         let rid = if let Ok(rid) = RepoId::from_urn(&self.repo) {
             rid
         } else {

@@ -160,15 +163,16 @@ impl Leaf for AddEvent {

             self.lookup_commit(rid, &self.commit)?
         };
         let name = format!(
             "refs/namespaces/{}/refs/heads/{}",
             self.lookup_nid()?,
             self.name.as_str()
         );
         let name = format!("refs/namespaces/{}/refs/heads/{}", nid, self.name.as_str());
         let name =
             RefString::try_from(name.clone()).map_err(|e| CibToolError::RefString(name, e))?;
         let event = BrokerEvent::new(&rid, &name, &oid, self.base);
         let event = if let Some(base) = &self.base {
             CiEvent::branch_updated(nid, rid, &name, oid, *base)
                 .map_err(CibToolError::BranchCreted)?
         } else {
             CiEvent::branch_created(nid, rid, &name, oid).map_err(CibToolError::BranchCreted)?
         };
         if let Some(output) = &self.output {
             let json = serde_json::to_string_pretty(&event)

@@ -177,7 +181,7 @@ impl Leaf for AddEvent {

                 .map_err(|e| CibToolError::Write(output.into(), e))?;
         } else {
             let db = args.open_db()?;
             let id = db.push_queued_event(event)?;
             let id = db.push_queued_ci_event(event)?;
             println!("{id}");
             if let Some(filename) = &self.id_file {

@@ -224,7 +228,7 @@ impl Leaf for ShowEvent {

             return Err(CibToolError::MissingId);
         };
         if let Some(event) = db.get_queued_event(&id)? {
         if let Some(event) = db.get_queued_ci_event(&id)? {
             if self.json {
                 let json = serde_json::to_string_pretty(&event.event())
                     .map_err(|e| CibToolError::EventToJson(event.event().clone(), e))?;

@@ -268,7 +272,7 @@ impl Leaf for RemoveEvent {

             return Err(CibToolError::MissingId);
         };
         db.remove_queued_event(&id)?;
         db.remove_queued_ci_event(&id)?;
         Ok(())
     }
 }

@@ -287,7 +291,7 @@ pub struct Shutdown {

 impl Leaf for Shutdown {
     fn run(&self, args: &Args) -> Result<(), CibToolError> {
         let db = args.open_db()?;
         let id = db.push_queued_event(BrokerEvent::Shutdown)?;
         let id = db.push_queued_ci_event(CiEvent::Shutdown)?;
         if let Some(filename) = &self.id_file {
             write(filename, id.to_string().as_bytes())

@@ -346,16 +350,16 @@ impl Leaf for RecordEvents {

     }
 }
 /// Convert node events into broker events.
 /// Convert node events into CI events.
 ///
 /// Node events are read from the specified file. Note that one node
 /// event can result in any number of broker events.
 ///
 /// The events are written to the standard output or to the specified
 /// file, as one JSON object per line.
 /// The CI events are written to the standard output or to the
 /// specified file, as one JSON object per line.
 #[derive(Parser)]
 pub struct BrokerEvents {
     /// Write broker events to this file.
 pub struct CiEvents {
     /// Write CI events to this file.
     #[clap(long)]
     output: Option<PathBuf>,

@@ -363,7 +367,7 @@ pub struct BrokerEvents {

     input: PathBuf,
 }
 impl Leaf for BrokerEvents {
 impl Leaf for CiEvents {
     fn run(&self, _args: &Args) -> Result<(), CibToolError> {
         let bytes = std::fs::read(&self.input)
             .map_err(|e| CibToolError::ReadEvents(self.input.clone(), e))?;

@@ -377,18 +381,18 @@ impl Leaf for BrokerEvents {

             node_events.push(event);
         }
         let mut broker_events = vec![];
         let mut ci_events: Vec<CiEvent> = vec![];
         for node_event in node_events.iter() {
             if let Some(mut bes) = BrokerEvent::from_event(node_event) {
                 broker_events.append(&mut bes);
             if let Ok(mut cevs) = CiEvent::from_node_event(node_event) {
                 ci_events.append(&mut cevs);
             }
         }
         let mut jsons = vec![];
         for be in broker_events.iter() {
         for ci_event in ci_events.iter() {
             jsons.push(
                 serde_json::to_string(be)
                     .map_err(|err| CibToolError::EventToJson(be.clone(), err))?,
                 serde_json::to_string(ci_event)
                     .map_err(|err| CibToolError::EventToJson(ci_event.clone(), err))?,
             );
         }

@@ -426,7 +430,7 @@ pub struct FilterEvents {

 impl Leaf for FilterEvents {
     fn run(&self, _args: &Args) -> Result<(), CibToolError> {
         let filters = EventFilter::from_yaml_file(&self.filters)
         let filters = EventFilter::from_file(&self.filters)
             .map_err(|e| CibToolError::ReadFilters(self.filters.clone(), e))?;
         let bytes = std::fs::read(&self.input)

@@ -434,19 +438,16 @@ impl Leaf for FilterEvents {

         let text = String::from_utf8(bytes)
             .map_err(|e| CibToolError::BrokerEventNotUtf8(self.input.clone(), e))?;
         let mut broker_events = vec![];
         let mut ci_events = vec![];
         for line in text.lines() {
             let event: BrokerEvent = serde_json::from_str(line)
             let event: CiEvent = serde_json::from_str(line)
                 .map_err(|e| CibToolError::JsonToNodeEvent(self.input.clone(), e))?;
             broker_events.push(event);
             ci_events.push(event);
         }
         for event in broker_events.iter() {
         for event in ci_events.iter() {
             for filter in filters.iter() {
                 if event
                     .is_allowed(filter)
                     .map_err(|e| CibToolError::EventIsAllowed(event.clone(), e))?
                 {
                 if filter.allows(event) {
                     let json = serde_json::to_string_pretty(event)
                         .map_err(|e| CibToolError::EventToJson(event.clone(), e))?;
                     println!("{json}");

modified src/bin/cibtoolcmd/trigger.rs

@@ -35,10 +35,11 @@ impl Leaf for TriggerCmd {

         let name = format!("refs/namespaces/{nid}/refs/heads/{}", self.name.as_str());
         let name =
             RefString::try_from(name.clone()).map_err(|e| CibToolError::RefString(name, e))?;
         let event = BrokerEvent::new(&rid, &name, &oid, Some(base));
         let event = CiEvent::branch_updated(nid, rid, &name, oid, base)
             .map_err(CibToolError::BranchCreted)?;
         let db = args.open_db()?;
         let id = db.push_queued_event(event)?;
         let id = db.push_queued_ci_event(event)?;
         println!("{id}");
         if let Some(filename) = &self.id_file {