//! The gateway: writes on owners' behalf, for collaborators and for batches.
//!
//! The gateway is an OAuth client of its owners' PDSes. An owner delegates
//! (`dev.knotless.putDelegation`) by approving the gateway's authorization
//! request, which grants it the knotless records and blobs and nothing else
//! of the account, and the gateway keeps the session. After that, anyone on a repository's collaborator list can push
//! (`dev.knotless.pushRefs`) or commit (`dev.knotless.createCommit`) with a service
//! token from their own PDS, and the gateway writes as the owner.
//!
//! Its operator's bots (knotless-run, knotless-triage) log in through it
//! too, at `/oauth/bot`, as a client of their own asking for all of
//! knotless's scope: its session lasts as a confidential client's does,
//! and reads private repositories as an app password cannot.
//!
//! Commits to one branch queue up while a write to it is in flight. The
//! next write takes the whole queue: each commit is built on the one before
//! it, and the chain goes out as one ref update. Sixteen agents committing
//! to one branch cost about two writes, not sixteen rounds of losing a race.

use crate::{App, AppError};
use axum::Json;
use axum::body::Bytes;
use axum::extract::{Query, State};
use axum::http::{HeaderMap, StatusCode, header};
use axum::response::Redirect;
use knotless::commit::{self, Committed, Spec};
use knotless::forge::Target;
use knotless::gateway::{
    self, AWAIT_DELEGATION, COMMIT, CommitRequest, CommitResponse, DELEGATE, PUSH, PushResults,
    RefResult,
};
use knotless::landing;
use knotless::manage;
use knotless::mirror::Mirror;
use knotless::push::{self, Update};
use knotless::queue;
use knotless::records::RepoRecord;
use knotless::remote::{Remote, Snapshot};
use knotless::session::{self, Session};
use knotless::url::RepoUrl;
use knotless::{identity, oauth, org};
use serde::Deserialize;
use serde_json::{Value, json};
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use tokio::sync::{oneshot, watch};

/// How many times a batch is rebuilt on a branch that moved under it.
const RETRIES: u32 = 8;

pub struct Gateway {
    pub did: String,
    pub endpoint: String,
    /// Batches written, each one ref update however many commits it holds.
    pub writes: AtomicU64,
    /// Batches rebuilt because the branch moved under them.
    pub rebuilt: AtomicU64,
    /// Where browsers and authorization servers reach the gateway as an
    /// OAuth client: the endpoint, unless that is served elsewhere too.
    pub oauth_url: String,
    /// How long `getDelegation` waits for an approval before it says to
    /// ask again.
    pub wait: Duration,
    repos: Mutex<HashMap<(String, String), Lock>>,
    branches: Mutex<HashMap<(String, String, String), Arc<Branch>>>,
    /// Delegations waiting for their owners' approval, by state.
    delegations: Mutex<HashMap<String, Delegation>>,
    /// Where the bots' sessions are kept, a config directory they share:
    /// none unless `--bots` names one.
    pub bots: Option<PathBuf>,
    /// Bot logins waiting for approval, by state.
    bot_logins: Mutex<HashMap<String, (Instant, oauth::Request)>>,
    /// Logins to the pages waiting for approval, by state.
    pub(crate) logins: Mutex<HashMap<String, crate::login::Pending>>,
    /// Logins the app finished, waiting for it to trade their code.
    pub(crate) app_codes: Mutex<HashMap<String, crate::login::AppCode>>,
    /// The PDS where the pages send someone making an account: one that
    /// takes anyone, [`SIGN_UP_PDS`] unless `--sign-up-pds` names another.
    pub sign_up_pds: String,
}

/// Where the pages send someone making an account, unless told otherwise.
pub const SIGN_UP_PDS: &str = "https://selfhosted.social";

type Outcome = watch::Sender<Option<Result<(), String>>>;

struct Delegation {
    did: String,
    started: Instant,
    /// Taken by the browser's return.
    request: Option<oauth::Request>,
    done: Arc<Outcome>,
}

/// Where the gateway's OAuth client metadata is, its client ID.
pub const METADATA: &str = "/oauth-client-metadata.json";
const CALLBACK: &str = "/oauth/callback";
/// Where the bots' client metadata is, their client ID.
pub const BOT_METADATA: &str = "/oauth/bot-client-metadata.json";
const BOT_CALLBACK: &str = "/oauth/bot-callback";
/// How long a delegation waits for approval before it is dropped.
const ABANDONED: Duration = Duration::from_secs(15 * 60);

/// Held while a repository is written.
type Lock = Arc<tokio::sync::Mutex<()>>;

#[derive(Default)]
struct Branch {
    queue: Mutex<Vec<Pending>>,
    turn: tokio::sync::Mutex<()>,
}

struct Pending {
    spec: Spec,
    /// Who asked for it.
    caller: String,
    reply: oneshot::Sender<Result<CommitResponse, AppError>>,
}

