checksum = "b3d1d046238990b9cf5bcde22a3fb3584ee5cf65fb2765f454ed428c7a0063da"

[[package]]

name = "arraydeque"

version = "0.5.1"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "f3b7eb4404b8195a9abb6356f4ac07d8ba267045c8d6d220ac4dc992e6cc75df"

[[package]]

name = "ctr"

version = "0.9.2"

source = "registry+https://github.com/rust-lang/crates.io-index"

]

[[package]]

name = "env_logger"

version = "0.10.2"

source = "registry+https://github.com/rust-lang/crates.io-index"

]

[[package]]

name = "equivalent"

version = "1.0.1"

source = "registry+https://github.com/rust-lang/crates.io-index"

dependencies = [

 "anyhow",

 "clap",

 "culpa",

 "html-page",

 "radicle",

 "radicle-git-ext",

 "radicle-surf",

 "serde_yml",

 "slog",

 "slog-json",

 "sqlite",

 "sqlite3-sys",

 "subplot-build",

]

[[package]]

name = "slug"

version = "0.1.6"

source = "registry+https://github.com/rust-lang/crates.io-index"

 "anyhow",

 "base64 0.22.1",

 "clap",

 "file_diff",

 "git-testament",

 "html-escape",

[dependencies]

anyhow = "1.0.86"

clap = { version = "4.5.11", features = ["derive", "wrap_help"] }

html-page = "0.4.0"

radicle-git-ext = "0.7.0"

radicle-surf = { version = "0.18.0", default-features = false, features = ["serde"] }

regex = "1.10.5"

serde_yml = "0.0.11"

slog = "2.7.0"

slog-json = "2.6.1"

sqlite = "0.32.0"

sqlite3-sys = "0.15.0"

subplotlib = "0.10.0"

features = ["default", "test"]

[dev-dependencies]

culpa = "1.0.2"

tempfile = { version = "3.10.1" }

    process::{Command, Stdio},

};

