import { Router } from 'express'; import { z } from 'zod'; import type { ActiveChat } from '../../services/chat.js'; import type { WebContext } from '../context.js'; import { buildConversationState, chooseConversationIntent, conversationMessageFingerprint, mergeConversationState, remainingAutomationTurns, type ConversationAutomationStatus, type ConversationPolicyMode, } from '../../services/conversation-state.js'; import { sweepConversationsForSellers, type SellerSweepConversation, type SellerSweepOptions, type SellerSweepSummary, } from '../../services/seller-sweep.js'; import type { MediaLibraryAsset } from '../../services/media-library-types.js'; import type { ConversationReadStateRecord, ConversationReadStateRepository, } from '../../database/repositories.js'; import type { RedditRecipientAccountStatusRecord } from '../../database/account-repositories.js'; const ShareConversationMediaRequestSchema = z.strictObject({ assetId: z.string().uuid() }); const MarkReadRequestSchema = z.strictObject({ throughTimestamp: z.number().int().nonnegative().optional(), }); const MarkUnreadRequestSchema = z.strictObject({ beforeTimestamp: z.number().int().nonnegative().optional(), }); const SetStatusRequestSchema = z.strictObject({ status: z.enum(['open', 'snoozed', 'done']), snoozeUntil: z.number().int().positive().nullable().optional(), }); const BulkMarkReadRequestSchema = z.strictObject({ conversationIds: z.array(z.string().min(1)).min(1).max(500), }); type OwnedActiveChat = ActiveChat & { owner?: { accountId: string; redditUsername: string }; automationStatus?: ConversationAutomationStatus; automationPolicyMode?: ConversationPolicyMode; automationHandoffReason?: string | null; }; /** An owned conversation with operator triage state projected onto it. */ type TriagedActiveChat = OwnedActiveChat & { hasUnreadInbound: boolean; needsReply: boolean; triageStatus: 'open' | 'snoozed' | 'done'; snoozedUntil: number | null; lastReadMessageTs: number; recipientAccountStatus?: 'active' | 'suspended' | 'deleted'; recipientAccountStatusCheckedAt?: number; }; /** * Projects operator read state onto a conversation summary. * * `hasUnreadInbound` is deliberately derived from the summary's last-message * fields rather than the full message list — the list endpoint does not always * have message history loaded, and a conversation whose newest message is ours * is by definition one we have seen. */ function annotateTriage( conversation: OwnedActiveChat, readState: ConversationReadStateRecord | undefined, recipientStatus?: RedditRecipientAccountStatusRecord | null, ): TriagedActiveChat { const lastReadMessageTs = readState?.lastReadMessageTs ?? 0; const status = readState?.status ?? 'open'; return { ...conversation, hasUnreadInbound: !conversation.isFromUs && conversation.lastMessageTime > lastReadMessageTs, needsReply: recipientStatus?.status !== 'deleted' && !conversation.isFromUs && status === 'open', triageStatus: status, snoozedUntil: readState?.snoozedUntil ?? null, lastReadMessageTs, ...(recipientStatus ? { recipientAccountStatus: recipientStatus.status, recipientAccountStatusCheckedAt: recipientStatus.checkedAt, } : {}), }; } /** * Reopens conversations whose snooze has elapsed and returns the current read * state for the account, keyed by conversation id. */ function currentReadState( repo: ConversationReadStateRepository | undefined, ): Map { if (!repo) return new Map(); repo.expireSnoozes(); return new Map(repo.list().map((record) => [record.conversationId, record])); } 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) }; } /** * Projects the durable conversation state onto a list row: the seller flag and * the reply-loop status. The status is what tells the inbox which * conversations the operator is now driving by hand rather than the bot. */ function withStateMarkers( conversation: OwnedActiveChat, stateRepo = ctx.services.conversationState, ): OwnedActiveChat { if (!conversation.roomId) return conversation; const state = stateRepo?.get(conversation.roomId); if (!state) return conversation; return { ...conversation, ...(state.sellerDetected === true ? { sellerDetected: true } : {}), automationStatus: state.automation.status, automationPolicyMode: state.automation.policyMode, automationHandoffReason: state.automation.handoffReason, }; } /** * A non-blocking heads-up shown next to the manual share panel when the * conversation carries safety flags. The operator can still share — this only * makes the flags visible so the decision is informed. */ function manualShareWarning( conversation: ActiveChat, runtime: NonNullable['get']>>, ): string | null { if (!conversation.roomId) return null; const messages = conversation.messages ?? []; const state = mergeConversationState( runtime.resources.conversationStateRepo.get(conversation.roomId), buildConversationState(messages.map(({ sender, text, isSelf }) => ({ sender, text, isSelf }))), ); const notes: string[] = []; if (state.sellerDetected) notes.push('this conversation is flagged as a seller'); if (state.boundaries.length > 0) notes.push(`a boundary was detected (${state.boundaries.join('; ')})`); return notes.length > 0 ? `Heads up: ${notes.join(', and ')}.` : null; } function toMediaLibraryResponse(asset: MediaLibraryAsset) { const { relativePath, ...publicAsset } = asset; return { ...publicAsset, urlPath: `/library-media/${relativePath.split(/[\\/]/).map(encodeURIComponent).join('/')}`, }; } router.get('/conversations', async (req, res) => { const chatService = ctx.services.chat; if (!chatService) { res.status(503).json({ error: 'Chat service not initialized' }); return; } try { const runtimes = ctx.services.accountRuntimes?.list().filter((runtime) => runtime.state === 'running') ?? []; const conversations: TriagedActiveChat[] = runtimes.length > 0 ? (await Promise.all(runtimes.map(async (snapshot) => { const runtime = ctx.services.accountRuntimes!.get(snapshot.accountId)!; const readState = currentReadState(runtime.resources.conversationReadStateRepo); return (await runtime.resources.chatService.getCachedConversations()).map((conversation) => annotateTriage( withStateMarkers({ ...conversation, owner: { accountId: snapshot.accountId, redditUsername: snapshot.redditUsername }, }, runtime.resources.conversationStateRepo), conversation.roomId ? readState.get(conversation.roomId) : undefined, runtime.resources.recipientAccountStatusRepo?.get(conversation.username), )); }))).flat() : await (async () => { const readState = currentReadState(ctx.services.conversationReadState); return (await chatService.getCachedConversations()).map((conversation) => annotateTriage( withStateMarkers(conversation), conversation.roomId ? readState.get(conversation.roomId) : undefined, )); })(); 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 }; }); const readStateRepo = resolved.runtime?.resources.conversationReadStateRepo ?? ctx.services.conversationReadState; readStateRepo?.expireSnoozes(); const triage = annotateTriage( conversation, conversation.roomId ? readStateRepo?.get(conversation.roomId) : undefined, resolved.runtime?.resources.recipientAccountStatusRepo?.get(conversation.username), ); res.json({ ...conversation, hasUnreadInbound: triage.hasUnreadInbound, needsReply: triage.needsReply, triageStatus: triage.triageStatus, snoozedUntil: triage.snoozedUntil, lastReadMessageTs: triage.lastReadMessageTs, recipientAccountStatus: triage.recipientAccountStatus, recipientAccountStatusCheckedAt: triage.recipientAccountStatusCheckedAt, sellerDetected: intelligence.state.sellerDetected, ...(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) }); } }); /** * Resolves the read-state repository that owns a conversation, plus the room * id to key it by. Conversations without a Matrix room cannot carry read * state — there is no stable identifier to hang it off. */ async function resolveReadState(id: string): Promise< | { repo: ConversationReadStateRepository; roomId: string; conversation: ActiveChat } | { error: { status: number; message: string } } > { const resolved = await findConversation(decodeURIComponent(id)); if (!resolved) return { error: { status: 404, message: 'Conversation not found' } }; const repo = resolved.runtime?.resources.conversationReadStateRepo ?? ctx.services.conversationReadState; if (!repo) return { error: { status: 503, message: 'Conversation read state is not initialized' } }; if (!resolved.conversation.roomId) { return { error: { status: 409, message: 'This conversation has no Matrix room.' } }; } return { repo, roomId: resolved.conversation.roomId, conversation: resolved.conversation }; } router.post('/conversations/:id/read', async (req, res) => { const parsed = MarkReadRequestSchema.safeParse(req.body ?? {}); if (!parsed.success) { res.status(400).json({ error: 'throughTimestamp must be a non-negative integer' }); return; } const resolved = await resolveReadState(req.params.id); if ('error' in resolved) { res.status(resolved.error.status).json({ error: resolved.error.message }); return; } const through = parsed.data.throughTimestamp ?? resolved.conversation.lastMessageTime; resolved.repo.markRead(resolved.roomId, through); res.json({ success: true }); }); router.post('/conversations/:id/unread', async (req, res) => { const parsed = MarkUnreadRequestSchema.safeParse(req.body ?? {}); if (!parsed.success) { res.status(400).json({ error: 'beforeTimestamp must be a non-negative integer' }); return; } const resolved = await resolveReadState(req.params.id); if ('error' in resolved) { res.status(resolved.error.status).json({ error: resolved.error.message }); return; } const before = parsed.data.beforeTimestamp ?? resolved.conversation.lastMessageTime; resolved.repo.markUnread(resolved.roomId, before); res.json({ success: true }); }); router.post('/conversations/:id/status', async (req, res) => { const parsed = SetStatusRequestSchema.safeParse(req.body ?? {}); if (!parsed.success) { res.status(400).json({ error: 'status must be one of open, snoozed, done' }); return; } if (parsed.data.status === 'snoozed' && !parsed.data.snoozeUntil) { res.status(400).json({ error: 'snoozeUntil is required when snoozing a conversation' }); return; } const resolved = await resolveReadState(req.params.id); if ('error' in resolved) { res.status(resolved.error.status).json({ error: resolved.error.message }); return; } resolved.repo.setStatus(resolved.roomId, parsed.data.status, parsed.data.snoozeUntil ?? null); res.json({ success: true }); }); /** * Marks many conversations read in one request so draining a filtered backlog * is a single action. Individual failures are reported rather than aborting * the whole batch — a stopped account should not block the rest. */ router.post('/conversations/bulk-read', async (req, res) => { const parsed = BulkMarkReadRequestSchema.safeParse(req.body ?? {}); if (!parsed.success) { res.status(400).json({ error: 'conversationIds must be a non-empty array of at most 500 ids' }); return; } const failed: string[] = []; let updated = 0; for (const id of parsed.data.conversationIds) { const resolved = await resolveReadState(id); if ('error' in resolved) { failed.push(id); continue; } resolved.repo.markRead(resolved.roomId, resolved.conversation.lastMessageTime); updated += 1; } res.json({ success: true, updated, failed }); }); /** * Builds the sweep input for one account's conversations. * * Cached summaries frequently carry no message history — delta sync only * populates room summaries — so the newest inbound snippet stands in for the * transcript rather than forcing a full lazy load of every room, which would * turn one button press into hundreds of Matrix round trips. */ function sweepInput(conversations: ActiveChat[]): SellerSweepConversation[] { const input: SellerSweepConversation[] = []; for (const conversation of conversations) { if (!conversation.roomId) continue; const messages = (conversation.messages ?? []).map(({ sender, text, isSelf }) => ({ sender, text, isSelf })); const effective = messages.length > 0 ? messages : conversation.isFromUs ? [] : [{ sender: conversation.username, text: conversation.lastMessageSnippet, isSelf: false }]; const latestInbound = [...effective].reverse().find((message) => !message.isSelf); input.push({ conversationId: conversation.roomId, username: conversation.username, messages: effective, watermark: latestInbound ? `${conversation.lastMessageTime}:${conversationMessageFingerprint(latestInbound.text)}` : `${conversation.lastMessageTime}`, }); } return input; } function sweepOptions(): SellerSweepOptions { return { // A forced sweep deliberately ignores `reddit.filterSellers` and // `conversations.aiReplyScreening.enabled`: the operator asked for this // pass explicitly. The thresholds and the kill policy still come from // configuration, so a swept conversation lands in the same state the // live reply loop would have put it in. sellerKeywords: ctx.config.reddit.sellerKeywords, sellerDomains: ctx.config.reddit.sellerDomains, killConfidence: ctx.config.conversations.aiReplyScreening.killConfidence, flagConfidence: ctx.config.conversations.aiReplyScreening.flagConfidence, killOnSellerKeyword: ctx.config.conversations.killOnSellerKeyword, }; } /** * Forces a seller pass over every active conversation. Deterministic rules * settle what they can for free, unchanged conversations are skipped against * their last clean watermark, and only the remainder is batched into shared * classifier requests. */ router.post('/conversations/seller-scan', async (_req, res) => { const chatService = ctx.services.chat; if (!chatService) { res.status(503).json({ error: 'Chat service not initialized' }); return; } try { const runtimes = ctx.services.accountRuntimes?.list().filter((runtime) => runtime.state === 'running') ?? []; const targets = runtimes.length > 0 ? runtimes.map((snapshot) => { const runtime = ctx.services.accountRuntimes!.get(snapshot.accountId)!; return { accountId: snapshot.accountId, chat: runtime.resources.chatService, states: runtime.resources.conversationStateRepo, triage: runtime.resources.conversationReadStateRepo, }; }) : [{ accountId: currentOwner()?.accountId, chat: chatService, states: ctx.services.conversationState, triage: ctx.services.conversationReadState, }]; const totals: SellerSweepSummary = { scanned: 0, flagged: [], alreadyFlagged: 0, unchanged: 0, deterministic: 0, aiScreened: 0, aiRequests: 0, budgetSkipped: 0, emptySkipped: 0, degraded: false, }; for (const target of targets) { if (!target.states) continue; const summary = await sweepConversationsForSellers( sweepInput(await target.chat.getCachedConversations()), sweepOptions(), { states: target.states, triage: target.triage, screener: ctx.services.ai, logger: ctx.logger }, ); totals.scanned += summary.scanned; totals.alreadyFlagged += summary.alreadyFlagged; totals.unchanged += summary.unchanged; totals.deterministic += summary.deterministic; totals.aiScreened += summary.aiScreened; totals.aiRequests += summary.aiRequests; totals.budgetSkipped += summary.budgetSkipped; totals.emptySkipped += summary.emptySkipped; totals.degraded ||= summary.degraded; for (const flag of summary.flagged) { totals.flagged.push({ ...flag, conversationId: target.accountId ? `${encodeURIComponent(target.accountId)}~${encodeURIComponent(flag.conversationId)}` : flag.conversationId, }); } } if (totals.flagged.length > 0) { ctx.broadcaster.push('info', `Seller sweep flagged ${totals.flagged.length} of ${totals.scanned} conversations`); } res.json({ success: true, scanned: totals.scanned, flagged: totals.flagged, alreadyFlagged: totals.alreadyFlagged, unchanged: totals.unchanged, deterministic: totals.deterministic, aiScreened: totals.aiScreened, aiRequests: totals.aiRequests, budgetSkipped: totals.budgetSkipped, emptySkipped: totals.emptySkipped, degraded: totals.degraded, }); } catch (error) { ctx.logger.error('Failed to sweep conversations for sellers:', error); res.status(500).json({ error: 'Failed to scan conversations for sellers' }); } }); router.get('/conversations/:id/shareable-media', async (req, res) => { try { const resolved = await findConversation(decodeURIComponent(req.params.id)); if (!resolved) return res.status(404).json({ error: 'Conversation not found' }); const exchange = resolved.runtime?.resources.mediaExchange; if (!resolved.runtime || !exchange) return res.status(503).json({ error: 'Media sharing is not initialized for this account' }); const conversation = resolved.conversation; if (!conversation.roomId) { return res.json({ requestKind: null, assets: [], reason: 'This conversation has no Matrix room.', warning: null }); } // Full manual control: every enabled, unused library image is offered // regardless of what (if anything) the latest message asked for. Safety // flags surface as a non-blocking warning, never a gate. const assets = exchange.listShareableManual(conversation.roomId); const latestInbound = (conversation.messages ?? []).filter((message) => !message.isSelf).at(-1); const requestKind = latestInbound ? exchange.detect(conversation.roomId, latestInbound.text) : null; const reason = assets.length === 0 ? 'No enabled, unused library images are available to share.' : null; return res.json({ requestKind, assets: assets.map(toMediaLibraryResponse), reason, warning: manualShareWarning(conversation, resolved.runtime), }); } catch (error) { ctx.logger.error('Failed to list shareable conversation media:', error); return res.status(500).json({ error: 'Failed to list shareable media' }); } }); router.post('/conversations/:id/share-media', async (req, res) => { const parsed = ShareConversationMediaRequestSchema.safeParse(req.body); if (!parsed.success) return res.status(400).json({ error: 'A valid library asset ID is required' }); try { const resolved = await findConversation(decodeURIComponent(req.params.id)); if (!resolved) return res.status(404).json({ error: 'Conversation not found' }); const runtime = resolved.runtime; const exchange = runtime?.resources.mediaExchange; if (!runtime || !exchange) return res.status(503).json({ error: 'Media sharing is not initialized for this account' }); const chatService = runtime.resources.chatService; if (!resolved.conversation.roomId) return res.status(409).json({ error: 'This conversation has no Matrix room.' }); const result = await runtime.resources.rateLimiter.throttle( async () => exchange.shareManual(resolved.conversation.roomId!, parsed.data.assetId), (outcome) => outcome.kind === 'sent' || (outcome.kind === 'human-required' && outcome.remoteSent === true), ); if (result.kind !== 'sent' && !(result.kind === 'human-required' && result.remoteSent === true)) { const reason = result.kind === 'not-requested' ? 'The latest inbound message does not ask for an image.' : result.kind === 'dry-run' ? 'Dry-run mode prevented the image from being sent.' : result.kind === 'retryable' ? 'The image send can be retried later.' : result.reason; return res.status(409).json({ error: reason }); } const roomId = resolved.conversation.roomId!; const state = mergeConversationState( runtime.resources.conversationStateRepo.get(roomId), buildConversationState((resolved.conversation.messages ?? []).map(({ sender, text, isSelf }) => ({ sender, text, isSelf }))), ); state.automation.status = 'user-overridden'; state.automation.handoffReason = 'An exchange-library image was manually shared in this conversation.'; runtime.resources.conversationStateRepo.save(roomId, state); runtime.resources.activityLogRepo.insert({ timestamp: Date.now(), type: 'sent', message: `Shared a library image with u/${resolved.conversation.username}`, recipient: resolved.conversation.username, }); ctx.broadcaster.push('info', `Library image shared with u/${resolved.conversation.username}`); try { await chatService.refreshTopChats(5); } catch (error) { ctx.logger.error('Failed to refresh chats after share-media:', error); } if (result.kind === 'human-required') return res.status(502).json({ error: result.reason }); return res.json({ success: true }); } catch (error) { ctx.logger.error('Failed to share conversation media:', error); return res.status(500).json({ error: 'Failed to share library image' }); } }); /** * 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; }