impl Gateway {
    /// A gateway reached at `endpoint`, an `https://` or `http://` URL, as
    /// the `did:web` of its host.
    pub fn new(endpoint: &str) -> Result<Gateway, String> {
        let endpoint = endpoint.trim_end_matches('/');
        let host = endpoint
            .strip_prefix("https://")
            .or_else(|| endpoint.strip_prefix("http://"))
            .filter(|h| !h.is_empty() && !h.contains('/'))
            .ok_or(format!(
                "{endpoint}: a gateway URL is http(s)://host[:port]"
            ))?;
        Ok(Gateway {
            did: format!("did:web:{}", host.replace(':', "%3A")),
            endpoint: endpoint.to_string(),
            writes: Default::default(),
            rebuilt: Default::default(),
            oauth_url: endpoint.to_string(),
            wait: Duration::from_secs(25),
            repos: Mutex::default(),
            branches: Mutex::default(),
            delegations: Mutex::default(),
            bots: None,
            bot_logins: Mutex::default(),
            logins: Mutex::default(),
            app_codes: Mutex::default(),
            sign_up_pds: SIGN_UP_PDS.into(),
        })
    }

    /// The gateway as an OAuth client, signing with the key in its config
    /// directory.
    fn client(&self, ctx: &knotless::env::Ctx) -> Result<oauth::Client, AppError> {
        self.client_at(ctx, METADATA)
    }

    /// The bots' client, with the same key.
    fn bot_client(&self, ctx: &knotless::env::Ctx) -> Result<oauth::Client, AppError> {
        self.client_at(ctx, BOT_METADATA)
    }

    pub(crate) fn client_at(
        &self,
        ctx: &knotless::env::Ctx,
        metadata: &str,
    ) -> Result<oauth::Client, AppError> {
        let key = oauth::client_key(ctx).map_err(AppError::internal)?;
        Ok(oauth::Client::confidential(
            &format!("{}{metadata}", self.oauth_url),
            key,
        ))
    }
}

fn forbidden(msg: impl Into<String>) -> AppError {
    AppError(StatusCode::FORBIDDEN, msg.into())
}

/// A write as an owner their PDS refused: for want of scope, as their
/// grant is older than what the gateway asks for now, saying so.
fn written(err: anyhow::Error) -> AppError {
    match oauth::out_of_scope(&err) {
        true => forbidden(oauth::GRANT_BEHIND),
        false => AppError::upstream(err),
    }
}

fn bad(msg: impl Into<String>) -> AppError {
    AppError(StatusCode::BAD_REQUEST, msg.into())
}

pub(crate) fn gateway(app: &App) -> Result<&Gateway, AppError> {
    app.gateway
        .as_ref()
        .ok_or_else(|| AppError(StatusCode::NOT_FOUND, "this server is not a gateway".into()))
}

/// The DID that signed the request's service token for `lxm`.
pub(crate) async fn caller(app: &App, headers: &HeaderMap, lxm: &str) -> Result<String, AppError> {
    let gw = gateway(app)?;
    caller_for(app, headers, &gw.did, lxm).await
}

/// [`caller`], of a token for `aud`: this server's DID, or one of the
/// services its DID document names.
pub(crate) async fn caller_for(
    app: &App,
    headers: &HeaderMap,
    aud: &str,
    lxm: &str,
) -> Result<String, AppError> {
    let token = headers
        .get(header::AUTHORIZATION)
        .and_then(|v| v.to_str().ok())
        .and_then(|v| v.strip_prefix("Bearer "))
        .ok_or_else(|| {
            AppError(
                StatusCode::UNAUTHORIZED,
                "a service token is required".into(),
            )
        })?;
    knotless::auth::verify(&app.ctx, token, aud, lxm)
        .await
        .map_err(|e| AppError(StatusCode::UNAUTHORIZED, format!("{e:#}")))
}

pub async fn did_document(State(app): State<Arc<App>>) -> Result<Json<Value>, AppError> {
    let gw = gateway(&app)?;
    Ok(Json(json!({
        "@context": ["https://www.w3.org/ns/did/v1"],
        "id": gw.did,
        "service": [{
            "id": gateway::SERVICE,
            "type": "KnotlessGateway",
            "serviceEndpoint": gw.endpoint,
        }, {
            "id": knotless::moderation::LABELER,
            "type": "AtprotoLabeler",
            "serviceEndpoint": gw.endpoint,
        }, {
            "id": crate::viewer::SERVICE,
            "type": "KnotlessAppView",
            "serviceEndpoint": gw.endpoint,
        }],
    })))
}

/// The gateway's OAuth client metadata, which its client ID names.
pub async fn client_metadata(State(app): State<Arc<App>>) -> Result<Json<Value>, AppError> {
    let gw = gateway(&app)?;
    let client = gw.client(&app.ctx)?;
    let back = format!("{}{CALLBACK}", gw.oauth_url);
    Ok(Json(metadata(
        &client,
        "knotless gateway",
        &back,
        gateway::SCOPE,
    )))
}

