From a7dbdb6be9bac1326ed897023a307dc08ee48ddd Mon Sep 17 00:00:00 2001 From: "Natalie B." <22222885+espeon@users.noreply.github.com> Date: Fri, 20 Mar 2026 23:58:48 -0500 Subject: [PATCH] Add pinned chat message feature Add UI pin controls (preset durations), a pinned comment notification banner, websocket/server support to broadcast the active pin, and DB/ lexicon/schema changes to record the pinner and TTL --- docs/pinned-comment-plan.md | 460 ------------------ .../src/components/chat/mod-view.tsx | 110 +++-- .../stream-notification/pin-notification.tsx | 135 +++++ js/components/src/lib/stream-notifications.ts | 28 ++ .../src/livestream-provider/index.tsx | 38 +- .../chat/place-stream-chat-defs.md | 18 +- .../chat/place-stream-chat-pinnedrecord.md | 6 + .../lex-reference/place-stream-livestream.md | 9 +- lexicons/place/stream/chat/defs.json | 2 +- lexicons/place/stream/chat/pinnedRecord.json | 5 + lexicons/place/stream/livestream.json | 3 +- pkg/api/websocket.go | 53 ++ pkg/atproto/sync.go | 51 +- pkg/model/pinned_record.go | 22 + pkg/streamplace/cbor_gen.go | 59 ++- pkg/streamplace/chatdefs.go | 13 +- pkg/streamplace/chatpinnedRecord.go | 2 + pkg/streamplace/streamlivestream.go | 8 + 18 files changed, 484 insertions(+), 538 deletions(-) delete mode 100644 docs/pinned-comment-plan.md create mode 100644 js/components/src/components/stream-notification/pin-notification.tsx diff --git a/docs/pinned-comment-plan.md b/docs/pinned-comment-plan.md deleted file mode 100644 index a6babef4..00000000 --- a/docs/pinned-comment-plan.md +++ /dev/null @@ -1,460 +0,0 @@ -# Pinned Comment Feature Plan - -## Overview - -Allow the streamer or a delegated moderator to pin an existing chat message. The pinned comment appears prominently in the chat UI and auto-expires when a TTL is reached or the stream ends. - -## Design Decisions - -- **Pinnable content**: Existing chat messages only (referenced by AT-URI) -- **Who can pin**: Streamer + mods with new `message.pin` permission -- **Expiration**: Optional TTL (`expiresAt` datetime) + auto-clear at stream end (client-side) -- **Storage pattern**: Record in streamer's AT Protocol repo, modeled after `place.stream.chat.gate` -- **Single active pin**: Only one pinned comment at a time. Creating a new pin replaces the previous one. -- **Viewer dismiss**: Any viewer can hide the pin from their own view (local state only). The pin remains for everyone else. - ---- - -## 1. New Lexicon: `place.stream.chat.pinnedRecord` - -**File**: `lexicons/place/stream/chat/pinnedRecord.json` - -```json -{ - "lexicon": 1, - "id": "place.stream.chat.pinnedRecord", - "defs": { - "main": { - "type": "record", - "key": "tid", - "description": "Record pinning a chat message for prominent display.", - "record": { - "type": "object", - "required": ["pinnedMessage", "createdAt"], - "properties": { - "pinnedMessage": { - "type": "string", - "format": "at-uri", - "description": "AT-URI of the pinned chat message." - }, - "createdAt": { - "type": "string", - "format": "datetime", - "description": "When this pin was created." - }, - "expiresAt": { - "type": "string", - "format": "datetime", - "description": "Optional expiration time. If set, the pin is considered inactive after this time." - } - } - } - } - } -} -``` - ---- - -## 2. Pinned Record View + Defs - -### 2a. Add `pinnedRecordView` to chat defs - -**File**: `lexicons/place/stream/chat/defs.json` (add new definition) - -```json -"pinnedRecordView": { - "type": "object", - "description": "View of a pinned chat record with hydrated message data.", - "required": ["uri", "cid", "record", "indexedAt"], - "properties": { - "uri": { "type": "string", "format": "at-uri" }, - "cid": { "type": "string", "format": "cid" }, - "record": { "type": "ref", "ref": "place.stream.chat.pinnedRecord" }, - "indexedAt": { "type": "string", "format": "datetime" }, - "pinnedBy": { "type": "ref", "ref": "app.bsky.actor.defs#profileViewBasic" }, - "message": { "type": "ref", "ref": "place.stream.chat.defs#messageView" } - } -} -``` - -This gives us a hydrated view that includes the full message data, so the frontend doesn't need to look it up from the chat index. - -### 2b. Go struct for bus delivery - -The bus will deliver a `PinnedRecordView` (not just the raw record) so the frontend can render the message text directly. Similar to how `ChatGate` records are published raw but the frontend processes them. - ---- - -## 3. New Permission: `message.pin` - -### 3a. Add to lexicon enum - -**File**: `lexicons/place/stream/moderation/permission.json` - -Add `"message.pin"` to the permissions enum: - -```json -"enum": ["ban", "hide", "livestream.manage", "message.pin"] -``` - -### 3b. Register in Go permission system - -**File**: `pkg/moderation/permissions.go` - -```go -const PermissionMessagePin = "message.pin" - -var ActionPermissions = map[string]string{ - // ... existing entries ... - "createPin": PermissionMessagePin, - "deletePin": PermissionMessagePin, -} -``` - -### 3c. Update frontend permissions - -**File**: `js/components/src/streamplace-store/moderation.tsx` - -Add `canPin: boolean` to `ModerationPermissions` interface. Derive from permissions array: - -```ts -canPin: isOwner || permissions.includes("message.pin"), -``` - ---- - -## 4. New RPC Procedures - -### 4a. Lexicon: `place.stream.moderation.createPin` - -**File**: `lexicons/place/stream/moderation/createPin.json` - -```json -{ - "lexicon": 1, - "id": "place.stream.moderation.createPin", - "defs": { - "main": { - "type": "procedure", - "description": "Pin a chat message on behalf of a streamer. Requires 'message.pin' permission. Creates a place.stream.chat.pinnedRecord in the streamer's repo, replacing any existing pin.", - "input": { - "encoding": "application/json", - "schema": { - "type": "object", - "required": ["streamer", "messageUri"], - "properties": { - "streamer": { "type": "string", "format": "did" }, - "messageUri": { "type": "string", "format": "at-uri" }, - "expiresAt": { "type": "string", "format": "datetime" } - } - } - }, - "output": { - "encoding": "application/json", - "schema": { - "type": "object", - "required": ["uri", "cid"], - "properties": { - "uri": { "type": "string", "format": "at-uri" }, - "cid": { "type": "string", "format": "cid" } - } - } - }, - "errors": [ - { "name": "Unauthorized" }, - { "name": "Forbidden" }, - { "name": "SessionNotFound" } - ] - } - } -} -``` - -### 4b. Lexicon: `place.stream.moderation.deletePin` - -**File**: `lexicons/place/stream/moderation/deletePin.json` - -Same pattern as `deleteGate`: - -- Input: `streamer` (DID), `pinUri` (at-uri) -- Output: empty -- Errors: Unauthorized, Forbidden, SessionNotFound - -### 4c. Go handlers - -**File**: `pkg/spxrpc/place_stream_moderation.go` - -Add `handlePlaceStreamModerationCreatePin`: - -1. Validate input (DID, AT-URI) -2. `GetDelegatedModerationContext(ctx, input.Streamer, "createPin")` -3. Before creating: list existing `place.stream.chat.pinnedRecord` records in streamer's repo, delete any existing ones (single-pin semantics) -4. Build `streamplace.ChatPinnedRecord` struct -5. Create via `com.atproto.repo.createRecord` on streamer's repo -6. Audit log -7. Return URI + CID - -Add `handlePlaceStreamModerationDeletePin`: - -1. Validate input -2. `GetDelegatedModerationContext(ctx, input.Streamer, "deletePin")` -3. Extract rkey, delete record via `com.atproto.repo.deleteRecord` -4. Audit log - -### 4d. Register handlers - -In the XRPC server setup where `createGate`/`deleteGate` are registered, add `createPin` and `deletePin`. - ---- - -## 5. Backend: Model + DB - -**File**: `pkg/model/pinned_record.go` - -```go -type PinnedRecord struct { - RKey string `gorm:"primaryKey;column:rkey"` - CID string `gorm:"column:cid"` - RepoDID string `gorm:"column:repo_did"` - Repo *Repo `gorm:"foreignKey:DID;references:RepoDID"` - PinnedMessage string `gorm:"column:pinned_message"` - ExpiresAt *time.Time `gorm:"column:expires_at"` - CreatedAt time.Time `gorm:"column:created_at"` -} -``` - -Methods: - -- `CreatePinnedRecord(ctx, pin)` - insert -- `GetPinnedRecord(ctx, rkey)` - single lookup -- `DeletePinnedRecord(ctx, rkey)` - delete by rkey -- `GetActivePinnedRecord(ctx, streamerDID)` - returns the most recent non-expired pin for a streamer -- `DeleteAllPinnedRecords(ctx, streamerDID)` - bulk delete (called before creating new pin) - -Add `PinnedRecord{}` to the AutoMigrate list in `pkg/model/model.go`. - ---- - -## 6. Constants + Firehose - -### 6a. Constants - -**File**: `pkg/constants/constants.go` - -Add: `var PLACE_STREAM_CHAT_PINNED_RECORD = "place.stream.chat.pinnedRecord"` - -### 6b. Sync (create/update) - -**File**: `pkg/atproto/sync.go` - `handleCreateUpdate` - -Add case `*streamplace.ChatPinnedRecord`: - -- Sync bluesky repo -- Delete existing pinned records for this streamer (single-pin enforcement at DB level) -- Create new `PinnedRecord` model entry -- Build a hydrated view (resolve the pinned message, include author info) and publish to bus on streamer's channel - -### 6c. Firehose (delete) - -**File**: `pkg/atproto/firehose.go` - `EvtKindDeleteRecord` - -Add handling for `constants.PLACE_STREAM_CHAT_PINNED_RECORD`: - -- `DeletePinnedRecord(ctx, rkey)` -- Publish deletion marker to bus - ---- - -## 7. Bus / WebSocket Delivery - -### Create event - -Publish a `PinnedRecordView`-like object to the streamer's bus channel: - -```json -{ - "$type": "place.stream.chat.defs#pinnedRecordView", - "uri": "at://...", - "cid": "...", - "record": { - "pinnedMessage": "at://...", - "createdAt": "...", - "expiresAt": "..." - }, - "indexedAt": "...", - "message": { - /* hydrated messageView with author, text, facets, etc */ - } -} -``` - -### Delete event - -Publish a deletion marker: - -```json -{ - "$type": "place.stream.chat.pinnedRecord", - "deleted": true, - "rkey": "..." -} -``` - ---- - -## 8. Frontend State - -### LivestreamState additions - -**File**: `js/components/src/livestream-store/livestream-state.tsx` - -```ts -pinnedComment: PinnedRecordView | null; -``` - -Where `PinnedRecordView` is a type with the hydrated message + pin metadata. - -### Store init - -**File**: `js/components/src/livestream-store/livestream-store.tsx` - -Add `pinnedComment: null` to initial state. - -### Chat hooks - -**File**: `js/components/src/livestream-store/chat.tsx` - -Add: - -- `usePinnedComment()` - selector: `state.pinnedComment` -- `usePinChatMessage()` - hook to pin a message: - - If streamer: direct `com.atproto.repo.createRecord` for `place.stream.chat.pinnedRecord` - - If mod: call `place.stream.moderation.createPin` XRPC - - On success, the bus event will update state automatically -- `useUnpinChatMessage()` - hook to unpin: - - If streamer: direct `com.atproto.repo.deleteRecord` - - If mod: call `place.stream.moderation.deletePin` XRPC - -### WebSocket consumer - -**File**: `js/components/src/livestream-store/websocket-consumer.tsx` - -Add handler for `place.stream.chat.defs#pinnedRecordView`: - -- Set `state.pinnedComment` to the hydrated view - -Add handler for `place.stream.chat.pinnedRecord` with `deleted: true`: - -- Set `state.pinnedComment` to `null` - -### Expiration - -In the Chat component, `useEffect` checks `pinnedComment.record.expiresAt`. If current time passes it, clear `pinnedComment` locally. Also clear when `livestream.record.endedAt` is set (stream ended). - ---- - -## 9. Frontend UI - -### Pinned comment as persistent stream notification - -The pinned comment uses the existing `StreamNotificationProvider` + `streamNotificationManager` system. - -**File**: `js/components/src/components/stream-notification/pinned-comment-notification.tsx` (new) - -A custom render function passed to `streamNotificationManager.show()` with `duration: 0` (manual dismiss only), making it a persistent notification that sits at the top of the notification stack. Other temporary notifications (teleport, etc.) will appear above/below it and dismiss independently. - -The render function receives `(isExiting, onDismiss, startTime)` and renders: - -- Pin icon + author name (with chatProfile color) + message text (with facets rendered) -- Two dismiss actions: - - **Unpin** (X icon, only visible if `canPin`): calls `useUnpinChatMessage()`, removes the pin globally, then calls `onDismiss("user")` to dismiss the notification - - **Hide** (eye-off icon, visible to all viewers): calls `onDismiss("user")` to dismiss the notification locally only. Does NOT call the server - the pin remains for other viewers -- Does NOT link to or scroll to the original message (chat history is limited) -- Auto-dismisses when `expiresAt` is reached (a `useEffect` watching the pinned comment state calls `streamNotificationManager.requestDismiss("pinned-comment", "auto")`) -- Auto-dismisses when stream ends - -**Integration point**: The `TeleportWatcher` / livestream provider already manages notification lifecycle. The pinned comment notification is managed similarly - when `state.pinnedComment` changes in the Zustand store, a hook or effect calls `streamNotificationManager.show()` to create/update the notification, or `streamNotificationManager.hide()` to remove it. - -The notification ID should be fixed (e.g., `"pinned-comment"`) so that replacing a pin updates the same notification rather than creating a new one. - -### Pin action in mod menu - -**File**: `js/components/src/components/chat/mod-view.tsx` - -Add "Pin this message" in the moderation actions group (visible when `canPin` is true): - -```tsx -{ - modPermissions.canPin && ( - { - /* pin message */ - }} - > - Pin this message - - ); -} -``` - ---- - -## 10. Moderator Panel - -**File**: `js/components/src/components/dashboard/moderator-panel.tsx` - -Add `"message.pin"` to the list of assignable permissions when adding/editing moderators. - ---- - -## 11. Code Generation - -After creating lexicon JSON files, run `lexgen` to generate Go types: - -- `pkg/streamplace/chatpinnedrecord.go` (auto-generated) -- Update generated TypeScript types in the `streamplace` package - ---- - -## File Change Summary - -| File | Action | -| ---------------------------------------------------------------------------------- | ------------------------------------------------------- | -| `lexicons/place/stream/chat/pinnedRecord.json` | **New** | -| `lexicons/place/stream/chat/defs.json` | **Edit** - add `pinnedRecordView` | -| `lexicons/place/stream/moderation/createPin.json` | **New** | -| `lexicons/place/stream/moderation/deletePin.json` | **New** | -| `lexicons/place/stream/moderation/permission.json` | **Edit** - add `message.pin` | -| `pkg/constants/constants.go` | **Edit** | -| `pkg/moderation/permissions.go` | **Edit** | -| `pkg/model/pinned_record.go` | **New** | -| `pkg/model/model.go` | **Edit** | -| `pkg/atproto/sync.go` | **Edit** | -| `pkg/atproto/firehose.go` | **Edit** | -| `pkg/spxrpc/place_stream_moderation.go` | **Edit** | -| `js/components/src/streamplace-store/moderation.tsx` | **Edit** | -| `js/components/src/streamplace-store/block.tsx` | **Edit** - add pin/unpin hooks | -| `js/components/src/livestream-store/livestream-state.tsx` | **Edit** | -| `js/components/src/livestream-store/livestream-store.tsx` | **Edit** | -| `js/components/src/livestream-store/chat.tsx` | **Edit** | -| `js/components/src/livestream-store/websocket-consumer.tsx` | **Edit** | -| `js/components/src/components/stream-notification/pinned-comment-notification.tsx` | **New** - notification render function | -| `js/components/src/livestream-provider/index.tsx` | **Edit** - manage pinned comment notification lifecycle | -| `js/components/src/components/chat/mod-view.tsx` | **Edit** - add pin action | -| `js/components/src/components/dashboard/moderator-panel.tsx` | **Edit** | - ---- - -## Implementation Order - -1. Lexicons (pinnedRecord, createPin, deletePin, permission, defs) -2. Code generation (lexgen) -3. Backend: constants, permissions, model, DB migration -4. Backend: sync/firehose integration -5. Backend: XRPC handlers + registration -6. Frontend: state types, store init, hooks -7. Frontend: WebSocket consumer handlers -8. Frontend: pinned-comment notification render function + lifecycle management -9. Frontend: mod-view pin action -10. Frontend: moderator panel permission diff --git a/js/components/src/components/chat/mod-view.tsx b/js/components/src/components/chat/mod-view.tsx index afdc1976..dc52daa6 100644 --- a/js/components/src/components/chat/mod-view.tsx +++ b/js/components/src/components/chat/mod-view.tsx @@ -237,53 +237,69 @@ function ModViewContent({ )} {modPermissions.canPin && message.author.did !== streamerDID && ( - - - Pin this message - - - { - 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 }, - ); - }); - }} - > - Pin indefinitely - - { - if (!streamerDID) return; - const expiresAt = new Date(); - expiresAt.setHours(expiresAt.getHours() + 1); // Set expiration to 1 hour from now - pinChatMessage(message.uri, streamerDID, expiresAt) - .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 }, - ); - }); - }} - > - Pin for 1 hour - - - + + + + Pin this message + + + + { + 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 }, + ); + }); + }} + > + Until stream end + + {[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 }, + ); + }); + }} + > + + {minutes < 60 ? `${minutes} min` : "1 hour"} + + + ))} + + + + )} {modPermissions.canBan && agent?.did && diff --git a/js/components/src/components/stream-notification/pin-notification.tsx b/js/components/src/components/stream-notification/pin-notification.tsx new file mode 100644 index 00000000..a56766b3 --- /dev/null +++ b/js/components/src/components/stream-notification/pin-notification.tsx @@ -0,0 +1,135 @@ +import { EyeOff, Pin, X } from "lucide-react-native"; +import { useEffect, useState } from "react"; +import { Linking, Pressable, View } from "react-native"; +import { PinnedRecordViewHydrated, PlaceStreamChatProfile } from "streamplace"; +import { + Text, + useCanModerate, + useLivestreamStore, + useTheme, + zero, +} from "../../"; +import { RichtextSegment, segmentize } from "../../lib/facet"; +import { formatHandleWithAt } from "../../utils/format-handle"; + +const getRgbColor = (color?: PlaceStreamChatProfile.Color) => + color ? `rgb(${color.red}, ${color.green}, ${color.blue})` : undefined; + +function renderSegment(segment: RichtextSegment, index: number) { + if (segment.features && segment.features.length > 0) { + const ftr = segment.features[0]; + if (ftr.$type === "app.bsky.richtext.facet#link") { + return ( + Linking.openURL((ftr as any).uri || "")} + > + {segment.text} + + ); + } + if (ftr.$type === "app.bsky.richtext.facet#mention") { + return ( + + {segment.text} + + ); + } + } + return {segment.text}; +} + +export function PinnedCommentNotification({ + pinnedComment, + onDismiss, + onUnpin, +}: { + pinnedComment: PinnedRecordViewHydrated; + onDismiss: () => void; + onUnpin: () => void; +}) { + const z = useTheme(); + const message = pinnedComment.message; + const pinnedByColor = (pinnedComment.pinnedBy as any)?.color || "#bebebe"; + const record = pinnedComment.record; + + const currentStreamer = useLivestreamStore((state) => state.profile?.did); + + console.log("checking if we can mod", currentStreamer); + + const canActuallyPin = useCanModerate(currentStreamer)?.canPin; + + const messageRecord = message?.record as any; + + const [expiresAt] = useState( + record.expiresAt ? new Date(record.expiresAt) : null, + ); + + useEffect(() => { + if (!expiresAt) return; + const remaining = expiresAt.getTime() - Date.now(); + if (remaining <= 0) { + onDismiss(); + return; + } + const timeout = setTimeout(onDismiss, remaining); + return () => clearTimeout(timeout); + }, [expiresAt, onDismiss]); + + const authorName = message ? formatHandleWithAt(message.author) : "unknown"; + const authorColor = getRgbColor((message as any)?.chatProfile?.color); + + const segments = messageRecord + ? segmentize(messageRecord.text, messageRecord.facets) + : []; + + return ( + + + + + + + + + {authorName} + + {segments.map((seg, i) => renderSegment(seg, i))} + + + + {canActuallyPin && ( + + + + )} + + + + + + + ); +} diff --git a/js/components/src/lib/stream-notifications.ts b/js/components/src/lib/stream-notifications.ts index e05c4d56..4bbf40ad 100644 --- a/js/components/src/lib/stream-notifications.ts +++ b/js/components/src/lib/stream-notifications.ts @@ -1,8 +1,36 @@ import React from "react"; +import { PinnedRecordViewHydrated } from "streamplace"; import { streamNotification } from "../components/stream-notification"; +import { PinnedCommentNotification } from "../components/stream-notification/pin-notification"; import { TeleportNotification } from "../components/stream-notification/teleport-notification"; export const StreamNotifications = { + pinnedComment: (params: { + pinnedComment: PinnedRecordViewHydrated; + onDismiss?: (reason?: "user" | "auto") => void; + onUnpin?: () => void; + }) => { + streamNotification.show({ + id: "pinned-comment", + render: (isExiting, onDismiss) => { + return React.createElement(PinnedCommentNotification, { + pinnedComment: params.pinnedComment, + onDismiss: () => onDismiss("user"), + onUnpin: () => { + params.onUnpin?.(); + onDismiss("user"); + }, + }); + }, + duration: 0, // manually dismissed or auto-dismissed by TTL + onDismiss: params.onDismiss, + }); + }, + + pinnedCommentDismiss: () => { + streamNotification.hide("pinned-comment"); + }, + teleport: (params: { targetHandle: string; targetDID: string; diff --git a/js/components/src/livestream-provider/index.tsx b/js/components/src/livestream-provider/index.tsx index 1a752cab..e40e2249 100644 --- a/js/components/src/livestream-provider/index.tsx +++ b/js/components/src/livestream-provider/index.tsx @@ -6,6 +6,8 @@ import { LivestreamContext, makeLivestreamStore, useLivestreamStore, + usePinnedComment, + useUnpinChatMessage, } from "../livestream-store"; import { useDID, usePDSAgent } from "../streamplace-store"; import { useLivestreamWebsocket } from "./websocket"; @@ -134,6 +136,39 @@ export function TeleportWatcher({ return <>; } +export function PinnedCommentWatcher() { + const pinnedComment = usePinnedComment(); + const streamerDID = useLivestreamStore((state) => state.profile?.did); + const unpinChatMessage = useUnpinChatMessage(); + const prevPinnedRef = useRef(null); + + // Show/hide notification when pinned comment changes + useEffect(() => { + const currentUri = pinnedComment?.uri ?? null; + if (currentUri === prevPinnedRef.current) return; + prevPinnedRef.current = currentUri; + + if (pinnedComment) { + StreamNotifications.pinnedComment({ + pinnedComment, + onDismiss: () => { + // local dismiss + }, + onUnpin: () => { + if (!streamerDID) return; + unpinChatMessage(pinnedComment.uri, streamerDID).catch((e) => { + console.error("Failed to unpin:", e); + }); + }, + }); + } else { + StreamNotifications.pinnedCommentDismiss(); + } + }, [pinnedComment, streamerDID, unpinChatMessage]); + + return <>; +} + export function LivestreamPoller({ children, src, @@ -143,12 +178,11 @@ export function LivestreamPoller({ src: string; onTeleport?: (targetHandle: string, targetDID: string) => void; }) { - // Websocket watcher is a sibling instead of a parent to avoid - // re-rendering when the websocket does stuff return ( <> + {children} ); diff --git a/js/docs/src/content/docs/lex-reference/chat/place-stream-chat-defs.md b/js/docs/src/content/docs/lex-reference/chat/place-stream-chat-defs.md index bb73679b..1d4a71ca 100644 --- a/js/docs/src/content/docs/lex-reference/chat/place-stream-chat-defs.md +++ b/js/docs/src/content/docs/lex-reference/chat/place-stream-chat-defs.md @@ -39,14 +39,14 @@ View of a pinned chat record with hydrated message data. **Properties:** -| Name | Type | Req'd | Description | Constraints | -| ----------- | ------------------------------------------------------------------------------------------------------------------------------------------------ | ----- | ----------- | ------------------ | -| `uri` | `string` | ✅ | | Format: `at-uri` | -| `cid` | `string` | ✅ | | Format: `cid` | -| `record` | [`place.stream.chat.pinnedRecord`](/lex-reference/place-stream-chat-pinnedrecord) | ✅ | | | -| `indexedAt` | `string` | ✅ | | Format: `datetime` | -| `pinnedBy` | [`app.bsky.actor.defs#profileViewBasic`](https://github.com/bluesky-social/atproto/tree/main/lexicons/app/bsky/actor/defs.json#profileViewBasic) | ❌ | | | -| `message` | [`#messageView`](#messageview) | ❌ | | | +| Name | Type | Req'd | Description | Constraints | +| ----------- | --------------------------------------------------------------------------------- | ----- | ----------- | ------------------ | +| `uri` | `string` | ✅ | | Format: `at-uri` | +| `cid` | `string` | ✅ | | Format: `cid` | +| `record` | [`place.stream.chat.pinnedRecord`](/lex-reference/place-stream-chat-pinnedrecord) | ✅ | | | +| `indexedAt` | `string` | ✅ | | Format: `datetime` | +| `pinnedBy` | [`place.stream.chat.profile`](/lex-reference/place-stream-chat-profile) | ❌ | | | +| `message` | [`#messageView`](#messageview) | ❌ | | | --- @@ -126,7 +126,7 @@ View of a pinned chat record with hydrated message data. }, "pinnedBy": { "type": "ref", - "ref": "app.bsky.actor.defs#profileViewBasic" + "ref": "place.stream.chat.profile" }, "message": { "type": "ref", 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 8935e957..fc88032a 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 @@ -22,6 +22,7 @@ Record pinning a chat message for prominent display. | 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` | @@ -47,6 +48,11 @@ Record pinning a chat message for prominent display. "format": "at-uri", "description": "AT-URI of the pinned chat message." }, + "pinnedBy": { + "type": "string", + "format": "did", + "description": "DID of the user who pinned the message." + }, "createdAt": { "type": "string", "format": "datetime", 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 2825bf4c..db8ee166 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 @@ -123,9 +123,9 @@ Record announcing a livestream is happening **Properties:** -| 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) | ✅ | | | +| 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)
  [`place.stream.chat.defs#pinnedRecordView`](/lex-reference/place-stream-chat-defs#pinnedrecordview) | ✅ | | | --- @@ -309,7 +309,8 @@ Record announcing a livestream is happening "place.stream.defs#blockView", "place.stream.defs#renditions", "place.stream.defs#rendition", - "place.stream.chat.defs#messageView" + "place.stream.chat.defs#messageView", + "place.stream.chat.defs#pinnedRecordView" ] } } diff --git a/lexicons/place/stream/chat/defs.json b/lexicons/place/stream/chat/defs.json index aa1dcd75..5f9c495d 100644 --- a/lexicons/place/stream/chat/defs.json +++ b/lexicons/place/stream/chat/defs.json @@ -48,7 +48,7 @@ "indexedAt": { "type": "string", "format": "datetime" }, "pinnedBy": { "type": "ref", - "ref": "app.bsky.actor.defs#profileViewBasic" + "ref": "place.stream.chat.profile" }, "message": { "type": "ref", diff --git a/lexicons/place/stream/chat/pinnedRecord.json b/lexicons/place/stream/chat/pinnedRecord.json index f43ac525..32b5b33e 100644 --- a/lexicons/place/stream/chat/pinnedRecord.json +++ b/lexicons/place/stream/chat/pinnedRecord.json @@ -15,6 +15,11 @@ "format": "at-uri", "description": "AT-URI of the pinned chat message." }, + "pinnedBy": { + "type": "string", + "format": "did", + "description": "DID of the user who pinned the message." + }, "createdAt": { "type": "string", "format": "datetime", diff --git a/lexicons/place/stream/livestream.json b/lexicons/place/stream/livestream.json index e02f79a7..19b73cf3 100644 --- a/lexicons/place/stream/livestream.json +++ b/lexicons/place/stream/livestream.json @@ -162,7 +162,8 @@ "place.stream.defs#blockView", "place.stream.defs#renditions", "place.stream.defs#rendition", - "place.stream.chat.defs#messageView" + "place.stream.chat.defs#messageView", + "place.stream.chat.defs#pinnedRecordView" ] } } diff --git a/pkg/api/websocket.go b/pkg/api/websocket.go index 418a406c..67ad07f6 100644 --- a/pkg/api/websocket.go +++ b/pkg/api/websocket.go @@ -259,6 +259,59 @@ func (a *StreamplaceAPI) HandleWebsocket(ctx context.Context) httprouter.Handle } }() + // get the latest active pinned message for the repo + go func() { + pin, err := a.Model.GetActivePinnedRecord(ctx, repoDID) + if err != nil { + log.Error(ctx, "could not get pinned record", "error", err) + return + } + if pin != nil { + prv, err := pin.ToStreamplacePinnedRecordView() + if err != nil { + log.Error(ctx, "could not convert pinned record to streamplace view", "error", err) + return + } + // look up the original message, pinner + msg, err := a.Model.GetChatMessage(prv.Record.PinnedMessage) + if err != nil { + log.Error(ctx, "failed to get pinned message", err) + return + } + // if the message was deleted, treat as no pinned message + if msg != nil && msg.DeletedAt != nil { + log.Log(ctx, "pinned message was deleted, skipping", "uri", msg.URI) + return + } + // if no pinned by, use the repo owner as the pinner + if prv.Record.PinnedBy == nil { + prv.Record.PinnedBy = &repoDID + } + profile, err := a.Model.GetChatProfile(ctx, *prv.Record.PinnedBy) + if err != nil { + log.Error(ctx, "failed to get chat profile", err) + return + } + if msg != nil { + msgView, err := msg.ToStreamplaceMessageView() + if err != nil { + log.Error(ctx, "failed to convert chat message: %w", err) + return + } + prv.Message = msgView + } + if profile != nil { + profileView, err := profile.ToStreamplaceChatProfile() + if err != nil { + log.Error(ctx, "failed to convert chat profile: %w", err) + return + } + prv.PinnedBy = profileView + } + initialBurst <- prv + } + }() + go func() { teleports, err := a.Model.GetActiveTeleportsToRepo(repoDID) if err != nil { diff --git a/pkg/atproto/sync.go b/pkg/atproto/sync.go index a739a695..53f93032 100644 --- a/pkg/atproto/sync.go +++ b/pkg/atproto/sync.go @@ -226,11 +226,10 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD return nil } log.Debug(ctx, "creating pinned record", "userDID", userDID, "pinnedMessage", rec.PinnedMessage) - // Delete existing pinned records for this streamer (single-pin semantics) - err = atsync.Model.DeleteAllPinnedRecords(ctx, userDID) - if err != nil { - log.Error(ctx, "failed to delete existing pinned records", "err", err) - } + // err = atsync.Model.DeleteAllPinnedRecords(ctx, userDID) + // if err != nil { + // log.Error(ctx, "failed to delete existing pinned records", "err", err) + // } // Parse optional expiresAt var expiresAt *time.Time if rec.ExpiresAt != nil { @@ -239,12 +238,27 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD expiresAt = &t } } + // serialise createdAt + createdAt, err := time.Parse(time.RFC3339, rec.CreatedAt) + if err != nil { + return fmt.Errorf("failed to parse createdAt: %w", err) + } + + var pinnedBy string + if rec.PinnedBy == nil { + pinnedBy = userDID + } else { + pinnedBy = *rec.PinnedBy + } + pin := &model.PinnedRecord{ RKey: rkey.String(), RepoDID: userDID, PinnedMessage: rec.PinnedMessage, + PinnedBy: pinnedBy, + IndexedAt: &now, CID: cid, - CreatedAt: now, + CreatedAt: createdAt, Repo: repo, ExpiresAt: expiresAt, } @@ -256,10 +270,33 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD if err != nil { return fmt.Errorf("failed to get pinned record after we just saved it: %w", err) } - pinnedView, err := pin.ToStreamplacePinnedRecord() + pinnedView, err := pin.ToStreamplacePinnedRecordView() if err != nil { return fmt.Errorf("failed to convert pinned record: %w", err) } + // look up the original message, pinner + msg, err := atsync.Model.GetChatMessage(pinnedView.Record.PinnedMessage) + if err != nil { + return fmt.Errorf("failed to get chat message: %w", err) + } + profile, err := atsync.Model.GetChatProfile(ctx, pinnedBy) + if err != nil { + return fmt.Errorf("failed to get chat profile: %w", err) + } + if msg != nil { + msgView, err := msg.ToStreamplaceMessageView() + if err != nil { + return fmt.Errorf("failed to convert chat message: %w", err) + } + pinnedView.Message = msgView + } + if profile != nil { + profileView, err := profile.ToStreamplaceChatProfile() + if err != nil { + return fmt.Errorf("failed to convert chat profile: %w", err) + } + pinnedView.PinnedBy = profileView + } go atsync.Bus.Publish(userDID, pinnedView) case *streamplace.ChatProfile: diff --git a/pkg/model/pinned_record.go b/pkg/model/pinned_record.go index 90add2f7..10789d9a 100644 --- a/pkg/model/pinned_record.go +++ b/pkg/model/pinned_record.go @@ -15,6 +15,8 @@ type PinnedRecord struct { RepoDID string `json:"repoDID" gorm:"column:repo_did"` Repo *Repo `json:"repo,omitempty" gorm:"foreignKey:DID;references:RepoDID"` PinnedMessage string `gorm:"column:pinned_message" json:"pinnedMessage"` + PinnedBy string `gorm:"column:pinned_by" json:"pinnedBy"` + 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"` } @@ -32,6 +34,26 @@ func (p *PinnedRecord) ToStreamplacePinnedRecord() (*streamplace.ChatPinnedRecor return rec, nil } +func (p *PinnedRecord) ToStreamplacePinnedRecordView() (*streamplace.ChatDefs_PinnedRecordView, error) { + pr := &streamplace.ChatPinnedRecord{ + LexiconTypeID: "place.stream.chat.pinnedRecord", + PinnedMessage: p.PinnedMessage, + CreatedAt: p.CreatedAt.UTC().Format(time.RFC3339), + } + if p.ExpiresAt != nil { + s := p.ExpiresAt.UTC().Format(time.RFC3339) + pr.ExpiresAt = &s + } + rec := &streamplace.ChatDefs_PinnedRecordView{ + LexiconTypeID: "place.stream.chat.defs#pinnedRecordView", + Record: pr, + Cid: p.CID, + IndexedAt: p.CreatedAt.UTC().Format(time.RFC3339Nano), + // message, pinnedby not included, will fill in later + Uri: "at://" + p.RepoDID + "/place.stream.chat.pinnedRecord/" + p.RKey, + } + return rec, nil +} func (m *DBModel) CreatePinnedRecord(ctx context.Context, pin *PinnedRecord) error { return m.DB.Create(pin).Error } diff --git a/pkg/streamplace/cbor_gen.go b/pkg/streamplace/cbor_gen.go index 197bd3b9..6d7456dc 100644 --- a/pkg/streamplace/cbor_gen.go +++ b/pkg/streamplace/cbor_gen.go @@ -3641,12 +3641,16 @@ func (t *ChatPinnedRecord) MarshalCBOR(w io.Writer) error { } cw := cbg.NewCborWriter(w) - fieldCount := 4 + fieldCount := 5 if t.ExpiresAt == nil { fieldCount-- } + if t.PinnedBy == nil { + fieldCount-- + } + if _, err := cw.Write(cbg.CborEncodeMajorType(cbg.MajMap, uint64(fieldCount))); err != nil { return err } @@ -3670,6 +3674,38 @@ func (t *ChatPinnedRecord) MarshalCBOR(w io.Writer) error { return err } + // t.PinnedBy (string) (string) + if t.PinnedBy != nil { + + if len("pinnedBy") > 1000000 { + return xerrors.Errorf("Value in field \"pinnedBy\" was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len("pinnedBy"))); err != nil { + return err + } + if _, err := cw.WriteString(string("pinnedBy")); err != nil { + return err + } + + if t.PinnedBy == nil { + if _, err := cw.Write(cbg.CborNull); err != nil { + return err + } + } else { + if len(*t.PinnedBy) > 1000000 { + return xerrors.Errorf("Value in field t.PinnedBy was too long") + } + + if err := cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len(*t.PinnedBy))); err != nil { + return err + } + if _, err := cw.WriteString(string(*t.PinnedBy)); err != nil { + return err + } + } + } + // t.CreatedAt (string) (string) if len("createdAt") > 1000000 { return xerrors.Errorf("Value in field \"createdAt\" was too long") @@ -3802,6 +3838,27 @@ func (t *ChatPinnedRecord) UnmarshalCBOR(r io.Reader) (err error) { t.LexiconTypeID = string(sval) } + // t.PinnedBy (string) (string) + case "pinnedBy": + + { + b, err := cr.ReadByte() + if err != nil { + return err + } + if b != cbg.CborNull[0] { + if err := cr.UnreadByte(); err != nil { + return err + } + + sval, err := cbg.ReadStringWithMax(cr, 1000000) + if err != nil { + return err + } + + t.PinnedBy = (*string)(&sval) + } + } // t.CreatedAt (string) (string) case "createdAt": diff --git a/pkg/streamplace/chatdefs.go b/pkg/streamplace/chatdefs.go index 0f2aedac..bbb89b04 100644 --- a/pkg/streamplace/chatdefs.go +++ b/pkg/streamplace/chatdefs.go @@ -59,10 +59,11 @@ func (t *ChatDefs_MessageView_ReplyTo) UnmarshalJSON(b []byte) error { // // View of a pinned chat record with hydrated message data. type ChatDefs_PinnedRecordView struct { - Cid string `json:"cid" cborgen:"cid"` - IndexedAt string `json:"indexedAt" cborgen:"indexedAt"` - Message *ChatDefs_MessageView `json:"message,omitempty" cborgen:"message,omitempty"` - PinnedBy *appbsky.ActorDefs_ProfileViewBasic `json:"pinnedBy,omitempty" cborgen:"pinnedBy,omitempty"` - Record *ChatPinnedRecord `json:"record" cborgen:"record"` - Uri string `json:"uri" cborgen:"uri"` + LexiconTypeID string `json:"$type" cborgen:"$type,const=place.stream.chat.defs#pinnedRecordView"` + Cid string `json:"cid" cborgen:"cid"` + IndexedAt string `json:"indexedAt" cborgen:"indexedAt"` + Message *ChatDefs_MessageView `json:"message,omitempty" cborgen:"message,omitempty"` + PinnedBy *ChatProfile `json:"pinnedBy,omitempty" cborgen:"pinnedBy,omitempty"` + Record *ChatPinnedRecord `json:"record" cborgen:"record"` + Uri string `json:"uri" cborgen:"uri"` } diff --git a/pkg/streamplace/chatpinnedRecord.go b/pkg/streamplace/chatpinnedRecord.go index b61cddfd..7c013214 100644 --- a/pkg/streamplace/chatpinnedRecord.go +++ b/pkg/streamplace/chatpinnedRecord.go @@ -18,6 +18,8 @@ type ChatPinnedRecord struct { CreatedAt string `json:"createdAt" cborgen:"createdAt"` // expiresAt: Optional expiration time. If set, the pin is considered inactive after this time. ExpiresAt *string `json:"expiresAt,omitempty" cborgen:"expiresAt,omitempty"` + // pinnedBy: DID of the user who pinned the message. + PinnedBy *string `json:"pinnedBy,omitempty" cborgen:"pinnedBy,omitempty"` // pinnedMessage: AT-URI of the pinned chat message. PinnedMessage string `json:"pinnedMessage" cborgen:"pinnedMessage"` } diff --git a/pkg/streamplace/streamlivestream.go b/pkg/streamplace/streamlivestream.go index aef26c79..11aefb67 100644 --- a/pkg/streamplace/streamlivestream.go +++ b/pkg/streamplace/streamlivestream.go @@ -73,6 +73,7 @@ type Livestream_StreamplaceAnything_Livestream struct { Defs_Renditions *Defs_Renditions Defs_Rendition *Defs_Rendition ChatDefs_MessageView *ChatDefs_MessageView + ChatDefs_PinnedRecordView *ChatDefs_PinnedRecordView } func (t *Livestream_StreamplaceAnything_Livestream) MarshalJSON() ([]byte, error) { @@ -108,6 +109,10 @@ func (t *Livestream_StreamplaceAnything_Livestream) MarshalJSON() ([]byte, error t.ChatDefs_MessageView.LexiconTypeID = "place.stream.chat.defs#messageView" return json.Marshal(t.ChatDefs_MessageView) } + if t.ChatDefs_PinnedRecordView != nil { + t.ChatDefs_PinnedRecordView.LexiconTypeID = "place.stream.chat.defs#pinnedRecordView" + return json.Marshal(t.ChatDefs_PinnedRecordView) + } return nil, fmt.Errorf("can not marshal empty union as JSON") } @@ -142,6 +147,9 @@ func (t *Livestream_StreamplaceAnything_Livestream) UnmarshalJSON(b []byte) erro case "place.stream.chat.defs#messageView": t.ChatDefs_MessageView = new(ChatDefs_MessageView) return json.Unmarshal(b, t.ChatDefs_MessageView) + case "place.stream.chat.defs#pinnedRecordView": + t.ChatDefs_PinnedRecordView = new(ChatDefs_PinnedRecordView) + return json.Unmarshal(b, t.ChatDefs_PinnedRecordView) default: return nil } -- 2.51.2