Init

.gitignore

@@ -0,0 +1,3 @@
+/target
+nostr.db/
+.idea/

Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "nostr_services_rs"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+tokio = { version = "1.38.1", features = ["rt", "rt-multi-thread", "macros"] }
+nostr = "0.33.0"
+rocket = { version = "0.5.1", features = ["json"] }
+serde = { version = "1.0.204", features = ["derive"] }
+log = "0.4.22"
+pretty_env_logger = "0.5.0"
+anyhow = "1.0.86"
+nostr-database = { version = "0.33.0", features = ["flatbuf"] }
+sled = "0.34.7"

src/events.rs

@@ -0,0 +1,44 @@
+use anyhow::Error;
+use nostr::{Event, EventId, JsonUtil, Kind};
+use nostr_database::{DatabaseError, DynNostrDatabase, NostrDatabase};
+use rocket::{Data, Route, State};
+use rocket::http::Status;
+use rocket::serde::json::Json;
+use crate::store::SledDatabase;
+
+pub fn routes() -> Vec<Route> {
+    routes![import_event, get_event]
+}
+
+#[rocket::post("/event", data = "<data>")]
+async fn import_event(
+    db: &State<SledDatabase>,
+    data: Json<Event>,
+) -> Status {
+    if data.verify().is_err() {
+        return Status::InternalServerError;
+    }
+    if let Ok(v) = db.save_event(&data).await {
+        match v {
+            true => Status::Ok,
+            false => Status::Conflict
+        }
+    } else {
+        Status::InternalServerError
+    }
+}
+
+#[rocket::get("/event/<id>")]
+async fn get_event(
+    db: &State<SledDatabase>,
+    id: &str,
+) -> Option<Json<Event>> {
+    let id = match EventId::parse(id) {
+        Ok(i) => i,
+        _ => return None
+    };
+    match db.event_by_id(id).await {
+        Ok(ev) => Some(Json::from(ev)),
+        _ => None
+    }
+}

src/main.rs

@@ -0,0 +1,33 @@
+#[macro_use]
+extern crate rocket;
+
+use anyhow::Error;
+use nostr::Event;
+use nostr_database::{DynNostrDatabase, NostrDatabase};
+use rocket::shield::Shield;
+
+use crate::store::SledDatabase;
+
+mod events;
+mod store;
+
+#[rocket::main]
+async fn main() -> Result<(), Error> {
+    pretty_env_logger::init();
+
+    let db = SledDatabase::new("nostr.db");
+
+    let rocket = rocket::Rocket::build()
+        .manage(db)
+        .attach(Shield::new()) // disable
+        .mount("/", events::routes())
+        .launch()
+        .await;
+
+    if let Err(e) = rocket {
+        error!("Rocker error {}", e);
+        Err(Error::from(e))
+    } else {
+        Ok(())
+    }
+}

src/store.rs

@@ -0,0 +1,84 @@
+use std::fmt::Debug;
+use std::sync::Arc;
+
+use nostr::{Event, EventId};
+use nostr_database::{FlatBufferBuilder, FlatBufferDecode, FlatBufferEncode};
+use sled::{Db, IVec};
+use tokio::sync::Mutex;
+
+#[derive(Clone, Debug)]
+pub struct SledDatabase {
+    db: Db,
+    fbb: Arc<Mutex<FlatBufferBuilder<'static>>>,
+}
+
+impl SledDatabase {
+    pub fn new(path: &str) -> Self {
+        Self {
+            db: sled::open(path).unwrap(),
+            fbb: Arc::new(Mutex::new(FlatBufferBuilder::with_capacity(70_000))),
+        }
+    }
+}
+
+impl SledDatabase {
+    pub async fn save_event(&self, event: &Event) -> Result<bool, anyhow::Error> {
+        let mut fbb = self.fbb.lock().await;
+        if let Err(e) = self.db.insert(event.id.as_bytes(), event.encode(&mut fbb)) {
+            return Err(anyhow::Error::new(e));
+        }
+        if event.kind.is_replaceable() {
+            self.write_replaceable_index(event)?;
+        }
+        Ok(true)
+    }
+
+    fn write_replaceable_index(&self, event: &Event) -> Result<bool, anyhow::Error> {
+        let rpk = Self::replaceable_index_key(event);
+
+        if let Err(e) = self.db.update_and_fetch(rpk, |prev| {
+            if let Some(prev) = prev {
+                let timestamp: u64 = u64::from_be_bytes(prev[..8].try_into().unwrap());
+                if timestamp < event.created_at.as_u64() {
+                    let new_val = Self::replaceable_index_value(event);
+                    Some(IVec::from(new_val.as_slice()))
+                } else {
+                    None
+                }
+            } else {
+                let new_val = Self::replaceable_index_value(event);
+                Some(IVec::from(new_val.as_slice()))
+            }
+        }) {
+            return Err(anyhow::Error::new(e));
+        }
+        Ok(false)
+    }
+
+    fn replaceable_index_key(event: &Event) -> [u8; 36] {
+        let mut rpk = [0; 4 + 32]; // kind:pubkey
+        rpk[..4].copy_from_slice(&event.kind.as_u32().to_be_bytes());
+        rpk[4..].copy_from_slice(&event.pubkey.to_bytes());
+        rpk
+    }
+
+    fn replaceable_index_value(event: &Event) -> [u8; 40] {
+        let mut new_val = [0; 8 + 32]; // timestamp:event_id
+        new_val[..8].copy_from_slice(&event.created_at.as_u64().to_be_bytes());
+        new_val[8..].copy_from_slice(event.id.as_bytes());
+        new_val
+    }
+
+    pub async fn event_by_id(&self, event_id: EventId) -> Result<Event, anyhow::Error> {
+        match self.db.get(event_id.as_bytes()) {
+            Ok(v) => match v {
+                Some(v) => match Event::decode(&v) {
+                    Ok(v) => Ok(v),
+                    Err(e) => Err(anyhow::Error::new(e))
+                },
+                None => Err(anyhow::Error::msg("Not Found"))
+            }
+            Err(e) => Err(anyhow::Error::new(e))
+        }
+    }
+}