wsb-daemon 0.258.6441

//! Peer continuity over HTTP, for the workbench.
//!
//! Every one of these is the same operation the command line performs, through
//! the same code: pairing, enabling, pausing, resolving a conflict. A button
//! and a command are one operation here, so there is no route that does
//! something the CLI cannot and no CLI verb that the page cannot reach.

use std::sync::Arc;

use axum::{
    extract::{Path, State},
    http::StatusCode,
    routing::{get, post},
    Json, Router,
};
use serde::{Deserialize, Serialize};

use wsb::peer::model::{
    Continuity, Desired, PeerId, RepositoryId, Service, Switch, SyncMode, WorkspaceId,
};

use crate::machine::MachineState;

type Failure = (StatusCode, Json<serde_json::Value>);

fn bad(status: StatusCode, message: impl Into<String>) -> Failure {
    (
        status,
        Json(serde_json::json!({ "error": message.into() })),
    )
}

pub fn routes() -> Router<Arc<MachineState>> {
    Router::new()
        .route(wsb::peer::route::PEERS, get(list))
        .route("/api/v1/peers/pair", post(pair_open))
        .route("/api/v1/peers/pair/join", post(pair_join))
        .route("/api/v1/peers/:peer", axum::routing::delete(remove))
        .route("/api/v1/peers/:peer/workspaces", get(peer_workspaces))
        .route("/api/v1/peers/:peer/sync", post(enable))
        .route("/api/v1/peers/:peer/pause", post(pause))
        .route("/api/v1/peers/:peer/resume", post(resume))
        .route("/api/v1/peers/:peer/status", get(status))
        .route("/api/v1/peers/:peer/conflicts", get(conflicts))
        .route("/api/v1/peers/:peer/diff", get(diff))
        .route("/api/v1/peers/:peer/conflicts/versions", get(versions))
        .route("/api/v1/peers/:peer/conflicts/resolve", post(resolve))
        .route("/api/v1/peers/:peer/services", get(services))
        .route("/api/v1/peers/:peer/services/:service", post(set_service))
        .route("/api/v1/peers/:peer/mode", post(set_mode))
        .route("/api/v1/peers/:peer/align", post(align))
        .route("/api/v1/peers/:peer/deltas", get(deltas))
        .route("/api/v1/peers/:peer/deltas/decide", post(decide))
        .route("/api/v1/peers/:peer/deltas/dismiss", post(dismiss))
        .route("/api/v1/peers/:peer/deltas/reconsider", post(reconsider))
}

#[derive(Debug, Serialize)]
pub struct PeerView {
    pub id: String,
    pub name: String,
    pub fingerprint: String,
    pub address: String,
    pub state: String,
    /// How long `state` has held. A `connecting` that never becomes
    /// `connected` looks identical to a healthy one without this.
    pub state_for_seconds: u64,
    pub detail: Option<String>,
    pub last_seen: Option<u64>,
    pub shared_workspaces: usize,
}

async fn list(State(machine): State<Arc<MachineState>>) -> Result<Json<Vec<PeerView>>, Failure> {
    let peers = machine
        .store
        .peers()
        .map_err(|e| bad(StatusCode::INTERNAL_SERVER_ERROR, e))?;
    let relationships = machine.store.relationships().unwrap_or_default();
    Ok(Json(
        peers
            .into_iter()
            .map(|peer| {
                let liveness = machine.peers.liveness(&peer.id);
                PeerView {
                    shared_workspaces: relationships
                        .iter()
                        .filter(|(id, _, desired)| *id == peer.id && *desired != Desired::Disabled)
                        .count(),
                    state: liveness.state.as_str().to_string(),
                    state_for_seconds: liveness.held_for(),
                    detail: liveness.detail,
                    last_seen: machine.store.last_seen(&peer.id).ok().flatten(),
                    id: peer.id.to_string(),
                    name: peer.name,
                    fingerprint: peer.fingerprint.to_string(),
                    address: peer.address,
                }
            })
            .collect(),
    ))
}

#[derive(Debug, Serialize)]
pub struct PairingView {
    pub code: String,
    pub expires_in_seconds: u64,
    pub machine_name: String,
    pub fingerprint: String,
    pub peer_port: u16,
}

