Files
p2p-chat/src/main.rs

389 lines
12 KiB
Rust

//! P2P Chat Application
//!
//! A Linux-only, terminal-based peer-to-peer communication app.
//! Chat is the primary feature, file transfer is first-class,
//! voice/camera/screen are optional, powered by PipeWire.
mod app_logic;
mod chat;
mod config;
mod file_transfer;
mod media;
mod net;
mod protocol;
mod tui;
mod web;
use std::io;
use std::path::PathBuf;
use anyhow::{Context, Result};
use clap::Parser;
use crossterm::event::{Event, EventStream};
use crossterm::execute;
use crossterm::terminal::{
disable_raw_mode, enable_raw_mode, EnterAlternateScreen, LeaveAlternateScreen,
};
use iroh::EndpointId;
use n0_future::StreamExt;
use ratatui::backend::CrosstermBackend;
use ratatui::Terminal;
use tokio::sync::mpsc;
use crate::chat::ChatState;
use crate::config::AppConfig;
use crate::file_transfer::FileTransferManager;
use crate::media::MediaState;
use crate::net::{NetEvent, NetworkManager, PeerInfo};
use crate::protocol::{CapabilitiesMessage, GossipMessage, PeerAnnounce};
use crate::tui::App;
/// P2P Chat — decentralized chat over QUIC
#[derive(Parser, Debug)]
#[command(name = "p2p-chat", about = "Peer-to-peer chat over QUIC")]
struct Cli {
/// Your display name
#[arg(short, long, default_value = "anon")]
name: String,
/// Peer endpoint ID to join (hex string)
#[arg(short, long)]
join: Option<String>,
/// Topic room ID (32-byte hex). Peers on the same topic can chat.
#[arg(short, long)]
topic: Option<String>, // Changed to Option to fallback to config
/// Download directory for received files
#[arg(short, long, default_value = "~/Downloads")]
download_dir: String,
/// Screen resolution for sharing (e.g., 1280x720, 1920x1080)
#[arg(long)]
screen_resolution: Option<String>, // Changed to Option to fallback to config
}
#[tokio::main]
async fn main() -> Result<()> {
// ... tracing init ...
// Initialize tracing to file (not stdout, since we use TUI)
let _tracing_guard = tracing_subscriber::fmt()
.with_env_filter(
tracing_subscriber::EnvFilter::from_default_env()
.add_directive("p2p_chat=debug".parse()?)
.add_directive("iroh=warn".parse()?)
.add_directive("iroh_gossip=warn".parse()?),
)
.with_writer(|| -> Box<dyn io::Write + Send> {
match std::fs::OpenOptions::new()
.create(true)
.append(true)
.open("p2p-chat.log")
{
Ok(f) => Box::new(f),
Err(_) => Box::new(io::sink()),
}
})
.with_ansi(false)
.init();
// Load config
let config = AppConfig::load().unwrap_or_else(|e| {
eprintln!("Warning: Failed to load config: {}", e);
AppConfig::default()
});
let cli = Cli::parse();
// Resolution: CLI > Config > Default
let res_str = cli
.screen_resolution
.as_deref()
.unwrap_or(&config.media.screen_resolution);
let screen_res = parse_resolution(res_str).unwrap_or((1280, 720));
// Topic: CLI > Config > Default
let topic_str = cli
.topic
.as_deref()
.or(config.network.topic.as_deref())
.unwrap_or("00000000000000000000000000000000");
let topic_bytes = parse_topic(topic_str)?;
// ... networking init ...
// Initialize networking
let (mut net_mgr, _net_tx, mut net_rx) = NetworkManager::new(topic_bytes)
.await
.context("Failed to start networking")?;
let our_id = net_mgr.our_id;
let our_id_short = format!("{}", our_id).chars().take(8).collect::<String>();
// Initialize application state
let mut chat = ChatState::new(cli.name.clone());
// Resolve download directory, handling ~
let download_path = if cli.download_dir.starts_with("~/") {
if let Ok(home) = std::env::var("HOME") {
PathBuf::from(home).join(&cli.download_dir[2..])
} else {
PathBuf::from(&cli.download_dir)
}
} else {
PathBuf::from(&cli.download_dir)
};
// Ensure download directory exists
tokio::fs::create_dir_all(&download_path).await?;
let file_mgr = FileTransferManager::new(download_path);
// Pass mic name from config if present
let media = MediaState::new(
screen_res,
config.media.mic_name.clone(),
config.media.speaker_name.clone(),
config.media.mic_bitrate,
);
// Initialize App with Theme
let theme = crate::config::Theme::from(config.ui.clone());
let mut app = App::new(theme);
// If a peer was specified, add it and bootstrap
let bootstrap_peers = if let Some(ref join_id) = cli.join {
let peer_id: EndpointId = join_id
.parse()
.context("Invalid peer ID. Expected a hex-encoded endpoint ID.")?;
vec![peer_id]
} else {
vec![]
};
// We need a sender for the gossip event loop — but we already have net_rx
// Recreate the channel and use it for gossip
let (gossip_event_tx, mut gossip_event_rx) = mpsc::channel::<NetEvent>(256);
// Join the gossip topic
net_mgr
.join_gossip(bootstrap_peers, gossip_event_tx)
.await
.context("Failed to join gossip")?;
// Announce ourselves
let announce = GossipMessage::PeerAnnounce(PeerAnnounce {
sender_name: cli.name.clone(),
});
// Don't fail if no peers yet
let _ = net_mgr.broadcast(&announce).await;
// Insert ourselves into the peer list
{
let mut peers = net_mgr.peers.lock().await;
peers.insert(
our_id,
PeerInfo {
id: our_id,
name: Some(cli.name.clone()),
capabilities: None,
is_self: true,
},
);
}
// Broadcast capabilities
let caps = GossipMessage::Capabilities(CapabilitiesMessage {
sender_name: cli.name.clone(),
..Default::default()
});
let _ = net_mgr.broadcast(&caps).await;
chat.add_system_message(format!("Welcome, {}! Your ID: {}", cli.name, our_id_short));
chat.add_system_message(format!("Full Endpoint ID: {}", our_id));
if cli.join.is_some() {
chat.add_system_message("Connecting to peer...".to_string());
} else {
chat.add_system_message(
"Waiting for peers. Share your Endpoint ID for others to join.".to_string(),
);
}
// Start Web Interface
// Start Web Interface
// Start Web Interface
tokio::spawn(crate::web::start_web_server(
media.broadcast_tx.clone(),
media.mic_broadcast.clone(),
media.cam_broadcast.clone(),
media.screen_broadcast.clone(),
));
// Initialize AppLogic
let mut app_logic = crate::app_logic::AppLogic::new(
chat,
file_mgr,
media,
net_mgr,
cli.name.clone(),
our_id_short,
);
// Setup terminal
enable_raw_mode()?;
let mut stdout = io::stdout();
execute!(stdout, EnterAlternateScreen)?;
let backend = CrosstermBackend::new(stdout);
let mut terminal = Terminal::new(backend)?;
// Main event loop
let result = run_event_loop(
&mut terminal,
&mut app,
&mut app_logic,
&mut net_rx,
&mut gossip_event_rx,
)
.await;
// Restore terminal
disable_raw_mode()?;
execute!(terminal.backend_mut(), LeaveAlternateScreen)?;
terminal.show_cursor()?;
// Shutdown networking
let _ = app_logic.net.shutdown().await;
result
}
async fn run_event_loop(
terminal: &mut Terminal<CrosstermBackend<io::Stdout>>,
app: &mut App,
logic: &mut crate::app_logic::AppLogic,
net_rx: &mut mpsc::Receiver<NetEvent>,
gossip_rx: &mut mpsc::Receiver<NetEvent>,
) -> Result<()> {
let mut event_stream = EventStream::new();
let mut interval = tokio::time::interval(std::time::Duration::from_millis(100));
loop {
// Collect peers for rendering — self is always first
let peers: Vec<_> = {
let p = logic.net.peers.lock().await;
let mut all: Vec<_> = p.values().cloned().collect();
all.sort_by_key(|p| !p.is_self); // self first
all
};
logic.connected = peers.iter().any(|p| !p.is_self);
terminal.draw(|f| {
tui::render(
f,
app,
&logic.chat,
&logic.file_mgr,
&logic.media,
&peers,
&logic.our_name,
&logic.our_id_short,
logic.connected,
);
})?;
// Wait for events
tokio::select! {
// Tick for animation handling
_ = interval.tick() => {
logic.file_mgr.check_timeouts();
}
// Terminal/keyboard events
maybe_event = event_stream.next() => {
match maybe_event {
Some(Ok(Event::Key(key))) => {
let cmd = app.handle_key(key);
if logic.handle_tui_command(cmd).await? {
return Ok(());
}
}
Some(Ok(Event::Resize(_, _))) => {
// Terminal resize — just redraw on next iteration
}
Some(Err(e)) => {
tracing::error!("Terminal event error: {}", e);
}
None => {
return Ok(());
}
_ => {}
}
}
// Network events from file transfer acceptor
Some(event) = net_rx.recv() => {
logic.handle_net_event(event).await;
}
// Gossip events
Some(event) = gossip_rx.recv() => {
logic.handle_net_event(event).await;
}
// Signal handling (Ctrl+C / SIGTERM)
_ = tokio::signal::ctrl_c() => {
// Broadcast disconnect to peers
let disconnect_msg = GossipMessage::Disconnect {
sender_name: logic.our_name.clone(),
};
let _ = logic.net.broadcast(&disconnect_msg).await;
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
logic.media.shutdown();
return Ok(());
}
}
}
}
fn parse_topic(hex_str: &str) -> Result<[u8; 32]> {
// If it's all zeros (default), generate a deterministic topic
let hex_str = hex_str.trim();
if hex_str.len() != 32 && hex_str.len() != 64 {
// Treat as a room name — hash it to get 32 bytes
use sha2::{Digest, Sha256};
let mut hasher = Sha256::new();
hasher.update(hex_str.as_bytes());
let result = hasher.finalize();
let mut bytes = [0u8; 32];
bytes.copy_from_slice(&result);
return Ok(bytes);
}
// Try parsing as hex
if hex_str.len() == 64 {
let mut bytes = [0u8; 32];
for i in 0..32 {
bytes[i] = u8::from_str_radix(&hex_str[i * 2..i * 2 + 2], 16)
.context("Invalid hex in topic")?;
}
Ok(bytes)
} else {
// 32-char string — treat as room name
use sha2::{Digest, Sha256};
let mut hasher = Sha256::new();
hasher.update(hex_str.as_bytes());
let result = hasher.finalize();
let mut bytes = [0u8; 32];
bytes.copy_from_slice(&result);
Ok(bytes)
}
}
fn parse_resolution(res: &str) -> Option<(u32, u32)> {
let parts: Vec<&str> = res.split('x').collect();
if parts.len() == 2 {
let w = parts[0].parse().ok()?;
let h = parts[1].parse().ok()?;
Some((w, h))
} else {
None
}
}