package controlplane

import (
	"context"
	"database/sql"
	"database/sql/driver"
	"errors"
	"strings"
	"testing"
	"time"

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

func TestWithApplyAllSourceRevisionAdvanceExactIncarnationLocksAndDrainsBeforeMutation(t *testing.T) {
	for _, test := range []struct {
		category packs.Category
		column   string
	}{
		{category: packs.AppPack, column: "COALESCE(e.CompanionPackInstallId, '')"},
		{category: packs.ConnectorPack, column: "e.InstalledPackId"},
	} {
		t.Run(string(test.category), func(t *testing.T) {
			queries := []string{}
			matcher := sqlmock.QueryMatcherFunc(func(expectedSQL, actualSQL string) error {
				if err := sqlmock.QueryMatcherRegexp.Match(expectedSQL, actualSQL); err != nil {
					return err
				}
				queries = append(queries, actualSQL)
				return nil
			})
			db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(matcher))
			if err != nil {
				t.Fatalf("sqlmock: %v", err)
			}
			t.Cleanup(func() { _ = db.Close() })
			db.SetMaxOpenConns(1)
			row := testApplyAllSourceRevisionAdvancePreconditionRow(
				"alpha", installedPackStatusActive, string(packs.PackSourceKindOfficialRepo), "old", "applied",
			)
			row.Category = string(test.category)
			expectApplyAllSourceRevisionAdvanceExactIncarnationBegin(mock, row, test.category, false)
			mock.ExpectExec("UPDATE transition_probe").WillReturnResult(sqlmock.NewResult(0, 1))
			expectApplyAllSourceRevisionAdvanceExactIncarnationCommit(mock)

			ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
			defer cancel()
			var mutationGuard string
			err = withApplyAllSourceRevisionAdvanceExactIncarnation(
				ctx, db, row, test.category,
				func(exec applyAllSourceRevisionAdvanceExactIncarnationExec, queueGuard string) error {
					mutationGuard = queueGuard
					_, err := exec(ctx, "UPDATE transition_probe SET value = 1")
					return err
				},
			)
			if err != nil {
				t.Fatalf("exact incarnation transition: %v", err)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("SQL expectations: %v", err)
			}
			lock := testApplyAllSourceRevisionAdvanceQueryContaining(t, queries, "FROM _System_InstalledPack p")
			if !strings.Contains(lock, applyAllSourceRevisionAdvanceExactStringPredicate("p.PackInstallId")) ||
				!strings.Contains(lock, "FOR UPDATE") {
				t.Fatalf("exact row lock is incomplete:\n%s", lock)
			}
			queue := testApplyAllSourceRevisionAdvanceQueryContaining(t, queries, "FROM _System_ConnectorWorkItem w")
			exactMatch := connectorInboundPackInstallIDMatch(test.column, "p.PackInstallId", true)
			if !strings.Contains(queue, exactMatch) ||
				!strings.Contains(queue, "ProjectionKey") ||
				!strings.Contains(mutationGuard, exactMatch) ||
				!strings.Contains(mutationGuard, "FROM _System_ConnectorWorkItem w") {
				t.Fatalf("queue recheck is not exact for %s:\n%s", test.column, queue)
			}
		})
	}
}

func TestWithApplyAllSourceRevisionAdvanceExactIncarnationRefusesOldQueue(t *testing.T) {
	for _, category := range []packs.Category{packs.AppPack, packs.ConnectorPack} {
		t.Run(string(category), func(t *testing.T) {
			db, mock, err := sqlmock.New()
			if err != nil {
				t.Fatalf("sqlmock: %v", err)
			}
			t.Cleanup(func() { _ = db.Close() })
			db.SetMaxOpenConns(1)
			row := testApplyAllSourceRevisionAdvancePreconditionRow(
				"alpha", installedPackStatusActive, string(packs.PackSourceKindOfficialRepo), "old", "applied",
			)
			row.Category = string(category)
			expectApplyAllSourceRevisionAdvanceExactIncarnationBegin(mock, row, category, true)
			expectApplyAllSourceRevisionAdvanceExactIncarnationRollback(mock)
			called := false
			err = withApplyAllSourceRevisionAdvanceExactIncarnation(
				context.Background(), db, row, category,
				func(applyAllSourceRevisionAdvanceExactIncarnationExec, string) error { called = true; return nil },
			)
			if err == nil || !strings.Contains(err.Error(), "queue is not drained") || called {
				t.Fatalf("queued transition = %v, called=%t, want refusal before mutation", err, called)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("SQL expectations: %v", err)
			}
		})
	}
}

