package serving

import (
	"context"
	"database/sql"
	"errors"
	"strings"
	"sync"
	"testing"
	"time"

	connectorbindingownerv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/connectorbindingownerv1"
	applicationresultv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/applicationresultv1"
	commandv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commandv1"
	commonv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commonv1"
	flowcommandv3 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv3"
)

func TestExecutionRepositorySchemaReadinessAgainstTiDB(t *testing.T) {
	fixture := newExecutionTiDBFixture(t, "schema")
	if err := fixture.repository.VerifySchema(t.Context()); err != nil {
		t.Fatalf("VerifySchema(): %v", err)
	}
	if _, err := fixture.db.ExecContext(t.Context(), `ALTER TABLE _System_ConnectorV2Execution
DROP INDEX idx_connector_v2_execution_claim`); err != nil {
		t.Fatalf("damage execution schema: %v", err)
	}
	if err := fixture.repository.VerifySchema(t.Context()); !errors.Is(err, ErrSchemaUnavailable) || !errors.Is(err, ErrExecutionSchemaUnavailable) {
		t.Fatalf("VerifySchema(drifted) = %v, want execution schema unavailable", err)
	}
}

func TestExecutionRepositoryAdmissionLocksCompleteAuthorityBeforeWriteAgainstTiDB(t *testing.T) {
	fixture := newExecutionTiDBFixture(t, "admission")

	wrongFenceInput := fixture.input
	wrongFenceInput.AccountServingFenceToken++
	wrongFenceInput.ExecutionID = "execution-wrong-fence"
	wrongFence, err := NewConformanceExecutionCandidate(wrongFenceInput)
	if err != nil {
		t.Fatalf("NewConformanceExecutionCandidate(wrong fence): %v", err)
	}
	if _, _, err := fixture.repository.AdmitConformance(t.Context(), wrongFence); !errors.Is(err, ErrExecutionAuthorityConflict) {
		t.Fatalf("AdmitConformance(wrong fence) = %v, want authority conflict", err)
	}
	assertExecutionTiDBRowCount(t, fixture.db, executionTable, 0)
	assertExecutionTiDBRowCount(t, fixture.db, executionAttemptTable, 0)

	stored, duplicate, err := fixture.repository.AdmitConformance(t.Context(), fixture.candidate)
	if err != nil || duplicate || !stored.equalImmutable(fixture.candidate.execution) {
		t.Fatalf("AdmitConformance() = equal:%t duplicate:%t err:%v",
			stored.equalImmutable(fixture.candidate.execution), duplicate, err)
	}
	replayed, duplicate, err := fixture.repository.AdmitConformance(t.Context(), fixture.candidate)
	if err != nil || !duplicate || !replayed.equalImmutable(stored) {
		t.Fatalf("AdmitConformance(exact replay) = equal:%t duplicate:%t err:%v",
			replayed.equalImmutable(stored), duplicate, err)
	}

	conflictInput := fixture.input
	conflictInput.CanonicalInputs = []byte(`{"record":{"Name":"Grace"}}`)
	conflict, err := NewConformanceExecutionCandidate(conflictInput)
	if err != nil {
		t.Fatalf("NewConformanceExecutionCandidate(conflict): %v", err)
	}
	if _, _, err := fixture.repository.AdmitConformance(t.Context(), conflict); !errors.Is(err, ErrExecutionIdentityConflict) {
		t.Fatalf("AdmitConformance(conflict) = %v, want identity conflict", err)
	}
	assertExecutionTiDBRowCount(t, fixture.db, executionTable, 1)
}

