package persistence

import (
	"context"
	"fmt"
	"sort"
	"strings"

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

const (
	scheduledRunReservationTable  = "_System_ScheduledRunReservation"
	scheduledRunBatchDigestColumn = "BatchIdentityDigest"
	rollupDeltaTable              = "_System_RollupDelta"
	rollupDeltaScopeDigestColumn  = "ScopeDigest"
)

var workerQueueRequiredIndexes = map[string]map[string][]string{
	scheduledRunReservationTable: {
		"idx_scheduled_run_reservation_scope_claim": {"OrgId", "AppInstallationId", "RuntimeCapabilityId", "Status", "LeaseExpiresAt"},
		"idx_scheduled_run_reservation_scope_due":   {"OrgId", "AppInstallationId", "RuntimeCapabilityId", "Status", "DueAt"},
		"idx_scheduled_run_reservation_batch":       {"BatchIdentityDigest"},
	},
	rollupDeltaTable: {
		"idx_rollup_delta_scope_claim":       {"ScopeDigest", "Status", "NextAttemptAt", "CreatedDate", "Id"},
		"idx_rollup_delta_scope_parent_lane": {"ScopeDigest", "ParentObject", "ParentRecordId", "Status", "NextAttemptAt", "CreatedDate"},
	},
}

func declaresWorkerQueueIndexes(def schema.TableDefinition) bool {
	want, ok := workerQueueRequiredIndexes[def.TableName]
	if !ok {
		return false
	}
	found := make(map[string]bool, len(want))
	for _, index := range def.Indices {
		columns, needed := want[index.Name]
		if !needed {
			continue
		}
		if index.Primary || index.Unique || !equalExactColumnNames(index.Columns, columns) {
			return false
		}
		found[index.Name] = true
	}
	return len(found) == len(want)
}

// reconcileWorkerQueueIndexes adds only catalog-declared indexes to warm
// installations. All existing names are preflighted before the first ALTER so
// an incompatible retained shape fails without a partial migration.
func (r *SchemaRepository) reconcileWorkerQueueIndexes(ctx context.Context, def schema.TableDefinition) error {
	want := workerQueueRequiredIndexes[def.TableName]
	declared := make(map[string]schema.IndexDefinition, len(want))
	for _, index := range def.Indices {
		if _, needed := want[index.Name]; needed {
			declared[index.Name] = index
		}
	}
	if len(declared) != len(want) || !declaresWorkerQueueIndexes(def) {
		return fmt.Errorf("reconcile worker queue indexes: %s catalog declaration is incomplete", def.TableName)
	}
	digestColumns, err := r.preflightWorkerQueueDigestColumns(ctx, def)
	if err != nil {
		return err
	}

	names := make([]string, 0, len(want))
	for name := range want {
		names = append(names, name)
	}
	sort.Strings(names)
	missing := make([]schema.IndexDefinition, 0, len(want))
	for _, name := range names {
		index := declared[name]
		columns, unique, hasPrefix, err := inspectManagedPhysicalIndex(ctx, r.db, def.TableName, name)
		if err != nil {
			return fmt.Errorf("reconcile worker queue index %s.%s: %w", def.TableName, name, err)
		}
		if len(columns) == 0 {
			missing = append(missing, index)
			continue
		}
		if unique || hasPrefix || !equalExactColumnNames(columns, index.Columns) {
			return fmt.Errorf(
				"reconcile worker queue index: physical %s.%s has incompatible shape (unique=%t, prefix=%t, columns=%v)",
				def.TableName, name, unique, hasPrefix, columns,
			)
		}
	}
	for _, digestColumn := range digestColumns {
		if _, err := r.db.ExecContext(ctx, fmt.Sprintf(
			"ALTER TABLE `%s` ADD COLUMN %s",
			def.TableName, r.buildColumnDDL(digestColumn, false),
		)); err != nil {
			return fmt.Errorf("reconcile worker queue digest column %s.%s: %w", def.TableName, digestColumn.Name, err)
		}
	}

	for _, index := range missing {
		if _, err := r.db.ExecContext(ctx, fmt.Sprintf(
			"ALTER TABLE `%s` ADD INDEX `%s` (`%s`)",
			def.TableName, index.Name, strings.Join(index.Columns, "`, `"),
		)); err != nil {
			return fmt.Errorf("reconcile worker queue index %s.%s: %w", def.TableName, index.Name, err)
		}
	}
	if def.TableName == scheduledRunReservationTable {
		if err := r.backfillScheduledRunBatchDigests(ctx); err != nil {
			return err
		}
	} else if def.TableName == rollupDeltaTable {
		if err := r.backfillRollupDeltaScopeDigests(ctx); err != nil {
			return err
		}
	}
	for _, name := range names {
		index := declared[name]
		if err := VerifyExactTableIndex(ctx, r.db, def.TableName, name, false, index.Columns); err != nil {
			return fmt.Errorf("verify worker queue index %s.%s: %w", def.TableName, name, err)
		}
	}
	return nil
}

func (r *SchemaRepository) preflightWorkerQueueDigestColumns(
	ctx context.Context,
	def schema.TableDefinition,
) ([]schema.ColumnDefinition, error) {
	wanted := []string(nil)
	switch def.TableName {
	case scheduledRunReservationTable:
		wanted = []string{scheduledRunBatchDigestColumn}
	case rollupDeltaTable:
		wanted = []string{rollupDeltaScopeDigestColumn}
	default:
		return nil, nil
	}
	missing := make([]schema.ColumnDefinition, 0, len(wanted))
	for _, name := range wanted {
		var declared schema.ColumnDefinition
		foundDeclaration := false
		for _, column := range def.Columns {
			if column.Name == name {
				declared = column
				foundDeclaration = true
				break
			}
		}
		if !foundDeclaration || !strings.EqualFold(declared.Type, "CHAR(64)") ||
			!declared.Nullable || declared.Default != "" || declared.PrimaryKey || declared.Unique ||
			!strings.EqualFold(declared.CollationName, "utf8mb4_bin") {
			return nil, fmt.Errorf("reconcile worker queue digest %s.%s: catalog declaration is incompatible", def.TableName, name)
		}
		physical, found, err := inspectIntegrationOutboundDependencyColumn(ctx, r.db, def.TableName, declared.Name)
		if err != nil {
			return nil, fmt.Errorf("reconcile worker queue digest column %s.%s: %w", def.TableName, name, err)
		}
		if !found {
			missing = append(missing, declared)
			continue
		}
		if physical.columnType != "char(64)" || physical.nullable != "YES" ||
			!strings.EqualFold(physical.collation, "utf8mb4_bin") || physical.defaultVal.Valid ||
			strings.TrimSpace(physical.extra) != "" {
			return nil, fmt.Errorf(
				"reconcile worker queue digest %s.%s: physical column has incompatible shape (type=%s, nullable=%s, collation=%s, default=%v, extra=%q)",
				def.TableName, name, physical.columnType, physical.nullable, physical.collation, physical.defaultVal, physical.extra,
			)
		}
	}
	return missing, nil
}

func (r *SchemaRepository) backfillScheduledRunBatchDigests(ctx context.Context) error {
	// scheduledRunIdentityDigest uses an eight-byte big-endian length before
	// each UTF-8 value. OCTET_LENGTH plus 16 hex digits reproduces that exact
	// framing in TiDB; explicit columns remain the authority checked at read.
	expressions := []string{"'scoped-batch'", "OrgId", "AppInstallationId", "RuntimeCapabilityId", "FlowDefinitionId", "FlowVersionId", "ScheduleWindowKey"}
	parts := make([]string, 0, len(expressions)*2)
	for _, expression := range expressions {
		parts = append(parts,
			fmt.Sprintf("UNHEX(LPAD(HEX(OCTET_LENGTH(%s)), 16, '0'))", expression),
			fmt.Sprintf("CONVERT(%s USING binary)", expression),
		)
	}
	if _, err := r.db.ExecContext(ctx, fmt.Sprintf(`UPDATE %s
SET %s = LOWER(SHA2(CONCAT(%s), 256))
WHERE %s IS NULL
  AND OrgId IS NOT NULL AND AppInstallationId IS NOT NULL AND RuntimeCapabilityId IS NOT NULL`,
		scheduledRunReservationTable,
		scheduledRunBatchDigestColumn,
		strings.Join(parts, ", "),
		scheduledRunBatchDigestColumn,
	)); err != nil {
		return fmt.Errorf("reconcile scheduled run batch digest backfill: %w", err)
	}
	var missing int
	if err := r.db.QueryRowContext(ctx, fmt.Sprintf(`SELECT COUNT(*) FROM %s
WHERE OrgId IS NOT NULL AND AppInstallationId IS NOT NULL AND RuntimeCapabilityId IS NOT NULL
  AND %s IS NULL`, scheduledRunReservationTable, scheduledRunBatchDigestColumn)).Scan(&missing); err != nil {
		return fmt.Errorf("verify scheduled run batch digest backfill: %w", err)
	}
	if missing != 0 {
		return fmt.Errorf("verify scheduled run batch digest backfill: %d scoped rows remain unindexed", missing)
	}
	return nil
}

func (r *SchemaRepository) backfillRollupDeltaScopeDigests(ctx context.Context) error {
	// RuntimeWorkScope.Digest uses a two-byte big-endian length before each
	// UTF-8 value. Runtime scope values are capped at 255 bytes, so four hex
	// digits reproduce the exact framing without a wide composite TiDB key.
	expressions := []string{"OrgId", "AppInstallationId", "RuntimeCapabilityId"}
	parts := make([]string, 0, len(expressions)*2)
	for _, expression := range expressions {
		parts = append(parts,
			fmt.Sprintf("UNHEX(LPAD(HEX(OCTET_LENGTH(%s)), 4, '0'))", expression),
			fmt.Sprintf("CONVERT(%s USING binary)", expression),
		)
	}
	if _, err := r.db.ExecContext(ctx, fmt.Sprintf(`UPDATE %s
SET %s = LOWER(SHA2(CONCAT(%s), 256))
WHERE %s IS NULL
  AND OrgId IS NOT NULL AND AppInstallationId IS NOT NULL AND RuntimeCapabilityId IS NOT NULL`,
		rollupDeltaTable, rollupDeltaScopeDigestColumn, strings.Join(parts, ", "),
		rollupDeltaScopeDigestColumn,
	)); err != nil {
		return fmt.Errorf("reconcile rollup delta scope digest backfill: %w", err)
	}
	var missing int
	if err := r.db.QueryRowContext(ctx, fmt.Sprintf(`SELECT COUNT(*) FROM %s
WHERE OrgId IS NOT NULL AND AppInstallationId IS NOT NULL AND RuntimeCapabilityId IS NOT NULL
  AND %s IS NULL`, rollupDeltaTable, rollupDeltaScopeDigestColumn)).Scan(&missing); err != nil {
		return fmt.Errorf("verify rollup delta scope digest backfill: %w", err)
	}
	if missing != 0 {
		return fmt.Errorf("verify rollup delta scope digest backfill: %d scoped rows remain unindexed", missing)
	}
	return nil
}