//! 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. While the job is not terminal the page is live: it carries a
//! `data-live` marker for its `.../events` stream, and nav.js re-renders it when the job's status
//! changes. `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::sync::Arc;
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 tokio::sync::watch;
use topcoat::Result;
use topcoat::context::{Cx, app_context};
use topcoat::router::content::sse::{Event, last_event_id};
use topcoat::router::error::not_found;
use topcoat::router::request::uri;
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::{CiLive, artifact_key, log_path};
use crate::i18n::{
    self, Lang,
    releases_ci::{self as g, K},
};
use crate::pages::ci_runs::{
    changed_event, events_response, job_href, job_status, 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;
/// The longest a follower waits for a change to its job before it reads again anyway. A change
/// wakes it at once; this only guards against a missed wake.
const WAKE_BACKSTOP: Duration = Duration::from_secs(60);

/// 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 length of the longest whole-UTF-8 prefix of `buf`. A character that a read cut short at the
/// end is left out, so the next read sends it whole. Other invalid bytes are kept.
fn whole_utf8_len(buf: &[u8]) -> usize {
    match std::str::from_utf8(buf) {
        Ok(_) => buf.len(),
        Err(e) if e.error_len().is_none() => e.valid_up_to(),
        Err(_) => buf.len(),
    }
}

/// The last `TAIL_BYTES` of a log, from a line boundary, whether anything was cut, and the byte
/// offset just past the text shown. A log that does not exist yet is empty.
async fn log_tail(path: &Path) -> Result<(String, bool, u64)> {
    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, 0)),
        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;
    let mut start = 0;
    if truncated {
        start = len - TAIL_BYTES;
        file.seek(SeekFrom::Start(start))
            .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);
        start += end as u64 + 1;
    }
    buf.truncate(whole_utf8_len(&buf));
    let offset = start + buf.len() as u64;
    Ok((
        String::from_utf8_lossy(&buf).into_owned(),
        truncated,
        offset,
    ))
}

/// Follows a log: yields what the file holds, then sends new bytes each time the job changes (its
/// log grew or its state did). 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.
pub(crate) fn follow_log(
    store: Store,
    live: Arc<CiLive>,
    job_id: String,
    path: PathBuf,
) -> impl Stream<Item = std::io::Result<Bytes>> + Send + 'static {
    struct Follow {
        store: Store,
        live: Arc<CiLive>,
        job_id: String,
        path: PathBuf,
        file: Option<tokio::fs::File>,
        /// Subscribed before the first read, so a change between the read and the wait is seen.
        wake: watch::Receiver<u64>,
    }
    let wake = live.subscribe_job(&job_id);
    futures_util::stream::unfold(
        Follow {
            store,
            live,
            job_id,
            path,
            file: None,
            wake,
        },
        |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;
                }
                // A closed signal means the job finished, which the next check sees as terminal.
                // It is subscribed again so this wait does not return at once.
                if let Ok(Err(_)) = tokio::time::timeout(WAKE_BACKSTOP, follow.wake.changed()).await
                {
                    follow.wake = follow.live.subscribe_job(&follow.job_id);
                }
            }
        },
    )
}

/// 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,
    /// The status word as the events carry it, for the live check.
    status: 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. A job that is not terminal carries `live`, the URL of its event stream, and
/// the status the page shows. The log tail carries the offset its text ends at.
#[component]
async fn job_body(
    head: RepoHead,
    lang: Lang,
    job: JobView,
    log: String,
    log_offset: String,
    truncated: bool,
    artifacts: Vec<ArtifactRow>,
    live: Option<String>,
) -> Result<impl View> {
    Ok(view! {
        repo_head(head: head)
        <div data-live=(live.as_deref()) data-live-state=(job.status.as_str())>
        <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" data-live-log="true" data-offset=(log_offset.as_str())>(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>
        }
        </div>
    })
}

/// 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_offset) = 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 live = (!job.state.is_terminal()).then(|| format!("{}/events", job_href(&title, &job.id)));
    let view = JobView {
        name: job.name.clone(),
        status: job.state.as_str().to_owned(),
        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, log_offset: log_offset.to_string(),
                truncated: truncated, artifacts: artifacts, live: live))
    })
}

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

/// Bytes of log per `log` event. A bigger append is sent as several events.
const LOG_EVENT_BYTES: u64 = 64 * 1024;

