package grpcli

import (
	"context"
	"errors"
	"net"
	"strconv"
	"testing"

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/pkgv2/bizerr"
	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	mgmtv1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/status"
	"google.golang.org/grpc/test/bufconn"
	"google.golang.org/protobuf/proto"
)

type fakeExportServiceClient struct {
	getFn    func(context.Context, *globalngv1.GetExportRequest) (*globalngv1.Export, error)
	createFn func(context.Context, *globalngv1.CreateExportRequest) (*globalngv1.Export, error)
	deleteFn func(context.Context, *globalngv1.DeleteExportRequest) (*globalngv1.Export, error)
	cancelFn func(context.Context, *globalngv1.CancelExportRequest) (*globalngv1.Export, error)
}

// testExportCaller is the acting identity every management Export call needs;
// the client rejects a zero one.
var testExportCaller = ExportCaller{OrganizationID: 1369847559692509704, UserID: 42}

func (f *fakeExportServiceClient) GetExport(
	ctx context.Context, req *globalngv1.GetExportRequest, _ ...grpc.CallOption,
) (*globalngv1.Export, error) {
	return f.getFn(ctx, req)
}

func (f *fakeExportServiceClient) CreateExport(
	ctx context.Context, req *globalngv1.CreateExportRequest, _ ...grpc.CallOption,
) (*globalngv1.Export, error) {
	return f.createFn(ctx, req)
}

func (f *fakeExportServiceClient) DeleteExport(
	ctx context.Context, req *globalngv1.DeleteExportRequest, _ ...grpc.CallOption,
) (*globalngv1.Export, error) {
	return f.deleteFn(ctx, req)
}

func (f *fakeExportServiceClient) CancelExport(
	ctx context.Context, req *globalngv1.CancelExportRequest, _ ...grpc.CallOption,
) (*globalngv1.Export, error) {
	return f.cancelFn(ctx, req)
}

func TestExportGrpcClientDelegatesWithoutTransformingRequests(t *testing.T) {
	t.Parallel()

	ctx := context.Background()
	wantExport := &globalngv1.Export{ExportId: "exp-test"}
	client := newExportGrpcClient(&fakeExportServiceClient{
		getFn: func(_ context.Context, req *globalngv1.GetExportRequest) (*globalngv1.Export, error) {
			require.Equal(t, "exp-test", req.GetExportId())
			return wantExport, nil
		},
		createFn: func(_ context.Context, req *globalngv1.CreateExportRequest) (*globalngv1.Export, error) {
			require.Equal(t, uint64(7)<<18, req.GetExport().GetSnapshotTso())
			require.Equal(t, "test-user", req.GetExport().GetUsername())
			return wantExport, nil
		},
		deleteFn: func(_ context.Context, req *globalngv1.DeleteExportRequest) (*globalngv1.Export, error) {
			require.Equal(t, "exp-test", req.GetExportId())
			return wantExport, nil
		},
		cancelFn: func(_ context.Context, req *globalngv1.CancelExportRequest) (*globalngv1.Export, error) {
			require.Equal(t, "exp-test", req.GetExportId())
			return wantExport, nil
		},
	})

	got, err := client.GetExport(ctx, testExportCaller, &globalngv1.GetExportRequest{ClusterId: "42", ExportId: "exp-test"})
	require.NoError(t, err)
	require.Same(t, wantExport, got)

	got, err = client.CreateExport(ctx, testExportCaller, &globalngv1.CreateExportRequest{
		ClusterId: "42",
		Export: &globalngv1.Export{
			Username: "test-user",
			Password: "test-password",
		},
	}, 7<<18)
	require.NoError(t, err)
	require.Same(t, wantExport, got)
	require.NoError(t, client.CancelExport(ctx, testExportCaller, &globalngv1.CancelExportRequest{ClusterId: "42", ExportId: "exp-test"}))
	require.NoError(t, client.DeleteExport(ctx, testExportCaller, &globalngv1.DeleteExportRequest{ClusterId: "42", ExportId: "exp-test"}))
}

func TestExportGrpcClientConvertsErrors(t *testing.T) {
	t.Parallel()

	client := newExportGrpcClient(&fakeExportServiceClient{
		getFn: func(context.Context, *globalngv1.GetExportRequest) (*globalngv1.Export, error) {
			return nil, status.Error(codes.NotFound, "missing")
		},
		createFn: func(context.Context, *globalngv1.CreateExportRequest) (*globalngv1.Export, error) {
			return nil, errors.New("transport closed")
		},
		deleteFn: func(context.Context, *globalngv1.DeleteExportRequest) (*globalngv1.Export, error) {
			return nil, status.Error(codes.PermissionDenied, "denied")
		},
		cancelFn: func(context.Context, *globalngv1.CancelExportRequest) (*globalngv1.Export, error) {
			return nil, status.Error(codes.InvalidArgument, "invalid")
		},
	})

	_, err := client.GetExport(context.Background(), testExportCaller, &globalngv1.GetExportRequest{})
	require.Equal(t, bizerr.ErrCodeRecordNotFound, err.RawErrCode())
	_, err = client.CreateExport(context.Background(), testExportCaller, &globalngv1.CreateExportRequest{}, 7<<18)
	require.NotNil(t, err)
	err = client.DeleteExport(context.Background(), testExportCaller, &globalngv1.DeleteExportRequest{})
	require.Equal(t, bizerr.ErrCodePermission, err.RawErrCode())
	err = client.CancelExport(context.Background(), testExportCaller, &globalngv1.CancelExportRequest{})
	require.Equal(t, bizerr.ErrCodeParams, err.RawErrCode())
}

