package server import ( "encoding/json" "log/slog" "strconv" "sync" "sync/atomic" "github.com/uopi/uopi/internal/controllogic" ) // DebugHub fans control-logic node-execution events out to the editors watching // them, and owns the per-client debug subscriptions. It implements // controllogic.DebugObserver; the engine (and simulate sandboxes) call Observe // on every node execution. // // Two subscription modes share one event shape: // - "live" — observe the running, enabled graph by its id. // - "simulate" — dry-run an unsaved graph in a throwaway sandbox. Each // simulate sub gets a unique route id so its events never mix with the live // graph's (which may share the same persisted id). // // Routing is by event GraphID: live events carry the real graph id (only // emitted while that id is in the engine's debugWatch set, which the hub keeps // in sync with its live subs), simulate events carry the sandbox's unique id. type DebugHub struct { engine *controllogic.Engine log *slog.Logger seq uint64 // unique simulate route ids mu sync.Mutex subs map[*wsClient]*debugSub // one debug sub per client routes map[string]map[*wsClient]struct{} // route id → watching clients liveCount map[string]int // live-subscribed graph id → count } type debugSub struct { route string // graph id (live) or unique sandbox id (simulate) live bool // true for "live", false for "simulate" stop func() // simulate sandbox teardown (nil for live) } // NewDebugHub builds an empty hub bound to the engine it observes. func NewDebugHub(engine *controllogic.Engine, log *slog.Logger) *DebugHub { return &DebugHub{ engine: engine, log: log, subs: map[*wsClient]*debugSub{}, routes: map[string]map[*wsClient]struct{}{}, liveCount: map[string]int{}, } } // debugOut is the client-bound JSON form of a node-execution event. type debugOut struct { Type string `json:"type"` // always "debugNode" GraphID string `json:"graphId"` NodeID string `json:"nodeId"` Value any `json:"value"` HasValue bool `json:"hasValue"` TS int64 `json:"ts"` } // Observe implements controllogic.DebugObserver: push the event to every client // watching its route (drop-on-full so a slow editor never stalls the engine). func (h *DebugHub) Observe(ev controllogic.DebugEvent) { h.mu.Lock() watchers := h.routes[ev.GraphID] if len(watchers) == 0 { h.mu.Unlock() return } targets := make([]*wsClient, 0, len(watchers)) for c := range watchers { targets = append(targets, c) } h.mu.Unlock() b, err := json.Marshal(debugOut{ Type: "debugNode", GraphID: ev.GraphID, NodeID: ev.NodeID, Value: ev.Value, HasValue: ev.HasValue, TS: ev.TS, }) if err != nil { return } for _, c := range targets { select { case c.outCh <- b: default: } } } // subscribeLive starts (or replaces) a client's live observation of graphID. func (h *DebugHub) subscribeLive(c *wsClient, graphID string) { h.unsubscribe(c) if graphID == "" { return } sub := &debugSub{route: graphID, live: true} h.mu.Lock() h.addRoute(c, sub) h.liveCount[graphID]++ watch := h.snapshotWatch() h.mu.Unlock() h.engine.SetDebugWatch(watch) } // subscribeSimulate starts (or replaces) a client's dry-run of an unsaved graph. func (h *DebugHub) subscribeSimulate(c *wsClient, g controllogic.Graph) { h.unsubscribe(c) id := "sim-" + strconv.FormatUint(atomic.AddUint64(&h.seq, 1), 10) g.ID = id // route sandbox events under a unique id (never collides with live) stop := h.engine.StartSimulate(g) sub := &debugSub{route: id, live: false, stop: stop} h.mu.Lock() h.addRoute(c, sub) h.mu.Unlock() } // fire forces nodeID's trigger to run in the graph the client is currently // debugging (live or simulate), routed by that session's id. No-op if the // client has no active debug session. func (h *DebugHub) fire(c *wsClient, nodeID string) { h.mu.Lock() sub := h.subs[c] h.mu.Unlock() if sub == nil { return } h.engine.FireTrigger(sub.route, nodeID) } // addRoute records sub for c; caller holds h.mu. func (h *DebugHub) addRoute(c *wsClient, sub *debugSub) { h.subs[c] = sub if h.routes[sub.route] == nil { h.routes[sub.route] = map[*wsClient]struct{}{} } h.routes[sub.route][c] = struct{}{} } // unsubscribe tears down a client's debug sub (stopping its sandbox if any) and // refreshes the engine's watch set. Safe when the client has no sub. func (h *DebugHub) unsubscribe(c *wsClient) { h.mu.Lock() sub := h.subs[c] if sub == nil { h.mu.Unlock() return } delete(h.subs, c) if m := h.routes[sub.route]; m != nil { delete(m, c) if len(m) == 0 { delete(h.routes, sub.route) } } if sub.live { if h.liveCount[sub.route]--; h.liveCount[sub.route] <= 0 { delete(h.liveCount, sub.route) } } watch := h.snapshotWatch() h.mu.Unlock() if sub.stop != nil { sub.stop() } h.engine.SetDebugWatch(watch) } // snapshotWatch builds a fresh immutable set of live-watched graph ids. Caller // holds h.mu; the result is published to the engine which reads it lock-free. func (h *DebugHub) snapshotWatch() map[string]bool { w := make(map[string]bool, len(h.liveCount)) for k := range h.liveCount { w[k] = true } return w }