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.
//
// Backward compat: commands without network_id route to the first joined network.
package ipc

import (
	"bufio"
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"log"
	"net"
	"net/http"
	"strings"
	"time"

	"nhooyr.io/websocket"

	"github.com/waste-go/internal/crypto"
	"github.com/waste-go/internal/invite"
	"github.com/waste-go/internal/netmgr"
	"github.com/waste-go/internal/proto"
	"github.com/waste-go/internal/shares"
)

// RunWS starts a WebSocket IPC server on 127.0.0.1:wsPort.
// Each WebSocket connection gets the same handleClient treatment as TCP.
// The OriginPatterns option allows connections from local dev servers.
func RunWS(mgr *netmgr.Manager, wsPort int) error {
	addr := fmt.Sprintf("127.0.0.1:%d", wsPort)
	log.Printf("ipc: WS listening on %s", addr)
	return http.ListenAndServe(addr, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{
			OriginPatterns: []string{"*"},
		})
		if err != nil {
			log.Printf("ipc: ws accept: %v", err)
			return
		}
		log.Printf("ipc: WS client connected")
		nc := websocket.NetConn(r.Context(), conn, websocket.MessageText)
		handleClient(nc, mgr)
	}))
}

// Run starts the IPC listener. Blocks until the listener fails.
func Run(mgr *netmgr.Manager, 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, mgr)
	}
}

