//go:build integration
// +build integration

package integration

import (
	"bytes"
	"context"
	"fmt"
	"io"
	"os"
	"path"
	"strings"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	"google.golang.org/grpc"

	ydb "github.com/ydb-platform/ydb-go-sdk/v3"
	"github.com/ydb-platform/ydb-go-sdk/v3/config"
	"github.com/ydb-platform/ydb-go-sdk/v3/internal/topic/topicwritetest"
	"github.com/ydb-platform/ydb-go-sdk/v3/internal/version"
	xtest "github.com/ydb-platform/ydb-go-sdk/v3/pkg/xtest"
	"github.com/ydb-platform/ydb-go-sdk/v3/topic"
	"github.com/ydb-platform/ydb-go-sdk/v3/topic/topicoptions"
	"github.com/ydb-platform/ydb-go-sdk/v3/topic/topictypes"
	"github.com/ydb-platform/ydb-go-sdk/v3/topic/topicwriter"
)

const defaultConnectionString = "grpc://localhost:2136/local"

const commonConsumerName = "consumer"

func TestTopicCreateDrop(t *testing.T) {
	ctx := xtest.Context(t)
	db := connect(t)
	topicPath := db.Name() + "/testtopic"

	_ = db.Topic().Drop(ctx, topicPath)
	err := db.Topic().Create(ctx, topicPath,
		topicoptions.CreateWithConsumer(
			topictypes.Consumer{
				Name: "test",
			},
		),
	)
	require.NoError(t, err)

	_, err = db.Topic().Describe(ctx, topicPath)
	require.NoError(t, err)

	err = db.Topic().Drop(ctx, topicPath)
	require.NoError(t, err)
}

func TestTopicDescribe(t *testing.T) {
	ctx := xtest.Context(t)
	db := connect(t)
	topicName := "test-topic-" + t.Name()

	var (
		supportedCodecs     = []topictypes.Codec{topictypes.CodecRaw, topictypes.CodecGzip}
		minActivePartitions = int64(2)
		// partitionCountLimit = int64(5) LOGBROKER-7800
		retentionPeriod = time.Hour
		writeSpeed      = int64(1023)
		burstBytes      = int64(222)
		consumers       = []topictypes.Consumer{
			{
				Name:            "c1",
				Important:       false,
				SupportedCodecs: []topictypes.Codec{topictypes.CodecRaw, topictypes.CodecGzip},
				ReadFrom:        time.Date(2022, 9, 11, 10, 1, 2, 0, time.UTC),
			},
			{
				Name:            "c2",
				SupportedCodecs: []topictypes.Codec{},
				ReadFrom:        time.Date(2021, 1, 2, 3, 4, 5, 0, time.UTC),
			},
		}
	)

	_ = db.Topic().Drop(ctx, topicName)
	err := db.Topic().Create(ctx, topicName,
		topicoptions.CreateWithSupportedCodecs(supportedCodecs...),
		topicoptions.CreateWithMinActivePartitions(minActivePartitions),
		topicoptions.CreateWithMaxActivePartitions(2),
		topicoptions.CreateWithAutoPartitioningSettings(topictypes.AutoPartitioningSettings{
			AutoPartitioningStrategy: topictypes.AutoPartitioningStrategyScaleUp,
			AutoPartitioningWriteSpeedStrategy: topictypes.AutoPartitioningWriteSpeedStrategy{
				StabilizationWindow:    15 * time.Second,
				UpUtilizationPercent:   80,
				DownUtilizationPercent: 20,
			},
		}),
		// topicoptions.CreateWithPartitionCountLimit(partitionCountLimit), LOGBROKER-7800
		topicoptions.CreateWithRetentionPeriod(retentionPeriod),
		// topicoptions.CreateWithRetentionStorageMB(...) - incompatible with retention period
		topicoptions.CreateWithPartitionWriteSpeedBytesPerSecond(writeSpeed),
		topicoptions.CreateWithPartitionWriteBurstBytes(burstBytes),
		topicoptions.CreateWithConsumer(consumers...),
		// topicoptions.CreateWithMeteringMode(topictypes.MeteringModeRequestUnits), - work with serverless only
	)
	require.NoError(t, err)

	topicDesc, err := db.Topic().Describe(ctx, topicName)
	require.NoError(t, err)

	expected := topictypes.TopicDescription{
		Path: topicName,
		PartitionSettings: topictypes.PartitionSettings{
			MinActivePartitions: minActivePartitions,
			MaxActivePartitions: 2,
			AutoPartitioningSettings: topictypes.AutoPartitioningSettings{
				AutoPartitioningStrategy: topictypes.AutoPartitioningStrategyScaleUp,
				AutoPartitioningWriteSpeedStrategy: topictypes.AutoPartitioningWriteSpeedStrategy{
					StabilizationWindow:    15 * time.Second,
					UpUtilizationPercent:   80,
					DownUtilizationPercent: 20,
				},
			},
			// PartitionCountLimit: partitionCountLimit, LOGBROKER-7800
		},
		Partitions: []topictypes.PartitionInfo{
			{
				PartitionID: 0,
				Active:      true,
			},
			{
				PartitionID: 1,
				Active:      true,
			},
		},
		RetentionPeriod:                   retentionPeriod,
		RetentionStorageMB:                0,
		SupportedCodecs:                   supportedCodecs,
		PartitionWriteBurstBytes:          burstBytes,
		PartitionWriteSpeedBytesPerSecond: writeSpeed,
		Attributes:                        nil,
		Consumers:                         consumers,
		MeteringMode:                      topictypes.MeteringModeUnspecified,
	}

	requireAndCleanSubset := func(checked *map[string]string, subset *map[string]string) {
		t.Helper()
		for k, subValue := range *subset {
			checkedValue, ok := (*checked)[k]
			require.True(t, ok, k)
			require.Equal(t, subValue, checkedValue)
		}
		*checked = nil
		*subset = nil
	}

	requireAndCleanSubset(&topicDesc.Attributes, &expected.Attributes)

	for i := range expected.Consumers {
		requireAndCleanSubset(&topicDesc.Consumers[i].Attributes, &expected.Consumers[i].Attributes)
	}

	for i := range topicDesc.Partitions {
		topicDesc.Partitions[i].FromBound = nil
		topicDesc.Partitions[i].ToBound = nil
	}
	require.Equal(t, expected, topicDesc)
}

