From 7b0ad91e24c3d85809e1e629fcd4e78e53d6ec9c Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Wed, 18 Feb 2026 15:46:03 -0800 Subject: [PATCH] playback: initial demo of authenticated playback --- Makefile | 4 +- go.mod | 2 +- go.sum | 4 +- .../components/mobile-player/use-webrtc.tsx | 63 +++++++++----- .../content/docs/lex-reference/openapi.json | 71 ++++++++++++++++ .../playback/place-stream-playback-whep.md | 82 +++++++++++++++++++ lexicons/place/stream/playback/whep.json | 36 ++++++++ pkg/api/api.go | 2 +- pkg/api/playback.go | 2 +- pkg/atproto/atproto.go | 6 +- pkg/atproto/migrate.go | 2 +- pkg/atproto/sync.go | 22 ++--- pkg/bus/segchanman.go | 1 + pkg/director/stream_session.go | 18 ++-- pkg/media/media.go | 33 +++++++- pkg/media/validate.go | 2 +- pkg/spxrpc/place_stream_playback.go | 34 ++++++++ pkg/spxrpc/spxrpc.go | 5 +- pkg/spxrpc/stubs.go | 18 ++++ pkg/streamplace/playbackwhep.go | 30 +++++++ 20 files changed, 385 insertions(+), 52 deletions(-) create mode 100644 js/docs/src/content/docs/lex-reference/playback/place-stream-playback-whep.md create mode 100644 lexicons/place/stream/playback/whep.json create mode 100644 pkg/spxrpc/place_stream_playback.go create mode 100644 pkg/streamplace/playbackwhep.go diff --git a/Makefile b/Makefile index d48cbe18..0c2d62e4 100644 --- a/Makefile +++ b/Makefile @@ -410,7 +410,7 @@ lexgen: .PHONY: lexgen-types lexgen-types: - go run github.com/bluesky-social/indigo/cmd/lexgen \ + go tool github.com/bluesky-social/indigo/cmd/lexgen \ -outdir ./pkg/spxrpc \ --build-file util/lexgen-types.json \ --external-lexicons subprojects/atproto/lexicons \ @@ -420,7 +420,7 @@ lexgen-types: .PHONY: lexgen-server lexgen-server: mkdir -p ./pkg/spxrpc \ - && go run github.com/bluesky-social/indigo/cmd/lexgen \ + && go tool github.com/bluesky-social/indigo/cmd/lexgen \ --gen-server \ --types-import place.stream:stream.place/streamplace/pkg/streamplace \ --types-import app.bsky:github.com/bluesky-social/indigo/api/bsky \ diff --git a/go.mod b/go.mod index 82ac35b1..75fca763 100644 --- a/go.mod +++ b/go.mod @@ -8,7 +8,7 @@ replace github.com/gocql/gocql => github.com/scylladb/gocql v1.14.4 replace github.com/AxisCommunications/go-dpop => github.com/streamplace/go-dpop v0.0.0-20250510031900-c897158a8ad4 -//replace github.com/livepeer/go-livepeer => ../go-livepeer +replace github.com/bluesky-social/indigo => github.com/streamplace/indigo v0.0.0-20260218231908-939cdaf0c507 tool github.com/bluesky-social/indigo/cmd/lexgen diff --git a/go.sum b/go.sum index a2f54da9..feeb902f 100644 --- a/go.sum +++ b/go.sum @@ -219,8 +219,6 @@ github.com/bluenviron/gortsplib/v5 v5.2.1 h1:VrFC5RU8npiyKiqLFKXZmdUFChSmbjO5eZB github.com/bluenviron/gortsplib/v5 v5.2.1/go.mod h1:sK4+00XQaSpU2iPIKjmhj6Yye+sVbNWEU2IJWYEZI9U= github.com/bluenviron/mediacommon/v2 v2.5.2 h1:eq7LHJFksDAVtVdTrwOUl7dO7LE8eKwLgYKYi5MmYaY= github.com/bluenviron/mediacommon/v2 v2.5.2/go.mod h1:5V15TiOfeaNVmZPVuOqAwqQSWyvMV86/dijDKu5q9Zs= -github.com/bluesky-social/indigo v0.0.0-20251206005924-d49b45419635 h1:kNeRrgGJH2g5OvjLqtaQ744YXqduliZYpFkJ/ld47c0= -github.com/bluesky-social/indigo v0.0.0-20251206005924-d49b45419635/go.mod h1:Pm2I1+iDXn/hLbF7XCg/DsZi6uDCiOo7hZGWprSM7k0= github.com/bmatcuk/doublestar v1.1.1/go.mod h1:UD6OnuiIn0yFxxA2le/rnRU1G4RaI4UvFv1sNto9p6w= github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 h1:DDGfHa7BWjL4YnC6+E63dPcxHo2sUxDIu8g3QgEJdRY= github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869/go.mod h1:Ekp36dRnpXw/yCqJaO+ZrUyxD+3VXMFFr56k5XYrpB4= @@ -1315,6 +1313,8 @@ github.com/streamplace/atproto-oauth-golang v0.0.0-20250619231223-a9c04fb888ac h github.com/streamplace/atproto-oauth-golang v0.0.0-20250619231223-a9c04fb888ac/go.mod h1:9LlKkqciiO5lRfbX0n4Wn5KNY9nvFb4R3by8FdW2TWc= github.com/streamplace/go-dpop v0.0.0-20250510031900-c897158a8ad4 h1:L1fS4HJSaAyNnkwfuZubgfeZy8rkWmA0cMtH5Z0HqNc= github.com/streamplace/go-dpop v0.0.0-20250510031900-c897158a8ad4/go.mod h1:bGUXY9Wd4mnd+XUrOYZr358J2f6z9QO/dLhL1SsiD+0= +github.com/streamplace/indigo v0.0.0-20260218231908-939cdaf0c507 h1:e8M3qPLr37NxEjlr18TaAwGP+OVyherVjgUG5VVmgWI= +github.com/streamplace/indigo v0.0.0-20260218231908-939cdaf0c507/go.mod h1:Pm2I1+iDXn/hLbF7XCg/DsZi6uDCiOo7hZGWprSM7k0= github.com/streamplace/oatproxy v0.0.0-20260130124113-420429019d3b h1:BB/R1egvkEqZhGeKL3tqAlTn0mkoOaaMY6r6s18XJYA= github.com/streamplace/oatproxy v0.0.0-20260130124113-420429019d3b/go.mod h1:pXi24hA7xBHj8eEywX6wGqJOR9FaEYlGwQ/72rN6okw= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= diff --git a/js/components/src/components/mobile-player/use-webrtc.tsx b/js/components/src/components/mobile-player/use-webrtc.tsx index 676d79dd..af7495ab 100644 --- a/js/components/src/components/mobile-player/use-webrtc.tsx +++ b/js/components/src/components/mobile-player/use-webrtc.tsx @@ -1,6 +1,7 @@ import { useEffect, useRef, useState } from "react"; import * as sdpTransform from "sdp-transform"; -import { PlayerStatus, usePlayerStore, useStreamKey } from "../.."; +import { StreamplaceAgent } from "streamplace"; +import { PlayerStatus, usePDSAgent, usePlayerStore, useStreamKey } from "../.."; import { RTCPeerConnection, RTCSessionDescription } from "./webrtc-primitives"; export default function useWebRTC( @@ -9,10 +10,14 @@ export default function useWebRTC( const [mediaStream, setMediaStream] = useState(null); const [stuck, setStuck] = useState(false); const setStatus = usePlayerStore((x) => x.setStatus); + let agent = usePDSAgent(); const lastChange = useRef(0); useEffect(() => { + if (!agent) { + return; + } const peerConnection = new RTCPeerConnection({ bundlePolicy: "max-bundle", }); @@ -44,7 +49,12 @@ export default function useWebRTC( } }); peerConnection.addEventListener("negotiationneeded", () => { - negotiateConnectionWithClientOffer(peerConnection, endpoint); + negotiateConnectionWithClientOffer( + peerConnection, + endpoint, + undefined, + agent, + ); }); let lastFramesReceived = 0; @@ -82,7 +92,7 @@ export default function useWebRTC( clearInterval(handle); peerConnection.close(); }; - }, [endpoint]); + }, [endpoint, agent]); return [mediaStream, stuck]; } @@ -102,6 +112,7 @@ export async function negotiateConnectionWithClientOffer( peerConnection: RTCPeerConnection, endpoint: string, bearerToken?: string, + agent?: StreamplaceAgent, ) { /** https://developer.mozilla.org/en-US/docs/Web/API/RTCPeerConnection/createOffer */ const offer = await peerConnection.createOffer({ @@ -134,23 +145,23 @@ export async function negotiateConnectionWithClientOffer( * This specifies how the client should communicate, * and what kind of media client and server have negotiated to exchange. */ - let response = await postSDPOffer(`${endpoint}`, ofr.sdp, bearerToken); - if (response.status === 201) { - let answerSDP = await response.text(); + let response = await postSDPOffer( + `${endpoint}`, + ofr.sdp, + bearerToken, + agent, + ); + let text = new TextDecoder().decode(response.data); + if (response.success) { if ((peerConnection.connectionState as string) === "closed") { return; } await peerConnection.setRemoteDescription( - new RTCSessionDescription({ type: "answer", sdp: answerSDP }), - ); - return response.headers.get("Location"); - } else if (response.status === 405) { - console.log( - "Remember to update the URL passed into the WHIP or WHEP client", + new RTCSessionDescription({ type: "answer", sdp: text }), ); + return "https://stream.place/example"; } else { - const errorMessage = await response.text(); - console.error(errorMessage); + console.error(text); } } catch (e) { console.error(`posting sdp offer failed: ${e}`); @@ -165,16 +176,26 @@ async function postSDPOffer( endpoint: string, data: string, bearerToken?: string, + agent?: StreamplaceAgent, ) { - return await fetch(endpoint, { - method: "POST", - mode: "cors", - headers: { - "content-type": "application/sdp", - ...(bearerToken ? { Authorization: `Bearer ${bearerToken}` } : {}), + if (!agent) { + throw new Error("No agent found"); + } + return await agent.place.stream.playback.whep(data, { + qp: { + rendition: "source", + streamer: agent.did!, }, - body: data, }); + // return await fetch(endpoint, { + // method: "POST", + // mode: "cors", + // headers: { + // "content-type": "application/sdp", + // ...(bearerToken ? { Authorization: `Bearer ${bearerToken}` } : {}), + // }, + // body: data, + // }); } /** diff --git a/js/docs/src/content/docs/lex-reference/openapi.json b/js/docs/src/content/docs/lex-reference/openapi.json index 7efe3598..44570b9f 100644 --- a/js/docs/src/content/docs/lex-reference/openapi.json +++ b/js/docs/src/content/docs/lex-reference/openapi.json @@ -517,6 +517,77 @@ } } }, + "/xrpc/place.stream.playback.whep": { + "post": { + "summary": "Play a stream over WebRTC using WHEP.", + "operationId": "place.stream.playback.whep", + "tags": ["place.stream.playback"], + "responses": { + "200": { + "description": "Success", + "content": { + "*/*": { + "schema": {} + } + } + }, + "400": { + "description": "Bad Request", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": ["error", "message"], + "properties": { + "error": { + "type": "string", + "oneOf": [ + { + "const": "Unauthorized" + } + ] + }, + "message": { + "type": "string" + } + } + } + } + } + } + }, + "parameters": [ + { + "name": "streamer", + "in": "query", + "required": true, + "description": "The DID of the streamer to play.", + "schema": { + "type": "string", + "description": "The DID of the streamer to play." + } + }, + { + "name": "rendition", + "in": "query", + "required": true, + "description": "The rendition of the stream to play.", + "schema": { + "type": "string", + "description": "The rendition of the stream to play." + } + } + ], + "requestBody": { + "required": true, + "content": { + "*/*": { + "schema": {} + } + } + } + } + }, "/xrpc/place.stream.multistream.createTarget": { "post": { "summary": "Create a new target for rebroadcasting a Streamplace stream.", diff --git a/js/docs/src/content/docs/lex-reference/playback/place-stream-playback-whep.md b/js/docs/src/content/docs/lex-reference/playback/place-stream-playback-whep.md new file mode 100644 index 00000000..519ac959 --- /dev/null +++ b/js/docs/src/content/docs/lex-reference/playback/place-stream-playback-whep.md @@ -0,0 +1,82 @@ +--- +title: place.stream.playback.whep +description: Reference for the place.stream.playback.whep lexicon +--- + +**Lexicon Version:** 1 + +## Definitions + + + +### `main` + +**Type:** `procedure` + +Play a stream over WebRTC using WHEP. + +**Parameters:** + +| Name | Type | Req'd | Description | Constraints | +| ----------- | -------- | ----- | ------------------------------------ | ----------- | +| `streamer` | `string` | ✅ | The DID of the streamer to play. | | +| `rendition` | `string` | ✅ | The rendition of the stream to play. | | + +**Input:** + +- **Encoding:** `*/*` +- **Schema:** + +_Schema not defined._ +**Output:** + +- **Encoding:** `*/*` +- **Schema:** + +_Schema not defined._ +**Possible Errors:** + +- `Unauthorized`: This user may not play this stream. + +--- + +## Lexicon Source + +```json +{ + "lexicon": 1, + "id": "place.stream.playback.whep", + "defs": { + "main": { + "type": "procedure", + "description": "Play a stream over WebRTC using WHEP.", + "parameters": { + "type": "params", + "required": ["streamer", "rendition"], + "properties": { + "streamer": { + "type": "string", + "description": "The DID of the streamer to play." + }, + "rendition": { + "type": "string", + "description": "The rendition of the stream to play." + } + } + }, + "input": { + "encoding": "*/*" + }, + "output": { + "encoding": "*/*" + }, + "errors": [ + { + "name": "Unauthorized", + "description": "This user may not play this stream." + } + ] + } + } +} +``` diff --git a/lexicons/place/stream/playback/whep.json b/lexicons/place/stream/playback/whep.json new file mode 100644 index 00000000..6415c8f8 --- /dev/null +++ b/lexicons/place/stream/playback/whep.json @@ -0,0 +1,36 @@ +{ + "lexicon": 1, + "id": "place.stream.playback.whep", + "defs": { + "main": { + "type": "procedure", + "description": "Play a stream over WebRTC using WHEP.", + "parameters": { + "type": "params", + "required": ["streamer", "rendition"], + "properties": { + "streamer": { + "type": "string", + "description": "The DID of the streamer to play." + }, + "rendition": { + "type": "string", + "description": "The rendition of the stream to play." + } + } + }, + "input": { + "encoding": "*/*" + }, + "output": { + "encoding": "*/*" + }, + "errors": [ + { + "name": "Unauthorized", + "description": "This user may not play this stream." + } + ] + } + } +} diff --git a/pkg/api/api.go b/pkg/api/api.go index 8c345030..ddf2d3ee 100644 --- a/pkg/api/api.go +++ b/pkg/api/api.go @@ -155,7 +155,7 @@ func (a *StreamplaceAPI) Handler(ctx context.Context) (http.Handler, error) { Recorder: metrics.NewRecorder(metrics.Config{}), }) var xrpc http.Handler - xrpc, err := spxrpc.NewServer(ctx, a.CLI, a.Model, a.StatefulDB, a.op, mdlw, a.ATSync, a.Bus, a.LocalDB) + xrpc, err := spxrpc.NewServer(ctx, a.CLI, a.Model, a.StatefulDB, a.op, mdlw, a.ATSync, a.Bus, a.LocalDB, a.MediaManager) if err != nil { return nil, err } diff --git a/pkg/api/playback.go b/pkg/api/playback.go index 95a30c10..f76959a4 100644 --- a/pkg/api/playback.go +++ b/pkg/api/playback.go @@ -28,7 +28,7 @@ func (a *StreamplaceAPI) NormalizeUser(ctx context.Context, user string) (string return user, nil } // only other allowed case is a bluesky handle - repo, err := a.ATSync.SyncBlueskyRepoCached(ctx, user, a.Model) + repo, err := a.ATSync.SyncBlueskyRepoCached(ctx, user) if err != nil { return "", err } diff --git a/pkg/atproto/atproto.go b/pkg/atproto/atproto.go index 7f31af9e..e4598d32 100644 --- a/pkg/atproto/atproto.go +++ b/pkg/atproto/atproto.go @@ -22,10 +22,10 @@ import ( var SyncGetRepo = comatproto.SyncGetRepo -func (atsync *ATProtoSynchronizer) SyncBlueskyRepoCached(ctx context.Context, handle string, mod model.Model) (*model.Repo, error) { +func (atsync *ATProtoSynchronizer) SyncBlueskyRepoCached(ctx context.Context, handle string) (*model.Repo, error) { ctx, span := otel.Tracer("signer").Start(ctx, "SyncBlueskyRepoCached") defer span.End() - repo, err := mod.GetRepoByHandleOrDID(handle) + repo, err := atsync.Model.GetRepoByHandleOrDID(handle) if err != nil { return nil, fmt.Errorf("failed to get repo for %s: %w", handle, err) } @@ -33,7 +33,7 @@ func (atsync *ATProtoSynchronizer) SyncBlueskyRepoCached(ctx context.Context, ha return repo, nil } - return atsync.SyncBlueskyRepo(ctx, handle, mod) + return atsync.SyncBlueskyRepo(ctx, handle, atsync.Model) } type mstNode struct { diff --git a/pkg/atproto/migrate.go b/pkg/atproto/migrate.go index 5c283df8..cff9a165 100644 --- a/pkg/atproto/migrate.go +++ b/pkg/atproto/migrate.go @@ -60,7 +60,7 @@ func (atsync *ATProtoSynchronizer) Migrate(ctx context.Context) error { currentDID := did g.Go(func() error { log.Debug(ctx, "syncing repo", "did", currentDID, "progress", currentIndex+1, "total", len(allDIDs)) - _, err := atsync.SyncBlueskyRepoCached(ctx, currentDID, atsync.Model) + _, err := atsync.SyncBlueskyRepoCached(ctx, currentDID) if err != nil { log.Error(ctx, "failed to sync repo", "did", currentDID, "err", err) syncErrorMu.Lock() diff --git a/pkg/atproto/sync.go b/pkg/atproto/sync.go index 465f7988..1ec2dabb 100644 --- a/pkg/atproto/sync.go +++ b/pkg/atproto/sync.go @@ -97,13 +97,13 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD } case *streamplace.ChatMessage: - repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID, atsync.Model) + repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID) if err != nil { return fmt.Errorf("failed to sync bluesky repo: %w", err) } go func() { - _, err = atsync.SyncBlueskyRepoCached(ctx, rec.Streamer, atsync.Model) + _, err = atsync.SyncBlueskyRepoCached(ctx, rec.Streamer) if err != nil { log.Error(ctx, "failed to sync bluesky repo", "err", err) } @@ -178,7 +178,7 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD } case *streamplace.ChatGate: - repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID, atsync.Model) + repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID) if err != nil { return fmt.Errorf("failed to sync bluesky repo: %w", err) } @@ -210,7 +210,7 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD go atsync.Bus.Publish(userDID, streamplaceGate) case *streamplace.ChatProfile: - repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID, atsync.Model) + repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID) if err != nil { return fmt.Errorf("failed to sync bluesky repo: %w", err) } @@ -225,7 +225,7 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD } case *streamplace.ServerSettings: - _, err := atsync.SyncBlueskyRepoCached(ctx, userDID, atsync.Model) + _, err := atsync.SyncBlueskyRepoCached(ctx, userDID) if err != nil { return fmt.Errorf("failed to sync bluesky repo: %w", err) } @@ -253,7 +253,7 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD } if livestream, ok := d["place.stream.livestream"]; ok { - repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID, atsync.Model) + repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID) if err != nil { return fmt.Errorf("failed to sync bluesky repo: %w", err) } @@ -292,7 +292,7 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD // log.Warn(ctx, "chat message detected", "uri", livestream.URI) // if this post is a reply to someone's livestream post // log.Warn(ctx, "chat message detected", "message", rec.Text) - repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID, atsync.Model) + repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID) if err != nil { return fmt.Errorf("failed to sync bluesky repo: %w", err) } @@ -483,11 +483,11 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD } case *streamplace.BroadcastOrigin: - repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID, atsync.Model) + repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID) if err != nil { return fmt.Errorf("failed to sync broadcast origin creator bluesky repo: %w", err) } - _, err = atsync.SyncBlueskyRepoCached(ctx, rec.Streamer, atsync.Model) + _, err = atsync.SyncBlueskyRepoCached(ctx, rec.Streamer) if err != nil { return fmt.Errorf("failed to sync broadcast origin streamer bluesky repo: %w", err) } @@ -508,7 +508,7 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD go atsync.Bus.Publish("", view) case *streamplace.MetadataConfiguration: - repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID, atsync.Model) + repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID) if err != nil { return fmt.Errorf("failed to sync bluesky repo: %w", err) } @@ -524,7 +524,7 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD } case *streamplace.ModerationPermission: - repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID, atsync.Model) + repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID) if err != nil { return fmt.Errorf("failed to sync bluesky repo: %w", err) } diff --git a/pkg/bus/segchanman.go b/pkg/bus/segchanman.go index efd47558..c820f027 100644 --- a/pkg/bus/segchanman.go +++ b/pkg/bus/segchanman.go @@ -16,6 +16,7 @@ type Seg struct { Filepath string Data []byte PacketizedData *PacketizedSegment + Published bool } type PacketizedSegment struct { diff --git a/pkg/director/stream_session.go b/pkg/director/stream_session.go index 431274da..d16abdae 100644 --- a/pkg/director/stream_session.go +++ b/pkg/director/stream_session.go @@ -208,11 +208,17 @@ func (ss *StreamSession) NewSegment(ctx context.Context, notif *media.NewSegment ss.bus.Publish(spseg.Creator, spseg) ss.Go(ctx, func() error { return ss.AddPlaybackSegment(ctx, spseg, "source", &bus.Seg{ - Filepath: notif.Segment.ID, - Data: notif.Data, + Filepath: notif.Segment.ID, + Data: notif.Data, + Published: notif.Metadata.Published, }) }) + // everything else is for published segments + if !notif.Metadata.Published { + return nil + } + if ss.cli.Thumbnail { ss.Go(ctx, func() error { return ss.Thumbnail(ctx, spseg.Creator, notif) @@ -721,9 +727,11 @@ func (ss *StreamSession) Transcode(ctx context.Context, spseg *streamplace.Segme } func (ss *StreamSession) AddPlaybackSegment(ctx context.Context, spseg *streamplace.Segment, rendition string, seg *bus.Seg) error { - ss.Go(ctx, func() error { - return ss.AddToHLS(ctx, spseg, rendition, seg.Data) - }) + if seg.Published { + ss.Go(ctx, func() error { + return ss.AddToHLS(ctx, spseg, rendition, seg.Data) + }) + } ss.Go(ctx, func() error { return ss.AddToWebRTC(ctx, spseg, rendition, seg) }) diff --git a/pkg/media/media.go b/pkg/media/media.go index 30e45ee8..82b428b0 100644 --- a/pkg/media/media.go +++ b/pkg/media/media.go @@ -197,24 +197,52 @@ type SegmentMetadata struct { DistributionPolicy *localdb.DistributionPolicy MetadataConfiguration *streamplace.MetadataConfiguration Livestream *streamplace.Livestream + Published bool } var ErrMissingMetadata = errors.New("missing segment metadata") var ErrInvalidMetadata = errors.New("invalid segment metadata") +var C2PAActionsV2Label = "c2pa.actions.v2" +var C2PAPublishedAction = "c2pa.published" func ParseSegmentAssertions(ctx context.Context, mani *c2patypes.Manifest) (*SegmentMetadata, error) { _, span := otel.Tracer("signer").Start(ctx, "ParseSegmentAssertions") defer span.End() var ass *c2patypes.ManifestAssertion + isPublished := false for _, a := range mani.Assertions { if a.Label == StreamplaceMetadata { ass = &a - break + continue } if a.Label == "place.stream.metadata" { // backwards compatibility for old manifests ass = &a - break + continue + } + if a.Label == C2PAActionsV2Label { + data, ok := a.Data.(map[string]any) + if !ok { + return nil, ErrInvalidMetadata + } + actions, ok := data["actions"].([]any) + if !ok { + return nil, ErrInvalidMetadata + } + for _, action := range actions { + actionMap, ok := action.(map[string]any) + if !ok { + return nil, ErrInvalidMetadata + } + actionType, ok := actionMap["action"].(string) + if !ok { + return nil, ErrInvalidMetadata + } + if actionType == C2PAPublishedAction { + isPublished = true + break + } + } } } if ass == nil { @@ -268,6 +296,7 @@ func ParseSegmentAssertions(ctx context.Context, mani *c2patypes.Manifest) (*Seg DistributionPolicy: distributionPolicy, MetadataConfiguration: metadataConfiguration, Livestream: livestream, + Published: isPublished, } return &out, nil } diff --git a/pkg/media/validate.go b/pkg/media/validate.go index 6d15447a..f28bae31 100644 --- a/pkg/media/validate.go +++ b/pkg/media/validate.go @@ -75,7 +75,7 @@ func (mm *MediaManager) ValidateMP4(ctx context.Context, input io.Reader, local signingKeyDID = meta.Creator repoDID = meta.Creator } else { - repo, err := mm.atsync.SyncBlueskyRepoCached(ctx, meta.Creator, mm.model) + repo, err := mm.atsync.SyncBlueskyRepoCached(ctx, meta.Creator) if err != nil { return err } diff --git a/pkg/spxrpc/place_stream_playback.go b/pkg/spxrpc/place_stream_playback.go new file mode 100644 index 00000000..787cbfa2 --- /dev/null +++ b/pkg/spxrpc/place_stream_playback.go @@ -0,0 +1,34 @@ +package spxrpc + +import ( + "bytes" + "context" + "io" + "net/http" + + "github.com/labstack/echo/v4" + "github.com/pion/webrtc/v4" +) + +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") + } + repo, err := s.ATSync.SyncBlueskyRepoCached(ctx, streamer) + if err != nil { + return nil, err + } + body, err := io.ReadAll(r) + if err != nil { + 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) + if err != nil { + return nil, echo.NewHTTPError(http.StatusInternalServerError, "error playing back", err) + } + return bytes.NewReader([]byte(answer.SDP)), nil +} diff --git a/pkg/spxrpc/spxrpc.go b/pkg/spxrpc/spxrpc.go index 68d0cff9..cf00df19 100644 --- a/pkg/spxrpc/spxrpc.go +++ b/pkg/spxrpc/spxrpc.go @@ -20,6 +20,7 @@ import ( "stream.place/streamplace/pkg/config" "stream.place/streamplace/pkg/localdb" "stream.place/streamplace/pkg/log" + "stream.place/streamplace/pkg/media" "stream.place/streamplace/pkg/model" "stream.place/streamplace/pkg/statedb" ) @@ -35,9 +36,10 @@ type Server struct { bus *bus.Bus op *oatproxy.OATProxy localDB localdb.LocalDB + mm *media.MediaManager } -func NewServer(ctx context.Context, cli *config.CLI, model model.Model, statefulDB *statedb.StatefulDB, op *oatproxy.OATProxy, mdlw middleware.Middleware, atsync *atproto.ATProtoSynchronizer, bus *bus.Bus, ldb localdb.LocalDB) (*Server, error) { +func NewServer(ctx context.Context, cli *config.CLI, model model.Model, statefulDB *statedb.StatefulDB, op *oatproxy.OATProxy, mdlw middleware.Middleware, atsync *atproto.ATProtoSynchronizer, bus *bus.Bus, ldb localdb.LocalDB, mm *media.MediaManager) (*Server, error) { e := echo.New() s := &Server{ e: e, @@ -50,6 +52,7 @@ func NewServer(ctx context.Context, cli *config.CLI, model model.Model, stateful bus: bus, op: op, localDB: ldb, + mm: mm, } e.Use(s.ErrorHandlingMiddleware()) e.Use(s.ContextPreservingMiddleware()) diff --git a/pkg/spxrpc/stubs.go b/pkg/spxrpc/stubs.go index 026f74c4..de19be2d 100644 --- a/pkg/spxrpc/stubs.go +++ b/pkg/spxrpc/stubs.go @@ -299,6 +299,7 @@ func (s *Server) RegisterHandlersPlaceStream(e *echo.Echo) error { e.POST("/xrpc/place.stream.multistream.deleteTarget", s.HandlePlaceStreamMultistreamDeleteTarget) e.GET("/xrpc/place.stream.multistream.listTargets", s.HandlePlaceStreamMultistreamListTargets) e.POST("/xrpc/place.stream.multistream.putTarget", s.HandlePlaceStreamMultistreamPutTarget) + e.POST("/xrpc/place.stream.playback.whep", s.HandlePlaceStreamPlaybackWhep) e.POST("/xrpc/place.stream.server.createWebhook", s.HandlePlaceStreamServerCreateWebhook) e.POST("/xrpc/place.stream.server.deleteWebhook", s.HandlePlaceStreamServerDeleteWebhook) e.GET("/xrpc/place.stream.server.getServerTime", s.HandlePlaceStreamServerGetServerTime) @@ -692,6 +693,23 @@ func (s *Server) HandlePlaceStreamMultistreamPutTarget(c echo.Context) error { return c.JSON(200, out) } +func (s *Server) HandlePlaceStreamPlaybackWhep(c echo.Context) error { + ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandlePlaceStreamPlaybackWhep") + defer span.End() + rendition := c.QueryParam("rendition") + streamer := c.QueryParam("streamer") + body := c.Request().Body + contentType := c.Request().Header.Get("Content-Type") + var out io.Reader + var handleErr error + // func (s *Server) handlePlaceStreamPlaybackWhep(ctx context.Context,rendition string,streamer string,r io.Reader,contentType string) (io.Reader, error) + out, handleErr = s.handlePlaceStreamPlaybackWhep(ctx, rendition, streamer, body, contentType) + if handleErr != nil { + return handleErr + } + return c.Stream(200, "application/octet-stream", out) +} + func (s *Server) HandlePlaceStreamServerCreateWebhook(c echo.Context) error { ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandlePlaceStreamServerCreateWebhook") defer span.End() diff --git a/pkg/streamplace/playbackwhep.go b/pkg/streamplace/playbackwhep.go new file mode 100644 index 00000000..86285df6 --- /dev/null +++ b/pkg/streamplace/playbackwhep.go @@ -0,0 +1,30 @@ +// Code generated by cmd/lexgen (see Makefile's lexgen); DO NOT EDIT. + +// Lexicon schema: place.stream.playback.whep + +package streamplace + +import ( + "bytes" + "context" + "io" + + lexutil "github.com/bluesky-social/indigo/lex/util" +) + +// PlaybackWhep calls the XRPC method "place.stream.playback.whep". +// +// rendition: The rendition of the stream to play. +// streamer: The DID of the streamer to play. +func PlaybackWhep(ctx context.Context, c lexutil.LexClient, input io.Reader, rendition string, streamer string) ([]byte, error) { + buf := new(bytes.Buffer) + + params := map[string]interface{}{} + params["rendition"] = rendition + params["streamer"] = streamer + if err := c.LexDo(ctx, lexutil.Procedure, "*/*", "place.stream.playback.whep", params, input, buf); err != nil { + return nil, err + } + + return buf.Bytes(), nil +} -- 2.51.2