package executor

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

	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/cqlerrors"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/metrics"
)

type fakeExec struct{}

func (fakeExec) ExecContext(context.Context, string, ...interface{}) (sql.Result, error) {
	return nil, nil
}

func (fakeExec) QueryContext(context.Context, string, ...interface{}) (*sql.Rows, error) {
	return nil, nil
}

func deadlockErr() error {
	return cqlerrors.WriteTimeoutRetryable("deadlock found", cqlerrors.RetryDeadlock)
}

func failWait(t *testing.T) backoffWaiter {
	return func(context.Context, time.Duration) error {
		t.Helper()
		t.Fatalf("wait should not be called")
		return nil
	}
}

func TestBackoffCeiling(t *testing.T) {
	base := 10 * time.Millisecond
	max := 100 * time.Millisecond
	cases := []struct {
		attempt int
		want    time.Duration
	}{
		{0, 10 * time.Millisecond},
		{1, 20 * time.Millisecond},
		{2, 40 * time.Millisecond},
		{3, 80 * time.Millisecond},
		{4, 100 * time.Millisecond},
		{10, 100 * time.Millisecond},
		{100, 100 * time.Millisecond},
	}
	for _, tc := range cases {
		if got := backoffCeiling(tc.attempt, base, max); got != tc.want {
			t.Fatalf("backoffCeiling(%d)=%v want %v", tc.attempt, got, tc.want)
		}
	}
}

func TestRetryOnDeadlockSucceedsAfterRetries(t *testing.T) {
	cfg := deadlockRetryConfig{enabled: true, maxAttempts: 3, baseBackoff: 10 * time.Millisecond, maxBackoff: 100 * time.Millisecond}
	success := &QueryResult{RowsAffected: 7}
	calls := 0
	var ceilings []time.Duration
	wait := func(_ context.Context, ceiling time.Duration) error {
		ceilings = append(ceilings, ceiling)
		return nil
	}
	res, attempts, err := retryOnDeadlock(context.Background(), cfg, wait, func() (*QueryResult, error) {
		calls++
		if calls < 3 {
			return nil, deadlockErr()
		}
		return success, nil
	})
	if err != nil {
		t.Fatalf("expected success, got %v", err)
	}
	if res != success {
		t.Fatalf("expected success result, got %v", res)
	}
	if calls != 3 || attempts != 3 {
		t.Fatalf("expected 3 calls/attempts, got calls=%d attempts=%d", calls, attempts)
	}
	if len(ceilings) != 2 || ceilings[0] != 10*time.Millisecond || ceilings[1] != 20*time.Millisecond {
		t.Fatalf("unexpected backoff ceilings: %v", ceilings)
	}
}

func TestRetryOnDeadlockDoesNotRetryNonDeadlock(t *testing.T) {
	cfg := deadlockRetryConfig{enabled: true, maxAttempts: 3, baseBackoff: time.Millisecond, maxBackoff: time.Millisecond}
	sentinel := errors.New("boom")
	calls := 0
	_, attempts, err := retryOnDeadlock(context.Background(), cfg, failWait(t), func() (*QueryResult, error) {
		calls++
		return nil, sentinel
	})
	if !errors.Is(err, sentinel) {
		t.Fatalf("expected sentinel error, got %v", err)
	}
	if calls != 1 || attempts != 1 {
		t.Fatalf("expected single attempt, got calls=%d attempts=%d", calls, attempts)
	}
}

func TestRetryOnDeadlockExhausts(t *testing.T) {
	cfg := deadlockRetryConfig{enabled: true, maxAttempts: 3, baseBackoff: 10 * time.Millisecond, maxBackoff: 100 * time.Millisecond}
	calls := 0
	waits := 0
	wait := func(_ context.Context, _ time.Duration) error {
		waits++
		return nil
	}
	_, attempts, err := retryOnDeadlock(context.Background(), cfg, wait, func() (*QueryResult, error) {
		calls++
		return nil, deadlockErr()
	})
	if cqlerrors.RetryKindOf(err) != cqlerrors.RetryDeadlock {
		t.Fatalf("expected deadlock error after exhaustion, got %v", err)
	}
	if calls != 3 || attempts != 3 {
		t.Fatalf("expected 3 attempts, got calls=%d attempts=%d", calls, attempts)
	}
	if waits != 2 {
		t.Fatalf("expected 2 backoff waits, got %d", waits)
	}
}

func TestRetryOnDeadlockStopsWhenWaitAborts(t *testing.T) {
	cfg := deadlockRetryConfig{enabled: true, maxAttempts: 5, baseBackoff: 10 * time.Millisecond, maxBackoff: 100 * time.Millisecond}
	aborted := errors.New("context canceled")
	calls := 0
	wait := func(_ context.Context, _ time.Duration) error {
		return aborted
	}
	_, attempts, err := retryOnDeadlock(context.Background(), cfg, wait, func() (*QueryResult, error) {
		calls++
		return nil, deadlockErr()
	})
	if cqlerrors.RetryKindOf(err) != cqlerrors.RetryDeadlock {
		t.Fatalf("expected the deadlock error to surface, got %v", err)
	}
	if calls != 1 || attempts != 1 {
		t.Fatalf("expected to stop after first attempt, got calls=%d attempts=%d", calls, attempts)
	}
}

