add amo data validate

go.mod

@@ -13,7 +13,7 @@ require (
 	github.com/twmb/franz-go v1.16.1
 	go.uber.org/zap v1.27.0
 	google.golang.org/protobuf v1.33.0
-	penahub.gitlab.yandexcloud.net/backend/quiz/common.git v0.0.0-20240507152904-b2e6000c1f40
+	penahub.gitlab.yandexcloud.net/backend/quiz/common.git v0.0.0-20240507175756-10399fe4c21f
 	penahub.gitlab.yandexcloud.net/backend/quiz/core.git v0.0.0-20240219174804-d78fd38511af
 )
 

go.sum

@@ -171,5 +171,9 @@ penahub.gitlab.yandexcloud.net/backend/penahub_common v0.0.0-20240223054633-6cb3
 penahub.gitlab.yandexcloud.net/backend/penahub_common v0.0.0-20240223054633-6cb3d5ce45b6/go.mod h1:lTmpjry+8evVkXWbEC+WMOELcFkRD1lFMc7J09mOndM=
 penahub.gitlab.yandexcloud.net/backend/quiz/common.git v0.0.0-20240507152904-b2e6000c1f40 h1:2/7GQGyMcwISIXVj0bapGEy2FDHrI9FskMSqc6peGjM=
 penahub.gitlab.yandexcloud.net/backend/quiz/common.git v0.0.0-20240507152904-b2e6000c1f40/go.mod h1:oRyhT55ctjqp/7ZxIzkR7OsQ7T/NLibsfrbb7Ytns64=
+penahub.gitlab.yandexcloud.net/backend/quiz/common.git v0.0.0-20240507173031-17e2f6acba38 h1:50i28g5lzVthJHS5r+3sPrmRBlfNag8lngN+CkdrbN0=
+penahub.gitlab.yandexcloud.net/backend/quiz/common.git v0.0.0-20240507173031-17e2f6acba38/go.mod h1:oRyhT55ctjqp/7ZxIzkR7OsQ7T/NLibsfrbb7Ytns64=
+penahub.gitlab.yandexcloud.net/backend/quiz/common.git v0.0.0-20240507175756-10399fe4c21f h1:xUo4CsauxNgFhiTfv+5BKfF4Ekk6SHeR+ohwrBuJIrU=
+penahub.gitlab.yandexcloud.net/backend/quiz/common.git v0.0.0-20240507175756-10399fe4c21f/go.mod h1:oRyhT55ctjqp/7ZxIzkR7OsQ7T/NLibsfrbb7Ytns64=
 penahub.gitlab.yandexcloud.net/backend/quiz/core.git v0.0.0-20240219174804-d78fd38511af h1:jQ7HaXSutDX5iepU7VRImxhikK7lV/lBKkiloOZ4Emo=
 penahub.gitlab.yandexcloud.net/backend/quiz/core.git v0.0.0-20240219174804-d78fd38511af/go.mod h1:5S5YwjSXWmnEKjBjG6MtyGtFmljjukDRS8CwHk/CF/I=

internal/controllers/pipelines.go

@@ -11,7 +11,7 @@ func (c *Controller) UpdateListPipelines(ctx *fiber.Ctx) error {
 		return ctx.Status(fiber.StatusUnauthorized).SendString("account id is required")
 	}
 
-	//accountID := "64f2cd7a7047f28fdabf6d9e"
+	//accountID := "654a8909725f47e926f0bebc"
 
 	err := c.service.UpdateListPipelines(ctx.Context(), accountID)
 

internal/controllers/steps.go

@@ -30,7 +30,7 @@ func (c *Controller) UpdateListSteps(ctx *fiber.Ctx) error {
 		return ctx.Status(fiber.StatusUnauthorized).SendString("account id is required")
 	}
 
-	//accountID := "64f2cd7a7047f28fdabf6d9e"
+	//accountID := "654a8909725f47e926f0bebc"
 
 	err := c.service.UpdateListSteps(ctx.Context(), accountID)
 

internal/controllers/tags.go

@@ -29,7 +29,7 @@ func (c *Controller) UpdateListTags(ctx *fiber.Ctx) error {
 		return ctx.Status(fiber.StatusUnauthorized).SendString("account id is required")
 	}
 
-	//accountID := "64f2cd7a7047f28fdabf6d9e"
+	//accountID := "654a8909725f47e926f0bebc"
 
 	err := c.service.UpdateListTags(ctx.Context(), accountID)
 

internal/workers_methods/methods.go

@@ -100,6 +100,13 @@ func (m *Methods) CheckUsers(ctx context.Context, allTokens []model.Token) error
 			m.logger.Error("error getting current account from db", zap.Error(err))
 			return err
 		}
