package executor

import (
	"context"
	"sync/atomic"
	"testing"
	"time"

	"go.uber.org/zap"
)

func TestExecutorBackgroundSchemaSyncLifecycle(t *testing.T) {
	refresher := &testSchemaRefresher{}
	invalidator := &testSchemaInvalidator{}
	sync := newTestSchemaSync(10*time.Millisecond, refresher, invalidator, nil, nil)

	exec := &Executor{
		logger:               zap.NewNop(),
		schemaSync:           sync,
		schemaSyncBackground: true,
	}
	exec.Start()
	defer exec.Close()

	time.Sleep(20 * time.Millisecond)
	if got := atomic.LoadInt32(&refresher.calls); got != 0 {
		exec.Close()
		t.Fatalf("expected no schema sync before credentials, got %d", got)
	}

	exec.SetCredentials("user", "pass")
	waitForCalls(t, &refresher.calls, 1, 200*time.Millisecond)

	exec.Close()
	calls := atomic.LoadInt32(&refresher.calls)
	time.Sleep(30 * time.Millisecond)
	if got := atomic.LoadInt32(&refresher.calls); got != calls {
		t.Fatalf("expected schema sync stopped after Close, got %d -> %d", calls, got)
	}
}

func TestExecutorCQLMetaPrecreateGated(t *testing.T) {
	var calls int32
	exec := &Executor{
		logger:           zap.NewNop(),
		cqlMetaPrecreate: true,
		ensureCQLMeta: func(_ context.Context, _, _ string) error {
			atomic.AddInt32(&calls, 1)
			return nil
		},
	}

	exec.SetCredentials("user", "pass")
	waitForCalls(t, &calls, 1, 200*time.Millisecond)

	exec.SetCredentials("user", "pass")
	time.Sleep(20 * time.Millisecond)
	if got := atomic.LoadInt32(&calls); got != 1 {
		t.Fatalf("expected precreate to run once, got %d calls", got)
	}
}

func waitForCalls(t *testing.T, counter *int32, min int32, timeout time.Duration) {
	t.Helper()
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		if atomic.LoadInt32(counter) >= min {
			return
		}
		time.Sleep(5 * time.Millisecond)
	}
	t.Fatalf("timeout waiting for %d calls (got %d)", min, atomic.LoadInt32(counter))
}