feat: persist settings and enrich conversation context

This commit is contained in:
jjbrunton committed 2026-08-08 18:18:28 +01:00
1 parent a36723a325
commit 2c100d72b6
84 files changed
+2248 -644

No files matched your search

+2
View File
@@ -5,6 +5,7 @@ export type {
SyncStatus,
SyncProgress,
SyncPhase,
SyncCompletion,
SyncProgressListener,
} from './service.js';
export type {
@@ -23,3 +24,4 @@ export {
MatrixTokenRefresher,
} from './matrix/token-refresher.js';
export { MatrixMediaDownloader } from './matrix/http-media-downloader.js';
export { MatrixUsernameResolver } from './matrix/username-resolver.js';
+46 -2
View File
@@ -190,7 +190,13 @@ export class MatrixClient {
}
if (res.status === 401 || res.status === 403) {
throw new MatrixAuthError(`Matrix ${method} ${path} → ${res.status} (token likely expired)`);
const failure = await readMatrixFailure(res);
if (isMatrixAuthenticationFailure(res.status, failure.errcode)) {
throw new MatrixAuthError(
`Matrix ${method} ${path} → ${res.status} ${failure.errcode ?? ''} (token rejected)`,
);
}
throw new Error(`Matrix ${method} ${path} → ${res.status}: ${failure.text.slice(0, 300)}`);
}
// Matrix-standard rate-limit response: { retry_after_ms } per spec.
@@ -251,6 +257,13 @@ export class MatrixClient {
return res.user_id;
}
async getUserProfile(mxid: string): Promise<{ displayname?: string; avatar_url?: string }> {
return this.request<{ displayname?: string; avatar_url?: string }>(
'GET',
`/_matrix/client/v3/profile/${encodeURIComponent(mxid)}`,
);
}
/**
* All rooms the user is currently joined to. Use this for backfill —
* `/sync` only returns rooms with recent activity.
@@ -502,7 +515,15 @@ export class MatrixClient {
throw new Error(`Matrix media download ${url} timed out (exceeded ${maxAttempts} attempts)`);
}
if (res.status === 401 || res.status === 403) {
throw new MatrixAuthError(`Matrix media download → ${res.status} (token likely expired)`);
const failure = await readMatrixFailure(res);
if (isMatrixAuthenticationFailure(res.status, failure.errcode)) {
throw new MatrixAuthError(
`Matrix media download → ${res.status} ${failure.errcode ?? ''} (token rejected)`,
);
}
throw new Error(
`Matrix media download ${url} → ${res.status}: ${failure.text.slice(0, 200)}`,
);
}
if (res.status === 429) {
const body = (await res.json().catch(() => ({}))) as { retry_after_ms?: number };
@@ -529,6 +550,29 @@ export class MatrixClient {
}
}
interface MatrixFailure {
errcode?: string;
text: string;
}
async function readMatrixFailure(response: Response): Promise<MatrixFailure> {
const text = await response.text().catch(() => '');
if (!text) return { text };
try {
const parsed = JSON.parse(text) as { errcode?: unknown };
return {
text,
...(typeof parsed.errcode === 'string' ? { errcode: parsed.errcode } : {}),
};
} catch {
return { text };
}
}
function isMatrixAuthenticationFailure(status: number, errcode?: string): boolean {
return status === 401 || errcode === 'M_UNKNOWN_TOKEN' || errcode === 'M_MISSING_TOKEN';
}
function parseMxc(mxc: string): { serverName: string; mediaId: string } | null {
const m = mxc.match(/^mxc:\/\/([^/]+)\/(.+)$/);
if (!m) return null;
@@ -16,6 +16,10 @@ export class MatrixMediaDownloader implements MediaDownloader {
private logger: Logger,
) {}
setMatrixClient(client: MatrixClient): void {
this.matrix = client;
}
async fetch(url: string): Promise<MediaDownloadResult | null> {
try {
if (url.startsWith('mxc://')) {
@@ -0,0 +1,29 @@
import type { UsernameResolver } from '../service.js';
interface MatrixProfileClient {
getUserProfile(mxid: string): Promise<{ displayname?: string }>;
}
export class MatrixUsernameResolver implements UsernameResolver {
constructor(private profiles: MatrixProfileClient) {}
setMatrixClient(client: MatrixProfileClient): void {
this.profiles = client;
}
async resolveT2Ids(t2Ids: string[]): Promise<Record<string, string>> {
const resolved: Record<string, string> = {};
await Promise.all(t2Ids.map(async (value) => {
const t2Id = value.startsWith('t2_') ? value : `t2_${value}`;
try {
const username = (await this.profiles.getUserProfile(`@${t2Id}:reddit.com`)).displayname?.trim();
if (username && !username.startsWith('t2_') && !username.startsWith('@')) {
resolved[t2Id] = username;
}
} catch {
// A missing profile should not prevent the rest of the batch resolving.
}
}));
return resolved;
}
}
+160 -77
View File
@@ -2,6 +2,7 @@ import type { Logger } from '../../utils/logger.js';
import type {
MessagesRepository,
RedditUsersRepository,
SentMessageRecord,
SuccessfulMessagesRepository,
} from '../../database/repositories.js';
import type { MediaService } from '../media.js';
@@ -40,10 +41,13 @@ class MatrixAuthenticationUnavailableError extends Error {
export interface UsernameResolver {
/** Map t2 IDs (with or without `t2_` prefix) to Reddit usernames. */
resolveT2Ids(t2Ids: string[]): Promise<Record<string, string>>;
/** Resolve a Reddit username to its t2 account ID when starting a DM. */
resolveUsernameToT2?(username: string): Promise<string | null>;
}
interface BuildChatOpts {
knownUsernames: Map<string, string>;
includeTimelineMessages?: boolean;
}
/**
@@ -98,8 +102,15 @@ export interface SyncStatus {
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) => void;
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: {
@@ -112,6 +123,7 @@ export type NewReplyListener = (info: {
export class ChatService {
private cache: ChatCache;
private queue: Promise<unknown> = Promise.resolve();
private deltaSyncInFlight: Promise<void> | null = null;
private readonly t2ToUsername = new Map<string, string>();
/** Rooms whose /joined_members lookup has already failed — don't retry. */
private readonly unresolvableRooms = new Set<string>();
@@ -263,6 +275,7 @@ export class ChatService {
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 merged = await this.syncOnce();
@@ -273,8 +286,11 @@ export class ChatService {
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();
this.endSync(failure);
}
});
}
@@ -284,9 +300,12 @@ export class ChatService {
* cache. The `limit` arg is preserved for API compatibility but Matrix
* already returns only rooms with new events, so it's a no-op.
*/
async refreshTopChats(_limit = 5): Promise<void> {
await this.enqueue(async () => {
refreshTopChats(_limit = 5): Promise<void> {
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 });
await this.syncOnce();
@@ -297,11 +316,19 @@ export class ChatService {
total: this.cache.get().length,
});
} catch (error) {
failure = error;
this.logger.error('[chat] refreshTopChats failed:', error);
throw error;
} finally {
this.endSync();
this.endSync(failure);
}
});
const tracked = operation.finally(() => {
this.deltaSyncInFlight = null;
});
this.deltaSyncInFlight = tracked;
return tracked;
}
/**
@@ -312,6 +339,7 @@ export class ChatService {
async fetchAllMessages(): Promise<ActiveChat[]> {
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();
@@ -326,24 +354,46 @@ export class ChatService {
const joinedIds = await this.withAuth(() => this.matrix.getJoinedRooms());
this.logger.info(`[chat] backfill: ${joinedIds.length} joined rooms`);
this.updateProgress({
phase: 'resolving-usernames',
current: 0,
total: joinedIds.length,
});
const knownUsernames = await this.resolveUsernamesForRooms(joinedIds);
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 });
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);
let done = 0;
for (const chat of chats) {
this.updateProgress({
@@ -377,10 +427,12 @@ export class ChatService {
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) throw error;
return this.cache.get();
} finally {
this.endSync();
this.endSync(failure);
}
});
}
@@ -446,12 +498,18 @@ export class ChatService {
this.emitProgress();
}
private endSync(): void {
private endSync(error?: unknown): void {
this.syncingKind = null;
this.progress = null;
if (this.progressListener) {
try {
this.progressListener(null);
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);
}
@@ -522,7 +580,11 @@ export class ChatService {
* 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): Promise<ActiveChat | null> {
async startConversation(
username: string,
message: string,
knownT2Id?: string,
): Promise<ActiveChat | null> {
if (!this.accountAuth?.isFullyAuthenticated()) return null;
const active = this.getInvitationCooldown();
if (active) {
@@ -534,7 +596,7 @@ export class ChatService {
}
return this.enqueue(async () => {
try {
const t2Id = await this.resolveUsernameToT2(username);
const t2Id = await this.resolveUsernameToT2(username, knownT2Id);
if (!t2Id) {
this.logger.error(`[chat] cannot start conversation: unknown user "${username}"`);
return null;
@@ -657,7 +719,7 @@ export class ChatService {
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));
const knownUsernames = await this.resolveUsernamesForRooms(Object.keys(joined), joined);
for (const [roomId, room] of Object.entries(joined)) {
const prev = existing.get(roomId);
@@ -786,38 +848,36 @@ export class ChatService {
if (firstSelf) chat.originMessage = firstSelf.text;
const hasReply = chat.messages.some((m) => !m.isSelf);
let performanceMessage: SentMessageRecord | undefined;
if (firstSelf && chat.roomId && this.messagesRepo) {
try {
const existing = this.messagesRepo.getByChatRoomId(chat.roomId);
const id = existing
? existing.id
: this.messagesRepo.insertChat({
recipient: chat.username,
subject: chat.username,
message: firstSelf.text,
sentAt: firstSelf.timestamp,
chatRoomId: chat.roomId,
});
if (hasReply && !existing?.gotResponse) {
this.messagesRepo.markGotResponse(id);
const hasPerformanceContext = Boolean(
existing?.postId || existing?.strategyId || existing?.subreddit,
);
if (existing && hasPerformanceContext) {
performanceMessage = existing;
if (hasReply && !existing.gotResponse) {
this.messagesRepo.markGotResponse(existing.id);
}
}
} catch (err) {
this.logger.warn('[chat] failed to record sent_message for stats:', err);
this.logger.warn('[chat] failed to associate reply with performance context:', err);
}
}
if (firstSelf && hasReply && this.successfulMessagesRepo) {
if (firstSelf && hasReply && performanceMessage && this.successfulMessagesRepo) {
if (!this.successfulMessagesRepo.existsByMessage(firstSelf.text)) {
const firstReply = chat.messages.find((m) => !m.isSelf);
try {
this.successfulMessagesRepo.insert({
sentMessageId: null,
sentMessageId: performanceMessage.id,
originalMessage: firstSelf.text,
imageAnalysis: null,
postTitle: `Chat with ${chat.username}`,
imageAnalysis: performanceMessage.postImageAnalysis ?? null,
postTitle: performanceMessage.postTitle ?? null,
responseSnippet: firstReply?.text.slice(0, 200) || null,
subreddit: null,
subreddit: performanceMessage.subreddit ?? null,
});
this.logger.info(`[chat] stored successful chat example from ${chat.username}`);
try {
@@ -916,6 +976,10 @@ export class ChatService {
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 {
@@ -928,6 +992,7 @@ export class ChatService {
unread,
url: ROOM_URL_PREFIX + encodeURIComponent(roomId),
roomId,
...(timelineMessages ? { messages: timelineMessages } : {}),
};
}
@@ -956,16 +1021,58 @@ export class ChatService {
* as a fallback. Failures are logged and ignored, with the t2 ID
* staying as the fallback display.
*/
private async resolveUsernamesForRooms(roomIds: string[]): Promise<Map<string, string>> {
// Collect all known t2 IDs (from the in-memory cache) and query
// Reddit's alfred API for reverse-resolution.
private async resolveUsernamesForRooms(
roomIds: string[],
rooms: Record<string, MatrixJoinedRoom> = {},
): Promise<Map<string, string>> {
const cached = new Map(this.t2ToUsername);
const candidateT2s = new Set(cached.keys());
const staleT2s = Array.from(cached.keys()).filter(id => {
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 {
@@ -1014,23 +1121,6 @@ export class ChatService {
}
}
// Also resolve MXID-based t2 IDs via the lifted UsernameResolver,
// while avoiding re-queries for IDs we already resolved via alfred.
const mxidT2s = Array.from(new Set(staleT2s.filter(id => !cached.get(id.replace(/^t2_/, '')))));
if (mxidT2s.length > 0 && this.usernameResolver) {
try {
const resolved = await this.usernameResolver.resolveT2Ids(mxidT2s);
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);
}
}
return cached;
}
@@ -1170,28 +1260,21 @@ export class ChatService {
if (dirty) await this.cache.save();
}
/**
* Look up the t2 ID for a Reddit username via the standard about.json
* endpoint. Returns `t2_xxx` or null.
*/
private async resolveUsernameToT2(username: string): Promise<string | null> {
try {
const res = await fetchWithTimeout(
`https://www.reddit.com/user/${encodeURIComponent(username)}/about.json`,
{ headers: { 'User-Agent': 'reddit-bot-chat/1.0' }, timeoutMs: 15_000 },
);
if (!res.ok) {
this.logger.warn(`[chat] /user/${username}/about.json → ${res.status}`);
return null;
}
const json = (await res.json()) as { data?: { id?: string } };
const id = json.data?.id;
if (!id) return null;
return id.startsWith('t2_') ? id : `t2_${id}`;
} catch (err) {
this.logger.error(`[chat] failed to resolve username "${username}":`, err);
return null;
/** Resolve an account ID from listing data, the durable cache, or Reddit auth. */
private async resolveUsernameToT2(
username: string,
knownT2Id?: string,
): Promise<string | null> {
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`. */
+1
View File
@@ -23,6 +23,7 @@ export interface ActiveChat {
roomId?: string;
messages?: ChatMessage[];
originMessage?: string;
hasReceivedImages?: boolean;
}
/**
+11 -1
View File
@@ -57,7 +57,17 @@ export class ResponseCheckerService {
results.push(result);
if (result.gotResponse) {
// Mark as got response
const hasPerformanceContext = Boolean(
message.postId || message.strategyId || message.subreddit,
);
if (!hasPerformanceContext) {
this.messagesRepo.markChecked(message.id);
this.logger.info(
`Response from u/${message.recipient} excluded from performance: no post, strategy, or subreddit context.`,
);
continue;
}
this.messagesRepo.markGotResponse(message.id);
// Get the original post info