explewd/waste-go
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 5 Wiki Activity

Multi-network foundation: netmgr, derived identities, additive IPC protocol

Browse Source 
YAW/2 peer wire protocol is unchanged. Changes are local only.

internal/crypto:
- DeriveForNetwork(master, networkHash) — HKDF-SHA256 from master seed + network hash;
  same master + same network always produces the same Ed25519 keypair (stable peer ID)

internal/proto:
- NetworkInfo type (network_id, network_name, local_peer)
- NetworkID field on IpcMessage (optional; commands default to first network when absent)
- Networks []NetworkInfo on state_snapshot (additive alongside existing local_peer)
- EvtNetworkJoined / EvtNetworkLeft events

internal/netmgr (new):
- Manager holds N independent Network contexts (derived identity, mesh, store, anchor)
- Join(name) creates context, derives identity, opens per-network DB, starts anchor client
- Leave(id) / LeaveAll() cancel contexts and close stores
- Resolve(netID) returns named network, or Default() when netID is empty (backward compat)
- Fan-out: Manager.Subscribe() receives tagged events from all networks
- Network IDs are the first 8 hex chars of SHA-256("yaw2-net:"+name) — stable and short

internal/ipc:
- Run(mgr, port) replaces Run(m *mesh.Mesh, port, anchorURL, joinFn)
- Commands without network_id route to mgr.Default() (backward compat)
- state_snapshot includes Networks array; local_peer/connected_peers still populated from first network
- generate_invite, get_file_list, send_message all respect network_id routing

cmd/daemon:
- Creates netmgr.Manager instead of mesh.Mesh directly
- --join and --share-dir pass through Config
- Auto-join via mgr.Join() before IPC starts

test-network.sh:
- Fix peer_name bash bug: $() with && inside triggers set -e; use if/elif/else instead

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Fredrik Johansson
2026-06-21 19:13:54 +02:00
parent 8d3ca9d331
commit 13fb7ba1fe
6 changed files with 437 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files

205
internal/ipc/ipc.go
View File

@@ -1,6 +1,8 @@
// 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 (
@@ -12,22 +14,16 @@ import (
"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/netmgr"
"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 {
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 {
@@ -35,62 +31,26 @@ func Run(m *mesh.Mesh, port int, anchorURL string, join JoinFunc) error {
}
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)
go handleClient(conn, mgr)
}
}
func handleClient(conn net.Conn, m *mesh.Mesh, anchorURL string, currentNetwork func() string, doJoin func(string), doLeave func()) {
func handleClient(conn net.Conn, mgr *netmgr.Manager) {
defer conn.Close()
events := m.Subscribe()
defer m.Unsubscribe(events)
events := mgr.Subscribe()
defer mgr.Unsubscribe(events)
writeCh := make(chan []byte, 128)
done := make(chan struct{})
// Writer goroutine — sole owner of the write side of the connection.
// Writer goroutine.
go func() {
w := bufio.NewWriter(conn)
for line := range writeCh {
@@ -102,9 +62,7 @@ func handleClient(conn net.Conn, m *mesh.Mesh, anchorURL string, currentNetwork
}
}()
// 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).
// Event pusher — forwards Manager events to the UI client.
go func() {
defer func() { recover() }() //nolint:errcheck
for {
@@ -140,12 +98,8 @@ func handleClient(conn net.Conn, m *mesh.Mesh, anchorURL string, currentNetwork
}
}
send(proto.IpcMessage{
Type: proto.EvtStateSnapshot,
LocalPeer: ptr(m.Identity.PeerInfo()),
ConnectedPeers: m.ConnectedPeers(),
Rooms: []string{"general"},
})
// Send initial state snapshot.
send(stateSnapshot(mgr))
scanner := bufio.NewScanner(conn)
for scanner.Scan() {
@@ -157,11 +111,39 @@ func handleClient(conn net.Conn, m *mesh.Mesh, anchorURL string, currentNetwork
switch cmd.Type {
case proto.CmdJoinNetwork:
if cmd.NetworkName == "" {
send(errMsg("join_network: network_name is required"))
continue
}
netID, err := mgr.Join(cmd.NetworkName)
if err != nil {
send(errMsg(fmt.Sprintf("join_network: %v", err)))
continue
}
// network_joined event is emitted by Manager.Join; nothing extra needed.
_ = 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
}
msg := &proto.ChatMessage{
Mid: randomHex(16),
ID: uuid.NewString(),
From: m.Identity.PeerID(),
From: n.Identity.PeerID(),
To: cmd.To,
Room: cmd.Room,
Body: cmd.Body,
@@ -172,68 +154,63 @@ func handleClient(conn net.Conn, m *mesh.Mesh, anchorURL string, currentNetwork
continue
}
if cmd.To != nil {
// DM — send only to the named recipient.
m.SendTo(*cmd.To, payload)
n.Mesh.SendTo(*cmd.To, payload)
} else {
m.Broadcast(payload)
n.Mesh.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"},
n.Mesh.SaveMessage(msg)
n.Mesh.Emit(proto.IpcMessage{
Type: proto.EvtMessageReceived,
NetworkID: n.ID,
Message: msg,
})
case proto.CmdGetState:
send(stateSnapshot(mgr))
case proto.CmdGetFileList:
if cmd.PeerID == nil || *cmd.PeerID == m.Identity.PeerID() {
// Local list — scan immediately.
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,
PeerID: ptr(m.Identity.PeerID()),
Files: m.ScanShareDir(),
Type: proto.EvtFileList,
NetworkID: n.ID,
PeerID: ptr(n.Identity.PeerID()),
Files: n.Mesh.ScanShareDir(),
})
} else {
// Remote list — send a request over the DataChannel.
// The response arrives as EvtFileList via the mesh event bus.
req, err := json.Marshal(proto.PeerMessage{Type: proto.MsgFileListReq})
if err != nil {
continue
}
if !m.SendTo(*cmd.PeerID, req) {
if !n.Mesh.SendTo(*cmd.PeerID, req) {
send(errMsg(fmt.Sprintf("get_file_list: peer %s not connected", (*cmd.PeerID).Short())))
}
}
case proto.CmdGenerateInvite:
net := currentNetwork()
if net == "" {
n := mgr.Resolve(cmd.NetworkID)
if n == nil {
send(errMsg("generate_invite: not currently joined to a network"))
continue
}
if anchorURL == "" {
if mgr.AnchorURL() == "" {
send(errMsg("generate_invite: daemon was started without -anchor flag"))
continue
}
inv, err := invite.Encode(anchorURL, net)
inv, err := invite.Encode(mgr.AnchorURL(), n.Name)
if err != nil {
send(errMsg(fmt.Sprintf("generate_invite: %v", err)))
continue
}
send(proto.IpcMessage{Type: proto.EvtInviteGenerated, InviteString: inv})
send(proto.IpcMessage{
Type: proto.EvtInviteGenerated,
NetworkID: n.ID,
InviteString: inv,
})
case proto.CmdSendFile:
send(errMsg("file transfer not yet implemented"))
@@ -248,6 +225,37 @@ func handleClient(conn net.Conn, m *mesh.Mesh, anchorURL string, currentNetwork
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()
msg := proto.IpcMessage{
Type: proto.EvtStateSnapshot,
Rooms: []string{"general"},
}
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,
})
}
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()
}
return msg
}
func errMsg(s string) proto.IpcMessage {
return proto.IpcMessage{Type: proto.EvtError, ErrorMessage: s}
}
@@ -259,3 +267,12 @@ func randomHex(n int) string {
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)
}
}