package serving

import (
	"bytes"
	"database/sql"
	"errors"
	"sync"
	"testing"
	"time"

	catalogv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
)

func TestServingEpochProfiledVerificationLifecycleAgainstTiDB(t *testing.T) {
	db, cleanup := openServingEpochTestTiDB(t)
	defer cleanup()
	createServingEpochTestTables(t, db)
	repository := newServingEpochTestRepository(t, db)

	exactFixture := newProfiledVerificationTiDBFixture(t, "epoch-verification-exact")
	rejectedFixture := newProfiledVerificationTiDBFixture(t, "epoch-verification-rejected")
	retryableFixture := newProfiledVerificationTiDBFixture(t, "epoch-verification-retryable")
	seedProfiledWriterTiDBOwnerRows(t, db, exactFixture.candidate, "")
	for _, candidate := range []ProfiledPreparedEpoch{
		exactFixture.candidate,
		rejectedFixture.candidate,
		retryableFixture.candidate,
	} {
		prepareProfiledWriterTiDBCandidate(t, repository, candidate)
		setProfiledVerificationTiDBOldTimestamp(t, db, candidate.ID())
	}

	exactService, exactReaders := newProfiledVerificationTiDBService(t, repository, exactFixture)
	exact, err := exactService.Verify(t.Context(), exactFixture.candidate.ID())
	if err != nil {
		t.Fatalf("Verify(exact): %v", err)
	}
	assertProfiledVerificationTiDBResult(
		t,
		exact,
		exactFixture.candidate,
		ProfiledEpochVerified,
		true,
	)
	assertProfiledVerificationTiDBStored(
		t,
		db,
		exactFixture.candidate,
		verifiedState,
		true,
	)
	exactOwnerReads := len(exactReaders.catalog.minimumReads)
	exactRetry, err := exactService.Verify(t.Context(), exactFixture.candidate.ID())
	if err != nil {
		t.Fatalf("Verify(exact terminal retry): %v", err)
	}
	assertProfiledVerificationTiDBResult(
		t,
		exactRetry,
		exactFixture.candidate,
		ProfiledEpochVerified,
		false,
	)
	if len(exactReaders.catalog.minimumReads) != exactOwnerReads {
		t.Fatal("verified terminal retry reread mutable external owners")
	}

	rejectedService, rejectedReaders := newProfiledVerificationTiDBService(
		t,
		repository,
		rejectedFixture,
	)
	rejectedReaders.catalog.found = false
	rejected, err := rejectedService.Verify(t.Context(), rejectedFixture.candidate.ID())
	if err != nil {
		t.Fatalf("Verify(definitive rejection): %v", err)
	}
	assertProfiledVerificationTiDBResult(
		t,
		rejected,
		rejectedFixture.candidate,
		ProfiledEpochRejected,
		true,
	)
	assertProfiledVerificationTiDBStored(
		t,
		db,
		rejectedFixture.candidate,
		rejectedState,
		true,
	)
	rejectedOwnerReads := len(rejectedReaders.catalog.minimumReads)
	rejectedRetry, err := rejectedService.Verify(t.Context(), rejectedFixture.candidate.ID())
	if err != nil {
		t.Fatalf("Verify(rejected terminal retry): %v", err)
	}
	assertProfiledVerificationTiDBResult(
		t,
		rejectedRetry,
		rejectedFixture.candidate,
		ProfiledEpochRejected,
		false,
	)
	if len(rejectedReaders.catalog.minimumReads) != rejectedOwnerReads {
		t.Fatal("rejected terminal retry reread mutable external owners")
	}

	retryableService, retryableReaders := newProfiledVerificationTiDBService(
		t,
		repository,
		retryableFixture,
	)
	ownerUnavailable := errors.New("Catalog temporarily unavailable")
	retryableReaders.catalog.err = ownerUnavailable
	retryable, err := retryableService.Verify(t.Context(), retryableFixture.candidate.ID())
	if !errors.Is(err, ErrExternalAuthorityUnavailable) || !errors.Is(err, ownerUnavailable) {
		t.Fatalf(
			"Verify(retryable owner failure) = %#v, %v, want retained external-authority error",
			retryable,
			err,
		)
	}
	if retryable != (ProfiledExternalAuthorityVerificationResult{}) {
		t.Fatalf("Verify(retryable owner failure) result = %#v, want zero", retryable)
	}
	assertProfiledVerificationTiDBStored(
		t,
		db,
		retryableFixture.candidate,
		preparedState,
		false,
	)
}

