package serving

import (
	"context"
	"database/sql"
	"errors"
	"fmt"

	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/credentialgeneration"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/serviceconfiguration"
	catalogv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	connectorbindingownerv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/connectorbindingownerv1"
)

var (
	errStoredLegacyServingEpoch           = errors.New("stored serving epoch is legacy-unprofiled")
	errStoredProfiledServingEpoch         = errors.New("stored serving epoch has profiled proof rows")
	errStoredDifferentProfileServingEpoch = errors.New("stored serving epoch uses a different digest profile")
	errStoredTerminalProfiledServingEpoch = errors.New("stored profiled serving epoch is terminal")
)

const profiledServingEpochHeaderSelect = `
SELECT ServingEpochId, ConnectorDigestProfileId,
       ServiceConfigurationGenerationProtocolVersion,
       ServiceConfigurationReceiptProtocolVersion
  FROM _System_ConnectorServingEpochReproducibility
 WHERE ServingEpochId = ? AND BINARY ServingEpochId = BINARY ?
 LIMIT 1`

const profiledServingEpochBindingOwnerSelect = `
SELECT owner.ServingEpochBindingId, owner.PackGenerationId,
       owner.DefinitionEnvelopeVersion, owner.DefinitionReceiptVersion,
       owner.CanonicalDefinitionDigest, owner.DefinitionOwnerReceipt,
       owner.DefinitionOwnerReceiptDigest
  FROM _System_ConnectorServingEpochBindingOwner AS owner
  JOIN _System_ConnectorServingEpochBinding AS binding
    ON binding.Id = owner.ServingEpochBindingId
   AND BINARY binding.Id = BINARY owner.ServingEpochBindingId
 WHERE binding.ServingEpochId = ? AND BINARY binding.ServingEpochId = BINARY ?
 ORDER BY binding.BindingOrdinal, owner.ServingEpochBindingId`

const profiledServingEpochCredentialSelect = `
SELECT ServingEpochId, CredentialOrdinal, CredentialGenerationId,
       ReceiptProtocolVersion, TransportConfigurationDigest,
       OwnerReceipt, OwnerReceiptDigest
  FROM _System_ConnectorServingEpochCredentialGeneration
 WHERE ServingEpochId = ? AND BINARY ServingEpochId = BINARY ?
 ORDER BY CredentialOrdinal, CredentialGenerationId`

type storedProfiledHeader struct {
	servingEpochID            string
	profileID                 ConnectorDigestProfileID
	generationProtocolVersion serviceconfiguration.GenerationProtocolVersion
	receiptProtocolVersion    serviceconfiguration.ReceiptProtocolVersion
}

type storedProfiledProofGraph struct {
	headerFound bool
	header      storedProfiledHeader
	owners      []storedBindingOwnerProof
	credentials []storedCredentialGenerationProof
}

// profiledPreActivationEpoch keeps lifecycle state separate from the exported
// prepared-only candidate type. candidate is the exact immutable graph
// normalized to prepared solely for canonical validation and byte comparison;
// state is the byte-exact persisted lifecycle fact.
type profiledPreActivationEpoch struct {
	candidate ProfiledPreparedEpoch
	state     string
}

func (epoch profiledPreActivationEpoch) copy() profiledPreActivationEpoch {
	epoch.candidate = epoch.candidate.copy()
	return epoch
}

// LoadProfiledPrepared loads one complete immutable V2 candidate by its exact
// binary epoch ID. A legacy candidate, partial proof graph, or corrupt graph is
// rejected rather than inferred, repaired, or completed from current owners.
func (r *Repository) LoadProfiledPrepared(
	ctx context.Context,
	id string,
) (ProfiledPreparedEpoch, bool, error) {
	if r == nil || r.db == nil {
		return ProfiledPreparedEpoch{}, false, fmt.Errorf("%w: database is nil", ErrDependencyMissing)
	}
	if ctx == nil {
		return ProfiledPreparedEpoch{}, false, fmt.Errorf("%w: context is nil", ErrDependencyMissing)
	}
	if !validOpaqueIdentifierWithin(id, connectorDigestProfileV2MaxIdentifierBytes) {
		return ProfiledPreparedEpoch{}, false, invalidCandidate("epoch ID is not canonical")
	}
	if err := r.VerifySchema(ctx); err != nil {
		return ProfiledPreparedEpoch{}, false, err
	}
	return loadProfiledPrepared(ctx, r.db, id, false)
}

func loadProfiledPrepared(
	ctx context.Context,
	queryer preparedEpochQueryer,
	id string,
	forUpdate bool,
) (ProfiledPreparedEpoch, bool, error) {
	epoch, found, err := loadProfiledPreActivationEpoch(ctx, queryer, id, forUpdate)
	if err != nil || !found {
		return ProfiledPreparedEpoch{}, found, err
	}
	if epoch.state != preparedState {
		return ProfiledPreparedEpoch{}, false, errors.Join(
			fmt.Errorf("%w: profiled candidate is already %q", ErrProfiledEpochStateConflict, epoch.state),
			errStoredTerminalProfiledServingEpoch,
		)
	}
	return epoch.candidate.copy(), true, nil
}