func TestExportGrpcClientUsesManagementServiceOnMgmtConnection(t *testing.T) {
	t.Parallel()

	listener := bufconn.Listen(1024 * 1024)
	calledMethods := make([]string, 0, 5)
	server := grpc.NewServer(grpc.UnaryInterceptor(
		func(
			ctx context.Context,
			req any,
			info *grpc.UnaryServerInfo,
			handler grpc.UnaryHandler,
		) (any, error) {
			calledMethods = append(calledMethods, info.FullMethod)
			return handler(ctx, req)
		},
	))
	service := &testManagementExportService{}
	mgmtv1.RegisterExportServiceServer(server, service)
	serveErr := make(chan error, 1)
	go func() {
		serveErr <- server.Serve(listener)
	}()
	t.Cleanup(func() {
		server.Stop()
		require.NoError(t, <-serveErr)
		require.NoError(t, listener.Close())
	})

	conn, err := grpc.DialContext(
		context.Background(),
		"bufnet",
		grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) {
			return listener.Dial()
		}),
		grpc.WithTransportCredentials(insecure.NewCredentials()),
	)
	require.NoError(t, err)
	t.Cleanup(func() { require.NoError(t, conn.Close()) })

	client := newMgmtSvcGrpcClient("bufnet", conn).ExportClient()
	got, bizErr := client.GetExport(context.Background(), testExportCaller, &globalngv1.GetExportRequest{
		ClusterId: "cluster-test",
		ExportId:  "export-test",
	})
	require.NoError(t, bizErr)
	require.Equal(t, "export-test", got.GetExportId())
	// The management service reads the acting identity out of the incoming
	// metadata and refuses a write that arrives without it, so the identity
	// has to survive the wire, not just the client struct.
	require.Equal(t,
		[]string{strconv.FormatUint(testExportCaller.OrganizationID, 10)},
		service.seenMetadata.Get("x-org-id"))
	require.Equal(t,
		[]string{strconv.FormatUint(testExportCaller.UserID, 10)},
		service.seenMetadata.Get("x-user-id"))
	// Without this the management service drops the identity instead of
	// forwarding it to the service that performs the write.
	require.Equal(t, []string{"true"}, service.seenMetadata.Get("x-no-project-required"))

	request := &globalngv1.CreateExportRequest{
		ClusterId:    "cluster-test",
		ValidateOnly: true,
		Export: &globalngv1.Export{
			DisplayName: "pipeline-test",
			Username:    "test-user",
			Password:    "test-password",
		},
	}
	const snapshotTSO = uint64(1234) << 18
	got, bizErr = client.CreateExport(context.Background(), testExportCaller, request, snapshotTSO)
	require.NoError(t, bizErr)
	require.Equal(t, "export-test", got.GetExportId())
	// The snapshot boundary survives a real proto round trip on the wire.
	require.Equal(t, snapshotTSO, service.request.GetExport().GetSnapshotTso())
	require.Zero(t, request.GetExport().GetSnapshotTso(), "the caller's request is never mutated")
	want := proto.Clone(request).(*globalngv1.CreateExportRequest)
	want.Export.SnapshotTso = snapshotTSO
	require.True(t, proto.Equal(want, service.request))

	require.NoError(t, client.CancelExport(context.Background(), testExportCaller, &globalngv1.CancelExportRequest{
		ClusterId: "cluster-test",
		ExportId:  "export-test",
	}))
	require.NoError(t, client.DeleteExport(context.Background(), testExportCaller, &globalngv1.DeleteExportRequest{
		ClusterId: "cluster-test",
		ExportId:  "export-test",
	}))

	require.Equal(t, []string{
		mgmtv1.ExportService_GetExport_FullMethodName,
		mgmtv1.ExportService_CreateExport_FullMethodName,
		mgmtv1.ExportService_CancelExport_FullMethodName,
		mgmtv1.ExportService_DeleteExport_FullMethodName,
	}, calledMethods)
	require.Equal(t, "cluster-test", service.getRequest.GetClusterId())
	require.Equal(t, "export-test", service.getRequest.GetExportId())
	require.Equal(t, "cluster-test", service.cancelRequest.GetClusterId())
	require.Equal(t, "export-test", service.cancelRequest.GetExportId())
	require.Equal(t, "cluster-test", service.deleteRequest.GetClusterId())
	require.Equal(t, "export-test", service.deleteRequest.GetExportId())
}

