package planner

import (
	"testing"

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
)

func TestPlanDeliveryIntervals(t *testing.T) {
	tests := map[string]struct {
		syncMode     domain.DataPipelineSyncMode
		total        int32
		notified     bool
		wantProducer int32
		wantConsumer int32
		wantErr      error
	}{
		"full sync has no interval": {
			syncMode: domain.DataPipelineSyncModeFull,
		},
		"full sync rejects interval": {
			syncMode: domain.DataPipelineSyncModeFull,
			total:    60,
			wantErr:  ErrUnsupportedFlushInterval,
		},
		"balanced split": {
			syncMode:     domain.DataPipelineSyncModeFullAndIncremental,
			total:        60,
			wantProducer: 30,
			wantConsumer: 30,
		},
		// The consumer cannot poll faster than its minimum, so at the low end
		// the producer absorbs whatever is left rather than the half split.
		"consumer minimum constrains producer": {
			syncMode:     domain.DataPipelineSyncModeFullAndIncremental,
			total:        7,
			wantProducer: 2,
			wantConsumer: 5,
		},
		// Symmetrically, the consumer maximum pushes the extra onto the producer.
		"consumer maximum constrains producer": {
			syncMode:     domain.DataPipelineSyncModeFullAndIncremental,
			total:        700,
			wantProducer: 400,
			wantConsumer: 300,
		},
		"largest feasible budget": {
			syncMode:     domain.DataPipelineSyncModeFullAndIncremental,
			total:        900,
			wantProducer: 600,
			wantConsumer: 300,
		},
		"budget below both minimums": {
			syncMode: domain.DataPipelineSyncModeFullAndIncremental,
			total:    6,
			wantErr:  ErrUnsupportedFlushInterval,
		},
		// A budget the halves cannot reach is a bound the pipeline beats, not
		// one it fails: it is spent as far as both maximums go.
		"budget above both maximums is spent to the ceiling": {
			syncMode:     domain.DataPipelineSyncModeFullAndIncremental,
			total:        901,
			wantProducer: 600,
			wantConsumer: 300,
		},
		"zero interval is invalid for incremental": {
			syncMode: domain.DataPipelineSyncModeFullAndIncremental,
			wantErr:  ErrUnsupportedFlushInterval,
		},
		"unknown mode is rejected": {
			syncMode: "unknown",
			total:    60,
			wantErr:  ErrUnsupportedSyncMode,
		},

		// A notified destination has nothing to poll, so the whole budget
		// becomes the flush interval instead of being halved.
		"notified spends the whole budget on the producer": {
			syncMode:     domain.DataPipelineSyncModeFullAndIncremental,
			total:        60,
			notified:     true,
			wantProducer: 60,
			wantConsumer: 0,
		},
		// Budgets the polling split cannot accept at either end are ordinary
		// for a notified pipeline: the consumer minimum and maximum are simply
		// not being paid.
		"notified accepts a budget below the polling minimum": {
			syncMode:     domain.DataPipelineSyncModeFullAndIncremental,
			total:        3,
			notified:     true,
			wantProducer: 3,
			wantConsumer: 0,
		},
		// Delivering sooner than asked satisfies a latency budget, so an
		// oversized budget clamps to the flush ceiling rather than failing.
		"notified clamps a budget above the flush ceiling": {
			syncMode:     domain.DataPipelineSyncModeFullAndIncremental,
			total:        1200,
			notified:     true,
			wantProducer: 600,
			wantConsumer: 0,
		},
		// Below the flush floor there is nothing left to shorten, and a budget
		// that cannot be met must fail rather than be quietly rounded up.
		"notified rejects a budget below the flush floor": {
			syncMode: domain.DataPipelineSyncModeFullAndIncremental,
			total:    1,
			notified: true,
			wantErr:  ErrUnsupportedFlushInterval,
		},
		"notified still rejects a zero interval": {
			syncMode: domain.DataPipelineSyncModeFullAndIncremental,
			notified: true,
			wantErr:  ErrUnsupportedFlushInterval,
		},
		// Full sync has no incremental stream, so there is nothing for a queue
		// to notify anyone about and the interval stays zero.
		"notified full sync still has no interval": {
			syncMode: domain.DataPipelineSyncModeFull,
			notified: true,
		},
	}

	for name, test := range tests {
		t.Run(name, func(t *testing.T) {
			plan, err := PlanDeliveryIntervals(test.syncMode, test.total, test.notified)
			if test.wantErr != nil {
				require.ErrorIs(t, err, test.wantErr)
				return
			}
			require.NoError(t, err)
			require.Equal(t, deliveryIntervalPlanVersion, plan.Version)
			require.Equal(t, test.total, plan.EndToEndSeconds)
			require.Equal(t, test.wantProducer, plan.ProducerFlushIntervalSeconds)
			require.Equal(t, test.wantConsumer, plan.ConsumerPollIntervalSeconds)
			// A polled pipeline spends the whole budget, or the whole of what
			// the two halves can reach when the budget is larger than that.
			if !test.notified && test.total > 0 {
				want := test.total
				if int64(want) > maxDeliveryBudgetSeconds {
					want = int32(maxDeliveryBudgetSeconds)
				}
				require.Equal(t, want,
					plan.ProducerFlushIntervalSeconds+plan.ConsumerPollIntervalSeconds)
			}
		})
	}
}

