waste-go/internal/ipc/ipc.go

// 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/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.
// Network join/leave state is daemon-scoped (shared across all IPC clients).
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)

	// networkCancel is shared across all clients — any client can join/leave,
	// and the join persists even after the commanding client disconnects.
	var (
		networkMu     sync.Mutex
		networkCancel context.CancelFunc
	)

	doJoin := func(name string) {
		networkMu.Lock()
		if networkCancel != nil {
			networkCancel()
		}
		ctx, cancel := context.WithCancel(context.Background())
		networkCancel = cancel
		networkMu.Unlock()
		go join(ctx, name)
	}

	doLeave := func() {
		networkMu.Lock()
		if networkCancel != nil {
			networkCancel()
			networkCancel = nil
		}
		networkMu.Unlock()
	}

	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, doJoin, doLeave)
	}
}

func handleClient(conn net.Conn, m *mesh.Mesh, 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
			}
			m.Broadcast(payload)
			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.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)
}