uopi/internal/datasource/epics/noop.go

//go:build !epics

// Package epics provides an EPICS Channel Access data source for uopi using a
// pure-Go CA implementation (no CGo, no EPICS Base installation required).
// When built WITH the "epics" build tag, the CGo-based implementation in
// epics.go is used instead; that version requires CGo and a working EPICS Base.
package epics

import (
	"context"
	"fmt"
	"log/slog" //nolint:depguard
	"strings"
	"sync"
	"time"

	ca "github.com/uopi/goca"
	"github.com/uopi/goca/proto"
	"github.com/uopi/uopi/internal/datasource"
)

// Available always returns true for the pure-Go build.
func Available() bool { return true }

// -------------------------------------------------------------------------- //
// EPICS data source (pure-Go implementation)                                  //
// -------------------------------------------------------------------------- //

// EPICS is the pure-Go Channel Access data source.
type EPICS struct {
	caAddrList     string
	archiveURL     string
	cfURL          string
	autoSyncFilter       string
	autoSyncFromArchiver bool
	pvNames              []string // pre-fetched at connect time for ListSignals

	client *ca.Client

	mu       sync.RWMutex
	metadata map[string]datasource.Metadata

	// Per-user CA clients for writes. EPICS Channel Access carries the client
	// username at the circuit (TCP) level, so attributing a write to a specific
	// end-user requires a dedicated client whose ClientName is that user. These
	// are created lazily on first write by each user and evicted when idle.
	ctx         context.Context // datasource lifetime; governs per-user clients
	baseCfg     ca.Config       // template config for per-user clients
	selfName    string          // the server's own CA client name
	userMu      sync.Mutex
	userClients map[string]*userClient
}

// userClient is a per-end-user CA client and its last-use time for idle eviction.
type userClient struct {
	cli      *ca.Client
	lastUsed time.Time
}

// userClientIdleTTL is how long an idle per-user CA client is kept before being
// closed. Writes are bursty, so a few minutes avoids reconnecting on every put
// without holding circuits open indefinitely.
const userClientIdleTTL = 10 * time.Minute

// New creates a new EPICS Channel Access data source.
//
// caAddrList is the EPICS_CA_ADDR_LIST value (space-separated addresses).
// Leave empty to use the environment variable.
// archiveURL is the base URL of an EPICS Archive Appliance instance; leave
// empty to disable history queries.
// pvNames is an optional list of PV names whose metadata will be pre-fetched
// at connect time so they appear in ListSignals immediately (useful for the
// edit-mode signal tree).
func New(caAddrList, archiveURL, cfURL, autoSyncFilter string, autoSyncFromArchiver bool, pvNames []string) datasource.DataSource {
	return &EPICS{
		caAddrList:           caAddrList,
		archiveURL:           archiveURL,
		cfURL:                cfURL,
		autoSyncFilter:       autoSyncFilter,
		autoSyncFromArchiver: autoSyncFromArchiver,
		pvNames:              pvNames,
		metadata:             make(map[string]datasource.Metadata),
	}
}

// Name implements datasource.DataSource.
func (e *EPICS) Name() string { return "epics" }

// Connect creates the CA client and starts background I/O goroutines.
// ctx governs the lifetime of the client; cancel it to shut everything down.
func (e *EPICS) Connect(ctx context.Context) error {
	cfg := ca.ConfigFromEnv()
	if e.caAddrList != "" {
		cfg.AddrList = strings.Fields(e.caAddrList)
		cfg.AutoAddrList = false
	}
	slog.Info("epics: connecting CA client", "addrs", cfg.AddrList, "auto_addr", cfg.AutoAddrList)
	cli, err := ca.NewClient(ctx, cfg)
	if err != nil {
		return fmt.Errorf("epics: %w", err)
	}
	e.client = cli
	e.ctx = ctx
	e.baseCfg = cfg
	e.selfName = cfg.ClientName
	e.userClients = make(map[string]*userClient)
	go e.evictIdleUserClients(ctx)
	slog.Info("epics: CA client started", "addrs", cfg.AddrList, "client_name", cfg.ClientName)

	// Perform background sync from Channel Finder if configured.
	if e.cfURL != "" && e.autoSyncFilter != "" {
		go e.syncFromChannelFinder(ctx)
	}

	// Perform background sync from Archive Appliance if configured.
	if e.archiveURL != "" && e.autoSyncFromArchiver {
		go e.syncFromArchiver(ctx)
	}

	// Pre-fetch metadata for any configured PV names.
	if len(e.pvNames) > 0 {
		go func() {
			for _, pv := range e.pvNames {
				if ctx.Err() != nil {
					return
				}
				e.prefetchPV(ctx, pv)
			}
		}()
	}

	return nil
}

