diff --git a/internal/ipc/ipc.go b/internal/ipc/ipc.go index d9fad16..244f791 100644 --- a/internal/ipc/ipc.go +++ b/internal/ipc/ipc.go @@ -12,6 +12,7 @@ import ( "fmt" "log" "net" + "sync" "time" "github.com/google/uuid" @@ -20,10 +21,10 @@ import ( ) // JoinFunc is called when the UI issues a join_network command. -// It should connect to the anchor and block until done or ctx is cancelled. type JoinFunc func(ctx context.Context, networkName string) // Run starts the IPC listener. Blocks until the listener fails. +// Network join/leave state is daemon-scoped (shared across all IPC clients). func Run(m *mesh.Mesh, port int, join JoinFunc) error { addr := fmt.Sprintf("127.0.0.1:%d", port) ln, err := net.Listen("tcp", addr) @@ -32,24 +33,53 @@ func Run(m *mesh.Mesh, port int, join JoinFunc) error { } log.Printf("ipc: listening on %s", addr) + // networkCancel is shared across all clients — any client can join/leave, + // and the join persists even after the commanding client disconnects. + var ( + networkMu sync.Mutex + networkCancel context.CancelFunc + ) + + doJoin := func(name string) { + networkMu.Lock() + if networkCancel != nil { + networkCancel() + } + ctx, cancel := context.WithCancel(context.Background()) + networkCancel = cancel + networkMu.Unlock() + go join(ctx, name) + } + + doLeave := func() { + networkMu.Lock() + if networkCancel != nil { + networkCancel() + networkCancel = nil + } + networkMu.Unlock() + } + 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, join) + go handleClient(conn, m, doJoin, doLeave) } } -func handleClient(conn net.Conn, m *mesh.Mesh, join JoinFunc) { +func handleClient(conn net.Conn, m *mesh.Mesh, doJoin func(string), doLeave func()) { defer conn.Close() events := m.Subscribe() defer m.Unsubscribe(events) writeCh := make(chan []byte, 128) + done := make(chan struct{}) + // Writer goroutine — sole owner of the write side of the connection. go func() { w := bufio.NewWriter(conn) for line := range writeCh { @@ -61,31 +91,51 @@ func handleClient(conn net.Conn, m *mesh.Mesh, join JoinFunc) { } }() + // 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). go func() { - for evt := range events { - line, err := json.Marshal(evt) - if err != nil { - continue - } + defer func() { recover() }() //nolint:errcheck + for { select { - case writeCh <- line: - default: + 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(writeCh, proto.IpcMessage{ + 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(proto.IpcMessage{ Type: proto.EvtStateSnapshot, LocalPeer: ptr(m.Identity.PeerInfo()), ConnectedPeers: m.ConnectedPeers(), Rooms: []string{"general"}, }) - // Track an active network join so we can cancel it on leave_network. - var ( - networkCancel context.CancelFunc - ) - scanner := bufio.NewScanner(conn) for scanner.Scan() { var cmd proto.IpcMessage @@ -115,24 +165,16 @@ func handleClient(conn net.Conn, m *mesh.Mesh, join JoinFunc) { case proto.CmdJoinNetwork: if cmd.NetworkName == "" { - send(writeCh, errMsg("join_network: network_name is required")) + send(errMsg("join_network: network_name is required")) continue } - if networkCancel != nil { - networkCancel() // leave any previous network - } - ctx, cancel := context.WithCancel(context.Background()) - networkCancel = cancel - go join(ctx, cmd.NetworkName) + doJoin(cmd.NetworkName) case proto.CmdLeaveNetwork: - if networkCancel != nil { - networkCancel() - networkCancel = nil - } + doLeave() case proto.CmdGetState: - send(writeCh, proto.IpcMessage{ + send(proto.IpcMessage{ Type: proto.EvtStateSnapshot, LocalPeer: ptr(m.Identity.PeerInfo()), ConnectedPeers: m.ConnectedPeers(), @@ -140,31 +182,18 @@ func handleClient(conn net.Conn, m *mesh.Mesh, join JoinFunc) { }) case proto.CmdSendFile: - send(writeCh, errMsg("file transfer not yet implemented")) + send(errMsg("file transfer not yet implemented")) default: - send(writeCh, errMsg(fmt.Sprintf("unknown command: %s", cmd.Type))) + send(errMsg(fmt.Sprintf("unknown command: %s", cmd.Type))) } } - if networkCancel != nil { - networkCancel() - } + close(done) close(writeCh) log.Printf("ipc: UI client disconnected") } -func send(ch chan<- []byte, msg proto.IpcMessage) { - line, err := json.Marshal(msg) - if err != nil { - return - } - select { - case ch <- line: - default: - } -} - func errMsg(s string) proto.IpcMessage { return proto.IpcMessage{Type: proto.EvtError, ErrorMessage: s} } diff --git a/test-network.sh b/test-network.sh new file mode 100755 index 0000000..58b1d17 --- /dev/null +++ b/test-network.sh @@ -0,0 +1,251 @@ +#!/usr/bin/env bash +# test-network.sh — spin up anchor + 3 peers, connect them, exchange messages. +# Usage: ./test-network.sh +# Requires: go, nc, jq + +set -euo pipefail + +# ── colours ─────────────────────────────────────────────────────────────────── +RED='\033[0;31m'; GREEN='\033[0;32m'; YELLOW='\033[1;33m' +CYAN='\033[0;36m'; MAGENTA='\033[0;35m'; BLUE='\033[0;34m' +BOLD='\033[1m'; DIM='\033[2m'; RESET='\033[0m' + +ANCHOR_COLOR=$YELLOW +ALICE_COLOR=$CYAN +BOB_COLOR=$MAGENTA +CHARLIE_COLOR=$GREEN + +# ── config ──────────────────────────────────────────────────────────────────── +ANCHOR_PORT=19339 +ALICE_IPC=19337 +BOB_IPC=19340 +CHARLIE_IPC=19343 +NETWORK_NAME="test-$(date +%s)" +DATA_ROOT=$(mktemp -d /tmp/waste-test-XXXXXX) + +# ── cleanup ─────────────────────────────────────────────────────────────────── +PIDS=() +cleanup() { + echo "" + echo -e "${DIM}── cleaning up ──────────────────────────────────────────${RESET}" + for pid in "${PIDS[@]:-}"; do + kill "$pid" 2>/dev/null || true + done + wait 2>/dev/null || true + rm -rf "$DATA_ROOT" + echo -e "${DIM}done.${RESET}" +} +trap cleanup EXIT INT TERM + +# ── helpers ─────────────────────────────────────────────────────────────────── +log() { + local color="$1"; shift + local label="$1"; shift + echo -e "${color}${BOLD}[${label}]${RESET} $*" +} + +# Pretty-print a raw IPC JSON line +pretty() { + local color="$1" + local label="$2" + local line="$3" + local type + type=$(echo "$line" | jq -r '.type // "?"' 2>/dev/null) + case "$type" in + state_snapshot) + local id nick + id=$(echo "$line" | jq -r '.local_peer.id[:16]' 2>/dev/null | sed 's/.\{4\}/& /g') + nick=$(echo "$line" | jq -r '.local_peer.alias' 2>/dev/null) + echo -e "${color}${BOLD}[${label}]${RESET} ${DIM}state_snapshot${RESET} id=${CYAN}${id}${RESET} alias=${BOLD}${nick}${RESET}" + ;; + session_ready) + local pid nick + pid=$(echo "$line" | jq -r 'if .peer_id then .peer_id[:16] else "?" end' 2>/dev/null | sed 's/.\{4\}/& /g') + nick=$(echo "$line" | jq -r '.nick // "?"' 2>/dev/null) + echo -e "${color}${BOLD}[${label}]${RESET} ${GREEN}✓ session_ready${RESET} peer=${BOLD}${nick}${RESET} (${DIM}${pid}${RESET})" + ;; + peer_connected) + local nick + nick=$(echo "$line" | jq -r '.peer.alias // "?"' 2>/dev/null) + echo -e "${color}${BOLD}[${label}]${RESET} ${GREEN}→ peer_connected${RESET} ${BOLD}${nick}${RESET}" + ;; + peer_disconnected) + local pid + pid=$(echo "$line" | jq -r 'if .peer_id then .peer_id[:8] else "?" end' 2>/dev/null) + echo -e "${color}${BOLD}[${label}]${RESET} ${RED}← peer_disconnected${RESET} ${DIM}${pid}${RESET}" + ;; + message_received) + local from body room + from=$(echo "$line" | jq -r '.message.from[:8] // "?"' 2>/dev/null) + body=$(echo "$line" | jq -r '.message.body // ""' 2>/dev/null) + room=$(echo "$line" | jq -r '.message.room // ""' 2>/dev/null) + echo -e "${color}${BOLD}[${label}]${RESET} ${BOLD}💬 #${room}${RESET} ${DIM}<${from}…>${RESET} ${body}" + ;; + error) + local msg + msg=$(echo "$line" | jq -r '.error_message // .' 2>/dev/null) + echo -e "${color}${BOLD}[${label}]${RESET} ${RED}✗ error:${RESET} ${msg}" + ;; + *) + echo -e "${color}${BOLD}[${label}]${RESET} ${DIM}${line}${RESET}" + ;; + esac +} + +# Subscribe to a peer's IPC events in the background, pretty-printing each line. +subscribe() { + local color="$1" + local label="$2" + local port="$3" + ( + set +e + nc 127.0.0.1 "$port" 2>/dev/null | while IFS= read -r line; do + pretty "$color" "$label" "$line" + done + ) & + PIDS+=($!) +} + +# Send a single JSON command to a peer's IPC port (fire-and-forget). +ipc() { + local port="$1" + local json="$2" + echo "$json" | nc -q 0 127.0.0.1 "$port" >/dev/null 2>&1 || true +} + +# Wait until a TCP port accepts connections (up to N seconds). +wait_port() { + local port="$1" + local label="$2" + local n=0 + while ! nc -z 127.0.0.1 "$port" 2>/dev/null; do + sleep 0.1 + n=$(( n + 1 )) + if [ "$n" -gt 80 ]; then + echo -e "${RED}timeout waiting for ${label} on :${port}${RESET}" >&2 + exit 1 + fi + done +} + +# ── build ───────────────────────────────────────────────────────────────────── +echo "" +echo -e "${BOLD}waste-go network test${RESET}" +echo -e "${DIM}────────────────────────────────────────────────────────${RESET}" +echo -e "${DIM}network : ${BOLD}${NETWORK_NAME}${RESET}" +echo -e "${DIM}data : ${DATA_ROOT}${RESET}" +echo "" + +echo -e "${DIM}building binaries…${RESET}" +go build -o "$DATA_ROOT/bin/waste-anchor" ./cmd/anchor +go build -o "$DATA_ROOT/bin/waste-daemon" ./cmd/daemon +echo -e "${GREEN}✓ built${RESET}" +echo "" + +# ── anchor ──────────────────────────────────────────────────────────────────── +log "$ANCHOR_COLOR" "anchor" "starting on :${ANCHOR_PORT}" +"$DATA_ROOT/bin/waste-anchor" -bind "127.0.0.1:${ANCHOR_PORT}" \ + 2> >(while IFS= read -r l; do echo -e "${ANCHOR_COLOR}${DIM}[anchor] ${l}${RESET}"; done) & +PIDS+=($!) +wait_port "$ANCHOR_PORT" "anchor" +log "$ANCHOR_COLOR" "anchor" "ready" +echo "" + +# ── daemons ─────────────────────────────────────────────────────────────────── +ANCHOR_URL="ws://127.0.0.1:${ANCHOR_PORT}/ws" + +log "$ALICE_COLOR" "alice" "starting daemon (ipc :${ALICE_IPC})" +"$DATA_ROOT/bin/waste-daemon" \ + -alias alice \ + -data-dir "$DATA_ROOT/alice" \ + -ipc-port "$ALICE_IPC" \ + -anchor "$ANCHOR_URL" \ + 2> >(while IFS= read -r l; do echo -e "${ALICE_COLOR}${DIM}[alice] ${l}${RESET}"; done) & +PIDS+=($!) + +log "$BOB_COLOR" "bob" "starting daemon (ipc :${BOB_IPC})" +"$DATA_ROOT/bin/waste-daemon" \ + -alias bob \ + -data-dir "$DATA_ROOT/bob" \ + -ipc-port "$BOB_IPC" \ + -anchor "$ANCHOR_URL" \ + 2> >(while IFS= read -r l; do echo -e "${BOB_COLOR}${DIM}[bob] ${l}${RESET}"; done) & +PIDS+=($!) + +log "$CHARLIE_COLOR" "charlie" "starting daemon (ipc :${CHARLIE_IPC})" +"$DATA_ROOT/bin/waste-daemon" \ + -alias charlie \ + -data-dir "$DATA_ROOT/charlie" \ + -ipc-port "$CHARLIE_IPC" \ + -anchor "$ANCHOR_URL" \ + 2> >(while IFS= read -r l; do echo -e "${CHARLIE_COLOR}${DIM}[charlie]${l}${RESET}"; done) & +PIDS+=($!) + +wait_port "$ALICE_IPC" "alice" +wait_port "$BOB_IPC" "bob" +wait_port "$CHARLIE_IPC" "charlie" +echo "" + +# ── subscribe ───────────────────────────────────────────────────────────────── +echo -e "${DIM}subscribing to IPC event streams…${RESET}" +subscribe "$ALICE_COLOR" "alice " "$ALICE_IPC" +subscribe "$BOB_COLOR" "bob " "$BOB_IPC" +subscribe "$CHARLIE_COLOR" "charlie" "$CHARLIE_IPC" +sleep 0.3 # let state_snapshot lines arrive + +# ── join network ────────────────────────────────────────────────────────────── +echo "" +echo -e "${DIM}────────────────────────────────────────────────────────${RESET}" +echo -e "joining all peers to network: ${BOLD}${NETWORK_NAME}${RESET}" +echo -e "${DIM}────────────────────────────────────────────────────────${RESET}" + +JOIN=$(printf '{"type":"join_network","network_name":"%s"}' "$NETWORK_NAME") +ipc "$ALICE_IPC" "$JOIN" +sleep 0.2 +ipc "$BOB_IPC" "$JOIN" +sleep 0.2 +ipc "$CHARLIE_IPC" "$JOIN" + +echo -e "${DIM}waiting for ICE / DataChannel setup (up to 10s)…${RESET}" +sleep 6 + +# ── chat ────────────────────────────────────────────────────────────────────── +echo "" +echo -e "${DIM}────────────────────────────────────────────────────────${RESET}" +echo -e "sending messages" +echo -e "${DIM}────────────────────────────────────────────────────────${RESET}" + +ipc "$ALICE_IPC" '{"type":"send_message","room":"general","body":"hey everyone, alice here"}' +sleep 0.4 +ipc "$BOB_IPC" '{"type":"send_message","room":"general","body":"hi alice! bob checking in"}' +sleep 0.4 +ipc "$CHARLIE_IPC" '{"type":"send_message","room":"general","body":"charlie online. the mesh works!"}' +sleep 0.4 +ipc "$ALICE_IPC" '{"type":"send_message","room":"general","body":"nice — three-way chat over WebRTC 🎉"}' +sleep 0.4 +ipc "$BOB_IPC" '{"type":"send_message","room":"general","body":"no central server, fully peer-to-peer"}' +sleep 0.4 + +# ── state snapshot ──────────────────────────────────────────────────────────── +echo "" +echo -e "${DIM}────────────────────────────────────────────────────────${RESET}" +echo -e "requesting state snapshots" +echo -e "${DIM}────────────────────────────────────────────────────────${RESET}" +ipc "$ALICE_IPC" '{"type":"get_state"}' +ipc "$BOB_IPC" '{"type":"get_state"}' +ipc "$CHARLIE_IPC" '{"type":"get_state"}' +sleep 0.5 + +# ── leave ───────────────────────────────────────────────────────────────────── +echo "" +echo -e "${DIM}────────────────────────────────────────────────────────${RESET}" +echo -e "charlie leaves the network" +echo -e "${DIM}────────────────────────────────────────────────────────${RESET}" +ipc "$CHARLIE_IPC" '{"type":"leave_network"}' +sleep 1 + +echo "" +echo -e "${DIM}────────────────────────────────────────────────────────${RESET}" +echo -e "${GREEN}${BOLD}test complete${RESET}" +echo -e "${DIM}────────────────────────────────────────────────────────${RESET}" +sleep 0.5