Maven/internal/voice/client.go

// voice/client.go — the client-side dialer.
//
// Used by cmd/mavenclient to round-trip a PushToTalk. One Client = one TCP
// conn = one frame at a time (singleplex — the reference client doesn't
// multiplex; production may, but the wire shape already carries IDs so
// multiplexing is a client-lib affair, not a wire change). When the user
// wants to receive proactive voice pushes, RunPushReceiver spawns a reader
// goroutine that calls the PushHandler for each server-initiated Push frame.
//
// The wire is symmetric: a Request from the client is answered by a
// Response with a matching ID, OR a server-initiated Push frame (no ID)
// may arrive interleaved. SendRequest loops reading frames, drops Push
// frames to the harness if a receiver is running (or silently if not),
// and returns the first Response with the matching ID.
package voice

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net"
	"sync"
	"sync/atomic"
	"time"

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

// PushHandler — receives server-initiated Push frames. cmd/mavenclient's
// -listen mode wires one that writes the audio to disk / aplay. The Push
// arrives on the reader goroutine; the handler runs there too, so a slow
// handler blocks subsequent frames (intentional — proactive voice playback
// should not queue up behind a stuck player; the next nudge replaces the
// stale one in the user's attention, not dogpiles on it).
type PushHandler interface {
	OnPush(p Push)
}

// Client — one connection to the voice.Server.
type Client struct {
	addr   string
	mu     sync.Mutex
	c      net.Conn
	nextID atomic.Uint64

	// pushCh fan-out: a reader goroutine (started by RunPushReceiver)
	// writes Push frames here; SendRequest also drains it when no reader
	// is running (drops the frame in that case).
	pushMu sync.Mutex
	pushH  PushHandler
}

// Dial returns a Client that will connect to addr on first use.
func Dial(addr string) *Client { return &Client{addr: addr} }

// Close releases the conn. Idempotent.
func (c *Client) Close() error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.c == nil {
		return nil
	}
	err := c.c.Close()
	c.c = nil
	return err
}

// PushToTalk is the convenience for fire-and-forget: send audio bytes +
// language, block for the reply. Used by cmd/mavenclient's one-shot mode.
func (c *Client) PushToTalk(ctx context.Context, a audio.Audio, lang string) (PushToTalkResp, error) {
	var out PushToTalkResp
	err := c.SendRequest(ctx, MethodPushToTalk, PushToTalkReq{Audio: a, Lang: lang, Surface: SurfacePCClient}, &out)
	return out, err
}

// SendRequest sends one Request frame and waits for the matching Response.
// Push frames received while waiting are dropped on the floor UNLESS a
// PushHandler has been wired via RunPushReceiver, in which case the handler
// is invoked inline (still synchronous with the SendRequest caller's
// read). For sanity, the reference client runs either one-shot (no
// receiver) or interactive (RunPushReceiver, no concurrent SendRequest).
func (c *Client) SendRequest(ctx context.Context, m Method, params any, out any) error {
	body, err := marshalParams(params)
	if err != nil {
		return err
	}
	id := c.nextID.Add(1)
	req := Request{ID: id, Method: m, Params: body}

	c.mu.Lock()
	if err := c.ensureConnLocked(ctx); err != nil {
		c.mu.Unlock()
		return err
	}
	conn := c.c
	c.mu.Unlock()

	if dl, ok := ctx.Deadline(); ok {
		_ = conn.SetDeadline(dl)
	} else {
		_ = conn.SetDeadline(time.Now().Add(120 * time.Second))
	}
	defer conn.SetDeadline(time.Time{})

	if err := writeFrame(conn, &req); err != nil {
		c.teardown()
		return err
	}
	for {
		resp, push, err := readOneFrame(conn)
		if err != nil {
			c.teardown()
			return err
		}
		if push != nil {
			c.deliverPush(*push)
			continue
		}
		if resp.ID != id {
			continue // not ours; ignore (singleplex ⇒ shouldn't happen)
		}
		if resp.Error != nil {
			return hydrate(resp.Error)
		}
		if out != nil {
			if err := json.Unmarshal(resp.Result, out); err != nil {
				return fmt.Errorf("voice: unmarshal result: %w", err)
			}
		}
		return nil
	}
}

// RunPushReceiver spawns a reader goroutine that delivers Push frames to h
// until the conn closes or Close is called. Today's reference client uses
// this in -listen mode (proactive voice playback). SendRequest and
// RunPushReceiver SHOULD NOT be used concurrently on the same Client — the
// wire is singleplex at the reference client's scale; production picks one
// mode per conn. Returns when the goroutine ends (ctx cancel or conn close).
func (c *Client) RunPushReceiver(ctx context.Context, h PushHandler) error {
	c.mu.Lock()
	if err := c.ensureConnLocked(ctx); err != nil {
		c.mu.Unlock()
		return err
	}
	conn := c.c
	c.mu.Unlock()

	c.pushMu.Lock()
	c.pushH = h
	c.pushMu.Unlock()

	defer func() {
		c.pushMu.Lock()
		c.pushH = nil
		c.pushMu.Unlock()
	}()

	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		default:
		}
		_, push, err := readOneFrame(conn)
		if err != nil {
			if errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) {
				return nil
			}
			return err
		}
		if push != nil {
			c.deliverPush(*push)
		}
	}
}

func (c *Client) deliverPush(p Push) {
	c.pushMu.Lock()
	h := c.pushH
	c.pushMu.Unlock()
	if h != nil {
		h.OnPush(p)
	}
}

func (c *Client) ensureConnLocked(ctx context.Context) error {
	if c.c != nil {
		return nil
	}
	d := net.Dialer{Timeout: 10 * time.Second}
	conn, err := d.DialContext(ctx, "tcp", c.addr)
	if err != nil {
		return fmt.Errorf("voice: dial %s: %w", c.addr, err)
	}
	c.c = conn
	return nil
}

func (c *Client) teardown() {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.c != nil {
		_ = c.c.Close()
		c.c = nil
	}
}

// readOneFrame — reads one frame and tries Response-then-Push shape. A
// frame with a non-zero `id` is a Response; a frame with `kind` and no `id`
// is a Push. Exactly one return value is non-nil.
func readOneFrame(r io.Reader) (*Response, *Push, error) {
	var raw struct {
		ID     uint64          `json:"id"`
		Result json.RawMessage `json:"r,omitempty"`
		Error  *RpcError       `json:"e,omitempty"`
		Kind   PushKind        `json:"kind,omitempty"`
		Params json.RawMessage `json:"p,omitempty"`
	}
	if err := readFrame(r, &raw); err != nil {
		return nil, nil, err
	}
	if raw.Kind != "" && raw.ID == 0 {
		return nil, &Push{Kind: raw.Kind, Params: raw.Params}, nil
	}
	return &Response{ID: raw.ID, Result: raw.Result, Error: raw.Error}, nil, nil
}

func marshalParams(v any) (json.RawMessage, error) {
	if v == nil {
		return nil, nil
	}
	b, err := json.Marshal(v)
	if err != nil {
		return nil, fmt.Errorf("voice: marshal params: %w", err)
	}
	return b, nil
}