func TestExecutionRepositoryConcurrentClaimsHaveOneFencedWinnerAgainstTiDB(t *testing.T) {
	const workers = 6
	fixture := newExecutionTiDBFixture(t, "claim-race")
	admitExecutionTiDBFixture(t, fixture)
	fixture.db.SetMaxOpenConns(workers * 3)

	type claimResult struct {
		claim ExecutionClaim
		err   error
	}
	start := make(chan struct{})
	results := make(chan claimResult, workers)
	var group sync.WaitGroup
	for index := 0; index < workers; index++ {
		group.Add(1)
		go func(index int) {
			defer group.Done()
			<-start
			claim, err := fixture.repository.ClaimConformance(
				t.Context(),
				fixture.input.ExecutionID,
				"claim-race-worker-"+string(rune('a'+index)),
				5*time.Second,
			)
			results <- claimResult{claim: claim, err: err}
		}(index)
	}
	close(start)
	group.Wait()
	close(results)

	winners := 0
	rejected := 0
	for result := range results {
		switch {
		case result.err == nil:
			winners++
			if result.claim.fenceToken != 1 || result.claim.catalogRevision != executionTiDBCatalogRevision {
				t.Errorf("winning claim = fence:%d revision:%d", result.claim.fenceToken, result.claim.catalogRevision)
			}
		case errors.Is(result.err, ErrExecutionNotClaimable):
			rejected++
		default:
			t.Errorf("concurrent ClaimConformance(): %v", result.err)
		}
	}
	if winners != 1 || rejected != workers-1 {
		t.Fatalf("claim race = %d winners/%d rejected, want 1/%d", winners, rejected, workers-1)
	}
	assertExecutionTiDBRowCount(t, fixture.db, executionAttemptTable, 1)
}

func TestExecutionRepositoryRuntimeResultFinalizesAfterLeaseReplacementAgainstTiDB(t *testing.T) {
	fixture := newExecutionTiDBFixture(t, "lease")
	execution := admitExecutionTiDBFixture(t, fixture)
	result := executionTiDBApplicationResult(t, execution)

	first, err := fixture.repository.ClaimConformance(
		t.Context(), execution.id, "lease-worker-first", 25*time.Millisecond,
	)
	if err != nil {
		t.Fatalf("ClaimConformance(first): %v", err)
	}

	time.Sleep(60 * time.Millisecond)
	second, err := fixture.repository.ClaimConformance(
		t.Context(), execution.id, "lease-worker-second", time.Second,
	)
	if err != nil {
		t.Fatalf("ClaimConformance(after expiry): %v", err)
	}
	if second.fenceToken != first.fenceToken+1 ||
		second.execution.runtimeCommandID != first.execution.runtimeCommandID ||
		second.execution.executionRefDigest != first.execution.executionRefDigest {
		t.Fatalf("replacement claim changed identity or fence: first=%#v second=%#v", first, second)
	}
	completed, err := fixture.repository.completeFromApplicationResult(t.Context(), first.execution, result)
	if err != nil {
		t.Fatalf("completeFromApplicationResult(stale claim observation): %v", err)
	}
	if completed.state != executionStateCompleted {
		t.Fatalf("completed state = %q, want completed", completed.state)
	}
	if digest, receipt, ok := completed.ResultCorrelation(); !ok || digest != result.ResultDigest || receipt != result.RuntimeCommitReceipt {
		t.Fatalf("completed correlation = %q/%q/%t", digest, receipt, ok)
	}
	assertExecutionTiDBRowCount(t, fixture.db, executionAttemptTable, 2)
}

