package sfcrmappservice

import (
	"bytes"
	"context"
	"encoding/json"
	"io"
	"net/http"
	"net/http/httptest"
	"reflect"
	"strings"
	"testing"
	"time"

	crmwire "github.com/pingcap-inc/nexuscrm-app-packs/sfcrm/wire"
	bulkingestsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/bulkingestv1"
	commonsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commonv1"
	querysemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/queryv1"
)

type bulkIngestHTTPTestRuntime struct {
	t          *testing.T
	job        bulkingestsemantic.Job
	bearers    []string
	upload     *bulkIngestHTTPTestUpload
	beginCalls int
	closeCalls int
}

func (runtime *bulkIngestHTTPTestRuntime) create(_ context.Context, bearer string, request bulkingestsemantic.CreateRequest) (bulkingestsemantic.CreateResponse, error) {
	runtime.bearers = append(runtime.bearers, bearer)
	runtime.job.ObjectID = request.ObjectID
	runtime.job.Operation = request.Operation
	runtime.job.ExternalIDFieldName = request.ExternalIDFieldName
	runtime.job.ClientVersion = request.ClientVersion
	return bulkingestsemantic.CreateResponse{Job: runtime.job}, nil
}

func (runtime *bulkIngestHTTPTestRuntime) read(_ context.Context, bearer string, request bulkingestsemantic.ReadRequest) (bulkingestsemantic.ReadResponse, error) {
	runtime.bearers = append(runtime.bearers, bearer)
	if request.JobID != runtime.job.ID {
		runtime.t.Fatalf("read job ID = %q", request.JobID)
	}
	return bulkingestsemantic.ReadResponse{Job: runtime.job}, nil
}

func (runtime *bulkIngestHTTPTestRuntime) close(_ context.Context, bearer string, request bulkingestsemantic.CloseRequest) (bulkingestsemantic.CloseResponse, error) {
	runtime.closeCalls++
	runtime.bearers = append(runtime.bearers, bearer)
	if request.JobID != runtime.job.ID || request.State != bulkingestsemantic.StateUploadComplete {
		runtime.t.Fatalf("close request = %#v", request)
	}
	runtime.job.State = request.State
	return bulkingestsemantic.CloseResponse{Job: runtime.job}, nil
}

func TestBulkIngestRejectsUnsupportedPatchStateBeforeRuntime(t *testing.T) {
	runtime := newBulkIngestHTTPTestRuntime(t)
	service := newBulkIngestHTTPTestService(t, runtime)
	request := bulkIngestHTTPRequest(http.MethodPatch, "/services/data/v63.0/jobs/ingest/job-1", strings.NewReader(`{"state":"InProgress"}`))
	recorder := httptest.NewRecorder()

	service.ServeHTTP(recorder, request)

	if recorder.Code != http.StatusBadRequest ||
		!strings.Contains(recorder.Body.String(), `"errorCode":"INVALID_REQUEST"`) ||
		runtime.closeCalls != 0 {
		t.Fatalf("unsupported PATCH state = %d calls=%d body=%s", recorder.Code, runtime.closeCalls, recorder.Body.String())
	}
}

func TestBulkIngestUsesTheAppServiceSessionFailureContract(t *testing.T) {
	runtime := newBulkIngestHTTPTestRuntime(t)
	service := newBulkIngestHTTPTestService(t, runtime)
	request := bulkIngestHTTPRequest(http.MethodGet, "/services/data/v63.0/jobs/ingest/job-1", nil)
	request.Header.Del("Authorization")
	recorder := httptest.NewRecorder()

	service.ServeHTTP(recorder, request)

	want := `[{"errorCode":"INVALID_SESSION_ID","message":"Session expired or invalid"}]`
	if recorder.Code != http.StatusUnauthorized || recorder.Body.String() != want || len(runtime.bearers) != 0 {
		t.Fatalf("unauthenticated Bulk Ingest = %d body=%q Runtime calls=%d", recorder.Code, recorder.Body.String(), len(runtime.bearers))
	}
}

