//! Pages: one CI job (`/{owner}/{repo}/ci/jobs/{job_id}`), its live log (`.../log.txt`) and its
//! artifacts (`.../artifacts/{name}`). Reading follows repo Read, like the run pages. The job page
//! shows the last 512 KiB of the log with no JavaScript: while the job is not terminal it refreshes
//! itself with a meta tag. `log.txt` streams the whole log and keeps following it until the job is
//! terminal and the file is read to its end. Artifacts are blobs under `ci/<artifact id>` and are
//! always served as attachments.

use std::path::{Path, PathBuf};
use std::time::Duration;

use bytes::Bytes;
use forge_core::Store;
use forge_core::ci::{CiArtifact, CiJob, CiRun};
use futures_util::{Stream, TryStreamExt};
use http::header::{CONTENT_DISPOSITION, CONTENT_LENGTH, CONTENT_TYPE};
use http_body::Frame;
use http_body_util::StreamBody;
use tokio::io::{AsyncReadExt, AsyncSeekExt, SeekFrom};
use topcoat::Result;
use topcoat::context::{Cx, app_context};
use topcoat::router::error::not_found;
use topcoat::router::response::Response;
use topcoat::router::{Body, HeaderName, HeaderValue, StatusCode, page, path_param, route};
use topcoat::view::{View, component, view};

use crate::ci::{artifact_key, log_path};
use crate::i18n::{
    self, Lang,
    releases_ci::{self as g, K},
};
use crate::pages::ci_runs::{job_href, run_href, state_text};
use crate::pages::repo::repo_head;
use crate::pages::repo_ctx::{RepoCtx, RepoHead, load_repo};
use crate::state::AppState;
use crate::views::{enc, human_size, shell};

path_param!(owner);
path_param!(repo);
path_param!(job_id);
path_param!(artifact_name);

/// The most of the log the job page shows, from its end.
pub const TAIL_BYTES: u64 = 512 * 1024;
/// Bytes read per chunk while following a log.
const CHUNK_BYTES: usize = 64 * 1024;
/// How long `log.txt` waits before it looks at the file again.
const FOLLOW_POLL: Duration = Duration::from_secs(1);

/// The job and its run, both checked to belong to this repo. Anything else is 404.
async fn job_of(state: &AppState, ctx: &RepoCtx, id: &str) -> Result<(CiJob, CiRun)> {
    let Some(job) = state.store.job_by_id(id).await? else {
        return Err(not_found().into());
    };
    let Some(run) = state.store.run_by_id(&job.run_id).await? else {
        return Err(not_found().into());
    };
    if run.repo_id != ctx.repo.id {
        return Err(not_found().into());
    }
    Ok((job, run))
}

/// The last `TAIL_BYTES` of a log, from a line boundary, and whether anything was cut. A log that
/// does not exist yet is empty.
async fn log_tail(path: &Path) -> Result<(String, bool)> {
    let mut file = match tokio::fs::File::open(path).await {
        Ok(file) => file,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok((String::new(), false)),
        Err(e) => return Err(forge_core::Error::Io(e).into()),
    };
    let len = file.metadata().await.map_err(forge_core::Error::Io)?.len();
    let truncated = len > TAIL_BYTES;
    if truncated {
        file.seek(SeekFrom::Start(len - TAIL_BYTES))
            .await
            .map_err(forge_core::Error::Io)?;
    }
    let mut buf = Vec::new();
    (&mut file)
        .take(TAIL_BYTES)
        .read_to_end(&mut buf)
        .await
        .map_err(forge_core::Error::Io)?;
    if truncated && let Some(end) = buf.iter().position(|byte| *byte == b'\n') {
        buf.drain(..=end);
    }
    Ok((String::from_utf8_lossy(&buf).into_owned(), truncated))
}

/// Follows a log: yields what the file holds, then polls for new bytes once a second. The stream
/// ends when the job is terminal and a read after seeing that returns nothing, so every byte the
/// runner wrote is sent. A job that is gone, or a read error, ends the stream.
fn follow_log(
    store: Store,
    job_id: String,
    path: PathBuf,
) -> impl Stream<Item = std::io::Result<Bytes>> + Send + 'static {
    struct Follow {
        store: Store,
        job_id: String,
        path: PathBuf,
        file: Option<tokio::fs::File>,
    }
    futures_util::stream::unfold(
        Follow {
            store,
            job_id,
            path,
            file: None,
        },
        |mut follow| async move {
            loop {
                // Checked before the read, so a read that finds nothing after this is the end.
                let terminal = match follow.store.job_by_id(&follow.job_id).await {
                    Ok(Some(job)) => job.state.is_terminal(),
                    Ok(None) => true,
                    Err(e) => {
                        tracing::error!(error = %e, "ci: log follow could not read the job");
                        return None;
                    }
                };
                if follow.file.is_none() {
                    follow.file = tokio::fs::File::open(&follow.path).await.ok();
                }
                if let Some(file) = follow.file.as_mut() {
                    let mut buf = vec![0u8; CHUNK_BYTES];
                    match file.read(&mut buf).await {
                        Ok(0) => {}
                        Ok(n) => {
                            buf.truncate(n);
                            return Some((Ok(Bytes::from(buf)), follow));
                        }
                        Err(e) => {
                            tracing::error!(error = %e, "ci: log follow could not read the file");
                            return None;
                        }
                    }
                }
                if terminal {
                    return None;
                }
                tokio::time::sleep(FOLLOW_POLL).await;
            }
        },
    )
}