/// Open a pairing window on this machine.
async fn pair_open(State(machine): State<Arc<MachineState>>) -> Result<Json<PairingView>, Failure> {
    // A code read out for a machine nothing can reach is a pairing that
    // cannot complete, and the other side would find out by timing out.
    let peer_port = machine.peers.peer_port();
    if peer_port == 0 {
        let why = machine
            .peers
            .refusal()
            .unwrap_or_else(|| "peer listening is off (--peer-port 0)".to_string());
        return Err(bad(StatusCode::CONFLICT, format!("peers cannot reach this machine, so it cannot pair: {why}")));
    }
    let code = machine
        .peers
        .open_pairing()
        .map_err(|e| bad(StatusCode::INTERNAL_SERVER_ERROR, e))?;
    Ok(Json(PairingView {
        code: code.to_string(),
        expires_in_seconds: wsb::peer::identity::PAIRING_WINDOW.as_secs(),
        machine_name: machine.identity.name().to_string(),
        fingerprint: machine.identity.fingerprint().to_string(),
        peer_port,
    }))
}

#[derive(Debug, Deserialize)]
pub struct JoinRequest {
    pub code: String,
    /// Where the other machine listens, as `host:port`.
    pub address: String,
}

/// Take a code to the machine that issued it.
async fn pair_join(
    State(machine): State<Arc<MachineState>>,
    Json(body): Json<JoinRequest>,
) -> Result<Json<PeerView>, Failure> {
    let code = wsb::peer::identity::PairingCode::parse(&body.code)
        .map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    let address = body.address.clone();
    let peers = machine.peers.clone();
    // Off the runtime: this dials another machine and waits for a handshake.
    let paired = tokio::task::spawn_blocking(move || peers.pair_with(&address, &code))
        .await
        .map_err(|e| bad(StatusCode::INTERNAL_SERVER_ERROR, format!("join error: {e}")))?
        .map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;

    Ok(Json(PeerView {
        id: paired.id.to_string(),
        name: paired.name,
        fingerprint: paired.fingerprint.to_string(),
        address: paired.address,
        state: wsb::peer::model::PeerState::Connected.as_str().to_string(),
        // Connected as of this handshake, which is what pairing just did.
        state_for_seconds: 0,
        detail: None,
        last_seen: Some(paired.paired_at),
        shared_workspaces: 0,
        // Asked on the next connection. The pairing handshake proves who they
        // are; what they RUN is a separate question and this is not the moment
        // it was answered.
    }))
}

async fn remove(
    State(machine): State<Arc<MachineState>>,
    Path(peer): Path<String>,
) -> Result<Json<serde_json::Value>, Failure> {
    let id = PeerId::parse(&peer).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    let removed = machine
        .store
        .remove_peer(&id)
        .map_err(|e| bad(StatusCode::INTERNAL_SERVER_ERROR, e))?;
    // What was observed about a machine that is no longer a peer is not a
    // fact about anything: a re-pairing under the same name must start from
    // nothing rather than from the last relationship's statuses.
    machine.forget_peer(&id);
    Ok(Json(serde_json::json!({ "removed": removed })))
}

/// What both machines have, side by side.
async fn peer_workspaces(
    State(machine): State<Arc<MachineState>>,
    Path(peer): Path<String>,
) -> Result<Json<serde_json::Value>, Failure> {
    let id = PeerId::parse(&peer).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    let mine = machine.summaries().await;
    let theirs = machine
        .peer_inventory(&id)
        .await
        .map_err(|e| bad(StatusCode::SERVICE_UNAVAILABLE, e))?;
    let shared = super::inventory::compare(&mine, &theirs);
    let relationships = machine.store.relationships().unwrap_or_default();
    let files = machine
        .store
        .service(&id, Service::Files)
        .unwrap_or_else(|_| Switch::default_for(&Service::Files));
    Ok(Json(serde_json::json!({
        "workspaces": shared
            .into_iter()
            .map(|workspace| {
                let desired = relationships
                    .iter()
                    .find(|(p, w, _)| *p == id && *w == workspace.id)
                    .map(|(_, _, desired)| *desired)
                    .unwrap_or(Desired::Disabled);
                let mode = machine.store.mode(&id, &workspace.id).unwrap_or_default();
                serde_json::json!({
                    "id": workspace.id.to_string(),
                    "display_name": workspace.display_name,
                    "presence": workspace.presence.as_str(),
                    "compatible": workspace.compatible,
                    "desired": desired.as_str(),
                    // How it would be carried, which is a separate question
                    // from whether it is carried at all — and the one that
                    // decides whether anything moves without being asked.
                    "mode": mode.as_str(),
                    "continuity": Continuity::of(files, desired, mode).as_str(),
                })
            })
            .collect::<Vec<_>>()
    })))
}

