package storage

import (
	"context"
	"errors"
	"fmt"
	"maps"
	"slices"
	"sort"
	"time"

	"github.com/hashicorp/go-memdb"
	"github.com/pingcap/log"
	metametrics "github.com/tidbcloud/aws-shared-provider/pkg/metaserver/metrics"
	"go.uber.org/zap"
	"gorm.io/gorm"

	"github.com/tidbcloud/aws-shared-provider/pkg/infra"
	serverlessinfra "github.com/tidbcloud/tidb-management-service/api/serverless_infra"
	inframysql "github.com/tidbcloud/tidb-management-service/api/serverless_infra/client/mysql"
)

const (
	clusterEventPollInterval         = 3 * time.Second
	clusterEventPollBatchSize        = 500
	clusterRetainedEventScanInterval = min(120*time.Second, serverlessinfra.DefaultClusterEventTTL/4)
	clusterLoadBatchSize             = 500
	clusterLookupBatchSize           = 500
)

const latestClusterEventTableWithIndexHint = "cluster_events FORCE INDEX (`" +
	inframysql.ClusterEventACKIndex + "`)"

var errEventSafetyScanDeadlineExceeded = errors.New("cluster event safety scan deadline exceeded")

func clusterMetaTableSchema() *memdb.DBSchema {
	return &memdb.DBSchema{
		Tables: map[string]*memdb.TableSchema{
			clusterMetaTableName: {
				Name: clusterMetaTableName,
				Indexes: map[string]*memdb.IndexSchema{
					"id": {
						Name:    "id",
						Unique:  true,
						Indexer: &memdb.StringFieldIndex{Field: "ClusterID"},
					},
					string(IndexClusterID): {
						Name:    string(IndexClusterID),
						Unique:  true,
						Indexer: &memdb.StringFieldIndex{Field: "ClusterID"},
					},
					string(IndexTenantID): {
						Name:         string(IndexTenantID),
						Unique:       false,
						AllowMissing: true,
						Indexer:      &memdb.StringFieldIndex{Field: "Tenant"},
					},
					string(IndexProjectID): {
						Name:         string(IndexProjectID),
						Unique:       false,
						AllowMissing: true,
						Indexer:      &memdb.StringFieldIndex{Field: "ProjectID"},
					},
					string(IndexKeyspaceToken): {
						Name:         string(IndexKeyspaceToken),
						Unique:       false,
						AllowMissing: true,
						Indexer:      &memdb.StringFieldIndex{Field: "KeyspaceToken"},
					},
					string(IndexKeyspaceID): {
						Name:         string(IndexKeyspaceID),
						Unique:       false,
						AllowMissing: true,
						Indexer:      &memdb.UintFieldIndex{Field: "KeyspaceID"},
					},
					string(IndexBillingPlan): {
						Name:         string(IndexBillingPlan),
						Unique:       false,
						AllowMissing: true,
						Indexer:      clusterBillingPlanIndexer{},
					},
					string(IndexZoneRedundantEnabled): {
						Name:         string(IndexZoneRedundantEnabled),
						Unique:       false,
						AllowMissing: true,
						Indexer:      zoneRedundantEnabledIndexer{},
					},
				},
			},
		},
	}
}

func (m *memDBStorage) LookupCluster(index index, key string) ([]*inframysql.Cluster, error) {
	if m.clusterMemDB == nil {
		return nil, fmt.Errorf("memdb not initialized for cluster meta")
	}
	txn := m.clusterMemDB.Txn(false)
	it, err := txn.Get(clusterMetaTableName, string(index), key)
	if err != nil {
		return nil, err
	}
	result := make([]*inframysql.Cluster, 0)
	for obj := it.Next(); obj != nil; obj = it.Next() {
		p := obj.(*inframysql.Cluster)
		result = append(result, p)
	}
	return result, nil
}

func (m *memDBStorage) GetClusterEventWindow() (ClusterEventWindow, error) {
	if m.mysqlDB == nil {
		return ClusterEventWindow{}, fmt.Errorf("mysql not initialized for cluster events")
	}

	snapshot := m.clusterEventWindow.Load()
	if snapshot == nil || snapshot.Epoch == "" {
		return ClusterEventWindow{}, fmt.Errorf("cluster event epoch is not initialized")
	}

	window := *snapshot
	window.Events = slices.Clone(snapshot.Events)
	return window, nil
}