/// A header value for a download's file name: printable ASCII, with quotes and backslashes cut.
fn attachment_name(name: &str) -> String {
    name.chars()
        .map(|c| {
            if c.is_ascii_graphic() && c != '"' && c != '\\' {
                c
            } else {
                '_'
            }
        })
        .collect()
}

/// The job page's values, already shown as text.
struct JobView {
    name: String,
    state: String,
    state_class: String,
    runner: String,
    exit_code: String,
    message: String,
    started: String,
    finished: String,
    run_href: String,
    run_number: i64,
    log_href: String,
}

/// One artifact row: its link and its size.
struct ArtifactRow {
    href: String,
    name: String,
    size: String,
}

/// The job page body. The meta refresh is in the body, since the shell has no head hook.
#[component]
async fn job_body(
    head: RepoHead,
    lang: Lang,
    job: JobView,
    log: String,
    truncated: bool,
    artifacts: Vec<ArtifactRow>,
    refresh: bool,
) -> Result<impl View> {
    Ok(view! {
        repo_head(head: head)
        if refresh {
            <meta http-equiv="refresh" content="3">
        }
        <p class="muted small"><a href=(job.run_href.as_str())>(g::t(lang, K::RunNumber)) (job.run_number)</a></p>
        <section class="card">
            <h2>(job.name.as_str())</h2>
            <p class="muted small">
                <span class=(job.state_class.as_str())>(job.state.as_str())</span>
                if !job.runner.is_empty() {
                    " \u{b7} " (g::t(lang, K::Runner)) " " (job.runner.as_str())
                }
                if !job.exit_code.is_empty() {
                    " \u{b7} " (g::t(lang, K::Exit)) " " (job.exit_code.as_str())
                }
            </p>
            if !job.started.is_empty() {
                <p class="muted small">(g::t(lang, K::Started)) " " (job.started.as_str())</p>
            }
            if !job.finished.is_empty() {
                <p class="muted small">(g::t(lang, K::Finished)) " " (job.finished.as_str())</p>
            }
            if !job.message.is_empty() {
                <p>(job.message.as_str())</p>
            }
        </section>
        <section class="card">
            <h2>(g::t(lang, K::Log))</h2>
            <p class="muted small"><a href=(job.log_href.as_str())>(g::t(lang, K::FullLog))</a></p>
            if truncated {
                <p class="muted small">(g::t(lang, K::LogTruncated))</p>
            }
            <pre class="ci-log">(log.as_str())</pre>
        </section>
        if !artifacts.is_empty() {
            <section class="card">
                <h2>(g::t(lang, K::Artifacts))</h2>
                <ul class="ci-artifacts">
                    for artifact in artifacts.iter() {
                        <li><a href=(artifact.href.as_str())>(artifact.name.as_str())</a> " (" (artifact.size.as_str()) ")"</li>
                    }
                </ul>
            </section>
        }
    })
}

/// One job: its state, the last 512 KiB of its log and its artifacts.
#[page("/{owner}/{repo}/ci/jobs/{job_id}")]
async fn job_page(cx: &Cx) -> Result<impl View> {
    let owner = path_param::<Owner>(cx);
    let name = path_param::<Repo>(cx);
    let id = path_param::<JobId>(cx).to_owned();
    let ctx = load_repo(cx, owner, name).await?;
    let lang = i18n::lang(cx, ctx.viewer.as_ref().map(|(user, _)| user));
    let state = app_context::<AppState>(cx);
    let (job, run) = job_of(state, &ctx, &id).await?;
    let (log, truncated) = log_tail(&log_path(&state.config.data_dir, &job.id)).await?;
    let artifacts = state
        .store
        .list_artifacts(&job.id)
        .await?
        .into_iter()
        .map(|artifact| ArtifactRow {
            href: format!(
                "{}/artifacts/{}",
                job_href(&title_of(&ctx), &job.id),
                enc(&artifact.name)
            ),
            size: human_size(lang, u64::try_from(artifact.size).unwrap_or(0)),
            name: artifact.name,
        })
        .collect();
    let runner = match job.runner_id.as_deref() {
        Some(runner_id) => state
            .store
            .list_runners()
            .await?
            .into_iter()
            .find(|runner| runner.id == runner_id)
            .map(|runner| runner.name)
            .unwrap_or_default(),
        None => String::new(),
    };
    let title = title_of(&ctx);
    let refresh = !job.state.is_terminal();
    let view = JobView {
        name: job.name.clone(),
        state: state_text(lang, job.state.as_str()),
        state_class: format!("ci-state ci-{}", job.state.as_str()),
        runner,
        exit_code: job
            .exit_code
            .map(|code| code.to_string())
            .unwrap_or_default(),
        message: job.message.clone().unwrap_or_default(),
        started: job.started_at.clone().unwrap_or_default(),
        finished: job.finished_at.clone().unwrap_or_default(),
        run_href: run_href(&title, run.number),
        run_number: run.number,
        log_href: format!("{}/log.txt", job_href(&title, &job.id)),
    };
    let head = ctx.head(state, "ci").await?;
    let page_title = format!("{title} \u{b7} CI \u{b7} {}", view.name);
    Ok(view! {
        shell(title: page_title.as_str(), viewer: ctx.viewer.as_ref(),
            job_body(head: head, lang: lang, job: view, log: log, truncated: truncated,
                artifacts: artifacts, refresh: refresh))
    })
}

