package bootstrapimpl

import (
	"os"
	"regexp"
	"strings"
	"testing"

	"github.com/DATA-DOG/go-sqlmock"
	"github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/persistence"
)

func recordEventScopeDigestTestColumns() *sqlmock.Rows {
	return sqlmock.NewRows([]string{"COLUMN_NAME"}).
		AddRow("Id").
		AddRow("OrgId").
		AddRow("AppInstallationId").
		AddRow("RuntimeCapabilityId").
		AddRow("ExecutionScopeDigest")
}

func expectRecordEventScopeDigestColumns(mock sqlmock.Sqlmock) {
	mock.ExpectQuery("SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS").
		WithArgs(recordEventOutboxTable, "Id", "OrgId", "AppInstallationId", "RuntimeCapabilityId", recordEventExecutionScopeDigestColumn).
		WillReturnRows(recordEventScopeDigestTestColumns())
}

func expectNoIncompleteRecordEventScopeDigest(mock sqlmock.Sqlmock) {
	mock.ExpectQuery(regexp.QuoteMeta("SELECT Id\nFROM _System_RecordEventOutbox FORCE INDEX (idx_record_event_outbox_pending)")).
		WillReturnRows(sqlmock.NewRows([]string{"Id"}))
}

func TestReconcileRecordEventExecutionScopeDigests(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	expectRecordEventScopeDigestColumns(mock)
	scope := persistence.RuntimeWorkScope{OrgID: "org", AppInstallationID: "app", RuntimeCapabilityID: "runtime.records.mutate.v1"}
	digest, err := scope.Digest()
	if err != nil {
		t.Fatal(err)
	}
	expectNoIncompleteRecordEventScopeDigest(mock)
	mock.ExpectQuery(regexp.QuoteMeta("SELECT OrgId, AppInstallationId, RuntimeCapabilityId,")).
		WillReturnRows(sqlmock.NewRows([]string{"OrgId", "AppInstallationId", "RuntimeCapabilityId", "Minimum", "Maximum", "Missing"}).
			AddRow(scope.OrgID, scope.AppInstallationID, scope.RuntimeCapabilityID, strings.Repeat("f", 64), strings.Repeat("f", 64), 0))
	mock.ExpectExec(regexp.QuoteMeta("UPDATE _System_RecordEventOutbox FORCE INDEX (idx_record_event_outbox_pending)")).
		WithArgs(digest, scope.OrgID, scope.AppInstallationID, scope.RuntimeCapabilityID, digest, recordEventScopeDigestBackfillBatch).
		WillReturnResult(sqlmock.NewResult(0, 1))
	if err := reconcileRecordEventExecutionScopeDigests(db); err != nil {
		t.Fatal(err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestReconcileRecordEventExecutionScopeDigestsIsIdempotent(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	expectRecordEventScopeDigestColumns(mock)
	scope := persistence.RuntimeWorkScope{OrgID: "org", AppInstallationID: "app", RuntimeCapabilityID: "runtime.records.mutate.v1"}
	digest, err := scope.Digest()
	if err != nil {
		t.Fatal(err)
	}
	expectNoIncompleteRecordEventScopeDigest(mock)
	mock.ExpectQuery(regexp.QuoteMeta("SELECT OrgId, AppInstallationId, RuntimeCapabilityId,")).
		WillReturnRows(sqlmock.NewRows([]string{"OrgId", "AppInstallationId", "RuntimeCapabilityId", "Minimum", "Maximum", "Missing"}).
			AddRow(scope.OrgID, scope.AppInstallationID, scope.RuntimeCapabilityID, digest, digest, 0))
	if err := reconcileRecordEventExecutionScopeDigests(db); err != nil {
		t.Fatal(err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestReconcileRecordEventExecutionScopeDigestsFlushesBoundedPages(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	expectRecordEventScopeDigestColumns(mock)
	scope := persistence.RuntimeWorkScope{OrgID: "org", AppInstallationID: "app", RuntimeCapabilityID: "runtime.records.mutate.v1"}
	digest, err := scope.Digest()
	if err != nil {
		t.Fatal(err)
	}
	expectNoIncompleteRecordEventScopeDigest(mock)
	mock.ExpectQuery(regexp.QuoteMeta("SELECT OrgId, AppInstallationId, RuntimeCapabilityId,")).
		WillReturnRows(sqlmock.NewRows([]string{"OrgId", "AppInstallationId", "RuntimeCapabilityId", "Minimum", "Maximum", "Missing"}).
			AddRow(scope.OrgID, scope.AppInstallationID, scope.RuntimeCapabilityID, nil, nil, recordEventScopeDigestBackfillBatch))
	mock.ExpectExec(regexp.QuoteMeta("UPDATE _System_RecordEventOutbox FORCE INDEX (idx_record_event_outbox_pending)")).
		WithArgs(digest, scope.OrgID, scope.AppInstallationID, scope.RuntimeCapabilityID, digest, recordEventScopeDigestBackfillBatch).
		WillReturnResult(sqlmock.NewResult(0, recordEventScopeDigestBackfillBatch))
	mock.ExpectExec(regexp.QuoteMeta("UPDATE _System_RecordEventOutbox FORCE INDEX (idx_record_event_outbox_pending)")).
		WithArgs(digest, scope.OrgID, scope.AppInstallationID, scope.RuntimeCapabilityID, digest, recordEventScopeDigestBackfillBatch).
		WillReturnResult(sqlmock.NewResult(0, 0))
	if err := reconcileRecordEventExecutionScopeDigests(db); err != nil {
		t.Fatal(err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestReconcileRecordEventExecutionScopeDigestsRejectsDigestWithoutAuthority(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	expectRecordEventScopeDigestColumns(mock)
	mock.ExpectQuery(regexp.QuoteMeta("SELECT Id\nFROM _System_RecordEventOutbox FORCE INDEX (idx_record_event_outbox_pending)")).
		WillReturnRows(sqlmock.NewRows([]string{"Id"}).AddRow("corrupt"))
	if err := reconcileRecordEventExecutionScopeDigests(db); err == nil {
		t.Fatal("backfill accepted a digest without recoverable explicit scope authority")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestBackfillRecordEventExecutionScopeDigestsSkipsUnscopedLegacyRowsByQueryContract(t *testing.T) {
	if !strings.Contains(recordEventIncompleteDigestScopeSQL, "ExecutionScopeDigest IS NOT NULL") ||
		!strings.Contains(recordEventIncompleteDigestScopeSQL, "OrgId IS NULL OR AppInstallationId IS NULL OR RuntimeCapabilityId IS NULL") {
		t.Fatal("digest-only rows must remain a fail-closed reconciliation error")
	}
	for _, query := range []string{
		recordEventIncompleteDigestScopeSQL,
		recordEventScopeDigestSummarySQL,
		recordEventScopeDigestRepairSQL,
	} {
		if !strings.Contains(query, "Status IN ('pending', 'processing', 'failed')") ||
			!strings.Contains(query, "FORCE INDEX (idx_record_event_outbox_pending)") ||
			strings.Contains(query, "'processed'") {
			t.Fatal("reconciliation must be bounded to executable or repairable rows through the lifecycle index")
		}
	}
	if !strings.Contains(recordEventScopeDigestSummarySQL, "OrgId IS NOT NULL") ||
		!strings.Contains(recordEventScopeDigestSummarySQL, "AppInstallationId IS NOT NULL") ||
		!strings.Contains(recordEventScopeDigestSummarySQL, "RuntimeCapabilityId IS NOT NULL") {
		t.Fatal("backfill query no longer keeps fully unscoped legacy rows out of the projection")
	}
	if !strings.Contains(recordEventScopeDigestSummarySQL, "MIN(ExecutionScopeDigest)") ||
		!strings.Contains(recordEventScopeDigestSummarySQL, "MAX(ExecutionScopeDigest)") {
		t.Fatal("scope summary must retain mismatch detection for materialized digests")
	}
}

func TestL0DDLOnlyLifecycleReconcilesRecordEventExecutionScopeProjection(t *testing.T) {
	data, err := os.ReadFile("schema.go")
	if err != nil {
		t.Fatalf("read schema.go: %v", err)
	}
	source := string(data)
	call := "reconcileRecordEventExecutionScopeProjection("
	if strings.Count(source, call) != 1 {
		t.Fatalf("record-event physical projection must have one schema lifecycle owner; calls=%d", strings.Count(source, call))
	}
	batchTables := strings.Index(source, `runPhase("phase1_batch_tables"`)
	projection := strings.Index(source, call)
	if batchTables < 0 || projection < batchTables {
		t.Fatal("record-event projection must reconcile after physical batch tables exist")
	}
	nextFullOnly := strings.Index(source[batchTables:], "if mode.includeSchemaReconciliation")
	if nextFullOnly < 0 || projection > batchTables+nextFullOnly {
		t.Fatal("DDL-only bootstrap must not skip the runtime-critical record-event projection")
	}
	if !strings.Contains(source[projection:], "rawDB, tableDefs, mode.includeManagedMetadata") {
		t.Fatal("DDL-only projection must reconcile DDL without running the managed data backfill")
	}
}

func TestRecordEventExecutionScopeDigestCatalogLifecycle(t *testing.T) {
	for _, def := range GetSystemTableDefinitions() {
		if def.TableName != recordEventOutboxTable {
			continue
		}
		columnFound := false
		for _, column := range def.Columns {
			if column.Name == recordEventExecutionScopeDigestColumn {
				columnFound = column.Nullable && strings.EqualFold(column.Type, "CHAR(64)")
				break
			}
		}
		if !columnFound {
			t.Fatal("record event execution scope digest must be a nullable CHAR(64) projection")
		}
		found := map[string]bool{}
		for _, index := range def.Indices {
			columns := strings.Join(index.Columns, ",")
			found[index.Name] = !index.Unique &&
				((index.Name == recordEventPendingIndex && columns == "Status,CreatedDate") ||
					(index.Name == recordEventClaimIndex && columns == "ExecutionScopeDigest,Status,CreatedDate,Id"))
		}
		if found[recordEventPendingIndex] && found[recordEventClaimIndex] {
			return
		}
		t.Fatal("record event lifecycle indexes are missing or have the wrong ordered columns")
	}
	t.Fatal("record event outbox table is missing from the kernel schema catalog")
}

func TestGenericIndexReconcilerSkipsRecordEventExecutionScopeIndexes(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()

	expectExistingTableSet(mock, recordEventOutboxTable)
	expectExistingIndexSet(mock, driverIndexRow{
		tableName: recordEventOutboxTable,
		indexName: "wrong_pending_owner",
		nonUnique: 1,
		columns:   "Status,CreatedDate",
	}, driverIndexRow{
		tableName: recordEventOutboxTable,
		indexName: "wrong_claim_owner",
		nonUnique: 1,
		columns:   "ExecutionScopeDigest,Status,CreatedDate,Id",
	})
	definition := schema.TableDefinition{
		TableName: recordEventOutboxTable,
		Indices: []schema.IndexDefinition{
			{
				Name:    recordEventPendingIndex,
				Columns: []string{"Status", "CreatedDate"},
			}, {
				Name: recordEventClaimIndex,
				Columns: []string{
					recordEventExecutionScopeDigestColumn, "Status", "CreatedDate", "Id",
				},
			},
		},
	}
	if err := reconcileSystemTableIndices(db, []schema.TableDefinition{definition}); err != nil {
		t.Fatalf("reconcileSystemTableIndices: %v", err)
	}
	assertSQLMockExpectations(t, mock)
}