//! The git processes the site runs: `git http-backend` for clones and
//! fetches, and `git cat-file` for the raw view. libgit2 answers the
//! site's questions in-process, but it cannot stream a packed object and
//! pack generation is git's job anyway, so bytes that leave the site in
//! bulk come from git itself. Every such process occupies one of a
//! bounded number of slots, and is killed once nothing has moved between
//! it and its client for longer than the idle timeout, so that a crowd of
//! slow or silent clients cannot pin processes for good.

use std::{
    io,
    path::{Path, PathBuf},
    process::Stdio,
    sync::{
        Arc,
        atomic::{AtomicU64, Ordering},
    },
    time::{Duration, Instant},
};

use anyhow::{Context as _, Error, anyhow};
use poem::{
    Body, IntoResponse,
    http::{StatusCode, header},
};
use tokio::{
    io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader, DuplexStream},
    process::{Child, ChildStdin, ChildStdout, Command},
    sync::{OwnedSemaphorePermit, Semaphore},
    time::timeout,
};

// Bytes moved per read or write between a process and its client.
const CHUNK: usize = 64 * 1024;

/// Where git processes are started from: the store they may touch, and
/// the bounds every one of them runs under.
pub struct Git {
    root: PathBuf,
    slots: Arc<Semaphore>,
    idle: Duration,
}

impl Git {
    /// Runs git against the repositories under `root`, at most `processes`
    /// at a time, none of them left stalled for longer than `idle`.
    pub fn new(root: PathBuf, processes: u32, idle: Duration) -> Self {
        Git {
            root,
            slots: Arc::new(Semaphore::new(processes as usize)),
            idle,
        }
    }

    /// The directory of repositories.
    pub fn root(&self) -> &Path {
        &self.root
    }

    /// Starts `git` in a free slot, waiting up to the idle timeout for
    /// one. `configure` adds the arguments and environment of the
    /// particular command to a process that otherwise sees nothing but
    /// `PATH` and the store declared safe. `input`, when given, is fed to
    /// the process as it arrives; otherwise the process has no input.
    /// `what` names the command in the journal.
    pub async fn start(
        &self,
        what: &'static str,
        input: Option<Body>,
        configure: impl FnOnce(&mut Command),
    ) -> Result<Process, StartError> {
        let slot = match timeout(self.idle, self.slots.clone().acquire_owned()).await {
            Ok(Ok(slot)) => slot,
            Ok(Err(closed)) => return Err(StartError::Failed(closed.into())),
            Err(_) => {
                return Err(StartError::Busy {
                    retry_after: self.idle,
                });
            }
        };
        let mut command = Command::new("git");
        command
            .env_clear()
            .env("PATH", std::env::var_os("PATH").unwrap_or_default())
            // The repositories belong to the hosting user, not to the one
            // serving them, and git refuses to touch a repository owned by
            // someone else unless it is declared safe. The configured
            // store is trusted by definition.
            .env("GIT_CONFIG_COUNT", "1")
            .env("GIT_CONFIG_KEY_0", "safe.directory")
            .env("GIT_CONFIG_VALUE_0", format!("{}/*", self.root.display()))
            .stdin(if input.is_some() {
                Stdio::piped()
            } else {
                Stdio::null()
            })
            .stdout(Stdio::piped())
            .stderr(Stdio::inherit())
            .kill_on_drop(true);
        configure(&mut command);
        let mut child = command
            .spawn()
            .with_context(|| format!("starting git {what}"))?;
        let stdout = child
            .stdout
            .take()
            .ok_or_else(|| anyhow!("no output pipe for git {what}"))?;
        let activity = Arc::new(Activity::new());
        if let Some(body) = input {
            let stdin = child
                .stdin
                .take()
                .ok_or_else(|| anyhow!("no input pipe for git {what}"))?;
            tokio::spawn(feed(body, stdin, activity.clone(), self.idle, what));
        }
        Ok(Process {
            child,
            output: BufReader::with_capacity(CHUNK, stdout),
            activity,
            _slot: slot,
            idle: self.idle,
            what,
        })
    }
}

