package topru

import (
	"bytes"
	"context"
	"fmt"
	"net/http"
	"os"
	"sort"
	"strconv"
	"strings"
	"text/template"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/pingcap/log"
	"github.com/tidbcloud/observability/o11ydiagnosis/pkg/storage"
	"github.com/tidbcloud/observability/o11ydiagnosis/pkg/topsql/planmeta"
	"github.com/tidbcloud/observability/o11ydiagnosis/pkg/topsqlv2"
	"go.uber.org/zap"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	perClusterQPSLimitDuckDB = 10.0
	perClusterLimiterB       = 10
	defaultTimeout           = time.Second * 180
)

// InitTopRUApi registers TopRU HTTP handlers.
func InitTopRUApi(g *gin.RouterGroup, s *Service) {
	g.GET("/topru/config", s.configHandler)
	g.GET("/topru/summary", s.summaryHandler)
	g.GET("/topru/summaryplan", s.summaryPlanHandler)
	g.GET("/topru/user_stats", s.userStatsHandler)
}

// Service implements TopRU API using Delta Lake table type=topsql_topru.
type Service struct {
	duckDBAdapter duckDBAdapter
	store         storage.Storage
}

type duckDBAdapter interface {
	SelectSlowlogInDLWithoutFilterPushdown(ctx context.Context, dest interface{}, tableAlias, table, query string) error
}

// NewService creates a TopRU service. Reuses topsqlv2 DuckDB adapter and rate limiting.
func NewService(cacheDir string, store storage.Storage, kubeCli client.Client, duckDBConfig topsqlv2.DuckDBConfig) *Service {
	adapter := topsqlv2.NewQueryDuckDBAdapter(duckDBConfig.Vendor, duckDBConfig.Region, store, kubeCli)
	rateLimited := newTopRURateLimitedQuery(adapter, perClusterQPSLimitDuckDB, perClusterLimiterB)
	return &Service{
		duckDBAdapter: rateLimited,
		store:         store,
	}
}

type topRURateLimitedQuery struct {
	inner *topsqlv2.DuckDBAdapter
}

func newTopRURateLimitedQuery(inner *topsqlv2.DuckDBAdapter, _ float64, _ int) *topRURateLimitedQuery {
	return &topRURateLimitedQuery{inner: inner}
}

func (q *topRURateLimitedQuery) SelectSlowlogInDLWithoutFilterPushdown(ctx context.Context, dest interface{}, tableAlias, table, query string) error {
	return q.inner.SelectSlowlogInDLWithoutFilterPushdown(ctx, dest, tableAlias, table, query)
}

type errorResponse struct {
	Error    bool   `json:"error"`
	Message  string `json:"message"`
	Code     string `json:"code"`
	FullText string `json:"full_text"`
}

type summaryEnvelope struct {
	Status string             `json:"status"`
	Data   []TopRUSummaryItem `json:"data"`
}

type summaryPlanEnvelope struct {
	Status string                 `json:"status"`
	Data   []TopRUSummaryPlanItem `json:"data"`
}

type userStatsEnvelope struct {
	Status string          `json:"status"`
	Data   []UserStatsItem `json:"data"`
}

type configResponse struct {
	Enable bool `json:"enable"`
}

// TopRU has no instance concept; path always uses instance=default.
// Table path: deltalake/org={tenantId}/cluster={clusterId}/type=topsql_topru/instance=default
func topRUTablePath(scheme, tenantID, clusterID string) string {
	return fmt.Sprintf("%s/deltalake/org=%s/cluster=%s/type=topsql/component=topru/instance=default", scheme, tenantID, clusterID)
}

// @Summary		Get config of TopRU
// @Tags			topru
// @Description	TopRU is always enabled when using Delta Lake data source
// @Produce		json
// @Success		200	{object}	configResponse
// @Router			/topru/config [get]
func (s *Service) configHandler(c *gin.Context) {
	c.JSON(http.StatusOK, configResponse{Enable: true})
}

// GetSummaryRequest for TopRU summary (aggregated by sql_digest only).
// User: nil if the "user" query key is omitted (no user filter). If the key is present, filter by its value;
// empty string means filter rows where user is empty or NULL.
type GetSummaryRequest struct {
	Start      string  `form:"start" binding:"required"`
	End        string  `form:"end" binding:"required"`
	Top        string  `form:"top" binding:"required"`
	OrderBy    string  `form:"order_by"` // ru, exec_count, duration; default ru
	User       *string `form:"-"`        // set from query: omit key = nil; present = filter (including "")
	DataSource string  `form:"data_source"`
	Step       int     `form:"step"` // bucket size in seconds; default 60 (1 minute)
}

func (q *GetSummaryRequest) Validate() error {
	if q.Start == "" {
		return fmt.Errorf("start not found")
	}
	if q.End == "" {
		return fmt.Errorf("end not found")
	}
	if q.Top == "" {
		return fmt.Errorf("top not found")
	}
	if q.Step == 0 {
		q.Step = 60
	}
	if q.OrderBy != "" && q.OrderBy != "ru" && q.OrderBy != "exec_count" && q.OrderBy != "duration" {
		return fmt.Errorf("invalid order_by: %s, must be one of: ru, exec_count, duration", q.OrderBy)
	}
	return nil
}

// GetSummaryPlanRequest is the digests+plans breakdown for one (or all) SQL digests.
type GetSummaryPlanRequest struct {
	GetSummaryRequest
	Digest string `form:"digest"` // optional; when set, only return plans for this sql_digest
}

