package executor

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

	"go.uber.org/zap"

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

const defaultSchemaSyncInterval = 5 * time.Second

type schemaSync struct {
	enabled        bool
	intervalNanos  int64
	lastCheckNanos int64
	inFlight       int32
	refresher      schemaVersionRefresher
	invalidator    schemaCacheInvalidator
	prepared       *preparedStore
	preparedCache  preparedCacheInvalidator
	logger         *zap.Logger
}

type schemaVersionRefresher interface {
	RefreshSchemaVersion(ctx context.Context, user, password string) (string, bool, error)
}

type schemaCacheInvalidator interface {
	InvalidateAll()
}

type preparedCacheInvalidator interface {
	ClearPrepared()
}

func newSchemaSync(cfg config.SchemaSyncConfig, versionCfg config.SchemaVersionConfig, invalidator schemaCacheInvalidator, refresher schemaVersionRefresher, prepared *preparedStore, preparedCache preparedCacheInvalidator, logger *zap.Logger) *schemaSync {
	if invalidator == nil || refresher == nil {
		return nil
	}
	if !cfg.EnabledValue() {
		return nil
	}
	strategy := strings.ToLower(strings.TrimSpace(versionCfg.Strategy))
	if strategy == "" {
		strategy = "cql_meta"
	}
	if strategy == "fixed" {
		return nil
	}

	interval := cfg.CheckInterval
	if interval <= 0 {
		interval = defaultSchemaSyncInterval
	}
	if logger == nil {
		logger = zap.NewNop()
	}

	return &schemaSync{
		enabled:       true,
		intervalNanos: interval.Nanoseconds(),
		refresher:     refresher,
		invalidator:   invalidator,
		prepared:      prepared,
		preparedCache: preparedCache,
		logger:        logger,
	}
}

func (s *schemaSync) MaybeSync(ctx context.Context, user, password string) {
	if s == nil || !s.enabled {
		return
	}
	now := time.Now().UnixNano()
	last := atomic.LoadInt64(&s.lastCheckNanos)
	if last != 0 && now-last < s.intervalNanos {
		return
	}
	if !atomic.CompareAndSwapInt32(&s.inFlight, 0, 1) {
		return
	}
	defer atomic.StoreInt32(&s.inFlight, 0)
	defer atomic.StoreInt64(&s.lastCheckNanos, now)

	changed, err := s.checkNow(ctx, user, password)
	if err != nil {
		s.logger.Warn("schema sync check failed", zap.Error(err))
		return
	}
	if changed {
		s.invalidator.InvalidateAll()
		if s.prepared != nil {
			s.prepared.Clear()
		}
		if s.preparedCache != nil {
			s.preparedCache.ClearPrepared()
		}
	}
}

func (s *schemaSync) checkNow(ctx context.Context, user, password string) (bool, error) {
	if s == nil || s.refresher == nil {
		return false, nil
	}
	start := time.Now()
	_, changed, err := s.refresher.RefreshSchemaVersion(ctx, user, password)
	metrics.ObserveStageFromContext(ctx, metrics.StageSchemaSync, time.Since(start))
	if err != nil {
		return false, err
	}
	return changed, nil
}

func (s *schemaSync) interval() time.Duration {
	if s == nil {
		return 0
	}
	nanos := atomic.LoadInt64(&s.intervalNanos)
	if nanos <= 0 {
		return defaultSchemaSyncInterval
	}
	return time.Duration(nanos)
}

func (e *Executor) maybeSyncSchema(ctx context.Context, user, password string) {
	if e == nil || e.schemaSync == nil {
		return
	}
	if e.schemaSyncBackground {
		return
	}
	e.schemaSync.MaybeSync(ctx, user, password)
}