package mysqlbackend

import (
	"context"
	"database/sql"
	"database/sql/driver"
	"errors"
	"fmt"
	"io"
	"strings"
	"sync"
	"testing"
	"time"

	mysql "github.com/go-sql-driver/mysql"
	"go.uber.org/zap"
	"go.uber.org/zap/zaptest/observer"
)

var registerMetadataRetryDriverOnce sync.Once

type metadataRetryDriver struct{}

type metadataRetryConn struct{}

type metadataRetryRows struct {
	done bool
}

type metadataRetryState struct {
	mu      sync.Mutex
	queries int
	errs    []error
	logs    *observer.ObservedLogs
}

var testMetadataRetryState metadataRetryState

func (metadataRetryDriver) Open(string) (driver.Conn, error) {
	return metadataRetryConn{}, nil
}

func (metadataRetryConn) Prepare(string) (driver.Stmt, error) {
	return nil, errors.New("prepare not implemented")
}

func (metadataRetryConn) Close() error {
	return nil
}

func (metadataRetryConn) Begin() (driver.Tx, error) {
	return nil, errors.New("begin not implemented")
}

func (metadataRetryConn) QueryContext(context.Context, string, []driver.NamedValue) (driver.Rows, error) {
	testMetadataRetryState.mu.Lock()
	testMetadataRetryState.queries++
	queries := testMetadataRetryState.queries
	var err error
	if queries <= len(testMetadataRetryState.errs) {
		err = testMetadataRetryState.errs[queries-1]
	}
	testMetadataRetryState.mu.Unlock()

	if err != nil {
		return nil, err
	}
	return &metadataRetryRows{}, nil
}

func (metadataRetryRows) Columns() []string {
	return []string{"v"}
}

func (metadataRetryRows) Close() error {
	return nil
}

func (r *metadataRetryRows) Next(dest []driver.Value) error {
	if r.done {
		return io.EOF
	}
	r.done = true
	dest[0] = int64(1)
	return nil
}

func TestSchemaManagerTTLConfig(t *testing.T) {
	manager := NewSchemaManager(nil, 10*time.Second, 20*time.Second)
	if manager.cache.ttl != 10*time.Second {
		t.Fatalf("expected schema cache ttl 10s, got %s", manager.cache.ttl)
	}
	if manager.metaCache.ttl != 20*time.Second {
		t.Fatalf("expected cql meta cache ttl 20s, got %s", manager.metaCache.ttl)
	}
}

func TestSchemaCacheLockLifecycle(t *testing.T) {
	cache := NewSchemaCache(5 * time.Second)
	key := userKey{user: "u", password: "p"}

	lock1 := cache.schemaLock(key)
	if lock1 == nil {
		t.Fatalf("expected schema lock")
	}
	lock2 := cache.schemaLock(key)
	if lock1 != lock2 {
		t.Fatalf("expected same lock instance for same user")
	}

	cache.InvalidateUser("u", "p")
	cache.refreshMu.Lock()
	_, ok := cache.schemaLocks[key]
	cache.refreshMu.Unlock()
	if ok {
		t.Fatalf("expected schema lock to be cleared after invalidation")
	}
}

func TestCQLMetaCacheLockLifecycle(t *testing.T) {
	cache := NewCQLMetaCache(5 * time.Second)
	key := userKey{user: "u", password: "p"}

	lock1 := cache.tableLock(key)
	if lock1 == nil {
		t.Fatalf("expected table lock")
	}
	lock2 := cache.tableLock(key)
	if lock1 != lock2 {
		t.Fatalf("expected same lock instance for same user")
	}

	cache.InvalidateUser("u", "p")
	cache.refreshMu.Lock()
	_, ok := cache.tableLocks[key]
	cache.refreshMu.Unlock()
	if ok {
		t.Fatalf("expected table lock to be cleared after invalidation")
	}
}

