package binding

import (
	"errors"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
)

func TestNormalizeScheduleModesAndFinalInputs(t *testing.T) {
	anchor := mustTime("2026-07-21T00:00:00+08:00")
	absolute := mustTime("2026-07-22T00:00:00+08:00")
	tests := []struct {
		name     string
		schedule ScheduleSource
	}{
		{name: "cron", schedule: ScheduleSource{Mode: ScheduleModeCron, CronExpression: "0 9 * * *", Timezone: "Asia/Shanghai", Inputs: []byte(`{"channel":"chn_1"}`)}},
		{name: "interval", schedule: ScheduleSource{Mode: ScheduleModeInterval, AnchorAt: anchor, IntervalSeconds: int64Pointer(60), Inputs: []byte(`{"channel":"chn_1"}`)}},
		{name: "absolute", schedule: ScheduleSource{Mode: ScheduleModeAbsolute, ScheduledAt: absolute, Inputs: []byte(`{"channel":"chn_1"}`)}},
		{name: "relative", schedule: ScheduleSource{Mode: ScheduleModeRelative, RelativeSeconds: int64Pointer(90), Inputs: []byte(`{"channel":"chn_1"}`)}},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			spec := RevisionSpec{
				PlanID: "pln_1", PlanVersionID: "plv_1",
				InputPolicy: InputPolicy{Defaults: []byte(`{}`), Constants: []byte(`{}`)},
				Source:      Source{Type: SourceTypeSchedule, Schedule: &test.schedule},
			}
			normalized, err := NormalizeRevision(spec, bindingInputContract())
			require.NoError(t, err)
			require.JSONEq(t, `{"channel":"chn_1"}`, string(normalized.Spec.Source.Schedule.Inputs))
		})
	}

	missing := tests[3].schedule
	missing.Inputs = []byte(`{}`)
	_, err := NormalizeRevision(RevisionSpec{
		PlanID: "pln_1", PlanVersionID: "plv_1",
		InputPolicy: InputPolicy{Defaults: []byte(`{}`), Constants: []byte(`{}`)},
		Source:      Source{Type: SourceTypeSchedule, Schedule: &missing},
	}, bindingInputContract())
	require.Error(t, err)
	var validation ValidationError
	require.True(t, errors.As(err, &validation))
	require.Equal(t, "binding.input.required", validation.Code)
}

func TestNormalizeScheduleRejectsModeOverlapAndImplicitIntervalAnchor(t *testing.T) {
	tests := []ScheduleSource{
		{Mode: ScheduleModeCron, CronExpression: "* * * * *", Timezone: "UTC", RelativeSeconds: int64Pointer(2), Inputs: []byte(`{"channel":"chn_1"}`)},
		{Mode: ScheduleModeInterval, IntervalSeconds: int64Pointer(60), Inputs: []byte(`{"channel":"chn_1"}`)},
		{Mode: ScheduleModeAbsolute, ScheduledAt: mustTime("2026-07-21T00:00:00Z"), Timezone: "UTC", Inputs: []byte(`{"channel":"chn_1"}`)},
		{Mode: ScheduleModeRelative, RelativeSeconds: int64Pointer(0), Inputs: []byte(`{"channel":"chn_1"}`)},
	}
	for _, schedule := range tests {
		spec := RevisionSpec{
			PlanID: "pln_1", PlanVersionID: "plv_1",
			InputPolicy: InputPolicy{Defaults: []byte(`{}`), Constants: []byte(`{}`)},
			Source:      Source{Type: SourceTypeSchedule, Schedule: &schedule},
		}
		_, err := NormalizeRevision(spec, bindingInputContract())
		require.Error(t, err)
	}
}

