package runtimeapi

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

	sharedcontract "github.com/nexuscrm/shared/pkg/contract"
	"github.com/nexuscrm/shared/pkg/models"
	bulkingestsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/bulkingestv1"
	commonsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commonv1"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/delegationv1"
	recordsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordsv1"
	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/codes"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/status"
)

type bulkIngestServerTestEngine struct {
	t              *testing.T
	job            *models.BulkJob
	err            error
	userID         string
	uploadBody     []byte
	contentLength  int64
	createRequest  sharedcontract.BulkIngestCreateRequest
	wrapReadError  bool
	failBeforeRead bool
}

func (e *bulkIngestServerTestEngine) CreateIngestJob(_ context.Context, request sharedcontract.BulkIngestCreateRequest, user *models.UserSession) (*models.BulkJob, error) {
	e.userID = user.ID
	e.createRequest = request
	if e.err != nil {
		return nil, e.err
	}
	job := *e.job
	job.Object = request.Object
	job.Operation = request.Operation
	job.ExternalIdFieldName = request.ExternalIDFieldName
	job.ApiVersion = request.APIVersion
	return &job, nil
}

func (e *bulkIngestServerTestEngine) GetIngestJob(_ context.Context, _ string, user *models.UserSession) (*models.BulkJob, error) {
	e.userID = user.ID
	return e.job, e.err
}

func (e *bulkIngestServerTestEngine) CloseIngestJob(_ context.Context, _ string, state models.BulkJobState, user *models.UserSession) (*models.BulkJob, error) {
	e.userID = user.ID
	if e.err != nil {
		return nil, e.err
	}
	job := *e.job
	job.State = state
	return &job, nil
}

func (e *bulkIngestServerTestEngine) UploadIngestData(_ context.Context, request sharedcontract.BulkIngestUploadRequest, user *models.UserSession) error {
	e.userID = user.ID
	e.contentLength = request.ContentLength
	if e.failBeforeRead {
		return e.err
	}
	body, err := io.ReadAll(request.Body)
	if err != nil {
		if e.wrapReadError {
			return &sharedcontract.BulkIngestError{
				Kind: sharedcontract.BulkIngestErrorInternal, Message: "object storage read failed", Cause: err,
			}
		}
		return err
	}
	e.uploadBody = body
	return e.err
}

func TestBulkIngestServerLifecycleUsesOnlyTheTrustedPrincipal(t *testing.T) {
	engine := newBulkIngestServerTestEngine(t)
	server, err := NewBulkIngestServer(engine)
	if err != nil {
		t.Fatal(err)
	}
	ctx := bulkIngestServerTestContext(t)
	request, err := mapping.ToWireCreateBulkIngestJobRequest(ctx, bulkingestsemantic.CreateRequest{
		ObjectID: "Account", Operation: bulkingestsemantic.OperationUpsert, ExternalIDFieldName: "External_Id__c", ClientVersion: 59,
	})
	if err != nil {
		t.Fatal(err)
	}
	response, err := server.CreateBulkIngestJob(ctx, request)
	if err != nil {
		t.Fatalf("CreateBulkIngestJob() = %v", err)
	}
	semanticResponse, err := mapping.FromWireCreateBulkIngestJobResponse(bulkingestsemantic.CreateRequest{
		ObjectID: "Account", Operation: bulkingestsemantic.OperationUpsert, ExternalIDFieldName: "External_Id__c", ClientVersion: 59,
	}, response)
	if err != nil {
		t.Fatal(err)
	}
	if engine.userID != "user-1" || engine.createRequest.APIVersion != 59 || semanticResponse.Job.ObjectID != "Account" || semanticResponse.Job.ExternalIDFieldName != "External_Id__c" || semanticResponse.Job.ClientVersion != 59 {
		t.Fatalf("trusted lifecycle = user %q, job %#v", engine.userID, semanticResponse.Job)
	}
}

func TestBulkIngestServerStreamsKnownAndChunkedHTTPBodiesWithoutBuffering(t *testing.T) {
	for _, test := range []struct {
		name          string
		lengthKnown   bool
		contentLength int64
	}{
		{name: "known length", lengthKnown: true, contentLength: 7},
		{name: "chunked HTTP length", contentLength: -1},
	} {
		t.Run(test.name, func(t *testing.T) {
			engine := newBulkIngestServerTestEngine(t)
			server, err := NewBulkIngestServer(engine)
			if err != nil {
				t.Fatal(err)
			}
			header := bulkingestsemantic.UploadHeader{JobID: "job-1", ContentLengthKnown: test.lengthKnown}
			if test.lengthKnown {
				header.ContentLength = 7
			}
			wireHeader, err := mapping.ToWireUploadBulkIngestChunk(bulkingestsemantic.UploadChunk{Header: &header})
			if err != nil {
				t.Fatal(err)
			}
			first, _ := mapping.ToWireUploadBulkIngestChunk(bulkingestsemantic.UploadChunk{Data: []byte("id,name")})
			stream := &bulkIngestUploadTestStream{ctx: bulkIngestServerTestContext(t), requests: []*runtimev1.UploadBulkIngestDataRequest{wireHeader, first}}
			if err := server.UploadBulkIngestData(stream); err != nil {
				t.Fatalf("UploadBulkIngestData() = %v", err)
			}
			if string(engine.uploadBody) != "id,name" || engine.contentLength != test.contentLength || stream.response == nil || stream.response.GetJobId() != "job-1" {
				t.Fatalf("upload = body %q, length %d, response %#v", engine.uploadBody, engine.contentLength, stream.response)
			}
		})
	}
}

