package bootstrapimpl

import (
	"database/sql"
	"fmt"
	"log"
	"strings"

	"github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/persistence"
)

const (
	recordEventOutboxTable                = "_System_RecordEventOutbox"
	recordEventExecutionScopeDigestColumn = "ExecutionScopeDigest"
	recordEventPendingIndex               = "idx_record_event_outbox_pending"
	recordEventClaimIndex                 = "idx_record_event_outbox_claim"
	recordEventScopeDigestBackfillBatch   = 256
)

const recordEventIncompleteDigestScopeSQL = `SELECT Id
FROM _System_RecordEventOutbox FORCE INDEX (idx_record_event_outbox_pending)
WHERE Status IN ('pending', 'processing', 'failed')
  AND ExecutionScopeDigest IS NOT NULL
  AND (OrgId IS NULL OR AppInstallationId IS NULL OR RuntimeCapabilityId IS NULL)
LIMIT 1`

const recordEventScopeDigestSummarySQL = `SELECT OrgId, AppInstallationId, RuntimeCapabilityId,
       MIN(ExecutionScopeDigest), MAX(ExecutionScopeDigest),
       SUM(CASE WHEN ExecutionScopeDigest IS NULL THEN 1 ELSE 0 END)
FROM _System_RecordEventOutbox FORCE INDEX (idx_record_event_outbox_pending)
WHERE Status IN ('pending', 'processing', 'failed')
  AND OrgId IS NOT NULL
  AND AppInstallationId IS NOT NULL
  AND RuntimeCapabilityId IS NOT NULL
GROUP BY OrgId, AppInstallationId, RuntimeCapabilityId`

const recordEventScopeDigestRepairSQL = `UPDATE _System_RecordEventOutbox FORCE INDEX (idx_record_event_outbox_pending)
SET ExecutionScopeDigest = ?
WHERE Status IN ('pending', 'processing', 'failed')
  AND OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
  AND NOT (ExecutionScopeDigest <=> ?)
LIMIT ?`

func reconcileRecordEventExecutionScopeProjection(
	rawDB *sql.DB,
	tableDefs []schema.TableDefinition,
	includeDataBackfill bool,
) error {
	definition, column, pendingIndex, claimIndex, err := recordEventExecutionScopeRequirement(tableDefs)
	if err != nil {
		return err
	}
	if err := ReconcileTableColumns(rawDB, []schema.TableDefinition{definition}); err != nil {
		return fmt.Errorf("reconcile %s.%s: %w", definition.TableName, column.Name, err)
	}
	if err := verifyRecordEventExecutionScopeColumn(rawDB, definition, column); err != nil {
		return err
	}
	if err := reconcileExactManagedIndex(rawDB, pendingIndex); err != nil {
		return fmt.Errorf("reconcile record-event lifecycle index: %w", err)
	}
	if err := reconcileExactManagedIndex(rawDB, claimIndex); err != nil {
		return fmt.Errorf("reconcile record-event claim index: %w", err)
	}
	if includeDataBackfill {
		if err := reconcileRecordEventExecutionScopeDigests(rawDB); err != nil {
			return err
		}
	}
	return nil
}

func recordEventExecutionScopeRequirement(
	tableDefs []schema.TableDefinition,
) (schema.TableDefinition, schema.ColumnDefinition, requiredManagedIndex, requiredManagedIndex, error) {
	var (
		definition  schema.TableDefinition
		column      schema.ColumnDefinition
		tableCount  int
		columnCount int
	)
	for _, candidate := range tableDefs {
		if !strings.EqualFold(candidate.TableName, recordEventOutboxTable) {
			continue
		}
		tableCount++
		definition = candidate
		for _, candidateColumn := range candidate.Columns {
			if strings.EqualFold(candidateColumn.Name, recordEventExecutionScopeDigestColumn) {
				columnCount++
				column = candidateColumn
			}
		}
	}
	if tableCount != 1 {
		return definition, column, requiredManagedIndex{}, requiredManagedIndex{}, fmt.Errorf(
			"managed catalog has %d definitions for required table %s, want 1",
			tableCount, recordEventOutboxTable,
		)
	}
	if columnCount != 1 {
		return definition, column, requiredManagedIndex{}, requiredManagedIndex{}, fmt.Errorf(
			"managed catalog has %d definitions for required column %s.%s, want 1",
			columnCount, recordEventOutboxTable, recordEventExecutionScopeDigestColumn,
		)
	}
	pendingIndex, err := managedIndexRequirement(
		tableDefs, recordEventOutboxTable, recordEventPendingIndex,
	)
	if err != nil {
		return definition, column, requiredManagedIndex{}, requiredManagedIndex{}, err
	}
	claimIndex, err := managedIndexRequirement(
		tableDefs, recordEventOutboxTable, recordEventClaimIndex,
	)
	return definition, column, pendingIndex, claimIndex, err
}

func isRecordEventExecutionScopeIndex(tableName string, index schema.IndexDefinition) bool {
	if !strings.EqualFold(tableName, recordEventOutboxTable) {
		return false
	}
	return strings.EqualFold(index.Name, recordEventPendingIndex) ||
		strings.EqualFold(index.Name, recordEventClaimIndex)
}

func verifyRecordEventExecutionScopeColumn(
	rawDB *sql.DB,
	definition schema.TableDefinition,
	column schema.ColumnDefinition,
) error {
	shapes, err := LoadPhysicalColumnShapes(rawDB)
	if err != nil {
		return fmt.Errorf("read back %s.%s: %w", definition.TableName, column.Name, err)
	}
	actual, ok := shapes[strings.ToLower(definition.TableName)][strings.ToLower(column.Name)]
	if !ok || !ColumnShapeMatches(actual, column) {
		return fmt.Errorf(
			"required projection column %s.%s did not converge to catalog shape %s nullable=%t",
			definition.TableName, column.Name, DesiredColumnType(column), column.Nullable,
		)
	}
	return nil
}