func TestWithApplyAllSourceRevisionAdvanceExactIncarnationRollsBackLostRowAndMutationFailure(t *testing.T) {
	mutationErr := errors.New("mutation failed")
	for _, test := range []struct {
		name       string
		lockExists bool
		mutate     func(applyAllSourceRevisionAdvanceExactIncarnationExec, string) error
		want       string
	}{
		{name: "lost row", mutate: func(applyAllSourceRevisionAdvanceExactIncarnationExec, string) error { return nil }, want: "changed before incarnation lock"},
		{name: "mutation failure", lockExists: true, mutate: func(applyAllSourceRevisionAdvanceExactIncarnationExec, string) error { return mutationErr }, want: mutationErr.Error()},
	} {
		t.Run(test.name, func(t *testing.T) {
			db, mock, err := sqlmock.New()
			if err != nil {
				t.Fatalf("sqlmock: %v", err)
			}
			t.Cleanup(func() { _ = db.Close() })
			row := testApplyAllSourceRevisionAdvancePreconditionRow(
				"alpha", installedPackStatusActive, string(packs.PackSourceKindOfficialRepo), "old", "applied",
			)
			expectConnectorWorkItemSchemaLookup(mock, true)
			expectApplyAllSourceRevisionAdvanceExactIncarnationTransaction(mock)
			_, lockArgs := applyAllSourceRevisionAdvanceExactInstalledRowPredicate(row, true)
			lock := mock.ExpectQuery(`(?s)SELECT 1.*FROM _System_InstalledPack p.*FOR UPDATE`).
				WithArgs(testApplyAllSourceRevisionAdvanceDriverValues(lockArgs)...)
			if test.lockExists {
				lock.WillReturnRows(sqlmock.NewRows([]string{"1"}).AddRow(1))
				_, queueArgs := applyAllSourceRevisionAdvanceExactInstalledRowPredicate(row, true)
				mock.ExpectQuery(`(?s)SELECT 1.*FROM _System_InstalledPack p.*FROM _System_ConnectorWorkItem w.*LIMIT 1`).
					WithArgs(testApplyAllSourceRevisionAdvanceDriverValues(queueArgs)...).
					WillReturnRows(sqlmock.NewRows([]string{"1"}))
			} else {
				lock.WillReturnError(sql.ErrNoRows)
			}
			expectApplyAllSourceRevisionAdvanceExactIncarnationRollback(mock)
			err = withApplyAllSourceRevisionAdvanceExactIncarnation(
				context.Background(), db, row, packs.AppPack, test.mutate,
			)
			if err == nil || !strings.Contains(err.Error(), test.want) {
				t.Fatalf("transition error = %v, want %q", err, test.want)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("SQL expectations: %v", err)
			}
		})
	}
}

func TestWithApplyAllSourceRevisionAdvanceExactIncarnationPreservesAmbiguousCommit(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock: %v", err)
	}
	t.Cleanup(func() { _ = db.Close() })
	row := testApplyAllSourceRevisionAdvancePreconditionRow(
		"alpha", installedPackStatusActive, string(packs.PackSourceKindOfficialRepo), "old", "applied",
	)
	expectApplyAllSourceRevisionAdvanceExactIncarnationBegin(mock, row, packs.AppPack, false)
	mock.ExpectExec("UPDATE transition_probe").WillReturnResult(sqlmock.NewResult(0, 1))
	commitErr := errors.New("commit acknowledgement lost")
	mock.ExpectCommit().WillReturnError(commitErr)

	err = withApplyAllSourceRevisionAdvanceExactIncarnation(
		context.Background(), db, row, packs.AppPack,
		func(exec applyAllSourceRevisionAdvanceExactIncarnationExec, _ string) error {
			_, err := exec(context.Background(), "UPDATE transition_probe SET value = 1")
			return err
		},
	)
	var ambiguous *applyAllSourceRevisionAdvanceExactIncarnationCommitError
	if !errors.As(err, &ambiguous) || !errors.Is(err, commitErr) {
		t.Fatalf("commit error = %v, want ambiguous outcome preserving cause", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("SQL expectations: %v", err)
	}
}