/// The job's live stream. It opens with the status, sends a `changed` event when the status changes
/// and a `log` event with each new run of log bytes. A `log` event's id is the offset past its text,
/// so a reconnect (`Last-Event-ID`) resumes there. A new stream starts at `?from=`, the offset the
/// page already shows. The stream ends once the job is terminal and its log is read to the end.
#[route(GET "/{owner}/{repo}/ci/jobs/{job_id}/events")]
async fn job_events(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 offset = last_event_id(cx)
        .and_then(|id| id.parse::<u64>().ok())
        .or_else(|| from_query(cx))
        .unwrap_or(0);
    // Subscribed before the first read, so a change in between is sent.
    let wake = state.ci_live.subscribe_job(&job.id);
    let follow = JobFollow {
        store: state.store.clone(),
        job_id: job.id.clone(),
        path: log_path(&state.config.data_dir, &job.id),
        wake,
        offset,
        status: None,
        terminal: false,
        more: false,
        primed: false,
        pending_status: None,
        pending_log: None,
    };
    events_response(cx, job_stream(follow))
}

/// The `from` offset of the events URL, if it is a number.
fn from_query(cx: &Cx) -> Option<u64> {
    uri(cx).query().and_then(|query| {
        form_urlencoded::parse(query.as_bytes())
            .find(|(key, _)| key == "from")
            .and_then(|(_, value)| value.parse::<u64>().ok())
    })
}

/// One job's stream state. Each read of the status and the log happens after a wake, and a read that
/// sent log bytes is followed by another read at once, so a long append drains in order.
struct JobFollow {
    store: Store,
    job_id: String,
    path: PathBuf,
    wake: watch::Receiver<u64>,
    /// The log offset read so far. The next read starts here.
    offset: u64,
    /// The status last read.
    status: Option<String>,
    /// The job was terminal at the last read.
    terminal: bool,
    /// The last read sent log bytes, so more may be waiting.
    more: bool,
    /// False until the first read. Later reads wait for a wake first, unless `more` is set.
    primed: bool,
    pending_status: Option<String>,
    /// The text and the offset past it.
    pending_log: Option<(String, u64)>,
}

fn job_stream(follow: JobFollow) -> impl Stream<Item = Result<Event>> + Send + 'static {
    futures_util::stream::unfold(follow, |mut f| async move {
        loop {
            if let Some(status) = f.pending_status.take() {
                return Some((changed_event(status), f));
            }
            if let Some((text, end)) = f.pending_log.take() {
                let event = Event::new().event("log").data(text).id(end.to_string());
                return Some((Ok(event), f));
            }
            if f.terminal && !f.more {
                return None;
            }
            if f.primed && !f.more && f.wake.changed().await.is_err() {
                return None;
            }
            f.primed = true;
            f.more = false;
            let (status, terminal) = match job_status(&f.store, &f.job_id).await {
                Ok(Some(found)) => found,
                Ok(None) => return None,
                Err(e) => {
                    tracing::error!(error = %e, "ci: job stream could not read the job");
                    return None;
                }
            };
            f.terminal = terminal;
            if f.status.as_deref() != Some(status.as_str()) {
                f.status = Some(status.clone());
                f.pending_status = Some(status);
            }
            match read_log(&f.path, f.offset).await {
                Ok(Some((text, end))) => {
                    f.offset = end;
                    f.more = true;
                    f.pending_log = Some((text, end));
                }
                Ok(None) => {}
                Err(e) => {
                    tracing::error!(error = %e, "ci: job stream could not read the log");
                    return None;
                }
            }
        }
    })
}

/// The whole text of the log from `offset`, at most `LOG_EVENT_BYTES`, and the offset past it. None
/// when there is nothing new (no file, no bytes past `offset`, or only part of a character).
async fn read_log(path: &Path, offset: u64) -> std::io::Result<Option<(String, u64)>> {
    let mut file = match tokio::fs::File::open(path).await {
        Ok(file) => file,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(e) => return Err(e),
    };
    file.seek(SeekFrom::Start(offset)).await?;
    let mut buf = Vec::new();
    (&mut file)
        .take(LOG_EVENT_BYTES)
        .read_to_end(&mut buf)
        .await?;
    let whole = whole_utf8_len(&buf);
    if whole == 0 {
        return Ok(None);
    }
    buf.truncate(whole);
    Ok(Some((
        String::from_utf8_lossy(&buf).into_owned(),
        offset + whole as u64,
    )))
}

/// 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(),
        Arc::clone(&state.ci_live),
        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)
}