package persistence

import (
	"crypto/sha256"
	"encoding/binary"
	"encoding/hex"
	"strings"
	"testing"

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

func TestWorkerQueueIndexesWarmUpgradeOnTiDB(t *testing.T) {
	for _, tableName := range []string{scheduledRunReservationTable, rollupDeltaTable} {
		t.Run(tableName, func(t *testing.T) {
			db, cleanup := openSchemaDDLTiDB(t)
			defer cleanup()
			canonical, legacy := workerQueueIndexMigrationDefinitions(t, tableName)
			repository := NewSchemaRepository(db, nil)
			if err := repository.CreatePhysicalTable(t.Context(), legacy); err != nil {
				t.Fatalf("create legacy queue table: %v", err)
			}
			if tableName == scheduledRunReservationTable {
				if _, err := db.Exec(`INSERT INTO _System_ScheduledRunReservation
(Id, OrgId, AppInstallationId, RuntimeCapabilityId, OccurrenceIdentityDigest,
 FlowDefinitionId, FlowVersionId, ScheduleWindowKey, WindowToken, InputKey, State, Status)
VALUES ('legacy-scoped', 'org-α', 'installation-1', 'runtime.automation.execute-flow.v1', ?,
        'definition-1', 'version-1', 'window-1', 'window-1', 'input-1', 'COMPLETED', 'completed')`,
					strings.Repeat("a", 64)); err != nil {
					t.Fatalf("seed scoped scheduled row: %v", err)
				}
			} else if tableName == rollupDeltaTable {
				if _, err := db.Exec(`INSERT INTO _System_RollupDelta
(Id, SourceObject, SourceRecordId, ParentObject, ParentRecordId, RollupField, Delta, DedupeKey,
 OrgId, AppInstallationId, RuntimeCapabilityId, Status)
VALUES ('legacy-scoped', 'Child', 'child-1', 'Parent', 'parent-1', 'Count', 1, ?,
        'org-α', 'installation-1', 'runtime.records.mutate.v1', 'pending')`, strings.Repeat("b", 64)); err != nil {
					t.Fatalf("seed scoped rollup row: %v", err)
				}
			}
			for pass := 1; pass <= 2; pass++ {
				if err := repository.CreatePhysicalTable(t.Context(), canonical); err != nil {
					t.Fatalf("warm upgrade pass %d: %v", pass, err)
				}
				for name, columns := range workerQueueRequiredIndexes[tableName] {
					if err := VerifyExactTableIndex(t.Context(), db, tableName, name, false, columns); err != nil {
						t.Fatalf("verify %s pass %d: %v", name, pass, err)
					}
				}
				if tableName == scheduledRunReservationTable {
					var digest string
					if err := db.QueryRow(`SELECT BatchIdentityDigest FROM _System_ScheduledRunReservation WHERE Id = 'legacy-scoped'`).Scan(&digest); err != nil {
						t.Fatalf("read backfilled batch digest pass %d: %v", pass, err)
					}
					want := scheduledBatchDigestForMigrationTest(
						"org-α", "installation-1", "runtime.automation.execute-flow.v1",
						"definition-1", "version-1", "window-1",
					)
					if digest != want {
						t.Fatalf("backfilled batch digest pass %d = %q, want %q", pass, digest, want)
					}
				} else if tableName == rollupDeltaTable {
					var digest string
					if err := db.QueryRow(`SELECT ScopeDigest FROM _System_RollupDelta WHERE Id = 'legacy-scoped'`).Scan(&digest); err != nil {
						t.Fatalf("read backfilled rollup scope digest pass %d: %v", pass, err)
					}
					want, err := (RuntimeWorkScope{
						OrgID: "org-α", AppInstallationID: "installation-1", RuntimeCapabilityID: "runtime.records.mutate.v1",
					}).Digest()
					if err != nil {
						t.Fatal(err)
					}
					if digest != want {
						t.Fatalf("backfilled rollup scope digest pass %d = %q, want %q", pass, digest, want)
					}
				}
			}
		})
	}
}

func TestScheduledRunQueueIndexWarmUpgradePreflightsAllNamesOnTiDB(t *testing.T) {
	db, cleanup := openSchemaDDLTiDB(t)
	defer cleanup()
	canonical, legacy := workerQueueIndexMigrationDefinitions(t, scheduledRunReservationTable)
	repository := NewSchemaRepository(db, nil)
	if err := repository.CreatePhysicalTable(t.Context(), legacy); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`ALTER TABLE _System_ScheduledRunReservation
ADD INDEX idx_scheduled_run_reservation_scope_due
  (AppInstallationId, OrgId, RuntimeCapabilityId, Status, DueAt)`); err != nil {
		t.Fatal(err)
	}
	if err := repository.CreatePhysicalTable(t.Context(), canonical); err == nil {
		t.Fatal("warm upgrade accepted an incompatible scope index")
	}
	for _, name := range []string{
		"idx_scheduled_run_reservation_batch",
		"idx_scheduled_run_reservation_scope_claim",
	} {
		columns, _, _, err := inspectManagedPhysicalIndex(t.Context(), db, scheduledRunReservationTable, name)
		if err != nil {
			t.Fatal(err)
		}
		if len(columns) != 0 {
			t.Fatalf("failed preflight partially added %s", name)
		}
	}
	var batchColumnCount int
	if err := db.QueryRow(`SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = '_System_ScheduledRunReservation'
  AND COLUMN_NAME = 'BatchIdentityDigest'`).Scan(&batchColumnCount); err != nil {
		t.Fatal(err)
	}
	if batchColumnCount != 0 {
		t.Fatal("failed index preflight partially added BatchIdentityDigest")
	}
}

