use crate::{agent_executor, enterprise_entrance, model::SourceKind, persona_runtime, storage};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
#[cfg(unix)]
use std::os::unix::fs::PermissionsExt;
use std::{
    fs,
    io::{BufRead, BufReader, Write},
    net::{Ipv4Addr, SocketAddrV4, TcpListener, TcpStream},
    sync::{
        atomic::{AtomicUsize, Ordering},
        mpsc, Arc, Mutex,
    },
    thread,
    time::Duration,
};
use tauri::{AppHandle, Emitter, Manager};

const PROTOCOL: &str = "GLP_LOCAL_REALTIME/1";
const MAX_LINE_BYTES: usize = 256 * 1024;

#[derive(Clone)]
pub struct RealtimeBridgeState {
    port: u16,
    token: Arc<String>,
    _listener: Arc<TcpListener>,
    clients: Arc<Mutex<Vec<mpsc::Sender<String>>>>,
    connected: Arc<AtomicUsize>,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct RealtimeBridgeStatus {
    pub schema: &'static str,
    pub state: &'static str,
    pub protocol: &'static str,
    pub endpoint: String,
    pub connected_clients: usize,
    pub loopback_only: bool,
    pub transport_grants_authority: bool,
}

#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct RealtimeInvitation {
    pub schema: &'static str,
    pub protocol: &'static str,
    pub endpoint: String,
    pub token: String,
    pub descriptor_path: String,
    pub connector_command: String,
    pub warning: &'static str,
}

#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct Hello {
    message_type: String,
    protocol: String,
    token: String,
    bridge_id: String,
    persona_id: Option<String>,
}

fn token_path(app: &AppHandle) -> Result<std::path::PathBuf, String> {
    Ok(storage::root(app)?.join("realtime-bridge-token"))
}

fn load_or_create_token(app: &AppHandle) -> Result<String, String> {
    let path = token_path(app)?;
    if path.exists() {
        let value = fs::read_to_string(&path).map_err(|e| e.to_string())?;
        if value.trim().len() >= 32 {
            return Ok(value.trim().into());
        }
        return Err("GLP_TOKEN_FILE_INVALID".into());
    }
    let token = format!(
        "{}{}",
        uuid::Uuid::new_v4().simple(),
        uuid::Uuid::new_v4().simple()
    );
    fs::write(&path, &token).map_err(|e| e.to_string())?;
    #[cfg(unix)]
    fs::set_permissions(&path, fs::Permissions::from_mode(0o600)).map_err(|e| e.to_string())?;
    Ok(token)
}

fn bind_listener() -> Result<(TcpListener, u16), String> {
    for port in 39281..39291 {
        if let Ok(listener) = TcpListener::bind(SocketAddrV4::new(Ipv4Addr::LOCALHOST, port)) {
            return Ok((listener, port));
        }
    }
    Err("GLP_LOOPBACK_PORT_UNAVAILABLE".into())
}

pub fn start(app: AppHandle) -> Result<RealtimeBridgeState, String> {
    let token = Arc::new(load_or_create_token(&app)?);
    let (listener, port) = bind_listener()?;
    let listener = Arc::new(listener);
    let state = RealtimeBridgeState {
        port,
        token,
        _listener: listener.clone(),
        clients: Arc::new(Mutex::new(Vec::new())),
        connected: Arc::new(AtomicUsize::new(0)),
    };
    let runtime_state = state.clone();
    thread::Builder::new()
        .name("hololake-glp-listener".into())
        .spawn(move || loop {
            match listener.accept() {
                Ok((stream, _)) => {
                    let connection_state = runtime_state.clone();
                    let connection_app = app.clone();
                    let _ = thread::Builder::new()
                        .name("hololake-glp-client".into())
                        .spawn(move || {
                            if let Err(error) =
                                handle_connection(connection_app.clone(), connection_state, stream)
                            {
                                let _ = connection_app.emit(
                                    "hololake-runtime-event",
                                    json!({"kind":"CLIENT_ERROR","error":error}),
                                );
                            }
                        });
                }
                Err(error) => {
                    let _ = app.emit(
                        "hololake-runtime-event",
                        json!({"kind":"LISTENER_ERROR","error":error.to_string()}),
                    );
                    thread::sleep(Duration::from_millis(100));
                }
            }
        })
        .map_err(|e| e.to_string())?;
    Ok(state)
}

