package service

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

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

	"data-platform/internal/dataplatform/store"
	"data-platform/model"
	"data-platform/pkg/errors"
	"data-platform/pkg/log"
)

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

	HandleSingleSF(ctx context.Context, req model.SfReqBody)
	SyncSingleSF(ctx context.Context, req model.SfReqBody) (*model.SfSyncResult, error)
	HandleLastInvoiceMonthSF(ctx context.Context, req model.SfReqBody)

	EventCenter(ctx context.Context, etype string)
}

type salesforceService struct {
	store                 store.Factory
	pricebookEntryMu      sync.Mutex
	pricebookEntryByScope map[string]string
}

var _ SalesforceSrv = (*salesforceService)(nil)

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

const (
	defaultProduct   = "01t5g000000liLxAAI"
	defaultPriceBook = "01s5g000001cOA9AAM"
	defaultCurrency  = "USD"
	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"

	collectionFileMaxBytes = 20 << 20

	tokenPrefix = "salesforce"

	// request targets for sendRequestTo / getInfoByIdTo / upSfFieldsTo.
	// Salesforce and NexusCRM speak the same SF-compatible REST API.
	targetSalesforce = "salesforce"
	targetNexusCRM   = "nexuscrm"
)

var (
	sem                           = make(chan int, 1)
	salesforceHTTPClient          = &http.Client{}
	salesforceRequestMaxAttempt   = 2
	salesforceRetryDelay          = 10 * time.Second
	salesforceOrderLocks          sync.Map
	errCRMRequestLimitExceeded    = stdErrors.New("CRM request limit exceeded")
	orderActivationMaxAttempt     = 3
	orderActivationRetryDelay     = 2 * time.Second
	orderLastMonthConcurrency     = 8
	collectionFileDownloadTimeout = 30 * time.Second
)

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 isNumericString(s string) bool {
	if s == "" {
		return false
	}
	for _, c := range s {
		if c < '0' || c > '9' {
			return false
		}
	}
	return true
}

func shouldSkipRevrecMainSync(revrec *model.CloudRevrec) bool {
	if revrec == nil || revrec.Sync == 1 {
		return true
	}
	return isNumericString(revrec.AccountNumber) && revrec.AccountNumber == revrec.AdjustTenantID
}

func (sfs *salesforceService) SyncSalesforce(ctx context.Context) error {
	log.L(ctx).Info("start CRM order sync and local financial completion")

	lastMonth := GetLastMonth()

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

		return err
	}
	var orderErrList []error
	for _, order := range orders {
		if order.Sync == 1 {
			continue
		}
		if isNumericString(order.AccountNumber) && order.AccountNumber == order.AdjustTenantID {
			log.L(ctx).Infof("skip order sync, account not in salesforce: %s--%s", order.BillMonth, order.AdjustTenantID)
			order.Sync = 1
			order.SyncTime = time.Now()
			sfs.store.CloudOrder().Update(ctx, order)
			continue
		}
		result, err := sfs.sfOrder(ctx, order)
		if err != nil {
			log.L(ctx).Error("sync order to salesforce failed", zap.Error(err))
			orderErrList = append(orderErrList, err)
		} else if result != nil && result.Status == "processing" {
			log.L(ctx).Errorf("sync order to salesforce pending:%s", result.Message)
			orderErrList = append(orderErrList, errors.New(result.Message))
		}
	}
	if len(orderErrList) > 0 {
		return fmt.Errorf("sync order to salesforce failed with %d errors: %v", len(orderErrList), orderErrList[0])
	}

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

		return err
	}

	log.L(ctx).Info("revrec Salesforce create disabled; completing local state without Order resolution")
	var revrecErrList []error
	for _, revrec := range revrecs {
		if err := sfs.syncRevrecLocally(ctx, revrec); err != nil {
			revrecErrList = append(revrecErrList, err)
		}
	}
	if len(revrecErrList) > 0 {
		return fmt.Errorf("local revrec completion failed with %d errors: %v", len(revrecErrList), revrecErrList[0])
	}

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

		return err
	}

	log.L(ctx).Info("collection Salesforce create disabled; completing local fields and state")
	var collectionErrList []error
	for _, collection := range collections {
		if !needsCollectionLocalSync(collection) {
			continue
		}
		if err := sfs.syncCollectionLocally(ctx, collection); err != nil {
			collectionErrList = append(collectionErrList, err)
		}
	}
	if len(collectionErrList) > 0 {
		return fmt.Errorf("local collection completion failed with %d errors: %v", len(collectionErrList), collectionErrList[0])
	}

	oderMap := make(map[string]*model.CloudOrder)
	revrecMap := make(map[string][]*model.CloudRevrec)
	collectionMap := make(map[string][]*model.CloudCollection)

	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)
	}

	// 更新自动创建的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.CloudCollection
	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.CloudOrder) (*model.SfSyncResult, error) {
	return sfs.sfOrderWithActivation(ctx, order, false)
}

func (sfs *salesforceService) sfOrderWithActivation(ctx context.Context, order *model.CloudOrder, deferActivation bool) (*model.SfSyncResult, error) {
	log.L(ctx).Infof("create order to salesforce:%s--%s", order.BillMonth, order.AdjustTenantID)
	crmOrder, err := sfs.orderPayloadSource(ctx, *order)
	if err != nil {
		return nil, err
	}

	result := &model.SfSyncResult{
		Type:              "order",
		ID:                order.ID,
		Status:            "processing",
		OrderID:           strings.TrimSpace(order.NewOrderID),
		OrderNumber:       strings.TrimSpace(order.NewOrderNumber),
		OrderNumberSource: "crm_readback",
		ReadyStatus:       "pending",
		ActivationStatus:  "pending",
	}

	sourceRef := cloudOrderSourceRef(order)
	createdOrder := false
	if strings.TrimSpace(order.NewOrderID) == "" {
		if existing, ok, err := sfs.findSyncedOrderBySourceRef(ctx, order); err != nil {
			return result, err
		} else if ok {
			order.NewOrderID = existing.NewOrderID
			order.NewOrderNumber = existing.NewOrderNumber
			crmOrder.NewOrderID = existing.NewOrderID
			crmOrder.NewOrderNumber = existing.NewOrderNumber
			if strings.TrimSpace(order.FakeNumber) == "" {
				order.FakeNumber = sourceRef
			}
			if strings.TrimSpace(crmOrder.FakeNumber) == "" {
				crmOrder.FakeNumber = sourceRef
			}
			order.Sync = 0
			if err := sfs.store.CloudOrder().Update(ctx, order); err != nil {
				return result, err
			}
		} else {
			orderID, orderName, err := sfs.createOrder(ctx, &crmOrder)
			if err != nil {
				log.L(ctx).Errorf("create order failed", zap.Error(err))

				return result, err
			}

			order.NewOrderNumber = orderName
			order.NewOrderID = orderID
			crmOrder.NewOrderNumber = orderName
			crmOrder.NewOrderID = orderID
			if strings.TrimSpace(order.FakeNumber) == "" && sourceRef != "" {
				order.FakeNumber = sourceRef
			}
			if strings.TrimSpace(crmOrder.FakeNumber) == "" && sourceRef != "" {
				crmOrder.FakeNumber = sourceRef
			}
			order.Sync = 0
			if err := sfs.store.CloudOrder().Update(ctx, order); err != nil {
				log.L(ctx).Errorf("persist created order identity failed", zap.Error(err))

				return result, err
			}
			createdOrder = true
		}
	} else {
		crmOrder.NewOrderID = order.NewOrderID
		crmOrder.NewOrderNumber = order.NewOrderNumber
	}

	result.OrderID = order.NewOrderID
	result.OrderNumber = order.NewOrderNumber
	if !createdOrder {
		if err := sfs.updateOrder(ctx, &crmOrder); err != nil {
			return result, err
		}
	}

	orderItemID, err := sfs.ensureOrderItem(ctx, &crmOrder)
	if err != nil {
		log.L(ctx).Errorf("ensure order item failed", zap.Error(err))

		return result, err
	}
	result.OrderItemID = orderItemID
	if deferActivation {
		result.Status = "created"
		result.Message = "order and order item are ready; activation deferred for database repair"
		result.ReadyStatus = "order_item_ready"
		result.ActivationStatus = "deferred"
		return result, nil
	}

	attempts, err := activateOrderWithBoundedRetry(ctx, order.NewOrderID)
	result.ActivationAttempts = attempts
	if err != nil {
		log.L(ctx).Errorf("update order status failed", zap.Error(err))

		result.Status = "processing"
		result.Message = "order and order item are ready; activation is pending CRM Flow readiness"
		result.ActivationStatus = "pending"
		result.ActivationError = err.Error()
		result.ReadyStatus = "pending"
		return result, nil
	}

	result.ActivationStatus = "activated"
	order.Sync = 1
	order.SyncTime = time.Now()
	if err := sfs.store.CloudOrder().Update(ctx, order); err != nil {
		log.L(ctx).Errorf("persist synced order state failed", zap.Error(err))

		return result, err
	}

	result.Status = "synced"
	result.Message = "order item ensured and order activated"
	result.ReadyStatus = "ready"
	log.L(ctx).Infof("create order success:%s", order.NewOrderID)

	return result, nil
}