func TestServingEpochProfiledVerificationConcurrencyAgainstTiDB(t *testing.T) {
	t.Run("same target converges", func(t *testing.T) {
		const workers = 6

		db, cleanup := openServingEpochTestTiDB(t)
		defer cleanup()
		createServingEpochTestTables(t, db)
		db.SetMaxOpenConns(workers * 3)
		repository := newServingEpochTestRepository(t, db)
		fixture := newProfiledVerificationTiDBFixture(t, "epoch-verification-concurrent-exact")
		seedProfiledWriterTiDBOwnerRows(t, db, fixture.candidate, "")
		prepareProfiledWriterTiDBCandidate(t, repository, fixture.candidate)
		setProfiledVerificationTiDBOldTimestamp(t, db, fixture.candidate.ID())

		services := make([]*ProfiledExternalAuthorityVerificationService, workers)
		for index := range services {
			services[index], _ = newProfiledVerificationTiDBService(t, repository, fixture)
		}
		start := make(chan struct{})
		results := make(chan profiledVerificationTiDBCall, workers)
		var group sync.WaitGroup
		for index := range services {
			group.Add(1)
			go func(service *ProfiledExternalAuthorityVerificationService) {
				defer group.Done()
				<-start
				result, err := service.Verify(t.Context(), fixture.candidate.ID())
				results <- profiledVerificationTiDBCall{result: result, err: err}
			}(services[index])
		}
		close(start)
		group.Wait()
		close(results)

		applied := 0
		converged := 0
		for call := range results {
			if call.err != nil {
				t.Errorf("concurrent Verify(exact): %v", call.err)
				continue
			}
			if call.result.State() != ProfiledEpochVerified ||
				call.result.EpochID() != fixture.candidate.ID() ||
				call.result.EpochDigest() != fixture.candidate.EpochDigest() {
				t.Errorf("concurrent Verify(exact) result = %#v", call.result)
				continue
			}
			if call.result.Applied() {
				applied++
			} else {
				converged++
			}
		}
		if applied != 1 || converged != workers-1 {
			t.Fatalf(
				"concurrent exact outcomes = %d applied/%d converged, want 1/%d",
				applied,
				converged,
				workers-1,
			)
		}
		assertProfiledVerificationTiDBStored(
			t,
			db,
			fixture.candidate,
			verifiedState,
			true,
		)
	})

	t.Run("opposite assessed targets cannot overwrite the winner", func(t *testing.T) {
		db, cleanup := openServingEpochTestTiDB(t)
		defer cleanup()
		createServingEpochTestTables(t, db)
		repository := newServingEpochTestRepository(t, db)
		fixture := newProfiledVerificationTiDBFixture(t, "epoch-verification-opposite-race")
		seedProfiledWriterTiDBOwnerRows(t, db, fixture.candidate, "")
		prepareProfiledWriterTiDBCandidate(t, repository, fixture.candidate)
		setProfiledVerificationTiDBOldTimestamp(t, db, fixture.candidate.ID())

		verifiedService, verifiedReaders := newProfiledVerificationTiDBService(t, repository, fixture)
		rejectedService, rejectedReaders := newProfiledVerificationTiDBService(t, repository, fixture)
		arrived := make(chan struct{}, 2)
		release := make(chan struct{})
		verifiedService.assessor.reconstruct = profiledVerificationTiDBBarrierReconstructor(
			arrived,
			release,
			false,
		)
		rejectedService.assessor.reconstruct = profiledVerificationTiDBBarrierReconstructor(
			arrived,
			release,
			true,
		)

		services := []*ProfiledExternalAuthorityVerificationService{verifiedService, rejectedService}
		results := make(chan profiledVerificationTiDBCall, len(services))
		var group sync.WaitGroup
		for _, service := range services {
			group.Add(1)
			go func(service *ProfiledExternalAuthorityVerificationService) {
				defer group.Done()
				result, err := service.Verify(t.Context(), fixture.candidate.ID())
				results <- profiledVerificationTiDBCall{result: result, err: err}
			}(service)
		}

		timer := time.NewTimer(10 * time.Second)
		defer timer.Stop()
		for count := 0; count < len(services); count++ {
			select {
			case <-arrived:
			case <-timer.C:
				close(release)
				group.Wait()
				t.Fatal("opposite assessments did not both reach reconstruction before the CAS")
			}
		}
		close(release)
		group.Wait()
		close(results)

		var winner ProfiledExternalAuthorityVerificationResult
		successes := 0
		conflicts := 0
		for call := range results {
			switch {
			case call.err == nil:
				successes++
				winner = call.result
				if !call.result.Applied() {
					t.Errorf("opposite-target winner was not the applied CAS: %#v", call.result)
				}
			case errors.Is(call.err, ErrProfiledEpochStateConflict):
				conflicts++
			default:
				t.Errorf("opposite-target Verify() error = %v", call.err)
			}
		}
		if successes != 1 || conflicts != 1 {
			t.Fatalf(
				"opposite-target outcomes = %d winner/%d conflict, want 1/1",
				successes,
				conflicts,
			)
		}
		if len(verifiedReaders.catalog.minimumReads) != 1 ||
			len(rejectedReaders.catalog.minimumReads) != 1 {
			t.Fatalf(
				"fresh Catalog reads = verified:%d rejected:%d, want both 1 before CAS",
				len(verifiedReaders.catalog.minimumReads),
				len(rejectedReaders.catalog.minimumReads),
			)
		}
		assertProfiledVerificationTiDBStored(
			t,
			db,
			fixture.candidate,
			string(winner.State()),
			true,
		)
	})
}

