From b8029da3287b41ccc6c4366e1f8ae054afd2fbe9 Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Mon, 19 Oct 2026 16:01:38 -0700 Subject: [PATCH] Single-user node: per-DID tracks from the MoQ relay, push to the station A node with --upstream-station is a single-user AppView: the usual indexing, scoped to its logged-in accounts. Its firehose now comes from --relay-host rather than an #atmoq service in the station's DID document (did:web:stream.place advertises none): each account's per-DID track on every relay, which must all be MoQ, and never the node's own PDS firehose. --atmoq-relay goes with the advertisement. Without --peers the node pushes its streams to the station, the way a node behind NAT has to. Desktop starts its node against moqt://streamplace.network, and its listener addresses can be overridden like the HTTP ones. The upstream CI job gets the OBS plugin it now needs to package Desktop, and runs only the upstream specs; the Desktop spec expects the /desktop home screen. --- .github/workflows/build.yaml | 17 ++++- js/desktop/README.md | 59 +++++++++-------- js/desktop/src/node.ts | 26 +++++--- .../installing/downloading-streamplace.md | 65 +++++++++---------- js/e2e-web/README.md | 2 +- .../flows/desktop-upstream-station.spec.ts | 3 +- pkg/api/feeds.go | 2 +- pkg/atproto/atproto.go | 44 +++++-------- pkg/atproto/firehose.go | 14 ++-- pkg/atproto/indexing_scope.go | 2 +- pkg/atproto/upstream_firehose.go | 28 ++++---- pkg/atproto/upstream_station.go | 17 ++--- pkg/atproto/upstream_station_test.go | 49 +++++++------- pkg/cmd/e2e.go | 5 +- pkg/cmd/e2e_upstream.go | 3 +- pkg/cmd/streamplace.go | 18 +++-- pkg/config/config.go | 21 +++--- pkg/config/config_check_test.go | 10 ++- pkg/spxrpc/com_atproto_repo.go | 4 +- pkg/spxrpc/pds_auth_test.go | 2 +- 20 files changed, 196 insertions(+), 195 deletions(-) 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{ -- 2.54.0