use citygen_server::{
    gateway::{router, App, Config},
    protocol::{self as p, client_message, server_message},
};
use futures_util::{SinkExt, StreamExt};
use prost::Message;
use std::{io::Write, time::Duration};

fn fixture(dir: &std::path::Path) {
    let folder = dir.join("cities/demo/tiles");
    std::fs::create_dir_all(&folder).unwrap();
    std::fs::write(dir.join("cities/demo/manifest.json"),r#"{"formatVersion":1,"city":{"id":"demo","name":"Demo","bounds":{"minX":0,"minY":0,"maxX":100,"maxY":100}},"tiles":[{"id":"0_0","url":"tiles/0_0.json.gz","bounds":{"minX":0,"minY":0,"maxX":256,"maxY":256}}]}"#).unwrap();
    let file = std::fs::File::create(folder.join("0_0.json.gz")).unwrap();
    let mut encoder = flate2::write::GzEncoder::new(file, flate2::Compression::default());
    encoder.write_all(br#"{"formatVersion":1,"cityId":"demo","id":"0_0","buildings":[],"roads":[{"id":"road","kind":"residential","pts":[[0,0],[100,0],[100,100]]}]}"#).unwrap();
    encoder.finish().unwrap();
}
async fn read(
    socket: &mut tokio_tungstenite::WebSocketStream<
        tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
    >,
) -> p::ServerMessage {
    loop {
        let frame = tokio::time::timeout(Duration::from_secs(5), socket.next())
            .await
            .unwrap()
            .unwrap()
            .unwrap();
        if let tokio_tungstenite::tungstenite::Message::Binary(bytes) = frame {
            return p::ServerMessage::decode(&bytes[..]).unwrap();
        }
    }
}
async fn http_request(
    address: std::net::SocketAddr,
    path: &str,
    method: &str,
    token: &str,
) -> String {
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    let mut socket = tokio::net::TcpStream::connect(address).await.unwrap();
    socket.write_all(format!("{method} {path} HTTP/1.1\r\nHost: localhost\r\nAuthorization: Bearer {token}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n").as_bytes()).await.unwrap();
    let mut response = String::new();
    socket.read_to_string(&mut response).await.unwrap();
    response
}
#[tokio::test]
async fn thirty_observers_snapshot_delta_vote_and_restart() {
    let dir = tempfile::tempdir().unwrap();
    fixture(dir.path());
    let app = App::new(Config {
        cities: dir.path().join("cities"),
        saves: dir.path().join("rooms"),
        admin_token: Some("test-only-administrator-token-32-chars".into()),
        ..Default::default()
    })
    .unwrap();
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let address = listener.local_addr().unwrap();
    let serve_app = app.clone();
    let server = tokio::spawn(async move {
        axum::serve(
            listener,
            router(serve_app).into_make_service_with_connect_info::<std::net::SocketAddr>(),
        )
        .await
        .unwrap();
    });
    // Minimal HTTP request avoids adding a second HTTP client dependency.
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    let mut http = tokio::net::TcpStream::connect(address).await.unwrap();
    let body = r#"{"cityId":"demo","seed":1}"#;
    http.write_all(format!("POST /api/rooms HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",body.len()).as_bytes()).await.unwrap();
    let mut response = String::new();
    http.read_to_string(&mut response).await.unwrap();
    assert!(response.starts_with("HTTP/1.1 200"), "{response}");
    let value: serde_json::Value =
        serde_json::from_str(response.split("\r\n\r\n").nth(1).unwrap()).unwrap();
    let id = value["id"].as_str().unwrap().to_string();
    assert!(http_request(address, "/api/admin", "GET", "wrong")
        .await
        .starts_with("HTTP/1.1 401"));
    assert!(http_request(
        address,
        "/api/admin",
        "GET",
        "test-only-administrator-token-32-chars"
    )
    .await
    .starts_with("HTTP/1.1 200"));
    let checkpoint = http_request(
        address,
        &format!("/api/admin/rooms/{id}/checkpoint"),
        "POST",
        "test-only-administrator-token-32-chars",
    )
    .await;
    assert!(checkpoint.starts_with("HTTP/1.1 200"), "{checkpoint}");
    assert!(citygen_server::storage::load(&dir.path().join("rooms"), &id).is_ok());
    let mut sockets = vec![];
    for i in 0..30 {
        let (mut socket, _) =
            tokio_tungstenite::connect_async(format!("ws://{address}/api/rooms/{id}/ws"))
                .await
                .unwrap();
        let join = p::ClientMessage {
            body: Some(client_message::Body::Join(p::Join {
                room: id.clone(),
                client_version: 1,
                nickname: format!("Player{i}"),
            })),
        };
        socket
            .send(tokio_tungstenite::tungstenite::Message::Binary(
                join.encode_to_vec().into(),
            ))
            .await
            .unwrap();
        assert!(matches!(
            read(&mut socket).await.body,
            Some(server_message::Body::Welcome(_))
        ));
        sockets.push(socket);
    }
    let s = &mut sockets[0];
    let subscribe = p::ClientMessage {
        body: Some(client_message::Body::Subscribe(p::Subscribe {
            rect: Some(p::Rect {
                min_x: -100,
                min_y: -100,
                max_x: 20000,
                max_y: 20000,
            }),
            lod: 0,
        })),
    };
    s.send(tokio_tungstenite::tungstenite::Message::Binary(
        subscribe.encode_to_vec().into(),
    ))
    .await
    .unwrap();
    loop {
        if let Some(server_message::Body::Snapshot(snapshot)) = read(s).await.body {
            assert_eq!(snapshot.entities.len(), 1000);
            break;
        }
    }
    loop {
        if matches!(read(s).await.body, Some(server_message::Body::Delta(_))) {
            break;
        }
    }
    // 31st observer must be rejected without affecting existing players.
    let command = p::ClientMessage {
        body: Some(client_message::Body::Intervene(p::Intervene {
            action: "meal".into(),
        })),
    }
    .encode_to_vec();
    s.send(tokio_tungstenite::tungstenite::Message::Binary(
        command.clone().into(),
    ))
    .await
    .unwrap();
    s.send(tokio_tungstenite::tungstenite::Message::Binary(
        command.into(),
    ))
    .await
    .unwrap();
    loop {
        if let Some(server_message::Body::Error(error)) = read(s).await.body {
            assert_eq!(error.code, "COMMAND_LIMIT");
            break;
        }
    }
    loop {
        if let Some(server_message::Body::Gameplay(state)) = read(s).await.body {
            assert!(
                state.pending.iter().any(|p| p == "meal")
                    || state.events.iter().any(|e| e.kind == "meal")
            );
            break;
        }
    }
    let (mut extra, _) =
        tokio_tungstenite::connect_async(format!("ws://{address}/api/rooms/{id}/ws"))
            .await
            .unwrap();
    extra
        .send(tokio_tungstenite::tungstenite::Message::Binary(
            p::ClientMessage {
                body: Some(client_message::Body::Join(p::Join {
                    room: id.clone(),
                    client_version: 1,
                    nickname: "extra".into(),
                })),
            }
            .encode_to_vec()
            .into(),
        ))
        .await
        .unwrap();
    assert!(
        matches!(read(&mut extra).await.body,Some(server_message::Body::Error(e)) if e.code=="ROOM_FULL")
    );
    let proposal = p::ClientMessage {
        body: Some(client_message::Body::Propose(p::Propose { speed: 0 })),
    };
    sockets[0]
        .send(tokio_tungstenite::tungstenite::Message::Binary(
            proposal.encode_to_vec().into(),
        ))
        .await
        .unwrap();
    let yes = p::ClientMessage {
        body: Some(client_message::Body::Vote(p::Vote { yes: true })),
    }
    .encode_to_vec();
    for socket in sockets.iter_mut().take(15).skip(1) {
        socket
            .send(tokio_tungstenite::tungstenite::Message::Binary(
                yes.clone().into(),
            ))
            .await
            .unwrap();
    }
    loop {
        if let Some(server_message::Body::Clock(clock)) = read(&mut sockets[0]).await.body {
            assert_eq!(clock.speed, 1, "15 of 30 is not a majority");
            break;
        }
    }
    sockets[15]
        .send(tokio_tungstenite::tungstenite::Message::Binary(yes.into()))
        .await
        .unwrap();
    loop {
        if let Some(server_message::Body::Clock(clock)) = read(&mut sockets[0]).await.body {
            if clock.speed == 0 {
                break;
            }
        }
    }
    for socket in &mut sockets {
        let _ = socket.close(None).await;
    }
    app.stop();
    server.abort();
    let saved = citygen_server::storage::load(&dir.path().join("rooms"), &id).unwrap();
    assert_eq!(saved.agents.len(), 1000);
    assert_eq!(saved.speed, 0);
    assert!(!saved.city_revision.is_empty());
    let mut expected = citygen_server::simulation::Simulation::restore(saved.clone()).unwrap();
    let expected_entities = expected
        .visible(
            &p::Rect {
                min_x: -100,
                min_y: -100,
                max_x: 20000,
                max_y: 20000,
            },
            1000,
        )
        .unwrap();
    let restarted = App::new(Config {
        cities: dir.path().join("cities"),
        saves: dir.path().join("rooms"),
        ..Default::default()
    })
    .unwrap();
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let address = listener.local_addr().unwrap();
    let serve_app = restarted.clone();
    let server = tokio::spawn(async move {
        axum::serve(
            listener,
            router(serve_app).into_make_service_with_connect_info::<std::net::SocketAddr>(),
        )
        .await
        .unwrap();
    });
    let (mut socket, _) =
        tokio_tungstenite::connect_async(format!("ws://{address}/api/rooms/{id}/ws"))
            .await
            .unwrap();
    socket
        .send(tokio_tungstenite::tungstenite::Message::Binary(
            p::ClientMessage {
                body: Some(client_message::Body::Join(p::Join {
                    room: id.clone(),
                    client_version: 1,
                    nickname: "Restored".into(),
                })),
            }
            .encode_to_vec()
            .into(),
        ))
        .await
        .unwrap();
    assert!(http_request(
        address,
        "/api/admin",
        "GET",
        "test-only-administrator-token-32-chars"
    )
    .await
    .starts_with("HTTP/1.1 404"));
    assert!(
        matches!(read(&mut socket).await.body,Some(server_message::Body::Welcome(w)) if w.tick>=saved.tick)
    );
    socket
        .send(tokio_tungstenite::tungstenite::Message::Binary(
            subscribe.encode_to_vec().into(),
        ))
        .await
        .unwrap();
    loop {
        if let Some(server_message::Body::Snapshot(snapshot)) = read(&mut socket).await.body {
            let entities: std::collections::HashMap<_, _> =
                snapshot.entities.into_iter().map(|a| (a.id, a)).collect();
            assert_eq!(entities, expected_entities);
            break;
        }
    }
    let _ = socket.close(None).await;
    restarted.stop();
    server.abort();
}