修改:版本 v0.4.0

@@ -3,6 +3,7 @@ module gitlab.fjmaimaimai.com/mmm-go/partner
 go 1.14
 go 1.14
 
 
 require (
 require (
+	github.com/Shopify/sarama v1.26.4
 	github.com/astaxie/beego v1.12.1
 	github.com/astaxie/beego v1.12.1
 	github.com/dgrijalva/jwt-go v3.2.0+incompatible
 	github.com/dgrijalva/jwt-go v3.2.0+incompatible
 	github.com/gin-gonic/gin v1.5.0
 	github.com/gin-gonic/gin v1.5.0
@@ -10,5 +11,5 @@ require (
 	github.com/linmadan/egglib-go v0.0.0-20191217144343-ca4539f95bf9
 	github.com/linmadan/egglib-go v0.0.0-20191217144343-ca4539f95bf9
 	github.com/shiena/ansicolor v0.0.0-20151119151921-a422bbe96644 // indirect
 	github.com/shiena/ansicolor v0.0.0-20151119151921-a422bbe96644 // indirect
 	github.com/shopspring/decimal v1.2.0
 	github.com/shopspring/decimal v1.2.0
-	github.com/tiptok/gocomm v1.0.1
+	github.com/tiptok/gocomm v1.0.2
 )
 )

@@ -168,7 +168,7 @@ func getStatistics(userId int64, transactionContext *transaction.TransactionCont
 	if count, e := PartnerInfoDao.PartnerStatic(map[string]interface{}{"inPartnerIds": partnerIds, "inPartnerCategory": domain.App}); e == nil {
 	if count, e := PartnerInfoDao.PartnerStatic(map[string]interface{}{"inPartnerIds": partnerIds, "inPartnerCategory": domain.App}); e == nil {
 		Statistics["appCount"] = count
 		Statistics["appCount"] = count
 	}
 	}
-	if bonus, e := OrderBaseDao.OrderBonusStatics(domain.OrderBonusQuery{InPartnerIds: partnerIds, OrderType: domain.OrderReal}); e == nil {
+	if bonus, e := OrderBaseDao.OrderBonusStatics(domain.OrderBonusQuery{InPartnerIds: partnerIds, OrderTypes: domain.UserOrderTypes(domain.Career)}); e == nil {
 		Statistics["careerOrdersMoney"] = bonus.TotalOrderAmount
 		Statistics["careerOrdersMoney"] = bonus.TotalOrderAmount
 		Statistics["careerDividend"] = bonus.Bonus
 		Statistics["careerDividend"] = bonus.Bonus
 	}
 	}

@@ -24,7 +24,7 @@ func Statistics(header *protocol.RequestHeader, request *protocol.DividendStatis
 	}()
 	}()
 
 
 	// 事业分红统计-查询订单
 	// 事业分红统计-查询订单
-	_, orderAll, e := OrderBaseResponsitory.Find(utils.ObjectJsonToMap(domain.OrderQueryOption{PartnerId: header.UserId, EndTime: time.Now(), SortByCreateTime: domain.DESC, OrderType: domain.OrderReal}))
+	_, orderAll, e := OrderBaseResponsitory.Find(utils.ObjectJsonToMap(domain.OrderQueryOption{PartnerId: header.UserId, EndTime: time.Now(), SortByCreateTime: domain.DESC, OrderTypes: domain.UserOrderTypes(domain.Career)}))
 	if e != nil {
 	if e != nil {
 		log.Error(e)
 		log.Error(e)
 	}
 	}
@@ -124,7 +124,7 @@ func OrderList(header *protocol.RequestHeader, request *protocol.DividendOrdersR
 	rsp = &protocol.DividendOrdersResponse{List: make([]*protocol.DividendOrderListItem, 0)}
 	rsp = &protocol.DividendOrdersResponse{List: make([]*protocol.DividendOrderListItem, 0)}
 
 
 	count, orders, err = OrderDao.DividendOrders(&domain.DividendOrdersQueryOption{
 	count, orders, err = OrderDao.DividendOrders(&domain.DividendOrdersQueryOption{
-		OrderType:      domain.OrderReal,
+		OrderTypes:     domain.UserOrderTypes(domain.Career),
 		PartnerId:      header.UserId,
 		PartnerId:      header.UserId,
 		DetailAction:   request.DetailAction,
 		DetailAction:   request.DetailAction,
 		DividendAction: request.DividendAction,
 		DividendAction: request.DividendAction,

+package factory
+
+//import (
+//	"fmt"
+//	"github.com/linmadan/egglib-go/core/application"
+//	"github.com/linmadan/egglib-go/message/publisher/kafkaMessage/sarama"
+//	localMessagePublisher "github.com/linmadan/egglib-go/message/publisher/local_message"
+//	localMessageReceiver "github.com/linmadan/egglib-go/message/receiver/local_message"
+//	"suplus-message/pkg/constant"
+//	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/log"
+//)
+//
+//func CreateMessagePublisher(options map[string]interface{}) (application.MessagePublisher, error) {
+//	if localMessageOptions, ok := options["localMessageOptions"]; ok {
+//		var storeType string
+//		var storeOptions map[string]interface{}
+//		if value, ok := localMessageOptions.(map[string]interface{})["storeType"]; ok {
+//			storeType = value.(string)
+//		} else {
+//			return nil, fmt.Errorf("LocalMessagePublisher缺少参数storeType")
+//		}
+//		if value, ok := localMessageOptions.(map[string]interface{})["storeOptions"]; ok {
+//			storeOptions = value.(map[string]interface{})
+//		} else {
+//			return nil, fmt.Errorf("LocalMessagePublisher缺少参数storeOptions")
+//		}
+//		return localMessagePublisher.NewLocalMessagePublisher(storeType, storeOptions)
+//	} else {
+//		return sarama.NewKafkaSaramaMessagePublisher(constant.KAFKA_HOSTS,log.Logger)
+//	}
+//}
+//
+//func CreateMessageReceiver(options map[string]interface{}) (application.MessageReceiver, error) {
+//	if localMessageOptions, ok := options["localMessageOptions"]; ok {
+//		var converterType string
+//		if value, ok := localMessageOptions.(map[string]interface{})["converterType"]; ok {
+//			converterType = value.(string)
+//		} else {
+//			return nil, fmt.Errorf("LocalMessageReceiver缺少参数converterType")
+//		}
+//		var storeType string
+//		var storeOptions map[string]interface{}
+//		if value, ok := localMessageOptions.(map[string]interface{})["storeType"]; ok {
+//			storeType = value.(string)
+//		} else {
+//			return nil, fmt.Errorf("LocalMessageReceiver缺少参数storeType")
+//		}
+//		if value, ok := localMessageOptions.(map[string]interface{})["storeOptions"]; ok {
+//			storeOptions = value.(map[string]interface{})
+//		} else {
+//			return nil, fmt.Errorf("LocalMessageReceiver缺少参数storeOptions")
+//		}
+//		return localMessageReceiver.NewLocalMessageReceiver(converterType, nil, storeType, storeOptions)
+//	} else {
+//		return nil, fmt.Errorf("缺少参数localMessageOptions")
+//	}
+//}

@@ -138,14 +138,14 @@ func Statistics(header *protocol.RequestHeader, request *protocol.OrderStatistic
 	if rsp.Statistics.TodayRealQuantity, rsp.Statistics.TodayRealMoney, err = OrderDao.OrderStatics(&domain.OrderStaticQuery{
 	if rsp.Statistics.TodayRealQuantity, rsp.Statistics.TodayRealMoney, err = OrderDao.OrderStatics(&domain.OrderStaticQuery{
 		BeginTime:  utils.GetDayBegin().Unix() * 1000,
 		BeginTime:  utils.GetDayBegin().Unix() * 1000,
 		EndTime:    utils.GetDayEnd().Unix() * 1000,
 		EndTime:    utils.GetDayEnd().Unix() * 1000,
-		OrderType: domain.OrderReal,
+		OrderTypes: domain.UserOrderTypes(domain.Career),
 		PartnerId:  header.UserId,
 		PartnerId:  header.UserId,
 	}); err != nil {
 	}); err != nil {
 		return
 		return
 	}
 	}
 	if rsp.Statistics.CumulativeQuantity, rsp.Statistics.CumulativeMoney, err = OrderDao.OrderStatics(&domain.OrderStaticQuery{
 	if rsp.Statistics.CumulativeQuantity, rsp.Statistics.CumulativeMoney, err = OrderDao.OrderStatics(&domain.OrderStaticQuery{
 		EndTime:    time.Now().Unix() * 1000,
 		EndTime:    time.Now().Unix() * 1000,
-		OrderType: domain.OrderReal,
+		OrderTypes: domain.UserOrderTypes(domain.Career),
 		PartnerId:  header.UserId,
 		PartnerId:  header.UserId,
 	}); err != nil {
 	}); err != nil {
 		return
 		return
@@ -207,6 +207,7 @@ func List(header *protocol.RequestHeader, request *protocol.OrderListRequest) (r
 		queryOption.EndTime = time.Unix(request.EndTime/1000, 0)
 		queryOption.EndTime = time.Unix(request.EndTime/1000, 0)
 	}
 	}
 	queryOption.OrderType = request.OrderType
 	queryOption.OrderType = request.OrderType
+	queryOption.OrderTypes = request.OrderTypes
 	total, orders, _ = OrderResponsitory.Find(utils.ObjectJsonToMap(queryOption))
 	total, orders, _ = OrderResponsitory.Find(utils.ObjectJsonToMap(queryOption))
 	if len(orders) != 0 {
 	if len(orders) != 0 {
 		for i := range orders {
 		for i := range orders {
@@ -222,7 +223,7 @@ func List(header *protocol.RequestHeader, request *protocol.OrderListRequest) (r
 		//累计实发订单
 		//累计实发订单
 		cumulativeQuantity, _, err = OrderDao.OrderStatics(&domain.OrderStaticQuery{
 		cumulativeQuantity, _, err = OrderDao.OrderStatics(&domain.OrderStaticQuery{
 			EndTime:    time.Now().Unix() * 1000,
 			EndTime:    time.Now().Unix() * 1000,
-			OrderType: domain.OrderReal,
+			OrderTypes: domain.UserOrderTypes(domain.Career),
 			PartnerId:  header.UserId,
 			PartnerId:  header.UserId,
 		})
 		})
 		rsp.Total = cumulativeQuantity
 		rsp.Total = cumulativeQuantity

@@ -65,7 +65,7 @@ func getDetail(userId int64, transactionContext *transaction.TransactionContext)
 		}
 		}
 	}
 	}
 	p.CooperateTime = partner.CooperateTime.Unix() * 1000
 	p.CooperateTime = partner.CooperateTime.Unix() * 1000
-	if bonus, e := OrderBaseDao.OrderBonusStatics(domain.OrderBonusQuery{PartnerId: userId, OrderType: domain.OrderReal}); e == nil {
+	if bonus, e := OrderBaseDao.OrderBonusStatics(domain.OrderBonusQuery{PartnerId: userId, OrderTypes: domain.UserOrderTypes(domain.Career)}); e == nil {
 		p.CareerOrdersCount = int(bonus.Total)
 		p.CareerOrdersCount = int(bonus.Total)
 		p.CareerOrdersMoney = utils.Decimal(bonus.TotalOrderAmount)
 		p.CareerOrdersMoney = utils.Decimal(bonus.TotalOrderAmount)
 		p.CareerDividend = utils.Decimal(bonus.Bonus)
 		p.CareerDividend = utils.Decimal(bonus.Bonus)

@@ -13,6 +13,7 @@ import (
 	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/log"
 	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/log"
 	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/protocol"
 	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/protocol"
 	protocolx "gitlab.fjmaimaimai.com/mmm-go/partner/pkg/protocol/auth"
 	protocolx "gitlab.fjmaimaimai.com/mmm-go/partner/pkg/protocol/auth"
+	userx "gitlab.fjmaimaimai.com/mmm-go/partner/pkg/protocol/user"
 	"strconv"
 	"strconv"
 	"strings"
 	"strings"
 )
 )
@@ -82,10 +83,6 @@ func UserInfo(header *protocol.RequestHeader, request *protocol.UserInfoRequest)
 				Name:  company.Name,
 				Name:  company.Name,
 				Phone: company.Phone,
 				Phone: company.Phone,
 			},
 			},
-			//JoinWay:       partnerInfo.PartnerCategoryInfo(),
-			//District:      map[string]interface{}{"id": partnerInfo.RegionInfo.RegionId, "name": partnerInfo.RegionInfo.RegionName},
-			//SerialNo:      partnerInfo.Id,
-			//CooperateTime: partnerInfo.CooperateTime.Unix() * 1000,
 		}
 		}
 	}
 	}
 	switch header.AdminType {
 	switch header.AdminType {
@@ -205,28 +202,6 @@ func UserInfoV2(header *protocol.RequestHeader, request *protocol.UserInfoReques
 		transactionContext.RollbackTransaction()
 		transactionContext.RollbackTransaction()
 	}()
 	}()
 	rsp = &protocol.UserInfoResponse{}
 	rsp = &protocol.UserInfoResponse{}
-
-	type xcompany struct {
-		Id    int64  `json:"id"`
-		Name  string `json:"name"`
-		Phone string `json:"phone"`
-		//合作区域
-		District interface{} `json:"district"`
-		//合作编码
-		SerialNo int64 `json:"serialNo"`
-		//合作时间
-		CooperateTime int64       `json:"cooperationTime"`
-		Salesman      interface{} `json:"salesman"`
-	}
-	type xuser struct {
-		Id int64 `json:"uid"`
-		//用户名称
-		PartnerName string `json:"uname"`
-		//手机号
-		Phone string `json:"phone"`
-		//合作公司
-		CooperateCompany xcompany `json:"company"`
-	}
 	rspMap := make(map[string]interface{})
 	rspMap := make(map[string]interface{})
 	funcPartnerInfo := func() {
 	funcPartnerInfo := func() {
 		if partnerInfo, err = PartnerInfoService.FindOne(map[string]interface{}{"id": header.UserId}); err != nil {
 		if partnerInfo, err = PartnerInfoService.FindOne(map[string]interface{}{"id": header.UserId}); err != nil {
@@ -237,17 +212,22 @@ func UserInfoV2(header *protocol.RequestHeader, request *protocol.UserInfoReques
 			return
 			return
 		}
 		}
 
 
-		u := xuser{
+		u := userx.User{
 			Id:          partnerInfo.Id,
 			Id:          partnerInfo.Id,
 			PartnerName: partnerInfo.PartnerName,
 			PartnerName: partnerInfo.PartnerName,
 			Phone:       partnerInfo.Account,
 			Phone:       partnerInfo.Account,
-			CooperateCompany: xcompany{
+			CooperateCompany: userx.Company{
 				Id:            company.Id,
 				Id:            company.Id,
 				Name:          company.Name,
 				Name:          company.Name,
 				Phone:         company.Phone,
 				Phone:         company.Phone,
 				SerialNo:      partnerInfo.Id,
 				SerialNo:      partnerInfo.Id,
 				CooperateTime: partnerInfo.CooperateTime.Unix() * 1000,
 				CooperateTime: partnerInfo.CooperateTime.Unix() * 1000,
-				//JoinWay:       partnerInfo.PartnerCategoryInfo(),
+				MiniProgram: map[string]interface{}{
+					"webpageUrl": "www.baidu.com",                                                                     //兼容低版本的网页链接
+					"userName":   "wx3055ea821e62fe8c",                                                                //小程序的原始id
+					"path":       fmt.Sprintf("%v?inviter_id=%v", "pages/index/index", partnerInfo.Id),                //小程序的path
+					"hdImageUrl": "http://suplus-business-admin-test.fjmaimaimai.com/images/default/default_logo.png", //小程序的预览图链接
+				},
 				District: map[string]interface{}{"id": partnerInfo.RegionInfo.RegionId, "name": partnerInfo.RegionInfo.RegionName},
 				District: map[string]interface{}{"id": partnerInfo.RegionInfo.RegionId, "name": partnerInfo.RegionInfo.RegionName},
 			},
 			},
 		}
 		}
@@ -267,14 +247,15 @@ func UserInfoV2(header *protocol.RequestHeader, request *protocol.UserInfoReques
 		if company, err = CompanyResponsitory.FindOne(map[string]interface{}{"id": header.CompanyId}); err != nil {
 		if company, err = CompanyResponsitory.FindOne(map[string]interface{}{"id": header.CompanyId}); err != nil {
 			return
 			return
 		}
 		}
-		rspMap["user"] = xuser{
+		rspMap["user"] = userx.User{
 			Id:          user.Id,
 			Id:          user.Id,
 			PartnerName: user.Name,
 			PartnerName: user.Name,
 			Phone:       user.Phone,
 			Phone:       user.Phone,
-			CooperateCompany: xcompany{
+			CooperateCompany: userx.Company{
 				Id:          company.Id,
 				Id:          company.Id,
 				Name:        company.Name,
 				Name:        company.Name,
 				Phone:       company.Phone,
 				Phone:       company.Phone,
+				MiniProgram: struct{}{},
 			},
 			},
 		}
 		}
 		rsp = rspMap
 		rsp = rspMap

+package constant
+
+const TOPIC_UCENT_USER_CHANGE_PHONE = "ucent-user-changePhone"
+
+const KAFKA_HOSTS = "106.52.15.41:9092"

@@ -11,6 +11,7 @@ const (
 const (
 const (
 	OrderReal          = iota + 1 //实发订单
 	OrderReal          = iota + 1 //实发订单
 	OrderIntention                //意向订单
 	OrderIntention                //意向订单
+	OrderAppletSeafood            //小程序-海鲜干货
 )
 )
 
 
 const (
 const (
@@ -21,3 +22,10 @@ const (
 var (
 var (
 	QueryNoRow = fmt.Errorf("not row found")
 	QueryNoRow = fmt.Errorf("not row found")
 )
 )
+
+// UserOrderTypes
+// @category  合伙人类型
+// @desc 根据用户类别获取用户能看到的订单类型
+func UserOrderTypes(category int) []int {
+	return []int{OrderReal, OrderAppletSeafood}
+}

@@ -141,6 +141,7 @@ func (m *OrderBase) OrderBonusStatic() *OrderStatics {
 type OrderQueryOption struct {
 type OrderQueryOption struct {
 	PartnerId   int64     `json:"partnerId,omitempty"`
 	PartnerId   int64     `json:"partnerId,omitempty"`
 	OrderType   int       `json:"orderType,omitempty"`
 	OrderType   int       `json:"orderType,omitempty"`
+	OrderTypes  []int     `json:"orderTypes,omitempty"`
 	OrderStatus int       `json:"orderStatus,omitempty"`
 	OrderStatus int       `json:"orderStatus,omitempty"`
 	BeginTime   time.Time `json:"beginTime,omitempty"`
 	BeginTime   time.Time `json:"beginTime,omitempty"`
 	EndTime     time.Time `json:"endTime,omitempty"`
 	EndTime     time.Time `json:"endTime,omitempty"`
@@ -154,6 +155,7 @@ type OrderQueryOption struct {
 type DividendOrdersQueryOption struct {
 type DividendOrdersQueryOption struct {
 	PartnerId        int64  `json:"partnerId"`
 	PartnerId        int64  `json:"partnerId"`
 	OrderType        int    `json:"orderType"` //订单类型
 	OrderType        int    `json:"orderType"` //订单类型
+	OrderTypes       []int  `json:"orderTypes,omitempty"`
 	DetailAction     int    `json:"detailAction"`   //明细类型（0已收明细、1未收明细）
 	DetailAction     int    `json:"detailAction"`   //明细类型（0已收明细、1未收明细）
 	DividendAction   int    `json:"dividendAction"` //分红类型（0累计分红、1分红支出）
 	DividendAction   int    `json:"dividendAction"` //分红类型（0累计分红、1分红支出）
 	IsDisable        string `json:"isDisable,omitempty"`
 	IsDisable        string `json:"isDisable,omitempty"`

@@ -11,6 +11,7 @@ type OrderStaticQuery struct {
 	EndTime     int64 `json:"endTime,omitempty"`
 	EndTime     int64 `json:"endTime,omitempty"`
 	OrderStatus int   `json:"orderStatus,omitempty"`
 	OrderStatus int   `json:"orderStatus,omitempty"`
 	OrderType   int   `json:"orderType,omitempty"`
 	OrderType   int   `json:"orderType,omitempty"`
+	OrderTypes  []int `json:"orderTypes,omitempty"`
 	//IsDisable   int   `json:"isDisable,omitempty"`
 	//IsDisable   int   `json:"isDisable,omitempty"`
 }
 }
 
 
@@ -28,6 +29,7 @@ type OrderBonusQuery struct {
 	InPartnerIds []int64 `json:"inPartnerIds,omitempty"`
 	InPartnerIds []int64 `json:"inPartnerIds,omitempty"`
 	IsDisable    int     `json:"isDisable,omitempty"`
 	IsDisable    int     `json:"isDisable,omitempty"`
 	OrderType    int     `json:"orderType,omitempty"`
 	OrderType    int     `json:"orderType,omitempty"`
+	OrderTypes   []int   `json:"orderTypes,omitempty"`
 }
 }
 
 
 // 订单分红统计-应答
 // 订单分红统计-应答

+package domain
+
+// SysMessageConsume
+type SysMessageConsume struct {
+	// 消息ID
+	Id int64 `json:"id"`
+	// 主题
+	Topic string `json:"topic"`
+	// 分区信息
+	Partition int `json:"partition"`
+	// 消息偏移序号
+	Offset int64 `json:"offset"`
+	// 键值
+	Key string `json:"key"`
+	// 消息内容
+	Value string `json:"value"`
+	// 消息时间
+	MsgTime int64 `json:"msgTime"`
+	// 创建时间
+	CreateAt int64 `json:"createAt"`
+	// 状态
+	Status int64 `json:"status"`
+}
+
+type SysMessageConsumeRepository interface {
+	Save(dm *SysMessageConsume) (*SysMessageConsume, error)
+	Remove(dm *SysMessageConsume) (*SysMessageConsume, error)
+	FindOne(queryOptions map[string]interface{}) (*SysMessageConsume, error)
+	Find(queryOptions map[string]interface{}) (int64, []*SysMessageConsume, error)
+}
+
+func (m *SysMessageConsume) Identify() interface{} {
+	if m.Id == 0 {
+		return nil
+	}
+	return m.Id
+}

+package domain
+
+// SysMessageProduce
+type SysMessageProduce struct {
+	// 消息ID
+	Id int64 `json:"id"`
+	// 主题
+	Topic string `json:"topic"`
+	// 分区信息
+	Partition int `json:"partition"`
+	// 消息内容
+	Value string `json:"value"`
+	// 消息时间
+	MsgTime int64 `json:"msgTime"`
+	// 状态
+	Status int64 `json:"status"`
+}
+
+type SysMessageProduceRepository interface {
+	Save(dm *SysMessageProduce) (*SysMessageProduce, error)
+	Remove(dm *SysMessageProduce) (*SysMessageProduce, error)
+	FindOne(queryOptions map[string]interface{}) (*SysMessageProduce, error)
+	Find(queryOptions map[string]interface{}) (int64, []*SysMessageProduce, error)
+}
+
+func (m *SysMessageProduce) Identify() interface{} {
+	if m.Id == 0 {
+		return nil
+	}
+	return m.Id
+}

@@ -30,6 +30,9 @@ func (dao *OrderBaseDao) OrderStatics(option *domain.OrderStaticQuery) (count in
 	if option.OrderType > 0 {
 	if option.OrderType > 0 {
 		q.Where(`"order_base".order_type =?`, option.OrderType)
 		q.Where(`"order_base".order_type =?`, option.OrderType)
 	}
 	}
+	if len(option.OrderTypes) > 0 {
+		q.Where(`"order_base".order_type in (?)`, pg.In(option.OrderTypes))
+	}
 	if option.BeginTime > 0 {
 	if option.BeginTime > 0 {
 		q.Where(`"order_base".create_time >=?`, time.Unix(option.BeginTime/1000, 0))
 		q.Where(`"order_base".create_time >=?`, time.Unix(option.BeginTime/1000, 0))
 	}
 	}
@@ -68,6 +71,9 @@ func (dao *OrderBaseDao) OrderBonusStatics(option domain.OrderBonusQuery) (rsp d
 	if option.OrderType > 0 {
 	if option.OrderType > 0 {
 		q.Where(`"order_base".order_type =?`, option.OrderType)
 		q.Where(`"order_base".order_type =?`, option.OrderType)
 	}
 	}
+	if len(option.OrderTypes) > 0 {
+		q.Where(`"order_base".order_type in (?)`, pg.In(option.OrderTypes))
+	}
 	err = q.Select(&rsp.Total, &rsp.Bonus, &rsp.BonusExpense, &rsp.TotalOrderAmount)
 	err = q.Select(&rsp.Total, &rsp.Bonus, &rsp.BonusExpense, &rsp.TotalOrderAmount)
 	return
 	return
 }
 }
@@ -81,6 +87,9 @@ func (dao *OrderBaseDao) DividendOrders(option *domain.DividendOrdersQueryOption
 	if option.OrderType > 0 {
 	if option.OrderType > 0 {
 		q.Where(`"order_base".order_type=?`, option.OrderType)
 		q.Where(`"order_base".order_type=?`, option.OrderType)
 	}
 	}
+	if len(option.OrderTypes) > 0 {
+		q.Where(`"order_base".order_type in (?)`, pg.In(option.OrderTypes))
+	}
 	if option.PartnerId > 0 {
 	if option.PartnerId > 0 {
 		q.Where(`"order_base".partner_id=?`, option.PartnerId)
 		q.Where(`"order_base".partner_id=?`, option.PartnerId)
 	}
 	}

@@ -112,7 +112,7 @@ A left join
 			plan_order_amount  amount, 
 			plan_order_amount  amount, 
 			(case when use_order_count>=0 then use_partner_bonus else plan_partner_bonus end) bonus,
 			(case when use_order_count>=0 then use_partner_bonus else plan_partner_bonus end) bonus,
 			partner_bonus_expense bonus_expense FROM "order_base" AS "order_base" 
 			partner_bonus_expense bonus_expense FROM "order_base" AS "order_base" 
-			WHERE (partner_id in (?)) and order_type =1
+			WHERE (partner_id in (?)) and (order_type in (?))
 			UNION ALL
 			UNION ALL
 			SELECT partner_info_id partner_id,
 			SELECT partner_info_id partner_id,
 			0 amount, bonus bonus, bonus_expense bonus_expense FROM business_bonus
 			0 amount, bonus bonus, bonus_expense bonus_expense FROM business_bonus
@@ -129,7 +129,7 @@ GROUP BY partner_id
 			sql.WriteString(fmt.Sprintf(" \nOFFSET %v", offset))
 			sql.WriteString(fmt.Sprintf(" \nOFFSET %v", offset))
 		}
 		}
 	}
 	}
-	_, err = tx.Query(&statics, sql.String(), pg.In(partnerIds), pg.In(partnerIds), pg.In(partnerIds))
+	_, err = tx.Query(&statics, sql.String(), pg.In(partnerIds), pg.In(partnerIds), pg.In(domain.UserOrderTypes(domain.Career)), pg.In(partnerIds))
 	return
 	return
 }
 }
 
 

@@ -104,7 +104,7 @@ func (svr *PgLoginService) PartnerStaticInfo() (interface{}, error) {
 	if len(companyList) == 0 {
 	if len(companyList) == 0 {
 		return response, nil
 		return response, nil
 	}
 	}
-	totalBonus, e := OrderDao.OrderBonusStatics(domain.OrderBonusQuery{InPartnerIds: doGetPartnerIds(), OrderType: domain.OrderReal})
+	totalBonus, e := OrderDao.OrderBonusStatics(domain.OrderBonusQuery{InPartnerIds: doGetPartnerIds(), OrderTypes: domain.UserOrderTypes(domain.Career)})
 	if e != nil {
 	if e != nil {
 		return response, e
 		return response, e
 	}
 	}
@@ -153,7 +153,7 @@ func (svr *PgLoginService) PartnerStaticInfo() (interface{}, error) {
 			continue
 			continue
 		}
 		}
 
 
-		bonus, _ := OrderDao.OrderBonusStatics(domain.OrderBonusQuery{PartnerId: partner.Id, OrderType: domain.OrderReal})
+		bonus, _ := OrderDao.OrderBonusStatics(domain.OrderBonusQuery{PartnerId: partner.Id, OrderTypes: domain.UserOrderTypes(domain.Career)})
 		if v, ok := mapPartnerBussinessBonus[partner.Id]; ok {
 		if v, ok := mapPartnerBussinessBonus[partner.Id]; ok {
 			bonus.Bonus += v.Bonus
 			bonus.Bonus += v.Bonus
 		}
 		}
@@ -196,7 +196,7 @@ func (svr *PgLoginService) ManagerStaticInfo() (interface{}, error) {
 	for i := range companyList {
 	for i := range companyList {
 		c := companyList[i]
 		c := companyList[i]
 
 
-		if constant.POSTGRESQL_DB_NAME != "partner_dev1" {
+		if constant.POSTGRESQL_DB_NAME != "partner_dev" {
 			//通过企业平台 校验模块权限
 			//通过企业平台 校验模块权限
 			var user *domain.Users
 			var user *domain.Users
 			for j := range svr.Users {
 			for j := range svr.Users {

+package message
+
+import (
+	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/infrastructure/message/kafkax"
+	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/infrastructure/message/models"
+)
+
+//新消费者-消费组
+func NewConsumer(kafkaHosts string, groupId string) models.Consumer {
+	return kafkax.NewSaramaConsumer(kafkaHosts, groupId)
+}

+package message
+
+import (
+	"fmt"
+	"github.com/Shopify/sarama"
+	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/constant"
+	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/infrastructure/message/models"
+	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/infrastructure/pg"
+	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/infrastructure/pg/transaction"
+	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/infrastructure/utils"
+	"log"
+	"testing"
+)
+
+// 消息持久化
+func TestNewConsumer(t *testing.T) {
+	consumer := NewConsumer(constant.KAFKA_HOSTS, "0")
+	consumer.WithMessageReceiver(NewPgMessageReceiverRepository(transaction.NewPGTransactionContext(pg.DB)))
+	consumer.WithTopicHandler("mmm_xcx_orders", func(message interface{}) error {
+		m, ok := message.(*sarama.Message)
+		if !ok {
+			return nil
+		}
+		if len(m.Value) > 0 {
+			var msg models.Message
+			utils.JsonUnmarshal(string(m.Value), &msg)
+			t.Log("handler message :", string(m.Value), msg.Id, msg.Topic, msg.Value)
+		}
+		return nil
+	})
+	consumer.StartConsume()
+}
+
+// 消息不需要持久化
+func TestNewConsumerNoRepository(t *testing.T) {
+	consumer := NewConsumer(constant.KAFKA_HOSTS, "0")
+	consumer.WithTopicHandler("mmm_xcx_orders", func(message interface{}) error {
+		m, ok := message.(*sarama.Message)
+		if !ok {
+			return nil
+		}
+		if len(m.Value) > 0 {
+			var msg models.Message
+			utils.JsonUnmarshal(string(m.Value), &msg)
+			t.Log("handler message :", string(m.Value), msg.Id, msg.Topic, msg.Value)
+		}
+		return nil
+	})
+	consumer.StartConsume()
+}
+
+type PgMessageReceiverRepository struct {
+	transactionContext *transaction.TransactionContext
+}
+
+func NewPgMessageReceiverRepository(transactionContext *transaction.TransactionContext) *PgMessageReceiverRepository {
+	return &PgMessageReceiverRepository{
+		transactionContext: transactionContext,
+	}
+}
+func (repository *PgMessageReceiverRepository) ReceiveMessage(params map[string]interface{}) error {
+	var num int
+	checkSql := `select count(0) from sys_message_consume where "offset" =? and topic=?`
+	_, err := repository.transactionContext.PgDd.Query(&num, checkSql, params["offset"], params["topic"])
+	if err != nil {
+		return err
+	}
+	if num > 0 {
+		return fmt.Errorf("receive repeate message [%v]", params)
+	}
+
+	sql := `insert into sys_message_consume(topic,partition,"offset",key,value,msg_time,create_at,status)values(?,?,?,?,?,?,?,?)`
+	_, err = repository.transactionContext.PgDd.Exec(sql, params["topic"], params["partition"], params["offset"], params["key"], params["value"], params["msg_time"], params["create_at"], params["status"])
+	return err
+}
+func (repository *PgMessageReceiverRepository) ConfirmReceive(params map[string]interface{}) error {
+	log.Println(params)
+	_, err := repository.transactionContext.PgDd.Exec(`update sys_message_consume set status=? where "offset" =? and topic=?`, int(models.Finished), params["offset"], params["topic"])
+	return err
+}

+package kafkax
+
+import (
+	"context"
+	"fmt"
+	"github.com/Shopify/sarama"
+	"github.com/tiptok/gocomm/identity/idgen"
+	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/infrastructure/message/models"
+	"log"
+	"strings"
+	"sync"
+	"time"
+)
+
+type SaramaConsumer struct {
+	ready             chan bool
+	messageHandlerMap map[string]func(message interface{}) error
+	//Logger            log.Logger
+	kafkaHosts string
+	groupId    string
+	topicMiss  map[string]string //记录未被消费的topic
+	receiver   models.MessageReceiverRepository
+}
+
+func (consumer *SaramaConsumer) Setup(sarama.ConsumerGroupSession) error {
+	close(consumer.ready)
+	return nil
+}
+func (consumer *SaramaConsumer) Cleanup(sarama.ConsumerGroupSession) error {
+	return nil
+}
+func (consumer *SaramaConsumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
+	var err error
+	for message := range claim.Messages() {
+		log.Printf("Message claimed:  timestamp = %v, topic = %s offset = %v value = %v", message.Timestamp, message.Topic, message.Offset, string(message.Value))
+		handler, ok := consumer.messageHandlerMap[message.Topic]
+		consumer.messageReceiveBefore(message)
+		if !ok {
+			continue
+		}
+		if err = handler(message); err == nil {
+			session.MarkMessage(message, "")
+		} else {
+			fmt.Println("Message claimed: kafka消息处理错误 topic =", message.Topic, message.Offset, err)
+		}
+		session.MarkMessage(message, "")
+		if err != nil {
+			continue
+		}
+		consumer.messageReceiveAfter(message)
+	}
+	return err
+}
+
+func (consumer *SaramaConsumer) messageReceiveBefore(message *sarama.ConsumerMessage) {
+	if consumer.receiver == nil {
+		return
+	}
+
+	var params = make(map[string]interface{})
+	var err error
+	_, ok := consumer.messageHandlerMap[message.Topic]
+	if !ok {
+		params["status"] = models.Ignore
+		_, topicMiss := consumer.topicMiss[message.Topic]
+		if !topicMiss {
+			fmt.Printf("topic:[%v] has not consumer handler", message.Topic)
+		}
+		return
+	}
+
+	_, err = consumer.storeMessage(params, message)
+	if err != nil {
+		log.Println("ConsumeClaim:", err)
+	}
+}
+func (consumer *SaramaConsumer) messageReceiveAfter(message *sarama.ConsumerMessage) {
+	if consumer.receiver == nil {
+		return
+	}
+	consumer.finishMessage(map[string]interface{}{"offset": message.Offset, "topic": message.Topic})
+}
+
+func (consumer *SaramaConsumer) storeMessage(params map[string]interface{}, message *sarama.ConsumerMessage) (id int64, err error) {
+	defer func() {
+		if e := recover(); e != nil {
+			log.Println(e)
+		}
+	}()
+	id = idgen.Next()
+	params = make(map[string]interface{})
+	params["id"] = message.Offset
+	params["topic"] = message.Topic
+	params["partition"] = message.Partition
+	params["offset"] = message.Offset
+	params["key"] = string(message.Key)
+	params["value"] = string(message.Value)
+	params["msg_time"] = message.Timestamp.Unix()
+	params["create_at"] = time.Now().Unix()
+	params["status"] = models.UnFinished //0:未完成 1:已完成 2：未命中
+	err = consumer.receiver.ReceiveMessage(params)
+	return
+}
+func (consumer *SaramaConsumer) finishMessage(params map[string]interface{}) error {
+	defer func() {
+		if e := recover(); e != nil {
+			log.Println(e)
+		}
+	}()
+	consumer.receiver.ConfirmReceive(params)
+	return nil
+}
+
+func (consumer *SaramaConsumer) StartConsume() error {
+	config := sarama.NewConfig()
+	config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRoundRobin
+	config.Consumer.Offsets.Initial = sarama.OffsetNewest
+	config.Version = sarama.V0_11_0_0
+	brokerList := strings.Split(consumer.kafkaHosts, ",")
+	consumerGroup, err := sarama.NewConsumerGroup(brokerList, consumer.groupId, config)
+	if err != nil {
+		return err
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
+	consumer.ready = make(chan bool)
+	go func() {
+		defer wg.Done()
+		for {
+			var topics []string
+			for key := range consumer.messageHandlerMap {
+				topics = append(topics, key)
+			}
+			if err := consumerGroup.Consume(ctx, topics, consumer); err != nil {
+				log.Println(err.Error())
+				return
+			}
+			if ctx.Err() != nil {
+				return
+			}
+		}
+	}()
+	<-consumer.ready
+	log.Println("Sarama consumer up and running!...")
+	select {
+	case <-ctx.Done():
+		log.Println("Sarama consumer : context cancelled")
+	}
+	cancel()
+	wg.Wait()
+	if err := consumerGroup.Close(); err != nil {
+		return err
+	}
+	return nil
+}
+func (consumer *SaramaConsumer) WithTopicHandler(topic string, handler func(message interface{}) error) { //*sarama.ConsumerMessage
+	consumer.messageHandlerMap[topic] = handler
+}
+func (consumer *SaramaConsumer) WithMessageReceiver(receiver models.MessageReceiverRepository) {
+	consumer.receiver = receiver
+}
+
+func NewSaramaConsumer(kafkaHosts string, groupId string) models.Consumer {
+	return &SaramaConsumer{
+		kafkaHosts:        kafkaHosts,
+		groupId:           groupId,
+		topicMiss:         make(map[string]string),
+		messageHandlerMap: make(map[string]func(message interface{}) error),
+	}
+}

+package kafkax
+
+import (
+	"encoding/json"
+	"fmt"
+	"github.com/Shopify/sarama"
+	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/infrastructure/message/models"
+	"log"
+	"strings"
+	"time"
+)
+
+// sarame kafka 消息生产
+type KafkaMessageProducer struct {
+	KafkaHosts string
+	LogInfo    models.LogInfo
+}
+
+// 同步发送
+func (engine *KafkaMessageProducer) Publish(messages []*models.Message, option map[string]interface{}) (*models.MessagePublishResult, error) {
+	config := sarama.NewConfig()
+	config.Producer.Return.Successes = true
+	config.Producer.Return.Errors = true
+	config.Producer.Partitioner = sarama.NewRandomPartitioner
+	config.Producer.Retry.Max = 10
+	config.Producer.RequiredAcks = sarama.WaitForAll
+	config.Version = sarama.V0_11_0_0
+	brokerList := strings.Split(engine.KafkaHosts, ",")
+	producer, err := sarama.NewSyncProducer(brokerList, config)
+	if err != nil {
+		return nil, err
+	}
+	defer func() {
+		if err := producer.Close(); err != nil {
+			log.Println(err)
+		}
+	}()
+	var successMessageIds []int64
+	var errMessageIds []int64
+	for _, message := range messages {
+		if value, err := json.Marshal(message); err == nil {
+			msg := &sarama.ProducerMessage{
+				Topic:     message.Topic,
+				Value:     sarama.StringEncoder(value),
+				Timestamp: time.Now(),
+			}
+			partition, offset, err := producer.SendMessage(msg)
+			if err != nil {
+				errMessageIds = append(errMessageIds, message.Id)
+				log.Println(err)
+			} else {
+				successMessageIds = append(successMessageIds, message.Id)
+				var append = make(map[string]interface{})
+				append["topic"] = message.Topic
+				append["partition"] = partition
+				append["offset"] = offset
+				log.Println("kafka消息发送", append)
+			}
+		}
+	}
+	return &models.MessagePublishResult{SuccessMessageIds: successMessageIds, ErrorMessageIds: errMessageIds}, nil
+}
+
+// 消息调度器
+type MessageDispatcher struct {
+	notifications     chan struct{}
+	messageChan       chan *models.Message
+	dispatchTicker    *time.Ticker
+	messageRepository models.MessageRepository
+	producer          models.MessageProducer
+}
+
+func (dispatcher *MessageDispatcher) MessagePublishedNotice() error {
+	time.Sleep(time.Second * 2)
+	dispatcher.notifications <- struct{}{}
+	return nil
+}
+
+func (dispatcher *MessageDispatcher) MessagePublish(messages []*models.Message) error {
+	for i := range messages {
+		dispatcher.messageChan <- messages[i]
+	}
+	return nil
+}
+
+// go dispatcher.Dispatch() 启动一个独立协程
+func (dispatcher *MessageDispatcher) Dispatch() {
+	for {
+		select {
+		case <-dispatcher.dispatchTicker.C:
+			go func(dispatcher *MessageDispatcher) {
+				dispatcher.notifications <- struct{}{}
+			}(dispatcher)
+		case <-dispatcher.notifications:
+			if dispatcher.messageRepository == nil {
+				continue
+			}
+			messages, _ := dispatcher.messageRepository.FindNoPublishedStoredMessages()
+			var messagesInProcessIds []int64
+			for i := range messages {
+				messagesInProcessIds = append(messagesInProcessIds, messages[i].Id)
+			}
+			if messages != nil && len(messages) > 0 {
+				dispatcher.messageRepository.FinishMessagesStatus(messagesInProcessIds, int(models.InProcess))
+
+				reuslt, err := dispatcher.producer.Publish(messages, nil)
+				if err == nil && len(reuslt.SuccessMessageIds) > 0 {
+					dispatcher.messageRepository.FinishMessagesStatus(reuslt.SuccessMessageIds, int(models.Finished))
+				}
+				//发送失败的消息ID列表 更新状态 进行中->未开始
+				if len(reuslt.ErrorMessageIds) > 0 {
+					dispatcher.messageRepository.FinishMessagesStatus(reuslt.ErrorMessageIds, int(models.UnFinished))
+				}
+			}
+		case msg := <-dispatcher.messageChan:
+			dispatcher.producer.Publish([]*models.Message{msg}, nil)
+		}
+	}
+}
+
+type MessageDirector struct {
+	messageRepository models.MessageRepository
+	dispatcher        *MessageDispatcher
+}
+
+func (d *MessageDirector) PublishMessages(messages []*models.Message) error {
+	if d.dispatcher == nil {
+		return fmt.Errorf("dispatcher还没有启动")
+	}
+	if d.messageRepository == nil {
+		d.dispatcher.MessagePublish(messages)
+		return nil
+	}
+	for _, message := range messages {
+		if err := d.messageRepository.SaveMessage(message); err != nil {
+			return err
+		}
+	}
+	if err := d.dispatcher.MessagePublishedNotice(); err != nil {
+		return err
+	}
+	return nil
+}
+
+// 消息发布器
+// options["kafkaHosts"]="localhost:9092"
+// options["timeInterval"]=time.Second*60*5
+func NewMessageDirector(messageRepository models.MessageRepository, options map[string]interface{}) *MessageDirector {
+	dispatcher := &MessageDispatcher{
+		notifications:     make(chan struct{}),
+		messageRepository: messageRepository,
+		messageChan:       make(chan *models.Message, 100),
+	}
+
+	var hosts string
+	if kafkaHosts, ok := options["kafkaHosts"]; ok {
+		hosts = kafkaHosts.(string)
+	} else {
+		hosts = "localhost:9092"
+	}
+	dispatcher.producer = &KafkaMessageProducer{KafkaHosts: hosts, LogInfo: models.DefaultLog}
+
+	if interval, ok := options["timeInterval"]; ok {
+		dispatcher.dispatchTicker = time.NewTicker(interval.(time.Duration))
+	} else {
+		dispatcher.dispatchTicker = time.NewTicker(time.Second * 60 * 5)
+	}
+	go dispatcher.Dispatch()
+
+	return &MessageDirector{
+		messageRepository: messageRepository,
+		dispatcher:        dispatcher,
+	}
+}

+package models
+
+import "log"
+
+// 消息存储-发布
+type MessageRepository interface {
+	SaveMessage(message *Message) error
+	FindNoPublishedStoredMessages() ([]*Message, error)
+	FinishMessagesStatus(messageIds []int64, finishStatus int) error
+}
+
+// 消息存储-接收
+type MessageReceiverRepository interface {
+	ReceiveMessage(params map[string]interface{}) error
+	ConfirmReceive(params map[string]interface{}) error
+}
+
+// 消费者
+type Consumer interface {
+	StartConsume() error
+	WithTopicHandler(topic string, handler func(message interface{}) error)
+	WithMessageReceiver(receiver MessageReceiverRepository)
+}
+
+// 生产者
+type MessageProducer interface {
+	Publish(messages []*Message, option map[string]interface{}) (*MessagePublishResult, error)
+}
+
+type LogInfo func(params ...interface{})
+
+var DefaultLog LogInfo = func(params ...interface{}) {
+	log.Println(params...)
+}

+package models
+
+type Message struct {
+	Id           int64  `json:"id"`
+	Topic        string `json:"topic"`
+	Value        string `json:"value"`
+	MsgTime      int64  `json:"msg_time"`
+	FinishStatus int    `json:"-"` //0:未完成 2：已完成 1：进行中  3：忽略
+}
+
+//结束状态
+type FinishStatus int
+
+const (
+	UnFinished FinishStatus = 0
+	InProcess  FinishStatus = 1
+	Finished   FinishStatus = 2
+	Ignore     FinishStatus = 3
+)
+
+type MessagePublishResult struct {
+	SuccessMessageIds []int64
+	ErrorMessageIds   []int64
+}

+package message
+
+import (
+	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/infrastructure/message/kafkax"
+	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/infrastructure/message/models"
+)
+
+// 消息发布器
+// options["kafkaHosts"]="localhost:9092"
+// options["timeInterval"]=time.Second*60*5
+func NewMessageProducer(messageRepository models.MessageRepository, options map[string]interface{}) *kafkax.MessageDirector {
+	dispatcher := kafkax.NewMessageDirector(messageRepository, options)
+	return dispatcher
+}

+package message
+
+import (
+	"github.com/go-pg/pg/v10"
+	"github.com/tiptok/gocomm/identity/idgen"
+	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/constant"
+	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/infrastructure/message/kafkax"
+	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/infrastructure/message/models"
+	pgDB "gitlab.fjmaimaimai.com/mmm-go/partner/pkg/infrastructure/pg"
+	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/infrastructure/pg/transaction"
+	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/infrastructure/utils"
+	"testing"
+	"time"
+)
+
+// 发布消息本地持久化
+func TestNewMessageProducer(t *testing.T) {
+	var (
+		transactionContext = transaction.NewPGTransactionContext(pgDB.DB)
+		err                error
+	)
+
+	producer := NewMessageProducer(NewPgMessageRepository(transactionContext), map[string]interface{}{"kafkaHosts": constant.KAFKA_HOSTS})
+	err = producer.PublishMessages([]*models.Message{
+		&models.Message{Id: idgen.Next(), Topic: "chat", MsgTime: time.Now().Unix(), Value: "hello world! tip tip!", FinishStatus: 0},
+	})
+
+	if err != nil {
+		return
+	}
+	time.Sleep(time.Second * 2)
+}
+
+// 发布消息无本地持久化
+func TestNewMessageProducerNoRepository(t *testing.T) {
+	var (
+		err error
+	)
+
+	producer := NewMessageProducer(nil, map[string]interface{}{"kafkaHosts": constant.KAFKA_HOSTS})
+	err = producer.PublishMessages([]*models.Message{
+		&models.Message{Id: idgen.Next(), Topic: "chat", MsgTime: time.Now().Unix(), Value: "hello world! tip tip!", FinishStatus: 0},
+	})
+
+	if err != nil {
+		return
+	}
+	time.Sleep(time.Second * 2)
+}
+
+// 简单发布消息
+func TestSampleProducer(t *testing.T) {
+	var producer models.MessageProducer = &kafkax.KafkaMessageProducer{
+		KafkaHosts: constant.KAFKA_HOSTS,
+	}
+	_, err := producer.Publish([]*models.Message{{Id: 22, Topic: "mmm_xcx_orders", MsgTime: time.Now().Unix(), Value: "hello ccc20201009!"}}, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+type PgMessageRepository struct {
+	transactionContext *transaction.TransactionContext
+}
+
+func (repository *PgMessageRepository) SaveMessage(message *models.Message) error {
+	sql := `insert into sys_message_produce (id,topic,value,msg_time,status)values(?,?,?,?,?)`
+	_, err := repository.transactionContext.PgDd.Exec(sql, message.Id, message.Topic, utils.JsonAssertString(message), message.MsgTime, int64(models.UnFinished))
+	return err
+}
+func (repository *PgMessageRepository) FindNoPublishedStoredMessages() ([]*models.Message, error) {
+	sql := `select value from sys_message_produce where status=?`
+	var values []string
+	_, e := repository.transactionContext.PgDd.Query(&values, sql, int64(models.UnFinished))
+	var messages = make([]*models.Message, 0)
+	if e != nil {
+		return messages, nil
+	}
+	for _, v := range values {
+		item := &models.Message{}
+		utils.JsonUnmarshal(v, item)
+		if item.Id != 0 {
+			messages = append(messages, item)
+		}
+	}
+	return messages, nil
+}
+func (repository *PgMessageRepository) FinishMessagesStatus(messageIds []int64, finishStatus int) error {
+	_, err := repository.transactionContext.PgDd.Exec("update sys_message_produce set status=? where id in (?)", finishStatus, pg.In(messageIds))
+	return err
+}
+func NewPgMessageRepository(transactionContext *transaction.TransactionContext) *PgMessageRepository {
+	return &PgMessageRepository{
+		transactionContext: transactionContext,
+	}
+}

@@ -28,6 +28,8 @@ func init() {
 			(*models.PartnerInfo)(nil),
 			(*models.PartnerInfo)(nil),
 			(*models.PartnerSubAccount)(nil),
 			(*models.PartnerSubAccount)(nil),
 			(*models.Company)(nil),
 			(*models.Company)(nil),
+			(*models.SysMessageConsume)(nil),
+			(*models.SysMessageProduce)(nil),
 			(*models.OrderBase)(nil),
 			(*models.OrderBase)(nil),
 			(*models.OrderGood)(nil),
 			(*models.OrderGood)(nil),
 			(*models.ImInfo)(nil),
 			(*models.ImInfo)(nil),

+package models
+
+// SysMessageConsume
+type SysMessageConsume struct {
+	tableName struct{} `pg:"sys_message_consume"`
+	//	消息ID
+	Id int64
+	//	主题
+	Topic string
+	//	分区信息
+	Partition int
+	//	消息偏移序号
+	Offset int64
+	//	键值
+	Key string
+	//	消息内容
+	Value string
+	//	消息时间
+	MsgTime int64
+	//	创建时间
+	CreateAt int64
+	//	状态
+	Status int64
+}

+package models
+
+// SysMessageProduce
+type SysMessageProduce struct {
+	tableName struct{} `pg:"sys_message_produce"`
+	//	消息ID
+	Id int64
+	//	主题
+	Topic string
+	//	分区信息
+	Partition int
+	//	消息内容
+	Value string
+	//	消息时间
+	MsgTime int64
+	//	状态
+	Status int64
+}

 package repository
 package repository
 
 
 import (
 import (
+	"github.com/go-pg/pg/v10"
 	"github.com/go-pg/pg/v10/orm"
 	"github.com/go-pg/pg/v10/orm"
 	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/domain"
 	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/domain"
 	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/infrastructure/pg/models"
 	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/infrastructure/pg/models"
@@ -84,6 +85,9 @@ func (repository *OrderBaseRepository) Find(queryOptions map[string]interface{})
 		SetLimit().
 		SetLimit().
 		SetOrder(`order_base.create_time`, "sortByCreateTime").
 		SetOrder(`order_base.create_time`, "sortByCreateTime").
 		SetOrder(`order_base.update_time`, "sortByUpdateTime")
 		SetOrder(`order_base.update_time`, "sortByUpdateTime")
+	if v, ok := queryOptions["orderTypes"]; ok {
+		query.Where(`"order_base".order_type in (?)`, pg.In(v))
+	}
 	var err error
 	var err error
 	if query.AffectRow, err = query.SelectAndCount(); err != nil {
 	if query.AffectRow, err = query.SelectAndCount(); err != nil {
 		return 0, OrderBases, err
 		return 0, OrderBases, err

+package repository
+
+import (
+	"fmt"
+	"github.com/tiptok/gocomm/common"
+	. "github.com/tiptok/gocomm/pkg/orm/pgx"
+	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/domain"
+	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/infrastructure/pg/models"
+	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/infrastructure/pg/transaction"
+)
+
+type SysMessageConsumeRepository struct {
+	transactionContext *transaction.TransactionContext
+}
+
+func (repository *SysMessageConsumeRepository) Save(dm *domain.SysMessageConsume) (*domain.SysMessageConsume, error) {
+	var (
+		err error
+		m   = &models.SysMessageConsume{}
+		tx  = repository.transactionContext.PgTx
+	)
+	if err = common.GobModelTransform(m, dm); err != nil {
+		return nil, err
+	}
+	if dm.Identify() == nil {
+		if err = tx.Insert(m); err != nil {
+			return nil, err
+		}
+		return dm, nil
+	}
+	if err = tx.Update(m); err != nil {
+		return nil, err
+	}
+	return dm, nil
+}
+
+func (repository *SysMessageConsumeRepository) Remove(SysMessageConsume *domain.SysMessageConsume) (*domain.SysMessageConsume, error) {
+	var (
+		tx                     = repository.transactionContext.PgTx
+		SysMessageConsumeModel = &models.SysMessageConsume{Id: SysMessageConsume.Identify().(int64)}
+	)
+	if _, err := tx.Model(SysMessageConsumeModel).Where("id = ?", SysMessageConsume.Id).Delete(); err != nil {
+		return SysMessageConsume, err
+	}
+	return SysMessageConsume, nil
+}
+
+func (repository *SysMessageConsumeRepository) FindOne(queryOptions map[string]interface{}) (*domain.SysMessageConsume, error) {
+	tx := repository.transactionContext.PgTx
+	SysMessageConsumeModel := new(models.SysMessageConsume)
+	query := NewQuery(tx.Model(SysMessageConsumeModel), queryOptions)
+	query.SetWhere("id = ?", "id")
+	if err := query.First(); err != nil {
+		return nil, fmt.Errorf("query row not found")
+	}
+	if SysMessageConsumeModel.Id == 0 {
+		return nil, fmt.Errorf("query row not found")
+	}
+	return repository.transformPgModelToDomainModel(SysMessageConsumeModel)
+}
+
+func (repository *SysMessageConsumeRepository) Find(queryOptions map[string]interface{}) (int64, []*domain.SysMessageConsume, error) {
+	tx := repository.transactionContext.PgTx
+	var SysMessageConsumeModels []*models.SysMessageConsume
+	SysMessageConsumes := make([]*domain.SysMessageConsume, 0)
+	query := NewQuery(tx.Model(&SysMessageConsumeModels), queryOptions).
+		SetOrder("create_at", "sortByCreateTime")
+	var err error
+	if query.AffectRow, err = query.SelectAndCount(); err != nil {
+		return 0, SysMessageConsumes, err
+	}
+	for _, SysMessageConsumeModel := range SysMessageConsumeModels {
+		if SysMessageConsume, err := repository.transformPgModelToDomainModel(SysMessageConsumeModel); err != nil {
+			return 0, SysMessageConsumes, err
+		} else {
+			SysMessageConsumes = append(SysMessageConsumes, SysMessageConsume)
+		}
+	}
+	return int64(query.AffectRow), SysMessageConsumes, nil
+}
+
+func (repository *SysMessageConsumeRepository) transformPgModelToDomainModel(SysMessageConsumeModel *models.SysMessageConsume) (*domain.SysMessageConsume, error) {
+	m := &domain.SysMessageConsume{}
+	err := common.GobModelTransform(m, SysMessageConsumeModel)
+	return m, err
+}
+
+func NewSysMessageConsumeRepository(transactionContext *transaction.TransactionContext) (*SysMessageConsumeRepository, error) {
+	if transactionContext == nil {
+		return nil, fmt.Errorf("transactionContext参数不能为nil")
+	}
+	return &SysMessageConsumeRepository{transactionContext: transactionContext}, nil
+}

+package repository
+
+import (
+	"fmt"
+	"github.com/tiptok/gocomm/common"
+	. "github.com/tiptok/gocomm/pkg/orm/pgx"
+	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/domain"
+	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/infrastructure/pg/models"
+	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/infrastructure/pg/transaction"
+)
+
+type SysMessageProduceRepository struct {
+	transactionContext *transaction.TransactionContext
+}
+
+func (repository *SysMessageProduceRepository) Save(dm *domain.SysMessageProduce) (*domain.SysMessageProduce, error) {
+	var (
+		err error
+		m   = &models.SysMessageProduce{}
+		tx  = repository.transactionContext.PgTx
+	)
+	if err = common.GobModelTransform(m, dm); err != nil {
+		return nil, err
+	}
+
+	if err = tx.Insert(m); err != nil {
+		return nil, err
+	}
+	return dm, nil
+}
+
+func (repository *SysMessageProduceRepository) Remove(SysMessageProduce *domain.SysMessageProduce) (*domain.SysMessageProduce, error) {
+	var (
+		tx                     = repository.transactionContext.PgTx
+		SysMessageProduceModel = &models.SysMessageProduce{Id: SysMessageProduce.Identify().(int64)}
+	)
+	if _, err := tx.Model(SysMessageProduceModel).Where("id = ?", SysMessageProduce.Id).Delete(); err != nil {
+		return SysMessageProduce, err
+	}
+	return SysMessageProduce, nil
+}
+
+func (repository *SysMessageProduceRepository) FindOne(queryOptions map[string]interface{}) (*domain.SysMessageProduce, error) {
+	tx := repository.transactionContext.PgTx
+	SysMessageProduceModel := new(models.SysMessageProduce)
+	query := NewQuery(tx.Model(SysMessageProduceModel), queryOptions)
+	query.SetWhere("id = ?", "id")
+	if err := query.First(); err != nil {
+		return nil, fmt.Errorf("query row not found")
+	}
+	if SysMessageProduceModel.Id == 0 {
+		return nil, fmt.Errorf("query row not found")
+	}
+	return repository.transformPgModelToDomainModel(SysMessageProduceModel)
+}
+
+func (repository *SysMessageProduceRepository) Find(queryOptions map[string]interface{}) (int64, []*domain.SysMessageProduce, error) {
+	tx := repository.transactionContext.PgDd
+	var SysMessageProduceModels []*models.SysMessageProduce
+	SysMessageProduces := make([]*domain.SysMessageProduce, 0)
+	query := NewQuery(tx.Model(&SysMessageProduceModels), queryOptions).
+		SetWhere("status = ?", "status").
+		SetOrder("update_time", "sortByUpdateTime")
+	var err error
+	if query.AffectRow, err = query.SelectAndCount(); err != nil {
+		return 0, SysMessageProduces, err
+	}
+	for _, SysMessageProduceModel := range SysMessageProduceModels {
+		if SysMessageProduce, err := repository.transformPgModelToDomainModel(SysMessageProduceModel); err != nil {
+			return 0, SysMessageProduces, err
+		} else {
+			SysMessageProduces = append(SysMessageProduces, SysMessageProduce)
+		}
+	}
+	return int64(query.AffectRow), SysMessageProduces, nil
+}
+
+func (repository *SysMessageProduceRepository) transformPgModelToDomainModel(SysMessageProduceModel *models.SysMessageProduce) (*domain.SysMessageProduce, error) {
+	m := &domain.SysMessageProduce{}
+	err := common.GobModelTransform(m, SysMessageProduceModel)
+	return m, err
+}
+
+func NewSysMessageProduceRepository(transactionContext *transaction.TransactionContext) (*SysMessageProduceRepository, error) {
+	if transactionContext == nil {
+		return nil, fmt.Errorf("transactionContext参数不能为nil")
+	}
+	return &SysMessageProduceRepository{transactionContext: transactionContext}, nil
+}

@@ -66,7 +66,7 @@ func (this *OrderController) OrderList() {
 		msg = m
 		msg = m
 		return
 		return
 	}
 	}
-	request.OrderType = domain.OrderReal
+	request.OrderTypes = domain.UserOrderTypes(domain.Career)
 	header := this.GetRequestHeader(this.Ctx)
 	header := this.GetRequestHeader(this.Ctx)
 	msg = protocol.NewReturnResponse(order.List(header, request))
 	msg = protocol.NewReturnResponse(order.List(header, request))
 }
 }

+package messageHandler
+
+import "github.com/Shopify/sarama"
+
+type UcenterMessageCommand struct {
+}
+
+func (c *UcenterMessageCommand) ChangePhoneHandler(message interface{}) error {
+	msg, ok := message.(*sarama.Message)
+	if !ok && msg == nil {
+		return nil
+	}
+	return nil
+}

+package sarama
+
+import (
+	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/constant"
+	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/infrastructure/message/kafkax"
+	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/log"
+	"gitlab.fjmaimaimai.com/mmm-go/partner/pkg/port/sarama/messageHandler"
+	//"suplus-message/pkg/constant"
+	//"suplus-message/pkg/port/sarama/messageHandler"
+)
+
+func Run() {
+	var (
+		ucenterMessage = &messageHandler.UcenterMessageCommand{}
+	)
+
+	saramaConsumer := kafkax.NewSaramaConsumer(constant.KAFKA_HOSTS, constant.SERVICE_NAME)
+	saramaConsumer.WithTopicHandler(constant.TOPIC_UCENT_USER_CHANGE_PHONE, ucenterMessage.ChangePhoneHandler)
+
+	err := saramaConsumer.StartConsume()
+	if err != nil {
+		log.Error(err)
+	}
+}

@@ -71,6 +71,7 @@ type OrderListRequest struct {
 	PageIndex   int   `json:"pageIndex"`
 	PageIndex   int   `json:"pageIndex"`
 	PageSize    int   `json:"pageSize" valid:"Required"`
 	PageSize    int   `json:"pageSize" valid:"Required"`
 	OrderType   int   `json:"-"`
 	OrderType   int   `json:"-"`
+	OrderTypes  []int `json:"-"`
 }
 }
 type OrderListResponse struct {
 type OrderListResponse struct {
 	List       []*OrderListItem       `json:"list"`
 	List       []*OrderListItem       `json:"list"`

+package user
+
+type Company struct {
+	Id    int64  `json:"id"`
+	Name  string `json:"name"`
+	Phone string `json:"phone"`
+	//合作区域
+	District interface{} `json:"district"`
+	//合作编码
+	SerialNo int64 `json:"serialNo"`
+	//合作时间
+	CooperateTime int64       `json:"cooperationTime"`
+	Salesman      interface{} `json:"salesman"`
+	MiniProgram   interface{} `json:"miniProgram"`
+}
+type User struct {
+	Id int64 `json:"uid"`
+	//用户名称
+	PartnerName string `json:"uname"`
+	//手机号
+	Phone string `json:"phone"`
+	//合作公司
+	CooperateCompany Company `json:"company"`
+}