179 lines
7.7 KiB
Go
179 lines
7.7 KiB
Go
package service
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"encoding/json"
|
||
"strings"
|
||
"time"
|
||
"unicode"
|
||
"unicode/utf8"
|
||
|
||
"git.yuntuo.hk/k2safe/orbit/internal/fault"
|
||
"git.yuntuo.hk/k2safe/orbit/internal/jsonvalue"
|
||
core "git.yuntuo.hk/k2safe/orbit/model"
|
||
"git.yuntuo.hk/k2safe/orbit/platform/auth"
|
||
pulse "git.yuntuo.hk/k2safe/orbit/pulse/model"
|
||
)
|
||
|
||
// ReviewInput 人工核查未知效果；声明与证据由获授权的核查主体负责。
|
||
type ReviewInput struct {
|
||
// ExpectedRevision 当前待核查任务版本，非零；重报保留原版本。
|
||
ExpectedRevision uint64 `json:"expected_revision,string"`
|
||
// Reason 核查原因与结论说明，最多1024个Unicode字符。
|
||
Reason string `json:"reason"`
|
||
// EvidenceRef 受控证据引用，1至512个无空白/控制字符的Unicode字符；平台不请求该地址。
|
||
EvidenceRef string `json:"evidence_ref"`
|
||
// ExecutionSettled 必须显式为true，声明已核实全部历史执行已结束且不会继续产生效果。
|
||
ExecutionSettled bool `json:"execution_settled" required:"true"`
|
||
// Outcome 核查结论，仅SUCCEEDED或FAILED；不能直接重开旧任务。
|
||
Outcome pulse.TaskState `json:"outcome"`
|
||
// Result 非空JSON结论或结果引用，最大256 KiB，保留精确数值。
|
||
Result json.RawMessage `json:"result"`
|
||
}
|
||
|
||
type ReviewResult struct {
|
||
// Task 核查提交时的任务快照；当前状态另行查询。
|
||
Task TaskView `json:"task"`
|
||
// AuditID 不可变核查审计记录标识，可通过任务操作历史读取。
|
||
AuditID core.ID `json:"audit_id"`
|
||
// Deduplicated 是否返回原核查操作回执。
|
||
Deduplicated bool `json:"deduplicated"`
|
||
}
|
||
|
||
// ManualRetryInput 为已确认失败/取消的独立任务建立新预算，不修改原任务历史。
|
||
type ManualRetryInput struct {
|
||
// ExpectedRevision 原任务的当前版本，重报保持原值。
|
||
ExpectedRevision uint64 `json:"expected_revision,string"`
|
||
// Reason 人工重新执行的原因，最多1024个Unicode字符。
|
||
Reason string `json:"reason"`
|
||
// EvidenceRef 重新执行的安全依据引用，最多512个字符，不包含秘密。
|
||
EvidenceRef string `json:"evidence_ref"`
|
||
// SafeToRetry 必须显式为true，声明旧执行已结束且按原业务身份重新执行安全。
|
||
SafeToRetry bool `json:"safe_to_retry" required:"true"`
|
||
// TaskTimeoutMS 新任务从受理起的总时限，必须显式指定，最多30天。
|
||
TaskTimeoutMS uint64 `json:"task_timeout_ms"`
|
||
// AvailableAt 新任务最早开始时间，空表示立即；不复用旧任务的过期绝对时间。
|
||
AvailableAt *time.Time `json:"available_at,omitempty"`
|
||
// StartDeadlineAt 新任务最晚开始时间，空表示使用新的总截止时间。
|
||
StartDeadlineAt *time.Time `json:"start_deadline_at,omitempty"`
|
||
}
|
||
|
||
type ManualRetryResult struct {
|
||
// SourceTaskID 被人工重新发起的原任务标识，其终态与尝试历史保持不变。
|
||
SourceTaskID core.ID `json:"source_task_id"`
|
||
// Task 新任务的原始受理快照，RetryOf指回原任务。
|
||
Task TaskView `json:"task"`
|
||
// AuditID 原任务上的人工重试审计标识，包含新任务关联。
|
||
AuditID core.ID `json:"audit_id"`
|
||
// Deduplicated 是否返回原人工重试回执。
|
||
Deduplicated bool `json:"deduplicated"`
|
||
}
|
||
|
||
type TaskOperationSummary struct {
|
||
// ID 不可改写的审计记录标识。
|
||
ID core.ID `json:"id"`
|
||
// TaskID 本页授权任务标识。
|
||
TaskID core.ID `json:"task_id"`
|
||
// OperationID 服务端管理操作身份，不是调用方原始幂等键。
|
||
OperationID core.ID `json:"operation_id"`
|
||
// ActorID 实际认证主体标识，不接受请求体自报。
|
||
ActorID string `json:"actor_id"`
|
||
// Action 管理动作，例如task.review、task.retry、task.cancel、task.result。
|
||
Action string `json:"action"`
|
||
// CreatedAt 不可变审计创建时间，UTC微秒精度。
|
||
CreatedAt time.Time `json:"created_at"`
|
||
}
|
||
|
||
type TaskOperationView struct {
|
||
TaskOperationSummary
|
||
// Reason 实际提交的操作原因。
|
||
Reason string `json:"reason"`
|
||
// EvidenceRef 操作证据引用；非核查类操作可能为空。
|
||
EvidenceRef string `json:"evidence_ref,omitempty"`
|
||
// Before 操作前的受控状态摘要，不包含完整业务输入。
|
||
Before json.RawMessage `json:"before"`
|
||
// After 操作后摘要、核查声明或关联新任务，不包含凭据秘密。
|
||
After json.RawMessage `json:"after"`
|
||
}
|
||
|
||
type TaskOperationPage struct {
|
||
// Items 按创建时间及审计ID倒序排列，不返回原因与摘要正文。
|
||
Items []TaskOperationSummary `json:"items"`
|
||
// NextCursor 必须沿相同任务继续使用，空表示本次已无下一页。
|
||
NextCursor string `json:"next_cursor,omitempty"`
|
||
}
|
||
|
||
func validateRecoveryControl(id core.ID, key string, revision uint64, reason, evidence string) error {
|
||
if id.Validate() != nil {
|
||
return fault.Invalid("invalid task id")
|
||
}
|
||
if err := ValidateKey(key); err != nil {
|
||
return err
|
||
}
|
||
if revision == 0 || !utf8.ValidString(reason) || strings.TrimSpace(reason) == "" || utf8.RuneCountInString(reason) > 1024 || strings.ContainsRune(reason, 0) {
|
||
return fault.Invalid("reason and nonzero expected_revision are required")
|
||
}
|
||
if !utf8.ValidString(evidence) || len(evidence) == 0 || utf8.RuneCountInString(evidence) > 512 {
|
||
return fault.Invalid("evidence_ref must contain 1 to 512 characters")
|
||
}
|
||
for _, r := range evidence {
|
||
if unicode.IsSpace(r) || unicode.IsControl(r) {
|
||
return fault.Invalid("evidence_ref must not contain whitespace or control characters")
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (s *Service) ReviewTask(ctx context.Context, t auth.Token, id core.ID, key string, in ReviewInput) (ReviewResult, error) {
|
||
if err := validateRecoveryControl(id, key, in.ExpectedRevision, in.Reason, in.EvidenceRef); err != nil {
|
||
return ReviewResult{}, err
|
||
}
|
||
if !in.ExecutionSettled || (in.Outcome != pulse.TaskSucceeded && in.Outcome != pulse.TaskFailed) {
|
||
return ReviewResult{}, fault.Invalid("execution_settled=true and SUCCEEDED or FAILED outcome are required")
|
||
}
|
||
if len(in.Result) == 0 || len(in.Result) > 256<<10 || bytes.Equal(bytes.TrimSpace(in.Result), []byte("null")) {
|
||
return ReviewResult{}, fault.Invalid("result must be non-null JSON up to 256 KiB")
|
||
}
|
||
if _, err := jsonvalue.Canonical(in.Result); err != nil {
|
||
return ReviewResult{}, fault.Invalid("invalid review result")
|
||
}
|
||
return s.store.ReviewTask(ctx, t, id, key, in)
|
||
}
|
||
|
||
func (s *Service) RetryTask(ctx context.Context, t auth.Token, id core.ID, key string, in ManualRetryInput) (ManualRetryResult, error) {
|
||
if err := validateRecoveryControl(id, key, in.ExpectedRevision, in.Reason, in.EvidenceRef); err != nil {
|
||
return ManualRetryResult{}, err
|
||
}
|
||
if !in.SafeToRetry || in.TaskTimeoutMS == 0 || in.TaskTimeoutMS > 30*24*60*60*1000 {
|
||
return ManualRetryResult{}, fault.Invalid("safe_to_retry=true and task_timeout_ms between 1 and 2592000000 are required")
|
||
}
|
||
for _, field := range []**time.Time{&in.AvailableAt, &in.StartDeadlineAt} {
|
||
if *field != nil {
|
||
v := (**field).UTC()
|
||
if v.Year() < 2000 || v.Year() > 9998 || v.Nanosecond()%1000 != 0 {
|
||
return ManualRetryResult{}, fault.Invalid("times must use supported years and microsecond precision")
|
||
}
|
||
*field = &v
|
||
}
|
||
}
|
||
return s.store.RetryTask(ctx, t, id, key, in)
|
||
}
|
||
|
||
func (s *Service) ListTaskOperations(ctx context.Context, t auth.Token, id core.ID, in HistoryPageInput) (TaskOperationPage, error) {
|
||
if err := validateHistoryIDs(id); err != nil {
|
||
return TaskOperationPage{}, err
|
||
}
|
||
in, err := normalizeHistoryPage(in)
|
||
if err != nil {
|
||
return TaskOperationPage{}, err
|
||
}
|
||
return s.store.ListTaskOperations(ctx, t, id, in)
|
||
}
|
||
func (s *Service) GetTaskOperation(ctx context.Context, t auth.Token, id, operation core.ID) (TaskOperationView, error) {
|
||
if err := validateHistoryIDs(id, operation); err != nil {
|
||
return TaskOperationView{}, err
|
||
}
|
||
return s.store.GetTaskOperation(ctx, t, id, operation)
|
||
}
|