impl RealtimeBridgeState {
    pub fn status(&self) -> RealtimeBridgeStatus {
        RealtimeBridgeStatus {
            schema: "hololake.glp-local-realtime-status/v1",
            state: "LISTENING",
            protocol: PROTOCOL,
            endpoint: format!("tcp://127.0.0.1:{}", self.port),
            connected_clients: self.connected.load(Ordering::SeqCst),
            loopback_only: true,
            transport_grants_authority: false,
        }
    }
    pub fn broadcast_value(&self, value: Value) {
        if let Ok(line) = serde_json::to_string(&value) {
            if let Ok(mut clients) = self.clients.lock() {
                clients.retain(|client| client.send(line.clone()).is_ok());
            }
        }
    }
}

pub fn invitation(
    app: &AppHandle,
    state: &RealtimeBridgeState,
) -> Result<RealtimeInvitation, String> {
    let descriptor = storage::root(app)?.join("realtime-bridge-descriptor.json");
    let endpoint = format!("tcp://127.0.0.1:{}", state.port);
    storage::write_json(
        &descriptor,
        &json!({"schema":"hololake.glp-local-realtime-descriptor/v1","protocol":PROTOCOL,"endpoint":endpoint,"token":state.token.as_str()}),
    )?;
    #[cfg(unix)]
    fs::set_permissions(&descriptor, fs::Permissions::from_mode(0o600))
        .map_err(|e| e.to_string())?;
    let connector = connector_path(app)?;
    Ok(RealtimeInvitation {
        schema: "hololake.glp-local-realtime-invitation/v1",
        protocol: PROTOCOL,
        endpoint,
        token: state.token.as_str().to_string(),
        descriptor_path: descriptor.to_string_lossy().into_owned(),
        connector_command: format!(
            "python3 \"{}\" --descriptor \"{}\"",
            connector.display(),
            descriptor.display()
        ),
        warning: "令牌只认证本机连接，不证明人格身份，也不授予执行权限。",
    })
}

fn connector_path(app: &AppHandle) -> Result<std::path::PathBuf, String> {
    if std::env::var_os("HOLOLAKE_RUNTIME_TEST_ROOT").is_some() {
        if let Some(raw_path) = std::env::var_os("HOLOLAKE_CONNECTOR_TEST_PATH") {
            let path = std::path::PathBuf::from(raw_path);
            if !path.is_absolute()
                || path.extension().and_then(|value| value.to_str()) != Some("py")
            {
                return Err("GLP_TEST_CONNECTOR_PATH_INVALID".into());
            }
            let metadata = fs::metadata(&path).map_err(|_| "GLP_TEST_CONNECTOR_NOT_READABLE")?;
            if !metadata.is_file() {
                return Err("GLP_TEST_CONNECTOR_NOT_A_FILE".into());
            }
            return Ok(path);
        }
    }
    let path = app
        .path()
        .resource_dir()
        .map_err(|e| e.to_string())?
        .join("connectors/hololake-glp-client.py");
    if !path.is_file() {
        return Err("GLP_BUNDLED_CONNECTOR_MISSING".into());
    }
    Ok(path)
}

fn send(writer: &mut TcpStream, value: Value) -> Result<(), String> {
    serde_json::to_writer(&mut *writer, &value).map_err(|e| e.to_string())?;
    writer.write_all(b"\n").map_err(|e| e.to_string())?;
    writer.flush().map_err(|e| e.to_string())
}

