trigger_test.go

package ruleflow

import (
	"context"
	"errors"
	"sync"
	"testing"
	"time"
)

func TestEmitNoMatch(t *testing.T) {
	js := []byte(`[
		{"id":"trig","type":"trigger","config":{"message":"device-report"},"outputs":{"ids":["a1"]}},
		{"id":"a1","type":"action","config":{"kind":"http"}}
	]`)
	ma := &mockAction{}
	ActionHandler = ma.Do
	f := newFlow()
	f.load(js)
	// 投递不匹配的消息
	f.Emit("click", nil)
	if len(ma.calls) != 0 {
		t.Fatalf("不匹配消息不应触发, got %d", len(ma.calls))
	}
}

func TestEmitMultipleTriggersSameMessage(t *testing.T) {
	js := []byte(`[
		{"id":"t1","type":"trigger","config":{"message":"device-report"},"outputs":{"ids":["a1"]}},
		{"id":"t2","type":"trigger","config":{"message":"device-report"},"outputs":{"ids":["a2"]}},
		{"id":"a1","type":"action","config":{"kind":"http"}},
		{"id":"a2","type":"action","config":{"kind":"mqtt"}}
	]`)
	ma := &mockAction{}
	ActionHandler = ma.Do
	f := newFlow()
	f.load(js)
	f.Emit("device-report", 1)
	if len(ma.calls) != 2 {
		t.Fatalf("同消息两个触发器都应激活, got %d", len(ma.calls))
	}
}

func TestSecondsParse(t *testing.T) {
	if seconds(float64(60)) != 60 || seconds(60) != 60 || seconds("30") != 30 || seconds("x") != 0 || seconds(nil) != 0 {
		t.Error("seconds 解析错误")
	}
}

// 计数动作(并发安全)
type countAction struct {
	mu sync.Mutex
	n  int
}

func (c *countAction) Do(msg string, params map[string]any) (any, error) {
	c.mu.Lock()
	c.n++
	c.mu.Unlock()
	return nil, nil
}
func (c *countAction) count() int {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.n
}

func TestStartInterval(t *testing.T) {
	// interval 用秒，测试用 1 秒过粗；这里直接验证 ctx 取消能停止(不实际等待多次触发)
	js := []byte(`[
		{"id":"t","type":"trigger","config":{"interval":1},"outputs":{"ids":["a"]}},
		{"id":"a","type":"action","config":{"kind":"http"}}
	]`)
	ca := &countAction{}
	ActionHandler = ca.Do
	f := newFlow()
	f.load(js)

	ctx, cancel := context.WithCancel(context.Background())
	f.Start(ctx)
	time.Sleep(1200 * time.Millisecond) // 约触发 1 次
	cancel()
	got := ca.count()
	if got < 1 {
		t.Fatalf("定时器应至少触发 1 次, got %d", got)
	}
	// 取消后不再增长
	time.Sleep(1200 * time.Millisecond)
	if ca.count() != got {
		t.Errorf("ctx 取消后不应再触发, before=%d after=%d", got, ca.count())
	}
}

func TestStartCountdown(t *testing.T) {
	js := []byte(`[
		{"id":"t","type":"trigger","config":{"countdown":1},"outputs":{"ids":["a"]}},
		{"id":"a","type":"action","config":{"kind":"http"}}
	]`)
	ca := &countAction{}
	ActionHandler = ca.Do
	f := newFlow()
	f.load(js)

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	f.Start(ctx)
	time.Sleep(1300 * time.Millisecond)
	if ca.count() != 1 {
		t.Fatalf("倒计时应只触发 1 次, got %d", ca.count())
	}
	// 再等不应重复触发
	time.Sleep(1200 * time.Millisecond)
	if ca.count() != 1 {
		t.Errorf("倒计时不应重复触发, got %d", ca.count())
	}
}

// cron 触发器：每秒触发一次(6 字段带秒)，ctx 取消后停止。
func TestCronTrigger(t *testing.T) {
	js := []byte(`[
		{"id":"t","type":"trigger","config":{"cron":"*/1 * * * * *"},"outputs":{"ids":["a"]}},
		{"id":"a","type":"action","config":{"kind":"http"}}
	]`)
	ca := &countAction{}
	ActionHandler = ca.Do
	f := newFlow()
	f.load(js)

	ctx, cancel := context.WithCancel(context.Background())
	f.Start(ctx)
	time.Sleep(2200 * time.Millisecond) // 约触发 2 次
	cancel()
	got := ca.count()
	if got < 1 {
		t.Fatalf("cron 每秒应至少触发 1 次, got %d", got)
	}
	// 取消后不再增长
	time.Sleep(1200 * time.Millisecond)
	if ca.count() != got {
		t.Errorf("ctx 取消后不应再触发, before=%d after=%d", got, ca.count())
	}
}

