// 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" "context" "crypto/rand" "encoding/hex" "encoding/json" "fmt" "log" "net" "time" "github.com/google/uuid" "github.com/waste-go/internal/mesh" "github.com/waste-go/internal/proto" ) // JoinFunc is called when the UI issues a join_network command. // It should connect to the anchor and block until done or ctx is cancelled. type JoinFunc func(ctx context.Context, networkName string) // Run starts the IPC listener. Blocks until the listener fails. func Run(m *mesh.Mesh, port int, join JoinFunc) 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, join) } } func handleClient(conn net.Conn, m *mesh.Mesh, join JoinFunc) { defer conn.Close() events := m.Subscribe() defer m.Unsubscribe(events) writeCh := make(chan []byte, 128) go func() { w := bufio.NewWriter(conn) for line := range writeCh { line = append(line, '\n') if _, err := w.Write(line); err != nil { return } w.Flush() } }() go func() { for evt := range events { line, err := json.Marshal(evt) if err != nil { continue } select { case writeCh <- line: default: } } }() send(writeCh, proto.IpcMessage{ Type: proto.EvtStateSnapshot, LocalPeer: ptr(m.Identity.PeerInfo()), ConnectedPeers: m.ConnectedPeers(), Rooms: []string{"general"}, }) // Track an active network join so we can cancel it on leave_network. var ( networkCancel context.CancelFunc ) 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 } switch cmd.Type { case proto.CmdSendMessage: msg := &proto.ChatMessage{ Mid: randomHex(16), 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 { continue } m.Broadcast(payload) m.Emit(proto.IpcMessage{Type: proto.EvtMessageReceived, Message: msg}) case proto.CmdJoinNetwork: if cmd.NetworkName == "" { send(writeCh, errMsg("join_network: network_name is required")) continue } if networkCancel != nil { networkCancel() // leave any previous network } ctx, cancel := context.WithCancel(context.Background()) networkCancel = cancel go join(ctx, cmd.NetworkName) case proto.CmdLeaveNetwork: if networkCancel != nil { networkCancel() networkCancel = nil } 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))) } } if networkCancel != nil { networkCancel() } close(writeCh) log.Printf("ipc: UI client disconnected") } 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 } func randomHex(n int) string { b := make([]byte, n) rand.Read(b) //nolint:errcheck return hex.EncodeToString(b) }