func TestConformanceExecutionReconcilerFinalizesLostResponseWhileHaltingWithoutResubmissionAgainstTiDB(t *testing.T) {
	fixture := newExecutionTiDBFixture(t, "lost-response")
	execution := admitExecutionTiDBFixture(t, fixture)
	runtime := &executionTiDBCrashRuntime{loseFirstResponse: true}
	reconciler, err := NewConformanceExecutionReconciler(fixture.repository, runtime)
	if err != nil {
		t.Fatalf("NewConformanceExecutionReconciler(): %v", err)
	}

	firstContext, cancelFirst := context.WithTimeout(context.Background(), 5*time.Second)
	_, firstErr := reconciler.Reconcile(
		firstContext,
		execution.id,
		"lost-response-worker-first",
		25*time.Millisecond,
	)
	cancelFirst()
	if firstErr == nil || !strings.Contains(firstErr.Error(), "response lost after Runtime commit") {
		t.Fatalf("Reconcile(first lost response) = %v", firstErr)
	}

	haltExecutionTiDBGate(t, fixture)
	secondContext, cancelSecond := context.WithTimeout(context.Background(), 5*time.Second)
	completed, err := reconciler.Reconcile(
		secondContext,
		execution.id,
		"lost-response-worker-second",
		time.Second,
	)
	cancelSecond()
	if err != nil {
		t.Fatalf("Reconcile(exact retry): %v", err)
	}
	if completed.state != executionStateCompleted {
		t.Fatalf("Reconcile(exact retry) state = %q", completed.state)
	}

	runtime.mu.Lock()
	requests := append([]commandv1.Request(nil), runtime.requests...)
	mutationCount := runtime.mutationCount
	resultReads := runtime.resultReads
	runtime.mu.Unlock()
	if len(requests) != 1 {
		t.Fatalf("Runtime submissions = %d, want exactly 1", len(requests))
	}
	if mutationCount != 1 {
		t.Fatalf("Runtime business mutations = %d, want exactly 1", mutationCount)
	}
	if resultReads != 2 {
		t.Fatalf("Runtime result reads = %d, want pre-submit plus halted recovery", resultReads)
	}
	if requests[0].CommandID != execution.runtimeCommandID ||
		requests[0].PayloadFormatVersion != flowcommandv3.PayloadFormatVersion {
		t.Fatalf("Runtime request identity/version = %q/%d", requests[0].CommandID, requests[0].PayloadFormatVersion)
	}
	assertExecutionTiDBRowCount(t, fixture.db, executionAttemptTable, 1)
}

func TestConformanceExecutionReconcilerDoesNotClaimOrSubmitPendingWorkWhileHaltingAgainstTiDB(t *testing.T) {
	fixture := newExecutionTiDBFixture(t, "halted-pending")
	execution := admitExecutionTiDBFixture(t, fixture)
	runtime := &executionTiDBCrashRuntime{knownPending: true}
	reconciler, err := NewConformanceExecutionReconciler(fixture.repository, runtime)
	if err != nil {
		t.Fatalf("NewConformanceExecutionReconciler(): %v", err)
	}
	haltExecutionTiDBGate(t, fixture)

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	_, err = reconciler.Reconcile(ctx, execution.id, "halted-pending-worker", time.Second)
	cancel()
	if !errors.Is(err, ErrExecutionAuthorityConflict) {
		t.Fatalf("Reconcile(pending while halting) = %v, want authority conflict", err)
	}

	runtime.mu.Lock()
	requestCount := len(runtime.requests)
	resultReads := runtime.resultReads
	runtime.mu.Unlock()
	if requestCount != 0 || resultReads != 2 {
		t.Fatalf("Runtime calls = %d submissions/%d result reads, want 0/2", requestCount, resultReads)
	}
	stored, found, loadErr := fixture.repository.LoadExecution(t.Context(), execution.id)
	if loadErr != nil || !found || stored.state != executionStateReady || stored.claimFenceToken != 0 {
		t.Fatalf("halted pending execution = found:%t state:%q fence:%d err:%v",
			found, stored.state, stored.claimFenceToken, loadErr)
	}
	assertExecutionTiDBRowCount(t, fixture.db, executionAttemptTable, 0)
}