func (m *memDBStorage) publishClusterEventWindow(events []ClusterEvent) {
	if events == nil {
		events = []ClusterEvent{}
	}
	m.clusterEventWindow.Store(&ClusterEventWindow{
		Epoch:            m.clusterEventEpoch,
		HighWatermark:    m.lastClusterEventID,
		RewindLowerBound: m.clusterEventRewindLowerBound,
		Events:           events,
	})
}

func (m *memDBStorage) queryAllClusters(tx *gorm.DB) (clusters []*inframysql.Cluster, err error) {
	startedAt := time.Now()
	defer func() {
		result := metametrics.ResultSuccess
		if err != nil {
			result = metametrics.ResultFailure
		}
		metametrics.ClusterFullLoadDuration.WithLabelValues(result).
			Observe(time.Since(startedAt).Seconds())
	}()

	var batch []*inframysql.Cluster
	query := tx.Preload("Spec").Preload("Status").
		Where("kubernetes = ?", m.kubeClusterName)
	if err = query.FindInBatches(&batch, clusterLoadBatchSize, func(_ *gorm.DB, _ int) error {
		clusters = append(clusters, batch...)
		return nil
	}).Error; err != nil {
		return nil, fmt.Errorf("failed to fetch clusters from mysql: %w", err)
	}
	return clusters, nil
}

func (m *memDBStorage) queryClustersByIDs(tx *gorm.DB, clusterIDs []string) ([]*inframysql.Cluster, error) {
	if len(clusterIDs) == 0 {
		return nil, nil
	}

	clusters := make([]*inframysql.Cluster, 0, len(clusterIDs))
	for start := 0; start < len(clusterIDs); start += clusterLookupBatchSize {
		end := min(start+clusterLookupBatchSize, len(clusterIDs))
		var batch []*inframysql.Cluster
		if err := tx.Preload("Spec").Preload("Status").
			Where("cluster_id IN ?", clusterIDs[start:end]).
			Where("kubernetes = ?", m.kubeClusterName).
			Find(&batch).Error; err != nil {
			return nil, fmt.Errorf("failed to fetch changed clusters from mysql: %w", err)
		}
		clusters = append(clusters, batch...)
	}
	return clusters, nil
}

func clusterEventPollQuery(tx *gorm.DB, kubeClusterName string, afterEventID uint64) *gorm.DB {
	return tx.Model(&inframysql.ClusterEvent{}).
		Select("event_id, name, event_type, resource_version").
		Where("kubernetes = ?", kubeClusterName).
		Where("event_id > ?", afterEventID).
		Order("event_id ASC").
		Limit(clusterEventPollBatchSize)
}

func latestClusterEventPageQuery(tx *gorm.DB, kubeClusterName, afterClusterName string) *gorm.DB {
	eventSource := tx.Model(&inframysql.ClusterEvent{})
	// SQLite backs unit tests but does not support MySQL index hints.
	if tx.Dialector.Name() == "mysql" {
		eventSource = tx.Table(latestClusterEventTableWithIndexHint)
	}

	latestEventIDs := eventSource.
		Select("name, MAX(event_id) AS event_id").
		Where("kubernetes = ?", kubeClusterName).
		Where("name <> ''")
	if afterClusterName != "" {
		latestEventIDs = latestEventIDs.Where("name > ?", afterClusterName)
	}
	latestEventIDs = latestEventIDs.
		Group("name").
		Order("name ASC").
		Limit(clusterEventPollBatchSize)

	return tx.Table("cluster_events AS cluster_event").
		Select("cluster_event.event_id, cluster_event.name, cluster_event.event_type, cluster_event.resource_version").
		Joins("JOIN (?) AS latest ON cluster_event.event_id = latest.event_id", latestEventIDs).
		Order("latest.name ASC")
}

func retainLatestClusterEvent(events map[string]ClusterEvent, event ClusterEvent) {
	if event.Name == "" {
		return
	}

	current, found := events[event.Name]
	if !found || event.EventID > current.EventID {
		events[event.Name] = event
	}
}

