feat: split steam categories into separate table and print time in offline message

ent/migrate/migrations/atlas.sum

@@ -1,3 +1,4 @@
-h1:x0Y0JlMrA7W3oMyLCbIpQ7QRqckGpga6Rw6vB0kwap0=
+h1:+Uq126QTEe4nFfdPa9fjq+ZlDcgw5trXo/I/+9/2Tgc=
 20230327181912_initial.sql h1:L6nniWh3O35p6lwgIG+NrRkkU2n2iG48Be5/zfPAz0I=
 20230401125338_TitleChangeNotification.sql h1:9u5qCYBNNL6RHtLdcW691tRhYyli7/pG2kBxYuHs1fA=
+20230408113330_add_stream_categories_history.sql h1:LeWjpeGt1nRgvVP3/mCXNeWKOuhxBhqAQEpqW9Aadwc=

ent/schema/stream.go

@@ -3,6 +3,7 @@ package schema
 import (
 	"entgo.io/ent"
 	"entgo.io/ent/dialect"
+	"entgo.io/ent/dialect/entsql"
 	"entgo.io/ent/schema/edge"
 	"entgo.io/ent/schema/field"
 	"github.com/google/uuid"
@@ -30,13 +31,6 @@ func (Stream) Fields() []ent.Field {
 		//	"postgres": "text[]",
 		//	"sqlite":   "text[]",
 		//}),
-		field.Other("categories", pq.StringArray{}).
-			SchemaType(map[string]string{
-				dialect.Postgres: "text[]",
-				dialect.SQLite:   "JSON",
-			}).
-			Default(pq.StringArray{}).
-			Optional(),
 
 		//SchemaType(map[string]string{
 		//	"postgres": "text[]",
@@ -56,5 +50,8 @@ func (Stream) Edges() []ent.Edge {
 			Required().
 			Unique().
 			Field("channel_id"),
+		edge.To("stream_categories", StreamCategory.Type).Annotations(entsql.Annotation{
+			OnDelete: entsql.Cascade,
+		}),
 	}
 }

ent/schema/stream_category.go

@@ -0,0 +1,33 @@
+package schema
+
+import (
+	"entgo.io/ent"
+	"entgo.io/ent/schema/edge"
+	"entgo.io/ent/schema/field"
+	"github.com/google/uuid"
+	"time"
+)
+
+type StreamCategory struct {
+	ent.Schema
+}
+
+func (StreamCategory) Fields() []ent.Field {
+	return []ent.Field{
+		field.UUID("id", uuid.UUID{}).Default(uuid.New),
+		field.String("stream_id"),
+		field.String("name"),
+
+		field.Time("setted_at").Optional().Default(time.Now().UTC),
+	}
+}
+
+func (StreamCategory) Edges() []ent.Edge {
+	return []ent.Edge{
+		edge.From("stream", Stream.Type).
+			Ref("stream_categories").
+			Required().
+			Unique().
+			Field("stream_id"),
+	}
+}

internal/db/db_models/stream.go

@@ -5,12 +5,17 @@ import (
 	"time"
 )
 
+type StreamCategory struct {
+	Name     string    `json:"name"`
+	SettedAt time.Time `json:"settedAt"`
+}
+
 type Stream struct {
-	ID         string     `json:"id,omitempty"`
-	ChannelID  uuid.UUID  `json:"channel_id,omitempty"`
-	Titles     []string   `json:"titles,omitempty"`
-	Categories []string   `json:"categories,omitempty"`
-	StartedAt  time.Time  `json:"started_at,omitempty"`
-	UpdatedAt  *time.Time `json:"updated_at,omitempty"`
-	EndedAt    *time.Time `json:"ended_at,omitempty"`
+	ID         string            `json:"id,omitempty"`
+	ChannelID  uuid.UUID         `json:"channel_id,omitempty"`
+	Titles     []string          `json:"titles,omitempty"`
+	Categories []*StreamCategory `json:"categories,omitempty"`
+	StartedAt  time.Time         `json:"started_at,omitempty"`
+	UpdatedAt  *time.Time        `json:"updated_at,omitempty"`
+	EndedAt    *time.Time        `json:"ended_at,omitempty"`
 }

internal/db/stream.go

@@ -4,13 +4,15 @@ import (
 	"context"
 	"github.com/google/uuid"
 	"github.com/satont/twitch-notifier/internal/db/db_models"
+	"time"
 )
 
 type StreamUpdateQuery struct {
-	StreamID string
-	IsLive   *bool
-	Category *string
-	Title    *string
+	StreamID  string
+	StartTime *time.Time
+	EndTime   *time.Time
+	Category  *string
+	Title     *string
 }
 
 type StreamInterface interface {

internal/db/stream_impl_ent.go

@@ -3,33 +3,49 @@ package db
 import (
 	"context"
 	"errors"
+	"time"
+
 	"github.com/google/uuid"
 	"github.com/lib/pq"
 	"github.com/satont/twitch-notifier/ent"
 	"github.com/satont/twitch-notifier/ent/channel"
 	"github.com/satont/twitch-notifier/ent/stream"
+	"github.com/satont/twitch-notifier/ent/streamcategory"
 	"github.com/satont/twitch-notifier/internal/db/db_models"
-	"time"
 )
 
 type StreamEntService struct {
 	entClient *ent.Client
 }
 
 func (s *StreamEntService) convertEntity(stream *ent.Stream) *db_models.Stream {
+	categories := make([]*db_models.StreamCategory, 0, len(stream.Edges.StreamCategories))
+
+	for _, category := range stream.Edges.StreamCategories {
+		categories = append(categories, &db_models.StreamCategory{
+			Name:     category.Name,
+			SettedAt: category.SettedAt,
+		})
+	}
+
 	return &db_models.Stream{
 		ID:         stream.ID,
 		ChannelID:  stream.ChannelID,
 		Titles:     stream.Titles,
-		Categories: stream.Categories,
+		Categories: categories,
 		StartedAt:  stream.StartedAt,
 		UpdatedAt:  stream.UpdatedAt,
 		EndedAt:    stream.EndedAt,
 	}
 }
 
 func (s *StreamEntService) GetByID(ctx context.Context, streamID string) (*db_models.Stream, error) {
-	str, err := s.entClient.Stream.Query().Where(stream.IDEQ(streamID)).Only(ctx)
+	str, err := s.entClient.Stream.Query().
+		Where(stream.IDEQ(streamID)).
+		WithStreamCategories(func(scq *ent.StreamCategoryQuery) {
+			scq.Order(ent.Desc(streamcategory.FieldSettedAt))
+		}).
+		Only(ctx)
 	if err != nil {
 		if ent.IsNotFound(err) {
 			return nil, nil
@@ -48,6 +64,9 @@ func (s *StreamEntService) GetLatestByChannelID(
 	str, err := s.entClient.Stream.
 		Query().
 		Where(stream.ChannelIDEQ(channelEntityID), stream.EndedAtIsNil()).
+		WithStreamCategories(func(scq *ent.StreamCategoryQuery) {
+			scq.Order(ent.Desc(streamcategory.FieldSettedAt))
+		}).
 		Order(ent.Desc(stream.FieldStartedAt)).
 		First(ctx)
 	if err != nil {
@@ -69,6 +88,9 @@ func (s *StreamEntService) GetManyByChannelID(
 	streams, err := s.entClient.Stream.
 		Query().
 		Where(stream.HasChannelWith(channel.IDEQ(channelEntityID))).
+		WithStreamCategories(func(scq *ent.StreamCategoryQuery) {
+			scq.Order(ent.Desc(streamcategory.FieldSettedAt))
+		}).
 		Order(ent.Desc(stream.FieldStartedAt)).
 		Limit(limit).
 		All(ctx)
@@ -100,30 +122,32 @@ func (s *StreamEntService) UpdateOneByStreamID(
 
 	query := s.entClient.Stream.UpdateOneID(str.ID)
 
-	if updateQuery.IsLive != nil && *updateQuery.IsLive {
-		query.SetStartedAt(time.Now().UTC())
+	if updateQuery.StartTime != nil {
+		query.SetStartedAt(*updateQuery.StartTime)
 	}
 
-	if updateQuery.IsLive != nil && !*updateQuery.IsLive {
-		query.SetEndedAt(time.Now().UTC())
+	if updateQuery.EndTime != nil {
+		query.SetEndedAt(*updateQuery.EndTime)
 	}
 
 	if updateQuery.Category != nil {
-		str.Categories = append(str.Categories, *updateQuery.Category)
-		query.SetCategories(str.Categories)
+		_, err = s.entClient.StreamCategory.Create().SetStreamID(str.ID).SetName(*updateQuery.Category).Save(ctx)
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	if updateQuery.Title != nil {
 		str.Titles = append(str.Titles, *updateQuery.Title)
 		query.SetTitles(str.Titles)
 	}
 
-	newStream, err := query.Save(ctx)
+	_, err = query.Save(ctx)
 	if err != nil {
 		return nil, err
 	}
 
-	return s.convertEntity(newStream), nil
+	return s.GetByID(ctx, streamID)
 }
 
 func (s *StreamEntService) CreateOneByChannelID(
@@ -142,16 +166,19 @@ func (s *StreamEntService) CreateOneByChannelID(
 		query.SetTitles(pq.StringArray{*data.Title})
 	}
 
-	if data.Category != nil {
-		query.SetCategories(pq.StringArray{*data.Category})
-	}
-
 	str, err := query.Save(ctx)
 	if err != nil {
 		return nil, err
 	}
 
-	return s.convertEntity(str), nil
+	if data.Category != nil {
+		_, err = s.entClient.StreamCategory.Create().SetStreamID(str.ID).SetName(*data.Category).Save(ctx)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return s.GetByID(ctx, str.ID)
 }
 
 func NewStreamEntService(entClient *ent.Client) *StreamEntService {