diff --git a/js/app/components/live-dashboard/livestream-panel.tsx b/js/app/components/live-dashboard/livestream-panel.tsx index b945bdab6..e7e35587f 100644 --- a/js/app/components/live-dashboard/livestream-panel.tsx +++ b/js/app/components/live-dashboard/livestream-panel.tsx @@ -209,11 +209,12 @@ function LivestreamPanel({ scrollable = true }: { scrollable?: boolean }) { const userIsLive = useLiveUser(); const captureFrame = useCaptureVideoFrame(); const profile = useUserProfile(); - const livestream = useLivestream(); + const livestream = useLivestream(true); const createStreamRecord = useCreateStreamRecord(); const updateStreamRecord = useUpdateStreamRecord(); const endLivestream = useEndLivestream(); const url = useUrl(); + const [endingLivestream, setEndingLivestream] = useState(false); const [title, setTitle] = useState(""); const [loading, setLoading] = useState(false); @@ -352,9 +353,23 @@ function LivestreamPanel({ scrollable = true }: { scrollable?: boolean }) { const handleEndLivestream = useCallback(async () => { if (!livestream) return; - await endLivestream(livestream); + setEndingLivestream(true); + try { + await endLivestream(livestream); + } catch (error) { + console.error("Error ending livestream:", error); + toast.show("Error", "Failed to end livestream", { + duration: 3, + }); + } }, [livestream, endLivestream]); + useEffect(() => { + if (livestream && livestream.record.endedAt !== undefined) { + setEndingLivestream(false); + } + }, [livestream]); + const handleImageSelect = useCallback(() => { // Default web file picker behavior const input = document.createElement("input"); @@ -419,6 +434,9 @@ function LivestreamPanel({ scrollable = true }: { scrollable?: boolean }) { } : {}; + const canEndLivestream = + livestream && livestream.record.endedAt === undefined; + return ( <> @@ -668,20 +686,24 @@ function LivestreamPanel({ scrollable = true }: { scrollable?: boolean }) { diff --git a/js/app/components/live-dashboard/stream-monitor.tsx b/js/app/components/live-dashboard/stream-monitor.tsx index 3b977c4ff..301963ec4 100644 --- a/js/app/components/live-dashboard/stream-monitor.tsx +++ b/js/app/components/live-dashboard/stream-monitor.tsx @@ -33,7 +33,7 @@ export default function StreamMonitor({ const isUserLive = useLiveUser(); const profile = useLivestreamStore((x) => x.profile); const ingestConnectionState = usePlayerStore((x) => x.ingestConnectionState); - const ls = useLivestream(); + let ls = useLivestream(); const segmentTiming = useSegmentTiming(); // Use hook data primarily, fallback to props @@ -88,6 +88,39 @@ export default function StreamMonitor({ return "red"; } }; + + const getStreamStatus = () => { + if (!isLive) return "OFFLINE"; + if (!ls) return "NOT LIVE"; + return "LIVE"; + }; + + const getStreamTitle = () => { + if (!ls) { + return ( + + Stream not live yet. Press "Announce Livestream" to start! + + ); + } + return ( + + {ls?.record.title || "Stream Title"} + + ); + }; + return ( - - - {ls?.record.title || "Stream Title"} - - + {getStreamTitle()} - {isLive ? "LIVE" : "OFFLINE"} + {getStreamStatus()} diff --git a/js/components/src/livestream-store/livestream-store.tsx b/js/components/src/livestream-store/livestream-store.tsx index 304d1be81..eff4e30d3 100644 --- a/js/components/src/livestream-store/livestream-store.tsx +++ b/js/components/src/livestream-store/livestream-store.tsx @@ -62,7 +62,13 @@ export const useProfile = () => useLivestreamStore((x) => x.profile); export const useViewers = () => useLivestreamStore((x) => x.viewers); -export const useLivestream = () => useLivestreamStore((x) => x.livestream); +export const useLivestream = (includeEnded: boolean = false) => + useLivestreamStore((x) => { + const ls = x.livestream; + if (!ls) return null; + if (!includeEnded && ls.record.endedAt !== undefined) return null; + return ls; + }); export const useSegment = () => useLivestreamStore((x) => x.segment); diff --git a/pkg/api/playback.go b/pkg/api/playback.go index f76959a43..cdae6e22a 100644 --- a/pkg/api/playback.go +++ b/pkg/api/playback.go @@ -56,7 +56,7 @@ func (a *StreamplaceAPI) HandleWebRTCPlayback(ctx context.Context) httprouter.Ha offer := webrtc.SessionDescription{Type: webrtc.SDPTypeOffer, SDP: string(body)} var answer *webrtc.SessionDescription if a.CLI.NewWebRTCPlayback { - answer, err = a.MediaManager.WebRTCPlayback2(ctx, user, rendition, &offer) + answer, err = a.MediaManager.WebRTCPlayback2(ctx, user, rendition, &offer, "") } else { answer, err = a.MediaManager.WebRTCPlayback(ctx, user, rendition, &offer) } diff --git a/pkg/media/webrtc_playback2.go b/pkg/media/webrtc_playback2.go index 5ee005afb..edb74cf17 100644 --- a/pkg/media/webrtc_playback2.go +++ b/pkg/media/webrtc_playback2.go @@ -14,7 +14,7 @@ import ( ) // This function remains in scope for the duration of a single users' playback -func (mm *MediaManager) WebRTCPlayback2(ctx context.Context, user string, rendition string, offer *webrtc.SessionDescription) (*webrtc.SessionDescription, error) { +func (mm *MediaManager) WebRTCPlayback2(ctx context.Context, user string, rendition string, offer *webrtc.SessionDescription, viewer string) (*webrtc.SessionDescription, error) { uu, err := uuid.NewV7() if err != nil { return nil, err @@ -93,6 +93,10 @@ func (mm *MediaManager) WebRTCPlayback2(ctx context.Context, user string, rendit return case file := <-segChan.C: log.Debug(ctx, "got segment", "file", file.Filepath) + if !file.Published && viewer != user { + log.Warn(ctx, "segment is not published and viewer is not the user", "viewer", viewer, "user", user) + continue + } latency += file.PacketizedData.Duration packetQueue <- file.PacketizedData } diff --git a/pkg/media/webrtc_playback2_test.go b/pkg/media/webrtc_playback2_test.go index 3512db8a0..53d117566 100644 --- a/pkg/media/webrtc_playback2_test.go +++ b/pkg/media/webrtc_playback2_test.go @@ -17,7 +17,7 @@ func TestWebRTCPlayback2(t *testing.T) { Type: webrtc.SDPTypeOffer, SDP: firefoxNoH264SDP, } - answer, err := mm.WebRTCPlayback2(context.Background(), "test-user", "test-rendition", offer) + answer, err := mm.WebRTCPlayback2(context.Background(), "test-user", "test-rendition", offer, "") require.ErrorContains(t, err, "RTPSender created with no codecs") require.Nil(t, answer) } diff --git a/pkg/spxrpc/place_stream_playback.go b/pkg/spxrpc/place_stream_playback.go index e092558d3..9cca974ce 100644 --- a/pkg/spxrpc/place_stream_playback.go +++ b/pkg/spxrpc/place_stream_playback.go @@ -8,15 +8,22 @@ import ( "github.com/labstack/echo/v4" "github.com/pion/webrtc/v4" + "github.com/streamplace/oatproxy/pkg/oatproxy" ) func (s *Server) handlePlaceStreamPlaybackWhep(ctx context.Context, rendition string, streamer string, r io.Reader, _contentType string) (io.Reader, error) { + if streamer == "" { return nil, echo.NewHTTPError(http.StatusBadRequest, "streamer is required") } if rendition == "" { return nil, echo.NewHTTPError(http.StatusBadRequest, "rendition is required") } + viewer := "" + session, _ := oatproxy.GetOAuthSession(ctx) + if session != nil { + viewer = session.DID + } repo, err := s.ATSync.SyncBlueskyRepoCached(ctx, streamer) if err != nil { return nil, err @@ -26,7 +33,7 @@ func (s *Server) handlePlaceStreamPlaybackWhep(ctx context.Context, rendition st return nil, echo.NewHTTPError(http.StatusBadRequest, "error reading body", err) } offer := webrtc.SessionDescription{Type: webrtc.SDPTypeOffer, SDP: string(body)} - answer, err := s.mm.WebRTCPlayback2(ctx, repo.DID, rendition, &offer) + answer, err := s.mm.WebRTCPlayback2(ctx, repo.DID, rendition, &offer, viewer) if err != nil { return nil, echo.NewHTTPError(http.StatusInternalServerError, "error playing back", err) }