package storage

import (
	"testing"

	"github.com/hashicorp/go-memdb"
	"github.com/pingcap/kvproto/pkg/keyspacepb"
	"github.com/stretchr/testify/require"
)

func TestListKeyspacesPaginatedFiltersState(t *testing.T) {
	t.Parallel()

	db, err := memdb.NewMemDB(keyspaceMetaTableSchema())
	require.NoError(t, err)
	store := &memDBStorage{keyspaceMemDB: db}

	txn := db.Txn(true)
	for _, meta := range []*KeyspaceMetaData{
		{KeyspaceID: "1", KeyspaceName: "archived-old", State: keyspacepb.KeyspaceState_ARCHIVED, StateChangedAt: 100},
		{KeyspaceID: "2", KeyspaceName: "archived-recent", State: keyspacepb.KeyspaceState_ARCHIVED, StateChangedAt: 300},
		{KeyspaceID: "3", KeyspaceName: "archived-missing-time", State: keyspacepb.KeyspaceState_ARCHIVED},
		{KeyspaceID: "4", KeyspaceName: "tombstone-old", State: keyspacepb.KeyspaceState_TOMBSTONE, StateChangedAt: 100},
	} {
		require.NoError(t, txn.Insert(keyspaceMetaTableName, meta))
	}
	txn.Commit()

	filter := KeyspaceListFilter{
		HasState: true,
		State:    keyspacepb.KeyspaceState_ARCHIVED,
	}
	items, total, err := store.ListKeyspacesPaginated(filter, 10, 0)
	require.NoError(t, err)
	require.Equal(t, 3, total)
	require.Len(t, items, 3)
	require.Equal(t, keyspacepb.KeyspaceState_ARCHIVED, items[0].State)
	require.Equal(t, keyspacepb.KeyspaceState_ARCHIVED, items[1].State)
	require.Equal(t, keyspacepb.KeyspaceState_ARCHIVED, items[2].State)
}

func TestListKeyspacesPaginatedAppliesFilteredPagination(t *testing.T) {
	t.Parallel()

	db, err := memdb.NewMemDB(keyspaceMetaTableSchema())
	require.NoError(t, err)
	store := &memDBStorage{keyspaceMemDB: db}

	txn := db.Txn(true)
	for _, meta := range []*KeyspaceMetaData{
		{KeyspaceID: "1", KeyspaceName: "archived-1", State: keyspacepb.KeyspaceState_ARCHIVED, StateChangedAt: 100},
		{KeyspaceID: "2", KeyspaceName: "archived-2", State: keyspacepb.KeyspaceState_ARCHIVED, StateChangedAt: 200},
		{KeyspaceID: "3", KeyspaceName: "archived-3", State: keyspacepb.KeyspaceState_ARCHIVED, StateChangedAt: 300},
	} {
		require.NoError(t, txn.Insert(keyspaceMetaTableName, meta))
	}
	txn.Commit()

	filter := KeyspaceListFilter{
		HasState: true,
		State:    keyspacepb.KeyspaceState_ARCHIVED,
	}
	items, total, err := store.ListKeyspacesPaginated(filter, 1, 1)
	require.NoError(t, err)
	require.Equal(t, 3, total)
	require.Len(t, items, 1)
	require.Equal(t, "archived-2", items[0].KeyspaceName)
}

func TestListKeyspacesUsesStateFilterForEnabledKeyspaces(t *testing.T) {
	t.Parallel()

	db, err := memdb.NewMemDB(keyspaceMetaTableSchema())
	require.NoError(t, err)
	store := &memDBStorage{keyspaceMemDB: db}

	txn := db.Txn(true)
	for _, meta := range []*KeyspaceMetaData{
		{KeyspaceID: "1", KeyspaceName: "enabled", State: keyspacepb.KeyspaceState_ENABLED},
		{KeyspaceID: "2", KeyspaceName: "archived", State: keyspacepb.KeyspaceState_ARCHIVED},
	} {
		require.NoError(t, txn.Insert(keyspaceMetaTableName, meta))
	}
	txn.Commit()

	filter := KeyspaceListFilter{
		HasState: true,
		State:    keyspacepb.KeyspaceState_ENABLED,
	}
	items, err := store.ListKeyspaces(filter)
	require.NoError(t, err)
	require.Len(t, items, 1)
	require.Equal(t, "enabled", items[0].KeyspaceName)
}