package ticdc

import (
	"context"
	"encoding/json"
	"strings"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	dataflowng "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	serverlessrpc "github.com/tidbcloud/serverless-service/idl/pbgen/proto/rpc"
	"google.golang.org/grpc"
	"google.golang.org/grpc/metadata"
	"google.golang.org/protobuf/proto"
	"google.golang.org/protobuf/types/known/timestamppb"
)

type dedicatedBusinessStub struct {
	listResponse *dataflow.Connectors
	detail       *dataflow.Connector
	listRequest  *dataflow.ConnectorListReq
	detailID     uint64
	metadata     metadata.MD
}

func (s *dedicatedBusinessStub) ListConnectors(ctx context.Context, request *dataflow.ConnectorListReq, _ ...grpc.CallOption) (*dataflow.Connectors, error) {
	s.listRequest = request
	s.metadata, _ = metadata.FromOutgoingContext(ctx)
	return s.listResponse, nil
}

func (s *dedicatedBusinessStub) GetConnector(ctx context.Context, request *dataflow.ConnectorID, _ ...grpc.CallOption) (*dataflow.Connector, error) {
	s.detailID = request.GetConnectorId()
	s.metadata, _ = metadata.FromOutgoingContext(ctx)
	return s.detail, nil
}

type dedicatedInternalStub struct {
	listResponse *dataflow.ListInternalChangefeedsResponse
	detail       *dataflow.InternalChangefeed
	listErr      error
}

func (s *dedicatedInternalStub) ListInternalChangefeeds(context.Context, *dataflow.ListInternalChangefeedsRequest, ...grpc.CallOption) (*dataflow.ListInternalChangefeedsResponse, error) {
	return s.listResponse, s.listErr
}

func (s *dedicatedInternalStub) GetInternalChangefeed(context.Context, *dataflow.GetInternalChangefeedRequest, ...grpc.CallOption) (*dataflow.InternalChangefeed, error) {
	return s.detail, nil
}

type nextGenBusinessStub struct {
	listResponse *dataflowng.Changefeeds
	listCalls    int
}

func (s *nextGenBusinessStub) ListChangefeeds(context.Context, *dataflowng.ChangefeedListReq, ...grpc.CallOption) (*dataflowng.Changefeeds, error) {
	s.listCalls++
	return s.listResponse, nil
}

func (s *nextGenBusinessStub) GetChangefeed(context.Context, *dataflowng.ChangefeedID, ...grpc.CallOption) (*dataflowng.Changefeed, error) {
	return nil, nil
}

type nextGenInternalStub struct {
	listResponse *dataflowng.ListInternalChangefeedsResponse
}

type essentialChangefeedStub struct {
	listResponse *serverlessrpc.ListClusterChangefeedsResponse
	getResponse  *serverlessrpc.ChangefeedInfo
	listRequest  *serverlessrpc.ListClusterChangefeedsRequest
	getRequest   *serverlessrpc.GetChangefeedRequest
	metadata     metadata.MD
}

func (s *essentialChangefeedStub) ListClusterChangefeeds(ctx context.Context, request *serverlessrpc.ListClusterChangefeedsRequest, _ ...grpc.CallOption) (*serverlessrpc.ListClusterChangefeedsResponse, error) {
	s.listRequest = request
	s.metadata, _ = metadata.FromOutgoingContext(ctx)
	return s.listResponse, nil
}

func (s *essentialChangefeedStub) GetChangefeed(ctx context.Context, request *serverlessrpc.GetChangefeedRequest, _ ...grpc.CallOption) (*serverlessrpc.ChangefeedInfo, error) {
	s.getRequest = request
	s.metadata, _ = metadata.FromOutgoingContext(ctx)
	return s.getResponse, nil
}

func (s *nextGenInternalStub) ListInternalChangefeeds(context.Context, *dataflowng.ListInternalChangefeedsRequest, ...grpc.CallOption) (*dataflowng.ListInternalChangefeedsResponse, error) {
	return s.listResponse, nil
}

func (s *nextGenInternalStub) GetInternalChangefeed(context.Context, *dataflowng.GetInternalChangefeedRequest, ...grpc.CallOption) (*dataflowng.InternalChangefeed, error) {
	return nil, nil
}

