// Package catalogconsumer projects authorized, immutable Catalog consumer
// views. It owns no metadata authoring or request authentication; callers must
// construct scope only from already-admitted workload authority.
package catalogconsumer

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"math"
	"slices"
	"sort"
	"strconv"
	"strings"
	"time"
	"unicode"
	"unicode/utf8"

	"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/layoutbundleslotv1"
	common "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commonv1"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/catalogconsumerauthority"
)

const (
	defaultWatchPollInterval    = 250 * time.Millisecond
	defaultWatchMaxPollInterval = time.Second
)

// AuthorizedConsumerScope is an opaque serving input minted from admitted
// authority. Its fields are deliberately private so request payloads cannot be
// assigned directly to Catalog scope.
type AuthorizedConsumerScope struct {
	orgID             string
	appInstallationID string
	workloadID        string
	capabilities      []string
}

// NewAuthorizedConsumerScope validates and defensively copies one admitted
// authority tuple. It does not authenticate the values; the gRPC boundary must
// call it only after workload admission.
func NewAuthorizedConsumerScope(
	orgID string,
	appInstallationID string,
	workloadID string,
	capabilities []string,
) (AuthorizedConsumerScope, error) {
	if !validOpaque(orgID, consumerv1.MaxIdentifierBytes) ||
		appauthority.ValidateAppInstallationID(appInstallationID) != nil ||
		!validOpaque(workloadID, consumerv1.MaxIdentifierBytes) {
		return AuthorizedConsumerScope{}, fmt.Errorf("Catalog consumer authority scope is invalid")
	}
	canonical := append([]string(nil), capabilities...)
	if !sort.StringsAreSorted(canonical) {
		return AuthorizedConsumerScope{}, fmt.Errorf("Catalog consumer authority capabilities must be sorted")
	}
	for index, capability := range canonical {
		if !validOpaque(capability, consumerv1.MaxIdentifierBytes) ||
			(index > 0 && capability == canonical[index-1]) {
			return AuthorizedConsumerScope{}, fmt.Errorf(
				"Catalog consumer authority capabilities must be valid and unique",
			)
		}
	}
	return AuthorizedConsumerScope{
		orgID:             orgID,
		appInstallationID: appInstallationID,
		workloadID:        workloadID,
		capabilities:      canonical,
	}, nil
}

// Reader serves views directly from Catalog's durable authorities.
type Reader struct {
	db              *sql.DB
	store           catalogsnapshot.Store
	pollInterval    time.Duration
	maxPollInterval time.Duration
}

func New(db *sql.DB, store catalogsnapshot.Store) (*Reader, error) {
	return newWithWatchPollIntervalRange(
		db, store,
		defaultWatchPollInterval,
		defaultWatchMaxPollInterval,
	)
}

func NewWithWatchPollInterval(db *sql.DB, store catalogsnapshot.Store, interval time.Duration) (*Reader, error) {
	return newWithWatchPollIntervalRange(db, store, interval, interval)
}

func newWithWatchPollIntervalRange(
	db *sql.DB,
	store catalogsnapshot.Store,
	interval time.Duration,
	maximum time.Duration,
) (*Reader, error) {
	if db == nil || store == nil {
		return nil, fmt.Errorf("Catalog consumer reader requires a database and snapshot store")
	}
	if interval <= 0 {
		return nil, fmt.Errorf("Catalog consumer watch poll interval must be positive")
	}
	if maximum < interval {
		return nil, fmt.Errorf("Catalog consumer watch maximum poll interval must not be shorter than its initial interval")
	}
	return &Reader{
		db:              db,
		store:           store,
		pollInterval:    interval,
		maxPollInterval: maximum,
	}, nil
}

type activeContext struct {
	orgID                 string
	appInstallationID     string
	appID                 string
	activeReleaseDigest   string
	installationRevision  uint64
	catalogRevision       uint64
	activeWorkloads       []string
	activeCapabilities    []string
	operationID           string
	catalogSnapshotDigest string
	schemaGeneration      uint64
	engineGeneration      uint64
	securityEpoch         uint64
}

// GetAppMetadataView reads one transaction-consistent authorized App view.
func (reader *Reader) GetAppMetadataView(
	ctx context.Context,
	scope AuthorizedConsumerScope,
	request consumerv1.GetRequest,
) (consumerv1.AppMetadataView, error) {
	if err := consumerv1.ValidateGetRequest(ctx, request); err != nil {
		return consumerv1.AppMetadataView{}, err
	}
	tx, err := reader.beginRead(ctx)
	if err != nil {
		return consumerv1.AppMetadataView{}, servingError(ctx, err)
	}
	defer rollback(tx)
	active, err := reader.readActiveContext(
		ctx, tx, scope, consumerv1.AppMetadataCapabilityName,
		request.MinCatalogRevision,
	)
	if err != nil {
		return consumerv1.AppMetadataView{}, err
	}
	components, err := readVerifiedAppReleaseComponents(ctx, tx, active)
	if err != nil {
		return consumerv1.AppMetadataView{}, servingError(ctx, err)
	}
	view := consumerv1.AppMetadataView{
		OrgID:             active.orgID,
		AppInstallationID: active.appInstallationID,
		AppID:             active.appID,
		AppReleaseDigest:  active.activeReleaseDigest,
		Lineage:           active.lineage(),
		Components:        components,
		Receipts:          active.receipts(consumerv1.AppMetadataCapabilityName),
	}
	view.Lineage.ViewDigest = consumerv1.ComputeAppMetadataViewDigest(view)
	if err := consumerv1.ValidateAppMetadataView(view); err != nil {
		return consumerv1.AppMetadataView{}, err
	}
	if err := tx.Commit(); err != nil {
		return consumerv1.AppMetadataView{}, servingError(ctx, err)
	}
	return view, nil
}

// GetAppMetadataViewV2 reads the AppRelease projection and the Catalog-derived
// effective projection under one repeatable-read transaction and one active
// Catalog snapshot lineage. There is no repository or global fallback.
func (reader *Reader) GetAppMetadataViewV2(
	ctx context.Context,
	scope AuthorizedConsumerScope,
	request consumerv1.GetRequest,
) (consumerv1.AppMetadataViewV2, error) {
	if err := consumerv1.ValidateGetRequest(ctx, request); err != nil {
		return consumerv1.AppMetadataViewV2{}, err
	}
	tx, err := reader.beginRead(ctx)
	if err != nil {
		return consumerv1.AppMetadataViewV2{}, servingError(ctx, err)
	}
	defer rollback(tx)
	active, err := reader.readActiveContext(
		ctx, tx, scope, consumerv1.AppMetadataV2CapabilityName,
		request.MinCatalogRevision,
	)
	if err != nil {
		return consumerv1.AppMetadataViewV2{}, err
	}
	appComponents, err := readVerifiedAppReleaseComponents(ctx, tx, active)
	if err != nil {
		return consumerv1.AppMetadataViewV2{}, servingError(ctx, err)
	}
	effectiveComponents, err := readVerifiedEffectiveMetadataComponents(
		ctx, tx, active, consumerv1.AppMetadataV2CapabilityName,
	)
	if err != nil {
		return consumerv1.AppMetadataViewV2{}, servingError(ctx, err)
	}
	view := consumerv1.AppMetadataViewV2{
		OrgID:               active.orgID,
		AppInstallationID:   active.appInstallationID,
		AppID:               active.appID,
		AppReleaseDigest:    active.activeReleaseDigest,
		Lineage:             active.lineage(),
		AppComponents:       appComponents,
		EffectiveComponents: effectiveComponents,
		Receipts: active.receipts(
			consumerv1.AppMetadataV2CapabilityName,
		),
	}
	view.Lineage.ViewDigest = consumerv1.ComputeAppMetadataViewV2Digest(view)
	if err := consumerv1.ValidateAppMetadataViewV2(view); err != nil {
		return consumerv1.AppMetadataViewV2{}, err
	}
	if err := tx.Commit(); err != nil {
		return consumerv1.AppMetadataViewV2{}, servingError(ctx, err)
	}
	return view, nil
}