func (runtime *bulkIngestHTTPTestRuntime) begin(ctx context.Context, bearer string, header bulkingestsemantic.UploadHeader) (RuntimeBulkIngestUpload, error) {
	runtime.bearers = append(runtime.bearers, bearer)
	runtime.beginCalls++
	state, err := bulkingestsemantic.NewUploadStreamState(ctx)
	if err != nil {
		return nil, err
	}
	if err := state.Observe(bulkingestsemantic.UploadChunk{Header: &header}); err != nil {
		return nil, err
	}
	runtime.upload = &bulkIngestHTTPTestUpload{state: state, header: header}
	return runtime.upload, nil
}

type bulkIngestHTTPTestUpload struct {
	state    *bulkingestsemantic.UploadStreamState
	header   bulkingestsemantic.UploadHeader
	body     bytes.Buffer
	maxChunk int
	canceled bool
}

func (upload *bulkIngestHTTPTestUpload) Send(data []byte) error {
	if len(data) > upload.maxChunk {
		upload.maxChunk = len(data)
	}
	if err := upload.state.Observe(bulkingestsemantic.UploadChunk{Data: data}); err != nil {
		return err
	}
	_, _ = upload.body.Write(data)
	return nil
}

func (upload *bulkIngestHTTPTestUpload) CloseAndRecv() (bulkingestsemantic.UploadResponse, error) {
	if err := upload.state.Complete(); err != nil {
		return bulkingestsemantic.UploadResponse{}, err
	}
	return bulkingestsemantic.UploadResponse{JobID: upload.header.JobID}, nil
}

func (upload *bulkIngestHTTPTestUpload) Cancel() { upload.canceled = true }

func TestBulkIngestPublicHTTPLifecycleReachesTheDurableRuntimeBoundary(t *testing.T) {
	runtime := newBulkIngestHTTPTestRuntime(t)
	service := newBulkIngestHTTPTestService(t, runtime)

	create := bulkIngestHTTPRequest(http.MethodPost, "/services/data/v59.0/jobs/ingest", strings.NewReader(`{"object":"Account","operation":"upsert","externalIdFieldName":"External_Id__c"}`))
	createRecorder := httptest.NewRecorder()
	service.ServeHTTP(createRecorder, create)
	wantCreate := `{"id":"job-1","object":"Account","operation":"upsert","state":"Open","concurrencyMode":"Parallel","createdById":"user-1","createdDate":"2026-08-27T10:00:00Z","systemModstamp":"2026-08-27T10:00:00Z","contentType":"CSV","columnDelimiter":"COMMA","lineEnding":"LF","apiVersion":59,"externalIdFieldName":"External_Id__c","numberRecordsProcessed":0,"numberRecordsFailed":0,"totalProcessingTime":0,"apiActiveProcessingTime":0,"apexProcessingTime":0}`
	if createRecorder.Code != http.StatusCreated || createRecorder.Body.String() != wantCreate {
		t.Fatalf("create = %d %s", createRecorder.Code, createRecorder.Body.String())
	}

	csv := "External_Id__c,Name\naccount-1,Acme\n"
	upload := bulkIngestHTTPRequest(http.MethodPut, "/services/data/v63.0/jobs/ingest/job-1/batches", strings.NewReader(csv))
	upload.ContentLength = -1 // Prove chunked HTTP does not require buffering.
	uploadRecorder := httptest.NewRecorder()
	service.ServeHTTP(uploadRecorder, upload)
	if uploadRecorder.Code != http.StatusCreated || runtime.upload == nil || runtime.upload.body.String() != csv || runtime.upload.header.ContentLengthKnown || runtime.upload.maxChunk > bulkingestsemantic.MaxChunkBytes {
		t.Fatalf("upload = %d header=%#v chunk=%d body=%q", uploadRecorder.Code, runtime.upload.header, runtime.upload.maxChunk, runtime.upload.body.String())
	}

	closeRequest := bulkIngestHTTPRequest(http.MethodPatch, "/services/data/v63.0/jobs/ingest/job-1", strings.NewReader(`{"state":"UploadComplete"}`))
	closeRecorder := httptest.NewRecorder()
	service.ServeHTTP(closeRecorder, closeRequest)
	if closeRecorder.Code != http.StatusOK || !strings.Contains(closeRecorder.Body.String(), `"state":"UploadComplete"`) || !strings.Contains(closeRecorder.Body.String(), `"apiVersion":59`) {
		t.Fatalf("close = %d %s", closeRecorder.Code, closeRecorder.Body.String())
	}

	readRecorder := httptest.NewRecorder()
	service.ServeHTTP(readRecorder, bulkIngestHTTPRequest(http.MethodGet, "/services/data/v63.0/jobs/ingest/job-1", nil))
	if readRecorder.Code != http.StatusOK || !strings.Contains(readRecorder.Body.String(), `"state":"UploadComplete"`) || !strings.Contains(readRecorder.Body.String(), `"apiVersion":59`) {
		t.Fatalf("read = %d %s", readRecorder.Code, readRecorder.Body.String())
	}
	if len(runtime.bearers) != 4 {
		t.Fatalf("Runtime lifecycle calls = %d, want 4", len(runtime.bearers))
	}
	for _, bearer := range runtime.bearers {
		if bearer != "user-secret" {
			t.Fatalf("Runtime bearer = %q", bearer)
		}
	}
}

