//go:build integration_service_integration

package integrationservice

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/base64"
	"encoding/hex"
	"fmt"
	"os"
	"sort"
	"strconv"
	"strings"
	"testing"
	"time"

	_ "github.com/go-sql-driver/mysql"
	"github.com/nexuscrm/kernel/pkg/schemacatalog"
	"github.com/nexuscrm/persistence"
	connectorevents "github.com/pingcap-inc/nexuscrm-connectors/features/connector/events"
	"github.com/pingcap-inc/nexuscrm-connectors/features/namedcredential"
	catalogv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	connectorbindingownerv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/connectorbindingownerv1"
	flowcommandv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv1"
	flowcommandv2 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv2"
)

// TestProductionCompositionStartsAgainstPublishedSchema proves the independent
// serving role uses the canonical stopped-upgrade schema, fails closed before
// the WorkItem cutover receipt, and owns one HTTP process plus the existing
// connector supervisor. Exact TLS Platform and launch-route execution remain in
// the final M2 cross-process gate rather than being simulated here.
func TestProductionCompositionStartsAgainstPublishedSchema(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("Integration Service production composition requires the declared TiDB environment")
	}
	db, cleanup := openIntegrationScratchDB(t)
	defer cleanup()
	createCanonicalSystemSchema(t, db)

	runtime := &fakePlatform{}
	runtime.ready.Store(true)
	config := processTestConfig()
	config.OrgID = "org-test"
	config.AppInstallationID = "01900000-0000-7000-8000-000000000001"
	config.AuthorizedWorkloadID = "integration-service"
	const (
		ownerPack   = "connector-test"
		packInstall = "connector-install-1"
	)
	digest := "nexus-pack-sha256-v1:" + strings.Repeat("a", 64)
	runtime.view = productionIntegrationBindingView(
		config, ownerPack, packInstall, digest, "flow-v1",
	)
	seedIntegrationServingGeneration(
		t, db, ownerPack, packInstall, digest, runtime.view.Bindings,
	)
	if _, err := db.ExecContext(t.Context(), `DELETE FROM _System_ConnectorProjectionProtocol WHERE InstalledPackId = ?`, packInstall); err != nil {
		t.Fatal(err)
	}
	if _, err := assembleProductionWithDatabase(t.Context(), "test-version", config, db, runtime, &fakeDatabase{}); err == nil ||
		!strings.Contains(err.Error(), "WorkItem cutover is not ready") {
		t.Fatalf("production composition without the WorkItem-only cutover receipt error = %v", err)
	}
	seedIntegrationWorkItemCutover(t, db)
	if _, err := assembleProductionWithDatabase(t.Context(), "test-version", config, db, runtime, &fakeDatabase{}); err == nil || !strings.Contains(err.Error(), "projection range protocol") {
		t.Fatalf("production composition without projection protocol: %v", err)
	}
	seedIntegrationProjectionProtocol(t, db, packInstall)
	t.Setenv(namedcredential.MasterKeyEnvVar, base64.StdEncoding.EncodeToString(make([]byte, 32)))
	process, err := assembleProductionWithDatabase(t.Context(), "test-version", config, db, runtime, &fakeDatabase{})
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(t.Context()); err != nil {
		t.Fatal(err)
	}
	assertHTTPStatus(t, "http://"+process.Address()+"/livez", 200)
	assertHTTPStatus(t, "http://"+process.Address()+"/readyz", 200)
	shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	if err := process.Shutdown(shutdownCtx); err != nil {
		t.Fatal(err)
	}
}

