gn_sync/
p2p.rs

1use anyhow::{Context, Result};
2use gn_core::{MergeStrategy, Namespace, NotesEngine};
3use serde::{Deserialize, Serialize};
4use std::net::{IpAddr, Ipv4Addr, ToSocketAddrs, UdpSocket};
5use std::path::{Path, PathBuf};
6use std::process::{Child, Command};
7use uuid::Uuid;
8
9#[derive(Debug, Clone, Serialize, Deserialize)]
10pub struct P2pReport {
11    pub peer: String,
12    pub fetched: usize,
13    pub merged: usize,
14}
15
16/// Helper structure managing a local Git P2P server process.
17pub struct P2pServer {
18    child: Option<Child>,
19    port: u16,
20    repo_path: PathBuf,
21}
22
23impl P2pServer {
24    /// Start a lightweight git daemon serving `refs/notes/*` for the given repository.
25    pub fn start(repo_path: &Path, port: u16) -> Result<Self> {
26        let abs_repo = if repo_path.is_relative() {
27            std::fs::canonicalize(repo_path)
28                .unwrap_or_else(|_| std::env::current_dir().unwrap_or_else(|_| repo_path.to_path_buf()))
29        } else {
30            repo_path.to_path_buf()
31        };
32
33        let child = Command::new("git")
34            .args([
35                "daemon",
36                "--reuseaddr",
37                "--export-all",
38                "--base-path-relaxed",
39                &format!("--base-path={}", abs_repo.display()),
40                &format!("--port={}", port),
41                "--enable=receive-pack",
42                abs_repo.to_str().unwrap_or("."),
43            ])
44            .spawn()
45            .with_context(|| format!("Failed to start git daemon on port {}", port))?;
46
47        Ok(Self {
48            child: Some(child),
49            port,
50            repo_path: abs_repo,
51        })
52    }
53
54    pub fn port(&self) -> u16 {
55        self.port
56    }
57
58    pub fn repo_path(&self) -> &Path {
59        &self.repo_path
60    }
61
62    /// Stop the server process.
63    pub fn stop(&mut self) -> Result<()> {
64        if let Some(mut child) = self.child.take() {
65            let _ = child.kill();
66            let _ = child.wait();
67        }
68        Ok(())
69    }
70
71    /// Wait for server process to exit.
72    pub fn wait(&mut self) -> Result<std::process::ExitStatus> {
73        if let Some(ref mut child) = self.child {
74            let status = child.wait()?;
75            Ok(status)
76        } else {
77            anyhow::bail!("Server is not running");
78        }
79    }
80}
81
82impl Drop for P2pServer {
83    fn drop(&mut self) {
84        let _ = self.stop();
85    }
86}
87
88/// Parse a peer address string into `(host, port)`.
89/// Supports:
90/// - `"192.168.1.50:9418"` -> `("192.168.1.50", 9418)`
91/// - `"192.168.1.50"` -> `("192.168.1.50", default_port)`
92/// - `"git://192.168.1.50:9418/"` -> `("192.168.1.50", 9418)`
93pub fn parse_peer_address(peer: &str, default_port: u16) -> (String, u16) {
94    let s = peer.trim();
95    let s = s.strip_prefix("git://").unwrap_or(s);
96    let s = s.trim_end_matches('/');
97
98    if let Some((host, port_str)) = s.split_once(':') {
99        if let Ok(p) = port_str.parse::<u16>() {
100            return (host.to_string(), p);
101        }
102    }
103    (s.to_string(), default_port)
104}
105
106/// Query local network IP addresses to assist peer pairing.
107pub fn get_local_ips() -> Vec<IpAddr> {
108    let mut ips = Vec::new();
109
110    // 1. Probe local routing gateway addresses (non-blocking UDP socket check)
111    for probe_target in &[
112        "192.168.1.1:80",
113        "192.168.0.1:80",
114        "10.0.0.1:80",
115        "172.16.0.1:80",
116        "8.8.8.8:80",
117    ] {
118        if let Ok(socket) = UdpSocket::bind("0.0.0.0:0") {
119            if socket.connect(probe_target).is_ok() {
120                if let Ok(local_addr) = socket.local_addr() {
121                    let ip = local_addr.ip();
122                    if !ip.is_loopback() && !ips.contains(&ip) {
123                        ips.push(ip);
124                    }
125                }
126            }
127        }
128    }
129
130    // 2. Resolve local hostname
131    if let Ok(hostname) = std::env::var("COMPUTERNAME").or_else(|_| std::env::var("HOSTNAME")) {
132        if let Ok(addrs) = format!("{}:0", hostname).to_socket_addrs() {
133            for addr in addrs {
134                let ip = addr.ip();
135                if ip.is_ipv4() && !ip.is_loopback() && !ips.contains(&ip) {
136                    ips.push(ip);
137                }
138            }
139        }
140    }
141
142    if ips.is_empty() {
143        ips.push(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)));
144    }
145
146    ips
147}
148
149/// Start a lightweight local P2P git notes sync server on the local LAN.
150pub fn serve_p2p(repo_path: &Path, port: u16) -> Result<()> {
151    let mut server = P2pServer::start(repo_path, port)?;
152    let local_ips = get_local_ips();
153
154    println!("\n\x1b[1;36m📡 git-notes P2P Server listening on port {}\x1b[0m", port);
155    println!("Repository: \x1b[33m{}\x1b[0m\n", server.repo_path().display());
156    println!("Share your connection address with your team on the local LAN:");
157
158    for ip in &local_ips {
159        println!("  • \x1b[32mgn sync p2p connect {}:{}\x1b[0m", ip, port);
160    }
161    println!("  • (Loopback): \x1b[2mgn sync p2p connect 127.0.0.1:{}\x1b[0m\n", port);
162    println!("\x1b[33m⚡ Ready! Press Ctrl+C at any time to stop serving.\x1b[0m\n");
163
164    let _ = server.wait();
165    println!("\x1b[32m✔ P2P server stopped.\x1b[0m");
166    Ok(())
167}
168
169/// Pull and merge notes directly from a peer developer's IP on the local Wi-Fi without needing internet or GitHub!
170pub fn connect_peer(
171    repo_path: &Path,
172    peer: &str,
173    strategy: &dyn MergeStrategy,
174) -> Result<P2pReport> {
175    let (host, port) = parse_peer_address(peer, 9418);
176    let remote_url = format!("git://{}:{}/", host, port);
177
178    let temp_token = format!("p2p_{}", Uuid::new_v4().simple());
179    let refspec = format!("refs/notes/*:refs/notes/{}/*", temp_token);
180
181    println!(
182        "\x1b[36m⏳ Connecting to peer at {} ({})...\x1b[0m",
183        peer, remote_url
184    );
185
186    let fetch_out = Command::new("git")
187        .current_dir(repo_path)
188        .args(["fetch", &remote_url, &refspec])
189        .output()
190        .context("Failed to connect to peer git daemon")?;
191
192    if !fetch_out.status.success() {
193        let err = String::from_utf8_lossy(&fetch_out.stderr);
194        anyhow::bail!(
195            "Failed to pull notes from peer {}: {}",
196            peer,
197            err.trim()
198        );
199    }
200
201    // Discover imported temporary refs
202    let temp_ref_prefix = format!("refs/notes/{}/", temp_token);
203    let list_out = Command::new("git")
204        .current_dir(repo_path)
205        .args(["for-each-ref", "--format=%(refname)", &temp_ref_prefix])
206        .output()
207        .context("Failed to inspect peer notes refs")?;
208
209    let imported_refs: Vec<String> = String::from_utf8_lossy(&list_out.stdout)
210        .lines()
211        .map(|l| l.trim().to_string())
212        .filter(|l| l.starts_with(&temp_ref_prefix))
213        .collect();
214
215    let engine = NotesEngine::new(repo_path);
216    let mut total_fetched = 0;
217    let mut total_merged = 0;
218
219    for temp_ref in &imported_refs {
220        let ns_suffix = temp_ref
221            .strip_prefix(&temp_ref_prefix)
222            .unwrap_or(temp_ref);
223
224        let target_ns = Namespace::from_str(ns_suffix);
225        let temp_ns = Namespace::Custom(format!("{}/{}", temp_token, ns_suffix));
226
227        let local_notes = engine.read_notes(&target_ns).unwrap_or_default();
228        let peer_notes = engine.read_notes(&temp_ns).unwrap_or_default();
229
230        total_fetched += peer_notes.len();
231
232        if !peer_notes.is_empty() {
233            let merged = strategy.merge(&local_notes, &peer_notes);
234            for note in &merged {
235                let mut note_to_write = note.clone();
236                note_to_write.namespace = target_ns.clone();
237                engine.write_note(&note_to_write)?;
238            }
239            total_merged += merged.len();
240        }
241
242        // Clean up temporary ref
243        let _ = Command::new("git")
244            .current_dir(repo_path)
245            .args(["update-ref", "-d", temp_ref])
246            .status();
247    }
248
249    Ok(P2pReport {
250        peer: format!("{}:{}", host, port),
251        fetched: total_fetched,
252        merged: total_merged,
253    })
254}
255
256#[cfg(test)]
257mod tests {
258    use super::*;
259    use gn_core::{LwwStrategy, Note};
260    use std::fs;
261
262    struct TestDir {
263        path: PathBuf,
264    }
265
266    impl TestDir {
267        fn new() -> Self {
268            let path = std::env::temp_dir().join(format!("gn_p2p_test_{}", Uuid::new_v4().simple()));
269            fs::create_dir_all(&path).unwrap();
270            Self { path }
271        }
272
273        fn path(&self) -> &Path {
274            &self.path
275        }
276    }
277
278    impl Drop for TestDir {
279        fn drop(&mut self) {
280            let _ = fs::remove_dir_all(&self.path);
281        }
282    }
283
284    fn run_git(dir: &Path, args: &[&str]) {
285        let status = Command::new("git")
286            .current_dir(dir)
287            .args(args)
288            .status()
289            .unwrap();
290        assert!(status.success(), "git {:?} failed", args);
291    }
292
293    #[test]
294    fn test_parse_peer_address() {
295        assert_eq!(
296            parse_peer_address("192.168.1.10:9418", 9418),
297            ("192.168.1.10".to_string(), 9418)
298        );
299        assert_eq!(
300            parse_peer_address("192.168.1.10", 9418),
301            ("192.168.1.10".to_string(), 9418)
302        );
303        assert_eq!(
304            parse_peer_address("git://10.0.0.5:8000/", 9418),
305            ("10.0.0.5".to_string(), 8000)
306        );
307        assert_eq!(
308            parse_peer_address("localhost:9999", 9418),
309            ("localhost".to_string(), 9999)
310        );
311    }
312
313    #[test]
314    fn test_p2p_sync_end_to_end() {
315        let port = 19428;
316        let repo_peer = TestDir::new();
317        let repo_local = TestDir::new();
318
319        // Setup peer repo
320        run_git(repo_peer.path(), &["init"]);
321        run_git(repo_peer.path(), &["config", "user.name", "Peer Dev"]);
322        run_git(repo_peer.path(), &["config", "user.email", "peer@lan.local"]);
323        fs::write(repo_peer.path().join("main.rs"), "fn main() {}\n").unwrap();
324        run_git(repo_peer.path(), &["add", "."]);
325        run_git(repo_peer.path(), &["commit", "-m", "Init"]);
326
327        let peer_engine = NotesEngine::new(repo_peer.path());
328        let peer_note = Note::new(
329            "HEAD".to_string(),
330            Some("main.rs".to_string()),
331            Some(1),
332            Some(1),
333            "LAN review comment from peer developer".to_string(),
334            "Peer Dev <peer@lan.local>".to_string(),
335            Namespace::Review,
336        );
337        peer_engine.write_note(&peer_note).unwrap();
338
339        // Setup local repo
340        run_git(repo_local.path(), &["init"]);
341        run_git(repo_local.path(), &["config", "user.name", "Local Dev"]);
342        run_git(repo_local.path(), &["config", "user.email", "local@lan.local"]);
343        fs::write(repo_local.path().join("main.rs"), "fn main() {}\n").unwrap();
344        run_git(repo_local.path(), &["add", "."]);
345        run_git(repo_local.path(), &["commit", "-m", "Init"]);
346
347        // Start P2P server on peer repo
348        let mut server = P2pServer::start(repo_peer.path(), port).unwrap();
349        std::thread::sleep(std::time::Duration::from_millis(800));
350
351        // Connect from local repo
352        let strategy = LwwStrategy;
353        let report = connect_peer(
354            repo_local.path(),
355            &format!("127.0.0.1:{}", port),
356            &strategy,
357        )
358        .unwrap();
359
360        assert_eq!(report.fetched, 1);
361        assert_eq!(report.merged, 1);
362
363        // Verify local repo received the note
364        let local_engine = NotesEngine::new(repo_local.path());
365        let notes = local_engine.read_notes(&Namespace::Review).unwrap();
366        assert_eq!(notes.len(), 1);
367        assert_eq!(notes[0].body, "LAN review comment from peer developer");
368
369        server.stop().unwrap();
370    }
371}
gn_sync/p2p.rs

gn_sync/
p2p.rs
