//! This server's own index of the private repositories its members read
//! here: every record in their spaces, synced with the session of the
//! member who last read the repository here ([`App::keep`]), or of its
//! owner, whose own are found as they log in or look at their profile
//! ([`discover`]), so their pages read as a public repository's do, from an
//! index, rather than from every member's repo at each look. A space whose
//! sync fails, as it does once that member is removed, is forgotten, with
//! everything it gave.
//!
//! Nothing written in a space reaches the firehose. Each followed space's
//! authority is asked to tell this server of every write there
//! (`registerNotify`), as is a new private repository's by the command
//! line that makes it, so that it is in the index before anyone looks
//! ([`written`]). Told of one, or every so often in case a notification
//! went missing, the authority is asked which members' repos were written
//! to since the last sync, and each of those is read on through its oplog
//! from the revision it was synced at, checked against the commit its
//! server vouches for ([`knotless::space_sync`]). A notification is only
//! ever a reason to look. One never synced, or whose writes do not check
//! out, is read whole and replaces what it gave before.

use crate::index::Index;
use crate::{App, AppError};
use anyhow::Result;
use atproto::space::LtHash;
use axum::Json;
use axum::extract::State;
use axum::http::{HeaderMap, StatusCode};
use knotless::crypto::PublicKey;
use knotless::env::Ctx;
use knotless::identity;
use knotless::records::nsid;
use knotless::remote::Remote;
use knotless::space::{self, Space};
use knotless::space_sync;
use knotless::url::RepoUrl;
use knotless::xrpc::{Xrpc, XrpcError};
use serde::Deserialize;
use serde_json::{Value, json};
use std::collections::{BTreeMap, HashMap, HashSet};
use std::sync::atomic::Ordering;
use std::sync::{Arc, LazyLock, Mutex};
use std::time::{Duration, Instant};

/// The collections a private repository's pages read, as the appview
/// indexes them from the firehose for public ones; and an organization's
/// records, as its space and its private repositories' spaces keep them,
/// naming its invitations and hidden members.
const COLLECTIONS: [&str; 20] = [
    nsid::REPO,
    nsid::REDIRECT,
    nsid::REF,
    nsid::CLAIM,
    nsid::CHECK,
    nsid::PULL,
    nsid::ISSUE,
    nsid::COMMENT,
    nsid::STATE,
    nsid::HIDE,
    nsid::BLOCK,
    nsid::QUEUE,
    nsid::AUTO_MERGE,
    nsid::LABEL,
    nsid::ASSIGNMENT,
    nsid::REACTION,
    nsid::ORG,
    nsid::ORG_MEMBER,
    nsid::ORG_ACCEPTANCE,
    // A private repository's site, served to its members.
    nsid::SITE,
];

/// How long a space may go unsynced before a member reading it syncs it
/// with their own credential: two of [`follow`]'s sweeps. A space's
/// syncer that fails, as one whose session here lapsed or lacks the
/// scope its space needs does, would otherwise leave it as it was until
/// they log in again, every write since missing from it, to every reader.
pub const STALE: Duration = Duration::from_secs(60);

/// Syncs each space `app` keeps an index of as its authority tells of
/// writes there, and every one of them every `every` besides, for as long
/// as it runs, each as the member it is synced with.
pub async fn follow(app: Arc<App>, every: Duration) {
    let Some(index) = app.spaces.clone() else {
        return;
    };
    app.following.store(true, Ordering::Relaxed);
    let mut swept: Option<Instant> = None;
    loop {
        let all = swept.is_none_or(|at| at.elapsed() >= every);
        if all {
            swept = Some(Instant::now());
        }
        let due = std::mem::take(&mut *app.due.lock().unwrap());
        for (uri, did) in to_sync(app.syncers(), all, &due) {
            let ctx = crate::login::syncer_ctx(&app, &did);
            // A credential asked for afresh now and then, which the owner's
            // PDS gives only to one still a member.
            if all {
                space::forget(&ctx);
            }
            let synced = sync(&ctx, &index, &uri).await;
            refused_for_scope(&did, &uri, &synced);
            let failed = synced.is_err();
            if !failed {
                app.synced_now(&uri);
            }
            // Refused, as once the space is deleted or its syncer no longer
            // a member, it is followed no more; its authority not reached,
            // as while it restarts, it is synced again at the next.
            let refused = synced.as_ref().err().is_some_and(space::denied);
            let gone = synced.as_ref().err().is_some_and(gone);
            // What was written there shows on the next look at it, as does
            // what a sync that failed read of its other repos.
            if !matches!(synced, Ok(false)) {
                app.moved(&uri);
            }
            crate::telemetry::space_synced(&uri, synced.map(drop));
            if gone {
                let _ = index.space_gone(&uri);
            }
            if refused {
                app.unfollow(&index, &uri);
            }
            if !failed {
                listen(&app, &ctx, &uri).await;
            }
        }
        let since = swept.map_or(every, |at| at.elapsed());
        let left = every.saturating_sub(since);
        tokio::select! {
            () = tokio::time::sleep(left) => {}
            () = app.told.notified() => {}
        }
    }
}

