johnandersen777.bsky.social/knotless

Switch repository

Loading…

//! Private repositories, on a server people log in to: each read as the
//! viewer asking, with the session they logged in to the pages with, and
//! never as this server.
//!
//! A query is answered within [`reading`], which names its viewer: the
//! account the session cookie or service token proves, or no one. A
//! repository no one may read publicly is then looked for in its space as
//! that viewer ([`App::as_viewer`]), whose own PDS mints the delegation
//! token and whose membership the owner's PDS checks as it issues the
//! credential. What is found is kept for that viewer alone: its mirror
//! under their cache, its entry under their DID. Anything else reads as no
//! one, and so finds public repositories only: a card, a clone, a task
//! spawned to refresh a mirror.
//!
//! Membership is asked again, with a fresh credential, once the last
//! answer is older than [`REPROVE`], so someone removed from a repository
//! stops reading it here within that; in between it is read as found
//! then, with no round trip to find it again. Logging out forgets what the
//! viewer was shown at once ([`forget`]).
//!
//! An answer that read anything private is the viewer's alone, kept by no
//! cache, the browser's included ([`unshared`]).
//!
//! A private repository read here is indexed here: its space is synced
//! into [`App::spaces`] with the session of the member who last read it
//! ([`crate::spaces`]), first as they read it, and its pulls, issues and
//! checks are read from that, as a public repository's are from the
//! appview's index, by viewers proven members as above. Its owner's own
//! are found as they log in or look at their profile
//! ([`crate::spaces::discover`]), so their list of repositories is read
//! from the index too, never from their PDS as they look.

use crate::{App, AppError, Located};
use axum::http::{HeaderValue, StatusCode, header};
use axum::response::Response;
use knotless::env::Ctx;
use knotless::remote::Remote;
use knotless::space::Space;
use knotless::url::RepoUrl;
use std::future::Future;
use std::sync::Mutex;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Duration, Instant};

/// How long a viewer's reading of a private repository is taken as
/// allowed before the owner's PDS is asked again: how long someone removed
/// from it may still read it here.
pub const REPROVE: Duration = Duration::from_secs(60);

/// Who a request is answered for, and whether the answer read anything
/// private.
struct Reading {
    viewer: Mutex<Option<String>>,
    delegated: Option<Delegated>,
    private: AtomicBool,
}

/// A delegation token a query came with, checked and spent here
/// ([`App::spend`]): it reads the one private repository whose space it
/// names, as the member it names, with no session of theirs here.
#[derive(Clone)]
pub struct Delegated {
    pub did: String,
    pub space: String,
    pub token: String,
}

tokio::task_local! {
    static READING: Reading;
}

/// `answer`, as `viewer` reads, or as `delegated` names once it proves
/// them a member, and whether it read anything private.
pub async fn reading<F: Future>(
    viewer: Option<String>,
    delegated: Option<Delegated>,
    answer: F,
) -> (F::Output, bool) {
    let reading = Reading {
        viewer: Mutex::new(viewer),
        delegated,
        private: AtomicBool::new(false),
    };
    let read = async {
        let out = answer.await;
        (out, READING.with(|r| r.private.load(Ordering::Relaxed)))
    };
    READING.scope(reading, read).await
}

/// The viewer of the request being answered: none outside [`reading`].
pub fn viewer() -> Option<String> {
    READING
        .try_with(|r| r.viewer.lock().unwrap().clone())
        .ok()
        .flatten()
}

/// The delegation token the request being answered came with, if it is
/// for the space `uri`, and from its viewer if it has one.
pub fn delegated(uri: &str) -> Option<Delegated> {
    let found = READING.try_with(|r| r.delegated.clone()).ok().flatten()?;
    let theirs = viewer().is_none_or(|v| v == found.did);
    (found.space == uri && theirs).then_some(found)
}

