worker/answerwc/to_client.go

package answerwc

import (
	"context"
	_ "embed"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/go-redis/redis/v8"
	"penahub.gitlab.yandexcloud.net/backend/quiz/common.git/dal"
	"penahub.gitlab.yandexcloud.net/backend/quiz/common.git/model"
	"penahub.gitlab.yandexcloud.net/backend/quiz/common.git/pj_errors"
	"penahub.gitlab.yandexcloud.net/backend/quiz/worker.git/senders"
	"penahub.gitlab.yandexcloud.net/backend/quiz/worker.git/wctools"
	"penahub.gitlab.yandexcloud.net/pena-services/customer/pkg/customer_clients"

	"time"
)

type DepsSendToClient struct {
	Redis           *redis.Client
	Dal             *dal.DAL
	LeadSenders     []senders.LeadSender
	CustomerService *customer_clients.CustomersClient
}

type SendToClient struct {
	redis           *redis.Client
	dal             *dal.DAL
	leadSenders     []senders.LeadSender
	customerService *customer_clients.CustomersClient
	errChan         chan<- error
}

type PendingTasks struct {
	Count      int64
	QuizConfig model.QuizConfig
}

//go:embed mail/to_client.tmpl
var toClientTemplate string

//go:embed mail/reminder.tmpl
var reminderTemplate string

func NewSendToClient(deps DepsSendToClient, errChan chan<- error) *SendToClient {
	return &SendToClient{
		redis:           deps.Redis,
		dal:             deps.Dal,
		customerService: deps.CustomerService,
		errChan:         errChan,
	}
}

func (w *SendToClient) Start(ctx context.Context) {
	answerTicker := time.NewTicker(30 * time.Second)
	defer answerTicker.Stop()

	for {
		select {
		case <-answerTicker.C:
			w.processPendingAnswer(ctx)

		case <-ctx.Done():
			return

		}
	}
}

func (w *SendToClient) processPendingAnswer(ctx context.Context) {
	pendingAnswers, err := w.redis.Keys(ctx, "answer:*").Result()
	if err != nil {
		fmt.Println("Error getting keys from redis")
		w.errChan <- err
		return
	}

	fmt.Println("ANS")

	for _, key := range pendingAnswers {
		func() {
			fmt.Println("ANS1", key)
			answerJSON, err := w.redis.GetDel(ctx, key).Result()
			if err == redis.Nil {
				return
			} else if err != nil {
				w.reportError(err, "Error getting and deleting data from redis")
				return
			}

			defer func() {
				if r := recover(); r != nil {
					w.reportError(nil, fmt.Sprintf("recovering from panic or error setting redis value %v", r))
					fmt.Println("ANS1ERRR", r)
					_ = w.redis.Set(ctx, key, answerJSON, 0).Err()
				}
			}()

			var answer model.Answer
			err = json.Unmarshal([]byte(answerJSON), &answer)
			fmt.Println("ANS2", err)
			if err != nil {
				w.reportError(err, "Error unmarshal answer")
				return
			}

			answerContent, err := wctools.ProcessAnswer(answer.Content)
			fmt.Println("ANS3", err)
			if err != nil {
				w.reportError(err, "Error unmarshal answer content")
				return
			}

			allAnswers, err := w.dal.AnswerRepo.GetAllAnswersByQuizID(ctx, answer.Session)
			fmt.Println("ANS4", err)
			if err != nil {
				w.reportError(err, "Error getting all answers by quizID")
				return
			}

			questionsMap, sortedallAnswers, err := w.dal.QuestionRepo.GetMapQuestions(ctx, allAnswers)
			fmt.Println("ANS5", err)
			if err != nil {
				w.reportError(err, "Error getting questionsMap")
				return
			}

			if answer.QuizId == 0 {
				return
			}

			quizConfig, accountId, err := w.dal.QuizRepo.GetQuizConfig(ctx, answer.QuizId)
			fmt.Println("ANS6", err)
			if err != nil {
				w.reportError(err, "Error getting quiz config")
				return
			}

			quiz, err := w.dal.QuizRepo.GetQuizById(ctx, accountId, answer.QuizId)
			fmt.Println("ANS60", err, accountId, answer.QuizId)
			if err != nil {
				w.reportError(err, "Error getting quiz")
				return
			}

			quizConfig.Mailing.Reply = quiz.Name

			if quizConfig.Mailing.Theme == "" {
				quizConfig.Mailing.Theme = quiz.Name
			}

			account, privileges, err := w.dal.AccountRepo.GetAccAndPrivilegeByEmail(ctx, accountId)
			fmt.Println("ANS7", err)
			if err != nil {
				w.reportError(err, "Error getting account and privileges by email")
				return
			}

			result, err := w.processAnswerWithPrivileges(ctx, ProcessAnsWithPriv{
				quiz:          quiz,
				quizConfig:    quizConfig,
				questionsMap:  questionsMap,
				privileges:    privileges,
				account:       account,
				allAnswers:    sortedallAnswers,
				answerContent: answerContent,
				answerTime:    answer.CreatedAt,
			})
			fmt.Println("ANS8", err, result, privileges)
			if err != nil {
				w.reportError(err, "Error process answer with privileges")
				return
			}
			if !result {
				err = w.redis.Set(ctx, fmt.Sprintf("%s:%s", account.ID, key), answerJSON, 0).Err()
				if err != nil {
					w.reportError(err, "Error setting redis value")
					return
				}
			}
		}()
	}
}

