commit 5a988fd14016aaa2db56cb6d00db7ca132d3c38b
parent 04811cc6530c75ef64f36f11ddf18453c5d95ad9
Author: quantumish <freifeld.david@gmail.com>
Date: Thu, 6 Oct 2022 11:00:26 -0700
MVP
Diffstat:
5 files changed, 354 insertions(+), 61 deletions(-)
diff --git a/README.md b/README.md
@@ -0,0 +1,39 @@
+# blatherskite
+
+# Dependencies
+You'll need to install CassandraDB for this: you can do that by running:
+```
+brew install cassandra
+```
+
+You also need to install the Rust language:
+```
+curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh
+```
+
+# Usage
+Start by launching the database in the background with
+```
+cassandra -f
+```
+Then, launch `chatterbox` - the service for sending/getting messages - and `scuttlebutt` - the service for everything else.
+```
+cargo run -p chatterbox &
+cargo run -p scuttlebutt &
+```
+
+## Scuttlebutt
+Scuttlebutt is an HTTP service that handles the creation, deletion, and updating of groups/channels/users as well as misc others.
+
+The various methods and objects are documented at `localhost:3000`, and the basic usage flow is something like:
+- `POST /api/user` to make a user, which will return a User object (see Schemas on the docs)
+- `GET /api/login` to login with said user. This will return a JWT that you'll use to authenticate future requests. This token will expire in a day!
+- Whatever requests you'd like at that point! Authenticate by including a `ScuttleKey` header with the token you got.
+
+## Chatterbox
+Chatterbox is a websocket service used for sending and receiving messages. To use:
+- Connect to the websocket at `ws://localhost:3001/ws/whee`
+- Send authentication in the form of `{"hash": "YOUR_PASSWORD_HASH", "id": "YOUR_ID"}`
+- Then use the websocket as normal!
+ - Send message requests in the form of `{"content": "whee", "channel": "CHANNEL_ID"}`
+ - Recieve messages!
diff --git a/chatterbox/Cargo.toml b/chatterbox/Cargo.toml
@@ -6,7 +6,14 @@ edition = "2021"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
+anyhow = "1.0.65"
+cassandra-cpp = "1.1.0"
+chrono = "0.4.22"
futures-util = "0.3.24"
+hex = "0.4.3"
poem = { version = "1.3.43", features = ["websocket"] }
+rustflake = "0.1.1"
+serde = "1.0.145"
+serde_json = "1.0.85"
tokio = { version = "1.21.1", features = ["full"] }
tracing-subscriber = "0.3.15"
diff --git a/chatterbox/src/main.rs b/chatterbox/src/main.rs
@@ -1,57 +1,132 @@
/// Currently a modified version of `poem`'s default websocket-chat example
+use cassandra_cpp::*;
use futures_util::{SinkExt, StreamExt};
use poem::{
- get, handler,
- listener::TcpListener,
- web::{
- websocket::{Message, WebSocket},
- Data, Path,
- },
- EndpointExt, IntoResponse, Route, Server,
+ get, handler,
+ listener::TcpListener,
+ web::{
+ websocket::{Message, WebSocket},
+ Data, Path,
+ },
+ EndpointExt, IntoResponse, Route, Server,
};
+use rustflake::Snowflake;
+use serde_json::Value;
+use std::result::Result;
+use serde::{Serialize, Deserialize};
+use chrono::{DateTime, Local};
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct MessageObj {
+ pub id: i64,
+ pub channel: i64,
+ pub author: i64,
+ pub content: String,
+}
+
+pub fn gen_id() -> i64 {
+ static STATE: std::sync::Mutex<Option<Snowflake>> = std::sync::Mutex::new(None);
+
+ STATE
+ .lock()
+ .unwrap()
+ .get_or_insert_with(|| Snowflake::new(1_564_790_400_000, 2, 1))
+ .generate()
+}
+
+const KEYSPC: &'static str = "bsk";
+
+fn setup_db() -> Session {
+ let contact_points = "127.0.0.1";
+ let mut cluster = Cluster::default();
+ cluster.set_contact_points(contact_points).unwrap();
+ cluster.set_load_balance_round_robin();
+ cluster.connect().unwrap()
+}
#[handler]
fn ws(
- Path(name): Path<String>,
- ws: WebSocket,
- sender: Data<&tokio::sync::broadcast::Sender<String>>,
+ Path(name): Path<String>,
+ ws: WebSocket,
+ sender: Data<&tokio::sync::broadcast::Sender<String>>,
) -> impl IntoResponse {
- let sender = sender.clone();
- let mut receiver = sender.subscribe();
- ws.on_upgrade(move |socket| async move {
- let (mut sink, mut stream) = socket.split();
-
- tokio::spawn(async move {
- while let Some(Ok(msg)) = stream.next().await {
- if let Message::Text(text) = msg {
- if sender.send(format!("{}: {}", name, text)).is_err() {
- break;
- }
- }
- }
- });
-
- tokio::spawn(async move {
- while let Ok(msg) = receiver.recv().await {
- if sink.send(Message::Text(msg)).await.is_err() {
- break;
- }
- }
- });
- })
+ let sender = sender.clone();
+ ws.on_upgrade(move |socket| async move {
+ let mut receiver = sender.subscribe();
+ let (mut sink, mut stream) = socket.split();
+
+ tokio::spawn(async move {
+ let sess = setup_db();
+ let mut user: Option<Value> = None;
+ while let Some(Ok(msg)) = stream.next().await {
+ if let Message::Text(auth) = msg {
+ let req: Value = serde_json::from_str(&auth).unwrap();
+ let res = sess.execute(&stmt!(&format!(
+ "SELECT hash FROM {}.users WHERE id={};",
+ KEYSPC, req["id"].as_i64().unwrap(),
+ ))).wait().unwrap();
+ let row = res.first_row().unwrap();
+ let db_hash: String = row.get(0).unwrap();
+ if hex::decode(db_hash).unwrap() != hex::decode(req["hash"].as_str().unwrap()).unwrap() {
+ return;
+ }
+ user = Some(req);
+ break
+ }
+ }
+ while let Some(Ok(mesg)) = stream.next().await {
+ if let Message::Text(text) = mesg {
+ let id = gen_id();
+ let req: Value = serde_json::from_str(&text).unwrap();
+ let msg = MessageObj {
+ id,
+ content: req["content"].as_str().unwrap().to_string(),
+ author: user.clone().unwrap()["id"].as_i64().unwrap(),
+ channel: req["channel"].as_i64().unwrap(),
+ };
+ let now = Local::now();
+ let res = sess.execute(&stmt!(&format!(
+ "INSERT INTO {}.messages (channel, id, author, time, content) VALUES ({},{},{},'{}','{}');",
+ KEYSPC, msg.channel, gen_id(), msg.author, now.to_rfc3339(), msg.content
+ ))).wait().unwrap();
+ if sender.send(serde_json::to_string(&msg).unwrap()).is_err() {
+ break;
+ }
+ }
+ }
+ });
+
+ tokio::spawn(async move {
+ let sess = setup_db();
+ while let Ok(msg) = receiver.recv().await {
+ let req: Value = serde_json::from_str(&msg).unwrap();
+ let res = sess.execute(&stmt!(&format!(
+ "SELECT members FROM {}.channels WHERE id={};", KEYSPC, req["channel"].as_i64().unwrap(),
+ ))).wait().unwrap();
+ let row = res.first_row().unwrap();
+ let members: SetIterator = row.get(0).unwrap();
+ if !members.map(|i| i.get_i64().unwrap()).collect::<Vec<i64>>().contains(&req["author"].as_i64().unwrap()) {
+ continue
+ }
+ if sink.send(Message::Text(msg)).await.is_err() {
+ break;
+ }
+ }
+ });
+ })
}
#[tokio::main]
async fn main() -> Result<(), std::io::Error> {
- if std::env::var_os("RUST_LOG").is_none() {
- std::env::set_var("RUST_LOG", "poem=debug");
- }
- tracing_subscriber::fmt::init();
+ if std::env::var_os("RUST_LOG").is_none() {
+ std::env::set_var("RUST_LOG", "poem=debug");
+ }
+ tracing_subscriber::fmt::init();
- let app = Route::new().at(
- "/ws/:name",
- get(ws.data(tokio::sync::broadcast::channel::<String>(32).0)),
- );
+ let app = Route::new().at(
+ "/ws/:name",
+ get(ws.data(tokio::sync::broadcast::channel::<String>(32).0)),
+ );
- Server::new(TcpListener::bind("127.0.0.1:3001")).run(app).await
+ Server::new(TcpListener::bind("127.0.0.1:3001")).run(app).await
}
diff --git a/scuttlebutt/src/main.rs b/scuttlebutt/src/main.rs
@@ -84,7 +84,7 @@ impl Api {
session.execute(&stmt!(&format!("CREATE TABLE IF NOT EXISTS {keyspc}.groups (id bigint PRIMARY KEY, name text, members set<bigint>, channels set<bigint>);"))).wait().unwrap();
session.execute(&stmt!(&format!("CREATE TABLE IF NOT EXISTS {keyspc}.channels (id bigint PRIMARY KEY, name text, group bigint, members set<bigint>);"))).wait().unwrap();
session.execute(&stmt!(&format!("CREATE TABLE IF NOT EXISTS {keyspc}.user_groups (id bigint PRIMARY KEY, groups set<bigint>);"))).wait().unwrap();
- session.execute(&stmt!(&format!("CREATE TABLE IF NOT EXISTS {keyspc}.messages (group bigint, channel bigint, author bigint, time timestamp, content text, PRIMARY KEY ((group, channel)));"))).wait().unwrap();
+ session.execute(&stmt!(&format!("CREATE TABLE IF NOT EXISTS {keyspc}.messages (channel bigint, id bigint, author bigint, time timestamp, content text, PRIMARY KEY (channel, id)) WITH CLUSTERING ORDER BY (id DESC);"))).wait().unwrap();
Api {
sess: session,
@@ -103,13 +103,13 @@ impl Api {
}
- async fn __remove_channel_member(&self, cid: i64, uid: i64) {
+ fn __remove_channel_member(&self, cid: i64, uid: i64) {
self.sess.execute(&stmt!(&format!(
"UPDATE {}.channels SET members = members - {{{}}} WHERE id={};", self.kspc, uid, cid
))).wait().unwrap();
}
- async fn __remove_group_member(&self, gid: i64, uid: i64) {
+ fn __remove_group_member(&self, gid: i64, uid: i64) {
self.sess.execute(&stmt!(&format!(
"UPDATE {}.groups SET members = members - {{{}}} WHERE id={};", self.kspc, uid, gid
))).wait().unwrap();
@@ -119,7 +119,7 @@ impl Api {
let row = res.first_row().unwrap();
let channels: SetIterator = row.get(0).unwrap();
for channel in channels {
- self.__remove_channel_member(channel.get_i64().unwrap(), uid).await;
+ self.__remove_channel_member(channel.get_i64().unwrap(), uid);
}
self.sess.execute(&stmt!(&format!(
"UPDATE {}.user_groups SET groups = groups - {{{}}} WHERE id={};", self.kspc, gid, uid
@@ -257,7 +257,6 @@ impl Api {
self.sess.execute(&stmt!(&format!(
"DELETE FROM {}.users WHERE id={};", self.kspc, id
))).wait().unwrap();
-
let res = self.sess.execute(&stmt!(&format!(
"SELECT groups FROM {}.user_groups WHERE id={};",
self.kspc, auth.0.id
@@ -267,7 +266,7 @@ impl Api {
_ => res.first_row().unwrap().get(0).unwrap(),
};
for group in groups {
- self.__remove_group_member(group.get_i64().unwrap(), id).await;
+ self.__remove_group_member(group.get_i64().unwrap(), id);
}
self.sess.execute(&stmt!(&format!(
"DELETE FROM {}.user_groups WHERE id={};", self.kspc, id
@@ -280,7 +279,7 @@ impl Api {
async fn get_groups(&self, auth: Authorization) -> GroupsResponse {
use GroupsResponse::*;
let res = self.sess.execute(&stmt!(&format!(
- "SELECT id, name, members, channels FROM {}.user_groups WHERE id={};",
+ "SELECT groups FROM {}.user_groups WHERE id={};",
self.kspc, auth.0.id
))).wait().unwrap();
@@ -312,7 +311,7 @@ impl Api {
if let Err(e) = self.validate_id("groups", gid.0) {
return NotFound(PlainText("Didn't find group or experienced database error.".to_string()));
}
- self.__remove_group_member(gid.0, auth.0.id).await;
+ self.__remove_group_member(gid.0, auth.0.id);
Success
}
@@ -469,6 +468,21 @@ impl Api {
"UPDATE {}.groups SET members = members + {{{}}} WHERE id = {};",
self.kspc, uid.0, gid.0
))).wait().unwrap();
+ let res = self.sess.execute(&stmt!(&format!(
+ "SELECT channels FROM {}.groups WHERE id = {};",
+ self.kspc, gid.0
+ ))).wait().unwrap();
+ let row = res.first_row().unwrap();
+ let mut channels: SetIterator = row.get(0).unwrap();
+ let cid: i64 = channels.next().unwrap().get_i64().unwrap();
+ self.sess.execute(&stmt!(&format!(
+ "UPDATE {}.channels SET members = members + {{{}}} WHERE id = {};",
+ self.kspc, uid.0, cid
+ ))).wait().unwrap();
+ self.sess.execute(&stmt!(&format!(
+ "UPDATE {}.user_groups SET groups = groups + {{{}}} WHERE id = {};",
+ self.kspc, gid.0, uid.0
+ ))).wait().unwrap();
Success
}
@@ -486,7 +500,7 @@ impl Api {
} else if let Err(_) = self.validate_id("users", uid.0) {
return NotFound(PlainText("User not found".to_string()))
}
- self.__remove_group_member(gid.0, uid.0).await;
+ self.__remove_group_member(gid.0, uid.0);
Success
}
@@ -584,7 +598,7 @@ impl Api {
use ChannelResponse::*;
let res = self.sess.execute(&stmt!(&format!(
"SELECT name, group, members FROM {}.channels WHERE id={};", self.kspc, id.0
- ))).wait().unwrap();
+ ))).wait().unwrap();
let (name, group, members): (String, i64, SetIterator) = match res.row_count() {
1 => {
let row = res.first_row().unwrap();
@@ -593,7 +607,6 @@ impl Api {
0 => return NotFound,
_ => return InternalError(PlainText(UNUSUAL_ROW_ERROR.to_string()))
};
-
Success(Json(Channel {
id: id.0,
name,
@@ -605,13 +618,49 @@ impl Api {
#[oai(path = "/channel", method = "delete")]
/// Deletes a channel
async fn delete_channel(&self, auth: Authorization, id: Query<i64>) -> DeleteResponse {
- todo!()
+ use DeleteResponse::*;
+ if let Err(_) = self.validate_id("channels", id.0) {
+ NotFound(PlainText("Channel not found.".to_string()))
+ } else {
+ let res = self.sess.execute(&stmt!(&format!(
+ "SELECT group FROM {}.channels WHERE id={};", self.kspc, id.0
+ ))).wait().unwrap();
+ let row = res.first_row().unwrap();
+ let group: i64 = row.get(0).unwrap();
+ self.sess.execute(&stmt!(&format!(
+ "UPDATE {}.groups SET channels = channels - {{{}}} WHERE id = {};",
+ self.kspc, id.0, group
+ ))).wait().unwrap();
+ self.sess.execute(&stmt!(&format!(
+ "DELETE FROM {}.channels WHERE id = {};",
+ self.kspc, id.0
+ ))).wait().unwrap();
+ Success
+ }
}
#[oai(path = "/channel/members", method = "get")]
/// Gets the members that can access a channel
async fn get_channel_members(&self, auth: Authorization, id: Query<i64>) -> MembersResponse {
- todo!()
+ use MembersResponse::*;
+ let res = self.sess.execute(&stmt!(&format!(
+ "SELECT members FROM {}.channels WHERE id={};", self.kspc, id.0
+ ))).wait().unwrap();
+ let row = res.first_row().unwrap();
+ let members: SetIterator = row.get(0).unwrap();
+ let members_objs = members.map(|_| {
+ let res = self.sess.execute(&stmt!(&format!(
+ "SELECT id, name, email FROM {}.users WHERE id={};",
+ self.kspc, id.0
+ ))).wait().unwrap();
+ let row = res.first_row().unwrap();
+ User {
+ id: id.0,
+ username: row.get(1).unwrap(),
+ email: row.get(2).unwrap(),
+ }
+ }).collect::<Vec<User>>();
+ Success(Json(members_objs))
}
#[oai(path = "/channel/members", method = "put")]
@@ -622,7 +671,16 @@ impl Api {
id: Query<i64>,
uid: Query<i64>,
) -> GenericResponse {
- todo!()
+ use GenericResponse::*;
+ if let Err(_) = self.validate_id("channels", id.0) {
+ return NotFound(PlainText("Channel not found".to_string()))
+ } else if let Err(_) = self.validate_id("users", uid.0) {
+ return NotFound(PlainText("User not found".to_string()))
+ }
+ self.sess.execute(&stmt!(&format!(
+ "UPDATE {}.channels SET members = members + {{{}}} WHERE id={};", self.kspc, uid.0, id.0
+ ))).wait().unwrap();
+ Success
}
#[oai(path = "/channel/members", method = "delete")]
@@ -639,12 +697,12 @@ impl Api {
} else if let Err(_) = self.validate_id("users", uid.0) {
return NotFound(PlainText("User not found".to_string()))
}
- self.__remove_channel_member(cid.0, uid.0).await;
+ self.__remove_channel_member(cid.0, uid.0);
Success
}
#[oai(path = "/channel/message", method = "get")]
- /// Returns batch of messages in channel containing "term" starting at offset
+ /// Returns batch of messages in channel containing "term" in the last 100 messages
async fn search_channel(
&self,
auth: Authorization,
@@ -652,7 +710,26 @@ impl Api {
term: Query<String>,
off: Query<u64>,
) -> MessagesResponse {
- todo!()
+ use MessagesResponse::*;
+ if let Err(_) = self.validate_id("channels", cid.0) {
+ return NotFound(PlainText("Channel not found".to_string()))
+ }
+ let res = self.sess.execute(&stmt!(&format!(
+ "SELECT * FROM {}.messages WHERE channel={} LIMIT 100;",
+ self.kspc, cid.0
+ ))).wait().unwrap();
+ let messages = res.iter().filter(|row| {
+ let content: String = row.get(4).unwrap();
+ content.contains(&term.0)
+ }).map(|row| {
+ Message {
+ id: row.get(0).unwrap(),
+ author: row.get(2).unwrap(),
+ channel: row.get(1).unwrap(),
+ content: row.get(4).unwrap()
+ }
+ }).collect::<Vec<Message>>();
+ Success(Json(messages))
}
#[oai(path = "/channel/messages", method = "get")]
@@ -665,7 +742,23 @@ impl Api {
cid: Query<i64>,
num_msgs: Query<u64>,
) -> MessagesResponse {
- todo!()
+ use MessagesResponse::*;
+ if let Err(_) = self.validate_id("channels", cid.0) {
+ return NotFound(PlainText("Channel not found".to_string()))
+ }
+ let res = self.sess.execute(&stmt!(&format!(
+ "SELECT * FROM {}.messages WHERE channel={} LIMIT {};",
+ self.kspc, cid.0, num_msgs.0,
+ ))).wait().unwrap();
+ let messages = res.iter().map(|row| {
+ Message {
+ id: row.get(0).unwrap(),
+ author: row.get(2).unwrap(),
+ channel: row.get(1).unwrap(),
+ content: row.get(4).unwrap()
+ }
+ }).collect::<Vec<Message>>();
+ Success(Json(messages))
}
}
@@ -698,7 +791,7 @@ async fn main() -> Result<(), std::io::Error> {
.nest("/", ui)
.data(ServerKey::new_from_slice(&key.as_bytes()).unwrap());
// let cli = poem::test::TestClient::new(app);
- // let resp = cli.post("/api/login?id=234").body("abc").send().await;
+ // let resp = cli.post("/api/login?id=234").body("abc").send();
// resp.assert_status_is_ok();
// Ok(())
Server::new(TcpListener::bind("127.0.0.1:3000"))
diff --git a/test.py b/test.py
@@ -0,0 +1,79 @@
+import sys
+import requests
+import asyncio
+import websockets
+
+HASH = "6c6e2b0cfda80007e693d52b5956083ea68770e1310d0ed02d195cb14113b284"
+if sys.argv[1] == "setup":
+ r = requests.post(f'http://localhost:3000/api/user?name=quantum&email=test@example.com&hash={HASH}')
+ quantum = r.json()["id"]
+ r = requests.post(f'http://localhost:3000/api/login?id={quantum}', data=HASH, headers={"Content-Type": "text/plain"})
+ tok = r.text
+
+ r = requests.post(f'http://localhost:3000/api/user?name=jemoka&email=test@example.com&hash={HASH}')
+ jemoka = r.json()["id"]
+ r = requests.post(f'http://localhost:3000/api/user?name=exr0n&email=test@example.com&hash={HASH}')
+ exr0n = r.json()["id"]
+ r = requests.post(f'http://localhost:3000/api/user?name=enquirer&email=test@example.com&hash={HASH}')
+ enquirer = r.json()["id"]
+ r = requests.post(f'http://localhost:3000/api/user?name=zbuster&email=test@example.com&hash={HASH}')
+ zbuster = r.json()["id"]
+
+ print({"quantum": quantum, "exr0n": exr0n, "jemoka": jemoka, "enquirer": enquirer, "zbuster": zbuster})
+
+ r = requests.post(f'http://localhost:3000/api/group?name=testing', headers={"ScuttleKey": tok})
+ print(r.text)
+ testing = r.json()["id"]
+
+ r = requests.put(f'http://localhost:3000/api/group/members?gid={testing}&uid={zbuster}', headers={"ScuttleKey": tok})
+ r = requests.put(f'http://localhost:3000/api/group/members?gid={testing}&uid={exr0n}', headers={"ScuttleKey": tok})
+ r = requests.put(f'http://localhost:3000/api/group/members?gid={testing}&uid={jemoka}', headers={"ScuttleKey": tok})
+ r = requests.get(f'http://localhost:3000/api/group?id={testing}', headers={"ScuttleKey": tok})
+
+ r = requests.post(f'http://localhost:3000/api/group?name=whoo', headers={"ScuttleKey": tok})
+ whoo = r.json()["id"]
+ r = requests.put(f'http://localhost:3000/api/group/members?gid={whoo}&uid={zbuster}', headers={"ScuttleKey": tok})
+ r = requests.put(f'http://localhost:3000/api/group/members?gid={whoo}&uid={enquirer}', headers={"ScuttleKey": tok})
+
+ r = requests.post(f'http://localhost:3000/api/group?name=whee', headers={"ScuttleKey": tok})
+ whee = r.json()["id"]
+ r = requests.put(f'http://localhost:3000/api/group/members?gid={whee}&uid={enquirer}', headers={"ScuttleKey": tok})
+ r = requests.put(f'http://localhost:3000/api/group/members?gid={whee}&uid={jemoka}', headers={"ScuttleKey": tok})
+
+ exit(0)
+
+
+async def main():
+ ids = {'quantum': 420625705584431104, 'exr0n': 420625705622179840, 'jemoka': 420625705609596928, 'enquirer': 420625705634762752, 'zbuster': 420625705643151360}
+ name = sys.argv[1]
+ my_id = ids[name]
+ r = requests.post(f'http://localhost:3000/api/login?id={my_id}', data=HASH, headers={"Content-Type": "text/plain"})
+ tok = r.text
+
+ while True:
+ cmd = input("$ ").split(" ")
+ if cmd[0] == "groups":
+ r = requests.get(f'http://localhost:3000/api/user/groups', headers={"ScuttleKey": tok})
+ print(r.text, my_id)
+ for g in r.json():
+ print(f"{g['name']} ({g['id']})")
+ elif cmd[0] == "group":
+ r = requests.get(f'http://localhost:3000/api/group/channels?gid={cmd[1]}', headers={"ScuttleKey": tok})
+ for c in r.json():
+ print(f"{c['name']} ({c['id']})")
+ elif cmd[0] == "channel":
+ async with websockets.connect("ws://localhost:3001/ws/whee") as websocket:
+ await websocket.send(f'{{"hash": "{HASH}", "id": {my_id}}}')
+ await websocket.send(f'{{"content": "{name} says whee", "channel": {cmd[1]}}}')
+ while True:
+ print(await websocket.recv())
+
+loop = asyncio.get_event_loop()
+loop.run_until_complete(main())
+
+
+
+
+
+
+
