// Package connectorbindingdefinitionauthority owns the one source-free read
// path for a complete immutable connector-binding owner definition set.
package connectorbindingdefinitionauthority

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"errors"
	"fmt"
	"strconv"

	consumerv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	connectorbindingownerv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/connectorbindingownerv1"
)

var (
	ErrPrecondition = errors.New("connector binding definition authority precondition failed")
	ErrConflict     = errors.New("connector binding definition authority conflict")
)

// Reader is the transaction-bound SQL surface required by the source-free
// definition-set verifier.
type Reader interface {
	QueryContext(context.Context, string, ...any) (*sql.Rows, error)
	QueryRowContext(context.Context, string, ...any) *sql.Row
}

type transactionReader interface {
	Reader
	Commit() error
	Rollback() error
}

// GenerationCoordinate identifies the exact immutable storage parent and its
// portable owner coordinate.
type GenerationCoordinate struct {
	StorageID          string
	OwnerPack          string
	PackInstallationID string
	PackPayloadDigest  string
}

// PackGenerationStorageID derives the durable storage identity for one exact
// pack generation.
func PackGenerationStorageID(ownerPack, packInstallationID string) string {
	return "packgen_" + hashStorageID(ownerPack+"\x00"+packInstallationID)
}

// DefinitionStorageID derives the durable storage identity for one child of a
// pack-generation definition set.
func DefinitionStorageID(
	packGenerationID string,
	kind consumerv1.IntegrationBindingKind,
	businessKey string,
) string {
	return "packgendef_" + hashStorageID(
		packGenerationID+"\x00"+string(kind)+"\x00"+businessKey,
	)
}

