package service

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"strconv"
	"strings"
	"time"

	"go.uber.org/zap"

	"cloud.google.com/go/bigquery"
	"data-platform/model"
	"data-platform/pkg/log"
	"encoding/base64"
	"google.golang.org/api/iterator"
)

const (
	create = 1001
	update = 1002
	public = 10001
	market = 10002
)

func (sfs *salesforceService) EventCenter(ctx context.Context, etype string) {
	// Salesforce retired: all event-center syncs (ps_reject / invoice_rewrite /
	// collection / ns_revrec / stripe collection) targeted Salesforce and are
	// disabled. Original dispatch kept (commented) for reference.
	log.L(ctx).Infof("EventCenter disabled (salesforce retired): %s", etype)
	/*
		cmsrv := commonService{
			store: sfs.store,
		}
		defer func() {
			if r := recover(); r != nil {
				errMsg := fmt.Sprintf("EventCenter panicked: %v", r)
				log.L(ctx).Error(errMsg)
				go cmsrv.saveLarkOps(ctx, errorMsg, errMsg)
			}
		}()

		go cmsrv.saveLarkOps(ctx, infoMsg, "start sync "+etype+" to salesforce")

		log.L(ctx).Infof("handle event: %v", etype)

		var err error
		switch etype {
		case "ps_reject":
			err = sfs.handlePsManDayReject(ctx)
		case "invoice_rewrite":
			err = sfs.handleInvoiceRewrite(ctx)
		case "mk_collection":
			err = sfs.handleMarketplaceCollection(ctx)
		case "pb_collection":
			err = sfs.handlePublicCollection(ctx)
		case "ns_revrec":
			err = sfs.handleNSRevrec(ctx)
		case "check_stripe_collection":
			err = sfs.handleCheckStripeCollection(ctx)
		}

		if err != nil {
			log.L(ctx).Errorf("EventCenter failed: %v", err)

			go cmsrv.saveLarkOps(ctx, errorMsg, "EventCenter failed: "+etype+" msg:"+err.Error())

			return
		}

		go cmsrv.saveLarkOps(ctx, successMsg, "sync "+etype+" to salesforce success")

		log.L(ctx).Info("EventCenter success")
	*/
}

func (sfs *salesforceService) handlePsManDayReject(ctx context.Context) error {
	log.L(ctx).Info("start EventCenter")

	psRds, err := sfs.store.Salesforce().ListPsRejectBySync(ctx)
	if err != nil {
		log.L(ctx).Errorf("ListBySync failed: %v", err)

		return err
	}

	if len(psRds) == 0 {
		log.L(ctx).Info("no ps_reject records to process")
		return nil
	}
	comSrv := commonService{
		store: sfs.store,
	}

	for _, psRd := range psRds {
		if psRd.Sync != 0 && psRd.Sync != 1 {
			continue
		}
		sync := 1
		if psRd.SalesforceID == "" {
			log.L(ctx).Error("salesforce id is empty")
			go comSrv.saveLarkOps(ctx, errorMsg, "psRd.SalesforceID is empty. psRd:"+strconv.Itoa(psRd.ID))

			sync = 2
			go sfs.store.Salesforce().UpdateRejectSync(ctx, *psRd, sync)

			continue
		}
		err = updateSfpsdays(ctx, psRd)
		if err != nil {
			log.L(ctx).Errorf("updateSfpsdays failed: %v", err)

			sync = 2
			go sfs.store.Salesforce().UpdateRejectSync(ctx, *psRd, sync)

			return err
		}

		err = sfs.store.Salesforce().UpdateRejectSync(ctx, *psRd, sync)
		if err != nil {
			log.L(ctx).Errorf("UpdateRejectSync failed: %v", err)

			return err
		}
	}

	log.L(ctx).Info("end EventCenter")

	return nil
}