func cloudOrderSourceRef(order *model.CloudOrder) string {
	if order == nil {
		return ""
	}
	if ref := strings.TrimSpace(order.FakeNumber); ref != "" {
		return ref
	}
	if ref := strings.TrimSpace(order.NewOrderNumber); ref != "" {
		return ref
	}
	if strings.TrimSpace(order.BillMonth) != "" || strings.TrimSpace(order.AdjustTenantID) != "" || strings.TrimSpace(order.AccountNumber) != "" {
		return strings.Join([]string{
			strings.TrimSpace(order.BillMonth),
			strings.TrimSpace(order.AdjustTenantID),
			strings.TrimSpace(order.AccountNumber),
		}, "|")
	}
	return ""
}

func cloudOrderLockKey(order *model.CloudOrder) string {
	if ref := cloudOrderSourceRef(order); ref != "" {
		if order != nil {
			return strings.Join([]string{
				strings.TrimSpace(order.BillMonth),
				strings.TrimSpace(order.AdjustTenantID),
				ref,
			}, "|")
		}
		return ref
	}
	if order != nil && order.ID != 0 {
		return fmt.Sprintf("id:%d", order.ID)
	}
	return "unknown"
}

func acquireSalesforceOrderLock(key string) (func(), bool) {
	value, _ := salesforceOrderLocks.LoadOrStore(key, &sync.Mutex{})
	mu := value.(*sync.Mutex)
	if !mu.TryLock() {
		return func() {}, false
	}
	return mu.Unlock, true
}

func (sfs *salesforceService) findSyncedOrderBySourceRef(ctx context.Context, order *model.CloudOrder) (model.CloudOrder, bool, error) {
	ref := cloudOrderSourceRef(order)
	if strings.TrimSpace(ref) == "" {
		return model.CloudOrder{}, false, nil
	}
	records, err := sfs.store.CloudOrder().ListBySourceRef(ctx, order.BillMonth, order.AdjustTenantID, ref)
	if err != nil {
		if err == gorm.ErrRecordNotFound {
			return model.CloudOrder{}, false, nil
		}
		return model.CloudOrder{}, false, err
	}
	for _, existing := range records {
		if order != nil && existing.ID == order.ID {
			continue
		}
		if strings.TrimSpace(existing.NewOrderID) != "" {
			return existing, true, nil
		}
	}
	return model.CloudOrder{}, false, nil
}

func (sfs *salesforceService) orderPayloadSource(ctx context.Context, order model.CloudOrder) (model.CloudOrder, error) {
	ref := cloudOrderSourceRef(&order)
	if ref == "" && strings.TrimSpace(order.NewOrderID) != "" {
		ref = strings.TrimSpace(order.NewOrderID)
	}
	if ref == "" {
		return order, nil
	}

	records, err := sfs.store.CloudOrder().ListBySourceRef(ctx, order.BillMonth, order.AdjustTenantID, ref)
	if err != nil {
		return model.CloudOrder{}, err
	}
	if len(records) <= 1 {
		return order, nil
	}

	aggregated := order
	aggregated.NetFee = "0"
	for _, record := range records {
		aggregated.NetFee = sumMoneyStrings(aggregated.NetFee, record.NetFee)
	}
	return aggregated, nil
}

func activateOrderWithBoundedRetry(ctx context.Context, orderID string) (int, error) {
	attempts := orderActivationMaxAttempt
	if attempts < 1 {
		attempts = 1
	}

	url := "/sobjects/Order"
	body := map[string]interface{}{
		"Status": "Activated",
	}

	var lastErr error
	for i := 0; i < attempts; i++ {
		if err := upSfFieldsTo(ctx, targetNexusCRM, url, orderID, body); err != nil {
			lastErr = err
		} else {
			return i + 1, nil
		}

		if i == attempts-1 {
			break
		}
		select {
		case <-ctx.Done():
			return i + 1, ctx.Err()
		case <-time.After(orderActivationRetryDelay):
		}
	}

	return attempts, fmt.Errorf("order activation not ready after %d attempts: %w", attempts, lastErr)
}

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

	if revrec.RevrecID != "" {
		return sfs.updateRevrec(ctx, revrec)
	}

	if err := sfs.prepareRevrecOrderForSalesforce(ctx, revrec); err != nil {
		log.L(ctx).Errorf("prepare revrec order failed", zap.Error(err))

		return err
	}

	existingID, existingName, err := getRevenueByOrderAndBillMonth(ctx, revrec.Order, revrec.BillMonth)
	if err != nil {
		log.L(ctx).Errorf("lookup existing revrec failed", zap.Error(err))

		return err
	}
	if existingID != "" {
		revrec.RevrecID = existingID
		revrec.RevrecNumber = existingName

		return sfs.updateRevrec(ctx, revrec)
	}

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

		return err
	}

	revrec.RevrecNumber = revrecName
	revrec.RevrecID = revrecID
	revrec.Sync = 1
	revrec.SyncTime = time.Now()

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

		return err
	}

	//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", revrec.RevrecNumber)

	return nil
}

func (sfs *salesforceService) syncRevrecLocally(ctx context.Context, revrec *model.CloudRevrec) error {
	if revrec == nil || revrec.Sync == 1 {
		return nil
	}

	// Downstream RevRec ingestion is keyed by the RevRec row itself and does not
	// depend on a CRM Order. Local completion must therefore preserve the source
	// Order references and must not block on Order lookup or synchronization.
	revrec.Sync = 1
	revrec.SyncTime = time.Now()
	if err := sfs.store.CloudRevrec().Update(ctx, revrec); err != nil {
		return err
	}

	return nil
}

func (sfs *salesforceService) prepareRevrecOrderForSalesforce(ctx context.Context, revrec *model.CloudRevrec) error {
	if revrec.Type != 0 {
		return nil
	}

	order, err := sfs.resolveCloudOrderForSalesforce(ctx, revrec.AdjustTenantID, revrec.BillMonth, revrec.AccountNumber, revrec.OrderNumber, revrec.Order)
	if err != nil {
		return err
	}

	revrec.Order = order.NewOrderID
	if !shouldKeepSelfServeOrderNumberBlank(revrec.AccountNumber, revrec.AdjustTenantID) {
		revrec.OrderNumber = order.NewOrderNumber
	}

	return nil
}

func (sfs *salesforceService) prepareCollectionOrderForSalesforce(ctx context.Context, collection *model.CloudCollection) error {
	if collection.Type != 0 {
		return nil
	}

	order, err := sfs.resolveCloudOrderForSalesforce(ctx, collection.AdjustTenantID, collection.BillMonth, collection.AccountNumber, collection.OrderNumber, collection.Order)
	if err != nil {
		return err
	}

	collection.Order = order.NewOrderID
	if !shouldKeepSelfServeOrderNumberBlank(collection.AccountNumber, collection.AdjustTenantID) {
		collection.OrderNumber = order.NewOrderNumber
	}

	return nil
}

func (sfs *salesforceService) resolveCloudOrderForSalesforce(ctx context.Context, adjustTenantID, billMonth, accountNumber string, refs ...string) (model.CloudOrder, error) {
	seen := make(map[string]struct{}, len(refs))
	for _, ref := range refs {
		ref = strings.TrimSpace(ref)
		if ref == "" {
			continue
		}
		if _, ok := seen[ref]; ok {
			continue
		}
		seen[ref] = struct{}{}

		order, err := sfs.store.CloudOrder().GetByFakeNumber(ctx, ref)
		if err == nil {
			if order.NewOrderID == "" {
				return order, fmt.Errorf("salesforce order %s is not synced", ref)
			}
			return order, nil
		}
	}

	accountNumber = strings.TrimSpace(accountNumber)
	if accountNumber != "" {
		order, err := sfs.store.CloudOrder().GetByAdjustIDAndAccount(ctx, adjustTenantID, billMonth, accountNumber)
		if err != nil {
			if stdErrors.Is(err, gorm.ErrRecordNotFound) {
				return order, fmt.Errorf("salesforce order for tenant %s account %s bill month %s is not found", adjustTenantID, accountNumber, billMonth)
			}
			return order, err
		}
		if order.NewOrderID == "" {
			return order, fmt.Errorf("salesforce order for tenant %s account %s bill month %s is not synced", adjustTenantID, accountNumber, billMonth)
		}
		return order, nil
	}

	order, err := sfs.store.CloudOrder().GetByAdjustID(ctx, adjustTenantID, billMonth)
	if err != nil {
		return order, err
	}
	if order.NewOrderID == "" {
		return order, fmt.Errorf("salesforce order for tenant %s bill month %s is not synced", adjustTenantID, billMonth)
	}

	return order, nil
}

