package service

import (
	"bytes"
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"io"
	"io/ioutil"
	"math/big"
	"net/http"
	"os"
	"reflect"
	"strconv"
	"strings"
	"time"

	"cloud.google.com/go/bigquery"
	"github.com/spf13/viper"
	"go.uber.org/zap"

	"github.com/pingcap-inc/data-ingestion-tools/internal/itcronjob/store"
	"github.com/pingcap-inc/data-ingestion-tools/model"
	"github.com/pingcap-inc/data-ingestion-tools/pkg/errors"
	"github.com/pingcap-inc/data-ingestion-tools/pkg/log"
)

type SalesforceSrv interface {
	StartSync(ctx context.Context)
	SyncSalesforce(ctx context.Context) error
	FixUp(ctx context.Context)
}

type salesforceService struct {
	store store.Factory
}

var _ SalesforceSrv = (*salesforceService)(nil)

func newSalesforceSrv(str *service) *salesforceService {
	return &salesforceService{store: str.store}
}

const (
	defaultProduct   = "01t5g000000liLxAAI"
	defaultPriceBook = "01s5g000001cOA9AAM"
	objectName       = "metadb-schema/monthly_order.csv"
	bqDatasetId      = "model_of_monthly_cloud_revrec_confirm"
	bqTableId        = "dwm_monthly_order_info_calculation12_of_revrec_tenant_lite_version"
	pdfPath          = "invoice.pdf"
)

var (
	sfToken = ""
	sem     = make(chan int, 1)
)

func (sfs *salesforceService) StartSync(ctx context.Context) {
	log.L(ctx).Info("start sync to salesforce")

	select {
	case sem <- 1:
		err := sfs.SyncSalesforce(ctx)
		if err != nil {
			log.L(ctx).Error("sync to salesforce failed", zap.Error(err))

			_ = FreshRunStatus(ctx, 1, err)

			return
		}

		_ = FreshRunStatus(ctx, 0, nil)

		<-sem
	default:
		_ = FreshRunStatus(ctx, 1, errors.New("sync to salesforce is running"))
	}
}

func (sfs *salesforceService) SyncSalesforce(ctx context.Context) error {
	log.L(ctx).Info("start sync to salesforce")
	sfToken = ""

	err := sfs.GetSfToken(ctx)
	if err != nil {
		return err
	}

	lastMonth := GetLastMonth()

	orders, err := sfs.store.CreatedOrder().ListCurrentMonth(ctx, lastMonth)
	if err != nil {
		log.L(ctx).Errorf("get created orders failed", zap.Error(err))

		return err
	}
	for _, order := range orders {
		if order.Sync == 1 {
			continue
		}
		sfs.sfOrder(ctx, order)
	}

	revrecs, err := sfs.store.CreatedRevrec().ListCurrentMonth(ctx, lastMonth)
	if err != nil {
		log.L(ctx).Errorf("get created revrecs failed", zap.Error(err))

		return err
	}

	for _, revrec := range revrecs {
		if revrec.Sync == 1 || revrec.Type == 0 {
			continue
		}
		sfs.sfRevrec(ctx, revrec)
	}

	collections, err := sfs.store.CreatedCollection().ListCurrentMonth(ctx, lastMonth)
	if err != nil {
		log.L(ctx).Errorf("get created collections failed", zap.Error(err))

		return err
	}

	for _, collection := range collections {
		if collection.Sync == 1 {
			continue
		}
		sfs.sfCollection(ctx, collection)
	}

	// prod
	// saf := viper.GetString("bq.sa")
	// f, err := os.Stat(saf)
	// if err != nil || f.IsDir() {
	// 	log.L(ctx).Error("Failed to get bg sa file: " + err.Error())
	//
	// 	return err
	// }

	// bqSrv, err := bigquery.NewClient(ctx, GcpProject, option.WithCredentialsFile(saf))
	// debug
	bqSrv, err := bigquery.NewClient(ctx, GcpProject)
	if err != nil {
		log.L(ctx).Error("Failed to get bigquery client: " + err.Error())

		return err
	}

	records, err := GetBq(ctx, bqSrv, false)
	if err != nil {
		return err
	}

	oderMap := make(map[string]*model.CreatedOrder)
	revrecMap := make(map[string][]*model.CreatedRevrec)
	collectionMap := make(map[string][]*model.CreatedCollection)

	// for _, order := range orders {
	// 	if order.Sync == 0 {
	// 		continue
	// 	}
	// 	oderMap[order.BillMonth+order.OriginalTenantID] = order
	// }
	//
	// for _, revrec := range revrecs {
	// 	if revrec.Sync == 0 {
	// 		continue
	// 	}
	// 	revrecMap[revrec.BillMonth+revrec.OriginalTenantID] = append(revrecMap[revrec.BillMonth+revrec.OriginalTenantID], revrec)
	// }

	for _, collection := range collections {
		if collection.Sync == 0 {
			continue
		}
		collectionMap[collection.BillMonth+collection.OriginalTenantID] = append(collectionMap[collection.BillMonth+collection.OriginalTenantID], collection)
	}

	for _, record := range records {
		updateBqRecord(ctx, record, oderMap, revrecMap, collectionMap)
	}

	err = WriteBqData(ctx, bqSrv, records)
	if err != nil {
		return err
	}

	// 更新自动创建的collection
	err = getAndDelBiRecords(ctx, collectionTableId)
	if err != nil {
		log.L(ctx).Errorf("get and del bi records failed", zap.Error(err))

		return err
	}

	var cls []model.CreatedCollection
	for _, collection := range collections {
		cls = append(cls, *collection)
	}

	list, err := sfs.store.BqOrder().List(ctx)
	if err != nil {
		log.L(ctx).Errorf("Failed to get order data from db: %v", err)

		return err
	}
	orderMap := make(map[string]string)
	for _, order := range list {
		orderMap[order.OrderID] = order.OrderName
	}

	err = writeBiCollections(ctx, cls, orderMap)
	if err != nil {
		log.L(ctx).Errorf("write bi collections failed", zap.Error(err))

		return err
	}

	log.L(ctx).Info("sync to salesforce success")

	return nil
}