/// Of `syncers`, each a space and the member it is synced as, those to
/// sync now: every one on a sweep (`all`), else those whose authority told
/// of a write there (`due`). None whose member was refused it for want of
/// scope lately ([`out_of_scope`]): their grant stays as it was until they
/// log in again, and the space's readers sync it meanwhile, once it goes
/// [`STALE`].
fn to_sync(
    syncers: Vec<(String, String)>,
    all: bool,
    due: &HashSet<String>,
) -> Vec<(String, String)> {
    let now = |(uri, _): &(String, String)| all || due.contains(uri);
    let asked = |(uri, did): &(String, String)| !out_of_scope(did, uri);
    syncers.into_iter().filter(now).filter(asked).collect()
}

/// Of [`space::denied`], what says the space itself is no more, rather
/// than that its syncer may no longer read it.
fn gone(err: &anyhow::Error) -> bool {
    let said = XrpcError::of(err).map(|e| e.error.as_str());
    matches!(said, Some("SpaceNotFound" | "SpaceDeleted"))
}

/// How long this server's asking a space's authority to tell it of writes
/// there is taken to hold before it asks again: the reference PDS keeps
/// one for a day.
pub const RENEW: Duration = Duration::from_secs(12 * 60 * 60);

/// Asks the authority of the space `uri`, as its syncer (`ctx`), to tell
/// this server of each write there, unless it did within [`RENEW`].
async fn listen(app: &App, ctx: &Ctx, uri: &str) {
    let mut registered = app.registered.lock().await;
    let asked = registered.get(uri).is_some_and(|at| at.elapsed() < RENEW);
    if asked {
        return;
    }
    let did = app.gateway.as_ref().map(|gw| gw.did.as_str());
    let service = format!("{}{}", did.unwrap_or_default(), space::APPVIEW);
    let heard = register(ctx, uri, &service).await;
    crate::telemetry::space_unheard(uri, heard);
    // Asked again after as long whether or not it was heard, so that an
    // authority without notifications is not asked at every sync.
    registered.insert(uri.to_string(), Instant::now());
}

/// [`listen`], whatever came before.
async fn register(ctx: &Ctx, uri: &str, service: &str) -> Result<String> {
    let joined = space::joined(ctx, uri).await?;
    let owner = identity::resolve(ctx, authority(uri)).await?;
    let host = Xrpc::new(ctx.http.clone(), &owner.pds);
    joined.register_notify(&host, &owner.did, service).await
}

/// The account a space is under, its authority.
fn authority(uri: &str) -> &str {
    let authority = uri.trim_start_matches("at://").split('/').next();
    authority.unwrap_or_default()
}

#[derive(Deserialize)]
pub struct Told {
    space: String,
}

/// `com.atproto.space.notifyWrite`, from a space's authority: a member's
/// repo there moved. A space followed is synced now; one not yet, as a
/// private repository just made is, is followed as its owner, and synced
/// now, if they are logged in here. Anyone else's is no one's to read here.
pub async fn written(
    State(app): State<Arc<App>>,
    headers: HeaderMap,
    Json(told): Json<Told>,
) -> Result<Json<Value>, AppError> {
    let (index, owner) = from_authority(&app, &headers, &told.space, NOTIFY_WRITE).await?;
    app.live.written(&told.space);
    let followed = app.syncers().iter().any(|(uri, _)| *uri == told.space);
    let ctx = crate::login::ctx_for(&app, &owner);
    let theirs = knotless::session::load(&ctx, &owner).is_some();
    if !followed && theirs {
        let follows = index.follow_space(&told.space, &owner);
        follows
            .map_err(anyhow::Error::from)
            .map_err(AppError::internal)?;
    }
    app.due.lock().unwrap().insert(told.space);
    app.told.notify_one();
    Ok(Json(json!({})))
}

