Maven/internal/voice/voice_test.go

package voice

import (
	"context"
	"encoding/json"
	"net"
	"os"
	"syscall"
	"testing"
	"time"

	"github.com/kami/maven/internal/audio"
)

// newTestListener returns a TCP listener on a random port with SO_REUSEADDR.
// It retries a few times if the port is temporarily unavailable (e.g., TIME_WAIT).
func newTestListener(t *testing.T) net.Listener {
	t.Helper()
	lc := net.ListenConfig{
		Control: func(network, address string, c syscall.RawConn) error {
			var err error
			c.Control(func(fd uintptr) {
				err = syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_REUSEADDR, 1)
			})
			return err
		},
	}
	var ln net.Listener
	var err error
	for i := 0; i < 5; i++ {
		ln, err = lc.Listen(context.Background(), "tcp", "127.0.0.1:0")
		if err == nil {
			break
		}
		if !isAddrInUse(err) {
			t.Fatalf("listen: %v", err)
		}
		time.Sleep(100 * time.Millisecond)
	}
	if err != nil {
		t.Fatalf("listen after retries: %v", err)
	}
	// Close the probe before returning: it only existed to reserve a free port
	// (127.0.0.1:0 → a concrete port). The Server rebinds that exact addr in its
	// own Listen(), which fails with EADDRINUSE while the probe still holds it.
	// Addr() keeps returning the address after Close, and an unconnected listener
	// leaves no TIME_WAIT, so the rebind is immediate and clean.
	_ = ln.Close()
	return ln
}

func isAddrInUse(err error) bool {
	if err == nil {
		return false
	}
	if opErr, ok := err.(*net.OpError); ok {
		if sysErr, ok := opErr.Err.(*os.SyscallError); ok {
			if err := sysErr.Err; err == syscall.EADDRINUSE {
				return true
			}
		}
	}
	return false
}

// wait a bit for OS to release the port after close.
func waitPort() { time.Sleep(500 * time.Millisecond) }

// stubHandler — satisfies voice.Handler for tests.
type stubHandler struct {
	lastReq PushToTalkReq
}

func (h *stubHandler) HandlePushToTalk(_ context.Context, req PushToTalkReq, _ uint64) (PushToTalkResp, error) {
	h.lastReq = req
	return PushToTalkResp{
		ReplyAudio: audio.Audio{Format: audio.PCM16kMono, Bytes: []byte("reply")},
		ReplyText:  "got it",
	}, nil
}

func TestServerAcceptsAndRemovesSession(t *testing.T) {
	l := newTestListener(t)
	sess := NewSessions()
	h := &stubHandler{}
	srv := NewServer(l.Addr().String(), h, sess)
	if err := srv.Listen(); err != nil {
		t.Fatalf("listen: %v", err)
	}
	defer func() {
		_ = srv.Close()
		waitPort()
	}()
	// Accept connections in the background; Serve blocks until Close. Without
	// it the listener binds but never accepts, so a client round-trip hangs.
	go func() { _ = srv.Serve() }()

	if sess.Active() != 0 {
		t.Fatalf("active before connect: %d", sess.Active())
	}

	conn, err := net.Dial("tcp", l.Addr().String())
	if err != nil {
		t.Fatalf("dial: %v", err)
	}
	conn.Close() // quick disconnect

	// Give serveConn a moment to register then remove.
	time.Sleep(50 * time.Millisecond)
	if sess.Active() != 0 {
		t.Fatalf("active after disconnect: %d, want 0", sess.Active())
	}
}

func TestClientPushToTalkRoundTrip(t *testing.T) {
	l := newTestListener(t)
	sess := NewSessions()
	h := &stubHandler{}
	srv := NewServer(l.Addr().String(), h, sess)
	if err := srv.Listen(); err != nil {
		t.Fatalf("listen: %v", err)
	}
	defer func() {
		_ = srv.Close()
		waitPort()
	}()
	// Accept connections in the background; Serve blocks until Close. Without
	// it the listener binds but never accepts, so a client round-trip hangs.
	go func() { _ = srv.Serve() }()

	c := Dial(l.Addr().String())
	defer c.Close()

	resp, err := c.PushToTalk(context.Background(), audio.Audio{Format: audio.PCM16kMono, Bytes: []byte("hello")}, "ru")
	if err != nil {
		t.Fatalf("PushToTalk: %v", err)
	}
	if resp.ReplyText != "got it" {
		t.Fatalf("ReplyText: %q, want %q", resp.ReplyText, "got it")
	}
	if len(resp.ReplyAudio.Bytes) == 0 {
		t.Fatalf("ReplyAudio empty")
	}
	if h.lastReq.Audio.Bytes == nil {
		t.Fatalf("handler never got audio")
	}
}

