protocol.rs 94.9%

559 of 589 lines run, 119 of 232 functions · run #507 on main · run missed
//! The wire protocol, embedded (hub spec §3.3, §5.4): `oxy-server`'s
//! `Server` over a backend that resolves `owner/name`, authorises reads
//! and pushes by account role and token scope, and records who pushed.

use crate::auth::{self, Access, Role};
use crate::db::{self, accounts::Account};
use crate::repos::{RepoName, Repos};
use crate::Shared;
use axum::body::Bytes;
use axum::extract::{Path, State};
use axum::http::{HeaderMap, Method, StatusCode};
use axum::response::{IntoResponse, Response};
use oxy_core::repo::{PushReport, Repo};
use oxy_server::{Backend, Endpoint, Failure};
use std::sync::Arc;

pub struct HubBackend {
    repos: Arc<Repos>,
    /// Wikis, reached as `OWNER/NAME.wiki` (hub spec §6.11).
    wikis: Arc<Repos>,
    pool: db::Pool,
    /// Site tokens from `hub.toml`: full access, for bootstrapping.
    site_tokens: Vec<String>,
    runtime: tokio::runtime::Handle,
    /// The application, set once it exists (the backend is built first).
    app_hint: Arc<std::sync::Mutex<Option<std::sync::Weak<crate::App>>>>,
    /// The hub's key, which countersigns when it accepted an update a key
    /// it certified signed (oxygenhub#241 step 4c, Oxygen spec §14.7).
    acceptor: Option<Arc<(oxy_core::identity::Identity, oxy_core::identity::SecretKey)>>,
    /// Push protection's scan budget (#268, `[limits]`).
    scan_budget: crate::push_protection::Budget,
    /// Repositories whose push protection scan stopped early, by id: the
    /// note goes on the push log entry of the push that moves references
    /// (the parts of a long push before it move none).
    scan_notes: Arc<std::sync::Mutex<std::collections::HashMap<i64, String>>>,
    /// The keys that signed what a push to each repository (by name)
    /// brought, noted when it is vetted and recorded once it is accepted
    /// (#286, Settings › Keys' "last used").
    signers: Arc<std::sync::Mutex<std::collections::HashMap<String, std::collections::BTreeSet<String>>>>,
}

/// Who a request's credential resolved to.
enum Caller {
    Anonymous,
    /// A site token from the configuration.
    Site,
    Token(Box<Account>, Box<db::accounts::Token>),
    /// A running OxygenFlow job: reads its own repository and the subrepos
    /// its state pins, nothing else (hub spec §13.2).
    Job {
        repo_id: i64,
        also: Vec<i64>,
    },
}

impl HubBackend {
    pub fn new(
        repos: Arc<Repos>,
        wikis: Arc<Repos>,
        pool: db::Pool,
        site_tokens: Vec<String>,
        runtime: tokio::runtime::Handle,
    ) -> HubBackend {
        HubBackend {
            repos,
            wikis,
            pool,
            site_tokens,
            runtime,
            app_hint: Arc::new(std::sync::Mutex::new(None)),
            acceptor: None,
            scan_budget: crate::push_protection::Budget::default(),
            scan_notes: Arc::default(),
            signers: Arc::default(),
        }
    }

    /// Scan pushes for secrets within `budget` (#268).
    pub fn with_scan_budget(mut self, budget: crate::push_protection::Budget) -> Self {
        self.scan_budget = budget;
        self
    }

    /// Who pushes, for the push log: the account (id 0 for a site token)
    /// and, for an agent's token, the agent's name.
    fn pusher(&self, headers: &HeaderMap) -> (Option<(i64, String)>, String) {
        match self.caller(headers) {
            Ok(Caller::Token(account, token)) => {
                (Some((account.id, account.username)), if token.agent { token.name.clone() } else { String::new() })
            }
            Ok(Caller::Site) => (Some((0, "site token".to_string())), String::new()),
            _ => (None, String::new()),
        }
    }

    /// Countersign accepted updates by keys this hub certified with `identity`.
    pub fn with_acceptor(mut self, identity: oxy_core::identity::Identity, secret: oxy_core::identity::SecretKey) -> Self {
        self.acceptor = Some(Arc::new((identity, secret)));
        self
    }

    /// The slot the application registers itself in after construction.
    pub fn app_slot(&self) -> Arc<std::sync::Mutex<Option<std::sync::Weak<crate::App>>>> {
        self.app_hint.clone()
    }