/// Why a git process could not be started.
#[derive(Debug)]
pub enum StartError {
    /// Every slot stayed taken for the whole idle timeout.
    Busy {
        retry_after: Duration,
    },
    Failed(Error),
}

impl From<Error> for StartError {
    fn from(err: Error) -> Self {
        StartError::Failed(err)
    }
}

// Being refused for lack of a slot is an answer to the client, not a
// failure of the site, so it travels as a complete response: a plain 503
// naming the wait the client should observe.
impl From<StartError> for poem::Error {
    fn from(err: StartError) -> Self {
        match err {
            StartError::Busy { retry_after } => poem::Error::from_response(
                "The site is serving as many clones and downloads as it can at once. Try again shortly.\n"
                    .with_status(StatusCode::SERVICE_UNAVAILABLE)
                    .with_header(header::RETRY_AFTER, retry_after.as_secs().max(1).to_string())
                    .into_response(),
            ),
            StartError::Failed(err) => err.into(),
        }
    }
}

/// A running git process in its slot. Dropping it kills the process and
/// frees the slot.
pub struct Process {
    child: Child,
    output: BufReader<ChildStdout>,
    activity: Arc<Activity>,
    _slot: OwnedSemaphorePermit,
    idle: Duration,
    what: &'static str,
}

impl Process {
    /// The next line of output without its line ending, or `None` once
    /// the output is over.
    pub async fn read_line(&mut self) -> Result<Option<String>, Error> {
        let mut line = String::new();
        let read = within(self.idle, self.what, self.output.read_line(&mut line)).await?;
        if read == 0 {
            return Ok(None);
        }
        line.truncate(line.trim_end_matches(['\r', '\n']).len());
        Ok(Some(line))
    }

    /// The output produced so far, left in place for whatever reads next;
    /// empty once the output is over.
    pub async fn peek(&mut self) -> Result<&[u8], Error> {
        within(self.idle, self.what, self.output.fill_buf()).await
    }

    /// The rest of the output, as a response body. The process is reaped
    /// once its output is over, and killed if it or the client stalls.
    pub fn into_body(self) -> Body {
        // A pipe whose ends notice each other going away: the body sees
        // the end of the output once the pump drops its end, and the pump
        // fails to write once the body has been dropped.
        let (ours, theirs) = tokio::io::duplex(2 * CHUNK);
        tokio::spawn(self.pump(ours));
        Body::from_async_read(theirs)
    }

    // Moves output to the client until it is over, then sees the process
    // out.
    async fn pump(mut self, mut sink: DuplexStream) {
        let outcome = self.copy_output(&mut sink).await;
        drop(sink);
        let stall = match outcome {
            Ok(()) => return self.finish().await,
            Err(stall) => stall,
        };
        let seconds = self.idle.as_secs();
        match stall {
            Stall::Silent => tracing::warn!(
                command = self.what,
                "nothing moved for {seconds}s; killing the process"
            ),
            Stall::ClientNotReading => tracing::info!(
                command = self.what,
                "client read nothing for {seconds}s; killing the process"
            ),
            Stall::ClientGone => {
                tracing::debug!(command = self.what, "client went away; killing the process")
            }
            Stall::ReadFailed(err) => tracing::warn!(
                command = self.what,
                error = %err,
                "reading git output failed; killing the process"
            ),
        }
        self.kill().await;
    }

