package domain

import (
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	dataflow "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	"google.golang.org/protobuf/types/known/timestamppb"
)

func TestBizChangefeed_GetCheckpointUTCTime(t *testing.T) {
	tests := []struct {
		name          string
		checkpointTSO uint64
		want          time.Time
	}{
		{
			name:          "valid TSO",
			checkpointTSO: 439804651110400000,       // A sample TSO
			want:          time.Unix(1677721600, 0), // Equivalent UTC time
		},
		{
			name:          "zero TSO",
			checkpointTSO: 0,
			want:          time.Unix(0, 0),
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			c := &BizChangefeed{
				CheckpointTSO: tt.checkpointTSO,
			}
			got := c.GetCheckpointUTCTime()
			// We compare only the Unix timestamp to avoid timezone issues
			assert.Equal(t, tt.want.Unix(), got.Unix())
		})
	}
}

func TestBizChangefeed_GetConnectorStatus(t *testing.T) {
	tests := []struct {
		name            string
		operationStatus entity.BizChangefeedOperationStatus
		kernelStatus    entity.ChangefeedKernelStatus
		want            dataflow.Changefeed_Status
	}{
		{
			name:            "normal operation status",
			operationStatus: entity.BizChangefeedOperationStatusNormal,
			kernelStatus:    entity.ChangefeedKernelStatusNormal,
			want:            dataflow.Changefeed_RUNNING,
		},
		{
			name:            "creating operation status",
			operationStatus: entity.BizChangefeedOperationStatusCreating,
			kernelStatus:    entity.ChangefeedKernelStatusNormal,
			want:            dataflow.Changefeed_CREATING,
		},
		{
			name:            "scaling operation status",
			operationStatus: entity.BizChangefeedOperationStatusScaling,
			kernelStatus:    entity.ChangefeedKernelStatusNormal,
			want:            dataflow.Changefeed_SCALING,
		},
		{
			name:            "paused operation status",
			operationStatus: entity.BizChangefeedOperationStatusPaused,
			kernelStatus:    entity.ChangefeedKernelStatusNormal,
			want:            dataflow.Changefeed_PAUSED,
		},
		{
			name:            "paused operation status with kernel failed",
			operationStatus: entity.BizChangefeedOperationStatusPaused,
			kernelStatus:    entity.ChangefeedKernelStatusFailed,
			want:            dataflow.Changefeed_ERROR,
		},
		{
			name:            "paused operation status with kernel error",
			operationStatus: entity.BizChangefeedOperationStatusPaused,
			kernelStatus:    entity.ChangefeedKernelStatusError,
			want:            dataflow.Changefeed_ERROR,
		},
		{
			name:            "failed operation status",
			operationStatus: entity.BizChangefeedOperationStatusFailed,
			kernelStatus:    entity.ChangefeedKernelStatusNormal,
			want:            dataflow.Changefeed_FAILED,
		},
		{
			name:            "deleting operation status",
			operationStatus: entity.BizChangefeedOperationStatusDeleting,
			kernelStatus:    entity.ChangefeedKernelStatusNormal,
			want:            dataflow.Changefeed_DELETING,
		},
		{
			name:            "deleted operation status",
			operationStatus: entity.BizChangefeedOperationStatusDeleted,
			kernelStatus:    entity.ChangefeedKernelStatusNormal,
			want:            dataflow.Changefeed_DELETED,
		},
		{
			name:            "kernel warning status",
			operationStatus: entity.BizChangefeedOperationStatusNormal,
			kernelStatus:    entity.ChangefeedKernelStatusWarning,
			want:            dataflow.Changefeed_WARNING,
		},
		{
			name:            "kernel error status",
			operationStatus: entity.BizChangefeedOperationStatusNormal,
			kernelStatus:    entity.ChangefeedKernelStatusError,
			want:            dataflow.Changefeed_ERROR,
		},
		{
			name:            "kernel failed status",
			operationStatus: entity.BizChangefeedOperationStatusNormal,
			kernelStatus:    entity.ChangefeedKernelStatusFailed,
			want:            dataflow.Changefeed_ERROR,
		},
		{
			name:            "kernel stopped status",
			operationStatus: entity.BizChangefeedOperationStatusNormal,
			kernelStatus:    entity.ChangefeedKernelStatusStopped,
			want:            dataflow.Changefeed_PAUSED,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			c := &BizChangefeed{
				OperationStatus:  tt.operationStatus,
				ChangefeedStatus: tt.kernelStatus,
			}
			got := c.GetConnectorStatus()
			assert.Equal(t, tt.want, got)
		})
	}
}

