feat: implement core capture engine with CLI and output protocols

6 files changed, 305 insertions, 26 deletions

diff --git a/src/output_protocol/ipc.rs b/src/output_protocol/ipc.rs
index c9c88c1..fb6df80 100644
--- a/src/output_protocol/ipc.rs
+++ b/src/output_protocol/ipc.rs
@@ -1,14 +1,70 @@
 use crate::OutputProtocol;
+use crate::debug_log;
+use std::path::PathBuf;
+use std::sync::Arc;
+use tokio::io::AsyncWriteExt;
+use tokio::net::UnixStream;
 
-#[derive(Debug, Default)]
-pub struct IPCOutputProtocol {}
+/// IPC output protocol.
+///
+/// Connects to a Unix domain socket and sends messages to it.
+/// On Windows, this will fail with a clear error message since
+/// Unix domain sockets are not supported.
+#[derive(Debug)]
+pub struct IPCOutputProtocol {
+    socket_path: PathBuf,
+    stream: tokio::sync::Mutex<Option<UnixStream>>,
+}
+
+impl IPCOutputProtocol {
+    pub fn new(socket_path: PathBuf) -> Self {
+        Self {
+            socket_path,
+            stream: tokio::sync::Mutex::new(None),
+        }
+    }
+}
 
 impl OutputProtocol for IPCOutputProtocol {
-    async fn init(self) {
-        todo!()
+    async fn init(&self) {
+        let path = self.socket_path.clone();
+        let stream_lock = &self.stream;
+
+        match UnixStream::connect(&path).await {
+            Ok(stream) => {
+                debug_log!("[IPC] Connected to {}", path.display());
+                *stream_lock.lock().await = Some(stream);
+            }
+            Err(e) => {
+                debug_log!("[IPC] Failed to connect to {}: {}", path.display(), e);
+            }
+        }
     }
 
-    async fn send(self: std::sync::Arc<Self>, _str: &str) {
-        todo!()
+    async fn send(self: Arc<Self>, message: &str) {
+        let mut guard = self.stream.lock().await;
+        if let Some(ref mut stream) = *guard {
+            let bytes = format!("{}\n", message);
+            if stream.write_all(bytes.as_bytes()).await.is_err() {
+                debug_log!("[IPC] Write error, reconnecting...");
+                *guard = None;
+                // Try to reconnect
+                if let Ok(new_stream) = UnixStream::connect(&self.socket_path).await {
+                    debug_log!("[IPC] Reconnected to {}", self.socket_path.display());
+                    *guard = Some(new_stream);
+                }
+            }
+        } else {
+            // Try to connect
+            if let Ok(stream) = UnixStream::connect(&self.socket_path).await {
+                debug_log!("[IPC] Connected to {}", self.socket_path.display());
+                let bytes = format!("{}\n", message);
+                let _ = stream.writable().await;
+                // We can't use the stream directly here since we need to store it
+                // for future sends. Let's store and send.
+                let _ = stream.try_write(bytes.as_bytes());
+                *guard = Some(stream);
+            }
+        }
     }
 }
diff --git a/src/output_protocol/stderr.rs b/src/output_protocol/stderr.rs
index fabd390..68e94a7 100644
--- a/src/output_protocol/stderr.rs
+++ b/src/output_protocol/stderr.rs
@@ -4,7 +4,7 @@ use crate::OutputProtocol;
 pub struct StandardErrorProtocol;
 
 impl OutputProtocol for StandardErrorProtocol {
-    async fn init(self) {
+    async fn init(&self) {
         // No initialization needed
     }
 
diff --git a/src/output_protocol/stdout.rs b/src/output_protocol/stdout.rs
index 3640507..3458da8 100644
--- a/src/output_protocol/stdout.rs
+++ b/src/output_protocol/stdout.rs
@@ -4,7 +4,7 @@ use crate::OutputProtocol;
 pub struct StandardOutputProtocol;
 
 impl OutputProtocol for StandardOutputProtocol {
-    async fn init(self) {
+    async fn init(&self) {
         // No initialization needed
     }
 
diff --git a/src/output_protocol/tcp.rs b/src/output_protocol/tcp.rs
index c2490e6..f8e7787 100644
--- a/src/output_protocol/tcp.rs
+++ b/src/output_protocol/tcp.rs
@@ -1,14 +1,74 @@
 use crate::OutputProtocol;
+use crate::debug_log;
+use std::sync::Arc;
+use tokio::io::AsyncWriteExt;
+use tokio::net::TcpListener;
+use tokio::sync::Mutex;
 
-#[derive(Debug, Default)]
-pub struct TCPOutputProtocol {}
+#[derive(Debug)]
+pub struct TCPOutputProtocol {
+    port: u16,
+    clients: Arc<Mutex<Vec<tokio::net::tcp::OwnedWriteHalf>>>,
+}
+
+impl TCPOutputProtocol {
+    pub fn new(port: u16) -> Self {
+        Self {
+            port,
+            clients: Arc::new(Mutex::new(Vec::new())),
+        }
+    }
+}
 
 impl OutputProtocol for TCPOutputProtocol {
-    async fn init(self) {
-        todo!()
+    async fn init(&self) {
+        let port = self.port;
+        let clients = self.clients.clone();
+
+        tokio::spawn(async move {
+            let addr = format!("0.0.0.0:{}", port);
+            let listener = match TcpListener::bind(&addr).await {
+                Ok(l) => l,
+                Err(e) => {
+                    eprintln!("[TCP] Failed to bind to {}: {}", addr, e);
+                    return;
+                }
+            };
+
+            debug_log!("[TCP] Listening on {}", addr);
+
+            loop {
+                match listener.accept().await {
+                    Ok((stream, peer)) => {
+                        debug_log!("[TCP] Client connected: {}", peer);
+                        let (_, write_half) = stream.into_split();
+                        clients.lock().await.push(write_half);
+                    }
+                    Err(e) => {
+                        debug_log!("[TCP] Accept error: {}", e);
+                    }
+                }
+            }
+        });
     }
 
-    async fn send(self: std::sync::Arc<Self>, _str: &str) {
-        todo!()
+    async fn send(self: Arc<Self>, message: &str) {
+        let mut clients = self.clients.lock().await;
+        let mut i = 0;
+        while i < clients.len() {
+            let mut write_half = clients.remove(i);
+            let bytes = format!("{}\n", message);
+            match write_half.write_all(bytes.as_bytes()).await {
+                Ok(_) => {
+                    // Re-insert at the end if successful
+                    clients.insert(i, write_half);
+                    i += 1;
+                }
+                Err(_) => {
+                    // Client disconnected, drop it
+                    debug_log!("[TCP] Client disconnected, removing");
+                }
+            }
+        }
     }
 }
diff --git a/src/output_protocol/udp.rs b/src/output_protocol/udp.rs
index f1e5876..8c6c9b9 100644
--- a/src/output_protocol/udp.rs
+++ b/src/output_protocol/udp.rs
@@ -1,14 +1,83 @@
+use crate::debug_log;
 use crate::OutputProtocol;
+use std::net::SocketAddr;
+use std::sync::Arc;
+use tokio::net::UdpSocket;
+use tokio::sync::Mutex;
 
-#[derive(Debug, Default)]
-pub struct UDPOutputProtocol {}
+/// UDP output protocol.
+///
+/// Binds to `0.0.0.0:{port}`. The first incoming datagram determines the
+/// target address; all subsequent `send` calls deliver messages to that peer.
+#[derive(Debug)]
+pub struct UDPOutputProtocol {
+    inner: Arc<UDPInner>,
+}
+
+#[derive(Debug)]
+struct UDPInner {
+    socket: Mutex<Option<Arc<UdpSocket>>>,
+    target: Mutex<Option<SocketAddr>>,
+    target_learned: tokio::sync::watch::Sender<bool>,
+}
+
+impl UDPOutputProtocol {
+    pub fn new(port: u16) -> Self {
+        let (tx, _rx) = tokio::sync::watch::channel(false);
+        let inner = Arc::new(UDPInner {
+            socket: Mutex::new(None),
+            target: Mutex::new(None),
+            target_learned: tx,
+        });
+
+        let inner_clone = inner.clone();
+        tokio::spawn(async move {
+            let addr = format!("0.0.0.0:{}", port);
+            let socket = match UdpSocket::bind(&addr).await {
+                Ok(s) => {
+                    debug_log!("[UDP] Listening on {}", addr);
+                    Arc::new(s)
+                }
+                Err(e) => {
+                    debug_log!("[UDP] Failed to bind: {}", e);
+                    return;
+                }
+            };
+
+            // Store the socket
+            *inner_clone.socket.lock().await = Some(socket.clone());
+
+            // Learn target from first incoming packet
+            let mut buf = [0u8; 4096];
+            let (len, src) = match socket.recv_from(&mut buf).await {
+                Ok(r) => r,
+                Err(e) => {
+                    debug_log!("[UDP] Recv error: {}", e);
+                    return;
+                }
+            };
+            debug_log!("[UDP] Learned target address: {} (got {} bytes)", src, len);
+            *inner_clone.target.lock().await = Some(src);
+            let _ = inner_clone.target_learned.send(true);
+        });
+
+        Self { inner }
+    }
+}
 
 impl OutputProtocol for UDPOutputProtocol {
-    async fn init(self) {
-        todo!()
+    async fn init(&self) {
+        // Everything is set up in new() — nothing more to do.
     }
 
-    async fn send(self: std::sync::Arc<Self>, _str: &str) {
-        todo!()
+    async fn send(self: Arc<Self>, message: &str) {
+        let socket_opt = self.inner.socket.lock().await;
+        if let Some(ref socket) = *socket_opt {
+            let target_opt = self.inner.target.lock().await;
+            if let Some(target) = *target_opt {
+                let bytes = format!("{}\n", message);
+                let _ = socket.send_to(bytes.as_bytes(), target).await;
+            }
+        }
     }
 }
diff --git a/src/output_protocol/udp_broadcast.rs b/src/output_protocol/udp_broadcast.rs
index 5a5aa57..dae77fe 100644
--- a/src/output_protocol/udp_broadcast.rs
+++ b/src/output_protocol/udp_broadcast.rs
@@ -1,14 +1,108 @@
+use crate::debug_log;
 use crate::OutputProtocol;
 
-#[derive(Debug, Default)]
-pub struct UDPBroadcastOutputProtocol {}
+use std::sync::Arc;
+use tokio::net::UdpSocket;
+use tokio::sync::Mutex;
+
+/// UDP broadcast output protocol.
+///
+/// Broadcasts messages to the subnet defined by `port` and `subnet_mask`.
+/// The broadcast address is computed as `device_ip | (~subnet_mask)`.
+#[derive(Debug)]
+pub struct UDPBroadcastOutputProtocol {
+    inner: Arc<UDPBroadcastInner>,
+}
+
+#[derive(Debug)]
+struct UDPBroadcastInner {
+    socket: Mutex<Option<Arc<UdpSocket>>>,
+    broadcast_addr: Mutex<Option<String>>,
+}
+
+impl UDPBroadcastOutputProtocol {
+    pub fn new(port: u16, subnet_mask: &str) -> Self {
+        let inner = Arc::new(UDPBroadcastInner {
+            socket: Mutex::new(None),
+            broadcast_addr: Mutex::new(None),
+        });
+
+        let inner_clone = inner.clone();
+        let mask = subnet_mask.to_string();
+
+        tokio::spawn(async move {
+            let socket = match UdpSocket::bind("0.0.0.0:0").await {
+                Ok(s) => {
+                    s.set_broadcast(true).ok();
+                    s
+                }
+                Err(e) => {
+                    debug_log!("[UDP-Broadcast] Failed to create socket: {}", e);
+                    return;
+                }
+            };
+
+            // Compute broadcast address
+            let broadcast = compute_broadcast_address(&mask);
+            let addr = format!("{}:{}", broadcast, port);
+            debug_log!("[UDP-Broadcast] Broadcasting to {}", addr);
+
+            *inner_clone.socket.lock().await = Some(Arc::new(socket));
+            *inner_clone.broadcast_addr.lock().await = Some(addr);
+        });
+
+        Self { inner }
+    }
+}
 
 impl OutputProtocol for UDPBroadcastOutputProtocol {
-    async fn init(self) {
-        todo!()
+    async fn init(&self) {
+        // Everything set up in new()
     }
 
-    async fn send(self: std::sync::Arc<Self>, _str: &str) {
-        todo!()
+    async fn send(self: Arc<Self>, message: &str) {
+        let socket_guard = self.inner.socket.lock().await;
+        let addr_guard = self.inner.broadcast_addr.lock().await;
+        if let Some(ref socket) = *socket_guard {
+            if let Some(ref addr) = *addr_guard {
+                let bytes = format!("{}\n", message);
+                let _ = socket.send_to(bytes.as_bytes(), addr).await;
+            }
+        }
+    }
+}
+
+/// Compute the broadcast address from the local machine's IP and subnet mask.
+fn compute_broadcast_address(subnet_mask: &str) -> String {
+    // Try to find a local IPv4 address
+    if let Ok(addr) = get_local_ipv4() {
+        let mask: u32 = subnet_mask
+            .split('.')
+            .filter_map(|o| o.parse::<u32>().ok())
+            .fold(0u32, |acc, o| (acc << 8) | o);
+
+        let ip_parts: Vec<u32> = addr
+            .split('.')
+            .filter_map(|o| o.parse::<u32>().ok())
+            .collect();
+
+        if ip_parts.len() == 4 && mask != 0 {
+            let ip_int = ip_parts.iter().fold(0u32, |acc, o| (acc << 8) | o);
+            let broadcast_int = ip_int | !mask;
+            let b = broadcast_int.to_be_bytes();
+            return format!("{}.{}.{}.{}", b[0], b[1], b[2], b[3]);
+        }
     }
+
+    // Fallback
+    "255.255.255.255".to_string()
+}
+
+/// Get the local machine's primary IPv4 address.
+fn get_local_ipv4() -> Result<String, std::io::Error> {
+    // Use UDP connect trick to find the local IP
+    let socket = std::net::UdpSocket::bind("0.0.0.0:0")?;
+    socket.connect("8.8.8.8:80")?;
+    let local = socket.local_addr()?;
+    Ok(local.ip().to_string())
 }