    fn block<T>(&self, f: impl std::future::Future<Output = T>) -> T {
        tokio::task::block_in_place(|| self.runtime.block_on(f))
    }

    fn row(&self, name: &RepoName) -> Result<db::repos::RepoRow, Failure> {
        let pool = self.pool.clone();
        match self.block(db::repos::get(&pool, &name.owner, &name.name)) {
            Ok(Some(r)) => Ok(r),
            // A transferred repository still answers to its old name (#108).
            Ok(None) => match self.block(db::repos::moved_to(&pool, &name.owner, &name.name)) {
                Ok(Some(r)) => Ok(r),
                Ok(None) => Err(not_found()),
                Err(e) => Err(Failure(StatusCode::INTERNAL_SERVER_ERROR, e.to_string())),
            },
            Err(e) => Err(Failure(StatusCode::INTERNAL_SERVER_ERROR, e.to_string())),
        }
    }

    fn caller(&self, headers: &HeaderMap) -> Result<Caller, Failure> {
        let Some(presented) = oxy_server::bearer_token(headers) else { return Ok(Caller::Anonymous) };
        if self.site_tokens.iter().any(|t| auth::mac_equal(t, presented)) {
            return Ok(Caller::Site);
        }
        let hash = auth::token_hash(presented);
        let pool = self.pool.clone();
        if presented.starts_with(crate::flow::JOB_TOKEN_PREFIX) {
            let job = self.block(db::flow::job_by_token(&pool, &hash, crate::flow::JOB_TOKEN_MAX_AGE_MS)).map_err(internal)?;
            let Some(job) = job else { return Err(Failure(StatusCode::UNAUTHORIZED, "this job token is not valid".into())) };
            let run = self.block(db::flow::run_by_id(&pool, job.run_id)).map_err(internal)?;
            let Some(run) = run else { return Err(Failure(StatusCode::UNAUTHORIZED, "this job token is not valid".into())) };
            return Ok(Caller::Job { repo_id: run.repo_id, also: job.also_read_ids() });
        }
        // An agent's grant token (keyless agents, #241): its owner's
        // account, as an agent's token, writing when the grant lets its
        // keys move references. Revoking the grant stops it at once.
        let grant_token = presented.starts_with(crate::keyless::GRANT_TOKEN_PREFIX);
        let token = if grant_token {
            let grant = self.block(db::accounts::agent_grant_by_token(&pool, &hash)).map_err(internal)?;
            let Some(g) = grant.filter(|g| g.live(db::now_ms())) else {
                return Err(Failure(StatusCode::UNAUTHORIZED, "this agent grant is not valid (revoked, ended or unknown)".into()));
            };
            let _ = self.block(db::accounts::touch_agent_grant(&pool, g.id));
            Some(db::accounts::Token {
                id: 0,
                account_id: g.account_id,
                name: g.agent.clone(),
                hash: String::new(),
                scopes: if g.scope.split(',').any(|s| s.trim() == "refs") { "repo:read,repo:write" } else { "repo:read" }.into(),
                repos: String::new(),
                created_ms: g.created_ms,
                expires_ms: Some(g.expires_ms),
                last_used_ms: g.last_used_ms,
                renew_ms: None,
                agent: true,
            })
        } else {
            self.block(db::accounts::token_by_hash(&pool, &hash)).map_err(internal)?
        };
        let Some(token) = token else { return Err(Failure(StatusCode::UNAUTHORIZED, "unknown token".into())) };
        if token.expired(db::now_ms()) {
            return Err(Failure(StatusCode::UNAUTHORIZED, "this token has expired".into()));
        }
        let account = self.block(db::accounts::account_by_id(&pool, token.account_id)).map_err(internal)?;
        let Some(account) = account else { return Err(Failure(StatusCode::UNAUTHORIZED, "unknown token".into())) };
        if account.banned {
            return Err(Failure(StatusCode::UNAUTHORIZED, "this account is banned".into()));
        }
        if account.disabled {
            return Err(Failure(StatusCode::UNAUTHORIZED, "this account is disabled; contact the hub's administrator".into()));
        }
        if !grant_token {
            let _ = self.block(db::accounts::touch_token(&pool, token.id));
        }
        Ok(Caller::Token(Box::new(account), Box::new(token)))
    }
}

fn internal(e: sqlx::Error) -> Failure {
    Failure(StatusCode::INTERNAL_SERVER_ERROR, e.to_string())
}

fn not_found() -> Failure {
    Failure(StatusCode::NOT_FOUND, "no such repository".into())
}

