package storage

import (
	"context"
	"fmt"
	"path/filepath"
	"sync"
	"testing"
	"time"

	"github.com/hashicorp/go-memdb"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/testutil"
	dto "github.com/prometheus/client_model/go"
	"github.com/stretchr/testify/require"
	metametrics "github.com/tidbcloud/aws-shared-provider/pkg/metaserver/metrics"
	serverlessinfra "github.com/tidbcloud/tidb-management-service/api/serverless_infra"
	inframysql "github.com/tidbcloud/tidb-management-service/api/serverless_infra/client/mysql"
	"gorm.io/driver/sqlite"
	"gorm.io/gorm"
)

func (m *memDBStorage) reconcileClusterEvents(ctx context.Context) error {
	return m.reconcileClusterEventsBefore(ctx, time.Time{})
}

func TestNewClusterEventEpoch(t *testing.T) {
	first, err := newClusterEventEpoch()
	require.NoError(t, err)
	require.Len(t, first, 32)

	second, err := newClusterEventEpoch()
	require.NoError(t, err)
	require.Len(t, second, 32)
	require.NotEqual(t, first, second)
}

func TestGetClusterEventWindowReturnsPublishedSortedCopyAndBounds(t *testing.T) {
	store := newTestClusterStore(t, openTestClusterDB(t))
	store.clusterEventRewindLowerBound = 2
	store.lastClusterEventID = 5
	store.lastClusterEvents = clusterEventMap(
		testClusterEvent(5, "updated", inframysql.EventTypeClusterUpdated, 2),
		testClusterEvent(1, "expired", inframysql.EventTypeClusterCreated, 1),
		testClusterEvent(4, "deleted", inframysql.EventTypeClusterDeleted, 3),
	)
	store.publishClusterEventWindow(sortedClusterEvents(store.lastClusterEvents))
	published := store.clusterEventWindow.Load()

	window, err := store.GetClusterEventWindow()
	require.NoError(t, err)
	require.Equal(t, ClusterEventWindow{
		Epoch:            "test-epoch",
		HighWatermark:    5,
		RewindLowerBound: 2,
		Events: []ClusterEvent{
			testClusterEvent(1, "expired", inframysql.EventTypeClusterCreated, 1),
			testClusterEvent(4, "deleted", inframysql.EventTypeClusterDeleted, 3),
			testClusterEvent(5, "updated", inframysql.EventTypeClusterUpdated, 2),
		},
	}, window)

	window.Events[0].Name = "mutated"
	windowAgain, err := store.GetClusterEventWindow()
	require.NoError(t, err)
	require.Equal(t, "expired", windowAgain.Events[0].Name)
	require.Same(t, published, store.clusterEventWindow.Load())
	require.Equal(t, "expired", published.Events[0].Name)

	_, err = newTestClusterStore(t, nil).GetClusterEventWindow()
	require.ErrorContains(t, err, "mysql not initialized")
}

func TestQueryAllClustersLoadsAssociationsInBatches(t *testing.T) {
	db := openTestClusterDB(t)
	expectedCount := clusterLoadBatchSize + 1
	const otherKubernetesID = "other-kubernetes"
	require.NoError(t, db.Transaction(func(tx *gorm.DB) error {
		for i := range expectedCount {
			insertTestCluster(t, tx, testCluster(fmt.Sprintf("cluster-%04d", i), uint64(i+1)))
		}
		otherKubernetes := testCluster(otherKubernetesID, 1)
		otherKubernetes.Kubernetes = "kube-b"
		insertTestCluster(t, tx, otherKubernetes)
		return nil
	}))

	queryCounts := make(map[string]int)
	require.NoError(t, db.Callback().Query().Before("gorm:query").
		Register("test:count_cluster_batch_queries", func(tx *gorm.DB) {
			if tx.Statement.Schema != nil {
				queryCounts[tx.Statement.Schema.Table]++
			}
		}))

	successCountBefore, successSumBefore := clusterFullLoadDurationMetric(t, metametrics.ResultSuccess)
	clusters, err := newTestClusterStore(t, db).queryAllClusters(db)
	require.NoError(t, err)
	require.Len(t, clusters, expectedCount)
	require.Equal(t, 2, queryCounts["clusters"])
	require.Equal(t, 2, queryCounts["cluster_specs"])
	require.Equal(t, 2, queryCounts["cluster_statuses"])
	successCountAfter, successSumAfter := clusterFullLoadDurationMetric(t, metametrics.ResultSuccess)
	require.Equal(t, successCountBefore+1, successCountAfter)
	require.Greater(t, successSumAfter, successSumBefore)

	clustersByID := make(map[string]*inframysql.Cluster, len(clusters))
	for _, cluster := range clusters {
		clustersByID[cluster.ClusterID] = cluster
	}
	require.NotContains(t, clustersByID, otherKubernetesID)
	for i := range expectedCount {
		clusterID := fmt.Sprintf("cluster-%04d", i)
		cluster, ok := clustersByID[clusterID]
		require.True(t, ok, "cluster %s was not loaded", clusterID)
		require.Equal(t, clusterID, cluster.Spec.ClusterID)
		require.Equal(t, clusterID, cluster.Status.ClusterID)
		require.Equal(t, uint64(i+1), cluster.Status.ResourceVersion)
	}
}

func TestQueryAllClustersRecordsFailureDuration(t *testing.T) {
	db := openTestClusterDB(t)
	require.NoError(t, db.Exec("DROP TABLE clusters").Error)

	failureCountBefore, failureSumBefore := clusterFullLoadDurationMetric(t, metametrics.ResultFailure)
	_, err := newTestClusterStore(t, db).queryAllClusters(db)
	require.ErrorContains(t, err, "failed to fetch clusters from mysql")
	failureCountAfter, failureSumAfter := clusterFullLoadDurationMetric(t, metametrics.ResultFailure)
	require.Equal(t, failureCountBefore+1, failureCountAfter)
	require.Greater(t, failureSumAfter, failureSumBefore)
}

func TestGetClusterEventWindowDoesNotWaitForReconcileLock(t *testing.T) {
	store := newTestClusterStore(t, openTestClusterDB(t))
	store.clusterEventMu.Lock()
	defer store.clusterEventMu.Unlock()

	result := make(chan ClusterEventWindow, 1)
	errCh := make(chan error, 1)
	go func() {
		window, err := store.GetClusterEventWindow()
		result <- window
		errCh <- err
	}()

	select {
	case window := <-result:
		require.Equal(t, "test-epoch", window.Epoch)
		require.NoError(t, <-errCh)
	case <-time.After(time.Second):
		t.Fatal("cluster event window read blocked on reconcile lock")
	}
}

func TestAdvanceClusterEventRewindLowerBound(t *testing.T) {
	previous := clusterEventMap(
		testClusterEvent(1, "pending", "", 0),
		testClusterEvent(10, "removed", "", 0),
		testClusterEvent(11, "updated", "", 0),
		testClusterEvent(12, "regressed", "", 0),
	)
	observed := clusterEventMap(
		testClusterEvent(1, "pending", "", 0),
		testClusterEvent(13, "updated", "", 0),
		testClusterEvent(2, "regressed", "", 0),
	)

	require.Equal(t, uint64(12), advanceClusterEventRewindLowerBound(0, previous, observed))
	require.Equal(t, uint64(20), advanceClusterEventRewindLowerBound(20, previous, observed))
	require.Equal(t, uint64(0), advanceClusterEventRewindLowerBound(0, nil, observed))
}

