package controlplane

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

	bootstate "github.com/nexuscrm/db/pkg/bootstate"
)

const migrationIdConnectorEventEnqueueOrdinalBackfill = "20260619_connector_event_enqueue_ordinal_backfill"

// MigrateConnectorEventEnqueueOrdinalBackfill stamps warm-upgrade
// _System_ConnectorEvent rows that inherited EnqueueOrdinal=0 when the column
// was added. The runtime claim guard treats EnqueueOrdinal as the same-source
// queue order, so leaving all legacy rows at 0 lets the Id tiebreak override the
// historical queue order.
func MigrateConnectorEventEnqueueOrdinalBackfill(ctx context.Context, db *sql.DB) (int, error) {
	if db == nil {
		return 0, fmt.Errorf("MigrateConnectorEventEnqueueOrdinalBackfill requires a database")
	}
	applied, err := bootstate.MigrationApplied(ctx, db, migrationIdConnectorEventEnqueueOrdinalBackfill)
	if err != nil {
		return 0, err
	}
	if applied {
		remaining, err := bootstate.ConnectorEventEnqueueOrdinalLiveZeroCount(ctx, db)
		if err != nil {
			return 0, err
		}
		if remaining == 0 {
			log.Println("   ⏭  MigrateConnectorEventEnqueueOrdinalBackfill: already applied; skipping")
			return 0, nil
		}
		log.Printf("   🧼 MigrateConnectorEventEnqueueOrdinalBackfill: completed receipt exists but %d live zero-ordinal row(s) remain; repairing", remaining)
	}

	ready, err := bootstate.ConnectorEventEnqueueOrdinalBackfillColumnsReady(ctx, db)
	if err != nil {
		return 0, err
	}
	if !ready {
		return 0, fmt.Errorf("MigrateConnectorEventEnqueueOrdinalBackfill: required _System_ConnectorEvent columns missing after reconciliation")
	}

	updated, err := backfillConnectorEventEnqueueOrdinals(ctx, db)
	if err != nil {
		return updated, fmt.Errorf("MigrateConnectorEventEnqueueOrdinalBackfill: %w", err)
	}
	if err := bootstate.RecordMigrationApplied(
		ctx,
		db,
		migrationIdConnectorEventEnqueueOrdinalBackfill,
		1,
		"Backfill _System_ConnectorEvent.EnqueueOrdinal for warm-upgrade rows before same-source queue ordering becomes authoritative.",
	); err != nil {
		return updated, err
	}
	if updated > 0 {
		log.Printf("   🧼 MigrateConnectorEventEnqueueOrdinalBackfill: backfilled %d connector event row(s)", updated)
	}
	return updated, nil
}

func backfillConnectorEventEnqueueOrdinals(ctx context.Context, db *sql.DB) (int, error) {
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return 0, fmt.Errorf("begin transaction: %w", err)
	}
	committed := false
	defer func() {
		if !committed {
			_ = tx.Rollback()
		}
	}()

	rows, err := tx.QueryContext(ctx, `
SELECT DISTINCT InstalledPackId
FROM _System_ConnectorEvent
WHERE EnqueueOrdinal = 0
  AND Status NOT IN ('processed', 'dead')
ORDER BY InstalledPackId ASC`)
	if err != nil {
		return 0, fmt.Errorf("query packs with zero enqueue ordinals: %w", err)
	}
	defer rows.Close()

	packOrder := make([]string, 0)
	for rows.Next() {
		var packInstallID string
		if err := rows.Scan(&packInstallID); err != nil {
			return 0, fmt.Errorf("scan pack with zero enqueue ordinal: %w", err)
		}
		packOrder = append(packOrder, packInstallID)
	}
	if err := rows.Err(); err != nil {
		return 0, fmt.Errorf("iterate packs with zero enqueue ordinals: %w", err)
	}
	if err := rows.Close(); err != nil {
		return 0, fmt.Errorf("close packs with zero enqueue ordinals: %w", err)
	}

	updated := 0
	for _, packInstallID := range packOrder {
		eventRows, err := tx.QueryContext(ctx, `
SELECT Id, EnqueueOrdinal
FROM _System_ConnectorEvent
WHERE InstalledPackId = ?
  AND Status NOT IN ('processed', 'dead')
ORDER BY CreatedDate ASC, DeliverySequence ASC, Id ASC`, packInstallID)
		if err != nil {
			return updated, fmt.Errorf("query enqueue ordinal space for pack %s: %w", packInstallID, err)
		}

		type eventOrdinal struct {
			id      string
			ordinal int64
		}
		events := make([]eventOrdinal, 0)
		for eventRows.Next() {
			var ev eventOrdinal
			if err := eventRows.Scan(&ev.id, &ev.ordinal); err != nil {
				_ = eventRows.Close()
				return updated, fmt.Errorf("scan enqueue ordinal row for pack %s: %w", packInstallID, err)
			}
			events = append(events, ev)
		}
		if err := eventRows.Err(); err != nil {
			_ = eventRows.Close()
			return updated, fmt.Errorf("iterate enqueue ordinal rows for pack %s: %w", packInstallID, err)
		}
		if err := eventRows.Close(); err != nil {
			return updated, fmt.Errorf("close enqueue ordinal rows for pack %s: %w", packInstallID, err)
		}

		for i, ev := range events {
			ordinal := int64(i + 1)
			if ev.ordinal == ordinal {
				continue
			}
			res, err := tx.ExecContext(ctx, `
UPDATE _System_ConnectorEvent
SET EnqueueOrdinal = ?, LastModifiedDate = NOW()
WHERE Id = ?`, ordinal, ev.id)
			if err != nil {
				return updated, fmt.Errorf("backfill enqueue ordinal for connector event %s: %w", ev.id, err)
			}
			if n, err := res.RowsAffected(); err != nil {
				return updated, fmt.Errorf("enqueue ordinal backfill rows affected for connector event %s: %w", ev.id, err)
			} else if n > 0 {
				updated += int(n)
			}
		}
	}

	var remaining int
	if err := tx.QueryRowContext(ctx, `
SELECT COUNT(*)
FROM _System_ConnectorEvent
WHERE EnqueueOrdinal = 0
  AND Status NOT IN ('processed', 'dead')`).Scan(&remaining); err != nil {
		return updated, fmt.Errorf("enqueue ordinal backfill closure check: %w", err)
	}
	if remaining != 0 {
		return updated, fmt.Errorf("enqueue ordinal backfill: %d legacy row(s) remain with EnqueueOrdinal=0", remaining)
	}

	if err := tx.Commit(); err != nil {
		return updated, fmt.Errorf("commit: %w", err)
	}
	committed = true
	return updated, nil
}