func handleClient(conn net.Conn, mgr *netmgr.Manager) {
	defer conn.Close()

	events := mgr.Subscribe()
	defer mgr.Unsubscribe(events)

	writeCh := make(chan []byte, 128)
	done := make(chan struct{})
	writerDone := make(chan struct{})

	// Writer goroutine.
	go func() {
		defer close(writerDone)
		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 Manager events to the UI client.
	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 initial state snapshot.
	send(stateSnapshot(mgr))
	// Send stored history for each room so the UI is populated on connect.
	sendStoredHistory(mgr, send)

	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.CmdJoinNetwork:
			var (
				netID string
				err   error
			)
			switch {
			case cmd.NetworkName != "":
				netID, err = mgr.Join(cmd.NetworkName, cmd.ShareDir)
			case len(cmd.NetworkHash) == 64:
				netID, err = mgr.JoinByHash(cmd.NetworkHash, cmd.ShareDir)
			default:
				send(errMsg("join_network: network_name or network_hash (64 hex chars) required"))
				continue
			}
			if err != nil {
				send(errMsg(fmt.Sprintf("join_network: %v", err)))
				continue
			}
			if n, ok := mgr.Get(netID); ok {
				if cmd.RequireInvite {
					n.Mesh.RequireInvite = true
				}
				if cmd.InviteString != "" {
					n.Mesh.InviteString = cmd.InviteString
				}
			}
			_ = netID

		case proto.CmdLeaveNetwork:
			if cmd.NetworkID != "" {
				mgr.Leave(cmd.NetworkID)
			} else {
				// Backward compat: leave the first joined network.
				if n := mgr.Default(); n != nil {
					mgr.Leave(n.ID)
				}
			}

		case proto.CmdSendMessage:
			n := mgr.Resolve(cmd.NetworkID)
			if n == nil {
				send(errMsg("send_message: not joined to any network"))
				continue
			}
			ts := time.Now().UnixMilli()
			if cmd.To != nil {
				// DM → spec "pm" type: flat {type, mid, text, ts} on the wire
				mid := randomHex(16)
				wire, err := json.Marshal(proto.PeerMessage{
					Type: proto.MsgPm,
					Mid:  mid,
					Text: cmd.Body,
					Ts:   ts,
				})
				if err != nil {
					continue
				}
				n.Mesh.SendTo(*cmd.To, wire)
				// Store locally with dm:<short-id> room convention
				local := &proto.ChatMessage{
					Mid:  mid,
					From: n.Identity.PeerID(),
					To:   cmd.To,
					Room: "dm:" + (*cmd.To).Short(),
					Text: cmd.Body,
					Ts:   ts,
				}
				n.Mesh.SaveMessage(local)
				n.Mesh.Emit(proto.IpcMessage{
					Type:      proto.EvtMessageReceived,
					NetworkID: n.ID,
					Message:   local,
				})
			} else {
				// Group chat → spec "chat" type: flat {type, mid, room, text, ts}
				mid := randomHex(16)
				msgID := proto.ComputeMsgID(n.Identity.PeerID(), cmd.Room, ts, cmd.Body)
				wire, err := json.Marshal(proto.PeerMessage{
					Type: proto.MsgChat,
					Mid:  mid,
					Room: cmd.Room,
					Text: cmd.Body,
					Ts:   ts,
				})
				if err != nil {
					continue
				}
				n.Mesh.Broadcast(wire)
				local := &proto.ChatMessage{
					Mid:   mid,
					MsgID: msgID,
					From:  n.Identity.PeerID(),
					Room:  cmd.Room,
					Text:  cmd.Body,
					Ts:    ts,
				}
				n.Mesh.SaveMessage(local)
				n.Mesh.Emit(proto.IpcMessage{
					Type:      proto.EvtMessageReceived,
					NetworkID: n.ID,
					Message:   local,
				})
			}

		case proto.CmdSendReaction:
			n := mgr.Resolve(cmd.NetworkID)
			if n == nil {
				send(errMsg("send_reaction: not joined to any network"))
				continue
			}
			if cmd.ReactionMID == "" || cmd.ReactionEmoji == "" {
				send(errMsg("send_reaction: reaction_mid and reaction_emoji are required"))
				continue
			}
			wire, err := json.Marshal(proto.PeerMessage{
				Type:          proto.MsgReaction,
				ReactionMID:   cmd.ReactionMID,
				ReactionEmoji: cmd.ReactionEmoji,
			})
			if err != nil {
				continue
			}
			n.Mesh.Broadcast(wire)
			n.Mesh.SaveReaction(cmd.ReactionMID, cmd.ReactionEmoji, string(n.Identity.PeerID()))
			n.Mesh.Emit(proto.IpcMessage{
				Type:          proto.EvtReaction,
				NetworkID:     n.ID,
				PeerID:        ptr(n.Identity.PeerID()),
				ReactionMID:   cmd.ReactionMID,
				ReactionEmoji: cmd.ReactionEmoji,
			})

		case proto.CmdCreateRoom:
			n := mgr.Resolve(cmd.NetworkID)
			if n == nil {
				send(errMsg("create_room: not joined to any network"))
				continue
			}
			name := strings.TrimSpace(cmd.Room)
			if name == "" || name == "general" {
				send(errMsg("create_room: room name is required and cannot be 'general'"))
				continue
			}
			if err := n.Store.SaveRoom(name); err != nil {
				send(errMsg(fmt.Sprintf("create_room: %v", err)))
				continue
			}
			send(proto.IpcMessage{Type: proto.EvtRoomCreated, NetworkID: n.ID, Room: name})

		case proto.CmdGetState:
			send(stateSnapshot(mgr))

		case proto.CmdGetFileList:
			n := mgr.Resolve(cmd.NetworkID)
			if n == nil {
				send(errMsg("get_file_list: not joined to any network"))
				continue
			}
			if cmd.PeerID == nil || *cmd.PeerID == n.Identity.PeerID() {
				send(proto.IpcMessage{
					Type:      proto.EvtFileList,
					NetworkID: n.ID,
					PeerID:    ptr(n.Identity.PeerID()),
					Files:     mgr.ScanAllShares(n.ID),
				})
			} else {
				req, err := json.Marshal(proto.PeerMessage{Type: proto.MsgFileListReq})
				if err != nil {
					continue
				}
				if !n.Mesh.SendTo(*cmd.PeerID, req) {
					send(errMsg(fmt.Sprintf("get_file_list: peer %s not connected", (*cmd.PeerID).Short())))
				}
			}

		case proto.CmdAddShare:
			if cmd.Path == "" {
				send(errMsg("add_share: path is required"))
				continue
			}
			networks := cmd.ShareNetworks
			if len(networks) == 0 {
				networks = []string{"*"}
			}
			if err := mgr.Shares.Add(shares.Share{Path: cmd.Path, Networks: networks}); err != nil {
				send(errMsg(fmt.Sprintf("add_share: %v", err)))
				continue
			}
			send(sharesListMsg(mgr))

		case proto.CmdRemoveShare:
			if cmd.Path == "" {
				send(errMsg("remove_share: path is required"))
				continue
			}
			if err := mgr.Shares.Remove(cmd.Path); err != nil {
				send(errMsg(fmt.Sprintf("remove_share: %v", err)))
				continue
			}
			send(sharesListMsg(mgr))

		case proto.CmdListShares:
			send(sharesListMsg(mgr))

		case proto.CmdGenerateInvite:
			n := mgr.Resolve(cmd.NetworkID)
			if n == nil {
				send(errMsg("generate_invite: not currently joined to a network"))
				continue
			}
			if mgr.AnchorURL() == "" {
				send(errMsg("generate_invite: daemon was started without -anchor flag"))
				continue
			}
			inv, err := invite.EncodeSigned(mgr.AnchorURL(), n.Name, n.Identity)
			if err != nil {
				send(errMsg(fmt.Sprintf("generate_invite: %v", err)))
				continue
			}
			send(proto.IpcMessage{
				Type:            proto.EvtInviteGenerated,
				NetworkID:       n.ID,
				InviteGenerated: inv,
			})

		case proto.CmdSetShareDir:
			n := mgr.Resolve(cmd.NetworkID)
			if n == nil {
				send(errMsg("set_share_dir: not joined to any network"))
				continue
			}
			if !mgr.SetShareDir(n.ID, cmd.Path) {
				send(errMsg("set_share_dir: network not found"))
				continue
			}

		case proto.CmdSetDownloadDir:
			n := mgr.Resolve(cmd.NetworkID)
			if n == nil {
				send(errMsg("set_download_dir: not joined to any network"))
				continue
			}
			if cmd.Path == "" {
				send(errMsg("set_download_dir: path is required"))
				continue
			}
			if !mgr.SetDownloadDir(n.ID, cmd.Path) {
				send(errMsg("set_download_dir: network not found"))
				continue
			}
			send(stateSnapshot(mgr))

		case proto.CmdSendFile:
			n := mgr.Resolve(cmd.NetworkID)
			if n == nil {
				send(errMsg("send_file: not joined to any network"))
				continue
			}
			if cmd.PeerID == nil {
				send(errMsg("send_file: peer_id is required"))
				continue
			}
			if cmd.Path == "" {
				send(errMsg("send_file: path is required"))
				continue
			}
			if err := n.Mesh.OfferFile(*cmd.PeerID, cmd.Path); err != nil {
				send(errMsg(fmt.Sprintf("send_file: %v", err)))
			}

		case proto.CmdExportIdentity:
			if cmd.Passphrase == "" {
				send(errMsg("export_identity: passphrase is required"))
				continue
			}
			blob, err := crypto.ExportIdentity(mgr.MasterIdentity(), cmd.Passphrase)
			if err != nil {
				send(errMsg(fmt.Sprintf("export_identity: %v", err)))
				continue
			}
			send(proto.IpcMessage{
				Type:   proto.EvtIdentityExported,
				Backup: string(blob),
			})

		case proto.CmdImportIdentity:
			if cmd.Passphrase == "" || cmd.Backup == "" {
				send(errMsg("import_identity: passphrase and backup are required"))
				continue
			}
			_, err := crypto.ImportIdentity([]byte(cmd.Backup), cmd.Passphrase)
			if err != nil {
				send(errMsg(fmt.Sprintf("import_identity: %v", err)))
				continue
			}
			// Import is intentionally read-only here: returns the decrypted identity
			// for the caller to verify before committing. Actual on-disk replacement
			// requires a daemon restart with --import flag (see cmd/daemon).
			send(proto.IpcMessage{Type: proto.EvtIdentityImported})

		default:
			send(errMsg(fmt.Sprintf("unknown command: %s", cmd.Type)))
		}
	}

	close(done)
	close(writeCh)
	<-writerDone // wait for writer to flush before conn.Close() fires
	log.Printf("ipc: UI client disconnected")
}

