podup 1.7.1

//! Image push to a registry (docker compose `push`).

use std::time::Duration;

use futures_util::{Stream, StreamExt};
use tracing::{info, warn};

use crate::compose::types::ComposeFile;
use crate::error::{ComposeError, Result};
use crate::libpod::types::image::ImagePullProgress;
use crate::libpod::{urlencoded, PodmanError, API_PREFIX};

use super::super::Engine;

/// Maximum time to wait for the next progress line from a push body stream
/// before treating the registry as unresponsive.
///
/// The client `READ_TIMEOUT` only bounds the request head, not this streamed
/// body, so without a per-line deadline a push to an unreachable/wedged registry
/// would block indefinitely while draining the response. Generous so a slow but
/// progressing upload is never aborted.
const PUSH_STALL_TIMEOUT: Duration = Duration::from_secs(120);

/// Drain a push progress stream, surfacing a mid-stream `error` line and bounding
/// each read by `stall` so an unresponsive registry fails with a clear timeout
/// instead of hanging. `quiet` (`-q/--quiet`) suppresses the per-line progress
/// and the final "pushed" line. Generic over the stream so it is unit-tested
/// without a live socket.
async fn drain_push_stream<S>(
	mut stream: S,
	image: &str,
	quiet: bool,
	stall: Duration,
) -> Result<()>
where
	S: Stream<Item = std::result::Result<ImagePullProgress, PodmanError>> + Unpin,
{
	let mut stream_error: Option<String> = None;
	loop {
		match tokio::time::timeout(stall, stream.next()).await {
			Ok(Some(Ok(progress))) => {
				if !progress.stream.is_empty() && !quiet {
					info!("{}", progress.stream.trim_end());
				}
				if !progress.error.is_empty() {
					stream_error = Some(progress.error.clone());
				}
			}
			Ok(Some(Err(e))) => stream_error = Some(e.to_string()),
			Ok(None) => break,
			Err(_elapsed) => {
				return Err(ComposeError::Build(format!(
					"push {image}: no progress from the registry for {}s; aborting \
					 (registry unreachable or unresponsive)",
					stall.as_secs()
				)));
			}
		}
	}
	match stream_error {
		Some(err) => Err(ComposeError::Build(format!("push {image}: {err}"))),
		None => {
			if quiet {
				tracing::debug!("pushed {image}");
			} else {
				info!("pushed {image}");
			}
			Ok(())
		}
	}
}

/// Options for [`Engine::push`], mirroring `docker compose push` (plus a Podman
/// `--tls-verify` escape hatch for insecure/local registries).
#[derive(Debug, Clone, Default)]
pub struct PushOptions {
	/// Continue with the remaining services after a push fails.
	pub ignore_failures: bool,
	/// Override TLS verification of the registry. `None` leaves Podman's default
	/// (verify on); `Some(false)` allows an insecure/HTTP registry.
	pub tls_verify: Option<bool>,
}

impl Engine {
	/// Push each service's image to its registry. Services without an `image:`
	/// (build-only or imageless) are skipped. Registry credentials come from
	/// Podman's auth file (`podman login`), so no auth handling is needed here.
	pub async fn push(
		&self,
		file: &ComposeFile,
		target_services: &[String],
		opts: PushOptions,
	) -> Result<()> {
		self.push_with_quiet(file, target_services, opts, false)
			.await
	}

	/// Push each service's image like [`Engine::push`], with `quiet` (`-q/--quiet`)
	/// suppressing the per-image progress output. Kept off the frozen
	/// [`PushOptions`] struct so the 1.0 library API stays stable.
	pub async fn push_with_quiet(
		&self,
		file: &ComposeFile,
		target_services: &[String],
		opts: PushOptions,
		quiet: bool,
	) -> Result<()> {
		for svc in target_services {
			if !file.services.contains_key(svc) {
				return Err(ComposeError::ServiceNotFound(svc.clone()));
			}
		}

		for (name, service) in &file.services {
			if !target_services.is_empty() && !target_services.iter().any(|t| t == name) {
				continue;
			}
			let Some(image) = service.image.as_deref() else {
				tracing::debug!("{name}: no image to push, skipping");
				continue;
			};
			if let Err(e) = self.push_image(image, &opts, quiet).await {
				if opts.ignore_failures {
					warn!("push {image} failed (ignored): {e}");
				} else {
					return Err(e);
				}
			}
		}
		Ok(())
	}

	/// Push a single image ref and drain its progress stream, surfacing a
	/// mid-stream `error` line as a failure.
	async fn push_image(&self, image: &str, opts: &PushOptions, quiet: bool) -> Result<()> {
		if quiet {
			tracing::debug!("pushing {image}");
		} else {
			info!("pushing {image}");
		}
		let mut query = format!("destination={}", urlencoded(image));
		if let Some(tls) = opts.tls_verify {
			query.push_str(&format!("&tlsVerify={tls}"));
		}
		let path = format!("{API_PREFIX}/images/{}/push?{query}", urlencoded(image));

		let resp = self
			.client
			.post_empty_stream(&path)
			.await
			.map_err(ComposeError::Podman)?;
		let stream = crate::libpod::parse_json_lines::<ImagePullProgress>(resp.into_body());
		drain_push_stream(stream, image, quiet, PUSH_STALL_TIMEOUT).await
	}
}

#[cfg(test)]
mod tests {
	use super::{drain_push_stream, ImagePullProgress};
	use crate::error::ComposeError;
	use crate::libpod::PodmanError;
	use futures_util::StreamExt;
	use std::time::Duration;

	fn progress(stream: &str, error: &str) -> ImagePullProgress {
		ImagePullProgress {
			stream: stream.to_string(),
			error: error.to_string(),
		}
	}

	#[tokio::test]
	async fn drain_ok_when_stream_completes_cleanly() {
		let items = vec![Ok(progress("pushing", "")), Ok(progress("done", ""))];
		let stream = futures_util::stream::iter(items);
		drain_push_stream(stream, "img", false, Duration::from_secs(5))
			.await
			.unwrap();
	}

	#[tokio::test]
	async fn drain_surfaces_mid_stream_error_line() {
		let items = vec![Ok(progress("", "denied: unauthorized"))];
		let stream = futures_util::stream::iter(items);
		let err = drain_push_stream(stream, "img", false, Duration::from_secs(5))
			.await
			.unwrap_err();
		assert!(matches!(err, ComposeError::Build(m) if m.contains("denied: unauthorized")));
	}

	#[tokio::test]
	async fn drain_times_out_on_an_unresponsive_stream() {
		// A stream that yields one line then never another stands in for a registry
		// that accepts the request then stalls — the per-line deadline must fire.
		let first = futures_util::stream::iter(vec![Ok(progress("pushing", ""))]);
		let stream = first.chain(futures_util::stream::pending::<
			std::result::Result<ImagePullProgress, PodmanError>,
		>());
		let err = drain_push_stream(stream, "img", false, Duration::from_millis(20))
			.await
			.unwrap_err();
		assert!(matches!(err, ComposeError::Build(m) if m.contains("no progress")));
	}
}