#[derive(Debug, Deserialize)]
pub struct WorkspaceBody {
    pub workspace: String,
}

async fn enable(
    State(machine): State<Arc<MachineState>>,
    Path(peer): Path<String>,
    Json(body): Json<WorkspaceBody>,
) -> Result<Json<serde_json::Value>, Failure> {
    set_desired(machine, peer, body.workspace, Desired::Enabled).await
}

async fn pause(
    State(machine): State<Arc<MachineState>>,
    Path(peer): Path<String>,
    Json(body): Json<WorkspaceBody>,
) -> Result<Json<serde_json::Value>, Failure> {
    set_desired(machine, peer, body.workspace, Desired::Paused).await
}

async fn resume(
    State(machine): State<Arc<MachineState>>,
    Path(peer): Path<String>,
    Json(body): Json<WorkspaceBody>,
) -> Result<Json<serde_json::Value>, Failure> {
    set_desired(machine, peer, body.workspace, Desired::Enabled).await
}

async fn set_desired(
    machine: Arc<MachineState>,
    peer: String,
    workspace: String,
    desired: Desired,
) -> Result<Json<serde_json::Value>, Failure> {
    let peer = PeerId::parse(&peer).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    let workspace = WorkspaceId::parse(&workspace).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    if machine.store.peer(&peer).map_err(|e| bad(StatusCode::INTERNAL_SERVER_ERROR, e))?.is_none() {
        return Err(bad(StatusCode::NOT_FOUND, format!("{peer} is not a paired peer")));
    }
    machine
        .store
        .set_desired(&peer, &workspace, desired, super::service::now())
        .map_err(|e| bad(StatusCode::INTERNAL_SERVER_ERROR, e))?;
    Ok(Json(serde_json::json!({ "desired": desired.as_str() })))
}

#[derive(Debug, Deserialize)]
pub struct StatusQuery {
    pub workspace: String,
}

/// What every repository of one relationship is doing.
async fn status(
    State(machine): State<Arc<MachineState>>,
    Path(peer): Path<String>,
    axum::extract::Query(query): axum::extract::Query<StatusQuery>,
) -> Result<Json<serde_json::Value>, Failure> {
    let peer = PeerId::parse(&peer).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    let workspace =
        WorkspaceId::parse(&query.workspace).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    let statuses = machine.statuses(&peer, &workspace);
    let liveness = machine.peers.liveness(&peer);
    Ok(Json(serde_json::json!({
        "peer": peer.to_string(),
        "workspace": workspace.to_string(),
        "peer_state": liveness.state.as_str(),
        "desired": machine.store.desired(&peer, &workspace).unwrap_or(Desired::Disabled).as_str(),
        "mode": machine.store.mode(&peer, &workspace).unwrap_or_default().as_str(),
        "repositories": statuses,
    })))
}

