package schedule import ( "context" "strings" "time" "unicode/utf8" "git.yuntuo.hk/k2safe/orbit/internal/fault" core "git.yuntuo.hk/k2safe/orbit/model" "git.yuntuo.hk/k2safe/orbit/platform/auth" "git.yuntuo.hk/k2safe/orbit/pulse/service" ) // Definition 是每次发布冻结的完整定义,业务内容仍由独立 Worker 解释。 type Definition struct { Rule Rule `json:"rule"` // Rule 时间规则;start_at 是基准,不允许追溯到版本生效前产生新任务。 Policy Policy `json:"policy"` // Policy 明确的漏跑、重叠和补跑边界。 Task service.SubmitInput `json:"task"` // Task 每次触发的模板,禁止 available_at/start_deadline_at 和非零 delay_ms。 } // CreateInput 创建持久化计划,暂停计划也保存完整的触发进度。 type CreateInput struct { Name string `json:"name"` // Name 命名空间内唯一的 ASCII 名称。 Definition Definition `json:"definition"` // Definition 首个不可变版本。 Paused bool `json:"paused"` // Paused 是否从暂停状态开始,恢复时处理期间漏跑。 } // PublishInput 发布立即生效的新版本;旧积压仍按当时版本处理。 type PublishInput struct { ExpectedRevision uint64 `json:"expected_revision,string"` // ExpectedRevision 当前控制版本,必须非零。 Reason string `json:"reason"` // Reason 必填操作原因,写入审计。 Definition Definition `json:"definition"` // Definition 完整替换定义;队列不能更换。 } // ControlInput 暂停/恢复计划,不能取消已产生的任务。 type ControlInput struct { ExpectedRevision uint64 `json:"expected_revision,string"` // ExpectedRevision 乐观并发版本。 Reason string `json:"reason"` // Reason 必填操作原因,最多 1024 字符。 } // VersionView 保留规则、模板和原生效时间,已生成任务不会采用后续版本。 type VersionView struct { Version uint64 `json:"version,string"` // Version 发布序号,从一递增。 EffectiveAt time.Time `json:"effective_at"` // EffectiveAt 数据库事务内取得的生效边界,包含此时间。 Definition Definition `json:"definition"` // Definition 已冻结内容。 } // Summary 用于计划列表,不包含大任务参数。 type Summary struct { ID core.ID `json:"id"` // ID 计划稳定身份。 NamespaceID core.ID `json:"namespace_id"` // NamespaceID 隔离范围。 QueueID core.ID `json:"queue_id"` // QueueID 计划终身绑定队列。 Name string `json:"name"` // Name 同命名空间内唯一名称。 Revision uint64 `json:"revision,string"` // Revision 控制和进度变更版本。 CurrentVersion uint64 `json:"current_version,string"` // CurrentVersion 最新发布版本。 ProcessingVersion uint64 `json:"processing_version,string"` // ProcessingVersion 尚在处理积压的历史版本。 Paused bool `json:"paused"` // Paused 为真时停止新触发。 NextFireAt *time.Time `json:"next_fire_at,omitempty"` // NextFireAt 原计划时间;积压时可在过去,待初始化或结束时为空,需结合 CheckAt。 CheckAt *time.Time `json:"check_at,omitempty"` // CheckAt 下次可扫描时间;为空表示没有扫描工作,非空也可能尚待初始化触发点。 CreatedAt time.Time `json:"created_at"` // CreatedAt UTC 创建时间。 UpdatedAt time.Time `json:"updated_at"` // UpdatedAt UTC 最近变更时间。 } // View 返回控制状态和最新版本,历史版本通过专用查询读取。 type View struct { Summary Current VersionView `json:"current"` // Current 最新完整发布版本,不一定是当前积压所用版本。 } // Result 是幂等命令的原始回执,当前状态应使用 Get 查询。 type Result struct { Schedule View `json:"schedule"` // Schedule 本次命令提交时的快照。 Deduplicated bool `json:"deduplicated"` // Deduplicated 是否取回原回执。 } // OccurrenceView 描述一个已决定的触发点或不重叠范围。 type OccurrenceView struct { ID core.ID `json:"id"` // ID 触发记录标识。 ScheduleID core.ID `json:"schedule_id"` // ScheduleID 所属计划。 ScheduleVersion uint64 `json:"schedule_version,string"` // ScheduleVersion 当时适用的不可变版本。 ScheduledAt time.Time `json:"scheduled_at"` // ScheduledAt 原时间,参与唯一去重。 Decision string `json:"decision"` // Decision FIRED、SKIPPED 或 COALESCED。 TaskID *core.ID `json:"task_id,omitempty"` // TaskID 已生成任务;跳过时为空。 CoveredUntil *time.Time `json:"covered_until,omitempty"` // CoveredUntil 范围排除边界,不一定是触发点。 Reason string `json:"reason"` // Reason 稳定决策分类。 CreatedAt time.Time `json:"created_at"` // CreatedAt 数据库记录时间。 } type Page struct { Items []Summary `json:"items"` // Items 当前页摘要,始终为数组。 NextCursor string `json:"next_cursor,omitempty"` // NextCursor 不透明后续游标。 } type OccurrencePage struct { Items []OccurrenceView `json:"items"` // Items 当前页触发历史。 NextCursor string `json:"next_cursor,omitempty"` // NextCursor 原样用于相同父资源查询。 } type PageInput struct { Limit int // Limit 默认 20,最大 100。 Cursor string // Cursor 最长 1024 字符,不能作为权限证明。 } // Store 必须原子执行命令授权、版本、游标、审计和幂等回执。 type Store interface { Create(context.Context, auth.Token, string, CreateInput) (Result, error) Get(context.Context, auth.Token, core.ID) (View, error) Publish(context.Context, auth.Token, core.ID, string, PublishInput) (Result, error) Control(context.Context, auth.Token, core.ID, string, ControlInput, bool) (Result, error) List(context.Context, auth.Token, core.ID, PageInput) (Page, error) Version(context.Context, auth.Token, core.ID, uint64) (VersionView, error) Occurrences(context.Context, auth.Token, core.ID, PageInput) (OccurrencePage, error) } type Service struct{ store Store } func New(store Store) *Service { return &Service{store: store} } func Normalize(in Definition) (Definition, error) { c, err := Compile(in.Rule) if err != nil { return in, fault.Invalid(err.Error()) } in.Rule = c.Rule() if err := in.Policy.Validate(); err != nil { return in, fault.Invalid(err.Error()) } if in.Task.AvailableAt != nil || in.Task.StartDeadlineAt != nil || in.Task.DelayMS != 0 { return in, fault.Invalid("schedule templates cannot contain task times or a nonzero delay_ms") } in.Task, err = service.NormalizeSubmit(in.Task) return in, err } func control(id core.ID, key string, revision uint64, reason string) error { if id.Validate() != nil || revision == 0 || !utf8.ValidString(reason) || strings.TrimSpace(reason) == "" || utf8.RuneCountInString(reason) > 1024 { return fault.Invalid("valid schedule id, expected_revision and reason are required") } return service.ValidateKey(key) } func page(in PageInput) (PageInput, error) { if in.Limit == 0 { in.Limit = 20 } if in.Limit < 1 || in.Limit > 100 || len(in.Cursor) > 1024 { return in, fault.Invalid("invalid schedule page") } return in, nil } func (s *Service) Create(ctx context.Context, t auth.Token, key string, in CreateInput) (Result, error) { if err := service.ValidateKey(key); err != nil { return Result{}, err } if !service.ValidName(in.Name, 128) { return Result{}, fault.Invalid("schedule name is invalid") } var err error in.Definition, err = Normalize(in.Definition) if err != nil { return Result{}, err } return s.store.Create(ctx, t, key, in) } func (s *Service) Get(ctx context.Context, t auth.Token, id core.ID) (View, error) { if id.Validate() != nil { return View{}, fault.Invalid("invalid schedule id") } return s.store.Get(ctx, t, id) } func (s *Service) Publish(ctx context.Context, t auth.Token, id core.ID, key string, in PublishInput) (Result, error) { if err := control(id, key, in.ExpectedRevision, in.Reason); err != nil { return Result{}, err } var err error in.Definition, err = Normalize(in.Definition) if err != nil { return Result{}, err } return s.store.Publish(ctx, t, id, key, in) } func (s *Service) Control(ctx context.Context, t auth.Token, id core.ID, key string, in ControlInput, paused bool) (Result, error) { if err := control(id, key, in.ExpectedRevision, in.Reason); err != nil { return Result{}, err } return s.store.Control(ctx, t, id, key, in, paused) } func (s *Service) List(ctx context.Context, t auth.Token, queue core.ID, in PageInput) (Page, error) { if queue.Validate() != nil { return Page{}, fault.Invalid("queue_id is required") } in, err := page(in) if err != nil { return Page{}, err } return s.store.List(ctx, t, queue, in) } func (s *Service) Version(ctx context.Context, t auth.Token, id core.ID, version uint64) (VersionView, error) { if id.Validate() != nil || version == 0 { return VersionView{}, fault.Invalid("invalid schedule id or version") } return s.store.Version(ctx, t, id, version) } func (s *Service) Occurrences(ctx context.Context, t auth.Token, id core.ID, in PageInput) (OccurrencePage, error) { if id.Validate() != nil { return OccurrencePage{}, fault.Invalid("invalid schedule id") } in, err := page(in) if err != nil { return OccurrencePage{}, err } return s.store.Occurrences(ctx, t, id, in) }