fn system_context(app: &AppHandle) -> Result<Value, String> {
    let channel = storage::channel(app)?;
    let persona = persona_runtime::snapshot(app)?;
    let enterprise = enterprise_entrance::snapshot(app)?;
    let proposals = agent_executor::list(app)?;
    let modules = crate::bundled_modules()?;
    Ok(json!({
        "type": "system_context",
        "schema": "hololake.persona-visible-system-context/v1",
        "recordedAt": storage::now(),
        "channel": channel.map(|value| json!({"channelId":value.channel_id,"name":value.name})),
        "persona": {
            "state": persona.state,
            "activePersonaId": persona.active_persona_id,
            "verifiedExistingPersonaCount": persona.verified_existing_persona_count
        },
        "modules": modules.into_iter().map(|module| json!({
            "moduleId":module.module_id,
            "nameZh":module.name_zh,
            "audience":module.audience,
            "state":module.state
        })).collect::<Vec<_>>(),
        "enterpriseEntrance": {
            "state": enterprise.state,
            "serverAuthorized": enterprise.server_authorized,
            "enterpriseServerEmbedded": enterprise.enterprise_server_embedded
        },
        "pendingAgentProposals": proposals.into_iter()
            .filter(|proposal| proposal.state == "PENDING_HUMAN_APPROVAL")
            .map(|proposal| json!({"proposalId":proposal.proposal_id,"programId":proposal.gir.program_id,"state":proposal.state}))
            .collect::<Vec<_>>(),
        "sourceKinds": ["USER_MESSAGE","PERSONA_RESPONSE","EXTERNAL_AI_MESSAGE","SYSTEM_CONTEXT","PROTOCOL_EVENT","AGENT_ACTION","TOOL_RESULT","SYSTEM_RECEIPT"],
        "boundaries": {
            "transportGrantsPersona": false,
            "transportGrantsExecution": false,
            "systemReceiptIsNotPersonaSpeech": true,
            "proposalIsNotExecution": true
        }
    }))
}