// GetAppMetadataViewV3 reads the same immutable App and effective projection
// as V2, but validates and receipts it under the independently versioned V3
// capability identity and digest domain.
func (reader *Reader) GetAppMetadataViewV3(
	ctx context.Context,
	scope AuthorizedConsumerScope,
	request consumerv1.GetRequest,
) (consumerv1.AppMetadataViewV3, error) {
	if err := consumerv1.ValidateGetRequest(ctx, request); err != nil {
		return consumerv1.AppMetadataViewV3{}, err
	}
	tx, err := reader.beginRead(ctx)
	if err != nil {
		return consumerv1.AppMetadataViewV3{}, servingError(ctx, err)
	}
	defer rollback(tx)
	active, err := reader.readActiveContext(
		ctx, tx, scope, consumerv1.AppMetadataV3CapabilityName,
		request.MinCatalogRevision,
	)
	if err != nil {
		return consumerv1.AppMetadataViewV3{}, err
	}
	appComponents, err := readVerifiedAppReleaseComponents(ctx, tx, active)
	if err != nil {
		return consumerv1.AppMetadataViewV3{}, servingError(ctx, err)
	}
	effectiveComponents, err := readVerifiedEffectiveMetadataComponents(
		ctx, tx, active, consumerv1.AppMetadataV3CapabilityName,
	)
	if err != nil {
		return consumerv1.AppMetadataViewV3{}, servingError(ctx, err)
	}
	view := consumerv1.AppMetadataViewV3{
		OrgID:               active.orgID,
		AppInstallationID:   active.appInstallationID,
		AppID:               active.appID,
		AppReleaseDigest:    active.activeReleaseDigest,
		Lineage:             active.lineage(),
		AppComponents:       appComponents,
		EffectiveComponents: effectiveComponents,
		Receipts:            active.receipts(consumerv1.AppMetadataV3CapabilityName),
	}
	view.Lineage.ViewDigest = consumerv1.ComputeAppMetadataViewV3Digest(view)
	if err := consumerv1.ValidateAppMetadataViewV3(view); err != nil {
		return consumerv1.AppMetadataViewV3{}, err
	}
	if err := tx.Commit(); err != nil {
		return consumerv1.AppMetadataViewV3{}, servingError(ctx, err)
	}
	return view, nil
}

// GetIntegrationBindingView reads one complete operation-bound reference set,
// verifies its immutable pack authorities, then discloses only the admitted
// workload's bindings.
func (reader *Reader) GetIntegrationBindingView(
	ctx context.Context,
	scope AuthorizedConsumerScope,
	request consumerv1.GetRequest,
) (consumerv1.IntegrationBindingView, error) {
	if err := consumerv1.ValidateGetRequest(ctx, request); err != nil {
		return consumerv1.IntegrationBindingView{}, err
	}
	tx, err := reader.beginRead(ctx)
	if err != nil {
		return consumerv1.IntegrationBindingView{}, servingError(ctx, err)
	}
	defer rollback(tx)
	active, err := reader.readActiveContext(
		ctx, tx, scope, consumerv1.IntegrationBindingCapabilityName,
		request.MinCatalogRevision,
	)
	if err != nil {
		return consumerv1.IntegrationBindingView{}, err
	}
	bindings, err := readVerifiedIntegrationBindings(ctx, tx, active, scope.workloadID)
	if err != nil {
		return consumerv1.IntegrationBindingView{}, servingError(ctx, err)
	}
	view := consumerv1.IntegrationBindingView{
		OrgID:                active.orgID,
		AppInstallationID:    active.appInstallationID,
		AuthorizedWorkloadID: scope.workloadID,
		Lineage:              active.lineage(),
		Bindings:             bindings,
		Receipts: active.receipts(
			consumerv1.IntegrationBindingCapabilityName,
		),
	}
	view.Lineage.ViewDigest = consumerv1.ComputeIntegrationBindingViewDigest(view)
	if err := consumerv1.ValidateIntegrationBindingView(view); err != nil {
		return consumerv1.IntegrationBindingView{}, err
	}
	if err := tx.Commit(); err != nil {
		return consumerv1.IntegrationBindingView{}, servingError(ctx, err)
	}
	return view, nil
}

func (reader *Reader) beginRead(ctx context.Context) (*sql.Tx, error) {
	// TiDB rejects START TRANSACTION READ ONLY unless its noop compatibility
	// switch is enabled. Keep the real repeatable-read boundary; this package's
	// closed query inventory and API surface enforce that the transaction has
	// no write path.
	return reader.db.BeginTx(ctx, &sql.TxOptions{
		Isolation: sql.LevelRepeatableRead,
	})
}

type rowQuerier interface {
	QueryRowContext(context.Context, string, ...any) *sql.Row
}

func readActiveInstallation(
	ctx context.Context,
	querier rowQuerier,
	scope AuthorizedConsumerScope,
	capability string,
) (activeContext, error) {
	var (
		active             activeContext
		state              string
		releaseDigest      sql.NullString
		rawCatalogRevision []byte
		rawWorkloads       []byte
		rawCapabilities    []byte
	)
	err := querier.QueryRowContext(ctx, `SELECT
OrgId, AppId, State, ActiveReleaseDigest, CatalogRevision,
ActiveAuthorizedWorkloads, ActiveGrantedCapabilities
FROM _System_CatalogAppInstallation
WHERE AppInstallationId = ?`, scope.appInstallationID).Scan(
		&active.orgID,
		&active.appID,
		&state,
		&releaseDigest,
		&rawCatalogRevision,
		&rawWorkloads,
		&rawCapabilities,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return active, failure(common.FailureMissing)
	}
	if err != nil {
		return active, servingError(ctx, err)
	}
	return validateActiveInstallation(
		active,
		state,
		releaseDigest,
		rawCatalogRevision,
		rawWorkloads,
		rawCapabilities,
		scope,
		capability,
	)
}