/// Takes `did`, whom a delegation token proved a member of what it reads,
/// as the request's viewer from now on.
fn proven(did: &str) {
    let _ = READING.try_with(|r| *r.viewer.lock().unwrap() = Some(did.to_string()));
}

/// Marks the answer being made as holding something private.
pub fn touched() {
    let _ = READING.try_with(|r| r.private.store(true, Ordering::Relaxed));
}

/// `res`, for its viewer alone, and kept by no cache at all.
pub fn unshared(mut res: Response) -> Response {
    res.headers_mut().insert(
        header::CACHE_CONTROL,
        HeaderValue::from_static("private, no-store"),
    );
    res
}

/// Forgets what `did` was shown of private repositories, and the
/// credentials and mirrors it was read with: as they log out. Which
/// spaces they were proven to read is kept, to be asked again at their
/// next login.
pub fn forget(app: &App, did: &str) {
    app.viewed.lock().unwrap().retain(|(v, _, _), _| v != did);
    if let Some(index) = &app.spaces {
        let _ = index.drop_syncer(did);
    }
    let ctx = crate::login::ctx_for(app, did);
    knotless::space::forget(&ctx);
    let _ = std::fs::remove_dir_all(&ctx.env.cache_dir);
}

type Viewed = (String, String, String);

impl App {
    /// Whether `viewer` was shown `did`'s repository `name` as private.
    pub(crate) fn shown_private(&self, viewer: &str, did: &str, name: &str) -> bool {
        let key = (viewer.to_string(), did.to_string(), name.to_string());
        self.viewed.lock().unwrap().contains_key(&key)
    }

    /// `did`'s private repository `url`, read as `viewer`: `missing` (or
    /// that there is no such repository) if they may not read it, or it is
    /// not private, so that a non-member learns nothing of it. With
    /// `delegated`, a delegation token of theirs for its space, they are
    /// proven a member with it rather than with a session of theirs here.
    pub(crate) async fn as_viewer(
        &self,
        viewer: &str,
        did: &str,
        url: &RepoUrl,
        missing: Option<AppError>,
        delegated: Option<&Delegated>,
    ) -> Result<Located, AppError> {
        let none = format!("{url}: no such repository");
        let missing = missing.unwrap_or(AppError(StatusCode::NOT_FOUND, none));
        let ctx = crate::login::ctx_for(self, viewer);
        let key = (viewer.to_string(), did.to_string(), url.name.clone());
        let entry = self
            .viewed
            .lock()
            .unwrap()
            .entry(key.clone())
            .or_default()
            .clone();
        let mut entry = entry.lock().await;
        let reprove = Duration::from_millis(self.reprove.load(Ordering::Relaxed));
        let stale = entry.proved.is_none_or(|at| at.elapsed() >= reprove);
        // Between provings the space is read as it was found: its owner's
        // session and a member's credential each last a minute more.
        let remote = match entry.remote.clone().filter(|_| !stale) {
            Some(remote) => remote,
            None => {
                // A credential is asked for afresh, which the owner's PDS
                // gives only to a member.
                let opened = match delegated {
                    Some(d) => Remote::delegated(&ctx, url, viewer, &d.token).await,
                    None => {
                        knotless::space::forget(&ctx);
                        Remote::open(&ctx, url).await
                    }
                };
                match opened {
                    Ok(remote) if remote.space.is_some() => remote,
                    // Not private, or not theirs to read: as if there were
                    // none.
                    other => {
                        let failed = other.err().filter(|e| !knotless::space::refused(e));
                        if failed.is_none() {
                            self.reads(viewer, &knotless::space::uri(did, &url.name), false);
                        }
                        let e = failed.map_or(missing, AppError::upstream);
                        return self.unview(&key, Err(e));
                    }
                }
            }
        };
        let space = entry.space.clone().unwrap_or_default();
        let moves = self.moves_of(&space);
        let entry_key = entry.key(did);
        let any = self.events_for(did);
        let events = self.stamp(did, entry_key.as_deref());
        // A space no one here has a session for is synced only as it is
        // read, with what proved its reader a member.
        let followed = self.space_followed(&space);
        let fresh = match (&entry.git, entry.synced) {
            // Nothing is written to a space but its sync finds, within
            // seconds, or the pages touch.
            (Some(_), Some(_)) if followed => entry.moves == moves && entry.events == events,
            (Some(_), Some(at)) => at.elapsed() < self.ttl && entry.events == events,
            _ => false,
        };
        // No `?` ends a block, which coverage would count as a line never
        // run.
        let mut read = Ok(());
        if !fresh {
            read = self.sync(&ctx, remote.clone(), url, &mut entry).await;
        }
        if !fresh && read.is_ok() {
            entry.moves = moves;
            let found = entry.key(did);
            entry.events = self.settled(did, entry_key.as_deref(), events, any, found.as_deref());
        }
        // Fresh or not, it is synced as whoever read it last: once its
        // syncer logs out, the next to read it takes over. A reader with a
        // delegation token has no session to sync it with after: it is
        // synced with their credential as they read it, when nobody's
        // session syncs it.
        if read.is_ok() {
            let space = entry.space.clone().unwrap_or_default();
            read = match &remote.space {
                Some(joined) if delegated.is_some() => self.keep_read(&ctx, joined).await,
                _ => self.keep(&space, viewer).await,
            };
        }
        self.unview(&key, read)?;
        // Theirs to read, it is in their lists from now on.
        if viewer != did {
            self.reads(viewer, entry.space.as_deref().unwrap_or_default(), true);
        }
        if stale {
            entry.proved = Some(Instant::now());
            entry.remote = Some(remote);
        }
        if delegated.is_some() {
            proven(viewer);
        }
        touched();
        Ok(self.found(&entry, did, ctx))
    }

