diff --git a/TODO.md b/TODO.md new file mode 100644 index 0000000..fff23d4 --- /dev/null +++ b/TODO.md @@ -0,0 +1,28 @@ +# TODO + +- [ ] Implement git style versioning for: synthetic variable, panels, control logic +- [ ] Implement configuration manager: + - configuration is a set of signals (that can be organized in group and subgroup) that are used to configure a system or a sub system + - with configuration manager user should be able to: + - Define and manage configuration sets: delete (keep server side copy of deleted config), create (specifying default to parameters, mandatory, optional etc), edit (with versioning git style) and compare set + - user can fork an existing config instance to create a new one + - user can compare two instances: show unified diff or side by side diff + - etc... + - Create, delete (keep server cache), edit (git style versioning) and compare configuration instances: meaning set actual value to a configuration set + - user can fork an existing config instance to create a new one + - user can compare two instances: show unified diff or side by side diff + - etc... +- [ ] Implement admin pane: create / manage groups, set users permits, manage auditors etc +- [ ] Implement new datasources: + - [ ] modbus tcp + - [ ] scpi tcp +- [ ] Improve UX: + - [ ] Synthetic editor: + - color code the node link by type + - hover on a block in error should show the reason + - add proper array functionality + - add shortcut to add Signals (S) and nodes (N) with HUD + - [ ] Panels: + - in view mode the widgets should have no border/bg but blend with background + - add widgets such +- [ ] Implement proper distributed server side nodes to balance load and have redundancy (if a node is not available anymore all its clients migrate seamelessly to another) diff --git a/cmd/uopi/main.go b/cmd/uopi/main.go index c802da0..30cb1a9 100644 --- a/cmd/uopi/main.go +++ b/cmd/uopi/main.go @@ -186,9 +186,15 @@ func main() { os.Exit(1) } ctrlEngine := controllogic.NewEngine(ctx, brk, ctrlStore, recorder, log) + + // Dialog hub: control-logic action.dialog nodes push notifications/inputs to + // connected clients (filtered by user/group) and route input responses back + // to server variables. Installed before Reload so running graphs can emit. + dialogs := server.NewDialogHub(brk, policy, recorder, log) + ctrlEngine.SetNotifier(dialogs) ctrlEngine.Reload() - srv := server.New(cfg.Server.Listen, webFS, brk, synthDS, store, policy, aclStore, ctrlStore, ctrlEngine, recorder, cfg.Datasource.EPICS.ChannelFinderURL, cfg.Datasource.EPICS.ArchiveURL, cfg.Server.TrustedUserHeader, log) + srv := server.New(cfg.Server.Listen, webFS, brk, synthDS, store, policy, aclStore, ctrlStore, ctrlEngine, dialogs, recorder, cfg.Datasource.EPICS.ChannelFinderURL, cfg.Datasource.EPICS.ArchiveURL, cfg.Server.TrustedUserHeader, log) if err := srv.Start(ctx); err != nil { fmt.Fprintf(os.Stderr, "server error: %v\n", err) diff --git a/internal/controllogic/engine.go b/internal/controllogic/engine.go index f307146..a863f28 100644 --- a/internal/controllogic/engine.go +++ b/internal/controllogic/engine.go @@ -8,6 +8,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "time" "github.com/uopi/uopi/internal/audit" @@ -34,11 +35,28 @@ type Engine struct { cancel context.CancelFunc // cancels the current generation wg *sync.WaitGroup // tracks the current generation's goroutines + // notifier delivers action.dialog requests to connected clients. Stored in + // an atomic so flow goroutines can read it without taking e.mu (which Reload + // holds while it waits for those same goroutines to drain). + notifier atomic.Value // notifierBox + // Shared live signal cache for the current generation (key "ds\0name"). liveMu sync.RWMutex live map[string]float64 } +// notifierBox wraps a Notifier so atomic.Value always sees one concrete type. +type notifierBox struct{ n Notifier } + +// dialogSeq generates unique action.dialog ids across all graphs. +var dialogSeq uint64 + +// SetNotifier installs the sink for action.dialog requests. Safe to call once +// at startup before or after Reload; it is read lock-free by running flows. +func (e *Engine) SetNotifier(n Notifier) { + e.notifier.Store(notifierBox{n: n}) +} + // NewEngine creates an engine bound to root. Call Reload to start it. rec records // the writes performed by flows; pass audit.Nop() to disable auditing. func NewEngine(root context.Context, brk *broker.Broker, store *Store, rec audit.Recorder, log *slog.Logger) *Engine { @@ -252,6 +270,29 @@ func (e *Engine) write(cg *compiledGraph, target string, val float64) { e.audit.Record(ev) } +// emitDialog delivers an action.dialog node's request to the installed Notifier +// (the WebSocket dialog hub). It is lock-free so it never deadlocks against a +// concurrent Reload that is waiting for this flow goroutine to finish. +func (e *Engine) emitDialog(n Node) { + box, _ := e.notifier.Load().(notifierBox) + if box.n == nil { + return + } + kind := strings.TrimSpace(n.param("kind")) + if kind != "error" && kind != "input" { + kind = "info" + } + box.n.Notify(Dialog{ + ID: strconv.FormatUint(atomic.AddUint64(&dialogSeq, 1), 10), + Kind: kind, + Title: n.param("title"), + Message: n.param("message"), + Target: strings.TrimSpace(n.param("target")), + Users: splitCSV(n.param("users")), + Groups: splitCSV(n.param("groups")), + }) +} + // ── compiled graph ───────────────────────────────────────────────────────────── type wireOut struct { @@ -658,6 +699,10 @@ func (cg *compiledGraph) run(nodeID string, ctx *runCtx) { cg.runLua(node.ID, ctx) cg.follow(node.ID, "out", ctx) + case "action.dialog": + cg.engine.emitDialog(node) + cg.follow(node.ID, "out", ctx) + default: cg.follow(node.ID, "out", ctx) } diff --git a/internal/controllogic/model.go b/internal/controllogic/model.go index 26541a0..0dc753d 100644 --- a/internal/controllogic/model.go +++ b/internal/controllogic/model.go @@ -29,6 +29,8 @@ package controllogic // action.delay — waits `ms` before continuing. // action.log — logs an expression value to the server log. // action.lua — runs a sandboxed Lua script with get/set/log host funcs. +// action.dialog — pushes an info/error/input dialog to connected clients +// filtered by user/group; input responses write `target`. // Node is a single node in a control-logic graph. Params are stored as strings // (matching the panel logic model) and parsed per-kind by the engine. diff --git a/internal/controllogic/notify.go b/internal/controllogic/notify.go new file mode 100644 index 0000000..59ab321 --- /dev/null +++ b/internal/controllogic/notify.go @@ -0,0 +1,37 @@ +package controllogic + +import "strings" + +// Dialog is a user-facing notification or input request emitted by an +// action.dialog node. It is delivered to connected clients whose identity +// matches Users/Groups (both empty = everyone). For an "input" dialog the +// client's response is written back to Target (a "ds:name" reference, e.g. +// "srv:approved") so control logic can read it on a later activation. +type Dialog struct { + ID string `json:"id"` + Kind string `json:"kind"` // "info" | "error" | "input" + Title string `json:"title"` + Message string `json:"message"` + Target string `json:"target,omitempty"` + Users []string `json:"users,omitempty"` + Groups []string `json:"groups,omitempty"` +} + +// Notifier delivers control-logic dialogs to connected clients. The server +// implements it; the engine calls Notify when an action.dialog node runs. +// Notify must not block (the hub fans out without waiting on slow clients). +type Notifier interface { + Notify(Dialog) +} + +// splitCSV parses a comma-separated user/group filter into trimmed, +// non-empty tokens. An empty string yields a nil slice (no filter). +func splitCSV(s string) []string { + var out []string + for _, p := range strings.Split(s, ",") { + if t := strings.TrimSpace(p); t != "" { + out = append(out, t) + } + } + return out +} diff --git a/internal/datasource/synthetic/definition.go b/internal/datasource/synthetic/definition.go index a5cf39f..2e1c807 100644 --- a/internal/datasource/synthetic/definition.go +++ b/internal/datasource/synthetic/definition.go @@ -9,6 +9,7 @@ type SignalDef struct { Signal string `json:"signal"` // upstream signal name (or "" for constant) Inputs []InputRef `json:"inputs"` // alternative multi-input format Pipeline []NodeDef `json:"pipeline"` // ordered list of DSP nodes + Graph *Graph `json:"graph,omitempty"` // DAG form (preferred when present) Meta MetaOverride `json:"meta"` // optional metadata overrides // Visibility controls who sees this signal in the signal tree: @@ -34,6 +35,37 @@ type NodeDef struct { Params map[string]any `json:"params,omitempty"` } +// Graph is the DAG form of a synthetic signal: a set of nodes (sources, ops and +// one output) wired together by explicit per-node ordered input lists. It +// supersedes the linear Inputs+Pipeline form: when SignalDef.Graph is set it is +// authoritative; otherwise the legacy linear fields are converted into an +// equivalent graph at load time (see toGraph). +type Graph struct { + Nodes []GraphNode `json:"nodes"` + Output string `json:"output"` // id of the output node +} + +// GraphNode is one node in a Graph. +// +// kind=="source": carries DS+Signal; has no Inputs (a graph root). +// kind=="op": carries Op + Params; Inputs lists upstream node IDs in the +// order the op receives them (input 0, 1, … e.g. a−b, a÷b). +// kind=="output": Inputs has a single upstream node whose value is the result. +// +// X/Y are the editor layout coordinates, persisted so a reloaded graph keeps its +// shape; they have no effect on evaluation. +type GraphNode struct { + ID string `json:"id"` + Kind string `json:"kind"` + Op string `json:"op,omitempty"` + Params map[string]any `json:"params,omitempty"` + DS string `json:"ds,omitempty"` + Signal string `json:"signal,omitempty"` + Inputs []string `json:"inputs,omitempty"` + X float64 `json:"x,omitempty"` + Y float64 `json:"y,omitempty"` +} + // MetaOverride allows the synthetic signal to override display metadata. type MetaOverride struct { Unit string `json:"unit,omitempty"` diff --git a/internal/datasource/synthetic/dsp_bridge.go b/internal/datasource/synthetic/dsp_bridge.go index 9156168..3d36da6 100644 --- a/internal/datasource/synthetic/dsp_bridge.go +++ b/internal/datasource/synthetic/dsp_bridge.go @@ -38,21 +38,32 @@ func stringParam(params map[string]any, key string) string { return s } -// BuildPipeline converts a []NodeDef (from JSON) into a []dsp.Node ready for -// execution. JSON numbers are float64, so all numeric params are handled as -// float64 regardless of the final type needed. -func BuildPipeline(defs []NodeDef) ([]dsp.Node, error) { - nodes := make([]dsp.Node, 0, len(defs)) - for i, d := range defs { - n, err := buildNode(d) - if err != nil { - return nil, fmt.Errorf("pipeline node %d (%q): %w", i, d.Type, err) - } - nodes = append(nodes, n) +// stringSliceParam extracts a []string from params; JSON arrays decode to []any, +// so each element is coerced via its string value. Returns nil if missing. +func stringSliceParam(params map[string]any, key string) []string { + if params == nil { + return nil } - return nodes, nil + v, ok := params[key] + if !ok { + return nil + } + arr, ok := v.([]any) + if !ok { + return nil + } + out := make([]string, 0, len(arr)) + for _, e := range arr { + if s, ok := e.(string); ok && s != "" { + out = append(out, s) + } + } + return out } +// buildNode converts a single NodeDef (from JSON) into a dsp.Node ready for +// execution. JSON numbers are float64, so all numeric params are handled as +// float64 regardless of the final type needed. func buildNode(d NodeDef) (dsp.Node, error) { p := d.Params switch d.Type { @@ -100,7 +111,7 @@ func buildNode(d NodeDef) (dsp.Node, error) { }, nil case "expr": - return &dsp.ExprNode{Expr: stringParam(p, "expr")}, nil + return &dsp.ExprNode{Expr: stringParam(p, "expr"), Vars: stringSliceParam(p, "vars")}, nil case "lowpass": order := int(floatParam(p, "order")) @@ -113,7 +124,7 @@ func buildNode(d NodeDef) (dsp.Node, error) { }, nil case "lua": - return &dsp.LuaNode{Script: stringParam(p, "script")}, nil + return &dsp.LuaNode{Script: stringParam(p, "script"), Vars: stringSliceParam(p, "vars")}, nil default: return nil, fmt.Errorf("unknown node type %q", d.Type) diff --git a/internal/datasource/synthetic/graph.go b/internal/datasource/synthetic/graph.go new file mode 100644 index 0000000..982976f --- /dev/null +++ b/internal/datasource/synthetic/graph.go @@ -0,0 +1,199 @@ +package synthetic + +import ( + "errors" + "fmt" + + "github.com/uopi/uopi/internal/broker" + "github.com/uopi/uopi/internal/dsp" +) + +// runtimeGraph is the executable form of a synthetic signal's DAG. Nodes are +// held in topological order so a single forward pass computes every value with +// each node's inputs already resolved. Op-node state maps persist across +// evaluations (for stateful nodes like moving_average / lua). +type runtimeGraph struct { + order []*rtNode // topological order (sources first, output last) + sources []rtSource // source nodes, in topological order + outputID string // id of the output node +} + +type rtNode struct { + id string + kind string // source | op | output + op dsp.Node // set for kind==op + state map[string]any // persistent per-node state (op only) + inputs []string // upstream node ids, in input order +} + +type rtSource struct { + id string + ref broker.SignalRef +} + +// sourceRefs returns the broker references for every source node, in a stable +// order matching rg.sources. +func (rg *runtimeGraph) sourceRefs() []broker.SignalRef { + refs := make([]broker.SignalRef, len(rg.sources)) + for i, s := range rg.sources { + refs[i] = s.ref + } + return refs +} + +// eval computes the output value given the latest value for each source node +// (keyed by source node id). Nodes are visited in topological order so every +// input is already present in vals by the time a node is processed. +func (rg *runtimeGraph) eval(sourceVals map[string]float64) (float64, error) { + vals := make(map[string]float64, len(rg.order)) + for id, v := range sourceVals { + vals[id] = v + } + for _, n := range rg.order { + switch n.kind { + case "op": + in := make([]float64, len(n.inputs)) + for i, id := range n.inputs { + in[i] = vals[id] + } + r, err := n.op.Process(in, n.state) + if err != nil { + return 0, fmt.Errorf("node %s (%s): %w", n.id, n.op.Type(), err) + } + vals[n.id] = r + case "output": + if len(n.inputs) > 0 { + vals[n.id] = vals[n.inputs[0]] + } + } + } + return vals[rg.outputID], nil +} + +// compileGraph converts a SignalDef into an executable runtimeGraph. When the +// def carries an explicit Graph it is used directly; otherwise the legacy +// Inputs+Pipeline form is converted to an equivalent linear graph (see toGraph). +func compileGraph(def SignalDef) (*runtimeGraph, error) { + g := toGraph(def) + if g == nil || len(g.Nodes) == 0 { + return &runtimeGraph{}, nil + } + order, err := topoOrder(g) + if err != nil { + return nil, err + } + rg := &runtimeGraph{outputID: g.Output} + for _, gn := range order { + switch gn.Kind { + case "source": + rg.sources = append(rg.sources, rtSource{id: gn.ID, ref: broker.SignalRef{DS: gn.DS, Name: gn.Signal}}) + case "op": + node, err := buildNode(NodeDef{Type: gn.Op, Params: gn.Params}) + if err != nil { + return nil, fmt.Errorf("node %q: %w", gn.ID, err) + } + rg.order = append(rg.order, &rtNode{id: gn.ID, kind: "op", op: node, state: map[string]any{}, inputs: gn.Inputs}) + case "output": + rg.outputID = gn.ID + rg.order = append(rg.order, &rtNode{id: gn.ID, kind: "output", inputs: gn.Inputs}) + default: + return nil, fmt.Errorf("node %q: unknown kind %q", gn.ID, gn.Kind) + } + } + return rg, nil +} + +// topoOrder returns the graph's nodes in a topological (dependency-first) order, +// treating each node's Inputs as its predecessors. It errors on dangling input +// references or cycles. +func topoOrder(g *Graph) ([]GraphNode, error) { + byID := make(map[string]GraphNode, len(g.Nodes)) + for _, n := range g.Nodes { + byID[n.ID] = n + } + indeg := make(map[string]int, len(g.Nodes)) + succ := make(map[string][]string, len(g.Nodes)) + for _, n := range g.Nodes { + if _, ok := indeg[n.ID]; !ok { + indeg[n.ID] = 0 + } + for _, in := range n.Inputs { + if _, ok := byID[in]; !ok { + return nil, fmt.Errorf("node %q references unknown input %q", n.ID, in) + } + indeg[n.ID]++ + succ[in] = append(succ[in], n.ID) + } + } + // Seed the queue with roots, preserving the node slice order for determinism. + queue := make([]string, 0, len(g.Nodes)) + for _, n := range g.Nodes { + if indeg[n.ID] == 0 { + queue = append(queue, n.ID) + } + } + order := make([]GraphNode, 0, len(g.Nodes)) + for len(queue) > 0 { + id := queue[0] + queue = queue[1:] + order = append(order, byID[id]) + for _, s := range succ[id] { + indeg[s]-- + if indeg[s] == 0 { + queue = append(queue, s) + } + } + } + if len(order) != len(g.Nodes) { + return nil, errors.New("graph contains a cycle") + } + return order, nil +} + +// toGraph returns the DAG for a SignalDef. If def.Graph is set it is returned +// as-is. Otherwise the legacy linear form is converted: each input signal +// becomes a source node, the pipeline becomes a chain of op nodes (the first op +// receiving every source, each later op the previous op's output), terminated by +// an output node. With no pipeline the output takes the first source directly, +// matching the old runPipeline behaviour. +func toGraph(def SignalDef) *Graph { + if def.Graph != nil && len(def.Graph.Nodes) > 0 { + return def.Graph + } + + inputs := def.Inputs + if len(inputs) == 0 && def.DS != "" && def.Signal != "" { + inputs = []InputRef{{DS: def.DS, Signal: def.Signal}} + } + + nodes := make([]GraphNode, 0, len(inputs)+len(def.Pipeline)+1) + srcIDs := make([]string, 0, len(inputs)) + for i, inp := range inputs { + id := fmt.Sprintf("s%d", i) + nodes = append(nodes, GraphNode{ID: id, Kind: "source", DS: inp.DS, Signal: inp.Signal}) + srcIDs = append(srcIDs, id) + } + + opIDs := make([]string, 0, len(def.Pipeline)) + for i, nd := range def.Pipeline { + id := fmt.Sprintf("p%d", i) + var ins []string + if i == 0 { + ins = srcIDs + } else { + ins = []string{opIDs[i-1]} + } + nodes = append(nodes, GraphNode{ID: id, Kind: "op", Op: nd.Type, Params: nd.Params, Inputs: ins}) + opIDs = append(opIDs, id) + } + + var outInputs []string + if len(opIDs) > 0 { + outInputs = []string{opIDs[len(opIDs)-1]} + } else if len(srcIDs) > 0 { + outInputs = []string{srcIDs[0]} + } + nodes = append(nodes, GraphNode{ID: "out", Kind: "output", Inputs: outInputs}) + + return &Graph{Nodes: nodes, Output: "out"} +} diff --git a/internal/datasource/synthetic/graph_test.go b/internal/datasource/synthetic/graph_test.go new file mode 100644 index 0000000..40abe3d --- /dev/null +++ b/internal/datasource/synthetic/graph_test.go @@ -0,0 +1,130 @@ +package synthetic + +import ( + "math" + "testing" +) + +// evalDef compiles a SignalDef and evaluates it against per-source values keyed +// by source node id. +func evalDef(t *testing.T, def SignalDef, srcVals map[string]float64) float64 { + t.Helper() + rg, err := compileGraph(def) + if err != nil { + t.Fatalf("compileGraph: %v", err) + } + out, err := rg.eval(srcVals) + if err != nil { + t.Fatalf("eval: %v", err) + } + return out +} + +// TestGraphMultiInputDAG verifies that an intermediate op can take two +// independently-wired sources — the capability the old linear pipeline lacked. +func TestGraphMultiInputDAG(t *testing.T) { + def := SignalDef{ + Name: "diff", + Graph: &Graph{ + Output: "out", + Nodes: []GraphNode{ + {ID: "a", Kind: "source", DS: "x", Signal: "left"}, + {ID: "b", Kind: "source", DS: "x", Signal: "right"}, + {ID: "sub", Kind: "op", Op: "subtract", Inputs: []string{"a", "b"}}, + {ID: "out", Kind: "output", Inputs: []string{"sub"}}, + }, + }, + } + got := evalDef(t, def, map[string]float64{"a": 10, "b": 3}) + if got != 7 { + t.Errorf("subtract DAG: want 7, got %v", got) + } +} + +// TestGraphExprNamedInputs verifies expr nodes bind arbitrary named inputs in +// wired order. +func TestGraphExprNamedInputs(t *testing.T) { + def := SignalDef{ + Name: "formula", + Graph: &Graph{ + Output: "out", + Nodes: []GraphNode{ + {ID: "a", Kind: "source", DS: "x", Signal: "p"}, + {ID: "b", Kind: "source", DS: "x", Signal: "q"}, + {ID: "e", Kind: "op", Op: "expr", Inputs: []string{"a", "b"}, + Params: map[string]any{"expr": "price * qty", "vars": []any{"price", "qty"}}}, + {ID: "out", Kind: "output", Inputs: []string{"e"}}, + }, + }, + } + got := evalDef(t, def, map[string]float64{"a": 4, "b": 2.5}) + if math.Abs(got-10) > 1e-9 { + t.Errorf("expr named inputs: want 10, got %v", got) + } +} + +// TestGraphFanInToExpr exercises a non-trivial DAG: two ops feeding one expr. +func TestGraphFanInToExpr(t *testing.T) { + def := SignalDef{ + Name: "combo", + Graph: &Graph{ + Output: "out", + Nodes: []GraphNode{ + {ID: "a", Kind: "source", DS: "x", Signal: "p"}, + {ID: "b", Kind: "source", DS: "x", Signal: "q"}, + {ID: "g", Kind: "op", Op: "gain", Inputs: []string{"a"}, Params: map[string]any{"gain": 2.0}}, + {ID: "o", Kind: "op", Op: "offset", Inputs: []string{"b"}, Params: map[string]any{"offset": 1.0}}, + {ID: "e", Kind: "op", Op: "expr", Inputs: []string{"g", "o"}, Params: map[string]any{"expr": "a + b"}}, + {ID: "out", Kind: "output", Inputs: []string{"e"}}, + }, + }, + } + // g = 5*2 = 10 ; o = 4+1 = 5 ; a+b = 15 + got := evalDef(t, def, map[string]float64{"a": 5, "b": 4}) + if math.Abs(got-15) > 1e-9 { + t.Errorf("fan-in DAG: want 15, got %v", got) + } +} + +// TestGraphLegacyConversion verifies the linear Inputs+Pipeline form still +// evaluates correctly via the graph runtime. +func TestGraphLegacyConversion(t *testing.T) { + def := SignalDef{ + Name: "legacy", + DS: "x", + Signal: "p", + Pipeline: []NodeDef{{Type: "gain", Params: map[string]any{"gain": 3.0}}}, + } + rg, err := compileGraph(def) + if err != nil { + t.Fatalf("compileGraph: %v", err) + } + if len(rg.sources) != 1 { + t.Fatalf("want 1 source, got %d", len(rg.sources)) + } + got, err := rg.eval(map[string]float64{rg.sources[0].id: 4}) + if err != nil { + t.Fatalf("eval: %v", err) + } + if got != 12 { + t.Errorf("legacy gain: want 12, got %v", got) + } +} + +// TestGraphCycleRejected ensures a cyclic graph is refused at compile time. +func TestGraphCycleRejected(t *testing.T) { + def := SignalDef{ + Name: "cyclic", + Graph: &Graph{ + Output: "out", + Nodes: []GraphNode{ + {ID: "a", Kind: "op", Op: "gain", Inputs: []string{"b"}, Params: map[string]any{"gain": 1.0}}, + {ID: "b", Kind: "op", Op: "gain", Inputs: []string{"a"}, Params: map[string]any{"gain": 1.0}}, + {ID: "out", Kind: "output", Inputs: []string{"a"}}, + }, + }, + } + if _, err := compileGraph(def); err == nil { + t.Error("expected cycle to be rejected") + } +} diff --git a/internal/datasource/synthetic/synthetic.go b/internal/datasource/synthetic/synthetic.go index 901d0e6..224b86e 100644 --- a/internal/datasource/synthetic/synthetic.go +++ b/internal/datasource/synthetic/synthetic.go @@ -13,16 +13,14 @@ import ( "github.com/uopi/uopi/internal/broker" "github.com/uopi/uopi/internal/datasource" - "github.com/uopi/uopi/internal/dsp" ) const definitionsFile = "synthetic.json" // signalState holds everything needed to run one synthetic signal. type signalState struct { - def SignalDef - nodes []dsp.Node - states []map[string]any // one map per node, persistent across calls + def SignalDef + rg *runtimeGraph // compiled DAG; op-node state persists across evaluations // cancel stops the goroutine driving this signal. cancel context.CancelFunc @@ -123,19 +121,25 @@ func (s *Synthetic) Subscribe(ctx context.Context, signal string, ch chan<- data return nil, datasource.ErrNotFound } - // Collect the upstream references for this signal. - refs := upstreamRefs(st.def) + // Collect the source node references for this signal's DAG. + refs := st.rg.sourceRefs() if len(refs) == 0 { return nil, fmt.Errorf("synthetic: signal %q has no upstream inputs", signal) } + // Source node ids, index-aligned with refs, so updates map to graph inputs. + srcIDs := make([]string, len(st.rg.sources)) + for i, s := range st.rg.sources { + srcIDs[i] = s.id + } ctx, cancel := context.WithCancel(ctx) go func() { defer cancel() - // Latest value per upstream input index. - latest := make([]float64, len(refs)) + // Latest value and timestamp per source node id. + latest := make(map[string]float64, len(refs)) + latestTs := make([]time.Time, len(refs)) ready := make([]bool, len(refs)) // Subscribe to every upstream ref via the broker. @@ -184,7 +188,8 @@ func (s *Synthetic) Subscribe(ctx context.Context, signal string, ch chan<- data return case upd := <-updateCh: - latest[upd.idx] = upd.val + latest[srcIDs[upd.idx]] = upd.val + latestTs[upd.idx] = upd.ts ready[upd.idx] = true // Only compute once we have at least one value for every input. @@ -199,7 +204,19 @@ func (s *Synthetic) Subscribe(ctx context.Context, signal string, ch chan<- data continue } - // Run the pipeline. + // The output is computed from the latest value of every input, so + // its timestamp is the most recent contributing sample time. Using + // the triggering update's timestamp instead would drag the output + // back in time whenever a slow/stale input fired, producing + // non-monotonic or duplicated timestamps on plots. + outTs := latestTs[0] + for _, ts := range latestTs[1:] { + if ts.After(outTs) { + outTs = ts + } + } + + // Evaluate the DAG. s.mu.RLock() cur, stillExists := s.signals[signal] s.mu.RUnlock() @@ -207,14 +224,14 @@ func (s *Synthetic) Subscribe(ctx context.Context, signal string, ch chan<- data return } - result, err := runPipeline(cur.nodes, cur.states, latest) + result, err := cur.rg.eval(latest) if err != nil { s.log.Warn("synthetic: pipeline error", "signal", signal, "err", err) continue } v := datasource.Value{ - Timestamp: upd.ts, + Timestamp: outTs, Data: result, Quality: datasource.QualityGood, } @@ -246,9 +263,9 @@ func (s *Synthetic) AddSignal(def SignalDef) error { return errors.New("signal name must not be empty") } - nodes, err := BuildPipeline(def.Pipeline) + rg, err := compileGraph(def) if err != nil { - return fmt.Errorf("build pipeline: %w", err) + return fmt.Errorf("compile graph: %w", err) } s.mu.Lock() @@ -257,16 +274,7 @@ func (s *Synthetic) AddSignal(def SignalDef) error { return fmt.Errorf("signal %q already exists", def.Name) } - states := make([]map[string]any, len(nodes)) - for i := range states { - states[i] = make(map[string]any) - } - - st := &signalState{ - def: def, - nodes: nodes, - states: states, - } + st := &signalState{def: def, rg: rg} s.signals[def.Name] = st s.mu.Unlock() @@ -320,14 +328,9 @@ func (s *Synthetic) UpdateSignal(def SignalDef) error { return errors.New("signal name must not be empty") } - nodes, err := BuildPipeline(def.Pipeline) + rg, err := compileGraph(def) if err != nil { - return fmt.Errorf("build pipeline: %w", err) - } - - states := make([]map[string]any, len(nodes)) - for i := range states { - states[i] = make(map[string]any) + return fmt.Errorf("compile graph: %w", err) } s.mu.Lock() @@ -339,7 +342,7 @@ func (s *Synthetic) UpdateSignal(def SignalDef) error { if old.cancel != nil { old.cancel() } - s.signals[def.Name] = &signalState{def: def, nodes: nodes, states: states} + s.signals[def.Name] = &signalState{def: def, rg: rg} s.mu.Unlock() if err := s.saveDefs(); err != nil { @@ -402,73 +405,22 @@ func (s *Synthetic) saveDefs() error { return os.WriteFile(s.defsFilePath(), data, 0o644) } -// startSignal builds the pipeline for def and registers the signalState. +// startSignal compiles the DAG for def and registers the signalState. // The actual goroutines are started lazily by Subscribe. func (s *Synthetic) startSignal(def SignalDef) error { - nodes, err := BuildPipeline(def.Pipeline) + rg, err := compileGraph(def) if err != nil { - return fmt.Errorf("build pipeline for %q: %w", def.Name, err) - } - - states := make([]map[string]any, len(nodes)) - for i := range states { - states[i] = make(map[string]any) + return fmt.Errorf("compile graph for %q: %w", def.Name, err) } s.mu.Lock() - s.signals[def.Name] = &signalState{ - def: def, - nodes: nodes, - states: states, - } + s.signals[def.Name] = &signalState{def: def, rg: rg} s.mu.Unlock() s.log.Info("synthetic: signal registered", "name", def.Name) return nil } -// runPipeline executes all nodes in sequence. The output of node N becomes -// input[0] of node N+1. For the first node, inputs is the full upstream slice. -func runPipeline(nodes []dsp.Node, states []map[string]any, inputs []float64) (float64, error) { - if len(nodes) == 0 { - if len(inputs) == 0 { - return 0, nil - } - return inputs[0], nil - } - - // First node receives all upstream inputs. - cur := inputs - var result float64 - var err error - - for i, node := range nodes { - result, err = node.Process(cur, states[i]) - if err != nil { - return 0, fmt.Errorf("node %d (%s): %w", i, node.Type(), err) - } - // Subsequent nodes receive only the single output of the previous node. - cur = []float64{result} - } - return result, nil -} - -// upstreamRefs returns the broker.SignalRef list for a SignalDef. -// If Inputs is set, those take precedence; otherwise DS+Signal is used. -func upstreamRefs(def SignalDef) []broker.SignalRef { - if len(def.Inputs) > 0 { - refs := make([]broker.SignalRef, len(def.Inputs)) - for i, inp := range def.Inputs { - refs[i] = broker.SignalRef{DS: inp.DS, Name: inp.Signal} - } - return refs - } - if def.DS != "" && def.Signal != "" { - return []broker.SignalRef{{DS: def.DS, Name: def.Signal}} - } - return nil -} - // defToMetadata converts a SignalDef into a datasource.Metadata. func defToMetadata(def SignalDef) datasource.Metadata { return datasource.Metadata{ diff --git a/internal/datasource/synthetic/timestamp_test.go b/internal/datasource/synthetic/timestamp_test.go new file mode 100644 index 0000000..778b493 --- /dev/null +++ b/internal/datasource/synthetic/timestamp_test.go @@ -0,0 +1,154 @@ +package synthetic + +import ( + "context" + "log/slog" + "os" + "testing" + "time" + + "github.com/uopi/uopi/internal/broker" + "github.com/uopi/uopi/internal/datasource" +) + +// seqSource is a test DataSource that emits a fixed sequence of values, each +// carrying its own timestamp, so tests can control upstream sample times. +type seqSource struct { + name string + seq []datasource.Value +} + +func (s *seqSource) Name() string { return s.name } +func (s *seqSource) Connect(context.Context) error { return nil } +func (s *seqSource) ListSignals(context.Context) ([]datasource.Metadata, error) { return nil, nil } +func (s *seqSource) GetMetadata(context.Context, string) (datasource.Metadata, error) { + return datasource.Metadata{Name: "x", Type: datasource.TypeFloat64}, nil +} +func (s *seqSource) Write(context.Context, string, any) error { return datasource.ErrNotWritable } +func (s *seqSource) History(context.Context, string, time.Time, time.Time, int) ([]datasource.Value, error) { + return nil, datasource.ErrHistoryUnavailable +} +func (s *seqSource) Subscribe(ctx context.Context, _ string, ch chan<- datasource.Value) (datasource.CancelFunc, error) { + go func() { + for _, v := range s.seq { + select { + case ch <- v: + case <-ctx.Done(): + return + } + time.Sleep(8 * time.Millisecond) + } + }() + return func() {}, nil +} + +// TestSubscribePreservesUpstreamTimestamp verifies a single-source synthetic +// emits each computed value with the upstream sample's timestamp. +func TestSubscribePreservesUpstreamTimestamp(t *testing.T) { + log := slog.New(slog.NewTextHandler(os.Stderr, nil)) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + base := time.Date(2026, 6, 19, 10, 0, 0, 0, time.UTC) + src := &seqSource{name: "src", seq: []datasource.Value{ + {Timestamp: base.Add(1 * time.Second), Data: 1.0, Quality: datasource.QualityGood}, + {Timestamp: base.Add(2 * time.Second), Data: 2.0, Quality: datasource.QualityGood}, + {Timestamp: base.Add(3 * time.Second), Data: 3.0, Quality: datasource.QualityGood}, + }} + + brk := broker.New(ctx, log) + brk.Register(src) + syn := New(t.TempDir(), brk, log) + if err := syn.Connect(ctx); err != nil { + t.Fatal(err) + } + if err := syn.AddSignal(SignalDef{ + Name: "g", DS: "src", Signal: "x", + Pipeline: []NodeDef{{Type: "gain", Params: map[string]any{"gain": 10.0}}}, + }); err != nil { + t.Fatal(err) + } + + ch := make(chan datasource.Value, 8) + if _, err := syn.Subscribe(ctx, "g", ch); err != nil { + t.Fatal(err) + } + + want := []time.Time{base.Add(1 * time.Second), base.Add(2 * time.Second), base.Add(3 * time.Second)} + for i, w := range want { + select { + case v := <-ch: + if !v.Timestamp.Equal(w) { + t.Errorf("emit #%d timestamp: want %s, got %s", i, w, v.Timestamp) + } + case <-time.After(2 * time.Second): + t.Fatalf("timeout waiting for emit #%d", i) + } + } +} + +// TestSubscribeMultiSourceUsesLatestTimestamp verifies that a synthetic combining +// two independent sources stamps each output with the MOST RECENT contributing +// sample time — not the timestamp of whichever source happened to trigger the +// computation. A slow source carrying a stale timestamp must not drag the output +// backwards in time (which previously produced wrong/non-monotonic plot points). +func TestSubscribeMultiSourceUsesLatestTimestamp(t *testing.T) { + log := slog.New(slog.NewTextHandler(os.Stderr, nil)) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + now := time.Date(2026, 6, 19, 12, 0, 0, 0, time.UTC) + // Fast source A with current timestamps. + a := &seqSource{name: "A", seq: []datasource.Value{ + {Timestamp: now.Add(10 * time.Second), Data: 1.0, Quality: datasource.QualityGood}, + {Timestamp: now.Add(11 * time.Second), Data: 2.0, Quality: datasource.QualityGood}, + {Timestamp: now.Add(12 * time.Second), Data: 3.0, Quality: datasource.QualityGood}, + {Timestamp: now.Add(13 * time.Second), Data: 4.0, Quality: datasource.QualityGood}, + }} + // Slow source B: a single sample with a much older timestamp. + b := &seqSource{name: "B", seq: []datasource.Value{ + {Timestamp: now.Add(1 * time.Second), Data: 100.0, Quality: datasource.QualityGood}, + }} + + brk := broker.New(ctx, log) + brk.Register(a) + brk.Register(b) + syn := New(t.TempDir(), brk, log) + if err := syn.Connect(ctx); err != nil { + t.Fatal(err) + } + if err := syn.AddSignal(SignalDef{ + Name: "diff", + Graph: &Graph{Output: "out", Nodes: []GraphNode{ + {ID: "sa", Kind: "source", DS: "A", Signal: "x"}, + {ID: "sb", Kind: "source", DS: "B", Signal: "x"}, + {ID: "sub", Kind: "op", Op: "subtract", Inputs: []string{"sa", "sb"}}, + {ID: "out", Kind: "output", Inputs: []string{"sub"}}, + }}, + }); err != nil { + t.Fatal(err) + } + + ch := make(chan datasource.Value, 16) + if _, err := syn.Subscribe(ctx, "diff", ch); err != nil { + t.Fatal(err) + } + + var last time.Time + for i := 0; i < 4; i++ { + select { + case v := <-ch: + // The stale source-B timestamp (t=1s) must never be used: every output + // is stamped with the newest input time, so emits stay monotonic. + if v.Timestamp.Equal(now.Add(1 * time.Second)) { + t.Errorf("emit #%d used the stale source-B timestamp %s", i, v.Timestamp) + } + if !last.IsZero() && v.Timestamp.Before(last) { + t.Errorf("emit #%d went backwards: %s before previous %s", i, v.Timestamp, last) + } + last = v.Timestamp + case <-time.After(2 * time.Second): + t.Fatalf("timeout waiting for emit #%d", i) + } + } +} diff --git a/internal/dsp/nodes.go b/internal/dsp/nodes.go index 4c22789..37d92cd 100644 --- a/internal/dsp/nodes.go +++ b/internal/dsp/nodes.go @@ -277,16 +277,28 @@ func (n *ThresholdNode) Process(inputs []float64, _ map[string]any) (float64, er // ── ExprNode ────────────────────────────────────────────────────────────────── -// ExprNode evaluates a simple arithmetic expression with variables a, b, c, d -// bound to inputs[0..3]. It uses a hand-written recursive descent parser. +// ExprNode evaluates a simple arithmetic expression. Inputs are bound to named +// variables: Vars[i] -> inputs[i]. When Vars is empty it defaults to a, b, c, d +// (bound to inputs[0..3]) for backward compatibility. It uses a hand-written +// recursive descent parser. type ExprNode struct { Expr string + Vars []string +} + +// defaultVarNames returns the variable names for an expr/lua node: the explicit +// list when set, otherwise the legacy a,b,c,d. +func defaultVarNames(vars []string) []string { + if len(vars) > 0 { + return vars + } + return []string{"a", "b", "c", "d"} } func (n *ExprNode) Type() string { return "expr" } func (n *ExprNode) Process(inputs []float64, _ map[string]any) (float64, error) { vars := map[string]float64{} - names := []string{"a", "b", "c", "d"} + names := defaultVarNames(n.Vars) for i, name := range names { if i < len(inputs) { vars[name] = inputs[i] @@ -507,13 +519,10 @@ func (p *exprParser) parseCall() (float64, error) { } } - // Not a function call — must be a single-letter variable. - if len(name) != 1 { - return 0, fmt.Errorf("unknown identifier %q (use a–d for variables, or a known function name)", name) - } + // Not a function call — must be a declared input variable. val, ok := p.vars[name] if !ok { - return 0, fmt.Errorf("unknown variable %q (allowed: a, b, c, d)", name) + return 0, fmt.Errorf("unknown variable %q (declare it as a named input)", name) } return val, nil } @@ -628,10 +637,12 @@ func (n *LowPassNode) Process(inputs []float64, state map[string]any) (float64, // ── LuaNode ─────────────────────────────────────────────────────────────────── // LuaNode runs a Lua script in a sandboxed gopher-lua VM. -// Inputs are bound to globals a, b, c, d. The script's return value is the output. +// Inputs are bound to globals named by Vars (Vars[i] -> inputs[i]); when Vars is +// empty it defaults to a, b, c, d. The script's return value is the output. // The os, io, package, and debug libraries are disabled. type LuaNode struct { Script string + Vars []string } func (n *LuaNode) Type() string { return "lua" } @@ -682,7 +693,7 @@ func (n *LuaNode) Process(inputs []float64, state map[string]any) (result float6 L.SetTop(0) // Bind inputs. - names := []string{"a", "b", "c", "d"} + names := defaultVarNames(n.Vars) for i, name := range names { if i < len(inputs) { L.SetGlobal(name, lua.LNumber(inputs[i])) diff --git a/internal/server/dialoghub.go b/internal/server/dialoghub.go new file mode 100644 index 0000000..e90e57f --- /dev/null +++ b/internal/server/dialoghub.go @@ -0,0 +1,199 @@ +package server + +import ( + "context" + "encoding/json" + "log/slog" + "strconv" + "strings" + "sync" + + "github.com/uopi/uopi/internal/access" + "github.com/uopi/uopi/internal/audit" + "github.com/uopi/uopi/internal/broker" + "github.com/uopi/uopi/internal/controllogic" + "github.com/uopi/uopi/internal/datasource" +) + +// DialogHub fans control-logic dialog requests out to connected WebSocket +// clients whose identity matches the dialog's user/group filter, and routes +// input responses back to the dialog's target server variable. +// +// It implements controllogic.Notifier; the engine calls Notify when an +// action.dialog node runs. Input dialogs are remembered as pending so a later +// dialogResponse can be correlated by id and validated against its recipient +// filter — this is why panels can write the response target even though direct +// srv writes are otherwise gated to control-logic editors. +type DialogHub struct { + broker *broker.Broker + policy *access.Policy + audit audit.Recorder + log *slog.Logger + + mu sync.Mutex + clients map[*wsClient]struct{} + pending map[string]controllogic.Dialog // input dialogs awaiting a response +} + +// NewDialogHub builds an empty hub. rec is never nil after construction. +func NewDialogHub(brk *broker.Broker, policy *access.Policy, rec audit.Recorder, log *slog.Logger) *DialogHub { + if rec == nil { + rec = audit.Nop() + } + return &DialogHub{ + broker: brk, + policy: policy, + audit: rec, + log: log, + clients: map[*wsClient]struct{}{}, + pending: map[string]controllogic.Dialog{}, + } +} + +func (h *DialogHub) add(c *wsClient) { + h.mu.Lock() + h.clients[c] = struct{}{} + h.mu.Unlock() +} + +func (h *DialogHub) remove(c *wsClient) { + h.mu.Lock() + delete(h.clients, c) + h.mu.Unlock() +} + +// dialogOut is the client-bound JSON form of a dialog request. +type dialogOut struct { + Type string `json:"type"` // always "dialog" + ID string `json:"id"` + Kind string `json:"kind"` // "info" | "error" | "input" + Title string `json:"title,omitempty"` + Message string `json:"message,omitempty"` +} + +// Notify implements controllogic.Notifier: serialise the dialog and push it to +// every connected client matching its user/group filter. +func (h *DialogHub) Notify(d controllogic.Dialog) { + b, err := json.Marshal(dialogOut{ + Type: "dialog", + ID: d.ID, + Kind: d.Kind, + Title: d.Title, + Message: d.Message, + }) + if err != nil { + return + } + + h.mu.Lock() + if d.Kind == "input" && strings.TrimSpace(d.Target) != "" { + h.pending[d.ID] = d + } + var targets []*wsClient + for c := range h.clients { + if h.matches(c.user, d) { + targets = append(targets, c) + } + } + h.mu.Unlock() + + for _, c := range targets { + select { + case c.outCh <- b: + default: + } + } +} + +// matches reports whether user is a recipient of d. An empty user+group filter +// targets everyone; otherwise the user must be named or in a named group. +func (h *DialogHub) matches(user string, d controllogic.Dialog) bool { + if len(d.Users) == 0 && len(d.Groups) == 0 { + return true + } + if h.policy != nil { + user = h.policy.ResolveUser(user) + } + for _, u := range d.Users { + if u == user { + return true + } + } + if len(d.Groups) > 0 && h.policy != nil { + groups := map[string]bool{} + for _, g := range h.policy.GroupsOf(user) { + groups[g] = true + } + for _, g := range d.Groups { + if groups[g] { + return true + } + } + } + return false +} + +// cancel drops a pending input dialog without writing (user dismissed it). +func (h *DialogHub) cancel(id string) { + h.mu.Lock() + delete(h.pending, id) + h.mu.Unlock() +} + +// respond writes an input dialog's response to its target server variable. The +// dialog must be pending and the responding client must have been a recipient; +// this gate replaces the usual srv write-permission check for sanctioned +// control-logic responses. +func (h *DialogHub) respond(ctx context.Context, c *wsClient, id string, value float64) { + h.mu.Lock() + d, ok := h.pending[id] + if ok { + delete(h.pending, id) + } + h.mu.Unlock() + if !ok || !h.matches(c.user, d) { + return + } + + ds, name, ok := parseDialogTarget(d.Target) + if !ok || ds == "local" { + return + } + src, ok := h.broker.Source(ds) + if !ok { + h.log.Warn("dialog response: unknown data source", "ds", ds, "target", d.Target) + return + } + + ev := audit.Event{ + Actor: c.user, + ActorType: audit.ActorUser, + Action: "signal.write", + DS: ds, + Signal: name, + Value: strconv.FormatFloat(value, 'g', -1, 64), + Detail: "control logic dialog response", + IP: c.ip, + Outcome: audit.OutcomeOK, + } + wctx := datasource.WithUser(ctx, c.user) + if err := src.Write(wctx, name, value); err != nil { + h.log.Warn("dialog response: write failed", "ds", ds, "signal", name, "err", err) + ev.Outcome = audit.OutcomeError + ev.Error = err.Error() + } + h.audit.Record(ev) +} + +// parseDialogTarget splits a "ds:name" dialog target on the first ':'. A bare +// name (no ':') defaults to the persistent server-variable source "srv". +func parseDialogTarget(t string) (ds, name string, ok bool) { + t = strings.TrimSpace(t) + if t == "" { + return "", "", false + } + if i := strings.IndexByte(t, ':'); i >= 0 { + return t[:i], t[i+1:], true + } + return "srv", t, true +} diff --git a/internal/server/server.go b/internal/server/server.go index 6e41fcc..73f70dc 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -28,7 +28,7 @@ type Server struct { // New creates the HTTP server, registers all routes, and returns a ready-to-start Server. // synth may be nil if the synthetic data source is not enabled. -func New(addr string, webFS fs.FS, brk *broker.Broker, synth *synthetic.Synthetic, store *storage.Store, policy *access.Policy, acl *panelacl.Store, ctrlLogic *controllogic.Store, ctrlEngine *controllogic.Engine, rec audit.Recorder, channelFinderURL, archiverURL, trustedUserHeader string, log *slog.Logger) *Server { +func New(addr string, webFS fs.FS, brk *broker.Broker, synth *synthetic.Synthetic, store *storage.Store, policy *access.Policy, acl *panelacl.Store, ctrlLogic *controllogic.Store, ctrlEngine *controllogic.Engine, dialogs *DialogHub, rec audit.Recorder, channelFinderURL, archiverURL, trustedUserHeader string, log *slog.Logger) *Server { if rec == nil { rec = audit.Nop() } @@ -41,7 +41,7 @@ func New(addr string, webFS fs.FS, brk *broker.Broker, synth *synthetic.Syntheti }) // WebSocket endpoint - mux.Handle("/ws", &wsHandler{broker: brk, log: log, userHeader: trustedUserHeader, policy: policy, audit: rec}) + mux.Handle("/ws", &wsHandler{broker: brk, log: log, userHeader: trustedUserHeader, policy: policy, audit: rec, dialogs: dialogs}) // Prometheus-format metrics mux.HandleFunc("/metrics", metrics.Handler(brk.ActiveSubscriptions)) diff --git a/internal/server/ws.go b/internal/server/ws.go index 26e1de5..7cf25cf 100644 --- a/internal/server/ws.go +++ b/internal/server/ws.go @@ -34,6 +34,9 @@ type inMsg struct { Name string `json:"name,omitempty"` Value json.RawMessage `json:"value,omitempty"` + // dialogResponse — id of the control-logic dialog being answered. + ID string `json:"id,omitempty"` + // history Start time.Time `json:"start"` End time.Time `json:"end"` @@ -101,6 +104,8 @@ type wsHandler struct { policy *access.Policy // audit records signal writes (never nil; audit.Nop when disabled). audit audit.Recorder + // dialogs fans control-logic dialogs to clients and routes responses. + dialogs *DialogHub } func (h *wsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { @@ -143,12 +148,19 @@ func (h *wsHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { ip: clientIP, policy: h.policy, audit: rec, + dialogs: h.dialogs, outCh: make(chan []byte, 512), updateCh: make(chan broker.Update, 1024), subs: make(map[broker.SignalRef]func()), log: h.log, } + // Register for control-logic dialog pushes for this client's identity. + if h.dialogs != nil { + h.dialogs.add(c) + defer h.dialogs.remove(c) + } + var wg sync.WaitGroup wg.Add(3) go func() { defer wg.Done(); c.readLoop(ctx, cancel) }() @@ -174,8 +186,9 @@ type wsClient struct { log *slog.Logger user string // end-user identity from the trusted proxy header ("" if none) ip string // client address, for audit attribution - policy *access.Policy // global access-level enforcement - audit audit.Recorder // signal-write audit recorder (never nil) + policy *access.Policy // global access-level enforcement + audit audit.Recorder // signal-write audit recorder (never nil) + dialogs *DialogHub // control-logic dialog fan-out (nil if disabled) outCh chan []byte // serialised outgoing messages updateCh chan broker.Update // raw updates from the broker @@ -272,6 +285,8 @@ func (c *wsClient) handleMessage(ctx context.Context, data []byte) { c.handleWrite(ctx, msg) case "history": c.handleHistory(ctx, msg) + case "dialogResponse": + c.handleDialogResponse(ctx, msg) default: c.sendError(ctx, "UNKNOWN_TYPE", "unknown message type: "+msg.Type) } @@ -389,6 +404,25 @@ func (c *wsClient) handleWrite(ctx context.Context, msg inMsg) { c.audit.Record(ev) } +// handleDialogResponse routes the answer to a control-logic input dialog. An +// empty/null value means the user dismissed the dialog (drop it without +// writing); otherwise the numeric value is written to the dialog's target. +func (c *wsClient) handleDialogResponse(ctx context.Context, msg inMsg) { + if c.dialogs == nil || msg.ID == "" { + return + } + if len(msg.Value) == 0 || string(msg.Value) == "null" { + c.dialogs.cancel(msg.ID) + return + } + var num float64 + if err := json.Unmarshal(msg.Value, &num); err != nil { + c.dialogs.cancel(msg.ID) + return + } + c.dialogs.respond(ctx, c, msg.ID, num) +} + func (c *wsClient) handleHistory(ctx context.Context, msg inMsg) { metrics.IncHistoryReqs() ds, ok := c.broker.Source(msg.DS) diff --git a/web/src/App.tsx b/web/src/App.tsx index b76e0fc..609a900 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -4,6 +4,7 @@ import { wsClient } from './lib/ws'; import ViewMode from './ViewMode'; import EditMode from './EditMode'; import FullscreenMode from './FullscreenMode'; +import ControlDialogs from './ControlDialogs'; import { applyZoom, getStoredZoom } from './ZoomControl'; import { AuthContext, DEFAULT_ME, fetchMe, canRead } from './lib/auth'; import type { Interface, Me } from './lib/types'; @@ -69,6 +70,7 @@ export default function App() { return (
+ {wsStatus === 'disconnected' && (
WebSocket disconnected — reconnecting…
)} diff --git a/web/src/ControlDialogs.tsx b/web/src/ControlDialogs.tsx new file mode 100644 index 0000000..3f9412f --- /dev/null +++ b/web/src/ControlDialogs.tsx @@ -0,0 +1,60 @@ +import { h, Fragment } from 'preact'; +import { useState, useEffect } from 'preact/hooks'; +import { wsClient } from './lib/ws'; +import { controlDialogs, dismissControlDialog, type ControlDialog } from './lib/controldialogs'; + +// Renders dialogs pushed by server-side control-logic action.dialog nodes over +// the WebSocket. Mounted once globally (App) since these are addressed to the +// connected user, independent of which panel is open. info/error dialogs are +// acknowledgements; input dialogs send a numeric response back to the server. +export default function ControlDialogs() { + const [reqs, setReqs] = useState([]); + useEffect(() => controlDialogs.subscribe(setReqs), []); + if (reqs.length === 0) return null; + return {reqs.map(r => )}; +} + +function ControlDialogModal({ req }: { req: ControlDialog; key?: string }) { + const isInput = req.kind === 'input'; + const [val, setVal] = useState(''); + + function ok() { + if (isInput) { + const n = parseFloat(val); + wsClient.sendDialogResponse(req.id, Number.isFinite(n) ? n : null); + } + dismissControlDialog(req.id); + } + function cancel() { + if (isInput) wsClient.sendDialogResponse(req.id, null); + dismissControlDialog(req.id); + } + + function onKey(e: KeyboardEvent) { + if (e.key === 'Enter') { e.preventDefault(); ok(); } + if (e.key === 'Escape') { e.preventDefault(); cancel(); } + } + + return ( +
+ +
+ ); +} diff --git a/web/src/ControlLogicEditor.tsx b/web/src/ControlLogicEditor.tsx index 97ce37b..a5ee04e 100644 --- a/web/src/ControlLogicEditor.tsx +++ b/web/src/ControlLogicEditor.tsx @@ -1,6 +1,6 @@ import { h, Fragment } from 'preact'; import { useState, useEffect, useRef } from 'preact/hooks'; -import SearchableSelect from './SearchableSelect'; +import SignalPicker, { SignalOption } from './SignalPicker'; import LuaEditor from './LuaEditor'; import { checkExpr } from './lib/expr'; @@ -18,7 +18,8 @@ type CLNodeKind = | 'action.write' | 'action.delay' | 'action.log' - | 'action.lua'; + | 'action.lua' + | 'action.dialog'; interface CLNode { id: string; @@ -55,6 +56,11 @@ const NODE_W = 11.5 * REM; const PORT_TOP = 2 * REM; const PORT_GAP = 1.375 * REM; const PORT_R = 0.375 * REM; +// Ports sit inside the node's padding box, offset from the node border-box origin +// by the node borders (3px colored top accent, 1px sides). Anchors add the same +// offsets so wires meet the port centers, not their top edges. +const BORDER = 1; +const BORDER_TOP = 3; interface PaletteEntry { kind: CLNodeKind; label: string; params: Record; } @@ -71,6 +77,7 @@ const PALETTE: PaletteEntry[] = [ { kind: 'action.delay', label: 'Delay', params: { ms: '500' } }, { kind: 'action.log', label: 'Log', params: { expr: '', label: '' } }, { kind: 'action.lua', label: 'Lua script', params: { script: '-- get("ds:name") reads, set("ds:name", v) writes,\n-- log("msg") logs to the server.\n' } }, + { kind: 'action.dialog', label: 'Dialog', params: { kind: 'info', title: '', message: '', users: '', groups: '', target: '' } }, ]; const KIND_LABEL: Record = { @@ -86,6 +93,7 @@ const KIND_LABEL: Record = { 'action.delay': 'Delay', 'action.log': 'Log', 'action.lua': 'Lua script', + 'action.dialog': 'Dialog', }; const THRESHOLD_OPS = ['>', '<', '>=', '<=', '==', '!='] as const; @@ -117,11 +125,11 @@ function nodeHeight(kind: CLNodeKind): number { return PORT_TOP + outputs(kind). function genId(): string { return 'n_' + Math.random().toString(36).slice(2, 9); } -function inAnchor(n: CLNode) { return { x: n.x, y: n.y + PORT_TOP }; } +function inAnchor(n: CLNode) { return { x: n.x + BORDER, y: n.y + BORDER_TOP + PORT_TOP }; } function outAnchor(n: CLNode, port: string) { const ports = outputs(n.kind); const i = Math.max(0, ports.findIndex(p => p.id === port)); - return { x: n.x + NODE_W, y: n.y + PORT_TOP + i * PORT_GAP }; + return { x: n.x + NODE_W - BORDER, y: n.y + BORDER_TOP + PORT_TOP + i * PORT_GAP }; } function wirePathStr(x1: number, y1: number, x2: number, y2: number): string { const dx = Math.max(40, Math.abs(x2 - x1) / 2); @@ -384,6 +392,15 @@ function FlowEditor({ graph, onChange }: { return dsSignals[ds] ?? []; } + // Flatten every known ds:name into a single list for the fuzzy SignalPicker, + // and a hook to lazily load all data sources' signals when a picker opens. + function allSignalOptions(): SignalOption[] { + const out: SignalOption[] = []; + for (const ds of dsOptions) for (const name of signalOptions(ds)) out.push({ ds, name }); + return out; + } + function openAllSignals() { dsOptions.forEach(ds => loadSignals(ds)); } + const selected = nodes.find(n => n.id === selectedNode) ?? null; useEffect(() => { @@ -603,19 +620,13 @@ function FlowEditor({ graph, onChange }: {
{KIND_LABEL[selected.kind]}
{(selected.kind === 'trigger.threshold' || selected.kind === 'trigger.change' || selected.kind === 'trigger.alarm') && (() => { - const { ds, name } = splitRef(selected.params.signal ?? ''); return ( -
- - { loadSignals(newDs); patchParams(selected.id, { signal: `${newDs}:` }); }} /> -
- patchParams(selected.id, { signal: `${ds}:${sig}` })} - placeholder={ds ? 'Search signals…' : 'Select data source first'} /> + patchParams(selected.id, { signal: ref })} />
{selected.kind === 'trigger.threshold' && (
@@ -691,7 +702,7 @@ function FlowEditor({ graph, onChange }: { patchParams(selected.id, { cond: v })} onInsert={(ds, sig) => insertRef(selected.id, 'cond', ds, sig)} - dsOptions={dsOptions} signalOptions={signalOptions} loadSignals={loadSignals} + allSignals={allSignalOptions()} onOpenSignals={openAllSignals} hint="True → 'then' port, false → 'else'. e.g. {stub:level} > 5 && temp < 100. {sys:time}=epoch seconds, {sys:dt}=seconds since the trigger last fired." /> )} @@ -715,50 +726,30 @@ function FlowEditor({ graph, onChange }: { patchParams(selected.id, { cond: v })} onInsert={(ds, sig) => insertRef(selected.id, 'cond', ds, sig)} - dsOptions={dsOptions} signalOptions={signalOptions} loadSignals={loadSignals} + allSignals={allSignalOptions()} onOpenSignals={openAllSignals} hint="Loops the 'body' port while true (capped). e.g. {stub:level} < 90" /> )}

Runs the body port repeatedly, then continues on done.

)} - {selected.kind === 'action.write' && (() => { - const { ds, name } = splitRef(selected.params.target ?? ''); - return ( - -
- - { loadSignals(newDs); patchParams(selected.id, { target: `${newDs}:` }); }} /> -
-
- - {(ds === 'local' || ds === 'srv') ? ( - - patchParams(selected.id, { target: `${ds}:${(e.target as HTMLInputElement).value}` })} /> - - {signalOptions(ds).map(o => -

{ds === 'srv' - ? 'Server variables are visible to interface panels (read-only unless the user has control-logic access).' - : 'Local variables live only inside this graph and reset on reload.'}

- - ) : ( - patchParams(selected.id, { target: `${ds}:${sig}` })} - placeholder={ds ? 'Search signals…' : 'Select data source first'} /> - )} -
- patchParams(selected.id, { expr: v })} - onInsert={(ds2, sig) => insertRef(selected.id, 'expr', ds2, sig)} - dsOptions={dsOptions} signalOptions={signalOptions} loadSignals={loadSignals} - hint="Write to a data source signal, a bare local var, or srv:name for a server variable panels can read. e.g. ({stub:a} - {stub:b}) / {sys:dt}." /> - - ); - })()} + {selected.kind === 'action.write' && ( + +
+ + patchParams(selected.id, { target: ref })} /> +

Pick a signal, or type local:name for a graph-local var + or srv:name for a server variable that panels can read.

+
+ patchParams(selected.id, { expr: v })} + onInsert={(ds2, sig) => insertRef(selected.id, 'expr', ds2, sig)} + allSignals={allSignalOptions()} onOpenSignals={openAllSignals} + hint="Write to a data source signal, a bare local var, or srv:name for a server variable panels can read. e.g. ({stub:a} - {stub:b}) / {sys:dt}." /> + + )} {selected.kind === 'action.delay' && (
@@ -779,7 +770,7 @@ function FlowEditor({ graph, onChange }: { patchParams(selected.id, { expr: v })} onInsert={(ds2, sig) => insertRef(selected.id, 'expr', ds2, sig)} - dsOptions={dsOptions} signalOptions={signalOptions} loadSignals={loadSignals} + allSignals={allSignalOptions()} onOpenSignals={openAllSignals} hint="Logs this value to the server log — handy for debugging a flow." /> )} @@ -797,6 +788,54 @@ function FlowEditor({ graph, onChange }: {
)} + {selected.kind === 'action.dialog' && ( + +
+ + +
+
+ + patchParams(selected.id, { title: (e.target as HTMLInputElement).value })} /> +
+
+ +