quoter/src/main.rs

use actix_web::{web, App, HttpResponse, HttpServer, Responder, web::Bytes};
use redis::{Client, AsyncCommands};
use std::collections::HashMap;
use std::env;
use futures::StreamExt;
use tokio::sync::broadcast::{self, Receiver};

pub mod data;

async fn sse_handler(
    token: web::Path<String>,
    rx: web::Data<Receiver<String>>,
    redis: web::Data<Client>,
) -> impl Responder {
    let author_id = match data::get_auth_id(&token).await {
        Ok(id) => id,
        Err(e) => {
            eprintln!("TOKEN check failed: {}", e);
            return HttpResponse::Unauthorized().finish();
        }
    };

    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();
        }
    };

    let _ = match con.sadd::<&str, &i32, usize>("authors-online", &author_id).await {
        Ok(_) => (),
        Err(e) => {
            eprintln!("Failed to add author to online list: {}", e);
            return HttpResponse::InternalServerError().finish();
        }
    };

    let chats: Vec<String> = match con.smembers::<String, Vec<String>>(format!("chats_by_author/{}", author_id)).await {
        Ok(chats) => {
            if chats.is_empty() {
                match data::create_first_chat(author_id, &mut con).await {
                    Ok(chat) => chat,
                    Err(e) => {
                        eprintln!("Failed to create first chat: {}", e);
                        return HttpResponse::InternalServerError().finish();
                    }
                }
            } else {
                chats
            }
        },
        Err(e) => {
            eprintln!("Failed to get chats by author: {}", e);
            match data::create_first_chat(author_id, &mut con).await {
                Ok(chat) => chat,
                Err(e) => {
                    eprintln!("Failed to create first chat: {}", e);
                    return HttpResponse::InternalServerError().finish();
                }
            }
        }
    };

    let mut pubsub = con.into_pubsub();
    for chat_id in &chats {
        if let Err(e) = pubsub.subscribe(format!("chat:{}", chat_id)).await {
            eprintln!("Failed to subscribe to chat: {}", e);
            return HttpResponse::InternalServerError().finish();
        }
    }

    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();
        }
    };

    let _ = match con.srem::<&str, &i32, usize>("authors-online", &author_id).await {
        Ok(_) => (),
        Err(e) => {
            eprintln!("Failed to remove author from online list: {}", e);
            return HttpResponse::InternalServerError().finish();
        }
    };

    // FIXME: cannot borrow data in an `Arc` as mutable
    // trait `DerefMut` is required to modify through a dereference, 
    // but it is not implemented for `Arc<tokio::sync::broadcast::Receiver<std::string::String>>`
    let server_event = match rx.recv().await { 
        Ok(event) => event,
        Err(e) => {
            eprintln!("Failed to receive server event: {}", e);
            return HttpResponse::InternalServerError().finish();
        }
    };
    let server_event_stream = futures::stream::once(async move { Ok::<_, actix_web::Error>(Bytes::from(server_event)) });

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

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    let (tx, _rx) = broadcast::channel(100);
    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");

    let _handle = tokio::spawn(async move {
        let mut conn = client.get_async_connection().await.expect("Failed to get async connection");
        let mut pubsub = conn.into_pubsub();

        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");

        while let Some(msg) = pubsub.on_message().next().await {
            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");
        }
    });

    HttpServer::new(move || {
        let rx = tx.subscribe();
        App::new()
            .app_data(web::Data::new(rx))
            .app_data(web::Data::new(client.clone()))
            .route("/presence/{token}", web::get().to(sse_handler))
    })
    .bind("127.0.0.1:8080")?
    .run()
    .await
}