    async fn copy_output(&mut self, sink: &mut DuplexStream) -> Result<(), Stall> {
        let mut buf = vec![0u8; CHUNK];
        loop {
            let read = loop {
                match timeout(self.idle, self.output.read(&mut buf)).await {
                    Ok(Ok(read)) => break read,
                    Ok(Err(err)) => return Err(Stall::ReadFailed(err)),
                    // Silence on the output is fine while input is still
                    // arriving: the process is working on it.
                    Err(_) if self.activity.idle_for() < self.idle => {}
                    Err(_) => return Err(Stall::Silent),
                }
            };
            if read == 0 {
                return Ok(());
            }
            match timeout(self.idle, sink.write_all(&buf[..read])).await {
                Ok(Ok(())) => self.activity.touch(),
                Ok(Err(_)) => return Err(Stall::ClientGone),
                Err(_) => return Err(Stall::ClientNotReading),
            }
        }
    }

    // Reaps a process whose output is over. One that keeps running
    // regardless is killed.
    async fn finish(mut self) {
        let waited = timeout(self.idle, self.child.wait()).await;
        match waited {
            Ok(Ok(status)) if status.success() => {}
            Ok(Ok(status)) => tracing::warn!(command = self.what, %status, "git failed"),
            Ok(Err(err)) => tracing::warn!(command = self.what, error = %err, "waiting for git"),
            Err(_) => {
                tracing::warn!(
                    command = self.what,
                    "git kept running for {}s after its output ended; killing it",
                    self.idle.as_secs()
                );
                self.kill().await;
            }
        }
    }

    async fn kill(mut self) {
        if let Err(err) = self.child.kill().await {
            tracing::debug!(command = self.what, error = %err, "killing git");
        }
    }
}

// How the output pump stopped short.
enum Stall {
    Silent,
    ClientNotReading,
    ClientGone,
    ReadFailed(io::Error),
}

// Waits for an operation on the process's output, for at most the idle
// timeout.
async fn within<T>(
    idle: Duration,
    what: &str,
    operation: impl Future<Output = io::Result<T>>,
) -> Result<T, Error> {
    match timeout(idle, operation).await {
        Ok(result) => Ok(result?),
        Err(_) => Err(anyhow!(
            "git {what} produced no output for {}s",
            idle.as_secs()
        )),
    }
}

// Feeds a request body to a process as it arrives, closing the process's
// input once the body is over or the client has sent nothing for the idle
// timeout. A process whose input ends early fails on its own, and the
// output side sees it out.
async fn feed(
    body: Body,
    mut stdin: ChildStdin,
    activity: Arc<Activity>,
    idle: Duration,
    what: &'static str,
) {
    let mut source = body.into_async_read();
    let mut buf = vec![0u8; CHUNK];
    loop {
        let read = match timeout(idle, source.read(&mut buf)).await {
            Ok(Ok(0)) => return,
            Ok(Ok(read)) => read,
            Ok(Err(err)) => {
                tracing::debug!(command = what, error = %err, "request body ended early");
                return;
            }
            Err(_) => {
                tracing::info!(
                    command = what,
                    "client sent no input for {}s; closing the process's input",
                    idle.as_secs()
                );
                return;
            }
        };
        activity.touch();
        match timeout(idle, stdin.write_all(&buf[..read])).await {
            Ok(Ok(())) => {}
            // The process is gone; the output side reports why.
            Ok(Err(_)) => return,
            Err(_) => {
                tracing::warn!(
                    command = what,
                    "git read no input for {}s; closing its input",
                    idle.as_secs()
                );
                return;
            }
        }
    }
}

// When bytes last moved in either direction, shared by both sides of a
// process.
struct Activity {
    started: Instant,
    last_millis: AtomicU64,
}

impl Activity {
    fn new() -> Self {
        Activity {
            started: Instant::now(),
            last_millis: AtomicU64::new(0),
        }
    }

    fn touch(&self) {
        self.last_millis
            .store(self.started.elapsed().as_millis() as u64, Ordering::Relaxed);
    }

    fn idle_for(&self) -> Duration {
        let last = Duration::from_millis(self.last_millis.load(Ordering::Relaxed));
        self.started.elapsed().saturating_sub(last)
    }
}