package alert

import (
	"context"
	"database/sql"
	"fmt"
	"github.com/go-openapi/strfmt"
	"github.com/jinzhu/gorm"
	"github.com/pingcap/log"
	"github.com/pingcap/tep/pkg/apis/apiserver"
	"github.com/pingcap/tep/pkg/operator"
	store "github.com/pingcap/tep/pkg/registry/apiserver"
	"github.com/pingcap/tep/pkg/utils"
	"github.com/pingcap/tep/pkg/utils/alert"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/wushilin/stream"
	"go.uber.org/zap"
	"k8s.io/apimachinery/pkg/api/errors"
	metainternalversion "k8s.io/apimachinery/pkg/apis/meta/internalversion"
	"k8s.io/apimachinery/pkg/apis/meta/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/types"
	"reflect"
	"sort"
	"strings"
	"time"
)

type AlertStorage struct {
	store         *store.Store
	groupResource schema.GroupResource
}

func NewAlertStorage(store *store.Store, groupResource schema.GroupResource) *AlertStorage {
	return &AlertStorage{
		store:         store,
		groupResource: groupResource,
	}
}

func (s *AlertStorage) Kind() string {
	return "Alerts"
}

func (s *AlertStorage) New() runtime.Object {
	return &apiserver.Alert{}
}

func (s *AlertStorage) NewList() runtime.Object {
	return &apiserver.AlertList{}
}

func (s *AlertStorage) NamespaceScoped() bool {
	return false
}

func (s *AlertStorage) NewExtraListOptions() runtime.Object {
	return &apiserver.AlertListOptions{}
}

func (s *AlertStorage) NeedParameter() bool {
	return true
}

func (s *AlertStorage) List(ctx context.Context, options *metainternalversion.ListOptions, extraOptions runtime.Object) (runtime.Object, error) {
	o := extraOptions.(*apiserver.AlertListOptions)

	var clauses []string

	if o.StartTimeInSec > o.EndTimeInSec {
		return nil, errors.NewBadRequest("invalid time range")
	}

	if o.StartTimeInSec > 0 && o.EndTimeInSec > 0 {
		clauses = append(clauses, fmt.Sprintf("(alerts.ends_at >= from_unixtime(%d) OR (alerts.ends_at IS NULL)) AND alerts.starts_at <= from_unixtime(%d)", o.StartTimeInSec, o.EndTimeInSec))
	}

	if o.Status != "" {
		clauses = append(clauses, fmt.Sprintf("alerts.status='%s'", o.Status))
	}

	labelClause := getLabelClause(o)
	if labelClause != "" {
		clauses = append(clauses, fmt.Sprintf("alerts.id in (%s)", labelClause))
	}

	var clauseStr string
	if len(clauses) > 0 {
		clauseStr = " where " + strings.Join(clauses, " and ")
	}

	filter := fmt.Sprintf("%s%s%s limit %d offset %d", SELECTALERTS, clauseStr, ORDERBY, utils.GetSize(o.ItemsPerPage), utils.GetStartIndex(o.Page, o.ItemsPerPage))
	rows, err := s.store.DB.DB().Query(fmt.Sprintf(JOINALERTS, filter))
	if err != nil {
		log.Error("query alerts failed", zap.Error(err))
		return nil, errors.NewBadRequest("query alerts failed")
	}

	alerts, err := scan(rows)
	if err != nil {
		return nil, errors.NewBadRequest("query alerts failed")
	}

	return s.getResult(alerts, s.store.DB, clauseStr), nil
}

func getLabelClause(o *apiserver.AlertListOptions) string {
	if o.Cluster == "" && o.Level == "" {
		return ""
	}

	if o.Cluster == "" {
		return fmt.Sprintf("select distinct(alert_id) from alert_labels where label = '%s' and value='%s'", alert.LEVELLABEL, o.Level)
	}

	if o.Level == "" {
		return fmt.Sprintf("select distinct(alert_id) from alert_labels where label = '%s' and value='%s'", alert.INSTANCELABEL, o.Cluster)
	}

	return fmt.Sprintf("select distinct(alert_id) from alert_labels where alert_id in (select distinct(alert_id) from alert_labels where label = '%s' and value='%s') and label='%s' and value = '%s'", alert.LEVELLABEL, o.Level, alert.INSTANCELABEL, o.Cluster)
}

func (s *AlertStorage) getAllClusters() map[string]string {
	opts := metav1.ListOptions{}
	opts.LabelSelector = fmt.Sprintf("%s=%s", operator.CreatedByLabelKey, operator.CreatedByLabelVal)

	m := make(map[string]string)

	tcl, err := s.store.Controller().GetTidbClusters(metav1.NamespaceAll, opts)
	if err != nil {
		log.Warn("failed to get instance list", zap.Error(err))
		return m
	}

	stream.FromArray(tcl.Items).Each(func(tc v1alpha1.TidbCluster) {
		clusterName, _ := s.store.Controller().GetNameAndTags(&tc)
		m[tc.Name] = clusterName
	})

	return m
}