func TestProductionCompositionBindsSelectedConnectorGeneration(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("Integration Service production composition requires the declared TiDB environment")
	}
	for name, payloadVersion := range map[string]uint32{
		"v1":            flowcommandv1.PayloadFormatVersion,
		"v2 dependency": flowcommandv2.PayloadFormatVersion,
	} {
		t.Run(name, func(t *testing.T) {
			db, cleanup := openIntegrationScratchDB(t)
			defer cleanup()
			createCanonicalSystemSchema(t, db)
			seedIntegrationWorkItemCutover(t, db)

			const (
				ownerPack   = "connector-test"
				packInstall = "connector-install-1"
			)
			digest := "nexus-pack-sha256-v1:" + strings.Repeat("a", 64)
			runtime := &fakePlatform{}
			runtime.ready.Store(true)
			config := processTestConfig()
			config.OrgID = "org-test"
			config.AppInstallationID = "01900000-0000-7000-8000-000000000001"
			config.AuthorizedWorkloadID = "integration-service"
			runtime.view = productionIntegrationBindingView(
				config, ownerPack, packInstall, digest, "flow-v1",
			)
			seedIntegrationServingGeneration(
				t, db, ownerPack, packInstall, digest, runtime.view.Bindings,
			)
			for index := range runtime.view.Bindings {
				if runtime.view.Bindings[index].RuntimeCommand != nil {
					runtime.view.Bindings[index].RuntimeCommand.PayloadFormatVersion =
						payloadVersion
				}
			}
			runtime.view.Lineage.ViewDigest =
				catalogv1.ComputeIntegrationBindingViewDigest(runtime.view)
			t.Setenv(namedcredential.MasterKeyEnvVar, base64.StdEncoding.EncodeToString(make([]byte, 32)))
			process, err := assembleProductionWithDatabase(t.Context(), "test-version", config, db, runtime, &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, cancel := context.WithTimeout(context.Background(), 10*time.Second)
			defer cancel()
			if err := process.Shutdown(shutdownCtx); err != nil {
				t.Fatal(err)
			}
		})
	}
}

func TestIntegrationServingAuthorityRejectsUnexecutableRuntimeCommandRoute(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("Integration Service production composition requires the declared TiDB environment")
	}
	for name, testCase := range map[string]struct {
		want   string
		mutate func(*testing.T, *sql.DB, *catalogv1.IntegrationBindingView)
	}{
		"active Flow version drift": {
			want: "does not match Catalog command version",
			mutate: func(t *testing.T, _ *sql.DB, view *catalogv1.IntegrationBindingView) {
				for index := range view.Bindings {
					if view.Bindings[index].RuntimeCommand != nil {
						view.Bindings[index].RuntimeCommand.DefinitionVersion = "flow-v2"
						return
					}
				}
				t.Fatal("test Integration view has no Runtime command route")
			},
		},
		"process-local same-pack Flow": {
			want: "process-local Flow",
			mutate: func(t *testing.T, db *sql.DB, _ *catalogv1.IntegrationBindingView) {
				if _, err := db.ExecContext(t.Context(), `
UPDATE FlowDefinition SET OwnerPack = 'connector-test' WHERE DeveloperName = 'BoundedInbound'`); err != nil {
					t.Fatal(err)
				}
			},
		},
	} {
		t.Run(name, func(t *testing.T) {
			db, cleanup := openIntegrationScratchDB(t)
			defer cleanup()
			createCanonicalSystemSchema(t, db)
			seedIntegrationWorkItemCutover(t, db)
			const ownerPack, packInstall = "connector-test", "connector-install-1"
			digest := "nexus-pack-sha256-v1:" + strings.Repeat("a", 64)
			config := processTestConfig()
			config.OrgID = "org-test"
			config.AppInstallationID = "01900000-0000-7000-8000-000000000001"
			config.AuthorizedWorkloadID = "integration-service"
			view := productionIntegrationBindingView(config, ownerPack, packInstall, digest, "flow-v1")
			seedIntegrationServingGeneration(
				t, db, ownerPack, packInstall, digest, view.Bindings,
			)
			testCase.mutate(t, db, &view)
			view.Lineage.ViewDigest = catalogv1.ComputeIntegrationBindingViewDigest(view)
			runtime := &fakePlatform{view: view}
			runtime.ready.Store(true)
			if _, err := assembleProductionWithDatabase(t.Context(), "test-version", config, db, runtime, &fakeDatabase{}); err == nil || !strings.Contains(err.Error(), testCase.want) {
				t.Fatalf("production composition error = %v, want %q", err, testCase.want)
			}
		})
	}
}

