Fix issues around echo & redaction handling in threads (#2286)

spec/test-utils/test-utils.ts

@@ -8,6 +8,7 @@ import { logger } from '../../src/logger';
 import { IContent, IEvent, IUnsigned, MatrixEvent, MatrixEventEvent } from "../../src/models/event";
 import { ClientEvent, EventType, MatrixClient } from "../../src";
 import { SyncState } from "../../src/sync";
+import { eventMapperFor } from "../../src/event-mapper";
 
 /**
  * Return a promise that is resolved when the client next emits a
@@ -79,6 +80,7 @@ interface IEventOpts {
     redacts?: string;
 }
 
+let testEventIndex = 1; // counter for events, easier for comparison of randomly generated events
 /**
  * Create an Event.
  * @param {Object} opts Values for the event.
@@ -88,9 +90,10 @@ interface IEventOpts {
  * @param {string} opts.skey Optional. The state key (auto inserts empty string)
  * @param {Object} opts.content The event.content
  * @param {boolean} opts.event True to make a MatrixEvent.
+ * @param {MatrixClient} client If passed along with opts.event=true will be used to set up re-emitters.
  * @return {Object} a JSON object representing this event.
  */
-export function mkEvent(opts: IEventOpts): object | MatrixEvent {
+export function mkEvent(opts: IEventOpts, client?: MatrixClient): object | MatrixEvent {
     if (!opts.type || !opts.content) {
         throw new Error("Missing .type or .content =>" + JSON.stringify(opts));
     }
@@ -100,7 +103,7 @@ export function mkEvent(opts: IEventOpts): object | MatrixEvent {
         sender: opts.sender || opts.user, // opts.user for backwards-compat
         content: opts.content,
         unsigned: opts.unsigned || {},
-        event_id: "$" + Math.random() + "-" + Math.random(),
+        event_id: "$" + testEventIndex++ + "-" + Math.random() + "-" + Math.random(),
         txn_id: "~" + Math.random(),
         redacts: opts.redacts,
     };
@@ -117,6 +120,11 @@ export function mkEvent(opts: IEventOpts): object | MatrixEvent {
     ].includes(opts.type)) {
         event.state_key = "";
     }
+
+    if (opts.event && client) {
+        return eventMapperFor(client, {})(event);
+    }
+
     return opts.event ? new MatrixEvent(event) : event;
 }
 
@@ -209,9 +217,10 @@ interface IMessageOpts {
  * @param {string} opts.user The user ID for the event.
  * @param {string} opts.msg Optional. The content.body for the event.
  * @param {boolean} opts.event True to make a MatrixEvent.
+ * @param {MatrixClient} client If passed along with opts.event=true will be used to set up re-emitters.
  * @return {Object|MatrixEvent} The event
  */
-export function mkMessage(opts: IMessageOpts): object | MatrixEvent {
+export function mkMessage(opts: IMessageOpts, client?: MatrixClient): object | MatrixEvent {
     const eventOpts: IEventOpts = {
         ...opts,
         type: EventType.RoomMessage,
@@ -224,7 +233,7 @@ export function mkMessage(opts: IMessageOpts): object | MatrixEvent {
     if (!eventOpts.content.body) {
         eventOpts.content.body = "Random->" + Math.random();
     }
-    return mkEvent(eventOpts);
+    return mkEvent(eventOpts, client);
 }
 
 /**

spec/unit/room.spec.ts

@@ -50,7 +50,7 @@ describe("Room", function() {
         event: true,
         user: userA,
         room: roomId,
-    }) as MatrixEvent;
+    }, room.client) as MatrixEvent;
 
     const mkReply = (target: MatrixEvent) => utils.mkEvent({
         event: true,
@@ -65,7 +65,7 @@ describe("Room", function() {
                 },
             },
         },
-    }) as MatrixEvent;
+    }, room.client) as MatrixEvent;
 
     const mkEdit = (target: MatrixEvent, salt = Math.random()) => utils.mkEvent({
         event: true,
@@ -82,7 +82,7 @@ describe("Room", function() {
                 event_id: target.getId(),
             },
         },
-    }) as MatrixEvent;
+    }, room.client) as MatrixEvent;
 
     const mkThreadResponse = (root: MatrixEvent) => utils.mkEvent({
         event: true,
@@ -99,7 +99,7 @@ describe("Room", function() {
                 "rel_type": "m.thread",
             },
         },
-    }) as MatrixEvent;
+    }, room.client) as MatrixEvent;
 
     const mkReaction = (target: MatrixEvent) => utils.mkEvent({
         event: true,
@@ -113,7 +113,7 @@ describe("Room", function() {
                 "key": Math.random().toString(),
             },
         },
-    }) as MatrixEvent;
+    }, room.client) as MatrixEvent;
 
     const mkRedaction = (target: MatrixEvent) => utils.mkEvent({
         event: true,
@@ -122,7 +122,7 @@ describe("Room", function() {
         room: roomId,
         redacts: target.getId(),
         content: {},
-    }) as MatrixEvent;
+    }, room.client) as MatrixEvent;
 
     beforeEach(function() {
         room = new Room(roomId, new TestClient(userA, "device").client, userA);
@@ -1899,6 +1899,7 @@ describe("Room", function() {
                 "@alice:example.com", "alicedevice",
             )).client;
             room = new Room(roomId, client, userA);
+            client.getRoom = () => room;
         });
 
         it("allow create threads without a root event", function() {
@@ -1938,11 +1939,7 @@ describe("Room", function() {
         });
 
         it("Edits update the lastReply event", async () => {
-            const client = (new TestClient(
-                "@alice:example.com", "alicedevice",
-            )).client;
-            client.supportsExperimentalThreads = () => true;
-            room = new Room(roomId, client, userA);
+            room.client.supportsExperimentalThreads = () => true;
 
             const randomMessage = mkMessage();
             const threadRoot = mkMessage();
@@ -1951,7 +1948,7 @@ describe("Room", function() {
             const threadResponseEdit = mkEdit(threadResponse);
             threadResponseEdit.localTimestamp += 2000;
 
-            client.fetchRoomEvent = (eventId: string) => Promise.resolve({
+            room.client.fetchRoomEvent = (eventId: string) => Promise.resolve({
                 ...threadRoot.event,
                 unsigned: {
                     "age": 123,
@@ -1975,6 +1972,121 @@ describe("Room", function() {
             await emitPromise(thread, ThreadEvent.Update);
             expect(thread.replyToEvent.getContent().body).toBe(threadResponseEdit.getContent()["m.new_content"].body);
         });
+
+        it("Redactions to thread responses decrement the length", async () => {
+            room.client.supportsExperimentalThreads = () => true;
+
+            const threadRoot = mkMessage();
+            const threadResponse1 = mkThreadResponse(threadRoot);
+            threadResponse1.localTimestamp += 1000;
+            const threadResponse2 = mkThreadResponse(threadRoot);
+            threadResponse2.localTimestamp += 2000;
+
+            room.client.fetchRoomEvent = (eventId: string) => Promise.resolve({
+                ...threadRoot.event,
+                unsigned: {
+                    "age": 123,
+                    "m.relations": {
+                        "m.thread": {
+                            latest_event: threadResponse2.event,
+                            count: 2,
+                            current_user_participated: true,
+                        },
+                    },
+                },
+            });
+
+            room.addLiveEvents([threadRoot, threadResponse1, threadResponse2]);
+            const thread = await emitPromise(room, ThreadEvent.New);
+
+            expect(thread).toHaveLength(2);
+            expect(thread.replyToEvent.getId()).toBe(threadResponse2.getId());
+
+            const threadResponse1Redaction = mkRedaction(threadResponse1);
+            room.addLiveEvents([threadResponse1Redaction]);
+            await emitPromise(thread, ThreadEvent.Update);
+            expect(thread).toHaveLength(1);
+            expect(thread.replyToEvent.getId()).toBe(threadResponse2.getId());
+        });
+
+        it("Redactions to reactions in threads do not decrement the length", async () => {
+            room.client.supportsExperimentalThreads = () => true;
+
+            const threadRoot = mkMessage();
+            const threadResponse1 = mkThreadResponse(threadRoot);
+            threadResponse1.localTimestamp += 1000;
+            const threadResponse2 = mkThreadResponse(threadRoot);
+            threadResponse2.localTimestamp += 2000;
+            const threadResponse2Reaction = mkReaction(threadResponse2);
+
+            room.client.fetchRoomEvent = (eventId: string) => Promise.resolve({
+                ...threadRoot.event,
+                unsigned: {
+                    "age": 123,
+                    "m.relations": {
+                        "m.thread": {
+                            latest_event: threadResponse2.event,
+                            count: 2,
+                            current_user_participated: true,
+                        },
+                    },
+                },
+            });
+
+            room.addLiveEvents([threadRoot, threadResponse1, threadResponse2, threadResponse2Reaction]);
+            const thread = await emitPromise(room, ThreadEvent.New);
+
+            expect(thread).toHaveLength(2);
+            expect(thread.replyToEvent.getId()).toBe(threadResponse2.getId());
+
+            const threadResponse2ReactionRedaction = mkRedaction(threadResponse2Reaction);
+            room.addLiveEvents([threadResponse2ReactionRedaction]);
+            await emitPromise(thread, ThreadEvent.Update);
+            expect(thread).toHaveLength(2);
+            expect(thread.replyToEvent.getId()).toBe(threadResponse2.getId());
+        });
+
+        it("Redacting the lastEvent finds a new lastEvent", async () => {
+            room.client.supportsExperimentalThreads = () => true;
+
+            const threadRoot = mkMessage();
+            const threadResponse1 = mkThreadResponse(threadRoot);
+            threadResponse1.localTimestamp += 1000;
+            const threadResponse2 = mkThreadResponse(threadRoot);
+            threadResponse2.localTimestamp += 2000;
+
+            room.client.fetchRoomEvent = (eventId: string) => Promise.resolve({
+                ...threadRoot.event,
+                unsigned: {
+                    "age": 123,
+                    "m.relations": {
+                        "m.thread": {
+                            latest_event: threadResponse2.event,
+                            count: 2,
+                            current_user_participated: true,
+                        },
+                    },
+                },
+            });
+
+            room.addLiveEvents([threadRoot, threadResponse1, threadResponse2]);
+            const thread = await emitPromise(room, ThreadEvent.New);
+
+            expect(thread).toHaveLength(2);
+            expect(thread.replyToEvent.getId()).toBe(threadResponse2.getId());
+
+            const threadResponse2Redaction = mkRedaction(threadResponse2);
+            room.addLiveEvents([threadResponse2Redaction]);
+            await emitPromise(thread, ThreadEvent.Update);
+            expect(thread).toHaveLength(1);
+            expect(thread.replyToEvent.getId()).toBe(threadResponse1.getId());
+
+            const threadResponse1Redaction = mkRedaction(threadResponse1);
+            room.addLiveEvents([threadResponse1Redaction]);
+            await emitPromise(thread, ThreadEvent.Update);
+            expect(thread).toHaveLength(0);
+            expect(thread.replyToEvent.getId()).toBe(threadRoot.getId());
+        });
     });
 
     describe("eventShouldLiveIn", () => {

src/event-mapper.ts

@@ -66,6 +66,9 @@ export function eventMapperFor(client: MatrixClient, options: MapperOpts): Event
                 MatrixEventEvent.Replaced,
                 MatrixEventEvent.VisibilityChange,
             ]);
+            room?.reEmitter.reEmit(event, [
+                MatrixEventEvent.BeforeRedaction,
+            ]);
         }
         return event;
     }

src/models/room.ts

@@ -23,7 +23,7 @@ import { Direction, EventTimeline } from "./event-timeline";
 import { getHttpUriForMxc } from "../content-repo";
 import * as utils from "../utils";
 import { defer, normalize } from "../utils";
-import { IEvent, IThreadBundledRelationship, MatrixEvent } from "./event";
+import { IEvent, IThreadBundledRelationship, MatrixEvent, MatrixEventEvent, MatrixEventHandlerMap } from "./event";
 import { EventStatus } from "./event-status";
 import { RoomMember } from "./room-member";
 import { IRoomSummary, RoomSummary } from "./room-summary";
@@ -171,7 +171,8 @@ type EmittedEvents = RoomEvent
     | ThreadEvent.Update
     | ThreadEvent.NewReply
     | RoomEvent.Timeline
-    | RoomEvent.TimelineReset;
+    | RoomEvent.TimelineReset
+    | MatrixEventEvent.BeforeRedaction;
 
 export type RoomEventHandlerMap = {
     [RoomEvent.MyMembership]: (room: Room, membership: string, prevMembership?: string) => void;
@@ -188,10 +189,10 @@ export type RoomEventHandlerMap = {
         oldStatus?: EventStatus,
     ) => void;
     [ThreadEvent.New]: (thread: Thread, toStartOfTimeline: boolean) => void;
-} & ThreadHandlerMap;
+} & ThreadHandlerMap & MatrixEventHandlerMap;
 
 export class Room extends TypedEventEmitter<EmittedEvents, RoomEventHandlerMap> {
-    private readonly reEmitter: TypedReEmitter<EmittedEvents, RoomEventHandlerMap>;
+    public readonly reEmitter: TypedReEmitter<EmittedEvents, RoomEventHandlerMap>;
     private txnToEvent: Record<string, MatrixEvent> = {}; // Pending in-flight requests { string: MatrixEvent }
     // receipts should clobber based on receipt_type and user_id pairs hence
     // the form of this structure. This is sub-optimal for the exposed APIs
@@ -383,7 +384,7 @@ export class Room extends TypedEventEmitter<EmittedEvents, RoomEventHandlerMap>
             return this.threadTimelineSetsPromise;
         }
 
-        if (this.client?.supportsExperimentalThreads) {
+        if (this.client?.supportsExperimentalThreads()) {
             try {
                 this.threadTimelineSetsPromise = Promise.all([
                     this.createThreadTimelineSet(),
@@ -1676,6 +1677,8 @@ export class Room extends TypedEventEmitter<EmittedEvents, RoomEventHandlerMap>
             thread = await this.threadPromises.get(threadId);
         }
 
+        events = events.filter(e => e.getId() !== threadId); // filter out any root events
+
         if (thread) {
             for (const event of events) {
                 await thread.addEvent(event, toStartOfTimeline);
@@ -1810,7 +1813,7 @@ export class Room extends TypedEventEmitter<EmittedEvents, RoomEventHandlerMap>
         }
     };
 
-    private processLiveEvent(event: MatrixEvent): Promise<void> {
+    private processLiveEvent(event: MatrixEvent): void {
         this.applyRedaction(event);
 
         // Implement MSC3531: hiding messages.
@@ -1827,7 +1830,6 @@ export class Room extends TypedEventEmitter<EmittedEvents, RoomEventHandlerMap>
             if (existingEvent) {
                 // remote echo of an event we sent earlier
                 this.handleRemoteEcho(event, existingEvent);
-                return;
             }
         }
     }

src/models/thread.ts

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-import { MatrixClient, RelationType, RoomEvent } from "../matrix";
+import { MatrixClient, MatrixEventEvent, RelationType, RoomEvent } from "../matrix";
 import { TypedReEmitter } from "../ReEmitter";
 import { IRelationsRequestOpts } from "../@types/requests";
 import { IThreadBundledRelationship, MatrixEvent } from "./event";
@@ -94,6 +94,7 @@ export class Thread extends TypedEventEmitter<EmittedEvents, EventHandlerMap> {
             RoomEvent.TimelineReset,
         ]);
 
+        this.room.on(MatrixEventEvent.BeforeRedaction, this.onBeforeRedaction);
         this.room.on(RoomEvent.LocalEchoUpdated, this.onEcho);
         this.timelineSet.on(RoomEvent.Timeline, this.onEcho);
 
@@ -114,7 +115,29 @@ export class Thread extends TypedEventEmitter<EmittedEvents, EventHandlerMap> {
         }
     }
 
+    private onBeforeRedaction = (event: MatrixEvent) => {
+        if (event?.isRelation(THREAD_RELATION_TYPE.name) &&
+            this.room.eventShouldLiveIn(event).threadId === this.id
+        ) {
+            this.replyCount--;
+            this.emit(ThreadEvent.Update, this);
+        }
+
+        if (this.lastEvent?.getId() === event.getId()) {
+            const events = [...this.timelineSet.getLiveTimeline().getEvents()].reverse();
+            this.lastEvent = events.find(e => (
+                !e.isRedacted() &&
+                e.getId() !== event.getId() &&
+                e.isRelation(THREAD_RELATION_TYPE.name)
+            )) ?? this.rootEvent;
+            this.emit(ThreadEvent.NewReply, this, this.lastEvent);
+        }
+    };
+
     private onEcho = (event: MatrixEvent) => {
+        if (event.threadRootId !== this.id) return; // ignore echoes for other timelines
+        if (this.lastEvent === event) return;
+
         // There is a risk that the `localTimestamp` approximation will not be accurate
         // when threads are used over federation. That could result in the reply
         // count value drifting away from the value returned by the server
@@ -135,9 +158,7 @@ export class Thread extends TypedEventEmitter<EmittedEvents, EventHandlerMap> {
             }
         }
 
-        if (this.timelineSet.eventIdToTimeline(event.getId())) {
-            this.emit(ThreadEvent.Update, this);
-        }
+        this.emit(ThreadEvent.Update, this);
     };
 
     public get roomState(): RoomState {
@@ -188,10 +209,9 @@ export class Thread extends TypedEventEmitter<EmittedEvents, EventHandlerMap> {
             this._currentUserParticipated = true;
         }
 
-        const isThreadReply = event.isRelation(THREAD_RELATION_TYPE.name);
         // If no thread support exists we want to count all thread relation
         // added as a reply. We can't rely on the bundled relationships count
-        if (!Thread.hasServerSideSupport && isThreadReply) {
+        if (!Thread.hasServerSideSupport && event.isRelation(THREAD_RELATION_TYPE.name)) {
             this.replyCount++;
         }