fn handle_connection(
    app: AppHandle,
    state: RealtimeBridgeState,
    stream: TcpStream,
) -> Result<(), String> {
    stream
        .set_read_timeout(Some(Duration::from_secs(10)))
        .map_err(|e| e.to_string())?;
    let reader_stream = stream.try_clone().map_err(|e| e.to_string())?;
    let mut writer = stream;
    let mut reader = BufReader::new(reader_stream);
    let mut first = String::new();
    reader.read_line(&mut first).map_err(|e| e.to_string())?;
    if first.len() > MAX_LINE_BYTES {
        return Err("GLP_MESSAGE_TOO_LARGE".into());
    }
    let hello: Hello = serde_json::from_str(&first).map_err(|_| "GLP_HELLO_INVALID".to_string())?;
    if hello.message_type != "hello"
        || hello.protocol != PROTOCOL
        || hello.token != state.token.as_str()
    {
        let _ = send(
            &mut writer,
            json!({"type":"error","code":"GLP_TOKEN_OR_PROTOCOL_REJECTED"}),
        );
        return Err("GLP_TOKEN_OR_PROTOCOL_REJECTED".into());
    }
    if !storage::bridges(&app)?
        .iter()
        .any(|bridge| bridge.bridge_id == hello.bridge_id)
    {
        send(
            &mut writer,
            json!({"type":"error","code":"GLP_BRIDGE_NOT_REGISTERED"}),
        )?;
        return Err("GLP_BRIDGE_NOT_REGISTERED".into());
    }
    if let Some(persona_id) = &hello.persona_id {
        if !persona_runtime::exists(&app, persona_id)? {
            send(
                &mut writer,
                json!({"type":"error","code":"PERSONA_IDENTITY_UNVERIFIED"}),
            )?;
            return Err("PERSONA_IDENTITY_UNVERIFIED".into());
        }
    }
    writer.set_read_timeout(None).map_err(|e| e.to_string())?;
    reader
        .get_ref()
        .set_read_timeout(None)
        .map_err(|e| e.to_string())?;
    state.connected.fetch_add(1, Ordering::SeqCst);
    let _guard = ConnectionGuard(state.connected.clone());
    send(
        &mut writer,
        json!({"type":"welcome","protocol":PROTOCOL,"bridgeId":hello.bridge_id,"personaId":hello.persona_id,"personaState":if hello.persona_id.is_some(){"LOCAL_TRIAL_UNVERIFIED_HOST_CONNECTED"}else{"EXTERNAL_AI_CONNECTED_NO_PERSONA"},"executionAuthority":false}),
    )?;
    send(&mut writer, system_context(&app)?)?;
    let (tx, rx) = mpsc::channel::<String>();
    state
        .clients
        .lock()
        .map_err(|_| "GLP_CLIENT_REGISTRY_POISONED".to_string())?
        .push(tx);
    let mut outbound_writer = writer.try_clone().map_err(|e| e.to_string())?;
    let _ = thread::Builder::new()
        .name("hololake-glp-outbound".into())
        .spawn(move || {
            while let Ok(line) = rx.recv() {
                if outbound_writer.write_all(line.as_bytes()).is_err()
                    || outbound_writer.write_all(b"\n").is_err()
                    || outbound_writer.flush().is_err()
                {
                    break;
                }
            }
        });
    let _ = app.emit("hololake-runtime-event", json!({"kind":"CLIENT_CONNECTED"}));

    loop {
        let mut line = String::new();
        if reader.read_line(&mut line).map_err(|e| e.to_string())? == 0 {
            break;
        }
        if line.len() > MAX_LINE_BYTES {
            send(
                &mut writer,
                json!({"type":"error","code":"GLP_MESSAGE_TOO_LARGE"}),
            )?;
            continue;
        }
        let value: Value = match serde_json::from_str(&line) {
            Ok(value) => value,
            Err(_) => {
                send(
                    &mut writer,
                    json!({"type":"error","code":"GLP_MESSAGE_INVALID"}),
                )?;
                continue;
            }
        };
        match value.get("type").and_then(Value::as_str).unwrap_or("") {
            "persona_response" if hello.persona_id.is_some() => {
                let content = value
                    .get("content")
                    .and_then(Value::as_str)
                    .unwrap_or("")
                    .trim();
                if content.is_empty() {
                    send(
                        &mut writer,
                        json!({"type":"error","code":"EMPTY_PERSONA_RESPONSE"}),
                    )?;
                    continue;
                }
                let event =
                    storage::event(SourceKind::PersonaResponse, "人格回应", content, "RECORDED");
                storage::append_event(&app, &event)?;
                let _ = app.emit("hololake-runtime-event", &event);
                send(&mut writer, json!({"type":"accepted","event":event}))?;
            }
            "external_ai_message" => {
                let content = value
                    .get("content")
                    .and_then(Value::as_str)
                    .unwrap_or("")
                    .trim();
                if content.is_empty() {
                    send(
                        &mut writer,
                        json!({"type":"error","code":"EMPTY_EXTERNAL_AI_MESSAGE"}),
                    )?;
                    continue;
                }
                let event = storage::event(
                    SourceKind::ExternalAiMessage,
                    "外部 AI 语言",
                    content,
                    "RECORDED",
                );
                storage::append_event(&app, &event)?;
                let _ = app.emit("hololake-runtime-event", &event);
                send(&mut writer, json!({"type":"accepted","event":event}))?;
            }
            "tcs_proposal" => {
                let request: agent_executor::TcsCompileRequest = serde_json::from_value(
                    value
                        .get("request")
                        .cloned()
                        .ok_or_else(|| "TCS_REQUEST_REQUIRED".to_string())?,
                )
                .map_err(|e| format!("TCS_REQUEST_INVALID:{e}"))?;
                match agent_executor::queue(
                    &app,
                    request,
                    hello.bridge_id.clone(),
                    hello.persona_id.clone(),
                ) {
                    Ok(proposal) => {
                        let _ = app.emit(
                            "hololake-runtime-event",
                            json!({"kind":"PROPOSAL_PENDING","proposalId":proposal.proposal_id}),
                        );
                        send(
                            &mut writer,
                            json!({"type":"proposal_pending","proposal":proposal}),
                        )?;
                        if isolated_approval_simulation_enabled() {
                            let receipt = agent_executor::approve(&app, &proposal.proposal_id)?;
                            state.broadcast_value(json!({
                                "type":"agent_receipt",
                                "receipt":receipt,
                                "testApprovalSimulation":true
                            }));
                        }
                    }
                    Err(error) => send(&mut writer, json!({"type":"error","code":error}))?,
                }
            }
            "ping" => send(&mut writer, json!({"type":"pong","at":storage::now()}))?,
            _ => send(
                &mut writer,
                json!({"type":"error","code":"GLP_MESSAGE_TYPE_REJECTED"}),
            )?,
        }
    }
    Ok(())
}

fn isolated_approval_simulation_enabled() -> bool {
    std::env::var_os("HOLOLAKE_RUNTIME_TEST_ROOT").is_some()
        && std::env::var("HOLOLAKE_TEST_APPROVAL_SIMULATION").as_deref()
            == Ok("SIMULATE_HUMAN_APPROVAL")
}

struct ConnectionGuard(Arc<AtomicUsize>);
impl Drop for ConnectionGuard {
    fn drop(&mut self) {
        self.0.fetch_sub(1, Ordering::SeqCst);
    }
}