// SummaryResponse for TopRU (by sql_digest).
type SummaryResponse struct {
	Data []TopRUSummaryItem `json:"data"`
}

// SummaryPlanResponse for TopRU (by sql_digest + plan_digest).
type SummaryPlanResponse struct {
	Data []TopRUSummaryPlanItem `json:"data"`
}

// TopRUSummaryItem one row per sql_digest (plans collapsed).
type TopRUSummaryItem struct {
	SQLDigest         string    `json:"sql_digest"`
	SQLText           string    `json:"sql_text,omitempty"`
	TotalRU           float64   `json:"total_ru"`
	ExecCount         int64     `json:"exec_count"`
	DurationNs        int64     `json:"duration_ns"`
	ExecCountPerSec   float64   `json:"exec_count_per_sec"`
	DurationPerExecMs float64   `json:"duration_per_exec_ms"`
	TimestampSec      []int64   `json:"timestamp_sec"`
	TotalRUByTime     []float64 `json:"total_ru_by_time,omitempty"`
}

// TopRUSummaryPlanItem one row per (sql_digest, plan_digest).
type TopRUSummaryPlanItem struct {
	SQLDigest         string    `json:"sql_digest"`
	PlanDigest        string    `json:"plan_digest"`
	SQLText           string    `json:"sql_text,omitempty"`
	PlanText          string    `json:"plan_text,omitempty"`
	TotalRU           float64   `json:"total_ru"`
	ExecCount         int64     `json:"exec_count"`
	DurationNs        int64     `json:"duration_ns"`
	ExecCountPerSec   float64   `json:"exec_count_per_sec"`
	DurationPerExecMs float64   `json:"duration_per_exec_ms"`
	TimestampSec      []int64   `json:"timestamp_sec"`
	TotalRUByTime     []float64 `json:"total_ru_by_time,omitempty"`
}

// topruQueryResult one row from DuckDB aggregation.
type topruQueryResult struct {
	SQLDigest       *string    `db:"sql_digest"`
	PlanDigest      *string    `db:"plan_digest"`
	TimeWindow      *time.Time `db:"time_window"`
	TotalRU         float64    `db:"total_ru"`
	TotalExecCount  *float64   `db:"total_exec_count"`
	TotalDurationNs *float64   `db:"total_duration_ns"`
}

// @Summary		Get TopRU summary (RU by SQL digest)
// @Tags			topru
// @Description	Aggregate ReadUnits (RU) grouped by sql_digest within the given time range. Plans are collapsed into one row per digest. Use /topru/summaryplan?digest=… for plan breakdown.
// @Produce		json
// @Param			X-Provider		header		string	true	"Cloud provider where RU data is stored (aws/gcp/azure)"
// @Param			X-Region		header		string	true	"Cloud region where RU data is stored"
// @Param			X-Org-Id		header		string	true	"Organization ID (x-org-id)"
// @Param			X-Project-Id	header		string	false	"Project ID (x-project-id; optional)"
// @Param			X-Cluster-Id	header		string	true	"Cluster ID (x-cluster-id)"
// @Param			X-Deploy-Type	header		string	true	"Deployment tier: dedicated/shared/nextgen/premium"
// @Param			start			query		string	true	"Start time in Unix seconds (inclusive)"
// @Param			end				query		string	true	"End time in Unix seconds (inclusive)"
// @Param			top				query		string	true	"Top N SQL digests to return"
// @Param			order_by		query		string	false	"Metric used to rank results: ru | exec_count | duration (default: ru)"
// @Param			user			query		string	false	"Omit to not filter by user. Pass empty (user=) to keep only rows with empty or NULL user; pass a name to filter by that user"
// @Param			data_source		query		string	false	"Data source identifier (expected: deltalake)"
// @Param			step			query		int		false	"Bucket size in seconds. Default: 60"
// @Success		200	{object}	summaryEnvelope
// @Router			/topru/summary [get]
func (s *Service) summaryHandler(c *gin.Context) {
	header, query, ok := bindTopRUSummaryRequest(c)
	if !ok {
		return
	}

	ctx, cancel := context.WithTimeout(c.Request.Context(), defaultTimeout)
	defer cancel()

	resp, err := s.queryTopRUSummary(ctx, header, query)
	if err != nil {
		log.Warn("failed to query topru summary", zap.Error(err))
		c.JSON(http.StatusInternalServerError, errorResponse{Error: true, Message: "Internal Server Error", FullText: err.Error()})
		return
	}

	c.JSON(http.StatusOK, summaryEnvelope{Status: "ok", Data: resp.Data})
}