func validateActiveInstallation(
	active activeContext,
	state string,
	releaseDigest sql.NullString,
	rawCatalogRevision []byte,
	rawWorkloads []byte,
	rawCapabilities []byte,
	scope AuthorizedConsumerScope,
	capability string,
) (activeContext, error) {
	active.appInstallationID = scope.appInstallationID
	if active.orgID != scope.orgID {
		return active, failure(common.FailureForbidden)
	}
	if state != catalogconsumerauthority.ActiveInstallationState || !releaseDigest.Valid {
		// A non-serving installation is not a usable revision. Do not expose
		// its lifecycle state or manufacture Catalog-stale details.
		return active, failure(common.FailureMissing)
	}
	active.activeReleaseDigest = releaseDigest.String
	if appauthority.ValidateAppID(active.appID) != nil ||
		appauthority.ValidateDigest(active.activeReleaseDigest) != nil {
		return active, failure(common.FailureInternal)
	}
	installationRevision, err := parseUint(
		"AppInstallation CatalogRevision", rawCatalogRevision,
	)
	if err != nil || installationRevision == 0 {
		return active, failure(common.FailureInternal)
	}
	active.installationRevision = installationRevision
	if err := json.Unmarshal(rawWorkloads, &active.activeWorkloads); err != nil {
		return active, failure(common.FailureInternal)
	}
	if err := json.Unmarshal(rawCapabilities, &active.activeCapabilities); err != nil {
		return active, failure(common.FailureInternal)
	}
	if !canonicalStrings(active.activeWorkloads) ||
		!canonicalStrings(active.activeCapabilities) {
		return active, failure(common.FailureInternal)
	}
	if !slices.Contains(scope.capabilities, capability) ||
		!slices.Contains(active.activeWorkloads, scope.workloadID) {
		return active, failure(common.FailureForbidden)
	}
	return active, nil
}

func (reader *Reader) readWatchRevision(
	ctx context.Context,
	scope AuthorizedConsumerScope,
	capability string,
) (uint64, error) {
	// This is one non-authoritative wakeup/liveness hint, not a view read. A
	// single statement snapshot is enough: every positive revision result is
	// followed by the complete repeatable-read load, which revalidates the full
	// grant-set digest, selected immutable snapshot, release and receipts.
	rows, err := reader.db.QueryContext(ctx, `SELECT
i.OrgId, i.AppId, i.State, i.ActiveReleaseDigest, i.CatalogRevision,
i.ActiveAuthorizedWorkloads, i.ActiveGrantedCapabilities,
o.OperationId, g.ConsumerGrantId
FROM _System_CatalogAppInstallation i
LEFT JOIN _System_CatalogAppOperation o
  ON o.AppInstallationId = i.AppInstallationId
 AND o.OrgId = i.OrgId
 AND o.AppId = i.AppId
 AND o.State = 'ACTIVE'
 AND o.Step = 'ACTIVATED'
 AND o.TargetReleaseDigest = i.ActiveReleaseDigest
 AND o.ExpectedCatalogRevision = i.CatalogRevision
LEFT JOIN _System_CatalogConsumerGrant g
  ON g.OperationId = o.OperationId
 AND g.OrgId = i.OrgId
 AND g.AppInstallationId = i.AppInstallationId
 AND g.AuthorizedWorkloadId = ?
 AND g.CapabilityId = ?
WHERE i.AppInstallationId = ?
ORDER BY BINARY o.OperationId
LIMIT 2`,
		scope.workloadID,
		capability,
		scope.appInstallationID,
	)
	if err != nil {
		return 0, servingError(ctx, err)
	}
	defer rows.Close()
	type probeRow struct {
		active                  activeContext
		state                   string
		releaseDigest           sql.NullString
		rawInstallationRevision []byte
		rawWorkloads            []byte
		rawCapabilities         []byte
		operationID             sql.NullString
		consumerGrantID         sql.NullString
	}
	var probes []probeRow
	for rows.Next() {
		var probe probeRow
		if err := rows.Scan(
			&probe.active.orgID,
			&probe.active.appID,
			&probe.state,
			&probe.releaseDigest,
			&probe.rawInstallationRevision,
			&probe.rawWorkloads,
			&probe.rawCapabilities,
			&probe.operationID,
			&probe.consumerGrantID,
		); err != nil {
			return 0, servingError(ctx, err)
		}
		probes = append(probes, probe)
	}
	if err := rows.Err(); err != nil {
		return 0, servingError(ctx, err)
	}
	if len(probes) == 0 {
		return 0, failure(common.FailureMissing)
	}
	if len(probes) != 1 {
		return 0, failure(common.FailureInternal)
	}
	probe := probes[0]
	active, err := validateActiveInstallation(
		probe.active,
		probe.state,
		probe.releaseDigest,
		probe.rawInstallationRevision,
		probe.rawWorkloads,
		probe.rawCapabilities,
		scope,
		capability,
	)
	if err != nil {
		return 0, err
	}
	if !probe.operationID.Valid {
		return 0, failure(common.FailureMissing)
	}
	if !probe.consumerGrantID.Valid {
		return 0, failure(common.FailureForbidden)
	}
	return active.installationRevision, nil
}

func readAuthorizedActiveInstallation(
	ctx context.Context,
	tx *sql.Tx,
	scope AuthorizedConsumerScope,
	capability string,
) (activeContext, error) {
	active, err := readActiveInstallation(ctx, tx, scope, capability)
	if err != nil {
		return active, err
	}
	active.operationID, err = readActiveOperationForRevision(ctx, tx, active)
	if err != nil {
		return active, servingError(ctx, err)
	}
	if err := verifyConsumerGrant(
		ctx, tx, active, scope.workloadID, capability,
	); err != nil {
		return active, servingError(ctx, err)
	}
	return active, nil
}