func (s *AlertStorage) getResult(alerts map[int64][]AlertInDB, db *gorm.DB, clauseStr string) *apiserver.AlertList {
	list := &apiserver.AlertList{}
	if len(alerts) == 0 {
		return list
	}

	m := s.getAllClusters()

	stream.FromMapEntries(alerts).Each(func(entry stream.MapEntry) {
		name := fmt.Sprintf("alert%d", entry.Key.(reflect.Value).Int())
		alert := &apiserver.Alert{
			ObjectMeta: v1.ObjectMeta{Name: name, UID: types.UID(name)},
		}

		stream.FromArray(entry.Value.([]AlertInDB)).Each(func(a AlertInDB) {
			alert.StartTime = strfmt.DateTime(a.StartAt).String()
			if a.EndAt != nil {
				end := strfmt.DateTime(*a.EndAt).String()
				alert.EndTime = end
			}
			alert.Status = a.Status

			if f, exist := AlertLables[a.Label]; exist {
				f(alert, a.Value)
			}
		})

		annotation, err := queryAnnotation(db, entry.Key.(reflect.Value).Int())
		if err != nil {
			log.Error("get annotation failed", zap.Error(err))
		}

		alert.Description = annotation.Value

		if v, ok := m[alert.Cluster]; ok {
			alert.ClusterName = v
		}

		list.Items = append(list.Items, alert)
	})

	total, err := getTotal(db, clauseStr)
	if err != nil {
		log.Error("get count failed", zap.Error(err))
		return list
	}

	list.Total = total

	sort.Slice(list.Items, func(i, j int) bool {
		ti, err := time.Parse(time.RFC3339, list.Items[i].StartTime)
		if err != nil {
			log.Warn("parse time failed", zap.Error(err))
			return false
		}

		tj, err := time.Parse(time.RFC3339, list.Items[j].StartTime)
		if err != nil {
			log.Warn("parse time failed", zap.Error(err))
			return false
		}

		return ti.After(tj)
	})

	return list
}

func getTotal(db *gorm.DB, clauseStr string) (int64, error) {
	rows, err := db.DB().Query(fmt.Sprintf("%s%s", COUNTALERTS, clauseStr))
	if err != nil {
		return 0, err
	}

	var total int64
	for rows.Next() {
		if err := rows.Scan(&total); err != nil {
			return 0, err
		}
	}

	return total, nil
}

var (
	JOINALERTS       = "select a.id, a.starts_at, a.ends_at, a.status, alert_labels.label, alert_labels.value from (%s) a inner join alert_labels on a.id=alert_labels.alert_id "
	SELECTALERTS     = fmt.Sprintf("select alerts.id, alerts.starts_at, alerts.ends_at, alerts.status from alerts")
	COUNTALERTS      = fmt.Sprintf("select count(*) from alerts")
	SELECTANNOTATION = fmt.Sprintf("select alert_id, annotation, value from alert_annotations where annotation = '%s' ", alert.DESCRIPTIONANNOTATION) + " and alert_id = %d "
	AlertLables      = map[string]func(alert *apiserver.Alert, value string){
		alert.INSTANCELABEL: func(alert *apiserver.Alert, value string) {
			alert.Cluster = value
		},
		alert.EXPRLABEL: func(alert *apiserver.Alert, value string) {
			alert.Rule = value
		},

		alert.LEVELLABEL: func(alert *apiserver.Alert, value string) {
			alert.Level = value
		},
		alert.ALERTNAMELABEL: func(alert *apiserver.Alert, value string) {
			alert.AlertName = value
		},
	}

	ORDERBY = " order by alerts.starts_at desc "
)

func queryAnnotation(db *gorm.DB, alertID int64) (AnnotationInDB, error) {
	rows, err := db.DB().Query(fmt.Sprintf(SELECTANNOTATION, alertID))
	if err != nil {
		return AnnotationInDB{}, err
	}

	var a AnnotationInDB
	for rows.Next() {
		err := rows.Scan(
			&a.AlertID,
			&a.Annotation,
			&a.Value,
		)

		if err != nil {
			return a, err
		}
	}

	return a, nil
}

func scan(rows *sql.Rows) (map[int64][]AlertInDB, error) {
	alerts := make(map[int64][]AlertInDB)
	for rows.Next() {
		var a AlertInDB
		err := rows.Scan(
			&a.ID,
			&a.StartAt,
			&a.EndAt,
			&a.Status,
			&a.Label,
			&a.Value,
		)

		if err != nil {
			return nil, err
		}

		_, exist := alerts[a.ID]
		if !exist {
			alerts[a.ID] = make([]AlertInDB, 0)
		}

		alerts[a.ID] = append(alerts[a.ID], a)
	}

	return alerts, nil
}

type AlertInDB struct {
	ID      int64
	StartAt time.Time
	EndAt   *time.Time
	Status  string
	Label   string
	Value   string
}

type AnnotationInDB struct {
	AlertID    int64
	Annotation string
	Value      string
}