//go:build integration_service_integration

package integrationservice

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"

	database "github.com/nexuscrm/db"
	"github.com/nexuscrm/persistence"
	connectorserviceconfiguration "github.com/pingcap-inc/nexuscrm-connectors/features/connector/serviceconfiguration"
	connectorserving "github.com/pingcap-inc/nexuscrm-connectors/features/connector/serving"
	"github.com/pingcap-inc/nexuscrm-connectors/features/namedcredential"
	catalogsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	connectorbindingownerv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/connectorbindingownerv1"
	applicationresultsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/applicationresultv1"
	commandsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commandv1"
	flowcommandsemanticv3 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv3"
)

func TestM2V2ExecutionConformanceCrossesSeparateRuntimeProcessWithoutProductionReachability(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("M2 V2 execution conformance requires the declared TiDB environment")
	}
	db, cleanup := openIntegrationScratchDB(t)
	defer cleanup()
	createCanonicalSystemSchema(t, db)
	seedIntegrationWorkItemCutover(t, db)

	const (
		ownerPack   = "connector-v2-conformance"
		packInstall = "connector-v2-conformance-installation"
	)
	packDigest := "nexus-pack-sha256-v1:" + strings.Repeat("a", 64)
	config := processTestConfig()
	config.OrgID = "org-v2-conformance"
	config.AppInstallationID = "01900000-0000-7000-8000-000000000002"
	config.AuthorizedWorkloadID = "integration-service"
	productionView := productionIntegrationBindingView(
		config,
		ownerPack,
		packInstall,
		packDigest,
		"flow-v1",
	)
	productionRuntimeView, _ := m2RuntimeConsumerViews(t, productionView)
	productionView.Lineage.SourceSnapshotDigest = string(productionRuntimeView.SourceSnapshotDigest())
	productionView.Receipts.SecurityEpoch = 1
	productionView.Lineage.ViewDigest = catalogsemantic.ComputeIntegrationBindingViewDigest(productionView)
	if err := catalogsemantic.ValidateIntegrationBindingView(productionView); err != nil {
		t.Fatalf("M2 production Integration binding fixture: %v", err)
	}
	conformanceView := productionIntegrationBindingView(
		config,
		ownerPack,
		packInstall,
		packDigest,
		"flow-v2-conformance",
	)
	for index := range conformanceView.Bindings {
		if conformanceView.Bindings[index].RuntimeCommand != nil {
			conformanceView.Bindings[index].RuntimeCommand.CommandKind = flowcommandsemanticv3.CommandKind
			conformanceView.Bindings[index].RuntimeCommand.PayloadFormatVersion = flowcommandsemanticv3.PayloadFormatVersion
		}
	}
	conformanceRuntimeView, _ := m2RuntimeConsumerViews(t, conformanceView)
	conformanceView.Lineage.SourceSnapshotDigest = string(conformanceRuntimeView.SourceSnapshotDigest())
	conformanceView.Receipts.SecurityEpoch = 1
	conformanceView.Lineage.ViewDigest = catalogsemantic.ComputeIntegrationBindingViewDigest(conformanceView)
	if err := catalogsemantic.ValidateIntegrationBindingView(conformanceView); err != nil {
		t.Fatalf("M2 V2 conformance Integration binding fixture: %v", err)
	}
	seedIntegrationServingGeneration(
		t,
		db,
		ownerPack,
		packInstall,
		packDigest,
		productionView.Bindings,
	)
	candidate := seedM2V2ExecutionAuthority(
		t,
		db,
		config,
		conformanceView,
		ownerPack,
		packInstall,
	)
	t.Setenv(
		namedcredential.MasterKeyEnvVar,
		"AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=",
	)

	testDir := t.TempDir()
	eventsFile := filepath.Join(testDir, "runtime-events.jsonl")
	productionViewFile := filepath.Join(testDir, "production-integration-view.json")
	conformanceViewFile := filepath.Join(testDir, "conformance-integration-view.json")
	caFile, certFile, keyFile := writeM2RuntimeTLSFiles(t, testDir)
	writeM2JSON(t, productionViewFile, productionView)
	writeM2JSON(t, conformanceViewFile, conformanceView)
	runtimeDSN := m2RuntimeDatabaseDSN(t, db)

	productionChild := startM2RuntimeChild(
		t,
		productionViewFile,
		certFile,
		keyFile,
		eventsFile,
		runtimeDSN,
	)
	productionClient := dialM2RuntimeClient(
		t,
		productionChild.address,
		caFile,
		config.AppInstallationID,
	)
	conformance, err := newIsolatedExecutionConformance(
		t.Context(),
		db,
		productionClient,
	)
	if err != nil {
		t.Fatal(err)
	}
	execution, duplicate, err := conformance.Admit(t.Context(), candidate)
	if err != nil || duplicate {
		t.Fatalf("isolated V2 Admit() duplicate=%v error=%v", duplicate, err)
	}

	process, err := assembleProductionWithDatabase(
		t.Context(),
		m2RuntimeProcessVersion,
		config,
		db,
		productionClient,
		&fakeDatabase{},
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(t.Context()); err != nil {
		t.Fatal(err)
	}
	assertHTTPStatus(t, "http://"+process.Address()+"/readyz", 200)
	shutdownCtx, cancelShutdown := context.WithTimeout(context.Background(), 10*time.Second)
	if err := process.Shutdown(shutdownCtx); err != nil {
		cancelShutdown()
		t.Fatal(err)
	}
	cancelShutdown()
	productionChild.stop(t)

	assertM2V2ExecutionState(t, db, execution.ID(), "ready", 0, 0)
	if got := countM2RuntimeEvents(readM2RuntimeEvents(t, eventsFile), "AcceptCommand"); got != 0 {
		t.Fatalf("production Integration composition submitted V2 work: commands=%d", got)
	}

	conformanceChild := startM2RuntimeChild(
		t,
		conformanceViewFile,
		certFile,
		keyFile,
		eventsFile,
		runtimeDSN,
	)
	conformanceClient := dialM2RuntimeClient(
		t,
		conformanceChild.address,
		caFile,
		config.AppInstallationID,
	)
	conformance, err = newIsolatedExecutionConformance(
		t.Context(),
		db,
		conformanceClient,
	)
	if err != nil {
		t.Fatal(err)
	}
	firstCtx, cancelFirst := context.WithTimeout(context.Background(), 5*time.Second)
	_, firstErr := conformance.Reconcile(
		firstCtx,
		execution.ID(),
		"m2-v2-conformance-first",
		time.Second,
	)
	cancelFirst()
	if !errors.Is(firstErr, connectorserving.ErrExecutionResultPending) {
		t.Fatalf("first isolated V2 Reconcile() = %v, want pending result", firstErr)
	}
	assertM2V2ExecutionState(t, db, execution.ID(), "ready", 1, 1)
	completeM2V2RuntimeCommand(t, db, config, execution.RuntimeCommandID())

	secondCtx, cancelSecond := context.WithTimeout(context.Background(), 5*time.Second)
	completed, err := conformance.Reconcile(
		secondCtx,
		execution.ID(),
		"m2-v2-conformance-second",
		time.Second,
	)
	cancelSecond()
	if err != nil {
		t.Fatal(err)
	}
	if completed.State() != "completed" {
		t.Fatalf("second isolated V2 Reconcile() state = %q", completed.State())
	}
	if _, _, ok := completed.ResultCorrelation(); !ok {
		t.Fatal("completed isolated V2 execution lacks Runtime result correlation")
	}
	assertM2V2ExecutionState(t, db, execution.ID(), "completed", 1, 1)
	events := readM2RuntimeEvents(t, eventsFile)
	commands := m2RuntimeCommands(events)
	if len(commands) != 1 {
		t.Fatalf("isolated V2 Runtime submissions = %d, want exactly 1", len(commands))
	}
	command := commands[0]
	if command.CommandID != execution.RuntimeCommandID() ||
		command.CommandKind != flowcommandsemanticv3.CommandKind ||
		command.PayloadFormatVersion != flowcommandsemanticv3.PayloadFormatVersion {
		t.Fatalf("isolated V2 Runtime command identity/version = %#v", command)
	}
	payload, err := flowcommandsemanticv3.Decode(command.Payload)
	if err != nil || payload.ExternalEffectsMode != flowcommandsemanticv3.SuppressExternalEffects {
		t.Fatalf("isolated V2 Runtime external-effects mode = %q error=%v", payload.ExternalEffectsMode, err)
	}
	assertM2CommandSharesPlatformEndpoint(t, events, command, conformanceView)
	if err := conformanceClient.Close(); err != nil {
		t.Fatal(err)
	}
	conformanceChild.stop(t)
}