func TestReloadAllClusterStateUsesAuthoritativeClusterState(t *testing.T) {
	db := openTestClusterDB(t)
	updated := testCluster("updated", 2)
	created := testCluster("created", 1)
	insertTestCluster(t, db, updated)
	insertTestCluster(t, db, created)
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         7,
		Name:            "updated",
		Namespace:       "default",
		Kubernetes:      "kube-a",
		EventType:       inframysql.EventTypeClusterUpdated,
		ResourceVersion: 2,
	})
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         8,
		Name:            "other-kube",
		Namespace:       "default",
		Kubernetes:      "kube-b",
		EventType:       inframysql.EventTypeClusterDeleted,
		ResourceVersion: 9,
	})

	store := newTestClusterStore(t, db)
	require.NoError(t, store.replaceClusterMemDB([]*inframysql.Cluster{
		testCluster("updated", 1),
		testCluster("deleted", 1),
	}))
	initialEpoch := requireClusterEventEpoch(t, store)

	require.NoError(t, store.reloadAllClusterState(t.Context()))
	require.Equal(t, initialEpoch, requireClusterEventEpoch(t, store))

	clusters, err := store.LookupCluster(IndexClusterID, "updated")
	require.NoError(t, err)
	require.Len(t, clusters, 1)
	require.Equal(t, uint64(2), clusters[0].Status.ResourceVersion)

	clusters, err = store.LookupCluster(IndexClusterID, "deleted")
	require.NoError(t, err)
	require.Empty(t, clusters)

	clusters, err = store.LookupCluster(IndexClusterID, "created")
	require.NoError(t, err)
	require.Len(t, clusters, 1)
	require.Equal(t, uint64(1), clusters[0].Status.ResourceVersion)

	require.Equal(t, []ClusterEvent{
		testClusterEvent(7, "updated", inframysql.EventTypeClusterUpdated, 2),
	}, snapshotLastClusterEvents(store))
	require.Equal(t, uint64(7), snapshotLastClusterEventID(store))
	require.Zero(t, snapshotClusterEventRewindLowerBound(store))
}

func TestFindNewClusterEventsUsesPerClusterMaxEventID(t *testing.T) {
	previous := clusterEventMap(
		testClusterEvent(11, "deleted", "", 0),
		testClusterEvent(12, "current", "", 0),
		testClusterEvent(14, "expired", "", 0),
	)
	observed := []ClusterEvent{
		testClusterEvent(10, "late-other-cluster", inframysql.EventTypeClusterDeleted, 0),
		testClusterEvent(10, "deleted", "", 0),
		testClusterEvent(11, "deleted", "", 0),
		testClusterEvent(12, "current", "", 0),
		testClusterEvent(13, "created", inframysql.EventTypeClusterCreated, 0),
	}

	newEvents := findNewClusterEvents(previous, observed)
	require.Equal(t, []ClusterEvent{
		testClusterEvent(10, "late-other-cluster", inframysql.EventTypeClusterDeleted, 0),
		testClusterEvent(13, "created", inframysql.EventTypeClusterCreated, 0),
	}, newEvents)
	require.Equal(t, []string{"created"}, clusterIDsToFetch(newEvents))
}

func TestClusterEventStateHelpers(t *testing.T) {
	events := clusterEventMap(
		testClusterEvent(10, "expired", "", 0),
		testClusterEvent(11, "current", "", 0),
		testClusterEvent(20, "latest", "", 0),
	)
	require.Equal(t, []ClusterEvent{
		testClusterEvent(10, "expired", "", 0),
		testClusterEvent(11, "current", "", 0),
		testClusterEvent(20, "latest", "", 0),
	}, sortedClusterEvents(events))
	require.Equal(t, uint64(20), latestClusterEventID(15, sortedClusterEvents(events)))

	merged := mergeClusterEvents(
		clusterEventMap(
			testClusterEvent(10, "retained", "", 0),
			testClusterEvent(12, "updated", "", 0),
		),
		[]ClusterEvent{
			testClusterEvent(11, "new", "", 0),
			testClusterEvent(12, "updated", "", 0),
		},
	)
	require.Equal(t, []ClusterEvent{
		testClusterEvent(10, "retained", "", 0),
		testClusterEvent(11, "new", "", 0),
		testClusterEvent(12, "updated", "", 0),
	}, sortedClusterEvents(merged))

	retained := clusterEventMap(
		testClusterEvent(9, "retained", "", 0),
		testClusterEvent(12, "updated", "", 0),
	)
	require.Equal(t, []ClusterEvent{
		testClusterEvent(9, "retained", "", 0),
		testClusterEvent(12, "updated", "", 0),
	}, sortedClusterEvents(retained))
	require.Empty(t, clusterEventMap())

	advanced := mergeClusterEvents(
		clusterEventMap(testClusterEvent(10, "updated", inframysql.EventTypeClusterCreated, 1)),
		[]ClusterEvent{
			testClusterEvent(11, "updated", inframysql.EventTypeClusterUpdated, 2),
		},
	)
	require.Equal(t, []ClusterEvent{
		testClusterEvent(11, "updated", inframysql.EventTypeClusterUpdated, 2),
	}, sortedClusterEvents(advanced))
}

func TestMergeClusterEventWindowEventsPreservesEventIDOrder(t *testing.T) {
	previous := []ClusterEvent{
		testClusterEvent(10, "retained", "", 0),
		testClusterEvent(12, "updated", "", 0),
		testClusterEvent(15, "untouched", "", 0),
	}
	observed := []ClusterEvent{
		testClusterEvent(20, "updated", "", 0),
		testClusterEvent(21, "created", "", 0),
	}

	require.Equal(t, []ClusterEvent{
		testClusterEvent(10, "retained", "", 0),
		testClusterEvent(15, "untouched", "", 0),
		testClusterEvent(20, "updated", "", 0),
		testClusterEvent(21, "created", "", 0),
	}, mergeClusterEventWindowEvents(previous, observed))
}

func TestReloadAllClusterStateKeepsCursorMonotonicAndRebuildsRetainedMarkers(t *testing.T) {
	db := openTestClusterDB(t)
	insertTestCluster(t, db, testCluster("cluster", 1))
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         10,
		Name:            "cluster",
		Kubernetes:      "kube-a",
		EventType:       inframysql.EventTypeClusterCreated,
		ResourceVersion: 1,
	})
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         11,
		Name:            "cluster",
		Kubernetes:      "kube-a",
		EventType:       inframysql.EventTypeClusterUpdated,
		ResourceVersion: 1,
	})

	store := newTestClusterStore(t, db)
	require.NoError(t, store.reloadAllClusterState(t.Context()))
	require.NoError(t, db.Exec("DELETE FROM cluster_events").Error)
	require.NoError(t, store.reloadAllClusterState(t.Context()))

	require.Equal(t, uint64(11), snapshotLastClusterEventID(store))
	require.Empty(t, snapshotLastClusterEvents(store))
	require.Equal(t, uint64(11), snapshotClusterEventRewindLowerBound(store))

	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         16,
		Name:            "cluster",
		Kubernetes:      "kube-a",
		EventType:       inframysql.EventTypeClusterUpdated,
		ResourceVersion: 1,
	})
	require.NoError(t, store.reloadAllClusterState(t.Context()))

	require.Equal(t, uint64(16), snapshotLastClusterEventID(store))
	require.Equal(t, []ClusterEvent{
		testClusterEvent(16, "cluster", inframysql.EventTypeClusterUpdated, 1),
	}, snapshotLastClusterEvents(store))
	require.Equal(t, uint64(11), snapshotClusterEventRewindLowerBound(store))
}

func TestPollClusterEventsAcrossStatusesAndKubernetes(t *testing.T) {
	db := openTestClusterDB(t)
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:    5,
		Name:       "pending",
		Kubernetes: "kube-a",
		Status:     inframysql.StatusPending,
	})
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:    7,
		Name:       "done",
		Kubernetes: "kube-a",
		Status:     inframysql.StatusDone,
	})
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:    9,
		Name:       "other-kube",
		Kubernetes: "kube-b",
		Status:     inframysql.StatusDone,
	})
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:    10,
		Name:       "",
		Kubernetes: "kube-a",
		Status:     inframysql.StatusDone,
	})

	store := newTestClusterStore(t, db)
	events, err := store.pollRetainedClusterEvents(db)
	require.NoError(t, err)
	require.Equal(t, []ClusterEvent{
		testClusterEvent(5, "pending", "", 0),
		testClusterEvent(7, "done", "", 0),
	}, events)

	events, err = store.pollClusterEvents(db, 5)
	require.NoError(t, err)
	require.Equal(t, []ClusterEvent{
		testClusterEvent(7, "done", "", 0),
	}, events)
}

