core/workers/tg_worker.go

package workers

import (
	"context"
	"encoding/json"
	"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/core/clients/telegram"
	"strconv"
	"time"
)

type Deps struct {
	BotID    int64
	Redis    *redis.Client
	Dal      *dal.DAL
	TgClient *telegram.TelegramClient
}

type TgListenerWorker struct {
	botID    int64
	redis    *redis.Client
	dal      *dal.DAL
	tgClient *telegram.TelegramClient
}

func NewTgListenerWC(deps Deps) *TgListenerWorker {
	return &TgListenerWorker{
		botID:    deps.BotID,
		redis:    deps.Redis,
		dal:      deps.Dal,
		tgClient: deps.TgClient,
	}
}

func (wc *TgListenerWorker) Start(ctx context.Context) {
	ticker := time.NewTicker(10 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			wc.processTasks(ctx)
		case <-ctx.Done():
			return
		}
	}
}

func (wc *TgListenerWorker) processTasks(ctx context.Context) {
	var cursor uint64
	for {
		var keys []string
		var err error
		keys, cursor, err = wc.redis.Scan(ctx, cursor, "telegram_task:*", 0).Result()
		if err != nil {
			fmt.Println("Failed scan for telegram tasks:", err)
			break
		}

		for _, key := range keys {
			func() {
				taskBytes, err := wc.redis.GetDel(ctx, key).Result()
				if err == redis.Nil {
					return
				} else if err != nil {
					fmt.Println("Failed getdel telegram task:", err)
					return
				}
				// todo logging into tg with trashlog
				defer func() {
					if r := recover(); r != nil {
						fmt.Println("recovering from panic or error setting redis value:", r)
						_ = wc.redis.Set(ctx, key, taskBytes, 0).Err()
					}
				}()

				var task model.TgRedisTask
				if json.Unmarshal([]byte(taskBytes), &task) != nil {
					fmt.Println("Failed unmarshal telegram task:", err)
					return
				}

				inviteLink, chatID, err := wc.tgClient.CreateChannel(task.Name, wc.botID)
				if err != nil {
					fmt.Println("Failed create tg channel:", err)
					return
				}

				_, err = wc.dal.AccountRepo.PostLeadTarget(ctx, model.LeadTarget{
					AccountID:  task.AccountID,
					Type:       model.LeadTargetTg,
					QuizID:     task.QuizID,
					Target:     strconv.Itoa(int(chatID)),
					InviteLink: inviteLink,
				})
				if err != nil {
					fmt.Println("Failed create lead target in db:", err)
					return
				}
			}()
		}
		if cursor == 0 {
			break
		}
	}
}