quoter/src/main.rs

use actix_web::{web, App, HttpResponse, HttpServer, Responder};
use redis::{Client, AsyncCommands};
use reqwest::Client as HTTPClient;
use serde::{Serialize, Deserialize};
use serde_json::Value;
use std::collections::HashMap;
use std::env;
use futures::FutureExt;
use tokio::sync::broadcast::{self, Receiver};

#[derive(Debug, Serialize, Deserialize)]
struct Payload {
    chat_id: Option<String>,
    shout_id: Option<i32>,

}

// Получаем id автора из токена
async fn get_auth_id(token: &str) -> Result<i32, Box<dyn std::error::Error>> {
    let api_base = env::var("API_BASE")?;
    let gql = match api_base.contains("auth") {
        true => r#"query { sessiom { user { id } } }"#,
        _ => r#"mutation { getSession { user { id } } }"#
    };
    let client = HTTPClient::new();
    let response = client
        .post(api_base)
        .bearer_auth(token) // NOTE: auth token is here
        .body(gql)
        .send()
        .await?;
    let response_body: Value = response.json().await?;
    let id = response_body["data"]["getSession"]["user"]["id"]
        .as_i64()
        .ok_or("Failed to get user id by token")? as i32;
    Ok(id)
}

// Обработчик SSE
async fn sse_handler(
    token: web::Path<String>,
    rx: web::Data<Receiver<String>>,
    redis: web::Data<Client>,
) -> impl Responder {
    let author_id = match get_author_id(&token).await {
        Ok(id) => id,
        Err(e) => {
            eprintln!("Не удалось проверить токен: {}", e);
            return HttpResponse::Unauthorized().finish();
        }
    };

    let mut con = redis.get_async_connection().await.unwrap();
    let _: () = con
        .sadd("authors-online", &author_id)
        .await
        .unwrap();

    // Получаем все чаты авторизованного пользователя
    let chats: Vec<String> = con
        .smembers(format!("chats_by_author/{}", author_id))
        .await
        .unwrap();

    // Подписываемся на каждый чат в Redis Pub/Sub
    let mut pubsub = con.into_pubsub();
    for chat_id in chats {
        pubsub.subscribe(format!("message:{}", chat_id)).await.unwrap();
    }

    let server_event = rx.get_ref().clone().into_stream().map(|result| {
        match result {
            Ok(payload) => {
                let payload: Payload = serde_json::from_str(&payload).unwrap();
                Ok::<_, actix_web::Error>(web::Bytes::from(format!("data: {:?}\n\n", payload.chat_id)))
            }
            Err(_) => Err(actix_web::Error::from(())),
        }
    });

    let _: () = con
        .srem("authors-online", &author_id)
        .await
        .unwrap();

    HttpResponse::Ok()
        .header("content-type", "text/event-stream")
        .streaming(server_event)
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    // Создаем одно соединение с Redis Pub/Sub и канал вещания
    let (tx, rx) = broadcast::channel(100);
    let redis_url = env::var("REDIS_URL").unwrap();
    let client = redis::Client::open(redis_url).unwrap();
    let _handle = tokio::spawn(async move {
        let mut conn = client.get_async_connection().await.unwrap();
        let mut pubsub = conn.into_pubsub();

        // новый подписчик автора
        pubsub.subscribe("new_follower").await.unwrap();
        // в теме, автор, сообщество
        pubsub.subscribe("new_shout").await.unwrap();
        // оценка, комментарий, одобрение в криках автора и реагировавших криках
        pubsub.subscribe("new_reaction").await.unwrap();

        while let Some(msg) = pubsub.on_message().next().await {
            let payload: HashMap<String, String> = msg.get_payload().unwrap();
            tx.send(serde_json::to_string(&payload).unwrap()).unwrap();
        }
    });

    HttpServer::new(move || {
        App::new()
            .data(rx.clone())
            .data(client.clone())
            .route("/aware/{token}", web::get().to(sse_handler))
    })
    .bind("127.0.0.1:8080")?
    .run()
    .await
}