jx-callback/business/controller/controller.go

package controller

import (
	"fmt"
	"time"

	"git.rosy.net.cn/baseapi/utils"
	"git.rosy.net.cn/baseapi/utils/routinepool"
	"git.rosy.net.cn/jx-callback/business/model"
	"git.rosy.net.cn/jx-callback/business/scheduler"
	_ "git.rosy.net.cn/jx-callback/business/scheduler/defsch" // 导入缺省订单调度器
	"git.rosy.net.cn/jx-callback/globals"
	"github.com/astaxie/beego/orm"
)

const (
	pendingOrderGapMax = 2 * 24 * time.Hour // 每次重启机子时，要检查几天内的订单状态
)

var (
	OrderManager   *OrderController
	WaybillManager *WaybillController
	routinePool    *routinepool.Pool
)

func init() {
	routinePool = routinepool.New(1000, 1000)
	OrderManager = NewOrderManager()
	WaybillManager = NewWaybillManager()
	scheduler.CurrentScheduler.RegisterOrderManager(OrderManager)
}

func addOrderOrWaybillStatus(status *model.OrderStatus, db orm.Ormer) (isDuplicated bool, err error) {
	status.ID = 0
	created, _, err := db.ReadOrCreate(status, "VendorOrderID", "VendorID", "OrderType", "VendorStatus", "StatusTime")
	if err == nil {
		if !created {
			globals.SugarLogger.Debugf("duplicated event:%v", status)
			isDuplicated = true
			status.DuplicatedCount++
			utils.CallFuncLogError(func() error {
				_, err = db.Update(status, "DuplicatedCount")
				return err
			}, "addOrderOrWaybillStatus update DuplicatedCount, status:%v", status)
		}
	}
	if err != nil {
		// todo 这里居然会有主键重复错误，逻辑上是不应该的
		globals.SugarLogger.Warnf("addOrderOrWaybillStatus status:%v, access db error:%v", status, err)
	}
	return isDuplicated, err
}

func CallMsgHandler(handler func(), primaryID string) {
	// handler()
	routinePool.CallFun(func() {
		handler()
	}, primaryID)
}

func GetDataCityCodeFromOrder(order *model.GoodsOrder) (retVal string, err error) {
	var sql string
	if order.VendorID == model.VendorIDJD {
		sql = `
		SELECT t2.tel_code
		FROM jxstoremap t0
		JOIN jxstore t1 ON t0.jxstoreid = t1.storeid
		JOIN city t2 ON t1.area = t2.citycode
		WHERE t0.jdstoreid = ?
		`
	} else if order.VendorID == model.VendorIDELM {
		sql = `
		SELECT t2.tel_code
		FROM jx_to_elm_store_map t0
		JOIN jxstore t1 ON t0.jx_store_id = t1.storeid
		JOIN city t2 ON t1.area = t2.citycode
		WHERE t0.elm_store_id = ?
		`
	} else {
		panic(fmt.Sprintf("wrong vendorid:%d", order.VendorID))
	}
	db := orm.NewOrm()
	var lists []orm.ParamsList
	num, err := db.Raw(sql, utils.Str2Int64(order.VendorStoreID)).ValuesList(&lists)
	if err != nil && num == 1 {
		retVal = lists[0][0].(string)
	} else {
		globals.SugarLogger.Errorf("GetDataCityCodeFromOrder can not find store info for vendorID:%d, store:%s", order.VendorID, order.VendorStoreID)
	}
	return retVal, err
}

// todo 可以考虑改成完全按StatusTime来发送事件
func LoadPendingOrders() {
	orders := OrderManager.LoadPendingOrders()
	globals.SugarLogger.Infof("LoadPendingOrders orders count:%d", len(orders))
	bills := WaybillManager.LoadPendingWaybills()
	globals.SugarLogger.Infof("LoadPendingOrders waybills count:%d", len(bills))

	billsMap := map[string][]*model.Waybill{}
	for _, v := range bills {
		savedBills := []*model.Waybill{v}
		if bills, ok := billsMap[v.VendorOrderID]; ok {
			savedBills = append(bills, v)
		}
		billsMap[v.VendorOrderID] = savedBills
	}
	for _, order := range orders {
		isNoNewSent := false
		orderNew := *order
		orderNew.Status = model.OrderStatusNew
		orderNew.StatusTime = order.OrderCreatedAt
		routinePool.CallFunAsync(func() {
			scheduler.CurrentScheduler.OnOrderNew(&orderNew)
		}, order.VendorOrderID)
		for _, bill := range billsMap[order.VendorOrderID] {
			if order.Status > model.OrderStatusNew && !isNoNewSent && order.StatusTime.Sub(bill.WaybillCreatedAt) < 0 {
				isNoNewSent = true
				order2 := *order
				routinePool.CallFunAsync(func() {
					scheduler.CurrentScheduler.OnOrderStatusChanged(model.Order2Status(&order2))
				}, order.VendorOrderID)
			}
			billNew := *bill
			billNew.Status = model.OrderStatusNew
			billNew.StatusTime = order.OrderCreatedAt
			routinePool.CallFunAsync(func() {
				scheduler.CurrentScheduler.OnWaybillStatusChanged(&billNew)
			}, bill.VendorOrderID)
			if order.Status > model.OrderStatusNew && !isNoNewSent && order.StatusTime.Sub(bill.StatusTime) < 0 {
				isNoNewSent = true
				order2 := *order
				routinePool.CallFunAsync(func() {
					scheduler.CurrentScheduler.OnOrderStatusChanged(model.Order2Status(&order2))
				}, order.VendorOrderID)
			}
			if bill.Status > model.WaybillStatusNew {
				bill2 := *bill
				routinePool.CallFunAsync(func() {
					scheduler.CurrentScheduler.OnWaybillStatusChanged(&bill2)
				}, bill.VendorOrderID)
			}
			if order.Status > model.OrderStatusNew && !isNoNewSent {
				isNoNewSent = true
				order2 := *order
				routinePool.CallFunAsync(func() {
					scheduler.CurrentScheduler.OnOrderStatusChanged(model.Order2Status(&order2))
				}, order.VendorOrderID)
			}
		}
		if order.Status > model.OrderStatusNew && !isNoNewSent {
			order2 := *order
			routinePool.CallFunAsync(func() {
				scheduler.CurrentScheduler.OnOrderStatusChanged(model.Order2Status(&order2))
			}, order.VendorOrderID)
		}
	}
}