package premium_dm

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"testing"

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/dataflow-service-ng/pkg/agent"
	"github.com/tidbcloud/dataflow-service-ng/pkg/grpc/grpcli"
	regionalapi "github.com/tidbcloud/dataflow-service-ng/pkg/regional-api"
	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
)

type mockDMOpenAPIClient struct {
	lastMethod string
	lastPath   string
	lastBody   []byte
	status     int
	respBody   []byte
	err        error
}

func (m *mockDMOpenAPIClient) Do(
	_ context.Context, method, apiPath string, body []byte,
) (int, []byte, error) {
	m.lastMethod = method
	m.lastPath = apiPath
	m.lastBody = body
	if m.err != nil {
		return 0, nil, m.err
	}
	return m.status, m.respBody, nil
}

func TestDMOpenAPIRouteFieldsServiceProxy(t *testing.T) {
	proxy := &PremiumDMProxy{
		useServiceProxy: true,
		DMNameSpace:     "dm42",
		dmClusterID:     "7002",
		gatewayBaseURL:  "https://gateway.example:8443",
	}

	fields := proxy.dmOpenAPIRouteFields(http.MethodPost, PremiumDMSourcePath)
	fieldMap := fieldsToMap(t, fields)

	require.Equal(t, http.MethodPost, fieldMap["method"])
	require.Equal(t, PremiumDMSourcePath, fieldMap["path"])
	require.Equal(t, "/proxy/cluster/42/dm/7002/api/v1/sources", fieldMap["proxy_path"])
	require.Equal(t, "https://gateway.example:8443/proxy/cluster/42/dm/7002/api/v1/sources", fieldMap["uri"])
	require.Equal(t, "gateway.example:8443", fieldMap["gateway_host"])
	require.Equal(t, "7002", fieldMap["dm_cluster_id"])
	require.Equal(t, "42", fieldMap["shared_pool_id"])
}

func TestDMOpenAPIRouteFieldsLegacy(t *testing.T) {
	proxy := &PremiumDMProxy{
		useServiceProxy: false,
		DMNameSpace:     "dm42",
	}

	fields := proxy.dmOpenAPIRouteFields(http.MethodGet, "/api/v1/sources/src-1")
	fieldMap := fieldsToMap(t, fields)

	require.Equal(t, http.MethodGet, fieldMap["method"])
	require.Equal(t, "/api/v1/sources/src-1", fieldMap["path"])
	require.Equal(t, "dm42", fieldMap["dm_namespace"])
	require.NotContains(t, fieldMap, "proxy_path")
}

func TestRedactedJSONBodyFieldMasksPassword(t *testing.T) {
	body := []byte(`{"source":{"user":"dmtestadmin","password":"secret-pwd","host":"mysql.example"}}`)
	field := redactedJSONBodyField(body)
	fieldMap := fieldsToMap(t, []zap.Field{field})

	redacted := fieldMap["body"]
	require.Contains(t, redacted, `"password":"redacted"`)
	require.Contains(t, redacted, `"user":"dmtestadmin"`)
	require.NotContains(t, redacted, "secret-pwd")
}

func fieldsToMap(t *testing.T, fields []zap.Field) map[string]string {
	t.Helper()

	enc := zapcore.NewMapObjectEncoder()
	for _, field := range fields {
		field.AddTo(enc)
	}
	result := make(map[string]string, len(enc.Fields))
	for key, value := range enc.Fields {
		result[key] = fmt.Sprint(value)
	}
	return result
}

func TestPremiumDMProxyRoutesOperatorV2ToServiceProxy(t *testing.T) {
	mockClient := &mockDMOpenAPIClient{status: http.StatusOK, respBody: []byte(`{"source_name":"src-1"}`)}
	proxy := &PremiumDMProxy{
		useServiceProxy: true,
		dmOpenAPI:       mockClient,
	}

	_, err := proxy.GetPremiumDMSource(context.Background(), "src-1")
	require.NoError(t, err)
	require.Equal(t, http.MethodGet, mockClient.lastMethod)
	require.Equal(t, "/api/v1/sources/src-1?with_status=true", mockClient.lastPath)
}

func TestNewPremiumDMProxyRejectsUnresolvedDMCluster(t *testing.T) {
	restore := stubPremiumDMProxyDeps(t)
	defer restore()

	newNGRegionalClientByClusterIDFunc = func(
		ctx context.Context, clusterID uint64, mgmtClient grpcli.MgmtSvcGrpcClient, cfg *agent.ClientConfig,
	) (*agent.NGRegionalClient, error) {
		require.Equal(t, uint64(123), clusterID)
		return &agent.NGRegionalClient{}, nil
	}
	getSharedPoolByClusterIDFunc = func(
		ctx context.Context, agentClient *agent.NGRegionalClient, clusterID uint64,
	) (*agent.SharedPool, error) {
		return &agent.SharedPool{Id: "42"}, nil
	}
	resolveDMClusterForTiDBIDFunc = func(
		ctx context.Context, tiDBID string, mgmtClient grpcli.MgmtSvcGrpcClient, agentCfg agent.ClientConfig,
	) (*agent.DMCluster, error) {
		require.Equal(t, "123", tiDBID)
		return nil, agent.DMClusterNotFoundErr
	}

	proxy, err := NewPremiumDMProxy(context.Background(), "123", nil, agent.ClientConfig{}, nil)
	require.Nil(t, proxy)
	require.ErrorContains(t, err, "resolve dm cluster for routing")
	require.ErrorContains(t, err, agent.DMClusterNotFoundErr.Error())
}

