RedditBot/src/bootstrap.ts

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,
  ConversationReadStateRepository,
  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 { PushNotifier } from './services/push-notifier.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,
  RedditAccountEstablishmentRepository,
} 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 { AccountGrowthRepository } from './database/growth-repository.js';
import { AccountBrowsingSimulationRepository } from './database/browsing-simulation-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;
  conversationReadStateRepo: ConversationReadStateRepository;
  conversationOutcomesRepo: ConversationOutcomesRepository;
  recipientDeduplicator: RecipientDeduplicator;
  candidateQueueRepo: CandidateQueueRepository;
  runtimeStateRepo: RuntimeStateRepository;
  growthRepo: AccountGrowthRepository;
  browsingRepo: AccountBrowsingSimulationRepository;

  redditApi: RedditApiClient;
  aiClient: OpenRouterClient;
  monitorService: MonitorService;
  processorService: ProcessorService;
  strategySelector: StrategySelector;
  responseChecker: ResponseCheckerService;
  mediaService: MediaService;
  chatService: ChatService;
  mediaSyncService: MediaSyncService;
  mediaLibraryService: MediaLibraryService;
  rateLimiter: RateLimiter;
  pushNotifier: PushNotifier;

  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<AppContainer> {
  // 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);
  }

  // Phone push notifications ride the same typed event stream the dashboard
  // consumes, so every notifiable event has exactly one publisher.
  const pushNotifier = new PushNotifier(() => config.pushNotifications, logger);
  pushNotifier.attach(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 conversationReadStateRepo = new ConversationReadStateRepository(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 growthRepo = new AccountGrowthRepository(db);
  const browsingRepo = new AccountBrowsingSimulationRepository(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`);
  }

  // Legacy single-account path: a 401 waits for a new extension dump. The
  // per-account runtimes self-refresh via BrowserMatrixTokenRefresher instead.
  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,
      getBotMode: () => webServer?.getMode() ?? 'running',
    }),
  );
  const redditAccounts = new RedditAccountsService(
    redditAccountsRepo,
    sessionImporter,
    accountAuth,
    accountRuntimeManager,
    browserClient,
    new RedditAccountEstablishmentRepository(db),
  );
  if (config.web.enabled) {
    webServer = new WebServer(
      config.web.port,
      config,
      postsRepo,
      messagesRepo,
      successfulMessagesRepo,
      activityLogRepo,
      logger,
      broadcaster,
      process.cwd(),
      aiUsageRepo,
      runtimeStateRepo.getBotMode(),
      (mode) => runtimeStateRepo.setBotMode(mode),
      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<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,
          });
        }
        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.setMediaReceivedListener(({ conversationId, username, count }) => {
      webServer!.logActivity('response', `Image received from u/${username}`, username);
      webServer!.pushLog('info', `🖼️ Image from u/${username}`);
      broadcaster.dashboardEvents.mediaReceived({ conversationId, username, count });
    });
    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.setConversationReadStateRepository(conversationReadStateRepo);
    webServer.setCandidateQueueRepository(candidateQueueRepo);
    webServer.setGrowthRepository(growthRepo);
    webServer.setBrowsingRepository(browsingRepo);
    webServer.setConfigStore(new ConfigStore(db));
    webServer.setPushNotifier(pushNotifier);
    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,
    conversationReadStateRepo,
    conversationOutcomesRepo,
    recipientDeduplicator,
    candidateQueueRepo,
    runtimeStateRepo,
    growthRepo,
    browsingRepo,
    redditApi,
    aiClient,
    monitorService,
    processorService,
    strategySelector,
    responseChecker,
    mediaService,
    chatService,
    mediaSyncService,
    mediaLibraryService,
    rateLimiter,
    pushNotifier,
    webServer,
  };
}