func (reader *Reader) readActiveContext(
	ctx context.Context,
	tx *sql.Tx,
	scope AuthorizedConsumerScope,
	capability string,
	minimum *uint64,
) (activeContext, error) {
	active, err := readAuthorizedActiveInstallation(ctx, tx, scope, capability)
	if err != nil {
		return active, err
	}

	// The activated operation owns the App's immutable metadata selection.
	// Layout publication is a different authority and is not a consumer input.
	// Read selection and committed security lineage in this same DB snapshot;
	// candidate operations cannot become visible before atomic activation.
	rows, err := tx.QueryContext(ctx, `SELECT
j.EvidenceDigest, o.ExpectedSecurityEpoch, s.CatalogRevision, s.LastSecurityEpoch
FROM _System_CatalogAppOperationJournal j
JOIN _System_CatalogAppOperation o ON o.OperationId = j.OperationId
JOIN _System_CatalogOrgState s ON s.OrgId = o.OrgId
WHERE j.OperationId = ? AND j.Step = 'METADATA_SELECTION_COMMITTED'
ORDER BY j.SequenceNumber
LIMIT 2`, active.operationID)
	if err != nil {
		return active, servingError(ctx, err)
	}
	var operationEpoch, orgRevision uint64
	count := 0
	for rows.Next() {
		count++
		var rawOperationEpoch, rawOrgRevision, rawSecurityEpoch []byte
		if err := rows.Scan(&active.catalogSnapshotDigest, &rawOperationEpoch, &rawOrgRevision, &rawSecurityEpoch); err != nil {
			rows.Close()
			return active, servingError(ctx, err)
		}
		operationEpoch, err = parseUint("operation SecurityEpoch", rawOperationEpoch)
		if err == nil {
			orgRevision, err = parseUint("committed org CatalogRevision", rawOrgRevision)
		}
		if err == nil {
			active.securityEpoch, err = parseUint("committed org SecurityEpoch", rawSecurityEpoch)
		}
		if err != nil {
			rows.Close()
			return active, failure(common.FailureInternal)
		}
	}
	if err := rows.Err(); err != nil {
		rows.Close()
		return active, servingError(ctx, err)
	}
	if err := rows.Close(); err != nil {
		return active, servingError(ctx, err)
	}
	if count != 1 || appauthority.ValidateDigest(active.catalogSnapshotDigest) != nil ||
		operationEpoch == 0 || active.securityEpoch < operationEpoch || orgRevision < active.installationRevision {
		return active, failure(common.FailureInternal)
	}
	artifact, err := reader.store.Get(ctx, catalogsnapshot.ArtifactCatalogSnapshot, catalogsnapshot.Digest(active.catalogSnapshotDigest))
	if err != nil {
		return active, servingError(ctx, err)
	}
	snapshot, err := catalogsnapshot.OpenCatalogSnapshot(artifact)
	if err != nil || string(artifact.Digest()) != active.catalogSnapshotDigest || snapshot.OrgID() != active.orgID {
		return active, failure(common.FailureInternal)
	}
	lineage := snapshot.Lineage()
	active.catalogRevision = uint64(lineage.CatalogRevision)
	active.schemaGeneration = uint64(lineage.SchemaGeneration)
	active.engineGeneration = uint64(lineage.EngineGeneration)
	if active.catalogRevision != active.installationRevision || active.schemaGeneration == 0 || active.engineGeneration == 0 {
		return active, failure(common.FailureInternal)
	}
	if minimum != nil && active.catalogRevision < *minimum {
		return active, catalogStaleFailure(active, capability, *minimum)
	}
	return active, nil
}

func readVerifiedAppReleaseComponents(
	ctx context.Context,
	tx *sql.Tx,
	active activeContext,
) ([]consumerv1.AppMetadataComponent, error) {
	var (
		publisherID     string
		semanticVersion string
		metadataDigest  string
		rawManifest     []byte
		rawBundle       []byte
	)
	err := tx.QueryRowContext(ctx, `SELECT
PublisherId, SemanticVersion, MetadataBundleDigest, CanonicalManifest,
CanonicalBundle
FROM _System_CatalogAppRelease
WHERE AppId = ? AND AppReleaseDigest = ?`,
		active.appID, active.activeReleaseDigest,
	).Scan(
		&publisherID, &semanticVersion, &metadataDigest, &rawManifest, &rawBundle,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, failure(common.FailureMissing)
	}
	if err != nil {
		return nil, err
	}
	manifest, err := appauthority.ParseAppManifest(rawManifest)
	if err != nil {
		return nil, failure(common.FailureInternal)
	}
	bundle, err := appauthority.ParseMetadataBundle(rawBundle, manifest)
	if err != nil {
		return nil, failure(common.FailureInternal)
	}
	rows, err := tx.QueryContext(ctx, `SELECT ComponentId, CanonicalBehavior
FROM _System_CatalogAppReleaseComponent
WHERE AppId = ? AND AppReleaseDigest = ?
ORDER BY ComponentId`, active.appID, active.activeReleaseDigest)
	if err != nil {
		return nil, err
	}
	var blobs []appauthority.ComponentBlob
	for rows.Next() {
		var blob appauthority.ComponentBlob
		if err := rows.Scan(&blob.ComponentID, &blob.CanonicalBehavior); err != nil {
			rows.Close()
			return nil, err
		}
		blobs = append(blobs, blob)
	}
	if err := rows.Err(); err != nil {
		rows.Close()
		return nil, err
	}
	if err := rows.Close(); err != nil {
		return nil, err
	}
	verified, _, err := layoutbundleslotv1.VerifyPackageDeclarations(
		appauthority.Package{
			Manifest:       manifest,
			Bundle:         bundle,
			ComponentBlobs: blobs,
		},
	)
	if err != nil ||
		verified.AppReleaseDigest != active.activeReleaseDigest ||
		publisherID != manifest.PublisherID ||
		semanticVersion != manifest.SemanticVersion ||
		metadataDigest != manifest.MetadataBundleDigest ||
		len(verified.Bundle.Components) != len(verified.ComponentBlobs) {
		return nil, failure(common.FailureInternal)
	}
	blobByID := make(map[string][]byte, len(verified.ComponentBlobs))
	for _, blob := range verified.ComponentBlobs {
		blobByID[blob.ComponentID] = append([]byte(nil), blob.CanonicalBehavior...)
	}
	components := make([]consumerv1.AppMetadataComponent, 0, len(verified.Bundle.Components))
	for _, component := range verified.Bundle.Components {
		behavior, found := blobByID[component.ComponentID]
		if !found {
			return nil, failure(common.FailureInternal)
		}
		components = append(components, consumerv1.AppMetadataComponent{
			ComponentID:            component.ComponentID,
			ComponentKind:          component.ComponentKind,
			ComponentFormatVersion: component.ComponentFormatVersion,
			NamespaceID:            component.NamespaceID,
			BusinessKey:            component.BusinessKey,
			DefinitionVersion:      component.DefinitionVersion,
			ContentDigest:          component.ContentDigest,
			CanonicalBehavior:      behavior,
		})
	}
	sort.Slice(components, func(left, right int) bool {
		return components[left].ComponentID < components[right].ComponentID
	})
	return components, nil
}

