uopi/internal/datasource/synthetic/synthetic.go

package synthetic

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"os"
	"path/filepath"
	"sync"
	"time"

	"github.com/uopi/uopi/internal/broker"
	"github.com/uopi/uopi/internal/datasource"
	"github.com/uopi/uopi/internal/dsp"
)

const definitionsFile = "synthetic.json"

// signalState holds everything needed to run one synthetic signal.
type signalState struct {
	def    SignalDef
	nodes  []dsp.Node
	states []map[string]any // one map per node, persistent across calls

	// cancel stops the goroutine driving this signal.
	cancel context.CancelFunc
}

// Synthetic is a DataSource that computes values from upstream signals via
// configurable DSP pipelines.
type Synthetic struct {
	brk       *broker.Broker
	storePath string
	log       *slog.Logger

	mu      sync.RWMutex
	signals map[string]*signalState

	// root context provided by Connect; used to start per-signal goroutines.
	ctx context.Context
}

// New creates a Synthetic data source. storePath is the directory where
// synthetic.json is stored. brk is used to subscribe to upstream signals.
func New(storePath string, brk *broker.Broker, log *slog.Logger) *Synthetic {
	return &Synthetic{
		brk:       brk,
		storePath: storePath,
		log:       log,
		signals:   make(map[string]*signalState),
	}
}

// Name implements datasource.DataSource.
func (s *Synthetic) Name() string { return "synthetic" }

// Connect loads the definitions file and starts all signal goroutines.
func (s *Synthetic) Connect(ctx context.Context) error {
	s.ctx = ctx

	defs, err := s.loadDefs()
	if err != nil {
		return fmt.Errorf("synthetic: load definitions: %w", err)
	}

	for _, def := range defs {
		if err := s.startSignal(def); err != nil {
			s.log.Warn("synthetic: failed to start signal", "name", def.Name, "err", err)
		}
	}
	return nil
}

// ListSignals returns metadata for all defined synthetic signals.
func (s *Synthetic) ListSignals(_ context.Context) ([]datasource.Metadata, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()

	out := make([]datasource.Metadata, 0, len(s.signals))
	for _, st := range s.signals {
		out = append(out, defToMetadata(st.def))
	}
	return out, nil
}

// FilteredMetadata returns metadata for every defined synthetic signal for
// which keep returns true. It lets the API layer apply per-caller visibility
// rules (which depend on SignalDef fields not present in datasource.Metadata).
func (s *Synthetic) FilteredMetadata(keep func(SignalDef) bool) []datasource.Metadata {
	s.mu.RLock()
	defer s.mu.RUnlock()

	out := make([]datasource.Metadata, 0, len(s.signals))
	for _, st := range s.signals {
		if keep(st.def) {
			out = append(out, defToMetadata(st.def))
		}
	}
	return out
}

// GetMetadata returns metadata for a single named signal.
func (s *Synthetic) GetMetadata(_ context.Context, signal string) (datasource.Metadata, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()

	st, ok := s.signals[signal]
	if !ok {
		return datasource.Metadata{}, datasource.ErrNotFound
	}
	return defToMetadata(st.def), nil
}

// Subscribe registers ch to receive computed values for the named signal.
// Values are pushed whenever an upstream signal changes.
func (s *Synthetic) Subscribe(ctx context.Context, signal string, ch chan<- datasource.Value) (datasource.CancelFunc, error) {
	s.mu.RLock()
	st, ok := s.signals[signal]
	s.mu.RUnlock()
	if !ok {
		return nil, datasource.ErrNotFound
	}

	// Collect the upstream references for this signal.
	refs := upstreamRefs(st.def)
	if len(refs) == 0 {
		return nil, fmt.Errorf("synthetic: signal %q has no upstream inputs", signal)
	}

	ctx, cancel := context.WithCancel(ctx)

	go func() {
		defer cancel()

		// Latest value per upstream input index.
		latest := make([]float64, len(refs))
		ready := make([]bool, len(refs))

		// Subscribe to every upstream ref via the broker.
		updateCh := make(chan indexedUpdate, 32*len(refs))

		unsubs := make([]func(), 0, len(refs))
		for i, ref := range refs {
			idx := i // capture
			perCh := make(chan broker.Update, 16)
			unsub, err := s.brk.Subscribe(ref, perCh)
			if err != nil {
				s.log.Warn("synthetic: upstream subscribe failed",
					"signal", signal, "upstream", ref, "err", err)
				// Continue; the slot will stay at zero.
			} else {
				unsubs = append(unsubs, unsub)
				go func() {
					for {
						select {
						case u, ok := <-perCh:
							if !ok {
								return
							}
							val := toFloat64(u.Value.Data)
							select {
							case updateCh <- indexedUpdate{idx: idx, val: val, ts: u.Value.Timestamp}:
							default:
							}
						case <-ctx.Done():
							return
						}
					}
				}()
			}
		}

		defer func() {
			for _, unsub := range unsubs {
				unsub()
			}
		}()

		for {
			select {
			case <-ctx.Done():
				return

			case upd := <-updateCh:
				latest[upd.idx] = upd.val
				ready[upd.idx] = true

				// Only compute once we have at least one value for every input.
				allReady := true
				for _, r := range ready {
					if !r {
						allReady = false
						break
					}
				}
				if !allReady {
					continue
				}

				// Run the pipeline.
				s.mu.RLock()
				cur, stillExists := s.signals[signal]
				s.mu.RUnlock()
				if !stillExists {
					return
				}

				result, err := runPipeline(cur.nodes, cur.states, latest)
				if err != nil {
					s.log.Warn("synthetic: pipeline error", "signal", signal, "err", err)
					continue
				}

				v := datasource.Value{
					Timestamp: upd.ts,
					Data:      result,
					Quality:   datasource.QualityGood,
				}
				select {
				case ch <- v:
				case <-ctx.Done():
					return
				}
			}
		}
	}()

	return datasource.CancelFunc(cancel), nil
}

