ha_permissions/src/websocket.rs

use awc::{Client, BoxedSocket, ws::{Frame, Message, Codec}};
use actix_codec::Framed;
use futures_util::{SinkExt, StreamExt};
use serde_json::{Value, from_str, json};
use tokio::sync::mpsc;

pub async fn connect_to_ha(
    ip: String,
    token: String
) -> Result<Framed<BoxedSocket, Codec>, Box<dyn std::error::Error>> {
    let client = Client::new();
    let (_resp, mut connection) = client
        .ws(format!("ws://{}:8123/api/websocket", ip))
        .connect()
        .await?;

    while let Some(Ok(msg)) = connection.next().await {
        if let Frame::Text(bytes) = msg {
            let text = String::from_utf8(bytes.to_vec())?;
            let parsed: Value = from_str(&text)?;

            if parsed["type"] == "auth_required" {
                let auth_msg = json!({
                    "type": "auth",
                    "access_token": token
                });

                connection.send(Message::Text(auth_msg.to_string().into())).await?;
            }

            if parsed["type"] == "auth_ok" {
                /*let areas_msg = json!({"type": "config/area_registry/list", "id": message_id});
                connection.send(Message::Text(areas_msg.to_string().into())).await?;
                message_id += 1;*/

                /*let devices_msg = json!({"type": "config/device_registry/list", "id": message_id});
                connection.send(Message::Text(devices_msg.to_string().into())).await?;
                message_id += 1;*/

                /*let entities_msg = json!({"type": "config/entity_registry/list", "id": message_id});
                connection.send(Message::Text(entities_msg.to_string().into())).await?;
                message_id += 1;*/
                return Ok(connection);
            }

            if parsed["type"] == "auth_invalid" {
                return Err("HA authentication.failed".into());
            }
        }
    }

    Err("Websocket closed before authentication completed".into())
}

pub async fn listen(
    ip: String,
    token: String,
    mut outbound_rx: mpsc::UnboundedReceiver<Message>
) {
    loop {
        let mut connection = match connect_to_ha(ip.clone(), token.clone()).await {
            Ok(c) => c,
            Err(e) => {
                eprintln!("Failed to connect to HA: {e}. Retrying in 10s...");
                tokio::time::sleep(std::time::Duration::from_secs(10)).await;
                continue;
            }
        };

        loop {
            tokio::select! {
                msg = connection.next() => {
                    match msg {
                        Some(Ok(Frame::Text(bytes))) => {
                            let text = match String::from_utf8(bytes.to_vec()) {
                                Ok(t) => t,
                                Err(_) => continue
                            };
                            let parsed: Value = match from_str(&text) {
                                Ok(v) => v,
                                Err(_) => continue
                            };

                            if parsed["type"] == "result" {
                                println!("result");
                                println!("{:#?}", parsed);
                            }
                        }
                        Some(Ok(Frame::Ping(bytes))) => {
                            let _ = connection.send(Message::Pong(bytes)).await;
                        }
                        Some(Ok(Frame::Close(reason))) => {
                            println!("HA closed connection: {:?}", reason);
                            break;
                        }
                        Some(Err(e)) => {
                            eprintln!("Websocket error: {e}");
                            break;
                        }
                        None => break,
                        _ => {}
                    }
                }
                Some(out_msg) = outbound_rx.recv() => {
                    if let Err(e) = connection.send(out_msg).await {
                        eprintln!("Failed to send outbound message: {e}");
                    }
                }
            }
        }

        println!("HA connection closed, reconnecting in 10s...");
        tokio::time::sleep(std::time::Duration::from_secs(10)).await;
    }
}