func (sfs *salesforceService) sfCollection(ctx context.Context, collection *model.CloudCollection) error {
	return sfs.syncCollectionLocally(ctx, collection)
}

func (sfs *salesforceService) syncCollectionLocally(ctx context.Context, collection *model.CloudCollection) error {
	if collection == nil {
		return nil
	}

	wasIncomplete := collection.Sync != 1
	if wasIncomplete &&
		!(isNumericString(collection.AccountNumber) && collection.AccountNumber == collection.AdjustTenantID) {
		if err := sfs.prepareCollectionOrderForSalesforce(ctx, collection); err != nil {
			return err
		}
	}

	if strings.TrimSpace(collection.Files) != "" && strings.TrimSpace(collection.FileContent) == "" {
		fileContent, err := downloadFileContentFromURL(ctx, collection.Files)
		if err != nil {
			return err
		}
		collection.FileContent = base64.StdEncoding.EncodeToString(fileContent)
	}

	collection.Sync = 1
	collection.SyncTime = time.Now()
	return sfs.store.CloudCollection().Update(ctx, collection)
}

func needsCollectionLocalSync(collection *model.CloudCollection) bool {
	if collection == nil {
		return false
	}
	return collection.Sync != 1 ||
		(strings.TrimSpace(collection.Files) != "" && strings.TrimSpace(collection.FileContent) == "")
}

func (sfs *salesforceService) sfCollectionToSalesforce(ctx context.Context, collection *model.CloudCollection) error {
	log.L(ctx).Infof("create collection to salesforce:%s--%s", collection.BillMonth, collection.AdjustTenantID)

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

		return err
	}
	collection.CollectionNumber = collectionNum
	collection.CollectionID = collectionID
	collection.Sync = 1
	collection.SyncTime = time.Now()
	if fileContent != nil {
		collection.FileContent = base64.StdEncoding.EncodeToString(fileContent)
	}
	err = sfs.store.CloudCollection().Update(ctx, collection)
	if err != nil {
		log.L(ctx).Errorf("update collection failed", zap.Error(err))

		return err
	}

	// update order is stripe
	body := map[string]interface{}{
		"Is_Stripe__c": "true",
	}
	err = upSfFields(ctx, "/sobjects/Order", collection.Order, body)
	if err != nil {
		log.L(ctx).Errorf("update order failed", zap.Error(err))

		return err
	}

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

	return nil
}

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

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

	url := "/sobjects/order"

	body := buildOrderPayload(order, true)

	b, _ := json.Marshal(body)
	resp, err := sendRequestTo(ctx, targetNexusCRM, 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 := getInfoByIdTo(ctx, targetNexusCRM, url, res.ID)
	if err != nil {
		return "", "", err
	}

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

	return res.ID, name, nil
}

func (sfs *salesforceService) updateOrder(ctx context.Context, order *model.CloudOrder) error {
	log.L(ctx).Infof("update order to salesforce:%s--%s", order.BillMonth, order.AdjustTenantID)

	url := "/sobjects/Order"
	body := buildOrderPayload(order, false)

	err := upSfFieldsTo(ctx, targetNexusCRM, url, order.NewOrderID, body)
	if err != nil {
		log.L(ctx).Errorf("update order failed", zap.Error(err))

		return err
	}

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

	return nil
}

func buildOrderPayload(order *model.CloudOrder, create bool) model.CreatedOrderReq {
	body := model.CreatedOrderReq{
		AccountID:              order.AccountID,
		OrderStartDate:         order.OrderStartDate,
		OrderEndDate:           order.OrderEndDate,
		Tax:                    order.Tax,
		ProductFamily:          "TiDB Cloud",
		AccumulateRevenueToOrg: order.AdjustTenantID,
	}
	if create {
		body.Status = "Draft"
		body.PriceBook2ID = defaultPriceBook
		body.CurrencyIsoCode = defaultCurrency
	}
	if order.RevrecEntity != "" {
		body.PingCAPRevRecEntity = order.RevrecEntity
	}
	return body
}

func (sfs *salesforceService) orderSingle(ctx context.Context, id uint64) error {
	result, err := sfs.orderSingleResult(ctx, id)
	if err != nil {
		return err
	}
	if result != nil && result.Status == "processing" {
		return errors.New(result.Message)
	}
	return nil
}

func (sfs *salesforceService) orderSingleResult(ctx context.Context, id uint64) (*model.SfSyncResult, error) {
	return sfs.orderSingleResultWithActivation(ctx, id, false)
}

func (sfs *salesforceService) orderSingleResultWithActivation(ctx context.Context, id uint64, deferActivation bool) (*model.SfSyncResult, error) {
	log.L(ctx).Infof("start sync order to salesforce:%d", id)

	order, err := sfs.store.CloudOrder().GetByID(ctx, id)
	if err != nil {
		log.L(ctx).Errorf("get order failed", zap.Error(err))

		return nil, err
	}

	sourceKey := cloudOrderLockKey(&order)
	unlockLocal, lockedLocal := acquireSalesforceOrderLock(sourceKey)
	if !lockedLocal {
		return &model.SfSyncResult{
			Type:              "order",
			ID:                id,
			Status:            "processing",
			Message:           "order sync already in progress for source reference",
			OrderID:           order.NewOrderID,
			OrderNumber:       order.NewOrderNumber,
			OrderNumberSource: "crm_readback",
			ReadyStatus:       "pending",
			ActivationStatus:  "pending",
		}, nil
	}
	defer unlockLocal()

	lockedDB, err := sfs.store.CloudOrder().TryLockByID(ctx, id)
	if err != nil {
		return nil, err
	}
	if !lockedDB {
		order, _ = sfs.store.CloudOrder().GetByID(ctx, id)
		return &model.SfSyncResult{
			Type:              "order",
			ID:                id,
			Status:            "processing",
			Message:           "order sync already in progress",
			OrderID:           order.NewOrderID,
			OrderNumber:       order.NewOrderNumber,
			OrderNumberSource: "crm_readback",
			ReadyStatus:       "pending",
			ActivationStatus:  "pending",
		}, nil
	}
	defer func() {
		if err := sfs.store.CloudOrder().UnlockByID(ctx, id); err != nil {
			log.L(ctx).Errorf("unlock order failed:%d", id, zap.Error(err))
		}
	}()

	order, err = sfs.store.CloudOrder().GetByID(ctx, id)
	if err != nil {
		log.L(ctx).Errorf("get locked order failed", zap.Error(err))

		return nil, err
	}

	if !IsFormatDate(order.OrderStartDate) || !IsFormatDate(order.OrderEndDate) {
		log.L(ctx).Errorf("order date format error", zap.Error(err))

		return nil, errors.New("order date format error")
	}

	var result *model.SfSyncResult
	if order.Sync == 1 {
		crmOrder, payloadErr := sfs.orderPayloadSource(ctx, order)
		if payloadErr != nil {
			return nil, payloadErr
		}
		err = sfs.updateOrder(ctx, &crmOrder)
		if err != nil {
			log.L(ctx).Errorf("update order failed", zap.Error(err))

			return nil, err
		}
		result = &model.SfSyncResult{
			Type:              "order",
			ID:                id,
			Status:            "synced",
			Message:           "existing Salesforce order updated",
			OrderID:           order.NewOrderID,
			OrderNumber:       order.NewOrderNumber,
			OrderNumberSource: "crm_readback",
			ReadyStatus:       "not_applicable",
			ActivationStatus:  "not_applicable",
		}
	} else {
		result, err = sfs.sfOrderWithActivation(ctx, &order, deferActivation)
		if err != nil {
			log.L(ctx).Errorf("create order failed", zap.Error(err))

			return result, err
		}
	}

	log.L(ctx).Infof("sync order to salesforce success:%d", id)

	return result, nil
}