async fn conflicts(
    State(machine): State<Arc<MachineState>>,
    Path(peer): Path<String>,
    axum::extract::Query(query): axum::extract::Query<StatusQuery>,
) -> Result<Json<serde_json::Value>, Failure> {
    let peer = PeerId::parse(&peer).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    let workspace =
        WorkspaceId::parse(&query.workspace).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    let rows = machine
        .store
        .conflicts(&peer, &workspace)
        .map_err(|e| bad(StatusCode::INTERNAL_SERVER_ERROR, e))?;
    Ok(Json(serde_json::json!({
        "conflicts": rows
            .into_iter()
            .map(|row| serde_json::json!({
                "repository": row.repository,
                "path": row.path,
                // What each side HOLDS, said in words rather than left to a
                // consumer to infer from a null hash. A deletion has no
                // content to hash, and "null" is not a state name.
                "here": if row.here_hash.is_some() { "modified" } else { "deleted" },
                "there": if row.there_hash.is_some() { "modified" } else { "deleted" },
                "here_hash": row.here_hash,
                "there_hash": row.there_hash,
                "base_hash": row.base_hash,
                "detected_at": row.detected_at,
            }))
            .collect::<Vec<_>>()
    })))
}

/// What differs between this machine and a peer, path by path.
///
/// NOT a git diff, and the vocabulary keeps them apart: this compares two
/// working trees and says nothing about either one's relationship to a
/// commit. A path appears here because the two machines hold different bytes
/// for it, whether or not either side has ever committed it.
async fn diff(
    State(machine): State<Arc<MachineState>>,
    Path(peer): Path<String>,
    axum::extract::Query(query): axum::extract::Query<StatusQuery>,
) -> Result<Json<serde_json::Value>, Failure> {
    let peer = PeerId::parse(&peer).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    let workspace =
        WorkspaceId::parse(&query.workspace).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    let plans = machine.plans(&peer, &workspace);
    let statuses = machine.statuses(&peer, &workspace);
    Ok(Json(serde_json::json!({
        "peer": peer.to_string(),
        "workspace": workspace.to_string(),
        "repositories": statuses
            .iter()
            .map(|status| {
                let repository = status.repository.as_str().to_string();
                serde_json::json!({
                    "repository": repository,
                    "label": status.repository.label(),
                    "observed": status.observed.as_str(),
                    // A blocked repository has no plan, and an empty list of
                    // moves would read as "nothing differs" — which is
                    // exactly what nobody knows while it is blocked.
                    "planned": status.blocked.is_none(),
                    "moves": plans.get(&repository).cloned().unwrap_or_default(),
                })
            })
            .collect::<Vec<_>>()
    })))
}

#[derive(Debug, Deserialize)]
pub struct VersionsQuery {
    pub workspace: String,
    pub repository: String,
    pub path: String,
}

/// Both versions of one conflicted path, and the diff between them.
///
/// A conflict is settled by a person choosing a side, and this is what they
/// choose between. Read-only: it reads this machine's file, asks the peer for
/// its own over the same authenticated session a transfer uses, and resolves
/// nothing.
async fn versions(
    State(machine): State<Arc<MachineState>>,
    Path(peer): Path<String>,
    axum::extract::Query(query): axum::extract::Query<VersionsQuery>,
) -> Result<Json<serde_json::Value>, Failure> {
    let peer = PeerId::parse(&peer).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    let workspace =
        WorkspaceId::parse(&query.workspace).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    let repository =
        RepositoryId::parse(&query.repository).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    let path = query.path.clone();

    // Off the runtime: this dials the peer and reads files, and both block.
    let state = machine.clone();
    let versions = tokio::task::spawn_blocking(move || {
        super::versions::versions(&state, &peer, &workspace, &repository, &path)
    })
    .await
    .map_err(|e| bad(StatusCode::INTERNAL_SERVER_ERROR, format!("read error: {e}")))?
    // A peer that cannot be reached is why there is no answer, and saying so
    // is the answer: the page offers to wait rather than to choose blind.
    .map_err(|e| bad(StatusCode::SERVICE_UNAVAILABLE, e))?;

    Ok(Json(serde_json::to_value(versions).map_err(|e| {
        bad(StatusCode::INTERNAL_SERVER_ERROR, format!("cannot encode versions: {e}"))
    })?))
}

#[derive(Debug, Deserialize)]
pub struct ResolveRequest {
    pub workspace: String,
    pub repository: String,
    pub path: String,
    /// `here` or `there`. There is no "newest": a conflict is resolved by a
    /// person choosing, and both choices are destructive to the other side.
    pub keep: String,
}