type sfTokenResp struct {
	AccessToken string `json:"access_token"`
}

func (sfs *salesforceService) sfOrder(ctx context.Context, order *model.CreatedOrder) {
	log.L(ctx).Infof("create order to salesforce:%s--%s", order.BillMonth, order.OriginalTenantID)

	orderID, orderName, err := sfs.createOrder(ctx, order)
	if err != nil {
		log.L(ctx).Errorf("create order failed", zap.Error(err))

		return
	}
	order.NewOrderNumber = orderName
	order.NewOrderID = orderID
	order.Sync = 1

	err = sfs.store.CreatedOrder().Update(ctx, order)
	if err != nil {
		log.L(ctx).Errorf("update order failed", zap.Error(err))

		return
	}
	orderJson, _ := json.Marshal(order)
	err = sfs.saveAuditLog(ctx, 0, order.NewOrderNumber, string(orderJson))
	if err != nil {
		log.L(ctx).Errorf("save audit log failed", zap.Error(err))

		return
	}

	opID, err := sfs.createOrderProduct(ctx, order)
	if err != nil {
		log.L(ctx).Errorf("create order product failed", zap.Error(err))

		return
	}

	// 更新order状态
	url := "/sobjects/Order"
	body := map[string]interface{}{
		"Status": "Activated",
	}
	upSfFields(ctx, url, orderID, body)

	err = sfs.saveAuditLog(ctx, 3, opID, string(orderJson))
	if err != nil {
		log.L(ctx).Errorf("save order product log failed", zap.Error(err))

		return
	}

	// 更新自动创建的revrec
	revrecID, err := getRevenueByOrderID(ctx, orderID)
	if err != nil {
		return
	}
	rvrc, err := sfs.store.CreatedRevrec().GetByOrderID(ctx, order.FakeNumber)
	if err != nil {
		log.L(ctx).Errorf("get revrec by order id failed", zap.Error(err))

		return
	}
	url = "/sobjects/Revenue__c"
	bd := map[string]interface{}{
		"Date__c":                             rvrc.Date,
		"Amount__c":                           rvrc.Amount,
		"Billing_End_Date__c":                 rvrc.BillingEndDate,
		"Status__c":                           "Confirmed",
		"Product__c":                          defaultProduct,
		"TiDB_Cloud_Dedicated_Amount_Raw__c":  rvrc.NewOrderDedicatedAmount,
		"TiDB_Cloud_Serverless_Amount_Raw__c": rvrc.NewOrderServerlessAmount,
		"TiDB_Cloud_Support_Amount__c":        rvrc.NewOrderSupportAmount,
	}
	upSfFields(ctx, url, revrecID, bd)

	rvrcName, err := getInfoById(ctx, url, revrecID)
	if err != nil {
		log.L(ctx).Errorf("get revrec by order id failed", zap.Error(err))

		return
	}

	rvrc.Order = orderID
	rvrc.OrderNumber = order.NewOrderNumber
	rvrc.Sync = 1
	rvrc.RevrecID = revrecID
	rvrc.RevrecNumber = rvrcName

	err = sfs.store.CreatedRevrec().Update(ctx, &rvrc)
	if err != nil {
		log.L(ctx).Errorf("update revrec failed", zap.Error(err))

		return
	}

	log.L(ctx).Infof("create order success:%s", orderID)

}

