package main

import (
	"context"
	orbit "git.yuntuo.hk/k2safe/orbit/sdk/go"
	"reflect"
	"time"
)

// slotObservation 只缓存一个固定时间槽的不可变触发和核验证据。
type slotObservation struct {
	occurrence                            *orbit.ScheduleOccurrence // occurrence 首次读到的快照；不得改写。
	verified                              bool                      // verified 是否已完成全部持久化核验。
	attemptID                             string                    // attemptID 唯一成功尝试，用于与本地调用关联。
	trigger, queue, execution, completion time.Duration             // 各阶段只使用数据库时间。
}

// planObservation 预分配固定槽位，内存上限由配置限制。
type planObservation struct {
	index     int               // index 计划在本次配置中的序号。
	start     time.Time         // start 固定触发基准。
	namespace string            // namespace 预检认证队列命名空间。
	slots     []slotObservation // slots 预期触发槽位，不随服务端输入增长。
}

// observations 合并轮询快照，跨计划检查唯一触发和任务身份。
type observations struct {
	cfg         config                      // cfg 固定样本配置。
	runID       string                      // runID 固定测量身份。
	plans       map[string]*planObservation // plans 创建时登记的有限计划集合。
	occurrences map[string]string           // occurrences 触发ID到计划ID，避免跨计划重复。
	tasks       map[string]*slotObservation // tasks 已观察任务到唯一槽位。
}