use crate::{

    db::{Db, DbError},

    msg::{MessageError, Request, Response},

    run::{Run, RunState},

};

    }

    pub fn run(&self, trigger: &Request, run: &mut Run, db: &Db) -> Result<(), AdapterError> {

        run.set_state(RunState::Triggered);

        db.update_run(run).map_err(AdapterError::UpdateRun)?;

        assert!(matches!(trigger, Request::Trigger { .. }));

        // Spawn the adapter sub-process.

        let mut child = Command::new(&self.bin)

            .stdin(Stdio::piped())

            .stdout(Stdio::piped())

        // Write the request to trigger a run to the child's stdin.

        // Then close the pipe to prevent the child from trying to

        // read another message that will never be sent.

        {

            let stdin = child.stdin.take().ok_or(AdapterError::StdinHandle)?;

            trigger

        if let Some(line) = lines.next() {

            let line = line.map_err(AdapterError::ReadLine)?;

            let resp = Response::from_str(&line).map_err(AdapterError::ParseResponse)?;

            match resp {

                Response::Triggered { run_id, info_url } => {

                    run.set_state(RunState::Running);

                _ => return Err(AdapterError::NotTriggered(resp)),

            }

        } else {

        }

        if let Some(line) = lines.next() {

            let line = line.map_err(AdapterError::ReadLine)?;

            let resp = Response::from_str(&line).map_err(AdapterError::ParseResponse)?;

            match resp {

                Response::Finished { result } => {

                    run.set_result(result);

                _ => return Err(AdapterError::NotFinished(resp)),

            }

        } else {

        }

        if let Some(line) = lines.next() {

            let line = line.map_err(AdapterError::ReadLine)?;

            let resp = Response::from_str(&line).map_err(AdapterError::ParseResponse)?;

            return Err(AdapterError::TooMany(resp));

        }

        let wait = child.wait().map_err(AdapterError::Wait)?;

        if let Some(exit) = wait.code() {

            if exit != 0 {

                let mut stderr = child.stderr.take().ok_or(AdapterError::StderrHandle)?;

                let mut buf = vec![];

                    .read_to_end(&mut buf)

                    .map_err(AdapterError::ReadStderr)?;

                let stderr = String::from_utf8_lossy(&buf);

                return Err(AdapterError::Failed(exit));

            }

        } else {

            return Err(AdapterError::Failed(NOT_EXITED));

        }

        adapter::AdapterError,

        msg::{MessageError, Response, RunResult},

        run::Whence,

    };

    fn db() -> anyhow::Result<Db> {

    #[test]

    fn adapter_reports_success() -> TestResult<()> {

        const ADAPTER: &str = r#"#!/bin/bash

read

echo '{"response":"triggered","run_id":{"id":"xyzzy"}}'

    #[test]

    fn adapter_reports_failure() -> TestResult<()> {

        const ADAPTER: &str = r#"#!/bin/bash

read

echo '{"response":"triggered","run_id":{"id":"xyzzy"}}'

    #[test]

    fn adapter_exits_nonzero() -> TestResult<()> {

        const ADAPTER: &str = r#"#!/bin/bash

read

echo '{"response":"triggered","run_id":{"id":"xyzzy"}}'

    #[test]

    fn adapter_is_killed_before_any_messages() -> TestResult<()> {

        const ADAPTER: &str = r#"#!/bin/bash

kill -9 $BASHPID

"#;

    #[test]

    fn adapter_is_killed_after_first_message() -> TestResult<()> {

        const ADAPTER: &str = r#"#!/bin/bash

read

echo '{"response":"triggered","run_id":{"id":"xyzzy"}}'

    #[test]

    fn adapter_is_killed_after_second_message() -> TestResult<()> {

        const ADAPTER: &str = r#"#!/bin/bash

read

echo '{"response":"triggered","run_id":{"id":"xyzzy"}}'

    #[test]

    fn adapter_produces_as_bad_message() -> TestResult<()> {

        const ADAPTER: &str = r#"#!/bin/bash

read

echo '{"response":"triggered","run_id":{"id":"xyzzy"}}'

    #[test]

    fn adapter_first_message_isnt_triggered() -> TestResult<()> {

        const ADAPTER: &str = r#"#!/bin/bash

read

echo '{"response":"finished","result":"success"}'

    #[test]

    fn adapter_outputs_too_many_messages() -> TestResult<()> {

        const ADAPTER: &str = r#"#!/bin/bash

read

echo '{"response":"triggered","run_id":{"id":"xyzzy"}}'

    #[test]

    fn adapter_does_not_exist() -> TestResult<()> {

        let tmp = tempdir()?;

        let bin = tmp.path().join("adapter.sh");

    #[test]

    fn adapter_is_not_executable() -> TestResult<()> {

        const ADAPTER: &str = r#"#!/bin/bash

read

echo '{"response":"triggered","run_id":{"id":"xyzzy"}}'

    #[test]

    fn adapter_has_bad_interpreter() -> TestResult<()> {

        // We test this with a shebang. However, the same kind of code

        // paths and errors should happen when a binary can't be

        // loaded due to missing dynamic linker or library or such.

#![allow(clippy::result_large_err)]

use std::{

    fs::write,

    path::{Path, PathBuf},

    process::exit,

};

use clap::Parser;

use radicle::Profile;

    broker::{Broker, BrokerError},

    config::{Config, ConfigError},

    db::{Db, DbError},

    notif::NotificationChannel,

    pages::{PageBuilder, PageError},

    queueadd::{AdderError, QueueAdderBuilder},

};

fn main() {

    if let Err(e) = fallible_main() {

        exit(1);

    }

}

fn fallible_main() -> Result<(), CibError> {

    let args = Args::parse();

    let config = Config::load(&args.config).map_err(|e| CibError::read_config(&args.config, e))?;

    args.run(&config)?;

            .join()

            .expect("wait for thread to finish")

            .map_err(CibError::add_events)?;

        Ok(())

    }

}

        let adapter = Adapter::new(&spec.command)

            .with_environment(spec.envs())

            .with_environment(spec.sensitive_envs());

        broker.set_default_adapter(&adapter);

        let mut event_notifications = NotificationChannel::default();

            .expect("wait for thread to finish")

            .map_err(CibError::process_queue)?;

        Ok(())

    }

}

        let adapter = Adapter::new(&spec.command)

            .with_environment(spec.envs())

            .with_environment(spec.sensitive_envs());

        broker.set_default_adapter(&adapter);

        let processor = QueueProcessorBuilder::default()

            .join()

            .expect("wait for processor thread to finish")

            .map_err(CibError::process_queue)?;

        // The page updating thread ends when the channel for run

        // notifications is closed by the processor thread ending.

        page_updater

            .join()

            .expect("wait for page updater thread to finish")

            .expect("page updater thread succeeded");

        Ok(())

    }

}

#[allow(clippy::result_large_err)]

fn fallible_main() -> Result<(), CibToolError> {

    let args = Args::parse();

    args.cmd.run(&args)?;

    Ok(())

}

use std::{

    collections::HashMap,

    path::{Path, PathBuf},

};

use time::{macros::format_description, OffsetDateTime};

use radicle::prelude::RepoId;

use crate::{

    adapter::Adapter,

    db::{Db, DbError},

    msg::{PatchEvent, PushEvent, Request},

    run::{Run, Whence},

};

impl Broker {

    #[allow(clippy::result_large_err)]

    pub fn new(db_filename: &Path) -> Result<Self, BrokerError> {

        Ok(Self {

            default_adapter: None,

            adapters: HashMap::new(),

    #[allow(clippy::result_large_err)]

    pub fn execute_ci(&mut self, trigger: &Request) -> Result<Run, BrokerError> {

        let run = match trigger {

            Request::Trigger {

                common,

                    // We run the adapter, but if that fails, we just

                    // log the error. The `Run` value records the

                    // result of the run.

                    if let Err(e) = adapter.run(trigger, &mut run, &self.db) {

                    }

                    run

                } else {

                    return Err(BrokerError::NoAdapter(*rid));

                }

            }

        };

        self.db.update_run(&run)?;

    use crate::{

        msg::{RunId, RunResult},

        run::RunState,

    };

    fn broker(filename: &Path) -> anyhow::Result<Broker> {

    #[test]

    fn adapter_fails() -> TestResult<()> {

        const ADAPTER: &str = r#"#!/bin/bash

read

echo '{"response":"triggered","run_id":{"id":"xyzzy"}}'

    time::Duration,

};

use sqlite::{Connection, State, Statement};

use time::{macros::format_description, OffsetDateTime};

use uuid::Uuid;

    /// If it is created, tables are created.

    pub fn new<P: AsRef<Path>>(filename: P) -> Result<Self, DbError> {

        let filename = filename.as_ref();

        let mut db = Db {

            filename: filename.into(),

            conn: sqlite::open(filename).map_err(|e| DbError::open(filename, e))?,

        debug_assert!(MAX_WAIT.as_millis() < usize::MAX as u128); // safe conversion

        let ms = MAX_WAIT.as_millis() as usize;

        db.conn

            .set_busy_timeout(ms)

            .map_err(|e| DbError::busy_timer(filename, e))?;

    // Prepare a statement for execution.

    fn prepare<'a>(&'a self, sql: &str) -> Result<Stmt<'a>, DbError> {

        match self.conn.prepare(sql) {

            Ok(stmt) => Ok(Stmt::new(sql, stmt)),

            Err(e) => Err(DbError::prepare(sql, &self.filename, e)),

    // Execute a statement that doesn't return any rows with values.

    // This means basically any statement except SELECT.

    fn execute_valueless(stmt: &mut Stmt) -> Result<(), DbError> {

        stmt.stmt.reset().map_err(DbError::reset)?;

        match stmt.stmt.next() {

            Ok(_) => Ok(()),

//! let e = Filters::try_from(filters).unwrap();

//! ```

use radicle::{

    node::{Event, Handle, NodeId},

    prelude::RepoId,

    time,

};

/// Source of events from the local Radicle node.

///

/// The events are filtered. Only events allowed by at least one

    /// Create a new source of node events, for a given Radicle

    /// profile.

    pub fn new(profile: &Profile) -> Result<Self, NodeEventError> {

        let socket = profile.socket();

        if !socket.exists() {

            return Err(NodeEventError::NoControlSocket(socket));

        }

        let node = radicle::Node::new(socket.clone());

        match node.subscribe(time::Duration::MAX) {

            Err(err) => {

                Err(NodeEventError::CannotSubscribe(socket.clone(), err))

            }

        }

    fn allowed(&self, event: &BrokerEvent) -> Result<bool, NodeEventError> {

        for filter in self.allowed.iter() {

            if !event.is_allowed(filter)? {

                return Ok(false);

            }

        }

        Ok(true)

    }

    /// no more events from this source, or there's an error.

    pub fn event(&mut self) -> Result<Vec<BrokerEvent>, NodeEventError> {

        loop {

            if let Some(event) = self.events.next() {

                match event {

                    Ok(event) => {

                        let mut result = vec![];

                        if let Some(broker_events) = BrokerEvent::from_event(&event) {

                            for e in broker_events {

                                if self.allowed(&e)? {

                                    result.push(e);

                                }

                            }

                            if !result.is_empty() {

                                return Ok(result);

                            }

                        }

                    }

                    Err(radicle::node::Error::Io(err))

                        if err.kind() == std::io::ErrorKind::ConnectionReset =>

                    {

                        return Err(NodeEventError::BrokenConnection);

                    }

                    Err(err) => {

                        return Err(NodeEventError::Node(err));

                    }

                }

            } else {

                return Ok(vec![]);

            }

        }

                    RefUpdate::Skipped { name, oid }

                        if name.as_str() == "shutdown" && oid.is_zero() =>

                    {

                        events.push(Self::shutdown());

                    }

                    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)));

                    }

                    _ => (),

    /// Is this broker event allowed by a filter?

    fn is_allowed(&self, filter: &EventFilter) -> Result<bool, NodeEventError> {

        Ok(res)

    }

        let allowed = match self {

            Self::Shutdown => true,

            Self::RefChanged {

                rid,

                name,

            } => {

                let parsed = parse_ref(name)?;

                match filter {

                    EventFilter::Repository(wanted) => rid == wanted,

                    }

                    EventFilter::And(conds) => conds

                        .iter()

                    EventFilter::Or(conds) => conds

                        .iter()

                    EventFilter::Not(conds) => !conds

                        .iter()

                }

            }

        };

        Ok(allowed)

    }

/// }

/// ```

pub fn parse_ref(s: &str) -> Result<Option<ParsedRef>, NodeEventError> {

    const PAT_PATCH: &str = r"^refs/namespaces/[^/]+/refs/heads/patches/([^/]+)$";

    let patch_re = Regex::new(PAT_PATCH).map_err(|e| NodeEventError::Regex(PAT_PATCH, e))?;

    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)

                .map_err(|e| NodeEventError::ParseOid(patch_id_str.into(), e))?;

            return Ok(Some(ParsedRef::Patch(oid)));

        }

    }

    const PAT_BRANCH: &str = r"^refs/namespaces/[^/]+/refs/heads/(.+)$";

    let push_re = Regex::new(PAT_BRANCH).map_err(|e| NodeEventError::Regex(PAT_BRANCH, e))?;

    if let Some(push_captures) = push_re.captures(s) {

        if let Some(branch) = push_captures.get(1) {

            return Ok(Some(ParsedRef::Push(branch.as_str().to_string())));

        }

    }

    Ok(None)

}

#[cfg(test)]

mod test_parse_ref {

    #[test]

    fn plain_branch_name_is_none() -> anyhow::Result<()> {

        assert_eq!(parse_ref("main")?, None);

        Ok(())

    }

    #[test]

    fn namespaced_branch() -> anyhow::Result<()> {

        assert_eq!(

            parse_ref(

                "refs/namespaces/z6MkfBU2cwcZfaE6Z1dLqb7Ve7u4pdgbSo9tP6qUVsqFn2xv/refs/heads/main"

    #[test]

    fn namespaced_branch_with_slashes() -> anyhow::Result<()> {

        assert_eq!(

            parse_ref(

                "refs/namespaces/z6MkfBU2cwcZfaE6Z1dLqb7Ve7u4pdgbSo9tP6qUVsqFn2xv/refs/heads/liw/cob/draft/v2"

    #[test]

    fn namespaced_patch() -> anyhow::Result<()> {

        const SHA: &str = "0a4c69183fc8b8d849f5ab977d70f2a1f4788bca";

        assert_eq!(

            parse_ref(&format!("refs/namespaces/NID/refs/heads/patches/{SHA}"))?,

#[cfg(test)]

mod test {

    #[test]

    fn test_parse_patch_ref() -> anyhow::Result<()> {

        let patch_ref =

            "refs/namespaces/NID/refs/heads/patches/9183ed6232687d3105482960cecb01a53018b80a";

    #[test]

    fn test_parse_push_ref() -> anyhow::Result<()> {

        let push_ref =

            "refs/namespaces/z6MkuhvCnrcow7vzkyQzkuFixzpTa42iC2Cfa4DA8HRLCmys/refs/heads/main";

        let parsed_ref = parse_ref(push_ref)?;

    #[test]

    fn test_parse_invalid_ref() -> anyhow::Result<()> {

        let invalid_ref = "invalid_ref";

        let parsed_ref = parse_ref(invalid_ref)?;

        assert!(parsed_ref.is_none());

    #[test]

    fn branch_is_not_patch_update() {

        assert_eq!(

            is_patch_update(

                "refs/namespaces/z6MkuhvCnrcow7vzkyQzkuFixzpTa42iC2Cfa4DA8HRLCmys/refs/heads/main"

    #[test]

    fn patch_branch_is_not_patch_update() {

        assert_eq!(

            is_patch_update(

                "refs/namespaces/z6MkuhvCnrcow7vzkyQzkuFixzpTa42iC2Cfa4DA8HRLCmys/refs/heads/patches/bbb54a2c9314a528a4fff9d6c2aae874ed098433"

    #[test]

    fn patch_update() {

        assert_eq!(

            is_patch_update(

                "refs/namespaces/z6MkuhvCnrcow7vzkyQzkuFixzpTa42iC2Cfa4DA8HRLCmys/refs/cobs/xyz.radicle.patch/bbb54a2c9314a528a4fff9d6c2aae874ed098433"

    #[test]

    fn get_push_branch() {

        assert_eq!(

            push_branch(

                "refs/namespaces/z6MkuhvCnrcow7vzkyQzkuFixzpTa42iC2Cfa4DA8HRLCmys/refs/heads/branch_name"

    #[test]

    fn get_no_push_branch() {

        assert_eq!(

            push_branch(

                "refs/namespaces/z6MkuhvCnrcow7vzkyQzkuFixzpTa42iC2Cfa4DA8HRLCmys/refs/rad/sigrefs"

//! A logger abstraction for the CI broker.

}

}

    }

}

};

use html_page::{Element, HtmlPage, Tag};

use serde::Serialize;

use time::{macros::format_description, OffsetDateTime};

use crate::{

    db::{Db, DbError, QueuedEvent},

    event::{parse_ref, BrokerEvent, ParsedRef},

    msg::RunId,

    notif::NotificationReceiver,

    run::{Run, RunState, Whence},

        for run in self.runs.iter() {

            runs.insert(run.adapter_run_id().unwrap().clone(), run.clone());

        }

        Ok(StatusPage::new())

    }

        db: Db,

        once: bool,

    ) -> JoinHandle<Result<(), PageError>> {

        if self.dirname.is_none() {

        }

        self.node_alias = profile.config.alias().to_string();

        spawn(move || {

            'processing_loop: loop {

                self.update_and_write(&profile, &db)?;

                    return Ok(());

                }

                match run_rx.recv_timeout(UPDATE_INTERVAL) {

                    Err(RecvTimeoutError::Disconnected) => {

                        break 'processing_loop;

                    }

                }

            }

            // Make sure we update reports and status JSON at least once.

            self.update_and_write(&profile, &db)?;

            Ok(())

        })

    }

    fn update_and_write(&mut self, profile: &Profile, db: &Db) -> Result<(), PageError> {

        if let Some(dirname) = &self.dirname {

            let runs = db.get_all_runs()?;

            // Create list of events, except ones for private

    }

    fn write_file(filename: &Path, text: &str) -> Result<(), PageError> {

        write(filename, text).map_err(|e| PageError::Write(filename.into(), e))?;

        Ok(())

    }

use radicle::Profile;

use crate::{

    db::{Db, DbError},

    event::{BrokerEvent, EventFilter, NodeEventError, NodeEventSource},

    notif::NotificationSender,

};

    }

    pub fn add_events(&self) -> Result<(), AdderError> {

        let profile = Profile::load()?;

        let mut source = NodeEventSource::new(&profile)?;

        for filter in self.filters.iter() {

            source.allow(filter.clone());

        }

        // This loop ends when there's an error, e.g., failure to read an

        // event from the node.

        'event_loop: loop {

            let events = source.event()?;

            if events.is_empty() {

                break 'event_loop;

            } else {

                for e in events {

                    self.push_event(e)?;

                }

            }

        // Add a shutdown event to the queue so that the queue

        // processing thread knows to stop.

        if self.push_shutdown {

            self.push_event(BrokerEvent::Shutdown)?;

        }

        Ok(())

    }

    time::Duration,

};

use radicle::Profile;

use crate::{

    broker::{Broker, BrokerError},

    db::{Db, DbError, QueueId, QueuedEvent},

    event::BrokerEvent,

    msg::{MessageError, RequestBuilder},

    notif::{NotificationReceiver, NotificationSender},

};

    }

    fn process_until_shutdown(&mut self) -> Result<(), QueueError> {

        let mut done = false;

        while !done {

            while let Some(qe) = self.pick_event()? {

                done = self.process_event(qe.event())?;

                self.drop_event(qe.id())?;

                self.run_tx.send(()).map_err(|_| QueueError::NotifyRun)?;

            }

            match self.events_rx.recv_timeout(RECV_TIMEOUT) {

                Err(RecvTimeoutError::Disconnected) => {

                    done = true;

                }

            }

        }

        Ok(())

    }

    fn pick_event(&self) -> Result<Option<QueuedEvent>, QueueError> {

        let ids = self.db.queued_events().map_err(QueueError::db)?;

        let mut queue = vec![];

        for id in ids.iter() {

                oid,

                old: _,

            } => {

                let trigger = RequestBuilder::default()

                    .profile(&self.profile)

                Ok(false)

            }

            BrokerEvent::Shutdown => {

                Ok(true)

            }

        }

    }

    fn drop_event(&mut self, id: &QueueId) -> Result<(), QueueError> {

        self.db.remove_queued_event(id).map_err(QueueError::db)?;

        Ok(())

    }

    Ok(Adapter::new(filename))

}