impl Backend for HubBackend {
    fn open(&self, name: &str) -> Result<Repo, Failure> {
        let parsed = RepoName::parse(name).ok_or_else(not_found)?;
        // `NAME.wiki` is the repository's wiki (§6.11); the row knows the
        // current name, so a transferred repository still answers to its
        // old one (#108).
        let (parent, is_wiki) = crate::wiki::parent_of(&parsed);
        let row = self.row(&parent)?;
        let name = RepoName::new(&row.owner, &row.name).ok_or_else(not_found)?;
        let store = if is_wiki { &self.wikis } else { &self.repos };
        let repo = store.open(&name).map_err(|e| match e {
            crate::repos::RepoError::NotFound => not_found(),
            other => Failure(StatusCode::INTERNAL_SERVER_ERROR, other.to_string()),
        })?;
        Ok(match &self.acceptor {
            Some(a) => repo.with_acceptor(a.0.clone(), a.1.clone()),
            None => repo,
        })
    }

    fn authorize(&self, name: &str, access: oxy_server::Access, headers: &HeaderMap) -> Result<(), Failure> {
        let parsed = RepoName::parse(name).ok_or_else(not_found)?;
        let (parent, _) = crate::wiki::parent_of(&parsed);
        let row = self.row(&parent)?;
        let caller = self.caller(headers)?;
        if let Caller::Job { repo_id, also } = &caller {
            // Only reads: of the job's own repository, and of the subrepos its
            // state pins that its pusher could read when the job was claimed.
            return match access {
                oxy_server::Access::Read if *repo_id == row.id || also.contains(&row.id) => Ok(()),
                oxy_server::Access::Read => Err(not_found()),
                oxy_server::Access::Push => Err(Failure(StatusCode::FORBIDDEN, "a job token cannot push".into())),
            };
        }
        let (role, scope_ok) = match &caller {
            Caller::Job { .. } => unreachable!(),
            Caller::Site => (Some(Role::Admin), true),
            Caller::Anonymous => (self.block(auth::roles::role_for(&self.pool, None, &row)).map_err(internal)?, true),
            Caller::Token(account, token) => {
                let role = self.block(auth::roles::role_for(&self.pool, Some(account), &row)).map_err(internal)?;
                let needed = match access {
                    oxy_server::Access::Read => {
                        token.has_scope("repo:read") || token.has_scope("repo:write") || token.has_scope("repo:admin")
                    }
                    oxy_server::Access::Push => token.has_scope("repo:write") || token.has_scope("repo:admin"),
                };
                let current = format!("{}/{}", row.owner, row.name);
                let covered = token.covers(&parsed.to_string()) || token.covers(&current) || token.covers(&parent.to_string());
                (role, needed && covered)
            }
        };
        // An unreadable repository does not exist for the caller (hub spec §4.4).
        if !auth::roles::allows(role, Access::Read) {
            return Err(not_found());
        }
        // A wiki is made, empty and unbound, when someone who may write to
        // its repository first reaches it, so the first page can be pushed
        // as well as written on the web (§6.11); the push gives it its id.
        let (_, is_wiki) = crate::wiki::parent_of(&parsed);
        let may_write = match &caller {
            Caller::Site => true,
            Caller::Token(_, token) => token.has_scope("repo:write") || token.has_scope("repo:admin"),
            _ => false,
        };
        if is_wiki && may_write && auth::roles::allows(role, Access::Write) && !row.archived {
            if let Some(name) = RepoName::new(&row.owner, &row.name) {
                if !self.wikis.exists(&name) {
                    self.wikis
                        .create(&name, None)
                        .map_err(|e| Failure(StatusCode::INTERNAL_SERVER_ERROR, format!("make the wiki: {e}")))?;
                }
            }
        }
        match access {
            oxy_server::Access::Read if scope_ok => Ok(()),
            oxy_server::Access::Read => Err(Failure(StatusCode::FORBIDDEN, "this token cannot read repositories".into())),
            oxy_server::Access::Push => {
                if matches!(caller, Caller::Anonymous) {
                    return Err(Failure(StatusCode::UNAUTHORIZED, "a personal access token is required to push".into()));
                }
                if !auth::roles::allows(role, Access::Write) {
                    return Err(Failure(StatusCode::FORBIDDEN, "you do not have write access to this repository".into()));
                }
                if row.archived {
                    return Err(Failure(StatusCode::FORBIDDEN, "this repository is archived and read-only".into()));
                }
                if !scope_ok {
                    return Err(Failure(StatusCode::FORBIDDEN, "this token lacks the repo:write scope for this repository".into()));
                }
                Ok(())
            }
        }
    }

