chadtmiller.com/knotless

//! `dev.knotless.getFeed`: what happened on the repositories the index
//! knows, newest first. The front page shows it.

use super::*;
use crate::index::{Index, cursor_at};
use knotless::records::nsid;
use std::collections::HashSet;

/// Events answered when the asker names no number.
const LIMIT: usize = 40;
/// Commits a push is walked for: past this, it says only that there were
/// at least as many.
const WALKED: usize = 100;
/// Commits a push lists; the rest are counted.
const LISTED: usize = 10;

#[derive(Serialize, TS)]
#[ts(export, rename = "Feed")]
pub struct Output {
    events: Vec<Event>,
    /// Where the next page starts, when there may be one.
    #[serde(skip_serializing_if = "Option::is_none")]
    #[ts(optional)]
    cursor: Option<String>,
}

/// Something an account did to a repository.
#[derive(Serialize, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export, rename = "FeedEvent")]
pub struct Event {
    created_at: String,
    actor: Actor,
    repo: RepoView,
    action: Action,
}

#[derive(Serialize, TS, Clone)]
#[ts(export, rename = "FeedActor")]
pub struct Actor {
    did: String,
    handle: String,
}

/// A repository, under the name it has now.
#[derive(Serialize, TS, Clone)]
#[ts(export, rename = "FeedRepo")]
pub struct RepoView {
    did: String,
    handle: String,
    name: String,
}

#[derive(Serialize, TS)]
#[ts(export, rename = "FeedCommit")]
pub struct CommitView {
    sha: String,
    message: String,
}

#[derive(Serialize, TS)]
#[serde(tag = "$type")]
#[ts(export, rename = "FeedAction")]
pub enum Action {
    /// A repository made, or forked from another.
    #[serde(rename = "dev.knotless.getFeed#created", rename_all = "camelCase")]
    Created {
        #[serde(skip_serializing_if = "Option::is_none")]
        #[ts(optional)]
        description: Option<String>,
        #[serde(skip_serializing_if = "Option::is_none")]
        #[ts(optional)]
        fork_of: Option<RepoView>,
    },
    /// A ref made or moved, and the newest of the commits that brought.
    #[serde(rename = "dev.knotless.getFeed#pushed")]
    Pushed {
        #[serde(rename = "ref")]
        refname: String,
        created: bool,
        count: usize,
        commits: Vec<CommitView>,
    },
    #[serde(rename = "dev.knotless.getFeed#deleted")]
    Deleted {
        #[serde(rename = "ref")]
        refname: String,
    },
    /// A pull or an issue opened.
    #[serde(rename = "dev.knotless.getFeed#opened")]
    Opened { subject: String, title: String },
    /// A comment, or a review with a verdict.
    #[serde(rename = "dev.knotless.getFeed#commented")]
    Commented {
        subject: String,
        title: String,
        text: String,
        #[serde(skip_serializing_if = "Option::is_none")]
        #[ts(optional)]
        verdict: Option<String>,
    },
    /// A pull or an issue moved to another state.
    #[serde(rename = "dev.knotless.getFeed#changed")]
    Changed {
        subject: String,
        title: String,
        state: String,
    },
}

#[derive(Deserialize, Default)]
pub struct Params {
    limit: Option<usize>,
    cursor: Option<String>,
}

pub async fn get_feed(
    State(app): State<Arc<App>>,
    Query(p): Query<Params>,
) -> Result<axum::Json<Output>> {
    let index = app.index.as_ref().ok_or(Shown(
        StatusCode::NOT_FOUND,
        "this server keeps no index".into(),
    ))?;
    let limit = p.limit.unwrap_or(LIMIT).clamp(1, 100);
    let (events, cursor) = events(&app, index, limit, p.cursor.as_deref()).await?;
    Ok(axum::Json(Output { events, cursor }))
}

/// The owner and name of the repository `uri` names.
fn owner_and_name(uri: &str) -> Option<(String, String)> {
    let (did, _, name) = forge::parse_uri(uri).ok()?;
    Some((did.to_string(), name.to_string()))
}

/// Where a page of `limit` ends, from the time and URI of each record it
/// could hold, newest `limit` of each kind: the last it holds, if it is
/// full.
fn cut<'a>(mut keys: Vec<(&'a str, &'a str)>, limit: usize) -> Option<(&'a str, &'a str)> {
    keys.sort_by(|a, b| b.cmp(a));
    keys.get(limit - 1).copied()
}