// @Summary		Get TopRU summary by plan (RU by SQL/plan digest)
// @Tags			topru
// @Description	Aggregate ReadUnits (RU) grouped by sql_digest and plan_digest. Optionally filter to one sql_digest via digest=.
// @Produce		json
// @Param			X-Provider		header		string	true	"Cloud provider where RU data is stored (aws/gcp/azure)"
// @Param			X-Region		header		string	true	"Cloud region where RU data is stored"
// @Param			X-Org-Id		header		string	true	"Organization ID (x-org-id)"
// @Param			X-Project-Id	header		string	false	"Project ID (x-project-id; optional)"
// @Param			X-Cluster-Id	header		string	true	"Cluster ID (x-cluster-id)"
// @Param			X-Deploy-Type	header		string	true	"Deployment tier: dedicated/shared/nextgen/premium"
// @Param			start			query		string	true	"Start time in Unix seconds (inclusive)"
// @Param			end				query		string	true	"End time in Unix seconds (inclusive)"
// @Param			top				query		string	true	"Top N SQL/plan groups to return"
// @Param			digest			query		string	false	"Filter to a single sql_digest (drill-down from /topru/summary)"
// @Param			order_by		query		string	false	"Metric used to rank results: ru | exec_count | duration (default: ru)"
// @Param			user			query		string	false	"Omit to not filter by user. Pass empty (user=) to keep only rows with empty or NULL user; pass a name to filter by that user"
// @Param			data_source		query		string	false	"Data source identifier (expected: deltalake)"
// @Param			step			query		int		false	"Bucket size in seconds. Default: 60"
// @Success		200	{object}	summaryPlanEnvelope
// @Router			/topru/summaryplan [get]
func (s *Service) summaryPlanHandler(c *gin.Context) {
	header, base, ok := bindTopRUSummaryRequest(c)
	if !ok {
		return
	}
	query := &GetSummaryPlanRequest{GetSummaryRequest: *base}
	query.Digest = strings.TrimSpace(c.Query("digest"))

	ctx, cancel := context.WithTimeout(c.Request.Context(), defaultTimeout)
	defer cancel()

	resp, err := s.queryTopRUSummaryPlan(ctx, header, query)
	if err != nil {
		log.Warn("failed to query topru summaryplan", zap.Error(err))
		c.JSON(http.StatusInternalServerError, errorResponse{Error: true, Message: "Internal Server Error", FullText: err.Error()})
		return
	}

	c.JSON(http.StatusOK, summaryPlanEnvelope{Status: "ok", Data: resp.Data})
}

func bindTopRUSummaryRequest(c *gin.Context) (*ClusterHeader, *GetSummaryRequest, bool) {
	var header ClusterHeader
	if err := c.BindHeader(&header); err != nil {
		log.Warn("failed to bind topru summary header", zap.Error(err))
		c.JSON(http.StatusBadRequest, errorResponse{Error: true, Message: "failed to bind topru summary header", FullText: err.Error()})
		return nil, nil, false
	}
	if err := header.Validate(); err != nil {
		log.Warn("invalid topru summary header", zap.Error(err))
		c.JSON(http.StatusBadRequest, errorResponse{Error: true, Message: "Bad Request", FullText: err.Error()})
		return nil, nil, false
	}

	var query GetSummaryRequest
	if err := c.BindQuery(&query); err != nil {
		log.Warn("failed to bind topru summary query", zap.Error(err))
		c.JSON(http.StatusBadRequest, errorResponse{Error: true, Message: "failed to bind topru summary query", FullText: err.Error()})
		return nil, nil, false
	}
	if _, ok := c.GetQuery("user"); ok {
		u := c.Query("user")
		query.User = &u
	}
	if err := query.Validate(); err != nil {
		log.Warn("invalid topru summary query", zap.Error(err))
		c.JSON(http.StatusBadRequest, errorResponse{Error: true, Message: "invalid topru summary query", FullText: err.Error()})
		return nil, nil, false
	}

	if *header.DeployType == "byoc" {
		*header.BizType = "byoc"
		*header.DeployType = TierNextGen
	}
	return &header, &query, true
}

func (s *Service) queryTopRUSummary(ctx context.Context, header *ClusterHeader, query *GetSummaryRequest) (*SummaryResponse, error) {
	scheme, tablePath, err := buildTopRUSchemeAndTablePath(header)
	if err != nil {
		return nil, err
	}

	startSecs, endSecs, timeRangeSecs, err := parseUnixTimeRange(query.Start, query.End)
	if err != nil {
		return nil, err
	}

	topN, err := strconv.Atoi(query.Top)
	if err != nil {
		return nil, fmt.Errorf("invalid top: %w", err)
	}

	orderField, err := topruOrderField(query.OrderBy)
	if err != nil {
		return nil, err
	}

	sqlQuery, err := buildTopRUSummarySQL(tablePath, startSecs, endSecs, orderField, topN, query.User, query.Step)
	if err != nil {
		return nil, err
	}

	results, err := s.queryTopRUSummaryRows(ctx, tablePath, sqlQuery)
	if err != nil {
		return nil, err
	}

	sqlDigestSet, _ := collectTopRUDigests(results)
	sqlMetaMap := s.queryTopRUSQLMeta(ctx, scheme, header, startSecs, endSecs, sqlDigestSet)
	return s.processTopRUSummaryResults(results, timeRangeSecs, sqlMetaMap)
}

func (s *Service) queryTopRUSummaryPlan(ctx context.Context, header *ClusterHeader, query *GetSummaryPlanRequest) (*SummaryPlanResponse, error) {
	scheme, tablePath, err := buildTopRUSchemeAndTablePath(header)
	if err != nil {
		return nil, err
	}

	startSecs, endSecs, timeRangeSecs, err := parseUnixTimeRange(query.Start, query.End)
	if err != nil {
		return nil, err
	}

	topN, err := strconv.Atoi(query.Top)
	if err != nil {
		return nil, fmt.Errorf("invalid top: %w", err)
	}

	orderField, err := topruOrderField(query.OrderBy)
	if err != nil {
		return nil, err
	}

	sqlQuery, err := buildTopRUSummaryPlanSQL(tablePath, startSecs, endSecs, orderField, topN, query.User, query.Step, query.Digest)
	if err != nil {
		return nil, err
	}

	results, err := s.queryTopRUSummaryRows(ctx, tablePath, sqlQuery)
	if err != nil {
		return nil, err
	}

	sqlMetaMap, planMetaMap := s.queryTopRUMetas(ctx, scheme, header, startSecs, endSecs, results)
	return s.processTopRUSummaryPlanResults(results, timeRangeSecs, sqlMetaMap, planMetaMap)
}

