wsb 0.258.6441

//! What two daemons say to each other.
//!
//! One request/response conversation per connection, over the encrypted
//! channel in [`super::transport`]. The messages are deliberately about STATE
//! rather than about commands: a peer says what it has and what it believes,
//! and each side decides for itself what to do. Nothing here tells the other
//! machine to write a file — a receiver applies a transfer only after it has
//! independently agreed that the transfer is safe.
//!
//! # Versioned, and refused rather than guessed
//!
//! Every conversation opens with [`VERSION`]. Two daemons that do not agree on
//! it stop, and say so. There is no negotiation and no compatibility layer: a
//! peer running a different wsb is a peer to update, and a half-understood
//! protocol moving files is the failure that would be worth avoiding at any
//! cost.

use std::collections::BTreeMap;

use serde::{Deserialize, Serialize};

use super::compatibility::{Epoch, RepositoryFacts};
use super::converge::Entry;
use super::model::{RepositoryId, WorkspaceId};
use super::scan::PolicyDigest;

/// The wire version. Bumped whenever any message changes shape.
///
/// 5: a machine no longer answers what platform it is; that served registering
/// a peer as a test host, which is not part of continuity.
/// 6: [`Request::Plugin`] carries a plugin's request to the same plugin on the
/// other machine; refusals say [`RefusalCode::NoSuchService`] and
/// [`RefusalCode::PluginAbsent`].
/// 7: a workspace id is `<family>@<worktree>`: each worktree of a repository
/// is carried with the worktree of the same name, not every worktree with
/// whichever one was opened first. [`Request::Apply`] says what the plan found
/// at each path it touches, so the receiver leaves alone a file edited since.
pub const VERSION: u32 = 7;

/// One workspace, as a peer advertises it.
///
/// No path. Where the workspace sits on the other machine is that machine's
/// business, and matching workspaces by path is the mistake this whole
/// identity scheme exists to prevent.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkspaceSummary {
    pub workspace: WorkspaceId,
    pub display_name: String,
    /// The superproject's root commit — the same value that identifies the
    /// repository itself, carried here so a workspace id collision (or a
    /// restored backup with a rewritten id) cannot pass for the same tree.
    pub root_commit: String,
}

/// Everything a peer knows about one repository of one workspace, when asked.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RepositoryReport {
    pub facts: RepositoryFacts,
    pub policy: PolicyDigest,
    /// The continuity set, keyed by repository-relative path.
    pub entries: BTreeMap<String, Entry>,
}

/// A request from one daemon to the other.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "request", rename_all = "snake_case")]
pub enum Request {
    /// What workspaces do you have?
    Inventory,
    /// What is the state of these repositories in this workspace?
    ///
    /// The repositories are named by the asker, which is what keeps the answer
    /// bounded: a workspace with forty repositories does not send forty
    /// working-tree scans to answer "are we on the same branch".
    Report {
        workspace: WorkspaceId,
        repositories: Vec<RepositoryId>,
    },
    /// Send me these paths' contents, as of this epoch.
    ///
    /// The epoch is the RECEIVER's, learned from its own report a moment ago.
    /// An epoch identifies one moment of one repository on one machine — it
    /// includes that machine's index — so the only side that can check one is
    /// the side it describes.
    Fetch {
        workspace: WorkspaceId,
        repository: RepositoryId,
        epoch: Epoch,
        paths: Vec<String>,
    },
    /// Take these paths. The receiver validates the epoch against its OWN
    /// current state before writing anything.
    Apply {
        workspace: WorkspaceId,
        repository: RepositoryId,
        epoch: Epoch,
        /// Content-carrying changes. Deletions are separate so an empty file
        /// and a deleted file can never be confused by an absent field.
        writes: Vec<Write>,
        deletions: Vec<String>,
        /// What the sender's plan read on THIS side at every path it writes or
        /// deletes, from the receiver's own report. A path the plan found
        /// nothing at is absent.
        ///
        /// A file edited here after that report is work the plan never saw,
        /// and the receiver leaves it alone rather than replace it: the epoch
        /// is git's state and says nothing about the working tree, and the
        /// baseline is what was agreed, not what is here now.
        found: BTreeMap<String, Entry>,
        /// What the sender believed the baseline was when it planned this.
        /// The receiver refuses a transfer planned against a baseline it does
        /// not share, rather than applying changes decided from a state that
        /// has since moved.
        baseline_digest: String,
    },
    /// Do we still agree about what we last agreed on?
    ///
    /// Asked before a machine changes its OWN tree from a plan — before it
    /// writes what a fetch brings back and, more importantly, before it
    /// deletes what the plan says the other side no longer has.
    ///
    /// A convergence is computed from three things: this side's scan, the
    /// other side's scan, and the baseline. The first two are exchanged. The
    /// baseline is held separately by each machine, and if the two copies have
    /// drifted apart then "the peer deleted this" can be a conclusion drawn
    /// entirely from this side's own copy — about a file the peer never had.
    /// A file was deleted from the machine that created it exactly that way.
    ///
    /// `Apply` and `Acknowledge` already carry the digest, so the paths where
    /// this machine changes the OTHER one were covered. This is the direction
    /// that was not: a pass whose plan was only deletions sent no request at
    /// all and deleted on an unconfirmed opinion.
    Confirm {
        workspace: WorkspaceId,
        repository: RepositoryId,
        baseline_digest: String,
    },
    /// This baseline is now agreed for these paths.
    Acknowledge {
        workspace: WorkspaceId,
        repository: RepositoryId,
        epoch: Epoch,
        paths: Vec<String>,
        baseline_digest: String,
    },
    /// A request from a plugin on the asking machine to the same plugin here.
    ///
    /// The body is the plugin's own bytes, base64, and nothing here reads it:
    /// wsb carries the request, proves who sent it, and hands it to the plugin
    /// registered for the service — if the service is switched on with the
    /// asking machine on this side, and the plugin is running.
    Plugin {
        service: crate::plugin::PluginService,
        body: String,
    },
}

