diff --git a/js/components/src/components/chat/mod-view.tsx b/js/components/src/components/chat/mod-view.tsx index 09151ad1..7be2933e 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 c80e5396..d5c6efe7 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 fc88032a..a39824a2 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 5c1ce46e..9a01bc38 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 09e35d02..f70f7686 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 d9b399e4..fdb7373f 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 32b5b33e..f46b2d1b 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 b7be8c50..45511859 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 a659a57f..4a0a2b66 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 a541aa4d..e2998e19 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 00000000..17255935 --- /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 7bbabfef..55c473c2 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 21dc9542..a56aa3ce 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 06b07bf4..3f9295de 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{