package bigquery

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io/ioutil"
	"log"
	"path/filepath"
	"reflect"
	"strings"
	"text/template"
	"time"

	googlebq "cloud.google.com/go/bigquery"
	"cloud.google.com/go/storage"
	"google.golang.org/api/iterator"
	"google.golang.org/api/option"

	multierror "github.com/hashicorp/go-multierror"

	"github.com/pingcap/crm-bot/pkg/config"
	"github.com/pingcap/crm-bot/salesforce"
)

type opfn func(*googlebq.TableMetadata)

var models []IModel

func init() {
	models = append(models, &CRMAccount{}, &PingcapSalesforceLead{}, &PingcapSalesforceOpportunity{},
		&PingcapSalesforceOrder{}, &PingcapSalesforceCollection{}, &PingcapSalesforceTask{},
		&PingcapSalesforceRevrec{}, &PingcapSalesforcePoc{}, &PingcapSalesforceOpportunityFieldHistory{})
}

func SetExpire(expire time.Time) opfn {
	return func(metadata *googlebq.TableMetadata) {
		metadata.ExpirationTime = expire
	}
}

func CreateTable(client *googlebq.Client, dataset string,
	tablename string, tableschema interface{}, ops ...opfn) error {
	ds := client.Dataset(dataset)
	if ds == nil {
		return errors.New(fmt.Sprintf("Dataset %s not exist", dataset))
	}

	ctx := context.Background()
	schema, err := googlebq.InferSchema(tableschema)
	if err != nil {
		return err
	}

	metadata := &googlebq.TableMetadata{
		Schema: schema,
	}
	for _, op := range ops {
		op(metadata)
	}

	err = ds.Table(tablename).Create(ctx, metadata)
	if err != nil {
		return err
	}
	return nil
}

func InsertTable(client *googlebq.Client, dataset string, tablename string, rows interface{}) error {
	ds := client.Dataset(dataset)
	if ds == nil {
		return errors.New(fmt.Sprintf("Dataset %s not exist", dataset))
	}

	tb := ds.Table(tablename)
	if tb == nil {
		return errors.New(fmt.Sprintf("Table %s not exist", tablename))
	}

	u := tb.Inserter()
	if err := u.Put(context.Background(), rows); err != nil {
		return err
	}
	return nil
}

func Query(client *googlebq.Client, query_string string, itemType reflect.Type) (interface{}, error) {
	q := client.Query(query_string)
	it, err := q.Read(context.Background())
	if err != nil {
		return nil, err
	}

	slice := reflect.New(reflect.SliceOf(itemType)).Elem()
	for {
		val := reflect.New(itemType)
		err = it.Next(val.Interface())
		if err == iterator.Done {
			break
		}

		if err != nil {
			return nil, err
		}
		slice = reflect.Append(slice, val.Elem())
	}

	return slice.Interface(), nil
}

// Do BigQuery query, query_string can be generate by template
func Exec(client *googlebq.Client, query_string string) error {
	log.Println("merge sql", query_string)
	q := client.Query(query_string)
	ctx := context.Background()
	job, err := q.Run(ctx)
	if err != nil {
		return err
	}

	result, err := job.Wait(ctx)
	if err != nil {
		return err
	}

	if err = result.Err(); err != nil {
		return err
	}

	return nil
}

func Init(tpls []BigQuerySFTpl) error {
	ctx := context.Background()
	client, err := googlebq.NewClient(ctx, config.GetString("bigquery.projectid"),
		option.WithCredentialsFile(config.GetString("bigquery.credentialsfilepath")))
	if err != nil {
		return err
	}
	defer client.Close()

	datasetId := config.GetString("bigquery.datasetid")
	for _, m := range models {
		err = CreateTable(client, datasetId, m.TableName(), m)
		if err != nil {
			log.Println("[error] create table", m.TableName(), err)
		}
	}

	sfDatas := querySalesforceData(tpls, true)

	time.Sleep(10 * time.Second)
	for table, rowarr := range sfDatas {
		for _, rows := range rowarr {
			err = InsertTable(client, datasetId, table, rows)
			if err != nil {
				log.Println("[error] insert table ", table, err)
			}
			time.Sleep(8 * time.Second)
		}
	}

	return nil
}

func SalesforceOppoFieldHistory() error {
	ctx := context.Background()
	client, err := googlebq.NewClient(ctx, config.GetString("bigquery.projectid"),
		option.WithCredentialsFile(config.GetString("bigquery.credentialsfilepath")))
	if err != nil {
		return err
	}
	defer client.Close()

	datasetId := config.GetString("bigquery.datasetid")
	model := PingcapSalesforceOpportunityFieldHistory{}
	CreateTable(client, datasetId, model.TableName(), model)

	time.Sleep(10 * time.Second)

	historys, err := salesforce.GetOpportunityFieldHistory(true)
	if err != nil {
		return err
	}
	bqHistorys, err := ModelConvert(historys, reflect.TypeOf([]PingcapSalesforceOpportunityFieldHistory{}))
	if err != nil {
		return err
	}
	return InsertTable(client, datasetId, model.TableName(), bqHistorys)
}

