feat: expand outreach targeting and messaging

src/services/reddit-user-messenger.ts

@@ -0,0 +1,140 @@
+import type { RedditAccountRuntime, RedditAccountRuntimeManager } from './reddit-account-runtime.js';
+import type { ActiveChat } from './chat/types.js';
+import { buildConversationState, mergeConversationState } from './conversation-state.js';
+
+export type RedditUserMessageResult = {
+  accountId: string;
+  redditUsername: string;
+  recipient: string;
+  conversationId: string;
+  reusedConversation: boolean;
+};
+
+type RuntimeConversation = { runtime: RedditAccountRuntime; conversation: ActiveChat };
+
+/** Selects an account and performs one explicit, rate-limited operator/API send. */
+export class RedditUserMessenger {
+  constructor(private readonly runtimes: RedditAccountRuntimeManager) {}
+
+  async send(username: string, message: string): Promise<RedditUserMessageResult> {
+    const recipient = username.trim().replace(/^u\//i, '');
+    const running = this.runningRuntimes();
+    if (running.length === 0) throw new RedditUserMessageError(503, 'No Reddit account is running');
+
+    const existing = await this.findExistingConversation(running, recipient);
+    if (existing) return this.sendExisting(existing, recipient, message);
+
+    const runtime = running
+      .filter((item) => item.resources.accountAuth.isFullyAuthenticated())
+      .filter((item) => !item.resources.chatService.getInvitationCooldown())
+      .sort(compareAvailableRuntimes)[0];
+    if (!runtime) throw new RedditUserMessageError(503, 'No authenticated Reddit account is available to start a conversation');
+
+    const started = await runtime.resources.rateLimiter.throttle(
+      () => runtime.resources.chatService.startConversationDetailed(recipient, message),
+      (result) => result.status === 'sent',
+    );
+    if (started.status !== 'sent') {
+      throw new RedditUserMessageError(started.status === 'terminal' ? 422 : 503, started.reason);
+    }
+    const conversationId = started.chat.roomId;
+    if (!conversationId) throw new RedditUserMessageError(502, 'Reddit confirmed the message without returning a conversation ID');
+    this.recordSend(runtime, recipient, `Started chat with u/${recipient}`);
+    runtime.resources.conversationStateRepo.save(
+      conversationId,
+      buildConversationState(started.chat.messages ?? [{ sender: 'Me', text: message, isSelf: true }]),
+    );
+    return resultFor(runtime, recipient, conversationId, false);
+  }
+
+  private runningRuntimes(): RedditAccountRuntime[] {
+    return this.runtimes.list()
+      .filter((snapshot) => snapshot.state === 'running')
+      .flatMap((snapshot) => {
+        const runtime = this.runtimes.get(snapshot.accountId);
+        return runtime ? [runtime] : [];
+      });
+  }
+
+  private async findExistingConversation(
+    runtimes: RedditAccountRuntime[],
+    username: string,
+  ): Promise<RuntimeConversation | null> {
+    const normalized = normalizeUsername(username);
+    const matches = (await Promise.all(runtimes.map(async (runtime) =>
+      (await runtime.resources.chatService.getCachedConversations())
+        .filter((conversation) => normalizeUsername(conversation.username) === normalized)
+        .map((conversation) => ({ runtime, conversation })),
+    ))).flat();
+    matches.sort((left, right) => right.conversation.lastMessageTime - left.conversation.lastMessageTime
+      || left.runtime.session.redditUsername.localeCompare(right.runtime.session.redditUsername));
+    return matches[0] ?? null;
+  }
+
+  private async sendExisting(
+    selected: RuntimeConversation,
+    username: string,
+    message: string,
+  ): Promise<RedditUserMessageResult> {
+    const { runtime, conversation } = selected;
+    if (!conversation.url || !conversation.roomId) {
+      throw new RedditUserMessageError(409, 'The existing conversation is missing its Matrix room details; synchronize chats and retry');
+    }
+    const sent = await runtime.resources.rateLimiter.throttle(
+      () => runtime.resources.chatService.sendChatMessage(conversation.url!, message),
+      (success) => success,
+    );
+    if (!sent) throw new RedditUserMessageError(502, 'Reddit did not confirm the message send');
+    const currentMessages = (conversation.messages ?? []).map(({ sender, text, isSelf }) => ({ sender, text, isSelf }));
+    const state = mergeConversationState(
+      runtime.resources.conversationStateRepo.get(conversation.roomId),
+      buildConversationState([...currentMessages, { sender: 'Me', text: message, isSelf: true }]),
+    );
+    state.automation.status = 'user-overridden';
+    state.automation.handoffReason = 'An API message was sent in this conversation.';
+    runtime.resources.conversationStateRepo.save(conversation.roomId, state);
+    this.recordSend(runtime, username, `Sent chat message to u/${username}`);
+    void runtime.resources.chatService.refreshTopChats(5).catch(() => undefined);
+    return resultFor(runtime, username, conversation.roomId, true);
+  }
+
+  private recordSend(runtime: RedditAccountRuntime, username: string, description: string): void {
+    runtime.resources.activityLogRepo.insert({
+      timestamp: Date.now(),
+      type: 'sent',
+      message: description,
+      recipient: username,
+    });
+  }
+}
+
+export class RedditUserMessageError extends Error {
+  constructor(readonly status: number, message: string) {
+    super(message);
+  }
+}
+
+function compareAvailableRuntimes(left: RedditAccountRuntime, right: RedditAccountRuntime): number {
+  return left.resources.rateLimiter.getStats().messagesThisHour
+    - right.resources.rateLimiter.getStats().messagesThisHour
+    || left.session.redditUsername.localeCompare(right.session.redditUsername);
+}
+
+function normalizeUsername(username: string): string {
+  return username.trim().replace(/^u\//i, '').toLocaleLowerCase('en-US');
+}
+
+function resultFor(
+  runtime: RedditAccountRuntime,
+  recipient: string,
+  conversationId: string,
+  reusedConversation: boolean,
+): RedditUserMessageResult {
+  return {
+    accountId: runtime.session.accountId,
+    redditUsername: runtime.session.redditUsername,
+    recipient: recipient.trim().replace(/^u\//i, ''),
+    conversationId,
+    reusedConversation,
+  };
+}