{error} diff --git a/lexicons/place/stream/server/defs.json b/lexicons/place/stream/server/defs.json index 5d5f0edc7..3b85510e0 100644 --- a/lexicons/place/stream/server/defs.json +++ b/lexicons/place/stream/server/defs.json @@ -119,6 +119,17 @@ "description": "Whether backup storage is currently active." } } + }, + "preferences": { + "type": "object", + "description": "A user's private preferences on this Streamplace node. The node stores them itself; they are not a record in the user's repository.", + "required": ["autoPublishVods"], + "properties": { + "autoPublishVods": { + "type": "boolean", + "description": "Whether the node publishes a VOD of each of the user's recorded livestreams as soon as the livestream ends." + } + } } } } diff --git a/lexicons/place/stream/server/getPreferences.json b/lexicons/place/stream/server/getPreferences.json new file mode 100644 index 000000000..24f374d71 --- /dev/null +++ b/lexicons/place/stream/server/getPreferences.json @@ -0,0 +1,23 @@ +{ + "lexicon": 1, + "id": "place.stream.server.getPreferences", + "defs": { + "main": { + "type": "query", + "description": "Get the authenticated user's private preferences on this node.", + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["preferences"], + "properties": { + "preferences": { + "type": "ref", + "ref": "place.stream.server.defs#preferences" + } + } + } + } + } + } +} diff --git a/lexicons/place/stream/server/putPreferences.json b/lexicons/place/stream/server/putPreferences.json new file mode 100644 index 000000000..db666b92b --- /dev/null +++ b/lexicons/place/stream/server/putPreferences.json @@ -0,0 +1,36 @@ +{ + "lexicon": 1, + "id": "place.stream.server.putPreferences", + "defs": { + "main": { + "type": "procedure", + "description": "Update the authenticated user's private preferences on this node. Omitted fields keep their current values.", + "input": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": [], + "properties": { + "autoPublishVods": { + "type": "boolean", + "description": "Whether the node publishes a VOD of each of the user's recorded livestreams as soon as the livestream ends." + } + } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["preferences"], + "properties": { + "preferences": { + "type": "ref", + "ref": "place.stream.server.defs#preferences" + } + } + } + } + } + } +} diff --git a/pkg/atproto/sync.go b/pkg/atproto/sync.go index a61e534e8..6824f065a 100644 --- a/pkg/atproto/sync.go +++ b/pkg/atproto/sync.go @@ -566,12 +566,20 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD // they're live somewhere but they don't have nothin' to do with us return nil } + if rec.EndedAt != nil { + // The livestream is over: publish its VOD if the streamer + // asked for that. + if err := atsync.StatefulDB.ScheduleAutoPublishVOD(ctx, userDID, aturi.String()); err != nil { + return fmt.Errorf("failed to schedule automatic VOD publishing: %w", err) + } + return nil + } log.Debug(ctx, "stream is allowed, queuing finalize task") // queue a task to clean up the livestream if it's been inactive for too long task := &statedb.FinalizeLivestreamTask{ LivestreamURI: aturi.String(), } - if rec.LastSeenAt == nil || rec.IdleTimeoutSeconds == nil || *rec.IdleTimeoutSeconds == 0 || rec.EndedAt != nil { + if rec.LastSeenAt == nil || rec.IdleTimeoutSeconds == nil || *rec.IdleTimeoutSeconds == 0 { return nil } scheduledAt, err := time.Parse(time.RFC3339, *rec.LastSeenAt) diff --git a/pkg/cmd/streamplace.go b/pkg/cmd/streamplace.go index 74bb64402..c0f6dd848 100644 --- a/pkg/cmd/streamplace.go +++ b/pkg/cmd/streamplace.go @@ -467,6 +467,15 @@ func runMain(ctx context.Context, build *config.BuildFlags, platformJobs []jobFu // (where the model is in scope) so pkg/vod stays free of pkg/model, then // concatenates the recorded MUXL objects into a VOD. state.SetLivestreamVODFinalizer(func(ctx context.Context, t statedb.FinalizeLivestreamVODTask) (string, error) { + // The finalize route refuses banned accounts up front; automatic + // publishing (statedb.AutoPublishVODTask) has no caller to refuse. + labels, err := mod.GetActiveLabels(t.RepoDID) + if err != nil { + return "", fmt.Errorf("finalize-livestream-vod: check account labels: %w", err) + } + if atproto.IsBanned(labels...) { + return "", fmt.Errorf("finalize-livestream-vod: account %s is banned; skipping upload %s", t.RepoDID, t.UploadID) + } signingKey, err := resolveLiveSigningKey(mod, t.RepoDID) if err != nil { return "", fmt.Errorf("finalize-livestream-vod: resolve signing key: %w", err) diff --git a/pkg/placestream/serverdefs.go b/pkg/placestream/serverdefs.go index b568b9388..07c6b7fed 100644 --- a/pkg/placestream/serverdefs.go +++ b/pkg/placestream/serverdefs.go @@ -11,6 +11,33 @@ import ( cbg "github.com/whyrusleeping/cbor-gen" ) +// ServerDefs_Preferences is a "preferences" in the place.stream.server.defs schema. +// +// A user's private preferences on this Streamplace node. The node stores them itself; they are not a record in the user's repository. +type ServerDefs_Preferences struct { + LexiconTypeID string `json:"$type,omitempty"` + // autoPublishVods: Whether the node publishes a VOD of each of the user's recorded livestreams as soon as the livestream ends. + AutoPublishVods bool `json:"autoPublishVods"` +} + +// RecordTypeID implements glex.Record. +func (t *ServerDefs_Preferences) RecordTypeID() string { return "place.stream.server.defs#preferences" } + +func (t *ServerDefs_Preferences) MarshalCBOR(w io.Writer) error { + if t == nil { + _, err := w.Write(cbg.CborNull) + return err + } + // stamp $type on a copy so marshal never mutates the record + cp := *t + cp.LexiconTypeID = "place.stream.server.defs#preferences" + return glex.MarshalCBOR(w, &cp) +} + +func (t *ServerDefs_Preferences) UnmarshalCBOR(r io.Reader) error { + return glex.UnmarshalCBOR(r, t) +} + // ServerDefs_RewriteRule is a "rewriteRule" in the place.stream.server.defs schema. type ServerDefs_RewriteRule struct { LexiconTypeID string `json:"$type,omitempty"` diff --git a/pkg/placestream/servergetpreferences.go b/pkg/placestream/servergetpreferences.go new file mode 100644 index 000000000..ae88e8de3 --- /dev/null +++ b/pkg/placestream/servergetpreferences.go @@ -0,0 +1,50 @@ +// Code generated by glex; DO NOT EDIT. + +// Lexicon schema: place.stream.server.getPreferences + +package placestream + +import ( + "context" + "io" + + glex "github.com/streamplace/glex/runtime" + cbg "github.com/whyrusleeping/cbor-gen" +) + +type ServerGetPreferences_Output struct { + LexiconTypeID string `json:"$type,omitempty"` + Preferences ServerDefs_Preferences `json:"preferences"` +} + +// RecordTypeID implements glex.Record. +func (t *ServerGetPreferences_Output) RecordTypeID() string { + return "place.stream.server.getPreferences" +} + +func (t *ServerGetPreferences_Output) MarshalCBOR(w io.Writer) error { + if t == nil { + _, err := w.Write(cbg.CborNull) + return err + } + // stamp $type on a copy so marshal never mutates the record + cp := *t + cp.LexiconTypeID = "place.stream.server.getPreferences" + return glex.MarshalCBOR(w, &cp) +} + +func (t *ServerGetPreferences_Output) UnmarshalCBOR(r io.Reader) error { + return glex.UnmarshalCBOR(r, t) +} + +// ServerGetPreferences calls the XRPC method "place.stream.server.getPreferences". +// +// Get the authenticated user's private preferences on this node. +func ServerGetPreferences(ctx context.Context, c glex.LexClient) (*ServerGetPreferences_Output, error) { + var out ServerGetPreferences_Output + + if err := c.LexDo(ctx, glex.Query, "", "place.stream.server.getPreferences", nil, nil, &out); err != nil { + return nil, err + } + return &out, nil +} diff --git a/pkg/placestream/serverputpreferences.go b/pkg/placestream/serverputpreferences.go new file mode 100644 index 000000000..f808ebad5 --- /dev/null +++ b/pkg/placestream/serverputpreferences.go @@ -0,0 +1,76 @@ +// Code generated by glex; DO NOT EDIT. + +// Lexicon schema: place.stream.server.putPreferences + +package placestream + +import ( + "context" + "io" + + glex "github.com/streamplace/glex/runtime" + cbg "github.com/whyrusleeping/cbor-gen" +) + +type ServerPutPreferences_Input struct { + LexiconTypeID string `json:"$type,omitempty"` + // autoPublishVods: Whether the node publishes a VOD of each of the user's recorded livestreams as soon as the livestream ends. + AutoPublishVods *bool `json:"autoPublishVods,omitempty"` +} + +// RecordTypeID implements glex.Record. +func (t *ServerPutPreferences_Input) RecordTypeID() string { + return "place.stream.server.putPreferences" +} + +func (t *ServerPutPreferences_Input) MarshalCBOR(w io.Writer) error { + if t == nil { + _, err := w.Write(cbg.CborNull) + return err + } + // stamp $type on a copy so marshal never mutates the record + cp := *t + cp.LexiconTypeID = "place.stream.server.putPreferences" + return glex.MarshalCBOR(w, &cp) +} + +func (t *ServerPutPreferences_Input) UnmarshalCBOR(r io.Reader) error { + return glex.UnmarshalCBOR(r, t) +} + +type ServerPutPreferences_Output struct { + LexiconTypeID string `json:"$type,omitempty"` + Preferences ServerDefs_Preferences `json:"preferences"` +} + +// RecordTypeID implements glex.Record. +func (t *ServerPutPreferences_Output) RecordTypeID() string { + return "place.stream.server.putPreferences" +} + +func (t *ServerPutPreferences_Output) MarshalCBOR(w io.Writer) error { + if t == nil { + _, err := w.Write(cbg.CborNull) + return err + } + // stamp $type on a copy so marshal never mutates the record + cp := *t + cp.LexiconTypeID = "place.stream.server.putPreferences" + return glex.MarshalCBOR(w, &cp) +} + +func (t *ServerPutPreferences_Output) UnmarshalCBOR(r io.Reader) error { + return glex.UnmarshalCBOR(r, t) +} + +// ServerPutPreferences calls the XRPC method "place.stream.server.putPreferences". +// +// Update the authenticated user's private preferences on this node. Omitted fields keep their current values. +func ServerPutPreferences(ctx context.Context, c glex.LexClient, input *ServerPutPreferences_Input) (*ServerPutPreferences_Output, error) { + var out ServerPutPreferences_Output + + if err := c.LexDo(ctx, glex.Procedure, "application/json", "place.stream.server.putPreferences", nil, input, &out); err != nil { + return nil, err + } + return &out, nil +} diff --git a/pkg/spxrpc/media_manage.go b/pkg/spxrpc/media_manage.go index 4dac84ac8..6d1f5211d 100644 --- a/pkg/spxrpc/media_manage.go +++ b/pkg/spxrpc/media_manage.go @@ -10,9 +10,7 @@ import ( "github.com/bluesky-social/indigo/atproto/syntax" "github.com/labstack/echo/v4" - "stream.place/streamplace/pkg/comatproto" "stream.place/streamplace/pkg/log" - "stream.place/streamplace/pkg/model" "stream.place/streamplace/pkg/placestream" "stream.place/streamplace/pkg/statedb" "stream.place/streamplace/pkg/vod" @@ -42,18 +40,11 @@ func (s *Server) requireLivestreamManage(ctx context.Context, streamer, action s return modCtx, nil } -// livestreamItem is one livestream record: the indexed row and its decoded -// record. -type livestreamItem struct { - ls *model.Livestream - rec *placestream.Livestream -} - // livestreamItems resolves livestream URIs to their indexed rows and decoded // records, checks they belong to one streamer, and orders them by creation: // recording order is record order, whatever order the caller listed them in. -func (s *Server) livestreamItems(uris []string) ([]livestreamItem, error) { - items := make([]livestreamItem, 0, len(uris)) +func (s *Server) livestreamItems(uris []string) ([]statedb.LivestreamItem, error) { + items := make([]statedb.LivestreamItem, 0, len(uris)) for _, u := range uris { ls, err := s.model.GetLivestream(u) if err != nil { @@ -70,14 +61,14 @@ func (s *Server) livestreamItems(uris []string) ([]livestreamItem, error) { if !ok { return nil, echo.NewHTTPError(http.StatusInternalServerError, "record is not a place.stream.livestream: "+u) } - items = append(items, livestreamItem{ls: ls, rec: rec}) + items = append(items, statedb.LivestreamItem{Livestream: ls, Record: rec}) } for _, it := range items[1:] { - if it.ls.RepoDID != items[0].ls.RepoDID { + if it.Livestream.RepoDID != items[0].Livestream.RepoDID { return nil, echo.NewHTTPError(http.StatusNotFound, "LivestreamNotFound: the livestreams belong to different streamers") } } - sort.SliceStable(items, func(i, j int) bool { return items[i].rec.CreatedAt < items[j].rec.CreatedAt }) + sort.SliceStable(items, func(i, j int) bool { return items[i].Record.CreatedAt < items[j].Record.CreatedAt }) return items, nil } @@ -108,46 +99,6 @@ func deref(p *string) string { return *p } -// videoDraftForLivestreams describes the place.stream.video record for the -// VOD of one or more livestream records: the first one's title (or the given -// one), description, tags and activity, connected to every livestream record -// the way the app's draft is connected to its one. Duration, source tracks -// and thumbnail are filled in at publish time from the finalized upload. -func videoDraftForLivestreams(items []livestreamItem, title, description string) *statedb.VideoDraft { - first := items[0].rec - title = strings.TrimSpace(title) - if title == "" { - title = strings.TrimSpace(first.Title) - } - if title == "" { - title = "Livestream" - } - v := &statedb.VideoDraft{ - Title: title, - Tags: first.Tags, - } - for _, it := range items { - v.Connections = append(v.Connections, placestream.Video_Connections_Elem{ - Video_Connection: &placestream.Video_Connection{ - LexiconTypeID: "place.stream.video#connection", - Ref: &comatproto.RepoStrongRef{Uri: it.ls.URI, Cid: it.ls.CID}, - }, - }) - } - if d := strings.TrimSpace(description); d != "" { - v.Description = &d - } - if first.Activity != nil { - switch { - case first.Activity.Defs_ActivityGame != nil: - v.Activity = &placestream.Video_Activity{Defs_ActivityGame: first.Activity.Defs_ActivityGame} - case first.Activity.Defs_ActivityLabel != nil: - v.Activity = &placestream.Video_Activity{Defs_ActivityLabel: first.Activity.Defs_ActivityLabel} - } - } - return v -} - // videoOwner is the repo a place.stream.video URI lives in. func videoOwner(uri string) (string, error) { aturi, err := syntax.ParseATURI(uri) diff --git a/pkg/spxrpc/media_manage_test.go b/pkg/spxrpc/media_manage_test.go index e8efde263..5a0986a76 100644 --- a/pkg/spxrpc/media_manage_test.go +++ b/pkg/spxrpc/media_manage_test.go @@ -1,72 +1,11 @@ package spxrpc import ( - "encoding/json" "testing" "github.com/stretchr/testify/require" - "stream.place/streamplace/pkg/model" - "stream.place/streamplace/pkg/placestream" - "stream.place/streamplace/pkg/statedb" ) -func TestVideoDraftForLivestreams(t *testing.T) { - a := livestreamItem{ - ls: &model.Livestream{URI: "at://did:plc:x/place.stream.livestream/first", CID: "bafyfirst"}, - rec: &placestream.Livestream{Title: " State of the Union ", Tags: []string{"politics"}, - Activity: &placestream.Livestream_Activity{Defs_ActivityLabel: &placestream.Defs_ActivityLabel{Label: "Talk"}}}, - } - b := livestreamItem{ - ls: &model.Livestream{URI: "at://did:plc:x/place.stream.livestream/second", CID: "bafysecond"}, - rec: &placestream.Livestream{Title: "State of the Union (continued)"}, - } - - v := videoDraftForLivestreams([]livestreamItem{a, b}, "", "") - require.Equal(t, "State of the Union", v.Title, "the first livestream's title by default") - require.Nil(t, v.Description) - require.Equal(t, []string{"politics"}, v.Tags) - require.NotNil(t, v.Activity.Defs_ActivityLabel) - require.Len(t, v.Connections, 2, "connected to every livestream it came from") - require.Equal(t, a.ls.URI, v.Connections[0].Video_Connection.Ref.Uri) - require.Equal(t, a.ls.CID, v.Connections[0].Video_Connection.Ref.Cid) - require.Equal(t, b.ls.URI, v.Connections[1].Video_Connection.Ref.Uri) - - v = videoDraftForLivestreams([]livestreamItem{a}, "Replay", " Full speech. ") - require.Equal(t, "Replay", v.Title) - require.Equal(t, "Full speech.", *v.Description) - - v = videoDraftForLivestreams([]livestreamItem{{ls: a.ls, rec: &placestream.Livestream{}}}, "", "") - require.Equal(t, "Livestream", v.Title, "a record with no title at all") -} - -// The draft rides in a task payload, so it has to survive JSON, and the -// record built from it has to be the one PublishVideo expects (a full Video -// with an empty source union does not marshal at all, which is why the -// payload carries a draft and not a record). -func TestVideoDraftPayloadRoundTrip(t *testing.T) { - a := livestreamItem{ - ls: &model.Livestream{URI: "at://did:plc:x/place.stream.livestream/first", CID: "bafyfirst"}, - rec: &placestream.Livestream{Title: "Speech", Tags: []string{"politics"}, - Activity: &placestream.Livestream_Activity{Defs_ActivityLabel: &placestream.Defs_ActivityLabel{Label: "Talk"}}}, - } - task := statedb.FinalizeLivestreamVODTask{UploadID: "u", RepoDID: "did:plc:x", LivestreamURI: a.ls.URI, - Publish: videoDraftForLivestreams([]livestreamItem{a}, "", "the whole thing")} - b, err := json.Marshal(task) - require.NoError(t, err) - var back statedb.FinalizeLivestreamVODTask - require.NoError(t, json.Unmarshal(b, &back)) - require.NotNil(t, back.Publish) - rec := back.Publish.Record() - require.Equal(t, "Speech", rec.Title) - require.Equal(t, "the whole thing", *rec.Description) - require.Equal(t, []string{"politics"}, rec.Tags) - require.Equal(t, "Talk", rec.Activity.Defs_ActivityLabel.Label) - require.Len(t, rec.Connections, 1) - require.Equal(t, a.ls.URI, rec.Connections[0].Video_Connection.Ref.Uri) - require.Equal(t, a.ls.CID, rec.Connections[0].Video_Connection.Ref.Cid) - require.Equal(t, "place.stream.video", rec.RecordTypeID()) -} - func TestUniqueURIs(t *testing.T) { one := " at://a " require.Equal(t, []string{"at://a", "at://b"}, uniqueURIs(&one, []string{"at://b", "", "at://a"})) diff --git a/pkg/spxrpc/place_stream_media_finalizelivestream.go b/pkg/spxrpc/place_stream_media_finalizelivestream.go index 86d63c148..49d247292 100644 --- a/pkg/spxrpc/place_stream_media_finalizelivestream.go +++ b/pkg/spxrpc/place_stream_media_finalizelivestream.go @@ -5,7 +5,6 @@ import ( "net/http" "strings" - "github.com/google/uuid" "github.com/labstack/echo/v4" "stream.place/streamplace/pkg/log" placestream "stream.place/streamplace/pkg/placestream" @@ -30,10 +29,10 @@ func (s *Server) handlePlaceStreamMediaFinalizeLivestream(ctx context.Context, b if err != nil { return nil, err } - streamer := items[0].ls.RepoDID + streamer := items[0].Livestream.RepoDID ordered := make([]string, len(items)) for i, it := range items { - ordered[i] = it.ls.URI + ordered[i] = it.Livestream.URI } modCtx, err := s.requireLivestreamManage(ctx, streamer, "finalizeLivestream") if err != nil { @@ -57,22 +56,10 @@ func (s *Server) handlePlaceStreamMediaFinalizeLivestream(ctx context.Context, b return nil, echo.NewHTTPError(http.StatusNotFound, "NoRecording: no completed recording objects for these livestreams") } - // Synthetic Upload row so the client reuses the getUploadStatus / - // publishVideo flow it already has for resumable uploads. - uu, err := uuid.NewV7() + uploadID, err := s.statefulDB.CreateLivestreamUpload(ctx, streamer, ordered[0]) if err != nil { return nil, echo.NewHTTPError(http.StatusInternalServerError, err.Error()) } - uploadID := uu.String() - if err := s.statefulDB.CreateUpload(ctx, &statedb.Upload{ - ID: uploadID, - RepoDID: streamer, - MimeType: "video/mp4", - Backend: "live", - Location: ordered[0], - }); err != nil { - return nil, echo.NewHTTPError(http.StatusInternalServerError, err.Error()) - } publish := body.Publish != nil && *body.Publish task := statedb.FinalizeLivestreamVODTask{UploadID: uploadID, RepoDID: streamer, LivestreamURI: ordered[0], LivestreamURIs: ordered} @@ -81,7 +68,7 @@ func (s *Server) handlePlaceStreamMediaFinalizeLivestream(ctx context.Context, b out.Objects = &objects // The video record, published now or drafted: the given title and // description over the first record's, connected to every record. - video := videoDraftForLivestreams(items, deref(body.Title), deref(body.Description)) + video := statedb.VideoDraftForLivestreams(items, deref(body.Title), deref(body.Description)) if publish { task.Publish = video } else { @@ -102,17 +89,17 @@ func (s *Server) handlePlaceStreamMediaFinalizeLivestream(ctx context.Context, b log.Log(ctx, "livestream VOD queued", "uploadId", uploadID, "objects", len(segs), "publish", publish) for _, it := range items { - if it.rec.EndedAt != nil { - out.Ended = append(out.Ended, it.ls.URI) + if it.Record.EndedAt != nil { + out.Ended = append(out.Ended, it.Livestream.URI) continue } if body.EndLivestream == nil || !*body.EndLivestream { continue } - if err := s.statefulDB.EndLivestreamRecord(ctx, it.ls, it.rec); err != nil { - log.Error(ctx, "could not end livestream record", "livestream", it.ls.URI, "error", err) + if err := s.statefulDB.EndLivestreamRecord(ctx, it.Livestream, it.Record); err != nil { + log.Error(ctx, "could not end livestream record", "livestream", it.Livestream.URI, "error", err) } else { - out.Ended = append(out.Ended, it.ls.URI) + out.Ended = append(out.Ended, it.Livestream.URI) } } return out, nil diff --git a/pkg/spxrpc/place_stream_media_publishvideo.go b/pkg/spxrpc/place_stream_media_publishvideo.go index 55ccd63bb..368ed0e2d 100644 --- a/pkg/spxrpc/place_stream_media_publishvideo.go +++ b/pkg/spxrpc/place_stream_media_publishvideo.go @@ -50,18 +50,18 @@ func (s *Server) handlePlaceStreamMediaPublishVideo(ctx context.Context, body *p if len(uris) == 0 && strings.HasPrefix(upload.Location, "at://") { uris = []string{upload.Location} } - var items []livestreamItem + var items []statedb.LivestreamItem if len(uris) > 0 { if items, err = s.livestreamItems(uris); err != nil { return nil, err } - if items[0].ls.RepoDID != upload.RepoDID { + if items[0].Livestream.RepoDID != upload.RepoDID { return nil, echo.NewHTTPError(http.StatusBadRequest, "the livestreams belong to a different streamer than the upload") } } var draft *statedb.VideoDraft if len(items) > 0 { - draft = videoDraftForLivestreams(items, deref(body.Title), deref(body.Description)) + draft = statedb.VideoDraftForLivestreams(items, deref(body.Title), deref(body.Description)) } else { title := strings.TrimSpace(deref(body.Title)) if title == "" { diff --git a/pkg/spxrpc/preferences.go b/pkg/spxrpc/preferences.go new file mode 100644 index 000000000..03f86c797 --- /dev/null +++ b/pkg/spxrpc/preferences.go @@ -0,0 +1,42 @@ +package spxrpc + +import ( + "context" + "net/http" + + "github.com/labstack/echo/v4" + "github.com/streamplace/oatproxy/pkg/oatproxy" + "stream.place/streamplace/pkg/log" + placestream "stream.place/streamplace/pkg/placestream" +) + +func (s *Server) handlePlaceStreamServerGetPreferences(ctx context.Context) (*placestream.ServerGetPreferences_Output, error) { + session, _ := oatproxy.GetOAuthSession(ctx) + if session == nil { + return nil, echo.NewHTTPError(http.StatusUnauthorized, "oauth session not found") + } + prefs, err := s.statefulDB.GetUserPreferences(ctx, session.DID) + if err != nil { + return nil, echo.NewHTTPError(http.StatusInternalServerError, "get preferences: "+err.Error()) + } + return &placestream.ServerGetPreferences_Output{Preferences: prefs.ToLexicon()}, nil +} + +func (s *Server) handlePlaceStreamServerPutPreferences(ctx context.Context, input *placestream.ServerPutPreferences_Input) (*placestream.ServerPutPreferences_Output, error) { + session, _ := oatproxy.GetOAuthSession(ctx) + if session == nil { + return nil, echo.NewHTTPError(http.StatusUnauthorized, "oauth session not found") + } + prefs, err := s.statefulDB.GetUserPreferences(ctx, session.DID) + if err != nil { + return nil, echo.NewHTTPError(http.StatusInternalServerError, "get preferences: "+err.Error()) + } + if input.AutoPublishVods != nil { + prefs.AutoPublishVODs = *input.AutoPublishVods + } + if err := s.statefulDB.PutUserPreferences(ctx, prefs); err != nil { + return nil, echo.NewHTTPError(http.StatusInternalServerError, "save preferences: "+err.Error()) + } + log.Log(ctx, "preferences updated", "did", session.DID, "autoPublishVods", prefs.AutoPublishVODs) + return &placestream.ServerPutPreferences_Output{Preferences: prefs.ToLexicon()}, nil +} diff --git a/pkg/spxrpc/stubs.go b/pkg/spxrpc/stubs.go index 11f8c795e..e8712a8e3 100644 --- a/pkg/spxrpc/stubs.go +++ b/pkg/spxrpc/stubs.go @@ -475,10 +475,12 @@ func (s *Server) RegisterHandlersPlacestream(e *echo.Echo) error { e.POST("/xrpc/place.stream.server.createWebhook", s.HandlePlaceStreamServerCreateWebhook) e.POST("/xrpc/place.stream.server.deleteStorage", s.HandlePlaceStreamServerDeleteStorage) e.POST("/xrpc/place.stream.server.deleteWebhook", s.HandlePlaceStreamServerDeleteWebhook) + e.GET("/xrpc/place.stream.server.getPreferences", s.HandlePlaceStreamServerGetPreferences) e.GET("/xrpc/place.stream.server.getServerTime", s.HandlePlaceStreamServerGetServerTime) e.GET("/xrpc/place.stream.server.getStorage", s.HandlePlaceStreamServerGetStorage) e.GET("/xrpc/place.stream.server.getWebhook", s.HandlePlaceStreamServerGetWebhook) e.GET("/xrpc/place.stream.server.listWebhooks", s.HandlePlaceStreamServerListWebhooks) + e.POST("/xrpc/place.stream.server.putPreferences", s.HandlePlaceStreamServerPutPreferences) e.POST("/xrpc/place.stream.server.updateWebhook", s.HandlePlaceStreamServerUpdateWebhook) e.POST("/xrpc/place.stream.server.upsertStorage", s.HandlePlaceStreamServerUpsertStorage) e.POST("/xrpc/place.stream.vod.createDraft", s.HandlePlaceStreamVodCreateDraft) @@ -1451,6 +1453,19 @@ func (s *Server) HandlePlaceStreamServerDeleteWebhook(c echo.Context) error { return c.JSON(200, out) } +func (s *Server) HandlePlaceStreamServerGetPreferences(c echo.Context) error { + ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandlePlaceStreamServerGetPreferences") + defer span.End() + var out *placestream.ServerGetPreferences_Output + var handleErr error + // func (s *Server) handlePlaceStreamServerGetPreferences(ctx context.Context) (*placestream.ServerGetPreferences_Output, error) + out, handleErr = s.handlePlaceStreamServerGetPreferences(ctx) + if handleErr != nil { + return handleErr + } + return c.JSON(200, out) +} + func (s *Server) HandlePlaceStreamServerGetServerTime(c echo.Context) error { ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandlePlaceStreamServerGetServerTime") defer span.End() @@ -1522,6 +1537,23 @@ func (s *Server) HandlePlaceStreamServerListWebhooks(c echo.Context) error { return c.JSON(200, out) } +func (s *Server) HandlePlaceStreamServerPutPreferences(c echo.Context) error { + ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandlePlaceStreamServerPutPreferences") + defer span.End() + var body placestream.ServerPutPreferences_Input + if err := c.Bind(&body); err != nil { + return err + } + var out *placestream.ServerPutPreferences_Output + var handleErr error + // func (s *Server) handlePlaceStreamServerPutPreferences(ctx context.Context,body *placestream.ServerPutPreferences_Input) (*placestream.ServerPutPreferences_Output, error) + out, handleErr = s.handlePlaceStreamServerPutPreferences(ctx, &body) + if handleErr != nil { + return handleErr + } + return c.JSON(200, out) +} + func (s *Server) HandlePlaceStreamServerUpdateWebhook(c echo.Context) error { ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandlePlaceStreamServerUpdateWebhook") defer span.End() diff --git a/pkg/statedb/auto_publish_vod.go b/pkg/statedb/auto_publish_vod.go new file mode 100644 index 000000000..8af9d6a38 --- /dev/null +++ b/pkg/statedb/auto_publish_vod.go @@ -0,0 +1,129 @@ +package statedb + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "stream.place/streamplace/pkg/log" + placestream "stream.place/streamplace/pkg/placestream" +) + +// AutoPublishVODTask publishes the VOD of a livestream that just ended, for +// a streamer who turned on UserPreferences.AutoPublishVODs. It is queued by +// ScheduleAutoPublishVOD when the livestream record's endedAt is indexed. +type AutoPublishVODTask struct { + LivestreamURI string `json:"livestreamURI"` + // Waits counts the passes that found part of the recording still being + // written. + Waits int `json:"waits,omitempty"` +} + +const ( + // autoPublishVODWait is how long the task waits after the livestream ends, + // and between passes that find the recording still being written. The + // recorder completes its last object when the next segment arrives or the + // stream session ends, a few seconds after the record ends. + autoPublishVODWait = 15 * time.Second + // autoPublishVODMaxWaits bounds that waiting: an object whose upload never + // completes (its node died mid-stream) would otherwise hold the VOD back + // forever, so after this many passes the VOD is made of the objects that + // did complete. + autoPublishVODMaxWaits = 8 +) + +// ScheduleAutoPublishVOD queues the publishing of an ended livestream's VOD +// if its streamer has turned automatic VOD publishing on. Once per +// livestream: indexing the same ended record again is a no-op. +func (state *StatefulDB) ScheduleAutoPublishVOD(ctx context.Context, repoDID, livestreamURI string) error { + prefs, err := state.GetUserPreferences(ctx, repoDID) + if err != nil { + return fmt.Errorf("get user preferences: %w", err) + } + if !prefs.AutoPublishVODs { + return nil + } + return state.enqueueAutoPublishVOD(ctx, AutoPublishVODTask{LivestreamURI: livestreamURI}) +} + +func (state *StatefulDB) enqueueAutoPublishVOD(ctx context.Context, t AutoPublishVODTask) error { + key := fmt.Sprintf("auto-publish-vod::%s::%d", t.LivestreamURI, t.Waits) + _, err := state.EnqueueTask(ctx, TaskAutoPublishVOD, t, WithTaskKey(key), WithScheduledAt(time.Now().Add(autoPublishVODWait).UTC())) + return err +} + +func (state *StatefulDB) processAutoPublishVODTask(ctx context.Context, task *AppTask) error { + ctx = log.WithLogValues(ctx, "func", "processAutoPublishVODTask") + var t AutoPublishVODTask + if err := json.Unmarshal(task.Payload, &t); err != nil { + return err + } + ctx = log.WithLogValues(ctx, "livestream", t.LivestreamURI) + ls, err := state.model.GetLivestream(t.LivestreamURI) + if err != nil { + return fmt.Errorf("get livestream: %w", err) + } + if ls == nil { + log.Warn(ctx, "livestream to publish a VOD of is not indexed; skipping") + return state.CompleteTask(ctx, task.ID) + } + ctx = log.WithLogValues(ctx, "did", ls.RepoDID) + + // The streamer may have turned it off since the livestream ended. + prefs, err := state.GetUserPreferences(ctx, ls.RepoDID) + if err != nil { + return fmt.Errorf("get user preferences: %w", err) + } + if !prefs.AutoPublishVODs { + log.Log(ctx, "automatic VOD publishing was turned off after the livestream ended; skipping") + return state.CompleteTask(ctx, task.ID) + } + + open, err := state.CountOpenS3Segments(ctx, ls.URI) + if err != nil { + return fmt.Errorf("count open recording objects: %w", err) + } + if open > 0 && t.Waits < autoPublishVODMaxWaits { + next := t + next.Waits++ + if err := state.enqueueAutoPublishVOD(ctx, next); err != nil { + return fmt.Errorf("reschedule automatic VOD publishing: %w", err) + } + log.Debug(ctx, "livestream recording still being written; waiting to publish its VOD", "open", open, "waits", next.Waits) + return state.CompleteTask(ctx, task.ID) + } + segs, err := state.ListS3SegmentsForLivestream(ctx, ls.URI) + if err != nil { + return fmt.Errorf("list recorded objects: %w", err) + } + if len(segs) == 0 { + log.Log(ctx, "livestream has no recording; no VOD to publish") + return state.CompleteTask(ctx, task.ID) + } + + view, err := ls.ToLivestreamView() + if err != nil { + return fmt.Errorf("decode livestream: %w", err) + } + rec, ok := view.Record.Val.(*placestream.Livestream) + if !ok { + return fmt.Errorf("record is not a place.stream.livestream: %s", ls.URI) + } + uploadID, err := state.CreateLivestreamUpload(ctx, ls.RepoDID, ls.URI) + if err != nil { + return fmt.Errorf("create upload: %w", err) + } + vodTask := FinalizeLivestreamVODTask{ + UploadID: uploadID, + RepoDID: ls.RepoDID, + LivestreamURI: ls.URI, + LivestreamURIs: []string{ls.URI}, + Publish: VideoDraftForLivestreams([]LivestreamItem{{Livestream: ls, Record: rec}}, "", ""), + } + if _, err := state.EnqueueTask(ctx, TaskFinalizeLivestreamVOD, vodTask, WithTaskKey("finalize-vod:"+uploadID)); err != nil { + return fmt.Errorf("enqueue VOD finalize: %w", err) + } + log.Log(ctx, "livestream ended; VOD queued for automatic publishing", "uploadId", uploadID, "objects", len(segs), "waits", t.Waits) + return state.CompleteTask(ctx, task.ID) +} diff --git a/pkg/statedb/auto_publish_vod_test.go b/pkg/statedb/auto_publish_vod_test.go new file mode 100644 index 000000000..88849e28e --- /dev/null +++ b/pkg/statedb/auto_publish_vod_test.go @@ -0,0 +1,184 @@ +package statedb + +import ( + "context" + "encoding/json" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/placestream" +) + +func TestUserPreferences(t *testing.T) { + WithAllDatabases(t, func(state *StatefulDB) { + ctx := context.Background() + prefs, err := state.GetUserPreferences(ctx, "did:plc:prefs") + require.NoError(t, err) + require.False(t, prefs.AutoPublishVODs, "off until the user opts in") + + prefs.AutoPublishVODs = true + require.NoError(t, state.PutUserPreferences(ctx, prefs)) + prefs, err = state.GetUserPreferences(ctx, "did:plc:prefs") + require.NoError(t, err) + require.True(t, prefs.AutoPublishVODs) + + prefs.AutoPublishVODs = false + require.NoError(t, state.PutUserPreferences(ctx, prefs), "saving over existing preferences") + prefs, err = state.GetUserPreferences(ctx, "did:plc:prefs") + require.NoError(t, err) + require.False(t, prefs.AutoPublishVODs) + + other, err := state.GetUserPreferences(ctx, "did:plc:other") + require.NoError(t, err) + require.False(t, other.AutoPublishVODs, "one user's preferences are not another's") + }) +} + +// endedLivestream seeds an ended livestream record for did, a streamer opted +// in to automatic VOD publishing. +func endedLivestream(t *testing.T, state *StatefulDB, did string) string { + t.Helper() + require.NoError(t, state.PutUserPreferences(context.Background(), &UserPreferences{RepoDID: did, AutoPublishVODs: true})) + return seedLivestream(t, state.model, did, strings.TrimPrefix(did, "did:plc:"), time.Hour, &placestream.Livestream{ + LexiconTypeID: "place.stream.livestream", + CreatedAt: time.Now().Add(-time.Hour).Format(time.RFC3339), + EndedAt: ptr(time.Now().Format(time.RFC3339)), + Title: "Speedrun night", + }) +} + +// recordObject records one live-recording object for the livestream, +// completed or still being uploaded. +func recordObject(t *testing.T, state *StatefulDB, did, uri, key string, completed bool) { + t.Helper() + ctx := context.Background() + id, err := state.RecordStart(ctx, did, "bucket", key, uri, time.Now()) + require.NoError(t, err) + if completed { + require.NoError(t, state.RecordComplete(ctx, id, 1, 100)) + } +} + +func pendingTasks(t *testing.T, state *StatefulDB, taskType string) []AppTask { + t.Helper() + tasks, err := state.ListTasks(context.Background(), TaskFilters{Type: taskType, Status: TaskStatusPending, Limit: 10}) + require.NoError(t, err) + return tasks +} + +// runAutoPublish schedules automatic publishing for the livestream and runs +// the task it queued. +func runAutoPublish(t *testing.T, state *StatefulDB, did, uri string) { + t.Helper() + require.NoError(t, state.ScheduleAutoPublishVOD(context.Background(), did, uri)) + tasks := pendingTasks(t, state, TaskAutoPublishVOD) + require.Len(t, tasks, 1) + require.NoError(t, state.processAutoPublishVODTask(context.Background(), &tasks[0])) +} + +func TestScheduleAutoPublishVOD(t *testing.T) { + WithAllDatabases(t, func(state *StatefulDB) { + ctx := context.Background() + require.NoError(t, state.ScheduleAutoPublishVOD(ctx, "did:plc:optedout", "at://did:plc:optedout/place.stream.livestream/a")) + require.Empty(t, pendingTasks(t, state, TaskAutoPublishVOD), "nothing for a streamer who has not opted in") + + did := "did:plc:optedin" + uri := endedLivestream(t, state, did) + before := time.Now() + require.NoError(t, state.ScheduleAutoPublishVOD(ctx, did, uri)) + require.NoError(t, state.ScheduleAutoPublishVOD(ctx, did, uri), "the ended record indexed again") + tasks := pendingTasks(t, state, TaskAutoPublishVOD) + require.Len(t, tasks, 1, "one VOD per livestream") + require.NotNil(t, tasks[0].ScheduledAt) + require.False(t, tasks[0].ScheduledAt.Before(before.Add(autoPublishVODWait)), "gives the recorder time to complete its last object") + }) +} + +func TestAutoPublishVODQueuesPublishingFinalize(t *testing.T) { + WithAllDatabases(t, func(state *StatefulDB) { + did := "did:plc:autovod" + uri := endedLivestream(t, state, did) + recordObject(t, state, did, uri, "a.m4s", true) + recordObject(t, state, did, uri, "b.m4s", true) + + runAutoPublish(t, state, did, uri) + + require.Empty(t, pendingTasks(t, state, TaskAutoPublishVOD)) + vods := pendingTasks(t, state, TaskFinalizeLivestreamVOD) + require.Len(t, vods, 1) + var vt FinalizeLivestreamVODTask + require.NoError(t, json.Unmarshal(vods[0].Payload, &vt)) + require.Equal(t, did, vt.RepoDID) + require.Equal(t, uri, vt.LivestreamURI) + require.NotNil(t, vt.Publish, "published, not left as a draft") + require.Equal(t, "Speedrun night", vt.Publish.Title) + require.Len(t, vt.Publish.Connections, 1) + require.Equal(t, uri, vt.Publish.Connections[0].Video_Connection.Ref.Uri) + + upload, err := state.GetUpload(context.Background(), vt.UploadID) + require.NoError(t, err) + require.Equal(t, did, upload.RepoDID) + require.Equal(t, uri, upload.Location) + }) +} + +func TestAutoPublishVODWaitsForRecording(t *testing.T) { + WithAllDatabases(t, func(state *StatefulDB) { + did := "did:plc:stillwriting" + uri := endedLivestream(t, state, did) + recordObject(t, state, did, uri, "a.m4s", true) + recordObject(t, state, did, uri, "b.m4s", false) + + runAutoPublish(t, state, did, uri) + + require.Empty(t, pendingTasks(t, state, TaskFinalizeLivestreamVOD), "not before the last object completes") + tasks := pendingTasks(t, state, TaskAutoPublishVOD) + require.Len(t, tasks, 1, "waits another pass") + var next AutoPublishVODTask + require.NoError(t, json.Unmarshal(tasks[0].Payload, &next)) + require.Equal(t, 1, next.Waits) + }) +} + +func TestAutoPublishVODStopsWaitingForAbandonedObject(t *testing.T) { + WithAllDatabases(t, func(state *StatefulDB) { + ctx := context.Background() + did := "did:plc:abandoned" + uri := endedLivestream(t, state, did) + recordObject(t, state, did, uri, "a.m4s", true) + recordObject(t, state, did, uri, "b.m4s", false) + + task, err := state.EnqueueTask(ctx, TaskAutoPublishVOD, AutoPublishVODTask{LivestreamURI: uri, Waits: autoPublishVODMaxWaits}) + require.NoError(t, err) + require.NoError(t, state.processAutoPublishVODTask(ctx, task)) + + require.Len(t, pendingTasks(t, state, TaskFinalizeLivestreamVOD), 1, "the VOD of the objects that completed") + }) +} + +func TestAutoPublishVODSkips(t *testing.T) { + WithAllDatabases(t, func(state *StatefulDB) { + ctx := context.Background() + + // No recording: the streamer was not recorded (not in the VOD beta, + // recording turned off, no S3 on this node). + did := "did:plc:unrecorded" + uri := endedLivestream(t, state, did) + runAutoPublish(t, state, did, uri) + require.Empty(t, pendingTasks(t, state, TaskFinalizeLivestreamVOD)) + + // Turned off between the livestream ending and the task running. + did = "did:plc:changedmind" + uri = endedLivestream(t, state, did) + recordObject(t, state, did, uri, "a.m4s", true) + require.NoError(t, state.ScheduleAutoPublishVOD(ctx, did, uri)) + require.NoError(t, state.PutUserPreferences(ctx, &UserPreferences{RepoDID: did, AutoPublishVODs: false})) + tasks := pendingTasks(t, state, TaskAutoPublishVOD) + require.Len(t, tasks, 1) + require.NoError(t, state.processAutoPublishVODTask(ctx, &tasks[0])) + require.Empty(t, pendingTasks(t, state, TaskFinalizeLivestreamVOD)) + require.Empty(t, pendingTasks(t, state, TaskAutoPublishVOD)) + }) +} diff --git a/pkg/statedb/livestream_vod.go b/pkg/statedb/livestream_vod.go new file mode 100644 index 000000000..c59d672a4 --- /dev/null +++ b/pkg/statedb/livestream_vod.go @@ -0,0 +1,80 @@ +package statedb + +import ( + "context" + "strings" + + "github.com/google/uuid" + "stream.place/streamplace/pkg/comatproto" + "stream.place/streamplace/pkg/model" + placestream "stream.place/streamplace/pkg/placestream" +) + +// LivestreamItem is one livestream record: the indexed row and its decoded +// record. +type LivestreamItem struct { + Livestream *model.Livestream + Record *placestream.Livestream +} + +// VideoDraftForLivestreams describes the place.stream.video record for the +// VOD of one or more livestream records: the first one's title (or the given +// one), description, tags and activity, connected to every livestream record +// the way the app's draft is connected to its one. Duration, source tracks +// and thumbnail are filled in at publish time from the finalized upload. +func VideoDraftForLivestreams(items []LivestreamItem, title, description string) *VideoDraft { + first := items[0].Record + title = strings.TrimSpace(title) + if title == "" { + title = strings.TrimSpace(first.Title) + } + if title == "" { + title = "Livestream" + } + v := &VideoDraft{ + Title: title, + Tags: first.Tags, + } + for _, it := range items { + v.Connections = append(v.Connections, placestream.Video_Connections_Elem{ + Video_Connection: &placestream.Video_Connection{ + LexiconTypeID: "place.stream.video#connection", + Ref: &comatproto.RepoStrongRef{Uri: it.Livestream.URI, Cid: it.Livestream.CID}, + }, + }) + } + if d := strings.TrimSpace(description); d != "" { + v.Description = &d + } + if first.Activity != nil { + switch { + case first.Activity.Defs_ActivityGame != nil: + v.Activity = &placestream.Video_Activity{Defs_ActivityGame: first.Activity.Defs_ActivityGame} + case first.Activity.Defs_ActivityLabel != nil: + v.Activity = &placestream.Video_Activity{Defs_ActivityLabel: first.Activity.Defs_ActivityLabel} + } + } + return v +} + +// CreateLivestreamUpload creates the synthetic Upload row a livestream VOD is +// finalized into, so clients follow it with the getUploadStatus / publishVideo +// flow they already have for resumable uploads. firstURI is the first +// livestream record of the recording. +func (state *StatefulDB) CreateLivestreamUpload(ctx context.Context, repoDID, firstURI string) (string, error) { + uu, err := uuid.NewV7() + if err != nil { + return "", err + } + uploadID := uu.String() + if err := state.CreateUpload(ctx, &Upload{ + ID: uploadID, + RepoDID: repoDID, + MimeType: "video/mp4", + Backend: "live", + Location: firstURI, + }); err != nil { + return "", err + } + return uploadID, nil +} diff --git a/pkg/statedb/livestream_vod_test.go b/pkg/statedb/livestream_vod_test.go new file mode 100644 index 000000000..6d74dfbcf --- /dev/null +++ b/pkg/statedb/livestream_vod_test.go @@ -0,0 +1,67 @@ +package statedb + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/model" + placestream "stream.place/streamplace/pkg/placestream" +) + +func TestVideoDraftForLivestreams(t *testing.T) { + a := LivestreamItem{ + Livestream: &model.Livestream{URI: "at://did:plc:x/place.stream.livestream/first", CID: "bafyfirst"}, + Record: &placestream.Livestream{Title: " State of the Union ", Tags: []string{"politics"}, + Activity: &placestream.Livestream_Activity{Defs_ActivityLabel: &placestream.Defs_ActivityLabel{Label: "Talk"}}}, + } + b := LivestreamItem{ + Livestream: &model.Livestream{URI: "at://did:plc:x/place.stream.livestream/second", CID: "bafysecond"}, + Record: &placestream.Livestream{Title: "State of the Union (continued)"}, + } + + v := VideoDraftForLivestreams([]LivestreamItem{a, b}, "", "") + require.Equal(t, "State of the Union", v.Title, "the first livestream's title by default") + require.Nil(t, v.Description) + require.Equal(t, []string{"politics"}, v.Tags) + require.NotNil(t, v.Activity.Defs_ActivityLabel) + require.Len(t, v.Connections, 2, "connected to every livestream it came from") + require.Equal(t, a.Livestream.URI, v.Connections[0].Video_Connection.Ref.Uri) + require.Equal(t, a.Livestream.CID, v.Connections[0].Video_Connection.Ref.Cid) + require.Equal(t, b.Livestream.URI, v.Connections[1].Video_Connection.Ref.Uri) + + v = VideoDraftForLivestreams([]LivestreamItem{a}, "Replay", " Full speech. ") + require.Equal(t, "Replay", v.Title) + require.Equal(t, "Full speech.", *v.Description) + + v = VideoDraftForLivestreams([]LivestreamItem{{Livestream: a.Livestream, Record: &placestream.Livestream{}}}, "", "") + require.Equal(t, "Livestream", v.Title, "a record with no title at all") +} + +// The draft rides in a task payload, so it has to survive JSON, and the +// record built from it has to be the one PublishVideo expects (a full Video +// with an empty source union does not marshal at all, which is why the +// payload carries a draft and not a record). +func TestVideoDraftPayloadRoundTrip(t *testing.T) { + a := LivestreamItem{ + Livestream: &model.Livestream{URI: "at://did:plc:x/place.stream.livestream/first", CID: "bafyfirst"}, + Record: &placestream.Livestream{Title: "Speech", Tags: []string{"politics"}, + Activity: &placestream.Livestream_Activity{Defs_ActivityLabel: &placestream.Defs_ActivityLabel{Label: "Talk"}}}, + } + task := FinalizeLivestreamVODTask{UploadID: "u", RepoDID: "did:plc:x", LivestreamURI: a.Livestream.URI, + Publish: VideoDraftForLivestreams([]LivestreamItem{a}, "", "the whole thing")} + b, err := json.Marshal(task) + require.NoError(t, err) + var back FinalizeLivestreamVODTask + require.NoError(t, json.Unmarshal(b, &back)) + require.NotNil(t, back.Publish) + rec := back.Publish.Record() + require.Equal(t, "Speech", rec.Title) + require.Equal(t, "the whole thing", *rec.Description) + require.Equal(t, []string{"politics"}, rec.Tags) + require.Equal(t, "Talk", rec.Activity.Defs_ActivityLabel.Label) + require.Len(t, rec.Connections, 1) + require.Equal(t, a.Livestream.URI, rec.Connections[0].Video_Connection.Ref.Uri) + require.Equal(t, a.Livestream.CID, rec.Connections[0].Video_Connection.Ref.Cid) + require.Equal(t, "place.stream.video", rec.RecordTypeID()) +} diff --git a/pkg/statedb/queue_processor.go b/pkg/statedb/queue_processor.go index 64fb7a7fc..03b873666 100644 --- a/pkg/statedb/queue_processor.go +++ b/pkg/statedb/queue_processor.go @@ -31,6 +31,7 @@ var TaskFinalizeLivestreamVOD = "finalize_livestream_vod" var TaskVODProcess = "vod_process" var TaskViewCountAggregate = "view_count_aggregate" var TaskCDNLogIngest = "cdn_log_ingest" +var TaskAutoPublishVOD = "auto_publish_vod" // nonVODTaskTypes is every task type handled by the general queue worker. // VOD processing runs on its own dedicated pool (see ProcessQueue) so a @@ -45,6 +46,7 @@ var nonVODTaskTypes = []string{ TaskFinalizeLivestream, TaskViewCountAggregate, TaskCDNLogIngest, + TaskAutoPublishVOD, } type NotificationTask struct { @@ -107,7 +109,8 @@ type FinalizeLivestreamVODTask struct { // Publish, when set, describes the place.stream.video record to // publish in the streamer's repo (with their stored session) as soon // as the VOD is finalized, instead of leaving a draft for them to - // publish from the app. Set by the operator's finalize route. + // publish from the app. Set by the operator's finalize route and by + // automatic VOD publishing (AutoPublishVODTask). Publish *VideoDraft `json:"publish,omitempty"` } @@ -248,6 +251,8 @@ func (state *StatefulDB) processTask(ctx context.Context, task *AppTask) error { return state.processViewCountAggregateTask(ctx, task) case TaskCDNLogIngest: return state.processCDNLogIngestTask(ctx, task) + case TaskAutoPublishVOD: + return state.processAutoPublishVODTask(ctx, task) default: return fmt.Errorf("unknown task type: %s", task.Type) } diff --git a/pkg/statedb/s3_segment.go b/pkg/statedb/s3_segment.go index fb4bf8cae..02af1346c 100644 --- a/pkg/statedb/s3_segment.go +++ b/pkg/statedb/s3_segment.go @@ -80,6 +80,16 @@ func (state *StatefulDB) ListS3SegmentsForLivestreams(ctx context.Context, lives return segs, nil } +// CountOpenS3Segments counts the objects recorded for one livestream whose +// uploads have not completed yet. +func (state *StatefulDB) CountOpenS3Segments(ctx context.Context, livestreamURI string) (int64, error) { + var n int64 + err := state.DB.WithContext(ctx).Model(&S3Segment{}). + Where("livestream_uri = ? AND completed_at IS NULL", livestreamURI). + Count(&n).Error + return n, err +} + // RecordComplete marks an S3 multipart upload as completed and records the // final part count and size. Implements s3.Recorder. func (state *StatefulDB) RecordComplete(ctx context.Context, id string, parts int32, size int64) error { diff --git a/pkg/statedb/statedb.go b/pkg/statedb/statedb.go index 2785bd175..4956d7d7e 100644 --- a/pkg/statedb/statedb.go +++ b/pkg/statedb/statedb.go @@ -86,6 +86,7 @@ var StatefulDBModels = []any{ Upload{}, DraftVideo{}, CDNLogPart{}, + UserPreferences{}, } var NoPostgresDatabaseCode = "3D000" diff --git a/pkg/statedb/user_preferences.go b/pkg/statedb/user_preferences.go new file mode 100644 index 000000000..99cf78558 --- /dev/null +++ b/pkg/statedb/user_preferences.go @@ -0,0 +1,56 @@ +package statedb + +import ( + "context" + "errors" + "time" + + "gorm.io/gorm" + "gorm.io/gorm/clause" + placestream "stream.place/streamplace/pkg/placestream" +) + +// UserPreferences are a user's private preferences on this node. Unlike the +// place.stream.server.settings record, they are not in the user's repo: the +// node keeps them here and serves them over place.stream.server.getPreferences +// and putPreferences. +type UserPreferences struct { + RepoDID string `gorm:"column:repo_did;primarykey"` + // AutoPublishVODs publishes the VOD of each of the user's recorded + // livestreams as soon as the livestream ends. + AutoPublishVODs bool `gorm:"column:auto_publish_vods;not null;default:false"` + CreatedAt time.Time `gorm:"column:created_at"` + UpdatedAt time.Time `gorm:"column:updated_at"` +} + +func (p *UserPreferences) TableName() string { + return "user_preferences" +} + +// GetUserPreferences returns repoDID's preferences, or the defaults if they +// have never set any. +func (state *StatefulDB) GetUserPreferences(ctx context.Context, repoDID string) (*UserPreferences, error) { + var prefs UserPreferences + err := state.DB.WithContext(ctx).Where("repo_did = ?", repoDID).First(&prefs).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + return &UserPreferences{RepoDID: repoDID}, nil + } + if err != nil { + return nil, err + } + return &prefs, nil +} + +// PutUserPreferences creates or replaces a user's preferences. +func (state *StatefulDB) PutUserPreferences(ctx context.Context, prefs *UserPreferences) error { + return state.DB.WithContext(ctx).Clauses(clause.OnConflict{ + Columns: []clause.Column{{Name: "repo_did"}}, + DoUpdates: clause.AssignmentColumns([]string{"auto_publish_vods", "updated_at"}), + }).Create(prefs).Error +} + +func (p *UserPreferences) ToLexicon() placestream.ServerDefs_Preferences { + return placestream.ServerDefs_Preferences{ + AutoPublishVods: p.AutoPublishVODs, + } +}