func TestWithApplyAllSourceRevisionAdvanceExactIncarnationRejectsInvalidIdentityBeforeSQL(t *testing.T) {
	row := testApplyAllSourceRevisionAdvancePreconditionRow(
		"alpha", installedPackStatusActive, string(packs.PackSourceKindOfficialRepo), "old", "applied",
	)
	for _, test := range []struct {
		name string
		edit func(*installedPackRow)
		want string
	}{
		{name: "missing row id", edit: func(row *installedPackRow) { row.ID = "" }, want: "identity are required"},
		{name: "missing incarnation", edit: func(row *installedPackRow) { row.PackInstallID = "" }, want: "identity are required"},
		{name: "category mismatch", edit: func(row *installedPackRow) { row.Category = string(packs.ConnectorPack) }, want: "does not match queue category"},
	} {
		t.Run(test.name, func(t *testing.T) {
			db, mock, err := sqlmock.New()
			if err != nil {
				t.Fatalf("sqlmock: %v", err)
			}
			t.Cleanup(func() { _ = db.Close() })
			got := row
			test.edit(&got)
			err = withApplyAllSourceRevisionAdvanceExactIncarnation(
				context.Background(), db, got, packs.AppPack,
				func(applyAllSourceRevisionAdvanceExactIncarnationExec, string) error { return nil },
			)
			if err == nil || !strings.Contains(err.Error(), test.want) {
				t.Fatalf("invalid exact incarnation identity error = %v, want %q", err, test.want)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("invalid identity reached SQL: %v", err)
			}
		})
	}
}

func TestApplyAllSourceRevisionAdvanceOldIncarnationQueueGuardUsesCanonicalExactAuthority(t *testing.T) {
	for _, test := range []struct {
		category packs.Category
		scope    connectorInboundReapplyScope
		column   string
	}{
		{category: packs.ConnectorPack, scope: connectorPackInboundReapplyScope, column: "e.InstalledPackId"},
		{category: packs.AppPack, scope: companionAppPackInboundReapplyScope, column: "COALESCE(e.CompanionPackInstallId, '')"},
	} {
		db, mock, err := sqlmock.New()
		if err != nil {
			t.Fatalf("sqlmock: %v", err)
		}
		expectConnectorWorkItemSchemaLookup(mock, true)
		blocker, err := resolveApplyAllSourceRevisionAdvanceOldIncarnationQueueBlocker(
			context.Background(), db, test.category,
		)
		if err != nil {
			_ = db.Close()
			t.Fatalf("resolve %s blocker: %v", test.category, err)
		}
		wantBlocker := connectorInboundExactIncarnationBlockerPredicate(test.scope, true)
		guard := applyAllSourceRevisionAdvanceOldIncarnationQueueGuard(blocker)
		if blocker != wantBlocker ||
			!strings.Contains(guard, connectorInboundPackInstallIDMatch(test.column, "p.PackInstallId", true)) ||
			!strings.Contains(guard, "FROM _System_ConnectorWorkItem w") ||
			!strings.Contains(guard, "ProjectionKey") {
			_ = db.Close()
			t.Fatalf("%s guard does not preserve canonical exact authority:\n%s", test.category, guard)
		}
		if err := mock.ExpectationsWereMet(); err != nil {
			_ = db.Close()
			t.Fatalf("%s SQL expectations: %v", test.category, err)
		}
		_ = db.Close()
	}
	if guard := applyAllSourceRevisionAdvanceOldIncarnationQueueGuard(""); guard != "" {
		t.Fatalf("empty blocker guard = %q, want empty", guard)
	}
}

