package common

import (
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"strconv"
	"testing"
	"time"

	"github.com/pingcap/kvproto/pkg/keyspacepb"
	"github.com/stretchr/testify/require"
)

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

	keyspaces := []*metaServerKeyspace{
		{
			KeyspaceID:     "1",
			KeyspaceName:   "ks-1",
			State:          keyspacepb.KeyspaceState_ARCHIVED,
			CreatedAt:      1710000000,
			StateChangedAt: 1710000100,
			Config: map[string]string{
				"meta_service_group_id": "meta-service-group-1",
			},
		},
		{
			KeyspaceID:     "2",
			KeyspaceName:   "ks-2",
			State:          keyspacepb.KeyspaceState_ARCHIVED,
			CreatedAt:      1710000200,
			StateChangedAt: 1710000300,
		},
	}
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		require.Equal(t, metaServerListKeyspacesPath, r.URL.Path)
		require.Equal(t, "1000", r.URL.Query().Get("limit"))
		require.Equal(t, keyspacepb.KeyspaceState_ARCHIVED.String(), r.URL.Query().Get("state"))

		offset, err := strconv.Atoi(r.URL.Query().Get("offset"))
		require.NoError(t, err)
		require.LessOrEqual(t, offset, len(keyspaces))

		items := keyspaces[offset:]
		if len(items) > 1 {
			items = items[:1]
		}
		err = json.NewEncoder(w).Encode(metaServerListKeyspacesResponse{
			Items:  items,
			Total:  len(keyspaces),
			Limit:  metaServerListKeyspacesLimit,
			Offset: offset,
		})
		require.NoError(t, err)
	}))
	defer server.Close()

	got, err := GetKeyspaceByState(context.Background(), server.URL, keyspacepb.KeyspaceState_ARCHIVED)
	require.NoError(t, err)
	require.Len(t, got, 2)
	require.Equal(t, uint32(1), got[0].Id)
	require.Equal(t, "ks-1", got[0].Name)
	require.Equal(t, keyspacepb.KeyspaceState_ARCHIVED, got[0].State)
	require.Equal(t, int64(1710000000), got[0].CreatedAt)
	require.Equal(t, int64(1710000100), got[0].StateChangedAt)
	require.Equal(t, "meta-service-group-1", got[0].Config["meta_service_group_id"])
	require.Equal(t, uint32(2), got[1].Id)
	require.Equal(t, keyspacepb.KeyspaceState_ARCHIVED, got[1].State)
}

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

	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		require.Equal(t, keyspacepb.KeyspaceState_ARCHIVED.String(), r.URL.Query().Get("state"))
		err := json.NewEncoder(w).Encode(metaServerListKeyspacesResponse{
			Items: []*metaServerKeyspace{
				{
					KeyspaceID:   "1",
					KeyspaceName: "ks-1",
					State:        keyspacepb.KeyspaceState_ARCHIVED,
				},
			},
			Total: 1,
		})
		require.NoError(t, err)
	}))
	defer server.Close()

	records, err := GetKeyspace(context.Background(), server.URL, 24*time.Hour, keyspacepb.KeyspaceState_ARCHIVED, 0)
	require.NoError(t, err)
	require.Empty(t, records)
}

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

	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		require.Equal(t, keyspacepb.KeyspaceState_ARCHIVED.String(), r.URL.Query().Get("state"))
		err := json.NewEncoder(w).Encode(metaServerListKeyspacesResponse{
			Items: []*metaServerKeyspace{
				{
					KeyspaceID:     "1",
					KeyspaceName:   "ks-1",
					State:          keyspacepb.KeyspaceState_ENABLED,
					StateChangedAt: 1710000100,
				},
			},
			Total: 1,
		})
		require.NoError(t, err)
	}))
	defer server.Close()

	records, err := GetKeyspace(context.Background(), server.URL, 24*time.Hour, keyspacepb.KeyspaceState_ARCHIVED, 0)
	require.NoError(t, err)
	require.Empty(t, records)
}

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

	now := time.Now()
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		require.Equal(t, keyspacepb.KeyspaceState_ARCHIVED.String(), r.URL.Query().Get("state"))
		err := json.NewEncoder(w).Encode(metaServerListKeyspacesResponse{
			Items: []*metaServerKeyspace{
				{
					KeyspaceID:     "3",
					KeyspaceName:   "ks-3",
					State:          keyspacepb.KeyspaceState_ARCHIVED,
					StateChangedAt: now.Add(-36 * time.Hour).Unix(),
				},
				{
					KeyspaceID:     "1",
					KeyspaceName:   "ks-1",
					State:          keyspacepb.KeyspaceState_ARCHIVED,
					StateChangedAt: now.Add(-72 * time.Hour).Unix(),
				},
				{
					KeyspaceID:     "2",
					KeyspaceName:   "ks-2",
					State:          keyspacepb.KeyspaceState_ARCHIVED,
					StateChangedAt: now.Add(-48 * time.Hour).Unix(),
				},
				{
					KeyspaceID:     "4",
					KeyspaceName:   "ks-4",
					State:          keyspacepb.KeyspaceState_ARCHIVED,
					StateChangedAt: now.Add(-time.Hour).Unix(),
				},
			},
			Total: 4,
		})
		require.NoError(t, err)
	}))
	defer server.Close()

	records, err := GetKeyspace(context.Background(), server.URL, 24*time.Hour, keyspacepb.KeyspaceState_ARCHIVED, 2)
	require.NoError(t, err)
	require.Len(t, records, 2)
	require.Equal(t, uint32(1), records[0].KeyspaceId)
	require.Equal(t, uint32(2), records[1].KeyspaceId)
}