package slowstats

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"io"
	"os"
	"path"
	"path/filepath"
	"strings"
	"time"

	"github.com/aws/smithy-go"
	_ "github.com/mattn/go-sqlite3"
	"github.com/pingcap/log"
	"go.uber.org/zap"

	"github.com/tidbcloud/observability/o11ydiagnosis/pkg/storage"
)

type ClusterInfo struct {
	Provider   string
	Region     string
	Env        string
	TenantID   string
	ProjectID  string
	ClusterID  string
	SharedTier bool
	NextGen    bool
	Byoc       bool
}

type SlowStats struct {
	DB            string  `json:"db"`
	SQLDigest     string  `json:"sql_digest"`
	SQLText       string  `json:"sql_text"`
	StatementType string  `json:"statement_type"`
	Count         int     `json:"count"`
	SumLatency    float64 `json:"sum_latency"`
	MaxLatency    float64 `json:"max_latency"`
	AvgLatency    float64 `json:"avg_latency"`
	SumMemory     float64 `json:"sum_memory"`
	MaxMemory     float64 `json:"max_memory"`
	AvgMemory     float64 `json:"avg_memory"`
	SumDisk       float64 `json:"sum_disk"`
	MaxDisk       float64 `json:"max_disk"`
	AvgDisk       float64 `json:"avg_disk"`
}

type MetricValues struct {
	Timestamps []int64 `json:"timestamps"`
	Values     []int64 `json:"values"`
}

type StatsQuery struct {
	localPath string
	store     storage.Storage
}

func NewStatsQuery(localPath string, store storage.Storage) *StatsQuery {
	s := &StatsQuery{
		localPath,
		store,
	}
	go s.cleanCacheLoop()
	return s
}

func (s *StatsQuery) QueryDatabases(
	ctx context.Context,
	cluster ClusterInfo,
	beginHourTs int64,
	hours int,
) ([]string, error) {
	db, err := s.openStatsDB(ctx, cluster, beginHourTs, hours)
	if err != nil {
		return nil, err
	}
	if db == nil {
		return []string{}, nil
	}
	rows, err := db.QueryContext(ctx, `
		SELECT
			DISTINCT "db"
		FROM
			"stats_by_db";
	`)
	if err != nil {
		return nil, err
	}
	if rows.Err() != nil {
		return nil, rows.Err()
	}
	defer func() {
		_ = rows.Close()
	}()
	var dbs []string
	for rows.Next() {
		var db sql.NullString
		if err := rows.Scan(&db); err != nil {
			return nil, err
		}
		if db.Valid {
			dbs = append(dbs, db.String)
		}
	}
	return dbs, nil
}

