Restructure chat service, web routes, and add test suite
Split monolithic chat service into modular structure with Matrix client support, extract web server into per-resource route modules, introduce bootstrap/main-loop separation, and add comprehensive vitest coverage across services, repositories, and utilities. Adds project docs and CLAUDE.md agent instructions.
This commit is contained in:
jjbrunton committed
1 parent
dfa0580859
commit
633bb0ec04
84 files changed
+19278
-3888
No files matched your search
@@ -1,199 +0,0 @@
|
||||
import type { Logger } from '../utils/logger.js';
|
||||
import type { Config } from '../config/index.js';
|
||||
import type { OpenRouterClient } from '../ai/openrouter.js';
|
||||
import type { ChatService, ActiveChat } from './chat.js';
|
||||
|
||||
export class AutoReplyService {
|
||||
private interval: NodeJS.Timeout | null = null;
|
||||
private processing = false;
|
||||
// Track which conversations we've already processed to avoid re-generating replies
|
||||
private processedConversations = new Set<string>();
|
||||
|
||||
constructor(
|
||||
private chatService: ChatService,
|
||||
private aiClient: OpenRouterClient,
|
||||
private config: Config,
|
||||
private logger: Logger,
|
||||
private onReplyGenerated?: (chat: ActiveChat) => void,
|
||||
private onUpsellDetected?: (chat: ActiveChat, keyword: string) => void,
|
||||
private onAutoSent?: (chat: ActiveChat) => void,
|
||||
) { }
|
||||
|
||||
async start(): Promise<void> {
|
||||
if (!this.config.autoReply.enabled) {
|
||||
this.logger.info('Auto-reply is disabled');
|
||||
return;
|
||||
}
|
||||
|
||||
// Seed processedConversations from cache so we don't re-generate on restart
|
||||
const cached = await this.chatService.getCachedConversations();
|
||||
for (const chat of cached) {
|
||||
if (chat.replyStatus) {
|
||||
const stateKey = `${chat.username}:${chat.lastMessageTime}:${chat.lastMessageSnippet}`;
|
||||
this.processedConversations.add(stateKey);
|
||||
}
|
||||
}
|
||||
this.logger.info(`Auto-reply: Seeded ${this.processedConversations.size} already-processed conversation(s) from cache`);
|
||||
|
||||
const intervalMs = this.config.autoReply.checkIntervalMs;
|
||||
this.logger.info(`Auto-reply service started (interval: ${intervalMs / 1000}s, autoSend: ${this.config.autoReply.autoSend})`);
|
||||
|
||||
// Run immediately then schedule
|
||||
this.processUnreadChats();
|
||||
this.interval = setInterval(() => this.processUnreadChats(), intervalMs);
|
||||
}
|
||||
|
||||
stop(): void {
|
||||
if (this.interval) {
|
||||
clearInterval(this.interval);
|
||||
this.interval = null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Manually trigger processing (e.g., after a chat refresh)
|
||||
*/
|
||||
async triggerProcess(): Promise<void> {
|
||||
if (!this.config.autoReply.enabled) return;
|
||||
await this.processUnreadChats();
|
||||
}
|
||||
|
||||
private async processUnreadChats(): Promise<void> {
|
||||
if (this.processing) return;
|
||||
this.processing = true;
|
||||
|
||||
try {
|
||||
const chats = await this.chatService.getCachedConversations();
|
||||
|
||||
// Find conversations where:
|
||||
// 1. The last message is NOT from us (they sent a message)
|
||||
// 2. The conversation is marked as unread OR the last message is from them
|
||||
// 3. We haven't already processed this state (using message snippet + time as key)
|
||||
const unreadChats = chats.filter(chat => {
|
||||
if (!chat.messages || chat.messages.length === 0) return false;
|
||||
if (chat.replyStatus) return false; // Already processed (pending, upsell_detected, or sent)
|
||||
|
||||
// The definitive check: the last message in the actual history must be from THEM, not us
|
||||
const lastMessage = chat.messages[chat.messages.length - 1];
|
||||
if (lastMessage.isSelf) return false;
|
||||
|
||||
// Create a unique key for this conversation state
|
||||
const stateKey = `${chat.username}:${chat.lastMessageTime}:${chat.lastMessageSnippet}`;
|
||||
if (this.processedConversations.has(stateKey)) return false;
|
||||
|
||||
return true;
|
||||
});
|
||||
|
||||
if (unreadChats.length === 0) {
|
||||
this.logger.debug('Auto-reply: No unread conversations to process');
|
||||
this.processing = false;
|
||||
return;
|
||||
}
|
||||
|
||||
this.logger.info(`Auto-reply: Processing ${unreadChats.length} unread conversation(s)...`);
|
||||
|
||||
for (const chat of unreadChats) {
|
||||
try {
|
||||
await this.processChat(chat);
|
||||
} catch (error) {
|
||||
this.logger.error(`Auto-reply: Error processing chat with ${chat.username}:`, error);
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
this.logger.error('Auto-reply: Error in processUnreadChats:', error);
|
||||
} finally {
|
||||
this.processing = false;
|
||||
}
|
||||
}
|
||||
|
||||
private async processChat(chat: ActiveChat): Promise<void> {
|
||||
const stateKey = `${chat.username}:${chat.lastMessageTime}:${chat.lastMessageSnippet}`;
|
||||
|
||||
this.logger.info(`Auto-reply: Checking conversation with ${chat.username}...`);
|
||||
|
||||
// Safety gate: verify the last message is truly from them, not us
|
||||
const lastMsg = chat.messages![chat.messages!.length - 1];
|
||||
if (lastMsg.isSelf) {
|
||||
this.logger.info(`Auto-reply: Skipping ${chat.username} — we sent the last message`);
|
||||
this.processedConversations.add(stateKey);
|
||||
return;
|
||||
}
|
||||
|
||||
// Step 1: Check for upsell/spam
|
||||
const upsellResult = this.aiClient.detectUpsell(
|
||||
chat.messages!.map(m => ({ text: m.text, isSelf: m.isSelf })),
|
||||
this.config.autoReply.upsellKeywords
|
||||
);
|
||||
|
||||
if (upsellResult.isUpsell) {
|
||||
this.logger.info(`Auto-reply: Upsell detected from ${chat.username} (keyword: "${upsellResult.matchedKeyword}")`);
|
||||
chat.replyStatus = 'upsell_detected';
|
||||
chat.pendingReply = undefined;
|
||||
this.processedConversations.add(stateKey);
|
||||
this.onUpsellDetected?.(chat, upsellResult.matchedKeyword!);
|
||||
|
||||
// Save the updated status to cache
|
||||
await this.chatService.saveCacheExternal();
|
||||
return;
|
||||
}
|
||||
|
||||
// Step 2: Generate a proposed reply
|
||||
this.logger.info(`Auto-reply: Generating reply for ${chat.username}...`);
|
||||
const suggestedReply = await this.aiClient.generateReply(
|
||||
chat.messages!.map(m => ({ sender: m.sender, text: m.text, isSelf: m.isSelf })),
|
||||
chat.username
|
||||
);
|
||||
|
||||
if (!suggestedReply) {
|
||||
this.logger.warn(`Auto-reply: Failed to generate reply for ${chat.username}`);
|
||||
this.processedConversations.add(stateKey);
|
||||
return;
|
||||
}
|
||||
|
||||
// Step 3: Auto-send or store as pending
|
||||
if (this.config.autoReply.autoSend) {
|
||||
if (chat.url) {
|
||||
this.logger.info(`Auto-reply: Auto-sending reply to ${chat.username}: "${suggestedReply.slice(0, 80)}..."`);
|
||||
const sent = await this.chatService.sendChatMessage(chat.url, suggestedReply);
|
||||
if (sent) {
|
||||
chat.replyStatus = 'sent';
|
||||
chat.pendingReply = suggestedReply;
|
||||
this.onAutoSent?.(chat);
|
||||
|
||||
// Refresh chats in background so cache reflects the sent message
|
||||
this.chatService.refreshTopChats(5).catch(err =>
|
||||
this.logger.error('Failed to refresh chats after auto-send:', err)
|
||||
);
|
||||
} else {
|
||||
this.logger.error(`Auto-reply: Failed to send message to ${chat.username}`);
|
||||
}
|
||||
} else {
|
||||
this.logger.warn(`Auto-reply: No chat URL for ${chat.username}, cannot auto-send`);
|
||||
}
|
||||
} else {
|
||||
// Store as pending approval
|
||||
chat.pendingReply = suggestedReply;
|
||||
chat.replyStatus = 'pending';
|
||||
this.logger.info(`Auto-reply: Reply pending approval for ${chat.username}: "${suggestedReply.slice(0, 80)}..."`);
|
||||
this.onReplyGenerated?.(chat);
|
||||
}
|
||||
|
||||
this.processedConversations.add(stateKey);
|
||||
|
||||
// Save the updated status to cache
|
||||
await this.chatService.saveCacheExternal();
|
||||
}
|
||||
|
||||
/**
|
||||
* Clear the processed state for a conversation so it can be re-evaluated
|
||||
*/
|
||||
clearProcessedState(username: string): void {
|
||||
const keysToRemove: string[] = [];
|
||||
for (const key of this.processedConversations) {
|
||||
if (key.startsWith(`${username}:`)) {
|
||||
keysToRemove.push(key);
|
||||
}
|
||||
}
|
||||
keysToRemove.forEach(k => this.processedConversations.delete(k));
|
||||
}
|
||||
}
|
||||
+1
-1178
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,108 @@
|
||||
import fs from 'fs/promises';
|
||||
import path from 'path';
|
||||
import type { Logger } from '../../utils/logger.js';
|
||||
import type { ActiveChat, ChatCacheFile } from './types.js';
|
||||
|
||||
export interface SyncMarkers {
|
||||
nextBatch: string | null;
|
||||
lastFullSyncAt: number | null;
|
||||
lastDeltaSyncAt: number | null;
|
||||
}
|
||||
|
||||
/**
|
||||
* In-memory + on-disk cache of Reddit chats plus the Matrix delta cursor
|
||||
* so the bot can resume incremental sync across restarts.
|
||||
*/
|
||||
export class ChatCache {
|
||||
private chats: ActiveChat[] = [];
|
||||
private lastFetchTime = 0;
|
||||
readonly cacheDurationMs = 60 * 1000;
|
||||
private readonly filePath: string;
|
||||
|
||||
private nextBatch: string | null = null;
|
||||
private lastFullSyncAt: number | null = null;
|
||||
private lastDeltaSyncAt: number | null = null;
|
||||
|
||||
constructor(private logger: Logger, filePath?: string) {
|
||||
this.filePath = filePath ?? path.join(process.cwd(), 'data', 'chat-cache.json');
|
||||
}
|
||||
|
||||
get(): ActiveChat[] {
|
||||
return this.chats;
|
||||
}
|
||||
|
||||
set(chats: ActiveChat[]): void {
|
||||
this.chats = chats;
|
||||
this.lastFetchTime = Date.now();
|
||||
}
|
||||
|
||||
isFresh(): boolean {
|
||||
return Date.now() - this.lastFetchTime < this.cacheDurationMs && this.chats.length > 0;
|
||||
}
|
||||
|
||||
getMarkers(): SyncMarkers {
|
||||
return {
|
||||
nextBatch: this.nextBatch,
|
||||
lastFullSyncAt: this.lastFullSyncAt,
|
||||
lastDeltaSyncAt: this.lastDeltaSyncAt,
|
||||
};
|
||||
}
|
||||
|
||||
setNextBatch(token: string | null): void {
|
||||
this.nextBatch = token;
|
||||
}
|
||||
|
||||
markFullSync(at: number = Date.now()): void {
|
||||
this.lastFullSyncAt = at;
|
||||
this.lastDeltaSyncAt = at;
|
||||
}
|
||||
|
||||
markDeltaSync(at: number = Date.now()): void {
|
||||
this.lastDeltaSyncAt = at;
|
||||
}
|
||||
|
||||
async load(): Promise<void> {
|
||||
try {
|
||||
const raw = await fs.readFile(this.filePath, 'utf-8');
|
||||
const parsed = JSON.parse(raw) as ChatCacheFile;
|
||||
if (!parsed || !Array.isArray(parsed.chats)) {
|
||||
this.logger.warn('Chat cache file is invalid, ignoring.');
|
||||
return;
|
||||
}
|
||||
this.chats = parsed.chats;
|
||||
this.lastFetchTime = Date.now();
|
||||
this.nextBatch = parsed.nextBatch ?? null;
|
||||
this.lastFullSyncAt = parsed.lastFullSyncAt ?? null;
|
||||
this.lastDeltaSyncAt = parsed.lastDeltaSyncAt ?? null;
|
||||
this.logger.info(
|
||||
`Loaded ${parsed.chats.length} chats from cache` +
|
||||
(this.lastFullSyncAt
|
||||
? ` (last full sync ${new Date(this.lastFullSyncAt).toISOString()})`
|
||||
: ' (no full sync recorded)'),
|
||||
);
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code === 'ENOENT') {
|
||||
this.logger.info('No chat cache found to load');
|
||||
return;
|
||||
}
|
||||
this.logger.warn('Failed to load chat cache:', error);
|
||||
}
|
||||
}
|
||||
|
||||
async save(chats: ActiveChat[] = this.chats): Promise<void> {
|
||||
const payload: ChatCacheFile = {
|
||||
version: 1,
|
||||
savedAt: Date.now(),
|
||||
chats,
|
||||
nextBatch: this.nextBatch,
|
||||
lastFullSyncAt: this.lastFullSyncAt,
|
||||
lastDeltaSyncAt: this.lastDeltaSyncAt,
|
||||
};
|
||||
try {
|
||||
await fs.mkdir(path.dirname(this.filePath), { recursive: true });
|
||||
await fs.writeFile(this.filePath, JSON.stringify(payload, null, 2), 'utf-8');
|
||||
} catch (error) {
|
||||
this.logger.warn('Failed to save chat cache:', error);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,24 @@
|
||||
export { ChatService } from './service.js';
|
||||
export type {
|
||||
TokenRefresher,
|
||||
UsernameResolver,
|
||||
SyncStatus,
|
||||
SyncProgress,
|
||||
SyncPhase,
|
||||
SyncProgressListener,
|
||||
} from './service.js';
|
||||
export type {
|
||||
ActiveChat,
|
||||
ChatMessage,
|
||||
ChatMedia,
|
||||
ChatCacheFile,
|
||||
ScrapedChatMessage,
|
||||
ScrapedRoom,
|
||||
} from './types.js';
|
||||
export { MatrixClient, MatrixAuthError } from './matrix/client.js';
|
||||
export type { MatrixCredentials } from './matrix/types.js';
|
||||
export {
|
||||
extractMatrixCredentials,
|
||||
loadCachedCredentials,
|
||||
} from './matrix/token-extractor.js';
|
||||
export { MatrixMediaDownloader } from './matrix/http-media-downloader.js';
|
||||
@@ -0,0 +1,529 @@
|
||||
import type { Logger } from '../../../utils/logger.js';
|
||||
import type {
|
||||
MatrixCredentials,
|
||||
MatrixSyncResponse,
|
||||
MatrixMessagesResponse,
|
||||
MatrixTimelineEvent,
|
||||
MatrixJoinedRoom,
|
||||
} from './types.js';
|
||||
|
||||
export class MatrixAuthError extends Error {
|
||||
constructor(message: string) {
|
||||
super(message);
|
||||
this.name = 'MatrixAuthError';
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Thrown when Reddit's homeserver returns a 429 with a `retry_after_ms`
|
||||
* longer than we're willing to block on. Caller (typically ChatService)
|
||||
* is expected to record this as a cooldown and stop trying until the
|
||||
* window expires.
|
||||
*/
|
||||
export class MatrixRateLimitError extends Error {
|
||||
constructor(
|
||||
message: string,
|
||||
public retryAfterMs: number,
|
||||
public errcode: string | null,
|
||||
public redditCode: string | null,
|
||||
) {
|
||||
super(message);
|
||||
this.name = 'MatrixRateLimitError';
|
||||
}
|
||||
}
|
||||
|
||||
export interface SyncOptions {
|
||||
/** Pagination token from a previous /sync. Omit for initial sync. */
|
||||
since?: string;
|
||||
/** Long-poll timeout. Default 0 = return immediately. */
|
||||
timeoutMs?: number;
|
||||
/** Filter blob (JSON-encoded) — omitted by default. */
|
||||
filter?: string;
|
||||
}
|
||||
|
||||
export interface RoomSummary {
|
||||
roomId: string;
|
||||
/** Sorted member IDs other than ourselves; useful for DMs (1 entry). */
|
||||
otherMembers: string[];
|
||||
/** From summary.m.heroes or m.room.name event content. */
|
||||
displayName: string | null;
|
||||
unreadCount: number;
|
||||
highlightCount: number;
|
||||
lastEvent: MatrixTimelineEvent | null;
|
||||
/** Raw timeline events from this sync — initial sync gives a small window. */
|
||||
recentEvents: MatrixTimelineEvent[];
|
||||
}
|
||||
|
||||
/**
|
||||
* Cap on how long we'll sit and sleep waiting out a 429. Anything longer
|
||||
* is surfaced as a `MatrixRateLimitError` so the main loop can back off
|
||||
* cleanly instead of silently blocking for minutes.
|
||||
*/
|
||||
const LONG_RETRY_THRESHOLD_MS = 5_000;
|
||||
|
||||
export interface MatrixClientOptions {
|
||||
/**
|
||||
* Minimum interval (ms) between *outbound* HTTP requests, applied across
|
||||
* all callers. Defaults to 30ms (~33 req/s) which keeps us comfortably
|
||||
* under Reddit's homeserver rate limit even with several parallel workers.
|
||||
*/
|
||||
minRequestIntervalMs?: number;
|
||||
/**
|
||||
* Per-request HTTP timeout. Without this a stalled connection to the
|
||||
* homeserver would hang the caller forever (see incident 2026-05-20).
|
||||
* An aborted fetch is treated as a transient error and retried with
|
||||
* the same backoff as a 5xx.
|
||||
*/
|
||||
requestTimeoutMs?: number;
|
||||
/** Test seam: inject a custom fetch (e.g. one that hangs). */
|
||||
fetchImpl?: typeof fetch;
|
||||
}
|
||||
|
||||
/**
|
||||
* Thin Matrix Client-Server v3 wrapper for the Reddit chat homeserver.
|
||||
* No SDK dependency — just `fetch`. Talks the same protocol as
|
||||
* matrix-js-sdk but only the few endpoints we need.
|
||||
*/
|
||||
export class MatrixClient {
|
||||
private nextBatch: string | null = null;
|
||||
private nextRequestSlot = 0;
|
||||
private readonly minRequestIntervalMs: number;
|
||||
private readonly requestTimeoutMs: number;
|
||||
private readonly fetchImpl: typeof fetch;
|
||||
|
||||
constructor(
|
||||
private creds: MatrixCredentials,
|
||||
private logger: Logger,
|
||||
options: MatrixClientOptions = {},
|
||||
) {
|
||||
this.minRequestIntervalMs = options.minRequestIntervalMs ?? 30;
|
||||
this.requestTimeoutMs = options.requestTimeoutMs ?? 30_000;
|
||||
this.fetchImpl = options.fetchImpl ?? fetch;
|
||||
}
|
||||
|
||||
/**
|
||||
* Block until we're allowed to fire the next outbound HTTP request.
|
||||
* Stamps `nextRequestSlot` atomically so concurrent acquirers get
|
||||
* staggered, non-overlapping slots.
|
||||
*/
|
||||
private async acquireRequestSlot(): Promise<void> {
|
||||
const now = Date.now();
|
||||
const slot = Math.max(now, this.nextRequestSlot);
|
||||
this.nextRequestSlot = slot + this.minRequestIntervalMs;
|
||||
const wait = slot - now;
|
||||
if (wait > 0) await new Promise((r) => setTimeout(r, wait));
|
||||
}
|
||||
|
||||
get userId(): string {
|
||||
return this.creds.userId;
|
||||
}
|
||||
|
||||
get homeserver(): string {
|
||||
return this.creds.homeserver;
|
||||
}
|
||||
|
||||
/** Replace the bearer (e.g. after a re-extract on 401). */
|
||||
updateCredentials(creds: MatrixCredentials): void {
|
||||
this.creds = creds;
|
||||
this.nextBatch = null;
|
||||
}
|
||||
|
||||
/**
|
||||
* `fetch` with a hard client-side timeout. Returns the Response on success,
|
||||
* or null if the request aborted — caller treats null as transient and
|
||||
* retries with backoff.
|
||||
*/
|
||||
private async timedFetch(
|
||||
url: string,
|
||||
init: RequestInit,
|
||||
): Promise<Response | null> {
|
||||
try {
|
||||
return await this.fetchImpl(url, {
|
||||
...init,
|
||||
signal: AbortSignal.timeout(this.requestTimeoutMs),
|
||||
});
|
||||
} catch (err) {
|
||||
const name = (err as { name?: string }).name;
|
||||
if (name === 'AbortError' || name === 'TimeoutError') {
|
||||
this.logger.warn(
|
||||
`[matrix] request to ${url} timed out after ${this.requestTimeoutMs}ms`,
|
||||
);
|
||||
return null;
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
private async request<T>(
|
||||
method: 'GET' | 'POST' | 'PUT',
|
||||
path: string,
|
||||
options: { query?: Record<string, string | number | undefined>; body?: unknown } = {},
|
||||
): Promise<T> {
|
||||
const url = new URL(this.creds.homeserver + path);
|
||||
if (options.query) {
|
||||
for (const [k, v] of Object.entries(options.query)) {
|
||||
if (v !== undefined) url.searchParams.set(k, String(v));
|
||||
}
|
||||
}
|
||||
|
||||
const maxAttempts = 10;
|
||||
for (let attempt = 1; attempt <= maxAttempts; attempt++) {
|
||||
await this.acquireRequestSlot();
|
||||
const res = await this.timedFetch(url.toString(), {
|
||||
method,
|
||||
headers: {
|
||||
Authorization: `Bearer ${this.creds.accessToken}`,
|
||||
'Content-Type': 'application/json',
|
||||
},
|
||||
body: options.body ? JSON.stringify(options.body) : undefined,
|
||||
});
|
||||
|
||||
if (res === null) {
|
||||
// Aborted — retry with the same backoff as a 5xx.
|
||||
if (attempt < maxAttempts) {
|
||||
const backoff =
|
||||
Math.min(2000, 200 * 2 ** (attempt - 1)) + Math.floor(Math.random() * 150);
|
||||
await new Promise((r) => setTimeout(r, backoff));
|
||||
continue;
|
||||
}
|
||||
throw new Error(`Matrix ${method} ${path} timed out (exceeded ${maxAttempts} attempts)`);
|
||||
}
|
||||
|
||||
if (res.status === 401 || res.status === 403) {
|
||||
throw new MatrixAuthError(`Matrix ${method} ${path} → ${res.status} (token likely expired)`);
|
||||
}
|
||||
|
||||
// Matrix-standard rate-limit response: { retry_after_ms } per spec.
|
||||
if (res.status === 429) {
|
||||
const body = (await res.json().catch(() => ({}))) as {
|
||||
retry_after_ms?: number;
|
||||
errcode?: string;
|
||||
error?: string;
|
||||
['com.reddit.error.code']?: string;
|
||||
['com.reddit.error.msg']?: string;
|
||||
};
|
||||
const baseMs = body.retry_after_ms ?? 500;
|
||||
// If the homeserver tells us to wait longer than this, don't block —
|
||||
// bubble out so the caller can record a cooldown and skip sends.
|
||||
// 5s covers a normal Matrix flood-control burst; anything bigger
|
||||
// (e.g. the ~10 min invitation_limit_mature_account) is a real
|
||||
// soft throttle we shouldn't sit on.
|
||||
if (baseMs > LONG_RETRY_THRESHOLD_MS) {
|
||||
const msg = `Matrix ${method} ${path} → 429 ${body['com.reddit.error.code'] ?? body.errcode ?? ''} retry_after=${baseMs}ms`;
|
||||
throw new MatrixRateLimitError(
|
||||
msg,
|
||||
baseMs,
|
||||
body.errcode ?? null,
|
||||
body['com.reddit.error.code'] ?? null,
|
||||
);
|
||||
}
|
||||
// Add tiny jitter so parallel workers don't wake in lockstep.
|
||||
const waitMs = baseMs + Math.floor(Math.random() * 100);
|
||||
if (attempt < maxAttempts) {
|
||||
await new Promise((r) => setTimeout(r, waitMs));
|
||||
continue;
|
||||
}
|
||||
throw new Error(
|
||||
`Matrix ${method} ${path} → 429 (exceeded ${maxAttempts} retries, last wait ${waitMs}ms)`,
|
||||
);
|
||||
}
|
||||
|
||||
// Transient upstream errors — Reddit's homeserver occasionally 502/504s
|
||||
// under load. Retry with exponential backoff + jitter.
|
||||
if (res.status >= 500 && res.status <= 599 && attempt < maxAttempts) {
|
||||
const backoff = Math.min(2000, 200 * 2 ** (attempt - 1)) + Math.floor(Math.random() * 150);
|
||||
await new Promise((r) => setTimeout(r, backoff));
|
||||
continue;
|
||||
}
|
||||
|
||||
if (!res.ok) {
|
||||
const text = await res.text().catch(() => '');
|
||||
throw new Error(`Matrix ${method} ${path} → ${res.status}: ${text.slice(0, 300)}`);
|
||||
}
|
||||
return (await res.json()) as T;
|
||||
}
|
||||
// Unreachable, but TS doesn't know.
|
||||
throw new Error(`Matrix ${method} ${path} exhausted retries`);
|
||||
}
|
||||
|
||||
async whoami(): Promise<string> {
|
||||
const res = await this.request<{ user_id: string }>('GET', '/_matrix/client/v3/account/whoami');
|
||||
return res.user_id;
|
||||
}
|
||||
|
||||
/**
|
||||
* All rooms the user is currently joined to. Use this for backfill —
|
||||
* `/sync` only returns rooms with recent activity.
|
||||
*/
|
||||
async getJoinedRooms(): Promise<string[]> {
|
||||
const res = await this.request<{ joined_rooms: string[] }>(
|
||||
'GET',
|
||||
'/_matrix/client/v3/joined_rooms',
|
||||
);
|
||||
return res.joined_rooms;
|
||||
}
|
||||
|
||||
/** Member map for a room (user_id → display name/avatar). */
|
||||
async getJoinedMembers(
|
||||
roomId: string,
|
||||
): Promise<Record<string, { display_name?: string; avatar_url?: string }>> {
|
||||
const res = await this.request<{
|
||||
joined: Record<string, { display_name?: string; avatar_url?: string }>;
|
||||
}>(
|
||||
'GET',
|
||||
`/_matrix/client/v3/rooms/${encodeURIComponent(roomId)}/joined_members`,
|
||||
);
|
||||
return res.joined;
|
||||
}
|
||||
|
||||
/** Returns the room's m.room.name content if set, else null. */
|
||||
async getRoomName(roomId: string): Promise<string | null> {
|
||||
try {
|
||||
const res = await this.request<{ name?: string }>(
|
||||
'GET',
|
||||
`/_matrix/client/v3/rooms/${encodeURIComponent(roomId)}/state/m.room.name/`,
|
||||
);
|
||||
return res.name ?? null;
|
||||
} catch (err) {
|
||||
// 404 = no name event (normal for DMs)
|
||||
if (err instanceof Error && err.message.includes('→ 404')) return null;
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
async sync(options: SyncOptions = {}): Promise<MatrixSyncResponse> {
|
||||
const since = options.since ?? this.nextBatch ?? undefined;
|
||||
const res = await this.request<MatrixSyncResponse>('GET', '/_matrix/client/v3/sync', {
|
||||
query: {
|
||||
since,
|
||||
timeout: options.timeoutMs ?? 0,
|
||||
filter: options.filter,
|
||||
},
|
||||
});
|
||||
this.nextBatch = res.next_batch;
|
||||
return res;
|
||||
}
|
||||
|
||||
/**
|
||||
* Convenience: do a /sync and project the joined rooms into a flat list.
|
||||
* For an initial sync (no `since`) Reddit returns each room with a small
|
||||
* recent-event window — usually enough for "active chats" views.
|
||||
*/
|
||||
async listRooms(options: SyncOptions = {}): Promise<RoomSummary[]> {
|
||||
const sync = await this.sync(options);
|
||||
const joined = sync.rooms?.join ?? {};
|
||||
const summaries: RoomSummary[] = [];
|
||||
|
||||
for (const [roomId, room] of Object.entries(joined)) {
|
||||
summaries.push(this.summarizeRoom(roomId, room));
|
||||
}
|
||||
summaries.sort((a, b) => {
|
||||
const at = a.lastEvent?.origin_server_ts ?? 0;
|
||||
const bt = b.lastEvent?.origin_server_ts ?? 0;
|
||||
return bt - at;
|
||||
});
|
||||
return summaries;
|
||||
}
|
||||
|
||||
private summarizeRoom(roomId: string, room: MatrixJoinedRoom): RoomSummary {
|
||||
const events = room.timeline?.events ?? [];
|
||||
const lastMessage = [...events].reverse().find((e) => e.type === 'm.room.message') ?? null;
|
||||
const lastEvent = events.length > 0 ? events[events.length - 1] : null;
|
||||
|
||||
const heroes = room.summary?.['m.heroes'] ?? [];
|
||||
const otherMembers = heroes.filter((h) => h !== this.creds.userId);
|
||||
|
||||
let displayName: string | null = null;
|
||||
const nameEvent = room.state?.events?.find((e) => e.type === 'm.room.name');
|
||||
if (nameEvent && typeof nameEvent.content.name === 'string') {
|
||||
displayName = nameEvent.content.name as string;
|
||||
} else if (otherMembers.length > 0) {
|
||||
displayName = otherMembers.map(stripMxid).join(', ');
|
||||
}
|
||||
|
||||
return {
|
||||
roomId,
|
||||
otherMembers,
|
||||
displayName,
|
||||
unreadCount: room.unread_notifications?.notification_count ?? 0,
|
||||
highlightCount: room.unread_notifications?.highlight_count ?? 0,
|
||||
lastEvent: lastMessage ?? lastEvent,
|
||||
recentEvents: events,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Paginate room history backwards. Pass the previous response's `end`
|
||||
* token as `from` to keep walking.
|
||||
*/
|
||||
async getRoomMessages(
|
||||
roomId: string,
|
||||
options: { from?: string; limit?: number; dir?: 'b' | 'f' } = {},
|
||||
): Promise<MatrixMessagesResponse> {
|
||||
return this.request<MatrixMessagesResponse>(
|
||||
'GET',
|
||||
`/_matrix/client/v3/rooms/${encodeURIComponent(roomId)}/messages`,
|
||||
{
|
||||
query: {
|
||||
dir: options.dir ?? 'b',
|
||||
limit: options.limit ?? 50,
|
||||
from: options.from,
|
||||
},
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a new Reddit DM room with the given recipient and return its
|
||||
* room_id. Mirrors exactly what chat.reddit.com sends when you click
|
||||
* "Start chat" in the UI:
|
||||
*
|
||||
* POST /_matrix/client/v3/createRoom
|
||||
* { "preset": "reddit_dm", "invite": [mxid] }
|
||||
*
|
||||
* `reddit_dm` is a Reddit-specific preset that the homeserver expands
|
||||
* server-side into the right power levels, history visibility, and
|
||||
* `com.reddit.chat.type` state event.
|
||||
*/
|
||||
async createDirectRoom(recipientMxid: string): Promise<string> {
|
||||
const res = await this.request<{ room_id: string }>('POST', '/_matrix/client/v3/createRoom', {
|
||||
body: { preset: 'reddit_dm', invite: [recipientMxid] },
|
||||
});
|
||||
this.logger.debug(`[matrix] created DM room ${res.room_id} with ${recipientMxid}`);
|
||||
return res.room_id;
|
||||
}
|
||||
|
||||
/**
|
||||
* One-shot: create a DM room with the recipient and send the opening
|
||||
* message. Returns the new room_id and the event_id of the message.
|
||||
*/
|
||||
async sendInitialDirectMessage(
|
||||
recipientMxid: string,
|
||||
body: string,
|
||||
): Promise<{ roomId: string; eventId: string }> {
|
||||
const roomId = await this.createDirectRoom(recipientMxid);
|
||||
const eventId = await this.sendTextMessage(roomId, body);
|
||||
return { roomId, eventId };
|
||||
}
|
||||
|
||||
/**
|
||||
* Convert a Reddit t2 user id (e.g. "t2_abc123") into the Matrix user id
|
||||
* Reddit's homeserver uses: "@t2_abc123:reddit.com".
|
||||
*/
|
||||
static t2ToMxid(t2Id: string): string {
|
||||
const clean = t2Id.startsWith('t2_') ? t2Id : `t2_${t2Id}`;
|
||||
return `@${clean}:reddit.com`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Send a plain-text message. Returns the event_id.
|
||||
* txnId must be unique per message to make sends idempotent on retry.
|
||||
*/
|
||||
async sendTextMessage(roomId: string, body: string, txnId?: string): Promise<string> {
|
||||
const id = txnId ?? `rbot-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;
|
||||
const res = await this.request<{ event_id: string }>(
|
||||
'PUT',
|
||||
`/_matrix/client/v3/rooms/${encodeURIComponent(roomId)}/send/m.room.message/${encodeURIComponent(id)}`,
|
||||
{
|
||||
body: { msgtype: 'm.text', body },
|
||||
},
|
||||
);
|
||||
this.logger.debug(`[matrix] sent message to ${roomId}: event_id=${res.event_id}`);
|
||||
return res.event_id;
|
||||
}
|
||||
|
||||
/**
|
||||
* Convert an mxc:// URI to an HTTPS download URL on this homeserver.
|
||||
* Reddit's media is served from the same Matrix server.
|
||||
*
|
||||
* Returns the *authenticated* v1 endpoint (Matrix v1.11+); use
|
||||
* `downloadMedia()` to actually fetch with the right headers.
|
||||
*/
|
||||
mxcToHttp(mxc: string): string | null {
|
||||
const parsed = parseMxc(mxc);
|
||||
if (!parsed) return null;
|
||||
return `${this.creds.homeserver}/_matrix/client/v1/media/download/${parsed.serverName}/${parsed.mediaId}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Download a media object referenced by an mxc:// URI.
|
||||
*
|
||||
* Tries the authenticated v1 endpoint first (current spec) and falls back
|
||||
* to the unauthenticated v3 endpoint (legacy) on 404 — different Matrix
|
||||
* server versions deploy these at different times.
|
||||
*/
|
||||
async downloadMedia(mxc: string): Promise<{ bytes: Buffer; contentType: string } | null> {
|
||||
const parsed = parseMxc(mxc);
|
||||
if (!parsed) return null;
|
||||
const { serverName, mediaId } = parsed;
|
||||
|
||||
// Reddit's homeserver serves the legacy unauthenticated v3 endpoint;
|
||||
// the modern authed v1 endpoint 404s. Try v3 first to save a round-trip
|
||||
// per image; v1 stays as a fallback for when Reddit upgrades.
|
||||
const candidates = [
|
||||
{
|
||||
url: `${this.creds.homeserver}/_matrix/media/v3/download/${serverName}/${mediaId}`,
|
||||
authed: false,
|
||||
},
|
||||
{
|
||||
url: `${this.creds.homeserver}/_matrix/client/v1/media/download/${serverName}/${mediaId}`,
|
||||
authed: true,
|
||||
},
|
||||
];
|
||||
|
||||
for (const { url, authed } of candidates) {
|
||||
const headers: Record<string, string> = {};
|
||||
if (authed) headers['Authorization'] = `Bearer ${this.creds.accessToken}`;
|
||||
|
||||
const maxAttempts = 10;
|
||||
for (let attempt = 1; attempt <= maxAttempts; attempt++) {
|
||||
await this.acquireRequestSlot();
|
||||
const res = await this.timedFetch(url, { headers });
|
||||
if (res === null) {
|
||||
if (attempt < maxAttempts) {
|
||||
const backoff =
|
||||
Math.min(2000, 200 * 2 ** (attempt - 1)) + Math.floor(Math.random() * 150);
|
||||
await new Promise((r) => setTimeout(r, backoff));
|
||||
continue;
|
||||
}
|
||||
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)`);
|
||||
}
|
||||
if (res.status === 429) {
|
||||
const body = (await res.json().catch(() => ({}))) as { retry_after_ms?: number };
|
||||
const waitMs = (body.retry_after_ms ?? 500) + Math.floor(Math.random() * 100);
|
||||
if (attempt < maxAttempts) {
|
||||
await new Promise((r) => setTimeout(r, waitMs));
|
||||
continue;
|
||||
}
|
||||
throw new Error(`Matrix media download → 429 after ${maxAttempts} retries`);
|
||||
}
|
||||
if (res.status === 404 || res.status === 405) break; // try next endpoint
|
||||
if (!res.ok) {
|
||||
const text = await res.text().catch(() => '');
|
||||
throw new Error(`Matrix media download ${url} → ${res.status}: ${text.slice(0, 200)}`);
|
||||
}
|
||||
const arrayBuffer = await res.arrayBuffer();
|
||||
return {
|
||||
bytes: Buffer.from(arrayBuffer),
|
||||
contentType: res.headers.get('content-type') ?? 'application/octet-stream',
|
||||
};
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
function parseMxc(mxc: string): { serverName: string; mediaId: string } | null {
|
||||
const m = mxc.match(/^mxc:\/\/([^/]+)\/(.+)$/);
|
||||
if (!m) return null;
|
||||
return { serverName: m[1], mediaId: m[2] };
|
||||
}
|
||||
|
||||
/** "@t2_abc123:reddit.com" → "t2_abc123" */
|
||||
function stripMxid(mxid: string): string {
|
||||
const m = mxid.match(/^@([^:]+):/);
|
||||
return m ? m[1] : mxid;
|
||||
}
|
||||
@@ -0,0 +1,44 @@
|
||||
import type { Logger } from '../../../utils/logger.js';
|
||||
import type { MediaDownloader, MediaDownloadResult } from '../../media.js';
|
||||
import type { MatrixClient } from './client.js';
|
||||
import { fetchWithTimeout } from '../../../utils/fetch-with-timeout.js';
|
||||
|
||||
/**
|
||||
* `MediaDownloader` for the existing `MediaService` that knows how to fetch
|
||||
* both raw HTTPS URLs (legacy chat snapshots, external links) and Matrix
|
||||
* `mxc://` URIs. The latter go through `MatrixClient.downloadMedia` so they
|
||||
* honour the same auth + rate-limiter machinery as the rest of the chat
|
||||
* code path; nothing in the chat flow needs Playwright any more.
|
||||
*/
|
||||
export class MatrixMediaDownloader implements MediaDownloader {
|
||||
constructor(
|
||||
private matrix: MatrixClient,
|
||||
private logger: Logger,
|
||||
) {}
|
||||
|
||||
async fetch(url: string): Promise<MediaDownloadResult | null> {
|
||||
try {
|
||||
if (url.startsWith('mxc://')) {
|
||||
const result = await this.matrix.downloadMedia(url);
|
||||
if (!result) return null;
|
||||
return { buffer: result.bytes, contentType: result.contentType };
|
||||
}
|
||||
|
||||
// Reddit chat sometimes embeds direct https://matrix.redditspace.com
|
||||
// URLs (e.g. /thumbnail/.../blur_*) — fall through to plain fetch.
|
||||
const res = await fetchWithTimeout(url, { timeoutMs: 30_000 });
|
||||
if (!res.ok) {
|
||||
this.logger.warn(`Media download failed (${res.status}): ${url}`);
|
||||
return null;
|
||||
}
|
||||
const ab = await res.arrayBuffer();
|
||||
return {
|
||||
buffer: Buffer.from(ab),
|
||||
contentType: res.headers.get('content-type') ?? null,
|
||||
};
|
||||
} catch (error) {
|
||||
this.logger.warn(`Media download error for ${url}:`, error);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,131 @@
|
||||
import fs from 'fs';
|
||||
import path from 'path';
|
||||
import type { BrowserContext } from 'playwright';
|
||||
import type { Logger } from '../../../utils/logger.js';
|
||||
import type { MatrixCredentials } from './types.js';
|
||||
import { fetchWithTimeout } from '../../../utils/fetch-with-timeout.js';
|
||||
|
||||
const DEFAULT_CACHE_PATH = path.resolve(process.cwd(), 'data/matrix-credentials.json');
|
||||
/** Reddit chat homeserver. The bearer token is host-scoped to this origin. */
|
||||
const MATRIX_HOST_RE = /matrix\.(redditspace|reddit)\.com$/i;
|
||||
|
||||
export interface TokenExtractorOptions {
|
||||
/** Where to persist the captured token. */
|
||||
cachePath?: string;
|
||||
/** Max ms to wait for a Matrix-bound request to appear. */
|
||||
timeoutMs?: number;
|
||||
}
|
||||
|
||||
export async function loadCachedCredentials(
|
||||
cachePath = DEFAULT_CACHE_PATH,
|
||||
): Promise<MatrixCredentials | null> {
|
||||
try {
|
||||
if (!fs.existsSync(cachePath)) return null;
|
||||
const raw = fs.readFileSync(cachePath, 'utf-8');
|
||||
const parsed = JSON.parse(raw) as MatrixCredentials;
|
||||
if (!parsed.accessToken || !parsed.homeserver || !parsed.userId) return null;
|
||||
return parsed;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
export function saveCredentials(
|
||||
creds: MatrixCredentials,
|
||||
cachePath = DEFAULT_CACHE_PATH,
|
||||
): void {
|
||||
const dir = path.dirname(cachePath);
|
||||
if (!fs.existsSync(dir)) fs.mkdirSync(dir, { recursive: true });
|
||||
fs.writeFileSync(cachePath, JSON.stringify(creds, null, 2), 'utf-8');
|
||||
}
|
||||
|
||||
/**
|
||||
* Sniffs the Matrix bearer token by opening chat.reddit.com in the supplied
|
||||
* (already-logged-in) browser context and watching outbound requests for any
|
||||
* call to the Matrix homeserver. The first request carrying an
|
||||
* `Authorization: Bearer ...` header is captured.
|
||||
*
|
||||
* Requires the context to already hold a valid Reddit session — call this
|
||||
* after BrowserClient.login() or with a persistent context that has cookies.
|
||||
*/
|
||||
export async function extractMatrixCredentials(
|
||||
context: BrowserContext,
|
||||
logger: Logger,
|
||||
options: TokenExtractorOptions = {},
|
||||
): Promise<MatrixCredentials> {
|
||||
const timeoutMs = options.timeoutMs ?? 30_000;
|
||||
const cachePath = options.cachePath ?? DEFAULT_CACHE_PATH;
|
||||
|
||||
const page = await context.newPage();
|
||||
try {
|
||||
let resolveCreds: (c: MatrixCredentials) => void;
|
||||
let rejectCreds: (err: Error) => void;
|
||||
const credsPromise = new Promise<MatrixCredentials>((resolve, reject) => {
|
||||
resolveCreds = resolve;
|
||||
rejectCreds = reject;
|
||||
});
|
||||
|
||||
let done = false;
|
||||
const onRequest = (request: import('playwright').Request) => {
|
||||
if (done) return;
|
||||
try {
|
||||
const url = new URL(request.url());
|
||||
if (!MATRIX_HOST_RE.test(url.hostname)) return;
|
||||
const headers = request.headers();
|
||||
const auth = headers['authorization'] || headers['Authorization'];
|
||||
if (!auth || !auth.toLowerCase().startsWith('bearer ')) return;
|
||||
const token = auth.slice('bearer '.length).trim();
|
||||
if (!token) return;
|
||||
const homeserver = `${url.protocol}//${url.host}`;
|
||||
done = true;
|
||||
logger.info(`[matrix] captured bearer token for ${homeserver}`);
|
||||
resolveCreds({
|
||||
homeserver,
|
||||
accessToken: token,
|
||||
userId: '',
|
||||
capturedAt: Date.now(),
|
||||
});
|
||||
} catch {
|
||||
// swallow malformed URLs
|
||||
}
|
||||
};
|
||||
|
||||
page.on('request', onRequest);
|
||||
|
||||
logger.info('[matrix] loading chat.reddit.com to capture access token...');
|
||||
await page.goto('https://chat.reddit.com/', {
|
||||
waitUntil: 'domcontentloaded',
|
||||
timeout: 30_000,
|
||||
});
|
||||
|
||||
const timeout = new Promise<MatrixCredentials>((_, reject) =>
|
||||
setTimeout(
|
||||
() => reject(new Error(`Matrix token capture timed out after ${timeoutMs}ms`)),
|
||||
timeoutMs,
|
||||
),
|
||||
);
|
||||
|
||||
const captured = await Promise.race([credsPromise, timeout]);
|
||||
|
||||
// Resolve the user_id by calling /whoami so the caller knows who we are.
|
||||
const whoamiRes = await fetchWithTimeout(
|
||||
`${captured.homeserver}/_matrix/client/v3/account/whoami`,
|
||||
{
|
||||
headers: { Authorization: `Bearer ${captured.accessToken}` },
|
||||
timeoutMs: 15_000,
|
||||
},
|
||||
);
|
||||
if (whoamiRes.ok) {
|
||||
const body = (await whoamiRes.json()) as { user_id?: string };
|
||||
if (body.user_id) captured.userId = body.user_id;
|
||||
} else {
|
||||
logger.warn(`[matrix] whoami returned ${whoamiRes.status}; userId left blank`);
|
||||
}
|
||||
|
||||
saveCredentials(captured, cachePath);
|
||||
logger.info(`[matrix] credentials saved to ${cachePath} (user=${captured.userId || 'unknown'})`);
|
||||
return captured;
|
||||
} finally {
|
||||
await page.close().catch(() => undefined);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,85 @@
|
||||
/**
|
||||
* Minimal subset of Matrix Client-Server API response shapes that the
|
||||
* Reddit chat client actually uses. Reddit runs its chat on a Matrix
|
||||
* homeserver (typically matrix.redditspace.com), so the same API works.
|
||||
*
|
||||
* Spec: https://spec.matrix.org/v1.11/client-server-api/
|
||||
*/
|
||||
|
||||
export interface MatrixCredentials {
|
||||
/** Homeserver base URL, e.g. "https://matrix.redditspace.com" */
|
||||
homeserver: string;
|
||||
/** Bearer access token */
|
||||
accessToken: string;
|
||||
/** Matrix user ID, e.g. "@t2_abc123:reddit.com" */
|
||||
userId: string;
|
||||
/** When the credentials were captured (ms). Used for staleness checks. */
|
||||
capturedAt: number;
|
||||
}
|
||||
|
||||
export interface MatrixEventContent {
|
||||
msgtype?: string;
|
||||
body?: string;
|
||||
/** mxc:// URI for image/file messages */
|
||||
url?: string;
|
||||
info?: {
|
||||
mimetype?: string;
|
||||
w?: number;
|
||||
h?: number;
|
||||
size?: number;
|
||||
};
|
||||
[k: string]: unknown;
|
||||
}
|
||||
|
||||
export interface MatrixTimelineEvent {
|
||||
event_id: string;
|
||||
type: string;
|
||||
sender: string;
|
||||
origin_server_ts: number;
|
||||
content: MatrixEventContent;
|
||||
unsigned?: { age?: number };
|
||||
}
|
||||
|
||||
export interface MatrixRoomTimeline {
|
||||
events: MatrixTimelineEvent[];
|
||||
limited?: boolean;
|
||||
prev_batch?: string;
|
||||
}
|
||||
|
||||
export interface MatrixRoomState {
|
||||
events: MatrixTimelineEvent[];
|
||||
}
|
||||
|
||||
export interface MatrixJoinedRoom {
|
||||
timeline?: MatrixRoomTimeline;
|
||||
state?: MatrixRoomState;
|
||||
unread_notifications?: {
|
||||
notification_count?: number;
|
||||
highlight_count?: number;
|
||||
};
|
||||
summary?: {
|
||||
'm.heroes'?: string[];
|
||||
'm.joined_member_count'?: number;
|
||||
'm.invited_member_count'?: number;
|
||||
};
|
||||
}
|
||||
|
||||
export interface MatrixSyncResponse {
|
||||
next_batch: string;
|
||||
rooms?: {
|
||||
join?: Record<string, MatrixJoinedRoom>;
|
||||
leave?: Record<string, unknown>;
|
||||
invite?: Record<string, unknown>;
|
||||
};
|
||||
}
|
||||
|
||||
export interface MatrixMessagesResponse {
|
||||
chunk: MatrixTimelineEvent[];
|
||||
start: string;
|
||||
end?: string;
|
||||
}
|
||||
|
||||
export interface MatrixWhoamiResponse {
|
||||
user_id: string;
|
||||
device_id?: string;
|
||||
}
|
||||
@@ -0,0 +1,856 @@
|
||||
import type { Logger } from '../../utils/logger.js';
|
||||
import type { SuccessfulMessagesRepository } from '../../database/repositories.js';
|
||||
import type { MediaService } from '../media.js';
|
||||
import { ChatCache } from './cache.js';
|
||||
import { MatrixClient, MatrixAuthError, MatrixRateLimitError } from './matrix/client.js';
|
||||
import type { MatrixJoinedRoom, MatrixTimelineEvent } from './matrix/types.js';
|
||||
import type { ActiveChat, ChatMessage, ScrapedChatMessage } from './types.js';
|
||||
import { fetchWithTimeout } from '../../utils/fetch-with-timeout.js';
|
||||
|
||||
const ACTIVE_WINDOW_MS = 96 * 60 * 60 * 1000;
|
||||
const ROOM_URL_PREFIX = 'https://chat.reddit.com/room/';
|
||||
|
||||
export interface TokenRefresher {
|
||||
/**
|
||||
* Re-extract a fresh Matrix bearer token (e.g. by re-running the browser
|
||||
* sniff against the persistent session) and apply it to the underlying
|
||||
* client. Resolves when the client is ready to be retried.
|
||||
*/
|
||||
refresh(): Promise<void>;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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<Record<string, string>>;
|
||||
}
|
||||
|
||||
interface BuildChatOpts {
|
||||
knownUsernames: Map<string, string>;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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/<txn>` 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;
|
||||
}
|
||||
|
||||
/** `null` signals "sync ended" (clear progress UI). */
|
||||
export type SyncProgressListener = (progress: SyncProgress | null) => void;
|
||||
|
||||
/** Fired when a chat we sent into first receives a reply from the other party. */
|
||||
export type NewReplyListener = (info: {
|
||||
username: string;
|
||||
replySnippet: string;
|
||||
}) => void;
|
||||
|
||||
/**
|
||||
* Set when Reddit's homeserver tells us to stop creating new DMs for a while
|
||||
* (e.g. the "Wow, you've created a lot of chats. Let's take a break." soft
|
||||
* throttle, errcode `rate.invitation_limit_mature_account`).
|
||||
*/
|
||||
export interface InvitationCooldown {
|
||||
/** Wall-clock ms when sends may resume. */
|
||||
until: number;
|
||||
/** Reddit-side error code, e.g. `rate.invitation_limit_mature_account`. */
|
||||
redditCode: string | null;
|
||||
/** Human-readable explanation suitable for the dashboard. */
|
||||
reason: string;
|
||||
/** Wall-clock ms when the cooldown was set. */
|
||||
setAt: number;
|
||||
}
|
||||
|
||||
export class ChatService {
|
||||
private cache: ChatCache;
|
||||
private queue: Promise<unknown> = Promise.resolve();
|
||||
private readonly t2ToUsername = new Map<string, string>();
|
||||
private syncingKind: 'full' | 'delta' | null = null;
|
||||
private progress: SyncProgress | null = null;
|
||||
private progressListener: SyncProgressListener | null = null;
|
||||
private newReplyListener: NewReplyListener | 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,
|
||||
) {
|
||||
this.cache = new ChatCache(logger);
|
||||
}
|
||||
|
||||
// ------------ scheduling + auth helpers ------------
|
||||
|
||||
/** Serialize work so concurrent callers don't clobber cache state. */
|
||||
private enqueue<T>(task: () => Promise<T>): Promise<T> {
|
||||
const result = this.queue.then(() => task());
|
||||
this.queue = result.catch(() => undefined);
|
||||
return result;
|
||||
}
|
||||
|
||||
/**
|
||||
* Run a Matrix call; if it 401/403s, re-extract the bearer via the
|
||||
* configured `TokenRefresher` and retry once. Single retry only — if the
|
||||
* refresh itself fails, we propagate.
|
||||
*/
|
||||
private async withAuth<T>(fn: () => Promise<T>): Promise<T> {
|
||||
try {
|
||||
return await fn();
|
||||
} catch (err) {
|
||||
if (err instanceof MatrixAuthError && this.tokenRefresher) {
|
||||
this.logger.warn('[chat] Matrix token rejected — refreshing via browser session');
|
||||
await this.tokenRefresher.refresh();
|
||||
return await fn();
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
// ------------ public API ------------
|
||||
|
||||
async loadCache(): Promise<void> {
|
||||
await this.cache.load();
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async saveCacheExternal(): Promise<void> {
|
||||
await this.cache.save();
|
||||
}
|
||||
|
||||
async getCachedConversations(): Promise<ActiveChat[]> {
|
||||
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<ActiveChat[]> {
|
||||
return this.enqueue(async () => {
|
||||
if (this.cache.isFresh()) return this.cache.get();
|
||||
this.beginSync('delta');
|
||||
try {
|
||||
this.updateProgress({ phase: 'starting', current: 0, total: 0 });
|
||||
const 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;
|
||||
} finally {
|
||||
this.endSync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 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.
|
||||
*/
|
||||
async refreshTopChats(_limit = 5): Promise<void> {
|
||||
await this.enqueue(async () => {
|
||||
this.beginSync('delta');
|
||||
try {
|
||||
this.updateProgress({ phase: 'starting', current: 0, total: 0 });
|
||||
await this.syncOnce();
|
||||
this.logger.info('[chat] incremental sync complete');
|
||||
this.updateProgress({
|
||||
phase: 'done',
|
||||
current: this.cache.get().length,
|
||||
total: this.cache.get().length,
|
||||
});
|
||||
} catch (error) {
|
||||
this.logger.error('[chat] refreshTopChats failed:', error);
|
||||
} finally {
|
||||
this.endSync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 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(): Promise<ActiveChat[]> {
|
||||
return this.enqueue(async () => {
|
||||
this.beginSync('full');
|
||||
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`);
|
||||
|
||||
this.updateProgress({
|
||||
phase: 'resolving-usernames',
|
||||
current: 0,
|
||||
total: joinedIds.length,
|
||||
});
|
||||
const knownUsernames = await this.resolveUsernamesForRooms(joinedIds);
|
||||
const chats: ActiveChat[] = [];
|
||||
|
||||
for (const roomId of joinedIds) {
|
||||
try {
|
||||
const room = seedJoined[roomId];
|
||||
const chat = this.roomToActiveChat(roomId, room, { knownUsernames });
|
||||
chats.push(chat);
|
||||
} catch (err) {
|
||||
this.logger.warn(`[chat] backfill: failed to seed ${roomId}: ${(err as Error).message}`);
|
||||
}
|
||||
}
|
||||
|
||||
let done = 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) {
|
||||
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);
|
||||
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) {
|
||||
this.logger.error('[chat] fetchAllMessages failed:', error);
|
||||
return this.cache.get();
|
||||
} finally {
|
||||
this.endSync();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 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();
|
||||
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();
|
||||
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;
|
||||
}
|
||||
|
||||
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<Omit<SyncProgress, 'kind' | 'startedAt'>>): void {
|
||||
if (!this.progress) return;
|
||||
this.progress = { ...this.progress, ...patch };
|
||||
this.emitProgress();
|
||||
}
|
||||
|
||||
private endSync(): void {
|
||||
this.syncingKind = null;
|
||||
this.progress = null;
|
||||
if (this.progressListener) {
|
||||
try {
|
||||
this.progressListener(null);
|
||||
} 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<void> {
|
||||
return this.loadDetails(chat);
|
||||
}
|
||||
|
||||
/**
|
||||
* Send a reply into an existing room. Accepts either a full
|
||||
* `https://chat.reddit.com/room/<roomId>` URL (web UI compatibility) or
|
||||
* a raw Matrix room ID like `!abc:reddit.com`.
|
||||
*/
|
||||
async sendChatMessage(chatUrlOrRoomId: string, message: string): Promise<boolean> {
|
||||
const active = this.getInvitationCooldown();
|
||||
if (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 false;
|
||||
}
|
||||
return this.enqueue(async () => {
|
||||
try {
|
||||
const roomId = ChatService.extractRoomId(chatUrlOrRoomId);
|
||||
if (!roomId) {
|
||||
this.logger.error(`[chat] cannot extract room ID from "${chatUrlOrRoomId}"`);
|
||||
return false;
|
||||
}
|
||||
const eventId = await this.withAuth(() => this.matrix.sendTextMessage(roomId, message));
|
||||
this.logger.info(`[chat] sent message to ${roomId} (event_id=${eventId})`);
|
||||
const chat = this.cache.get().find((c) => c.roomId === roomId);
|
||||
if (chat) {
|
||||
chat.lastMessageSnippet = message.slice(0, 200);
|
||||
chat.lastMessageTime = Date.now();
|
||||
chat.isFromUs = true;
|
||||
}
|
||||
return true;
|
||||
} catch (error) {
|
||||
if (error instanceof MatrixRateLimitError) {
|
||||
this.setInvitationCooldown(error);
|
||||
return false;
|
||||
}
|
||||
this.logger.error('[chat] sendChatMessage failed:', error);
|
||||
return false;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 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> {
|
||||
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 null;
|
||||
}
|
||||
return this.enqueue(async () => {
|
||||
try {
|
||||
const t2Id = await this.resolveUsernameToT2(username);
|
||||
if (!t2Id) {
|
||||
this.logger.error(`[chat] cannot start conversation: unknown user "${username}"`);
|
||||
return null;
|
||||
}
|
||||
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);
|
||||
|
||||
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: [
|
||||
{ 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 chat;
|
||||
} catch (error) {
|
||||
if (error instanceof MatrixRateLimitError) {
|
||||
this.setInvitationCooldown(error);
|
||||
return null;
|
||||
}
|
||||
this.logger.error('[chat] startConversation failed:', error);
|
||||
return null;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 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<boolean> {
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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<unknown> {
|
||||
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;
|
||||
return null;
|
||||
}
|
||||
return this.invitationCooldown;
|
||||
}
|
||||
|
||||
/** Record a cooldown after a `MatrixRateLimitError`. Test seam + reuse. */
|
||||
setInvitationCooldown(error: MatrixRateLimitError): 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,
|
||||
redditCode: error.redditCode,
|
||||
reason:
|
||||
error.redditCode === 'rate.invitation_limit_mature_account'
|
||||
? "Reddit is throttling new chat invitations (\"you've created a lot of chats, take a break\")."
|
||||
: `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'})`,
|
||||
);
|
||||
}
|
||||
|
||||
// ------------ internals ------------
|
||||
|
||||
/** Single delta /sync that updates cache + nextBatch and returns the union. */
|
||||
private async syncOnce(): Promise<ActiveChat[]> {
|
||||
const sync = await this.withAuth(() =>
|
||||
this.matrix.sync({ since: this.cache.getMarkers().nextBatch ?? undefined, timeoutMs: 0 }),
|
||||
);
|
||||
this.cache.setNextBatch(sync.next_batch);
|
||||
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));
|
||||
|
||||
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);
|
||||
}
|
||||
|
||||
const merged = Array.from(existing.values()).sort(
|
||||
(a, b) => b.lastMessageTime - a.lastMessageTime,
|
||||
);
|
||||
this.cache.set(merged);
|
||||
this.cache.markDeltaSync();
|
||||
await this.cache.save(merged);
|
||||
return merged;
|
||||
}
|
||||
|
||||
private async loadDetails(chat: ActiveChat): Promise<void> {
|
||||
if (!chat.roomId) {
|
||||
this.logger.warn(`[chat] no roomId for ${chat.roomName}, skipping details`);
|
||||
return;
|
||||
}
|
||||
|
||||
const events: MatrixTimelineEvent[] = [];
|
||||
let from: string | undefined = undefined;
|
||||
for (let i = 0; i < 50; i++) {
|
||||
const res = await this.withAuth(() =>
|
||||
this.matrix.getRoomMessages(chat.roomId!, { from, limit: 100, dir: 'b' }),
|
||||
);
|
||||
events.push(...res.chunk);
|
||||
if (!res.end || res.end === from || res.chunk.length === 0) break;
|
||||
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 mxc = typeof e.content.url === 'string' ? (e.content.url as string) : null;
|
||||
return {
|
||||
sender: e.sender === this.matrix.userId ? 'Me' : this.shortenSender(e.sender),
|
||||
text: body || (mxc ? '[Image]' : ''),
|
||||
timestamp: e.origin_server_ts,
|
||||
isSelf: e.sender === this.matrix.userId,
|
||||
mediaUrls: mxc ? [mxc] : [],
|
||||
};
|
||||
});
|
||||
|
||||
// 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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
chat.messages = this.mediaService
|
||||
? await this.mediaService.attachMedia(chat, scraped)
|
||||
: scraped.map(({ mediaUrls: _ignored, ...m }) => m as ChatMessage);
|
||||
|
||||
if (chat.messages.length === 0) {
|
||||
this.logger.debug(`[chat] no messages in ${chat.roomName}`);
|
||||
return;
|
||||
}
|
||||
|
||||
chat.lastMessageSnippet = chat.messages[chat.messages.length - 1].text;
|
||||
chat.lastMessageTime = chat.messages[chat.messages.length - 1].timestamp;
|
||||
|
||||
const firstSelf = chat.messages.find((m) => m.isSelf);
|
||||
if (firstSelf) chat.originMessage = firstSelf.text;
|
||||
|
||||
const hasReply = chat.messages.some((m) => !m.isSelf);
|
||||
if (firstSelf && hasReply && this.successfulMessagesRepo) {
|
||||
if (!this.successfulMessagesRepo.existsByMessage(firstSelf.text)) {
|
||||
const firstReply = chat.messages.find((m) => !m.isSelf);
|
||||
try {
|
||||
this.successfulMessagesRepo.insert({
|
||||
sentMessageId: null,
|
||||
originalMessage: firstSelf.text,
|
||||
imageAnalysis: null,
|
||||
postTitle: `Chat with ${chat.username}`,
|
||||
responseSnippet: firstReply?.text.slice(0, 200) || null,
|
||||
subreddit: null,
|
||||
});
|
||||
this.logger.info(`[chat] stored successful chat example from ${chat.username}`);
|
||||
try {
|
||||
this.newReplyListener?.({
|
||||
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 store successful chat example:', err);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 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;
|
||||
if (membership === 'join' && stateKey && stateKey !== this.matrix.userId) {
|
||||
memberJoins.push(stateKey);
|
||||
}
|
||||
}
|
||||
const candidateMxids = [...new Set([...heroes, ...memberJoins])];
|
||||
const otherT2 = candidateMxids
|
||||
.map((h) => h.match(/^@(t2_[^:]+):/)?.[1])
|
||||
.filter((s): s is string => !!s);
|
||||
|
||||
let username = prev?.username ?? null;
|
||||
if (!username || 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 unread = (room?.unread_notifications?.notification_count ?? 0) > 0;
|
||||
|
||||
return {
|
||||
...(prev ?? {}),
|
||||
username,
|
||||
roomName,
|
||||
lastMessageTime,
|
||||
lastMessageSnippet,
|
||||
isFromUs,
|
||||
unread,
|
||||
url: ROOM_URL_PREFIX + encodeURIComponent(roomId),
|
||||
roomId,
|
||||
};
|
||||
}
|
||||
|
||||
/** 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 `@t2_abc:reddit.com` → `t2_abc`. */
|
||||
private shortenSender(mxid: string): string {
|
||||
const m = mxid.match(/^@([^:]+):/);
|
||||
return m ? m[1] : mxid;
|
||||
}
|
||||
|
||||
/**
|
||||
* Best-effort batch resolution of t2 IDs to usernames. Uses the injected
|
||||
* `UsernameResolver` (if any); failures are logged and ignored, with the
|
||||
* t2 ID staying as the fallback display.
|
||||
*/
|
||||
private async resolveUsernamesForRooms(_roomIds: string[]): Promise<Map<string, string>> {
|
||||
// We don't have the recipient mxids until rooms are projected, so this
|
||||
// method returns the in-memory cache and lets `roomToActiveChat` fall
|
||||
// back to the t2 ID. Resolution happens lazily via `enrichUsernames()`
|
||||
// which the caller can invoke after building the chat list.
|
||||
return new Map(this.t2ToUsername);
|
||||
}
|
||||
|
||||
/**
|
||||
* 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<void> {
|
||||
if (!this.usernameResolver) return;
|
||||
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) return;
|
||||
try {
|
||||
const got = await this.usernameResolver.resolveT2Ids(needed);
|
||||
let updated = 0;
|
||||
for (const [t2, name] of Object.entries(got)) {
|
||||
const bare = t2.replace(/^t2_/, '');
|
||||
this.t2ToUsername.set(bare, 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`);
|
||||
await this.cache.save();
|
||||
}
|
||||
} catch (err) {
|
||||
this.logger.debug('[chat] enrichUsernames failed:', err);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 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;
|
||||
}
|
||||
}
|
||||
|
||||
/** 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;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,51 @@
|
||||
export interface ChatMessage {
|
||||
sender: string;
|
||||
text: string;
|
||||
timestamp: number;
|
||||
isSelf: boolean;
|
||||
media?: ChatMedia[];
|
||||
}
|
||||
|
||||
export interface ChatMedia {
|
||||
originalUrl: string;
|
||||
localPath: string;
|
||||
fileName: string;
|
||||
}
|
||||
|
||||
export interface ActiveChat {
|
||||
username: string;
|
||||
roomName: string;
|
||||
lastMessageTime: number;
|
||||
lastMessageSnippet: string;
|
||||
isFromUs: boolean;
|
||||
unread: boolean;
|
||||
url?: string;
|
||||
roomId?: string;
|
||||
messages?: ChatMessage[];
|
||||
originMessage?: string;
|
||||
}
|
||||
|
||||
export interface ChatCacheFile {
|
||||
version: number;
|
||||
savedAt: number;
|
||||
chats: ActiveChat[];
|
||||
/** Matrix `next_batch` cursor — present after any successful sync. */
|
||||
nextBatch?: string | null;
|
||||
/** Wall-clock timestamp of the last successful full backfill. Null/missing
|
||||
* → the bot has never done a full sync on this machine, so startup
|
||||
* should run one. */
|
||||
lastFullSyncAt?: number | null;
|
||||
/** Timestamp of the most recent delta sync. */
|
||||
lastDeltaSyncAt?: number | null;
|
||||
}
|
||||
|
||||
export type ScrapedChatMessage = Omit<ChatMessage, 'media'> & { mediaUrls: string[] };
|
||||
|
||||
export interface ScrapedRoom {
|
||||
href: string;
|
||||
ariaLabel: string | null;
|
||||
roomName: string | null;
|
||||
lastMessage: string | null;
|
||||
timeText: string | null;
|
||||
unread: boolean;
|
||||
}
|
||||
@@ -57,9 +57,10 @@ export class MediaService {
|
||||
|
||||
constructor(
|
||||
private downloader: MediaDownloader,
|
||||
private logger: Logger
|
||||
private logger: Logger,
|
||||
mediaRoot?: string,
|
||||
) {
|
||||
this.mediaRoot = path.join(process.cwd(), 'data', 'media');
|
||||
this.mediaRoot = mediaRoot ?? path.join(process.cwd(), 'data', 'media');
|
||||
this.indexPath = path.join(this.mediaRoot, 'index.json');
|
||||
}
|
||||
|
||||
@@ -84,7 +85,7 @@ export class MediaService {
|
||||
|
||||
async getMediaItems(): Promise<MediaRecord[]> {
|
||||
await this.loadIndex();
|
||||
return (this.index?.items || []).slice().sort((a, b) => b.createdAt - a.createdAt);
|
||||
return (this.index?.items || []).slice().sort((a, b) => b.timestamp - a.timestamp);
|
||||
}
|
||||
|
||||
private async downloadMessageMedia(chat: ActiveChat, message: ScrapedChatMessage): Promise<ChatMedia[]> {
|
||||
|
||||
+75
-13
@@ -2,25 +2,32 @@ import type { RedditApiClient, RedditPost } from '../reddit/api.js';
|
||||
import type { PostsRepository, MessagesRepository } from '../database/repositories.js';
|
||||
import type { Config } from '../config/index.js';
|
||||
import type { Logger } from '../utils/logger.js';
|
||||
import { parseGenderTag } from '../utils/gender-tag.js';
|
||||
|
||||
export class MonitorService {
|
||||
// Flips on each call so that when both pools have candidates the bot
|
||||
// alternates between image and chat instead of always favouring whichever
|
||||
// type the random shuffle put first.
|
||||
private chatGoesFirst = false;
|
||||
|
||||
constructor(
|
||||
private redditApi: RedditApiClient,
|
||||
private postsRepo: PostsRepository,
|
||||
private messagesRepo: MessagesRepository,
|
||||
private config: Config,
|
||||
private logger: Logger
|
||||
private logger: Logger,
|
||||
private delayBetweenFetchesMs: number = 1000,
|
||||
) { }
|
||||
|
||||
async getUnprocessedPosts(): Promise<RedditPost[]> {
|
||||
// Create a shuffled copy of subreddits to check
|
||||
const subreddits = [...this.config.reddit.subreddits];
|
||||
this.shuffleArray(subreddits);
|
||||
const subreddits = this.buildCheckOrder();
|
||||
|
||||
const targetGenders = this.config.reddit.targetGenders;
|
||||
|
||||
for (const subreddit of subreddits) {
|
||||
try {
|
||||
this.logger.debug(`Checking r/${subreddit} for new posts...`);
|
||||
const posts = await this.redditApi.getNewPosts(subreddit);
|
||||
this.logger.debug(`Checking r/${subreddit.name} (${subreddit.type}) for new posts...`);
|
||||
const posts = await this.redditApi.getNewPosts(subreddit.name, 25, subreddit.type);
|
||||
|
||||
// Filter for valid, unprocessed posts
|
||||
const validPosts: RedditPost[] = [];
|
||||
@@ -34,11 +41,28 @@ export class MonitorService {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Chat subs almost always have a self-declared gender tag in the
|
||||
// title/body (e.g. [25F], M4F). Drop posts that don't match the
|
||||
// configured targetGenders here so they never enter the processor.
|
||||
if (subreddit.type === 'chat' && targetGenders.length > 0) {
|
||||
const tag = parseGenderTag(`${post.title}\n${post.selftext ?? ''}`);
|
||||
if (!tag) {
|
||||
this.logger.debug(`Skipping chat post ${post.id} - no gender tag in title/body`);
|
||||
continue;
|
||||
}
|
||||
if (!targetGenders.includes(tag)) {
|
||||
this.logger.debug(
|
||||
`Skipping chat post ${post.id} - gender tag "${tag}" not in target list [${targetGenders.join(', ')}]`,
|
||||
);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
validPosts.push(post);
|
||||
}
|
||||
|
||||
if (validPosts.length > 0) {
|
||||
this.logger.info(`Found ${validPosts.length} new posts in r/${subreddit}`);
|
||||
this.logger.info(`Found ${validPosts.length} new posts in r/${subreddit.name}`);
|
||||
|
||||
// Sort by creation time (newest first)
|
||||
validPosts.sort((a, b) => b.created_utc - a.created_utc);
|
||||
@@ -49,21 +73,59 @@ export class MonitorService {
|
||||
return [validPosts[0]];
|
||||
}
|
||||
} catch (error) {
|
||||
this.logger.error(`Failed to fetch posts from r/${subreddit}:`, error);
|
||||
this.logger.error(`Failed to fetch posts from r/${subreddit.name}:`, error);
|
||||
}
|
||||
|
||||
// Small delay between subreddit fetches to be nice to Reddit
|
||||
await this.sleep(1000);
|
||||
if (this.delayBetweenFetchesMs > 0) {
|
||||
await this.sleep(this.delayBetweenFetchesMs);
|
||||
}
|
||||
}
|
||||
|
||||
return [];
|
||||
}
|
||||
|
||||
private shuffleArray<T>(array: T[]): void {
|
||||
for (let i = array.length - 1; i > 0; i--) {
|
||||
const j = Math.floor(Math.random() * (i + 1));
|
||||
[array[i], array[j]] = [array[j], array[i]];
|
||||
/**
|
||||
* Build an interleaved check order that balances image and chat subs.
|
||||
*
|
||||
* Each pool is shuffled independently, then zipped together so a chat sub
|
||||
* is checked roughly every other position even when there are far more
|
||||
* image subs. `chatGoesFirst` flips per call so neither type is
|
||||
* permanently advantaged when both pools have fresh posts.
|
||||
*/
|
||||
private buildCheckOrder(): Config['reddit']['subreddits'] {
|
||||
const images = this.weightedShuffle(
|
||||
this.config.reddit.subreddits.filter((s) => s.type === 'image'),
|
||||
);
|
||||
const chats = this.weightedShuffle(
|
||||
this.config.reddit.subreddits.filter((s) => s.type === 'chat'),
|
||||
);
|
||||
|
||||
const [a, b] = this.chatGoesFirst ? [chats, images] : [images, chats];
|
||||
this.chatGoesFirst = !this.chatGoesFirst;
|
||||
|
||||
const out: Config['reddit']['subreddits'] = [];
|
||||
const max = Math.max(a.length, b.length);
|
||||
for (let i = 0; i < max; i++) {
|
||||
if (i < a.length) out.push(a[i]);
|
||||
if (i < b.length) out.push(b[i]);
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
// Weighted random ordering (Efraimidis–Spirakis): key = random^(1/weight),
|
||||
// sorted descending. Heavier-weighted subs are more likely to land earlier
|
||||
// each cycle, while every sub still has a non-zero chance of any position.
|
||||
private weightedShuffle<T extends { weight?: number }>(items: T[]): T[] {
|
||||
return items
|
||||
.map((item) => {
|
||||
const w = item.weight && item.weight > 0 ? item.weight : 1;
|
||||
const r = Math.random();
|
||||
const key = r === 0 ? -Infinity : Math.log(r) / w;
|
||||
return { item, key };
|
||||
})
|
||||
.sort((a, b) => b.key - a.key)
|
||||
.map(({ item }) => item);
|
||||
}
|
||||
|
||||
private sleep(ms: number): Promise<void> {
|
||||
|
||||
+100
-13
@@ -3,7 +3,8 @@ import type { OpenRouterClient, DetectedGender } from '../ai/openrouter.js';
|
||||
import type { Logger } from '../utils/logger.js';
|
||||
import type { Config } from '../config/index.js';
|
||||
import type { StrategySelector, SelectedStrategy } from './strategy-selector.js';
|
||||
import { downloadImage, bufferToBase64, isValidImageMimeType } from '../utils/image.js';
|
||||
import { downloadImage, bufferToBase64, isValidImageMimeType, type DownloadedImage } from '../utils/image.js';
|
||||
import { parseGenderTag } from '../utils/gender-tag.js';
|
||||
|
||||
export interface ProcessedPost {
|
||||
post: RedditPost;
|
||||
@@ -13,16 +14,33 @@ export interface ProcessedPost {
|
||||
strategyId: string | null;
|
||||
}
|
||||
|
||||
export interface SkipResult {
|
||||
skipReason: string;
|
||||
}
|
||||
|
||||
export type ProcessResult = ProcessedPost | SkipResult;
|
||||
|
||||
export function isSkip(result: ProcessResult): result is SkipResult {
|
||||
return 'skipReason' in result;
|
||||
}
|
||||
|
||||
export type ImageFetcher = (url: string, userAgent: string) => Promise<DownloadedImage>;
|
||||
|
||||
export class ProcessorService {
|
||||
private successfulExamples: string[] = [];
|
||||
private strategySelector: StrategySelector | null = null;
|
||||
private chatStrategySelector: StrategySelector | null = null;
|
||||
private fetchImage: ImageFetcher;
|
||||
|
||||
constructor(
|
||||
private redditApi: RedditApiClient,
|
||||
private aiClient: OpenRouterClient,
|
||||
private config: Config,
|
||||
private logger: Logger
|
||||
) {}
|
||||
private logger: Logger,
|
||||
fetchImage?: ImageFetcher,
|
||||
) {
|
||||
this.fetchImage = fetchImage ?? downloadImage;
|
||||
}
|
||||
|
||||
setStrategySelector(selector: StrategySelector): void {
|
||||
this.strategySelector = selector;
|
||||
@@ -31,6 +49,18 @@ export class ProcessorService {
|
||||
}
|
||||
}
|
||||
|
||||
setChatStrategySelector(selector: StrategySelector): void {
|
||||
this.chatStrategySelector = selector;
|
||||
if (selector.hasStrategies()) {
|
||||
this.logger.info(`Chat strategy A/B testing enabled with ${this.config.chatStrategies.variants.length} variants`);
|
||||
}
|
||||
}
|
||||
|
||||
private getSubredditType(name: string): 'image' | 'chat' {
|
||||
const entry = this.config.reddit.subreddits.find((s) => s.name.toLowerCase() === name.toLowerCase());
|
||||
return entry?.type ?? 'image';
|
||||
}
|
||||
|
||||
setSuccessfulExamples(examples: string[]): void {
|
||||
this.successfulExamples = examples;
|
||||
if (examples.length > 0) {
|
||||
@@ -38,12 +68,16 @@ export class ProcessorService {
|
||||
}
|
||||
}
|
||||
|
||||
async processPost(post: RedditPost): Promise<ProcessedPost | null> {
|
||||
async processPost(post: RedditPost): Promise<ProcessResult> {
|
||||
if (this.getSubredditType(post.subreddit) === 'chat') {
|
||||
return this.processChatPost(post);
|
||||
}
|
||||
|
||||
const imageUrl = this.redditApi.getImageUrl(post);
|
||||
|
||||
if (!imageUrl) {
|
||||
this.logger.warn(`No image URL found for post ${post.id}`);
|
||||
return null;
|
||||
return { skipReason: 'no image URL on post' };
|
||||
}
|
||||
|
||||
this.logger.info(`Processing post: "${post.title.slice(0, 50)}..." by u/${post.author}`);
|
||||
@@ -51,22 +85,23 @@ export class ProcessorService {
|
||||
// Download image
|
||||
let imageData;
|
||||
try {
|
||||
imageData = await downloadImage(imageUrl, this.config.reddit.userAgent);
|
||||
imageData = await this.fetchImage(imageUrl, this.config.reddit.userAgent);
|
||||
} catch (error) {
|
||||
const msg = error instanceof Error ? error.message : String(error);
|
||||
this.logger.error(`Failed to download image from ${imageUrl}:`, error);
|
||||
return null;
|
||||
return { skipReason: `image download failed: ${msg}` };
|
||||
}
|
||||
|
||||
// Validate image type
|
||||
if (!isValidImageMimeType(imageData.mimeType)) {
|
||||
this.logger.warn(`Invalid image type ${imageData.mimeType} for post ${post.id}`);
|
||||
return null;
|
||||
return { skipReason: `unsupported image type (${imageData.mimeType})` };
|
||||
}
|
||||
|
||||
// Check image size (skip very small images like thumbnails)
|
||||
if (imageData.buffer.length < 5000) {
|
||||
this.logger.warn(`Image too small (${imageData.buffer.length} bytes) for post ${post.id}`);
|
||||
return null;
|
||||
return { skipReason: `image too small (${imageData.buffer.length} bytes)` };
|
||||
}
|
||||
|
||||
// Convert to base64
|
||||
@@ -81,7 +116,9 @@ export class ProcessorService {
|
||||
this.logger.info(
|
||||
`Skipping post ${post.id} - detected gender "${detectedGender}" not in target list [${targetGenders.join(', ')}]`
|
||||
);
|
||||
return null;
|
||||
return {
|
||||
skipReason: `detected gender "${detectedGender}" not in target list [${targetGenders.join(', ')}]`,
|
||||
};
|
||||
}
|
||||
|
||||
this.logger.debug(`Gender check passed: ${detectedGender}`);
|
||||
@@ -111,14 +148,16 @@ export class ProcessorService {
|
||||
} catch (error) {
|
||||
if (error instanceof Error && error.message.startsWith('AI_REFUSAL')) {
|
||||
this.logger.warn(`AI refused to analyze post ${post.id} by u/${post.author} - skipping`);
|
||||
return null;
|
||||
return { skipReason: 'AI refused to analyze image' };
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
|
||||
if (!message || message.length < 10) {
|
||||
this.logger.warn(`Generated message too short or empty for post ${post.id}`);
|
||||
return null;
|
||||
this.logger.warn(
|
||||
`Generated message too short or empty for post ${post.id} (${message?.length ?? 0} chars): ${JSON.stringify(message?.slice(0, 200) ?? '')}`,
|
||||
);
|
||||
return { skipReason: `generated message too short (${message?.length ?? 0} chars)` };
|
||||
}
|
||||
|
||||
return {
|
||||
@@ -129,4 +168,52 @@ export class ProcessorService {
|
||||
strategyId: selectedStrategy?.id || null,
|
||||
};
|
||||
}
|
||||
|
||||
private async processChatPost(post: RedditPost): Promise<ProcessResult> {
|
||||
this.logger.info(`Processing chat post: "${post.title.slice(0, 50)}..." by u/${post.author}`);
|
||||
|
||||
// Gender pre-filtering for chat subs happens in MonitorService — by the
|
||||
// time a chat post reaches the processor it's already passed the filter.
|
||||
// We still parse the tag here so it can be threaded into the opener prompt.
|
||||
const haystack = `${post.title}\n${post.selftext ?? ''}`;
|
||||
const detectedGender = parseGenderTag(haystack);
|
||||
|
||||
let selectedStrategy: SelectedStrategy | null = null;
|
||||
if (this.chatStrategySelector) {
|
||||
selectedStrategy = this.chatStrategySelector.selectStrategy();
|
||||
if (selectedStrategy) {
|
||||
this.logger.info(`Using chat strategy "${selectedStrategy.id}" (${selectedStrategy.reason})`);
|
||||
}
|
||||
}
|
||||
|
||||
let message: string;
|
||||
try {
|
||||
message = await this.aiClient.generateChatOpener(
|
||||
post.title,
|
||||
post.selftext ?? '',
|
||||
post.author,
|
||||
detectedGender ?? 'unknown',
|
||||
this.successfulExamples,
|
||||
selectedStrategy?.promptOverride,
|
||||
);
|
||||
} catch (error) {
|
||||
this.logger.error(`Chat opener generation failed for post ${post.id}:`, error);
|
||||
throw error;
|
||||
}
|
||||
|
||||
if (!message || message.length < 10) {
|
||||
this.logger.warn(
|
||||
`Generated chat opener too short or empty for post ${post.id} (${message?.length ?? 0} chars): ${JSON.stringify(message?.slice(0, 200) ?? '')}`,
|
||||
);
|
||||
return { skipReason: `generated chat opener too short (${message?.length ?? 0} chars)` };
|
||||
}
|
||||
|
||||
return {
|
||||
post,
|
||||
imageAnalysis: `[chat] ${post.title}${post.selftext ? ` — ${post.selftext.slice(0, 200)}` : ''}`,
|
||||
detectedGender: detectedGender ?? 'unknown',
|
||||
generatedMessage: message,
|
||||
strategyId: selectedStrategy?.id || null,
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -16,11 +16,13 @@ export class StrategySelector {
|
||||
constructor(
|
||||
private config: Config,
|
||||
private messagesRepo: MessagesRepository,
|
||||
private logger: Logger
|
||||
private logger: Logger,
|
||||
section: 'strategies' | 'chatStrategies' = 'strategies',
|
||||
) {
|
||||
this.variants = config.strategies.variants;
|
||||
this.explorationRate = config.strategies.explorationRate;
|
||||
this.minSamples = config.strategies.minSamples;
|
||||
const cfg = config[section];
|
||||
this.variants = cfg.variants;
|
||||
this.explorationRate = cfg.explorationRate;
|
||||
this.minSamples = cfg.minSamples;
|
||||
}
|
||||
|
||||
selectStrategy(): SelectedStrategy | null {
|
||||
|
||||
Reference in new issue
Block a user