Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
50 kB · 957 lines
Rust
at master
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958use crate::{ error::ApiError, reader::{ CreateMaterialRequest, Material, bounded_text, get_material, index_article_tx, sync_catalog_tx, validate_sources, }, reader_pdf::{asset_path, enqueue_index, file_response, safe_relative_path}, state::AppState,};use axum::{ Json, Router, extract::{Multipart, Path, State}, http::{HeaderMap, StatusCode, header}, response::{IntoResponse, Response}, routing::{get, post},};use chrono::{DateTime, Utc};use serde_json::{Value, json};use sha2::{Digest, Sha256};use sqlx::{Postgres, Transaction};use std::{ collections::{BTreeMap, HashMap}, io::{Cursor, Read, Write}, path::{Path as FsPath, PathBuf},};use tokio::fs;use uuid::Uuid;use zip::{ZipArchive, ZipWriter, write::SimpleFileOptions};
const MAX_UPLOAD: usize = 256 * 1024 * 1024;type Files = BTreeMap<String, Vec<u8>>;
pub fn routes() -> Router<AppState> { Router::new() .route("/api/reader/import", post(import_file)) .route("/api/reader/backup", get(backup)) .route("/api/reader/materials/{id}/export", get(export_material))}fn invalid(message: impl Into<String>) -> ApiError { ApiError::BadRequest(message.into())}fn zip_error(error: impl std::fmt::Display) -> ApiError { invalid(format!("invalid lesson archive: {error}"))}
fn read_zip(bytes: Vec<u8>) -> Result<Files, ApiError> { let mut archive = ZipArchive::new(Cursor::new(bytes)).map_err(zip_error)?; if archive.len() > 20000 { return Err(invalid("archive contains too many files")); } let backup = archive.file_names().any(|name| name == "odre-backup.json"); let (max_file, max_total) = if backup { (256_000_000u64, 1_000_000_000u64) } else { (32_000_000, 256_000_000) }; let mut total = 0u64; let mut files = Files::new(); for i in 0..archive.len() { let mut file = archive.by_index(i).map_err(zip_error)?; let name = file.name().to_owned(); let clean = name.strip_suffix('/').unwrap_or(&name); if !safe_relative_path(clean) { return Err(invalid(format!("unsafe archive path: {name}"))); } if file .unix_mode() .is_some_and(|mode| mode & 0o170000 == 0o120000) { return Err(invalid( "symbolic links are not supported in lesson archives", )); } total = total .checked_add(file.size()) .ok_or_else(|| invalid("archive is too large"))?; if file.size() > max_file || total > max_total { return Err(invalid("archive exceeds the expanded size limit")); } if file.is_dir() { continue; } if files.contains_key(&name) { return Err(invalid("archive contains duplicate paths")); } let mut bytes = Vec::new(); (&mut file) .take(max_file + 1) .read_to_end(&mut bytes) .map_err(zip_error)?; if bytes.len() as u64 > max_file { return Err(invalid("archive file exceeds its size limit")); } files.insert(name, bytes); } Ok(files)}
fn write_zip(files: Files) -> Result<Vec<u8>, ApiError> { let mut writer = ZipWriter::new(Cursor::new(Vec::new())); let options = SimpleFileOptions::default() .compression_method(zip::CompressionMethod::Deflated) .unix_permissions(0o600); for (name, bytes) in files { if !safe_relative_path(&name) { return Err(invalid("invalid export path")); } writer.start_file(name, options).map_err(zip_error)?; writer.write_all(&bytes)?; } Ok(writer.finish().map_err(zip_error)?.into_inner())}
fn supported_asset(name: &str) -> bool { safe_relative_path(name) && matches!( name.rsplit('.') .next() .unwrap_or("") .to_ascii_lowercase() .as_str(), "png" | "jpg" | "jpeg" | "webp" | "gif" | "svg" | "avif" | "md" | "json" )}
async fn import_file( State(state): State<AppState>, mut multipart: Multipart,) -> Result<Response, ApiError> { let mut upload = None; while let Some(field) = multipart .next_field() .await .map_err(|e| invalid(e.to_string()))? { if field.name() != Some("file") { continue; } if upload.is_some() { return Err(invalid("only one file is supported")); } let name = field.file_name().unwrap_or("lesson.md").to_owned(); let bytes = field.bytes().await.map_err(|e| invalid(e.to_string()))?; if bytes.len() > MAX_UPLOAD { return Err(invalid("the file exceeds the 256 MiB upload limit")); } upload = Some((name, bytes.to_vec())); } let (name, bytes) = upload.ok_or_else(|| invalid("choose a Markdown file, lesson ZIP, backup ZIP or PDF"))?; let extension = name.rsplit('.').next().unwrap_or("").to_ascii_lowercase(); match extension.as_str() { "pdf" => Ok(( StatusCode::CREATED, Json(import_pdf(&state, &name, bytes).await?), ) .into_response()), "zip" => { let files = tokio::task::spawn_blocking(move || read_zip(bytes)) .await .map_err(anyhow::Error::from)??; if files.contains_key("odre-backup.json") { Ok(Json(restore_backup(&state, files).await?).into_response()) } else { Ok(( StatusCode::CREATED, Json(import_bundle(&state, files).await?), ) .into_response()) } } "md" => { let content = String::from_utf8(bytes).map_err(|_| invalid("Markdown must be UTF-8"))?; let title = FsPath::new(&name) .file_stem() .and_then(|s| s.to_str()) .unwrap_or("Article") .to_owned(); Ok(( StatusCode::CREATED, Json( crate::reader::create_article_record( &state, CreateMaterialRequest { title, subject: None, content, sources: None, }, ) .await?, ), ) .into_response()) } "json" => { let input: CreateMaterialRequest = serde_json::from_slice(&bytes) .map_err(|e| invalid(format!("invalid lesson JSON: {e}")))?; Ok(( StatusCode::CREATED, Json(crate::reader::create_article_record(&state, input).await?), ) .into_response()) } _ => Err(invalid("supported formats: .md, .json, .zip and .pdf")), }}
async fn write_owned_directory( state: &AppState, key: &str, files: &Files,) -> Result<PathBuf, ApiError> { if !safe_relative_path(key) { return Err(invalid("unsafe destination path")); } let destination = state.config.storage_dir.join(key); if let Some(parent) = destination.parent() { fs::create_dir_all(parent).await?; } let root = fs::canonicalize(&state.config.storage_dir).await?; if let Some(parent) = destination.parent() { if !fs::canonicalize(parent).await?.starts_with(&root) { return Err(invalid("asset destination is outside storage")); } } fs::create_dir(&destination).await?; let result = async { for (name, bytes) in files { if !safe_relative_path(name) { return Err(invalid("unsafe asset path")); } let path = destination.join(name); if let Some(parent) = path.parent() { fs::create_dir_all(parent).await?; } fs::write(path, bytes).await?; } Ok::<_, ApiError>(()) } .await; if let Err(error) = result { let _ = fs::remove_dir_all(&destination).await; return Err(error); } Ok(destination)}
async fn import_bundle(state: &AppState, files: Files) -> Result<Material, ApiError> { let manifest_name = files .keys() .find(|name| name.as_str() == "lesson.json") .or_else(|| { files .keys() .find(|name| name.ends_with("/lesson.json") && name.split('/').count() == 2) }) .ok_or_else(|| invalid("the archive must contain lesson.json and content.md"))?; let prefix = manifest_name .strip_suffix("lesson.json") .unwrap_or_default(); let manifest: Value = serde_json::from_slice(&files[manifest_name]) .map_err(|e| invalid(format!("invalid lesson.json: {e}")))?; let content_file = manifest["contentFile"].as_str().unwrap_or("content.md"); if !supported_asset(content_file) { return Err(invalid("unsafe lesson content path")); } let content = files .get(&format!("{prefix}{content_file}")) .ok_or_else(|| invalid("the lesson content file is missing"))?; let body = String::from_utf8(content.clone()).map_err(|_| invalid("lesson content must be UTF-8"))?; let title = manifest["title"] .as_str() .ok_or_else(|| invalid("lesson title is required"))?; let subject = manifest["subject"].as_str().unwrap_or_default(); let sources = manifest .get("sources") .cloned() .unwrap_or_else(|| json!([])); bounded_text(title, "title", 200, true)?; bounded_text(subject, "subject", 200, false)?; bounded_text(&body, "content", 4_000_000, false)?; validate_sources(&sources)?; let mut assets = Files::new(); for (name, bytes) in &files { let relative = name .strip_prefix(prefix) .ok_or_else(|| invalid("lesson files must share one root directory"))?; if !supported_asset(relative) { return Err(invalid(format!("unsafe lesson asset: {relative}"))); } if relative != "lesson.json" && relative != content_file && relative != "content.md" { assets.insert(relative.to_owned(), bytes.clone()); } } let id = Uuid::now_v7(); let destination = write_owned_directory(state, &format!("reader/{id}"), &assets).await?; let result=async{ let mut tx=state.platform_db.begin().await?; let material=sqlx::query_as::<_,Material>("insert into reader_materials(id,kind,title,subject,body,sources) values($1,'article',$2,$3,$4,$5) returning *").bind(id).bind(title.trim()).bind(subject).bind(body).bind(sources).fetch_one(&mut *tx).await?; index_article_tx(&mut tx,id,&material.body).await?;let material=sync_catalog_tx(&mut tx,&material).await?;tx.commit().await?;Ok::<_,ApiError>(material) }.await; if result.is_err() { let _ = fs::remove_dir_all(destination).await; } result}
pub(crate) async fn import_pdf( state: &AppState, name: &str, bytes: Vec<u8>,) -> Result<Material, ApiError> { if !bytes[..bytes.len().min(1024)] .windows(5) .any(|s| s == b"%PDF-") { return Err(invalid("this file is not a valid PDF")); } if bytes.len() > MAX_UPLOAD { return Err(invalid("PDF exceeds the 256 MiB upload limit")); } let asset_id = Uuid::now_v7(); let mut files = Files::new(); files.insert("book.pdf".into(), bytes.clone()); let directory = write_owned_directory(state, &asset_id.to_string(), &files).await?; let filename = FsPath::new(name) .file_name() .and_then(|s| s.to_str()) .unwrap_or("book.pdf") .to_owned(); let title = FsPath::new(&filename) .file_stem() .and_then(|s| s.to_str()) .unwrap_or("Book") .chars() .take(200) .collect::<String>(); let checksum = format!("{:x}", Sha256::digest(&bytes)); let result=async{ let mut tx=state.platform_db.begin().await?; sqlx::query("insert into assets(id,kind,filename,mime_type,byte_size,storage_key,checksum_sha256,metadata) values($1,'book',$2,'application/pdf',$3,$4,$5,'{}')").bind(asset_id).bind(&filename).bind(bytes.len() as i64).bind(format!("{asset_id}/book.pdf")).bind(checksum).execute(&mut *tx).await?; let material=sqlx::query_as::<_,Material>("insert into reader_materials(kind,title,filename,status,asset_id) values('book',$1,$2,'indexing',$3) returning *").bind(title).bind(filename).bind(asset_id).fetch_one(&mut *tx).await?; sync_catalog_tx(&mut tx,&material).await?;tx.commit().await?;Ok::<_,ApiError>(material) }.await; if let Ok(material) = &result { enqueue_index(state.clone(), material.id); } else { let _ = fs::remove_dir_all(directory).await; } result}
async fn collect_files(root: &FsPath) -> Result<Files, ApiError> { let mut result = Files::new(); if !fs::try_exists(root).await? { return Ok(result); } let mut pending = vec![(root.to_owned(), String::new())]; while let Some((dir, prefix)) = pending.pop() { let mut entries = fs::read_dir(dir).await?; while let Some(entry) = entries.next_entry().await? { let name = entry.file_name().to_string_lossy().into_owned(); let relative = format!("{prefix}{name}"); let kind = entry.file_type().await?; if kind.is_symlink() { continue; } if kind.is_dir() { pending.push((entry.path(), format!("{relative}/"))); } else if kind.is_file() { if !safe_relative_path(&relative) { return Err(invalid("unsafe stored asset path")); } result.insert(relative, fs::read(entry.path()).await?); } } } Ok(result)}
fn zip_response(bytes: Vec<u8>, name: &str) -> Response { ( [ (header::CONTENT_TYPE, "application/zip"), (header::CONTENT_DISPOSITION, name), ], bytes, ) .into_response()}async fn export_material( State(state): State<AppState>, Path(id): Path<Uuid>, headers: HeaderMap,) -> Result<Response, ApiError> { let material = get_material(&state, id).await?; if material.kind == "book" { let (_, path) = asset_path( &state, material .asset_id .ok_or_else(|| invalid("book PDF is missing"))?, ) .await?; return file_response(path, "application/pdf", &headers, Some("book.pdf")).await; } let mut files = collect_files(&state.config.storage_dir.join("reader").join(id.to_string())).await?; files.insert("lesson.json".into(),serde_json::to_vec_pretty(&json!({"version":1,"title":material.title,"subject":material.subject,"contentFile":"content.md","sources":material.sources})).map_err(anyhow::Error::from)?); files.insert("content.md".into(), material.body.into_bytes()); let bytes = tokio::task::spawn_blocking(move || write_zip(files)) .await .map_err(anyhow::Error::from)??; Ok(zip_response(bytes, "attachment; filename=\"lesson.zip\""))}
fn encode_legacy_json(row: &mut Value, keys: &[&str]) { if let Some(object) = row.as_object_mut() { for key in keys { if let Some(value) = object.get_mut(*key) { *value = Value::String(value.to_string()); } } }}async fn query_rows(tx: &mut Transaction<'_, Postgres>, sql: &str) -> Result<Vec<Value>, ApiError> { Ok(sqlx::query_scalar(sql).fetch_all(&mut **tx).await?)}
async fn backup(State(state): State<AppState>) -> Result<Response, ApiError> { let mut tx = state.platform_db.begin().await?; sqlx::query("set transaction isolation level repeatable read") .execute(&mut *tx) .await?; let mut materials = query_rows(&mut tx, "select to_jsonb(m) from reader_materials m").await?; let chunks = query_rows( &mut tx, "select to_jsonb(c)-'search_vector' from reader_chunks c", ) .await?; let chats=query_rows(&mut tx,"select jsonb_build_object('id',c.id,'material_id',c.context_id,'title',c.title,'provider_id',c.provider,'model',c.model,'created_at',c.created_at,'updated_at',c.updated_at) from agent_conversations c where context_type='reader'").await?; let mut messages=query_rows(&mut tx,"select jsonb_build_object('id',m.id,'chat_id',m.conversation_id,'role',m.role,'content',m.content,'status',coalesce(m.metadata->>'status','complete'),'model',m.metadata->>'model','citations',coalesce(m.metadata->'citations','[]'::jsonb),'context',coalesce(m.metadata->'context','{}'::jsonb),'created_at',m.created_at) from agent_messages m join agent_conversations c on c.id=m.conversation_id where c.context_type='reader'").await?; let mut highlights = query_rows(&mut tx, "select to_jsonb(h) from reader_highlights h").await?; let mut notes = query_rows(&mut tx, "select to_jsonb(n) from reader_notes n").await?; let cards = query_rows(&mut tx, "select to_jsonb(c) from reader_review_cards c").await?; let attempts = query_rows( &mut tx, "select to_jsonb(a) from reader_exercise_attempts a", ) .await?; let revisions = query_rows( &mut tx, "select to_jsonb(r) from reader_material_revisions r", ) .await?; let native_notes=query_rows(&mut tx,"select to_jsonb(n) || jsonb_build_object('tags',coalesce((select jsonb_agg(t.display_name order by t.display_name) from note_tags nt join tags t on t.id=nt.tag_id where nt.note_id=n.id),'[]'::jsonb)) from notes n join reader_materials m on m.note_id=n.id").await?; let native_note_revisions=query_rows(&mut tx,"select to_jsonb(r) from note_revisions r join reader_materials m on m.note_id=r.note_id order by r.note_id,r.revision_number").await?; let native_annotations = query_rows( &mut tx, "select to_jsonb(a) from annotations a join reader_materials m on m.note_id=a.note_id", ) .await?; let providers=query_rows(&mut tx,"select jsonb_build_object('id',provider,'name',coalesce(name,provider),'baseUrl',coalesce(base_url,case when provider='anthropic' then 'https://api.anthropic.com' else 'https://api.openai.com' end),'hasKey',api_key_ciphertext is not null,'models',models,'options',options) from provider_settings").await?; let mut files = Files::new(); for material in &materials { let id = material["id"] .as_str() .and_then(|id| Uuid::parse_str(id).ok()) .ok_or_else(|| invalid("invalid material id"))?; let prefix = format!("files/{id}/"); for (name, bytes) in collect_files(&state.config.storage_dir.join("reader").join(id.to_string())).await? { files.insert(format!("{prefix}{name}"), bytes); } if material["kind"].as_str() == Some("book") { let asset = material["asset_id"] .as_str() .and_then(|id| Uuid::parse_str(id).ok()) .ok_or_else(|| invalid("book asset is missing"))?; let (_, path) = asset_path(&state, asset).await?; files.insert(format!("{prefix}book.pdf"), fs::read(path).await?); } } tx.commit().await?; for material in &mut materials { encode_legacy_json(material, &["sources", "position"]); } for message in &mut messages { encode_legacy_json(message, &["citations", "context"]); } for highlight in &mut highlights { encode_legacy_json(highlight, &["rects"]); } for note in &mut notes { encode_legacy_json(note, &["citations"]); } files.insert("odre-backup.json".into(),serde_json::to_vec(&json!({"version":1,"createdAt":Utc::now(),"materials":materials,"chunks":chunks,"chats":chats,"messages":messages,"highlights":highlights,"notes":notes,"review_cards":cards,"exercise_attempts":attempts,"providers":providers,"reader_material_revisions":revisions,"native_notes":native_notes,"native_note_revisions":native_note_revisions,"native_annotations":native_annotations})).map_err(anyhow::Error::from)?); let bytes = tokio::task::spawn_blocking(move || write_zip(files)) .await .map_err(anyhow::Error::from)??; Ok(zip_response( bytes, "attachment; filename=\"odre-backup.zip\"", ))}
fn field<'a>(row: &'a Value, key: &str) -> Result<&'a str, ApiError> { row[key] .as_str() .ok_or_else(|| invalid(format!("backup field {key} must be text")))}fn text_default<'a>(row: &'a Value, key: &str) -> &'a str { row[key].as_str().unwrap_or_default()}fn source_page(row: &Value) -> Result<Option<i32>, ApiError> { match row.get("page") { None | Some(Value::Null) => Ok(None), Some(value) => { let page = value .as_i64() .ok_or_else(|| invalid("invalid source page"))?; if !(1..=5000).contains(&page) { return Err(invalid("invalid source page")); } Ok(Some(page as i32)) } }}fn legacy_json(row: &Value, key: &str, fallback: Value) -> Result<Value, ApiError> { match row.get(key) { None | Some(Value::Null) => Ok(fallback), Some(Value::String(value)) => { serde_json::from_str(value).map_err(|_| invalid(format!("invalid backup JSON {key}"))) } Some(value) => Ok(value.clone()), }}fn time(row: &Value, key: &str) -> Result<DateTime<Utc>, ApiError> { let value = field(row, key)?; DateTime::parse_from_rfc3339(value) .map(|time| time.with_timezone(&Utc)) .map_err(|_| invalid(format!("invalid backup date {key}")))}fn rows<'a>(backup: &'a Value, key: &str) -> Result<&'a [Value], ApiError> { match backup.get(key) { None => Ok(&[]), Some(Value::Array(rows)) => Ok(rows), _ => Err(invalid(format!("backup {key} must be an array"))), }}fn mapped(map: &HashMap<String, Uuid>, row: &Value, key: &str) -> Result<Uuid, ApiError> { map.get(field(row, key)?) .copied() .ok_or_else(|| invalid(format!("invalid backup reference {key}")))}fn nullable_mapped( map: &HashMap<String, Uuid>, row: &Value, key: &str,) -> Result<Option<Uuid>, ApiError> { match row.get(key) { None | Some(Value::Null) => Ok(None), Some(Value::String(value)) if value.is_empty() => Ok(None), Some(Value::String(value)) => map .get(value) .copied() .map(Some) .ok_or_else(|| invalid(format!("invalid backup reference {key}"))), _ => Err(invalid(format!("invalid backup reference {key}"))), }}fn new_id_map(rows: &[Value]) -> Result<HashMap<String, Uuid>, ApiError> { let mut map = HashMap::new(); for row in rows { let id = field(row, "id")?; if id.is_empty() || map.insert(id.to_owned(), Uuid::now_v7()).is_some() { return Err(invalid("duplicate or empty backup ID")); } } Ok(map)}fn map_citations(value: &mut Value, materials: &HashMap<String, Uuid>) -> Result<(), ApiError> { let array = value .as_array_mut() .ok_or_else(|| invalid("citations must be an array"))?; for citation in array { if let Some(old) = citation .get("materialId") .and_then(Value::as_str) .map(ToOwned::to_owned) { if let Some(new) = materials.get(&old) { citation["materialId"] = json!(new); } else if let Some(object) = citation.as_object_mut() { object.remove("materialId"); object.insert("unavailableMaterialId".into(), json!(old)); } } } Ok(())}
fn remap_context(context: &mut Value, materials: &HashMap<String, Uuid>) -> Result<(), ApiError> { let object = context .as_object_mut() .ok_or_else(|| invalid("invalid message context"))?; if let Some(mut comparison) = object.remove("comparison").filter(|value| !value.is_null()) { let old = comparison["materialId"] .as_str() .ok_or_else(|| invalid("invalid comparison material"))?; if let Some(id) = materials.get(old) { comparison["materialId"] = json!(id); object.insert("comparison".into(), comparison); } else { object.insert("unavailableComparison".into(), comparison); } } Ok(())}
fn remap_native_document( value: &mut Value, materials: &HashMap<String, Uuid>, annotations: &HashMap<String, Uuid>,) { match value { Value::Object(object) => { if let Some(old) = object.get("annotationId").and_then(Value::as_str) { if let Some(id) = annotations.get(old) { object.insert("annotationId".into(), json!(id)); } } for value in object.values_mut() { remap_native_document(value, materials, annotations); } } Value::Array(values) => { for value in values { remap_native_document(value, materials, annotations); } } Value::String(text) => { if let Some(rest) = text.strip_prefix("/api/reader/files/") { if let Some((old, path)) = rest.split_once('/') { if let Some(id) = materials.get(old) { *text = format!("/api/reader/files/{id}/{path}"); } } } } _ => {} }}
async fn restore_native_backup_tx( tx: &mut Transaction<'_, Postgres>, backup: &Value, materials: &HashMap<String, Uuid>,) -> Result<(), ApiError> { let native = rows(backup, "native_notes")?; if native.is_empty() { return Ok(()); } let annotation_ids = new_id_map(rows(backup, "native_annotations")?)?; let revisions = rows(backup, "native_note_revisions")?; let revision_ids = new_id_map(revisions)?; let mut note_ids = HashMap::new(); for row in native { let old = field(row, "id")?; let source = rows(backup, "materials")? .iter() .find(|m| m["note_id"].as_str() == Some(old)) .ok_or_else(|| invalid("native Note has no linked material"))?; let material_id = mapped(materials, source, "id")?; let material = sqlx::query_as::<_, Material>("select * from reader_materials where id=$1") .bind(material_id) .fetch_one(&mut **tx) .await?; let note_id = material .note_id .ok_or_else(|| invalid("native Note must belong to an article"))?; if note_ids.insert(old.to_owned(), note_id).is_some() { return Err(invalid("duplicate native Note ID")); } let mut document = legacy_json(row, "document", json!({}))?; remap_native_document(&mut document, materials, &annotation_ids); let body = crate::reader_native::document_markdown(&document, material_id)?; let kind = field(row, "kind")?; if !matches!(kind, "card" | "post") { return Err(invalid("invalid native Note kind")); } let version = row["version"] .as_i64() .filter(|v| *v >= 1) .ok_or_else(|| invalid("invalid native Note version"))?; if version != material.version || row["title"].as_str() != Some(&material.title) { return Err(invalid("native Note and article disagree in backup")); } sqlx::query("update notes set title=$2,kind=$3,excerpt=$4,document=$5,version=$6,created_at=$7,updated_at=$8 where id=$1") .bind(note_id).bind(&material.title).bind(kind).bind(text_default(row,"excerpt")).bind(&document).bind(version).bind(time(row,"created_at")?).bind(time(row,"updated_at")?).execute(&mut **tx).await?; let tags = row .get("tags") .and_then(Value::as_array) .map(|values| { values .iter() .map(|v| { v.as_str() .map(str::to_owned) .ok_or_else(|| invalid("invalid native Note tag")) }) .collect::<Result<Vec<_>, _>>() }) .transpose()? .unwrap_or_default(); crate::notes::replace_tags(tx, note_id, tags).await?; sqlx::query("delete from note_revisions where note_id=$1") .bind(note_id) .execute(&mut **tx) .await?; if crate::reader_native::document_markdown( &crate::reader_native::native_document(&material.body, material_id), material_id, )? != body { sqlx::query("update reader_materials set body=$2 where id=$1") .bind(material_id) .bind(&body) .execute(&mut **tx) .await?; index_article_tx(tx, material_id, &body).await?; } } // One statement permits parent IDs from later rows without introducing a // dependency on the ordering chosen by the archive producer. let mut imported = Vec::new(); for row in revisions { let mut document = legacy_json(row, "document", json!({}))?; remap_native_document(&mut document, materials, &annotation_ids); imported.push(json!({"id":mapped(&revision_ids,row,"id")?,"note_id":mapped(¬e_ids,row,"note_id")?,"parent_revision_id":nullable_mapped(&revision_ids,row,"parent_revision_id")?,"revision_number":row["revision_number"],"title":field(row,"title")?,"excerpt":text_default(row,"excerpt"),"document":document,"reason":text_default(row,"reason"),"created_at":time(row,"created_at")?})); } if !imported.is_empty() { sqlx::query("insert into note_revisions select * from jsonb_populate_recordset(null::note_revisions,$1)").bind(json!(imported)).execute(&mut **tx).await?; } for note_id in note_ids.values() { sqlx::query("insert into note_revisions(note_id,revision_number,title,excerpt,document,reason) select id,version,title,excerpt,document,'restored native backup' from notes where id=$1 on conflict(note_id,revision_number) do nothing").bind(note_id).execute(&mut **tx).await?; } for row in rows(backup, "native_annotations")? { sqlx::query("insert into annotations(id,note_id,block_id,anchor,messages,created_at,updated_at) values($1,$2,$3,$4,$5,$6,$7)") .bind(mapped(&annotation_ids,row,"id")?).bind(mapped(¬e_ids,row,"note_id")?).bind(field(row,"block_id")?).bind(row.get("anchor")).bind(legacy_json(row,"messages",json!([]))?).bind(time(row,"created_at")?).bind(time(row,"updated_at")?).execute(&mut **tx).await?; } Ok(())}
async fn restore_backup(state: &AppState, files: Files) -> Result<Value, ApiError> { let mut digest = Sha256::new(); for (path, bytes) in &files { digest.update((path.len() as u64).to_le_bytes()); digest.update(path.as_bytes()); digest.update((bytes.len() as u64).to_le_bytes()); digest.update(bytes); } let fingerprint = format!("{:x}", digest.finalize()); let backup: Value = serde_json::from_slice( files .get("odre-backup.json") .ok_or_else(|| invalid("backup manifest missing"))?, ) .map_err(|e| invalid(format!("invalid backup manifest: {e}")))?; if backup["version"].as_i64() != Some(1) || !backup["materials"].is_array() || !backup["messages"].is_array() { return Err(invalid("this is not a supported Odre backup")); } let mut tx = state.platform_db.begin().await?; sqlx::query("select pg_advisory_xact_lock(739271580119::bigint)") .execute(&mut *tx) .await?; if let Some(result) = sqlx::query_scalar::<_, Value>( "select result from reader_legacy_imports where fingerprint=$1", ) .bind(&fingerprint) .fetch_optional(&mut *tx) .await? { return Ok(result); } let material_rows = rows(&backup, "materials")?; let material_ids = new_id_map(material_rows)?; let chat_ids = new_id_map(rows(&backup, "chats")?)?; let message_ids = new_id_map(rows(&backup, "messages")?)?; let note_ids = new_id_map(rows(&backup, "notes")?)?; let mut directory_paths = Vec::new(); let mut book_assets = HashMap::new(); let result=async{ for material in material_rows{ let old=field(material,"id")?;if !old.chars().all(|c|c.is_ascii_alphanumeric()||c=='-'){return Err(invalid("invalid backup material path"));} let id=material_ids[old];let prefix=format!("files/{old}/");let mut assets=Files::new(); for (name,bytes) in &files{if let Some(relative)=name.strip_prefix(&prefix){if !safe_relative_path(relative){return Err(invalid("invalid backup asset path"));}assets.insert(relative.to_owned(),bytes.clone());}} if field(material,"kind")?=="book"{ let bytes=assets.remove("book.pdf").ok_or_else(||invalid("the backup is missing a book PDF"))?; if !bytes[..bytes.len().min(1024)].windows(5).any(|s|s==b"%PDF-"){return Err(invalid("backup contains an invalid PDF"));} let asset=Uuid::now_v7();let mut owned=Files::new();owned.insert("book.pdf".into(),bytes.clone());let path=write_owned_directory(state,&asset.to_string(),&owned).await?;directory_paths.push(path);book_assets.insert(id,(asset,bytes)); } if !assets.is_empty(){directory_paths.push(write_owned_directory(state,&format!("reader/{id}"),&assets).await?);} } for name in files.keys(){if name=="odre-backup.json"{continue;}let Some(rest)=name.strip_prefix("files/")else{return Err(invalid("invalid backup asset path"));};let Some((id,path))=rest.split_once('/')else{return Err(invalid("invalid backup asset path"));};if !material_ids.contains_key(id)||!safe_relative_path(path){return Err(invalid("backup asset refers to an unknown material"));}} let mut material_records=HashMap::new(); let mut provider_ids=HashMap::new(); for provider in rows(&backup,"providers")?{ let old=field(provider,"id")?;let name=field(provider,"name")?;bounded_text(name,"provider name",100,true)?;let base=field(provider,"baseUrl")?; let url=reqwest::Url::parse(base).map_err(|_|invalid("invalid provider URL"))?;if !matches!(url.scheme(),"http"|"https"){return Err(invalid("invalid provider URL"));} let models=legacy_json(provider,"models",json!([]))?;let model_array=models.as_array().ok_or_else(||invalid("provider models must be an array"))?;if model_array.len()>100||model_array.iter().any(|model|model.as_str().is_none()){return Err(invalid("invalid provider models"));} let mut options=legacy_json(provider,"options",json!({}))?;let object=options.as_object_mut().ok_or_else(||invalid("provider options must be an object"))?;object.entry("protocol").or_insert(json!("chat")); let new=format!("odre-import-{}",Uuid::now_v7());let model=model_array.first().and_then(Value::as_str).unwrap_or_default(); sqlx::query("insert into provider_settings(provider,model,base_url,enabled,name,models,options) values($1,$2,$3,false,$4,$5,$6)").bind(&new).bind(model).bind(base.trim_end_matches('/')).bind(name).bind(&models).bind(options).execute(&mut *tx).await?; if provider_ids.insert(old.to_owned(),new).is_some(){return Err(invalid("duplicate backup provider ID"));} } for row in material_rows{ let id=mapped(&material_ids,row,"id")?;let kind=field(row,"kind")?;if !matches!(kind,"article"|"book"){return Err(invalid("invalid material kind"));}let title=field(row,"title")?;bounded_text(title,"title",200,true)?; let subject=text_default(row,"subject");bounded_text(subject,"subject",200,false)?;let body=text_default(row,"body");bounded_text(body,"content",4_000_000,false)?; let sources=legacy_json(row,"sources",json!([]))?;validate_sources(&sources)?;let position=legacy_json(row,"position",json!({}))?;if !position.is_object(){return Err(invalid("invalid material position"));} let page_count=row["page_count"].as_i64().unwrap_or(0);if !(0..=5000).contains(&page_count){return Err(invalid("invalid PDF page count"));} let status=row["status"].as_str().unwrap_or("ready");if !matches!(status,"ready"|"indexing"|"image-only"|"error"){return Err(invalid("invalid indexing status"));}let progress=row["progress"].as_i64().unwrap_or(0);if progress<0||progress>5000{return Err(invalid("invalid indexing progress"));} let asset=book_assets.get(&id).map(|(id,_)|*id); if let Some((asset,bytes))=book_assets.get(&id){sqlx::query("insert into assets(id,kind,filename,mime_type,byte_size,storage_key,checksum_sha256,metadata) values($1,'book',$2,'application/pdf',$3,$4,$5,'{}')").bind(asset).bind(row["filename"].as_str().unwrap_or("book.pdf")).bind(bytes.len() as i64).bind(format!("{asset}/book.pdf")).bind(format!("{:x}",Sha256::digest(bytes))).execute(&mut *tx).await?;} let material=sqlx::query_as::<_,Material>("insert into reader_materials(id,kind,title,subject,body,sources,filename,page_count,status,progress,error,position,version,asset_id,created_at,updated_at) values($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16) returning *") .bind(id).bind(kind).bind(title).bind(subject).bind(body).bind(sources).bind(row["filename"].as_str()).bind(page_count as i32).bind(status).bind(progress as i32).bind(row["error"].as_str()).bind(position).bind(row["version"].as_i64().unwrap_or(1).max(1)).bind(asset).bind(time(row,"created_at")?).bind(time(row,"updated_at")?).fetch_one(&mut *tx).await?; crate::reader::validate_position(&material.position,&material)?; let material=sync_catalog_tx(&mut tx,&material).await?; material_records.insert(id,material); } for row in rows(&backup,"chats")?{ let id=mapped(&chat_ids,row,"id")?;let material=mapped(&material_ids,row,"material_id")?; let provider=if let Some(old)=row["provider_id"].as_str().filter(|id|!id.is_empty()) { if let Some(existing)=provider_ids.get(old){existing.clone()}else{ let new=format!("odre-import-{}",Uuid::now_v7()); let model=text_default(row,"model");let models=if model.is_empty(){json!([])}else{json!([model])}; sqlx::query("insert into provider_settings(provider,model,enabled,name,models,options) values($1,$2,false,'Imported provider (unavailable)',$3,$4)").bind(&new).bind(model).bind(models).bind(json!({"protocol":"chat","legacyProviderId":old,"unavailable":true})).execute(&mut *tx).await?; provider_ids.insert(old.to_owned(),new.clone());new } }else{"openai".to_owned()}; bounded_text(field(row,"title")?,"chat title",160,true)?; sqlx::query("insert into agent_conversations(id,title,context_type,context_id,provider,model,created_at,updated_at) values($1,$2,'reader',$3,$4,$5,$6,$7)").bind(id).bind(field(row,"title")?).bind(material).bind(provider).bind(text_default(row,"model")).bind(time(row,"created_at")?).bind(time(row,"updated_at")?).execute(&mut *tx).await?; } for row in rows(&backup,"messages")?{ let id=mapped(&message_ids,row,"id")?;let chat=mapped(&chat_ids,row,"chat_id")?;let role=field(row,"role")?;if !matches!(role,"user"|"assistant"){return Err(invalid("invalid backup message role"));} let mut context=legacy_json(row,"context",json!({}))?;remap_context(&mut context,&material_ids)?; let mut citations=legacy_json(row,"citations",json!([]))?;map_citations(&mut citations,&material_ids)?; let status=row["status"].as_str().unwrap_or("complete");let status=if status=="streaming"{"interrupted"}else{status};if !matches!(status,"complete"|"interrupted"|"error"|"cancelled"){return Err(invalid("invalid message status"));} let metadata=json!({"status":status,"model":row["model"],"citations":citations,"context":context}); sqlx::query("insert into agent_messages(id,conversation_id,role,content,metadata,created_at) values($1,$2,$3,$4,$5,$6)").bind(id).bind(chat).bind(role).bind(field(row,"content")?).bind(metadata).bind(time(row,"created_at")?).execute(&mut *tx).await?; } for row in rows(&backup,"chunks")?{let material=mapped(&material_ids,row,"material_id")?;let page=source_page(row)?;crate::reader_learning::validate_location(&material_records[&material],"",page)?;sqlx::query("insert into reader_chunks(material_id,page,heading,text) values($1,$2,$3,$4)").bind(material).bind(page).bind(text_default(row,"heading")).bind(field(row,"text")?).execute(&mut *tx).await?;} for row in material_rows{let id=mapped(&material_ids,row,"id")?;let count:i64=sqlx::query_scalar("select count(*) from reader_chunks where material_id=$1").bind(id).fetch_one(&mut *tx).await?;if count==0&&row["kind"].as_str()==Some("article"){index_article_tx(&mut tx,id,text_default(row,"body")).await?;}} for row in rows(&backup,"highlights")?{ let material=mapped(&material_ids,row,"material_id")?;let rects=legacy_json(row,"rects",json!([]))?; let source=sqlx::query_as::<_,Material>("select * from reader_materials where id=$1").bind(material).fetch_one(&mut *tx).await?; crate::reader::validate_highlight(&json!({"quote":row["quote"],"prefix":row.get("prefix").cloned().unwrap_or(json!("")),"suffix":row.get("suffix").cloned().unwrap_or(json!("")),"page":row["page"],"rects":rects,"note":row.get("note").cloned().unwrap_or(json!(""))}),&source)?; sqlx::query("insert into reader_highlights(material_id,quote,prefix,suffix,page,rects,note,content_version,created_at) values($1,$2,$3,$4,$5,$6,$7,$8,$9)").bind(material).bind(field(row,"quote")?).bind(text_default(row,"prefix")).bind(text_default(row,"suffix")).bind(row["page"].as_i64().map(|n|n as i32)).bind(rects).bind(text_default(row,"note")).bind(row["content_version"].as_i64().unwrap_or(1)).bind(time(row,"created_at")?).execute(&mut *tx).await?; } let mut note_materials=HashMap::new(); for row in rows(&backup,"notes")?{ let id=mapped(¬e_ids,row,"id")?;let material=mapped(&material_ids,row,"material_id")?;let chat=nullable_mapped(&chat_ids,row,"chat_id")?;let message=nullable_mapped(&message_ids,row,"message_id")?; bounded_text(field(row,"title")?,"note title",200,true)?;bounded_text(field(row,"body")?,"note body",100000,true)?;bounded_text(text_default(row,"quote"),"quote",12000,false)?; crate::reader_learning::validate_location(&material_records[&material],text_default(row,"quote"),source_page(row)?)?; if let Some(chat)=chat{let same:bool=sqlx::query_scalar("select exists(select 1 from agent_conversations where id=$1 and context_id=$2 and context_type='reader')").bind(chat).bind(material).fetch_one(&mut *tx).await?;if !same{return Err(invalid("note conversation does not belong to its material"));}} if let Some(message)=message{let same:bool=sqlx::query_scalar("select exists(select 1 from agent_messages where id=$1 and conversation_id=$2 and role='assistant' and coalesce(metadata->>'status','complete')='complete')").bind(message).bind(chat).fetch_one(&mut *tx).await?;if !same{return Err(invalid("note message does not belong to a completed assistant answer"));}} let mut citations=legacy_json(row,"citations",json!([]))?;map_citations(&mut citations,&material_ids)?; sqlx::query("insert into reader_notes(id,material_id,title,body,quote,page,chat_id,message_id,citations,created_at,updated_at) values($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)").bind(id).bind(material).bind(field(row,"title")?).bind(field(row,"body")?).bind(text_default(row,"quote")).bind(row["page"].as_i64().map(|n|n as i32)).bind(chat).bind(message).bind(citations).bind(time(row,"created_at")?).bind(time(row,"updated_at")?).execute(&mut *tx).await?; note_materials.insert(id,material); } for row in rows(&backup,"review_cards")?{ let material=mapped(&material_ids,row,"material_id")?;let note=nullable_mapped(¬e_ids,row,"note_id")?;if note.is_some_and(|id|note_materials.get(&id)!=Some(&material)){return Err(invalid("card source note does not belong to its material"));} bounded_text(field(row,"question")?,"question",12000,true)?;bounded_text(field(row,"answer")?,"answer",20000,true)?;bounded_text(text_default(row,"quote"),"quote",12000,false)?; crate::reader_learning::validate_location(&material_records[&material],text_default(row,"quote"),source_page(row)?)?; let interval=row["interval_days"].as_f64().ok_or_else(||invalid("invalid card interval"))?;let repetitions=row["repetitions"].as_i64().ok_or_else(||invalid("invalid card repetitions"))?;if !interval.is_finite()||interval<0.0||!(0..=i32::MAX as i64).contains(&repetitions){return Err(invalid("invalid review schedule"));} sqlx::query("insert into reader_review_cards(material_id,question,answer,quote,page,note_id,due_at,interval_days,repetitions,created_at,updated_at) values($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)").bind(material).bind(field(row,"question")?).bind(field(row,"answer")?).bind(text_default(row,"quote")).bind(row["page"].as_i64().map(|n|n as i32)).bind(note).bind(time(row,"due_at")?).bind(interval).bind(repetitions as i32).bind(time(row,"created_at")?).bind(time(row,"updated_at")?).execute(&mut *tx).await?; } for row in rows(&backup,"exercise_attempts")?{let material=mapped(&material_ids,row,"material_id")?;bounded_text(field(row,"exercise_key")?,"exercise key",12000,true)?;bounded_text(field(row,"answer")?,"answer",20000,false)?;sqlx::query("insert into reader_exercise_attempts(material_id,exercise_key,answer,updated_at) values($1,$2,$3,$4)").bind(material).bind(field(row,"exercise_key")?).bind(field(row,"answer")?).bind(time(row,"updated_at")?).execute(&mut *tx).await?;} for row in rows(&backup,"reader_material_revisions")?{let material=mapped(&material_ids,row,"material_id")?;let sources=legacy_json(row,"sources",json!([]))?;validate_sources(&sources)?;sqlx::query("insert into reader_material_revisions(material_id,version,title,subject,body,sources,reason,created_at) values($1,$2,$3,$4,$5,$6,$7,$8)").bind(material).bind(row["version"].as_i64().unwrap_or(1)).bind(field(row,"title")?).bind(text_default(row,"subject")).bind(field(row,"body")?).bind(sources).bind(text_default(row,"reason")).bind(time(row,"created_at")?).execute(&mut *tx).await?;} restore_native_backup_tx(&mut tx,&backup,&material_ids).await?; let result=json!({"restored":material_ids.len()}); sqlx::query("insert into reader_legacy_imports(fingerprint,result) values($1,$2)").bind(&fingerprint).bind(&result).execute(&mut *tx).await?; tx.commit().await?; Ok::<_,ApiError>(result) }.await; if result.is_err() { for path in directory_paths { let _ = fs::remove_dir_all(path).await; } } else { for row in material_rows { if row["kind"].as_str() == Some("book") && row["status"].as_str() == Some("indexing") { enqueue_index(state.clone(), mapped(&material_ids, row, "id")?); } } } result}
#[cfg(test)]mod tests { use super::*; #[test] fn archive_traversal_and_expansion_are_rejected() { let mut writer = ZipWriter::new(Cursor::new(Vec::new())); writer .start_file("../secret", SimpleFileOptions::default()) .unwrap(); writer.write_all(b"x").unwrap(); let bytes = writer.finish().unwrap().into_inner(); assert!(read_zip(bytes).is_err()); let mut writer = ZipWriter::new(Cursor::new(Vec::new())); writer .start_file("content.md", SimpleFileOptions::default()) .unwrap(); writer.write_all(b"small").unwrap(); let mut advertised_large = writer.finish().unwrap().into_inner(); let central = advertised_large .windows(4) .position(|bytes| bytes == b"PK\x01\x02") .unwrap(); advertised_large[central + 24..central + 28].copy_from_slice(&33_000_000u32.to_le_bytes()); assert!(read_zip(advertised_large).is_err()); let files = BTreeMap::from([ ("assets/figure.svg".into(), b"<svg/>".to_vec()), ("content.md".into(), b"lesson".to_vec()), ]); let encoded = write_zip(files.clone()).unwrap(); assert_eq!(read_zip(encoded).unwrap(), files); } #[test] fn citations_keep_identity_when_material_ids_change() { let old = "original"; let id = Uuid::new_v4(); let ids = HashMap::from([(old.into(), id)]); let mut citations = json!([{"source":2,"url":"https://example.org/original","materialId":old,"comparison":true}]); map_citations(&mut citations, &ids).unwrap(); assert_eq!(citations[0]["materialId"], json!(id)); assert_eq!(citations[0]["source"], 2); let mut unavailable = json!([{"materialId":"missing","excerpt":"Retained source text"}]); map_citations(&mut unavailable, &ids).unwrap(); assert_eq!(unavailable[0]["unavailableMaterialId"], "missing"); assert_eq!(unavailable[0]["excerpt"], "Retained source text"); assert!(unavailable[0].get("materialId").is_none()); let mut context = json!({"comparison":{"materialId":"missing","page":4}}); remap_context(&mut context, &ids).unwrap(); assert!(context.get("comparison").is_none()); assert_eq!(context["unavailableComparison"]["materialId"], "missing"); }}