fix(eventsub): add categoryId update on channel update

Author: Satont

apps/eventsub/internal/handler/channel_update.go

@@ -52,6 +52,7 @@ func (c *Handler) handleChannelUpdate(
 			map[string]any{
 				"title":    event.Title,
 				"gameName": event.CategoryName,
+				"gameId":   event.CategoryID,
 			},
 		).Error
 	if err != nil {

apps/scheduler/internal/timers/online.go apps/scheduler/internal/timers/online_chatters.go

@@ -0,0 +1,87 @@
+package postgres
+
+import (
+	"context"
+
+	"github.com/Masterminds/squirrel"
+	trmpgx "github.com/avito-tech/go-transaction-manager/drivers/pgxv5/v2"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/twirapp/twir/libs/repositories"
+	"github.com/twirapp/twir/libs/repositories/streams"
+	"github.com/twirapp/twir/libs/repositories/streams/model"
+)
+
+type Opts struct {
+	PgxPool *pgxpool.Pool
+}
+
+func New(opts Opts) *Pgx {
+	return &Pgx{
+		pool:   opts.PgxPool,
+		getter: trmpgx.DefaultCtxGetter,
+	}
+}
+
+func NewFx(pool *pgxpool.Pool) *Pgx {
+	return New(Opts{PgxPool: pool})
+}
+
+var _ streams.Repository = (*Pgx)(nil)
+var sq = squirrel.StatementBuilder.PlaceholderFormat(squirrel.Dollar)
+
+type Pgx struct {
+	pool   *pgxpool.Pool
+	getter *trmpgx.CtxGetter
+}
+
+func (c *Pgx) GetList(ctx context.Context) ([]model.Stream, error) {
+	query := `
+SELECT id, "userId", "userLogin", "userName", "gameId", "gameName", "communityIds", type, title, "viewerCount", "startedAt", "language", "thumbnailUrl", "tagIds", tags, "isMature"
+FROM channels_streams
+`
+
+	conn := c.getter.DefaultTrOrDB(ctx, c.pool)
+	rows, err := conn.Query(ctx, query)
+	if err != nil {
+		return nil, err
+	}
+
+	result, err := pgx.CollectRows(rows, pgx.RowToStructByName[model.Stream])
+	if err != nil {
+		return nil, err
+	}
+
+	return result, nil
+}
+
+func (c *Pgx) Update(ctx context.Context, channelID string, input streams.UpdateInput) error {
+	updateBuilder := sq.Update("channels_streams").
+		Where(squirrel.Eq{`"userId"`: channelID})
+
+	updateBuilder = repositories.SquirrelApplyPatch(
+		updateBuilder, map[string]interface{}{
+			`"gameId"`:       input.GameId,
+			`"gameName"`:     input.GameName,
+			`"communityIds"`: input.CommunityIds,
+			`"type"`:         input.Type,
+			`"title"`:        input.Title,
+			`"viewerCount"`:  input.ViewerCount,
+			`"startedAt"`:    input.StartedAt,
+			`"language"`:     input.Language,
+			`"thumbnailUrl"`: input.ThumbnailUrl,
+			`"tagIds"`:       input.TagIds,
+			`"tags"`:         input.Tags,
+			`"isMature"`:     input.IsMature,
+		},
+	)
+
+	query, args, err := updateBuilder.ToSql()
+	if err != nil {
+		return err
+	}
+
+	conn := c.getter.DefaultTrOrDB(ctx, c.pool)
+	_, err = conn.Exec(ctx, query, args...)
+	return err
+}

libs/repositories/streams/datasource/postgres/postgres.go

@@ -0,0 +1,26 @@
+package model
+
+import (
+	"time"
+)
+
+type Stream struct {
+	ID           string
+	UserId       string
+	UserLogin    string
+	UserName     string
+	GameId       string
+	GameName     string
+	CommunityIds []string
+	Type         string
+	Title        string
+	ViewerCount  int
+	StartedAt    time.Time
+	Language     string
+	ThumbnailUrl string
+	TagIds       []string
+	Tags         []string
+	IsMature     bool
+}
+
+var Nil = Stream{}

libs/repositories/streams/model/stream.go

@@ -0,0 +1,28 @@
+package streams
+
+import (
+	"context"
+	"time"
+
+	"github.com/twirapp/twir/libs/repositories/streams/model"
+)
+
+type Repository interface {
+	GetList(ctx context.Context) ([]model.Stream, error)
+	Update(ctx context.Context, channelID string, input UpdateInput) error
+}
+
+type UpdateInput struct {
+	GameId       *string
+	GameName     *string
+	CommunityIds []string
+	Type         *string
+	Title        *string
+	ViewerCount  *int
+	StartedAt    *time.Time
+	Language     *string
+	ThumbnailUrl *string
+	TagIds       []string
+	Tags         []string
+	IsMature     *bool
+}