// Write is not supported for synthetic signals.
func (s *Synthetic) Write(_ context.Context, _ string, _ any) error {
	return datasource.ErrNotWritable
}

// History is not supported for synthetic signals.
func (s *Synthetic) History(_ context.Context, _ string, _, _ time.Time, _ int) ([]datasource.Value, error) {
	return nil, datasource.ErrHistoryUnavailable
}

// AddSignal adds a new synthetic signal definition at runtime and persists it.
func (s *Synthetic) AddSignal(def SignalDef) error {
	if def.Name == "" {
		return errors.New("signal name must not be empty")
	}

	nodes, err := BuildPipeline(def.Pipeline)
	if err != nil {
		return fmt.Errorf("build pipeline: %w", err)
	}

	s.mu.Lock()
	if _, exists := s.signals[def.Name]; exists {
		s.mu.Unlock()
		return fmt.Errorf("signal %q already exists", def.Name)
	}

	states := make([]map[string]any, len(nodes))
	for i := range states {
		states[i] = make(map[string]any)
	}

	st := &signalState{
		def:    def,
		nodes:  nodes,
		states: states,
	}
	s.signals[def.Name] = st
	s.mu.Unlock()

	if err := s.saveDefs(); err != nil {
		// Roll back the in-memory addition.
		s.mu.Lock()
		delete(s.signals, def.Name)
		s.mu.Unlock()
		return fmt.Errorf("persist definitions: %w", err)
	}

	return nil
}

// RemoveSignal removes a synthetic signal definition at runtime and persists.
func (s *Synthetic) RemoveSignal(name string) error {
	s.mu.Lock()
	st, ok := s.signals[name]
	if !ok {
		s.mu.Unlock()
		return datasource.ErrNotFound
	}
	// Stop the goroutine if it is running.
	if st.cancel != nil {
		st.cancel()
	}
	delete(s.signals, name)
	s.mu.Unlock()

	if err := s.saveDefs(); err != nil {
		return fmt.Errorf("persist definitions: %w", err)
	}
	return nil
}

// GetSignal returns the definition of a single named signal.
func (s *Synthetic) GetSignal(name string) (SignalDef, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()
	st, ok := s.signals[name]
	if !ok {
		return SignalDef{}, datasource.ErrNotFound
	}
	return st.def, nil
}

// UpdateSignal replaces the pipeline of an existing synthetic signal at runtime
// and persists the change. The signal must already exist.
func (s *Synthetic) UpdateSignal(def SignalDef) error {
	if def.Name == "" {
		return errors.New("signal name must not be empty")
	}

	nodes, err := BuildPipeline(def.Pipeline)
	if err != nil {
		return fmt.Errorf("build pipeline: %w", err)
	}

	states := make([]map[string]any, len(nodes))
	for i := range states {
		states[i] = make(map[string]any)
	}

	s.mu.Lock()
	old, exists := s.signals[def.Name]
	if !exists {
		s.mu.Unlock()
		return datasource.ErrNotFound
	}
	if old.cancel != nil {
		old.cancel()
	}
	s.signals[def.Name] = &signalState{def: def, nodes: nodes, states: states}
	s.mu.Unlock()

	if err := s.saveDefs(); err != nil {
		return fmt.Errorf("persist definitions: %w", err)
	}
	return nil
}

