package persistence

import (
	"context"
	"database/sql"
	"errors"
	"sync"
	"testing"
	"time"

	"github.com/nexuscrm/db"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/kernel/pkg/schemacatalog"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
)

func TestHistoryTrackingExactOwnerTwoSessionMatrix(t *testing.T) {
	rawDB, cleanup := openSchemaDDLTiDB(t)
	defer cleanup()
	rawDB.SetMaxOpenConns(8)
	prepareHistoryOwnerCatalog(t, rawDB)

	t.Run("enable_enable_typed_commit_conflict_and_generic_rollback", func(t *testing.T) {
		resetHistoryOwnerField(t, rawDB, false, 7)
		errs := runConcurrentHistoryOwnerAttempts(t, rawDB, []bool{true, true}, []string{"attempt-a", "attempt-b"})
		t.Logf("enable/enable exact owner outcomes: %v", errs)
		winner := assertOneHistoryOwnerWinnerOneConflict(t, errs)
		assertHistoryOwnerFinalState(t, rawDB, true, 8, 1, 1)
		assertHistoryOwnerHelpText(t, rawDB, []string{"attempt-a", "attempt-b"}[winner])
		startedAt := historyOwnerOpenStartedAt(t, rawDB)

		if err := runHistoryOwnerAttempt(rawDB, true, []string{"attempt-a", "attempt-b"}[winner]); err != nil {
			t.Fatalf("fresh owner retry: %v", err)
		}
		assertHistoryOwnerFinalState(t, rawDB, true, 8, 1, 1)
		if got := historyOwnerOpenStartedAt(t, rawDB); !got.Equal(startedAt) {
			t.Fatalf("fresh no-op changed open boundary: before=%s after=%s", startedAt, got)
		}
	})

	t.Run("pack_batch_failure_rolls_back_transition_and_metadata", func(t *testing.T) {
		resetHistoryOwnerField(t, rawDB, false, 7)
		rollbackErr := errors.New("later pack batch failure")
		tm := NewTransactionManager(database.NewConnectionFromDB(rawDB))
		repo := NewSchemaRepository(rawDB, nil)
		err := tm.WithTransaction(context.Background(), func(txCtx context.Context) error {
			field := historyOwnerFieldCandidate(true, "must roll back")
			if err := repo.BatchSaveFieldMetadataWithCustomizationPolicy(txCtx, []schema.FieldWithContext{field}, tm.ExtractTx(txCtx), false); err != nil {
				return err
			}
			return rollbackErr
		})
		if !errors.Is(err, rollbackErr) {
			t.Fatalf("pack batch rollback error = %v", err)
		}
		assertHistoryOwnerFinalState(t, rawDB, false, 7, 0, 0)
		var sequences int
		if err := rawDB.QueryRow(`SELECT COUNT(*) FROM _System_FieldHistorySequence WHERE FieldId = 'fld-history'`).Scan(&sequences); err != nil || sequences != 0 {
			t.Fatalf("pack rollback sequences = %d, err = %v", sequences, err)
		}
		var helpText sql.NullString
		if err := rawDB.QueryRow(`SELECT HelpText FROM _System_Field WHERE Id = 'fld-history'`).Scan(&helpText); err != nil || helpText.Valid {
			t.Fatalf("pack rollback HelpText = %#v, err = %v", helpText, err)
		}
	})

	t.Run("optimistic_noop_claim_conflicts_after_concurrent_transition", func(t *testing.T) {
		resetHistoryOwnerField(t, rawDB, true, 7)
		ctx := context.Background()
		conn, err := rawDB.Conn(ctx)
		if err != nil {
			t.Fatalf("open optimistic owner connection: %v", err)
		}
		defer conn.Close()
		if _, err := conn.ExecContext(ctx, "SET SESSION tidb_txn_mode = 'optimistic'"); err != nil {
			t.Fatalf("set optimistic owner transaction mode: %v", err)
		}
		defer func() {
			_, _ = conn.ExecContext(ctx, "SET SESSION tidb_txn_mode = DEFAULT")
		}()

		tx, err := conn.BeginTx(ctx, nil)
		if err != nil {
			t.Fatalf("begin optimistic owner transaction: %v", err)
		}
		defer tx.Rollback()
		states, err := loadHistoryTrackingStates(ctx, tx, []string{"fld-history"})
		if err != nil {
			t.Fatalf("prefetch optimistic no-op state: %v", err)
		}
		state := states["fld-history"]
		if err := runHistoryOwnerAttempt(rawDB, false, "concurrent-disable"); err != nil {
			t.Fatalf("concurrent disable: %v", err)
		}

		err = claimHistoryTrackingNoOps(ctx, tx, []historyTrackingNoOpClaim{{
			fieldID: state.fieldID, tracked: state.trackHistory,
			revision: state.revision, customized: state.customized,
		}})
		if err == nil {
			err = tx.Commit()
		}
		if !errors.Is(err, ports.ErrHistoryTrackingConflict) && !isOptimisticWriteConflict(err) {
			t.Fatalf("stale optimistic no-op claim error = %v, want history/write conflict", err)
		}
		assertHistoryOwnerFinalState(t, rawDB, false, 8, 1, 0)
	})

	tests := []struct {
		name       string
		initial    bool
		desired    []bool
		outcomes   []historyOwnerOutcome
		noOpBounds bool
	}{
		{
			name: "enable_disable", initial: false, desired: []bool{true, false},
			outcomes: []historyOwnerOutcome{{tracked: true, revision: 8, intervals: 1, open: 1}, {tracked: false, revision: 9, intervals: 1, open: 0}},
		},
		{
			name: "disable_reenable", initial: true, desired: []bool{false, true},
			outcomes: []historyOwnerOutcome{{tracked: false, revision: 8, intervals: 1, open: 0}, {tracked: true, revision: 9, intervals: 2, open: 1}},
		},
		{
			name: "reapply_reapply", initial: true, desired: []bool{true, true},
			outcomes: []historyOwnerOutcome{{tracked: true, revision: 7, intervals: 1, open: 1}}, noOpBounds: true,
		},
		{
			name: "reapply_toggle", initial: true, desired: []bool{true, false},
			outcomes: []historyOwnerOutcome{{tracked: false, revision: 8, intervals: 1, open: 0}, {tracked: true, revision: 9, intervals: 2, open: 1}},
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			resetHistoryOwnerField(t, rawDB, test.initial, 7)
			var startedAt time.Time
			if test.noOpBounds {
				startedAt = historyOwnerOpenStartedAt(t, rawDB)
			}
			errs := runConcurrentHistoryOwnerAttempts(t, rawDB, test.desired, []string{"attempt-a", "attempt-b"})
			for _, err := range errs {
				if err != nil && !errors.Is(err, ports.ErrHistoryTrackingConflict) && !ports.IsHistorySerializationConflict(err) {
					t.Fatalf("owner attempt error = %v", err)
				}
			}
			assertHistoryOwnerLegalOutcome(t, rawDB, test.outcomes)
			if test.noOpBounds {
				if got := historyOwnerOpenStartedAt(t, rawDB); !got.Equal(startedAt) {
					t.Fatalf("identical reapply changed open boundary: before=%s after=%s", startedAt, got)
				}
			}
		})
	}
}