/// The bots' OAuth client metadata.
pub async fn bot_metadata(State(app): State<Arc<App>>) -> Result<Json<Value>, AppError> {
    let (gw, _) = bots(&app)?;
    let client = gw.bot_client(&app.ctx)?;
    let back = format!("{}{BOT_CALLBACK}", gw.oauth_url);
    Ok(Json(metadata(
        &client,
        "knotless bots",
        &back,
        &oauth::SCOPE,
    )))
}

pub(crate) fn metadata(client: &oauth::Client, name: &str, back: &str, scope: &str) -> Value {
    json!({
        "client_id": client.id,
        "client_name": name,
        "application_type": "web",
        "grant_types": ["authorization_code", "refresh_token"],
        "response_types": ["code"],
        "redirect_uris": [back],
        "scope": scope,
        "token_endpoint_auth_method": "private_key_jwt",
        "token_endpoint_auth_signing_alg": "ES256",
        "dpop_bound_access_tokens": true,
        "jwks": client.jwks(),
    })
}

/// The gateway, and where it keeps the bots' sessions.
fn bots(app: &App) -> Result<(&Gateway, &Path), AppError> {
    let gw = gateway(app)?;
    let dir = gw.bots.as_deref().ok_or_else(|| {
        AppError(
            StatusCode::NOT_FOUND,
            "this server keeps no bots' sessions".into(),
        )
    })?;
    Ok((gw, dir))
}

#[derive(Deserialize)]
pub struct BotLogin {
    account: String,
}

/// Starts logging a bot in as `account`, one of KNOTLESS_BOT_DIDS: sends
/// the browser to its PDS, to approve as that account.
pub async fn bot_login(
    State(app): State<Arc<App>>,
    Query(query): Query<BotLogin>,
) -> Result<Redirect, AppError> {
    let (gw, _) = bots(&app)?;
    let id = identity::resolve(&app.ctx, &query.account)
        .await
        .map_err(AppError::upstream)?;
    if !app.ctx.env.bot_dids.contains(&id.did) {
        return Err(forbidden(format!(
            "{} is not one of this server's bots",
            id.did
        )));
    }
    let did = id.did.clone();
    let back = format!("{}{BOT_CALLBACK}", gw.oauth_url);
    let client = gw.bot_client(&app.ctx)?;
    let request = oauth::authorize(&app.ctx, &client, id, &did, &back, &oauth::SCOPE)
        .await
        .map_err(AppError::upstream)?;
    let to = Redirect::to(&request.url);
    let mut logins = gw.bot_logins.lock().unwrap();
    logins.retain(|_, (started, _)| started.elapsed() < ABANDONED);
    logins.insert(request.state.clone(), (Instant::now(), request));
    Ok(to)
}

/// Where the browser comes back to from a bot's login: keeps the session,
/// and the key it refreshes with, in the bots' config directory.
pub async fn bot_callback(
    State(app): State<Arc<App>>,
    Query(params): Query<HashMap<String, String>>,
) -> Result<String, AppError> {
    let (gw, dir) = bots(&app)?;
    let state = params.get("state").map_or("", String::as_str);
    let waiting = gw.bot_logins.lock().unwrap().remove(state);
    let Some((_, request)) = waiting else {
        return Err(bad("no bot login is waiting for this"));
    };
    let client = gw.bot_client(&app.ctx)?;
    let redeemed = request.redeem(&app.ctx, &client, &params).await;
    let session = redeemed.map_err(|e| forbidden(format!("the bot was not logged in: {e:#}")))?;
    // Its handle names the commits it makes, as the merge queue's.
    let named = oauth::named(&app.ctx, session).await;
    let session = named.map_err(AppError::upstream)?;
    let mut ctx = app.ctx.clone();
    ctx.env.config_dir = dir.to_path_buf();
    let key = oauth::client_key(&app.ctx).map_err(AppError::internal)?;
    oauth::set_client_key(&ctx, &key).map_err(AppError::internal)?;
    session::save(&ctx, &session).map_err(AppError::internal)?;
    crate::telemetry::said("a bot logged in", Some(&session.did));
    Ok(format!(
        "{} is logged in for this server's bots. You can close this tab.",
        session.did
    ))
}

/// Starts delegating the caller's writes here: asks their PDS for a grant,
/// and answers with the page where they approve it.
pub async fn delegate(
    State(app): State<Arc<App>>,
    headers: HeaderMap,
) -> Result<Json<Value>, AppError> {
    let did = caller(&app, &headers, DELEGATE).await?;
    let gw = gateway(&app)?;
    let id = identity::resolve(&app.ctx, &did)
        .await
        .map_err(AppError::upstream)?;
    let back = format!("{}{CALLBACK}", gw.oauth_url);
    let client = gw.client(&app.ctx)?;
    let request = oauth::authorize(&app.ctx, &client, id, &did, &back, gateway::SCOPE)
        .await
        .map_err(AppError::upstream)?;
    let out = json!({ "url": request.url, "state": request.state });
    let mut delegations = gw.delegations.lock().unwrap();
    delegations.retain(|_, d| d.started.elapsed() < ABANDONED);
    delegations.insert(
        request.state.clone(),
        Delegation {
            did,
            started: Instant::now(),
            request: Some(request),
            done: Arc::new(watch::channel(None).0),
        },
    );
    Ok(Json(out))
}