+
+		currentUserUsers, err := m.repo.AmoRepo.GetUserUsersByID(ctx, mainAccount.Amouserid)
+		if err != nil {
+			m.logger.Error("error getting user users by amo user id", zap.Error(err))
+			return err
+		}
+
 		for _, user := range users {
 			if user.ID == mainAccount.AmoID {
 				err := m.repo.AmoRepo.CheckMainUser(ctx, model.User{
@@ -124,6 +131,29 @@ func (m *Methods) CheckUsers(ctx context.Context, allTokens []model.Token) error
 				Amouserid: mainAccount.Amouserid,
 			})
 		}
+
+		var deletedUserIDs []int64
+		for _, currentUserUser := range currentUserUsers {
+			found := false
+			for _, user := range users {
+				if currentUserUser.AmoID == user.ID {
+					found = true
+					break
+				}
+			}
+
+			if !found {
+				deletedUserIDs = append(deletedUserIDs, currentUserUser.ID)
+			}
+		}
+
+		if len(deletedUserIDs) > 0 {
+			err := m.repo.AmoRepo.DeleteUsers(ctx, deletedUserIDs)
+			if err != nil {
+				m.logger.Error("error deleting users in db", zap.Error(err))
+				return err
+			}
+		}
 	}
 
 	err := m.repo.AmoRepo.CheckAndUpdateUsers(ctx, usersForUpdateAndCreate)
