RedditBot/src/services/reddit-account-runtime.ts

import path from 'node:path';
import type { Cookie } from 'playwright';
import { z } from 'zod';
import type { BrowserClient } from '../browser/client.js';
import type {
  RedditAccountPrivateRecord,
  RedditAccountsRepository,
  AccountContentRepository,
  RedditRecipientAccountStatusRepository,
} from '../database/account-repositories.js';
import type {
  ConversationOutcomesRepository,
  ActivityLogRepository,
  ConversationStateRepository,
  ConversationReadStateRepository,
  DeliveryOutboxRepository,
  MessagesRepository,
  PostsRepository,
  RedditUsersRepository,
  SuccessfulMessagesRepository,
} from '../database/repositories.js';
import type { RedditApiClient } from '../reddit/api.js';
import type { RateLimiter } from '../utils/rate-limiter.js';
import type { AccountAuthContext } from './account-auth-context.js';
import type { ChatCache } from './chat/cache.js';
import type { MatrixClient } from './chat/matrix/client.js';
import type { MatrixCredentials } from './chat/matrix/types.js';
import type { ChatService } from './chat/service.js';
import type { MediaService } from './media.js';
import type { MediaSyncService } from './media-sync.js';
import type { MediaLibraryExchangeService } from './media-library-exchange.js';
import type { MonitorService } from './monitor.js';
import type { ProcessorService } from './processor.js';
import type { ResponseCheckerService } from './response-checker.js';
import type { RedditPost } from '../reddit/api.js';
import type { ConversationAutomationService } from './conversation-automation.js';
import type { RecipientAccountStatusService } from './recipient-account-status.js';
import type { AccountEstablishmentService } from './account-establishment.js';
import type { AccountGrowthService } from './account-growth.js';
import type { AccountBrowsingSimulationService } from './account-browsing-simulation.js';
import type { AccountSuspensionService } from './account-suspension.js';
import type { ProcessedPost, ProcessResult } from './processor.js';
import type { ConversationEventRepository } from '../database/conversation-event-repository.js';
import type { ConversationLearningRepository } from '../database/conversation-learning-repository.js';

/** Credentials passed to a factory. They must never be exposed by list/snapshot APIs. */
export interface VerifiedAccountSession {
  accountId: string;
  redditUsername: string;
  matrixUserId: string;
  cookies: Cookie[];
  matrixCredentials: MatrixCredentials;
  updatedAt: number;
}

/**
 * The complete set of stateful dependencies owned by one account runtime.
 *
 * Nothing in this object may be shared with another account runtime except the
 * underlying SQLite connection and stateless configuration/AI clients. In
 * particular, BrowserClient, AccountAuthContext, ChatCache, MatrixClient,
 * RedditApiClient and RateLimiter are intentionally per-account instances.
 */
export interface RedditAccountRuntimeResources {
  browserProfileDir: string;
  browserClient: BrowserClient;
  accountAuth: AccountAuthContext;
  matrixClient: MatrixClient;
  chatCache: ChatCache;
  redditApi: RedditApiClient;

  postsRepo: PostsRepository;
  messagesRepo: MessagesRepository;
  successfulMessagesRepo: SuccessfulMessagesRepository;
  deliveryOutboxRepo: DeliveryOutboxRepository;
  conversationStateRepo: ConversationStateRepository;
  conversationReadStateRepo: ConversationReadStateRepository;
  conversationOutcomesRepo: ConversationOutcomesRepository;
  conversationEventRepo: ConversationEventRepository;
  conversationLearningRepo: ConversationLearningRepository;
  redditUsersRepo: RedditUsersRepository;
  accountContentRepo: AccountContentRepository;
  recipientAccountStatusRepo: RedditRecipientAccountStatusRepository;
  activityLogRepo: ActivityLogRepository;

  monitorService: MonitorService;
  processorService: ProcessorService;
  responseChecker: ResponseCheckerService;
  conversationAutomation: ConversationAutomationService;
  chatService: ChatService;
  mediaService: MediaService;
  mediaSyncService: MediaSyncService;
  recipientAccountStatusService: RecipientAccountStatusService;
  establishmentService: AccountEstablishmentService;
  /** Present in concrete runtimes; optional for narrow lifecycle test fixtures. */
  suspensionService?: AccountSuspensionService;
  /**
   * Comment-based account growth. Optional because a runtime is fully
   * functional without it, and it is disabled per account by default.
   */
  growthService?: AccountGrowthService;
  /** Simulation-only browsing selection. It has no Reddit write dependency. */
  browsingService?: AccountBrowsingSimulationService;
  mediaExchange?: MediaLibraryExchangeService;
  rateLimiter: RateLimiter;
}