func TestPollRetainedClusterEventsReturnsLatestEventPerCluster(t *testing.T) {
	db := openTestClusterDB(t)
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         5,
		Name:            "cluster-a",
		Kubernetes:      "kube-a",
		EventType:       inframysql.EventTypeClusterCreated,
		ResourceVersion: 1,
	})
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         6,
		Name:            "cluster-b",
		Kubernetes:      "kube-a",
		EventType:       inframysql.EventTypeClusterUpdated,
		ResourceVersion: 4,
	})
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         7,
		Name:            "cluster-a",
		Kubernetes:      "kube-a",
		EventType:       inframysql.EventTypeClusterUpdated,
		ResourceVersion: 2,
	})

	store := newTestClusterStore(t, db)
	events, err := store.pollRetainedClusterEvents(db)
	require.NoError(t, err)
	require.Equal(t, []ClusterEvent{
		testClusterEvent(6, "cluster-b", inframysql.EventTypeClusterUpdated, 4),
		testClusterEvent(7, "cluster-a", inframysql.EventTypeClusterUpdated, 2),
	}, events)
}

func TestRetainLatestClusterEvent(t *testing.T) {
	events := make(map[string]ClusterEvent)
	retainLatestClusterEvent(events,
		testClusterEvent(5, "cluster", inframysql.EventTypeClusterCreated, 1))
	retainLatestClusterEvent(events,
		testClusterEvent(7, "cluster", inframysql.EventTypeClusterUpdated, 2))
	retainLatestClusterEvent(events,
		testClusterEvent(6, "cluster", inframysql.EventTypeClusterUpdated, 4))
	retainLatestClusterEvent(events,
		testClusterEvent(8, "", inframysql.EventTypeClusterDeleted, 3))
	require.Equal(t,
		testClusterEvent(7, "cluster", inframysql.EventTypeClusterUpdated, 2),
		events["cluster"])
	require.Len(t, events, 1)
}

func TestApplyClusterEventsUsesEventType(t *testing.T) {
	store := newTestClusterStore(t, nil)
	require.NoError(t, store.replaceClusterMemDB([]*inframysql.Cluster{
		testCluster("updated", 1),
		testCluster("deleted", 1),
		testCluster("unknown", 1),
	}))

	cacheDelta, err := store.applyClusterEvents(
		[]ClusterEvent{
			testClusterEvent(10, "updated", inframysql.EventTypeClusterUpdated, 2),
			testClusterEvent(11, "created", inframysql.EventTypeClusterCreated, 1),
			testClusterEvent(12, "deleted", inframysql.EventTypeClusterDeleted, 2),
			testClusterEvent(13, "unknown", "UNKNOWN", 2),
		},
		[]*inframysql.Cluster{
			testCluster("updated", 2),
			testCluster("created", 1),
		},
	)
	require.NoError(t, err)
	require.Zero(t, cacheDelta)

	clusters, err := store.LookupCluster(IndexClusterID, "updated")
	require.NoError(t, err)
	require.Len(t, clusters, 1)
	require.Equal(t, uint64(2), clusters[0].Status.ResourceVersion)

	clusters, err = store.LookupCluster(IndexClusterID, "created")
	require.NoError(t, err)
	require.Len(t, clusters, 1)

	clusters, err = store.LookupCluster(IndexClusterID, "deleted")
	require.NoError(t, err)
	require.Empty(t, clusters)

	clusters, err = store.LookupCluster(IndexClusterID, "unknown")
	require.NoError(t, err)
	require.Len(t, clusters, 1)
}

func TestApplyClusterEventsRemovesMissingClusterForCreateOrUpdate(t *testing.T) {
	for _, eventType := range []string{
		inframysql.EventTypeClusterCreated,
		inframysql.EventTypeClusterUpdated,
	} {
		t.Run(eventType, func(t *testing.T) {
			store := newTestClusterStore(t, nil)
			require.NoError(t, store.replaceClusterMemDB([]*inframysql.Cluster{
				testCluster("missing", 1),
			}))

			cacheDelta, err := store.applyClusterEvents(
				[]ClusterEvent{
					testClusterEvent(10, "missing", eventType, 2),
				},
				nil,
			)
			require.NoError(t, err)
			require.Equal(t, -1, cacheDelta)

			clusters, err := store.LookupCluster(IndexClusterID, "missing")
			require.NoError(t, err)
			require.Empty(t, clusters)
		})
	}
}

func TestPollRetainedClusterEventsPaginates(t *testing.T) {
	db := openTestClusterDB(t)
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         1,
		Name:            "late-page-cluster",
		Kubernetes:      "kube-a",
		EventType:       inframysql.EventTypeClusterCreated,
		ResourceVersion: 1,
	})
	for eventID := 2; eventID <= clusterEventPollBatchSize+1; eventID++ {
		status := inframysql.StatusPending
		if eventID%2 == 0 {
			status = inframysql.StatusDone
		}
		insertTestEvent(t, db, &inframysql.ClusterEvent{
			EventID:         uint64(eventID),
			Name:            fmt.Sprintf("cluster-%04d", eventID),
			Kubernetes:      "kube-a",
			EventType:       inframysql.EventTypeClusterUpdated,
			Status:          status,
			ResourceVersion: uint64(eventID),
		})
	}
	latestEventID := uint64(clusterEventPollBatchSize + 2)
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         latestEventID,
		Name:            "late-page-cluster",
		Kubernetes:      "kube-a",
		EventType:       inframysql.EventTypeClusterUpdated,
		ResourceVersion: 2,
	})

	queryCount := 0
	require.NoError(t, db.Callback().Query().After("gorm:query").
		Register("test:count_retained_event_pages", func(tx *gorm.DB) {
			if !tx.DryRun {
				queryCount++
			}
		}))
	store := newTestClusterStore(t, db)
	events, err := store.pollRetainedClusterEvents(db)
	require.NoError(t, err)
	require.Len(t, events, clusterEventPollBatchSize+1)
	require.Equal(t, 2, queryCount)
	require.Equal(t,
		testClusterEvent(
			latestEventID,
			"late-page-cluster",
			inframysql.EventTypeClusterUpdated,
			2,
		),
		events[len(events)-1],
	)
}

func TestPollRetainedClusterEventsRecordsFullScanMetrics(t *testing.T) {
	resetClusterEventFullScanMetrics(t)

	db := openTestClusterDB(t)
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         5,
		Name:            "cluster",
		Kubernetes:      "kube-a",
		EventType:       inframysql.EventTypeClusterCreated,
		ResourceVersion: 1,
	})
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         7,
		Name:            "cluster",
		Kubernetes:      "kube-a",
		EventType:       inframysql.EventTypeClusterUpdated,
		ResourceVersion: 2,
	})

	events, err := newTestClusterStore(t, db).pollRetainedClusterEvents(db)
	require.NoError(t, err)
	require.Equal(t, []ClusterEvent{
		testClusterEvent(7, "cluster", inframysql.EventTypeClusterUpdated, 2),
	}, events)

	rowsCount, rowsSum := successfulClusterEventFullScanMetric(
		t,
		metametrics.ClusterEventFullScanReturnedRows,
	)
	require.Equal(t, uint64(1), rowsCount)
	require.Equal(t, float64(1), rowsSum)
	pagesCount, pagesSum := successfulClusterEventFullScanMetric(
		t,
		metametrics.ClusterEventFullScanPages,
	)
	require.Equal(t, uint64(1), pagesCount)
	require.Equal(t, float64(1), pagesSum)
}

func TestPollRetainedClusterEventsRecordsFailureMetrics(t *testing.T) {
	resetClusterEventFullScanMetrics(t)

	db := openTestClusterDB(t)
	sqlDB, err := db.DB()
	require.NoError(t, err)
	require.NoError(t, sqlDB.Close())

	events, err := newTestClusterStore(t, db).pollRetainedClusterEvents(db)
	require.Error(t, err)
	require.Empty(t, events)

	durationCount, _ := clusterEventFullScanMetric(
		t,
		metametrics.ClusterEventFullScanDuration,
		metametrics.ResultFailure,
	)
	require.Equal(t, uint64(1), durationCount)
	rowsCount, rowsSum := clusterEventFullScanMetric(
		t,
		metametrics.ClusterEventFullScanReturnedRows,
		metametrics.ResultFailure,
	)
	require.Equal(t, uint64(1), rowsCount)
	require.Zero(t, rowsSum)
}