func TestBizChangefeed_ToDataflowConnector(t *testing.T) {
	now := time.Now()
	checkpointTSO := uint64(439804651110400000)

	c := &BizChangefeed{
		BizChangefeedID:     12345,
		Name:                "test-changefeed",
		CreatedAt:           now,
		SpecificationCode:   "spec-code-1",
		DownstreamType:      entity.TiCDCDownStreamTypeMySQL,
		DownstreamParameter: "param",
		NetworkType:         entity.ChangefeedDownstreamNetworkTypeInternet,
		NetworkInfo:         "network-info",
		OperationStatus:     entity.BizChangefeedOperationStatusNormal,
		ChangefeedStatus:    entity.ChangefeedKernelStatusNormal,
		CheckpointTSO:       checkpointTSO,
		ErrorMessage:        "no error",
		ClusterID:           67890,
		TenantID:            11111,
	}

	got := c.ToDataflowConnector()

	assert.Equal(t, c.BizChangefeedID, got.ChangefeedId)
	assert.Equal(t, dataflow.ChangefeedType_MYSQL, got.ChangefeedType)
	assert.Equal(t, c.Name, got.Name)
	assert.Equal(t, c.CheckpointTSO, got.CheckpointTso)
	assert.Equal(t, dataflow.Changefeed_RUNNING, got.Status)
	assert.Equal(t, c.ErrorMessage, got.Error)
	assert.Equal(t, c.SpecificationCode, got.ChangefeedSpec)

	// Check timestamps
	assert.Equal(t, timestamppb.New(c.CreatedAt), got.CreatedAt)
	expectedCheckpointUTC := timestamppb.New(c.GetCheckpointUTCTime())
	assert.Equal(t, expectedCheckpointUTC, got.CheckpointUtc)
}

func TestBizChangefeed_ToDataflowConnector_ZeroCheckpoint(t *testing.T) {
	c := &BizChangefeed{
		BizChangefeedID:     12345,
		Name:                "test-changefeed",
		CreatedAt:           time.Now(),
		SpecificationCode:   "spec-code-1",
		DownstreamType:      entity.TiCDCDownStreamTypeMySQL,
		DownstreamParameter: "param",
		NetworkType:         entity.ChangefeedDownstreamNetworkTypeInternet,
		NetworkInfo:         "network-info",
		OperationStatus:     entity.BizChangefeedOperationStatusNormal,
		ChangefeedStatus:    entity.ChangefeedKernelStatusNormal,
		CheckpointTSO:       0, // Zero checkpoint TSO
		ErrorMessage:        "no error",
		ClusterID:           67890,
		TenantID:            11111,
	}

	got := c.ToDataflowConnector()

	assert.Equal(t, c.BizChangefeedID, got.ChangefeedId)
	assert.Equal(t, dataflow.ChangefeedType_MYSQL, got.ChangefeedType)
	assert.Equal(t, c.Name, got.Name)
	assert.Equal(t, uint64(0), got.CheckpointTso)
	assert.Equal(t, dataflow.Changefeed_RUNNING, got.Status)
	assert.Equal(t, c.ErrorMessage, got.Error)
	assert.Equal(t, c.SpecificationCode, got.ChangefeedSpec)

	// With zero checkpoint TSO, CheckpointUtc should not be set
	assert.Nil(t, got.CheckpointUtc)
}