func TestConformanceExecutionReconcilerPreHaltClaimSubmitsOnceWhileHaltingAgainstTiDB(t *testing.T) {
	fixture := newExecutionTiDBFixture(t, "claim-before-halt")
	execution := admitExecutionTiDBFixture(t, fixture)
	runtime := &executionTiDBCrashRuntime{}
	reconciler, err := NewConformanceExecutionReconciler(fixture.repository, runtime)
	if err != nil {
		t.Fatalf("NewConformanceExecutionReconciler(): %v", err)
	}

	claim, err := fixture.repository.ClaimConformance(
		t.Context(),
		execution.id,
		"claim-before-halt-worker",
		time.Second,
	)
	if err != nil {
		t.Fatalf("ClaimConformance(before halt): %v", err)
	}
	haltExecutionTiDBGate(t, fixture)

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	completed, err := reconciler.submitClaimedExecution(ctx, claim)
	cancel()
	if err != nil {
		t.Fatalf("submitClaimedExecution(after halt): %v", err)
	}
	if completed.state != executionStateCompleted {
		t.Fatalf("submitClaimedExecution(after halt) state = %q, want completed", completed.state)
	}
	if _, err := fixture.repository.ClaimConformance(
		t.Context(),
		execution.id,
		"claim-after-halt-worker",
		time.Second,
	); !errors.Is(err, ErrExecutionAuthorityConflict) {
		t.Fatalf("ClaimConformance(after halt) = %v, want authority conflict", err)
	}

	runtime.mu.Lock()
	requestCount := len(runtime.requests)
	resultReads := runtime.resultReads
	mutationCount := runtime.mutationCount
	runtime.mu.Unlock()
	if requestCount != 1 || resultReads != 1 || mutationCount != 1 {
		t.Fatalf(
			"Runtime calls = %d submissions/%d result reads/%d mutations, want 1/1/1",
			requestCount,
			resultReads,
			mutationCount,
		)
	}
	assertExecutionTiDBRowCount(t, fixture.db, executionAttemptTable, 1)
}

const executionTiDBCatalogRevision = uint64(41)

type executionTiDBFixture struct {
	db         *sql.DB
	repository *ExecutionRepository
	input      ConformanceExecutionInput
	candidate  ConformanceExecutionCandidate
}

