waste-go/internal/mesh/mesh.go

// Package mesh manages the set of live peer connections and broadcasts events.
package mesh

import (
	"log"
	"sync"

	"github.com/waste-go/internal/crypto"
	"github.com/waste-go/internal/proto"
	"github.com/waste-go/internal/store"
)

// PeerConn is a live connection to one peer.
type PeerConn struct {
	Info proto.PeerInfo
	// Send a line of JSON to this peer (pre-encrypted by the sender goroutine).
	Send chan<- []byte
}

// Mesh is the shared state of the local node.
// All methods are safe to call from multiple goroutines.
type Mesh struct {
	Identity *crypto.Identity
	Store    *store.Store // may be nil if persistence is disabled

	mu    sync.RWMutex
	peers map[proto.PeerID]*PeerConn

	// subscribers receive a copy of every event (fan-out to IPC clients)
	subMu sync.Mutex
	subs  []chan proto.IpcMessage
}

// New creates an empty mesh with the given identity.
// Pass a non-nil store to enable message and peer persistence.
func New(id *crypto.Identity, st *store.Store) *Mesh {
	return &Mesh{
		Identity: id,
		Store:    st,
		peers:    make(map[proto.PeerID]*PeerConn),
	}
}

// ── Peer management ───────────────────────────────────────────────────────────

// AddPeer registers a connected peer and notifies subscribers.
func (m *Mesh) AddPeer(conn *PeerConn) {
	m.mu.Lock()
	m.peers[conn.Info.ID] = conn
	m.mu.Unlock()

	if m.Store != nil && conn.Info.Alias != "" {
		if err := m.Store.SavePeer(conn.Info.ID, conn.Info.Alias); err != nil {
			log.Printf("mesh: store peer %s: %v", conn.Info.ID.Short(), err)
		}
	}

	m.emit(proto.IpcMessage{
		Type: proto.EvtPeerConnected,
		Peer: &conn.Info,
	})
}

// SaveMessage persists a chat message if a store is configured.
// Duplicate mids are silently dropped.
func (m *Mesh) SaveMessage(msg *proto.ChatMessage) {
	if m.Store == nil || msg == nil {
		return
	}
	if err := m.Store.SaveMessage(msg); err != nil {
		log.Printf("mesh: store message %s: %v", msg.Mid, err)
	}
}

// UpdatePeerAlias updates the cached alias for a peer after hello verification.
func (m *Mesh) UpdatePeerAlias(id proto.PeerID, alias string) {
	if m.Store != nil && alias != "" {
		if err := m.Store.SavePeer(id, alias); err != nil {
			log.Printf("mesh: update peer alias %s: %v", id.Short(), err)
		}
	}
}

// RemovePeer unregisters a peer and notifies subscribers.
func (m *Mesh) RemovePeer(id proto.PeerID) {
	m.mu.Lock()
	delete(m.peers, id)
	m.mu.Unlock()

	m.emit(proto.IpcMessage{
		Type:   proto.EvtPeerDisconnected,
		PeerID: &id,
	})
}

// ConnectedPeers returns a snapshot of current peer infos.
func (m *Mesh) ConnectedPeers() []proto.PeerInfo {
	m.mu.RLock()
	defer m.mu.RUnlock()
	out := make([]proto.PeerInfo, 0, len(m.peers))
	for _, c := range m.peers {
		out = append(out, c.Info)
	}
	return out
}

// SendTo delivers a raw JSON payload to a specific peer.
// Returns false if the peer isn't connected.
func (m *Mesh) SendTo(id proto.PeerID, payload []byte) bool {
	m.mu.RLock()
	conn, ok := m.peers[id]
	m.mu.RUnlock()
	if !ok {
		return false
	}
	select {
	case conn.Send <- payload:
		return true
	default:
		return false // channel full — peer is slow
	}
}

// Broadcast delivers a raw JSON payload to every connected peer.
func (m *Mesh) Broadcast(payload []byte) {
	m.mu.RLock()
	defer m.mu.RUnlock()
	for _, conn := range m.peers {
		select {
		case conn.Send <- payload:
		default:
		}
	}
}

// ── Event fan-out ─────────────────────────────────────────────────────────────

// Subscribe returns a channel that receives every IPC event.
// The caller must drain it; a full channel is silently dropped.
func (m *Mesh) Subscribe() <-chan proto.IpcMessage {
	ch := make(chan proto.IpcMessage, 64)
	m.subMu.Lock()
	m.subs = append(m.subs, ch)
	m.subMu.Unlock()
	return ch
}

// Unsubscribe removes and closes a subscription channel.
func (m *Mesh) Unsubscribe(ch <-chan proto.IpcMessage) {
	m.subMu.Lock()
	defer m.subMu.Unlock()
	for i, s := range m.subs {
		if s == ch {
			m.subs = append(m.subs[:i], m.subs[i+1:]...)
			close(s)
			return
		}
	}
}

// Emit sends an event to all IPC subscribers (exported for ipc/nat packages).
func (m *Mesh) Emit(msg proto.IpcMessage) {
	m.emit(msg)
}

func (m *Mesh) emit(msg proto.IpcMessage) {
	m.subMu.Lock()
	defer m.subMu.Unlock()
	for _, ch := range m.subs {
		select {
		case ch <- msg:
		default:
		}
	}
}