From 3fd9bd7dcc06abaf63c429bde82e777d1d3b4c7f Mon Sep 17 00:00:00 2001 From: Natalie Bridgers Date: Fri, 11 Sep 2026 00:19:28 -0500 Subject: [PATCH] web: revoke moderation permissions in real time --- .../src/streamplace-store/moderation.tsx | 31 ++++---- js/core/src/livestream-store/state.ts | 13 +++- .../websocket-consumer.test.ts | 77 ++++++++++++++++++- .../livestream-store/websocket-consumer.ts | 69 +++++++++++++---- js/web/src/hooks/use-can-moderate.ts | 3 +- js/web/src/lib/moderation.test.ts | 52 +++++++++++++ js/web/src/lib/moderation.ts | 32 +++++--- pkg/atproto/firehose.go | 5 +- pkg/atproto/moderation_test.go | 31 ++++++++ pkg/atproto/sync.go | 1 + 10 files changed, 266 insertions(+), 48 deletions(-) diff --git a/js/components/src/streamplace-store/moderation.tsx b/js/components/src/streamplace-store/moderation.tsx index 92d96cb27..e8fc143bd 100644 --- a/js/components/src/streamplace-store/moderation.tsx +++ b/js/components/src/streamplace-store/moderation.tsx @@ -3,6 +3,16 @@ import { place } from "streamplace"; import { useLivestreamStore } from "../livestream-store"; import { usePDSAgent } from "./xrpc"; +function permissionRecordsFromListRecords( + records: Array<{ uri?: string; value?: any }>, +): place.stream.moderation.permission.Main[] { + return records + .filter( + (record) => record.value?.$type === "place.stream.moderation.permission", + ) + .map((record) => ({ ...record.value, uri: record.uri })); +} + export interface ModerationPermissions { canBan: boolean; canHide: boolean; @@ -75,13 +85,9 @@ export function useCanModerate( limit: 100, }); - const records = result.data.records || []; - const permissionRecords: place.stream.moderation.permission.Main[] = - records - .map((r: { value: any }) => r.value) - .filter( - (v: any) => v && v.$type === "place.stream.moderation.permission", - ); + const permissionRecords = permissionRecordsFromListRecords( + result.data.records || [], + ); // Store all permissions in the livestream store // WebSocket updates will keep this in sync @@ -119,14 +125,9 @@ export function useCanModerate( limit: 100, }); - const records = result.data.records || []; - const permissionRecords: place.stream.moderation.permission.Main[] = - records - .map((r: { value: any }) => r.value) - .filter( - (v: any) => - v && v.$type === "place.stream.moderation.permission", - ); + const permissionRecords = permissionRecordsFromListRecords( + result.data.records || [], + ); setModerationPermissions(permissionRecords); } catch (err) { diff --git a/js/core/src/livestream-store/state.ts b/js/core/src/livestream-store/state.ts index b56fc483f..79b54f5be 100644 --- a/js/core/src/livestream-store/state.ts +++ b/js/core/src/livestream-store/state.ts @@ -6,6 +6,15 @@ import { place, } from "streamplace"; +/** + * A permission record plus its AT-URI, which is needed to apply deletion + * events to records loaded before the websocket connection was established. + */ +export type LivestreamModerationPermission = + place.stream.moderation.permission.Main & { + uri?: string; + }; + export interface LivestreamState { profile: AppBskyActorDefs.ProfileViewBasic | null; chatIndex: { [key: string]: ChatMessageViewHydrated }; @@ -34,9 +43,9 @@ export interface LivestreamState { websocketConnected: boolean; hasReceivedSegment: boolean; pinnedComment: PinnedRecordViewHydrated | null; - moderationPermissions: place.stream.moderation.permission.Main[]; + moderationPermissions: LivestreamModerationPermission[]; setModerationPermissions: ( - permissions: place.stream.moderation.permission.Main[], + permissions: LivestreamModerationPermission[], ) => void; localLivestreamURI: string | null; setLocalLivestreamURI: (uri: string | null) => void; diff --git a/js/core/src/livestream-store/websocket-consumer.test.ts b/js/core/src/livestream-store/websocket-consumer.test.ts index 3d8cdfb2a..01452e974 100644 --- a/js/core/src/livestream-store/websocket-consumer.test.ts +++ b/js/core/src/livestream-store/websocket-consumer.test.ts @@ -36,6 +36,10 @@ function makeState(overrides: Partial = {}): LivestreamState { const MODERATOR_DID = "did:plc:mod"; const OTHER_MODERATOR_DID = "did:plc:othermod"; +const PERMISSION_URI = + "at://did:plc:streamer/place.stream.moderation.permission/3kq"; +const OTHER_PERMISSION_URI = + "at://did:plc:streamer/place.stream.moderation.permission/3kz"; function permissionRecord(overrides: Record = {}) { return { @@ -43,6 +47,7 @@ function permissionRecord(overrides: Record = {}) { moderator: MODERATOR_DID, permissions: ["ban", "hide"], createdAt: "2024-01-01T00:00:00.000Z", + uri: PERMISSION_URI, ...overrides, }; } @@ -50,13 +55,22 @@ function permissionRecord(overrides: Record = {}) { function permissionView(record = permissionRecord()) { return { $type: "place.stream.moderation.defs#permissionView", - uri: "at://did:plc:streamer/place.stream.moderation.permission/3kq", + uri: PERMISSION_URI, cid: "bafytest", author: { did: "did:plc:streamer", handle: "streamer.bsky.social" }, record, }; } +function deletedPermission(overrides: Record = {}) { + return { + $type: "place.stream.moderation.permission", + deleted: true, + uri: PERMISSION_URI, + ...overrides, + }; +} + describe("handleWebSocketMessages: moderation permission views", () => { it("merges a permission view into an empty store", () => { const state = makeState(); @@ -67,6 +81,7 @@ describe("handleWebSocketMessages: moderation permission views", () => { "ban", "hide", ]); + expect(result.moderationPermissions[0].uri).toBe(PERMISSION_URI); }); it("replaces the record fetched earlier for the same moderator and createdAt", () => { @@ -83,7 +98,10 @@ describe("handleWebSocketMessages: moderation permission views", () => { it("keeps records belonging to other moderators", () => { const state = makeState({ moderationPermissions: [ - permissionRecord({ moderator: OTHER_MODERATOR_DID }) as any, + permissionRecord({ + moderator: OTHER_MODERATOR_DID, + uri: OTHER_PERMISSION_URI, + }) as any, ], }); const result = handleWebSocketMessages(state, [permissionView()]); @@ -97,13 +115,66 @@ describe("handleWebSocketMessages: moderation permission views", () => { it("treats records with a different createdAt as separate delegations", () => { const state = makeState({ moderationPermissions: [ - permissionRecord({ createdAt: "2023-06-01T00:00:00.000Z" }) as any, + permissionRecord({ + createdAt: "2023-06-01T00:00:00.000Z", + uri: OTHER_PERMISSION_URI, + }) as any, ], }); const result = handleWebSocketMessages(state, [permissionView()]); expect(result.moderationPermissions).toHaveLength(2); }); + it("replaces an updated permission view for the same URI", () => { + const state = makeState({ + moderationPermissions: [ + permissionRecord({ permissions: ["ban"] }) as any, + ], + }); + const result = handleWebSocketMessages(state, [ + permissionView( + permissionRecord({ + permissions: ["hide"], + createdAt: "2024-01-02T00:00:00.000Z", + }), + ), + ]); + + expect(result.moderationPermissions).toHaveLength(1); + expect(result.moderationPermissions[0].permissions).toEqual(["hide"]); + }); + + it("removes only the permission named by a deletion event", () => { + const state = makeState({ + moderationPermissions: [ + permissionRecord() as any, + permissionRecord({ + moderator: OTHER_MODERATOR_DID, + uri: OTHER_PERMISSION_URI, + }) as any, + ], + }); + const result = handleWebSocketMessages(state, [deletedPermission()]); + + expect(result.moderationPermissions).toHaveLength(1); + expect(result.moderationPermissions[0].moderator).toBe(OTHER_MODERATOR_DID); + }); + + it("supports the server deletion marker identified by streamer and rkey", () => { + const state = makeState({ + moderationPermissions: [permissionRecord() as any], + }); + const result = handleWebSocketMessages(state, [ + deletedPermission({ + uri: undefined, + streamer: "did:plc:streamer", + rkey: "3kq", + }), + ]); + + expect(result.moderationPermissions).toEqual([]); + }); + it("leaves moderationPermissions alone for unrelated messages", () => { const state = makeState(); const result = handleWebSocketMessages(state, [ diff --git a/js/core/src/livestream-store/websocket-consumer.ts b/js/core/src/livestream-store/websocket-consumer.ts index 1ebf65006..e2ee19059 100644 --- a/js/core/src/livestream-store/websocket-consumer.ts +++ b/js/core/src/livestream-store/websocket-consumer.ts @@ -9,9 +9,37 @@ import { formatHandleWithAt } from "../lib/format-handle"; import { SystemMessages } from "../lib/system-messages"; import { reduceChat } from "./chat-reducer"; import { findProblems } from "./problems"; -import { LivestreamState } from "./state"; +import { LivestreamModerationPermission, LivestreamState } from "./state"; const MAX_RECENT_SEGMENTS = 10; +const MODERATION_PERMISSION_TYPE = "place.stream.moderation.permission"; + +type ModerationPermissionDeletion = { + $type?: unknown; + deleted?: unknown; + uri?: unknown; + streamer?: unknown; + rkey?: unknown; +}; + +function moderationPermissionURI( + message: ModerationPermissionDeletion, +): string | null { + if (typeof message.uri === "string" && message.uri.length > 0) { + return message.uri; + } + + if ( + typeof message.streamer === "string" && + typeof message.rkey === "string" && + message.streamer.length > 0 && + message.rkey.length > 0 + ) { + return `at://${message.streamer}/${MODERATION_PERMISSION_TYPE}/${message.rkey}`; + } + + return null; +} export const handleWebSocketMessages = ( state: LivestreamState, @@ -136,23 +164,38 @@ export const handleWebSocketMessages = ( ...state, pinnedComment: null, }; + } else if ( + (message as ModerationPermissionDeletion).$type === + MODERATION_PERMISSION_TYPE && + (message as ModerationPermissionDeletion).deleted === true + ) { + const deletedURI = moderationPermissionURI( + message as ModerationPermissionDeletion, + ); + if (deletedURI) { + state = { + ...state, + moderationPermissions: state.moderationPermissions.filter( + (permission) => permission.uri !== deletedURI, + ), + }; + } } else if ( place.stream.moderation.defs.permissionView.isTypeOf(message) ) { const view = message as place.stream.moderation.defs.PermissionView; - const record = - view.record as unknown as place.stream.moderation.permission.Main; + const record = { + ...(view.record as unknown as place.stream.moderation.permission.Main), + uri: view.uri, + } as LivestreamModerationPermission; if (record?.moderator) { - // Permission records are immutable, so moderator+createdAt - // identifies a record across the initial listRecords fetch and - // websocket pushes; a re-delivered view replaces its earlier copy - // instead of stacking a duplicate. - const withoutRecord = state.moderationPermissions.filter( - (perm) => - !( - perm.moderator === record.moderator && - perm.createdAt === record.createdAt - ), + const withoutRecord = state.moderationPermissions.filter((perm) => + record.uri + ? perm.uri !== record.uri + : !( + perm.moderator === record.moderator && + perm.createdAt === record.createdAt + ), ); state = { ...state, diff --git a/js/web/src/hooks/use-can-moderate.ts b/js/web/src/hooks/use-can-moderate.ts index 903c4dc08..2cecfc271 100644 --- a/js/web/src/hooks/use-can-moderate.ts +++ b/js/web/src/hooks/use-can-moderate.ts @@ -37,7 +37,8 @@ async function fetchDelegations( * Moderation capabilities of the logged-in viewer for the stream this store * is connected to. The streamer's delegation records are fetched once per * (agent, streamer) pair into the store; websocket permissionView pushes - * keep the store current while the page is open. + * keep the store current while the page is open, including permission + * revocations. */ export function useCanModerate(store: LivestreamStore): UseCanModerateResult { const { pdsAgent, did } = useSession(); diff --git a/js/web/src/lib/moderation.test.ts b/js/web/src/lib/moderation.test.ts index fc90e4ad4..eaea1a35e 100644 --- a/js/web/src/lib/moderation.test.ts +++ b/js/web/src/lib/moderation.test.ts @@ -1,3 +1,7 @@ +import { + handleWebSocketMessages, + makeLivestreamStore, +} from "@streamplace/core"; import { describe, expect, it } from "vitest"; import { moderationPermissionsFor, @@ -29,6 +33,12 @@ describe("permissionRecordsFromListRecords", () => { ]); expect(records).toHaveLength(1); expect(records[0].moderator).toBe(MODERATOR); + expect(records[0].uri).toBeUndefined(); + + const recordsWithURI = permissionRecordsFromListRecords([ + { value: permissionRecord(), uri: "at://streamer/permission/3kq" }, + ]); + expect(recordsWithURI[0].uri).toBe("at://streamer/permission/3kq"); }); }); @@ -121,4 +131,46 @@ describe("moderationPermissionsFor", () => { const perms = moderationPermissionsFor(records, MODERATOR, STREAMER); expect(perms.canBan).toBe(true); }); + + it("removes a moderator's controls when the websocket revokes their permission", () => { + const store = makeLivestreamStore(); + const permission = permissionRecord({ + permissions: ["ban", "hide", "message.pin", "livestream.manage"], + uri: "at://did:plc:streamer/place.stream.moderation.permission/3kq", + }) as ModerationPermissionRecord; + store.setState({ moderationPermissions: [permission] }); + + expect( + moderationPermissionsFor( + store.getState().moderationPermissions, + MODERATOR, + STREAMER, + ).canBan, + ).toBe(true); + + store.setState((state) => + handleWebSocketMessages(state, [ + { + $type: "place.stream.moderation.permission", + deleted: true, + uri: permission.uri, + }, + ]), + ); + + expect(store.getState().moderationPermissions).toEqual([]); + expect( + moderationPermissionsFor( + store.getState().moderationPermissions, + MODERATOR, + STREAMER, + ), + ).toMatchObject({ + canBan: false, + canHide: false, + canPin: false, + canManageLivestream: false, + isOwner: false, + }); + }); }); diff --git a/js/web/src/lib/moderation.ts b/js/web/src/lib/moderation.ts index 36eb75a8f..56f6f3d24 100644 --- a/js/web/src/lib/moderation.ts +++ b/js/web/src/lib/moderation.ts @@ -1,7 +1,6 @@ -import type { place } from "streamplace"; +import type { LivestreamModerationPermission } from "@streamplace/core"; -export type ModerationPermissionRecord = - place.stream.moderation.permission.Main; +export type ModerationPermissionRecord = LivestreamModerationPermission; const PERMISSION_RECORD_TYPE = "place.stream.moderation.permission"; @@ -18,16 +17,25 @@ export interface ModerationPermissions { * permission record values. */ export function permissionRecordsFromListRecords( - records: Array<{ value: unknown } | null | undefined>, + records: Array<{ value: unknown; uri?: string } | null | undefined>, ): ModerationPermissionRecord[] { - return records - .map((r) => r?.value) - .filter( - (value): value is ModerationPermissionRecord => - !!value && - typeof value === "object" && - (value as { $type?: unknown }).$type === PERMISSION_RECORD_TYPE, - ); + return records.flatMap((record) => { + const value = record?.value; + if ( + !value || + typeof value !== "object" || + (value as { $type?: unknown }).$type !== PERMISSION_RECORD_TYPE + ) { + return []; + } + + return [ + { + ...(value as ModerationPermissionRecord), + ...(record.uri ? { uri: record.uri } : {}), + }, + ]; + }); } /** diff --git a/pkg/atproto/firehose.go b/pkg/atproto/firehose.go index d4162b2d5..37b5ba268 100644 --- a/pkg/atproto/firehose.go +++ b/pkg/atproto/firehose.go @@ -777,12 +777,13 @@ func (atsync *ATProtoSynchronizer) handleIndexedOps(ctx context.Context, evt *in err := atsync.Model.DeleteModerationDelegation(ctx, rkey.String()) if err != nil { log.Error(ctx, "failed to delete moderation delegation", "err", err) + continue } // Publish deletion to WebSocket bus for real-time updates - // Create a deleted record marker to notify frontend deletedRecord := map[string]any{ - "$type": "place.stream.moderation.permission", + "$type": constants.PLACE_STREAM_MODERATION_PERMISSION, "deleted": true, + "uri": uri, "rkey": rkey.String(), "streamer": evt.Repo, } diff --git a/pkg/atproto/moderation_test.go b/pkg/atproto/moderation_test.go index 0f0492955..21bcf79ba 100644 --- a/pkg/atproto/moderation_test.go +++ b/pkg/atproto/moderation_test.go @@ -58,6 +58,8 @@ func TestDelegatedModeration(t *testing.T) { streamer := dev.CreateAccount(t) moderator := dev.CreateAccount(t) user := dev.CreateAccount(t) + permissionEvents := b.Subscribe(streamer.DID) + defer b.Unsubscribe(streamer.DID, permissionEvents) // Test 1: Create delegation record with expiration time t.Log("Test 1: Creating delegation record with expiration time") @@ -89,6 +91,7 @@ func TestDelegatedModeration(t *testing.T) { return nil }) require.NoError(t, err) + t.Log("✓ Delegation record ingested successfully") // Verify delegation details including expiration time @@ -97,6 +100,20 @@ func TestDelegatedModeration(t *testing.T) { require.NotNil(t, view) require.Equal(t, streamer.DID, view.Author.Did) + var permissionViewEvent placestream.ModerationDefs_PermissionView + require.Eventually(t, func() bool { + select { + case event := <-permissionEvents: + var ok bool + permissionViewEvent, ok = event.(placestream.ModerationDefs_PermissionView) + return ok + default: + return false + } + }, 5*time.Second, 10*time.Millisecond, "permission creation should reach the websocket bus") + require.Equal(t, "place.stream.moderation.defs#permissionView", permissionViewEvent.LexiconTypeID) + require.Equal(t, view.Uri, permissionViewEvent.Uri) + delegation := view.Record.Val.(*placestream.ModerationPermission) require.NotNil(t, delegation) require.Equal(t, moderator.DID, delegation.Moderator) @@ -231,6 +248,20 @@ func TestDelegatedModeration(t *testing.T) { return nil }) require.NoError(t, err) + + var deletionEvent map[string]any + require.Eventually(t, func() bool { + select { + case event := <-permissionEvents: + var ok bool + deletionEvent, ok = event.(map[string]any) + return ok && deletionEvent["deleted"] == true + default: + return false + } + }, 5*time.Second, 10*time.Millisecond, "permission deletion should reach the websocket bus") + require.Equal(t, constants.PLACE_STREAM_MODERATION_PERMISSION, deletionEvent["$type"]) + require.Equal(t, view.Uri, deletionEvent["uri"]) t.Log("✓ Delegation record deleted successfully") t.Log("All moderation tests passed!") diff --git a/pkg/atproto/sync.go b/pkg/atproto/sync.go index 4e3aca5bf..706054680 100644 --- a/pkg/atproto/sync.go +++ b/pkg/atproto/sync.go @@ -783,6 +783,7 @@ func (atsync *ATProtoSynchronizer) handleCreateUpdate(ctx context.Context, userD } view := placestream.ModerationDefs_PermissionView{ + LexiconTypeID: "place.stream.moderation.defs#permissionView", Uri: aturi.String(), Cid: cid, Author: appbsky.ActorDefs_ProfileViewBasic{ -- knotmirror 0.0.1