diff --git a/pkg/liveStream/BUILD.bazel b/pkg/liveStream/BUILD.bazel new file mode 100644 index 0000000..3235c20 --- /dev/null +++ b/pkg/liveStream/BUILD.bazel @@ -0,0 +1,23 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "liveStream", + srcs = ["liveStream.go"], + importpath = "github.com/eat-pray-ai/yutu/pkg/liveStream", + visibility = ["//visibility:public"], + deps = [ + "//pkg/common", + "@com_github_jedib0t_go_pretty_v6//table", + "@org_golang_google_api//youtube/v3:youtube", + ], +) + +go_test( + name = "liveStream_test", + srcs = ["liveStream_test.go"], + embed = [":liveStream"], + deps = [ + "//pkg/common", + "@org_golang_google_api//youtube/v3:youtube", + ], +) diff --git a/pkg/liveStream/liveStream.go b/pkg/liveStream/liveStream.go new file mode 100644 index 0000000..d08c3ff --- /dev/null +++ b/pkg/liveStream/liveStream.go @@ -0,0 +1,276 @@ +// Copyright 2026 eat-pray-ai & OpenWaygate +// SPDX-License-Identifier: Apache-2.0 + +package liveStream + +import ( + "errors" + "fmt" + "io" + + "github.com/eat-pray-ai/yutu/pkg/common" + "github.com/jedib0t/go-pretty/v6/table" + "google.golang.org/api/youtube/v3" +) + +var ( + errGetLiveStream = errors.New("failed to get live stream") + errInsertLiveStream = errors.New("failed to insert live stream") + errUpdateLiveStream = errors.New("failed to update live stream") + errDeleteLiveStream = errors.New("failed to delete live stream") +) + +type LiveStream struct { + common.Fields + Title string `yaml:"title" json:"title,omitempty"` + Description string `yaml:"description" json:"description,omitempty"` + Mine *bool `yaml:"mine" json:"mine,omitempty"` + FrameRate string `yaml:"frame_rate" json:"frame_rate,omitempty"` + IngestionType string `yaml:"ingestion_type" json:"ingestion_type,omitempty"` + Resolution string `yaml:"resolution" json:"resolution,omitempty"` + + OnBehalfOfContentOwnerChannel string `yaml:"on_behalf_of_content_owner_channel" json:"on_behalf_of_content_owner_channel,omitempty"` +} + +type ILiveStream[T any] interface { + List(io.Writer) error + Get() ([]*T, error) + Insert(io.Writer) error + Update(io.Writer) error + Delete(io.Writer) error +} + +type Option func(*LiveStream) + +func NewLiveStream(opts ...Option) ILiveStream[youtube.LiveStream] { + s := &LiveStream{Fields: common.Fields{}} + for _, opt := range opts { + opt(s) + } + return s +} + +func (s *LiveStream) Get() ([]*youtube.LiveStream, error) { + if err := s.EnsureService(); err != nil { + return nil, err + } + call := s.Service.LiveStreams.List(s.Parts) + if len(s.Ids) > 0 { + call = call.Id(s.Ids...) + } + if s.Mine != nil && *s.Mine { + call = call.Mine(true) + } + if s.OnBehalfOfContentOwner != "" { + call = call.OnBehalfOfContentOwner(s.OnBehalfOfContentOwner) + } + if s.OnBehalfOfContentOwnerChannel != "" { + call = call.OnBehalfOfContentOwnerChannel(s.OnBehalfOfContentOwnerChannel) + } + + return common.Paginate( + &s.Fields, call, + func(r *youtube.LiveStreamListResponse) ([]*youtube.LiveStream, string) { + return r.Items, r.NextPageToken + }, errGetLiveStream, + ) +} + +func (s *LiveStream) List(writer io.Writer) error { + streams, err := s.Get() + if err != nil && streams == nil { + return err + } + + common.PrintList( + s.Output, streams, writer, + table.Row{"ID", "Title", "Status"}, + func(stream *youtube.LiveStream) table.Row { + title := "" + status := "" + if stream.Snippet != nil { + title = stream.Snippet.Title + } + if stream.Status != nil { + status = stream.Status.StreamStatus + } + return table.Row{stream.Id, title, status} + }, + ) + return err +} + +func (s *LiveStream) Insert(writer io.Writer) error { + if err := s.EnsureService(); err != nil { + return err + } + stream := &youtube.LiveStream{ + Snippet: &youtube.LiveStreamSnippet{ + Title: s.Title, + Description: s.Description, + }, + Cdn: &youtube.CdnSettings{ + FrameRate: s.FrameRate, + IngestionType: s.IngestionType, + Resolution: s.Resolution, + }, + } + + call := s.Service.LiveStreams.Insert(s.Parts, stream) + if s.OnBehalfOfContentOwner != "" { + call = call.OnBehalfOfContentOwner(s.OnBehalfOfContentOwner) + } + if s.OnBehalfOfContentOwnerChannel != "" { + call = call.OnBehalfOfContentOwnerChannel(s.OnBehalfOfContentOwnerChannel) + } + + res, err := call.Do() + if err != nil { + return errors.Join(errInsertLiveStream, err) + } + + common.PrintResult( + s.Output, res, writer, "Live stream inserted: %s\n", res.Id, + ) + return nil +} + +func (s *LiveStream) Update(writer io.Writer) error { + if err := s.EnsureService(); err != nil { + return err + } + s.Parts = []string{"id", "snippet", "cdn"} + streams, err := s.Get() + if err != nil { + return errors.Join(errUpdateLiveStream, err) + } + if len(streams) == 0 { + return errGetLiveStream + } + + original := streams[0] + stream := &youtube.LiveStream{ + Id: original.Id, + Snippet: &youtube.LiveStreamSnippet{ + Title: original.Snippet.Title, + Description: original.Snippet.Description, + }, + Cdn: &youtube.CdnSettings{}, + } + if original.Cdn != nil { + stream.Cdn.FrameRate = original.Cdn.FrameRate + stream.Cdn.IngestionType = original.Cdn.IngestionType + stream.Cdn.Resolution = original.Cdn.Resolution + } + + if s.Title != "" { + stream.Snippet.Title = s.Title + } + if s.Description != "" { + stream.Snippet.Description = s.Description + } + if s.FrameRate != "" { + stream.Cdn.FrameRate = s.FrameRate + } + if s.IngestionType != "" { + stream.Cdn.IngestionType = s.IngestionType + } + if s.Resolution != "" { + stream.Cdn.Resolution = s.Resolution + } + + call := s.Service.LiveStreams.Update([]string{"snippet", "cdn"}, stream) + if s.OnBehalfOfContentOwner != "" { + call = call.OnBehalfOfContentOwner(s.OnBehalfOfContentOwner) + } + if s.OnBehalfOfContentOwnerChannel != "" { + call = call.OnBehalfOfContentOwnerChannel(s.OnBehalfOfContentOwnerChannel) + } + + res, err := call.Do() + if err != nil { + return errors.Join(errUpdateLiveStream, err) + } + + common.PrintResult( + s.Output, res, writer, "Live stream updated: %s\n", res.Id, + ) + return nil +} + +func (s *LiveStream) Delete(writer io.Writer) error { + if err := s.EnsureService(); err != nil { + return err + } + for _, id := range s.Ids { + call := s.Service.LiveStreams.Delete(id) + if s.OnBehalfOfContentOwner != "" { + call = call.OnBehalfOfContentOwner(s.OnBehalfOfContentOwner) + } + if s.OnBehalfOfContentOwnerChannel != "" { + call = call.OnBehalfOfContentOwnerChannel(s.OnBehalfOfContentOwnerChannel) + } + + err := call.Do() + if err != nil { + return errors.Join(errDeleteLiveStream, err) + } + + _, _ = fmt.Fprintf(writer, "Live stream %s deleted\n", id) + } + return nil +} + +func WithTitle(title string) Option { + return func(s *LiveStream) { + s.Title = title + } +} + +func WithDescription(description string) Option { + return func(s *LiveStream) { + s.Description = description + } +} + +func WithMine(mine *bool) Option { + return func(s *LiveStream) { + if mine != nil { + s.Mine = mine + } + } +} + +func WithFrameRate(frameRate string) Option { + return func(s *LiveStream) { + s.FrameRate = frameRate + } +} + +func WithIngestionType(ingestionType string) Option { + return func(s *LiveStream) { + s.IngestionType = ingestionType + } +} + +func WithResolution(resolution string) Option { + return func(s *LiveStream) { + s.Resolution = resolution + } +} + +func WithOnBehalfOfContentOwnerChannel(channel string) Option { + return func(s *LiveStream) { + s.OnBehalfOfContentOwnerChannel = channel + } +} + +var ( + WithMaxResults = common.WithMaxResults[*LiveStream] + WithParts = common.WithParts[*LiveStream] + WithOutput = common.WithOutput[*LiveStream] + WithService = common.WithService[*LiveStream] + WithIds = common.WithIds[*LiveStream] + + WithOnBehalfOfContentOwner = common.WithOnBehalfOfContentOwner[*LiveStream] +) diff --git a/pkg/liveStream/liveStream_test.go b/pkg/liveStream/liveStream_test.go new file mode 100644 index 0000000..5d5414d --- /dev/null +++ b/pkg/liveStream/liveStream_test.go @@ -0,0 +1,493 @@ +// Copyright 2026 eat-pray-ai & OpenWaygate +// SPDX-License-Identifier: Apache-2.0 + +package liveStream + +import ( + "bytes" + "encoding/json" + "io" + "math" + "net/http" + "reflect" + "testing" + + "github.com/eat-pray-ai/yutu/pkg/common" + "google.golang.org/api/youtube/v3" +) + +func TestNewLiveStream(t *testing.T) { + type args struct { + opts []Option + } + + svc := &youtube.Service{} + mine := true + + tests := []struct { + name string + args args + want ILiveStream[youtube.LiveStream] + }{ + { + name: "with all options", + args: args{ + opts: []Option{ + WithIds([]string{"stream1", "stream2"}), + WithTitle("Test Stream"), + WithDescription("A test stream"), + WithMine(&mine), + WithFrameRate("60fps"), + WithIngestionType("rtmp"), + WithResolution("1080p"), + WithMaxResults(50), + WithParts([]string{"snippet", "cdn", "status"}), + WithOutput("json"), + WithService(svc), + WithOnBehalfOfContentOwner("owner123"), + WithOnBehalfOfContentOwnerChannel("channel123"), + }, + }, + want: &LiveStream{ + Fields: common.Fields{ + Service: svc, + Parts: []string{"snippet", "cdn", "status"}, + Output: "json", + MaxResults: 50, + Ids: []string{"stream1", "stream2"}, + OnBehalfOfContentOwner: "owner123", + }, + Title: "Test Stream", + Description: "A test stream", + Mine: &mine, + FrameRate: "60fps", + IngestionType: "rtmp", + Resolution: "1080p", + OnBehalfOfContentOwnerChannel: "channel123", + }, + }, + { + name: "with no options", + args: args{ + opts: []Option{}, + }, + want: &LiveStream{Fields: common.Fields{}}, + }, + { + name: "with zero max results", + args: args{ + opts: []Option{ + WithMaxResults(0), + }, + }, + want: &LiveStream{ + Fields: common.Fields{MaxResults: math.MaxInt64}, + }, + }, + { + name: "with negative max results", + args: args{ + opts: []Option{ + WithMaxResults(-10), + }, + }, + want: &LiveStream{ + Fields: common.Fields{MaxResults: 1}, + }, + }, + { + name: "with empty string values", + args: args{ + opts: []Option{ + WithTitle(""), + WithDescription(""), + WithFrameRate(""), + WithIngestionType(""), + WithResolution(""), + }, + }, + want: &LiveStream{ + Fields: common.Fields{}, + Title: "", + Description: "", + FrameRate: "", + IngestionType: "", + Resolution: "", + }, + }, + { + name: "with partial options", + args: args{ + opts: []Option{ + WithTitle("My Stream"), + WithIngestionType("rtmp"), + WithMaxResults(25), + WithParts([]string{"snippet"}), + }, + }, + want: &LiveStream{ + Fields: common.Fields{ + Parts: []string{"snippet"}, + MaxResults: 25, + }, + Title: "My Stream", + IngestionType: "rtmp", + }, + }, + } + + for _, tt := range tests { + t.Run( + tt.name, func(t *testing.T) { + if got := NewLiveStream(tt.args.opts...); !reflect.DeepEqual( + got, tt.want, + ) { + t.Errorf("%s\nNewLiveStream() = %v\nwant %v", tt.name, got, tt.want) + } + }, + ) + } +} + +func TestLiveStream_Get(t *testing.T) { + tests := []struct { + name string + opts []Option + verify func(*http.Request) + wantLen int + wantErr bool + }{ + { + name: "get live streams by id", + opts: []Option{ + WithIds([]string{"stream-1"}), + WithMaxResults(1), + }, + verify: func(r *http.Request) { + q := r.URL.Query() + if q.Get("id") != "stream-1" { + t.Errorf( + "expected id=stream-1, got %s", + q.Get("id"), + ) + } + }, + wantLen: 1, + wantErr: false, + }, + { + name: "get live streams with mine", + opts: func() []Option { + mine := true + return []Option{ + WithMine(&mine), + WithMaxResults(1), + } + }(), + verify: func(r *http.Request) { + q := r.URL.Query() + if q.Get("mine") != "true" { + t.Errorf( + "expected mine=true, got %s", + q.Get("mine"), + ) + } + }, + wantLen: 1, + wantErr: false, + }, + } + + for _, tt := range tests { + t.Run( + tt.name, func(t *testing.T) { + svc := common.NewTestService( + t, http.HandlerFunc( + func(w http.ResponseWriter, r *http.Request) { + if tt.verify != nil { + tt.verify(r) + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write( + []byte(`{ + "items": [ + { + "id": "stream-1", + "snippet": { + "title": "Test Stream", + "description": "A test stream" + }, + "status": { + "streamStatus": "active" + } + } + ] + }`), + ) + }, + ), + ) + + opts := append([]Option{WithService(svc)}, tt.opts...) + s := NewLiveStream(opts...) + got, err := s.Get() + if (err != nil) != tt.wantErr { + t.Errorf("LiveStream.Get() error = %v, wantErr %v", err, tt.wantErr) + return + } + if len(got) != tt.wantLen { + t.Errorf( + "LiveStream.Get() got length = %v, want %v", len(got), tt.wantLen, + ) + } + }, + ) + } +} + +func TestLiveStream_Get_Pagination(t *testing.T) { + svc := common.NewTestService(t, common.PaginationHandler("stream")) + + mine := true + s := NewLiveStream( + WithService(svc), + WithMine(&mine), + WithMaxResults(22), + ) + got, err := s.Get() + if err != nil { + t.Errorf("LiveStream.Get() error = %v", err) + } + if len(got) != 22 { + t.Errorf("LiveStream.Get() got length = %v, want 22", len(got)) + } +} + +func TestLiveStream_List(t *testing.T) { + mockResponse := `{ + "items": [ + { + "id": "stream-1", + "snippet": { + "title": "Test Stream", + "description": "A test stream" + }, + "status": { + "streamStatus": "active" + } + } + ] + }` + + mine := true + common.RunListTest( + t, mockResponse, + func(svc *youtube.Service, output string) func(io.Writer) error { + s := NewLiveStream( + WithService(svc), + WithMine(&mine), + WithOutput(output), + WithMaxResults(1), + ) + return s.List + }, + ) +} + +func TestLiveStream_Insert(t *testing.T) { + tests := []struct { + name string + opts []Option + verify func(*http.Request) + wantErr bool + }{ + { + name: "insert live stream", + opts: []Option{ + WithTitle("New Stream"), + WithDescription("A new live stream"), + WithFrameRate("60fps"), + WithIngestionType("rtmp"), + WithResolution("1080p"), + WithParts([]string{"snippet", "cdn"}), + }, + verify: func(r *http.Request) { + if r.Method != "POST" { + t.Errorf("expected POST, got %s", r.Method) + } + + defer func() { _ = r.Body.Close() }() + var body struct { + Snippet struct { + Title string `json:"title"` + Description string `json:"description"` + } `json:"snippet"` + Cdn struct { + FrameRate string `json:"frameRate"` + IngestionType string `json:"ingestionType"` + Resolution string `json:"resolution"` + } `json:"cdn"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + t.Fatalf("failed to decode request body: %v", err) + } + + if body.Snippet.Title != "New Stream" { + t.Errorf( + "expected snippet.title=New Stream, got %s", body.Snippet.Title, + ) + } + if body.Snippet.Description != "A new live stream" { + t.Errorf( + "expected snippet.description=A new live stream, got %s", + body.Snippet.Description, + ) + } + if body.Cdn.FrameRate != "60fps" { + t.Errorf("expected cdn.frameRate=60fps, got %s", body.Cdn.FrameRate) + } + if body.Cdn.IngestionType != "rtmp" { + t.Errorf( + "expected cdn.ingestionType=rtmp, got %s", body.Cdn.IngestionType, + ) + } + if body.Cdn.Resolution != "1080p" { + t.Errorf("expected cdn.resolution=1080p, got %s", body.Cdn.Resolution) + } + }, + wantErr: false, + }, + } + + for _, tt := range tests { + t.Run( + tt.name, func(t *testing.T) { + svc := common.NewTestService( + t, http.HandlerFunc( + func(w http.ResponseWriter, r *http.Request) { + if tt.verify != nil { + tt.verify(r) + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"id": "new-stream-id"}`)) + }, + ), + ) + + opts := append([]Option{WithService(svc)}, tt.opts...) + s := NewLiveStream(opts...) + var buf bytes.Buffer + if err := s.Insert(&buf); (err != nil) != tt.wantErr { + t.Errorf( + "LiveStream.Insert() error = %v, wantErr %v", err, tt.wantErr, + ) + } + }, + ) + } +} + +func TestLiveStream_Update(t *testing.T) { + tests := []struct { + name string + opts []Option + wantErr bool + }{ + { + name: "update live stream", + opts: []Option{ + WithIds([]string{"stream-1"}), + WithTitle("Updated Stream"), + WithDescription("Updated description"), + WithMaxResults(1), + }, + wantErr: false, + }, + } + + for _, tt := range tests { + t.Run( + tt.name, func(t *testing.T) { + callCount := 0 + svc := common.NewTestService( + t, http.HandlerFunc( + func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + callCount++ + if callCount == 1 { + _, _ = w.Write( + []byte(`{ + "items": [{ + "id": "stream-1", + "snippet": {"title": "Old Stream", "description": "Old description"}, + "cdn": {"frameRate": "30fps", "ingestionType": "rtmp", "resolution": "720p"} + }] + }`), + ) + } else { + _, _ = w.Write([]byte(`{"id": "stream-1"}`)) + } + }, + ), + ) + + opts := append([]Option{WithService(svc)}, tt.opts...) + s := NewLiveStream(opts...) + var buf bytes.Buffer + if err := s.Update(&buf); (err != nil) != tt.wantErr { + t.Errorf( + "LiveStream.Update() error = %v, wantErr %v", err, tt.wantErr, + ) + } + }, + ) + } +} + +func TestLiveStream_Delete(t *testing.T) { + tests := []struct { + name string + opts []Option + verify func(*http.Request) + wantErr bool + }{ + { + name: "delete live stream", + opts: []Option{ + WithIds([]string{"stream-id"}), + }, + verify: func(r *http.Request) { + if r.Method != "DELETE" { + t.Errorf("expected DELETE, got %s", r.Method) + } + }, + wantErr: false, + }, + } + + for _, tt := range tests { + t.Run( + tt.name, func(t *testing.T) { + svc := common.NewTestService( + t, http.HandlerFunc( + func(w http.ResponseWriter, r *http.Request) { + if tt.verify != nil { + tt.verify(r) + } + w.WriteHeader(http.StatusNoContent) + }, + ), + ) + + opts := append([]Option{WithService(svc)}, tt.opts...) + s := NewLiveStream(opts...) + var buf bytes.Buffer + if err := s.Delete(&buf); (err != nil) != tt.wantErr { + t.Errorf( + "LiveStream.Delete() error = %v, wantErr %v", err, tt.wantErr, + ) + } + }, + ) + } +}