package binding

import (
	"fmt"
	"math"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/store/schedule"
	"github.com/tidbcloud/AgentTeam/internal/workflow/plan"
)

const maxScheduleSeconds = int64(math.MaxInt64 / int64(time.Second))

// Binding source cursors are persisted as DATETIME(6). Normalize authored and
// materialized instants to that exact precision and range before they enter a
// revision digest or a source row. This keeps the immutable configuration,
// activation receipt, and durable cursor from disagreeing after a database
// round trip.
var (
	minScheduleInstant = time.Date(1000, 1, 1, 0, 0, 0, 0, time.UTC)
	maxScheduleInstant = time.Date(9999, 12, 31, 23, 59, 59, 999999000, time.UTC)
)

func normalizeScheduleSource(source ScheduleSource, contract plan.ArtifactContract) (ScheduleSource, plan.NormalizedInput, error) {
	source.Mode = ScheduleMode(strings.TrimSpace(string(source.Mode)))
	// The cron parser treats runs of whitespace identically. Collapse them so
	// semantically equivalent schedules produce one deterministic revision digest.
	source.CronExpression = strings.Join(strings.Fields(source.CronExpression), " ")
	source.Timezone = strings.TrimSpace(source.Timezone)
	if source.AnchorAt != nil {
		anchor, err := normalizeScheduleInstant(*source.AnchorAt)
		if err != nil {
			return ScheduleSource{}, plan.NormalizedInput{}, invalid("binding.schedule_datetime", "$.source.schedule.anchorAt", err.Error())
		}
		source.AnchorAt = &anchor
	}
	if source.ScheduledAt != nil {
		scheduled, err := normalizeScheduleInstant(*source.ScheduledAt)
		if err != nil {
			return ScheduleSource{}, plan.NormalizedInput{}, invalid("binding.schedule_datetime", "$.source.schedule.scheduledAt", err.Error())
		}
		source.ScheduledAt = &scheduled
	}
	inputs, inputsJSON, err := normalizePolicyObject(contract, source.Inputs, "$.source.schedule.inputs")
	if err != nil {
		return ScheduleSource{}, plan.NormalizedInput{}, err
	}
	source.Inputs = inputsJSON

	switch source.Mode {
	case ScheduleModeCron:
		if source.CronExpression == "" || source.Timezone == "" {
			return ScheduleSource{}, plan.NormalizedInput{}, invalid("binding.schedule_cron", "$.source.schedule", "cron schedule requires cronExpression and timezone")
		}
		// time.LoadLocation treats "Local" as a process-global special value.
		// Accepting it would make an otherwise immutable Binding revision run at
		// different instants depending on the scheduler host. Require an explicit,
		// durable timezone instead (UTC and IANA zone names remain valid).
		if strings.EqualFold(source.Timezone, "Local") {
			return ScheduleSource{}, plan.NormalizedInput{}, invalid("binding.schedule_timezone", "$.source.schedule.timezone", "cron timezone must be UTC or an explicit IANA timezone, not Local")
		}
		if source.IntervalSeconds != nil || source.AnchorAt != nil || source.ScheduledAt != nil || source.RelativeSeconds != nil {
			return ScheduleSource{}, plan.NormalizedInput{}, invalid("binding.schedule_union", "$.source.schedule", "cron schedule cannot contain interval, absolute, or relative fields")
		}
		if _, err := schedule.NextCronOccurrenceAfter(source.CronExpression, source.Timezone, time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC)); err != nil {
			return ScheduleSource{}, plan.NormalizedInput{}, invalid("binding.schedule_cron", "$.source.schedule.cronExpression", err.Error())
		}
	case ScheduleModeInterval:
		if source.IntervalSeconds == nil || *source.IntervalSeconds <= 0 || *source.IntervalSeconds > maxScheduleSeconds || source.AnchorAt == nil {
			return ScheduleSource{}, plan.NormalizedInput{}, invalid("binding.schedule_interval", "$.source.schedule", "interval schedule requires explicit anchorAt and positive intervalSeconds")
		}
		if source.AnchorAt.IsZero() {
			return ScheduleSource{}, plan.NormalizedInput{}, invalid("binding.schedule_interval", "$.source.schedule.anchorAt", "interval anchorAt must be a non-zero datetime")
		}
		if source.CronExpression != "" || source.Timezone != "" || source.ScheduledAt != nil || source.RelativeSeconds != nil {
			return ScheduleSource{}, plan.NormalizedInput{}, invalid("binding.schedule_union", "$.source.schedule", "interval schedule cannot contain cron, absolute, or relative fields")
		}
	case ScheduleModeAbsolute:
		if source.ScheduledAt == nil {
			return ScheduleSource{}, plan.NormalizedInput{}, invalid("binding.schedule_absolute", "$.source.schedule.scheduledAt", "absolute schedule requires scheduledAt")
		}
		if source.ScheduledAt.IsZero() {
			return ScheduleSource{}, plan.NormalizedInput{}, invalid("binding.schedule_absolute", "$.source.schedule.scheduledAt", "absolute scheduledAt must be a non-zero datetime")
		}
		if source.CronExpression != "" || source.Timezone != "" || source.IntervalSeconds != nil || source.AnchorAt != nil || source.RelativeSeconds != nil {
			return ScheduleSource{}, plan.NormalizedInput{}, invalid("binding.schedule_union", "$.source.schedule", "absolute schedule cannot contain cron, interval, or relative fields")
		}
	case ScheduleModeRelative:
		if source.RelativeSeconds == nil || *source.RelativeSeconds <= 0 || *source.RelativeSeconds > maxScheduleSeconds {
			return ScheduleSource{}, plan.NormalizedInput{}, invalid("binding.schedule_relative", "$.source.schedule.relativeSeconds", "relative schedule requires positive relativeSeconds")
		}
		if source.CronExpression != "" || source.Timezone != "" || source.IntervalSeconds != nil || source.AnchorAt != nil || source.ScheduledAt != nil {
			return ScheduleSource{}, plan.NormalizedInput{}, invalid("binding.schedule_union", "$.source.schedule", "relative schedule cannot contain cron, interval, or absolute fields")
		}
	default:
		return ScheduleSource{}, plan.NormalizedInput{}, invalid("binding.schedule_mode", "$.source.schedule.mode", "schedule mode must be cron, interval, absolute, or relative")
	}
	return source, inputs, nil
}