func TestTopicDescribePartitionStats(t *testing.T) {
	ctx := xtest.Context(t)
	db := connect(t)
	topicName := "test-topic-" + t.Name()

	var (
		supportedCodecs      = []topictypes.Codec{topictypes.CodecRaw}
		minActivePartitions  = int64(1)
		maxActivePartitions  = int64(1)
		partitioningSettings = topictypes.AutoPartitioningSettings{
			AutoPartitioningStrategy: topictypes.AutoPartitioningStrategyScaleUp,
			AutoPartitioningWriteSpeedStrategy: topictypes.AutoPartitioningWriteSpeedStrategy{
				StabilizationWindow:    15 * time.Second,
				UpUtilizationPercent:   80,
				DownUtilizationPercent: 20,
			},
		}
	)

	_ = db.Topic().Drop(ctx, topicName)
	err := db.Topic().Create(ctx, topicName,
		topicoptions.CreateWithSupportedCodecs(supportedCodecs...),
		topicoptions.CreateWithMinActivePartitions(minActivePartitions),
		topicoptions.CreateWithMaxActivePartitions(maxActivePartitions),
		topicoptions.CreateWithAutoPartitioningSettings(partitioningSettings),
	)
	require.NoError(t, err)

	writer, err := db.Topic().StartWriter(topicName, topicoptions.WithWriterWaitServerAck(true))
	require.NoError(t, err)
	topicMessage := topicwriter.Message{
		Data: bytes.NewReader([]byte{128}),
	}
	err = writer.Write(ctx, topicMessage)
	require.NoError(t, err)

	topicDesc, err := db.Topic().Describe(ctx, topicName, topicoptions.IncludePartitionStats())
	require.NoError(t, err)

	expected := topictypes.TopicDescription{
		Path: topicName,
		PartitionSettings: topictypes.PartitionSettings{
			MinActivePartitions:      minActivePartitions,
			MaxActivePartitions:      maxActivePartitions,
			AutoPartitioningSettings: partitioningSettings,
		},
		Partitions: []topictypes.PartitionInfo{
			{
				PartitionID: 0,
				Active:      true,
				PartitionStats: topictypes.PartitionStats{
					PartitionsOffset: topictypes.OffsetRange{
						Start: 0,
						End:   1,
					},
					BytesWritten: topictypes.MultipleWindowsStat{
						PerMinute: 1,
						PerHour:   1,
						PerDay:    1,
					},
				},
			},
		},
		RetentionPeriod:                   86400000000000,
		RetentionStorageMB:                0,
		SupportedCodecs:                   supportedCodecs,
		PartitionWriteBurstBytes:          1048576,
		PartitionWriteSpeedBytesPerSecond: 1048576,
		Attributes:                        nil,
		Consumers:                         []topictypes.Consumer{},
		MeteringMode:                      topictypes.MeteringModeUnspecified,
	}

	requireAndCleanSubset := func(checked *map[string]string, subset *map[string]string) {
		t.Helper()
		for k, subValue := range *subset {
			checkedValue, ok := (*checked)[k]
			require.True(t, ok, k)
			require.Equal(t, subValue, checkedValue)
		}
		*checked = nil
		*subset = nil
	}

	topicDesc.Partitions[0].PartitionStats.LastWriteTime = nil
	bw := &topicDesc.Partitions[0].PartitionStats.BytesWritten
	sign := func(x int64) int64 {
		if x > 0 {
			return 1
		}
		return 0
	}
	topicDesc.Partitions[0].PartitionStats.BytesWritten = topictypes.MultipleWindowsStat{
		PerMinute: sign(bw.PerMinute),
		PerHour:   sign(bw.PerHour),
		PerDay:    sign(bw.PerDay),
	}

	requireAndCleanSubset(&topicDesc.Attributes, &expected.Attributes)
	require.Equal(t, expected, topicDesc)
}

