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 dialect = detect_dialect(file)?;
87    let node = dialect.parse(&source)?;
88    let hash = repo.nodes.put(&node)?;
89    repo.mst.insert(hash)?;
90    repo.attestations
91        .add(Attestation::create(&keypair, hash))?;
92    repo.flush()?;
93
94    Ok(IngestResult {
95        hash,
96        did: keypair.did(),
97    })
98}
99
100/// Detecta el dialecto desde la extensión del archivo. Error si la
101/// extensión no corresponde a un lenguaje soportado.
102fn detect_dialect(file: &Path) -> Result<parse::Dialect, CliError> {
103    let ext = file
104        .extension()
105        .and_then(|e| e.to_str())
106        .unwrap_or("");
107    parse::detect_by_extension(ext).ok_or_else(|| {
108        CliError::UnsupportedLanguage {
109            path: file.to_path_buf(),
110            extension: ext.to_string(),
111        }
112    })
113}
114
115/// `minga listen <addr>`: arranca el peer, escucha en `addr`, y
116/// acepta sincronizaciones entrantes hasta que el proceso se cierre.
117pub async fn cmd_listen(
118    repo_path: &Path,
119    passphrase: &str,
120    addr: &str,
121) -> Result<Multiaddr, CliError> {
122    let keypair = keypair_file::load(repo_path.join(KEYPAIR_FILENAME), passphrase)?;
123    let did = keypair.did();
124    let peer = MingaPeer::open(keypair, repo_path.join(REPO_DIRNAME))?;
125    let multi: Multiaddr = addr
126        .parse()
127        .map_err(|e: libp2p::multiaddr::Error| CliError::Multiaddr(e.to_string()))?;
128    let actual = peer.listen(multi).await;
129    let _accept = peer.run_passive_accept();
130
131    // Bloqueamos para siempre mientras la task de accept procesa
132    // sincronizaciones. El usuario cierra con Ctrl+C.
133    println!("Escuchando en: {}", actual);
134    println!("DID Minga: {}", did);
135    println!("PeerID libp2p: {}", peer.peer_id());
136    futures::future::pending::<()>().await;
137
138    Ok(actual)
139}
140
141/// `minga sync <multiaddr>`: dializa al peer y ejecuta una
142/// sincronización completa con él.
143pub async fn cmd_sync(
144    repo_path: &Path,
145    passphrase: &str,
146    target: &str,
147) -> Result<(), CliError> {
148    let keypair = keypair_file::load(repo_path.join(KEYPAIR_FILENAME), passphrase)?;
149    let peer = MingaPeer::open(keypair, repo_path.join(REPO_DIRNAME))?;
150
151    let multi: Multiaddr = target
152        .parse()
153        .map_err(|e: libp2p::multiaddr::Error| CliError::Multiaddr(e.to_string()))?;
154    let peer_id = extract_peer_id(&multi).ok_or(CliError::NoPeerIdInMultiaddr)?;
155
156    peer.dial(multi);
157
158    let deadline = std::time::Instant::now() + Duration::from_secs(10);
159    loop {
160        if peer.sync_with(peer_id).await.is_ok() {
161            return Ok(());
162        }
163        if std::time::Instant::now() >= deadline {
164            return Err(CliError::SyncTimeout);
165        }
166        tokio::time::sleep(Duration::from_millis(100)).await;
167    }
168}
169
170fn extract_peer_id(addr: &Multiaddr) -> Option<PeerId> {
171    addr.iter().find_map(|p| match p {
172        Protocol::P2p(peer_id) => Some(peer_id),
173        _ => None,
174    })
175}
176
177/// `minga watch <dir>`: vigila un directorio, re-parsea y re-ingesta
178/// cualquier archivo `.rs` que se cree o modifique. Convierte Minga en
179/// un VCS de fondo: el usuario escribe en su editor habitual y el
180/// código queda versionado y firmado en el repo automáticamente.
181pub async fn cmd_watch(
182    repo_path: &Path,
183    passphrase: &str,
184    watch_dir: &Path,
185) -> Result<(), CliError> {
186    let keypair = keypair_file::load(repo_path.join(KEYPAIR_FILENAME), passphrase)?;
187    let repo = PersistentRepo::open(repo_path.join(REPO_DIRNAME))?;
188
189    // Pasada inicial: ingerimos todos los .rs ya presentes para que
190    // el repo arranque sincronizado con el contenido actual del
191    // directorio (no solo con cambios futuros).
192    initial_scan(&repo, &keypair, watch_dir);
193
194    // Canal entre el callback síncrono de notify y el bucle async.
195    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
196    let mut watcher = notify::recommended_watcher(
197        move |res: Result<notify::Event, notify::Error>| {
198            if let Ok(event) = res {
199                let _ = tx.send(event);
200            }
201        },
202    )?;
203    notify::Watcher::watch(&mut watcher, watch_dir, notify::RecursiveMode::Recursive)?;
204
205    while let Some(event) = rx.recv().await {
206        if !is_relevant_event(&event) {
207            continue;
208        }
209        for path in &event.paths {
210            if is_supported_source(path) {
211                match ingest_into_repo(&repo, &keypair, path) {
212                    Ok(hash) => {
213                        eprintln!("ingerido: {} → {}", path.display(), hash);
214                    }
215                    Err(e) => {
216                        eprintln!("warning: {} no se pudo ingerir: {}", path.display(), e);
217                    }
218                }
219            }
220        }
221    }
222
223    Ok(())
224}
225
226fn initial_scan(repo: &PersistentRepo, keypair: &Keypair, dir: &Path) {
227    let Ok(entries) = fs::read_dir(dir) else {
228        return;
229    };
230    for entry in entries.flatten() {
231        let p = entry.path();
232        if is_supported_source(&p) {
233            let _ = ingest_into_repo(repo, keypair, &p);
234        }
235    }
236}
237
238fn ingest_into_repo(
239    repo: &PersistentRepo,
240    keypair: &Keypair,
241    file: &Path,
242) -> Result<ContentHash, CliError> {
243    let source = fs::read_to_string(file)?;
244    let dialect = detect_dialect(file)?;
245    let node = dialect.parse(&source)?;
246    let hash = repo.nodes.put(&node)?;
247    repo.mst.insert(hash)?;
248    repo.attestations
249        .add(Attestation::create(keypair, hash))?;
250    repo.flush()?;
251    Ok(hash)
252}
253
254/// Detecta si un archivo debe ingerirse: existe, es regular, y su
255/// extensión corresponde a un dialecto soportado.
256fn is_supported_source(path: &Path) -> bool {
257    if !path.is_file() {
258        return false;
259    }
260    let ext = path.extension().and_then(|e| e.to_str()).unwrap_or("");
261    parse::detect_by_extension(ext).is_some()
262}
263
264fn is_relevant_event(event: &notify::Event) -> bool {
265    matches!(
266        event.kind,
267        notify::EventKind::Create(_) | notify::EventKind::Modify(_)
268    )
269}