ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-24 11:25:50 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(matrix): preserve media, reaction, and member state (#116890)

Browse Source 
Co-authored-by: Peter Steinberger <steipete@macos.shared>
This commit is contained in:
Peter Steinberger
2026-07-31 08:08:32 -07:00
committed by GitHub
parent 39d4ca8f4e
commit 9922ea02e9
9 changed files with 373 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
extensions/matrix/src/matrix/actions/reactions.test.ts
+102 -15
View File
@@ -11,21 +11,27 @@ function createReactionsClient(params: {
}>;
userId?: string | null;
}) {
const doRequest = vi.fn(async (_method: string, _path: string, _query: unknown) => ({
chunk: params.chunk.map((item) => ({
event_id: item.event_id ?? "",
sender: item.sender ?? "",
content: item.key
? {
"m.relates_to": {
rel_type: "m.annotation",
event_id: "$target",
key: item.key,
},
}
: {},
})),
}));
const doRequest = vi.fn(
async (
_method: string,
_path: string,
_query: unknown,
): Promise<{ chunk: Array<Record<string, unknown>>; next_batch?: string }> => ({
chunk: params.chunk.map((item) => ({
event_id: item.event_id ?? "",
sender: item.sender ?? "",
content: item.key
? {
"m.relates_to": {
rel_type: "m.annotation",
event_id: "$target",
key: item.key,
},
}
: {},
})),
}),
);
const getUserId = vi.fn(async () => params.userId ?? null);
const redactEvent = vi.fn(async () => undefined);
@@ -94,6 +100,87 @@ describe("matrix reaction actions", () => {
expect(redactEvent).toHaveBeenCalledWith("!room:example.org", "$1");
});
it("removes current-user reactions found after the first relations page", async () => {
const { client, doRequest, redactEvent } = createReactionsClient({
chunk: [],
userId: "@me:example.org",
});
doRequest
.mockResolvedValueOnce({
chunk: [
{
event_id: "$other",
sender: "@other:example.org",
content: {
"m.relates_to": { rel_type: "m.annotation", event_id: "$msg", key: "👍" },
},
},
],
next_batch: "older-reactions",
})
.mockResolvedValueOnce({
chunk: [
{
event_id: "$mine",
sender: "@me:example.org",
content: {
"m.relates_to": { rel_type: "m.annotation", event_id: "$msg", key: "👍" },
},
},
],
});
await expect(
removeMatrixReactions("!room:example.org", "$msg", { client, emoji: "👍" }),
).resolves.toEqual({ removed: 1 });
expect(doRequest).toHaveBeenNthCalledWith(
2,
"GET",
"/_matrix/client/v1/rooms/!room%3Aexample.org/relations/%24msg/m.annotation/m.reaction",
{ dir: "b", limit: 200, from: "older-reactions" },
);
expect(redactEvent).toHaveBeenCalledWith("!room:example.org", "$mine");
});
it("continues listing across empty relation pages", async () => {
const { client, doRequest } = createReactionsClient({
chunk: [],
userId: "@me:example.org",
});
doRequest
.mockResolvedValueOnce({ chunk: [], next_batch: "older-reactions" })
.mockResolvedValueOnce({
chunk: [
{
event_id: "$mine",
sender: "@me:example.org",
content: {
"m.relates_to": { rel_type: "m.annotation", event_id: "$msg", key: "👍" },
},
},
],
});
await expect(listMatrixReactions("!room:example.org", "$msg", { client })).resolves.toEqual([
{ key: "👍", count: 1, users: ["@me:example.org"] },
]);
expect(doRequest).toHaveBeenCalledTimes(2);
});
it("fails visibly when the relations server repeats a pagination cursor", async () => {
const { client, doRequest, redactEvent } = createReactionsClient({
chunk: [],
userId: "@me:example.org",
});
doRequest.mockResolvedValue({ chunk: [], next_batch: "same-cursor" });
await expect(removeMatrixReactions("!room:example.org", "$msg", { client })).rejects.toThrow(
"repeated cursor",
);
expect(doRequest).toHaveBeenCalledTimes(2);
expect(redactEvent).not.toHaveBeenCalled();
});
it("returns removed=0 when current user id is unavailable", async () => {
const { client, redactEvent } = createReactionsClient({
chunk: [{ event_id: "$1", sender: "@me:example.org", key: "👍" }],
extensions/matrix/src/matrix/actions/reactions.ts
+32 -6
View File
@@ -15,12 +15,34 @@ async function listMatrixReactionEvents(
roomId: string,
messageId: string,
limit: number,
opts: { allPages?: boolean } = {},
): Promise<MatrixRawEvent[]> {
const res = (await client.doRequest("GET", buildMatrixReactionRelationsPath(roomId, messageId), {
dir: "b",
limit,
})) as { chunk?: MatrixRawEvent[] };
return Array.isArray(res.chunk) ? res.chunk : [];
const events: MatrixRawEvent[] = [];
const seenCursors = new Set<string>();
let cursor: string | undefined;
while (true) {
const res = (await client.doRequest(
"GET",
buildMatrixReactionRelationsPath(roomId, messageId),
{
dir: "b",
limit,
...(cursor ? { from: cursor } : {}),
},
)) as { chunk?: MatrixRawEvent[]; next_batch?: unknown };
if (Array.isArray(res.chunk)) {
events.push(...res.chunk);
}
const nextCursor = typeof res.next_batch === "string" ? res.next_batch.trim() : "";
if (!nextCursor || (!opts.allPages && events.length >= limit)) {
return events;
}
if (seenCursors.has(nextCursor)) {
throw new Error("Matrix reaction pagination returned a repeated cursor");
}
seenCursors.add(nextCursor);
cursor = nextCursor;
}
}
export async function listMatrixReactions(
@@ -41,7 +63,11 @@ export async function removeMatrixReactions(
opts: MatrixActionClientOpts & { emoji?: string } = {},
): Promise<{ removed: number }> {
return await withResolvedRoomAction(roomId, opts, async (client, resolvedRoom) => {
const chunk = await listMatrixReactionEvents(client, resolvedRoom, messageId, 200);
// A message can have hundreds of newer reactions; the bot's own reaction may
// be on a later page, so fetch all pages before mutating any server state.
const chunk = await listMatrixReactionEvents(client, resolvedRoom, messageId, 200, {
allPages: true,
});
const userId = await client.getUserId();
if (!userId) {
return { removed: 0 };
extensions/matrix/src/matrix/monitor/events.test.ts
+24 -2
View File
@@ -135,6 +135,7 @@ function createHarness(params?: {
);
const sendMessage = vi.fn(async (_roomId: string, _payload: { body?: string }) => "$notice");
const invalidateRoom = vi.fn();
const invalidateMemberDisplayName = vi.fn();
const rememberInvite = vi.fn();
const logger = { info: vi.fn(), warn: vi.fn(), error: vi.fn() };
const formatNativeDependencyHint = vi.fn(() => "install hint");
@@ -200,6 +201,7 @@ function createHarness(params?: {
invalidateRoom,
rememberInvite,
},
invalidateMemberDisplayName,
logVerboseMessage,
warnedEncryptedRooms: new Set<string>(),
warnedCryptoMissingRooms: new Set<string>(),
@@ -223,6 +225,7 @@ function createHarness(params?: {
onRoomMessage,
sendMessage,
invalidateRoom,
invalidateMemberDisplayName,
rememberInvite,
roomEventListener,
listVerifications,
@@ -318,8 +321,8 @@ describe("registerMatrixMonitorEvents verification routing", () => {
expect(sendMessage).not.toHaveBeenCalled();
});
it("invalidates direct-room membership cache on room member events", () => {
const { invalidateRoom, roomEventListener } = createHarness();
it("invalidates direct-room and observed member-display-name caches on room member events", () => {
const { invalidateRoom, invalidateMemberDisplayName, roomEventListener } = createHarness();
roomEventListener("!room:example.org", {
event_id: "$member1",
@@ -333,6 +336,25 @@ describe("registerMatrixMonitorEvents verification routing", () => {
});
expect(invalidateRoom).toHaveBeenCalledWith("!room:example.org");
expect(invalidateMemberDisplayName).toHaveBeenCalledWith(
"!room:example.org",
"@mallory:example.org",
);
});
it("does not invalidate a member display name without an authoritative state key", () => {
const { invalidateRoom, invalidateMemberDisplayName, roomEventListener } = createHarness();
roomEventListener("!room:example.org", {
event_id: "$member-no-state-key",
sender: "@alice:example.org",
type: EventType.RoomMember,
origin_server_ts: Date.now(),
content: { membership: "join" },
});
expect(invalidateRoom).toHaveBeenCalledWith("!room:example.org");
expect(invalidateMemberDisplayName).not.toHaveBeenCalled();
});
it("remembers invite provenance on room invites", () => {
extensions/matrix/src/matrix/monitor/events.ts
+5
View File
@@ -179,6 +179,7 @@ export function registerMatrixMonitorEvents(params: {
invalidateRoom: (roomId: string) => void;
rememberInvite?: (roomId: string, remoteUserId: string) => void;
};
invalidateMemberDisplayName?: (roomId: string, userId: string) => void;
logVerboseMessage: (message: string) => void;
warnedEncryptedRooms: Set<string>;
warnedCryptoMissingRooms: Set<string>;
@@ -199,6 +200,7 @@ export function registerMatrixMonitorEvents(params: {
dmPolicy,
readStoreAllowFrom,
directTracker,
invalidateMemberDisplayName,
logVerboseMessage,
warnedEncryptedRooms,
warnedCryptoMissingRooms,
@@ -386,6 +388,9 @@ export function registerMatrixMonitorEvents(params: {
directTracker?.invalidateRoom(roomId);
const membership = (event?.content as { membership?: string } | undefined)?.membership;
const stateKey = (event as { state_key?: string }).state_key ?? "";
if (stateKey) {
invalidateMemberDisplayName?.(roomId, stateKey);
}
logVerboseMessage(
`matrix: member event room=${roomId} stateKey=${stateKey} membership=${membership ?? "unknown"}`,
);
extensions/matrix/src/matrix/monitor/index.ts
+3 -1
View File
@@ -317,7 +317,8 @@ export async function monitorMatrixProvider(opts: MonitorMatrixOpts = {}): Promi
// Cold starts should ignore old room history, but once we have a persisted
// /sync cursor we want restart backlogs to replay just like other channels.
const dropPreStartupMessages = !client.hasPersistedSyncState();
const { getRoomInfo, getMemberDisplayName } = createMatrixRoomInfoResolver(client);
const { getRoomInfo, getMemberDisplayName, invalidateMemberDisplayName } =
createMatrixRoomInfoResolver(client);
const isExplicitlyConfiguredRoom = async (roomId: string): Promise<boolean> => {
const roomInfoForConfig = needsRoomAliasesForConfig
? await getRoomInfo(roomId, { includeAliases: true })
@@ -440,6 +441,7 @@ export async function monitorMatrixProvider(opts: MonitorMatrixOpts = {}): Promi
})
.catch(() => []),
directTracker,
invalidateMemberDisplayName,
logVerboseMessage,
warnedEncryptedRooms,
warnedCryptoMissingRooms,
extensions/matrix/src/matrix/monitor/room-info.test.ts
+31 -4
View File
@@ -173,9 +173,12 @@ describe("createMatrixRoomInfoResolver", () => {
});
});
it("caches fallback user IDs when member display-name lookups fail", async () => {
it("retries member display names after transient state lookup failures", async () => {
const client = createRoomStateClient(async () => {
throw new Error("member lookup failed");
if (client.getRoomStateEvent.mock.calls.length === 1) {
throw new Error("member lookup failed");
}
return { displayname: "Recovered Alice" };
});
const resolver = createMatrixRoomInfoResolver(client);
@@ -184,9 +187,33 @@ describe("createMatrixRoomInfoResolver", () => {
).resolves.toBe("@alice:example.org");
await expect(
resolver.getMemberDisplayName("!room:example.org", "@alice:example.org"),
).resolves.toBe("@alice:example.org");
).resolves.toBe("Recovered Alice");
expect(client.getRoomStateEvent).toHaveBeenCalledTimes(1);
expect(client.getRoomStateEvent).toHaveBeenCalledTimes(2);
});
it("refreshes only the observed member after a room membership state change", async () => {
let aliceDisplayName = "Original Alice";
const client = createRoomStateClient(async (_roomId, _eventType, stateKey) => ({
displayname: stateKey === "@alice:example.org" ? aliceDisplayName : "Bob",
}));
const resolver = createMatrixRoomInfoResolver(client);
await expect(
resolver.getMemberDisplayName("!room:example.org", "@alice:example.org"),
).resolves.toBe("Original Alice");
await resolver.getMemberDisplayName("!room:example.org", "@bob:example.org");
aliceDisplayName = "Renamed Alice";
resolver.invalidateMemberDisplayName("!room:example.org", "@alice:example.org");
await expect(
resolver.getMemberDisplayName("!room:example.org", "@alice:example.org"),
).resolves.toBe("Renamed Alice");
await expect(
resolver.getMemberDisplayName("!room:example.org", "@bob:example.org"),
).resolves.toBe("Bob");
expect(client.getRoomStateEvent).toHaveBeenCalledTimes(3);
});
it("bounds cached room and member entries", async () => {
extensions/matrix/src/matrix/monitor/room-info.ts
+13 -3
View File
@@ -97,18 +97,28 @@ export function createMatrixRoomInfoResolver(client: MatrixClient) {
if (memberDisplayNameCache.has(cacheKey)) {
return memberDisplayNameCache.get(cacheKey) ?? userId;
}
const memberState = await client
.getRoomStateEvent(roomId, "m.room.member", userId)
.catch(() => null);
let memberState: Record<string, unknown>;
try {
memberState = await client.getRoomStateEvent(roomId, "m.room.member", userId);
} catch {
// A transient homeserver failure is not authoritative room state; retry
// the next lookup instead of pinning the fallback user ID for the session.
return userId;
}
const displayName =
memberState && typeof memberState.displayname === "string" ? memberState.displayname : userId;
setBoundedMap(memberDisplayNameCache, cacheKey, displayName, MAX_MEMBER_DISPLAY_NAMES);
return displayName;
};
const invalidateMemberDisplayName = (roomId: string, userId: string): void => {
memberDisplayNameCache.delete(`${roomId}:${userId}`);
};
return {
getRoomAliases,
getRoomInfo,
getMemberDisplayName,
invalidateMemberDisplayName,
};
}
extensions/matrix/src/matrix/send/media.test.ts
+102
View File
@@ -0,0 +1,102 @@
import { describe, expect, it } from "vitest";
import { resolveMediaDurationMs } from "./media.js";
const AIFC_SAMPLE_RATE = 44_100;
const AIFC_PACKET_COUNT = 441;
function buildAiffChunk(type: string, content: Buffer): Buffer {
const header = Buffer.alloc(8);
header.write(type, 0, "ascii");
header.writeUInt32BE(content.length, 4);
return Buffer.concat([header, content, ...(content.length % 2 ? [Buffer.alloc(1)] : [])]);
}
function buildAifcFixture(
params: {
channels?: number;
codec?: string;
decodedFrameCount?: boolean;
malformedSoundData?: boolean;
} = {},
): Buffer {
const channels = params.channels ?? 1;
const codec = params.codec ?? "ima4";
const compressed = codec === "ima4";
const samplesPerPacket = compressed ? 64 : 1;
const bytesPerPacket = compressed ? channels * 34 : channels;
const declaredFrameCount =
params.decodedFrameCount || !compressed
? AIFC_PACKET_COUNT * samplesPerPacket
: AIFC_PACKET_COUNT;
const common = Buffer.alloc(22);
common.writeUInt16BE(channels, 0);
common.writeUInt32BE(declaredFrameCount, 2);
common.writeUInt16BE(compressed ? 0 : 8, 6);
common.writeUInt16BE(0x400e, 8);
common.writeUInt16BE(AIFC_SAMPLE_RATE, 10);
common.write(codec, 18, "ascii");
const sound = Buffer.alloc(
8 + AIFC_PACKET_COUNT * bytesPerPacket + (params.malformedSoundData ? 1 : 0),
);
const chunks = Buffer.concat([
// Odd-sized chunks exercise the IFF padding rule before locating COMM.
buildAiffChunk("JUNK", Buffer.from([0])),
buildAiffChunk("COMM", common),
buildAiffChunk("SSND", sound),
]);
const form = Buffer.alloc(12);
form.write("FORM", 0, "ascii");
form.writeUInt32BE(chunks.length + 4, 4);
form.write("AIFC", 8, "ascii");
return Buffer.concat([form, chunks]);
}
describe("resolveMediaDurationMs", () => {
it.each([1, 2])(
"corrects Apple IMA4 packet-count durations for %i channel(s)",
async (channels) => {
await expect(
resolveMediaDurationMs({
buffer: buildAifcFixture({ channels }),
contentType: "audio/aiff",
fileName: "clip.aifc",
kind: "audio",
}),
).resolves.toBe(640);
},
);
it("preserves IMA4 durations when COMM already contains decoded sample frames", async () => {
await expect(
resolveMediaDurationMs({
buffer: buildAifcFixture({ decodedFrameCount: true }),
contentType: "audio/aiff",
fileName: "clip.aifc",
kind: "audio",
}),
).resolves.toBe(640);
});
it.each(["alaw", "ulaw"])("preserves non-IMA4 AIFC %s durations", async (codec) => {
await expect(
resolveMediaDurationMs({
buffer: buildAifcFixture({ codec }),
contentType: "audio/aiff",
fileName: "clip.aifc",
kind: "audio",
}),
).resolves.toBe(10);
});
it("does not reinterpret malformed IMA4 sound data as complete packets", async () => {
await expect(
resolveMediaDurationMs({
buffer: buildAifcFixture({ malformedSoundData: true }),
contentType: "audio/aiff",
fileName: "clip.aifc",
kind: "audio",
}),
).resolves.toBe(10);
});
});
extensions/matrix/src/matrix/send/media.ts
+61 -1
View File
@@ -108,6 +108,64 @@ export function buildMediaContent(params: {
const THUMBNAIL_MAX_SIDE = 800;
const THUMBNAIL_QUALITY = 80;
const AIFC_IMA4_BYTES_PER_CHANNEL_PACKET = 34;
const AIFC_IMA4_FRAMES_PER_PACKET = 64;
function resolveAifcIma4DurationSeconds(buffer: Buffer, sampleRate?: number): number | undefined {
if (
!sampleRate ||
!Number.isFinite(sampleRate) ||
buffer.length < 12 ||
buffer.toString("ascii", 0, 4) !== "FORM" ||
buffer.toString("ascii", 8, 12) !== "AIFC"
) {
return undefined;
}
let channels: number | undefined;
let declaredFrameCount: number | undefined;
let soundDataBytes: number | undefined;
for (let offset = 12; offset + 8 <= buffer.length;) {
const chunkType = buffer.toString("ascii", offset, offset + 4);
const chunkSize = buffer.readUInt32BE(offset + 4);
const chunkStart = offset + 8;
if (chunkSize > buffer.length - chunkStart) {
return undefined;
}
if (chunkType === "COMM") {
if (chunkSize < 22 || buffer.toString("ascii", chunkStart + 18, chunkStart + 22) !== "ima4") {
return undefined;
}
channels = buffer.readUInt16BE(chunkStart);
declaredFrameCount = buffer.readUInt32BE(chunkStart + 2);
} else if (chunkType === "SSND") {
if (chunkSize < 8) {
return undefined;
}
const soundDataOffset = buffer.readUInt32BE(chunkStart);
if (soundDataOffset > chunkSize - 8) {
return undefined;
}
soundDataBytes = chunkSize - 8 - soundDataOffset;
}
offset = chunkStart + chunkSize + (chunkSize & 1);
}
if (!channels || !declaredFrameCount || !soundDataBytes) {
return undefined;
}
const packetSize = channels * AIFC_IMA4_BYTES_PER_CHANNEL_PACKET;
if (soundDataBytes % packetSize !== 0) {
return undefined;
}
const packetCount = soundDataBytes / packetSize;
if (packetCount !== declaredFrameCount) {
return undefined;
}
// Apple AIFC stores IMA4 packet count in COMM; each packet decodes to 64
// sample frames. Other encoders can store decoded frames, so verify SSND first.
return (packetCount * AIFC_IMA4_FRAMES_PER_PACKET) / sampleRate;
}
export async function prepareImageInfo(params: {
buffer: Buffer;
@@ -180,7 +238,9 @@ export async function resolveMediaDurationMs(params: {
duration: true,
skipCovers: true,
});
const durationSeconds = metadata.format.duration;
const durationSeconds =
resolveAifcIma4DurationSeconds(params.buffer, metadata.format.sampleRate) ??
metadata.format.duration;
if (typeof durationSeconds === "number" && Number.isFinite(durationSeconds)) {
return Math.max(0, Math.round(durationSeconds * 1000));
}