package query

import (
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/aws/aws-sdk-go-v2/service/s3/types"
	"github.com/gin-gonic/gin"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

func TestBuildHiveListPrefix(t *testing.T) {
	fields := logsDivePath()
	rel, err := buildHiveListPrefix(fields[:3], Scope{
		"env":         ScopeEq("prod"),
		"deploy_type": ScopeEq("dedicated"),
		"date":        ScopeEq("2026-07-17"),
	})
	require.NoError(t, err)
	assert.Equal(t, "env=prod/deployType=dedicated/date=2026-07-17", rel)

	_, err = buildHiveListPrefix(fields[:3], Scope{"env": ScopeEq("prod")})
	require.Error(t, err)
	assert.Contains(t, err.Error(), "deploy_type")
}

func TestScopeFieldMetasOrdered(t *testing.T) {
	logs := scopeFieldMetas(logsDivePath())
	require.Len(t, logs, 4)
	assert.Equal(t, []string{"env", "deploy_type", "date", "components"},
		[]string{logs[0].Name, logs[1].Name, logs[2].Name, logs[3].Name})
	assert.Equal(t, ProjectionInjected, logs[3].Projection)
	assert.Equal(t, "component", logs[3].PathKey)

	stats := scopeFieldMetas(logStatsDivePath())
	require.Len(t, stats, 3)
	assert.Equal(t, "date", stats[0].Name)
	assert.Equal(t, "org", stats[1].Name)
	assert.Equal(t, "cluster", stats[2].Name)

	topsql := scopeFieldMetas(topsqlDivePath())
	require.Len(t, topsql, 5)
	assert.Equal(t, []string{"org", "cluster", "component", "instance", "date"},
		[]string{topsql[0].Name, topsql[1].Name, topsql[2].Name, topsql[3].Name, topsql[4].Name})

	stmt := scopeFieldMetas(sqlstatementDivePath())
	require.Len(t, stmt, 3)
	assert.Equal(t, "org", stmt[0].Name)
	assert.Equal(t, "date", stmt[2].Name)

	slow := scopeFieldMetas(slowlogsDivePath())
	require.Len(t, slow, 3)
	assert.Equal(t, "org", slow[0].Name)
	assert.Equal(t, "cluster", slow[1].Name)
	assert.Equal(t, "date", slow[2].Name)
	assert.True(t, slow[0].BarePath)
	assert.True(t, slow[1].BarePath)
	assert.False(t, slow[2].BarePath)
	assert.Equal(t, ProjectionInteger, slow[2].Projection)
}

func TestBuildHiveListPrefixSlowlogsBarePath(t *testing.T) {
	fields := slowlogsDivePath()
	rel, err := buildHiveListPrefix(fields[:1], Scope{"org": ScopeEq("1372813089188151296")})
	require.NoError(t, err)
	assert.Equal(t, "1372813089188151296", rel)

	rel, err = buildHiveListPrefix(fields[:2], Scope{
		"org":     ScopeEq("1372813089188151296"),
		"cluster": ScopeEq("019b23f6-ffc9-731d-be49-eb8f9e651855"),
	})
	require.NoError(t, err)
	assert.Equal(t, "1372813089188151296/019b23f6-ffc9-731d-be49-eb8f9e651855/slowlogs", rel)

	rel, err = buildHiveListPrefix(fields, Scope{
		"org":     ScopeEq("1372813089188151296"),
		"cluster": ScopeEq("019b23f6-ffc9-731d-be49-eb8f9e651855"),
		"date":    ScopeEq("2026-07-27"),
	})
	require.NoError(t, err)
	assert.Equal(t, "1372813089188151296/019b23f6-ffc9-731d-be49-eb8f9e651855/slowlogs/date=20260727", rel)
}

func TestParseScopeListSegmentBare(t *testing.T) {
	field := slowlogsDivePath()[0]
	v, ok := parseScopeListSegment(field, "1372813089188151296")
	require.True(t, ok)
	assert.Equal(t, "1372813089188151296", v)

	_, ok = parseScopeListSegment(field, "org=1372")
	assert.False(t, ok)
	_, ok = parseScopeListSegment(field, "global")
	assert.False(t, ok)
}

func TestBuildHiveListPrefixTopsqlPathSuffix(t *testing.T) {
	fields := topsqlDivePath()
	rel, err := buildHiveListPrefix(fields[:2], Scope{
		"org":     ScopeEq("o1"),
		"cluster": ScopeEq("c1"),
	})
	require.NoError(t, err)
	assert.Equal(t, "org=o1/cluster=c1/type=topsql", rel)

	rel, err = buildHiveListPrefix(fields[:3], Scope{
		"org": ScopeEq("o1"), "cluster": ScopeEq("c1"), "component": ScopeEq("tidb"),
	})
	require.NoError(t, err)
	assert.Equal(t, "org=o1/cluster=c1/type=topsql/component=tidb", rel)

	stmtFields := sqlstatementDivePath()
	rel, err = buildHiveListPrefix(stmtFields[:2], Scope{
		"org": ScopeEq("o1"), "cluster": ScopeEq("c1"),
	})
	require.NoError(t, err)
	assert.Equal(t, "org=o1/cluster=c1/type=sqlstatement/component=sqlstatement", rel)
}

type fakeS3List struct {
	lastIn *s3.ListObjectsV2Input
	out    *s3.ListObjectsV2Output
	err    error
}

func (f *fakeS3List) ListObjectsV2(_ context.Context, params *s3.ListObjectsV2Input, _ ...func(*s3.Options)) (*s3.ListObjectsV2Output, error) {
	f.lastIn = params
	if f.err != nil {
		return nil, f.err
	}
	return f.out, nil
}

func TestCatalogListScopeValues(t *testing.T) {
	fake := &fakeS3List{
		out: &s3.ListObjectsV2Output{
			CommonPrefixes: []types.CommonPrefix{
				{Prefix: aws.String("deltalake/global/logs/env=prod/deployType=dedicated/date=2026-07-17/component=pd/")},
				{Prefix: aws.String("deltalake/global/logs/env=prod/deployType=dedicated/date=2026-07-17/component=tidb/")},
				{Prefix: aws.String("deltalake/global/logs/env=prod/deployType=dedicated/date=2026-07-17/component=tikv/")},
			},
			IsTruncated: aws.Bool(false),
		},
	}
	cfg := Config{Vendor: "aws", Region: "us-west-2", LogsBucket: "o11y-prod-shared-us-west-2"}
	cat := &Catalog{
		cfg:    cfg,
		tables: resolveRegistryTables(cfg),
		s3:     fake,
	}
	res, err := cat.ListScopeValues(context.Background(), ListScopeRequest{
		Dataset: DatasetLogs,
		Field:   "components",
		Scope: Scope{
			"env":         ScopeEq("prod"),
			"deploy_type": ScopeEq("dedicated"),
			"date":        ScopeEq("2026-07-17"),
		},
	})
	require.NoError(t, err)
	assert.Equal(t, []string{"pd", "tidb", "tikv"}, res.Values)
	assert.Equal(t, "components", res.Field)
	assert.False(t, res.HasMore)
	require.NotNil(t, fake.lastIn)
	assert.Equal(t, "o11y-prod-shared-us-west-2", aws.ToString(fake.lastIn.Bucket))
	assert.Equal(t, "deltalake/global/logs/env=prod/deployType=dedicated/date=2026-07-17/", aws.ToString(fake.lastIn.Prefix))
	assert.Equal(t, "/", aws.ToString(fake.lastIn.Delimiter))
	assert.Equal(t, int32(1000), aws.ToInt32(fake.lastIn.MaxKeys))
}

func TestCatalogListScopeValuesMissingAncestor(t *testing.T) {
	cfg := Config{Vendor: "aws", Region: "us-west-2", LogsBucket: "b"}
	cat := &Catalog{
		cfg:    cfg,
		tables: resolveRegistryTables(cfg),
		s3:     &fakeS3List{out: &s3.ListObjectsV2Output{}},
	}
	_, err := cat.ListScopeValues(context.Background(), ListScopeRequest{
		Dataset: DatasetLogStats,
		Field:   "org",
		Scope:   Scope{},
	})
	require.Error(t, err)
	assert.Contains(t, err.Error(), "date")
}

func TestCatalogListScopeValuesFirstLevel(t *testing.T) {
	fake := &fakeS3List{
		out: &s3.ListObjectsV2Output{
			CommonPrefixes: []types.CommonPrefix{
				{Prefix: aws.String("deltalake/global/logs/env=dev/")},
				{Prefix: aws.String("deltalake/global/logs/env=prod/")},
			},
		},
	}
	cfg := Config{Vendor: "aws", Region: "us-west-2", LogsBucket: "o11y-prod-shared-us-west-2"}
	cat := &Catalog{cfg: cfg, tables: resolveRegistryTables(cfg), s3: fake}
	res, err := cat.ListScopeValues(context.Background(), ListScopeRequest{
		Dataset: DatasetLogs,
		Field:   "env",
	})
	require.NoError(t, err)
	assert.Equal(t, []string{"dev", "prod"}, res.Values)
	assert.Equal(t, "deltalake/global/logs/", aws.ToString(fake.lastIn.Prefix))
}

func TestHandleListScopeNoS3(t *testing.T) {
	gin.SetMode(gin.TestMode)
	s := newTestService(&fakeEngine{}, "us-east-1")
	r := gin.New()
	InitOlapApi(r.Group("/api/v1"), s)

	body, err := json.Marshal(map[string]any{
		"dataset": "logs",
		"field":   "env",
	})
	require.NoError(t, err)
	req := httptest.NewRequest(http.MethodPost, "/api/v1/olap/scope/list", strings.NewReader(string(body)))
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("x-provider", "aws")
	req.Header.Set("x-region", "us-east-1")
	req.Header.Set("x-cluster-id", "c1")
	w := httptest.NewRecorder()
	r.ServeHTTP(w, req)
	assert.Equal(t, http.StatusNotImplemented, w.Code)
}

func TestHandleListTablesIncludesOrderedScopeFields(t *testing.T) {
	gin.SetMode(gin.TestMode)
	s := newTestService(&fakeEngine{}, "us-east-1")
	r := gin.New()
	InitOlapApi(r.Group("/api/v1"), s)

	req := httptest.NewRequest(http.MethodGet, "/api/v1/olap/tables", nil)
	req.Header.Set("x-provider", "aws")
	req.Header.Set("x-region", "us-east-1")
	req.Header.Set("x-cluster-id", "cluster-1")
	w := httptest.NewRecorder()
	r.ServeHTTP(w, req)
	require.Equal(t, http.StatusOK, w.Code)
	body := w.Body.String()
	assert.Contains(t, body, `"name":"env"`)
	assert.Contains(t, body, `"projection":"injected"`)
	assert.Contains(t, body, `"path_key":"deployType"`)
}