func TestRetainedSafetyScanPrunesMarkersByClusterName(t *testing.T) {
	db := openTestClusterDB(t)
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:    1,
		Name:       "pending",
		Kubernetes: "kube-a",
		Status:     inframysql.StatusPending,
	})
	for eventID := 2; eventID <= 11; eventID++ {
		insertTestEvent(t, db, &inframysql.ClusterEvent{
			EventID:    uint64(eventID),
			Name:       fmt.Sprintf("done-%d", eventID),
			Kubernetes: "kube-a",
			Status:     inframysql.StatusDone,
		})
	}

	store := newTestClusterStore(t, db)
	require.NoError(t, store.reloadAllClusterState(t.Context()))
	require.Len(t, snapshotLastClusterEvents(store), 11)

	require.NoError(t, db.Exec(
		"DELETE FROM cluster_events WHERE status = ?",
		inframysql.StatusDone,
	).Error)
	require.NoError(t, store.reconcileClusterEvents(t.Context()))
	require.Len(t, snapshotLastClusterEvents(store), 11,
		"fast cursor scans retain markers until the next full safety scan")

	store.clusterEventMu.Lock()
	store.lastRetainedEventScanSuccess = time.Now().Add(-clusterRetainedEventScanInterval)
	store.clusterEventMu.Unlock()
	require.NoError(t, store.reconcileClusterEvents(t.Context()))
	require.Equal(t, []ClusterEvent{
		testClusterEvent(1, "pending", "", 0),
	}, snapshotLastClusterEvents(store))
	require.Equal(t, uint64(11), snapshotLastClusterEventID(store))
	require.Equal(t, uint64(11), snapshotClusterEventRewindLowerBound(store))
}

func TestReconcileClusterEventsUsesCursorFastPath(t *testing.T) {
	db := openTestClusterDB(t)
	insertTestCluster(t, db, testCluster("updated", 1))
	insertTestCluster(t, db, testCluster("deleted", 1))
	insertTestCluster(t, db, testCluster("untouched", 1))

	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         11,
		Name:            "updated",
		Namespace:       "default",
		Kubernetes:      "kube-a",
		EventType:       inframysql.EventTypeClusterUpdated,
		Status:          inframysql.StatusPending,
		ResourceVersion: 1,
	})

	store := newTestClusterStore(t, db)
	require.NoError(t, store.reloadAllClusterState(t.Context()))
	initialWindow := store.clusterEventWindow.Load()
	require.Equal(t, []ClusterEvent{
		testClusterEvent(11, "updated", inframysql.EventTypeClusterUpdated, 1),
	}, snapshotLastClusterEvents(store))
	require.Equal(t, float64(3), testutil.ToFloat64(metametrics.ClusterCacheEntries))

	updateTestClusterResourceVersion(t, db, "updated", 2)
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         12,
		Name:            "updated",
		Namespace:       "default",
		Kubernetes:      "kube-a",
		EventType:       inframysql.EventTypeClusterUpdated,
		Status:          inframysql.StatusDone,
		ResourceVersion: 2,
	})
	insertTestCluster(t, db, testCluster("created", 1))
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         13,
		Name:            "created",
		Namespace:       "default",
		Kubernetes:      "kube-a",
		EventType:       inframysql.EventTypeClusterCreated,
		Status:          inframysql.StatusPending,
		ResourceVersion: 1,
	})
	deleteTestCluster(t, db, "deleted")
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         14,
		Name:            "deleted",
		Namespace:       "default",
		Kubernetes:      "kube-a",
		EventType:       inframysql.EventTypeClusterDeleted,
		Status:          inframysql.StatusDone,
		ResourceVersion: 2,
	})
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         15,
		Name:            "untouched",
		Namespace:       "default",
		Kubernetes:      "kube-b",
		EventType:       inframysql.EventTypeClusterDeleted,
		Status:          inframysql.StatusDone,
		ResourceVersion: 1,
	})

	currentEvents, err := store.pollClusterEvents(
		db,
		snapshotLastClusterEventID(store),
	)
	require.NoError(t, err)
	newEvents := findNewClusterEvents(snapshotLastClusterEventMap(store), currentEvents)
	require.Equal(t, []ClusterEvent{
		testClusterEvent(12, "updated", inframysql.EventTypeClusterUpdated, 2),
		testClusterEvent(13, "created", inframysql.EventTypeClusterCreated, 1),
		testClusterEvent(14, "deleted", inframysql.EventTypeClusterDeleted, 2),
	}, newEvents)
	require.Equal(t, []string{"updated", "created"}, clusterIDsToFetch(newEvents))

	canceledCtx, cancel := context.WithCancel(t.Context())
	cancel()
	failuresBefore := testutil.ToFloat64(metametrics.ClusterSyncOperationsTotal.WithLabelValues(
		metametrics.ClusterSyncOperationReconcile,
		metametrics.ResultFailure,
	))
	require.Error(t, store.reconcileClusterEvents(canceledCtx))
	require.Equal(t, []ClusterEvent{
		testClusterEvent(11, "updated", inframysql.EventTypeClusterUpdated, 1),
	}, snapshotLastClusterEvents(store), "a failed poll must not advance the event list")
	require.Equal(t, uint64(11), snapshotLastClusterEventID(store),
		"a failed poll must not advance the last event ID")
	require.Same(t, initialWindow, store.clusterEventWindow.Load(),
		"a failed poll must not publish a new event window")
	require.Equal(t, failuresBefore+1, testutil.ToFloat64(metametrics.ClusterSyncOperationsTotal.WithLabelValues(
		metametrics.ClusterSyncOperationReconcile,
		metametrics.ResultFailure,
	)))

	require.NoError(t, store.reconcileClusterEvents(t.Context()))

	clusters, err := store.LookupCluster(IndexClusterID, "updated")
	require.NoError(t, err)
	require.Len(t, clusters, 1)
	require.Equal(t, uint64(2), clusters[0].Status.ResourceVersion)

	clusters, err = store.LookupCluster(IndexClusterID, "deleted")
	require.NoError(t, err)
	require.Empty(t, clusters)

	clusters, err = store.LookupCluster(IndexClusterID, "created")
	require.NoError(t, err)
	require.Len(t, clusters, 1)
	require.Equal(t, uint64(1), clusters[0].Status.ResourceVersion)

	clusters, err = store.LookupCluster(IndexClusterID, "untouched")
	require.NoError(t, err)
	require.Len(t, clusters, 1, "events from another kubernetes must not change the local cache")

	require.Equal(t, currentEvents, snapshotLastClusterEvents(store))
	require.Equal(t, uint64(14), snapshotLastClusterEventID(store))
	require.Equal(t, float64(3), testutil.ToFloat64(metametrics.ClusterCacheEntries))

	require.Equal(t, []ClusterEvent{
		testClusterEvent(12, "updated", inframysql.EventTypeClusterUpdated, 2),
		testClusterEvent(13, "created", inframysql.EventTypeClusterCreated, 1),
		testClusterEvent(14, "deleted", inframysql.EventTypeClusterDeleted, 2),
	}, snapshotLastClusterEvents(store), "the event state must be sorted and exclude other kubernetes")
	updatedWindow := store.clusterEventWindow.Load()
	require.NotSame(t, initialWindow, updatedWindow)
	window, err := store.GetClusterEventWindow()
	require.NoError(t, err)
	require.Equal(t, ClusterEventWindow{
		Epoch:         "test-epoch",
		HighWatermark: 14,
		Events: []ClusterEvent{
			testClusterEvent(12, "updated", inframysql.EventTypeClusterUpdated, 2),
			testClusterEvent(13, "created", inframysql.EventTypeClusterCreated, 1),
			testClusterEvent(14, "deleted", inframysql.EventTypeClusterDeleted, 2),
		},
	}, window)

	require.NoError(t, db.Exec("DELETE FROM cluster_events WHERE event_id = ?", 11).Error)
	require.NoError(t, store.reconcileClusterEvents(t.Context()))
	require.Equal(t, []ClusterEvent{
		testClusterEvent(12, "updated", inframysql.EventTypeClusterUpdated, 2),
		testClusterEvent(13, "created", inframysql.EventTypeClusterCreated, 1),
		testClusterEvent(14, "deleted", inframysql.EventTypeClusterDeleted, 2),
	}, snapshotLastClusterEvents(store), "removing an older Event does not change the latest per-cluster markers")
	require.Same(t, updatedWindow, store.clusterEventWindow.Load(),
		"an empty cursor poll must reuse the published event window")

	updateTestClusterResourceVersion(t, db, "updated", 3)
	require.NoError(t, store.syncClusterEvents(t.Context()))
	clusters, err = store.LookupCluster(IndexClusterID, "updated")
	require.NoError(t, err)
	require.Len(t, clusters, 1)
	require.Equal(t, uint64(2), clusters[0].Status.ResourceVersion,
		"a cluster change without a new Event must not be processed")
}

