From 507fe4497894352858bac47876f06e95663f1907 Mon Sep 17 00:00:00 2001 From: Natalie Bridgers Date: Sat, 1 Aug 2026 16:05:37 -0500 Subject: [PATCH] Add duration and livestream scope to pinned messages Signed-off-by: Natalie Bridgers --- .../src/components/chat/mod-view.tsx | 41 ++- js/components/src/livestream-store/chat.tsx | 11 +- .../chat/place-stream-chat-pinnedrecord.md | 24 +- .../place-stream-moderation-createpin.md | 22 +- .../content/docs/lex-reference/openapi.json | 9 + lexicons.json | 2 +- lexicons/place/stream/chat/pinnedRecord.json | 10 + .../place/stream/moderation/createPin.json | 10 + pkg/atproto/sync.go | 6 + pkg/model/pinned_record.go | 57 ++++- pkg/model/pinned_record_test.go | 238 ++++++++++++++++++ pkg/placestream/chatpinnedrecord.go | 4 + pkg/placestream/moderationcreatepin.go | 4 + pkg/spxrpc/place_stream_moderation.go | 2 + 14 files changed, 416 insertions(+), 24 deletions(-) create mode 100644 pkg/model/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 09151ad1f..7be2933e8 100644 --- a/js/components/src/components/chat/mod-view.tsx +++ b/js/components/src/components/chat/mod-view.tsx @@ -16,6 +16,7 @@ import { Linking } from "react-native"; import { ChatMessageViewHydrated } from "streamplace"; import { useDeleteChatMessage, + useLivestream, useLivestreamStore, usePinChatMessage, } from "../../livestream-store"; @@ -60,6 +61,7 @@ export const ModView = forwardRef(() => { let { createBlock, isLoading: isBlockLoading } = useCreateBlockRecord(); let { createHideChat, isLoading: isHideLoading } = useCreateHideChatRecord(); const pinChatMessage = usePinChatMessage(); + const livestream = useLivestream(); const setReportModalOpen = usePlayerStore((x) => x.setReportModalOpen); const setReportSubject = usePlayerStore((x) => x.setReportSubject); @@ -121,6 +123,7 @@ export const ModView = forwardRef(() => { modPermissions={modPermissions} agent={agent} streamerDID={streamerDID} + livestreamUri={livestream?.uri} hasAvailableActions={hasAvailableActions} isHideLoading={isHideLoading} isBlockLoading={isBlockLoading} @@ -146,6 +149,7 @@ interface ModViewContentProps { modPermissions: ModerationPermissions; agent: ReturnType; streamerDID?: string; + livestreamUri?: string; hasAvailableActions: boolean; isHideLoading: boolean; isBlockLoading: boolean; @@ -156,7 +160,7 @@ interface ModViewContentProps { pinChatMessage: ( messageUri: string, streamerDID: string, - expiresAt?: string, + options?: { expiresAt?: string; duration?: string; livestream?: string }, ) => Promise; toast: ReturnType; setReportModalOpen: (open: boolean) => void; @@ -169,6 +173,7 @@ function ModViewContent({ modPermissions, agent, streamerDID, + livestreamUri, hasAvailableActions, isHideLoading, isBlockLoading, @@ -248,7 +253,9 @@ function ModViewContent({ { if (!streamerDID) return; - pinChatMessage(message.uri, streamerDID) + pinChatMessage(message.uri, streamerDID, { + livestream: livestreamUri, + }) .then(() => { toast.show("Comment pinned", "", { duration: 3 }); onOpenChange?.(false); @@ -264,6 +271,27 @@ function ModViewContent({ > Until stream end + { + if (!streamerDID) return; + pinChatMessage(message.uri, streamerDID, { + duration: "forever", + }) + .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 }, + ); + }); + }} + > + Forever + {[5, 10, 15, 30, 60].map((minutes) => ( { toast.show("Comment pinned", "", { duration: 3 }); onOpenChange?.(false); diff --git a/js/components/src/livestream-store/chat.tsx b/js/components/src/livestream-store/chat.tsx index c80e53964..d5c6efe7d 100644 --- a/js/components/src/livestream-store/chat.tsx +++ b/js/components/src/livestream-store/chat.tsx @@ -491,19 +491,24 @@ export const usePinChatMessage = () => { return async ( messageUri: string, streamerDID: string, - expiresAt?: string, + options?: { expiresAt?: string; duration?: string; livestream?: string }, ) => { if (!agent || !agent.did) { throw new Error("No PDS agent or user DID found"); } + const extra: Record = {}; + if (options?.expiresAt) extra.expiresAt = options.expiresAt; + if (options?.duration) extra.duration = options.duration; + if (options?.livestream) extra.livestream = options.livestream; + // If streamer, create directly if (agent.did === streamerDID) { const record = { $type: "place.stream.chat.pinnedRecord", pinnedMessage: messageUri, createdAt: new Date().toISOString(), - ...(expiresAt ? { expiresAt } : {}), + ...extra, }; const result = await agent.com.atproto.repo.createRecord({ @@ -518,7 +523,7 @@ export const usePinChatMessage = () => { const result = await agent.client.call(place.stream.moderation.createPin, { streamer: streamerDID, messageUri, - ...(expiresAt ? { expiresAt } : {}), + ...extra, } as any); return result; }; 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..a39824a27 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,14 @@ 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. If set, the pin is considered inactive after this time. | Format: `datetime` | +| `duration` | `string` | ❌ | How long this pin lasts when expiresAt is not set. 'streamEnd' (the default when unset) keeps the pin active until the livestream referenced by 'livestream' ends. 'forever' keeps the pin active across streams until manually unpinned. | Known Values: `streamEnd`, `forever` | +| `livestream` | `string` | ❌ | AT-URI of the place.stream.livestream record this pin is scoped to. Set when duration is 'streamEnd' (including the unset default); the pin is inactive unless that livestream exists and has not ended. | Format: `at-uri` | --- @@ -62,6 +64,16 @@ Record pinning a chat message for prominent display. "type": "string", "format": "datetime", "description": "Optional expiration time. If set, the pin is considered inactive after this time." + }, + "duration": { + "type": "string", + "knownValues": ["streamEnd", "forever"], + "description": "How long this pin lasts when expiresAt is not set. 'streamEnd' (the default when unset) keeps the pin active until the livestream referenced by 'livestream' ends. 'forever' keeps the pin active across streams until manually unpinned." + }, + "livestream": { + "type": "string", + "format": "at-uri", + "description": "AT-URI of the place.stream.livestream record this pin is scoped to. Set when duration is 'streamEnd' (including the unset default); the pin is inactive unless that livestream exists and has not ended." } } } 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..9a01bc383 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,13 @@ 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. | Format: `datetime` | +| `duration` | `string` | ❌ | Pin duration, passed through to the pinned record. Defaults to 'streamEnd' when unset. | Known Values: `streamEnd`, `forever` | +| `livestream` | `string` | ❌ | AT-URI of the livestream this pin is scoped to, passed through to the pinned record. Required for 'streamEnd' pins to be active. | Format: `at-uri` | **Output:** @@ -80,6 +82,16 @@ Pin a chat message on behalf of a streamer. Requires 'message.pin' permission. C "type": "string", "format": "datetime", "description": "Optional expiration time for this pin." + }, + "duration": { + "type": "string", + "knownValues": ["streamEnd", "forever"], + "description": "Pin duration, passed through to the pinned record. Defaults to 'streamEnd' when unset." + }, + "livestream": { + "type": "string", + "format": "at-uri", + "description": "AT-URI of the livestream this pin is scoped to, passed through to the pinned record. Required for 'streamEnd' pins to be active." } } } diff --git a/js/docs/src/content/docs/lex-reference/openapi.json b/js/docs/src/content/docs/lex-reference/openapi.json index 09e35d024..f70f76865 100644 --- a/js/docs/src/content/docs/lex-reference/openapi.json +++ b/js/docs/src/content/docs/lex-reference/openapi.json @@ -2128,6 +2128,15 @@ "type": "string", "description": "Optional expiration time for this pin.", "format": "date-time" + }, + "duration": { + "type": "string", + "description": "Pin duration, passed through to the pinned record. Defaults to 'streamEnd' when unset." + }, + "livestream": { + "type": "string", + "description": "AT-URI of the livestream this pin is scoped to, passed through to the pinned record. Required for 'streamEnd' pins to be active.", + "format": "uri" } }, "required": ["streamer", "messageUri"] diff --git a/lexicons.json b/lexicons.json index d9b399e4d..fdb7373fc 100644 --- a/lexicons.json +++ b/lexicons.json @@ -227,4 +227,4 @@ "cid": "bafyreic3l2rmh2ugirt3jz372wcvy333m7t2ynlyzj2k54oshijs6lxdfu" } } -} \ No newline at end of file +} diff --git a/lexicons/place/stream/chat/pinnedRecord.json b/lexicons/place/stream/chat/pinnedRecord.json index 32b5b33e0..f46b2d1b4 100644 --- a/lexicons/place/stream/chat/pinnedRecord.json +++ b/lexicons/place/stream/chat/pinnedRecord.json @@ -29,6 +29,16 @@ "type": "string", "format": "datetime", "description": "Optional expiration time. If set, the pin is considered inactive after this time." + }, + "duration": { + "type": "string", + "knownValues": ["streamEnd", "forever"], + "description": "How long this pin lasts when expiresAt is not set. 'streamEnd' (the default when unset) keeps the pin active until the livestream referenced by 'livestream' ends. 'forever' keeps the pin active across streams until manually unpinned." + }, + "livestream": { + "type": "string", + "format": "at-uri", + "description": "AT-URI of the place.stream.livestream record this pin is scoped to. Set when duration is 'streamEnd' (including the unset default); the pin is inactive unless that livestream exists and has not ended." } } } diff --git a/lexicons/place/stream/moderation/createPin.json b/lexicons/place/stream/moderation/createPin.json index b7be8c50f..45511859b 100644 --- a/lexicons/place/stream/moderation/createPin.json +++ b/lexicons/place/stream/moderation/createPin.json @@ -25,6 +25,16 @@ "type": "string", "format": "datetime", "description": "Optional expiration time for this pin." + }, + "duration": { + "type": "string", + "knownValues": ["streamEnd", "forever"], + "description": "Pin duration, passed through to the pinned record. Defaults to 'streamEnd' when unset." + }, + "livestream": { + "type": "string", + "format": "at-uri", + "description": "AT-URI of the livestream this pin is scoped to, passed through to the pinned record. Required for 'streamEnd' pins to be active." } } } diff --git a/pkg/atproto/sync.go b/pkg/atproto/sync.go index a659a57f8..4a0a2b666 100644 --- a/pkg/atproto/sync.go +++ b/pkg/atproto/sync.go @@ -286,6 +286,12 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD Repo: repo, ExpiresAt: expiresAt, } + if rec.Duration != nil { + pin.Duration = *rec.Duration + } + if rec.Livestream != nil { + pin.LivestreamURI = *rec.Livestream + } err = atsync.Model.CreatePinnedRecord(ctx, pin) if errors.Is(err, model.ErrAlreadyIndexed) { return nil diff --git a/pkg/model/pinned_record.go b/pkg/model/pinned_record.go index a541aa4db..e2998e190 100644 --- a/pkg/model/pinned_record.go +++ b/pkg/model/pinned_record.go @@ -5,6 +5,8 @@ import ( "errors" "time" + glex "github.com/streamplace/glex/runtime" + "github.com/bluesky-social/indigo/util" "gorm.io/gorm" "stream.place/streamplace/pkg/placestream" @@ -20,6 +22,8 @@ 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"` + Duration string `gorm:"column:duration" json:"duration,omitempty"` + LivestreamURI string `gorm:"column:livestream_uri" json:"livestreamURI,omitempty"` } func (p *PinnedRecord) ToStreamplacePinnedRecord() (placestream.ChatPinnedRecord, error) { @@ -32,6 +36,12 @@ func (p *PinnedRecord) ToStreamplacePinnedRecord() (placestream.ChatPinnedRecord s := p.ExpiresAt.UTC().Format(util.ISO8601) rec.ExpiresAt = &s } + if p.Duration != "" { + rec.Duration = &p.Duration + } + if p.LivestreamURI != "" { + rec.Livestream = &p.LivestreamURI + } return rec, nil } @@ -45,6 +55,12 @@ func (p *PinnedRecord) ToStreamplacePinnedRecordView() (placestream.ChatDefs_Pin s := p.ExpiresAt.UTC().Format(util.ISO8601) pr.ExpiresAt = &s } + if p.Duration != "" { + pr.Duration = &p.Duration + } + if p.LivestreamURI != "" { + pr.Livestream = &p.LivestreamURI + } rec := placestream.ChatDefs_PinnedRecordView{ LexiconTypeID: "place.stream.chat.defs#pinnedRecordView", Record: pr, @@ -81,9 +97,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 +107,43 @@ func (m *DBModel) GetActivePinnedRecord(ctx context.Context, streamerDID string) if err != nil { return nil, err } + active, err := m.isPinActive(&pin) + if err != nil { + return nil, err + } + if !active { + return nil, nil + } return &pin, nil } + +// isPinActive checks whether a pin should currently be shown. The most recent +// pin record for a streamer is authoritative: if it is expired, or scoped to +// a livestream (duration="streamEnd" or unset) whose stream has ended, the +// streamer has no active pin — older records are not resurrected. +func (m *DBModel) isPinActive(pin *PinnedRecord) (bool, error) { + if pin.ExpiresAt != nil && pin.ExpiresAt.Before(time.Now()) { + return false, nil + } + if pin.Duration == "forever" { + return true, nil + } + // Legacy pins with no livestream scope are always active. + if pin.LivestreamURI == "" { + return true, nil + } + ls, err := m.GetLivestream(pin.LivestreamURI) + if err != nil { + return false, err + } + if ls == nil { + return false, nil + } + if ls.Livestream != nil { + var rec placestream.Livestream + if err := glex.DecodeCBOR(*ls.Livestream, &rec); err == nil && rec.EndedAt != nil { + return false, nil + } + } + return true, nil +} diff --git a/pkg/model/pinned_record_test.go b/pkg/model/pinned_record_test.go new file mode 100644 index 000000000..172559356 --- /dev/null +++ b/pkg/model/pinned_record_test.go @@ -0,0 +1,238 @@ +package model + +import ( + "bytes" + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/placestream" +) + +// putLivestream creates a Livestream row whose CBOR blob is the given record. +// endedAt may be nil for an active stream. +func putLivestream(t *testing.T, m Model, uri string, endedAt *string) { + t.Helper() + ctx := context.Background() + rec := placestream.Livestream{ + LexiconTypeID: "place.stream.livestream", + CreatedAt: time.Now().UTC().Format(time.RFC3339), + EndedAt: endedAt, + Title: "test stream", + } + var buf bytes.Buffer + require.NoError(t, rec.MarshalCBOR(&buf)) + ls := &Livestream{ + URI: uri, + CID: "bafytest", + CreatedAt: time.Now().UTC(), + Livestream: ptrTo(buf.Bytes()), + RepoDID: "did:plc:streamer", + } + require.NoError(t, m.CreateLivestream(ctx, ls)) +} + +func ptrTo[T any](v T) *T { return &v } + +// putPinnedRecord inserts a PinnedRecord directly for testing. +func putPinnedRecord(t *testing.T, m Model, uri, streamerDID string, createdAt time.Time, duration, livestreamURI string, expiresAt *time.Time) { + t.Helper() + pin := &PinnedRecord{ + Uri: uri, + CID: "bafypin", + RepoDID: streamerDID, + PinnedMessage: "at://did:plc:user/place.stream.chat.message/msg1", + PinnedBy: streamerDID, + CreatedAt: createdAt, + Duration: duration, + LivestreamURI: livestreamURI, + ExpiresAt: expiresAt, + } + require.NoError(t, m.CreatePinnedRecord(context.Background(), pin)) +} + +func TestGetActivePinnedRecord_LegacyPinAlwaysActive(t *testing.T) { + m, err := MakeDB(":memory:") + require.NoError(t, err) + ctx := context.Background() + + const streamer = "did:plc:streamer" + putPinnedRecord(t, m, "at://did:plc:streamer/place.stream.chat.pinnedRecord/p1", streamer, time.Now(), "", "", nil) + + pin, err := m.GetActivePinnedRecord(ctx, streamer) + require.NoError(t, err) + require.NotNil(t, pin, "legacy pin with no duration/livestream should be active") + require.Equal(t, "at://did:plc:streamer/place.stream.chat.pinnedRecord/p1", pin.Uri) +} + +func TestGetActivePinnedRecord_ForeverPinAlwaysActive(t *testing.T) { + m, err := MakeDB(":memory:") + require.NoError(t, err) + ctx := context.Background() + + const streamer = "did:plc:streamer" + const lsURI = "at://did:plc:streamer/place.stream.livestream/ls1" + // Create an ended livestream — the forever pin should still be active. + putLivestream(t, m, lsURI, ptrTo(time.Now().UTC().Format(time.RFC3339))) + putPinnedRecord(t, m, "at://did:plc:streamer/place.stream.chat.pinnedRecord/p1", streamer, time.Now(), "forever", lsURI, nil) + + pin, err := m.GetActivePinnedRecord(ctx, streamer) + require.NoError(t, err) + require.NotNil(t, pin, "forever pin should be active even if livestream ended") +} + +func TestGetActivePinnedRecord_StreamEndWithActiveLivestream(t *testing.T) { + m, err := MakeDB(":memory:") + require.NoError(t, err) + ctx := context.Background() + + const streamer = "did:plc:streamer" + const lsURI = "at://did:plc:streamer/place.stream.livestream/ls1" + putLivestream(t, m, lsURI, nil) // active stream + putPinnedRecord(t, m, "at://did:plc:streamer/place.stream.chat.pinnedRecord/p1", streamer, time.Now(), "", lsURI, nil) + + pin, err := m.GetActivePinnedRecord(ctx, streamer) + require.NoError(t, err) + require.NotNil(t, pin, "streamEnd pin with active livestream should be active") +} + +func TestGetActivePinnedRecord_StreamEndWithEndedLivestream(t *testing.T) { + m, err := MakeDB(":memory:") + require.NoError(t, err) + ctx := context.Background() + + const streamer = "did:plc:streamer" + const lsURI = "at://did:plc:streamer/place.stream.livestream/ls1" + putLivestream(t, m, lsURI, ptrTo(time.Now().UTC().Format(time.RFC3339))) // ended + putPinnedRecord(t, m, "at://did:plc:streamer/place.stream.chat.pinnedRecord/p1", streamer, time.Now(), "", lsURI, nil) + + pin, err := m.GetActivePinnedRecord(ctx, streamer) + require.NoError(t, err) + require.Nil(t, pin, "streamEnd pin with ended livestream should be inactive") +} + +func TestGetActivePinnedRecord_StreamEndWithMissingLivestream(t *testing.T) { + m, err := MakeDB(":memory:") + require.NoError(t, err) + ctx := context.Background() + + const streamer = "did:plc:streamer" + putPinnedRecord(t, m, "at://did:plc:streamer/place.stream.chat.pinnedRecord/p1", streamer, time.Now(), "", "at://did:plc:streamer/place.stream.livestream/nonexistent", nil) + + pin, err := m.GetActivePinnedRecord(ctx, streamer) + require.NoError(t, err) + require.Nil(t, pin, "streamEnd pin with missing livestream should be inactive") +} + +func TestGetActivePinnedRecord_NewestSupersedesOlder(t *testing.T) { + m, err := MakeDB(":memory:") + require.NoError(t, err) + ctx := context.Background() + + const streamer = "did:plc:streamer" + const ls1URI = "at://did:plc:streamer/place.stream.livestream/ls1" + + // ls1 has ended. + putLivestream(t, m, ls1URI, ptrTo(time.Now().UTC().Format(time.RFC3339))) + + // Pin for ended stream (newest) supersedes the older forever pin. + putPinnedRecord(t, m, "at://did:plc:streamer/place.stream.chat.pinnedRecord/p1", streamer, time.Now().Add(-1*time.Minute), "forever", "", nil) + putPinnedRecord(t, m, "at://did:plc:streamer/place.stream.chat.pinnedRecord/p2", streamer, time.Now(), "", ls1URI, nil) + + pin, err := m.GetActivePinnedRecord(ctx, streamer) + require.NoError(t, err) + require.Nil(t, pin, "most recent pin's stream ended, so no older pin should be resurrected") +} + +// TestGetActivePinnedRecord_PinForeverThenPinStream_StreamEnds covers the +// product scenario: pin a message forever, then pin a message scoped to the +// current stream. When the stream ends, the stream pin goes inactive and the +// forever pin must NOT come back — the most recent pin record is +// authoritative. +func TestGetActivePinnedRecord_PinForeverThenPinStream_StreamEnds(t *testing.T) { + m, err := MakeDB(":memory:") + require.NoError(t, err) + ctx := context.Background() + + const streamer = "did:plc:streamer" + const lsURI = "at://did:plc:streamer/place.stream.livestream/ls1" + putLivestream(t, m, lsURI, ptrTo(time.Now().UTC().Format(time.RFC3339))) // ended + + putPinnedRecord(t, m, "at://did:plc:streamer/place.stream.chat.pinnedRecord/p1", streamer, time.Now().Add(-2*time.Minute), "forever", "", nil) + putPinnedRecord(t, m, "at://did:plc:streamer/place.stream.chat.pinnedRecord/p2", streamer, time.Now(), "", lsURI, nil) + + pin, err := m.GetActivePinnedRecord(ctx, streamer) + require.NoError(t, err) + require.Nil(t, pin, "stream pin superseded the forever pin; after stream end there is no pin") +} + +func TestGetActivePinnedRecord_NewestActiveWins(t *testing.T) { + m, err := MakeDB(":memory:") + require.NoError(t, err) + ctx := context.Background() + + const streamer = "did:plc:streamer" + const lsURI = "at://did:plc:streamer/place.stream.livestream/ls1" + putLivestream(t, m, lsURI, nil) // active + + putPinnedRecord(t, m, "at://did:plc:streamer/place.stream.chat.pinnedRecord/p1", streamer, time.Now().Add(-2*time.Minute), "forever", "", nil) + putPinnedRecord(t, m, "at://did:plc:streamer/place.stream.chat.pinnedRecord/p2", streamer, time.Now(), "", lsURI, nil) + + pin, err := m.GetActivePinnedRecord(ctx, streamer) + require.NoError(t, err) + require.NotNil(t, pin, "most recent pin is active, so it wins over the older forever pin") + require.Equal(t, "at://did:plc:streamer/place.stream.chat.pinnedRecord/p2", pin.Uri) +} + +func TestGetActivePinnedRecord_ExpiredPinSkipped(t *testing.T) { + m, err := MakeDB(":memory:") + require.NoError(t, err) + ctx := context.Background() + + const streamer = "did:plc:streamer" + past := time.Now().Add(-1 * time.Hour) + putPinnedRecord(t, m, "at://did:plc:streamer/place.stream.chat.pinnedRecord/p1", streamer, time.Now(), "forever", "", &past) + + pin, err := m.GetActivePinnedRecord(ctx, streamer) + require.NoError(t, err) + require.Nil(t, pin, "expired pin should not be returned even if duration is forever") +} + +func TestGetActivePinnedRecord_NewestExpiredSupersedesOlder(t *testing.T) { + m, err := MakeDB(":memory:") + require.NoError(t, err) + ctx := context.Background() + + const streamer = "did:plc:streamer" + past := time.Now().Add(-1 * time.Hour) + + putPinnedRecord(t, m, "at://did:plc:streamer/place.stream.chat.pinnedRecord/p1", streamer, time.Now().Add(-2*time.Minute), "forever", "", nil) + putPinnedRecord(t, m, "at://did:plc:streamer/place.stream.chat.pinnedRecord/p2", streamer, time.Now(), "forever", "", &past) + + pin, err := m.GetActivePinnedRecord(ctx, streamer) + require.NoError(t, err) + require.Nil(t, pin, "expired newest pin supersedes the older active one") +} + +func TestGetActivePinnedRecord_DurationAndLivestreamRoundTrip(t *testing.T) { + m, err := MakeDB(":memory:") + require.NoError(t, err) + ctx := context.Background() + + const streamer = "did:plc:streamer" + const lsURI = "at://did:plc:streamer/place.stream.livestream/ls1" + putLivestream(t, m, lsURI, nil) + putPinnedRecord(t, m, "at://did:plc:streamer/place.stream.chat.pinnedRecord/p1", streamer, time.Now(), "forever", lsURI, nil) + + pin, err := m.GetActivePinnedRecord(ctx, streamer) + require.NoError(t, err) + require.NotNil(t, pin) + + pr, err := pin.ToStreamplacePinnedRecord() + require.NoError(t, err) + require.NotNil(t, pr.Duration) + require.Equal(t, "forever", *pr.Duration) + require.NotNil(t, pr.Livestream) + require.Equal(t, lsURI, *pr.Livestream) +} diff --git a/pkg/placestream/chatpinnedrecord.go b/pkg/placestream/chatpinnedrecord.go index 7bbabfef8..55c473c2a 100644 --- a/pkg/placestream/chatpinnedrecord.go +++ b/pkg/placestream/chatpinnedrecord.go @@ -21,8 +21,12 @@ type ChatPinnedRecord struct { LexiconTypeID string `json:"$type,omitempty"` // createdAt: When this pin was created. CreatedAt string `json:"createdAt"` + // duration: How long this pin lasts when expiresAt is not set. 'streamEnd' (the default when unset) keeps the pin active until the livestream referenced by 'livestream' ends. 'forever' keeps the pin active across streams until manually unpinned. + Duration *string `json:"duration,omitempty"` // expiresAt: Optional expiration time. If set, the pin is considered inactive after this time. ExpiresAt *string `json:"expiresAt,omitempty"` + // livestream: AT-URI of the place.stream.livestream record this pin is scoped to. Set when duration is 'streamEnd' (including the unset default); the pin is inactive unless that livestream exists and has not ended. + 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/moderationcreatepin.go b/pkg/placestream/moderationcreatepin.go index 21dc95422..a56aa3ce9 100644 --- a/pkg/placestream/moderationcreatepin.go +++ b/pkg/placestream/moderationcreatepin.go @@ -14,8 +14,12 @@ import ( type ModerationCreatePin_Input struct { LexiconTypeID string `json:"$type,omitempty"` + // duration: Pin duration, passed through to the pinned record. Defaults to 'streamEnd' when unset. + Duration *string `json:"duration,omitempty"` // expiresAt: Optional expiration time for this pin. ExpiresAt *string `json:"expiresAt,omitempty"` + // livestream: AT-URI of the livestream this pin is scoped to, passed through to the pinned record. Required for 'streamEnd' pins to be active. + 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/place_stream_moderation.go b/pkg/spxrpc/place_stream_moderation.go index 06b07bf4f..3f9295de7 100644 --- a/pkg/spxrpc/place_stream_moderation.go +++ b/pkg/spxrpc/place_stream_moderation.go @@ -376,6 +376,8 @@ func (s *Server) handlePlaceStreamModerationCreatePin(ctx context.Context, input PinnedMessage: input.MessageUri, CreatedAt: time.Now().UTC().Format(time.RFC3339), ExpiresAt: input.ExpiresAt, + Duration: input.Duration, + Livestream: input.Livestream, } createInput := comatproto.RepoCreateRecord_Input{ -- 2.51.2