package mysql

import (
	"context"
	"database/sql"
	"errors"
	"sort"
	"time"
)

const operationalMetricSampleLimit = 10000
const operationalMetricEventWindow = 5 * time.Minute

// OperationalSnapshot 是进程观测使用的有限数据库快照。
// Counts 只包含固定状态的有界计数；达到上限时 Truncated 为 true，不能解释为精确总量。
type OperationalSnapshot struct {
	// TaskStates 是 Pulse 任务按状态的有界计数。
	TaskStates map[string]uint64 `json:"task_states"`
	// TaskOldestAgeSeconds 是运行中任务中最早 available_at 的年龄。
	TaskOldestAgeSeconds float64 `json:"task_oldest_age_seconds"`
	// DeliveryStates 是结果交付按状态的有界计数。
	DeliveryStates map[string]uint64 `json:"delivery_states"`
	// DeliveryOldestAgeSeconds 是待处理交付中最早 available_at 的年龄。
	DeliveryOldestAgeSeconds float64 `json:"delivery_oldest_age_seconds"`
	// WorkflowStates 是 Weave 实例按状态的有界计数。
	WorkflowStates map[string]uint64 `json:"workflow_states"`
	// WorkflowOldestAgeSeconds 是未终结流程中最早 updated_at 的年龄。
	WorkflowOldestAgeSeconds float64 `json:"workflow_oldest_age_seconds"`
	// InboxStates 是 Weave 内部事件按状态的有界计数。
	InboxStates map[string]uint64 `json:"inbox_states"`
	// InboxOldestAgeSeconds 是待消费内部事件中最早 available_at 的年龄。
	InboxOldestAgeSeconds float64 `json:"inbox_oldest_age_seconds"`
	// Truncated 表示至少一个状态达到有界采样上限。
	Truncated bool `json:"truncated"`
	// RecentEvents 是固定事件类别在最近窗口内的有界数量。
	RecentEvents map[string]uint64 `json:"recent_events"`
	// TaskTerminalLatencySeconds 是最近窗口终态任务从创建到完成的平均耗时样本。
	TaskTerminalLatencySeconds float64 `json:"task_terminal_latency_seconds"`
	// TaskTerminalLatencyMaxSeconds 是最近窗口终态任务样本的最大耗时。
	TaskTerminalLatencyMaxSeconds float64 `json:"task_terminal_latency_max_seconds"`
	// WorkflowTerminalLatencySeconds 是最近窗口终态流程从创建到完成的平均耗时样本。
	WorkflowTerminalLatencySeconds float64 `json:"workflow_terminal_latency_seconds"`
	// WorkflowTerminalLatencyMaxSeconds 是最近窗口终态流程样本的最大耗时。
	WorkflowTerminalLatencyMaxSeconds float64 `json:"workflow_terminal_latency_max_seconds"`
	// TaskAcceptToStartLatencySeconds 是最近窗口任务从受理事件到开始事件的平均耗时样本。
	TaskAcceptToStartLatencySeconds float64 `json:"task_accept_to_start_latency_seconds"`
	// TaskAcceptToStartLatencyMaxSeconds 是最近窗口任务受理到开始耗时样本的最大值。
	TaskAcceptToStartLatencyMaxSeconds float64 `json:"task_accept_to_start_latency_max_seconds"`
	// TaskAcceptToTerminalLatencySeconds 是最近窗口任务从受理到确定终态事件的平均耗时样本。
	TaskAcceptToTerminalLatencySeconds float64 `json:"task_accept_to_terminal_latency_seconds"`
	// TaskAcceptToTerminalLatencyMaxSeconds 是最近窗口任务受理到确定终态耗时样本的最大值。
	TaskAcceptToTerminalLatencyMaxSeconds float64 `json:"task_accept_to_terminal_latency_max_seconds"`
	// WorkflowStartToTerminalLatencySeconds 是最近窗口流程从开始到停止事件的平均耗时样本。
	WorkflowStartToTerminalLatencySeconds float64 `json:"workflow_start_to_terminal_latency_seconds"`
	// WorkflowStartToTerminalLatencyMaxSeconds 是最近窗口流程开始到停止耗时样本的最大值。
	WorkflowStartToTerminalLatencyMaxSeconds float64 `json:"workflow_start_to_terminal_latency_max_seconds"`
}

