package controlplane

import (
	"context"
	"database/sql"
	"fmt"
	"strings"
	"time"

	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
)

// canonicalizeApplyAllSourceRevisionAdvanceInstalledRow changes only the raw
// persisted source alias for an already-active target. It preserves the row,
// installation incarnation, lifecycle state, hashes, and runtime projections.
// The ApplyAll source-revision orchestrator invokes it for canonicalization-only entries.
func canonicalizeApplyAllSourceRevisionAdvanceInstalledRow(
	ctx context.Context,
	admission applyAllSourceRevisionAdvanceExecutionAdmission,
	pendingOrdinal int,
) error {
	db, admitted, err := applyAllSourceRevisionAdvanceAdmittedPendingEntry(admission, pendingOrdinal)
	if err != nil {
		return err
	}
	entry := admitted.current
	advance := entry.advance
	if entry.pending == nil || entry.pending.Kind != ApplyAllSourceRevisionAdvancePendingCanonicalize {
		return applyAllSourceRevisionAdvanceCanonicalizeError(advance, fmt.Errorf("admitted pending kind is not canonicalize"))
	}
	if strings.TrimSpace(entry.row.ID) == "" || strings.TrimSpace(entry.row.PackInstallID) == "" {
		return applyAllSourceRevisionAdvanceCanonicalizeError(advance, fmt.Errorf("installed row and pack-install incarnation identity are required"))
	}
	if entry.row.Name != advance.Name || entry.row.IsDeleted != 0 {
		return applyAllSourceRevisionAdvanceCanonicalizeError(advance, fmt.Errorf("installed row is not the active planned pack"))
	}
	pending, err := classifyApplyAllSourceRevisionAdvance(entry.row, advance.SourceRevisionAdvanceAuthorization)
	if err != nil || !equalApplyAllSourceRevisionAdvancePending(pending, entry.pending) {
		if err == nil {
			err = fmt.Errorf("pending classification does not match the frozen row")
		}
		return applyAllSourceRevisionAdvanceCanonicalizeError(advance, err)
	}
	if err := validateApplyAllSourceRevisionAdvanceFrozenCurrentTargetEntry(entry, admitted.currentTarget); err != nil {
		return applyAllSourceRevisionAdvanceCanonicalizeError(advance, fmt.Errorf("validate admitted current target: %w", err))
	}
	if err := validateApplyAllSourceRevisionAdvanceTarget(advance, admitted.installedCompatibility); err != nil {
		return applyAllSourceRevisionAdvanceCanonicalizeError(advance, fmt.Errorf("validate admitted installed target: %w", err))
	}
	if err := packs.ValidateSourceBackedExecutionAtomTarget(admitted.atom, admitted.installedCompatibility); err != nil {
		return applyAllSourceRevisionAdvanceCanonicalizeError(advance, fmt.Errorf("validate admitted execution atom: %w", err))
	}

	includeHostDigest, err := installedPackHasHostRequirementsDigestColumn(ctx, db)
	if err != nil {
		return applyAllSourceRevisionAdvanceCanonicalizeError(advance, err)
	}
	predicate, predicateArgs := applyAllSourceRevisionAdvanceExactInstalledRowPredicate(entry.row, includeHostDigest)
	args := []any{advance.Kind, advance.URI, time.Now().UTC()}
	args = append(args, predicateArgs...)
	result, err := db.ExecContext(ctx, `UPDATE _System_InstalledPack p
   SET SourceKind = ?,
       SourceURI = ?,
       LastModifiedDate = ?
 WHERE `+predicate, args...)
	if err != nil {
		return applyAllSourceRevisionAdvanceCanonicalizeError(advance, fmt.Errorf("write canonical source identity: %w", err))
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return applyAllSourceRevisionAdvanceCanonicalizeError(advance, fmt.Errorf("canonical source identity rows affected: %w", err))
	}
	if affected != 1 {
		return applyAllSourceRevisionAdvanceCanonicalizeError(
			advance, fmt.Errorf("canonical source identity affected %d rows; frozen installed row changed", affected),
		)
	}
	return nil
}