func (e *EPICS) syncFromChannelFinder(ctx context.Context) {
	time.Sleep(500 * time.Millisecond)
	slog.Info("epics: syncing signals from Channel Finder", "url", e.cfURL, "filter", e.autoSyncFilter)
	channels, err := queryChannelFinder(e.cfURL, e.autoSyncFilter)
	if err != nil {
		slog.Error("epics: Channel Finder sync failed", "err", err)
		return
	}

	for _, ch := range channels {
		if ctx.Err() != nil {
			return
		}
		// Convert CFS properties and tags to datasource.Metadata
		props := make(map[string]string)
		for _, p := range ch.Properties {
			props[p.Name] = p.Value
		}
		tags := make([]string, len(ch.Tags))
		for i, t := range ch.Tags {
			tags[i] = t.Name
		}

		// Initialize metadata with CFS info. Full metadata (type, units, etc.)
		// will be populated on first access via DBR_CTRL GET.
		e.mu.Lock()
		if _, exists := e.metadata[ch.Name]; !exists {
			e.metadata[ch.Name] = datasource.Metadata{
				Name:       ch.Name,
				Properties: props,
				Tags:       tags,
			}
		}
		e.mu.Unlock()

		// Background pre-fetch of actual record info (type, etc.)
		go e.prefetchPV(ctx, ch.Name)
	}
	slog.Info("epics: synced signals from Channel Finder", "count", len(channels))
}

func (e *EPICS) prefetchPV(ctx context.Context, pv string) {
	tvCh := make(chan proto.TimeValue, 1)
	subCtx, subCancel := context.WithTimeout(ctx, 15*time.Second)
	defer subCancel()
	cancel, err := e.client.Subscribe(subCtx, pv, tvCh)
	if err != nil {
		return
	}
	defer cancel()
	// Channel is now connected; fetch CTRL metadata.
	mctx, mcancel := context.WithTimeout(ctx, 5*time.Second)
	defer mcancel()
	_, _ = e.GetMetadata(mctx, pv)
}

// -------------------------------------------------------------------------- //
// Subscribe                                                                    //
// -------------------------------------------------------------------------- //

// Subscribe registers ch to receive live value updates for signal.
// It blocks until the CA channel is connected (or ctx expires).
// A background goroutine forwards proto.TimeValue updates to ch as
// datasource.Value.  Call the returned CancelFunc to unsubscribe.
func (e *EPICS) Subscribe(ctx context.Context, signal string, ch chan<- datasource.Value) (datasource.CancelFunc, error) {
	slog.Info("epics: subscribing", "pv", signal)
	tvCh := make(chan proto.TimeValue, 16)

	clientCancel, err := e.client.Subscribe(ctx, signal, tvCh)
	if err != nil {
		slog.Error("epics: subscribe failed", "pv", signal, "err", err)
		return nil, fmt.Errorf("epics: subscribe %q: %w", signal, err)
	}
	slog.Info("epics: subscribed OK", "pv", signal)

	// Fetch and cache metadata now that the channel is connected.
	// Send MetaUpdate so the dispatcher re-broadcasts metadata to clients.
	go func() {
		// Small delay gives ACCESS_RIGHTS message time to arrive before GetCtrl.
		time.Sleep(50 * time.Millisecond)
		mctx, mcancel := context.WithTimeout(ctx, 5*time.Second)
		defer mcancel()
		if _, err := e.GetMetadata(mctx, signal); err == nil {
			select {
			case ch <- datasource.Value{MetaUpdate: true}:
			default:
			}
		}
	}()

	done := make(chan struct{})

	// Forward goroutine: convert proto.TimeValue → datasource.Value.
	go func() {
		for {
			select {
			case tv := <-tvCh:
				slog.Debug("epics: value received", "pv", signal, "double", tv.Double, "severity", tv.Severity)
				val := e.timeValueToDS(signal, tv)
				select {
				case ch <- val:
				default: // drop if consumer is slow
				}
			case <-done:
				return
			}
		}
	}()

	cancel := func() {
		clientCancel()
		close(done)
	}
	return datasource.CancelFunc(cancel), nil
}

