package clustersearch

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

	"github.com/pingcap/log"
	"go.uber.org/zap"

	"github.com/pingcap-inc/clinic/pkg/db"
)

const (
	largeOrgClusterThreshold = int64(50000)
	refreshInterval          = 24 * time.Hour
)

var ErrTextSearchPolicyUnavailable = errors.New("cluster text search policy unavailable; please search by cluster ID")

type cacheSnapshot struct {
	initialized bool
	loaded      bool
	largeOrgIDs map[uint]struct{}
	refreshedAt time.Time
}

var (
	snapshot atomic.Value
	startMu  sync.Mutex
	started  bool
)

func init() {
	snapshot.Store(cacheSnapshot{
		initialized: false,
		loaded:      true,
		largeOrgIDs: map[uint]struct{}{},
	})
}

// StartLargeOrgTextSearchCache loads the large-org snapshot once at startup and refreshes it daily.
func StartLargeOrgTextSearchCache(ctx context.Context) {
	startMu.Lock()
	if started {
		startMu.Unlock()
		return
	}
	started = true
	startMu.Unlock()

	if err := RefreshLargeOrgTextSearchCache(ctx); err != nil {
		log.Error("refresh large-org text-search cache failed at startup", zap.Error(err))
		failClosedIfUninitialized()
	}

	go func() {
		ticker := time.NewTicker(refreshInterval)
		defer ticker.Stop()

		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				if err := RefreshLargeOrgTextSearchCache(ctx); err != nil {
					log.Error("refresh large-org text-search cache failed", zap.Error(err))
				}
			}
		}
	}()
}

// RefreshLargeOrgTextSearchCache rebuilds the snapshot from DB and swaps it in atomically.
func RefreshLargeOrgTextSearchCache(ctx context.Context) error {
	if db.GetDB() == nil {
		return nil
	}
	queryCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()

	type row struct {
		OrgID uint `gorm:"column:org_id"`
	}

	var rows []row
	err := db.GetDB().WithContext(queryCtx).
		Unscoped().
		Table("clusters").
		Select("org_id").
		Group("org_id").
		Having("COUNT(*) >= ?", largeOrgClusterThreshold).
		Scan(&rows).Error
	if err != nil {
		return err
	}

	largeOrgIDs := make(map[uint]struct{}, len(rows))
	for _, r := range rows {
		largeOrgIDs[r.OrgID] = struct{}{}
	}
	snapshot.Store(cacheSnapshot{
		initialized: true,
		loaded:      true,
		largeOrgIDs: largeOrgIDs,
		refreshedAt: time.Now(),
	})
	log.Info("refreshed large-org text-search cache",
		zap.Int("large_org_count", len(largeOrgIDs)),
		zap.Int64("cluster_threshold", largeOrgClusterThreshold),
	)
	return nil
}

// AllowTextSearch reports whether LIKE-based text search is allowed for an org.
func AllowTextSearch(orgID uint) bool {
	s := currentSnapshot()
	if !s.initialized {
		return true
	}
	if !s.loaded {
		return false
	}
	_, ok := s.largeOrgIDs[orgID]
	return !ok
}

func TextSearchPolicyAvailable() bool {
	s := currentSnapshot()
	return !s.initialized || s.loaded
}

func RequirePolicyForQuery(query string) error {
	query = strings.TrimSpace(query)
	if query == "" || IsLikelyIDSearch(query) || TextSearchPolicyAvailable() {
		return nil
	}
	return ErrTextSearchPolicyUnavailable
}

func IsLikelyIDSearch(query string) bool {
	id, err := strconv.ParseUint(strings.TrimSpace(query), 10, 64)
	return err == nil && id > 0
}

// DisabledOrgIDsForTextSearch returns the org IDs that must be excluded from
// text-search predicates. The bool is false when text search should fail closed.
func DisabledOrgIDsForTextSearch() ([]uint, bool) {
	s := currentSnapshot()
	if !s.initialized {
		return nil, true
	}
	if !s.loaded {
		return nil, false
	}
	orgIDs := make([]uint, 0, len(s.largeOrgIDs))
	for orgID := range s.largeOrgIDs {
		orgIDs = append(orgIDs, orgID)
	}
	return orgIDs, true
}

func failClosedIfUninitialized() {
	s := currentSnapshot()
	if s.initialized {
		return
	}
	snapshot.Store(cacheSnapshot{
		initialized: true,
		loaded:      false,
		largeOrgIDs: map[uint]struct{}{},
		refreshedAt: time.Now(),
	})
}

func currentSnapshot() cacheSnapshot {
	return snapshot.Load().(cacheSnapshot)
}

// SetSnapshotForTest replaces the cache snapshot and returns a cleanup function.
func SetSnapshotForTest(initialized, loaded bool, largeOrgIDs []uint) func() {
	old := currentSnapshot()
	next := cacheSnapshot{
		initialized: initialized,
		loaded:      loaded,
		largeOrgIDs: make(map[uint]struct{}, len(largeOrgIDs)),
		refreshedAt: time.Now(),
	}
	for _, orgID := range largeOrgIDs {
		next.largeOrgIDs[orgID] = struct{}{}
	}
	snapshot.Store(next)
	return func() {
		snapshot.Store(old)
	}
}