func TestRetryOnDeadlockDisabledRunsOnce(t *testing.T) {
	cfg := deadlockRetryConfig{enabled: false, maxAttempts: 3, baseBackoff: time.Millisecond, maxBackoff: time.Millisecond}
	calls := 0
	_, attempts, err := retryOnDeadlock(context.Background(), cfg, failWait(t), func() (*QueryResult, error) {
		calls++
		return nil, deadlockErr()
	})
	if cqlerrors.RetryKindOf(err) != cqlerrors.RetryDeadlock {
		t.Fatalf("expected deadlock error surfaced once, got %v", err)
	}
	if calls != 1 || attempts != 1 {
		t.Fatalf("expected single attempt when disabled, got calls=%d attempts=%d", calls, attempts)
	}
}

func TestNewDeadlockRetryConfigDefaults(t *testing.T) {
	cfg := newDeadlockRetryConfig(nil, 0, 0, 0)
	if !cfg.enabled {
		t.Fatalf("expected enabled by default when unset")
	}
	if cfg.maxAttempts != 3 || cfg.baseBackoff != 10*time.Millisecond || cfg.maxBackoff != 100*time.Millisecond {
		t.Fatalf("unexpected defaults: %+v", cfg)
	}
}

func TestNewDeadlockRetryConfigExplicitValues(t *testing.T) {
	disabled := false
	cfg := newDeadlockRetryConfig(&disabled, 5, time.Second, 2*time.Second)
	if cfg.enabled {
		t.Fatalf("expected disabled when enabled=false")
	}
	if cfg.maxAttempts != 5 || cfg.baseBackoff != time.Second || cfg.maxBackoff != 2*time.Second {
		t.Fatalf("expected provided values preserved: %+v", cfg)
	}
}

func TestNewDeadlockRetryConfigClampsMaxBelowBase(t *testing.T) {
	cfg := newDeadlockRetryConfig(nil, 2, 50*time.Millisecond, 10*time.Millisecond)
	if cfg.maxBackoff != cfg.baseBackoff {
		t.Fatalf("expected maxBackoff clamped up to baseBackoff, got %+v", cfg)
	}
}

func TestDeadlockRetryOutcome(t *testing.T) {
	cases := []struct {
		name       string
		attempts   int
		err        error
		wantRecord bool
		wantOut    string
	}{
		{"single-attempt-success", 1, nil, false, ""},
		{"single-attempt-failure", 1, deadlockErr(), false, ""},
		{"succeeded-after-retry", 3, nil, true, metrics.WriteRetryOutcomeSucceeded},
		{"exhausted", 3, deadlockErr(), true, metrics.WriteRetryOutcomeExhausted},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			out, record := deadlockRetryOutcome(tc.attempts, tc.err)
			if record != tc.wantRecord {
				t.Fatalf("record=%v want %v", record, tc.wantRecord)
			}
			if record && out != tc.wantOut {
				t.Fatalf("outcome=%q want %q", out, tc.wantOut)
			}
		})
	}
}

func TestExecuteWriteWithRetryRetriesDeadlockForAutocommit(t *testing.T) {
	e := &Executor{deadlockRetry: deadlockRetryConfig{enabled: true, maxAttempts: 3, baseBackoff: time.Nanosecond, maxBackoff: time.Nanosecond}}
	calls := 0
	_, err := e.executeWriteWithRetry(context.Background(), fakeExec{}, func() (*QueryResult, error) {
		calls++
		return nil, deadlockErr()
	})
	if cqlerrors.RetryKindOf(err) != cqlerrors.RetryDeadlock {
		t.Fatalf("expected deadlock error, got %v", err)
	}
	if calls != 3 {
		t.Fatalf("expected 3 attempts for autocommit write, got %d", calls)
	}
}

func TestExecuteWriteWithRetryDoesNotRetryInBatchContext(t *testing.T) {
	e := &Executor{deadlockRetry: deadlockRetryConfig{enabled: true, maxAttempts: 3, baseBackoff: time.Nanosecond, maxBackoff: time.Nanosecond}}
	ctx := markBatchContext(context.Background())
	calls := 0
	_, err := e.executeWriteWithRetry(ctx, fakeExec{}, func() (*QueryResult, error) {
		calls++
		return nil, deadlockErr()
	})
	if cqlerrors.RetryKindOf(err) != cqlerrors.RetryDeadlock {
		t.Fatalf("expected deadlock error, got %v", err)
	}
	if calls != 1 {
		t.Fatalf("expected single attempt in batch context, got %d", calls)
	}
}

func TestExecuteWriteWithRetryDoesNotRetryInsideTransaction(t *testing.T) {
	e := &Executor{deadlockRetry: deadlockRetryConfig{enabled: true, maxAttempts: 3, baseBackoff: time.Nanosecond, maxBackoff: time.Nanosecond}}
	var tx *sql.Tx // typed nil identifies a caller-owned transaction
	calls := 0
	_, err := e.executeWriteWithRetry(context.Background(), tx, func() (*QueryResult, error) {
		calls++
		return nil, deadlockErr()
	})
	if cqlerrors.RetryKindOf(err) != cqlerrors.RetryDeadlock {
		t.Fatalf("expected deadlock error, got %v", err)
	}
	if calls != 1 {
		t.Fatalf("expected single attempt inside tx, got %d", calls)
	}
}

func TestDeadlockBackoffWaitRespectsCanceledContext(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	if err := deadlockBackoffWait(ctx, time.Hour); err == nil {
		t.Fatalf("expected context error, got nil")
	}
}

func TestDeadlockBackoffWaitReturnsAfterShortCeiling(t *testing.T) {
	if err := deadlockBackoffWait(context.Background(), time.Millisecond); err != nil {
		t.Fatalf("expected nil, got %v", err)
	}
}