From 0724ef56d6aa72836c42ad601c0d5ac0b65feccc Mon Sep 17 00:00:00 2001 From: Natalie Bridgers Date: Sat, 17 Oct 2026 07:03:19 +0000 Subject: [PATCH] Scope chat pins to broadcasts across chapters --- .../src/components/chat/mod-view.tsx | 72 ++++++----- .../stream-notification/pin-notification.tsx | 18 +-- .../src/livestream-provider/index.tsx | 8 +- js/components/src/livestream-store/hooks.tsx | 6 +- .../src/livestream-store/use-store.tsx | 3 +- .../src/streamplace-store/stream.tsx | 1 + .../tests/pinned-comment-watcher.test.tsx | 78 ++++++++++++ js/core/src/livestream-store/index.tsx | 1 + .../src/livestream-store/pinned-comment.ts | 17 +++ .../websocket-consumer.test.ts | 96 ++++++++++++++ .../chat/place-stream-chat-pinnedrecord.md | 20 +-- .../place-stream-moderation-createpin.md | 18 ++- .../content/docs/lex-reference/openapi.json | 7 +- .../lex-reference/place-stream-livestream.md | 36 +++--- js/e2e-web/flows/pin-broadcast.spec.ts | 118 ++++++++++++++++++ js/i18n/locales/en-US/common.ftl | 1 + .../components/stream/chat-message-menu.tsx | 26 +++- .../stream/stream-notifications.test.tsx | 65 +++++++++- .../stream/stream-notifications.tsx | 17 ++- js/web/src/hooks/use-moderation-actions.ts | 8 +- lexicons/place/stream/chat/pinnedRecord.json | 7 +- lexicons/place/stream/livestream.json | 5 + .../place/stream/moderation/createPin.json | 7 +- pkg/atproto/broadcast_test.go | 107 ++++++++++++++++ pkg/atproto/sync.go | 15 +++ pkg/model/pinned_record.go | 46 +++++-- pkg/model/pinned_record_test.go | 110 ++++++++++++++++ pkg/placestream/chatpinnedrecord.go | 4 +- pkg/placestream/livestream.go | 2 + pkg/placestream/moderationcreatepin.go | 4 +- pkg/spxrpc/pinned_record_test.go | 33 +++++ pkg/spxrpc/place_stream_moderation.go | 30 +++++ 32 files changed, 877 insertions(+), 109 deletions(-) create mode 100644 js/components/tests/pinned-comment-watcher.test.tsx create mode 100644 js/core/src/livestream-store/pinned-comment.ts create mode 100644 js/e2e-web/flows/pin-broadcast.spec.ts create mode 100644 pkg/atproto/broadcast_test.go create mode 100644 pkg/model/pinned_record_test.go create mode 100644 pkg/spxrpc/pinned_record_test.go diff --git a/js/components/src/components/chat/mod-view.tsx b/js/components/src/components/chat/mod-view.tsx index 1e3a8ceb5..9d92c7929 100644 --- a/js/components/src/components/chat/mod-view.tsx +++ b/js/components/src/components/chat/mod-view.tsx @@ -73,6 +73,11 @@ export function ModMenuContent({ const deleteChatMessage = useDeleteChatMessage(); // Get the streamer's DID from the livestream profile const streamerDID = useLivestreamStore((x) => x.profile?.did); + const livestream = useLivestreamStore((x) => x.livestream); + const broadcastURI = + livestream && !livestream.record.endedAt + ? (livestream.record.broadcast ?? livestream.uri) + : undefined; // Check moderation permissions for the current user on this streamer's channel const modPermissions = useCanModerate(streamerDID); @@ -95,6 +100,7 @@ export function ModMenuContent({ modPermissions={modPermissions} agent={agent} streamerDID={streamerDID} + broadcastURI={broadcastURI} hasAvailableActions={hasAvailableActions} isHideLoading={isHideLoading} isBlockLoading={isBlockLoading} @@ -149,6 +155,7 @@ interface ModViewContentProps { modPermissions: ModerationPermissions; agent: ReturnType; streamerDID?: string; + broadcastURI?: string; hasAvailableActions: boolean; isHideLoading: boolean; isBlockLoading: boolean; @@ -156,11 +163,7 @@ interface ModViewContentProps { setMessageRemoved: (removed: boolean) => void; createHideChat: (uri: string, streamerDID?: string) => Promise; createBlock: (did: string, streamerDID?: string) => Promise; - pinChatMessage: ( - messageUri: string, - streamerDID: string, - expiresAt?: string, - ) => Promise; + pinChatMessage: ReturnType; toast: ReturnType; setReportModalOpen: (open: boolean) => void; setReportSubject: (subject: any) => void; @@ -172,6 +175,7 @@ function ModViewContent({ modPermissions, agent, streamerDID, + broadcastURI, hasAvailableActions, isHideLoading, isBlockLoading, @@ -190,6 +194,25 @@ function ModViewContent({ const profileUrl = useNetworkProfileUrl(); const networkName = useNetworkName(); + const pinMessage = (options?: { + expiresAt?: string; + livestream?: string; + }) => { + if (!streamerDID) return; + pinChatMessage(message.uri, streamerDID, options) + .then(() => { + toast.show("Comment pinned", "", { duration: 3 }); + onOpenChange?.(false); + }) + .catch((e) => { + toast.show( + "Error pinning comment", + e instanceof Error ? e.message : "Failed to pin", + { duration: 5 }, + ); + }); + }; + return ( <> @@ -252,50 +275,25 @@ function ModViewContent({ { - if (!streamerDID) return; - pinChatMessage(message.uri, streamerDID) - .then(() => { - toast.show("Comment pinned", "", { duration: 3 }); - onOpenChange?.(false); - }) - .catch((e) => { - toast.show( - "Error pinning comment", - e instanceof Error ? e.message : "Failed to pin", - { duration: 5 }, - ); - }); + if (broadcastURI) + pinMessage({ livestream: broadcastURI }); }} > Until stream end + pinMessage()}> + Until unpinned + {[5, 10, 15, 30, 60].map((minutes) => ( { - if (!streamerDID) return; const expiresAt = new Date( Date.now() + minutes * 60 * 1000, ); - pinChatMessage( - message.uri, - streamerDID, - expiresAt.toISOString(), - ) - .then(() => { - toast.show("Comment pinned", "", { duration: 3 }); - onOpenChange?.(false); - }) - .catch((e) => { - toast.show( - "Error pinning comment", - e instanceof Error - ? e.message - : "Failed to pin", - { duration: 5 }, - ); - }); + pinMessage({ expiresAt: expiresAt.toISOString() }); }} > diff --git a/js/components/src/components/stream-notification/pin-notification.tsx b/js/components/src/components/stream-notification/pin-notification.tsx index 4550b85b5..3b4f25209 100644 --- a/js/components/src/components/stream-notification/pin-notification.tsx +++ b/js/components/src/components/stream-notification/pin-notification.tsx @@ -1,6 +1,6 @@ import { RichtextSegment, segmentize } from "@streamplace/core"; import { EyeOff, Pin, X } from "lucide-react-native"; -import { useEffect, useState } from "react"; +import { useEffect } from "react"; import { Linking, Pressable, View } from "react-native"; import { PinnedRecordViewHydrated, place } from "streamplace"; import { @@ -67,13 +67,14 @@ export function PinnedCommentNotification({ const messageRecord = message?.record as any; - const [expiresAt] = useState( - record.expiresAt ? new Date(record.expiresAt) : null, - ); + const expiresAt = + !record.livestream && record.expiresAt + ? Date.parse(record.expiresAt) + : null; useEffect(() => { - if (!expiresAt) return; - const remaining = expiresAt.getTime() - Date.now(); + if (expiresAt === null) return; + const remaining = expiresAt - Date.now(); if (remaining <= 0) { onDismiss(); return; @@ -90,7 +91,10 @@ export function PinnedCommentNotification({ : []; return ( - + { - store.setState({ pinnedComment: null }); + const current = store.getState().pinnedComment; + if ( + current?.uri === pinnedComment.uri && + current.cid === pinnedComment.cid + ) { + store.setState({ pinnedComment: null }); + } }, onUnpin: () => { if (!streamerDID) return; diff --git a/js/components/src/livestream-store/hooks.tsx b/js/components/src/livestream-store/hooks.tsx index 522699946..894faff0d 100644 --- a/js/components/src/livestream-store/hooks.tsx +++ b/js/components/src/livestream-store/hooks.tsx @@ -266,7 +266,7 @@ export const usePinChatMessage = () => { return async ( messageUri: string, streamerDID: string, - expiresAt?: string, + options?: { expiresAt?: string; livestream?: string }, ) => { if (!agent || !agent.did) { throw new Error("No PDS agent or user DID found"); @@ -278,7 +278,7 @@ export const usePinChatMessage = () => { $type: "place.stream.chat.pinnedRecord", pinnedMessage: messageUri, createdAt: new Date().toISOString(), - ...(expiresAt ? { expiresAt } : {}), + ...options, }; const result = await agent.com.atproto.repo.createRecord({ @@ -293,7 +293,7 @@ export const usePinChatMessage = () => { const result = await agent.client.call(place.stream.moderation.createPin, { streamer: streamerDID as any, messageUri: messageUri as any, - ...(expiresAt ? { expiresAt } : {}), + ...options, } as any); return result; }; diff --git a/js/components/src/livestream-store/use-store.tsx b/js/components/src/livestream-store/use-store.tsx index ed7db7b41..03f586bbe 100644 --- a/js/components/src/livestream-store/use-store.tsx +++ b/js/components/src/livestream-store/use-store.tsx @@ -3,6 +3,7 @@ // re-renders on changes. useLivestreamStoreOptional never throws when // there's no provider above (useful for mode-generic hooks like useTitle). import { + getActivePinnedComment, handleWebSocketMessages, makeLivestreamStore, type LivestreamState, @@ -57,7 +58,7 @@ export const useHandleWebsocketMessages = () => { export const useChat = () => useLivestreamStore((x) => x.chat); export const usePinnedComment = () => - useLivestreamStore((x) => x.pinnedComment); + useLivestreamStore(getActivePinnedComment); export const useProfile = () => useLivestreamStore((x) => x.profile); diff --git a/js/components/src/streamplace-store/stream.tsx b/js/components/src/streamplace-store/stream.tsx index 3db5c200e..85ba11855 100644 --- a/js/components/src/streamplace-store/stream.tsx +++ b/js/components/src/streamplace-store/stream.tsx @@ -261,6 +261,7 @@ export function useUpdateStreamRecord(customUrl: string | null = null) { url: finalUrl as any, createdAt: new Date().toISOString() as any, post: oldRecordValue.post, + broadcast: oldRecordValue.broadcast, thumb: thumbnail as any, activity: activity, tags: tags?.length ? tags : undefined, diff --git a/js/components/tests/pinned-comment-watcher.test.tsx b/js/components/tests/pinned-comment-watcher.test.tsx new file mode 100644 index 000000000..8b407e3c0 --- /dev/null +++ b/js/components/tests/pinned-comment-watcher.test.tsx @@ -0,0 +1,78 @@ +import { makeLivestreamStore } from "@streamplace/core"; +import { act } from "react"; +import { createRoot } from "react-dom/client"; +import { expect, it, vi } from "vitest"; +import { streamNotificationManager } from "../src/components/stream-notification/stream-notification-manager"; +import { PinnedCommentWatcher } from "../src/livestream-provider"; +import { LivestreamContext } from "../src/livestream-store/context"; + +vi.mock("../src/hooks", () => ({ useAvatars: () => ({}) })); +vi.mock("../src/streamplace-store", () => ({})); +vi.mock("../src/livestream-provider/websocket", () => ({})); +vi.mock("../src/livestream-store/hooks", () => ({ + useUnpinChatMessage: () => () => {}, +})); +vi.mock( + "../src/components/stream-notification", + () => + import("../src/components/stream-notification/stream-notification-manager"), +); +vi.mock("../src/components/stream-notification/pin-notification", () => ({ + PinnedCommentNotification: () => null, +})); +vi.mock("../src/components/stream-notification/teleport-notification", () => ({ + TeleportNotification: () => null, +})); + +it("keeps a new broadcast's pin when hiding the previous broadcast's notification", async () => { + vi.stubGlobal("IS_REACT_ACT_ENVIRONMENT", true); + const log = vi.spyOn(console, "log").mockImplementation(() => {}); + const store = makeLivestreamStore(); + const broadcast = (key: string) => + `at://did:plc:streamer/place.stream.livestream/${key}`; + const livestream = (key: string) => + ({ + uri: broadcast(key), + record: {}, + }) as never; + const pin = (key: string) => + ({ + uri: `at://did:plc:streamer/place.stream.chat.pinnedRecord/${key}`, + cid: key, + record: { livestream: broadcast(key) }, + }) as never; + const first = pin("first"); + const next = pin("next"); + store.setState({ livestream: livestream("first"), pinnedComment: first }); + const container = document.createElement("div"); + const root = createRoot(container); + try { + await act(async () => + root.render( + + + , + ), + ); + expect(streamNotificationManager.getAll()).toHaveLength(1); + + await act(async () => { + store.setState({ pinnedComment: next }); + }); + expect(streamNotificationManager.getAll()).toHaveLength(0); + expect(store.getState().pinnedComment).toBe(next); + + await act(async () => { + store.setState({ livestream: livestream("next") }); + }); + expect(streamNotificationManager.getAll()).toHaveLength(1); + await act(async () => { + streamNotificationManager.hide("pinned-comment"); + }); + expect(store.getState().pinnedComment).toBeNull(); + } finally { + await act(async () => root.unmount()); + log.mockRestore(); + vi.unstubAllGlobals(); + } +}); diff --git a/js/core/src/livestream-store/index.tsx b/js/core/src/livestream-store/index.tsx index c164cfeb8..ebee653c7 100644 --- a/js/core/src/livestream-store/index.tsx +++ b/js/core/src/livestream-store/index.tsx @@ -5,6 +5,7 @@ export * from "./chat-expiry"; export * from "./chat-reducer"; export * from "./connect"; +export * from "./pinned-comment"; export * from "./problems"; export * from "./state"; export * from "./store"; diff --git a/js/core/src/livestream-store/pinned-comment.ts b/js/core/src/livestream-store/pinned-comment.ts new file mode 100644 index 000000000..f351ceeeb --- /dev/null +++ b/js/core/src/livestream-store/pinned-comment.ts @@ -0,0 +1,17 @@ +import type { LivestreamState } from "./state"; + +export function getActivePinnedComment(state: LivestreamState) { + const { pinnedComment, livestream } = state; + const scope = pinnedComment?.record.livestream; + // Snapshots can deliver the pin before its livestream. Keep the received + // record in the store so a matching chapter can make it visible later. + if ( + scope && + (!livestream || + livestream.record.endedAt !== undefined || + (livestream.record.broadcast ?? livestream.uri) !== scope) + ) { + return null; + } + return pinnedComment; +} diff --git a/js/core/src/livestream-store/websocket-consumer.test.ts b/js/core/src/livestream-store/websocket-consumer.test.ts index 683d503a8..1d079f675 100644 --- a/js/core/src/livestream-store/websocket-consumer.test.ts +++ b/js/core/src/livestream-store/websocket-consumer.test.ts @@ -1,4 +1,5 @@ import { describe, expect, it } from "vitest"; +import { getActivePinnedComment } from "./pinned-comment"; import type { LivestreamState } from "./state"; import { makeLivestreamStore } from "./store"; import { handleWebSocketMessages } from "./websocket-consumer"; @@ -40,6 +41,101 @@ function makeState(overrides: Partial = {}): LivestreamState { } as LivestreamState; } +describe("broadcast-scoped pins", () => { + const root = "at://did:plc:streamer/place.stream.livestream/1"; + const chapter = "at://did:plc:streamer/place.stream.livestream/2"; + const endedAt = "2026-01-01T01:00:00Z"; + const livestream = (uri: string, broadcast?: string, ended?: string) => ({ + $type: "place.stream.livestream#livestreamView", + uri, + cid: "chapter", + author: { did: "did:plc:streamer", handle: "streamer.test" }, + indexedAt: "2026-01-01T00:00:00Z", + record: { + $type: "place.stream.livestream", + title: "chapter", + createdAt: "2026-01-01T00:00:00Z", + broadcast, + endedAt: ended, + }, + }); + const pin = (scope?: string) => ({ + $type: "place.stream.chat.defs#pinnedRecordView", + uri: "at://did:plc:streamer/place.stream.chat.pinnedRecord/1", + cid: "pin", + indexedAt: "2026-01-01T00:00:00Z", + record: { + $type: "place.stream.chat.pinnedRecord", + pinnedMessage: "at://did:plc:viewer/place.stream.chat.message/1", + createdAt: "2026-01-01T00:00:00Z", + livestream: scope, + }, + }); + + it.each([ + { + name: "a new chapter", + scope: root, + next: livestream(chapter, root), + keep: true, + }, + { + name: "the broadcast ending", + scope: root, + next: livestream(chapter, root, endedAt), + keep: false, + }, + { + name: "a separate broadcast", + scope: root, + next: livestream(chapter), + keep: false, + }, + { + name: "a permanent pin at stream end", + scope: undefined, + next: livestream(root, undefined, endedAt), + keep: true, + }, + ])("handles $name for connected viewers", ({ scope, next, keep }) => { + const pinned = pin(scope); + const state = handleWebSocketMessages(makeState(), [ + livestream(root), + pinned, + ]); + const result = handleWebSocketMessages(state, [next]); + expect(getActivePinnedComment(result)).toEqual(keep ? pinned : null); + }); + + it("ignores a scoped pin arriving after the broadcast has ended", () => { + const state = handleWebSocketMessages(makeState(), [ + livestream(chapter, root, endedAt), + ]); + const result = handleWebSocketMessages(state, [pin(root)]); + expect(getActivePinnedComment(result)).toBeNull(); + }); + + it("keeps a pin across an initial snapshot with the pin before its chapter", () => { + const pinned = pin(root); + const result = handleWebSocketMessages(makeState(), [ + pinned, + livestream(chapter, root), + ]); + expect(getActivePinnedComment(result)).toEqual(pinned); + }); + + it("retains a new broadcast's pin before its reconnect chapter arrives", () => { + const pinned = pin(chapter); + const disconnected = handleWebSocketMessages(makeState(), [ + livestream(root, undefined, endedAt), + ]); + const withPin = handleWebSocketMessages(disconnected, [pinned]); + expect(getActivePinnedComment(withPin)).toBeNull(); + const reconnected = handleWebSocketMessages(withPin, [livestream(chapter)]); + expect(getActivePinnedComment(reconnected)).toEqual(pinned); + }); +}); + const MODERATOR_DID = "did:plc:mod"; const OTHER_MODERATOR_DID = "did:plc:othermod"; const PERMISSION_URI = diff --git a/js/docs/src/content/docs/lex-reference/chat/place-stream-chat-pinnedrecord.md b/js/docs/src/content/docs/lex-reference/chat/place-stream-chat-pinnedrecord.md index fc88032a4..f25ed1848 100644 --- a/js/docs/src/content/docs/lex-reference/chat/place-stream-chat-pinnedrecord.md +++ b/js/docs/src/content/docs/lex-reference/chat/place-stream-chat-pinnedrecord.md @@ -19,12 +19,13 @@ Record pinning a chat message for prominent display. **Record Properties:** -| Name | Type | Req'd | Description | Constraints | -| --------------- | -------- | ----- | --------------------------------------------------------------------------------- | ------------------ | -| `pinnedMessage` | `string` | ✅ | AT-URI of the pinned chat message. | Format: `at-uri` | -| `pinnedBy` | `string` | ❌ | DID of the user who pinned the message. | Format: `did` | -| `createdAt` | `string` | ✅ | When this pin was created. | Format: `datetime` | -| `expiresAt` | `string` | ❌ | Optional expiration time. If set, the pin is considered inactive after this time. | Format: `datetime` | +| Name | Type | Req'd | Description | Constraints | +| --------------- | -------- | ----- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------ | +| `pinnedMessage` | `string` | ✅ | AT-URI of the pinned chat message. | Format: `at-uri` | +| `pinnedBy` | `string` | ❌ | DID of the user who pinned the message. | Format: `did` | +| `createdAt` | `string` | ✅ | When this pin was created. | Format: `datetime` | +| `expiresAt` | `string` | ❌ | Optional expiration time, used when livestream is not set. | Format: `datetime` | +| `livestream` | `string` | ❌ | AT-URI of the first chapter of the streamer's broadcast. The pin remains active across chapter changes until that broadcast ends. Takes precedence over expiresAt. Without either field, the pin remains until manually unpinned. | Format: `at-uri` | --- @@ -61,7 +62,12 @@ Record pinning a chat message for prominent display. "expiresAt": { "type": "string", "format": "datetime", - "description": "Optional expiration time. If set, the pin is considered inactive after this time." + "description": "Optional expiration time, used when livestream is not set." + }, + "livestream": { + "type": "string", + "format": "at-uri", + "description": "AT-URI of the first chapter of the streamer's broadcast. The pin remains active across chapter changes until that broadcast ends. Takes precedence over expiresAt. Without either field, the pin remains until manually unpinned." } } } diff --git a/js/docs/src/content/docs/lex-reference/moderation/place-stream-moderation-createpin.md b/js/docs/src/content/docs/lex-reference/moderation/place-stream-moderation-createpin.md index 5c1ce46eb..f7f6c7dc8 100644 --- a/js/docs/src/content/docs/lex-reference/moderation/place-stream-moderation-createpin.md +++ b/js/docs/src/content/docs/lex-reference/moderation/place-stream-moderation-createpin.md @@ -24,11 +24,12 @@ Pin a chat message on behalf of a streamer. Requires 'message.pin' permission. C **Schema Type:** `object` -| Name | Type | Req'd | Description | Constraints | -| ------------ | -------- | ----- | -------------------------------------- | ------------------ | -| `streamer` | `string` | ✅ | The DID of the streamer. | Format: `did` | -| `messageUri` | `string` | ✅ | The AT-URI of the chat message to pin. | Format: `at-uri` | -| `expiresAt` | `string` | ❌ | Optional expiration time for this pin. | Format: `datetime` | +| Name | Type | Req'd | Description | Constraints | +| ------------ | -------- | ----- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------ | +| `streamer` | `string` | ✅ | The DID of the streamer. | Format: `did` | +| `messageUri` | `string` | ✅ | The AT-URI of the chat message to pin. | Format: `at-uri` | +| `expiresAt` | `string` | ❌ | Optional expiration time for this pin, used when livestream is not set. | Format: `datetime` | +| `livestream` | `string` | ❌ | AT-URI of the first chapter of the streamer's current broadcast. The pin remains active across chapter changes until that broadcast ends. Without either livestream or expiresAt, the pin remains until manually unpinned. | Format: `at-uri` | **Output:** @@ -79,7 +80,12 @@ Pin a chat message on behalf of a streamer. Requires 'message.pin' permission. C "expiresAt": { "type": "string", "format": "datetime", - "description": "Optional expiration time for this pin." + "description": "Optional expiration time for this pin, used when livestream is not set." + }, + "livestream": { + "type": "string", + "format": "at-uri", + "description": "AT-URI of the first chapter of the streamer's current broadcast. The pin remains active across chapter changes until that broadcast ends. Without either livestream or expiresAt, the pin remains until manually unpinned." } } } diff --git a/js/docs/src/content/docs/lex-reference/openapi.json b/js/docs/src/content/docs/lex-reference/openapi.json index 73f18aef1..92472761a 100644 --- a/js/docs/src/content/docs/lex-reference/openapi.json +++ b/js/docs/src/content/docs/lex-reference/openapi.json @@ -2157,8 +2157,13 @@ }, "expiresAt": { "type": "string", - "description": "Optional expiration time for this pin.", + "description": "Optional expiration time for this pin, used when livestream is not set.", "format": "date-time" + }, + "livestream": { + "type": "string", + "description": "AT-URI of the first chapter of the streamer's current broadcast. The pin remains active across chapter changes until that broadcast ends. Without either livestream or expiresAt, the pin remains until manually unpinned.", + "format": "uri" } }, "required": ["streamer", "messageUri"] 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 3e504ee3d..375f1b226 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 @@ -19,21 +19,22 @@ Record announcing a livestream is happening **Record Properties:** -| Name | Type | Req'd | Description | Constraints | -| ---------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------- | -| `title` | `string` | ✅ | The title of the livestream, as it will be announced to followers. | Max Length: 1400
Max Graphemes: 140 | -| `url` | `string` | ❌ | The URL where this stream can be found. This is primarily a hint for other Streamplace nodes to locate and replicate the stream. | Format: `uri` | -| `createdAt` | `string` | ✅ | Client-declared timestamp when this livestream started. | Format: `datetime` | -| `lastSeenAt` | `string` | ❌ | Client-declared timestamp when this livestream was last seen by the Streamplace station. | Format: `datetime` | -| `endedAt` | `string` | ❌ | Client-declared timestamp when this livestream ended. Ended livestreams are not supposed to start up again. | Format: `datetime` | -| `idleTimeoutSeconds` | `integer` | ❌ | Time in seconds after which this livestream should be automatically ended if idle. Zero means no timeout. | | -| `post` | [`com.atproto.repo.strongRef`](https://github.com/bluesky-social/atproto/tree/main/lexicons/com/atproto/repo/strongref.json#undefined) | ❌ | The post that announced this livestream. | | -| `agent` | `string` | ❌ | The source of the livestream, if available, in a User Agent format: ` / ` e.g. Streamplace/0.7.5 iOS | | -| `canonicalUrl` | `string` | ❌ | The primary URL where this livestream can be viewed, if available. | Format: `uri` | -| `thumb` | `blob` | ❌ | | Accept: `image/*`
Max Size: 1000000 bytes | -| `notificationSettings` | [`place.stream.livestream#notificationSettings`](/lex-reference/place-stream-livestream#notificationsettings) | ❌ | | | -| `activity` | Union of:
  [`place.stream.defs#activityGame`](/lex-reference/place-stream-defs#activitygame)
  [`place.stream.defs#activityLabel`](/lex-reference/place-stream-defs#activitylabel) | ❌ | The game or activity being streamed. | | -| `tags` | Array of `string` | ❌ | Freeform tags for this stream. Each tag must be alphanumeric (a-z, A-Z, 0-9) plus colon. Tags with colons indicate a specific tag group (e.g. 'lang:en' indicates the stream's primary language). | Max Items: 10 | +| Name | Type | Req'd | Description | Constraints | +| ---------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | --------------------------------------------- | +| `title` | `string` | ✅ | The title of the livestream, as it will be announced to followers. | Max Length: 1400
Max Graphemes: 140 | +| `url` | `string` | ❌ | The URL where this stream can be found. This is primarily a hint for other Streamplace nodes to locate and replicate the stream. | Format: `uri` | +| `createdAt` | `string` | ✅ | Client-declared timestamp when this livestream started. | Format: `datetime` | +| `broadcast` | `string` | ❌ | AT-URI of the first chapter in this continuous broadcast, owned by the same repo. Omitted on the first chapter, whose own URI identifies the broadcast. Later chapters preserve this reference; a separate broadcast starts with a new identity. | Format: `at-uri` | +| `lastSeenAt` | `string` | ❌ | Client-declared timestamp when this livestream was last seen by the Streamplace station. | Format: `datetime` | +| `endedAt` | `string` | ❌ | Client-declared timestamp when this livestream ended. Ended livestreams are not supposed to start up again. | Format: `datetime` | +| `idleTimeoutSeconds` | `integer` | ❌ | Time in seconds after which this livestream should be automatically ended if idle. Zero means no timeout. | | +| `post` | [`com.atproto.repo.strongRef`](https://github.com/bluesky-social/atproto/tree/main/lexicons/com/atproto/repo/strongref.json#undefined) | ❌ | The post that announced this livestream. | | +| `agent` | `string` | ❌ | The source of the livestream, if available, in a User Agent format: ` / ` e.g. Streamplace/0.7.5 iOS | | +| `canonicalUrl` | `string` | ❌ | The primary URL where this livestream can be viewed, if available. | Format: `uri` | +| `thumb` | `blob` | ❌ | | Accept: `image/*`
Max Size: 1000000 bytes | +| `notificationSettings` | [`place.stream.livestream#notificationSettings`](/lex-reference/place-stream-livestream#notificationsettings) | ❌ | | | +| `activity` | Union of:
  [`place.stream.defs#activityGame`](/lex-reference/place-stream-defs#activitygame)
  [`place.stream.defs#activityLabel`](/lex-reference/place-stream-defs#activitylabel) | ❌ | The game or activity being streamed. | | +| `tags` | Array of `string` | ❌ | Freeform tags for this stream. Each tag must be alphanumeric (a-z, A-Z, 0-9) plus colon. Tags with colons indicate a specific tag group (e.g. 'lang:en' indicates the stream's primary language). | Max Items: 10 | --- @@ -163,6 +164,11 @@ Record announcing a livestream is happening "format": "datetime", "description": "Client-declared timestamp when this livestream started." }, + "broadcast": { + "type": "string", + "format": "at-uri", + "description": "AT-URI of the first chapter in this continuous broadcast, owned by the same repo. Omitted on the first chapter, whose own URI identifies the broadcast. Later chapters preserve this reference; a separate broadcast starts with a new identity." + }, "lastSeenAt": { "type": "string", "format": "datetime", diff --git a/js/e2e-web/flows/pin-broadcast.spec.ts b/js/e2e-web/flows/pin-broadcast.spec.ts new file mode 100644 index 000000000..37a3b6f2e --- /dev/null +++ b/js/e2e-web/flows/pin-broadcast.spec.ts @@ -0,0 +1,118 @@ +import { expect, test, type Page } from "@playwright/test"; + +const pdsUrl = process.env.PDS_HTTPS_URL; +test.skip(!pdsUrl, "needs the harness's HTTPS PDS"); + +async function pds(page: Page, method: string, input: object, token?: string) { + return page.evaluate( + async ({ url, method, input, token }) => { + const response = await fetch(`${url}/xrpc/${method}`, { + method: "POST", + headers: { + "Content-Type": "application/json", + ...(token ? { Authorization: `Bearer ${token}` } : {}), + }, + body: JSON.stringify(input), + }); + if (!response.ok) + throw new Error( + `${method}: ${response.status} ${await response.text()}`, + ); + return response.json(); + }, + { url: pdsUrl!, method, input, token }, + ); +} + +for (const frontend of ["app", "web"]) { + test(`${frontend}: pins survive chapters and expire for connected and returning viewers`, async ({ + page, + }) => { + if (frontend === "web") { + await page + .context() + .addCookies([ + { name: "sp_web_beta", value: "1", url: process.env.SERVER_URL! }, + ]); + } + await page.goto("/"); + const suffix = process.env.ACCOUNT_HANDLE!.split(".").slice(1).join("."); + const account = await pds(page, "com.atproto.server.createAccount", { + handle: `pin-${Date.now()}.${suffix}`, + email: `pin-${Date.now()}@example.invalid`, + password: "local-pin-test-password", + }); + const write = (collection: string, record: object) => + pds( + page, + "com.atproto.repo.createRecord", + { repo: account.did, collection, record }, + account.accessJwt, + ); + const record = { + $type: "place.stream.livestream", + title: "pin test broadcast", + createdAt: new Date().toISOString(), + }; + const first = await write("place.stream.livestream", record); + const message = await write("place.stream.chat.message", { + $type: "place.stream.chat.message", + text: "pin across chapters", + streamer: account.did, + createdAt: new Date().toISOString(), + }); + await write("place.stream.chat.pinnedRecord", { + $type: "place.stream.chat.pinnedRecord", + pinnedMessage: message.uri, + livestream: first.uri, + createdAt: new Date().toISOString(), + }); + + await page.goto(`/${account.did}`); + const pinned = page.getByTestId("pinned-message"); + await expect(pinned).toContainText("pin across chapters"); + + const chapterRecord = { + ...record, + title: "pin test next chapter", + broadcast: first.uri, + createdAt: new Date().toISOString(), + }; + const chapter = await write("place.stream.livestream", chapterRecord); + await expect(page.getByText("pin test next chapter").first()).toBeVisible(); + await expect(pinned).toContainText("pin across chapters"); + await page.reload(); + await expect(pinned).toContainText("pin across chapters"); + + await pds( + page, + "com.atproto.repo.putRecord", + { + repo: account.did, + collection: "place.stream.livestream", + rkey: chapter.uri.split("/").pop(), + record: { ...chapterRecord, endedAt: new Date().toISOString() }, + }, + account.accessJwt, + ); + await expect(pinned).toHaveCount(0); + await page.reload(); + await expect(page.getByText("pin test next chapter").first()).toBeVisible(); + await expect(pinned).toHaveCount(0); + + await write("place.stream.livestream", { + ...record, + title: "pin test separate broadcast", + createdAt: new Date().toISOString(), + }); + await expect( + page.getByText("pin test separate broadcast").first(), + ).toBeVisible(); + await expect(pinned).toHaveCount(0); + await page.reload(); + await expect( + page.getByText("pin test separate broadcast").first(), + ).toBeVisible(); + await expect(pinned).toHaveCount(0); + }); +} diff --git a/js/i18n/locales/en-US/common.ftl b/js/i18n/locales/en-US/common.ftl index afb35b282..9a35f8ab7 100644 --- a/js/i18n/locales/en-US/common.ftl +++ b/js/i18n/locales/en-US/common.ftl @@ -127,6 +127,7 @@ chat-close = Close chat chat-open = Open chat chat-moderation-menu = Message options chat-pin-until-stream-end = Until stream end +chat-pin-forever = Until unpinned chat-pin-duration-minutes = { $count } min chat-hide-message = Hide message chat-block-user = Block user diff --git a/js/web/src/components/stream/chat-message-menu.tsx b/js/web/src/components/stream/chat-message-menu.tsx index cab1e1be4..46c584c7c 100644 --- a/js/web/src/components/stream/chat-message-menu.tsx +++ b/js/web/src/components/stream/chat-message-menu.tsx @@ -15,6 +15,7 @@ import { import { useCallback, useEffect, useRef, useState } from "react"; import { useTranslation } from "react-i18next"; import type { ChatMessageViewHydrated } from "streamplace"; +import { useStore } from "zustand"; import { DropdownMenu, DropdownMenuContent, @@ -52,6 +53,11 @@ export function ChatMessageMenu({ const { pdsAgent, did } = useSession(); const toast = useToast(); const { blockUser, hideMessage, pinMessage } = useModerationActions(); + const livestream = useStore(store, (s) => s.livestream); + const broadcast = + livestream && !livestream.record.endedAt + ? (livestream.record.broadcast ?? livestream.uri) + : undefined; const [confirmingDelete, setConfirmingDelete] = useState(false); const deleteTimer = useRef | null>(null); @@ -124,10 +130,10 @@ export function ChatMessageMenu({ ]); const handlePin = useCallback( - async (expiresAt?: string) => { + async (options?: { expiresAt?: string; livestream?: string }) => { if (!streamerDid) return; try { - await pinMessage(message.uri, streamerDid, expiresAt); + await pinMessage(message.uri, streamerDid, options); toast.show(t("moderation-message-pinned"), "", { duration: 3000 }); } catch (e) { fail(t("moderation-pin-failed"), e); @@ -183,18 +189,26 @@ export function ChatMessageMenu({ - handlePin()}> + { + if (broadcast) handlePin({ livestream: broadcast }); + }} + > {t("chat-pin-until-stream-end")} + handlePin()}> + {t("chat-pin-forever")} + {PIN_DURATIONS_MINUTES.map((minutes) => ( - handlePin( - new Date( + handlePin({ + expiresAt: new Date( Date.now() + minutes * 60 * 1000, ).toISOString(), - ) + }) } > {t("chat-pin-duration-minutes", { count: minutes })} diff --git a/js/web/src/components/stream/stream-notifications.test.tsx b/js/web/src/components/stream/stream-notifications.test.tsx index 47882eab7..24413bd13 100644 --- a/js/web/src/components/stream/stream-notifications.test.tsx +++ b/js/web/src/components/stream/stream-notifications.test.tsx @@ -1,4 +1,7 @@ -import { makeLivestreamStore } from "@streamplace/core"; +import { + handleWebSocketMessages, + makeLivestreamStore, +} from "@streamplace/core"; import { act } from "react"; import { createRoot, type Root } from "react-dom/client"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; @@ -70,10 +73,10 @@ describe("TeleportNotification", () => { expect(container.textContent).toBe(""); }); - function pin(text: string, expiresAt?: string) { + function pin(text: string, expiresAt?: string, livestream?: string) { return { uri: `at://did:plc:streamer/place.stream.chat.pinnedRecord/${text}`, - record: { expiresAt }, + record: { expiresAt, livestream }, message: { author: { did: "did:plc:author", handle: "author.example" }, record: { text }, @@ -81,6 +84,62 @@ describe("TeleportNotification", () => { } as never; } + function livestream(broadcast: string, endedAt?: string) { + return { + $type: "place.stream.livestream#livestreamView", + uri: `${broadcast}-chapter`, + author: { did: "did:plc:streamer" }, + record: { broadcast, endedAt }, + } as never; + } + + it("keeps a broadcast-scoped pin when its optional timestamp has expired", async () => { + store.setState({ + livestream: livestream("at://did:plc:streamer/place.stream.livestream/1"), + pinnedComment: pin( + "broadcast pin", + new Date(Date.now() - 1_000).toISOString(), + "at://did:plc:streamer/place.stream.livestream/1", + ), + }); + await act(async () => render()); + expect(container.textContent).toContain("broadcast pin"); + }); + + it("shows a reconnect pin after its matching chapter and hides it at broadcast end", async () => { + const broadcast = "at://did:plc:streamer/place.stream.livestream/2"; + store.setState({ + livestream: livestream("at://did:plc:streamer/place.stream.livestream/1"), + pinnedComment: pin("new broadcast pin", undefined, broadcast), + }); + await act(async () => render()); + expect(container.textContent).toBe(""); + + await act(async () => { + store.setState((state) => + handleWebSocketMessages(state, [livestream(broadcast)]), + ); + }); + expect(container.textContent).toContain("new broadcast pin"); + + await act(async () => { + store.setState((state) => + handleWebSocketMessages(state, [ + livestream(broadcast, new Date().toISOString()), + ]), + ); + }); + expect(container.textContent).toBe(""); + }); + + it("expires a timed pin even at the Unix epoch", async () => { + store.setState({ + pinnedComment: pin("expired pin", new Date(0).toISOString()), + }); + await act(async () => render()); + expect(container.textContent).toBe(""); + }); + it("keeps a pin visible when the teleport countdown expires", async () => { store.setState({ pinnedComment: pin("pinned message"), diff --git a/js/web/src/components/stream/stream-notifications.tsx b/js/web/src/components/stream/stream-notifications.tsx index da182eb24..7518cc404 100644 --- a/js/web/src/components/stream/stream-notifications.tsx +++ b/js/web/src/components/stream/stream-notifications.tsx @@ -3,6 +3,7 @@ import type { Link as FacetLink } from "@atproto/api/dist/client/types/app/bsky/ import type { LivestreamStore } from "@streamplace/core"; import { deleteTeleport, + getActivePinnedComment, segmentize, type Facet, type FacetFeature, @@ -51,7 +52,7 @@ export function StreamNotifications({ const state = useStore( store, useShallow((s) => ({ - pinnedComment: s.pinnedComment, + pinnedComment: getActivePinnedComment(s), activeTeleport: s.activeTeleport, })), ); @@ -136,10 +137,13 @@ function PinnedNotification({ const [dismissed, setDismissed] = useState(false); // Handle TTL expiry - const expiresAt = record.expiresAt ? new Date(record.expiresAt) : null; + const expiresAt = + !record.livestream && record.expiresAt + ? Date.parse(record.expiresAt) + : null; useEffect(() => { - if (!expiresAt) return; - const remaining = expiresAt.getTime() - Date.now(); + if (expiresAt === null) return; + const remaining = expiresAt - Date.now(); if (remaining <= 0) { setDismissed(true); return; @@ -175,7 +179,10 @@ function PinnedNotification({ if (dismissed) return null; return ( -
+
{ const agent = requireAgent(); if (agent.did === streamerDid) { @@ -69,14 +69,15 @@ export function useModerationActions() { $type: "place.stream.chat.pinnedRecord", pinnedMessage: messageUri, createdAt: new Date().toISOString(), - ...(expiresAt ? { expiresAt } : {}), + ...options, }, }); } return agent.client.call(place.stream.moderation.createPin, { streamer: streamerDid as any, messageUri: messageUri as any, - ...(expiresAt ? { expiresAt: expiresAt as any } : {}), + ...(options?.expiresAt ? { expiresAt: options.expiresAt as any } : {}), + ...(options?.livestream ? { livestream: options.livestream as any } : {}), }); }; @@ -116,6 +117,7 @@ export function useModerationActions() { place.stream.livestream, { ...(value as object), + broadcast: (value.broadcast ?? livestreamUri) as any, title, createdAt: new Date().toISOString() as any, }, diff --git a/lexicons/place/stream/chat/pinnedRecord.json b/lexicons/place/stream/chat/pinnedRecord.json index 32b5b33e0..5571e91c9 100644 --- a/lexicons/place/stream/chat/pinnedRecord.json +++ b/lexicons/place/stream/chat/pinnedRecord.json @@ -28,7 +28,12 @@ "expiresAt": { "type": "string", "format": "datetime", - "description": "Optional expiration time. If set, the pin is considered inactive after this time." + "description": "Optional expiration time, used when livestream is not set." + }, + "livestream": { + "type": "string", + "format": "at-uri", + "description": "AT-URI of the first chapter of the streamer's broadcast. The pin remains active across chapter changes until that broadcast ends. Takes precedence over expiresAt. Without either field, the pin remains until manually unpinned." } } } diff --git a/lexicons/place/stream/livestream.json b/lexicons/place/stream/livestream.json index ed7c50bc8..a99d82042 100644 --- a/lexicons/place/stream/livestream.json +++ b/lexicons/place/stream/livestream.json @@ -26,6 +26,11 @@ "format": "datetime", "description": "Client-declared timestamp when this livestream started." }, + "broadcast": { + "type": "string", + "format": "at-uri", + "description": "AT-URI of the first chapter in this continuous broadcast, owned by the same repo. Omitted on the first chapter, whose own URI identifies the broadcast. Later chapters preserve this reference; a separate broadcast starts with a new identity." + }, "lastSeenAt": { "type": "string", "format": "datetime", diff --git a/lexicons/place/stream/moderation/createPin.json b/lexicons/place/stream/moderation/createPin.json index b7be8c50f..299b6b928 100644 --- a/lexicons/place/stream/moderation/createPin.json +++ b/lexicons/place/stream/moderation/createPin.json @@ -24,7 +24,12 @@ "expiresAt": { "type": "string", "format": "datetime", - "description": "Optional expiration time for this pin." + "description": "Optional expiration time for this pin, used when livestream is not set." + }, + "livestream": { + "type": "string", + "format": "at-uri", + "description": "AT-URI of the first chapter of the streamer's current broadcast. The pin remains active across chapter changes until that broadcast ends. Without either livestream or expiresAt, the pin remains until manually unpinned." } } } diff --git a/pkg/atproto/broadcast_test.go b/pkg/atproto/broadcast_test.go new file mode 100644 index 000000000..a922424ba --- /dev/null +++ b/pkg/atproto/broadcast_test.go @@ -0,0 +1,107 @@ +package atproto + +import ( + "bytes" + "context" + "testing" + + "github.com/bluesky-social/indigo/atproto/syntax" + "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/model" + "stream.place/streamplace/pkg/placestream" + "stream.place/streamplace/pkg/spid" +) + +func TestLivestreamBroadcastReference(t *testing.T) { + const did = "did:plc:aaaaaaaaaaaaaaaaaaaaaaaa" + for _, tt := range []struct { + name string + uri string + want bool + }{ + {name: "first chapter", want: true}, + {name: "continuing broadcast", uri: "at://" + did + "/place.stream.livestream/3lfirst00000", want: true}, + {name: "another streamer", uri: "at://did:plc:bbbbbbbbbbbbbbbbbbbbbbbb/place.stream.livestream/3lfirst00000"}, + {name: "another collection", uri: "at://" + did + "/place.stream.chat.message/3lfirst00000"}, + {name: "missing record key", uri: "at://" + did + "/place.stream.livestream"}, + {name: "malformed URI", uri: "not-an-at-uri"}, + } { + t.Run(tt.name, func(t *testing.T) { + ctx := context.Background() + atsync, mod, _ := offlineSynchronizer(t) + require.NoError(t, mod.UpdateRepo(&model.Repo{DID: did, Handle: "streamer.test", Version: "3lrev00000000"})) + rec := &placestream.Livestream{ + LexiconTypeID: "place.stream.livestream", + CreatedAt: "2026-01-01T00:00:00Z", + Title: "chapter", + } + if tt.uri != "" { + rec.Broadcast = &tt.uri + } + var buf bytes.Buffer + require.NoError(t, rec.MarshalCBOR(&buf)) + body := buf.Bytes() + cid, err := spid.GetCID(rec) + require.NoError(t, err) + err = atsync.handleCreateUpdate(ctx, did, syntax.RecordKey("3lchapter0000"), &body, cid.String(), syntax.NSID("place.stream.livestream"), false, true, "") + if tt.want { + require.NoError(t, err) + } else { + require.Error(t, err) + } + ls, err := mod.GetLatestLivestreamForRepo(did) + require.NoError(t, err) + if tt.want { + require.NotNil(t, ls) + } else { + require.Nil(t, ls) + } + }) + } +} + +func TestPinnedBroadcastReference(t *testing.T) { + const did = "did:plc:aaaaaaaaaaaaaaaaaaaaaaaa" + for _, tt := range []struct { + name string + uri string + want bool + }{ + {name: "own broadcast", uri: "at://" + did + "/place.stream.livestream/3lfirst00000", want: true}, + {name: "another streamer", uri: "at://did:plc:bbbbbbbbbbbbbbbbbbbbbbbb/place.stream.livestream/3lfirst00000"}, + {name: "another collection", uri: "at://" + did + "/place.stream.chat.message/3lfirst00000"}, + {name: "missing record key", uri: "at://" + did + "/place.stream.livestream"}, + {name: "malformed URI", uri: "not-an-at-uri"}, + } { + t.Run(tt.name, func(t *testing.T) { + ctx := context.Background() + atsync, mod, _ := offlineSynchronizer(t) + require.NoError(t, mod.UpdateRepo(&model.Repo{DID: did, Handle: "streamer.test", Version: "3lrev00000000"})) + rec := &placestream.ChatPinnedRecord{ + LexiconTypeID: "place.stream.chat.pinnedRecord", + CreatedAt: "2026-01-01T00:00:00Z", + PinnedMessage: "at://" + did + "/place.stream.chat.message/3lmessage000", + Livestream: &tt.uri, + } + var buf bytes.Buffer + require.NoError(t, rec.MarshalCBOR(&buf)) + body := buf.Bytes() + cid, err := spid.GetCID(rec) + require.NoError(t, err) + err = atsync.handleCreateUpdate(ctx, did, syntax.RecordKey("3lpin0000000"), &body, cid.String(), syntax.NSID("place.stream.chat.pinnedRecord"), false, true, "") + if tt.want { + require.NoError(t, err) + } else { + require.Error(t, err) + } + pin, err := mod.GetPinnedRecord(ctx, "at://"+did+"/place.stream.chat.pinnedRecord/3lpin0000000") + require.NoError(t, err) + if tt.want { + require.NotNil(t, pin) + require.Equal(t, tt.uri, pin.LivestreamURI) + } else { + require.Nil(t, pin) + } + }) + } +} diff --git a/pkg/atproto/sync.go b/pkg/atproto/sync.go index ac84541a5..eb1341ada 100644 --- a/pkg/atproto/sync.go +++ b/pkg/atproto/sync.go @@ -309,6 +309,12 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD go atsync.Bus.Publish(userDID, streamplaceGate) case *placestream.ChatPinnedRecord: + if rec.Livestream != nil { + uri, err := syntax.ParseATURI(*rec.Livestream) + if err != nil || uri.Authority().String() != userDID || uri.Collection().String() != constants.PLACE_STREAM_LIVESTREAM || uri.RecordKey().String() == "" { + return fmt.Errorf("pin livestream must reference the streamer's broadcast") + } + } repo, err := atsync.SyncBlueskyRepoCached(ctx, userDID) if err != nil { return fmt.Errorf("failed to sync bluesky repo: %w", err) @@ -353,6 +359,9 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD Repo: repo, ExpiresAt: expiresAt, } + if rec.Livestream != nil { + pin.LivestreamURI = *rec.Livestream + } err = atsync.Model.CreatePinnedRecord(ctx, pin) if errors.Is(err, model.ErrAlreadyIndexed) { return nil @@ -526,6 +535,12 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD // we don't know about this repo return nil } + if rec.Broadcast != nil { + uri, err := syntax.ParseATURI(*rec.Broadcast) + if err != nil || uri.Authority().String() != userDID || uri.Collection().String() != constants.PLACE_STREAM_LIVESTREAM || uri.RecordKey().String() == "" { + return fmt.Errorf("broadcast must reference the streamer's first chapter") + } + } createdAt, err := time.Parse(time.RFC3339, rec.CreatedAt) if err != nil { log.Error(ctx, "failed to parse createdAt", "err", err) diff --git a/pkg/model/pinned_record.go b/pkg/model/pinned_record.go index a541aa4db..4a78671f1 100644 --- a/pkg/model/pinned_record.go +++ b/pkg/model/pinned_record.go @@ -3,9 +3,11 @@ package model import ( "context" "errors" + "fmt" "time" "github.com/bluesky-social/indigo/util" + glex "github.com/streamplace/glex/runtime" "gorm.io/gorm" "stream.place/streamplace/pkg/placestream" ) @@ -20,6 +22,7 @@ type PinnedRecord struct { IndexedAt *time.Time `gorm:"column:indexed_at" json:"indexedAt"` ExpiresAt *time.Time `gorm:"column:expires_at" json:"expiresAt"` CreatedAt time.Time `gorm:"column:created_at" json:"createdAt"` + LivestreamURI string `gorm:"column:livestream_uri" json:"livestreamURI,omitempty"` } func (p *PinnedRecord) ToStreamplacePinnedRecord() (placestream.ChatPinnedRecord, error) { @@ -32,18 +35,16 @@ func (p *PinnedRecord) ToStreamplacePinnedRecord() (placestream.ChatPinnedRecord s := p.ExpiresAt.UTC().Format(util.ISO8601) rec.ExpiresAt = &s } + if p.LivestreamURI != "" { + rec.Livestream = &p.LivestreamURI + } return rec, nil } func (p *PinnedRecord) ToStreamplacePinnedRecordView() (placestream.ChatDefs_PinnedRecordView, error) { - pr := placestream.ChatPinnedRecord{ - LexiconTypeID: "place.stream.chat.pinnedRecord", - PinnedMessage: p.PinnedMessage, - CreatedAt: p.CreatedAt.UTC().Format(util.ISO8601), - } - if p.ExpiresAt != nil { - s := p.ExpiresAt.UTC().Format(util.ISO8601) - pr.ExpiresAt = &s + pr, err := p.ToStreamplacePinnedRecord() + if err != nil { + return placestream.ChatDefs_PinnedRecordView{}, err } rec := placestream.ChatDefs_PinnedRecordView{ LexiconTypeID: "place.stream.chat.defs#pinnedRecordView", @@ -81,9 +82,8 @@ func (m *DBModel) DeleteAllPinnedRecords(ctx context.Context, streamerDID string func (m *DBModel) GetActivePinnedRecord(ctx context.Context, streamerDID string) (*PinnedRecord, error) { var pin PinnedRecord - now := time.Now() err := m.DB.Preload("Repo"). - Where("repo_did = ? AND (expires_at IS NULL OR expires_at > ?)", streamerDID, now). + Where("repo_did = ?", streamerDID). Order("created_at DESC"). First(&pin).Error if errors.Is(err, gorm.ErrRecordNotFound) { @@ -92,5 +92,31 @@ func (m *DBModel) GetActivePinnedRecord(ctx context.Context, streamerDID string) if err != nil { return nil, err } + // The newest pin supersedes older pins, including after it expires. + if pin.LivestreamURI != "" { + latest, err := m.GetLatestLivestreamForRepo(streamerDID) + if err != nil { + return nil, err + } + if latest == nil { + return nil, nil + } + if latest.Livestream == nil { + return nil, errors.New("livestream record is nil") + } + var rec placestream.Livestream + if err := glex.DecodeCBOR(*latest.Livestream, &rec); err != nil { + return nil, fmt.Errorf("decoding pinned broadcast: %w", err) + } + broadcast := latest.URI + if rec.Broadcast != nil { + broadcast = *rec.Broadcast + } + if rec.EndedAt != nil || broadcast != pin.LivestreamURI { + return nil, nil + } + } else if pin.ExpiresAt != nil && !pin.ExpiresAt.After(time.Now()) { + return nil, nil + } return &pin, nil } diff --git a/pkg/model/pinned_record_test.go b/pkg/model/pinned_record_test.go new file mode 100644 index 000000000..6d224f572 --- /dev/null +++ b/pkg/model/pinned_record_test.go @@ -0,0 +1,110 @@ +package model + +import ( + "bytes" + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/placestream" +) + +func TestExpiredPinDoesNotRestoreOlderPin(t *testing.T) { + db := indexedTestDB(t) + ctx := context.Background() + now := time.Now().UTC() + expired := now.Add(-time.Minute) + for _, pin := range []PinnedRecord{ + {Uri: "at://did:plc:streamer/place.stream.chat.pinnedRecord/1", RepoDID: "did:plc:streamer", CID: "old", CreatedAt: now.Add(-time.Hour)}, + {Uri: "at://did:plc:streamer/place.stream.chat.pinnedRecord/2", RepoDID: "did:plc:streamer", CID: "new", CreatedAt: now, ExpiresAt: &expired}, + } { + require.NoError(t, db.CreatePinnedRecord(ctx, &pin)) + } + + pin, err := db.GetActivePinnedRecord(ctx, "did:plc:streamer") + require.NoError(t, err) + require.Nil(t, pin, "the newest pin supersedes older pins even after it expires") +} + +func TestBroadcastScopedPin(t *testing.T) { + const streamer = "did:plc:streamer" + root := "at://did:plc:streamer/place.stream.livestream/1" + const chapter = "at://did:plc:streamer/place.stream.livestream/2" + ended := "2026-01-01T01:00:00Z" + expired := time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC) + for _, tt := range []struct { + name string + latestURI string + broadcast *string + endedAt *string + expiresAt *time.Time + want bool + }{ + {name: "first chapter", latestURI: root, want: true}, + {name: "later chapter keeps the pin", latestURI: chapter, broadcast: &root, want: true}, + {name: "broadcast scope takes precedence over expiry", latestURI: chapter, broadcast: &root, expiresAt: &expired, want: true}, + {name: "last chapter ends the broadcast", latestURI: chapter, broadcast: &root, endedAt: &ended}, + {name: "new broadcast does not restore the pin", latestURI: chapter}, + {name: "no livestream"}, + } { + t.Run(tt.name, func(t *testing.T) { + db := indexedTestDB(t) + ctx := context.Background() + require.NoError(t, db.UpdateRepo(&Repo{DID: streamer, Handle: "streamer.test"})) + if tt.latestURI != "" { + rec := placestream.Livestream{ + LexiconTypeID: "place.stream.livestream", + Title: "chapter", + CreatedAt: "2026-01-01T00:00:00Z", + Broadcast: tt.broadcast, + EndedAt: tt.endedAt, + } + var buf bytes.Buffer + require.NoError(t, rec.MarshalCBOR(&buf)) + body := buf.Bytes() + require.NoError(t, db.CreateLivestream(ctx, &Livestream{ + URI: tt.latestURI, CID: "chapter", RepoDID: streamer, + CreatedAt: time.Now(), Livestream: &body, + })) + } + require.NoError(t, db.CreatePinnedRecord(ctx, &PinnedRecord{ + Uri: "at://did:plc:streamer/place.stream.chat.pinnedRecord/1", CID: "pin", + RepoDID: streamer, CreatedAt: time.Now(), LivestreamURI: root, ExpiresAt: tt.expiresAt, + })) + + pin, err := db.GetActivePinnedRecord(ctx, streamer) + require.NoError(t, err) + if !tt.want { + require.Nil(t, pin) + return + } + require.NotNil(t, pin) + record, err := pin.ToStreamplacePinnedRecord() + require.NoError(t, err) + require.NotNil(t, record.Livestream) + require.Equal(t, root, *record.Livestream) + view, err := pin.ToStreamplacePinnedRecordView() + require.NoError(t, err) + require.NotNil(t, view.Record.Livestream) + require.Equal(t, root, *view.Record.Livestream) + }) + } +} + +func TestBroadcastPinDoesNotIgnoreInvalidLivestreamData(t *testing.T) { + db := indexedTestDB(t) + ctx := context.Background() + body := []byte("invalid CBOR") + root := "at://did:plc:streamer/place.stream.livestream/1" + require.NoError(t, db.CreateLivestream(ctx, &Livestream{ + URI: root, CID: "broken", RepoDID: "did:plc:streamer", Livestream: &body, + })) + require.NoError(t, db.CreatePinnedRecord(ctx, &PinnedRecord{ + Uri: "at://did:plc:streamer/place.stream.chat.pinnedRecord/1", CID: "pin", + RepoDID: "did:plc:streamer", LivestreamURI: root, + })) + pin, err := db.GetActivePinnedRecord(ctx, "did:plc:streamer") + require.Error(t, err) + require.Nil(t, pin) +} diff --git a/pkg/placestream/chatpinnedrecord.go b/pkg/placestream/chatpinnedrecord.go index 7bbabfef8..b7bea4c01 100644 --- a/pkg/placestream/chatpinnedrecord.go +++ b/pkg/placestream/chatpinnedrecord.go @@ -21,8 +21,10 @@ type ChatPinnedRecord struct { LexiconTypeID string `json:"$type,omitempty"` // createdAt: When this pin was created. CreatedAt string `json:"createdAt"` - // expiresAt: Optional expiration time. If set, the pin is considered inactive after this time. + // expiresAt: Optional expiration time, used when livestream is not set. ExpiresAt *string `json:"expiresAt,omitempty"` + // livestream: AT-URI of the first chapter of the streamer's broadcast. The pin remains active across chapter changes until that broadcast ends. Takes precedence over expiresAt. Without either field, the pin remains until manually unpinned. + Livestream *string `json:"livestream,omitempty"` // pinnedBy: DID of the user who pinned the message. PinnedBy *string `json:"pinnedBy,omitempty"` // pinnedMessage: AT-URI of the pinned chat message. diff --git a/pkg/placestream/livestream.go b/pkg/placestream/livestream.go index 98d4027c5..325d9324d 100644 --- a/pkg/placestream/livestream.go +++ b/pkg/placestream/livestream.go @@ -26,6 +26,8 @@ type Livestream struct { Activity *Livestream_Activity `json:"activity,omitempty"` // agent: The source of the livestream, if available, in a User Agent format: ` / ` e.g. Streamplace/0.7.5 iOS Agent *string `json:"agent,omitempty"` + // broadcast: AT-URI of the first chapter in this continuous broadcast, owned by the same repo. Omitted on the first chapter, whose own URI identifies the broadcast. Later chapters preserve this reference; a separate broadcast starts with a new identity. + Broadcast *string `json:"broadcast,omitempty"` // canonicalUrl: The primary URL where this livestream can be viewed, if available. CanonicalUrl *string `json:"canonicalUrl,omitempty"` // createdAt: Client-declared timestamp when this livestream started. diff --git a/pkg/placestream/moderationcreatepin.go b/pkg/placestream/moderationcreatepin.go index 21dc95422..fc867e432 100644 --- a/pkg/placestream/moderationcreatepin.go +++ b/pkg/placestream/moderationcreatepin.go @@ -14,8 +14,10 @@ import ( type ModerationCreatePin_Input struct { LexiconTypeID string `json:"$type,omitempty"` - // expiresAt: Optional expiration time for this pin. + // expiresAt: Optional expiration time for this pin, used when livestream is not set. ExpiresAt *string `json:"expiresAt,omitempty"` + // livestream: AT-URI of the first chapter of the streamer's current broadcast. The pin remains active across chapter changes until that broadcast ends. Without either livestream or expiresAt, the pin remains until manually unpinned. + Livestream *string `json:"livestream,omitempty"` // messageUri: The AT-URI of the chat message to pin. MessageUri string `json:"messageUri"` // streamer: The DID of the streamer. diff --git a/pkg/spxrpc/pinned_record_test.go b/pkg/spxrpc/pinned_record_test.go new file mode 100644 index 000000000..51cf7080e --- /dev/null +++ b/pkg/spxrpc/pinned_record_test.go @@ -0,0 +1,33 @@ +package spxrpc + +import ( + "context" + "net/http" + "testing" + + "github.com/labstack/echo/v4" + "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/placestream" +) + +func TestCreatePinRejectsInvalidBroadcast(t *testing.T) { + for _, uri := range []string{ + "not-an-at-uri", + "at://did:plc:other/place.stream.livestream/1", + "at://did:plc:streamer/place.stream.chat.message/1", + "at://did:plc:streamer/place.stream.livestream", + } { + t.Run(uri, func(t *testing.T) { + s := &Server{} + _, err := s.handlePlaceStreamModerationCreatePin(context.Background(), &placestream.ModerationCreatePin_Input{ + Streamer: "did:plc:streamer", + MessageUri: "at://did:plc:viewer/place.stream.chat.message/1", + Livestream: &uri, + }) + var httpErr *echo.HTTPError + require.ErrorAs(t, err, &httpErr) + require.Equal(t, http.StatusBadRequest, httpErr.Code) + require.Equal(t, "livestream must reference the streamer's broadcast", httpErr.Message) + }) + } +} diff --git a/pkg/spxrpc/place_stream_moderation.go b/pkg/spxrpc/place_stream_moderation.go index 06b07bf4f..f2aa622e2 100644 --- a/pkg/spxrpc/place_stream_moderation.go +++ b/pkg/spxrpc/place_stream_moderation.go @@ -264,6 +264,9 @@ func (s *Server) handlePlaceStreamModerationUpdateLivestream(ctx context.Context return nil, echo.NewHTTPError(http.StatusInternalServerError, "failed to decode livestream record") } livestream := *existing + if livestream.Broadcast == nil { + livestream.Broadcast = &input.LivestreamUri + } // Create new record (don't edit existing - old records serve as "chapter markers") // Copy fields from existing record and update title @@ -363,12 +366,38 @@ func (s *Server) handlePlaceStreamModerationCreatePin(ctx context.Context, input if err := validateATURI(input.MessageUri); err != nil { return nil, echo.NewHTTPError(http.StatusBadRequest, fmt.Sprintf("invalid messageUri: %v", err)) } + if input.Livestream != nil { + uri, err := syntax.ParseATURI(*input.Livestream) + if err != nil || uri.Authority().String() != input.Streamer || uri.Collection().String() != constants.PLACE_STREAM_LIVESTREAM || uri.RecordKey().String() == "" { + return nil, echo.NewHTTPError(http.StatusBadRequest, "livestream must reference the streamer's broadcast") + } + } // Get delegated moderation context (validates OAuth, permission, and returns client) modCtx, err := s.GetDelegatedModerationContext(ctx, input.Streamer, "createPin") if err != nil { return nil, err } + if input.Livestream != nil { + latest, err := s.model.GetLatestLivestreamForRepo(input.Streamer) + if err != nil { + return nil, echo.NewHTTPError(http.StatusInternalServerError, "failed to get current broadcast").SetInternal(err) + } + if latest == nil || latest.Livestream == nil { + return nil, echo.NewHTTPError(http.StatusBadRequest, "no current broadcast") + } + var rec placestream.Livestream + if err := glex.DecodeCBOR(*latest.Livestream, &rec); err != nil { + return nil, echo.NewHTTPError(http.StatusInternalServerError, "failed to decode current broadcast").SetInternal(err) + } + broadcast := latest.URI + if rec.Broadcast != nil { + broadcast = *rec.Broadcast + } + if rec.EndedAt != nil || broadcast != *input.Livestream { + return nil, echo.NewHTTPError(http.StatusBadRequest, "livestream is not the current active broadcast") + } + } // Create the pinned record (old pins persist as history) pinnedRecord := placestream.ChatPinnedRecord{ @@ -376,6 +405,7 @@ func (s *Server) handlePlaceStreamModerationCreatePin(ctx context.Context, input PinnedMessage: input.MessageUri, CreatedAt: time.Now().UTC().Format(time.RFC3339), ExpiresAt: input.ExpiresAt, + Livestream: input.Livestream, } createInput := comatproto.RepoCreateRecord_Input{ -- knotmirror 0.0.1