// stateSnapshot builds a state_snapshot covering all joined networks.
// Backward compat: local_peer and connected_peers are populated from the first network.
func stateSnapshot(mgr *netmgr.Manager) proto.IpcMessage {
	all := mgr.All()

	master := mgr.MasterIdentity()
	msg := proto.IpcMessage{
		Type:        proto.EvtStateSnapshot,
		Rooms:       []string{"general"},
		MasterAlias: master.Alias,
		MasterID:    string(master.PeerID()),
	}

	var netInfos []proto.NetworkInfo
	for _, n := range all {
		pi := n.Identity.PeerInfo()
		netInfos = append(netInfos, proto.NetworkInfo{
			NetworkID:   n.ID,
			NetworkName: n.Name,
			LocalPeer:   &pi,
			ShareDir:    n.Mesh.ShareDir,
			DownloadDir: n.Mesh.DownloadDir,
		})
	}
	msg.Networks = netInfos

	// Backward-compat fields — populated from the first network when one exists.
	if len(all) > 0 {
		pi := all[0].Identity.PeerInfo()
		msg.LocalPeer = &pi
		msg.ConnectedPeers = all[0].Mesh.ConnectedPeers()
		if extra, err := all[0].Store.Rooms(); err == nil {
			for _, r := range extra {
				msg.Rooms = append(msg.Rooms, r)
			}
		}
		// Include all historically-known peers so the UI can resolve aliases in history.
		if known, err := all[0].Store.KnownPeers(); err == nil {
			connected := map[proto.PeerID]bool{}
			for _, p := range msg.ConnectedPeers {
				connected[p.ID] = true
			}
			for id, alias := range known {
				if connected[id] {
					continue // already in ConnectedPeers
				}
				msg.KnownPeers = append(msg.KnownPeers, proto.PeerInfo{
					ID:    id,
					Alias: alias,
				})
			}
		}
	}

	return msg
}