// ReadVerifiedDefinitionSet reads and verifies the explicit complete set for
// coordinate. It distinguishes a canonical empty set from a missing set and
// validates every stored child before returning any projected field.
func ReadVerifiedDefinitionSet(
	ctx context.Context,
	reader Reader,
	coordinate GenerationCoordinate,
) (connectorbindingownerv1.DefinitionSet, error) {
	if ctx == nil || reader == nil {
		return connectorbindingownerv1.DefinitionSet{}, fmt.Errorf(
			"%w: context and reader are required",
			ErrPrecondition,
		)
	}
	if _, ok := reader.(transactionReader); !ok {
		return connectorbindingownerv1.DefinitionSet{}, fmt.Errorf(
			"%w: definition authority reads require one SQL transaction",
			ErrPrecondition,
		)
	}
	portableCoordinate := connectorbindingownerv1.DefinitionSetCoordinate{
		OwnerPack:          coordinate.OwnerPack,
		PackInstallationID: coordinate.PackInstallationID,
		PackPayloadDigest:  coordinate.PackPayloadDigest,
	}
	if coordinate.StorageID != PackGenerationStorageID(
		coordinate.OwnerPack,
		coordinate.PackInstallationID,
	) {
		return connectorbindingownerv1.DefinitionSet{}, fmt.Errorf(
			"%w: pack %q generation storage identity is malformed",
			ErrPrecondition,
			coordinate.OwnerPack,
		)
	}
	if _, err := connectorbindingownerv1.NewDefinitionSet(portableCoordinate, nil); err != nil {
		return connectorbindingownerv1.DefinitionSet{}, fmt.Errorf(
			"%w: pack %q generation coordinate is malformed: %v",
			ErrPrecondition,
			coordinate.OwnerPack,
			err,
		)
	}
	if err := verifyPackGenerationParent(ctx, reader, coordinate); err != nil {
		return connectorbindingownerv1.DefinitionSet{}, err
	}

	var (
		storedGenerationID string
		protocol           string
		rawCount           []byte
		setDigest          string
	)
	// Keep the ordinary equality as the clustered-index access predicate while
	// the binary equality preserves the byte-exact authority boundary.
	err := reader.QueryRowContext(ctx, `SELECT
PackGenerationId, Protocol, DefinitionCount, DefinitionSetDigest
FROM _System_PackGenerationConnectorBindingDefinitionSet
WHERE PackGenerationId = ?
  AND BINARY PackGenerationId = BINARY ?
FOR UPDATE`, coordinate.StorageID, coordinate.StorageID).Scan(
		&storedGenerationID,
		&protocol,
		&rawCount,
		&setDigest,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return connectorbindingownerv1.DefinitionSet{}, fmt.Errorf(
			"%w: pack %q generation %q is missing its connector binding definition set",
			ErrConflict,
			coordinate.OwnerPack,
			coordinate.PackInstallationID,
		)
	}
	if err != nil {
		return connectorbindingownerv1.DefinitionSet{}, fmt.Errorf(
			"lock pack %q connector binding definition set: %w",
			coordinate.OwnerPack,
			err,
		)
	}
	if storedGenerationID != coordinate.StorageID {
		return connectorbindingownerv1.DefinitionSet{}, fmt.Errorf(
			"%w: pack %q connector binding definition set names a different generation",
			ErrConflict,
			coordinate.OwnerPack,
		)
	}
	definitionCount, err := parseCanonicalUint64(rawCount, true)
	if err != nil {
		return connectorbindingownerv1.DefinitionSet{}, fmt.Errorf(
			"%w: pack %q connector binding definition count is malformed: %v",
			ErrConflict,
			coordinate.OwnerPack,
			err,
		)
	}
	if definitionCount > connectorbindingownerv1.MaxDefinitionCount {
		return connectorbindingownerv1.DefinitionSet{}, fmt.Errorf(
			"%w: pack %q connector binding definition count exceeds %d",
			ErrConflict,
			coordinate.OwnerPack,
			connectorbindingownerv1.MaxDefinitionCount,
		)
	}

	rows, err := reader.QueryContext(ctx, `SELECT
Id, PackGenerationId, PackInstallId, DefinitionOrdinal, Kind, OwnerPack,
PackPayloadDigest, BusinessKey, DefinitionVersion, DefinitionEnvelopeDigest,
DefinitionEnvelope
FROM _System_PackGenerationConnectorBindingDefinition
WHERE PackGenerationId = ?
ORDER BY DefinitionOrdinal
FOR UPDATE`, coordinate.StorageID)
	if err != nil {
		return connectorbindingownerv1.DefinitionSet{}, fmt.Errorf(
			"lock pack %q connector binding definitions: %w",
			coordinate.OwnerPack,
			err,
		)
	}
	entries := make([]connectorbindingownerv1.DefinitionSetEntry, 0, definitionCount)
	definitionSetBytes := uint64(0)
	for rows.Next() {
		if len(entries) >= connectorbindingownerv1.MaxDefinitionCount {
			_ = rows.Close()
			return connectorbindingownerv1.DefinitionSet{}, fmt.Errorf(
				"%w: pack %q connector binding definition rows exceed %d",
				ErrConflict,
				coordinate.OwnerPack,
				connectorbindingownerv1.MaxDefinitionCount,
			)
		}
		var (
			id                 string
			packGenerationID   string
			rawOrdinal         []byte
			rawKind            string
			rawVersion         []byte
			definitionEnvelope sql.RawBytes
			entry              connectorbindingownerv1.DefinitionSetEntry
		)
		if err := rows.Scan(
			&id,
			&packGenerationID,
			&entry.PackInstallationID,
			&rawOrdinal,
			&rawKind,
			&entry.OwnerPack,
			&entry.PackPayloadDigest,
			&entry.BusinessKey,
			&rawVersion,
			&entry.EnvelopeDigest,
			&definitionEnvelope,
		); err != nil {
			_ = rows.Close()
			return connectorbindingownerv1.DefinitionSet{}, fmt.Errorf(
				"scan pack %q connector binding definition: %w",
				coordinate.OwnerPack,
				err,
			)
		}
		// PackGenerationId retains the parent schema collation for FK
		// compatibility. The query retrieves every collation-equivalent row;
		// this comparison rejects any byte-distinct hidden child.
		if packGenerationID != coordinate.StorageID {
			_ = rows.Close()
			return connectorbindingownerv1.DefinitionSet{}, fmt.Errorf(
				"%w: pack %q connector binding definition names a different generation",
				ErrConflict,
				coordinate.OwnerPack,
			)
		}
		entry.Kind = consumerv1.IntegrationBindingKind(rawKind)
		entry.Ordinal, err = parseCanonicalUint64(rawOrdinal, true)
		if err != nil {
			_ = rows.Close()
			return connectorbindingownerv1.DefinitionSet{}, fmt.Errorf(
				"%w: pack %q connector binding definition ordinal is malformed: %v",
				ErrConflict,
				coordinate.OwnerPack,
				err,
			)
		}
		entry.DefinitionVersion, err = parseCanonicalUint64(rawVersion, false)
		if err != nil {
			_ = rows.Close()
			return connectorbindingownerv1.DefinitionSet{}, fmt.Errorf(
				"%w: pack %q connector binding definition version is malformed: %v",
				ErrConflict,
				coordinate.OwnerPack,
				err,
			)
		}
		nextDefinitionSetBytes, withinBudget := AddEnvelopeBytes(
			definitionSetBytes,
			len(definitionEnvelope),
		)
		if !withinBudget {
			_ = rows.Close()
			return connectorbindingownerv1.DefinitionSet{}, fmt.Errorf(
				"%w: pack %q connector binding definition bytes exceed %d",
				ErrConflict,
				coordinate.OwnerPack,
				connectorbindingownerv1.MaxDefinitionSetBytes,
			)
		}
		definitionSetBytes = nextDefinitionSetBytes
		entry.EnvelopeJSON = append([]byte(nil), definitionEnvelope...)
		if id != DefinitionStorageID(
			coordinate.StorageID,
			entry.Kind,
			entry.BusinessKey,
		) {
			_ = rows.Close()
			return connectorbindingownerv1.DefinitionSet{}, fmt.Errorf(
				"%w: pack %q connector binding definition storage identity is malformed",
				ErrConflict,
				coordinate.OwnerPack,
			)
		}
		entries = append(entries, entry)
	}
	if err := rows.Err(); err != nil {
		_ = rows.Close()
		return connectorbindingownerv1.DefinitionSet{}, fmt.Errorf(
			"iterate pack %q connector binding definitions: %w",
			coordinate.OwnerPack,
			err,
		)
	}
	if err := rows.Close(); err != nil {
		return connectorbindingownerv1.DefinitionSet{}, fmt.Errorf(
			"close pack %q connector binding definitions: %w",
			coordinate.OwnerPack,
			err,
		)
	}

	set := connectorbindingownerv1.DefinitionSet{
		Header: connectorbindingownerv1.DefinitionSetHeader{
			Protocol:                protocol,
			DefinitionSetCoordinate: portableCoordinate,
			DefinitionCount:         definitionCount,
			OrderedSetDigest:        setDigest,
		},
		Entries: entries,
	}
	if _, err := connectorbindingownerv1.ValidateDefinitionSet(set); err != nil {
		return connectorbindingownerv1.DefinitionSet{}, fmt.Errorf(
			"%w: pack %q connector binding definition set is malformed: %v",
			ErrConflict,
			coordinate.OwnerPack,
			err,
		)
	}
	return set, nil
}