func prepareHistoryOwnerCatalog(t *testing.T, rawDB *sql.DB) {
	t.Helper()
	definitions, err := schemacatalog.SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}
	wanted := []string{constants.TableObject, constants.TableField, constants.TableFieldHistorySequence, constants.TableFieldHistoryTrackingInterval}
	repo := NewSchemaRepository(rawDB, nil)
	for _, tableName := range wanted {
		var definition schema.TableDefinition
		for _, candidate := range definitions {
			if candidate.TableName == tableName {
				definition = candidate
				break
			}
		}
		if definition.TableName == "" {
			t.Fatalf("missing canonical table %s", tableName)
		}
		if err := repo.CreatePhysicalTable(context.Background(), definition); err != nil {
			t.Fatalf("create canonical table %s: %v", tableName, err)
		}
	}
	if _, err := rawDB.Exec(`INSERT INTO _System_Object (Id, ApiName, Label) VALUES ('obj-account', 'Account', 'Account')`); err != nil {
		t.Fatalf("insert owner object: %v", err)
	}
	if _, err := rawDB.Exec(`CREATE TABLE Account (Name VARCHAR(255) NULL)`); err != nil {
		t.Fatalf("create owner physical table: %v", err)
	}
}

func runConcurrentHistoryOwnerAttempts(t *testing.T, rawDB *sql.DB, desired []bool, helpTexts []string) []error {
	t.Helper()
	errs := make([]error, len(desired))
	start := make(chan struct{})
	var wg sync.WaitGroup
	for i := range desired {
		wg.Add(1)
		go func(index int) {
			defer wg.Done()
			<-start
			errs[index] = runHistoryOwnerAttempt(rawDB, desired[index], helpTexts[index])
		}(i)
	}
	close(start)
	wg.Wait()
	return errs
}