func TestServiceDedicatedListUsesBusinessPageAndEnrichesAndSorts(t *testing.T) {
	business := &dedicatedBusinessStub{listResponse: &dataflow.Connectors{Items: []*dataflow.Connector{
		{ConnectorId: 10, Name: "older", CreatedAt: timestamppb.New(time.Unix(10, 0)), Request: &dataflow.ConnectorCreateReq{ClusterId: 456}},
		{ConnectorId: 20, Name: "newer", CreatedAt: timestamppb.New(time.Unix(20, 0)), Request: &dataflow.ConnectorCreateReq{ClusterId: 456}},
	}}}
	internal := &dedicatedInternalStub{listResponse: &dataflow.ListInternalChangefeedsResponse{Changefeeds: []*dataflow.InternalChangefeed{
		{Id: 100, DisplayId: 10},
	}}}
	service := NewService(&Clients{dedicatedBusiness: business, dedicatedInternal: internal})

	response, err := service.List(context.Background(), 123, 0, 456, ArchitectureDedicated)
	require.NoError(t, err)
	require.Equal(t, uint32(changefeedListPageSize), business.listRequest.GetPerPage())
	require.Equal(t, uint32(1), business.listRequest.GetPage())
	require.Equal(t, []string{"newer", "older"}, []string{response.Changefeeds[0].DisplayName, response.Changefeeds[1].DisplayName})
	require.Empty(t, response.Changefeeds[0].ID)
	require.Equal(t, "100", response.Changefeeds[1].ID)
	require.Len(t, business.metadata.Get(dataflowUserMetadataKey), 1)
}

func TestServiceDedicatedDetailRejectsCrossClusterKernelID(t *testing.T) {
	business := &dedicatedBusinessStub{detail: &dataflow.Connector{
		ConnectorId: 10,
		Request:     &dataflow.ConnectorCreateReq{ClusterId: 999},
	}}
	internal := &dedicatedInternalStub{detail: &dataflow.InternalChangefeed{Id: 100, DisplayId: 10}}
	service := NewService(&Clients{dedicatedBusiness: business, dedicatedInternal: internal})

	response, err := service.Detail(context.Background(), 123, 456, 456, "100", ArchitectureDedicated)
	require.Nil(t, response)
	require.ErrorIs(t, err, ErrNotFound)
	require.Equal(t, uint64(10), business.detailID)
}

func TestServiceEssentialListUsesServerlessWithoutDataflowMetadata(t *testing.T) {
	essential := &essentialChangefeedStub{listResponse: &serverlessrpc.ListClusterChangefeedsResponse{Items: []*serverlessrpc.ChangefeedInfo{
		{ClusterId: "456", ChangefeedId: "essential-new", DisplayName: "newer", CreatedAt: "2026-08-31T04:00:00Z", Rcu: 4},
		{ClusterId: "456", ChangefeedId: "essential-old", DisplayName: "older", CreatedAt: "2026-08-30 04:00:00", Rcu: 2},
	}}}
	service := NewService(&Clients{essential: essential})

	response, err := service.List(context.Background(), 123, 789, 456, ArchitectureEssential)
	require.NoError(t, err)
	require.Equal(t, "456", essential.listRequest.GetClusterId())
	require.Equal(t, int32(changefeedListPageSize), essential.listRequest.GetLimit())
	require.Zero(t, essential.listRequest.GetOffset())
	require.Empty(t, essential.metadata.Get(dataflowUserMetadataKey))
	require.Equal(t, []string{"essential-new", "essential-old"}, []string{response.Changefeeds[0].ID, response.Changefeeds[1].ID})
	require.Equal(t, "4", response.Changefeeds[0].RCU)
	require.Equal(t, "CHANGEFEED_ESSENTIAL", response.Changefeeds[0].ClusterType)
}

func TestServiceEssentialDetailRejectsCrossClusterChangefeed(t *testing.T) {
	essential := &essentialChangefeedStub{getResponse: &serverlessrpc.ChangefeedInfo{
		ClusterId: "999", ChangefeedId: "essential-changefeed",
	}}
	service := NewService(&Clients{essential: essential})

	response, err := service.Detail(context.Background(), 123, 789, 456, "essential-changefeed", ArchitectureEssential)
	require.Nil(t, response)
	require.ErrorIs(t, err, ErrNotFound)
	require.Equal(t, "456", essential.getRequest.GetClusterId())
	require.Equal(t, "essential-changefeed", essential.getRequest.GetChangefeedId())
	require.Empty(t, essential.metadata.Get(dataflowUserMetadataKey))
}

func TestServiceNextGenListUsesNextGenClients(t *testing.T) {
	business := &nextGenBusinessStub{listResponse: &dataflowng.Changefeeds{Items: []*dataflowng.Changefeed{
		{ChangefeedId: 20, Name: "premium", Request: &dataflowng.ChangefeedCreateReq{ClusterId: 456}},
	}}}
	internal := &nextGenInternalStub{listResponse: &dataflowng.ListInternalChangefeedsResponse{Changefeeds: []*dataflowng.InternalChangefeed{
		{Id: 200, DisplayId: 20},
	}}}
	service := NewService(&Clients{nextGenBusiness: business, nextGenInternal: internal})

	response, err := service.List(context.Background(), 123, 456, 456, ArchitectureNextGen)
	require.NoError(t, err)
	require.Equal(t, 1, business.listCalls)
	require.Len(t, response.Changefeeds, 1)
	require.Equal(t, "200", response.Changefeeds[0].ID)
	require.Equal(t, string(ArchitectureNextGen), response.Changefeeds[0].Arch)
}

