quoter/src/main.rs

205 lines
6.8 KiB
Rust
Raw Normal View History

2023-10-02 15:36:02 +00:00
use actix_web::{web, App, HttpResponse, HttpServer, Responder, web::Bytes};
2023-10-02 09:22:04 +00:00
use redis::{Client, AsyncCommands};
use reqwest::Client as HTTPClient;
use serde::{Serialize, Deserialize};
2023-09-28 10:12:07 +00:00
use serde_json::Value;
2023-09-27 23:08:48 +00:00
use std::collections::HashMap;
use std::env;
2023-10-02 10:16:57 +00:00
use std::error::Error;
2023-10-02 12:37:56 +00:00
use futures::StreamExt;
2023-10-02 09:22:04 +00:00
use tokio::sync::broadcast::{self, Receiver};
2023-10-02 13:55:31 +00:00
use uuid::Uuid;
use chrono::Utc;
2023-09-27 23:08:48 +00:00
2023-10-02 10:16:57 +00:00
#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
enum PayloadKind {
NewMessage,
NewFollower,
NewShout,
NewApproval,
NewComment,
NewRate
}
2023-10-02 09:22:04 +00:00
#[derive(Debug, Serialize, Deserialize)]
2023-09-28 10:12:07 +00:00
struct Payload {
2023-10-02 09:22:04 +00:00
chat_id: Option<String>,
shout_id: Option<i32>,
2023-10-02 10:16:57 +00:00
author_id: Option<i32>,
topic_id: Option<i32>,
reaction_id: Option<i32>,
community_id: Option<i32>,
kind: PayloadKind,
body: String,
2023-09-27 23:08:48 +00:00
}
2023-10-02 10:16:57 +00:00
async fn get_auth_id(token: &str) -> Result<i32, Box<dyn Error>> {
2023-09-28 10:12:07 +00:00
let api_base = env::var("API_BASE")?;
2023-10-02 10:16:57 +00:00
let gql = match api_base.contains("v2") {
true => r#"mutation { getSession { user { id } } }"#, // v2
_ => r#"query { sessiom { user { id } } }"# // authorizer
2023-10-02 09:22:04 +00:00
};
let client = HTTPClient::new();
2023-09-28 10:12:07 +00:00
let response = client
.post(api_base)
2023-10-02 09:22:04 +00:00
.bearer_auth(token) // NOTE: auth token is here
2023-09-28 10:12:07 +00:00
.body(gql)
2023-09-27 23:08:48 +00:00
.send()
.await?;
2023-10-02 09:22:04 +00:00
let response_body: Value = response.json().await?;
2023-09-28 10:12:07 +00:00
let id = response_body["data"]["getSession"]["user"]["id"]
.as_i64()
.ok_or("Failed to get user id by token")? as i32;
Ok(id)
2023-09-27 23:08:48 +00:00
}
2023-10-02 18:55:10 +00:00
// Функция создает первый чат для пользователя. В чате два участника: пользователь и автор с идентификатором 1.
// Данные чата сохраняются в Redis.
async fn create_first_chat(author_id: i32, con: &mut redis::aio::Connection) -> Result<Vec<String>, Box<dyn Error>> {
2023-10-02 20:12:33 +00:00
let chat_id = Uuid::new_v4().to_string();
let members = vec![author_id.to_string(), "1".to_string()];
let timestamp = Utc::now().timestamp();
let chat = serde_json::json!({
"id": chat_id.clone(),
"admins": members.clone(),
"members": members.clone(),
"title": "",
"createdBy": author_id,
"createdAt": timestamp,
"updatedAt": timestamp,
});
2023-10-02 13:47:22 +00:00
let _: () = redis::pipe()
.atomic()
2023-10-02 16:18:23 +00:00
.cmd("SADD")
.arg(format!("chats_by_author/{}", author_id))
.arg(&chat_id)
.ignore()
2023-10-02 13:47:22 +00:00
.set(format!("chats/{}", chat_id), chat.to_string())
2023-10-02 16:18:23 +00:00
.ignore()
2023-10-02 13:47:22 +00:00
.set(format!("chats/{}/next_message_id", chat_id), "0")
2023-10-02 16:18:23 +00:00
.ignore()
2023-10-02 18:55:10 +00:00
.query_async(con)
.await?;
2023-10-02 20:12:33 +00:00
2023-10-02 18:55:10 +00:00
Ok(vec![chat.to_string()])
2023-10-02 13:47:22 +00:00
}
2023-09-28 10:12:07 +00:00
async fn sse_handler(
token: web::Path<String>,
2023-10-02 12:37:56 +00:00
mut rx: web::Data<Receiver<String>>,
2023-10-02 09:22:04 +00:00
redis: web::Data<Client>,
2023-09-28 10:12:07 +00:00
) -> impl Responder {
2023-10-02 10:16:57 +00:00
let author_id = match get_auth_id(&token).await {
2023-09-27 23:08:48 +00:00
Ok(id) => id,
Err(e) => {
2023-10-02 13:55:31 +00:00
eprintln!("TOKEN check failed: {}", e);
2023-10-02 09:22:04 +00:00
return HttpResponse::Unauthorized().finish();
2023-09-27 23:08:48 +00:00
}
};
2023-10-02 18:55:10 +00:00
let mut con = match redis.get_async_connection().await {
Ok(con) => con,
Err(e) => {
eprintln!("Failed to get async connection: {}", e);
return HttpResponse::InternalServerError().finish();
}
};
2023-10-02 20:27:17 +00:00
let _ = match con.sadd("authors-online", &author_id).await {
2023-10-02 18:55:10 +00:00
Ok(_) => (),
Err(e) => {
eprintln!("Failed to add author to online list: {}", e);
return HttpResponse::InternalServerError().finish();
}
};
2023-10-02 09:22:04 +00:00
2023-10-02 20:12:33 +00:00
let chats: Vec<String> = match con.smembers::<String, Vec<String>>(format!("chats_by_author/{}", author_id)).await {
2023-10-02 13:47:22 +00:00
Ok(chats) => {
if chats.is_empty() {
2023-10-02 18:55:10 +00:00
match create_first_chat(author_id, &mut con).await {
Ok(chat) => chat,
Err(e) => {
eprintln!("Failed to create first chat: {}", e);
return HttpResponse::InternalServerError().finish();
}
}
2023-10-02 13:47:22 +00:00
} else {
chats
}
},
2023-10-02 18:55:10 +00:00
Err(e) => {
eprintln!("Failed to get chats by author: {}", e);
match create_first_chat(author_id, &mut con).await {
2023-10-02 20:22:05 +00:00
Ok(chat) => chat,
Err(e) => {
eprintln!("Failed to create first chat: {}", e);
return HttpResponse::InternalServerError().finish();
2023-10-02 18:55:10 +00:00
}
2023-10-02 20:22:05 +00:00
}
2023-10-02 18:55:10 +00:00
}
2023-10-02 13:47:22 +00:00
};
2023-09-28 10:12:07 +00:00
2023-10-02 09:22:04 +00:00
let mut pubsub = con.into_pubsub();
for chat_id in chats {
2023-10-02 18:55:10 +00:00
if let Err(e) = pubsub.subscribe(format!("chat:{}", chat_id)).await {
eprintln!("Failed to subscribe to chat: {}", e);
return HttpResponse::InternalServerError().finish();
}
2023-09-28 10:12:07 +00:00
}
2023-10-02 18:55:10 +00:00
let _: () = match con.srem("authors-online", &author_id).await {
Ok(_) => (),
Err(e) => {
eprintln!("Failed to remove author from online list: {}", e);
return HttpResponse::InternalServerError().finish();
}
};
2023-10-02 09:22:04 +00:00
2023-10-02 18:55:10 +00:00
let server_event = match rx.recv().await {
Ok(event) => event,
Err(e) => {
eprintln!("Failed to receive server event: {}", e);
return HttpResponse::InternalServerError().finish();
}
};
2023-10-02 15:27:55 +00:00
let server_event_stream = futures::stream::once(async move { Ok::<_, actix_web::Error>(Bytes::from(server_event)) });
2023-10-02 09:22:04 +00:00
HttpResponse::Ok()
2023-10-02 11:24:45 +00:00
.append_header(("content-type", "text/event-stream"))
2023-10-02 15:27:55 +00:00
.streaming(server_event_stream)
2023-09-27 23:08:48 +00:00
}
#[actix_web::main]
async fn main() -> std::io::Result<()> {
2023-10-02 12:37:56 +00:00
let (tx, _rx) = broadcast::channel(100);
2023-10-02 18:55:10 +00:00
let redis_url = env::var("REDIS_URL").expect("REDIS_URL must be set");
let client = redis::Client::open(redis_url).expect("Failed to open Redis client");
2023-09-28 10:12:07 +00:00
let _handle = tokio::spawn(async move {
2023-10-02 18:55:10 +00:00
let mut conn = client.get_async_connection().await.expect("Failed to get async connection");
2023-09-28 10:12:07 +00:00
let mut pubsub = conn.into_pubsub();
2023-10-02 18:55:10 +00:00
pubsub.subscribe("new_follower").await.expect("Failed to subscribe to new_follower");
pubsub.subscribe("new_shout").await.expect("Failed to subscribe to new_shout");
pubsub.subscribe("new_reaction").await.expect("Failed to subscribe to new_reaction");
2023-09-28 10:12:07 +00:00
2023-10-02 12:30:53 +00:00
while let Some(msg) = pubsub.on_message().next().await {
2023-10-02 18:55:10 +00:00
let payload: HashMap<String, String> = msg.get_payload().expect("Failed to get payload");
tx.send(serde_json::to_string(&payload).expect("Failed to serialize payload")).expect("Failed to send payload");
2023-09-28 10:12:07 +00:00
}
});
2023-09-27 23:08:48 +00:00
HttpServer::new(move || {
2023-10-02 12:35:16 +00:00
let rx = tx.subscribe();
2023-09-27 23:08:48 +00:00
App::new()
2023-10-02 12:35:16 +00:00
.app_data(web::Data::new(rx))
2023-10-02 11:48:27 +00:00
.app_data(web::Data::new(client.clone()))
2023-10-02 18:55:10 +00:00
.route("/presence/{token}", web::get().to(sse_handler))
2023-09-27 23:08:48 +00:00
})
.bind("127.0.0.1:8080")?
.run()
.await
}