tawasuyu/brahman

1//! Implementaciones de los subcomandos. Funciones puras que retornan
2//! datos estructurados — el binario las llama y formatea la salida.
3//!
4//! Layout en disco bajo `repo_path/`:
5//! - `keypair`     — la `Keypair` del peer cifrada con passphrase.
6//! - `repo/`       — directorio sled con `nodes`, `attestations`, `mst`.
7
8use std::fs;
9use std::path::Path;
10use std::time::Duration;
11
12use libp2p::{multiaddr::Protocol, Multiaddr, PeerId};
13use minga_core::{parse, Attestation, ContentHash, Did, Keypair};
14use minga_p2p::MingaPeer;
15use minga_store::{keypair_file, PersistentRepo};
16
17use crate::error::CliError;
18
19pub const KEYPAIR_FILENAME: &str = "keypair";
20pub const REPO_DIRNAME: &str = "repo";
21
22#[derive(Debug, Clone)]
23pub struct RepoStatus {
24    pub did: Did,
25    pub mst_len: usize,
26    pub nodes_len: usize,
27    pub attestations_len: usize,
28}
29
30#[derive(Debug, Clone)]
31pub struct IngestResult {
32    pub hash: ContentHash,
33    pub did: Did,
34}
35
36/// `minga init`: genera un keypair fresco, crea el repo persistente,
37/// y guarda el keypair cifrado.
38pub fn cmd_init(repo_path: &Path, passphrase: &str) -> Result<Did, CliError> {
39    if repo_path.exists() {
40        // Si el directorio existe pero está vacío, lo aceptamos.
41        // Si tiene cualquier cosa, abortamos para no pisar un repo.
42        let mut entries = fs::read_dir(repo_path)?;
43        if entries.next().is_some() {
44            return Err(CliError::AlreadyExists(repo_path.to_path_buf()));
45        }
46    } else {
47        fs::create_dir_all(repo_path)?;
48    }
49
50    let keypair = Keypair::generate();
51    keypair_file::save(&keypair, repo_path.join(KEYPAIR_FILENAME), passphrase)?;
52
53    // Crear el repo sled vacío. Se cierra al final del scope; el
54    // siguiente comando lo reabre.
55    let _ = PersistentRepo::open(repo_path.join(REPO_DIRNAME))?;
56
57    Ok(keypair.did())
58}
59
60/// `minga status`: descifra el keypair, abre el repo, devuelve
61/// estadísticas básicas.
62pub fn cmd_status(repo_path: &Path, passphrase: &str) -> Result<RepoStatus, CliError> {
63    let keypair = keypair_file::load(repo_path.join(KEYPAIR_FILENAME), passphrase)?;
64    let repo = PersistentRepo::open(repo_path.join(REPO_DIRNAME))?;
65
66    Ok(RepoStatus {
67        did: keypair.did(),
68        mst_len: repo.mst.len(),
69        nodes_len: repo.nodes.len(),
70        attestations_len: repo.attestations.len(),
71    })
72}
73
74/// `minga ingest <file>`: parsea un archivo Rust con tree-sitter,
75/// inserta el AST en el store, lo añade al MST, y crea una atestación
76/// firmada por el dueño del keypair (auto-firma de autoría).
77pub fn cmd_ingest(
78    repo_path: &Path,
79    passphrase: &str,
80    file: &Path,
81) -> Result<IngestResult, CliError> {
82    let keypair = keypair_file::load(repo_path.join(KEYPAIR_FILENAME), passphrase)?;
83    let repo = PersistentRepo::open(repo_path.join(REPO_DIRNAME))?;
84
85    let source = fs::read_to_string(file)?;
86    let node = parse::rust(&source)?;
87    let hash = repo.nodes.put(&node)?;
88    repo.mst.insert(hash)?;
89    repo.attestations
90        .add(Attestation::create(&keypair, hash))?;
91    repo.flush()?;
92
93    Ok(IngestResult {
94        hash,
95        did: keypair.did(),
96    })
97}
98
99/// `minga listen <addr>`: arranca el peer, escucha en `addr`, y
100/// acepta sincronizaciones entrantes hasta que el proceso se cierre.
101pub async fn cmd_listen(
102    repo_path: &Path,
103    passphrase: &str,
104    addr: &str,
105) -> Result<Multiaddr, CliError> {
106    let keypair = keypair_file::load(repo_path.join(KEYPAIR_FILENAME), passphrase)?;
107    let did = keypair.did();
108    let peer = MingaPeer::open(keypair, repo_path.join(REPO_DIRNAME))?;
109    let multi: Multiaddr = addr
110        .parse()
111        .map_err(|e: libp2p::multiaddr::Error| CliError::Multiaddr(e.to_string()))?;
112    let actual = peer.listen(multi).await;
113    let _accept = peer.run_passive_accept();
114
115    // Bloqueamos para siempre mientras la task de accept procesa
116    // sincronizaciones. El usuario cierra con Ctrl+C.
117    println!("Escuchando en: {}", actual);
118    println!("DID Minga: {}", did);
119    println!("PeerID libp2p: {}", peer.peer_id());
120    futures::future::pending::<()>().await;
121
122    Ok(actual)
123}
124
125/// `minga sync <multiaddr>`: dializa al peer y ejecuta una
126/// sincronización completa con él.
127pub async fn cmd_sync(
128    repo_path: &Path,
129    passphrase: &str,
130    target: &str,
131) -> Result<(), CliError> {
132    let keypair = keypair_file::load(repo_path.join(KEYPAIR_FILENAME), passphrase)?;
133    let peer = MingaPeer::open(keypair, repo_path.join(REPO_DIRNAME))?;
134
135    let multi: Multiaddr = target
136        .parse()
137        .map_err(|e: libp2p::multiaddr::Error| CliError::Multiaddr(e.to_string()))?;
138    let peer_id = extract_peer_id(&multi).ok_or(CliError::NoPeerIdInMultiaddr)?;
139
140    peer.dial(multi);
141
142    let deadline = std::time::Instant::now() + Duration::from_secs(10);
143    loop {
144        if peer.sync_with(peer_id).await.is_ok() {
145            return Ok(());
146        }
147        if std::time::Instant::now() >= deadline {
148            return Err(CliError::SyncTimeout);
149        }
150        tokio::time::sleep(Duration::from_millis(100)).await;
151    }
152}
153
154fn extract_peer_id(addr: &Multiaddr) -> Option<PeerId> {
155    addr.iter().find_map(|p| match p {
156        Protocol::P2p(peer_id) => Some(peer_id),
157        _ => None,
158    })
159}
160
161/// `minga watch <dir>`: vigila un directorio, re-parsea y re-ingesta
162/// cualquier archivo `.rs` que se cree o modifique. Convierte Minga en
163/// un VCS de fondo: el usuario escribe en su editor habitual y el
164/// código queda versionado y firmado en el repo automáticamente.
165pub async fn cmd_watch(
166    repo_path: &Path,
167    passphrase: &str,
168    watch_dir: &Path,
169) -> Result<(), CliError> {
170    let keypair = keypair_file::load(repo_path.join(KEYPAIR_FILENAME), passphrase)?;
171    let repo = PersistentRepo::open(repo_path.join(REPO_DIRNAME))?;
172
173    // Pasada inicial: ingerimos todos los .rs ya presentes para que
174    // el repo arranque sincronizado con el contenido actual del
175    // directorio (no solo con cambios futuros).
176    initial_scan(&repo, &keypair, watch_dir);
177
178    // Canal entre el callback síncrono de notify y el bucle async.
179    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
180    let mut watcher = notify::recommended_watcher(
181        move |res: Result<notify::Event, notify::Error>| {
182            if let Ok(event) = res {
183                let _ = tx.send(event);
184            }
185        },
186    )?;
187    notify::Watcher::watch(&mut watcher, watch_dir, notify::RecursiveMode::Recursive)?;
188
189    while let Some(event) = rx.recv().await {
190        if !is_relevant_event(&event) {
191            continue;
192        }
193        for path in &event.paths {
194            if is_rs_file(path) {
195                match ingest_into_repo(&repo, &keypair, path) {
196                    Ok(hash) => {
197                        eprintln!("ingerido: {} → {}", path.display(), hash);
198                    }
199                    Err(e) => {
200                        eprintln!("warning: {} no se pudo ingerir: {}", path.display(), e);
201                    }
202                }
203            }
204        }
205    }
206
207    Ok(())
208}
209
210fn initial_scan(repo: &PersistentRepo, keypair: &Keypair, dir: &Path) {
211    let Ok(entries) = fs::read_dir(dir) else {
212        return;
213    };
214    for entry in entries.flatten() {
215        let p = entry.path();
216        if is_rs_file(&p) {
217            let _ = ingest_into_repo(repo, keypair, &p);
218        }
219    }
220}
221
222fn ingest_into_repo(
223    repo: &PersistentRepo,
224    keypair: &Keypair,
225    file: &Path,
226) -> Result<ContentHash, CliError> {
227    let source = fs::read_to_string(file)?;
228    let node = parse::rust(&source)?;
229    let hash = repo.nodes.put(&node)?;
230    repo.mst.insert(hash)?;
231    repo.attestations
232        .add(Attestation::create(keypair, hash))?;
233    repo.flush()?;
234    Ok(hash)
235}
236
237fn is_rs_file(path: &Path) -> bool {
238    path.extension().and_then(|e| e.to_str()) == Some("rs") && path.is_file()
239}
240
241fn is_relevant_event(event: &notify::Event) -> bool {
242    matches!(
243        event.kind,
244        notify::EventKind::Create(_) | notify::EventKind::Modify(_)
245    )
246}