func TestClientListenModeReceivesPush(t *testing.T) {
	l := newTestListener(t)
	sess := NewSessions()
	h := &stubHandler{}
	srv := NewServer(l.Addr().String(), h, sess)
	if err := srv.Listen(); err != nil {
		t.Fatalf("listen: %v", err)
	}
	defer func() {
		_ = srv.Close()
		waitPort()
	}()
	// Accept connections in the background; Serve blocks until Close. Without
	// it the listener binds but never accepts, so a client round-trip hangs.
	go func() { _ = srv.Serve() }()

	c := Dial(l.Addr().String())
	defer c.Close()

	// Run the push receiver in the background: it blocks reading frames until
	// the conn closes (ctx cancel alone can't interrupt a blocking read). A
	// proactive push from the server is delivered to the handler, which hands
	// the audio to a channel the test waits on.
	got := make(chan audio.Audio, 1)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go func() {
		_ = c.RunPushReceiver(ctx, pushHandlerFunc(func(p Push) {
			if p.Kind != PushKindAudioNudge {
				return
			}
			var ap AudioNudgePush
			if err := json.Unmarshal(p.Params, &ap); err == nil {
				select {
				case got <- ap.Audio:
				default:
				}
			}
		}))
	}()

	// serveConn registers the session on accept, but Accept→Add is async — wait
	// for it before pushing, or PushToMostRecent finds no session.
	for i := 0; i < 100 && sess.Active() < 1; i++ {
		time.Sleep(10 * time.Millisecond)
	}
	if sess.Active() < 1 {
		t.Fatal("session never registered")
	}

	// Server pushes to the session.
	push := AudioNudgePush{
		RuleName: "test",
		Severity: 3,
		Audio:    audio.Audio{Format: audio.PCM16kMono, Bytes: []byte("proactive")},
		Text:     "proactive text",
		Ts:       time.Now(),
	}
	if err := sess.PushToMostRecent(context.Background(), push); err != nil {
		t.Fatalf("PushToMostRecent: %v", err)
	}

	select {
	case received := <-got:
		if string(received.Bytes) != "proactive" {
			t.Fatalf("received audio mismatch: %q", string(received.Bytes))
		}
	case <-time.After(2 * time.Second):
		t.Fatal("never received pushed audio")
	}
}

type pushHandlerFunc func(Push)

func (f pushHandlerFunc) OnPush(p Push) { f(p) }

// The whole point of the per-conn reader (V-671): mavwaked speaks utterances
// and must hear nudges, and the server routes a nudge to the session that
// spoke most recently. A second listening conn would never be picked, so both
// directions have to share one conn.
func TestClientSendsAndListensOnOneConn(t *testing.T) {
	l := newTestListener(t)
	sess := NewSessions()
	h := &stubHandler{}
	srv := NewServer(l.Addr().String(), h, sess)
	if err := srv.Listen(); err != nil {
		t.Fatalf("listen: %v", err)
	}
	defer func() {
		_ = srv.Close()
		waitPort()
	}()
	go func() { _ = srv.Serve() }()

	c := Dial(l.Addr().String())
	defer c.Close()

	got := make(chan audio.Audio, 4)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go func() {
		_ = c.RunPushReceiver(ctx, pushHandlerFunc(func(p Push) {
			var ap AudioNudgePush
			if err := json.Unmarshal(p.Params, &ap); err == nil {
				got <- ap.Audio
			}
		}))
	}()

	for i := 0; i < 100 && sess.Active() < 1; i++ {
		time.Sleep(10 * time.Millisecond)
	}
	if sess.Active() != 1 {
		t.Fatalf("active sessions = %d, want exactly 1", sess.Active())
	}

	// A round-trip while the receiver is running. Before the reader owned the
	// conn, this and the receiver raced for every frame.
	resp, err := c.PushToTalk(context.Background(), audio.Audio{Format: audio.PCM16kMono, Bytes: []byte("hello")}, "ru")
	if err != nil {
		t.Fatalf("PushToTalk with a receiver running: %v", err)
	}
	if resp.ReplyText != "got it" {
		t.Fatalf("ReplyText = %q, want %q", resp.ReplyText, "got it")
	}

	// And the nudge still arrives, on the session that just spoke.
	err = sess.PushToMostRecent(context.Background(), AudioNudgePush{
		RuleName: "after-speaking",
		Audio:    audio.Audio{Format: audio.PCM16kMono, Bytes: []byte("proactive")},
	})
	if err != nil {
		t.Fatalf("PushToMostRecent: %v", err)
	}
	select {
	case a := <-got:
		if string(a.Bytes) != "proactive" {
			t.Fatalf("received %q, want the nudge audio", string(a.Bytes))
		}
	case <-time.After(2 * time.Second):
		t.Fatal("nudge never reached the handler after the client had spoken")
	}
}

// A request abandoned by its context must not leave its slot behind, or a
// long-running client leaks one channel per timeout.
func TestClientForgetsAbandonedRequests(t *testing.T) {
	l := newTestListener(t)
	sess := NewSessions()
	srv := NewServer(l.Addr().String(), &blockingHandler{}, sess)
	if err := srv.Listen(); err != nil {
		t.Fatalf("listen: %v", err)
	}
	defer func() {
		_ = srv.Close()
		waitPort()
	}()
	go func() { _ = srv.Serve() }()

	c := Dial(l.Addr().String())
	defer c.Close()

	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
	defer cancel()
	_, err := c.PushToTalk(ctx, audio.Audio{Format: audio.PCM16kMono, Bytes: []byte("x")}, "ru")
	if err == nil {
		t.Fatal("expected the round-trip to fail on its context")
	}

	c.mu.Lock()
	n := len(c.pending)
	c.mu.Unlock()
	if n != 0 {
		t.Fatalf("pending = %d after an abandoned request, want 0", n)
	}
}

// blockingHandler never answers, so the client's context is what ends the
// round-trip.
type blockingHandler struct{}

func (blockingHandler) HandlePushToTalk(ctx context.Context, _ PushToTalkReq, _ uint64) (PushToTalkResp, error) {
	<-ctx.Done()
	return PushToTalkResp{}, ctx.Err()
}