func selectLastMonthOrderBatch(orders []*model.CloudOrder, req model.SfReqBody) ([]*model.CloudOrder, error) {
	if err := req.ValidateLastMonthBatchScope(); err != nil {
		return nil, err
	}

	availableOrders := make([]*model.CloudOrder, 0, len(orders))
	for _, order := range orders {
		if order != nil {
			availableOrders = append(availableOrders, order)
		}
	}
	if req.Limit == 0 && len(req.OrderIDs) == 0 {
		return availableOrders, nil
	}

	sort.SliceStable(availableOrders, func(i, j int) bool {
		return availableOrders[i].ID < availableOrders[j].ID
	})
	isCRMEligible := func(order *model.CloudOrder) bool {
		return order.Sync != 1 && !(isNumericString(order.AccountNumber) && order.AccountNumber == order.AdjustTenantID)
	}

	if len(req.OrderIDs) > 0 {
		requested := make(map[uint64]struct{}, len(req.OrderIDs))
		for _, id := range req.OrderIDs {
			requested[id] = struct{}{}
		}

		selected := make([]*model.CloudOrder, 0, len(req.OrderIDs))
		found := make(map[uint64]struct{}, len(req.OrderIDs))
		for _, order := range availableOrders {
			if _, ok := requested[order.ID]; !ok {
				continue
			}
			found[order.ID] = struct{}{}
			if !isCRMEligible(order) {
				return nil, fmt.Errorf("requested order id %d is not eligible for CRM creation", order.ID)
			}
			selected = append(selected, order)
		}

		missing := make([]uint64, 0)
		for _, id := range req.OrderIDs {
			if _, ok := found[id]; !ok {
				missing = append(missing, id)
			}
		}
		if len(missing) > 0 {
			sort.Slice(missing, func(i, j int) bool { return missing[i] < missing[j] })
			return nil, fmt.Errorf("requested order_ids are not active unsynced last-month rows: %v", missing)
		}
		return selected, nil
	}

	selected := make([]*model.CloudOrder, 0, req.Limit)
	for _, order := range availableOrders {
		if !isCRMEligible(order) {
			continue
		}
		selected = append(selected, order)
		if len(selected) == req.Limit {
			break
		}
	}
	if len(selected) == 0 {
		return nil, fmt.Errorf("no active unsynced last-month orders are eligible for CRM creation")
	}
	return selected, nil
}

func cloudOrderIDs(orders []*model.CloudOrder) []uint64 {
	ids := make([]uint64, 0, len(orders))
	for _, order := range orders {
		if order != nil {
			ids = append(ids, order.ID)
		}
	}
	return ids
}

func processLastMonthOrderBatch(
	ctx context.Context,
	orders []*model.CloudOrder,
	concurrency int,
	process func(context.Context, *model.CloudOrder) error,
) []error {
	if len(orders) == 0 {
		return nil
	}
	if concurrency < 1 {
		concurrency = 1
	}
	if concurrency > len(orders) {
		concurrency = len(orders)
	}

	type job struct {
		index int
		order *model.CloudOrder
	}
	jobs := make(chan job)
	errorsByIndex := make([]error, len(orders))
	stopDispatch := make(chan struct{})
	var stopOnce sync.Once
	var workers sync.WaitGroup
	workers.Add(concurrency)
	for range concurrency {
		go func() {
			defer workers.Done()
			for current := range jobs {
				err := process(ctx, current.order)
				errorsByIndex[current.index] = err
				if stdErrors.Is(err, errCRMRequestLimitExceeded) {
					stopOnce.Do(func() { close(stopDispatch) })
				}
			}
		}()
	}

dispatchLoop:
	for index, order := range orders {
		select {
		case <-stopDispatch:
			break dispatchLoop
		default:
		}
		select {
		case jobs <- job{index: index, order: order}:
		case <-stopDispatch:
			break dispatchLoop
		}
	}
	close(jobs)
	workers.Wait()

	errList := make([]error, 0)
	for _, err := range errorsByIndex {
		if err != nil {
			errList = append(errList, err)
		}
	}
	return errList
}

func (sfs *salesforceService) orderLastInvoiceMonth(ctx context.Context, req model.SfReqBody) error {
	log.L(ctx).Info("start sync order to salesforce")
	if err := req.ValidateLastMonthBatchScope(); err != nil {
		return err
	}

	lastMonth := GetLastMonth()

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

		return err
	}
	orders, err = selectLastMonthOrderBatch(orders, req)
	if err != nil {
		return err
	}
	log.L(ctx).Info("selected last-month order batch",
		zap.String("bill_month", lastMonth),
		zap.Int("requested_limit", req.Limit),
		zap.Uint64s("requested_order_ids", req.OrderIDs),
		zap.Bool("defer_activation", req.DeferActivation),
		zap.Int("selected_count", len(orders)),
		zap.Uint64s("selected_order_ids", cloudOrderIDs(orders)))

	if _, err := tokenCache.GetToken(ctx, "nexuscrm"); err != nil {
		return fmt.Errorf("prepare NexusCRM token for last-month order batch: %w", err)
	}

	errList := processLastMonthOrderBatch(ctx, orders, orderLastMonthConcurrency, func(ctx context.Context, order *model.CloudOrder) error {
		log.L(ctx).Info("processing last-month order",
			zap.Uint64("order_id", order.ID))
		if order.Sync == 1 {
			return nil
		}
		if isNumericString(order.AccountNumber) && order.AccountNumber == order.AdjustTenantID {
			log.L(ctx).Infof("skip order sync, account not in salesforce: %s--%s", order.BillMonth, order.AdjustTenantID)
			order.Sync = 1
			order.SyncTime = time.Now()
			return sfs.store.CloudOrder().Update(ctx, order)
		}
		result, err := sfs.orderSingleResultWithActivation(ctx, order.ID, req.DeferActivation)
		if err != nil {
			return err
		}
		if result != nil && result.Status == "processing" {
			return errors.New(result.Message)
		}
		return nil
	})

	if len(errList) > 0 {
		log.L(ctx).Error("sync to salesforce failed", zap.Any("errList", errList))

		// return errorList information - return first 3 errors
		var errMsgs []string
		maxErrors := 3
		if len(errList) < maxErrors {
			maxErrors = len(errList)
		}

		for i := 0; i < maxErrors; i++ {
			errMsgs = append(errMsgs, errList[i].Error())
		}

		return fmt.Errorf("sync order to salesforce failed with %d errors. First %d errors: %v",
			len(errList), maxErrors, errMsgs)
	}

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

	return nil
}

type nexusCRMPricebookEntryRecord struct {
	ID string `json:"Id"`
}

type nexusCRMPricebookEntryQuery struct {
	TotalSize int                            `json:"totalSize"`
	Done      bool                           `json:"done"`
	Records   []nexusCRMPricebookEntryRecord `json:"records"`
}

type nexusCRMOrderScopeRecord struct {
	ID              string `json:"Id"`
	Pricebook2ID    string `json:"Pricebook2Id"`
	CurrencyIsoCode string `json:"CurrencyIsoCode"`
}

type nexusCRMOrderScopeQuery struct {
	TotalSize int                        `json:"totalSize"`
	Done      bool                       `json:"done"`
	Records   []nexusCRMOrderScopeRecord `json:"records"`
}

type nexusCRMOrderItemRecord struct {
	ID               string      `json:"Id"`
	OrderID          string      `json:"OrderId"`
	PricebookEntryID string      `json:"PricebookEntryId"`
	Quantity         json.Number `json:"Quantity"`
	UnitPrice        json.Number `json:"UnitPrice"`
	ServiceDate      string      `json:"ServiceDate"`
	EndDate          string      `json:"EndDate"`
	NetFee           json.Number `json:"Net_fees__c"`
	AccountID        string      `json:"Account__c"`
	IsDeleted        bool        `json:"IsDeleted"`
}

type nexusCRMOrderItemQuery struct {
	TotalSize int                       `json:"totalSize"`
	Done      bool                      `json:"done"`
	Records   []nexusCRMOrderItemRecord `json:"records"`
}

func queryOrderItemScope(ctx context.Context, orderID string) (nexusCRMOrderScopeRecord, error) {
	query := fmt.Sprintf(
		"SELECT Id, Pricebook2Id, CurrencyIsoCode FROM Order WHERE Id = '%s' LIMIT 1",
		escapeSOQLString(orderID),
	)
	resp, err := sendRequestTo(ctx, targetNexusCRM, "/query", http.MethodGet, map[string]string{"q": query}, nil)
	if err != nil {
		return nexusCRMOrderScopeRecord{}, err
	}

	var result nexusCRMOrderScopeQuery
	if err := json.Unmarshal(resp, &result); err != nil {
		return nexusCRMOrderScopeRecord{}, fmt.Errorf("decode NexusCRM Order scope query: %w", err)
	}
	if !result.Done || result.TotalSize != 1 || len(result.Records) != 1 || strings.TrimSpace(result.Records[0].ID) == "" {
		return nexusCRMOrderScopeRecord{}, fmt.Errorf(
			"expected exactly one NexusCRM Order scope record for %s, got totalSize=%d records=%d done=%t",
			orderID,
			result.TotalSize,
			len(result.Records),
			result.Done,
		)
	}
	return result.Records[0], nil
}