func (s *StatsQuery) QueryStats(
	ctx context.Context,
	cluster ClusterInfo,
	beginHourTs int64,
	hours int,
	statementTypes []string,
	statementTypesNotIn bool,
	databases []string,
	internal bool,
	order string,
	limit int,
) ([]SlowStats, error) {
	if order == "" {
		order = "count"
	}
	if limit == 0 {
		limit = 10
	}
	db, err := s.openStatsDB(ctx, cluster, beginHourTs, hours)
	if err != nil {
		return nil, err
	}
	if db == nil {
		return []SlowStats{}, nil
	}
	if err := fillStmtType(ctx, db); err != nil {
		return nil, err
	}
	if len(databases) == 0 {
		var args []any
		whereStmtTypeIn := ""
		if len(statementTypes) > 0 {
			var placeholders []string
			for _, st := range statementTypes {
				args = append(args, st)
				placeholders = append(placeholders, "?")
			}
			sqlIn := "IN"
			if statementTypesNotIn {
				sqlIn = "NOT IN"
			}
			whereStmtTypeIn = fmt.Sprintf(`"stmt_type" %s (%s) AND`, sqlIn, strings.Join(placeholders, ", "))
		}
		args = append(args, internal, limit)
		rows, err := db.QueryContext(ctx, fmt.Sprintf(`
			SELECT
				"stats_by_db"."digest" AS "sql_digest",
				"meta"."text" AS "sql_text",
				"meta"."stmt_type" AS "statement_type",
				sum("count") as "count",
				sum("query_time_sum") AS "sum_latency",
				max("query_time_max") AS "max_latency",
				sum("query_time_sum") / sum("count") AS "avg_latency",
				sum("mem_max_sum") AS "sum_memory",
				max("mem_max_max") AS "max_memory",
				sum("mem_max_sum") / sum("count") AS "avg_memory",
				sum("disk_max_sum") AS "sum_disk",
				max("disk_max_max") AS "max_disk",
				sum("disk_max_sum") / sum("count") AS "avg_disk"
			FROM
				"stats_by_db" JOIN "meta" ON "stats_by_db"."digest" = "meta"."digest"
			WHERE
				%s
				"internal" IS ?
			GROUP BY
				"stats_by_db"."digest"
			ORDER BY
				"%s" DESC
			LIMIT
				?;
		`, whereStmtTypeIn, order), args...)
		if err != nil {
			return nil, err
		}
		if rows.Err() != nil {
			return nil, rows.Err()
		}
		defer func() {
			_ = rows.Close()
		}()
		var rs []SlowStats
		for rows.Next() {
			var s SlowStats
			if err := rows.Scan(
				&s.SQLDigest,
				&s.SQLText,
				&s.StatementType,
				&s.Count,
				&s.SumLatency,
				&s.MaxLatency,
				&s.AvgLatency,
				&s.SumMemory,
				&s.MaxMemory,
				&s.AvgMemory,
				&s.SumDisk,
				&s.MaxDisk,
				&s.AvgDisk,
			); err != nil {
				return nil, err
			}
			rs = append(rs, s)
		}
		return rs, nil
	} else {
		var args []any
		whereStmtTypeIn := ""
		if len(statementTypes) > 0 {
			placeholders := []string{}
			for _, st := range statementTypes {
				args = append(args, st)
				placeholders = append(placeholders, "?")
			}
			sqlIn := "IN"
			if statementTypesNotIn {
				sqlIn = "NOT IN"
			}
			whereStmtTypeIn = fmt.Sprintf(`"stmt_type" %s (%s) AND`, sqlIn, strings.Join(placeholders, ", "))
		}
		placeholders := []string{}
		for _, db := range databases {
			args = append(args, db)
			placeholders = append(placeholders, "?")
		}
		dbPlaceholders := strings.Join(placeholders, ", ")
		args = append(args, internal, limit)
		rows, err := db.QueryContext(ctx, fmt.Sprintf(`
				SELECT
					"db",
					"stats_by_db"."digest" AS "sql_digest",
					"meta"."text" AS "sql_text",
					"meta"."stmt_type" AS "statement_type",
					sum("count") as "count",
					sum("query_time_sum") AS "sum_latency",
					max("query_time_max") AS "max_latency",
					sum("query_time_sum") / sum("count") AS "avg_latency",
					sum("mem_max_sum") AS "sum_memory",
					max("mem_max_max") AS "max_memory",
					sum("mem_max_sum") / sum("count") AS "avg_memory",
					sum("disk_max_sum") AS "sum_disk",
					max("disk_max_max") AS "max_disk",
					sum("disk_max_sum") / sum("count") AS "avg_disk"
				FROM
					"stats_by_db" JOIN "meta" ON "stats_by_db"."digest" = "meta"."digest"
				WHERE
					%s
					"db" IN (%s) AND
					"internal" IS ?
				GROUP BY
					"db", "stats_by_db"."digest"
				ORDER BY
					"%s" DESC
				LIMIT
					?;
			`, whereStmtTypeIn, dbPlaceholders, order), args...)
		if err != nil {
			return nil, err
		}
		if rows.Err() != nil {
			return nil, rows.Err()
		}
		defer func() {
			_ = rows.Close()
		}()
		var rs []SlowStats
		for rows.Next() {
			var s SlowStats
			if err := rows.Scan(
				&s.DB,
				&s.SQLDigest,
				&s.SQLText,
				&s.StatementType,
				&s.Count,
				&s.SumLatency,
				&s.MaxLatency,
				&s.AvgLatency,
				&s.SumMemory,
				&s.MaxMemory,
				&s.AvgMemory,
				&s.SumDisk,
				&s.MaxDisk,
				&s.AvgDisk,
			); err != nil {
				return nil, err
			}
			rs = append(rs, s)
		}
		return rs, nil
	}
}

func (s *StatsQuery) QueryMetric(
	ctx context.Context,
	cluster ClusterInfo,
	beginHourTs int64,
	hours int,
	name string,
) ([][]int64, error) {
	if name == "" {
		name = "count_per_minute"
	}
	db, err := s.openStatsDB(ctx, cluster, beginHourTs, hours)
	if err != nil {
		return nil, err
	}
	if db == nil {
		return [][]int64{}, nil
	}
	rows, err := db.QueryContext(ctx, fmt.Sprintf(`
		SELECT
			"ts", "count"
		FROM
			"metric_%s"
		WHERE
			"digest" == ''
		ORDER BY
			"ts" ASC;
	`, name))
	if err != nil {
		return nil, err
	}
	if rows.Err() != nil {
		return nil, rows.Err()
	}
	defer func() {
		_ = rows.Close()
	}()
	var items [][]int64
	for rows.Next() {
		var ts, count int64
		if err := rows.Scan(&ts, &count); err != nil {
			return nil, err
		}
		if hours < 6 || len(items) == 0 {
			items = append(items, []int64{ts, count})
			continue
		}
		step := int64(60)
		switch hours {
		case 6:
			step = 3 * 60
		case 12:
			step = 6 * 60
		case 24:
			step = 12 * 60
		case 72:
			step = 30 * 60
		case 168:
			step = 60 * 60
		}
		ts -= ts % step
		if ts == items[len(items)-1][0] {
			items[len(items)-1][1] += count
		} else {
			items = append(items, []int64{ts, count})
		}
	}
	return items, nil
}

