feat(api-gql): add ws router for future scaling (#741)

* feat(api-gql): add ws router for future scaling

* kruto

* kruto

* kruto

* kruto

apps/api-gql/cmd/main.go

@@ -11,6 +11,7 @@ import (
 	pubclicroutes "github.com/twirapp/twir/apps/api-gql/internal/routes/public"
 	"github.com/twirapp/twir/apps/api-gql/internal/routes/webhooks"
 	"github.com/twirapp/twir/apps/api-gql/internal/sessions"
+	"github.com/twirapp/twir/apps/api-gql/internal/wsrouter"
 	"github.com/twirapp/twir/libs/baseapp"
 	buscore "github.com/twirapp/twir/libs/bus-core"
 	commandscache "github.com/twirapp/twir/libs/cache/commands"
@@ -38,6 +39,10 @@ func main() {
 			commandscache.New,
 			keywordscacher.New,
 			buscore.NewNatsBusFx("api-gql"),
+			fx.Annotate(
+				wsrouter.NewNatsSubscription,
+				fx.As(new(wsrouter.WsRouter)),
+			),
 			subscriptions_store.New,
 			resolvers.New,
 			directives.New,

apps/api-gql/internal/gql/resolvers/resolver.go

@@ -13,6 +13,7 @@ import (
 	"github.com/satont/twir/libs/twitch"
 	subscriptions_store "github.com/twirapp/twir/apps/api-gql/internal/gql/subscriptions-store"
 	"github.com/twirapp/twir/apps/api-gql/internal/sessions"
+	"github.com/twirapp/twir/apps/api-gql/internal/wsrouter"
 	bus_core "github.com/twirapp/twir/libs/bus-core"
 	generic_cacher "github.com/twirapp/twir/libs/cache/generic-cacher"
 	twitchcahe "github.com/twirapp/twir/libs/cache/twitch"
@@ -39,6 +40,7 @@ type Resolver struct {
 	redis                *redis.Client
 	keywordsCacher       *generic_cacher.GenericCacher[[]model.ChannelsKeywords]
 	tokensClient         tokens.TokensClient
+	wsRouter             wsrouter.WsRouter
 }
 
 type Opts struct {
@@ -56,6 +58,7 @@ type Opts struct {
 	Logger               logger.Logger
 	Redis                *redis.Client
 	KeywordsCacher       *generic_cacher.GenericCacher[[]model.ChannelsKeywords]
+	WsRouter             wsrouter.WsRouter
 }
 
 func New(opts Opts) (*Resolver, error) {
@@ -78,6 +81,7 @@ func New(opts Opts) (*Resolver, error) {
 		cachedCommandsClient: opts.CachedCommandsClient,
 		keywordsCacher:       opts.KeywordsCacher,
 		tokensClient:         opts.TokensGrpc,
+		wsRouter:             opts.WsRouter,
 	}, nil
 }
 

apps/api-gql/internal/gql/resolvers/subscriptions.keys.go

@@ -0,0 +1,5 @@
+package resolvers
+
+const (
+	notificationsSubscriptionKey = "api.newNotifications"
+)

apps/api-gql/internal/wsrouter/nats_wsrouter.go

@@ -0,0 +1,76 @@
+package wsrouter
+
+import (
+	"github.com/goccy/go-json"
+	"github.com/nats-io/nats.go"
+)
+
+func NewNatsSubscription(opts Opts) (*WsRouterNats, error) {
+	nc, err := nats.Connect(opts.Config.NatsUrl)
+	if err != nil {
+		return nil, err
+	}
+
+	return &WsRouterNats{
+		nc: nc,
+	}, nil
+}
+
+type WsRouterNats struct {
+	nc *nats.Conn
+}
+
+var _ WsRouter = &WsRouterNats{}
+
+type WsRouterNatsSubscription struct {
+	subs      []*nats.Subscription
+	dataChann chan []byte
+}
+
+func (c *WsRouterNatsSubscription) Unsubscribe() error {
+	for _, sub := range c.subs {
+		if err := sub.Unsubscribe(); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (c *WsRouterNatsSubscription) GetChannel() chan []byte {
+	return c.dataChann
+}
+
+func (c *WsRouterNats) Subscribe(keys []string) (WsRouterSubscription, error) {
+	ch := make(chan []byte)
+	subs := make([]*nats.Subscription, 0, len(keys))
+
+	for _, key := range keys {
+		sub, err := c.nc.Subscribe(
+			key,
+			func(msg *nats.Msg) {
+				ch <- msg.Data
+			},
+		)
+
+		if err != nil {
+			return nil, err
+		}
+
+		subs = append(subs, sub)
+	}
+
+	return &WsRouterNatsSubscription{
+		subs:      subs,
+		dataChann: ch,
+	}, nil
+}
+
+func (c *WsRouterNats) Publish(key string, data any) error {
+	dataBytes, err := json.Marshal(data)
+	if err != nil {
+		return err
+	}
+
+	return c.nc.Publish(key, dataBytes)
+}

apps/api-gql/internal/wsrouter/wsrouter.go

@@ -0,0 +1,22 @@
+package wsrouter
+
+import (
+	config "github.com/satont/twir/libs/config"
+	"go.uber.org/fx"
+)
+
+type Opts struct {
+	fx.In
+
+	Config config.Config
+}
+
+type WsRouter interface {
+	Subscribe(keys []string) (WsRouterSubscription, error)
+	Publish(key string, data any) error
+}
+
+type WsRouterSubscription interface {
+	GetChannel() chan []byte
+	Unsubscribe() error
+}

frontend/dashboard/src/features/admin-panel/manage-notifications/components/notifications-form.vue

@@ -79,15 +79,15 @@ const { textareaRef, applyModifier } = useTextarea()
 					variant="secondary"
 					@click="notificationsForm.onReset"
 				>
-					<template v-if="notificationsForm.editableMessageId">
+					<template v-if="notificationsForm.editableMessageId.value">
 						{{ t('sharedButtons.cancel') }}
 					</template>
 					<template v-else>
 						{{ t('sharedButtons.reset') }}
 					</template>
 				</Button>
 				<Button type="submit">
-					<template v-if="notificationsForm.editableMessageId">
+					<template v-if="notificationsForm.editableMessageId.value">
 						{{ t('sharedButtons.edit') }}
 					</template>
 					<template v-else>