func (sfs *salesforceService) resolveDefaultPricebookEntryID(ctx context.Context, orderID string) (string, error) {
	scope, err := queryOrderItemScope(ctx, strings.TrimSpace(orderID))
	if err != nil {
		return "", err
	}
	if scope.Pricebook2ID != defaultPriceBook || scope.CurrencyIsoCode != defaultCurrency {
		return "", fmt.Errorf(
			"NexusCRM Order %s has unsupported OrderItem scope pricebook=%s currency=%s; expected pricebook=%s currency=%s",
			scope.ID,
			scope.Pricebook2ID,
			scope.CurrencyIsoCode,
			defaultPriceBook,
			defaultCurrency,
		)
	}

	cacheKey := strings.Join([]string{scope.Pricebook2ID, defaultProduct, scope.CurrencyIsoCode}, "|")
	sfs.pricebookEntryMu.Lock()
	defer sfs.pricebookEntryMu.Unlock()

	if pricebookEntryID := sfs.pricebookEntryByScope[cacheKey]; pricebookEntryID != "" {
		return pricebookEntryID, nil
	}

	query := fmt.Sprintf(
		"SELECT Id FROM PricebookEntry WHERE Pricebook2Id = '%s' AND Product2Id = '%s' AND CurrencyIsoCode = '%s' AND IsActive = true LIMIT 2",
		escapeSOQLString(scope.Pricebook2ID),
		escapeSOQLString(defaultProduct),
		escapeSOQLString(scope.CurrencyIsoCode),
	)
	resp, err := sendRequestTo(ctx, targetNexusCRM, "/query", http.MethodGet, map[string]string{"q": query}, nil)
	if err != nil {
		return "", err
	}

	var result nexusCRMPricebookEntryQuery
	if err := json.Unmarshal(resp, &result); err != nil {
		return "", fmt.Errorf("decode NexusCRM PricebookEntry query: %w", err)
	}
	if !result.Done || result.TotalSize != 1 || len(result.Records) != 1 || strings.TrimSpace(result.Records[0].ID) == "" {
		return "", fmt.Errorf(
			"expected exactly one active PricebookEntry for pricebook %s, product %s, and currency %s, got totalSize=%d records=%d done=%t",
			scope.Pricebook2ID,
			defaultProduct,
			scope.CurrencyIsoCode,
			result.TotalSize,
			len(result.Records),
			result.Done,
		)
	}

	if sfs.pricebookEntryByScope == nil {
		sfs.pricebookEntryByScope = make(map[string]string)
	}
	pricebookEntryID := strings.TrimSpace(result.Records[0].ID)
	sfs.pricebookEntryByScope[cacheKey] = pricebookEntryID
	return pricebookEntryID, nil
}

func buildOrderItemPayload(order *model.CloudOrder, pricebookEntryID string) (model.CreatedOrderItemReq, error) {
	if order == nil {
		return model.CreatedOrderItemReq{}, errors.New("cloud order is required")
	}

	orderID := strings.TrimSpace(order.NewOrderID)
	accountID := strings.TrimSpace(order.AccountID)
	serviceDate := strings.TrimSpace(order.OrderStartDate)
	endDate := strings.TrimSpace(order.OrderEndDate)
	netFee := strings.TrimSpace(order.NetFee)
	pricebookEntryID = strings.TrimSpace(pricebookEntryID)
	if orderID == "" || accountID == "" || serviceDate == "" || endDate == "" || netFee == "" || pricebookEntryID == "" {
		return model.CreatedOrderItemReq{}, errors.New("order item requires order, account, dates, net fee, and PricebookEntry identities")
	}
	if _, ok := new(big.Rat).SetString(netFee); !ok {
		return model.CreatedOrderItemReq{}, fmt.Errorf("order item net fee is not an exact decimal: %q", netFee)
	}

	return model.CreatedOrderItemReq{
		OrderID:          orderID,
		PricebookEntryID: pricebookEntryID,
		Quantity:         json.Number("1"),
		UnitPrice:        json.Number(netFee),
		ServiceDate:      serviceDate,
		EndDate:          endDate,
		NetFee:           json.Number(netFee),
		AccountID:        accountID,
	}, nil
}

func equalOrderItemDecimal(actual, expected json.Number) bool {
	actualValue, actualOK := new(big.Rat).SetString(string(actual))
	expectedValue, expectedOK := new(big.Rat).SetString(string(expected))
	return actualOK && expectedOK && actualValue.Cmp(expectedValue) == 0
}

func validateOrderItemReadback(record nexusCRMOrderItemRecord, expected model.CreatedOrderItemReq) error {
	var mismatches []string
	if strings.TrimSpace(record.ID) == "" {
		mismatches = append(mismatches, "Id")
	}
	if record.IsDeleted {
		mismatches = append(mismatches, "IsDeleted")
	}
	if record.OrderID != expected.OrderID {
		mismatches = append(mismatches, "OrderId")
	}
	if record.PricebookEntryID != expected.PricebookEntryID {
		mismatches = append(mismatches, "PricebookEntryId")
	}
	if !equalOrderItemDecimal(record.Quantity, expected.Quantity) {
		mismatches = append(mismatches, "Quantity")
	}
	if !equalOrderItemDecimal(record.UnitPrice, expected.UnitPrice) {
		mismatches = append(mismatches, "UnitPrice")
	}
	if record.ServiceDate != expected.ServiceDate {
		mismatches = append(mismatches, "ServiceDate")
	}
	if record.EndDate != expected.EndDate {
		mismatches = append(mismatches, "EndDate")
	}
	if !equalOrderItemDecimal(record.NetFee, expected.NetFee) {
		mismatches = append(mismatches, "Net_fees__c")
	}
	if record.AccountID != expected.AccountID {
		mismatches = append(mismatches, "Account__c")
	}
	if len(mismatches) != 0 {
		return fmt.Errorf("NexusCRM OrderItem readback mismatch: %s", strings.Join(mismatches, ", "))
	}
	return nil
}

func queryOrderItems(ctx context.Context, orderID, pricebookEntryID string) ([]nexusCRMOrderItemRecord, error) {
	query := fmt.Sprintf(
		"SELECT Id, OrderId, PricebookEntryId, Quantity, UnitPrice, ServiceDate, EndDate, Net_fees__c, Account__c, IsDeleted FROM OrderItem WHERE OrderId = '%s' AND PricebookEntryId = '%s' LIMIT 2",
		escapeSOQLString(orderID),
		escapeSOQLString(pricebookEntryID),
	)
	resp, err := sendRequestTo(ctx, targetNexusCRM, "/query", http.MethodGet, map[string]string{"q": query}, nil)
	if err != nil {
		return nil, err
	}

	var result nexusCRMOrderItemQuery
	if err := json.Unmarshal(resp, &result); err != nil {
		return nil, fmt.Errorf("decode NexusCRM OrderItem query: %w", err)
	}
	if !result.Done || result.TotalSize != len(result.Records) {
		return nil, fmt.Errorf("incomplete NexusCRM OrderItem query: totalSize=%d records=%d done=%t", result.TotalSize, len(result.Records), result.Done)
	}
	return result.Records, nil
}