func (sfs *salesforceService) handleInvoiceRewrite(ctx context.Context) error {
	log.L(ctx).Info("handle invoice rewrite")

	comSrv := commonService{
		store: sfs.store,
	}

	invoices, err := sfs.store.Salesforce().ListInvoicesNotSync(ctx)
	if err != nil {
		log.L(ctx).Errorf("ListInvoicesNotSync failed: %v", err)

		return err
	}

	for _, invoice := range invoices {
		if invoice.Sync != 0 && invoice.Sync != 1 {
			continue
		}

		// japan invoice
		if invoice.PdfContent == "" && invoice.PdfContentByte == "" {
			continue
		}
		nowTime := time.Now()
		invoice.SyncTime = nowTime
		invoice.UpdatedTime = nowTime

		if invoice.CollectionID == "" {
			log.L(ctx).Errorf(" handleInvoiceRewrite order id or content is empty" + strconv.Itoa(invoice.ID))
			go comSrv.saveLarkOps(ctx, errorMsg, "handleInvoiceRewrite order or content id is empty. invoice:"+strconv.Itoa(invoice.ID))

			invoice.Sync = 2
			go sfs.store.Salesforce().UpdateInvoice(ctx, *invoice)

			continue
		}

		var pdfContent string

		if invoice.PdfContent != "" {
			// Download PDF from URL
			pdfURL := invoice.PdfContent
			resp, err := http.Get(pdfURL)
			if err != nil {
				log.L(ctx).Errorf("Failed to download PDF from URL: %v", err)
				invoice.Sync = 2
				go sfs.store.Salesforce().UpdateInvoice(ctx, *invoice)
				continue
			}
			defer resp.Body.Close()

			if resp.StatusCode != http.StatusOK {
				log.L(ctx).Errorf("Failed to download PDF, status code: %d", resp.StatusCode)
				invoice.Sync = 2
				go sfs.store.Salesforce().UpdateInvoice(ctx, *invoice)
				continue
			}

			// Read PDF content
			pdfBody, err := io.ReadAll(resp.Body)
			if err != nil {
				log.L(ctx).Errorf("Failed to read PDF content: %v", err)
				invoice.Sync = 2
				go sfs.store.Salesforce().UpdateInvoice(ctx, *invoice)
				continue
			}

			pdfContent = base64.StdEncoding.EncodeToString(pdfBody)
		} else {
			pdfContent = invoice.PdfContentByte
		}

		pdfName := invoice.InvoiceNumberToUse + ".pdf"
		err = updateInvoice(ctx, pdfContent, pdfName, invoice.CollectionID)
		if err != nil {
			log.L(ctx).Errorf("updateInvoice failed: %v", err)

			invoice.Sync = 2
			go sfs.store.Salesforce().UpdateInvoice(ctx, *invoice)

			return err
		}
		invoice.Sync = 1

		err = sfs.store.Salesforce().UpdateInvoice(ctx, *invoice)
		if err != nil {
			log.L(ctx).Errorf("UpdateInvoice failed: %v", err)

			return err
		}
	}

	log.L(ctx).Info("handle invoice rewrite success")

	return nil
}

