notifier/internal/workers/consumer.go

package workers

import (
	"context"
	"encoding/json"
	"github.com/twmb/franz-go/pkg/kgo"
	"go.uber.org/zap"
	"mailnotifier/internal/models"
	"mailnotifier/internal/repository"
	"time"
)

type Consumer struct {
	repo        *repository.Repository
	kafkaClient *kgo.Client
	logger      *zap.Logger
}

type ConsumerDeps struct {
	Repo        *repository.Repository
	KafkaClient *kgo.Client
	Logger      *zap.Logger
}

func NewConsumerWC(deps ConsumerDeps) *Consumer {
	return &Consumer{
		repo:        deps.Repo,
		kafkaClient: deps.KafkaClient,
		logger:      deps.Logger,
	}
}

func (c *Consumer) Start(ctx context.Context) {
	ticker := time.NewTicker(30 * time.Minute)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			c.consumeMessages(ctx)

		case <-ctx.Done():
			return
		}
	}

}

// тут только консюмим по тикеру и инзертим в монгу
func (c *Consumer) consumeMessages(ctx context.Context) {
	fetches := c.kafkaClient.PollFetches(ctx)
	iter := fetches.RecordIter()

	for !iter.Done() {
		record := iter.Next()
		var message models.Message

		err := json.Unmarshal(record.Value, &message)
		if err != nil {
			c.logger.Error("error unmarshal kafka message:", zap.Error(err))
		}

		err = c.repo.Insert(ctx, message)
		if err != nil {
			c.logger.Error("error insert kafka data to mongo:", zap.Error(err))
		}
	}
}