package gcp

import (
	"context"
	"fmt"
	"os"
	"text/template"

	"cloud.google.com/go/bigquery"
	"github.com/pingcap/log"
	gcpauth "github.com/tidbcloud/cost-collect/pkg/cloud/gcp/auth"
	"github.com/tidbcloud/cost-collect/pkg/global"
	"google.golang.org/api/iterator"
	"google.golang.org/api/option"
)

type BigQueryParam struct {
	global.Request

	Config
}

type Config struct {
	AWSBillTemplate       *template.Template
	GCPBillTemplate       *template.Template
	PreAggregatedAWSTable string
	PreAggregatedGCPTable string
}

type BigQueryRows []map[string]bigquery.Value

type BigQueryClient struct {
	client   *bigquery.Client
	location string
	Config   Config
}

// NewBigQueryClient new a bigquery client
func NewBigQueryClient(providerProjectID string, location string, sharedParam Config) (*BigQueryClient, error) {
	if providerProjectID == "" || location == "" ||
		sharedParam.PreAggregatedAWSTable == "" || sharedParam.PreAggregatedGCPTable == "" {
		err := fmt.Errorf("gcp config error")
		return nil, err
	}
	ctx := context.Background()
	var opts []option.ClientOption
	if os.Getenv("GOOGLE_APPLICATION_CREDENTIALS") == "" && os.Getenv("USE_DEFAULT_GCP_CREDENTIAL") != "true" {
		// init opts for kubernetes iam
		var err error
		err = gcpauth.InitWebIdentityEnv()
		if err != nil {
			return nil, err
		}
		opts, err = gcpauth.DefaultClientOptions()
		if err != nil {
			return nil, err
		}
	}
	client, err := bigquery.NewClient(ctx, providerProjectID, opts...)
	if err != nil {
		return nil, fmt.Errorf("bigquery.NewClient: %w", err)
	}

	c := BigQueryClient{
		client:   client,
		location: location,
		Config:   sharedParam,
	}
	return &c, nil
}

func (t BigQueryClient) query(ctx context.Context, sql string) (BigQueryRows, error) {
	q := t.client.Query(sql)
	q.Location = t.location

	job, err := q.Run(ctx)
	if err != nil {
		return nil, err
	}
	// wait job
	status, err := job.Wait(ctx)
	if err != nil {
		return nil, err
	}
	// check status
	if err := status.Err(); err != nil {
		return nil, err
	}

	var rows BigQueryRows

	it, err := job.Read(ctx)
	if err != nil {
		return nil, err
	}
	for {
		var row map[string]bigquery.Value
		err2 := it.Next(&row)
		if err2 == iterator.Done {
			break
		}
		if err2 != nil {
			return nil, err2
		}
		rows = append(rows, row)
	}
	return rows, nil
}

func (t BigQueryClient) ExecQuery(ctx context.Context, sql string) (BigQueryRows, error) {
	rows, err := t.query(ctx, sql)
	if err != nil {
		log.S().Errorf("sql:%s, err: %s", sql, err)
		return nil, err
	}
	return rows, err
}