diff --git a/pkg/api/api.go b/pkg/api/api.go index cbdaa86c..9309331d 100644 --- a/pkg/api/api.go +++ b/pkg/api/api.go @@ -418,12 +418,12 @@ func (a *StreamplaceAPI) NotFoundLinkingHandler(ctx context.Context, linker *lin return } lsv, err := ls.ToLivestreamView() - if err != nil || false { + if err != nil || lsv == nil { log.Error(ctx, "no livestream view found", "repoDID", repo.DID) defaultHandler.ServeHTTP(w, req) return } - bs, err := linker.GenerateStreamerCard(ctx, req.URL, &lsv, a.CLI.SentryDSN) + bs, err := linker.GenerateStreamerCard(ctx, req.URL, lsv, a.CLI.SentryDSN) if err != nil { log.Error(ctx, "error generating html", "error", err) defaultHandler.ServeHTTP(w, req) diff --git a/pkg/api/websocket.go b/pkg/api/websocket.go index 8f996a12..8c33b308 100644 --- a/pkg/api/websocket.go +++ b/pkg/api/websocket.go @@ -304,7 +304,7 @@ func (a *StreamplaceAPI) HandleWebsocket(ctx context.Context) httprouter.Handle if msgView.Author.Handle == "" || msgView.Author.Handle == "handle.invalid" { msgView.Author.Handle = a.ATSync.ResolveAuthorHandle(ctx, msgView.Author.Did) } - prv.Message = &msgView + prv.Message = msgView } if profile != nil { profileView, err := profile.ToStreamplaceChatProfile() diff --git a/pkg/atproto/lexicon_repo.go b/pkg/atproto/lexicon_repo.go index e1f86eec..32a94307 100644 --- a/pkg/atproto/lexicon_repo.go +++ b/pkg/atproto/lexicon_repo.go @@ -270,7 +270,7 @@ func MakeLexiconRepo(ctx context.Context, cli *config.CLI, mod model.Model, stat } cidLink := glex.Link(*newCid) - oldCid, _, err := LexiconRepo.GetRecord(ctx, rpath) + oldCid, _, err := LexiconRepo.GetRecordBytes(ctx, rpath) if errors.Is(err, mst.ErrNotFound) { _, err = LexiconRepo.PutRecord(ctx, rpath, sfw) if err != nil { diff --git a/pkg/atproto/lexicon_repo_queries.go b/pkg/atproto/lexicon_repo_queries.go index 410668c6..fa9c6258 100644 --- a/pkg/atproto/lexicon_repo_queries.go +++ b/pkg/atproto/lexicon_repo_queries.go @@ -111,7 +111,7 @@ func LexiconRepoGetRecord(ctx context.Context, repo string, collection string, r if err != nil { return nil, fmt.Errorf("handleComAtprotoRepoGetRecord: failed to open repo: %w", err) } - outCID, _, err := r.GetRecord(ctx, fmt.Sprintf("%s/%s", collection, rkey)) + outCID, _, err := r.GetRecordBytes(ctx, fmt.Sprintf("%s/%s", collection, rkey)) if err != nil { return nil, err } diff --git a/pkg/atproto/lexicon_repo_test.go b/pkg/atproto/lexicon_repo_test.go index 14c611e1..e7176f82 100644 --- a/pkg/atproto/lexicon_repo_test.go +++ b/pkg/atproto/lexicon_repo_test.go @@ -109,6 +109,6 @@ func TestLexiconRepo(t *testing.T) { require.NoError(t, err) newCommit, err := evts[1].ToCommitEvent() require.NoError(t, err) - require.Equal(t, newCommit.Since, &oldCommit.Rev) + require.Equal(t, newCommit.Since, oldCommit.Rev) require.Equal(t, newCommit.PrevData.String(), evts[0].SignedData) } diff --git a/pkg/atproto/server_repo.go b/pkg/atproto/server_repo.go index 830f9ac8..7ba16c5e 100644 --- a/pkg/atproto/server_repo.go +++ b/pkg/atproto/server_repo.go @@ -325,7 +325,7 @@ func CommitServerRepoRecord(ctx context.Context, cli *config.CLI, collection str rpath := fmt.Sprintf("%s/%s", collection, rkey) var recordCid cid.Cid var action string - _, _, err = r.GetRecord(ctx, rpath) + _, _, err = r.GetRecordBytes(ctx, rpath) if err != nil { // Record doesn't exist, create it recordCid, err = r.PutRecord(ctx, rpath, value) @@ -615,7 +615,7 @@ func ServerRepoGetRecord(ctx context.Context, repo string, collection string, rk if err != nil { return nil, fmt.Errorf("ServerRepoGetRecord: failed to open repo: %w", err) } - outCID, _, err := r.GetRecord(ctx, fmt.Sprintf("%s/%s", collection, rkey)) + outCID, _, err := r.GetRecordBytes(ctx, fmt.Sprintf("%s/%s", collection, rkey)) if err != nil { return nil, err } diff --git a/pkg/atproto/sync.go b/pkg/atproto/sync.go index 652fbe96..84996d55 100644 --- a/pkg/atproto/sync.go +++ b/pkg/atproto/sync.go @@ -167,7 +167,7 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD // Add mod badge if the author is a moderator issuerDID := fmt.Sprintf("did:web:%s", atsync.CLI.BroadcasterHost) - err = AddModBadgeIfApplicable(ctx, &scm, rec.Streamer, issuerDID, atsync.Model) + err = AddModBadgeIfApplicable(ctx, scm, rec.Streamer, issuerDID, atsync.Model) if err != nil { log.Error(ctx, "failed to add mod badge", "err", err) } @@ -181,7 +181,7 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD if !isUpdate && !isFirstSync { task := &statedb.ChatTask{ - MessageView: scm, + MessageView: *scm, } _, err = atsync.StatefulDB.EnqueueTask(ctx, statedb.TaskChat, task, statedb.WithTaskKey(fmt.Sprintf("chat-message::%s", aturi.String()))) @@ -293,7 +293,7 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD if err != nil { return fmt.Errorf("failed to convert chat message: %w", err) } - pinnedView.Message = &msgView + pinnedView.Message = msgView } if profile != nil { profileView, err := profile.ToStreamplaceChatProfile() diff --git a/pkg/badges/badges.go b/pkg/badges/badges.go index b33320ac..7e1813c7 100644 --- a/pkg/badges/badges.go +++ b/pkg/badges/badges.go @@ -53,7 +53,7 @@ func GetValidBadges(ctx context.Context, userDID, streamerDID, issuerDID string, } spChatProfile, err := chatProfile.ToStreamplaceChatProfile() - if err != nil || false { + if err != nil { return badges, nil } diff --git a/pkg/director/stream_session.go b/pkg/director/stream_session.go index d2fe186c..cf6605de 100644 --- a/pkg/director/stream_session.go +++ b/pkg/director/stream_session.go @@ -342,12 +342,12 @@ func (ss *StreamSession) NewSegment(ctx context.Context, notif *media.NewSegment if err != nil { return fmt.Errorf("failed to convert livestream to streamplace livestream: %w", err) } - if !shouldNotify(&lsv) { + if !shouldNotify(lsv) { log.Debug(ctx, "is not set to notify", "repoDID", spseg.Creator) return nil } task := &statedb.NotificationTask{ - Livestream: lsv, + Livestream: *lsv, PDSURL: r.PDS, } cp, err := ss.mod.GetChatProfile(ctx, spseg.Creator) diff --git a/pkg/model/chat_message.go b/pkg/model/chat_message.go index fa5885ba..8e4e03d8 100644 --- a/pkg/model/chat_message.go +++ b/pkg/model/chat_message.go @@ -38,10 +38,10 @@ func hashString(s string) int { return int(h.Sum32()) } -func (m *ChatMessage) ToStreamplaceMessageView() (placestream.ChatDefs_MessageView, error) { +func (m *ChatMessage) ToStreamplaceMessageView() (*placestream.ChatDefs_MessageView, error) { var msg placestream.ChatMessage if err := glex.DecodeCBOR(*m.ChatMessage, &msg); err != nil { - return placestream.ChatDefs_MessageView{}, fmt.Errorf("error decoding chat message: %w", err) + return nil, fmt.Errorf("error decoding chat message: %w", err) } // Truncate overlong message text if uniseg.GraphemeClusterCount(msg.Text) > 300 { @@ -69,7 +69,7 @@ func (m *ChatMessage) ToStreamplaceMessageView() (placestream.ChatDefs_MessageVi if m.ChatProfile != nil { scp, err := m.ChatProfile.ToStreamplaceChatProfile() if err != nil { - return placestream.ChatDefs_MessageView{}, fmt.Errorf("error converting chat profile to streamplace chat profile: %w", err) + return nil, fmt.Errorf("error converting chat profile to streamplace chat profile: %w", err) } message.ChatProfile = &scp } else { @@ -83,13 +83,13 @@ func (m *ChatMessage) ToStreamplaceMessageView() (placestream.ChatDefs_MessageVi if m.ReplyTo != nil { replyTo, err := m.ReplyTo.ToStreamplaceMessageView() if err != nil { - return placestream.ChatDefs_MessageView{}, fmt.Errorf("error converting reply to to streamplace message view: %w", err) + return nil, fmt.Errorf("error converting reply to to streamplace message view: %w", err) } message.ReplyTo = &placestream.ChatDefs_MessageView_ReplyTo{ - ChatDefs_MessageView: &replyTo, + ChatDefs_MessageView: replyTo, } } - return message, nil + return &message, nil } func (m *DBModel) CreateChatMessage(ctx context.Context, message *ChatMessage) error { @@ -160,7 +160,7 @@ func (m *DBModel) MostRecentChatMessages(repoDID string) ([]placestream.ChatDefs if err != nil { return nil, fmt.Errorf("error converting feed post to bsky post view: %w", err) } - spmessages = append(spmessages, spmessage) + spmessages = append(spmessages, *spmessage) } return spmessages, nil } diff --git a/pkg/model/livestream.go b/pkg/model/livestream.go index 9aab2d37..76bf1109 100644 --- a/pkg/model/livestream.go +++ b/pkg/model/livestream.go @@ -26,13 +26,13 @@ type Livestream struct { PostURI string `json:"postURI" gorm:"column:post_uri;index:idx_post_uri"` } -func (ls *Livestream) ToLivestreamView() (placestream.Livestream_LivestreamView, error) { +func (ls *Livestream) ToLivestreamView() (*placestream.Livestream_LivestreamView, error) { if ls == nil || ls.Livestream == nil { - return placestream.Livestream_LivestreamView{}, fmt.Errorf("livestream record is nil") + return nil, fmt.Errorf("livestream record is nil") } var rec placestream.Livestream if err := glex.DecodeCBOR(*ls.Livestream, &rec); err != nil { - return placestream.Livestream_LivestreamView{}, fmt.Errorf("error decoding livestream record: %w", err) + return nil, fmt.Errorf("error decoding livestream record: %w", err) } rec.Tags = moderation.FilterTags(rec.Tags) postView := placestream.Livestream_LivestreamView{ @@ -46,7 +46,7 @@ func (ls *Livestream) ToLivestreamView() (placestream.Livestream_LivestreamView, Record: &glex.LexiconTypeDecoder{Val: &rec}, IndexedAt: time.Now().Format(time.RFC3339), } - return postView, nil + return &postView, nil } func (m *DBModel) CreateLivestream(ctx context.Context, ls *Livestream) error { diff --git a/pkg/model/livestream_test.go b/pkg/model/livestream_test.go index a0b30afb..c7e2d014 100644 --- a/pkg/model/livestream_test.go +++ b/pkg/model/livestream_test.go @@ -15,7 +15,7 @@ func TestToLivestreamViewNil(t *testing.T) { if err == nil { t.Fatalf("expected error for nil receiver, got nil") } - if true { + if view != nil { t.Fatalf("expected nil view for nil receiver, got %+v", view) } }) @@ -26,7 +26,7 @@ func TestToLivestreamViewNil(t *testing.T) { if err == nil { t.Fatalf("expected error for nil livestream blob, got nil") } - if true { + if view != nil { t.Fatalf("expected nil view for nil livestream blob, got %+v", view) } }) diff --git a/pkg/model/media_view_count.go b/pkg/model/media_view_count.go index 831e86cc..590606d1 100644 --- a/pkg/model/media_view_count.go +++ b/pkg/model/media_view_count.go @@ -36,12 +36,12 @@ type MediaViewCount struct { } // ToRecord decodes the stored CBOR into the typed lexicon struct. -func (v *MediaViewCount) ToRecord() (placestream.MediaViewCount, error) { +func (v *MediaViewCount) ToRecord() (*placestream.MediaViewCount, error) { var vc placestream.MediaViewCount if err := glex.DecodeCBOR(v.Record, &vc); err != nil { - return placestream.MediaViewCount{}, fmt.Errorf("decode view-count record: %w", err) + return nil, fmt.Errorf("decode view-count record: %w", err) } - return vc, nil + return &vc, nil } func (m *DBModel) UpsertMediaViewCount(ctx context.Context, rec placestream.MediaViewCount, aturi syntax.ATURI) error { @@ -72,14 +72,14 @@ func (m *DBModel) DeleteMediaViewCount(ctx context.Context, uri string) error { return m.DB.WithContext(ctx).Where("uri = ?", uri).Delete(&MediaViewCount{}).Error } -func (m *DBModel) GetMediaViewCountByURI(ctx context.Context, uri string) (placestream.MediaViewCount, error) { +func (m *DBModel) GetMediaViewCountByURI(ctx context.Context, uri string) (*placestream.MediaViewCount, error) { var row MediaViewCount err := m.DB.WithContext(ctx).Where("uri = ?", uri).First(&row).Error if errors.Is(err, gorm.ErrRecordNotFound) { - return placestream.MediaViewCount{}, nil + return nil, nil } if err != nil { - return placestream.MediaViewCount{}, fmt.Errorf("get view count by uri: %w", err) + return nil, fmt.Errorf("get view count by uri: %w", err) } return row.ToRecord() } diff --git a/pkg/model/model.go b/pkg/model/model.go index d0e20cd5..ad23ff8e 100644 --- a/pkg/model/model.go +++ b/pkg/model/model.go @@ -161,7 +161,7 @@ type Model interface { UpsertMediaViewCount(ctx context.Context, rec placestream.MediaViewCount, aturi syntax.ATURI) error DeleteMediaViewCount(ctx context.Context, uri string) error - GetMediaViewCountByURI(ctx context.Context, uri string) (placestream.MediaViewCount, error) + GetMediaViewCountByURI(ctx context.Context, uri string) (*placestream.MediaViewCount, error) GetVideoView(ctx context.Context, uri string) (*placestream.MediaGetVideo_VideoView, error) GetVideoList(ctx context.Context, repoDID string, limit int, cursor string, hostedByServerDID string) (placestream.MediaGetVideoList_Output, error) diff --git a/pkg/spxrpc/og.go b/pkg/spxrpc/og.go index be05074a..0b2993b5 100644 --- a/pkg/spxrpc/og.go +++ b/pkg/spxrpc/og.go @@ -278,7 +278,7 @@ func (s *Server) generateOGImage(ctx context.Context, username string) ([]byte, streamplaceChatProfile, err := chatProfile.ToStreamplaceChatProfile() if err != nil { log.Warn(ctx, "failed to decode chat profile", "did", userDID, "error", err) - } else if true && streamplaceChatProfile.Color != nil { + } else if streamplaceChatProfile.Color != nil { userColor = color.RGBA{ R: uint8(streamplaceChatProfile.Color.Red), G: uint8(streamplaceChatProfile.Color.Green), diff --git a/pkg/spxrpc/place_stream_badge.go b/pkg/spxrpc/place_stream_badge.go index 69fd6ac1..42d9372f 100644 --- a/pkg/spxrpc/place_stream_badge.go +++ b/pkg/spxrpc/place_stream_badge.go @@ -60,7 +60,7 @@ func (s *Server) handlePlaceStreamBadgeGetIssuedBadges(ctx context.Context, stre } if chatProfile != nil { spcp, err := chatProfile.ToStreamplaceChatProfile() - if err == nil && true && spcp.Badges != nil { + if err == nil && spcp.Badges != nil { for _, sel := range spcp.Badges.Streamer { if sel.Badge.Uri != "" { selectedURIs[sel.Badge.Uri] = true diff --git a/pkg/spxrpc/place_stream_live.go b/pkg/spxrpc/place_stream_live.go index c21af6b9..2afa7f61 100644 --- a/pkg/spxrpc/place_stream_live.go +++ b/pkg/spxrpc/place_stream_live.go @@ -282,7 +282,7 @@ func (s *Server) getLiveUsersRanked(ctx context.Context, limit int, userDID stri LexiconTypeID: "place.stream.livestream#viewerCount", Count: int64(s.bus.GetViewerCount(stream.Author.Did)), } - streams[i] = stream + streams[i] = *stream } liveUsers := placestream.LiveGetLiveUsers_Output{Streams: streams} @@ -324,7 +324,7 @@ func (s *Server) getLiveUsersLatest(ctx context.Context, before string, limit in LexiconTypeID: "place.stream.livestream#viewerCount", Count: int64(s.bus.GetViewerCount(stream.Author.Did)), } - streams[i] = stream + streams[i] = *stream } liveUsers := placestream.LiveGetLiveUsers_Output{Streams: streams} diff --git a/pkg/spxrpc/place_stream_vod_drafts.go b/pkg/spxrpc/place_stream_vod_drafts.go index 62f1dfcd..09f93eb4 100644 --- a/pkg/spxrpc/place_stream_vod_drafts.go +++ b/pkg/spxrpc/place_stream_vod_drafts.go @@ -98,7 +98,7 @@ func (s *Server) handlePlaceStreamVodUpdateDraft(ctx context.Context, body *plac // Apply only the editable fields present in the partial input. The server- // authoritative fields (source, durationMs, status, error) are never touched // here — the closure only mutates editable metadata. - updated, err := s.statefulDB.UpdateDraftMetadata(ctx, body.Uri, func(rec placestream.VodDraftVideo) { + updated, err := s.statefulDB.UpdateDraftMetadata(ctx, body.Uri, func(rec *placestream.VodDraftVideo) { if body.Title != nil { rec.Title = *body.Title } @@ -177,7 +177,7 @@ func (s *Server) handlePlaceStreamVodCreateDraft(ctx context.Context, body *plac } else if banned { return nil, echo.NewHTTPError(http.StatusForbidden, "account is not permitted to publish videos") } - dv, err := s.statefulDB.CreateDraft(ctx, session.DID, "", placestream.VodDraftVideo{ + dv, err := s.statefulDB.CreateDraft(ctx, session.DID, "", &placestream.VodDraftVideo{ LexiconTypeID: "place.stream.vod.draftVideo", Title: "Untitled", Status: "processing", @@ -263,7 +263,7 @@ func (s *Server) createLivestreamDraft(ctx context.Context, did, uploadID string }, }} - return s.statefulDB.CreateDraft(ctx, did, uploadID, draftRec) + return s.statefulDB.CreateDraft(ctx, did, uploadID, &draftRec) } // livestreamActivityToDraft maps a livestream's activity union onto the draft's diff --git a/pkg/statedb/draft_video.go b/pkg/statedb/draft_video.go index 741a8254..d2ef46e3 100644 --- a/pkg/statedb/draft_video.go +++ b/pkg/statedb/draft_video.go @@ -65,7 +65,7 @@ func DraftURI(did, tid string) string { } // marshalDraft CBOR-encodes a draft record and computes its CID. -func marshalDraft(rec placestream.VodDraftVideo) (data []byte, cidStr string, err error) { +func marshalDraft(rec *placestream.VodDraftVideo) (data []byte, cidStr string, err error) { buf := bytes.NewBuffer(nil) if err := rec.MarshalCBOR(buf); err != nil { return nil, "", fmt.Errorf("marshal draft CBOR: %w", err) @@ -79,18 +79,18 @@ func marshalDraft(rec placestream.VodDraftVideo) (data []byte, cidStr string, er } // unmarshalDraft CBOR-decodes a draft record body. -func unmarshalDraft(data []byte) (placestream.VodDraftVideo, error) { +func unmarshalDraft(data []byte) (*placestream.VodDraftVideo, error) { var rec placestream.VodDraftVideo if err := rec.UnmarshalCBOR(bytes.NewReader(data)); err != nil { - return placestream.VodDraftVideo{}, fmt.Errorf("unmarshal draft CBOR: %w", err) + return nil, fmt.Errorf("unmarshal draft CBOR: %w", err) } - return rec, nil + return &rec, nil } // CreateDraft stores a new draft record. The record's CID and the row's URI are // derived from the record; originUploadID ties it to its processing job (may be // empty for drafts created outside an upload/finalize flow). -func (state *StatefulDB) CreateDraft(ctx context.Context, did, originUploadID string, rec placestream.VodDraftVideo) (*DraftVideo, error) { +func (state *StatefulDB) CreateDraft(ctx context.Context, did, originUploadID string, rec *placestream.VodDraftVideo) (*DraftVideo, error) { data, cidStr, err := marshalDraft(rec) if err != nil { return nil, err @@ -182,7 +182,7 @@ func (state *StatefulDB) ListDrafts(ctx context.Context, did string, limit int, // UpdateDraftMetadata applies a partial update of editable fields only. It never // touches source/durationMs/status/error — those are server-authoritative. The // CID is recomputed from the new CBOR body. Returns the updated row. -func (state *StatefulDB) UpdateDraftMetadata(ctx context.Context, uri string, apply func(rec placestream.VodDraftVideo)) (*DraftVideo, error) { +func (state *StatefulDB) UpdateDraftMetadata(ctx context.Context, uri string, apply func(rec *placestream.VodDraftVideo)) (*DraftVideo, error) { var dv DraftVideo err := state.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error { // FOR UPDATE (Postgres) so a concurrent SetDraftReady/SetDraftError @@ -224,10 +224,10 @@ func (state *StatefulDB) UpdateDraftMetadata(ctx context.Context, uri string, ap // fields (source, durationMs) in the CBOR body, plus content_cid on the SQL row. // sourceTracks is the JSON string of {uri,cid} track refs from the Upload row. // The caller is expected to have built the source union member from that JSON. -func (state *StatefulDB) SetDraftReady(ctx context.Context, originUploadID string, source placestream.VodDraftVideo_Source, durationMs int64, contentCID string) error { - return state.updateDraftByUpload(ctx, originUploadID, func(rec placestream.VodDraftVideo) { +func (state *StatefulDB) SetDraftReady(ctx context.Context, originUploadID string, source *placestream.VodDraftVideo_Source, durationMs int64, contentCID string) error { + return state.updateDraftByUpload(ctx, originUploadID, func(rec *placestream.VodDraftVideo) { rec.Status = "ready" - rec.Source = &source + rec.Source = source rec.DurationMs = &durationMs rec.Error = nil }, func(dv *DraftVideo) { @@ -265,22 +265,22 @@ func (state *StatefulDB) markDraftReadyFromUpload(ctx context.Context, uploadID // the {uri,cid} JSON array the Upload row stores — the same conversion // vod.sourceTracksFromUpload performs for the publishVideo path, replicated // here because pkg/statedb can't import pkg/vod (cycle). -func draftSourceFromTrackURIs(trackURIsJSON string) (placestream.VodDraftVideo_Source, error) { +func draftSourceFromTrackURIs(trackURIsJSON string) (*placestream.VodDraftVideo_Source, error) { if trackURIsJSON == "" { - return placestream.VodDraftVideo_Source{}, nil + return nil, nil } var refs []struct { URI string `json:"uri"` CID string `json:"cid"` } if err := json.Unmarshal([]byte(trackURIsJSON), &refs); err != nil { - return placestream.VodDraftVideo_Source{}, fmt.Errorf("decode track refs: %w", err) + return nil, fmt.Errorf("decode track refs: %w", err) } tracks := make([]comatproto.RepoStrongRef, 0, len(refs)) for _, r := range refs { tracks = append(tracks, comatproto.RepoStrongRef{Uri: r.URI, Cid: r.CID}) } - return placestream.VodDraftVideo_Source{ + return &placestream.VodDraftVideo_Source{ MediaDefs_SourceTracks: &placestream.MediaDefs_SourceTracks{ LexiconTypeID: "place.stream.media.defs#sourceTracks", Tracks: tracks, @@ -290,7 +290,7 @@ func draftSourceFromTrackURIs(trackURIsJSON string) (placestream.VodDraftVideo_S // SetDraftError flips a draft to status "error" with an error message. func (state *StatefulDB) SetDraftError(ctx context.Context, originUploadID, errMsg string) error { - return state.updateDraftByUpload(ctx, originUploadID, func(rec placestream.VodDraftVideo) { + return state.updateDraftByUpload(ctx, originUploadID, func(rec *placestream.VodDraftVideo) { rec.Status = "error" rec.Error = &errMsg }, nil) @@ -298,7 +298,7 @@ func (state *StatefulDB) SetDraftError(ctx context.Context, originUploadID, errM // updateDraftByUpload rewrites the CBOR body (via apply) and, optionally, the // denormalized SQL columns (via applyRow) for the draft tied to originUploadID. -func (state *StatefulDB) updateDraftByUpload(ctx context.Context, originUploadID string, apply func(rec placestream.VodDraftVideo), applyRow func(dv *DraftVideo)) error { +func (state *StatefulDB) updateDraftByUpload(ctx context.Context, originUploadID string, apply func(rec *placestream.VodDraftVideo), applyRow func(dv *DraftVideo)) error { return state.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error { // FOR UPDATE (Postgres) — see UpdateDraftMetadata. A concurrent user // metadata edit blocks until this ready/error transition commits, so it @@ -359,6 +359,6 @@ func (dv *DraftVideo) ToDraftView() (placestream.VodDraftDefs_DraftView, error) return placestream.VodDraftDefs_DraftView{ Uri: dv.URI, Cid: dv.CID, - Record: &glex.LexiconTypeDecoder{Val: &rec}, + Record: &glex.LexiconTypeDecoder{Val: rec}, }, nil } diff --git a/pkg/statedb/draft_video_test.go b/pkg/statedb/draft_video_test.go index b2590a0c..68a377f1 100644 --- a/pkg/statedb/draft_video_test.go +++ b/pkg/statedb/draft_video_test.go @@ -8,8 +8,8 @@ import ( "stream.place/streamplace/pkg/placestream" ) -func newDraftRec(title, status string) placestream.VodDraftVideo { - return placestream.VodDraftVideo{ +func newDraftRec(title, status string) *placestream.VodDraftVideo { + return &placestream.VodDraftVideo{ LexiconTypeID: "place.stream.vod.draftVideo", Title: title, Status: status, @@ -71,7 +71,7 @@ func TestDraftVideoUpdateMetadataRecomputesCID(t *testing.T) { origCID := dv.CID // Partial update of an editable field. - updated, err := state.UpdateDraftMetadata(ctx, dv.URI, func(rec placestream.VodDraftVideo) { + updated, err := state.UpdateDraftMetadata(ctx, dv.URI, func(rec *placestream.VodDraftVideo) { rec.Title = "Edited Title" }) require.NoError(t, err) @@ -125,7 +125,7 @@ func TestSetDraftReadyAndError(t *testing.T) { dv, err := state.CreateDraft(ctx, did, "up-ready", newDraftRec("Ready me", "processing")) require.NoError(t, err) - sourceTracks := placestream.VodDraftVideo_Source{ + sourceTracks := &placestream.VodDraftVideo_Source{ MediaDefs_SourceTracks: &placestream.MediaDefs_SourceTracks{ LexiconTypeID: "place.stream.media.defs#sourceTracks", Tracks: nil, @@ -163,7 +163,7 @@ func TestSetDraftReadyAndError(t *testing.T) { func TestSetDraftOnMissingUploadIsNoOp(t *testing.T) { WithAllDatabases(t, func(state *StatefulDB) { // A pre-drafts-era upload (no draft row) must not error. - err := state.SetDraftReady(t.Context(), "no-such-upload", placestream.VodDraftVideo_Source{}, 0, "") + err := state.SetDraftReady(t.Context(), "no-such-upload", nil, 0, "") require.NoError(t, err) err = state.SetDraftError(t.Context(), "no-such-upload", "x") require.NoError(t, err) diff --git a/pkg/statedb/queue_processor_draft_test.go b/pkg/statedb/queue_processor_draft_test.go index 378b8715..21f4201c 100644 --- a/pkg/statedb/queue_processor_draft_test.go +++ b/pkg/statedb/queue_processor_draft_test.go @@ -26,7 +26,7 @@ func TestDraftLifecycleThroughVODProcessor(t *testing.T) { require.NoError(t, state.CreateUpload(ctx, &Upload{ ID: "up-lifecycle", RepoDID: did, MimeType: "video/mp4", Backend: "file", })) - _, err := state.CreateDraft(ctx, did, "up-lifecycle", placestream.VodDraftVideo{ + _, err := state.CreateDraft(ctx, did, "up-lifecycle", &placestream.VodDraftVideo{ LexiconTypeID: "place.stream.vod.draftVideo", Title: "from upload", Status: "processing", @@ -84,7 +84,7 @@ func TestDraftLifecycleErrorFlipsDraft(t *testing.T) { ctx := context.Background() require.NoError(t, state.CreateUpload(ctx, &Upload{ID: "up-fail", RepoDID: "did:plc:err", Backend: "file"})) - _, err = state.CreateDraft(ctx, "did:plc:err", "up-fail", placestream.VodDraftVideo{ + _, err = state.CreateDraft(ctx, "did:plc:err", "up-fail", &placestream.VodDraftVideo{ LexiconTypeID: "place.stream.vod.draftVideo", Title: "will fail", Status: "processing", CreatedAt: "2026-01-01T00:00:00Z", }) diff --git a/pkg/upload/upload.go b/pkg/upload/upload.go index 799449d3..a5297395 100644 --- a/pkg/upload/upload.go +++ b/pkg/upload/upload.go @@ -336,7 +336,7 @@ func (m *Manager) onComplete(ctx context.Context, ev tushandler.HookEvent) { // supply a draftUri. A failed create is non-fatal: the upload still // processes. if existing, _ := m.state.GetDraftByUpload(ctx, row.ID); existing == nil { - if _, err := m.state.CreateDraft(ctx, row.RepoDID, row.ID, placestream.VodDraftVideo{ + if _, err := m.state.CreateDraft(ctx, row.RepoDID, row.ID, &placestream.VodDraftVideo{ LexiconTypeID: "place.stream.vod.draftVideo", Title: filenameOrDefault(row.Filename), Status: "processing", diff --git a/pkg/vod/publish_draft_test.go b/pkg/vod/publish_draft_test.go index 56fb821e..b58a88cd 100644 --- a/pkg/vod/publish_draft_test.go +++ b/pkg/vod/publish_draft_test.go @@ -41,7 +41,7 @@ func TestPublishDraftNotFoundForOtherUsersDraft(t *testing.T) { ctx := context.Background() // Alice owns the draft; Bob tries to publish it. - dv, err := state.CreateDraft(ctx, "did:plc:alice", "up-1", placestream.VodDraftVideo{ + dv, err := state.CreateDraft(ctx, "did:plc:alice", "up-1", &placestream.VodDraftVideo{ LexiconTypeID: "place.stream.vod.draftVideo", Title: "Alice's draft", Status: "ready", @@ -59,7 +59,7 @@ func TestPublishDraftNotReadyWhileProcessing(t *testing.T) { require.NoError(t, err) ctx := context.Background() - dv, err := state.CreateDraft(ctx, "did:plc:alice", "up-2", placestream.VodDraftVideo{ + dv, err := state.CreateDraft(ctx, "did:plc:alice", "up-2", &placestream.VodDraftVideo{ LexiconTypeID: "place.stream.vod.draftVideo", Title: "Still cooking", Status: "processing", @@ -77,7 +77,7 @@ func TestPublishDraftNotReadyWhenErrored(t *testing.T) { require.NoError(t, err) ctx := context.Background() - dv, err := state.CreateDraft(ctx, "did:plc:alice", "up-3", placestream.VodDraftVideo{ + dv, err := state.CreateDraft(ctx, "did:plc:alice", "up-3", &placestream.VodDraftVideo{ LexiconTypeID: "place.stream.vod.draftVideo", Title: "Failed", Status: "error",