This commit is contained in:
Timothy Jaeryang Baek 2026-10-16 01:36:10 +04:00
parent 4a70253bc8
commit 4dedaaf6e2
7 changed files with 259 additions and 42 deletions

View file

@ -232,6 +232,45 @@ def get_session_pool_batches():
return [list(SESSION_POOL.items())]
async def publish_chat_presence(chat_id: str):
# Room maps only contain this worker's sockets; the session pool spans workers.
lock = (
REDIS.lock(f'{REDIS_KEY_PREFIX}:chat:{chat_id}:presence:lock',
timeout=WEBSOCKET_REDIS_LOCK_TIMEOUT,
blocking_timeout=WEBSOCKET_REDIS_LOCK_TIMEOUT)
if REDIS is not None
else SOCKET_EVENT_LOCKS.setdefault(('chat', chat_id), asyncio.Lock())
)
try:
async with lock, asyncio.timeout(WEBSOCKET_REDIS_LOCK_TIMEOUT / 2):
now = time.time()
people = {}
for batch in get_session_pool_batches():
for _, session in batch:
if (
session
and chat_id in (session.get('chat_ids') or [])
and now - session.get('last_seen_at', 0) <= SESSION_POOL_TIMEOUT
):
people[session['id']] = {'id': session['id'], 'name': session.get('name', '')}
if REDIS is not None and not await lock.owned():
return
await sio.emit(
'events',
{
'chat_id': chat_id,
'shared': True,
'data': {
'type': 'chat:presence',
'data': {'users': sorted(people.values(), key=lambda person: person['id'])},
},
},
room=f'chat:{chat_id}',
)
except Exception:
log.exception('Failed to publish presence for chat %s', chat_id)
async def periodic_session_pool_cleanup():
"""Reap orphaned SESSION_POOL entries that missed heartbeats (e.g. crashed instance)."""
retry_delay = random.uniform(WEBSOCKET_REDIS_LOCK_TIMEOUT / 2, WEBSOCKET_REDIS_LOCK_TIMEOUT)
@ -258,11 +297,20 @@ async def periodic_session_pool_cleanup():
]
if expired:
log.warning('Reaping %d orphaned session(s) from the session pool', len(expired))
chat_ids = {
chat_id for sid, entry in batch if sid in expired for chat_id in entry.get('chat_ids', [])
}
if WEBSOCKET_MANAGER == 'redis':
SESSION_POOL.delete_many(*expired)
else:
for sid in expired:
SESSION_POOL.pop(sid, None)
for sid, entry in batch:
if sid in expired:
for chat_id in entry.get('chat_ids', []):
await sio.leave_room(sid, f'chat:{chat_id}')
for chat_id in chat_ids:
await publish_chat_presence(chat_id)
await asyncio.sleep(0) # don't hold the loop for the whole sweep
next_cleanup_at = time.monotonic() + SESSION_POOL_TIMEOUT
@ -567,12 +615,12 @@ async def user_join(sid, data):
async def refresh_chat_access(chat_id=None):
# The pool includes sessions on other workers; Socket.IO routes room changes via Redis.
access = {}
chat_ids = set()
for batch in get_session_pool_batches():
for sid, session in batch:
if not session:
continue
chat_ids = set(session.get('chat_ids') or [])
for cid in list(chat_ids):
for cid in session.get('chat_ids') or []:
if chat_id and cid != chat_id:
continue
key = (cid, session['id'])
@ -581,12 +629,18 @@ async def refresh_chat_access(chat_id=None):
access[key] = bool(user and await Chats.get_accessible_chat_by_id(cid, user))
if not access[key]:
await sio.leave_room(sid, f'chat:{cid}')
chat_ids.discard(cid)
chat_ids.add(cid)
# Preserve joins/leaves and heartbeats that happened during access checks.
current = SESSION_POOL.get(sid)
if current:
SESSION_POOL[sid] = {
**current, 'chat_ids': [id for id in current.get('chat_ids', []) if id != cid]
}
await sio.emit(
'events', {'chat_id': cid, 'shared': True, 'data': {'type': 'chat:access', 'data': {}}}, to=sid
)
if chat_ids != set(session.get('chat_ids') or []):
SESSION_POOL[sid] = {**session, 'chat_ids': list(chat_ids)}
for cid in chat_ids:
await publish_chat_presence(cid)
@sio.on('heartbeat')
@ -750,18 +804,25 @@ async def chat_events(sid, data):
chat_id = data.get('chat_id')
if not isinstance(chat_id, str) or not is_saved_chat_id(chat_id):
return False
session = SESSION_POOL.get(sid) or user
chat_ids = set(session.get('chat_ids') or [])
if event_type == 'leave':
await sio.leave_room(sid, f'chat:{chat_id}')
chat_ids.discard(chat_id)
else:
reader = await Users.get_user_by_id(user['id'])
if not reader or not await Chats.get_accessible_chat_by_id(chat_id, reader):
async with SOCKET_EVENT_LOCKS.setdefault(('session', sid), asyncio.Lock()):
if event_type == 'leave':
await sio.leave_room(sid, f'chat:{chat_id}')
else:
reader = await Users.get_user_by_id(user['id'])
if not reader or not await Chats.get_accessible_chat_by_id(chat_id, reader):
return False
await sio.enter_room(sid, f'chat:{chat_id}')
session = SESSION_POOL.get(sid)
if not session:
await sio.leave_room(sid, f'chat:{chat_id}')
return False
await sio.enter_room(sid, f'chat:{chat_id}')
chat_ids.add(chat_id)
SESSION_POOL[sid] = {**session, 'chat_ids': list(chat_ids)}
chat_ids = set(session.get('chat_ids') or [])
if event_type == 'leave':
chat_ids.discard(chat_id)
else:
chat_ids.add(chat_id)
SESSION_POOL[sid] = {**session, 'chat_ids': list(chat_ids)}
await publish_chat_presence(chat_id)
return True
if event_type == 'last_read_at':
@ -1136,7 +1197,11 @@ async def disconnect(sid, reason=None):
await encryption.disconnect(sid)
LOCAL_AUTHENTICATED_SIDS.discard(sid)
if sid in SESSION_POOL:
chat_ids = (SESSION_POOL.get(sid) or {}).get('chat_ids', [])
del SESSION_POOL[sid]
for chat_id in chat_ids:
await sio.leave_room(sid, f'chat:{chat_id}')
await publish_chat_presence(chat_id)
# Clean up USAGE_POOL entries for this session
for model_id, connections in list(USAGE_POOL.items()):