func loadProfiledPreActivationEpoch(
	ctx context.Context,
	queryer preparedEpochQueryer,
	id string,
	forUpdate bool,
) (profiledPreActivationEpoch, bool, error) {
	base, found, err := loadPreparedBase(ctx, queryer, id, forUpdate)
	if err != nil || !found {
		return profiledPreActivationEpoch{}, found, err
	}
	proofs, err := loadProfiledProofGraph(ctx, queryer, id, forUpdate)
	if err != nil {
		return profiledPreActivationEpoch{}, false, err
	}
	if !proofs.headerFound {
		if len(proofs.owners) != 0 || len(proofs.credentials) != 0 {
			return profiledPreActivationEpoch{}, false, storedCandidateInvalid(
				"profiled candidate header is missing from a partial proof graph",
			)
		}
		return profiledPreActivationEpoch{}, false, errors.Join(
			storedCandidateInvalid("epoch is a legacy candidate without a reproducibility profile"),
			errStoredLegacyServingEpoch,
		)
	}
	if proofs.header.profileID != ConnectorDigestProfileV2ID {
		return profiledPreActivationEpoch{}, false, errors.Join(
			storedCandidateInvalid(
				"stored digest profile %q is not Profile V2",
				proofs.header.profileID,
			),
			errStoredDifferentProfileServingEpoch,
		)
	}
	epoch, err := profiledPreActivationFromStored(base, proofs, id)
	if err != nil {
		return profiledPreActivationEpoch{}, false, err
	}
	return epoch.copy(), true, nil
}

func profiledPreActivationFromStored(
	base PreparedEpoch,
	proofs storedProfiledProofGraph,
	id string,
) (profiledPreActivationEpoch, error) {
	state := base.state
	if !isProfiledPreActivationState(state) || base.activeScopeDigest != nil {
		return profiledPreActivationEpoch{}, storedCandidateInvalid(
			"profiled pre-activation lifecycle state is invalid",
		)
	}
	base.state = preparedState
	candidate, err := profiledCandidateFromStored(base, proofs, id)
	if err != nil {
		return profiledPreActivationEpoch{}, err
	}
	return profiledPreActivationEpoch{candidate: candidate.copy(), state: state}, nil
}

func isProfiledPreActivationState(state string) bool {
	return state == preparedState || state == verifiedState || state == rejectedState
}

func profiledCandidateFromStored(
	base PreparedEpoch,
	proofs storedProfiledProofGraph,
	id string,
) (ProfiledPreparedEpoch, error) {
	if !proofs.headerFound || proofs.header.servingEpochID != id || base.id != id {
		return ProfiledPreparedEpoch{}, storedCandidateInvalid("profiled header epoch ID is not byte-exact")
	}
	if len(proofs.owners) != len(base.bindings) {
		return ProfiledPreparedEpoch{}, storedCandidateInvalid(
			"profiled binding-owner proof count = %d, want %d", len(proofs.owners), len(base.bindings),
		)
	}

	bindings := make([]profiledStoredBinding, len(base.bindings))
	for index := range base.bindings {
		bindings[index] = profiledStoredBinding{
			binding: base.bindings[index].copy(),
			owner:   proofs.owners[index].copy(),
		}
	}
	credentials := make([]storedCredentialGenerationProof, len(proofs.credentials))
	for index := range proofs.credentials {
		credentials[index] = proofs.credentials[index].copy()
	}
	epoch := ProfiledPreparedEpoch{
		id:                 base.id,
		epochDigest:        base.epochDigest,
		servingScopeDigest: base.servingScopeDigest,
		state:              base.state,
		activeScopeDigest:  copyOptionalString(base.activeScopeDigest),
		profileID:          proofs.header.profileID,
		catalog:            base.catalog,
		serviceConfiguration: storedServiceConfigurationProof{
			generationProtocolVersion: proofs.header.generationProtocolVersion,
			receiptProtocolVersion:    proofs.header.receiptProtocolVersion,
			generationID:              base.l2.configurationGenerationRef,
			generationDigest:          base.l2.configurationDigest,
			ownerReceipt:              append([]byte(nil), base.l2.verificationReceipt...),
			ownerReceiptDigest:        base.l2.verificationReceiptDigest,
		},
		bindings:    bindings,
		credentials: credentials,
	}
	if err := epoch.validateStored(); err != nil {
		return ProfiledPreparedEpoch{}, err
	}
	return epoch, nil
}