/// Resolve one conflict by choosing a side.
async fn resolve(
    State(machine): State<Arc<MachineState>>,
    Path(peer): Path<String>,
    Json(body): Json<ResolveRequest>,
) -> Result<Json<serde_json::Value>, Failure> {
    let peer = PeerId::parse(&peer).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    let workspace =
        WorkspaceId::parse(&body.workspace).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    let repository =
        RepositoryId::parse(&body.repository).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    let keep = match body.keep.as_str() {
        "here" => super::conflict::Keep::Here,
        "there" => super::conflict::Keep::There,
        other => {
            return Err(bad(
                StatusCode::BAD_REQUEST,
                format!("'{other}' is not a side to keep — 'here' or 'there'"),
            ))
        }
    };
    super::conflict::resolve(&machine, &peer, &workspace, &repository, &body.path, keep)
        .await
        .map_err(|e| bad(StatusCode::CONFLICT, e))?;
    Ok(Json(serde_json::json!({ "resolved": body.path })))
}

/// Which services run with one peer.
///
/// Every service there is, always — including the ones that are on. A page
/// that listed only what was switched off could not offer to switch anything
/// off, and a person looking for "why is this machine not answering for
/// secrets" needs to see the switch that says so.
async fn services(
    State(machine): State<Arc<MachineState>>,
    Path(peer): Path<String>,
) -> Result<Json<wsb::peer::model::PeerServices>, Failure> {
    let peer = PeerId::parse(&peer).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    let switches = machine
        .store
        .services(&peer)
        .map_err(|e| bad(StatusCode::INTERNAL_SERVER_ERROR, e))?;
    // wsb's own services, every service a plugin running here offers, and
    // any service somebody has switched with this peer before — a plugin
    // that is not running now still has a switch somebody may want to see.
    let mut listed: std::collections::BTreeSet<Service> = Service::BUILT_IN.into_iter().collect();
    listed.extend(switches.keys().cloned());
    for registration in machine
        .plugins
        .running()
        .map_err(|e| bad(StatusCode::INTERNAL_SERVER_ERROR, e))?
    {
        listed.extend(registration.services.into_iter().map(Service::Plugin));
    }
    Ok(Json(wsb::peer::model::PeerServices {
        peer: peer.to_string(),
        services: listed
            .iter()
            .map(|service| wsb::peer::model::ServiceSwitch {
                service: service.as_str().to_string(),
                switch: switches
                    .get(service)
                    .copied()
                    .unwrap_or_else(|| Switch::default_for(service)),
                describes: service.describe(),
            })
            .collect(),
    }))
}

#[derive(Debug, Deserialize)]
pub struct SwitchBody {
    /// `on` or `off`.
    pub switch: String,
}

/// Switch one service on or off with one peer.
///
/// One service at a time and nothing else touched: switching file sync off
/// must leave secret sharing exactly as it was, which is the entire reason
/// these are separate rows rather than one relationship state.
async fn set_service(
    State(machine): State<Arc<MachineState>>,
    Path((peer, service)): Path<(String, String)>,
    Json(body): Json<SwitchBody>,
) -> Result<Json<serde_json::Value>, Failure> {
    let peer = PeerId::parse(&peer).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    let service = Service::parse(&service).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    let switch = Switch::parse(&body.switch).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    if machine
        .store
        .peer(&peer)
        .map_err(|e| bad(StatusCode::INTERNAL_SERVER_ERROR, e))?
        .is_none()
    {
        return Err(bad(StatusCode::NOT_FOUND, format!("{peer} is not a paired peer")));
    }
    machine
        .store
        .set_service(&peer, service.clone(), switch, super::service::now())
        .map_err(|e| bad(StatusCode::INTERNAL_SERVER_ERROR, e))?;
    Ok(Json(serde_json::json!({
        "service": service.as_str(),
        "switch": switch.as_str(),
    })))
}

#[derive(Debug, Deserialize)]
pub struct ModeBody {
    pub workspace: String,
    /// `manual` or `auto`.
    pub mode: String,
}

