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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110//! Pushing changes to clients: every mutation publishes a [`ChangeSet`] of//! the records it touched, broadcast over `GET /api/events` (SSE).//! `GET /api/snapshot` gives clients a consistent starting point, stamped//! with the revision the events continue from.
use std::{convert::Infallible, sync::atomic::Ordering};
use axum::{ Json, extract::State, response::{ IntoResponse, sse::{Event, KeepAlive, Sse}, },};use entity::NanoId;use fil_api::{Change, ChangeSet};use fil_core::{Kasten, ZettelId};use tokio_stream::{Stream, StreamExt as _, wrappers::BroadcastStream};use tracing::{debug, instrument};
use super::{FilState, error::ApiError};
/// How often an idle event stream sends a keep-alive comment.const KEEP_ALIVE: std::time::Duration = std::time::Duration::from_secs(15);
impl FilState { /// Broadcasts `changes` to every subscribed client under the next /// revision. /// /// Takes the `Kasten` only to prove the write lock is still held: /// publishing under it keeps revisions in mutation order and lets /// snapshots read a revision that matches their data. pub fn publish(&self, _locked: &mut Kasten, changes: Vec<Change>) { let rev = self.rev.fetch_add(1, Ordering::SeqCst) + 1; debug!(rev, count = changes.len(), "publishing changes"); // no subscribers is fine: nobody needs telling let _ = self.diff_tx.send(ChangeSet { epoch: self.epoch, rev, changes, }); }}
/// `GET /api/snapshot`: every record, plus the revision they reflect, read/// under one lock so they are consistent.#[axum::debug_handler]#[instrument(skip(state))]pub async fn snapshot(State(state): State<FilState>) -> Result<impl IntoResponse, ApiError> { let kt = state.kh.read().await; let rev = state.rev.load(Ordering::SeqCst); let snapshot = fil_api::snapshot(&kt, state.epoch, rev).await?; drop(kt); Ok(Json(snapshot))}
/// `GET /api/events`: an SSE stream of `changes` events, each a/// [`ChangeSet`] as JSON. If this subscriber falls too far behind it gets a/// `resync` event instead, telling it to fetch a fresh snapshot.#[instrument(skip(state))]pub async fn events( State(state): State<FilState>,) -> Sse<impl Stream<Item = Result<Event, Infallible>>> { let stream = BroadcastStream::new(state.diff_tx.subscribe()).map(|set| { let event = set .ok() .and_then(|set| Event::default().event("changes").json_data(set).ok()); Ok(event.unwrap_or_else(|| Event::default().event("resync").data(""))) }); // clients time a stalled stream out (see `client::READ_TIMEOUT`), so keep // this well under that Sse::new(stream).keep_alive(KeepAlive::new().interval(KEEP_ALIVE))}
/// The current state of task `id`, as a change to publish.pub async fn task_change(id: &NanoId, kt: &Kasten) -> Result<Change, ApiError> { fil_api::fetch_task(id, kt) .await? .map(Change::Task) .ok_or_else(|| ApiError::NotFound(format!("no task with id {id}")))}
/// The current state of group `id`, as a change to publish.pub async fn group_change(id: &NanoId, kt: &Kasten) -> Result<Change, ApiError> { fil_api::fetch_group(id, kt) .await? .map(Change::Group) .ok_or_else(|| ApiError::NotFound(format!("no group with id {id}")))}
/// The current state of tag `id`, as a change to publish.pub async fn tag_change(id: &NanoId, kt: &Kasten) -> Result<Change, ApiError> { fil_api::fetch_tag(id, kt) .await? .map(Change::Tag) .ok_or_else(|| ApiError::NotFound(format!("no tag with id {id}")))}
/// The current state of zettel `id`, preceded by its tags (a client may not/// have them yet, e.g. a tag just typed into the front matter).pub async fn zettel_changes(id: &ZettelId, kt: &Kasten) -> Result<Vec<Change>, ApiError> { let (zettel, tags) = fil_api::fetch_zettel(id, kt) .await? .ok_or_else(|| ApiError::NotFound(format!("no zettel with id {id}")))?; let mut changes: Vec<Change> = tags.into_iter().map(Change::Tag).collect(); changes.push(Change::Zettel(zettel)); Ok(changes)}