package records

import (
	"context"
	"encoding/csv"
	"errors"
	"io"
	"strings"
	"testing"

	appErrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/models"
	semantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordmutationv1"
)

type failingBulkCSVReader struct {
	data []byte
	err  error
}

func (r *failingBulkCSVReader) Read(buffer []byte) (int, error) {
	if len(r.data) == 0 {
		return 0, r.err
	}
	read := copy(buffer, r.data)
	r.data = r.data[read:]
	return read, nil
}

func TestReadBulkCSVChunkPreservesQuotedNewlinesAndExactCursorUnits(t *testing.T) {
	reader := csv.NewReader(strings.NewReader("Acme,\"line one\nline two\"\nBeta,plain\n"))
	rows, failures, eof, err := readBulkCSVChunk(reader, []string{"name", "description"}, 1)
	if err != nil || failures != 0 || eof || len(rows) != 1 {
		t.Fatalf("first chunk = %#v failures=%d eof=%t err=%v", rows, failures, eof, err)
	}
	if rows[0]["name"] != "Acme" || rows[0]["description"] != "line one\nline two" {
		t.Fatalf("quoted record = %#v", rows[0])
	}
	rows, failures, eof, err = readBulkCSVChunk(reader, []string{"name", "description"}, 10)
	if err != nil || failures != 0 || !eof || len(rows) != 1 || rows[0]["name"] != "Beta" {
		t.Fatalf("second chunk = %#v failures=%d eof=%t err=%v", rows, failures, eof, err)
	}
}

func TestBulkDispositionConsumesOnlyTheTypedAbortRace(t *testing.T) {
	rolledBackAbort := errors.Join(
		persistence.ErrBulkClaimTransactionRolledBack,
		persistence.ErrBulkIngestJobAborted,
	)
	if err := consumeBulkAbort(rolledBackAbort); err != nil {
		t.Fatalf("typed abort reached the shared controller: %v", err)
	}
	infrastructure := errors.New("database connection disappeared")
	if err := consumeBulkAbort(infrastructure); !errors.Is(err, infrastructure) {
		t.Fatalf("infrastructure failure was consumed: %v", err)
	}
}

type bulkMutationExecutorFake struct {
	insertOptions BulkInsertOptions
	insertResult  *BulkInsertResult
	insertErr     error
	updated       []models.SObject
	deleted       []string
	upsertField   string
	upsertValues  []string
	failureID     string
}

func (*bulkMutationExecutorFake) LoadUserSession(context.Context, string) (*models.UserSession, error) {
	return &models.UserSession{ID: "005-owner"}, nil
}

func (*bulkMutationExecutorFake) RecordMutationTransactionObjectNames(_ context.Context, objectName string, _ semantic.OperationKind) []string {
	return []string{objectName}
}

func (*bulkMutationExecutorFake) RunRecordMutationTransactionWithOutcome(
	ctx context.Context,
	_ []string,
	fn func(context.Context) error,
) (ports.TransactionCompletionOutcome, error) {
	err := fn(ctx)
	if err != nil {
		return ports.TransactionCompletionRolledBack, err
	}
	return ports.TransactionCompletionCommitted, nil
}

func (f *bulkMutationExecutorFake) BulkInsert(_ context.Context, _ string, rows []models.SObject, _ *models.UserSession, options BulkInsertOptions) (BulkInsertResult, error) {
	f.insertOptions = options
	if f.insertResult != nil || f.insertErr != nil {
		var result BulkInsertResult
		if f.insertResult != nil {
			result = *f.insertResult
		}
		return result, f.insertErr
	}
	return BulkInsertResult{SuccessCount: len(rows)}, nil
}

func (f *bulkMutationExecutorFake) Update(_ context.Context, _ string, id string, row models.SObject, _ *models.UserSession) error {
	f.updated = append(f.updated, row)
	if id == f.failureID {
		return appErrors.NewValidationError("Id", "update rejected")
	}
	return nil
}

func (f *bulkMutationExecutorFake) Delete(_ context.Context, _ string, id string, _ *models.UserSession) error {
	f.deleted = append(f.deleted, id)
	if id == f.failureID {
		return appErrors.NewValidationError("Id", "delete rejected")
	}
	return nil
}