var (
	operationalTaskStates     = []string{"WAITING", "LEASED", "RUNNING", "NEEDS_REVIEW", "SUCCEEDED", "FAILED", "CANCELLED"}
	operationalDeliveryStates = []string{"PENDING", "LEASED", "ACKED", "EXPIRED"}
	operationalWorkflowStates = []string{"RUNNING", "WAITING", "CANCELLING", "NEEDS_REVIEW", "SUCCEEDED", "FAILED", "CANCELLED"}
	operationalInboxStates    = []string{"PENDING", "LEASED", "PROCESSED", "QUARANTINED"}
	operationalEventTypes     = []string{"task.accepted", "task.started", "task.succeeded", "task.failed", "task.cancelled", "task.needs_review", "task.requeued", "workflow.started", "workflow.advanced", "workflow.recovered", "workflow.blocked", "workflow.stopped"}
	phaseEventTypes           = []string{"task.accepted", "task.started", "task.succeeded", "task.failed", "task.cancelled", "workflow.started", "workflow.stopped"}
)

// ReadOperationalSnapshot performs only bounded, indexed reads. It is intended
// for a background sampler with a short context deadline, never for a request.
func ReadOperationalSnapshot(ctx context.Context, db *sql.DB) (OperationalSnapshot, error) {
	if db == nil {
		return OperationalSnapshot{}, errors.New("database is required")
	}
	now, err := readDatabaseNow(ctx, db)
	if err != nil {
		return OperationalSnapshot{}, err
	}
	tasks, taskTruncated, err := boundedStateCounts(ctx, db, "pulse_tasks", "ix_task_observe_state_available", operationalTaskStates)
	if err != nil {
		return OperationalSnapshot{}, err
	}
	deliveries, deliveryTruncated, err := boundedStateCounts(ctx, db, "orbit_deliveries", "ix_delivery_observe_state_available", operationalDeliveryStates)
	if err != nil {
		return OperationalSnapshot{}, err
	}
	workflows, workflowTruncated, err := boundedStateCounts(ctx, db, "weave_instances", "ix_workflow_observe_state_updated", operationalWorkflowStates)
	if err != nil {
		return OperationalSnapshot{}, err
	}
	inbox, inboxTruncated, err := boundedStateCounts(ctx, db, "weave_inbox", "ix_inbox_due", operationalInboxStates)
	if err != nil {
		return OperationalSnapshot{}, err
	}
	taskAge, err := oldestAge(ctx, db, "pulse_tasks", "ix_task_observe_state_available", []string{"WAITING", "LEASED", "RUNNING", "NEEDS_REVIEW"}, "available_at", now)
	if err != nil {
		return OperationalSnapshot{}, err
	}
	inboxAge, err := oldestAge(ctx, db, "weave_inbox", "ix_inbox_due", []string{"PENDING", "LEASED"}, "available_at", now)
	if err != nil {
		return OperationalSnapshot{}, err
	}
	deliveryAge, err := oldestAge(ctx, db, "orbit_deliveries", "ix_delivery_observe_state_available", []string{"PENDING", "LEASED"}, "available_at", now)
	if err != nil {
		return OperationalSnapshot{}, err
	}
	workflowAge, err := oldestAge(ctx, db, "weave_instances", "ix_workflow_observe_state_updated", []string{"RUNNING", "WAITING", "CANCELLING", "NEEDS_REVIEW"}, "updated_at", now)
	if err != nil {
		return OperationalSnapshot{}, err
	}
	events, eventTruncated, err := boundedRecentEventCounts(ctx, db, now)
	if err != nil {
		return OperationalSnapshot{}, err
	}
	taskLatency, err := sampledTerminalLatency(ctx, db, "pulse_tasks", "ix_task_observe_terminal_completed", []string{"SUCCEEDED", "FAILED", "CANCELLED"}, now.Add(-operationalMetricEventWindow))
	if err != nil {
		return OperationalSnapshot{}, err
	}
	workflowLatency, err := sampledTerminalLatency(ctx, db, "weave_instances", "ix_workflow_observe_terminal_completed", []string{"SUCCEEDED", "FAILED", "CANCELLED"}, now.Add(-operationalMetricEventWindow))
	if err != nil {
		return OperationalSnapshot{}, err
	}
	phaseLatency, err := sampledPhaseLatency(ctx, db, now.Add(-operationalMetricEventWindow))
	if err != nil {
		return OperationalSnapshot{}, err
	}
	return OperationalSnapshot{
		TaskStates: tasks, TaskOldestAgeSeconds: taskAge,
		DeliveryStates: deliveries, DeliveryOldestAgeSeconds: deliveryAge,
		WorkflowStates: workflows, WorkflowOldestAgeSeconds: workflowAge,
		InboxStates: inbox, InboxOldestAgeSeconds: inboxAge,
		RecentEvents:               events,
		TaskTerminalLatencySeconds: taskLatency.average, TaskTerminalLatencyMaxSeconds: taskLatency.maximum,
		WorkflowTerminalLatencySeconds: workflowLatency.average, WorkflowTerminalLatencyMaxSeconds: workflowLatency.maximum,
		TaskAcceptToStartLatencySeconds: phaseLatency.taskAcceptToStart.average, TaskAcceptToStartLatencyMaxSeconds: phaseLatency.taskAcceptToStart.maximum,
		TaskAcceptToTerminalLatencySeconds: phaseLatency.taskAcceptToTerminal.average, TaskAcceptToTerminalLatencyMaxSeconds: phaseLatency.taskAcceptToTerminal.maximum,
		WorkflowStartToTerminalLatencySeconds: phaseLatency.workflowStartToTerminal.average, WorkflowStartToTerminalLatencyMaxSeconds: phaseLatency.workflowStartToTerminal.maximum,
		Truncated: taskTruncated || deliveryTruncated || workflowTruncated || inboxTruncated || eventTruncated || taskLatency.truncated || workflowLatency.truncated || phaseLatency.truncated,
	}, nil
}

