feat(eventsub): add vip add and remove handlers

Author: Satont

apps/api-gql/internal/delivery/gql/mappers/admin-actions.go

@@ -1,17 +1,17 @@
 package mappers
 
 import (
+	model "github.com/satont/twir/libs/gomodels"
 	"github.com/twirapp/twir/apps/api-gql/internal/delivery/gql/gqlmodel"
-	"github.com/twirapp/twir/apps/api-gql/internal/entity"
 )
 
-var eventSubConditionTypeGqlToEntity = map[gqlmodel.EventsubSubscribeConditionInput]entity.EventsubSubscribeCondition{
-	gqlmodel.EventsubSubscribeConditionInputChannel:                entity.EventsubSubscribeConditionChannel,
-	gqlmodel.EventsubSubscribeConditionInputUser:                   entity.EventsubSubscribeConditionUser,
-	gqlmodel.EventsubSubscribeConditionInputChannelWithModeratorID: entity.EventsubSubscribeConditionChannelWithModeratorID,
-	gqlmodel.EventsubSubscribeConditionInputChannelWithBotID:       entity.EventsubSubscribeConditionChannelWithBotID,
+var eventSubConditionTypeGqlToEntity = map[gqlmodel.EventsubSubscribeConditionInput]model.EventsubConditionType{
+	gqlmodel.EventsubSubscribeConditionInputChannel:                model.EventsubConditionTypeBroadcasterUserID,
+	gqlmodel.EventsubSubscribeConditionInputUser:                   model.EventsubConditionTypeUserID,
+	gqlmodel.EventsubSubscribeConditionInputChannelWithModeratorID: model.EventsubConditionTypeBroadcasterWithModeratorID,
+	gqlmodel.EventsubSubscribeConditionInputChannelWithBotID:       model.EventsubConditionTypeBroadcasterWithUserID,
 }
 
-func ConditionTypeGqlToEntity(conditionType gqlmodel.EventsubSubscribeConditionInput) entity.EventsubSubscribeCondition {
+func ConditionTypeGqlToEntity(conditionType gqlmodel.EventsubSubscribeConditionInput) model.EventsubConditionType {
 	return eventSubConditionTypeGqlToEntity[conditionType]
 }

apps/api-gql/internal/services/admin-actions/admin-actions.go

@@ -7,7 +7,6 @@ import (
 	"github.com/redis/go-redis/v9"
 	"github.com/samber/lo"
 	model "github.com/satont/twir/libs/gomodels"
-	"github.com/twirapp/twir/apps/api-gql/internal/entity"
 	buscore "github.com/twirapp/twir/libs/bus-core"
 	"github.com/twirapp/twir/libs/bus-core/eventsub"
 	"github.com/twirapp/twir/libs/bus-core/timers"
@@ -68,9 +67,9 @@ func (c *Service) DropAllAuthSessions(ctx context.Context) error {
 }
 
 type EventSubSubscribeInput struct {
-	Type      string                            `json:"type"`
-	Version   string                            `json:"version"`
-	Condition entity.EventsubSubscribeCondition `json:"condition"`
+	Type      string                      `json:"type"`
+	Version   string                      `json:"version"`
+	Condition model.EventsubConditionType `json:"condition"`
 }
 
 func (c *Service) EventSubSubscribe(ctx context.Context, input EventSubSubscribeInput) error {

apps/eventsub/go.mod

@@ -32,7 +32,7 @@ require (
 	github.com/twirapp/twir/libs/grpc v0.0.0-20241105005614-f68dbf11ade1
 	github.com/twirapp/twir/libs/integrations v0.0.0-20241105005614-f68dbf11ade1
 	github.com/twirapp/twir/libs/uptrace v0.0.0-20241105005614-f68dbf11ade1
-	github.com/twirapp/twitch-eventsub-framework v1.3.7
+	github.com/twirapp/twitch-eventsub-framework v1.3.8
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.58.0
 	go.opentelemetry.io/otel v1.33.0
 	go.opentelemetry.io/otel/trace v1.33.0

apps/eventsub/go.sum

@@ -152,17 +152,22 @@ github.com/twirapp/twir/libs/cache v0.0.0-20241105005614-f68dbf11ade1 h1:LfNuHsD
 github.com/twirapp/twir/libs/cache v0.0.0-20241105005614-f68dbf11ade1/go.mod h1:h4TtQax+U9vXDy815YWOnggdD82xK2tEln2/YfRYVbI=
 github.com/twirapp/twitch-eventsub-framework v1.3.7 h1:HwpolVPgZRVCJCD+AzlLol2nRZId+lLLR41qXilDYKY=
 github.com/twirapp/twitch-eventsub-framework v1.3.7/go.mod h1:G67t1kwDexqC38efhh97md/HcnEZFqPFs+B1HeZkBtc=
+github.com/twirapp/twitch-eventsub-framework v1.3.8 h1:ODAFrPYVxLq4R/bFBhNF8laRnRkBoqNRH1VxXlOQFK4=
+github.com/twirapp/twitch-eventsub-framework v1.3.8/go.mod h1:G67t1kwDexqC38efhh97md/HcnEZFqPFs+B1HeZkBtc=
 github.com/uptrace/uptrace-go v1.31.0 h1:i+AsiXVZ5NHHYWvlmJyTqg5l9sLqYN/KSZeLWr3kAM8=
 github.com/uptrace/uptrace-go v1.31.0/go.mod h1:zX5JqvEoLpFbbtGN+G1/vf276rfoHiTFrh7yt1XdVCY=
 github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU=
 github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E=
 go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA=
+go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A=
 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.56.0 h1:yMkBS9yViCc7U7yeLzJPM2XizlfdVvBRSmsQDWu6qc0=
 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.56.0/go.mod h1:n8MR6/liuGB5EmTETUBeU5ZgqMOlqKRxUaqPQBOANZ8=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.58.0 h1:yd02MEjBdJkG3uabWP9apV+OuWRIXGDuJEUJbOHmCFU=
+go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.58.0/go.mod h1:umTcuxiv1n/s/S6/c2AT/g2CQ7u5C59sHDNmfSwgz7Q=
 go.opentelemetry.io/contrib/instrumentation/runtime v0.56.0 h1:s7wHG+t8bEoH7ibWk1nk682h7EoWLJ5/8j+TSO3bX/o=
 go.opentelemetry.io/contrib/instrumentation/runtime v0.56.0/go.mod h1:Q8Hsv3d9DwryfIl+ebj4mY81IYVRSPy4QfxroVZwqLo=
 go.opentelemetry.io/otel v1.33.0 h1:/FerN9bax5LoK51X/sI0SVYrjSE0/yUL7DpxW4K3FWw=
+go.opentelemetry.io/otel v1.33.0/go.mod h1:SUUkR6csvUQl+yjReHu5uM3EtVV7MBm5FHKRlNx4I8I=
 go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp v0.7.0 h1:mMOmtYie9Fx6TSVzw4W+NTpvoaS1JWWga37oI1a/4qQ=
 go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp v0.7.0/go.mod h1:yy7nDsMMBUkD+jeekJ36ur5f3jJIrmCwUrY67VFhNpA=
 go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.31.0 h1:ZsXq73BERAiNuuFXYqP4MR5hBrjXfMGSO+Cx7qoOZiM=
@@ -176,13 +181,15 @@ go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.31.0/go.mod h1:fcwWuDua
 go.opentelemetry.io/otel/log v0.7.0 h1:d1abJc0b1QQZADKvfe9JqqrfmPYQCz2tUSO+0XZmuV4=
 go.opentelemetry.io/otel/log v0.7.0/go.mod h1:2jf2z7uVfnzDNknKTO9G+ahcOAyWcp1fJmk/wJjULRo=
 go.opentelemetry.io/otel/metric v1.33.0 h1:r+JOocAyeRVXD8lZpjdQjzMadVZp2M4WmQ+5WtEnklQ=
+go.opentelemetry.io/otel/metric v1.33.0/go.mod h1:L9+Fyctbp6HFTddIxClbQkjtubW6O9QS3Ann/M82u6M=
 go.opentelemetry.io/otel/sdk v1.31.0 h1:xLY3abVHYZ5HSfOg3l2E5LUj2Cwva5Y7yGxnSW9H5Gk=
 go.opentelemetry.io/otel/sdk v1.31.0/go.mod h1:TfRbMdhvxIIr/B2N2LQW2S5v9m3gOQ/08KsbbO5BPT0=
 go.opentelemetry.io/otel/sdk/log v0.7.0 h1:dXkeI2S0MLc5g0/AwxTZv6EUEjctiH8aG14Am56NTmQ=
 go.opentelemetry.io/otel/sdk/log v0.7.0/go.mod h1:oIRXpW+WD6M8BuGj5rtS0aRu/86cbDV/dAfNaZBIjYM=
 go.opentelemetry.io/otel/sdk/metric v1.31.0 h1:i9hxxLJF/9kkvfHppyLL55aW7iIJz4JjxTeYusH7zMc=
 go.opentelemetry.io/otel/sdk/metric v1.31.0/go.mod h1:CRInTMVvNhUKgSAMbKyTMxqOBC0zgyxzW55lZzX43Y8=
 go.opentelemetry.io/otel/trace v1.33.0 h1:cCJuF7LRjUFso9LPnEAHJDB2pqzp+hbO8eu1qqW2d/s=
+go.opentelemetry.io/otel/trace v1.33.0/go.mod h1:uIcdVUZMpTAmz0tI1z04GoVSezK37CbGV4fr1f2nBck=
 go.opentelemetry.io/proto/otlp v1.3.1 h1:TrMUixzpM0yuc/znrFTP9MMRh8trP93mkCiDVeXrui0=
 go.opentelemetry.io/proto/otlp v1.3.1/go.mod h1:0X1WI4de4ZsLrrJNLAQbFeLCm3T7yBkR0XqQ7niQU+8=
 go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
@@ -204,18 +211,24 @@ golang.ngrok.com/muxado/v2 v2.0.1/go.mod h1:wzxJYX4xiAtmwumzL+QsukVwFRXmPNv86vB8
 golang.ngrok.com/ngrok v1.11.0 h1:lvbBcoOvH+Ek15wgrjvxpCB+PBM7vinU6jQPsrCdOLw=
 golang.ngrok.com/ngrok v1.11.0/go.mod h1:1/gLOyOJm7ygHJlcEbtldFLQwQnQ42z+rucpLE2YsvA=
 golang.org/x/crypto v0.31.0 h1:ihbySMvVjLAeSH1IbfcRTkD/iNscyz8rGzjF/E5hV6U=
+golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk=
 golang.org/x/exp v0.0.0-20241009180824-f66d83c29e7c h1:7dEasQXItcW1xKJ2+gg5VOiBnqWrJc+rq0DPKyvvdbY=
 golang.org/x/exp v0.0.0-20241009180824-f66d83c29e7c/go.mod h1:NQtJDoLvd6faHhE7m4T/1IY708gDefGGjR/iUW8yQQ8=
 golang.org/x/mod v0.21.0 h1:vvrHzRwRfVKSiLrG+d4FMl/Qi4ukBCE6kZlTUkDYRT0=
 golang.org/x/mod v0.21.0/go.mod h1:6SkKJ3Xj0I0BrPOZoBy3bdMptDDU9oJrpohJ3eWZ1fY=
 golang.org/x/net v0.33.0 h1:74SYHlV8BIgHIFC/LrYkOGIwL19eTYXQ5wc6TBuO36I=
+golang.org/x/net v0.33.0/go.mod h1:HXLR5J+9DxmrqMwG9qjGCxZ+zKXxBru04zlTvWlWuN4=
 golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ=
+golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA=
+golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/term v0.27.0 h1:WP60Sv1nlK1T6SupCHbXzSaN0b9wUmsPoRS9b61A23Q=
+golang.org/x/term v0.27.0/go.mod h1:iMsnZpn0cago0GOrHO2+Y7u7JPn5AylBrcoWkElMTSM=
 golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
+golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
 golang.org/x/time v0.7.0 h1:ntUhktv3OPE6TgYxXWv9vKvUSJyIFJlyohwbkEwPrKQ=
 golang.org/x/tools v0.26.0 h1:v/60pFQmzmT9ExmjDv2gGIfi3OqfKoEP6I5+umXlbnQ=
 golang.org/x/tools v0.26.0/go.mod h1:TPVVj70c7JJ3WCazhD8OdXcZg/og+b9+tH/KxylGwH0=

apps/eventsub/internal/handler/handler.go

@@ -123,6 +123,8 @@ func New(opts Opts) *Handler {
 	handler.HandleChannelPointsRewardAdd = myHandler.handleChannelPointsRewardAdd
 	handler.HandleChannelPointsRewardUpdate = myHandler.handleChannelPointsRewardUpdate
 	handler.HandleChannelPointsRewardRemove = myHandler.handleChannelPointsRewardRemove
+	handler.HandleChannelVipAdd = myHandler.handleChannelVipAdd
+	handler.HandleChannelVipRemove = myHandler.handleChannelVipRemove
 
 	httpHandler := otelhttp.NewHandler(handler, "")
 

apps/eventsub/internal/handler/vip.go

@@ -0,0 +1,44 @@
+package handler
+
+import (
+	"log/slog"
+
+	model "github.com/satont/twir/libs/gomodels"
+	eventsub_bindings "github.com/twirapp/twitch-eventsub-framework/esb"
+)
+
+func (c *Handler) handleChannelVipAdd(
+	_ *eventsub_bindings.ResponseHeaders,
+	event *eventsub_bindings.ChannelVipAdd,
+) {
+	c.logger.Info(
+		"channel vip add",
+		slog.String("channelId", event.BroadcasterUserId),
+		slog.String("userId", event.UserId),
+		slog.String("userName", event.UserLogin),
+	)
+
+	if err := c.gorm.Model(&model.UsersStats{}).
+		Where(`"userId" = ? and "channelId" = ?`, event.UserId, event.BroadcasterUserId).
+		Update(`"is_vip"`, true).Error; err != nil {
+		c.logger.Error(err.Error(), slog.Any("err", err))
+	}
+}
+
+func (c *Handler) handleChannelVipRemove(
+	_ *eventsub_bindings.ResponseHeaders,
+	event *eventsub_bindings.ChannelVipRemove,
+) {
+	c.logger.Info(
+		"channel vip remove",
+		slog.String("channelId", event.BroadcasterUserId),
+		slog.String("userId", event.UserId),
+		slog.String("userName", event.UserLogin),
+	)
+
+	if err := c.gorm.Model(&model.UsersStats{}).
+		Where(`"userId" = ? and "channelId" = ?`, event.UserId, event.BroadcasterUserId).
+		Update(`"is_vip"`, false).Error; err != nil {
+		c.logger.Error(err.Error(), slog.Any("err", err))
+	}
+}

libs/migrations/migrations/20250309054456_add_eventsub_vip_topics.sql

@@ -0,0 +1,12 @@
+-- +goose Up
+-- +goose StatementBegin
+SELECT 'up SQL query';
+INSERT INTO eventsub_topics (topic, version, condition_type)
+VALUES ('channel.vip.add', '1', 'BROADCASTER_USER_ID'),
+       ('channel.vip.remove', '1', 'BROADCASTER_USER_ID');
+-- +goose StatementEnd
+
+-- +goose Down
+-- +goose StatementBegin
+SELECT 'down SQL query';
+-- +goose StatementEnd