uopi/pkg/ca/client.go

package ca

import (
	"context"
	"fmt"
	"os"
	"os/user"
	"strings"
	"sync"

	"github.com/uopi/goca/proto"
)

// -------------------------------------------------------------------------- //
// Config                                                                      //
// -------------------------------------------------------------------------- //

// Config holds the configuration for a Client.
// All fields are optional; sensible defaults are applied by NewClient.
type Config struct {
	// AddrList is the list of CA server addresses ("host" or "host:port").
	// Corresponds to EPICS_CA_ADDR_LIST.
	AddrList []string

	// AutoAddrList, when true, appends the IPv4 broadcast address of every
	// local network interface to AddrList.  Defaults to true.
	// Corresponds to EPICS_CA_AUTO_ADDR_LIST.
	AutoAddrList bool

	// ClientName is announced to every CA server in the CLIENT_NAME message.
	// Defaults to the executable base name.
	ClientName string

	// HostName is announced to every CA server in the HOST_NAME message.
	// Defaults to the OS hostname.
	HostName string
}

// ConfigFromEnv reads the standard EPICS CA environment variables and returns
// a ready-to-use Config.
//
//	EPICS_CA_ADDR_LIST        — space-separated list of server addresses
//	EPICS_CA_AUTO_ADDR_LIST   — "NO" disables automatic broadcast addresses
func ConfigFromEnv() Config {
	// CA security ACLs match on the client username, not the program name.
	// Use the OS username (same as libca), falling back to $USER, then "user".
	clientName := "user"
	if u, err := user.Current(); err == nil && u.Username != "" {
		clientName = u.Username
	} else if v := os.Getenv("USER"); v != "" {
		clientName = v
	}
	host, _ := os.Hostname()
	cfg := Config{
		AutoAddrList: true,
		ClientName:   clientName,
		HostName:     host,
	}
	if v := os.Getenv("EPICS_CA_ADDR_LIST"); v != "" {
		cfg.AddrList = strings.Fields(v)
	}
	if strings.EqualFold(os.Getenv("EPICS_CA_AUTO_ADDR_LIST"), "no") {
		cfg.AutoAddrList = false
	}
	return cfg
}

// -------------------------------------------------------------------------- //
// Client                                                                      //
// -------------------------------------------------------------------------- //

// Client is a thread-safe EPICS Channel Access client.
//
// A single Client should serve an entire application.  It maintains a pool of
// persistent TCP circuits (one per IOC) and a shared UDP search engine.
// Channels and subscriptions survive IOC restarts automatically.
//
// Usage:
//
//	cli, err := ca.NewClient(ctx, ca.ConfigFromEnv())
//	defer cli.Close()
//
//	ch := make(chan proto.TimeValue, 16)
//	cancel, err := cli.Subscribe(ctx, "MY:PV", ch)
//	defer cancel()
//
//	for tv := range ch { fmt.Println(tv.Double) }
type Client struct {
	cfg    Config
	ctx    context.Context
	cancel context.CancelFunc
	search *searchEngine

	mu       sync.Mutex
	circuits map[string]*circuit // IOC TCP addr → circuit
}

// NewClient creates a new CA client and starts background I/O.
// ctx governs the lifetime of the client; cancelling it is equivalent to
// calling Close.
//
// Returns an error only if no search addresses can be derived from cfg.
func NewClient(ctx context.Context, cfg Config) (*Client, error) {
	addrs := resolveAddrs(cfg.AddrList, proto.DefaultPort)
	if cfg.AutoAddrList {
		addrs = append(addrs, localBroadcastAddrs(proto.DefaultPort)...)
	}
	if len(addrs) == 0 {
		return nil, fmt.Errorf("ca: no search addresses (set EPICS_CA_ADDR_LIST or enable AutoAddrList)")
	}

	if cfg.ClientName == "" {
		if u, err := user.Current(); err == nil && u.Username != "" {
			cfg.ClientName = u.Username
		} else if v := os.Getenv("USER"); v != "" {
			cfg.ClientName = v
		} else {
			cfg.ClientName = "user"
		}
	}
	if cfg.HostName == "" {
		cfg.HostName, _ = os.Hostname()
	}

	cctx, cancel := context.WithCancel(ctx)
	se := newSearchEngine(addrs)
	if err := se.start(cctx); err != nil {
		cancel()
		return nil, err
	}

	return &Client{
		cfg:      cfg,
		ctx:      cctx,
		cancel:   cancel,
		search:   se,
		circuits: make(map[string]*circuit),
	}, nil
}

// Close shuts down all circuits and background goroutines.
// Any in-flight Subscribe, Get, or Put calls will unblock with an error.
func (c *Client) Close() {
	c.cancel()
}