func runHistoryOwnerAttempt(rawDB *sql.DB, desired bool, helpText string) error {
	tm := NewTransactionManager(database.NewConnectionFromDB(rawDB))
	repo := NewSchemaRepository(rawDB, nil)
	return tm.WithTransaction(context.Background(), func(txCtx context.Context) error {
		tx := tm.ExtractTx(txCtx)
		if tx == nil {
			return errors.New("missing exact owner transaction executor")
		}
		field := historyOwnerFieldCandidate(desired, helpText)
		if err := repo.BatchSaveFieldMetadataWithCustomizationPolicy(txCtx, []schema.FieldWithContext{field}, tx, false); err != nil {
			return err
		}
		return nil
	})
}

func historyOwnerFieldCandidate(desired bool, helpText string) schema.FieldWithContext {
	return schema.FieldWithContext{
		TableName:        "Account",
		ObjectID:         "obj-account",
		FieldID:          "fld-history",
		OwnerKind:        constants.FieldOwnerKindPack,
		OwnerPack:        "history-pack",
		ColumnDefinition: schema.ColumnDefinition{Name: "Name", Type: "VARCHAR(255)", Nullable: true},
		Field: &models.FieldMetadata{
			ID: "fld-history", APIName: "Name", Label: "Name", Type: constants.FieldTypeString,
			HelpText: &helpText, TrackHistory: desired,
		},
	}
}

func resetHistoryOwnerField(t *testing.T, rawDB *sql.DB, tracked bool, revision int64) {
	t.Helper()
	if _, err := rawDB.Exec("DELETE FROM _System_FieldHistoryTrackingInterval"); err != nil {
		t.Fatalf("reset owner intervals: %v", err)
	}
	if _, err := rawDB.Exec("DELETE FROM _System_FieldHistorySequence"); err != nil {
		t.Fatalf("reset owner sequences: %v", err)
	}
	if _, err := rawDB.Exec("DELETE FROM _System_Field"); err != nil {
		t.Fatalf("reset owner field: %v", err)
	}
	if _, err := rawDB.Exec(`INSERT INTO _System_Field (Id, ObjectId, ApiName, Label, Type, TrackHistory, HistoryTrackingRevision, OwnerKind, OwnerPack, IsCustomized, IsDeleted) VALUES ('fld-history', 'obj-account', 'Name', 'Name', 'string', ?, ?, 'pack', 'history-pack', 0, 0)`, tracked, revision); err != nil {
		t.Fatalf("insert owner field: %v", err)
	}
	if tracked {
		if _, err := rawDB.Exec(`INSERT INTO _System_FieldHistorySequence (FieldId, CurrentSequence) VALUES ('fld-history', 1)`); err != nil {
			t.Fatalf("insert owner sequence: %v", err)
		}
		if _, err := rawDB.Exec(`INSERT INTO _System_FieldHistoryTrackingInterval (Id, FieldId, OpenFieldId, StartedSequence, StartedAt) VALUES ('interval-initial', 'fld-history', 'fld-history', 1, UTC_TIMESTAMP(6) - INTERVAL 1 HOUR)`); err != nil {
			t.Fatalf("insert owner interval: %v", err)
		}
	}
}