func TestNormalizeScheduleNormalizesCronWhitespaceAndRejectsZeroTimes(t *testing.T) {
	left := RevisionSpec{
		PlanID: "pln_1", PlanVersionID: "plv_1",
		InputPolicy: InputPolicy{Defaults: []byte(`{}`), Constants: []byte(`{}`)},
		Source: Source{Type: SourceTypeSchedule, Schedule: &ScheduleSource{
			Mode: ScheduleModeCron, CronExpression: "  0   9  * *  * ", Timezone: " UTC ", Inputs: []byte(`{"channel":"chn_1"}`),
		}},
	}
	right := left
	right.Source.Schedule = &ScheduleSource{
		Mode: ScheduleModeCron, CronExpression: "0 9 * * *", Timezone: "UTC", Inputs: []byte(`{"channel":"chn_1"}`),
	}
	normalizedLeft, err := NormalizeRevision(left, bindingInputContract())
	require.NoError(t, err)
	normalizedRight, err := NormalizeRevision(right, bindingInputContract())
	require.NoError(t, err)
	require.Equal(t, "0 9 * * *", normalizedLeft.Spec.Source.Schedule.CronExpression)
	require.Equal(t, normalizedLeft.Digest, normalizedRight.Digest)

	zero := time.Time{}
	for _, schedule := range []ScheduleSource{
		{Mode: ScheduleModeInterval, AnchorAt: &zero, IntervalSeconds: int64Pointer(60), Inputs: []byte(`{"channel":"chn_1"}`)},
		{Mode: ScheduleModeAbsolute, ScheduledAt: &zero, Inputs: []byte(`{"channel":"chn_1"}`)},
	} {
		_, err := NormalizeRevision(RevisionSpec{
			PlanID: "pln_1", PlanVersionID: "plv_1",
			InputPolicy: InputPolicy{Defaults: []byte(`{}`), Constants: []byte(`{}`)},
			Source:      Source{Type: SourceTypeSchedule, Schedule: &schedule},
		}, bindingInputContract())
		require.Error(t, err)
	}
	_, err = InitialOccurrence(*normalizedLeft.Spec.Source.Schedule, time.Time{})
	require.Error(t, err)
}

func TestNormalizeScheduleRejectsProcessLocalTimezone(t *testing.T) {
	for _, timezone := range []string{"Local", " local "} {
		schedule := ScheduleSource{
			Mode: ScheduleModeCron, CronExpression: "0 9 * * *", Timezone: timezone,
			Inputs: []byte(`{"channel":"chn_1"}`),
		}
		_, err := NormalizeRevision(RevisionSpec{
			PlanID: "pln_1", PlanVersionID: "plv_1",
			InputPolicy: InputPolicy{Defaults: []byte(`{}`), Constants: []byte(`{}`)},
			Source:      Source{Type: SourceTypeSchedule, Schedule: &schedule},
		}, bindingInputContract())
		require.Error(t, err)
		var validation ValidationError
		require.True(t, errors.As(err, &validation))
		require.Equal(t, "binding.schedule_timezone", validation.Code)
		require.Equal(t, "$.source.schedule.timezone", validation.Path)
	}

	// Stable, explicit zones remain valid.
	for _, timezone := range []string{"UTC", "Asia/Shanghai"} {
		schedule := ScheduleSource{
			Mode: ScheduleModeCron, CronExpression: "0 9 * * *", Timezone: timezone,
			Inputs: []byte(`{"channel":"chn_1"}`),
		}
		_, err := NormalizeRevision(RevisionSpec{
			PlanID: "pln_1", PlanVersionID: "plv_1",
			InputPolicy: InputPolicy{Defaults: []byte(`{}`), Constants: []byte(`{}`)},
			Source:      Source{Type: SourceTypeSchedule, Schedule: &schedule},
		}, bindingInputContract())
		require.NoError(t, err)
	}
}

func TestRelativeOccurrenceMaterializesAtActivationAndDigestIsClockFree(t *testing.T) {
	schedule := ScheduleSource{Mode: ScheduleModeRelative, RelativeSeconds: int64Pointer(90), Inputs: []byte(`{"channel":"chn_1"}`)}
	spec := RevisionSpec{
		PlanID: "pln_1", PlanVersionID: "plv_1",
		InputPolicy: InputPolicy{Defaults: []byte(`{}`), Constants: []byte(`{}`)},
		Source:      Source{Type: SourceTypeSchedule, Schedule: &schedule},
	}
	first, err := NormalizeRevision(spec, bindingInputContract())
	require.NoError(t, err)
	second, err := NormalizeRevision(spec, bindingInputContract())
	require.NoError(t, err)
	require.Equal(t, first.Digest, second.Digest)
	require.Nil(t, first.Spec.Source.Schedule.AnchorAt)
	require.Nil(t, first.Spec.Source.Schedule.ScheduledAt)

	activatedA := time.Date(2026, 7, 21, 0, 0, 0, 0, time.UTC)
	activatedB := activatedA.Add(time.Hour)
	occurrenceA, err := InitialOccurrence(*first.Spec.Source.Schedule, activatedA)
	require.NoError(t, err)
	occurrenceB, err := InitialOccurrence(*first.Spec.Source.Schedule, activatedB)
	require.NoError(t, err)
	require.Equal(t, activatedA.Add(90*time.Second), *occurrenceA)
	require.Equal(t, activatedB.Add(90*time.Second), *occurrenceB)
	next, err := NextOccurrenceAfter(*first.Spec.Source.Schedule, *occurrenceA)
	require.NoError(t, err)
	require.Nil(t, next)
}

