package catalogconsumer

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"math"
	"sort"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/DATA-DOG/go-sqlmock"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/appauthority"
	consumerv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	common "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commonv1"
	"github.com/pingcap-inc/nexuscrm-controlplane/catalogapp"
)

const (
	testOrgID               = "org-catalog-consumer"
	testInstallationID      = "018f0000-0000-7000-9000-000000000001"
	testAppID               = "io.example.app"
	testIntegrationWorkload = "workload-integration"
	testOtherWorkload       = "workload-sfcrm"
	testReleaseDigest       = "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
	testPackPayloadDigest   = "nexus-pack-sha256-v1:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
	testOperationID         = "018f0000-0000-7000-9000-000000000002"
)

func TestDefaultWatchPollIntervalBacksOffToOneSecond(t *testing.T) {
	db, _, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	reader, err := New(db, testSnapshotStore(t))
	if err != nil {
		t.Fatal(err)
	}
	if reader.pollInterval != 250*time.Millisecond {
		t.Fatalf("initial poll interval = %s, want 250ms", reader.pollInterval)
	}
	if reader.maxPollInterval != time.Second {
		t.Fatalf("maximum poll interval = %s, want 1s", reader.maxPollInterval)
	}
	interval := reader.pollInterval
	for index, want := range []time.Duration{
		500 * time.Millisecond,
		time.Second,
		time.Second,
	} {
		interval = nextWatchPollInterval(interval, reader.maxPollInterval)
		if interval != want {
			t.Fatalf("backoff step %d = %s, want %s", index+1, interval, want)
		}
	}
}

func TestExplicitWatchPollIntervalRemainsFixed(t *testing.T) {
	db, _, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	reader, err := NewWithWatchPollInterval(db, testSnapshotStore(t), 17*time.Millisecond)
	if err != nil {
		t.Fatal(err)
	}
	if reader.pollInterval != 17*time.Millisecond ||
		reader.maxPollInterval != 17*time.Millisecond {
		t.Fatalf(
			"explicit poll interval range = %s..%s, want fixed 17ms",
			reader.pollInterval,
			reader.maxPollInterval,
		)
	}
}

func TestWatchPollIntervalRangeRejectsMaximumBelowInitial(t *testing.T) {
	db, _, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	if _, err := newWithWatchPollIntervalRange(
		db, testSnapshotStore(t),
		time.Second,
		500*time.Millisecond,
	); err == nil {
		t.Fatal("newWithWatchPollIntervalRange accepted maximum below initial")
	}
}

func TestGetIntegrationBindingViewScopesAndVerifiesCompleteAuthority(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	reader, err := New(db, testSnapshotStore(t))
	if err != nil {
		t.Fatal(err)
	}
	scope, err := NewAuthorizedConsumerScope(
		testOrgID,
		testInstallationID,
		testIntegrationWorkload,
		[]string{consumerv1.IntegrationBindingCapabilityName},
	)
	if err != nil {
		t.Fatal(err)
	}
	selection := testSelection(t, 8)
	allSelections := []catalogapp.IntegrationBindingSelection{
		testBindingSelection(
			testIntegrationWorkload,
			consumerv1.BindingConnectorInboundRoute,
			"route",
		),
		testBindingSelection(
			testOtherWorkload,
			consumerv1.BindingConnectorPushContract,
			"push",
		),
	}
	allSelections[0].Binding.RuntimeCommand = &consumerv1.RuntimeCommandBinding{
		CommandKind:          "runtime.flow.execute",
		PayloadFormatVersion: 1,
		DefinitionRef:        "FlowRoute",
		DefinitionVersion:    "flow-version-1",
	}
	setDigest, err := catalogapp.ComputeIntegrationBindingSetDigest(allSelections)
	if err != nil {
		t.Fatal(err)
	}

	mock.ExpectBegin()
	expectActiveInstallation(
		mock,
		8,
		testReleaseDigest,
		[]string{testIntegrationWorkload, testOtherWorkload},
		[]string{"runtime.records.query.v1"},
	)
	expectConsumerGrantAuthority(mock, 8, []catalogapp.ConsumerGrant{{
		AuthorizedWorkloadID: testIntegrationWorkload,
		CapabilityID:         consumerv1.IntegrationBindingCapabilityName,
	}})
	expectActiveSelection(mock, selection)
	mock.ExpectQuery("(?s)SELECT.*FROM _System_CatalogIntegrationBindingSet").
		WithArgs(testOperationID).
		WillReturnRows(sqlmock.NewRows([]string{
			"AppInstallationId", "BindingSetDigest", "BindingCount",
		}).AddRow(testInstallationID, setDigest, []byte("2")))
	bindingRows := sqlmock.NewRows([]string{
		"OperationId", "OrgId", "AppInstallationId",
		"AuthorizedWorkloadId", "Kind", "OwnerPack",
		"PackInstallId", "PackPayloadDigest", "BusinessKey",
		"DefinitionVersion", "CredentialRefs", "CommandKind",
		"CommandPayloadFormatVersion", "CommandDefinitionRef",
		"CommandDefinitionVersion", "ActivePackInstallId",
		"PreviousPackInstallId", "ActivationEpoch",
		"ActivatedDate", "ActivationDeleted", "GenerationPackName", "GenerationInstallId",
		"GenerationStatus", "GenerationDigest", "GenerationDeleted",
	})
	for _, selection := range allSelections {
		var commandKind, commandPayloadVersion, commandDefinitionRef,
			commandDefinitionVersion any
		if selection.Binding.RuntimeCommand != nil {
			commandKind = selection.Binding.RuntimeCommand.CommandKind
			commandPayloadVersion = []byte("1")
			commandDefinitionRef = selection.Binding.RuntimeCommand.DefinitionRef
			commandDefinitionVersion =
				selection.Binding.RuntimeCommand.DefinitionVersion
		}
		bindingRows.AddRow(
			testOperationID,
			testOrgID,
			testInstallationID,
			selection.AuthorizedWorkloadID,
			selection.Binding.Kind,
			selection.Binding.OwnerPack,
			selection.Binding.PackInstallationID,
			selection.Binding.PackPayloadDigest,
			selection.Binding.BusinessKey,
			[]byte("1"),
			[]byte(`[]`),
			commandKind,
			commandPayloadVersion,
			commandDefinitionRef,
			commandDefinitionVersion,
			selection.Binding.PackInstallationID,
			nil,
			[]byte("2"),
			time.Now(),
			0,
			selection.Binding.OwnerPack,
			selection.Binding.PackInstallationID,
			"active",
			selection.Binding.PackPayloadDigest,
			0,
		)
	}
	mock.ExpectQuery("(?s)SELECT.*FROM _System_CatalogIntegrationBinding AS b").
		WithArgs(testOperationID).
		WillReturnRows(bindingRows)
	mock.ExpectCommit()

	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	view, err := reader.GetIntegrationBindingView(
		ctx, scope, consumerv1.GetRequest{},
	)
	if err != nil {
		t.Fatal(err)
	}
	if view.OrgID != testOrgID ||
		view.AppInstallationID != testInstallationID ||
		view.AuthorizedWorkloadID != testIntegrationWorkload ||
		view.Lineage.CatalogRevision != 8 ||
		view.Lineage.SourceSnapshotDigest != selection.CatalogSnapshotDigest ||
		view.Receipts.SecurityEpoch != selection.SecurityEpoch {
		t.Fatalf("unexpected view authority: %#v", view)
	}
	if len(view.Bindings) != 1 ||
		view.Bindings[0].Kind != consumerv1.BindingConnectorInboundRoute ||
		view.Bindings[0].BusinessKey != "route" ||
		view.Bindings[0].RuntimeCommand == nil ||
		view.Bindings[0].RuntimeCommand.DefinitionRef != "FlowRoute" {
		t.Fatalf("cross-workload disclosure or missing binding: %#v", view.Bindings)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestIntegrationRuntimeCommandFromSQLFailsClosed(t *testing.T) {
	valid := func(value string) sql.NullString {
		return sql.NullString{String: value, Valid: true}
	}
	tests := []struct {
		name        string
		kind        sql.NullString
		version     []byte
		definition  sql.NullString
		exact       sql.NullString
		wantNil     bool
		wantFailure bool
	}{
		{name: "N-1 row remains absent", wantNil: true},
		{
			name: "complete command",
			kind: valid("runtime.flow.execute"), version: []byte("1"),
			definition: valid("FlowRoute"), exact: valid("flow-version-1"),
		},
		{
			name: "partial columns",
			kind: valid("runtime.flow.execute"), version: []byte("1"),
			wantFailure: true,
		},
		{
			name:       "uint32 overflow",
			kind:       valid("runtime.flow.execute"),
			version:    []byte(strconv.FormatUint(uint64(math.MaxUint32)+1, 10)),
			definition: valid("FlowRoute"), exact: valid("flow-version-1"),
			wantFailure: true,
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			got, err := integrationRuntimeCommandFromSQL(
				test.kind, test.version, test.definition, test.exact,
			)
			if (err != nil) != test.wantFailure {
				t.Fatalf("integrationRuntimeCommandFromSQL error = %v", err)
			}
			if test.wantNil && got != nil {
				t.Fatalf("integrationRuntimeCommandFromSQL = %#v, want nil", got)
			}
			if !test.wantNil && !test.wantFailure &&
				(got == nil || got.PayloadFormatVersion != 1 ||
					got.DefinitionVersion != "flow-version-1") {
				t.Fatalf("integrationRuntimeCommandFromSQL = %#v", got)
			}
		})
	}
}

func TestReadVerifiedIntegrationBindingsRejectsMalformedActivationAuthority(t *testing.T) {
	selection := testBindingSelection(
		testIntegrationWorkload,
		consumerv1.BindingConnectorInboundRoute,
		"route",
	)
	setDigest, err := catalogapp.ComputeIntegrationBindingSetDigest(
		[]catalogapp.IntegrationBindingSelection{selection},
	)
	if err != nil {
		t.Fatal(err)
	}
	for _, test := range []struct {
		name            string
		previousInstall any
		activationEpoch []byte
	}{
		{
			name:            "reserved previous pointer",
			previousInstall: "install-pack-0",
			activationEpoch: []byte("2"),
		},
		{
			name:            "zero epoch",
			activationEpoch: []byte("0"),
		},
		{
			name:            "malformed epoch",
			activationEpoch: []byte("not-an-epoch"),
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			db, mock, err := sqlmock.New()
			if err != nil {
				t.Fatal(err)
			}
			defer db.Close()
			mock.ExpectBegin()
			mock.ExpectQuery("(?s)SELECT.*FROM _System_CatalogIntegrationBindingSet").
				WithArgs(testOperationID).
				WillReturnRows(sqlmock.NewRows([]string{
					"AppInstallationId", "BindingSetDigest", "BindingCount",
				}).AddRow(testInstallationID, setDigest, []byte("1")))
			mock.ExpectQuery("(?s)SELECT.*FROM _System_CatalogIntegrationBinding AS b").
				WithArgs(testOperationID).
				WillReturnRows(sqlmock.NewRows([]string{
					"OperationId", "OrgId", "AppInstallationId",
					"AuthorizedWorkloadId", "Kind", "OwnerPack",
					"PackInstallId", "PackPayloadDigest", "BusinessKey",
					"DefinitionVersion", "CredentialRefs", "CommandKind",
					"CommandPayloadFormatVersion", "CommandDefinitionRef",
					"CommandDefinitionVersion", "ActivePackInstallId",
					"PreviousPackInstallId", "ActivationEpoch",
					"ActivatedDate", "ActivationDeleted", "GenerationPackName",
					"GenerationInstallId", "GenerationStatus", "GenerationDigest",
					"GenerationDeleted",
				}).AddRow(
					testOperationID,
					testOrgID,
					testInstallationID,
					selection.AuthorizedWorkloadID,
					selection.Binding.Kind,
					selection.Binding.OwnerPack,
					selection.Binding.PackInstallationID,
					selection.Binding.PackPayloadDigest,
					selection.Binding.BusinessKey,
					[]byte("1"),
					[]byte(`[]`),
					nil,
					nil,
					nil,
					nil,
					selection.Binding.PackInstallationID,
					test.previousInstall,
					test.activationEpoch,
					time.Now(),
					0,
					selection.Binding.OwnerPack,
					selection.Binding.PackInstallationID,
					"active",
					selection.Binding.PackPayloadDigest,
					0,
				))
			mock.ExpectRollback()
			tx, err := db.BeginTx(context.Background(), nil)
			if err != nil {
				t.Fatal(err)
			}
			_, err = readVerifiedIntegrationBindings(
				context.Background(),
				tx,
				activeContext{
					orgID:             testOrgID,
					appInstallationID: testInstallationID,
					activeWorkloads:   []string{testIntegrationWorkload},
					operationID:       testOperationID,
				},
				testIntegrationWorkload,
			)
			var semantic *consumerv1.Failure
			if !errors.As(err, &semantic) ||
				semantic.Category != common.FailureMissing {
				t.Fatalf("malformed activation authority error = %v, want missing", err)
			}
			_ = tx.Rollback()
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatal(err)
			}
		})
	}
}

