varde/varde-proto/src/lib.rs

//! Wire types for the varde socket API.
//!
//! The protocol is JSON Lines over a unix stream socket: the client writes
//! one [`Request`] per line, the daemon answers with one [`Envelope`] per
//! line. A [`Request::Subscribe`] switches the connection into event mode,
//! after which the daemon writes one [`Event`] per line until the client
//! disconnects.
//!
//! Hashes, tickets and node ids travel as strings (BLAKE3 hex, iroh ticket
//! base32, z-base-32 NodeId respectively) so this crate stays free of iroh
//! dependencies and the protocol stays trivially consumable from any
//! language.

use std::collections::BTreeMap;

use serde::{Deserialize, Serialize};

/// Version of the socket protocol described by this crate.
///
/// Bumped on incompatible changes; reported in [`GlobalStatus`].
pub const PROTOCOL_VERSION: u32 = 2;

/// The most bytes a [`Request::Put`] or a [`Request::Read`] reply carries
/// inline. Larger content goes through `Add`/`Materialize` and a path.
pub const MAX_INLINE_BYTES: u64 = 16 * 1024 * 1024;

/// A request from a client to the daemon. One JSON object per line.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum Request {
    /// Import a file or directory at `path` into the store.
    ///
    /// `path` must be absolute (the daemon has no meaningful cwd).
    /// Returns [`ResponseData::Added`] with the root hash — for a directory
    /// this is the HashSeq root, for a file the blob hash.
    Add {
        /// Absolute path of the file or directory to import.
        path: String,
        /// Import directories recursively. Adding a directory with
        /// `recursive: false` is an error.
        #[serde(default)]
        recursive: bool,
    },
    /// Standing intent: keep this content, fetch it if absent, serve it to
    /// peers according to `policy`.
    Pin {
        /// BLAKE3 hash (hex) of the blob or HashSeq root to pin.
        hash: String,
        /// Replication policy for this pin.
        #[serde(default)]
        policy: PinPolicy,
    },
    /// Remove a standing pin. Does not delete data — see [`Request::Gc`].
    Unpin {
        /// Hash (hex) of the pin to remove.
        hash: String,
    },
    /// Export content from the store to a filesystem path.
    Materialize {
        /// Hash (hex) of the blob or HashSeq root to export.
        hash: String,
        /// Absolute destination path.
        dest: String,
        /// How to get bytes out of the store.
        #[serde(default)]
        mode: MaterializeMode,
    },
    /// Import bytes carried in the request, base64 (standard alphabet),
    /// at most [`MAX_INLINE_BYTES`]. Replies [`ResponseData::Added`].
    ///
    /// For a client that shares no filesystem with the daemon: the system
    /// daemon runs as its own user with `ProtectHome`, so it can neither
    /// read a file a desktop app wrote for `Add` nor write one back for
    /// `Materialize`. Added in varde 0.1.4.
    Put {
        /// The content, base64.
        data: String,
    },
    /// Read a complete raw blob back in the reply, base64, at most
    /// [`MAX_INLINE_BYTES`]. Replies [`ResponseData::Bytes`]. The
    /// counterpart of [`Request::Put`]; added in varde 0.1.4.
    Read {
        /// Hash (hex) of the blob.
        hash: String,
    },
    /// Query daemon status, globally or for one hash.
    Status {
        /// If set, report on this hash; otherwise report global status.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        hash: Option<String>,
    },
    /// List all pins with their policies and completeness.
    List {},
    /// Produce an iroh blob ticket for out-of-band sharing of `hash`.
    TicketExport {
        /// Hash (hex) of the content to share.
        hash: String,
    },
    /// Import an iroh blob ticket: pin the content and fetch it from the
    /// provider embedded in the ticket.
    TicketImport {
        /// The ticket string, as produced by [`Request::TicketExport`].
        ticket: String,
        /// Policy for the pin created by the import.
        #[serde(default)]
        pin_policy: PinPolicy,
    },
    /// Trust a peer: content is served to and fetched from trusted peers
    /// only (unless a pin is `open_lan`).
    PeerTrust {
        /// The peer's iroh NodeId (z-base-32).
        node_id: String,
    },
    /// Revoke trust in a peer.
    PeerUntrust {
        /// The peer's iroh NodeId (z-base-32).
        node_id: String,
    },
    /// List trusted peers and their current connectivity.
    PeerList {},
    /// Push fully-present content to a trusted peer, unprompted.
    ///
    /// Consent is mutual: this daemon only pushes to peers it trusts, and
    /// the receiver only accepts pushes from peers *it* trusts. On the
    /// receiving side an accepted push is recorded as a pin (default
    /// policy), so the content survives gc and shows up in `List`.
    Push {
        /// Hash (hex) of the fully-present blob or HashSeq root to push.
        hash: String,
        /// The receiving peer's iroh NodeId (z-base-32).
        node_id: String,
    },
    /// Drop all blobs not reachable from a pin. Explicit, never automatic.
    Gc {},
    /// Switch this connection to an event stream. The daemon acknowledges
    /// with an ok envelope, then writes one [`Event`] per line.
    Subscribe {},
}