// A queue changes how quickly a new object is noticed, not which budgets can
// be expressed. Rejecting an over-large budget on one path and clamping it on
// the other made the same number valid or invalid depending on a setting that
// has nothing to do with it: a Console asking for 1800 was accepted with a
// queue and refused without one.
func TestDeliveryBudgetCeilingBehavesTheSameWithAndWithoutAQueue(t *testing.T) {
	t.Parallel()

	for _, total := range []int32{901, 1800, 3600} {
		polled, polledErr := PlanDeliveryIntervals(
			domain.DataPipelineSyncModeFullAndIncremental, total, false)
		require.NoError(t, polledErr, "a polled staging area must accept total=%d", total)

		notified, notifiedErr := PlanDeliveryIntervals(
			domain.DataPipelineSyncModeFullAndIncremental, total, true)
		require.NoError(t, notifiedErr, "a notified staging area must accept total=%d", total)

		// The requested budget is recorded as asked on both paths, so a reader
		// can still see what was wanted next to what will be delivered.
		require.Equal(t, total, polled.EndToEndSeconds)
		require.Equal(t, total, notified.EndToEndSeconds)

		// Both spend the producer to its ceiling; only the polled one pays for
		// discovery on top.
		require.Equal(t, int32(producerFlushMaxSeconds), polled.ProducerFlushIntervalSeconds)
		require.Equal(t, int32(consumerPollMaxSeconds), polled.ConsumerPollIntervalSeconds)
		require.Equal(t, int32(producerFlushMaxSeconds), notified.ProducerFlushIntervalSeconds)
		require.Zero(t, notified.ConsumerPollIntervalSeconds)
	}
}

// The floor is refused rather than clamped, on both paths: delivering slower
// than the budget the caller was told was accepted would be a broken promise,
// while delivering faster is not.
func TestDeliveryBudgetFloorIsRefusedNotClamped(t *testing.T) {
	t.Parallel()

	_, err := PlanDeliveryIntervals(domain.DataPipelineSyncModeFullAndIncremental, 6, false)
	require.ErrorIs(t, err, ErrUnsupportedFlushInterval)
	require.Contains(t, err.Error(), "below the smallest budget")
	require.Contains(t, err.Error(), "7 seconds")

	_, err = PlanDeliveryIntervals(domain.DataPipelineSyncModeFullAndIncremental, 1, true)
	require.ErrorIs(t, err, ErrUnsupportedFlushInterval)
}