func TestDescribePartitionSettings(t *testing.T) {
	scope := newScope(t)
	ctx := scope.Ctx

	const (
		defaultRetentionPeriod = 24 * time.Hour
		writeSpeed             = int64(1024)
		burstBytes             = int64(2048)
	)

	var (
		supportedCodecs          = []topictypes.Codec{topictypes.CodecRaw}
		minActivePartitions      = int64(2)
		maxActivePartitions      = int64(5)
		topicName                = "test-topic-" + t.Name()
		topicPath                = scope.Driver().Name() + "/" + topicName
		autoPartitioningSettings = topictypes.AutoPartitioningSettings{
			AutoPartitioningStrategy: topictypes.AutoPartitioningStrategyScaleUp,
			AutoPartitioningWriteSpeedStrategy: topictypes.AutoPartitioningWriteSpeedStrategy{
				StabilizationWindow:    10 * time.Second,
				UpUtilizationPercent:   80,
				DownUtilizationPercent: 20,
			},
		}
	)

	_ = scope.Driver().Topic().Drop(ctx, topicPath)
	err := scope.Driver().Topic().Create(ctx, topicPath,
		topicoptions.CreateWithSupportedCodecs(supportedCodecs...),
		topicoptions.CreateWithMinActivePartitions(minActivePartitions),
		topicoptions.CreateWithMaxActivePartitions(maxActivePartitions),
		topicoptions.CreateWithAutoPartitioningSettings(autoPartitioningSettings),
		topicoptions.CreateWithPartitionWriteSpeedBytesPerSecond(writeSpeed),
		topicoptions.CreateWithPartitionWriteBurstBytes(burstBytes),
	)
	require.NoError(t, err)

	topicDesc, err := scope.Driver().Topic().Describe(ctx, topicPath)
	require.NoError(t, err)

	expected := topictypes.TopicDescription{
		Path: topicName,
		PartitionSettings: topictypes.PartitionSettings{
			MinActivePartitions:      minActivePartitions,
			MaxActivePartitions:      maxActivePartitions,
			AutoPartitioningSettings: autoPartitioningSettings,
		},
		Partitions: []topictypes.PartitionInfo{
			{
				PartitionID: 0,
				Active:      true,
			},
			{
				PartitionID: 1,
				Active:      true,
			},
		},
		RetentionPeriod:                   defaultRetentionPeriod,
		RetentionStorageMB:                0,
		SupportedCodecs:                   supportedCodecs,
		PartitionWriteBurstBytes:          burstBytes,
		PartitionWriteSpeedBytesPerSecond: writeSpeed,
		Attributes:                        nil,
		Consumers:                         []topictypes.Consumer{},
		MeteringMode:                      topictypes.MeteringModeUnspecified,
	}

	requireAndCleanSubset := func(checked *map[string]string, subset *map[string]string) {
		t.Helper()
		for k, subValue := range *subset {
			checkedValue, ok := (*checked)[k]
			require.True(t, ok, k)
			require.Equal(t, subValue, checkedValue)
		}
		*checked = nil
		*subset = nil
	}

	requireAndCleanSubset(&topicDesc.Attributes, &expected.Attributes)
	for i := range topicDesc.Partitions {
		topicDesc.Partitions[i].FromBound = nil
		topicDesc.Partitions[i].ToBound = nil
	}
	require.Equal(t, expected, topicDesc)
}

// directWriteStreamChecker records how StreamWrite streams are opened on the client.
// It is used when the pinned partition is known up front (WithWriterPartitionID).
// On a single-node cluster matching nodeID to DescribeTopic is tautological; the
// check still verifies that the SDK disables fallback and passes the resolved node ID.
// Multi-writer and probe-and-rebind flows are covered by unit tests and grpc mocks.
type directWriteStreamChecker struct {
	*topicwritetest.StreamWriteRecorder
}

func newDirectWriteStreamChecker() *directWriteStreamChecker {
	return &directWriteStreamChecker{
		StreamWriteRecorder: topicwritetest.NewStreamWriteRecorder(),
	}
}

