indexer-node: Initial commit

Signed-off-by: xphoniex <[email protected]>

Author: xphoniex

Cargo.lock

@@ -4,6 +4,7 @@ members = [
   "org-node",
   "git-server",
   "service-init",
+  "indexer-node",
   "shared",
 ]
 

Cargo.toml

@@ -0,0 +1,24 @@
+[package]
+name = "radicle-indexer-node"
+license = "MIT OR Apache-2.0"
+version = "0.0.1"
+authors = ["Alexis Sellier <[email protected]>"]
+edition = "2018"
+build = "../build.rs"
+default-run = "radicle-indexer-node"
+
+[dependencies]
+shared = { path = "../shared", default-features = false }
+async-trait = { version = "0.1" }
+serde = { version = "1", features = ["derive"] }
+serde_json = { version = "1" }
+librad = { version = "0" }
+tokio = { version = "1.2", features = ["macros", "rt", "sync"] }
+thiserror = { version = "1" }
+argh = { version = "0.1.4" }
+tracing = "0.1"
+anyhow = "1.0"
+warp = { version = "0.3.1", features = ["tls"] }
+rocksdb = "0.17.0"
+url = { version = "*" }
+git2 = "0.13"

indexer-node/Cargo.toml

@@ -0,0 +1,102 @@
+use rocksdb::{DBWithThreadMode, Direction, Error, IteratorMode, MultiThreaded, DB};
+
+use std::sync::Arc;
+
+#[derive(Debug, Clone)]
+pub struct Handle {
+    pub db: Arc<DBWithThreadMode<MultiThreaded>>,
+}
+
+impl Handle {
+    /// Creates a DB handle and stores its data at the path (folder).
+    pub fn from_path(path: &str) -> Result<Handle, Error> {
+        type MDB = DBWithThreadMode<MultiThreaded>;
+        let db = MDB::open_default(path)?;
+        //let db = DB::open_default(path)?;
+        Ok(Handle { db: Arc::new(db) })
+    }
+
+    /// Key should hold alias::urn, and value holds the server
+    /// in which repo is hosted.
+    pub fn add_repository<K, V>(&self, k: K, v: V) -> Result<(), Error>
+    where
+        K: AsRef<[u8]>,
+        V: AsRef<[u8]>,
+    {
+        let k = [b"repo::", k.as_ref()].concat();
+        self.db.put(k, v)
+    }
+
+    /// Iterates through all keys starting with prefix and returns (key, value).
+    fn iterate_prefix<P>(&self, prefix: P) -> impl Iterator<Item = (String, String)> + '_
+    where
+        P: AsRef<[u8]> + 'static,
+    {
+        self.db
+            .iterator(IteratorMode::From(prefix.as_ref(), Direction::Forward))
+            .into_iter()
+            .take_while(move |(k, _)| k.starts_with(prefix.as_ref()))
+            // This is safe because inputs are checked on insertion
+            // and leaking the Box leaves sole ownership to String.
+            .map(|(k, v)| unsafe {
+                let k = Box::leak(k);
+                let v = Box::leak(v);
+                (
+                    String::from_raw_parts(k.as_mut_ptr(), k.len(), k.len()),
+                    String::from_raw_parts(v.as_mut_ptr(), v.len(), v.len()),
+                )
+            })
+    }
+
+    /// Iterates through all keys starting *from* prefix and returns (key, value).
+    pub fn iterate_from_prefix<P>(&self, prefix: P) -> impl Iterator<Item = (String, String)> + '_
+    where
+        P: AsRef<[u8]> + 'static,
+    {
+        self.db
+            .iterator(IteratorMode::From(prefix.as_ref(), Direction::Forward))
+            .into_iter()
+            // This is safe because inputs are checked on insertion
+            // and leaking the Box leaves sole ownership to String.
+            .map(|(k, v)| unsafe {
+                let k = Box::leak(k);
+                let v = Box::leak(v);
+                (
+                    String::from_raw_parts(k.as_mut_ptr(), k.len(), k.len()),
+                    String::from_raw_parts(v.as_mut_ptr(), v.len(), v.len()),
+                )
+            })
+    }
+
+    /// Lists all repositories' alias::urn (key) and servers (value) hosting them.
+    pub fn list_repositories(&self) -> impl Iterator<Item = (String, String)> + '_ {
+        let prefix = b"repo::";
+        self.iterate_prefix(prefix)
+    }
+
+    /// Lists all repositories starting with alias::<any_urn>.
+    pub fn repos_starting_with(&self, alias: &str) -> impl Iterator<Item = (String, String)> + '_ {
+        let prefix = [b"repo::", alias.as_bytes()].concat();
+        self.iterate_prefix(prefix)
+    }
+
+    /// Lists all users starting with username::<any_urn>.
+    pub fn users_starting_with(
+        &self,
+        username: &str,
+    ) -> impl Iterator<Item = (String, String)> + '_ {
+        let prefix = [b"user::", username.as_bytes()].concat();
+        self.iterate_prefix(prefix)
+    }
+
+    /// Key should hold username::urn, and value holds the server
+    /// in which user is active.
+    pub fn add_user<K, V>(&self, k: K, v: V) -> Result<(), Error>
+    where
+        K: AsRef<[u8]>,
+        V: AsRef<[u8]>,
+    {
+        let k = [b"user::", k.as_ref()].concat();
+        self.db.put(k, v)
+    }
+}

indexer-node/src/db.rs

@@ -0,0 +1,13 @@
+/// Errors that may occur when interacting with [`librad::net::peer::Peer`].
+#[derive(Debug, thiserror::Error)]
+pub enum Error {
+    /// The entity was not found.
+    #[error("entity not found")]
+    NotFound,
+
+    /// An error occured while parsing URL.
+    #[error(transparent)]
+    ParseError(#[from] url::ParseError),
+}
+
+impl warp::reject::Reject for Error {}

indexer-node/src/error.rs

@@ -0,0 +1,22 @@
+use std::process::Command;
+
+pub fn git<S: AsRef<std::ffi::OsStr>>(
+    repo: &std::path::Path,
+    args: impl IntoIterator<Item = S>,
+) -> Result<String, anyhow::Error> {
+    let output = Command::new("git").current_dir(repo).args(args).output()?;
+
+    if output.status.success() {
+        let out = if output.stdout.is_empty() {
+            &output.stderr
+        } else {
+            &output.stdout
+        };
+        return Ok(String::from_utf8_lossy(out).into());
+    }
+
+    Err(anyhow::Error::new(std::io::Error::new(
+        std::io::ErrorKind::Other,
+        String::from_utf8_lossy(&output.stderr),
+    )))
+}