func TestReconcileClusterEventsRemovesClusterMissingFromAuthoritativeQuery(t *testing.T) {
	db := openTestClusterDB(t)
	insertTestCluster(t, db, testCluster("missing", 1))
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         1,
		Name:            "missing",
		Namespace:       "default",
		Kubernetes:      "kube-a",
		EventType:       inframysql.EventTypeClusterCreated,
		Status:          inframysql.StatusDone,
		ResourceVersion: 1,
	})

	store := newTestClusterStore(t, db)
	require.NoError(t, store.reloadAllClusterState(t.Context()))

	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         2,
		Name:            "missing",
		Namespace:       "default",
		Kubernetes:      "kube-a",
		EventType:       inframysql.EventTypeClusterUpdated,
		Status:          inframysql.StatusDone,
		ResourceVersion: 2,
	})
	deleteTestCluster(t, db, "missing")

	require.NoError(t, store.reconcileClusterEvents(t.Context()))

	clusters, err := store.LookupCluster(IndexClusterID, "missing")
	require.NoError(t, err)
	require.Empty(t, clusters)
}

func TestSyncClusterEventsRecordsPollGapEveryTick(t *testing.T) {
	db := openTestClusterDB(t)
	insertTestCluster(t, db, testCluster("cluster", 1))
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         1,
		Name:            "cluster",
		Namespace:       "default",
		Kubernetes:      "kube-a",
		EventType:       inframysql.EventTypeClusterCreated,
		Status:          inframysql.StatusDone,
		ResourceVersion: 1,
	})

	store := newTestClusterStore(t, db)
	countBefore, sumBefore := clusterEventPollGapMetric(t)
	require.NoError(t, store.syncClusterEvents(t.Context()))
	countAfterReload, sumAfterReload := clusterEventPollGapMetric(t)
	require.Equal(t, countBefore+1, countAfterReload)
	require.Equal(t, sumBefore, sumAfterReload)

	store.clusterEventMu.Lock()
	store.lastClusterEventPollSuccess = time.Now().Add(-time.Second)
	store.clusterEventMu.Unlock()
	require.NoError(t, store.syncClusterEvents(t.Context()))
	countAfterReconcile, sumAfterReconcile := clusterEventPollGapMetric(t)
	require.Equal(t, countAfterReload+1, countAfterReconcile)
	require.GreaterOrEqual(t, sumAfterReconcile-sumAfterReload, time.Second.Seconds())
}

func TestReconcileClusterEventsRejectsResultAfterSafetyScanDeadline(t *testing.T) {
	db := openTestClusterDB(t)
	insertTestCluster(t, db, testCluster("cluster", 1))
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         1,
		Name:            "cluster",
		Namespace:       "default",
		Kubernetes:      "kube-a",
		EventType:       inframysql.EventTypeClusterCreated,
		Status:          inframysql.StatusDone,
		ResourceVersion: 1,
	})

	store := newTestClusterStore(t, db)
	require.NoError(t, store.reloadAllClusterState(t.Context()))
	previousLastSuccess := store.lastClusterEventPollSuccess

	updateTestClusterResourceVersion(t, db, "cluster", 2)
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         2,
		Name:            "cluster",
		Namespace:       "default",
		Kubernetes:      "kube-a",
		EventType:       inframysql.EventTypeClusterUpdated,
		Status:          inframysql.StatusDone,
		ResourceVersion: 2,
	})

	queryStarted, releaseQuery := blockFirstClusterQuery(t, db)
	deadline := time.Now().Add(100 * time.Millisecond)
	errCh := make(chan error, 1)
	go func() {
		errCh <- store.reconcileClusterEventsBefore(t.Context(), deadline)
	}()

	select {
	case <-queryStarted:
	case <-time.After(2 * time.Second):
		t.Fatal("timed out waiting for cluster Event query")
	}
	time.Sleep(max(time.Until(deadline), 0) + 10*time.Millisecond)
	close(releaseQuery)
	require.ErrorIs(t, <-errCh, errEventSafetyScanDeadlineExceeded)

	clusters, err := store.LookupCluster(IndexClusterID, "cluster")
	require.NoError(t, err)
	require.Len(t, clusters, 1)
	require.Equal(t, uint64(1), clusters[0].Status.ResourceVersion)
	require.Equal(t, []ClusterEvent{
		testClusterEvent(1, "cluster", inframysql.EventTypeClusterCreated, 1),
	}, snapshotLastClusterEvents(store))
	require.Equal(t, uint64(1), snapshotLastClusterEventID(store))
	store.clusterEventMu.RLock()
	require.Equal(t, previousLastSuccess, store.lastClusterEventPollSuccess)
	store.clusterEventMu.RUnlock()
}

func TestClusterRetainedEventScanInterval(t *testing.T) {
	require.Equal(t, 120*time.Second, clusterRetainedEventScanInterval)
	require.Less(t, clusterRetainedEventScanInterval, serverlessinfra.DefaultClusterEventTTL)
}

func TestReconcileClusterEventsRunsRetainedSafetyScan(t *testing.T) {
	db := openTestClusterDB(t)
	insertTestCluster(t, db, testCluster("current", 1))
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         100,
		Name:            "current",
		Namespace:       "default",
		Kubernetes:      "kube-a",
		EventType:       inframysql.EventTypeClusterCreated,
		Status:          inframysql.StatusDone,
		ResourceVersion: 1,
	})

	store := newTestClusterStore(t, db)
	require.NoError(t, store.reloadAllClusterState(t.Context()))
	require.False(t, snapshotLastRetainedEventScanSuccess(store).IsZero())

	insertTestCluster(t, db, testCluster("late", 1))
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         50,
		Name:            "late",
		Namespace:       "default",
		Kubernetes:      "kube-a",
		EventType:       inframysql.EventTypeClusterCreated,
		Status:          inframysql.StatusDone,
		ResourceVersion: 1,
	})

	fastScanBaseline := time.Now().Add(-clusterRetainedEventScanInterval + time.Second)
	store.clusterEventMu.Lock()
	store.lastRetainedEventScanSuccess = fastScanBaseline
	store.clusterEventMu.Unlock()
	fastWindow := store.clusterEventWindow.Load()
	require.NoError(t, store.reconcileClusterEvents(t.Context()))

	clusters, err := store.LookupCluster(IndexClusterID, "late")
	require.NoError(t, err)
	require.Empty(t, clusters, "fast cursor scan must not see an Event below its lower bound")
	require.Equal(t, fastScanBaseline, snapshotLastRetainedEventScanSuccess(store))
	require.Same(t, fastWindow, store.clusterEventWindow.Load())

	fullScanBaseline := time.Now().Add(-clusterRetainedEventScanInterval)
	store.clusterEventMu.Lock()
	store.lastRetainedEventScanSuccess = fullScanBaseline
	store.clusterEventMu.Unlock()
	require.NoError(t, store.reconcileClusterEvents(t.Context()))

	clusters, err = store.LookupCluster(IndexClusterID, "late")
	require.NoError(t, err)
	require.Len(t, clusters, 1)
	require.Equal(t, uint64(100), snapshotLastClusterEventID(store))
	require.Equal(t, []ClusterEvent{
		testClusterEvent(50, "late", inframysql.EventTypeClusterCreated, 1),
		testClusterEvent(100, "current", inframysql.EventTypeClusterCreated, 1),
	}, snapshotLastClusterEvents(store))
	require.NotSame(t, fastWindow, store.clusterEventWindow.Load())
	window, err := store.GetClusterEventWindow()
	require.NoError(t, err)
	require.Equal(t, []ClusterEvent{
		testClusterEvent(50, "late", inframysql.EventTypeClusterCreated, 1),
		testClusterEvent(100, "current", inframysql.EventTypeClusterCreated, 1),
	}, window.Events)
	require.Zero(t, snapshotClusterEventRewindLowerBound(store))
	require.True(t, snapshotLastRetainedEventScanSuccess(store).After(fullScanBaseline))
}