/// How one workspace is carried with one peer: by hand, or by itself.
async fn set_mode(
    State(machine): State<Arc<MachineState>>,
    Path(peer): Path<String>,
    Json(body): Json<ModeBody>,
) -> Result<Json<serde_json::Value>, Failure> {
    let peer = PeerId::parse(&peer).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    let workspace =
        WorkspaceId::parse(&body.workspace).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    let mode = SyncMode::parse(&body.mode).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    if machine
        .store
        .peer(&peer)
        .map_err(|e| bad(StatusCode::INTERNAL_SERVER_ERROR, e))?
        .is_none()
    {
        return Err(bad(StatusCode::NOT_FOUND, format!("{peer} is not a paired peer")));
    }
    machine
        .store
        .set_mode(&peer, &workspace, mode, super::service::now())
        .map_err(|e| bad(StatusCode::INTERNAL_SERVER_ERROR, e))?;
    Ok(Json(serde_json::json!({ "mode": mode.as_str() })))
}

#[derive(Debug, Deserialize)]
pub struct AlignBody {
    pub workspace: String,
    pub repository: String,
    /// The commit to move to. Checked against what the peer was last observed
    /// to be on, so this is "go to where that machine is" and not a general
    /// checkout endpoint reachable through a peer's name.
    pub commit: String,
    /// Consent to lose the working tree's uncommitted changes.
    ///
    /// Asked for rather than assumed: moving a branch rewrites the worktree,
    /// and a person looking at a "different commit" line has not necessarily
    /// remembered what they were editing.
    #[serde(default)]
    pub discard_changes: bool,
    /// Consent to take commits off this branch.
    ///
    /// Separate from `discard_changes` because they are different losses: one
    /// is an edit nobody committed, the other is work that IS committed and
    /// will only be findable in the reflog afterwards. Approving one must
    /// never approve the other.
    #[serde(default)]
    pub discard_commits: bool,
}

/// Put one repository at the commit a peer is on.
///
/// The action behind "you are on a different commit" — the state the peer
/// machinery reports as `head_mismatch` and then refuses to compare anything
/// through, so it is the one blocked reason a person can clear from here.
///
/// What it costs is computed BEFORE anything is touched and refused unless the
/// caller has consented to exactly that cost. The two costs are asked for
/// separately because they are different losses: uncommitted edits, and
/// commits that stop being reachable from this branch.
async fn align(
    State(machine): State<Arc<MachineState>>,
    Path(peer): Path<String>,
    Json(body): Json<AlignBody>,
) -> Result<Json<serde_json::Value>, Failure> {
    let peer = PeerId::parse(&peer).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    let workspace =
        WorkspaceId::parse(&body.workspace).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    let repository =
        RepositoryId::parse(&body.repository).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;

    // The commit must be the one that peer was last seen on. Anything else is
    // a checkout request wearing a peer's name, and this endpoint is not that.
    let observed = machine.statuses(&peer, &workspace);
    let theirs = observed
        .iter()
        .find(|status| status.repository == repository)
        .and_then(|status| status.head.clone())
        .ok_or_else(|| {
            bad(
                StatusCode::CONFLICT,
                format!(
                    "nothing has been observed about {repository} with this machine yet, so \
                     there is no commit of theirs to move to"
                ),
            )
        })?;
    if theirs != body.commit {
        return Err(bad(
            StatusCode::CONFLICT,
            format!(
                "that machine is on {theirs}, not {}. It moved since the page was drawn — \
                 look again before deciding.",
                body.commit
            ),
        ));
    }

    let state = machine.clone();
    let outcome = tokio::task::spawn_blocking(move || {
        align_repository(&state, &workspace, &repository, &theirs, body.discard_changes, body.discard_commits)
    })
    .await
    .map_err(|e| bad(StatusCode::INTERNAL_SERVER_ERROR, format!("align error: {e}")))?;

    match outcome {
        Ok(value) => Ok(Json(value)),
        Err((code, message)) => Err(bad(code, message)),
    }
}

