"regex-automata",

 "regex-automata",

 "regex-syntax",

name = "fnv"

version = "1.0.7"

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

checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1"

[[package]]

 "regex-syntax",

version = "0.2.0"

checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9"

 "regex-automata",

version = "0.50.3"

checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5"

 "windows-sys 0.60.2",

name = "prometheus"

version = "0.14.0"

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

checksum = "3ca5326d8d0b950a9acd87e6a3f94745394f62e4dae1b1ee22b2bc0c394af43a"

dependencies = [

 "cfg-if",

 "fnv",

 "lazy_static",

 "memchr",

 "parking_lot",

 "protobuf",

 "thiserror 2.0.12",

]

[[package]]

name = "protobuf"

version = "3.7.2"

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

checksum = "d65a1d4ddae7d8b5de68153b48f6aa3bba8cb002b243dbdbc55a5afbc98f99f4"

dependencies = [

 "once_cell",

 "protobuf-support",

 "thiserror 1.0.69",

]

[[package]]

name = "protobuf-support"

version = "3.7.2"

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

checksum = "3e36c2f31e0a47f9280fb347ef5e461ffcd2c52dd520d8e216b52f93b0b0d7d6"

dependencies = [

 "thiserror 1.0.69",

]

[[package]]

name = "radicle-metrics"

version = "0.1.0"

dependencies = [

 "anyhow",

 "clap",

 "prometheus",

 "radicle",

 "serde",

 "serde_json",

]

[[package]]

 "regex-automata",

 "regex-syntax",

 "regex-syntax",

version = "0.1.43"

checksum = "2d15d90a0b5c19378952d479dc858407149d7bb45a14de0142f6c534b16fc647"

version = "0.1.35"

checksum = "7a04e24fab5c89c6a36eb8558c9656f30d81de51dfa4d3b45f26b21d61fa0a6c"

version = "0.3.22"

checksum = "2f30143827ddab0d256fd843b7a66d164e9f271cfa0dde49142c5ca0ca291f1e"

 "regex-automata",

 "regex-syntax",

[package]

name = "radicle-metrics"

description = "Radicle Metrics"

homepage.workspace = true

license.workspace = true

version = "0.1.0"

authors = ["shishqa <shishqa.main@gmail.com>"]

edition.workspace = true

rust-version.workspace = true

[[bin]]

name = "rad-metrics"

path = "src/main.rs"

[dependencies]

anyhow = "1"

clap = { version = "4.5.44", features = ["derive"] }

prometheus = "0.14.0"

radicle = { workspace = true }

serde = { workspace = true }

serde_json = { workspace = true }

[lints]

workspace = true

pub mod metrics;

pub mod walk;

use anyhow::Context;

use clap::Parser;

use radicle::cob;

use radicle_metrics::metrics::MetricsWindow;

use radicle_metrics::walk::CobHistoryWalk;

pub const NAME: &str = "rad-metrics";

#[derive(Parser, Debug)]

#[command(name = NAME)]

struct CliArgs {

    #[arg(long = "type", short, value_name = "TYPENAME")]

    type_names: Vec<cob::TypeName>,

    #[arg(long, short, value_name = "MAX_WINDOW_SIZE")]

    max_window_size: Option<usize>,

    #[arg(long, short, value_name = "POLL_INTERVAL")]

    poll_interval: Option<u64>,

}

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

    let CliArgs {

        type_names,

        max_window_size,

        poll_interval,

    } = CliArgs::parse();

    let profile = radicle::profile::Profile::load().context("profile::load")?;

    let storage = &profile.storage;

    let mut window = MetricsWindow::new(max_window_size)?;

    let mut walk = CobHistoryWalk::new(storage, type_names);

    if let Some(poll_interval) = poll_interval {

        loop {

            walk.scan_new_records(|record| {

                window.add(record);

            })?;

            window.print_aggregated()?;

            if max_window_size.is_some() {

                window.print_window()?;

            }

            std::thread::sleep(std::time::Duration::from_secs(poll_interval));

        }

    } else {

        walk.scan_new_records(|record| {

            window.add(record);

        })?;

        window.print_window()?;

    }

    Ok(())

}

use prometheus::{Encoder, IntGaugeVec, Opts, Registry, TextEncoder};

use serde::Serialize;

use std::collections::{BinaryHeap, HashMap};

use radicle::cob;

