package query

import (
	"context"
	"fmt"
	"strings"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/s3"
)

const (
	defaultScopeListMaxKeys = int32(1000)
	maxScopeListMaxKeys     = int32(1000)
)

// ListScopeRequest lists one Hive dive-path level under the given ancestor scope.
type ListScopeRequest struct {
	Dataset   string `json:"dataset"`
	Field     string `json:"field"`
	Scope     Scope  `json:"scope"`
	MaxKeys   *int   `json:"max_keys"`
	PageToken string `json:"page_token"`
}

// ListScopeResult is the paginated S3 delimiter listing for one scope field.
type ListScopeResult struct {
	Dataset       string   `json:"dataset"`
	Field         string   `json:"field"`
	Values        []string `json:"values"`
	HasMore       bool     `json:"has_more"`
	NextPageToken string   `json:"next_page_token,omitempty"`
}

// ListScopeValues lists distinct Hive partition values for field under ancestor scope.
// Uses S3 ListObjectsV2 with Delimiter="/" (lexicographic / Hive path order). Max 1000 per call.
func (c *Catalog) ListScopeValues(ctx context.Context, req ListScopeRequest) (*ListScopeResult, error) {
	if c == nil || c.s3 == nil {
		return nil, fmt.Errorf("scope listing requires AWS S3 (not available for this deployment)")
	}
	ds := normalizeDataset(req.Dataset)
	fields, err := divePathForDataset(ds)
	if err != nil {
		return nil, err
	}
	field, idx, ok := findDiveField(fields, req.Field)
	if !ok {
		return nil, fmt.Errorf("unknown scope field %q for dataset %q", req.Field, ds)
	}

	table, ok := c.findResolvedByDataset(ds)
	if !ok {
		return nil, fmt.Errorf("no registered table for dataset %q", ds)
	}

	bucket, rootKey, err := parseS3URI(strings.TrimSuffix(table.LocationPrefix, "/") + "/")
	if err != nil {
		return nil, err
	}
	rel, err := buildHiveListPrefix(fields[:idx], req.Scope)
	if err != nil {
		return nil, err
	}
	listPrefix := rootKey
	if !strings.HasSuffix(listPrefix, "/") {
		listPrefix += "/"
	}
	if rel != "" {
		listPrefix += strings.TrimSuffix(rel, "/") + "/"
	}

	maxKeys := defaultScopeListMaxKeys
	if req.MaxKeys != nil {
		if *req.MaxKeys <= 0 || int32(*req.MaxKeys) > maxScopeListMaxKeys {
			return nil, fmt.Errorf("max_keys must be 1..%d", maxScopeListMaxKeys)
		}
		maxKeys = int32(*req.MaxKeys)
	}

	in := &s3.ListObjectsV2Input{
		Bucket:    aws.String(bucket),
		Prefix:    aws.String(listPrefix),
		Delimiter: aws.String("/"),
		MaxKeys:   aws.Int32(maxKeys),
	}
	if tok := strings.TrimSpace(req.PageToken); tok != "" {
		in.ContinuationToken = aws.String(tok)
	}

	out, err := c.s3.ListObjectsV2(ctx, in)
	if err != nil {
		return nil, fmt.Errorf("s3 ListObjectsV2: %w", err)
	}

	values := make([]string, 0, len(out.CommonPrefixes))
	for _, cp := range out.CommonPrefixes {
		p := aws.ToString(cp.Prefix)
		rel := strings.TrimPrefix(p, listPrefix)
		rel = strings.TrimSuffix(rel, "/")
		if rel == "" || strings.Contains(rel, "/") {
			continue
		}
		val, ok := parseScopeListSegment(field, rel)
		if !ok {
			continue
		}
		values = append(values, val)
	}

	result := &ListScopeResult{
		Dataset: ds,
		Field:   field.Name,
		Values:  values,
		HasMore: aws.ToBool(out.IsTruncated),
	}
	if result.HasMore {
		result.NextPageToken = aws.ToString(out.NextContinuationToken)
	}
	return result, nil
}

func (c *Catalog) findResolvedByDataset(ds string) (*ResolvedTable, bool) {
	ds = normalizeDataset(ds)
	for i := range c.tables {
		if normalizeDataset(c.tables[i].Definition.Dataset) == ds {
			return &c.tables[i], true
		}
	}
	return nil, false
}

// buildHiveListPrefix returns the relative path (no leading slash) for ancestors.
func buildHiveListPrefix(ancestors []divePathField, scope Scope) (string, error) {
	if len(ancestors) == 0 {
		return "", nil
	}
	parts := make([]string, 0, len(ancestors)*2)
	for _, f := range ancestors {
		val, err := f.resolveEquality(scope)
		if err != nil {
			return "", fmt.Errorf("list field requires ancestor %s: %w", f.Name, err)
		}
		parts = append(parts, pathListSegment(f, val))
		if suf := strings.Trim(f.PathSuffix, "/"); suf != "" {
			parts = append(parts, suf)
		}
	}
	return strings.Join(parts, "/"), nil
}

// parseScopeListSegment extracts a scope value from one S3 common-prefix segment.
func parseScopeListSegment(field divePathField, rel string) (string, bool) {
	if field.BarePath || strings.TrimSpace(field.PathKey) == "" {
		// Bare segments: skip Hive-style key=value dirs (e.g. org=…, type=…) and known non-tenant roots.
		if strings.Contains(rel, "=") {
			return "", false
		}
		switch rel {
		case "global", "slowlogs", "topsql", "sqlstatement", "logs", "log-stats":
			return "", false
		}
		return rel, true
	}
	wantKey := field.PathKey + "="
	if !strings.HasPrefix(rel, wantKey) {
		return "", false
	}
	val := strings.TrimPrefix(rel, wantKey)
	if val == "" {
		return "", false
	}
	return val, true
}