// verifyPackGenerationParent locks and proves the immutable parent before any
// derived definition-set row is trusted. Keeping the parent, header, and child
// reads on the same transaction-bound Reader prevents a concurrent or corrupt
// PackGeneration tuple from authenticating a stale, internally consistent set.
func verifyPackGenerationParent(
	ctx context.Context,
	reader Reader,
	coordinate GenerationCoordinate,
) error {
	var (
		stored        GenerationCoordinate
		appliedDigest sql.NullString
	)
	// Keep the ordinary equality as the clustered-index access predicate while
	// the binary equality preserves the byte-exact authority boundary.
	err := reader.QueryRowContext(ctx, `SELECT
Id, PackName, PackInstallId, AppliedPayloadHash
FROM _System_PackGeneration
WHERE Id = ?
  AND BINARY Id = BINARY ?
LIMIT 1
FOR UPDATE`, coordinate.StorageID, coordinate.StorageID).Scan(
		&stored.StorageID,
		&stored.OwnerPack,
		&stored.PackInstallationID,
		&appliedDigest,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return fmt.Errorf(
			"%w: pack %q generation %q is missing its immutable parent",
			ErrConflict,
			coordinate.OwnerPack,
			coordinate.PackInstallationID,
		)
	}
	if err != nil {
		return fmt.Errorf(
			"lock pack %q immutable generation parent: %w",
			coordinate.OwnerPack,
			err,
		)
	}
	stored.PackPayloadDigest = appliedDigest.String
	if !appliedDigest.Valid || stored != coordinate {
		return fmt.Errorf(
			"%w: pack %q immutable generation parent does not match its exact storage coordinate",
			ErrConflict,
			coordinate.OwnerPack,
		)
	}
	return nil
}

