refactor: add single subscribe function to eventsub

apps/api/internal/impl_protected/bot/bot.go

@@ -152,8 +152,8 @@ func (c *Bot) BotJoinPart(ctx context.Context, request *bots.BotJoinPartRequest)
 	}
 
 	if dbChannel.IsEnabled {
-		c.Bus.EventSub.Subscribe.Publish(
-			eventsub.EventsubSubscribeRequest{ChannelID: dashboardId},
+		c.Bus.EventSub.SubscribeToAllEvents.Publish(
+			eventsub.EventsubSubscribeToAllEventsRequest{ChannelID: dashboardId},
 		)
 	}
 

apps/api/internal/impl_unprotected/auth/auth.go

@@ -165,8 +165,8 @@ func (c *Auth) AuthPostCode(ctx context.Context, request *auth.PostCodeRequest)
 	c.SessionManager.Put(ctx, "twitchUser", &twitchUser)
 	c.SessionManager.Put(ctx, "dashboardId", dbUser.ID)
 
-	if err := c.Bus.EventSub.Subscribe.Publish(
-		eventsub.EventsubSubscribeRequest{
+	if err := c.Bus.EventSub.SubscribeToAllEvents.Publish(
+		eventsub.EventsubSubscribeToAllEventsRequest{
 			ChannelID: dbUser.ID,
 		},
 	); err != nil {

apps/eventsub/internal/bus-listener/bus-listener.go

@@ -40,9 +40,24 @@ func New(opts Opts) (*BusListener, error) {
 	opts.Lc.Append(
 		fx.Hook{
 			OnStart: func(ctx context.Context) error {
-				return impl.bus.EventSub.Subscribe.SubscribeGroup("eventsub", impl.subscribeToEvents)
+				if err := impl.bus.EventSub.SubscribeToAllEvents.SubscribeGroup(
+					"eventsub",
+					impl.subscribeToAllEvents,
+				); err != nil {
+					return err
+				}
+
+				if err := impl.bus.EventSub.Subscribe.SubscribeGroup(
+					"eventsub",
+					impl.subscribe,
+				); err != nil {
+					return err
+				}
+
+				return nil
 			},
 			OnStop: func(ctx context.Context) error {
+				impl.bus.EventSub.SubscribeToAllEvents.Unsubscribe()
 				impl.bus.EventSub.Subscribe.Unsubscribe()
 				return nil
 			},
@@ -52,9 +67,9 @@ func New(opts Opts) (*BusListener, error) {
 	return impl, nil
 }
 
-func (c *BusListener) subscribeToEvents(
+func (c *BusListener) subscribeToAllEvents(
 	ctx context.Context,
-	msg eventsub.EventsubSubscribeRequest,
+	msg eventsub.EventsubSubscribeToAllEventsRequest,
 ) struct{} {
 	channel := model.Channels{}
 	err := c.gorm.
@@ -85,3 +100,20 @@ func (c *BusListener) subscribeToEvents(
 
 	return struct{}{}
 }
+
+func (c *BusListener) subscribe(
+	ctx context.Context,
+	msg eventsub.EventsubSubscribeRequest,
+) struct{} {
+	if err := c.eventSubClient.SubscribeToEvent(
+		ctx,
+		msg.ConditionType,
+		msg.Topic,
+		msg.Version,
+		msg.ChannelID,
+	); err != nil {
+		c.logger.Error("error subscribing to event", err)
+	}
+
+	return struct{}{}
+}

apps/eventsub/internal/manager/condition.go

@@ -4,7 +4,7 @@ import (
 	model "github.com/satont/twir/libs/gomodels"
 )
 
-func getTypeCondition(
+func GetTypeCondition(
 	t model.EventsubConditionType,
 	topic,
 	channelID,

apps/eventsub/internal/manager/manager.go

@@ -3,6 +3,7 @@ package manager
 import (
 	"context"
 	"errors"
+	"fmt"
 	"log/slog"
 	"sync"
 
@@ -146,7 +147,7 @@ func (c *Manager) SubscribeToNeededEvents(
 		topic := topic
 		go func() {
 			defer wg.Done()
-			condition := getTypeCondition(topic.ConditionType, topic.Topic, broadcasterId, botId)
+			condition := GetTypeCondition(topic.ConditionType, topic.Topic, broadcasterId, botId)
 			if condition == nil {
 				c.logger.Error(
 					"failed to get condition",
@@ -198,3 +199,47 @@ func (c *Manager) SubscribeToNeededEvents(
 
 	return nil
 }
+
+func (c *Manager) SubscribeToEvent(
+	ctx context.Context,
+	conditionType,
+	topic,
+	version,
+	channelId string,
+) error {
+	channel := model.Channels{}
+	err := c.gorm.
+		WithContext(ctx).
+		Where(
+			`"id" = ?`,
+			channelId,
+		).First(&channel).Error
+	if err != nil {
+		return err
+	}
+
+	convertedCondition := model.FindEventsubCondition(conditionType)
+	if conditionType == "" {
+		return errors.New("condition type not found")
+	}
+
+	condition := GetTypeCondition(convertedCondition, topic, channel.ID, channel.BotID)
+
+	fmt.Println(conditionType, topic, version, channelId)
+	if condition == nil {
+		return errors.New("condition not found")
+	}
+
+	_, err = c.SubscribeWithLimits(
+		ctx,
+		&eventsub_framework.SubRequest{
+			Type:      topic,
+			Condition: condition,
+			Callback:  c.tunnel.GetAddr(),
+			Secret:    c.config.TwitchClientSecret,
+			Version:   version,
+		},
+	)
+
+	return err
+}

libs/bus-core/bus-services.go

@@ -52,7 +52,8 @@ type evalBus struct {
 }
 
 type eventSubBus struct {
-	Subscribe Queue[eventsub.EventsubSubscribeRequest, struct{}]
+	SubscribeToAllEvents Queue[eventsub.EventsubSubscribeToAllEventsRequest, struct{}]
+	Subscribe            Queue[eventsub.EventsubSubscribeRequest, struct{}]
 }
 
 type schedulerBus struct {

libs/bus-core/bus.go

@@ -165,6 +165,12 @@ func NewNatsBus(nc *nats.Conn) *Bus {
 		},
 
 		EventSub: &eventSubBus{
+			SubscribeToAllEvents: NewNatsQueue[eventsub.EventsubSubscribeToAllEventsRequest, struct{}](
+				nc,
+				eventsub.EventsubSubscribeSubject,
+				1*time.Minute,
+				nats.GOB_ENCODER,
+			),
 			Subscribe: NewNatsQueue[eventsub.EventsubSubscribeRequest, struct{}](
 				nc,
 				eventsub.EventsubSubscribeSubject,

libs/bus-core/eventsub/eventsub.go

@@ -4,6 +4,13 @@ const (
 	EventsubSubscribeSubject = "eventsub.subscribe"
 )
 
-type EventsubSubscribeRequest struct {
+type EventsubSubscribeToAllEventsRequest struct {
 	ChannelID string
 }
+
+type EventsubSubscribeRequest struct {
+	ChannelID     string
+	Topic         string
+	ConditionType string
+	Version       string
+}