func readDatabaseNow(ctx context.Context, db *sql.DB) (time.Time, error) {
	var now time.Time
	if err := db.QueryRowContext(ctx, "SELECT UTC_TIMESTAMP(6)").Scan(&now); err != nil {
		return time.Time{}, err
	}
	return now.UTC(), nil
}

func boundedStateCounts(ctx context.Context, db *sql.DB, table, index string, states []string) (map[string]uint64, bool, error) {
	counts := make(map[string]uint64, len(states))
	truncated := false
	for _, state := range states {
		query := "SELECT COUNT(*) FROM (SELECT id FROM " + table + " FORCE INDEX (" + index + ") WHERE state=? LIMIT ?) AS bounded"
		var n uint64
		if err := db.QueryRowContext(ctx, query, state, operationalMetricSampleLimit+1).Scan(&n); err != nil {
			return nil, false, err
		}
		if n > operationalMetricSampleLimit {
			n = operationalMetricSampleLimit
			truncated = true
		}
		counts[state] = n
	}
	return counts, truncated, nil
}

func boundedRecentEventCounts(ctx context.Context, db *sql.DB, now time.Time) (map[string]uint64, bool, error) {
	counts := make(map[string]uint64, len(operationalEventTypes))
	truncated := false
	cutoff := now.Add(-operationalMetricEventWindow)
	for _, eventType := range operationalEventTypes {
		query := "SELECT COUNT(*) FROM (SELECT id FROM orbit_events FORCE INDEX (ix_event_observe_type_occurred) WHERE type=? AND occurred_at>=? LIMIT ?) AS bounded"
		var n uint64
		if err := db.QueryRowContext(ctx, query, eventType, cutoff, operationalMetricSampleLimit+1).Scan(&n); err != nil {
			return nil, false, err
		}
		if n > operationalMetricSampleLimit {
			n = operationalMetricSampleLimit
			truncated = true
		}
		counts[eventType] = n
	}
	return counts, truncated, nil
}

type terminalLatencySample struct {
	average   float64
	maximum   float64
	truncated bool
}

type phaseLatencySamples struct {
	taskAcceptToStart       terminalLatencySample
	taskAcceptToTerminal    terminalLatencySample
	workflowStartToTerminal terminalLatencySample
	truncated               bool
}

type phaseEvent struct {
	kind     string
	id       string
	typeName string
	at       time.Time
}

// sampledPhaseLatency pairs only fixed event types in the recent bounded
// window. Missing history, retries and events outside the window are excluded;
// the result is a trend sample rather than an end-to-end SLO.
func sampledPhaseLatency(ctx context.Context, db *sql.DB, cutoff time.Time) (phaseLatencySamples, error) {
	placeholders := "?"
	args := make([]any, 0, len(phaseEventTypes)+2)
	args = append(args, phaseEventTypes[0])
	for _, eventType := range phaseEventTypes[1:] {
		placeholders += ",?"
		args = append(args, eventType)
	}
	args = append([]any{cutoff}, args...)
	args = append(args, operationalMetricSampleLimit+1)
	query := "SELECT aggregate_kind,aggregate_id,type,occurred_at FROM orbit_events FORCE INDEX (ix_event_observe_type_occurred) WHERE occurred_at>=? AND type IN (" + placeholders + ") LIMIT ?"
	rows, err := db.QueryContext(ctx, query, args...)
	if err != nil {
		return phaseLatencySamples{}, err
	}
	defer rows.Close()
	tasks := make(map[string][]phaseEvent)
	workflows := make(map[string][]phaseEvent)
	read := 0
	for rows.Next() {
		read++
		var event phaseEvent
		var id []byte
		if err := rows.Scan(&event.kind, &id, &event.typeName, &event.at); err != nil {
			return phaseLatencySamples{}, err
		}
		if read <= operationalMetricSampleLimit {
			event.id = string(id)
			switch event.kind {
			case "task":
				tasks[event.id] = append(tasks[event.id], event)
			case "workflow":
				workflows[event.id] = append(workflows[event.id], event)
			}
		}
	}
	if err := rows.Err(); err != nil {
		return phaseLatencySamples{}, err
	}
	result := phaseLatencySamples{truncated: read > operationalMetricSampleLimit}
	result.taskAcceptToStart = pairPhaseLatencies(tasks, "task.accepted", "task.started")
	result.taskAcceptToTerminal = pairPhaseLatencies(tasks, "task.accepted", "task.succeeded", "task.failed", "task.cancelled")
	result.workflowStartToTerminal = pairPhaseLatencies(workflows, "workflow.started", "workflow.stopped")
	result.taskAcceptToStart.truncated = false
	result.taskAcceptToTerminal.truncated = false
	result.workflowStartToTerminal.truncated = false
	return result, nil
}

