// r8r: rust port of n8n.io
// Initial author: Michael Cochran <mcochran@cochranblock.org>
// Built for: Tanner Woodrum, intended cofounder
// License: Unlicense (public domain dedication, see UNLICENSE)
//
// `ssh` — connect to a remote host and execute a command. Pure-Rust
// via russh — no libssh2 / system C library. Federal-friendly: same
// supply-chain story as the rest of the binary.
//
// Credential: ssh = {
//   host = "10.0.0.5"
//   port = 22                         (optional, default 22)
//   username = "ops"
//   password = "..."                  (one of password/private_key required)
//   private_key = "-----BEGIN ..."    (PEM-encoded ed25519 / RSA key)
//   accept_unknown_host = true        (optional, default true — see comment below)
// }
//
// Parameters:
//   command       the shell command to run
//   timeout_ms    max wait for command completion (default 30000)
//
// Returns: { exit_code, stdout, stderr, success } per item.
//
// Host key checking: v0.1 accepts any host key by default. For
// production / federal deployments, set known_hosts via the credential
// (v0.2 — current implementation always accepts).

use async_trait::async_trait;
use crate::core::error::NodeError;
use crate::core::types::*;
use crate::engine::{ExecuteCtx, Node, NodeFactory};
use russh::client::{self, Config, Handle, Handler};
use russh::{ChannelMsg, Disconnect};
use russh_keys::{decode_secret_key, key};
use serde_json::json;
use std::sync::Arc;
use std::time::Duration;

pub struct SshNode;

static META: NodeMetadata = NodeMetadata {
    name: "ssh",
    display_name: "SSH",
    version: &[1],
    group: &["transform"],
    description: "Connect to a remote host via SSH and execute a command",
    inputs: 1,
    outputs: 1,
    properties: &[
        PropertyDef {
            name: "command",
            display_name: "Command",
            kind: PropertyKind::String,
            required: true,
            default: None,
        },
        PropertyDef {
            name: "timeout_ms",
            display_name: "Timeout (ms)",
            kind: PropertyKind::Number,
            required: false,
            default: Some("30000"),
        },
    ],
    credentials: &[CredentialRef { name: "ssh", required: true }],
};

#[async_trait]
impl Node for SshNode {
    fn metadata(&self) -> &'static NodeMetadata { &META }

    async fn execute(&self, ctx: &mut ExecuteCtx<'_>) -> Result<NodeOutput, NodeError> {
        let host = ctx.credential_field("host")?
            .as_str().ok_or_else(|| NodeError::MissingCredential("ssh.host".into()))?
            .to_string();
        let port = ctx.credential_field("port").ok()
            .and_then(|v| v.as_u64()).map(|n| n as u16).unwrap_or(22);
        let username = ctx.credential_field("username")?
            .as_str().ok_or_else(|| NodeError::MissingCredential("ssh.username".into()))?
            .to_string();
        let password = ctx.credential_field("password").ok()
            .and_then(|v| v.as_str().map(|s| s.to_string()));
        let private_key = ctx.credential_field("private_key").ok()
            .and_then(|v| v.as_str().map(|s| s.to_string()));

        let command = ctx.param_str("command")?;
        let timeout_ms = ctx.node_def.parameters.get("timeout_ms")
            .and_then(|v| v.as_u64()).unwrap_or(30_000);

        let result = tokio::time::timeout(
            Duration::from_millis(timeout_ms),
            run_ssh(&host, port, &username, password.as_deref(), private_key.as_deref(), &command),
        ).await
        .map_err(|_| NodeError::Other(format!("ssh: command timed out after {}ms", timeout_ms)))??;

        Ok(vec![vec![NodeItem {
            json: json!({
                "exit_code": result.exit_code,
                "stdout": result.stdout,
                "stderr": result.stderr,
                "success": result.exit_code == 0,
            }),
            ..Default::default()
        }]])
    }
}

struct SshResult {
    exit_code: i32,
    stdout: String,
    stderr: String,
}

async fn run_ssh(
    host: &str,
    port: u16,
    username: &str,
    password: Option<&str>,
    private_key: Option<&str>,
    command: &str,
) -> Result<SshResult, NodeError> {
    let config = Arc::new(Config::default());
    let handler = AcceptAllHandler;
    let mut session: Handle<AcceptAllHandler> = client::connect(config, (host, port), handler).await
        .map_err(|e| NodeError::Other(format!("ssh: connect {}:{}: {}", host, port, e)))?;

    // Auth: pubkey takes precedence if both are provided.
    let authed = if let Some(key_pem) = private_key {
        let kp = decode_secret_key(key_pem, None)
            .map_err(|e| NodeError::Other(format!("ssh: parse private_key: {}", e)))?;
        session.authenticate_publickey(username, Arc::new(kp)).await
            .map_err(|e| NodeError::Other(format!("ssh: auth pubkey: {}", e)))?
    } else if let Some(pw) = password {
        session.authenticate_password(username, pw).await
            .map_err(|e| NodeError::Other(format!("ssh: auth password: {}", e)))?
    } else {
        return Err(NodeError::MissingCredential("ssh.password or ssh.private_key".into()));
    };
    if !authed {
        return Err(NodeError::Other("ssh: authentication rejected".into()));
    }

    let mut channel = session.channel_open_session().await
        .map_err(|e| NodeError::Other(format!("ssh: open channel: {}", e)))?;
    channel.exec(true, command).await
        .map_err(|e| NodeError::Other(format!("ssh: exec: {}", e)))?;

    let mut stdout = Vec::new();
    let mut stderr = Vec::new();
    let mut exit_code: i32 = -1;
    while let Some(msg) = channel.wait().await {
        match msg {
            ChannelMsg::Data { ref data } => stdout.extend_from_slice(data),
            ChannelMsg::ExtendedData { ref data, ext: 1 } => stderr.extend_from_slice(data),
            ChannelMsg::ExitStatus { exit_status } => exit_code = exit_status as i32,
            ChannelMsg::Eof => break,
            _ => {}
        }
    }

    let _ = session.disconnect(Disconnect::ByApplication, "", "en").await;

    Ok(SshResult {
        exit_code,
        stdout: String::from_utf8_lossy(&stdout).into_owned(),
        stderr: String::from_utf8_lossy(&stderr).into_owned(),
    })
}

/// Accept any host key. v0.2 will load `~/.ssh/known_hosts` style config
/// from credential.known_hosts and verify.
struct AcceptAllHandler;

#[async_trait]
impl Handler for AcceptAllHandler {
    type Error = russh::Error;

    async fn check_server_key(
        &mut self,
        _server_public_key: &key::PublicKey,
    ) -> Result<bool, Self::Error> {
        Ok(true)
    }
}

inventory::submit! { NodeFactory::new("ssh", || Arc::new(SshNode)) }
