uopi/internal/controllogic/engine.go

package controllogic

import (
	"context"
	"log/slog"
	"math"
	"regexp"
	"strconv"
	"strings"
	"sync"
	"time"

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

// Guards against runaway flows (cycles / pathological loops).
const (
	maxSteps = 100000
	maxLoop  = 100000
)

// Engine runs all enabled control-logic graphs continuously under a root
// context. Reload tears down the current generation (subscriptions, timers,
// in-flight flows) and rebuilds from the store's enabled graphs.
type Engine struct {
	broker *broker.Broker
	store  *Store
	log    *slog.Logger
	root   context.Context

	mu       sync.Mutex
	cancel   context.CancelFunc // cancels the current generation
	wg       *sync.WaitGroup    // tracks the current generation's goroutines

	// Shared live signal cache for the current generation (key "ds\0name").
	liveMu sync.RWMutex
	live   map[string]float64
}

// NewEngine creates an engine bound to root. Call Reload to start it.
func NewEngine(root context.Context, brk *broker.Broker, store *Store, log *slog.Logger) *Engine {
	return &Engine{
		broker: brk,
		store:  store,
		log:    log,
		root:   root,
		live:   map[string]float64{},
	}
}

// ── reference / value helpers ──────────────────────────────────────────────────

func refKey(ds, name string) string { return ds + "\x00" + name }

// parseRef splits a "ds:name" target on the FIRST ':' (EPICS PV names contain
// ':'). A bare name (no ':') is a graph-local variable in data source "local".
func parseRef(target string) (ds, name string, ok bool) {
	t := strings.TrimSpace(target)
	if t == "" {
		return "", "", false
	}
	i := strings.IndexByte(t, ':')
	if i < 0 {
		return "local", t, true
	}
	return t[:i], t[i+1:], true
}

func toNum(v any) float64 {
	switch x := v.(type) {
	case float64:
		return x
	case float32:
		return float64(x)
	case int64:
		return float64(x)
	case int:
		return float64(x)
	case bool:
		if x {
			return 1
		}
		return 0
	case string:
		f, err := strconv.ParseFloat(strings.TrimSpace(x), 64)
		if err != nil {
			return math.NaN()
		}
		return f
	default:
		return math.NaN()
	}
}

// ── lifecycle ───────────────────────────────────────────────────────────────

// Reload rebuilds the engine from the store. Safe to call repeatedly (after any
// graph mutation). It is a no-op-safe full restart of the running generation.
func (e *Engine) Reload() {
	e.mu.Lock()
	defer e.mu.Unlock()

	// Tear down the previous generation and wait for its goroutines to exit.
	if e.cancel != nil {
		e.cancel()
		e.cancel = nil
	}
	if e.wg != nil {
		e.wg.Wait()
		e.wg = nil
	}

	e.liveMu.Lock()
	e.live = map[string]float64{}
	e.liveMu.Unlock()

	graphs := e.store.List()
	var compiled []*compiledGraph
	refs := map[string]RefLite{}
	for i := range graphs {
		g := graphs[i]
		if !g.Enabled {
			continue
		}
		cg := compile(g)
		compiled = append(compiled, cg)
		for k, r := range cg.refs {
			refs[k] = r
		}
	}
	if len(compiled) == 0 {
		return
	}

	genCtx, cancel := context.WithCancel(e.root)
	wg := &sync.WaitGroup{}
	e.cancel = cancel
	e.wg = wg

	for _, cg := range compiled {
		cg.engine = e
		cg.genCtx = genCtx
		cg.wg = wg
	}

	// One shared updates channel feeds a single dispatch goroutine; every
	// subscription delivers into it. Subscriptions are released on teardown.
	updates := make(chan broker.Update, 128)
	var unsubs []func()
	for _, r := range refs {
		unsub, err := e.broker.Subscribe(broker.SignalRef{DS: r.DS, Name: r.Name}, updates)
		if err != nil {
			e.log.Warn("control logic: subscribe failed", "ds", r.DS, "signal", r.Name, "err", err)
			continue
		}
		unsubs = append(unsubs, unsub)
	}

	wg.Add(1)
	go func() {
		defer wg.Done()
		<-genCtx.Done()
		for _, u := range unsubs {
			u()
		}
	}()

	// Dispatch goroutine: keep the live cache fresh and drive level/edge triggers.
	wg.Add(1)
	go func() {
		defer wg.Done()
		for {
			select {
			case u := <-updates:
				val := toNum(u.Value.Data)
				key := refKey(u.Ref.DS, u.Ref.Name)
				e.liveMu.Lock()
				e.live[key] = val
				e.liveMu.Unlock()
				for _, cg := range compiled {
					cg.onSignal(key, val)
				}
			case <-genCtx.Done():
				return
			}
		}
	}()

	// Start timer and cron triggers.
	for _, cg := range compiled {
		cg.startTriggers()
	}

	e.log.Info("control logic engine reloaded", "graphs", len(compiled), "signals", len(refs))
}

// liveGet reads the current value of a signal from the shared cache.
func (e *Engine) liveGet(ds, name string) float64 {
	if ds == "sys" {
		if name == "time" {
			return float64(time.Now().UnixNano()) / 1e9
		}
		return math.NaN() // sys:dt handled per-activation
	}
	e.liveMu.RLock()
	defer e.liveMu.RUnlock()
	v, ok := e.live[refKey(ds, name)]
	if !ok {
		return math.NaN()
	}
	return v
}

// write applies an action.write/lua-set to a target: a bare name updates a
// graph-local var; a ds:name target writes to the data source.
func (e *Engine) write(cg *compiledGraph, target string, val float64) {
	ds, name, ok := parseRef(target)
	if !ok || math.IsNaN(val) {
		return
	}
	if ds == "local" {
		cg.setLocal(name, val)
		return
	}
	src, ok := e.broker.Source(ds)
	if !ok {
		e.log.Warn("control logic: write to unknown data source", "ds", ds, "signal", name)
		return
	}
	if err := src.Write(e.root, name, val); err != nil {
		e.log.Warn("control logic: write failed", "ds", ds, "signal", name, "err", err)
	}
}

// ── compiled graph ─────────────────────────────────────────────────────────────

type wireOut struct {
	to   string
	port string
}

// compiledGraph holds the runtime state for one enabled graph.
type compiledGraph struct {
	engine *Engine
	genCtx context.Context
	wg     *sync.WaitGroup

	name  string
	byId  map[string]Node
	out   map[string][]wireOut
	inc   map[string][]string // incoming source ids per node (for gates)
	refs  map[string]RefLite  // unique signals to subscribe (excl. sys/local)

	watchers map[string][]string // signal key → trigger node ids
	luaNodes map[string]*luaRuntime

	stateMu    sync.Mutex
	levelState map[string]bool    // current truth of level triggers (threshold/alarm)
	prevBool   map[string]bool    // edge detection for threshold/alarm
	prevVal    map[string]float64 // last value for change triggers
	hasVal     map[string]bool
	lastFire   map[string]int64 // ns wall clock each trigger last fired
	locals     map[string]float64
}

func compile(g Graph) *compiledGraph {
	cg := &compiledGraph{
		name:       g.Name,
		byId:       map[string]Node{},
		out:        map[string][]wireOut{},
		inc:        map[string][]string{},
		refs:       map[string]RefLite{},
		watchers:   map[string][]string{},
		luaNodes:   map[string]*luaRuntime{},
		levelState: map[string]bool{},
		prevBool:   map[string]bool{},
		prevVal:    map[string]float64{},
		hasVal:     map[string]bool{},
		lastFire:   map[string]int64{},
		locals:     map[string]float64{},
	}
	for _, n := range g.Nodes {
		cg.byId[n.ID] = n
	}
	for _, w := range g.Wires {
		port := w.FromPort
		if port == "" {
			port = "out"
		}
		cg.out[w.From] = append(cg.out[w.From], wireOut{to: w.To, port: port})
		cg.inc[w.To] = append(cg.inc[w.To], w.From)
	}

	want := func(ds, name string) {
		if ds == "sys" || ds == "local" {
			return
		}
		if name == "" {
			return
		}
		cg.refs[refKey(ds, name)] = RefLite{DS: ds, Name: name}
	}
	wantExpr := func(expr string) {
		for _, r := range CollectRefs(expr) {
			want(r.DS, r.Name)
		}
	}

	for _, n := range g.Nodes {
		switch n.Kind {
		case "trigger.threshold", "trigger.change", "trigger.alarm":
			if ds, name, ok := parseRef(n.param("signal")); ok {
				want(ds, name)
				key := refKey(ds, name)
				cg.watchers[key] = append(cg.watchers[key], n.ID)
			}
		case "flow.if":
			wantExpr(n.param("cond"))
		case "flow.loop":
			if n.param("mode") == "while" {
				wantExpr(n.param("cond"))
			}
		case "action.write", "action.log":
			wantExpr(n.param("expr"))
		case "action.lua":
			cg.luaNodes[n.ID] = newLuaRuntime(n.param("script"))
			for _, r := range luaGetRefs(n.param("script")) {
				want(r.DS, r.Name)
			}
		}
	}
	return cg
}

func (cg *compiledGraph) setLocal(name string, v float64) {
	cg.stateMu.Lock()
	cg.locals[name] = v
	cg.stateMu.Unlock()
}

func (cg *compiledGraph) getLocal(name string) float64 {
	cg.stateMu.Lock()
	defer cg.stateMu.Unlock()
	v, ok := cg.locals[name]
	if !ok {
		return 0
	}
	return v
}

// startTriggers launches timer and cron trigger goroutines for the generation.
func (cg *compiledGraph) startTriggers() {
	hasCron := false
	for _, n := range cg.byId {
		switch n.Kind {
		case "trigger.timer":
			node := n
			cg.wg.Add(1)
			go func() {
				defer cg.wg.Done()
				d := intervalOf(node)
				t := time.NewTicker(d)
				defer t.Stop()
				for {
					select {
					case <-t.C:
						cg.activate(node.ID)
					case <-cg.genCtx.Done():
						return
					}
				}
			}()
		case "trigger.cron":
			hasCron = true
		}
	}
	if hasCron {
		cg.startCron()
	}
}

func (cg *compiledGraph) startCron() {
	type cronNode struct {
		id    string
		sched *Schedule
	}
	var crons []cronNode
	for _, n := range cg.byId {
		if n.Kind != "trigger.cron" {
			continue
		}
		sched, err := ParseSchedule(n.param("spec"))
		if err != nil {
			cg.engine.log.Warn("control logic: bad cron spec", "graph", cg.name, "spec", n.param("spec"), "err", err)
			continue
		}
		crons = append(crons, cronNode{id: n.ID, sched: sched})
	}
	if len(crons) == 0 {
		return
	}
	cg.wg.Add(1)
	go func() {
		defer cg.wg.Done()
		t := time.NewTicker(time.Second)
		defer t.Stop()
		lastMinute := -1
		for {
			select {
			case now := <-t.C:
				minute := now.Hour()*60 + now.Minute()
				if minute == lastMinute {
					continue // fire at most once per minute
				}
				lastMinute = minute
				for _, c := range crons {
					if c.sched.Match(now) {
						cg.activate(c.id)
					}
				}
			case <-cg.genCtx.Done():
				return
			}
		}
	}()
}

func intervalOf(n Node) time.Duration {
	ms, err := strconv.Atoi(strings.TrimSpace(n.param("interval")))
	if err != nil || ms < 50 {
		ms = 1000
	}
	return time.Duration(ms) * time.Millisecond
}

// ── trigger evaluation ─────────────────────────────────────────────────────────

// onSignal drives threshold/alarm (rising edge) and change triggers when a
// watched signal updates.
func (cg *compiledGraph) onSignal(key string, value float64) {
	for _, id := range cg.watchers[key] {
		node, ok := cg.byId[id]
		if !ok {
			continue
		}
		switch node.Kind {
		case "trigger.threshold":
			cur := testThreshold(value, node.param("op"), parseFloat(node.param("value")))
			cg.stateMu.Lock()
			cg.levelState[id] = cur
			prev := cg.prevBool[id]
			cg.prevBool[id] = cur
			cg.stateMu.Unlock()
			if cur && !prev {
				cg.activate(id)
			}
		case "trigger.alarm":
			lo := parseFloat(node.param("min"))
			hi := parseFloat(node.param("max"))
			cur := !math.IsNaN(value) && (value < lo || value > hi)
			cg.stateMu.Lock()
			cg.levelState[id] = cur
			prev := cg.prevBool[id]
			cg.prevBool[id] = cur
			cg.stateMu.Unlock()
			if cur && !prev {
				cg.activate(id)
			}
		case "trigger.change":
			cg.stateMu.Lock()
			had := cg.hasVal[id]
			prev := cg.prevVal[id]
			cg.prevVal[id] = value
			cg.hasVal[id] = true
			cg.stateMu.Unlock()
			if had && value != prev {
				cg.activate(id)
			}
		}
	}
}

func testThreshold(val float64, op string, cmp float64) bool {
	if math.IsNaN(val) {
		return false
	}
	switch op {
	case "<":
		return val < cmp
	case ">=":
		return val >= cmp
	case "<=":
		return val <= cmp
	case "==":
		return val == cmp
	case "!=":
		return val != cmp
	default: // ">"
		return val > cmp
	}
}

func parseFloat(s string) float64 {
	f, err := strconv.ParseFloat(strings.TrimSpace(s), 64)
	if err != nil {
		return 0
	}
	return f
}

// ── execution ──────────────────────────────────────────────────────────────────

type runCtx struct {
	fired   string
	steps   int
	resolve Resolver
}

// activate spawns a flow run for a trigger on its own goroutine so that
// action.delay does not block signal dispatch or other flows.
func (cg *compiledGraph) activate(triggerID string) {
	now := time.Now().UnixNano()
	cg.stateMu.Lock()
	last, had := cg.lastFire[triggerID]
	cg.lastFire[triggerID] = now
	cg.stateMu.Unlock()
	dt := 0.0
	if had {
		dt = float64(now-last) / 1e9
	}

	resolve := func(ds, name string) float64 {
		switch ds {
		case "sys":
			if name == "dt" {
				return dt
			}
			return cg.engine.liveGet("sys", name)
		case "local":
			return cg.getLocal(name)
		default:
			return cg.engine.liveGet(ds, name)
		}
	}

	cg.wg.Add(1)
	go func() {
		defer cg.wg.Done()
		ctx := &runCtx{fired: triggerID, resolve: resolve}
		cg.follow(triggerID, "out", ctx)
	}()
}

func (cg *compiledGraph) follow(fromID, port string, ctx *runCtx) {
	for _, w := range cg.out[fromID] {
		if w.port == port {
			cg.run(w.to, ctx)
		}
	}
}

func (cg *compiledGraph) run(nodeID string, ctx *runCtx) {
	if ctx.steps > maxSteps {
		return
	}
	ctx.steps++
	node, ok := cg.byId[nodeID]
	if !ok {
		return
	}

	select {
	case <-cg.genCtx.Done():
		return
	default:
	}

	switch node.Kind {
	case "gate.and":
		if cg.gateSatisfied(node.ID, ctx.fired) {
			cg.follow(node.ID, "out", ctx)
		}

	case "flow.if":
		branch := "else"
		if EvalBool(node.param("cond"), ctx.resolve) {
			branch = "then"
		}
		cg.follow(node.ID, branch, ctx)

	case "flow.loop":
		if node.param("mode") == "while" {
			for i := 0; i < maxLoop && ctx.steps <= maxSteps && EvalBool(node.param("cond"), ctx.resolve); i++ {
				cg.follow(node.ID, "body", ctx)
			}
		} else {
			n := int(EvalExpr(node.param("count"), ctx.resolve))
			if n < 0 {
				n = 0
			}
			if n > maxLoop {
				n = maxLoop
			}
			for i := 0; i < n && ctx.steps <= maxSteps; i++ {
				cg.follow(node.ID, "body", ctx)
			}
		}
		cg.follow(node.ID, "done", ctx)

	case "action.write":
		val := EvalExpr(node.param("expr"), ctx.resolve)
		cg.engine.write(cg, node.param("target"), val)
		cg.follow(node.ID, "out", ctx)

	case "action.delay":
		ms := 0
		if v, err := strconv.Atoi(strings.TrimSpace(node.param("ms"))); err == nil && v > 0 {
			ms = v
		}
		if ms > 0 {
			t := time.NewTimer(time.Duration(ms) * time.Millisecond)
			select {
			case <-t.C:
			case <-cg.genCtx.Done():
				t.Stop()
				return
			}
		}
		cg.follow(node.ID, "out", ctx)

	case "action.log":
		val := EvalExpr(node.param("expr"), ctx.resolve)
		label := strings.TrimSpace(node.param("label"))
		cg.engine.log.Info("control logic log", "graph", cg.name, "label", label, "value", val)
		cg.follow(node.ID, "out", ctx)

	case "action.lua":
		cg.runLua(node.ID, ctx)
		cg.follow(node.ID, "out", ctx)

	default:
		cg.follow(node.ID, "out", ctx)
	}
}

// gateSatisfied: every incoming trigger must currently be satisfied. The firing
// trigger counts as satisfied; level triggers (threshold/alarm) use their truth.
func (cg *compiledGraph) gateSatisfied(gateID, fired string) bool {
	inputs := cg.inc[gateID]
	if len(inputs) == 0 {
		return false
	}
	cg.stateMu.Lock()
	defer cg.stateMu.Unlock()
	for _, src := range inputs {
		if src == fired || cg.levelState[src] {
			continue
		}
		return false
	}
	return true
}

// luaGetRefs scans a Lua script for get("ds:name") / get('ds:name') literals so
// the engine can subscribe to the signals the script reads.
var luaGetRe = regexp.MustCompile(`get\s*\(\s*["']([^"']+)["']`)

func luaGetRefs(script string) []RefLite {
	var out []RefLite
	for _, m := range luaGetRe.FindAllStringSubmatch(script, -1) {
		if ds, name, ok := parseRef(m[1]); ok {
			out = append(out, RefLite{DS: ds, Name: name})
		}
	}
	return out
}

func (cg *compiledGraph) runLua(nodeID string, ctx *runCtx) {
	lr := cg.luaNodes[nodeID]
	if lr == nil {
		return
	}
	lr.run(ctx.resolve, func(target string, val float64) {
		cg.engine.write(cg, target, val)
	}, func(msg string) {
		cg.engine.log.Info("control logic lua", "graph", cg.name, "msg", msg)
	})
}