@@ -137,13 +167,33 @@ func (m *Methods) CheckUsers(ctx context.Context, allTokens []model.Token) error
 
 func (m *Methods) CheckPipelinesAndSteps(ctx context.Context, tokens []model.Token) error {
 	for _, token := range tokens {
+		user, err := m.repo.AmoRepo.GetCurrentAccount(ctx, token.AccountID)
+		if err != nil {
+			m.logger.Error("error getting amoUserInfo by account quiz id", zap.Error(err))
+			return err
+		}
+		currentUserPipelines, err := m.repo.AmoRepo.GetUserPipelinesByID(ctx, user.AmoID)
+		if err != nil {
+			m.logger.Error("error getting user pipelines by amo id", zap.Error(err))
+			return err
+		}
+
+		currentUserSteps, err := m.repo.AmoRepo.GetUserStepsByID(ctx, user.AmoID)
+		if err != nil {
+			m.logger.Error("error getting user steps by amo id", zap.Error(err))
+			return err
+		}
+
+		var receivedSteps []model.Step
+
 		pipelines, err := m.amoClient.GetListPipelines(token.AccessToken)
 		if err != nil {
 			m.logger.Error("error fetching list pipelines from amo", zap.Error(err))
 			continue
 		}
 		if len(pipelines.Embedded.Pipelines) > 0 {
-			err = m.repo.AmoRepo.CheckPipelines(ctx, tools.ToPipeline(pipelines.Embedded.Pipelines))
+			receivedPipelines := tools.ToPipeline(pipelines.Embedded.Pipelines)
+			err = m.repo.AmoRepo.CheckPipelines(ctx, receivedPipelines)
 			if err != nil {
 				m.logger.Error("error update list pipelines in db:", zap.Error(err))
 				return err
@@ -156,12 +206,58 @@ func (m *Methods) CheckPipelinesAndSteps(ctx context.Context, tokens []model.Tok
 					continue
 				}
 
-				err = m.repo.AmoRepo.CheckSteps(ctx, tools.ToStep(steps.Embedded.Statuses))
+				receivedStep := tools.ToStep(steps.Embedded.Statuses)
+				receivedSteps = append(receivedSteps, receivedStep...)
+				err = m.repo.AmoRepo.CheckSteps(ctx, receivedStep)
 				if err != nil {
 					m.logger.Error("error update pipeline steps in db:", zap.Error(err))
 					return err
 				}
 			}
+
+			var deletedPipelineIDs []int64
+			for _, currentUserPipeline := range currentUserPipelines {
+				found := false
+				for _, receivedPipeline := range receivedPipelines {
+					if currentUserPipeline.Amoid == receivedPipeline.Amoid && currentUserPipeline.AccountID == user.AmoID {
+						found = true
+						break
+					}
+				}
+				if !found {
+					deletedPipelineIDs = append(deletedPipelineIDs, currentUserPipeline.ID)
+				}
+			}
+
+			if len(deletedPipelineIDs) > 0 {
+				err := m.repo.AmoRepo.DeletePipelines(ctx, deletedPipelineIDs)
+				if err != nil {
+					m.logger.Error("error deleting pipelines in db", zap.Error(err))
+					return err
+				}
+			}
+
+			var deletedStepIDs []int64
+			for _, currentUserStep := range currentUserSteps {
+				found := false
+				for _, receivedStep := range receivedSteps {
+					if currentUserStep.Amoid == receivedStep.Amoid && currentUserStep.Accountid == user.AmoID && currentUserStep.Pipelineid == receivedStep.Pipelineid {
+						found = true
+						break
+					}
+				}
+				if !found {
+					deletedStepIDs = append(deletedStepIDs, currentUserStep.ID)
+				}
+			}
+
+			if len(deletedStepIDs) > 0 {
+				err := m.repo.AmoRepo.DeleteSteps(ctx, deletedStepIDs)
+				if err != nil {
+					m.logger.Error("error deleting steps in db", zap.Error(err))
+					return err
+				}
+			}
 		}
 	}
 
@@ -170,6 +266,18 @@ func (m *Methods) CheckPipelinesAndSteps(ctx context.Context, tokens []model.Tok
 
 func (m *Methods) CheckTags(ctx context.Context, tokens []model.Token) error {
 	for _, token := range tokens {
+		user, err := m.repo.AmoRepo.GetCurrentAccount(ctx, token.AccountID)
+		if err != nil {
+			m.logger.Error("error getting amoUserInfo by account quiz id", zap.Error(err))
+			return err
+		}
+
+		currentUserTags, err := m.repo.AmoRepo.GetUserTagsByID(ctx, user.AmoID)
+		if err != nil {
+			m.logger.Error("error getting user tags by amo id", zap.Error(err))
+			return err
+		}
+
 		var wg sync.WaitGroup
 		wg.Add(4)
 
@@ -206,6 +314,39 @@ func (m *Methods) CheckTags(ctx context.Context, tokens []model.Token) error {
 
 		wg.Wait()
 
+		var deletedTagIDs []int64
+		for _, currentUserTag := range currentUserTags {
+			found := false
+			for _, entityType := range entityTypes {
+				if tags, ok := tagsMap.Load(entityType); ok {
+					if len(tags.([]models.Tag)) > 0 {
+						receivedTags := tools.ToTag(tags.([]models.Tag), entityType)
+						for _, tag := range receivedTags {
+							if currentUserTag.Amoid == tag.Amoid && currentUserTag.Accountid == user.AmoID && currentUserTag.Entity == entityType {
+								found = true
+								break
+							}
+						}
+					}
+				}
+				if found {
+					break
+				}
+			}
+
+			if !found {
+				deletedTagIDs = append(deletedTagIDs, currentUserTag.ID)
+			}
+		}
+
+		if len(deletedTagIDs) > 0 {
+			err = m.repo.AmoRepo.DeleteTags(ctx, deletedTagIDs)
+			if err != nil {
+				m.logger.Error("error deleting tags in db", zap.Error(err))
+				return err
+			}
+		}
+
 		for _, entityType := range entityTypes {
 			if tags, ok := tagsMap.Load(entityType); ok {
 				if len(tags.([]models.Tag)) > 0 {
@@ -235,6 +376,18 @@ func (m *Methods) CheckTags(ctx context.Context, tokens []model.Token) error {
 
 func (m *Methods) CheckFields(ctx context.Context, tokens []model.Token) error {
 	for _, token := range tokens {
+		user, err := m.repo.AmoRepo.GetCurrentAccount(ctx, token.AccountID)
+		if err != nil {
+			m.logger.Error("error getting amoUserInfo by account quiz id", zap.Error(err))
+			return err
+		}
+
+		currentUserFields, err := m.repo.AmoRepo.GetUserFieldsByID(ctx, user.AmoID)
+		if err != nil {
+			m.logger.Error("error getting user fields by amo id", zap.Error(err))
+			return err
+		}
+
 		var wg sync.WaitGroup
 		wg.Add(4)
 
@@ -271,6 +424,39 @@ func (m *Methods) CheckFields(ctx context.Context, tokens []model.Token) error {
 
 		wg.Wait()
 
+		var deletedFieldIDs []int64
+		for _, currentUserField := range currentUserFields {
+			found := false
+			for _, entityType := range entityTypes {
+				if fields, ok := fieldsMap.Load(entityType); ok {
+					if len(fields.([]models.CustomField)) > 0 {
+						receivedFields := tools.ToField(fields.([]models.CustomField), entityType)
+						for _, field := range receivedFields {
+							if currentUserField.Amoid == field.Amoid && currentUserField.Accountid == user.AmoID && currentUserField.Entity == entityType {
+								found = true
+								break
+							}
+						}
+					}
+				}
+				if found {
+					break
+				}
+			}
+
+			if !found {
+				deletedFieldIDs = append(deletedFieldIDs, currentUserField.ID)
+			}
+		}
+
+		if len(deletedFieldIDs) > 0 {
+			err = m.repo.AmoRepo.DeleteFields(ctx, deletedFieldIDs)
+			if err != nil {
+				m.logger.Error("error deleting fields in db", zap.Error(err))
+				return err
+			}
+		}
+
 		for _, entityType := range entityTypes {
 			if fields, ok := fieldsMap.Load(entityType); ok {
 				if len(fields.([]models.CustomField)) > 0 {