package controlplane

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

	"github.com/DATA-DOG/go-sqlmock"
	"github.com/pingcap/tidb/pkg/parser"
	_ "github.com/pingcap/tidb/pkg/parser/test_driver"
)

func TestConnectorInboundReapplyBlockerPredicateUsesWorkItemAuthority(t *testing.T) {
	for _, tc := range []struct {
		name             string
		scope            connectorInboundReapplyScope
		workItemPackBind string
		eventPackBind    string
	}{
		{
			name:             "connector pack",
			scope:            connectorPackInboundReapplyScope,
			workItemPackBind: "w.InstalledPackId = p.PackInstallId",
			eventPackBind:    "e.InstalledPackId = p.PackInstallId",
		},
		{
			name:             "companion app pack",
			scope:            companionAppPackInboundReapplyScope,
			workItemPackBind: "COALESCE(queued_event.CompanionPackInstallId, '') = p.PackInstallId",
			eventPackBind:    "COALESCE(e.CompanionPackInstallId, '') = p.PackInstallId",
		},
	} {
		t.Run(tc.name, func(t *testing.T) {
			predicate := connectorInboundReapplyBlockerPredicate(tc.scope, true)
			for _, want := range []string{
				"FROM _System_ConnectorWorkItem w",
				"queued_event.InstalledPackId = w.InstalledPackId",
				tc.workItemPackBind,
				"(w.Attempt > 0 OR queued_event.ExecutionOwner = 'work_item')",
				"COALESCE(w.State, '') NOT IN ('processed', 'skipped', 'dead')",
				tc.eventPackBind,
				"(direct_work.Attempt > 0 OR e.ExecutionOwner = 'work_item')",
				"direct_work.WorkId IS NULL",
				"(projection.Attempt > 0 OR selected.ExecutionOwner = 'work_item')",
				"NULLIF(COALESCE(projection.ProjectionKey, ''), '') IS NOT NULL",
				"projection.WorkId LIKE 'csp:%'",
				"COALESCE(projection.SourceLaneKey, '') = COALESCE(e.SourceLaneKey, '')",
				"selected.EnqueueOrdinal",
				"selected.DeliverySequence",
			} {
				if !strings.Contains(predicate, want) {
					t.Fatalf("WorkItem authority predicate missing %q:\n%s", want, predicate)
				}
			}
		})
	}
}

func TestConnectorInboundReapplyBlockerPredicateCoversInitialWorkItemOwnership(t *testing.T) {
	predicate := connectorInboundReapplyBlockerPredicate(connectorPackInboundReapplyScope, true)

	// A newly owned pending WorkItem can still have Attempt=0 before its first
	// claim. The Event CAS owner must therefore be an independent authority arm.
	if !strings.Contains(predicate, "(w.Attempt > 0 OR queued_event.ExecutionOwner = 'work_item')") {
		t.Fatalf("initial WorkItem ownership can escape the blocker:\n%s", predicate)
	}
	// Conversely, an Attempt=0 compatibility mirror whose Event remains owned
	// by the legacy executor must not suppress the Event fallback.
	if !strings.Contains(predicate, "(direct_work.Attempt > 0 OR e.ExecutionOwner = 'work_item')") {
		t.Fatalf("direct mirror authority is not fenced by Event owner:\n%s", predicate)
	}
	if strings.Contains(predicate, "AND e.ExecutionOwner = 'event'\n          AND e.Status") {
		t.Fatalf("Event fallback excludes owner-mismatch fail-closed rows:\n%s", predicate)
	}
}

func TestConnectorInboundWorkItemSchemaReadyRequiresEveryReferencedColumn(t *testing.T) {
	for _, tc := range []struct {
		name  string
		count int
		want  bool
	}{
		{name: "complete schema", count: len(connectorInboundWorkItemRequiredColumns), want: true},
		{name: "incomplete additive schema", count: len(connectorInboundWorkItemRequiredColumns) - 1, want: false},
		{name: "table absent", count: 0, want: false},
	} {
		t.Run(tc.name, func(t *testing.T) {
			db, mock, err := sqlmock.New()
			if err != nil {
				t.Fatalf("sqlmock: %v", err)
			}
			defer db.Close()

			args := []driver.Value{"_System_ConnectorWorkItem"}
			for _, column := range connectorInboundWorkItemRequiredColumns {
				args = append(args, column)
			}
			mock.ExpectQuery(`SELECT COUNT\(DISTINCT COLUMN_NAME\)[\s\S]*FROM INFORMATION_SCHEMA\.COLUMNS`).
				WithArgs(args...).
				WillReturnRows(sqlmock.NewRows([]string{"COUNT(DISTINCT COLUMN_NAME)"}).AddRow(tc.count))

			got, err := connectorInboundWorkItemSchemaReady(context.Background(), db)
			if err != nil {
				t.Fatalf("connectorInboundWorkItemSchemaReady: %v", err)
			}
			if got != tc.want {
				t.Fatalf("connectorInboundWorkItemSchemaReady = %t, want %t", got, tc.want)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatalf("sql expectations: %v", err)
			}
		})
	}
}