func (sfs *salesforceService) handleMarketplaceCollection(ctx context.Context) error {
	log.L(ctx).Info("handle marketplace collection")

	// get all collection records
	collectionMap, err := GetInvoiceCollection(ctx)
	if err != nil {
		log.L(ctx).Errorf("GetInvoiceCollection failed: %v", err)

		return err
	}
	comSrv := commonService{
		store: sfs.store,
	}

	createdMc, err := sfs.store.Salesforce().ListCreateMc(ctx)
	if err != nil {
		log.L(ctx).Errorf("ListMc failed: %v", err)

		return err
	}

	for _, mc := range createdMc {
		if mc.Sync != 0 && mc.Sync != 1 {
			continue
		}

		// internal user
		if strings.Contains(mc.AccountName, "内部员工") || strings.Contains(mc.ReasonOfRefund, "内部员工") || strings.Contains(mc.AccountName, "内部用户") || strings.Contains(mc.ReasonOfRefund, "内部用户") {
			log.L(ctx).Errorf("handleMarketplaceCollection internal user" + strconv.Itoa(mc.ID))

			// go comSrv.saveLarkOps(ctx, errorMsg, "handleMarketplaceCollection internal user. mc:"+strconv.Itoa(mc.ID))
			go sfs.store.Salesforce().CreatedMcStatus(ctx, model.MarketplaceCollectionStatus{
				Sync:        3,
				SyncTime:    time.Now(),
				RelatedMkID: mc.MkID,
			})

			continue
		}

		if mc.OrderID == "" {
			log.L(ctx).Errorf("handleMarketplaceCollection order id is empty" + strconv.Itoa(mc.ID))

			go comSrv.saveLarkOps(ctx, errorMsg, "handleMarketplaceCollection order id is empty. mc:"+strconv.Itoa(mc.ID))
			go sfs.store.Salesforce().CreatedMcStatus(ctx, model.MarketplaceCollectionStatus{
				Sync:        2,
				SyncTime:    time.Now(),
				RelatedMkID: mc.MkID,
			})

			continue
		}

		id, name, err := createCollection(ctx, mc, market, create)
		if err != nil {
			log.L(ctx).Errorf("createMcStatus failed: %v", err)
			go sfs.store.Salesforce().CreatedMcStatus(ctx, model.MarketplaceCollectionStatus{
				Sync:             2,
				SyncTime:         time.Now(),
				CollectionID:     id,
				CollectionNumber: name,
				RelatedMkID:      mc.MkID,
			})

			return err
		}

		if id == "" || name == "" {
			log.L(ctx).Errorf("empty id or name")

			return errors.New("empty id or name")
		}

		err = sfs.store.Salesforce().CreatedMcStatus(ctx, model.MarketplaceCollectionStatus{
			Sync:             1,
			SyncTime:         time.Now(),
			CollectionID:     id,
			CollectionNumber: name,
			RelatedMkID:      mc.MkID,
		})
		if err != nil {
			log.L(ctx).Errorf("CreatedMcStatus failed: %v", err)

			return err
		}
	}

	updatedMc, err := sfs.store.Salesforce().ListUpdateMc(ctx)
	if err != nil {
		log.L(ctx).Errorf("ListUpdateMc failed: %v", err)

		return err
	}
	for _, mc := range updatedMc {
		if mc.Sync != 0 && mc.Sync != 1 {
			continue
		}

		mkStatus, err := sfs.store.Salesforce().GetMcStatusByMkID(ctx, mc.MkID)
		if mkStatus == nil && err != nil {
			mkStatus = &model.MarketplaceCollectionStatus{}
		}

		mkStatus.RelatedMkID = mc.MkID
		mkStatus.SyncTime = time.Now()

		if mc.OrderID == "" {
			log.L(ctx).Errorf("handleMarketplaceCollection order id is empty" + strconv.Itoa(mc.ID))

			go comSrv.saveLarkOps(ctx, errorMsg, "handleMarketplaceCollection order id is empty. mc:"+strconv.Itoa(mc.ID))
			if mc.CollectionID == "" {
				mkStatus.Sync = 2
				go sfs.store.Salesforce().SaveMcStatus(ctx, mkStatus)
			}

			continue
		}

		if mc.CreatedCollectionID == "" {
			sfID := collectionMap[mc.CollectionID]
			mc.CreatedCollectionID = sfID
			mc.CreatedCollectionNumber = mc.CollectionID
		}

		id, name, err := createCollection(ctx, mc, market, update)
		if err != nil {
			log.L(ctx).Errorf("createMcStatus failed: %v", err)

			mkStatus.Sync = 2
			go sfs.store.Salesforce().SaveMcStatus(ctx, mkStatus)
			return err
		}

		if id == "" || name == "" {
			log.L(ctx).Errorf("empty id or name")

			return errors.New("empty id or name")
		}

		mkStatus.Sync = 1
		mkStatus.CollectionID = id
		mkStatus.CollectionNumber = name

		err = sfs.store.Salesforce().SaveMcStatus(ctx, mkStatus)
		if err != nil {
			log.L(ctx).Errorf("CreatedMcStatus failed: %v", err)

			return err
		}
	}

	// copy make collection to bak table
	err = sfs.store.Salesforce().CopyMc(ctx)
	if err != nil {
		log.L(ctx).Errorf("CopyMc failed: %v", err)

		return err
	}

	log.L(ctx).Info("handle marketplace collection success")

	return nil
}