func readVerifiedEffectiveMetadataComponents(
	ctx context.Context,
	tx *sql.Tx,
	active activeContext,
	capabilityName string,
) ([]consumerv1.EffectiveMetadataComponent, error) {
	var (
		orgID             string
		appInstallationID string
		snapshotDigest    string
		setDigest         string
		rawCount          []byte
	)
	err := tx.QueryRowContext(ctx, `SELECT
OrgId, AppInstallationId, CatalogSnapshotDigest, ComponentSetDigest,
ComponentCount
FROM _System_CatalogEffectiveMetadataSet
WHERE OperationId = ?`, active.operationID).Scan(
		&orgID, &appInstallationID, &snapshotDigest, &setDigest, &rawCount,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, failure(common.FailureMissing)
	}
	if err != nil {
		return nil, err
	}
	if orgID != active.orgID ||
		appInstallationID != active.appInstallationID ||
		snapshotDigest != active.catalogSnapshotDigest {
		return nil, failure(common.FailureInternal)
	}
	count, err := parseUintAllowZero(
		"EffectiveMetadata ComponentCount", rawCount,
	)
	if err != nil {
		return nil, failure(common.FailureInternal)
	}
	rows, err := tx.QueryContext(ctx, `SELECT
OperationId, OrgId, AppInstallationId, ComponentKind,
ComponentFormatVersion, BusinessKey, ContentDigest, CanonicalBehavior
FROM _System_CatalogEffectiveMetadataComponent
WHERE OperationId = ?
ORDER BY BINARY ComponentKind, BINARY BusinessKey`, active.operationID)
	if err != nil {
		return nil, err
	}
	var components []consumerv1.EffectiveMetadataComponent
	previousKey := ""
	for rows.Next() {
		var (
			operationID      string
			rowOrgID         string
			rowInstallation  string
			value            consumerv1.EffectiveMetadataComponent
			rawFormatVersion []byte
		)
		if err := rows.Scan(
			&operationID, &rowOrgID, &rowInstallation,
			&value.ComponentKind, &rawFormatVersion, &value.BusinessKey,
			&value.ContentDigest, &value.CanonicalBehavior,
		); err != nil {
			rows.Close()
			return nil, err
		}
		formatVersion, err := parseUint(
			"EffectiveMetadata ComponentFormatVersion",
			rawFormatVersion,
		)
		if err != nil || formatVersion > uint64(^uint32(0)) {
			rows.Close()
			return nil, failure(common.FailureInternal)
		}
		value.ComponentFormatVersion = uint32(formatVersion)
		key := value.ComponentKind + "\x00" + value.BusinessKey
		if operationID != active.operationID ||
			rowOrgID != active.orgID ||
			rowInstallation != active.appInstallationID ||
			(previousKey != "" && key <= previousKey) ||
			consumerv1.ValidateEffectiveMetadataComponentForCapability(
				value, capabilityName,
			) != nil {
			rows.Close()
			return nil, failure(common.FailureInternal)
		}
		previousKey = key
		components = append(components, value)
	}
	if err := rows.Err(); err != nil {
		rows.Close()
		return nil, err
	}
	if err := rows.Close(); err != nil {
		return nil, err
	}
	if count != uint64(len(components)) ||
		validateEffectiveMetadataComponentSet(components, capabilityName) != nil {
		return nil, failure(common.FailureInternal)
	}
	currentSetDigest, err :=
		catalogconsumerauthority.ComputeEffectiveMetadataSetDigest(components)
	if err != nil || currentSetDigest != setDigest {
		return nil, failure(common.FailureInternal)
	}
	return components, nil
}

func validateEffectiveMetadataComponentSet(
	components []consumerv1.EffectiveMetadataComponent,
	capabilityName string,
) error {
	var applicationNavigation, objectSearch bool
	for _, component := range components {
		switch component.ComponentKind {
		case consumerv1.EffectiveApplicationNavigationComponentKind:
			if applicationNavigation ||
				component.BusinessKey !=
					consumerv1.EffectiveApplicationNavigationComponentBusinessKey ||
				component.ComponentFormatVersion != 1 {
				return failure(common.FailureInternal)
			}
			applicationNavigation = true
		case consumerv1.EffectiveObjectSearchComponentKind:
			if objectSearch ||
				component.BusinessKey !=
					consumerv1.EffectiveObjectSearchComponentBusinessKey ||
				!readerObjectSearchFormatAccepted(
					component.ComponentFormatVersion,
					capabilityName,
				) {
				return failure(common.FailureInternal)
			}
			objectSearch = true
		default:
			if err := validateEffectiveMetadataTailComponent(
				component, capabilityName,
			); err != nil {
				return err
			}
		}
	}
	if !applicationNavigation || !objectSearch {
		return failure(common.FailureInternal)
	}
	return nil
}

func readerObjectSearchFormatAccepted(format uint32, capabilityName string) bool {
	switch capabilityName {
	case consumerv1.AppMetadataV2CapabilityName:
		return format == 1 || format == 2
	case consumerv1.AppMetadataV3CapabilityName:
		return format == 1 || format == 2 || format == 3 || format == 4 || format == 5 || format == 6
	default:
		return false
	}
}

func validateEffectiveMetadataTailComponent(
	component consumerv1.EffectiveMetadataComponent,
	capabilityName string,
) error {
	if component.ComponentFormatVersion != 1 {
		return failure(common.FailureInternal)
	}
	switch component.ComponentKind {
	case consumerv1.EffectiveStandardFlowComponentKind:
		_, err := consumerv1.DecodeEffectiveStandardFlowDocument(
			component.CanonicalBehavior,
		)
		return err
	case consumerv1.EffectivePageLayoutComponentKind:
		if capabilityName != consumerv1.AppMetadataV3CapabilityName ||
			len(component.CanonicalBehavior) == 0 ||
			component.CanonicalBehavior[0] != '{' ||
			!json.Valid(component.CanonicalBehavior) {
			return failure(common.FailureInternal)
		}
		return nil
	case consumerv1.EffectiveStandardActionComponentKind:
		if capabilityName != consumerv1.AppMetadataV3CapabilityName ||
			len(component.CanonicalBehavior) == 0 ||
			component.CanonicalBehavior[0] != '[' ||
			!json.Valid(component.CanonicalBehavior) {
			return failure(common.FailureInternal)
		}
		return nil
	default:
		// AppMetadataV3 is intentionally extensible. The contract-level
		// validator above has already proven the opaque component's identity,
		// format, digest, size, and canonical bytes. A consumer reader must not
		// turn its local knowledge of selected component kinds into a second
		// closed metadata registry. V2 retains its historical closed surface.
		if capabilityName == consumerv1.AppMetadataV3CapabilityName {
			return nil
		}
		return failure(common.FailureInternal)
	}
}

