2 Commits

Author SHA1 Message Date
Fredrik Johansson
13fb7ba1fe Multi-network foundation: netmgr, derived identities, additive IPC protocol
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>
2026-06-21 19:13:54 +02:00
Fredrik Johansson
8d3ca9d331 Add file listing: share-dir flag, file_list_req/resp DataChannel messages
- proto: FileEntry, FileListResp types; MsgFileListReq/Resp msg types;
  CmdGetFileList + EvtFileList IPC types; Files field on IpcMessage
- mesh: ShareDir field + ScanShareDir(); on DataChannel open, auto-send
  MsgFileListReq to new peer; handle MsgFileListReq (scan + reply) and
  MsgFileListResp (emit EvtFileList to IPC subscribers)
- ipc: get_file_list command — own list returned immediately; remote peer
  list requested via DataChannel (response arrives as EvtFileList event)
- daemon: -share-dir flag wired to mesh.ShareDir
- test scripts: pass -share-dir /home/frejoh/Downloads/{alice,bob,charlie};
  test-network.sh verifies each peer's own file list via get_file_list
- FUTURE.md: document per-network share directories and multi-network design

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-21 19:07:11 +02:00
11 changed files with 584 additions and 148 deletions

7
.claude/settings.json Normal file
View File

@@ -0,0 +1,7 @@
{
"permissions": {
"allow": [
"Bash(go build *)"
]
}
}

View File

@@ -59,6 +59,15 @@ No DHT needed at small group scale (1050 nodes). Keep it simple:
- Public key = stable identity, not a mutable nickname - Public key = stable identity, not a mutable nickname
- No phone number, no central registry — closer to Signal's model than WASTE's original unregistered aliases - No phone number, no central registry — closer to Signal's model than WASTE's original unregistered aliases
### Per-Network Share Directories
The current `-share-dir` flag is a single global directory. Eventually, each network should have its own independent share set:
- Alice shares `/home/alice/Downloads/work-files` on the "work" network
- Alice shares `/home/alice/Music` on the "friends" network
- Peers on "work" never see Alice's music, and vice versa
When multi-network support lands (see below), the `ShareDir` field on `networkCtx` replaces the current global `Mesh.ShareDir`. The IPC `get_file_list` and daemon `-share-dir` flag should move to be per-network configuration — either via a config file or via an IPC command `set_share_dir` scoped to a `network_id`.
### Multi-Network Support ### Multi-Network Support
A single client should be able to participate in multiple networks simultaneously (e.g. "work" and "friends") without leaking that both identities belong to the same person. A single client should be able to participate in multiple networks simultaneously (e.g. "work" and "friends") without leaking that both identities belong to the same person.

View File