func TestIntervalOccurrenceKeepsExplicitAnchor(t *testing.T) {
	source := ScheduleSource{
		Mode: ScheduleModeInterval, AnchorAt: mustTime("2026-07-21T00:00:00Z"),
		IntervalSeconds: int64Pointer(60), Inputs: []byte(`{}`),
	}
	initial, err := InitialOccurrence(source, time.Date(2026, 7, 21, 0, 2, 30, 0, time.UTC))
	require.NoError(t, err)
	require.Equal(t, time.Date(2026, 7, 21, 0, 3, 0, 0, time.UTC), *initial)
	next, err := NextOccurrenceAfter(source, *initial)
	require.NoError(t, err)
	require.Equal(t, time.Date(2026, 7, 21, 0, 4, 0, 0, time.UTC), *next)
}

func TestIntervalOccurrenceAcrossDurationHorizonKeepsAnchorPhase(t *testing.T) {
	// This span is greater than time.Duration's roughly 292-year range. The
	// activation instant is exactly on the anchor phase, so strict scheduling
	// must select the following day rather than returning activation itself.
	source := ScheduleSource{
		Mode: ScheduleModeInterval, AnchorAt: mustTime("1500-01-01T12:34:56.123456Z"),
		IntervalSeconds: int64Pointer(24 * 60 * 60), Inputs: []byte(`{}`),
	}
	activation := mustTime("2026-01-01T12:34:56.123456Z")
	initial, err := InitialOccurrence(source, *activation)
	require.NoError(t, err)
	require.Equal(t, mustTime("2026-01-02T12:34:56.123456Z"), initial)
	require.True(t, initial.After(*activation))

	next, err := NextOccurrenceAfter(source, *initial)
	require.NoError(t, err)
	require.Equal(t, mustTime("2026-01-03T12:34:56.123456Z"), next)
}

func TestIntervalOccurrenceRangeBoundariesDoNotOverflow(t *testing.T) {
	min := minScheduleInstant
	max := maxScheduleInstant

	// A future anchor at the upper boundary itself is still a valid next
	// occurrence when the reference instant precedes it.
	beforeMax := max.Add(-time.Microsecond)
	next, err := nextIntervalOccurrence(max, 1, beforeMax)
	require.NoError(t, err)
	require.Equal(t, max, next)

	// Advancing from the upper boundary is a range error, not an integer wrap
	// into an earlier, apparently valid timestamp.
	_, err = nextIntervalOccurrence(max, 1, max)
	require.ErrorContains(t, err, "outside the durable range")

	// Exercise both a multi-millennial span and the largest accepted interval.
	// The mathematical result lies beyond DATETIME's upper bound and must fail
	// deterministically rather than overflowing time.Duration multiplication.
	_, err = nextIntervalOccurrence(min, maxScheduleSeconds, max)
	require.ErrorContains(t, err, "outside the durable range")

	_, err = nextIntervalOccurrence(min, maxScheduleSeconds+1, min)
	require.ErrorContains(t, err, "intervalSeconds")
}

