import { Router } from 'express'; import type { ActiveChat } from '../../services/chat.js'; import type { WebContext } from '../context.js'; import { buildConversationState, chooseConversationIntent, conversationMessageFingerprint, mergeConversationState, remainingAutomationTurns, } from '../../services/conversation-state.js'; type OwnedActiveChat = ActiveChat & { owner?: { accountId: string; redditUsername: string } }; export function createConversationsRouter(ctx: WebContext): Router { const router = Router(); function currentOwner() { const snapshot = ctx.services.accountAuth?.getSnapshot(); return snapshot?.accountId && snapshot.redditUsername ? { accountId: snapshot.accountId, redditUsername: snapshot.redditUsername } : undefined; } function parseOwnedId(id: string): { accountId: string; conversationId: string } | null { const separator = id.indexOf('~'); if (separator < 1) return null; try { return { accountId: decodeURIComponent(id.slice(0, separator)), conversationId: decodeURIComponent(id.slice(separator + 1)), }; } catch { return null; } } async function findConversation(id: string): Promise<{ conversation: ActiveChat; runtime: NonNullable['get']>> | null; } | undefined> { const owned = parseOwnedId(id); if (!owned && (ctx.services.accountRuntimes?.list().length ?? 0) > 0) return undefined; const runtime = owned ? ctx.services.accountRuntimes?.get(owned.accountId) ?? null : null; if (owned && (!runtime || runtime.snapshot().state !== 'running')) return undefined; const chat = runtime?.resources.chatService ?? ctx.services.chat; if (!chat) return undefined; const conversations = await chat.getCachedConversations(); const conversationId = owned?.conversationId ?? id; const conversation = conversations.find( (c) => c.roomId === conversationId || c.username === conversationId || c.roomName === conversationId, ); return conversation ? { conversation, runtime } : undefined; } function getConversationIntelligence( conversation: ActiveChat, stateRepo = ctx.services.conversationState, requestedIntent?: 'media-ask', ) { const messages = (conversation.messages ?? []).map((message) => ({ sender: message.sender, text: message.text, isSelf: message.isSelf, })); const saved = conversation.roomId ? stateRepo?.get(conversation.roomId) : undefined; const state = mergeConversationState(saved, buildConversationState(messages)); if (conversation.roomId) stateRepo?.save(conversation.roomId, state); return { messages, state, intent: chooseConversationIntent(messages, state, requestedIntent) }; } router.get('/conversations', async (req, res) => { if (!ctx.services.chat) { res.status(503).json({ error: 'Chat service not initialized' }); return; } try { const runtimes = ctx.services.accountRuntimes?.list().filter((runtime) => runtime.state === 'running') ?? []; const conversations: OwnedActiveChat[] = runtimes.length > 0 ? (await Promise.all(runtimes.map(async (snapshot) => { const runtime = ctx.services.accountRuntimes!.get(snapshot.accountId)!; return (await runtime.resources.chatService.getCachedConversations()).map((conversation) => ({ ...conversation, owner: { accountId: snapshot.accountId, redditUsername: snapshot.redditUsername }, })); }))).flat() : await ctx.services.chat.getCachedConversations(); if (req.query.includeImageMetadata !== 'true') { res.json(conversations.map((conversation) => ({ ...conversation, owner: conversation.owner ?? currentOwner() }))); return; } const receivedMediaRooms = new Set(); const receivedMediaUsers = new Set(); if (ctx.services.media) { try { const mediaByAccount = runtimes.length > 0 ? await Promise.all(runtimes.map(async (snapshot) => ({ accountId: snapshot.accountId, items: await ctx.services.accountRuntimes!.get(snapshot.accountId)!.resources.mediaService.getMediaItems(), }))) : [{ accountId: currentOwner()?.accountId ?? 'legacy', items: await ctx.services.media.getMediaItems() }]; for (const source of mediaByAccount) { for (const item of source.items) { if (item.sender.toLowerCase() === 'me') continue; if (item.chatRoomId) receivedMediaRooms.add(`${source.accountId}:${item.chatRoomId}`); if (item.chatUsername) receivedMediaUsers.add(`${source.accountId}:${item.chatUsername.toLowerCase()}`); } } } catch (error) { ctx.logger.warn('Failed to read media index for conversation markers:', error); } } res.json(conversations.map((conversation) => ({ ...conversation, owner: conversation.owner ?? currentOwner(), hasReceivedImages: receivedMediaRooms.has(`${conversation.owner?.accountId ?? currentOwner()?.accountId ?? 'legacy'}:${conversation.roomId ?? ''}`) || receivedMediaUsers.has(`${conversation.owner?.accountId ?? currentOwner()?.accountId ?? 'legacy'}:${conversation.username.toLowerCase()}`) || (conversation.messages ?? []).some( (message) => !message.isSelf && (message.media?.length ?? 0) > 0, ), }))); } catch (error) { ctx.logger.error('Failed to fetch conversations:', error); res.status(500).json({ error: 'Failed to fetch conversations' }); } }); router.get('/conversations/:id', async (req, res) => { if (!ctx.services.chat) { res.status(404).json({ error: 'Chat service not initialized' }); return; } try { // Conversation reads never initiate a sync. Sync lifecycle events can // invalidate this query, so starting a sync here would create a loop. const resolved = await findConversation(decodeURIComponent(req.params.id)); if (!resolved) { res.status(404).json({ error: 'Conversation not found' }); return; } const chatService = resolved.runtime?.resources.chatService ?? ctx.services.chat; const syncStatus = chatService.getSyncStatus(); const syncInFlight = syncStatus.syncing; const conversation = resolved.conversation; // Delta sync only populates room summaries — full message history is // fetched on demand the first time a conversation is opened, and // re-fetched when the summary's last-activity timestamp is newer than // anything we have cached (i.e. they've replied since we last loaded). const cachedLastTs = conversation.messages?.length ? conversation.messages[conversation.messages.length - 1].timestamp : 0; const hasNewerActivity = (conversation.lastMessageTime ?? 0) > cachedLastTs; const needsLazyLoad = !conversation.messages || conversation.messages.length === 0 || hasNewerActivity; if (needsLazyLoad && !syncInFlight) { try { await chatService.loadConversationDetails(conversation); await chatService.saveCacheExternal(); } catch (err) { ctx.logger.warn(`Lazy-load failed for conversation ${conversation.roomId}:`, err); } } // Attach the post that triggered this outreach so the modal can link to it. let triggeringPost: { id: string; subreddit?: string; title?: string; url: string; imageUrl: string | null; postedAt?: number; sentAt: number; } | null = null; let automatedOpener: string | null = null; try { const messagesRepo = resolved.runtime?.resources.messagesRepo ?? ctx.repos.messages; const sent = conversation.roomId ? messagesRepo.getByChatRoomId(conversation.roomId) ?? messagesRepo.getLatestByRecipient(conversation.username) : messagesRepo.getLatestByRecipient(conversation.username); automatedOpener = sent?.message ?? null; if (sent && sent.postId) { const url = sent.postUrl ?? (sent.subreddit ? `https://www.reddit.com/r/${sent.subreddit}/comments/${sent.postId}/` : `https://redd.it/${sent.postId}`); triggeringPost = { id: sent.postId, subreddit: sent.subreddit, title: sent.postTitle, url, imageUrl: sent.postImageUrl ?? null, ...(sent.postCreatedAt ? { postedAt: sent.postCreatedAt } : {}), sentAt: sent.sentAt, }; } } catch (err) { ctx.logger.warn(`Failed to look up triggering post for ${conversation.username}:`, err); } const stateRepo = resolved.runtime?.resources.conversationStateRepo ?? ctx.services.conversationState; const intelligence = getConversationIntelligence(conversation, stateRepo); const receivedImage = (conversation.messages ?? []).some( (message) => !message.isSelf && (message.media?.length ?? 0) > 0, ); if (intelligence.state.automation.status === 'active' && receivedImage) { intelligence.state.automation.status = 'image-received'; intelligence.state.automation.handoffReason = 'The user shared an image.'; } else if (intelligence.state.automation.status === 'active' && intelligence.state.boundaries.length > 0) { intelligence.state.automation.status = 'boundary'; intelligence.state.automation.handoffReason = intelligence.state.boundaries.join('; '); } if (conversation.roomId) stateRepo?.save(conversation.roomId, intelligence.state); const aiFingerprints = new Map(); if (automatedOpener) { aiFingerprints.set(conversationMessageFingerprint(automatedOpener), 1); } for (const fingerprint of intelligence.state.automation.aiMessageFingerprints) { aiFingerprints.set(fingerprint, (aiFingerprints.get(fingerprint) ?? 0) + 1); } const annotatedMessages = conversation.messages?.map((message) => { if (!message.isSelf) return message; const fingerprint = conversationMessageFingerprint(message.text); const remainingMatches = aiFingerprints.get(fingerprint) ?? 0; if (remainingMatches === 0) return message; aiFingerprints.set(fingerprint, remainingMatches - 1); return { ...message, isAi: true }; }); res.json({ ...conversation, ...(annotatedMessages ? { messages: annotatedMessages } : {}), owner: resolved.runtime ? { accountId: resolved.runtime.session.accountId, redditUsername: resolved.runtime.session.redditUsername } : currentOwner(), triggeringPost, automation: { turnsUsed: intelligence.state.automation.turnsUsed, remainingTurns: remainingAutomationTurns(intelligence.state), status: intelligence.state.automation.status, handoffReason: intelligence.state.automation.handoffReason, }, syncInProgress: syncInFlight ? { kind: syncStatus.syncingKind, progress: syncStatus.progress, staleMessages: needsLazyLoad, } : null, }); } catch (error) { ctx.logger.error('Failed to fetch conversation:', error); res.status(500).json({ error: String(error) }); } }); router.post('/conversations/:id/suggest-reply', async (req, res) => { if (!ctx.services.chat) { res.status(500).json({ error: 'Chat service not initialized' }); return; } if (!ctx.services.ai) { res.status(500).json({ error: 'AI client not initialized' }); return; } try { const resolved = await findConversation(decodeURIComponent(req.params.id)); if (!resolved) { res.status(404).json({ error: 'Conversation not found' }); return; } const conversation = resolved.conversation; if (!conversation.messages || conversation.messages.length === 0) { res.status(400).json({ error: 'No messages in conversation to generate a reply from' }); return; } if (!conversation.messages.some((m) => !m.isSelf)) { res.status(400).json({ error: 'No reply from user yet - nothing to respond to' }); return; } const intelligence = getConversationIntelligence(conversation, resolved.runtime?.resources.conversationStateRepo); const suggestedReply = await ctx.services.ai.generateReply( intelligence.messages, conversation.username, intelligence.state, intelligence.intent, ); res.json({ suggestedReply, intent: intelligence.intent, conversationState: intelligence.state }); } catch (error) { ctx.logger.error('Failed to generate suggested reply:', error); res.status(500).json({ error: String(error) }); } }); router.post('/conversations/:id/suggest-media-ask', async (req, res) => { if (!ctx.services.chat) { res.status(500).json({ error: 'Chat service not initialized' }); return; } if (!ctx.services.ai) { res.status(500).json({ error: 'AI client not initialized' }); return; } try { const resolved = await findConversation(decodeURIComponent(req.params.id)); if (!resolved) { res.status(404).json({ error: 'Conversation not found' }); return; } const conversation = resolved.conversation; if (!conversation.messages || conversation.messages.length === 0) { res.status(400).json({ error: 'No messages in conversation to generate from' }); return; } if (!conversation.messages.some((m) => !m.isSelf)) { res.status(400).json({ error: 'No reply from user yet - nothing to respond to' }); return; } const intelligence = getConversationIntelligence(conversation, resolved.runtime?.resources.conversationStateRepo, 'media-ask'); if (intelligence.intent.kind !== 'media-ask') { res.status(409).json({ error: intelligence.intent.reason, intent: intelligence.intent, conversationState: intelligence.state }); return; } const suggestedReply = await ctx.services.ai.generateMediaAsk( intelligence.messages, conversation.username, intelligence.state, intelligence.intent, ); res.json({ suggestedReply, intent: intelligence.intent, conversationState: intelligence.state }); } catch (error) { ctx.logger.error('Failed to generate media-ask message:', error); res.status(500).json({ error: String(error) }); } }); router.post('/conversations/:id/send-reply', async (req, res) => { if (!ctx.services.chat) { res.status(500).json({ error: 'Chat service not initialized' }); return; } try { const { message } = req.body; if (!message || typeof message !== 'string' || message.trim().length === 0) { res.status(400).json({ error: 'Message is required' }); return; } const resolved = await findConversation(decodeURIComponent(req.params.id)); if (!resolved) { res.status(404).json({ error: 'Conversation not found' }); return; } const conversation = resolved.conversation; const chatService = resolved.runtime?.resources.chatService ?? ctx.services.chat; const stateRepo = resolved.runtime?.resources.conversationStateRepo ?? ctx.services.conversationState; if (!conversation.url) { res.status(400).json({ error: 'No chat URL for this conversation' }); return; } const success = await chatService.sendChatMessage(conversation.url, message.trim()); if (success) { if (conversation.roomId) { const currentMessages = (conversation.messages ?? []).map((item) => ({ sender: item.sender, text: item.text, isSelf: item.isSelf, })); const saved = stateRepo?.get(conversation.roomId); const updatedState = mergeConversationState( saved, buildConversationState([...currentMessages, { sender: 'Me', text: message.trim(), isSelf: true }]), ); updatedState.automation.status = 'user-overridden'; updatedState.automation.handoffReason = 'A manual reply was sent in this conversation.'; stateRepo?.save(conversation.roomId, updatedState); } if (resolved.runtime) { resolved.runtime.resources.activityLogRepo.insert({ timestamp: Date.now(), type: 'sent', message: `Sent chat reply to u/${conversation.username}`, recipient: conversation.username, }); } else { ctx.activity.push('sent', `Sent chat reply to u/${conversation.username}`, conversation.username); } ctx.broadcaster.push('info', `Chat reply sent to u/${conversation.username}`); try { // Wait for Matrix /sync to merge the sent event into the cached // message history. The dashboard invalidates its queries when this // request completes, so returning earlier can immediately re-read // the stale pre-send conversation. await chatService.refreshTopChats(5); } catch (err) { // Delivery is already confirmed by Matrix's send response. A sync // failure must not turn a successful send into a reported failure. ctx.logger.error('Failed to refresh chats after send-reply:', err); } res.json({ success: true }); } else { res.status(500).json({ error: 'Failed to send chat message' }); } } catch (error) { ctx.logger.error('Failed to send chat reply:', error); res.status(500).json({ error: String(error) }); } }); /** * Start a brand-new DM with a Reddit user and send the opening message. * Body: { username: string, message: string } */ router.post('/conversations/start', async (req, res) => { if (!ctx.services.chat) { res.status(500).json({ error: 'Chat service not initialized' }); return; } try { const { accountId, username, message } = req.body ?? {}; if (!accountId || typeof accountId !== 'string') { res.status(400).json({ error: 'accountId is required' }); return; } if (!username || typeof username !== 'string') { res.status(400).json({ error: 'username is required' }); return; } if (!message || typeof message !== 'string' || message.trim().length === 0) { res.status(400).json({ error: 'message is required' }); return; } const runtime = ctx.services.accountRuntimes?.get(accountId); if (!runtime || runtime.snapshot().state !== 'running') { res.status(409).json({ error: 'Selected Reddit account is not running' }); return; } const chat = await runtime.resources.chatService.startConversation(username, message.trim()); if (!chat) { res.status(500).json({ error: 'Failed to start conversation (unknown user or send failed)' }); return; } runtime.resources.activityLogRepo.insert({ timestamp: Date.now(), type: 'sent', message: `Started chat with u/${chat.username}`, recipient: chat.username, }); if (chat.roomId) { runtime.resources.conversationStateRepo.save( chat.roomId, buildConversationState(chat.messages ?? [{ sender: 'Me', text: message.trim(), isSelf: true }]), ); } ctx.broadcaster.push('info', `Started new chat with u/${chat.username}`); res.json({ success: true, chat: { ...chat, owner: { accountId, redditUsername: runtime.session.redditUsername } } }); } catch (error) { ctx.logger.error('Failed to start conversation:', error); res.status(500).json({ error: String(error) }); } }); return router; }