type ProcessAnsWithPriv struct {
	quiz          *model.Quiz
	quizConfig    model.QuizConfig
	questionsMap  map[uint64]string
	privileges    []model.ShortPrivilege
	account       model.Account
	allAnswers    []model.ResultAnswer
	answerContent model.ResultContent
	answerTime    time.Time
}

func (w *SendToClient) processAnswerWithPrivileges(ctx context.Context, data ProcessAnsWithPriv) (bool, error) {

	err := w.notificationCustomer(ctx, data.account, data.privileges)
	fmt.Println("ANS81", err)
	if err != nil {
		return false, err
	}

	if wctools.HasUnlimitedPrivilege(data.privileges) {
		err := w.ProcessMessageToClient(ctx, DepsProcessMsgToClient{
			quizConfig:    data.quizConfig,
			questionsMap:  data.questionsMap,
			account:       data.account,
			allAnswers:    data.allAnswers,
			answerContent: data.answerContent,
			answerTime:    data.answerTime,
			quiz:          data.quiz,
		})
		if err != nil {
			return false, err
		}
		return true, nil
	}
	privilege := wctools.HasQuizCntPrivilege(data.privileges)
	if privilege != nil {
		err := w.ProcessMessageToClient(ctx, DepsProcessMsgToClient{
			quizConfig:    data.quizConfig,
			questionsMap:  data.questionsMap,
			account:       data.account,
			allAnswers:    data.allAnswers,
			answerContent: data.answerContent,
			answerTime:    data.answerTime,
			quiz:          data.quiz,
		})
		fmt.Println("PMC", err)
		if err != nil {
			return true, err
		}

		privilege.Amount--
		err = w.dal.AccountRepo.UpdatePrivilegeAmount(ctx, privilege.ID, privilege.Amount)
		if err != nil {
			return false, err
		}
		return true, nil
	} else {
		w.checkAndSendTaskReminders(ctx, sendTaskRemindersDeps{
			quiz:    data.quiz,
			account: data.account,
			theme:   data.quiz.Name,
			config: model.QuizConfig{
				Mailing: model.ResultInfo{
					When:     "email",
					Theme:    fmt.Sprintf("не удалось отправить заявку по опросу\"%s\"", data.quiz.Name),
					Reply:    "noreply@pena.digital",
					ReplName: "Reminder",
				},
			},
		})
		return false, nil
	}
}

func (w *SendToClient) recordPendingTasks(ctx context.Context, Email string, quizConfig model.QuizConfig) error {
	key := fmt.Sprintf("pending_tasks:%s", Email)

	var pendingTasks PendingTasks
	val, err := w.redis.HGet(ctx, key, "data").Result()
	if err == nil {
		err := json.Unmarshal([]byte(val), &pendingTasks)
		if err != nil {
			return err
		}
		pendingTasks.Count++
	} else {
		pendingTasks = PendingTasks{
			Count:      1,
			QuizConfig: quizConfig,
		}
	}

	pendingTasksJSON, err := json.Marshal(pendingTasks)
	if err != nil {
		return err
	}

	err = w.redis.HSet(ctx, key, "data", string(pendingTasksJSON)).Err()
	if err != nil {
		return err
	}

	return nil
}

type sendTaskRemindersDeps struct {
	account model.Account
	theme   string
	config  model.QuizConfig
	quiz    *model.Quiz
}

func (w *SendToClient) checkAndSendTaskReminders(ctx context.Context, data sendTaskRemindersDeps) {
	err := w.processReminderToClient(ctx, data.account, data.config, data.quiz)
	fmt.Println("PMC1", err)
	if err != nil {
		w.reportError(err, "Error sending tasks reminder email")
	}
}

func (w *SendToClient) notificationCustomer(ctx context.Context, account model.Account, privileges []model.ShortPrivilege) error {
	for _, privilege := range privileges {
		fmt.Println("NOTIFIC", privilege.PrivilegeID, privilege.Amount, !wctools.IsPrivilegeExpired(privilege))
		if privilege.PrivilegeID == "quizUnlimTime" && !wctools.IsPrivilegeExpired(privilege) {
			historyData := customer_clients.InsertHistoryDeps{
				UserID:  account.UserID,
				Comment: fmt.Sprintf("%s privilege has expired, it was created at %d", privilege.PrivilegeID, privilege.CreatedAt.Unix()),
				Key:     "privilege_expired",
			}

			err := w.customerService.InsertHistory(ctx, historyData)
			if err != nil {
				return err
			}

		}

		if privilege.PrivilegeID == "quizCnt" && privilege.Amount == 0 {
			historyData := customer_clients.InsertHistoryDeps{
				UserID:  account.UserID,
				Comment: fmt.Sprintf("%s privilege has expired, it was created at %d", privilege.PrivilegeID, privilege.CreatedAt.Unix()),
				Key:     "privilege_expired",
			}

			err := w.customerService.InsertHistory(ctx, historyData)
			if err != nil {
				return err
			}
		}
	}

	return nil
}