func (s *StatsQuery) openStatsDB(ctx context.Context, cluster ClusterInfo, beginHourTs int64, hours int) (*sql.DB, error) {
	beginHourTs -= beginHourTs % 3600
	path := path.Join(s.localPath, fmt.Sprintf("%s-%dh-%s.db", cluster.ClusterID, hours, toUtcHour(beginHourTs)))
	if _, err := os.Stat(path); err != nil || (hours == 1 && time.Since(time.Unix(beginHourTs, 0)) <= 2*time.Hour) {
		if err != nil && !os.IsNotExist(err) {
			return nil, err
		}
		object, err := s.store.Object(ctx, drhouseBucket(cluster.Region), drhouseSlowStatsKey(cluster.ClusterID, beginHourTs, hours))
		if err != nil {
			if err == storage.ErrNotFound {
				return nil, nil
			}
			var ae smithy.APIError
			if errors.As(err, &ae) && ae.ErrorCode() == "NoSuchKey" {
				return nil, nil
			}
			return nil, err
		}
		defer object.Reader.Close()
		if err := os.MkdirAll(filepath.Dir(path), os.ModePerm); err != nil {
			return nil, err
		}
		file, err := os.Create(path)
		if err != nil {
			return nil, err
		}
		defer func() {
			_ = file.Close()
		}()
		if _, err := io.Copy(file, object.Reader); err != nil {
			return nil, err
		}
	}
	return sql.Open("sqlite3", path)
}

func (s *StatsQuery) cleanCacheLoop() {
	tick := time.NewTicker(time.Hour)
	defer tick.Stop()
	for range tick.C {
		entries, err := os.ReadDir(s.localPath)
		if err != nil {
			log.Error("failed to read dir", zap.Error(err))
			continue
		}
		for _, entry := range entries {
			if entry.IsDir() {
				continue
			}
			if filepath.Ext(entry.Name()) == ".db" {
				info, err := entry.Info()
				if err != nil {
					log.Error("failed to read file info", zap.Error(err))
					continue
				}
				if time.Since(info.ModTime()) > 12*time.Hour {
					if err := os.Remove(entry.Name()); err != nil {
						if strings.Contains(err.Error(), "no such file or directory") {
							log.Info("file already removed", zap.String("filename", entry.Name()))
						}
						log.Error("failed to remove old file", zap.Error(err), zap.String("filename", entry.Name()))
					} else {
						log.Info("removed old file", zap.String("filename", entry.Name()))
					}
				}
			}
		}
	}
}

// func hasTable(ctx context.Context, db *sql.DB, table string) (bool, error) {
// 	var hasStatsByDb int
// 	row := db.QueryRowContext(ctx, fmt.Sprintf(`
// 		SELECT
// 			count(*)
// 		FROM
// 			sqlite_master
// 		WHETE
// 			name = '%s' AND
// 			type = 'table';
// 	`, table))
// 	if err := row.Scan(&hasStatsByDb); err != nil {
// 		return false, err
// 	}
// 	return hasStatsByDb == 1, nil
// }

func hasColumn(ctx context.Context, db *sql.DB, table, column string) (bool, error) {
	var hasColumn int
	row := db.QueryRowContext(ctx, fmt.Sprintf(`
		SELECT
			count(*)
		FROM
			pragma_table_info("%s")
		WHERE
			name = '%s';
	`, table, column))
	if err := row.Scan(&hasColumn); err != nil {
		return false, err
	}
	return hasColumn == 1, nil
}

func toUtcHour(ts int64) string {
	return time.Unix(ts, 0).UTC().Format("2006010215")
}

func drhouseBucket(region string) string {
	return "o11y-prod-shared-" + strings.ToLower(region)
}

func drhouseSlowStatsKey(cluster string, beginHourTs int64, hours int) string {
	return fmt.Sprintf("diagnosis/data/%s/slowlog-stats/%dh/%s.db", cluster, hours, toUtcHour(beginHourTs))
}