fix: add workaround for missing first message #208
Conversation
WalkthroughAdds a pending-invite workflow: new pendingInvite flag on staged events, a PendingInviteRepository with DI wiring, updates to event staging queries and creation, EventService logic to mark/check/unmark pending invites, InviteService to add pending invites when not hosting, and RoomService to clear them on successful joins. Changes
Sequence Diagram(s)sequenceDiagram
autonumber
actor Ext as Remote HS
participant IS as InviteService
participant ES as EventService
participant PIR as PendingInviteRepo
participant ESR as EventStagingRepo
Ext->>IS: Incoming invite (not host)
IS->>ES: addPendingInvite(inviteEvent)
ES->>PIR: add(eventId, event)
ES->>ESR: create(eventId, origin, event, pendingInvite=true)
note over ESR: Event staged with pendingInvite=true
sequenceDiagram
autonumber
participant Proc as EventService.process
participant PIR as PendingInviteRepo
participant ESR as EventStagingRepo
participant Q as RoomQueue
Proc->>PIR: isSenderInvitePending(sender, roomId)?
PIR-->>Proc: true/false
alt Pending invite exists
Proc->>ESR: create(..., pendingInvite=true)
Proc-->>Proc: Skip further processing
else No pending invite
Proc->>ESR: create(..., pendingInvite=false)
Proc->>Q: Enqueue room for processing
end
sequenceDiagram
autonumber
participant RS as RoomService.joinUser
participant ES as EventService
participant PIR as PendingInviteRepo
participant ESR as EventStagingRepo
participant L as RoomLock
RS->>RS: Persist final join event
RS->>ES: removePendingInvite(eventId, roomId)
ES->>PIR: remove(eventId)
ES->>ESR: unmarkInvitePending(eventId)
ES->>L: Try acquire room lock and enqueue
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~30 minutes Possibly related PRs
Suggested reviewers
Poem
Pre-merge checks and finishing touches❌ Failed checks (2 warnings)
✅ Passed checks (1 passed)
✨ Finishing touches
🧪 Generate unit tests
Comment |
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #208 +/- ##
=======================================
Coverage 81.01% 81.01%
=======================================
Files 63 63
Lines 4692 4692
=======================================
Hits 3801 3801
Misses 891 891 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 7
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
packages/federation-sdk/src/repositories/event-staging.repository.ts (1)
49-55: Legacy docs hidden: treat missing pendingInvite as false.Filtering strictly on
pendingInvite: falseexcludes older staged events (no field), stalling processing.Apply:
return this.collection.findOne( { - roomId, - pendingInvite: false, + roomId, + $or: [{ pendingInvite: false }, { pendingInvite: { $exists: false } }], }, { sort: { createdAt: 1 }, }, );
🧹 Nitpick comments (9)
packages/federation-sdk/src/services/room.service.ts (1)
806-814: Also clear pending invite on local-join path.After persisting a local join, mirror the same cleanup so locally hosted joins also unblock staged events.
packages/federation-sdk/src/services/invite.service.ts (2)
68-69: Avoid broad type cast if possible.
roomInformation.room_version as RoomVersionis okay but brittle. Prefer returning aRoomVersionfromgetRoomInformationor narrowing via a guard.
111-112: Same as above: minimize casts.Consider narrowing earlier to avoid repeated
as RoomVersion.packages/federation-sdk/src/repositories/event-staging.repository.ts (2)
13-14: Index for new query shape.Add a compound index to support the new filter.
) { - this.collection.createIndex({ roomId: 1, createdAt: 1 }); + this.collection.createIndex({ roomId: 1, createdAt: 1 }); + this.collection.createIndex({ roomId: 1, pendingInvite: 1, createdAt: 1 }); }
16-43: Optional: relax origin in upsert filter.Using
{ _id, origin }can cause duplicate-key errors if the same event is seen from another origin. Filtering only by_idis safer; keeporiginin$setOnInsert.- return this.collection.updateOne( - { - _id: eventId, - origin, - }, + return this.collection.updateOne( + { _id: eventId }, { $setOnInsert: { roomId: event.room_id, createdAt: new Date(), + origin, }, $set: { event, pendingInvite, }, }, { upsert: true, }, );packages/federation-sdk/src/container.ts (1)
111-116: Collection binding LGTM. Consider TTL via repo to avoid growth.packages/federation-sdk/src/repositories/pending-invite.repository.ts (3)
11-16: Create indexes (lookup + TTL).Speed up lookups and auto-expire stale invites.
export class PendingInviteRepository { constructor( @inject('PendingInviteCollection') private readonly collection: Collection<PendingInvite>, - ) {} + ) { + // Lookup index + void this.collection.createIndex({ + 'event.room_id': 1, + 'event.state_key': 1, + 'event.type': 1, + }); + // TTL (14 days) to avoid unbounded growth if never resolved + void this.collection.createIndex({ createdAt: 1 }, { expireAfterSeconds: 14 * 24 * 60 * 60 }); + }
26-35: Tighten query to membership=invite.Be explicit so future non-invite member events don’t match.
return this.collection.findOne({ 'event.type': 'm.room.member', + 'event.content.membership': 'invite', 'event.state_key': userId, 'event.room_id': roomId, });
37-39: Optional convenience: remove by user+room.Helps the service avoid an extra read.
async remove(eventId: EventID): Promise<void> { await this.collection.deleteOne({ _id: eventId }); } + + async removeByUserIdAndRoomId(userId: string, roomId: string): Promise<void> { + await this.collection.deleteOne({ + 'event.type': 'm.room.member', + 'event.content.membership': 'invite', + 'event.state_key': userId, + 'event.room_id': roomId, + }); + }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
Disabled knowledge base sources:
- Jira integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (7)
packages/core/src/models/event.model.ts(1 hunks)packages/federation-sdk/src/container.ts(3 hunks)packages/federation-sdk/src/repositories/event-staging.repository.ts(4 hunks)packages/federation-sdk/src/repositories/pending-invite.repository.ts(1 hunks)packages/federation-sdk/src/services/event.service.ts(5 hunks)packages/federation-sdk/src/services/invite.service.ts(4 hunks)packages/federation-sdk/src/services/room.service.ts(1 hunks)
🧰 Additional context used
🧬 Code graph analysis (4)
packages/federation-sdk/src/repositories/pending-invite.repository.ts (2)
packages/room/src/types/v3-11.ts (1)
Pdu(729-729)packages/room/src/types/_common.ts (1)
EventID(8-8)
packages/federation-sdk/src/container.ts (1)
packages/federation-sdk/src/repositories/pending-invite.repository.ts (1)
PendingInvite(5-9)
packages/federation-sdk/src/services/event.service.ts (4)
packages/room/src/manager/event-wrapper.ts (4)
event(102-111)origin(82-88)roomId(72-74)sender(76-78)packages/room/src/manager/v3.ts (1)
eventId(22-34)packages/room/src/types/v3-11.ts (1)
Pdu(729-729)packages/room/src/types/_common.ts (1)
EventID(8-8)
packages/federation-sdk/src/services/invite.service.ts (1)
packages/room/src/manager/type.ts (1)
RoomVersion(14-14)
🔇 Additional comments (9)
packages/federation-sdk/src/services/invite.service.ts (1)
2-2: Import consolidation LGTM.packages/federation-sdk/src/repositories/event-staging.repository.ts (2)
20-21: Extending create() with pendingInvite is fine.
35-37: Ensure pendingInvite is always written on upsert.Good that
$setwrites it even on re-receive. Keep this to allow clearing later.packages/federation-sdk/src/container.ts (2)
20-23: Type registration for PendingInvite repo LGTM.
130-130: Approve — DI wiring LGTM; no directnew EventService(usages found. EventService is registered via container.registerSingleton in packages/federation-sdk/src/container.ts:131 and consumers use container.resolve.packages/federation-sdk/src/services/event.service.ts (4)
25-26: Adding PersistentEventBase import LGTM.
36-66: Constructor DI extended with PendingInviteRepository LGTM.
828-831: addPendingInvite LGTM.No further comments.
852-859: isSenderInvitePending LGTM.Consider inlining membership=invite filter in repo (proposed).
|
|
||
| pendingInvite: boolean; | ||
| } |
There was a problem hiding this comment.
Backfill or tolerant query needed for new required field.
Adding pendingInvite: boolean makes legacy staged docs (without this field) invisible to queries that match pendingInvite: false. Ensure downstream queries treat missing as false or run a one-off backfill/migration.
I’ve proposed tolerant query changes under event-staging.repository.ts.
🤖 Prompt for AI Agents
In packages/core/src/models/event.model.ts around lines 29 to 31, adding the new
required field `pendingInvite: boolean` will cause legacy documents that lack
this property to be excluded by queries that expect `pendingInvite: false`; to
fix, make the model tolerate missing values and/or backfill existing docs:
update the TypeScript model to allow the field to be optional (or provide a
default), add tolerant query logic in repositories (e.g., treat missing as false
via an `$or`/`$exists` check or include both `false` and `null`/missing in
filters), and run a one‑time migration to set `pendingInvite: false` on staged
documents so future queries can be simple and non-tolerant.
| async unmarkInvitePending(eventId: EventID): Promise<UpdateResult> { | ||
| return this.collection.updateOne( | ||
| { _id: eventId }, | ||
| { $set: { pendingInvite: false } }, | ||
| ); | ||
| } |
There was a problem hiding this comment.
Unmarking by eventId is insufficient; add a bulk clear by room+sender.
Multiple PDUs may be flagged. Provide an updateMany helper.
async unmarkInvitePending(eventId: EventID): Promise<UpdateResult> {
return this.collection.updateOne(
{ _id: eventId },
{ $set: { pendingInvite: false } },
);
}
+
+ async unmarkInvitePendingForRoomAndSender(
+ roomId: string,
+ sender: string,
+ ): Promise<UpdateResult> {
+ return this.collection.updateMany(
+ { roomId, 'event.sender': sender, pendingInvite: true },
+ { $set: { pendingInvite: false } },
+ );
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| async unmarkInvitePending(eventId: EventID): Promise<UpdateResult> { | |
| return this.collection.updateOne( | |
| { _id: eventId }, | |
| { $set: { pendingInvite: false } }, | |
| ); | |
| } | |
| async unmarkInvitePending(eventId: EventID): Promise<UpdateResult> { | |
| return this.collection.updateOne( | |
| { _id: eventId }, | |
| { $set: { pendingInvite: false } }, | |
| ); | |
| } | |
| async unmarkInvitePendingForRoomAndSender( | |
| roomId: string, | |
| sender: string, | |
| ): Promise<UpdateResult> { | |
| return this.collection.updateMany( | |
| { roomId, 'event.sender': sender, pendingInvite: true }, | |
| { $set: { pendingInvite: false } }, | |
| ); | |
| } |
🤖 Prompt for AI Agents
In packages/federation-sdk/src/repositories/event-staging.repository.ts around
lines 65 to 70, the current unmarkInvitePending method only updates a single
document by eventId which misses other PDUs flagged for the same invite; add a
new public method (e.g., unmarkInvitePendingByRoomAndSender or
unmarkPendingInvites) that calls collection.updateMany with a filter for roomId
and sender (and any other identifying fields like stateKey if applicable) and
sets pendingInvite to false, return the UpdateResult; keep existing updateOne
method if needed but use updateMany to clear all matching PDUs.
| async add(eventId: EventID, event: Pdu): Promise<void> { | ||
| await this.collection.insertOne({ | ||
| _id: eventId, | ||
| event, | ||
| createdAt: new Date(), | ||
| }); | ||
| } |
There was a problem hiding this comment.
Make add idempotent.
Use upsert to tolerate retries.
- async add(eventId: EventID, event: Pdu): Promise<void> {
- await this.collection.insertOne({
- _id: eventId,
- event,
- createdAt: new Date(),
- });
- }
+ async add(eventId: EventID, event: Pdu): Promise<void> {
+ await this.collection.updateOne(
+ { _id: eventId },
+ { $setOnInsert: { _id: eventId, event, createdAt: new Date() } },
+ { upsert: true },
+ );
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| async add(eventId: EventID, event: Pdu): Promise<void> { | |
| await this.collection.insertOne({ | |
| _id: eventId, | |
| event, | |
| createdAt: new Date(), | |
| }); | |
| } | |
| async add(eventId: EventID, event: Pdu): Promise<void> { | |
| await this.collection.updateOne( | |
| { _id: eventId }, | |
| { $setOnInsert: { _id: eventId, event, createdAt: new Date() } }, | |
| { upsert: true }, | |
| ); | |
| } |
🤖 Prompt for AI Agents
In packages/federation-sdk/src/repositories/pending-invite.repository.ts around
lines 18 to 24, the add method currently does an insertOne which will fail on
retries; make it idempotent by replacing insertOne with an updateOne (or
replace/upsert) using upsert: true and $setOnInsert for the event and createdAt
so repeated calls don’t error or overwrite existing records (e.g.,
updateOne({_id: eventId}, {$setOnInsert: {event, createdAt: new Date()}},
{upsert: true})).
| const pendingInvite = await this.isSenderInvitePending( | ||
| event.sender, | ||
| event.room_id, | ||
| ); | ||
|
|
||
| // save the event as staged to be processed | ||
| await this.eventStagingRepository.create(eventId, origin, event); | ||
| await this.eventStagingRepository.create( | ||
| eventId, | ||
| origin, | ||
| event, | ||
| pendingInvite, | ||
| ); | ||
|
|
||
| if (pendingInvite) { | ||
| continue; | ||
| } | ||
|
|
There was a problem hiding this comment.
🧩 Analysis chain
Pending-invite short-circuit: confirm join events aren’t blocked.
This will stage and skip any PDU from a sender with a pending invite. Ensure membership “join” events aren’t accidentally deferred, or they may never clear the block.
Run a quick scan to see if join PDUs enter this path:
🏁 Script executed:
#!/bin/bash
rg -nP --type=ts -C3 "processIncomingPDUs|persistStateEvent\(.+m\.room\.member.+join" packagesLength of output: 1134
🏁 Script executed:
#!/bin/bash
rg -n -C30 "isSenderInvitePending" packages/federation-sdk/src/services/event.service.tsLength of output: 2991
Add join-event exemption to pending-invite short-circuit
In processIncomingPDUs (packages/federation-sdk/src/services/event.service.ts lines 203–219), you currently skip all PDUs when pendingInvite is true. That’ll also defer membership-“join” events and prevent removePendingInvite from ever running. Wrap the continue in a check so joins bypass the block, e.g.:
const isJoin = event.type === 'm.room.member'
&& (event.content as any).membership === 'join';
if (pendingInvite && !isJoin) {
continue;
}This ensures join events are processed immediately.
🤖 Prompt for AI Agents
In packages/federation-sdk/src/services/event.service.ts around lines 203-219,
the code short-circuits all PDUs when pendingInvite is true which also skips
processing of membership "join" events and prevents removePendingInvite from
running; modify the conditional so that you only continue when pendingInvite is
true AND the event is not a membership join — i.e., detect isJoin by checking
event.type === 'm.room.member' and event.content.membership === 'join' (or
equivalent) and only call continue when pendingInvite && !isJoin so join events
bypass the short-circuit and are processed immediately.
| async removePendingInvite(eventId: EventID, roomId: string): Promise<void> { | ||
| await Promise.all([ | ||
| this.pendingInviteRepository.remove(eventId), | ||
| this.eventStagingRepository.unmarkInvitePending(eventId), | ||
| ]); | ||
|
|
||
| // acquire a lock for processing the event | ||
| const lock = await this.lockRepository.getLock( | ||
| roomId, | ||
| this.configService.instanceId, | ||
| ); | ||
| if (!lock) { | ||
| this.logger.debug(`Couldn't acquire a lock for room ${roomId}`); | ||
| return; | ||
| } | ||
|
|
||
| // TODO change this to call stagingAreaService directly | ||
| this.stagingAreaQueue.enqueue(roomId); | ||
| } |
There was a problem hiding this comment.
Fix: remove by user+room and unmark all staged events.
Current implementation removes by eventId and unmarks a single staged doc, which doesn’t map to how invites and staged PDUs are keyed.
- async removePendingInvite(eventId: EventID, roomId: string): Promise<void> {
- await Promise.all([
- this.pendingInviteRepository.remove(eventId),
- this.eventStagingRepository.unmarkInvitePending(eventId),
- ]);
-
- // acquire a lock for processing the event
- const lock = await this.lockRepository.getLock(
- roomId,
- this.configService.instanceId,
- );
- if (!lock) {
- this.logger.debug(`Couldn't acquire a lock for room ${roomId}`);
- return;
- }
-
- // TODO change this to call stagingAreaService directly
- this.stagingAreaQueue.enqueue(roomId);
- }
+ async resolvePendingInvitesForUser(roomId: string, userId: string): Promise<void> {
+ // Remove the invite entry (by user+room) if present
+ const invite = await this.pendingInviteRepository.findByUserIdAndRoomId(userId, roomId);
+ if (invite?._id) {
+ await this.pendingInviteRepository.remove(invite._id);
+ }
+ // Unblock all staged PDUs from this user in this room
+ await this.eventStagingRepository.unmarkInvitePendingForRoomAndSender(roomId, userId);
+
+ const lock = await this.lockRepository.getLock(roomId, this.configService.instanceId);
+ if (!lock) {
+ this.logger.debug(`Couldn't acquire a lock for room ${roomId}`);
+ return;
+ }
+ this.stagingAreaQueue.enqueue(roomId);
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| async removePendingInvite(eventId: EventID, roomId: string): Promise<void> { | |
| await Promise.all([ | |
| this.pendingInviteRepository.remove(eventId), | |
| this.eventStagingRepository.unmarkInvitePending(eventId), | |
| ]); | |
| // acquire a lock for processing the event | |
| const lock = await this.lockRepository.getLock( | |
| roomId, | |
| this.configService.instanceId, | |
| ); | |
| if (!lock) { | |
| this.logger.debug(`Couldn't acquire a lock for room ${roomId}`); | |
| return; | |
| } | |
| // TODO change this to call stagingAreaService directly | |
| this.stagingAreaQueue.enqueue(roomId); | |
| } | |
| async resolvePendingInvitesForUser(roomId: string, userId: string): Promise<void> { | |
| // Remove the invite entry (by user+room) if present | |
| const invite = await this.pendingInviteRepository.findByUserIdAndRoomId(userId, roomId); | |
| if (invite?._id) { | |
| await this.pendingInviteRepository.remove(invite._id); | |
| } | |
| // Unblock all staged PDUs from this user in this room | |
| await this.eventStagingRepository.unmarkInvitePendingForRoomAndSender(roomId, userId); | |
| const lock = await this.lockRepository.getLock(roomId, this.configService.instanceId); | |
| if (!lock) { | |
| this.logger.debug(`Couldn't acquire a lock for room ${roomId}`); | |
| return; | |
| } | |
| this.stagingAreaQueue.enqueue(roomId); | |
| } |
| // nor are we part of the room now. | ||
|
|
||
| await this.eventService.addPendingInvite(inviteEvent); | ||
|
|
There was a problem hiding this comment.
Idempotency: guard against duplicate inserts.
addPendingInvite ultimately does insertOne(_id=inviteEventId). Retries can raise E11000. Use upsert in the repo or catch/ignore duplicate key errors.
Apply in repository (see pending-invite.repository.ts comment) or wrap call:
- await this.eventService.addPendingInvite(inviteEvent);
+ try {
+ await this.eventService.addPendingInvite(inviteEvent);
+ } catch (e: any) {
+ if (!/E11000/.test(String(e?.code) + String(e?.message))) throw e;
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| // nor are we part of the room now. | |
| await this.eventService.addPendingInvite(inviteEvent); | |
| // nor are we part of the room now. | |
| try { | |
| await this.eventService.addPendingInvite(inviteEvent); | |
| } catch (e: any) { | |
| if (!/E11000/.test(String(e?.code) + String(e?.message))) throw e; | |
| } |
🤖 Prompt for AI Agents
In packages/federation-sdk/src/services/invite.service.ts around lines 183 to
186, the call to this.eventService.addPendingInvite(inviteEvent) can cause
duplicate-key E11000 on retries because addPendingInvite does an insertOne;
modify the repository to perform an upsert (insert-or-update) for the
pending-invite record or change addPendingInvite to catch duplicate key errors
and ignore them (only rethrow non-duplicate errors). Ensure the chosen fix is
applied in pending-invite.repository.ts (prefer upsert with the invite ID as the
filter) or wrap the service call in a try/catch that suppresses E11000.
| await this.eventService.removePendingInvite( | ||
| joinEventFinal.eventId, | ||
| joinEventFinal.roomId, | ||
| ); |
There was a problem hiding this comment.
Bug: wrong identifier used to clear pending invite and unblock staged events.
removePendingInvite(joinEventFinal.eventId, …) won’t remove the pending invite (stored by inviteEventId) and won’t unmark staged events (flagged by sender+room). Clear by userId+roomId and unmark all staged events for that pair.
Apply:
- await this.eventService.removePendingInvite(
- joinEventFinal.eventId,
- joinEventFinal.roomId,
- );
+ await this.eventService.resolvePendingInvitesForUser(
+ joinEventFinal.roomId,
+ userId,
+ );Supporting changes are proposed in event.service.ts and event-staging.repository.ts comments below.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| await this.eventService.removePendingInvite( | |
| joinEventFinal.eventId, | |
| joinEventFinal.roomId, | |
| ); | |
| await this.eventService.resolvePendingInvitesForUser( | |
| joinEventFinal.roomId, | |
| userId, | |
| ); |
🤖 Prompt for AI Agents
In packages/federation-sdk/src/services/room.service.ts around lines 992 to 995,
the call to removePendingInvite uses joinEventFinal.eventId which is the wrong
identifier; change it to remove the pending invite by userId and roomId (e.g.,
joinEventFinal.userId, joinEventFinal.roomId) and ensure you also call or extend
the event staging cleanup to unmark all staged events for that sender+room pair
so staged events are cleared for that user in that room; adapt the call
signature and invocation to pass userId+roomId and trigger unmarking of all
staged events for that pair (see related suggested changes in event.service.ts
and event-staging.repository.ts).
|
Should be replaced by #212 |
Summary by CodeRabbit