package executor

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

	"go.uber.org/zap"

	"github.com/everything-on-mysql/cassandra-on-mysql/internal/config"
)

type testSchemaRefresher struct {
	calls   int32
	changed bool
	err     error
}

func (t *testSchemaRefresher) RefreshSchemaVersion(_ context.Context, _, _ string) (string, bool, error) {
	atomic.AddInt32(&t.calls, 1)
	return "schema", t.changed, t.err
}

type testSchemaInvalidator struct {
	calls int32
}

func (t *testSchemaInvalidator) InvalidateAll() {
	atomic.AddInt32(&t.calls, 1)
}

type testPreparedCacheInvalidator struct {
	calls int32
}

func (t *testPreparedCacheInvalidator) ClearPrepared() {
	atomic.AddInt32(&t.calls, 1)
}

func newTestSchemaSync(interval time.Duration, refresher *testSchemaRefresher, invalidator *testSchemaInvalidator, prepared *preparedStore, preparedCache *testPreparedCacheInvalidator) *schemaSync {
	enabled := true
	cfg := config.SchemaSyncConfig{
		Enabled:       &enabled,
		CheckInterval: interval,
	}
	versionCfg := config.SchemaVersionConfig{Strategy: "cql_meta"}
	return newSchemaSync(cfg, versionCfg, invalidator, refresher, prepared, preparedCache, zap.NewNop())
}

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

	sync.MaybeSync(context.Background(), "user", "pass")
	sync.MaybeSync(context.Background(), "user", "pass")

	if got := atomic.LoadInt32(&refresher.calls); got != 1 {
		t.Fatalf("expected 1 refresh call, got %d", got)
	}
}

func TestSchemaSyncInvalidatesOnChange(t *testing.T) {
	refresher := &testSchemaRefresher{changed: true}
	invalidator := &testSchemaInvalidator{}
	preparedCache := &testPreparedCacheInvalidator{}
	prepared := newPreparedStore(zap.NewNop())
	prepared.putIfAbsent(&PreparedStatement{ID: "id", CreatedAt: time.Now()})
	sync := newTestSchemaSync(time.Second, refresher, invalidator, prepared, preparedCache)

	sync.MaybeSync(context.Background(), "user", "pass")

	if got := atomic.LoadInt32(&invalidator.calls); got != 1 {
		t.Fatalf("expected 1 invalidation call, got %d", got)
	}
	if got := len(prepared.stmts); got != 0 {
		t.Fatalf("expected prepared cache cleared, found %d entries", got)
	}
	if got := atomic.LoadInt32(&preparedCache.calls); got != 1 {
		t.Fatalf("expected prepared cache invalidation, got %d", got)
	}
}

func TestSchemaSyncSkipsOnError(t *testing.T) {
	refresher := &testSchemaRefresher{err: errors.New("boom")}
	invalidator := &testSchemaInvalidator{}
	sync := newTestSchemaSync(time.Hour, refresher, invalidator, nil, nil)

	sync.MaybeSync(context.Background(), "user", "pass")
	sync.MaybeSync(context.Background(), "user", "pass")

	if got := atomic.LoadInt32(&refresher.calls); got != 1 {
		t.Fatalf("expected 1 refresh call after error, got %d", got)
	}
	if got := atomic.LoadInt32(&invalidator.calls); got != 0 {
		t.Fatalf("expected no invalidation on error, got %d", got)
	}
}