func TestIsMySQLConnectionErrorIncludesBusyBuffer(t *testing.T) {
	tests := []struct {
		name string
		err  error
		want bool
	}{
		{name: "invalid conn", err: mysql.ErrInvalidConn, want: true},
		{name: "busy buffer", err: mysql.ErrBusyBuffer, want: true},
		{name: "wrapped busy buffer", err: fmt.Errorf("metadata query failed: %w", mysql.ErrBusyBuffer), want: true},
		{name: "string busy buffer", err: errors.New("failed to query cql_meta.columns: busy buffer"), want: true},
		{name: "ordinary error", err: errors.New("table does not exist"), want: false},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			if got := IsMySQLConnectionError(tt.err); got != tt.want {
				t.Fatalf("IsMySQLConnectionError() = %v, want %v", got, tt.want)
			}
		})
	}
}

func TestQueryMetadataContextWithRetryUsesFreshConnAfterBusyBuffer(t *testing.T) {
	db, pm := newMetadataRetryPool(t, mysql.ErrBusyBuffer)
	pinnedConn := newMetadataRetryPinnedConn(t, db)

	_, rows, release, err := queryMetadataContextWithRetry(WithConn(context.Background(), pinnedConn), pm, "u", "p", "SELECT 1")
	if err != nil {
		t.Fatalf("queryMetadataContextWithRetry err=%v", err)
	}

	if !rows.Next() {
		t.Fatal("expected retry rows")
	}
	var got int64
	if err := rows.Scan(&got); err != nil {
		t.Fatalf("scan retry row: %v", err)
	}
	if got != 1 {
		t.Fatalf("retry row value = %d, want 1", got)
	}
	releaseRows(rows, release)

	testMetadataRetryState.mu.Lock()
	queries := testMetadataRetryState.queries
	testMetadataRetryState.mu.Unlock()
	if queries != 2 {
		t.Fatalf("query count = %d, want 2", queries)
	}
	busyBufferLogs := observedMetadataRetryLogs().FilterMessage("metadata query hit busy buffer, retrying with fresh connection")
	if n := busyBufferLogs.Len(); n != 1 {
		t.Fatalf("busy buffer retry log count = %d, want 1", n)
	}
	if level := busyBufferLogs.All()[0].Level; level != zap.WarnLevel {
		t.Fatalf("busy buffer retry log level = %s, want %s", level, zap.WarnLevel)
	}

	pinnedRows, err := pinnedConn.QueryContext(context.Background(), "SELECT 1")
	if err != nil {
		t.Fatalf("pinned connection should remain usable after metadata retry: %v", err)
	}
	_ = pinnedRows.Close()
}

func TestQueryMetadataContextWithRetryReturnsJoinedErrorWhenFreshConnFails(t *testing.T) {
	db, pm := newMetadataRetryPool(t, mysql.ErrBusyBuffer, errors.New("retry failed"))
	pinnedConn := newMetadataRetryPinnedConn(t, db)

	_, rows, release, err := queryMetadataContextWithRetry(WithConn(context.Background(), pinnedConn), pm, "u", "p", "SELECT 1")
	releaseRows(rows, release)
	if err == nil {
		t.Fatal("expected error")
	}
	if !errors.Is(err, mysql.ErrBusyBuffer) {
		t.Fatalf("expected joined error to include busy buffer, got %v", err)
	}
	if !strings.Contains(err.Error(), "retry failed") {
		t.Fatalf("expected joined error to include retry failure, got %v", err)
	}

	testMetadataRetryState.mu.Lock()
	queries := testMetadataRetryState.queries
	testMetadataRetryState.mu.Unlock()
	if queries != 2 {
		t.Fatalf("query count = %d, want 2", queries)
	}
}