func newObservations(cfg config, runID string) *observations {
	return &observations{cfg: cfg, runID: runID, plans: map[string]*planObservation{}, occurrences: map[string]string{}, tasks: map[string]*slotObservation{}}
}
func (o *observations) addPlan(id string, index int, start time.Time, namespace string) {
	o.plans[id] = &planObservation{index: index, start: start, namespace: namespace, slots: make([]slotObservation, o.cfg.Requests)}
}
func (o *observations) ingest(id string, page orbit.ScheduleOccurrencePage) string {
	p := o.plans[id]
	if p == nil || page.Items == nil || page.NextCursor != "" || len(page.Items) > len(p.slots) {
		return "OCCURRENCE_PAGE_MISMATCH"
	}
	seen := map[string]bool{}
	for _, item := range page.Items {
		delta := item.ScheduledAt.Sub(p.start)
		if !validID(item.ID) || item.ScheduleID != id || item.ScheduleVersion != 1 || delta < 0 || delta%o.cfg.Interval != 0 || delta/o.cfg.Interval >= time.Duration(len(p.slots)) || item.CreatedAt.Before(item.ScheduledAt) {
			return "OCCURRENCE_MISMATCH"
		}
		if seen[item.ID] {
			return "DUPLICATE_OCCURRENCE"
		}
		seen[item.ID] = true
		slot := &p.slots[int(delta/o.cfg.Interval)]
		if slot.occurrence != nil {
			if !reflect.DeepEqual(*slot.occurrence, item) {
				return "OCCURRENCE_CHANGED"
			}
			continue
		}
		if _, exists := o.occurrences[item.ID]; exists {
			return "DUPLICATE_OCCURRENCE"
		}
		// 保存已观察证据再判定策略异常；未验证的触发不会充当成功分母。
		snapshot := item
		slot.occurrence = &snapshot
		o.occurrences[item.ID] = id
		if item.Decision != "FIRED" || item.CoveredUntil != nil || item.Reason != "SCHEDULED" {
			return "OCCURRENCE_DECISION_MISMATCH"
		}
		if item.TaskID == nil || !validID(*item.TaskID) {
			return "OCCURRENCE_TASK_MISMATCH"
		}
		if _, exists := o.tasks[*item.TaskID]; exists {
			return "DUPLICATE_TASK"
		}
		o.tasks[*item.TaskID] = slot
	}
	// 此样本总量不超过单页上限，历史不可消失；分页异常不降级继续。
	known := 0
	for _, s := range p.slots {
		if s.occurrence != nil {
			known++
		}
	}
	if len(page.Items) != known {
		return "OCCURRENCE_DISAPPEARED"
	}
	return ""
}
func (o *observations) verify(ctx context.Context, api scheduleAPI, id string) (bool, string) {
	p := o.plans[id]
	complete := true
	for i := range p.slots {
		slot := &p.slots[i]
		if slot.occurrence == nil {
			complete = false
			continue
		}
		if slot.verified {
			continue
		}
		if ctx.Err() != nil {
			return false, "VERIFY_" + classify(ctx.Err())
		}
		item := slot.occurrence
		if item.TaskID == nil || o.tasks[*item.TaskID] != slot {
			return false, "OCCURRENCE_TASK_MISMATCH"
		}
		call, stop := context.WithTimeout(ctx, o.cfg.Timeout)
		task, err := api.GetTask(call, *item.TaskID)
		stop()
		if err != nil {
			return false, "TASK_" + classify(err)
		}
		if task.ID != *item.TaskID || task.NamespaceID != p.namespace || !matchesTask(task, o.cfg, o.runID, p.index) || !task.CreatedAt.Equal(item.CreatedAt) || !task.AvailableAt.Equal(task.CreatedAt) || !task.DeadlineAt.Equal(task.CreatedAt.Add(time.Duration(o.cfg.RunTimeout.Milliseconds())*time.Millisecond)) {
			return false, "TASK_MISMATCH"
		}
		switch task.State {
		case orbit.TaskWaiting, orbit.TaskLeased, orbit.TaskRunning:
			complete = false
			continue
		case orbit.TaskFailed, orbit.TaskCancelled, orbit.TaskNeedsReview:
			return false, "TASK_" + string(task.State)
		case orbit.TaskSucceeded:
		default:
			return false, "UNKNOWN_TASK_STATE"
		}
		output, valid := decodePayload(task.Result)
		if !valid || output != (schedulePayload{RunID: o.runID, Schedule: p.index}) || task.CompletedAt == nil || task.EffectUncertain || task.AttemptCount != 1 || task.CancelRequestedAt != nil {
			return false, "COMPLETION_MISMATCH"
		}
		call, stop = context.WithTimeout(ctx, o.cfg.Timeout)
		history, err := api.ListTaskAttempts(call, task.ID, orbit.HistoryPageRequest{Limit: 2})
		stop()
		if err != nil {
			return false, "HISTORY_" + classify(err)
		}
		if len(history.Items) != 1 || history.NextCursor != "" {
			return false, "ATTEMPT_COUNT_MISMATCH"
		}
		a := history.Items[0]
		if !validID(a.ID) || a.TaskID != task.ID || a.Epoch != 1 || a.State != orbit.AttemptSucceeded || a.AuthorizedAt == nil || a.EndedAt == nil || a.ErrorCode != "" || a.Retryable || a.ConcurrencyToken != 0 {
			return false, "ATTEMPT_MISMATCH"
		}
		if a.CreatedAt.Before(task.AvailableAt) || a.AuthorizedAt.Before(a.CreatedAt) || a.EndedAt.Before(*a.AuthorizedAt) || !a.EndedAt.Equal(*task.CompletedAt) {
			return false, "TIMESTAMP_MISMATCH"
		}
		call, stop = context.WithTimeout(ctx, o.cfg.Timeout)
		source, err := api.GetTaskSource(call, task.ID)
		stop()
		if err != nil {
			return false, "SOURCE_" + classify(err)
		}
		s := source.Schedule
		if source.TaskID != task.ID || source.Kind != "SCHEDULE" || source.Workflow != nil || source.RetryOf != nil || s == nil || s.ScheduleID != id || s.ScheduleVersion != 1 || s.OccurrenceID != item.ID || !s.ScheduledAt.Equal(item.ScheduledAt) || s.Decision != "FIRED" || s.CoveredUntil != nil {
			return false, "SOURCE_MISMATCH"
		}
		slot.verified = true
		slot.attemptID = a.ID
		slot.trigger = task.CreatedAt.Sub(item.ScheduledAt)
		slot.queue = a.AuthorizedAt.Sub(task.CreatedAt)
		slot.execution = a.EndedAt.Sub(*a.AuthorizedAt)
		slot.completion = task.CompletedAt.Sub(item.ScheduledAt)
	}
	return complete, ""
}
func matchesTask(t orbit.Task, c config, runID string, index int) bool {
	payload, ok := decodePayload(t.Payload)
	return ok && payload == (schedulePayload{RunID: runID, Schedule: index}) && validID(t.ID) && t.QueueID == c.QueueID && t.Handler == handlerName(runID) && t.HandlerVersion == "v1" && t.PayloadVersion == "v1" && t.BusinessKey == businessKey(runID, index) && t.ConcurrencyKey == "" && t.Priority == 0 && t.RetryOf == nil && t.StartDeadlineAt == nil && t.Retry == (orbit.RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: uint64(c.RunTimeout.Milliseconds())}) && t.RecoverySafety == orbit.RecoveryReconcileOnly
}
func (o *observations) summarize(r *scheduleReport) {
	var trigger, queue, execution, completion []time.Duration
	for i := range r.Plans {
		p := &r.Plans[i]
		data := o.plans[p.ID]
		if data == nil {
			continue
		}
		for j := range data.slots {
			s := &data.slots[j]
			if s.occurrence != nil {
				p.Occurrences++
			}
			if s.occurrence != nil && s.occurrence.TaskID != nil && o.tasks[*s.occurrence.TaskID] == s {
				p.Fired++
			}
			if s.verified {
				p.VerifiedSuccess++
				trigger = append(trigger, s.trigger)
				queue = append(queue, s.queue)
				execution = append(execution, s.execution)
				completion = append(completion, s.completion)
			}
		}
		r.Occurrences += p.Occurrences
		r.Fired += p.Fired
		r.VerifiedSuccess += p.VerifiedSuccess
	}
	r.TriggerLag = distributionOf(trigger)
	r.QueueWait = distributionOf(queue)
	r.Execution = distributionOf(execution)
	r.Completion = distributionOf(completion)
}