trigger.go

package ruleflow

import (
	"context"
	"encoding/json"
	"strconv"
	"time"

	"github.com/robfig/cron/v3"
)

// cronParser 解析定时器触发器的 cron 表达式：SecondOptional 兼容 5 字段标准 cron
// (分 时 日 月 周，最小粒度 1 分钟) 与 6 字段带秒 (秒 分 时 日 月 周)。
var cronParser = cron.NewParser(
	cron.SecondOptional | cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow,
)

// marshal 把任意值序列化为 JSON 字节。
func marshal(v any) ([]byte, error) {
	return json.Marshal(v)
}

// Emit 投递一条消息，激活所有监听该消息的触发器节点，向其下游流转。
// 消息名如 "device-report" / "device-event" / "device-connect" / "click" / 自定义。
// 触发器节点通过 config.message 声明监听的消息名。
//
// data 整体作为触发器的 msg.value(消息类触发器的 value 就是整个消息对象)。
// data 为对象且含 deviceId 时，参与触发器 config.deviceId 的过滤。
// 每次投递都是一次新的传播：payload 从 {} 开始。
func (f *Flow) Emit(message string, data any) {
	f.emitWithin(newExecContext(), message, data)
}

// emitWithin 在给定因果链上投递消息。emit 动作走这里：被激活的链路是一次新的传播
// (payload 从 {} 开始、传播路径从头计起)，但沿用同一条因果链的消息名集合，
// 使 emit 成环(m1 → 触发器 → 再发 m1)能收敛。
func (f *Flow) emitWithin(ctx *execContext, message string, data any) {
	if !ctx.enterEmit(message) {
		return // 本链上正投递着该消息：拦住，避免无限循环
	}
	defer ctx.leaveEmit(message)
	f.mu.RLock()
	ids := make([]string, len(f.order))
	copy(ids, f.order)
	f.mu.RUnlock()

	deviceId := deviceIDOf(data)
	for _, id := range ids {
		n := f.node(id)
		if n == nil || n.Type != TypeTrigger {
			continue
		}
		if n.ConfigString("message") != message {
			continue
		}
		// trigger 声明了 deviceId 时，只接匹配设备的消息
		if want := n.ConfigString("deviceId"); want != "" && want != deviceId {
			continue
		}
		_ = f.execNode(ctx.derive(), n, f.entryInbound(n, &TriggerInput{Value: data}))
	}
}

// deviceIDOf 从消息体里取 deviceId(消息为扁平对象，deviceId 是其中一个业务字段)。
// 非对象或无该字段时返回空串。
func deviceIDOf(data any) string {
	m, ok := data.(map[string]any)
	if !ok {
		return ""
	}
	s, _ := m["deviceId"].(string)
	return s
}

// emitMessage 执行 emit 动作：按节点配置组装一条扁平消息并投递到本场景。
//
// 字段按顺序合并、后者覆盖前者：
//  1. name = config.message(消息名)
//  2. config.data 的各字段(静态内容)
//  3. 本节点各输入参数(按 key 索引，无 key 的参数跳过)——动态内容从这里带出
//
// 被激活的链路是一次新的传播(payload 从 {} 开始)，不继承发送方的流作用域数据——
// 这也是跨分支传数据的推荐方式。
func (f *Flow) emitMessage(ctx *execContext, n *Node, in *inbound) {
	name := n.ConfigString("message")
	msg := map[string]any{"name": name}
	if data, ok := n.Config["data"].(map[string]any); ok {
		for k, v := range data {
			msg[k] = v
		}
	}
	byKey, _ := resolveInputs(n, in)
	for k, v := range byKey {
		msg[k] = v
	}
	f.emitWithin(ctx, name, msg)
}

// Start 启动定时/倒计时类触发器，ctx 取消即停。非阻塞(内部起 goroutine)。
// 定时器: config.interval(秒) 周期触发；倒计时: config.countdown(秒) 到点触发一次。
func (f *Flow) Start(ctx context.Context) {
	f.mu.RLock()
	ids := make([]string, len(f.order))
	copy(ids, f.order)
	f.mu.RUnlock()

	for _, id := range ids {
		n := f.node(id)
		if n == nil || n.Type != TypeTrigger {
			continue
		}
		if n.ConfigString("source") != "" {
			continue // 第三方触发器由 StartTriggers 的订阅器驱动，interval 归其解释
		}
		cfg := n.Config
		if expr := n.ConfigString("cron"); expr != "" {
			go f.runCron(ctx, n, expr)
		} else if iv := seconds(cfg["interval"]); iv > 0 {
			go f.runInterval(ctx, n, iv)
		} else if cd := seconds(cfg["countdown"]); cd > 0 {
			go f.runCountdown(ctx, n, cd)
		}
	}
}

// fire 定时/倒计时类触发器到点：新建一条 msg(value 为 nil)开始一次新的传播。
func (f *Flow) fire(n *Node) {
	_ = f.execNode(newExecContext(), n, f.entryInbound(n, nil))
}

func (f *Flow) runInterval(ctx context.Context, n *Node, sec int) {
	t := time.NewTicker(time.Duration(sec) * time.Second)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
			f.fire(n)
		}
	}
}

// runCron 按 cron 表达式定时触发节点 n，ctx 取消即停。解析失败静默返回。
func (f *Flow) runCron(ctx context.Context, n *Node, expr string) {
	sched, err := cronParser.Parse(expr)
	if err != nil {
		return // cron 表达式非法：不启动
	}
	for {
		next := sched.Next(time.Now())
		if next.IsZero() {
			return // 无后续触发时刻
		}
		t := time.NewTimer(time.Until(next))
		select {
		case <-ctx.Done():
			t.Stop()
			return
		case <-t.C:
			f.fire(n)
		}
	}
}

func (f *Flow) runCountdown(ctx context.Context, n *Node, sec int) {
	t := time.NewTimer(time.Duration(sec) * time.Second)
	defer t.Stop()
	select {
	case <-ctx.Done():
	case <-t.C:
		f.fire(n)
	}
}

// seconds 把 config 值转为秒(支持数字/数字字符串)。
func seconds(v any) int {
	switch x := v.(type) {
	case float64:
		return int(x)
	case int:
		return x
	case string:
		n, _ := strconv.Atoi(x)
		return n
	default:
		return 0
	}
}