func retainLatestClusterEvents(events map[string]ClusterEvent, observed []ClusterEvent) {
	for _, event := range observed {
		retainLatestClusterEvent(events, event)
	}
}

func (m *memDBStorage) pollClusterEvents(
	tx *gorm.DB,
	lowerExclusive uint64,
) ([]ClusterEvent, error) {
	latestEvents := make(map[string]ClusterEvent)
	cursor := lowerExclusive
	for {
		var batch []ClusterEvent
		if err := clusterEventPollQuery(tx, m.kubeClusterName, cursor).Find(&batch).Error; err != nil {
			return nil, fmt.Errorf("failed to poll cluster events: %w", err)
		}

		for _, event := range batch {
			cursor = event.EventID
			retainLatestClusterEvent(latestEvents, event)
		}
		if len(batch) < clusterEventPollBatchSize {
			break
		}
	}
	return sortedClusterEvents(latestEvents), nil
}

func (m *memDBStorage) pollRetainedClusterEvents(tx *gorm.DB) (events []ClusterEvent, err error) {
	startedAt := time.Now()
	returnedRows := 0
	pages := 0
	defer func() {
		result := metametrics.ResultSuccess
		if err != nil {
			result = metametrics.ResultFailure
		}
		metametrics.ClusterEventFullScanDuration.WithLabelValues(result).
			Observe(time.Since(startedAt).Seconds())
		metametrics.ClusterEventFullScanReturnedRows.WithLabelValues(result).
			Observe(float64(returnedRows))
		metametrics.ClusterEventFullScanPages.WithLabelValues(result).
			Observe(float64(pages))
	}()

	events = make([]ClusterEvent, 0)
	cursor := ""
	for {
		var batch []ClusterEvent
		if err = latestClusterEventPageQuery(tx, m.kubeClusterName, cursor).Find(&batch).Error; err != nil {
			return nil, fmt.Errorf("failed to poll latest cluster events: %w", err)
		}
		pages++
		returnedRows += len(batch)

		events = append(events, batch...)
		if len(batch) < clusterEventPollBatchSize {
			break
		}
		cursor = batch[len(batch)-1].Name
	}
	sort.Slice(events, func(i, j int) bool {
		return events[i].EventID < events[j].EventID
	})
	return events, nil
}

func latestClusterEventID(previous uint64, events []ClusterEvent) uint64 {
	lastEventID := previous
	for _, event := range events {
		lastEventID = max(lastEventID, event.EventID)
	}
	return lastEventID
}

func advanceClusterEventRewindLowerBound(
	current uint64,
	previous map[string]ClusterEvent,
	observed map[string]ClusterEvent,
) uint64 {
	lowerBound := current
	for name, previousEvent := range previous {
		observedEvent, found := observed[name]
		if found && observedEvent.EventID >= previousEvent.EventID {
			continue
		}
		lowerBound = max(lowerBound, previousEvent.EventID)
	}
	return lowerBound
}

func sortedClusterEvents(events map[string]ClusterEvent) []ClusterEvent {
	retained := make([]ClusterEvent, 0, len(events))
	for _, event := range events {
		retained = append(retained, event)
	}
	sort.Slice(retained, func(i, j int) bool {
		return retained[i].EventID < retained[j].EventID
	})
	return retained
}

func mergeClusterEvents(
	previous map[string]ClusterEvent,
	observed []ClusterEvent,
) map[string]ClusterEvent {
	events := make(map[string]ClusterEvent, len(previous)+len(observed))
	maps.Copy(events, previous)
	retainLatestClusterEvents(events, observed)
	return events
}

func mergeClusterEventWindowEvents(previous, observed []ClusterEvent) []ClusterEvent {
	// Fast polling returns one event per cluster in EventID order, and every
	// observed EventID is above the previous window's high watermark.
	replacedNames := make(map[string]struct{}, len(observed))
	for _, event := range observed {
		replacedNames[event.Name] = struct{}{}
	}

	events := make([]ClusterEvent, 0, len(previous)+len(observed))
	for _, event := range previous {
		if _, replaced := replacedNames[event.Name]; replaced {
			continue
		}
		events = append(events, event)
	}
	return append(events, observed...)
}

