Something went wrong. Try again.
My personal-knowledge-system, with deeply integrated task tracking and long term goal planning capabilities.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386//! HTTP client for the `fil serve` daemon, shared by every surface that//! talks to it (the Elm TUI, the LSP, the CLI). The records it exchanges are//! `fil_api`'s.
use std::{ path::{Path, PathBuf}, sync::Arc, time::Duration,};
use color_eyre::eyre::{Result, eyre};use entity::{DateTime, NanoId};use eventsource_stream::Eventsource as _;pub use fil_api::Snapshot;use fil_api::{ChangeSet, GroupDto, TagDto, TaskDto, ZettelDto};use fil_core::{Priority, ZettelId};use futures::{Stream, StreamExt as _};use serde_json::json;use tokio::io::AsyncWriteExt as _;
/// Header `GET /api/export` reports the archive's file count in; kept in/// step with `server::export::FILES_HEADER`.const EXPORT_FILES_HEADER: &str = "x-fil-export-files";
/// How long to wait for the daemon to accept a connection.const CONNECT_TIMEOUT: Duration = Duration::from_secs(5);/// How long any one read may stall; comfortably above the event stream's/// 15s keep-alive.const READ_TIMEOUT: Duration = Duration::from_secs(45);
/// An inspector edit to send to the daemon. `is_group` picks between the/// task and group routes for fields both have.#[derive(Debug)]pub enum TodoUpdate { Name { id: NanoId, is_group: bool, name: String, }, Priority { id: NanoId, is_group: bool, priority: Priority, }, /// `None` clears the due date. Tasks only. Due { id: NanoId, due: Option<DateTime>, }, RandomColor { group_id: NanoId, },}
/// Something to delete through the daemon.#[derive(Debug, Clone, PartialEq, Eq)]pub enum Deletion { /// A zettel that holds no task's or group's notes. Zettel(ZettelId), /// A task, with its zettel. Task(NanoId), /// A group, with everything beneath it. Group(NanoId),}
/// One event from `GET /api/events`.#[derive(Debug)]pub enum StreamEvent { /// The daemon published changes. Changes(ChangeSet), /// The daemon dropped changes this subscriber was too slow to receive; /// fetch a fresh snapshot. Resync,}
/// HTTP client for the `fil serve` daemon at `url`, with one method per/// endpoint. Cheap to clone (both fields are `Arc`-backed), so each request/// can run on its own task.#[derive(Debug, Clone)]pub struct Client { http: reqwest::Client, url: Arc<str>,}
impl Client { pub fn new(url: &str) -> Self { // The event stream carries a keep-alive comment every 15s, so a read // that stalls well past that means the daemon is gone without having // closed the connection (sleep, network loss); give up on it so the // stream reconnects. let http = reqwest::Client::builder() .connect_timeout(CONNECT_TIMEOUT) .read_timeout(READ_TIMEOUT) .build() .unwrap_or_default(); Self { http, url: url.trim_end_matches('/').into(), } }
pub async fn fetch_snapshot(&self) -> Result<Snapshot> { self.get_json("/api/snapshot").await }
/// Every zettel's metadata; bodies come from [`Client::fetch_body`]. pub async fn fetch_zettels(&self) -> Result<Vec<ZettelDto>> { self.get_json("/api/zettels").await }
/// The zettel's body: its file without the front matter. pub async fn fetch_body(&self, id: &ZettelId) -> Result<String> { Ok(self .http .get(format!("{}/api/zettels/{id}/body", self.url)) .send() .await? .checked() .await? .text() .await?) }
/// Ids of the zettels matching `query`, best match first. The daemon /// ranks them, since it has every body indexed. pub async fn search(&self, query: &str) -> Result<Vec<ZettelId>> { Ok(self .http .get(format!("{}/api/zettels/search", self.url)) .query(&[("q", query)]) .send() .await? .checked() .await? .json() .await?) }
/// The zettel with `id`, including where its file lives. pub async fn fetch_zettel(&self, id: &ZettelId) -> Result<ZettelDto> { self.get_json(&format!("/api/zettels/{id}")).await }
pub async fn fetch_tags(&self) -> Result<Vec<TagDto>> { self.get_json("/api/tags").await }
/// Opens `GET /api/events`. Resolves once the stream is open, then yields /// an item per event until the daemon closes it. Events that don't parse /// are skipped: the revision gap they leave triggers a resync. pub async fn events(&self) -> Result<impl Stream<Item = Result<StreamEvent>>> { let response = self .http .get(format!("{}/api/events", self.url)) .send() .await? .checked() .await?;
Ok(response .bytes_stream() .eventsource() .filter_map(|event| async move { let event = match event { Ok(event) => event, Err(error) => return Some(Err(eyre!("event stream: {error}"))), }; match event.event.as_str() { "changes" => serde_json::from_str(&event.data) .ok() .map(|set| Ok(StreamEvent::Changes(set))), "resync" => Some(Ok(StreamEvent::Resync)), _ => None, } })) } /// Downloads an archive of the whole vault (see `fil import`) to `dest`, /// via `dest.partial` so a failed download never leaves a truncated /// archive behind. Returns how many files it holds. pub async fn export(&self, dest: &Path) -> Result<usize> { let response = self .http .get(format!("{}/api/export", self.url)) .send() .await? .checked() .await?; let files = response .headers() .get(EXPORT_FILES_HEADER) .and_then(|count| count.to_str().ok()?.parse().ok()) .unwrap_or_default();
let mut partial = dest.as_os_str().to_owned(); partial.push(".partial"); let partial = PathBuf::from(partial);
let written = async { let mut file = tokio::fs::File::create(&partial).await?; let mut chunks = response.bytes_stream(); while let Some(chunk) = chunks.next().await { file.write_all(&chunk?).await?; } file.flush().await?; tokio::fs::rename(&partial, dest).await?; Ok::<_, color_eyre::Report>(()) } .await; if written.is_err() { let _ = tokio::fs::remove_file(&partial).await; } written.map(|()| files) }
pub async fn fetch_file(&self, id: &ZettelId) -> Result<String> { let contents = self .http .get(format!("{}/api/zettels/{id}/file", self.url)) .send() .await? .checked() .await? .text() .await?; Ok(contents) }
pub async fn save_file(&self, id: &ZettelId, contents: String) -> Result<ZettelDto> { let zettel = self .http .put(format!("{}/api/zettels/{id}/file", self.url)) .body(contents) .send() .await? .checked() .await? .json::<ZettelDto>() .await?; Ok(zettel) }
pub async fn create_zettel(&self, title: String) -> Result<ZettelDto> { let zettel = self .http .post(format!("{}/api/zettels", self.url)) .json(&json!({ "title": title, "tags": [] })) .send() .await? .checked() .await? .json::<ZettelDto>() .await?; Ok(zettel) }
pub async fn create_task(&self, name: String, group_id: NanoId) -> Result<TaskDto> { let task = self .http .post(format!("{}/api/todo/task", self.url)) .json(&json!({ "name": name, "parent_id": group_id, "due": null, "priority": Priority::default(), })) .send() .await? .checked() .await? .json::<TaskDto>() .await?; Ok(task) }
pub async fn create_group(&self, name: String, parent_id: Option<NanoId>) -> Result<GroupDto> { let group = self .http .post(format!("{}/api/todo/group", self.url)) .json(&json!({ "name": name, "parent_id": parent_id })) .send() .await? .checked() .await? .json::<GroupDto>() .await?; Ok(group) }
pub async fn update_todo(&self, update: &TodoUpdate) -> Result<()> { let kind = |is_group| if is_group { "group" } else { "task" }; let request = match update { TodoUpdate::Name { id, is_group, name } => self .http .put(format!( "{}/api/todo/{}/{id}/name", self.url, kind(*is_group) )) .json(&json!({ "name": name })), TodoUpdate::Priority { id, is_group, priority, } => self .http .put(format!( "{}/api/todo/{}/{id}/priority", self.url, kind(*is_group) )) .json(&json!({ "priority": priority })), TodoUpdate::Due { id, due } => self .http .put(format!("{}/api/todo/task/{id}/due", self.url)) .json(&json!({ "due": due })), TodoUpdate::RandomColor { group_id } => self.http.post(format!( "{}/api/todo/group/{group_id}/randomize-color", self.url )), }; request.send().await?.checked().await?; Ok(()) }
pub async fn delete(&self, deletion: &Deletion) -> Result<()> { let path = match deletion { Deletion::Zettel(id) => format!("/api/zettels/{id}"), Deletion::Task(id) => format!("/api/todo/task/{id}"), Deletion::Group(id) => format!("/api/todo/group/{id}"), }; self.http .delete(format!("{}{path}", self.url)) .send() .await? .checked() .await?; Ok(()) }
pub async fn toggle_task(&self, id: &NanoId) -> Result<()> { self.http .post(format!("{}/api/todo/task/{id}/toggle", self.url)) .send() .await? .checked() .await?; Ok(()) }}
impl Client { async fn get_json<T: serde::de::DeserializeOwned>(&self, path: &str) -> Result<T> { Ok(self .http .get(format!("{}{path}", self.url)) .send() .await? .checked() .await? .json::<T>() .await?) }}
trait Checked { /// Like `error_for_status`, but keeps the daemon's explanation (the /// response body), e.g. `404 Not Found: no group with id …`. async fn checked(self) -> Result<reqwest::Response>;}
impl Checked for reqwest::Response { async fn checked(self) -> Result<Self> { let status = self.status(); if !status.is_client_error() && !status.is_server_error() { return Ok(self); } let body = self.text().await.unwrap_or_default(); let body = body.trim(); if body.is_empty() { Err(eyre!("daemon replied {status}")) } else { Err(eyre!("daemon replied {status}: {body}")) } }}