func buildTopRUSchemeAndTablePath(header *ClusterHeader) (scheme, tablePath string, err error) {
	env := "prod"
	if header.Env != nil && *header.Env != "" {
		env = *header.Env
	}
	sharedTier := *header.DeployType == TierShared
	nextGen := *header.DeployType == TierNextGen || *header.DeployType == TierPremium
	bucket := storage.BucketName(env, *header.Region, sharedTier, nextGen)

	switch strings.ToLower(*header.Provider) {
	case "aws", "alicloud":
		scheme = fmt.Sprintf("s3://%s", bucket)
	case "gcp":
		scheme = fmt.Sprintf("gs://%s", bucket)
	case "azure":
		account := os.Getenv("AZURE_STORAGE_ACCOUNT")
		scheme = fmt.Sprintf("abfss://%s@%s.dfs.core.windows.net", bucket, account)
	case "local":
		scheme = "./"
	default:
		return "", "", fmt.Errorf("unsupported cloud provider: %s", *header.Provider)
	}

	return scheme, topRUTablePath(scheme, *header.TenantID, *header.ClusterID), nil
}

func parseUnixTimeRange(startStr, endStr string) (startSecs int64, endSecs int64, rangeSecs int64, err error) {
	startSecs, err = strconv.ParseInt(startStr, 10, 64)
	if err != nil {
		return 0, 0, 0, fmt.Errorf("invalid start time: %w", err)
	}
	endSecs, err = strconv.ParseInt(endStr, 10, 64)
	if err != nil {
		return 0, 0, 0, fmt.Errorf("invalid end time: %w", err)
	}
	rangeSecs = endSecs - startSecs
	if rangeSecs <= 0 {
		return 0, 0, 0, fmt.Errorf("invalid time range")
	}
	return startSecs, endSecs, rangeSecs, nil
}

func topruOrderField(orderBy string) (string, error) {
	if orderBy == "" {
		orderBy = "ru"
	}
	switch orderBy {
	case "ru":
		return "topru_total_ru", nil
	case "exec_count":
		return "topru_exec_count", nil
	case "duration":
		return "topru_exec_duration", nil
	default:
		return "", fmt.Errorf("invalid order_by: %s", orderBy)
	}
}

// formatStepInterval converts seconds to a time interval format for time_bucket
func formatStepInterval(stepSeconds int) string {
	if stepSeconds <= 0 {
		return "60 seconds" // default
	}

	return fmt.Sprintf("%d seconds", stepSeconds)
}

func buildTopRUSummarySQL(tablePath string, startSecs, endSecs int64, orderField string, topN int, user *string, stepSeconds int) (string, error) {
	// Aggregate by sql_digest only (plans collapsed).
	userFilter := buildTopRUUserFilter(user)
	bucketSize := formatStepInterval(stepSeconds)

	tmplData := map[string]interface{}{
		"TablePath":       tablePath,
		"StartSecs":       startSecs,
		"EndSecs":         endSecs,
		"OrderField":      orderField,
		"Top":             topN,
		"BucketSize":      bucketSize,
		"UserFilter":      userFilter,
		"TotalRUSelect":   "SUM(topru_total_ru) AS total_ru",
		"TotalExecSelect": "SUM(topru_exec_count) AS total_exec_count",
		"TotalDurSelect":  "SUM(topru_exec_duration) AS total_duration_ns",
	}

	const sqlTemplate = `
WITH raw_data AS (
	SELECT
		sql_digest, timestamps,
		topru_total_ru, topru_exec_count, topru_exec_duration
	FROM delta_scan('{{.TablePath}}')
	WHERE {{.OrderField}} > 0 AND timestamps BETWEEN {{.StartSecs}} AND {{.EndSecs}}{{.UserFilter}}
),
top_groups AS (
	SELECT sql_digest
	FROM (
		SELECT sql_digest, SUM({{.OrderField}}) AS total_order_value
		FROM raw_data
		WHERE COALESCE(sql_digest, '') <> ''
		GROUP BY sql_digest
		ORDER BY total_order_value DESC
		LIMIT {{.Top}}
	) sub
),
tagged_data AS (
	SELECT
		CASE WHEN g.sql_digest IS NULL THEN '' ELSE COALESCE(r.sql_digest, '') END AS sql_digest,
		time_bucket('{{.BucketSize}}', to_timestamp(timestamps)) AS time_window,
		r.topru_total_ru,
		r.topru_exec_count,
		r.topru_exec_duration
	FROM raw_data r
	LEFT JOIN top_groups g
		ON r.sql_digest = g.sql_digest
),
time_bucket_data AS (
	SELECT
		sql_digest, time_window,
		{{.TotalRUSelect}},
		{{.TotalExecSelect}},
		{{.TotalDurSelect}}
	FROM tagged_data
	GROUP BY sql_digest, time_window
)
SELECT
	t.sql_digest, CAST(NULL AS VARCHAR) AS plan_digest, t.time_window,
	t.total_ru, t.total_exec_count, t.total_duration_ns
FROM time_bucket_data t
ORDER BY t.sql_digest, t.time_window ASC
`

	tmpl, err := template.New("topru_summary").Parse(sqlTemplate)
	if err != nil {
		return "", fmt.Errorf("parse SQL template: %w", err)
	}
	var buf bytes.Buffer
	if err := tmpl.Execute(&buf, tmplData); err != nil {
		return "", fmt.Errorf("execute SQL template: %w", err)
	}
	return buf.String(), nil
}