@@ -3,29 +3,33 @@
package main package main
import ( import (
"context"
"flag" "flag"
"log" "log"
"os" "os"
"path/filepath"
"github.com/waste-go/internal/anchor"
"github.com/waste-go/internal/crypto" "github.com/waste-go/internal/crypto"
"github.com/waste-go/internal/invite"
"github.com/waste-go/internal/ipc" "github.com/waste-go/internal/ipc"
"github.com/waste-go/internal/mesh" "github.com/waste-go/internal/invite"
"github.com/waste-go/internal/proto" "github.com/waste-go/internal/netmgr"
"github.com/waste-go/internal/store"
) )
func main() { func main() {
dataDir := flag.String("data-dir", "~/.waste", "path to identity/config directory") dataDir := flag.String("data-dir", "~/.waste", "path to identity/config directory")
alias := flag.String("alias", "anon", "display name shown to peers (advisory only)") alias := flag.String("alias", "anon", "display name shown to peers (advisory only)")
ipcPort := flag.Int("ipc-port", 17337, "port for local IPC (UI connects here)") ipcPort := flag.Int("ipc-port", 17337, "port for local IPC (UI connects here)")
anchorURL := flag.String("anchor", "", "anchor WebSocket URL, e.g. ws://your-vps:17339/ws") anchorURL := flag.String("anchor", "", "anchor WebSocket URL, e.g. ws://your-vps:17339/ws")
shareDir := flag.String("share-dir", "", "directory to share with peers on the network")
joinInvite := flag.String("join", "", "waste: invite string — sets anchor URL and auto-joins the network on startup") joinInvite := flag.String("join", "", "waste: invite string — sets anchor URL and auto-joins the network on startup")
flag.Parse() flag.Parse()
dir := expandHome(*dataDir)
id, err := crypto.LoadOrCreate(dir, *alias)
if err != nil {
log.Fatalf("identity: %v", err)
}
log.Printf("daemon: local peer id: %s alias: %s", id.PeerID().Short(), id.Alias)
// --join overrides/sets the anchor URL and triggers an auto-join. // --join overrides/sets the anchor URL and triggers an auto-join.
var autoJoinNetwork string var autoJoinNetwork string
if *joinInvite != "" { if *joinInvite != "" {
@@ -38,41 +42,20 @@ func main() {
log.Printf("daemon: invite decoded — anchor=%s network=%s", inv.Anchor, inv.Network) log.Printf("daemon: invite decoded — anchor=%s network=%s", inv.Anchor, inv.Network)
} }
dir := expandHome(*dataDir) mgr := netmgr.New(netmgr.Config{
id, err := crypto.LoadOrCreate(dir, *alias) MasterIdentity: id,
if err != nil { StoreDir: dir,
log.Fatalf("identity: %v", err) AnchorURL: *anchorURL,
} ShareDir: expandHome(*shareDir),
log.Printf("daemon: local peer id: %s alias: %s", id.PeerID().Short(), id.Alias) })
st, err := store.Open(filepath.Join(dir, "messages.db"))
if err != nil {
log.Fatalf("store: %v", err)
}
defer st.Close()
m := mesh.New(id, st)
// joinFn is passed to the IPC layer; it's called when the UI sends join_network.
joinFn := func(ctx context.Context, networkName string) {
if *anchorURL == "" {
log.Printf("daemon: join_network: no -anchor flag set")
m.Emit(proto.IpcMessage{Type: proto.EvtError, ErrorMessage: "no anchor configured — start daemon with -anchor <url>"})
return
}
log.Printf("daemon: joining network %q via %s", networkName, *anchorURL)
anchor.Run(ctx, *anchorURL, networkName, id, m)
log.Printf("daemon: left network %q", networkName)
}
// Auto-join from --join flag before starting IPC (non-blocking).
if autoJoinNetwork != "" { if autoJoinNetwork != "" {
ctx, cancel := context.WithCancel(context.Background()) if _, err := mgr.Join(autoJoinNetwork); err != nil {
_ = cancel // lifecycle managed by the joinFn / ipc.Run leave log.Fatalf("auto-join: %v", err)
go joinFn(ctx, autoJoinNetwork) }
} }
if err := ipc.Run(m, *ipcPort, *anchorURL, joinFn); err != nil { if err := ipc.Run(mgr, *ipcPort); err != nil {
log.Fatalf("ipc: %v", err) log.Fatalf("ipc: %v", err)
} }
} }
@@ -85,4 +68,3 @@ func expandHome(path string) string {
} }
return path return path
} }

View File