func (sfs *salesforceService) ensureOrderItem(ctx context.Context, order *model.CloudOrder) (string, error) {
	if order == nil {
		return "", errors.New("cloud order is required")
	}
	log.L(ctx).Infof("ensure standard order item in NexusCRM:%s--%s", order.BillMonth, order.AdjustTenantID)

	pricebookEntryID, err := sfs.resolveDefaultPricebookEntryID(ctx, order.NewOrderID)
	if err != nil {
		return "", err
	}
	expected, err := buildOrderItemPayload(order, pricebookEntryID)
	if err != nil {
		return "", err
	}

	records, err := queryOrderItems(ctx, expected.OrderID, expected.PricebookEntryID)
	if err != nil {
		return "", err
	}
	if len(records) > 1 {
		return "", fmt.Errorf("found %d active NexusCRM OrderItems for order %s and PricebookEntry %s", len(records), expected.OrderID, expected.PricebookEntryID)
	}
	if len(records) == 1 {
		if err := validateOrderItemReadback(records[0], expected); err != nil {
			return "", err
		}
		return records[0].ID, nil
	}

	body, err := json.Marshal(expected)
	if err != nil {
		return "", fmt.Errorf("encode NexusCRM OrderItem: %w", err)
	}
	resp, err := sendRequestTo(ctx, targetNexusCRM, "/sobjects/OrderItem", http.MethodPost, nil, body)
	if err != nil {
		return "", err
	}
	var created model.CommonSalesResp
	if err := json.Unmarshal(resp, &created); err != nil {
		return "", fmt.Errorf("decode NexusCRM OrderItem create response: %w", err)
	}
	if !created.Success || strings.TrimSpace(created.ID) == "" {
		return "", errors.New("create NexusCRM OrderItem failed")
	}

	readbackResp, err := sendRequestTo(ctx, targetNexusCRM, "/sobjects/OrderItem/"+created.ID, http.MethodGet, nil, nil)
	if err != nil {
		return "", err
	}
	var readback nexusCRMOrderItemRecord
	if err := json.Unmarshal(readbackResp, &readback); err != nil {
		return "", fmt.Errorf("decode NexusCRM OrderItem readback: %w", err)
	}
	if readback.ID != created.ID {
		return "", fmt.Errorf("NexusCRM OrderItem readback identity mismatch: created=%s readback=%s", created.ID, readback.ID)
	}
	if err := validateOrderItemReadback(readback, expected); err != nil {
		return "", err
	}

	records, err = queryOrderItems(ctx, expected.OrderID, expected.PricebookEntryID)
	if err != nil {
		return "", err
	}
	if len(records) != 1 || records[0].ID != created.ID {
		return "", fmt.Errorf("NexusCRM OrderItem idempotency readback expected only %s, got %d records", created.ID, len(records))
	}
	if err := validateOrderItemReadback(records[0], expected); err != nil {
		return "", err
	}

	return created.ID, nil
}

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

	url := "/sobjects/Revenue__c"

	var name, id string

	if revrec.Type == 0 && revrec.Order == "" {
		if err := sfs.prepareRevrecOrderForSalesforce(ctx, revrec); err != nil {
			log.L(ctx).Errorf("prepare revrec order failed", zap.Error(err))

			return "", "", err
		}
		//body := model.CreatedRevrecReq{
		//	Date:                     revrec.Date,
		//	Amount:                   revrec.AdjustedRevenueAmount,
		//	TiDBCloudBillingEndDate:  revrec.BillingEndDate,
		//	NewOrderStarterAmount: revrec.AdjustedStarterAmount,
		//	NewOrderDedicatedAmount:  revrec.AdjustedDedicatedAmount,
		//	NewOrderSupportAmount:    revrec.AdjustedSupportAmount,
		//	CreditAmount:             revrec.CreditAmount,
		//	AmountBeforeAdjustUse:    revrec.RevenueAmount,
		//	CurrencyIsoCode:          revrec.CurrencyIsoCode,
		//	ExchangeRateToUse:        revrec.ExchangeRateToUse,
		//}
		//err = upSfFields(ctx, url, revrec.RevrecID, body)
		//if err != nil {
		//	log.L(ctx).Errorf("update revrec failed", zap.Error(err))
		//
		//	return "", "", err
		//}
		//name = revrec.RevrecNumber
		//id = revrec.RevrecID
	}
	body := buildRevrecPayload(revrec)
	body.Order = revrec.Order
	body.Status = "Confirmed"
	body.Product = defaultProduct

	b, _ := json.Marshal(body)

	resp, err := sendRequest(ctx, url, http.MethodPost, nil, b)
	if err != nil {
		log.L(ctx).Errorf("create revrec failed", zap.Error(err))

		return "", "", err
	}

	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")
	}

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

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

	return id, name, nil
}

func (sfs *salesforceService) updateRevrec(ctx context.Context, rvrc *model.CloudRevrec) error {
	log.L(ctx).Infof("update revrec to salesforce:%s--%s", rvrc.BillMonth, rvrc.AdjustTenantID)

	if strings.TrimSpace(rvrc.RevrecID) == "" {
		return errors.New("cannot update revrec without salesforce revrec id")
	}

	url := "/sobjects/Revenue__c"
	body := buildRevrecPayload(rvrc)
	err := upSfFields(ctx, url, rvrc.RevrecID, body)
	if err != nil {
		log.L(ctx).Errorf("update revrec failed", zap.Error(err))

		return err
	}

	rvrc.Sync = 1
	rvrc.SyncTime = time.Now()
	err = sfs.store.CloudRevrec().Update(ctx, rvrc)
	if err != nil {
		log.L(ctx).Errorf("update revrec failed", zap.Error(err))

		return err
	}

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

	return nil
}

func buildRevrecPayload(revrec *model.CloudRevrec) model.CreatedRevrecReq {
	body := model.CreatedRevrecReq{
		Date:                    revrec.Date,
		TiDBCloudBillingEndDate: revrecBillingEndDateForSalesforce(revrec),
		NewOrderStarterAmount:   revrec.AdjustedStarterAmount,
		NewOrderDedicatedAmount: revrec.AdjustedDedicatedAmount,
		NewOrderSupportAmount:   revrec.AdjustedSupportAmount,
		NewOrderEssentialAmount: revrec.AdjustedEssentialAmount,
		NewOrderPremiumAmount:   revrec.AdjustedPremiumAmount,
		NewOrderLakeAmount:      revrec.AdjustedLakeAmount,
		NewOrderByocAmount:      revrec.AdjustedByocAmount,
		CreditAmount:            revrec.CreditAmount,
		AmountBeforeAdjustUse:   revrec.RevenueAmount,
		CurrencyIsoCode:         revrec.CurrencyIsoCode,
		ExchangeRateToUse:       revrec.ExchangeRateToUse,
	}

	if revrec.RevenueRecognitionType == "actual" || revrec.ActualRevenueRecognitionDate != "" {
		body.Amount = firstNonEmpty(revrec.ActualRevenueAmount, revrec.AdjustedRevenueAmount)
		return body
	}

	body.EstimatedRevenueAmount = firstNonEmpty(revrec.EstimatedRevenueAmount, revrec.AdjustedRevenueAmount)
	return body
}

func revrecBillingEndDateForSalesforce(revrec *model.CloudRevrec) string {
	if revrec.BillMonth != "" {
		if date := billMonthEndDate(revrec.BillMonth); date != "" {
			return date
		}
	}
	return revrec.BillingEndDate
}

func firstNonEmpty(values ...string) string {
	for _, value := range values {
		if value != "" {
			return value
		}
	}
	return ""
}

func (sfs *salesforceService) revrecSingle(ctx context.Context, id uint64) error {
	log.L(ctx).Infof("start local revrec completion:%d", id)

	revrec, err := sfs.store.CloudRevrec().GetByID(ctx, id)
	if err != nil {
		log.L(ctx).Errorf("get revrec failed", zap.Error(err))

		return err
	}

	if !IsFormatDate(revrec.Date) || !IsFormatDate(revrec.BillingEndDate) {
		log.L(ctx).Errorf("revrec date format error", zap.Error(err))

		return errors.New("revrec date format error")
	}

	if err = sfs.syncRevrecLocally(ctx, &revrec); err != nil {
		log.L(ctx).Errorf("complete revrec locally failed", zap.Error(err))

		return err
	}

	log.L(ctx).Infof("local revrec completion success:%d", id)

	return nil
}

func (sfs *salesforceService) revrecLastInvoiceMonth(ctx context.Context) error {
	log.L(ctx).Info("start local revrec completion for last invoice month")

	lastMonth := GetLastMonth()

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

		return err
	}
	// err list
	var errList []error
	for i, revrec := range revrecs {
		// count the number of revrecs
		log.L(ctx).Infof("revrec count:%d", i+1)

		if revrec.Sync == 1 {
			continue
		}
		err = sfs.syncRevrecLocally(ctx, revrec)
		if err != nil {
			errList = append(errList, err)
		}
	}

	if len(errList) > 0 {
		log.L(ctx).Error("sync to salesforce failed", zap.Any("errList", errList))

		// return errorList information - return first 3 errors
		var errMsgs []string
		maxErrors := 3
		if len(errList) < maxErrors {
			maxErrors = len(errList)
		}

		for i := 0; i < maxErrors; i++ {
			errMsgs = append(errMsgs, errList[i].Error())
		}

		return fmt.Errorf("local revrec completion failed with %d errors. First %d errors: %v",
			len(errList), maxErrors, errMsgs)
	}

	log.L(ctx).Info("local revrec completion success")

	return nil
}

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

	url := "/sobjects/Invoice_Collection__c"

	if collection.Type == 0 {
		if err := sfs.prepareCollectionOrderForSalesforce(ctx, collection); err != nil {
			log.L(ctx).Errorf("prepare collection order failed", zap.Error(err))

			return "", "", nil, err
		}
	}
	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

	respMsg := string(resp)
	err = json.Unmarshal(resp, &res)
	if err != nil {
		log.L(ctx).Errorf("unmarshal response failed: %s body:%s", err.Error(), respMsg)

		return "", "", nil, err
	}
	if !res.Success {
		log.L(ctx).Errorf("create collection failed: %s", res.Errors)

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

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

	var fileContent []byte
	// 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 "", "", nil, err
		}

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

			return "", "", nil, err
		}

		// blob
		fileContent, err = readFile(pdfPath)
	}

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

	return res.ID, name, fileContent, err
}