// reconcileRecordEventExecutionScopeDigests materializes the bounded query
// projection used by independently deployed Runtime Workers. The three
// explicit scope columns remain the recoverable authority; unscoped legacy
// rows stay NULL and are not adopted by any Runtime partition.
//
// Caller must hold the schema reconciliation advisory lock and run this after
// the physical table, columns, and claim index exist. TiDB groups only rows
// that can still execute (pending/processing) or be explicitly repaired
// (failed), using the lifecycle index, and returns one digest summary per
// runtime partition. Processed rows are terminal; replay authors a new row, so
// completed history is outside this projection's reconciliation boundary. Go
// remains the sole digest implementation, and only a scope whose summary is
// missing or disagrees with it enters the bounded repair path.
func reconcileRecordEventExecutionScopeDigests(rawDB *sql.DB) error {
	ready, err := recordEventScopeDigestBackfillColumnsReady(rawDB)
	if err != nil {
		return err
	}
	if !ready {
		return fmt.Errorf("%s execution-scope projection columns are not present after schema reconciliation", recordEventOutboxTable)
	}

	var incompleteID string
	err = rawDB.QueryRow(recordEventIncompleteDigestScopeSQL).Scan(&incompleteID)
	if err != nil && err != sql.ErrNoRows {
		return fmt.Errorf("querying %s incomplete execution scopes: %w", recordEventOutboxTable, err)
	}
	if err == nil {
		return fmt.Errorf("%s row %s has a digest without a complete explicit execution scope", recordEventOutboxTable, incompleteID)
	}

	rows, err := rawDB.Query(recordEventScopeDigestSummarySQL)
	if err != nil {
		return fmt.Errorf("querying %s execution scope summaries: %w", recordEventOutboxTable, err)
	}
	type scopeRepair struct {
		scope    persistence.RuntimeWorkScope
		expected string
	}
	repairs := make([]scopeRepair, 0)
	for rows.Next() {
		var scope persistence.RuntimeWorkScope
		var minimum, maximum sql.NullString
		var missing int64
		if err := rows.Scan(
			&scope.OrgID, &scope.AppInstallationID, &scope.RuntimeCapabilityID,
			&minimum, &maximum, &missing,
		); err != nil {
			_ = rows.Close()
			return fmt.Errorf("scanning %s execution scope summary: %w", recordEventOutboxTable, err)
		}
		expected, err := scope.Digest()
		if err != nil {
			_ = rows.Close()
			return fmt.Errorf("%s has an invalid explicit execution scope: %w", recordEventOutboxTable, err)
		}
		if missing > 0 || !minimum.Valid || !maximum.Valid || minimum.String != expected || maximum.String != expected {
			repairs = append(repairs, scopeRepair{scope: scope, expected: expected})
		}
	}
	if err := rows.Err(); err != nil {
		_ = rows.Close()
		return fmt.Errorf("iterating %s execution scope summaries: %w", recordEventOutboxTable, err)
	}
	if err := rows.Close(); err != nil {
		return fmt.Errorf("closing %s execution scope summaries: %w", recordEventOutboxTable, err)
	}

	updated := int64(0)
	for _, repair := range repairs {
		for {
			result, err := rawDB.Exec(
				recordEventScopeDigestRepairSQL,
				repair.expected,
				repair.scope.OrgID,
				repair.scope.AppInstallationID,
				repair.scope.RuntimeCapabilityID,
				repair.expected,
				recordEventScopeDigestBackfillBatch,
			)
			if err != nil {
				return fmt.Errorf("repairing %s execution scope digest: %w", recordEventOutboxTable, err)
			}
			count, err := result.RowsAffected()
			if err != nil {
				return fmt.Errorf("reading %s execution scope repair count: %w", recordEventOutboxTable, err)
			}
			updated += count
			if count < recordEventScopeDigestBackfillBatch {
				break
			}
		}
	}
	if updated > 0 {
		log.Printf("   ✅ Reconciled %d _System_RecordEventOutbox ExecutionScopeDigest projection values", updated)
	}
	return nil
}

func recordEventScopeDigestBackfillColumnsReady(rawDB *sql.DB) (bool, error) {
	required := map[string]bool{
		"id":                  false,
		"orgid":               false,
		"appinstallationid":   false,
		"runtimecapabilityid": false,
		strings.ToLower(recordEventExecutionScopeDigestColumn): false,
	}
	rows, err := rawDB.Query(
		"SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND COLUMN_NAME IN (?, ?, ?, ?, ?)",
		recordEventOutboxTable, "Id", "OrgId", "AppInstallationId", "RuntimeCapabilityId", recordEventExecutionScopeDigestColumn,
	)
	if err != nil {
		return false, fmt.Errorf("querying %s columns for execution scope digest backfill: %w", recordEventOutboxTable, err)
	}
	defer rows.Close()
	for rows.Next() {
		var name string
		if err := rows.Scan(&name); err != nil {
			return false, fmt.Errorf("scanning %s column for execution scope digest backfill: %w", recordEventOutboxTable, err)
		}
		key := strings.ToLower(name)
		if _, ok := required[key]; ok {
			required[key] = true
		}
	}
	if err := rows.Err(); err != nil {
		return false, fmt.Errorf("iterating %s columns for execution scope digest backfill: %w", recordEventOutboxTable, err)
	}
	for _, present := range required {
		if !present {
			return false, nil
		}
	}
	return true, nil
}