/// Where the owner's browser comes back to: trades the code for the
/// session the gateway writes with.
pub async fn callback(
    State(app): State<Arc<App>>,
    Query(params): Query<HashMap<String, String>>,
) -> Result<String, AppError> {
    let gw = gateway(&app)?;
    let state = params.get("state").map_or("", String::as_str);
    let waiting = gw
        .delegations
        .lock()
        .unwrap()
        .get_mut(state)
        .and_then(|d| Some((d.request.take()?, d.done.clone())));
    let Some((request, done)) = waiting else {
        return Err(bad("no delegation is waiting for this"));
    };
    let result = match request
        .redeem(&app.ctx, &gw.client(&app.ctx)?, &params)
        .await
    {
        Ok(session) => {
            app.event(
                "gateway.delegated",
                "",
                true,
                Duration::ZERO,
                Some(&session.did),
            );
            let saved = session::save(&app.ctx, &session);
            crate::spaces::in_scope_again(&session.did);
            saved
        }
        Err(err) => Err(err),
    };
    let page = match &result {
        Ok(()) => "The gateway now writes for you. You can close this tab.".to_string(),
        Err(err) => format!("The gateway was not delegated to: {err:#}"),
    };
    done.send_replace(Some(result.map_err(|e| format!("{e:#}"))));
    Ok(page)
}

#[derive(Deserialize)]
pub struct Awaited {
    state: String,
}

/// Waits a while for the caller's delegation to be approved.
pub async fn await_delegation(
    State(app): State<Arc<App>>,
    headers: HeaderMap,
    Query(query): Query<Awaited>,
) -> Result<Json<Value>, AppError> {
    let did = caller(&app, &headers, AWAIT_DELEGATION).await?;
    let gw = gateway(&app)?;
    let mut done = gw
        .delegations
        .lock()
        .unwrap()
        .get(&query.state)
        .filter(|d| d.did == did)
        .map(|d| d.done.subscribe())
        .ok_or_else(|| AppError(StatusCode::NOT_FOUND, "no such delegation".into()))?;
    let waited = tokio::time::timeout(gw.wait, done.wait_for(Option::is_some)).await;
    let Ok(Ok(outcome)) = waited else {
        return Ok(Json(json!({ "done": false })));
    };
    let outcome = outcome.clone().expect("waited for an outcome");
    gw.delegations.lock().unwrap().remove(&query.state);
    outcome.map_err(forbidden)?;
    Ok(Json(json!({ "done": true })))
}

/// Whether this server's gateway writes for `did`: it delegated writes
/// here, and the gateway keeps its session.
pub(crate) fn writes_for(app: &App, did: &str) -> bool {
    app.gateway.is_some() && session::load(&app.ctx, did).is_some()
}

/// Whether the gateway writes for `did` under a grant older than what it
/// asks for now, which `did` delegating again brings up.
pub(crate) fn grant_behind(app: &App, did: &str) -> bool {
    let kept = app.gateway.as_ref().and(session::load(&app.ctx, did));
    kept.is_some_and(|s| oauth::behind(&s, gateway::SCOPE))
}

/// The gateway's own context, reading as `owner`, whose session it keeps
/// as they delegated: what decides a push reads a private repository's
/// space as its owner, since nobody is logged in here.
fn as_owner(app: &App, owner: &str) -> knotless::env::Ctx {
    let mut ctx = app.ctx.clone();
    ctx.env.identifier = Some(owner.to_string());
    ctx.env.password = None;
    ctx
}

/// The repository, as its owner, for a caller who may write to it.
async fn writable(
    app: &App,
    caller: &str,
    owner: &str,
    name: &str,
) -> Result<(Remote, Session, Snapshot), AppError> {
    match writer(app, caller, owner, name).await? {
        (_, Writer::Runner) => Err(forbidden(format!(
            "{caller} is not a collaborator on knotless://{owner}/{name}"
        ))),
        (found, Writer::Maintainer) => Ok(found),
    }
}

/// Who a caller who may write to a repository is to it.
enum Writer {
    /// The owner or a collaborator, who may push anything the branch rules
    /// let them.
    Maintainer,
    /// A runner it names that is neither, which lands its merge queue and
    /// pushes nothing else ([`knotless::queue::landing_refusal`]).
    Runner,
}