// Driver opens a dedicated cached driver instance that includes the StreamWrite interceptor.
// Do not pass the option to [scopeT.Driver]: fixenv caches drivers by name only, so opts
// on the default driver are ignored after the first connect.
// Each subtest must pass a unique driverName: the interceptor is bound to this checker
// instance and a reused driver name would record sessions into another checker.
func (c *directWriteStreamChecker) Driver(scope *scopeT, driverName string) *ydb.Driver {
	return scope.driverNamed(driverName, ydb.With(config.WithGrpcOptions(
		grpc.WithChainStreamInterceptor(c.Interceptor()),
	)))
}

func topicPartitionLocation(
	ctx context.Context,
	topicClient topic.Client,
	topicPath string,
	partitionID int64,
) (nodeID uint32, generation int64, err error) {
	desc, err := topicClient.Describe(ctx, topicPath, topicoptions.IncludeLocation())
	if err != nil {
		return 0, 0, err
	}
	for i := range desc.Partitions {
		p := &desc.Partitions[i]
		if p.PartitionID == partitionID {
			if p.Location.NodeID == 0 {
				return 0, 0, fmt.Errorf(
					"partition %d has empty location (IncludeLocation missing on server?)",
					partitionID,
				)
			}

			return p.Location.NodeID, p.Location.Generation, nil
		}
	}

	return 0, 0, fmt.Errorf("partition %d not found in topic %q", partitionID, topicPath)
}

func TestTopicDirectWrite(t *testing.T) {
	scope := newScope(t)
	ctx := scope.Ctx
	const partitionID = int64(0)

	topicPath := scope.TopicPath(
		topicoptions.CreateWithMinActivePartitions(2),
		topicoptions.CreateWithMaxActivePartitions(2),
	)

	t.Run("WriteAndRead", func(t *testing.T) {
		hostNode, generation, err := topicPartitionLocation(ctx, scope.Driver().Topic(), topicPath, partitionID)
		require.NoError(t, err)

		routing := newDirectWriteStreamChecker()
		writer, err := routing.Driver(scope, "direct-write-pinned").Topic().StartWriter(
			topicPath,
			topicoptions.WithWriterPartitionID(partitionID),
			topicoptions.WithWriterDirectWrite(true),
			topicoptions.WithWriterWaitServerAck(true),
		)
		require.NoError(t, err)

		payload := []byte("direct-write-payload")
		require.NoError(t, writer.Write(ctx, topicwriter.Message{Data: bytes.NewReader(payload)}))
		routing.RequireRoutedToNode(t, hostNode)
		routing.RequireInitGeneration(t, partitionID, generation)
		require.NoError(t, writer.Close(ctx))

		reader, err := scope.Driver().Topic().StartReader(
			scope.TopicConsumerName(),
			topicoptions.ReadSelectors{
				{Path: topicPath, Partitions: []int64{partitionID}},
			},
		)
		require.NoError(t, err)
		defer func() { _ = reader.Close(ctx) }()

		msg, err := reader.ReadMessage(ctx)
		require.NoError(t, err)
		require.Equal(t, partitionID, msg.PartitionID())

		got, err := io.ReadAll(msg)
		require.NoError(t, err)
		require.Equal(t, payload, got)
	})

	t.Run("WithoutDirectWrite", func(t *testing.T) {
		routing := newDirectWriteStreamChecker()
		writer, err := routing.Driver(scope, "proxy-write-pinned").Topic().StartWriter(
			topicPath,
			topicoptions.WithWriterPartitionID(partitionID),
			topicoptions.WithWriterWaitServerAck(true),
		)
		require.NoError(t, err)
		defer func() { _ = writer.Close(ctx) }()

		require.NoError(t, writer.Write(ctx, topicwriter.Message{
			Data: bytes.NewReader([]byte("proxy-write")),
		}))
		routing.RequireInitPartitionWithoutGeneration(t, partitionID)
	})

	t.Run("WithAutoProducer", func(t *testing.T) {
		writer, err := scope.Driver().Topic().StartWriter(
			topicPath,
			topicoptions.WithWriterDirectWrite(true),
			topicoptions.WithWriterWaitServerAck(true),
		)
		require.NoError(t, err)
		defer func() { _ = writer.Close(ctx) }()

		require.NoError(t, writer.WaitInit(ctx))
		require.NoError(t, writer.Write(ctx, topicwriter.Message{Data: bytes.NewReader([]byte("direct-write-auto-producer"))}))
	})
}