func newExecutionTiDBFixture(t *testing.T, suffix string) executionTiDBFixture {
	t.Helper()
	db, cleanup := openServingEpochTestTiDB(t)
	t.Cleanup(cleanup)
	createServingEpochTestTables(t, db)
	for _, statement := range []string{
		`ALTER TABLE _System_PackGeneration
 ADD COLUMN PackInstallId VARCHAR(255) NULL,
 ADD COLUMN PackName VARCHAR(255) NULL,
 ADD COLUMN AppliedPayloadHash VARCHAR(128) NULL`,
		`ALTER TABLE _System_PackGenerationConnectorBindingDefinitionSet
 ADD COLUMN Protocol VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NULL,
 ADD COLUMN DefinitionCount DECIMAL(20,0) NULL,
 ADD COLUMN DefinitionSetDigest CHAR(71) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NULL`,
		accountServingGateTestDDL,
		executionTiDBExecutionDDL,
		executionTiDBAttemptDDL,
	} {
		if _, err := db.ExecContext(t.Context(), statement); err != nil {
			t.Fatalf("create execution fixture schema: %v", err)
		}
	}

	digest := func(character string) string { return "sha256:" + strings.Repeat(character, 64) }
	const (
		packGenerationID  = "execution-pack-generation"
		packInstallID     = "execution-pack-installation"
		ownerPack         = "execution-pack"
		accountID         = "execution-account"
		gateID            = "execution-gate"
		epochID           = "execution-epoch"
		bindingID         = "execution-binding"
		definitionRef     = "Execution_Flow"
		definitionVersion = "flow-version-7"
	)
	scopeDigest := strings.Repeat("1", 64)
	epochDigest := digest("2")
	packPayloadDigest := "nexus-pack-sha256-v1:" + strings.Repeat("3", 64)
	identity := connectorbindingownerv1.Identity{
		Kind:               "CONNECTOR_INBOUND_ROUTE",
		OwnerPack:          ownerPack,
		PackInstallationID: packInstallID,
		PackPayloadDigest:  packPayloadDigest,
		BusinessKey:        "execution-pack/contact.created",
		DefinitionVersion:  7,
	}
	envelope, err := connectorbindingownerv1.NewDefinitionEnvelope(
		connectorbindingownerv1.AuthoringInput{
			Identity:            identity,
			CanonicalDefinition: []byte(`{"flow":"Execution_Flow"}`),
		},
	)
	if err != nil {
		t.Fatalf("NewDefinitionEnvelope(): %v", err)
	}

	execExecutionTiDBSQL(t, db, `INSERT INTO _System_PackGeneration
(Id, PackInstallId, PackName, AppliedPayloadHash) VALUES (?, ?, ?, ?)`,
		packGenerationID, packInstallID, ownerPack, packPayloadDigest)
	execExecutionTiDBSQL(t, db, `INSERT INTO _System_PackGenerationConnectorBindingDefinitionSet
(PackGenerationId, Protocol, DefinitionCount, DefinitionSetDigest) VALUES (?, ?, ?, ?)`,
		packGenerationID,
		connectorbindingownerv1.DefinitionSetProtocolV1,
		uint64(1),
		digest("4"),
	)
	execExecutionTiDBSQL(t, db, `INSERT INTO _System_ConnectorServingEpoch
(Id, EpochDigest, ServingScopeDigest, ActiveScopeDigest, State,
 OrgId, AppInstallationId, AuthorizedWorkloadId,
 CatalogRevision, SchemaGeneration, EngineGeneration,
 SourceSnapshotDigest, BindingViewDigest,
 ReceiptProtocolVersion, ReceiptCapabilityVersion, SecurityEpoch,
 L2ConfigurationGenerationRef, L2ConfigurationDigest,
 L2VerificationReceipt, L2VerificationReceiptDigest)
VALUES (?, ?, ?, ?, 'active', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		epochID, epochDigest, scopeDigest, scopeDigest,
		"execution-org", "execution-app-installation", "execution-workload",
		executionTiDBCatalogRevision, uint64(7), uint64(8),
		digest("5"), digest("6"),
		"1.0", "catalog.consumer.integration-binding-view.v1", uint64(9),
		"execution-service-configuration", digest("7"),
		[]byte("execution-service-configuration-receipt"), digest("8"),
	)
	execExecutionTiDBSQL(t, db, `INSERT INTO _System_ConnectorServingEpochReproducibility
(ServingEpochId, ConnectorDigestProfileId,
 ServiceConfigurationGenerationProtocolVersion,
 ServiceConfigurationReceiptProtocolVersion)
VALUES (?, ?, ?, ?)`,
		epochID,
		ConnectorDigestProfileV2ID,
		"nexuscrm.connector-service-configuration-generation/v1",
		"nexuscrm.connector-service-configuration-owner-receipt/v1",
	)
	execExecutionTiDBSQL(t, db, `INSERT INTO _System_ConnectorServingEpochBinding
(Id, ServingEpochId, BindingOrdinal, BindingIdentityDigest,
 Kind, OwnerPack, PackInstallationId, PackPayloadDigest,
 BusinessKey, DefinitionVersion, CredentialRefs,
 RuntimeCommandKind, RuntimePayloadFormatVersion,
 RuntimeDefinitionRef, RuntimeDefinitionVersion)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		bindingID, epochID, uint64(0), strings.Repeat("9", 64),
		identity.Kind, ownerPack, packInstallID, packPayloadDigest,
		identity.BusinessKey, identity.DefinitionVersion, []byte("[]"),
		flowcommandv3.CommandKind, flowcommandv3.PayloadFormatVersion,
		definitionRef, definitionVersion,
	)
	execExecutionTiDBSQL(t, db, `INSERT INTO _System_ConnectorServingEpochBindingOwner
(ServingEpochBindingId, PackGenerationId,
 DefinitionEnvelopeVersion, DefinitionReceiptVersion,
 CanonicalDefinitionDigest, DefinitionOwnerReceipt, DefinitionOwnerReceiptDigest)
VALUES (?, ?, ?, ?, ?, ?, ?)`,
		bindingID, packGenerationID,
		envelope.EnvelopeVersion, envelope.ReceiptVersion,
		envelope.CanonicalDefinitionDigest, envelope.OwnerReceipt, envelope.OwnerReceiptDigest,
	)
	seedAccountServingGateTestRow(
		t, db, gateID, scopeDigest, accountID, 7, epochID, epochDigest,
	)

	input := ConformanceExecutionInput{
		ExecutionID:              "execution-" + suffix,
		ServingScopeDigest:       scopeDigest,
		ConnectorAccountID:       accountID,
		AccountServingGateID:     gateID,
		AccountServingFenceToken: 7,
		ServingEpochID:           epochID,
		ServingEpochDigest:       epochDigest,
		ServingEpochBindingID:    bindingID,
		PackGenerationID:         packGenerationID,
		DefinitionRef:            definitionRef,
		DefinitionVersion:        definitionVersion,
		CanonicalInputs:          []byte(`{"record":{"Name":"Ada"}}`),
	}
	candidate, err := NewConformanceExecutionCandidate(input)
	if err != nil {
		t.Fatalf("NewConformanceExecutionCandidate(): %v", err)
	}
	repository, err := NewExecutionRepository(db)
	if err != nil {
		t.Fatalf("NewExecutionRepository(): %v", err)
	}
	return executionTiDBFixture{db: db, repository: repository, input: input, candidate: candidate}
}

