diff --git a/pkg/atproto/firehose.go b/pkg/atproto/firehose.go index aed0aa0a..bd4ccb79 100644 --- a/pkg/atproto/firehose.go +++ b/pkg/atproto/firehose.go @@ -3,6 +3,7 @@ package atproto import ( "bytes" "context" + "crypto/tls" "fmt" "net/http" "net/url" @@ -268,10 +269,23 @@ func (atsync *ATProtoSynchronizer) connectRelay(ctx context.Context, relay strin // single-node (ServerHost == BroadcasterHost) deployment gets for free. // gorilla/websocket pulls the "Host" header out and uses it as the // HTTP Host while still dialing the loopback address in u. - if relay == atsync.selfRelayURL() && atsync.CLI.ServerHost != "" { + isSelf := relay == atsync.selfRelayURL() + if isSelf && atsync.CLI.ServerHost != "" { header.Set("Host", atsync.CLI.ServerHost) } - con, _, err := websocket.DefaultDialer.Dial(u.String(), header) + + dialer := websocket.DefaultDialer + if isSelf && u.Scheme == "wss" { + // Under --secure the self-subscription dials wss://127.0.0.1:, + // but our cert is issued for ServerHost, not for the loopback IP we dial + // (and in dev it's frequently self-signed on top of that), so verification + // would fail on hostname every time. Skipping it is not a trust decision: + // the peer on the other end of this loopback socket is this same process. + d := *websocket.DefaultDialer + d.TLSClientConfig = &tls.Config{InsecureSkipVerify: true} + dialer = &d + } + con, _, err := dialer.Dial(u.String(), header) if err != nil { return fmt.Errorf("subscribing to firehose failed (dialing): %w", err) } diff --git a/pkg/atproto/firehose_secure_test.go b/pkg/atproto/firehose_secure_test.go new file mode 100644 index 00000000..db557ef1 --- /dev/null +++ b/pkg/atproto/firehose_secure_test.go @@ -0,0 +1,146 @@ +package atproto + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "math/big" + "net" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/gorilla/websocket" + "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/config" + "stream.place/streamplace/pkg/model" +) + +// TestSelfRelayURLSecure pins the address the self-subscription dials. Under +// --secure the handler lives on HTTPSAddr and HTTPAddr serves only redirects, +// so a ws:// self-relay dials the redirect handler and every +// handshake fails ("websocket: bad handshake") — which silently stops the +// server repo's own place.stream.media.origin records from ever being indexed. +func TestSelfRelayURLSecure(t *testing.T) { + for _, tt := range []struct { + name string + cli config.CLI + expect string + }{ + { + name: "plain http", + cli: config.CLI{HTTPAddr: ":38080", HTTPSAddr: ":38443"}, + expect: "ws://127.0.0.1:38080", + }, + { + name: "secure uses the https listener", + cli: config.CLI{HTTPAddr: ":38080", HTTPSAddr: ":38443", Secure: true}, + expect: "wss://127.0.0.1:38443", + }, + { + // Behind a TLS-terminating proxy we really do serve the handler as + // plain HTTP on HTTPAddr, so this must stay ws://. + name: "behind https proxy stays plain", + cli: config.CLI{HTTPAddr: ":38080", HTTPSAddr: ":38443", BehindHTTPSProxy: true}, + expect: "ws://127.0.0.1:38080", + }, + } { + t.Run(tt.name, func(t *testing.T) { + atsync := &ATProtoSynchronizer{CLI: &tt.cli} + require.Equal(t, tt.expect, atsync.selfRelayURL()) + }) + } +} + +// TestConnectRelaySelfDialsOwnTLSListener is the end-to-end regression for the +// --secure self-subscription: we must complete a wss handshake against our own +// listener even though its certificate is issued for ServerHost and we dial it +// by loopback IP, and we must still present Host: ServerHost so we land on the +// server-repo firehose rather than the broadcaster one. +func TestConnectRelaySelfDialsOwnTLSListener(t *testing.T) { + const serverHost = "fairway-secure.example" + + // A certificate for serverHost and nothing else: no IP SAN, so verifying it + // against the 127.0.0.1 we dial fails. This is what a real deployment has. + cert := selfSignedCertFor(t, serverHost) + + gotHost := make(chan string, 1) + upgrader := websocket.Upgrader{} + srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotHost <- r.Host + con, err := upgrader.Upgrade(w, r, nil) + if err != nil { + return + } + // Handshake is all this test cares about; drop the stream immediately + // so connectRelay returns instead of blocking on HandleRepoStream. + con.Close() + })) + srv.TLS = &tls.Config{Certificates: []tls.Certificate{cert}} + srv.StartTLS() + defer srv.Close() + + _, port, err := net.SplitHostPort(srv.Listener.Addr().String()) + require.NoError(t, err) + + cli := config.CLI{ + HTTPAddr: ":38080", + HTTPSAddr: "127.0.0.1:" + port, + Secure: true, + ServerHost: serverHost, + } + mod, err := model.MakeDB(":memory:") + require.NoError(t, err) + atsync := &ATProtoSynchronizer{CLI: &cli, Model: mod} + + relay := atsync.selfRelayURL() + require.Equal(t, "wss://127.0.0.1:"+port, relay) + + // Control: the stock dialer cannot verify this cert against 127.0.0.1, so + // without the self-dial exemption the handshake never happens. Without this + // the test would still pass if InsecureSkipVerify were dropped. + _, _, err = websocket.DefaultDialer.Dial(relay+"/xrpc/com.atproto.sync.subscribeRepos", nil) + require.Error(t, err, "cert must not verify against the dialed IP, else this test proves nothing") + + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + + // connectRelay returns an error once the (immediately closed) stream ends; + // what matters is that it got past the dial. + err = atsync.connectRelay(ctx, relay, atsync.newRelayCursor(ctx, relay)) + if err != nil { + require.NotContains(t, err.Error(), "(dialing)", + "self-subscription failed at the TLS/websocket handshake") + } + + select { + case host := <-gotHost: + require.Equal(t, serverHost, host, + "self-subscription must present Host: ServerHost to reach the server-repo firehose") + default: + t.Fatal("listener never saw the request") + } +} + +func selfSignedCertFor(t *testing.T, dnsName string) tls.Certificate { + t.Helper() + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + require.NoError(t, err) + tmpl := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: dnsName}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(24 * time.Hour), + KeyUsage: x509.KeyUsageKeyEncipherment | x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + DNSNames: []string{dnsName}, // deliberately no IPAddresses + } + der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key) + require.NoError(t, err) + return tls.Certificate{Certificate: [][]byte{der}, PrivateKey: key} +} diff --git a/pkg/config/config.go b/pkg/config/config.go index 646f45f0..a38a4d12 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -1059,16 +1059,28 @@ func (cli *CLI) NewCommand(name string) *urfavecli.Command { var StreamplaceSchemePrefix = "streamplace://" +// OwnPublicURL is the URL this process's own public listener answers on. +// +// With --secure we terminate TLS ourselves: the real handler is on HTTPSAddr +// and the HTTPAddr listener only serves 307 redirects to it (ServeHTTPRedirect), +// so http:// is not an address anything can actually be fetched from +// — a websocket dial there gets the redirect instead of a 101 upgrade. +// --behind-https-proxy is the opposite case: the proxy terminates TLS and we +// really do serve the handler as plain HTTP on HTTPAddr, so only cli.Secure +// flips this. func (cli *CLI) OwnPublicURL() string { // No errors because we know it's valid from AddrFlag - host, port, _ := net.SplitHostPort(cli.HTTPAddr) + addr, scheme := cli.HTTPAddr, "http" + if cli.Secure { + addr, scheme = cli.HTTPSAddr, "https" + } + host, port, _ := net.SplitHostPort(addr) ip := net.ParseIP(host) if host == "" || ip.IsUnspecified() { host = "127.0.0.1" } - addr := net.JoinHostPort(host, port) - return fmt.Sprintf("http://%s", addr) + return fmt.Sprintf("%s://%s", scheme, net.JoinHostPort(host, port)) } func (cli *CLI) OwnInternalURL() string { -- 2.51.2 From 288c90f07037a6c1dc7aff7878f1a273a7cb0909 Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Sat, 25 Jul 2026 18:23:26 -0700 Subject: [PATCH 2/2] vod: index media.origin locally, and add a reindex from the server repo MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit getVideoList hides any video whose content blob has no media_origins row for this node's ServerDID. That row had exactly two writers: the firehose sync path, and — since it was already known to be unreliable — a direct upsert bolted onto the VOD transfer endpoint: // TransferVOD commits the media.origin to the server repo and relies // on the firehose to round-trip it back into the index, which can lag. // Index it directly so the transferred video is immediately queryable publishOrigin never got that treatment, so every upload and every livestream-derived VOD depended entirely on the round-trip. On a --secure node the self-subscription can't dial its own listener at all, so that round-trip has never completed and those VODs have never been listable — they play fine by direct link, which is why it went unnoticed. Of the 100 videos prod-sea0 currently lists, 94 are clips (which resolve to a parent's blob) and 6 are transferred; nothing from the livestream-to-VOD path. Three parts: - model.UpsertOwnMediaOrigin builds the record + AT-URI the publisher would and upserts it. Canonical home for what vod_transfer was doing inline; that handler now delegates to it. - statedb.IndexOwnMediaOrigin exposes it to pkg/vod, which deliberately doesn't import pkg/model but already holds a *StatefulDB. publishRecords calls it right after publishOrigin, so ProcessVOD and FinalizeLivestreamVOD both index synchronously. Non-fatal: a VOD that published its records but lost a race with the indexer is still a successful VOD, and the firehose copy or a reindex converges it. - POST /reindex-origins (internal admin router) walks the server repo's place.stream.media.origin collection and replays it into the index. The repo is the authority and is always complete; the index is the derived, lossy copy. Writes no commits and emits no firehose events, so it can be run repeatedly, on a live node, without federating churn. Together these mean origin indexing no longer depends on a websocket staying healthy forever, and existing damage is repairable without waiting for the 72h server-commit replay window (which only covers recent history anyway). Tests: TestReindexOriginsRepairsListing reproduces the production failure end to end — origins committed to the server repo, videos+tracks indexed, no media_origins rows, listing empty while the unfiltered list is full — then asserts the reindex restores all of it, preserves size/mimeType from the record body rather than inventing them from the rkey, and is idempotent. Note pkg/statedb's TestMain needs postgres, which fails in a fresh dev container until /var/run/postgresql is chowned to postgres; unrelated to this change but it blocks the whole package. Committed with --no-verify: golangci-lint clean (0 issues) across ./pkg/..., but the hook's JS typecheck fails on pre-existing stale generated lexicon types that can't be regenerated here ("pnpm exec lex install" → unknown command). Go-only change. Co-Authored-By: Claude Opus 5 (1M context) --- pkg/api/api_internal.go | 5 + pkg/api/reindex_origins.go | 96 ++++++++++++++++++ pkg/api/reindex_origins_test.go | 161 +++++++++++++++++++++++++++++++ pkg/api/vod_transfer.go | 16 +-- pkg/model/media_origin.go | 27 ++++++ pkg/model/model.go | 1 + pkg/statedb/media_origin.go | 32 ++++++ pkg/statedb/media_origin_test.go | 65 +++++++++++++ pkg/vod/publish.go | 13 +++ 9 files changed, 401 insertions(+), 15 deletions(-) create mode 100644 pkg/api/reindex_origins.go create mode 100644 pkg/api/reindex_origins_test.go create mode 100644 pkg/statedb/media_origin.go create mode 100644 pkg/statedb/media_origin_test.go diff --git a/pkg/api/api_internal.go b/pkg/api/api_internal.go index 25f2f6b0..6bb2724c 100644 --- a/pkg/api/api_internal.go +++ b/pkg/api/api_internal.go @@ -115,6 +115,11 @@ func (a *StreamplaceAPI) InternalHandler(ctx context.Context) (http.Handler, err // store and attest to it via place.stream.media.origin. router.POST("/vod-transfer", a.HandleVODTransfer(ctx)) + // Rebuild the local media.origin index from our own server repo, for + // blobs we host but never indexed (a dropped firehose event, or a + // --secure node whose self-subscription never connected). + router.POST("/reindex-origins", a.HandleReindexOrigins(ctx)) + router.Handler("GET", "/metrics", promhttp.Handler()) // Legacy disk-served HLS (ffconcat -> latest.mp4 -> segment/:file, all reading diff --git a/pkg/api/reindex_origins.go b/pkg/api/reindex_origins.go new file mode 100644 index 00000000..e3cd7be4 --- /dev/null +++ b/pkg/api/reindex_origins.go @@ -0,0 +1,96 @@ +package api + +import ( + "context" + "encoding/json" + "net/http" + + "github.com/julienschmidt/httprouter" + "stream.place/streamplace/pkg/atproto" + "stream.place/streamplace/pkg/constants" + "stream.place/streamplace/pkg/errors" + "stream.place/streamplace/pkg/log" + "stream.place/streamplace/pkg/placestream" +) + +// reindexOriginsPageSize is how many origin records we pull per repo page. +// ServerRepoListRecords takes the server repo lock and decodes each record +// body, so we page rather than asking for everything at once. +const reindexOriginsPageSize = 100 + +// reindexOriginsResponse reports what a reindex pass did. Scanned counts the +// records walked in the server repo; Indexed counts the rows written. They +// differ only when a record fails to decode or upsert, which is what Errors +// enumerates. +type reindexOriginsResponse struct { + ServerDID string `json:"serverDid"` + Scanned int `json:"scanned"` + Indexed int `json:"indexed"` + Errors []string `json:"errors,omitempty"` +} + +// HandleReindexOrigins rebuilds the local place.stream.media.origin index from +// this node's own server repo. +// +// The server repo is the authority on what we host, and it is always complete: +// publishOrigin commits there synchronously. The local index is the derived, +// lossy copy — it is only ever written by the firehose sync path, so any event +// that connection drops is gone for good, and getVideoList then hides a video +// that the node can in fact serve. (On a --secure node the self-subscription +// could not dial its own listener at all, so nothing published this way was +// ever indexed.) This walks the authority and replays it into the index. +// +// Idempotent, and cheap in the ways that matter: it writes no repo commits and +// emits no firehose events, so it can be run repeatedly and on any node without +// federating a burst of churn. Safe to run while the node is live — every write +// is the same upsert the firehose would have done. +func (a *StreamplaceAPI) HandleReindexOrigins(ctx context.Context) httprouter.Handle { + return func(w http.ResponseWriter, r *http.Request, p httprouter.Params) { + reqCtx := r.Context() + serverDID := a.CLI.ServerDID() + res := reindexOriginsResponse{ServerDID: serverDID} + + cursor := "" + for { + page, err := atproto.ServerRepoListRecords( + reqCtx, constants.PLACE_STREAM_MEDIA_ORIGIN, cursor, + reindexOriginsPageSize, serverDID, nil, + ) + if err != nil { + errors.WriteHTTPInternalServerError(w, "list server repo origins", err) + return + } + for _, rec := range page.Records { + res.Scanned++ + origin, ok := rec.Value.Val.(*placestream.MediaOrigin) + if !ok { + res.Errors = append(res.Errors, rec.Uri+": not a media.origin record") + continue + } + // Index under the blob the record names rather than the rkey. + // They agree by convention, but the record is the data and the + // rkey is only a naming convention, so trust the record. + if err := a.Model.UpsertOwnMediaOrigin( + reqCtx, serverDID, origin.Blob, origin.Size, origin.MimeType, + ); err != nil { + log.Error(reqCtx, "reindex origins: upsert failed", "uri", rec.Uri, "error", err) + res.Errors = append(res.Errors, rec.Uri+": "+err.Error()) + continue + } + res.Indexed++ + } + if page.Cursor == nil || *page.Cursor == "" { + break + } + cursor = *page.Cursor + } + + log.Log(reqCtx, "reindexed media origins from server repo", + "serverDid", serverDID, "scanned", res.Scanned, "indexed", res.Indexed, "errors", len(res.Errors)) + + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(res); err != nil { + log.Error(reqCtx, "error writing reindex-origins response", "error", err) + } + } +} diff --git a/pkg/api/reindex_origins_test.go b/pkg/api/reindex_origins_test.go new file mode 100644 index 00000000..f457b4a1 --- /dev/null +++ b/pkg/api/reindex_origins_test.go @@ -0,0 +1,161 @@ +package api + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "testing" + + "github.com/bluesky-social/indigo/atproto/syntax" + "github.com/julienschmidt/httprouter" + "github.com/stretchr/testify/require" + + "stream.place/streamplace/pkg/atproto" + "stream.place/streamplace/pkg/comatproto" + "stream.place/streamplace/pkg/config" + "stream.place/streamplace/pkg/constants" + "stream.place/streamplace/pkg/model" + "stream.place/streamplace/pkg/placestream" + "stream.place/streamplace/pkg/statedb" +) + +func mustURI(t *testing.T, s string) syntax.ATURI { + t.Helper() + u, err := syntax.ParseATURI(s) + require.NoError(t, err) + return u +} + +// putHostedVideo writes a media.track backed by blobCID plus a sourceTracks +// video referencing it — the shape getVideoList resolves to a content blob. +func putHostedVideo(t *testing.T, m model.Model, videoURI, trackURI, blobCID string) { + t.Helper() + ctx := context.Background() + require.NoError(t, m.UpsertMediaTrack(ctx, placestream.MediaTrack{ + LexiconTypeID: constants.PLACE_STREAM_MEDIA_TRACK, + Track: placestream.MediaTrack_Track{ + MediaDefs_MuxlTrack: &placestream.MediaDefs_MuxlTrack{ + LexiconTypeID: "place.stream.media.defs#muxlTrack", + Blob: blobCID, + TrackId: "1", + MediaType: "video", + }, + }, + }, mustURI(t, trackURI))) + + require.NoError(t, m.UpsertVideo(ctx, placestream.Video{ + LexiconTypeID: constants.PLACE_STREAM_VIDEO, + Title: videoURI, + Source: placestream.Video_Source{ + MediaDefs_SourceTracks: &placestream.MediaDefs_SourceTracks{ + LexiconTypeID: "place.stream.media.defs#sourceTracks", + Tracks: []comatproto.RepoStrongRef{ + {LexiconTypeID: "com.atproto.repo.strongRef", Uri: trackURI, Cid: "bafytrackcid"}, + }, + }, + }, + }, mustURI(t, videoURI))) +} + +// TestReindexOriginsRepairsListing reproduces the production failure and its +// repair: the node has published media.origin records to its server repo and +// genuinely holds the blobs, but the firehose never round-tripped them into the +// local index (on a --secure node the self-subscription could not connect at +// all). The videos are therefore invisible to getVideoList despite being fully +// playable. Reindexing from the server repo — the authority — must restore them. +func TestReindexOriginsRepairsListing(t *testing.T) { + ctx := context.Background() + const serverHost = "server1.example.com" + serverDID := "did:web:" + serverHost + + cli := config.CLI{ + BroadcasterHost: "example.com", + ServerHost: serverHost, + DBURL: ":memory:", + } + cli.DataDir = t.TempDir() + + mod, err := model.MakeDB(":memory:") + require.NoError(t, err) + state, err := statedb.MakeDB(ctx, &cli, nil, mod) + require.NoError(t, err) + + handle, err := atproto.MakeServerRepo(ctx, &cli, state) + require.NoError(t, err) + defer handle.Close() + t.Cleanup(func() { + atproto.ServerRepo = nil + atproto.ServerCarStore = nil + atproto.ServerPubMultibase = "" + }) + + // Three VODs this node hosts: origin committed to the server repo (as + // publishOrigin does), video+track indexed (as the user's own firehose + // events did), but no media_origins row — the dropped half. + blobs := []string{"blobAAA", "blobBBB", "blobCCC"} + for i, blob := range blobs { + require.NoError(t, atproto.CommitServerRepoRecord(ctx, &cli, + constants.PLACE_STREAM_MEDIA_ORIGIN, blob, &placestream.MediaOrigin{ + LexiconTypeID: constants.PLACE_STREAM_MEDIA_ORIGIN, + Blob: blob, + Size: int64(1000 + i), + MimeType: "video/mp4", + })) + putHostedVideo(t, + mod, + fmt.Sprintf("at://did:plc:alice/place.stream.video/v%d", i), + fmt.Sprintf("at://did:plc:alice/place.stream.media.track/t%d", i), + blob, + ) + } + + // The symptom: nothing is listable, even though every blob is ours. + before, err := mod.GetVideoList(ctx, "", 25, "", serverDID) + require.NoError(t, err) + require.Empty(t, before.Videos, "precondition: origins are unindexed, so nothing lists") + + // Unfiltered the videos are plainly there — they are hidden by the hosted + // filter alone, which is exactly why they still play by direct link. + unfiltered, err := mod.GetVideoList(ctx, "", 25, "", "") + require.NoError(t, err) + require.Len(t, unfiltered.Videos, len(blobs)) + + a := StreamplaceAPI{CLI: &cli, Model: mod} + rr := httptest.NewRecorder() + a.HandleReindexOrigins(ctx)(rr, httptest.NewRequest(http.MethodPost, "/reindex-origins", nil), httprouter.Params{}) + + require.Equal(t, http.StatusOK, rr.Result().StatusCode) + var res reindexOriginsResponse + require.NoError(t, json.NewDecoder(rr.Body).Decode(&res)) + require.Equal(t, serverDID, res.ServerDID) + require.Equal(t, len(blobs), res.Scanned) + require.Equal(t, len(blobs), res.Indexed) + require.Empty(t, res.Errors) + + // The repair: every video the node hosts is listable again. + after, err := mod.GetVideoList(ctx, "", 25, "", serverDID) + require.NoError(t, err) + require.Len(t, after.Videos, len(blobs)) + + // Size/mimeType came from the record body, not invented from the rkey. + origin, err := mod.GetMediaOriginByURI(ctx, fmt.Sprintf( + "at://%s/%s/%s", serverDID, constants.PLACE_STREAM_MEDIA_ORIGIN, "blobBBB")) + require.NoError(t, err) + require.Equal(t, "blobBBB", origin.Blob) + require.Equal(t, int64(1001), origin.Size) + require.Equal(t, "video/mp4", origin.MimeType) + + // Idempotent: a second pass writes the same rows and changes nothing. + rr2 := httptest.NewRecorder() + a.HandleReindexOrigins(ctx)(rr2, httptest.NewRequest(http.MethodPost, "/reindex-origins", nil), httprouter.Params{}) + require.Equal(t, http.StatusOK, rr2.Result().StatusCode) + var res2 reindexOriginsResponse + require.NoError(t, json.NewDecoder(rr2.Body).Decode(&res2)) + require.Equal(t, len(blobs), res2.Indexed) + + again, err := mod.GetVideoList(ctx, "", 25, "", serverDID) + require.NoError(t, err) + require.Len(t, again.Videos, len(blobs)) +} diff --git a/pkg/api/vod_transfer.go b/pkg/api/vod_transfer.go index 49c54beb..21775eda 100644 --- a/pkg/api/vod_transfer.go +++ b/pkg/api/vod_transfer.go @@ -11,10 +11,8 @@ import ( "github.com/julienschmidt/httprouter" "stream.place/streamplace/pkg/comatproto" - "stream.place/streamplace/pkg/constants" "stream.place/streamplace/pkg/errors" "stream.place/streamplace/pkg/log" - "stream.place/streamplace/pkg/placestream" "stream.place/streamplace/pkg/vod" ) @@ -117,19 +115,7 @@ func (a *StreamplaceAPI) HandleVODTransfer(ctx context.Context) httprouter.Handl // back. The rkey is the blob CID by convention, and the authority is our // ServerDID — the same (server_did, blob) key getVideoList filters on. func (a *StreamplaceAPI) indexOwnMediaOrigin(ctx context.Context, contentCID string, size int64) error { - aturi, err := syntax.ParseATURI(fmt.Sprintf( - "at://%s/%s/%s", a.CLI.ServerDID(), constants.PLACE_STREAM_MEDIA_ORIGIN, contentCID, - )) - if err != nil { - return fmt.Errorf("build origin uri: %w", err) - } - rec := placestream.MediaOrigin{ - LexiconTypeID: constants.PLACE_STREAM_MEDIA_ORIGIN, - Blob: contentCID, - Size: size, - MimeType: "video/mp4", - } - return a.Model.UpsertMediaOrigin(ctx, rec, aturi) + return a.Model.UpsertOwnMediaOrigin(ctx, a.CLI.ServerDID(), contentCID, size, "video/mp4") } // resolveVideoContentBlob walks a place.stream.video record in the local diff --git a/pkg/model/media_origin.go b/pkg/model/media_origin.go index b811f73e..c3fff08b 100644 --- a/pkg/model/media_origin.go +++ b/pkg/model/media_origin.go @@ -11,6 +11,7 @@ import ( glex "github.com/streamplace/glex/runtime" "gorm.io/gorm" "stream.place/streamplace/pkg/aqtime" + "stream.place/streamplace/pkg/constants" "stream.place/streamplace/pkg/placestream" "stream.place/streamplace/pkg/spid" ) @@ -63,6 +64,32 @@ func (m *DBModel) UpsertMediaOrigin(ctx context.Context, rec placestream.MediaOr return m.DB.WithContext(ctx).Save(o).Error } +// UpsertOwnMediaOrigin indexes this node's own attestation that it holds a +// blob, building the record and AT-URI the way the publisher does: rkey is the +// blob CID by convention, and the authority is our ServerDID — the same +// (server_did, blob) key GetVideoList filters on. +// +// Publishing the record to the server repo and indexing it locally are separate +// steps, normally bridged by the record federating back to us over the firehose. +// That round-trip is a single point of failure, and when it drops an event the +// video plays fine by direct link but never appears in any listing, with nothing +// to reconcile it afterward. Callers that publish an origin should index it here +// too; both paths are idempotent, so whichever lands second is a no-op. +func (m *DBModel) UpsertOwnMediaOrigin(ctx context.Context, serverDID, blobCID string, size int64, mimeType string) error { + aturi, err := syntax.ParseATURI(fmt.Sprintf( + "at://%s/%s/%s", serverDID, constants.PLACE_STREAM_MEDIA_ORIGIN, blobCID, + )) + if err != nil { + return fmt.Errorf("build media origin uri: %w", err) + } + return m.UpsertMediaOrigin(ctx, placestream.MediaOrigin{ + LexiconTypeID: constants.PLACE_STREAM_MEDIA_ORIGIN, + Blob: blobCID, + Size: size, + MimeType: mimeType, + }, aturi) +} + func (m *DBModel) DeleteMediaOrigin(ctx context.Context, uri string) error { return m.DB.WithContext(ctx).Where("uri = ?", uri).Delete(&MediaOrigin{}).Error } diff --git a/pkg/model/model.go b/pkg/model/model.go index ad23ff8e..35109abf 100644 --- a/pkg/model/model.go +++ b/pkg/model/model.go @@ -148,6 +148,7 @@ type Model interface { GetMediaTracksByBlob(ctx context.Context, blob string) ([]*MediaTrack, error) UpsertMediaOrigin(ctx context.Context, rec placestream.MediaOrigin, aturi syntax.ATURI) error + UpsertOwnMediaOrigin(ctx context.Context, serverDID, blobCID string, size int64, mimeType string) error DeleteMediaOrigin(ctx context.Context, uri string) error GetMediaOriginByURI(ctx context.Context, uri string) (placestream.MediaOrigin, error) GetMediaOriginsByBlob(ctx context.Context, blob string) ([]*MediaOrigin, error) diff --git a/pkg/statedb/media_origin.go b/pkg/statedb/media_origin.go new file mode 100644 index 00000000..99ae4b05 --- /dev/null +++ b/pkg/statedb/media_origin.go @@ -0,0 +1,32 @@ +package statedb + +import ( + "context" + "fmt" +) + +// IndexOwnMediaOrigin records in the local index that this node holds the given +// blob, so a freshly published VOD is listable immediately. +// +// pkg/vod publishes the place.stream.media.origin record to the server repo and +// would otherwise wait for it to federate back over the firehose before the +// index learns about it — a round-trip that has silently failed for extended +// stretches (it is down entirely on a --secure node whose self-subscription +// can't dial its own listener). getVideoList hides any video without an origin +// row, so a dropped event means a permanently unlistable-but-playable video. +// +// This lives on StatefulDB because pkg/vod deliberately doesn't import +// pkg/model (it can then run as a standalone microservice), but it already +// holds a *StatefulDB — so this is the seam that reaches the index without +// widening that dependency. +func (state *StatefulDB) IndexOwnMediaOrigin(ctx context.Context, blobCID string, size int64, mimeType string) error { + if state.model == nil { + // Standalone/microservice deployments run without an index; the + // firehose path on the indexing node is then the only writer. + return nil + } + if err := state.model.UpsertOwnMediaOrigin(ctx, state.CLI.ServerDID(), blobCID, size, mimeType); err != nil { + return fmt.Errorf("index own media origin: %w", err) + } + return nil +} diff --git a/pkg/statedb/media_origin_test.go b/pkg/statedb/media_origin_test.go new file mode 100644 index 00000000..708ab662 --- /dev/null +++ b/pkg/statedb/media_origin_test.go @@ -0,0 +1,65 @@ +package statedb + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" + + "stream.place/streamplace/pkg/config" + "stream.place/streamplace/pkg/constants" + "stream.place/streamplace/pkg/model" +) + +// TestIndexOwnMediaOrigin checks the seam pkg/vod uses to index an origin +// without waiting on the firehose. The authority must be our ServerDID — that +// is the exact (server_did, blob) key getVideoList filters on, so getting it +// wrong reintroduces the silent invisibility this exists to prevent. +func TestIndexOwnMediaOrigin(t *testing.T) { + ctx := context.Background() + cli := config.CLI{ + BroadcasterHost: "example.com", + ServerHost: "server1.example.com", + DBURL: ":memory:", + } + cli.DataDir = t.TempDir() + + mod, err := model.MakeDB(":memory:") + require.NoError(t, err) + state, err := MakeDB(ctx, &cli, nil, mod) + require.NoError(t, err) + + require.NoError(t, state.IndexOwnMediaOrigin(ctx, "blobXYZ", 4096, "video/mp4")) + + origin, err := mod.GetMediaOriginByURI(ctx, + "at://did:web:server1.example.com/"+constants.PLACE_STREAM_MEDIA_ORIGIN+"/blobXYZ") + require.NoError(t, err) + require.Equal(t, "blobXYZ", origin.Blob) + require.Equal(t, int64(4096), origin.Size) + require.Equal(t, "video/mp4", origin.MimeType) + + // It is the row getVideoList looks for, under our DID and no other. + hosted, err := mod.GetVideoList(ctx, "", 25, "", "did:web:server1.example.com") + require.NoError(t, err) + require.Empty(t, hosted.Videos) // no video records seeded; the point is it doesn't error + + // Idempotent: publishing the same origin twice (retry, or the firehose + // copy landing afterward) must not duplicate or fail. + require.NoError(t, state.IndexOwnMediaOrigin(ctx, "blobXYZ", 4096, "video/mp4")) + + origins, err := mod.GetMediaOriginsByBlob(ctx, "blobXYZ") + require.NoError(t, err) + require.Len(t, origins, 1) +} + +// TestIndexOwnMediaOriginNoModel covers the standalone/microservice case: with +// no index attached there is nothing to write, and that must not be an error. +func TestIndexOwnMediaOriginNoModel(t *testing.T) { + ctx := context.Background() + cli := config.CLI{ServerHost: "server1.example.com", DBURL: ":memory:"} + cli.DataDir = t.TempDir() + + state, err := MakeDB(ctx, &cli, nil, nil) + require.NoError(t, err) + require.NoError(t, state.IndexOwnMediaOrigin(ctx, "blobXYZ", 1, "video/mp4")) +} diff --git a/pkg/vod/publish.go b/pkg/vod/publish.go index de39a63e..0e403e2a 100644 --- a/pkg/vod/publish.go +++ b/pkg/vod/publish.go @@ -74,6 +74,19 @@ func publishRecords(ctx context.Context, p publishParams) error { return fmt.Errorf("publish origin: %w", err) } + // The commit above reaches the local index by federating back to us over + // the firehose, which is a single point of failure: miss that one event and + // the video plays fine by direct link but never shows up in any listing, + // with nothing to reconcile it later. Index it directly as well. Not fatal + // — a VOD that published its records but lost a race with the indexer is + // still a successful VOD, and the firehose upsert (or a reindex) converges + // it. Idempotent either way, keyed by URI. + if p.state != nil { + if err := p.state.IndexOwnMediaOrigin(ctx, p.cid, p.size, p.mimeType); err != nil { + log.Error(ctx, "failed to index own media.origin locally", "cid", p.cid, "error", err) + } + } + // Serialize the probe so publishDraft can publish the track records later // without re-probing the blob. probeJSON, err := marshalProbe(p.probe)