// 定时类触发器的 msg.value 恒为 nil，流到下游不应顶掉下游参数的缺省值。
// 用 Trigger(nil) 而非等真实 cron：fire 与 Trigger(nil) 走的是同一个 entryInbound(n, nil)。
//
// 回归的是一整条静默失效的链路：cron → randInt(min=0,max=10) → 动作。
// nil 覆盖缺省值时 randInt 的 min 被顶成 nil，argFloat 报"参数[0] 非数值"使函数节点
// 走异常出口，下游动作根本不执行——现场表现为"定时器在跑但设备属性毫无变化"。
func TestTriggerNilKeepsDownstreamDefaults(t *testing.T) {
	js := []byte(`[
		{"id":"t","type":"trigger","config":{"cron":"0/5 * * * * *"},"inputs":[],
		 "outputs":{"ids":[{"id":"rnd","key":"min"}]}},
		{"id":"rnd","type":"function","config":{"fn":"randInt"},
		 "inputs":[{"key":"min","value":0},{"key":"max","value":10}],
		 "outputs":{"ids":[{"id":"a","key":"value"}]}},
		{"id":"a","type":"action","config":{"kind":"device-update-property"},
		 "inputs":[{"key":"value","value":12}]}
	]`)
	ma := &mockAction{}
	ActionHandler = ma.Do
	var nodeErr error
	ErrorHandler = func(_ string, err error) { nodeErr = err }
	defer func() { ErrorHandler = nil }()

	f := newFlow()
	if err := f.load(js); err != nil {
		t.Fatal(err)
	}
	if err := f.Trigger("t", nil); err != nil {
		t.Fatal(err)
	}
	if nodeErr != nil {
		t.Fatalf("链路不应有节点异常, got %v", nodeErr)
	}
	if len(ma.calls) != 1 {
		t.Fatalf("动作应被调用一次(链路未断), got %d 次", len(ma.calls))
	}
	// randInt 拿到 min=0/max=10 正常出值，动作的 value 参数被该输出覆盖
	got, ok := toFloat(ma.calls[0].inputs["value"])
	if !ok {
		t.Fatalf("动作应收到 randInt 的数值输出, got %#v", ma.calls[0].inputs["value"])
	}
	if got < 0 || got > 10 {
		t.Errorf("randInt 输出应在 [0,10], got %v", got)
	}
}

// cron 表达式非法：不启动、不触发、不 panic。
func TestCronTriggerInvalid(t *testing.T) {
	js := []byte(`[
		{"id":"t","type":"trigger","config":{"cron":"not a cron"},"outputs":{"ids":["a"]}},
		{"id":"a","type":"action","config":{"kind":"http"}}
	]`)
	ca := &countAction{}
	ActionHandler = ca.Do
	f := newFlow()
	f.load(js)
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	f.Start(ctx)
	time.Sleep(300 * time.Millisecond)
	if ca.count() != 0 {
		t.Fatalf("非法 cron 不应触发, got %d", ca.count())
	}
}

// 设备/消息类触发器：监听 device-report/event/connect/disconnect/disable、click、自定义消息名，
// Emit 对应消息各触发下游一次。
func TestDeviceMessageTriggers(t *testing.T) {
	messages := []string{
		"device-report", "device-event", "device-connect",
		"device-disconnect", "device-disable", "click", "my-custom-msg",
	}
	for _, msg := range messages {
		t.Run(msg, func(t *testing.T) {
			js := []byte(`[
				{"id":"trig","type":"trigger","config":{"message":"` + msg + `"},"outputs":{"ids":["act"]}},
				{"id":"act","type":"action","config":{"kind":"message"}}
			]`)
			ma := &mockAction{}
			ActionHandler = ma.Do
			f := newFlow()
			f.load(js)
			f.Emit(msg, 1)
			if len(ma.calls) != 1 {
				t.Fatalf("消息 %q 应触发下游一次, got %d", msg, len(ma.calls))
			}
			// 不匹配的消息不触发
			ma.calls = nil
			f.Emit("other-message", 1)
			if len(ma.calls) != 0 {
				t.Fatalf("不匹配消息不应触发, got %d", len(ma.calls))
			}
		})
	}
}

// 手动触发的场景：两个手动触发器各接一个动作，用于验证按节点 id 精确触发。
// 这正是 Emit 做不到的——Emit 按 config.message 匹配，两个都会被激活。
const manualScene = `[
	{"id":"m1","type":"trigger","config":{},"outputs":{"ids":["a1"]}},
	{"id":"m2","type":"trigger","config":{},"outputs":{"ids":["a2"]}},
	{"id":"a1","type":"action","config":{"kind":"message","text":"一号"}},
	{"id":"a2","type":"action","config":{"kind":"message","text":"二号"}}
]`

