import type { Logger } from '../../utils/logger.js'; import type { ConversationOutcomesRepository, ConversationStateRepository, MessagesRepository, RedditUsersRepository, SentMessageRecord, SuccessfulMessagesRepository, } from '../../database/repositories.js'; import { recordConversationReply } from '../conversation-outcome-recorder.js'; import type { MediaService } from '../media.js'; import { ChatCache } from './cache.js'; import { MatrixClient, MatrixAuthError, MatrixFeatureDisabledError, MatrixRateLimitError, } from './matrix/client.js'; import type { MatrixJoinedRoom, MatrixTimelineEvent } from './matrix/types.js'; import type { ActiveChat, ChatMessage, ScrapedChatMessage, InvitationCooldown } from './types.js'; export type { InvitationCooldown } from './types.js'; import { fetchWithTimeout } from '../../utils/fetch-with-timeout.js'; import type { AccountAuthGate } from '../account-auth-context.js'; import { containsMediaRequest } from '../conversation-boundaries.js'; import type { ConversationLearningObserver } from '../conversation-learning-observer.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 { /** * Requests replacement Matrix credentials. Extension-only authentication * deliberately rejects this until the user imports a fresh dump. */ refresh(): Promise; } class MatrixAuthenticationUnavailableError extends Error { constructor(cause: unknown) { super('Matrix authentication could not be restored', { cause }); this.name = 'MatrixAuthenticationUnavailableError'; } } /** * Optional Reddit-API hook for resolving t2 IDs to real usernames. Without * it, conversations show up as `t2_abc123` (stable but ugly). With it, the * dashboard sees real names. */ export interface UsernameResolver { /** Map t2 IDs (with or without `t2_` prefix) to Reddit usernames. */ resolveT2Ids(t2Ids: string[]): Promise>; /** Resolve a Reddit username to its t2 account ID when starting a DM. */ resolveUsernameToT2?(username: string): Promise; } export type ConversationStartResult = | { status: 'sent'; chat: ActiveChat } | { status: 'terminal'; reason: string } | { status: 'retryable'; reason: string; retryAt: number }; export interface ChatSendReceipt { eventId: string; sentAt: number; } interface BuildChatOpts { knownUsernames: Map; includeTimelineMessages?: boolean; } /** * Public chat API. Internally a thin wrapper around `MatrixClient`: * * - `/sync` (delta cursor) drives `getActiveChats` / `refreshTopChats` — * cheap, just-changed rooms only. * - `/joined_rooms` + per-room `/messages` paginate every historical * conversation when `fetchAllMessages` is called. * - Sends use `PUT .../send/m.room.message/` directly — no UI. * * The public surface (`getActiveChats`, `sendChatMessage`, etc.) is * preserved from the old Playwright implementation so callers (web routes, * MediaSyncService) don't change. */ export type SyncPhase = | 'starting' | 'seeding' | 'listing-rooms' | 'resolving-usernames' | 'loading-details' | 'finalizing' | 'done'; export interface SyncProgress { kind: 'full' | 'delta'; phase: SyncPhase; /** Number of items completed in the current phase (e.g. rooms processed). */ current: number; /** Total items in the current phase, if known. */ total: number; /** Optional label for the in-flight item (e.g. current room name). */ label?: string; /** Wall-clock ms when the sync started — for elapsed-time display. */ startedAt: number; } export interface SyncStatus { /** True if a sync is currently running. */ syncing: boolean; /** Kind of sync currently running, if any. */ syncingKind: 'full' | 'delta' | null; /** Wall-clock ms of the last successful full backfill, or null. */ lastFullSyncAt: number | null; /** Wall-clock ms of the last delta sync, or null. */ lastDeltaSyncAt: number | null; /** Has the bot ever completed a full sync on this machine? */ hasEverFullSynced: boolean; /** Number of chats currently cached. */ cachedChats: number; /** Live progress for the in-flight sync, if any. */ progress: SyncProgress | null; } export type SyncCompletion = | { status: 'completed' } | { status: 'failed'; error: string }; /** `null` signals "sync ended" (clear progress UI). */ export type SyncProgressListener = ( progress: SyncProgress | null, completion?: SyncCompletion, ) => void; /** Fired when a chat we sent into first receives a reply from the other party. */ export type NewReplyListener = (info: { conversationId: string; replyId: string; username: string; replySnippet: string; }) => void; /** Fired the first time a conversation we opened receives inbound media. */ export type MediaReceivedListener = (info: { conversationId: string; username: string; count: number; }) => void; export class ChatService { private cache: ChatCache; private queue: Promise = Promise.resolve(); private deltaSyncInFlight: Promise | null = null; private readonly t2ToUsername = new Map(); /** Rooms whose /joined_members lookup has already failed — don't retry. */ private readonly unresolvableRooms = new Set(); private roomStateLookupAvailable = true; private syncingKind: 'full' | 'delta' | null = null; private progress: SyncProgress | null = null; private progressListener: SyncProgressListener | null = null; private newReplyListener: NewReplyListener | null = null; private mediaReceivedListener: MediaReceivedListener | null = null; private invitationCooldown: InvitationCooldown | null = null; constructor( private matrix: MatrixClient, private logger: Logger, private successfulMessagesRepo?: SuccessfulMessagesRepository, private mediaService?: MediaService, private tokenRefresher?: TokenRefresher, private usernameResolver?: UsernameResolver, cache?: ChatCache, private messagesRepo?: MessagesRepository, private redditUsersRepo?: RedditUsersRepository, private accountAuth?: AccountAuthGate, private outcomesRepo?: ConversationOutcomesRepository, private conversationStateRepo?: ConversationStateRepository, private conversationLearningObserver?: ConversationLearningObserver, ) { this.cache = cache ?? new ChatCache(logger); if (this.redditUsersRepo) { try { for (const u of this.redditUsersRepo.getAll()) { const bare = u.t2Id.replace(/^t2_/, ''); this.t2ToUsername.set(bare, u.username); } } catch (err) { this.logger.warn('[chat] failed to seed username cache from db:', err); } } } /** Persist a t2→username resolution; safe to call with no repo configured. */ private persistUsername(t2Id: string, username: string): void { if (!this.redditUsersRepo) return; try { this.redditUsersRepo.upsert({ t2Id, username }); } catch (err) { this.logger.debug('[chat] failed to persist reddit user mapping:', err); } } /** Resolve `@t2_abc:reddit.com` or `t2_abc` to a real username if known. */ resolveSenderDisplay(mxidOrT2: string): string { const mxidMatch = mxidOrT2.match(/^@(t2_[^:]+):/); const t2 = mxidMatch ? mxidMatch[1] : mxidOrT2; const bare = t2.replace(/^t2_/, ''); return this.t2ToUsername.get(bare) ?? t2; } // ------------ scheduling + auth helpers ------------ /** Serialize work so concurrent callers don't clobber cache state. */ private enqueue(task: () => Promise): Promise { const result = this.queue.then(() => task()); this.queue = result.catch(() => undefined); return result; } /** * Run a Matrix call; if it 401/403s, invalidate the shared account gate. * The configured refresher reports that a new extension dump is required. */ private async withAuth(fn: () => Promise): Promise { if (!this.accountAuth?.isFullyAuthenticated()) { throw new MatrixAuthenticationUnavailableError( new Error('Full account authentication is required'), ); } try { return await fn(); } catch (err) { if (err instanceof MatrixAuthError && this.tokenRefresher) { this.logger.warn('[chat] Matrix token rejected — waiting for a new extension dump'); try { await this.tokenRefresher.refresh(); } catch (refreshError) { this.accountAuth?.invalidate('Matrix authentication could not be restored'); throw new MatrixAuthenticationUnavailableError(refreshError); } try { return await fn(); } catch (retryError) { this.accountAuth?.invalidate('Matrix authentication was rejected after refresh'); throw retryError; } } throw err; } } // ------------ public API ------------ async loadCache(): Promise { 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_')) { this.t2ToUsername.set(t2, chat.username); this.persistUsername(t2, chat.username); } } // Upgrade any `t2_xxx` cache entries using DB-known mappings so the UI // never has to display a raw t2 ID when we already know the username. let upgraded = 0; for (const chat of this.cache.get()) { const t2 = this.extractT2(chat); if (!t2) continue; const known = this.t2ToUsername.get(t2); if (known) { chat.username = known; if (!chat.roomName || chat.roomName.startsWith('t2_')) chat.roomName = known; upgraded++; } } if (upgraded > 0) { this.logger.info(`[chat] upgraded ${upgraded} cached chats from db username map`); } const persisted = this.cache.getCooldown(); if (persisted && persisted.until > Date.now()) { this.invitationCooldown = persisted; const minutesLeft = Math.ceil((persisted.until - Date.now()) / 60_000); this.logger.warn( `[chat] restored send cooldown from cache: ${persisted.reason} (~${minutesLeft}m left, code=${persisted.redditCode ?? 'n/a'})`, ); } else if (persisted) { this.cache.setCooldown(null); } } async switchAccount(accountId: string): Promise { await this.enqueue(async () => { await this.cache.switchAccount(accountId); this.t2ToUsername.clear(); this.unresolvableRooms.clear(); this.invitationCooldown = this.cache.getCooldown(); }); } async saveCacheExternal(): Promise { await this.cache.save(); } async getCachedConversations(): Promise { if (this.cache.get().length > 0) return this.cache.get(); await this.cache.load(); return this.cache.get(); } /** * Returns chats with activity in the last 96h. Uses the persisted * `next_batch` cursor — first call is essentially the initial sync, * subsequent calls are delta-only. */ async getActiveChats(): Promise { return this.enqueue(async () => { if (this.cache.isFresh()) return this.cache.get(); this.beginSync('delta'); let failure: unknown; try { this.updateProgress({ phase: 'starting', current: 0, total: 0 }); const { chats: merged } = await this.syncOnce(); const cutoff = Date.now() - ACTIVE_WINDOW_MS; const active = merged.filter((c) => c.lastMessageTime >= cutoff); this.cache.set(active); await this.cache.save(merged); this.logger.info(`[chat] sync → ${merged.length} known rooms, ${active.length} active`); this.updateProgress({ phase: 'done', current: merged.length, total: merged.length }); return active; } catch (error) { failure = error; throw error; } finally { this.endSync(failure); } }); } /** * Incremental refresh — calls /sync once and applies the deltas to the * cache. The `limit` arg is preserved for API compatibility but Matrix * already returns only rooms with new events, so it's a no-op. */ refreshTopChats(_limit = 5): Promise { if (this.deltaSyncInFlight) return this.deltaSyncInFlight; const operation = this.enqueue(async () => { this.beginSync('delta'); let failure: unknown; try { this.updateProgress({ phase: 'starting', current: 0, total: 0 }); const { changedRooms } = await this.syncOnce(); this.logger.info(`[chat] incremental sync complete (${changedRooms} changed rooms)`); this.updateProgress({ phase: 'done', current: this.cache.get().length, total: this.cache.get().length, }); return changedRooms; } catch (error) { failure = error; this.logger.error('[chat] refreshTopChats failed:', error); throw error; } finally { this.endSync(failure); } }); const tracked = operation.finally(() => { this.deltaSyncInFlight = null; }); this.deltaSyncInFlight = tracked; return tracked; } /** * Full backfill — walks every room via /joined_rooms and paginates each * one's full history. Slow but rare; intended for "rebuild the cache from * scratch" or initial provisioning. */ async fetchAllMessages(options: { requireComplete?: boolean } = {}): Promise { return this.enqueue(async () => { this.beginSync('full'); let failure: unknown; try { this.logger.info('[chat] full backfill starting (via /joined_rooms)...'); const t0 = Date.now(); this.updateProgress({ phase: 'seeding', current: 0, total: 0 }); // Seed display-name hints from one /sync (heroes / m.room.name). const seed = await this.withAuth(() => this.matrix.sync({ timeoutMs: 0 })); this.cache.setNextBatch(seed.next_batch); const seedJoined = seed.rooms?.join ?? {}; this.updateProgress({ phase: 'listing-rooms', current: 0, total: 0 }); const joinedIds = await this.withAuth(() => this.matrix.getJoinedRooms()); this.logger.info(`[chat] backfill: ${joinedIds.length} joined rooms`); const knownUsernames = new Map(this.t2ToUsername); const chats: ActiveChat[] = []; for (const roomId of joinedIds) { try { const room = seedJoined[roomId]; const chat = this.roomToActiveChat(roomId, room, { knownUsernames, includeTimelineMessages: true, }); chats.push(chat); } catch (err) { this.logger.warn(`[chat] backfill: failed to seed ${roomId}: ${(err as Error).message}`); } } // Make the inbox usable as soon as Matrix has listed the rooms. A // full history backfill can take many minutes for large accounts, so // keeping these summaries local until the final room has loaded makes // a healthy sync look like an empty inbox for the entire run. chats.sort((a, b) => b.lastMessageTime - a.lastMessageTime); this.cache.set(chats); await this.cache.save(chats); this.updateProgress({ phase: 'resolving-usernames', current: 0, total: joinedIds.length, }); const resolvedUsernames = await this.resolveUsernamesForRooms(joinedIds, seedJoined); for (const chat of chats) { const t2 = this.extractT2(chat); if (!t2) continue; const resolved = resolvedUsernames.get(t2); if (!resolved) continue; chat.username = resolved; if (!chat.roomName || chat.roomName.startsWith('t2_')) chat.roomName = resolved; } await this.cache.save(chats); await this.repairDirectRoomMappings(chats); let done = 0; let failedRooms = 0; for (const chat of chats) { this.updateProgress({ phase: 'loading-details', current: done, total: chats.length, label: chat.roomName || chat.username, }); try { await this.loadDetails(chat); } catch (err) { if (err instanceof MatrixAuthenticationUnavailableError) throw err; failedRooms++; this.logger.error(`[chat] backfill: failed details for ${chat.roomName}`, err); } done++; } this.updateProgress({ phase: 'finalizing', current: chats.length, total: chats.length, }); chats.sort((a, b) => b.lastMessageTime - a.lastMessageTime); this.cache.set(chats); if (failedRooms > 0) { await this.cache.save(chats); const error = new Error( `Full chat backfill incomplete: ${failedRooms} of ${chats.length} conversations failed`, ); this.logger.error(`[chat] ${error.message}`); if (options.requireComplete) throw error; return chats; } this.cache.markFullSync(); await this.cache.save(chats); this.logger.info( `[chat] full backfill done in ${Math.round((Date.now() - t0) / 1000)}s — ${chats.length} chats`, ); this.updateProgress({ phase: 'done', current: chats.length, total: chats.length }); return chats; } catch (error) { failure = error; this.logger.error('[chat] fetchAllMessages failed:', error); if (error instanceof MatrixAuthenticationUnavailableError || options.requireComplete) throw error; return this.cache.get(); } finally { this.endSync(failure); } }); } /** * Decide whether to run a full backfill or just a delta sync based on * whether the bot has ever completed a full sync on this machine. * Called once on bootstrap. */ async runStartupSync(): Promise<{ kind: 'full' | 'delta'; chats: number }> { const { lastFullSyncAt } = this.cache.getMarkers(); if (!lastFullSyncAt || this.cache.get().length === 0) { this.logger.info('[chat] startup: no prior full sync recorded → running full backfill'); const chats = await this.fetchAllMessages(); this.backfillConversationOutcomes(); return { kind: 'full', chats: chats.length }; } this.logger.info( `[chat] startup: full sync was ${new Date(lastFullSyncAt).toISOString()} → running delta`, ); const chats = await this.getActiveChats(); this.backfillConversationOutcomes(); return { kind: 'delta', chats: chats.length }; } /** Lightweight status for the dashboard's sync controls. */ getSyncStatus(): SyncStatus { const markers = this.cache.getMarkers(); return { syncing: this.syncingKind !== null, syncingKind: this.syncingKind, lastFullSyncAt: markers.lastFullSyncAt, lastDeltaSyncAt: markers.lastDeltaSyncAt, hasEverFullSynced: !!markers.lastFullSyncAt, cachedChats: this.cache.get().length, progress: this.progress, }; } /** Subscribe to live sync-progress updates. Single listener; replaces any prior one. */ setProgressListener(listener: SyncProgressListener | null): void { this.progressListener = listener; } /** Subscribe to new-reply notifications (fires once per chat the first time it gets a reply). */ setNewReplyListener(listener: NewReplyListener | null): void { this.newReplyListener = listener; } /** Subscribe to inbound-media notifications (fires once per chat, on the first image). */ setMediaReceivedListener(listener: MediaReceivedListener | null): void { this.mediaReceivedListener = listener; } private beginSync(kind: 'full' | 'delta'): void { this.syncingKind = kind; this.progress = { kind, phase: 'starting', current: 0, total: 0, startedAt: Date.now(), }; this.emitProgress(); } private updateProgress(patch: Partial>): void { if (!this.progress) return; this.progress = { ...this.progress, ...patch }; this.emitProgress(); } private endSync(error?: unknown): void { this.syncingKind = null; this.progress = null; if (this.progressListener) { try { const completion: SyncCompletion = error === undefined ? { status: 'completed' } : { status: 'failed', error: error instanceof Error ? error.message : String(error), }; this.progressListener(null, completion); } catch (err) { this.logger.debug('progress listener error:', err); } } } private emitProgress(): void { if (!this.progress || !this.progressListener) return; try { this.progressListener(this.progress); } catch (err) { this.logger.debug('progress listener error:', err); } } /** * Load full message history (+ media) for a single chat. Mutates * `chat.messages`. */ async loadConversationDetails(chat: ActiveChat): Promise { if (!this.accountAuth?.isFullyAuthenticated()) return; return this.loadDetails(chat); } /** * Send a reply into an existing room. Accepts either a full * `https://chat.reddit.com/room/` URL (web UI compatibility) or * a raw Matrix room ID like `!abc:reddit.com`. */ async sendChatMessage(chatUrlOrRoomId: string, message: string, txnId?: string): Promise { return (await this.sendChatMessageDetailed(chatUrlOrRoomId, message, txnId)) !== null; } /** Send a reply and retain the authoritative Matrix event identity. */ async sendChatMessageDetailed( chatUrlOrRoomId: string, message: string, txnId?: string, ): Promise { if (!this.accountAuth?.isFullyAuthenticated()) return null; const active = this.getInvitationCooldown(); if (active && this.blocksExistingRoomSends(active)) { const secondsLeft = Math.ceil((active.until - Date.now()) / 1000); this.logger.warn( `[chat] skipping sendChatMessage — send cooldown active for ${secondsLeft}s (${active.redditCode ?? 'rate-limited'})`, ); return null; } return this.enqueue(async () => { try { const roomId = ChatService.extractRoomId(chatUrlOrRoomId); if (!roomId) { this.logger.error(`[chat] cannot extract room ID from "${chatUrlOrRoomId}"`); return null; } const eventId = await this.withAuth(() => this.matrix.sendTextMessage(roomId, message, txnId)); this.logger.info(`[chat] sent message to ${roomId} (event_id=${eventId})`); const chat = this.cache.get().find((c) => c.roomId === roomId); const sentAt = Date.now(); if (chat) { chat.lastMessageSnippet = message.slice(0, 200); chat.lastMessageTime = sentAt; chat.isFromUs = true; } return { eventId, sentAt }; } catch (error) { if (error instanceof MatrixRateLimitError) { this.setInvitationCooldown(error, 'all-sends'); return null; } this.logger.error('[chat] sendChatMessage failed:', error); return null; } }); } async sendChatImage( chatUrlOrRoomId: string, image: { bytes: Buffer; mimeType: string; fileName: string; width: number; height: number }, txnId: string, existingMxcUri?: string | null, ): Promise<{ eventId: string; mxcUri: string } | null> { if (!this.accountAuth?.isFullyAuthenticated()) return null; const active = this.getInvitationCooldown(); if (active && this.blocksExistingRoomSends(active)) return null; return this.enqueue(async () => { try { const roomId = ChatService.extractRoomId(chatUrlOrRoomId); if (!roomId) return null; const mxcUri = existingMxcUri ?? await this.withAuth(() => this.matrix.uploadImage(image.bytes, image.mimeType, image.fileName)); const eventId = await this.withAuth(() => this.matrix.sendImageMessage( roomId, image.fileName, mxcUri, { mimeType: image.mimeType, size: image.bytes.length, width: image.width, height: image.height }, txnId, )); this.logger.info(`[chat] sent image to ${roomId} (event_id=${eventId})`); const chat = this.cache.get().find((item) => item.roomId === roomId); if (chat) { chat.lastMessageSnippet = '[Image]'; chat.lastMessageTime = Date.now(); chat.isFromUs = true; } return { eventId, mxcUri }; } catch (error) { if (error instanceof MatrixRateLimitError) { this.setInvitationCooldown(error, 'all-sends'); return null; } this.logger.error('[chat] sendChatImage failed:', error); return null; } }); } /** * Start a brand-new DM with the given Reddit user and send the opening * line. Returns the new chat (or null on failure). */ async startConversation( username: string, message: string, knownT2Id?: string, ): Promise { const result = await this.startConversationDetailed(username, message, knownT2Id); return result.status === 'sent' ? result.chat : null; } /** Start a DM while preserving terminal versus retryable failure semantics. */ async startConversationDetailed( username: string, message: string, knownT2Id?: string, ): Promise { if (!this.accountAuth?.isFullyAuthenticated()) { return { status: 'retryable', reason: 'Account authentication is temporarily unavailable', retryAt: Date.now() + 60_000, }; } const active = this.getInvitationCooldown(); if (active) { const secondsLeft = Math.ceil((active.until - Date.now()) / 1000); this.logger.warn( `[chat] skipping startConversation(${username}) — invitation cooldown active for ${secondsLeft}s (${active.redditCode ?? 'rate-limited'})`, ); return { status: 'retryable', reason: active.reason, retryAt: active.until }; } return this.enqueue(async () => { try { const t2Id = await this.resolveUsernameToT2(username, knownT2Id); if (!t2Id) { this.logger.error(`[chat] cannot start conversation: unknown user "${username}"`); return { status: 'terminal', reason: 'Reddit user no longer exists' }; } const mxid = MatrixClient.t2ToMxid(t2Id); const { roomId, eventId } = await this.withAuth(() => this.matrix.sendInitialDirectMessage(mxid, message), ); this.logger.info( `[chat] started conversation with ${username} (${t2Id}) → room=${roomId} event=${eventId}`, ); this.t2ToUsername.set(t2Id.replace(/^t2_/, ''), username); this.persistUsername(t2Id, username); const chat: ActiveChat = { username, roomName: username, lastMessageTime: Date.now(), lastMessageSnippet: message.slice(0, 200), isFromUs: true, unread: false, url: ROOM_URL_PREFIX + encodeURIComponent(roomId), roomId, messages: [ { eventId, sender: 'Me', text: message, timestamp: Date.now(), isSelf: true }, ], originMessage: message, }; const existing = this.cache.get(); const merged = [chat, ...existing.filter((c) => c.roomId !== roomId)]; this.cache.set(merged); await this.cache.save(merged); return { status: 'sent', chat }; } catch (error) { if (error instanceof MatrixRateLimitError) { this.setInvitationCooldown(error, 'room-creation'); const cooldown = this.getInvitationCooldown(); return { status: 'retryable', reason: cooldown?.reason ?? 'Reddit temporarily rate-limited chat invitations', retryAt: cooldown?.until ?? Date.now() + error.retryAfterMs + 5_000, }; } this.logger.error('[chat] startConversation failed:', error); const message = error instanceof Error ? error.message : String(error); if (/do not accept chat requests/i.test(message)) { return { status: 'terminal', reason: 'Recipient does not accept chat requests' }; } if (/unknown user|not found|M_NOT_FOUND/i.test(message)) { return { status: 'terminal', reason: 'Reddit user is unavailable' }; } return { status: 'retryable', reason: 'Reddit chat invitation failed temporarily', retryAt: Date.now() + 60_000, }; } }); } /** * Matrix sends are authoritative — the event_id in the PUT response * confirms delivery. Kept on the public API for compatibility with the * old Playwright sender; always returns true. */ async verifyMessageSent(_recipient: string, _messageText: string): Promise { return this.accountAuth?.isFullyAuthenticated() ?? false; } /** * Legacy debug hook. There's no DOM to inspect on the Matrix backend; * returns a small status payload so any UI still wired to this endpoint * stops 500-ing. */ async debugInspectDom(_chatUrl: string): Promise { return { backend: 'matrix', note: 'DOM inspection is not available on the Matrix backend — chat data is sourced from the Reddit Matrix homeserver API directly.', }; } /** Public so bootstrap can swap in a fresh client after token refresh. */ setMatrixClient(client: MatrixClient): void { this.matrix = client; } /** * Active invitation cooldown (i.e. Reddit's "you've created a lot of chats, * take a break" throttle), or null if we're clear to send. Auto-expires * by wall clock — callers don't have to clear it manually. */ getInvitationCooldown(): InvitationCooldown | null { if (!this.invitationCooldown) return null; if (Date.now() >= this.invitationCooldown.until) { this.invitationCooldown = null; this.cache.setCooldown(null); void this.cache.persistCooldown(); return null; } return this.invitationCooldown; } /** Record a cooldown after a `MatrixRateLimitError`. Test seam + reuse. */ setInvitationCooldown( error: MatrixRateLimitError, scope: NonNullable = this.inferCooldownScope(error.redditCode), ): void { const now = Date.now(); // Add a small safety margin so we don't probe again the very second // the homeserver thinks the window expired. const margin = 5_000; this.invitationCooldown = { until: now + error.retryAfterMs + margin, scope, redditCode: error.redditCode, reason: scope === 'room-creation' ? 'Reddit is throttling new chat invitations; replies in existing chats can continue.' : `Matrix rate-limited: ${error.errcode ?? '429'}`, setAt: now, }; const minutes = Math.ceil(error.retryAfterMs / 60_000); this.logger.warn( `[chat] invitation cooldown set: ${this.invitationCooldown.reason} (~${minutes}m, code=${error.redditCode ?? error.errcode ?? '429'})`, ); this.cache.setCooldown(this.invitationCooldown); void this.cache.persistCooldown(); } private blocksExistingRoomSends(cooldown: InvitationCooldown): boolean { return (cooldown.scope ?? this.inferCooldownScope(cooldown.redditCode)) === 'all-sends'; } private inferCooldownScope(redditCode: string | null): NonNullable { return redditCode === 'rate.room_creation_limit' || redditCode === 'rate.invitation_limit_mature_account' ? 'room-creation' : 'all-sends'; } // ------------ internals ------------ /** Single delta /sync that updates cache + nextBatch and returns the union. */ private async syncOnce(): Promise<{ chats: ActiveChat[]; changedRooms: number }> { const sync = await this.withAuth(() => this.matrix.sync({ since: this.cache.getMarkers().nextBatch ?? undefined, timeoutMs: 0 }), ); const joined = sync.rooms?.join ?? {}; const existing = new Map(this.cache.get().map((c) => [c.roomId!, c] as const)); const knownUsernames = await this.resolveUsernamesForRooms(Object.keys(joined), joined); for (const [roomId, room] of Object.entries(joined)) { const prev = existing.get(roomId); const next = this.roomToActiveChat(roomId, room, { knownUsernames }, prev); existing.set(roomId, next); } for (const [roomId, room] of Object.entries(joined)) { const chat = existing.get(roomId); if (!chat) continue; // A limited timeline means Matrix omitted events between the saved // cursor and this response. Walk the room history before advancing the // cursor so media in that gap cannot become permanently unreachable. if (room?.timeline?.limited) { this.logger.info(`[chat] delta timeline limited for ${roomId}; backfilling room history`); await this.loadDetails(chat); continue; } const scraped = this.scrapeMessageEvents(room?.timeline?.events ?? []); if (scraped.length === 0) continue; let newMessages: ChatMessage[] = scraped.map( ({ mediaUrls: _ignored, ...message }) => message, ); if (this.mediaService && scraped.some((message) => message.mediaUrls.length > 0)) { newMessages = await this.mediaService.attachMedia(chat, scraped); } this.mergeNewMessages(chat, newMessages); this.recordConversationPerformance(chat); } const merged = Array.from(existing.values()).sort( (a, b) => b.lastMessageTime - a.lastMessageTime, ); this.cache.set(merged); this.cache.setNextBatch(sync.next_batch); this.cache.markDeltaSync(); await this.cache.save(merged); // Resolve any t2_-prefixed usernames the delta just introduced, so the // dashboard doesn't show raw Matrix-style IDs for new rooms. await this.enrichUsernames(); await this.repairDirectRoomMappings(merged); return { chats: merged, changedRooms: Object.keys(joined).length }; } /** * Reconcile cached one-to-one chats with Matrix account data. The durable * username repository lets delta sync repair older rooms even when Reddit * omits their member state from the latest /sync response. */ private async repairDirectRoomMappings(chats: ActiveChat[]): Promise { const mappings: Array<{ recipientMxid: string; roomId: string }> = []; const seen = new Set(); for (const chat of chats) { if (!chat.roomId) continue; const rawT2 = this.extractT2(chat); const persistedT2 = rawT2 ? `t2_${rawT2}` : this.redditUsersRepo?.getByUsername(chat.username)?.t2Id; if (!persistedT2) continue; const recipientMxid = MatrixClient.t2ToMxid(persistedT2); const key = `${recipientMxid}|${chat.roomId}`; if (seen.has(key)) continue; seen.add(key); mappings.push({ recipientMxid, roomId: chat.roomId }); } if (mappings.length === 0) return; try { const repaired = await this.withAuth(() => this.matrix.repairDirectRoomMappings(mappings), ); if (repaired > 0) { this.logger.info(`[chat] repaired ${repaired} missing direct-room mapping(s)`); } } catch (error) { // Sync remains useful even if Reddit temporarily rejects account-data // maintenance. A later delta/full sync will try the additive repair again. this.logger.warn('[chat] direct-room mapping repair failed:', error); } } private scrapeMessageEvents(events: MatrixTimelineEvent[]): ScrapedChatMessage[] { return events .filter((e) => e.type === 'm.room.message') .map((e) => { const body = (e.content.body as string | undefined) ?? ''; const mediaUrls = this.extractMediaUrls(e); return { eventId: e.event_id, sender: e.sender === this.matrix.userId ? 'Me' : this.shortenSender(e.sender), text: body || (mediaUrls.length > 0 ? '[Image]' : ''), timestamp: e.origin_server_ts, isSelf: e.sender === this.matrix.userId, mediaUrls, }; }); } private extractMediaUrls(event: MatrixTimelineEvent): string[] { if (typeof event.content.url === 'string' && event.content.url.length > 0) { return [event.content.url]; } const thumbnail = event.content.thumbnail_url ?? event.content.info?.thumbnail_url; return typeof thumbnail === 'string' && thumbnail.length > 0 ? [thumbnail] : []; } private mergeNewMessages(chat: ActiveChat, newMessages: ChatMessage[]): void { if (newMessages.length === 0) return; 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(); const fallbackKeys = new Set(); 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 { if (!chat.roomId) { this.logger.warn(`[chat] no roomId for ${chat.roomName}, skipping details`); return; } const events: MatrixTimelineEvent[] = []; let from: string | undefined = undefined; const seenPaginationTokens = new Set(); while (true) { const res = await this.withAuth(() => this.matrix.getRoomMessages(chat.roomId!, { from, limit: 100, dir: 'b' }), ); events.push(...res.chunk); if (!res.end || res.chunk.length === 0) break; if (res.end === from || seenPaginationTokens.has(res.end)) { throw new Error(`Matrix history pagination stalled for ${chat.roomId}`); } seenPaginationTokens.add(res.end); from = res.end; } // Pagination yields newest-first; flip to chronological. const messages = events.reverse().filter((e) => e.type === 'm.room.message'); const scraped: ScrapedChatMessage[] = messages.map((e) => { const body = (e.content.body as string | undefined) ?? ''; const mediaUrls = this.extractMediaUrls(e); return { eventId: e.event_id, sender: e.sender === this.matrix.userId ? 'Me' : this.shortenSender(e.sender), text: body || (mediaUrls.length > 0 ? '[Image]' : ''), timestamp: e.origin_server_ts, isSelf: e.sender === this.matrix.userId, mediaUrls, }; }); // Backfill the recipient from message senders BEFORE attachMedia stamps // it into media records. /sync's heroes + state.events miss inactive // rooms; the senders of historical messages don't. if (!chat.username || chat.username === 'unknown' || chat.username.startsWith('@')) { const firstOther = messages.find((e) => e.sender !== this.matrix.userId); if (firstOther) { const t2 = firstOther.sender.match(/^@(t2_[^:]+):/)?.[1]; if (t2) { const resolved = this.t2ToUsername.get(t2.replace(/^t2_/, '')) ?? t2; chat.username = resolved; if (!chat.roomName || chat.roomName === 'unknown') chat.roomName = resolved; } } } 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}`); return; } const newest = chat.messages[chat.messages.length - 1]; chat.lastMessageSnippet = newest.text; chat.lastMessageTime = newest.timestamp; // Re-derive authorship from the authoritative history. Room summaries built // from /sync default `isFromUs` to false when the room carries no timeline // events (Reddit omits them for idle rooms), which would otherwise leave an // unanswered opener of ours sitting in the operator's "Needs reply" queue. chat.isFromUs = newest.isSelf; const firstSelf = chat.messages.find((m) => m.isSelf); if (firstSelf) chat.originMessage = firstSelf.text; this.recordConversationPerformance(chat); } /** * Project a cached or newly-synced conversation into durable performance * state. Safe to call repeatedly: outcome rows are upserted and successful * examples are de-duplicated by their original message. */ private recordConversationPerformance(chat: ActiveChat): boolean { const messages = chat.messages ?? []; if (!chat.roomId || !this.messagesRepo) return false; let performanceMessage: SentMessageRecord | undefined; try { const existing = this.messagesRepo.getByChatRoomId(chat.roomId); const hasPerformanceContext = Boolean( existing?.postId || existing?.strategyId || existing?.subreddit, ); if (existing && hasPerformanceContext) performanceMessage = existing; } catch (err) { this.logger.warn('[chat] failed to associate reply with performance context:', err); return false; } if (!performanceMessage) return false; const firstSelf = messages.find( (message) => message.isSelf && message.timestamp >= performanceMessage.sentAt - 1_000, ); const conversationStart = firstSelf?.timestamp ?? performanceMessage.sentAt; // Ignore messages from an older exchange in a reused room. The one-second // tolerance accommodates the optimistic opener timestamp being reconciled // with Matrix's authoritative event timestamp. Legacy caches can omit the // opener, so the durable sent timestamp is the fallback boundary. const conversationMessages = messages.filter( (message) => message.timestamp >= conversationStart - 1_000, ); const incomingMessages = conversationMessages.filter((message) => !message.isSelf); const hasReply = incomingMessages.length > 0; const incomingMediaCount = incomingMessages.reduce( (total, message) => total + (message.media?.length ?? 0), 0, ); const hasReceivedImage = incomingMediaCount > 0; const firstAttributedReply = hasReply && !performanceMessage.gotResponse; if (hasReceivedImage && !performanceMessage.gotImage) { this.messagesRepo.markGotImage(performanceMessage.id); // Guarded by the durable `gotImage` flag, so re-syncing a conversation // cannot re-notify for media that was already reported. try { this.mediaReceivedListener?.({ conversationId: chat.roomId, username: chat.username, count: incomingMediaCount, }); } catch (err) { this.logger.warn('[chat] mediaReceivedListener threw:', err); } } if (hasReply && this.successfulMessagesRepo && this.outcomesRepo) { const firstReply = incomingMessages[0]; const latestReply = incomingMessages.at(-1); const cachedSelfCount = conversationMessages.filter((message) => message.isSelf).length; const selfCount = cachedSelfCount + (firstSelf ? 0 : 1); const otherCount = incomingMessages.length; const wasMissingOutcome = this.outcomesRepo.get(performanceMessage.id) === undefined; const conversationState = this.conversationStateRepo?.get(chat.roomId); try { recordConversationReply( performanceMessage, latestReply?.text.slice(0, 200) ?? '', { messageCount: selfCount + otherCount, reciprocal: selfCount >= 2 && otherCount >= 2, responses: incomingMessages.map((message) => message.text), learningEligible: conversationState !== undefined && conversationState.automation.status !== 'active', receivedImage: hasReceivedImage, mediaWasRequested: conversationState?.askedForMedia ?? conversationMessages.some((message) => message.isSelf && containsMediaRequest(message.text)), authoritativeTranscript: true, }, { messages: this.messagesRepo, successful: this.successfulMessagesRepo, outcomes: this.outcomesRepo, logger: this.logger, }, ); if (firstAttributedReply) { try { this.newReplyListener?.({ conversationId: chat.roomId, replyId: `${chat.roomId}:${firstReply?.timestamp ?? Date.now()}`, username: chat.username, replySnippet: firstReply?.text.slice(0, 200) ?? '', }); } catch (err) { this.logger.warn('[chat] newReplyListener threw:', err); } } } catch (err) { this.logger.warn('[chat] failed to record conversation outcome:', err); return false; } this.conversationLearningObserver?.observe(chat, performanceMessage, conversationState); return wasMissingOutcome; } else if (hasReply && !performanceMessage.gotResponse) { this.messagesRepo.markGotResponse(performanceMessage.id); this.logger.warn('[chat] response recorded without canonical outcome repository; excluded from learning'); } this.conversationLearningObserver?.observe( chat, performanceMessage, this.conversationStateRepo?.get(chat.roomId), ); return false; } /** Repair cached replies that predate canonical outcome recording. */ private backfillConversationOutcomes(): void { let repaired = 0; for (const chat of this.cache.get()) { if (this.recordConversationPerformance(chat)) repaired += 1; } if (repaired > 0) { this.logger.info(`[chat] backfilled ${repaired} missing conversation outcome(s)`); } } /** * Project a Matrix room (from /sync) into an ActiveChat. If `prev` is * supplied, fields it carries (messages, pendingReply, originMessage) * are preserved. */ private roomToActiveChat( roomId: string, room: MatrixJoinedRoom | undefined, opts: BuildChatOpts, prev?: ActiveChat, ): ActiveChat { const timeline = room?.timeline?.events ?? []; const lastMessage = [...timeline].reverse().find((e) => e.type === 'm.room.message'); // Prefer summary.m.heroes (cheap), but fall back to scanning the room's // member-join state events — Reddit often omits heroes for older rooms // even though the recipient is in state.events as an m.room.member join. const heroes = (room?.summary?.['m.heroes'] ?? []).filter((h) => h !== this.matrix.userId); const memberJoins: string[] = []; for (const ev of room?.state?.events ?? []) { if (ev.type !== 'm.room.member') continue; const stateKey = (ev as { state_key?: string }).state_key; const membership = (ev.content as { membership?: string }).membership; // Include 'invite' too — for outgoing chats the recipient never // accepted, their MXID is only in state as an invite, not a join. // chat.reddit.com's web UI reads the same to display the invitee. if ( (membership === 'join' || membership === 'invite') && stateKey && stateKey !== this.matrix.userId ) { memberJoins.push(stateKey); } } // Reddit's homeserver routinely omits heroes/member state from /sync // for chats that have been idle for a while — but the timeline events // it does return always carry full MXIDs in their `sender` field, so // mine those too. const timelineSenders = timeline .map((e) => e.sender) .filter((s): s is string => !!s && s !== this.matrix.userId); const candidateMxids = [...new Set([...heroes, ...memberJoins, ...timelineSenders])]; const otherT2 = candidateMxids .map((h) => h.match(/^@(t2_[^:]+):/)?.[1]) .filter((s): s is string => !!s); let username = prev?.username ?? null; if ( !username || username === 'unknown' || username.startsWith('t2_') || username.startsWith('@') ) { const resolved = otherT2 .map((id) => opts.knownUsernames.get(id.replace(/^t2_/, ''))) .find((x): x is string => !!x); if (resolved) username = resolved; else if (otherT2[0]) username = otherT2[0]; } if (!username) username = prev?.username ?? 'unknown'; const nameEvent = room?.state?.events?.find((e) => e.type === 'm.room.name'); let roomName = prev?.roomName ?? username; if (nameEvent && typeof nameEvent.content.name === 'string' && nameEvent.content.name) { roomName = nameEvent.content.name as string; } else if (!prev) { roomName = username; } let lastMessageTime = prev?.lastMessageTime ?? 0; let lastMessageSnippet = prev?.lastMessageSnippet ?? ''; let isFromUs = prev?.isFromUs ?? false; if (lastMessage) { lastMessageTime = lastMessage.origin_server_ts; const body = (lastMessage.content.body as string | undefined) ?? ''; lastMessageSnippet = body || (lastMessage.content.url ? '[Image]' : ''); isFromUs = lastMessage.sender === this.matrix.userId; } const timelineMessages = opts.includeTimelineMessages && !prev?.messages ? this.scrapeMessageEvents(timeline).map(({ mediaUrls: _ignored, ...message }) => message) : prev?.messages; const unread = (room?.unread_notifications?.notification_count ?? 0) > 0; return { ...(prev ?? {}), username, roomName, lastMessageTime, lastMessageSnippet, isFromUs, unread, url: ROOM_URL_PREFIX + encodeURIComponent(roomId), roomId, ...(timelineMessages ? { messages: timelineMessages } : {}), }; } /** Returns the t2 id portion of a stored `t2_xxx` username. */ private extractT2(chat: ActiveChat): string | null { const m = chat.username?.match(/^t2_(.+)$/); return m ? m[1] : null; } /** * Convert a Matrix sender into a display label. Prefers the resolved Reddit * username when known (so message bubbles read "alice" rather than * "t2_abc123"); falls back to the bare t2 form, then the raw MXID. */ private shortenSender(mxid: string): string { const m = mxid.match(/^@([^:]+):/); const bare = m ? m[1] : mxid; const t2 = bare.replace(/^t2_/, ''); return this.t2ToUsername.get(t2) ?? bare; } /** * Best-effort batch resolution of t2 IDs to usernames. Queries Reddit's * `alfred` API to look up usernames from t2 IDs, and also resolves * room-state MXIDs via `UsernameResolver`. Uses the in-memory cache * as a fallback. Failures are logged and ignored, with the t2 ID * staying as the fallback display. */ private async resolveUsernamesForRooms( roomIds: string[], rooms: Record = {}, ): Promise> { const cached = new Map(this.t2ToUsername); const candidateT2s = new Set(cached.keys()); for (const roomId of roomIds) { const room = rooms[roomId]; const mxids = [ ...(room?.summary?.['m.heroes'] ?? []), ...(room?.state?.events ?? []).map((event) => (event as MatrixTimelineEvent & { state_key?: string }).state_key, ), ...(room?.timeline?.events ?? []).map((event) => event.sender), ]; for (const mxid of mxids) { if (!mxid || mxid === this.matrix.userId) continue; const t2 = mxid.match(/^@t2_([^:]+):/)?.[1]; if (t2) candidateT2s.add(t2); } } let staleT2s = Array.from(candidateT2s).filter(id => { const name = cached.get(id); return !name || name.startsWith('t2_') || name.startsWith('@') || this.isStaleUsername(name); }); // Matrix already knows the display name attached to each room member, so // prefer that authenticated source before trying Reddit's legacy lookup. if (staleT2s.length > 0 && this.usernameResolver) { try { const resolved = await this.usernameResolver.resolveT2Ids( staleT2s.map((id) => `t2_${id}`), ); for (const [t2, name] of Object.entries(resolved)) { const bare = t2.replace(/^t2_/, ''); cached.set(bare, name); this.t2ToUsername.set(bare, name); this.persistUsername(t2, name); } } catch (err) { this.logger.debug('[chat] MXID resolution failed:', err); } staleT2s = staleT2s.filter((id) => { const name = cached.get(id); return !name || this.isStaleUsername(name); }); } // Fall back to Reddit's alfred API for profiles without a Matrix display // name. This is best-effort and leaves the stable t2 ID visible on error. if (staleT2s.length > 0) { const numericIds = staleT2s.map(id => id.replace(/^t2_/, '')); try { const res = await fetchWithTimeout( 'https://www.reddit.com/alfred/users/by_id.json', { method: 'POST', headers: { 'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8', 'User-Agent': 'reddit-bot-chat/1.0', 'X-Requested-With': 'XMLHttpRequest', }, body: new URLSearchParams({ alchemyType: 'Account', ids: numericIds.join(','), currencyType: '1', max_rating: '100', size: '100', }).toString(), timeoutMs: 15_000, }, ); if (res.ok) { const body = await res.json() as { results?: Array> }; for (const obj of body.results ?? []) { const t2 = `t2_${obj.Id ?? obj.id ?? ''}`; const idKey = t2.replace(/^t2_/, ''); const displayName = (obj.name ?? obj.alias ?? '') as string; if (displayName) { cached.set(idKey, displayName); this.t2ToUsername.set(idKey, displayName); this.persistUsername(idKey, displayName); for (const chat of this.cache.get()) { if (chat.username === `t2_${idKey}`) { chat.username = displayName; if (!chat.roomName || chat.roomName.startsWith('t2_')) { chat.roomName = displayName; } } } } } } } catch (err) { this.logger.debug('[chat] alfred reverse-resolution failed:', err); } } return cached; } /** Return true if `name` looks like it isn't a real Reddit username. */ private isStaleUsername(name: string): boolean { return !name || /^[t2_0-9@:]{3,}$/.test(name); } /** * After a sync, look up real usernames for any t2-only entries in the * cache. Best-effort; safe to await or fire-and-forget. */ async enrichUsernames(): Promise { if (!this.accountAuth?.isFullyAuthenticated()) { this.logger.debug('[chat] username enrichment skipped until full account authentication'); return; } if (!this.usernameResolver) return; let dirty = false; // Pass 1 — upgrade any t2_-prefixed chat whose t2 we *already* know how // to resolve. Without this, a delta that re-creates a t2-only chat // wouldn't get upgraded until the bot restarts, because the resolver // below skips known mappings. for (const chat of this.cache.get()) { const t2 = this.extractT2(chat); if (!t2) continue; const known = this.t2ToUsername.get(t2); if (known && known !== chat.username) { chat.username = known; if (!chat.roomName || chat.roomName.startsWith('t2_') || chat.roomName === 'unknown') { chat.roomName = known; } dirty = true; } } // Pass 2 — chats still stuck on 'unknown' (or a bare MXID) didn't get a // t2 ID from /sync heroes/state/timeline. Try /state first — it returns // every m.room.member event including 'invite'-membership ones, which is // how chat.reddit.com's web UI names rooms the recipient never accepted. // Fall back to mining a /messages page if /state somehow yields nothing. for (const chat of this.cache.get()) { if (!chat.roomId) continue; if (this.unresolvableRooms.has(chat.roomId)) continue; const needsLookup = !chat.username || chat.username === 'unknown' || chat.username.startsWith('@'); if (!needsLookup) continue; let t2: string | undefined; if (this.roomStateLookupAvailable) { try { const state = await this.withAuth(() => this.matrix.getRoomFullState(chat.roomId!)); const otherMember = state.find((e) => { if (e.type !== 'm.room.member') return false; const sk = (e as { state_key?: string }).state_key; const membership = (e.content as { membership?: string }).membership; return ( sk && sk !== this.matrix.userId && (membership === 'join' || membership === 'invite') ); }) as { state_key?: string } | undefined; t2 = otherMember?.state_key?.match(/^@(t2_[^:]+):/)?.[1]; } catch (err) { if (err instanceof MatrixAuthenticationUnavailableError) { this.logger.warn('[chat] username enrichment paused until Reddit login is restored'); return; } if (err instanceof MatrixFeatureDisabledError) { this.roomStateLookupAvailable = false; this.logger.debug('[chat] room state lookup is disabled upstream; using message history instead'); } else { this.logger.debug(`[chat] /state lookup failed for ${chat.roomId}:`, err); } } } if (!t2) { try { const res = await this.withAuth(() => this.matrix.getRoomMessages(chat.roomId!, { limit: 20, dir: 'b' }), ); const otherSender = res.chunk?.find( (e) => e.sender && e.sender !== this.matrix.userId, )?.sender; t2 = otherSender?.match(/^@(t2_[^:]+):/)?.[1]; } catch (err) { if (err instanceof MatrixAuthenticationUnavailableError) { this.logger.warn('[chat] username enrichment paused until Reddit login is restored'); return; } this.logger.debug(`[chat] /messages lookup failed for ${chat.roomId}:`, err); } } if (t2) { const bare = t2.replace(/^t2_/, ''); const known = this.t2ToUsername.get(bare); chat.username = known ?? `t2_${bare}`; if (!chat.roomName || chat.roomName === 'unknown' || chat.roomName.startsWith('@')) { chat.roomName = chat.username; } dirty = true; } else { // Neither /state nor /messages surfaced a t2 — give up on this room // so we don't keep hammering the homeserver every sync. this.unresolvableRooms.add(chat.roomId); } } // Pass 3 — for every chat still on a raw t2 ID, batch-resolve via the // injected resolver (www.reddit.com/api/user_data_by_account_ids.json). const needed: string[] = []; for (const chat of this.cache.get()) { const t2 = this.extractT2(chat); if (t2 && !this.t2ToUsername.has(t2)) needed.push(`t2_${t2}`); } if (needed.length > 0) { try { const got = await this.usernameResolver.resolveT2Ids(needed); let updated = 0; for (const [t2, name] of Object.entries(got)) { const bare = t2.replace(/^t2_/, ''); if (!name) continue; this.t2ToUsername.set(bare, name); this.persistUsername(t2, name); for (const chat of this.cache.get()) { if (chat.username === `t2_${bare}`) { chat.username = name; chat.roomName = chat.roomName?.startsWith('t2_') ? name : chat.roomName; updated++; } } } if (updated > 0) { this.logger.info(`[chat] enriched ${updated} chats with usernames`); dirty = true; } } catch (err) { this.logger.debug('[chat] enrichUsernames failed:', err); } } if (dirty) await this.cache.save(); } /** Resolve an account ID from listing data, the durable cache, or Reddit auth. */ private async resolveUsernameToT2( username: string, knownT2Id?: string, ): Promise { if (knownT2Id && /^t2_[a-z0-9]+$/i.test(knownT2Id)) return knownT2Id; const cached = this.redditUsersRepo?.getByUsername(username); if (cached) return cached.t2Id; if (this.usernameResolver?.resolveUsernameToT2) { return this.usernameResolver.resolveUsernameToT2(username); } return null; } /** Parse a chat URL like `…/room/!abc%3Areddit.com` → `!abc:reddit.com`. */ static extractRoomId(chatUrlOrRoomId: string): string | null { if (chatUrlOrRoomId.startsWith('!')) return chatUrlOrRoomId; const m = chatUrlOrRoomId.match(/\/room\/([^/?#]+)/); if (!m) return null; try { return decodeURIComponent(m[1]); } catch { return null; } } }