/// `com.atproto.space.notifySpaceDeleted`, from a space's authority: what
/// it gave goes, and it is followed no more.
pub async fn deleted(
    State(app): State<Arc<App>>,
    headers: HeaderMap,
    Json(told): Json<Told>,
) -> Result<Json<Value>, AppError> {
    let (index, _) = from_authority(&app, &headers, &told.space, NOTIFY_DELETED).await?;
    app.unfollow(&index, &told.space);
    let gone = index.space_gone(&told.space);
    gone.map_err(anyhow::Error::from)
        .map_err(AppError::internal)?;
    app.registered.lock().await.remove(&told.space);
    Ok(Json(json!({})))
}

pub const NOTIFY_WRITE: &str = "com.atproto.space.notifyWrite";
pub const NOTIFY_DELETED: &str = "com.atproto.space.notifySpaceDeleted";

/// This server's index of spaces, and the account that signed the
/// request's token for `lxm`, to this server's [`space::APPVIEW`]: the
/// authority of `space`, a private repository's or its organization's, or
/// it is refused.
async fn from_authority(
    app: &App,
    headers: &HeaderMap,
    space: &str,
    lxm: &str,
) -> Result<(Arc<Index>, String), AppError> {
    let none = AppError(StatusCode::NOT_FOUND, "this server syncs no spaces".into());
    let index = app.spaces.clone().ok_or(none)?;
    let did = app.gateway.as_ref().map(|gw| gw.did.as_str());
    let aud = format!("{}{}", did.unwrap_or_default(), space::APPVIEW);
    let caller = crate::gateway::caller_for(app, headers, &aud, lxm).await?;
    let theirs = space.starts_with(&space::uri(&caller, ""));
    if !theirs && space != knotless::org::space_uri(&caller) {
        let why = "only a private repository's authority tells of its space";
        return Err(AppError(StatusCode::FORBIDDEN, why.into()));
    }
    Ok((index, caller))
}

/// How often an owner's spaces are listed again, as they log in or look at
/// their profile, for private repositories made since.
pub const DISCOVER: Duration = Duration::from_secs(60);

/// Follows, and syncs, each space `did` owns that this server does not yet,
/// as them: their private repositories, so that their profile lists them
/// from the index, and keeps whether their PDS has spaces at all, for the
/// pages to say whether they can make one ([`App::has_spaces`]). Of the
/// spaces followed that are someone else's, each they may be a member of
/// ([`Index::maybe_member`]) is asked of its authority as them, so the
/// repositories shared with them are listed to them before they open one,
/// and no longer once they are refused. In the background, at most once
/// every [`DISCOVER`], so whoever looks never waits on it, and one at a
/// time for each account, so that an answer asked before a removal is
/// never kept after one asked since.
pub fn discover(app: &Arc<App>, did: &str) {
    let Some(index) = app.spaces.clone().filter(|_| app.discovering(did)) else {
        return;
    };
    let app = app.clone();
    let did = did.to_string();
    tokio::spawn(async move {
        let found = owned(&app, &index, &did).await;
        crate::telemetry::space_synced(&did, found);
    });
}