func TestResolveApplyAllSourceRevisionAdvanceOldIncarnationQueueBlockerPreservesLegacyFallback(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock: %v", err)
	}
	t.Cleanup(func() { _ = db.Close() })
	expectConnectorWorkItemSchemaLookup(mock, false)
	blocker, err := resolveApplyAllSourceRevisionAdvanceOldIncarnationQueueBlocker(
		context.Background(), db, packs.ConnectorPack,
	)
	if err != nil {
		t.Fatalf("resolve legacy blocker: %v", err)
	}
	if blocker != connectorInboundExactIncarnationBlockerPredicate(connectorPackInboundReapplyScope, false) ||
		strings.Contains(blocker, "_System_ConnectorWorkItem") ||
		!strings.Contains(blocker, "e.Status IN ('pending', 'failed', 'processing')") {
		t.Fatalf("legacy blocker = %q", blocker)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("SQL expectations: %v", err)
	}
}

func expectApplyAllSourceRevisionAdvanceExactIncarnationBegin(
	mock sqlmock.Sqlmock,
	row installedPackRow,
	category packs.Category,
	queued bool,
) {
	expectConnectorWorkItemSchemaLookup(mock, true)
	expectApplyAllSourceRevisionAdvanceExactIncarnationTransaction(mock)
	_, lockArgs := applyAllSourceRevisionAdvanceExactInstalledRowPredicate(row, true)
	mock.ExpectQuery(`(?s)SELECT 1.*FROM _System_InstalledPack p.*FOR UPDATE`).
		WithArgs(testApplyAllSourceRevisionAdvanceDriverValues(lockArgs)...).
		WillReturnRows(sqlmock.NewRows([]string{"1"}).AddRow(1))
	_, queueArgs := applyAllSourceRevisionAdvanceExactInstalledRowPredicate(row, true)
	queueRows := sqlmock.NewRows([]string{"1"})
	if queued {
		queueRows.AddRow(1)
	}
	mock.ExpectQuery(`(?s)SELECT 1.*FROM _System_InstalledPack p.*FROM _System_ConnectorWorkItem w.*LIMIT 1`).
		WithArgs(testApplyAllSourceRevisionAdvanceDriverValues(queueArgs)...).
		WillReturnRows(queueRows)
}

func expectApplyAllSourceRevisionAdvanceExactIncarnationTransaction(mock sqlmock.Sqlmock) {
	mock.ExpectExec("SET SESSION tidb_txn_mode = 'pessimistic'").WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectExec("SET SESSION transaction_isolation = 'READ-COMMITTED'").WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectBegin()
}

func expectApplyAllSourceRevisionAdvanceExactIncarnationCommit(mock sqlmock.Sqlmock) {
	mock.ExpectCommit()
	expectApplyAllSourceRevisionAdvanceExactIncarnationSessionReset(mock)
}

func expectApplyAllSourceRevisionAdvanceExactIncarnationRollback(mock sqlmock.Sqlmock) {
	mock.ExpectRollback()
	expectApplyAllSourceRevisionAdvanceExactIncarnationSessionReset(mock)
}

func expectApplyAllSourceRevisionAdvanceExactIncarnationSessionReset(mock sqlmock.Sqlmock) {
	mock.ExpectExec("SET SESSION transaction_isolation = DEFAULT").WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectExec("SET SESSION tidb_txn_mode = DEFAULT").WillReturnResult(sqlmock.NewResult(0, 0))
}

func testApplyAllSourceRevisionAdvanceDriverValues(args []any) []driver.Value {
	values := make([]driver.Value, len(args))
	for index := range args {
		values[index] = args[index]
	}
	return values
}

func testApplyAllSourceRevisionAdvanceQueryContaining(t *testing.T, queries []string, fragment string) string {
	t.Helper()
	for _, query := range queries {
		if strings.Contains(query, fragment) {
			return query
		}
	}
	t.Fatalf("query containing %q was not executed: %#v", fragment, queries)
	return ""
}

func testApplyAllSourceRevisionAdvanceExactQueuePackMatch(category packs.Category) string {
	column := "e.InstalledPackId"
	if category == packs.AppPack {
		column = "COALESCE(e.CompanionPackInstallId, '')"
	}
	return connectorInboundPackInstallIDMatch(column, "p.PackInstallId", true)
}