func UpdateSalesforce(tpls []BigQuerySFTpl) error {
	ctx := context.Background()
	client, err := googlebq.NewClient(ctx, config.GetString("bigquery.projectid"),
		option.WithCredentialsFile(config.GetString("bigquery.credentialsfilepath")))
	if err != nil {
		return err
	}
	defer client.Close()

	datasetId := config.GetString("bigquery.datasetid")
	sfDatas := querySalesforceData(tpls, false)

	// create tmp table
	for _, m := range models {
		tableName := m.TableName() + "_tmp"
		err := CreateTable(client, datasetId, tableName,
			m, SetExpire(time.Now().Add(10*time.Minute)))
		if err != nil {
			log.Println("[error] create table", tableName, err)
		}
	}

	// sleep for table flush
	time.Sleep(15 * time.Second)

	for tableName, rowarr := range sfDatas {
		tableName := tableName + "_tmp"
		for _, rows := range rowarr {
			err = InsertTable(client, datasetId, tableName, rows)
			if err != nil {
				log.Println("[error] insert table ", tableName, err)
			}
			time.Sleep(3 * time.Second)
		}
	}

	funcMap := template.FuncMap{
		"dec": func(i int) int {
			return i - 1
		},
	}
	tplbytes, _ := ioutil.ReadFile("./tpl/merge.tpl")
	tpl, err := template.New("merge").Funcs(funcMap).Parse(string(tplbytes))
	if err != nil {
		return err
	}

	// do BigQuery merge exec
	for _, model := range models {
		sql, err := genMergeSQL(model, tpl, datasetId)
		if err != nil {
			return err
		}

		log.Println(sql)
		err = Exec(client, sql)
		if err != nil {
			return err
		}
	}

	return nil
}

func genMergeSQL(model IModel, tpl *template.Template, datesetId string) (string, error) {
	reftype := reflect.TypeOf(model)
	reftype = reftype.Elem()
	fields := make([]interface{}, reftype.NumField())
	for i := 0; i < reftype.NumField(); i++ {
		fields[i] = reftype.Field(i).Name
	}

	buf := bytes.NewBuffer(nil)
	err := tpl.Execute(buf, struct {
		DataSet     string
		SourceTable string
		TmpTable    string
		PrimaryKey  string
		Fields      []interface{}
	}{datesetId, model.TableName(),
		model.TableName() + "_tmp", model.PrimaryId(), fields})
	if err != nil {
		return "", err
	}

	return buf.String(), nil
}

func UpdateTablesByGCS(prefix string) error {
	errs := new(multierror.Error)
	ctx := context.Background()
	client, err := googlebq.NewClient(ctx, config.GetString("bigquery.projectid"),
		option.WithCredentialsFile(config.GetString("bigquery.credentialsfilepath")))
	if err != nil {
		return err
	}
	defer client.Close()

	storageClient, err := storage.NewClient(ctx)
	if err != nil {
		return err
	}
	defer storageClient.Close()

	buckets := []GCSBucketInfo{}
	config.UnmarshalKey("gcsbuckets", &buckets)
	for _, bucket := range buckets {
		if bucket.Prefix == prefix {
			it := storageClient.Bucket(bucket.Bucket).Objects(ctx, &storage.Query{
				Prefix: bucket.Prefix,
			})
			dataset := bucket.DataSet

			for {
				attrs, err := it.Next()
				if err == iterator.Done {
					break
				}

				if err != nil {
					multierror.Append(errs, err)
					continue
				}

				tablename := filepath.Base(attrs.Name)
				if strings.Contains(tablename, " ") {
					continue
				}
				tablename = strings.TrimSuffix(strings.ReplaceAll(tablename, "-", "_"), ".csv")
				log.Println("Process table: ", tablename)
				if err = client.Dataset(dataset).
					Table(tablename).Delete(context.Background()); err != nil {
					multierror.Append(errs, err)
				}
				time.Sleep(10 * time.Second)

				gcsRef := googlebq.NewGCSReference(fmt.Sprintf("gs://%s/%s", attrs.Bucket, attrs.Name))
				gcsRef.SkipLeadingRows = 1
				gcsRef.AutoDetect = true
				loader := client.Dataset(dataset).Table(tablename).LoaderFrom(gcsRef)

				job, err := loader.Run(ctx)
				if err != nil {
					multierror.Append(errs, err)
					continue
				}
				status, err := job.Wait(ctx)
				if err != nil {
					multierror.Append(errs, err)
					continue
				}

				if status.Err() != nil {
					multierror.Append(errs, status.Err())
					continue
				}
			}
		}
	}

	return errs.ErrorOrNil()
}

var NonprodAccounts = []string{"013690006788",
	"166681434446",
	"178851224597",
	"180941277423",
	"200683210682",
	"239023060546",
	"378955295993",
	"385595570414",
	"417272767211",
	"448141240262",
	"470438934515",
	"487575298759",
	"494090988690",
	"495580073302",
	"510387698298",
	"552185537300",
	"557537366020",
	"577523860935",
	"646528577659",
	"676546228356",
	"729581434105",
	"801626783489",
	"946646677266",
	"986330900858"}

func CalNonProdAccountsBilling() error {
	errs := new(multierror.Error)
	ctx := context.Background()
	client, err := googlebq.NewClient(ctx, config.GetString("bigquery.projectid"),
		option.WithCredentialsFile(config.GetString("bigquery.credentialsfilepath")))
	if err != nil {
		return err
	}
	defer client.Close()

	err = Exec(client, "delete from `billing.aws_nonprod_billing_gather` where 1=1")
	if err != nil {
		return err
	}

	datas := []AWSNonprodBillingGather{}
	for _, account := range NonprodAccounts {
		log.Println("Process Account:", account)
		slice, err := Query(client, fmt.Sprintf("select '%s' as accountId, concat(year,'-',month) as month, sum(line_item_unblended_cost) as cost from `gcp-digital-bi.aws_nonprod_billing.%s_aws_billing_data` group by 1,2", account, account),
			reflect.TypeOf(AWSNonprodBillingGather{}))
		if err != nil {
			multierror.Append(errs, err)
		}
		datas = append(datas, slice.([]AWSNonprodBillingGather)...)
	}

	err = InsertTable(client, "billing", "aws_nonprod_billing_gather", datas)
	if err != nil {
		multierror.Append(errs, err)
	}

	return errs.ErrorOrNil()
}