From d6aa4cf151fbf974cdc76308ed3ae9fb74e77215 Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Wed, 16 Sep 2026 03:29:58 -0700 Subject: [PATCH] live-to-VOD: the operator's finalize enqueues again The finalize task carried the whole place.stream.video record to publish, but that record's source union has no value until the finalizer has the MUXL CID, and an empty union refuses to marshal, so every /finalize-livestream call died at enqueue with "cannot marshal empty union Video_Source as JSON". The task now carries a VideoDraft (title, description, tags, activity, connections: what is known before the VOD exists) and the publisher builds the record from it at publish time, where PublishVideo fills in the source tracks, duration and thumbnail as before. A round-trip test marshals the task payload the way the queue does (cherry picked from commit 8b9a301c37ca17767d400b454ddc93e436d602ef) --- pkg/api/admin_livestream.go | 21 +++++++++--------- pkg/api/admin_livestream_test.go | 38 ++++++++++++++++++++++++++++---- pkg/cmd/streamplace.go | 2 +- pkg/statedb/queue_processor.go | 38 +++++++++++++++++++++++++++----- 4 files changed, 78 insertions(+), 21 deletions(-) diff --git a/pkg/api/admin_livestream.go b/pkg/api/admin_livestream.go index d5f813809..ee059a303 100644 --- a/pkg/api/admin_livestream.go +++ b/pkg/api/admin_livestream.go @@ -163,7 +163,7 @@ func (a *StreamplaceAPI) HandleFinalizeLivestream(ctx context.Context) httproute return } task := statedb.FinalizeLivestreamVODTask{UploadID: uploadID, RepoDID: repoDID, LivestreamURI: ordered[0], LivestreamURIs: ordered} - video := videoRecordForLivestreams(items, req.Title, req.Description) + video := videoDraftForLivestreams(items, req.Title, req.Description) if publish { task.Publish = video } @@ -204,12 +204,12 @@ func contains(list []string, v string) bool { return false } -// videoRecordForLivestreams is 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 videoRecordForLivestreams(items []livestreamItem, title, description string) *placestream.Video { +// 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 == "" { @@ -218,10 +218,9 @@ func videoRecordForLivestreams(items []livestreamItem, title, description string if title == "" { title = "Livestream" } - v := &placestream.Video{ - LexiconTypeID: "place.stream.video", - Title: title, - Tags: first.Tags, + v := &statedb.VideoDraft{ + Title: title, + Tags: first.Tags, } for _, it := range items { v.Connections = append(v.Connections, placestream.Video_Connections_Elem{ diff --git a/pkg/api/admin_livestream_test.go b/pkg/api/admin_livestream_test.go index ffa338d0b..d7b8e61ba 100644 --- a/pkg/api/admin_livestream_test.go +++ b/pkg/api/admin_livestream_test.go @@ -1,14 +1,16 @@ package api 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 TestVideoRecordForLivestreams(t *testing.T) { +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"}, @@ -19,7 +21,7 @@ func TestVideoRecordForLivestreams(t *testing.T) { rec: &placestream.Livestream{Title: "State of the Union (continued)"}, } - v := videoRecordForLivestreams([]livestreamItem{a, b}, "", "") + 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) @@ -29,10 +31,38 @@ func TestVideoRecordForLivestreams(t *testing.T) { 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 = videoRecordForLivestreams([]livestreamItem{a}, "Replay", " Full speech. ") + v = videoDraftForLivestreams([]livestreamItem{a}, "Replay", " Full speech. ") require.Equal(t, "Replay", v.Title) require.Equal(t, "Full speech.", *v.Description) - v = videoRecordForLivestreams([]livestreamItem{{ls: a.ls, rec: &placestream.Livestream{}}}, "", "") + 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()) +} diff --git a/pkg/cmd/streamplace.go b/pkg/cmd/streamplace.go index 24ab8be6c..d63cc5334 100644 --- a/pkg/cmd/streamplace.go +++ b/pkg/cmd/streamplace.go @@ -474,7 +474,7 @@ func runMain(ctx context.Context, build *config.BuildFlags, platformJobs []jobFu if vodStore == nil { return "", "", fmt.Errorf("finalize-livestream-vod: no VOD store configured") } - return vod.PublishVideo(ctx, state, vodStore, t.RepoDID, t.UploadID, t.Publish) + return vod.PublishVideo(ctx, state, vodStore, t.RepoDID, t.UploadID, t.Publish.Record()) }) // View-count aggregator runs the log → record pipeline for one // window. Same function-pointer pattern as the VOD processor so diff --git a/pkg/statedb/queue_processor.go b/pkg/statedb/queue_processor.go index 38a73878d..64fb7a7fc 100644 --- a/pkg/statedb/queue_processor.go +++ b/pkg/statedb/queue_processor.go @@ -104,11 +104,39 @@ type FinalizeLivestreamVODTask struct { // recordings make up the VOD, in order (LivestreamURI is the first); // empty means just LivestreamURI. LivestreamURIs []string `json:"livestreamURIs,omitempty"` - // Publish, when set, is 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 *placestream.Video `json:"publish,omitempty"` + // 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 *VideoDraft `json:"publish,omitempty"` +} + +// VideoDraft is the part of a place.stream.video record known before the +// VOD exists: what the operator (or the livestream record) says about it. +// It is not a placestream.Video because that record's source union has no +// value until the finalizer has the MUXL CID, and an empty union will not +// marshal, so a Video cannot sit in a task payload; the publisher builds +// the record from this with Record() and fills in the source, duration and +// thumbnail from the finished upload. +type VideoDraft struct { + Title string `json:"title"` + Description *string `json:"description,omitempty"` + Tags []string `json:"tags,omitempty"` + Activity *placestream.Video_Activity `json:"activity,omitempty"` + Connections []placestream.Video_Connections_Elem `json:"connections,omitempty"` +} + +// Record is the place.stream.video record for the draft, minus the fields +// that come from the finalized upload. +func (d *VideoDraft) Record() *placestream.Video { + return &placestream.Video{ + LexiconTypeID: "place.stream.video", + Title: d.Title, + Description: d.Description, + Tags: d.Tags, + Activity: d.Activity, + Connections: d.Connections, + } } // VideoPublisher publishes a finalized livestream VOD's place.stream.video -- knotmirror 0.0.1