func seedM2V2ExecutionAuthority(
	t *testing.T,
	db *sql.DB,
	config Config,
	view catalogsemantic.IntegrationBindingView,
	ownerPack string,
	packInstall string,
) connectorserving.ConformanceExecutionCandidate {
	t.Helper()
	var runtimeBinding *catalogsemantic.IntegrationBinding
	for index := range view.Bindings {
		if view.Bindings[index].RuntimeCommand == nil {
			continue
		}
		if runtimeBinding != nil {
			t.Fatal("M2 V2 conformance fixture has multiple Runtime command bindings")
		}
		binding := view.Bindings[index]
		runtimeBinding = &binding
	}
	if runtimeBinding == nil {
		t.Fatal("M2 V2 conformance fixture lacks a Runtime command binding")
	}
	envelope, err := connectorbindingownerv1.NewDefinitionEnvelope(
		connectorbindingownerv1.AuthoringInput{
			Identity:            connectorbindingownerv1.IdentityFromBinding(*runtimeBinding),
			CanonicalDefinition: []byte(`{}`),
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	credentialRefs, err := json.Marshal(runtimeBinding.CredentialRefs)
	if err != nil {
		t.Fatal(err)
	}

	digest := func(character string) string {
		return "sha256:" + strings.Repeat(character, 64)
	}
	const (
		epochID   = "m2-v2-conformance-epoch"
		bindingID = "m2-v2-conformance-binding"
		gateID    = "m2-v2-conformance-gate"
		accountID = "m2-v2-conformance-account"
	)
	scopeDigest := strings.Repeat("b", 64)
	epochDigest := digest("c")
	packGenerationID := integrationPackGenerationStorageID(ownerPack, packInstall)
	for _, statement := range []struct {
		query string
		args  []any
	}{
		{
			query: `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', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
			args: []any{
				epochID, epochDigest, scopeDigest, scopeDigest,
				config.OrgID, config.AppInstallationID, config.AuthorizedWorkloadID,
				view.Lineage.CatalogRevision, view.Lineage.SchemaGeneration, view.Lineage.EngineGeneration,
				view.Lineage.SourceSnapshotDigest, view.Lineage.ViewDigest,
				view.Receipts.ProtocolVersion, view.Receipts.CapabilityVersion, view.Receipts.SecurityEpoch,
				"m2-v2-service-configuration", digest("d"),
				[]byte("m2-v2-service-configuration-receipt"), digest("e"),
			},
		},
		{
			query: `INSERT INTO _System_ConnectorServingEpochReproducibility
(ServingEpochId, ConnectorDigestProfileId,
 ServiceConfigurationGenerationProtocolVersion,
 ServiceConfigurationReceiptProtocolVersion)
VALUES (?, ?, ?, ?)`,
			args: []any{
				epochID,
				connectorserving.ConnectorDigestProfileV2ID,
				connectorserviceconfiguration.GenerationProtocolV1,
				connectorserviceconfiguration.ReceiptProtocolV1,
			},
		},
		{
			query: `INSERT INTO _System_ConnectorServingEpochBinding
(Id, ServingEpochId, BindingOrdinal, BindingIdentityDigest,
 Kind, OwnerPack, PackInstallationId, PackPayloadDigest,
 BusinessKey, DefinitionVersion, CredentialRefs,
 RuntimeCommandKind, RuntimePayloadFormatVersion,
 RuntimeDefinitionRef, RuntimeDefinitionVersion)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
			args: []any{
				bindingID, epochID, uint64(0), strings.Repeat("f", 64),
				runtimeBinding.Kind, runtimeBinding.OwnerPack,
				runtimeBinding.PackInstallationID, runtimeBinding.PackPayloadDigest,
				runtimeBinding.BusinessKey, runtimeBinding.DefinitionVersion, credentialRefs,
				runtimeBinding.RuntimeCommand.CommandKind,
				runtimeBinding.RuntimeCommand.PayloadFormatVersion,
				runtimeBinding.RuntimeCommand.DefinitionRef,
				runtimeBinding.RuntimeCommand.DefinitionVersion,
			},
		},
		{
			query: `INSERT INTO _System_ConnectorServingEpochBindingOwner
(ServingEpochBindingId, PackGenerationId,
 DefinitionEnvelopeVersion, DefinitionReceiptVersion,
 CanonicalDefinitionDigest, DefinitionOwnerReceipt, DefinitionOwnerReceiptDigest)
VALUES (?, ?, ?, ?, ?, ?, ?)`,
			args: []any{
				bindingID, packGenerationID,
				envelope.EnvelopeVersion, envelope.ReceiptVersion,
				envelope.CanonicalDefinitionDigest, envelope.OwnerReceipt, envelope.OwnerReceiptDigest,
			},
		},
		{
			query: `INSERT INTO _System_ConnectorAccountServingGate
(Id, ServingScopeDigest, ConnectorAccountId, Mode, FenceToken,
 ServingEpochId, ServingEpochDigest)
VALUES (?, ?, ?, 'v2_serving', ?, ?, ?)`,
			args: []any{gateID, scopeDigest, accountID, int64(7), epochID, epochDigest},
		},
	} {
		if _, err := db.ExecContext(t.Context(), statement.query, statement.args...); err != nil {
			t.Fatalf("seed M2 V2 execution authority: %v", err)
		}
	}
	candidate, err := connectorserving.NewConformanceExecutionCandidate(
		connectorserving.ConformanceExecutionInput{
			ExecutionID:              "m2-v2-conformance-execution",
			ServingScopeDigest:       scopeDigest,
			ConnectorAccountID:       accountID,
			AccountServingGateID:     gateID,
			AccountServingFenceToken: 7,
			ServingEpochID:           epochID,
			ServingEpochDigest:       epochDigest,
			ServingEpochBindingID:    bindingID,
			PackGenerationID:         packGenerationID,
			DefinitionRef:            runtimeBinding.RuntimeCommand.DefinitionRef,
			DefinitionVersion:        runtimeBinding.RuntimeCommand.DefinitionVersion,
			CanonicalInputs:          []byte(`{"record":{"Name":"Ada"}}`),
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	return candidate
}

func assertM2V2ExecutionState(
	t *testing.T,
	db *sql.DB,
	executionID string,
	wantState string,
	wantFence int64,
	wantAttempts int,
) {
	t.Helper()
	var state string
	var fence int64
	if err := db.QueryRowContext(
		t.Context(),
		`SELECT State, ClaimFenceToken
FROM _System_ConnectorV2Execution
WHERE Id = ? AND BINARY Id = BINARY ?`,
		executionID,
		executionID,
	).Scan(&state, &fence); err != nil {
		t.Fatal(err)
	}
	if state != wantState || fence != wantFence {
		t.Fatalf("M2 V2 execution state/fence = %q/%d, want %q/%d", state, fence, wantState, wantFence)
	}
	var attempts int
	if err := db.QueryRowContext(
		t.Context(),
		`SELECT COUNT(*) FROM _System_ConnectorV2ExecutionAttempt
WHERE ExecutionId = ? AND BINARY ExecutionId = BINARY ?`,
		executionID,
		executionID,
	).Scan(&attempts); err != nil {
		t.Fatal(err)
	}
	if attempts != wantAttempts {
		t.Fatalf("M2 V2 execution attempts = %d, want %d", attempts, wantAttempts)
	}
}

func completeM2V2RuntimeCommand(
	t *testing.T,
	db *sql.DB,
	config Config,
	commandID string,
) {
	t.Helper()
	control, err := persistence.NewRuntimeWorkOwnershipControl(db)
	if err != nil {
		t.Fatal(err)
	}
	owner, err := control.Activate(
		t.Context(),
		persistence.RuntimeWorkOwnershipActivation{
			OperationID:         "activate-m2-v2-conformance-runtime-owner",
			WorkFamily:          persistence.RuntimePlatformCommandExecutionWorkFamily,
			DeploymentID:        "m2-v2-conformance-runtime-owner",
			WorkerReleaseDigest: "sha256:" + strings.Repeat("9", 64),
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	workScope := persistence.RuntimeWorkScope{
		OrgID: config.OrgID, AppInstallationID: config.AppInstallationID,
		RuntimeCapabilityID: commandsemantic.CapabilityName,
	}
	transactions := persistence.NewTransactionManager(database.NewConnectionFromDB(db))
	store, err := persistence.NewRuntimePlatformCommandExecutionStore(
		db,
		m2RuntimeTransactionOwner{TransactionManager: transactions},
		owner,
		workScope,
		30*time.Second,
		3,
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := store.CheckReady(t.Context()); err != nil {
		t.Fatal(err)
	}
	lease, found, err := store.Claim(t.Context())
	if err != nil || !found || lease.Row.CommandID != commandID {
		t.Fatalf(
			"M2 V2 Runtime Claim() = command:%q found:%v error:%v",
			lease.Row.CommandID,
			found,
			err,
		)
	}
	if err := store.ExecuteApplication(
		t.Context(),
		lease,
		func(context.Context) (persistence.RuntimePlatformCommandApplicationCompletion, error) {
			return persistence.RuntimePlatformCommandApplicationCompletion{
				Outcome: applicationresultsemantic.OutcomeApplied,
				Body:    json.RawMessage(`{"saved":true}`),
			}, nil
		},
	); err != nil {
		t.Fatal(err)
	}
}