func pairPhaseLatencies(groups map[string][]phaseEvent, from string, to ...string) terminalLatencySample {
	targets := make(map[string]struct{}, len(to))
	for _, eventType := range to {
		targets[eventType] = struct{}{}
	}
	var total, maximum float64
	var count int
	for _, events := range groups {
		sort.Slice(events, func(i, j int) bool {
			if events[i].at.Equal(events[j].at) {
				return events[i].id < events[j].id
			}
			return events[i].at.Before(events[j].at)
		})
		var start *time.Time
		var finish *time.Time
		for _, event := range events {
			if event.typeName == from && start == nil {
				at := event.at.UTC()
				start = &at
				continue
			}
			if start != nil {
				if _, ok := targets[event.typeName]; ok && !event.at.Before(*start) {
					at := event.at.UTC()
					finish = &at
					break
				}
			}
		}
		if start == nil || finish == nil {
			continue
		}
		seconds := finish.Sub(*start).Seconds()
		if seconds < 0 {
			seconds = 0
		}
		total += seconds
		if seconds > maximum {
			maximum = seconds
		}
		count++
	}
	result := terminalLatencySample{maximum: maximum}
	if count > 0 {
		result.average = total / float64(count)
	}
	return result
}

func sampledTerminalLatency(ctx context.Context, db *sql.DB, table, index string, states []string, cutoff time.Time) (terminalLatencySample, error) {
	placeholders := "?"
	args := make([]any, 0, len(states)+2)
	args = append(args, states[0])
	for _, state := range states[1:] {
		placeholders += ",?"
		args = append(args, state)
	}
	args = append(args, cutoff, operationalMetricSampleLimit+1)
	query := "SELECT created_at,completed_at FROM " + table + " FORCE INDEX (" + index + ") WHERE state IN (" + placeholders + ") AND completed_at>=? ORDER BY completed_at,id LIMIT ?"
	rows, err := db.QueryContext(ctx, query, args...)
	if err != nil {
		return terminalLatencySample{}, err
	}
	defer rows.Close()
	var total, maximum float64
	var read, valid int
	for rows.Next() {
		read++
		if read > operationalMetricSampleLimit {
			continue
		}
		var created, completed time.Time
		if err := rows.Scan(&created, &completed); err != nil {
			return terminalLatencySample{}, err
		}
		seconds := completed.Sub(created).Seconds()
		if seconds < 0 {
			seconds = 0
		}
		total += seconds
		if seconds > maximum {
			maximum = seconds
		}
		valid++
	}
	if err := rows.Err(); err != nil {
		return terminalLatencySample{}, err
	}
	result := terminalLatencySample{maximum: maximum, truncated: read > operationalMetricSampleLimit}
	if valid > 0 {
		result.average = total / float64(valid)
	}
	return result, nil
}

func oldestAge(ctx context.Context, db *sql.DB, table, index string, states []string, column string, now time.Time) (float64, error) {
	placeholders := "?"
	args := make([]any, 0, len(states))
	args = append(args, states[0])
	for _, state := range states[1:] {
		placeholders += ",?"
		args = append(args, state)
	}
	query := "SELECT " + column + " FROM " + table + " FORCE INDEX (" + index + ") WHERE state IN (" + placeholders + ") ORDER BY " + column + ",id LIMIT 1"
	var occurred time.Time
	if err := db.QueryRowContext(ctx, query, args...).Scan(&occurred); err != nil {
		if err == sql.ErrNoRows {
			return 0, nil
		}
		return 0, err
	}
	age := now.Sub(occurred.UTC()).Seconds()
	if age < 0 {
		return 0, nil
	}
	return age, nil
}