    /// Keeps an index of the private repository in the space `joined`, as
    /// a member reads it with a delegation token: synced now with their
    /// credential if it never was, or if no session here syncs it. A client
    /// shows the appview each private repository it writes to so
    /// ([`knotless::space::show_appview`]), so what it wrote is heard of.
    async fn keep_read(&self, ctx: &Ctx, joined: &Space) -> Result<(), AppError> {
        let none = "this server keeps no index of private repositories";
        let none = AppError(StatusCode::INTERNAL_SERVER_ERROR, none.into());
        let index = self.spaces.as_ref().ok_or(none)?;
        let uri = joined.uri.as_str();
        let kept = index.keep_space(uri).map_err(anyhow::Error::from);
        kept.map_err(AppError::internal)?;
        let synced = index.space_synced(uri).map_err(anyhow::Error::from);
        let synced = synced.map_err(AppError::internal)?;
        // Synced as the reader, too, when its syncer has not lately.
        let sync = !synced || self.syncer_of(uri).is_none() || self.sync_stale(uri);
        if !sync {
            return Ok(());
        }
        let read = crate::spaces::sync_with(ctx, index, joined).await;
        self.read_now(uri, read, synced)
    }

    /// What a reader's sync of the space `uri` makes of their read, with
    /// `synced` whether it was ever synced before. A sync that failed
    /// counts as one made, so it is tried again only after
    /// [`STALE`](crate::spaces::STALE), not at each read; and a space
    /// already synced is then read from its index as it was, the failure
    /// logged (`space not synced`), rather than failing the read. One never
    /// synced has nothing to be read from, and its failure is the read's.
    fn read_now(
        &self,
        uri: &str,
        read: anyhow::Result<bool>,
        synced: bool,
    ) -> Result<(), AppError> {
        if read.is_ok() || synced {
            self.synced_now(uri);
        }
        if matches!(read, Ok(true)) {
            self.moved(uri);
        }
        let read = read.map(drop);
        let (logged, read) = if synced {
            (read, Ok(()))
        } else {
            (Ok(()), read)
        };
        crate::telemetry::space_synced(uri, logged);
        read.map_err(AppError::upstream)
    }