func clusterEventMap(events ...ClusterEvent) map[string]ClusterEvent {
	eventsByName := make(map[string]ClusterEvent, len(events))
	retainLatestClusterEvents(eventsByName, events)
	return eventsByName
}

func findNewClusterEvents(
	previous map[string]ClusterEvent,
	observed []ClusterEvent,
) []ClusterEvent {
	newEvents := make([]ClusterEvent, 0)
	for _, event := range observed {
		previousEvent, found := previous[event.Name]
		if found && event.EventID <= previousEvent.EventID {
			continue
		}
		newEvents = append(newEvents, event)
	}
	return newEvents
}

func clusterIDsToFetch(events []ClusterEvent) []string {
	clusterIDs := make([]string, 0, len(events))
	for _, event := range events {
		switch event.EventType {
		case inframysql.EventTypeClusterCreated, inframysql.EventTypeClusterUpdated:
			clusterIDs = append(clusterIDs, event.Name)
		}
	}
	return clusterIDs
}

func (m *memDBStorage) replaceClusterMemDB(clusters []*inframysql.Cluster) error {
	txn := m.clusterMemDB.Txn(true)
	defer txn.Abort()

	it, err := txn.Get(clusterMetaTableName, "id")
	if err != nil {
		return fmt.Errorf("failed to list existing cluster meta: %w", err)
	}
	var existing []*inframysql.Cluster
	for obj := it.Next(); obj != nil; obj = it.Next() {
		existing = append(existing, obj.(*inframysql.Cluster))
	}
	for _, cluster := range existing {
		if err := txn.Delete(clusterMetaTableName, cluster); err != nil && !errors.Is(err, memdb.ErrNotFound) {
			return fmt.Errorf("failed to delete existing cluster %s from memdb: %w", cluster.ClusterID, err)
		}
	}
	for _, cluster := range clusters {
		if err := txn.Insert(clusterMetaTableName, cluster); err != nil {
			return fmt.Errorf("failed to insert cluster meta: %w", err)
		}
	}
	txn.Commit()
	return nil
}

func (m *memDBStorage) reloadAllClusterState(ctx context.Context) error {
	return m.reloadAllClusterStateWithEpoch(ctx, false)
}

func (m *memDBStorage) reloadAllClusterStateWithEpoch(ctx context.Context, rotateEventEpoch bool) error {
	m.clusterEventMu.Lock()
	defer m.clusterEventMu.Unlock()

	nextEventEpoch := m.clusterEventEpoch
	if rotateEventEpoch {
		var err error
		nextEventEpoch, err = nextClusterEventEpoch(m.clusterEventEpoch)
		if err != nil {
			return err
		}
	}

	pollStartedAt := time.Now()
	result := metametrics.ResultFailure
	defer func() {
		metametrics.ClusterSyncOperationsTotal.WithLabelValues(
			metametrics.ClusterSyncOperationReload,
			result,
		).Inc()
		metametrics.ClusterSyncDuration.WithLabelValues(
			metametrics.ClusterSyncOperationReload,
		).Observe(time.Since(pollStartedAt).Seconds())
	}()

	log.Info("[storage] reloading all cluster state from mysql")

	var (
		clusters       []*inframysql.Cluster
		observedEvents []ClusterEvent
	)
	previousLastEventID := m.lastClusterEventID
	err := m.runMySQLTransaction(ctx, func(tx *gorm.DB) error {
		var err error
		clusters, err = m.queryAllClusters(tx)
		if err != nil {
			return err
		}
		observedEvents, err = m.pollRetainedClusterEvents(tx)
		return err
	})
	if err != nil {
		return err
	}

	if err := m.replaceClusterMemDB(clusters); err != nil {
		return err
	}
	lastEventID := latestClusterEventID(previousLastEventID, observedEvents)
	events := clusterEventMap(observedEvents...)
	rewindLowerBound := advanceClusterEventRewindLowerBound(
		m.clusterEventRewindLowerBound,
		m.lastClusterEvents,
		events,
	)
	if rotateEventEpoch {
		m.clusterEventEpoch = nextEventEpoch
		rewindLowerBound = 0
	}
	m.clusterEventRewindLowerBound = rewindLowerBound
	m.lastClusterEventID = lastEventID
	m.lastClusterEvents = events
	m.lastClusterEventPollSuccess = pollStartedAt
	m.lastRetainedEventScanSuccess = pollStartedAt
	m.publishClusterEventWindow(observedEvents)
	result = metametrics.ResultSuccess

	metametrics.ClusterCacheEntries.Set(float64(len(clusters)))
	metametrics.ClusterEventSnapshotEntries.Set(float64(len(events)))
	metametrics.ClusterEventPollLastSuccessTimestamp.Set(float64(pollStartedAt.Unix()))

	log.Debug("[storage] all cluster state reloaded",
		zap.Int("totalCluster", len(clusters)),
		zap.Uint64("previousLastEventID", previousLastEventID),
		zap.Uint64("lastEventID", lastEventID),
		zap.Uint64("rewindLowerBound", rewindLowerBound),
		zap.Bool("fullSafetyScan", true),
		zap.Bool("rotatedEventEpoch", rotateEventEpoch),
		zap.Int("eventCount", len(events)),
		zap.Duration("duration", time.Since(pollStartedAt)),
	)
	return nil
}