func TestIntegrationServingAuthorityRejectsNonActiveSelectedGeneration(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("Integration Service production composition requires the declared TiDB environment")
	}
	db, cleanup := openIntegrationScratchDB(t)
	defer cleanup()
	createCanonicalSystemSchema(t, db)

	const (
		ownerPack       = "connector-test"
		selectedInstall = "connector-install-1"
		activeInstall   = "connector-install-2"
	)
	selectedDigest := "nexus-pack-sha256-v1:" + strings.Repeat("a", 64)
	activeDigest := "nexus-pack-sha256-v1:" + strings.Repeat("b", 64)
	view := integrationServingTestView([]catalogv1.IntegrationBinding{
		integrationServingTestBinding(
			catalogv1.BindingConnectorInboundRoute,
			ownerPack,
			selectedInstall,
			ownerPack+"/thing.updated",
			"a",
		),
	})
	setIntegrationServingRouteCommand(view.Bindings, "BoundedInbound")
	view.Lineage.ViewDigest = catalogv1.ComputeIntegrationBindingViewDigest(view)
	seedIntegrationServingGeneration(
		t, db, ownerPack, selectedInstall, selectedDigest, view.Bindings,
	)
	if _, err := db.ExecContext(t.Context(), `
INSERT INTO _System_PackGeneration
  (Id, PackInstallId, PackName, Category, Status, DesiredPayloadHash, AppliedPayloadHash)
VALUES (?, ?, ?, 'connector-pack', 'active', ?, ?)`,
		"generation-row-2", activeInstall, ownerPack, activeDigest, activeDigest,
	); err != nil {
		t.Fatal(err)
	}
	if _, err := db.ExecContext(t.Context(), `
UPDATE _System_PackActivation
SET ActivePackInstallId = ?, ActivationEpoch = 2, ActivatedDate = NOW()
WHERE PackName = ?`, activeInstall, ownerPack); err != nil {
		t.Fatal(err)
	}

	authority, err := newIntegrationServingAuthority(view)
	if err != nil {
		t.Fatal(err)
	}
	if err := authority.VerifyDatabase(t.Context(), db); err == nil || !strings.Contains(err.Error(), "does not match the exact active pack generation") {
		t.Fatalf("VerifyDatabase() error = %v, want active-generation mismatch", err)
	}
}

func seedIntegrationWorkItemCutover(t *testing.T, db *sql.DB) {
	t.Helper()
	if _, err := db.ExecContext(t.Context(), `
INSERT INTO _System_Migration (Id, Version, Description, AppliedAt, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, NOW(), NOW(), NOW())`,
		connectorevents.WorkItemOnlyExecutionCutoverMigrationID,
		connectorevents.WorkItemOnlyExecutionCutoverMigrationVersion,
		connectorevents.WorkItemOnlyExecutionCutoverMigrationDescription,
	); err != nil {
		t.Fatal(err)
	}
}

