package planner

import (
	"fmt"

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

const deliveryIntervalPlanVersion uint16 = 1

// The two halves of the budget are bounded by the wire contracts they configure:
// TiCDC's object storage flush interval and the destination's poll interval.
// They are properties of those systems, not per-deployment tuning.
const (
	producerFlushMinSeconds int64 = 2
	producerFlushMaxSeconds int64 = 600
	consumerPollMinSeconds  int64 = 5
	consumerPollMaxSeconds  int64 = 300

	// What a polled staging area can express, being the two halves added up.
	// A budget outside this is not a different configuration, it is one the
	// two wire contracts have no way to represent.
	minDeliveryBudgetSeconds = producerFlushMinSeconds + consumerPollMinSeconds
	maxDeliveryBudgetSeconds = producerFlushMaxSeconds + consumerPollMaxSeconds
)

// PlanDeliveryIntervals divides the end-to-end budget between TiCDC's object
// storage flush interval and the destination consumer poll interval.
//
// A notified destination has no discovery delay to pay for, so the whole budget
// goes to the producer and the consumer half disappears. That also means the
// two configurations do not accept the same budgets: a polling pipeline needs
// room for both halves, while a notified one is capped by the flush interval
// alone. Both directions are handled below rather than left to the caller,
// because the difference is a property of the two wire contracts and not
// something a caller could derive.
func PlanDeliveryIntervals(
	syncMode domain.DataPipelineSyncMode,
	totalSeconds int32,
	notified bool,
) (domain.DeliveryIntervalPlan, error) {
	if syncMode == domain.DataPipelineSyncModeFull {
		if totalSeconds != 0 {
			return domain.DeliveryIntervalPlan{}, fmt.Errorf(
				"%w: full sync requires a zero interval", ErrUnsupportedFlushInterval)
		}
		return domain.DeliveryIntervalPlan{Version: deliveryIntervalPlanVersion}, nil
	}
	if syncMode != domain.DataPipelineSyncModeFullAndIncremental {
		return domain.DeliveryIntervalPlan{}, fmt.Errorf(
			"%w: %q", ErrUnsupportedSyncMode, syncMode)
	}
	if totalSeconds <= 0 {
		return domain.DeliveryIntervalPlan{}, fmt.Errorf(
			"%w: interval must be positive", ErrUnsupportedFlushInterval)
	}
	total := int64(totalSeconds)
	if notified {
		// Only the ceiling is clamped, and the two ends are not symmetric.
		// Delivering sooner than asked still satisfies a latency budget, so a
		// budget above the flush ceiling is spent as far as it can be. A budget
		// below the floor cannot be honored at all - clamping it up would
		// quietly deliver slower than the budget the caller was told was
		// accepted.
		if total < producerFlushMinSeconds {
			return domain.DeliveryIntervalPlan{}, fmt.Errorf(
				"%w: total=%d producer=[%d,%d]",
				ErrUnsupportedFlushInterval,
				totalSeconds, producerFlushMinSeconds, producerFlushMaxSeconds)
		}
		producer := minInt64(total, producerFlushMaxSeconds)
		return domain.DeliveryIntervalPlan{
			Version:                      deliveryIntervalPlanVersion,
			EndToEndSeconds:              totalSeconds,
			ProducerFlushIntervalSeconds: int32(producer),
			// Zero is not "poll constantly": it records that there is no poll
			// at all, and the destination builder reads it that way.
			ConsumerPollIntervalSeconds: 0,
		}, nil
	}
	// The floor and the ceiling are not symmetric, for the same reason they are
	// not symmetric on the notified path above. A budget below the floor cannot
	// be honored at all, so it is refused rather than quietly tightened into a
	// promise the pipeline cannot keep. A budget above the ceiling is a latency
	// bound the pipeline beats: it is spent as far as the two halves reach, and
	// the data simply arrives sooner than asked.
	//
	// Refusing the ceiling instead would also make the two staging modes
	// disagree about the same number - a queue changes how quickly a new object
	// is noticed, not which budgets can be expressed - which is how a request
	// for 1800 came to be accepted with a queue and rejected without one.
	if total < minDeliveryBudgetSeconds {
		return domain.DeliveryIntervalPlan{}, fmt.Errorf(
			"%w: total=%d is below the smallest budget a polled staging area can"+
				" deliver, which is %d seconds (producer=[%d,%d] consumer=[%d,%d])",
			ErrUnsupportedFlushInterval,
			totalSeconds, minDeliveryBudgetSeconds,
			producerFlushMinSeconds, producerFlushMaxSeconds,
			consumerPollMinSeconds, consumerPollMaxSeconds,
		)
	}

	var producer, consumer int64
	if total > maxDeliveryBudgetSeconds {
		producer, consumer = producerFlushMaxSeconds, consumerPollMaxSeconds
	} else {
		low := maxInt64(producerFlushMinSeconds, total-consumerPollMaxSeconds)
		high := minInt64(producerFlushMaxSeconds, total-consumerPollMinSeconds)
		producer = clampInt64(total/2, low, high)
		consumer = total - producer
	}
	return domain.DeliveryIntervalPlan{
		Version:                      deliveryIntervalPlanVersion,
		EndToEndSeconds:              totalSeconds,
		ProducerFlushIntervalSeconds: int32(producer),
		ConsumerPollIntervalSeconds:  int32(consumer),
	}, nil
}

func clampInt64(value, low, high int64) int64 {
	return maxInt64(low, minInt64(value, high))
}

func minInt64(left, right int64) int64 {
	if left < right {
		return left
	}
	return right
}

func maxInt64(left, right int64) int64 {
	if left > right {
		return left
	}
	return right
}