// 场景未运行(未 Start / 已 Stop)时 TriggerNode 返回 ErrSceneNotRunning，
// 与"节点不存在"区分：前者是部署/状态问题，后者是规则链数据问题。
func TestTriggerNodeNotRunning(t *testing.T) {
	cleanupReg(t)
	if err := TriggerNode("从未注册", "m1", nil); !errors.Is(err, ErrSceneNotRunning) {
		t.Fatalf("未注册场景应返回 ErrSceneNotRunning, got %v", err)
	}
	if err := Start("s-manual", []byte(manualScene)); err != nil {
		t.Fatal(err)
	}
	Stop("s-manual")
	if err := TriggerNode("s-manual", "m1", nil); !errors.Is(err, ErrSceneNotRunning) {
		t.Fatalf("已 Stop 的场景应返回 ErrSceneNotRunning, got %v", err)
	}
}

// 按节点 id 精确触发：只有被点的那个手动节点的下游执行。
func TestTriggerNodeByID(t *testing.T) {
	cleanupReg(t)
	ma := &mockAction{}
	ActionHandler = ma.Do
	if err := Start("s-manual", []byte(manualScene)); err != nil {
		t.Fatal(err)
	}
	if err := TriggerNode("s-manual", "m2", nil); err != nil {
		t.Fatal(err)
	}
	if len(ma.calls) != 1 {
		t.Fatalf("应只触发一个动作, got %d", len(ma.calls))
	}
	if got := ma.calls[0].config["text"]; got != "二号" {
		t.Errorf("应触发 m2 的下游动作, got %v", got)
	}
	// 节点不存在：报错而非静默成功，否则编辑器点了没反应无从排查
	err := TriggerNode("s-manual", "不存在的节点", nil)
	if err == nil || errors.Is(err, ErrSceneNotRunning) {
		t.Fatalf("节点不存在应返回节点级错误, got %v", err)
	}
}

// TriggerNode 不限节点类型：可从链路中间任一节点起跑，Value 落该节点第一个参数。
// 调长链时不必每次从头喂数据，这是手动触发之外白送的调试能力。
func TestTriggerNodeMidChain(t *testing.T) {
	cleanupReg(t)
	js := []byte(`[
		{"id":"m","type":"trigger","config":{},"outputs":{"ids":[{"id":"f","key":"a"}]}},
		{"id":"f","type":"function","config":{"fn":"add"},
		 "inputs":[{"key":"a","label":"加数","value":0},{"key":"b","label":"被加数","value":8}],
		 "outputs":{"ids":[{"id":"act","key":"value"}]}},
		{"id":"act","type":"action","config":{"kind":"message"},"inputs":[{"key":"value","label":"值"}]}
	]`)
	ma := &mockAction{}
	ActionHandler = ma.Do
	if err := Start("s-mid", js); err != nil {
		t.Fatal(err)
	}
	// 直接从中间的 function 节点起跑，42 落它的第一个参数 a
	if err := TriggerNode("s-mid", "f", &TriggerInput{Value: 42}); err != nil {
		t.Fatal(err)
	}
	if len(ma.calls) != 1 {
		t.Fatalf("下游动作应执行一次, got %d", len(ma.calls))
	}
	if got, _ := toFloat(ma.calls[0].inputs["value"]); got != 50 {
		t.Errorf("应为 42+8=50, got %v", ma.calls[0].inputs["value"])
	}
}

// midChainScene 从中间起跑用的两参数场景：f 把 msg.value 记进 payload 再返回 a+b，
// 一次就能同时看清 TriggerInput 三个字段各落到了哪里。b 的缺省值 8 用于验"未喂到的参数"。
const midChainScene = `[
	{"id":"f","type":"function","config":{"script":"payload.gotValue = value; return a + b"},
	 "inputs":[{"key":"a","label":"加数","value":0},{"key":"b","label":"被加数","value":8}],
	 "outputs":{"ids":[{"id":"act","key":"value"}]}},
	{"id":"act","type":"action","config":{"kind":"message"},"inputs":[{"key":"value","label":"值"}]}
]`

