diff --git a/.github/workflows/build.yaml b/.github/workflows/build.yaml index 7a6f946aa..1110bf42a 100644 --- a/.github/workflows/build.yaml +++ b/.github/workflows/build.yaml @@ -105,7 +105,7 @@ jobs: # and the 20260726 runner image update made that pairing fatal. sudo curl -fsSL -o /usr/bin/crun https://github.com/containers/crun/releases/download/1.28/crun-1.28-linux-amd64 sudo chmod +x /usr/bin/crun - make in-container BUILDER_TARGET=builder-no-darwin DOCKER_PWD_MOUNT_PATH=/app DOCKER_BUILD_OPTS="--layers --cache-to ghcr.io/streamplace/streamplace --cache-from ghcr.io/streamplace/streamplace" DOCKER_OPTS="-e STREAMPLACE_IGNORE_LEAKS=true -e STREAMPLACE_TEST_COUNT=1 -e CI=true -e GITHUB_ACTION=true" IN_CONTAINER_CMD="make linux-amd64 && make test-vod && make static-test && make check && make ci-lexicons" + make in-container BUILDER_TARGET=builder-no-darwin DOCKER_PWD_MOUNT_PATH=/app DOCKER_BUILD_OPTS="--layers --cache-to ghcr.io/streamplace/streamplace --cache-from ghcr.io/streamplace/streamplace" DOCKER_OPTS="-e STREAMPLACE_IGNORE_LEAKS=true -e STREAMPLACE_TEST_COUNT=1 -e CI=true -e GITHUB_ACTION=true" IN_CONTAINER_CMD="make linux-amd64 && make obs-plugin-linux-amd64 && make test-vod && make static-test && make check && make ci-lexicons" - name: Publish Test Report if: success() || failure() @@ -124,6 +124,14 @@ jobs: path: ./build-linux-amd64/streamplace retention-days: 1 + # Desktop bundles the plugin, so the upstream e2e needs it to package. + - name: Upload OBS plugin + uses: actions/upload-artifact@v4 + with: + name: obs-plugin-linux-amd64-${{ github.run_id }} + path: ./bin/obs-streamplace-*-linux-amd64.tar.gz + retention-days: 1 + # Reuse the binary built by linux-amd64 instead of provisioning another # frontend and native build just to run Playwright. e2e-web: @@ -178,6 +186,12 @@ jobs: run: | cargo install --git https://github.com/streamplace/atmoq --rev 09a604b0b00b1eb063295746de1878a258e88d5b --locked --root "$RUNNER_TEMP/atmoq-bin" atmoq echo "$RUNNER_TEMP/atmoq-bin/bin" >> "$GITHUB_PATH" + - name: Download OBS plugin + if: matrix.mode == 'upstream' + uses: actions/download-artifact@v4 + with: + name: obs-plugin-linux-amd64-${{ github.run_id }} + path: ./bin - name: Build Desktop e2e entry point if: matrix.mode == 'upstream' run: | @@ -196,6 +210,7 @@ jobs: # "Music" stem), so that ingest path and audio stems are covered from # push to web playback. - name: Run web stream e2e tests over fMP4 ingest + if: matrix.mode == 'standalone' run: hack/e2e-web-local.sh flows/stream.spec.ts flows/audio-stems.spec.ts env: diff --git a/js/desktop/README.md b/js/desktop/README.md index bf45d5c25..06f58b497 100644 --- a/js/desktop/README.md +++ b/js/desktop/README.md @@ -22,32 +22,39 @@ last package run unpacked into `out/obs-plugin/`. Unit tests for the OBS plugin installer and profile writer. -## Upstream station mode - -Desktop starts its bundled node with `SP_UPSTREAM_STATION=did:web:stream.place`. -Set `SP_UPSTREAM_STATION` to another station DID to override it, or explicitly -set it to an empty string to run the standalone server mode. - -The station must advertise both `#streamplace` (an HTTPS origin) and `#atmoq` -(a raw-QUIC MoQ origin) in its DID document. Missing or invalid advertisements -stop startup; Desktop never falls back to the full public firehose. - -The local node subscribes to exact-DID tracks only for successfully authenticated -local users. Startup, sweeps, history backfill and event handling use the same -persisted OAuth eligibility; previously indexed repos do not authorize new -scraping. Logout removes eligibility when the last local grant ends. Existing -indexed records remain on disk, but no longer receive updates. -After subscribing or reconnecting, the node queues a head check and shallow -repair for that DID behind any active sweep. Reconnect bursts coalesce into one -pending repair per DID; a gap during an active repair retains one follow-up. -Sparse tracks do not depend on firehose replay or wait for the next periodic -sweep to recover missed writes. - -OAuth has three independent DPoP sessions: frontend → loopback node → HTTPS -station → PDS. Only the station needs a public HTTPS callback and retains the -long-lived PDS session. The local node's issuer and frontend callback use its -actual loopback port; `SP_HTTP_ADDR` and `SP_HTTP_INTERNAL_ADDR` can override -the Desktop listener defaults. +## Single-user node + +Desktop's bundled node is a single-user AppView: it runs the same indexing as +any node, but only for the accounts logged into it. It starts with +`SP_UPSTREAM_STATION=did:web:stream.place` and +`SP_RELAY_HOST=moqt://streamplace.network`. Set `SP_UPSTREAM_STATION` to +another station DID to override it, or explicitly to an empty string to run an +ordinary standalone node (which then keeps the default relays). + +With a station: + +- **Firehose.** The node subscribes to one per-DID track per logged-in + account on each `--relay-host`, which must all be MoQ relays (atmoq serves a + per-DID track on demand). It never takes the aggregate firehose, and its own + PDS firehose is left out. +- **Indexing.** Startup, sweeps, history backfill and event handling use the + same persisted OAuth eligibility; previously indexed repos do not authorize + new scraping, and no other account's identity is resolved. Logout removes + eligibility when the last local grant ends. Existing indexed records remain + on disk, but no longer receive updates. After subscribing or reconnecting, + the node queues a head check and shallow repair for that DID behind any + active sweep. Reconnect bursts coalesce into one pending repair per DID; a + gap during an active repair retains one follow-up. +- **OAuth.** Three independent DPoP sessions: frontend → loopback node → HTTPS + station → PDS. Only the station needs a public HTTPS callback and retains the + long-lived PDS session, so logins last as long as a confidential client's. + The station is the `#streamplace` HTTPS origin in its DID document; startup + stops if it is missing. The local node's issuer and frontend callback use its + actual loopback port; `SP_HTTP_ADDR` and `SP_HTTP_INTERNAL_ADDR` can override + the Desktop listener defaults. +- **Streams.** Unless `SP_PEERS` is set, the node pushes the streams it + ingests to the station over MoQ (see `docs/moq.md`). +- **Labelers** are not followed: their firehoses carry other accounts. For development with the embedded frontend, set `AQD_NODE_FRONTEND=true` and `SP_DEV_FRONTEND_PROXY=` after `make app && make dev`. diff --git a/js/desktop/src/node.ts b/js/desktop/src/node.ts index 24f82ff07..3bad33b78 100644 --- a/js/desktop/src/node.ts +++ b/js/desktop/src/node.ts @@ -46,20 +46,30 @@ export default async function makeNode(opts: { opts.env.SP_HTTP_INTERNAL_ADDR ?? process.env.SP_HTTP_INTERNAL_ADDR ?? "127.0.0.1:39092"; + // A single-user node: it indexes only the accounts logged into it, signs + // them in and pushes their streams through the station, and takes their + // firehose events from per-DID tracks on the station's MoQ relay. An empty + // SP_UPSTREAM_STATION runs an ordinary standalone node instead. + const upstreamStation = + opts.env.SP_UPSTREAM_STATION ?? + process.env.SP_UPSTREAM_STATION ?? + "did:web:stream.place"; + const relayHost = + opts.env.SP_RELAY_HOST ?? + process.env.SP_RELAY_HOST ?? + (upstreamStation ? "moqt://streamplace.network" : undefined); const proc = spawn(exe, [], { stdio: "inherit", env: { ...process.env, - SP_UPSTREAM_STATION: - opts.env.SP_UPSTREAM_STATION ?? - process.env.SP_UPSTREAM_STATION ?? - "did:web:stream.place", + SP_UPSTREAM_STATION: upstreamStation, + ...(relayHost !== undefined && { SP_RELAY_HOST: relayHost }), SP_HTTP_ADDR: addr, SP_HTTP_INTERNAL_ADDR: internalAddr, - // Every listener on loopback: the node takes any valid stream key, so - // it must only hear this machine's encoders. - SP_RTMP_ADDR: "127.0.0.1:1935", - SP_MOQ_ADDR: "127.0.0.1:38443", + // Every listener on loopback by default: the node takes any valid + // stream key, so it must only hear this machine's encoders. + SP_RTMP_ADDR: process.env.SP_RTMP_ADDR ?? "127.0.0.1:1935", + SP_MOQ_ADDR: process.env.SP_MOQ_ADDR ?? "127.0.0.1:38443", ...opts.env, }, windowsHide: true, diff --git a/js/docs/src/content/docs/guides/installing/downloading-streamplace.md b/js/docs/src/content/docs/guides/installing/downloading-streamplace.md index 115789158..e7d14d04b 100644 --- a/js/docs/src/content/docs/guides/installing/downloading-streamplace.md +++ b/js/docs/src/content/docs/guides/installing/downloading-streamplace.md @@ -92,44 +92,37 @@ host networking. So that command would look something like: Binaries for all platforms are available to download from [our GitLab server](https://git.stream.place/streamplace/streamplace/-/releases). -## Local node with an upstream station +## Single-user node with an upstream station ```shell streamplace --http-addr=127.0.0.1:38080 \ - --upstream-station=did:web:stream.place + --upstream-station=did:web:stream.place \ + --relay-host=moqt://streamplace.network ``` -`SP_UPSTREAM_STATION` is the environment equivalent. Without it, normal server -indexing and relay selection remain unchanged. With it, the node discovers the -station's HTTPS OAuth origin and MoQ firehose from its DID document. There is no -fallback to `bsky.network` or the local PDS firehose. - -The local node indexes only users with a successfully completed, non-revoked -local OAuth grant. Pending logins, unrelated followed users and repos left over -from a previous standalone run do not permit PDS scraping. Per-DID subscriptions -start at login and end when the last eligible grant ends. Reconnects reconcile -the same scoped users with their actual PDS; historical records remain on disk -after logout. - -Local HTTP OAuth requires a loopback listener. The local node delegates to the -station using its own DPoP key; the station independently authenticates to the -PDS using its public HTTPS callback. No public callback or development-only -OAuth flag is needed on the local node. - -### Preparing an upstream station - -Run an `atmoq serve` bridge against the station's upstream atproto relay, with -a valid TLS certificate and reachable QUIC UDP port. Advertise its origin: - -```ini -SP_ATMOQ_RELAY=moqt://firehose.example.com:443 -``` - -The station's DID document then includes an `#atmoq` service of type -`AtprotoMoQRelay`, alongside the existing `#streamplace` HTTPS origin. Clients -fail startup if either advertisement is absent or malformed. Use a DNS name -matching the certificate: released atmoq v0.1.1 requires TLS SNI for raw MoQ, -which Go does not send for numeric-IP URLs. - -Deploy the station's chained-OAuth support and MoQ advertisement before -releasing Desktop clients that default to `did:web:stream.place`. +`SP_UPSTREAM_STATION` and `SP_RELAY_HOST` are the environment equivalents. +Without a station, normal server indexing and relay selection remain unchanged. +With one, the node is a single-user AppView: it runs the usual indexing, but +only for the accounts logged into it. + +- It takes each logged-in account's events from that account's per-DID track + on every `--relay-host`, which must all be MoQ relays: an + [atmoq](https://github.com/streamplace/atmoq) relay serves a per-DID track on + demand. There is no aggregate firehose, no `bsky.network` and no local PDS + firehose. +- It indexes only users with a successfully completed, non-revoked local OAuth + grant. Pending logins, unrelated followed users and repos left over from a + previous standalone run do not permit PDS scraping. Per-DID subscriptions + start at login and end when the last eligible grant ends. Reconnects + reconcile the same scoped users with their actual PDS; historical records + remain on disk after logout. +- It signs users in through the station: the `#streamplace` HTTPS origin in + the station's DID document. The local node delegates to the station using + its own DPoP key; the station independently authenticates to the PDS using + its public HTTPS callback. Local HTTP OAuth requires a loopback listener; no + public callback or development-only OAuth flag is needed on the local node. +- It pushes the streams it ingests to the station over MoQ, unless `--peers` + names other nodes. + +Deploy the station's chained-OAuth support before releasing Desktop clients +that default to `did:web:stream.place`. diff --git a/js/e2e-web/README.md b/js/e2e-web/README.md index e8f8afb42..e2d1590a5 100644 --- a/js/e2e-web/README.md +++ b/js/e2e-web/README.md @@ -175,7 +175,7 @@ SP_E2E_UPSTREAM_STATION=true SP_E2E_ATMOQ_BIN=/path/to/atmoq \ The harness starts the existing PDS/PLC, an HTTPS upstream station, a separate HTTP loopback node configured with `SP_UPSTREAM_STATION=did:web:`, and `atmoq serve` bridging the PDS's actual WebSocket firehose to QUIC. -The station advertises the bridge through `SP_ATMOQ_RELAY`. QUIC verifies the +The loopback node takes the bridge as its `SP_RELAY_HOST`. QUIC verifies the throwaway CA and the leaf's loopback IP SAN; there is no insecure TLS override, mock frame source or fake protocol server. The bridge uses a one-frame group for sparse per-user traffic. URLs printed as evidence are `SERVER_URL`, diff --git a/js/e2e-web/flows/desktop-upstream-station.spec.ts b/js/e2e-web/flows/desktop-upstream-station.spec.ts index cbd93d4c0..756e33f19 100644 --- a/js/e2e-web/flows/desktop-upstream-station.spec.ts +++ b/js/e2e-web/flows/desktop-upstream-station.spec.ts @@ -62,6 +62,7 @@ test("desktop-upstream-station: login and writes survive closing Desktop", async AQD_NODE_FRONTEND: "true", AQD_NO_UPDATE: "true", SP_UPSTREAM_STATION: `did:web:${new URL(process.env.UPSTREAM_STATION_URL!).hostname}`, + SP_RELAY_HOST: process.env.ATMOQ_RELAY_URL!, SP_HTTP_ADDR: localHost, SP_HTTP_INTERNAL_ADDR: `127.0.0.1:${await freePort()}`, SP_RTMP_ADDR: `127.0.0.1:${await freePort()}`, @@ -105,7 +106,7 @@ test("desktop-upstream-station: login and writes survive closing Desktop", async try { app = await launch(); let page = await app.firstWindow(); - await expect(page).toHaveURL(nodeURL + "/", { timeout: 60_000 }); + await expect(page).toHaveURL(nodeURL + "/desktop", { timeout: 60_000 }); await loginThroughPds(page, nodeURL); await page.goto(nodeURL + "/settings/privacy"); const toggleName = `Allow ${localHost} to record your livestream for debugging and improving the service`; diff --git a/pkg/api/feeds.go b/pkg/api/feeds.go index 814b36dca..f00e5ff14 100644 --- a/pkg/api/feeds.go +++ b/pkg/api/feeds.go @@ -22,7 +22,7 @@ func (a *StreamplaceAPI) HandleDidJSON(ctx context.Context) httprouter.Handle { host = a.CLI.BroadcasterHost pubKey = atproto.LexiconPubMultibase } - didJSON := atproto.DIDDoc(host, pubKey, a.CLI.AtmoqRelay) + didJSON := atproto.DIDDoc(host, pubKey) w.WriteHeader(200) w.Header().Set("Content-Type", "application/json") bs, err := json.Marshal(didJSON) diff --git a/pkg/atproto/atproto.go b/pkg/atproto/atproto.go index bcb790b9d..93e5bbfe1 100644 --- a/pkg/atproto/atproto.go +++ b/pkg/atproto/atproto.go @@ -482,31 +482,7 @@ func (atsync *ATProtoSynchronizer) IdentityDirectory() identity.Directory { return atsync.directory(true) } -func DIDDoc(host string, pubMultibase string, atmoqRelay string) map[string]any { - services := []map[string]any{ - { - "id": "#bsky_fg", - "type": "BskyFeedGenerator", - "serviceEndpoint": fmt.Sprintf("https://%s", host), - }, - { - "id": "#atproto_pds", - "type": "AtprotoPersonalDataServer", - "serviceEndpoint": fmt.Sprintf("https://%s", host), - }, - { - "id": "#" + StreamplaceServiceID, - "type": "StreamplaceServer", - "serviceEndpoint": fmt.Sprintf("https://%s", host), - }, - } - if atmoqRelay != "" { - services = append(services, map[string]any{ - "id": "#" + AtmoqServiceID, - "type": "AtprotoMoQRelay", - "serviceEndpoint": atmoqRelay, - }) - } +func DIDDoc(host string, pubMultibase string) map[string]any { return map[string]any{ "@context": []string{ "https://www.w3.org/ns/did/v1", @@ -515,7 +491,23 @@ func DIDDoc(host string, pubMultibase string, atmoqRelay string) map[string]any }, "id": fmt.Sprintf("did:web:%s", host), "alsoKnownAs": []string{}, - "service": services, + "service": []map[string]any{ + { + "id": "#bsky_fg", + "type": "BskyFeedGenerator", + "serviceEndpoint": fmt.Sprintf("https://%s", host), + }, + { + "id": "#atproto_pds", + "type": "AtprotoPersonalDataServer", + "serviceEndpoint": fmt.Sprintf("https://%s", host), + }, + { + "id": "#" + StreamplaceServiceID, + "type": "StreamplaceServer", + "serviceEndpoint": fmt.Sprintf("https://%s", host), + }, + }, "verificationMethod": []map[string]any{ { "id": fmt.Sprintf("did:web:%s#atproto", host), diff --git a/pkg/atproto/firehose.go b/pkg/atproto/firehose.go index beeabaa31..557214adc 100644 --- a/pkg/atproto/firehose.go +++ b/pkg/atproto/firehose.go @@ -58,7 +58,6 @@ type ATProtoSynchronizer struct { CachedPLCDirectory identity.Directory dirMu sync.Mutex OATProxy *oatproxy.OATProxy - UpstreamStation *UpstreamStation // firehose liveness, written from every relay consumer concurrently // (unix nanos). @@ -98,14 +97,9 @@ func sinceNanos(ns int64) time.Duration { // records published to our built-in PDS are always indexed immediately (rather // than waiting for an external relay to crawl them back to us). We only add // ourselves when we actually have an HTTP listener to dial — i.e. never in -// tests that run without the server. +// tests that run without the server — and never with an upstream station, +// whose node's users publish to their own PDSes rather than ours. func (atsync *ATProtoSynchronizer) relayHosts() []string { - if atsync.CLI.UpstreamStation != "" { - if atsync.UpstreamStation == nil { - return nil - } - return []string{atsync.UpstreamStation.FirehoseURL} - } seen := map[string]struct{}{} var hosts []string add := func(h string) { @@ -122,7 +116,7 @@ func (atsync *ATProtoSynchronizer) relayHosts() []string { for _, h := range strings.Split(atsync.CLI.RelayHost, ",") { add(h) } - if self := atsync.selfRelayURL(); self != "" { + if self := atsync.selfRelayURL(); self != "" && !atsync.upstreamMode() { add(self) } return hosts @@ -168,7 +162,7 @@ func (atsync *ATProtoSynchronizer) StartFirehose(ctx context.Context) error { atsync.markSeen() atsync.markEvent(time.Now()) if atsync.upstreamMode() { - return atsync.startUpstreamFirehose(ctx) + return atsync.startUpstreamFirehose(ctx, relays) } log.Log(ctx, "starting firehose consumers", "relays", relays) diff --git a/pkg/atproto/indexing_scope.go b/pkg/atproto/indexing_scope.go index 280a32895..204decf6e 100644 --- a/pkg/atproto/indexing_scope.go +++ b/pkg/atproto/indexing_scope.go @@ -14,7 +14,7 @@ import ( var ErrRepoNotLocal = errors.New("repo has no authenticated local session") func (atsync *ATProtoSynchronizer) upstreamMode() bool { - return atsync.UpstreamStation != nil || (atsync.CLI != nil && atsync.CLI.UpstreamStation != "") + return atsync.CLI != nil && atsync.CLI.UpstreamStation != "" } // indexingDID authorizes before identity resolution or a cache hit. Resolve diff --git a/pkg/atproto/upstream_firehose.go b/pkg/atproto/upstream_firehose.go index caa093768..9aa55e5f3 100644 --- a/pkg/atproto/upstream_firehose.go +++ b/pkg/atproto/upstream_firehose.go @@ -12,14 +12,14 @@ import ( const OAuthSessionTopic = "atproto:oauth-sessions" -// The set of users comes from persisted successful local logins. The map here -// owns only live subscriptions, never authorization or an independent allowlist. -func (atsync *ATProtoSynchronizer) startUpstreamFirehose(ctx context.Context) error { - if atsync.UpstreamStation == nil { - return fmt.Errorf("upstream station has not been resolved") +// startUpstreamFirehose subscribes to each logged-in user's per-DID track on +// every relay. The set of users comes from persisted successful local logins; +// the map here owns only live subscriptions, never authorization or an +// independent allowlist. +func (atsync *ATProtoSynchronizer) startUpstreamFirehose(ctx context.Context, relays []string) error { + for _, relay := range relays { + spmetrics.TouchRelay(relay, relayProtocol(relay)) } - relay := atsync.UpstreamStation.FirehoseURL - spmetrics.TouchRelay(relay, "moq") changes := atsync.Bus.Subscribe(OAuthSessionTopic) defer atsync.Bus.Unsubscribe(OAuthSessionTopic, changes) // Login and revocation wake immediately; the timer also observes session @@ -55,12 +55,14 @@ func (atsync *ATProtoSynchronizer) startUpstreamFirehose(ctx context.Context) er } userCtx, cancel := context.WithCancel(ctx) active[did] = cancel - workers.Add(1) - go func() { - defer workers.Done() - atsync.consumeRelay(userCtx, relay, did) - }() - log.Log(ctx, "subscribing to upstream user", "did", did, "relay", relay) + for _, relay := range relays { + workers.Add(1) + go func() { + defer workers.Done() + atsync.consumeRelay(userCtx, relay, did) + }() + } + log.Log(ctx, "subscribing to upstream user", "did", did, "relays", relays) } select { case <-ctx.Done(): diff --git a/pkg/atproto/upstream_station.go b/pkg/atproto/upstream_station.go index a8ce300e1..aa7636fdc 100644 --- a/pkg/atproto/upstream_station.go +++ b/pkg/atproto/upstream_station.go @@ -6,16 +6,12 @@ import ( "net/url" "github.com/bluesky-social/indigo/atproto/syntax" - "stream.place/streamplace/pkg/config" ) -const AtmoqServiceID = "atmoq" - // UpstreamStation is resolved once before any relay or OAuth session is opened. type UpstreamStation struct { - DID string - URL string - FirehoseURL string + DID string + URL string } // ResolveUpstreamStation looks the station up through plcURL's directory, the @@ -35,13 +31,8 @@ func ResolveUpstreamStation(ctx context.Context, plcURL, did string) (*UpstreamS if err != nil || u.Scheme != "https" || u.Hostname() == "" || u.User != nil || u.RawQuery != "" || u.Fragment != "" || (u.Path != "" && u.Path != "/") { return nil, fmt.Errorf("upstream station %s must advertise an HTTPS origin in #%s", did, StreamplaceServiceID) } - firehoseURL := ident.GetServiceEndpoint(AtmoqServiceID) - if err := config.ValidateMoQOrigin(firehoseURL); err != nil { - return nil, fmt.Errorf("upstream station %s service #%s: %w", did, AtmoqServiceID, err) - } return &UpstreamStation{ - DID: did, - URL: u.Scheme + "://" + u.Host, - FirehoseURL: firehoseURL, + DID: did, + URL: u.Scheme + "://" + u.Host, }, nil } diff --git a/pkg/atproto/upstream_station_test.go b/pkg/atproto/upstream_station_test.go index 09f51edc2..09de941b7 100644 --- a/pkg/atproto/upstream_station_test.go +++ b/pkg/atproto/upstream_station_test.go @@ -15,26 +15,24 @@ import ( ) func TestResolveUpstreamStation(t *testing.T) { - var stationURL, firehoseURL string - var advertiseRelay = true + stationEndpoint := "https://station.streamplace.team" station := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.URL.Path != "/.well-known/did.json" { t.Errorf("unexpected discovery request: %s", r.URL.Path) http.NotFound(w, r) return } - relay := firehoseURL - if !advertiseRelay { - relay = "" + doc := DIDDoc("station.streamplace.team", "zQ3shRvJrnKp4SWvoxkFyZqijahHcmvJLdViKzkwQKj7cWxVm") + for _, svc := range doc["service"].([]map[string]any) { + if svc["id"] == "#"+StreamplaceServiceID { + svc["serviceEndpoint"] = stationEndpoint + } } - doc := DIDDoc("station.streamplace.team", "zQ3shRvJrnKp4SWvoxkFyZqijahHcmvJLdViKzkwQKj7cWxVm", relay) if err := json.NewEncoder(w).Encode(doc); err != nil { t.Error(err) } })) defer station.Close() - stationURL = "https://station.streamplace.team" - firehoseURL = "moqt://127.0.0.1:4443" original := aqhttp.Client client := station.Client() transport := client.Transport.(*http.Transport).Clone() @@ -51,26 +49,23 @@ func TestResolveUpstreamStation(t *testing.T) { resolved, err := ResolveUpstreamStation(context.Background(), "", did) require.NoError(t, err) - require.Equal(t, stationURL, resolved.URL) - require.Equal(t, firehoseURL, resolved.FirehoseURL) + require.Equal(t, "https://station.streamplace.team", resolved.URL) - // A missing/bad advertisement must not quietly select the global firehose. - advertiseRelay = false - _, err = ResolveUpstreamStation(context.Background(), "", did) - require.Error(t, err) - advertiseRelay = true - firehoseURL = "wss://bsky.network" - _, err = ResolveUpstreamStation(context.Background(), "", did) - require.Error(t, err) + // OAuth is delegated to this origin, so anything but a bare HTTPS origin + // stops startup. + for _, endpoint := range []string{"http://station.streamplace.team", "https://station.streamplace.team/oauth", ""} { + stationEndpoint = endpoint + _, err = ResolveUpstreamStation(context.Background(), "", did) + require.Error(t, err, endpoint) + } } -func TestUpstreamRelayNeverFallsBack(t *testing.T) { - atsync := &ATProtoSynchronizer{CLI: &config.CLI{ - UpstreamStation: "did:web:station.example", - RelayHost: "wss://bsky.network", - HTTPAddr: "127.0.0.1:38080", - }} - require.Empty(t, atsync.relayHosts()) - atsync.UpstreamStation = &UpstreamStation{FirehoseURL: "moqt://station.example"} - require.Equal(t, []string{"moqt://station.example"}, atsync.relayHosts()) +// With an upstream station the node's own PDS firehose is left out: its +// users' records live on their own PDSes, reached through the MoQ relay. +func TestUpstreamRelayHosts(t *testing.T) { + cli := &config.CLI{RelayHost: "moqt://streamplace.network", HTTPAddr: "127.0.0.1:38080"} + atsync := &ATProtoSynchronizer{CLI: cli} + require.Equal(t, []string{"moqt://streamplace.network", "ws://127.0.0.1:38080"}, atsync.relayHosts()) + cli.UpstreamStation = "did:web:station.example" + require.Equal(t, []string{"moqt://streamplace.network"}, atsync.relayHosts()) } diff --git a/pkg/cmd/e2e.go b/pkg/cmd/e2e.go index e4e804a44..d085fc0e5 100644 --- a/pkg/cmd/e2e.go +++ b/pkg/cmd/e2e.go @@ -391,9 +391,6 @@ func runE2E(ctx context.Context, devEnvPath, httpsPDSHost, httpsStationHost stri if tlsEnv != nil { nodeEnv = append(nodeEnv, tlsEnv.NodeEnv()...) } - if upstream { - nodeEnv = append(nodeEnv, "SP_ATMOQ_RELAY="+bridgeURL) - } if appBundleID != "" { nodeEnv = append(nodeEnv, "SP_APP_BUNDLE_ID="+appBundleID) } @@ -489,7 +486,7 @@ func runE2E(ctx context.Context, devEnvPath, httpsPDSHost, httpsStationHost stri var local *e2eLocalNode var unrelated *e2eUnrelatedAccount if upstream { - local, err = startE2ELocalNode(ctx, self, tlsEnv, env, out.Did, appBundleID) + local, err = startE2ELocalNode(ctx, self, tlsEnv, env, out.Did, appBundleID, bridgeURL) if err != nil { return err } diff --git a/pkg/cmd/e2e_upstream.go b/pkg/cmd/e2e_upstream.go index d5f4bf8e1..e59eb292e 100644 --- a/pkg/cmd/e2e_upstream.go +++ b/pkg/cmd/e2e_upstream.go @@ -202,7 +202,7 @@ func (n *e2eLocalNode) Control(ctx context.Context, did string) { } } -func startE2ELocalNode(ctx context.Context, self string, tlsEnv *e2eHTTPS, env e2eDevEnv, did, appBundleID string) (*e2eLocalNode, error) { +func startE2ELocalNode(ctx context.Context, self string, tlsEnv *e2eHTTPS, env e2eDevEnv, did, appBundleID, relayURL string) (*e2eLocalNode, error) { ports := make([]int, 3) for i := range ports { port, err := freePort() @@ -237,6 +237,7 @@ func startE2ELocalNode(ctx context.Context, self string, tlsEnv *e2eHTTPS, env e "SP_WEBSOCKET_URL=ws://"+n.addr, "SP_PLC_URL="+env.PLCURL, "SP_UPSTREAM_STATION=did:web:"+tlsEnv.stationHost, + "SP_RELAY_HOST="+relayURL, "SP_TRUST_PRIVATE_NETWORK=true", "SP_ADMIN_DIDS="+did, "SP_STREAM_SESSION_TIMEOUT=30s", diff --git a/pkg/cmd/streamplace.go b/pkg/cmd/streamplace.go index 8fe0ed915..9c95f2eb0 100644 --- a/pkg/cmd/streamplace.go +++ b/pkg/cmd/streamplace.go @@ -285,15 +285,19 @@ func runMain(ctx context.Context, build *config.BuildFlags, platformJobs []jobFu if err != nil { return err } - log.Log(ctx, "using upstream station", "did", upstreamStation.DID, "url", upstreamStation.URL, "firehose", upstreamStation.FirehoseURL) + // The station is where this node's streams go out from: push them + // there, as a node behind NAT has to (see docs/moq.md). + if len(cli.Peers) == 0 { + cli.Peers = []string{upstreamStation.URL} + } + log.Log(ctx, "using upstream station", "did", upstreamStation.DID, "url", upstreamStation.URL, "relays", cli.RelayHost, "peers", cli.Peers) } atsync := &atproto.ATProtoSynchronizer{ - CLI: cli, - Model: mod, - StatefulDB: state, - Noter: noter, - Bus: b, - UpstreamStation: upstreamStation, + CLI: cli, + Model: mod, + StatefulDB: state, + Noter: noter, + Bus: b, } mm, err := media.MakeMediaManager(ctx, cli, signer, mod, b, atsync, ldb) if err != nil { diff --git a/pkg/config/config.go b/pkg/config/config.go index 850770567..cac36e5ad 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -105,7 +105,6 @@ type CLI struct { StreamerName string RelayHost string UpstreamStation string - AtmoqRelay string Debug map[string]map[string]int AllowedStreams []string WideOpen bool @@ -689,16 +688,10 @@ func (cli *CLI) NewCommand(name string) *urfavecli.Command { }, &urfavecli.StringFlag{ Name: "upstream-station", - Usage: "DID of an upstream Streamplace station; consume its MoQ firehose, delegate OAuth to it, and scrape only locally authenticated users", + Usage: "DID of an upstream Streamplace station, for a node that indexes only its own logged-in users: OAuth goes through the station, streams are pushed to it (unless --peers is set), and those users' firehose events come from per-DID tracks on the MoQ --relay-host", Destination: &cli.UpstreamStation, Sources: urfavecli.EnvVars("SP_UPSTREAM_STATION"), }, - &urfavecli.StringFlag{ - Name: "atmoq-relay", - Usage: "MoQ firehose URL to advertise in this station's DID document for upstream-station clients", - Destination: &cli.AtmoqRelay, - Sources: urfavecli.EnvVars("SP_ATMOQ_RELAY"), - }, &urfavecli.StringFlag{ Name: "color", Usage: "'true' to enable colorized logging, 'false' to disable", @@ -1567,6 +1560,13 @@ func (cli *CLI) CheckConfig() error { if _, err := syntax.ParseDID(cli.UpstreamStation); err != nil { errs = append(errs, fmt.Errorf("--upstream-station: %w", err)) } + // Only a MoQ relay can serve one account's events (a per-DID track); + // a WebSocket relay would hand this node every account's. + for _, relay := range strings.Split(cli.RelayHost, ",") { + if err := ValidateMoQOrigin(strings.TrimSpace(relay)); err != nil { + errs = append(errs, fmt.Errorf("--upstream-station needs MoQ --relay-host entries: %w", err)) + } + } } if cli.UpstreamStation != "" && !cli.Secure { // The advertised URL maps a wildcard bind (":38080") to 127.0.0.1, @@ -1582,11 +1582,6 @@ func (cli *CLI) CheckConfig() error { errs = append(errs, fmt.Errorf("local OAuth requires a loopback --http-addr or --secure")) } } - if cli.AtmoqRelay != "" { - if err := ValidateMoQOrigin(cli.AtmoqRelay); err != nil { - errs = append(errs, fmt.Errorf("--atmoq-relay: %w", err)) - } - } return errors.Join(errs...) } diff --git a/pkg/config/config_check_test.go b/pkg/config/config_check_test.go index ea83923c1..c01bc3dad 100644 --- a/pkg/config/config_check_test.go +++ b/pkg/config/config_check_test.go @@ -39,15 +39,19 @@ func TestCheckConfig(t *testing.T) { bad(CLI{DataDir: t.TempDir(), DuplicateMistTest: true, RTMPServerAddon: "127.0.0.1:31935"}, "--secure") ok(CLI{DataDir: t.TempDir(), DuplicateMistTest: true, Secure: true, RTMPServerAddon: "127.0.0.1:31935"}) - bad(CLI{DataDir: t.TempDir(), UpstreamStation: "did:web:station.test", HTTPAddr: "192.0.2.1:38080"}, + const moqRelay = "moqt://streamplace.network" + bad(CLI{DataDir: t.TempDir(), UpstreamStation: "did:web:station.test", RelayHost: moqRelay, HTTPAddr: "192.0.2.1:38080"}, "loopback --http-addr") - ok(CLI{DataDir: t.TempDir(), UpstreamStation: "did:web:station.test", HTTPAddr: "localhost:38080"}) + ok(CLI{DataDir: t.TempDir(), UpstreamStation: "did:web:station.test", RelayHost: moqRelay, HTTPAddr: "localhost:38080"}) // A wildcard bind serves the plain-HTTP OAuth endpoints to the network, // though the advertised URL is loopback. for _, addr := range []string{":38080", "0.0.0.0:38080", "[::]:38080"} { - bad(CLI{DataDir: t.TempDir(), UpstreamStation: "did:web:station.test", HTTPAddr: addr}, + bad(CLI{DataDir: t.TempDir(), UpstreamStation: "did:web:station.test", RelayHost: moqRelay, HTTPAddr: addr}, "loopback --http-addr") } + // A WebSocket relay, the default, would carry every account's events. + bad(CLI{DataDir: t.TempDir(), UpstreamStation: "did:web:station.test", RelayHost: moqRelay + ",wss://bsky.network", HTTPAddr: "127.0.0.1:38080"}, + "MoQ --relay-host") bad(CLI{}, "data-dir") bad(CLI{DataDir: t.TempDir(), LivepeerGateway: true, LivepeerGatewayURL: "http://127.0.0.1:8935"}, "livepeer-gateway") diff --git a/pkg/spxrpc/com_atproto_repo.go b/pkg/spxrpc/com_atproto_repo.go index 57fdca439..5fa9127d3 100644 --- a/pkg/spxrpc/com_atproto_repo.go +++ b/pkg/spxrpc/com_atproto_repo.go @@ -97,7 +97,7 @@ func (s *Server) handleComAtprotoRepoDescribeRepo(ctx context.Context, repo stri if err != nil { return nil, fmt.Errorf("list server repo collections: %w", err) } - didDoc, err := glex.Unknown(atproto.DIDDoc(s.cli.ServerHost, atproto.ServerPubMultibase, s.cli.AtmoqRelay)) + didDoc, err := glex.Unknown(atproto.DIDDoc(s.cli.ServerHost, atproto.ServerPubMultibase)) if err != nil { return nil, fmt.Errorf("marshal did doc: %w", err) } @@ -110,7 +110,7 @@ func (s *Server) handleComAtprotoRepoDescribeRepo(ctx context.Context, repo stri }, nil } - didDoc, err := glex.Unknown(atproto.DIDDoc(s.cli.BroadcasterHost, atproto.LexiconPubMultibase, s.cli.AtmoqRelay)) + didDoc, err := glex.Unknown(atproto.DIDDoc(s.cli.BroadcasterHost, atproto.LexiconPubMultibase)) if err != nil { return nil, fmt.Errorf("marshal did doc: %w", err) } diff --git a/pkg/spxrpc/pds_auth_test.go b/pkg/spxrpc/pds_auth_test.go index 029768f20..4aca77c53 100644 --- a/pkg/spxrpc/pds_auth_test.go +++ b/pkg/spxrpc/pds_auth_test.go @@ -425,7 +425,7 @@ func TestAuthMiddlewareStack(t *testing.T) { } func TestDIDDocAdvertisesStreamplaceService(t *testing.T) { - doc := atproto.DIDDoc(pdsAuthTestHost, "zQ3shRvJrnKp4SWvoxkFyZqijahHcmvJLdViKzkwQKj7cWxVm", "") + doc := atproto.DIDDoc(pdsAuthTestHost, "zQ3shRvJrnKp4SWvoxkFyZqijahHcmvJLdViKzkwQKj7cWxVm") services, ok := doc["service"].([]map[string]any) require.True(t, ok) require.Contains(t, services, map[string]any{