import type { Config } from './config/index.js'; import { attachLogSink, type Logger } from './utils/logger.js'; import { LogBroadcaster } from './web/log-broadcaster.js'; import type Database from 'better-sqlite3'; import { createDatabase, runMigrations } from './database/client.js'; import { PostsRepository, MessagesRepository, SuccessfulMessagesRepository, ActivityLogRepository, RedditUsersRepository, ConversationStateRepository, ConversationOutcomesRepository, DeliveryOutboxRepository, RecipientHistoryRepository, } from './database/repositories.js'; import { BrowserClient } from './browser/client.js'; import { RedditApiClient } from './reddit/api.js'; import { RedditRequestCoordinator } from './reddit/request-coordinator.js'; import { OpenRouterClient } from './ai/openrouter.js'; import { MonitorService } from './services/monitor.js'; import { ProcessorService } from './services/processor.js'; import { ResponseCheckerService } from './services/response-checker.js'; import { StrategySelector } from './services/strategy-selector.js'; import { ChatService, MatrixClient, MatrixMediaDownloader, MatrixUsernameResolver, MatrixTokenRefresher, loadCachedCredentials, saveCredentials, type UsernameResolver, type MatrixCredentials, } from './services/chat/index.js'; import { MediaService } from './services/media.js'; import { MediaSyncService } from './services/media-sync.js'; import { RateLimiter } from './utils/rate-limiter.js'; import { WebServer } from './web/server.js'; import { reportReply } from './web/event-reporting.js'; import { RedditSessionImporter } from './services/reddit-session-importer.js'; import { AccountAuthContext, authenticateCurrentAccount } from './services/account-auth-context.js'; import { ConfigStore } from './config/persistence.js'; import { RedditAccountsRepository, AccountContentRepository } from './database/account-repositories.js'; import { RedditAccountsService } from './services/reddit-accounts.js'; import type { Cookie } from 'playwright'; 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 { AiUsageRepository } from './database/ai-usage-repository.js'; import { importLegacyAccountContent } from './database/legacy-content-import.js'; import { RecipientDeduplicator } from './services/recipient-deduplicator.js'; import { createConfigBackedSubredditPollingControl } from './services/subreddit-polling-control.js'; import { CandidateQueueRepository, RuntimeStateRepository } from './database/candidate-queue-repository.js'; import { MediaLibraryRepository } from './database/media-library-repository.js'; import { MediaLibraryService } from './services/media-library.js'; export interface AppContainer { config: Config; logger: Logger; db: Database.Database; browserClient: BrowserClient; accountAuth: AccountAuthContext; redditAccountsRepo: RedditAccountsRepository; accountRuntimeManager: RedditAccountRuntimeManager; postsRepo: PostsRepository; messagesRepo: MessagesRepository; successfulMessagesRepo: SuccessfulMessagesRepository; deliveryOutboxRepo: DeliveryOutboxRepository; conversationStateRepo: ConversationStateRepository; conversationOutcomesRepo: ConversationOutcomesRepository; recipientDeduplicator: RecipientDeduplicator; candidateQueueRepo: CandidateQueueRepository; runtimeStateRepo: RuntimeStateRepository; redditApi: RedditApiClient; aiClient: OpenRouterClient; monitorService: MonitorService; processorService: ProcessorService; strategySelector: StrategySelector; responseChecker: ResponseCheckerService; mediaService: MediaService; chatService: ChatService; mediaSyncService: MediaSyncService; mediaLibraryService: MediaLibraryService; rateLimiter: RateLimiter; webServer: WebServer | null; } /** * Builds and wires every service required for the bot. * * Performs: * 1. DB + repos * 2. Managed browser + imported-session verification * 3. Three pages (compose / response-check / web-UI) * 4. All services * 5. Optional WebServer * * Returns a populated container. Caller owns lifecycle (start loop, handle * shutdown). */ export async function bootstrap( config: Config, logger: Logger, existingDatabase?: Database.Database, ): Promise { // Wire winston → dashboard log buffer immediately so the matrix-token // sniff and other early bootstrap diagnostics are visible in /api/logs. const broadcaster = new LogBroadcaster(); if (config.web.enabled) { attachLogSink(logger, broadcaster); } // Database const db = existingDatabase ?? createDatabase(config); runMigrations(db); const bootstrapAccountsRepo = new RedditAccountsRepository(db); const bootstrapAccount = bootstrapAccountsRepo.listPublic() .filter((account) => account.status === 'verified') .sort((left, right) => right.updatedAt - left.updatedAt)[0]; const bootstrapAccountId = bootstrapAccount?.accountId; const postsRepo = new PostsRepository(db, bootstrapAccountId); const messagesRepo = new MessagesRepository(db, bootstrapAccountId); const successfulMessagesRepo = new SuccessfulMessagesRepository(db, bootstrapAccountId); const activityLogRepo = new ActivityLogRepository(db, bootstrapAccountId); const redditUsersRepo = new RedditUsersRepository(db); 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 candidateQueueRepo = new CandidateQueueRepository(db); const runtimeStateRepo = new RuntimeStateRepository(db); const redditAccountsRepo = bootstrapAccountsRepo; const accountContentRepo = new AccountContentRepository(db); const aiUsageRepo = new AiUsageRepository(db); const legacyImport = importLegacyAccountContent(db); if (legacyImport.conversations > 0 || legacyImport.media > 0) { logger.info(`Imported legacy unassigned content into SQLite (${legacyImport.conversations} conversations, ${legacyImport.media} media items)`); } logger.info('Database initialized'); // Browser const browserClient = new BrowserClient(config, logger); const accountAuth = new AccountAuthContext(logger); await browserClient.initialize(); const storedAccount = bootstrapAccount; if (storedAccount) { const privateAccount = redditAccountsRepo.get(storedAccount.accountId); if (privateAccount) { try { await browserClient.importRedditCookies(JSON.parse(privateAccount.cookiesJson) as Cookie[]); } catch (error) { logger.warn(`Could not restore stored Reddit cookies for u/${storedAccount.redditUsername}:`, error); } } } // The extension import is the only session acquisition path. A missing // session is recoverable: the dashboard remains available while all // account-bound work waits for a verified dump. logger.info('Verifying Reddit login status...'); const redditLoggedIn = await browserClient.isSessionValid(); if (!redditLoggedIn) { logger.warn('Reddit session is logged out. Account work is paused until a session dump is imported from the extension.'); } // Reddit API client (shares browser context for auth) const redditApi = new RedditApiClient(config, logger, new RedditRequestCoordinator()); redditApi.setBrowserContext(browserClient.getContext()!); // Dedicated page for the response checker (background scraping that // doesn't conflict with the dashboard). const chatPage = await browserClient.getContext()!.newPage(); // AI + services const aiClient = new OpenRouterClient(config, logger, undefined, (usage) => { aiUsageRepo.record({ ...usage, accountId: accountAuth.getSnapshot().accountId }); }); const mediaLibraryService = new MediaLibraryService( new MediaLibraryRepository(db), aiClient, logger, ); const monitorService = new MonitorService( redditApi, postsRepo, recipientDeduplicator, config, logger, 1000, createConfigBackedSubredditPollingControl(config, new ConfigStore(db), logger), ); const processorService = new ProcessorService( redditApi, aiClient, config, logger, undefined, (sourceType) => messagesRepo .getRecent(20) .filter((message) => message.sourceType === sourceType || (sourceType === 'image' && !message.sourceType)) .map((message) => message.message), ); const strategySelector = new StrategySelector(config, messagesRepo, logger); processorService.setStrategySelector(strategySelector); const chatStrategySelector = new StrategySelector(config, messagesRepo, logger, 'chatStrategies'); processorService.setChatStrategySelector(chatStrategySelector); const responseChecker = new ResponseCheckerService( chatPage, messagesRepo, successfulMessagesRepo, postsRepo, logger, accountAuth, conversationOutcomesRepo, ); // Matrix credentials come only from a verified extension dump (or its // persisted cache on restart). The backend never opens Reddit Chat to // sniff or refresh a token. const storedPrivateAccount = storedAccount ? redditAccountsRepo.get(storedAccount.accountId) : null; let matrixCreds: MatrixCredentials | null = storedPrivateAccount ? JSON.parse(storedPrivateAccount.matrixCredentialsJson) as MatrixCredentials : await loadCachedCredentials(); if (matrixCreds) { logger.info(`Loaded cached Matrix credentials (user=${matrixCreds.userId})`); } if (!matrixCreds) { matrixCreds = { homeserver: 'https://matrix.redditspace.com', accessToken: '', userId: '', capturedAt: Date.now(), }; } let activeMatrixClient = new MatrixClient(matrixCreds, logger); accountAuth.setAuthenticator(() => authenticateCurrentAccount({ getRedditIdentity: () => browserClient.getAuthenticatedIdentity(), refreshMatrixCredentials: async () => { try { const userId = await activeMatrixClient.whoami(); return userId ? { userId } : null; } catch { return null; } }, })); const fullyAuthenticated = redditLoggedIn && await accountAuth.ensureFullyAuthenticated(); if (fullyAuthenticated) logger.info('Reddit and Matrix authentication verified'); broadcaster.dashboardEvents.redditSession({ state: fullyAuthenticated ? 'authenticated' : 'login-required', username: fullyAuthenticated ? accountAuth.getSnapshot().redditUsername ?? undefined : undefined, }); // 2. Build the media downloader + service on top of the Matrix client so // `mxc://` URIs resolve through the same auth + rate-limiter pipeline. const mediaDownloader = new MatrixMediaDownloader(activeMatrixClient, logger); const mediaService = new MediaService( mediaDownloader, logger, undefined, accountContentRepo, () => accountAuth.getSnapshot().accountId, ); const indexedMedia = await mediaService.getMediaItems(); const receivedImageRooms = new Set( indexedMedia .filter((item) => item.sender.toLowerCase() !== 'me' && item.chatRoomId) .map((item) => item.chatRoomId as string), ); let backfilledImageOutcomes = 0; for (const roomId of receivedImageRooms) { const message = messagesRepo.getByChatRoomId(roomId); const hasPerformanceContext = Boolean(message?.postId || message?.strategyId || message?.subreddit); if (!message || !hasPerformanceContext || message.gotImage) continue; messagesRepo.markGotImage(message.id); backfilledImageOutcomes += 1; } if (backfilledImageOutcomes > 0) { logger.info(`Backfilled ${backfilledImageOutcomes} received-image performance outcome(s) from the media index`); } // A 401 waits for a new extension dump instead of opening a browser login // or token-sniff flow. const tokenRefresher = new MatrixTokenRefresher(); // 4. Username resolver — Matrix profiles expose Reddit display names using // the same authenticated session that powers chat sync. const matrixUsernameResolver = new MatrixUsernameResolver(activeMatrixClient); const usernameResolver: UsernameResolver = { resolveT2Ids: (ids) => matrixUsernameResolver.resolveT2Ids(ids), resolveUsernameToT2: (username) => redditApi.resolveUsernameToT2(username), }; const chatService = new ChatService( activeMatrixClient, logger, successfulMessagesRepo, mediaService, tokenRefresher, usernameResolver, new ChatCache(logger, undefined, accountContentRepo, storedAccount?.accountId ?? null), messagesRepo, redditUsersRepo, accountAuth, conversationOutcomesRepo, conversationStateRepo, ); const mediaSyncService = new MediaSyncService(chatService, logger, config.media.syncIntervalMs, accountAuth); const rateLimiter = new RateLimiter(config, logger, undefined, { sentMessageHistory: async (since) => messagesRepo.getSentTimestampsSince(since), }); await rateLimiter.initialize(); await chatService.loadCache(); const sessionImporter = new RedditSessionImporter( browserClient, { setMatrixClient(client) { activeMatrixClient = client; chatService.setMatrixClient(client); mediaDownloader.setMatrixClient(client); matrixUsernameResolver.setMatrixClient(client); }, switchAccount(accountId) { return chatService.switchAccount(accountId); }, }, (credentials) => new MatrixClient(credentials, logger), saveCredentials, accountAuth, { save(input) { redditAccountsRepo.upsertVerified({ accountId: input.accountId, redditUsername: input.redditUsername, matrixUserId: input.matrixUserId, cookiesJson: JSON.stringify(input.cookies), matrixCredentialsJson: JSON.stringify(input.credentials), }); }, }, ); let webServer: WebServer | null = null; const accountRuntimeManager = new RedditAccountRuntimeManager( redditAccountsRepo, createConcreteRedditAccountRuntimeFactory({ config, logger, db, isBotRunning: () => webServer?.isRunning() ?? true, }), ); const redditAccounts = new RedditAccountsService( redditAccountsRepo, sessionImporter, accountAuth, accountRuntimeManager, browserClient, ); if (config.web.enabled) { webServer = new WebServer( config.web.port, config, postsRepo, messagesRepo, successfulMessagesRepo, activityLogRepo, logger, broadcaster, process.cwd(), aiUsageRepo, runtimeStateRepo.isBotRunning(), (running) => runtimeStateRepo.setBotRunning(running), accountContentRepo, ); 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, completion) => { 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) { const id = `chat-${lastSyncProgress.startedAt}`; if (completion?.status === 'failed') { broadcaster.dashboardEvents.sync({ phase: 'failed', id, kind: lastSyncProgress.kind, error: completion.error, }); } else { broadcaster.dashboardEvents.sync({ phase: 'completed', id, 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); webServer.setMediaLibraryService(mediaLibraryService); webServer.setResponseChecker(responseChecker); webServer.setRateLimiter(rateLimiter); webServer.setAiClient(aiClient); webServer.setStrategySelector(strategySelector); webServer.setBrowserClient(browserClient); webServer.setSessionImporter(sessionImporter); webServer.setAccountAuth(accountAuth); webServer.setRedditAccountsService(redditAccounts); webServer.setAccountRuntimeManager(accountRuntimeManager); webServer.setConversationStateRepository(conversationStateRepo); webServer.setCandidateQueueRepository(candidateQueueRepo); webServer.setConfigStore(new ConfigStore(db)); webServer.setOnConfigChange(() => logger.info('Runtime config updated via dashboard')); } // Start background services if (webServer) webServer.start(); await accountRuntimeManager.startEnabled().catch((error) => { logger.error('One or more Reddit account runtimes failed to start:', error); }); if (accountRuntimeManager.list().length > 0) await browserClient.close(); // Prime processor with successful examples already on disk. const existingExamples = responseChecker.getSuccessfulExamples(3); processorService.setSuccessfulExamples(existingExamples); const successCount = responseChecker.getSuccessCount(); if (successCount > 0) { logger.info(`Loaded ${successCount} successful message(s) from history`); } return { config, logger, db, browserClient, accountAuth, redditAccountsRepo, accountRuntimeManager, postsRepo, messagesRepo, successfulMessagesRepo, deliveryOutboxRepo, conversationStateRepo, conversationOutcomesRepo, recipientDeduplicator, candidateQueueRepo, runtimeStateRepo, redditApi, aiClient, monitorService, processorService, strategySelector, responseChecker, mediaService, chatService, mediaSyncService, mediaLibraryService, rateLimiter, webServer, }; }