package serving

import (
	"context"
	"errors"
	"regexp"
	"strings"
	"testing"

	"github.com/DATA-DOG/go-sqlmock"
)

const profiledVerificationUpdateTestSQL = `
UPDATE _System_ConnectorServingEpoch
   SET State = ?, LastModifiedDate = CURRENT_TIMESTAMP(6)
 WHERE Id = ? AND BINARY Id = BINARY ?
   AND EpochDigest = ? AND BINARY EpochDigest = BINARY ?
   AND State = ? AND BINARY State = BINARY ?
   AND ActiveScopeDigest IS NULL`

func TestProfiledExternalAuthorityVerificationAppliesDurableAssessment(t *testing.T) {
	tests := []struct {
		name          string
		target        ProfiledEpochVerificationState
		rejectCatalog bool
	}{
		{name: "verified", target: ProfiledEpochVerified},
		{name: "rejected", target: ProfiledEpochRejected, rejectCatalog: true},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			db, mock := newServingSchemaMock(t)
			repository := newServingEpochTestRepository(t, db)
			fixture := newExternalAuthorityAssessmentFixture(t)
			assessor, readers := newExternalAuthorityTestAssessor(t, fixture)
			if test.rejectCatalog {
				readers.catalog.found = false
			}
			service, err := newProfiledExternalAuthorityVerificationService(repository, assessor)
			if err != nil {
				t.Fatalf("newProfiledExternalAuthorityVerificationService(): %v", err)
			}

			expectServingSchema(t, mock, exactServingSchemaFixture(), len(servingTableSchemas)*3)
			expectProfiledReaderTestRows(
				mock,
				fixture.candidate.ID(),
				profiledReaderTestExactRows(fixture.candidate),
				false,
			)
			expectServingSchema(t, mock, exactServingSchemaFixture(), len(servingTableSchemas)*3)
			expectProfiledVerificationBegin(mock)
			expectProfiledReaderTestRows(
				mock,
				fixture.candidate.ID(),
				profiledReaderTestExactRows(fixture.candidate),
				true,
			)
			expectProfiledVerificationUpdate(mock, fixture.candidate, test.target, 1, nil)
			terminal := profiledVerificationTerminalRows(fixture.candidate, test.target)
			expectProfiledReaderTestRows(mock, fixture.candidate.ID(), terminal, true)
			mock.ExpectCommit()
			expectProfiledVerificationCleanup(mock)

			result, err := service.Verify(t.Context(), fixture.candidate.ID())
			if err != nil {
				t.Fatalf("Verify(): %v", err)
			}
			assertProfiledVerificationResult(t, result, fixture.candidate, test.target, true)
			if len(readers.epoch.calls) != 0 {
				t.Fatalf("verification reread candidate through assessor: %#v", readers.epoch.calls)
			}
			if len(readers.catalog.minimumReads) != 1 {
				t.Fatalf("Catalog reads = %#v, want one fresh assessment", readers.catalog.minimumReads)
			}
			if test.rejectCatalog {
				if len(readers.binding.selections) != 0 || len(readers.service.ids) != 0 ||
					len(readers.credentials.calls) != 0 {
					t.Fatalf(
						"definitive Catalog rejection continued to downstream owners: binding=%d service=%d credentials=%d",
						len(readers.binding.selections),
						len(readers.service.ids),
						len(readers.credentials.calls),
					)
				}
			} else if len(readers.binding.selections) != 1 || len(readers.service.ids) != 1 ||
				len(readers.credentials.calls) != 1 {
				t.Fatalf(
					"exact assessment owner reads = binding:%d service:%d credentials:%d, want one each",
					len(readers.binding.selections),
					len(readers.service.ids),
					len(readers.credentials.calls),
				)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("sql expectations: %v", err)
			}
		})
	}
}

