RedditBot/src/services/conversation-automation.ts

import type { OpenRouterClient } from '../ai/openrouter.js';
import type { ConversationStateRepository, SentMessageRecord } from '../database/repositories.js';
import type { Logger } from '../utils/logger.js';
import type { RateLimiter } from '../utils/rate-limiter.js';
import { findSellerSignal } from '../reddit/api.js';
import type { ActiveChat, ChatMessage } from './chat/types.js';
import type { ChatService } from './chat/service.js';
import {
  buildConversationState,
  chooseConversationIntent,
  conversationMessageFingerprint,
  MAX_AUTOMATED_CONVERSATION_TURNS,
  mergeConversationState,
  remainingAutomationTurns,
  type ConversationState,
} from './conversation-state.js';

export type ConversationAutomationResultKind =
  | 'sent'
  | 'human-required'
  | 'boundary'
  | 'image-received';

export interface ConversationAutomationResult {
  conversationId: string;
  username: string;
  kind: ConversationAutomationResultKind;
  remainingTurns: number;
}

export interface ConversationOpenerHistory {
  getByChatRoomId(chatRoomId: string): SentMessageRecord | undefined;
}

export interface ConversationAutomationClaims {
  tryClaim(conversationId: string, incomingKey: string): boolean;
  markSent(conversationId: string, incomingKey: string): boolean;
}

/**
 * Sends at most one reply for each newly received user message. Durable state
 * makes retries and restarts idempotent, and every conversation stops after
 * three automated turns or immediately when consent is withdrawn.
 */
export class ConversationAutomationService {
  constructor(
    private readonly chat: ChatService,
    private readonly ai: OpenRouterClient,
    private readonly states: ConversationStateRepository,
    private readonly openers: ConversationOpenerHistory,
    private readonly claims: ConversationAutomationClaims,
    private readonly rateLimiter: RateLimiter,
    private readonly logger: Logger,
    private readonly dryRun = false,
    private readonly sellerKeywords: readonly string[] = [],
    private readonly sellerDomains: readonly string[] = [],
  ) {}

  async processUnreadConversations(signal?: AbortSignal): Promise<ConversationAutomationResult[]> {
    await this.chat.refreshTopChats(5);
    const conversations = await this.chat.getCachedConversations();
    const results: ConversationAutomationResult[] = [];
    for (const conversation of conversations) {
      const result = await this.processConversation(conversation, signal);
      if (result) results.push(result);
    }
    return results;
  }

  async processConversation(
    conversation: ActiveChat,
    signal?: AbortSignal,
  ): Promise<ConversationAutomationResult | null> {
    if (!conversation.unread || conversation.isFromUs || !conversation.roomId || !conversation.url) return null;
    const messages = conversation.messages ?? [];
    const latest = messages.at(-1);
    if (!latest || latest.isSelf) return null;

    const conversationId = conversation.roomId;
    const opener = this.openers.getByChatRoomId(conversationId);
    if (!opener || latest.timestamp <= opener.sentAt) {
      this.logger.debug?.(
        `[automation] Ignoring unverified inbound event for u/${conversation.username}: no confirmed post-opener response in this room`,
      );
      return null;
    }
    const mapped = messages.map(({ sender, text, isSelf }) => ({ sender, text, isSelf }));
    const state = mergeConversationState(
      this.states.get(conversationId),
      buildConversationState(mapped),
    );
    const incomingKey = messageKey(latest);
    if (state.automation.lastIncomingKey === incomingKey) return null;
    if (state.automation.status !== 'active') return null;

    const sellerSignal = findSellerSignal(latest.text, [...this.sellerKeywords], [...this.sellerDomains]);
    if (sellerSignal) {
      return this.finish(
        conversation,
        conversationId,
        state,
        incomingKey,
        'human-required',
        `Potential upsell detected (${sellerSignal.kind}: ${sellerSignal.value}).`,
      );
    }
    if (hasReceivedImage(messages)) {
      return this.finish(conversation, conversationId, state, incomingKey, 'image-received', 'The user shared an image.');
    }
    if (state.boundaries.length > 0) {
      return this.finish(conversation, conversationId, state, incomingKey, 'boundary', state.boundaries.join('; '));
    }
    if (remainingAutomationTurns(state) === 0) {
      return this.finish(conversation, conversationId, state, incomingKey, 'human-required', 'Three automated turns have been used.');
    }

    const intent = chooseConversationIntent(mapped, state, 'media-ask');
    if (intent.kind !== 'media-ask') {
      return this.finish(conversation, conversationId, state, incomingKey, 'human-required', intent.reason);
    }
    const reply = await this.ai.generateMediaAsk(mapped, conversation.username, state, intent);
    if (this.dryRun) {
      this.logger.info(`[automation] Dry run prepared turn ${state.automation.turnsUsed + 1} for u/${conversation.username}`);
      return null;
    }

    if (!this.claims.tryClaim(conversationId, incomingKey)) return null;

    // Persist the full intended state before the remote send. If the process
    // dies after Matrix accepts the message, both this state and the durable
    // event claim make the retry fail closed instead of sending twice.
    state.askedForMedia = true;
    state.automation.turnsUsed += 1;
    state.automation.lastIncomingKey = incomingKey;
    state.automation.aiMessageFingerprints.push(conversationMessageFingerprint(reply));
    const exhausted = state.automation.turnsUsed >= MAX_AUTOMATED_CONVERSATION_TURNS;
    if (exhausted) {
      state.automation.status = 'human-required';
      state.automation.handoffReason = 'Three automated turns have been used.';
    }
    this.states.save(conversationId, state);

    const sent = await this.rateLimiter.throttle(
      () => this.chat.sendChatMessage(conversation.url!, reply),
      (success) => success,
      signal,
    );
    if (!sent) return null;
    this.claims.markSent(conversationId, incomingKey);
    this.logger.info(
      `[automation] Sent turn ${state.automation.turnsUsed}/${MAX_AUTOMATED_CONVERSATION_TURNS} to u/${conversation.username}`,
    );
    return {
      conversationId,
      username: conversation.username,
      kind: exhausted ? 'human-required' : 'sent',
      remainingTurns: remainingAutomationTurns(state),
    };
  }

  private finish(
    conversation: ActiveChat,
    conversationId: string,
    state: ConversationState,
    incomingKey: string,
    status: 'image-received' | 'human-required' | 'boundary',
    reason: string,
  ): ConversationAutomationResult {
    state.automation.status = status;
    state.automation.lastIncomingKey = incomingKey;
    state.automation.handoffReason = reason;
    this.states.save(conversationId, state);
    return {
      conversationId,
      username: conversation.username,
      kind: status,
      remainingTurns: remainingAutomationTurns(state),
    };
  }
}

function hasReceivedImage(messages: ChatMessage[]): boolean {
  return messages.some((message) => !message.isSelf && (message.media?.length ?? 0) > 0);
}

function messageKey(message: ChatMessage): string {
  return message.eventId ?? `${message.timestamp}:${message.sender}:${message.text}`;
}