/// [`discover`], waited on.
async fn owned(app: &App, index: &Index, did: &str) -> Result<()> {
    let lock = app.discover_lock(did);
    let _held = lock.lock().await;
    let ctx = crate::login::ctx_for(app, did);
    let me = knotless::session::me(&ctx).await?;
    let followed: HashSet<String> = index.spaces()?.into_iter().map(|(uri, _)| uri).collect();
    let owned = Remote::own_spaces(&ctx, &me).await?;
    app.found_spaces(did, owned.is_some());
    // One its authority said is deleted, which a PDS can go on naming, is
    // not followed again only to fail.
    let gone = index.gone_spaces()?;
    for uri in owned
        .into_iter()
        .flatten()
        .filter(|uri| !gone.contains(uri))
    {
        index.follow_space(&uri, did)?;
        let new = !followed.contains(&uri);
        let synced = if new {
            sync(&ctx, index, &uri).await.map(drop)
        } else {
            Ok(())
        };
        // One that fails keeps none of the rest from being looked at; its
        // sync failing, [`follow`] forgets it.
        crate::telemetry::space_synced(&uri, synced);
    }
    // Asked afresh, as a reading is, since a kept credential outlives a
    // removal.
    space::forget(&ctx);
    let members = index.maybe_member(did)?.into_iter();
    for uri in members.filter(|uri| !out_of_scope(did, uri)) {
        let joined = space::joined(&ctx, &uri).await;
        refused_for_scope(did, &uri, &joined);
        // Anything but an answer leaves it as it was, to ask again.
        if joined.is_ok() {
            index.read_by(&uri, did)?;
            // Followed by no one, as after the index is rebuilt, it is
            // synced as them, so what they read is listed again.
            if !followed.contains(&uri) {
                index.follow_space(&uri, did)?;
                crate::telemetry::space_synced(&uri, sync(&ctx, index, &uri).await.map(drop));
            }
        }
        // Refused by its authority, they read it no more; its authority not
        // reached, or the session not let ask, says nothing either way.
        crate::telemetry::reader_checked(&uri, did, &joined);
        if joined.as_ref().err().is_some_and(space::denied) {
            index.unread_by(&uri, did)?;
        }
    }
    Ok(())
}

/// The spaces each account's session was refused for want of scope, by
/// account and space, and when: not asked again until [`SCOPE_ASKED`]
/// passes, or it logs in again ([`in_scope_again`]), as its grant stays as
/// it was until then, and each asking costs a call to its PDS, every
/// [`DISCOVER`] as a reader, every sweep of [`follow`] as a syncer.
static OUT_OF_SCOPE: LazyLock<Mutex<HashMap<(String, String), Instant>>> =
    LazyLock::new(Mutex::default);

/// How long a space a session was refused for want of scope is not asked.
const SCOPE_ASKED: Duration = Duration::from_secs(60 * 60);

/// Whether `did`'s session was refused the space `uri` for want of scope
/// within [`SCOPE_ASKED`].
fn out_of_scope(did: &str, uri: &str) -> bool {
    let refused = OUT_OF_SCOPE.lock().unwrap();
    let at = refused.get(&(did.to_string(), uri.to_string()));
    at.is_some_and(|at| at.elapsed() < SCOPE_ASKED)
}

/// Keeps that `did`'s session was refused the space `uri` for want of
/// scope, if `joined` says it was.
fn refused_for_scope<T>(did: &str, uri: &str, joined: &Result<T>) {
    if joined
        .as_ref()
        .err()
        .is_some_and(knotless::oauth::out_of_scope)
    {
        let key = (did.to_string(), uri.to_string());
        OUT_OF_SCOPE.lock().unwrap().insert(key, Instant::now());
    }
}

/// Forgets the spaces `did`'s session was refused for want of scope: it
/// was granted anew, as by logging in again, so each is asked again.
pub fn in_scope_again(did: &str) {
    OUT_OF_SCOPE.lock().unwrap().retain(|(of, _), _| of != did);
}

/// Takes in what was written to the space `uri` since it was last synced,
/// and says whether anything was.
pub async fn sync(ctx: &Ctx, index: &Index, uri: &str) -> Result<bool> {
    let joined = space::joined(ctx, uri).await?;
    sync_with(ctx, index, &joined).await
}

