dwarf-go/dwarfctl/internal/transport/client.go

package transport

import (
	"fmt"
	"log"
	"sync"
	"time"

	"github.com/gorilla/websocket"
	pb "github.com/antitbone/dwarfctl/proto"
	"google.golang.org/protobuf/proto"
)

const (
	WsMajorVersion = 2
	WsMinorVersion = 3
	WsPort         = 9900
)

// MsgType mirrors WsMessageType: request=0, response=1, notification=2, reply=3.
type MsgType uint32

const (
	MsgRequest      MsgType = 0
	MsgResponse     MsgType = 1
	MsgNotification MsgType = 2
	MsgReply        MsgType = 3
)

// Client is a WebSocket client for the DWARF telescope control plane.
type Client struct {
	conn      *websocket.Conn
	clientID  string
	deviceID  uint32
	mu        sync.Mutex
	pending   map[uint32]chan *pb.WsPacket
	notifyChs []chan *pb.WsPacket
	done      chan struct{}
	Debug     bool
}

// NewClient creates a client with the given client_id and device_id.
func NewClient(clientID string, deviceID uint32) *Client {
	return &Client{
		clientID: clientID,
		deviceID: deviceID,
		pending:  make(map[uint32]chan *pb.WsPacket),
		done:     make(chan struct{}),
	}
}

// Connect opens the WebSocket to the telescope.
func (c *Client) Connect(ip string) error {
	url := fmt.Sprintf("ws://%s:%d/?client_id=%s", ip, WsPort, c.clientID)
	dialer := websocket.Dialer{
		HandshakeTimeout: 10 * time.Second,
	}
	conn, _, err := dialer.Dial(url, nil)
	if err != nil {
		return fmt.Errorf("ws dial %s: %w", url, err)
	}
	c.conn = conn
	go c.readLoop()
	return nil
}

// Close shuts down the connection.
func (c *Client) Close() error {
	close(c.done)
	if c.conn != nil {
		return c.conn.Close()
	}
	return nil
}

// IsConnected returns true if the WebSocket is open.
func (c *Client) IsConnected() bool {
	return c.conn != nil
}

// readLoop continuously reads WsPacket frames and dispatches them.
func (c *Client) readLoop() {
	for {
		select {
		case <-c.done:
			return
		default:
		}
		_, data, err := c.conn.ReadMessage()
		if err != nil {
			if c.Debug {
				log.Printf("[WS] read error: %v", err)
			}
			return
		}
		pkt := &pb.WsPacket{}
		if err := proto.Unmarshal(data, pkt); err != nil {
			if c.Debug {
				log.Printf("[WS] unmarshal error: %v (raw %d bytes)", err, len(data))
			}
			continue
		}
		if c.Debug {
			log.Printf("[WS] recv cmd=%d module=%d type=%d data=%d bytes",
				pkt.GetCmd(), pkt.GetModuleId(), pkt.GetType(), len(pkt.GetData()))
		}
		// Dispatch based on type. Many telescopes send responses with
		// type=0 (default proto3 value) instead of type=1. So we try
		// response dispatch for type 0, 1, and 3.
		switch MsgType(pkt.GetType()) {
		case MsgNotification:
			c.dispatchNotification(pkt)
			// Also try response dispatch — some notifications double as acks
			c.dispatchResponse(pkt)
		default:
			// Covers type=0 (unset), type=1 (response), type=3 (reply)
			c.dispatchResponse(pkt)
			c.dispatchNotification(pkt)
		}
	}
}

// dispatchResponse delivers a response to the pending request waiter.
func (c *Client) dispatchResponse(pkt *pb.WsPacket) {
	c.mu.Lock()
	ch, ok := c.pending[pkt.GetCmd()]
	if ok {
		delete(c.pending, pkt.GetCmd())
	}
	c.mu.Unlock()
	if ok {
		ch <- pkt
	}
}

// dispatchNotification fans out to all subscribers.
func (c *Client) dispatchNotification(pkt *pb.WsPacket) {
	c.mu.Lock()
	chs := make([]chan *pb.WsPacket, len(c.notifyChs))
	copy(chs, c.notifyChs)
	c.mu.Unlock()
	for _, ch := range chs {
		select {
		case ch <- pkt:
		default:
		}
	}
}

// SubscribeNotifications returns a channel for receiving NOTIFY packets.
func (c *Client) SubscribeNotifications() chan *pb.WsPacket {
	ch := make(chan *pb.WsPacket, 64)
	c.mu.Lock()
	c.notifyChs = append(c.notifyChs, ch)
	c.mu.Unlock()
	return ch
}

// Send sends a raw command with the given module_id and cmd, carrying
// the serialized inner proto message as data. It returns the response packet.
func (c *Client) Send(moduleID uint32, cmd uint32, data []byte, timeout time.Duration) (*pb.WsPacket, error) {
	pkt := &pb.WsPacket{
		MajorVersion: WsMajorVersion,
		MinorVersion: WsMinorVersion,
		DeviceId:     c.deviceID,
		ModuleId:     moduleID,
		Cmd:          cmd,
		Type:         uint32(MsgRequest),
		Data:         data,
		ClientId:     c.clientID,
	}
	raw, err := proto.Marshal(pkt)
	if err != nil {
		return nil, fmt.Errorf("marshal WsPacket: %w", err)
	}

	respCh := make(chan *pb.WsPacket, 1)
	c.mu.Lock()
	c.pending[cmd] = respCh
	c.mu.Unlock()

	if err := c.conn.WriteMessage(websocket.BinaryMessage, raw); err != nil {
		c.mu.Lock()
		delete(c.pending, cmd)
		c.mu.Unlock()
		return nil, fmt.Errorf("ws write: %w", err)
	}

	select {
	case resp := <-respCh:
		return resp, nil
	case <-time.After(timeout):
		c.mu.Lock()
		delete(c.pending, cmd)
		c.mu.Unlock()
		return nil, fmt.Errorf("timeout waiting for response to cmd %d", cmd)
	case <-c.done:
		return nil, fmt.Errorf("connection closed")
	}
}

// SendNotify sends a command without waiting for a response (fire-and-forget).
func (c *Client) SendNotify(moduleID uint32, cmd uint32, data []byte) error {
	pkt := &pb.WsPacket{
		MajorVersion: WsMajorVersion,
		MinorVersion: WsMinorVersion,
		DeviceId:     c.deviceID,
		ModuleId:     moduleID,
		Cmd:          cmd,
		Type:         uint32(MsgRequest),
		Data:         data,
		ClientId:     c.clientID,
	}
	raw, err := proto.Marshal(pkt)
	if err != nil {
		return err
	}
	return c.conn.WriteMessage(websocket.BinaryMessage, raw)
}