added opportunity re-login in webhook

internal/models/kafkaMess.go

@@ -28,4 +28,5 @@ const (
 	UserCreate      MessageType = "userCreate"
 	AllDataUpdate   MessageType = "allDataUpdate"
 	RuleCheck       MessageType = "ruleCheck"
+	UserReLogin     MessageType = "userReLogin"
 )

internal/service/webhook.go

@@ -3,6 +3,7 @@ package service
 import (
 	"amocrm/internal/models"
 	"context"
+	"database/sql"
 	"go.uber.org/zap"
 )
 
@@ -15,16 +16,38 @@ type ParamsWebhookCreate struct {
 }
 
 func (s *Service) WebhookCreate(ctx context.Context, req ParamsWebhookCreate) error {
+	_, err := s.GetCurrentAccount(ctx, req.AccountID)
+	if err != nil && err != sql.ErrNoRows {
+		s.logger.Error("error checking current account in amo in webhook create")
+		return err
+	}
+
+	if err == sql.ErrNoRows {
+		message := models.KafkaMessage{
+			AccountID:  req.AccountID,
+			AuthCode:   req.Code,
+			RefererURL: req.Referer,
+			Type:       models.UserCreate,
+		}
+
+		err = s.producer.ToKafkaUpdate(ctx, message)
+		if err != nil {
+			s.logger.Error("failed to send message to kafka on service webhook create", zap.Error(err))
+			return err
+		}
+		return nil
+	}
+
 	message := models.KafkaMessage{
 		AccountID:  req.AccountID,
 		AuthCode:   req.Code,
 		RefererURL: req.Referer,
-		Type:       models.UserCreate,
+		Type:       models.UserReLogin,
 	}
 
-	err := s.producer.ToKafkaUpdate(ctx, message)
+	err = s.producer.ToKafkaUpdate(ctx, message)
 	if err != nil {
-		s.logger.Error("failed to send message to kafka on service webhook create", zap.Error(err))
+		s.logger.Error("failed to send message to kafka on service webhook create, user re-login", zap.Error(err))
 		return err
 	}
 

internal/workers/queueUpdater/queue_updater.go

@@ -218,6 +218,12 @@ func (wc *QueueUpdater) processMessages(ctx context.Context, message models.Kafk
 				return err
 			}
 		}
+	case models.UserReLogin:
+		err := wc.methods.UserReLogin(ctx, message)
+		if err != nil {
+			wc.logger.Error("error update user information in re-login method", zap.Error(err))
+			return err
+		}
 
 	default:
 		wc.logger.Error("incorrect message type", zap.Any("Type:", message))

internal/workers_methods/methods.go

@@ -712,3 +712,51 @@ func (m *Methods) CheckFieldRule(ctx context.Context, token string, msg models.K
 
 	return nil
 }
+
+func (m *Methods) UserReLogin(ctx context.Context, msg models.KafkaMessage) error {
+	forGetTokens := models.CreateWebHookReq{
+		GrantType: "authorization_code",
+		Code:      msg.AuthCode,
+	}
+
+	tokens, err := m.amoClient.CreateWebHook(&forGetTokens, msg.RefererURL)
+	if err != nil {
+		m.logger.Error("error getting tokens in method user re-login:", zap.Error(err))
+		return err
+	}
+
+	userInfo, err := m.amoClient.GetUserInfo(tokens.AccessToken, msg.RefererURL)
+	if err != nil {
+		m.logger.Error("error getting UserInfo in method user re-login:", zap.Error(err))
+		return err
+	}
+
+	toUpdate := model.AmoAccount{
+		AccountID: msg.AccountID,
+		AmoID:     userInfo.ID,
+		Name:      userInfo.Name,
+		Subdomain: msg.RefererURL,
+		Country:   userInfo.Country,
+		DriveURL:  userInfo.DriveUrl,
+	}
+
+	err = m.repo.AmoRepo.UpdateCurrentAccount(ctx, toUpdate)
+	if err != nil {
+		m.logger.Error("error update account in db in method user re-login", zap.Error(err))
+		return err
+	}
+
+	err = m.repo.AmoRepo.WebhookUpdate(ctx, model.Token{
+		RefreshToken: tokens.RefreshToken,
+		AccessToken:  tokens.AccessToken,
+		AccountID:    msg.AccountID,
+		Expiration:   time.Now().Unix() + tokens.ExpiresIn,
+		CreatedAt:    time.Now().Unix(),
+	})
+	if err != nil {
+		m.logger.Error("error update tokens in db in method user re-login", zap.Error(err))
+		return err
+	}
+
+	return nil
+}