func (sfs *salesforceService) handlePublicCollection(ctx context.Context) error {
	log.L(ctx).Info("handle public collection")

	comSrv := commonService{
		store: sfs.store,
	}

	pbcs, err := sfs.store.Salesforce().ListPbcNotSync(ctx)
	if err != nil {
		log.L(ctx).Errorf("ListPbcNotSync failed: %v", err)

		return err
	}

	sfOrders, err := sfs.store.Salesforce().ListOrder(ctx)

	// order_map
	orderMap := make(map[string]string)
	for _, order := range sfOrders {
		if order.SalesforceID != "" {
			orderMap[order.OrderNumberName] = order.SalesforceID
		}
	}

	for _, pbc := range pbcs {
		if pbc.Sync != 0 && pbc.Sync != 1 {
			continue
		}

		nowTime := time.Now()
		pbc.SyncTime = nowTime
		pbc.UpdatedTime = nowTime

		oderNumber := strings.Replace(pbc.OrderNumber, "Old", "", -1)
		sfOrderID := orderMap[oderNumber]

		if sfOrderID == "" {
			log.L(ctx).Errorf("handlePublicCollection order id is empty" + strconv.Itoa(pbc.ID))

			go comSrv.saveLarkOps(ctx, errorMsg, "handlePublicCollection order id is empty. pbc:"+strconv.Itoa(pbc.ID))

			pbc.Sync = 2
			go sfs.store.Salesforce().UpdatePbc(ctx, *pbc)

			continue
		}

		pbc.OrderID = sfOrderID

		op := create
		if pbc.Sync == 1 {
			op = update
		}
		id, name, err := createCollection(ctx, pbc, public, op)
		pbc.CollectionID = id
		pbc.CollectionNumber = name
		pbc.Sync = 1

		if err != nil {
			log.L(ctx).Errorf("updatePublicCollection failed: %v", err)

			pbc.Sync = 2
			go sfs.store.Salesforce().UpdatePbc(ctx, *pbc)

			return err
		}

		err = sfs.store.Salesforce().UpdatePbc(ctx, *pbc)
		if err != nil {
			log.L(ctx).Errorf("UpdatePbc failed: %v", err)

			return err
		}

	}

	log.L(ctx).Info("handle public collection success")

	return nil
}

