package serving

import (
	"bytes"
	"errors"

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

var (
	// ErrDependencyMissing reports that the serving repository cannot be
	// constructed because a required dependency is absent.
	ErrDependencyMissing = errors.New("connector serving: dependency missing")
	// ErrInvalidCandidate reports that caller-supplied candidate material is not
	// a closed, canonical ServingEpoch identity.
	ErrInvalidCandidate = errors.New("connector serving: invalid candidate")
	// ErrIdentityConflict reports that an existing epoch ID names materially
	// different immutable candidate bytes.
	ErrIdentityConflict = errors.New("connector serving: identity conflict")
	// ErrStoredCandidateInvalid reports that persisted candidate rows do not
	// reconstruct one complete, canonical prepared epoch.
	ErrStoredCandidateInvalid = errors.New("connector serving: stored candidate invalid")
	// ErrSchemaUnavailable reports that the catalog-owned serving schema is not
	// available to the Connector repository.
	ErrSchemaUnavailable = errors.New("connector serving: schema unavailable")
	// ErrProfiledEpochStateConflict reports that an exact immutable profiled
	// epoch exists in a valid lifecycle state that cannot perform the requested
	// prepared-only operation.
	ErrProfiledEpochStateConflict = errors.New("connector serving: profiled epoch state conflict")
)

const (
	preparedState = "prepared"
	verifiedState = "verified"
	rejectedState = "rejected"
)

// L2Snapshot is the immutable, versioned L2 configuration generation admitted
// into a ServingEpoch candidate. Its receipt is opaque to L0 and is defensively
// copied at every public boundary.
type L2Snapshot struct {
	configurationGenerationRef string
	configurationDigest        string
	verificationReceipt        []byte
	verificationReceiptDigest  string
}

// ConfigurationGenerationRef returns the exact versioned L2 generation.
func (snapshot L2Snapshot) ConfigurationGenerationRef() string {
	return snapshot.configurationGenerationRef
}

// ConfigurationDigest returns the L2-owner-authored configuration digest.
func (snapshot L2Snapshot) ConfigurationDigest() string {
	return snapshot.configurationDigest
}

// VerificationReceipt returns a copy of the opaque L2 verification receipt.
func (snapshot L2Snapshot) VerificationReceipt() []byte {
	return append([]byte(nil), snapshot.verificationReceipt...)
}

// VerificationReceiptDigest returns the Connector-owned canonical digest of
// the exact receipt bytes.
func (snapshot L2Snapshot) VerificationReceiptDigest() string {
	return snapshot.verificationReceiptDigest
}

func (snapshot L2Snapshot) copy() L2Snapshot {
	snapshot.verificationReceipt = append([]byte(nil), snapshot.verificationReceipt...)
	return snapshot
}

func (snapshot L2Snapshot) equal(other L2Snapshot) bool {
	return snapshot.configurationGenerationRef == other.configurationGenerationRef &&
		snapshot.configurationDigest == other.configurationDigest &&
		snapshot.verificationReceiptDigest == other.verificationReceiptDigest &&
		bytes.Equal(snapshot.verificationReceipt, other.verificationReceipt)
}

// catalogReference is the exact immutable Catalog identity admitted into a
// candidate. It deliberately retains no IntegrationBindingView and exposes no
// Catalog-shaped read API: Catalog remains the sole definition authority.
type catalogReference struct {
	orgID                string
	appInstallationID    string
	authorizedWorkloadID string
	lineage              catalogv1.ViewLineage
	receipts             catalogv1.Receipts
}

// PreparedEpoch is a legacy-unprofiled immutable candidate. It deliberately
// exposes no state-transition method: this slice can only prepare and load it.
// All fields remain private so callers cannot manufacture derived digests or a
// partially populated persisted representation.
type PreparedEpoch struct {
	id                 string
	epochDigest        string
	servingScopeDigest string
	state              string
	activeScopeDigest  *string
	catalog            catalogReference
	l2                 L2Snapshot
	bindings           []storedBinding
}

// ID returns the stable Connector-owned epoch identity.
func (epoch PreparedEpoch) ID() string { return epoch.id }

// EpochDigest returns the digest of the complete immutable candidate.
func (epoch PreparedEpoch) EpochDigest() string { return epoch.epochDigest }

// ServingScopeDigest returns the raw hexadecimal digest of the authorized
// Catalog consumer scope.
func (epoch PreparedEpoch) ServingScopeDigest() string { return epoch.servingScopeDigest }

// State returns the persisted candidate state. This slice only produces
// prepared epochs.
func (epoch PreparedEpoch) State() string { return epoch.state }

// ActiveScopeDigest reports the active-scope key, if one exists. Prepared
// candidates always return ok=false; the pair makes accidental future exposure
// of a mutable pointer impossible.
func (epoch PreparedEpoch) ActiveScopeDigest() (value string, ok bool) {
	if epoch.activeScopeDigest == nil {
		return "", false
	}
	return *epoch.activeScopeDigest, true
}

// L2Snapshot returns a deep copy of the admitted L2 generation receipt.
func (epoch PreparedEpoch) L2Snapshot() L2Snapshot { return epoch.l2.copy() }

// BindingCount returns the number of ordered exact-generation references.
func (epoch PreparedEpoch) BindingCount() int { return len(epoch.bindings) }

func (epoch PreparedEpoch) copy() PreparedEpoch {
	epoch.l2 = epoch.l2.copy()
	if epoch.activeScopeDigest != nil {
		value := *epoch.activeScopeDigest
		epoch.activeScopeDigest = &value
	}
	epoch.bindings = copyStoredBindings(epoch.bindings)
	return epoch
}

// equal performs the byte-exact duplicate comparison. It compares canonical
// persisted binding bytes rather than Go slice nilness, so two equivalent
// Catalog zero-length lists converge to one identity.
func (epoch PreparedEpoch) equal(other PreparedEpoch) bool {
	if epoch.id != other.id || epoch.epochDigest != other.epochDigest ||
		epoch.servingScopeDigest != other.servingScopeDigest ||
		epoch.state != other.state ||
		!equalOptionalString(epoch.activeScopeDigest, other.activeScopeDigest) ||
		epoch.catalog != other.catalog ||
		!epoch.l2.equal(other.l2) ||
		len(epoch.bindings) != len(other.bindings) {
		return false
	}
	for i := range epoch.bindings {
		if !epoch.bindings[i].equal(other.bindings[i]) {
			return false
		}
	}
	return true
}

// storedBinding is the exact-generation reference projection that mirrors one
// immutable SQL row. It contains identity only, never editable definitions.
// Nullable Runtime fields remain independent pointers so a partial tuple
// survives scanning and is rejected instead of being silently normalized.
type storedBinding struct {
	id                          string
	servingEpochID              string
	ordinal                     uint64
	identityDigest              string
	kind                        catalogv1.IntegrationBindingKind
	ownerPack                   string
	packInstallationID          string
	packPayloadDigest           string
	businessKey                 string
	definitionVersion           uint64
	credentialRefs              []byte
	runtimeCommandKind          *string
	runtimePayloadFormatVersion *uint32
	runtimeDefinitionRef        *string
	runtimeDefinitionVersion    *string
}

func (binding storedBinding) copy() storedBinding {
	binding.credentialRefs = append([]byte(nil), binding.credentialRefs...)
	binding.runtimeCommandKind = copyOptionalString(binding.runtimeCommandKind)
	binding.runtimePayloadFormatVersion = copyOptionalUint32(binding.runtimePayloadFormatVersion)
	binding.runtimeDefinitionRef = copyOptionalString(binding.runtimeDefinitionRef)
	binding.runtimeDefinitionVersion = copyOptionalString(binding.runtimeDefinitionVersion)
	return binding
}

func (binding storedBinding) equal(other storedBinding) bool {
	return binding.id == other.id &&
		binding.servingEpochID == other.servingEpochID &&
		binding.ordinal == other.ordinal &&
		binding.identityDigest == other.identityDigest &&
		binding.kind == other.kind &&
		binding.ownerPack == other.ownerPack &&
		binding.packInstallationID == other.packInstallationID &&
		binding.packPayloadDigest == other.packPayloadDigest &&
		binding.businessKey == other.businessKey &&
		binding.definitionVersion == other.definitionVersion &&
		bytes.Equal(binding.credentialRefs, other.credentialRefs) &&
		equalOptionalString(binding.runtimeCommandKind, other.runtimeCommandKind) &&
		equalOptionalUint32(binding.runtimePayloadFormatVersion, other.runtimePayloadFormatVersion) &&
		equalOptionalString(binding.runtimeDefinitionRef, other.runtimeDefinitionRef) &&
		equalOptionalString(binding.runtimeDefinitionVersion, other.runtimeDefinitionVersion)
}

func copyStoredBindings(bindings []storedBinding) []storedBinding {
	if bindings == nil {
		return nil
	}
	result := make([]storedBinding, len(bindings))
	for i := range bindings {
		result[i] = bindings[i].copy()
	}
	return result
}

func copyOptionalString(value *string) *string {
	if value == nil {
		return nil
	}
	result := *value
	return &result
}

func copyOptionalUint32(value *uint32) *uint32 {
	if value == nil {
		return nil
	}
	result := *value
	return &result
}

func equalOptionalString(first, second *string) bool {
	return (first == nil && second == nil) ||
		(first != nil && second != nil && *first == *second)
}

func equalOptionalUint32(first, second *uint32) bool {
	return (first == nil && second == nil) ||
		(first != nil && second != nil && *first == *second)
}