feat: recover missing conversations, fix media attachment, add manual inbox tab
Missing conversations: recipient Reddit accounts that are deleted or suspended left conversations unaccounted for. A new RecipientAccountStatusService polls Reddit for the users behind cached conversations and records their account status durably, with progressive backfill and delta-media fixes in ChatService so history and images are no longer dropped mid-sync. Media: attachMedia now surfaces download failures as MediaAttachmentError so the sync cursor can be retried instead of silently losing an image, recognizes image bytes when Matrix returns a generic content type, and keeps separate gallery occurrences when the same asset appears in more than one event. Manual inbox tab: the conversations list now carries the reply-loop status, so the inbox can separate conversations the operator is driving by hand from the ones automation is still handling. A Manual filter pill and a per-row chip explain why each conversation was handed over. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
1 parent
725b5fc4e6
commit
ac12a3146a
35 files changed
+1164
-129
No files matched your search
@@ -22,11 +22,14 @@ export interface MatrixEventContent {
|
||||
body?: string;
|
||||
/** mxc:// URI for image/file messages */
|
||||
url?: string;
|
||||
/** Unencrypted attachment thumbnail; Matrix clients use both locations. */
|
||||
thumbnail_url?: string;
|
||||
info?: {
|
||||
mimetype?: string;
|
||||
w?: number;
|
||||
h?: number;
|
||||
size?: number;
|
||||
thumbnail_url?: string;
|
||||
};
|
||||
[k: string]: unknown;
|
||||
}
|
||||
|
||||
@@ -138,7 +138,7 @@ export type NewReplyListener = (info: {
|
||||
export class ChatService {
|
||||
private cache: ChatCache;
|
||||
private queue: Promise<unknown> = Promise.resolve();
|
||||
private deltaSyncInFlight: Promise<void> | null = null;
|
||||
private deltaSyncInFlight: Promise<number> | null = null;
|
||||
private readonly t2ToUsername = new Map<string, string>();
|
||||
/** Rooms whose /joined_members lookup has already failed — don't retry. */
|
||||
private readonly unresolvableRooms = new Set<string>();
|
||||
@@ -318,7 +318,7 @@ export class ChatService {
|
||||
let failure: unknown;
|
||||
try {
|
||||
this.updateProgress({ phase: 'starting', current: 0, total: 0 });
|
||||
const merged = await this.syncOnce();
|
||||
const { chats: merged } = await this.syncOnce();
|
||||
const cutoff = Date.now() - ACTIVE_WINDOW_MS;
|
||||
const active = merged.filter((c) => c.lastMessageTime >= cutoff);
|
||||
this.cache.set(active);
|
||||
@@ -340,7 +340,7 @@ export class ChatService {
|
||||
* cache. The `limit` arg is preserved for API compatibility but Matrix
|
||||
* already returns only rooms with new events, so it's a no-op.
|
||||
*/
|
||||
refreshTopChats(_limit = 5): Promise<void> {
|
||||
refreshTopChats(_limit = 5): Promise<number> {
|
||||
if (this.deltaSyncInFlight) return this.deltaSyncInFlight;
|
||||
|
||||
const operation = this.enqueue(async () => {
|
||||
@@ -348,13 +348,14 @@ export class ChatService {
|
||||
let failure: unknown;
|
||||
try {
|
||||
this.updateProgress({ phase: 'starting', current: 0, total: 0 });
|
||||
await this.syncOnce();
|
||||
this.logger.info('[chat] incremental sync complete');
|
||||
const { changedRooms } = await this.syncOnce();
|
||||
this.logger.info(`[chat] incremental sync complete (${changedRooms} changed rooms)`);
|
||||
this.updateProgress({
|
||||
phase: 'done',
|
||||
current: this.cache.get().length,
|
||||
total: this.cache.get().length,
|
||||
});
|
||||
return changedRooms;
|
||||
} catch (error) {
|
||||
failure = error;
|
||||
this.logger.error('[chat] refreshTopChats failed:', error);
|
||||
@@ -376,7 +377,7 @@ export class ChatService {
|
||||
* one's full history. Slow but rare; intended for "rebuild the cache from
|
||||
* scratch" or initial provisioning.
|
||||
*/
|
||||
async fetchAllMessages(): Promise<ActiveChat[]> {
|
||||
async fetchAllMessages(options: { requireComplete?: boolean } = {}): Promise<ActiveChat[]> {
|
||||
return this.enqueue(async () => {
|
||||
this.beginSync('full');
|
||||
let failure: unknown;
|
||||
@@ -436,6 +437,7 @@ export class ChatService {
|
||||
await this.repairDirectRoomMappings(chats);
|
||||
|
||||
let done = 0;
|
||||
let failedRooms = 0;
|
||||
for (const chat of chats) {
|
||||
this.updateProgress({
|
||||
phase: 'loading-details',
|
||||
@@ -447,6 +449,7 @@ export class ChatService {
|
||||
await this.loadDetails(chat);
|
||||
} catch (err) {
|
||||
if (err instanceof MatrixAuthenticationUnavailableError) throw err;
|
||||
failedRooms++;
|
||||
this.logger.error(`[chat] backfill: failed details for ${chat.roomName}`, err);
|
||||
}
|
||||
done++;
|
||||
@@ -459,6 +462,15 @@ export class ChatService {
|
||||
});
|
||||
chats.sort((a, b) => b.lastMessageTime - a.lastMessageTime);
|
||||
this.cache.set(chats);
|
||||
if (failedRooms > 0) {
|
||||
await this.cache.save(chats);
|
||||
const error = new Error(
|
||||
`Full chat backfill incomplete: ${failedRooms} of ${chats.length} conversations failed`,
|
||||
);
|
||||
this.logger.error(`[chat] ${error.message}`);
|
||||
if (options.requireComplete) throw error;
|
||||
return chats;
|
||||
}
|
||||
this.cache.markFullSync();
|
||||
await this.cache.save(chats);
|
||||
|
||||
@@ -470,7 +482,7 @@ export class ChatService {
|
||||
} catch (error) {
|
||||
failure = error;
|
||||
this.logger.error('[chat] fetchAllMessages failed:', error);
|
||||
if (error instanceof MatrixAuthenticationUnavailableError) throw error;
|
||||
if (error instanceof MatrixAuthenticationUnavailableError || options.requireComplete) throw error;
|
||||
return this.cache.get();
|
||||
} finally {
|
||||
this.endSync(failure);
|
||||
@@ -842,11 +854,10 @@ export class ChatService {
|
||||
// ------------ internals ------------
|
||||
|
||||
/** Single delta /sync that updates cache + nextBatch and returns the union. */
|
||||
private async syncOnce(): Promise<ActiveChat[]> {
|
||||
private async syncOnce(): Promise<{ chats: ActiveChat[]; changedRooms: number }> {
|
||||
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));
|
||||
@@ -859,22 +870,26 @@ export class ChatService {
|
||||
}
|
||||
|
||||
for (const [roomId, room] of Object.entries(joined)) {
|
||||
const scraped = this.scrapeMessageEvents(room?.timeline?.events ?? []);
|
||||
if (scraped.length === 0) continue;
|
||||
const chat = existing.get(roomId);
|
||||
if (!chat) continue;
|
||||
|
||||
// A limited timeline means Matrix omitted events between the saved
|
||||
// cursor and this response. Walk the room history before advancing the
|
||||
// cursor so media in that gap cannot become permanently unreachable.
|
||||
if (room?.timeline?.limited) {
|
||||
this.logger.info(`[chat] delta timeline limited for ${roomId}; backfilling room history`);
|
||||
await this.loadDetails(chat);
|
||||
continue;
|
||||
}
|
||||
|
||||
const scraped = this.scrapeMessageEvents(room?.timeline?.events ?? []);
|
||||
if (scraped.length === 0) continue;
|
||||
|
||||
let newMessages: ChatMessage[] = scraped.map(
|
||||
({ mediaUrls: _ignored, ...message }) => message,
|
||||
);
|
||||
if (this.mediaService && scraped.some((message) => message.mediaUrls.length > 0)) {
|
||||
try {
|
||||
newMessages = await this.mediaService.attachMedia(chat, scraped);
|
||||
} catch (err) {
|
||||
this.logger.warn(
|
||||
`[chat] delta media attach failed for ${roomId}: ${(err as Error).message}`,
|
||||
);
|
||||
}
|
||||
newMessages = await this.mediaService.attachMedia(chat, scraped);
|
||||
}
|
||||
this.mergeNewMessages(chat, newMessages);
|
||||
this.recordConversationPerformance(chat);
|
||||
@@ -884,13 +899,14 @@ export class ChatService {
|
||||
(a, b) => b.lastMessageTime - a.lastMessageTime,
|
||||
);
|
||||
this.cache.set(merged);
|
||||
this.cache.setNextBatch(sync.next_batch);
|
||||
this.cache.markDeltaSync();
|
||||
await this.cache.save(merged);
|
||||
// Resolve any t2_-prefixed usernames the delta just introduced, so the
|
||||
// dashboard doesn't show raw Matrix-style IDs for new rooms.
|
||||
await this.enrichUsernames();
|
||||
await this.repairDirectRoomMappings(merged);
|
||||
return merged;
|
||||
return { chats: merged, changedRooms: Object.keys(joined).length };
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -937,18 +953,26 @@ export class ChatService {
|
||||
.filter((e) => e.type === 'm.room.message')
|
||||
.map((e) => {
|
||||
const body = (e.content.body as string | undefined) ?? '';
|
||||
const mxc = typeof e.content.url === 'string' ? (e.content.url as string) : null;
|
||||
const mediaUrls = this.extractMediaUrls(e);
|
||||
return {
|
||||
eventId: e.event_id,
|
||||
sender: e.sender === this.matrix.userId ? 'Me' : this.shortenSender(e.sender),
|
||||
text: body || (mxc ? '[Image]' : ''),
|
||||
text: body || (mediaUrls.length > 0 ? '[Image]' : ''),
|
||||
timestamp: e.origin_server_ts,
|
||||
isSelf: e.sender === this.matrix.userId,
|
||||
mediaUrls: mxc ? [mxc] : [],
|
||||
mediaUrls,
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
private extractMediaUrls(event: MatrixTimelineEvent): string[] {
|
||||
if (typeof event.content.url === 'string' && event.content.url.length > 0) {
|
||||
return [event.content.url];
|
||||
}
|
||||
const thumbnail = event.content.thumbnail_url ?? event.content.info?.thumbnail_url;
|
||||
return typeof thumbnail === 'string' && thumbnail.length > 0 ? [thumbnail] : [];
|
||||
}
|
||||
|
||||
private mergeNewMessages(chat: ActiveChat, newMessages: ChatMessage[]): void {
|
||||
if (newMessages.length === 0) return;
|
||||
chat.messages = this.normalizeMessages([...(chat.messages ?? []), ...newMessages]);
|
||||
@@ -1042,12 +1066,17 @@ export class ChatService {
|
||||
|
||||
const events: MatrixTimelineEvent[] = [];
|
||||
let from: string | undefined = undefined;
|
||||
for (let i = 0; i < 50; i++) {
|
||||
const seenPaginationTokens = new Set<string>();
|
||||
while (true) {
|
||||
const res = await this.withAuth(() =>
|
||||
this.matrix.getRoomMessages(chat.roomId!, { from, limit: 100, dir: 'b' }),
|
||||
);
|
||||
events.push(...res.chunk);
|
||||
if (!res.end || res.end === from || res.chunk.length === 0) break;
|
||||
if (!res.end || res.chunk.length === 0) break;
|
||||
if (res.end === from || seenPaginationTokens.has(res.end)) {
|
||||
throw new Error(`Matrix history pagination stalled for ${chat.roomId}`);
|
||||
}
|
||||
seenPaginationTokens.add(res.end);
|
||||
from = res.end;
|
||||
}
|
||||
|
||||
@@ -1056,14 +1085,14 @@ export class ChatService {
|
||||
|
||||
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;
|
||||
const mediaUrls = this.extractMediaUrls(e);
|
||||
return {
|
||||
eventId: e.event_id,
|
||||
sender: e.sender === this.matrix.userId ? 'Me' : this.shortenSender(e.sender),
|
||||
text: body || (mxc ? '[Image]' : ''),
|
||||
text: body || (mediaUrls.length > 0 ? '[Image]' : ''),
|
||||
timestamp: e.origin_server_ts,
|
||||
isSelf: e.sender === this.matrix.userId,
|
||||
mediaUrls: mxc ? [mxc] : [],
|
||||
mediaUrls,
|
||||
};
|
||||
});
|
||||
|
||||
|
||||
@@ -74,8 +74,7 @@ export class MediaSyncService {
|
||||
this.logger.info('Media sync: sync already in progress, skipping trigger');
|
||||
return false;
|
||||
}
|
||||
await this.runSync(full);
|
||||
return true;
|
||||
return this.runSync(full);
|
||||
}
|
||||
|
||||
getStatus(): MediaSyncStatus {
|
||||
@@ -91,12 +90,12 @@ export class MediaSyncService {
|
||||
};
|
||||
}
|
||||
|
||||
private async runSync(full = false): Promise<void> {
|
||||
private async runSync(full = false): Promise<boolean> {
|
||||
if (!this.accountAuth.isFullyAuthenticated()) {
|
||||
this.logger.debug('Media sync: skipped because the account is not fully authenticated');
|
||||
return;
|
||||
return false;
|
||||
}
|
||||
if (this.syncing) return;
|
||||
if (this.syncing) return false;
|
||||
this.syncing = true;
|
||||
this.lastError = null;
|
||||
const start = Date.now();
|
||||
@@ -104,12 +103,11 @@ export class MediaSyncService {
|
||||
try {
|
||||
if (full) {
|
||||
this.logger.info('Media sync: running full sync (all conversations)...');
|
||||
const chats = await this.chatService.fetchAllMessages();
|
||||
const chats = await this.chatService.fetchAllMessages({ requireComplete: true });
|
||||
this.lastSyncChats = chats.length;
|
||||
} else {
|
||||
this.logger.info('Media sync: running incremental sync (top 10 chats)...');
|
||||
await this.chatService.refreshTopChats(10);
|
||||
this.lastSyncChats = 10;
|
||||
this.logger.info('Media sync: running incremental Matrix sync...');
|
||||
this.lastSyncChats = await this.chatService.refreshTopChats();
|
||||
}
|
||||
|
||||
this.lastSyncAt = Date.now();
|
||||
@@ -118,10 +116,12 @@ export class MediaSyncService {
|
||||
`Media sync: completed in ${Math.round(this.lastSyncDurationMs / 1000)}s` +
|
||||
` (${this.lastSyncChats} chats)`
|
||||
);
|
||||
return true;
|
||||
} catch (error) {
|
||||
this.lastError = error instanceof Error ? error.message : String(error);
|
||||
this.lastSyncDurationMs = Date.now() - start;
|
||||
this.logger.error('Media sync: failed:', error);
|
||||
return false;
|
||||
} finally {
|
||||
this.syncing = false;
|
||||
}
|
||||
|
||||
+128
-41
@@ -2,6 +2,7 @@ import fs from 'fs/promises';
|
||||
import path from 'path';
|
||||
import crypto from 'crypto';
|
||||
import { z } from 'zod';
|
||||
import sharp from 'sharp';
|
||||
import type { Logger } from '../utils/logger.js';
|
||||
import type { ActiveChat, ChatMedia, ChatMessage } from './chat.js';
|
||||
import type { AccountContentRepository } from '../database/account-repositories.js';
|
||||
@@ -28,6 +29,15 @@ export interface MediaRecord {
|
||||
createdAt: number;
|
||||
}
|
||||
|
||||
export class MediaAttachmentError extends Error {
|
||||
constructor(public readonly failedDownloads: number) {
|
||||
super(`Failed to retrieve ${failedDownloads} conversation media item(s)`);
|
||||
this.name = 'MediaAttachmentError';
|
||||
}
|
||||
}
|
||||
|
||||
class MediaDownloadError extends Error {}
|
||||
|
||||
const MediaIndexSchema = z.object({
|
||||
version: z.number(),
|
||||
savedAt: z.number(),
|
||||
@@ -72,9 +82,16 @@ export class MediaService {
|
||||
const databaseBacked = Boolean(this.databaseStore && this.getAccountId());
|
||||
if (!databaseBacked) await this.loadIndex();
|
||||
const updated: ChatMessage[] = [];
|
||||
let failedDownloads = 0;
|
||||
|
||||
for (const message of messages) {
|
||||
const media = await this.downloadMessageMedia(chat, message);
|
||||
let media: ChatMedia[] = [];
|
||||
try {
|
||||
media = await this.downloadMessageMedia(chat, message);
|
||||
} catch (error) {
|
||||
if (!(error instanceof MediaDownloadError)) throw error;
|
||||
failedDownloads++;
|
||||
}
|
||||
updated.push({
|
||||
eventId: message.eventId,
|
||||
sender: message.sender,
|
||||
@@ -86,6 +103,7 @@ export class MediaService {
|
||||
}
|
||||
|
||||
if (!databaseBacked) await this.saveIndex();
|
||||
if (failedDownloads > 0) throw new MediaAttachmentError(failedDownloads);
|
||||
return updated;
|
||||
}
|
||||
|
||||
@@ -155,7 +173,13 @@ export class MediaService {
|
||||
// Try the full-resolution URL first, fall back to original
|
||||
const fullUrl = this.resolveFullMediaUrl(url);
|
||||
const accountId = this.getAccountId();
|
||||
const id = crypto.createHash('sha1').update(`${accountId ?? 'legacy'}\0${fullUrl}`).digest('hex');
|
||||
const ownerKey = accountId ?? 'legacy';
|
||||
const assetId = crypto.createHash('sha1').update(`${ownerKey}\0${fullUrl}`).digest('hex');
|
||||
const occurrenceKey = message.eventId
|
||||
?? `${chat.roomId ?? ''}\0${message.timestamp}\0${message.sender}`;
|
||||
const id = crypto.createHash('sha1')
|
||||
.update(`${ownerKey}\0${occurrenceKey}\0${fullUrl}`)
|
||||
.digest('hex');
|
||||
if (this.databaseStore && accountId) {
|
||||
const existingRecord = this.databaseStore.getMedia(accountId, id);
|
||||
if (existingRecord) {
|
||||
@@ -166,6 +190,23 @@ export class MediaService {
|
||||
this.databaseStore.deleteMedia(accountId, id);
|
||||
}
|
||||
}
|
||||
|
||||
const reusable = this.databaseStore.getMediaByOriginalUrl(accountId, fullUrl);
|
||||
if (reusable) {
|
||||
try {
|
||||
await fs.access(path.join(this.mediaRoot, reusable.relativePath));
|
||||
const record = this.buildRecord(
|
||||
id, chat, message, fullUrl, reusable.fileName, reusable.relativePath, accountId,
|
||||
);
|
||||
this.databaseStore.upsertMedia(this.toStoredRecord(record, accountId));
|
||||
if (reusable.mediaId === assetId && assetId !== id) {
|
||||
this.databaseStore.deleteMedia(accountId, assetId);
|
||||
}
|
||||
return record;
|
||||
} catch {
|
||||
this.databaseStore.deleteMedia(accountId, reusable.mediaId);
|
||||
}
|
||||
}
|
||||
}
|
||||
const existing = this.index?.items.find((item) => item.id === id);
|
||||
if (existing) {
|
||||
@@ -181,23 +222,45 @@ export class MediaService {
|
||||
}
|
||||
}
|
||||
|
||||
const reusable = this.index?.items.find((item) => item.originalUrl === fullUrl);
|
||||
if (reusable) {
|
||||
try {
|
||||
await fs.access(path.join(this.mediaRoot, reusable.relativePath));
|
||||
const record = this.buildRecord(
|
||||
id, chat, message, fullUrl, reusable.fileName, reusable.relativePath, accountId,
|
||||
);
|
||||
this.index!.items.push(record);
|
||||
if (reusable.id === assetId && assetId !== id) {
|
||||
this.index!.items = this.index!.items.filter((item) => item.id !== assetId);
|
||||
}
|
||||
return record;
|
||||
} catch {
|
||||
this.index!.items = this.index!.items.filter((item) => item.id !== reusable.id);
|
||||
}
|
||||
}
|
||||
|
||||
let download = await this.downloader.fetch(fullUrl);
|
||||
if (!download && fullUrl !== url) {
|
||||
this.logger.debug(`Full-res download failed, trying original URL: ${url}`);
|
||||
let extension = download
|
||||
? await this.pickExtension(url, download.contentType, download.buffer)
|
||||
: null;
|
||||
if ((!download || !extension) && fullUrl !== url) {
|
||||
this.logger.debug(`Full-res media was unavailable or invalid, trying original URL: ${url}`);
|
||||
download = await this.downloader.fetch(url);
|
||||
extension = download
|
||||
? await this.pickExtension(url, download.contentType, download.buffer)
|
||||
: null;
|
||||
}
|
||||
if (!download) {
|
||||
this.logger.warn(`Failed to download chat media from ${fullUrl}`);
|
||||
return null;
|
||||
throw new MediaDownloadError('Conversation media download failed');
|
||||
}
|
||||
|
||||
const extension = this.pickExtension(url, download.contentType);
|
||||
if (!extension) {
|
||||
this.logger.debug(`Skipping non-image media from ${url}`);
|
||||
return null;
|
||||
}
|
||||
|
||||
const fileName = `${id}${extension}`;
|
||||
const fileName = `${assetId}${extension}`;
|
||||
const accountDirectory = accountId
|
||||
? crypto.createHash('sha256').update(accountId).digest('hex').slice(0, 16)
|
||||
: '';
|
||||
@@ -207,32 +270,10 @@ export class MediaService {
|
||||
await fs.mkdir(path.dirname(fullPath), { recursive: true });
|
||||
await fs.writeFile(fullPath, download.buffer);
|
||||
|
||||
const record: MediaRecord = {
|
||||
accountId: accountId ?? undefined,
|
||||
id,
|
||||
chatUsername: chat.username || null,
|
||||
chatRoomId: chat.roomId || null,
|
||||
sender: message.sender,
|
||||
timestamp: message.timestamp,
|
||||
originalUrl: fullUrl,
|
||||
fileName,
|
||||
relativePath,
|
||||
createdAt: Date.now(),
|
||||
};
|
||||
const record = this.buildRecord(id, chat, message, fullUrl, fileName, relativePath, accountId);
|
||||
|
||||
if (this.databaseStore && accountId) {
|
||||
this.databaseStore.upsertMedia({
|
||||
accountId,
|
||||
mediaId: id,
|
||||
chatUsername: record.chatUsername,
|
||||
chatRoomId: record.chatRoomId,
|
||||
sender: record.sender,
|
||||
timestamp: record.timestamp,
|
||||
originalUrl: record.originalUrl,
|
||||
fileName: record.fileName,
|
||||
relativePath: record.relativePath,
|
||||
createdAt: record.createdAt,
|
||||
});
|
||||
this.databaseStore.upsertMedia(this.toStoredRecord(record, accountId));
|
||||
return record;
|
||||
}
|
||||
|
||||
@@ -246,26 +287,72 @@ export class MediaService {
|
||||
return record;
|
||||
}
|
||||
|
||||
private pickExtension(url: string, contentType: string | null): string | null {
|
||||
private buildRecord(
|
||||
id: string,
|
||||
chat: ActiveChat,
|
||||
message: ScrapedChatMessage,
|
||||
originalUrl: string,
|
||||
fileName: string,
|
||||
relativePath: string,
|
||||
accountId: string | null,
|
||||
): MediaRecord {
|
||||
return {
|
||||
accountId: accountId ?? undefined,
|
||||
id,
|
||||
chatUsername: chat.username || null,
|
||||
chatRoomId: chat.roomId || null,
|
||||
sender: message.sender,
|
||||
timestamp: message.timestamp,
|
||||
originalUrl,
|
||||
fileName,
|
||||
relativePath,
|
||||
createdAt: Date.now(),
|
||||
};
|
||||
}
|
||||
|
||||
private toStoredRecord(record: MediaRecord, accountId: string) {
|
||||
return {
|
||||
accountId,
|
||||
mediaId: record.id,
|
||||
chatUsername: record.chatUsername,
|
||||
chatRoomId: record.chatRoomId,
|
||||
sender: record.sender,
|
||||
timestamp: record.timestamp,
|
||||
originalUrl: record.originalUrl,
|
||||
fileName: record.fileName,
|
||||
relativePath: record.relativePath,
|
||||
createdAt: record.createdAt,
|
||||
};
|
||||
}
|
||||
|
||||
private async pickExtension(url: string, contentType: string | null, buffer: Buffer): Promise<string | null> {
|
||||
const supportedExtensions = new Set(['.jpg', '.jpeg', '.png', '.gif', '.webp', '.avif']);
|
||||
let urlExt = '';
|
||||
try {
|
||||
urlExt = path.extname(new URL(url).pathname || '').toLowerCase();
|
||||
} catch {
|
||||
urlExt = path.extname(url.split('?')[0] || '').toLowerCase();
|
||||
}
|
||||
if (urlExt && urlExt.length <= 5) {
|
||||
if (supportedExtensions.has(urlExt)) {
|
||||
return urlExt;
|
||||
}
|
||||
|
||||
if (!contentType) {
|
||||
return null;
|
||||
}
|
||||
if (contentType?.includes('image/jpeg')) return '.jpg';
|
||||
if (contentType?.includes('image/png')) return '.png';
|
||||
if (contentType?.includes('image/gif')) return '.gif';
|
||||
if (contentType?.includes('image/webp')) return '.webp';
|
||||
if (contentType?.includes('image/avif')) return '.avif';
|
||||
|
||||
if (contentType.includes('image/jpeg')) return '.jpg';
|
||||
if (contentType.includes('image/png')) return '.png';
|
||||
if (contentType.includes('image/gif')) return '.gif';
|
||||
if (contentType.includes('image/webp')) return '.webp';
|
||||
if (contentType.includes('image/avif')) return '.avif';
|
||||
try {
|
||||
const format = (await sharp(buffer).metadata()).format;
|
||||
if (format === 'jpeg') return '.jpg';
|
||||
if (format === 'png') return '.png';
|
||||
if (format === 'gif') return '.gif';
|
||||
if (format === 'webp') return '.webp';
|
||||
if (format === 'heif') return '.avif';
|
||||
} catch {
|
||||
// Unsupported/corrupt payloads are not gallery images.
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,97 @@
|
||||
import type { Logger } from '../utils/logger.js';
|
||||
import type { RedditUserAccountStatus } from '../reddit/api.js';
|
||||
import { RedditRateLimitError } from '../reddit/api.js';
|
||||
import type { RedditRecipientAccountStatusRepository } from '../database/account-repositories.js';
|
||||
import type { ChatService } from './chat.js';
|
||||
import { denyAccountAuth, type AccountAuthGate } from './account-auth-context.js';
|
||||
|
||||
export const RECIPIENT_ACCOUNT_CHECK_INTERVAL_MS = 4 * 60 * 60_000;
|
||||
const INITIAL_CHECK_DELAY_MS = 60_000;
|
||||
|
||||
export interface RedditAccountStatusLookup {
|
||||
getUserAccountStatus(username: string): Promise<RedditUserAccountStatus>;
|
||||
}
|
||||
|
||||
/** Periodically records whether the Reddit users behind cached conversations still exist. */
|
||||
export class RecipientAccountStatusService {
|
||||
private intervalHandle: NodeJS.Timeout | null = null;
|
||||
private initialHandle: NodeJS.Timeout | null = null;
|
||||
private checking = false;
|
||||
|
||||
constructor(
|
||||
private readonly chatService: Pick<ChatService, 'getCachedConversations'>,
|
||||
private readonly reddit: RedditAccountStatusLookup,
|
||||
private readonly repository: RedditRecipientAccountStatusRepository,
|
||||
private readonly logger: Logger,
|
||||
private readonly accountAuth: AccountAuthGate = denyAccountAuth,
|
||||
private readonly intervalMs = RECIPIENT_ACCOUNT_CHECK_INTERVAL_MS,
|
||||
private readonly now: () => number = Date.now,
|
||||
) {}
|
||||
|
||||
start(): void {
|
||||
if (this.intervalHandle) return;
|
||||
this.initialHandle = setTimeout(() => {
|
||||
this.initialHandle = null;
|
||||
void this.checkNow();
|
||||
}, Math.min(INITIAL_CHECK_DELAY_MS, this.intervalMs));
|
||||
this.intervalHandle = setInterval(() => void this.checkNow(), this.intervalMs);
|
||||
this.logger.info(`Recipient account check: scheduled every ${Math.round(this.intervalMs / 3_600_000)} hours`);
|
||||
}
|
||||
|
||||
stop(): void {
|
||||
if (this.initialHandle) clearTimeout(this.initialHandle);
|
||||
if (this.intervalHandle) clearInterval(this.intervalHandle);
|
||||
this.initialHandle = null;
|
||||
this.intervalHandle = null;
|
||||
}
|
||||
|
||||
async checkNow(): Promise<number> {
|
||||
if (this.checking || !this.accountAuth.isFullyAuthenticated()) return 0;
|
||||
this.checking = true;
|
||||
let checked = 0;
|
||||
const startedAt = this.now();
|
||||
try {
|
||||
const chats = await this.chatService.getCachedConversations();
|
||||
const uniqueUsernames = new Map<string, string>();
|
||||
for (const chat of chats) {
|
||||
const username = chat.username.trim();
|
||||
if (isCheckableUsername(username) && !uniqueUsernames.has(username.toLowerCase())) {
|
||||
uniqueUsernames.set(username.toLowerCase(), username);
|
||||
}
|
||||
}
|
||||
const usernames = [...uniqueUsernames.values()];
|
||||
|
||||
for (const username of usernames) {
|
||||
const previous = this.repository.get(username);
|
||||
if (previous && previous.checkedAt > startedAt - this.intervalMs) continue;
|
||||
try {
|
||||
const status = await this.reddit.getUserAccountStatus(username);
|
||||
this.repository.upsert(username, status, this.now());
|
||||
checked += 1;
|
||||
if (status === 'deleted' && previous?.status !== 'deleted') {
|
||||
this.logger.info(`Recipient account check: u/${username} is deleted`);
|
||||
}
|
||||
} catch (error) {
|
||||
if (error instanceof RedditRateLimitError) {
|
||||
this.logger.warn('Recipient account check paused because Reddit rate-limited profile lookups');
|
||||
break;
|
||||
}
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
this.logger.warn(`Recipient account check failed for u/${username}: ${message}`);
|
||||
}
|
||||
}
|
||||
return checked;
|
||||
} finally {
|
||||
this.checking = false;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function isCheckableUsername(username: string): boolean {
|
||||
const normalized = username.toLowerCase();
|
||||
return username.length > 0
|
||||
&& normalized !== 'unknown'
|
||||
&& normalized !== '[deleted]'
|
||||
&& !normalized.startsWith('t2_')
|
||||
&& !username.startsWith('@');
|
||||
}
|
||||
@@ -13,7 +13,11 @@ import {
|
||||
ActivityLogRepository,
|
||||
RecipientHistoryRepository,
|
||||
} from '../database/repositories.js';
|
||||
import { AccountContentRepository, RedditAccountsRepository } from '../database/account-repositories.js';
|
||||
import {
|
||||
AccountContentRepository,
|
||||
RedditAccountsRepository,
|
||||
RedditRecipientAccountStatusRepository,
|
||||
} from '../database/account-repositories.js';
|
||||
import { RedditApiClient, RedditRateLimitError, type RedditPost } from '../reddit/api.js';
|
||||
import { RedditRequestCoordinator } from '../reddit/request-coordinator.js';
|
||||
import { OpenRouterClient } from '../ai/openrouter.js';
|
||||
@@ -48,6 +52,7 @@ import { ConversationAutomationService } from './conversation-automation.js';
|
||||
import { AiUsageRepository } from '../database/ai-usage-repository.js';
|
||||
import { MediaLibraryRepository } from '../database/media-library-repository.js';
|
||||
import { MediaLibraryExchangeService } from './media-library-exchange.js';
|
||||
import { RecipientAccountStatusService } from './recipient-account-status.js';
|
||||
|
||||
export interface ConcreteRuntimeFactoryOptions {
|
||||
config: Config;
|
||||
@@ -212,6 +217,14 @@ async function buildRuntime(
|
||||
);
|
||||
await chatService.loadCache();
|
||||
const mediaSyncService = new MediaSyncService(chatService, logger, accountConfig.media.syncIntervalMs, accountAuth);
|
||||
const recipientAccountStatusRepo = new RedditRecipientAccountStatusRepository(db);
|
||||
const recipientAccountStatusService = new RecipientAccountStatusService(
|
||||
chatService,
|
||||
redditApi,
|
||||
recipientAccountStatusRepo,
|
||||
logger,
|
||||
accountAuth,
|
||||
);
|
||||
const responseChecker = new ResponseCheckerService(
|
||||
chatPage, messagesRepo, successfulMessagesRepo, postsRepo, logger,
|
||||
accountAuth, conversationOutcomesRepo,
|
||||
@@ -250,10 +263,11 @@ async function buildRuntime(
|
||||
browserProfileDir, browserClient, accountAuth, matrixClient, chatCache, redditApi,
|
||||
postsRepo, messagesRepo, successfulMessagesRepo, deliveryOutboxRepo,
|
||||
conversationStateRepo, conversationReadStateRepo, conversationOutcomesRepo,
|
||||
redditUsersRepo, accountContentRepo,
|
||||
redditUsersRepo, accountContentRepo, recipientAccountStatusRepo,
|
||||
activityLogRepo,
|
||||
monitorService, processorService, responseChecker, conversationAutomation, chatService, mediaService,
|
||||
mediaSyncService, mediaExchange, rateLimiter,
|
||||
recipientAccountStatusService,
|
||||
};
|
||||
const lifecycle = new AccountRuntimeLoop(
|
||||
resources, accountConfig, logger, isBotRunning, recipientDeduplicator,
|
||||
@@ -293,6 +307,7 @@ class AccountRuntimeLoop {
|
||||
if (this.loop) return;
|
||||
this.controller = new AbortController();
|
||||
this.resources.mediaSyncService.start();
|
||||
this.resources.recipientAccountStatusService.start();
|
||||
void this.resources.chatService.runStartupSync()
|
||||
.then(() => this.resources.chatService.enrichUsernames())
|
||||
.catch((error) => this.logger.error(`[account ${this.resources.accountAuth.getSnapshot().redditUsername}] startup sync failed:`, error));
|
||||
@@ -302,6 +317,7 @@ class AccountRuntimeLoop {
|
||||
async stop(): Promise<void> {
|
||||
this.controller?.abort();
|
||||
this.resources.mediaSyncService.stop();
|
||||
this.resources.recipientAccountStatusService.stop();
|
||||
await this.loop?.catch((error) => {
|
||||
if (!(error instanceof DOMException && error.name === 'AbortError')) throw error;
|
||||
});
|
||||
|
||||
@@ -6,6 +6,7 @@ import type {
|
||||
RedditAccountPrivateRecord,
|
||||
RedditAccountsRepository,
|
||||
AccountContentRepository,
|
||||
RedditRecipientAccountStatusRepository,
|
||||
} from '../database/account-repositories.js';
|
||||
import type {
|
||||
ConversationOutcomesRepository,
|
||||
@@ -33,6 +34,7 @@ import type { ProcessorService } from './processor.js';
|
||||
import type { ResponseCheckerService } from './response-checker.js';
|
||||
import type { RedditPost } from '../reddit/api.js';
|
||||
import type { ConversationAutomationService } from './conversation-automation.js';
|
||||
import type { RecipientAccountStatusService } from './recipient-account-status.js';
|
||||
import type { ProcessedPost, ProcessResult } from './processor.js';
|
||||
|
||||
/** Credentials passed to a factory. They must never be exposed by list/snapshot APIs. */
|
||||
@@ -70,6 +72,7 @@ export interface RedditAccountRuntimeResources {
|
||||
conversationOutcomesRepo: ConversationOutcomesRepository;
|
||||
redditUsersRepo: RedditUsersRepository;
|
||||
accountContentRepo: AccountContentRepository;
|
||||
recipientAccountStatusRepo: RedditRecipientAccountStatusRepository;
|
||||
activityLogRepo: ActivityLogRepository;
|
||||
|
||||
monitorService: MonitorService;
|
||||
@@ -79,6 +82,7 @@ export interface RedditAccountRuntimeResources {
|
||||
chatService: ChatService;
|
||||
mediaService: MediaService;
|
||||
mediaSyncService: MediaSyncService;
|
||||
recipientAccountStatusService: RecipientAccountStatusService;
|
||||
mediaExchange?: MediaLibraryExchangeService;
|
||||
rateLimiter: RateLimiter;
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user