func TestBulkIngestJobProjectionCoversTheCompleteNeutralContract(t *testing.T) {
	jobType := reflect.TypeOf(bulkingestsemantic.Job{})
	gotFields := make([]string, 0, jobType.NumField())
	for index := 0; index < jobType.NumField(); index++ {
		gotFields = append(gotFields, jobType.Field(index).Name)
	}
	wantFields := []string{
		"ID", "ObjectID", "Operation", "State", "CreatedByID", "CreatedAt", "ModifiedAt", "FailureCode",
		"ExternalIDFieldName", "NumberRecordsProcessed", "NumberRecordsFailed", "TotalProcessingTime", "ActiveProcessingTime", "ClientVersion",
	}
	if !reflect.DeepEqual(gotFields, wantFields) {
		t.Fatalf("Runtime Bulk Ingest Job fields changed; review the public projection: got %v want %v", gotFields, wantFields)
	}

	projected, err := json.Marshal(projectBulkIngestJob(bulkingestsemantic.Job{
		ID: "job-complete", ObjectID: "Account", Operation: bulkingestsemantic.OperationUpsert, State: bulkingestsemantic.StateFailed,
		CreatedByID: "user-1", CreatedAt: "2026-08-27T10:00:00Z", ModifiedAt: "2026-08-27T10:05:00Z",
		FailureCode: bulkingestsemantic.FailureProcessingFailed, ExternalIDFieldName: "External_Id__c",
		NumberRecordsProcessed: 7, NumberRecordsFailed: 2, TotalProcessingTime: 1234, ActiveProcessingTime: 987, ClientVersion: 62,
	}))
	if err != nil {
		t.Fatal(err)
	}
	want := `{"id":"job-complete","object":"Account","operation":"upsert","state":"Failed","concurrencyMode":"Parallel","createdById":"user-1","createdDate":"2026-08-27T10:00:00Z","systemModstamp":"2026-08-27T10:05:00Z","errorMessage":"Bulk ingest processing failed","contentType":"CSV","columnDelimiter":"COMMA","lineEnding":"LF","apiVersion":62,"externalIdFieldName":"External_Id__c","numberRecordsProcessed":7,"numberRecordsFailed":2,"totalProcessingTime":1234,"apiActiveProcessingTime":987,"apexProcessingTime":0}`
	if string(projected) != want {
		t.Fatalf("complete Bulk Ingest projection = %s, want %s", projected, want)
	}
}