func readVerifiedIntegrationBindings(
	ctx context.Context,
	tx *sql.Tx,
	active activeContext,
	workloadID string,
) ([]consumerv1.IntegrationBinding, error) {
	var (
		committedInstallationID string
		committedSetDigest      string
		rawCommittedCount       []byte
	)
	err := tx.QueryRowContext(ctx, `SELECT
AppInstallationId, BindingSetDigest, BindingCount
FROM _System_CatalogIntegrationBindingSet
WHERE OperationId = ?`, active.operationID).Scan(
		&committedInstallationID, &committedSetDigest, &rawCommittedCount,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, failure(common.FailureMissing)
	}
	if err != nil {
		return nil, err
	}
	if committedInstallationID != active.appInstallationID {
		return nil, failure(common.FailureInternal)
	}
	committedCount, err := parseUintAllowZero(
		"IntegrationBinding BindingCount", rawCommittedCount,
	)
	if err != nil {
		return nil, failure(common.FailureInternal)
	}

	rows, err := tx.QueryContext(ctx, `SELECT
b.OperationId, b.OrgId, b.AppInstallationId,
b.AuthorizedWorkloadId, b.Kind, b.OwnerPack, b.PackInstallId,
b.PackPayloadDigest, b.BusinessKey, b.DefinitionVersion, b.CredentialRefs,
b.CommandKind, b.CommandPayloadFormatVersion, b.CommandDefinitionRef,
b.CommandDefinitionVersion,
a.ActivePackInstallId, a.PreviousPackInstallId, a.ActivationEpoch,
a.ActivatedDate, COALESCE(a.IsDeleted, 0),
g.PackName, g.PackInstallId, g.Status, g.AppliedPayloadHash,
COALESCE(g.IsDeleted, 0)
FROM _System_CatalogIntegrationBinding AS b
LEFT JOIN _System_PackActivation AS a
  ON a.PackName = b.OwnerPack
LEFT JOIN _System_PackGeneration AS g
  ON g.PackInstallId = b.PackInstallId
WHERE b.OperationId = ?
ORDER BY BINARY b.AuthorizedWorkloadId, BINARY b.Kind, BINARY b.OwnerPack,
         BINARY b.PackInstallId, BINARY b.BusinessKey`,
		active.operationID,
	)
	if err != nil {
		return nil, err
	}
	var complete []catalogconsumerauthority.IntegrationBindingSelection
	previousSelectionKey := ""
	for rows.Next() {
		var (
			rowOperationID    string
			rowOrgID          string
			rowInstallationID string
			selection         catalogconsumerauthority.IntegrationBindingSelection
			rawKind           string
			rawVersion        []byte
			rawCredentials    []byte
			commandKind       sql.NullString
			rawCommandVersion []byte
			commandDefinitionRef,
			commandDefinitionVersion sql.NullString
			activePackInstallID sql.NullString
			previousPackInstall sql.NullString
			rawActivationEpoch  []byte
			activatedDate       sql.NullTime
			activationDeleted   int
			generationPackName  sql.NullString
			generationInstallID sql.NullString
			generationStatus    sql.NullString
			generationDigest    sql.NullString
			generationDeleted   int
		)
		if err := rows.Scan(
			&rowOperationID,
			&rowOrgID,
			&rowInstallationID,
			&selection.AuthorizedWorkloadID,
			&rawKind,
			&selection.Binding.OwnerPack,
			&selection.Binding.PackInstallationID,
			&selection.Binding.PackPayloadDigest,
			&selection.Binding.BusinessKey,
			&rawVersion,
			&rawCredentials,
			&commandKind,
			&rawCommandVersion,
			&commandDefinitionRef,
			&commandDefinitionVersion,
			&activePackInstallID,
			&previousPackInstall,
			&rawActivationEpoch,
			&activatedDate,
			&activationDeleted,
			&generationPackName,
			&generationInstallID,
			&generationStatus,
			&generationDigest,
			&generationDeleted,
		); err != nil {
			rows.Close()
			return nil, err
		}
		selection.Binding.Kind = consumerv1.IntegrationBindingKind(rawKind)
		selection.Binding.DefinitionVersion, err = parseUint(
			"IntegrationBinding DefinitionVersion", rawVersion,
		)
		if err != nil ||
			json.Unmarshal(rawCredentials, &selection.Binding.CredentialRefs) != nil {
			rows.Close()
			return nil, failure(common.FailureInternal)
		}
		selection.Binding.RuntimeCommand, err = integrationRuntimeCommandFromSQL(
			commandKind,
			rawCommandVersion,
			commandDefinitionRef,
			commandDefinitionVersion,
		)
		if err != nil {
			rows.Close()
			return nil, failure(common.FailureInternal)
		}
		if err := catalogconsumerauthority.ValidateIntegrationBindingSelection(selection); err != nil {
			rows.Close()
			return nil, failure(common.FailureInternal)
		}
		if _, err := parseUint(
			"PackActivation ActivationEpoch", rawActivationEpoch,
		); err != nil {
			rows.Close()
			return nil, failure(common.FailureMissing)
		}
		selectionKey := integrationSelectionKey(selection)
		if previousSelectionKey != "" && selectionKey <= previousSelectionKey {
			rows.Close()
			return nil, failure(common.FailureInternal)
		}
		previousSelectionKey = selectionKey
		if rowOperationID != active.operationID ||
			rowOrgID != active.orgID ||
			rowInstallationID != active.appInstallationID ||
			!activePackInstallID.Valid ||
			activePackInstallID.String != selection.Binding.PackInstallationID ||
			previousPackInstall.Valid ||
			!activatedDate.Valid ||
			activationDeleted != 0 ||
			!generationPackName.Valid ||
			generationPackName.String != selection.Binding.OwnerPack ||
			!generationInstallID.Valid ||
			generationInstallID.String != selection.Binding.PackInstallationID ||
			!generationStatus.Valid ||
			generationStatus.String != "active" ||
			!generationDigest.Valid ||
			generationDigest.String != selection.Binding.PackPayloadDigest ||
			generationDeleted != 0 {
			rows.Close()
			// The exact immutable authority selected by this App operation is
			// no longer present. Do not disclose whether another generation
			// exists, and do not mislabel pack identity as Catalog revision
			// staleness.
			return nil, failure(common.FailureMissing)
		}
		complete = append(complete, selection)
	}
	if err := rows.Err(); err != nil {
		rows.Close()
		return nil, err
	}
	if err := rows.Close(); err != nil {
		return nil, err
	}
	if committedCount != uint64(len(complete)) {
		return nil, failure(common.FailureInternal)
	}
	currentSetDigest, err := catalogconsumerauthority.ComputeIntegrationBindingSetDigest(complete)
	if err != nil || currentSetDigest != committedSetDigest {
		return nil, failure(common.FailureInternal)
	}
	var disclosed []consumerv1.IntegrationBinding
	for _, selection := range complete {
		if !slices.Contains(active.activeWorkloads, selection.AuthorizedWorkloadID) {
			return nil, failure(common.FailureInternal)
		}
		if selection.AuthorizedWorkloadID == workloadID {
			disclosed = append(disclosed, selection.Binding)
		}
	}
	return disclosed, nil
}

func integrationRuntimeCommandFromSQL(
	commandKind sql.NullString,
	rawPayloadFormatVersion []byte,
	definitionRef sql.NullString,
	definitionVersion sql.NullString,
) (*consumerv1.RuntimeCommandBinding, error) {
	present := 0
	for _, isPresent := range []bool{
		commandKind.Valid,
		len(rawPayloadFormatVersion) != 0,
		definitionRef.Valid,
		definitionVersion.Valid,
	} {
		if isPresent {
			present++
		}
	}
	if present == 0 {
		return nil, nil
	}
	if present != 4 {
		return nil, errors.New("IntegrationBinding Runtime command is incomplete")
	}
	payloadFormatVersion, err := parseUint(
		"IntegrationBinding CommandPayloadFormatVersion",
		rawPayloadFormatVersion,
	)
	if err != nil || payloadFormatVersion > math.MaxUint32 {
		return nil, errors.New("IntegrationBinding Runtime command version is invalid")
	}
	return &consumerv1.RuntimeCommandBinding{
		CommandKind:          commandKind.String,
		PayloadFormatVersion: uint32(payloadFormatVersion),
		DefinitionRef:        definitionRef.String,
		DefinitionVersion:    definitionVersion.String,
	}, nil
}