func TestScheduleDatetimeUsesDurableMicrosecondPrecision(t *testing.T) {
	anchor := mustTime("2026-07-21T00:00:00.123456789+08:00")
	scheduled := mustTime("2026-07-22T00:00:00.987654321+08:00")
	for _, schedule := range []ScheduleSource{
		{Mode: ScheduleModeInterval, AnchorAt: anchor, IntervalSeconds: int64Pointer(60), Inputs: []byte(`{"channel":"chn_1"}`)},
		{Mode: ScheduleModeAbsolute, ScheduledAt: scheduled, Inputs: []byte(`{"channel":"chn_1"}`)},
	} {
		normalized, err := NormalizeRevision(RevisionSpec{
			PlanID: "pln_1", PlanVersionID: "plv_1",
			InputPolicy: InputPolicy{Defaults: []byte(`{}`), Constants: []byte(`{}`)},
			Source:      Source{Type: SourceTypeSchedule, Schedule: &schedule},
		}, bindingInputContract())
		require.NoError(t, err)
		stored := normalized.Spec.Source.Schedule
		if stored.AnchorAt != nil {
			require.Equal(t, 123456000, stored.AnchorAt.Nanosecond())
		}
		if stored.ScheduledAt != nil {
			require.Equal(t, 987654000, stored.ScheduledAt.Nanosecond())
		}
	}

	relative := ScheduleSource{Mode: ScheduleModeRelative, RelativeSeconds: int64Pointer(1), Inputs: []byte(`{}`)}
	activation := mustTime("2026-07-21T00:00:00.123456789Z")
	occurrence, err := InitialOccurrence(relative, *activation)
	require.NoError(t, err)
	require.Equal(t, mustTime("2026-07-21T00:00:01.123456Z"), occurrence)
}

func TestScheduleDatetimeRejectsOutsideDurableRange(t *testing.T) {
	tooEarly := time.Date(999, 12, 31, 23, 59, 59, 0, time.UTC)
	for _, schedule := range []ScheduleSource{
		{Mode: ScheduleModeInterval, AnchorAt: &tooEarly, IntervalSeconds: int64Pointer(60), Inputs: []byte(`{"channel":"chn_1"}`)},
		{Mode: ScheduleModeAbsolute, ScheduledAt: &tooEarly, Inputs: []byte(`{"channel":"chn_1"}`)},
	} {
		_, err := NormalizeRevision(RevisionSpec{
			PlanID: "pln_1", PlanVersionID: "plv_1",
			InputPolicy: InputPolicy{Defaults: []byte(`{}`), Constants: []byte(`{}`)},
			Source:      Source{Type: SourceTypeSchedule, Schedule: &schedule},
		}, bindingInputContract())
		require.Error(t, err)
		var validation ValidationError
		require.True(t, errors.As(err, &validation))
		require.Equal(t, "binding.schedule_datetime", validation.Code)
	}

	_, err := InitialOccurrence(
		ScheduleSource{Mode: ScheduleModeRelative, RelativeSeconds: int64Pointer(1)},
		tooEarly,
	)
	require.Error(t, err)

	nearMax := time.Date(9999, 12, 31, 23, 59, 59, 999999000, time.UTC)
	_, err = InitialOccurrence(
		ScheduleSource{Mode: ScheduleModeRelative, RelativeSeconds: int64Pointer(1)},
		nearMax,
	)
	require.Error(t, err)
}

func TestNormalizeScheduleInvocationAppliesNormalizedPrecedence(t *testing.T) {
	schedule := ScheduleSource{
		Mode: ScheduleModeRelative, RelativeSeconds: int64Pointer(30),
		Inputs: []byte(`{"channel":"chn_schedule","topic":"occurrence"}`),
	}
	normalized, err := NormalizeRevision(RevisionSpec{
		PlanID: "pln_1", PlanVersionID: "plv_1",
		InputPolicy: InputPolicy{
			Defaults:  []byte(`{"count":1}`),
			Constants: []byte(`{"secret":"fixed","topic":"constant"}`),
		},
		Source: Source{Type: SourceTypeSchedule, Schedule: &schedule},
	}, bindingInputContract())
	require.NoError(t, err)

	input, err := NormalizeScheduleInvocationInput(bindingInputContract(), normalized.Spec)
	require.NoError(t, err)
	payload, err := input.DeterministicJSON()
	require.NoError(t, err)
	require.JSONEq(t, `{"channel":"chn_schedule","count":1,"secret":"fixed","topic":"constant"}`, string(payload))

	_, err = NormalizeScheduleInvocationInput(bindingInputContract(), RevisionSpec{
		Source: Source{Type: SourceType("manual")},
	})
	require.Error(t, err)
	var validation ValidationError
	require.True(t, errors.As(err, &validation))
	require.Equal(t, "binding.source_type", validation.Code)
}