增加职位信息同步

@@ -97,9 +97,9 @@ func (c SyncDataCompanyService) addCompany(param *command.SaveCompanyCommand) er
 		AdminType: param.User.AdminType,
 		Name:      param.User.Name,
 		Status:    param.User.Status,
-		UpdateAt:  nowTime,
-		DeleteAt:  nil,
-		CreateAt:  nowTime,
+		UpdatedAt: nowTime,
+		DeletedAt: nil,
+		CreatedAt: nowTime,
 	}
 	companyRepo := factory.CreateCompanyRepository(map[string]interface{}{
 		"transactionContext": transactionContext,
@@ -172,7 +172,7 @@ func (c SyncDataCompanyService) editCompany(param *command.SaveCompanyCommand) e
 		newUser = userList[0]
 	} else {
 		newUser = &domain.User{
-			CreateAt: nowTime,
+			CreatedAt: nowTime,
 		}
 	}
 
@@ -189,7 +189,7 @@ func (c SyncDataCompanyService) editCompany(param *command.SaveCompanyCommand) e
 	newUser.AdminType = param.User.AdminType
 	newUser.Name = param.User.Name
 	newUser.Status = param.User.Status
-	newUser.UpdateAt = nowTime
+	newUser.UpdatedAt = nowTime
 	if len(companyList) > 0 {
 		_, err = companyRepo.Update(newCompany)
 		if err != nil {
@@ -287,7 +287,7 @@ func (srv SyncDataCompanyService) changeAdmin(param *command.ChangeAdminCommand)
 	//修改旧管理员 为普通用户
 	for i := range userList {
 		userList[i].AdminType = domain.UserTypeCommon
-		userList[i].UpdateAt = time.Now()
+		userList[i].UpdatedAt = time.Now()
 		_, err := userRepo.Update(userList[i])
 		if err != nil {
 			return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
@@ -305,7 +305,7 @@ func (srv SyncDataCompanyService) changeAdmin(param *command.ChangeAdminCommand)
 	//修改为管理员用户
 	for i := range userList2 {
 		userList[i].AdminType = domain.UserTypeManager
-		userList[i].UpdateAt = time.Now()
+		userList[i].UpdatedAt = time.Now()
 		_, err := userRepo.Update(userList[i])
 		if err != nil {
 			return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())

@@ -57,6 +57,14 @@ func CreateDepartmentRepository(options map[string]interface{}) domain.Departmen
 	return repository.NewDepartmentRepository(transactionContext)
 }
 
+func CreatePositionRepository(options map[string]interface{}) domain.PositionRepository {
+	var transactionContext *pg.TransactionContext
+	if value, ok := options["transactionContext"]; ok {
+		transactionContext = value.(*pg.TransactionContext)
+	}
+	return repository.NewPositionRepository(transactionContext)
+}
+
 func CreateRoleRepository(options map[string]interface{}) domain.RoleRepository {
 	var transactionContext *pg.TransactionContext
 	if value, ok := options["transactionContext"]; ok {

+package command
+
+type BatchDeletePositionCommand struct {
+	Ids []int64 `json:"ids"`
+}

+package command
+
+type SavePositionCommand struct {
+	//职位ID
+	Id int64 `json:"id"`
+	//公司ID
+	CompanyId int64 `json:"companyId"`
+	//职位名称
+	Name string `json:"name"`
+	//父级职位ID
+	ParentId int64 `json:"parentId"`
+	//职位路径
+	Path string `json:"path"`
+	//职位层级
+	Level int `json:"level"`
+}

+package service
+
+import (
+	"encoding/json"
+	"github.com/linmadan/egglib-go/core/application"
+	"gitlab.fjmaimaimai.com/allied-creation/performance/pkg/application/factory"
+	"gitlab.fjmaimaimai.com/allied-creation/performance/pkg/application/position/command"
+	"gitlab.fjmaimaimai.com/allied-creation/performance/pkg/domain"
+)
+
+type SyncDataPositionService struct{}
+
+func (service SyncDataPositionService) FromBusinessAdmin(param *domain.MessageBody) error {
+	var err error
+	switch param.Action {
+	//新增-编辑职位
+	case "add", "edit":
+		createPositionCommand := &command.SavePositionCommand{}
+		err = json.Unmarshal(param.Data, createPositionCommand)
+		if err != nil {
+			return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
+		}
+		err = service.CreateOrUpdatePosition(createPositionCommand)
+	//批量删除职位
+	case "batchDelete":
+		batchDeletePositionCommand := &command.BatchDeletePositionCommand{}
+		err = json.Unmarshal(param.Data, batchDeletePositionCommand)
+		if err != nil {
+			return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
+		}
+		err = service.BatchDeletePosition(batchDeletePositionCommand)
+	}
+	return err
+}
+
+// CreateOrUpdatePosition 新增职位
+func (service SyncDataPositionService) CreateOrUpdatePosition(positionCommand *command.SavePositionCommand) error {
+	transactionContext, err := factory.CreateTransactionContext(nil)
+	if err != nil {
+		return application.ThrowError(application.TRANSACTION_ERROR, err.Error())
+	}
+	if err := transactionContext.StartTransaction(); err != nil {
+		return application.ThrowError(application.TRANSACTION_ERROR, err.Error())
+	}
+	defer func() {
+		_ = transactionContext.RollbackTransaction()
+	}()
+	positionRepository := factory.CreatePositionRepository(map[string]interface{}{
+		"transactionContext": transactionContext,
+	})
+	position := &domain.Position{
+		Id:        positionCommand.Id,
+		CompanyId: positionCommand.CompanyId,
+		Name:      positionCommand.Name,
+		ParentId:  positionCommand.ParentId,
+		Path:      positionCommand.Path,
+		Level:     positionCommand.Level,
+	}
+	pdm, _ := positionRepository.FindOne(map[string]interface{}{"id": position.Id})
+	if pdm.Id > 0 {
+		_, err = positionRepository.Update(position)
+	} else {
+		_, err = positionRepository.Insert(position)
+	}
+	if err != nil {
+		return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
+	}
+	if err := transactionContext.CommitTransaction(); err != nil {
+		return application.ThrowError(application.TRANSACTION_ERROR, err.Error())
+	}
+	return nil
+}
+
+// BatchDeletePosition 批量删除职位
+func (service SyncDataPositionService) BatchDeletePosition(batchDeletePositionCommand *command.BatchDeletePositionCommand) error {
+	transactionContext, err := factory.CreateTransactionContext(nil)
+	if err != nil {
+		return application.ThrowError(application.TRANSACTION_ERROR, err.Error())
+	}
+	if err := transactionContext.StartTransaction(); err != nil {
+		return application.ThrowError(application.TRANSACTION_ERROR, err.Error())
+	}
+	defer func() {
+		_ = transactionContext.RollbackTransaction()
+	}()
+	positionRepository := factory.CreatePositionRepository(map[string]interface{}{
+		"transactionContext": transactionContext,
+	})
+	err = positionRepository.Remove(batchDeletePositionCommand.Ids)
+	if err != nil {
+		return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
+	}
+	if err := transactionContext.CommitTransaction(); err != nil {
+		return application.ThrowError(application.TRANSACTION_ERROR, err.Error())
+	}
+	return nil
+}

@@ -8,6 +8,7 @@ type SaveUserCommand struct {
 	AdminType       int    `json:"admin_type"` // 1普通员工  2 主管理员
 	Name            string `json:"name"`       // 用户姓名
 	Status          int    `json:"status"`     // 用户状态（1正常 2禁用）
+	Email           string `json:"email"`      // 邮箱
 	UserDepartments []struct {
 		DepartmentId int `json:"department_id" `
 	} `json:"user_departments"` //用户的组织ids

@@ -83,7 +83,7 @@ func (srv SyncDataUserService) AddUser(param *command.SaveUserCommand) error {
 	defer func() {
 		_ = transactionContext.RollbackTransaction()
 	}()
-	var departmentIds []int
+	departmentIds := make([]int, 0)
 	for _, v := range param.UserDepartments {
 		departmentIds = append(departmentIds, v.DepartmentId)
 	}
@@ -96,10 +96,10 @@ func (srv SyncDataUserService) AddUser(param *command.SaveUserCommand) error {
 		AdminType:    param.AdminType,
 		DepartmentId: departmentIds,
 		Name:         param.Name,
+		Email:        param.Email,
 		Status:       param.Status,
-		UpdateAt:     nowTime,
-		DeleteAt:     nil,
-		CreateAt:     nowTime,
+		UpdatedAt:    nowTime,
+		CreatedAt:    nowTime,
 	}
 	userRepo := factory.CreateUserRepository(map[string]interface{}{
 		"transactionContext": transactionContext,
@@ -146,7 +146,7 @@ func (srv SyncDataUserService) UpdateUser(param *command.SaveUserCommand) error
 		newUser = userList[0]
 	} else {
 		newUser = &domain.User{
-			CreateAt: nowTime,
+			CreatedAt: nowTime,
 		}
 	}
 	newUser.Id = param.Id
@@ -159,7 +159,7 @@ func (srv SyncDataUserService) UpdateUser(param *command.SaveUserCommand) error
 	newUser.PositionId = make([]int, 0)
 	newUser.DepartmentId = make([]int, 0)
 
-	newUser.UpdateAt = nowTime
+	newUser.UpdatedAt = nowTime
 	if len(userList) > 0 {
 		_, err = userRepo.Update(newUser)
 		if err != nil {
@@ -290,7 +290,7 @@ func (srv SyncDataUserService) importUser(param *command.ImportUserCommand) erro
 		editUserList[i].Name = mVal.Name
 		editUserList[i].Status = mVal.Status
 		editUserList[i].CompanyId = mVal.CompanyId
-		editUserList[i].UpdateAt = nowTime
+		editUserList[i].UpdatedAt = nowTime
 		_, err = userRepo.Update(editUserList[i])
 		if err != nil {
 			return err
@@ -308,9 +308,9 @@ func (srv SyncDataUserService) importUser(param *command.ImportUserCommand) erro
 			AdminType: param.AddUsers[i].AdminType,
 			Name:      param.AddUsers[i].Name,
 			Status:    param.AddUsers[i].Status,
-			UpdateAt:  nowTime,
-			DeleteAt:  nil,
-			CreateAt:  nowTime,
+			UpdatedAt: nowTime,
+			DeletedAt: nil,
+			CreatedAt: nowTime,
 		}
 		_, err := userRepo.Insert(&tempUser)
 		if err != nil {

+package domain
+
+type Position struct {
+	//职位ID
+	Id int64 `json:"id"`
+	//公司ID
+	CompanyId int64 `json:"companyId"`
+	//职位名称
+	Name string `json:"name"`
+	//父级职位ID
+	ParentId int64 `json:"parentId"`
+	//职位路径
+	Path string `json:"path"`
+	//职位层级
+	Level int `json:"level"`
+}
+
+type PositionRepository interface {
+	Insert(position *Position) (*Position, error)
+	Update(position *Position) (*Position, error)
+	Remove(ids []int64) error
+	FindOne(queryOption map[string]interface{}) (*Position, error)
+}

@@ -13,9 +13,9 @@ type User struct {
 	Status       int       // 用户状态（1正常 2禁用）
 	DepartmentId []int     // 用户归属的部门
 	PositionId   []int     //用户职位
-	UpdateAt     time.Time // 更新时间
-	DeleteAt     *time.Time
-	CreateAt     time.Time
+	UpdatedAt    time.Time // 更新时间
+	DeletedAt    *time.Time
+	CreatedAt    time.Time
 }
 
 // 1普通员工 2 主管理员

@@ -33,7 +33,7 @@ func init() {
 			&models.ReceivedMessage{},
 			&models.Role{},
 			&models.RoleUser{},
-			&models.Positions{},
+			&models.Position{},
 		}
 		for _, model := range tables {
 			err := DB.Model(model).CreateTable(&orm.CreateTableOptions{

+package models
+
+import "time"
+
+type Position struct {
+	tableName struct{} `pg:"position"`
+	//职位ID
+	Id int64 `pg:",pk"`
+	//公司ID
+	CompanyId int64 `comment:"公司ID"`
+	//职位名称
+	Name string `comment:"职位名称"`
+	//父级职位ID
+	ParentId int64 `comment:"父级职位ID" pg:",use_zero"`
+	//职位路径
+	Path string `comment:"职位路径"`
+	//职位层级
+	Level int `comment:"职位层级" pg:",use_zero"`
+	//创建时间
+	CreatedAt time.Time `comment:"创建时间" pg:"default:now()"`
+	//更新时间
+	UpdatedAt time.Time `comment:"更新时间" pg:"default:now()"`
+	//删除时间
+	DeletedAt *time.Time `comment:"删除时间" pg:",soft_delete"`
+}

-package models
-
-import "time"
-
-type Positions struct {
-	Id        int64  //ID
-	CompanyId int64  //公司ID
-	Name      string //职位名称
-	ParentId  int64  `pg:",use_zero"` //父级职位
-	Path      string //职位路径
-	Level     int    `pg:",use_zero"` //层级
-	Remarks   string //备注
-	CreatedAt time.Time
-	UpdatedAt time.Time
-	DeletedAt *time.Time `pg:",soft_delete"`
-}

+package repository
+
+import (
+	"github.com/go-pg/pg/v10"
+	pgTransaction "github.com/linmadan/egglib-go/transaction/pg"
+	"gitlab.fjmaimaimai.com/allied-creation/performance/pkg/domain"
+	"gitlab.fjmaimaimai.com/allied-creation/performance/pkg/infrastructure/pg/models"
+	"time"
+)
+
+type PositionRepository struct {
+	transactionContext *pgTransaction.TransactionContext
+}
+
+func NewPositionRepository(tx *pgTransaction.TransactionContext) *PositionRepository {
+	return &PositionRepository{
+		transactionContext: tx,
+	}
+}
+
+// Insert 新增职位
+func (repository *PositionRepository) Insert(position *domain.Position) (*domain.Position, error) {
+	model := &models.Position{
+		Id:        position.Id,
+		CompanyId: position.CompanyId,
+		Name:      position.Name,
+		ParentId:  position.ParentId,
+		Path:      position.Path,
+		Level:     position.Level,
+	}
+	_, err := repository.transactionContext.PgTx.Model(model).Insert()
+	return position, err
+}
+
+// Update 更新
+func (repository *PositionRepository) Update(position *domain.Position) (*domain.Position, error) {
+	model := &models.Position{
+		Id:        position.Id,
+		CompanyId: position.CompanyId,
+		Name:      position.Name,
+		ParentId:  position.ParentId,
+		Path:      position.Path,
+		Level:     position.Level,
+		UpdatedAt: time.Now(),
+	}
+	_, err := repository.transactionContext.PgTx.Model(model).WherePK().Update()
+	return position, err
+}
+
+// Remove 删除
+func (repository *PositionRepository) Remove(ids []int64) error {
+	_, err := repository.transactionContext.PgTx.Model(&models.Position{}).Where("id in (?)", pg.In(ids)).Delete()
+	return err
+}
+
+func (repository *PositionRepository) FindOne(queryOption map[string]interface{}) (*domain.Position, error) {
+	position := &models.Position{}
+	query := repository.transactionContext.PgTx.Model(position)
+	if id, ok := queryOption["id"]; ok {
+		query.Where("id = ?", id)
+	}
+	err := query.First()
+	if err != nil {
+		return nil, err
+	}
+	return repository.transferToDomain(position), nil
+}
+
+func (repository *PositionRepository) transferToDomain(model *models.Position) *domain.Position {
+	return &domain.Position{
+		Id:        model.Id,
+		CompanyId: model.CompanyId,
+		Name:      model.Name,
+		ParentId:  model.ParentId,
+		Path:      model.Path,
+		Level:     model.Level,
+	}
+}

 package repository
 
 import (
-	"time"
-
 	"github.com/go-pg/pg/v10"
 	pgTransaction "github.com/linmadan/egglib-go/transaction/pg"
 	"gitlab.fjmaimaimai.com/allied-creation/performance/pkg/domain"
@@ -30,9 +28,9 @@ func (repo *UserRepository) Insert(user *domain.User) (*domain.User, error) {
 		AdminType: user.AdminType,
 		Name:      user.Name,
 		Status:    user.Status,
-		UpdatedAt: user.UpdateAt,
-		CreatedAt: user.CreateAt,
-		DeletedAt: user.DeleteAt,
+		UpdatedAt: user.UpdatedAt,
+		CreatedAt: user.CreatedAt,
+		DeletedAt: user.DeletedAt,
 	}
 	tx := repo.transactionContext.PgTx
 	_, err := tx.Model(&userModel).Insert()
@@ -53,9 +51,9 @@ func (repo *UserRepository) Update(user *domain.User) (*domain.User, error) {
 		Name:      user.Name,
 		Email:     user.Email,
 		Status:    user.Status,
-		UpdatedAt: user.UpdateAt,
-		CreatedAt: user.CreateAt,
-		DeletedAt: user.DeleteAt,
+		UpdatedAt: user.UpdatedAt,
+		CreatedAt: user.CreatedAt,
+		DeletedAt: user.DeletedAt,
 	}
 	tx := repo.transactionContext.PgTx
 	_, err := tx.Model(&userModel).WherePK().Update()
@@ -66,13 +64,9 @@ func (repo *UserRepository) Update(user *domain.User) (*domain.User, error) {
 }
 
 func (repo *UserRepository) Remove(userId []int64) error {
-	nowTime := time.Now()
 	tx := repo.transactionContext.PgTx
 	uModel := models.User{}
-	_, err := tx.Model(&uModel).
-		Set("delete_at", nowTime).
-		Where("id in (?)", pg.In(userId)).
-		Update()
+	_, err := tx.Model(&uModel).Where("id in (?)", pg.In(userId)).Delete()
 	return err
 }
 
@@ -145,8 +139,8 @@ func (repo *UserRepository) TransformToUserDomain(user *models.User) *domain.Use
 		Name:      user.Name,
 		Email:     user.Email,
 		Status:    user.Status,
-		UpdateAt:  user.UpdatedAt,
-		CreateAt:  user.CreatedAt,
-		DeleteAt:  user.DeletedAt,
+		UpdatedAt: user.UpdatedAt,
+		CreatedAt: user.CreatedAt,
+		DeletedAt: user.DeletedAt,
 	}
 }

@@ -2,6 +2,7 @@ package handle
 
 import (
 	"encoding/json"
+	"gitlab.fjmaimaimai.com/allied-creation/performance/pkg/application/position/service"
 	"strconv"
 
 	"github.com/Shopify/sarama"
@@ -87,6 +88,17 @@ func SyncDataBusinessAdmin(msgData *sarama.ConsumerMessage) error {
 			})
 			return err
 		}
+	case "position":
+		positionSrv := service.SyncDataPositionService{}
+		err = positionSrv.FromBusinessAdmin(&msgBody)
+		if err != nil {
+			log.Logger.Error("处理position消息失败"+err.Error(), map[string]interface{}{
+				"module": msgBody.Module,
+				"action": msgBody.Action,
+				"data":   string(msgBody.Data),
+			})
+			return err
+		}
 	}
 	err = msgRepo.SaveMessage(&receivedMsg)
 	if err != nil {