Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
16 kB · 546 lines
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547package media
import ( "bytes" "context" "crypto" "encoding/json" "errors" "fmt" "io" "sync" "sync/atomic" "time"
"github.com/google/uuid" "github.com/pion/webrtc/v4" "go.opentelemetry.io/otel" "stream.place/streamplace/pkg/aqtime" "stream.place/streamplace/pkg/atproto" "stream.place/streamplace/pkg/bus" c2patypes "stream.place/streamplace/pkg/c2patypes" "stream.place/streamplace/pkg/config" "stream.place/streamplace/pkg/gstinit" "stream.place/streamplace/pkg/livehls" "stream.place/streamplace/pkg/localdb" "stream.place/streamplace/pkg/model" "stream.place/streamplace/pkg/placestream"
"stream.place/streamplace/pkg/log"
"github.com/piprate/json-gold/ld"
irohStreamplace "stream.place/streamplace/pkg/iroh/generated/iroh_streamplace"
_ "stream.place/streamplace/pkg/streamplacedeps")
const CertFile = "cert.pem"const SegmentsDir = "segments"
const StreamplaceMetadata = "cawg.metadata"
type MediaManager struct { cli *config.CLI liveWindows map[string]*livehls.Writer liveWindowsMut sync.Mutex // liveWindowPublished is, per streamer, whether the latest segment fed // into the window was published; guarded by liveWindowsMut. liveWindowPublished map[string]bool // modBuffers holds a short in-memory ring of each live user's most recent // canonical segments, the source for moderation/report clips now that // segments are no longer archived to disk. Keyed by repoDID. See // moderation_buffer.go. modBuffers map[string]*modBuffer modBuffersMut sync.Mutex httpPipes map[string]io.Writer httpPipesMutex sync.Mutex newSegmentSubs []*segmentSubscriber // Recent replicated source segments' Opus tracks, for pairing with the // rendition addenda that follow them (see PublishRenditionsForPlayback). sourceAudios map[string][]sourceAudio sourceAudioMu sync.Mutex newSegmentSubsMutex sync.RWMutex model model.Model bus *bus.Bus atsync *atproto.ATProtoSynchronizer webrtcAPI *webrtc.API webrtcConfig webrtc.Configuration localDB localdb.LocalDB
// hlsSessions tracks live HLS playback sessions by (streamer, sid) so // stateless HLS requests feed the viewer count. See hls_sessions.go. hlsSessions *hlsSessionTracker
// Node S2PA transcode signer (cert + PKCS#8 key PEM), built once from the // server-repo key. Used to sign transcode-completed audio tracks under the // node's own did:web identity, signed in-wasm (the node key is software). // See transcode.go. nodeSignerOnce sync.Once nodeCert []byte nodeKeyPEM []byte nodeSignerErr error
// Per-stream continuous audio transcoders, keyed by repoDID. A single-codec // stream's segments are fed here in order; each completed dual-codec segment // is distributed asynchronously (~1 GoP later). See transcode_stream.go. transcoders map[string]*streamTranscoder // renditionVideoTID remembers, per rendition name, the track id a // transcoder's MP4s carry their video under (see canonicalRenditionTrack). renditionVideoTID renditionTIDCache transcodersMu sync.Mutex
// Monotonic ingest-session epoch. Each live ingest session (one // SegmentAndSignElem) claims a fresh value, stamped onto its context, so the // per-DID transcoder rebuilds when a streamer reconnects rather than feeding // the restarted media timeline into the previous session's continuous encoder. // See withIngestSession / feedStreamTranscoder. ingestSessionSeq atomic.Uint64}
// nextIngestSession claims a fresh monotonic ingest-session epoch for a new live// session. Epochs are strictly increasing, so a newer session always wins over a// transcoder built for an older one.func (mm *MediaManager) nextIngestSession() uint64 { return mm.ingestSessionSeq.Add(1)}
type NewSegmentNotification struct { Segment *localdb.Segment // Data is the presentation flat MP4 (ftyp+moov+mdat envelope) consumed by // the GStreamer pipelines (WebRTC packetize, thumbnail). Data []byte // Muxl is the bare canonical .m4s: blindly concatenatable signed segments // with no container header. Consumers synthesize whatever wrapper they // need. The long-term wire format; Data retires once all consumers are MUXL. Muxl []byte Metadata *SegmentMetadata Local bool}
func RunSelfTest(ctx context.Context) error { gstinit.InitGST() return SelfTest(ctx)}
func MakeMediaManager(ctx context.Context, cli *config.CLI, signer crypto.Signer, mod model.Model, bus *bus.Bus, atsync *atproto.ATProtoSynchronizer, ldb localdb.LocalDB) (*MediaManager, error) { gstinit.InitGST() err := SelfTest(ctx) if err != nil { return nil, fmt.Errorf("error in gstreamer self-test: %w", err) }
api, config, err := newWebRTCAPI() if err != nil { return nil, err } mm := &MediaManager{ cli: cli, liveWindows: map[string]*livehls.Writer{}, liveWindowPublished: map[string]bool{}, modBuffers: map[string]*modBuffer{}, httpPipes: map[string]io.Writer{}, model: mod, bus: bus, atsync: atsync, webrtcAPI: api, webrtcConfig: config, localDB: ldb, transcoders: map[string]*streamTranscoder{}, } mm.hlsSessions = newHLSSessionTracker(hlsSessionTTL, func(streamer string) { mm.IncrementViewerCount(streamer, "hls") }, func(streamer string) { mm.DecrementViewerCount(streamer, "hls") }, ) return mm, nil}
func (mm *MediaManager) HandleData(node *irohStreamplace.PublicKey, data []byte) { r := bytes.NewReader(data) ctx := context.Background() err := mm.ValidateMP4(ctx, r, true) if err != nil { log.Log(ctx, "invalid incoming segment", "error", err) }}
// replacement for os.Pipe that works on windowsfunc (mm *MediaManager) HTTPPipe() (string, io.ReadCloser, func(), error) { uu, err := uuid.NewV7() if err != nil { return "", nil, nil, err } mm.httpPipesMutex.Lock() defer mm.httpPipesMutex.Unlock() u := fmt.Sprintf("%s/http-pipe/%s", mm.cli.OwnInternalURL(), uu.String()) done := func() { mm.httpPipesMutex.Lock() defer mm.httpPipesMutex.Unlock() delete(mm.httpPipes, uu.String()) } r, w := io.Pipe() mm.httpPipes[uu.String()] = w return u, r, done, nil}
func (mm *MediaManager) GetHTTPPipeWriter(uu string) io.Writer { mm.httpPipesMutex.Lock() defer mm.httpPipesMutex.Unlock() return mm.httpPipes[uu]}
// segmentQueueSize is how many validated segments may wait for one// subscriber before further ones are dropped for it.const segmentQueueSize = 256
// A segmentSubscriber receives every validated segment, in validation// order: notifications queue per subscriber and one goroutine forwards// them. (A goroutine per notification, as before, handed a burst of tiny// segments to the director in whatever order the scheduler ran them.)type segmentSubscriber struct { ch chan *NewSegmentNotification queue chan *NewSegmentNotification}
func (s *segmentSubscriber) forward() { for not := range s.queue { select { case s.ch <- not: case <-time.After(time.Minute): log.Warn(context.Background(), "segment subscriber did not take a segment within a minute, dropping it", "streamer", not.Segment.RepoDID, "segmentID", not.Segment.ID) } }}
// register a handler for all new segments that come infunc (mm *MediaManager) NewSegment() <-chan *NewSegmentNotification { sub := &segmentSubscriber{ ch: make(chan *NewSegmentNotification), queue: make(chan *NewSegmentNotification, segmentQueueSize), } go sub.forward() mm.newSegmentSubsMutex.Lock() defer mm.newSegmentSubsMutex.Unlock() mm.newSegmentSubs = append(mm.newSegmentSubs, sub) return sub.ch}
// notifySubscribers hands a validated segment to every subscriber, in// order, without waiting on any of them.func (mm *MediaManager) notifySubscribers(ctx context.Context, not *NewSegmentNotification) { mm.newSegmentSubsMutex.RLock() defer mm.newSegmentSubsMutex.RUnlock() for _, sub := range mm.newSegmentSubs { select { case sub.queue <- not: default: log.Error(ctx, "segment subscriber is not keeping up, dropping a segment", "streamer", not.Segment.RepoDID, "segmentID", not.Segment.ID, "behind", len(sub.queue)) } }}
type obj map[string]any
type StringVal struct { Value string `json:"@value"`}
type ExpandedSchemaOrg []struct { Creator []StringVal `json:"http://purl.org/dc/elements/1.1/creator"` Date []StringVal `json:"http://purl.org/dc/elements/1.1/date"` Title []StringVal `json:"http://purl.org/dc/elements/1.1/title"`}
type SegmentMetadata struct { StartTime aqtime.AQTime Title string Creator string ContentWarnings []string ContentRights *localdb.ContentRights DistributionPolicy *localdb.DistributionPolicy MetadataConfiguration *placestream.MetadataConfiguration Livestream *placestream.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 continue } if a.Label == "place.stream.metadata" { // backwards compatibility for old manifests ass = &a 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 { return nil, ErrMissingMetadata } proc := ld.NewJsonLdProcessor() options := ld.NewJsonLdOptions("") flat, err := proc.Expand(ass.Data, options) if err != nil { return nil, err } bs, err := json.Marshal(flat) if err != nil { return nil, err } var metas ExpandedSchemaOrg err = json.Unmarshal(bs, &metas) if err != nil { return nil, err } if len(metas) != 1 { return nil, ErrInvalidMetadata } meta := metas[0] if len(meta.Creator) == 0 { return nil, ErrInvalidMetadata } if len(meta.Title) != 1 { return nil, ErrInvalidMetadata } if len(meta.Date) != 1 { return nil, ErrInvalidMetadata } start, err := aqtime.FromString(meta.Date[0].Value) if err != nil { return nil, err }
contentWarnings := extractContentWarnings(mani) contentRights := extractContentRights(mani) distributionPolicy := extractDistributionPolicy(mani, start) metadataConfiguration := extractMetadataConfiguration(mani) livestream := extractLivestream(mani)
out := SegmentMetadata{ StartTime: start, Title: meta.Title[0].Value, Creator: meta.Creator[0].Value, ContentWarnings: contentWarnings, ContentRights: contentRights, DistributionPolicy: distributionPolicy, MetadataConfiguration: metadataConfiguration, Livestream: livestream, Published: isPublished, } return &out, nil}
// findAssertion finds an assertion by labelfunc findAssertion(mani *c2patypes.Manifest, label string) *c2patypes.ManifestAssertion { for _, a := range mani.Assertions { if a.Label == label { return &a } } return nil}
// extractContentWarnings extracts content warnings from the C2PA manifestfunc extractContentWarnings(mani *c2patypes.Manifest) []string { ass := findAssertion(mani, StreamplaceMetadata) if ass == nil { return nil }
data, ok := ass.Data.(map[string]interface{}) if !ok { return nil }
warnings, ok := data["Iptc4xmpExt:ContentWarning"] if !ok { return nil }
warningList, ok := warnings.([]interface{}) if !ok { return nil }
result := make([]string, 0, len(warningList)) for _, warning := range warningList { if warningStr, ok := warning.(string); ok { result = append(result, warningStr) } }
return result}
// extractContentRights extracts content rights from the C2PA manifestfunc extractContentRights(mani *c2patypes.Manifest) *localdb.ContentRights { ass := findAssertion(mani, StreamplaceMetadata) if ass == nil { return nil }
data, ok := ass.Data.(map[string]interface{}) if !ok { return nil }
rights := &localdb.ContentRights{}
// Extract copyright notice if notice, ok := data["dc:rights"]; ok { if noticeStr, ok := notice.(string); ok { rights.CopyrightNotice = ¬iceStr } }
// Extract copyright year if year, ok := data["Iptc4xmpExt:CopyrightYear"]; ok { if yearNum, ok := year.(float64); ok { yearInt := int64(yearNum) rights.CopyrightYear = &yearInt } }
// Extract creator if creator, ok := data["dc:creator"]; ok { if creatorStr, ok := creator.(string); ok { rights.Creator = &creatorStr } }
// Extract credit line if credit, ok := data["photoshop:Credit"]; ok { if creditStr, ok := credit.(string); ok { rights.CreditLine = &creditStr } }
// Extract license information if license, ok := data["Iptc4xmpExt:LinkedEncRightsExpr"]; ok { if licenseStr, ok := license.(string); ok { rights.License = &licenseStr } } else if usageTerms, ok := data["xmpRights:UsageTerms"]; ok { if usageStr, ok := usageTerms.(string); ok { rights.License = &usageStr } }
// Return nil if no rights information was found if rights.CopyrightNotice == nil && rights.CopyrightYear == nil && rights.Creator == nil && rights.CreditLine == nil && rights.License == nil { return nil }
return rights}
// extractDistributionPolicy extracts distribution policy from the C2PA manifestfunc extractDistributionPolicy(mani *c2patypes.Manifest, segmentStart aqtime.AQTime) *localdb.DistributionPolicy { metadataConfig := extractMetadataConfiguration(mani) if metadataConfig == nil { return nil }
if metadataConfig.DistributionPolicy == nil { return nil }
policy := &localdb.DistributionPolicy{ AllowGenAiTraining: metadataConfig.DistributionPolicy.AllowGenAiTraining, }
if metadataConfig.DistributionPolicy.DeleteAfter != nil { // deleteAfter contains an offset in seconds from creation time deleteAfterSeconds := *metadataConfig.DistributionPolicy.DeleteAfter policy.DeleteAfterSeconds = &deleteAfterSeconds }
if policy.DeleteAfterSeconds == nil && policy.AllowGenAiTraining == nil { return nil }
return policy}
// extractMetadataConfiguration extracts the place.stream.metadata.configuration from the C2PA manifestfunc extractMetadataConfiguration(mani *c2patypes.Manifest) *placestream.MetadataConfiguration { ass := findAssertion(mani, "place.stream.metadata.configuration") if ass == nil { return nil }
bs, err := json.Marshal(ass.Data) if err != nil { return nil } var metadataConfiguration placestream.MetadataConfiguration err = json.Unmarshal(bs, &metadataConfiguration) if err != nil { return nil } return &metadataConfiguration}
func extractLivestream(mani *c2patypes.Manifest) *placestream.Livestream { ass := findAssertion(mani, "place.stream.livestream") if ass == nil { return nil } bs, err := json.Marshal(ass.Data) if err != nil { return nil }
var livestream placestream.Livestream err = json.Unmarshal(bs, &livestream) if err != nil { return nil } return &livestream}
// NewOffline is a MediaManager with no node behind it — just the// configuration and the in-memory windows — for harnesses that run the// transcode and rendition code paths against stored segments.func NewOffline(cli *config.CLI) *MediaManager { return &MediaManager{ cli: cli, liveWindows: map[string]*livehls.Writer{}, liveWindowPublished: map[string]bool{}, }}