func TestProfiledExternalAuthorityVerificationLeavesUnavailableOwnerPrepared(t *testing.T) {
	db, mock := newServingSchemaMock(t)
	repository := newServingEpochTestRepository(t, db)
	fixture := newExternalAuthorityAssessmentFixture(t)
	assessor, readers := newExternalAuthorityTestAssessor(t, fixture)
	dependencyFailure := errors.New("Catalog temporarily unavailable")
	readers.catalog.err = dependencyFailure
	service, err := newProfiledExternalAuthorityVerificationService(repository, assessor)
	if err != nil {
		t.Fatalf("newProfiledExternalAuthorityVerificationService(): %v", err)
	}

	expectServingSchema(t, mock, exactServingSchemaFixture(), len(servingTableSchemas)*3)
	expectProfiledReaderTestRows(
		mock,
		fixture.candidate.ID(),
		profiledReaderTestExactRows(fixture.candidate),
		false,
	)

	result, err := service.Verify(t.Context(), fixture.candidate.ID())
	if !errors.Is(err, ErrExternalAuthorityUnavailable) || !errors.Is(err, dependencyFailure) {
		t.Fatalf("Verify() error = %v, want external-unavailable retaining cause", err)
	}
	if result != (ProfiledExternalAuthorityVerificationResult{}) {
		t.Fatalf("Verify() result = %#v, want zero result on retryable owner failure", result)
	}
	if len(readers.epoch.calls) != 0 || len(readers.catalog.minimumReads) != 1 ||
		len(readers.binding.selections) != 0 || len(readers.service.ids) != 0 ||
		len(readers.credentials.calls) != 0 {
		t.Fatalf(
			"owner calls = epoch:%d catalog:%d binding:%d service:%d credentials:%d",
			len(readers.epoch.calls),
			len(readers.catalog.minimumReads),
			len(readers.binding.selections),
			len(readers.service.ids),
			len(readers.credentials.calls),
		)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations (a write or transaction was attempted): %v", err)
	}
}

func TestProfiledExternalAuthorityVerificationReturnsTerminalWithoutOwnerReads(t *testing.T) {
	for _, state := range []ProfiledEpochVerificationState{
		ProfiledEpochVerified,
		ProfiledEpochRejected,
	} {
		t.Run(string(state), func(t *testing.T) {
			db, mock := newServingSchemaMock(t)
			repository := newServingEpochTestRepository(t, db)
			fixture := newExternalAuthorityAssessmentFixture(t)
			assessor, readers := newExternalAuthorityTestAssessor(t, fixture)
			service, err := newProfiledExternalAuthorityVerificationService(repository, assessor)
			if err != nil {
				t.Fatalf("newProfiledExternalAuthorityVerificationService(): %v", err)
			}

			expectServingSchema(t, mock, exactServingSchemaFixture(), len(servingTableSchemas)*3)
			expectProfiledReaderTestRows(
				mock,
				fixture.candidate.ID(),
				profiledVerificationTerminalRows(fixture.candidate, state),
				false,
			)

			result, err := service.Verify(t.Context(), fixture.candidate.ID())
			if err != nil {
				t.Fatalf("Verify(): %v", err)
			}
			assertProfiledVerificationResult(t, result, fixture.candidate, state, false)
			if len(readers.epoch.calls) != 0 || len(readers.catalog.minimumReads) != 0 ||
				len(readers.binding.selections) != 0 || len(readers.service.ids) != 0 ||
				len(readers.credentials.calls) != 0 {
				t.Fatalf(
					"terminal retry read external owners: epoch:%d catalog:%d binding:%d service:%d credentials:%d",
					len(readers.epoch.calls),
					len(readers.catalog.minimumReads),
					len(readers.binding.selections),
					len(readers.service.ids),
					len(readers.credentials.calls),
				)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("sql expectations (a transition was attempted): %v", err)
			}
		})
	}
}

func TestTransitionProfiledPreparedConvergesOnlyOnSameTerminalState(t *testing.T) {
	tests := []struct {
		name        string
		target      ProfiledEpochVerificationState
		stored      ProfiledEpochVerificationState
		wantApplied bool
		wantErr     error
	}{
		{name: "verified onto verified", target: ProfiledEpochVerified, stored: ProfiledEpochVerified},
		{name: "rejected onto rejected", target: ProfiledEpochRejected, stored: ProfiledEpochRejected},
		{
			name:    "verified onto rejected",
			target:  ProfiledEpochVerified,
			stored:  ProfiledEpochRejected,
			wantErr: ErrProfiledEpochStateConflict,
		},
		{
			name:    "rejected onto verified",
			target:  ProfiledEpochRejected,
			stored:  ProfiledEpochVerified,
			wantErr: ErrProfiledEpochStateConflict,
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			db, mock := newServingSchemaMock(t)
			repository := newServingEpochTestRepository(t, db)
			candidate := mustNewProfiledPreparedEpoch(t, newProfiledCanonicalFixture(t))

			expectServingSchema(t, mock, exactServingSchemaFixture(), len(servingTableSchemas)*3)
			expectProfiledVerificationBegin(mock)
			expectProfiledReaderTestRows(
				mock,
				candidate.ID(),
				profiledVerificationTerminalRows(candidate, test.stored),
				true,
			)
			if test.wantErr == nil {
				mock.ExpectCommit()
			} else {
				mock.ExpectRollback()
			}
			expectProfiledVerificationCleanup(mock)

			result, err := repository.transitionProfiledPrepared(t.Context(), candidate, test.target)
			if test.wantErr != nil {
				if !errors.Is(err, test.wantErr) {
					t.Fatalf("transitionProfiledPrepared() error = %v, want %v", err, test.wantErr)
				}
				if result != (ProfiledExternalAuthorityVerificationResult{}) {
					t.Fatalf("transitionProfiledPrepared() result = %#v, want zero on conflict", result)
				}
			} else {
				if err != nil {
					t.Fatalf("transitionProfiledPrepared(): %v", err)
				}
				assertProfiledVerificationResult(t, result, candidate, test.stored, false)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("sql expectations: %v", err)
			}
		})
	}
}

