grpctestify 1.6.1

// Call command - execute gRPC call without assertions (like curl)

use anyhow::Result;
use futures::stream::StreamExt;
use serde_json::Value;
use std::fs::File;
use std::io::Write;
use std::path::Path;
use std::time::Instant;
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;

use crate::cli::args::CallArgs;
use crate::execution::runner_helpers;
use crate::grpc::{GrpcClient, GrpcClientConfig, client::StreamItem, proxy::ProxyEnv};
use crate::parser;

struct CallOptions<'a> {
    include_headers: bool,
    verbose: bool,
    very_verbose: bool,
    output_file: &'a mut Option<File>,
    header_file: &'a mut Option<File>,
    silent: bool,
    show_error: bool,
    connect_timeout: u64,
    max_time: u64,
}

pub async fn handle_call(args: &CallArgs) -> Result<()> {
    if args.doc_index == Some(0) {
        anyhow::bail!("--doc-index must be >= 1");
    }

    let cwd = std::env::current_dir()?;
    let file_path = if args.file.is_absolute() {
        args.file.clone()
    } else {
        cwd.join(&args.file)
    };

    if !file_path.exists() {
        return Err(anyhow::anyhow!("File not found: {}", file_path.display()));
    }

    let parse_result = parser::parse_with_recovery(&file_path);
    let doc = parse_result.document;

    let mut output_file: Option<File> = if let Some(ref path) = args.output {
        Some(File::create(path)?)
    } else {
        None
    };

    let mut header_file: Option<File> = if let Some(ref path) = args.dump_header {
        Some(File::create(path)?)
    } else {
        None
    };

    let verbose = args.verbose || args.very_verbose;

    let mut matched_docs = 0usize;
    for (idx, d) in doc.iter_chain().enumerate() {
        let doc_index = idx + 1;
        if let Some(selected) = args.doc_index
            && selected != doc_index
        {
            continue;
        }
        matched_docs += 1;

        let opts = CallOptions {
            include_headers: args.include,
            verbose,
            very_verbose: args.very_verbose,
            output_file: &mut output_file,
            header_file: &mut header_file,
            silent: args.silent,
            show_error: args.show_error,
            connect_timeout: args.connect_timeout,
            max_time: args.max_time,
        };

        handle_call_document(d, &file_path, opts).await?;
    }

    if matched_docs == 0 {
        if let Some(selected) = args.doc_index {
            anyhow::bail!(
                "Document index {} is out of range (total documents: {})",
                selected,
                doc.document_count()
            );
        }
        anyhow::bail!("No documents found in file");
    }

    Ok(())
}

/// Print an info line to stderr (curl `*` prefix).
fn vinfo(silent: bool, line: &str) {
    if !silent {
        eprintln!("* {}", line);
    }
}

/// Print a sent-header line to stderr (curl `>` prefix).
fn vsend(silent: bool, line: &str) {
    if !silent {
        eprintln!("> {}", line);
    }
}

/// Print a received-header line to stderr (curl `<` prefix).
fn vrecv(silent: bool, line: &str) {
    if !silent {
        eprintln!("< {}", line);
    }
}

fn print_send_metadata(silent: bool, entries: &std::collections::HashMap<String, String>) {
    let mut pairs: Vec<_> = entries.iter().collect();
    pairs.sort_by_key(|(k, _)| k.as_str());
    for (k, v) in pairs {
        vsend(silent, &format!("{}: {}", k, v));
    }
}

fn print_recv_metadata(silent: bool, entries: &std::collections::HashMap<String, String>) {
    let mut pairs: Vec<_> = entries.iter().collect();
    pairs.sort_by_key(|(k, _)| k.as_str());
    for (k, v) in pairs {
        vrecv(silent, &format!("{}: {}", k, v));
    }
}

