package topsql

import (
	"fmt"

	"github.com/pingcap/log"
	"go.uber.org/zap"
	"gorm.io/gorm"
)

type TopSQLModel interface {
	ListInstances(clusterID, beginTs, endTs uint64) ([]InstanceRow, error)
	GetTopCPUTimeDigestsWindowed(clusterID uint64, windowSize, topN uint32, beginTs, endTs uint64, instanceID string) ([]TopCPUTimeDigestsWindowedRow, error)
	GetNormalizedSQLText(clusterID uint64, sqlDigestList []string) ([]SQLMetaRow, error)
}

var _ TopSQLModel = &TopSQLModelImpl{}

type TopSQLModelImpl struct {
	db *gorm.DB
}

func NewTopSQLModel(db *gorm.DB) *TopSQLModelImpl {
	return &TopSQLModelImpl{db: db}
}

const (
	queryTopCPUTimeDigestWindowed = `
	SELECT time_window, cpu_time_agg, cpu_time_sum_in_window, sql_digest
	FROM (SELECT *,
				 SUM(windowed.cpu_time_agg) OVER (PARTITION BY windowed.time_window ORDER BY windowed.cpu_time_agg DESC, windowed.sql_digest DESC ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS cpu_time_sum_in_window,
				 RANK() OVER (PARTITION BY windowed.time_window ORDER BY windowed.cpu_time_agg DESC, windowed.sql_digest DESC) AS rk
		  FROM (SELECT instance_id,
					   sql_digest,
					   cast(timestamp / ? as unsigned) AS time_window,
					   sum(cpu_time_ms) AS cpu_time_agg
				FROM tidb%d_metrics
				WHERE timestamp >= ?
				  AND timestamp < ?
				  AND instance_id = ?
				GROUP BY time_window, sql_digest) windowed) topsql
	WHERE topsql.rk <= ?;
	`
)

// GetTopCPUTimeDigestsWindowed may returns more than topN rows for each time window, because there are multiple rows with the same rank/cpu time
func (m *TopSQLModelImpl) GetTopCPUTimeDigestsWindowed(clusterID uint64, windowSize, topN uint32, beginTs, endTs uint64, instanceID string) ([]TopCPUTimeDigestsWindowedRow, error) {
	ret := make([]TopCPUTimeDigestsWindowedRow, 0)
	query := fmt.Sprintf(queryTopCPUTimeDigestWindowed, clusterID)
	log.Debug("GetTopCPUTimeDigestsWindowed", zap.String("sql", query))
	rows, err := m.db.Raw(query, windowSize, beginTs, endTs, instanceID, topN).Rows()
	if err != nil {
		return ret, err
	}
	var result TopCPUTimeDigestsWindowedRow
	for rows.Next() {
		if err := m.db.ScanRows(rows, &result); err != nil {
			log.Error("failed to call ScanRows", zap.Error(err))
		}
		ret = append(ret, result)
	}
	log.Debug("GetTopCPUTimeDigestsWindowed", zap.Any("result", ret))
	// this is for bypassing the golang-ci sqlclosecheck
	defer func() {
		err = rows.Close()
	}()
	return ret, err
}

func (m *TopSQLModelImpl) GetNormalizedSQLText(clusterID uint64, sqlDigestList []string) ([]SQLMetaRow, error) {
	query := fmt.Sprintf("SELECT sql_digest, normalized_sql FROM tidb%d_sql_meta WHERE sql_digest IN (?);", clusterID)
	ret := make([]SQLMetaRow, 0, len(sqlDigestList))
	rows, err := m.db.Raw(query, sqlDigestList).Rows()
	if err != nil {
		return ret, err
	}
	var result SQLMetaRow
	for rows.Next() {
		if err := m.db.ScanRows(rows, &result); err != nil {
			log.Error("failed to call ScanRows", zap.Error(err))
		}
		ret = append(ret, result)
	}
	defer func() {
		err = rows.Close()
	}()
	return ret, err
}

func (m *TopSQLModelImpl) ListInstances(clusterID, beginTs, endTs uint64) ([]InstanceRow, error) {
	query := fmt.Sprintf("SELECT id, name, component FROM tidb%d_instance;", clusterID)
	ret := make([]InstanceRow, 0)
	rows, err := m.db.Raw(query).Rows()
	if err != nil {
		return ret, err
	}
	var result InstanceRow
	for rows.Next() {
		if err := m.db.ScanRows(rows, &result); err != nil {
			log.Error("failed to call ScanRows", zap.Error(err))
		}
		ret = append(ret, result)
	}
	defer func() {
		err = rows.Close()
	}()
	return ret, err
}