/// The blocking half of [`align`].
fn align_repository(
    machine: &Arc<MachineState>,
    workspace: &WorkspaceId,
    repository: &RepositoryId,
    commit: &str,
    discard_changes: bool,
    discard_commits: bool,
) -> Result<serde_json::Value, (StatusCode, String)> {
    // Blocking on the async accessor, deliberately: this whole function runs
    // on a blocking thread precisely because everything after it does git.
    let (_, local) = futures::executor::block_on(machine.continuity(workspace))
        .map_err(|e| (StatusCode::CONFLICT, e))?
        .ok_or_else(|| (StatusCode::NOT_FOUND, format!("no worktree open here is {workspace}")))?;
    let repo = local
        .repository(repository)
        .ok_or_else(|| {
            (
                StatusCode::NOT_FOUND,
                format!("{repository} is not a repository of this workspace"),
            )
        })?;
    let opened = wsb::git_backend::open_repo(&repo.workdir)
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("{e}")))?;

    // The object has to be here. A peer that has committed without pushing is
    // holding a commit this machine cannot reach, and saying so is the answer
    // — inventing one from the peer's files would be a different commit with
    // the same message.
    let present = wsb::git_backend::has_commit(&opened, commit)
        .map_err(|e| (StatusCode::BAD_REQUEST, format!("{e}")))?;
    if !present {
        let _ = wsb::git_backend::fetch(&opened, Some("origin"));
        let present = wsb::git_backend::has_commit(&opened, commit)
            .map_err(|e| (StatusCode::BAD_REQUEST, format!("{e}")))?;
        if !present {
            return Err((
                StatusCode::CONFLICT,
                format!(
                    "{commit} is not in this repository and origin does not have it either. \
                     That machine has not pushed it, so there is nothing here to move to."
                ),
            ));
        }
    }

    let cost = wsb::git_backend::move_cost(&opened, commit)
        .map_err(|e| (StatusCode::CONFLICT, format!("{e}")))?;
    use wsb::git_backend::MoveCost;
    if matches!(cost, MoveCost::Already) {
        return Ok(serde_json::json!({ "moved": false, "reason": "already on that commit" }));
    }
    if let MoveCost::Discards { commits } = cost {
        if !discard_commits {
            return Err((
                StatusCode::CONFLICT,
                format!(
                    "moving to {commit} takes {commits} commit(s) off this branch. They would \
                     survive only in the reflog. Say so explicitly to proceed."
                ),
            ));
        }
    }

    // The working tree, asked separately. `own` files only: a moved gitlink is
    // not an edit anybody loses, and a container repository is otherwise never
    // alignable at all.
    let dirty = wsb::fleet::git::GitOps::get_uncommitted_own_files(&opened)
        .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("{e}")))?;
    if !dirty.is_empty() && !discard_changes {
        return Err((
            StatusCode::CONFLICT,
            format!(
                "there are uncommitted changes here that moving would overwrite: {}. \
                 Say so explicitly to proceed.",
                wsb::fleet::git::GitOps::describe_uncommitted(&dirty)
            ),
        ));
    }

    wsb::git_backend::move_branch_to(&opened, commit)
        .map_err(|e| (StatusCode::CONFLICT, format!("{e}")))?;

    Ok(serde_json::json!({
        "moved": true,
        "commit": commit,
        "discarded_changes": dirty,
        "discarded_commits": match cost {
            MoveCost::Discards { commits } => commits,
            _ => 0,
        },
    }))
}