/// The repository, as its owner, for a caller who may write to it, and who
/// they are to it.
async fn writer(
    app: &App,
    caller: &str,
    owner: &str,
    name: &str,
) -> Result<((Remote, Session, Snapshot), Writer), AppError> {
    let url =
        RepoUrl::parse(&format!("knotless://{owner}/{name}")).map_err(|e| bad(e.to_string()))?;
    let remote = Remote::open(&app.ctx, &url)
        .await
        .map_err(AppError::upstream)?;
    let session = session::ensure(&app.ctx, &remote.identity)
        .await
        .map_err(|_| forbidden(format!("{owner} has not delegated writes to this gateway")))?;
    let snap = remote
        .snapshot(&app.ctx)
        .await
        .map_err(AppError::upstream)?;
    let Some(repo) = &snap.repo else {
        return Err(AppError(
            StatusCode::NOT_FOUND,
            format!("{url}: no such repository"),
        ));
    };
    if let Some(why) = repo.value.archived() {
        return Err(forbidden(why));
    }
    // A member of the organization the owner is pushes too, hidden or not.
    let member = app.org_standing(remote.did(), caller);
    let member = member.is_some_and(|s| s.standing.joined());
    let named = caller == remote.did() || repo.value.is_collaborator(caller);
    let writer = if named || member {
        Writer::Maintainer
    } else if repo.value.is_runner(caller) {
        Writer::Runner
    } else {
        return Err(forbidden(format!(
            "{caller} is not a collaborator on {url}"
        )));
    };
    Ok(((remote, session, snap), writer))
}

async fn mirror(app: &App, remote: &Remote, snap: &Snapshot) -> Result<Mirror, AppError> {
    let key = snap.key().expect("a writable repository exists");
    let mirror = Mirror::open(&app.ctx, remote.did(), key)
        .await
        .map_err(AppError::internal)?;
    mirror
        .refresh(remote, snap)
        .await
        .map_err(AppError::upstream)?;
    crate::aside(&mirror);
    Ok(mirror)
}

pub async fn push(
    State(app): State<Arc<App>>,
    headers: HeaderMap,
    body: Bytes,
) -> Result<Json<PushResults>, AppError> {
    let started = Instant::now();
    let caller = caller(&app, &headers, PUSH).await?;
    let (header, pack) = gateway::split_push(&body).map_err(|e| bad(format!("{e:#}")))?;
    let mut updates = Vec::new();
    for u in &header.updates {
        if u.src.as_deref().is_some_and(|s| !gateway::is_sha(s)) || !u.dst.starts_with("refs/") {
            return Err(bad(format!("bad update of {}", u.dst)));
        }
        updates.push(Update {
            src: u.src.clone(),
            dst: u.dst.clone(),
            force: u.force,
        });
    }
    let lock = gateway(&app)?
        .repos
        .lock()
        .unwrap()
        .entry((header.repo.clone(), header.name.clone()))
        .or_default()
        .clone();
    let _turn = lock.lock().await;
    let found = writer(&app, &caller, &header.repo, &header.name).await?;
    let ((remote, session, snap), writer) = found;
    let mirror = mirror(&app, &remote, &snap).await?;
    if !pack.is_empty() {
        mirror
            .git
            .index_pack(pack)
            .await
            .map_err(|e| bad(format!("{e:#}")))?;
    }
    let tips: Vec<String> = updates.iter().filter_map(|u| u.src.clone()).collect();
    if let Some(sha) = mirror
        .git
        .missing(&tips)
        .await
        .map_err(AppError::internal)?
        .into_iter()
        .next()
    {
        return Err(bad(format!("the pack does not hold {sha}")));
    }
    // Protected branches hold whoever pushes here; the owner can write
    // past the gateway. A collaborator's merge is held to the merge gate,
    // read as the repository is now: as `writer` read it, under this
    // push's turn, so it is not read again, for the gate or a runner's
    // landing. Read as the owner, whose space a private one is in.
    let ctx = as_owner(&app, remote.did());
    let record = snap.repo.as_ref().map(|r| &r.value);
    let reviewed = |r: &RepoRecord| r.rules.iter().any(|x| x.reviews.is_some());
    let gated = record.is_some_and(|r| !r.runners.is_empty() || reviewed(r));
    let gated = gated && caller != remote.did();
    let runner = matches!(writer, Writer::Runner);
    let opened = match runner || gated {
        true => Some(Target::read(&ctx, remote.clone(), snap.clone()).await),
        false => None,
    };
    let opened = opened.transpose().map_err(AppError::upstream)?;
    if let (true, Some(target)) = (runner, &opened) {
        let refusal = queue::landing_refusal(&ctx, target, &mirror.git, &updates).await;
        if let Some(why) = refusal.map_err(AppError::internal)? {
            return Err(forbidden(format!(
                "{caller} runs {}'s checks, and lands its merge queue and pulls set to merge when their checks pass here, nothing else: {why}",
                remote.url
            )));
        }
    }
    let target = opened.filter(|_| gated);
    let (mut allowed, mut refused) = (Vec::new(), Vec::new());
    for update in updates {
        let refusal = gateway::refusal(&mirror.git, &snap, &update, remote.did(), &caller).await;
        let mut refusal = refusal.map_err(AppError::internal)?;
        if let (None, Some(t)) = (&refusal, &target) {
            let held = landing::gate_refusal(&ctx, t, &mirror.git, &update, &caller).await;
            refusal = held.map_err(AppError::upstream)?;
        }
        match refusal {
            Some(reason) => refused.push((update.dst, Err(reason))),
            None => allowed.push(update),
        }
    }
    let updates = allowed;
    let mut report = push::push_by(
        &app.ctx,
        &remote,
        &snap,
        &mirror.git,
        &session.auth(),
        &updates,
        Some(&caller),
    )
    .await
    .map_err(written)?;
    report.extend(refused);
    app.event("gateway.push", "", true, started.elapsed(), Some(&caller));
    // By the owner it wrote for, whoever asked: what delegating led to,
    // once a ref moved.
    let moved = report.iter().any(|(_, res)| res.is_ok());
    if moved {
        let owner = Some(remote.did());
        app.event("gateway.served", "", true, started.elapsed(), owner);
    }
    if moved && remote.space.is_some() && !pack.is_empty() {
        seed(&app, &caller, remote.did(), &snap, body.slice_ref(pack));
    }
    Ok(Json(PushResults {
        results: report
            .into_iter()
            .map(|(name, res)| RefResult {
                name,
                error: res.err(),
            })
            .collect(),
    }))
}