func (m *memDBStorage) applyClusterEvents(
	events []ClusterEvent,
	clusters []*inframysql.Cluster,
) (int, error) {
	if len(events) == 0 {
		return 0, nil
	}

	clustersByID := make(map[string]*inframysql.Cluster, len(clusters))
	for _, cluster := range clusters {
		clustersByID[cluster.ClusterID] = cluster
	}

	txn := m.clusterMemDB.Txn(true)
	defer txn.Abort()

	cacheDelta := 0
	for _, event := range events {
		switch event.EventType {
		case inframysql.EventTypeClusterCreated, inframysql.EventTypeClusterUpdated:
			cluster, found := clustersByID[event.Name]
			if !found {
				existed, err := deleteClusterFromMemDB(txn, event.Name)
				if err != nil {
					return 0, fmt.Errorf("failed to remove missing cluster %s from memdb: %w", event.Name, err)
				}
				if existed {
					cacheDelta--
				}
				log.Warn("[storage] cluster data not found for create/update event; ensured cache entry is absent",
					zap.String("clusterID", event.Name),
					zap.Uint64("eventID", event.EventID),
					zap.Bool("removedFromCache", existed))
				continue
			}

			it, err := txn.Get(clusterMetaTableName, "id", event.Name)
			if err != nil {
				return 0, fmt.Errorf("failed to lookup existing cluster %s in memdb: %w", event.Name, err)
			}
			existsInCache := it.Next() != nil
			if err := txn.Insert(clusterMetaTableName, cluster); err != nil {
				return 0, fmt.Errorf("failed to insert/update cluster %s in memdb: %w", event.Name, err)
			}
			if !existsInCache {
				cacheDelta++
			}

		case inframysql.EventTypeClusterDeleted:
			existed, err := deleteClusterFromMemDB(txn, event.Name)
			if err != nil {
				return 0, fmt.Errorf("failed to delete cluster %s from memdb: %w", event.Name, err)
			}
			if existed {
				cacheDelta--
			}

		default:
			log.Warn("[storage] unknown cluster event type",
				zap.String("type", event.EventType),
				zap.Uint64("eventID", event.EventID))
		}
	}
	txn.Commit()
	return cacheDelta, nil
}

func deleteClusterFromMemDB(txn *memdb.Txn, clusterID string) (bool, error) {
	it, err := txn.Get(clusterMetaTableName, "id", clusterID)
	if err != nil {
		return false, err
	}

	exists := it.Next() != nil
	cluster := &inframysql.Cluster{ClusterID: clusterID}
	if err := txn.Delete(clusterMetaTableName, cluster); err != nil && !errors.Is(err, memdb.ErrNotFound) {
		return false, err
	}
	return exists, nil
}