func (sfs *salesforceService) sfRevrec(ctx context.Context, revrec *model.CreatedRevrec) {
	log.L(ctx).Infof("create revrec to salesforce:%s--%s", revrec.BillMonth, revrec.OriginalTenantID)

	revrecID, revrecName, err := sfs.createRevrec(ctx, revrec)
	if err != nil {
		log.L(ctx).Errorf("create revrec failed", zap.Error(err))

		return
	}
	revrec.RevrecNumber = revrecName
	revrec.RevrecID = revrecID
	revrec.Sync = 1
	err = sfs.store.CreatedRevrec().Update(ctx, revrec)
	if err != nil {
		log.L(ctx).Errorf("update revrec failed", zap.Error(err))

		return
	}

	revrecJson, _ := json.Marshal(revrec)
	err = sfs.saveAuditLog(ctx, 1, revrec.RevrecNumber, string(revrecJson))
	if err != nil {
		log.L(ctx).Errorf("save audit log failed", zap.Error(err))
	}

	log.L(ctx).Infof("create revrec success:%s", revrecID)
}

func (sfs *salesforceService) sfCollection(ctx context.Context, collection *model.CreatedCollection) {
	log.L(ctx).Infof("create collection to salesforce:%s--%s", collection.BillMonth, collection.OriginalTenantID)

	collectionNum, err := sfs.createCollection(ctx, collection)
	if err != nil {
		log.L(ctx).Errorf("create collection failed", zap.Error(err))

		return
	}
	collection.CollectionNumber = collectionNum
	collection.Sync = 1
	err = sfs.store.CreatedCollection().Update(ctx, collection)
	if err != nil {
		log.L(ctx).Errorf("update collection failed", zap.Error(err))

		return
	}

	collectionJson, _ := json.Marshal(collection)
	err = sfs.saveAuditLog(ctx, 2, collection.CollectionNumber, string(collectionJson))
	if err != nil {
		log.L(ctx).Errorf("save audit log failed", zap.Error(err))
	}

	log.L(ctx).Infof("create collection success:%s", collectionNum)
}

func updateBqRecord(ctx context.Context, record *model.OrgBq, orderMap map[string]*model.CreatedOrder, revrecMap map[string][]*model.CreatedRevrec, collectionMap map[string][]*model.CreatedCollection) {
	key := record.BillMonth.StringVal + record.OriginalTenantID.StringVal
	autoType := ""

	if order, ok := orderMap[key]; ok {
		record.NewOrderNumber =
			bigquery.NullString{
				StringVal: order.NewOrderNumber,
				Valid:     true,
			}
		autoType += "1"
	}
	if revs, ok := revrecMap[key]; ok {
		if len(revs) > 3 {
			log.L(ctx).Errorf("revrecs length is more than 3", zap.Any("revrecs", revs)) // debug
		}
		for _, revrec := range revs {
			switch revrec.Type {
			case 0:
				record.NewOrderRevrecNumber = bigquery.NullString{
					StringVal: revrec.RevrecNumber,
					Valid:     true,
				}
			case 1:
				record.OrderFormRevrecNumber = bigquery.NullString{
					StringVal: revrec.RevrecNumber,
					Valid:     true,
				}
				autoType += "2"
			case 2:
				record.OrderForm2RevrecNumber = bigquery.NullString{
					StringVal: revrec.RevrecNumber,
					Valid:     true,
				}
				autoType += "3"
			}
		}
	}
	if collections, ok := collectionMap[key]; ok {
		cNumbers := make([]string, 0)

		tpMap := make(map[int]int)
		for _, collection := range collections {
			cNumbers = append(cNumbers, collection.CollectionNumber)

			tpMap[collection.Type] = 1
		}

		for k := range tpMap {
			autoType += strconv.Itoa(k + 4)
		}

		record.CollectionsNumber = bigquery.NullString{
			StringVal: strings.Join(cNumbers, ","),
			Valid:     true,
		}
	}

	// type
	if autoType != "" {
		record.AutoMateType = bigquery.NullString{
			StringVal: autoType,
			Valid:     true,
		}
	}
}