/// Puts `pack`, as `caller` pushed it to `did`'s private repository
/// `snap`, in the mirror of it the pages keep for them, if they keep one:
/// aside, as the push is answered. Their next look then reads its records
/// and not this pack again from the PDS. Only what they sent goes in,
/// theirs to read already, so their mirror is still theirs alone
/// (`docs/design/access.md`). A pack whose bases that mirror lacks is not
/// taken in, and that look fetches it as before.
fn seed(app: &App, caller: &str, did: &str, snap: &Snapshot, pack: Bytes) {
    let key = snap.key().expect("a writable repository exists");
    let ctx = crate::login::ctx_for(app, caller);
    let dir = knotless::mirror::path(&ctx, did, key);
    if dir.join("HEAD").exists() {
        tokio::spawn(async move { knotless::git::Git::new(dir).index_pack(&pack).await.ok() });
    }
}

pub async fn commit(
    State(app): State<Arc<App>>,
    headers: HeaderMap,
    Json(req): Json<CommitRequest>,
) -> Result<Json<CommitResponse>, AppError> {
    let started = Instant::now();
    let caller = caller(&app, &headers, COMMIT).await?;
    if req.branch.is_empty() || req.message.is_empty() {
        return Err(bad("a commit needs a branch and a message"));
    }
    // Checked before queueing, so a batch holds only commits that may land.
    let (remote, _, snap) = writable(&app, &caller, &req.repo, &req.name).await?;
    let record = snap.repo.as_ref().map(|r| &r.value);
    if record.is_some_and(|r| r.held_to_queue(remote.did(), &caller, &req.branch)) {
        return Err(forbidden(format!(
            "{} takes pulls only through the merge queue: commit to another branch and open a pull",
            req.branch
        )));
    }
    if record.is_some_and(|r| r.held_to_reviews(remote.did(), &caller, &req.branch)) {
        return Err(forbidden(format!(
            "{} takes changes only through reviewed pulls: commit to another branch and open a pull",
            req.branch
        )));
    }
    let branch = gateway(&app)?
        .branches
        .lock()
        .unwrap()
        .entry((req.repo.clone(), req.name.clone(), req.branch.clone()))
        .or_default()
        .clone();
    let (reply, answer) = oneshot::channel();
    branch.queue.lock().unwrap().push(Pending {
        spec: req.spec(),
        caller: caller.clone(),
        reply,
    });
    {
        let _turn = branch.turn.lock().await;
        let batch = std::mem::take(&mut *branch.queue.lock().unwrap());
        // Empty when the write before this one took this commit along.
        if !batch.is_empty() {
            land(&app, &caller, &req, batch).await;
        }
    }
    let answered = answer.await.expect("every queued commit is answered");
    let ok = answered.is_ok();
    app.event("gateway.commit", "", ok, started.elapsed(), Some(&caller));
    if ok {
        let owner = Some(remote.did());
        app.event("gateway.served", "", true, started.elapsed(), owner);
    }
    answered.map(Json)
}

/// Builds the batch on the branch and writes it, rebuilding on a branch
/// that moved, and answers every commit in it.
async fn land(app: &App, caller: &str, req: &CommitRequest, batch: Vec<Pending>) {
    match attempt(app, caller, req, &batch).await {
        Ok(results) => {
            for (pending, result) in batch.into_iter().zip(results) {
                let _ = pending.reply.send(result);
            }
        }
        Err(err) => {
            for pending in batch {
                let _ = pending.reply.send(Err(AppError(err.0, err.1.clone())));
            }
        }
    }
}

type Answers = Vec<Result<CommitResponse, AppError>>;