func (f *bulkMutationExecutorFake) UpsertByExternalID(_ context.Context, _ string, field, value string, _ models.SObject, _ *models.UserSession) (models.SObject, bool, error) {
	f.upsertField = field
	f.upsertValues = append(f.upsertValues, value)
	if value == f.failureID {
		return nil, false, appErrors.NewValidationError(field, "upsert rejected")
	}
	return models.SObject{"Id": value}, true, nil
}

func TestBulkWorkSourceExecutesEveryAcceptedOperation(t *testing.T) {
	rows := []models.SObject{{"id": "001-ok", "external_id__c": "ext-ok"}, {"Id": "001-fail", "external_id__c": "ext-fail"}}
	user := &models.UserSession{ID: "005-owner"}

	t.Run("insert preserves the established CSV normalization path", func(t *testing.T) {
		executor := &bulkMutationExecutorFake{}
		source := &BulkWorkSource{records: executor}
		succeeded, failed, messages, err := source.applyChunk(t.Context(), &models.BulkJob{Object: "Account", Operation: models.OperationInsert}, rows, user)
		if err != nil || succeeded != 2 || failed != 0 || len(messages) != 0 || !executor.insertOptions.NormalizeFromCSV || executor.insertOptions.BatchSize != 100 {
			t.Fatalf("insert = %d/%d messages=%v options=%+v err=%v", succeeded, failed, messages, executor.insertOptions, err)
		}
	})

	t.Run("update executes one independent row", func(t *testing.T) {
		executor := &bulkMutationExecutorFake{}
		source := &BulkWorkSource{records: executor}
		succeeded, failed, _, err := source.applyChunk(t.Context(), &models.BulkJob{Object: "Account", Operation: models.OperationUpdate}, rows[:1], user)
		if err != nil || succeeded != 1 || failed != 0 || len(executor.updated) != 1 {
			t.Fatalf("update = %d/%d rows=%v err=%v", succeeded, failed, executor.updated, err)
		}
		if _, stillPresent := executor.updated[0]["id"]; stillPresent {
			t.Fatalf("update payload still contains identity field: %#v", executor.updated[0])
		}
	})

	t.Run("delete executes one independent row", func(t *testing.T) {
		executor := &bulkMutationExecutorFake{}
		source := &BulkWorkSource{records: executor}
		succeeded, failed, messages, err := source.applyChunk(t.Context(), &models.BulkJob{Object: "Account", Operation: models.OperationDelete}, rows[:1], user)
		if err != nil || succeeded != 1 || failed != 0 || len(messages) != 0 || len(executor.deleted) != 1 {
			t.Fatalf("delete = %d/%d messages=%v ids=%v err=%v", succeeded, failed, messages, executor.deleted, err)
		}
	})

	t.Run("upsert uses the declared external ID", func(t *testing.T) {
		executor := &bulkMutationExecutorFake{}
		source := &BulkWorkSource{records: executor}
		succeeded, failed, messages, err := source.applyChunk(t.Context(), &models.BulkJob{Object: "Account", Operation: models.OperationUpsert, ExternalIdFieldName: "External_Id__c"}, rows[:1], user)
		if err != nil || succeeded != 1 || failed != 0 || len(messages) != 0 || executor.upsertField != "external_id__c" || len(executor.upsertValues) != 1 {
			t.Fatalf("upsert = %d/%d messages=%v field=%q values=%v err=%v", succeeded, failed, messages, executor.upsertField, executor.upsertValues, err)
		}
	})
}

func TestBulkWorkSourceIndependentMutationClassifiesOnlyKnownDomainFailures(t *testing.T) {
	job := &models.BulkJob{Object: "Account", Operation: models.OperationUpdate}
	domain := appErrors.NewValidationError("Name", "rejected")
	var wrapped bulkRowDomainFailure
	if err := classifyBulkRowMutationError(job, "001-row", domain); !errors.As(err, &wrapped) {
		t.Fatalf("validation failure = %T %v, want bulkRowDomainFailure", err, err)
	}
	infrastructure := errors.New("database connection disappeared")
	if err := classifyBulkRowMutationError(job, "001-row", infrastructure); !errors.Is(err, infrastructure) || errors.As(err, &wrapped) {
		t.Fatalf("unknown infrastructure failure = %T %v, want unchanged retryable boundary error", err, err)
	}
}