func TestTopicMultiWriterDirectWrite(t *testing.T) {
	scope := newScope(t)
	ctx := scope.Ctx
	topicPath := scope.TopicPath(
		topicoptions.CreateWithMinActivePartitions(2),
		topicoptions.CreateWithMaxActivePartitions(2),
	)

	writer, err := scope.Driver().Topic().StartWriter(
		topicPath,
		topicoptions.WithWriterWaitServerAck(true),
		topicoptions.WithWriteToManyPartitions(
			topicoptions.WithWriterPartitionByPartitionID(),
			topicoptions.WithMultiWriterDirectWrite(true),
		),
	)
	require.NoError(t, err)

	const messageText = "multi-writer-direct"
	require.NoError(t, writer.Write(ctx,
		topicwriter.Message{PartitionID: 0, Data: strings.NewReader(messageText + "-0")},
		topicwriter.Message{PartitionID: 1, Data: strings.NewReader(messageText + "-1")},
	))
	require.NoError(t, writer.Close(ctx))

	reader, err := scope.Driver().Topic().StartReader(
		scope.TopicConsumerName(),
		topicoptions.ReadSelectors{
			{Path: topicPath, Partitions: []int64{0, 1}},
		},
	)
	require.NoError(t, err)
	defer func() { _ = reader.Close(ctx) }()

	gotByPartition := map[int64]string{}
	for range 2 {
		msg, err := reader.ReadMessage(ctx)
		require.NoError(t, err)
		body, err := io.ReadAll(msg)
		require.NoError(t, err)
		gotByPartition[msg.PartitionID()] = string(body)
	}
	require.Equal(t, messageText+"-0", gotByPartition[0])
	require.Equal(t, messageText+"-1", gotByPartition[1])
}

func TestDescribeTopicConsumer(t *testing.T) {
	ctx := xtest.Context(t)
	db := connect(t)
	topicName := "test-topic-" + t.Name()
	var (
		supportedCodecs     = []topictypes.Codec{topictypes.CodecRaw, topictypes.CodecGzip}
		minActivePartitions = int64(2)
		consumers           = []topictypes.Consumer{
			{
				Name:            "c1",
				Important:       true,
				SupportedCodecs: []topictypes.Codec{topictypes.CodecRaw, topictypes.CodecGzip},
				ReadFrom:        time.Date(2022, 9, 11, 10, 1, 2, 0, time.UTC),
			},
			{
				Name:            "c2",
				Important:       false,
				SupportedCodecs: []topictypes.Codec{},
				ReadFrom:        time.Date(2021, 1, 2, 3, 4, 5, 0, time.UTC),
			},
		}
	)

	_ = db.Topic().Drop(ctx, topicName)
	err := db.Topic().Create(ctx, topicName,
		topicoptions.CreateWithSupportedCodecs(supportedCodecs...),
		topicoptions.CreateWithMinActivePartitions(minActivePartitions),
		topicoptions.CreateWithConsumer(consumers...),
	)
	require.NoError(t, err)

	consumer, err := db.Topic().DescribeTopicConsumer(ctx, topicName, "c1", topicoptions.IncludeConsumerStats())
	require.NoError(t, err)

	zeroTime := time.Time{}
	zeroDuration := time.Duration(0)
	expectedConsumerDesc := topictypes.TopicConsumerDescription{
		Path: path.Join(topicName, "c1"),
		Consumer: topictypes.Consumer{
			Name:            "c1",
			Important:       true,
			SupportedCodecs: []topictypes.Codec{topictypes.CodecRaw, topictypes.CodecGzip},
			ReadFrom:        time.Date(2022, 9, 11, 10, 1, 2, 0, time.UTC),
			Attributes: map[string]string{
				"_service_type": "data-streams",
			},
		},
		Partitions: []topictypes.DescribeConsumerPartitionInfo{
			{
				PartitionID: 0,
				Active:      true,
				PartitionStats: topictypes.PartitionStats{
					PartitionsOffset: topictypes.OffsetRange{},
					StoreSizeBytes:   0,
					LastWriteTime:    nil,
					MaxWriteTimeLag:  &zeroDuration,
					BytesWritten:     topictypes.MultipleWindowsStat{},
				},
				PartitionConsumerStats: topictypes.PartitionConsumerStats{
					LastReadOffset:                 0,
					CommittedOffset:                0,
					ReadSessionID:                  "",
					PartitionReadSessionCreateTime: &zeroTime,
					LastReadTime:                   &zeroTime,
					MaxReadTimeLag:                 &zeroDuration,
					MaxWriteTimeLag:                &zeroDuration,
					BytesRead:                      topictypes.MultipleWindowsStat{},
					ReaderName:                     "",
				},
			},
			{
				PartitionID: 1,
				Active:      true,
				PartitionStats: topictypes.PartitionStats{
					PartitionsOffset: topictypes.OffsetRange{},
					StoreSizeBytes:   0,
					LastWriteTime:    nil,
					MaxWriteTimeLag:  &zeroDuration,
					BytesWritten:     topictypes.MultipleWindowsStat{},
				},
				PartitionConsumerStats: topictypes.PartitionConsumerStats{
					LastReadOffset:                 0,
					CommittedOffset:                0,
					ReadSessionID:                  "",
					PartitionReadSessionCreateTime: &zeroTime,
					LastReadTime:                   &zeroTime,
					MaxReadTimeLag:                 &zeroDuration,
					MaxWriteTimeLag:                &zeroDuration,
					BytesRead:                      topictypes.MultipleWindowsStat{},
					ReaderName:                     "",
				},
			},
		},
	}

	requireAndCleanSubset := func(checked *map[string]string, subset *map[string]string) {
		t.Helper()
		for k, subValue := range *subset {
			checkedValue, ok := (*checked)[k]
			require.True(t, ok, k)
			require.Equal(t, subValue, checkedValue)
		}
		*checked = nil
		*subset = nil
	}

	requireAndCleanSubset(&consumer.Consumer.Attributes, &expectedConsumerDesc.Consumer.Attributes)

	for i := range consumer.Partitions {
		// Fields that are checked here are dynamic and they change with time, so we need to set them to expected values
		// to make the comparison possible.
		p := &consumer.Partitions[i]

		require.NotNil(t, p.PartitionStats.LastWriteTime)

		p.PartitionStats.LastWriteTime = expectedConsumerDesc.Partitions[i].PartitionStats.LastWriteTime
		p.PartitionStats.MaxWriteTimeLag = expectedConsumerDesc.Partitions[i].PartitionStats.MaxWriteTimeLag

		require.NotNil(t, p.PartitionConsumerStats.LastReadTime)
		require.NotNil(t, p.PartitionConsumerStats.MaxReadTimeLag)
		require.NotNil(t, p.PartitionConsumerStats.MaxWriteTimeLag)

		p.PartitionConsumerStats.PartitionReadSessionCreateTime = expectedConsumerDesc.Partitions[i].PartitionConsumerStats.PartitionReadSessionCreateTime
		p.PartitionConsumerStats.LastReadTime = expectedConsumerDesc.Partitions[i].PartitionConsumerStats.LastReadTime
		p.PartitionConsumerStats.MaxReadTimeLag = expectedConsumerDesc.Partitions[i].PartitionConsumerStats.MaxReadTimeLag
		p.PartitionConsumerStats.MaxWriteTimeLag = expectedConsumerDesc.Partitions[i].PartitionConsumerStats.MaxWriteTimeLag
	}

	require.Equal(t, expectedConsumerDesc, consumer)
}

