// 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" "sync" "time" "github.com/google/uuid" "github.com/waste-go/internal/invite" "github.com/waste-go/internal/mesh" "github.com/waste-go/internal/proto" ) // JoinFunc is called when the UI issues a join_network command. type JoinFunc func(ctx context.Context, networkName string) // Run starts the IPC listener. Blocks until the listener fails. // anchorURL is the configured anchor WebSocket URL (used for invite generation). // Network join/leave state is daemon-scoped (shared across all IPC clients). func Run(m *mesh.Mesh, port int, anchorURL string, 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) // networkCancel / networkName are shared across all clients — any client // can join/leave, and the join persists after the commanding client disconnects. var ( networkMu sync.Mutex networkCancel context.CancelFunc networkName string ) doJoin := func(name string) { networkMu.Lock() if networkCancel != nil { networkCancel() } ctx, cancel := context.WithCancel(context.Background()) networkCancel = cancel networkName = name networkMu.Unlock() go join(ctx, name) } doLeave := func() { networkMu.Lock() if networkCancel != nil { networkCancel() networkCancel = nil networkName = "" } networkMu.Unlock() } currentNetwork := func() string { networkMu.Lock() defer networkMu.Unlock() return networkName } 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, anchorURL, currentNetwork, doJoin, doLeave) } } func handleClient(conn net.Conn, m *mesh.Mesh, anchorURL string, currentNetwork func() string, doJoin func(string), doLeave func()) { defer conn.Close() events := m.Subscribe() defer m.Unsubscribe(events) writeCh := make(chan []byte, 128) done := make(chan struct{}) // Writer goroutine — sole owner of the write side of the connection. 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 — forwards mesh events to the UI client. // recover() guards against the rare race where writeCh is closed while a // send is in flight (closed channel panics even inside select). go func() { defer func() { recover() }() //nolint:errcheck for { select { case evt, ok := <-events: if !ok { return } line, err := json.Marshal(evt) if err != nil { continue } select { case writeCh <- line: case <-done: return } case <-done: return } } }() send := func(msg proto.IpcMessage) { defer func() { recover() }() //nolint:errcheck line, err := json.Marshal(msg) if err != nil { return } select { case writeCh <- line: case <-done: } } send(proto.IpcMessage{ Type: proto.EvtStateSnapshot, LocalPeer: ptr(m.Identity.PeerInfo()), ConnectedPeers: m.ConnectedPeers(), Rooms: []string{"general"}, }) 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 } if cmd.To != nil { // DM — send only to the named recipient. m.SendTo(*cmd.To, payload) } else { m.Broadcast(payload) } m.SaveMessage(msg) m.Emit(proto.IpcMessage{Type: proto.EvtMessageReceived, Message: msg}) case proto.CmdJoinNetwork: if cmd.NetworkName == "" { send(errMsg("join_network: network_name is required")) continue } doJoin(cmd.NetworkName) case proto.CmdLeaveNetwork: doLeave() case proto.CmdGetState: send(proto.IpcMessage{ Type: proto.EvtStateSnapshot, LocalPeer: ptr(m.Identity.PeerInfo()), ConnectedPeers: m.ConnectedPeers(), Rooms: []string{"general"}, }) case proto.CmdGenerateInvite: net := currentNetwork() if net == "" { send(errMsg("generate_invite: not currently joined to a network")) continue } if anchorURL == "" { send(errMsg("generate_invite: daemon was started without -anchor flag")) continue } inv, err := invite.Encode(anchorURL, net) if err != nil { send(errMsg(fmt.Sprintf("generate_invite: %v", err))) continue } send(proto.IpcMessage{Type: proto.EvtInviteGenerated, InviteString: inv}) case proto.CmdSendFile: send(errMsg("file transfer not yet implemented")) default: send(errMsg(fmt.Sprintf("unknown command: %s", cmd.Type))) } } close(done) close(writeCh) log.Printf("ipc: UI client disconnected") } 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) }