func TestQueryMetadataContextWithRetryLogsRetryBusyBuffer(t *testing.T) {
	db, pm := newMetadataRetryPool(t, mysql.ErrBusyBuffer, mysql.ErrBusyBuffer)
	pinnedConn := newMetadataRetryPinnedConn(t, db)

	_, rows, release, err := queryMetadataContextWithRetry(WithConn(context.Background(), pinnedConn), pm, "u", "p", "SELECT 1")
	releaseRows(rows, release)
	if err == nil {
		t.Fatal("expected error")
	}
	if !errors.Is(err, mysql.ErrBusyBuffer) {
		t.Fatalf("expected joined error to include busy buffer, got %v", err)
	}

	logs := observedMetadataRetryLogs()
	if n := logs.FilterMessage("metadata query hit busy buffer, retrying with fresh connection").Len(); n != 1 {
		t.Fatalf("initial busy buffer log count = %d, want 1", n)
	}
	retryLogs := logs.FilterMessage("metadata retry query hit busy buffer")
	if n := retryLogs.Len(); n != 1 {
		t.Fatalf("retry busy buffer log count = %d, want 1", n)
	}
	if level := retryLogs.All()[0].Level; level != zap.WarnLevel {
		t.Fatalf("retry busy buffer log level = %s, want %s", level, zap.WarnLevel)
	}
}

func TestQueryMetadataContextWithRetryDoesNotRetryOrdinaryMetadataError(t *testing.T) {
	ordinaryErr := errors.New("table does not exist")
	_, pm := newMetadataRetryPool(t, ordinaryErr)

	_, rows, release, err := queryMetadataContextWithRetry(context.Background(), pm, "u", "p", "SELECT 1")
	releaseRows(rows, release)
	if !errors.Is(err, ordinaryErr) {
		t.Fatalf("expected ordinary metadata error, got %v", err)
	}

	testMetadataRetryState.mu.Lock()
	queries := testMetadataRetryState.queries
	testMetadataRetryState.mu.Unlock()
	if queries != 1 {
		t.Fatalf("query count = %d, want 1", queries)
	}
	logs := observedMetadataRetryLogs()
	if n := logs.FilterMessage("metadata query hit busy buffer, retrying with fresh connection").Len(); n != 0 {
		t.Fatalf("busy buffer retry log count = %d, want 0", n)
	}
	if n := logs.FilterMessage("metadata query failed, retrying with fresh connection").Len(); n != 0 {
		t.Fatalf("generic retry log count = %d, want 0", n)
	}
	nonConnLogs := logs.FilterMessage("metadata query failed with non connection error")
	if n := nonConnLogs.Len(); n != 1 {
		t.Fatalf("non connection error log count = %d, want 1", n)
	}
	if level := nonConnLogs.All()[0].Level; level != zap.ErrorLevel {
		t.Fatalf("non connection error log level = %s, want %s", level, zap.ErrorLevel)
	}
}

func newMetadataRetryPool(t *testing.T, errs ...error) (*sql.DB, *PoolManager) {
	t.Helper()
	registerMetadataRetryDriverOnce.Do(func() {
		sql.Register("metadata-retry-test", metadataRetryDriver{})
	})
	testMetadataRetryState.mu.Lock()
	testMetadataRetryState.queries = 0
	testMetadataRetryState.errs = append([]error(nil), errs...)
	testMetadataRetryState.logs = nil
	testMetadataRetryState.mu.Unlock()

	db, err := sql.Open("metadata-retry-test", "")
	if err != nil {
		t.Fatalf("open fake db: %v", err)
	}
	t.Cleanup(func() { _ = db.Close() })

	pm := &PoolManager{
		logger: newMetadataRetryLogger(),
		pools: map[string]*UserPool{
			"u:p": {username: "u", db: db},
		},
	}
	return db, pm
}

func newMetadataRetryLogger() *zap.Logger {
	core, logs := observer.New(zap.WarnLevel)
	testMetadataRetryState.mu.Lock()
	testMetadataRetryState.logs = logs
	testMetadataRetryState.mu.Unlock()
	return zap.New(core)
}

func observedMetadataRetryLogs() *observer.ObservedLogs {
	testMetadataRetryState.mu.Lock()
	defer testMetadataRetryState.mu.Unlock()
	if testMetadataRetryState.logs == nil {
		_, logs := observer.New(zap.WarnLevel)
		return logs
	}
	return testMetadataRetryState.logs
}

func newMetadataRetryPinnedConn(t *testing.T, db *sql.DB) *sql.Conn {
	t.Helper()
	conn, err := db.Conn(context.Background())
	if err != nil {
		t.Fatalf("get pinned conn: %v", err)
	}
	t.Cleanup(func() { _ = conn.Close() })
	return conn
}