func TestConnectorWorkItemSchemaReadinessSeparatesReapplyAuthorityFromCutoverMutation(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock: %v", err)
	}
	defer db.Close()

	authorityArgs := []driver.Value{"_System_ConnectorWorkItem"}
	for _, column := range connectorInboundWorkItemRequiredColumns {
		authorityArgs = append(authorityArgs, column)
	}
	mock.ExpectQuery(`SELECT COUNT\(DISTINCT COLUMN_NAME\)[\s\S]*FROM INFORMATION_SCHEMA\.COLUMNS`).
		WithArgs(authorityArgs...).
		WillReturnRows(sqlmock.NewRows([]string{"COUNT(DISTINCT COLUMN_NAME)"}).AddRow(len(connectorInboundWorkItemRequiredColumns)))

	cutoverArgs := []driver.Value{"_System_ConnectorWorkItem"}
	for _, column := range connectorWorkItemCutoverRequiredColumns {
		cutoverArgs = append(cutoverArgs, column)
	}
	mock.ExpectQuery(`SELECT COUNT\(DISTINCT COLUMN_NAME\)[\s\S]*FROM INFORMATION_SCHEMA\.COLUMNS`).
		WithArgs(cutoverArgs...).
		WillReturnRows(sqlmock.NewRows([]string{"COUNT(DISTINCT COLUMN_NAME)"}).AddRow(len(connectorInboundWorkItemRequiredColumns)))

	authorityReady, err := connectorInboundWorkItemSchemaReady(context.Background(), db)
	if err != nil {
		t.Fatalf("connectorInboundWorkItemSchemaReady: %v", err)
	}
	if !authorityReady {
		t.Fatal("reapply authority schema is not ready when every authority column exists")
	}

	cutoverReady, err := connectorWorkItemCutoverSchemaReady(context.Background(), db)
	if err != nil {
		t.Fatalf("connectorWorkItemCutoverSchemaReady: %v", err)
	}
	if cutoverReady {
		t.Fatal("cutover mutation schema is ready without every lease and terminal column")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestConnectorInboundReapplyBlockerPredicateFallsBackToLegacyEventStatus(t *testing.T) {
	for _, scope := range []connectorInboundReapplyScope{
		connectorPackInboundReapplyScope,
		companionAppPackInboundReapplyScope,
	} {
		predicate := connectorInboundReapplyBlockerPredicate(scope, false)
		if strings.Contains(predicate, "_System_ConnectorWorkItem") {
			t.Fatalf("legacy fallback unexpectedly requires WorkItem table:\n%s", predicate)
		}
		if !strings.Contains(predicate, "e.Status IN ('pending', 'failed', 'processing')") {
			t.Fatalf("legacy fallback missing Event status guard:\n%s", predicate)
		}
	}
}

func TestConnectorInboundReapplyBlockerPredicateParsesWithTiDBParser(t *testing.T) {
	p := parser.New()
	for _, scope := range []connectorInboundReapplyScope{
		connectorPackInboundReapplyScope,
		companionAppPackInboundReapplyScope,
	} {
		for _, workItemAuthority := range []bool{false, true} {
			for _, exactIdentity := range []bool{false, true} {
				predicate := connectorInboundReapplyBlockerPredicateWithIdentity(
					scope, workItemAuthority, exactIdentity,
				)
				query := "SELECT 1 FROM _System_InstalledPack p WHERE NOT " + predicate
				if _, _, err := p.Parse(query, "", ""); err != nil {
					t.Fatalf("reapply blocker predicate does not parse for scope=%d workItemAuthority=%t exactIdentity=%t: %v\nSQL:\n%s", scope, workItemAuthority, exactIdentity, err, query)
				}
			}
		}
	}
}

func TestConnectorInboundExactIncarnationBlockerUsesBinaryPackIdentityThroughout(t *testing.T) {
	for _, scope := range []connectorInboundReapplyScope{
		connectorPackInboundReapplyScope,
		companionAppPackInboundReapplyScope,
	} {
		predicate := connectorInboundExactIncarnationBlockerPredicate(scope, true)
		for _, match := range [][2]string{
			{"queued_event.InstalledPackId", "w.InstalledPackId"},
			{"direct_work.InstalledPackId", "e.InstalledPackId"},
			{"projection.InstalledPackId", "e.InstalledPackId"},
			{"selected.InstalledPackId", "e.InstalledPackId"},
			{"COALESCE(selected.CompanionPackInstallId, '')", "COALESCE(e.CompanionPackInstallId, '')"},
		} {
			want := connectorInboundPackInstallIDMatch(match[0], match[1], true)
			if !strings.Contains(predicate, want) {
				t.Fatalf("exact blocker for scope=%d is missing %q:\n%s", scope, want, predicate)
			}
		}
	}
}