package changefeed

import (
	"fmt"
	"time"

	"github.com/tidbcloud/dataflow-service-e2e/cases"
	"github.com/tidbcloud/dataflow-service-e2e/cases/essential/common"
	. "github.com/tidbcloud/dataflow-service-e2e/internal/model"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/essential"
)

func expectScaled(c *cases.BaseCase, client essential.CDCClient, cluster *TiDBServerlessEntity, id string, rcu int64) {
	c.Expect("scaled changefeed reports target RCU", func() error {
		detail, err := client.Get(c.Context(), cluster, id)
		if err != nil {
			return err
		}
		if detail.RCU != rcu {
			return fmt.Errorf("changefeed RCU=%d, want %d", detail.RCU, rcu)
		}
		return common.ValidateChangefeedState(detail.State, "RUNNING")
	}, 60, 15*time.Second)
}

func verifyScaledEvent(cluster *TiDBServerlessEntity, kafka *KafkaEntity, topic, marker string) error {
	if _, err := common.WriteMarker(cluster, marker); err != nil {
		return err
	}
	return common.ConsumeKafkaEvent(kafka, topic, marker)
}