/// One file's worth of a transfer.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Write {
    pub path: String,
    pub entry: Entry,
    /// The bytes, base64. For a symlink, the target text.
    pub content: String,
}

impl Write {
    /// The bytes this write carries, checked against the hash the entry
    /// claims.
    ///
    /// Verified on arrival rather than trusted: a peer is authenticated, which
    /// says who it is, not that every byte survived the journey — and a write
    /// whose content does not match its own entry would poison the baseline
    /// with a hash nothing on disk has.
    pub fn decoded(&self) -> Result<Vec<u8>, String> {
        let bytes = super::identity::base64_decode(&self.content)
            .map_err(|e| format!("{}: {e}", self.path))?;
        let actual = super::converge::hash_bytes(&bytes);
        if actual != self.entry.hash {
            return Err(format!(
                "{}: the content does not match the hash it was sent with",
                self.path
            ));
        }
        if bytes.len() as u64 != self.entry.size {
            return Err(format!(
                "{}: the content is {} bytes and was sent as {}",
                self.path,
                bytes.len(),
                self.entry.size
            ));
        }
        Ok(bytes)
    }
}

/// What a daemon answers.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "response", rename_all = "snake_case")]
pub enum Response {
    Inventory {
        workspaces: Vec<WorkspaceSummary>,
    },
    Report {
        /// Absent when the peer does not have the workspace at all — which is
        /// a state, not a failure.
        #[serde(skip_serializing_if = "Option::is_none")]
        workspace_present: Option<bool>,
        repositories: Vec<RepositoryReport>,
    },
    Fetched {
        writes: Vec<Write>,
    },
    /// The paths that were actually written, so the sender advances its
    /// baseline over exactly those and no more.
    Applied {
        applied: Vec<String>,
    },
    Acknowledged,
    /// The plugin's reply, base64, for the plugin that asked.
    Plugin {
        body: String,
    },
    /// The request could not be honoured, and this is why. A refusal is not an
    /// error in the connection: the connection is fine and the state was
    /// wrong, which is a thing the coordinator handles rather than a thing
    /// that drops the peer.
    Refused {
        code: RefusalCode,
        detail: String,
    },
}

/// Why a request was refused, as data.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RefusalCode {
    /// The epoch the request was prepared against is not the one the
    /// repository is in now. The sender re-plans; nothing is wrong.
    EpochMoved,
    /// The sender planned against a baseline this side does not share.
    BaselineMismatch,
    /// This side does not have that workspace.
    UnknownWorkspace,
    /// This side does not have that repository, or it is not initialised.
    UnknownRepository,
    /// Continuity is not enabled for this relationship on this side.
    NotEnabled,
    /// The service the request belongs to is switched off with this peer on
    /// the answering side.
    ///
    /// Distinct from `NotEnabled`, which is about one workspace's continuity,
    /// and deliberately distinct from answering "not held" to a secret query:
    /// a machine that has switched off secret sharing with this peer is not
    /// saying it lacks the secret, and the asking side has to be able to say
    /// which of the two happened.
    ServiceOff,
    /// The request named a path that may never be synchronised, or one that
    /// escapes the repository.
    ForbiddenPath,
    /// The two sides declare different continuity policy.
    PolicyMismatch,
    /// The request was malformed.
    Malformed,
    /// Something failed on this side that is nobody's fault but this side's.
    Internal,
    /// No plugin on this side offers the service asked for: none is enabled
    /// under that name, or the one that is has not registered it.
    NoSuchService,
    /// The plugin that offers the service is registered here and did not
    /// answer: it has stopped, or its endpoint refused the call.
    PluginAbsent,
    /// The plugin here answered and declined this request; the detail is its
    /// own words.
    PluginRefused,
}

impl RefusalCode {
    pub fn as_str(self) -> &'static str {
        match self {
            Self::EpochMoved => "epoch_moved",
            Self::BaselineMismatch => "baseline_mismatch",
            Self::UnknownWorkspace => "unknown_workspace",
            Self::UnknownRepository => "unknown_repository",
            Self::NotEnabled => "not_enabled",
            Self::ServiceOff => "service_off",
            Self::ForbiddenPath => "forbidden_path",
            Self::PolicyMismatch => "policy_mismatch",
            Self::Malformed => "malformed",
            Self::Internal => "internal",
            Self::NoSuchService => "no_such_service",
            Self::PluginAbsent => "plugin_absent",
            Self::PluginRefused => "plugin_refused",
        }
    }
}

/// A digest of a baseline, so two sides can tell whether they are planning
/// against the same one without sending it.
///
/// Order-independent by construction: the map is sorted, so two daemons that
/// hold the same baseline compute the same digest whatever order they learned
/// it in.
pub fn baseline_digest(baseline: &super::converge::Baseline) -> String {
    use sha2::{Digest, Sha256};
    let mut hasher = Sha256::new();
    hasher.update(b"wsb-peer-baseline-v1");
    for (path, entry) in baseline {
        hasher.update(path.as_bytes());
        hasher.update([0]);
        hasher.update(entry.hash.as_bytes());
        hasher.update([0]);
        hasher.update(match entry.kind {
            super::converge::EntryKind::File => b"f".as_slice(),
            super::converge::EntryKind::Symlink => b"l".as_slice(),
        });
        hasher.update(if entry.executable { b"x" } else { b"-" });
        hasher.update([0]);
    }
    format!("{:x}", hasher.finalize())
}

#[cfg(test)]
#[path = "protocol_tests.rs"]
mod tests;