package bulkingestv1

import (
	"context"
	"errors"
	"math"
	"testing"
	"time"

	common "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commonv1"
)

func TestUploadStreamOwnsCompleteBoundedContract(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	header := UploadHeader{JobID: "job-1", ContentLength: 3, ContentLengthKnown: true}
	state, err := NewUploadStreamState(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if err := state.Observe(UploadChunk{Header: &header}); err != nil {
		t.Fatal(err)
	}
	if err := state.Observe(UploadChunk{Data: []byte("abc")}); err != nil {
		t.Fatal(err)
	}
	if err := state.Complete(); err != nil {
		t.Fatalf("Complete() = %v", err)
	}

	for name, run := range map[string]func(*UploadStreamState) error{
		"data before header": func(s *UploadStreamState) error { return s.Observe(UploadChunk{Data: []byte("a")}) },
		"duplicate header": func(s *UploadStreamState) error {
			if err := s.Observe(UploadChunk{Header: &header}); err != nil {
				return err
			}
			return s.Observe(UploadChunk{Header: &header})
		},
		"overflow": func(s *UploadStreamState) error {
			if err := s.Observe(UploadChunk{Header: &header}); err != nil {
				return err
			}
			return s.Observe(UploadChunk{Data: []byte("abcd")})
		},
		"incomplete": func(s *UploadStreamState) error {
			if err := s.Observe(UploadChunk{Header: &header}); err != nil {
				return err
			}
			if err := s.Observe(UploadChunk{Data: []byte("a")}); err != nil {
				return err
			}
			return s.Complete()
		},
	} {
		t.Run(name, func(t *testing.T) {
			probe, probeErr := NewUploadStreamState(ctx)
			if probeErr != nil {
				t.Fatal(probeErr)
			}
			if err := run(probe); err == nil {
				t.Fatal("invalid stream was accepted")
			}
		})
	}

	tooLarge := UploadHeader{JobID: "job-1", ContentLength: MaxUploadBytes + 1, ContentLengthKnown: true}
	err = ValidateUploadHeader(tooLarge)
	var failure *common.Failure
	if !errors.As(err, &failure) || failure.Category != common.FailureResourceExhausted {
		t.Fatalf("oversized header = %#v", err)
	}
	if ValidateUploadHeader(UploadHeader{JobID: "job-1", ContentLength: 3}) == nil {
		t.Fatal("unknown-length header with a declared length was accepted")
	}
}

func TestUploadStreamAcceptsUnknownLengthWithoutLosingTheAggregateLimit(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	state, err := NewUploadStreamState(ctx)
	if err != nil {
		t.Fatal(err)
	}
	header := UploadHeader{JobID: "job-1"}
	if err := state.Observe(UploadChunk{Header: &header}); err != nil {
		t.Fatal(err)
	}
	if err := state.Observe(UploadChunk{Data: []byte("chunked")}); err != nil {
		t.Fatal(err)
	}
	if err := state.Complete(); err != nil {
		t.Fatalf("Complete() = %v", err)
	}

	state, err = NewUploadStreamState(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if err := state.Observe(UploadChunk{Header: &header}); err != nil {
		t.Fatal(err)
	}
	state.received = MaxUploadBytes
	err = state.Observe(UploadChunk{Data: []byte("x")})
	var failure *common.Failure
	if !errors.As(err, &failure) || failure.Category != common.FailureResourceExhausted {
		t.Fatalf("aggregate overflow = %#v", err)
	}
}

func TestLifecycleValidationIsClosedAndResponseIsReceiptBound(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	if err := ValidateCreateRequest(ctx, CreateRequest{ObjectID: "Account", Operation: OperationInsert, ClientVersion: 63}); err != nil {
		t.Fatal(err)
	}
	if ValidateCreateRequest(ctx, CreateRequest{ObjectID: "Account", Operation: OperationInsert, ExternalIDFieldName: "External_Id__c", ClientVersion: 63}) == nil {
		t.Fatal("external identity on insert was accepted")
	}
	if err := ValidateCreateRequest(ctx, CreateRequest{ObjectID: "Account", Operation: OperationInsert}); err != nil {
		t.Fatalf("N-1 create without the additive client version = %v", err)
	}
	for name, version := range map[string]float64{
		"negative":          -1,
		"NaN":               math.NaN(),
		"positive infinity": math.Inf(1),
		"negative infinity": math.Inf(-1),
	} {
		t.Run("invalid client version "+name, func(t *testing.T) {
			if ValidateCreateRequest(ctx, CreateRequest{ObjectID: "Account", Operation: OperationInsert, ClientVersion: version}) == nil {
				t.Fatal("invalid client version was accepted")
			}
		})
	}
	if ValidateCloseRequest(ctx, CloseRequest{JobID: "job-1", State: StateInProgress}) == nil {
		t.Fatal("caller-selected in-progress transition was accepted")
	}
	response := ReadResponse{Job: Job{
		ID: "job-1", ObjectID: "Account", Operation: OperationInsert, State: StateOpen,
		CreatedByID: "user-1", CreatedAt: "2026-08-27T10:00:00Z", ModifiedAt: "2026-08-27T10:00:00Z",
		ClientVersion: 63,
	}, Receipts: common.Receipts{ProtocolVersion: "1.0", CapabilityVersion: CapabilityName}}
	request := ReadRequest{JobID: "job-1"}
	if err := ValidateReadResponse(request, response); err != nil {
		t.Fatalf("valid response = %v", err)
	}
	response.Receipts.CapabilityVersion = "runtime.records.other.v1"
	if ValidateReadResponse(request, response) == nil {
		t.Fatal("foreign receipts were accepted")
	}
}

func TestResponsesAreBoundToRequestsAndPersistedOperationShape(t *testing.T) {
	receipts := common.Receipts{ProtocolVersion: "1.0", CapabilityVersion: CapabilityName}
	job := Job{ID: "job-1", ObjectID: "Account", Operation: OperationUpsert, State: StateOpen,
		CreatedByID: "user-1", CreatedAt: "2026-08-27T10:00:00Z", ModifiedAt: "2026-08-27T10:00:00Z", ExternalIDFieldName: "External_Id__c", ClientVersion: 62}
	create := CreateRequest{ObjectID: "Account", Operation: OperationUpsert, ExternalIDFieldName: "External_Id__c", ClientVersion: 62}
	if err := ValidateCreateResponse(create, CreateResponse{Job: job, Receipts: receipts}); err != nil {
		t.Fatal(err)
	}
	job.ClientVersion = 63
	if ValidateCreateResponse(create, CreateResponse{Job: job, Receipts: receipts}) == nil {
		t.Fatal("create response changed its L1-authored client version")
	}
	job.ClientVersion = 62
	job.ExternalIDFieldName = ""
	if ValidateCreateResponse(create, CreateResponse{Job: job, Receipts: receipts}) == nil {
		t.Fatal("upsert response without its external identity was accepted")
	}
	job.ExternalIDFieldName = "External_Id__c"
	job.State = StateInProgress
	if ValidateCreateResponse(create, CreateResponse{Job: job, Receipts: receipts}) == nil {
		t.Fatal("create response with a non-open job was accepted")
	}
	job.State = StateOpen
	if ValidateReadResponse(ReadRequest{JobID: "foreign-job"}, ReadResponse{Job: job, Receipts: receipts}) == nil {
		t.Fatal("read response for a foreign job was accepted")
	}
	job.State = StateFailed
	if ValidateReadResponse(ReadRequest{JobID: job.ID}, ReadResponse{Job: job, Receipts: receipts}) == nil {
		t.Fatal("failed job without a disclosure-safe failure code was accepted")
	}
	job.FailureCode = FailureProcessingFailed
	if err := ValidateReadResponse(ReadRequest{JobID: job.ID}, ReadResponse{Job: job, Receipts: receipts}); err != nil {
		t.Fatalf("safe failed job = %v", err)
	}
	job.State = StateOpen
	job.FailureCode = FailureCode("internal storage topology")
	if ValidateReadResponse(ReadRequest{JobID: job.ID}, ReadResponse{Job: job, Receipts: receipts}) == nil {
		t.Fatal("unknown failure text on a non-failed job was accepted")
	}

	uploadRequest := UploadHeader{JobID: "job-1"}
	if err := ValidateUploadResponse(uploadRequest, UploadResponse{JobID: "job-1", Receipts: receipts}); err != nil {
		t.Fatalf("valid upload response = %v", err)
	}
	if ValidateUploadResponse(uploadRequest, UploadResponse{JobID: "other-job", Receipts: receipts}) == nil {
		t.Fatal("upload response for a foreign job was accepted")
	}
}