func admitExecutionTiDBFixture(t *testing.T, fixture executionTiDBFixture) Execution {
	t.Helper()
	execution, duplicate, err := fixture.repository.AdmitConformance(t.Context(), fixture.candidate)
	if err != nil || duplicate {
		t.Fatalf("AdmitConformance() = duplicate:%t err:%v", duplicate, err)
	}
	return execution
}

func executionTiDBApplicationResult(
	t *testing.T,
	execution Execution,
) applicationresultv1.ApplicationResult {
	t.Helper()
	resultDigest, err := applicationresultv1.ComputeResultDigest([]byte(`{"saved":true}`))
	if err != nil {
		t.Fatalf("ComputeResultDigest(): %v", err)
	}
	result := applicationresultv1.ApplicationResult{
		CommandID:          execution.runtimeCommandID,
		ExecutionRefDigest: execution.executionRefDigest,
		Outcome:            applicationresultv1.OutcomeApplied,
		ResultDigest:       resultDigest,
	}
	result.RuntimeCommitReceipt, err = applicationresultv1.ComputeRuntimeCommitReceipt(result)
	if err != nil {
		t.Fatalf("ComputeRuntimeCommitReceipt(): %v", err)
	}
	return result
}

type executionTiDBCrashRuntime struct {
	mu                sync.Mutex
	requests          []commandv1.Request
	result            *applicationresultv1.ApplicationResult
	knownPending      bool
	loseFirstResponse bool
	resultReads       int
	mutationCount     int
}

func (runtime *executionTiDBCrashRuntime) AcceptCommand(
	_ context.Context,
	request commandv1.Request,
) (commandv1.Response, error) {
	runtime.mu.Lock()
	defer runtime.mu.Unlock()
	request.Payload = append([]byte(nil), request.Payload...)
	if request.MinCatalogRevision != nil {
		revision := *request.MinCatalogRevision
		request.MinCatalogRevision = &revision
	}
	runtime.requests = append(runtime.requests, request)
	if runtime.result == nil {
		payload, err := flowcommandv3.Decode(request.Payload)
		if err != nil {
			return commandv1.Response{}, err
		}
		refDigest, err := applicationresultv1.ExecutionRefDigest(payload.ExecutionRef)
		if err != nil {
			return commandv1.Response{}, err
		}
		resultDigest, err := applicationresultv1.ComputeResultDigest([]byte(`{"saved":true}`))
		if err != nil {
			return commandv1.Response{}, err
		}
		result := applicationresultv1.ApplicationResult{
			CommandID:          request.CommandID,
			ExecutionRefDigest: refDigest,
			Outcome:            applicationresultv1.OutcomeApplied,
			ResultDigest:       resultDigest,
		}
		result.RuntimeCommitReceipt, err = applicationresultv1.ComputeRuntimeCommitReceipt(result)
		if err != nil {
			return commandv1.Response{}, err
		}
		runtime.result = &result
		runtime.mutationCount++
		if runtime.loseFirstResponse {
			return commandv1.Response{}, errors.New("response lost after Runtime commit")
		}
		return commandv1.Response{
			CommandID: request.CommandID,
			Receipts: commonv1.Receipts{
				ProtocolVersion:   "1.0",
				CatalogRevision:   executionTiDBCatalogRevision,
				CapabilityVersion: commandv1.CapabilityName,
			},
		}, nil
	}
	return commandv1.Response{
		CommandID: request.CommandID,
		Duplicate: true,
		Receipts: commonv1.Receipts{
			ProtocolVersion:   "1.0",
			CatalogRevision:   executionTiDBCatalogRevision,
			CapabilityVersion: commandv1.CapabilityName,
		},
	}, nil
}

