// Code generated by candi v1.4.0.

package workerhandler

import (
	"context"

	"monorepo/services/user-service/internal/modules/member/usecase"
	"monorepo/services/user-service/pkg/shared"

	taskqueueworker "pkg.agungdp.dev/candi/codebase/app/task_queue_worker"
	"pkg.agungdp.dev/candi/codebase/factory/types"
	"pkg.agungdp.dev/candi/codebase/interfaces"
	"pkg.agungdp.dev/candi/tracer"
)

// KafkaHandler struct
type KafkaHandler struct {
	uc        usecase.MemberUsecase
	validator interfaces.Validator
}

// NewKafkaHandler constructor
func NewKafkaHandler(uc usecase.MemberUsecase, validator interfaces.Validator) *KafkaHandler {
	return &KafkaHandler{
		uc:        uc,
		validator: validator,
	}
}

// MountHandlers mount handler group
func (h *KafkaHandler) MountHandlers(group *types.WorkerHandlerGroup) {
	group.Add(shared.GetEnv().KafkaTopicAutoGenerateMember, h.handleMember) // handling topic "KafkaTopicAutoGenerateMember"
}

// ProcessMessage from kafka consumer
func (h *KafkaHandler) handleMember(ctx context.Context, message []byte) error {
	trace := tracer.StartTrace(ctx, "MemberDeliveryKafka:HandleMember")
	defer trace.Finish()
	ctx = trace.Context()

	return taskqueueworker.AddJob(shared.GetEnv().TaskAddMember, 10, message)
}