/// Replication policy attached to a pin.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct PinPolicy {
    /// Serve this content to *any* LAN peer, not just trusted ones.
    /// Defaults to off: trust gates participation.
    #[serde(default)]
    pub open_lan: bool,
}

/// How [`Request::Materialize`] gets bytes out of the store.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum MaterializeMode {
    /// Try a reflink (`FICLONE`) first, fall back to a plain copy.
    #[default]
    ReflinkOrCopy,
    /// Always copy bytes.
    Copy,
}

/// The envelope wrapped around every daemon reply.
///
/// `ok: true` replies carry `data`, `ok: false` replies carry `error`.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Envelope {
    /// Whether the request succeeded.
    pub ok: bool,
    /// Payload on success.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub data: Option<ResponseData>,
    /// Structured error on failure.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub error: Option<ApiError>,
}

impl Envelope {
    /// Build a success envelope carrying `data`.
    pub fn ok(data: ResponseData) -> Self {
        Envelope {
            ok: true,
            data: Some(data),
            error: None,
        }
    }

    /// Build an error envelope.
    pub fn err(code: ErrorCode, message: impl Into<String>) -> Self {
        Envelope {
            ok: false,
            data: None,
            error: Some(ApiError {
                code,
                message: message.into(),
            }),
        }
    }
}

/// A structured error. Never a bare string.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ApiError {
    /// Machine-readable error class.
    pub code: ErrorCode,
    /// Human-readable detail.
    pub message: String,
}

/// Machine-readable error classes.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ErrorCode {
    /// The request line was not valid JSON or not a known request.
    BadRequest,
    /// A hash, ticket or node id failed to parse.
    InvalidArgument,
    /// The referenced content is not in the store.
    NotFound,
    /// Filesystem-level failure (permissions, missing path, ...).
    Io,
    /// The store or endpoint failed internally.
    Internal,
    /// The feature exists in the protocol but this daemon build or
    /// milestone does not implement it yet.
    Unimplemented,
}

/// Success payloads, tagged by response kind.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum ResponseData {
    /// Content was imported. Reply to [`Request::Add`].
    Added {
        /// Root hash (hex) of the imported content.
        hash: String,
        /// Total bytes imported.
        bytes: u64,
    },
    /// A pin was created or updated. Reply to [`Request::Pin`] and
    /// [`Request::TicketImport`].
    Pinned {
        /// The pinned root hash (hex).
        hash: String,
    },
    /// Generic acknowledgement for requests with nothing to report
    /// (`Unpin`, `PeerTrust`, `PeerUntrust`, `Subscribe` ack).
    Done {},
    /// Content was exported. Reply to [`Request::Materialize`].
    Materialized {
        /// Bytes written to the destination.
        bytes: u64,
        /// Whether at least one file was reflinked rather than copied.
        reflinked: bool,
    },
    /// A blob's content, base64. Reply to [`Request::Read`].
    Bytes {
        /// The content, base64 (standard alphabet).
        data: String,
    },
    /// Global daemon status. Reply to [`Request::Status`] without a hash.
    Status(GlobalStatus),
    /// Per-hash status. Reply to [`Request::Status`] with a hash.
    HashStatus(HashStatus),
    /// All pins. Reply to [`Request::List`].
    Pins {
        /// One entry per pin.
        pins: Vec<PinInfo>,
    },
    /// A serialized iroh ticket. Reply to [`Request::TicketExport`].
    Ticket {
        /// The ticket string.
        ticket: String,
    },
    /// Trusted peers. Reply to [`Request::PeerList`].
    Peers {
        /// One entry per trusted peer.
        peers: Vec<PeerInfo>,
    },
    /// Garbage collection result. Reply to [`Request::Gc`].
    GcDone {
        /// Number of blobs removed.
        blobs_removed: u64,
    },
    /// Content was pushed to a peer. Reply to [`Request::Push`].
    Pushed {
        /// The pushed root hash (hex).
        hash: String,
        /// Payload bytes written to the peer.
        bytes: u64,
    },
}

