//! End-to-end regression test for the SSH backpressure deadlock.
//!
//! This is the test that was missing when the push hang shipped. The bug could
//! not be reached from a unit test: it lives in russh's session loop, and
//! `ChannelId` and `Channel<Msg>` cannot be constructed outside that crate. So
//! this drives the real thing — a real `git push`, over the real `ssh` client,
//! against [`MnwServer`] and [`MnwHandler`], with only the MNW API stubbed.
//!
//! WHAT IT PINS. russh hands every inbound `CHANNEL_DATA` first to the
//! `Channel`'s bounded queue (`channel_buffer_size`, 100 messages) with a
//! blocking `send().await`, and only then to `Handler::data`. A `Channel` the
//! handler parks and never reads therefore wedges the whole session loop once
//! 100 packets have arrived, and the client hangs forever mid-pack. The push
//! below is deliberately larger than that buffer: incompressible random bytes,
//! so the pack cannot shrink under it. Against the pre-fix handler this test
//! hangs until its timeout; against the fixed one it finishes in about a
//! second.
//!
//! Requires `git` and `ssh` on PATH.

use std::path::Path;
use std::process::Stdio;
use std::sync::Arc;
use std::time::Duration;

use russh::server::Server as _;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;
use tokio::process::Command;

use crate::api::MnwApiClient;
use crate::rate_limit::AuthRateLimiter;
use crate::ssh::MnwServer;

/// Comfortably past russh's 100-message channel buffer. 6 MiB of random bytes
/// packs to roughly 6 MiB, which is ~190 packets at the 32 KiB maximum packet
/// size. `shop`, the repo that first hit this, was 4.5 MiB.
const PAYLOAD_BYTES: usize = 6 * 1024 * 1024;

/// Generous enough that a slow machine does not produce a false red, short
/// enough that a real regression fails the suite rather than hanging CI. The
/// fixed path takes about a second.
const PUSH_TIMEOUT: Duration = Duration::from_secs(90);

async fn run(cmd: &str, args: &[&str], cwd: &Path) -> anyhow::Result<()> {
    let out = Command::new(cmd)
        .args(args)
        .current_dir(cwd)
        .output()
        .await?;
    anyhow::ensure!(
        out.status.success(),
        "{cmd} {args:?} failed: {}",
        String::from_utf8_lossy(&out.stderr)
    );
    Ok(())
}

/// Minimal stand-in for the MNW server's two internal endpoints. Accepts every
/// key and authorizes every operation: this test is about transport
/// backpressure, and the authorization paths have their own coverage.
async fn spawn_stub_api(repo_path: String) -> anyhow::Result<String> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let addr = listener.local_addr()?;

    tokio::spawn(async move {
        loop {
            let Ok((mut sock, _)) = listener.accept().await else {
                break;
            };
            let repo_path = repo_path.clone();
            tokio::spawn(async move {
                // Read up to the end of the headers; that carries the request
                // line, which is all this stub dispatches on. Any body after it
                // is ignored, and the connection is closed per response, so
                // there is nothing to keep parsing.
                let mut buf = vec![0u8; 8192];
                let Ok(n) = sock.read(&mut buf).await else {
                    return;
                };
                let req = String::from_utf8_lossy(&buf[..n]);

                let body = if req.contains("/api/internal/git/authorize") {
                    format!(r#"{{"repo_path":"{repo_path}"}}"#)
                } else {
                    r#"{"user_id":"test-user","username":"max","display_name":null,
                        "creator_tier":null,"can_create_projects":true,"suspended":false,
                        "actor_token":"stub-actor-token"}"#
                        .replace(['\n', ' '], "")
                };

                let resp = format!(
                    "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
                    body.len()
                );
                let _ = sock.write_all(resp.as_bytes()).await;
                let _ = sock.flush().await;
            });
        }
    });

    Ok(format!("http://{addr}"))
}