    fn default_bookmark(&self, name: &str) -> Option<String> {
        let parsed = RepoName::parse(name)?;
        if crate::wiki::parent_of(&parsed).1 {
            return Some(crate::wiki::BOOKMARK.into());
        }
        self.row(&parsed).ok().map(|r| r.default_bookmark)
    }

    fn add_log(&self, name: &str, refs: &mut oxy_core::protocol::RefsResponse) {
        let Some(app) = self.app_hint.lock().unwrap().as_ref().and_then(|w| w.upgrade()) else { return };
        let row = RepoName::parse(name).and_then(|p| self.row(&crate::wiki::parent_of(&p).0).ok()).map(|r| r.id);
        let backfill = crate::tlog::add_log(&app, row, refs);
        if !backfill.is_empty() {
            self.block(crate::tlog::log(&app, backfill));
        }
    }

    /// Tailor the refs answer to the client asking (oxygen#131): one that
    /// does not declare `certified-tags` in `Oxy-Features` (oxy 0.5.0 and
    /// older) is not sent a tag that only a certified key's certificate
    /// authorizes, which it would refuse and, with it, the whole clone.
    fn adjust_refs(&self, _name: &str, repo: &Repo, refs: &mut oxy_core::protocol::RefsResponse, headers: &HeaderMap) {
        oxy_server::tailor_refs(repo, refs, headers);
    }

    /// Push protection (hub spec §15.4, #268): a push adding what looks
    /// like a secret is refused, saying where, unless the repository turned
    /// it off or a maintainer allowed that finding's fingerprint.
    fn vet_push(
        &self,
        name: &str,
        repo: &Repo,
        bundle: &oxy_core::protocol::Bundle,
        updates: &[oxy_core::protocol::RefUpdate],
        headers: &HeaderMap,
    ) -> Result<(), Failure> {
        let keys: std::collections::BTreeSet<String> = bundle
            .changes
            .iter()
            .flat_map(|(_, sigs)| sigs.iter())
            .chain(bundle.attests.iter().flat_map(|a| a.sigs.iter()))
            .map(|sig| sig.key.to_hex())
            .collect();
        if !keys.is_empty() {
            self.signers.lock().unwrap().entry(name.to_string()).or_default().extend(keys);
        }
        if bundle.changes.is_empty() {
            return Ok(());
        }
        let parsed = RepoName::parse(name).ok_or_else(not_found)?;
        let row = self.row(&crate::wiki::parent_of(&parsed).0)?;
        if !row.push_protection {
            return Ok(());
        }
        let scan = crate::push_protection::scan(bundle, self.scan_budget);
        if let Some(note) = scan.note() {
            tracing::info!("push to {name}: {note}");
            self.scan_notes.lock().unwrap().insert(row.id, note);
        }
        if scan.findings.is_empty() {
            return Ok(());
        }
        let allowed = self.block(db::push_protection::allowed(&self.pool, row.id)).map_err(internal)?;
        let mut findings: Vec<_> = scan.findings.into_iter().filter(|f| !allowed.contains(&f.fingerprint)).collect();
        if findings.is_empty() {
            return Ok(());
        }
        crate::push_protection::locate(repo, bundle, &mut findings);
        let (pusher, via) = self.pusher(headers);
        let base = self.app_hint.lock().unwrap().as_ref().and_then(|w| w.upgrade()).map(|a| a.config.base_url().to_string());
        let settings =
            if row.is_drop() { String::new() } else { format!("{}/{}/{}/settings", base.unwrap_or_default(), row.owner, row.name) };
        let text = crate::push_protection::refusal(&findings, &settings, &via);
        // Recorded for the maintainers (Settings › Push protection) and in
        // the push log, as every refused push is.
        let records: Vec<db::push_protection::NewFinding> = findings
            .iter()
            .map(|f| db::push_protection::NewFinding {
                rule: &f.rule,
                description: &f.description,
                path: &f.path,
                line: f.line as i64,
                fingerprint: &f.fingerprint,
                excerpt: &f.excerpt,
            })
            .collect();
        let who = pusher.as_ref().map(|(id, n)| (*id, n.as_str()));
        if let Err(e) = self.block(db::push_protection::record(&self.pool, row.id, who.map(|w| w.1).unwrap_or(""), &via, &records)) {
            tracing::warn!("recording push protection findings for {name}: {e}");
        }
        let reason = format!(
            "push protection: {} possible secret{} ({}){}",
            findings.len(),
            if findings.len() == 1 { "" } else { "s" },
            findings.iter().take(3).map(|f| format!("{}:{}", f.path, f.line)).collect::<Vec<_>>().join(", "),
            if row.is_drop() { "" } else { "; see Settings › Push protection" }
        );
        let refs: Vec<String> = if updates.is_empty() {
            vec![oxy_core::repo::OBJECTS_RESULT.to_string()]
        } else {
            updates.iter().map(|u| u.name.clone()).collect()
        };
        let outcomes: Vec<db::repos::PushOutcome> = refs
            .into_iter()
            .map(|reference| db::repos::PushOutcome {
                reference,
                ok: false,
                reason: reason.clone(),
                old_state: String::new(),
                new_state: String::new(),
                heads: 0,
                changes: 0,
            })
            .collect();
        if let Err(e) = self.block(db::repos::record_push_via(&self.pool, row.id, who, &via, &outcomes)) {
            tracing::warn!("recording refused push to {name}: {e}");
        }
        tracing::info!("push to {name} refused by push protection: {} finding(s)", findings.len());
        Err(Failure(StatusCode::UNPROCESSABLE_ENTITY, text))
    }

