// 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/hmac" "crypto/sha1" "crypto/sha256" "encoding/base64" "encoding/hex" "fmt" "log" "os" "path/filepath" "strconv" "sync" "time" "github.com/pion/webrtc/v3" "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/shares" "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 DownloadDir string // base directory for received files; defaults to StoreDir if empty AnchorURL string // WebSocket anchor URL used for all networks ShareDir string // default share directory; overridden per network via Join or SetShareDir TurnURL string // optional TURN server URL, e.g. "turn:your-vps:3478" TurnSecret string // shared secret for coturn use-auth-secret HMAC credential } // 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 } // ShareDir returns the share directory for this network (from the mesh). func (n *Network) ShareDir() string { return n.Mesh.ShareDir } // 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 Shares *shares.Store // persistent multi-share configuration } // New creates a Manager from the given config. // Loads shares.json from StoreDir if it exists. func New(cfg Config) *Manager { sh, err := shares.Load(cfg.StoreDir) if err != nil { log.Printf("netmgr: loading shares.json: %v (starting empty)", err) sh, _ = shares.Load("") // fallback to empty } return &Manager{ networks: make(map[string]*Network), cfg: cfg, Shares: sh, } } // Join creates or rejoins a named network. Returns the network ID. // If the network is already joined the existing ID is returned immediately. // shareDir overrides the global default for this network; pass "" to use the default. func (mgr *Manager) Join(name, shareDir 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) // Per-network share dir takes precedence over the global default. if shareDir != "" { m.ShareDir = shareDir } else if mgr.cfg.ShareDir != "" { m.ShareDir = mgr.cfg.ShareDir } m.DownloadDir = filepath.Join(mgr.downloadBase(), "downloads-"+netID_full) capturedNetID := netID m.ScanFiles = func() []proto.FileEntry { return mgr.ScanAllShares(capturedNetID) } if ice := mgr.turnICEServers(); ice != nil { m.ICEServers = ice } // 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()), ShareDir: m.ShareDir, }) return netID, nil } // JoinByHash joins a network using its pre-computed full 64-char hex hash // (yaw2 `net` field) instead of the plaintext name. The network is stored // with an empty name; the network_id (first 8 bytes) is used for display. // This enables joining networks whose names are unknown — e.g. from a yaw2 // invite URL that only contains the hash. func (mgr *Manager) JoinByHash(netHash64, shareDir string) (string, error) { if len(netHash64) != 64 { return "", fmt.Errorf("netHash must be 64 hex chars, got %d", len(netHash64)) } netID := netHash64[:16] // first 8 bytes = 16 hex chars mgr.mu.Lock() if _, exists := mgr.networks[netID]; exists { mgr.mu.Unlock() return netID, nil } mgr.mu.Unlock() derived, err := crypto.DeriveForNetwork(mgr.cfg.MasterIdentity, netHash64) if err != nil { return "", fmt.Errorf("derive identity for net %s: %w", netID, err) } dbPath := filepath.Join(mgr.cfg.StoreDir, "messages-"+netID+".db") st, err := store.Open(dbPath) if err != nil { return "", fmt.Errorf("open store for net %s: %w", netID, err) } m := mesh.New(derived, st) if shareDir != "" { m.ShareDir = shareDir } else if mgr.cfg.ShareDir != "" { m.ShareDir = mgr.cfg.ShareDir } m.DownloadDir = filepath.Join(mgr.downloadBase(), "downloads-"+netID) capturedNetID2 := netID m.ScanFiles = func() []proto.FileEntry { return mgr.ScanAllShares(capturedNetID2) } if ice := mgr.turnICEServers(); ice != nil { m.ICEServers = ice } 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: netID, // display as short hash when name is unknown Hash: netHash64, 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 by hash id=%s peer=%s", netID, derived.PeerID().Short()) if mgr.cfg.AnchorURL != "" { go func() { anchor.RunByHash(ctx, mgr.cfg.AnchorURL, netHash64, derived, m) log.Printf("netmgr: left network %s", netID) }() } mgr.emit(proto.IpcMessage{ Type: proto.EvtNetworkJoined, NetworkID: netID, NetworkName: netID, LocalPeer: peerPtr(derived.PeerInfo()), ShareDir: m.ShareDir, }) 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}) } // SetShareDir updates the share directory for an already-joined network. // Changes take effect immediately for subsequent file-list requests and offers. func (mgr *Manager) SetShareDir(netID, path string) bool { mgr.mu.RLock() net, ok := mgr.networks[netID] mgr.mu.RUnlock() if !ok { return false } net.Mesh.ShareDir = path log.Printf("netmgr: share dir for %q set to %q", net.Name, path) return true } // downloadBase returns the configured download base directory, falling back to StoreDir. func (mgr *Manager) downloadBase() string { if mgr.cfg.DownloadDir != "" { return mgr.cfg.DownloadDir } return mgr.cfg.StoreDir } // SetDownloadDir updates the download directory for an already-joined network. // Changes take effect for the next incoming file transfer on that network. func (mgr *Manager) SetDownloadDir(netID, path string) bool { mgr.mu.RLock() net, ok := mgr.networks[netID] mgr.mu.RUnlock() if !ok { return false } net.Mesh.DownloadDir = path log.Printf("netmgr: download dir for %q set to %q", net.Name, path) return true } // 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 } // ScanAllShares returns file entries from all share roots visible to networkID, // combining the network's legacy ShareDir with entries from shares.json. func (mgr *Manager) ScanAllShares(netID string) []proto.FileEntry { var files []proto.FileEntry // Legacy single-dir share from the network mesh. if n := mgr.Resolve(netID); n != nil { files = append(files, n.Mesh.ScanShareDir()...) } // Additional shares from shares.json. if mgr.Shares != nil { for _, sh := range mgr.Shares.ForNetwork(netID) { files = append(files, scanDir(sh.Path)...) } } return files } // scanDir recursively walks a directory and returns FileEntry for each file. func scanDir(root string) []proto.FileEntry { var files []proto.FileEntry _ = filepath.WalkDir(root, func(path string, d os.DirEntry, err error) error { if err != nil || d.IsDir() { return nil } info, err := d.Info() if err != nil { return nil } rel, err := filepath.Rel(root, path) if err != nil { rel = d.Name() } files = append(files, proto.FileEntry{ Name: d.Name(), SizeBytes: info.Size(), Path: rel, }) return nil }) return files } // AnchorURL returns the configured anchor URL. func (mgr *Manager) AnchorURL() string { return mgr.cfg.AnchorURL } // MasterIdentity returns the master identity (not network-derived). func (mgr *Manager) MasterIdentity() *crypto.Identity { return mgr.cfg.MasterIdentity } // 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: } } } // turnICEServers returns TURN ICE servers if TurnURL and TurnSecret are set, // using coturn's use-auth-secret HMAC-SHA1 time-limited credential scheme. // Returns nil if TURN is not configured. func (mgr *Manager) turnICEServers() []webrtc.ICEServer { if mgr.cfg.TurnURL == "" || mgr.cfg.TurnSecret == "" { return nil } // Username = Unix timestamp 1 hour from now. expiry := strconv.FormatInt(time.Now().Add(time.Hour).Unix(), 10) mac := hmac.New(sha1.New, []byte(mgr.cfg.TurnSecret)) mac.Write([]byte(expiry)) credential := base64.StdEncoding.EncodeToString(mac.Sum(nil)) return []webrtc.ICEServer{{ URLs: []string{mgr.cfg.TurnURL}, Username: expiry, Credential: credential, CredentialType: webrtc.ICECredentialTypePassword, }} } // ── 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 }