/// Every difference with one peer, and what has been said about each.
///
/// The manual-mode surface. `diff` answers the same question for a machine —
/// what would move — while this answers it for a person: which differences are
/// waiting, which are about to move because somebody chose, and which were
/// left alone. All three are the same list, distinguished by what is on each
/// row rather than by three endpoints that could disagree.
async fn deltas(
    State(machine): State<Arc<MachineState>>,
    Path(peer): Path<String>,
    axum::extract::Query(query): axum::extract::Query<StatusQuery>,
) -> Result<Json<serde_json::Value>, Failure> {
    let peer = PeerId::parse(&peer).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    let workspace =
        WorkspaceId::parse(&query.workspace).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    let plans = machine.plans(&peer, &workspace);
    let statuses = machine.statuses(&peer, &workspace);
    let desired = machine
        .store
        .desired(&peer, &workspace)
        .unwrap_or(Desired::Disabled);
    let files = machine
        .store
        .service(&peer, Service::Files)
        .unwrap_or_else(|_| Switch::default_for(&Service::Files));
    let mode = machine.store.mode(&peer, &workspace).unwrap_or_default();
    let liveness = machine.peers.liveness(&peer);

    Ok(Json(serde_json::json!({
        "peer": peer.to_string(),
        "workspace": workspace.to_string(),
        "peer_state": liveness.state.as_str(),
        "desired": desired.as_str(),
        "mode": mode.as_str(),
        "files": files.as_str(),
        "continuity": Continuity::of(files, desired, mode).as_str(),
        "repositories": statuses
            .iter()
            .map(|status| {
                let repository = status.repository.as_str().to_string();
                serde_json::json!({
                    "repository": repository,
                    "label": status.repository.label(),
                    "observed": status.observed.as_str(),
                    "blocked": status.blocked,
                    "undecided": status.undecided,
                    "conflicts": status.conflicts,
                    // Nothing is known about a blocked repository's files —
                    // the comparison never happened — and an empty list would
                    // read as "nothing differs".
                    "compared": status.blocked.is_none(),
                    "deltas": plans.get(&repository).cloned().unwrap_or_default(),
                })
            })
            .collect::<Vec<_>>(),
    })))
}

#[derive(Debug, Deserialize)]
pub struct DeltaBody {
    pub workspace: String,
    pub repository: String,
    pub path: String,
    /// `here` or `there`. Only for a decision.
    #[serde(default)]
    pub keep: Option<String>,
}

fn delta_target(body: &DeltaBody) -> Result<(WorkspaceId, RepositoryId), Failure> {
    let workspace =
        WorkspaceId::parse(&body.workspace).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    let repository =
        RepositoryId::parse(&body.repository).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    Ok((workspace, repository))
}

/// Say which side of one difference wins.
async fn decide(
    State(machine): State<Arc<MachineState>>,
    Path(peer): Path<String>,
    Json(body): Json<DeltaBody>,
) -> Result<Json<serde_json::Value>, Failure> {
    let peer = PeerId::parse(&peer).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    let (workspace, repository) = delta_target(&body)?;
    let keep = match body.keep.as_deref() {
        Some("here") => super::conflict::Keep::Here,
        Some("there") => super::conflict::Keep::There,
        other => {
            return Err(bad(
                StatusCode::BAD_REQUEST,
                format!(
                    "'{}' is not a side to keep — 'here' or 'there'",
                    other.unwrap_or("")
                ),
            ))
        }
    };
    super::conflict::decide(&machine, &peer, &workspace, &repository, &body.path, keep)
        .await
        .map_err(|e| bad(StatusCode::CONFLICT, e))?;
    Ok(Json(serde_json::json!({
        "decided": body.path,
        "keep": body.keep,
    })))
}

/// Leave one difference alone, as it stands.
async fn dismiss(
    State(machine): State<Arc<MachineState>>,
    Path(peer): Path<String>,
    Json(body): Json<DeltaBody>,
) -> Result<Json<serde_json::Value>, Failure> {
    let peer = PeerId::parse(&peer).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    let (workspace, repository) = delta_target(&body)?;
    super::conflict::dismiss(&machine, &peer, &workspace, &repository, &body.path)
        .await
        .map_err(|e| bad(StatusCode::CONFLICT, e))?;
    Ok(Json(serde_json::json!({ "dismissed": body.path })))
}

/// Take back a decision or a dismissal.
async fn reconsider(
    State(machine): State<Arc<MachineState>>,
    Path(peer): Path<String>,
    Json(body): Json<DeltaBody>,
) -> Result<Json<serde_json::Value>, Failure> {
    let peer = PeerId::parse(&peer).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;
    let (workspace, repository) = delta_target(&body)?;
    super::conflict::reconsider(&machine, &peer, &workspace, &repository, &body.path)
        .await
        .map_err(|e| bad(StatusCode::CONFLICT, e))?;
    Ok(Json(serde_json::json!({ "reconsidered": body.path })))
}