diff --git a/docs/services.md b/docs/services.md index 921e1fb..18ba4f0 100644 --- a/docs/services.md +++ b/docs/services.md @@ -11,6 +11,7 @@ This is an ownership map, not an API dump. Read the owning implementation and te | `services/reddit-account-runtime.ts` | Immutable account runtime/factory/manager lifecycle | Same-account operations serialize; session refresh replaces one runtime. | | `services/reddit-user-messenger.ts` | Account selection for explicit API/MCP messages | Prefer the newest existing conversation owner; otherwise choose the authenticated running account with the fewest hourly sends. | | `services/reddit-account-runtime-builder.ts` | Concrete account dependency graph and account loop | Account-specific browser, Matrix, repos, limiter, automation, chat, and media live here. | +| `services/account-runtime-event-wiring.ts` | Per-account chat-to-realtime event bridge | Every account runtime publishes reply, media, and sync events into the shared dashboard/push stream; sync IDs remain account-scoped. | | `services/reddit-accounts.ts` | Account import/activation orchestration | Verification must bind Reddit cookies and Matrix identity before activation. | | `services/account-auth-context.ts` | Current verified identity snapshot | Account work pauses unless both sides authenticate. | | `services/session-bound-runtime.ts` | Legacy auth-bound background lifecycle | Prevents background work while logged out. | diff --git a/src/bootstrap.ts b/src/bootstrap.ts index 8998951..dfcdd45 100644 --- a/src/bootstrap.ts +++ b/src/bootstrap.ts @@ -373,6 +373,7 @@ export async function bootstrap( logger, db, getBotMode: () => webServer?.getMode() ?? 'running', + events: broadcaster.dashboardEvents, }), ); const redditAccounts = new RedditAccountsService( diff --git a/src/services/account-runtime-event-wiring.ts b/src/services/account-runtime-event-wiring.ts new file mode 100644 index 0000000..bac0fe8 --- /dev/null +++ b/src/services/account-runtime-event-wiring.ts @@ -0,0 +1,82 @@ +import { + type ChatService, + type SyncCompletion, + type SyncProgress, +} from './chat/index.js'; + +/** Events emitted by one account-owned chat service into the shared UI/push hub. */ +export interface AccountRuntimeEventReporter { + conversation(input: { id: string; state: 'replied'; unread: number }): unknown; + reply(input: { conversationId: string; id: string; username: string; body?: string }): unknown; + mediaReceived(input: { conversationId: string; username: string; count: number }): unknown; + sync(input: + | { phase: 'started'; id: string; kind: 'delta' | 'full' } + | { phase: 'progress'; id: string; kind: 'delta' | 'full'; current: number; total?: number; label?: string } + | { phase: 'completed'; id: string; kind: 'delta' | 'full'; processed: number } + | { phase: 'failed'; id: string; kind: 'delta' | 'full'; error: string } + ): unknown; +} + +type AccountChatEventSource = Pick< + ChatService, + 'setProgressListener' | 'setNewReplyListener' | 'setMediaReceivedListener' +>; + +/** Connects one account-owned ChatService to the process-wide event stream. */ +export function wireAccountChatEvents( + chatService: AccountChatEventSource, + events: AccountRuntimeEventReporter, + accountId: string, +): void { + let activeSync: SyncProgress | null = null; + + chatService.setProgressListener((progress, completion) => { + if (progress) { + const id = accountSyncId(accountId, progress.startedAt); + if (progress.phase === 'starting') { + events.sync({ phase: 'started', id, kind: progress.kind }); + } else { + events.sync({ + phase: 'progress', + id, + kind: progress.kind, + current: progress.current, + total: progress.total, + label: progress.label ?? progress.phase, + }); + } + activeSync = progress; + return; + } + + if (!activeSync) return; + reportSyncCompletion(events, accountId, activeSync, completion); + activeSync = null; + }); + + chatService.setMediaReceivedListener(({ conversationId, username, count }) => { + events.mediaReceived({ conversationId, username, count }); + }); + chatService.setNewReplyListener(({ conversationId, replyId, username, replySnippet }) => { + events.conversation({ id: conversationId, state: 'replied', unread: 1 }); + events.reply({ conversationId, id: replyId, username, body: replySnippet }); + }); +} + +function reportSyncCompletion( + events: AccountRuntimeEventReporter, + accountId: string, + progress: SyncProgress, + completion?: SyncCompletion, +): void { + const id = accountSyncId(accountId, progress.startedAt); + if (completion?.status === 'failed') { + events.sync({ phase: 'failed', id, kind: progress.kind, error: completion.error }); + return; + } + events.sync({ phase: 'completed', id, kind: progress.kind, processed: progress.current }); +} + +function accountSyncId(accountId: string, startedAt: number): string { + return `chat-${accountId}-${startedAt}`; +} diff --git a/src/services/reddit-account-runtime-builder.ts b/src/services/reddit-account-runtime-builder.ts index a27cfaa..4590754 100644 --- a/src/services/reddit-account-runtime-builder.ts +++ b/src/services/reddit-account-runtime-builder.ts @@ -66,12 +66,17 @@ import { AccountGrowthRepository } from '../database/growth-repository.js'; import { AccountBrowsingSimulationRepository } from '../database/browsing-simulation-repository.js'; import { AccountBrowsingSimulationService } from './account-browsing-simulation.js'; import { allowsConversationSync, allowsSending, type BotMode } from './bot-mode.js'; +import { + wireAccountChatEvents, + type AccountRuntimeEventReporter, +} from './account-runtime-event-wiring.js'; export interface ConcreteRuntimeFactoryOptions { config: Config; logger: Logger; db: Database.Database; getBotMode?: () => BotMode; + events?: AccountRuntimeEventReporter; } export function materializePreparedPostForAccount( @@ -106,7 +111,7 @@ export function createConcreteRedditAccountRuntimeFactory( } async function buildRuntime( - { config, logger: rootLogger, db, getBotMode = () => 'running' }: ConcreteRuntimeFactoryOptions, + { config, logger: rootLogger, db, getBotMode = () => 'running', events }: ConcreteRuntimeFactoryOptions, session: VerifiedAccountSession, recipientDeduplicator: RecipientDeduplicator, redditRequestCoordinator: RedditRequestCoordinator, @@ -228,6 +233,7 @@ async function buildRuntime( tokenRefresher, usernameResolver, chatCache, messagesRepo, redditUsersRepo, accountAuth, conversationOutcomesRepo, conversationStateRepo, ); + if (events) wireAccountChatEvents(chatService, events, session.accountId); await chatService.loadCache(); const mediaSyncService = new MediaSyncService(chatService, logger, accountConfig.media.syncIntervalMs, accountAuth); const recipientAccountStatusRepo = new RedditRecipientAccountStatusRepository(db); diff --git a/tests/account-runtime-event-wiring.test.ts b/tests/account-runtime-event-wiring.test.ts new file mode 100644 index 0000000..9bac6cf --- /dev/null +++ b/tests/account-runtime-event-wiring.test.ts @@ -0,0 +1,87 @@ +import { describe, expect, it, vi } from 'vitest'; +import { + wireAccountChatEvents, + type AccountRuntimeEventReporter, +} from '../src/services/account-runtime-event-wiring.js'; +import type { + MediaReceivedListener, + NewReplyListener, + SyncProgressListener, +} from '../src/services/chat/index.js'; + +function harness() { + let progressListener: SyncProgressListener | null = null; + let replyListener: NewReplyListener | null = null; + let mediaListener: MediaReceivedListener | null = null; + const chatService = { + setProgressListener: vi.fn((listener: SyncProgressListener | null) => { progressListener = listener; }), + setNewReplyListener: vi.fn((listener: NewReplyListener | null) => { replyListener = listener; }), + setMediaReceivedListener: vi.fn((listener: MediaReceivedListener | null) => { mediaListener = listener; }), + }; + const events = { + conversation: vi.fn(), + reply: vi.fn(), + mediaReceived: vi.fn(), + sync: vi.fn(), + } satisfies AccountRuntimeEventReporter; + + wireAccountChatEvents(chatService, events, '@t2_account:reddit.com'); + return { + events, + progress: (value: Parameters[0], completion?: Parameters[1]) => { + if (!progressListener) throw new Error('Progress listener was not installed'); + progressListener(value, completion); + }, + reply: (value: Parameters[0]) => { + if (!replyListener) throw new Error('Reply listener was not installed'); + replyListener(value); + }, + media: (value: Parameters[0]) => { + if (!mediaListener) throw new Error('Media listener was not installed'); + mediaListener(value); + }, + }; +} + +describe('wireAccountChatEvents', () => { + it('publishes replies and media received by an account-owned chat service', () => { + const fixture = harness(); + + fixture.reply({ + conversationId: '!room:reddit.com', replyId: 'event-1', username: 'alice', replySnippet: 'hello', + }); + fixture.media({ conversationId: '!room:reddit.com', username: 'alice', count: 2 }); + + expect(fixture.events.conversation).toHaveBeenCalledWith({ + id: '!room:reddit.com', state: 'replied', unread: 1, + }); + expect(fixture.events.reply).toHaveBeenCalledWith({ + conversationId: '!room:reddit.com', id: 'event-1', username: 'alice', body: 'hello', + }); + expect(fixture.events.mediaReceived).toHaveBeenCalledWith({ + conversationId: '!room:reddit.com', username: 'alice', count: 2, + }); + }); + + it('publishes account-scoped sync progress and failures', () => { + const fixture = harness(); + const progress = { + kind: 'delta', phase: 'starting', current: 0, total: 4, startedAt: 123, + } as const; + + fixture.progress(progress); + fixture.progress({ ...progress, phase: 'listing-rooms', current: 2, label: 'Loading rooms' }); + fixture.progress(null, { status: 'failed', error: 'Matrix unavailable' }); + + expect(fixture.events.sync).toHaveBeenNthCalledWith(1, { + phase: 'started', id: 'chat-@t2_account:reddit.com-123', kind: 'delta', + }); + expect(fixture.events.sync).toHaveBeenNthCalledWith(2, { + phase: 'progress', id: 'chat-@t2_account:reddit.com-123', kind: 'delta', + current: 2, total: 4, label: 'Loading rooms', + }); + expect(fixture.events.sync).toHaveBeenNthCalledWith(3, { + phase: 'failed', id: 'chat-@t2_account:reddit.com-123', kind: 'delta', error: 'Matrix unavailable', + }); + }); +});