// GetDefs returns a copy of all current signal definitions (for the REST API).
func (s *Synthetic) GetDefs() []SignalDef {
	s.mu.RLock()
	defer s.mu.RUnlock()

	out := make([]SignalDef, 0, len(s.signals))
	for _, st := range s.signals {
		out = append(out, st.def)
	}
	return out
}

// ── internal helpers ──────────────────────────────────────────────────────────

func (s *Synthetic) defsFilePath() string {
	return filepath.Join(s.storePath, definitionsFile)
}

func (s *Synthetic) loadDefs() ([]SignalDef, error) {
	path := s.defsFilePath()
	data, err := os.ReadFile(path)
	if errors.Is(err, os.ErrNotExist) {
		// Create an empty definitions file.
		if werr := os.WriteFile(path, []byte("[]\n"), 0o644); werr != nil {
			s.log.Warn("synthetic: could not create empty definitions file", "path", path, "err", werr)
		}
		return nil, nil
	}
	if err != nil {
		return nil, err
	}

	var defs []SignalDef
	if err := json.Unmarshal(data, &defs); err != nil {
		return nil, fmt.Errorf("parse %s: %w", path, err)
	}
	return defs, nil
}

func (s *Synthetic) saveDefs() error {
	s.mu.RLock()
	defs := make([]SignalDef, 0, len(s.signals))
	for _, st := range s.signals {
		defs = append(defs, st.def)
	}
	s.mu.RUnlock()

	data, err := json.MarshalIndent(defs, "", "  ")
	if err != nil {
		return err
	}
	return os.WriteFile(s.defsFilePath(), data, 0o644)
}

// startSignal builds the pipeline for def and registers the signalState.
// The actual goroutines are started lazily by Subscribe.
func (s *Synthetic) startSignal(def SignalDef) error {
	nodes, err := BuildPipeline(def.Pipeline)
	if err != nil {
		return fmt.Errorf("build pipeline for %q: %w", def.Name, err)
	}

	states := make([]map[string]any, len(nodes))
	for i := range states {
		states[i] = make(map[string]any)
	}

	s.mu.Lock()
	s.signals[def.Name] = &signalState{
		def:    def,
		nodes:  nodes,
		states: states,
	}
	s.mu.Unlock()

	s.log.Info("synthetic: signal registered", "name", def.Name)
	return nil
}

// runPipeline executes all nodes in sequence. The output of node N becomes
// input[0] of node N+1. For the first node, inputs is the full upstream slice.
func runPipeline(nodes []dsp.Node, states []map[string]any, inputs []float64) (float64, error) {
	if len(nodes) == 0 {
		if len(inputs) == 0 {
			return 0, nil
		}
		return inputs[0], nil
	}

	// First node receives all upstream inputs.
	cur := inputs
	var result float64
	var err error

	for i, node := range nodes {
		result, err = node.Process(cur, states[i])
		if err != nil {
			return 0, fmt.Errorf("node %d (%s): %w", i, node.Type(), err)
		}
		// Subsequent nodes receive only the single output of the previous node.
		cur = []float64{result}
	}
	return result, nil
}

// upstreamRefs returns the broker.SignalRef list for a SignalDef.
// If Inputs is set, those take precedence; otherwise DS+Signal is used.
func upstreamRefs(def SignalDef) []broker.SignalRef {
	if len(def.Inputs) > 0 {
		refs := make([]broker.SignalRef, len(def.Inputs))
		for i, inp := range def.Inputs {
			refs[i] = broker.SignalRef{DS: inp.DS, Name: inp.Signal}
		}
		return refs
	}
	if def.DS != "" && def.Signal != "" {
		return []broker.SignalRef{{DS: def.DS, Name: def.Signal}}
	}
	return nil
}

// defToMetadata converts a SignalDef into a datasource.Metadata.
func defToMetadata(def SignalDef) datasource.Metadata {
	return datasource.Metadata{
		Name:        def.Name,
		Type:        datasource.TypeFloat64,
		Unit:        def.Meta.Unit,
		Description: def.Meta.Description,
		DisplayLow:  def.Meta.DisplayLow,
		DisplayHigh: def.Meta.DisplayHigh,
		Writable:    def.Meta.Writable,
	}
}

// toFloat64 coerces any numeric value from a datasource.Value.Data to float64.
func toFloat64(v any) float64 {
	switch val := v.(type) {
	case float64:
		return val
	case float32:
		return float64(val)
	case int64:
		return float64(val)
	case int32:
		return float64(val)
	case int:
		return float64(val)
	case bool:
		if val {
			return 1
		}
		return 0
	default:
		return 0
	}
}

// indexedUpdate carries a value from one upstream goroutine to the pipeline runner.
type indexedUpdate struct {
	idx int
	val float64
	ts  time.Time
}