Improve Reddit chat and media reliability
This commit is contained in:
1 parent
a44768c089
commit
345ec99348
14 files changed
+323
-676
No files matched your search
@@ -17,6 +17,7 @@ import type { AccountAuthGate } from '../account-auth-context.js';
|
||||
|
||||
const ACTIVE_WINDOW_MS = 96 * 60 * 60 * 1000;
|
||||
const ROOM_URL_PREFIX = 'https://chat.reddit.com/room/';
|
||||
const OPTIMISTIC_RECONCILIATION_WINDOW_MS = 2_000;
|
||||
|
||||
export interface TokenRefresher {
|
||||
/**
|
||||
@@ -221,6 +222,19 @@ export class ChatService {
|
||||
|
||||
async loadCache(): Promise<void> {
|
||||
await this.cache.load();
|
||||
let repairedDuplicateMessages = 0;
|
||||
for (const chat of this.cache.get()) {
|
||||
if (!chat.messages || chat.messages.length < 2) continue;
|
||||
const normalized = this.normalizeMessages(chat.messages, true);
|
||||
repairedDuplicateMessages += chat.messages.length - normalized.length;
|
||||
chat.messages = normalized;
|
||||
}
|
||||
if (repairedDuplicateMessages > 0) {
|
||||
this.logger.info(
|
||||
`[chat] removed ${repairedDuplicateMessages} legacy optimistic message duplicate(s) from cache`,
|
||||
);
|
||||
await this.cache.save();
|
||||
}
|
||||
for (const chat of this.cache.get()) {
|
||||
const t2 = this.extractT2(chat);
|
||||
if (t2 && chat.username && !chat.username.startsWith('t2_')) {
|
||||
@@ -632,7 +646,7 @@ export class ChatService {
|
||||
url: ROOM_URL_PREFIX + encodeURIComponent(roomId),
|
||||
roomId,
|
||||
messages: [
|
||||
{ sender: 'Me', text: message, timestamp: Date.now(), isSelf: true },
|
||||
{ eventId, sender: 'Me', text: message, timestamp: Date.now(), isSelf: true },
|
||||
],
|
||||
originMessage: message,
|
||||
};
|
||||
@@ -822,6 +836,7 @@ export class ChatService {
|
||||
const body = (e.content.body as string | undefined) ?? '';
|
||||
const mxc = typeof e.content.url === 'string' ? (e.content.url as string) : null;
|
||||
return {
|
||||
eventId: e.event_id,
|
||||
sender: e.sender === this.matrix.userId ? 'Me' : this.shortenSender(e.sender),
|
||||
text: body || (mxc ? '[Image]' : ''),
|
||||
timestamp: e.origin_server_ts,
|
||||
@@ -833,13 +848,87 @@ export class ChatService {
|
||||
|
||||
private mergeNewMessages(chat: ActiveChat, newMessages: ChatMessage[]): void {
|
||||
if (!chat.messages || chat.messages.length === 0) return;
|
||||
const seen = new Set(chat.messages.map((m) => `${m.timestamp}|${m.sender}`));
|
||||
for (const msg of newMessages) {
|
||||
const key = `${msg.timestamp}|${msg.sender}`;
|
||||
if (seen.has(key)) continue;
|
||||
chat.messages.push(msg);
|
||||
seen.add(key);
|
||||
chat.messages = this.normalizeMessages([...chat.messages, ...newMessages]);
|
||||
}
|
||||
|
||||
/**
|
||||
* Matrix event IDs are authoritative. The content/time fallback only
|
||||
* exists to repair cache entries written before ChatMessage stored them.
|
||||
*/
|
||||
private normalizeMessages(
|
||||
messages: ChatMessage[],
|
||||
repairLegacyOpeningDuplicate = false,
|
||||
): ChatMessage[] {
|
||||
const normalized: ChatMessage[] = [];
|
||||
const eventIndexes = new Map<string, number>();
|
||||
const fallbackKeys = new Set<string>();
|
||||
|
||||
for (const message of messages) {
|
||||
if (message.eventId) {
|
||||
const existingIndex = eventIndexes.get(message.eventId);
|
||||
if (existingIndex !== undefined) {
|
||||
const existing = normalized[existingIndex];
|
||||
normalized[existingIndex] = {
|
||||
...existing,
|
||||
...message,
|
||||
media: message.media ?? existing.media,
|
||||
};
|
||||
continue;
|
||||
}
|
||||
|
||||
const optimisticIndex = normalized.findIndex((candidate) =>
|
||||
!candidate.eventId
|
||||
&& candidate.isSelf
|
||||
&& message.isSelf
|
||||
&& candidate.sender === message.sender
|
||||
&& candidate.text === message.text
|
||||
&& Math.abs(candidate.timestamp - message.timestamp)
|
||||
<= OPTIMISTIC_RECONCILIATION_WINDOW_MS,
|
||||
);
|
||||
if (optimisticIndex !== -1) {
|
||||
normalized[optimisticIndex] = {
|
||||
...normalized[optimisticIndex],
|
||||
...message,
|
||||
media: message.media ?? normalized[optimisticIndex].media,
|
||||
};
|
||||
eventIndexes.set(message.eventId, optimisticIndex);
|
||||
continue;
|
||||
}
|
||||
|
||||
eventIndexes.set(message.eventId, normalized.length);
|
||||
normalized.push(message);
|
||||
continue;
|
||||
}
|
||||
|
||||
const fallbackKey = `${message.timestamp}|${message.sender}|${message.text}`;
|
||||
if (fallbackKeys.has(fallbackKey)) continue;
|
||||
fallbackKeys.add(fallbackKey);
|
||||
normalized.push(message);
|
||||
}
|
||||
|
||||
normalized.sort((a, b) => a.timestamp - b.timestamp);
|
||||
|
||||
// Before event IDs were persisted, startConversation cached a Date.now()
|
||||
// copy and /sync appended the same opening send with the server timestamp.
|
||||
// Restrict legacy repair to the opening pair and self-authored text so two
|
||||
// genuine repeated messages elsewhere in a conversation remain intact.
|
||||
if (repairLegacyOpeningDuplicate && normalized.length >= 2) {
|
||||
const first = normalized[0];
|
||||
const second = normalized[1];
|
||||
if (
|
||||
!first.eventId
|
||||
&& !second.eventId
|
||||
&& first.isSelf
|
||||
&& second.isSelf
|
||||
&& first.sender === second.sender
|
||||
&& first.text === second.text
|
||||
&& second.timestamp - first.timestamp <= OPTIMISTIC_RECONCILIATION_WINDOW_MS
|
||||
) {
|
||||
normalized.splice(1, 1);
|
||||
}
|
||||
}
|
||||
|
||||
return normalized;
|
||||
}
|
||||
|
||||
private async loadDetails(chat: ActiveChat): Promise<void> {
|
||||
@@ -866,6 +955,7 @@ export class ChatService {
|
||||
const body = (e.content.body as string | undefined) ?? '';
|
||||
const mxc = typeof e.content.url === 'string' ? (e.content.url as string) : null;
|
||||
return {
|
||||
eventId: e.event_id,
|
||||
sender: e.sender === this.matrix.userId ? 'Me' : this.shortenSender(e.sender),
|
||||
text: body || (mxc ? '[Image]' : ''),
|
||||
timestamp: e.origin_server_ts,
|
||||
@@ -890,9 +980,10 @@ export class ChatService {
|
||||
}
|
||||
}
|
||||
|
||||
chat.messages = this.mediaService
|
||||
const attachedMessages = this.mediaService
|
||||
? await this.mediaService.attachMedia(chat, scraped)
|
||||
: scraped.map(({ mediaUrls: _ignored, ...m }) => m as ChatMessage);
|
||||
chat.messages = this.normalizeMessages(attachedMessages);
|
||||
|
||||
if (chat.messages.length === 0) {
|
||||
this.logger.debug(`[chat] no messages in ${chat.roomName}`);
|
||||
|
||||
@@ -1,4 +1,6 @@
|
||||
export interface ChatMessage {
|
||||
/** Stable Matrix identity used to reconcile optimistic sends with /sync. */
|
||||
eventId?: string;
|
||||
sender: string;
|
||||
text: string;
|
||||
timestamp: number;
|
||||
|
||||
Reference in new issue
Block a user