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 any `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 Value, 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() }) } }