/// Start the real SSH server on an ephemeral port. Returns the port.
///
/// `git_user` is empty on purpose: it puts `spawn_git_process` on its direct
/// path, so the test needs no sudo grant. Everything else is production code.
async fn spawn_ssh_server(api_url: String) -> anyhow::Result<u16> {
    let listener = TcpListener::bind("127.0.0.1:0").await?;
    let port = listener.local_addr()?.port();

    let host_key =
        russh::keys::PrivateKey::random(&mut rand::rng(), russh::keys::Algorithm::Ed25519)?;

    let mut methods = russh::MethodSet::empty();
    methods.push(russh::MethodKind::PublicKey);

    let config = Arc::new(russh::server::Config {
        methods,
        keys: vec![host_key],
        auth_rejection_time: Duration::from_millis(0),
        auth_rejection_time_initial: Some(Duration::from_millis(0)),
        ..Default::default()
    });

    let api = MnwApiClient::new(api_url, "stub-service-token".to_string());
    let staging = Arc::new(std::env::temp_dir().join("mnw-cli-backpressure-staging"));
    let mut server = MnwServer::new(
        api,
        staging,
        String::new(),
        Arc::new(AuthRateLimiter::new()),
    );

    tokio::spawn(async move {
        let _ = server.run_on_socket(config, &listener).await;
    });

    Ok(port)
}

/// A push larger than russh's channel buffer completes instead of hanging.
///
/// The whole point is the size: shrink the payload under ~100 packets and this
/// test passes against the deadlocked handler too, which is exactly why the bug
/// survived `makeover-touch` pushing fine the day before `shop` stalled.
#[tokio::test]
async fn a_push_larger_than_the_channel_buffer_completes() {
    let tmp = tempfile::tempdir().unwrap();
    let root = tmp.path();

    // Client identity. The stub API accepts whatever fingerprint it is asked
    // about, so the key only has to exist and be offered.
    let key_path = root.join("id_ed25519");
    run(
        "ssh-keygen",
        &[
            "-t",
            "ed25519",
            "-N",
            "",
            "-C",
            "backpressure-test",
            "-f",
            key_path.to_str().unwrap(),
        ],
        root,
    )
    .await
    .expect("ssh-keygen; is openssh installed?");

    // The bare repo the push lands in, standing in for what the server would
    // have auto-created.
    let bare = root.join("max").join("backpressure.git");
    std::fs::create_dir_all(bare.parent().unwrap()).unwrap();
    run(
        "git",
        &["init", "--bare", "-b", "main", bare.to_str().unwrap()],
        root,
    )
    .await
    .unwrap();

    // Source repo carrying incompressible data, so the pack stays big.
    let src = root.join("src");
    std::fs::create_dir_all(&src).unwrap();
    run("git", &["init", "-b", "main"], &src).await.unwrap();
    run("git", &["config", "user.email", "t@example.com"], &src)
        .await
        .unwrap();
    run("git", &["config", "user.name", "test"], &src)
        .await
        .unwrap();

    let mut blob = vec![0u8; PAYLOAD_BYTES];
    rand::fill(&mut blob[..]);
    std::fs::write(src.join("payload.bin"), &blob).unwrap();
    run("git", &["add", "."], &src).await.unwrap();
    run("git", &["commit", "-m", "payload"], &src)
        .await
        .unwrap();

    let api_url = spawn_stub_api(bare.to_str().unwrap().to_string())
        .await
        .unwrap();
    let port = spawn_ssh_server(api_url).await.unwrap();

    let ssh_command = format!(
        "ssh -p {port} -i {} -o IdentitiesOnly=yes -o StrictHostKeyChecking=no \
         -o UserKnownHostsFile=/dev/null -o LogLevel=ERROR",
        key_path.display()
    );

    let push = Command::new("git")
        .args([
            "push",
            &format!("ssh://git@127.0.0.1:{port}/max/backpressure.git"),
            "main",
        ])
        .current_dir(&src)
        .env("GIT_SSH_COMMAND", &ssh_command)
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .output();

    let out = tokio::time::timeout(PUSH_TIMEOUT, push)
        .await
        .expect(
            "push hung: the session loop stopped reading mid-pack, which is the \
             deadlock this test exists to catch",
        )
        .unwrap();

    assert!(
        out.status.success(),
        "push failed: {}",
        String::from_utf8_lossy(&out.stderr)
    );

    // The bytes actually landed, not merely a session that failed politely.
    let refs = Command::new("git")
        .args(["--git-dir", bare.to_str().unwrap(), "rev-parse", "main"])
        .output()
        .await
        .unwrap();
    assert!(
        refs.status.success(),
        "pushed ref missing from the bare repo: {}",
        String::from_utf8_lossy(&refs.stderr)
    );
}