View file

@ -1303,6 +1303,8 @@
};
let joinedChatId = '';
let liveParticipants: { id: string; name: string }[] = [];
const clearPresence = () => (liveParticipants = []);
let chatRefresh: { id: string; promise: Promise<void> } | null = null;
let typingUsers: { id: string; name: string }[] = [];
let typingUsersTimeout: Record<string, ReturnType<typeof setTimeout>> = {};
@ -1388,6 +1390,7 @@
mergeChatMessages(fresh.chat.history, baseline);
} catch (error) {
if ($chatId === id && joinedChatId === id) {
clearPresence();
chat = null;
history = { messages: {}, currentId: null };
if (!embedded) await goto('/');
@ -1401,20 +1404,32 @@
};
const syncChatRoom = (socket: typeof $socket, id: string, temporary: boolean) => {
const next = socket && id && !temporary && !isTemporaryChatId(id) ? id : '';
const next = socket?.connected && id && !temporary && !isTemporaryChatId(id) ? id : '';
if (next === joinedChatId) return;
clearTyping();
if (joinedChatId)
socket?.emit('events:chat', { chat_id: joinedChatId, data: { type: 'leave' } });
clearPresence();
if (joinedChatId && socket?.connected)
socket.emit('events:chat', { chat_id: joinedChatId, data: { type: 'leave' } });
joinedChatId = next;
if (next)
socket?.emit('events:chat', { chat_id: next, data: { type: 'join' } }, () => {
if ($chatId === next) void refreshChat();
const connectionId = socket?.id;
if (next && socket)
socket.emit('events:chat', { chat_id: next, data: { type: 'join' } }, (accepted: boolean) => {
if (!socket.connected || socket.id !== connectionId || $chatId !== next) return;
if (accepted) void refreshChat();
else clearPresence();
});
};
$: syncChatRoom($socket, $chatId, $temporaryChatEnabled);
const chatEventHandler = async (event, cb) => {
if (event?.data?.type === 'chat:presence') {
if (!$socket?.connected || event.chat_id !== $chatId || event.chat_id !== joinedChatId)
return;
liveParticipants = (event.data.data?.users ?? []).filter(
(person: { id: string; name: string }) => person.id !== $user?.id
);
return;
}
console.log(event);
if (event?.data?.type === 'voice:play_animation') {
cb?.(
@ -1850,6 +1865,7 @@
$socket?.on('events', chatEventHandler);
$socket?.on('connect', handleSocketConnect);
$socket?.on('disconnect', clearTyping);
$socket?.on('disconnect', clearPresence);
$socket?.on('disconnect', cancelBrowserInteractions);
$audioQueue?.destroy();
@ -1947,8 +1963,9 @@
// to the admin panel), otherwise the previously-viewed chat stays selected
// in the sidebar and deleting/archiving it wrongly navigates away.
clearTyping();
if (joinedChatId)
$socket?.emit('events:chat', { chat_id: joinedChatId, data: { type: 'leave' } });
clearPresence();
if (joinedChatId && $socket?.connected)
$socket.emit('events:chat', { chat_id: joinedChatId, data: { type: 'leave' } });
joinedChatId = '';
chatId.set('');
chatTitle.set('');
@ -1957,6 +1974,7 @@
$socket?.off('events', chatEventHandler);
$socket?.off('connect', handleSocketConnect);
$socket?.off('disconnect', clearTyping);
$socket?.off('disconnect', clearPresence);
$socket?.off('disconnect', cancelBrowserInteractions);
dismissContextCompactionToast();
audioQueueInstance?.destroy();
@ -4990,6 +5008,7 @@
</div>
{:else}
<Navbar
{liveParticipants}
bind:this={navbarElement}
readOnly={isShared}
chat={{

View file

@ -36,11 +36,13 @@
import ChatCheck from '../icons/ChatCheck.svelte';
import Knobs from '../icons/Knobs.svelte';
import { isTemporaryChatId } from '$lib/utils/chatId';
import ChatParticipants from '$lib/components/common/ChatParticipants.svelte';
const i18n: any = getContext('i18n');
export let initNewChat: Function;
export let readOnly: boolean = false;
export let liveParticipants: { id: string; name: string }[] = [];
export let shareEnabled: boolean = false;
export let scrollTop = 0;
export let scrollToTop: (() => void) | null = null;
@ -113,7 +115,9 @@
{/if}
<div
class="flex-1 overflow-hidden max-w-full mt-0.5 py-0.5 pl-1 {$showSidebar ? 'ml-1' : ''}"
class="flex-1 min-w-0 overflow-hidden max-w-full mt-0.5 py-0.5 pl-1 {$showSidebar
? 'ml-1'
: ''}"
>
{#if chat?.id}
<div class="flex max-w-full min-w-0 items-center gap-2 mr-2">
@ -174,7 +178,7 @@
{/if}
</div>
<div class="lg:mr-1 flex flex-none items-center gap-2 self-center">
<div class="lg:mr-1 flex flex-none items-center gap-4 self-center">
<!-- <div class="md:hidden flex self-center w-[0.0625rem] h-5 mx-2 bg-gray-300 dark:bg-stone-700" /> -->
{#if $user?.role === 'user' ? ($user?.permissions?.chat?.temporary ?? true) && !($user?.permissions?.chat?.temporary_enforced ?? false) : true}
@ -243,6 +247,14 @@
</Tooltip>
{/if}
{#if liveParticipants.length > 0}
<ChatParticipants
participants={liveParticipants}
mode="presence"
avatarClass="size-5"
/>
{/if}
{#if $user?.role === 'admin' || ($user?.permissions.chat?.controls ?? true)}
<Tooltip content={$i18n.t('Controls')}>
<button

View file

@ -12,10 +12,15 @@
export let avatarClass = 'size-4';
export let activeProfileId: string | null = null;
export let overflowOpen = false;
export let mode: 'contributors' | 'presence' = 'contributors';
// Preserve the base overflow label alongside its generated plural forms.
// $i18n.t('Show all {{count}} contributors');
$: people = participants.length ? participants : owner?.id ? [owner] : [];
$: people = participants.length
? participants
: mode === 'contributors' && owner?.id
? [owner]
: [];
$: if (activeProfileId && !people.some((person) => person.id === activeProfileId)) {
activeProfileId = null;
}
@ -47,6 +52,8 @@
{#each people.slice(0, 3) as person (person.id)}
<div class="relative rounded-full hover:z-10 focus-within:z-10">
<ProfilePreview
side={mode === 'presence' ? 'bottom' : 'right'}
align={mode === 'presence' ? 'end' : 'center'}
user={{ id: person.id, name: person.name || $i18n.t('Unknown User') }}
bind:openPreview={
() => !overflowOpen && activeProfileId === person.id,
@ -55,7 +62,11 @@
>
{@render avatar(
person,
`${avatarClass} bg-white ring-1 ring-white dark:bg-gray-900 dark:ring-gray-900`
`${avatarClass} bg-white dark:bg-gray-900 ${
mode === 'presence'
? `ring-2 ${['ring-red-400', 'ring-sky-400', 'ring-lime-500', 'ring-amber-400', 'ring-violet-400'][person.id.charCodeAt(0) % 5]}`
: 'ring-1 ring-white dark:ring-gray-900'
}`
)}
</ProfilePreview>
</div>
@ -65,8 +76,12 @@
{#if people.length > 3}
<Popover.Root bind:open={overflowOpen} onOpenChange={() => (activeProfileId = null)}>
<Popover.Trigger
class="shrink-0 rounded px-0.5 text-[0.625rem] text-gray-500 hover:bg-gray-100 focus-visible:outline dark:text-gray-400 dark:hover:bg-gray-800"
aria-label={$i18n.t('Show all {{count}} contributors', { count: people.length })}
class="shrink-0 rounded px-0.5 {mode === 'presence'
? 'h-6 text-xs'
: 'text-[0.625rem]'} text-gray-500 hover:bg-gray-100 focus-visible:outline dark:text-gray-400 dark:hover:bg-gray-800"
aria-label={mode === 'presence'
? $i18n.t('People here')
: $i18n.t('Show all {{count}} contributors', { count: people.length })}
>
+{people.length - 3}
</Popover.Trigger>
@ -75,15 +90,22 @@
role="dialog"
trapFocus={false}
side="bottom"
align="start"
align={mode === 'presence' ? 'end' : 'start'}
sideOffset={6}
class="z-[9999] w-60 max-w-[calc(100vw-1.5rem)] max-h-80 overflow-y-auto rounded-xl bg-white p-2 text-gray-900 shadow-lg ring-1 ring-black/5 dark:bg-gray-850 dark:text-gray-100 dark:ring-white/10"
aria-label={$i18n.t('Contributors')}
aria-label={mode === 'presence' ? $i18n.t('People here') : $i18n.t('Contributors')}
>
{#if mode === 'presence'}
<div class="px-2 py-1 text-xs text-gray-500 dark:text-gray-400">
{$i18n.t('People here')}
</div>
{/if}
<ul class="space-y-1">
{#each people as person (person.id)}
<li class="rounded-lg p-1 hover:bg-gray-50 dark:hover:bg-gray-800">
<ProfilePreview
side={mode === 'presence' ? 'bottom' : 'right'}
align={mode === 'presence' ? 'end' : 'center'}
user={{ id: person.id, name: person.name || $i18n.t('Unknown User') }}
bind:openPreview={
() => overflowOpen && activeProfileId === person.id,

View file

@ -29,6 +29,7 @@ export type State = {
status: 'joining' | 'active' | 'ended';
messages: Message[];
members: Member[];
typing: string[];
self: string;
host: string;
};
@ -73,8 +74,22 @@ export function request(socket: Socket, action: string, data: object): Promise<a
}
export class Conversation {
state = writable<State>({ status: 'joining', messages: [], members: [], self: '', host: '' });
private current: State = { status: 'joining', messages: [], members: [], self: '', host: '' };
state = writable<State>({
status: 'joining',
messages: [],
members: [],
typing: [],
self: '',
host: ''
});
private current: State = {
status: 'joining',
messages: [],
members: [],
typing: [],
self: '',
host: ''
};
private worker: Worker;
private pending = new Map<
number,
@ -89,6 +104,8 @@ export class Conversation {
private timer?: ReturnType<typeof setInterval>;
private joinTimer?: ReturnType<typeof setTimeout>;
private messageIds = new Set<string>();
private lastTyping?: number;
private typingTimers = new Map<string, ReturnType<typeof setTimeout>>();
private disposers: (() => void)[] = [];
constructor(
@ -232,6 +249,9 @@ export class Conversation {
const members: Member[] = JSON.parse(await this.crypto('members'));
if (this.stopped) return;
const previous = new Set(this.current.members.map((m) => m.identity.participation));
for (const id of this.current.typing) {
if (!members.some((m) => m.identity.participation === id)) this.removeTyping(id);
}
this.update({
members: members.map((m) => ({
...m,
@ -280,6 +300,7 @@ export class Conversation {
await this.crypto('join', event.packet, bindings);
await this.activate();
} else if (event.kind === 'departed') {
this.removeTyping(event.participation);
if (event.admitted && this.participation === event.host) {
const packet = await this.crypto('remove', event.participation, bindings);
await this.send('packet', { kind: 'mls', packet });
@ -319,7 +340,24 @@ export class Conversation {
!/^[a-f0-9-]{36}$/.test(data.id)
)
throw new Error();
if (data.type === 'receipt') {
if (data.type === 'typing') {
if (typeof data.typing !== 'boolean') throw new Error();
const id = sender.participation;
if (
id === this.participation ||
!this.current.members.some((m) => m.identity.participation === id)
)
return;
if (data.typing) {
clearTimeout(this.typingTimers.get(id));
this.typingTimers.set(
id,
setTimeout(() => this.removeTyping(id), 5000)
);
if (!this.current.typing.includes(id))
this.update({ typing: [...this.current.typing, id] });
} else this.removeTyping(id);
} else if (data.type === 'receipt') {
const messages = this.current.messages.map((m) =>
m.id === data.id &&
m.sender === this.participation &&
@ -336,6 +374,7 @@ export class Conversation {
)
throw new Error();
this.messageIds.add(data.id);
this.removeTyping(sender.participation);
// MLS rejects replay; the bounded ID set additionally rejects repeated app envelopes.
if (this.messageIds.size > 1000)
this.messageIds.delete(this.messageIds.values().next().value!);
@ -359,6 +398,36 @@ export class Conversation {
await this.send('packet', { kind: 'mls', packet });
} else throw new Error();
}
private removeTyping(id: string) {
clearTimeout(this.typingTimers.get(id));
this.typingTimers.delete(id);
if (this.current.typing.includes(id))
this.update({ typing: this.current.typing.filter((participant) => participant !== id) });
}
sendTyping(typing: boolean) {
if (this.current.status !== 'active') return;
const now = Date.now();
if (
typing
? this.lastTyping !== undefined && now - this.lastTyping < 2000
: this.lastTyping === undefined
)
return;
this.lastTyping = typing ? now : undefined;
return this.enqueue(async () => {
const packet = await this.crypto(
'encrypt',
JSON.stringify({
type: 'typing',
id: crypto.randomUUID(),
typing,
channel: this.channel,
session: this.view!.session
})
);
await this.send('packet', { kind: 'mls', packet });
});
}
sendText(text: string) {
if (
this.current.status !== 'active' ||
@ -366,6 +435,7 @@ export class Conversation {
new TextEncoder().encode(text).length > 16384
)
return;
this.sendTyping(false);
return this.enqueue(async () => {
const id = crypto.randomUUID();
const expected = this.current.members
@ -407,6 +477,9 @@ export class Conversation {
}
this.pending.clear();
this.messageIds.clear();
for (const timer of this.typingTimers.values()) clearTimeout(timer);
this.typingTimers.clear();
this.lastTyping = undefined;
if (this.view && this.participation && connected(this.socket)) {
void request(this.socket, 'leave', {
channel: this.channel,
@ -415,7 +488,7 @@ export class Conversation {
}).catch(() => {});
}
this.view = undefined;
this.current = { status: 'ended', messages: [], members: [], self: '', host: '' };
this.current = { status: 'ended', messages: [], members: [], typing: [], self: '', host: '' };
this.state.set(this.current);
}
}

View file

@ -1937,6 +1937,7 @@
"pdf, docx, pptx, xlsx": "",
"Peak tokens": "",
"pending": "",
"People here": "People here",
"Pending": "",
"Per automation": "",
"Per chat": "",

View file

@ -11,6 +11,7 @@
import RichTextInput from '$lib/components/common/RichTextInput.svelte';
import type { Editor } from '@tiptap/core';
import Markdown from '$lib/components/chat/Messages/Markdown.svelte';
import Skeleton from '$lib/components/chat/Messages/Skeleton.svelte';
import Modal from '$lib/components/common/Modal.svelte';
import Tooltip from '$lib/components/common/Tooltip.svelte';
import ProfileImage from '$lib/components/chat/Messages/ProfileImage.svelte';
@ -33,7 +34,10 @@
is_private: true
};
let conversation: Conversation | undefined;
let state: State = { status: 'ended', messages: [], members: [], self: '', host: '' };
let state: State = { status: 'ended', messages: [], members: [], typing: [], self: '', host: '' };
$: typingMembers = state.members.filter((member) =>
state.typing.includes(member.identity.participation)
);
let draft = '';
let inputEditor: Editor | null = null;
let verified = new Set<string>();
@ -58,7 +62,7 @@
});
function startSession(channel: string) {
stopSession();
state = { status: 'ended', messages: [], members: [], self: '', host: '' };
state = { status: 'ended', messages: [], members: [], typing: [], self: '', host: '' };
const intent = consumeEntry(channel);
const connection = get(socket);
const account = get(user);
@ -71,15 +75,17 @@
return;
}
const unsubscribe = conversation.state.subscribe((next) => {
const messagesChanged = state.messages !== next.messages;
state = next;
if (next.status === 'ended') {
showParticipants = false;
draft = '';
verified = new Set();
}
void tick().then(() => {
if (list) list.scrollTop = list.scrollHeight;
});
if (messagesChanged)
void tick().then(() => {
if (list) list.scrollTop = list.scrollHeight;
});
});
const logout = user.subscribe((current) => {
if (current?.id !== account.id || !canUseEncryption(current, get(config)?.features)) close();
@ -220,7 +226,25 @@
{/each}
</div>
</div>
<form class="shrink-0 pb-3 px-3" on:submit|preventDefault={send}>
<form
class="shrink-0 pb-3 px-3"
on:submit|preventDefault={send}
on:focusout={() => conversation?.sendTyping(false)}
>
{#if typingMembers.length > 0}
<div
class="flex items-center gap-1.5 px-1 pb-2 text-xs text-gray-500 dark:text-gray-400"
role="status"
>
<Skeleton size="xs" />
<div>
<span class="font-normal text-black dark:text-white"
>{typingMembers.map((member) => member.name).join(', ')}</span
>
{typingMembers.length === 1 ? $i18n.t('is typing...') : $i18n.t('are typing...')}
</div>
</div>
{/if}
<div
id="message-input-container"
class="flex-1 flex flex-col relative w-full shadow-sm rounded-2xl border border-gray-100/60 dark:border-gray-800/50 hover:border-gray-100 focus-within:border-gray-100 hover:dark:border-gray-800 focus-within:dark:border-gray-800 transition bg-white dark:bg-gray-400/5 dark:text-gray-100"
@ -248,6 +272,7 @@
floatingMenuPlacement="top-start"
onChange={({ md }) => {
draft = state.status === 'active' ? md : '';
void conversation?.sendTyping(Boolean(draft.trim()));
}}
on:keydown={({ detail: { event } }) => {
if (event.isComposing || event.defaultPrevented || event.key !== 'Enter') return;