func TestNewPremiumDMProxyUsesLegacyPathWhenResolvedBackendIsLegacy(t *testing.T) {
	restore := stubPremiumDMProxyDeps(t)
	defer restore()

	newNGRegionalClientByClusterIDFunc = func(
		ctx context.Context, clusterID uint64, mgmtClient grpcli.MgmtSvcGrpcClient, cfg *agent.ClientConfig,
	) (*agent.NGRegionalClient, error) {
		return &agent.NGRegionalClient{}, nil
	}
	getSharedPoolByClusterIDFunc = func(
		ctx context.Context, agentClient *agent.NGRegionalClient, clusterID uint64,
	) (*agent.SharedPool, error) {
		return &agent.SharedPool{Id: "42"}, nil
	}
	resolveDMClusterForTiDBIDFunc = func(
		ctx context.Context, tiDBID string, mgmtClient grpcli.MgmtSvcGrpcClient, agentCfg agent.ClientConfig,
	) (*agent.DMCluster, error) {
		return &agent.DMCluster{Id: "7001", Backend: regionalapi.BackendLegacy}, nil
	}

	proxy, err := NewPremiumDMProxy(context.Background(), "123", nil, agent.ClientConfig{}, nil)
	require.NoError(t, err)
	require.NotNil(t, proxy)
	require.False(t, proxy.useServiceProxy)
	require.Equal(t, "dm42", proxy.DMNameSpace)
	require.Empty(t, proxy.dmClusterID)
}

func TestNewPremiumDMProxyUsesServiceProxyWhenResolvedBackendIsOperatorV2(t *testing.T) {
	restore := stubPremiumDMProxyDeps(t)
	defer restore()

	newNGRegionalClientByClusterIDFunc = func(
		ctx context.Context, clusterID uint64, mgmtClient grpcli.MgmtSvcGrpcClient, cfg *agent.ClientConfig,
	) (*agent.NGRegionalClient, error) {
		return &agent.NGRegionalClient{}, nil
	}
	getSharedPoolByClusterIDFunc = func(
		ctx context.Context, agentClient *agent.NGRegionalClient, clusterID uint64,
	) (*agent.SharedPool, error) {
		return &agent.SharedPool{
			Id:             "42",
			GatewayBaseUrl: "https://gateway.example",
			Provider:       &agent.Provider{AuthToken: "token"},
		}, nil
	}
	resolveDMClusterForTiDBIDFunc = func(
		ctx context.Context, tiDBID string, mgmtClient grpcli.MgmtSvcGrpcClient, agentCfg agent.ClientConfig,
	) (*agent.DMCluster, error) {
		return &agent.DMCluster{Id: "7002", Backend: regionalapi.BackendOperatorV2}, nil
	}

	var capturedCfg regionalapi.DMConfig
	newDMOpenAPIClientFunc = func(cfg regionalapi.DMConfig) (regionalapi.DMOpenAPIClient, error) {
		capturedCfg = cfg
		return &mockDMOpenAPIClient{}, nil
	}

	proxy, err := NewPremiumDMProxy(context.Background(), "123", nil, agent.ClientConfig{}, nil)
	require.NoError(t, err)
	require.NotNil(t, proxy)
	require.True(t, proxy.useServiceProxy)
	require.Equal(t, "7002", proxy.dmClusterID)
	require.Equal(t, regionalapi.DMConfig{
		GatewayBaseURL: "https://gateway.example",
		AuthToken:      "token",
		SharedPoolID:   "42",
		DMClusterID:    "7002",
	}, capturedCfg)
}

func TestIsOperatorV2BackendAlignment(t *testing.T) {
	require.True(t, regionalapi.IsOperatorV2Backend("operator-v2"))
	require.False(t, regionalapi.IsOperatorV2Backend("legacy"))
}

func TestPatchDMTargetConn(t *testing.T) {
	proxy := &PremiumDMProxy{
		agentCli: &agent.NGRegionalClient{},
	}
	original := []byte(`{"task":{"name":"task-1","target_config":{"port":4000}}}`)

	proxy.useServiceProxy = true
	proxy.agentCli = nil

	originalResolver := getTiDBHostFromAgentFunc
	getTiDBHostFromAgentFunc = func(ctx context.Context, agentCli *agent.NGRegionalClient, clusterID string) (string, error) {
		require.Equal(t, "123", clusterID)
		return "tg-1-tidb.tidb42.svc", nil
	}
	defer func() {
		getTiDBHostFromAgentFunc = originalResolver
	}()

	patched, err := proxy.patchDMTargetConn(context.Background(), original, "123")
	require.NoError(t, err)

	var payload map[string]any
	require.NoError(t, json.Unmarshal(patched, &payload))
	task := payload["task"].(map[string]any)
	targetConfig := task["target_config"].(map[string]any)
	require.Equal(t, "tg-1-tidb.tidb42.svc", targetConfig["host"])
	require.Equal(t, float64(4000), targetConfig["port"])
}

func stubPremiumDMProxyDeps(t *testing.T) func() {
	t.Helper()

	originalNewClient := newNGRegionalClientByClusterIDFunc
	originalGetSharedPool := getSharedPoolByClusterIDFunc
	originalResolveDMCluster := resolveDMClusterForTiDBIDFunc
	originalNewDMOpenAPIClient := newDMOpenAPIClientFunc

	return func() {
		newNGRegionalClientByClusterIDFunc = originalNewClient
		getSharedPoolByClusterIDFunc = originalGetSharedPool
		resolveDMClusterForTiDBIDFunc = originalResolveDMCluster
		newDMOpenAPIClientFunc = originalNewDMOpenAPIClient
	}
}