func seedIntegrationServingGeneration(
	t *testing.T,
	db *sql.DB,
	ownerPack, packInstall, digest string,
	bindings []catalogv1.IntegrationBinding,
) {
	t.Helper()
	generationID := integrationPackGenerationStorageID(ownerPack, packInstall)
	if _, err := db.ExecContext(t.Context(), `
INSERT INTO _System_InstalledPack
  (Id, Name, PackInstallId, Category, Status, DesiredPayloadHash, AppliedPayloadHash)
VALUES (?, ?, ?, 'connector-pack', 'active', ?, ?)`,
		"pack-row-1", ownerPack, packInstall, digest, digest,
	); err != nil {
		t.Fatal(err)
	}
	// A newly installed serving generation has a native protocol checkpoint.
	// Gate-rejection tests explicitly remove the prerequisite they exercise.
	seedIntegrationProjectionProtocol(t, db, packInstall)
	if _, err := db.ExecContext(t.Context(), `
INSERT INTO _System_InstalledPack
  (Id, Name, PackInstallId, Category, Status)
VALUES ('app-pack-row-1', 'app-test', 'app-install-1', 'app-pack', 'active')`); err != nil {
		t.Fatal(err)
	}
	if _, err := db.ExecContext(t.Context(), `
INSERT INTO FlowDefinition
  (Id, DeveloperName, MasterLabel, ActiveVersionId, LatestVersionId, IsCustomized, OwnerPack)
VALUES ('flow-definition-1', 'BoundedInbound', 'Bounded Inbound', 'flow-v1', 'flow-v1', 0, 'app-test')`); err != nil {
		t.Fatal(err)
	}
	if _, err := db.ExecContext(t.Context(), `
INSERT INTO _System_PackGeneration
  (Id, PackInstallId, PackName, Category, Status, DesiredPayloadHash, AppliedPayloadHash)
VALUES (?, ?, ?, 'connector-pack', 'active', ?, ?)`,
		generationID, packInstall, ownerPack, digest, digest,
	); err != nil {
		t.Fatal(err)
	}
	if _, err := db.ExecContext(t.Context(), `
INSERT INTO _System_PackActivation
  (Id, PackName, ActivePackInstallId, PreviousPackInstallId, ActivationEpoch, ActivatedDate)
VALUES (?, ?, ?, NULL, 1, NOW())`,
		"activation-row-1", ownerPack, packInstall,
	); err != nil {
		t.Fatal(err)
	}
	seedIntegrationBindingDefinitionSet(t, db, generationID, bindings)
	for _, capability := range []struct {
		kind, config string
	}{
		{kind: "signature_verifier", config: `{"scheme":"hmac-sha256-template","signingTemplate":"{Timestamp}.{RawBody}","timestampHeader":"X-Test-Timestamp","signatureHeader":"X-Test-Signature","signatureEncoding":"hex","skewWindow":"5m","secretName":"default"}`},
		{kind: "envelope_parser", config: `{"eventArrayPath":".events","rules":[{"eventTypeLiteral":"thing.updated","namespaceFromCtx":true,"objectKind":"thing","sourceIdPath":".objectId","idempotencyKeyPath":".eventId","payloadProjection":[{"name":"objectId","path":".objectId"}]}]}`},
	} {
		if _, err := db.ExecContext(t.Context(), `
INSERT INTO _System_ConnectorCapabilityDeclaration
  (Id, InstalledPackName, Namespace, Kind, PayloadHash)
VALUES (?, ?, ?, ?, ?)`,
			"capability-"+capability.kind, ownerPack, ownerPack, capability.kind, digest,
		); err != nil {
			t.Fatal(err)
		}
		if _, err := db.ExecContext(t.Context(), `
INSERT INTO _System_ConnectorInboundHandlerConfig
  (Id, InstalledPackId, Namespace, CapabilityKind, ConfigJSON)
VALUES (?, ?, ?, ?, ?)`,
			"handler-"+capability.kind, packInstall, ownerPack, capability.kind, capability.config,
		); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := db.ExecContext(t.Context(), `
INSERT INTO _System_ConnectorInboundRoute
  (Id, Namespace, InstalledPackId, EventKind, Version, FlowDeveloperName)
VALUES (?, ?, ?, ?, 1, ?)`,
		"route-row-1", ownerPack, packInstall, "thing.updated", "BoundedInbound",
	); err != nil {
		t.Fatal(err)
	}
	if _, err := db.ExecContext(t.Context(), `
INSERT INTO _System_ConnectorCalloutProfile
  (Id, InstalledPackId, OwnerPack, ProfileName, CredentialRef, EndpointClass,
   RetryPolicy, PageSpec, BatchSpec, ObjectPathMap, Description, IsDeleted)
VALUES (?, ?, ?, 'default', 'credential-test', 'records.read',
        '{"MaxAttempts":1}', '{}', '{}', '{}', '', 0)`,
		"profile-row-1", packInstall, ownerPack,
	); err != nil {
		t.Fatal(err)
	}
}

func seedIntegrationBindingDefinitionSet(
	t *testing.T,
	db *sql.DB,
	generationID string,
	bindings []catalogv1.IntegrationBinding,
) {
	t.Helper()
	if len(bindings) == 0 {
		t.Fatal("Integration definition-set fixture requires bindings")
	}
	definitions := make([]connectorbindingownerv1.DefinitionEnvelope, 0, len(bindings))
	for _, binding := range bindings {
		definition, err := connectorbindingownerv1.NewDefinitionEnvelope(
			connectorbindingownerv1.AuthoringInput{
				Identity:            connectorbindingownerv1.IdentityFromBinding(binding),
				CanonicalDefinition: []byte(`{}`),
			},
		)
		if err != nil {
			t.Fatalf("build %s/%s definition envelope: %v", binding.Kind, binding.BusinessKey, err)
		}
		definitions = append(definitions, definition)
	}
	set, err := connectorbindingownerv1.NewDefinitionSet(
		connectorbindingownerv1.DefinitionSetCoordinate{
			OwnerPack:          bindings[0].OwnerPack,
			PackInstallationID: bindings[0].PackInstallationID,
			PackPayloadDigest:  bindings[0].PackPayloadDigest,
		},
		definitions,
	)
	if err != nil {
		t.Fatalf("build Integration definition set: %v", err)
	}
	if _, err := db.ExecContext(t.Context(), `INSERT INTO _System_PackGenerationConnectorBindingDefinitionSet
(PackGenerationId, Protocol, DefinitionCount, DefinitionSetDigest)
VALUES (?, ?, ?, ?)`, generationID, set.Header.Protocol,
		strconv.FormatUint(set.Header.DefinitionCount, 10), set.Header.OrderedSetDigest); err != nil {
		t.Fatalf("insert Integration definition set: %v", err)
	}
	for _, entry := range set.Entries {
		if _, err := db.ExecContext(t.Context(), `INSERT INTO _System_PackGenerationConnectorBindingDefinition
(Id, PackGenerationId, PackInstallId, DefinitionOrdinal, Kind, OwnerPack,
PackPayloadDigest, BusinessKey, DefinitionVersion, DefinitionEnvelopeDigest,
DefinitionEnvelope)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
			integrationBindingDefinitionStorageID(generationID, entry.Kind, entry.BusinessKey),
			generationID,
			entry.PackInstallationID,
			strconv.FormatUint(entry.Ordinal, 10),
			string(entry.Kind),
			entry.OwnerPack,
			entry.PackPayloadDigest,
			entry.BusinessKey,
			strconv.FormatUint(entry.DefinitionVersion, 10),
			entry.EnvelopeDigest,
			entry.EnvelopeJSON,
		); err != nil {
			t.Fatalf("insert %s/%s Integration definition: %v", entry.Kind, entry.BusinessKey, err)
		}
	}
}

func integrationPackGenerationStorageID(ownerPack, packInstall string) string {
	return "packgen_" + integrationStorageDigest(ownerPack+"\x00"+packInstall)
}

func integrationBindingDefinitionStorageID(
	generationID string,
	kind catalogv1.IntegrationBindingKind,
	businessKey string,
) string {
	return "packgendef_" + integrationStorageDigest(
		generationID+"\x00"+string(kind)+"\x00"+businessKey,
	)
}

func integrationStorageDigest(value string) string {
	digest := sha256.Sum256([]byte(value))
	return hex.EncodeToString(digest[:])
}

func productionIntegrationBindingView(
	config Config,
	ownerPack, packInstall, digest, flowVersion string,
) catalogv1.IntegrationBindingView {
	view := validIntegrationBindingView(config)
	view.Bindings = []catalogv1.IntegrationBinding{
		{
			Kind: catalogv1.BindingConnectorCalloutProfile, OwnerPack: ownerPack,
			PackInstallationID: packInstall, PackPayloadDigest: digest,
			BusinessKey: ownerPack + "/default", DefinitionVersion: 1,
			CredentialRefs: []string{"credential-test"},
		},
		{
			Kind: catalogv1.BindingConnectorInboundRoute, OwnerPack: ownerPack,
			PackInstallationID: packInstall, PackPayloadDigest: digest,
			BusinessKey: ownerPack + "/thing.updated", DefinitionVersion: 1,
			CredentialRefs: []string{}, RuntimeCommand: &catalogv1.RuntimeCommandBinding{
				CommandKind: flowcommandv1.CommandKind, PayloadFormatVersion: flowcommandv1.PayloadFormatVersion,
				DefinitionRef: "BoundedInbound", DefinitionVersion: flowVersion,
			},
		},
	}
	sort.Slice(view.Bindings, func(i, j int) bool {
		left := view.Bindings[i]
		right := view.Bindings[j]
		return string(left.Kind)+"\x00"+left.OwnerPack+"\x00"+left.PackInstallationID+"\x00"+left.BusinessKey <
			string(right.Kind)+"\x00"+right.OwnerPack+"\x00"+right.PackInstallationID+"\x00"+right.BusinessKey
	})
	view.Lineage.ViewDigest = catalogv1.ComputeIntegrationBindingViewDigest(view)
	return view
}

func openIntegrationScratchDB(t *testing.T) (*sql.DB, func()) {
	t.Helper()
	host, port := os.Getenv("TIDB_HOST"), os.Getenv("TIDB_PORT")
	if port == "" {
		port = "4000"
	}
	user, password := os.Getenv("TIDB_USER"), os.Getenv("TIDB_PASSWORD")
	admin, err := sql.Open("mysql", fmt.Sprintf("%s:%s@tcp(%s:%s)/?parseTime=true&loc=UTC", user, password, host, port))
	if err != nil {
		t.Fatal(err)
	}
	name := fmt.Sprintf("nexuscrm_integration_service_%d", time.Now().UnixNano())
	if _, err := admin.Exec("CREATE DATABASE `" + name + "`"); err != nil {
		_ = admin.Close()
		t.Fatal(err)
	}
	db, err := sql.Open("mysql", fmt.Sprintf("%s:%s@tcp(%s:%s)/%s?parseTime=true&loc=UTC", user, password, host, port, name))
	if err != nil {
		_, _ = admin.Exec("DROP DATABASE IF EXISTS `" + name + "`")
		_ = admin.Close()
		t.Fatal(err)
	}
	return db, func() {
		_ = db.Close()
		_, _ = admin.Exec("DROP DATABASE IF EXISTS `" + name + "`")
		_ = admin.Close()
	}
}

func createCanonicalSystemSchema(t *testing.T, db *sql.DB) {
	t.Helper()
	definitions, err := schemacatalog.SystemTableDefinitions()
	if err != nil {
		t.Fatal(err)
	}
	repository := persistence.NewSchemaRepository(db, nil)
	for _, definition := range definitions {
		if err := repository.CreatePhysicalTable(t.Context(), definition); err != nil {
			t.Fatalf("create canonical table %s: %v", definition.TableName, err)
		}
	}
}

func seedIntegrationProjectionProtocol(t *testing.T, db *sql.DB, packID string) {
	t.Helper()
	if _, err := db.ExecContext(t.Context(), `INSERT INTO _System_ConnectorProjectionProtocol (InstalledPackId,ProtocolVersion,LegacyMaxOrdinal) VALUES (?,1,0)`, packID); err != nil {
		t.Fatal(err)
	}
}