diff --git a/radicle-node/src/control.rs b/radicle-node/src/control.rs index f778f2da..2fb2542c 100644 --- a/radicle-node/src/control.rs +++ b/radicle-node/src/control.rs @@ -126,6 +126,15 @@ fn command>( } } } + CommandName::TrackedRepos => match handle.tracked_repos() { + Ok(ts) => { + for t in ts.into_iter() { + serde_json::to_writer(&mut writer, &t)?; + writeln!(writer)?; + } + } + Err(e) => return Err(CommandError::Runtime(e)), + }, CommandName::TrackNode => { let (node, alias) = match cmd.args.as_slice() { [node] => (node.as_str(), None), @@ -157,6 +166,15 @@ fn command>( } } } + CommandName::TrackedNodes => match handle.tracked_nodes() { + Ok(ts) => { + for t in ts.into_iter() { + serde_json::to_writer(&mut writer, &t)?; + writeln!(writer)?; + } + } + Err(e) => return Err(CommandError::Runtime(e)), + }, CommandName::AnnounceRefs => { let rid: Id = parse::arg(cmd)?; @@ -184,8 +202,8 @@ fn command>( } CommandName::Routing => match handle.routing() { Ok(c) => { - for (id, seed) in c.iter() { - writeln!(writer, "{id} {seed}")?; + for route in c.into_iter() { + serde_json::to_writer(&mut writer, &route)?; } } Err(e) => return Err(CommandError::Runtime(e)), diff --git a/radicle-node/src/runtime/handle.rs b/radicle-node/src/runtime/handle.rs index fa223ca6..13822f02 100644 --- a/radicle-node/src/runtime/handle.rs +++ b/radicle-node/src/runtime/handle.rs @@ -14,6 +14,7 @@ use crate::identity::Id; use crate::node::{Command, FetchResult}; use crate::profile::Home; use crate::service; +use crate::service::tracking; use crate::service::{CommandError, QueryState}; use crate::service::{NodeId, Sessions}; use crate::wire; @@ -109,6 +110,10 @@ impl radicle::node::Handle for Handle { type Sessions = Sessions; type Error = Error; + type Routing = chan::Receiver<(Id, NodeId)>; + type TrackedRepos = chan::Receiver; + type TrackedNodes = chan::Receiver; + fn is_running(&self) -> bool { true } @@ -131,6 +136,40 @@ impl radicle::node::Handle for Handle { receiver.recv().map_err(Error::from) } + fn tracked_repos(&self) -> Result { + let (sender, receiver) = chan::unbounded(); + let query: Arc = Arc::new(move |state| { + for t in state.tracked_repos()? { + if sender.send(t).is_err() { + break; + } + } + Ok(()) + }); + let (err_sender, err_receiver) = chan::bounded(1); + self.command(service::Command::QueryState(query, err_sender))?; + err_receiver.recv()??; + + Ok(receiver) + } + + fn tracked_nodes(&self) -> Result { + let (sender, receiver) = chan::unbounded(); + let query: Arc = Arc::new(move |state| { + for t in state.tracked_nodes()? { + if sender.send(t).is_err() { + break; + } + } + Ok(()) + }); + let (err_sender, err_receiver) = chan::bounded(1); + self.command(service::Command::QueryState(query, err_sender))?; + err_receiver.recv()??; + + Ok(receiver) + } + fn track_node(&mut self, id: NodeId, alias: Option) -> Result { let (sender, receiver) = chan::bounded(1); self.command(service::Command::TrackNode(id, alias, sender))?; @@ -169,7 +208,7 @@ impl radicle::node::Handle for Handle { receiver.recv().map_err(Error::from) } - fn routing(&self) -> Result, Error> { + fn routing(&self) -> Result { let (sender, receiver) = chan::unbounded(); let query: Arc = Arc::new(move |state| { for (id, node) in state.routing().entries()? { diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs index 45a30fde..43e65a14 100644 --- a/radicle-node/src/service.rs +++ b/radicle-node/src/service.rs @@ -151,6 +151,8 @@ pub enum CommandError { Storage(#[from] storage::Error), #[error(transparent)] Routing(#[from] routing::Error), + #[error(transparent)] + Tracking(#[from] tracking::Error), } #[derive(Debug)] @@ -285,8 +287,7 @@ where self.filter = Filter::new( self.tracking .repo_entries()? - .filter(|(_, _, policy)| *policy == tracking::Policy::Track) - .map(|(e, _, _)| e), + .filter_map(|t| (t.policy == tracking::Policy::Track).then_some(t.id)), ); Ok(updated) } @@ -376,8 +377,7 @@ where self.filter = Filter::new( self.tracking .repo_entries()? - .filter(|(_, _, policy)| *policy == tracking::Policy::Track) - .map(|(e, _, _)| e), + .filter_map(|t| (t.policy == tracking::Policy::Track).then_some(t.id)), ); Ok(()) @@ -1319,6 +1319,10 @@ pub trait ServiceState { fn config(&self) -> &Config; /// Get reference to routing table. fn routing(&self) -> &dyn routing::Store; + /// Get the tracked repos. + fn tracked_repos(&self) -> Result, tracking::Error>; + /// Get the tracked nodes. + fn tracked_nodes(&self) -> Result, tracking::Error>; } impl ServiceState for Service @@ -1354,6 +1358,14 @@ where fn routing(&self) -> &dyn routing::Store { &self.routing } + + fn tracked_repos(&self) -> Result, tracking::Error> { + Ok(self.tracking.repo_entries()?.collect()) + } + + fn tracked_nodes(&self) -> Result, tracking::Error> { + Ok(self.tracking.node_entries()?.collect()) + } } /// Disconnect reason. diff --git a/radicle-node/src/service/tracking.rs b/radicle-node/src/service/tracking.rs index edc7ab33..363e577a 100644 --- a/radicle-node/src/service/tracking.rs +++ b/radicle-node/src/service/tracking.rs @@ -1,72 +1,15 @@ mod store; -use std::str::FromStr; -use std::{fmt, ops}; +use std::ops; use crate::prelude::Id; use crate::service::NodeId; +pub use crate::node::tracking::{Alias, Node, Policy, Repo, Scope}; + pub use store::Config as Store; pub use store::Error; -/// Node alias. -pub type Alias = String; - -/// Tracking policy. -#[derive(Default, Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] -pub enum Policy { - /// The resource is tracked. - Track, - /// The resource is blocked. - #[default] - Block, -} - -impl fmt::Display for Policy { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - match self { - Self::Track => write!(f, "track"), - Self::Block => write!(f, "block"), - } - } -} - -impl FromStr for Policy { - type Err = String; - - fn from_str(s: &str) -> Result { - match s { - "track" => Ok(Self::Track), - "block" => Ok(Self::Block), - _ => Err(s.to_owned()), - } - } -} - -/// Tracking scope of a repository tracking policy. -#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] -pub enum Scope { - /// Track remotes of nodes that are already tracked. - Trusted, - /// Track remotes of repository delegates. - DelegatesOnly, - /// Track all remotes. - All, -} - -impl FromStr for Scope { - type Err = (); - - fn from_str(s: &str) -> Result { - match s { - "trusted" => Ok(Self::Trusted), - "delegates-only" => Ok(Self::DelegatesOnly), - "all" => Ok(Self::All), - _ => Err(()), - } - } -} - /// Tracking configuration. #[derive(Debug)] pub struct Config { diff --git a/radicle-node/src/service/tracking/store.rs b/radicle-node/src/service/tracking/store.rs index 46997f1c..5ac028d5 100644 --- a/radicle-node/src/service/tracking/store.rs +++ b/radicle-node/src/service/tracking/store.rs @@ -1,6 +1,5 @@ #![allow(clippy::type_complexity)] use std::path::Path; -use std::str::FromStr; use std::{fmt, io}; use sqlite as sql; @@ -9,7 +8,7 @@ use thiserror::Error; use crate::prelude::Id; use crate::service::NodeId; -use super::{Alias, Policy, Scope}; +use super::{Alias, Node, Policy, Repo, Scope}; #[derive(Error, Debug)] pub enum Error { @@ -21,63 +20,6 @@ pub enum Error { Internal(#[from] sql::Error), } -impl sqlite::BindableWithIndex for Scope { - fn bind(self, stmt: &mut sql::Statement<'_>, i: I) -> sql::Result<()> { - let s = match self { - Self::Trusted => "trusted", - Self::DelegatesOnly => "delegates-only", - Self::All => "all", - }; - s.bind(stmt, i) - } -} - -impl TryFrom<&sql::Value> for Scope { - type Error = sql::Error; - - fn try_from(value: &sql::Value) -> Result { - let message = Some("invalid remote scope".to_owned()); - - match value { - sql::Value::String(scope) => Scope::from_str(scope).map_err(|_| sql::Error { - code: None, - message, - }), - _ => Err(sql::Error { - code: None, - message, - }), - } - } -} - -impl sqlite::BindableWithIndex for Policy { - fn bind(self, stmt: &mut sql::Statement<'_>, i: I) -> sql::Result<()> { - match self { - Self::Track => "track", - Self::Block => "block", - } - .bind(stmt, i) - } -} - -impl TryFrom<&sql::Value> for Policy { - type Error = sql::Error; - - fn try_from(value: &sql::Value) -> Result { - let message = Some("sql: invalid policy".to_owned()); - - match value { - sql::Value::String(s) if s == "track" => Ok(Policy::Track), - sql::Value::String(s) if s == "block" => Ok(Policy::Block), - _ => Err(sql::Error { - code: None, - message, - }), - } - } -} - /// Tracking configuration. pub struct Config { db: sql::Connection, @@ -248,7 +190,7 @@ impl Config { } /// Get node tracking entries. - pub fn node_entries(&self) -> Result>, Error> { + pub fn node_entries(&self) -> Result>, Error> { let mut stmt = self .db .prepare("SELECT id, alias, policy FROM `node-policies`")? @@ -257,16 +199,17 @@ impl Config { while let Some(Ok(row)) = stmt.next() { let id = row.read("id"); - let alias = row.read::<&str, _>("alias"); + let alias = row.read::<&str, _>("alias").to_owned(); let policy = row.read::("policy"); - entries.push((id, alias.to_owned(), policy)); + entries.push(Node { id, alias, policy }); } Ok(Box::new(entries.into_iter())) } + // TODO: see if sql can return iterator directly /// Get repository tracking entries. - pub fn repo_entries(&self) -> Result>, Error> { + pub fn repo_entries(&self) -> Result>, Error> { let mut stmt = self .db .prepare("SELECT id, scope, policy FROM `repo-policies`")? @@ -278,7 +221,7 @@ impl Config { let scope = row.read("scope"); let policy = row.read::("policy"); - entries.push((id, scope, policy)); + entries.push(Repo { id, scope, policy }); } Ok(Box::new(entries.into_iter())) } @@ -324,9 +267,9 @@ mod test { assert!(db.track_node(id, None).unwrap()); } let mut entries = db.node_entries().unwrap(); - assert_matches!(entries.next(), Some((id, _, _)) if id == ids[0]); - assert_matches!(entries.next(), Some((id, _, _)) if id == ids[1]); - assert_matches!(entries.next(), Some((id, _, _)) if id == ids[2]); + assert_matches!(entries.next(), Some(Node { id, .. }) if id == ids[0]); + assert_matches!(entries.next(), Some(Node { id, .. }) if id == ids[1]); + assert_matches!(entries.next(), Some(Node { id, .. }) if id == ids[2]); } #[test] @@ -338,9 +281,9 @@ mod test { assert!(db.track_repo(id, Scope::All).unwrap()); } let mut entries = db.repo_entries().unwrap(); - assert_matches!(entries.next(), Some((id, _, _)) if id == ids[0]); - assert_matches!(entries.next(), Some((id, _, _)) if id == ids[1]); - assert_matches!(entries.next(), Some((id, _, _)) if id == ids[2]); + assert_matches!(entries.next(), Some(Repo { id, .. }) if id == ids[0]); + assert_matches!(entries.next(), Some(Repo { id, .. }) if id == ids[1]); + assert_matches!(entries.next(), Some(Repo { id, .. }) if id == ids[2]); } #[test] diff --git a/radicle-node/src/test/handle.rs b/radicle-node/src/test/handle.rs index 1d97a692..ab0d7b29 100644 --- a/radicle-node/src/test/handle.rs +++ b/radicle-node/src/test/handle.rs @@ -6,8 +6,8 @@ use crossbeam_channel as chan; use crate::identity::Id; use crate::node::{FetchResult, Seeds}; use crate::runtime::HandleError; -use crate::service; use crate::service::NodeId; +use crate::service::{self, tracking}; use crate::storage::RefUpdate; #[derive(Default, Clone)] @@ -20,6 +20,9 @@ pub struct Handle { impl radicle::node::Handle for Handle { type Error = HandleError; type Sessions = service::Sessions; + type Routing = Vec<(Id, NodeId)>; + type TrackedRepos = Vec; + type TrackedNodes = Vec; fn is_running(&self) -> bool { true @@ -37,6 +40,32 @@ impl radicle::node::Handle for Handle { Ok(FetchResult::from(Ok::, Self::Error>(vec![]))) } + fn tracked_repos(&self) -> Result { + Ok(self + .tracking_repos + .iter() + .copied() + .map(|id| tracking::Repo { + id, + scope: tracking::Scope::All, + policy: tracking::Policy::Track, + }) + .collect()) + } + + fn tracked_nodes(&self) -> Result { + Ok(self + .tracking_nodes + .iter() + .copied() + .map(|id| tracking::Node { + id, + alias: "".to_string(), + policy: tracking::Policy::Track, + }) + .collect()) + } + fn track_repo(&mut self, id: Id) -> Result { Ok(self.tracking_repos.insert(id)) } @@ -67,7 +96,7 @@ impl radicle::node::Handle for Handle { unimplemented!() } - fn routing(&self) -> Result, Self::Error> { + fn routing(&self) -> Result { unimplemented!(); } diff --git a/radicle/src/node.rs b/radicle/src/node.rs index 9efcfb2f..495cab54 100644 --- a/radicle/src/node.rs +++ b/radicle/src/node.rs @@ -1,4 +1,5 @@ mod features; +pub mod tracking; use std::collections::BTreeSet; use std::io::{BufRead, BufReader}; @@ -124,10 +125,14 @@ pub enum CommandName { TrackRepo, /// Untrack the given repository. UntrackRepo, + /// Get the tracked repositories. + TrackedRepos, /// Track the given node. TrackNode, /// Untrack the given node. UntrackNode, + /// Get the tracked nodes. + TrackedNodes, /// Get the node's inventory. Inventory, /// Get the node's routing table. @@ -374,6 +379,10 @@ pub trait Handle { /// The error returned by all methods. type Error: std::error::Error + Send + Sync + 'static; + type Routing: IntoIterator; + type TrackedRepos: IntoIterator; + type TrackedNodes: IntoIterator; + /// Check if the node is running. to a peer. fn is_running(&self) -> bool; /// Connect to a peer. @@ -391,6 +400,10 @@ pub trait Handle { fn untrack_repo(&mut self, id: Id) -> Result; /// Untrack the given node. fn untrack_node(&mut self, id: NodeId) -> Result; + /// Get the tracking information for all tracked repos in storage. + fn tracked_repos(&self) -> Result; + /// Get the tracking information for all tracked nodes in storage. + fn tracked_nodes(&self) -> Result; /// Notify the service that a project has been updated, and announce local refs. fn announce_refs(&mut self, id: Id) -> Result<(), Self::Error>; /// Announce local inventory. @@ -400,7 +413,7 @@ pub trait Handle { /// Ask the service to shutdown. fn shutdown(self) -> Result<(), Self::Error>; /// Query the routing table entries. - fn routing(&self) -> Result, Self::Error>; + fn routing(&self) -> Result; /// Query the peer session state. fn sessions(&self) -> Result; /// Query the inventory. @@ -446,10 +459,16 @@ impl Node { } } +// TODO(finto): tracked_repos, tracked_nodes, and routing should all +// attempt to return iterators instead of allocating vecs. impl Handle for Node { type Sessions = (); type Error = Error; + type TrackedRepos = Vec; + type TrackedNodes = Vec; + type Routing = Vec<(Id, NodeId)>; + fn is_running(&self) -> bool { let Ok(mut lines) = self.call::<&str, CommandResult>(CommandName::Status, []) else { return false; @@ -486,6 +505,24 @@ impl Handle for Node { Ok(result) } + fn tracked_repos(&self) -> Result, Self::Error> { + let mut repos = Vec::new(); + for result in self.call::<&str, _>(CommandName::TrackedRepos, [])? { + let repo = result?; + repos.push(repo); + } + Ok(repos) + } + + fn tracked_nodes(&self) -> Result, Self::Error> { + let mut repos = Vec::new(); + for result in self.call::<&str, _>(CommandName::TrackedNodes, [])? { + let repo = result?; + repos.push(repo); + } + Ok(repos) + } + fn track_node(&mut self, id: NodeId, alias: Option) -> Result { let id = id.to_human(); let args = if let Some(alias) = alias.as_deref() { @@ -552,8 +589,13 @@ impl Handle for Node { response.into() } - fn routing(&self) -> Result, Error> { - todo!(); + fn routing(&self) -> Result { + let mut routes = Vec::new(); + for result in self.call::<&str, _>(CommandName::Routing, [])? { + let route = result?; + routes.push(route); + } + Ok(routes) } fn sessions(&self) -> Result { diff --git a/radicle/src/node/tracking.rs b/radicle/src/node/tracking.rs new file mode 100644 index 00000000..46262217 --- /dev/null +++ b/radicle/src/node/tracking.rs @@ -0,0 +1,164 @@ +use std::fmt; +use std::str::FromStr; + +use serde::{Deserialize, Serialize}; +use thiserror::Error; + +use crate::prelude::Id; + +use super::NodeId; + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct Repo { + pub id: Id, + pub scope: Scope, + pub policy: Policy, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct Node { + pub id: NodeId, + pub alias: Alias, + pub policy: Policy, +} + +/// Node alias. +pub type Alias = String; + +/// Tracking policy. +#[derive(Default, Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] +pub enum Policy { + /// The resource is tracked. + Track, + /// The resource is blocked. + #[default] + Block, +} + +impl fmt::Display for Policy { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Track => write!(f, "track"), + Self::Block => write!(f, "block"), + } + } +} + +impl FromStr for Policy { + type Err = String; + + fn from_str(s: &str) -> Result { + match s { + "track" => Ok(Self::Track), + "block" => Ok(Self::Block), + _ => Err(s.to_owned()), + } + } +} + +#[cfg(feature = "sql")] +impl sqlite::BindableWithIndex for Policy { + fn bind( + self, + stmt: &mut sqlite::Statement<'_>, + i: I, + ) -> sqlite::Result<()> { + match self { + Self::Track => "track", + Self::Block => "block", + } + .bind(stmt, i) + } +} + +#[cfg(feature = "sql")] +impl TryFrom<&sqlite::Value> for Policy { + type Error = sqlite::Error; + + fn try_from(value: &sqlite::Value) -> Result { + let message = Some("sql: invalid policy".to_owned()); + + match value { + sqlite::Value::String(s) if s == "track" => Ok(Policy::Track), + sqlite::Value::String(s) if s == "block" => Ok(Policy::Block), + _ => Err(sqlite::Error { + code: None, + message, + }), + } + } +} + +/// Tracking scope of a repository tracking policy. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] +pub enum Scope { + /// Track remotes of nodes that are already tracked. + Trusted, + /// Track remotes of repository delegates. + DelegatesOnly, + /// Track all remotes. + All, +} + +impl fmt::Display for Scope { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Scope::Trusted => f.write_str("trusted"), + Scope::DelegatesOnly => f.write_str("delegates-only"), + Scope::All => f.write_str("all"), + } + } +} + +#[derive(Debug, Error)] +#[error("invalid tracking scope: {0:?}")] +pub struct ParseScopeError(String); + +impl FromStr for Scope { + type Err = ParseScopeError; + + fn from_str(s: &str) -> Result { + match s { + "trusted" => Ok(Self::Trusted), + "delegates-only" => Ok(Self::DelegatesOnly), + "all" => Ok(Self::All), + _ => Err(ParseScopeError(s.to_string())), + } + } +} + +#[cfg(feature = "sql")] +impl sqlite::BindableWithIndex for Scope { + fn bind( + self, + stmt: &mut sqlite::Statement<'_>, + i: I, + ) -> sqlite::Result<()> { + let s = match self { + Self::Trusted => "trusted", + Self::DelegatesOnly => "delegates-only", + Self::All => "all", + }; + s.bind(stmt, i) + } +} + +#[cfg(feature = "sql")] +impl TryFrom<&sqlite::Value> for Scope { + type Error = sqlite::Error; + + fn try_from(value: &sqlite::Value) -> Result { + let message = Some("invalid remote scope".to_owned()); + + match value { + sqlite::Value::String(scope) => Scope::from_str(scope).map_err(|_| sqlite::Error { + code: None, + message, + }), + _ => Err(sqlite::Error { + code: None, + message, + }), + } + } +}