func (sfs *salesforceService) updateCollection(ctx context.Context, collection *model.CloudCollection) error {
	log.L(ctx).Infof("update collection to salesforce:%s--%s", collection.BillMonth, collection.AdjustTenantID)
	date := collection.CollectionsDate
	if collection.CollectedDate != "" {
		date = collection.CollectedDate
	}
	url := "/sobjects/Invoice_Collection__c"
	body := model.CreatedCollectionReq{
		PayableAmount:     collection.PayableAmount,
		InvoiceNumber:     collection.InvoiceNumber,
		PayableDate:       collection.PayableDate,
		ActualBillingDate: collection.ActualBillingDate,
		CollectionsDate:   date,
		CollectionStatus:  collection.CollectionStatus,
		CollectionsAmount: collection.CollectionsAmount,
		StripeInvoicePDF:  collection.Files,
	}
	err := upSfFields(ctx, url, collection.CollectionID, body)
	if err != nil {
		log.L(ctx).Errorf("update collection failed", zap.Error(err))

		return err
	}

	collection.Sync = 1
	collection.SyncTime = time.Now()
	err = sfs.store.CloudCollection().Update(ctx, collection)
	if err != nil {
		log.L(ctx).Errorf("update collection failed", zap.Error(err))

		return err
	}

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

	return nil
}

func (sfs *salesforceService) collectionSingle(ctx context.Context, id uint64) error {
	log.L(ctx).Infof("start local collection completion:%d", id)

	collection, err := sfs.store.CloudCollection().GetByID(ctx, id)
	if err != nil {
		log.L(ctx).Errorf("get collection failed", zap.Error(err))

		return err
	}

	if !IsFormatDate(collection.PayableDate) || !IsFormatDate(collection.ActualBillingDate) || (!IsFormatDate(collection.CollectionsDate) || !IsFormatDate(collection.CollectedDate)) {
		log.L(ctx).Errorf("collection date format error", zap.Error(err))

		return errors.New("collection date format error")
	}

	if err = sfs.syncCollectionLocally(ctx, &collection); err != nil {
		log.L(ctx).Errorf("complete collection locally failed", zap.Error(err))

		return err
	}

	log.L(ctx).Infof("local collection completion success:%d", id)

	return nil
}

func (sfs *salesforceService) collectionLastInvoiceMonth(ctx context.Context) error {
	log.L(ctx).Info("start local collection completion for last invoice month")

	lastMonth := GetLastMonth()

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

		return err
	}
	// err list
	var errList []error
	for i, collection := range collections {
		// count the number of collections
		log.L(ctx).Infof("collection count:%d", i+1)

		if !needsCollectionLocalSync(collection) {
			continue
		}
		err = sfs.syncCollectionLocally(ctx, collection)
		if err != nil {
			errList = append(errList, err)
		}
	}

	if len(errList) > 0 {
		log.L(ctx).Error("sync to salesforce failed", zap.Any("errList", errList))

		// return errorList information - return first 3 errors
		var errMsgs []string
		maxErrors := 3
		if len(errList) < maxErrors {
			maxErrors = len(errList)
		}

		for i := 0; i < maxErrors; i++ {
			errMsgs = append(errMsgs, errList[i].Error())
		}

		return fmt.Errorf("local collection completion failed with %d errors. First %d errors: %v",
			len(errList), maxErrors, errMsgs)
	}

	log.L(ctx).Info("local collection completion success")

	return nil
}

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 getRevenueByOrderAndBillMonth(ctx context.Context, orderID, billMonth string) (string, string, error) {
	if strings.TrimSpace(orderID) == "" {
		return "", "", nil
	}
	billingEndDate := billMonthEndDate(billMonth)
	if billingEndDate == "" {
		return "", "", nil
	}

	url := "/query"
	q := fmt.Sprintf(
		"select Id, Name, LastModifiedDate from Revenue__c where IsDeleted = false and Order__c = '%s' and Billing_End_Date__c = %s Order by LastModifiedDate desc limit 1",
		escapeSOQLString(orderID),
		billingEndDate,
	)
	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 and bill month 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 {
		return "", "", nil
	}

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

func escapeSOQLString(value string) string {
	value = strings.ReplaceAll(value, `\`, `\\`)
	return strings.ReplaceAll(value, `'`, `\'`)
}

func getInfoById(ctx context.Context, url, id string) (string, error) {
	return getInfoByIdTo(ctx, targetSalesforce, url, id)
}

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

	if id != "" {
		url += "/" + id
	}
	resp, err := sendRequestTo(ctx, target, 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
	}

	return res.Name, nil
}

func upSfFields(ctx context.Context, url, id string, body interface{}) error {
	return upSfFieldsTo(ctx, targetSalesforce, url, id, body)
}

func upSfFieldsTo(ctx context.Context, target, url, id string, body interface{}) error {
	log.L(ctx).Infof("update status by id from %s:%s", target, id)

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

	b, _ := json.Marshal(body)

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

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

		return err
	}

	log.L(ctx).Infof("update status success:%s", string(resp))

	return nil
}

func deleteCo(ctx context.Context, url, id string, body interface{}) error {
	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.MethodDelete, nil, b)

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

		return err
	}

	log.L(ctx).Infof("update status success:%s", string(resp))

	return nil
}

// sendRequest posts to the Salesforce-compatible API. Kept for all existing
// callers (revrec/collection/etc.); delegates to sendRequestTo with the
// salesforce target.
func sendRequest(ctx context.Context, url string, method string, params map[string]string, body []byte) ([]byte, error) {
	return sendRequestTo(ctx, targetSalesforce, url, method, params, body)
}

// sendRequestTo sends a request to the chosen backend. Salesforce and NexusCRM
// share the same Salesforce-compatible REST API; only the domain and OAuth
// token differ, selected by target.
func sendRequestTo(ctx context.Context, target string, url string, method string, params map[string]string, body []byte) ([]byte, error) {
	domainKey, tokenService := "sf.domain", "salesforce"
	urlBase := ""
	if target == targetNexusCRM {
		domainKey, tokenService = "nexuscrm.domain", "nexuscrm"
		if baseURL := strings.TrimRight(viper.GetString("nexuscrm.base_url"), "/"); baseURL != "" {
			apiVersion := viper.GetString("nexuscrm.api_version")
			if apiVersion == "" {
				apiVersion = "v63.0"
			}
			urlBase = baseURL + "/services/data/" + apiVersion
		}
	}

	log.L(ctx).Infof("send request to %s:%s", target, url)
	log.L(ctx).Infof("params:%v", params)

	if urlBase == "" {
		domain := viper.GetString(domainKey)
		if domain == "" {
			return nil, fmt.Errorf("%s is not configured", domainKey)
		}
		urlBase = "https://" + domain + "/services/data/v59.0"
	}

	token, err := tokenCache.GetToken(ctx, tokenService)
	if err != nil {
		return nil, err
	}

	var lastErr error

	// 失败等待10s重试
	for i := 0; i < salesforceRequestMaxAttempt; i++ {
		req, err := http.NewRequestWithContext(ctx, method, urlBase+url, bytes.NewReader(body))
		if err != nil {
			return nil, err
		}

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

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

		resp, err := salesforceHTTPClient.Do(req)
		if err != nil {
			lastErr = fmt.Errorf("send request to %s failed:%s err:%w", target, url, err)
		} else {
			b, readErr := io.ReadAll(resp.Body)
			resp.Body.Close()
			if readErr != nil {
				return nil, readErr
			}
			if resp.StatusCode >= 200 && resp.StatusCode < 300 {
				log.L(ctx).Infof("send request to %s success:%s", target, url)

				return b, nil
			}
			if target == targetNexusCRM && resp.StatusCode == http.StatusTooManyRequests {
				return nil, fmt.Errorf("%w: target=%s path=%s", errCRMRequestLimitExceeded, target, url)
			}
			lastErr = fmt.Errorf("send request to %s failed:%s  code:%v  err:%s", target, url, resp.StatusCode, string(b))
		}

		if i < salesforceRequestMaxAttempt-1 {
			time.Sleep(salesforceRetryDelay)
		}
	}

	if lastErr != nil {
		return nil, lastErr
	}

	return nil, fmt.Errorf("send request to %s failed:%s", target, url)
}

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 downloadFileContentFromURL(ctx context.Context, url string) ([]byte, error) {
	downloadCtx, cancel := context.WithTimeout(ctx, collectionFileDownloadTimeout)
	defer cancel()

	req, err := http.NewRequestWithContext(downloadCtx, http.MethodGet, url, nil)
	if err != nil {
		return nil, err
	}

	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()

	if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices {
		return nil, fmt.Errorf("download file failed: status %d", resp.StatusCode)
	}

	fileContent, err := io.ReadAll(io.LimitReader(resp.Body, collectionFileMaxBytes+1))
	if err != nil {
		return nil, err
	}
	if len(fileContent) > collectionFileMaxBytes {
		return nil, fmt.Errorf("download file exceeded %d byte limit", collectionFileMaxBytes)
	}

	return fileContent, 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,
	}

	if err := sfs.store.AuditLog().Create(auLog); err != nil {
		log.L(ctx).Error("save audit log failed", zap.Error(err))
		return err
	}

	return nil
}