func buildTopRUSummaryPlanSQL(tablePath string, startSecs, endSecs int64, orderField string, topN int, user *string, stepSeconds int, digest string) (string, error) {
	// Aggregate by sql_digest + plan_digest; optional digest filter for drill-down.
	userFilter := buildTopRUUserFilter(user)
	digestFilter := ""
	if strings.TrimSpace(digest) != "" {
		digestEscaped := strings.ReplaceAll(strings.TrimSpace(digest), "'", "''")
		digestFilter = fmt.Sprintf(" AND sql_digest = '%s'", digestEscaped)
	}
	bucketSize := formatStepInterval(stepSeconds)

	tmplData := map[string]interface{}{
		"TablePath":       tablePath,
		"StartSecs":       startSecs,
		"EndSecs":         endSecs,
		"OrderField":      orderField,
		"Top":             topN,
		"BucketSize":      bucketSize,
		"UserFilter":      userFilter,
		"DigestFilter":    digestFilter,
		"TotalRUSelect":   "SUM(topru_total_ru) AS total_ru",
		"TotalExecSelect": "SUM(topru_exec_count) AS total_exec_count",
		"TotalDurSelect":  "SUM(topru_exec_duration) AS total_duration_ns",
	}

	const sqlTemplate = `
WITH raw_data AS (
	SELECT
		sql_digest, plan_digest, timestamps,
		topru_total_ru, topru_exec_count, topru_exec_duration
	FROM delta_scan('{{.TablePath}}')
	WHERE {{.OrderField}} > 0 AND timestamps BETWEEN {{.StartSecs}} AND {{.EndSecs}}{{.UserFilter}}{{.DigestFilter}}
),
top_groups AS (
	SELECT sql_digest, plan_digest
	FROM (
		SELECT sql_digest, plan_digest, SUM({{.OrderField}}) AS total_order_value
		FROM raw_data
		WHERE COALESCE(sql_digest, '') <> ''
		GROUP BY sql_digest, plan_digest
		ORDER BY total_order_value DESC
		LIMIT {{.Top}}
	) sub
),
tagged_data AS (
	SELECT
		CASE WHEN g.sql_digest IS NULL THEN '' ELSE COALESCE(r.sql_digest, '') END AS sql_digest,
		CASE WHEN g.sql_digest IS NULL THEN '' ELSE COALESCE(r.plan_digest, '') END AS plan_digest,
		time_bucket('{{.BucketSize}}', to_timestamp(timestamps)) AS time_window,
		r.topru_total_ru,
		r.topru_exec_count,
		r.topru_exec_duration
	FROM raw_data r
	LEFT JOIN top_groups g
		ON r.sql_digest = g.sql_digest AND COALESCE(r.plan_digest, '') = COALESCE(g.plan_digest, '')
),
time_bucket_data AS (
	SELECT
		sql_digest, plan_digest, time_window,
		{{.TotalRUSelect}},
		{{.TotalExecSelect}},
		{{.TotalDurSelect}}
	FROM tagged_data
	GROUP BY sql_digest, plan_digest, time_window
)
SELECT
	t.sql_digest, t.plan_digest, t.time_window,
	t.total_ru, t.total_exec_count, t.total_duration_ns
FROM time_bucket_data t
ORDER BY t.sql_digest, t.plan_digest, t.time_window ASC
`

	tmpl, err := template.New("topru_summaryplan").Parse(sqlTemplate)
	if err != nil {
		return "", fmt.Errorf("parse SQL template: %w", err)
	}
	var buf bytes.Buffer
	if err := tmpl.Execute(&buf, tmplData); err != nil {
		return "", fmt.Errorf("execute SQL template: %w", err)
	}
	return buf.String(), nil
}

func (s *Service) queryTopRUSummaryRows(ctx context.Context, tablePath, sqlQuery string) ([]topruQueryResult, error) {
	var results []topruQueryResult
	if err := s.duckDBAdapter.SelectSlowlogInDLWithoutFilterPushdown(ctx, &results, "", tablePath, sqlQuery); err != nil {
		return nil, fmt.Errorf("query topru summary: %w", err)
	}
	return results, nil
}

func (s *Service) queryTopRUMetas(ctx context.Context, scheme string, header *ClusterHeader, startSecs, endSecs int64, results []topruQueryResult) (map[string]string, map[string]string) {
	sqlDigestSet, planDigestSet := collectTopRUDigests(results)
	sqlMetaMap := s.queryTopRUSQLMeta(ctx, scheme, header, startSecs, endSecs, sqlDigestSet)
	planMetaMap := s.queryTopRUPlanMeta(ctx, scheme, header, startSecs, endSecs, planDigestSet)
	return sqlMetaMap, planMetaMap
}

func collectTopRUDigests(results []topruQueryResult) (map[string]struct{}, map[string]struct{}) {
	sqlDigestSet := make(map[string]struct{})
	planDigestSet := make(map[string]struct{})
	for _, row := range results {
		if row.SQLDigest != nil && *row.SQLDigest != "" {
			sqlDigestSet[*row.SQLDigest] = struct{}{}
		}
		if row.PlanDigest != nil && *row.PlanDigest != "" {
			planDigestSet[*row.PlanDigest] = struct{}{}
		}
	}
	return sqlDigestSet, planDigestSet
}