// Inputs 按 Input.Key 精确落槽，且优先于 Value 落进第一个槽的那个值。
// 这是多参数节点从中间起跑的唯一可靠喂法——只给 Value 的话另一个参数会去取
// 上次运行残留的缓存值(见 store.take)，同一次触发的结果不可重现。
func TestTriggerInputBySlotKey(t *testing.T) {
	ma := &mockAction{}
	ActionHandler = ma.Do
	f := newFlow()
	if err := f.load([]byte(midChainScene)); err != nil {
		t.Fatal(err)
	}
	if err := f.Trigger("f", &TriggerInput{Value: 1, Inputs: map[string]any{"a": 3, "b": 4}}); err != nil {
		t.Fatal(err)
	}
	if len(ma.calls) != 1 {
		t.Fatalf("下游动作应执行一次, got %d", len(ma.calls))
	}
	if got, _ := toFloat(ma.calls[0].inputs["value"]); got != 7 {
		t.Errorf("Inputs 应覆盖 Value 落的第一个槽(3+4=7), got %v", ma.calls[0].inputs["value"])
	}
	// Value 照旧落 msg.value——被 Inputs 覆盖的只是槽，不是 value 本身
	if got, _ := toFloat(ma.calls[0].payload["gotValue"]); got != 1 {
		t.Errorf("msg.value 应为 Value 的原值 1, got %v", ma.calls[0].payload["gotValue"])
	}
}

// Inputs 里匹配不到参数的 key 与 nil 值都忽略：前者与 deliver 的落槽规则一致，
// 后者与引擎各处"nil = 上游没有值可给"的处理一致——都不该顶掉参数自己的缺省值。
func TestTriggerInputIgnoresUnknownKeyAndNil(t *testing.T) {
	ma := &mockAction{}
	ActionHandler = ma.Do
	f := newFlow()
	if err := f.load([]byte(midChainScene)); err != nil {
		t.Fatal(err)
	}
	in := &TriggerInput{Inputs: map[string]any{"a": 5, "查无此参数": 99, "b": nil}}
	if err := f.Trigger("f", in); err != nil {
		t.Fatal(err)
	}
	if got, _ := toFloat(ma.calls[0].inputs["value"]); got != 13 {
		t.Errorf("b 应保留缺省值 8(5+8=13), got %v", ma.calls[0].inputs["value"])
	}
}

// Payload 作流作用域初值，让"从链路中间起跑"能还原上游写进 payload 的现场；
// 且进流的是拷贝——链上节点写 payload 不会改到调用方的 map，两次触发也互不串。
func TestTriggerInputPayload(t *testing.T) {
	js := []byte(`[
		{"id":"f","type":"function","config":{"script":"payload.mark = (payload.mark || 0) + 1; return 0"},
		 "outputs":{"ids":["act"]}},
		{"id":"act","type":"action","config":{"kind":"message"}}
	]`)
	ma := &mockAction{}
	ActionHandler = ma.Do
	f := newFlow()
	if err := f.load(js); err != nil {
		t.Fatal(err)
	}
	caller := map[string]any{"deviceId": "d-1"}
	for i := 0; i < 2; i++ {
		if err := f.Trigger("f", &TriggerInput{Payload: caller}); err != nil {
			t.Fatal(err)
		}
	}
	if len(ma.calls) != 2 {
		t.Fatalf("应触发两次, got %d", len(ma.calls))
	}
	for i, c := range ma.calls {
		if c.payload["deviceId"] != "d-1" {
			t.Errorf("第 %d 次: 下游应看到初始化的流作用域, got %v", i+1, c.payload)
		}
		// 每次都是 1 才说明拿到的是各自的拷贝：共享同一张 map 的话第二次会累加成 2
		if got, _ := toFloat(c.payload["mark"]); got != 1 {
			t.Errorf("第 %d 次: 两次触发的 payload 应互相隔离, got %v", i+1, c.payload["mark"])
		}
	}
	if _, ok := caller["mark"]; ok {
		t.Errorf("调用方的 map 不应被链上节点写脏, got %v", caller)
	}
}

// 三个字段全零值(以及 in 为 nil)等同"只触发、不喂数据"：参数取自己的缺省值、
// payload 为空对象。定时/cron 触发器走的就是这条路(fire 传 nil)。
func TestTriggerInputZeroValue(t *testing.T) {
	ma := &mockAction{}
	ActionHandler = ma.Do
	f := newFlow()
	if err := f.load([]byte(midChainScene)); err != nil {
		t.Fatal(err)
	}
	for _, in := range []*TriggerInput{nil, {}} {
		ma.calls = nil
		if err := f.Trigger("f", in); err != nil {
			t.Fatal(err)
		}
		if got, _ := toFloat(ma.calls[0].inputs["value"]); got != 8 {
			t.Errorf("in=%v: 应全取缺省值(0+8=8), got %v", in, ma.calls[0].inputs["value"])
		}
		if ma.calls[0].payload["gotValue"] != nil {
			t.Errorf("in=%v: msg.value 应为 nil, got %v", in, ma.calls[0].payload["gotValue"])
		}
	}
}