func (runtime *executionTiDBCrashRuntime) GetApplicationResult(
	_ context.Context,
	request applicationresultv1.GetResultRequest,
) (applicationresultv1.GetResultResponse, error) {
	runtime.mu.Lock()
	defer runtime.mu.Unlock()
	runtime.resultReads++
	if runtime.result == nil {
		if !runtime.knownPending && len(runtime.requests) == 0 {
			return applicationresultv1.GetResultResponse{}, &commonv1.Failure{
				Category: commonv1.FailureMissing,
			}
		}
		return applicationresultv1.GetResultResponse{
			CommandID:          request.CommandID,
			ExecutionRefDigest: request.ExecutionRefDigest,
			State:              applicationresultv1.ResultPending,
		}, nil
	}
	result := *runtime.result
	return applicationresultv1.GetResultResponse{
		CommandID:          request.CommandID,
		ExecutionRefDigest: request.ExecutionRefDigest,
		State:              applicationresultv1.ResultTerminal,
		Result:             &result,
	}, nil
}

func haltExecutionTiDBGate(t *testing.T, fixture executionTiDBFixture) {
	t.Helper()
	result, err := fixture.db.ExecContext(t.Context(), `UPDATE _System_ConnectorAccountServingGate
SET Mode = ?, FenceToken = FenceToken + 1, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE Id = ? AND BINARY Id = BINARY ?
  AND ServingScopeDigest = ? AND BINARY ServingScopeDigest = BINARY ?
  AND ConnectorAccountId = ? AND BINARY ConnectorAccountId = BINARY ?
  AND Mode = ? AND BINARY Mode = BINARY ?
  AND ServingEpochId = ? AND BINARY ServingEpochId = BINARY ?
  AND ServingEpochDigest = ? AND BINARY ServingEpochDigest = BINARY ?`,
		accountServingModeHaltingV2,
		fixture.input.AccountServingGateID, fixture.input.AccountServingGateID,
		fixture.input.ServingScopeDigest, fixture.input.ServingScopeDigest,
		fixture.input.ConnectorAccountID, fixture.input.ConnectorAccountID,
		accountServingModeV2Serving, accountServingModeV2Serving,
		fixture.input.ServingEpochID, fixture.input.ServingEpochID,
		fixture.input.ServingEpochDigest, fixture.input.ServingEpochDigest,
	)
	if err != nil {
		t.Fatalf("halt execution fixture account gate: %v", err)
	}
	if err := requireOneExecutionRow(result, "halt execution fixture account gate"); err != nil {
		t.Fatal(err)
	}
}

func execExecutionTiDBSQL(t *testing.T, db *sql.DB, query string, args ...any) {
	t.Helper()
	if _, err := db.ExecContext(t.Context(), query, args...); err != nil {
		t.Fatalf("seed execution fixture: %v", err)
	}
}

func assertExecutionTiDBRowCount(t *testing.T, db *sql.DB, table string, want int) {
	t.Helper()
	var count int
	if err := db.QueryRowContext(t.Context(), "SELECT COUNT(*) FROM "+table).Scan(&count); err != nil {
		t.Fatalf("count %s: %v", table, err)
	}
	if count != want {
		t.Fatalf("%s rows = %d, want %d", table, count, want)
	}
}