func TestBulkIngestServerRejectsIncompleteKnownLengthBeforeSuccess(t *testing.T) {
	engine := newBulkIngestServerTestEngine(t)
	// The production object-storage path wraps the pipe error as an internal
	// BulkIngestError. The protocol owner must still return the original
	// known-length violation rather than the cleanup echo from the engine.
	engine.wrapReadError = true
	server, err := NewBulkIngestServer(engine)
	if err != nil {
		t.Fatal(err)
	}
	header := bulkingestsemantic.UploadHeader{JobID: "job-1", ContentLength: 8, ContentLengthKnown: true}
	wireHeader, _ := mapping.ToWireUploadBulkIngestChunk(bulkingestsemantic.UploadChunk{Header: &header})
	data, _ := mapping.ToWireUploadBulkIngestChunk(bulkingestsemantic.UploadChunk{Data: []byte("short")})
	stream := &bulkIngestUploadTestStream{ctx: bulkIngestServerTestContext(t), requests: []*runtimev1.UploadBulkIngestDataRequest{wireHeader, data}}
	if err := server.UploadBulkIngestData(stream); status.Code(err) != codes.InvalidArgument {
		t.Fatalf("UploadBulkIngestData() = %v, want InvalidArgument", err)
	}
	if stream.response != nil {
		t.Fatal("incomplete upload reported success")
	}
}

func TestBulkIngestServerPreservesOversizedFirstHeaderClassification(t *testing.T) {
	engine := newBulkIngestServerTestEngine(t)
	server, err := NewBulkIngestServer(engine)
	if err != nil {
		t.Fatal(err)
	}
	tooLarge := bulkingestsemantic.MaxUploadBytes + 1
	stream := &bulkIngestUploadTestStream{
		ctx: bulkIngestServerTestContext(t),
		requests: []*runtimev1.UploadBulkIngestDataRequest{{Value: &runtimev1.UploadBulkIngestDataRequest_Header{
			Header: &runtimev1.UploadBulkIngestDataHeader{JobId: "job-1", ContentLength: &tooLarge},
		}}},
	}
	if err := server.UploadBulkIngestData(stream); status.Code(err) != codes.ResourceExhausted {
		t.Fatalf("UploadBulkIngestData() = %v, want ResourceExhausted", err)
	}
	if engine.userID != "" || stream.response != nil {
		t.Fatal("oversized first header reached the engine or reported success")
	}
}

func TestBulkIngestServerReturnsEarlyEngineFailureWithoutWaitingForMoreClientFrames(t *testing.T) {
	engine := newBulkIngestServerTestEngine(t)
	engine.failBeforeRead = true
	engine.err = &sharedcontract.BulkIngestError{
		Kind: sharedcontract.BulkIngestErrorNotFound, Message: "Job not found",
	}
	server, err := NewBulkIngestServer(engine)
	if err != nil {
		t.Fatal(err)
	}
	header := bulkingestsemantic.UploadHeader{JobID: "missing-job"}
	wireHeader, err := mapping.ToWireUploadBulkIngestChunk(bulkingestsemantic.UploadChunk{Header: &header})
	if err != nil {
		t.Fatal(err)
	}
	wait := make(chan struct{})
	stream := &bulkIngestUploadTestStream{
		ctx: bulkIngestServerTestContext(t), requests: []*runtimev1.UploadBulkIngestDataRequest{wireHeader},
		waitAfterRequests: wait,
	}
	result := make(chan error, 1)
	go func() { result <- server.UploadBulkIngestData(stream) }()

	select {
	case callErr := <-result:
		close(wait)
		if status.Code(callErr) != codes.NotFound {
			t.Fatalf("UploadBulkIngestData() = %v, want NotFound", callErr)
		}
	case <-time.After(100 * time.Millisecond):
		close(wait)
		t.Fatal("UploadBulkIngestData waited for another client frame after the engine had failed")
	}
}

