更新

@@ -39,6 +39,7 @@ func (c CompanyServices) addCompany(param *command.SaveCompanyCommand) error {
 		Status:        param.Comapany.Status,
 		Status:        param.Comapany.Status,
 		UpdateAt:      nowTime,
 		UpdateAt:      nowTime,
 		CreateAt:      nowTime,
 		CreateAt:      nowTime,
+		ChargeUserIds: []int64{},
 		DeleteAt:      nil,
 		DeleteAt:      nil,
 	}
 	}
 
 
@@ -169,3 +170,100 @@ func (c CompanyServices) editCompany(param *command.SaveCompanyCommand) error {
 	}
 	}
 	return nil
 	return nil
 }
 }
+
+func (srv CompanyServices) setCompanyCharge(param *command.SetCompanyCharge) 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()
+	}()
+
+	companyRepo := factory.CreateCompanyRepository(map[string]interface{}{
+		"transactionContext": transactionContext,
+	})
+
+	_, companyList, err := companyRepo.Find(map[string]interface{}{
+		"id":    param.CompanyId,
+		"limit": 1,
+	})
+
+	if err != nil {
+		return err
+	}
+	for i := range companyList {
+		companyList[i].ChargeUserIds = param.ChargeUserIds
+		companyList[i].UpdateAt = time.Now()
+		_, err = companyRepo.Update(companyList[i])
+		if err != nil {
+			return err
+		}
+	}
+	if err := transactionContext.CommitTransaction(); err != nil {
+		return application.ThrowError(application.TRANSACTION_ERROR, err.Error())
+	}
+	return nil
+}
+
+//changeAdmin
+//从BusinessAdmins 接收消息 变更主管
+func (srv CompanyServices) changeAdmin(param *command.ChangeAdminCommand) 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()
+	}()
+
+	userRepo := factory.CreateUserRepository(map[string]interface{}{
+		"transactionContext": transactionContext,
+	})
+	//查找公司对应的管理员用户
+	_, userList, err := userRepo.Find(map[string]interface{}{
+		"limit":     10,
+		"companyId": param.CompanyId,
+		"adminType": domain.UserTypeManager,
+	})
+	if err != nil {
+		return err
+	}
+	//修改旧管理员 为普通用户
+	for i := range userList {
+		userList[i].AdminType = domain.UserTypeCommon
+		userList[i].UpdateAt = time.Now()
+		_, err := userRepo.Update(userList[i])
+		if err != nil {
+			return err
+		}
+	}
+	//获取新管理员
+	_, userList2, err := userRepo.Find(map[string]interface{}{
+		"limit":     1,
+		"companyId": param.CompanyId,
+		"account":   param.UserAccount,
+	})
+	if err != nil {
+		return err
+	}
+	//修改为管理员用户
+	for i := range userList2 {
+		userList[i].AdminType = domain.UserTypeManager
+		userList[i].UpdateAt = time.Now()
+		_, err := userRepo.Update(userList[i])
+		if err != nil {
+			return err
+		}
+	}
+	if err := transactionContext.CommitTransaction(); err != nil {
+		return application.ThrowError(application.TRANSACTION_ERROR, err.Error())
+	}
+	return nil
+}

+package command
+
+type AddDepartmentCommand struct {
+	Id            int64   `json:"id" `        // 组织id
+	CompanyId     int64   `json:"company_id"` // 公司编号
+	Level         int     `json:"level"`      // 组织名称
+	Name          string  `json:"name"`       // 组织名称
+	ParentId      int64   `json:"parent_id"`  // 组织父级id
+	ChargeUserIds []int64 `json:"charge"`     // 主管uids
+	Path          string  `json:"path"`       // 组织路径
+}

+package department
+
+type SyncDataDepartmentService struct{}
+
+//AddDepartment
+//从BusinessAdmins 接收消息 添加部门
+//module="department" action="add"
+func (srv SyncDataDepartmentService) AddDepartment() error {
+	return nil
+}
+
+//EditDepartment
+//从BusinessAdmins 接收消息 编辑部门
+//module="department" action="edit"
+func (srv SyncDataDepartmentService) EditDepartment() error {
+	return nil
+}
+
+//batchDelete
+//从BusinessAdmins 接收消息 删除部门
+//module="department" action="batchDelete"
+func (srv SyncDataDepartmentService) batchDelete() error {
+	return nil
+}
+
+//importDepartment
+//从BusinessAdmins 接收消息 导入部门数据
+//module="department" action="import"
+func (srv SyncDataDepartmentService) importDepartment() error {
+	return nil
+}