// timeValueToDS converts a proto.TimeValue to a datasource.Value.
func (e *EPICS) timeValueToDS(signal string, tv proto.TimeValue) datasource.Value {
	var data any
	if tv.Doubles != nil {
		if len(tv.Doubles) == 1 {
			data = tv.Doubles[0]
		} else {
			data = tv.Doubles
		}
	} else if tv.Str != "" {
		data = tv.Str
	} else {
		data = tv.Double
	}

	quality := datasource.QualityGood
	switch tv.Severity {
	case proto.SeverityMinor:
		quality = datasource.QualityUncertain
	case proto.SeverityMajor, proto.SeverityInvalid:
		quality = datasource.QualityBad
	}

	return datasource.Value{
		Timestamp: tv.Timestamp,
		Data:      data,
		Quality:   quality,
	}
}

// -------------------------------------------------------------------------- //
// GetMetadata                                                                  //
// -------------------------------------------------------------------------- //

// GetMetadata returns metadata for signal, using a cached value when available.
// On the first call for a signal it performs a DBR_CTRL_* GET to retrieve
// units, display limits, enum strings, and access rights.
func (e *EPICS) GetMetadata(ctx context.Context, signal string) (datasource.Metadata, error) {
	e.mu.RLock()
	if m, ok := e.metadata[signal]; ok {
		e.mu.RUnlock()
		return m, nil
	}
	e.mu.RUnlock()

	slog.Info("epics: fetching metadata", "pv", signal)
	ci, err := e.client.GetCtrl(ctx, signal)
	if err != nil {
		slog.Error("epics: GetMetadata failed", "pv", signal, "err", err)
		return datasource.Metadata{}, fmt.Errorf("epics: GetMetadata %q: %w", signal, err)
	}

	m := e.ctrlToMeta(signal, ci)
	slog.Info("epics: metadata fetched", "pv", signal, "type", m.Type, "unit", m.Unit, "writable", m.Writable)

	e.mu.Lock()
	e.metadata[signal] = m
	e.mu.Unlock()

	return m, nil
}

// ctrlToMeta converts a ca.CtrlInfo to a datasource.Metadata.
func (e *EPICS) ctrlToMeta(name string, ci ca.CtrlInfo) datasource.Metadata {
	m := datasource.Metadata{
		Name:     name,
		Writable: ci.Access&proto.AccessWrite != 0,
	}

	switch {
	case ci.Double != nil:
		m.Type = datasource.TypeFloat64
		if ci.Count > 1 {
			m.Type = datasource.TypeFloat64Array
		}
		m.Unit = ci.Double.Units
		m.DisplayLow = ci.Double.LowerDispLimit
		m.DisplayHigh = ci.Double.UpperDispLimit
		m.DriveLow = ci.Double.LowerCtrlLimit
		m.DriveHigh = ci.Double.UpperCtrlLimit

	case ci.Long != nil:
		m.Type = datasource.TypeInt64
		m.Unit = ci.Long.Units
		m.DisplayLow = float64(ci.Long.LowerDispLimit)
		m.DisplayHigh = float64(ci.Long.UpperDispLimit)
		m.DriveLow = float64(ci.Long.LowerCtrlLimit)
		m.DriveHigh = float64(ci.Long.UpperCtrlLimit)

	case ci.Enum != nil:
		m.Type = datasource.TypeEnum
		m.EnumStrings = ci.Enum.Strings

	case ci.Str != nil:
		m.Type = datasource.TypeString
	}

	return m
}

// -------------------------------------------------------------------------- //
// ListSignals                                                                  //
// -------------------------------------------------------------------------- //