func applyAllSourceRevisionAdvanceAdmittedPendingEntry(
	admission applyAllSourceRevisionAdvanceExecutionAdmission,
	pendingOrdinal int,
) (*sql.DB, applyAllSourceRevisionAdvanceExecutionProofEntry, error) {
	identity := admission.identity
	if err := validateApplyAllSourceRevisionAdvanceExecutionAdmission(admission); err != nil {
		return nil, applyAllSourceRevisionAdvanceExecutionProofEntry{}, err
	}
	proof := identity.proof
	pendingIndexes := make([]int, 0, len(proof.entries))
	for index, entry := range proof.entries {
		if entry.current.pending != nil {
			pendingIndexes = append(pendingIndexes, index)
		}
	}
	if pendingOrdinal < 0 || pendingOrdinal >= len(pendingIndexes) {
		return nil, applyAllSourceRevisionAdvanceExecutionProofEntry{}, applyAllSourceRevisionAdvancePlanError(
			ApplyAllSourceRevisionAdvancePlanState, -1, "", fmt.Errorf("pending admission ordinal %d is out of range", pendingOrdinal),
		)
	}
	entry := cloneApplyAllSourceRevisionAdvanceExecutionProofEntry(
		proof.entries[pendingIndexes[pendingOrdinal]],
	)
	return identity.provenance.activeRuntimeDatabase, entry, nil
}

func applyAllSourceRevisionAdvanceExactInstalledRowPredicate(
	row installedPackRow,
	includeHostDigest bool,
) (string, []any) {
	columns := []string{
		"p.Id", "p.Name", "p.PackInstallId", "p.Category", "p.Status",
	}
	args := []any{row.ID, row.Name, row.PackInstallID, row.Category, row.Status}
	predicates := make([]string, 0, 16)
	for _, column := range columns {
		predicates = append(predicates, applyAllSourceRevisionAdvanceExactStringPredicate(column))
	}
	predicates = append(predicates, "COALESCE(p.IsDeleted, 0) = ?")
	args = append(args, row.IsDeleted)
	for _, field := range []struct {
		column string
		value  sql.NullString
	}{
		{column: "p.PackVersion", value: row.PackVersion},
		{column: "p.ManifestFormatVersion", value: row.ManifestFormatVersion},
		{column: "p.PlatformCompatibility", value: row.PlatformCompatibility},
		{column: "p.DesiredPayloadHash", value: row.DesiredPayloadHash},
		{column: "p.AppliedPayloadHash", value: row.AppliedPayloadHash},
		{column: "p.PayloadHashAlgorithm", value: row.PayloadHashAlgorithm},
		{column: "p.SourceKind", value: row.SourceKind},
		{column: "p.SourceURI", value: row.SourceURI},
		{column: "p.SourceRevision", value: row.SourceRevision},
		{column: "p.HostRequirementsDigest", value: row.HostRequirementsDigest},
	} {
		if field.column == "p.HostRequirementsDigest" && !includeHostDigest {
			continue
		}
		predicates = append(predicates, applyAllSourceRevisionAdvanceExactStringPredicate(field.column))
		args = append(args, nullStringArg(field.value))
	}
	return strings.Join(predicates, "\n   AND "), args
}

func applyAllSourceRevisionAdvanceExactStringPredicate(column string) string {
	return "CAST(" + column + " AS BINARY) <=> CAST(? AS BINARY)"
}

func applyAllSourceRevisionAdvanceExactJSONPredicate(column string) string {
	return "CAST(" + column + " AS JSON) <=> CAST(? AS JSON)"
}

func applyAllSourceRevisionAdvanceCanonicalizeError(
	advance normalizedApplyAllSourceRevisionAdvance,
	err error,
) error {
	return applyAllSourceRevisionAdvancePlanError(
		ApplyAllSourceRevisionAdvancePlanState, advance.index, advance.Name,
		fmt.Errorf("canonicalize installed source identity: %w", err),
	)
}