+package command
+
+type BatchDeleteCommand struct {
+	// 用户账号
+	Uids []int64 `json:"ids"`
+}

+package command
+
+// 批量禁用、启用用户
+type BatchForbidCommand struct {
+	// 用户账号
+	Uids []int64 `json:"ids"`
+	//用户状态
+	Status int `json:"status"`
+}

+package command
+
+type ImportUserCommand struct {
+	AddUsers  []SaveUserCommand `json:"add"`
+	EditUsers []SaveUserCommand `json:"edit"`
+}

@@ -9,11 +9,12 @@ import (
 	"gitlab.fjmaimaimai.com/allied-creation/performance/pkg/domain"
 	"gitlab.fjmaimaimai.com/allied-creation/performance/pkg/domain"
 )
 )
 
 
-type UserService struct{}
+type SyncDataUserService struct{}
 
 
 //AddUser
 //AddUser
 //从BusinessAdmins 接收消息 添加用户
 //从BusinessAdmins 接收消息 添加用户
-func (srv UserService) addUser(param command.SaveUserCommand) error {
+//module="employee" action="add"
+func (srv SyncDataUserService) AddUser(param command.SaveUserCommand) error {
 	transactionContext, err := factory.CreateTransactionContext(nil)
 	transactionContext, err := factory.CreateTransactionContext(nil)
 	if err != nil {
 	if err != nil {
 		return application.ThrowError(application.TRANSACTION_ERROR, err.Error())
 		return application.ThrowError(application.TRANSACTION_ERROR, err.Error())
@@ -52,7 +53,8 @@ func (srv UserService) addUser(param command.SaveUserCommand) error {
 
 
 //UpdateUser
 //UpdateUser
 //从BusinessAdmins 接收消息 更新用户
 //从BusinessAdmins 接收消息 更新用户
-func (srv UserService) updateUser(param command.SaveUserCommand) error {
+//module="employee" action="edit"
+func (srv SyncDataUserService) UpdateUser(param command.SaveUserCommand) error {
 	transactionContext, err := factory.CreateTransactionContext(nil)
 	transactionContext, err := factory.CreateTransactionContext(nil)
 	if err != nil {
 	if err != nil {
 		return application.ThrowError(application.TRANSACTION_ERROR, err.Error())
 		return application.ThrowError(application.TRANSACTION_ERROR, err.Error())
@@ -109,9 +111,13 @@ func (srv UserService) updateUser(param command.SaveUserCommand) error {
 	return nil
 	return nil
 }
 }
 
 
-//changeAdmin
-//从BusinessAdmins 接收消息 变更主管
-func (srv UserService) changeAdmin(param *command.ChangeAdminCommand) error {
+//batchDelete
+//从BusinessAdmins 接收消息 删除用户
+//module="employee" action="batchDelete"
+func (srv SyncDataUserService) batchDelete(param command.BatchDeleteCommand) error {
+	if len(param.Uids) == 0 {
+		return nil
+	}
 	transactionContext, err := factory.CreateTransactionContext(nil)
 	transactionContext, err := factory.CreateTransactionContext(nil)
 	if err != nil {
 	if err != nil {
 		return application.ThrowError(application.TRANSACTION_ERROR, err.Error())
 		return application.ThrowError(application.TRANSACTION_ERROR, err.Error())
@@ -122,14 +128,64 @@ func (srv UserService) changeAdmin(param *command.ChangeAdminCommand) error {
 	defer func() {
 	defer func() {
 		_ = transactionContext.RollbackTransaction()
 		_ = transactionContext.RollbackTransaction()
 	}()
 	}()
+	userRepo := factory.CreateUserRepository(map[string]interface{}{
+		"transactionContext": transactionContext,
+	})
 
 
+	err = userRepo.Remove(param.Uids)
+	if err != nil {
+		return err
+	}
 	if err := transactionContext.CommitTransaction(); err != nil {
 	if err := transactionContext.CommitTransaction(); err != nil {
+		return err
+	}
+	return nil
+}
+
+//batchForbid
+//从BusinessAdmins 接收消息 禁用,启用用户
+//module="employee" action="batchForbid"
+func (srv SyncDataUserService) batchForbid(param command.BatchForbidCommand) error {
+	if len(param.Uids) == 0 {
+		return nil
+	}
+	transactionContext, err := factory.CreateTransactionContext(nil)
+	if err != nil {
 		return application.ThrowError(application.TRANSACTION_ERROR, err.Error())
 		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()
+	}()
+	userRepo := factory.CreateUserRepository(map[string]interface{}{
+		"transactionContext": transactionContext,
+	})
+	_, userList, err := userRepo.Find(map[string]interface{}{
+		"ids":   param.Uids,
+		"limit": len(param.Uids),
+	})
+	if err != nil {
+		return err
+	}
+	for i := range userList {
+		userList[i].Status = param.Status
+		_, err = userRepo.Update(userList[i])
+		if err != nil {
+			return err
+		}
+	}
+	if err := transactionContext.CommitTransaction(); err != nil {
+		return err
+	}
 	return nil
 	return nil
 }
 }
 
 
-func (srv UserService) setCompanyCharge(param *command.ChangeAdminCommand) error {
+//importUser
+//从BusinessAdmins 接收消息 导入用户数据
+//module="employee" action="import"
+func (srv SyncDataUserService) importUser(param command.ImportUserCommand) error {
 	transactionContext, err := factory.CreateTransactionContext(nil)
 	transactionContext, err := factory.CreateTransactionContext(nil)
 	if err != nil {
 	if err != nil {
 		return application.ThrowError(application.TRANSACTION_ERROR, err.Error())
 		return application.ThrowError(application.TRANSACTION_ERROR, err.Error())
@@ -140,9 +196,64 @@ func (srv UserService) setCompanyCharge(param *command.ChangeAdminCommand) error
 	defer func() {
 	defer func() {
 		_ = transactionContext.RollbackTransaction()
 		_ = transactionContext.RollbackTransaction()
 	}()
 	}()
+	userRepo := factory.CreateUserRepository(map[string]interface{}{
+		"transactionContext": transactionContext,
+	})
 
 
+	editUserMap := map[int64]command.SaveUserCommand{}
+	var editUserIds []int64
+	for i := range param.EditUsers {
+		editUserIds = append(editUserIds, param.EditUsers[i].Id)
+		editUserMap[param.EditUsers[i].Id] = param.EditUsers[i]
+	}
+	_, editUserList, err := userRepo.Find(map[string]interface{}{
+		"ids": editUserIds,
+	})
+	if err != nil {
+		return err
+	}
+	nowTime := time.Now()
+	for i := range editUserList {
+		mVal, ok := editUserMap[editUserList[i].Id]
+		if !ok {
+			continue
+		}
+		editUserList[i].Account = mVal.Phone
+		editUserList[i].AdminType = mVal.AdminType
+		editUserList[i].AvatarUrl = mVal.Avatar
+		editUserList[i].Name = mVal.Name
+		editUserList[i].Status = mVal.Status
+		editUserList[i].CompanyId = mVal.CompanyId
+		editUserList[i].UpdateAt = nowTime
+		_, err = userRepo.Update(editUserList[i])
+		if err != nil {
+			return err
+		}
+	}
+	var (
+		tempUser domain.User
+	)
+	for i := range param.AddUsers {
+		tempUser = domain.User{
+			Id:        param.AddUsers[i].Id,
+			Account:   param.AddUsers[i].Phone,
+			AvatarUrl: param.AddUsers[i].Avatar,
+			CompanyId: param.AddUsers[i].CompanyId,
+			AdminType: param.AddUsers[i].AdminType,
+			Name:      param.AddUsers[i].Name,
+			Status:    param.AddUsers[i].Status,
+			UpdateAt:  nowTime,
+			DeleteAt:  nil,
+			CreateAt:  nowTime,
+		}
+		_, err := userRepo.Insert(&tempUser)
+		if err != nil {
+			return err
+		}
+	}
 	if err := transactionContext.CommitTransaction(); err != nil {
 	if err := transactionContext.CommitTransaction(); err != nil {
-		return application.ThrowError(application.TRANSACTION_ERROR, err.Error())
+		return err
 	}
 	}
+
 	return nil
 	return nil
 }
 }

@@ -4,7 +4,7 @@ package constant
 
 
 import "os"
 import "os"
 
 
-var KAFKA_HOSTS = "" // 1.116.151.79:9092
+var KAFKA_HOSTS = "127.7.8.1:9094" // 1.116.151.79:9092
 
 
 var KAFKA_GROUP_ID = "performance_dev"
 var KAFKA_GROUP_ID = "performance_dev"
 
 

@@ -6,6 +6,7 @@ type Company struct {
 	Id            int64     //公司编号
 	Id            int64     //公司编号
 	Logo          string    //公司logo
 	Logo          string    //公司logo
 	Name          string    //公司名称
 	Name          string    //公司名称
+	ChargeUserIds []int64   //公司级别的部门主管uids
 	Status        int       //公司状态，1正常 2禁用
 	Status        int       //公司状态，1正常 2禁用
 	UpdateAt      time.Time //更新时间
 	UpdateAt      time.Time //更新时间
 	CreateAt      time.Time //创建时间
 	CreateAt      time.Time //创建时间

+package domain
+
+import "time"
+
+type Department struct {
+	Id            int64      // 组织id
+	CompanyId     int64      // 公司编号
+	Level         int        // 组织级别
+	Name          string     // 组织名称
+	ParentId      int64      // 组织父级id
+	ChargeUserIds []int64    // 主管uids
+	Path          string     // 组织路径
+	CreateAt      time.Time  // 创建时间
+	UpdateAt      time.Time  // 更新时间
+	DeleteAt      *time.Time // 删除时间
+}
+type DepartmentRepository interface {
+	Insert(param *Department) (*Department, error)
+	Update(param *Department) (*Department, error)
+	Remove(param *Department) (*Department, error)
+	FindOne(queryOptions map[string]interface{}) (*Department, error)
+	Find(queryOptions map[string]interface{}) (int, []*Department, error)
+}

-package domain
-
-import "time"
-
-type Organization struct {
-	OrganizationId       int64     // 组织id
-	CompanyId            int64     // 公司编号
-	OrganizationLevel    int       // 组织名称
-	OrganizationName     string    // 组织名称
-	OrganizationParentId int64     // 组织父级id
-	OrganizationPath     []int64   // 组织路径
-	ChargeUserIds        []int64   // 主管uids
-	Path                 string    // 组织路径
-	CreateAt             time.Time // 创建时间
-	UpdateAt             time.Time // 更新时间
-}

@@ -15,10 +15,16 @@ type User struct {
 	CreateAt  time.Time
 	CreateAt  time.Time
 }
 }
 
 
+//1普通员工 2 主管理员
+const (
+	UserTypeCommon  int = 1
+	UserTypeManager int = 2
+)
+
 type UserRepository interface {
 type UserRepository interface {
 	Insert(user *User) (*User, error)
 	Insert(user *User) (*User, error)
 	Update(user *User) (*User, error)
 	Update(user *User) (*User, error)
-	Remove(user *User) (*User, error)
+	Remove(userId []int64) error
 	FindOne(queryOptions map[string]interface{}) (*User, error)
 	FindOne(queryOptions map[string]interface{}) (*User, error)
 	Find(queryOptions map[string]interface{}) (int, []*User, error)
 	Find(queryOptions map[string]interface{}) (int, []*User, error)
 }
 }

@@ -7,6 +7,7 @@ type Company struct {
 	Id            int64      `pg:"pk:id"` //公司id
 	Id            int64      `pg:"pk:id"` //公司id
 	Logo          string     //公司logo
 	Logo          string     //公司logo
 	Name          string     //公司名称
 	Name          string     //公司名称
+	ChargeUserIds []int64    //公司级别的部门主管uids
 	Status        int        //公司状态，1正常 2禁用
 	Status        int        //公司状态，1正常 2禁用
 	UpdateAt      time.Time  //更新时间
 	UpdateAt      time.Time  //更新时间
 	CreateAt      time.Time  //创建时间
 	CreateAt      time.Time  //创建时间

+package models
+
+import (
+	"time"
+)
+
+type Department struct {
+	tableName     struct{}   `pg:"department"`
+	Id            int64      `pg:"pk:id"` // 组织id
+	CompanyId     int64      // 公司编号
+	Level         int        // 组织级别
+	Name          string     // 组织名称
+	ParentId      int64      // 组织父级id
+	ChargeUserIds []int64    // 主管uids
+	Path          string     // 组织路径
+	CreateAt      time.Time  // 创建时间
+	UpdateAt      time.Time  // 更新时间
+	DeleteAt      *time.Time // 删除时间
+}

@@ -27,6 +27,7 @@ func (repo *CompanyRepository) Insert(u *domain.Company) (*domain.Company, error
 		Logo:          u.Logo,
 		Logo:          u.Logo,
 		Name:          u.Name,
 		Name:          u.Name,
 		Status:        u.Status,
 		Status:        u.Status,
+		ChargeUserIds: u.ChargeUserIds,
 		UpdateAt:      u.UpdateAt,
 		UpdateAt:      u.UpdateAt,
 		CreateAt:      u.CreateAt,
 		CreateAt:      u.CreateAt,
 		DeleteAt:      u.DeleteAt,
 		DeleteAt:      u.DeleteAt,
@@ -46,6 +47,7 @@ func (repo *CompanyRepository) Update(u *domain.Company) (*domain.Company, error
 		Logo:          u.Logo,
 		Logo:          u.Logo,
 		Name:          u.Name,
 		Name:          u.Name,
 		Status:        u.Status,
 		Status:        u.Status,
+		ChargeUserIds: u.ChargeUserIds,
 		UpdateAt:      u.UpdateAt,
 		UpdateAt:      u.UpdateAt,
 		CreateAt:      u.CreateAt,
 		CreateAt:      u.CreateAt,
 		DeleteAt:      u.DeleteAt,
 		DeleteAt:      u.DeleteAt,

+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"
+	"gitlab.fjmaimaimai.com/allied-creation/performance/pkg/infrastructure/pg/models"
+)
+
+type DepartmentRepository struct {
+	transactionContext *pgTransaction.TransactionContext
+}
+
+var _ domain.DepartmentRepository = (*DepartmentRepository)(nil)
+
+func NewDepartmentRepository(tx *pgTransaction.TransactionContext) *CompanyRepository {
+	return &CompanyRepository{
+		transactionContext: tx,
+	}
+}
+
+func (repo *DepartmentRepository) Insert(u *domain.Department) (*domain.Department, error) {
+	departmentModel := models.Department{
+		Id:            u.Id,
+		CompanyId:     u.CompanyId,
+		Level:         u.Level,
+		Name:          u.Name,
+		ParentId:      u.ParentId,
+		ChargeUserIds: u.ChargeUserIds,
+		Path:          u.Path,
+		CreateAt:      u.CreateAt,
+		UpdateAt:      u.UpdateAt,
+		DeleteAt:      nil,
+	}
+	tx := repo.transactionContext.PgTx
+	_, err := tx.Model(&departmentModel).Insert()
+	if err != nil {
+		return nil, err
+	}
+	u.Id = departmentModel.Id
+	return u, nil
+}
+
+func (repo *DepartmentRepository) Update(u *domain.Department) (*domain.Department, error) {
+	departmentModel := models.Department{
+		Id:            u.Id,
+		CompanyId:     u.CompanyId,
+		Level:         u.Level,
+		Name:          u.Name,
+		ParentId:      u.ParentId,
+		ChargeUserIds: u.ChargeUserIds,
+		Path:          u.Path,
+		CreateAt:      u.CreateAt,
+		UpdateAt:      u.UpdateAt,
+		DeleteAt:      nil,
+	}
+	tx := repo.transactionContext.PgTx
+	_, err := tx.Model(&departmentModel).WherePK().Update()
+	if err != nil {
+		return nil, err
+	}
+	return u, nil
+}
+
+func (repo *DepartmentRepository) Remove(u *domain.Department) (*domain.Department, error) {
+	nowTime := time.Now()
+	u.DeleteAt = &nowTime
+	_, err := repo.Update(u)
+	return u, err
+}
+
+func (repo *DepartmentRepository) FindOne(queryOptions map[string]interface{}) (*domain.Department, error) {
+	tx := repo.transactionContext.PgTx
+	departmentModel := models.Department{}
+	query := tx.Model(&departmentModel)
+	if v, ok := queryOptions["id"]; ok {
+		query.Where("id=?", v)
+	}
+	err := query.First()
+	if err == pg.ErrNoRows {
+		return nil, ErrNoRows
+	}
+	if err != nil {
+		return nil, err
+	}
+	result := repo.TransformToCompanyDomain(&departmentModel)
+	return result, nil
+}
+
+func (repo *DepartmentRepository) Find(queryOptions map[string]interface{}) (int, []*domain.Department, error) {
+	tx := repo.transactionContext.PgTx
+	dparmentModel := []models.Department{}
+	query := tx.Model(&dparmentModel).
+		Limit(20)
+	if v, ok := queryOptions["id"]; ok {
+		query.Where("id=?", v)
+	}
+	if v, ok := queryOptions["limit"]; ok {
+		query.Limit(v.(int))
+	}
+	if v, ok := queryOptions["offset"]; ok {
+		query.Offset(v.(int))
+	}
+	cnt, err := query.SelectAndCount()
+	if err != nil {
+		return 0, nil, err
+	}
+	var resultList []*domain.Department
+	for i := range dparmentModel {
+		result := repo.TransformToCompanyDomain(&dparmentModel[i])
+		resultList = append(resultList, result)
+	}
+	return cnt, resultList, nil
+}
+
+func (repo *DepartmentRepository) TransformToCompanyDomain(u *models.Department) *domain.Department {
+	return &domain.Department{
+		Id:            u.Id,
+		CompanyId:     u.CompanyId,
+		Level:         u.Level,
+		Name:          u.Name,
+		ParentId:      u.ParentId,
+		ChargeUserIds: u.ChargeUserIds,
+		Path:          u.Path,
+		CreateAt:      u.CreateAt,
+		UpdateAt:      u.UpdateAt,
+		DeleteAt:      u.DeleteAt,
+	}
+}

@@ -64,11 +64,15 @@ func (repo *UserRepository) Update(user *domain.User) (*domain.User, error) {
 	return user, nil
 	return user, nil
 }
 }
 
 
-func (repo *UserRepository) Remove(user *domain.User) (*domain.User, error) {
+func (repo *UserRepository) Remove(userId []int64) error {
 	nowTime := time.Now()
 	nowTime := time.Now()
-	user.DeleteAt = &nowTime
-	_, err := repo.Update(user)
-	return user, err
+	tx := repo.transactionContext.PgTx
+	uModel := models.User{}
+	_, err := tx.Model(&uModel).
+		Set("delete_at", nowTime).
+		Where("id in (?)", pg.In(userId)).
+		Update()
+	return err
 }
 }
 
 
 func (repo *UserRepository) FindOne(queryOptions map[string]interface{}) (*domain.User, error) {
 func (repo *UserRepository) FindOne(queryOptions map[string]interface{}) (*domain.User, error) {
@@ -92,17 +96,29 @@ func (repo *UserRepository) FindOne(queryOptions map[string]interface{}) (*domai
 func (repo *UserRepository) Find(queryOptions map[string]interface{}) (int, []*domain.User, error) {
 func (repo *UserRepository) Find(queryOptions map[string]interface{}) (int, []*domain.User, error) {
 	tx := repo.transactionContext.PgTx
 	tx := repo.transactionContext.PgTx
 	userModel := []models.User{}
 	userModel := []models.User{}
-	query := tx.Model(&userModel).
+	query := tx.Model(&userModel).Where("delete_at isnull").
 		Limit(20)
 		Limit(20)
 	if v, ok := queryOptions["id"]; ok {
 	if v, ok := queryOptions["id"]; ok {
 		query.Where("id=?", v)
 		query.Where("id=?", v)
 	}
 	}
-	if v, ok := queryOptions["limit"]; ok {
-		query.Limit(v.(int))
+	if v, ok := queryOptions["ids"]; ok {
+		query.Where("id in(?)", pg.In(v))
+	}
+	if v, ok := queryOptions["companyId"]; ok {
+		query.Where("company_id=?", v)
+	}
+	if v, ok := queryOptions["adminType"]; ok {
+		query.Where("admin_type=?", v)
+	}
+	if v, ok := queryOptions["account"]; ok {
+		query.Where("account like ?", v)
 	}
 	}
 	if v, ok := queryOptions["offset"]; ok {
 	if v, ok := queryOptions["offset"]; ok {
 		query.Offset(v.(int))
 		query.Offset(v.(int))
 	}
 	}
+	if v, ok := queryOptions["limit"]; ok {
+		query.Limit(v.(int))
+	}
 	cnt, err := query.SelectAndCount()
 	cnt, err := query.SelectAndCount()
 	if err != nil {
 	if err != nil {
 		return 0, nil, err
 		return 0, nil, err

@@ -10,7 +10,9 @@ import (
 func Run() {
 func Run() {
 	messageHandlerMap := make(map[string]func(message *sarama.ConsumerMessage) error)
 	messageHandlerMap := make(map[string]func(message *sarama.ConsumerMessage) error)
 	messageHandlerMap["demo-v1"] = Demo
 	messageHandlerMap["demo-v1"] = Demo
-	saramaConsumer.StartConsume(constant.KAFKA_HOSTS, constant.SERVICE_NAME, messageHandlerMap, log.Logger)
+
+	err := saramaConsumer.StartConsume(constant.KAFKA_HOSTS, constant.SERVICE_NAME, messageHandlerMap, log.Logger)
+	log.Logger.Error(err.Error())
 }
 }
 
 
 func Demo(message *sarama.ConsumerMessage) error {
 func Demo(message *sarama.ConsumerMessage) error {