/// What happened, newest first, after `cursor`: repositories made, refs
/// moved, and what was opened, said and decided on them; and where the
/// next page starts. A page is the newest `limit` records of all three
/// kinds, by time and then URI, less those not shown.
async fn events(
    app: &App,
    index: &Index,
    limit: usize,
    cursor: Option<&str>,
) -> Result<(Vec<Event>, Option<String>)> {
    let (mut repos, _) = index
        .repos_page("", limit, cursor)
        .map_err(anyhow::Error::from)
        .map_err(AppError::internal)?;
    let mut pushes = index
        .pushes(limit, cursor)
        .map_err(anyhow::Error::from)
        .map_err(AppError::internal)?;
    let mut said = index
        .recent(limit, cursor)
        .map_err(anyhow::Error::from)
        .map_err(AppError::internal)?;
    let at = |e: &forge::Entry| {
        e.value["createdAt"]
            .as_str()
            .unwrap_or_default()
            .to_string()
    };
    let keys = repos
        .iter()
        .map(|r| (r.created_at.as_str(), r.uri.as_str()))
        .chain(
            pushes
                .iter()
                .map(|p| (p.created_at.as_str(), p.uri.as_str())),
        )
        .chain(said.iter().map(|(e, _, _)| {
            let at = e.value["createdAt"].as_str().unwrap_or_default();
            (at, e.uri.as_str())
        }))
        .collect();
    let end = cut(keys, limit).map(|(at, uri)| (at.to_string(), uri.to_string()));
    let next = end.as_ref().map(|(at, uri)| cursor_at(at, uri));
    if let Some(end) = &end {
        let held = |at: &str, uri: &str| (at, uri) >= (end.0.as_str(), end.1.as_str());
        repos.retain(|r| held(&r.created_at, &r.uri));
        pushes.retain(|p| held(&p.created_at, &p.uri));
        said.retain(|(e, _, _)| held(&at(e), &e.uri));
    }

    // Each account named, resolved together.
    let parents: Vec<(String, String)> = repos
        .iter()
        .filter_map(|r| r.fork_of.as_deref())
        .filter_map(owner_and_name)
        .collect();
    let mut dids: HashSet<&str> = parents.iter().map(|(did, _)| did.as_str()).collect();
    dids.extend(repos.iter().map(|r| r.did.as_str()));
    dids.extend(pushes.iter().map(|p| p.did.as_str()));
    dids.extend(
        said.iter()
            .flat_map(|(e, did, _)| [e.author.as_str(), did.as_str()]),
    );
    let handles: HashMap<&str, String> = futures::future::join_all(
        dids.into_iter()
            .map(|did| async move { (did, handle(app, did).await) }),
    )
    .await
    .into_iter()
    .collect();
    let actor = |did: &str| Actor {
        did: did.to_string(),
        handle: handles[did].clone(),
    };
    let repo = |did: &str, name: &str| RepoView {
        did: did.to_string(),
        handle: handles[did].clone(),
        name: name.to_string(),
    };

    let mut events: Vec<(String, Event)> = repos
        .iter()
        .map(|r| {
            (
                r.uri.clone(),
                Event {
                    created_at: r.created_at.clone(),
                    actor: actor(&r.did),
                    repo: repo(&r.did, &r.name),
                    action: Action::Created {
                        description: r.description.clone(),
                        fork_of: r
                            .fork_of
                            .as_deref()
                            .and_then(owner_and_name)
                            .map(|(did, name)| repo(&did, &name)),
                    },
                },
            )
        })
        .collect();

    // The mirror of each repository pushed to, opened once.
    let wanted: HashSet<(String, String)> = pushes
        .iter()
        .map(|p| (p.did.clone(), p.name.clone()))
        .collect();
    let mirrors: HashMap<(String, String), Repo> =
        futures::future::join_all(wanted.into_iter().map(|at| async move {
            let repo = Repo::open(app, &at.0, &at.1).await;
            (at, repo)
        }))
        .await
        .into_iter()
        .filter_map(|(at, repo)| Some((at, repo.ok()?)))
        .collect();
    for p in &pushes {
        let mirror = mirrors.get(&(p.did.clone(), p.name.clone()));
        let action = match &p.to {
            None => Action::Deleted {
                refname: p.refname.clone(),
            },
            Some(to) => {
                // A ref that moved brought what it did not reach before; a
                // new one, what the default branch did not reach just
                // before, nor the default branch of a fork's source, which
                // a fork's is named for.
                let head = mirror.map(|r| r.head.as_str()).unwrap_or_default();
                let branch = format!("refs/heads/{head}");
                let before = |at: &str| index.tip_before(at, &branch, &p.created_at);
                let known: Vec<String> = match &p.from {
                    Some(from) => vec![from.clone()],
                    None => [
                        (branch != p.refname).then_some(p.repo.as_str()),
                        p.fork_of.as_deref(),
                    ]
                    .into_iter()
                    .flatten()
                    .map(before)
                    .collect::<rusqlite::Result<Vec<_>>>()
                    .map_err(anyhow::Error::from)
                    .map_err(AppError::internal)?
                    .into_iter()
                    .flatten()
                    .collect(),
                };
                let known: Vec<&str> = known.iter().map(String::as_str).collect();
                let brought = mirror
                    .and_then(|r| r.objects().ok())
                    .and_then(|o| o.brought(to, &known, WALKED).ok())
                    .unwrap_or_default();
                // A fork's copy of its source's refs is the fork itself.
                if p.fork_of.is_some() && p.from.is_none() && brought.is_empty() {
                    continue;
                }
                Action::Pushed {
                    refname: p.refname.clone(),
                    created: p.from.is_none(),
                    count: brought.len(),
                    commits: brought
                        .into_iter()
                        .take(LISTED)
                        .map(|c| CommitView {
                            sha: c.sha,
                            message: c.message,
                        })
                        .collect(),
                }
            }
        };
        let event = Event {
            created_at: p.created_at.clone(),
            actor: actor(&p.did),
            repo: repo(&p.did, &p.name),
            action,
        };
        events.push((p.uri.clone(), event));
    }

    // Pulls and issues, as their repositories' owners let them stand, by
    // title.
    let mut titles: HashMap<String, String> = HashMap::new();
    let on: HashSet<(&str, &str)> = said
        .iter()
        .map(|(_, did, key)| (did.as_str(), key.as_str()))
        .collect();
    let mut named = HashMap::new();
    for (did, key) in on {
        let name = index
            .name(did, key)
            .map_err(anyhow::Error::from)
            .map_err(AppError::internal)?
            .unwrap_or_default();
        let entries = index
            .forge(did, key)
            .map_err(anyhow::Error::from)
            .map_err(AppError::internal)?;
        for e in entries {
            let collection = forge::parse_uri(&e.uri).unwrap_or_default().1;
            if matches!(collection, nsid::PULL | nsid::ISSUE) {
                let title = e.value["title"].as_str().unwrap_or_default().to_string();
                titles.insert(e.uri, title);
            }
        }
        named.insert((did, key), repo(did, &name));
    }
    for (e, did, key) in &said {
        // A pull or an issue is its own subject. One the repository's owner
        // keeps out is not shown, nor what is said about it.
        let subject = if titles.contains_key(&e.uri) {
            e.uri.as_str()
        } else {
            e.value["subject"].as_str().unwrap_or_default()
        };
        let shown = titles.contains_key(subject);
        let title = titles.get(subject).cloned().unwrap_or_default();
        let text = |field: &str| e.value[field].as_str().unwrap_or_default().to_string();
        let subject = subject.to_string();
        let action = match forge::parse_uri(&e.uri).unwrap_or_default().1 {
            nsid::PULL | nsid::ISSUE => Action::Opened { subject, title },
            nsid::COMMENT => Action::Commented {
                subject,
                title,
                text: text("body"),
                verdict: e.value["verdict"].as_str().map(str::to_string),
            },
            _ => Action::Changed {
                subject,
                title,
                state: text("state"),
            },
        };
        let event = Event {
            created_at: text("createdAt"),
            actor: actor(&e.author),
            repo: named[&(did.as_str(), key.as_str())].clone(),
            action,
        };
        events.extend(shown.then(|| (e.uri.clone(), event)));
    }
    events.sort_by(|(a, x), (b, y)| (&y.created_at, b).cmp(&(&x.created_at, a)));
    Ok((events.into_iter().map(|(_, e)| e).collect(), next))
}

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

    #[test]
    fn owners_and_names() {
        assert_eq!(
            owner_and_name("at://did:a/dev.knotless.repo/r"),
            Some(("did:a".into(), "r".into()))
        );
        assert_eq!(owner_and_name("nope"), None);
    }
}