/// Who moves the branch for a batch: its one caller, or no one but the
/// owner when it gathered several.
fn pusher(batch: &[Pending]) -> Option<&str> {
    let first = batch.first()?.caller.as_str();
    batch.iter().all(|p| p.caller == first).then_some(first)
}

async fn attempt(
    app: &App,
    caller: &str,
    req: &CommitRequest,
    batch: &[Pending],
) -> Result<Answers, AppError> {
    let refname = format!("refs/heads/{}", req.branch);
    let mut round = 0;
    loop {
        let (remote, session, snap) = writable(app, caller, &req.repo, &req.name).await?;
        let mirror = mirror(app, &remote, &snap).await?;
        let start = commit::start(&snap, &batch[0].spec);
        let mut tip = start.clone();
        let mut built: Vec<Result<Committed, String>> = Vec::new();
        for pending in batch {
            if let Err(err) = commit::conflicts(&mirror.git, &pending.spec, tip.as_deref()).await {
                built.push(Err(format!("{err:#}")));
                continue;
            }
            // An edit that no longer finds its text fails its own commit,
            // not the batch.
            let spec = match commit::resolve(&mirror.git, tip.as_deref(), &pending.spec).await {
                Ok(spec) => spec,
                Err(err) => {
                    built.push(Err(format!("{err:#}")));
                    continue;
                }
            };
            let sha = commit::build(&app.ctx, &mirror.git, tip.as_deref(), &spec, remote.did())
                .await
                .map_err(AppError::input)?;
            let parent = tip.clone();
            if let Some(sha) = &sha {
                tip = Some(sha.clone());
            }
            built.push(Ok(Committed {
                sha: tip.clone().unwrap_or_default(),
                parent,
                attempts: 1,
                changed: sha.is_some(),
            }));
        }
        let landed = built.iter().flatten().filter(|c| c.changed).count();
        if tip != start {
            let update = Update {
                src: tip.clone(),
                dst: refname.clone(),
                force: true,
            };
            // What a push of this tip would be refused, a commit is: a new
            // branch under a linear rule takes no merge commits from its base.
            let owner = remote.did();
            let held = gateway::refusal(&mirror.git, &snap, &update, owner, caller).await;
            if let Some(why) = held.map_err(AppError::internal)? {
                return Err(forbidden(format!("{refname}: {why}")));
            }
            let report = push::push_by(
                &app.ctx,
                &remote,
                &snap,
                &mirror.git,
                &session.auth(),
                &[update],
                pusher(batch),
            )
            .await
            .map_err(written)?;
            gateway(app)?.writes.fetch_add(1, Ordering::Relaxed);
            // Forced against the generation read above, so the one refusal
            // is that the branch moved: rebuild on where it went.
            if let Err(reason) = &report[0].1 {
                let gave_up = AppError(StatusCode::CONFLICT, format!("{refname}: {reason}"));
                (round < RETRIES).then_some(()).ok_or(gave_up)?;
                gateway(app)?.rebuilt.fetch_add(1, Ordering::Relaxed);
                round += 1;
                continue;
            }
        }
        let answers = built
            .into_iter()
            .map(|c| match c {
                Ok(c) => Ok(CommitResponse {
                    sha: c.sha,
                    parent: c.parent,
                    changed: c.changed,
                    batched: landed,
                }),
                Err(msg) => Err(AppError(StatusCode::CONFLICT, msg)),
            })
            .collect();
        return Ok(answers);
    }
}

/// A call to an organization's gateway naming only the organization.
#[derive(Deserialize)]
pub struct OrgCall {
    org: String,
}

/// A member's call to be taken in, saying which of their join records
/// counts, if they keep both.
#[derive(Deserialize)]
pub struct JoinCall {
    org: String,
    visibility: Option<org::Visibility>,
}

/// An admin's call to an organization's gateway about one account.
#[derive(Deserialize)]
pub struct MemberCall {
    org: String,
    member: String,
    role: Option<String>,
}

/// The organization `org`'s session, which this gateway keeps.
pub(crate) async fn organization(app: &App, org: &str) -> Result<Session, AppError> {
    gateway(app)?;
    let id = identity::resolve(&app.ctx, org)
        .await
        .map_err(AppError::upstream)?;
    session::ensure(&app.ctx, &id)
        .await
        .map_err(|_| forbidden(format!("{org} has not delegated writes to this gateway")))
}

/// The organization `org`'s session, for `caller`, who must be the
/// organization or one of its admins who joined.
async fn administered(app: &App, caller: &str, org: &str) -> Result<Session, AppError> {
    let session = organization(app, org).await?;
    match app.org_admits(&session.did, caller) {
        true => Ok(session),
        false => Err(forbidden(format!("{caller} is not an admin of {org}"))),
    }
}