    /// Notes whether `viewer` may read the private repository in the
    /// space `uri`, for their lists of repositories ([`Index::read_of`]).
    ///
    /// [`Index::read_of`]: crate::index::Index::read_of
    pub(crate) fn reads(&self, viewer: &str, uri: &str, may: bool) {
        let Some(index) = &self.spaces else { return };
        let _ = match may {
            true => index.read_by(uri, viewer),
            false => index.unread_by(uri, viewer),
        };
    }

    /// Whether the space `uri` is synced as its authority tells of writes
    /// there, so that it is current until a sync of it finds one
    /// ([`App::moves_of`]).
    pub(crate) fn space_followed(&self, uri: &str) -> bool {
        self.following.load(Ordering::Relaxed) && self.syncer_of(uri).is_some()
    }

    /// The member whose session here syncs the space `uri`, if any.
    fn syncer_of(&self, uri: &str) -> Option<String> {
        let index = self.spaces.as_ref()?;
        index.syncer_of(uri).ok().flatten()
    }

    /// Keeps an index of the private repository in `space`, synced as
    /// `syncer`, a member, who reads it: synced now if it never was, so its
    /// pages read from the index from their first look.
    pub async fn keep(&self, space: &str, syncer: &str) -> Result<(), AppError> {
        let none = "this server keeps no index of private repositories";
        let none = AppError(StatusCode::INTERNAL_SERVER_ERROR, none.into());
        let index = self.spaces.as_ref().ok_or(none)?;
        let followed = index.follow_space(space, syncer);
        followed
            .map_err(anyhow::Error::from)
            .map_err(AppError::internal)?;
        self.follow_org(space, syncer);
        let synced = index.space_synced(space);
        let synced = synced.map_err(anyhow::Error::from);
        let synced = synced.map_err(AppError::internal)?;
        // Synced, and lately: what was written since its sync will reach it.
        if synced && !self.sync_stale(space) {
            return Ok(());
        }
        let ctx = crate::login::syncer_ctx(self, syncer);
        // Synced before, it is read as it was while it syncs aside, as a
        // sync reads the whole space and can take seconds: only one never
        // synced has nothing to read until its sync is done. Taken as
        // synced now, so reads meanwhile start no other.
        if synced {
            self.synced_now(space);
            let (moves, last) = (self.moves.clone(), self.last_synced.clone());
            let (index, space) = (index.clone(), space.to_string());
            tokio::spawn(async move {
                let read = crate::spaces::sync(&ctx, &index, &space).await;
                last.lock().unwrap().insert(space.clone(), Instant::now());
                if matches!(read, Ok(true)) {
                    *moves.lock().unwrap().entry(space.clone()).or_default() += 1;
                }
                crate::telemetry::space_synced(&space, read.map(drop));
            });
            return Ok(());
        }
        let read = crate::spaces::sync(&ctx, index, space).await;
        self.read_now(space, read, synced)
    }

    /// Stops syncing the space `uri`, and forgets what it gave: its sync
    /// failed, as it does once its syncer is no longer a member.
    pub fn unfollow(&self, index: &crate::index::Index, uri: &str) {
        let _ = index.forget_space(uri);
        self.moved(uri);
    }

    /// The spaces to sync, each with the member whose session it is synced
    /// with.
    pub fn syncers(&self) -> Vec<(String, String)> {
        let index = self.spaces.as_ref();
        let syncers = index.map(|i| i.syncers().unwrap_or_default());
        syncers.unwrap_or_default()
    }

    /// `read`, having forgotten that the viewer `key` names was shown its
    /// repository if reading it again failed.
    fn unview<T>(&self, key: &Viewed, read: Result<T, AppError>) -> Result<T, AppError> {
        if read.is_err() {
            self.viewed.lock().unwrap().remove(key);
        }
        read
    }
}