func (s *Service) queryTopRUSQLMeta(ctx context.Context, scheme string, header *ClusterHeader, startSecs, endSecs int64, sqlDigestSet map[string]struct{}) map[string]string {
	sqlMetaMap := make(map[string]string)
	if len(sqlDigestSet) == 0 {
		return sqlMetaMap
	}

	sqlMetaPath := fmt.Sprintf("%s/deltalake/org=%s/cluster=%s/type=topsql/component=topsql_sql_meta", scheme, *header.TenantID, *header.ClusterID)
	startDate, endDate := toUTCDateRange(startSecs, endSecs)

	sqlDigestList := make([]string, 0, len(sqlDigestSet))
	for d := range sqlDigestSet {
		sqlDigestList = append(sqlDigestList, fmt.Sprintf("'%s'", strings.ReplaceAll(d, "'", "''")))
	}

	sqlMetaQuery := fmt.Sprintf(`
		SELECT sql_digest, normalized_sql
		FROM delta_scan('%s')
		WHERE date >= '%s' AND date <= '%s' AND sql_digest IN (%s)
	`, sqlMetaPath, startDate, endDate, strings.Join(sqlDigestList, ", "))

	type sqlMetaResult struct {
		SQLDigest     string `db:"sql_digest"`
		NormalizedSQL string `db:"normalized_sql"`
	}
	var sqlMetaResults []sqlMetaResult
	if err := s.duckDBAdapter.SelectSlowlogInDLWithoutFilterPushdown(ctx, &sqlMetaResults, "", sqlMetaPath, sqlMetaQuery); err != nil {
		log.Warn("Failed to query TopRU SQL metadata", zap.Error(err))
		return sqlMetaMap
	}
	for _, m := range sqlMetaResults {
		sqlMetaMap[m.SQLDigest] = m.NormalizedSQL
	}
	return sqlMetaMap
}

func (s *Service) queryTopRUPlanMeta(ctx context.Context, scheme string, header *ClusterHeader, startSecs, endSecs int64, planDigestSet map[string]struct{}) map[string]string {
	planMetaMap := make(map[string]string)
	if len(planDigestSet) == 0 {
		return planMetaMap
	}

	planMetaPath := fmt.Sprintf("%s/deltalake/org=%s/cluster=%s/type=topsql/component=topsql_plan_meta", scheme, *header.TenantID, *header.ClusterID)
	startDate, endDate := toUTCDateRange(startSecs, endSecs)

	planDigestList := make([]string, 0, len(planDigestSet))
	for d := range planDigestSet {
		planDigestList = append(planDigestList, fmt.Sprintf("'%s'", strings.ReplaceAll(d, "'", "''")))
	}

	planMetaQuery := fmt.Sprintf(`
		SELECT plan_digest, normalized_plan, encoded_normalized_plan
		FROM delta_scan('%s')
		WHERE date >= '%s' AND date <= '%s' AND plan_digest IN (%s)
	`, planMetaPath, startDate, endDate, strings.Join(planDigestList, ", "))

	type planMetaResult struct {
		PlanDigest            string `db:"plan_digest"`
		NormalizedPlan        string `db:"normalized_plan"`
		EncodedNormalizedPlan string `db:"encoded_normalized_plan"`
	}
	var planMetaResults []planMetaResult
	if err := s.duckDBAdapter.SelectSlowlogInDLWithoutFilterPushdown(ctx, &planMetaResults, "", planMetaPath, planMetaQuery); err != nil {
		log.Warn("Failed to query TopRU plan metadata", zap.Error(err))
		return planMetaMap
	}
	for _, m := range planMetaResults {
		if m.PlanDigest == "" {
			continue
		}
		text, err := planmeta.ResolvePlanText(m.NormalizedPlan, m.EncodedNormalizedPlan)
		if err != nil {
			log.Warn("Failed to resolve TopRU plan text",
				zap.String("plan_digest", m.PlanDigest),
				zap.Error(err))
			continue
		}
		if text == "" {
			continue
		}
		planMetaMap[m.PlanDigest] = text
	}
	return planMetaMap
}

func toUTCDateRange(startSecs, endSecs int64) (startDate string, endDate string) {
	startTime := time.Unix(startSecs, 0).UTC()
	endTime := time.Unix(endSecs, 0).UTC()
	return startTime.Format("2006-01-02"), endTime.Format("2006-01-02")
}

// GetUserStatsRequest for TopRU user RU statistics.
type GetUserStatsRequest struct {
	Start string `form:"start" binding:"required"`
	End   string `form:"end" binding:"required"`
	Top   string `form:"top"` // optional, default 20
}

func (q *GetUserStatsRequest) Validate() error {
	if q.Start == "" {
		return fmt.Errorf("start not found")
	}
	if q.End == "" {
		return fmt.Errorf("end not found")
	}
	return nil
}

const (
	systemMaintenanceUserLabel = "System Maintenance"
	cloudAdminUserSuffix       = ".cloud_admin@%"
)

// displayUserStatsUser maps internal cloud admin accounts to a friendly label in API responses.
func displayUserStatsUser(user string) string {
	if strings.HasSuffix(user, cloudAdminUserSuffix) {
		return systemMaintenanceUserLabel
	}
	return user
}

// buildTopRUUserFilter returns SQL AND-clause for optional user filter in summary queries.
// When the client passes the display label "System Maintenance", match any user whose name ends
// with ".cloud_admin@%" since the exact account prefix is cluster-specific and unknown.
func buildTopRUUserFilter(user *string) string {
	if user == nil {
		return ""
	}
	if *user == systemMaintenanceUserLabel {
		suffixEscaped := strings.ReplaceAll(cloudAdminUserSuffix, "'", "''")
		return fmt.Sprintf(" AND ends_with(COALESCE(user, ''), '%s')", suffixEscaped)
	}
	userEscaped := strings.ReplaceAll(*user, "'", "''")
	return fmt.Sprintf(" AND (user = '%s' OR (user IS NULL AND '%s' = ''))", userEscaped, userEscaped)
}

