package controlplane

import (
	"context"
	"database/sql"
	"regexp"
	"strings"
	"testing"

	"github.com/DATA-DOG/go-sqlmock"
	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
)

func TestCanonicalizeApplyAllSourceRevisionAdvanceInstalledRowUsesExactCAS(t *testing.T) {
	for _, hostColumn := range []bool{true, false} {
		name := "legacy schema"
		if hostColumn {
			name = "current schema"
		}
		t.Run(name, func(t *testing.T) {
			db, mock, err := sqlmock.New()
			if err != nil {
				t.Fatalf("sqlmock: %v", err)
			}
			t.Cleanup(func() { _ = db.Close() })
			admission := testApplyAllSourceRevisionAdvanceCanonicalizeAdmission(t, db)
			expectApplyAllSourceRevisionAdvanceHostDigestColumn(mock, hostColumn)
			mock.ExpectExec(regexp.QuoteMeta(`UPDATE _System_InstalledPack p
   SET SourceKind = ?,
       SourceURI = ?,
       LastModifiedDate = ?
 WHERE `)).WillReturnResult(sqlmock.NewResult(0, 1))

			if err := canonicalizeApplyAllSourceRevisionAdvanceInstalledRow(
				context.Background(), admission, 0,
			); err != nil {
				t.Fatalf("canonicalize exact row: %v", err)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("SQL expectations: %v", err)
			}
		})
	}
}

func TestApplyAllSourceRevisionAdvanceExactInstalledRowPredicateFreezesProjection(t *testing.T) {
	row := testApplyAllSourceRevisionAdvanceCanonicalizeAdmission(t, &sql.DB{}).
		identity.proof.entries[0].current.row
	row.PlatformCompatibility = sql.NullString{}
	row.AppliedPayloadHash = sql.NullString{Valid: true, String: ""}
	predicate, args := applyAllSourceRevisionAdvanceExactInstalledRowPredicate(row, true)
	for _, column := range []string{
		"p.Id", "p.Name", "p.PackInstallId", "p.Category", "p.Status",
		"p.PackVersion", "p.ManifestFormatVersion", "p.PlatformCompatibility",
		"p.DesiredPayloadHash", "p.AppliedPayloadHash", "p.PayloadHashAlgorithm",
		"p.SourceKind", "p.SourceURI", "p.SourceRevision", "p.HostRequirementsDigest",
	} {
		want := applyAllSourceRevisionAdvanceExactStringPredicate(column)
		if !strings.Contains(predicate, want) {
			t.Fatalf("predicate missing byte-exact field %q:\n%s", column, predicate)
		}
	}
	if !strings.Contains(predicate, "COALESCE(p.IsDeleted, 0) = ?") || len(args) != 16 {
		t.Fatalf("predicate/args do not freeze the 16-field projection: %s / %#v", predicate, args)
	}
	if args[8] != nil || args[10] != "" {
		t.Fatalf("NULL-safe args collapsed NULL and empty string: %#v", args)
	}
	withoutHost, withoutHostArgs := applyAllSourceRevisionAdvanceExactInstalledRowPredicate(row, false)
	if strings.Contains(withoutHost, "HostRequirementsDigest") || len(withoutHostArgs) != 15 {
		t.Fatalf("legacy predicate retained unavailable host column: %s / %#v", withoutHost, withoutHostArgs)
	}
}

func TestCanonicalizeApplyAllSourceRevisionAdvanceInstalledRowRequiresOneRow(t *testing.T) {
	for _, affected := range []int64{0, 2} {
		t.Run(string(rune('0'+affected)), func(t *testing.T) {
			db, mock, err := sqlmock.New()
			if err != nil {
				t.Fatalf("sqlmock: %v", err)
			}
			t.Cleanup(func() { _ = db.Close() })
			admission := testApplyAllSourceRevisionAdvanceCanonicalizeAdmission(t, db)
			entry := admission.identity.proof.entries[0].current
			expectApplyAllSourceRevisionAdvanceHostDigestColumn(mock, true)
			mock.ExpectExec(regexp.QuoteMeta("UPDATE _System_InstalledPack p")).
				WillReturnResult(sqlmock.NewResult(0, affected))

			err = canonicalizeApplyAllSourceRevisionAdvanceInstalledRow(context.Background(), admission, 0)
			assertApplyAllSourceRevisionAdvancePreconditionError(
				t, err, ApplyAllSourceRevisionAdvancePlanState, entry.advance.index, entry.advance.Name,
			)
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("SQL expectations: %v", err)
			}
		})
	}
}

