diff --git a/src/bootstrap.ts b/src/bootstrap.ts index f17135a..b9f0681 100644 --- a/src/bootstrap.ts +++ b/src/bootstrap.ts @@ -33,6 +33,7 @@ import { MediaSyncService } from './services/media-sync.js'; import { RateLimiter } from './utils/rate-limiter.js'; import { fetchWithTimeout } from './utils/fetch-with-timeout.js'; import { WebServer } from './web/server.js'; +import { reportReply } from './web/event-reporting.js'; export interface AppContainer { config: Config; @@ -126,6 +127,10 @@ export async function bootstrap(config: Config, logger: Logger): Promise { webServer!.broadcastSyncProgress(progress as Record | null); + if (progress) { + const syncId = `chat-${progress.startedAt}`; + if (progress.phase === 'starting') { + broadcaster.dashboardEvents.sync({ phase: 'started', id: syncId, kind: progress.kind }); + } else { + broadcaster.dashboardEvents.sync({ + phase: 'progress', + id: syncId, + kind: progress.kind, + current: progress.current, + total: progress.total, + label: progress.label ?? progress.phase, + }); + } + lastSyncProgress = progress; + } else if (lastSyncProgress) { + broadcaster.dashboardEvents.sync({ + phase: 'completed', + id: `chat-${lastSyncProgress.startedAt}`, + kind: lastSyncProgress.kind, + processed: lastSyncProgress.current, + }); + lastSyncProgress = null; + } }); - chatService.setNewReplyListener(({ username, replySnippet }) => { + chatService.setNewReplyListener(({ conversationId, replyId, username, replySnippet }) => { webServer!.logActivity( 'response', `Reply from u/${username}: ${replySnippet}`, username, ); webServer!.pushLog('info', `💬 Reply from u/${username}`); + reportReply(broadcaster.dashboardEvents, { + conversationId, + replyId, + username, + body: replySnippet, + }); }); webServer.setMediaService(mediaService); webServer.setMediaSyncService(mediaSyncService); diff --git a/src/main-loop.ts b/src/main-loop.ts index d36d57f..f79bdd3 100644 --- a/src/main-loop.ts +++ b/src/main-loop.ts @@ -1,6 +1,7 @@ import type { AppContainer } from './bootstrap.js'; import { withRetry } from './utils/retry.js'; import { isSkip } from './services/processor.js'; +import { reportDeliveryResult, reportInvitationCooldown } from './web/event-reporting.js'; function sleep(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); @@ -93,6 +94,7 @@ export async function runMainLoop(app: AppContainer): Promise { let lastPausedLog = 0; let lastCooldownLog = 0; + let reportedCooldownUntil: number | null = null; const COOLDOWN_LOG_INTERVAL_MS = 5 * 60_000; while (!isShuttingDown) { @@ -110,6 +112,10 @@ export async function runMainLoop(app: AppContainer): Promise { // pull new posts — they'd just queue up and skip. Wait the cooldown out. const cooldown = chatService.getInvitationCooldown(); if (cooldown) { + if (webServer && reportedCooldownUntil !== cooldown.until) { + reportInvitationCooldown(webServer.dashboardEvents, cooldown); + reportedCooldownUntil = cooldown.until; + } const minutesLeft = Math.ceil((cooldown.until - Date.now()) / 60_000); if (Date.now() - lastCooldownLog > COOLDOWN_LOG_INTERVAL_MS) { logger.warn(`Chat invitation cooldown active — ~${minutesLeft}m left. ${cooldown.reason}`); @@ -122,6 +128,10 @@ export async function runMainLoop(app: AppContainer): Promise { await sleep(config.reddit.pollIntervalMs); continue; } + if (webServer && reportedCooldownUntil !== null) { + reportInvitationCooldown(webServer.dashboardEvents, null); + reportedCooldownUntil = null; + } lastCooldownLog = 0; const posts = await monitorService.getUnprocessedPosts(); @@ -242,6 +252,14 @@ export async function runMainLoop(app: AppContainer): Promise { ); webServer?.pushLog('info', `✅ Message sent to u/${post.author} (${currentCount}/${config.rateLimit.maxMessagesPerHour} this hour)`); webServer?.updateStats(currentCount); + if (webServer) { + reportDeliveryResult(webServer.dashboardEvents, { + success: true, + messageId: post.id, + username: post.author, + conversationId: chat.roomId, + }); + } } else { postsRepo.setError(post.id, 'Failed to send message'); webServer?.logActivity( @@ -251,6 +269,14 @@ export async function runMainLoop(app: AppContainer): Promise { postImageUrl ?? undefined, ); webServer?.pushLog('error', `❌ Failed to send message to u/${post.author}`); + if (webServer) { + reportDeliveryResult(webServer.dashboardEvents, { + success: false, + messageId: post.id, + username: post.author, + error: 'Chat service did not create a conversation', + }); + } } return success; diff --git a/src/services/chat/service.ts b/src/services/chat/service.ts index 0506e67..8339e9f 100644 --- a/src/services/chat/service.ts +++ b/src/services/chat/service.ts @@ -103,6 +103,8 @@ 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: { + conversationId: string; + replyId: string; username: string; replySnippet: string; }) => void; @@ -806,6 +808,8 @@ export class ChatService { this.logger.info(`[chat] stored successful chat example from ${chat.username}`); try { this.newReplyListener?.({ + conversationId: chat.roomId, + replyId: `${chat.roomId}:${firstReply?.timestamp ?? Date.now()}`, username: chat.username, replySnippet: firstReply?.text.slice(0, 200) ?? '', }); diff --git a/src/web/event-reporting.ts b/src/web/event-reporting.ts new file mode 100644 index 0000000..eb4d7f1 --- /dev/null +++ b/src/web/event-reporting.ts @@ -0,0 +1,53 @@ +interface DeliveryEvents { + messageSent(input: { id: string; username: string; conversationId?: string }): unknown; + messageFailed(input: { username: string; error: unknown; messageId?: string; conversationId?: string }): unknown; +} + +interface ConversationEvents { + conversation(input: { id: string; state: 'replied'; unread?: number }): unknown; + reply(input: { conversationId: string; id: string; username: string; body?: string }): unknown; +} + +interface RateLimitEvents { + rateLimit(input: { blocked: boolean; resetAt?: number; reason?: string }): unknown; +} + +interface DeliveryResult { + success: boolean; + messageId: string; + username: string; + conversationId?: string; + error?: unknown; +} + +export function reportDeliveryResult(events: DeliveryEvents, result: DeliveryResult): void { + if (result.success) { + events.messageSent({ id: result.messageId, username: result.username, conversationId: result.conversationId }); + return; + } + events.messageFailed({ + messageId: result.messageId, + username: result.username, + conversationId: result.conversationId, + error: result.error ?? 'Message delivery failed', + }); +} + +export function reportReply( + events: ConversationEvents, + reply: { conversationId: string; replyId: string; username: string; body?: string }, +): void { + events.conversation({ id: reply.conversationId, state: 'replied', unread: 1 }); + events.reply({ conversationId: reply.conversationId, id: reply.replyId, username: reply.username, body: reply.body }); +} + +export function reportInvitationCooldown( + events: RateLimitEvents, + cooldown: { until: number; reason: string } | null, +): void { + if (cooldown) { + events.rateLimit({ blocked: true, resetAt: cooldown.until, reason: cooldown.reason }); + return; + } + events.rateLimit({ blocked: false }); +} diff --git a/src/web/server.ts b/src/web/server.ts index 405c6a2..c9e2b7c 100644 --- a/src/web/server.ts +++ b/src/web/server.ts @@ -163,6 +163,10 @@ export class WebServer { return this.botRunning; } + get dashboardEvents(): LogBroadcaster['dashboardEvents'] { + return this.broadcaster.dashboardEvents; + } + setChatService(chatService: ChatService): void { this.services.chat = chatService; } diff --git a/tests/web/event-reporting.test.ts b/tests/web/event-reporting.test.ts new file mode 100644 index 0000000..659a192 --- /dev/null +++ b/tests/web/event-reporting.test.ts @@ -0,0 +1,31 @@ +import { describe, expect, it, vi } from 'vitest'; +import { + reportDeliveryResult, + reportInvitationCooldown, + reportReply, +} from '../../src/web/event-reporting.js'; + +describe('dashboard event reporting', () => { + it('reports authoritative delivery success and failure', () => { + const events = { messageSent: vi.fn(), messageFailed: vi.fn() }; + + reportDeliveryResult(events, { success: true, messageId: 'post-1', username: 'alice', conversationId: 'room-1' }); + reportDeliveryResult(events, { success: false, messageId: 'post-2', username: 'bob', error: 'Matrix rejected the message' }); + + expect(events.messageSent).toHaveBeenCalledWith({ id: 'post-1', username: 'alice', conversationId: 'room-1' }); + expect(events.messageFailed).toHaveBeenCalledWith({ messageId: 'post-2', username: 'bob', error: 'Matrix rejected the message' }); + }); + + it('reports replies and invitation cooldown state', () => { + const events = { conversation: vi.fn(), reply: vi.fn(), rateLimit: vi.fn() }; + + reportReply(events, { conversationId: 'room-1', replyId: 'reply-1', username: 'alice', body: 'Hello there' }); + reportInvitationCooldown(events, { until: 10_000, reason: 'Reddit asked us to slow down' }); + reportInvitationCooldown(events, null); + + expect(events.conversation).toHaveBeenCalledWith({ id: 'room-1', state: 'replied', unread: 1 }); + expect(events.reply).toHaveBeenCalledWith({ conversationId: 'room-1', id: 'reply-1', username: 'alice', body: 'Hello there' }); + expect(events.rateLimit).toHaveBeenNthCalledWith(1, { blocked: true, resetAt: 10_000, reason: 'Reddit asked us to slow down' }); + expect(events.rateLimit).toHaveBeenNthCalledWith(2, { blocked: false }); + }); +});