func (m *memDBStorage) reconcileClusterEventsBefore(
	ctx context.Context,
	safetyScanDeadline time.Time,
) error {
	m.clusterEventMu.Lock()
	defer m.clusterEventMu.Unlock()

	pollStartedAt := time.Now()
	result := metametrics.ResultFailure
	defer func() {
		metametrics.ClusterSyncOperationsTotal.WithLabelValues(
			metametrics.ClusterSyncOperationReconcile,
			result,
		).Inc()
		metametrics.ClusterSyncDuration.WithLabelValues(
			metametrics.ClusterSyncOperationReconcile,
		).Observe(time.Since(pollStartedAt).Seconds())
	}()

	var (
		observedEvents   []ClusterEvent
		newEvents        []ClusterEvent
		clustersToUpdate []*inframysql.Cluster
	)
	previousLastEventID := m.lastClusterEventID
	fullSafetyScan := m.lastRetainedEventScanSuccess.IsZero() ||
		pollStartedAt.Sub(m.lastRetainedEventScanSuccess) >= clusterRetainedEventScanInterval
	err := m.runMySQLTransaction(ctx, func(tx *gorm.DB) error {
		var err error
		// TMS serializes event-producing mutations for one cluster through its
		// clusters row: existing rows are locked before the Event insert, while
		// Create establishes the unique parent row before inserting its Event.
		// Event IDs are therefore commit-monotonic within a cluster. Different
		// clusters do not share that lock: AUTO_INCREMENT allocation can still
		// precede commit, so a lower-ID Event for another cluster may become
		// visible late. The fast path advances from the global EventID cursor,
		// while the periodic full retained-Event scan catches lower-ID commits
		// before Event TTL cleanup removes them.
		if fullSafetyScan {
			observedEvents, err = m.pollRetainedClusterEvents(tx)
		} else {
			observedEvents, err = m.pollClusterEvents(tx, previousLastEventID)
		}
		if err != nil {
			return err
		}

		newEvents = findNewClusterEvents(m.lastClusterEvents, observedEvents)
		clustersToUpdate, err = m.queryClustersByIDs(tx, clusterIDsToFetch(newEvents))
		return err
	})
	if err != nil {
		return err
	}
	if !safetyScanDeadline.IsZero() && !time.Now().Before(safetyScanDeadline) {
		return errEventSafetyScanDeadlineExceeded
	}

	cacheDelta, err := m.applyClusterEvents(newEvents, clustersToUpdate)
	if err != nil {
		return err
	}
	lastEventID := latestClusterEventID(previousLastEventID, observedEvents)
	events := m.lastClusterEvents
	var publishedEvents []ClusterEvent
	if fullSafetyScan {
		events = clusterEventMap(observedEvents...)
		publishedEvents = observedEvents
	} else if len(observedEvents) > 0 {
		retainLatestClusterEvents(events, observedEvents)
		publishedEvents = mergeClusterEventWindowEvents(
			m.clusterEventWindow.Load().Events,
			observedEvents,
		)
	}
	if fullSafetyScan {
		m.clusterEventRewindLowerBound = advanceClusterEventRewindLowerBound(
			m.clusterEventRewindLowerBound,
			m.lastClusterEvents,
			events,
		)
	}
	m.lastClusterEventID = lastEventID
	m.lastClusterEvents = events
	m.lastClusterEventPollSuccess = pollStartedAt
	if fullSafetyScan {
		m.lastRetainedEventScanSuccess = pollStartedAt
	}
	if publishedEvents != nil {
		m.publishClusterEventWindow(publishedEvents)
	}
	result = metametrics.ResultSuccess

	metametrics.ClusterCacheEntries.Add(float64(cacheDelta))
	metametrics.ClusterEventSnapshotEntries.Set(float64(len(events)))
	metametrics.ClusterEventPollLastSuccessTimestamp.Set(float64(pollStartedAt.Unix()))
	metametrics.ClusterEventChangedClusters.Observe(float64(len(newEvents)))

	log.Debug("[storage] reconciled cluster state from event diff",
		zap.Uint64("previousLastEventID", previousLastEventID),
		zap.Uint64("lastEventID", lastEventID),
		zap.Uint64("rewindLowerBound", m.clusterEventRewindLowerBound),
		zap.Bool("fullSafetyScan", fullSafetyScan),
		zap.Int("eventCount", len(events)),
		zap.Int("newEventCount", len(newEvents)),
		zap.Int("changedClusterCount", len(newEvents)),
		zap.Int("clusterUpdateCount", len(clustersToUpdate)),
		zap.Duration("duration", time.Since(pollStartedAt)),
	)
	return nil
}