func TestBulkWorkSourceClassifiesRealPersistenceDeleteDenialAsRowFailure(t *testing.T) {
	service, user := runtimeDeleteProjectionFixture(true, true)
	err := service.Delete(t.Context(), "ProbeObject", "a00-hidden", user)
	if err == nil {
		t.Fatal("real PersistenceService.Delete denial = nil")
	}

	classified := classifyBulkRowMutationError(
		&models.BulkJob{Object: "ProbeObject", Operation: models.OperationDelete},
		"a00-hidden",
		err,
	)
	var rowFailure bulkRowDomainFailure
	if !errors.As(classified, &rowFailure) {
		t.Fatalf("real delete denial classified as %T %v, want durable row failure", classified, classified)
	}
}

func TestBulkWorkSourceInsertRequiresCompleteRowOutcomeBeforeCheckpoint(t *testing.T) {
	rows := []models.SObject{{"name": "invalid"}, {"name": "valid"}}
	topLevelFailure := errors.New("transaction failed after validating one row")
	executor := &bulkMutationExecutorFake{
		insertResult: &BulkInsertResult{FailedCount: 1, Errors: []string{"record 0: invalid"}},
		insertErr:    topLevelFailure,
	}
	source := &BulkWorkSource{records: executor}

	_, _, rowErrors, err := source.applyChunk(
		t.Context(), &models.BulkJob{Object: "Account", Operation: models.OperationInsert},
		rows, &models.UserSession{ID: "005-owner"},
	)
	if !errors.Is(err, topLevelFailure) || len(rowErrors) != 0 {
		t.Fatalf("incomplete insert outcome = errors:%v err:%v, want transaction failure only", rowErrors, err)
	}
}

func TestBulkWorkSourceInsertCarriesCompleteRejectionsThroughRequiredRollback(t *testing.T) {
	rows := []models.SObject{{"name": "invalid-one"}, {"name": "invalid-two"}}
	domainErr := appErrors.NewValidationError("", "all 2 records failed validation")
	executor := &bulkMutationExecutorFake{
		insertResult: &BulkInsertResult{FailedCount: 2, Errors: []string{"record 0: invalid", "record 1: invalid"}},
		insertErr:    domainErr,
	}
	source := &BulkWorkSource{records: executor}

	succeeded, failed, rowErrors, err := source.applyChunk(
		t.Context(), &models.BulkJob{Object: "Account", Operation: models.OperationInsert},
		rows, &models.UserSession{ID: "005-owner"},
	)
	var chunk *bulkChunkDomainFailure
	if !errors.As(err, &chunk) || !errors.Is(err, domainErr) || succeeded != 0 || failed != 0 || len(rowErrors) != 0 ||
		chunk.failed != 2 || len(chunk.messages) != 2 {
		t.Fatalf("rollback-required rejection = %d/%d errors:%v err:%v chunk=%#v", succeeded, failed, rowErrors, err, chunk)
	}
}

func TestBulkWorkSourceInsertRejectsCompleteLookingSuccessesWhenTransactionFailed(t *testing.T) {
	rows := []models.SObject{{"name": "apparently-valid"}, {"name": "invalid"}}
	topLevelFailure := errors.New("post-write lifecycle flush failed")
	executor := &bulkMutationExecutorFake{
		insertResult: &BulkInsertResult{SuccessCount: 1, FailedCount: 1, Errors: []string{"record 1: invalid"}},
		insertErr:    topLevelFailure,
	}
	source := &BulkWorkSource{records: executor}

	succeeded, failed, rowErrors, err := source.applyChunk(
		t.Context(), &models.BulkJob{Object: "Account", Operation: models.OperationInsert},
		rows, &models.UserSession{ID: "005-owner"},
	)
	if !errors.Is(err, topLevelFailure) || succeeded != 0 || failed != 0 || len(rowErrors) != 0 {
		t.Fatalf("transaction-failed insert = %d/%d errors:%v err:%v, want rollback signal only", succeeded, failed, rowErrors, err)
	}
}