// UserStatsItem one row per user.
type UserStatsItem struct {
	User       string  `json:"user"`
	TotalRU    float64 `json:"total_ru"`
	ExecCount  int64   `json:"exec_count"`
	DurationNs int64   `json:"duration_ns"`
}

type userStatsQueryResult struct {
	User          *string  `db:"user"`
	TotalRU       float64  `db:"total_ru"`
	TotalExecCnt  *float64 `db:"total_exec_count"`
	TotalDuration *float64 `db:"total_duration_ns"`
}

// @Summary		Get TopRU user statistics (RU per user)
// @Tags			topru
// @Description	Aggregate RU and execution stats grouped by user over the given time range.
// @Produce		json
// @Param			X-Provider		header		string	true	"Cloud provider where RU data is stored (aws/gcp/azure)"
// @Param			X-Region		header		string	true	"Cloud region where RU data is stored"
// @Param			X-Org-Id		header		string	true	"Organization ID (x-org-id)"
// @Param			X-Project-Id	header		string	false	"Project ID (x-project-id; optional)"
// @Param			X-Cluster-Id	header		string	true	"Cluster ID (x-cluster-id)"
// @Param			X-Deploy-Type	header		string	true	"Deployment tier"
// @Param			start			query		string	true	"Start time in Unix seconds (inclusive)"
// @Param			end				query		string	true	"End time in Unix seconds (inclusive)"
// @Param			top				query		string	false	"Top N users to return (default: 20)"
// @Success		200	{object}	userStatsEnvelope
// @Router			/topru/user_stats [get]
func (s *Service) userStatsHandler(c *gin.Context) {
	var header ClusterHeader
	if err := c.BindHeader(&header); err != nil {
		log.Warn("failed to bind topru user_stats header", zap.Error(err))
		c.JSON(http.StatusBadRequest, errorResponse{Error: true, Message: "failed to bind topru user_stats header", FullText: err.Error()})
		return
	}
	if err := header.Validate(); err != nil {
		log.Warn("invalid topru user_stats header", zap.Error(err))
		c.JSON(http.StatusBadRequest, errorResponse{Error: true, Message: "Bad Request", FullText: err.Error()})
		return
	}

	var query GetUserStatsRequest
	if err := c.BindQuery(&query); err != nil {
		log.Warn("failed to bind topru user_stats query", zap.Error(err))
		c.JSON(http.StatusBadRequest, errorResponse{Error: true, Message: "failed to bind topru user_stats query", FullText: err.Error()})
		return
	}
	if err := query.Validate(); err != nil {
		log.Warn("invalid topru user_stats query", zap.Error(err))
		c.JSON(http.StatusBadRequest, errorResponse{Error: true, Message: "invalid topru user_stats query", FullText: err.Error()})
		return
	}

	if *header.DeployType == "byoc" {
		*header.BizType = "byoc"
		*header.DeployType = TierNextGen
	}

	ctx, cancel := context.WithTimeout(c.Request.Context(), defaultTimeout)
	defer cancel()

	data, err := s.queryTopRUUserStats(ctx, &header, &query)
	if err != nil {
		log.Warn("failed to query topru user_stats", zap.Error(err))
		c.JSON(http.StatusInternalServerError, errorResponse{Error: true, Message: "Internal Server Error", FullText: err.Error()})
		return
	}

	c.JSON(http.StatusOK, userStatsEnvelope{Status: "ok", Data: data})
}

func (s *Service) queryTopRUUserStats(ctx context.Context, header *ClusterHeader, query *GetUserStatsRequest) ([]UserStatsItem, error) {
	env := "prod"
	if header.Env != nil && *header.Env != "" {
		env = *header.Env
	}
	sharedTier := *header.DeployType == TierShared
	nextGen := *header.DeployType == TierNextGen || *header.DeployType == TierPremium
	bucket := storage.BucketName(env, *header.Region, sharedTier, nextGen)

	var scheme string
	switch strings.ToLower(*header.Provider) {
	case "aws", "alicloud":
		scheme = fmt.Sprintf("s3://%s", bucket)
	case "gcp":
		scheme = fmt.Sprintf("gs://%s", bucket)
	case "azure":
		account := os.Getenv("AZURE_STORAGE_ACCOUNT")
		scheme = fmt.Sprintf("abfss://%s@%s.dfs.core.windows.net", bucket, account)
	case "local":
		scheme = "./"
	default:
		return nil, fmt.Errorf("unsupported cloud provider: %s", *header.Provider)
	}

	tablePath := topRUTablePath(scheme, *header.TenantID, *header.ClusterID)

	startSecs, err := strconv.ParseInt(query.Start, 10, 64)
	if err != nil {
		return nil, fmt.Errorf("invalid start time: %w", err)
	}
	endSecs, err := strconv.ParseInt(query.End, 10, 64)
	if err != nil {
		return nil, fmt.Errorf("invalid end time: %w", err)
	}

	topN := 20
	if query.Top != "" {
		topN, err = strconv.Atoi(query.Top)
		if err != nil {
			return nil, fmt.Errorf("invalid top: %w", err)
		}
	}

	// Group by user, sum RU / exec_count / duration, order by total_ru desc
	sqlQuery := fmt.Sprintf(`
SELECT 
	COALESCE(user, '') AS user,
	SUM(topru_total_ru) AS total_ru,
	SUM(topru_exec_count) AS total_exec_count,
	SUM(topru_exec_duration) AS total_duration_ns
FROM delta_scan('%s')
WHERE timestamps BETWEEN %d AND %d
GROUP BY COALESCE(user, '')
ORDER BY total_ru DESC
LIMIT %d
`, tablePath, startSecs, endSecs, topN)

	var results []userStatsQueryResult
	if err := s.duckDBAdapter.SelectSlowlogInDLWithoutFilterPushdown(ctx, &results, "", tablePath, sqlQuery); err != nil {
		return nil, fmt.Errorf("query topru user_stats: %w", err)
	}

	list := make([]UserStatsItem, 0, len(results))
	for _, r := range results {
		user := ""
		if r.User != nil {
			user = *r.User
		}
		execCnt := int64(0)
		if r.TotalExecCnt != nil {
			execCnt = int64(*r.TotalExecCnt)
		}
		dur := int64(0)
		if r.TotalDuration != nil {
			dur = int64(*r.TotalDuration)
		}
		list = append(list, UserStatsItem{
			User:       displayUserStatsUser(user),
			TotalRU:    r.TotalRU,
			ExecCount:  execCnt,
			DurationNs: dur,
		})
	}
	return list, nil
}

