package compatibility_test

import (
	"context"
	"io"
	"testing"
	"time"

	semantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/bulkingestv1"
	commonsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commonv1"
	runtimev1 "github.com/pingcap-inc/nexuscrm-contract/wire-go/gen/runtime/v1"
	"github.com/pingcap-inc/nexuscrm-contract/wire-go/mapping"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/proto"
	"google.golang.org/protobuf/reflect/protodesc"
	"google.golang.org/protobuf/reflect/protoreflect"
	"google.golang.org/protobuf/reflect/protoregistry"
	"google.golang.org/protobuf/types/dynamicpb"
)

func TestGeneratedBulkIngestClientUsesCompleteLifecycle(t *testing.T) {
	service := &bulkIngestService{}
	connection, cleanup := startBufconnServer(t, func(server *grpc.Server) {
		runtimev1.RegisterBulkIngestServiceServer(server, service)
	})
	defer cleanup()
	client := runtimev1.NewBulkIngestServiceClient(connection)
	ctx, cancel := context.WithTimeout(withCapability(context.Background(), semantic.CapabilityName), time.Second)
	defer cancel()

	created, err := client.CreateBulkIngestJob(ctx, &runtimev1.CreateBulkIngestJobRequest{ObjectId: "Account"})
	if err != nil || created.GetJob().GetId() != "job-1" {
		t.Fatalf("create = %#v, error = %v", created, err)
	}
	read, err := client.ReadBulkIngestJob(ctx, &runtimev1.ReadBulkIngestJobRequest{JobId: "job-1"})
	if err != nil || read.GetJob().GetId() != "job-1" {
		t.Fatalf("read = %#v, error = %v", read, err)
	}
	closed, err := client.CloseBulkIngestJob(ctx, &runtimev1.CloseBulkIngestJobRequest{JobId: "job-1"})
	if err != nil || closed.GetJob().GetId() != "job-1" {
		t.Fatalf("close = %#v, error = %v", closed, err)
	}
	upload, err := client.UploadBulkIngestData(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if err := upload.Send(&runtimev1.UploadBulkIngestDataRequest{Value: &runtimev1.UploadBulkIngestDataRequest_Header{
		Header: &runtimev1.UploadBulkIngestDataHeader{JobId: "job-1"},
	}}); err != nil {
		t.Fatal(err)
	}
	if err := upload.Send(&runtimev1.UploadBulkIngestDataRequest{Value: &runtimev1.UploadBulkIngestDataRequest_Data{Data: []byte("a,b\n")}}); err != nil {
		t.Fatal(err)
	}
	uploaded, err := upload.CloseAndRecv()
	if err != nil || uploaded.GetJobId() != "job-1" || service.uploaded != 4 {
		t.Fatalf("upload = %#v bytes=%d, error = %v", uploaded, service.uploaded, err)
	}
	if service.createCalls != 1 || service.readCalls != 1 || service.closeCalls != 1 || service.uploadCalls != 1 {
		t.Fatalf("calls create/read/close/upload = %d/%d/%d/%d", service.createCalls, service.readCalls, service.closeCalls, service.uploadCalls)
	}
}

func TestBulkIngestNClientFailsClosedAgainstNMinusOneEndpoint(t *testing.T) {
	connection, cleanup := startBufconnServer(t, func(*grpc.Server) {})
	defer cleanup()
	client := runtimev1.NewBulkIngestServiceClient(connection)
	ctx, cancel := context.WithTimeout(withCapability(context.Background(), semantic.CapabilityName), time.Second)
	defer cancel()

	for name, call := range map[string]func() error{
		"create": func() error {
			_, err := client.CreateBulkIngestJob(ctx, &runtimev1.CreateBulkIngestJobRequest{})
			return err
		},
		"read": func() error {
			_, err := client.ReadBulkIngestJob(ctx, &runtimev1.ReadBulkIngestJobRequest{})
			return err
		},
		"close": func() error {
			_, err := client.CloseBulkIngestJob(ctx, &runtimev1.CloseBulkIngestJobRequest{})
			return err
		},
	} {
		t.Run(name, func(t *testing.T) {
			if err := call(); status.Code(err) != codes.Unimplemented {
				t.Fatalf("N/N-1 error = %v", err)
			}
		})
	}
	upload, err := client.UploadBulkIngestData(ctx)
	if err != nil {
		t.Fatal(err)
	}
	response, err := upload.CloseAndRecv()
	if response != nil || status.Code(err) != codes.Unimplemented {
		t.Fatalf("N/N-1 upload = %#v, error = %v", response, err)
	}
}

func TestBulkIngestClientVersionIsAdditiveAcrossNAndNMinusOne(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	oldFile := nMinusOneBulkIngestFile(t)

	request := semantic.CreateRequest{ObjectID: "Account", Operation: semantic.OperationInsert, ClientVersion: 59}
	wireRequest, err := mapping.ToWireCreateBulkIngestJobRequest(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	encodedRequest, err := proto.Marshal(wireRequest)
	if err != nil {
		t.Fatal(err)
	}
	oldRequest := dynamicpb.NewMessage(oldFile.Messages().ByName("CreateBulkIngestJobRequest"))
	if err := (proto.UnmarshalOptions{DiscardUnknown: true}).Unmarshal(encodedRequest, oldRequest); err != nil {
		t.Fatalf("N-1 decode of N create request: %v", err)
	}
	oldRequestBytes, err := proto.Marshal(oldRequest)
	if err != nil {
		t.Fatal(err)
	}
	roundWireRequest := &runtimev1.CreateBulkIngestJobRequest{}
	if err := proto.Unmarshal(oldRequestBytes, roundWireRequest); err != nil {
		t.Fatalf("N decode of N-1 create request: %v", err)
	}
	roundRequest, err := mapping.FromWireCreateBulkIngestJobRequest(ctx, roundWireRequest)
	if err != nil {
		t.Fatalf("N mapping rejected N-1 create request: %v", err)
	}
	if roundRequest.ObjectID != request.ObjectID || roundRequest.Operation != request.Operation || roundRequest.ClientVersion != 0 {
		t.Fatalf("N-1 create request mapping = %#v", roundRequest)
	}

	job := semantic.Job{
		ID: "job-1", ObjectID: "Account", Operation: semantic.OperationInsert, State: semantic.StateOpen,
		CreatedByID: "user-1", CreatedAt: "2026-08-27T10:00:00Z", ModifiedAt: "2026-08-27T10:00:00Z", ClientVersion: 59,
	}
	readRequest := semantic.ReadRequest{JobID: job.ID}
	receipts := commonsemantic.Receipts{ProtocolVersion: "1.0", CapabilityVersion: semantic.CapabilityName}
	wireResponse, err := mapping.ToWireReadBulkIngestJobResponse(readRequest, semantic.ReadResponse{Job: job, Receipts: receipts})
	if err != nil {
		t.Fatal(err)
	}
	encodedResponse, err := proto.Marshal(wireResponse)
	if err != nil {
		t.Fatal(err)
	}
	oldResponse := dynamicpb.NewMessage(oldFile.Messages().ByName("ReadBulkIngestJobResponse"))
	if err := (proto.UnmarshalOptions{DiscardUnknown: true}).Unmarshal(encodedResponse, oldResponse); err != nil {
		t.Fatalf("N-1 decode of N read response: %v", err)
	}
	jobField := oldResponse.Descriptor().Fields().ByName("job")
	idField := oldFile.Messages().ByName("BulkIngestJob").Fields().ByName("id")
	if jobField == nil || idField == nil || oldResponse.Get(jobField).Message().Get(idField).String() != job.ID {
		t.Fatal("N-1 client lost the existing job identity")
	}
	oldResponseBytes, err := proto.Marshal(oldResponse)
	if err != nil {
		t.Fatal(err)
	}
	roundWireResponse := &runtimev1.ReadBulkIngestJobResponse{}
	if err := proto.Unmarshal(oldResponseBytes, roundWireResponse); err != nil {
		t.Fatalf("N decode of N-1 read response: %v", err)
	}
	roundResponse, err := mapping.FromWireReadBulkIngestJobResponse(readRequest, roundWireResponse)
	if err != nil {
		t.Fatalf("N mapping rejected N-1 read response: %v", err)
	}
	if roundResponse.Job.ID != job.ID || roundResponse.Job.ClientVersion != 0 {
		t.Fatalf("N-1 read response mapping = %#v", roundResponse.Job)
	}
}

func nMinusOneBulkIngestFile(t *testing.T) protoreflect.FileDescriptor {
	t.Helper()
	file := protodesc.ToFileDescriptorProto(runtimev1.File_runtime_v1_bulk_ingest_proto)
	for _, message := range file.GetMessageType() {
		removeNumber := int32(0)
		switch message.GetName() {
		case "BulkIngestJob":
			removeNumber = 14
		case "CreateBulkIngestJobRequest":
			removeNumber = 4
		}
		if removeNumber == 0 {
			continue
		}
		fields := message.Field[:0]
		for _, field := range message.GetField() {
			if field.GetNumber() != removeNumber {
				fields = append(fields, field)
			}
		}
		message.Field = fields
	}
	oldFile, err := protodesc.NewFile(file, protoregistry.GlobalFiles)
	if err != nil {
		t.Fatal(err)
	}
	return oldFile
}

type bulkIngestService struct {
	runtimev1.UnimplementedBulkIngestServiceServer
	createCalls int
	readCalls   int
	closeCalls  int
	uploadCalls int
	uploaded    int
}

func (s *bulkIngestService) CreateBulkIngestJob(context.Context, *runtimev1.CreateBulkIngestJobRequest) (*runtimev1.CreateBulkIngestJobResponse, error) {
	s.createCalls++
	return &runtimev1.CreateBulkIngestJobResponse{Job: &runtimev1.BulkIngestJob{Id: "job-1"}}, nil
}

func (s *bulkIngestService) ReadBulkIngestJob(context.Context, *runtimev1.ReadBulkIngestJobRequest) (*runtimev1.ReadBulkIngestJobResponse, error) {
	s.readCalls++
	return &runtimev1.ReadBulkIngestJobResponse{Job: &runtimev1.BulkIngestJob{Id: "job-1"}}, nil
}

func (s *bulkIngestService) CloseBulkIngestJob(context.Context, *runtimev1.CloseBulkIngestJobRequest) (*runtimev1.CloseBulkIngestJobResponse, error) {
	s.closeCalls++
	return &runtimev1.CloseBulkIngestJobResponse{Job: &runtimev1.BulkIngestJob{Id: "job-1"}}, nil
}

func (s *bulkIngestService) UploadBulkIngestData(stream grpc.ClientStreamingServer[runtimev1.UploadBulkIngestDataRequest, runtimev1.UploadBulkIngestDataResponse]) error {
	s.uploadCalls++
	jobID := ""
	for {
		request, err := stream.Recv()
		if err == io.EOF {
			return stream.SendAndClose(&runtimev1.UploadBulkIngestDataResponse{JobId: jobID})
		}
		if err != nil {
			return err
		}
		if header := request.GetHeader(); header != nil {
			jobID = header.GetJobId()
		}
		s.uploaded += len(request.GetData())
	}
}