func readActiveOperationForRevision(
	ctx context.Context,
	tx *sql.Tx,
	active activeContext,
) (string, error) {
	rows, err := tx.QueryContext(ctx, `SELECT OperationId
FROM _System_CatalogAppOperation
WHERE AppInstallationId = ? AND OrgId = ? AND AppId = ?
AND State = 'ACTIVE' AND Step = 'ACTIVATED'
AND TargetReleaseDigest = ? AND ExpectedCatalogRevision = ?
ORDER BY OperationId
LIMIT 2`,
		active.appInstallationID,
		active.orgID,
		active.appID,
		active.activeReleaseDigest,
		strconv.FormatUint(active.installationRevision, 10),
	)
	if err != nil {
		return "", err
	}
	var operationIDs []string
	for rows.Next() {
		var operationID string
		if err := rows.Scan(&operationID); err != nil {
			rows.Close()
			return "", err
		}
		operationIDs = append(operationIDs, operationID)
	}
	if err := rows.Err(); err != nil {
		rows.Close()
		return "", err
	}
	if err := rows.Close(); err != nil {
		return "", err
	}
	if len(operationIDs) == 0 {
		return "", failure(common.FailureMissing)
	}
	if len(operationIDs) != 1 {
		return "", failure(common.FailureInternal)
	}
	return operationIDs[0], nil
}

func verifyConsumerGrant(
	ctx context.Context,
	tx *sql.Tx,
	active activeContext,
	workloadID string,
	capability string,
) error {
	var (
		committedInstallationID string
		committedSetDigest      string
		rawCommittedCount       []byte
	)
	err := tx.QueryRowContext(ctx, `SELECT
AppInstallationId, GrantSetDigest, GrantCount
FROM _System_CatalogConsumerGrantSet
WHERE OperationId = ?`, active.operationID).Scan(
		&committedInstallationID, &committedSetDigest, &rawCommittedCount,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return failure(common.FailureMissing)
	}
	if err != nil {
		return err
	}
	if committedInstallationID != active.appInstallationID {
		return failure(common.FailureInternal)
	}
	committedCount, err := parseUintAllowZero(
		"ConsumerGrant GrantCount", rawCommittedCount,
	)
	if err != nil {
		return failure(common.FailureInternal)
	}
	rows, err := tx.QueryContext(ctx, `SELECT
OperationId, OrgId, AppInstallationId, AuthorizedWorkloadId, CapabilityId
FROM _System_CatalogConsumerGrant
WHERE OperationId = ?
ORDER BY BINARY AuthorizedWorkloadId, BINARY CapabilityId`, active.operationID)
	if err != nil {
		return err
	}
	var complete []catalogconsumerauthority.ConsumerGrant
	authorized := false
	previousGrantKey := ""
	for rows.Next() {
		var (
			operationID    string
			orgID          string
			installationID string
			grant          catalogconsumerauthority.ConsumerGrant
		)
		if err := rows.Scan(
			&operationID,
			&orgID,
			&installationID,
			&grant.AuthorizedWorkloadID,
			&grant.CapabilityID,
		); err != nil {
			rows.Close()
			return err
		}
		if operationID != active.operationID ||
			orgID != active.orgID ||
			installationID != active.appInstallationID ||
			!slices.Contains(active.activeWorkloads, grant.AuthorizedWorkloadID) {
			rows.Close()
			return failure(common.FailureInternal)
		}
		if err := catalogconsumerauthority.ValidateConsumerGrant(grant); err != nil {
			rows.Close()
			return failure(common.FailureInternal)
		}
		grantKey := grant.AuthorizedWorkloadID + "\x00" + grant.CapabilityID
		if previousGrantKey != "" && grantKey <= previousGrantKey {
			rows.Close()
			return failure(common.FailureInternal)
		}
		previousGrantKey = grantKey
		if grant.AuthorizedWorkloadID == workloadID &&
			grant.CapabilityID == capability {
			authorized = true
		}
		complete = append(complete, grant)
	}
	if err := rows.Err(); err != nil {
		rows.Close()
		return err
	}
	if err := rows.Close(); err != nil {
		return err
	}
	if committedCount != uint64(len(complete)) {
		return failure(common.FailureInternal)
	}
	currentSetDigest, err := catalogconsumerauthority.ComputeConsumerGrantSetDigest(complete)
	if err != nil || currentSetDigest != committedSetDigest {
		return failure(common.FailureInternal)
	}
	if !authorized {
		return failure(common.FailureForbidden)
	}
	return nil
}

func integrationSelectionKey(
	value catalogconsumerauthority.IntegrationBindingSelection,
) string {
	return value.AuthorizedWorkloadID + "\x00" +
		string(value.Binding.Kind) + "\x00" +
		value.Binding.OwnerPack + "\x00" +
		value.Binding.PackInstallationID + "\x00" +
		value.Binding.BusinessKey
}

func (active activeContext) lineage() consumerv1.ViewLineage {
	return consumerv1.ViewLineage{
		CatalogRevision:      active.catalogRevision,
		SchemaGeneration:     active.schemaGeneration,
		EngineGeneration:     active.engineGeneration,
		SourceSnapshotDigest: active.catalogSnapshotDigest,
	}
}

func (active activeContext) receipts(capability string) common.Receipts {
	return common.Receipts{
		ProtocolVersion:   "1.0",
		CatalogRevision:   active.catalogRevision,
		SchemaGeneration:  active.schemaGeneration,
		EngineGeneration:  active.engineGeneration,
		SecurityEpoch:     active.securityEpoch,
		CapabilityVersion: capability,
	}
}

// NextAppMetadataView waits for and returns exactly one strictly newer
// transaction-consistent App view. A gRPC handler repeats this operation after
// each successful Send, so backpressure cannot accumulate hidden emissions.
func (reader *Reader) NextAppMetadataView(
	ctx context.Context,
	scope AuthorizedConsumerScope,
	request consumerv1.WatchRequest,
) (consumerv1.AppMetadataView, error) {
	if err := consumerv1.ValidateWatchRequest(ctx, request); err != nil {
		return consumerv1.AppMetadataView{}, err
	}
	after := uint64(0)
	if request.AfterCatalogRevision != nil {
		after = *request.AfterCatalogRevision
	}
	if after == math.MaxUint64 {
		return consumerv1.AppMetadataView{}, invalidRequestFailure()
	}
	return nextCatalogView(
		reader, ctx, scope, consumerv1.AppMetadataCapabilityName, after,
		reader.GetAppMetadataView,
	)
}

func (reader *Reader) NextAppMetadataViewV2(
	ctx context.Context,
	scope AuthorizedConsumerScope,
	request consumerv1.WatchRequest,
) (consumerv1.AppMetadataViewV2, error) {
	if err := consumerv1.ValidateWatchRequest(ctx, request); err != nil {
		return consumerv1.AppMetadataViewV2{}, err
	}
	after := uint64(0)
	if request.AfterCatalogRevision != nil {
		after = *request.AfterCatalogRevision
	}
	if after == math.MaxUint64 {
		return consumerv1.AppMetadataViewV2{}, invalidRequestFailure()
	}
	return nextCatalogView(
		reader, ctx, scope, consumerv1.AppMetadataV2CapabilityName, after,
		reader.GetAppMetadataViewV2,
	)
}