func TestBulkIngestRouteUsesTheCanonicalSupportedAPIVersions(t *testing.T) {
	for _, path := range []string{
		"/services/data/v59/jobs/ingest",
		"/services/data/v59.00/jobs/ingest",
		"/services/data/v58.0/jobs/ingest",
		"/services/data/v64.0/jobs/ingest",
	} {
		if _, matched := matchBulkIngestRoute(path); matched {
			t.Fatalf("non-canonical or unsupported route matched: %s", path)
		}
	}
	for _, version := range []string{"v59.0", "v60.0", "v61.0", "v62.0", "v63.0"} {
		if route, matched := matchBulkIngestRoute("/services/data/" + version + "/jobs/ingest"); !matched || route.version == 0 {
			t.Fatalf("supported route did not match: %s", version)
		}
	}
}

func TestBulkIngestRejectsOversizedKnownBodyBeforeRuntimeStream(t *testing.T) {
	runtime := newBulkIngestHTTPTestRuntime(t)
	service := newBulkIngestHTTPTestService(t, runtime)
	request := bulkIngestHTTPRequest(http.MethodPut, "/services/data/v63.0/jobs/ingest/job-1/batches", strings.NewReader("not read"))
	request.ContentLength = int64(bulkingestsemantic.MaxUploadBytes) + 1
	recorder := httptest.NewRecorder()
	service.ServeHTTP(recorder, request)
	if recorder.Code != http.StatusRequestEntityTooLarge || runtime.beginCalls != 0 || !strings.Contains(recorder.Body.String(), `"errorCode":"FILE_TOO_LARGE"`) {
		t.Fatalf("oversized upload = %d calls=%d body=%s", recorder.Code, runtime.beginCalls, recorder.Body.String())
	}
}

func TestBulkIngestRequiresTheCompleteRuntimeLifecycle(t *testing.T) {
	runtime := newBulkIngestHTTPTestRuntime(t)
	config := baseBulkIngestHTTPTestConfig(t)
	config.RuntimeCreateBulkIngestJob = runtime.create
	if service, err := NewService(config); err == nil || service != nil {
		t.Fatalf("partial Bulk Ingest composition = %#v, %v", service, err)
	}
}

func TestBulkIngestMapsRuntimeOperationFailuresByPublicRequestContext(t *testing.T) {
	tests := []struct {
		name      string
		operation bulkIngestFailureOperation
		wantCode  string
	}{
		{name: "create rejects an unsupported operation", operation: bulkIngestFailureCreate, wantCode: crmwire.SFCodeInvalidOperation},
		{name: "close rejects an invalid job transition", operation: bulkIngestFailureClose, wantCode: crmwire.SFCodeInvalidJobState},
		{name: "upload rejects an invalid job transition", operation: bulkIngestFailureUpload, wantCode: crmwire.SFCodeInvalidJobState},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			recorder := httptest.NewRecorder()
			writeBulkIngestFailure(recorder, &commonsemantic.Failure{
				Category: commonsemantic.FailureInvalidRequest, InvalidComponent: commonsemantic.InvalidComponentOperation,
			}, tt.operation)
			if recorder.Code != http.StatusBadRequest || !strings.Contains(recorder.Body.String(), `"errorCode":"`+tt.wantCode+`"`) {
				t.Fatalf("failure projection = %d %s", recorder.Code, recorder.Body.String())
			}
		})
	}
}

func TestBulkIngestPreservesPublicPermissionAndJobAccessErrors(t *testing.T) {
	for _, test := range []struct {
		name      string
		operation bulkIngestFailureOperation
		wantCode  string
	}{
		{name: "create permission", operation: bulkIngestFailureCreate, wantCode: crmwire.SFCodeInsufficientPerms},
		{name: "existing job access", operation: bulkIngestFailureRead, wantCode: crmwire.SFCodeForbidden},
	} {
		t.Run(test.name, func(t *testing.T) {
			recorder := httptest.NewRecorder()
			writeBulkIngestFailure(recorder, &commonsemantic.Failure{
				Category: commonsemantic.FailureForbidden,
			}, test.operation)
			if recorder.Code != http.StatusForbidden || !strings.Contains(recorder.Body.String(), `"errorCode":"`+test.wantCode+`"`) {
				t.Fatalf("forbidden projection = %d %s", recorder.Code, recorder.Body.String())
			}
		})
	}
}