func TestSchemeList(t *testing.T) {
	ctx := xtest.Context(t)
	db := connect(t)

	topicPath := createTopic(ctx, t, db)
	list, err := db.Scheme().ListDirectory(ctx, db.Name())
	require.NoError(t, err)

	topicName := path.Base(topicPath)

	hasTopic := false
	for _, e := range list.Children {
		if e.IsTopic() && topicName == e.Name {
			hasTopic = true
		}
	}
	require.True(t, hasTopic)
}

func TestReaderWithoutConsumer(t *testing.T) {
	t.Run("OK", func(t *testing.T) {
		scope := newScope(t)
		ctx := scope.Ctx

		reader1, err := scope.Driver().Topic().StartReader(
			"",
			topicoptions.ReadSelectors{
				{
					Path:       scope.TopicPath(),
					Partitions: []int64{0},
				},
			},
			topicoptions.WithReaderWithoutConsumer(false),
		)
		require.NoError(t, err)

		reader2, err := scope.Driver().Topic().StartReader(
			"",
			topicoptions.ReadSelectors{
				{
					Path:       scope.TopicPath(),
					Partitions: []int64{0},
				},
			},
			topicoptions.WithReaderWithoutConsumer(false),
		)
		require.NoError(t, err)

		err = scope.TopicWriter().Write(ctx, topicwriter.Message{Data: strings.NewReader("123")})
		require.NoError(t, err)

		msg1, err := reader1.ReadMessage(ctx)
		require.NoError(t, err)
		require.Equal(t, int64(1), msg1.SeqNo)

		msg1data, err := io.ReadAll(msg1)
		require.NoError(t, err)
		require.Equal(t, "123", string(msg1data))

		msg2, err := reader2.ReadMessage(ctx)
		require.NoError(t, err)
		require.Equal(t, int64(1), msg2.SeqNo)

		msg2data, err := io.ReadAll(msg2)
		require.NoError(t, err)
		require.Equal(t, "123", string(msg2data))

		_ = reader1.Close(ctx)
		_ = reader2.Close(ctx)
	})
	t.Run("NoNameNoOptionErr", func(t *testing.T) {
		scope := newScope(t)
		topicReader, err := scope.Driver().Topic().StartReader("", topicoptions.ReadTopic(scope.TopicPath()))
		require.Error(t, err)
		require.Nil(t, topicReader)
	})
	t.Run("NameAndOption", func(t *testing.T) {
		scope := newScope(t)
		topicReader, err := scope.Driver().Topic().StartReader(
			scope.TopicConsumerName(),
			topicoptions.ReadTopic(scope.TopicPath()),
			topicoptions.WithReaderWithoutConsumer(false),
		)
		require.Error(t, err)
		require.Nil(t, topicReader)
	})
}