func TestServiceListKeepsBusinessResultsWhenInternalEnrichmentFails(t *testing.T) {
	business := &dedicatedBusinessStub{listResponse: &dataflow.Connectors{Items: []*dataflow.Connector{
		{ConnectorId: 10, Name: "available", Request: &dataflow.ConnectorCreateReq{ClusterId: 456}},
	}}}
	internal := &dedicatedInternalStub{listErr: context.DeadlineExceeded}
	service := NewService(&Clients{dedicatedBusiness: business, dedicatedInternal: internal})

	response, err := service.List(context.Background(), 123, 456, 456, ArchitectureDedicated)
	require.NoError(t, err)
	require.Len(t, response.Changefeeds, 1)
	require.Equal(t, "available", response.Changefeeds[0].DisplayName)
	require.Empty(t, response.Changefeeds[0].ID)
}

func TestDedicatedDownstreamConfigRemovesCredentialsAndDuplicateSections(t *testing.T) {
	changefeed, err := fullDedicatedChangefeed(&dataflow.Connector{
		ConnectorId:   10,
		ConnectorType: dataflow.ConnectorType_MYSQL,
		Request: &dataflow.ConnectorCreateReq{
			ClusterId: 456,
			MysqlInfo: &dataflow.MySQLInfo{
				Connection:    &dataflow.Connection{Endpoint: "db.internal:4000", Username: "cdc", Password: "super-secret"},
				TableConfig:   &dataflow.TableConfig{FilterRule: []string{"app.*"}},
				StartPosition: &dataflow.StartPosition{Tso: 123},
			},
		},
	})
	require.NoError(t, err)
	require.Equal(t, "db.internal:4000", changefeed.DownstreamConfig["connection"].(map[string]interface{})["endpoint"])
	require.NotContains(t, changefeed.DownstreamConfig["connection"].(map[string]interface{}), "password")
	require.NotContains(t, changefeed.DownstreamConfig, "table_config")
	require.NotContains(t, changefeed.DownstreamConfig, "start_position")

	encoded, err := json.Marshal(changefeed)
	require.NoError(t, err)
	require.NotContains(t, strings.ToLower(string(encoded)), "super-secret")
}

func TestSanitizeDownstreamConfigRemovesKnownSinkCredentials(t *testing.T) {
	tests := []struct {
		name    string
		message proto.Message
		secrets []string
	}{
		{
			name: "S3 access keys",
			message: &dataflow.S3Info{Endpoint: &dataflow.S3Endpoint{
				S3Uri: "s3://bucket/path", AccessKey: "access-key-value", SecretKey: "secret-key-value",
			}},
			secrets: []string{"access-key-value", "secret-key-value"},
		},
		{
			name: "Azure SAS token",
			message: &dataflow.AzureBlobStorageInfo{Endpoint: &dataflow.AzureBlobStorageEndpoint{
				BlobUri: "https://account.blob.core.windows.net/container", SasToken: "sas-token-value",
			}},
			secrets: []string{"sas-token-value"},
		},
		{
			name: "Pulsar password and token",
			message: &dataflow.PulsarInfo{Authentication: &dataflow.PulsarAuthentication{
				UserName: "pulsar-user", Password: "pulsar-password", Token: "pulsar-token",
			}},
			secrets: []string{"pulsar-password", "pulsar-token"},
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			config, err := sanitizeDownstreamConfig(test.message)
			require.NoError(t, err)
			encoded, err := json.Marshal(config)
			require.NoError(t, err)
			for _, secret := range test.secrets {
				require.NotContains(t, string(encoded), secret)
			}
		})
	}
}

func TestSanitizeConfigMapRemovesNestedCredentialNames(t *testing.T) {
	config := map[string]interface{}{
		"safe": "visible",
		"nested": map[string]interface{}{
			"private_key":  "private-key-value",
			"clientSecret": "client-secret-value",
			"credentials":  "credential-value",
		},
	}

	sanitizeConfigMap(config)
	require.Equal(t, "visible", config["safe"])
	require.Empty(t, config["nested"].(map[string]interface{}))
}

func TestChangefeedStatusConversionPreservesWarningAndError(t *testing.T) {
	require.Equal(t, "warning", dedicatedStatus(dataflow.Connector_WARNING))
	require.Equal(t, "error", dedicatedStatus(dataflow.Connector_ERROR))
	require.Equal(t, "warning", nextGenStatus(dataflowng.Changefeed_WARNING))
	require.Equal(t, "error", nextGenStatus(dataflowng.Changefeed_ERROR))
}