func TestGetIntegrationBindingViewFailsClosedBeforeMixedRevisionRead(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	reader, err := New(db, testSnapshotStore(t))
	if err != nil {
		t.Fatal(err)
	}
	scope, err := NewAuthorizedConsumerScope(
		testOrgID,
		testInstallationID,
		testIntegrationWorkload,
		[]string{consumerv1.IntegrationBindingCapabilityName},
	)
	if err != nil {
		t.Fatal(err)
	}
	selection := testSelection(t, 7)
	mock.ExpectBegin()
	expectActiveInstallation(
		mock,
		7,
		testReleaseDigest,
		[]string{testIntegrationWorkload, testOtherWorkload},
		[]string{"runtime.records.query.v1"},
	)
	expectConsumerGrantAuthority(mock, 7, []catalogapp.ConsumerGrant{{
		AuthorizedWorkloadID: testIntegrationWorkload,
		CapabilityID:         consumerv1.IntegrationBindingCapabilityName,
	}})
	expectActiveSelection(mock, selection)
	mock.ExpectRollback()
	minimum := uint64(8)
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	_, err = reader.GetIntegrationBindingView(
		ctx, scope, consumerv1.GetRequest{MinCatalogRevision: &minimum},
	)
	var semantic *consumerv1.Failure
	if !errors.As(err, &semantic) ||
		semantic.Category != common.FailureStaleState ||
		semantic.StaleAxis != common.StaleAxisCatalog ||
		semantic.RequiredRevision != 8 ||
		semantic.ServedRevision != 7 ||
		semantic.Receipts == nil ||
		!common.ValidReceipts(
			*semantic.Receipts,
			consumerv1.IntegrationBindingCapabilityName,
		) {
		t.Fatalf("stale authority error = %v; SQL expectations: %v", err, mock.ExpectationsWereMet())
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestGetIntegrationBindingViewRejectsMissingDurableConsumerGrant(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	reader, err := New(db, testSnapshotStore(t))
	if err != nil {
		t.Fatal(err)
	}
	scope, err := NewAuthorizedConsumerScope(
		testOrgID,
		testInstallationID,
		testIntegrationWorkload,
		[]string{consumerv1.IntegrationBindingCapabilityName},
	)
	if err != nil {
		t.Fatal(err)
	}
	mock.ExpectBegin()
	expectActiveInstallation(
		mock,
		7,
		testReleaseDigest,
		[]string{testIntegrationWorkload},
		[]string{"runtime.records.query.v1"},
	)
	expectConsumerGrantAuthority(mock, 7, nil)
	mock.ExpectRollback()
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	_, err = reader.GetIntegrationBindingView(ctx, scope, consumerv1.GetRequest{})
	var semantic *consumerv1.Failure
	if !errors.As(err, &semantic) ||
		semantic.Category != common.FailureForbidden {
		t.Fatalf("missing durable consumer grant error = %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestGetAppMetadataViewRejectsCorruptCommittedGrantSet(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	reader, err := New(db, testSnapshotStore(t))
	if err != nil {
		t.Fatal(err)
	}
	scope, err := NewAuthorizedConsumerScope(
		testOrgID,
		testInstallationID,
		testOtherWorkload,
		[]string{consumerv1.AppMetadataCapabilityName},
	)
	if err != nil {
		t.Fatal(err)
	}
	mock.ExpectBegin()
	expectActiveInstallation(
		mock,
		7,
		testReleaseDigest,
		[]string{testIntegrationWorkload, testOtherWorkload},
		[]string{"runtime.records.query.v1"},
	)
	expectConsumerGrantAuthority(mock, 7, []catalogapp.ConsumerGrant{
		{
			AuthorizedWorkloadID: testIntegrationWorkload,
			CapabilityID:         "catalog.consumer.unknown.v1",
		},
		{
			AuthorizedWorkloadID: testOtherWorkload,
			CapabilityID:         consumerv1.AppMetadataCapabilityName,
		},
	})
	mock.ExpectRollback()
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	_, err = reader.GetAppMetadataView(ctx, scope, consumerv1.GetRequest{})
	var semantic *consumerv1.Failure
	if !errors.As(err, &semantic) ||
		semantic.Category != common.FailureInternal {
		t.Fatalf("corrupt committed consumer grant error = %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestGetAppMetadataViewFailsClosedOnActiveOperationRowError(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	reader, err := New(db, testSnapshotStore(t))
	if err != nil {
		t.Fatal(err)
	}
	scope, err := NewAuthorizedConsumerScope(
		testOrgID,
		testInstallationID,
		testOtherWorkload,
		[]string{consumerv1.AppMetadataCapabilityName},
	)
	if err != nil {
		t.Fatal(err)
	}
	mock.ExpectBegin()
	expectActiveInstallation(
		mock,
		7,
		testReleaseDigest,
		[]string{testOtherWorkload},
		[]string{"runtime.records.query.v1"},
	)
	mock.ExpectQuery("(?s)SELECT OperationId.*FROM _System_CatalogAppOperation").
		WithArgs(
			testInstallationID, testOrgID, testAppID,
			testReleaseDigest, "7",
		).
		WillReturnRows(
			sqlmock.NewRows([]string{"OperationId"}).
				AddRow(testOperationID).
				AddRow("018f0000-0000-7000-9000-000000000003").
				RowError(1, errors.New("operation scan interrupted")),
		)
	mock.ExpectRollback()
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	_, err = reader.GetAppMetadataView(ctx, scope, consumerv1.GetRequest{})
	var semantic *consumerv1.Failure
	if !errors.As(err, &semantic) ||
		semantic.Category != common.FailureInternal {
		t.Fatalf("active operation row error = %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestGetAppMetadataViewDoesNotClassifySelectionReadErrorAsMissing(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	reader, err := New(db, testSnapshotStore(t))
	if err != nil {
		t.Fatal(err)
	}
	scope, err := NewAuthorizedConsumerScope(
		testOrgID,
		testInstallationID,
		testOtherWorkload,
		[]string{consumerv1.AppMetadataCapabilityName},
	)
	if err != nil {
		t.Fatal(err)
	}
	mock.ExpectBegin()
	expectActiveInstallation(
		mock,
		7,
		testReleaseDigest,
		[]string{testOtherWorkload},
		[]string{"runtime.records.query.v1"},
	)
	expectConsumerGrantAuthority(mock, 7, []catalogapp.ConsumerGrant{{
		AuthorizedWorkloadID: testOtherWorkload,
		CapabilityID:         consumerv1.AppMetadataCapabilityName,
	}})
	mock.ExpectQuery("(?s)SELECT.*FROM _System_CatalogAppOperationJournal").
		WithArgs(testOperationID).
		WillReturnError(errors.New("metadata selection authority unavailable"))
	mock.ExpectRollback()
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	_, err = reader.GetAppMetadataView(
		ctx, scope, consumerv1.GetRequest{},
	)
	var semantic *consumerv1.Failure
	if !errors.As(err, &semantic) ||
		semantic.Category != common.FailureInternal {
		t.Fatalf("publication read error = %v, want internal", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestNextCatalogConsumerViewRejectsRevokedGrantWithoutRevisionAdvance(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	reader, err := NewWithWatchPollInterval(db, testSnapshotStore(t), time.Hour)
	if err != nil {
		t.Fatal(err)
	}
	scope, err := NewAuthorizedConsumerScope(
		testOrgID,
		testInstallationID,
		testOtherWorkload,
		[]string{consumerv1.AppMetadataCapabilityName},
	)
	if err != nil {
		t.Fatal(err)
	}
	expectWatchRevisionProbe(
		mock,
		7,
		testReleaseDigest,
		testOtherWorkload,
		consumerv1.AppMetadataCapabilityName,
		false,
	)
	after := uint64(7)
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	_, err = reader.NextAppMetadataView(
		ctx, scope,
		consumerv1.WatchRequest{AfterCatalogRevision: &after},
	)
	var semantic *consumerv1.Failure
	if !errors.As(err, &semantic) || semantic.Category != common.FailureForbidden {
		t.Fatalf("revoked watch grant error = %#v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestNextCatalogConsumerViewsRejectRevisionOverflowWithWireSafeShape(
	t *testing.T,
) {
	db, _, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	reader, err := New(db, testSnapshotStore(t))
	if err != nil {
		t.Fatal(err)
	}
	scope, err := NewAuthorizedConsumerScope(
		testOrgID,
		testInstallationID,
		testIntegrationWorkload,
		[]string{
			consumerv1.AppMetadataCapabilityName,
			consumerv1.IntegrationBindingCapabilityName,
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	after := uint64(^uint64(0))
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	for name, invoke := range map[string]func() error{
		"app metadata": func() error {
			_, err := reader.NextAppMetadataView(
				ctx,
				scope,
				consumerv1.WatchRequest{AfterCatalogRevision: &after},
			)
			return err
		},
		"integration binding": func() error {
			_, err := reader.NextIntegrationBindingView(
				ctx,
				scope,
				consumerv1.WatchRequest{AfterCatalogRevision: &after},
			)
			return err
		},
	} {
		t.Run(name, func(t *testing.T) {
			err := invoke()
			var semantic *consumerv1.Failure
			if !errors.As(err, &semantic) ||
				semantic.Category != common.FailureInvalidRequest ||
				semantic.InvalidComponent != common.InvalidComponentResponse ||
				semantic.CanonicalField != "" {
				t.Fatalf("revision overflow error = %#v", semantic)
			}
		})
	}
}

func TestNextCatalogConsumerViewsPollActivatedInstallationRevisionWhileUnchanged(
	t *testing.T,
) {
	tests := []struct {
		name       string
		capability string
		invoke     func(
			*Reader,
			context.Context,
			AuthorizedConsumerScope,
			consumerv1.WatchRequest,
		) error
	}{
		{
			name:       "AppMetadataV1",
			capability: consumerv1.AppMetadataCapabilityName,
			invoke: func(
				reader *Reader,
				ctx context.Context,
				scope AuthorizedConsumerScope,
				request consumerv1.WatchRequest,
			) error {
				_, err := reader.NextAppMetadataView(ctx, scope, request)
				return err
			},
		},
		{
			name:       "AppMetadataV2",
			capability: consumerv1.AppMetadataV2CapabilityName,
			invoke: func(
				reader *Reader,
				ctx context.Context,
				scope AuthorizedConsumerScope,
				request consumerv1.WatchRequest,
			) error {
				_, err := reader.NextAppMetadataViewV2(ctx, scope, request)
				return err
			},
		},
		{
			name:       "AppMetadataV3",
			capability: consumerv1.AppMetadataV3CapabilityName,
			invoke: func(
				reader *Reader,
				ctx context.Context,
				scope AuthorizedConsumerScope,
				request consumerv1.WatchRequest,
			) error {
				_, err := reader.NextAppMetadataViewV3(ctx, scope, request)
				return err
			},
		},
		{
			name:       "IntegrationBinding",
			capability: consumerv1.IntegrationBindingCapabilityName,
			invoke: func(
				reader *Reader,
				ctx context.Context,
				scope AuthorizedConsumerScope,
				request consumerv1.WatchRequest,
			) error {
				_, err := reader.NextIntegrationBindingView(ctx, scope, request)
				return err
			},
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			db, mock, err := sqlmock.New()
			if err != nil {
				t.Fatal(err)
			}
			defer db.Close()
			reader, err := NewWithWatchPollInterval(db, testSnapshotStore(t), time.Hour)
			if err != nil {
				t.Fatal(err)
			}
			scope, err := NewAuthorizedConsumerScope(
				testOrgID,
				testInstallationID,
				testIntegrationWorkload,
				[]string{test.capability},
			)
			if err != nil {
				t.Fatal(err)
			}
			expectWatchRevisionProbe(
				mock,
				7,
				testReleaseDigest,
				testIntegrationWorkload,
				test.capability,
				true,
			)
			after := uint64(7)
			ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
			defer cancel()
			err = test.invoke(
				reader,
				ctx,
				scope,
				consumerv1.WatchRequest{AfterCatalogRevision: &after},
			)
			var semantic *consumerv1.Failure
			if !errors.As(err, &semantic) ||
				semantic.Category != common.FailureDeadlineExceeded {
				t.Fatalf("unchanged watch cancellation error = %v", err)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatal(err)
			}
		})
	}
}

func TestNextCatalogViewInitialEmissionSkipsRevisionProbe(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	reader, err := New(db, testSnapshotStore(t))
	if err != nil {
		t.Fatal(err)
	}
	scope, err := NewAuthorizedConsumerScope(
		testOrgID,
		testInstallationID,
		testOtherWorkload,
		[]string{consumerv1.AppMetadataCapabilityName},
	)
	if err != nil {
		t.Fatal(err)
	}
	loadCalls := 0
	view, err := nextCatalogView(
		reader,
		context.Background(),
		scope,
		consumerv1.AppMetadataCapabilityName,
		0,
		func(
			_ context.Context,
			gotScope AuthorizedConsumerScope,
			request consumerv1.GetRequest,
		) (string, error) {
			loadCalls++
			if gotScope.appInstallationID != testInstallationID ||
				request.MinCatalogRevision == nil ||
				*request.MinCatalogRevision != 1 {
				t.Fatalf("initial load scope/request = %#v, %#v", gotScope, request)
			}
			return "initial-view", nil
		},
	)
	if err != nil || view != "initial-view" || loadCalls != 1 {
		t.Fatalf("initial view = %q, calls = %d, error = %v", view, loadCalls, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestNextCatalogViewInitialEmissionRetriesStaleStateWithoutRevisionProbe(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	reader, err := NewWithWatchPollInterval(db, testSnapshotStore(t), time.Nanosecond)
	if err != nil {
		t.Fatal(err)
	}
	scope, err := NewAuthorizedConsumerScope(
		testOrgID,
		testInstallationID,
		testOtherWorkload,
		[]string{consumerv1.AppMetadataCapabilityName},
	)
	if err != nil {
		t.Fatal(err)
	}
	loadCalls := 0
	view, err := nextCatalogView(
		reader,
		context.Background(),
		scope,
		consumerv1.AppMetadataCapabilityName,
		0,
		func(
			_ context.Context,
			_ AuthorizedConsumerScope,
			_ consumerv1.GetRequest,
		) (string, error) {
			loadCalls++
			if loadCalls == 1 {
				return "", &consumerv1.Failure{Category: common.FailureStaleState}
			}
			return "initial-view", nil
		},
	)
	if err != nil || view != "initial-view" || loadCalls != 2 {
		t.Fatalf("initial stale retry view = %q, calls = %d, error = %v", view, loadCalls, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestNextAppMetadataViewLoadsFullVerifiedViewAfterRevisionAdvance(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	reader, err := New(db, testSnapshotStore(t))
	if err != nil {
		t.Fatal(err)
	}
	scope, err := NewAuthorizedConsumerScope(
		testOrgID,
		testInstallationID,
		testOtherWorkload,
		[]string{consumerv1.AppMetadataCapabilityName},
	)
	if err != nil {
		t.Fatal(err)
	}
	verified := testVerifiedAppPackage(t)
	selection := testSelection(t, 8)
	expectWatchRevisionProbe(
		mock,
		8,
		verified.AppReleaseDigest,
		testOtherWorkload,
		consumerv1.AppMetadataCapabilityName,
		true,
	)
	mock.ExpectBegin()
	expectActiveInstallation(
		mock,
		8,
		verified.AppReleaseDigest,
		[]string{testOtherWorkload},
		[]string{"runtime.records.query.v1"},
	)
	expectConsumerGrantAuthority(mock, 8, []catalogapp.ConsumerGrant{{
		AuthorizedWorkloadID: testOtherWorkload,
		CapabilityID:         consumerv1.AppMetadataCapabilityName,
	}})
	expectActiveSelection(mock, selection)
	expectVerifiedAppRelease(mock, verified)
	mock.ExpectCommit()
	after := uint64(7)
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	view, err := reader.NextAppMetadataView(
		ctx,
		scope,
		consumerv1.WatchRequest{AfterCatalogRevision: &after},
	)
	if err != nil {
		t.Fatal(err)
	}
	if view.Lineage.CatalogRevision != 8 ||
		view.AppReleaseDigest != verified.AppReleaseDigest {
		t.Fatalf("advanced AppMetadataView = %#v", view)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestAuthorizedConsumerScopeRejectsRequestShapedAuthority(t *testing.T) {
	for _, test := range []struct {
		name         string
		orgID        string
		installation string
		workload     string
		capabilities []string
	}{
		{
			name:         "invalid installation",
			orgID:        testOrgID,
			installation: "request-installation",
			workload:     testIntegrationWorkload,
			capabilities: []string{consumerv1.IntegrationBindingCapabilityName},
		},
		{
			name:         "unsorted capabilities",
			orgID:        testOrgID,
			installation: testInstallationID,
			workload:     testIntegrationWorkload,
			capabilities: []string{"z", "a"},
		},
		{
			name:         "control character",
			orgID:        testOrgID,
			installation: testInstallationID,
			workload:     "workload\nforged",
			capabilities: []string{consumerv1.IntegrationBindingCapabilityName},
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			if _, err := NewAuthorizedConsumerScope(
				test.orgID,
				test.installation,
				test.workload,
				test.capabilities,
			); err == nil {
				t.Fatal("invalid authority scope was accepted")
			}
		})
	}
}

func TestGetAppMetadataViewReadsVerifiedImmutableRelease(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	reader, err := New(db, testSnapshotStore(t))
	if err != nil {
		t.Fatal(err)
	}
	scope, err := NewAuthorizedConsumerScope(
		testOrgID,
		testInstallationID,
		testOtherWorkload,
		[]string{consumerv1.AppMetadataCapabilityName},
	)
	if err != nil {
		t.Fatal(err)
	}
	verified := testVerifiedAppPackage(t)
	selection := testSelection(t, 7)
	mock.ExpectBegin()
	expectActiveInstallation(
		mock,
		7,
		verified.AppReleaseDigest,
		[]string{testOtherWorkload},
		[]string{"runtime.records.query.v1"},
	)
	expectConsumerGrantAuthority(mock, 7, []catalogapp.ConsumerGrant{{
		AuthorizedWorkloadID: testOtherWorkload,
		CapabilityID:         consumerv1.AppMetadataCapabilityName,
	}})
	expectActiveSelection(mock, selection)
	mock.ExpectQuery("(?s)SELECT.*FROM _System_CatalogAppRelease").
		WithArgs(testAppID, verified.AppReleaseDigest).
		WillReturnRows(sqlmock.NewRows([]string{
			"PublisherId", "SemanticVersion", "MetadataBundleDigest",
			"CanonicalManifest", "CanonicalBundle",
		}).AddRow(
			verified.Manifest.PublisherID,
			verified.Manifest.SemanticVersion,
			verified.Manifest.MetadataBundleDigest,
			verified.CanonicalManifest,
			verified.CanonicalBundle,
		))
	mock.ExpectQuery("(?s)SELECT ComponentId, CanonicalBehavior.*FROM _System_CatalogAppReleaseComponent").
		WithArgs(testAppID, verified.AppReleaseDigest).
		WillReturnRows(sqlmock.NewRows([]string{
			"ComponentId", "CanonicalBehavior",
		}).AddRow(
			verified.ComponentBlobs[0].ComponentID,
			verified.ComponentBlobs[0].CanonicalBehavior,
		))
	mock.ExpectCommit()
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	view, err := reader.GetAppMetadataView(
		ctx, scope, consumerv1.GetRequest{},
	)
	if err != nil {
		t.Fatal(err)
	}
	if view.AppReleaseDigest != verified.AppReleaseDigest ||
		view.AppID != testAppID ||
		len(view.Components) != 1 ||
		view.Components[0].ComponentID != verified.Bundle.Components[0].ComponentID ||
		string(view.Components[0].CanonicalBehavior) != `{"fields":["Name"]}` {
		t.Fatalf("unexpected AppMetadataView: %#v", view)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestGetAppMetadataViewV2ReadsHistoricalObjectSearchFormatOne(
	t *testing.T,
) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	reader, err := New(db, testSnapshotStore(t))
	if err != nil {
		t.Fatal(err)
	}
	scope, err := NewAuthorizedConsumerScope(
		testOrgID,
		testInstallationID,
		testOtherWorkload,
		[]string{consumerv1.AppMetadataV2CapabilityName},
	)
	if err != nil {
		t.Fatal(err)
	}
	verified := testVerifiedAppPackage(t)
	selection := testSelection(t, 7)
	effective := testRequiredEffectiveMetadataComponents()
	effective[1].ComponentFormatVersion = 1
	setDigest, err := catalogapp.ComputeEffectiveMetadataSetDigest(
		effective,
	)
	if err != nil {
		t.Fatal(err)
	}

	mock.ExpectBegin()
	expectActiveInstallation(
		mock, 7, verified.AppReleaseDigest,
		[]string{testOtherWorkload},
		[]string{"runtime.records.query.v1"},
	)
	expectConsumerGrantAuthority(mock, 7, []catalogapp.ConsumerGrant{{
		AuthorizedWorkloadID: testOtherWorkload,
		CapabilityID:         consumerv1.AppMetadataV2CapabilityName,
	}})
	expectActiveSelection(mock, selection)
	expectVerifiedAppRelease(mock, verified)
	mock.ExpectQuery(
		"(?s)SELECT.*FROM _System_CatalogEffectiveMetadataSet",
	).WithArgs(testOperationID).WillReturnRows(sqlmock.NewRows([]string{
		"OrgId", "AppInstallationId", "CatalogSnapshotDigest",
		"ComponentSetDigest", "ComponentCount",
	}).AddRow(
		testOrgID, testInstallationID,
		selection.CatalogSnapshotDigest, setDigest, []byte("2"),
	))
	mock.ExpectQuery(
		"(?s)SELECT.*FROM _System_CatalogEffectiveMetadataComponent",
	).WithArgs(testOperationID).WillReturnRows(sqlmock.NewRows([]string{
		"OperationId", "OrgId", "AppInstallationId", "ComponentKind",
		"ComponentFormatVersion", "BusinessKey", "ContentDigest",
		"CanonicalBehavior",
	}).AddRow(
		testOperationID, testOrgID, testInstallationID,
		effective[0].ComponentKind,
		[]byte(strconv.FormatUint(uint64(effective[0].ComponentFormatVersion), 10)),
		effective[0].BusinessKey,
		effective[0].ContentDigest, effective[0].CanonicalBehavior,
	).AddRow(
		testOperationID, testOrgID, testInstallationID,
		effective[1].ComponentKind,
		[]byte(strconv.FormatUint(uint64(effective[1].ComponentFormatVersion), 10)),
		effective[1].BusinessKey,
		effective[1].ContentDigest, effective[1].CanonicalBehavior,
	))
	mock.ExpectCommit()
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	view, err := reader.GetAppMetadataViewV2(
		ctx, scope, consumerv1.GetRequest{},
	)
	if err != nil {
		t.Fatal(err)
	}
	if view.Lineage.SourceSnapshotDigest !=
		selection.CatalogSnapshotDigest ||
		view.Receipts.CapabilityVersion !=
			consumerv1.AppMetadataV2CapabilityName ||
		len(view.AppComponents) != 1 ||
		len(view.EffectiveComponents) != 2 ||
		view.EffectiveComponents[1].ComponentFormatVersion != 1 ||
		string(view.EffectiveComponents[0].CanonicalBehavior) !=
			`{"applications":[]}` ||
		string(view.EffectiveComponents[1].CanonicalBehavior) !=
			`{"objects":[]}` {
		t.Fatalf("unexpected AppMetadataViewV2: %#v", view)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestGetAppMetadataViewV3UsesV3ComponentCeiling(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	reader, err := New(db, testSnapshotStore(t))
	if err != nil {
		t.Fatal(err)
	}
	scope, err := NewAuthorizedConsumerScope(
		testOrgID,
		testInstallationID,
		testOtherWorkload,
		[]string{consumerv1.AppMetadataV3CapabilityName},
	)
	if err != nil {
		t.Fatal(err)
	}
	verified := testVerifiedAppPackage(t)
	selection := testSelection(t, 7)
	effective := testRequiredEffectiveMetadataComponents()
	effective[1].CanonicalBehavior = []byte(strings.Repeat(
		"x", consumerv1.MaxSemanticResponseBytes+1,
	))
	effective[1].ContentDigest =
		consumerv1.ComputeEffectiveMetadataComponentDigest(
			effective[1].CanonicalBehavior,
		)
	setDigest, err := catalogapp.ComputeEffectiveMetadataSetDigest(effective)
	if err != nil {
		t.Fatal(err)
	}

	mock.ExpectBegin()
	expectActiveInstallation(
		mock, 7, verified.AppReleaseDigest,
		[]string{testOtherWorkload},
		[]string{"runtime.records.query.v1"},
	)
	expectConsumerGrantAuthority(mock, 7, []catalogapp.ConsumerGrant{{
		AuthorizedWorkloadID: testOtherWorkload,
		CapabilityID:         consumerv1.AppMetadataV3CapabilityName,
	}})
	expectActiveSelection(mock, selection)
	expectVerifiedAppRelease(mock, verified)
	mock.ExpectQuery(
		"(?s)SELECT.*FROM _System_CatalogEffectiveMetadataSet",
	).WithArgs(testOperationID).WillReturnRows(sqlmock.NewRows([]string{
		"OrgId", "AppInstallationId", "CatalogSnapshotDigest",
		"ComponentSetDigest", "ComponentCount",
	}).AddRow(
		testOrgID, testInstallationID,
		selection.CatalogSnapshotDigest, setDigest, []byte("2"),
	))
	mock.ExpectQuery(
		"(?s)SELECT.*FROM _System_CatalogEffectiveMetadataComponent",
	).WithArgs(testOperationID).WillReturnRows(sqlmock.NewRows([]string{
		"OperationId", "OrgId", "AppInstallationId", "ComponentKind",
		"ComponentFormatVersion", "BusinessKey", "ContentDigest",
		"CanonicalBehavior",
	}).AddRow(
		testOperationID, testOrgID, testInstallationID,
		effective[0].ComponentKind,
		[]byte(strconv.FormatUint(uint64(effective[0].ComponentFormatVersion), 10)),
		effective[0].BusinessKey,
		effective[0].ContentDigest, effective[0].CanonicalBehavior,
	).AddRow(
		testOperationID, testOrgID, testInstallationID,
		effective[1].ComponentKind,
		[]byte(strconv.FormatUint(uint64(effective[1].ComponentFormatVersion), 10)),
		effective[1].BusinessKey,
		effective[1].ContentDigest, effective[1].CanonicalBehavior,
	))
	mock.ExpectCommit()

	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	view, err := reader.GetAppMetadataViewV3(
		ctx, scope, consumerv1.GetRequest{},
	)
	if err != nil {
		t.Fatal(err)
	}
	if view.Receipts.CapabilityVersion !=
		consumerv1.AppMetadataV3CapabilityName ||
		len(view.EffectiveComponents) != 2 ||
		view.EffectiveComponents[1].ComponentFormatVersion != 2 ||
		len(view.EffectiveComponents[1].CanonicalBehavior) !=
			consumerv1.MaxSemanticResponseBytes+1 {
		t.Fatalf("unexpected AppMetadataViewV3: %#v", view)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestEffectiveMetadataTailComponentIsCapabilityBound(t *testing.T) {
	flowBehavior, err := consumerv1.EncodeEffectiveStandardFlowDocument(
		"1", []byte(`{"status":"Active"}`),
	)
	if err != nil {
		t.Fatal(err)
	}
	flow := catalogapp.EffectiveMetadataComponent{
		ComponentKind:          consumerv1.EffectiveStandardFlowComponentKind,
		ComponentFormatVersion: 1, CanonicalBehavior: flowBehavior,
	}
	if err := validateEffectiveMetadataTailComponent(
		flow, consumerv1.AppMetadataV2CapabilityName,
	); err != nil {
		t.Fatalf("V2 rejected effective Flow: %v", err)
	}
	layout := catalogapp.EffectiveMetadataComponent{
		ComponentKind:          consumerv1.EffectivePageLayoutComponentKind,
		ComponentFormatVersion: 1,
		CanonicalBehavior:      []byte(`{"objectApiName":"Lead","layoutName":"Default"}`),
	}
	if err := validateEffectiveMetadataTailComponent(
		layout, consumerv1.AppMetadataV3CapabilityName,
	); err != nil {
		t.Fatalf("V3 rejected effective page layout: %v", err)
	}
	if err := validateEffectiveMetadataTailComponent(
		layout, consumerv1.AppMetadataV2CapabilityName,
	); err == nil {
		t.Fatal("V2 accepted a V3-only effective page layout")
	}
	layout.CanonicalBehavior = []byte(`[]`)
	if err := validateEffectiveMetadataTailComponent(
		layout, consumerv1.AppMetadataV3CapabilityName,
	); err == nil {
		t.Fatal("V3 accepted a non-object page layout")
	}
	opaque := catalogapp.EffectiveMetadataComponent{
		ComponentKind:          "metadata.effective.object-action.v1",
		ComponentFormatVersion: 1,
		CanonicalBehavior:      []byte(`{"objectApiName":"Contract","actions":[]}`),
	}
	if err := validateEffectiveMetadataTailComponent(
		opaque, consumerv1.AppMetadataV3CapabilityName,
	); err != nil {
		t.Fatalf("V3 rejected a contract-validated opaque component: %v", err)
	}
	if err := validateEffectiveMetadataTailComponent(
		opaque, consumerv1.AppMetadataV2CapabilityName,
	); err == nil {
		t.Fatal("V2 accepted a V3 opaque component")
	}
}

func TestEffectiveMetadataRequiredComponentsAreLocatedByIdentity(t *testing.T) {
	components := testRequiredEffectiveMetadataComponents()
	opaque := catalogapp.EffectiveMetadataComponent{
		ComponentKind:          "metadata.effective.object-action.v1",
		ComponentFormatVersion: 1,
		BusinessKey:            "Contract",
		CanonicalBehavior:      []byte(`{"objectApiName":"Contract","actions":[]}`),
	}
	components = append(components, opaque)
	sort.Slice(components, func(left, right int) bool {
		leftKey := components[left].ComponentKind + "\x00" + components[left].BusinessKey
		rightKey := components[right].ComponentKind + "\x00" + components[right].BusinessKey
		return leftKey < rightKey
	})
	if components[1].ComponentKind != opaque.ComponentKind {
		t.Fatalf("fixture does not place opaque component between required kinds: %#v", components)
	}
	if err := validateEffectiveMetadataComponentSet(
		components, consumerv1.AppMetadataV3CapabilityName,
	); err != nil {
		t.Fatalf("V3 rejected identity-located required components: %v", err)
	}
	if err := validateEffectiveMetadataComponentSet(
		components, consumerv1.AppMetadataV2CapabilityName,
	); err == nil {
		t.Fatal("V2 accepted the V3-only component set")
	}
}

func TestEffectiveMetadataRequiredComponentsAcceptHistoricalObjectSearchFormat(
	t *testing.T,
) {
	components := testRequiredEffectiveMetadataComponents()
	components[1].ComponentFormatVersion = 1
	if err := validateEffectiveMetadataComponentSet(
		components, consumerv1.AppMetadataV3CapabilityName,
	); err != nil {
		t.Fatalf("historical object-search format 1 error = %v", err)
	}
}

func TestEffectiveMetadataRequiredComponentsRejectUnknownObjectSearchFormat(
	t *testing.T,
) {
	components := testRequiredEffectiveMetadataComponents()
	components[1].ComponentFormatVersion = 7
	if err := validateEffectiveMetadataComponentSet(
		components, consumerv1.AppMetadataV3CapabilityName,
	); err == nil {
		t.Fatal("object-search format 7 was accepted")
	}
}

func TestEffectiveMetadataObjectSearchFormatThreeIsV3Only(t *testing.T) {
	components := testRequiredEffectiveMetadataComponents()
	components[1].ComponentFormatVersion = 3
	if err := validateEffectiveMetadataComponentSet(
		components, consumerv1.AppMetadataV3CapabilityName,
	); err != nil {
		t.Fatalf("V3 rejected object-search format 3: %v", err)
	}
	if err := validateEffectiveMetadataComponentSet(
		components, consumerv1.AppMetadataV2CapabilityName,
	); err == nil {
		t.Fatal("V2 accepted object-search format 3")
	}
}

func TestEffectiveMetadataObjectSearchFormatFourIsV3Only(t *testing.T) {
	components := testRequiredEffectiveMetadataComponents()
	components[1].ComponentFormatVersion = 4
	if err := validateEffectiveMetadataComponentSet(
		components, consumerv1.AppMetadataV3CapabilityName,
	); err != nil {
		t.Fatalf("V3 rejected object-search format 4: %v", err)
	}
	if err := validateEffectiveMetadataComponentSet(
		components, consumerv1.AppMetadataV2CapabilityName,
	); err == nil {
		t.Fatal("V2 accepted object-search format 4")
	}
}

func TestEffectiveMetadataObjectSearchFormatFiveIsV3Only(t *testing.T) {
	components := testRequiredEffectiveMetadataComponents()
	components[1].ComponentFormatVersion = 5
	if err := validateEffectiveMetadataComponentSet(
		components, consumerv1.AppMetadataV3CapabilityName,
	); err != nil {
		t.Fatalf("V3 rejected object-search format 5: %v", err)
	}
	if err := validateEffectiveMetadataComponentSet(
		components, consumerv1.AppMetadataV2CapabilityName,
	); err == nil {
		t.Fatal("V2 accepted object-search format 5")
	}
}

func TestEffectiveMetadataObjectSearchFormatSixIsV3Only(t *testing.T) {
	components := testRequiredEffectiveMetadataComponents()
	components[1].ComponentFormatVersion = 6
	if err := validateEffectiveMetadataComponentSet(components, consumerv1.AppMetadataV3CapabilityName); err != nil {
		t.Fatalf("V3 rejected object-search format 6: %v", err)
	}
	if err := validateEffectiveMetadataComponentSet(components, consumerv1.AppMetadataV2CapabilityName); err == nil {
		t.Fatal("V2 accepted object-search format 6")
	}
}

func testRequiredEffectiveMetadataComponents() []catalogapp.EffectiveMetadataComponent {
	applicationNavigation := catalogapp.EffectiveMetadataComponent{
		ComponentKind:          consumerv1.EffectiveApplicationNavigationComponentKind,
		ComponentFormatVersion: 1,
		BusinessKey:            consumerv1.EffectiveApplicationNavigationComponentBusinessKey,
		CanonicalBehavior:      []byte(`{"applications":[]}`),
	}
	applicationNavigation.ContentDigest =
		consumerv1.ComputeEffectiveMetadataComponentDigest(
			applicationNavigation.CanonicalBehavior,
		)
	objectSearch := catalogapp.EffectiveMetadataComponent{
		ComponentKind:          consumerv1.EffectiveObjectSearchComponentKind,
		ComponentFormatVersion: 2,
		BusinessKey:            consumerv1.EffectiveObjectSearchComponentBusinessKey,
		CanonicalBehavior:      []byte(`{"objects":[]}`),
	}
	objectSearch.ContentDigest =
		consumerv1.ComputeEffectiveMetadataComponentDigest(
			objectSearch.CanonicalBehavior,
		)
	return []catalogapp.EffectiveMetadataComponent{
		applicationNavigation,
		objectSearch,
	}
}

func TestReadVerifiedEffectiveMetadataFailsClosedOnMissingOrMixedRevision(
	t *testing.T,
) {
	for _, test := range []struct {
		name         string
		rows         *sqlmock.Rows
		wantCategory common.FailureCategory
	}{
		{
			name: "missing",
			rows: sqlmock.NewRows([]string{
				"OrgId", "AppInstallationId", "CatalogSnapshotDigest",
				"ComponentSetDigest", "ComponentCount",
			}),
			wantCategory: common.FailureMissing,
		},
		{
			name: "mixed snapshot",
			rows: sqlmock.NewRows([]string{
				"OrgId", "AppInstallationId", "CatalogSnapshotDigest",
				"ComponentSetDigest", "ComponentCount",
			}).AddRow(
				testOrgID, testInstallationID, testSHA256("9"),
				testSHA256("8"), []byte("1"),
			),
			wantCategory: common.FailureInternal,
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			db, mock, err := sqlmock.New()
			if err != nil {
				t.Fatal(err)
			}
			defer db.Close()
			mock.ExpectBegin()
			mock.ExpectQuery(
				"(?s)SELECT.*FROM _System_CatalogEffectiveMetadataSet",
			).WithArgs(testOperationID).WillReturnRows(test.rows)
			mock.ExpectRollback()
			tx, err := db.BeginTx(context.Background(), nil)
			if err != nil {
				t.Fatal(err)
			}
			_, err = readVerifiedEffectiveMetadataComponents(
				context.Background(), tx, activeContext{
					orgID:                 testOrgID,
					appInstallationID:     testInstallationID,
					operationID:           testOperationID,
					catalogSnapshotDigest: testSHA256("1"),
				}, consumerv1.AppMetadataV2CapabilityName,
			)
			var semantic *consumerv1.Failure
			if !errors.As(err, &semantic) ||
				semantic.Category != test.wantCategory {
				t.Fatalf("error = %v, want %s", err, test.wantCategory)
			}
			_ = tx.Rollback()
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatal(err)
			}
		})
	}
}

func expectVerifiedAppRelease(
	mock sqlmock.Sqlmock,
	verified appauthority.VerifiedPackage,
) {
	mock.ExpectQuery("(?s)SELECT.*FROM _System_CatalogAppRelease").
		WithArgs(testAppID, verified.AppReleaseDigest).
		WillReturnRows(sqlmock.NewRows([]string{
			"PublisherId", "SemanticVersion", "MetadataBundleDigest",
			"CanonicalManifest", "CanonicalBundle",
		}).AddRow(
			verified.Manifest.PublisherID,
			verified.Manifest.SemanticVersion,
			verified.Manifest.MetadataBundleDigest,
			verified.CanonicalManifest,
			verified.CanonicalBundle,
		))
	mock.ExpectQuery(
		"(?s)SELECT ComponentId, CanonicalBehavior.*FROM _System_CatalogAppReleaseComponent",
	).WithArgs(testAppID, verified.AppReleaseDigest).
		WillReturnRows(sqlmock.NewRows([]string{
			"ComponentId", "CanonicalBehavior",
		}).AddRow(
			verified.ComponentBlobs[0].ComponentID,
			verified.ComponentBlobs[0].CanonicalBehavior,
		))
}

func expectConsumerGrantAuthority(
	mock sqlmock.Sqlmock,
	installationRevision uint64,
	grants []catalogapp.ConsumerGrant,
) {
	digest, err := catalogapp.ComputeConsumerGrantSetDigest(grants)
	if err != nil {
		panic(err)
	}
	mock.ExpectQuery("(?s)SELECT OperationId.*FROM _System_CatalogAppOperation").
		WithArgs(
			testInstallationID, testOrgID, testAppID,
			sqlmock.AnyArg(), strconv.FormatUint(installationRevision, 10),
		).
		WillReturnRows(
			sqlmock.NewRows([]string{"OperationId"}).AddRow(testOperationID),
		)
	mock.ExpectQuery("(?s)SELECT.*FROM _System_CatalogConsumerGrantSet").
		WithArgs(testOperationID).
		WillReturnRows(sqlmock.NewRows([]string{
			"AppInstallationId", "GrantSetDigest", "GrantCount",
		}).AddRow(testInstallationID, digest, []byte(strconv.Itoa(len(grants)))))
	rows := sqlmock.NewRows([]string{
		"OperationId", "OrgId", "AppInstallationId",
		"AuthorizedWorkloadId", "CapabilityId",
	})
	for _, grant := range grants {
		rows.AddRow(
			testOperationID,
			testOrgID,
			testInstallationID,
			grant.AuthorizedWorkloadID,
			grant.CapabilityID,
		)
	}
	mock.ExpectQuery("(?s)SELECT.*FROM _System_CatalogConsumerGrant").
		WithArgs(testOperationID).
		WillReturnRows(rows)
}

func expectActiveInstallation(
	mock sqlmock.Sqlmock,
	revision uint64,
	releaseDigest string,
	workloadValues []string,
	capabilityValues []string,
) {
	workloads, _ := json.Marshal(workloadValues)
	capabilities, _ := json.Marshal(capabilityValues)
	mock.ExpectQuery("(?s)SELECT.*FROM _System_CatalogAppInstallation").
		WithArgs(testInstallationID).
		WillReturnRows(sqlmock.NewRows([]string{
			"OrgId", "AppId", "State", "ActiveReleaseDigest",
			"CatalogRevision", "ActiveAuthorizedWorkloads",
			"ActiveGrantedCapabilities",
		}).AddRow(
			testOrgID,
			testAppID,
			"ACTIVE",
			releaseDigest,
			[]byte(strconv.FormatUint(revision, 10)),
			workloads,
			capabilities,
		))
}

func expectActiveSelection(mock sqlmock.Sqlmock, selection consumerTestSelection) {
	mock.ExpectQuery("(?s)SELECT.*FROM _System_CatalogAppOperationJournal").WithArgs(testOperationID).
		WillReturnRows(sqlmock.NewRows([]string{"EvidenceDigest", "ExpectedSecurityEpoch", "CatalogRevision", "LastSecurityEpoch"}).
			AddRow(selection.CatalogSnapshotDigest, []byte("17"), []byte(strconv.FormatUint(selection.CatalogRevision, 10)), []byte(strconv.FormatUint(selection.SecurityEpoch, 10))))
}

func expectWatchRevisionProbe(
	mock sqlmock.Sqlmock,
	installationRevision uint64,
	releaseDigest string,
	workloadID string,
	capability string,
	authorized bool,
) {
	workloads, _ := json.Marshal([]string{workloadID})
	capabilities, _ := json.Marshal([]string{"runtime.records.query.v1"})
	consumerGrantID := any(nil)
	if authorized {
		consumerGrantID = testSHA256("watch-consumer-grant")
	}
	mock.ExpectQuery(
		"(?s)SELECT.*FROM _System_CatalogAppInstallation i.*LEFT JOIN.*_System_CatalogAppOperation.*LEFT JOIN.*_System_CatalogConsumerGrant.*WHERE i.AppInstallationId = \\?",
	).WithArgs(
		workloadID,
		capability,
		testInstallationID,
	).WillReturnRows(sqlmock.NewRows([]string{
		"OrgId", "AppId", "State", "ActiveReleaseDigest",
		"CatalogRevision", "ActiveAuthorizedWorkloads",
		"ActiveGrantedCapabilities", "OperationId", "ConsumerGrantId",
	}).AddRow(
		testOrgID,
		testAppID,
		"ACTIVE",
		releaseDigest,
		[]byte(strconv.FormatUint(installationRevision, 10)),
		workloads,
		capabilities,
		testOperationID,
		consumerGrantID,
	))
}

type consumerTestSelection struct {
	CatalogRevision       uint64
	CatalogSnapshotDigest string
	SecurityEpoch         uint64
}

func testSnapshot(t *testing.T, orgID string, revision uint64) catalogsnapshot.Artifact {
	t.Helper()
	artifact, err := catalogsnapshot.CompileCatalogSnapshot(catalogsnapshot.CompileInput{
		OrgID: orgID, Lineage: catalogsnapshot.Lineage{CatalogRevision: catalogsnapshot.CatalogRevision(revision), SchemaGeneration: 11, EngineGeneration: 13},
		Objects:     []catalogsnapshot.ObjectInput{{CatalogRevision: catalogsnapshot.CatalogRevision(revision), Metadata: &models.ObjectMetadata{ID: "obj-issue", APIName: "Issue", Label: "Issue", PluralLabel: "Issues", SharingModel: constants.SharingModelPrivate, Fields: []models.FieldMetadata{{ID: "field-id", APIName: "Id", Label: "ID", Type: constants.FieldTypeReference, PhysicalName: "id", IsSystem: true}}}}},
		Projections: []catalogsnapshot.RuntimeProjectionInput{{CatalogRevision: catalogsnapshot.CatalogRevision(revision), AppInstallationID: testInstallationID, CapabilityID: "runtime.records.read-selected-root.v1", Profile: catalogsnapshot.ProjectionSelectedRootReadV1, ObjectAPINames: []string{"Issue"}}},
	})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := catalogsnapshot.OpenCatalogSnapshot(artifact); err != nil {
		t.Fatalf("snapshot fixture: %v", err)
	}
	return artifact
}

func testSelection(t *testing.T, revision uint64) consumerTestSelection {
	t.Helper()
	return consumerTestSelection{CatalogRevision: revision, CatalogSnapshotDigest: string(testSnapshot(t, testOrgID, revision).Digest()), SecurityEpoch: 17}
}

type consumerTestStore map[catalogsnapshot.Digest]catalogsnapshot.Artifact

func (store consumerTestStore) Put(_ context.Context, artifact catalogsnapshot.Artifact) error {
	store[artifact.Digest()] = artifact
	return nil
}
func (store consumerTestStore) Get(_ context.Context, _ catalogsnapshot.ArtifactKind, digest catalogsnapshot.Digest) (catalogsnapshot.Artifact, error) {
	if artifact, ok := store[digest]; ok {
		return artifact, nil
	}
	return catalogsnapshot.Artifact{}, catalogsnapshot.ErrArtifactNotFound
}
func testSnapshotStore(t *testing.T) consumerTestStore {
	t.Helper()
	store := consumerTestStore{}
	for _, revision := range []uint64{1, 2, 7, 8} {
		artifact := testSnapshot(t, testOrgID, revision)
		store[artifact.Digest()] = artifact
	}
	return store
}

func testBindingSelection(
	workloadID string,
	kind consumerv1.IntegrationBindingKind,
	businessKey string,
) catalogapp.IntegrationBindingSelection {
	return catalogapp.IntegrationBindingSelection{
		AuthorizedWorkloadID: workloadID,
		Binding: consumerv1.IntegrationBinding{
			Kind:               kind,
			OwnerPack:          "pack-integration",
			PackInstallationID: "install-pack-1",
			PackPayloadDigest:  testPackPayloadDigest,
			BusinessKey:        businessKey,
			DefinitionVersion:  1,
			CredentialRefs:     []string{},
		},
	}
}

func testSHA256(character string) string {
	return "sha256:" + strings.Repeat(character, 64)
}

func testVerifiedAppPackage(t *testing.T) appauthority.VerifiedPackage {
	return testVerifiedAppPackageForApp(t, testAppID)
}

func testVerifiedAppPackageForApp(t *testing.T, appID string) appauthority.VerifiedPackage {
	t.Helper()
	behavior := []byte(`{"fields":["Name"]}`)
	contentDigest, err := appauthority.ComponentDigest(
		"metadata.object.v1", 1, behavior,
	)
	if err != nil {
		t.Fatal(err)
	}
	component := appauthority.Component{
		ComponentKind:          "metadata.object.v1",
		ComponentFormatVersion: 1,
		NamespaceID:            appID,
		BusinessKey:            "Issue",
		DefinitionVersion:      "1",
		LocalDependencies:      []string{},
		ExternalComponentRefs:  []appauthority.ExternalComponentRef{},
		OverridePolicy:         appauthority.OverridePolicyPreserveAuthorized,
		RemovalPolicy:          appauthority.RemovalPolicyBlockIfReferencedOrOverridden,
		ContentDigest:          contentDigest,
	}
	component.ComponentID = appauthority.ComponentID(
		component.NamespaceID, component.ComponentKind, component.BusinessKey,
	)
	bundle := appauthority.MetadataBundle{
		SchemaVersion: appauthority.MetadataBundleSchemaVersion,
		Components:    []appauthority.Component{component},
	}
	manifest := appauthority.AppManifest{
		SchemaVersion:        appauthority.AppManifestSchemaVersion,
		AppID:                appID,
		PublisherID:          "io.example",
		SemanticVersion:      "1.0.0",
		MetadataBundleDigest: testSHA256("0"),
		Dependencies:         []appauthority.Dependency{},
		OwnedNamespaces:      []string{appID},
		RequiredProtocols: []appauthority.ProtocolRequirement{{
			ProtocolID: "nexuscrm.catalog.consumer.v1",
			Major:      1,
		}},
		RequiredEngineCapabilities: []string{consumerv1.AppMetadataCapabilityName, consumerv1.AppMetadataV2CapabilityName, consumerv1.IntegrationBindingCapabilityName, "runtime.records.query.v1"},
		ProvidedCapabilities:       []appauthority.ProvidedCapability{},
		FrontendCompatibility:      appauthority.FrontendCompatibility{Mode: appauthority.FrontendModeNone},
	}
	sort.Strings(manifest.RequiredEngineCapabilities)
	manifest.MetadataBundleDigest, err = appauthority.MetadataBundleDigest(
		bundle, manifest,
	)
	if err != nil {
		t.Fatal(err)
	}
	verified, err := appauthority.VerifyPackage(appauthority.Package{
		Manifest: manifest,
		Bundle:   bundle,
		ComponentBlobs: []appauthority.ComponentBlob{{
			ComponentID:       component.ComponentID,
			CanonicalBehavior: behavior,
		}},
	})
	if err != nil {
		t.Fatal(err)
	}
	return verified
}