89109d2
use std::collections::{HashMap, VecDeque};
use std::sync::{Arc, Mutex, OnceLock};
use hayro::hayro_interpret::InterpreterSettings;
use hayro::hayro_syntax::Pdf;
use hayro::{RenderCache, RenderSettings, render};
use serde_json::json;
use tokio::sync::mpsc;
use crate::rpcgit::commit::types::TreeEntryMode;
use crate::rpcgit::error::BabyError;
use crate::lfs;
use crate::service::GitService;
use crate::service::blobs;
use crate::share::errors::{GitError, map_baby};
/// Page rasterisation for the in-repo PDF viewer.
///
/// The browser's own PDF surface cannot be reached from an embedded same-origin
/// frame, so the server rasterises the document and streams it back one page at
/// a time. Rendering is CPU-bound and runs on a blocking task; the channel is
/// shallow, so a reader that stops consuming also stops the renderer instead of
/// rasterising all 200+ pages for nothing.
pub const PAGE_SCALE_MIN: f32 = 0.5;
pub const PAGE_SCALE_MAX: f32 = 3.0;
pub const PAGE_SCALE_DEFAULT: f32 = 2.0;
/// Document ceiling: past this the stream ends early and says so.
pub const PAGE_LIMIT: usize = 400;
/// Frames the renderer may run ahead of the reader.
const CHANNEL_DEPTH: usize = 2;
/// Rasterised pages kept in the process for repeat views, shared by every
/// viewer of the same blob and scale.
const CACHE_BUDGET_BYTES: usize = 96 * 1024 * 1024;
pub fn clamp_scale(scale: Option<f32>) -> f32 {
match scale {
Some(v) if v.is_finite() => v.clamp(PAGE_SCALE_MIN, PAGE_SCALE_MAX),
_ => PAGE_SCALE_DEFAULT,
}
}
/// Frame the payload as `u32 header_len | header | u32 payload_len | payload`,
/// all big-endian. The header is JSON tagged by `kind`:
/// `doc` (once, upfront), `page` (PNG payload), `page_error`, `done`. Frames are
/// self-describing, so a single failed page can be reported inline instead of
/// killing the stream, and the reader needs no lookahead beyond the two lengths.
fn encode_frame(header: &str, payload: &[u8]) -> Vec<u8> {
let mut out = Vec::with_capacity(header.len() + payload.len() + 8);
out.extend_from_slice(&(header.len() as u32).to_be_bytes());
out.extend_from_slice(header.as_bytes());
out.extend_from_slice(&(payload.len() as u32).to_be_bytes());
out.extend_from_slice(payload);
out
}
fn frame(value: serde_json::Value, payload: &[u8]) -> Vec<u8> {
encode_frame(&value.to_string(), payload)
}
struct CachedPage {
png: Arc<Vec<u8>>,
width: u16,
height: u16,
}
#[derive(Default)]
struct PageCache {
pages: HashMap<String, CachedPage>,
order: VecDeque<String>,
bytes: usize,
}
fn page_cache() -> &'static Mutex<PageCache> {
static CACHE: OnceLock<Mutex<PageCache>> = OnceLock::new();
CACHE.get_or_init(|| Mutex::new(PageCache::default()))
}
fn cache_get(key: &str) -> Option<(Arc<Vec<u8>>, u16, u16)> {
let cache = page_cache().lock().ok()?;
let hit = cache.pages.get(key)?;
Some((hit.png.clone(), hit.width, hit.height))
}
fn cache_put(key: String, png: Arc<Vec<u8>>, width: u16, height: u16) {
if png.len() > CACHE_BUDGET_BYTES {
return;
}
let Ok(mut cache) = page_cache().lock() else {
return;
};
if let Some(prev) = cache.pages.remove(&key) {
cache.bytes = cache.bytes.saturating_sub(prev.png.len());
cache.order.retain(|k| k != &key);
}
while cache.bytes + png.len() > CACHE_BUDGET_BYTES {
let Some(oldest) = cache.order.pop_front() else {
break;
};
if let Some(evicted) = cache.pages.remove(&oldest) {
cache.bytes = cache.bytes.saturating_sub(evicted.png.len());
}
}
cache.bytes += png.len();
cache.order.push_back(key.clone());
cache.pages.insert(key, CachedPage { png, width, height });
}
/// KenShell resolves refs through its git backend, so an unknown revision surfaces as a
/// non-zero exit rather than a typed error. On this read path the only thing
/// that matters is the resulting status: an unresolvable ref is a 404, exactly
/// like a missing path.
fn is_missing_ref(error: &BabyError) -> bool {
match error {
BabyError::RevisionNotFound(_)
| BabyError::RevisionInvalid(_)
| BabyError::RefNotFound(_)
| BabyError::PathNotFound { .. } => true,
BabyError::Command(cmd) => has_missing_ref_text(&cmd.to_string()),
// KenShell only classifies the git message shapes it knows; anything
// else (such as `fatal: Not a valid object name`) lands here, so the
// stderr has to be inspected once more.
BabyError::Spawn { stderr, .. } => has_missing_ref_text(stderr),
_ => false,
}
}
fn has_missing_ref_text(text: &str) -> bool {
let text = text.to_ascii_lowercase();
text.contains("not a valid object name")
|| text.contains("unknown revision")
|| text.contains("bad revision")
|| text.contains("does not exist")
|| text.contains("needed a single revision")
}
/// Where the bytes of a previewable file actually live, and under which cache
/// key they belong.
struct Source {
bytes: Vec<u8>,
/// Content-addressed identity: the git blob id, or the LFS object id. A
/// branch that moves or a tag that is rewritten can therefore never replay
/// an older file's pages.
key: String,
}
/// Whole bytes for a PDF, resolving through Git LFS when the blob is a pointer.
///
/// A pointer blob is ~130 bytes of text, so an LFS-tracked `paper.pdf` must be
/// followed to object storage or the rasteriser only ever sees the pointer. A
/// partial file cannot be parsed either way, so the size cap rejects the
/// document instead of truncating it.
async fn pdf_bytes(
service: &GitService,
space: &str,
repo: &str,
revision: &str,
path: &str,
) -> Result<Source, GitError> {
let ctx = service.context(space.to_string(), repo.to_string()).await?;
let baby = service.baby(&ctx);
// A missing ref, a missing path and a directory all read as not-found here;
// only a file that really is not a readable PDF is a bad request.
let entry = match baby
.tree_entry_at(revision, std::path::Path::new(path))
.await
{
Ok(Some(entry)) => entry,
Ok(None) => return Err(GitError::NotFound),
Err(e) if is_missing_ref(&e) => return Err(GitError::NotFound),
Err(e) => return Err(map_baby(e)),
};
if !matches!(
entry.mode,
TreeEntryMode::Blob | TreeEntryMode::Exec | TreeEntryMode::Symlink
) {
return Err(GitError::NotFound);
}
let max = service.state.config.git.max_blob_size;
let size = baby
.get_blob_size(entry.oid.clone())
.await
.map_err(map_baby)?;
// Sniff before reading the whole file: `parse_lfs_pointer` only needs its
// first lines, and a plain multi-hundred-megabyte PDF should be rejected by
// the size cap rather than paged into memory first.
let sniff = baby
.get_blob(
entry.oid.clone(),
crate::rpcgit::blob::types::GetBlobOptions {
limit: Some(blobs::SNIFF_LIMIT),
},
)
.await
.map_err(map_baby)?;
if let Some(pointer) = blobs::parse_lfs_pointer(&sniff.data) {
return lfs_source(service, &ctx, space, repo, pointer).await;
}
if size.max(0) as u64 > max {
return Err(GitError::InvalidArgument(format!(
"pdf blob is {size} bytes, above the {max} byte preview limit"
)));
}
let blob = baby
.get_blob(
entry.oid.clone(),
crate::rpcgit::blob::types::GetBlobOptions { limit: None },
)
.await
.map_err(map_baby)?;
Ok(Source {
bytes: blob.data,
key: entry.oid.to_string(),
})
}
/// Follow one LFS pointer to object storage.
async fn lfs_source(
service: &GitService,
ctx: &crate::context::facade::GitContext,
space: &str,
repo: &str,
pointer: blobs::LfsPointerDto,
) -> Result<Source, GitError> {
if !ctx.model.enable_lfs {
return Err(GitError::InvalidArgument(
"file is an lfs pointer but lfs is disabled on this repo".to_string(),
));
}
let raw = pointer
.oid
.ok_or_else(|| GitError::InvalidArgument("lfs pointer has no object id".to_string()))?;
let oid = lfs::canonical_oid(&raw)?;
let max = service.state.config.git.max_blob_size;
if let Some(size) = pointer.size
&& size.max(0) as u64 > max
{
return Err(GitError::InvalidArgument(format!(
"lfs object is {size} bytes, above the {max} byte preview limit"
)));
}
let mut bytes = Vec::new();
let written = service
.lfs_download_to(space.to_string(), repo.to_string(), oid.clone(), &mut bytes)
.await?;
if written > max {
return Err(GitError::InvalidArgument(format!(
"lfs object is {written} bytes, above the {max} byte preview limit"
)));
}
Ok(Source { bytes, key: oid })
}
/// Stream `path` as framed pages. `Err` covers everything that has to fail
/// before the response starts (missing blob, unreadable document).
pub async fn render_stream(
service: &GitService,
space: &str,
repo: &str,
revision: &str,
path: &str,
scale: f32,
) -> Result<mpsc::Receiver<Vec<u8>>, GitError> {
let source = pdf_bytes(service, space, repo, revision, path).await?;
let pdf = tokio::task::spawn_blocking(move || Pdf::new(source.bytes))
.await
.map_err(|e| GitError::Internal(format!("pdf task failed: {e}")))?
.map_err(|e| GitError::InvalidArgument(format!("unreadable pdf: {e:?}")))?;
let (tx, rx) = mpsc::channel::<Vec<u8>>(CHANNEL_DEPTH);
// Content-addressed cache key (git oid, or LFS object id): two paths that
// hold the same bytes share one render, and no revision name can replay an
// older file's pages.
let prefix = format!("pdf.{space}/{repo}/{}@{scale}", source.key);
tokio::task::spawn_blocking(move || {
let pages = pdf.pages();
let available = pages.len();
let total = available.min(PAGE_LIMIT);
let doc = json!({
"kind": "doc",
"pages": total,
"available": available,
"truncated": available > total,
});
if tx.blocking_send(frame(doc, &[])).is_err() {
return;
}
let settings = RenderSettings {
x_scale: scale,
y_scale: scale,
bg_color: hayro::vello_cpu::color::palette::css::WHITE,
..Default::default()
};
let cache = RenderCache::new();
let interpreter = InterpreterSettings::default();
for index in 0..total {
let key = format!("{prefix}:{index}");
let (png, width, height) = match cache_get(&key) {
Some(hit) => hit,
None => {
let pixmap = render(&pages[index], &cache, &interpreter, &settings);
let (width, height) = (pixmap.width(), pixmap.height());
match pixmap.into_png() {
Ok(data) => {
let png = Arc::new(data);
cache_put(key, png.clone(), width, height);
(png, width, height)
}
Err(e) => {
let err = json!({
"kind": "page_error",
"page": index + 1,
"total": total,
"error": e.to_string(),
});
if tx.blocking_send(frame(err, &[])).is_err() {
return;
}
continue;
}
}
}
};
let header = json!({
"kind": "page",
"page": index + 1,
"total": total,
"width": width,
"height": height,
});
if tx.blocking_send(frame(header, &png)).is_err() {
return;
}
}
let done = json!({ "kind": "done", "pages": total });
let _ = tx.blocking_send(frame(done, &[]));
});
Ok(rx)
}