/// [`sync`], with `joined`, the space as some member reads it. One sync
/// of a space at a time, so that none puts back what it read over what a
/// later one did. A repo that fails keeps none of the rest from being
/// read: it is the error, and the space's cursor stays where it was, so
/// it is listed, and read, again at the next.
pub async fn sync_with(ctx: &Ctx, index: &Index, joined: &Space) -> Result<bool> {
    let uri = joined.uri.as_str();
    let lock = index.syncing(uri);
    let _held = lock.lock().await;
    let owner = identity::resolve(ctx, authority(uri)).await?;
    let host = Xrpc::new(ctx.http.clone(), &owner.pds);
    let since = index.spaces()?.into_iter().find(|(u, _)| u == uri);
    let since = since.and_then(|(_, cursor)| cursor);
    let (writers, now) = joined
        .writers_since(&host, &owner.did, since.as_deref())
        .await?;
    // A repo written to again while the listing was read is listed twice:
    // the later says where it is now. One read to there already has
    // nothing new.
    let writers: BTreeMap<String, String> =
        writers.into_iter().map(|w| (w.did, w.repo_rev)).collect();
    let fresh = |(did, rev): &(String, String)| {
        let kept = index.space_writer(uri, did).ok().flatten();
        kept.is_none_or(|(at, _)| at != *rev)
    };
    let writers: Vec<_> = writers.into_iter().filter(fresh).collect();
    let wrote = !writers.is_empty();
    let mut failed = None;
    for (did, _) in writers {
        let read = member(ctx, index, joined, &did).await;
        failed = failed.or(read.err());
    }
    failed.map_or(Ok(()), Err)?;
    if let Some(now) = now {
        index.set_space_cursor(uri, &now)?;
    }
    Ok(wrote)
}

/// Takes in `did`'s repo in `joined`: read on from where it was last
/// synced to, or whole, or, should it not come to its commit whole, from
/// the start of its oplog, where each record's CID is vouched for apart
/// from its bytes.
async fn member(ctx: &Ctx, index: &Index, joined: &Space, did: &str) -> Result<()> {
    let theirs = remote(ctx, joined, did).await?;
    let kem = atproto::did::kem_key(&identity::document(ctx, did).await?)?;
    let uri = &joined.uri;
    let read = match index.space_writer(uri, did)? {
        Some((rev, state)) => {
            let read = read_on(index, uri, &theirs, &kem, &rev, &state).await;
            crate::telemetry::read_whole(uri, did, &read);
            read.is_ok()
        }
        None => false,
    };
    if !read {
        let read = whole(index, uri, &theirs, &kem).await;
        crate::telemetry::read_from_start(uri, did, &read);
        if read.is_err() {
            from_start(index, uri, &theirs, &kem).await?;
        }
    }
    Ok(())
}

/// `did`'s repo in the space `joined`, read with its credential.
pub async fn remote(ctx: &Ctx, joined: &Space, did: &str) -> Result<Remote> {
    let id = identity::resolve(ctx, did).await?;
    Ok(Remote {
        xrpc: Xrpc::new(ctx.http.clone(), &id.pds),
        identity: id,
        url: RepoUrl {
            authority: did.to_string(),
            name: String::new(),
        },
        space: Some(joined.at(did)),
    })
}

/// Whether the pages read records of `collection`.
fn indexed(collection: &str) -> bool {
    COLLECTIONS.contains(&collection)
}

/// Reads on from `rev` in `theirs`'s oplog, where its set hash's state was
/// `state`: each write since, checked against the commit they come to,
/// and each record they leave, against the CID its last write gave it.
async fn read_on(
    index: &Index,
    uri: &str,
    theirs: &Remote,
    kem: &PublicKey,
    rev: &str,
    state: &[u8],
) -> Result<()> {
    let hash = LtHash::from_state(state)?;
    let read = space_sync::read_on(theirs, kem, Some(rev), hash).await?;
    let did = theirs.did();
    let (changes, unreadable) = read.changes(indexed)?;
    crate::telemetry::unreadable(uri, did, &unreadable);
    for change in changes {
        let (collection, rkey) = (&change.collection, &change.rkey);
        match &change.value {
            Some(value) => index.put_from_space(uri, did, collection, rkey, value)?,
            None => index.delete_from_space(uri, did, collection, rkey)?,
        }
    }
    index.set_space_writer(uri, did, &read.commit.rev, &read.hash.state())?;
    Ok(())
}

/// Reads `theirs` whole, in place of all it gave before: what it no longer
/// has is taken out.
async fn whole(index: &Index, uri: &str, theirs: &Remote, kem: &PublicKey) -> Result<()> {
    let whole = space_sync::whole(theirs, kem).await?;
    let did = theirs.did();
    crate::telemetry::unreadable(uri, did, &whole.unreadable);
    let records = whole.records.iter().filter(|r| indexed(&r.collection));
    let records = records.map(|r| (&r.collection, &r.rkey, &r.value));
    let commit = (whole.commit.rev.as_str(), &whole.hash);
    replace(index, uri, did, records, commit)
}