async fn handle_call_document(
    doc: &parser::GctfDocument,
    gctf_file: &Path,
    opts: CallOptions<'_>,
) -> Result<()> {
    let (package, service, method) = match doc.parse_endpoint() {
        Some(e) => e,
        None => return Err(anyhow::anyhow!("Invalid ENDPOINT format")),
    };
    let full_service = runner_helpers::full_service_name(&package, &service);

    let address = runner_helpers::effective_address(doc);
    let tls_config = runner_helpers::build_tls_config(doc, gctf_file);
    let tls_label = if tls_config.is_some() {
        "TLS"
    } else {
        "plaintext"
    };

    if opts.verbose {
        // Log proxy env vars curl-style before the connection attempt
        let proxy = ProxyEnv::from_env();
        if let Some(v) = &proxy.no_proxy {
            vinfo(
                opts.silent,
                &format!("Uses proxy env variable NO_PROXY == '{}'", v),
            );
        }
        if let Some(v) = &proxy.https_proxy {
            vinfo(
                opts.silent,
                &format!("Uses proxy env variable HTTPS_PROXY == '{}'", v),
            );
            vinfo(
                opts.silent,
                "WARNING: gRPC transport does not support HTTP CONNECT proxies; HTTPS_PROXY is ignored",
            );
        }
        if let Some(v) = &proxy.http_proxy {
            vinfo(
                opts.silent,
                &format!("Uses proxy env variable HTTP_PROXY == '{}'", v),
            );
            vinfo(
                opts.silent,
                "WARNING: gRPC transport does not support HTTP CONNECT proxies; HTTP_PROXY is ignored",
            );
        }

        vinfo(
            opts.silent,
            &format!("Trying {} ({})...", address, tls_label),
        );
        vinfo(
            opts.silent,
            &format!("gRPC method: {}/{}", full_service, method),
        );

        if let Some(headers) = doc.get_request_headers()
            && !headers.is_empty()
        {
            vsend(opts.silent, "");
            print_send_metadata(opts.silent, &headers);
        }
        vsend(opts.silent, "");
    }

    let timeout_seconds = opts.connect_timeout.max(opts.max_time);

    let config = GrpcClientConfig {
        address: address.clone(),
        timeout_seconds,
        tls_config,
        proto_config: None,
        metadata: doc.get_request_headers(),
        target_service: Some(full_service.clone()),
        compression: Default::default(),
    };

    let start = Instant::now();
    let mut client = GrpcClient::new(config).await?;

    let requests: Vec<Value> = doc.get_requests();
    if requests.is_empty() {
        return Err(anyhow::anyhow!("No REQUEST section found"));
    }

    if opts.very_verbose {
        let size: usize = requests
            .iter()
            .filter_map(|r| serde_json::to_vec(r).ok())
            .map(|v| v.len())
            .sum();
        vinfo(
            opts.silent,
            &format!("Estimated request size: {} bytes", size),
        );
    }

    let (tx, rx) = mpsc::channel::<Value>(100);
    let mut tx = Some(tx);
    let request_stream = ReceiverStream::new(rx);

    if let Some(tx_ref) = tx.as_mut() {
        for req in requests {
            tx_ref.send(req).await?;
        }
    }
    drop(tx.take());

    let (headers, mut response_stream) = client
        .call_stream(&full_service, &method, request_stream)
        .await?;

    if opts.verbose && !headers.is_empty() {
        print_recv_metadata(opts.silent, &headers);
        vrecv(opts.silent, "");
    }

    // -i: print response headers to stdout before body
    if opts.include_headers && !opts.silent && !headers.is_empty() {
        let mut pairs: Vec<_> = headers.iter().collect();
        pairs.sort_by_key(|(k, _)| k.as_str());
        for (k, v) in &pairs {
            println!("< {}: {}", k, v);
        }
        println!();
    }

    // -D: dump response headers to file
    if let Some(f) = opts.header_file.as_mut() {
        let mut pairs: Vec<_> = headers.iter().collect();
        pairs.sort_by_key(|(k, _)| k.as_str());
        for (k, v) in &pairs {
            writeln!(f, "{}: {}", k, v)?;
        }
    }

    let mut msg_count = 0usize;
    while let Some(item_res) = response_stream.next().await {
        match item_res {
            Ok(StreamItem::Message(msg)) => {
                msg_count += 1;
                if opts.very_verbose {
                    let size = serde_json::to_vec(&msg).map(|v| v.len()).unwrap_or(0);
                    vinfo(
                        opts.silent,
                        &format!("Estimated response size: {} bytes", size),
                    );
                }
                let output = serde_json::to_string_pretty(&msg)?;
                if let Some(f) = opts.output_file.as_mut() {
                    writeln!(f, "{}", output)?;
                } else if !opts.silent {
                    println!("{}", output);
                }
            }
            Ok(StreamItem::Trailers(trailers)) => {
                if opts.verbose && !trailers.is_empty() {
                    print_recv_metadata(opts.silent, &trailers);
                }

                // -i: print trailers to stdout after body
                if opts.include_headers && !opts.silent && !trailers.is_empty() {
                    let mut pairs: Vec<_> = trailers.iter().collect();
                    pairs.sort_by_key(|(k, _)| k.as_str());
                    for (k, v) in &pairs {
                        println!("< {}: {}", k, v);
                    }
                }

                // -D: dump trailers to file
                if let Some(f) = opts.header_file.as_mut() {
                    let mut pairs: Vec<_> = trailers.iter().collect();
                    pairs.sort_by_key(|(k, _)| k.as_str());
                    for (k, v) in &pairs {
                        writeln!(f, "{}: {}", k, v)?;
                    }
                }

                let grpc_status = trailers
                    .get("grpc-status")
                    .and_then(|s| s.parse::<u32>().ok())
                    .unwrap_or(0);

                if opts.verbose {
                    if grpc_status == 0 {
                        vinfo(opts.silent, "Connection #0 left intact");
                    } else {
                        let grpc_msg = trailers
                            .get("grpc-message")
                            .map(|s| s.as_str())
                            .unwrap_or("");
                        vinfo(
                            opts.silent,
                            &format!("gRPC error: code={} message={}", grpc_status, grpc_msg),
                        );
                    }
                }

                if opts.very_verbose {
                    let elapsed = start.elapsed();
                    vinfo(
                        opts.silent,
                        &format!(
                            "Elapsed: {:.3}s, messages received: {}",
                            elapsed.as_secs_f64(),
                            msg_count
                        ),
                    );
                }
            }
            Err(err) => {
                if opts.show_error && !opts.silent {
                    eprintln!("* gRPC error: {}", err);
                }
                return Err(anyhow::anyhow!(err));
            }
        }
    }

    Ok(())
}