func TestSyncClusterEventsReloadsWhenRetainedSafetyScanExpires(t *testing.T) {
	db := openTestClusterDB(t)
	insertTestCluster(t, db, testCluster("cluster", 1))
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         1,
		Name:            "cluster",
		Namespace:       "default",
		Kubernetes:      "kube-a",
		EventType:       inframysql.EventTypeClusterCreated,
		Status:          inframysql.StatusDone,
		ResourceVersion: 1,
	})

	store := newTestClusterStore(t, db)
	require.NoError(t, store.reloadAllClusterState(t.Context()))
	epochBeforeGap := requireClusterEventEpoch(t, store)
	updateTestClusterResourceVersion(t, db, "cluster", 2)

	expiredSafetyScan := time.Now().Add(-serverlessinfra.DefaultClusterEventTTL - time.Second)
	store.clusterEventMu.Lock()
	store.lastClusterEventPollSuccess = time.Now()
	store.lastRetainedEventScanSuccess = expiredSafetyScan
	store.clusterEventRewindLowerBound = 1
	store.clusterEventMu.Unlock()
	gapReloadsBefore := testutil.ToFloat64(metametrics.ClusterEventGapReloadsTotal)
	reloadsBefore := testutil.ToFloat64(metametrics.ClusterSyncOperationsTotal.WithLabelValues(
		metametrics.ClusterSyncOperationReload,
		metametrics.ResultSuccess,
	))

	require.NoError(t, store.syncClusterEvents(t.Context()))

	clusters, err := store.LookupCluster(IndexClusterID, "cluster")
	require.NoError(t, err)
	require.Len(t, clusters, 1)
	require.Equal(t, uint64(2), clusters[0].Status.ResourceVersion)
	require.True(t, snapshotLastRetainedEventScanSuccess(store).After(expiredSafetyScan))
	require.Equal(t, gapReloadsBefore+1, testutil.ToFloat64(metametrics.ClusterEventGapReloadsTotal))
	require.Equal(t, reloadsBefore+1, testutil.ToFloat64(metametrics.ClusterSyncOperationsTotal.WithLabelValues(
		metametrics.ClusterSyncOperationReload,
		metametrics.ResultSuccess,
	)))
	epochAfterGap := requireClusterEventEpoch(t, store)
	require.NotEqual(t, epochBeforeGap, epochAfterGap)
	require.Zero(t, snapshotClusterEventRewindLowerBound(store))
}

func TestSyncClusterEventsFailedGapReloadKeepsEpoch(t *testing.T) {
	db := openTestClusterDB(t)
	insertTestCluster(t, db, testCluster("cluster", 1))
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         1,
		Name:            "cluster",
		Namespace:       "default",
		Kubernetes:      "kube-a",
		EventType:       inframysql.EventTypeClusterCreated,
		Status:          inframysql.StatusDone,
		ResourceVersion: 1,
	})

	store := newTestClusterStore(t, db)
	require.NoError(t, store.reloadAllClusterState(t.Context()))
	epochBeforeGap := requireClusterEventEpoch(t, store)

	store.clusterEventMu.Lock()
	store.lastRetainedEventScanSuccess = time.Now().Add(-serverlessinfra.DefaultClusterEventTTL - time.Second)
	store.clusterEventMu.Unlock()

	canceledCtx, cancel := context.WithCancel(t.Context())
	cancel()
	require.Error(t, store.syncClusterEvents(canceledCtx))
	require.Equal(t, epochBeforeGap, requireClusterEventEpoch(t, store))
}

func TestSyncClusterEventsReloadsAfterRetainedSafetyScanCrossesDeadline(t *testing.T) {
	db := openTestClusterDB(t)
	insertTestCluster(t, db, testCluster("cluster", 1))
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         1,
		Name:            "cluster",
		Namespace:       "default",
		Kubernetes:      "kube-a",
		EventType:       inframysql.EventTypeClusterCreated,
		Status:          inframysql.StatusDone,
		ResourceVersion: 1,
	})

	store := newTestClusterStore(t, db)
	require.NoError(t, store.reloadAllClusterState(t.Context()))
	updateTestClusterResourceVersion(t, db, "cluster", 2)
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         2,
		Name:            "cluster",
		Namespace:       "default",
		Kubernetes:      "kube-a",
		EventType:       inframysql.EventTypeClusterUpdated,
		Status:          inframysql.StatusDone,
		ResourceVersion: 2,
	})

	lastRetainedScanSuccess := time.Now().Add(-serverlessinfra.DefaultClusterEventTTL + 200*time.Millisecond)
	store.clusterEventMu.Lock()
	store.lastClusterEventPollSuccess = time.Now()
	store.lastRetainedEventScanSuccess = lastRetainedScanSuccess
	store.clusterEventMu.Unlock()
	reconcileFailuresBefore := testutil.ToFloat64(metametrics.ClusterSyncOperationsTotal.WithLabelValues(
		metametrics.ClusterSyncOperationReconcile,
		metametrics.ResultFailure,
	))
	gapReloadsBefore := testutil.ToFloat64(metametrics.ClusterEventGapReloadsTotal)
	reloadsBefore := testutil.ToFloat64(metametrics.ClusterSyncOperationsTotal.WithLabelValues(
		metametrics.ClusterSyncOperationReload,
		metametrics.ResultSuccess,
	))

	queryStarted, releaseQuery := blockFirstClusterQuery(t, db)
	errCh := make(chan error, 1)
	go func() {
		errCh <- store.syncClusterEvents(t.Context())
	}()

	select {
	case <-queryStarted:
	case <-time.After(2 * time.Second):
		t.Fatal("timed out waiting for retained cluster Event scan")
	}
	deadline := lastRetainedScanSuccess.Add(serverlessinfra.DefaultClusterEventTTL)
	time.Sleep(max(time.Until(deadline), 0) + 10*time.Millisecond)
	close(releaseQuery)
	require.NoError(t, <-errCh)

	clusters, err := store.LookupCluster(IndexClusterID, "cluster")
	require.NoError(t, err)
	require.Len(t, clusters, 1)
	require.Equal(t, uint64(2), clusters[0].Status.ResourceVersion)
	require.Equal(t, reconcileFailuresBefore+1,
		testutil.ToFloat64(metametrics.ClusterSyncOperationsTotal.WithLabelValues(
			metametrics.ClusterSyncOperationReconcile,
			metametrics.ResultFailure,
		)))
	require.Equal(t, gapReloadsBefore+1, testutil.ToFloat64(metametrics.ClusterEventGapReloadsTotal))
	require.Equal(t, reloadsBefore+1, testutil.ToFloat64(metametrics.ClusterSyncOperationsTotal.WithLabelValues(
		metametrics.ClusterSyncOperationReload,
		metametrics.ResultSuccess,
	)))
	require.True(t, snapshotLastRetainedEventScanSuccess(store).After(lastRetainedScanSuccess))
}