func (sfs *salesforceService) GetSfToken(ctx context.Context) error {
	domain := viper.GetString("sf.tokenDomain")

	url := "https://" + domain + "/services/oauth2/token"

	body := fmt.Sprintf("grant_type=password&client_id=%s&client_secret=%s&username=%s&password=%s",
		viper.GetString("sf.clientId"),
		viper.GetString("sf.clientSecret"),
		viper.GetString("sf.username"),
		viper.GetString("sf.password"),
	)

	resp, err := http.Post(url, "application/x-www-form-urlencoded", bytes.NewReader([]byte(body)))

	if err != nil {
		log.L(ctx).Error("get sf token failed", zap.Error(err))

		return err
	}

	var res sfTokenResp
	err = json.NewDecoder(resp.Body).Decode(&res)
	if err != nil {
		log.L(ctx).Error("unmarshal response failed", zap.Error(err))

		return err
	}

	if res.AccessToken == "" {
		log.L(ctx).Error("get sf token failed", zap.Any("response", res))

		return errors.New("get sf token failed")
	}

	sfToken = res.AccessToken

	return nil
}

func (sfs *salesforceService) createOrder(ctx context.Context, order *model.CreatedOrder) (string, string, error) {
	log.L(ctx).Infof("create order to salesforce:%s--%s", order.BillMonth, order.OriginalTenantID)

	if order.Sync == 1 {
		return order.NewOrderID, order.NewOrderNumber, nil
	}

	url := "/sobjects/order"

	body := model.CreatedOrderReq{
		AccountID:              order.AccountID,
		OrderStartDate:         order.OrderStartDate,
		OrderEndDate:           order.OrderEndDate,
		Status:                 "Draft",
		Tax:                    order.Tax,
		PriceBook2ID:           defaultPriceBook,
		ProductFamily:          "TiDB Cloud",
		AccumulateRevenueToOrg: order.AdjustTenantID,
	}
	b, _ := json.Marshal(body)
	resp, err := sendRequest(ctx, url, http.MethodPost, nil, b)
	if err != nil {
		return "", "", err
	}

	var res model.CreatedOrderResp
	err = json.Unmarshal(resp, &res)
	if err != nil {
		log.L(ctx).Errorf("unmarshal response failed", zap.Error(err), zap.String("response", string(resp)))

		return "", "", err
	}
	if !res.Success {
		log.L(ctx).Errorf("create order failed", res.Errors)

		return "", "", errors.New("create order failed")
	}

	url = "/sobjects/Order"

	name, err := getInfoById(ctx, url, res.ID)
	if err != nil {
		return "", "", err
	}

	log.L(ctx).Info("create order success")

	return res.ID, name, nil
}

func (sfs *salesforceService) createOrderProduct(ctx context.Context, order *model.CreatedOrder) (string, error) {
	log.L(ctx).Infof("create order product to salesforce:%s--%s", order.BillMonth, order.OriginalTenantID)

	url := "/sobjects/Order_Product__c"

	body := model.CreatedOrderProductReq{
		Order:            order.NewOrderID,
		ServiceStartDate: order.OrderStartDate,
		ServiceEndDate:   order.OrderEndDate,
		NetFee:           order.NetFee,
		Product:          defaultProduct,
		Quantity:         "1.0",
	}

	b, _ := json.Marshal(body)

	resp, err := sendRequest(ctx, url, http.MethodPost, nil, b)

	var res model.CommonSalesResp
	err = json.Unmarshal(resp, &res)
	if err != nil {
		log.L(ctx).Errorf("unmarshal response failed", zap.Error(err), zap.String("response", string(resp)))

		return "", err
	}
	if !res.Success {
		log.L(ctx).Error("create order product failed", zap.Any("response", res))

		return "", errors.New("create order product failed")
	}

	return res.ID, err
}