fn title_of(ctx: &RepoCtx) -> String {
    format!("{}/{}", ctx.repo.owner_name, ctx.repo.name)
}

/// Streams the job's log as plain text. It sends the file so far and then follows it until the job
/// is terminal.
#[route(GET "/{owner}/{repo}/ci/jobs/{job_id}/log.txt")]
async fn job_log(cx: &Cx) -> Result<Response> {
    let owner = path_param::<Owner>(cx);
    let name = path_param::<Repo>(cx);
    let id = path_param::<JobId>(cx).to_owned();
    let ctx = load_repo(cx, owner, name).await?;
    let state = app_context::<AppState>(cx);
    let (job, _) = job_of(state, &ctx, &id).await?;
    let path = log_path(&state.config.data_dir, &job.id);
    let chunks = follow_log(state.store.clone(), job.id.clone(), path).map_ok(Frame::data);
    let mut res = Response::new(Body::new(StreamBody::new(chunks)));
    *res.status_mut() = StatusCode::OK;
    let headers = res.headers_mut();
    headers.insert(
        CONTENT_TYPE,
        HeaderValue::from_static("text/plain; charset=utf-8"),
    );
    headers.insert(
        HeaderName::from_static("x-content-type-options"),
        HeaderValue::from_static("nosniff"),
    );
    headers.insert(
        HeaderName::from_static("cache-control"),
        HeaderValue::from_static("no-store"),
    );
    Ok(res)
}

/// Downloads an artifact of a job as an attachment.
#[route(GET "/{owner}/{repo}/ci/jobs/{job_id}/artifacts/{artifact_name}")]
async fn artifact_download(cx: &Cx) -> Result<Response> {
    let owner = path_param::<Owner>(cx);
    let name = path_param::<Repo>(cx);
    let id = path_param::<JobId>(cx).to_owned();
    let file_name = path_param::<ArtifactName>(cx).to_owned();
    let ctx = load_repo(cx, owner, name).await?;
    let state = app_context::<AppState>(cx);
    let (job, _) = job_of(state, &ctx, &id).await?;
    let Some(artifact) = state.store.artifact(&job.id, &file_name).await? else {
        return Err(not_found().into());
    };
    let Some(read) = state.blobs.read(&artifact_key(&artifact.id)).await? else {
        tracing::error!(artifact_id = %artifact.id, "ci: artifact row without a blob");
        return Err(not_found().into());
    };
    artifact_response(read.size, read.stream, &artifact)
}

/// The attachment response for an artifact's bytes.
fn artifact_response(
    size: u64,
    stream: futures_util::stream::BoxStream<'static, std::io::Result<Bytes>>,
    artifact: &CiArtifact,
) -> Result<Response> {
    let chunks = stream.map_ok(Frame::data);
    let mut res = Response::new(Body::new(StreamBody::new(chunks)));
    *res.status_mut() = StatusCode::OK;
    let headers = res.headers_mut();
    headers.insert(
        CONTENT_TYPE,
        HeaderValue::from_static("application/octet-stream"),
    );
    headers.insert(CONTENT_LENGTH, HeaderValue::from(size));
    if let Ok(value) = HeaderValue::from_str(&format!(
        "attachment; filename=\"{}\"",
        attachment_name(&artifact.name)
    )) {
        headers.insert(CONTENT_DISPOSITION, value);
    }
    headers.insert(
        HeaderName::from_static("x-content-type-options"),
        HeaderValue::from_static("nosniff"),
    );
    headers.insert(
        HeaderName::from_static("cache-control"),
        HeaderValue::from_static("private, no-cache"),
    );
    Ok(res)
}