diff --git a/apps/dashboard/src/features/conversations/conversations-page.tsx b/apps/dashboard/src/features/conversations/conversations-page.tsx index 94c9d01..8d67998 100644 --- a/apps/dashboard/src/features/conversations/conversations-page.tsx +++ b/apps/dashboard/src/features/conversations/conversations-page.tsx @@ -1,4 +1,4 @@ -import { useMemo, useState } from 'react'; +import { useEffect, useMemo, useState } from 'react'; import type { ActiveChat } from '../../../../../packages/api-contracts/src/index.js'; import { RedditUsernameLink, redditUsernameLabel } from '../../components/reddit-username-link.js'; import { ConversationDetail } from './conversation-detail.js'; @@ -8,6 +8,7 @@ import { ProcessedUserLookup } from './processed-user-lookup.js'; import { AccountOwnerBadge } from '../../components/account-owner-badge.js'; import './conversations.css'; import { ownedConversationId } from './conversation-identity.js'; +import { formatExactTime, formatShortRelativeTime } from './relative-time.js'; export type ConversationFilter = 'all' | 'unread' | 'awaiting-reply' | 'received-images'; @@ -30,9 +31,10 @@ function matchesFilter(conversation: ActiveChat, filter: ConversationFilter): bo return true; } -function ConversationListItem({ conversation, isSelected, onSelect }: { +function ConversationListItem({ conversation, isSelected, now, onSelect }: { conversation: ActiveChat; isSelected: boolean; + now: number; onSelect(): void; }) { const state = conversation.unread ? 'Unread' : conversation.isFromUs ? 'Awaiting reply' : 'Up to date'; @@ -45,7 +47,15 @@ function ConversationListItem({ conversation, isSelected, onSelect }: { onClick={onSelect} type="button" /> - {state} + + + + {state} + + + {conversation.hasReceivedImages ? Image received : null} {conversation.lastMessageSnippet || 'No message preview'} @@ -58,9 +68,15 @@ export function ConversationsPage({ selectedConversationId, onSelectedConversati const [internalSelectedId, setInternalSelectedId] = useState(null); const [localFilters, setLocalFilters] = useState({ search: '', filter: 'all' }); const [isStartingConversation, setIsStartingConversation] = useState(false); + const [now, setNow] = useState(() => Date.now()); const activeConversationId = selectedConversationId !== undefined ? selectedConversationId : internalSelectedId; const activeFilters = filters ?? localFilters; + useEffect(() => { + const interval = window.setInterval(() => setNow(Date.now()), 60_000); + return () => window.clearInterval(interval); + }, []); + function updateFilters(next: ConversationFilters) { if (filters === undefined) setLocalFilters(next); onFiltersChange?.(next); @@ -68,12 +84,14 @@ export function ConversationsPage({ selectedConversationId, onSelectedConversati const visibleConversations = useMemo(() => { const normalizedSearch = activeFilters.search.trim().toLowerCase(); - return (conversationsQuery.data ?? []).filter((conversation) => { - const matchesSearch = normalizedSearch.length === 0 - || conversation.username.toLowerCase().includes(normalizedSearch) - || conversation.lastMessageSnippet.toLowerCase().includes(normalizedSearch); - return matchesSearch && matchesFilter(conversation, activeFilters.filter); - }); + return (conversationsQuery.data ?? []) + .filter((conversation) => { + const matchesSearch = normalizedSearch.length === 0 + || conversation.username.toLowerCase().includes(normalizedSearch) + || conversation.lastMessageSnippet.toLowerCase().includes(normalizedSearch); + return matchesSearch && matchesFilter(conversation, activeFilters.filter); + }) + .sort((left, right) => right.lastMessageTime - left.lastMessageTime); }, [activeFilters, conversationsQuery.data]); function selectConversation(id: string | null) { @@ -144,6 +162,7 @@ export function ConversationsPage({ selectedConversationId, onSelectedConversati conversation={conversation} isSelected={ownedConversationId(conversation) === activeConversationId} key={ownedConversationId(conversation)} + now={now} onSelect={() => selectConversation(ownedConversationId(conversation))} /> ))} diff --git a/apps/dashboard/src/features/conversations/conversations.css b/apps/dashboard/src/features/conversations/conversations.css index 493323b..5b66e49 100644 --- a/apps/dashboard/src/features/conversations/conversations.css +++ b/apps/dashboard/src/features/conversations/conversations.css @@ -43,6 +43,8 @@ .conversation-list-item > span:last-child { overflow: hidden; color: var(--muted); font-size: .82rem; text-overflow: ellipsis; white-space: nowrap; } .conversation-list-heading { display: flex; align-items: center; justify-content: space-between; gap: .5rem; } .conversation-list-heading small { color: var(--accent); } +.conversation-list-meta { display: flex; align-items: center; gap: .45rem; } +.conversation-list-meta time { color: var(--muted); font-size: .7rem; white-space: nowrap; } .conversation-image-marker { justify-self: start; padding: .18rem .45rem; color: #d8e8ff !important; background: #18324d; border: 1px solid #315d85; border-radius: 999px; font-size: .68rem !important; font-weight: 750; } .conversation-list-state, .conversation-detail-state { display: grid; place-items: center; align-content: center; gap: .5rem; min-height: 16rem; padding: 1.5rem; color: var(--muted); text-align: center; } .conversation-list-state strong, .conversation-detail-state strong { color: #e9f1ec; } diff --git a/apps/dashboard/src/features/conversations/relative-time.ts b/apps/dashboard/src/features/conversations/relative-time.ts new file mode 100644 index 0000000..a315431 --- /dev/null +++ b/apps/dashboard/src/features/conversations/relative-time.ts @@ -0,0 +1,20 @@ +const MINUTE_MS = 60_000; +const HOUR_MS = 60 * MINUTE_MS; +const DAY_MS = 24 * HOUR_MS; +const MONTH_MS = 30 * DAY_MS; +const YEAR_MS = 365 * DAY_MS; + +export function formatShortRelativeTime(timestamp: number, now: number): string { + const elapsed = Math.max(0, now - timestamp); + + if (elapsed < MINUTE_MS) return 'just now'; + if (elapsed < HOUR_MS) return `${Math.floor(elapsed / MINUTE_MS)}m ago`; + if (elapsed < DAY_MS) return `${Math.floor(elapsed / HOUR_MS)}h ago`; + if (elapsed < MONTH_MS) return `${Math.floor(elapsed / DAY_MS)}d ago`; + if (elapsed < YEAR_MS) return `${Math.floor(elapsed / MONTH_MS)}mo ago`; + return `${Math.floor(elapsed / YEAR_MS)}y ago`; +} + +export function formatExactTime(timestamp: number): string { + return new Intl.DateTimeFormat(undefined, { dateStyle: 'medium', timeStyle: 'short' }).format(timestamp); +} diff --git a/apps/dashboard/src/features/media/filters.ts b/apps/dashboard/src/features/media/filters.ts index d2d7630..cc18003 100644 --- a/apps/dashboard/src/features/media/filters.ts +++ b/apps/dashboard/src/features/media/filters.ts @@ -23,11 +23,13 @@ export function filterMedia(items: MediaItem[], filters: MediaFilters, now: numb '30d': 30 * 86_400_000, }[filters.dateRange]; - return items.filter((item) => { - if (filters.conversation !== 'all' && mediaConversation(item) !== filters.conversation) return false; - if (filters.sender !== 'all' && item.sender !== filters.sender) return false; - if (filters.type !== 'all' && mediaType(item) !== filters.type) return false; - if (rangeMs !== null && item.timestamp < now - rangeMs) return false; - return true; - }); + return items + .filter((item) => { + if (filters.conversation !== 'all' && mediaConversation(item) !== filters.conversation) return false; + if (filters.sender !== 'all' && item.sender !== filters.sender) return false; + if (filters.type !== 'all' && mediaType(item) !== filters.type) return false; + if (rangeMs !== null && item.timestamp < now - rangeMs) return false; + return true; + }) + .sort((left, right) => right.timestamp - left.timestamp); } diff --git a/src/bootstrap.ts b/src/bootstrap.ts index 229e320..8d63023 100644 --- a/src/bootstrap.ts +++ b/src/bootstrap.ts @@ -12,6 +12,7 @@ import { ConversationStateRepository, ConversationOutcomesRepository, DeliveryOutboxRepository, + RecipientHistoryRepository, } from './database/repositories.js'; import { BrowserClient } from './browser/client.js'; import { RedditApiClient } from './reddit/api.js'; @@ -46,6 +47,7 @@ import { ChatCache } from './services/chat/cache.js'; import { RedditAccountRuntimeManager } from './services/reddit-account-runtime.js'; import { createConcreteRedditAccountRuntimeFactory } from './services/reddit-account-runtime-builder.js'; import { importLegacyAccountContent } from './database/legacy-content-import.js'; +import { RecipientDeduplicator } from './services/recipient-deduplicator.js'; export interface AppContainer { config: Config; @@ -63,6 +65,7 @@ export interface AppContainer { deliveryOutboxRepo: DeliveryOutboxRepository; conversationStateRepo: ConversationStateRepository; conversationOutcomesRepo: ConversationOutcomesRepository; + recipientDeduplicator: RecipientDeduplicator; redditApi: RedditApiClient; aiClient: OpenRouterClient; @@ -119,6 +122,7 @@ export async function bootstrap( const deliveryOutboxRepo = new DeliveryOutboxRepository(db, bootstrapAccountId); const conversationStateRepo = new ConversationStateRepository(db, bootstrapAccountId); const conversationOutcomesRepo = new ConversationOutcomesRepository(db, bootstrapAccountId); + const recipientDeduplicator = new RecipientDeduplicator(new RecipientHistoryRepository(db)); const redditAccountsRepo = bootstrapAccountsRepo; const accountContentRepo = new AccountContentRepository(db); const legacyImport = importLegacyAccountContent(db); @@ -163,7 +167,7 @@ export async function bootstrap( // AI + services const aiClient = new OpenRouterClient(config, logger); - const monitorService = new MonitorService(redditApi, postsRepo, messagesRepo, config, logger); + const monitorService = new MonitorService(redditApi, postsRepo, recipientDeduplicator, config, logger); const processorService = new ProcessorService(redditApi, aiClient, config, logger); const strategySelector = new StrategySelector(config, messagesRepo, logger); processorService.setStrategySelector(strategySelector); @@ -436,6 +440,7 @@ export async function bootstrap( deliveryOutboxRepo, conversationStateRepo, conversationOutcomesRepo, + recipientDeduplicator, redditApi, aiClient, monitorService, diff --git a/src/database/repositories.ts b/src/database/repositories.ts index 8d571f6..6f9706e 100644 --- a/src/database/repositories.ts +++ b/src/database/repositories.ts @@ -109,6 +109,24 @@ export interface DeliveryOutboxRecord { sentMessageId: number | null; } +/** Read-only recipient history shared by every Reddit account runtime. */ +export class RecipientHistoryRepository { + private readonly hasSentToStmt: Database.Statement; + + constructor(db: Database.Database) { + this.hasSentToStmt = db.prepare(` + SELECT 1 + FROM sent_messages + WHERE recipient = ? COLLATE NOCASE + LIMIT 1 + `); + } + + hasSentTo(recipient: string): boolean { + return this.hasSentToStmt.get(recipient) !== undefined; + } +} + /** * Durable state for an outbound delivery. The caller must only perform the * remote send after claiming an item, then acknowledge it using the same diff --git a/src/main-loop.ts b/src/main-loop.ts index 4101244..73d9514 100644 --- a/src/main-loop.ts +++ b/src/main-loop.ts @@ -85,6 +85,7 @@ export async function runMainLoop(app: AppContainer): Promise { accountAuth, postsRepo, messagesRepo, + recipientDeduplicator, redditApi, monitorService, processorService, @@ -290,17 +291,35 @@ export async function runMainLoop(app: AppContainer): Promise { await rateLimiter.throttle( async () => { + if (!recipientDeduplicator.tryReserve(post.author)) { + const reason = 'User already contacted across accounts'; + logger.info(`Skipping u/${post.author} - ${reason.toLowerCase()}`); + postsRepo.setError(post.id, reason); + webServer?.logActivity( + 'skipped', + `Skipped u/${post.author} (already contacted across accounts)`, + post.author, + ); + return false; + } + logger.info(`Sending message to u/${post.author}...`); webServer?.pushLog('info', `Sending message to u/${post.author}...`); - const chat = await chatService.startConversation( - post.author, - processed.generatedMessage, - post.author_fullname, - ); + let chat; + try { + chat = await chatService.startConversation( + post.author, + processed.generatedMessage, + post.author_fullname, + ); + } catch (error) { + recipientDeduplicator.release(post.author); + throw error; + } const success = chat !== null; - if (success) { + if (chat) { postsRepo.markMessageSent(post.id); // Refresh top chats in background so dashboard is up-to-date. @@ -340,6 +359,7 @@ export async function runMainLoop(app: AppContainer): Promise { }); } } else { + recipientDeduplicator.release(post.author); postsRepo.setError(post.id, 'Failed to send message'); webServer?.logActivity( 'error', diff --git a/src/services/chat/matrix/client.ts b/src/services/chat/matrix/client.ts index 131bf3e..01545f5 100644 --- a/src/services/chat/matrix/client.ts +++ b/src/services/chat/matrix/client.ts @@ -14,6 +14,18 @@ export class MatrixAuthError extends Error { } } +/** + * Thrown when Reddit's homeserver explicitly disables a Matrix endpoint. + * Callers can remember the missing capability instead of retrying it for + * every room while still using the remaining Matrix API normally. + */ +export class MatrixFeatureDisabledError extends Error { + constructor(message: string) { + super(message); + this.name = 'MatrixFeatureDisabledError'; + } +} + /** * Thrown when Reddit's homeserver returns a 429 with a `retry_after_ms` * longer than we're willing to block on. Caller (typically ChatService) @@ -201,6 +213,11 @@ export class MatrixClient { `Matrix ${method} ${path} → ${res.status} ${failure.errcode ?? ''} (token rejected)`, ); } + if (res.status === 403 && isFeatureDisabledFailure(failure.text)) { + throw new MatrixFeatureDisabledError( + `Matrix ${method} ${path} → 403 (disabled by configuration)`, + ); + } throw new Error(`Matrix ${method} ${path} → ${res.status}: ${failure.text.slice(0, 300)}`); } @@ -626,6 +643,16 @@ function isMatrixAuthenticationFailure(status: number, errcode?: string): boolea return status === 401 || errcode === 'M_UNKNOWN_TOKEN' || errcode === 'M_MISSING_TOKEN'; } +function isFeatureDisabledFailure(responseText: string): boolean { + try { + const parsed = JSON.parse(responseText) as { message?: unknown }; + return typeof parsed.message === 'string' + && parsed.message.toLowerCase() === 'disabled by configuration'; + } catch { + return false; + } +} + function parseMxc(mxc: string): { serverName: string; mediaId: string } | null { const m = mxc.match(/^mxc:\/\/([^/]+)\/(.+)$/); if (!m) return null; diff --git a/src/services/chat/service.ts b/src/services/chat/service.ts index b1b1d31..ea52e27 100644 --- a/src/services/chat/service.ts +++ b/src/services/chat/service.ts @@ -7,7 +7,12 @@ import type { } from '../../database/repositories.js'; import type { MediaService } from '../media.js'; import { ChatCache } from './cache.js'; -import { MatrixClient, MatrixAuthError, MatrixRateLimitError } from './matrix/client.js'; +import { + MatrixClient, + MatrixAuthError, + MatrixFeatureDisabledError, + MatrixRateLimitError, +} from './matrix/client.js'; import type { MatrixJoinedRoom, MatrixTimelineEvent } from './matrix/types.js'; import type { ActiveChat, ChatMessage, ScrapedChatMessage, InvitationCooldown } from './types.js'; @@ -128,6 +133,7 @@ export class ChatService { private readonly t2ToUsername = new Map(); /** Rooms whose /joined_members lookup has already failed — don't retry. */ private readonly unresolvableRooms = new Set(); + private roomStateLookupAvailable = true; private syncingKind: 'full' | 'delta' | null = null; private progress: SyncProgress | null = null; private progressListener: SyncProgressListener | null = null; @@ -1341,25 +1347,32 @@ export class ChatService { chat.username.startsWith('@'); if (!needsLookup) continue; let t2: string | undefined; - try { - const state = await this.withAuth(() => this.matrix.getRoomFullState(chat.roomId!)); - const otherMember = state.find((e) => { - if (e.type !== 'm.room.member') return false; - const sk = (e as { state_key?: string }).state_key; - const membership = (e.content as { membership?: string }).membership; - return ( - sk && - sk !== this.matrix.userId && - (membership === 'join' || membership === 'invite') - ); - }) as { state_key?: string } | undefined; - t2 = otherMember?.state_key?.match(/^@(t2_[^:]+):/)?.[1]; - } catch (err) { - if (err instanceof MatrixAuthenticationUnavailableError) { - this.logger.warn('[chat] username enrichment paused until Reddit login is restored'); - return; + if (this.roomStateLookupAvailable) { + try { + const state = await this.withAuth(() => this.matrix.getRoomFullState(chat.roomId!)); + const otherMember = state.find((e) => { + if (e.type !== 'm.room.member') return false; + const sk = (e as { state_key?: string }).state_key; + const membership = (e.content as { membership?: string }).membership; + return ( + sk && + sk !== this.matrix.userId && + (membership === 'join' || membership === 'invite') + ); + }) as { state_key?: string } | undefined; + t2 = otherMember?.state_key?.match(/^@(t2_[^:]+):/)?.[1]; + } catch (err) { + if (err instanceof MatrixAuthenticationUnavailableError) { + this.logger.warn('[chat] username enrichment paused until Reddit login is restored'); + return; + } + if (err instanceof MatrixFeatureDisabledError) { + this.roomStateLookupAvailable = false; + this.logger.debug('[chat] room state lookup is disabled upstream; using message history instead'); + } else { + this.logger.debug(`[chat] /state lookup failed for ${chat.roomId}:`, err); + } } - this.logger.debug(`[chat] /state lookup failed for ${chat.roomId}:`, err); } if (!t2) { try { diff --git a/src/services/monitor.ts b/src/services/monitor.ts index 9218688..3c790f6 100644 --- a/src/services/monitor.ts +++ b/src/services/monitor.ts @@ -1,5 +1,5 @@ import { RedditRateLimitError, type RedditApiClient, type RedditPost } from '../reddit/api.js'; -import type { PostsRepository, MessagesRepository } from '../database/repositories.js'; +import type { PostsRepository } 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'; @@ -13,7 +13,7 @@ export class MonitorService { constructor( private redditApi: RedditApiClient, private postsRepo: PostsRepository, - private messagesRepo: MessagesRepository, + private recipientHistory: { hasSentTo(recipient: string): boolean }, private config: Config, private logger: Logger, private delayBetweenFetchesMs: number = 1000, @@ -46,7 +46,7 @@ export class MonitorService { continue; } - if (this.messagesRepo.hasSentTo(post.author)) { + if (this.recipientHistory.hasSentTo(post.author)) { this.logger.debug(`Skipping post ${post.id} from u/${post.author} - already messaged this user`); continue; } diff --git a/src/services/recipient-deduplicator.ts b/src/services/recipient-deduplicator.ts new file mode 100644 index 0000000..db52a05 --- /dev/null +++ b/src/services/recipient-deduplicator.ts @@ -0,0 +1,39 @@ +export interface RecipientHistory { + hasSentTo(recipient: string): boolean; +} + +/** + * Coordinates recipient selection across account runtimes. + * + * SQLite history prevents duplicates across restarts. The in-memory + * reservation closes the smaller race where two live runtimes both pass the + * history check before either has persisted its successful send. + */ +export class RecipientDeduplicator { + private readonly reservations = new Set(); + + constructor(private readonly history: RecipientHistory) {} + + hasSentTo(recipient: string): boolean { + const key = normalizeRecipient(recipient); + return !key || this.reservations.has(key) || this.history.hasSentTo(recipient); + } + + tryReserve(recipient: string): boolean { + const key = normalizeRecipient(recipient); + if (!key || this.reservations.has(key) || this.history.hasSentTo(recipient)) { + return false; + } + this.reservations.add(key); + return true; + } + + release(recipient: string): void { + const key = normalizeRecipient(recipient); + if (key) this.reservations.delete(key); + } +} + +function normalizeRecipient(recipient: string): string { + return recipient.trim().toLocaleLowerCase('en-US'); +} diff --git a/src/services/reddit-account-runtime-builder.ts b/src/services/reddit-account-runtime-builder.ts index b91da1e..99359b2 100644 --- a/src/services/reddit-account-runtime-builder.ts +++ b/src/services/reddit-account-runtime-builder.ts @@ -9,6 +9,7 @@ import { DeliveryOutboxRepository, ConversationStateRepository, ConversationOutcomesRepository, RedditUsersRepository, ActivityLogRepository, + RecipientHistoryRepository, } from '../database/repositories.js'; import { AccountContentRepository } from '../database/account-repositories.js'; import { RedditApiClient } from '../reddit/api.js'; @@ -36,6 +37,7 @@ import { accountBrowserProfileDir, type RedditAccountRuntimeResources, type VerifiedAccountSession, } from './reddit-account-runtime.js'; +import { RecipientDeduplicator } from './recipient-deduplicator.js'; export interface ConcreteRuntimeFactoryOptions { config: Config; @@ -47,12 +49,18 @@ export interface ConcreteRuntimeFactoryOptions { export function createConcreteRedditAccountRuntimeFactory( options: ConcreteRuntimeFactoryOptions, ): RedditAccountRuntimeFactory { - return new RedditAccountRuntimeFactory((session) => buildRuntime(options, session)); + const recipientDeduplicator = new RecipientDeduplicator( + new RecipientHistoryRepository(options.db), + ); + return new RedditAccountRuntimeFactory( + (session) => buildRuntime(options, session, recipientDeduplicator), + ); } async function buildRuntime( { config, logger, db, isBotRunning = () => true }: ConcreteRuntimeFactoryOptions, session: VerifiedAccountSession, + recipientDeduplicator: RecipientDeduplicator, ): Promise { const browserProfileDir = accountBrowserProfileDir( path.resolve(process.cwd(), config.browser.sessionDir), @@ -103,7 +111,9 @@ async function buildRuntime( redditApi.setBrowserContext(browserClient.getContext()!); const chatPage = await browserClient.getContext()!.newPage(); const aiClient = new OpenRouterClient(accountConfig, logger); - const monitorService = new MonitorService(redditApi, postsRepo, messagesRepo, accountConfig, logger); + const monitorService = new MonitorService( + redditApi, postsRepo, recipientDeduplicator, accountConfig, logger, + ); const processorService = new ProcessorService(redditApi, aiClient, accountConfig, logger); processorService.setStrategySelector(new StrategySelector(accountConfig, messagesRepo, logger)); processorService.setChatStrategySelector(new StrategySelector(accountConfig, messagesRepo, logger, 'chatStrategies')); @@ -146,7 +156,9 @@ async function buildRuntime( monitorService, processorService, responseChecker, chatService, mediaService, mediaSyncService, rateLimiter, }; - const lifecycle = new AccountRuntimeLoop(resources, accountConfig, logger, isBotRunning); + const lifecycle = new AccountRuntimeLoop( + resources, accountConfig, logger, isBotRunning, recipientDeduplicator, + ); return new RedditAccountRuntime(session, resources, lifecycle); } catch (error) { await browserClient.close(); @@ -175,6 +187,7 @@ class AccountRuntimeLoop { private readonly config: Config, private readonly logger: Logger, private readonly isBotRunning: () => boolean, + private readonly recipientDeduplicator: RecipientDeduplicator, ) {} start(): void { @@ -257,8 +270,26 @@ class AccountRuntimeLoop { } postsRepo.updateAnalysis(post.id, processed.imageAnalysis, processed.generatedMessage); await rateLimiter.throttle(async () => { - const chat = await chatService.startConversation(post.author, processed.generatedMessage, post.author_fullname); - if (!chat) return false; + if (!this.recipientDeduplicator.tryReserve(post.author)) { + postsRepo.setError(post.id, 'User already contacted across accounts'); + this.resources.activityLogRepo.insert({ + timestamp: Date.now(), type: 'skipped', + message: `Skipped u/${post.author} (already contacted across accounts)`, + recipient: post.author, + }); + return false; + } + let chat; + try { + chat = await chatService.startConversation(post.author, processed.generatedMessage, post.author_fullname); + } catch (error) { + this.recipientDeduplicator.release(post.author); + throw error; + } + if (!chat) { + this.recipientDeduplicator.release(post.author); + return false; + } postsRepo.markMessageSent(post.id); messagesRepo.insert({ postId: post.id, recipient: post.author, subject: post.author, diff --git a/src/web/events/express-sse-adapter.ts b/src/web/events/express-sse-adapter.ts index 052f637..01e62a2 100644 --- a/src/web/events/express-sse-adapter.ts +++ b/src/web/events/express-sse-adapter.ts @@ -197,6 +197,14 @@ export function createExpressSseHandler( } try { + // A first-time subscriber repairs authoritative dashboard snapshots on + // open and only needs events published from this point onwards. Replaying + // the entire buffer here can overflow the bounded response queue and + // force EventSource into a reconnect loop. A genuine reconnect supplies + // Last-Event-ID and still receives every buffered event after its cursor. + const requestedLastEventId = request.get('Last-Event-ID'); + const replayCursor = requestedLastEventId + ?? options.hub.recent(1).at(-1)?.id; subscription = options.hub.subscribe( (event) => { if (!active) { @@ -208,7 +216,7 @@ export function createExpressSseHandler( close(true); } }, - { lastEventId: request.get('Last-Event-ID') }, + { lastEventId: replayCursor }, ); } catch { close(true); diff --git a/tests/chat-service-username-persist.test.ts b/tests/chat-service-username-persist.test.ts index 06f09db..45d848e 100644 --- a/tests/chat-service-username-persist.test.ts +++ b/tests/chat-service-username-persist.test.ts @@ -6,7 +6,11 @@ import { runMigrations } from '../src/database/client.js'; import { RedditUsersRepository } from '../src/database/repositories.js'; import { ChatService } from '../src/services/chat/service.js'; import { ChatCache } from '../src/services/chat/cache.js'; -import { MatrixAuthError, MatrixClient } from '../src/services/chat/matrix/client.js'; +import { + MatrixAuthError, + MatrixClient, + MatrixFeatureDisabledError, +} from '../src/services/chat/matrix/client.js'; import { createMockLogger } from './mockLogger.js'; import { authenticatedAccountAuth } from './testUtils.js'; @@ -232,4 +236,54 @@ describe('ChatService username persistence', () => { expect(matrix.getRoomFullState).toHaveBeenCalledOnce(); expect(matrix.getRoomMessages).not.toHaveBeenCalled(); }); + + it('stops room state lookups after the homeserver reports the feature is disabled', async () => { + const logger = createMockLogger(); + const cache = tmpCache(logger); + cache.set(['first', 'second'].map((name) => ({ + username: 'unknown', + roomName: 'unknown', + lastMessageTime: Date.now(), + lastMessageSnippet: '', + isFromUs: false, + unread: false, + roomId: `!${name}:reddit.com`, + messages: [], + }))); + + const matrix = buildMatrix(); + matrix.getRoomFullState = vi.fn().mockRejectedValue( + new MatrixFeatureDisabledError('disabled by configuration'), + ); + matrix.getRoomMessages = vi.fn().mockImplementation(async (roomId: string) => ({ + start: 'start', + chunk: [{ + event_id: `$${roomId}`, + type: 'm.room.message', + sender: `@t2_${roomId.includes('first') ? 'first' : 'second'}:reddit.com`, + origin_server_ts: Date.now(), + content: { msgtype: 'm.text', body: 'hello' }, + }], + })); + const usernameResolver = { resolveT2Ids: vi.fn().mockResolvedValue({}) }; + const svc = new ChatService( + matrix, + logger, + undefined, undefined, undefined, usernameResolver, + cache, + undefined, + undefined, + authenticatedAccountAuth, + ); + + await svc.enrichUsernames(); + + expect(matrix.getRoomFullState).toHaveBeenCalledOnce(); + expect(matrix.getRoomMessages).toHaveBeenCalledTimes(2); + expect(logger.logs.filter((entry) => + entry.message.includes('room state lookup is disabled upstream'), + )).toHaveLength(1); + expect((await svc.getCachedConversations()).map((chat) => chat.username)) + .toEqual(['t2_first', 't2_second']); + }); }); diff --git a/tests/dashboard/conversations.test.tsx b/tests/dashboard/conversations.test.tsx index a4c60dd..ca3f213 100644 --- a/tests/dashboard/conversations.test.tsx +++ b/tests/dashboard/conversations.test.tsx @@ -12,6 +12,7 @@ import { ConversationsPage } from '../../apps/dashboard/src/features/conversatio interface ChatFixtureOptions { username: string; roomId: string; + lastMessageTime?: number; unread?: boolean; isFromUs?: boolean; lastMessageSnippet?: string; @@ -24,7 +25,7 @@ function chatFixture(options: ChatFixtureOptions) { roomName: options.username, roomId: options.roomId, url: `https://chat.reddit.com/room/${options.roomId}`, - lastMessageTime: 1_786_000_000_000, + lastMessageTime: options.lastMessageTime ?? 1_786_000_000_000, lastMessageSnippet: options.lastMessageSnippet ?? `Latest from ${options.username}`, isFromUs: options.isFromUs ?? false, unread: options.unread ?? false, @@ -141,6 +142,36 @@ describe('ConversationsPage', () => { ); }); + it('orders conversations by latest message activity', async () => { + const oldest = chatFixture({ username: 'oldest', roomId: 'room-oldest', lastMessageTime: 100 }); + const newest = chatFixture({ username: 'newest', roomId: 'room-newest', lastMessageTime: 300 }); + const middle = chatFixture({ username: 'middle', roomId: 'room-middle', lastMessageTime: 200 }); + server.use(http.get('/api/conversations', () => HttpResponse.json([oldest, newest, middle]))); + renderConversations(); + + await screen.findByRole('button', { name: /u\/newest/ }); + expect(screen.getAllByRole('button', { name: /u\/(?:oldest|newest|middle)/ }).map((button) => button.getAttribute('aria-label'))) + .toEqual([ + expect.stringContaining('u/newest'), + expect.stringContaining('u/middle'), + expect.stringContaining('u/oldest'), + ]); + }); + + it('shows short relative times for the latest message', async () => { + const now = Date.now(); + const minutesAgo = chatFixture({ username: 'minutes', roomId: 'room-minutes', lastMessageTime: now - (5 * 60_000) }); + const daysAgo = chatFixture({ username: 'days', roomId: 'room-days', lastMessageTime: now - (2 * 86_400_000) }); + server.use(http.get('/api/conversations', () => HttpResponse.json([daysAgo, minutesAgo]))); + renderConversations(); + + const recentTime = await screen.findByText('5m ago'); + expect(recentTime.tagName).toBe('TIME'); + expect(recentTime).toHaveAttribute('dateTime', new Date(minutesAgo.lastMessageTime).toISOString()); + expect(recentTime).toHaveAttribute('title'); + expect(screen.getByText('2d ago')).toBeInTheDocument(); + }); + it('marks and filters conversations with received images and renders the images in chat', async () => { const imageChat = chatFixture({ username: 'alice', diff --git a/tests/dashboard/media.test.tsx b/tests/dashboard/media.test.tsx index 87cf5ff..24f1a12 100644 --- a/tests/dashboard/media.test.tsx +++ b/tests/dashboard/media.test.tsx @@ -150,6 +150,20 @@ describe('media page', () => { expect(screen.getByText('No media matches these filters.')).toBeInTheDocument(); }); + it('orders media by received time', async () => { + server.use( + http.get('/api/media', () => HttpResponse.json([mediaItems[2], mediaItems[0], mediaItems[1]])), + http.get('/api/media/sync', () => HttpResponse.json(idleStatus)), + ); + renderMedia(); + + expect((await screen.findAllByRole('img')).map((item) => item.getAttribute('alt'))).toEqual([ + 'Media from alice', + 'Media from charlie', + 'Media from dave', + ]); + }); + it('shows empty and recoverable error states', async () => { let requests = 0; server.use( diff --git a/tests/matrix-client.test.ts b/tests/matrix-client.test.ts index fa1ca6e..a07a5c4 100644 --- a/tests/matrix-client.test.ts +++ b/tests/matrix-client.test.ts @@ -2,6 +2,7 @@ import { describe, it, expect, vi } from 'vitest'; import { MatrixAuthError, MatrixClient, + MatrixFeatureDisabledError, MatrixRateLimitError, } from '../src/services/chat/matrix/client.js'; import { createMockLogger } from './mockLogger.js'; @@ -153,6 +154,18 @@ describe('MatrixClient direct room registration', () => { }); describe('MatrixClient authentication classification', () => { + it('classifies an explicitly disabled endpoint as an unavailable feature', async () => { + const client = new MatrixClient(creds, createMockLogger(), { + minRequestIntervalMs: 0, + fetchImpl: vi.fn().mockResolvedValue(new Response(JSON.stringify({ + message: 'disabled by configuration', + }), { status: 403, headers: { 'content-type': 'application/json' } })), + }); + + await expect(client.getRoomFullState('!room:example.test')) + .rejects.toBeInstanceOf(MatrixFeatureDisabledError); + }); + it('does not classify a room-level 403 as an expired account token', async () => { const client = new MatrixClient(creds, createMockLogger(), { minRequestIntervalMs: 0, diff --git a/tests/recipient-deduplicator.test.ts b/tests/recipient-deduplicator.test.ts new file mode 100644 index 0000000..adc2592 --- /dev/null +++ b/tests/recipient-deduplicator.test.ts @@ -0,0 +1,29 @@ +import { describe, expect, it, vi } from 'vitest'; +import { RecipientDeduplicator } from '../src/services/recipient-deduplicator.js'; + +describe('RecipientDeduplicator', () => { + it('reserves a recipient once across username casing variants', () => { + const history = { hasSentTo: vi.fn(() => false) }; + const deduplicator = new RecipientDeduplicator(history); + + expect(deduplicator.tryReserve('Some_User')).toBe(true); + expect(deduplicator.hasSentTo('some_user')).toBe(true); + expect(deduplicator.tryReserve('SOME_USER')).toBe(false); + }); + + it('rejects recipients found in persisted cross-account history', () => { + const history = { hasSentTo: vi.fn((recipient: string) => recipient === 'contacted') }; + const deduplicator = new RecipientDeduplicator(history); + + expect(deduplicator.tryReserve('contacted')).toBe(false); + expect(deduplicator.tryReserve('new-user')).toBe(true); + }); + + it('allows another account to retry after a failed send releases its reservation', () => { + const deduplicator = new RecipientDeduplicator({ hasSentTo: () => false }); + + expect(deduplicator.tryReserve('retry-user')).toBe(true); + deduplicator.release('RETRY-USER'); + expect(deduplicator.tryReserve('retry-user')).toBe(true); + }); +}); diff --git a/tests/repository-account-isolation.test.ts b/tests/repository-account-isolation.test.ts index 8097c01..e92d340 100644 --- a/tests/repository-account-isolation.test.ts +++ b/tests/repository-account-isolation.test.ts @@ -9,6 +9,7 @@ import { DeliveryOutboxRepository, MessagesRepository, PostsRepository, + RecipientHistoryRepository, SuccessfulMessagesRepository, } from '../src/database/repositories.js'; import { buildConversationState } from '../src/services/conversation-state.js'; @@ -75,6 +76,8 @@ describe('account-bound domain repositories', () => { expect(messagesB.getByPostId('same-post')).toMatchObject({ id: messageB, message: 'from B' }); expect(messagesA.hasSentTo('only-a')).toBe(true); expect(messagesB.hasSentTo('only-a')).toBe(false); + const recipientHistory = new RecipientHistoryRepository(db); + expect(recipientHistory.hasSentTo('ONLY-A')).toBe(true); expect(messagesA.getByChatRoomId('same-room')?.recipient).toBe('only-a'); expect(messagesB.getByChatRoomId('same-room')).toBeUndefined(); diff --git a/tests/web/events/express-sse-adapter.test.ts b/tests/web/events/express-sse-adapter.test.ts index d5e640c..657ed09 100644 --- a/tests/web/events/express-sse-adapter.test.ts +++ b/tests/web/events/express-sse-adapter.test.ts @@ -122,7 +122,7 @@ describe('createExpressSseHandler', () => { ]); }); - it('replays the full available window when Last-Event-ID is absent', () => { + it('starts at the live edge when Last-Event-ID is absent', () => { const hub = createHub(); hub.publish('log.created', { level: 'info', message: 'one' }); hub.publish('log.created', { level: 'warn', message: 'two' }); @@ -134,10 +134,12 @@ describe('createExpressSseHandler', () => { vi.fn(), ); + expect(response.frames).toEqual([': connected\n\n']); + + hub.publish('log.created', { level: 'info', message: 'live' }); expect(response.frames.map((frame) => frame.split('\n')[0])).toEqual([ ': connected', - 'id: event-1', - 'id: event-2', + 'id: event-3', ]); });