geth/crates/geth-node/src/lib.rs

use geth_auth::AuthExplanation;
use geth_cas::{LocalCas, hash_path};
use geth_config::GethPaths;
use geth_control::{
    CasBlob, ControlRequest, ControlResponse, KeychainStatusResponse, NodeIdResponse,
    StatusResponse,
};
use geth_crypto::AgentKey;
use geth_resource::ResourceDescriptor;
use geth_store::{Store, StoredResource};
use geth_types::{ResourceId, ResourceKind, ResourceName};
use std::path::Path;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::{UnixListener, UnixStream};

#[derive(Debug, thiserror::Error)]
pub enum NodeError {
    #[error("config error: {0}")]
    Config(#[from] geth_config::ConfigError),
    #[error("crypto error: {0}")]
    Crypto(#[from] geth_crypto::CryptoError),
    #[error("store error: {0}")]
    Store(#[from] geth_store::StoreError),
    #[error("cas error: {0}")]
    Cas(#[from] geth_cas::CasError),
    #[error("control error: {0}")]
    Control(#[from] geth_control::ControlError),
    #[error("io error: {0}")]
    Io(#[from] std::io::Error),
    #[error("invalid resource kind: {0}")]
    InvalidResourceKind(String),
}

#[derive(Clone, Debug)]
pub struct LocalNode {
    pub paths: GethPaths,
    pub agent_id: String,
    pub node_id: String,
}

pub fn init_node(paths: &GethPaths) -> Result<LocalNode, NodeError> {
    paths.ensure_base_dirs()?;
    if !paths.config_file().exists() {
        std::fs::write(
            paths.config_file(),
            "# geth local node config\n# Remote node-to-node communication is Iroh-only.\n",
        )?;
    }
    let key = AgentKey::load_or_create(&paths.agent_key())?;
    let agent_id = key.agent_id().to_string();
    let node_id = stable_node_id(&agent_id);
    let store = Store::open(&paths.metadata_db())?;
    store.upsert_agent(&agent_id, &key.public_key_hex())?;
    store.upsert_node(&node_id, "local", &agent_id)?;
    store.insert_resource(&StoredResource {
        resource_id: "resource:cas:local".to_owned(),
        kind: ResourceKind::Cas.to_string(),
        name: "local-cas".to_owned(),
        status: "active".to_owned(),
    })?;
    Ok(LocalNode {
        paths: paths.clone(),
        agent_id,
        node_id,
    })
}

pub fn open_node(paths: &GethPaths) -> Result<LocalNode, NodeError> {
    init_node(paths)
}

pub async fn run_daemon(paths: GethPaths) -> Result<(), NodeError> {
    let node = init_node(&paths)?;
    if Path::new(&paths.socket_path()).exists() {
        std::fs::remove_file(paths.socket_path())?;
    }
    let listener = UnixListener::bind(paths.socket_path())?;
    tracing::info!(socket = %paths.socket_path().display(), "geth daemon listening");

    loop {
        let (stream, _) = listener.accept().await?;
        let node = node.clone();
        tokio::spawn(async move {
            if let Err(error) = handle_stream(node, stream).await {
                tracing::warn!(%error, "control request failed");
            }
        });
    }
}

pub async fn send_control(
    paths: &GethPaths,
    request: ControlRequest,
) -> Result<ControlResponse, NodeError> {
    let mut stream = UnixStream::connect(paths.socket_path()).await?;
    stream
        .write_all(geth_control::encode_request(&request)?.as_bytes())
        .await?;
    stream.shutdown().await?;
    let mut reader = BufReader::new(stream);
    let mut line = String::new();
    reader.read_line(&mut line).await?;
    Ok(geth_control::decode_response(&line)?)
}

async fn handle_stream(node: LocalNode, stream: UnixStream) -> Result<(), NodeError> {
    let mut reader = BufReader::new(stream);
    let mut line = String::new();
    reader.read_line(&mut line).await?;
    let request = geth_control::decode_request(&line)?;
    let response = match handle_request(&node, request) {
        Ok(response) => response,
        Err(error) => ControlResponse::Error {
            message: error.to_string(),
        },
    };
    let mut stream = reader.into_inner();
    stream
        .write_all(geth_control::encode_response(&response)?.as_bytes())
        .await?;
    Ok(())
}

pub fn handle_request(
    node: &LocalNode,
    request: ControlRequest,
) -> Result<ControlResponse, NodeError> {
    let store = Store::open(&node.paths.metadata_db())?;
    match request {
        ControlRequest::Status => Ok(ControlResponse::Status(StatusResponse {
            home: node.paths.home().to_path_buf(),
            socket: node.paths.socket_path(),
            agent_id: node.agent_id.clone(),
            node_id: node.node_id.clone(),
            iroh: "scaffolded; no remote endpoint is started in bootstrap".to_owned(),
        })),
        ControlRequest::NodeId => Ok(ControlResponse::NodeId(NodeIdResponse {
            agent_id: node.agent_id.clone(),
            node_id: node.node_id.clone(),
            endpoint_id: None,
        })),
        ControlRequest::ResourceList => Ok(ControlResponse::ResourceList {
            resources: store
                .list_resources()?
                .into_iter()
                .map(stored_resource_to_descriptor)
                .collect::<Result<Vec<_>, _>>()?,
        }),
        ControlRequest::ResourceCreate { kind, name } => {
            let kind = kind
                .parse::<ResourceKind>()
                .map_err(|_| NodeError::InvalidResourceKind(kind.clone()))?;
            let id = format!("resource:{}:{}", kind, name);
            let stored = StoredResource {
                resource_id: id,
                kind: kind.to_string(),
                name,
                status: "active".to_owned(),
            };
            store.insert_resource(&stored)?;
            Ok(ControlResponse::ResourceCreated {
                resource: stored_resource_to_descriptor(stored)?,
            })
        }
        ControlRequest::CasAdd { path } => {
            let cas = LocalCas::new(node.paths.cas_dir());
            let info = cas.add_path(&path)?;
            store.record_cas_object(
                info.hash.as_str(),
                info.size_bytes,
                &info.path.to_string_lossy(),
            )?;
            Ok(ControlResponse::CasAdded {
                hash: info.hash,
                size_bytes: info.size_bytes,
            })
        }
        ControlRequest::CasGet { hash, out } => {
            let cas = LocalCas::new(node.paths.cas_dir());
            let size_bytes = cas.get_to_path(&hash, &out)?;
            Ok(ControlResponse::CasGot {
                hash,
                out,
                size_bytes,
            })
        }
        ControlRequest::CasHash { path } => Ok(ControlResponse::CasHash {
            hash: hash_path(&path)?,
        }),
        ControlRequest::CasHas { hash } => {
            let cas = LocalCas::new(node.paths.cas_dir());
            let present = cas.has(&hash)?;
            Ok(ControlResponse::CasHas { hash, present })
        }
        ControlRequest::CasList => {
            let cas = LocalCas::new(node.paths.cas_dir());
            Ok(ControlResponse::CasList {
                blobs: cas
                    .list()?
                    .into_iter()
                    .map(|blob| CasBlob {
                        hash: blob.hash,
                        size_bytes: blob.size_bytes,
                    })
                    .collect(),
            })
        }
        ControlRequest::KeychainStatus => {
            Ok(ControlResponse::KeychainStatus(KeychainStatusResponse {
                initialized: false,
                admin_keys: 0,
                users: 0,
                devices: 0,
                nodes: 1,
            }))
        }
        ControlRequest::AuthExplain {
            subject,
            resource,
            capability,
        } => Ok(ControlResponse::AuthExplain(AuthExplanation::stub(
            subject, resource, capability,
        ))),
        ControlRequest::ModuleStub { module, command } => {
            Ok(ControlResponse::NotImplemented { module, command })
        }
    }
}

fn stored_resource_to_descriptor(stored: StoredResource) -> Result<ResourceDescriptor, NodeError> {
    let kind = stored
        .kind
        .parse::<ResourceKind>()
        .map_err(|_| NodeError::InvalidResourceKind(stored.kind.clone()))?;
    Ok(ResourceDescriptor::local(
        ResourceId::new(stored.resource_id),
        kind,
        ResourceName::new(stored.name),
    ))
}

fn stable_node_id(agent_id: &str) -> String {
    format!("node:{agent_id}")
}