uopi/internal/datasource/servervar/servervar.go

// Package servervar provides a small persistent key/value data source for
// "server variables": named scalar values that the server-side control-logic
// engine writes (e.g. the state of a sequence) and that interface panels can
// read live. Panels may read any variable; writes from panels are gated to
// control-logic editors in the WebSocket write handler.
package servervar

import (
	"context"
	"encoding/json"
	"os"
	"path/filepath"
	"sort"
	"sync"
	"time"

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

const fileName = "servervars.json"

type variable struct {
	value float64
	ts    time.Time
}

// Source is the "srv" data source. Variables are created on first write and
// persisted so their last value survives a restart.
type Source struct {
	path string

	mu     sync.RWMutex
	vars   map[string]*variable
	subs   map[string]map[int]chan<- datasource.Value
	nextID int
}

// New opens (or initialises) the server-variable store under storageDir.
func New(storageDir string) (*Source, error) {
	s := &Source{
		path: filepath.Join(storageDir, fileName),
		vars: map[string]*variable{},
		subs: map[string]map[int]chan<- datasource.Value{},
	}
	if err := s.load(); err != nil {
		return nil, err
	}
	return s, nil
}

func (s *Source) load() error {
	data, err := os.ReadFile(s.path)
	if os.IsNotExist(err) {
		return nil
	}
	if err != nil {
		return err
	}
	var raw map[string]float64
	if err := json.Unmarshal(data, &raw); err != nil {
		return err
	}
	now := time.Now()
	for name, v := range raw {
		s.vars[name] = &variable{value: v, ts: now}
	}
	return nil
}

// saveLocked persists the current values atomically. Caller holds s.mu.
func (s *Source) saveLocked() {
	raw := make(map[string]float64, len(s.vars))
	for name, v := range s.vars {
		raw[name] = v.value
	}
	data, err := json.MarshalIndent(raw, "", "  ")
	if err != nil {
		return
	}
	tmp := s.path + ".tmp"
	if err := os.WriteFile(tmp, data, 0o644); err != nil {
		return
	}
	_ = os.Rename(tmp, s.path)
}

func meta(name string) datasource.Metadata {
	return datasource.Metadata{
		Name:        name,
		Type:        datasource.TypeFloat64,
		Description: "Server variable",
		Writable:    true,
	}
}

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

// Connect is a no-op — the store is opened in New.
func (s *Source) Connect(_ context.Context) error { return nil }

// ListSignals returns metadata for every defined server variable.
func (s *Source) ListSignals(_ context.Context) ([]datasource.Metadata, error) {
	s.mu.RLock()
	defer s.mu.RUnlock()
	names := make([]string, 0, len(s.vars))
	for name := range s.vars {
		names = append(names, name)
	}
	sort.Strings(names)
	out := make([]datasource.Metadata, 0, len(names))
	for _, name := range names {
		out = append(out, meta(name))
	}
	return out, nil
}

// GetMetadata returns metadata for a single variable. Unknown names still report
// writable metadata so that control logic / authorised panels may create them.
func (s *Source) GetMetadata(_ context.Context, signal string) (datasource.Metadata, error) {
	return meta(signal), nil
}

// Subscribe registers ch for updates and immediately delivers the current value.
func (s *Source) Subscribe(ctx context.Context, signal string, ch chan<- datasource.Value) (datasource.CancelFunc, error) {
	s.mu.Lock()
	if s.subs[signal] == nil {
		s.subs[signal] = map[int]chan<- datasource.Value{}
	}
	id := s.nextID
	s.nextID++
	s.subs[signal][id] = ch
	cur, ok := s.vars[signal]
	var first datasource.Value
	if ok {
		first = datasource.Value{Timestamp: cur.ts, Data: cur.value, Quality: datasource.QualityGood}
	}
	s.mu.Unlock()

	if ok {
		go func() {
			select {
			case ch <- first:
			case <-ctx.Done():
			}
		}()
	}

	return func() {
		s.mu.Lock()
		if m := s.subs[signal]; m != nil {
			delete(m, id)
			if len(m) == 0 {
				delete(s.subs, signal)
			}
		}
		s.mu.Unlock()
	}, nil
}

func toFloat64(v any) float64 {
	switch x := v.(type) {
	case float64:
		return x
	case float32:
		return float64(x)
	case int:
		return float64(x)
	case int64:
		return float64(x)
	case bool:
		if x {
			return 1
		}
		return 0
	case json.Number:
		f, _ := x.Float64()
		return f
	default:
		return 0
	}
}

// Write sets a variable (creating it if needed), persists, and fans the new value
// out to all subscribers.
func (s *Source) Write(_ context.Context, signal string, value any) error {
	v := toFloat64(value)

	s.mu.Lock()
	now := time.Now()
	s.vars[signal] = &variable{value: v, ts: now}
	s.saveLocked()
	subs := make([]chan<- datasource.Value, 0, len(s.subs[signal]))
	for _, ch := range s.subs[signal] {
		subs = append(subs, ch)
	}
	s.mu.Unlock()

	upd := datasource.Value{Timestamp: now, Data: v, Quality: datasource.QualityGood}
	for _, ch := range subs {
		select {
		case ch <- upd:
		default:
		}
	}
	return nil
}

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