type DepsProcessMsgToClient struct {
	quizConfig    model.QuizConfig
	questionsMap  map[uint64]string
	account       model.Account
	allAnswers    []model.ResultAnswer
	answerContent model.ResultContent
	answerTime    time.Time
	quiz          *model.Quiz
}

// сделал экспортируемым для теста
func (w *SendToClient) ProcessMessageToClient(ctx context.Context, constructData DepsProcessMsgToClient) error {
	leadTargetForAll, err := w.dal.AccountRepo.GetLeadTarget(ctx, constructData.quiz.AccountId, 0)
	if err != nil && !errors.Is(err, pj_errors.ErrNotFound) {
		return err
	}
	leadTargetForQuiz, err := w.dal.AccountRepo.GetLeadTarget(ctx, constructData.quiz.AccountId, int32(constructData.quiz.Id))
	if err != nil && !errors.Is(err, pj_errors.ErrNotFound) {
		return err
	}
	if len(leadTargetForQuiz) > 0 {
		leadTargetForAll = append(leadTargetForAll, leadTargetForQuiz...)
	}

	theme := constructData.quizConfig.Mailing.Theme
	constructData.quizConfig.Mailing.Theme = constructData.quizConfig.Mailing.Reply

	data := senders.TemplateData{
		QuizConfig:    constructData.quizConfig.Mailing,
		AnswerContent: constructData.answerContent,
		AllAnswers:    constructData.allAnswers,
		QuestionsMap:  constructData.questionsMap,
	}

	dayOfWeek := wctools.DaysOfWeek[constructData.answerTime.Format("Monday")]
	monthOfYear := wctools.MonthsOfYear[constructData.answerTime.Format("January")]

	formattedTime := fmt.Sprintf("%s, %d %s %d г., %02d:%02d (UTC%s)",
		dayOfWeek,
		constructData.answerTime.Day(),
		monthOfYear,
		constructData.answerTime.Year(),
		constructData.answerTime.Hour(),
		constructData.answerTime.Minute(),
		constructData.answerTime.Format("-07:00"),
	)

	data.AnswerTime = formattedTime

	mapLeadTarget := make(map[string][]senders.LeadData) // ключ имя сендера, модель отправки
	for _, leadTarget := range leadTargetForAll {
		mapLeadTarget[string(leadTarget.Type)] = append(mapLeadTarget[string(leadTarget.Type)], senders.LeadData{
			To:           leadTarget.Target,
			Subject:      theme,
			TemplateData: data,
		})
	}

	for _, sender := range w.leadSenders {
		for _, sendData := range mapLeadTarget[sender.Name()] {
			err := sender.SendLead(sendData)
			if err != nil {
				w.reportError(err, fmt.Sprintf("Error sending lead through %s", sender.Name()))
			}
		}
	}

	return nil
}

func (w *SendToClient) processReminderToClient(ctx context.Context, account model.Account, quizConfig model.QuizConfig, quiz *model.Quiz) error {
	leadTargetForAll, err := w.dal.AccountRepo.GetLeadTarget(ctx, account.ID, 0)
	if err != nil && !errors.Is(err, pj_errors.ErrNotFound) {
		return err
	}

	leadTargetForQuiz, err := w.dal.AccountRepo.GetLeadTarget(ctx, account.ID, int32(quiz.Id))
	if err != nil && !errors.Is(err, pj_errors.ErrNotFound) {
		return err
	}
	if len(leadTargetForQuiz) > 0 {
		leadTargetForAll = append(leadTargetForAll, leadTargetForQuiz...)
	}

	mapLeadTarget := make(map[string][]senders.LeadData)
	for _, leadTarget := range leadTargetForAll {
		data := senders.TemplateData{
			QuizConfig: model.ResultInfo{
				When:     quizConfig.Mailing.When,
				Theme:    quizConfig.Mailing.Theme,
				Reply:    leadTarget.Target,
				ReplName: quizConfig.Mailing.ReplName,
			},
			AnswerContent: model.ResultContent{},
			AllAnswers:    []model.ResultAnswer{},
			QuestionsMap:  nil,
		}

		fmt.Println("PRTC", data, leadTarget.Target, quizConfig)

		mapLeadTarget[string(leadTarget.Type)] = append(mapLeadTarget[string(leadTarget.Type)], senders.LeadData{
			To:           leadTarget.Target,
			Subject:      quizConfig.Mailing.Theme,
			Template:     reminderTemplate,
			TemplateData: data,
		})
	}

	for _, sender := range w.leadSenders {
		for _, sendData := range mapLeadTarget[sender.Name()] {
			err := sender.SendLead(sendData)
			if err != nil {
				w.reportError(err, fmt.Sprintf("Error sending lead through %s", sender.Name()))
			}
		}
	}

	return nil
}

func (w *SendToClient) reportError(err error, message string) {
	if err != nil {
		fmt.Println(message + ": " + err.Error())
		w.errChan <- err
	}
}