    /// A tag a release names is not withdrawn (#301): the release would
    /// point at nothing. Delete the release first.
    fn vet_tag_withdrawal(&self, name: &str, tag: &str) -> Result<(), String> {
        let Some(parsed) = RepoName::parse(name) else { return Ok(()) };
        let Ok(row) = self.row(&parsed) else { return Ok(()) };
        match self.block(db::releases::get(&self.pool, row.id, tag)) {
            Ok(Some(rel)) => Err(format!("the release {:?} uses tags/{tag}: delete the release first", rel.title)),
            Ok(None) => Ok(()),
            Err(e) => Err(format!("could not check the releases of {name}: {e}")),
        }
    }

    /// A push that moved the bookmark of a proposal that has landed, with
    /// none open from it now, says so beside the bookmark (#329): those
    /// changes reach no target until a new proposal takes them.
    fn push_notes(&self, name: &str, report: &PushReport, _headers: &HeaderMap) -> Vec<(String, String)> {
        let Some(parsed) = RepoName::parse(name) else { return Vec::new() };
        if report.updated.is_empty() || crate::wiki::parent_of(&parsed).1 {
            return Vec::new();
        }
        let Ok(row) = self.row(&parsed) else { return Vec::new() };
        let Ok(proposals) = self.block(db::proposals::list(&self.pool, row.id, &[])) else { return Vec::new() };
        let moved: Vec<&str> = report.updated.iter().map(|u| u.name.as_str()).collect();
        landed_notes(&proposals, row.id, &moved)
    }