/// Runs `change` to the organization `session` is, about `member`, alone
/// among changes to it, then writes what it changed into the indexes.
async fn changing<T>(
    app: &App,
    session: &Session,
    member: &str,
    change: impl std::future::Future<Output = anyhow::Result<T>>,
) -> Result<Json<T>, AppError> {
    let lock = app.org_lock(&session.did);
    let _held = lock.lock().await;
    let done = change.await;
    app.reindex_org(&session.did, member).await;
    done.map(Json).map_err(unchanged)
}

/// The account `member` names, by DID, as the organization's records name
/// it.
async fn account(app: &App, member: &str) -> Result<String, AppError> {
    let found = identity::resolve(&app.ctx, member).await;
    found.map(|id| id.did).map_err(|e| bad(format!("{e:#}")))
}

/// A change to an organization the gateway was refused, or refused itself.
fn unchanged(err: anyhow::Error) -> AppError {
    match oauth::out_of_scope(&err) {
        true => forbidden(oauth::GRANT_BEHIND),
        false => AppError(StatusCode::CONFLICT, format!("{err:#}")),
    }
}

pub async fn put_org_member(
    State(app): State<Arc<App>>,
    headers: HeaderMap,
    Json(call): Json<MemberCall>,
) -> Result<Json<org::Invited>, AppError> {
    let caller = caller(&app, &headers, gateway::PUT_ORG_MEMBER).await?;
    let session = administered(&app, &caller, &call.org).await?;
    let member = account(&app, &call.member).await?;
    let role = org::Role::of(call.role.as_deref());
    let gateway = gateway(&app)?;
    let done = org::invite_as(&app.ctx, &session, &member, role, &gateway.did);
    changing(&app, &session, &member, done).await
}

pub async fn remove_org_member(
    State(app): State<Arc<App>>,
    headers: HeaderMap,
    Json(call): Json<MemberCall>,
) -> Result<Json<org::Removed>, AppError> {
    let caller = caller(&app, &headers, gateway::REMOVE_ORG_MEMBER).await?;
    let session = administered(&app, &caller, &call.org).await?;
    let member = account(&app, &call.member).await?;
    let done = org::remove_as(&app.ctx, &session, &member);
    changing(&app, &session, &member, done).await
}

/// Takes the caller in, once they wrote their join record.
pub async fn join_org(
    State(app): State<Arc<App>>,
    headers: HeaderMap,
    Json(call): Json<JoinCall>,
) -> Result<Json<org::Joined>, AppError> {
    let caller = caller(&app, &headers, gateway::JOIN_ORG).await?;
    let session = organization(&app, &call.org).await?;
    let done = org::admit_as(&app.ctx, &session, &caller, call.visibility);
    changing(&app, &session, &caller, done).await
}

/// Takes the caller out: they ended the membership as they took back their
/// join record, and this takes back what it let them read.
pub async fn leave_org(
    State(app): State<Arc<App>>,
    headers: HeaderMap,
    Json(call): Json<OrgCall>,
) -> Result<Json<org::Removed>, AppError> {
    let caller = caller(&app, &headers, gateway::LEAVE_ORG).await?;
    let session = organization(&app, &call.org).await?;
    let done = org::remove_as(&app.ctx, &session, &caller);
    changing(&app, &session, &caller, done).await
}

/// An admin's call to make a repository as an organization.
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct OrgRepoCall {
    org: String,
    name: String,
    #[serde(default)]
    private: bool,
    description: Option<String>,
    default_branch: Option<String>,
    runners: Option<Vec<org::Account>>,
}

/// Makes a repository as the organization, for one of its admins.
pub async fn create_org_repo(
    State(app): State<Arc<App>>,
    headers: HeaderMap,
    Json(call): Json<OrgRepoCall>,
) -> Result<Json<org::Made>, AppError> {
    let caller = caller(&app, &headers, gateway::CREATE_ORG_REPO).await?;
    let session = administered(&app, &caller, &call.org).await?;
    // Made alone, as its members are added: one taken in or out meanwhile
    // would be left out of it, or in.
    let lock = app.org_lock(&session.did);
    let _held = lock.lock().await;
    let url = RepoUrl {
        authority: call.org,
        name: call.name,
    };
    let new = manage::NewRepo {
        private: call.private,
        description: call.description,
        default_branch: call.default_branch,
        runners: call.runners.map(|r| r.into_iter().map(|a| a.did).collect()),
        ..Default::default()
    };
    let made = manage::create(&app.ctx, &url, &new).await;
    made.map(|m| Json(m.into())).map_err(unchanged)
}

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

    #[test]
    fn writes_refused_for_want_of_scope_say_to_delegate_again() {
        let refused = || anyhow::anyhow!("InvalidToken (403): Scope does not permit create");
        for answered in [written(refused()), unchanged(refused())] {
            assert_eq!(answered.0, StatusCode::FORBIDDEN);
            assert_eq!(answered.1, oauth::GRANT_BEHIND);
        }
        assert_eq!(written(anyhow::anyhow!("down")).0, StatusCode::BAD_GATEWAY);
        assert_eq!(unchanged(anyhow::anyhow!("no")).0, StatusCode::CONFLICT);
    }
}