diff --git a/js/components/src/components/chat/chat-box.tsx b/js/components/src/components/chat/chat-box.tsx
index 225a897a9..3eb1adcea 100644
--- a/js/components/src/components/chat/chat-box.tsx
+++ b/js/components/src/components/chat/chat-box.tsx
@@ -119,11 +119,15 @@ export function ChatBox({
useEffect(() => {
if (pdsAgent && userDID) {
- registerTeleportCommand(pdsAgent, userDID, setActiveTeleportUri, () =>
- setShowTeleportModal(true),
+ registerTeleportCommand(
+ pdsAgent,
+ userDID,
+ () => (linfo ? { uri: linfo.uri, cid: linfo.cid } : null),
+ setActiveTeleportUri,
+ () => setShowTeleportModal(true),
);
}
- }, [pdsAgent, userDID, setActiveTeleportUri]);
+ }, [pdsAgent, userDID, linfo, setActiveTeleportUri]);
const authors = useMemo(() => {
if (!chat) return null;
@@ -141,11 +145,12 @@ export function ChatBox({
registerTeleportCommand(
pdsAgent,
pdsAgent.did,
+ () => (linfo ? { uri: linfo.uri, cid: linfo.cid } : null),
setActiveTeleportUri,
() => setShowTeleportModal(true),
);
}
- }, [pdsAgent, linfo?.author?.did, setActiveTeleportUri]);
+ }, [pdsAgent, linfo, setActiveTeleportUri]);
const handleMentionSelect = (handle: string) => {
const beforeAt = message.slice(0, message.lastIndexOf("@"));
@@ -177,6 +182,7 @@ export function ChatBox({
userDID,
targetHandle,
countdownSeconds,
+ linfo ? { uri: linfo.uri, cid: linfo.cid } : { uri: "", cid: "" },
setActiveTeleportUri,
);
diff --git a/js/components/src/lib/slash-commands/teleport.ts b/js/components/src/lib/slash-commands/teleport.ts
index 5add08f88..981d8ce08 100644
--- a/js/components/src/lib/slash-commands/teleport.ts
+++ b/js/components/src/lib/slash-commands/teleport.ts
@@ -25,6 +25,7 @@ export async function createTeleport(
userDID: string,
targetHandle: string,
countdownSeconds: number,
+ livestream: { uri: string; cid: string },
setActiveTeleportUri?: (uri: string | null) => void,
): Promise<{ success: boolean; error?: string }> {
if (countdownSeconds < 5 || countdownSeconds > 300) {
@@ -34,6 +35,13 @@ export async function createTeleport(
};
}
+ if (!livestream?.uri || !livestream?.cid) {
+ return {
+ success: false,
+ error: "No active livestream to teleport from",
+ };
+ }
+
let targetDID: string;
try {
const resolution = await pdsAgent.resolveHandle({
@@ -62,6 +70,10 @@ export async function createTeleport(
{
streamer: targetDID as any,
startsAt: startsAt as any,
+ livestream: {
+ uri: livestream.uri,
+ cid: livestream.cid,
+ } as any,
},
{ repo: userDID as any },
);
@@ -82,6 +94,7 @@ export async function createTeleport(
export function registerTeleportCommand(
pdsAgent: StreamplaceAgent,
userDID: string,
+ getLivestream: () => { uri: string; cid: string } | null,
setActiveTeleportUri?: (uri: string | null) => void,
onOpenModal?: () => void,
) {
@@ -131,6 +144,14 @@ export function registerTeleportCommand(
countdownSeconds = parsedDuration;
}
+ const livestream = getLivestream();
+ if (!livestream?.uri || !livestream?.cid) {
+ return {
+ handled: true,
+ error: "No active livestream to teleport from",
+ };
+ }
+
let targetDID: string;
try {
const resolution = await pdsAgent.resolveHandle({
@@ -161,6 +182,10 @@ export function registerTeleportCommand(
{
streamer: targetDID as any,
startsAt: startsAt as any,
+ livestream: {
+ uri: livestream.uri,
+ cid: livestream.cid,
+ } as any,
},
{ repo: userDID as any },
);
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
index 39713fa07..e4389ebd8 100644
--- 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
@@ -19,11 +19,12 @@ Record defining a 'teleport', that is active during a certain time.
**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 |
+| 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` |
+| `livestream` | [`com.atproto.repo.strongRef`](https://github.com/bluesky-social/atproto/tree/main/lexicons/com/atproto/repo/strongref.json#undefined) | ✅ | The source livestream this teleport is sending viewers away from. When the teleport fires, this is the livestream that gets ended (the same update place.stream.live.stopLivestream performs), so the source streamer returns to pre-live. Teleports without an origin livestream are treated as a no-op. | |
+| `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 |
---
@@ -40,7 +41,7 @@ Record defining a 'teleport', that is active during a certain time.
"description": "Record defining a 'teleport', that is active during a certain time.",
"record": {
"type": "object",
- "required": ["streamer", "startsAt"],
+ "required": ["streamer", "startsAt", "livestream"],
"properties": {
"streamer": {
"type": "string",
@@ -52,6 +53,11 @@ Record defining a 'teleport', that is active during a certain time.
"format": "datetime",
"description": "The time the teleport becomes active."
},
+ "livestream": {
+ "type": "ref",
+ "ref": "com.atproto.repo.strongRef",
+ "description": "The source livestream this teleport is sending viewers away from. When the teleport fires, this is the livestream that gets ended (the same update place.stream.live.stopLivestream performs), so the source streamer returns to pre-live. Teleports without an origin livestream are treated as a no-op."
+ },
"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).",
diff --git a/lexicons/place/stream/live/teleport.json b/lexicons/place/stream/live/teleport.json
index 1f70ea69c..a100db17f 100644
--- a/lexicons/place/stream/live/teleport.json
+++ b/lexicons/place/stream/live/teleport.json
@@ -8,7 +8,7 @@
"description": "Record defining a 'teleport', that is active during a certain time.",
"record": {
"type": "object",
- "required": ["streamer", "startsAt"],
+ "required": ["streamer", "startsAt", "livestream"],
"properties": {
"streamer": {
"type": "string",
@@ -20,6 +20,11 @@
"format": "datetime",
"description": "The time the teleport becomes active."
},
+ "livestream": {
+ "type": "ref",
+ "ref": "com.atproto.repo.strongRef",
+ "description": "The source livestream this teleport is sending viewers away from. When the teleport fires, this is the livestream that gets ended (the same update place.stream.live.stopLivestream performs), so the source streamer returns to pre-live. Teleports without an origin livestream are treated as a no-op."
+ },
"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).",
diff --git a/pkg/atproto/sync.go b/pkg/atproto/sync.go
index 8850ff25a..159b9c74a 100644
--- a/pkg/atproto/sync.go
+++ b/pkg/atproto/sync.go
@@ -573,9 +573,12 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD
// navigate back. End the source streamer's livestream here (the
// same record update place.stream.live.stopLivestream performs,
// setting endedAt so the streamer returns to "pre-live"), now
- // that viewers have been sent over. Best-effort: a failure only
- // logs and never blocks the arrival notification.
- atsync.endLivestreamForTeleport(ctx, userDID)
+ // that viewers have been sent over. The exact stream to end is
+ // pinned by the teleport record's `livestream` strongRef, so a
+ // newer stream the streamer may have started in the meantime is
+ // never terminated by mistake. Best-effort: a failure only logs
+ // and never blocks the arrival notification.
+ atsync.endLivestreamForTeleport(ctx, userDID, rec.Livestream)
})
case *placestream.Key:
diff --git a/pkg/atproto/teleport_endstream.go b/pkg/atproto/teleport_endstream.go
index a0b597e3d..9e0ace198 100644
--- a/pkg/atproto/teleport_endstream.go
+++ b/pkg/atproto/teleport_endstream.go
@@ -17,13 +17,22 @@ import (
comatproto "stream.place/streamplace/pkg/comatproto"
)
-// endLivestreamForTeleport ends the source streamer's current livestream when a
-// teleport fires — the same record update that place.stream.live.stopLivestream
-// performs (set endedAt on the latest place.stream.livestream record via a
-// getRecord/putRecord swap). A teleport sends the source streamer's viewers to
-// a target; without this, the source stream stays live and viewers can navigate
-// back. Setting endedAt returns the source streamer to "pre-live" and stops the
-// manifest from publishing.
+// endLivestreamForTeleport ends the source streamer's livestream when a teleport
+// fires — the same record update that place.stream.live.stopLivestream performs
+// (set endedAt on the place.stream.livestream record via a getRecord/putRecord
+// swap). A teleport sends the source streamer's viewers to a target; without
+// this, the source stream stays live and viewers can navigate back. Setting
+// endedAt returns the source streamer to "pre-live" and stops the manifest from
+// publishing.
+//
+// The livestream to end is identified by the teleport record's `livestream`
+// strongRef (uri + cid), NOT by "the streamer's latest livestream". The latest
+// record is a race: if the streamer started a new stream before this delayed
+// arrival callback runs, GetLatestLivestreamForRepo would select the wrong
+// (new) record and terminate an unrelated broadcast. The strongRef pins the
+// exact origin stream, so we always end the right one. Teleports whose record
+// predates the `livestream` field (or otherwise lacks an origin stream) are
+// treated as a no-op.
//
// The firehose path has no OAuth session, so the streamer's own stored session
// is looked up and an XRPC client is built from it — the same delegated-client
@@ -31,10 +40,19 @@ import (
//
// Best-effort: any failure only logs and returns. It never blocks the teleport
// arrival notification (the caller publishes that before invoking this). It is a
-// no-op when there is no stored session, no active livestream, or the livestream
-// is already ended.
-func (atsync *ATProtoSynchronizer) endLivestreamForTeleport(ctx context.Context, repoDID string) {
- ctx = log.WithLogValues(ctx, "func", "endLivestreamForTeleport", "repoDID", repoDID)
+// no-op when there is no stored session, no origin livestream strongRef, the
+// referenced record is gone, or the livestream is already ended.
+func (atsync *ATProtoSynchronizer) endLivestreamForTeleport(ctx context.Context, repoDID string, livestreamRef comatproto.RepoStrongRef) {
+ ctx = log.WithLogValues(ctx, "func", "endLivestreamForTeleport", "repoDID", repoDID, "livestreamUri", livestreamRef.Uri)
+
+ // A teleport without an origin livestream strongRef (e.g. a record from
+ // before the field existed) cannot be tied to a specific stream. Rather
+ // than guess via "latest livestream" — which can terminate an unrelated,
+ // newer broadcast — treat it as a no-op.
+ if livestreamRef.Uri == "" {
+ log.Debug(ctx, "teleport has no origin livestream strongRef, skipping stream-end")
+ return
+ }
// A teleport record can arrive before the streamer has ever logged in to
// this node (e.g. multi-node setups). Without a stored session we have no
@@ -66,39 +84,49 @@ func (atsync *ATProtoSynchronizer) endLivestreamForTeleport(ctx context.Context,
return
}
- if err := atsync.endStreamersLivestream(ctx, repoDID, client); err != nil {
+ if err := atsync.endReferencedLivestream(ctx, repoDID, livestreamRef, client); err != nil {
log.Error(ctx, "failed to end livestream for teleport", "err", err)
return
}
- log.Log(ctx, "ended source livestream for teleport", "repoDID", repoDID)
+ log.Log(ctx, "ended source livestream for teleport", "repoDID", repoDID, "livestreamUri", livestreamRef.Uri)
}
-// endStreamersLivestream sets endedAt on the streamer's latest un-ended
-// place.stream.livestream record. It mirrors the record-ending half of
-// stopLivestream / endPriorLivestream: fetch the latest livestream, bail if
-// none or already ended, getRecord for a fresh swap CID, set endedAt, putRecord
-// with the swap so a concurrent update is rejected rather than clobbered.
+// endReferencedLivestream sets endedAt on the specific place.stream.livestream
+// record named by livestreamRef. It mirrors the record-ending half of
+// stopLivestream / endPriorLivestream, but targets the strongRef's record rather
+// than "the latest livestream": parse the strongRef URI for its rkey, getRecord
+// for a fresh swap CID (the record may have changed since the teleport was
+// created, e.g. a title update), set endedAt, putRecord with the swap so a
+// concurrent update is rejected rather than clobbered.
//
// Split from endLivestreamForTeleport so the session/client plumbing is
// testable independently of the record update.
-func (atsync *ATProtoSynchronizer) endStreamersLivestream(ctx context.Context, repoDID string, client *oatproxy.XrpcClient) error {
- livestream, err := atsync.Model.GetLatestLivestreamForRepo(repoDID)
+func (atsync *ATProtoSynchronizer) endReferencedLivestream(ctx context.Context, repoDID string, livestreamRef comatproto.RepoStrongRef, client *oatproxy.XrpcClient) error {
+ aturi, err := syntax.ParseATURI(livestreamRef.Uri)
if err != nil {
- return fmt.Errorf("get latest livestream: %w", err)
- }
- if livestream == nil || livestream.Livestream == nil {
- // Nothing live to end.
- return nil
+ return fmt.Errorf("parse livestream strongRef URI: %w", err)
}
+ rkey := aturi.RecordKey().String()
- livestreamView, err := livestream.ToLivestreamView()
+ // Fetch the current record (and its current CID to swap on) so we don't
+ // clobber a concurrent update, and so the putRecord is rejected if the
+ // record changed since the teleport was created.
+ getOutput := comatproto.RepoGetRecord_Output{}
+ err = client.Do(ctx, xrpc.Query, "application/json", "com.atproto.repo.getRecord", map[string]any{
+ "repo": repoDID,
+ "collection": "place.stream.livestream",
+ "rkey": rkey,
+ }, nil, &getOutput)
if err != nil {
- return fmt.Errorf("convert livestream to view: %w", err)
+ return fmt.Errorf("get livestream record: %w", err)
+ }
+ if getOutput.Value == nil {
+ return fmt.Errorf("getRecord returned no value for livestream %s", livestreamRef.Uri)
}
- rec, ok := livestreamView.Record.Val.(*placestream.Livestream)
+ rec, ok := getOutput.Value.Val.(*placestream.Livestream)
if !ok {
- return fmt.Errorf("livestream is not a streamplace livestream")
+ return fmt.Errorf("referenced record is not a streamplace livestream")
}
if rec.EndedAt != nil {
@@ -106,34 +134,15 @@ func (atsync *ATProtoSynchronizer) endStreamersLivestream(ctx context.Context, r
return nil
}
- aturi, err := syntax.ParseATURI(livestreamView.Uri)
- if err != nil {
- return fmt.Errorf("parse livestream URI: %w", err)
- }
-
- // Fetch the current CID to swap on, so we don't clobber a concurrent
- // update (and so the putRecord is rejected if the record changed).
- var swapRecord *string
- getOutput := comatproto.RepoGetRecord_Output{}
- err = client.Do(ctx, xrpc.Query, "application/json", "com.atproto.repo.getRecord", map[string]any{
- "repo": repoDID,
- "collection": "place.stream.livestream",
- "rkey": aturi.RecordKey().String(),
- }, nil, &getOutput)
- if err != nil {
- return fmt.Errorf("get livestream record: %w", err)
- }
- swapRecord = getOutput.Cid
-
now := time.Now().UTC().Format(util.ISO8601)
rec.EndedAt = &now
inp := comatproto.RepoPutRecord_Input{
Collection: "place.stream.livestream",
Record: &glex.LexiconTypeDecoder{Val: rec},
- Rkey: aturi.RecordKey().String(),
+ Rkey: rkey,
Repo: repoDID,
- SwapRecord: swapRecord,
+ SwapRecord: getOutput.Cid,
}
var out comatproto.RepoPutRecord_Output
if err := client.Do(ctx, xrpc.Procedure, "application/json", "com.atproto.repo.putRecord", map[string]any{}, inp, &out); err != nil {
diff --git a/pkg/placestream/liveteleport.go b/pkg/placestream/liveteleport.go
index 3292d1f55..2196758de 100644
--- a/pkg/placestream/liveteleport.go
+++ b/pkg/placestream/liveteleport.go
@@ -10,6 +10,7 @@ import (
glex "github.com/streamplace/glex/runtime"
cbg "github.com/whyrusleeping/cbor-gen"
+ comatproto "stream.place/streamplace/pkg/comatproto"
)
func init() {
@@ -21,6 +22,8 @@ type LiveTeleport struct {
LexiconTypeID string `json:"$type,omitempty"`
// 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"`
+ // livestream: The source livestream this teleport is sending viewers away from. When the teleport fires, this is the livestream that gets ended (the same update place.stream.live.stopLivestream performs), so the source streamer returns to pre-live. Teleports without an origin livestream are treated as a no-op.
+ Livestream comatproto.RepoStrongRef `json:"livestream"`
// startsAt: The time the teleport becomes active.
StartsAt string `json:"startsAt"`
// streamer: The DID of the streamer to teleport to.