export interface RedditAccountRuntimeLifecycle {
  /** Start this account's main loop, response schedule and media sync. */
  start(): void | Promise<void>;
  /** Stop and await this account's schedules/loop, then close its browser. */
  stop(): void | Promise<void>;
  /** Immediately halt account work after a definitive suspension signal. */
  pause?(reason: string): void;
  /** Process a post assigned by the shared discovery scheduler. */
  dispatchPost?(post: RedditPost, signal?: AbortSignal): Promise<void>;
  preparePost?(post: RedditPost, signal?: AbortSignal): Promise<ProcessResult>;
  dispatchPreparedPost?(post: ProcessedPost, signal?: AbortSignal): Promise<PreparedDeliveryResult>;
}

export type PreparedDeliveryResult =
  | { status: 'sent'; conversationId: string }
  | { status: 'terminal'; reason: string }
  | { status: 'retryable'; reason: string; retryAt: number };

export type RedditAccountRuntimeState =
  | 'created'
  | 'starting'
  | 'running'
  | 'stopping'
  | 'paused'
  | 'stopped'
  | 'failed';

export interface RedditAccountRuntimeSnapshot {
  accountId: string;
  redditUsername: string;
  matrixUserId: string;
  state: RedditAccountRuntimeState;
  updatedAt: number;
  failure: string | null;
}

/**
 * One immutable Reddit identity and every mutable service associated with it.
 * There is deliberately no switchAccount method: a session refresh replaces
 * the matching runtime instead of mutating shared live state.
 */
export class RedditAccountRuntime {
  private state: RedditAccountRuntimeState = 'created';
  private failure: string | null = null;
  private transition: Promise<void> = Promise.resolve();

  constructor(
    readonly session: VerifiedAccountSession,
    readonly resources: RedditAccountRuntimeResources,
    private readonly lifecycle: RedditAccountRuntimeLifecycle,
  ) {
    if (session.accountId !== session.matrixUserId) {
      throw new Error('Runtime account ID must match the verified Matrix user ID');
    }
    if (!session.accountId || !session.redditUsername) {
      throw new Error('Runtime requires a verified Reddit and Matrix identity');
    }
  }

  start(): Promise<void> {
    return this.enqueue(async () => {
      if (this.state === 'running') return;
      this.state = 'starting';
      this.failure = null;
      try {
        await this.lifecycle.start();
        this.state = 'running';
      } catch (error) {
        this.state = 'failed';
        this.failure = errorMessage(error);
        try {
          await this.lifecycle.stop();
        } catch (cleanupError) {
          throw new AggregateError([error, cleanupError], 'Account runtime failed to start and clean up');
        }
        throw error;
      }
    });
  }

  stop(): Promise<void> {
    return this.enqueue(async () => {
      if (this.state === 'created' || this.state === 'stopped') {
        this.state = 'stopped';
        return;
      }
      this.state = 'stopping';
      try {
        await this.lifecycle.stop();
        this.state = 'stopped';
      } catch (error) {
        this.state = 'failed';
        this.failure = errorMessage(error);
        throw error;
      }
    });
  }

  /**
   * Suspension can be discovered by work that this runtime is currently
   * awaiting. Keep this synchronous so it cannot wait on its own loop.
   */
  pause(reason: string): void {
    if (this.state === 'stopped' || this.state === 'failed') return;
    this.failure = reason;
    this.state = 'paused';
    this.lifecycle.pause?.(reason);
  }

  snapshot(): RedditAccountRuntimeSnapshot {
    return {
      accountId: this.session.accountId,
      redditUsername: this.session.redditUsername,
      matrixUserId: this.session.matrixUserId,
      state: this.state,
      updatedAt: this.session.updatedAt,
      failure: this.failure,
    };
  }

  async canAcceptPost(): Promise<boolean> {
    if (this.state !== 'running') return false;
    if (!this.resources.accountAuth.isFullyAuthenticated()
      && !(await this.resources.accountAuth.ensureFullyAuthenticated())) return false;
    // An account Reddit has refused invitations from would burn the candidate,
    // so it takes no new posts until its re-test window opens.
    if (!this.resources.establishmentService.isSendable()) return false;
    if (await this.resources.conversationAutomation.hasActionableReplies()) return false;
    if (this.resources.chatService.getInvitationCooldown()) return false;
    return this.resources.rateLimiter.getTimers().nextMessageAvailableAt === null;
  }

