use std::convert::Infallible;
use std::io::{BufRead, BufReader, Seek, SeekFrom};
use anyhow::{Result, ensure};
use askama::Template;
use axum::extract::Path;
use axum::http::{HeaderMap, StatusCode};
use axum::response::{
IntoResponse, Response, Sse,
sse::{Event, KeepAlive},
};
use axum::{Extension, Json};
use serde::Serialize;
use super::conversation;
use crate::config::UserConfig;
use crate::logs::catalog::{self, Log};
#[derive(Template)]
#[template(path = "console/logs.html")]
struct LogsPage {
page: super::PageContext,
logs: Vec<Log>,
owner: String,
}
#[derive(Template)]
#[template(path = "console/reader.html")]
struct ReaderPage {
page: super::PageContext,
log: Log,
html: String,
offset: u64,
}
pub(super) async fn index() -> Response {
inventory(None).await
}
pub(super) async fn command(Path(id): Path<String>) -> Response {
inventory(Some(id)).await
}
async fn inventory(owner: Option<String>) -> Response {
super::page(
tokio::task::spawn_blocking(move || -> Result<String> {
let config = UserConfig::load()?;
let logs = catalog::load(&config, owner.as_deref())?;
Ok(LogsPage {
page: super::PageContext::new(&config),
logs,
owner: owner.unwrap_or_default(),
}
.render()?)
})
.await,
)
}
pub(super) async fn reader(Path(id): Path<String>) -> Response {
let result = tokio::task::spawn_blocking(move || -> Result<Option<String>> {
let config = UserConfig::load()?;
let Some(log) = catalog::load(&config, None)?
.into_iter()
.find(|log| log.id == id)
else {
return Ok(None);
};
let lines = read_lines(&log, 0, usize::MAX)?;
let html = conversation::render(&String::from_utf8_lossy(&lines.bytes), log.kind);
Ok(Some(
ReaderPage {
page: super::PageContext::new(&config),
log,
html,
offset: lines.offset,
}
.render()?,
))
})
.await;
match result {
Ok(Ok(Some(html))) => super::page(Ok(Ok(html))),
Ok(Ok(None)) => StatusCode::NOT_FOUND.into_response(),
Ok(Err(error)) => super::page(Ok(Err(error))),
Err(error) => super::page(Err(error)),
}
}
#[derive(Debug, Serialize)]
struct Batch {
html: String,
offset: u64,
done: bool,
}
fn appended(config: &UserConfig, id: &str, offset: u64) -> Result<Batch> {
let log = catalog::find(config, id)?;
let lines = read_lines(&log, offset, 256 * 1024)?;
Ok(Batch {
html: conversation::render(&String::from_utf8_lossy(&lines.bytes), log.kind),
offset: lines.offset,
done: lines.done,
})
}
struct Lines {
bytes: Vec<u8>,
offset: u64,
done: bool,
}
fn read_lines(log: &Log, offset: u64, budget: usize) -> Result<Lines> {
let mut file = BufReader::new(std::fs::File::open(&log.path)?);
let length = file.get_ref().metadata()?.len();
ensure!(offset <= length, "log was truncated; reopen the reader");
file.seek(SeekFrom::Start(offset))?;
let mut bytes = Vec::new();
let mut next = offset;
while next < length && bytes.len() < budget {
let mut line = Vec::new();
if file.read_until(b'\n', &mut line)? == 0 {
break;
}
if log.live && !line.ends_with(b"\n") {
break;
}
next += line.len() as u64;
bytes.extend(line);
}
let done = !log.live && next >= file.get_ref().metadata()?.len();
Ok(Lines {
bytes,
offset: next,
done,
})
}
pub(super) async fn stream(
Path((id, offset)): Path<(String, u64)>,
Extension(changes): Extension<super::changes::Receiver>,
Extension(notifier): Extension<super::changes::Notifier>,
Extension(viewer): Extension<super::auth::Viewer>,
headers: HeaderMap,
) -> Response {
if headers
.get("accept")
.is_some_and(|value| value == "application/json")
{
return match tokio::task::spawn_blocking(move || {
appended(&UserConfig::load()?, &id, offset)
})
.await
{
Ok(Ok(batch)) => Json(batch).into_response(),
Ok(Err(error)) => super::page(Ok(Err(error))),
Err(error) => super::page(Err(error)),
};
}
let offset = headers
.get("last-event-id")
.and_then(|value| value.to_str().ok())
.and_then(|value| value.parse().ok())
.unwrap_or(offset);
let checked = tokio::task::spawn_blocking(move || -> Result<_> {
let config = UserConfig::load()?;
let log = catalog::find(&config, &id)?;
ensure!(
offset <= std::fs::metadata(log.path)?.len(),
"invalid log offset"
);
Ok(LogStream {
config: std::sync::Arc::new(config),
id,
offset,
done: false,
viewer,
_following: notifier.follow(changes.clone()),
changes,
accounts_revision: None,
})
})
.await;
let state = match checked {
Ok(Ok(state)) => state,
_ => return StatusCode::NOT_FOUND.into_response(),
};
let events = futures_util::stream::unfold(state, |mut state| async move {
if state.done {
return None;
}
loop {
if state.changes.has_changed().is_err() {
return None;
}
let revision = state.changes.borrow_and_update().accounts;
let authorize = state.accounts_revision != Some(revision);
state.accounts_revision = Some(revision);
let root = state.config.clone();
let selected = state.id.clone();
let signed_in = state.viewer.clone();
let offset = state.offset;
let result = tokio::task::spawn_blocking(move || {
if authorize && !signed_in.still_allowed()? {
return Ok(None);
}
appended(&root, &selected, offset).map(Some)
})
.await;
match result {
Ok(Ok(None)) => return None,
Ok(Ok(Some(batch))) if batch.offset != state.offset || batch.done => {
state.offset = batch.offset;
state.done = batch.done;
let event = Event::default()
.event("lines")
.id(state.offset.to_string())
.data(serde_json::to_string(&batch).expect("serializable log batch"));
return Some((Ok::<_, Infallible>(event), state));
}
Ok(Ok(Some(_))) => {
if state.changes.changed().await.is_err() {
return None;
}
}
error => {
eprintln!("thirdshift Console: log stream: {error:?}");
let event = Event::default()
.event("log-error")
.data("Can't read log; reopen the reader. Check the Console's stderr.");
state.done = true;
return Some((Ok(event), state));
}
}
}
});
let mut response = Sse::new(events)
.keep_alive(KeepAlive::default())
.into_response();
response
.headers_mut()
.insert("x-accel-buffering", "no".parse().unwrap());
response
.headers_mut()
.insert("cache-control", "no-store".parse().unwrap());
response
}
struct LogStream {
config: std::sync::Arc<UserConfig>,
id: String,
offset: u64,
done: bool,
viewer: super::auth::Viewer,
changes: super::changes::Receiver,
accounts_revision: Option<u64>,
_following: std::sync::Arc<super::watching::Following>,
}