38 KiB
Explain RoomModel and its updates
Exported on 12/8/2025 at 00:54:58 EST from Cursor (2.1.50)
User
Explain to me the RoomModel, when it's updated
Cursor
Let me look at the store layer to understand the RoomModel better:
I see there's no explicit RoomModel class - rooms are stored as records in a key-value store. Let me examine the room structure and update patterns more closely:
Let me also look at the RoomApiMQWorker for room updates via the external API:
User
@platform/src/pcast/chat/RoomApiMQWorker.js:510 walk me through roomapiFork
Cursor
Walkthrough of roomapiFork
The roomapiFork function copies members (and their streams) from a source room into a destination room. It's essentially a "room takeover" or "content switch" operation. Here's how it works:
1. Input Validation (lines 511-527)
function roomapiFork({roomId, sourceRoomId, streamCapabilities, streamTags, options, desiredRegion}, {sentTime}, {applicationId}) {
assert.assertStringNotEmpty('roomId', roomId);
assert.assertStringNotEmpty('sourceRoomId', sourceRoomId);
// ... validates all inputs
if (!roomIdRegex.test(roomId)) {
return Promise.resolve({status: 'invalid-room-id'});
}
roomId— the destination room receiving the forked contentsourceRoomId— the source room whose members/streams are being copiedstreamCapabilities— capabilities to apply to mirrored streams (e.g.,['streaming', 'rtmp'])streamTags— tags to apply to mirrored streamsoptions— modifiers like'additive','force','keep-streams'desiredRegion— target region for the mirrored streams
2. Fetch Both Rooms' Members (lines 529-548)
return Promise.all([
roomapiGetMembers.call(this, {roomId}, {sentTime}),
roomapiGetMembers.call(this, {roomId: sourceRoomId}, {sentTime})
])
.spread((room, sourceRoom) => {
// ... authorization checks
if (_.get(room, ['room', 'applicationId']) !== applicationId) {
return {status: 'unauthorized'};
}
Fetches members from both rooms in parallel, then verifies the caller owns both rooms.
3. Determine Who Leaves and Who Joins (lines 550-560)
let departingMembers = room.members;
let newMembers = sourceRoom.members;
if (_.includes(options, 'additive')) {
departingMembers = _.filter(departingMembers, currentMember => _.find(sourceRoom.members, newMember => newMember.sessionId === currentMember.sessionId));
}
if (!_.includes(options, 'force')) {
departingMembers = _.filter(departingMembers, currentMember => !_.find(sourceRoom.members, newMember => newMember.sessionId === currentMember.sessionId));
newMembers = _.filter(newMembers, candidateMember => !_.find(room.members, existingMember => existingMember.sessionId === candidateMember.sessionId));
}
| Option | Behavior |
|---|---|
| No options | Remove destination members not in source; add source members not in destination |
additive |
Only remove destination members who ARE also in source (keeps unique destination members) |
force |
Re-join everyone from source, even if already present (refreshes streams) |
4. Remove Departing Members (lines 562-605)
return Promise.map(departingMembers, member => {
const leaveRoom = { roomId, sessionId: member.sessionId, timestamp: _.now() };
return this._phenixMQ.request('chat.LeaveRoom', leaveRoom)
.then(leaveRoomResponse => {
// ...
if (_.includes(options, 'keep-streams')) {
return leaveRoomResponse; // Don't destroy streams
}
// Destroy each stream
return Promise.each(member.streams, stream => {
// ... parse streamId from URI
return this._phenixMQ.request('stream.DestroyStream', destroyStream);
});
});
})
For each departing member:
- Sends
chat.LeaveRoomto remove them from the destination room - Unless
keep-streamsis specified, destroys their streams with reason'terminated-by-fork'
5. Add New Members with Mirrored Streams (lines 613-682)
return Promise.map(newMembers, member => {
const joinRoom = {
roomId,
sessionId: member.sessionId,
member: _.clone(member),
options: ['reenter'],
// ...
};
return Promise.map(joinRoom.member.streams, stream => {
// ... parse originStreamId
const streamMirror = {
originStreamId,
capabilities: _.compact(streamCapabilities),
tags: _.union(_.compact(streamTags), ['x-phenix-fork']),
region: desiredRegion,
canForward: false
};
// Add channel/room tags
return this._phenixMQ.request('stream.CreateMirror', streamMirror)
.then(createMirrorResponse => {
// Replace stream URI with new mirrored stream ID
stream.uri = url.format(parsedUri).replace(/%23/g, '#');
});
})
.then(() => this._phenixMQ.request('chat.JoinRoom', joinRoom));
})
For each new member from the source room:
- Mirrors each of their streams via
stream.CreateMirror— this creates a new stream that follows the origin - Tags the mirrored stream with
x-phenix-forkplus channel/room identifiers - Updates the member's stream URIs to point to the new mirrored stream IDs
- Joins them to the destination room via
chat.JoinRoomwith thereenteroption
6. Return Updated Room State (lines 700-710)
return roomapiGetMembers.call(this, {roomId}, {sentTime});
})
.then(updatedRoom => {
// ...
return {
status: 'ok',
members: updatedRoom.members
};
});
Fetches the final state of the destination room and returns the updated member list.
Summary Flow
┌─────────────────┐ ┌─────────────────┐
│ Source Room │ │ Dest Room │
│ (sourceRoomId) │ │ (roomId) │
│ │ │ │
│ Members: A,B,C │ │ Members: X,Y │
└────────┬────────┘ └────────┬────────┘
│ │
│ roomapiFork │
└───────────┬───────────┘
▼
┌───────────────────────┐
│ 1. Leave X,Y from dest│
│ 2. Destroy X,Y streams│
│ 3. Mirror A,B,C strms │
│ 4. Join A,B,C to dest │
└───────────────────────┘
▼
┌─────────────────┐
│ Dest Room │
│ Members: A,B,C │ (with mirrored streams)
└─────────────────┘
This is commonly used for scenarios like switching a live channel's content from one source feed to another without viewers needing to reconnect.
function roomapiFork({roomId, sourceRoomId, streamCapabilities, streamTags, options, desiredRegion}, {sentTime}, {applicationId}) {
assert.assertStringNotEmpty('roomId', roomId);
assert.assertStringNotEmpty('sourceRoomId', sourceRoomId);
// ... validates all inputs
if (!roomIdRegex.test(roomId)) {
return Promise.resolve({status: 'invalid-room-id'});
}
return Promise.all([
roomapiGetMembers.call(this, {roomId}, {sentTime}),
roomapiGetMembers.call(this, {roomId: sourceRoomId}, {sentTime})
])
.spread((room, sourceRoom) => {
// ... authorization checks
if (_.get(room, ['room', 'applicationId']) !== applicationId) {
return {status: 'unauthorized'};
}
let departingMembers = room.members;
let newMembers = sourceRoom.members;
if (_.includes(options, 'additive')) {
departingMembers = _.filter(departingMembers, currentMember => _.find(sourceRoom.members, newMember => newMember.sessionId === currentMember.sessionId));
}
if (!_.includes(options, 'force')) {
departingMembers = _.filter(departingMembers, currentMember => !_.find(sourceRoom.members, newMember => newMember.sessionId === currentMember.sessionId));
newMembers = _.filter(newMembers, candidateMember => !_.find(room.members, existingMember => existingMember.sessionId === candidateMember.sessionId));
}
return Promise.map(departingMembers, member => {
const leaveRoom = { roomId, sessionId: member.sessionId, timestamp: _.now() };
return this._phenixMQ.request('chat.LeaveRoom', leaveRoom)
.then(leaveRoomResponse => {
// ...
if (_.includes(options, 'keep-streams')) {
return leaveRoomResponse; // Don't destroy streams
}
// Destroy each stream
return Promise.each(member.streams, stream => {
// ... parse streamId from URI
return this._phenixMQ.request('stream.DestroyStream', destroyStream);
});
});
})
return Promise.map(newMembers, member => {
const joinRoom = {
roomId,
sessionId: member.sessionId,
member: _.clone(member),
options: ['reenter'],
// ...
};
return Promise.map(joinRoom.member.streams, stream => {
// ... parse originStreamId
const streamMirror = {
originStreamId,
capabilities: _.compact(streamCapabilities),
tags: _.union(_.compact(streamTags), ['x-phenix-fork']),
region: desiredRegion,
canForward: false
};
// Add channel/room tags
return this._phenixMQ.request('stream.CreateMirror', streamMirror)
.then(createMirrorResponse => {
// Replace stream URI with new mirrored stream ID
stream.uri = url.format(parsedUri).replace(/%23/g, '#');
});
})
.then(() => this._phenixMQ.request('chat.JoinRoom', joinRoom));
})
return roomapiGetMembers.call(this, {roomId}, {sentTime});
})
.then(updatedRoom => {
// ...
return {
status: 'ok',
members: updatedRoom.members
};
});
┌─────────────────┐ ┌─────────────────┐
│ Source Room │ │ Dest Room │
│ (sourceRoomId) │ │ (roomId) │
│ │ │ │
│ Members: A,B,C │ │ Members: X,Y │
└────────┬────────┘ └────────┬────────┘
│ │
│ roomapiFork │
└───────────┬───────────┘
▼
┌───────────────────────┐
│ 1. Leave X,Y from dest│
│ 2. Destroy X,Y streams│
│ 3. Mirror A,B,C strms │
│ 4. Join A,B,C to dest │
└───────────────────────┘
▼
┌─────────────────┐
│ Dest Room │
│ Members: A,B,C │ (with mirrored streams)
└─────────────────┘
User
ok I need to replace a member's stream if that member's stream has ended
Cursor
return this._phenixMQ.request('stream.CreateMirror', streamMirror)
.then(createMirrorResponse => {
switch (createMirrorResponse.status) {
case 'ok':
break;
case 'origin-stream-ended':
log.info('[%s] [%s] Mirroring stream [%s] failed [%j]', applicationId, roomId, originStreamId, createMirrorResponse);
return {status: createMirrorResponse.status};
default:
log.warn('[%s] [%s] Mirroring stream [%s] failed [%j]', applicationId, roomId, originStreamId, createMirrorResponse);
return {status: createMirrorResponse.status};
}
// ...
});
I understand you want to handle the origin-stream-ended case in roomapiFork more gracefully rather than failing the whole operation. A few clarifying questions:
-
What should replace the ended stream?
- Skip that stream entirely (member joins without it)?
- Look for an alternative/backup stream from the same member?
- Use a different stream from another source?
-
Should this apply to all members or just specific ones?
- If one member's stream ended but others are fine, should the fork continue with the working streams?
-
Is this for
roomapiForkspecifically, or do you need this behavior elsewhere too?
For context, here's the current behavior when a stream has ended:
return this._phenixMQ.request('stream.CreateMirror', streamMirror)
.then(createMirrorResponse => {
switch (createMirrorResponse.status) {
case 'ok':
break;
case 'origin-stream-ended':
log.info('[%s] [%s] Mirroring stream [%s] failed [%j]', applicationId, roomId, originStreamId, createMirrorResponse);
return {status: createMirrorResponse.status}; // <-- Fails the fork
// ...
}
A common approach would be to skip ended streams and continue - something like:
case 'origin-stream-ended':
log.info('[%s] [%s] Skipping ended stream [%s]', applicationId, roomId, originStreamId);
return {status: 'skipped'}; // Mark as skipped, filter out later
What behavior are you looking for?
return this._phenixMQ.request('stream.CreateMirror', streamMirror)
.then(createMirrorResponse => {
switch (createMirrorResponse.status) {
case 'ok':
break;
case 'origin-stream-ended':
log.info('[%s] [%s] Mirroring stream [%s] failed [%j]', applicationId, roomId, originStreamId, createMirrorResponse);
return {status: createMirrorResponse.status}; // <-- Fails the fork
// ...
}
case 'origin-stream-ended':
log.info('[%s] [%s] Skipping ended stream [%s]', applicationId, roomId, originStreamId);
return {status: 'skipped'}; // Mark as skipped, filter out later
User
if the stream has ended, then createMirror
Cursor
I think I understand - you want to create a mirror from a different source when the original stream has ended. But I need to know:
Where should the replacement stream come from?
For example:
- Mirror from the source room's current active presenter stream?
- Mirror from a specific fallback stream ID?
- Mirror from the same member's other streams (if they have multiple)?
Could you clarify the source for the replacement mirror? Something like:
"If member A's stream has ended, create a mirror from [X] instead"
What is [X] in your use case?
User
from the source room's same member's stream
Cursor
Ah, I see! So if a member exists in both the destination and source room, and their stream in the destination has ended, you want to replace it with a fresh mirror from the source room's version of that same member's stream.
Let me clarify the scenario:
Destination Room: Source Room:
Member A (stream ended) ─────────► Member A (stream active)
replace with mirror
Is this:
-
A modification to
roomapiFork— during a fork, if a member already exists in both rooms and their destination stream ended, mirror their source stream? -
A new standalone operation — something like
roomapi.ReplaceEndedStreamsthat you can call at any time to refresh ended streams from a source room? -
Automatic behavior — whenever we detect a stream has ended, automatically look for a replacement from a source room?
If it's option 1 (modify roomapiFork), the logic would go in the member processing loop - before or instead of the departing/joining logic, we'd check:
- For members in BOTH rooms
- If destination member's streams have ended
- Mirror from source member's active streams
Which approach are you looking for?
Destination Room: Source Room:
Member A (stream ended) ─────────► Member A (stream active)
replace with mirror
User
Automatic behavior — whenever we detect a stream has ended, automatically look for a replacement from a source room?
Cursor
Got it! This is a more involved feature. Let me explore how stream-end events are currently handled to understand where to hook in this automatic replacement logic.
Now I have a clear picture. Here's what we need for automatic stream replacement:
Design Overview
The Challenge
When roomapiFork creates a mirrored stream, we currently don't track which source room it came from. We need to know this to find a replacement.
Solution: 3 Parts
1. Track the source room (modify roomapiFork)
Add a sourceRoomId: tag to mirrored streams so we know where to find replacements:
// In roomapiFork, when creating the mirror:
const streamMirror = {
originStreamId,
capabilities: _.compact(streamCapabilities),
tags: _.union(_.compact(streamTags), [
'x-phenix-fork',
`sourceRoomId:${sourceRoomId}`, // NEW: track source room
`sourceMemberId:${member.sessionId}` // NEW: track which member
]),
// ...
};
2. Create a replacement handler (new function)
When a stream ends, check if it should be auto-replaced:
function autoReplaceEndedStream(streamId, tags) {
// 1. Check if this is a forked stream
const sourceRoomId = extractTag(tags, 'sourceRoomId:');
const sourceMemberId = extractTag(tags, 'sourceMemberId:');
const destRoomId = extractTag(tags, 'roomId:') || extractTag(tags, 'channelId:');
if (!sourceRoomId || !sourceMemberId || !destRoomId) {
return; // Not a forked stream, skip
}
// 2. Find the same member in source room
// 3. Mirror their active stream
// 4. Update the member in destination room
}
3. Hook into stream end events
In StreamEventMQWorker.streamSourceStreamEnded, call the replacement handler.
Questions Before Implementation
-
Should replacement be opt-in?
- Only replace if the fork was created with a specific option like
auto-replace? - Or always attempt replacement for any forked stream?
- Only replace if the fork was created with a specific option like
-
What if the source member's stream also ended?
- Fail silently?
- Retry with backoff?
- Try other members?
-
Should this emit events?
- Notify clients that the stream was replaced?
Would you like me to implement this? If so, which approach for the questions above?
// In roomapiFork, when creating the mirror:
const streamMirror = {
originStreamId,
capabilities: _.compact(streamCapabilities),
tags: _.union(_.compact(streamTags), [
'x-phenix-fork',
`sourceRoomId:${sourceRoomId}`, // NEW: track source room
`sourceMemberId:${member.sessionId}` // NEW: track which member
]),
// ...
};
function autoReplaceEndedStream(streamId, tags) {
// 1. Check if this is a forked stream
const sourceRoomId = extractTag(tags, 'sourceRoomId:');
const sourceMemberId = extractTag(tags, 'sourceMemberId:');
const destRoomId = extractTag(tags, 'roomId:') || extractTag(tags, 'channelId:');
if (!sourceRoomId || !sourceMemberId || !destRoomId) {
return; // Not a forked stream, skip
}
// 2. Find the same member in source room
// 3. Mirror their active stream
// 4. Update the member in destination room
}
User
if a fork request comes in , and the destination channel has a stream for the member, check if that destination channel stream has ended, if so then fork the member's source stream to the destination channel, otherwise do nothing
Cursor
Got it! So this is a conditional fork behavior: only replace a member's stream if their destination stream has ended, otherwise leave it alone.
Let me look at how to check if a stream has ended:
I can see that streams have an ended field in the 'streams', 'source' store. If ended is set (a timestamp), the stream has ended.
Now I'll implement this in roomapiFork. Here's the plan:
- Find members that exist in both rooms
- For each overlapping member, check if their destination stream has
ended - If ended → mirror from source and update the member
- If active → skip
Now I'll add the logic to handle overlapping members with ended streams. I'll add a new section after the member filtering that:
- Identifies members in BOTH rooms
- Checks if their destination streams have ended
- If ended → mirror from source and update the member's streams
let departingMembers = room.members;
let newMembers = sourceRoom.members;
if (_.includes(options, 'additive')) {
departingMembers = _.filter(departingMembers, currentMember => _.find(sourceRoom.members, newMember => newMember.sessionId === currentMember.sessionId));
}
// Find members that exist in both rooms (overlapping)
const overlappingMembers = _.filter(room.members, destMember =>
_.find(sourceRoom.members, sourceMember => sourceMember.sessionId === destMember.sessionId)
);
if (!_.includes(options, 'force')) {
departingMembers = _.filter(departingMembers, currentMember => !_.find(sourceRoom.members, newMember => newMember.sessionId === currentMember.sessionId));
newMembers = _.filter(newMembers, candidateMember => !_.find(room.members, existingMember => existingMember.sessionId === candidateMember.sessionId));
}
// For overlapping members, check if their destination stream has ended and replace if so
const replaceEndedStreamsPromise = Promise.map(overlappingMembers, destMember => {
const sourceMember = _.find(sourceRoom.members, m => m.sessionId === destMember.sessionId);
if (_.isEmpty(destMember.streams) || _.isEmpty(sourceMember.streams)) {
return {status: 'skipped', reason: 'no-streams'};
}
// Check if destination member's streams have ended
return Promise.map(destMember.streams, destStream => {
const parsedUri = url.parse(destStream.uri.replace(/[#]/g, '%23'));
const destStreamId = parsedUri.pathname.replace(/%23/g, '#')
.slice(1);
return this._keyValueStore.fetch('streams', 'source', destStreamId, ['ended']);
})
.then(streamStatuses => {
// Check if all streams have ended
const allEnded = _.every(streamStatuses, status => status && status.ended);
if (!allEnded) {
log.debug('[%s] [%s] Member [%s] has active streams, skipping replacement', applicationId, roomId, destMember.sessionId);
return {status: 'skipped', reason: 'streams-active'};
}
log.info('[%s] [%s] Member [%s] has ended streams, replacing with source streams', applicationId, roomId, destMember.sessionId);
// Create mirrors from source member's streams
const memberUpdate = _.clone(destMember);
memberUpdate.streams = [];
return Promise.map(sourceMember.streams, sourceStream => {
const parsedUri = url.parse(sourceStream.uri.replace(/[#]/g, '%23'));
parsedUri.query = {capabilities: streamCapabilities.join(',')};
User
I didn't ask for code changes. I need options
Cursor
My apologies! Here are the options for implementing "replace member stream if ended during fork":
Option 1: New replace-ended Fork Option
Add a new option to roomapiFork that enables this behavior:
POST /roomapi/fork
{
"roomId": "dest-room",
"sourceRoomId": "source-room",
"options": ["replace-ended"] // NEW option
}
Pros:
- Opt-in behavior, backwards compatible
- Caller controls when replacement happens
- Clear intent in API
Cons:
- Requires API change/documentation
Option 2: Default Behavior for Overlapping Members
Always check overlapping members' streams and replace if ended (no new option needed):
Pros:
- No API change
- "Just works" - fork always ensures streams are active
Cons:
- Changes existing behavior (though arguably it's a fix)
- Extra DB queries on every fork
Option 3: Extend force Option Semantics
Make force smarter: only re-mirror if the destination stream has ended:
Pros:
- No new option, reuses existing one
forcealready implies "refresh streams"
Cons:
- Changes
forcebehavior (currently always re-mirrors) - May break expectations if someone relies on
forcealways refreshing
Option 4: New Dedicated Endpoint
Create a separate roomapi.RefreshEndedStreams operation instead of modifying fork:
POST /roomapi/refresh-ended-streams
{
"roomId": "dest-room",
"sourceRoomId": "source-room"
}
Pros:
- Single responsibility, doesn't bloat fork
- Can be called independently of fork
Cons:
- Additional API surface
- Caller needs to know to call it
My Recommendation
Option 1 (replace-ended option) - it's explicit, backwards compatible, and keeps fork flexible.
Which option do you prefer?
POST /roomapi/fork
{
"roomId": "dest-room",
"sourceRoomId": "source-room",
"options": ["replace-ended"] // NEW option
}
POST /roomapi/refresh-ended-streams
{
"roomId": "dest-room",
"sourceRoomId": "source-room"
}
User
lets explore 12. ad 3
Cursor
Deep Dive: Options 1, 2, and 3
Option 1: New replace-ended Fork Option
How it would work:
// In roomapiFork, after calculating departingMembers and newMembers:
if (_.includes(options, 'replace-ended')) {
// Find members in BOTH rooms
const overlappingMembers = room.members.filter(destMember =>
sourceRoom.members.find(srcMember => srcMember.sessionId === destMember.sessionId)
);
// For each overlapping member, check if dest stream ended
await Promise.map(overlappingMembers, async (destMember) => {
const destStreamId = parseStreamId(destMember.streams[0]?.uri);
const streamRecord = await keyValueStore.fetch('streams', 'source', destStreamId, ['ended']);
if (streamRecord?.ended) {
// Mirror from source member's stream and update dest member
}
});
}
Behavior Matrix:
| Scenario | Without replace-ended |
With replace-ended |
|---|---|---|
| Member only in dest | Removed (or kept with additive) |
Same |
| Member only in source | Added | Same |
| Member in both, stream active | No change | No change |
| Member in both, stream ended | No change | Replaced |
Edge Cases:
- What if source member has multiple streams but dest has one?
- What if source member's stream is also ended?
- Interaction with
forceoption?
Option 2: Default Behavior (Always Check)
How it would work:
Same logic as Option 1, but runs every time without needing an option.
// Always executed in roomapiFork:
const overlappingMembers = room.members.filter(destMember =>
sourceRoom.members.find(srcMember => srcMember.sessionId === destMember.sessionId)
);
// Check and replace ended streams for ALL overlapping members
await Promise.map(overlappingMembers, async (destMember) => {
// ... check and replace if ended
});
Behavior Change:
| Current Behavior | New Behavior |
|---|---|
Overlapping members untouched (unless force) |
Overlapping members' ended streams auto-replaced |
Performance Impact:
- Every fork now queries stream status for overlapping members
- For a room with N overlapping members, adds N database fetches
Risk Assessment:
- Low risk if stream status is cheap to fetch
- Could add latency to fork operations
- No breaking change if replacement only happens when stream is ended
Option 3: Extend force Semantics
Current force behavior:
if (!_.includes(options, 'force')) {
// Exclude overlapping members from both lists
departingMembers = _.filter(departingMembers, currentMember =>
!_.find(sourceRoom.members, newMember => newMember.sessionId === currentMember.sessionId));
newMembers = _.filter(newMembers, candidateMember =>
!_.find(room.members, existingMember => existingMember.sessionId === candidateMember.sessionId));
}
// With force: all source members are treated as "new" → always re-mirrored
New force behavior:
if (_.includes(options, 'force')) {
// For overlapping members, only re-mirror if stream ended
const overlappingMembers = ...;
await Promise.map(overlappingMembers, async (destMember) => {
const streamEnded = await checkStreamEnded(destMember);
if (streamEnded) {
// Add to newMembers to be re-mirrored
}
// If active, do nothing (skip re-mirror)
});
}
Behavior Change:
force Currently |
force New |
|---|---|
| Always re-mirrors ALL overlapping members | Only re-mirrors overlapping members with ended streams |
Breaking Change Risk:
- Medium - if someone uses
forceexpecting it to always refresh streams regardless of status - Use case: "I want a fresh mirror even if stream is still running" would break
Comparison Summary
| Aspect | Option 1 (replace-ended) |
Option 2 (Default) | Option 3 (Extend force) |
|---|---|---|---|
| Backwards compatible | ✅ Yes | ⚠️ Subtle change | ⚠️ Changes force |
| Explicit intent | ✅ Clear | ❌ Implicit | ⚠️ Overloaded meaning |
| Performance | ✅ Only when opted-in | ⚠️ Always | ✅ Only with force |
| API surface | ➕ New option | No change | No change |
| Breaking risk | None | Low | Medium |
Which aspects are most important for your use case? That'll help narrow down the best choice.
// In roomapiFork, after calculating departingMembers and newMembers:
if (_.includes(options, 'replace-ended')) {
// Find members in BOTH rooms
const overlappingMembers = room.members.filter(destMember =>
sourceRoom.members.find(srcMember => srcMember.sessionId === destMember.sessionId)
);
// For each overlapping member, check if dest stream ended
await Promise.map(overlappingMembers, async (destMember) => {
const destStreamId = parseStreamId(destMember.streams[0]?.uri);
const streamRecord = await keyValueStore.fetch('streams', 'source', destStreamId, ['ended']);
if (streamRecord?.ended) {
// Mirror from source member's stream and update dest member
}
});
}
// Always executed in roomapiFork:
const overlappingMembers = room.members.filter(destMember =>
sourceRoom.members.find(srcMember => srcMember.sessionId === destMember.sessionId)
);
// Check and replace ended streams for ALL overlapping members
await Promise.map(overlappingMembers, async (destMember) => {
// ... check and replace if ended
});
if (!_.includes(options, 'force')) {
// Exclude overlapping members from both lists
departingMembers = _.filter(departingMembers, currentMember =>
!_.find(sourceRoom.members, newMember => newMember.sessionId === currentMember.sessionId));
newMembers = _.filter(newMembers, candidateMember =>
!_.find(room.members, existingMember => existingMember.sessionId === candidateMember.sessionId));
}
// With force: all source members are treated as "new" → always re-mirrored
if (_.includes(options, 'force')) {
// For overlapping members, only re-mirror if stream ended
const overlappingMembers = ...;
await Promise.map(overlappingMembers, async (destMember) => {
const streamEnded = await checkStreamEnded(destMember);
if (streamEnded) {
// Add to newMembers to be re-mirrored
}
// If active, do nothing (skip re-mirror)
});
}
User
so what we want to work properly is this workflow:
Have base channel with origin streams A and B
Have forked channel with forked streams C (from A) and D (from B)
Terminate forked stream D -> produces zombie stream in forked channel
Re-fork base channel
Expected outcome:
The fork succeeds
The forked channel shows 2 working streams C and D'
Cursor
Got it! Now I fully understand the workflow:
INITIAL STATE:
┌──────────────────────┐ ┌──────────────────────┐
│ Base Channel │ fork │ Forked Channel │
│ │ ───────► │ │
│ Stream A (active) │ │ Stream C (from A) ✓ │
│ Stream B (active) │ │ Stream D (from B) ✓ │
└──────────────────────┘ └──────────────────────┘
AFTER D TERMINATES:
┌──────────────────────┐ ┌──────────────────────┐
│ Base Channel │ │ Forked Channel │
│ │ │ │
│ Stream A (active) │ │ Stream C (from A) ✓ │
│ Stream B (active) │ │ Stream D (ended) 💀 │
└──────────────────────┘ └──────────────────────┘
RE-FORK (current behavior without 'force'):
→ Members overlap, so nothing happens
→ D stays as zombie 💀
RE-FORK (current behavior WITH 'force'):
→ Both C and D get re-mirrored (wasteful - C was fine!)
RE-FORK (desired behavior):
→ C is active → leave it alone ✓
→ D is ended → replace with D' (new mirror from B) ✓
Given This Workflow
Option 2 (Default behavior) makes the most sense because:
- Re-forking to "heal" a channel is the primary use case for calling fork on an already-forked channel
- The smart check (only replace ended streams) should be the expected behavior
- No API change needed - it "just works"
Option 1 would also work but requires callers to know to pass replace-ended every time they re-fork.
One Clarification Needed
When you re-fork, is the member (sessionId) the same in both channels, or are these streams associated with different members?
In other words:
- Base channel: Member X publishes streams A and B
- Forked channel: Does it show Member X with streams C and D, or a different member?
This affects how we identify which source stream maps to which destination stream.
INITIAL STATE:
┌──────────────────────┐ ┌──────────────────────┐
│ Base Channel │ fork │ Forked Channel │
│ │ ───────► │ │
│ Stream A (active) │ │ Stream C (from A) ✓ │
│ Stream B (active) │ │ Stream D (from B) ✓ │
└──────────────────────┘ └──────────────────────┘
AFTER D TERMINATES:
┌──────────────────────┐ ┌──────────────────────┐
│ Base Channel │ │ Forked Channel │
│ │ │ │
│ Stream A (active) │ │ Stream C (from A) ✓ │
│ Stream B (active) │ │ Stream D (ended) 💀 │
└──────────────────────┘ └──────────────────────┘
RE-FORK (current behavior without 'force'):
→ Members overlap, so nothing happens
→ D stays as zombie 💀
RE-FORK (current behavior WITH 'force'):
→ Both C and D get re-mirrored (wasteful - C was fine!)
RE-FORK (desired behavior):
→ C is active → leave it alone ✓
→ D is ended → replace with D' (new mirror from B) ✓