use crate::walk::{Event, Record};

#[derive(Serialize, PartialEq, Eq)]

struct Point {

    value: i64,

    name: String,

    timestamp: cob::common::Timestamp,

    labels: HashMap<String, String>,

}

impl std::cmp::PartialOrd for Point {

    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {

        Some(self.cmp(other))

    }

}

impl std::cmp::Ord for Point {

    fn cmp(&self, other: &Self) -> std::cmp::Ordering {

        self.timestamp.cmp(&other.timestamp).reverse()

    }

}

pub struct MetricsWindow {

    registry: Registry,

    open_issues: IntGaugeVec,

    max_points: Option<usize>,

    points: BinaryHeap<Point>,

}

impl MetricsWindow {

    pub fn new(max_points: Option<usize>) -> anyhow::Result<Self> {

        let open_issues = IntGaugeVec::new(

            Opts::new(

                "radicle_open_issues",

                "counter for open issues across all repositories",

            ),

            &["rid"],

        )?;

        let registry = Registry::new();

        registry.register(Box::new(open_issues.clone()))?;

        Ok(Self {

            registry,

            open_issues,

            max_points,

            points: Default::default(),

        })

    }

    pub fn add(&mut self, record: Record) {

        if record.op.manifest.type_name != *cob::issue::TYPENAME {

            // TODO: process other COB types

            return;

        }

        let mut new_points = Vec::<Point>::new();

        match record.event {

            Event::Created => {

                new_points.push(Point {

                    timestamp: record.op.timestamp,

                    name: "radicle_open_issues".to_string(),

                    labels: HashMap::from([("rid".to_string(), record.rid.to_string())]),

                    value: 1,

                });

            }

            Event::Deleted => {

                new_points.push(Point {

                    timestamp: record.op.timestamp,

                    name: "radicle_open_issues".to_string(),

                    labels: HashMap::from([("rid".to_string(), record.rid.to_string())]),

                    value: -1,

                });

            }

            _ => {}

        }

        for action in record.op.actions {

            let action: cob::issue::Action = serde_json::from_value(action).unwrap();

            if let cob::issue::Action::Lifecycle { state } = action {

                match state {

                    cob::issue::State::Open => {

                        new_points.push(Point {

                            timestamp: record.op.timestamp,

                            name: "radicle_open_issues".to_string(),

                            labels: HashMap::from([("rid".to_string(), record.rid.to_string())]),

                            value: 1,

                        });

                    }

                    cob::issue::State::Closed { .. } => {

                        new_points.push(Point {

                            timestamp: record.op.timestamp,

                            name: "radicle_open_issues".to_string(),

                            labels: HashMap::from([("rid".to_string(), record.rid.to_string())]),

                            value: -1,

                        });

                    }

                }

            }

        }

        for point in new_points {

            if self

                .max_points

                .map(|m| self.points.len() >= m)

                .unwrap_or(false)

            {

                match self.points.peek() {

                    Some(oldest_point) if oldest_point < &point => {

                        let rid = oldest_point

                            .labels

                            .get("rid")

                            .cloned()

                            .unwrap_or("-".to_string());

                        self.open_issues

                            .with_label_values(&[rid])

                            .add(oldest_point.value);

                        self.points.pop();

                        self.points.push(point);

                    }

                    _ => {

                        let rid = point.labels.get("rid").cloned().unwrap_or("-".to_string());

                        self.open_issues.with_label_values(&[rid]).add(point.value);

                    }

                }

            } else {

                self.points.push(point);

            }

        }

    }

    pub fn print_window(&self) -> anyhow::Result<()> {

        for point in &self.points {

            println!("{}", serde_json::to_string(&point)?);

        }

        Ok(())

    }

    pub fn print_aggregated(&self) -> anyhow::Result<()> {

        let mut buffer = vec![];

        let encoder = TextEncoder::new();

        let metric_families = self.registry.gather();

        encoder.encode(&metric_families, &mut buffer).unwrap();

        println!("{}", String::from_utf8(buffer).unwrap());

        Ok(())

    }

}

use anyhow;

use anyhow::Context;

use std::collections::{HashMap, HashSet};

use radicle::cob;

use radicle::cob::object::Storage as _;

use radicle::cob::stream::CobStream as _;

use radicle::identity::RepoId;

use radicle::storage::git::Storage;

use radicle::storage::ReadStorage;