func (sfs *salesforceService) createRevrec(ctx context.Context, revrec *model.CreatedRevrec) (string, string, error) {
	log.L(ctx).Infof("create revrec to salesforce:%s--%s", revrec.BillMonth, revrec.OriginalTenantID)

	url := "/sobjects/Revenue__c"

	if revrec.Type == 0 {
		order, err := sfs.store.CreatedOrder().GetByFakeNumber(ctx, revrec.Order)
		if err != nil {
			log.L(ctx).Errorf("get order failed", zap.Error(err))

			return "", "", err
		}

		revrec.Order = order.NewOrderID
		revrec.OrderNumber = order.NewOrderNumber
	}
	body := model.CreatedRevrecReq{
		Order:                    revrec.Order,
		Date:                     revrec.Date,
		Amount:                   revrec.Amount,
		TiDBCloudBillingEndDate:  revrec.BillingEndDate,
		Status:                   "Confirmed",
		Product:                  defaultProduct,
		NewOrderServerlessAmount: revrec.NewOrderServerlessAmount,
		NewOrderDedicatedAmount:  revrec.NewOrderDedicatedAmount,
		NewOrderSupportAmount:    revrec.NewOrderSupportAmount,
	}

	b, _ := json.Marshal(body)

	resp, err := sendRequest(ctx, url, http.MethodPost, nil, b)

	var res model.CommonSalesResp
	err = json.Unmarshal(resp, &res)
	if err != nil {
		log.L(ctx).Errorf("unmarshal response failed", zap.Error(err), zap.String("response", string(resp)))

		return "", "", err
	}
	if !res.Success {
		log.L(ctx).Error("create revrec failed", zap.Any("response", res))

		return "", "", errors.New("create revrec failed")
	}

	Name, err := getInfoById(ctx, url, res.ID)
	if err != nil {
		return "", "", err
	}

	log.L(ctx).Info("create revrec success")

	return res.ID, Name, err
}

func (sfs *salesforceService) createCollection(ctx context.Context, collection *model.CreatedCollection) (string, error) {
	log.L(ctx).Infof("create collection to salesforce:%s--%s", collection.BillMonth, collection.OriginalTenantID)

	url := "/sobjects/Invoice_Collection__c"

	if collection.Type == 0 {
		order, err := sfs.store.CreatedOrder().GetByFakeNumber(ctx, collection.Order)
		if err != nil {
			log.L(ctx).Errorf("get order failed", zap.Error(err))

			return "", err
		}

		collection.Order = order.NewOrderID
		collection.OrderNumber = order.NewOrderNumber
	}
	body := model.CreatedCollectionReq{
		Order:             collection.Order,
		PayableAmount:     collection.PayableAmount,
		InvoiceNumber:     collection.InvoiceNumber,
		PayableDate:       collection.PayableDate,
		ActualBillingDate: collection.ActualBillingDate,
		CollectionsDate:   collection.CollectionsDate,
		CollectionStatus:  collection.CollectionStatus,
		CollectionsAmount: collection.CollectionsAmount,
		StripeInvoicePDF:  collection.Files,
		InvoiceStatus:     "Issued",
	}

	b, _ := json.Marshal(body)

	resp, err := sendRequest(ctx, url, http.MethodPost, nil, b)

	var res model.CommonSalesResp
	err = json.Unmarshal(resp, &res)
	if err != nil {
		log.L(ctx).Error("unmarshal response failed", zap.Error(err), zap.String("response", string(resp)))

		return "", err
	}
	if !res.Success {
		log.L(ctx).Error("create collection failed", zap.Any("response", res))

		return "", errors.New("create collection failed")
	}

	name, err := getInfoById(ctx, url, res.ID)
	if err != nil {
		return "", err
	}

	// upload invoice pdf
	if collection.Files != "" {
		err := downloadFromUrl(collection.Files, pdfPath)
		if err != nil {
			log.L(ctx).Errorf("download pdf failed", zap.Error(err))

			return "", err
		}

		err = uploadPDF(ctx, pdfPath, res.ID)
		if err != nil {
			log.L(ctx).Errorf("upload pdf failed", zap.Error(err))

			return "", err
		}
	}

	log.L(ctx).Info("create collection success")

	return name, err
}