func (s *Service) processTopRUSummaryResults(rows []topruQueryResult, timeRangeSecs int64, sqlMetaMap map[string]string) (*SummaryResponse, error) {
	agg := make(map[string]*TopRUSummaryItem)

	for _, r := range rows {
		sqlD := ""
		if r.SQLDigest != nil {
			sqlD = *r.SQLDigest
		}
		if _, ok := agg[sqlD]; !ok {
			agg[sqlD] = &TopRUSummaryItem{
				SQLDigest:     sqlD,
				TimestampSec:  []int64{},
				TotalRUByTime: []float64{},
			}
		}
		item := agg[sqlD]
		item.TotalRU += r.TotalRU
		if r.TotalExecCount != nil {
			item.ExecCount += int64(*r.TotalExecCount)
		}
		if r.TotalDurationNs != nil {
			item.DurationNs += int64(*r.TotalDurationNs)
		}
		if r.TimeWindow != nil {
			item.TimestampSec = append(item.TimestampSec, r.TimeWindow.Unix())
			item.TotalRUByTime = append(item.TotalRUByTime, r.TotalRU)
		}
	}

	list := make([]TopRUSummaryItem, 0, len(agg))
	for _, item := range agg {
		item.SQLText = sqlMetaMap[item.SQLDigest]
		if timeRangeSecs > 0 {
			item.ExecCountPerSec = float64(item.ExecCount) / float64(timeRangeSecs)
		}
		if item.ExecCount > 0 && item.DurationNs > 0 {
			item.DurationPerExecMs = float64(item.DurationNs) / float64(item.ExecCount) / 1e6
		}
		list = append(list, *item)
	}
	sort.Slice(list, func(i, j int) bool {
		iIsOthers := list[i].SQLDigest == ""
		jIsOthers := list[j].SQLDigest == ""
		if iIsOthers != jIsOthers {
			return !iIsOthers
		}
		return list[i].TotalRU > list[j].TotalRU
	})

	return &SummaryResponse{Data: list}, nil
}

func (s *Service) processTopRUSummaryPlanResults(rows []topruQueryResult, timeRangeSecs int64, sqlMetaMap, planMetaMap map[string]string) (*SummaryPlanResponse, error) {
	type key struct{ sql, plan string }
	agg := make(map[key]*TopRUSummaryPlanItem)

	for _, r := range rows {
		sqlD := ""
		if r.SQLDigest != nil {
			sqlD = *r.SQLDigest
		}
		planD := ""
		if r.PlanDigest != nil {
			planD = *r.PlanDigest
		}
		k := key{sql: sqlD, plan: planD}
		if _, ok := agg[k]; !ok {
			agg[k] = &TopRUSummaryPlanItem{
				SQLDigest:     sqlD,
				PlanDigest:    planD,
				TimestampSec:  []int64{},
				TotalRUByTime: []float64{},
			}
		}
		item := agg[k]
		item.TotalRU += r.TotalRU
		if r.TotalExecCount != nil {
			item.ExecCount += int64(*r.TotalExecCount)
		}
		if r.TotalDurationNs != nil {
			item.DurationNs += int64(*r.TotalDurationNs)
		}
		if r.TimeWindow != nil {
			item.TimestampSec = append(item.TimestampSec, r.TimeWindow.Unix())
			item.TotalRUByTime = append(item.TotalRUByTime, r.TotalRU)
		}
	}

	list := make([]TopRUSummaryPlanItem, 0, len(agg))
	for _, item := range agg {
		item.SQLText = sqlMetaMap[item.SQLDigest]
		item.PlanText = planMetaMap[item.PlanDigest]
		if timeRangeSecs > 0 {
			item.ExecCountPerSec = float64(item.ExecCount) / float64(timeRangeSecs)
		}
		if item.ExecCount > 0 && item.DurationNs > 0 {
			item.DurationPerExecMs = float64(item.DurationNs) / float64(item.ExecCount) / 1e6
		}
		list = append(list, *item)
	}
	sort.Slice(list, func(i, j int) bool {
		iIsOthers := list[i].SQLDigest == "" && list[i].PlanDigest == ""
		jIsOthers := list[j].SQLDigest == "" && list[j].PlanDigest == ""
		if iIsOthers != jIsOthers {
			return !iIsOthers
		}
		return list[i].TotalRU > list[j].TotalRU
	})

	return &SummaryPlanResponse{Data: list}, nil
}