func (m *memDBStorage) reloadAllClusterStateAfterEventSafetyGap(
	ctx context.Context,
	lastSuccess time.Time,
) error {
	gap := time.Since(lastSuccess)
	metametrics.ClusterEventGapReloadsTotal.Inc()
	log.Warn("[storage] retained cluster Event safety scan gap exceeded event TTL; reloading all cluster state",
		zap.Time("lastSuccess", lastSuccess),
		zap.Duration("gap", gap),
		zap.Duration("eventTTL", serverlessinfra.DefaultClusterEventTTL),
	)
	return m.reloadAllClusterStateWithEpoch(ctx, true)
}

func (m *memDBStorage) syncClusterEvents(ctx context.Context) error {
	now := time.Now()

	m.clusterEventMu.RLock()
	lastSuccess := m.lastClusterEventPollSuccess
	lastSafetyScanSuccess := m.lastRetainedEventScanSuccess
	m.clusterEventMu.RUnlock()

	gap := now.Sub(lastSuccess)
	if lastSuccess.IsZero() {
		gap = 0
	}
	metametrics.ClusterEventPollGap.Observe(gap.Seconds())

	safetyScanDeadline := lastSafetyScanSuccess.Add(serverlessinfra.DefaultClusterEventTTL)
	if lastSafetyScanSuccess.IsZero() || !now.Before(safetyScanDeadline) {
		if lastSafetyScanSuccess.IsZero() {
			return m.reloadAllClusterState(ctx)
		}
		return m.reloadAllClusterStateAfterEventSafetyGap(ctx, lastSafetyScanSuccess)
	}

	err := m.reconcileClusterEventsBefore(ctx, safetyScanDeadline)
	if errors.Is(err, errEventSafetyScanDeadlineExceeded) {
		return m.reloadAllClusterStateAfterEventSafetyGap(ctx, lastSafetyScanSuccess)
	}
	return err
}

func (m *memDBStorage) watchClusterEvents(ctx context.Context) {
	eventPollTicker := time.NewTicker(clusterEventPollInterval)
	defer eventPollTicker.Stop()

	log.Info("[storage] starting cluster event watcher",
		zap.Duration("interval", clusterEventPollInterval))

	for {
		select {
		case <-ctx.Done():
			log.Info("[storage] stopping cluster event watcher")
			return
		case <-eventPollTicker.C:
			if err := m.syncClusterEvents(ctx); err != nil {
				log.Error("[storage] failed to poll cluster events", zap.Error(err))
			}
		}
	}
}

type zoneRedundantEnabledIndexer struct{}

func (zoneRedundantEnabledIndexer) FromArgs(args ...any) ([]byte, error) {
	if len(args) != 1 {
		return nil, fmt.Errorf("invalid args for zone redundant enabled indexer")
	}
	zoneRedundantEnabled, ok := args[0].(string)
	if !ok {
		return nil, fmt.Errorf("invalid args for zone redundant enabled indexer")
	}
	return []byte(zoneRedundantEnabled + "\x00"), nil
}

func (zoneRedundantEnabledIndexer) FromObject(raw any) (bool, []byte, error) {
	cluster := raw.(*inframysql.Cluster)
	if cluster == nil {
		return false, nil, nil
	}
	if cluster.Status.ZoneRedundantEnabled == "yes" {
		return true, []byte(zoneRedundantEnabled + "\x00"), nil
	}

	return false, nil, nil
}

type clusterBillingPlanIndexer struct{}

func (clusterBillingPlanIndexer) FromArgs(args ...any) ([]byte, error) {
	if len(args) != 1 {
		return nil, fmt.Errorf("invalid args for cluster billing plan indexer")
	}
	billingPlan, ok := args[0].(string)
	if !ok {
		return nil, fmt.Errorf("invalid args for cluster billing plan indexer")
	}
	return []byte(billingPlan + "\x00"), nil
}

func (clusterBillingPlanIndexer) FromObject(raw any) (bool, []byte, error) {
	cluster := raw.(*inframysql.Cluster)
	if infra.IsEssentialClusterMysql(cluster) {
		return true, []byte(essentialPlanName + "\x00"), nil
	}
	return false, nil, nil
}