// InitialOccurrence materializes source runtime state at activation time. It
// is deliberately separate from revision normalization: relative schedules
// retain only their delay in the immutable digest and are re-anchored on every
// activation, while intervals keep their explicit semantic anchor.
func InitialOccurrence(source ScheduleSource, activatedAt time.Time) (*time.Time, error) {
	var err error
	activatedAt, err = normalizeScheduleInstant(activatedAt)
	if err != nil {
		return nil, fmt.Errorf("invalid activatedAt: %w", err)
	}
	switch source.Mode {
	case ScheduleModeCron:
		next, err := schedule.NextCronOccurrenceAfter(source.CronExpression, source.Timezone, activatedAt)
		if err != nil {
			return nil, fmt.Errorf("compute initial cron occurrence: %w", err)
		}
		return normalizedSchedulePointer(next)
	case ScheduleModeInterval:
		if source.AnchorAt == nil || source.IntervalSeconds == nil {
			return nil, fmt.Errorf("interval schedule requires anchorAt and intervalSeconds")
		}
		anchor, err := normalizeScheduleInstant(*source.AnchorAt)
		if err != nil {
			return nil, fmt.Errorf("invalid interval anchorAt: %w", err)
		}
		next, err := nextIntervalOccurrence(anchor, *source.IntervalSeconds, activatedAt)
		if err != nil {
			return nil, fmt.Errorf("compute initial interval occurrence: %w", err)
		}
		return normalizedSchedulePointer(next)
	case ScheduleModeAbsolute:
		if source.ScheduledAt == nil {
			return nil, fmt.Errorf("absolute schedule requires scheduledAt")
		}
		return normalizedSchedulePointer(*source.ScheduledAt)
	case ScheduleModeRelative:
		if source.RelativeSeconds == nil || *source.RelativeSeconds <= 0 || *source.RelativeSeconds > maxScheduleSeconds {
			return nil, fmt.Errorf("relative schedule requires positive relativeSeconds")
		}
		return normalizedSchedulePointer(activatedAt.Add(time.Duration(*source.RelativeSeconds) * time.Second))
	default:
		return nil, fmt.Errorf("unsupported schedule mode %q", source.Mode)
	}
}

// NextOccurrenceAfter advances a recurring schedule strictly after one
// occurrence. One-shot schedules return nil.
func NextOccurrenceAfter(source ScheduleSource, after time.Time) (*time.Time, error) {
	var err error
	after, err = normalizeScheduleInstant(after)
	if err != nil {
		return nil, fmt.Errorf("invalid after: %w", err)
	}
	switch source.Mode {
	case ScheduleModeCron:
		next, err := schedule.NextCronOccurrenceAfter(source.CronExpression, source.Timezone, after)
		if err != nil {
			return nil, fmt.Errorf("compute next cron occurrence: %w", err)
		}
		return normalizedSchedulePointer(next)
	case ScheduleModeInterval:
		if source.AnchorAt == nil || source.IntervalSeconds == nil {
			return nil, fmt.Errorf("interval schedule requires anchorAt and intervalSeconds")
		}
		anchor, err := normalizeScheduleInstant(*source.AnchorAt)
		if err != nil {
			return nil, fmt.Errorf("invalid interval anchorAt: %w", err)
		}
		next, err := nextIntervalOccurrence(anchor, *source.IntervalSeconds, after)
		if err != nil {
			return nil, fmt.Errorf("compute next interval occurrence: %w", err)
		}
		return normalizedSchedulePointer(next)
	case ScheduleModeAbsolute, ScheduleModeRelative:
		return nil, nil
	default:
		return nil, fmt.Errorf("unsupported schedule mode %q", source.Mode)
	}
}