func TestCanonicalizeApplyAllSourceRevisionAdvanceInstalledRowRejectsInvalidCapabilityBeforeSQL(t *testing.T) {
	admission := testApplyAllSourceRevisionAdvanceCanonicalizeAdmission(t, &sql.DB{})
	admission.identity.proof.entries[0].current.pending.Kind = ApplyAllSourceRevisionAdvancePendingAdvance
	if err := canonicalizeApplyAllSourceRevisionAdvanceInstalledRow(
		context.Background(), admission, 0,
	); err == nil {
		t.Fatal("non-canonicalize capability was accepted")
	}
	admission = testApplyAllSourceRevisionAdvanceCanonicalizeAdmission(t, &sql.DB{})
	admission.identity.proof.entries[0].current.row.PackInstallID = ""
	if err := canonicalizeApplyAllSourceRevisionAdvanceInstalledRow(
		context.Background(), admission, 0,
	); err == nil {
		t.Fatal("blank installation incarnation was accepted")
	}
	admission = testApplyAllSourceRevisionAdvanceCanonicalizeAdmission(t, &sql.DB{})
	admission.identity.proof.provenance.entryCount = 0
	if err := canonicalizeApplyAllSourceRevisionAdvanceInstalledRow(
		context.Background(), admission, 0,
	); err == nil {
		t.Fatal("tampered proof cardinality was accepted")
	}
	if err := canonicalizeApplyAllSourceRevisionAdvanceInstalledRow(
		context.Background(), applyAllSourceRevisionAdvanceExecutionAdmission{}, 0,
	); err == nil {
		t.Fatal("unminted admission was accepted")
	}
}

func testApplyAllSourceRevisionAdvanceCanonicalizeAdmission(
	t *testing.T,
	db *sql.DB,
) applyAllSourceRevisionAdvanceExecutionAdmission {
	t.Helper()
	advance := testApplyAllSourceRevisionAdvance("alpha", 3)
	target := testApplyAllSourceRevisionTarget(
		t, advance.Name, "1.0.0", advance.Kind, advance.URI, advance.ToRevision,
	)
	current := testApplyAllSourceRevisionAdvanceFrozenCurrentTargetProof(
		t, advance, target, installedPackStatusActive, "in_repo",
		testCurrentTargetV1Hash, testCurrentTargetV1Hash,
	)
	atom, err := packs.FreezeSourceBackedExecutionAtom(target)
	if err != nil {
		t.Fatalf("freeze canonicalize fixture atom: %v", err)
	}
	entry := applyAllSourceRevisionAdvanceExecutionProofEntry{
		name: advance.Name, current: current.snapshot.entries[0],
		currentTarget: target, installedCompatibility: target, atom: atom,
	}
	return testApplyAllSourceRevisionAdvanceExecutionAdmissionFromEntries(
		t, db, []applyAllSourceRevisionAdvanceExecutionProofEntry{entry},
	)
}

func expectApplyAllSourceRevisionAdvanceHostDigestColumn(mock sqlmock.Sqlmock, present bool) {
	count := 0
	if present {
		count = 1
	}
	mock.ExpectQuery("SELECT COUNT\\(\\*\\)\\s+FROM INFORMATION_SCHEMA\\.COLUMNS").
		WithArgs("_System_InstalledPack", "HostRequirementsDigest").
		WillReturnRows(sqlmock.NewRows([]string{"COUNT(*)"}).AddRow(count))
}