func getRevenueByOrderID(ctx context.Context, orderID string) (string, error) {
	url := "/query"

	q := "select id, LastModifiedDate from Revenue__c where Order__c = '" + orderID + "' Order by LastModifiedDate desc limit 1"
	params := map[string]string{
		"q": q,
	}
	resp, err := sendRequest(ctx, url, http.MethodGet, params, nil)
	if err != nil {
		log.L(ctx).Errorf("get revenue by order id failed", zap.Error(err))

		return "", err
	}

	var res model.CommonQueryResp
	err = json.Unmarshal(resp, &res)
	if err != nil {
		log.L(ctx).Errorf("unmarshal response failed", zap.Error(err), zap.String("response", string(resp)))

		return "", err
	}

	if len(res.Records) == 0 {
		log.L(ctx).Errorf("get revenue by order id failed", zap.String("response", string(resp)))

		return "", nil
	}

	return res.Records[0].ID, nil
}

func getInfoById(ctx context.Context, url, id string) (string, error) {
	log.L(ctx).Infof("get info by id from salesforce:%s", id)

	if id != "" {
		url += "/" + id
	}
	resp, err := sendRequest(ctx, url, http.MethodGet, nil, nil)
	if err != nil {
		return "", err
	}

	var res model.CommonGetResp
	err = json.Unmarshal(resp, &res)
	if err != nil {
		log.L(ctx).Errorf("unmarshal response failed", zap.Error(err), zap.String("response", string(resp)))

		return "", err
	}

	log.L(ctx).Info("get order success")

	return res.Name, nil
}

func upSfFields(ctx context.Context, url, id string, body map[string]interface{}) {
	log.L(ctx).Infof("update status by id from salesforce:%s", id)

	if id != "" {
		url += "/" + id
	}

	b, _ := json.Marshal(body)

	resp, err := sendRequest(ctx, url, http.MethodPatch, nil, b)

	fmt.Println(string(resp))
	if err != nil {
		log.L(ctx).Errorf("update status failed %s", id, zap.Error(err))
	}

	log.L(ctx).Info("update status success")
}

func sendRequest(ctx context.Context, url string, method string, params map[string]string, body []byte) ([]byte, error) {
	log.L(ctx).Infof("send request to salesforce:%s", url)

	client := &http.Client{}

	urlBase := "https://" + viper.GetString("sf.domain") + "/services/data/v59.0"

	req, err := http.NewRequest(method, urlBase+url, bytes.NewReader(body))

	query := req.URL.Query()
	for k, v := range params {
		query.Add(k, v)
	}
	req.URL.RawQuery = query.Encode()

	req.Header.Set("Authorization", "Bearer "+sfToken)
	req.Header.Set("Content-Type", "application/json")

	if err != nil {
		return nil, err
	}

	var resp *http.Response
	var respBodyStr []byte
	// 失败等待2s重试
	for i := 0; i < 3; i++ {
		resp, err = client.Do(req)
		// code 以2开头的都是成功的
		if err == nil && (resp.StatusCode >= 200 && resp.StatusCode < 300) {
			break
		}
		log.L(ctx).Warnf("retry send request to salesforce:%d-%s", i, url)
		time.Sleep(3 * time.Second)

		if i == 0 {
			respBodyStr, _ = io.ReadAll(resp.Body)
		}
		if i == 2 {
			errCode := resp.StatusCode
			return nil, fmt.Errorf("send request to salesforce failed:%s  code:%v  err:%s", url, errCode, string(respBodyStr))
		}
	}

	defer resp.Body.Close()

	b, _ := io.ReadAll(resp.Body)

	log.L(ctx).Infof("send request to salesforce success:%s", string(b))

	return b, nil
}

func SchemaFromFile(filename string) (bigquery.Schema, error) {
	content, err := ioutil.ReadFile(filename)
	if err != nil {
		return nil, err
	}
	var schema bigquery.Schema
	if err := json.Unmarshal(content, &schema); err != nil {
		return nil, err
	}
	return schema, nil
}

// ValueToString 把 bigquery.NullX 类型的字段转换成相应的字符串表示
func ValueToString(value reflect.Value) string {
	switch value.Interface().(type) {
	case bigquery.NullString:
		ns := value.Interface().(bigquery.NullString)
		if ns.Valid {
			return ns.StringVal
		}
	case bigquery.NullDate:
		nd := value.Interface().(bigquery.NullDate)
		if nd.Valid {
			return nd.Date.String() // 或者您希望的任何其他格式
		}
	case bigquery.NullInt64:
		ni := value.Interface().(bigquery.NullInt64)
		if ni.Valid {
			return strconv.FormatInt(ni.Int64, 10)
		}
	case bigquery.NullFloat64:
		nf := value.Interface().(bigquery.NullFloat64)
		if nf.Valid {
			// float64
			return strconv.FormatFloat(nf.Float64, 'f', -1, 64)
		}
	case bigquery.NullBool:
		nb := value.Interface().(bigquery.NullBool)
		if nb.Valid {
			return strconv.FormatBool(nb.Bool)
		}
	case *big.Rat:
		r := value.Interface().(*big.Rat)
		if r != nil {
			return r.FloatString(2)
		}
	}
	return ""
}