// -------------------------------------------------------------------------- //
// Internal helpers                                                             //
// -------------------------------------------------------------------------- //

// getOrCreateCircuit returns the circuit for addr, creating one if necessary.
func (c *Client) getOrCreateCircuit(addr string) *circuit {
	c.mu.Lock()
	defer c.mu.Unlock()
	if circ, ok := c.circuits[addr]; ok {
		return circ
	}
	circ := newCircuit(c.ctx, addr, c.cfg.ClientName, c.cfg.HostName)
	c.circuits[addr] = circ
	return circ
}

// resolve finds the IOC for pvName via UDP search, then waits for the TCP
// channel to be fully established.  Returns (circuit, chanState) on success.
func (c *Client) resolve(ctx context.Context, pvName string) (*circuit, *chanState, error) {
	addr, err := c.search.lookup(ctx, pvName)
	if err != nil {
		return nil, nil, err
	}
	circ := c.getOrCreateCircuit(addr)
	cs := circ.getOrCreateChannel(pvName)
	if err := cs.waitReady(ctx); err != nil {
		return nil, nil, fmt.Errorf("ca: %q: %w", pvName, err)
	}
	return circ, cs, nil
}

// -------------------------------------------------------------------------- //
// Public API                                                                  //
// -------------------------------------------------------------------------- //

// Subscribe registers ch to receive live monitor updates for pvName.
//
// Values are pushed as proto.TimeValue structs (DBR_TIME_* encoding).
// ch should be buffered; slow consumers will have updates silently dropped.
//
// The returned CancelFunc unsubscribes from the PV and must always be called
// to avoid leaking resources.
//
// Subscribe blocks until the channel is connected or ctx expires.
func (c *Client) Subscribe(ctx context.Context, pvName string, ch chan<- proto.TimeValue) (context.CancelFunc, error) {
	circ, cs, err := c.resolve(ctx, pvName)
	if err != nil {
		return nil, err
	}

	cs.mu.RLock()
	dbfType := cs.dbfType
	count := cs.count
	cs.mu.RUnlock()

	ms := &monState{
		subID:   circ.nextID(),
		dbrType: proto.NativeTimeType(dbfType, int(count)),
		count:   count,
		ch:      ch,
	}
	circ.addMonitor(cs, ms)

	return func() { circ.removeMonitor(cs, ms.subID) }, nil
}

// Get performs a one-shot READ_NOTIFY for pvName and returns its current value.
// It blocks until the reply arrives or ctx expires.
func (c *Client) Get(ctx context.Context, pvName string) (proto.TimeValue, error) {
	circ, cs, err := c.resolve(ctx, pvName)
	if err != nil {
		return proto.TimeValue{}, err
	}

	cs.mu.RLock()
	dbfType := cs.dbfType
	count := cs.count
	cs.mu.RUnlock()

	dbrType := proto.NativeTimeType(dbfType, int(count))
	return circ.get(ctx, cs, dbrType, count)
}

// Put writes value to pvName using CA_PROTO_WRITE (fire-and-forget).
//
// value is automatically encoded to match the PV's native field type.
// Supported Go types: float64, float32, int64, int32, int, int16, string, bool.
//
// Put blocks only until the message is queued; it does not wait for an IOC
// acknowledgement.  Use a timeout context to bound the wait for connection.
func (c *Client) Put(ctx context.Context, pvName string, value any) error {
	circ, cs, err := c.resolve(ctx, pvName)
	if err != nil {
		return err
	}

	cs.mu.RLock()
	dbfType := cs.dbfType
	cs.mu.RUnlock()

	dbrType, payload, err := encodePut(dbfType, value)
	if err != nil {
		return fmt.Errorf("ca: put %q: %w", pvName, err)
	}
	return circ.put(ctx, cs, dbrType, payload)
}

// -------------------------------------------------------------------------- //
// Control metadata                                                             //
// -------------------------------------------------------------------------- //

// CtrlInfo holds the full control-block metadata for a PV.
// Exactly one of the Double, Long, Enum, Str pointer fields is non-nil,
// depending on the PV's native field type.
type CtrlInfo struct {
	DBFType int    // proto.DBF* constant
	Count   uint32 // element count
	Access  uint32 // proto.AccessRead | proto.AccessWrite bitmask

	Double *proto.CtrlDouble // non-nil for DBFDouble / DBFFloat
	Long   *proto.CtrlLong   // non-nil for DBFLong / DBFShort / DBFChar
	Enum   *proto.CtrlEnum   // non-nil for DBFEnum
	Str    *proto.CtrlString // non-nil for DBFString
}

