110 lines
4.6 KiB
Go
110 lines
4.6 KiB
Go
package mysql
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"errors"
|
|
"time"
|
|
|
|
"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"
|
|
"git.yuntuo.hk/k2safe/orbit/weave/engine"
|
|
)
|
|
|
|
func (s *TaskStore) GetTaskSource(ctx context.Context, t auth.Token, id core.ID) (service.TaskSourceView, error) {
|
|
out := service.TaskSourceView{TaskID: id}
|
|
err := s.transaction(ctx, t, "tasks:read", func(tx *sql.Tx, p auth.Permissions) error {
|
|
var queue core.ID
|
|
var owner sql.NullString
|
|
var ownerID, gate, retry *core.ID
|
|
var scheduled, activated, polled bool
|
|
err := tx.QueryRowContext(ctx, `SELECT queue_id,owner_kind,owner_id,gate_id,retry_of,
|
|
EXISTS(SELECT 1 FROM pulse_schedule_occurrences o WHERE o.namespace_id=t.namespace_id AND o.task_id=t.id),
|
|
EXISTS(SELECT 1 FROM weave_activations a WHERE a.namespace_id=t.namespace_id AND a.task_id=t.id),
|
|
EXISTS(SELECT 1 FROM weave_poll_rounds r WHERE r.namespace_id=t.namespace_id AND r.task_id=t.id)
|
|
FROM pulse_tasks t WHERE namespace_id=? AND id=?`, t.NamespaceID, id).Scan(&queue, &owner, &ownerID, &gate, &retry, &scheduled, &activated, &polled)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return fault.NotFound
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !p.AllowsQueue(queue) {
|
|
return fault.NotFound
|
|
}
|
|
// 反向关联同样必须存在且种类唯一；丢失所有者不能把流程任务降级为普通受理。
|
|
if owner.Valid != (ownerID != nil) {
|
|
return fault.Unavailable
|
|
}
|
|
if owner.Valid && owner.String == engine.TaskOwnerKind {
|
|
if gate == nil || scheduled || activated == polled {
|
|
return fault.Unavailable
|
|
}
|
|
source, err := readTaskWorkflowSource(ctx, tx, t.NamespaceID, id, queue, *ownerID, *gate, activated)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
out.Kind, out.Workflow = "WORKFLOW", &source
|
|
return nil
|
|
}
|
|
if gate != nil || activated || polled || owner.Valid && retry != nil {
|
|
return fault.Unavailable
|
|
}
|
|
if owner.Valid {
|
|
if owner.String != "schedule" {
|
|
return fault.Unavailable
|
|
}
|
|
var source service.TaskScheduleSource
|
|
// 每条关联均显式限定命名空间、计划、版本与任务队列，不依赖数据库外键。
|
|
err := tx.QueryRowContext(ctx, `SELECT o.schedule_id,o.schedule_version,o.id,o.scheduled_at,o.decision,o.covered_until
|
|
FROM pulse_schedule_occurrences o
|
|
JOIN pulse_schedules s ON s.namespace_id=o.namespace_id AND s.id=o.schedule_id AND s.queue_id=?
|
|
JOIN pulse_schedule_versions v ON v.namespace_id=s.namespace_id AND v.schedule_id=s.id AND v.version=o.schedule_version AND v.version<=s.current_version
|
|
WHERE o.namespace_id=? AND o.task_id=? AND o.schedule_id=? AND JSON_UNQUOTE(JSON_EXTRACT(v.task_template,'$.queue_id'))=?`, queue, t.NamespaceID, id, *ownerID, string(queue)).Scan(&source.ScheduleID, &source.ScheduleVersion, &source.OccurrenceID, &source.ScheduledAt, &source.Decision, &source.CoveredUntil)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return fault.Unavailable
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if source.ScheduleVersion == 0 || !sourceTimeValid(source.ScheduledAt) || source.CoveredUntil != nil && !sourceTimeValid(*source.CoveredUntil) || source.Decision == "FIRED" && source.CoveredUntil != nil || source.Decision == "COALESCED" && (source.CoveredUntil == nil || !source.CoveredUntil.After(source.ScheduledAt)) || source.Decision != "FIRED" && source.Decision != "COALESCED" {
|
|
return fault.Unavailable
|
|
}
|
|
out.Kind = "SCHEDULE"
|
|
out.Schedule = &source
|
|
return nil
|
|
}
|
|
if scheduled {
|
|
return fault.Unavailable
|
|
}
|
|
if retry == nil {
|
|
out.Kind = "DIRECT"
|
|
return nil
|
|
}
|
|
var parent core.ID
|
|
err = tx.QueryRowContext(ctx, `SELECT id FROM pulse_tasks t WHERE namespace_id=? AND id=? AND id<>? AND queue_id=? AND owner_kind IS NULL AND owner_id IS NULL AND gate_id IS NULL AND state IN ('FAILED','CANCELLED')
|
|
AND NOT EXISTS(SELECT 1 FROM pulse_schedule_occurrences o WHERE o.namespace_id=t.namespace_id AND o.task_id=t.id)
|
|
AND NOT EXISTS(SELECT 1 FROM weave_activations a WHERE a.namespace_id=t.namespace_id AND a.task_id=t.id)
|
|
AND NOT EXISTS(SELECT 1 FROM weave_poll_rounds r WHERE r.namespace_id=t.namespace_id AND r.task_id=t.id)`, t.NamespaceID, *retry, id, queue).Scan(&parent)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return fault.Unavailable
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
out.Kind = "MANUAL_RETRY"
|
|
out.RetryOf = &parent
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return service.TaskSourceView{}, err
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func sourceTimeValid(t time.Time) bool {
|
|
return t.Year() >= 2000 && t.Year() <= 9998 && t.Nanosecond()%1000 == 0
|
|
}
|