// ReadVerifiedBinding verifies the complete stored set before selecting one
// exact portable binding identity from it.
func ReadVerifiedBinding(
	ctx context.Context,
	reader Reader,
	coordinate GenerationCoordinate,
	binding consumerv1.IntegrationBinding,
) error {
	_, err := ReadVerifiedBindingDefinition(ctx, reader, coordinate, binding)
	return err
}

// ReadVerifiedBindingDefinition verifies the complete stored set before
// selecting and returning one exact portable binding definition from it. The
// returned envelope is detached from the stored set and remains authenticated
// by the owner's complete generation commitment.
func ReadVerifiedBindingDefinition(
	ctx context.Context,
	reader Reader,
	coordinate GenerationCoordinate,
	binding consumerv1.IntegrationBinding,
) (connectorbindingownerv1.DefinitionEnvelope, error) {
	identity := connectorbindingownerv1.IdentityFromBinding(binding)
	if err := connectorbindingownerv1.ValidateIdentity(identity); err != nil {
		return connectorbindingownerv1.DefinitionEnvelope{}, fmt.Errorf(
			"%w: connector binding identity is invalid: %v",
			ErrPrecondition,
			err,
		)
	}
	if identity.OwnerPack != coordinate.OwnerPack ||
		identity.PackInstallationID != coordinate.PackInstallationID ||
		identity.PackPayloadDigest != coordinate.PackPayloadDigest {
		return connectorbindingownerv1.DefinitionEnvelope{}, fmt.Errorf(
			"%w: connector binding %s/%s names a different generation",
			ErrConflict,
			binding.OwnerPack,
			binding.BusinessKey,
		)
	}
	set, err := ReadVerifiedDefinitionSet(ctx, reader, coordinate)
	if err != nil {
		return connectorbindingownerv1.DefinitionEnvelope{}, err
	}
	for _, entry := range set.Entries {
		if entry.Kind != binding.Kind || entry.BusinessKey != binding.BusinessKey {
			continue
		}
		if entry.OwnerPack == binding.OwnerPack &&
			entry.PackInstallationID == binding.PackInstallationID &&
			entry.PackPayloadDigest == binding.PackPayloadDigest &&
			entry.DefinitionVersion == binding.DefinitionVersion {
			definition, err := connectorbindingownerv1.ParseDefinitionEnvelopeJSON(
				entry.EnvelopeJSON,
			)
			if err != nil {
				return connectorbindingownerv1.DefinitionEnvelope{}, fmt.Errorf(
					"%w: connector binding %s/%s envelope is invalid after set verification: %v",
					ErrConflict,
					binding.OwnerPack,
					binding.BusinessKey,
					err,
				)
			}
			return connectorbindingownerv1.CloneDefinitionEnvelope(definition), nil
		}
		break
	}
	return connectorbindingownerv1.DefinitionEnvelope{}, fmt.Errorf(
		"%w: connector binding %s/%s is absent from its complete definition set",
		ErrConflict,
		binding.OwnerPack,
		binding.BusinessKey,
	)
}

// AddEnvelopeBytes performs overflow-safe aggregate byte accounting.
func AddEnvelopeBytes(total uint64, envelopeSize int) (uint64, bool) {
	if total > connectorbindingownerv1.MaxDefinitionSetBytes || envelopeSize < 0 {
		return 0, false
	}
	next := uint64(envelopeSize)
	if next > connectorbindingownerv1.MaxDefinitionSetBytes-total {
		return 0, false
	}
	return total + next, true
}

func parseCanonicalUint64(raw []byte, allowZero bool) (uint64, error) {
	if len(raw) == 0 {
		return 0, fmt.Errorf("value is empty")
	}
	value, err := strconv.ParseUint(string(raw), 10, 64)
	if err != nil || strconv.FormatUint(value, 10) != string(raw) {
		return 0, fmt.Errorf("value is not a canonical unsigned decimal")
	}
	if !allowZero && value == 0 {
		return 0, fmt.Errorf("value must be positive")
	}
	return value, nil
}

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