func (sfs *salesforceService) handleNSRevrec(ctx context.Context) error {
	log.L(ctx).Info("handle NS Revrec")

	comSrv := commonService{
		store: sfs.store,
	}

	nsRevrecs, err := sfs.store.Salesforce().ListNSRevrecNotSync(ctx)
	if err != nil {
		log.L(ctx).Errorf("ListNSRevrecNotSync failed: %v", err)

		return err
	}

	for _, nsRevrec := range nsRevrecs {
		if nsRevrec.Sync != 0 && nsRevrec.Sync != 1 {
			continue
		}

		nowTime := time.Now()

		var revrecStatus *model.NSRevRecStatus
		revrecStatus, _ = sfs.store.Salesforce().GetNSRevrecStatusByRowID(ctx, nsRevrec.RowID)
		if revrecStatus == nil || revrecStatus.ID == 0 {
			revrecStatus = &model.NSRevRecStatus{
				CreatedAt: nowTime,
			}
		}

		revrecStatus.RowID = nsRevrec.RowID
		revrecStatus.SyncTime = nowTime
		revrecStatus.UpdatedTime = nowTime

		if nsRevrec.OrderID == "" {
			log.L(ctx).Errorf("handleNSRevrec order id is empty" + strconv.Itoa(nsRevrec.ID))

			go comSrv.saveLarkOps(ctx, errorMsg, "handleNSRevrec order id is empty. nsRevrec:"+strconv.Itoa(nsRevrec.ID))

			revrecStatus.Sync = 2

			go sfs.store.Salesforce().UpdateNSRevrecStatus(ctx, revrecStatus)

			continue
		}

		op := create
		if nsRevrec.Sync == 1 {
			op = update
		}

		id, name, err := createNSRevrec(ctx, nsRevrec, op)
		revrecStatus.SfRevrecNumber = name
		revrecStatus.SfRevrecID = id
		revrecStatus.Sync = 1

		if err != nil {
			log.L(ctx).Errorf("updateNSRevrec failed: %v", err)

			revrecStatus.Sync = 2
			go sfs.store.Salesforce().UpdateNSRevrecStatus(ctx, revrecStatus)
			return err
		}

		err = sfs.store.Salesforce().UpdateNSRevrecStatus(ctx, revrecStatus)
		if err != nil {
			log.L(ctx).Errorf("UpdateNSRevrec failed: %v", err)

			return err
		}
	}

	log.L(ctx).Info("handle NS Revrec success")

	return nil
}

func updateSfpsdays(ctx context.Context, psReject *model.PsManDaysReject) error {
	log.L(ctx).Infof("update psReject: %v", psReject)

	url := "/sobjects/Service_Man_day__c"
	body := model.SfPsManDayReject{
		RejectReason: psReject.RejectReason,
	}
	err := upSfFields(ctx, url, psReject.SalesforceID, body)
	if err != nil {
		log.L(ctx).Errorf("update psReject failed: %v", err)

		return err
	}

	log.L(ctx).Infof("update psReject success: %v", psReject)

	return nil
}

func updateInvoice(ctx context.Context, base64pdf, title, orderID string) error {
	log.L(ctx).Infof("upload pdf to oder: %v", orderID)

	url := "/sobjects/ContentVersion"

	body := map[string]interface{}{
		"Title":                  strings.Split(title, ".")[0],
		"PathOnClient":           title,
		"VersionData":            base64pdf,
		"FirstPublishLocationId": orderID,
		"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 createCollection(ctx context.Context, collection interface{}, tp, op int) (string, string, error) {
	log.L(ctx).Infof("update collection: %v", collection)

	url := "/sobjects/Invoice_Collection__c"

	var body interface{}
	var id, name string

	if tp == market {
		mkCollection := collection.(*model.MarketplaceCollection)

		tBody := model.CreatedCollectionReq{
			PayableAmount:    mkCollection.PayableAmount,
			PayableDate:      mkCollection.PayableDate,
			CollectionStatus: collectStatus(mkCollection.CollectionStatus),
		}

		if op == update {
			url = "/sobjects/Invoice_Collection__c/" + mkCollection.CreatedCollectionID
			id = mkCollection.CreatedCollectionID
			name = mkCollection.CreatedCollectionNumber

			if mkCollection.Schange {
				tBody.CollectionsDate = mkCollection.CollectionsDate
				tBody.CollectionsAmount = mkCollection.PayableAmount
			}
		} else {
			if strings.ToLower(mkCollection.CollectionStatus) == "collected" {
				tBody.CollectionsDate = mkCollection.CollectionsDate
				tBody.CollectionsAmount = mkCollection.PayableAmount
			}
			tBody.Order = mkCollection.OrderID
		}

		body = tBody
	} else {
		pbc := collection.(*model.PublicCollection)

		payDate, _ := formatDate(pbc.PayableDate)
		actualDate, _ := formatDate(pbc.ActualBillingDate)
		collectionsDate, _ := formatDate(pbc.CollectionsDate)
		collectionStatus := collectStatus(pbc.CollectionStatus)

		exchangeRateToUse := pbc.ExchangeRateToUse
		if pbc.CurrencyISOCode == "USD" {
			exchangeRateToUse = "1"
		}

		tbody := model.CreatedCollectionReq{
			PayableAmount:     pbc.PayableAmount,
			PayableDate:       payDate,
			ExchangeRateUse:   exchangeRateToUse,
			CurrencyIsoCode:   pbc.CurrencyISOCode,
			CollectionsDate:   collectionsDate,
			CollectionsAmount: pbc.CollectionsAmount,
			InvoiceNumber:     pbc.InvoiceNumber,
			CollectionStatus:  collectionStatus,
			ActualBillingDate: actualDate,
		}

		if op == update {
			url = "/sobjects/Invoice_Collection__c/" + pbc.CollectionID
			id = pbc.CollectionID
			name = pbc.CollectionNumber
		} else {
			tbody.Order = pbc.OrderID
		}

		body = tbody
	}

	b, _ := json.Marshal(body)

	method := http.MethodPost
	if op == update {
		method = http.MethodPatch
	}

	resp, err := sendRequest(ctx, url, method, nil, b)
	if err != nil {
		log.L(ctx).Error("send request failed", zap.Error(err))

		return "", "", err
	}
	var res model.CommonSalesResp
	if len(resp) != 0 {
		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("update collection failed")
		}

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

			id = res.ID
		}
	}

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

	return id, name, nil
}