func TestReadBulkCSVChunkAccountsMalformedRowsWithoutLosingFollowingRows(t *testing.T) {
	reader := csv.NewReader(strings.NewReader("bad,too,many\ngood,value\n"))
	reader.FieldsPerRecord = 2
	rows, failures, eof, err := readBulkCSVChunk(reader, []string{"name", "value"}, 10)
	if err != nil || failures != 1 || !eof || len(rows) != 1 || rows[0]["name"] != "good" {
		t.Fatalf("chunk = %#v failures=%d eof=%t err=%v", rows, failures, eof, err)
	}
}

func TestSkipBulkCSVRowsReplaysCursorAcrossMalformedRows(t *testing.T) {
	reader := csv.NewReader(strings.NewReader("bad,too,many\ngood,value\n"))
	reader.FieldsPerRecord = 2
	if err := skipBulkCSVRows(reader, 1); err != nil {
		t.Fatalf("skip malformed durable row: %v", err)
	}
	row, err := reader.Read()
	if err != nil || len(row) != 2 || row[0] != "good" {
		t.Fatalf("row after durable cursor = %#v, %v", row, err)
	}
}

func TestReadBulkCSVChunkDoesNotCheckpointAcrossObjectReadFailure(t *testing.T) {
	readFailure := errors.New("object storage read interrupted")
	reader := csv.NewReader(&failingBulkCSVReader{
		data: []byte("good,value\npartial"),
		err:  readFailure,
	})
	reader.FieldsPerRecord = 2

	rows, failures, eof, err := readBulkCSVChunk(reader, []string{"name", "value"}, 10)
	if !errors.Is(err, readFailure) || rows != nil || failures != 0 || eof {
		t.Fatalf("interrupted chunk = %#v failures=%d eof=%t err=%v", rows, failures, eof, err)
	}
}

func TestSkipBulkCSVRowsPropagatesObjectReadFailureAfterMalformedRow(t *testing.T) {
	readFailure := io.ErrUnexpectedEOF
	reader := csv.NewReader(&failingBulkCSVReader{
		data: []byte("bad,too,many\npartial"),
		err:  readFailure,
	})
	reader.FieldsPerRecord = 2

	if err := skipBulkCSVRows(reader, 2); !errors.Is(err, readFailure) {
		t.Fatalf("skip swallowed object read failure: %v", err)
	}
}

func TestBulkCSVRecordIDIsCaseInsensitiveForSalesforceCSV(t *testing.T) {
	for _, row := range []models.SObject{{"id": "001-lower"}, {"Id": "001-upper"}} {
		if got := bulkCSVRecordID(row); !strings.HasPrefix(got, "001-") {
			t.Fatalf("bulkCSVRecordID(%v) = %q", row, got)
		}
	}
}

func TestNextBulkBatchCheckpointIsReplaySafeUntilCommit(t *testing.T) {
	base := &models.BulkBatch{
		NextRowOffset: 10, RecordsProcessed: 8, RecordsFailed: 2, TotalRecords: 10,
	}
	first := nextBulkBatchCheckpoint(base, 5, 4, 1, []string{"one rejected row"}, false)
	second := nextBulkBatchCheckpoint(base, 5, 4, 1, []string{"one rejected row"}, false)

	if first != second {
		t.Fatalf("transaction replay changed checkpoint: first=%+v second=%+v", first, second)
	}
	if base.NextRowOffset != 10 || base.RecordsProcessed != 8 || base.RecordsFailed != 2 || base.TotalRecords != 10 {
		t.Fatalf("candidate calculation mutated durable base before commit: %+v", base)
	}
	if first.nextRowOffset != 15 || first.recordsProcessed != 12 || first.recordsFailed != 3 ||
		first.totalRecords != 15 || first.state != "InProgress" || first.lastError != "one rejected row" {
		t.Fatalf("checkpoint = %+v", first)
	}
}