type profiledVerificationTiDBCall struct {
	result ProfiledExternalAuthorityVerificationResult
	err    error
}

func newProfiledVerificationTiDBFixture(
	t *testing.T,
	epochID string,
) externalAuthorityAssessmentFixture {
	t.Helper()
	fixture := newExternalAuthorityAssessmentFixture(t)
	if len(fixture.bindingOwners) != 1 ||
		len(fixture.bindingOwners[0]) != len(fixture.candidate.bindings) {
		t.Fatalf("external-authority fixture binding groups = %#v", fixture.bindingOwners)
	}
	proofs := make([]BindingOwnerProof, len(fixture.candidate.bindings))
	for index := range proofs {
		proof, err := NewBindingOwnerProof(
			fixture.candidate.bindings[index].owner.packGenerationID,
			fixture.bindingOwners[0][index],
		)
		if err != nil {
			t.Fatalf("NewBindingOwnerProof(%d): %v", index, err)
		}
		proofs[index] = proof
	}
	candidate, err := newProfiledPreparedEpoch(
		ConnectorDigestProfileV2ID,
		epochID,
		fixture.view,
		fixture.configuration,
		proofs,
		fixture.credentials,
	)
	if err != nil {
		t.Fatalf("newProfiledPreparedEpoch(%q): %v", epochID, err)
	}
	fixture.candidate = candidate
	return fixture
}

func newProfiledVerificationTiDBService(
	t *testing.T,
	repository *Repository,
	fixture externalAuthorityAssessmentFixture,
) (*ProfiledExternalAuthorityVerificationService, externalAuthorityTestReaders) {
	t.Helper()
	assessor, readers := newExternalAuthorityTestAssessor(t, fixture)
	assessor.epochs = repository
	service, err := newProfiledExternalAuthorityVerificationService(repository, assessor)
	if err != nil {
		t.Fatalf("newProfiledExternalAuthorityVerificationService(): %v", err)
	}
	return service, readers
}