func TestSyncClusterEventsReloadsAllStateAfterSafetyScanTTLGap(t *testing.T) {
	db := openTestClusterDB(t)
	insertTestCluster(t, db, testCluster("updated", 1))
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         1,
		Name:            "updated",
		Namespace:       "default",
		Kubernetes:      "kube-a",
		EventType:       inframysql.EventTypeClusterCreated,
		Status:          inframysql.StatusDone,
		ResourceVersion: 1,
	})

	store := newTestClusterStore(t, db)
	require.NoError(t, store.reloadAllClusterState(t.Context()))

	updateTestClusterResourceVersion(t, db, "updated", 2)
	insertTestEvent(t, db, &inframysql.ClusterEvent{
		EventID:         2,
		Name:            "updated",
		Namespace:       "default",
		Kubernetes:      "kube-a",
		EventType:       inframysql.EventTypeClusterUpdated,
		Status:          inframysql.StatusDone,
		ResourceVersion: 2,
	})
	require.NoError(t, db.Exec("DELETE FROM cluster_events WHERE event_id = ?", 2).Error)

	gapReloadsBefore := testutil.ToFloat64(metametrics.ClusterEventGapReloadsTotal)
	gapCountBefore, gapSumBefore := clusterEventPollGapMetric(t)
	reloadsBefore := testutil.ToFloat64(metametrics.ClusterSyncOperationsTotal.WithLabelValues(
		metametrics.ClusterSyncOperationReload,
		metametrics.ResultSuccess,
	))
	store.clusterEventMu.Lock()
	expiredPollSuccess := time.Now().Add(-serverlessinfra.DefaultClusterEventTTL - time.Second)
	store.lastClusterEventPollSuccess = expiredPollSuccess
	store.lastRetainedEventScanSuccess = expiredPollSuccess
	store.clusterEventMu.Unlock()

	require.NoError(t, store.syncClusterEvents(t.Context()))

	clusters, err := store.LookupCluster(IndexClusterID, "updated")
	require.NoError(t, err)
	require.Len(t, clusters, 1)
	require.Equal(t, uint64(2), clusters[0].Status.ResourceVersion)

	store.clusterEventMu.Lock()
	require.True(t, store.lastClusterEventPollSuccess.After(expiredPollSuccess))
	store.clusterEventMu.Unlock()
	require.Equal(t, []ClusterEvent{
		testClusterEvent(1, "updated", inframysql.EventTypeClusterCreated, 1),
	}, snapshotLastClusterEvents(store))
	require.Equal(t, uint64(1), snapshotLastClusterEventID(store))
	require.Equal(t, float64(1), testutil.ToFloat64(metametrics.ClusterCacheEntries))
	require.Equal(t, float64(1), testutil.ToFloat64(metametrics.ClusterEventSnapshotEntries))
	require.Positive(t, testutil.ToFloat64(metametrics.ClusterEventPollLastSuccessTimestamp))
	require.Equal(t, gapReloadsBefore+1, testutil.ToFloat64(metametrics.ClusterEventGapReloadsTotal))
	gapCountAfter, gapSumAfter := clusterEventPollGapMetric(t)
	require.Equal(t, gapCountBefore+1, gapCountAfter)
	require.GreaterOrEqual(t, gapSumAfter-gapSumBefore, serverlessinfra.DefaultClusterEventTTL.Seconds())
	require.Equal(t, reloadsBefore+1, testutil.ToFloat64(metametrics.ClusterSyncOperationsTotal.WithLabelValues(
		metametrics.ClusterSyncOperationReload,
		metametrics.ResultSuccess,
	)))
}

func TestClusterEventPollSuccessUsesOperationStartTime(t *testing.T) {
	tests := []struct {
		name       string
		initialize bool
		run        func(*memDBStorage, context.Context) error
	}{
		{
			name: "reload",
			run: func(store *memDBStorage, ctx context.Context) error {
				return store.reloadAllClusterState(ctx)
			},
		},
		{
			name:       "reconcile",
			initialize: true,
			run: func(store *memDBStorage, ctx context.Context) error {
				return store.reconcileClusterEvents(ctx)
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			db := openTestClusterDB(t)
			insertTestCluster(t, db, testCluster("cluster-1", 1))
			insertTestEvent(t, db, &inframysql.ClusterEvent{
				EventID:    1,
				Name:       "cluster-1",
				Namespace:  "default",
				Kubernetes: "kube-a",
			})
			store := newTestClusterStore(t, db)
			if tt.initialize {
				require.NoError(t, store.reloadAllClusterState(t.Context()))
			}

			queryStarted := make(chan struct{})
			releaseQuery := make(chan struct{})
			var blockOnce sync.Once
			blockQuery := func(*gorm.DB) {
				blockOnce.Do(func() {
					close(queryStarted)
					<-releaseQuery
				})
			}
			require.NoError(t, db.Callback().Query().Before("gorm:query").Register("test:block_query", blockQuery))
			require.NoError(t, db.Callback().Row().Before("gorm:row").Register("test:block_row", blockQuery))

			errCh := make(chan error, 1)
			go func() {
				errCh <- tt.run(store, t.Context())
			}()

			select {
			case <-queryStarted:
			case <-time.After(2 * time.Second):
				t.Fatal("timed out waiting for cluster query")
			}
			queryReleasedAt := time.Now()
			close(releaseQuery)
			require.NoError(t, <-errCh)

			store.clusterEventMu.Lock()
			lastSuccess := store.lastClusterEventPollSuccess
			store.clusterEventMu.Unlock()
			require.True(t, lastSuccess.Before(queryReleasedAt),
				"poll success must use the operation start time, got %s after query release %s",
				lastSuccess, queryReleasedAt)
		})
	}
}

func TestClusterEventPollQuery(t *testing.T) {
	db := openTestClusterDB(t).Session(&gorm.Session{DryRun: true})

	var events []ClusterEvent
	query := clusterEventPollQuery(db, "kube-a", 5).Find(&events)
	require.NoError(t, query.Error)
	pollSQL := query.Statement.SQL.String()
	require.Contains(t, pollSQL, "SELECT event_id, name, event_type, resource_version")
	require.Contains(t, pollSQL, "WHERE kubernetes = ? AND event_id > ?")
	require.Contains(t, pollSQL, "ORDER BY event_id ASC")
	require.Contains(t, pollSQL, "LIMIT")
	require.NotContains(t, pollSQL, "status")
	require.NotContains(t, pollSQL, "namespace")
	require.NotContains(t, pollSQL, "MAX(")
	require.NotContains(t, pollSQL, "GROUP BY")
	require.NotContains(t, pollSQL, "JOIN")
	require.Equal(t, []any{"kube-a", uint64(5)}, query.Statement.Vars)
}

func TestLatestClusterEventPageQuery(t *testing.T) {
	db := openTestClusterDB(t).Session(&gorm.Session{DryRun: true})

	var events []ClusterEvent
	query := latestClusterEventPageQuery(db, "kube-a", "cluster-005").Find(&events)
	require.NoError(t, query.Error)
	windowSQL := query.Statement.SQL.String()
	require.Contains(t, windowSQL,
		"SELECT cluster_event.event_id, cluster_event.name, cluster_event.event_type, cluster_event.resource_version")
	require.Contains(t, windowSQL, "MAX(event_id) AS event_id")
	require.Contains(t, windowSQL, "name <> ''")
	require.Contains(t, windowSQL, "kubernetes = ?")
	require.Contains(t, windowSQL, "name > ?")
	require.Contains(t, windowSQL, "GROUP BY `name`")
	require.Contains(t, windowSQL, "ORDER BY name ASC")
	require.Contains(t, windowSQL, "LIMIT")
	require.Contains(t, windowSQL, "JOIN")
	require.Contains(t, windowSQL, "ORDER BY latest.name ASC")
	require.NotContains(t, windowSQL, "latest.event_id >")
	require.NotContains(t, windowSQL, "status")
	require.NotContains(t, windowSQL, "namespace")
	require.Equal(t, []any{"kube-a", "cluster-005"}, query.Statement.Vars)
}