    fn after_push(&self, name: &str, report: &PushReport, headers: &HeaderMap) {
        // The keys that signed what it brought were used (#286): linked
        // ones show it under Settings › Keys.
        let signers = self.signers.lock().unwrap().remove(name);
        if let Some(keys) = signers.filter(|_| report.results.iter().any(|r| r.ok)) {
            let pool = self.pool.clone();
            let keys: Vec<String> = keys.into_iter().collect();
            tokio::spawn(async move {
                if let Err(e) = db::accounts::identities_used(&pool, &keys).await {
                    tracing::warn!("noting keys used: {e}");
                }
            });
        }
        let Some(parsed) = RepoName::parse(name) else { return };
        // A wiki push starts no run and rings no webhook.
        if crate::wiki::parent_of(&parsed).1 {
            return;
        }
        let Ok(row) = self.row(&parsed) else { return };
        let Some(parsed) = RepoName::new(&row.owner, &row.name) else { return };
        let (pusher, via) = self.pusher(headers);
        let pool = self.pool.clone();
        // A part of a long push (objects alone) moves nothing: the push it
        // belongs to is recorded when its last part names the references.
        let mut results: Vec<db::repos::PushOutcome> = report
            .results
            .iter()
            .filter(|r| r.name != oxy_core::repo::OBJECTS_RESULT)
            .map(|r| {
                let updated = report.updated.iter().find(|u| u.name == r.name);
                db::repos::PushOutcome {
                    reference: r.name.clone(),
                    ok: r.ok,
                    reason: r.reason.clone(),
                    old_state: updated.map(|u| u.old_state.to_hex()).unwrap_or_default(),
                    new_state: updated.map(|u| u.new_state.to_hex()).unwrap_or_default(),
                    heads: updated.map(|u| u.heads.len() as i64).unwrap_or(0),
                    changes: if r.ok { report.imported.len() as i64 } else { 0 },
                }
            })
            .collect();
        // Push protection's scan stopped early on this push or a part of it
        // (#268): its accepted references say so in the log.
        if !results.is_empty() {
            if let Some(note) = self.scan_notes.lock().unwrap().remove(&row.id) {
                for r in results.iter_mut().filter(|r| r.ok) {
                    r.reason = if r.reason.is_empty() { note.clone() } else { format!("{}; {note}", r.reason) };
                }
            }
        }
        // Messages of the changes stored: `closes #N` when the default
        // bookmark moved, and `#N` references on any bookmark (hub spec §8).
        let bookmarks: Vec<String> = report
            .updated
            .iter()
            .filter(|u| !u.name.starts_with("tags/") && report.results.iter().any(|r| r.ok && r.name == u.name))
            .map(|u| u.name.clone())
            .collect();
        let imported: Vec<(String, String)> = if bookmarks.is_empty() {
            vec![]
        } else {
            self.repos
                .open(&parsed)
                .map(|r| {
                    report.imported.iter().filter_map(|h| r.store().get_change_meta(h).ok().map(|c| (h.to_hex(), c.message))).collect()
                })
                .unwrap_or_default()
        };
        let messages: Vec<(String, String)> = if bookmarks.contains(&row.default_bookmark) { imported.clone() } else { vec![] };
        // Authors whose identity is a successor (a rotation, a renewal, a
        // delegation) of an earlier one: each step kept the key pair, so
        // if the chain reaches a linked key it is the same person, and the
        // successor is linked too, without re-uploading it (hub spec §5.3).
        let successors: Vec<(String, String, String, Vec<u8>, String)> = self
            .repos
            .open(&parsed)
            .map(|r| {
                let mut seen = std::collections::HashSet::new();
                let mut out = Vec::new();
                for h in &report.imported {
                    let Ok(meta) = r.store().get_change_meta(h) else { continue };
                    let mut key = meta.author;
                    while !key.is_zero() && seen.insert(key) {
                        let Ok(id) = r.store().get_identity(&key) else { break };
                        let Some(prev) = id.previous else { break };
                        let Ok(before) = r.store().get_identity(&prev) else { break };
                        if before.ed25519_pk != id.ed25519_pk || before.mldsa65_pk != id.mldsa65_pk {
                            break;
                        }
                        out.push((key.to_hex(), id.name.clone(), id.email.clone(), id.encode(), prev.to_hex()));
                        key = prev;
                    }
                }
                out
            })
            .unwrap_or_default();
        let on = if bookmarks.contains(&row.default_bookmark) {
            row.default_bookmark.clone()
        } else {
            bookmarks.first().cloned().unwrap_or_default()
        };
        // Bookmarks that moved, with their new heads, for OxygenFlow runs.
        let moved: Vec<(String, String, Vec<oxy_core::hash::ChangeHash>)> = report
            .updated
            .iter()
            // A bookmark sent again unchanged starts nothing.
            .filter(|u| u.old_state != u.new_state && report.results.iter().any(|r| r.ok && r.name == u.name))
            .map(|u| (u.name.clone(), u.new_state.to_hex(), u.heads.clone()))
            .collect();
        // The transparency log (#238): what the push brought, logged before
        // the push is answered.
        let app = self.app_hint.lock().unwrap().as_ref().and_then(|w| w.upgrade());
        if let Some(app) = app {
            let to_log = self.repos.open(&parsed).map(|r| crate::tlog::push_entries(&r, Some(row.id), report)).unwrap_or_default();
            if !to_log.is_empty() {
                self.block(crate::tlog::log(&app, to_log));
            }
        }
        let app_hint = self.app_hint.clone();
        self.runtime.spawn(async move {
            let who = pusher.as_ref().map(|(id, name)| (*id, name.as_str()));
            if let Err(e) = db::repos::record_push_via(&pool, row.id, who, &via, &results).await {
                tracing::warn!("recording push to {}: {e}", parsed);
            }
            // Oldest first: a link made for one step lets the next step link.
            for (key, name, email, object, prev) in successors.iter().rev() {
                let Ok(None) = db::accounts::identity_by_key(&pool, key).await else { continue };
                let Ok(Some(earlier)) = db::accounts::identity_by_key(&pool, prev).await else { continue };
                match db::accounts::link_identity(&pool, earlier.account_id, key, name, email, object).await {
                    Ok(()) => {
                        tracing::info!("linked identity {} to account {} as a successor of {}", &key[..12], earlier.account_id, &prev[..12]);
                        db::accounts::audit(&pool, None, "identity.rotated", key, &format!("successor of {}", &prev[..12])).await.ok();
                    }
                    Err(e) => tracing::warn!("linking successor identity {}: {e}", &key[..12]),
                }
            }
            // A drop (#279) has no issues, webhooks, mirror, Diffusion or
            // runs: its push is recorded, and that is all.
            if row.is_drop() {
                return;
            }
            let app = app_hint.lock().unwrap().as_ref().and_then(|w| w.upgrade());
            if let Some(app) = app {
                // Issues close from every change newly on the default
                // bookmark, not only those this push imported: a client's
                // landing moves main to changes the hub already held.
                if let Some(r) = results.iter().find(|r| r.ok && r.reference == row.default_bookmark) {
                    let new_heads = moved.iter().find(|(b, _, _)| *b == row.default_bookmark).map(|(_, _, h)| h.clone()).unwrap_or_default();
                    let old = r.old_state.clone();
                    let old_heads = crate::repos::RepoName::new(&row.owner, &row.name);
                    let old_heads = match (old_heads, oxy_core::hash::ObjectRef::from_hex(&old)) {
                        (Some(n), Some(s)) => app.repos.read(&n, move |repo, _| Ok(repo.store().get_state(&s).map(|st| st.heads).unwrap_or_default())).await.unwrap_or_default(),
                        _ => Vec::new(),
                    };
                    crate::web::issues::close_from_arrived(&app, &row, old_heads, new_heads, None).await;
                } else {
                    crate::web::issues::close_from_messages(&app, &row, &messages).await;
                }
                let actor = pusher.as_ref().map(|(_, n)| n.clone()).unwrap_or_default();
                crate::web::issues::references_from_messages(&app, &row, &imported, &on, &actor).await;
                if results.iter().any(|r| r.ok && r.reference == row.default_bookmark) {
                    // Failed, it is done again from the work queue (#276).
                    if crate::mirror::refresh(&app, &row).await.is_err() {
                        crate::queue::queue_mirror(&app, row.id).await;
                    }
                }
                let accepted: Vec<serde_json::Value> = results
                    .iter()
                    .filter(|r| r.ok)
                    .map(|r| serde_json::json!({ "reference": r.reference, "old_state": r.old_state, "new_state": r.new_state, "heads": r.heads, "changes": r.changes }))
                    .collect();
                if !accepted.is_empty() {
                    let actor = pusher.as_ref().map(|(_, n)| n.clone()).unwrap_or_default();
                    crate::hooks::emit(&app, &row, "push", &actor, serde_json::json!({ "updates": accepted })).await;
                }
                let actor = pusher.as_ref().map(|(_, n)| n.clone()).unwrap_or_default();
                // The moved bookmarks' briefs, worked out now so no page
                // waits for them (#145).
                if let Some(name) = crate::repos::RepoName::new(&row.owner, &row.name) {
                    let all: Vec<Vec<oxy_core::hash::ChangeHash>> = moved.iter().map(|(_, _, h)| h.clone()).collect();
                    let warm = app.clone();
                    tokio::spawn(async move {
                        warm.repos.read(&name, move |r, cache| Ok(cache.briefs(r, &all).map(|_| ())?)).await.ok();
                    });
                }
                crate::diffusion::after_push(&app, &row, &moved).await;
                // An upstream's default bookmark moved: consumers following
                // every move are bumped now, those following green when
                // its run passes (#148).
                if let Some((_, _, heads)) = moved.iter().find(|(b, _, _)| *b == row.default_bookmark) {
                    let (app, row, heads) = (app.clone(), row.clone(), heads.clone());
                    tokio::spawn(async move { crate::diffusion::propagate(&app, &row, heads, false).await });
                }
                for (bookmark, state, heads) in &moved {
                    crate::flow::on_push(&app, &row, bookmark, state, heads, &actor).await;
                    crate::web::proposals::settle_after_push(&app, &row, bookmark, &actor).await;
                }
            }
        });
    }
}

