//go:build integration

package catalogconsumer

import (
	"context"
	"crypto/ed25519"
	"crypto/sha256"
	"database/sql"
	"errors"
	"fmt"
	"os"
	"regexp"
	"strconv"
	"testing"
	"time"

	"github.com/go-sql-driver/mysql"
	database "github.com/nexuscrm/db"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/appauthority"
	consumerv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/catalogpublicationv1"
	connectorbindingownerv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/connectorbindingownerv1"
	common "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commonv1"
	"github.com/pingcap-inc/nexuscrm-controlplane/bootstrapimpl"
	"github.com/pingcap-inc/nexuscrm-controlplane/catalogapp"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/catalogwriter"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/connectorbindingdefinitionauthority"
	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
	"github.com/pingcap-inc/nexuscrm-controlplane/releaseartifact"
)

func TestCatalogConsumerViewsTiDBReadOneAuthorizedRevision(t *testing.T) {
	db, cleanup := openCatalogConsumerTiDB(t)
	t.Cleanup(cleanup)
	authority, err := catalogapp.New(db, catalogapp.Options{})
	if err != nil {
		t.Fatal(err)
	}
	if err := authority.EnsureSchema(context.Background()); err != nil {
		t.Fatal(err)
	}
	verified := testVerifiedAppPackage(t)
	grants := []catalogapp.ConsumerGrant{
		{
			AuthorizedWorkloadID: testIntegrationWorkload,
			CapabilityID:         consumerv1.IntegrationBindingCapabilityName,
		},
		{
			AuthorizedWorkloadID: testOtherWorkload,
			CapabilityID:         consumerv1.AppMetadataCapabilityName,
		},
		{
			AuthorizedWorkloadID: testOtherWorkload,
			CapabilityID:         consumerv1.AppMetadataV2CapabilityName,
		},
		{
			AuthorizedWorkloadID: testOtherWorkload,
			CapabilityID:         consumerv1.IntegrationBindingCapabilityName,
		},
	}
	bindingSelections := []catalogapp.IntegrationBindingSelection{
		testBindingSelection(
			testIntegrationWorkload,
			consumerv1.BindingConnectorInboundRoute,
			"route",
		),
		testBindingSelection(
			testOtherWorkload,
			consumerv1.BindingConnectorPushContract,
			"push",
		),
	}
	bindingSelections[0].Binding.RuntimeCommand =
		&consumerv1.RuntimeCommandBinding{
			CommandKind:          "runtime.flow.execute",
			PayloadFormatVersion: 1,
			DefinitionRef:        "FlowRoute",
			DefinitionVersion:    "flow-version-1",
		}
	ensureCatalogConsumerPackAuthorityTables(t, db)
	now := time.Now().UTC()
	seed := sha256.Sum256([]byte("catalog-consumer-release-test-key"))
	privateKey := ed25519.NewKeyFromSeed(seed[:])
	publicKey := privateKey.Public().(ed25519.PublicKey)
	publisherEpoch, err := authority.RegisterPublisher(
		context.Background(),
		catalogapp.PublisherTrust{
			PublisherID: verified.Manifest.PublisherID,
			Active:      true,
			Keys: []catalogapp.PublisherKey{{
				KeyID:      "key-test",
				Algorithm:  releaseartifact.SignatureAlgorithm,
				PublicKey:  publicKey,
				ValidFrom:  now.Add(-time.Hour),
				ValidUntil: now.Add(time.Hour),
			}},
		},
		0,
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := authority.ReserveAppID(
		context.Background(),
		verified.Manifest.PublisherID,
		verified.Manifest.AppID,
		publisherEpoch,
	); err != nil {
		t.Fatal(err)
	}
	registered := importCatalogConsumerRelease(t, authority, verified, now, privateKey)
	if registered.AppReleaseDigest != verified.AppReleaseDigest {
		t.Fatalf("registered AppRelease = %#v", registered)
	}
	generationID := connectorbindingdefinitionauthority.PackGenerationStorageID(
		bindingSelections[0].Binding.OwnerPack,
		bindingSelections[0].Binding.PackInstallationID,
	)
	previousGenerationID := connectorbindingdefinitionauthority.PackGenerationStorageID(
		bindingSelections[0].Binding.OwnerPack,
		"pack-install-integration-0",
	)
	if _, err := db.Exec(`INSERT INTO _System_PackGeneration
	(Id, PackName, PackInstallId, Category, Status, DesiredPayloadHash, AppliedPayloadHash,
	IsDeleted)
	VALUES
	(?, ?, ?, ?, 'active', ?, ?, 0),
	(?, ?, ?, ?, 'active', ?, ?, 0)`,
		previousGenerationID,
		bindingSelections[0].Binding.OwnerPack,
		"pack-install-integration-0",
		string(packs.ConnectorPack),
		bindingSelections[0].Binding.PackPayloadDigest,
		bindingSelections[0].Binding.PackPayloadDigest,
		generationID,
		bindingSelections[0].Binding.OwnerPack,
		bindingSelections[0].Binding.PackInstallationID,
		string(packs.ConnectorPack),
		bindingSelections[0].Binding.PackPayloadDigest,
		bindingSelections[0].Binding.PackPayloadDigest,
	); err != nil {
		t.Fatal(err)
	}
	insertCatalogConsumerDefinitionSet(t, db, generationID, bindingSelections)
	if _, err := db.Exec(`INSERT INTO _System_PackActivation
	(Id, PackName, ActivePackInstallId, PreviousPackInstallId, ActivationEpoch,
	ActivatedDate, IsDeleted)
	VALUES (?, ?, ?, NULL, 2, CURRENT_TIMESTAMP(6), 0)`,
		"catalog-consumer-test-activation",
		bindingSelections[0].Binding.OwnerPack,
		bindingSelections[0].Binding.PackInstallationID,
	); err != nil {
		t.Fatal(err)
	}

	operation, err := authority.BeginOperation(
		context.Background(),
		catalogapp.BeginOperationRequest{
			Kind:                    catalogapp.OperationPrepare,
			OrgID:                   testOrgID,
			AppID:                   testAppID,
			TargetAppReleaseDigest:  verified.AppReleaseDigest,
			IdempotencyKey:          "catalog-consumer-lifecycle",
			RuntimeAudience:         "runtime",
			AuthorizedWorkloads:     []string{testIntegrationWorkload, testOtherWorkload},
			GrantedCapabilities:     verified.Manifest.RequiredEngineCapabilities,
			ResolvedDependencies:    []catalogapp.ResolvedDependency{},
			ConsumerGrants:          grants,
			IntegrationBindings:     bindingSelections,
			ExpectedCatalogRevision: 0,
			ExpectedSecurityEpoch:   0,
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	store, err := catalogsnapshot.NewDirectoryStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	selected := testSnapshot(t, testOrgID, operation.TargetCatalogRevision)
	if err := store.Put(context.Background(), selected); err != nil {
		t.Fatal(err)
	}
	effectiveComponents := testRequiredEffectiveMetadataComponents()
	if _, err := authority.
		CommitOperationMetadataSelectionWithEffectiveComponents(
			context.Background(), operation.OperationID, string(selected.Digest()),
			effectiveComponents,
		); err != nil {
		t.Fatal(err)
	}
	if _, err := authority.CommitServingArtifacts(
		context.Background(), operation.OperationID,
	); err != nil {
		t.Fatal(err)
	}
	if err := authority.PublishRuntimeSecurityProjection(
		context.Background(),
		operation.OperationID,
		catalogConsumerProjectionWriter{},
	); err != nil {
		t.Fatal(err)
	}
	operation, err = authority.CompleteOperation(
		context.Background(), operation.OperationID, nil,
	)
	if err != nil {
		t.Fatal(err)
	}
	if operation.TargetCatalogRevision != 1 ||
		operation.TargetSecurityEpoch != 1 ||
		operation.AppInstallationID == "" ||
		operation.Step != "ACTIVATED" {
		t.Fatalf("completed Catalog operation = %#v", operation)
	}
	publication, publicationDigest := testPublication(
		t, operation.TargetCatalogRevision,
	)
	if _, err := db.Exec(`INSERT INTO _System_CatalogPublication
(PublicationDigest, BindingVersion, OrgId, CatalogRevision, CatalogSnapshotDigest,
SchemaGeneration, EngineGeneration, LayoutValidationViewDigest,
LayoutBundleOwnerVerificationReceiptSetDigest, LayoutActivationSetDigest,
SecurityEpoch, LayoutRoutingSecurityViewDigest,
CreatedDate)
VALUES (?, ?, ?, ?, ?, 11, 13, ?, NULL, ?, 17, ?, CURRENT_TIMESTAMP(6))`,
		publicationDigest,
		catalogpublicationv1.CanonicalBindingDomain,
		testOrgID,
		operation.TargetCatalogRevision,
		publication.CatalogSnapshotDigest,
		publication.LayoutValidationViewDigest,
		publication.LayoutActivationSetDigest,
		publication.LayoutRoutingSecurityViewDigest,
	); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`INSERT INTO _System_CatalogPublicationActive
(OrgId, ActivePublicationPresent, ActivePublicationDigest,
SelectedByCommandId, HistorySequence, CreatedDate, LastModifiedDate)
VALUES (?, TRUE, ?, ?, 1, CURRENT_TIMESTAMP(6), CURRENT_TIMESTAMP(6))`,
		testOrgID,
		publicationDigest,
		"018f0000-0000-7000-9000-000000000003",
	); err != nil {
		t.Fatal(err)
	}

	reader, err := New(db, store)
	if err != nil {
		t.Fatal(err)
	}
	scope, err := NewAuthorizedConsumerScope(
		testOrgID,
		operation.AppInstallationID,
		testOtherWorkload,
		[]string{consumerv1.AppMetadataCapabilityName},
	)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	view, err := reader.GetAppMetadataView(
		ctx, scope, consumerv1.GetRequest{},
	)
	if err != nil {
		diagnosticTx, beginErr := reader.beginRead(ctx)
		if beginErr != nil {
			t.Fatalf("GetAppMetadataView: %v; diagnostic begin: %v", err, beginErr)
		}
		defer rollback(diagnosticTx)
		active, activeErr := reader.readActiveContext(
			ctx,
			diagnosticTx,
			scope,
			consumerv1.AppMetadataCapabilityName,
			nil,
		)
		if activeErr != nil {
			t.Fatalf(
				"GetAppMetadataView: %v; active authority diagnostic: %v",
				err, activeErr,
			)
		}
		_, componentErr := readVerifiedAppReleaseComponents(
			ctx, diagnosticTx, active,
		)
		t.Fatalf(
			"GetAppMetadataView: %v; component diagnostic: %v",
			err, componentErr,
		)
	}
	if view.Lineage.CatalogRevision != operation.TargetCatalogRevision ||
		view.Receipts.SecurityEpoch != 1 ||
		view.AppReleaseDigest != verified.AppReleaseDigest ||
		len(view.Components) != 1 {
		t.Fatalf("TiDB AppMetadataView = %#v", view)
	}
	v2Scope, err := NewAuthorizedConsumerScope(
		testOrgID,
		operation.AppInstallationID,
		testOtherWorkload,
		[]string{consumerv1.AppMetadataV2CapabilityName},
	)
	if err != nil {
		t.Fatal(err)
	}
	v2View, err := reader.GetAppMetadataViewV2(
		ctx, v2Scope, consumerv1.GetRequest{},
	)
	if err != nil {
		t.Fatal(err)
	}
	if v2View.Lineage.SourceSnapshotDigest != string(selected.Digest()) ||
		len(v2View.AppComponents) != 1 ||
		len(v2View.EffectiveComponents) != 2 ||
		string(v2View.EffectiveComponents[0].CanonicalBehavior) !=
			`{"applications":[]}` ||
		string(v2View.EffectiveComponents[1].CanonicalBehavior) !=
			`{"objects":[]}` {
		t.Fatalf("TiDB AppMetadataViewV2 = %#v", v2View)
	}

	absentScope, err := NewAuthorizedConsumerScope(
		testOrgID,
		"018f0000-0000-7000-9000-000000000009",
		testOtherWorkload,
		[]string{consumerv1.AppMetadataCapabilityName},
	)
	if err != nil {
		t.Fatal(err)
	}
	_, err = reader.GetAppMetadataView(
		ctx, absentScope, consumerv1.GetRequest{},
	)
	var absentFailure *consumerv1.Failure
	if !errors.As(err, &absentFailure) ||
		absentFailure.Category != common.FailureMissing {
		t.Fatalf(
			"absent AppInstallation error = %v, want missing without fallback",
			err,
		)
	}

	integrationScope, err := NewAuthorizedConsumerScope(
		testOrgID,
		operation.AppInstallationID,
		testIntegrationWorkload,
		[]string{consumerv1.IntegrationBindingCapabilityName},
	)
	if err != nil {
		t.Fatal(err)
	}
	bindingView, err := reader.GetIntegrationBindingView(
		ctx, integrationScope, consumerv1.GetRequest{},
	)
	if err != nil {
		t.Fatal(err)
	}
	if bindingView.Lineage.CatalogRevision != operation.TargetCatalogRevision ||
		bindingView.AuthorizedWorkloadID != testIntegrationWorkload ||
		len(bindingView.Bindings) != 1 ||
		bindingView.Bindings[0].Kind != consumerv1.BindingConnectorInboundRoute ||
		bindingView.Bindings[0].BusinessKey != "route" ||
		bindingView.Bindings[0].RuntimeCommand == nil ||
		bindingView.Bindings[0].RuntimeCommand.DefinitionVersion !=
			"flow-version-1" {
		t.Fatalf(
			"TiDB IntegrationBindingView leaks another workload or lost authority: %#v",
			bindingView,
		)
	}

	for _, corruption := range []struct {
		name string
		sql  string
	}{
		{
			name: "reserved previous pointer",
			sql: `UPDATE _System_PackActivation
SET PreviousPackInstallId = 'reserved-previous'
WHERE PackName = ?`,
		},
		{
			name: "zero activation epoch",
			sql: `UPDATE _System_PackActivation
SET PreviousPackInstallId = NULL, ActivationEpoch = 0
WHERE PackName = ?`,
		},
	} {
		if _, err := db.Exec(
			corruption.sql,
			bindingSelections[0].Binding.OwnerPack,
		); err != nil {
			t.Fatal(err)
		}
		_, err = reader.GetIntegrationBindingView(
			ctx, integrationScope, consumerv1.GetRequest{},
		)
		var malformed *consumerv1.Failure
		if !errors.As(err, &malformed) ||
			malformed.Category != common.FailureMissing {
			t.Fatalf(
				"%s error = %v, want missing",
				corruption.name,
				err,
			)
		}
	}
	if _, err := db.Exec(`UPDATE _System_PackActivation
SET PreviousPackInstallId = NULL, ActivationEpoch = 2
WHERE PackName = ?`,
		bindingSelections[0].Binding.OwnerPack,
	); err != nil {
		t.Fatal(err)
	}

	// Pack authority is revalidated on every read. A replacement must fail
	// closed rather than serving the operation-bound reference or falling back
	// to mutable installed-pack state.
	if _, err := db.Exec(`UPDATE _System_PackActivation
SET ActivePackInstallId = 'different-pack-install'
WHERE PackName = ?`,
		bindingSelections[0].Binding.OwnerPack,
	); err != nil {
		t.Fatal(err)
	}
	_, err = reader.GetIntegrationBindingView(
		ctx, integrationScope, consumerv1.GetRequest{},
	)
	var failure *consumerv1.Failure
	if !errors.As(err, &failure) ||
		failure.Category != common.FailureMissing {
		t.Fatalf("mixed pack generation error = %v, want missing", err)
	}

	// An unrelated Layout publication must not relabel this App's effective
	// metadata. This test leaves a nonempty publication binding in place, then
	// changes its selected digest independently of the activated App operation.
	newerPublication, newerPublicationDigest := testPublication(t, 2)
	if _, err := db.Exec(`INSERT INTO _System_CatalogPublication
(PublicationDigest, BindingVersion, OrgId, CatalogRevision, CatalogSnapshotDigest,
SchemaGeneration, EngineGeneration, LayoutValidationViewDigest,
LayoutBundleOwnerVerificationReceiptSetDigest, LayoutActivationSetDigest,
SecurityEpoch, LayoutRoutingSecurityViewDigest,
CreatedDate)
VALUES (?, ?, ?, 2, ?, 11, 13, ?, NULL, ?, 17, ?, CURRENT_TIMESTAMP(6))`,
		newerPublicationDigest,
		catalogpublicationv1.CanonicalBindingDomain,
		testOrgID,
		newerPublication.CatalogSnapshotDigest,
		newerPublication.LayoutValidationViewDigest,
		newerPublication.LayoutActivationSetDigest,
		newerPublication.LayoutRoutingSecurityViewDigest,
	); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`UPDATE _System_CatalogPublicationActive
SET ActivePublicationDigest = ?, SelectedByCommandId = ?,
HistorySequence = 2, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE OrgId = ?`,
		newerPublicationDigest,
		"018f0000-0000-7000-9000-000000000004",
		testOrgID,
	); err != nil {
		t.Fatal(err)
	}

	unchanged, err := reader.GetAppMetadataViewV2(ctx, v2Scope, consumerv1.GetRequest{})
	if err != nil {
		t.Fatal(err)
	}
	if unchanged.Lineage != v2View.Lineage || unchanged.Receipts.SecurityEpoch != 1 {
		t.Fatalf("unrelated Layout publication changed App lineage: %#v", unchanged)
	}
	minimum := uint64(2)
	_, err = reader.GetAppMetadataView(ctx, scope, consumerv1.GetRequest{MinCatalogRevision: &minimum})
	var stale *consumerv1.Failure
	if !errors.As(err, &stale) || stale.Category != common.FailureStaleState || stale.RequiredRevision != 2 || stale.ServedRevision != 1 {
		t.Fatalf("minimum App revision: %v", err)
	}
	after := uint64(1)
	watchCtx, stopWatch := context.WithTimeout(ctx, 50*time.Millisecond)
	_, err = reader.NextAppMetadataView(watchCtx, scope, consumerv1.WatchRequest{AfterCatalogRevision: &after})
	stopWatch()
	var timedOut *consumerv1.Failure
	if !errors.As(err, &timedOut) || timedOut.Category != common.FailureDeadlineExceeded {
		t.Fatalf("unrelated publication woke App watch: %v", err)
	}

	// Security-only commit reissues receipts without inventing a new content
	// revision. Use the existing org writer, as the real lifecycle does.
	securityTx, err := db.BeginTx(ctx, nil)
	if err != nil {
		t.Fatal(err)
	}
	defer securityTx.Rollback()
	state, err := catalogwriter.Lock(ctx, securityTx, testOrgID)
	if err != nil {
		t.Fatal(err)
	}
	if err := catalogwriter.Reserve(ctx, securityTx, state, "consumer-security-only"); err != nil {
		t.Fatal(err)
	}
	if err := catalogwriter.Commit(ctx, securityTx, testOrgID, "consumer-security-only", state.CatalogRevision, state.SecurityEpoch+1); err != nil {
		t.Fatal(err)
	}
	if err := securityTx.Commit(); err != nil {
		t.Fatal(err)
	}
	restarted, err := New(db, store)
	if err != nil {
		t.Fatal(err)
	}
	secured, err := restarted.GetAppMetadataViewV2(ctx, v2Scope, consumerv1.GetRequest{})
	if err != nil {
		t.Fatal(err)
	}
	if secured.Lineage != v2View.Lineage || secured.Receipts.SecurityEpoch != 2 {
		t.Fatalf("security-only/restart view: %#v", secured)
	}

	// A real second App operation selects different immutable metadata. Its
	// candidate and activation must never replace the first App's lineage.
	other := testVerifiedAppPackageForApp(t, "io.example.other")
	if err := authority.ReserveAppID(ctx, other.Manifest.PublisherID, other.Manifest.AppID, publisherEpoch); err != nil {
		t.Fatal(err)
	}
	importCatalogConsumerRelease(t, authority, other, now, privateKey)
	otherOperation, err := authority.BeginOperation(ctx, catalogapp.BeginOperationRequest{
		Kind: catalogapp.OperationPrepare, OrgID: testOrgID, AppID: other.Manifest.AppID, TargetAppReleaseDigest: other.AppReleaseDigest,
		IdempotencyKey: "consumer-other-app", RuntimeAudience: "runtime", AuthorizedWorkloads: []string{testOtherWorkload}, GrantedCapabilities: other.Manifest.RequiredEngineCapabilities, ResolvedDependencies: []catalogapp.ResolvedDependency{},
		ConsumerGrants: []catalogapp.ConsumerGrant{{AuthorizedWorkloadID: testOtherWorkload, CapabilityID: consumerv1.AppMetadataCapabilityName}}, ExpectedCatalogRevision: 1, ExpectedSecurityEpoch: 2,
	})
	if err != nil {
		t.Fatal(err)
	}
	otherSnapshot := testSnapshot(t, testOrgID, otherOperation.TargetCatalogRevision)
	if err := store.Put(ctx, otherSnapshot); err != nil {
		t.Fatal(err)
	}
	if _, err := authority.CommitServingArtifacts(ctx, otherOperation.OperationID); !errors.Is(err, catalogapp.ErrNotReady) {
		t.Fatalf("V1 consumer serving without metadata selection = %v, want not ready", err)
	}
	if _, err := authority.CommitOperationMetadataSelection(ctx, otherOperation.OperationID, string(otherSnapshot.Digest())); err != nil {
		t.Fatal(err)
	}
	duringCandidate, err := reader.GetAppMetadataViewV2(ctx, v2Scope, consumerv1.GetRequest{})
	if err != nil || duringCandidate.Lineage != v2View.Lineage {
		t.Fatalf("other App candidate affected active selection: %#v %v", duringCandidate.Lineage, err)
	}
	if _, err := authority.CommitServingArtifacts(ctx, otherOperation.OperationID); err != nil {
		t.Fatal(err)
	}
	if err := authority.PublishRuntimeSecurityProjection(ctx, otherOperation.OperationID, catalogConsumerProjectionWriter{}); err != nil {
		t.Fatal(err)
	}
	for range 2 {
		if _, err := authority.CompleteOperation(ctx, otherOperation.OperationID, nil); err != nil {
			t.Fatal(err)
		}
	}
	otherScope, err := NewAuthorizedConsumerScope(testOrgID, otherOperation.AppInstallationID, testOtherWorkload, []string{consumerv1.AppMetadataCapabilityName})
	if err != nil {
		t.Fatal(err)
	}
	otherView, err := restarted.GetAppMetadataView(ctx, otherScope, consumerv1.GetRequest{})
	if err != nil {
		t.Fatal(err)
	}
	firstView, err := restarted.GetAppMetadataViewV2(ctx, v2Scope, consumerv1.GetRequest{})
	if err != nil {
		t.Fatal(err)
	}
	if otherView.Lineage.CatalogRevision != 2 || otherView.Lineage.SourceSnapshotDigest != string(otherSnapshot.Digest()) || otherView.AppID != other.Manifest.AppID ||
		firstView.Lineage != v2View.Lineage || firstView.Receipts.SecurityEpoch != 3 || otherView.Receipts.SecurityEpoch != 3 {
		t.Fatalf("App selections mixed: first=%#v other=%#v", firstView.Lineage, otherView.Lineage)
	}
	t.Log("real App A/B activation and repeated completion preserve distinct immutable selections")

	// Older low-level Authority callers could activate V1 consumers without
	// selecting metadata. Reproduce that persisted shape, then recover through
	// the existing same-release refresh; never rewrite the old operation.
	if _, err := db.Exec(`DELETE FROM _System_CatalogAppOperationJournal
WHERE OperationId = ? AND Step = 'METADATA_SELECTION_COMMITTED'`, otherOperation.OperationID); err != nil {
		t.Fatal(err)
	}
	if _, err := restarted.GetAppMetadataView(ctx, otherScope, consumerv1.GetRequest{}); err == nil {
		t.Fatal("legacy consumer unexpectedly fell back to another App's snapshot")
	}
	refresh, err := authority.BeginOperation(ctx, catalogapp.BeginOperationRequest{
		Kind: catalogapp.OperationUpgrade, OrgID: testOrgID, AppID: other.Manifest.AppID,
		AppInstallationID: otherOperation.AppInstallationID, TargetAppReleaseDigest: other.AppReleaseDigest,
		RefreshActiveRelease: true, ContinuationIntentDigest: testSHA256("e"),
		IdempotencyKey: "consumer-selection-refresh", RuntimeAudience: "runtime",
		AuthorizedWorkloads: []string{testOtherWorkload}, GrantedCapabilities: other.Manifest.RequiredEngineCapabilities,
		ResolvedDependencies:    []catalogapp.ResolvedDependency{},
		ConsumerGrants:          []catalogapp.ConsumerGrant{{AuthorizedWorkloadID: testOtherWorkload, CapabilityID: consumerv1.AppMetadataCapabilityName}},
		ExpectedCatalogRevision: 2, ExpectedSecurityEpoch: 3,
	})
	if err != nil {
		t.Fatal(err)
	}
	refreshedSnapshot := testSnapshot(t, testOrgID, refresh.TargetCatalogRevision)
	if err := store.Put(ctx, refreshedSnapshot); err != nil {
		t.Fatal(err)
	}
	if _, err := authority.CommitOperationMetadataSelection(ctx, refresh.OperationID, string(refreshedSnapshot.Digest())); err != nil {
		t.Fatal(err)
	}
	if _, err := authority.CommitServingArtifacts(ctx, refresh.OperationID); err != nil {
		t.Fatal(err)
	}
	if err := authority.PublishRuntimeSecurityProjection(ctx, refresh.OperationID, catalogConsumerProjectionWriter{}); err != nil {
		t.Fatal(err)
	}
	if _, err := authority.CompleteOperation(ctx, refresh.OperationID, nil); err != nil {
		t.Fatal(err)
	}
	recovered, err := restarted.GetAppMetadataView(ctx, otherScope, consumerv1.GetRequest{})
	if err != nil {
		t.Fatal(err)
	}
	if recovered.AppReleaseDigest != otherView.AppReleaseDigest || recovered.Lineage.CatalogRevision != 3 ||
		recovered.Lineage.SourceSnapshotDigest != string(refreshedSnapshot.Digest()) || recovered.Receipts.SecurityEpoch != 4 {
		t.Fatalf("same-release consumer recovery: %#v", recovered)
	}
	if _, err := authority.ReadOperationMetadataSelection(ctx, otherOperation.OperationID); !errors.Is(err, catalogapp.ErrNotFound) {
		t.Fatalf("refresh rewrote legacy operation history: %v", err)
	}
	t.Log("same-release refresh restores a legacy V1 consumer without backfilling history")

	// The cheap unchanged-revision probe must still stop this exact consumer
	// when its grant disappears. It does not need to hydrate or recompute the
	// complete grant set to make that liveness decision.
	if _, err := db.Exec(`DELETE FROM _System_CatalogConsumerGrant
WHERE OperationId = ? AND AuthorizedWorkloadId = ? AND CapabilityId = ?`,
		operation.OperationID,
		testOtherWorkload,
		consumerv1.AppMetadataCapabilityName,
	); err != nil {
		t.Fatal(err)
	}
	afterRevocation := uint64(2)
	_, err = reader.NextAppMetadataView(
		ctx,
		scope,
		consumerv1.WatchRequest{AfterCatalogRevision: &afterRevocation},
	)
	var revoked *consumerv1.Failure
	if !errors.As(err, &revoked) || revoked.Category != common.FailureForbidden {
		t.Fatalf("revoked TiDB watch error = %v, want forbidden", err)
	}
}

type catalogConsumerProjectionWriter struct{}

func importCatalogConsumerRelease(t *testing.T, authority *catalogapp.Authority, verified appauthority.VerifiedPackage, issuedAt time.Time, privateKey ed25519.PrivateKey) catalogapp.RegisteredRelease {
	t.Helper()
	_, packageRaw, err := releaseartifact.BuildPackage(verified.Package)
	if err != nil {
		t.Fatal(err)
	}
	packageHash := sha256.Sum256(packageRaw)
	_, attestationRaw, err := releaseartifact.SignAttestation(verified, releaseartifact.Provenance{
		SchemaVersion:         releaseartifact.ProvenanceSchemaVersion,
		SourceRepository:      "example/catalog-consumer",
		SourceRevision:        "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
		SourcePath:            "metadata",
		SourceTreeDigest:      "sha256:" + fmt.Sprintf("%x", packageHash),
		BuilderID:             "catalog-consumer-test/v1",
		ContractModuleVersion: "v0.1.8",
	}, "key-test", issuedAt, privateKey)
	if err != nil {
		t.Fatal(err)
	}
	protocols := make([]releaseartifact.SupportedProtocol, len(verified.Manifest.RequiredProtocols))
	for i, requirement := range verified.Manifest.RequiredProtocols {
		protocols[i] = releaseartifact.SupportedProtocol{
			ProtocolID: requirement.ProtocolID, Major: requirement.Major, Minor: requirement.MinimumMinor,
		}
	}
	componentKinds := make([]releaseartifact.SupportedComponentKind, len(verified.Bundle.Components))
	for i, component := range verified.Bundle.Components {
		componentKinds[i] = releaseartifact.SupportedComponentKind{
			ComponentKind: component.ComponentKind, ComponentFormatVersion: component.ComponentFormatVersion,
		}
	}
	registered, err := authority.ImportReleaseArtifact(context.Background(), packageRaw, attestationRaw, releaseartifact.HostCompatibility{
		Protocols: protocols, EngineCapabilities: append([]string(nil), verified.Manifest.RequiredEngineCapabilities...), ComponentKinds: componentKinds,
	})
	if err != nil {
		t.Fatal(err)
	}
	return registered
}

func (catalogConsumerProjectionWriter) ActivateProjection(
	context.Context,
	uint64,
	catalogapp.RuntimeSecurityProjection,
) error {
	return nil
}

func TestCatalogIntegrationBindingRuntimeCommandStoppedUpgradeTiDB(t *testing.T) {
	db, cleanup := openCatalogConsumerTiDB(t)
	t.Cleanup(cleanup)
	if _, err := db.Exec(`
CREATE TABLE _System_CatalogIntegrationBinding (
IntegrationBindingId CHAR(64) PRIMARY KEY,
OrgId VARCHAR(128) NOT NULL,
AppInstallationId CHAR(36) NOT NULL,
OperationId CHAR(36) NOT NULL,
AuthorizedWorkloadId VARBINARY(255) NOT NULL,
Kind VARBINARY(64) NOT NULL,
OwnerPack VARBINARY(255) NOT NULL,
PackInstallId VARBINARY(255) NOT NULL,
PackPayloadDigest VARBINARY(128) NOT NULL,
BusinessKey VARBINARY(255) NOT NULL,
DefinitionVersion DECIMAL(20,0) NOT NULL,
CredentialRefs JSON NOT NULL,
CreatedDate DATETIME(6) NOT NULL,
INDEX idx_catalog_integration_operation (OperationId)
)`); err != nil {
		t.Fatal(err)
	}
	authority, err := catalogapp.New(db, catalogapp.Options{})
	if err != nil {
		t.Fatal(err)
	}
	if err := authority.EnsureSchema(context.Background()); err != nil {
		t.Fatal(err)
	}
	rows, err := db.Query(`
SELECT COLUMN_NAME, IS_NULLABLE
FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_SCHEMA = DATABASE()
  AND TABLE_NAME = '_System_CatalogIntegrationBinding'
  AND COLUMN_NAME IN (
    'CommandKind',
    'CommandPayloadFormatVersion',
    'CommandDefinitionRef',
    'CommandDefinitionVersion'
  )
ORDER BY COLUMN_NAME`)
	if err != nil {
		t.Fatal(err)
	}
	defer rows.Close()
	columns := make(map[string]string)
	for rows.Next() {
		var name, nullable string
		if err := rows.Scan(&name, &nullable); err != nil {
			t.Fatal(err)
		}
		columns[name] = nullable
	}
	if err := rows.Err(); err != nil {
		t.Fatal(err)
	}
	if len(columns) != 4 {
		t.Fatalf("Runtime command columns = %#v, want four", columns)
	}
	for name, nullable := range columns {
		if nullable != "YES" {
			t.Fatalf("Runtime command column %s nullable = %q, want YES", name, nullable)
		}
	}
}

func ensureCatalogConsumerPackAuthorityTables(t *testing.T, db *sql.DB) {
	t.Helper()
	if err := bootstrapimpl.EnsureL0SchemaDDL(database.NewConnectionFromDB(db)); err != nil {
		t.Fatalf("initialize canonical L0 schema: %v", err)
	}
}

func insertCatalogConsumerDefinitionSet(
	t *testing.T,
	db *sql.DB,
	generationID string,
	selections []catalogapp.IntegrationBindingSelection,
) {
	t.Helper()
	if len(selections) == 0 {
		t.Fatal("Catalog consumer definition-set fixture requires bindings")
	}
	definitions := make([]connectorbindingownerv1.DefinitionEnvelope, 0, len(selections))
	for _, selection := range selections {
		binding := selection.Binding
		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)
	}
	first := selections[0].Binding
	set, err := connectorbindingownerv1.NewDefinitionSet(
		connectorbindingownerv1.DefinitionSetCoordinate{
			OwnerPack:          first.OwnerPack,
			PackInstallationID: first.PackInstallationID,
			PackPayloadDigest:  first.PackPayloadDigest,
		},
		definitions,
	)
	if err != nil {
		t.Fatalf("build Catalog consumer definition set: %v", err)
	}
	now := time.Now().UTC()
	if _, err := db.Exec(`INSERT INTO _System_PackGenerationConnectorBindingDefinitionSet
(PackGenerationId, Protocol, DefinitionCount, DefinitionSetDigest, CreatedDate,
LastModifiedDate)
VALUES (?, ?, ?, ?, ?, ?)`,
		generationID,
		set.Header.Protocol,
		strconv.FormatUint(set.Header.DefinitionCount, 10),
		set.Header.OrderedSetDigest,
		now,
		now,
	); err != nil {
		t.Fatalf("insert Catalog consumer definition set: %v", err)
	}
	for _, entry := range set.Entries {
		if _, err := db.Exec(`INSERT INTO _System_PackGenerationConnectorBindingDefinition
(Id, PackGenerationId, PackInstallId, DefinitionOrdinal, Kind, OwnerPack,
PackPayloadDigest, BusinessKey, DefinitionVersion, DefinitionEnvelopeDigest,
DefinitionEnvelope, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
			connectorbindingdefinitionauthority.DefinitionStorageID(
				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,
			now,
			now,
		); err != nil {
			t.Fatalf("insert %s/%s Catalog consumer definition: %v", entry.Kind, entry.BusinessKey, err)
		}
	}
}

func openCatalogConsumerTiDB(t *testing.T) (*sql.DB, func()) {
	t.Helper()
	host := os.Getenv("TIDB_HOST")
	if host == "" {
		host = "127.0.0.1"
	}
	port := os.Getenv("TIDB_PORT")
	if port == "" {
		port = "4000"
	}
	cfg := mysql.Config{
		User:                 "root",
		Passwd:               os.Getenv("TIDB_PASSWORD"),
		Net:                  "tcp",
		Addr:                 host + ":" + port,
		Collation:            "utf8mb4_unicode_ci",
		ParseTime:            true,
		Loc:                  time.UTC,
		AllowNativePasswords: true,
	}
	admin, err := sql.Open("mysql", cfg.FormatDSN())
	if err != nil {
		t.Fatal(err)
	}
	if err := admin.Ping(); err != nil {
		_ = admin.Close()
		if os.Getenv("NEXUSCRM_REQUIRE_CATALOG_CONSUMER_TIDB") == "1" {
			t.Fatalf("required TiDB is unavailable at %s:%s: %v", host, port, err)
		}
		t.Skipf("TiDB is unavailable at %s:%s: %v", host, port, err)
	}
	runID := os.Getenv("NEXUSCRM_CATALOG_CONSUMER_RUN_ID")
	if runID != "" && !regexp.MustCompile(`^[A-Za-z0-9_]{1,12}$`).MatchString(runID) {
		t.Fatal("NEXUSCRM_CATALOG_CONSUMER_RUN_ID must be 1-12 ASCII letters, digits, or underscores")
	}
	database := fmt.Sprintf(
		"nexuscrm_catalog_consumer_%s_%d", runID, time.Now().UnixNano(),
	)
	if _, err := admin.Exec(
		"CREATE DATABASE `" + database + "` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci",
	); err != nil {
		_ = admin.Close()
		t.Fatal(err)
	}
	testConfig := cfg
	testConfig.DBName = database
	db, err := sql.Open("mysql", testConfig.FormatDSN())
	if err != nil {
		_, _ = admin.Exec("DROP DATABASE IF EXISTS `" + database + "`")
		_ = admin.Close()
		t.Fatal(err)
	}
	cleanup := func() {
		_ = db.Close()
		_, _ = admin.Exec("DROP DATABASE IF EXISTS `" + database + "`")
		_ = admin.Close()
	}
	return db, cleanup
}

// Only the unrelated Layout authority fixture uses a publication binding.
func testPublication(t *testing.T, revision uint64) (catalogpublicationv1.Binding, string) {
	t.Helper()
	binding := catalogpublicationv1.Binding{OrgID: testOrgID, CatalogRevision: revision, CatalogSnapshotDigest: testSHA256("1"), SchemaGeneration: 11, EngineGeneration: 13, LayoutValidationViewDigest: testSHA256("2"), LayoutActivationSetDigest: testSHA256("3"), SecurityEpoch: 17, LayoutRoutingSecurityViewDigest: testSHA256("4")}
	digest, err := catalogpublicationv1.BindingDigest(binding)
	if err != nil {
		t.Fatal(err)
	}
	return binding, digest
}