// ListSignals returns metadata for all signals with cached information.
func (e *EPICS) ListSignals(_ context.Context) ([]datasource.Metadata, error) {
	e.mu.RLock()
	defer e.mu.RUnlock()
	slog.Debug("epics: listing signals", "count", len(e.metadata))
	out := make([]datasource.Metadata, 0, len(e.metadata))
	for _, m := range e.metadata {
		out = append(out, m)
	}
	return out, nil
}

// -------------------------------------------------------------------------- //
// Write                                                                        //
// -------------------------------------------------------------------------- //

// Write puts a new value onto the named CA channel.
// value must be one of: float64, float32, int64, int32, int, int16, string, bool.
//
// When the request context carries an end-user identity (see datasource.WithUser)
// that differs from the server's own CA client name, the put is performed on a
// dedicated CA client whose ClientName is that user, so the IOC's access-security
// rules see the actual end-user rather than the server process.
func (e *EPICS) Write(ctx context.Context, signal string, value any) error {
	cli := e.client
	if user, ok := datasource.UserFrom(ctx); ok && user != e.selfName {
		uc, err := e.clientForUser(user)
		if err != nil {
			return fmt.Errorf("epics: write %q as %q: %w", signal, user, err)
		}
		cli = uc
	}
	if err := cli.Put(ctx, signal, value); err != nil {
		return fmt.Errorf("epics: write %q: %w", signal, err)
	}
	return nil
}

// clientForUser returns a CA client whose ClientName is user, creating and
// caching one on first use. The client is bound to the datasource lifetime ctx,
// not the per-request ctx, so it survives across writes until evicted when idle.
func (e *EPICS) clientForUser(user string) (*ca.Client, error) {
	e.userMu.Lock()
	defer e.userMu.Unlock()
	if uc, ok := e.userClients[user]; ok {
		uc.lastUsed = time.Now()
		return uc.cli, nil
	}
	cfg := e.baseCfg
	cfg.ClientName = user
	cli, err := ca.NewClient(e.ctx, cfg)
	if err != nil {
		return nil, err
	}
	e.userClients[user] = &userClient{cli: cli, lastUsed: time.Now()}
	slog.Info("epics: created per-user CA client", "user", user)
	return cli, nil
}

// evictIdleUserClients closes per-user CA clients that have not been used within
// userClientIdleTTL. It runs until ctx is cancelled.
func (e *EPICS) evictIdleUserClients(ctx context.Context) {
	t := time.NewTicker(time.Minute)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case now := <-t.C:
			e.userMu.Lock()
			for user, uc := range e.userClients {
				if now.Sub(uc.lastUsed) > userClientIdleTTL {
					uc.cli.Close()
					delete(e.userClients, user)
					slog.Info("epics: evicted idle per-user CA client", "user", user)
				}
			}
			e.userMu.Unlock()
		}
	}
}

// -------------------------------------------------------------------------- //
// History                                                                      //
// -------------------------------------------------------------------------- //

// History delegates to the Archive Appliance client, or returns
// ErrHistoryUnavailable if no archive URL is configured.
func (e *EPICS) History(ctx context.Context, signal string, start, end time.Time, maxPoints int) ([]datasource.Value, error) {
	if e.archiveURL == "" {
		return nil, datasource.ErrHistoryUnavailable
	}
	return fetchArchiveHistory(ctx, e.archiveURL, signal, start, end, maxPoints)
}

func (e *EPICS) syncFromArchiver(ctx context.Context) {
	time.Sleep(1 * time.Second) // wait for network
	slog.Info("epics: syncing signals from Archiver", "url", e.archiveURL)
	pvs, err := searchArchivePVs(ctx, e.archiveURL, "*")
	if err != nil {
		slog.Error("epics: Archiver sync failed", "err", err)
		return
	}

	for _, name := range pvs {
		if ctx.Err() != nil {
			return
		}
		e.mu.Lock()
		if _, exists := e.metadata[name]; !exists {
			e.metadata[name] = datasource.Metadata{Name: name}
		}
		e.mu.Unlock()
		// Background pre-fetch of record info
		go e.prefetchPV(ctx, name)
	}
	slog.Info("epics: synced signals from Archiver", "count", len(pvs))
}