func loadProfiledProofGraph(
	ctx context.Context,
	queryer preparedEpochQueryer,
	id string,
	forUpdate bool,
) (storedProfiledProofGraph, error) {
	headerQuery := profiledServingEpochHeaderSelect
	ownerQuery := profiledServingEpochBindingOwnerSelect
	credentialQuery := profiledServingEpochCredentialSelect
	if forUpdate {
		headerQuery += " FOR UPDATE"
		ownerQuery += " FOR UPDATE"
		credentialQuery += " FOR UPDATE"
	}

	var graph storedProfiledProofGraph
	headerRow := queryer.QueryRowContext(ctx, headerQuery, id, id)
	if err := headerRow.Err(); err != nil {
		return storedProfiledProofGraph{}, fmt.Errorf("load profiled serving epoch header: %w", err)
	}
	err := headerRow.Scan(
		&graph.header.servingEpochID,
		&graph.header.profileID,
		&graph.header.generationProtocolVersion,
		&graph.header.receiptProtocolVersion,
	)
	if err == nil {
		graph.headerFound = true
	} else if !errors.Is(err, sql.ErrNoRows) {
		return storedProfiledProofGraph{}, fmt.Errorf("%w: scan profiled header: %w", ErrStoredCandidateInvalid, err)
	}

	graph.owners, err = loadProfiledBindingOwners(ctx, queryer, ownerQuery, id)
	if err != nil {
		return storedProfiledProofGraph{}, err
	}
	graph.credentials, err = loadProfiledCredentials(ctx, queryer, credentialQuery, id)
	if err != nil {
		return storedProfiledProofGraph{}, err
	}
	return graph, nil
}

func loadProfiledBindingOwners(
	ctx context.Context,
	queryer preparedEpochQueryer,
	query string,
	id string,
) ([]storedBindingOwnerProof, error) {
	rows, err := queryer.QueryContext(ctx, query, id, id)
	if err != nil {
		return nil, fmt.Errorf("load profiled binding-owner proofs: %w", err)
	}
	defer rows.Close()
	owners := make([]storedBindingOwnerProof, 0)
	var receiptBytes uint64
	for rows.Next() {
		if len(owners) >= catalogv1.MaxBindingCount {
			return nil, storedCandidateInvalid("binding-owner proof count exceeds Catalog maximum")
		}
		var owner storedBindingOwnerProof
		if err := rows.Scan(
			&owner.servingEpochBindingID,
			&owner.packGenerationID,
			&owner.definitionEnvelopeVersion,
			&owner.definitionReceiptVersion,
			&owner.canonicalDefinitionDigest,
			&owner.definitionOwnerReceipt,
			&owner.definitionOwnerReceiptDigest,
		); err != nil {
			return nil, fmt.Errorf("%w: scan binding-owner proof: %w", ErrStoredCandidateInvalid, err)
		}
		if !validOpaqueIdentifierWithin(owner.packGenerationID, connectorDigestProfileV2MaxIdentifierBytes) {
			return nil, storedCandidateInvalid("binding-owner PackGeneration ID is not canonical")
		}
		var ok bool
		receiptBytes, ok = addWithinLimit(
			receiptBytes,
			uint64(len(owner.definitionOwnerReceipt)),
			connectorbindingownerv1.MaxDefinitionSetBytes,
		)
		if !ok {
			return nil, storedCandidateInvalid("binding-owner receipts exceed their owner bound")
		}
		owner.definitionOwnerReceipt = append([]byte(nil), owner.definitionOwnerReceipt...)
		owners = append(owners, owner)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("load profiled binding-owner proofs: %w", err)
	}
	if err := rows.Close(); err != nil {
		return nil, fmt.Errorf("close profiled binding-owner proof rows: %w", err)
	}
	return owners, nil
}

func loadProfiledCredentials(
	ctx context.Context,
	queryer preparedEpochQueryer,
	query string,
	id string,
) ([]storedCredentialGenerationProof, error) {
	rows, err := queryer.QueryContext(ctx, query, id, id)
	if err != nil {
		return nil, fmt.Errorf("load profiled credential-generation proofs: %w", err)
	}
	defer rows.Close()
	proofs := make([]storedCredentialGenerationProof, 0)
	var receiptBytes uint64
	for rows.Next() {
		if len(proofs) >= credentialgeneration.MaxExactIdentitySetEntries {
			return nil, storedCandidateInvalid("credential-generation proof count exceeds owner maximum")
		}
		var (
			proof   storedCredentialGenerationProof
			epochID string
		)
		if err := rows.Scan(
			&epochID,
			&proof.ordinal,
			&proof.generationID,
			&proof.receiptProtocolVersion,
			&proof.transportConfigurationDigest,
			&proof.ownerReceipt,
			&proof.ownerReceiptDigest,
		); err != nil {
			return nil, fmt.Errorf("%w: scan credential-generation proof: %w", ErrStoredCandidateInvalid, err)
		}
		if epochID != id {
			return nil, storedCandidateInvalid("credential-generation proof epoch ID is not byte-exact")
		}
		var ok bool
		receiptBytes, ok = addWithinLimit(
			receiptBytes,
			uint64(len(proof.ownerReceipt)),
			credentialgeneration.MaxExactIdentitySetPayloadBytes,
		)
		if !ok {
			return nil, storedCandidateInvalid("credential-generation receipts exceed their owner bound")
		}
		proof.ownerReceipt = append([]byte(nil), proof.ownerReceipt...)
		proofs = append(proofs, proof)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("load profiled credential-generation proofs: %w", err)
	}
	if err := rows.Close(); err != nil {
		return nil, fmt.Errorf("close profiled credential-generation proof rows: %w", err)
	}
	return proofs, nil
}