func TestRollupDeltaQueueIndexWarmUpgradePreflightsAllNamesOnTiDB(t *testing.T) {
	db, cleanup := openSchemaDDLTiDB(t)
	defer cleanup()
	canonical, legacy := workerQueueIndexMigrationDefinitions(t, rollupDeltaTable)
	repository := NewSchemaRepository(db, nil)
	if err := repository.CreatePhysicalTable(t.Context(), legacy); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`ALTER TABLE _System_RollupDelta
ADD INDEX idx_rollup_delta_scope_parent_lane
  (ParentObject, ParentRecordId, Status, NextAttemptAt, CreatedDate)`); err != nil {
		t.Fatal(err)
	}
	if err := repository.CreatePhysicalTable(t.Context(), canonical); err == nil {
		t.Fatal("warm upgrade accepted an incompatible parent-lane index")
	}
	columns, _, _, err := inspectManagedPhysicalIndex(t.Context(), db, rollupDeltaTable, "idx_rollup_delta_scope_claim")
	if err != nil {
		t.Fatal(err)
	}
	if len(columns) != 0 {
		t.Fatal("failed preflight partially added the Rollup scope-claim index")
	}
	var digestColumnCount int
	if err := db.QueryRow(`SELECT COUNT(*) FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = '_System_RollupDelta'
  AND COLUMN_NAME = 'ScopeDigest'`).Scan(&digestColumnCount); err != nil {
		t.Fatal(err)
	}
	if digestColumnCount != 0 {
		t.Fatal("failed index preflight partially added Rollup ScopeDigest")
	}
}

func workerQueueIndexMigrationDefinitions(
	t *testing.T,
	tableName string,
) (schema.TableDefinition, schema.TableDefinition) {
	t.Helper()
	definitions, err := schemacatalog.SystemTableDefinitions()
	if err != nil {
		t.Fatal(err)
	}
	for _, canonical := range definitions {
		if canonical.TableName != tableName {
			continue
		}
		legacy := canonical
		legacy.Columns = nil
		for _, column := range canonical.Columns {
			if (tableName == scheduledRunReservationTable && column.Name == scheduledRunBatchDigestColumn) ||
				(tableName == rollupDeltaTable && column.Name == rollupDeltaScopeDigestColumn) {
				continue
			}
			legacy.Columns = append(legacy.Columns, column)
		}
		legacy.Indices = nil
		for _, index := range canonical.Indices {
			if _, managed := workerQueueRequiredIndexes[tableName][index.Name]; !managed {
				legacy.Indices = append(legacy.Indices, index)
			}
		}
		return canonical, legacy
	}
	t.Fatalf("canonical catalog is missing %s", tableName)
	return schema.TableDefinition{}, schema.TableDefinition{}
}

func scheduledBatchDigestForMigrationTest(parts ...string) string {
	digest := sha256.New()
	var size [8]byte
	for _, part := range append([]string{"scoped-batch"}, parts...) {
		binary.BigEndian.PutUint64(size[:], uint64(len(part)))
		_, _ = digest.Write(size[:])
		_, _ = digest.Write([]byte(part))
	}
	return hex.EncodeToString(digest.Sum(nil))
}