func assertOneHistoryOwnerWinnerOneConflict(t *testing.T, errs []error) int {
	t.Helper()
	winner := -1
	conflicts := 0
	for index, err := range errs {
		switch {
		case err == nil:
			if winner != -1 {
				t.Fatalf("multiple exact owner winners: errors=%v", errs)
			}
			winner = index
		case errors.Is(err, ports.ErrHistoryTrackingConflict), ports.IsHistorySerializationConflict(err):
			conflicts++
		default:
			t.Fatalf("unexpected exact owner error: %v", err)
		}
	}
	if winner == -1 || conflicts != 1 {
		t.Fatalf("exact owner outcomes winner=%d conflicts=%d errors=%v", winner, conflicts, errs)
	}
	return winner
}

type historyOwnerOutcome struct {
	tracked   bool
	revision  int64
	intervals int
	open      int
}

func assertHistoryOwnerFinalState(t *testing.T, rawDB *sql.DB, tracked bool, revision int64, intervals, open int) {
	t.Helper()
	assertHistoryTrackingFinalState(t, rawDB, tracked, revision, intervals, open)
}

func assertHistoryOwnerLegalOutcome(t *testing.T, rawDB *sql.DB, outcomes []historyOwnerOutcome) {
	t.Helper()
	var got historyOwnerOutcome
	if err := rawDB.QueryRow(`SELECT TrackHistory, HistoryTrackingRevision FROM _System_Field WHERE Id = 'fld-history'`).Scan(&got.tracked, &got.revision); err != nil {
		t.Fatalf("read exact owner field state: %v", err)
	}
	var overlaps int
	if err := rawDB.QueryRow(`SELECT COUNT(*), COALESCE(SUM(CASE WHEN EndedSequence IS NULL AND OpenFieldId = FieldId THEN 1 ELSE 0 END), 0) FROM _System_FieldHistoryTrackingInterval WHERE FieldId = 'fld-history'`).Scan(&got.intervals, &got.open); err != nil {
		t.Fatalf("read exact owner interval state: %v", err)
	}
	if err := rawDB.QueryRow(`SELECT COUNT(*) FROM _System_FieldHistoryTrackingInterval a JOIN _System_FieldHistoryTrackingInterval b ON a.FieldId = b.FieldId AND a.Id < b.Id AND a.StartedSequence < COALESCE(b.EndedSequence, 9223372036854775807) AND b.StartedSequence < COALESCE(a.EndedSequence, 9223372036854775807) WHERE a.FieldId = 'fld-history'`).Scan(&overlaps); err != nil {
		t.Fatalf("read exact owner overlaps: %v", err)
	}
	if overlaps != 0 {
		t.Fatalf("exact owner intervals overlap: %d", overlaps)
	}
	for _, want := range outcomes {
		if got == want {
			return
		}
	}
	t.Fatalf("exact owner outcome = %+v, want one of %+v", got, outcomes)
}

func assertHistoryOwnerHelpText(t *testing.T, rawDB *sql.DB, want string) {
	t.Helper()
	var got sql.NullString
	if err := rawDB.QueryRow(`SELECT HelpText FROM _System_Field WHERE Id = 'fld-history'`).Scan(&got); err != nil {
		t.Fatalf("read owner HelpText: %v", err)
	}
	if !got.Valid || got.String != want {
		t.Fatalf("owner HelpText = %#v, want committed winner %q", got, want)
	}
}

func historyOwnerOpenStartedAt(t *testing.T, rawDB *sql.DB) time.Time {
	t.Helper()
	var startedAt time.Time
	if err := rawDB.QueryRow(`SELECT StartedAt FROM _System_FieldHistoryTrackingInterval WHERE FieldId = 'fld-history' AND EndedSequence IS NULL`).Scan(&startedAt); err != nil {
		t.Fatalf("read owner open boundary: %v", err)
	}
	return startedAt
}