/// `/{owner}/{name}/oxy/v1/{endpoint}`.
pub async fn dispatch(
    State(app): State<Shared>,
    Path((owner, name, endpoint)): Path<(String, String, String)>,
    method: Method,
    headers: HeaderMap,
    body: Bytes,
) -> Response {
    let Some(endpoint) = Endpoint::parse(&endpoint) else {
        return Failure(StatusCode::NOT_FOUND, "no such endpoint".into()).into_response();
    };
    let Some(repo) = RepoName::new(&owner, &name) else {
        return not_found().into_response();
    };
    let _guard = match endpoint {
        Endpoint::Push | Endpoint::Checkpoint => Guard::Write(app.repos.write_guard(&repo).await),
        _ => Guard::Read(app.repos.read_guard(&repo).await),
    };
    let (name, received) = (endpoint.name(), body.len() as u64);
    let res = app.server.handle(repo.to_string(), endpoint, &method, &headers, body).await;
    let sent = axum::body::HttpBody::size_hint(res.body()).exact().unwrap_or(0);
    crate::metrics::protocol_bytes(name, received, sent);
    res
}

/// `/{owner}/drops/{id}/oxy/v1/{endpoint}`: a drop is a repository to the
/// wire protocol, named `owner/drops/ID` (#279).
pub async fn dispatch_drop(
    State(app): State<Shared>,
    Path((owner, id, endpoint)): Path<(String, String, String)>,
    method: Method,
    headers: HeaderMap,
    body: Bytes,
) -> Response {
    let Some(name) = RepoName::drop(&owner, &id) else {
        return not_found().into_response();
    };
    dispatch(State(app), Path((name.owner, name.name, endpoint)), method, headers, body).await
}