func createNSRevrec(ctx context.Context, nsRevrec *model.NSRevRec, op int) (string, string, error) {
	log.L(ctx).Infof("create nsRevrec: %v", nsRevrec)

	var id, name string

	url := "/sobjects/Revenue__c"

	amount := strconv.FormatFloat(nsRevrec.FxAmount, 'f', -1, 64)
	// exchangeRate := strconv.FormatFloat(nsRevrec.ExchangeRate, 'f', -1, 64)
	trandate, err := formatDate(nsRevrec.TranDate)
	if err != nil {
		log.L(ctx).Errorf("format date failed", zap.Error(err))

		return "", "", err
	}

	body := model.CreatedRevrecReq{
		Amount:            amount,
		CurrencyIsoCode:   nsRevrec.Currency,
		Date:              trandate,
		ExchangeRateToUse: "1",
		Status:            "Confirmed",
		Product:           defaultProduct,
	}

	if nsRevrec.OpOrderExchangeRate != "" {
		body.ExchangeRateToUse = nsRevrec.OpOrderExchangeRate
	}

	method := http.MethodPost
	if op == update {
		url = "/sobjects/Revenue__c/" + nsRevrec.SfRevrecID
		method = http.MethodPatch
		id = nsRevrec.SfRevrecID
		name = nsRevrec.SfRevrecNumber
	} else {
		body.Order = nsRevrec.OrderID
	}

	b, _ := json.Marshal(body)

	resp, err := sendRequest(ctx, url, method, nil, b)
	if err != nil {
		log.L(ctx).Errorf("send request failed: %v", err)

		return "", "", err
	}

	var res model.CommonSalesResp

	if len(resp) != 0 {
		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")
		}

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

			id = res.ID
		}
	}
	log.L(ctx).Infof("create nsrecrec success: %s", nsRevrec.NSOrderNumber)

	return id, name, nil
}

func formatDate(input string) (string, error) {
	if input == "" {
		return "", nil
	}

	// 检查输入日期是否已经是 yyyy-mm-dd 格式
	if _, err := time.Parse("2006-01-02", input); err == nil {
		return input, nil
	}

	// 定义输入日期的格式
	const inputFormat = "01/02/2006" // Go 的参考时间是 2006-01-02 15:04:05

	// 解析输入日期
	t, err := time.Parse(inputFormat, input)
	if err != nil {
		return "", fmt.Errorf("failed to parse date: %v", err)
	}

	// 定义输出日期的格式
	const outputFormat = "2006-01-02"

	// 格式化日期
	return t.Format(outputFormat), nil
}

