feat(web): wire domain events into realtime stream

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<AppCont
   if (loggedIn) {
     logger.info('Reddit login verified');
   }
+  broadcaster.dashboardEvents.redditSession({
+    state: loggedIn ? 'authenticated' : 'login-required',
+    username: loggedIn ? config.reddit.username : undefined,
+  });
 
   // Reddit API client (shares browser context for auth)
   const redditApi = new RedditApiClient(config, logger);
@@ -273,16 +278,47 @@ export async function bootstrap(config: Config, logger: Logger): Promise<AppCont
     webServer.setChatService(chatService);
     // Stream sync-progress frames over the dashboard's SSE channel so the
     // full-sync UI can show live phase + per-room progress.
+    let lastSyncProgress: { kind: 'full' | 'delta'; current: number; startedAt: number } | null = null;
     chatService.setProgressListener((progress) => {
       webServer!.broadcastSyncProgress(progress as Record<string, unknown> | 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,
           });
-    chatService.setNewReplyListener(({ username, replySnippet }) => {
+        }
+        lastSyncProgress = progress;
+      } else if (lastSyncProgress) {
+        broadcaster.dashboardEvents.sync({
+          phase: 'completed',
+          id: `chat-${lastSyncProgress.startedAt}`,
+          kind: lastSyncProgress.kind,
+          processed: lastSyncProgress.current,
+        });
+        lastSyncProgress = null;
+      }
+    });
+    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);

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<void> {
   return new Promise((resolve) => setTimeout(resolve, ms));
@@ -93,6 +94,7 @@ export async function runMainLoop(app: AppContainer): Promise<void> {
 
   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<void> {
       // 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<void> {
         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<void> {
                 );
                 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<void> {
                   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;

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) ?? '',
             });

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 });
+}

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;
   }

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 });
+  });
+});