// NormalizeScheduleInvocationInput assembles the immutable input for one due
// schedule occurrence. Schedule sources do not accept caller-authored values:
// the occurrence layer is the normalized input persisted in the immutable
// Binding revision. The schedule precedence is Plan defaults < Binding
// defaults < occurrence input < Binding constants.
func NormalizeScheduleInvocationInput(contract plan.ArtifactContract, revision RevisionSpec) (plan.NormalizedInput, error) {
	if revision.Source.Type != SourceTypeSchedule || revision.Source.Schedule == nil {
		return plan.NormalizedInput{}, invalid("binding.source_type", "$.source", "Binding revision is not a schedule source")
	}
	defaults, _, err := normalizePolicyObject(contract, revision.InputPolicy.Defaults, "$.inputPolicy.defaults")
	if err != nil {
		return plan.NormalizedInput{}, err
	}
	constants, _, err := normalizePolicyObject(contract, revision.InputPolicy.Constants, "$.inputPolicy.constants")
	if err != nil {
		return plan.NormalizedInput{}, err
	}
	occurrence, _, err := normalizePolicyObject(contract, revision.Source.Schedule.Inputs, "$.source.schedule.inputs")
	if err != nil {
		return plan.NormalizedInput{}, err
	}
	return plan.FinalizeInput(contract, defaults, occurrence, constants)
}

func timePointer(value time.Time) *time.Time { return &value }

func normalizeScheduleInstant(value time.Time) (time.Time, error) {
	if value.IsZero() {
		return time.Time{}, fmt.Errorf("datetime must be non-zero")
	}
	value = value.UTC().Truncate(time.Microsecond)
	if value.Before(minScheduleInstant) || value.After(maxScheduleInstant) {
		return time.Time{}, fmt.Errorf("datetime must be within the durable range %s through %s",
			minScheduleInstant.Format(time.RFC3339Nano), maxScheduleInstant.Format(time.RFC3339Nano))
	}
	return value, nil
}

func normalizedSchedulePointer(value time.Time) (*time.Time, error) {
	value, err := normalizeScheduleInstant(value)
	if err != nil {
		return nil, err
	}
	return timePointer(value), nil
}

// nextIntervalOccurrence returns the first occurrence strictly after after,
// preserving the phase established by anchor. It intentionally computes in
// Unix microseconds rather than with time.Time.Sub or time.Duration: Go's
// duration range is only about 292 years, while MySQL DATETIME supports the
// complete years 1000 through 9999 used by Binding schedules.
func nextIntervalOccurrence(anchor time.Time, intervalSeconds int64, after time.Time) (time.Time, error) {
	if intervalSeconds <= 0 || intervalSeconds > maxScheduleSeconds {
		return time.Time{}, fmt.Errorf("intervalSeconds must be positive and within the supported range")
	}

	var err error
	anchor, err = normalizeScheduleInstant(anchor)
	if err != nil {
		return time.Time{}, fmt.Errorf("invalid anchorAt: %w", err)
	}
	after, err = normalizeScheduleInstant(after)
	if err != nil {
		return time.Time{}, fmt.Errorf("invalid after: %w", err)
	}
	if after.Before(anchor) {
		return anchor, nil
	}

	// Every supported DATETIME instant fits comfortably in Unix microseconds
	// (roughly -3e16 through 2.6e17), as does their difference. The configured
	// interval is duration-bounded, so converting it to microseconds is safe.
	const microsPerSecond = int64(time.Second / time.Microsecond)
	intervalMicros := intervalSeconds * microsPerSecond
	anchorMicros := anchor.UnixMicro()
	afterMicros := after.UnixMicro()
	elapsedMicros := afterMicros - anchorMicros
	steps := elapsedMicros/intervalMicros + 1
	if steps > math.MaxInt64/intervalMicros {
		return time.Time{}, fmt.Errorf("next interval occurrence overflows supported arithmetic")
	}
	offsetMicros := steps * intervalMicros
	if anchorMicros > math.MaxInt64-offsetMicros {
		return time.Time{}, fmt.Errorf("next interval occurrence overflows supported arithmetic")
	}

	next, err := normalizeScheduleInstant(time.UnixMicro(anchorMicros + offsetMicros).UTC())
	if err != nil {
		return time.Time{}, fmt.Errorf("next interval occurrence is outside the durable range: %w", err)
	}
	if !next.After(after) {
		return time.Time{}, fmt.Errorf("next interval occurrence must be strictly after the reference instant")
	}
	return next, nil
}