func TestBulkIngestResourceLimitsPreserveTheirOwningWireSemantics(t *testing.T) {
	for _, test := range []struct {
		name           string
		failure        *commonsemantic.Failure
		wantStatus     int
		wantCode       string
		wantRetryAfter string
	}{
		{
			name: "upload bytes", failure: &commonsemantic.Failure{
				Category: commonsemantic.FailureResourceExhausted, LimitKind: commonsemantic.LimitUploadBytes,
			}, wantStatus: http.StatusRequestEntityTooLarge, wantCode: crmwire.SFCodeFileTooLarge,
		},
		{
			name: "runtime admission", failure: &commonsemantic.Failure{
				Category: commonsemantic.FailureResourceExhausted, LimitKind: commonsemantic.LimitAdmission, RetryAfterMillis: 50,
			}, wantStatus: http.StatusServiceUnavailable, wantCode: crmwire.SFCodeUnknownException, wantRetryAfter: "1",
		},
		{
			name: "other semantic limit", failure: &commonsemantic.Failure{
				Category: commonsemantic.FailureResourceExhausted, LimitKind: commonsemantic.LimitSemanticRequest,
			}, wantStatus: http.StatusBadRequest, wantCode: crmwire.SFCodeLimitExceeded,
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			recorder := httptest.NewRecorder()
			writeBulkIngestFailure(recorder, test.failure, bulkIngestFailureUpload)
			if recorder.Code != test.wantStatus ||
				!strings.Contains(recorder.Body.String(), `"errorCode":"`+test.wantCode+`"`) ||
				recorder.Header().Get("Retry-After") != test.wantRetryAfter {
				t.Fatalf("failure projection = status %d retry-after %q body %s", recorder.Code, recorder.Header().Get("Retry-After"), recorder.Body.String())
			}
		})
	}
}

func newBulkIngestHTTPTestRuntime(t *testing.T) *bulkIngestHTTPTestRuntime {
	t.Helper()
	return &bulkIngestHTTPTestRuntime{t: t, job: bulkingestsemantic.Job{
		ID: "job-1", ObjectID: "Account", Operation: bulkingestsemantic.OperationInsert, State: bulkingestsemantic.StateOpen,
		CreatedByID: "user-1", CreatedAt: "2026-08-27T10:00:00Z", ModifiedAt: "2026-08-27T10:00:00Z",
	}}
}

func newBulkIngestHTTPTestService(t *testing.T, runtime *bulkIngestHTTPTestRuntime) *Service {
	t.Helper()
	config := baseBulkIngestHTTPTestConfig(t)
	config.RuntimeCreateBulkIngestJob = runtime.create
	config.RuntimeReadBulkIngestJob = runtime.read
	config.RuntimeCloseBulkIngestJob = runtime.close
	config.RuntimeBeginBulkUpload = runtime.begin
	service, err := NewService(config)
	if err != nil {
		t.Fatal(err)
	}
	return service
}

func baseBulkIngestHTTPTestConfig(t *testing.T) ServiceConfig {
	t.Helper()
	runtime := &fakeRuntime{}
	runtime.ready.Store(true)
	return ServiceConfig{
		RuntimeEvaluate: runtime.Evaluate, RuntimeReady: runtime.Ready, RuntimeMetrics: runtime.Metrics,
		RuntimeQuery: func(context.Context, string, querysemantic.Request) (querysemantic.Response, error) {
			return querysemantic.Response{}, nil
		},
		RuntimeHistoryQuery: func(context.Context, string, querysemantic.HistoryRequest) (querysemantic.HistoryResponse, error) {
			return querysemantic.HistoryResponse{}, nil
		},
		RequestTimeout: time.Second, MaxRequestBytes: 64 * 1024,
	}
}

func bulkIngestHTTPRequest(method, target string, body io.Reader) *http.Request {
	request := httptest.NewRequest(method, target, body)
	request.Header.Set("Authorization", "Bearer user-secret")
	return request
}