uopi/pkg/ca/conn.go

package ca

import (
	"context"
	"fmt"
	"io"
	"net"
	"sync"
	"sync/atomic"
	"time"

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

const (
	writeQueueDepth = 256
	echoInterval    = 15 * time.Second
	connectTimeout  = 5 * time.Second
	maxReconnDelay  = 60 * time.Second
)

// -------------------------------------------------------------------------- //
// reply — async GET callback payload                                          //
// -------------------------------------------------------------------------- //

// reply is delivered to a pending GET waiter via a buffered channel.
// If ok is false the circuit died before the reply arrived.
type reply struct {
	hdr     proto.Header
	payload []byte
	ok      bool
}

// -------------------------------------------------------------------------- //
// monState — one active EVENT_ADD subscription                               //
// -------------------------------------------------------------------------- //

// monState survives reconnections; the circuit re-sends EVENT_ADD with the
// new SID each time a CREATE_CHAN reply is received.
type monState struct {
	subID   uint32
	dbrType uint16
	count   uint32
	ch      chan<- proto.TimeValue // caller-owned, never closed here
}

// -------------------------------------------------------------------------- //
// chanState — one CA channel (persists across reconnections)                 //
// -------------------------------------------------------------------------- //

// chanState field invariants:
//   - cid, pvName: immutable after construction.
//   - sid, dbfType, count, access: valid only while readyC is closed.
//   - readyC: closed when CREATE_CHAN reply received; replaced on reconnect.
//   - monitors: append-only (except removeMonitor); never cleared on reconnect.
//   - pending: ioid → GET reply channel; drained (channels closed) on reconnect.
type chanState struct {
	cid    uint32
	pvName string

	mu       sync.RWMutex
	sid      uint32         // server-assigned channel ID (0 until CREATE_CHAN reply)
	dbfType  int            // native DBF field type
	count    uint32         // element count
	access   uint32         // AccessRead | AccessWrite bitmask
	readyC   chan struct{}   // closed once CREATE_CHAN reply is received
	monitors []*monState    // active subscriptions
	pending  map[uint32]chan reply // ioid → one-shot GET callback
}

func newChanState(cid uint32, pvName string) *chanState {
	return &chanState{
		cid:    cid,
		pvName: pvName,
		readyC: make(chan struct{}),
		pending: make(map[uint32]chan reply),
	}
}

// resetForReconnect prepares cs for a new TCP connection.
// It closes the current readyC (waking any waitReady callers so they can
// re-wait on the fresh channel) and then replaces it.
// Must be called before the circuit sends CREATE_CHAN on the new conn.
func (cs *chanState) resetForReconnect() {
	cs.mu.Lock()
	cs.sid = 0
	old := cs.readyC
	cs.readyC = make(chan struct{})
	// Drain pending GETs — they will receive zero reply (ok=false).
	for id, ch := range cs.pending {
		close(ch)
		delete(cs.pending, id)
	}
	cs.mu.Unlock()
	// Close the old readyC outside the lock to avoid deadlock with waitReady.
	select {
	case <-old:
		// Already closed (CmdCreateFail path or double-reset guard).
	default:
		close(old)
	}
}

// waitReady blocks until sid != 0 (CREATE_CHAN reply received for the current
// connection) or ctx expires.  It loops through reconnections automatically:
// when resetForReconnect closes the current readyC the goroutine wakes, sees
// sid == 0, and waits on the freshly created channel.
func (cs *chanState) waitReady(ctx context.Context) error {
	for {
		cs.mu.RLock()
		sid := cs.sid
		ready := cs.readyC
		cs.mu.RUnlock()

		if sid != 0 {
			return nil
		}

		select {
		case <-ready:
			// State changed — either CREATE_CHAN reply (sid set) or reconnect
			// started (sid cleared, new readyC installed).  Loop to check.
		case <-ctx.Done():
			return ctx.Err()
		}
	}
}

// -------------------------------------------------------------------------- //
// circuit — persistent TCP connection to one CA server                       //
// -------------------------------------------------------------------------- //

// circuit manages a persistent, auto-reconnecting TCP connection to a single
// CA server.  All channels and monitors created on a circuit survive
// reconnections; the circuit re-creates them transparently.
//
// Goroutine model:
//   - run() goroutine: reconnect loop; drives the write loop for each conn.
//   - readLoop() goroutine: one per active conn; dispatches inbound messages.
//
// Lock ordering: circuit.mu → chanState.mu (never the reverse).
type circuit struct {
	addr       string
	clientName string
	hostName   string

	ctx    context.Context
	cancel context.CancelFunc

	mu       sync.RWMutex
	channels []*chanState          // append-only; survive reconnect
	byCID    map[uint32]*chanState // cid → chanState (fast lookup)
	bySubID  map[uint32]*monState  // subID → monState (fast event dispatch)

	writeQ chan []byte    // serialised outbound message queue
	seq    atomic.Uint32 // shared ID sequence (cid, ioid, subID)
}

func newCircuit(ctx context.Context, addr, clientName, hostName string) *circuit {
	cctx, cancel := context.WithCancel(ctx)
	c := &circuit{
		addr:       addr,
		clientName: clientName,
		hostName:   hostName,
		ctx:        cctx,
		cancel:     cancel,
		byCID:      make(map[uint32]*chanState),
		bySubID:    make(map[uint32]*monState),
		writeQ:     make(chan []byte, writeQueueDepth),
	}
	go c.run()
	return c
}

func (c *circuit) close() { c.cancel() }

func (c *circuit) nextID() uint32 { return c.seq.Add(1) }

// -------------------------------------------------------------------------- //
// Reconnect loop                                                               //
// -------------------------------------------------------------------------- //

func (c *circuit) run() {
	delay := time.Second
	for {
		if c.ctx.Err() != nil {
			return
		}

		dbg("CA circuit connecting", "addr", c.addr)
		conn, err := c.dial()
		if err != nil {
			dbg("CA circuit dial failed", "addr", c.addr, "err", err)
			select {
			case <-time.After(delay):
			case <-c.ctx.Done():
				return
			}
			delay = min(delay*2, maxReconnDelay)
			continue
		}
		dbg("CA circuit connected", "addr", c.addr)
		delay = time.Second // reset back-off on successful connect

		// Snapshot channels and reset their per-connection state.
		c.mu.Lock()
		for _, cs := range c.channels {
			cs.resetForReconnect()
		}
		chs := make([]*chanState, len(c.channels))
		copy(chs, c.channels)
		c.mu.Unlock()

		// Drain stale messages from the write queue (they carry old SIDs).
		for len(c.writeQ) > 0 {
			<-c.writeQ
		}

		// Send VERSION + HOST_NAME + CLIENT_NAME.
		if err := c.sendHandshake(conn); err != nil {
			conn.Close()
			continue
		}

		// Re-create all known channels on the new connection.
		setupOK := true
		for _, cs := range chs {
			if _, err := conn.Write(buildCreateChanMsg(cs.cid, cs.pvName)); err != nil {
				setupOK = false
				break
			}
		}
		if !setupOK {
			conn.Close()
			continue
		}

		dbg("CA handshake sent", "addr", c.addr, "channels", len(chs))

		// Start read loop.
		readDone := make(chan struct{})
		go func() {
			c.readLoop(conn)
			dbg("CA read loop exited", "addr", c.addr)
			close(readDone)
		}()

		// Write loop (run goroutine acts as the write pump).
		c.writeLoop(conn, readDone)

		conn.Close()
		<-readDone // drain read goroutine before reconnecting

		if c.ctx.Err() != nil {
			return
		}

		select {
		case <-time.After(delay):
		case <-c.ctx.Done():
			return
		}
		delay = min(delay*2, maxReconnDelay)
	}
}

func (c *circuit) writeLoop(conn net.Conn, readDone <-chan struct{}) {
	ticker := time.NewTicker(echoInterval)
	defer ticker.Stop()
	for {
		select {
		case msg := <-c.writeQ:
			if _, err := conn.Write(msg); err != nil {
				return
			}
		case <-ticker.C:
			// Periodic heartbeat; server echoes it back.
			hb := proto.BuildMessage(proto.Header{Command: proto.CmdEcho}, nil)
			if _, err := conn.Write(hb); err != nil {
				return
			}
		case <-readDone:
			return
		case <-c.ctx.Done():
			return
		}
	}
}

// -------------------------------------------------------------------------- //
// Dial + handshake                                                             //
// -------------------------------------------------------------------------- //

func (c *circuit) dial() (net.Conn, error) {
	d := net.Dialer{Timeout: connectTimeout}
	conn, err := d.DialContext(c.ctx, "tcp4", c.addr)
	if err != nil {
		return nil, fmt.Errorf("ca: dial %s: %w", c.addr, err)
	}
	return conn, nil
}

func (c *circuit) sendHandshake(conn net.Conn) error {
	ver := proto.BuildMessage(proto.Header{
		Command:   proto.CmdVersion,
		DataCount: proto.MinorVersion,
	}, nil)
	host := proto.BuildMessage(
		proto.Header{Command: proto.CmdHostName},
		proto.BuildStringPayload(c.hostName),
	)
	client := proto.BuildMessage(
		proto.Header{Command: proto.CmdClientName},
		proto.BuildStringPayload(c.clientName),
	)
	out := make([]byte, 0, len(ver)+len(host)+len(client))
	out = append(out, ver...)
	out = append(out, host...)
	out = append(out, client...)
	_, err := conn.Write(out)
	return err
}

// -------------------------------------------------------------------------- //
// Read loop + message dispatch                                                 //
// -------------------------------------------------------------------------- //

func (c *circuit) readLoop(conn net.Conn) {
	for {
		hdr, _, err := proto.DecodeHeader(conn)
		if err != nil {
			if c.ctx.Err() == nil {
				dbg("CA read loop error (header)", "addr", c.addr, "err", err)
			}
			return
		}
		dbg("CA recv", "addr", c.addr, "cmd", hdr.Command, "dataType", hdr.DataType,
			"dataCount", hdr.DataCount, "payloadSize", hdr.PayloadSize,
			"p1", hdr.Parameter1, "p2", hdr.Parameter2)
		var payload []byte
		if hdr.PayloadSize > 0 {
			payload = make([]byte, hdr.PayloadSize)
			if _, err = io.ReadFull(conn, payload); err != nil {
				if c.ctx.Err() == nil {
					dbg("CA read loop error (payload)", "addr", c.addr,
						"cmd", hdr.Command, "wantBytes", hdr.PayloadSize, "err", err)
				}
				return
			}
		}
		c.dispatch(conn, hdr, payload)
	}
}

// dispatch handles one inbound CA message.
func (c *circuit) dispatch(conn net.Conn, hdr proto.Header, payload []byte) {
	switch hdr.Command {

	case proto.CmdVersion:
		// Server version negotiation reply — nothing to do.

	case proto.CmdCreateChan:
		// Parameter1 = cid (echoed), Parameter2 = SID assigned by server.
		c.mu.RLock()
		cs, ok := c.byCID[hdr.Parameter1]
		c.mu.RUnlock()
		if !ok {
			return
		}
		cs.mu.Lock()
		cs.sid = hdr.Parameter2
		cs.dbfType = int(hdr.DataType)
		cs.count = hdr.DataCount
		ready := cs.readyC
		// Snapshot monitors for EVENT_ADD re-subscription.
		mons := make([]*monState, len(cs.monitors))
		copy(mons, cs.monitors)
		cs.mu.Unlock()

		dbg("CA CREATE_CHAN reply", "pv", cs.pvName, "sid", hdr.Parameter2,
			"dbfType", hdr.DataType, "count", hdr.DataCount, "monitors", len(mons))

		// Re-subscribe all monitors with the new SID.
		for _, ms := range mons {
			msg := buildEventAddMsg(ms.subID, hdr.Parameter2, ms.dbrType, ms.count)
			dbg("CA EVENT_ADD sent", "pv", cs.pvName, "subID", ms.subID,
				"sid", hdr.Parameter2, "dbrType", ms.dbrType, "count", ms.count)
			select {
			case c.writeQ <- msg:
			default:
			}
		}
		close(ready) // unblock waitReady callers

	case proto.CmdCreateFail:
		// Server does not host this PV (or quota exceeded).
		c.mu.RLock()
		cs, ok := c.byCID[hdr.Parameter1]
		c.mu.RUnlock()
		if !ok {
			return
		}
		dbg("CA CREATE_FAIL", "pv", cs.pvName, "cid", hdr.Parameter1)
		cs.mu.RLock()
		ready := cs.readyC
		cs.mu.RUnlock()
		// readyC is only closed once per connection cycle by this goroutine,
		// so the select here is safe (no concurrent close).
		select {
		case <-ready:
			// Already closed (shouldn't happen, but guard against it).
		default:
			close(ready)
		}

	case proto.CmdAccessRights:
		// Parameter1 = cid, Parameter2 = access bitmask.
		c.mu.RLock()
		cs, ok := c.byCID[hdr.Parameter1]
		c.mu.RUnlock()
		if ok {
			cs.mu.Lock()
			cs.access = hdr.Parameter2
			cs.mu.Unlock()
		}

	case proto.CmdEventAdd:
		// Monitor update: Parameter1 = subscriptionID.
		c.mu.RLock()
		ms, ok := c.bySubID[hdr.Parameter1]
		c.mu.RUnlock()
		if !ok {
			dbg("CA EVENT_ADD for unknown subID", "subID", hdr.Parameter1)
			return
		}
		tv, ok := proto.DecodeTimeValue(hdr.DataType, hdr.DataCount, payload)
		if !ok {
			dbg("CA EVENT_ADD decode failed", "subID", hdr.Parameter1,
				"dbrType", hdr.DataType, "count", hdr.DataCount, "payloadLen", len(payload))
			return
		}
		dbg("CA EVENT_ADD value", "subID", hdr.Parameter1, "dbrType", hdr.DataType,
			"double", tv.Double, "severity", tv.Severity)
		select {
		case ms.ch <- tv:
		default: // drop if consumer is slow
		}

	case proto.CmdReadNotify:
		// GET reply: Parameter1 = ioid.
		ioid := hdr.Parameter1
		c.mu.RLock()
		var replyCh chan reply
		var found bool
		for _, cs := range c.channels {
			cs.mu.Lock()
			if ch, exists := cs.pending[ioid]; exists {
				delete(cs.pending, ioid)
				replyCh = ch
				found = true
			}
			cs.mu.Unlock()
			if found {
				break
			}
		}
		c.mu.RUnlock()
		if found {
			select {
			case replyCh <- reply{hdr: hdr, payload: payload, ok: true}:
			default:
			}
		}

	case proto.CmdEcho:
		// Server heartbeat — no response needed (client sends its own echoes).

	case proto.CmdServerDisc:
		// Server is shutting down — close conn to trigger reconnect.
		conn.Close()
	}
}

// -------------------------------------------------------------------------- //
// Message builders                                                             //
// -------------------------------------------------------------------------- //

func buildCreateChanMsg(cid uint32, pvName string) []byte {
	return proto.BuildMessage(proto.Header{
		Command:    proto.CmdCreateChan,
		Parameter1: cid,
		Parameter2: proto.MinorVersion,
	}, proto.BuildStringPayload(pvName))
}

func buildEventAddMsg(subID, sid uint32, dbrType uint16, count uint32) []byte {
	return proto.BuildMessage(proto.Header{
		Command:    proto.CmdEventAdd,
		DataType:   dbrType,
		DataCount:  count,
		Parameter1: sid,   // m_cid = channel SID (per CA spec)
		Parameter2: subID, // m_available = client subscription ID
	}, proto.EncodeEventMask(proto.DBEDefault))
}

func buildEventCancelMsg(subID, sid uint32, dbrType uint16) []byte {
	return proto.BuildMessage(proto.Header{
		Command:    proto.CmdEventCancel,
		DataType:   dbrType,
		Parameter1: sid,   // m_cid = channel SID
		Parameter2: subID, // m_available = client subscription ID
	}, nil)
}

// -------------------------------------------------------------------------- //
// Channel and monitor management                                               //
// -------------------------------------------------------------------------- //

// getOrCreateChannel returns the chanState for pvName, creating it if needed.
// If the circuit is currently connected, CREATE_CHAN is queued immediately;
// otherwise the reconnect loop will send it when it next connects.
func (c *circuit) getOrCreateChannel(pvName string) *chanState {
	c.mu.Lock()
	defer c.mu.Unlock()
	for _, cs := range c.channels {
		if cs.pvName == pvName {
			return cs
		}
	}
	cid := c.nextID()
	cs := newChanState(cid, pvName)
	c.channels = append(c.channels, cs)
	c.byCID[cid] = cs
	// Best-effort: deliver CREATE_CHAN if circuit is up.
	msg := buildCreateChanMsg(cid, pvName)
	select {
	case c.writeQ <- msg:
	default:
	}
	return cs
}

// addMonitor registers ms on cs and queues EVENT_ADD if the channel is ready.
// If the channel is not yet ready the CREATE_CHAN reply handler will send
// EVENT_ADD when the connection is established.
func (c *circuit) addMonitor(cs *chanState, ms *monState) {
	c.mu.Lock()
	c.bySubID[ms.subID] = ms
	c.mu.Unlock()

	cs.mu.Lock()
	cs.monitors = append(cs.monitors, ms)
	sid := cs.sid
	cs.mu.Unlock()

	if sid != 0 {
		dbg("CA EVENT_ADD sent (addMonitor)", "pv", cs.pvName, "subID", ms.subID,
			"sid", sid, "dbrType", ms.dbrType, "count", ms.count)
		msg := buildEventAddMsg(ms.subID, sid, ms.dbrType, ms.count)
		select {
		case c.writeQ <- msg:
		default:
		}
	}
}

// removeMonitor unregisters a monitor and queues EVENT_CANCEL if connected.
func (c *circuit) removeMonitor(cs *chanState, subID uint32) {
	c.mu.Lock()
	delete(c.bySubID, subID)
	c.mu.Unlock()

	cs.mu.Lock()
	var dbrType uint16
	remaining := cs.monitors[:0:0] // fresh backing array
	for _, ms := range cs.monitors {
		if ms.subID == subID {
			dbrType = ms.dbrType
		} else {
			remaining = append(remaining, ms)
		}
	}
	cs.monitors = remaining
	sid := cs.sid
	cs.mu.Unlock()

	if sid != 0 {
		msg := buildEventCancelMsg(subID, sid, dbrType)
		select {
		case c.writeQ <- msg:
		default:
		}
	}
}

// -------------------------------------------------------------------------- //
// Channel operations: get and put                                              //
// -------------------------------------------------------------------------- //

// getRaw performs a READ_NOTIFY and returns the undecoded header + payload.
// Use this when you need to decode a type that get() does not handle
// (e.g. DBR_CTRL_* for metadata retrieval).
func (c *circuit) getRaw(ctx context.Context, cs *chanState, dbrType uint16, count uint32) (proto.Header, []byte, error) {
	if err := cs.waitReady(ctx); err != nil {
		return proto.Header{}, nil, fmt.Errorf("ca: %q: %w", cs.pvName, err)
	}

	cs.mu.RLock()
	sid := cs.sid
	cs.mu.RUnlock()
	if sid == 0 {
		return proto.Header{}, nil, fmt.Errorf("ca: %q: channel not connected", cs.pvName)
	}

	ioid := c.nextID()
	replyCh := make(chan reply, 1)

	cs.mu.Lock()
	cs.pending[ioid] = replyCh
	cs.mu.Unlock()

	msg := proto.BuildMessage(proto.Header{
		Command:    proto.CmdReadNotify,
		DataType:   dbrType,
		DataCount:  count,
		Parameter1: sid,  // m_cid = channel SID (per CA spec)
		Parameter2: ioid, // m_available = request IOId
	}, nil)

	select {
	case c.writeQ <- msg:
	case <-ctx.Done():
		cs.mu.Lock()
		delete(cs.pending, ioid)
		cs.mu.Unlock()
		return proto.Header{}, nil, ctx.Err()
	}

	select {
	case r := <-replyCh:
		if !r.ok {
			return proto.Header{}, nil, fmt.Errorf("ca: %q: circuit disconnected", cs.pvName)
		}
		return r.hdr, r.payload, nil
	case <-ctx.Done():
		cs.mu.Lock()
		delete(cs.pending, ioid)
		cs.mu.Unlock()
		return proto.Header{}, nil, ctx.Err()
	}
}

// get performs a READ_NOTIFY (async GET) and returns the decoded value.
// It blocks until the reply arrives, the channel is not yet ready, or ctx expires.
func (c *circuit) get(ctx context.Context, cs *chanState, dbrType uint16, count uint32) (proto.TimeValue, error) {
	if err := cs.waitReady(ctx); err != nil {
		return proto.TimeValue{}, fmt.Errorf("ca: %q: %w", cs.pvName, err)
	}

	cs.mu.RLock()
	sid := cs.sid
	cs.mu.RUnlock()
	if sid == 0 {
		return proto.TimeValue{}, fmt.Errorf("ca: %q: channel not connected", cs.pvName)
	}

	ioid := c.nextID()
	replyCh := make(chan reply, 1)

	cs.mu.Lock()
	cs.pending[ioid] = replyCh
	cs.mu.Unlock()

	msg := proto.BuildMessage(proto.Header{
		Command:    proto.CmdReadNotify,
		DataType:   dbrType,
		DataCount:  count,
		Parameter1: sid,  // m_cid = channel SID (per CA spec)
		Parameter2: ioid, // m_available = request IOId
	}, nil)

	select {
	case c.writeQ <- msg:
	case <-ctx.Done():
		cs.mu.Lock()
		delete(cs.pending, ioid)
		cs.mu.Unlock()
		return proto.TimeValue{}, ctx.Err()
	}

	select {
	case r := <-replyCh:
		if !r.ok {
			return proto.TimeValue{}, fmt.Errorf("ca: %q: circuit disconnected during GET", cs.pvName)
		}
		tv, ok := proto.DecodeTimeValue(r.hdr.DataType, r.hdr.DataCount, r.payload)
		if !ok {
			return proto.TimeValue{}, fmt.Errorf("ca: %q: failed to decode GET reply", cs.pvName)
		}
		return tv, nil
	case <-ctx.Done():
		cs.mu.Lock()
		delete(cs.pending, ioid)
		cs.mu.Unlock()
		return proto.TimeValue{}, ctx.Err()
	}
}

// put queues a CA_PROTO_WRITE (fire-and-forget put).
// It waits for the channel to be ready and access rights to be confirmed.
func (c *circuit) put(ctx context.Context, cs *chanState, dbrType uint16, payload []byte) error {
	if err := cs.waitReady(ctx); err != nil {
		return fmt.Errorf("ca: %q: %w", cs.pvName, err)
	}

	cs.mu.RLock()
	sid := cs.sid
	access := cs.access
	cs.mu.RUnlock()

	if sid == 0 {
		return fmt.Errorf("ca: %q: channel not connected", cs.pvName)
	}
	if access&proto.AccessWrite == 0 {
		return fmt.Errorf("ca: %q: read-only", cs.pvName)
	}

	// CA_PROTO_WRITE: DataType=dbrType, DataCount=1, Parameter1=SID, Parameter2=0 (no ioid).
	msg := proto.BuildMessage(proto.Header{
		Command:    proto.CmdWrite,
		DataType:   dbrType,
		DataCount:  1,
		Parameter1: sid,
		Parameter2: 0,
	}, payload)

	select {
	case c.writeQ <- msg:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}