amocrm/internal/workers/data_updater/data_updater.go

package data_updater

import (
	"amocrm/internal/models"
	"amocrm/internal/tools"
	"amocrm/pkg/amoClient"
	"context"
	"go.uber.org/zap"
	"penahub.gitlab.yandexcloud.net/backend/quiz/common.git/dal"
	"penahub.gitlab.yandexcloud.net/backend/quiz/common.git/model"
	"time"
)

type Deps struct {
	AmoClient *amoClient.Amo
	Repo      *dal.AmoDal
	Logger    *zap.Logger
}

type DataUpdater struct {
	amoClient *amoClient.Amo
	repo      *dal.AmoDal
	logger    *zap.Logger
}

func NewDataUpdaterWC(deps Deps) *DataUpdater {
	return &DataUpdater{
		amoClient: deps.AmoClient,
		repo:      deps.Repo,
		logger:    deps.Logger,
	}
}

func (wc *DataUpdater) Start(ctx context.Context) {
	nextStart := calculateTime()
	ticker := time.NewTicker(time.Second * time.Duration(nextStart))
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			wc.processTasks(ctx)
			nextStart = calculateTime()
			ticker.Reset(time.Second * time.Duration(nextStart))
		case <-ctx.Done():
			return
		}
	}
}

func (wc *DataUpdater) processTasks(ctx context.Context) {
	// обновляем информацию о пользователях
	err := wc.UserUpdater(ctx)
	if err != nil {
		wc.logger.Error("some error from UserUpdater", zap.Error(err))
	}
	// сначала получаем список токенов
	allTokens, err := wc.repo.AmoRepo.GetAllTokens(ctx)
	if err != nil {
		wc.logger.Error("error fetch all tokens from mongo:", zap.Error(err))
		return
	}
	for _, token := range allTokens {
		accountID := 0 // id аккаунта в амо для тегов и кастомных полей так как там не приходит
		// pipelines
		pipelines, err := wc.amoClient.GetListPipelines(token.AccessToken)
		if err != nil {
			wc.logger.Error("error getting list pipelines:", zap.Error(err))
		}
		err = wc.repo.AmoRepo.CheckPipelines(ctx, token.AccountID, tools.ToPipeline(pipelines.Embedded.Pipelines))
		if err != nil {
			wc.logger.Error("error update pipelines in mongo:", zap.Error(err))
		}

		// steps
		for _, pipeline := range pipelines.Embedded.Pipelines {
			accountID = pipeline.AccountID
			steps, err := wc.amoClient.GetListSteps(pipeline.ID, token.AccessToken)
			if err != nil {
				wc.logger.Error("error getting list steps pipeline:", zap.Error(err))
				continue
			}

			err = wc.repo.AmoRepo.CheckSteps(ctx, token.AccountID, tools.ToStep(steps.Embedded.Statuses))
			if err != nil {
				wc.logger.Error("error update pipeline steps in mongo:", zap.Error(err))
			}
		}

		// tags
		var leadsTags []models.Tag
		var contactsTags []models.Tag
		var companiesTags []models.Tag
		var customersTags []models.Tag

		entityTypes := []model.EntityType{model.LeadsTags, model.ContactsTags, model.CompaniesTags, model.CustomersTags}
		for _, entityType := range entityTypes {
			page := 1
			limit := 250

			for {
				req := models.GetListTagsReq{
					Page:       page,
					Limit:      limit,
					EntityType: entityType,
				}
				tags, err := wc.amoClient.GetListTags(req, token.AccessToken)
				if err != nil {
					wc.logger.Error("error getting list of tags", zap.Error(err))
					break
				}

				if tags == nil || len(tags.Embedded.Tags) == 0 {
					break
				}

				switch entityType {
				case model.LeadsTags:
					leadsTags = append(leadsTags, tags.Embedded.Tags...)
				case model.ContactsTags:
					contactsTags = append(contactsTags, tags.Embedded.Tags...)
				case model.CompaniesTags:
					companiesTags = append(companiesTags, tags.Embedded.Tags...)
				case model.CustomersTags:
					customersTags = append(customersTags, tags.Embedded.Tags...)
				}

				page++
			}
		}

		for _, entityType := range entityTypes {
			switch entityType {
			case model.LeadsTags:
				err := wc.repo.AmoRepo.CheckTags(ctx, tools.ToTag(leadsTags, entityType, accountID), token.AccountID)
				if err != nil {
					wc.logger.Error("error update leads tags")
					continue
				}
			case model.ContactsTags:
				err := wc.repo.AmoRepo.CheckTags(ctx, tools.ToTag(contactsTags, entityType, accountID), token.AccountID)
				if err != nil {
					wc.logger.Error("error update contacts tags")
					continue
				}
			case model.CompaniesTags:
				err := wc.repo.AmoRepo.CheckTags(ctx, tools.ToTag(companiesTags, entityType, accountID), token.AccountID)
				if err != nil {
					wc.logger.Error("error update companies tags")
					continue
				}
			case model.CustomersTags:
				err := wc.repo.AmoRepo.CheckTags(ctx, tools.ToTag(customersTags, entityType, accountID), token.AccountID)
				if err != nil {
					wc.logger.Error("error update customer tags")
					continue
				}
			}
		}

		// fields
		var leadsFields []models.CustomField
		var contactsFields []models.CustomField
		var companiesFields []models.CustomField
		var customersFields []models.CustomField
		for _, entityType := range entityTypes {
			page := 1
			limit := 50

			for {
				req := models.GetListFieldsReq{
					Page:       page,
					Limit:      limit,
					EntityType: entityType,
				}
				fields, err := wc.amoClient.GetListFields(req, token.AccessToken)
				if err != nil {
					wc.logger.Error("error getting list of fields", zap.Error(err))
					break
				}

				if fields == nil || len(fields.Embedded.CustomFields) == 0 {
					break
				}

				switch entityType {
				case model.LeadsTags:
					leadsFields = append(leadsFields, fields.Embedded.CustomFields...)
				case model.ContactsTags:
					contactsFields = append(contactsFields, fields.Embedded.CustomFields...)
				case model.CompaniesTags:
					companiesFields = append(companiesFields, fields.Embedded.CustomFields...)
				case model.CustomersTags:
					customersFields = append(customersFields, fields.Embedded.CustomFields...)
				}

				page++
			}
		}

		for _, entityType := range entityTypes {
			switch entityType {
			case model.LeadsTags:
				err := wc.repo.AmoRepo.CheckFields(ctx, tools.ToField(leadsFields, entityType, accountID), token.AccountID)
				if err != nil {
					wc.logger.Error("error update leads fields")
					continue
				}
			case model.ContactsTags:
				err := wc.repo.AmoRepo.CheckFields(ctx, tools.ToField(contactsFields, entityType, accountID), token.AccountID)
				if err != nil {
					wc.logger.Error("error update contacts fields")
					continue
				}
			case model.CompaniesTags:
				err := wc.repo.AmoRepo.CheckFields(ctx, tools.ToField(companiesFields, entityType, accountID), token.AccountID)
				if err != nil {
					wc.logger.Error("error update companies fields")
					continue
				}
			case model.CustomersTags:
				err := wc.repo.AmoRepo.CheckFields(ctx, tools.ToField(companiesFields, entityType, accountID), token.AccountID)
				if err != nil {
					wc.logger.Error("error update customer fields")
					continue
				}
			}
		}
	}
}

func (wc *DataUpdater) UserUpdater(ctx context.Context) error {
	var listUser []models.Users
	page := 1
	limit := 250
	for {
		userData, err := wc.amoClient.GetUserList(models.RequestGetListUsers{
			Page:  page,
			Limit: limit,
		})
		if err != nil {
			wc.logger.Error("error getting user list", zap.Error(err))
			break
		}

		if userData == nil || len(userData.Embedded.Users) == 0 {
			break
		}

		listUser = append(listUser, userData.Embedded.Users...)

		page++
	}

	for _, user := range listUser {
		onlyOneUser, err := wc.amoClient.GetUserByID(user.ID)
		if err != nil {
			wc.logger.Error("error getting user by id", zap.Error(err))
			continue
		}
		err = wc.repo.AmoRepo.CheckUsers(ctx, user.ID, model.User{
			Name:  onlyOneUser.Name,
			Group: tools.ConvertGroups(user),
			Role:  *onlyOneUser.Role,
			Email: onlyOneUser.Email,
		})
	}

	return nil
}

func (wc *DataUpdater) Stop(ctx context.Context) error {
	return nil
}