/// Reads `theirs` from the start of its oplog, in place of all it gave
/// before, as [`whole`] does: for a repo whose archive does not come to
/// its commit, as one holding a record its server keeps cut short, whose
/// bytes no longer come to the CID it was written with. The oplog gives
/// each record's CID apart from its bytes, which the commit vouches for
/// all together, and each record's value is checked against its own:
/// one that does not come to it is left out. It comes to the commit only
/// if the oplog still holds every write the repo was made of.
async fn from_start(index: &Index, uri: &str, theirs: &Remote, kem: &PublicKey) -> Result<()> {
    let read = space_sync::read_on(theirs, kem, None, LtHash::default()).await?;
    let did = theirs.did();
    let (changes, unreadable) = read.changes(indexed)?;
    crate::telemetry::unreadable(uri, did, &unreadable);
    let records = changes.iter().filter_map(|c| {
        let value = c.value.as_ref()?;
        Some((&c.collection, &c.rkey, value))
    });
    replace(
        index,
        uri,
        did,
        records,
        (read.commit.rev.as_str(), &read.hash),
    )
}

/// Puts `records` in the index as all `did`'s repo in the space `uri`
/// holds, at the revision and set hash its commit came to: what it gave
/// before and no longer has is taken out.
fn replace<'a>(
    index: &Index,
    uri: &str,
    did: &str,
    records: impl Iterator<Item = (&'a String, &'a String, &'a Value)>,
    (rev, hash): (&str, &LtHash),
) -> Result<()> {
    let mut gone: HashSet<(String, String)> = index.space_records(uri, did)?.into_iter().collect();
    for (collection, rkey, value) in records {
        index.put_from_space(uri, did, collection, rkey, value)?;
        gone.remove(&(collection.clone(), rkey.clone()));
    }
    for (collection, rkey) in gone {
        index.delete_from_space(uri, did, &collection, &rkey)?;
    }
    index.set_space_writer(uri, did, rev, &hash.state())?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn a_space_out_of_scope_is_not_asked_again_a_while() {
        let (did, uri) = ("did:plc:s", "at://did:plc:o/space/dev.knotless.org/self");
        refused_for_scope::<()>(did, uri, &Ok(()));
        refused_for_scope::<()>(did, uri, &Err(anyhow::anyhow!("connection refused")));
        assert!(!out_of_scope(did, uri));
        let scope = anyhow::anyhow!("InvalidToken (403): Scope does not permit reading {uri}");
        refused_for_scope::<()>(did, uri, &Err(scope));
        assert!(out_of_scope(did, uri));
        assert!(!out_of_scope("did:plc:t", uri));
        let key = (did.to_string(), uri.to_string());
        let long_ago = Instant::now() - SCOPE_ASKED;
        OUT_OF_SCOPE.lock().unwrap().insert(key, long_ago);
        assert!(!out_of_scope(did, uri));
    }

    #[test]
    fn syncers_out_of_scope_wait_till_they_log_in_again() {
        let did = "did:plc:syncer";
        let (uri, other) = (
            "at://did:plc:o/space/dev.knotless.space/r",
            "at://did:plc:o/space/dev.knotless.space/s",
        );
        let syncers = || vec![(uri.into(), did.into()), (other.into(), did.into())];
        let uris = |s: Vec<(String, String)>| s.into_iter().map(|(u, _)| u).collect::<Vec<_>>();
        let none = HashSet::new();
        assert_eq!(uris(to_sync(syncers(), true, &none)), [uri, other]);
        let due = HashSet::from([other.to_string()]);
        assert_eq!(uris(to_sync(syncers(), false, &due)), [other]);
        let scope = anyhow::anyhow!("InvalidToken (403): Scope does not permit reading {uri}");
        refused_for_scope::<()>(did, uri, &Err(scope));
        assert_eq!(uris(to_sync(syncers(), true, &none)), [other]);
        in_scope_again(did);
        assert_eq!(uris(to_sync(syncers(), true, &none)), [uri, other]);
    }
}