uopi/internal/controllogic/debug.go

package controllogic

import (
	"context"
	"sync"
	"time"

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

// DebugEvent reports a single node execution for the live debug view. Value is
// only meaningful when HasValue is true (e.g. an action.write's written value or
// a flow.if branch as 0/1); otherwise the event just marks the node as active.
type DebugEvent struct {
	GraphID  string  `json:"graphId"`
	NodeID   string  `json:"nodeId"`
	Value    float64 `json:"value"`
	HasValue bool    `json:"hasValue"`
	TS       int64   `json:"ts"` // unix millis
}

// DebugObserver receives node-execution events from running (and simulated)
// graphs. The server implements it; Observe must not block (the hub fans out
// drop-on-full so a slow editor never stalls the engine).
type DebugObserver interface {
	Observe(DebugEvent)
}

// debugObsBox wraps a DebugObserver so atomic.Value always sees one type.
type debugObsBox struct{ o DebugObserver }

// SetDebugObserver installs the sink for node-execution events. Safe to call
// once at startup; read lock-free by running flows.
func (e *Engine) SetDebugObserver(o DebugObserver) {
	e.debugObs.Store(debugObsBox{o: o})
}

// SetDebugWatch replaces the set of graph ids with at least one live debug
// subscriber. emitDebug short-circuits for graphs absent from this set, so the
// common (nobody watching) case costs a single atomic load. The hub owns the
// map and must not mutate it after publishing (it is read without a lock).
func (e *Engine) SetDebugWatch(ids map[string]bool) {
	e.debugWatch.Store(ids)
}

// registerFire records a compiled graph's manual-fire channel under its route id
// (live graph id or simulate sandbox id).
func (e *Engine) registerFire(id string, ch chan string) {
	e.fireMu.Lock()
	e.fireChs[id] = ch
	e.fireMu.Unlock()
}

// unregisterFire removes id's fire channel, but only if it still points at ch —
// so a newer generation that reused the same id (live reload) is not clobbered
// by the old generation's teardown.
func (e *Engine) unregisterFire(id string, ch chan string) {
	e.fireMu.Lock()
	if e.fireChs[id] == ch {
		delete(e.fireChs, id)
	}
	e.fireMu.Unlock()
}

// FireTrigger asks the graph behind a debug route (graphID) to run triggerID's
// flow now, as if the trigger had fired. Non-blocking and best-effort: returns
// false if the route is gone or its fire buffer is full. The receiving graph
// validates that triggerID is actually one of its trigger nodes.
func (e *Engine) FireTrigger(graphID, triggerID string) bool {
	e.fireMu.Lock()
	ch := e.fireChs[graphID]
	e.fireMu.Unlock()
	if ch == nil || triggerID == "" {
		return false
	}
	select {
	case ch <- triggerID:
		return true
	default:
		return false
	}
}

// emitDebug reports a node execution to the observer when the graph is watched
// (or the graph is a simulate sandbox, which is always its own subscriber).
func (cg *compiledGraph) emitDebug(nodeID string, value float64, hasValue bool) {
	if !cg.alwaysDebug {
		w, _ := cg.engine.debugWatch.Load().(map[string]bool)
		if !w[cg.id] {
			return
		}
	}
	box, _ := cg.engine.debugObs.Load().(debugObsBox)
	if box.o == nil {
		return
	}
	box.o.Observe(DebugEvent{
		GraphID:  cg.id,
		NodeID:   nodeID,
		Value:    value,
		HasValue: hasValue,
		TS:       time.Now().UnixMilli(),
	})
}

// StartSimulate runs g in a throwaway sandbox generation: real side effects are
// suppressed (data-source writes, config mutations, dialogs) but local vars,
// triggers, timers and the flow itself execute normally, emitting debug events
// the editor can visualise. It is independent of Reload's live generation. The
// returned stop func cancels the sandbox and waits for its goroutines to drain;
// it is safe to call more than once.
func (e *Engine) StartSimulate(g Graph) func() {
	cg := compile(g)
	cg.engine = e
	cg.dryRun = true
	cg.alwaysDebug = true

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

	// Sandbox-local live cache so simulate reads don't disturb the live engine.
	updates := make(chan broker.Update, 128)
	var unsubs []func()
	for _, r := range cg.refs {
		unsub, err := e.broker.Subscribe(broker.SignalRef{DS: r.DS, Name: r.Name}, updates)
		if err != nil {
			e.log.Warn("control logic simulate: subscribe failed", "ds", r.DS, "signal", r.Name, "err", err)
			continue
		}
		unsubs = append(unsubs, unsub)
	}

	e.registerFire(cg.id, cg.fireCh)

	wg.Add(1)
	go func() {
		defer wg.Done()
		<-ctx.Done()
		for _, u := range unsubs {
			u()
		}
		e.unregisterFire(cg.id, cg.fireCh)
	}()

	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()
				cg.onSignal(key, val)
			case <-ctx.Done():
				return
			}
		}
	}()

	cg.startTriggers()

	var once sync.Once
	return func() {
		once.Do(func() {
			cancel()
			wg.Wait()
		})
	}
}