const executionTiDBExecutionDDL = `CREATE TABLE _System_ConnectorV2Execution (
  Id VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  ExecutionProtocolVersion VARCHAR(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  ServingScopeDigest CHAR(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  ConnectorAccountId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  AccountServingGateId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  AccountServingFenceToken BIGINT NOT NULL,
  ServingEpochId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  ServingEpochDigest CHAR(71) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  ServingEpochBindingId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  PackGenerationId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
  DefinitionRef VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  DefinitionVersion VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  CanonicalInputs MEDIUMBLOB NOT NULL,
  ExternalEffectsMode VARCHAR(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  InputDigest CHAR(71) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  ExecutionRefDigest CHAR(71) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  RuntimeCommandId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  State VARCHAR(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL DEFAULT 'ready',
  ClaimFenceToken BIGINT NOT NULL DEFAULT 0,
  ClaimHolderId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NULL,
  ClaimLeaseUntil DATETIME(6) NULL,
  ResultDigest CHAR(71) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NULL,
  RuntimeCommitReceipt CHAR(71) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NULL,
  QuarantineReasonCode VARCHAR(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NULL,
  CreatedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
  LastModifiedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
  PRIMARY KEY (Id),
  UNIQUE KEY ux_connector_v2_execution_ref (ExecutionRefDigest),
  UNIQUE KEY ux_connector_v2_execution_runtime_command (RuntimeCommandId),
  KEY idx_connector_v2_execution_claim (ConnectorAccountId, State, ClaimLeaseUntil, Id),
  KEY idx_connector_v2_execution_epoch (ServingEpochId, State, Id),
  KEY idx_connector_v2_execution_gate (AccountServingGateId, State, Id),
  CONSTRAINT fk_connector_v2_execution_gate FOREIGN KEY (AccountServingGateId)
    REFERENCES _System_ConnectorAccountServingGate (Id) ON UPDATE RESTRICT ON DELETE RESTRICT,
  CONSTRAINT fk_connector_v2_execution_epoch FOREIGN KEY (ServingEpochId)
    REFERENCES _System_ConnectorServingEpoch (Id) ON UPDATE RESTRICT ON DELETE RESTRICT,
  CONSTRAINT fk_connector_v2_execution_binding_owner FOREIGN KEY (ServingEpochBindingId)
    REFERENCES _System_ConnectorServingEpochBindingOwner (ServingEpochBindingId) ON UPDATE RESTRICT ON DELETE RESTRICT,
  CONSTRAINT fk_connector_v2_execution_pack_generation FOREIGN KEY (PackGenerationId)
    REFERENCES _System_PackGenerationConnectorBindingDefinitionSet (PackGenerationId) ON UPDATE RESTRICT ON DELETE RESTRICT
) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_bin`

const executionTiDBAttemptDDL = `CREATE TABLE _System_ConnectorV2ExecutionAttempt (
  Id VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  ExecutionId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  AttemptOrdinal BIGINT NOT NULL,
  ClaimFenceToken BIGINT NOT NULL,
  ClaimHolderId VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  LeaseUntil DATETIME(6) NOT NULL,
  CreatedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
  LastModifiedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
  PRIMARY KEY (Id),
  UNIQUE KEY ux_connector_v2_execution_attempt_ordinal (ExecutionId, AttemptOrdinal),
  UNIQUE KEY ux_connector_v2_execution_attempt_fence (ExecutionId, ClaimFenceToken),
  KEY idx_connector_v2_execution_attempt_holder (ClaimHolderId, LeaseUntil, Id),
  CONSTRAINT fk_connector_v2_execution_attempt_execution FOREIGN KEY (ExecutionId)
    REFERENCES _System_ConnectorV2Execution (Id) ON UPDATE RESTRICT ON DELETE RESTRICT
) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_bin`