/// Global daemon status.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct GlobalStatus {
    /// Daemon version (crate version).
    pub version: String,
    /// Socket protocol version, see [`PROTOCOL_VERSION`].
    pub protocol: u32,
    /// This daemon's iroh NodeId (z-base-32), if the endpoint is up.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub node_id: Option<String>,
    /// Path of the blob store directory.
    pub store_path: String,
    /// Number of standing pins.
    pub pins: u64,
    /// Number of connected trusted peers.
    pub connected_peers: u64,
}

/// Status of one hash.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct HashStatus {
    /// The hash (hex) this reports on.
    pub hash: String,
    /// Bytes present locally.
    pub have_bytes: u64,
    /// Total size if known (unknown while a fetch has not seen the size).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub total_bytes: Option<u64>,
    /// Whether the content is fully present and verified.
    pub complete: bool,
    /// Whether a pin exists for this hash.
    pub pinned: bool,
    /// Node ids of peers currently transferring this hash with us.
    pub active_peers: Vec<String>,
}

/// One pin as reported by [`ResponseData::Pins`].
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct PinInfo {
    /// The pinned root hash (hex).
    pub hash: String,
    /// The pin's replication policy.
    pub policy: PinPolicy,
    /// Whether the content is fully present locally.
    pub complete: bool,
    /// Bytes present locally.
    pub have_bytes: u64,
    /// Total size if known.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub total_bytes: Option<u64>,
}

/// One trusted peer as reported by [`ResponseData::Peers`].
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PeerInfo {
    /// The peer's iroh NodeId (z-base-32).
    pub node_id: String,
    /// Whether we currently hold a connection to this peer.
    pub connected: bool,
}

/// Events streamed after [`Request::Subscribe`]. One JSON object per line.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "event", rename_all = "snake_case")]
pub enum Event {
    /// Bytes moved for a hash.
    TransferProgress {
        /// The hash being transferred (hex).
        hash: String,
        /// Direction of the transfer.
        direction: Direction,
        /// Bytes transferred so far.
        bytes_done: u64,
        /// Total bytes if known.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        bytes_total: Option<u64>,
    },
    /// A trusted peer became reachable.
    PeerJoined {
        /// The peer's NodeId (z-base-32).
        node_id: String,
    },
    /// A trusted peer went away.
    PeerLeft {
        /// The peer's NodeId (z-base-32).
        node_id: String,
    },
    /// A pin finished fetching and verified completely.
    PinComplete {
        /// The pinned root hash (hex).
        hash: String,
    },
    /// A trusted peer pushed content to this daemon; it has been pinned
    /// with the default policy.
    PushReceived {
        /// The pushed root hash (hex).
        hash: String,
        /// NodeId (z-base-32) of the pushing peer.
        from: String,
    },
    /// A signed announcement was received on a subscribed discovery topic.
    Announcement {
        /// The announced HashSeq root (hex).
        root: String,
        /// The announcing author's ed25519 public key (hex).
        author: String,
        /// Free-form metadata carried by the announcement.
        meta: BTreeMap<String, String>,
    },
}

/// Transfer direction for [`Event::TransferProgress`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Direction {
    /// We are fetching.
    Download,
    /// We are serving.
    Upload,
}

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

    #[test]
    fn request_wire_shape() {
        let req = Request::Add {
            path: "/tmp/x".into(),
            recursive: true,
        };
        let json = serde_json::to_string(&req).unwrap();
        assert_eq!(json, r#"{"type":"add","path":"/tmp/x","recursive":true}"#);
        let back: Request = serde_json::from_str(&json).unwrap();
        assert_eq!(back, req);
    }

    #[test]
    fn status_request_defaults() {
        let req: Request = serde_json::from_str(r#"{"type":"status"}"#).unwrap();
        assert_eq!(req, Request::Status { hash: None });
    }

    #[test]
    fn envelope_ok_carries_flag() {
        let env = Envelope::ok(ResponseData::Done {});
        let json = serde_json::to_string(&env).unwrap();
        assert!(json.starts_with(r#"{"ok":true"#));
        let back: Envelope = serde_json::from_str(&json).unwrap();
        assert!(back.ok && back.error.is_none());
    }

    #[test]
    fn envelope_err_is_structured() {
        let env = Envelope::err(ErrorCode::NotFound, "no such hash");
        let json = serde_json::to_string(&env).unwrap();
        let back: Envelope = serde_json::from_str(&json).unwrap();
        let err = back.error.unwrap();
        assert_eq!(err.code, ErrorCode::NotFound);
        assert!(!back.ok);
    }

    #[test]
    fn pin_policy_defaults_closed() {
        let req: Request = serde_json::from_str(r#"{"type":"pin","hash":"abc"}"#).unwrap();
        match req {
            Request::Pin { policy, .. } => assert!(!policy.open_lan),
            _ => panic!("wrong variant"),
        }
    }
}