// fix
func (sfs *salesforceService) FixUp(ctx context.Context) {
	// Salesforce retired: revrec fix disabled. Original logic kept (commented)
	// for reference if revrec is later migrated.
	log.L(ctx).Info("FixUp disabled (salesforce retired)")
	/*
		lastMonth := GetLastMonth()
		records, err := sfs.store.CloudRevrec().ListByMonth(ctx, lastMonth, 0)
		if err != nil {
			log.L(ctx).Errorf("get created revrecs failed", zap.Error(err))

			return
		}
		url := "/sobjects/Revenue__c"
		for _, record := range records {
			if record.Sync == 0 {
				sfs.sfRevrec(ctx, record)
			} else {
				bd := buildRevrecPayload(record)
				upSfFields(ctx, url, record.RevrecID, bd)
			}
			log.L(ctx).Infof("fix revrec success:%s", record.RevrecID)
		}
	*/
}

func (sfs *salesforceService) HandleSingleSF(ctx context.Context, req model.SfReqBody) {
	logger := log.L(ctx)

	// Get the common service to send Lark messages
	cmSrv := newCommon(sfs.store)
	go cmSrv.saveLarkOps(ctx, infoMsg, "sync to salesforce: "+req.Types+" "+strconv.Itoa(int(req.ID)))

	// Sync the entity
	result, err := sfs.SyncSingleSF(ctx, req)
	if err != nil {
		logger.Error("single CRM entity processing failed",
			zap.String("type", req.Types),
			zap.Uint64("id", req.ID),
			zap.Error(err))

		go cmSrv.saveLarkOps(ctx, errorMsg, "single CRM entity processing failed: "+req.Types+" "+strconv.Itoa(int(req.ID))+" msg:"+err.Error())
		return
	}
	if result != nil && result.Status == "processing" {
		go cmSrv.saveLarkOps(ctx, infoMsg, "single CRM entity processing: "+req.Types+" "+strconv.Itoa(int(req.ID))+" msg:"+result.Message)
		return
	}

	go cmSrv.saveLarkOps(ctx, successMsg, "single CRM entity processing success: "+req.Types+" "+strconv.Itoa(int(req.ID)))
}

func (sfs *salesforceService) salesforceSingle(ctx context.Context, req model.SfReqBody) error {
	result, err := sfs.SyncSingleSF(ctx, req)
	if err != nil {
		return err
	}
	if result != nil && result.Status == "processing" {
		return errors.New(result.Message)
	}
	return nil
}

func (sfs *salesforceService) SyncSingleSF(ctx context.Context, req model.SfReqBody) (*model.SfSyncResult, error) {
	logger := log.L(ctx)
	logger.Info("processing single CRM entity",
		zap.String("type", req.Types),
		zap.Uint64("id", req.ID))

	// Handle the entity based on its type
	var result *model.SfSyncResult
	var err error
	switch req.Type {
	case model.Order:
		// Order single-sync now targets NexusCRM (orderSingleResult -> sfOrder).
		result, err = sfs.orderSingleResult(ctx, req.ID)
	case model.Revrec:
		err = sfs.revrecSingle(ctx, req.ID)
		result = &model.SfSyncResult{
			Type:    req.Types,
			ID:      req.ID,
			Status:  "synced",
			Message: "revrec finalized locally; Salesforce create skipped",
		}
	case model.Collection:
		err = sfs.collectionSingle(ctx, req.ID)
		result = &model.SfSyncResult{
			Type:    req.Types,
			ID:      req.ID,
			Status:  "synced",
			Message: "collection finalized locally; Salesforce create skipped",
		}
	default:
		logger.Error("salesforce type error", zap.String("type", req.Types))
		return nil, errors.New("salesforce type error")
	}

	if err != nil {
		logger.Error("single CRM entity processing failed",
			zap.String("type", req.Types),
			zap.Uint64("id", req.ID),
			zap.Error(err))
		return result, err
	}

	logger.Info("single CRM entity processed successfully",
		zap.String("type", req.Types),
		zap.Uint64("id", req.ID))
	return result, nil
}

func (sfs *salesforceService) HandleLastInvoiceMonthSF(ctx context.Context, req model.SfReqBody) {
	logger := log.L(ctx)

	// Get the common service to send Lark messages
	cmSrv := newCommon(sfs.store)
	go cmSrv.saveLarkOps(ctx, infoMsg, "start processing last month "+req.Types+"!")

	// Sync the entities
	err := sfs.SalesforceLastInvoiceMonth(ctx, req)
	if err != nil {
		logger.Error("last month CRM entity processing failed",
			zap.String("type", req.Types),
			zap.Error(err))

		go cmSrv.saveLarkOps(ctx, errorMsg, "processing last month "+req.Types+" failed!"+err.Error())
		return
	}

	go cmSrv.saveLarkOps(ctx, successMsg, "processing last month "+req.Types+" success!")
}

func (sfs *salesforceService) SalesforceLastInvoiceMonth(ctx context.Context, req model.SfReqBody) error {
	logger := log.L(ctx)
	logger.Info("processing last month CRM entities", zap.String("type", req.Types))

	// Save an audit log
	//if err := sfs.saveAuditLog(ctx, 22, "", "type-"+req.Types); err != nil {
	//	logger.Error("save audit log failed", zap.Error(err))
	//	return err
	//}

	// Handle the entities based on their type
	var err error
	switch req.Type {
	case model.Order:
		// Order now syncs to NexusCRM (see createOrder/updateOrder/sfOrder).
		err = sfs.orderLastInvoiceMonth(ctx, req)
	case model.Revrec:
		// Salesforce creation is retired. Downstream ingestion is keyed by RevRec,
		// so completion neither resolves nor rewrites a parent Order.
		err = sfs.revrecLastInvoiceMonth(ctx)
	case model.Collection:
		// Salesforce creation is retired, but local field enrichment, invoice file
		// content, and completion state remain part of the downstream contract.
		err = sfs.collectionLastInvoiceMonth(ctx)
	default:
		logger.Error("salesforce type error", zap.String("type", req.Types))
		return errors.New("salesforce type error")
	}

	if err != nil {
		log.L(ctx).Errorf("last month CRM entity processing failed:%s", req.Types)

		return err
	}

	log.L(ctx).Infof("last month CRM entity processing success:%s", req.Types)

	return nil
}

// string 日期是否是YYYY-MM-DD格式
func IsFormatDate(date string) bool {
	if date == "" {
		return true
	}
	_, err := time.Parse("2006-01-02", date)
	if err != nil {
		return false
	}

	return true
}

func readFile(filePath string) ([]byte, error) {
	file, err := os.Open(filePath)
	if err != nil {
		return nil, fmt.Errorf("无法打开文件: %w", err)
	}
	defer file.Close()

	data, err := ioutil.ReadAll(file)
	if err != nil {
		return nil, fmt.Errorf("无法读取文件内容: %w", err)
	}

	return data, nil
}

func (sfs *salesforceService) Fix(ctx context.Context) {
	// Salesforce retired: revrec fix disabled. Original logic kept (commented)
	// for reference if revrec is later migrated.
	log.L(ctx).Info("Fix disabled (salesforce retired)")
	/*
		lastMonth := GetLastMonth()
		records, err := sfs.store.CloudRevrec().ListByMonth(ctx, lastMonth, -1)
		if err != nil {
			log.L(ctx).Errorf("get created revrecs failed", zap.Error(err))

			return
		}
		url := "/sobjects/Revenue__c"
		for _, record := range records {
			// 只更新已同步到 Salesforce 的 revrec
			if record.Sync == 1 && record.RevrecID != "" {
				bd := model.CreatedRevrecReq{
					Date:                    record.Date,
					TiDBCloudBillingEndDate: record.BillingEndDate,
				}
				err := upSfFields(ctx, url, record.RevrecID, bd)
				if err != nil {
					log.L(ctx).Errorf("update revrec failed: %s, error: %v", record.RevrecID, err)
					continue
				}
				log.L(ctx).Infof("fix revrec success:%s", record.RevrecID)
			}
		}
	*/
}