// sendStoredHistory pushes recent messages for all known rooms to a newly-connected IPC client.
func sendStoredHistory(mgr *netmgr.Manager, send func(proto.IpcMessage)) {
	all := mgr.All()
	if len(all) == 0 {
		return
	}
	n := all[0] // use first network; multi-network history follows same pattern
	if n.Store == nil {
		return
	}
	rooms := []string{"general"}
	if extra, err := n.Store.Rooms(); err == nil {
		rooms = append(rooms, extra...)
	}
	for _, room := range rooms {
		msgs, err := n.Store.RecentMessagesSince(room, 0, 200)
		if err != nil || len(msgs) == 0 {
			continue
		}
		send(proto.IpcMessage{
			Type:      proto.EvtHistoryLoaded,
			NetworkID: n.ID,
			Room:      room,
			Messages:  msgs,
		})
		// Send stored reactions for this room's messages.
		rxns, err := n.Store.ReactionsForRoom(room)
		if err != nil {
			continue
		}
		for mid, byEmoji := range rxns {
			for emoji, fromPeers := range byEmoji {
				for _, fromPeer := range fromPeers {
					pid := proto.PeerID(fromPeer)
					send(proto.IpcMessage{
						Type:          proto.EvtReaction,
						NetworkID:     n.ID,
						PeerID:        &pid,
						ReactionMID:   mid,
						ReactionEmoji: emoji,
					})
				}
			}
		}
	}
}

func errMsg(s string) proto.IpcMessage {
	return proto.IpcMessage{Type: proto.EvtError, ErrorMessage: s}
}

func sharesListMsg(mgr *netmgr.Manager) proto.IpcMessage {
	all := mgr.Shares.All()
	entries := make([]proto.ShareEntry, len(all))
	for i, sh := range all {
		entries[i] = proto.ShareEntry{Path: sh.Path, Networks: sh.Networks}
	}
	return proto.IpcMessage{Type: proto.EvtSharesList, Shares: entries}
}

// ensure shares import is used
var _ = shares.Share{}

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)
}

// autoJoin is called by the daemon when --join is provided at startup.
// It joins the network before the IPC listener starts accepting clients.
func AutoJoin(ctx context.Context, mgr *netmgr.Manager, networkName string) {
	_ = ctx // Manager owns the context internally
	if _, err := mgr.Join(networkName, ""); err != nil {
		log.Printf("ipc: auto-join %q failed: %v", networkName, err)
	}
}