package controlplane

import (
	"context"
	"fmt"
	"log"
	"strings"

	"github.com/nexuscrm/automation/triggersub"
	"github.com/nexuscrm/db"
)

type triggerSubscriptionIdentity struct {
	flowDefinitionID  string
	keyPrefix         string
	eventType         string
	recordTriggerType string
}

type triggerSubscriptionRowLoader func(context.Context, triggersub.Executor) ([]triggersub.Row, error)

func VerifyTriggerSubscriptionConsistency(ctx context.Context, db *database.TiDBConnection) error {
	if db == nil {
		return fmt.Errorf("trigger subscription consistency verify requires a database connection")
	}
	rawDB := db.DB()
	if rawDB == nil {
		return fmt.Errorf("trigger subscription consistency verify requires a database handle")
	}

	return verifyTriggerSubscriptionConsistencyWithSnapshot(
		ctx,
		func(ctx context.Context) (triggersub.Executor, func() error, func() error, error) {
			// TiDB rejects START TRANSACTION READ ONLY unless noop functions are enabled.
			// This verify path still takes one consistent snapshot by running both reads
			// inside a single transaction and never issuing writes through it.
			tx, err := rawDB.BeginTx(ctx, nil)
			if err != nil {
				return nil, nil, nil, err
			}
			return tx, tx.Commit, tx.Rollback, nil
		},
		triggersub.ExpectedActiveRows,
		loadActiveTriggerSubscriptionRows,
	)
}

func verifyTriggerSubscriptionConsistencyWithSnapshot(
	ctx context.Context,
	beginSnapshot func(context.Context) (triggersub.Executor, func() error, func() error, error),
	loadExpectedRows triggerSubscriptionRowLoader,
	loadActualRows triggerSubscriptionRowLoader,
) error {
	if beginSnapshot == nil {
		return fmt.Errorf("trigger subscription consistency verify requires a snapshot reader")
	}
	if loadExpectedRows == nil || loadActualRows == nil {
		return fmt.Errorf("trigger subscription consistency verify requires row loaders")
	}

	snapshot, commit, rollback, err := beginSnapshot(ctx)
	if err != nil {
		return fmt.Errorf("trigger subscription consistency: begin read-only snapshot: %w", err)
	}
	defer func() {
		if rollback != nil {
			_ = rollback()
		}
	}()

	expectedRows, err := loadExpectedRows(ctx, snapshot)
	if err != nil {
		return fmt.Errorf("trigger subscription consistency: resolve canonical rows: %w", err)
	}
	actualRows, err := loadActualRows(ctx, snapshot)
	if err != nil {
		return fmt.Errorf("trigger subscription consistency: %w", err)
	}
	if commit != nil {
		if err := commit(); err != nil {
			return fmt.Errorf("trigger subscription consistency: commit read-only snapshot: %w", err)
		}
		rollback = nil
	}

	expectedByIdentity := make(map[triggerSubscriptionIdentity]triggersub.Row, len(expectedRows))
	for _, row := range expectedRows {
		expectedByIdentity[triggerSubscriptionIdentityFor(row)] = row
	}
	actualByIdentity := make(map[triggerSubscriptionIdentity]triggersub.Row, len(actualRows))
	for _, row := range actualRows {
		actualByIdentity[triggerSubscriptionIdentityFor(row)] = row
	}

	var missing []string
	var stale []string
	var tupleMismatches []string

	for identity, expected := range expectedByIdentity {
		actual, ok := actualByIdentity[identity]
		if !ok {
			missing = append(missing, formatTriggerSubscriptionRow(expected))
			continue
		}
		if strings.TrimSpace(actual.FlowVersionId) != strings.TrimSpace(expected.FlowVersionId) {
			tupleMismatches = append(tupleMismatches,
				fmt.Sprintf("%s actual.FlowVersionId=%q want=%q",
					formatTriggerSubscriptionIdentity(identity),
					actual.FlowVersionId,
					expected.FlowVersionId,
				),
			)
		}
	}
	for identity, actual := range actualByIdentity {
		if _, ok := expectedByIdentity[identity]; !ok {
			stale = append(stale, formatTriggerSubscriptionRow(actual))
		}
	}

	if len(missing) == 0 && len(stale) == 0 && len(tupleMismatches) == 0 {
		log.Println("✅ trigger subscription shadow state matches canonical projection")
		return nil
	}

	var details []string
	if len(missing) > 0 {
		details = append(details, "missing rows: ["+strings.Join(capStrings(missing), "; ")+"]")
	}
	if len(stale) > 0 {
		details = append(details, "stale rows: ["+strings.Join(capStrings(stale), "; ")+"]")
	}
	if len(tupleMismatches) > 0 {
		details = append(details, "tuple mismatches: ["+strings.Join(capStrings(tupleMismatches), "; ")+"]")
	}

	return fmt.Errorf("trigger subscription canonical consistency failed — %s — run 'cmd/setup apply' to rebuild shadow state",
		strings.Join(details, " | "))
}

func loadActiveTriggerSubscriptionRows(ctx context.Context, exec triggersub.Executor) ([]triggersub.Row, error) {
	rows, err := exec.QueryContext(ctx, `
		SELECT Id, KeyPrefix, EventType, RecordTriggerType, FlowDefinitionId, FlowVersionId, IsActive
		FROM _System_TriggerSubscription
		WHERE IsActive = 1
		ORDER BY FlowDefinitionId, KeyPrefix, EventType, RecordTriggerType`)
	if err != nil {
		return nil, fmt.Errorf("query active trigger subscription rows: %w", err)
	}
	defer rows.Close()

	var out []triggersub.Row
	for rows.Next() {
		var row triggersub.Row
		if err := rows.Scan(&row.Id, &row.KeyPrefix, &row.EventType, &row.RecordTriggerType, &row.FlowDefinitionId, &row.FlowVersionId, &row.IsActive); err != nil {
			return nil, fmt.Errorf("scan active trigger subscription row: %w", err)
		}
		out = append(out, row)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate active trigger subscription rows: %w", err)
	}
	return out, nil
}

func triggerSubscriptionIdentityFor(row triggersub.Row) triggerSubscriptionIdentity {
	return triggerSubscriptionIdentity{
		flowDefinitionID:  strings.TrimSpace(row.FlowDefinitionId),
		keyPrefix:         strings.TrimSpace(row.KeyPrefix),
		eventType:         strings.TrimSpace(row.EventType),
		recordTriggerType: strings.TrimSpace(row.RecordTriggerType),
	}
}

func formatTriggerSubscriptionIdentity(identity triggerSubscriptionIdentity) string {
	return fmt.Sprintf("FlowDefinitionId=%q KeyPrefix=%q EventType=%q RecordTriggerType=%q",
		identity.flowDefinitionID,
		identity.keyPrefix,
		identity.eventType,
		identity.recordTriggerType,
	)
}

func formatTriggerSubscriptionRow(row triggersub.Row) string {
	return fmt.Sprintf("%s FlowVersionId=%q",
		formatTriggerSubscriptionIdentity(triggerSubscriptionIdentityFor(row)),
		row.FlowVersionId,
	)
}

func capStrings(items []string) []string {
	if len(items) <= 10 {
		return items
	}
	return append(items[:10], fmt.Sprintf("... %d more", len(items)-10))
}