// Package ipc implements the local IPC server. // The UI (or any local tool) connects to 127.0.0.1:17337 and speaks // newline-delimited JSON: send IpcMessage commands, receive IpcMessage events. package ipc import ( "bufio" "encoding/json" "fmt" "log" "net" "time" "github.com/google/uuid" "github.com/waste-go/internal/mesh" "github.com/waste-go/internal/proto" ) // Run starts the IPC listener. Blocks until the listener fails. func Run(m *mesh.Mesh, port int) error { addr := fmt.Sprintf("127.0.0.1:%d", port) ln, err := net.Listen("tcp", addr) if err != nil { return fmt.Errorf("ipc listen on %s: %w", addr, err) } log.Printf("ipc: listening on %s", addr) for { conn, err := ln.Accept() if err != nil { return fmt.Errorf("ipc accept: %w", err) } log.Printf("ipc: UI client connected from %s", conn.RemoteAddr()) go handleClient(conn, m) } } func handleClient(conn net.Conn, m *mesh.Mesh) { defer conn.Close() // Subscribe to mesh events before anything else so we miss nothing. events := m.Subscribe() defer m.Unsubscribe(events) // Channel to serialize writes from two goroutines (event pusher + reply sender). writeCh := make(chan []byte, 128) // Writer goroutine — single goroutine owns the connection write side. go func() { w := bufio.NewWriter(conn) for line := range writeCh { line = append(line, '\n') if _, err := w.Write(line); err != nil { return } w.Flush() } }() // Event pusher goroutine — forwards mesh events to the UI. go func() { for evt := range events { line, err := json.Marshal(evt) if err != nil { continue } select { case writeCh <- line: default: } } }() // Send an initial state snapshot so the UI has something to render. send(writeCh, proto.IpcMessage{ Type: proto.EvtStateSnapshot, LocalPeer: ptr(m.Identity.PeerInfo()), ConnectedPeers: m.ConnectedPeers(), Rooms: []string{"general"}, }) // Command reader loop. scanner := bufio.NewScanner(conn) for scanner.Scan() { var cmd proto.IpcMessage if err := json.Unmarshal(scanner.Bytes(), &cmd); err != nil { log.Printf("ipc: bad command: %v", err) continue } handleCommand(cmd, m, writeCh) } close(writeCh) log.Printf("ipc: UI client disconnected") } func handleCommand(cmd proto.IpcMessage, m *mesh.Mesh, writeCh chan<- []byte) { switch cmd.Type { case proto.CmdSendMessage: msg := &proto.ChatMessage{ ID: uuid.NewString(), From: m.Identity.PeerID(), To: cmd.To, Room: cmd.Room, Body: cmd.Body, SentAt: time.Now(), } payload, err := json.Marshal(proto.PeerMessage{ Type: proto.MsgChat, Chat: msg, }) if err != nil { return } m.Broadcast(payload) // Echo locally so the sender sees their own message. m.Emit(proto.IpcMessage{Type: proto.EvtMessageReceived, Message: msg}) case proto.CmdConnect: if cmd.Addr == "" { send(writeCh, errMsg("connect: addr is required")) return } go func() { log.Printf("ipc: connecting to peer at %s", cmd.Addr) conn, err := net.DialTimeout("tcp", cmd.Addr, 10*time.Second) if err != nil { log.Printf("ipc: dial %s failed: %v", cmd.Addr, err) m.Emit(errMsg(fmt.Sprintf("connect to %s failed: %v", cmd.Addr, err))) return } mesh.HandleConn(conn, m, true) }() case proto.CmdGetState: send(writeCh, proto.IpcMessage{ Type: proto.EvtStateSnapshot, LocalPeer: ptr(m.Identity.PeerInfo()), ConnectedPeers: m.ConnectedPeers(), Rooms: []string{"general"}, }) case proto.CmdSendFile: send(writeCh, errMsg("file transfer not yet implemented")) default: send(writeCh, errMsg(fmt.Sprintf("unknown command: %s", cmd.Type))) } } func send(ch chan<- []byte, msg proto.IpcMessage) { line, err := json.Marshal(msg) if err != nil { return } select { case ch <- line: default: } } func errMsg(s string) proto.IpcMessage { return proto.IpcMessage{Type: proto.EvtError, ErrorMessage: s} } func ptr[T any](v T) *T { return &v }