func TestListBizChangefeedsResp_ToDataflowConnectors(t *testing.T) {
	now := time.Now()

	resp := &ListBizChangefeedsResp{
		BizChangefeeds: []*BizChangefeed{
			{
				BizChangefeedID:     12345,
				Name:                "test-changefeed-1",
				CreatedAt:           now,
				SpecificationCode:   "spec-code-1",
				DownstreamType:      entity.TiCDCDownStreamTypeMySQL,
				DownstreamParameter: "param1",
				NetworkType:         entity.ChangefeedDownstreamNetworkTypeInternet,
				NetworkInfo:         "network-info-1",
				OperationStatus:     entity.BizChangefeedOperationStatusNormal,
				ChangefeedStatus:    entity.ChangefeedKernelStatusNormal,
				CheckpointTSO:       439804651110400000,
				ErrorMessage:        "no error",
				ClusterID:           67890,
				TenantID:            11111,
			},
			{
				BizChangefeedID:     12346,
				Name:                "test-changefeed-2",
				CreatedAt:           now,
				SpecificationCode:   "spec-code-2",
				DownstreamType:      entity.TiCDCDownStreamTypeKafka,
				DownstreamParameter: "param2",
				NetworkType:         entity.ChangefeedDownstreamNetworkTypeInternet,
				NetworkInfo:         "network-info-2",
				OperationStatus:     entity.BizChangefeedOperationStatusCreating,
				ChangefeedStatus:    entity.ChangefeedKernelStatusNormal,
				CheckpointTSO:       439804651110400001,
				ErrorMessage:        "no error",
				ClusterID:           67891,
				TenantID:            11112,
			},
		},
		Total: 2,
	}

	got := resp.ToDataflowConnectors()

	assert.Equal(t, resp.Total, got.Total)
	assert.Equal(t, len(resp.BizChangefeeds), len(got.Items))

	// Check first item
	assert.Equal(t, resp.BizChangefeeds[0].BizChangefeedID, got.Items[0].ChangefeedId)
	assert.Equal(t, dataflow.ChangefeedType_MYSQL, got.Items[0].ChangefeedType)
	assert.Equal(t, resp.BizChangefeeds[0].Name, got.Items[0].Name)
	assert.Equal(t, dataflow.Changefeed_RUNNING, got.Items[0].Status)

	// Check second item
	assert.Equal(t, resp.BizChangefeeds[1].BizChangefeedID, got.Items[1].ChangefeedId)
	assert.Equal(t, dataflow.ChangefeedType_KAFKA, got.Items[1].ChangefeedType)
	assert.Equal(t, resp.BizChangefeeds[1].Name, got.Items[1].Name)
	assert.Equal(t, dataflow.Changefeed_CREATING, got.Items[1].Status)
}

func TestGetClusterBizChangefeedRCUResp_ToDataflowResp(t *testing.T) {
	resp := &GetClusterBizChangefeedRCUResp{
		RCUNumber: 100,
	}

	got := resp.ToDataflowResp()

	assert.Equal(t, uint64(resp.RCUNumber), got.TcuNumber)
}

func TestCalBizChangefeedFeeResp_ToDataflowResp(t *testing.T) {
	resp := &CalBizChangefeedFeeResp{
		Fee:    100.50,
		RawFee: 90.25,
	}

	got := resp.ToDataflowResp()

	assert.Equal(t, resp.Fee, got.Fee)
	assert.Equal(t, resp.RawFee, got.RawFee)
}

func TestListBizChangefeedSpecificationsResp_ToDataflowResp(t *testing.T) {
	resp := &ListBizChangefeedSpecificationsResp{
		Specifications: []*BizChangefeedSpecification{
			{
				SpecificationCode: "spec-1",
				RCUNumber:         10,
				Replicas:          2,
				RPSLimit:          1000,
			},
			{
				SpecificationCode: "spec-2",
				RCUNumber:         20,
				Replicas:          4,
				RPSLimit:          2000,
			},
		},
		Total: 2,
	}

	got := resp.ToDataflowResp()

	assert.Equal(t, resp.Total, got.Total)
	assert.Equal(t, len(resp.Specifications), len(got.Items))

	// Check first item
	assert.Equal(t, resp.Specifications[0].SpecificationCode, got.Items[0].Name)
	assert.Equal(t, resp.Specifications[0].RCUNumber, got.Items[0].RcuNum)
	assert.Equal(t, resp.Specifications[0].RPSLimit, got.Items[0].RpsLimit)

	// Check second item
	assert.Equal(t, resp.Specifications[1].SpecificationCode, got.Items[1].Name)
	assert.Equal(t, resp.Specifications[1].RCUNumber, got.Items[1].RcuNum)
	assert.Equal(t, resp.Specifications[1].RPSLimit, got.Items[1].RpsLimit)
}

func TestBizChangefeedSpecification_ToDataflowInstanceType(t *testing.T) {
	spec := &BizChangefeedSpecification{
		SpecificationCode: "spec-code",
		RCUNumber:         100,
		Replicas:          2,
		RPSLimit:          1000,
	}

	got := spec.ToDataflowInstanceType()

	assert.Equal(t, spec.SpecificationCode, got.Name)
	assert.Equal(t, spec.RCUNumber, got.RcuNum)
	assert.Equal(t, spec.RPSLimit, got.RpsLimit)
}