func openTestClusterDB(t *testing.T) *gorm.DB {
	t.Helper()

	db, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "cluster.db")), &gorm.Config{})
	require.NoError(t, err)
	require.NoError(t, db.Exec(`
		CREATE TABLE clusters (
			cluster_id TEXT PRIMARY KEY,
			namespace TEXT,
			kubernetes TEXT
		);
	`).Error)
	require.NoError(t, db.Exec(`
		CREATE TABLE cluster_specs (
			cluster_id TEXT PRIMARY KEY
		);
	`).Error)
	require.NoError(t, db.Exec(`
		CREATE TABLE cluster_statuses (
			cluster_id TEXT PRIMARY KEY,
			resource_version INTEGER NOT NULL
		);
	`).Error)
	require.NoError(t, db.Exec(`
		CREATE TABLE cluster_events (
			event_id INTEGER PRIMARY KEY,
			namespace TEXT,
			name TEXT,
			kubernetes TEXT,
			event_type TEXT,
			status TEXT,
			resource_version INTEGER NOT NULL
		);
	`).Error)
	return db
}

func insertTestCluster(t *testing.T, db *gorm.DB, cluster *inframysql.Cluster) {
	t.Helper()

	require.NoError(t, db.Exec(
		"INSERT INTO clusters (cluster_id, namespace, kubernetes) VALUES (?, ?, ?)",
		cluster.ClusterID,
		cluster.Namespace,
		cluster.Kubernetes,
	).Error)
	require.NoError(t, db.Exec(
		"INSERT INTO cluster_specs (cluster_id) VALUES (?)",
		cluster.ClusterID,
	).Error)
	require.NoError(t, db.Exec(
		"INSERT INTO cluster_statuses (cluster_id, resource_version) VALUES (?, ?)",
		cluster.ClusterID,
		cluster.Status.ResourceVersion,
	).Error)
}

func updateTestClusterResourceVersion(t *testing.T, db *gorm.DB, clusterID string, resourceVersion uint64) {
	t.Helper()

	require.NoError(t, db.Exec(
		"UPDATE cluster_statuses SET resource_version = ? WHERE cluster_id = ?",
		resourceVersion,
		clusterID,
	).Error)
}

func deleteTestCluster(t *testing.T, db *gorm.DB, clusterID string) {
	t.Helper()

	require.NoError(t, db.Exec("DELETE FROM cluster_statuses WHERE cluster_id = ?", clusterID).Error)
	require.NoError(t, db.Exec("DELETE FROM cluster_specs WHERE cluster_id = ?", clusterID).Error)
	require.NoError(t, db.Exec("DELETE FROM clusters WHERE cluster_id = ?", clusterID).Error)
}

func insertTestEvent(t *testing.T, db *gorm.DB, event *inframysql.ClusterEvent) {
	t.Helper()

	status := event.Status
	if status == "" {
		status = inframysql.StatusPending
	}
	require.NoError(t, db.Exec(
		`INSERT INTO cluster_events
			(event_id, namespace, name, kubernetes, event_type, status, resource_version)
			VALUES (?, ?, ?, ?, ?, ?, ?)`,
		event.EventID,
		event.Namespace,
		event.Name,
		event.Kubernetes,
		event.EventType,
		status,
		event.ResourceVersion,
	).Error)
}

func snapshotLastClusterEvents(store *memDBStorage) []ClusterEvent {
	store.clusterEventMu.RLock()
	defer store.clusterEventMu.RUnlock()

	return sortedClusterEvents(store.lastClusterEvents)
}

func snapshotLastClusterEventMap(store *memDBStorage) map[string]ClusterEvent {
	store.clusterEventMu.RLock()
	defer store.clusterEventMu.RUnlock()

	events := make(map[string]ClusterEvent, len(store.lastClusterEvents))
	for name, event := range store.lastClusterEvents {
		events[name] = event
	}
	return events
}

func snapshotLastClusterEventID(store *memDBStorage) uint64 {
	store.clusterEventMu.RLock()
	defer store.clusterEventMu.RUnlock()
	return store.lastClusterEventID
}

func snapshotLastRetainedEventScanSuccess(store *memDBStorage) time.Time {
	store.clusterEventMu.RLock()
	defer store.clusterEventMu.RUnlock()
	return store.lastRetainedEventScanSuccess
}

func snapshotClusterEventRewindLowerBound(store *memDBStorage) uint64 {
	store.clusterEventMu.RLock()
	defer store.clusterEventMu.RUnlock()
	return store.clusterEventRewindLowerBound
}

func requireClusterEventEpoch(t *testing.T, store *memDBStorage) string {
	t.Helper()

	store.clusterEventMu.RLock()
	defer store.clusterEventMu.RUnlock()
	epoch := store.clusterEventEpoch
	require.NotEmpty(t, epoch)
	return epoch
}

func clusterEventPollGapMetric(t *testing.T) (uint64, float64) {
	t.Helper()

	var metric dto.Metric
	require.NoError(t, metametrics.ClusterEventPollGap.Write(&metric))
	return metric.GetHistogram().GetSampleCount(), metric.GetHistogram().GetSampleSum()
}

func clusterFullLoadDurationMetric(t *testing.T, result string) (uint64, float64) {
	t.Helper()

	var metric dto.Metric
	collector, ok := metametrics.ClusterFullLoadDuration.WithLabelValues(result).(prometheus.Metric)
	require.True(t, ok)
	require.NoError(t, collector.Write(&metric))
	return metric.GetHistogram().GetSampleCount(), metric.GetHistogram().GetSampleSum()
}

func successfulClusterEventFullScanMetric(
	t *testing.T,
	metricVec *prometheus.HistogramVec,
) (uint64, float64) {
	t.Helper()
	return clusterEventFullScanMetric(t, metricVec, metametrics.ResultSuccess)
}

func clusterEventFullScanMetric(
	t *testing.T,
	metricVec *prometheus.HistogramVec,
	result string,
) (uint64, float64) {
	t.Helper()

	var metric dto.Metric
	collector, ok := metricVec.WithLabelValues(result).(prometheus.Metric)
	require.True(t, ok)
	require.NoError(t, collector.Write(&metric))
	return metric.GetHistogram().GetSampleCount(), metric.GetHistogram().GetSampleSum()
}

func resetClusterEventFullScanMetrics(t *testing.T) {
	t.Helper()

	results := []string{metametrics.ResultSuccess, metametrics.ResultFailure}
	reset := func() {
		for _, result := range results {
			metametrics.ClusterEventFullScanDuration.DeleteLabelValues(result)
			metametrics.ClusterEventFullScanReturnedRows.DeleteLabelValues(result)
			metametrics.ClusterEventFullScanPages.DeleteLabelValues(result)
		}
	}
	reset()
	t.Cleanup(reset)
}

func blockFirstClusterQuery(t *testing.T, db *gorm.DB) (<-chan struct{}, chan<- struct{}) {
	t.Helper()

	queryStarted := make(chan struct{})
	releaseQuery := make(chan struct{})
	var blockOnce sync.Once
	blockQuery := func(*gorm.DB) {
		blockOnce.Do(func() {
			close(queryStarted)
			<-releaseQuery
		})
	}
	require.NoError(t, db.Callback().Query().Before("gorm:query").
		Register("test:block_first_cluster_query", blockQuery))
	return queryStarted, releaseQuery
}

func testClusterEvent(eventID uint64, name, eventType string, resourceVersion uint64) ClusterEvent {
	return ClusterEvent{
		EventID:         eventID,
		Name:            name,
		EventType:       eventType,
		ResourceVersion: resourceVersion,
	}
}

func testCluster(name string, resourceVersion uint64) *inframysql.Cluster {
	return &inframysql.Cluster{
		Namespace:  "default",
		ClusterID:  name,
		Kubernetes: "kube-a",
		Spec: inframysql.ClusterSpec{
			ClusterID: name,
		},
		Status: inframysql.ClusterStatus{
			ClusterID:       name,
			ResourceVersion: resourceVersion,
		},
	}
}

func newTestClusterStore(t *testing.T, db *gorm.DB) *memDBStorage {
	t.Helper()

	clusterDB, err := memdb.NewMemDB(clusterMetaTableSchema())
	require.NoError(t, err)
	store := &memDBStorage{
		mysqlDB:           db,
		clusterMemDB:      clusterDB,
		kubeClusterName:   "kube-a",
		clusterEventEpoch: "test-epoch",
		lastClusterEvents: make(map[string]ClusterEvent),
	}
	store.publishClusterEventWindow(nil)
	return store
}