terraform_wrapper/

streaming.rs

//! Streaming JSON output from `terraform plan` and `terraform apply`.
//!
//! When run with `-json`, Terraform produces one JSON object per line (NDJSON),
//! each representing an event like resource creation, progress, or completion.
//!
//! This module provides [`stream_terraform`] which yields [`JsonLogLine`] events
//! as they arrive, useful for progress reporting in orchestration tools.
//!
//! # Example
//!
//! ```rust,no_run
//! use terraform_wrapper::prelude::*;
//! use terraform_wrapper::streaming::{stream_terraform, JsonLogLine};
//!
//! # async fn example() -> terraform_wrapper::error::Result<()> {
//! let tf = Terraform::builder().working_dir("./infra").build()?;
//!
//! let result = stream_terraform(&tf, ApplyCommand::new().auto_approve().json(), |line| {
//!     println!("[{}] {}", line.log_type, line.message);
//! }).await?;
//!
//! assert!(result.success);
//! # Ok(())
//! # }
//! ```

use std::process::Stdio;

use serde::{Deserialize, Serialize};
use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::process::Command as TokioCommand;
use tracing::{debug, trace, warn};

use crate::Terraform;
use crate::command::TerraformCommand;
use crate::error::{Error, Result};
use crate::exec::CommandOutput;

/// A single JSON log line from Terraform's streaming output.
///
/// Terraform emits these as NDJSON (one per line) when commands are run
/// with the `-json` flag.
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct JsonLogLine {
    /// Log level: "info", "warn", "error".
    #[serde(rename = "@level")]
    pub level: String,
    /// Human-readable message.
    #[serde(rename = "@message")]
    pub message: String,
    /// Module that emitted the message (e.g., "terraform.ui").
    #[serde(rename = "@module", default)]
    pub module: String,
    /// ISO 8601 timestamp.
    #[serde(rename = "@timestamp", default)]
    pub timestamp: String,
    /// Event type: "version", "planned_change", "change_summary",
    /// "apply_start", "apply_progress", "apply_complete",
    /// "apply_errored", "outputs", etc.
    #[serde(rename = "type", default)]
    pub log_type: String,
    /// Change details (for planned_change events).
    #[serde(default)]
    pub change: serde_json::Value,
    /// Hook details (for apply_start/apply_complete/apply_progress events).
    #[serde(default)]
    pub hook: serde_json::Value,
    /// Change summary (for change_summary events).
    #[serde(default)]
    pub changes: serde_json::Value,
    /// Output values (for outputs events).
    #[serde(default)]
    pub outputs: serde_json::Value,
}

/// Execute a Terraform command with streaming JSON output.
///
/// Spawns the terraform process and calls `handler` with each [`JsonLogLine`]
/// as it arrives on stdout. Lines that fail to parse as JSON are logged and
/// skipped.
///
/// Returns a [`CommandOutput`] with the complete stderr and exit code after
/// the process finishes. Stdout is empty since lines were consumed by the
/// handler.
///
/// The command should have `.json()` enabled. If not, lines won't parse and
/// will be skipped.
pub async fn stream_terraform<C, F>(
    tf: &Terraform,
    command: C,
    mut handler: F,
) -> Result<CommandOutput>
where
    C: TerraformCommand,
    F: FnMut(JsonLogLine),
{
    let mut args = command.args();

    // Inject -input=false for commands that support it (same as the command's execute())
    if tf.no_input {
        let subcmd = args.first().map(String::as_str).unwrap_or("");
        if matches!(subcmd, "init" | "plan" | "apply" | "destroy" | "import") {
            args.insert(1, "-input=false".to_string());
        }
    }

    let mut cmd = TokioCommand::new(&tf.binary);

    if let Some(ref working_dir) = tf.working_dir {
        cmd.arg(format!("-chdir={}", working_dir.display()));
    }

    for arg in &args {
        cmd.arg(arg);
    }

    // Global args at end (same ordering as exec.rs)
    for arg in &tf.global_args {
        cmd.arg(arg);
    }

    for (key, value) in &tf.env {
        cmd.env(key, value);
    }

    cmd.stdout(Stdio::piped());
    cmd.stderr(Stdio::piped());

    trace!(binary = ?tf.binary, args = ?args, "streaming terraform command");

    let mut child = cmd.spawn().map_err(|e| {
        if e.kind() == std::io::ErrorKind::NotFound {
            Error::NotFound
        } else {
            Error::Io {
                message: format!("failed to spawn terraform: {e}"),
                source: e,
            }
        }
    })?;

    let stdout = child.stdout.take().ok_or_else(|| Error::Io {
        message: "failed to capture stdout".to_string(),
        source: std::io::Error::other("no stdout"),
    })?;

    let mut reader = BufReader::new(stdout).lines();

    while let Some(line) = reader.next_line().await.map_err(|e| Error::Io {
        message: format!("failed to read stdout line: {e}"),
        source: e,
    })? {
        trace!(%line, "stream line");
        match serde_json::from_str::<JsonLogLine>(&line) {
            Ok(log_line) => handler(log_line),
            Err(e) => {
                warn!(%line, error = %e, "failed to parse streaming json line, skipping");
            }
        }
    }

    let output = child.wait_with_output().await.map_err(|e| Error::Io {
        message: format!("failed to wait for terraform: {e}"),
        source: e,
    })?;

    let stderr = String::from_utf8_lossy(&output.stderr).to_string();
    let exit_code = output.status.code().unwrap_or(-1);
    let success = output.status.success();

    debug!(exit_code, success, "streaming terraform command completed");

    if !success {
        return Err(Error::CommandFailed {
            command: args.first().cloned().unwrap_or_default(),
            exit_code,
            stdout: String::new(),
            stderr,
        });
    }

    Ok(CommandOutput {
        stdout: String::new(), // consumed by handler
        stderr,
        exit_code,
        success,
    })
}