Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
9.8 kB · 255 lines
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256package vod
import ( "bytes" "context" "errors" "fmt" "io" "strings"
"go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/codes" "go.opentelemetry.io/otel/trace"
"stream.place/streamplace/pkg/blob" "stream.place/streamplace/pkg/config" "stream.place/streamplace/pkg/log" "stream.place/streamplace/pkg/media" "stream.place/streamplace/pkg/muxl" s3pkg "stream.place/streamplace/pkg/s3" "stream.place/streamplace/pkg/statedb")
// FinalizeInput drives FinalizeLivestreamVOD. It's resolved by the bootstrap// from the FinalizeLivestreamVODTask payload (plus a model lookup for the// signing key), so this package needn't import pkg/model.type FinalizeInput struct { // UploadID is the synthetic Upload row created by the finalizeLivestream // procedure; the getUploadStatus/publishVideo client contract keys off it. UploadID string // RepoDID is the streamer whose repo the media.track records land in. RepoDID string // LivestreamURI selects which S3Segment objects belong to this stream. LivestreamURI string // SigningKey is the did:key whose private half C2PA-signed the live // segments (the streamer's place.stream.key). Recorded on each // place.stream.media.track so playback can verify signatures. SigningKey string}
// FinalizeLivestreamVOD turns a finished livestream's recorded MUXL objects// into a VOD, reusing the upload pipeline's tail half. The live segments are// already C2PA-signed canonical MUXL, so — unlike ProcessVOD — there is no// gstreamer remux and no re-signing: we concatenate the objects under one// synthesized init header into a content-addressed blob, derive the playback// sidecars from it with `muxl unwrap` (exactly as VOD transfer does), and// publish the same origin + track records.//// The content blob is shaped [init][segments…] — byte-structurally identical// to an uploaded VOD — so the metafile offset math is the proven path and a// live-derived VOD is indistinguishable from an uploaded one downstream// (playback, transfer, view-count). The bulk of the bytes are assembled// server-side via S3 UploadPartCopy; only the ~5 MB header part is uploaded.func FinalizeLivestreamVOD(ctx context.Context, cli *config.CLI, state *statedb.StatefulDB, store blob.Store, in FinalizeInput) (string, error) { ctx = log.WithLogValues(ctx, "func", "FinalizeLivestreamVOD", "uploadId", in.UploadID, "did", in.RepoDID, "livestream", in.LivestreamURI) ctx, span := vodTracer.Start(ctx, "vod.FinalizeLivestreamVOD", trace.WithAttributes( attribute.String("upload_id", in.UploadID), attribute.String("did", in.RepoDID), attribute.String("livestream", in.LivestreamURI), )) defer span.End()
segs, err := state.ListS3SegmentsForLivestream(ctx, in.LivestreamURI) if err != nil { recordErr(span, "list_segments", err) return "", fmt.Errorf("list s3 segments: %w", err) } if len(segs) == 0 { err := fmt.Errorf("no recorded S3 segments for livestream %s", in.LivestreamURI) recordErr(span, "list_segments", err) return "", err } keys := make([]string, len(segs)) for i, s := range segs { keys[i] = s.Key } span.SetAttributes(attribute.Int("object_count", len(keys))) log.Log(ctx, "finalizing livestream VOD", "objects", len(keys))
// 1. Synthesize the flat-MP4 faststart header from the fragments' metafile. flatHeader, err := synthFlatHeaderForObjects(ctx, store, keys) if err != nil { recordErr(span, "synth_header", err) return "", fmt.Errorf("synthesize flat header: %w", err) } span.SetAttributes(attribute.Int("flat_header_bytes", len(flatHeader)))
// 2. Hash the bare fragments → MUXL CID (the stable, header-independent id) // and build the fragment-relative HLS metafile + per-track init blobs. muxlCID, fragSize, metafile, err := hashAndBuildFragmentMetafile(ctx, store, keys) if err != nil { recordErr(span, "build_metafile", err) return "", err } metafile.FlatHeaderSize = int64(len(flatHeader)) blobSize := int64(len(flatHeader)) + fragSize contentKey := BlobsPrefix + muxlCID + ".mp4" span.SetAttributes( attribute.String("muxl_cid", muxlCID), attribute.Int64("blob_size", blobSize), attribute.String("content_key", contentKey), )
// 3. Assemble [flat-header][fragments] at blobs/<muxlCID>.mp4 (the flat-header // is uploaded, the fragments server-side-copied after it). if err := runVODStage(ctx, "concat_assemble", func(ctx context.Context) error { return assembleContentBlob(ctx, store, flatHeader, keys, contentKey) }); err != nil { recordErr(span, "concat_assemble", err) return "", fmt.Errorf("assemble content blob: %w", err) }
if err := runVODStage(ctx, stageMetafile, func(ctx context.Context) error { return writeMetafile(ctx, store, muxlCID, metafile) }); err != nil { recordErr(span, stageMetafile, err) return "", fmt.Errorf("write metafile: %w", err) }
// 4. Publish origin + track records (referencing the MUXL CID) and store // TrackURIs on the Upload row, reusing the upload publish path unchanged. probe := metafileToVODResult(metafile) if err := runVODStage(ctx, stagePublish, func(ctx context.Context) error { return publishRecords(ctx, publishParams{ cli: cli, state: state, in: Input{UploadID: in.UploadID, RepoDID: in.RepoDID}, cid: muxlCID, size: blobSize, mimeType: "video/mp4", probe: probe, signingKey: in.SigningKey, }) }); err != nil { recordErr(span, stagePublish, err) return "", fmt.Errorf("publish records: %w", err) }
span.SetStatus(codes.Ok, "") log.Log(ctx, "livestream VOD finalized", "muxlCid", muxlCID, "blobSize", blobSize, "objects", len(keys), "duration_ms", probe.DurationMS) return muxlCID, nil}
// captureInitHeader synthesizes the per-stream init segment (ftyp+moov) from// the first recorded object via `muxl wrap --init-only`. The live objects are// bare canonical segments whose embedded catalogs carry everything needed to// derive the init; prepending this header to their concatenation yields the// proven [init][segments…] VOD blob shape.//// RunMuxlWrapInit writes straight to a buffer (no event channel), so — unlike// an unwrap pass — there's no stdout pipe to back up and no mid-stream cancel,// which is what previously deadlocked here.func captureInitHeader(ctx context.Context, store blob.Store, key string) ([]byte, error) { r, err := store.Open(ctx, key) if err != nil { return nil, fmt.Errorf("open first object %s: %w", key, err) } defer r.Close()
var buf bytes.Buffer if err := muxl.RunMuxlWrapInit(ctx, io.NewSectionReader(r, 0, r.Size()), &buf); err != nil { return nil, fmt.Errorf("synthesize init header: %w", err) } if buf.Len() == 0 { return nil, errors.New("muxl produced an empty init header") } return buf.Bytes(), nil}
// assembleContentBlob writes header ++ objects to contentKey. For an S3 store// this is a near-entirely server-side concat (uniform UploadPartCopy windows;// only header/boundary windows pass through this process); any non-S3 store// falls back to a full rewrite.func assembleContentBlob(ctx context.Context, store blob.Store, header []byte, keys []string, contentKey string) error { if s3store, ok := store.(*blob.S3Store); ok { return s3pkg.ConcatWithHeader(ctx, s3store.Client(), s3store.Bucket(), header, keys, contentKey, "video/mp4") } return assembleViaWriter(ctx, store, header, keys, contentKey)}
// assembleViaWriter streams header ++ objects through a store writer. The// universal fallback: correct for any blob.Store, but re-uploads every byte.func assembleViaWriter(ctx context.Context, store blob.Store, header []byte, keys []string, contentKey string) error { w, err := store.NewWriter(ctx, contentKey, "video/mp4") if err != nil { return fmt.Errorf("open content writer: %w", err) } defer w.Close() if _, err := w.Write(header); err != nil { return fmt.Errorf("write header: %w", err) } for _, key := range keys { r, err := store.Open(ctx, key) if err != nil { return fmt.Errorf("open object %s: %w", key, err) } _, err = io.Copy(w, io.NewSectionReader(r, 0, r.Size())) _ = r.Close() if err != nil { return fmt.Errorf("copy object %s: %w", key, err) } } return w.Complete()}
// metafileToVODResult derives the probe metadata publishTrack needs from the// regenerated metafile, standing in for the gstreamer probe that an upload// would have produced. publishTrack hardcodes the video codec to h264 and only// uses width/height (+ optional framerate), so we leave framerate unset.func metafileToVODResult(meta *Metafile) media.VODResult { var res media.VODResult for _, t := range meta.Tracks { switch t.Type { case "video": res.Video = &media.VODVideoTrack{ Codec: t.Codec, Width: int(t.Width), Height: int(t.Height), } case "audio": res.Audio = &media.VODAudioTrack{ Codec: audioProbeCodec(t.Codec), Rate: int(t.SampleRate), Channels: int(t.Channels), } } if d := metafileTrackDurationMS(t); d > res.DurationMS { res.DurationMS = d } } return res}
// metafileTrackDurationMS sums a track's segment durations (in its timescale)// and converts to milliseconds.func metafileTrackDurationMS(t MetafileTrack) int64 { if t.Timescale == 0 { return 0 } var ticks uint64 for _, s := range t.Segments { ticks += s.DurationTicks } return int64(ticks * 1000 / uint64(t.Timescale))}
// audioProbeCodec maps a metafile (CMAF catalog) audio codec name to the// gstreamer-caps-style value audioCodecForLexicon (publish.go) expects, so the// lexicon enum comes out right ("opus" vs "aac").func audioProbeCodec(metafileCodec string) string { if strings.Contains(strings.ToLower(metafileCodec), "opus") { return "x-opus" } return "mpeg" // audioCodecForLexicon maps "mpeg" -> "aac"}