func (reader *Reader) NextAppMetadataViewV3(
	ctx context.Context,
	scope AuthorizedConsumerScope,
	request consumerv1.WatchRequest,
) (consumerv1.AppMetadataViewV3, error) {
	if err := consumerv1.ValidateWatchRequest(ctx, request); err != nil {
		return consumerv1.AppMetadataViewV3{}, err
	}
	after := uint64(0)
	if request.AfterCatalogRevision != nil {
		after = *request.AfterCatalogRevision
	}
	if after == math.MaxUint64 {
		return consumerv1.AppMetadataViewV3{}, invalidRequestFailure()
	}
	return nextCatalogView(
		reader, ctx, scope, consumerv1.AppMetadataV3CapabilityName, after,
		reader.GetAppMetadataViewV3,
	)
}

// NextIntegrationBindingView is the corresponding one-emission primitive for
// Integration binding watch streams.
func (reader *Reader) NextIntegrationBindingView(
	ctx context.Context,
	scope AuthorizedConsumerScope,
	request consumerv1.WatchRequest,
) (consumerv1.IntegrationBindingView, error) {
	if err := consumerv1.ValidateWatchRequest(ctx, request); err != nil {
		return consumerv1.IntegrationBindingView{}, err
	}
	after := uint64(0)
	if request.AfterCatalogRevision != nil {
		after = *request.AfterCatalogRevision
	}
	if after == math.MaxUint64 {
		return consumerv1.IntegrationBindingView{}, invalidRequestFailure()
	}
	return nextCatalogView(
		reader, ctx, scope, consumerv1.IntegrationBindingCapabilityName, after,
		reader.GetIntegrationBindingView,
	)
}

func nextCatalogView[T any](
	reader *Reader,
	ctx context.Context,
	scope AuthorizedConsumerScope,
	capability string,
	after uint64,
	load func(
		context.Context,
		AuthorizedConsumerScope,
		consumerv1.GetRequest,
	) (T, error),
) (T, error) {
	minimum := after + 1
	pollInterval := reader.pollInterval
	if after == 0 {
		for {
			view, err := load(
				ctx,
				scope,
				consumerv1.GetRequest{MinCatalogRevision: &minimum},
			)
			if err == nil {
				return view, nil
			}
			if !isFailureCategory(err, common.FailureStaleState) {
				var zero T
				return zero, err
			}
			if err := reader.waitForPoll(ctx, pollInterval); err != nil {
				var zero T
				return zero, err
			}
			pollInterval = nextWatchPollInterval(
				pollInterval,
				reader.maxPollInterval,
			)
		}
	}
	for {
		revision, err := reader.readWatchRevision(ctx, scope, capability)
		if err != nil {
			var zero T
			return zero, err
		}
		if revision > after {
			view, loadErr := load(
				ctx,
				scope,
				consumerv1.GetRequest{MinCatalogRevision: &minimum},
			)
			if loadErr == nil {
				return view, nil
			}
			if !isFailureCategory(loadErr, common.FailureStaleState) {
				var zero T
				return zero, loadErr
			}
		}
		if err := reader.waitForPoll(ctx, pollInterval); err != nil {
			var zero T
			return zero, err
		}
		pollInterval = nextWatchPollInterval(
			pollInterval,
			reader.maxPollInterval,
		)
	}
}

func nextWatchPollInterval(current, maximum time.Duration) time.Duration {
	if current >= maximum || current > maximum-current {
		return maximum
	}
	return current * 2
}

func (reader *Reader) waitForPoll(ctx context.Context, interval time.Duration) error {
	timer := time.NewTimer(interval)
	defer timer.Stop()
	select {
	case <-ctx.Done():
		return servingError(ctx, ctx.Err())
	case <-timer.C:
		return nil
	}
}

func canonicalStrings(values []string) bool {
	if !sort.StringsAreSorted(values) {
		return false
	}
	for index, value := range values {
		if !validOpaque(value, consumerv1.MaxIdentifierBytes) ||
			(index > 0 && value == values[index-1]) {
			return false
		}
	}
	return true
}

func validOpaque(value string, maximum int) bool {
	if value == "" || len(value) > maximum || !utf8.ValidString(value) ||
		strings.TrimSpace(value) != value {
		return false
	}
	for _, character := range value {
		if unicode.IsControl(character) {
			return false
		}
	}
	return true
}

func parseUint(label string, raw []byte) (uint64, error) {
	value, err := parseUintAllowZero(label, raw)
	if err != nil {
		return 0, err
	}
	if value == 0 {
		return 0, fmt.Errorf("%s must be positive", label)
	}
	return value, nil
}

func parseUintAllowZero(label string, raw []byte) (uint64, error) {
	if len(raw) == 0 {
		return 0, fmt.Errorf("%s is empty", label)
	}
	for _, character := range raw {
		if character < '0' || character > '9' {
			return 0, fmt.Errorf("%s is not an unsigned decimal", label)
		}
	}
	value, err := strconv.ParseUint(string(raw), 10, 64)
	if err != nil {
		return 0, fmt.Errorf("%s exceeds uint64: %w", label, err)
	}
	return value, nil
}

func failure(category common.FailureCategory) error {
	return &consumerv1.Failure{Category: category}
}

func invalidRequestFailure() error {
	return &consumerv1.Failure{
		Category:         common.FailureInvalidRequest,
		InvalidComponent: common.InvalidComponentResponse,
	}
}

func catalogStaleFailure(
	active activeContext,
	capability string,
	requiredRevision uint64,
) error {
	receipts := active.receipts(capability)
	return &consumerv1.Failure{
		Category:         common.FailureStaleState,
		Receipts:         &receipts,
		StaleAxis:        common.StaleAxisCatalog,
		RequiredRevision: requiredRevision,
		ServedRevision:   active.catalogRevision,
	}
}

func isFailureCategory(err error, category common.FailureCategory) bool {
	var failureValue *consumerv1.Failure
	return errors.As(err, &failureValue) && failureValue.Category == category
}

func servingError(ctx context.Context, err error) error {
	if err == nil {
		return nil
	}
	if errors.Is(err, context.Canceled) || errors.Is(ctx.Err(), context.Canceled) {
		return failure(common.FailureCanceled)
	}
	if errors.Is(err, context.DeadlineExceeded) ||
		errors.Is(ctx.Err(), context.DeadlineExceeded) {
		return failure(common.FailureDeadlineExceeded)
	}
	var semantic *consumerv1.Failure
	if errors.As(err, &semantic) {
		return err
	}
	return failure(common.FailureInternal)
}

func rollback(tx *sql.Tx) {
	if tx != nil {
		_ = tx.Rollback()
	}
}