enum Guard {
    Read(#[allow(dead_code)] crate::repos::ReadGuard),
    Write(#[allow(dead_code)] crate::repos::WriteGuard),
}

/// For each bookmark in `moved` whose latest proposal in this repository
/// (`repo_id`) has landed and none is open from it, a note with what to do
/// (#329).
fn landed_notes(proposals: &[db::proposals::ProposalRow], repo_id: i64, moved: &[&str]) -> Vec<(String, String)> {
    let mut notes = Vec::new();
    for bookmark in moved {
        let from: Vec<&db::proposals::ProposalRow> = proposals
            .iter()
            .filter(|p| p.source_kind == "bookmark" && p.source_repo_id.is_none_or(|id| id == repo_id) && p.source_bookmark == *bookmark)
            .collect();
        if from.iter().any(|p| p.state == "open") {
            continue;
        }
        if let Some(p) = from.iter().filter(|p| p.state == "landed").max_by_key(|p| p.landed_ms) {
            notes.push((
                bookmark.to_string(),
                format!(
                    "its proposal #{} has landed in {}, and these changes are not on {}: open a new proposal with `oh proposal create --from {bookmark}`",
                    p.number, p.target, p.target
                ),
            ));
        }
    }
    notes
}

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

    fn proposal(number: i64, state: &str, source: &str, landed_ms: i64) -> db::proposals::ProposalRow {
        db::proposals::ProposalRow {
            id: number,
            repo_id: 1,
            number,
            state: state.into(),
            target: "develop".into(),
            source_kind: "bookmark".into(),
            source_bookmark: source.into(),
            landed_ms,
            ..Default::default()
        }
    }

    #[test]
    fn a_push_to_a_landed_proposals_bookmark_says_what_to_do() {
        let ps = [proposal(3, "landed", "audio-irq", 10), proposal(5, "landed", "audio-irq", 20), proposal(7, "open", "camera", 0)];
        let notes = landed_notes(&ps, 1, &["audio-irq", "camera", "fresh"]);
        assert_eq!(notes.len(), 1, "{notes:?}");
        assert_eq!(notes[0].0, "audio-irq");
        assert!(
            notes[0].1.contains("#5 has landed in develop") && notes[0].1.contains("oh proposal create --from audio-irq"),
            "{}",
            notes[0].1
        );
        // An open proposal from it takes the changes: nothing to say.
        let reopened = [proposal(3, "landed", "audio-irq", 10), proposal(9, "open", "audio-irq", 0)];
        assert!(landed_notes(&reopened, 1, &["audio-irq"]).is_empty());
        // A fork's proposal names another repository's bookmark.
        let mut fork = proposal(4, "landed", "audio-irq", 10);
        fork.source_repo_id = Some(2);
        assert!(landed_notes(&[fork], 1, &["audio-irq"]).is_empty());
    }
}
`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet
Coverage / crates/hub/src/protocol.rs 94.9%

Keyboard