// GetCtrl performs a READ_NOTIFY with a DBR_CTRL_* type and returns the full
// control-block metadata (units, display limits, enum strings, etc.) for pvName.
//
// GetCtrl blocks until the reply arrives or ctx expires.
func (c *Client) GetCtrl(ctx context.Context, pvName string) (CtrlInfo, error) {
	circ, cs, err := c.resolve(ctx, pvName)
	if err != nil {
		return CtrlInfo{}, err
	}

	cs.mu.RLock()
	dbfType := cs.dbfType
	count := cs.count
	access := cs.access
	cs.mu.RUnlock()

	ctrlType := proto.NativeCtrlType(dbfType)
	_, payload, err := circ.getRaw(ctx, cs, ctrlType, count)
	if err != nil {
		return CtrlInfo{}, err
	}

	ci := CtrlInfo{DBFType: dbfType, Count: count, Access: access}

	switch ctrlType {
	case proto.DBRCtrlDouble:
		cd, ok := proto.DecodeCtrlDouble(payload)
		if !ok {
			return CtrlInfo{}, fmt.Errorf("ca: %q: DecodeCtrlDouble failed (payload len %d)", pvName, len(payload))
		}
		ci.Double = &cd

	case proto.DBRCtrlLong:
		cl, ok := proto.DecodeCtrlLong(payload)
		if !ok {
			return CtrlInfo{}, fmt.Errorf("ca: %q: DecodeCtrlLong failed (payload len %d)", pvName, len(payload))
		}
		ci.Long = &cl

	case proto.DBRCtrlEnum:
		ce, ok := proto.DecodeCtrlEnum(payload)
		if !ok {
			return CtrlInfo{}, fmt.Errorf("ca: %q: DecodeCtrlEnum failed (payload len %d)", pvName, len(payload))
		}
		ci.Enum = &ce

	case proto.DBRCtrlString:
		cs2, ok := proto.DecodeCtrlString(payload)
		if !ok {
			return CtrlInfo{}, fmt.Errorf("ca: %q: DecodeCtrlString failed (payload len %d)", pvName, len(payload))
		}
		ci.Str = &cs2

	default:
		return CtrlInfo{}, fmt.Errorf("ca: %q: unhandled ctrl type %d", pvName, ctrlType)
	}

	return ci, nil
}

// -------------------------------------------------------------------------- //
// Value encoding for Put                                                      //
// -------------------------------------------------------------------------- //

// encodePut converts a Go value to a DBR wire payload matching the PV's
// native field type.
func encodePut(dbfType int, value any) (dbrType uint16, payload []byte, err error) {
	switch dbfType {
	case proto.DBFDouble, proto.DBFFloat:
		f, e := toFloat64(value)
		if e != nil {
			return 0, nil, e
		}
		return proto.DBRDouble, proto.EncodeDouble(f), nil

	case proto.DBFLong:
		n, e := toInt32(value)
		if e != nil {
			return 0, nil, e
		}
		return proto.DBRLong, proto.EncodeLong(n), nil

	case proto.DBFShort, proto.DBFChar, proto.DBFEnum:
		n, e := toInt32(value)
		if e != nil {
			return 0, nil, e
		}
		return proto.DBRShort, proto.EncodeShort(int16(n)), nil

	case proto.DBFString:
		s, e := toString(value)
		if e != nil {
			return 0, nil, e
		}
		return proto.DBRString, proto.EncodeString(s), nil

	default:
		return 0, nil, fmt.Errorf("unsupported DBF type %d", dbfType)
	}
}

func toFloat64(v any) (float64, error) {
	switch x := v.(type) {
	case float64:
		return x, nil
	case float32:
		return float64(x), nil
	case int:
		return float64(x), nil
	case int64:
		return float64(x), nil
	case int32:
		return float64(x), nil
	case int16:
		return float64(x), nil
	case uint64:
		return float64(x), nil
	case uint32:
		return float64(x), nil
	case bool:
		if x {
			return 1, nil
		}
		return 0, nil
	default:
		return 0, fmt.Errorf("cannot convert %T to float64", v)
	}
}

func toInt32(v any) (int32, error) {
	switch x := v.(type) {
	case int:
		return int32(x), nil
	case int64:
		return int32(x), nil
	case int32:
		return x, nil
	case int16:
		return int32(x), nil
	case float64:
		return int32(x), nil
	case float32:
		return int32(x), nil
	case bool:
		if x {
			return 1, nil
		}
		return 0, nil
	default:
		return 0, fmt.Errorf("cannot convert %T to int32", v)
	}
}

func toString(v any) (string, error) {
	switch x := v.(type) {
	case string:
		return x, nil
	case []byte:
		return string(x), nil
	default:
		return fmt.Sprintf("%v", x), nil
	}
}