// CsvRecordFromStruct Generate CSV record based on the CSV headers and struct tags.
func CsvRecordFromStruct(record interface{}, csvHeaders []string) ([]string, error) {
	rValue := reflect.ValueOf(record)
	rType := rValue.Type()

	var csvRecord []string

	// Iterate over the CSV headers to establish order
	for _, header := range csvHeaders {
		var found bool
		// Iterate over the struct fields to find the matching header
		for i := 0; i < rValue.NumField(); i++ {
			fieldValue := rValue.Field(i)
			fieldType := rType.Field(i)
			// Check if the struct tag matches the header
			if bigqueryTag := fieldType.Tag.Get("bigquery"); bigqueryTag == header {
				csvRecord = append(csvRecord, ValueToString(fieldValue))
				found = true
				break
			}
		}
		if !found {
			// The struct does not have a field for the CSV header.
			return nil, fmt.Errorf("CSV header '%s' does not match any struct tags", header)
		}
	}

	return csvRecord, nil
}

func uploadPDF(ctx context.Context, pdfPath, collectID string) error {
	url := "/sobjects/ContentVersion"

	pdfBytes, err := ioutil.ReadFile(pdfPath)
	if err != nil {
		log.L(ctx).Errorf("Error reading PDF file:", zap.Error(err))

		return err
	}

	// Encode the PDF file to base64
	encodedPdf := base64.StdEncoding.EncodeToString(pdfBytes)

	body := map[string]interface{}{
		"Title":                  "invoice",
		"PathOnClient":           "invoice.pdf",
		"VersionData":            encodedPdf,
		"FirstPublishLocationId": collectID,
		"ContentLocation":        "S",
	}

	jsonBody, _ := json.Marshal(body)
	_, err = sendRequest(ctx, url, http.MethodPost, nil, jsonBody)
	if err != nil {
		log.L(ctx).Errorf("upload pdf failed", zap.Error(err))

		return err
	}

	return nil
}

func downloadFromUrl(url string, filepath string) error {
	resp, err := http.Get(url)
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	file, err := os.Create(filepath)
	if err != nil {
		return err
	}
	defer file.Close()

	_, err = io.Copy(file, resp.Body)
	if err != nil {
		return err
	}

	return nil
}

func (sfs *salesforceService) saveAuditLog(ctx context.Context, typ int, objNum, extraInfo string) error {
	auLog := &model.AuditLog{
		CreatedAt:    time.Now(),
		Action:       typ,
		ObjectNumber: objNum,
		ExtraInfo:    extraInfo,
	}
	err := sfs.store.AuditLog().Create(auLog)
	if err != nil {
		log.L(ctx).Errorf("save audit log failed", zap.Error(err))

		return err
	}

	return nil
}

// fix
func (sfs *salesforceService) FixUp(ctx context.Context) {
	lastMonth := GetLastMonth()
	records, err := sfs.store.CreatedRevrec().ListCurrentMonth(ctx, lastMonth)
	if err != nil {
		log.L(ctx).Errorf("get created revrecs failed", zap.Error(err))

		return
	}
	sfs.GetSfToken(ctx)
	url := "/sobjects/Revenue__c"
	for _, record := range records {
		if record.Sync == 0 {
			sfs.sfRevrec(ctx, record)
		} else {
			bd := map[string]interface{}{
				"Date__c":                             record.Date,
				"Amount__c":                           record.Amount,
				"Billing_End_Date__c":                 record.BillingEndDate,
				"Status__c":                           "Confirmed",
				"Product__c":                          defaultProduct,
				"TiDB_Cloud_Dedicated_Amount_Raw__c":  record.NewOrderDedicatedAmount,
				"TiDB_Cloud_Serverless_Amount_Raw__c": record.NewOrderServerlessAmount,
				"TiDB_Cloud_Support_Amount__c":        record.NewOrderSupportAmount,
			}
			upSfFields(ctx, url, record.RevrecID, bd)
		}
		log.L(ctx).Infof("fix revrec success:%s", record.RevrecID)
	}
}