9ad3c96d43
history_loaded events were fired when peers exchanged history gossip, but the browser UI often connects to the daemon after that handshake has already happened. Now the daemon pushes recent messages for all known rooms immediately after the state_snapshot on each new IPC connection, so the browser always gets history regardless of timing. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
527 lines
14 KiB
Go
527 lines
14 KiB
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.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)
|
|
}
|
|
}
|
|
}
|
|
|
|
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,
|
|
})
|
|
}
|
|
}
|
|
|
|
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)
|
|
}
|
|
}
|