func TestAlterTopic(t *testing.T) {
	scope := newScope(t)
	ctx := scope.Ctx

	const (
		defaultRetentionPeriod = 24 * time.Hour
		writeSpeed             = int64(1024)
		burstBytes             = int64(2048)
	)

	var (
		supportedCodecs     = []topictypes.Codec{topictypes.CodecRaw}
		minActivePartitions = int64(2)
		maxActivePartitions = int64(5)
		topicName           = "test-topic-" + t.Name()
		topicPath           = scope.Driver().Name() + "/" + topicName
	)

	_ = scope.Driver().Topic().Drop(ctx, topicPath)
	err := scope.Driver().Topic().Create(ctx, topicPath,
		topicoptions.CreateWithSupportedCodecs(supportedCodecs...),
		topicoptions.CreateWithMinActivePartitions(minActivePartitions),
		topicoptions.CreateWithMaxActivePartitions(maxActivePartitions),
		topicoptions.CreateWithPartitionWriteSpeedBytesPerSecond(writeSpeed),
		topicoptions.CreateWithPartitionWriteBurstBytes(burstBytes),
	)
	require.NoError(t, err)

	// First alter: Change auto partitioning settings
	err = scope.Driver().Topic().Alter(ctx, topicPath,
		topicoptions.AlterWithAutoPartitioningStrategy(topictypes.AutoPartitioningStrategyScaleUp),
		topicoptions.AlterWithAutoPartitioningWriteSpeedStabilizationWindow(10*time.Second),
		topicoptions.AlterWithAutoPartitioningWriteSpeedUpUtilizationPercent(80),
		topicoptions.AlterWithAutoPartitioningWriteSpeedDownUtilizationPercent(20),
	)
	require.NoError(t, err)

	// Second alter: Change retention period and supported codecs
	newRetentionPeriod := 48 * time.Hour
	newSupportedCodecs := []topictypes.Codec{topictypes.CodecRaw, topictypes.CodecGzip}
	err = scope.Driver().Topic().Alter(ctx, topicPath,
		topicoptions.AlterWithRetentionPeriod(newRetentionPeriod),
		topicoptions.AlterWithSupportedCodecs(newSupportedCodecs...),
	)
	require.NoError(t, err)

	// Third alter: Change write speed and burst bytes
	newWriteSpeed := int64(2048)
	newBurstBytes := int64(4096)
	err = scope.Driver().Topic().Alter(ctx, topicPath,
		topicoptions.AlterWithPartitionWriteSpeedBytesPerSecond(newWriteSpeed),
		topicoptions.AlterWithPartitionWriteBurstBytes(newBurstBytes),
	)
	require.NoError(t, err)

	// Verify all changes
	topicDesc, err := scope.Driver().Topic().Describe(ctx, topicPath)
	require.NoError(t, err)

	expected := topictypes.TopicDescription{
		Path: topicName,
		PartitionSettings: topictypes.PartitionSettings{
			MinActivePartitions: minActivePartitions,
			MaxActivePartitions: maxActivePartitions,
			AutoPartitioningSettings: topictypes.AutoPartitioningSettings{
				AutoPartitioningStrategy: topictypes.AutoPartitioningStrategyScaleUp,
				AutoPartitioningWriteSpeedStrategy: topictypes.AutoPartitioningWriteSpeedStrategy{
					StabilizationWindow:    10 * time.Second,
					UpUtilizationPercent:   80,
					DownUtilizationPercent: 20,
				},
			},
		},
		Partitions: []topictypes.PartitionInfo{
			{
				PartitionID: 0,
				Active:      true,
			},
			{
				PartitionID: 1,
				Active:      true,
			},
		},
		RetentionPeriod:                   newRetentionPeriod,
		RetentionStorageMB:                0,
		SupportedCodecs:                   newSupportedCodecs,
		PartitionWriteBurstBytes:          newBurstBytes,
		PartitionWriteSpeedBytesPerSecond: newWriteSpeed,
		Attributes:                        nil,
		Consumers:                         []topictypes.Consumer{},
		MeteringMode:                      topictypes.MeteringModeUnspecified,
	}

	requireAndCleanSubset := func(checked *map[string]string, subset *map[string]string) {
		t.Helper()
		for k, subValue := range *subset {
			checkedValue, ok := (*checked)[k]
			require.True(t, ok, k)
			require.Equal(t, subValue, checkedValue)
		}
		*checked = nil
		*subset = nil
	}

	requireAndCleanSubset(&topicDesc.Attributes, &expected.Attributes)
	for i := range topicDesc.Partitions {
		topicDesc.Partitions[i].FromBound = nil
		topicDesc.Partitions[i].ToBound = nil
	}
	require.Equal(t, expected, topicDesc)

	// Test altering max active partitions
	newMaxActivePartitions := int64(10)
	err = scope.Driver().Topic().Alter(ctx, topicPath,
		topicoptions.AlterWithMaxActivePartitions(newMaxActivePartitions),
	)
	require.NoError(t, err)

	// Verify max active partitions
	topicDesc, err = scope.Driver().Topic().Describe(ctx, topicPath)
	require.NoError(t, err)
	require.Equal(t, newMaxActivePartitions, topicDesc.PartitionSettings.MaxActivePartitions)

	// Test altering min active partitions
	newMinActivePartitions := int64(3)
	err = scope.Driver().Topic().Alter(ctx, topicPath,
		topicoptions.AlterWithMinActivePartitions(newMinActivePartitions),
	)
	require.NoError(t, err)

	time.Sleep(3 * time.Second)

	// Verify min active partitions
	xtest.SpinWaitConditionWithTimeout(t, nil, time.Second, func() bool {
		topicDesc, err = scope.Driver().Topic().Describe(ctx, topicPath)
		require.NoError(t, err)
		return newMinActivePartitions == topicDesc.PartitionSettings.MinActivePartitions
	})
	topicDesc, err = scope.Driver().Topic().Describe(ctx, topicPath)
	require.NoError(t, err)
	require.Equal(t, newMinActivePartitions, topicDesc.PartitionSettings.MinActivePartitions)
}