func profiledVerificationTiDBBarrierReconstructor(
	arrived chan<- struct{},
	release <-chan struct{},
	reject bool,
) profiledPreparedEpochReconstructor {
	return func(
		profileID ConnectorDigestProfileID,
		epochID string,
		view catalogv1.IntegrationBindingView,
		configuration profiledServiceConfigurationInput,
		bindingOwners []BindingOwnerProof,
		credentials []profiledCredentialGenerationInput,
	) (ProfiledPreparedEpoch, error) {
		arrived <- struct{}{}
		<-release
		if reject {
			return ProfiledPreparedEpoch{}, nil
		}
		return reconstructProfiledPreparedEpochForRead(
			profileID,
			epochID,
			view,
			configuration,
			bindingOwners,
			credentials,
		)
	}
}

func assertProfiledVerificationTiDBResult(
	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(),
		)
	}
}

func assertProfiledVerificationTiDBStored(
	t *testing.T,
	db *sql.DB,
	candidate ProfiledPreparedEpoch,
	wantState string,
	wantTimestampAdvanced bool,
) {
	t.Helper()
	stored, found, err := loadProfiledPreActivationEpoch(
		t.Context(),
		db,
		candidate.ID(),
		false,
	)
	if err != nil || !found || stored.state != wantState ||
		!stored.candidate.equalImmutable(candidate) {
		t.Fatalf(
			"stored verification epoch = found:%t state:%q immutable-equal:%t err:%v",
			found,
			stored.state,
			stored.candidate.equalImmutable(candidate),
			err,
		)
	}
	if !bytes.Equal(
		stored.candidate.serviceConfiguration.ownerReceipt,
		candidate.serviceConfiguration.ownerReceipt,
	) {
		t.Fatal("verification changed the opaque ServiceConfiguration receipt bytes")
	}
	for index := range candidate.bindings {
		if !bytes.Equal(
			stored.candidate.bindings[index].owner.definitionOwnerReceipt,
			candidate.bindings[index].owner.definitionOwnerReceipt,
		) {
			t.Fatalf("verification changed binding-owner receipt %d", index)
		}
	}
	for index := range candidate.credentials {
		if !bytes.Equal(
			stored.candidate.credentials[index].ownerReceipt,
			candidate.credentials[index].ownerReceipt,
		) {
			t.Fatalf("verification changed credential-generation receipt %d", index)
		}
	}
	assertProfiledWriterTiDBGraphCounts(t, db, candidate.ID(), profiledWriterTiDBGraphCounts{
		epochs:      1,
		bindings:    len(candidate.bindings),
		headers:     1,
		owners:      len(candidate.bindings),
		credentials: len(candidate.credentials),
	})

	var state string
	var activeScope sql.NullString
	var modified time.Time
	if err := db.QueryRowContext(t.Context(), `
SELECT State, ActiveScopeDigest, LastModifiedDate
  FROM _System_ConnectorServingEpoch
 WHERE Id = ? AND BINARY Id = BINARY ?`, candidate.ID(), candidate.ID()).Scan(
		&state,
		&activeScope,
		&modified,
	); err != nil {
		t.Fatalf("read verification lifecycle row: %v", err)
	}
	if state != wantState || activeScope.Valid {
		t.Fatalf("verification lifecycle = state:%q active:%q/%t", state, activeScope.String, activeScope.Valid)
	}
	if wantTimestampAdvanced && modified.Year() == 2000 {
		t.Fatalf("verification did not advance LastModifiedDate: %s", modified)
	}
	if !wantTimestampAdvanced && modified.Year() != 2000 {
		t.Fatalf("non-transition changed LastModifiedDate: %s", modified)
	}
}

func setProfiledVerificationTiDBOldTimestamp(t *testing.T, db *sql.DB, epochID string) {
	t.Helper()
	result, err := db.ExecContext(t.Context(), `
UPDATE _System_ConnectorServingEpoch
   SET LastModifiedDate = '2000-01-01 00:00:00.000000'
 WHERE Id = ? AND BINARY Id = BINARY ?`, epochID, epochID)
	if err != nil {
		t.Fatalf("set old verification timestamp for %q: %v", epochID, err)
	}
	rows, err := result.RowsAffected()
	if err != nil || rows != 1 {
		t.Fatalf("set old verification timestamp rows = %d, %v", rows, err)
	}
}