From cb275890b60625dfd0843df38636ba23b3a2d200 Mon Sep 17 00:00:00 2001 From: "Natalie B." <22222885+espeon@users.noreply.github.com> Date: Tue, 9 Dec 2025 17:21:18 -0600 Subject: [PATCH] add teleport arrival backend, add cancelling teleport --- js/components/src/lib/slash-commands.ts | 3 +- .../src/lib/slash-commands/teleport.ts | 24 ++++- .../src/livestream-provider/index.tsx | 38 ++++++- .../src/livestream-store/livestream-state.tsx | 2 + .../src/livestream-store/livestream-store.tsx | 2 + .../livestream-store/websocket-consumer.tsx | 19 ++++ .../src/streamplace-provider/index.tsx | 1 + .../live/place-stream-live-denyteleport.md | 99 +++++++++++++++++++ .../live/place-stream-live-teleport.md | 66 +++++++++++++ .../content/docs/lex-reference/openapi.json | 71 +++++++++++++ .../lex-reference/place-stream-livestream.md | 81 ++++++++++++++- lexicons/place/stream/live/denyTeleport.json | 47 +++++++++ lexicons/place/stream/livestream.json | 43 ++++++++ pkg/api/websocket.go | 29 ++++++ pkg/atproto/firehose.go | 8 ++ pkg/atproto/sync.go | 43 ++++++++ pkg/constants/constants.go | 1 + pkg/model/model.go | 4 + pkg/model/teleport.go | 47 +++++++++ pkg/spxrpc/place_stream_live.go | 45 +++++++++ pkg/spxrpc/stubs.go | 19 ++++ pkg/streamplace/livedenyTeleport.go | 33 +++++++ pkg/streamplace/liveteleport.go | 14 +-- pkg/streamplace/streamlivestream.go | 50 ++++++++-- 24 files changed, 768 insertions(+), 21 deletions(-) create mode 100644 js/docs/src/content/docs/lex-reference/live/place-stream-live-denyteleport.md create mode 100644 js/docs/src/content/docs/lex-reference/live/place-stream-live-teleport.md create mode 100644 lexicons/place/stream/live/denyTeleport.json create mode 100644 pkg/streamplace/livedenyTeleport.go diff --git a/js/components/src/lib/slash-commands.ts b/js/components/src/lib/slash-commands.ts index 84bbf05e..d64e24c3 100644 --- a/js/components/src/lib/slash-commands.ts +++ b/js/components/src/lib/slash-commands.ts @@ -44,7 +44,8 @@ export async function handleSlashCommand( const command = commands.get(commandName); if (!command) { return { - handled: true, + // for now - return false + handled: false, error: `Unknown command: /${commandName}`, }; } diff --git a/js/components/src/lib/slash-commands/teleport.ts b/js/components/src/lib/slash-commands/teleport.ts index 9430623b..aadde8e4 100644 --- a/js/components/src/lib/slash-commands/teleport.ts +++ b/js/components/src/lib/slash-commands/teleport.ts @@ -1,9 +1,26 @@ import { PlaceStreamLiveTeleport, StreamplaceAgent } from "streamplace"; import { registerSlashCommand, SlashCommandResult } from "../slash-commands"; +export async function deleteTeleport( + pdsAgent: StreamplaceAgent, + userDID: string, + uri: string, +) { + const rkey = uri.split("/").pop(); + if (!rkey) { + throw new Error("No rkey found in teleport URI"); + } + return await pdsAgent.com.atproto.repo.deleteRecord({ + repo: userDID, + collection: "place.stream.live.teleport", + rkey: rkey, + }); +} + export function registerTeleportCommand( pdsAgent: StreamplaceAgent, userDID: string, + setActiveTeleportUri?: (uri: string | null) => void, ) { registerSlashCommand({ name: "teleport", @@ -72,12 +89,17 @@ export function registerTeleportCommand( }; try { - await pdsAgent.com.atproto.repo.createRecord({ + const result = await pdsAgent.com.atproto.repo.createRecord({ repo: userDID, collection: "place.stream.live.teleport", record, }); + // store the URI in the livestream store + if (setActiveTeleportUri) { + setActiveTeleportUri(result.data.uri); + } + return { handled: true }; } catch (err) { return { diff --git a/js/components/src/livestream-provider/index.tsx b/js/components/src/livestream-provider/index.tsx index df2e9f32..41139f44 100644 --- a/js/components/src/livestream-provider/index.tsx +++ b/js/components/src/livestream-provider/index.tsx @@ -1,11 +1,13 @@ import React, { useContext, useEffect, useRef } from "react"; import { useAvatars } from "../hooks"; +import { deleteTeleport } from "../lib/slash-commands/teleport"; import { StreamNotifications } from "../lib/stream-notifications"; import { LivestreamContext, makeLivestreamStore, useLivestreamStore, } from "../livestream-store"; +import { useDID, usePDSAgent } from "../streamplace-store"; import { useLivestreamWebsocket } from "./websocket"; export function LivestreamProvider({ @@ -50,7 +52,13 @@ export function TeleportWatcher({ onTeleport?: (targetHandle: string, targetDID: string) => void; }) { const activeTeleport = useLivestreamStore((state) => state.activeTeleport); + const activeTeleportUri = useLivestreamStore( + (state) => state.activeTeleportUri, + ); const profile = useAvatars(activeTeleport ? [activeTeleport.streamer] : []); + const pdsAgent = usePDSAgent(); + const userDID = useDID(); + const prevActiveTeleportRef = useRef(activeTeleport); useEffect(() => { if (!activeTeleport || !profile[activeTeleport.streamer]) return; @@ -70,8 +78,14 @@ export function TeleportWatcher({ targetHandle: targetHandle, targetDID: activeTeleport.streamer, countdown: countdown, - onCancel: () => { - console.log("Teleport cancelled by user"); + onCancel: async () => { + if (activeTeleportUri && pdsAgent && userDID) { + try { + await deleteTeleport(pdsAgent, userDID, activeTeleportUri); + } catch (err) { + console.error("Failed to delete teleport:", err); + } + } }, onAutoDismiss: () => { console.log("Teleport dismissed bestie!"); @@ -81,7 +95,25 @@ export function TeleportWatcher({ } }, }); - }, [activeTeleport, profile, onTeleport]); + }, [ + activeTeleport, + activeTeleportUri, + profile, + onTeleport, + pdsAgent, + userDID, + ]); + + useEffect(() => { + if ( + prevActiveTeleportRef.current && + !activeTeleport && + !activeTeleportUri + ) { + StreamNotifications.teleportCancelled(); + } + prevActiveTeleportRef.current = activeTeleport; + }, [activeTeleport, activeTeleportUri]); return <>; } diff --git a/js/components/src/livestream-store/livestream-state.tsx b/js/components/src/livestream-store/livestream-state.tsx index 610eaa14..c41976b4 100644 --- a/js/components/src/livestream-store/livestream-state.tsx +++ b/js/components/src/livestream-store/livestream-state.tsx @@ -24,6 +24,8 @@ export interface LivestreamState { streamKey: string | null; setStreamKey: (key: string | null) => void; activeTeleport: PlaceStreamLiveTeleport.Record | null; + activeTeleportUri: string | null; + setActiveTeleportUri: (uri: string | null) => void; websocketConnected: boolean; hasReceivedSegment: boolean; moderationPermissions: PlaceStreamModerationPermission.Record[]; diff --git a/js/components/src/livestream-store/livestream-store.tsx b/js/components/src/livestream-store/livestream-store.tsx index 2f434ed6..304d1be8 100644 --- a/js/components/src/livestream-store/livestream-store.tsx +++ b/js/components/src/livestream-store/livestream-store.tsx @@ -23,6 +23,8 @@ export const makeLivestreamStore = (): StoreApi => { recentSegments: [], problems: [], activeTeleport: null, + activeTeleportUri: null, + setActiveTeleportUri: (uri) => set({ activeTeleportUri: uri }), websocketConnected: false, hasReceivedSegment: false, moderationPermissions: [], diff --git a/js/components/src/livestream-store/websocket-consumer.tsx b/js/components/src/livestream-store/websocket-consumer.tsx index 63773d43..a70062fd 100644 --- a/js/components/src/livestream-store/websocket-consumer.tsx +++ b/js/components/src/livestream-store/websocket-consumer.tsx @@ -122,6 +122,25 @@ export const handleWebSocketMessages = ( pendingHides: newPendingHides, }; state = reduceChat(state, [], [], [hiddenMessageUri]); + } else if (PlaceStreamLiveTeleport.isRecord(message)) { + const teleportRecord = message as PlaceStreamLiveTeleport.Record; + state = { + ...state, + activeTeleport: teleportRecord, + }; + } else if (PlaceStreamLivestream.isTeleportArrival(message)) { + const arrival = message as PlaceStreamLivestream.TeleportArrival; + // when receiving a teleportArrival, we're the target + // the source is teleporting to us + console.log("Received teleport arrival", arrival); + // TODO: show notification or UI for incoming teleport + } else if (PlaceStreamLivestream.isTeleportCanceled(message)) { + // teleport was canceled (deleted or denied) + state = { + ...state, + activeTeleport: null, + activeTeleportUri: null, + }; } } } diff --git a/js/components/src/streamplace-provider/index.tsx b/js/components/src/streamplace-provider/index.tsx index 671b425a..a5fea376 100644 --- a/js/components/src/streamplace-provider/index.tsx +++ b/js/components/src/streamplace-provider/index.tsx @@ -19,6 +19,7 @@ export function StreamplaceProvider({ url: string; oauthSession?: SessionManager | null; }) { + console.log("yeh"); // todo: handle url changes? const store = useRef(makeStreamplaceStore({ url })).current; diff --git a/js/docs/src/content/docs/lex-reference/live/place-stream-live-denyteleport.md b/js/docs/src/content/docs/lex-reference/live/place-stream-live-denyteleport.md new file mode 100644 index 00000000..66634109 --- /dev/null +++ b/js/docs/src/content/docs/lex-reference/live/place-stream-live-denyteleport.md @@ -0,0 +1,99 @@ +--- +title: place.stream.live.denyTeleport +description: Reference for the place.stream.live.denyTeleport lexicon +--- + +**Lexicon Version:** 1 + +## Definitions + + + +### `main` + +**Type:** `procedure` + +Deny an incoming teleport request. + +**Parameters:** _(None defined)_ + +**Input:** + +- **Encoding:** `application/json` +- **Schema:** + +**Schema Type:** `object` + +| Name | Type | Req'd | Description | Constraints | +| ----- | -------- | ----- | --------------------------------------- | ---------------- | +| `uri` | `string` | ✅ | The URI of the teleport record to deny. | Format: `at-uri` | + +**Output:** + +- **Encoding:** `application/json` +- **Schema:** + +**Schema Type:** `object` + +| Name | Type | Req'd | Description | Constraints | +| --------- | --------- | ----- | --------------------------------------------- | ----------- | +| `success` | `boolean` | ✅ | Whether the teleport was successfully denied. | | + +**Possible Errors:** + +- `TeleportNotFound`: The specified teleport was not found. +- `Unauthorized`: The authenticated user is not the target of this teleport. + +--- + +## Lexicon Source + +```json +{ + "lexicon": 1, + "id": "place.stream.live.denyTeleport", + "defs": { + "main": { + "type": "procedure", + "description": "Deny an incoming teleport request.", + "input": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["uri"], + "properties": { + "uri": { + "type": "string", + "format": "at-uri", + "description": "The URI of the teleport record to deny." + } + } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["success"], + "properties": { + "success": { + "type": "boolean", + "description": "Whether the teleport was successfully denied." + } + } + } + }, + "errors": [ + { + "name": "TeleportNotFound", + "description": "The specified teleport was not found." + }, + { + "name": "Unauthorized", + "description": "The authenticated user is not the target of this teleport." + } + ] + } + } +} +``` diff --git a/js/docs/src/content/docs/lex-reference/live/place-stream-live-teleport.md b/js/docs/src/content/docs/lex-reference/live/place-stream-live-teleport.md new file mode 100644 index 00000000..39713fa0 --- /dev/null +++ b/js/docs/src/content/docs/lex-reference/live/place-stream-live-teleport.md @@ -0,0 +1,66 @@ +--- +title: place.stream.live.teleport +description: Reference for the place.stream.live.teleport lexicon +--- + +**Lexicon Version:** 1 + +## Definitions + + + +### `main` + +**Type:** `record` + +Record defining a 'teleport', that is active during a certain time. + +**Record Key:** `tid` + +**Record Properties:** + +| Name | Type | Req'd | Description | Constraints | +| ----------------- | --------- | ----- | ---------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------- | +| `streamer` | `string` | ✅ | The DID of the streamer to teleport to. | Format: `did` | +| `startsAt` | `string` | ✅ | The time the teleport becomes active. | Format: `datetime` | +| `durationSeconds` | `integer` | ❌ | The time limit in seconds for the teleport. If not set, the teleport is permanent. Must be at least 60 seconds, and no more than 32,400 seconds (9 hours). | Min: 60
Max: 32400 | + +--- + +## Lexicon Source + +```json +{ + "lexicon": 1, + "id": "place.stream.live.teleport", + "defs": { + "main": { + "type": "record", + "key": "tid", + "description": "Record defining a 'teleport', that is active during a certain time.", + "record": { + "type": "object", + "required": ["streamer", "startsAt"], + "properties": { + "streamer": { + "type": "string", + "format": "did", + "description": "The DID of the streamer to teleport to." + }, + "startsAt": { + "type": "string", + "format": "datetime", + "description": "The time the teleport becomes active." + }, + "durationSeconds": { + "type": "integer", + "description": "The time limit in seconds for the teleport. If not set, the teleport is permanent. Must be at least 60 seconds, and no more than 32,400 seconds (9 hours).", + "minimum": 60, + "maximum": 32400 + } + } + } + } + } +} +``` diff --git a/js/docs/src/content/docs/lex-reference/openapi.json b/js/docs/src/content/docs/lex-reference/openapi.json index 2e80021b..9272df1f 100644 --- a/js/docs/src/content/docs/lex-reference/openapi.json +++ b/js/docs/src/content/docs/lex-reference/openapi.json @@ -517,6 +517,77 @@ } } }, + "/xrpc/place.stream.live.denyTeleport": { + "post": { + "summary": "Deny an incoming teleport request.", + "operationId": "place.stream.live.denyTeleport", + "tags": ["place.stream.live"], + "responses": { + "200": { + "description": "Success", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "success": { + "type": "boolean", + "description": "Whether the teleport was successfully denied." + } + }, + "required": ["success"] + } + } + } + }, + "400": { + "description": "Bad Request", + "content": { + "application/json": { + "schema": { + "type": "object", + "required": ["error", "message"], + "properties": { + "error": { + "type": "string", + "oneOf": [ + { + "const": "TeleportNotFound" + }, + { + "const": "Unauthorized" + } + ] + }, + "message": { + "type": "string" + } + } + } + } + } + } + }, + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "uri": { + "type": "string", + "description": "The URI of the teleport record to deny.", + "format": "uri" + } + }, + "required": ["uri"] + } + } + } + } + } + }, "/xrpc/place.stream.multistream.createTarget": { "post": { "summary": "Create a new target for rebroadcasting a Streamplace stream.", diff --git a/js/docs/src/content/docs/lex-reference/place-stream-livestream.md b/js/docs/src/content/docs/lex-reference/place-stream-livestream.md index 5701d07a..490f06bd 100644 --- a/js/docs/src/content/docs/lex-reference/place-stream-livestream.md +++ b/js/docs/src/content/docs/lex-reference/place-stream-livestream.md @@ -79,6 +79,38 @@ Record announcing a livestream is happening --- + + +### `teleportArrival` + +**Type:** `object` + +**Properties:** + +| Name | Type | Req'd | Description | Constraints | +| ------------- | ------------------------------------------------------------------------------------------------------------------------------------------------ | ----- | -------------------------------------------------- | ------------------ | +| `teleportUri` | `string` | ✅ | The URI of the teleport record | Format: `at-uri` | +| `source` | [`app.bsky.actor.defs#profileViewBasic`](https://github.com/bluesky-social/atproto/tree/main/lexicons/app/bsky/actor/defs.json#profileViewBasic) | ✅ | The streamer who is teleporting their viewers here | | +| `viewerCount` | `integer` | ✅ | How many viewers are arriving from this teleport | | +| `startsAt` | `string` | ✅ | When this teleport started | Format: `datetime` | + +--- + + + +### `teleportCanceled` + +**Type:** `object` + +**Properties:** + +| Name | Type | Req'd | Description | Constraints | +| ------------- | -------- | ----- | ------------------------------------------------ | ------------------------------------ | +| `teleportUri` | `string` | ✅ | The URI of the teleport record that was canceled | Format: `at-uri` | +| `reason` | `string` | ✅ | Why this teleport was canceled | Enum: `deleted`, `denied`, `expired` | + +--- + ### `streamplaceAnything` @@ -87,9 +119,9 @@ Record announcing a livestream is happening **Properties:** -| Name | Type | Req'd | Description | Constraints | -| ------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----- | ----------- | ----------- | -| `livestream` | Union of:
  [`#livestreamView`](#livestreamview)
  [`#viewerCount`](#viewercount)
  [`place.stream.defs#blockView`](/lex-reference/place-stream-defs#blockview)
  [`place.stream.defs#renditions`](/lex-reference/place-stream-defs#renditions)
  [`place.stream.defs#rendition`](/lex-reference/place-stream-defs#rendition)
  [`place.stream.chat.defs#messageView`](/lex-reference/place-stream-chat-defs#messageview) | ✅ | | | +| Name | Type | Req'd | Description | Constraints | +| ------------ | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----- | ----------- | ----------- | +| `livestream` | Union of:
  [`#livestreamView`](#livestreamview)
  [`#viewerCount`](#viewercount)
  [`#teleportArrival`](#teleportarrival)
  [`#teleportCanceled`](#teleportcanceled)
  [`place.stream.defs#blockView`](/lex-reference/place-stream-defs#blockview)
  [`place.stream.defs#renditions`](/lex-reference/place-stream-defs#renditions)
  [`place.stream.defs#rendition`](/lex-reference/place-stream-defs#rendition)
  [`place.stream.chat.defs#messageView`](/lex-reference/place-stream-chat-defs#messageview) | ✅ | | | --- @@ -199,6 +231,47 @@ Record announcing a livestream is happening } } }, + "teleportArrival": { + "type": "object", + "required": ["teleportUri", "source", "viewerCount", "startsAt"], + "properties": { + "teleportUri": { + "type": "string", + "format": "at-uri", + "description": "The URI of the teleport record" + }, + "source": { + "type": "ref", + "ref": "app.bsky.actor.defs#profileViewBasic", + "description": "The streamer who is teleporting their viewers here" + }, + "viewerCount": { + "type": "integer", + "description": "How many viewers are arriving from this teleport" + }, + "startsAt": { + "type": "string", + "format": "datetime", + "description": "When this teleport started" + } + } + }, + "teleportCanceled": { + "type": "object", + "required": ["teleportUri", "reason"], + "properties": { + "teleportUri": { + "type": "string", + "format": "at-uri", + "description": "The URI of the teleport record that was canceled" + }, + "reason": { + "type": "string", + "enum": ["deleted", "denied", "expired"], + "description": "Why this teleport was canceled" + } + } + }, "streamplaceAnything": { "type": "object", "required": ["livestream"], @@ -208,6 +281,8 @@ Record announcing a livestream is happening "refs": [ "#livestreamView", "#viewerCount", + "#teleportArrival", + "#teleportCanceled", "place.stream.defs#blockView", "place.stream.defs#renditions", "place.stream.defs#rendition", diff --git a/lexicons/place/stream/live/denyTeleport.json b/lexicons/place/stream/live/denyTeleport.json new file mode 100644 index 00000000..3e563dfa --- /dev/null +++ b/lexicons/place/stream/live/denyTeleport.json @@ -0,0 +1,47 @@ +{ + "lexicon": 1, + "id": "place.stream.live.denyTeleport", + "defs": { + "main": { + "type": "procedure", + "description": "Deny an incoming teleport request.", + "input": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["uri"], + "properties": { + "uri": { + "type": "string", + "format": "at-uri", + "description": "The URI of the teleport record to deny." + } + } + } + }, + "output": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["success"], + "properties": { + "success": { + "type": "boolean", + "description": "Whether the teleport was successfully denied." + } + } + } + }, + "errors": [ + { + "name": "TeleportNotFound", + "description": "The specified teleport was not found." + }, + { + "name": "Unauthorized", + "description": "The authenticated user is not the target of this teleport." + } + ] + } + } +} diff --git a/lexicons/place/stream/livestream.json b/lexicons/place/stream/livestream.json index 195b2045..39f4efa8 100644 --- a/lexicons/place/stream/livestream.json +++ b/lexicons/place/stream/livestream.json @@ -88,6 +88,47 @@ "count": { "type": "integer" } } }, + "teleportArrival": { + "type": "object", + "required": ["teleportUri", "source", "viewerCount", "startsAt"], + "properties": { + "teleportUri": { + "type": "string", + "format": "at-uri", + "description": "The URI of the teleport record" + }, + "source": { + "type": "ref", + "ref": "app.bsky.actor.defs#profileViewBasic", + "description": "The streamer who is teleporting their viewers here" + }, + "viewerCount": { + "type": "integer", + "description": "How many viewers are arriving from this teleport" + }, + "startsAt": { + "type": "string", + "format": "datetime", + "description": "When this teleport started" + } + } + }, + "teleportCanceled": { + "type": "object", + "required": ["teleportUri", "reason"], + "properties": { + "teleportUri": { + "type": "string", + "format": "at-uri", + "description": "The URI of the teleport record that was canceled" + }, + "reason": { + "type": "string", + "enum": ["deleted", "denied", "expired"], + "description": "Why this teleport was canceled" + } + } + }, "streamplaceAnything": { "type": "object", "required": ["livestream"], @@ -97,6 +138,8 @@ "refs": [ "#livestreamView", "#viewerCount", + "#teleportArrival", + "#teleportCanceled", "place.stream.defs#blockView", "place.stream.defs#renditions", "place.stream.defs#rendition", diff --git a/pkg/api/websocket.go b/pkg/api/websocket.go index 96d9e79c..1cacc7b0 100644 --- a/pkg/api/websocket.go +++ b/pkg/api/websocket.go @@ -7,6 +7,7 @@ import ( "net/http" "time" + bsky "github.com/bluesky-social/indigo/api/bsky" "github.com/google/uuid" "github.com/gorilla/websocket" "github.com/julienschmidt/httprouter" @@ -241,6 +242,34 @@ func (a *StreamplaceAPI) HandleWebsocket(ctx context.Context) httprouter.Handle } }() + go func() { + teleports, err := a.Model.GetActiveTeleportsToRepo(repoDID) + if err != nil { + log.Error(ctx, "could not get active teleports", "error", err) + return + } + log.Log(ctx, "found active teleports in initial burst", "count", len(teleports), "targetDID", repoDID) + for _, tp := range teleports { + if tp.Repo == nil { + log.Error(ctx, "teleport repo is nil", "uri", tp.URI) + continue + } + viewerCount := a.Bus.GetViewerCount(tp.RepoDID) + arrivalMsg := streamplace.Livestream_TeleportArrival{ + LexiconTypeID: "place.stream.livestream#teleportArrival", + TeleportUri: tp.URI, + Source: &bsky.ActorDefs_ProfileViewBasic{ + Did: tp.RepoDID, + Handle: tp.Repo.Handle, + }, + ViewerCount: int64(viewerCount), + StartsAt: tp.StartsAt.Format(time.RFC3339), + } + log.Log(ctx, "sending teleport arrival in initial burst", "from", tp.RepoDID, "to", repoDID) + initialBurst <- arrivalMsg + } + }() + for { messageType, message, err := conn.ReadMessage() if err != nil { diff --git a/pkg/atproto/firehose.go b/pkg/atproto/firehose.go index 400d29bd..541db46c 100644 --- a/pkg/atproto/firehose.go +++ b/pkg/atproto/firehose.go @@ -305,6 +305,14 @@ func (atsync *ATProtoSynchronizer) handleCommitEventOps(ctx context.Context, evt atsync.Bus.Publish(msg.StreamerRepoDID, mv) } + if collection.String() == constants.PLACE_STREAM_LIVE_TELEPORT { + log.Warn(ctx, "deleting teleport", "userDID", evt.Repo, "uri", uri) + err := atsync.Model.DeleteTeleport(ctx, uri) + if err != nil { + log.Error(ctx, "failed to delete teleport", "err", err) + } + } + if collection.String() == constants.PLACE_STREAM_MODERATION_PERMISSION { log.Debug(ctx, "deleting moderation delegation", "userDID", evt.Repo, "rkey", rkey.String()) err := atsync.Model.DeleteModerationDelegation(ctx, rkey.String()) diff --git a/pkg/atproto/sync.go b/pkg/atproto/sync.go index ddfa01ee..ddc34419 100644 --- a/pkg/atproto/sync.go +++ b/pkg/atproto/sync.go @@ -404,6 +404,49 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD } go atsync.Bus.Publish(userDID, rec) + // schedule arrival notification 10 seconds after startsAt + arrivalTime := startsAt.Add(10 * time.Second) + waitDuration := time.Until(arrivalTime) + if waitDuration < 0 { + waitDuration = 0 + } + + time.AfterFunc(waitDuration, func() { + // verify the teleport still exists + existingTp, err := atsync.Model.GetTeleportByURI(aturi.String()) + if err != nil { + log.Error(ctx, "failed to get teleport by uri", "err", err) + return + } + if existingTp == nil || existingTp.Denied { + log.Debug(ctx, "teleport no longer active, skipping arrival notification", "uri", aturi.String()) + return + } + + // get the source profile + sourceRepo, err := atsync.Model.GetRepo(userDID) + if err != nil { + log.Error(ctx, "failed to get source repo", "err", err) + return + } + + viewerCount := atsync.Bus.GetViewerCount(userDID) + + arrivalMsg := &streamplace.Livestream_TeleportArrival{ + LexiconTypeID: "place.stream.livestream#teleportArrival", + TeleportUri: aturi.String(), + Source: &bsky.ActorDefs_ProfileViewBasic{ + Did: userDID, + Handle: sourceRepo.Handle, + }, + ViewerCount: int64(viewerCount), + StartsAt: rec.StartsAt, + } + + log.Log(ctx, "sending teleport arrival notification", "from", userDID, "to", rec.Streamer, "uri", aturi.String()) + atsync.Bus.Publish(rec.Streamer, arrivalMsg) + }) + case *streamplace.Key: log.Debug(ctx, "creating key", "key", rec) time, err := aqtime.FromString(rec.CreatedAt) diff --git a/pkg/constants/constants.go b/pkg/constants/constants.go index 3fade0fa..92474d55 100644 --- a/pkg/constants/constants.go +++ b/pkg/constants/constants.go @@ -5,6 +5,7 @@ var PLACE_STREAM_LIVESTREAM = "place.stream.livestream" // var PLACE_STREAM_CHAT_MESSAGE = "place.stream.chat.message" //nolint:all var PLACE_STREAM_CHAT_PROFILE = "place.stream.chat.profile" //nolint:all var PLACE_STREAM_SERVER_SETTINGS = "place.stream.server.settings" //nolint:all +var PLACE_STREAM_LIVE_TELEPORT = "place.stream.live.teleport" //nolint:all var PLACE_STREAM_MODERATION_PERMISSION = "place.stream.moderation.permission" //nolint:all var STREAMPLACE_SIGNING_KEY = "signingKey" //nolint:all var APP_BSKY_GRAPH_FOLLOW = "app.bsky.graph.follow" //nolint:all diff --git a/pkg/model/model.go b/pkg/model/model.go index f0600f66..88e60d43 100644 --- a/pkg/model/model.go +++ b/pkg/model/model.go @@ -77,6 +77,10 @@ type Model interface { CreateTeleport(ctx context.Context, tp *Teleport) error GetLatestTeleportForRepo(repoDID string) (*Teleport, error) GetActiveTeleportsForRepo(repoDID string) ([]Teleport, error) + GetActiveTeleportsToRepo(targetDID string) ([]Teleport, error) + GetTeleportByURI(uri string) (*Teleport, error) + DeleteTeleport(ctx context.Context, uri string) error + DenyTeleport(ctx context.Context, uri string) error CreateBlock(ctx context.Context, block *Block) error GetBlock(ctx context.Context, rkey string) (*Block, error) diff --git a/pkg/model/teleport.go b/pkg/model/teleport.go index 88d536cf..c221d65e 100644 --- a/pkg/model/teleport.go +++ b/pkg/model/teleport.go @@ -18,6 +18,7 @@ type Teleport struct { Teleport *[]byte `json:"teleport"` RepoDID string `json:"repoDID" gorm:"column:repo_did;index:idx_repo_starts,priority:1"` TargetDID string `json:"targetDID" gorm:"column:target_did;index:idx_target_did"` + Denied bool `json:"denied" gorm:"column:denied;default:false"` Repo *Repo `json:"repo,omitempty" gorm:"foreignKey:DID;references:RepoDID"` Target *Repo `json:"target,omitempty" gorm:"foreignKey:DID;references:TargetDID"` } @@ -53,6 +54,7 @@ func (m *DBModel) GetActiveTeleportsForRepo(repoDID string) ([]Teleport, error) Preload("Repo"). Preload("Target"). Where("repo_did = ?", repoDID). + Where("denied = ?", false). Where("starts_at <= ?", now). Where("(duration_seconds IS NULL OR DATE_ADD(starts_at, INTERVAL duration_seconds SECOND) > ?)", now). Order("starts_at DESC"). @@ -65,3 +67,48 @@ func (m *DBModel) GetActiveTeleportsForRepo(repoDID string) ([]Teleport, error) } return teleports, nil } + +func (m *DBModel) GetActiveTeleportsToRepo(targetDID string) ([]Teleport, error) { + now := time.Now() + var teleports []Teleport + err := m.DB. + Preload("Repo"). + Preload("Target"). + Where("target_did = ?", targetDID). + Where("denied = ?", false). + Where("starts_at <= ?", now). + Where("(duration_seconds IS NULL OR datetime(starts_at, '+' || duration_seconds || ' seconds') > ?)", now). + Order("starts_at DESC"). + Find(&teleports).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("error retrieving active teleports to repo: %w", err) + } + return teleports, nil +} + +func (m *DBModel) GetTeleportByURI(uri string) (*Teleport, error) { + var teleport Teleport + err := m.DB. + Preload("Repo"). + Preload("Target"). + Where("uri = ?", uri). + First(&teleport).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("error retrieving teleport by uri: %w", err) + } + return &teleport, nil +} + +func (m *DBModel) DeleteTeleport(ctx context.Context, uri string) error { + return m.DB.Where("uri = ?", uri).Delete(&Teleport{}).Error +} + +func (m *DBModel) DenyTeleport(ctx context.Context, uri string) error { + return m.DB.Model(&Teleport{}).Where("uri = ?", uri).Update("denied", true).Error +} diff --git a/pkg/spxrpc/place_stream_live.go b/pkg/spxrpc/place_stream_live.go index 9dae6bba..b4b36c24 100644 --- a/pkg/spxrpc/place_stream_live.go +++ b/pkg/spxrpc/place_stream_live.go @@ -9,6 +9,7 @@ import ( "github.com/bluesky-social/indigo/lex/util" "github.com/gorilla/websocket" "github.com/labstack/echo/v4" + "github.com/streamplace/oatproxy/pkg/oatproxy" "stream.place/streamplace/pkg/log" "stream.place/streamplace/pkg/spid" "stream.place/streamplace/pkg/spmetrics" @@ -16,6 +17,50 @@ import ( placestreamtypes "stream.place/streamplace/pkg/streamplace" ) +func (s *Server) handlePlaceStreamLiveDenyTeleport(ctx context.Context, input *placestreamtypes.LiveDenyTeleport_Input) (*placestreamtypes.LiveDenyTeleport_Output, error) { + session, _ := oatproxy.GetOAuthSession(ctx) + if session == nil { + return nil, echo.NewHTTPError(http.StatusUnauthorized, "oauth session not found") + } + + if input.Uri == "" { + return nil, echo.NewHTTPError(http.StatusBadRequest, "URI is required") + } + + teleport, err := s.model.GetTeleportByURI(input.Uri) + if err != nil { + log.Error(ctx, "failed to get teleport", "err", err) + return nil, echo.NewHTTPError(http.StatusInternalServerError, "Failed to retrieve teleport") + } + + if teleport == nil { + return nil, echo.NewHTTPError(http.StatusNotFound, "Teleport not found") + } + + if teleport.TargetDID != session.DID { + return nil, echo.NewHTTPError(http.StatusForbidden, "You are not the target of this teleport") + } + + err = s.model.DenyTeleport(ctx, input.Uri) + if err != nil { + log.Error(ctx, "failed to deny teleport", "err", err) + return nil, echo.NewHTTPError(http.StatusInternalServerError, "Failed to deny teleport") + } + + cancelMsg := &placestreamtypes.Livestream_TeleportCanceled{ + LexiconTypeID: "place.stream.livestream#teleportCanceled", + TeleportUri: input.Uri, + Reason: "denied", + } + + s.bus.Publish(teleport.RepoDID, cancelMsg) + s.bus.Publish(teleport.TargetDID, cancelMsg) + + return &placestreamtypes.LiveDenyTeleport_Output{ + Success: true, + }, nil +} + var replicationUpgrader = websocket.Upgrader{ ReadBufferSize: 1024, WriteBufferSize: 1024 * 1024 * 10, // 10MB diff --git a/pkg/spxrpc/stubs.go b/pkg/spxrpc/stubs.go index 4cad22c0..f922a8e8 100644 --- a/pkg/spxrpc/stubs.go +++ b/pkg/spxrpc/stubs.go @@ -268,6 +268,7 @@ func (s *Server) RegisterHandlersPlaceStream(e *echo.Echo) error { e.POST("/xrpc/place.stream.branding.updateBlob", s.HandlePlaceStreamBrandingUpdateBlob) e.GET("/xrpc/place.stream.broadcast.getBroadcaster", s.HandlePlaceStreamBroadcastGetBroadcaster) e.GET("/xrpc/place.stream.graph.getFollowingUser", s.HandlePlaceStreamGraphGetFollowingUser) + e.POST("/xrpc/place.stream.live.denyTeleport", s.HandlePlaceStreamLiveDenyTeleport) e.GET("/xrpc/place.stream.live.getLiveUsers", s.HandlePlaceStreamLiveGetLiveUsers) e.GET("/xrpc/place.stream.live.getProfileCard", s.HandlePlaceStreamLiveGetProfileCard) e.GET("/xrpc/place.stream.live.getRecommendations", s.HandlePlaceStreamLiveGetRecommendations) @@ -384,6 +385,24 @@ func (s *Server) HandlePlaceStreamGraphGetFollowingUser(c echo.Context) error { return c.JSON(200, out) } +func (s *Server) HandlePlaceStreamLiveDenyTeleport(c echo.Context) error { + ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandlePlaceStreamLiveDenyTeleport") + defer span.End() + + var body placestream.LiveDenyTeleport_Input + if err := c.Bind(&body); err != nil { + return err + } + var out *placestream.LiveDenyTeleport_Output + var handleErr error + // func (s *Server) handlePlaceStreamLiveDenyTeleport(ctx context.Context,body *placestream.LiveDenyTeleport_Input) (*placestream.LiveDenyTeleport_Output, error) + out, handleErr = s.handlePlaceStreamLiveDenyTeleport(ctx, &body) + if handleErr != nil { + return handleErr + } + return c.JSON(200, out) +} + func (s *Server) HandlePlaceStreamLiveGetLiveUsers(c echo.Context) error { ctx, span := otel.Tracer("server").Start(c.Request().Context(), "HandlePlaceStreamLiveGetLiveUsers") defer span.End() diff --git a/pkg/streamplace/livedenyTeleport.go b/pkg/streamplace/livedenyTeleport.go new file mode 100644 index 00000000..edbd5c92 --- /dev/null +++ b/pkg/streamplace/livedenyTeleport.go @@ -0,0 +1,33 @@ +// Code generated by cmd/lexgen (see Makefile's lexgen); DO NOT EDIT. + +// Lexicon schema: place.stream.live.denyTeleport + +package streamplace + +import ( + "context" + + lexutil "github.com/bluesky-social/indigo/lex/util" +) + +// LiveDenyTeleport_Input is the input argument to a place.stream.live.denyTeleport call. +type LiveDenyTeleport_Input struct { + // uri: The URI of the teleport record to deny. + Uri string `json:"uri" cborgen:"uri"` +} + +// LiveDenyTeleport_Output is the output of a place.stream.live.denyTeleport call. +type LiveDenyTeleport_Output struct { + // success: Whether the teleport was successfully denied. + Success bool `json:"success" cborgen:"success"` +} + +// LiveDenyTeleport calls the XRPC method "place.stream.live.denyTeleport". +func LiveDenyTeleport(ctx context.Context, c lexutil.LexClient, input *LiveDenyTeleport_Input) (*LiveDenyTeleport_Output, error) { + var out LiveDenyTeleport_Output + if err := c.LexDo(ctx, lexutil.Procedure, "application/json", "place.stream.live.denyTeleport", nil, input, &out); err != nil { + return nil, err + } + + return &out, nil +} diff --git a/pkg/streamplace/liveteleport.go b/pkg/streamplace/liveteleport.go index c14bf458..4e297425 100644 --- a/pkg/streamplace/liveteleport.go +++ b/pkg/streamplace/liveteleport.go @@ -1,19 +1,19 @@ // Code generated by cmd/lexgen (see Makefile's lexgen); DO NOT EDIT. -package streamplace +// Lexicon schema: place.stream.live.teleport -// schema: place.stream.live.teleport +package streamplace import ( - "github.com/bluesky-social/indigo/lex/util" + lexutil "github.com/bluesky-social/indigo/lex/util" ) func init() { - util.RegisterType("place.stream.live.teleport", &LiveTeleport{}) -} // -// RECORDTYPE: LiveTeleport + lexutil.RegisterType("place.stream.live.teleport", &LiveTeleport{}) +} + type LiveTeleport struct { - LexiconTypeID string `json:"$type,const=place.stream.live.teleport" cborgen:"$type,const=place.stream.live.teleport"` + LexiconTypeID string `json:"$type" cborgen:"$type,const=place.stream.live.teleport"` // durationSeconds: The time limit in seconds for the teleport. If not set, the teleport is permanent. Must be at least 60 seconds, and no more than 32,400 seconds (9 hours). DurationSeconds *int64 `json:"durationSeconds,omitempty" cborgen:"durationSeconds,omitempty"` // startsAt: The time the teleport becomes active. diff --git a/pkg/streamplace/streamlivestream.go b/pkg/streamplace/streamlivestream.go index dcb9f3df..438ab47c 100644 --- a/pkg/streamplace/streamlivestream.go +++ b/pkg/streamplace/streamlivestream.go @@ -59,12 +59,14 @@ type Livestream_StreamplaceAnything struct { } type Livestream_StreamplaceAnything_Livestream struct { - Livestream_LivestreamView *Livestream_LivestreamView - Livestream_ViewerCount *Livestream_ViewerCount - Defs_BlockView *Defs_BlockView - Defs_Renditions *Defs_Renditions - Defs_Rendition *Defs_Rendition - ChatDefs_MessageView *ChatDefs_MessageView + Livestream_LivestreamView *Livestream_LivestreamView + Livestream_ViewerCount *Livestream_ViewerCount + Livestream_TeleportArrival *Livestream_TeleportArrival + Livestream_TeleportCanceled *Livestream_TeleportCanceled + Defs_BlockView *Defs_BlockView + Defs_Renditions *Defs_Renditions + Defs_Rendition *Defs_Rendition + ChatDefs_MessageView *ChatDefs_MessageView } func (t *Livestream_StreamplaceAnything_Livestream) MarshalJSON() ([]byte, error) { @@ -76,6 +78,14 @@ func (t *Livestream_StreamplaceAnything_Livestream) MarshalJSON() ([]byte, error t.Livestream_ViewerCount.LexiconTypeID = "place.stream.livestream#viewerCount" return json.Marshal(t.Livestream_ViewerCount) } + if t.Livestream_TeleportArrival != nil { + t.Livestream_TeleportArrival.LexiconTypeID = "place.stream.livestream#teleportArrival" + return json.Marshal(t.Livestream_TeleportArrival) + } + if t.Livestream_TeleportCanceled != nil { + t.Livestream_TeleportCanceled.LexiconTypeID = "place.stream.livestream#teleportCanceled" + return json.Marshal(t.Livestream_TeleportCanceled) + } if t.Defs_BlockView != nil { t.Defs_BlockView.LexiconTypeID = "place.stream.defs#blockView" return json.Marshal(t.Defs_BlockView) @@ -108,6 +118,12 @@ func (t *Livestream_StreamplaceAnything_Livestream) UnmarshalJSON(b []byte) erro case "place.stream.livestream#viewerCount": t.Livestream_ViewerCount = new(Livestream_ViewerCount) return json.Unmarshal(b, t.Livestream_ViewerCount) + case "place.stream.livestream#teleportArrival": + t.Livestream_TeleportArrival = new(Livestream_TeleportArrival) + return json.Unmarshal(b, t.Livestream_TeleportArrival) + case "place.stream.livestream#teleportCanceled": + t.Livestream_TeleportCanceled = new(Livestream_TeleportCanceled) + return json.Unmarshal(b, t.Livestream_TeleportCanceled) case "place.stream.defs#blockView": t.Defs_BlockView = new(Defs_BlockView) return json.Unmarshal(b, t.Defs_BlockView) @@ -125,6 +141,28 @@ func (t *Livestream_StreamplaceAnything_Livestream) UnmarshalJSON(b []byte) erro } } +// Livestream_TeleportArrival is a "teleportArrival" in the place.stream.livestream schema. +type Livestream_TeleportArrival struct { + LexiconTypeID string `json:"$type" cborgen:"$type,const=place.stream.livestream#teleportArrival"` + // source: The streamer who is teleporting their viewers here + Source *appbsky.ActorDefs_ProfileViewBasic `json:"source" cborgen:"source"` + // startsAt: When this teleport started + StartsAt string `json:"startsAt" cborgen:"startsAt"` + // teleportUri: The URI of the teleport record + TeleportUri string `json:"teleportUri" cborgen:"teleportUri"` + // viewerCount: How many viewers are arriving from this teleport + ViewerCount int64 `json:"viewerCount" cborgen:"viewerCount"` +} + +// Livestream_TeleportCanceled is a "teleportCanceled" in the place.stream.livestream schema. +type Livestream_TeleportCanceled struct { + LexiconTypeID string `json:"$type" cborgen:"$type,const=place.stream.livestream#teleportCanceled"` + // reason: Why this teleport was canceled + Reason string `json:"reason" cborgen:"reason"` + // teleportUri: The URI of the teleport record that was canceled + TeleportUri string `json:"teleportUri" cborgen:"teleportUri"` +} + // Livestream_ViewerCount is a "viewerCount" in the place.stream.livestream schema. type Livestream_ViewerCount struct { LexiconTypeID string `json:"$type" cborgen:"$type,const=place.stream.livestream#viewerCount"` -- 2.51.2