orbit/pulse/schedule/rule.go

// Package schedule 计算计划触发点和有限的漏跑决策；不运行内存 Cron 或业务处理器。
package schedule

import (
	"errors"
	"strings"
	"time"

	"git.yuntuo.hk/k2safe/orbit/internal/timezone"

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

// Rule 是不可变计划版本中的时间规则。所有边界精确到 UTC 微秒。
type Rule struct {
	CalendarVersion string     `json:"calendar_version"`      // CalendarVersion 冻结的时区快照版本；提交为空时采用 v1 固定默认快照，升级不能静默改变。
	Kind            string     `json:"kind"`                  // Kind CRON 或 FIXED_RATE；不支持以完成时间推进的 fixed-delay。
	Timezone        string     `json:"timezone"`              // Timezone 显式 IANA 名称，禁止依赖节点 Local 设置。
	StartAt         time.Time  `json:"start_at"`              // StartAt 包含此边界；固定间隔以此为唯一基准。
	EndAt           *time.Time `json:"end_at,omitempty"`      // EndAt 排除此边界；为空时运行至支持的日期上限。
	Cron            string     `json:"cron,omitempty"`        // Cron 五段表达式：分钟、小时、月日、月、星期。
	IntervalMS      uint64     `json:"interval_ms,omitempty"` // IntervalMS 固定间隔毫秒数，范围 1000–2592000000。
}

// Calendar 持有已验证且不可变的计算规则，可并发读取，不包含当前进度。
type Calendar struct {
	rule Rule
	zone *time.Location
	cron *cron.SpecSchedule
}

func validTime(t time.Time) bool {
	return t.Year() >= 2000 && t.Year() <= 9998 && t.Nanosecond()%1000 == 0
}

// Compile 规范化规则并拒绝无解、混合字段和隐式本地时区。
func Compile(in Rule) (*Calendar, error) {
	in.StartAt = in.StartAt.UTC()
	if !validTime(in.StartAt) {
		return nil, errors.New("start_at must use UTC-compatible microseconds and years 2000 to 9998")
	}
	if in.EndAt != nil {
		end := in.EndAt.UTC()
		if !validTime(end) || !end.After(in.StartAt) {
			return nil, errors.New("end_at must be after start_at")
		}
		in.EndAt = &end
	}
	if in.Timezone == "" || in.Timezone == "Local" || len(in.Timezone) > 64 {
		return nil, errors.New("explicit IANA timezone required")
	}
	if in.CalendarVersion == "" {
		in.CalendarVersion = timezone.Version
	}
	zone, err := timezone.Load(in.Timezone, in.CalendarVersion)
	if err != nil {
		return nil, errors.New("invalid IANA timezone")
	}
	c := &Calendar{rule: in, zone: zone}
	switch in.Kind {
	case "FIXED_RATE":
		if in.Cron != "" || in.IntervalMS < 1000 || in.IntervalMS > 2592000000 {
			return nil, errors.New("fixed rate requires interval_ms 1000 to 2592000000 and no cron")
		}
	case "CRON":
		fields := strings.Fields(in.Cron)
		if in.IntervalMS != 0 || len(in.Cron) > 256 || len(fields) != 5 || strings.HasPrefix(fields[0], "@") || strings.Contains(fields[0], "=") {
			return nil, errors.New("cron requires exactly five fields and no interval or embedded timezone")
		}
		c.rule.Cron = strings.Join(fields, " ")
		parsed, err := cron.NewParser(cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow).Parse(c.rule.Cron)
		if err != nil {
			return nil, errors.New("invalid five-field cron")
		}
		c.cron = parsed.(*cron.SpecSchedule)
		// 仅用上游计算 UTC 日历；实际时区转换由恒定偏移区间处理，避免非整小时夏令时跳变遗漏。
		c.cron.Location = time.UTC
	default:
		return nil, errors.New("rule kind must be CRON or FIXED_RATE")
	}
	first, err := c.AtOrAfter(in.StartAt)
	if err != nil {
		return nil, err
	}
	if first == nil {
		return nil, errors.New("rule has no occurrence in its time bounds")
	}
	return c, nil
}

// Rule 返回独立规则副本，不暴露内部可变时间指针。
func (c *Calendar) Rule() Rule {
	out := c.rule
	if out.EndAt != nil {
		v := *out.EndAt
		out.EndAt = &v
	}
	return out
}

// AtOrAfter 返回不早于指定时刻的第一个原计划触发点；无后续返回 nil。
// DST 缺失时刻跳过，重复时刻分别按两个 UTC 身份触发；固定间隔按真实经过时间推进。
func (c *Calendar) AtOrAfter(at time.Time) (*time.Time, error) {
	at = at.UTC()
	if at.Before(c.rule.StartAt) {
		at = c.rule.StartAt
	}
	end := time.Date(9999, 1, 1, 0, 0, 0, 0, time.UTC)
	if c.rule.EndAt != nil {
		end = *c.rule.EndAt
	}
	if !at.Before(end) {
		return nil, nil
	}
	if c.cron == nil {
		// 使用微秒整数而非 time.Sub，避免跨越 292 年时 Duration 饱和。
		start := c.rule.StartAt.UnixMicro()
		step := int64(c.rule.IntervalMS) * 1000
		delta := at.UnixMicro() - start
		n := delta / step
		if delta%step != 0 || at.Nanosecond()%1000 != 0 {
			n++
		}
		next := time.UnixMicro(start + n*step).UTC()
		if !next.Before(end) {
			return nil, nil
		}
		return &next, nil
	}
	// 五段 Gregorian 规则的最大日期间隔为世纪闰年边界的八年；十年是计算保护界。
	limit := at.AddDate(10, 0, 0)
	if end.Before(limit) {
		limit = end
	}
	for segment := 0; segment < 256 && at.Before(limit); segment++ {
		local := at.In(c.zone)
		_, offset := local.Zone()
		_, until := local.ZoneBounds()
		wall := at.Add(time.Duration(offset) * time.Second)
		candidate := c.cron.Next(wall.Add(-time.Nanosecond))
		if candidate.IsZero() {
			// 上游单次只查五年；2096→2104 等合法闰日规则需要第二个区间。
			candidate = c.cron.Next(wall.AddDate(5, 0, 0).Add(-time.Nanosecond))
		}
		if candidate.IsZero() {
			return nil, nil
		}
		candidate = candidate.Add(-time.Duration(offset) * time.Second).UTC()
		if !until.IsZero() && !candidate.Before(until) {
			if !until.After(at) {
				return nil, errors.New("timezone interval did not advance")
			}
			at = until.UTC()
			continue
		}
		if candidate.Before(at) {
			return nil, errors.New("calendar moved backwards")
		}
		if !candidate.Before(limit) {
			return nil, nil
		}
		return &candidate, nil
	}
	if at.Before(limit) {
		return nil, errors.New("timezone transition budget exceeded")
	}
	return nil, nil
}

// After 返回严格晚于该触发点的下一个 UTC 身份。
func (c *Calendar) After(at time.Time) (*time.Time, error) {
	return c.AtOrAfter(at.Add(time.Microsecond))
}