func TestTransitionProfiledPreparedRejectsImmutableDriftUnderLock(t *testing.T) {
	db, mock := newServingSchemaMock(t)
	repository := newServingEpochTestRepository(t, db)
	candidate := mustNewProfiledPreparedEpoch(t, newProfiledCanonicalFixture(t))
	driftFixture := newProfiledCanonicalFixture(t)
	driftFixture.configuration.generationID += "-drift"
	drift := mustNewProfiledPreparedEpoch(t, driftFixture)
	if drift.ID() != candidate.ID() || drift.equalImmutable(candidate) {
		t.Fatal("drift fixture is not a different immutable graph under the same ID")
	}

	expectServingSchema(t, mock, exactServingSchemaFixture(), len(servingTableSchemas)*3)
	expectProfiledVerificationBegin(mock)
	expectProfiledReaderTestRows(mock, candidate.ID(), profiledReaderTestExactRows(drift), true)
	mock.ExpectRollback()
	expectProfiledVerificationCleanup(mock)

	_, err := repository.transitionProfiledPrepared(
		t.Context(),
		candidate,
		ProfiledEpochVerified,
	)
	if !errors.Is(err, ErrIdentityConflict) || errors.Is(err, ErrProfiledEpochStateConflict) {
		t.Fatalf("transitionProfiledPrepared() error = %v, want only identity conflict", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations (an update was attempted after drift): %v", err)
	}
}

func TestTransitionProfiledPreparedRollsBackUpdateFailure(t *testing.T) {
	execFailure := errors.New("verification update failed")
	tests := []struct {
		name         string
		rowsAffected int64
		execErr      error
		wantErr      error
	}{
		{name: "CAS lost", rowsAffected: 0, wantErr: ErrProfiledEpochStateConflict},
		{name: "database error", execErr: execFailure, wantErr: execFailure},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			db, mock := newServingSchemaMock(t)
			repository := newServingEpochTestRepository(t, db)
			candidate := mustNewProfiledPreparedEpoch(t, newProfiledCanonicalFixture(t))

			expectServingSchema(t, mock, exactServingSchemaFixture(), len(servingTableSchemas)*3)
			expectProfiledVerificationBegin(mock)
			expectProfiledReaderTestRows(
				mock,
				candidate.ID(),
				profiledReaderTestExactRows(candidate),
				true,
			)
			expectProfiledVerificationUpdate(
				mock,
				candidate,
				ProfiledEpochVerified,
				test.rowsAffected,
				test.execErr,
			)
			mock.ExpectRollback()
			expectProfiledVerificationCleanup(mock)

			_, err := repository.transitionProfiledPrepared(
				t.Context(),
				candidate,
				ProfiledEpochVerified,
			)
			if !errors.Is(err, test.wantErr) {
				t.Fatalf("transitionProfiledPrepared() error = %v, want %v", err, test.wantErr)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("sql expectations: %v", err)
			}
		})
	}
}