  async canScan(): Promise<boolean> {
    if (this.state !== 'running') return false;
    return this.resources.accountAuth.isFullyAuthenticated()
      || await this.resources.accountAuth.ensureFullyAuthenticated();
  }

  preparePost(post: RedditPost, signal?: AbortSignal): Promise<ProcessResult> {
    if (this.state !== 'running' || !this.lifecycle.preparePost) {
      throw new Error(`Runtime ${this.session.redditUsername} cannot prepare candidates`);
    }
    return this.lifecycle.preparePost(post, signal);
  }

  dispatchPreparedPost(post: ProcessedPost, signal?: AbortSignal): Promise<PreparedDeliveryResult> {
    if (this.state !== 'running' || !this.lifecycle.dispatchPreparedPost) {
      throw new Error(`Runtime ${this.session.redditUsername} cannot dispatch prepared candidates`);
    }
    return this.lifecycle.dispatchPreparedPost(post, signal);
  }

  dispatchPost(post: RedditPost, signal?: AbortSignal): Promise<void> {
    if (this.state !== 'running') {
      throw new Error(`Cannot assign a post to ${this.session.redditUsername}: runtime is not running`);
    }
    if (!this.lifecycle.dispatchPost) {
      throw new Error(`Runtime ${this.session.redditUsername} does not support shared post dispatch`);
    }
    return this.lifecycle.dispatchPost(post, signal);
  }

  private enqueue(operation: () => Promise<void>): Promise<void> {
    const result = this.transition.then(operation, operation);
    this.transition = result.catch(() => undefined);
    return result;
  }
}

export type RedditAccountRuntimeBuilder = (
  session: VerifiedAccountSession,
) => Promise<RedditAccountRuntime>;

/** Narrow factory wrapper used as the bootstrap integration seam. */
export class RedditAccountRuntimeFactory {
  constructor(private readonly buildRuntime: RedditAccountRuntimeBuilder) {}

  async create(record: RedditAccountPrivateRecord): Promise<RedditAccountRuntime> {
    return this.buildRuntime(parseStoredSession(record));
  }

  async createSession(session: VerifiedAccountSession): Promise<RedditAccountRuntime> {
    return this.buildRuntime(session);
  }
}

export interface RedditAccountRuntimeManagerSnapshot {
  runtimes: RedditAccountRuntimeSnapshot[];
}

/**
 * Owns all enabled account runtimes. Work for different account IDs runs in
 * parallel; operations for the same ID are serialized to avoid duplicate
 * loops and browser-profile contention.
 */
export class RedditAccountRuntimeManager {
  private readonly runtimes = new Map<string, RedditAccountRuntime>();
  private readonly accountOperations = new Map<string, Promise<void>>();

  constructor(
    private readonly accounts: Pick<RedditAccountsRepository, 'listPublic' | 'get'>,
    private readonly factory: RedditAccountRuntimeFactory,
  ) {}

  async startEnabled(): Promise<RedditAccountRuntimeManagerSnapshot> {
    const enabled = this.accounts.listPublic().filter(isEnabledAccount);
    // A broken or suspended session belongs to one account. Do not prevent
    // other verified accounts from starting because of it.
    await Promise.allSettled(enabled.map((account) => this.start(account.accountId)));
    return this.snapshot();
  }

  start(accountId: string): Promise<void> {
    return this.forAccount(accountId, async () => {
      const existing = this.runtimes.get(accountId);
      if (existing?.snapshot().state === 'running') return;
      const account = this.requireEnabledAccount(accountId);
      const runtime = existing ?? await this.factory.create(account);
      await runtime.start();
      this.runtimes.set(accountId, runtime);
    });
  }

