✨ Add liveStream resource package

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",
+    ],
+)

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]
+)

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,
+					)
+				}
+			},
+		)
+	}
+}