func TestTransitionProfiledPreparedRollsBackCorruptReadback(t *testing.T) {
	db, mock := newServingSchemaMock(t)
	repository := newServingEpochTestRepository(t, db)
	candidate := mustNewProfiledPreparedEpoch(t, newProfiledCanonicalFixture(t))
	corrupt := profiledVerificationTerminalRows(candidate, ProfiledEpochVerified)
	corrupt.owners = corrupt.owners[:len(corrupt.owners)-1]

	expectServingSchema(t, mock, exactServingSchemaFixture(), len(servingTableSchemas)*3)
	expectProfiledVerificationBegin(mock)
	expectProfiledReaderTestRows(
		mock,
		candidate.ID(),
		profiledReaderTestExactRows(candidate),
		true,
	)
	expectProfiledVerificationUpdate(mock, candidate, ProfiledEpochVerified, 1, nil)
	expectProfiledReaderTestRows(mock, candidate.ID(), corrupt, true)
	mock.ExpectRollback()
	expectProfiledVerificationCleanup(mock)

	_, err := repository.transitionProfiledPrepared(
		t.Context(),
		candidate,
		ProfiledEpochVerified,
	)
	if !errors.Is(err, ErrStoredCandidateInvalid) {
		t.Fatalf("transitionProfiledPrepared() error = %v, want stored-candidate-invalid", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestProfiledExternalAuthorityVerificationRejectsInvalidDependenciesAndInput(t *testing.T) {
	if _, err := NewProfiledExternalAuthorityVerificationService(nil, nil, nil, nil, nil); !errors.Is(err, ErrDependencyMissing) {
		t.Fatalf("NewProfiledExternalAuthorityVerificationService(nil) error = %v", err)
	}
	if _, err := newProfiledExternalAuthorityVerificationService(&Repository{}, nil); !errors.Is(err, ErrDependencyMissing) {
		t.Fatalf("newProfiledExternalAuthorityVerificationService(nil assessor) error = %v", err)
	}

	var nilService *ProfiledExternalAuthorityVerificationService
	if _, err := nilService.Verify(context.Background(), "epoch-profile-v2"); !errors.Is(err, ErrDependencyMissing) {
		t.Fatalf("nil service Verify() error = %v", err)
	}

	db, mock := newServingSchemaMock(t)
	repository := newServingEpochTestRepository(t, db)
	fixture := newExternalAuthorityAssessmentFixture(t)
	assessor, _ := newExternalAuthorityTestAssessor(t, fixture)
	service, err := newProfiledExternalAuthorityVerificationService(repository, assessor)
	if err != nil {
		t.Fatalf("newProfiledExternalAuthorityVerificationService(): %v", err)
	}
	if _, err := service.Verify(nil, fixture.candidate.ID()); !errors.Is(err, ErrDependencyMissing) {
		t.Fatalf("Verify(nil context) error = %v", err)
	}
	if _, err := service.Verify(t.Context(), ""); !errors.Is(err, ErrInvalidCandidate) {
		t.Fatalf("Verify(empty ID) error = %v", err)
	}
	if _, err := repository.transitionProfiledPrepared(
		nil,
		fixture.candidate,
		ProfiledEpochVerified,
	); !errors.Is(err, ErrDependencyMissing) {
		t.Fatalf("transitionProfiledPrepared(nil context) error = %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations (invalid input reached storage): %v", err)
	}
}

func profiledVerificationTerminalRows(
	candidate ProfiledPreparedEpoch,
	state ProfiledEpochVerificationState,
) profiledReaderTestRows {
	rows := profiledReaderTestExactRows(candidate)
	rows.base.state = string(state)
	return rows
}

func expectProfiledVerificationBegin(mock sqlmock.Sqlmock) {
	mock.ExpectExec(regexp.QuoteMeta("SET SESSION tidb_txn_mode = 'pessimistic'")).
		WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectBegin()
}

func expectProfiledVerificationCleanup(mock sqlmock.Sqlmock) {
	mock.ExpectExec(regexp.QuoteMeta("SET SESSION tidb_txn_mode = DEFAULT")).
		WillReturnResult(sqlmock.NewResult(0, 0))
}

func expectProfiledVerificationUpdate(
	mock sqlmock.Sqlmock,
	candidate ProfiledPreparedEpoch,
	target ProfiledEpochVerificationState,
	rowsAffected int64,
	execErr error,
) {
	expectation := mock.ExpectExec(regexp.QuoteMeta(strings.TrimSpace(profiledVerificationUpdateTestSQL))).
		WithArgs(
			string(target),
			candidate.ID(),
			candidate.ID(),
			candidate.EpochDigest(),
			candidate.EpochDigest(),
			preparedState,
			preparedState,
		)
	if execErr != nil {
		expectation.WillReturnError(execErr)
		return
	}
	expectation.WillReturnResult(sqlmock.NewResult(0, rowsAffected))
}

func assertProfiledVerificationResult(
	t *testing.T,
	result ProfiledExternalAuthorityVerificationResult,
	candidate ProfiledPreparedEpoch,
	state ProfiledEpochVerificationState,
	applied bool,
) {
	t.Helper()
	if result.EpochID() != candidate.ID() ||
		result.EpochDigest() != candidate.EpochDigest() ||
		result.State() != state || result.Applied() != applied {
		t.Fatalf(
			"verification result = id:%q digest:%q state:%q applied:%t",
			result.EpochID(),
			result.EpochDigest(),
			result.State(),
			result.Applied(),
		)
	}
}