@@ -15,6 +15,7 @@ import (
"encoding/json" "encoding/json"
"errors" "errors"
"fmt" "fmt"
"io"
"os" "os"
"path/filepath" "path/filepath"
"time" "time"
@@ -22,6 +23,7 @@ import (
"filippo.io/edwards25519" "filippo.io/edwards25519"
"golang.org/x/crypto/chacha20poly1305" "golang.org/x/crypto/chacha20poly1305"
"golang.org/x/crypto/curve25519" "golang.org/x/crypto/curve25519"
"golang.org/x/crypto/hkdf"
"golang.org/x/crypto/nacl/box" "golang.org/x/crypto/nacl/box"
"github.com/waste-go/internal/proto" "github.com/waste-go/internal/proto"
@@ -179,6 +181,24 @@ func SignalingOpen(b64box string, senderPub, recipientPriv *[32]byte) ([]byte, e
return out, nil return out, nil
} }
// DeriveForNetwork returns a new in-memory Identity derived from the master key
// and the given network hash (hex). Same master + same network hash always
// produces the same keypair, so the peer ID is stable across restarts.
// Different networks produce different peer IDs, preventing cross-network correlation.
func DeriveForNetwork(master *Identity, networkHash string) (*Identity, error) {
r := hkdf.New(sha256.New, master.privateKey[:32], []byte(networkHash), []byte("yaw2-net-identity"))
var seed [32]byte
if _, err := io.ReadFull(r, seed[:]); err != nil {
return nil, fmt.Errorf("deriving network identity: %w", err)
}
priv := ed25519.NewKeyFromSeed(seed[:])
return &Identity{
privateKey: priv,
PublicKey: priv.Public().(ed25519.PublicKey),
Alias: master.Alias,
}, nil
}
// ── X25519 ECDH ─────────────────────────────────────────────────────────────── // ── X25519 ECDH ───────────────────────────────────────────────────────────────
// EphemeralKey is an X25519 keypair used for a single session. // EphemeralKey is an X25519 keypair used for a single session.

View File

@@ -1,6 +1,8 @@
// Package ipc implements the local IPC server. // Package ipc implements the local IPC server.
// The UI (or any local tool) connects to 127.0.0.1:17337 and speaks // The UI (or any local tool) connects to 127.0.0.1:17337 and speaks
// newline-delimited JSON: send IpcMessage commands, receive IpcMessage events. // newline-delimited JSON: send IpcMessage commands, receive IpcMessage events.
//
// Backward compat: commands without network_id route to the first joined network.
package ipc package ipc
import ( import (
@@ -12,22 +14,16 @@ import (
"fmt" "fmt"
"log" "log"
"net" "net"
"sync"
"time" "time"
"github.com/google/uuid" "github.com/google/uuid"
"github.com/waste-go/internal/invite" "github.com/waste-go/internal/invite"
"github.com/waste-go/internal/mesh" "github.com/waste-go/internal/netmgr"
"github.com/waste-go/internal/proto" "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. // Run starts the IPC listener. Blocks until the listener fails.
// anchorURL is the configured anchor WebSocket URL (used for invite generation). func Run(mgr *netmgr.Manager, port int) error {
// 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) addr := fmt.Sprintf("127.0.0.1:%d", port)
ln, err := net.Listen("tcp", addr) ln, err := net.Listen("tcp", addr)
if err != nil { 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) 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 { for {
conn, err := ln.Accept() conn, err := ln.Accept()
if err != nil { if err != nil {
return fmt.Errorf("ipc accept: %w", err) return fmt.Errorf("ipc accept: %w", err)
} }
log.Printf("ipc: UI client connected from %s", conn.RemoteAddr()) 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() defer conn.Close()
events := m.Subscribe() events := mgr.Subscribe()
defer m.Unsubscribe(events) defer mgr.Unsubscribe(events)
writeCh := make(chan []byte, 128) writeCh := make(chan []byte, 128)
done := make(chan struct{}) done := make(chan struct{})
// Writer goroutine — sole owner of the write side of the connection. // Writer goroutine.
go func() { go func() {
w := bufio.NewWriter(conn) w := bufio.NewWriter(conn)
for line := range writeCh { 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. // Event pusher — forwards Manager 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() { go func() {
defer func() { recover() }() //nolint:errcheck defer func() { recover() }() //nolint:errcheck
for { for {
@@ -140,12 +98,8 @@ func handleClient(conn net.Conn, m *mesh.Mesh, anchorURL string, currentNetwork
} }
} }
send(proto.IpcMessage{ // Send initial state snapshot.
Type: proto.EvtStateSnapshot, send(stateSnapshot(mgr))
LocalPeer: ptr(m.Identity.PeerInfo()),
ConnectedPeers: m.ConnectedPeers(),
Rooms: []string{"general"},
})
scanner := bufio.NewScanner(conn) scanner := bufio.NewScanner(conn)
for scanner.Scan() { for scanner.Scan() {
@@ -157,11 +111,39 @@ func handleClient(conn net.Conn, m *mesh.Mesh, anchorURL string, currentNetwork
switch cmd.Type { 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: case proto.CmdSendMessage:
n := mgr.Resolve(cmd.NetworkID)
if n == nil {
send(errMsg("send_message: not joined to any network"))
continue
}
msg := &proto.ChatMessage{ msg := &proto.ChatMessage{
Mid: randomHex(16), Mid: randomHex(16),
ID: uuid.NewString(), ID: uuid.NewString(),
From: m.Identity.PeerID(), From: n.Identity.PeerID(),
To: cmd.To, To: cmd.To,
Room: cmd.Room, Room: cmd.Room,
Body: cmd.Body, Body: cmd.Body,
@@ -172,48 +154,63 @@ func handleClient(conn net.Conn, m *mesh.Mesh, anchorURL string, currentNetwork
continue continue
} }
if cmd.To != nil { if cmd.To != nil {
// DM — send only to the named recipient. n.Mesh.SendTo(*cmd.To, payload)
m.SendTo(*cmd.To, payload)
} else { } else {
m.Broadcast(payload) n.Mesh.Broadcast(payload)
} }
m.SaveMessage(msg) n.Mesh.SaveMessage(msg)
m.Emit(proto.IpcMessage{Type: proto.EvtMessageReceived, Message: msg}) n.Mesh.Emit(proto.IpcMessage{
Type: proto.EvtMessageReceived,
case proto.CmdJoinNetwork: NetworkID: n.ID,
if cmd.NetworkName == "" { Message: msg,
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.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: n.Mesh.ScanShareDir(),
})
} 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.CmdGenerateInvite: case proto.CmdGenerateInvite:
net := currentNetwork() n := mgr.Resolve(cmd.NetworkID)
if net == "" { if n == nil {
send(errMsg("generate_invite: not currently joined to a network")) send(errMsg("generate_invite: not currently joined to a network"))
continue continue
} }
if anchorURL == "" { if mgr.AnchorURL() == "" {
send(errMsg("generate_invite: daemon was started without -anchor flag")) send(errMsg("generate_invite: daemon was started without -anchor flag"))
continue continue
} }
inv, err := invite.Encode(anchorURL, net) inv, err := invite.Encode(mgr.AnchorURL(), n.Name)
if err != nil { if err != nil {
send(errMsg(fmt.Sprintf("generate_invite: %v", err))) send(errMsg(fmt.Sprintf("generate_invite: %v", err)))
continue continue
} }
send(proto.IpcMessage{Type: proto.EvtInviteGenerated, InviteString: inv}) send(proto.IpcMessage{
Type: proto.EvtInviteGenerated,
NetworkID: n.ID,
InviteString: inv,
})
case proto.CmdSendFile: case proto.CmdSendFile:
send(errMsg("file transfer not yet implemented")) send(errMsg("file transfer not yet implemented"))
@@ -228,6 +225,37 @@ func handleClient(conn net.Conn, m *mesh.Mesh, anchorURL string, currentNetwork
log.Printf("ipc: UI client disconnected") 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 { func errMsg(s string) proto.IpcMessage {
return proto.IpcMessage{Type: proto.EvtError, ErrorMessage: s} return proto.IpcMessage{Type: proto.EvtError, ErrorMessage: s}
} }
@@ -239,3 +267,12 @@ func randomHex(n int) string {
rand.Read(b) //nolint:errcheck rand.Read(b) //nolint:errcheck
return hex.EncodeToString(b) 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)
}
}

View File

@@ -3,6 +3,7 @@ package mesh
import ( import (
"log" "log"
"os"
"sync" "sync"
"github.com/waste-go/internal/crypto" "github.com/waste-go/internal/crypto"
@@ -22,6 +23,7 @@ type PeerConn struct {
type Mesh struct { type Mesh struct {
Identity *crypto.Identity Identity *crypto.Identity
Store *store.Store // may be nil if persistence is disabled Store *store.Store // may be nil if persistence is disabled
ShareDir string // directory whose contents are shared with peers; "" = no sharing
mu sync.RWMutex mu sync.RWMutex
peers map[proto.PeerID]*PeerConn peers map[proto.PeerID]*PeerConn
@@ -41,6 +43,31 @@ func New(id *crypto.Identity, st *store.Store) *Mesh {
} }
} }
// ScanShareDir returns the list of files in the local share directory.
// Returns an empty slice if ShareDir is unset or the directory is empty.
func (m *Mesh) ScanShareDir() []proto.FileEntry {
if m.ShareDir == "" {
return nil
}
entries, err := os.ReadDir(m.ShareDir)
if err != nil {
log.Printf("mesh: scan share dir %s: %v", m.ShareDir, err)
return nil
}
var files []proto.FileEntry
for _, e := range entries {
if e.IsDir() {
continue
}
info, err := e.Info()
if err != nil {
continue
}
files = append(files, proto.FileEntry{Name: e.Name(), SizeBytes: info.Size()})
}
return files
}
// ── Peer management ─────────────────────────────────────────────────────────── // ── Peer management ───────────────────────────────────────────────────────────
// AddPeer registers a connected peer and notifies subscribers. // AddPeer registers a connected peer and notifies subscribers.

View File

@@ -56,6 +56,11 @@ func WireDataChannel(
} }
m.AddPeer(peerConn) m.AddPeer(peerConn)
// Request the peer's file list immediately after connect.
if req, err := json.Marshal(proto.PeerMessage{Type: proto.MsgFileListReq}); err == nil {
sendCh <- req
}
go func() { go func() {
for payload := range sendCh { for payload := range sendCh {
if err := dc.SendText(string(payload)); err != nil { if err := dc.SendText(string(payload)); err != nil {
@@ -147,6 +152,28 @@ func handleDCMessage(data []byte, from proto.PeerID, localID *crypto.Identity, m
func dispatchPeerMessage(msg proto.PeerMessage, from proto.PeerID, m *Mesh) { func dispatchPeerMessage(msg proto.PeerMessage, from proto.PeerID, m *Mesh) {
switch msg.Type { switch msg.Type {
case proto.MsgFileListReq:
// Peer wants our file list — reply directly on their send channel.
files := m.ScanShareDir()
resp, err := json.Marshal(proto.PeerMessage{
Type: proto.MsgFileListResp,
FileListResp: &proto.FileListResp{Files: files},
})
if err != nil {
return
}
m.SendTo(from, resp)
case proto.MsgFileListResp:
// Received a remote peer's file list — forward to IPC subscribers.
if msg.FileListResp != nil {
m.Emit(proto.IpcMessage{
Type: proto.EvtFileList,
PeerID: peerIDPtr(from),
Files: msg.FileListResp.Files,
})
}
case proto.MsgChat: case proto.MsgChat:
if msg.Chat != nil { if msg.Chat != nil {
m.SaveMessage(msg.Chat) m.SaveMessage(msg.Chat)

264
internal/netmgr/manager.go Normal file
View File

@@ -0,0 +1,264 @@
// Package netmgr manages multiple concurrent network contexts.
// Each network gets its own derived identity, mesh, store, and anchor connection.
// The Manager fans out events from all networks to IPC subscribers, tagging each
// event with the network_id so clients can route them appropriately.
package netmgr
import (
"context"
"crypto/sha256"
"encoding/hex"
"fmt"
"log"
"path/filepath"
"sync"
"github.com/waste-go/internal/anchor"
"github.com/waste-go/internal/crypto"
"github.com/waste-go/internal/mesh"
"github.com/waste-go/internal/proto"
"github.com/waste-go/internal/store"
)
// Config holds the Manager's startup configuration.
type Config struct {
MasterIdentity *crypto.Identity
StoreDir string // base directory for per-network SQLite files
AnchorURL string // WebSocket anchor URL used for all networks
ShareDir string // shared file directory (global for now; per-network in future)
}
// Network is a single joined network context.
type Network struct {
ID string // first 8 hex chars of networkHash — stable, short, opaque
Name string
Hash string // full SHA-256("yaw2-net:"+name), hex
Identity *crypto.Identity
Mesh *mesh.Mesh
Store *store.Store
cancel context.CancelFunc
}
// Manager owns all joined networks and fans out their events to IPC subscribers.
type Manager struct {
cfg Config
mu sync.RWMutex
networks map[string]*Network // keyed by Network.ID
order []string // insertion order for "default" lookups
subsMu sync.Mutex
subs []chan proto.IpcMessage
}
// New creates a Manager from the given config.
func New(cfg Config) *Manager {
return &Manager{
networks: make(map[string]*Network),
cfg: cfg,
}
}
// Join creates or rejoins a named network. Returns the network ID.
// If the network is already joined the existing ID is returned immediately.
func (mgr *Manager) Join(name string) (string, error) {
netHash := hashNetName(name)
netID := netHash[:8]
mgr.mu.Lock()
if _, exists := mgr.networks[netID]; exists {
mgr.mu.Unlock()
return netID, nil
}
mgr.mu.Unlock()
// Derive a network-specific identity.
netID_full := netID
derived, err := crypto.DeriveForNetwork(mgr.cfg.MasterIdentity, netHash)
if err != nil {
return "", fmt.Errorf("derive identity for %q: %w", name, err)
}
// Open per-network store.
dbPath := filepath.Join(mgr.cfg.StoreDir, "messages-"+netID_full+".db")
st, err := store.Open(dbPath)
if err != nil {
return "", fmt.Errorf("open store for %q: %w", name, err)
}
m := mesh.New(derived, st)
if mgr.cfg.ShareDir != "" {
m.ShareDir = mgr.cfg.ShareDir
}
// Forward all mesh events to the Manager's fan-out, tagging with network_id.
meshEvents := m.Subscribe()
go func() {
for evt := range meshEvents {
evt.NetworkID = netID
mgr.emit(evt)
}
}()
ctx, cancel := context.WithCancel(context.Background())
net := &Network{
ID: netID,
Name: name,
Hash: netHash,
Identity: derived,
Mesh: m,
Store: st,
cancel: cancel,
}
mgr.mu.Lock()
mgr.networks[netID] = net
mgr.order = append(mgr.order, netID)
mgr.mu.Unlock()
log.Printf("netmgr: joining network %q (id=%s peer=%s)", name, netID, derived.PeerID().Short())
if mgr.cfg.AnchorURL != "" {
go func() {
anchor.Run(ctx, mgr.cfg.AnchorURL, name, derived, m)
log.Printf("netmgr: left network %q", name)
}()
}
mgr.emit(proto.IpcMessage{
Type: proto.EvtNetworkJoined,
NetworkID: netID,
NetworkName: name,
LocalPeer: peerPtr(derived.PeerInfo()),
})
return netID, nil
}
// Leave cancels a network context by ID. Closes its store.
func (mgr *Manager) Leave(netID string) {
mgr.mu.Lock()
net, ok := mgr.networks[netID]
if ok {
delete(mgr.networks, netID)
mgr.order = removeStr(mgr.order, netID)
}
mgr.mu.Unlock()
if !ok {
return
}
net.cancel()
net.Store.Close()
net.Mesh.Unsubscribe(net.Mesh.Subscribe()) // drain + close subscription
log.Printf("netmgr: left network %q (id=%s)", net.Name, netID)
mgr.emit(proto.IpcMessage{Type: proto.EvtNetworkLeft, NetworkID: netID})
}
// LeaveAll leaves every joined network.
func (mgr *Manager) LeaveAll() {
mgr.mu.RLock()
ids := make([]string, len(mgr.order))
copy(ids, mgr.order)
mgr.mu.RUnlock()
for _, id := range ids {
mgr.Leave(id)
}
}
// Get returns a network by ID.
func (mgr *Manager) Get(netID string) (*Network, bool) {
mgr.mu.RLock()
defer mgr.mu.RUnlock()
n, ok := mgr.networks[netID]
return n, ok
}
// Default returns the first joined network, or nil if none are joined.
// Used to route commands that carry no network_id (backward compat).
func (mgr *Manager) Default() *Network {
mgr.mu.RLock()
defer mgr.mu.RUnlock()
if len(mgr.order) == 0 {
return nil
}
return mgr.networks[mgr.order[0]]
}
// Resolve returns the network for netID, or the default if netID is empty.
func (mgr *Manager) Resolve(netID string) *Network {
if netID == "" {
return mgr.Default()
}
n, _ := mgr.Get(netID)
return n
}
// All returns a snapshot of all joined networks in join order.
func (mgr *Manager) All() []*Network {
mgr.mu.RLock()
defer mgr.mu.RUnlock()
out := make([]*Network, 0, len(mgr.order))
for _, id := range mgr.order {
out = append(out, mgr.networks[id])
}
return out
}
// AnchorURL returns the configured anchor URL.
func (mgr *Manager) AnchorURL() string { return mgr.cfg.AnchorURL }
// Subscribe returns a channel that receives tagged events from all networks.
func (mgr *Manager) Subscribe() <-chan proto.IpcMessage {
ch := make(chan proto.IpcMessage, 128)
mgr.subsMu.Lock()
mgr.subs = append(mgr.subs, ch)
mgr.subsMu.Unlock()
return ch
}
// Unsubscribe removes and closes a subscription channel.
func (mgr *Manager) Unsubscribe(ch <-chan proto.IpcMessage) {
mgr.subsMu.Lock()
defer mgr.subsMu.Unlock()
for i, s := range mgr.subs {
if s == ch {
mgr.subs = append(mgr.subs[:i], mgr.subs[i+1:]...)
close(s)
return
}
}
}
func (mgr *Manager) emit(msg proto.IpcMessage) {
mgr.subsMu.Lock()
defer mgr.subsMu.Unlock()
for _, ch := range mgr.subs {
select {
case ch <- msg:
default:
}
}
}
// ── helpers ───────────────────────────────────────────────────────────────────
func hashNetName(name string) string {
h := sha256.Sum256([]byte("yaw2-net:" + name))
return hex.EncodeToString(h[:])
}
func removeStr(ss []string, s string) []string {
out := ss[:0]
for _, v := range ss {
if v != s {
out = append(out, v)
}
}
return out
}
func peerPtr(p proto.PeerInfo) *proto.PeerInfo { return &p }

View File

@@ -34,13 +34,15 @@ type PeerInfo struct {
type MsgType string type MsgType string
const ( const (
MsgChat MsgType = "chat" MsgChat MsgType = "chat"
MsgPeerGossip MsgType = "peer_gossip" MsgPeerGossip MsgType = "peer_gossip"
MsgFileOffer MsgType = "file_offer" MsgFileListReq MsgType = "file_list_req"
MsgFileResp MsgType = "file_response" MsgFileListResp MsgType = "file_list_resp"
MsgFileDone MsgType = "file_done" MsgFileOffer MsgType = "file_offer"
MsgPing MsgType = "ping" MsgFileResp MsgType = "file_response"
MsgPong MsgType = "pong" MsgFileDone MsgType = "file_done"
MsgPing MsgType = "ping"
MsgPong MsgType = "pong"
) )
// PeerMessage is the top-level container sent over the "yaw" DataChannel. // PeerMessage is the top-level container sent over the "yaw" DataChannel.
@@ -49,12 +51,13 @@ type PeerMessage struct {
Type MsgType `json:"type"` Type MsgType `json:"type"`
// Only one of these will be set, depending on Type. // Only one of these will be set, depending on Type.
Chat *ChatMessage `json:"chat,omitempty"` Chat *ChatMessage `json:"chat,omitempty"`
Gossip *PeerGossip `json:"gossip,omitempty"` Gossip *PeerGossip `json:"gossip,omitempty"`
FileOffer *FileOffer `json:"file_offer,omitempty"` FileListResp *FileListResp `json:"file_list_resp,omitempty"`
FileResp *FileResponse `json:"file_response,omitempty"` FileOffer *FileOffer `json:"file_offer,omitempty"`
FileDone *FileDone `json:"file_done,omitempty"` FileResp *FileResponse `json:"file_response,omitempty"`
Seq *uint64 `json:"seq,omitempty"` // for ping/pong FileDone *FileDone `json:"file_done,omitempty"`
Seq *uint64 `json:"seq,omitempty"` // for ping/pong
} }
// ChatMessage is a message to a room or a DM. // ChatMessage is a message to a room or a DM.
@@ -80,6 +83,18 @@ type GossipEntry struct {
LastSeen time.Time `json:"last_seen"` LastSeen time.Time `json:"last_seen"`
} }
// FileEntry describes a single file in a peer's shared directory.
type FileEntry struct {
Name string `json:"name"`
SizeBytes int64 `json:"size_bytes"`
}
// FileListResp is the payload for MsgFileListResp.
// MsgFileListReq carries no payload — it is a zero-field request.
type FileListResp struct {
Files []FileEntry `json:"files"`
}
// FileOffer initiates a file transfer. // FileOffer initiates a file transfer.
type FileOffer struct { type FileOffer struct {
Mid string `json:"mid"` // dedup id Mid string `json:"mid"` // dedup id
@@ -187,6 +202,7 @@ const (
CmdGetState IpcMsgType = "get_state" CmdGetState IpcMsgType = "get_state"
CmdSendFile IpcMsgType = "send_file" CmdSendFile IpcMsgType = "send_file"
CmdGenerateInvite IpcMsgType = "generate_invite" CmdGenerateInvite IpcMsgType = "generate_invite"
CmdGetFileList IpcMsgType = "get_file_list"
// Events (daemon → UI) // Events (daemon → UI)
EvtMessageReceived IpcMsgType = "message_received" EvtMessageReceived IpcMsgType = "message_received"
@@ -198,12 +214,26 @@ const (
EvtStateSnapshot IpcMsgType = "state_snapshot" EvtStateSnapshot IpcMsgType = "state_snapshot"
EvtError IpcMsgType = "error" EvtError IpcMsgType = "error"
EvtInviteGenerated IpcMsgType = "invite_generated" EvtInviteGenerated IpcMsgType = "invite_generated"
EvtFileList IpcMsgType = "file_list"
EvtNetworkJoined IpcMsgType = "network_joined"
EvtNetworkLeft IpcMsgType = "network_left"
) )
// NetworkInfo summarises one joined network for state_snapshot and network_joined events.
type NetworkInfo struct {
NetworkID string `json:"network_id"`
NetworkName string `json:"network_name"`
LocalPeer *PeerInfo `json:"local_peer,omitempty"`
}
// IpcMessage covers both commands and events. // IpcMessage covers both commands and events.
type IpcMessage struct { type IpcMessage struct {
Type IpcMsgType `json:"type"` Type IpcMsgType `json:"type"`
// optional: scopes a command/event to a specific network.
// When absent, defaults to the first (or only) joined network.
NetworkID string `json:"network_id,omitempty"`
// send_message // send_message
Room string `json:"room,omitempty"` Room string `json:"room,omitempty"`
To *PeerID `json:"to,omitempty"` To *PeerID `json:"to,omitempty"`
@@ -224,9 +254,13 @@ type IpcMessage struct {
TransferID string `json:"transfer_id,omitempty"` TransferID string `json:"transfer_id,omitempty"`
BytesReceived int64 `json:"bytes_received,omitempty"` BytesReceived int64 `json:"bytes_received,omitempty"`
TotalBytes int64 `json:"total_bytes,omitempty"` TotalBytes int64 `json:"total_bytes,omitempty"`
LocalPeer *PeerInfo `json:"local_peer,omitempty"` // state_snapshot fields (existing shape preserved for backward compat)
ConnectedPeers []PeerInfo `json:"connected_peers,omitempty"` LocalPeer *PeerInfo `json:"local_peer,omitempty"`
Rooms []string `json:"rooms,omitempty"` ConnectedPeers []PeerInfo `json:"connected_peers,omitempty"`
Rooms []string `json:"rooms,omitempty"`
// multi-network: all joined networks (additive)
Networks []NetworkInfo `json:"networks,omitempty"`
ErrorMessage string `json:"error_message,omitempty"` ErrorMessage string `json:"error_message,omitempty"`
InviteString string `json:"invite,omitempty"` InviteString string `json:"invite,omitempty"`
Files []FileEntry `json:"files,omitempty"`
} }

View File

@@ -192,6 +192,7 @@ log "$ALICE_COLOR" "alice" "starting daemon (ipc :${ALICE_IPC})"
-data-dir "$DATA_ROOT/alice" \ -data-dir "$DATA_ROOT/alice" \
-ipc-port "$ALICE_IPC" \ -ipc-port "$ALICE_IPC" \
-anchor "$ANCHOR_URL" \ -anchor "$ANCHOR_URL" \
-share-dir "/home/frejoh/Downloads/alice" \
2> >(while IFS= read -r l; do echo -e "${ALICE_COLOR}${DIM}[alice] ${l}${RESET}"; done) & 2> >(while IFS= read -r l; do echo -e "${ALICE_COLOR}${DIM}[alice] ${l}${RESET}"; done) &
PIDS+=($!) PIDS+=($!)
@@ -201,6 +202,7 @@ log "$BOB_COLOR" "bob" "starting daemon (ipc :${BOB_IPC})"
-data-dir "$DATA_ROOT/bob" \ -data-dir "$DATA_ROOT/bob" \
-ipc-port "$BOB_IPC" \ -ipc-port "$BOB_IPC" \
-anchor "$ANCHOR_URL" \ -anchor "$ANCHOR_URL" \
-share-dir "/home/frejoh/Downloads/bob" \
2> >(while IFS= read -r l; do echo -e "${BOB_COLOR}${DIM}[bob] ${l}${RESET}"; done) & 2> >(while IFS= read -r l; do echo -e "${BOB_COLOR}${DIM}[bob] ${l}${RESET}"; done) &
PIDS+=($!) PIDS+=($!)
@@ -210,6 +212,7 @@ log "$CHARLIE_COLOR" "charlie" "starting daemon (ipc :${CHARLIE_IPC})"
-data-dir "$DATA_ROOT/charlie" \ -data-dir "$DATA_ROOT/charlie" \
-ipc-port "$CHARLIE_IPC" \ -ipc-port "$CHARLIE_IPC" \
-anchor "$ANCHOR_URL" \ -anchor "$ANCHOR_URL" \
-share-dir "/home/frejoh/Downloads/charlie" \
2> >(while IFS= read -r l; do echo -e "${CHARLIE_COLOR}${DIM}[charlie]${l}${RESET}"; done) & 2> >(while IFS= read -r l; do echo -e "${CHARLIE_COLOR}${DIM}[charlie]${l}${RESET}"; done) &
PIDS+=($!) PIDS+=($!)
@@ -309,6 +312,29 @@ ipc "$ALICE_IPC" "$(jq -cn \
'{"type":"send_message","room":$room,"body":$body,"to":$to}')" '{"type":"send_message","room":$room,"body":$body,"to":$to}')"
sleep 0.4 sleep 0.4
# ── file list check ───────────────────────────────────────────────────────────
echo ""
echo -e "${DIM}────────────────────────────────────────────────────────${RESET}"
echo -e "file listing"
echo -e "${DIM}────────────────────────────────────────────────────────${RESET}"
for peer_ipc in "$ALICE_IPC" "$BOB_IPC" "$CHARLIE_IPC"; do
if [ "$peer_ipc" = "$ALICE_IPC" ]; then peer_name="alice"
elif [ "$peer_ipc" = "$BOB_IPC" ]; then peer_name="bob"
else peer_name="charlie"; fi
result=$(echo '{"type":"get_file_list"}' \
| timeout 2 nc 127.0.0.1 "$peer_ipc" 2>/dev/null \
| grep '"type":"file_list"' | head -1)
if [ -n "$result" ]; then
count=$(echo "$result" | jq '.files | length' 2>/dev/null || echo "?")
files=$(echo "$result" | jq -r '.files[].name' 2>/dev/null | tr '\n' ' ')
echo -e " ${BOLD}${peer_name}${RESET}: ${count} file(s) — ${files}"
else
echo -e " ${RED}${peer_name}: no file_list response${RESET}"
fi
done
sleep 0.5
# ── leave ───────────────────────────────────────────────────────────────────── # ── leave ─────────────────────────────────────────────────────────────────────
echo "" echo ""
echo -e "${DIM}────────────────────────────────────────────────────────${RESET}" echo -e "${DIM}────────────────────────────────────────────────────────${RESET}"

View File

@@ -77,15 +77,18 @@ ANCHOR_URL="ws://127.0.0.1:${ANCHOR_PORT}/ws"
# Daemons # Daemons
"$DATA_ROOT/bin/waste-daemon" -alias alice -data-dir "$DATA_ROOT/alice" \ "$DATA_ROOT/bin/waste-daemon" -alias alice -data-dir "$DATA_ROOT/alice" \
-ipc-port "$ALICE_IPC" -anchor "$ANCHOR_URL" 2>/dev/null & -ipc-port "$ALICE_IPC" -anchor "$ANCHOR_URL" \
-share-dir "/home/frejoh/Downloads/alice" 2>/dev/null &
PIDS+=($!) PIDS+=($!)
"$DATA_ROOT/bin/waste-daemon" -alias bob -data-dir "$DATA_ROOT/bob" \ "$DATA_ROOT/bin/waste-daemon" -alias bob -data-dir "$DATA_ROOT/bob" \
-ipc-port "$BOB_IPC" -anchor "$ANCHOR_URL" 2>/dev/null & -ipc-port "$BOB_IPC" -anchor "$ANCHOR_URL" \
-share-dir "/home/frejoh/Downloads/bob" 2>/dev/null &
PIDS+=($!) PIDS+=($!)
"$DATA_ROOT/bin/waste-daemon" -alias charlie -data-dir "$DATA_ROOT/charlie" \ "$DATA_ROOT/bin/waste-daemon" -alias charlie -data-dir "$DATA_ROOT/charlie" \
-ipc-port "$CHARLIE_IPC" -anchor "$ANCHOR_URL" 2>/dev/null & -ipc-port "$CHARLIE_IPC" -anchor "$ANCHOR_URL" \
-share-dir "/home/frejoh/Downloads/charlie" 2>/dev/null &
PIDS+=($!) PIDS+=($!)
wait_port "$ALICE_IPC" wait_port "$ALICE_IPC"