func TestTopicMetricsLevel(t *testing.T) {
	if ydbVersion := os.Getenv("YDB_VERSION"); ydbVersion != "" &&
		ydbVersion != "nightly" && ydbVersion != "edge" && ydbVersion != "latest" &&
		version.Lt(ydbVersion, "25.4") {
		t.Skip("require support for topic metrics_level")
	}

	scope := newScope(t)
	ctx := scope.Ctx

	const initialLevel = uint32(3)

	topicPath := scope.TopicPath(
		topicoptions.CreateWithMinActivePartitions(1),
		topicoptions.CreateWithMaxActivePartitions(1),
		topicoptions.CreateWithMetricsLevel(initialLevel),
	)

	desc, err := scope.Driver().Topic().Describe(ctx, topicPath)
	require.NoError(t, err)
	require.NotNil(t, desc.MetricsLevel, "metrics_level must be populated after create")
	require.Equal(t, initialLevel, *desc.MetricsLevel)

	const updatedLevel = uint32(5)
	err = scope.Driver().Topic().Alter(ctx, topicPath,
		topicoptions.AlterWithSetMetricsLevel(updatedLevel),
	)
	require.NoError(t, err)

	desc, err = scope.Driver().Topic().Describe(ctx, topicPath)
	require.NoError(t, err)
	require.NotNil(t, desc.MetricsLevel, "metrics_level must be populated after alter set")
	require.Equal(t, updatedLevel, *desc.MetricsLevel)

	err = scope.Driver().Topic().Alter(ctx, topicPath,
		topicoptions.AlterWithResetMetricsLevel(),
	)
	require.NoError(t, err)

	desc, err = scope.Driver().Topic().Describe(ctx, topicPath)
	require.NoError(t, err)
	require.Nil(t, desc.MetricsLevel, "metrics_level must be reset to the database default")
}

func connect(t testing.TB, opts ...ydb.Option) *ydb.Driver {
	return connectWithLogOption(t, false, opts...)
}

func connectWithLogOption(t testing.TB, logGRPC bool, opts ...ydb.Option) *ydb.Driver {
	connectionString := defaultConnectionString
	if cs := os.Getenv("YDB_CONNECTION_STRING"); cs != "" {
		connectionString = cs
	}

	var grpcOptions []grpc.DialOption
	const needLogGRPCMessages = true
	if logGRPC {
		grpcOptions = append(grpcOptions,
			grpc.WithChainUnaryInterceptor(xtest.NewGrpcLogger(t).UnaryClientInterceptor),
			grpc.WithChainStreamInterceptor(xtest.NewGrpcLogger(t).StreamClientInterceptor),
		)
	}

	ydbOpts := []ydb.Option{
		ydb.WithDialTimeout(time.Second),
		ydb.WithAccessTokenCredentials(os.Getenv("YDB_ACCESS_TOKEN_CREDENTIALS")),
		ydb.With(config.WithGrpcOptions(grpcOptions...)),
	}
	ydbOpts = append(ydbOpts, opts...)

	db, err := ydb.Open(context.Background(), connectionString, ydbOpts...)
	require.NoError(t, err)

	t.Cleanup(func() {
		_ = db.Close(context.Background())
	})
	return db
}