func collectStatus(status string) string {
	if status == "" {
		return ""
	}
	status = strings.ToLower(status)
	if status == "collect" || status == "received" || status == "collected" {
		return "Received"
	}

	return "Uncollected"
}

func (sfs *salesforceService) handleCheckStripeCollection(ctx context.Context) error {
	log.L(ctx).Info("handle check stripe collection")

	// 1. Get uncollected records from cloud collection
	uncollectedRecords, err := sfs.store.CloudCollection().ListUncollected(ctx)
	if err != nil {
		log.L(ctx).Errorf("ListUncollected failed: %v", err)
		return err
	}

	// 2. Get collection data from BigQuery
	bqSrv, err := bigquery.NewClient(ctx, GcpProject)
	if err != nil {
		log.L(ctx).Errorf("Failed to create bigquery client: %v", err)
		return err
	}

	query := `SELECT * FROM gcp-digital-bi.salesforce.invoice_collection_view `
	q := bqSrv.Query(query)
	it, err := q.Read(ctx)
	if err != nil {
		log.L(ctx).Errorf("Failed to query bigquery: %v", err)
		return err
	}

	type InvoiceCollectionData struct {
		ID                string
		CollectionsDate   string
		CollectionsAmount string
		CollectionStatus  string
	}
	collectionStatusMap := make(map[string]InvoiceCollectionData)
	for {
		var value model.InvoiceCollection
		err := it.Next(&value)
		if errors.Is(err, iterator.Done) {
			break
		}
		if err != nil {
			log.L(ctx).Errorf("Failed to read row: %v", err)

			return err
		}

		collectionStatusMap[value.Name.StringVal] = InvoiceCollectionData{
			ID:                value.ID.StringVal,
			CollectionsDate:   value.CollectionsDate.String(),
			CollectionsAmount: value.CollectionsAmount.StringVal,
			CollectionStatus:  value.CollectionsStatus.StringVal,
		}
	}

	// 3. Compare and update records
	for _, record := range uncollectedRecords {
		if record.CollectionNumber == "" {
			continue
		}

		// Check if collection status in BigQuery is "Received"
		if collectionData, exists := collectionStatusMap[record.CollectionNumber]; exists && collectionData.CollectionStatus == "Received" {
			// Update the record status
			record.CollectionStatus = "Received"
			record.CollectionsDate = collectionData.CollectionsDate
			record.CollectionsAmount = collectionData.CollectionsAmount
			record.StripePaid = "True"

			err = sfs.UpdateHistoryCollection(ctx, record)
			if err != nil {
				log.L(ctx).Errorf("Update history collection failed for record %d: %v", record.ID, err)
				continue
			}
			log.L(ctx).Infof("Updated collection status to Received for record %d", record.ID)
		}
	}

	log.L(ctx).Info("handle check stripe collection success")
	return nil
}

func (sfs *salesforceService) UpdateHistoryCollection(ctx context.Context, record *model.CloudCollection) error {
	log.L(ctx).Infof("Update history collection for record %d", record.ID)

	// Update only specific fields
	updates := map[string]interface{}{
		"collection_status":  record.CollectionStatus,
		"collections_date":   record.CollectionsDate,
		"collections_amount": record.CollectionsAmount,
		"stripe_paid":        record.StripePaid,
		"updated_at":         time.Now().Format("2006-01-02 15:04:05"),
	}

	err := sfs.store.CloudCollection().UpdateFields(ctx, record.ID, updates)
	if err != nil {
		log.L(ctx).Errorf("Update history collection failed for record %d: %v", record.ID, err)
		return err
	}

	log.L(ctx).Infof("Successfully updated history collection for record %d", record.ID)
	return nil
}