/// Represents the type of change that occurred to a COB.

pub enum Event {

    Created,

    Changed,

    Deleted,

}

/// A single record of a COB event with associated metadata.

pub struct Record {

    /// The type of event that occurred.

    pub event: Event,

    /// The repository ID where the COB exists.

    pub rid: RepoId,

    /// The unique identifier of the COB object.

    pub object_id: cob::ObjectId,

    /// The operation that caused the event, with its JSON payload.

    pub op: cob::op::Op<serde_json::Value>,

}

/// A checkpoint that tracks the last known state of COBs for incremental updates.

#[derive(Default)]

pub struct Checkpoint {

    /// The COB types that are being tracked.

    type_names: Vec<cob::TypeName>,

    /// Last known operation for each COB object, keyed by object ID.

    /// The value is a tuple of (repository ID, last operation).

    last_ops: HashMap<cob::ObjectId, (RepoId, cob::op::Op<serde_json::Value>)>,

}

/// A walker that traverses the history of COBs across repositories.

///

/// This structure provides incremental scanning of COB history, allowing

/// callers to process only new or changed records since the last scan.

pub struct CobHistoryWalk<'a> {

    storage: &'a Storage,

    checkpoint: Checkpoint,

}

impl<'a> CobHistoryWalk<'a> {

    pub fn new(storage: &'a Storage, type_names: Vec<cob::TypeName>) -> Self {

        Self {

            storage,

            checkpoint: Checkpoint {

                type_names,

                last_ops: Default::default(),

            },

        }

    }

    pub fn new_with_checkpoint(storage: &'a Storage, checkpoint: Checkpoint) -> Self {

        Self {

            storage,

            checkpoint,

        }

    }

    pub fn get_checkpoint(&self) -> &Checkpoint {

        &self.checkpoint

    }

    /// Scans for new or changed COB records since the last checkpoint.

    ///

    /// This method:

    /// 1. Iterates through all repositories in storage

    /// 2. For each repository, scans the specified COB types

    /// 3. For each COB object, processes operations that occurred since the last known operation

    /// 4. Updates the checkpoint with the latest operations

    /// 5. Identifies and reports deleted COBs (objects that no longer exist)

    pub fn scan_new_records<F>(&mut self, mut sink_fn: F) -> anyhow::Result<()>

    where

        F: FnMut(Record),

    {

        // XXX: This could be turned into an iterator to better control the portion of records.

        let mut visited_objects = HashSet::<cob::ObjectId>::new();

        for repo in self.storage.repositories()? {

            let rid = repo.rid.to_string();

            let repo = self

                .storage

                .repository(repo.rid)

                .context(format!("load {}", rid))?;

            for type_name in &self.checkpoint.type_names {

                let references = repo.types(type_name)?;

                for (object_id, _) in references {

                    let history = cob::stream::CobRange::new(type_name, &object_id);

                    let stream = cob::stream::Stream::<serde_json::Value>::new(

                        &repo.backend,

                        history,

                        type_name.clone(),

                    );

                    let (iter, is_new) = match self.checkpoint.last_ops.get(&object_id) {

                        Some(last_op) => (stream.since(last_op.1.id)?, false),

                        None => (stream.all()?, true),

                    };

                    let mut last_op = None;

                    let mut first_op = None;

                    // Walk in historical order.

                    for op in iter.skip(if is_new { 0 } else { 1 }) {

                        let op = op?;

                        first_op.get_or_insert(op.clone());

                        last_op = Some(op.clone());

                        sink_fn(Record {

                            event: Event::Changed,

                            rid: repo.id,

                            object_id,

                            op,

                        });

                    }

                    if let Some(first_op) = first_op {

                        if is_new {

                            sink_fn(Record {

                                event: Event::Created,

                                rid: repo.id,

                                object_id,

                                op: first_op.clone(),

                            });

                        }

                    }

                    if let Some(last_op) = last_op {

                        self.checkpoint

                            .last_ops

                            .insert(object_id, (repo.id, last_op));

                    }

                    visited_objects.insert(object_id);

                }

            }

        }

        self.checkpoint.last_ops.retain(|k, v| {

            if visited_objects.contains(k) {

                return true;

            }

            sink_fn(Record {

                event: Event::Deleted,

                rid: v.0,

                object_id: *k,

                op: v.1.clone(),

            });

            false

        });

        Ok(())

    }

}