func TestBulkIngestEngineCancellationClassificationUsesRequestAuthority(t *testing.T) {
	for _, test := range []struct {
		name string
		ctx  func() context.Context
		err  error
		want codes.Code
	}{
		{
			name: "owner cancellation wrapped by Bulk error while request remains live",
			ctx:  context.Background,
			err:  &sharedcontract.BulkIngestError{Kind: sharedcontract.BulkIngestErrorInternal, Message: "storage canceled", Cause: context.Canceled},
			want: codes.Unavailable,
		},
		{
			name: "owner deadline wrapped by Bulk error while request remains live",
			ctx:  context.Background,
			err:  &sharedcontract.BulkIngestError{Kind: sharedcontract.BulkIngestErrorInternal, Message: "storage deadline", Cause: context.DeadlineExceeded},
			want: codes.Unavailable,
		},
		{
			name: "caller cancellation owns an opaque engine error",
			ctx: func() context.Context {
				ctx, cancel := context.WithCancel(context.Background())
				cancel()
				return ctx
			},
			err:  &sharedcontract.BulkIngestError{Kind: sharedcontract.BulkIngestErrorInternal, Message: "engine stopped"},
			want: codes.Canceled,
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			classified := classifyBulkIngestEngineError(test.ctx(), test.err, commonsemantic.InvalidComponentCommandData)
			wireErr := mapping.ToPlatformStatusError(classified, bulkingestsemantic.CapabilityName)
			if got := status.Code(wireErr); got != test.want {
				t.Fatalf("gRPC code = %v, want %v; error = %v", got, test.want, wireErr)
			}
		})
	}
}

func newBulkIngestServerTestEngine(t *testing.T) *bulkIngestServerTestEngine {
	t.Helper()
	now := time.Date(2026, 8, 27, 10, 0, 0, 0, time.UTC)
	return &bulkIngestServerTestEngine{t: t, job: &models.BulkJob{
		ID: "job-1", Object: "Account", Operation: models.OperationInsert, State: models.JobStateOpen,
		CreatedById: "user-1", CreatedDate: now, SystemModstamp: now,
	}}
}

func bulkIngestServerTestContext(t *testing.T) context.Context {
	t.Helper()
	base, cancel := context.WithTimeout(context.Background(), time.Second)
	t.Cleanup(cancel)
	ctx := context.WithValue(base, runtimeExecutionScopeContextKey{}, RuntimeExecutionScope{
		view: queryResponseTestView(t), orgID: testOrgID, appInstallationID: testInstallationID,
		workloadID: testWorkloadID, capability: bulkingestsemantic.CapabilityName,
		receipts: recordsemantic.Receipts{ProtocolVersion: "1.0", CapabilityVersion: bulkingestsemantic.CapabilityName, CatalogRevision: 17, SchemaGeneration: 8, EngineGeneration: 9, SecurityEpoch: 3},
	})
	return context.WithValue(ctx, trustedCallContextKey{}, newTrustedCall(delegationv1.Claims{
		DelegationID: "delegation-1", OrgID: testOrgID, PrincipalID: "user-1", SessionID: "session-1",
		CallerWorkloadID: testWorkloadID, AppInstallationID: testInstallationID,
		RuntimeAudience: "runtime-test", Capabilities: []string{bulkingestsemantic.CapabilityName}, AuthEpoch: 2, SecurityEpoch: 3,
	}, models.UserSession{ID: "user-1"}))
}

type bulkIngestUploadTestStream struct {
	ctx               context.Context
	requests          []*runtimev1.UploadBulkIngestDataRequest
	waitAfterRequests <-chan struct{}
	response          *runtimev1.UploadBulkIngestDataResponse
}

func (s *bulkIngestUploadTestStream) Recv() (*runtimev1.UploadBulkIngestDataRequest, error) {
	if len(s.requests) == 0 {
		if s.waitAfterRequests != nil {
			select {
			case <-s.waitAfterRequests:
			case <-s.ctx.Done():
				return nil, s.ctx.Err()
			}
		}
		return nil, io.EOF
	}
	request := s.requests[0]
	s.requests = s.requests[1:]
	return request, nil
}

func (s *bulkIngestUploadTestStream) SendAndClose(response *runtimev1.UploadBulkIngestDataResponse) error {
	s.response = response
	return nil
}

func (*bulkIngestUploadTestStream) SetHeader(metadata.MD) error  { return nil }
func (*bulkIngestUploadTestStream) SendHeader(metadata.MD) error { return nil }
func (*bulkIngestUploadTestStream) SetTrailer(metadata.MD)       {}
func (s *bulkIngestUploadTestStream) Context() context.Context   { return s.ctx }
func (*bulkIngestUploadTestStream) SendMsg(any) error            { return nil }
func (*bulkIngestUploadTestStream) RecvMsg(any) error            { return nil }