func TestExportGrpcClientSendsRequestLevelSnapshotTSO(t *testing.T) {
	t.Parallel()

	var sent *globalngv1.CreateExportRequest
	client := newExportGrpcClient(&fakeExportServiceClient{
		createFn: func(_ context.Context, req *globalngv1.CreateExportRequest) (*globalngv1.Export, error) {
			sent = req
			return &globalngv1.Export{ExportId: "exp-test"}, nil
		},
	})

	const snapshotTSO = uint64(123) << 18
	caller := &globalngv1.CreateExportRequest{Export: &globalngv1.Export{Username: "test-user"}}
	_, err := client.CreateExport(context.Background(), testExportCaller, caller, snapshotTSO)
	require.NoError(t, err)
	require.Equal(t, snapshotTSO, sent.GetExport().GetSnapshotTso())
	// The caller's request is never mutated, so a workflow retry rebuilds the
	// same value instead of accumulating state across attempts.
	require.Zero(t, caller.GetExport().GetSnapshotTso())

	// A missing boundary is rejected, not silently omitted: exporting at a
	// timestamp the regional service picks would break the alignment with the
	// incremental changefeed.
	sent = nil
	_, err = client.CreateExport(
		context.Background(),
		testExportCaller,
		&globalngv1.CreateExportRequest{Export: &globalngv1.Export{}},
		0,
	)
	require.Equal(t, bizerr.ErrCodeParams, err.RawErrCode())
	require.Nil(t, sent)
}

func TestExportGrpcClientRejectsOutputOnlySnapshotTSO(t *testing.T) {
	t.Parallel()

	createCalled := false
	client := newExportGrpcClient(&fakeExportServiceClient{
		createFn: func(context.Context, *globalngv1.CreateExportRequest) (*globalngv1.Export, error) {
			createCalled = true
			return &globalngv1.Export{}, nil
		},
	})

	// A missing boundary is rejected, not silently omitted.
	_, err := client.CreateExport(
		context.Background(),
		testExportCaller,
		&globalngv1.CreateExportRequest{Export: &globalngv1.Export{}},
		0,
	)
	require.Equal(t, bizerr.ErrCodeParams, err.RawErrCode())
	require.False(t, createCalled)

	// So is a request with no Export to carry it.
	_, err = client.CreateExport(
		context.Background(), testExportCaller, &globalngv1.CreateExportRequest{}, 123<<18)
	require.Equal(t, bizerr.ErrCodeParams, err.RawErrCode())
	require.False(t, createCalled)

	_, err = client.CreateExport(context.Background(), testExportCaller, nil, 123<<18)
	require.Equal(t, bizerr.ErrCodeParams, err.RawErrCode())
	require.False(t, createCalled)
}

// testManagementExportService is registered through the generated
// tidb_mgmt_service.v1 service descriptor, so these tests exercise the real
// method names and message types instead of a hand-maintained copy.
type testManagementExportService struct {
	mgmtv1.UnimplementedExportServiceServer

	getRequest    *globalngv1.GetExportRequest
	request       *globalngv1.CreateExportRequest
	deleteRequest *globalngv1.DeleteExportRequest
	cancelRequest *globalngv1.CancelExportRequest

	// metadata seen by the most recent call, so tests can assert the acting
	// identity actually reaches the wire rather than only the client struct.
	seenMetadata metadata.MD
}

func (s *testManagementExportService) record(ctx context.Context) {
	s.seenMetadata, _ = metadata.FromIncomingContext(ctx)
}

func (s *testManagementExportService) GetExport(
	ctx context.Context,
	req *globalngv1.GetExportRequest,
) (*globalngv1.Export, error) {
	s.record(ctx)
	s.getRequest = req
	return &globalngv1.Export{ExportId: req.GetExportId()}, nil
}

func (s *testManagementExportService) CreateExport(
	ctx context.Context,
	req *globalngv1.CreateExportRequest,
) (*globalngv1.Export, error) {
	s.record(ctx)
	s.request = req
	return &globalngv1.Export{ExportId: "export-test"}, nil
}

func (s *testManagementExportService) DeleteExport(
	ctx context.Context,
	req *globalngv1.DeleteExportRequest,
) (*globalngv1.Export, error) {
	s.record(ctx)
	s.deleteRequest = req
	return &globalngv1.Export{ExportId: req.GetExportId()}, nil
}

func (s *testManagementExportService) CancelExport(
	ctx context.Context,
	req *globalngv1.CancelExportRequest,
) (*globalngv1.Export, error) {
	s.record(ctx)
	s.cancelRequest = req
	return &globalngv1.Export{ExportId: req.GetExportId()}, nil
}