  /**
   * Rebuild only the refreshed account. Other account loops are untouched.
   * The old runtime is stopped before concrete construction because a builder
   * may open the account's persistent browser profile immediately.
   */
  replace(accountId: string): Promise<void> {
    return this.forAccount(accountId, async () => {
      const account = this.requireEnabledAccount(accountId);
      const previous = this.runtimes.get(accountId);
      if (previous) await previous.stop();
      try {
        // Concrete construction opens the account's persistent browser
        // profile, so the prior owner must be closed before factory.create.
        const replacement = await this.factory.create(account);
        await replacement.start();
        this.runtimes.set(accountId, replacement);
      } catch (error) {
        if (previous) {
          try {
            const restored = await this.factory.createSession(previous.session);
            await restored.start();
            this.runtimes.set(accountId, restored);
          } catch (rollbackError) {
            this.runtimes.delete(accountId);
            throw new AggregateError(
              [error, rollbackError],
              `Could not replace or restore Reddit account runtime ${accountId}`,
            );
          }
        } else {
          this.runtimes.delete(accountId);
        }
        throw error;
      }
    });
  }

  stop(accountId: string): Promise<void> {
    return this.forAccount(accountId, async () => {
      const runtime = this.runtimes.get(accountId);
      if (!runtime) return;
      await runtime.stop();
      this.runtimes.delete(accountId);
    });
  }

  async stopAll(): Promise<void> {
    await Promise.all([...this.runtimes.keys()].map((accountId) => this.stop(accountId)));
  }

  get(accountId: string): RedditAccountRuntime | null {
    return this.runtimes.get(accountId) ?? null;
  }

  list(): RedditAccountRuntimeSnapshot[] {
    return this.snapshot().runtimes;
  }

  snapshot(): RedditAccountRuntimeManagerSnapshot {
    return {
      runtimes: [...this.runtimes.values()]
        .map((runtime) => runtime.snapshot())
        .sort((left, right) => left.redditUsername.localeCompare(right.redditUsername)),
    };
  }

  private requireEnabledAccount(accountId: string): RedditAccountPrivateRecord {
    const account = this.accounts.get(accountId);
    if (!account || !isEnabledAccount(account)) {
      throw new Error(`Reddit account ${accountId} is not enabled with a verified session`);
    }
    return account;
  }

  private forAccount(accountId: string, operation: () => Promise<void>): Promise<void> {
    const previous = this.accountOperations.get(accountId) ?? Promise.resolve();
    const result = previous.then(operation, operation);
    const settled = result.catch(() => undefined).finally(() => {
      if (this.accountOperations.get(accountId) === settled) {
        this.accountOperations.delete(accountId);
      }
    });
    this.accountOperations.set(accountId, settled);
    return result;
  }
}

/**
 * Deterministic, filesystem-safe profile path for one immutable account ID.
 * Bootstrap should clone config and set browser.sessionDir to this value
 * before constructing that account's BrowserClient.
 */
export function accountBrowserProfileDir(baseSessionDir: string, accountId: string): string {
  if (!accountId) throw new Error('Browser profile requires an account ID');
  const encodedAccountId = Buffer.from(accountId, 'utf8').toString('base64url');
  return path.join(baseSessionDir, 'accounts', encodedAccountId);
}

function parseStoredSession(record: RedditAccountPrivateRecord): VerifiedAccountSession {
  const cookies = StoredCookiesSchema.parse(JSON.parse(record.cookiesJson)) as Cookie[];
  const matrixCredentials = StoredMatrixCredentialsSchema.parse(
    JSON.parse(record.matrixCredentialsJson),
  ) as MatrixCredentials;
  if (record.matrixUserId !== record.accountId || matrixCredentials.userId !== record.accountId) {
    throw new Error('Stored Reddit account and Matrix identities do not match');
  }
  return {
    accountId: record.accountId,
    redditUsername: record.redditUsername,
    matrixUserId: record.accountId,
    cookies,
    matrixCredentials,
    updatedAt: record.updatedAt,
  };
}

const StoredCookiesSchema = z.array(z.object({
  name: z.string().min(1),
  value: z.string(),
  domain: z.string().optional(),
  path: z.string().optional(),
  expires: z.number().optional(),
  httpOnly: z.boolean().optional(),
  secure: z.boolean().optional(),
  sameSite: z.enum(['Strict', 'Lax', 'None']).optional(),
  url: z.string().url().optional(),
}));

const StoredMatrixCredentialsSchema = z.object({
  homeserver: z.string().url(),
  accessToken: z.string().min(1),
  userId: z.string().min(1),
  capturedAt: z.number(),
});

function isEnabledAccount(account: {
  status: string;
  isActive: boolean;
}): boolean {
  return account.status === 'verified' && account.isActive;
}

function errorMessage(error: unknown): string {
  return error instanceof Error ? error.message : String(error);
}