diff --git a/pkg/api/api_internal.go b/pkg/api/api_internal.go index bddbc8a5..84e203d1 100644 --- a/pkg/api/api_internal.go +++ b/pkg/api/api_internal.go @@ -102,6 +102,10 @@ func (a *StreamplaceAPI) InternalHandler(ctx context.Context) (http.Handler, err w.WriteHeader(204) }) + // Pull a VOD blob from another Streamplace node into our playback + // store and attest to it via place.stream.media.origin. + router.POST("/vod-transfer", a.HandleVODTransfer(ctx)) + router.Handler("GET", "/metrics", promhttp.Handler()) router.GET("/playback/:user/:rendition/concat", func(w http.ResponseWriter, r *http.Request, p httprouter.Params) { diff --git a/pkg/api/vod_transfer.go b/pkg/api/vod_transfer.go new file mode 100644 index 00000000..31265387 --- /dev/null +++ b/pkg/api/vod_transfer.go @@ -0,0 +1,187 @@ +package api + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "time" + + comatproto "github.com/bluesky-social/indigo/api/atproto" + "github.com/bluesky-social/indigo/atproto/syntax" + "github.com/julienschmidt/httprouter" + + "stream.place/streamplace/pkg/errors" + "stream.place/streamplace/pkg/log" + "stream.place/streamplace/pkg/vod" +) + +// vodTransferRequest is the body of POST /vod-transfer. Either resolve the +// content blob from a place.stream.video record (Source + URI) or name the +// blob directly (CID + DID); Source is always required. +type vodTransferRequest struct { + // Source is the base URL of the node to pull the blob from + // (e.g. "https://source.example"). + Source string `json:"source"` + // URI is a place.stream.video AT-URI. When set, the content CID and + // owning DID are resolved from the local index. + URI string `json:"uri,omitempty"` + // CID is the content blob's BDASL CID, an alternative to URI for when + // the record isn't locally indexed. Requires DID. + CID string `json:"cid,omitempty"` + // DID is the account that owns a track in the blob; required alongside + // CID (and ignored when URI is set, since it's derived from the record). + DID string `json:"did,omitempty"` +} + +// vodTransferHTTPClient is used for the (potentially long, multi-gigabyte) +// blob fetch. No client-level timeout — the request context bounds it — but +// a generous dial/handshake timeout so an unreachable source fails fast. +var vodTransferHTTPClient = &http.Client{ + Transport: &http.Transport{ + ResponseHeaderTimeout: 30 * time.Second, + }, +} + +// HandleVODTransfer is the internal admin endpoint that pulls a VOD blob +// from another Streamplace node into this node's playback store and +// publishes a place.stream.media.origin for it. See vod.TransferVOD for the +// mechanics; this just resolves the request into (contentCID, did) and +// delegates. +func (a *StreamplaceAPI) HandleVODTransfer(ctx context.Context) httprouter.Handle { + return func(w http.ResponseWriter, r *http.Request, p httprouter.Params) { + reqCtx := r.Context() + + if a.PlaybackStore == nil { + errors.WriteHTTPInternalServerError(w, "playback store not configured", nil) + return + } + + var req vodTransferRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + errors.WriteHTTPBadRequest(w, "invalid request body", err) + return + } + if req.Source == "" { + errors.WriteHTTPBadRequest(w, "source is required", nil) + return + } + + contentCID := req.CID + did := req.DID + switch { + case req.URI != "": + var err error + contentCID, did, err = a.resolveVideoContentBlob(reqCtx, req.URI) + if err != nil { + errors.WriteHTTPBadRequest(w, fmt.Sprintf("resolve %s", req.URI), err) + return + } + case req.CID != "" && req.DID != "": + // Used as supplied. + default: + errors.WriteHTTPBadRequest(w, "provide either uri, or both cid and did", nil) + return + } + + log.Log(reqCtx, "vod transfer requested", "source", req.Source, "uri", req.URI, "cid", contentCID, "did", did) + + result, err := vod.TransferVOD(reqCtx, a.CLI, a.PlaybackStore, vodTransferHTTPClient, req.Source, contentCID, did) + if err != nil { + errors.WriteHTTPInternalServerError(w, "vod transfer failed", err) + return + } + + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(result); err != nil { + log.Error(reqCtx, "error writing vod-transfer response", "error", err) + } + } +} + +// resolveVideoContentBlob walks a place.stream.video record in the local +// index to the content blob a transfer should fetch, returning the blob's +// BDASL CID and the DID that owns a track in it (required by the source's +// getVideoBlob). +// +// Mirrors the playback resolver (pkg/spxrpc resolveVideoBlob): a +// sourceTracks record resolves to its first track's muxlTrack.blob; a +// sourceClip resolves through to its parent video (one level only), whose +// blob is the one actually backing the clip. +func (a *StreamplaceAPI) resolveVideoContentBlob(ctx context.Context, rawURI string) (cid string, did string, err error) { + aturi, err := syntax.ParseATURI(rawURI) + if err != nil { + return "", "", fmt.Errorf("invalid AT-URI: %w", err) + } + + rec, err := a.Model.GetVideoByURI(ctx, aturi.String()) + if err != nil { + return "", "", fmt.Errorf("get video: %w", err) + } + if rec == nil { + return "", "", fmt.Errorf("video not indexed locally (pass cid + did instead)") + } + + switch { + case rec.Source != nil && rec.Source.MediaDefs_SourceTracks != nil: + cid, err = a.firstTrackBlobCID(ctx, rec.Source.MediaDefs_SourceTracks.Tracks) + if err != nil { + return "", "", err + } + return cid, aturi.Authority().String(), nil + + case rec.Source != nil && rec.Source.MediaDefs_SourceClip != nil: + clip := rec.Source.MediaDefs_SourceClip + if clip.Video == "" { + return "", "", fmt.Errorf("sourceClip missing parent video URI") + } + parentURI, err := syntax.ParseATURI(clip.Video) + if err != nil { + return "", "", fmt.Errorf("invalid parent video URI: %w", err) + } + parent, err := a.Model.GetVideoByURI(ctx, parentURI.String()) + if err != nil { + return "", "", fmt.Errorf("get parent video: %w", err) + } + if parent == nil { + return "", "", fmt.Errorf("parent video %s not indexed locally", parentURI.String()) + } + if parent.Source == nil || parent.Source.MediaDefs_SourceTracks == nil { + return "", "", fmt.Errorf("sourceClip parent must be a sourceTracks video (clip-of-clip unsupported)") + } + cid, err = a.firstTrackBlobCID(ctx, parent.Source.MediaDefs_SourceTracks.Tracks) + if err != nil { + return "", "", err + } + // The blob lives in the parent's tracks, so attribute to the parent's owner. + return cid, parentURI.Authority().String(), nil + + default: + return "", "", fmt.Errorf("video record has no playable source") + } +} + +// firstTrackBlobCID resolves the muxlTrack.blob CID of the first track ref +// in a sourceTracks bundle via the local index. The metafile keyed at that +// CID catalogs every track of the container, so the first is enough. +func (a *StreamplaceAPI) firstTrackBlobCID(ctx context.Context, tracks []*comatproto.RepoStrongRef) (string, error) { + if len(tracks) == 0 { + return "", fmt.Errorf("video record has no tracks") + } + first := tracks[0] + if first == nil || first.Uri == "" { + return "", fmt.Errorf("first track ref is empty") + } + track, err := a.Model.GetMediaTrackByURI(ctx, first.Uri) + if err != nil { + return "", fmt.Errorf("get track %s: %w", first.Uri, err) + } + if track == nil || track.Track == nil || track.Track.MediaDefs_MuxlTrack == nil { + return "", fmt.Errorf("track %s not indexed or not a muxlTrack", first.Uri) + } + blob := track.Track.MediaDefs_MuxlTrack.Blob + if blob == "" { + return "", fmt.Errorf("track %s has no blob CID", first.Uri) + } + return blob, nil +} diff --git a/pkg/vod/transfer.go b/pkg/vod/transfer.go new file mode 100644 index 00000000..0b207b1b --- /dev/null +++ b/pkg/vod/transfer.go @@ -0,0 +1,300 @@ +package vod + +import ( + "context" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "sort" + "strings" + + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + "go.opentelemetry.io/otel/trace" + + "stream.place/streamplace/pkg/bdasl" + "stream.place/streamplace/pkg/blob" + "stream.place/streamplace/pkg/config" + "stream.place/streamplace/pkg/log" + "stream.place/streamplace/pkg/muxl" +) + +// transferMimeType is the MIME we record on the media.origin for a +// transferred blob. Every VOD content + init blob is a fragmented MP4. +const transferMimeType = "video/mp4" + +// TransferResult summarizes a completed VOD transfer for the caller (the +// internal admin endpoint). +type TransferResult struct { + // ContentCID is the BDASL CID of the primary blob we now host. + ContentCID string `json:"contentCid"` + // Size is the primary blob's size in bytes. + Size int64 `json:"size"` + // DID is the account whose track the blob was fetched for (used for + // the source's egress attribution + labeler check). + DID string `json:"did"` + // InitCIDs are the per-track init segment CIDs we regenerated and + // wrote locally, deduplicated and sorted. + InitCIDs []string `json:"initCids"` + // TrackCount is the number of tracks in the regenerated metafile. + TrackCount int `json:"trackCount"` + // Downloaded is false when the content blob was already present + // locally and the network fetch was skipped. + Downloaded bool `json:"downloaded"` +} + +// TransferVOD pulls a content-addressed VOD blob from a remote Streamplace +// node into this node's playback store, regenerates the playback sidecars +// (metafile + per-track init segments) locally from the blob, and publishes +// a place.stream.media.origin attestation that this node now hosts it. +// +// Only the primary content blob crosses the network. The metafile and the +// per-track init segments are deterministically derivable from it (via +// `muxl unwrap`), so we rebuild them locally rather than fetch them — the +// wire protocol is a single content-addressed GET against the source's +// existing place.stream.playback.getVideoBlob endpoint, and no +// non-content-addressed surface is required on the source. +// +// The fetched bytes are verified against contentCID before they're +// published; a source that serves the wrong bytes fails the transfer +// rather than poisoning the store. The published origin is byte-for-byte +// what the normal processing pipeline emits (rkey = CID), so it's +// idempotent across retries and indistinguishable from a locally-produced +// VOD once indexed. +// +// sourceBaseURL is the origin node's base URL (e.g. "https://source.example"). +// contentCID is the primary blob's BDASL CID. did is an account that owns a +// track in the blob — the source's getVideoBlob requires it for egress +// attribution and labeler enforcement on content blobs. +func TransferVOD(ctx context.Context, cli *config.CLI, store blob.Store, httpClient *http.Client, sourceBaseURL, contentCID, did string) (*TransferResult, error) { + ctx, span := vodTracer.Start(ctx, "vod.TransferVOD", trace.WithAttributes( + attribute.String("cid", contentCID), + attribute.String("did", did), + )) + defer span.End() + + if _, err := bdasl.Parse(contentCID); err != nil { + return nil, fmt.Errorf("invalid content CID %q: %w", contentCID, err) + } + if did == "" { + return nil, fmt.Errorf("did is required") + } + base, err := url.Parse(sourceBaseURL) + if err != nil || base.Scheme == "" || base.Host == "" { + return nil, fmt.Errorf("invalid source base URL %q", sourceBaseURL) + } + if httpClient == nil { + httpClient = http.DefaultClient + } + + key := BlobsPrefix + contentCID + ".mp4" + + // Skip the (potentially multi-gigabyte) download if we already hold + // the blob. Sidecars are regenerated unconditionally below so a prior + // transfer that died after the download still converges. + downloaded := false + size, err := existingBlobSize(ctx, store, key) + switch { + case errors.Is(err, blob.ErrNotFound): + size, err = downloadContentBlob(ctx, httpClient, store, base, key, contentCID, did) + if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, "download") + return nil, err + } + downloaded = true + case err != nil: + return nil, fmt.Errorf("stat existing blob: %w", err) + default: + log.Log(ctx, "vod transfer: content blob already present, skipping download", "cid", contentCID, "size", size) + } + + meta, initCIDs, err := ensureSidecars(ctx, store, contentCID, size) + if err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, "regenerate") + return nil, fmt.Errorf("regenerate sidecars: %w", err) + } + + if err := publishOrigin(ctx, cli, contentCID, size, transferMimeType); err != nil { + span.RecordError(err) + span.SetStatus(codes.Error, "origin") + return nil, fmt.Errorf("publish origin: %w", err) + } + + span.SetAttributes( + attribute.Int64("size", size), + attribute.Int("track_count", len(meta.Tracks)), + attribute.Bool("downloaded", downloaded), + ) + span.SetStatus(codes.Ok, "") + log.Log(ctx, "vod transfer complete", + "cid", contentCID, + "size", size, + "tracks", len(meta.Tracks), + "initCids", len(initCIDs), + "downloaded", downloaded, + ) + return &TransferResult{ + ContentCID: contentCID, + Size: size, + DID: did, + InitCIDs: initCIDs, + TrackCount: len(meta.Tracks), + Downloaded: downloaded, + }, nil +} + +// existingBlobSize returns the size of the blob at key, or blob.ErrNotFound +// (unwrapped via errors.Is by the caller) if it isn't present. +func existingBlobSize(ctx context.Context, store blob.Store, key string) (int64, error) { + r, err := store.Open(ctx, key) + if err != nil { + return 0, err + } + defer r.Close() + return r.Size(), nil +} + +// downloadContentBlob streams the blob from the source node's getVideoBlob +// endpoint into the store, hashing as it goes. The store write is only +// Completed (made visible) once the running hash matches contentCID, so a +// source that serves the wrong or corrupt bytes leaves nothing behind. +func downloadContentBlob(ctx context.Context, httpClient *http.Client, store blob.Store, base *url.URL, key, contentCID, did string) (int64, error) { + endpoint := strings.TrimRight(base.String(), "/") + "/xrpc/place.stream.playback.getVideoBlob" + u, err := url.Parse(endpoint) + if err != nil { + return 0, fmt.Errorf("build getVideoBlob URL: %w", err) + } + u.RawQuery = url.Values{"cid": {contentCID}, "did": {did}}.Encode() + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil) + if err != nil { + return 0, fmt.Errorf("build request: %w", err) + } + resp, err := httpClient.Do(req) + if err != nil { + return 0, fmt.Errorf("fetch blob from %s: %w", base.Redacted(), err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(io.LimitReader(resp.Body, 2048)) + return 0, fmt.Errorf("source returned %s for cid %s: %s", resp.Status, contentCID, strings.TrimSpace(string(body))) + } + + w, err := store.NewWriter(ctx, key, transferMimeType) + if err != nil { + return 0, fmt.Errorf("open store writer: %w", err) + } + // Close aborts the staged bytes unless Complete ran first, so a CID + // mismatch or copy error discards the partial blob. + defer w.Close() + + hasher := bdasl.NewWriter() + n, err := io.Copy(io.MultiWriter(w, hasher), resp.Body) + if err != nil { + return 0, fmt.Errorf("stream blob to store: %w", err) + } + if got := hasher.CID(); got != contentCID { + return 0, fmt.Errorf("CID mismatch: source served %s, expected %s (%d bytes)", got, contentCID, n) + } + if err := w.Complete(); err != nil { + return 0, fmt.Errorf("complete store write: %w", err) + } + log.Log(ctx, "vod transfer: downloaded content blob", "cid", contentCID, "bytes", n, "source", base.Redacted()) + return n, nil +} + +// ensureSidecars returns the playback sidecars for an already-stored +// content blob, regenerating them only if they're not already present. The +// sidecars are a deterministic function of the (CID-verified) content blob, +// so an existing metafile is necessarily correct — and since the builder +// writes the per-track init blobs before the metafile, a present metafile +// implies the inits are present too. This fast path matters because +// re-deriving is a full `muxl unwrap` pass over a blob that can be many +// gigabytes; an idempotent re-run shouldn't pay for it twice. +func ensureSidecars(ctx context.Context, store blob.Store, contentCID string, size int64) (*Metafile, []string, error) { + meta, err := readMetafile(ctx, store, contentCID) + if err == nil { + log.Log(ctx, "vod transfer: sidecars already present, skipping regeneration", "cid", contentCID) + return meta, dedupedInitCIDs(meta), nil + } + if !errors.Is(err, blob.ErrNotFound) { + return nil, nil, fmt.Errorf("check existing metafile: %w", err) + } + return regenerateSidecars(ctx, store, contentCID, size) +} + +// regenerateSidecars rebuilds the playback sidecars for an already-stored +// content blob: the per-track init segments and the metafile JSON. It feeds +// the stored blob through `muxl unwrap --events` (which re-derives the +// per-track event stream byte-for-byte) into the same metafileBuilder the +// processing pipeline uses, so the regenerated sidecars are identical to +// what the originating node produced. +func regenerateSidecars(ctx context.Context, store blob.Store, contentCID string, size int64) (*Metafile, []string, error) { + r, err := store.Open(ctx, BlobsPrefix+contentCID+".mp4") + if err != nil { + return nil, nil, fmt.Errorf("open content blob: %w", err) + } + defer r.Close() + + mb := newMetafileBuilder(ctx, store) + + // RunMuxlUnwrapEvents writes events synchronously and does not close + // the channel (the caller owns it), so the producer runs in its own + // goroutine while we drain on this one. + eventCh := make(chan *muxl.MuxlEvent, 16) + producerErr := make(chan error, 1) + go func() { + producerErr <- muxl.RunMuxlUnwrapEvents(ctx, io.NewSectionReader(r, 0, size), eventCh) + close(eventCh) + }() + + var obsErr error + for ev := range eventCh { + if e := mb.Observe(ev); e != nil && obsErr == nil { + obsErr = e + } + } + if perr := <-producerErr; perr != nil { + return nil, nil, fmt.Errorf("muxl unwrap: %w", perr) + } + if obsErr != nil { + return nil, nil, fmt.Errorf("metafile build: %w", obsErr) + } + // `muxl unwrap` returns nil (not an error) when the context is + // cancelled mid-stream, which would otherwise let a truncated event + // stream produce a partial metafile. Refuse to publish that. + if err := ctx.Err(); err != nil { + return nil, nil, err + } + + meta := mb.Finalize(contentCID, size) + if err := writeMetafile(ctx, store, contentCID, meta); err != nil { + return nil, nil, err + } + + return meta, dedupedInitCIDs(meta), nil +} + +// dedupedInitCIDs returns the distinct per-track init CIDs from a metafile, +// sorted for a stable result. The builder already wrote each init blob to +// the store; this is just the summary for the transfer result. +func dedupedInitCIDs(meta *Metafile) []string { + seen := map[string]struct{}{} + out := make([]string, 0, len(meta.Tracks)) + for _, tr := range meta.Tracks { + if tr.InitCID == "" { + continue + } + if _, ok := seen[tr.InitCID]; ok { + continue + } + seen[tr.InitCID] = struct{}{} + out = append(out, tr.InitCID) + } + sort.Strings(out) + return out +} diff --git a/pkg/vod/transfer_test.go b/pkg/vod/transfer_test.go new file mode 100644 index 00000000..314ca016 --- /dev/null +++ b/pkg/vod/transfer_test.go @@ -0,0 +1,109 @@ +package vod + +import ( + "bytes" + "context" + "os" + "testing" + "time" + + "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/bdasl" + "stream.place/streamplace/pkg/blob" + "stream.place/streamplace/pkg/log" +) + +// TestRegenerateSidecars is the load-bearing test for VOD transfer: it +// proves that the metafile + per-track init segments a node produces while +// *processing* a VOD can be reproduced byte-for-identically from just the +// finished content blob (which is the only thing TransferVOD pulls over the +// network). +// +// It runs the real processing pipeline to get a ground-truth metafile + +// init blobs in store A, then feeds only the resulting content blob through +// regenerateSidecars into a fresh store B and asserts the regenerated +// metafile matches exactly — same track set, same init CIDs, same segment +// byte offsets/sizes/durations — and that every init blob + the metafile +// landed in store B. If `muxl unwrap` ever stopped emitting per-track init +// bytes, or the offset accounting drifted, this fails. +func TestRegenerateSidecars(t *testing.T) { + warmGST() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + ctx = log.WithLogValues(ctx, "test", "TestRegenerateSidecars") + + fixture, err := os.ReadFile(getFixture("5sec.mp4")) + require.NoError(t, err) + + signer, err := newUploadSigner(time.Now()) + require.NoError(t, err) + + // --- store A: ground truth from the real processing pipeline --------- + storeA, err := blob.NewFileStore(t.TempDir()) + require.NoError(t, err) + + out := &bytes.Buffer{} + hasher := bdasl.NewWriter() + dst := teeWriter{hasher, out} + + mbA := newMetafileBuilder(ctx, storeA) + _, err = streamThroughMuxl(ctx, bytes.NewReader(fixture), int64(len(fixture)), dst, mbA, signer.SignerInput) + require.NoError(t, err) + + cid := hasher.CID() + blobBytes := out.Bytes() + metaA := mbA.Finalize(cid, int64(len(blobBytes))) + require.NotEmpty(t, metaA.Tracks) + + // --- store B: regenerate from only the content blob ------------------ + storeB, err := blob.NewFileStore(t.TempDir()) + require.NoError(t, err) + + // Place the content blob exactly where a download would have left it. + key := BlobsPrefix + cid + ".mp4" + w, err := storeB.NewWriter(ctx, key, transferMimeType) + require.NoError(t, err) + _, err = w.Write(blobBytes) + require.NoError(t, err) + require.NoError(t, w.Complete()) + + metaB, initCIDs, err := regenerateSidecars(ctx, storeB, cid, int64(len(blobBytes))) + require.NoError(t, err) + + // The regenerated metafile must equal the one the processing pipeline + // produced. BlobCID/BlobSize are derived inputs; the meat is the track + // table (codecs, init CIDs, and especially segment byte ranges). + require.Equal(t, metaA.BlobCID, metaB.BlobCID) + require.Equal(t, metaA.BlobSize, metaB.BlobSize) + require.Equal(t, metaA.Tracks, metaB.Tracks, "regenerated metafile tracks differ from processing-time metafile") + + // Every init CID reported back was actually written into store B, and + // matches the init the pipeline wrote into store A. + require.NotEmpty(t, initCIDs) + for _, tr := range metaB.Tracks { + require.NotEmpty(t, tr.InitCID, "track missing initCid") + require.Contains(t, initCIDs, tr.InitCID) + + rb, err := storeB.Open(ctx, BlobsPrefix+tr.InitCID+".mp4") + require.NoError(t, err, "init blob %s not written to store B", tr.InitCID) + bBytes := make([]byte, rb.Size()) + _, _ = rb.ReadAt(bBytes, 0) + require.NoError(t, rb.Close()) + + ra, err := storeA.Open(ctx, BlobsPrefix+tr.InitCID+".mp4") + require.NoError(t, err, "init blob %s missing from store A", tr.InitCID) + aBytes := make([]byte, ra.Size()) + _, _ = ra.ReadAt(aBytes, 0) + require.NoError(t, ra.Close()) + + require.Equal(t, aBytes, bBytes, "init blob %s bytes differ between stores", tr.InitCID) + require.NoError(t, bdasl.Verify(tr.InitCID, bBytes), "init blob %s fails its own CID", tr.InitCID) + } + + // The metafile JSON sidecar landed at blobs/.json in store B. + mr, err := storeB.Open(ctx, BlobsPrefix+cid+".json") + require.NoError(t, err, "metafile not written to store B") + require.Positive(t, mr.Size()) + require.NoError(t, mr.Close()) +}