add triplestore example

.github/workflows/test.yml

@@ -31,6 +31,7 @@ jobs:
           workspaces: >
             . -> target
             examples/actix-kv -> target
+            examples/actix-triplestore -> target
       - name: Build
         run: cargo build -v
       - name: Format

examples/actix-kv/src/main.rs

@@ -8,7 +8,6 @@ use actix_web::{
     App, HttpResponse, HttpServer,
 };
 use error::MyResult;
-use log::info;
 use lsm_tree::{Config, Tree};
 use serde::{Deserialize, Serialize};
 use serde_json::{json, Value};
@@ -137,13 +136,13 @@ async fn main() -> std::io::Result<()> {
     let port = port.parse::<u16>().expect("invalid port");
 
     let data_folder = std::env::var("DATA_FOLDER").unwrap_or(".data".into());
-    info!("Opening database at {data_folder}");
+    log::info!("Opening database at {data_folder}");
     let db = Config::new(&data_folder)
         .block_cache_size(25_600) // 100 MB
         .open()
         .expect("failed to open db");
 
-    info!("Starting on port {port}");
+    log::info!("Starting on port {port}");
 
     HttpServer::new(move || {
         App::new()

examples/actix-triplestore/.gitignore

@@ -0,0 +1,3 @@
+/target
+.data
+

examples/actix-triplestore/Cargo.toml

@@ -0,0 +1,14 @@
+[package]
+name = "actix-triplestore"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+actix-web = "4"
+log = { version = "0.4", features = ["release_max_level_info"] }
+env_logger = "0.10.0"
+lsm_tree = { path = "../../" }
+serde = { version = "1.0.164", features = ["derive"] }
+serde_json = "1.0.99"

examples/actix-triplestore/README.md

@@ -0,0 +1,37 @@
+# actix-triplestore
+
+This example uses `lsm-tree`, `actix_web` and `serde_json` to provide a simple triplestore with a JSON REST API.
+
+## REST API
+
+### `POST /{subject}`
+
+Upserts a subject.
+
+```json
+{
+  "item": {
+    "name": "John"
+  }
+}
+```
+
+### `POST /{subject}/{verb}/{object}`
+
+Upserts a relation between a subject and an object.
+
+```json
+{
+  "item": {
+    "created_at": 12345
+  }
+}
+```
+
+### `GET /{subject}`
+
+Returns a subject if it exists.
+
+### `GET /{subject}/{verb}?limit=1234`
+
+Returns a list of edges, their data and the corresponding nodes.

examples/actix-triplestore/src/error.rs

@@ -0,0 +1,24 @@
+use actix_web::{HttpResponse, ResponseError};
+
+#[derive(Debug)]
+pub struct MyError(lsm_tree::Error);
+
+impl std::fmt::Display for MyError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{:?}", self.0)
+    }
+}
+
+impl ResponseError for MyError {
+    fn error_response(&self) -> HttpResponse {
+        HttpResponse::InternalServerError().body("Internal Server Error")
+    }
+}
+
+impl From<lsm_tree::Error> for MyError {
+    fn from(value: lsm_tree::Error) -> Self {
+        Self(value)
+    }
+}
+
+pub type MyResult<T> = Result<T, MyError>;

examples/actix-triplestore/src/main.rs

@@ -0,0 +1,196 @@
+mod error;
+
+use actix_web::{
+    get,
+    middleware::Logger,
+    post,
+    web::{self},
+    App, HttpResponse, HttpServer,
+};
+use error::MyResult;
+use lsm_tree::{Config, Tree};
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+
+#[derive(Deserialize)]
+struct PrefixQueryParams {
+    limit: Option<u32>,
+}
+
+// This struct represents state
+struct AppState {
+    db: Tree,
+}
+
+#[derive(Serialize, Deserialize)]
+struct InsertBody {
+    item: Value,
+}
+
+#[derive(Serialize, Deserialize)]
+struct BulkBody {
+    upsert: Option<Vec<(String, String, String, Value)>>,
+    remove: Option<Vec<(String, String, String)>>,
+}
+
+#[post("/{subject}")]
+async fn insert_subject(
+    data: web::Data<AppState>,
+    body: web::Json<InsertBody>,
+    path: web::Path<String>,
+) -> MyResult<HttpResponse> {
+    eprintln!("INSERT SUBJECT");
+
+    let before = std::time::Instant::now();
+
+    let subject_key = path.into_inner();
+
+    data.db.insert(
+        format!("s:{subject_key}"),
+        serde_json::to_string(&body.item).unwrap(),
+    )?;
+
+    data.db.flush()?;
+
+    Ok(HttpResponse::Ok()
+        .append_header(("x-took-ms", before.elapsed().as_millis().to_string()))
+        .body("OK"))
+}
+
+#[post("/{subject}/{verb}/{object}")]
+async fn insert_relation(
+    data: web::Data<AppState>,
+    body: web::Json<InsertBody>,
+    path: web::Path<(String, String, String)>,
+) -> MyResult<HttpResponse> {
+    eprintln!("INSERT RELATION");
+
+    let before = std::time::Instant::now();
+
+    let (subject_key, verb_key, object_key) = path.into_inner();
+
+    data.db.insert(
+        format!("v:s:{subject_key}:v:{verb_key}:o:{object_key}"),
+        serde_json::to_string(&body.item).unwrap(),
+    )?;
+
+    data.db.flush()?;
+
+    Ok(HttpResponse::Ok()
+        .append_header(("x-took-ms", before.elapsed().as_millis().to_string()))
+        .body("OK"))
+}
+
+#[get("/{subject}")]
+async fn get_subject(path: web::Path<String>, data: web::Data<AppState>) -> MyResult<HttpResponse> {
+    let before = std::time::Instant::now();
+
+    let subject_key = path.into_inner();
+    let key = format!("s:{subject_key}");
+
+    match data.db.get(key)? {
+        Some(item) => Ok(HttpResponse::Ok()
+            .append_header(("x-took-ms", before.elapsed().as_millis().to_string()))
+            .content_type("application/json; utf-8")
+            .body(item.value)),
+        None => Ok(HttpResponse::NotFound()
+            .append_header(("x-took-ms", before.elapsed().as_millis().to_string()))
+            .content_type("text:html; utf-8")
+            .body("Not found")),
+    }
+}
+
+#[get("/{subject}/{verb}")]
+async fn list_by_verb(
+    path: web::Path<(String, String)>,
+    data: web::Data<AppState>,
+    query: web::Query<PrefixQueryParams>,
+) -> MyResult<HttpResponse> {
+    let (subject, verb) = path.into_inner();
+
+    let before = std::time::Instant::now();
+
+    let all = data
+        .db
+        .prefix(format!("v:s:{subject}:v:{verb}:"))?
+        .into_iter()
+        .take(query.limit.unwrap_or(10_000) as usize)
+        .collect::<Vec<_>>();
+
+    // let mut joined = String::with_capacity(100);
+
+    let count = all.len();
+
+    let mut edges = vec![];
+
+    for item in all {
+        let item = item?;
+
+        let key = std::str::from_utf8(&item.key).unwrap();
+        let verb_key = key.split(':').nth(4).unwrap();
+        let object_key = key.split(':').nth(6).unwrap();
+        let relation_data =
+            serde_json::from_str::<serde_json::Value>(std::str::from_utf8(&item.value).unwrap())
+                .unwrap();
+
+        let object_data = data
+            .db
+            .get(format!("s:{object_key}"))?
+            .map(|x| {
+                serde_json::from_str::<serde_json::Value>(std::str::from_utf8(&x.value).unwrap())
+                    .unwrap()
+            })
+            .unwrap_or(serde_json::Value::Null);
+
+        edges.push(serde_json::json!({
+            "key": verb_key,
+            "data": relation_data,
+            "node": {
+                "key": object_key,
+                "data": object_data,
+            },
+        }));
+    }
+
+    let body = serde_json::json!({
+        "edges": edges,
+    });
+
+    Ok(HttpResponse::Ok()
+        .append_header(("x-count", count.to_string()))
+        .append_header(("x-took-ms", before.elapsed().as_millis().to_string()))
+        .content_type("application/json; utf-8")
+        .body(serde_json::to_string(&body).unwrap()))
+}
+
+#[actix_web::main]
+async fn main() -> std::io::Result<()> {
+    env_logger::Builder::new()
+        .filter_level(log::LevelFilter::Info)
+        .init();
+
+    let port = std::env::var("PORT").unwrap_or("8000".into());
+    let port = port.parse::<u16>().expect("invalid port");
+
+    let data_folder = std::env::var("DATA_FOLDER").unwrap_or(".data".into());
+    log::info!("Opening database at {data_folder}");
+    let db = Config::new(&data_folder)
+        .block_cache_size(25_600) // 100 MB
+        .open()
+        .expect("failed to open db");
+
+    log::info!("Starting on port {port}");
+
+    HttpServer::new(move || {
+        App::new()
+            .wrap(Logger::new("%r %s - %{User-Agent}i"))
+            .app_data(web::Data::new(AppState { db: db.clone() }))
+            .service(insert_subject)
+            .service(insert_relation)
+            .service(get_subject)
+            .service(list_by_verb)
+    })
+    .bind(("127.0.0.1", port))?
+    .run()
+    .await
+}