diff --git a/.changeset/seven-squids-carry.md b/.changeset/seven-squids-carry.md new file mode 100644 index 0000000000..a845151cc8 --- /dev/null +++ b/.changeset/seven-squids-carry.md @@ -0,0 +1,2 @@ +--- +--- diff --git a/server/src/addie/jobs/slack-history-backfill.ts b/server/src/addie/jobs/slack-history-backfill.ts new file mode 100644 index 0000000000..15e9429548 --- /dev/null +++ b/server/src/addie/jobs/slack-history-backfill.ts @@ -0,0 +1,410 @@ +/** + * Slack History Backfill Job + * + * Indexes historical Slack messages for Addie's search_slack tool. + * This job fetches message history from channels and stores them + * in the addie_knowledge table for full-text search. + * + * Features: + * - Fetches both top-level messages and thread replies + * - Supports public and private channels + * - Excludes sensitive channels (admin, billing) by default + * - Idempotent - running multiple times won't create duplicates + */ + +import { logger } from '../../logger.js'; +import { AddieDatabase } from '../../db/addie-db.js'; +import { + getSlackChannels, + getFullChannelHistory, + getThreadReplies, + getSlackUser, + getChannelInfo, + type SlackHistoryMessage, +} from '../../slack/client.js'; + +const addieDb = new AddieDatabase(); + +/** + * Default channels to exclude from indexing (sensitive/admin channels) + * These contain billing info, admin discussions, etc. that shouldn't be searchable + */ +const DEFAULT_EXCLUDED_CHANNELS = [ + 'aao-admin', + 'aao-billing', + 'admin', + 'billing', + 'finance', + 'hr', + 'legal', + 'executive', + 'board', + 'confidential', +]; + +export interface BackfillResult { + channelsProcessed: number; + messagesIndexed: number; + threadRepliesIndexed: number; + messagesSkipped: number; + errors: number; + channels: Array<{ + id: string; + name: string; + isPrivate: boolean; + messagesIndexed: number; + threadRepliesIndexed: number; + messagesSkipped: number; + }>; +} + +export interface BackfillOptions { + /** Number of days of history to fetch (default: 90) */ + daysBack?: number; + /** Maximum messages per channel (default: 1000) */ + maxMessagesPerChannel?: number; + /** Specific channel IDs to backfill (default: all accessible channels) */ + channelIds?: string[]; + /** Include private channels (default: false) */ + includePrivateChannels?: boolean; + /** Channel names to exclude (default: admin/billing channels) */ + excludeChannelNames?: string[]; + /** Minimum message length to index (default: 20) */ + minMessageLength?: number; + /** Include thread replies (default: true) */ + includeThreadReplies?: boolean; + /** Progress callback */ + onProgress?: (status: { + channel: string; + messagesProcessed: number; + totalChannels: number; + currentChannel: number; + phase: 'messages' | 'threads'; + }) => void; +} + +/** + * Run the Slack history backfill job + * + * Fetches historical messages from channels and indexes them + * for Addie's search functionality. + */ +export async function runSlackHistoryBackfill(options: BackfillOptions = {}): Promise { + const { + daysBack = 90, + maxMessagesPerChannel = 1000, + channelIds, + includePrivateChannels = false, + excludeChannelNames = DEFAULT_EXCLUDED_CHANNELS, + minMessageLength = 20, + includeThreadReplies = true, + onProgress, + } = options; + + logger.info({ + daysBack, + maxMessagesPerChannel, + channelIds, + includePrivateChannels, + excludeChannelNames, + includeThreadReplies, + }, 'Starting Slack history backfill'); + + const result: BackfillResult = { + channelsProcessed: 0, + messagesIndexed: 0, + threadRepliesIndexed: 0, + messagesSkipped: 0, + errors: 0, + channels: [], + }; + + // Calculate oldest timestamp + const oldestDate = new Date(); + oldestDate.setDate(oldestDate.getDate() - daysBack); + const oldestTs = (oldestDate.getTime() / 1000).toString(); + + // Normalize excluded channel names for comparison + const excludedNamesLower = excludeChannelNames.map(n => n.toLowerCase()); + + // Get channels to process + let channels: Array<{ id: string; name: string; isPrivate: boolean }>; + + if (channelIds && channelIds.length > 0) { + // Use specific channels + channels = []; + for (const id of channelIds) { + const channelInfo = await getChannelInfo(id); + if (channelInfo) { + const name = channelInfo.name || 'unknown'; + // Check exclusion list even for specific channel IDs + if (excludedNamesLower.includes(name.toLowerCase())) { + logger.info({ channelId: id, name }, 'Skipping excluded channel'); + continue; + } + channels.push({ + id, + name, + isPrivate: channelInfo.is_private || false, + }); + } else { + logger.warn({ channelId: id }, 'Could not fetch channel info, skipping'); + } + } + } else { + // Get channels based on type preference + const channelTypes = includePrivateChannels + ? 'public_channel,private_channel' + : 'public_channel'; + + const allChannels = await getSlackChannels({ + types: channelTypes, + exclude_archived: true, + }); + + // Filter out excluded channels + channels = allChannels + .filter(c => !excludedNamesLower.includes((c.name || '').toLowerCase())) + .map(c => ({ + id: c.id, + name: c.name || 'unknown', + isPrivate: c.is_private || false, + })); + } + + logger.info({ + channelCount: channels.length, + publicCount: channels.filter(c => !c.isPrivate).length, + privateCount: channels.filter(c => c.isPrivate).length, + }, 'Backfilling channels'); + + // User cache to avoid repeated lookups + const userCache = new Map(); + + async function getUserDisplayName(userId: string): Promise { + if (userCache.has(userId)) { + return userCache.get(userId)?.displayName || 'unknown'; + } + + try { + const user = await getSlackUser(userId); + if (user) { + const displayName = user.profile?.display_name || user.profile?.real_name || user.name || 'unknown'; + userCache.set(userId, { displayName }); + return displayName; + } + } catch (error) { + logger.debug({ error, userId }, 'Failed to fetch user info'); + } + + userCache.set(userId, null); + return 'unknown'; + } + + async function indexMessage( + channelId: string, + channelName: string, + message: { user?: string; text?: string; ts: string; bot_id?: string; subtype?: string }, + channelResult: { messagesIndexed: number; threadRepliesIndexed: number; messagesSkipped: number }, + isThreadReply: boolean = false + ): Promise { + // Skip bot messages, subtypes (edits, deletes, etc.), and messages without text + if (message.bot_id || message.subtype || !message.text || !message.user) { + channelResult.messagesSkipped++; + return false; + } + + // Skip short messages + if (message.text.length < minMessageLength) { + channelResult.messagesSkipped++; + return false; + } + + try { + const username = await getUserDisplayName(message.user); + + // Construct permalink using env var or default workspace + const workspaceUrl = process.env.SLACK_WORKSPACE_URL || 'https://agenticads.slack.com'; + const tsForLink = message.ts.replace('.', ''); + const permalink = `${workspaceUrl}/archives/${channelId}/p${tsForLink}`; + + await addieDb.indexSlackMessage({ + channel_id: channelId, + channel_name: channelName, + user_id: message.user, + username, + ts: message.ts, + text: message.text, + permalink, + }); + + if (isThreadReply) { + channelResult.threadRepliesIndexed++; + } else { + channelResult.messagesIndexed++; + } + return true; + } catch (error) { + const errorMessage = error instanceof Error ? error.message : String(error); + + // Unique constraint violations are expected for duplicates - log at debug + if (errorMessage.includes('duplicate') || errorMessage.includes('unique') || errorMessage.includes('violates unique constraint')) { + logger.debug({ channelId, ts: message.ts }, 'Skipping duplicate message'); + } else { + // Unexpected error - log at warning level + logger.warn({ error, channelId, ts: message.ts }, 'Unexpected error indexing message'); + } + + channelResult.messagesSkipped++; + return false; + } + } + + // Process each channel + for (let i = 0; i < channels.length; i++) { + const channel = channels[i]; + const channelResult = { + id: channel.id, + name: channel.name, + isPrivate: channel.isPrivate, + messagesIndexed: 0, + threadRepliesIndexed: 0, + messagesSkipped: 0, + }; + + try { + logger.info({ + channel: channel.name, + channelId: channel.id, + isPrivate: channel.isPrivate, + }, 'Fetching channel history'); + + // Fetch messages + const messages = await getFullChannelHistory(channel.id, { + oldest: oldestTs, + maxMessages: maxMessagesPerChannel, + onProgress: (count) => { + if (onProgress) { + onProgress({ + channel: channel.name, + messagesProcessed: count, + totalChannels: channels.length, + currentChannel: i + 1, + phase: 'messages', + }); + } + }, + }); + + logger.info({ channel: channel.name, messageCount: messages.length }, 'Processing messages'); + + // Track threads that need reply fetching + const threadsToFetch: string[] = []; + + // Index each message + for (const message of messages) { + await indexMessage(channel.id, channel.name, message, channelResult, false); + + // Track threads with replies for later fetching + if (includeThreadReplies && message.thread_ts === message.ts && 'reply_count' in message) { + const replyCount = (message as SlackHistoryMessage & { reply_count?: number }).reply_count; + if (replyCount && replyCount > 0) { + threadsToFetch.push(message.ts); + } + } + } + + // Fetch and index thread replies + if (includeThreadReplies && threadsToFetch.length > 0) { + logger.info({ + channel: channel.name, + threadCount: threadsToFetch.length, + }, 'Fetching thread replies'); + + for (let t = 0; t < threadsToFetch.length; t++) { + const threadTs = threadsToFetch[t]; + + if (onProgress) { + onProgress({ + channel: channel.name, + messagesProcessed: t + 1, + totalChannels: channels.length, + currentChannel: i + 1, + phase: 'threads', + }); + } + + try { + const replies = await getThreadReplies(channel.id, threadTs); + + // Skip the first message (it's the parent, already indexed) + for (const reply of replies.slice(1)) { + await indexMessage(channel.id, channel.name, reply, channelResult, true); + } + } catch (error) { + logger.debug({ error, channelId: channel.id, threadTs }, 'Failed to fetch thread replies'); + } + + // Rate limit between thread fetches to avoid hitting Slack API limits + await new Promise(resolve => setTimeout(resolve, 200)); + } + } + + result.channelsProcessed++; + result.messagesIndexed += channelResult.messagesIndexed; + result.threadRepliesIndexed += channelResult.threadRepliesIndexed; + result.messagesSkipped += channelResult.messagesSkipped; + result.channels.push(channelResult); + + logger.info({ + channel: channel.name, + isPrivate: channel.isPrivate, + indexed: channelResult.messagesIndexed, + threadReplies: channelResult.threadRepliesIndexed, + skipped: channelResult.messagesSkipped, + }, 'Channel backfill complete'); + + } catch (error) { + logger.error({ error, channel: channel.name, channelId: channel.id }, 'Failed to process channel'); + result.errors++; + result.channels.push({ + ...channelResult, + messagesIndexed: 0, + threadRepliesIndexed: 0, + messagesSkipped: 0, + }); + } + } + + logger.info({ + channelsProcessed: result.channelsProcessed, + messagesIndexed: result.messagesIndexed, + threadRepliesIndexed: result.threadRepliesIndexed, + messagesSkipped: result.messagesSkipped, + errors: result.errors, + }, 'Slack history backfill complete'); + + return result; +} + +/** + * Get current backfill status + * Returns stats about what's currently indexed + */ +export async function getBackfillStatus(): Promise<{ + totalMessages: number; + channelCounts: Array<{ channel: string; count: number }>; + oldestMessage: Date | null; + newestMessage: Date | null; +}> { + const count = await addieDb.getSlackMessageCount(); + + // Get channel breakdown - this requires a custom query + // For now, return basic stats + return { + totalMessages: count, + channelCounts: [], // Would need to add a method to AddieDatabase + oldestMessage: null, + newestMessage: null, + }; +} diff --git a/server/src/addie/mcp/knowledge-search.ts b/server/src/addie/mcp/knowledge-search.ts index 9a7608660b..9e4b30bba1 100644 --- a/server/src/addie/mcp/knowledge-search.ts +++ b/server/src/addie/mcp/knowledge-search.ts @@ -229,23 +229,51 @@ export const KNOWLEDGE_TOOLS: AddieTool[] = [ { name: 'search_slack', description: - 'Search Slack messages from public channels in the AAO workspace. Use this when you need community discussions, Q&A threads, or real-world implementation examples. Recent messages are searched instantly from local index; older messages may fall back to live API (slower). Cite the Slack permalink when using information from results.', - usage_hints: 'use for community Q&A, "what did someone say about X?", real-world discussions', + 'Search Slack messages from public channels in the AAO workspace. Use this when you need community discussions, Q&A threads, or real-world implementation examples. When asked about a specific channel or working group (e.g., "Governance working group"), use the channel parameter to filter results. When asked to summarize discussions, search for relevant keywords then synthesize the results. Cite the Slack permalink when using information from results.', + usage_hints: 'use for community Q&A, "what did someone say about X?", channel summaries, working group discussions', input_schema: { type: 'object', properties: { query: { type: 'string', - description: 'Search query - keywords or phrases to find in Slack messages', + description: 'Search query - keywords or phrases to find in Slack messages. Use broad terms when summarizing a channel (e.g., "governance" for governance discussions).', + }, + channel: { + type: 'string', + description: 'Optional channel name to filter results (e.g., "governance-wg", "general"). Partial matches work.', }, limit: { type: 'number', - description: 'Maximum number of results (default 5, max 10)', + description: 'Maximum number of results (default 10, max 25 for summaries)', }, }, required: ['query'], }, }, + { + name: 'get_channel_activity', + description: + 'Get recent messages from a specific Slack channel. Use this when asked to summarize channel activity, see what a working group has been discussing, or get an overview of conversations in a channel. Returns messages sorted by recency. After getting results, synthesize them into a summary for the user.', + usage_hints: 'use for "summarize the governance channel", "what has the X working group been discussing?", channel overviews', + input_schema: { + type: 'object', + properties: { + channel: { + type: 'string', + description: 'Channel name to get activity from (e.g., "governance-wg", "general"). Partial matches work.', + }, + days: { + type: 'number', + description: 'How many days back to look (default 30, max 90)', + }, + limit: { + type: 'number', + description: 'Maximum number of messages to return (default 25, max 50)', + }, + }, + required: ['channel'], + }, + }, { name: 'search_resources', description: @@ -610,12 +638,13 @@ ${excerpt}`; }); handlers.set('search_slack', async (input) => { - const query = input.query as string; - const limit = Math.min((input.limit as number) || 5, 10); + const searchQuery = input.query as string; + const channel = input.channel as string | undefined; + const limit = Math.min((input.limit as number) || 10, 25); try { - // First, try local database search (instant, ~100ms) - const localResults = await addieDb.searchSlackMessages(query, { limit }); + // Search local database with optional channel filter + const localResults = await addieDb.searchSlackMessages(searchQuery, { limit, channel }); if (localResults.length > 0) { const formatted = localResults @@ -634,17 +663,80 @@ ${excerpt}`; }) .join('\n\n'); - return `Found ${localResults.length} Slack messages (from local index):\n\n${formatted}\n\n**Remember to cite the Slack permalink when using this information.**`; + const channelNote = channel ? ` in channels matching "${channel}"` : ''; + return `Found ${localResults.length} Slack messages${channelNote}:\n\n${formatted}\n\n**Remember to cite the Slack permalink when using this information.**`; } - // No local results found - return `No Slack discussions found for: "${query}"\n\nTry search_docs for documentation or web_search for external sources.`; + // No local results found - provide helpful guidance + const channelNote = channel ? ` in channel "${channel}"` : ''; + return `No Slack discussions found for: "${searchQuery}"${channelNote}\n\nTry:\n- Broader search terms\n- Removing the channel filter\n- search_docs for documentation`; } catch (error) { - logger.error({ error, query }, 'Addie: Slack search failed'); + logger.error({ error, query: searchQuery, channel }, 'Addie: Slack search failed'); return `Slack search failed: ${error instanceof Error ? error.message : 'Unknown error'}`; } }); + handlers.set('get_channel_activity', async (input) => { + const channel = input.channel as string; + const days = input.days as number | undefined; + const limit = input.limit as number | undefined; + + try { + const messages = await addieDb.getChannelActivity(channel, { days, limit }); + + if (messages.length === 0) { + return `No recent activity found in channels matching "${channel}".\n\nThis could mean:\n- The channel name might be different (try partial matches like "govern" for "governance-wg")\n- No messages in the last ${days ?? 30} days\n- The channel may not be indexed yet`; + } + + // Group messages by user to help with "who's most active" analysis + const userCounts = new Map(); + for (const msg of messages) { + userCounts.set(msg.username, (userCounts.get(msg.username) || 0) + 1); + } + const topUsers = [...userCounts.entries()] + .sort((a, b) => b[1] - a[1]) + .slice(0, 5) + .map(([name, count]) => `@${name} (${count})`) + .join(', '); + + const formatted = messages + .map((msg, i) => { + const cleanText = msg.text + .replace(/\s+/g, ' ') + .trim() + .substring(0, 400); + const truncated = cleanText.length < msg.text.length ? '...' : ''; + const date = new Date(msg.created_at).toLocaleDateString('en-US', { + month: 'short', + day: 'numeric', + }); + + return `### ${i + 1}. @${msg.username} (${date}) +"${cleanText}${truncated}" + +**Source:** ${msg.permalink}`; + }) + .join('\n\n'); + + const channelName = messages[0]?.channel_name || channel; + return `## Recent activity in #${channelName} + +**${messages.length} messages** from the last ${days ?? 30} days +**Most active:** ${topUsers} + +--- + +${formatted} + +--- + +**When summarizing:** Focus on key themes, decisions, and who contributed to each topic. Cite specific messages using their Slack permalinks.`; + } catch (error) { + logger.error({ error, channel }, 'Addie: get_channel_activity failed'); + return `Failed to get channel activity: ${error instanceof Error ? error.message : 'Unknown error'}`; + } + }); + handlers.set('search_resources', async (input) => { const query = input.query as string; const limit = Math.min((input.limit as number) || 5, 10); diff --git a/server/src/db/addie-db.ts b/server/src/db/addie-db.ts index e469ad8a7c..4fa5bf481f 100644 --- a/server/src/db/addie-db.ts +++ b/server/src/db/addie-db.ts @@ -616,8 +616,19 @@ export class AddieDatabase { */ async searchSlackMessages(searchQuery: string, options: { limit?: number; + channel?: string; } = {}): Promise { const limit = options.limit ?? 10; + const channel = options.channel; + + // Build query with optional channel filter + const channelFilter = channel + ? `AND LOWER(slack_channel_name) LIKE LOWER($3)` + : ''; + const params: (string | number)[] = [searchQuery, limit]; + if (channel) { + params.push(`%${channel}%`); + } const result = await query( `SELECT @@ -633,9 +644,10 @@ export class AddieDatabase { WHERE is_active = TRUE AND source_type = 'slack' AND search_vector @@ websearch_to_tsquery('english', $1) + ${channelFilter} ORDER BY rank DESC LIMIT $2`, - [searchQuery, limit] + params ); return result.rows; } @@ -650,6 +662,47 @@ export class AddieDatabase { return parseInt(result.rows[0]?.count ?? '0', 10); } + /** + * Get recent messages from a channel (no keyword search, just by recency) + */ + async getChannelActivity(channel: string, options: { + days?: number; + limit?: number; + } = {}): Promise> { + const days = Math.min(options.days ?? 30, 90); + const limit = Math.min(options.limit ?? 25, 50); + + const result = await query<{ + text: string; + channel_name: string; + username: string; + permalink: string; + created_at: Date; + }>( + `SELECT + content as text, + slack_channel_name as channel_name, + slack_username as username, + slack_permalink as permalink, + created_at + FROM addie_knowledge + WHERE is_active = TRUE + AND source_type = 'slack' + AND LOWER(slack_channel_name) LIKE LOWER($1) + AND created_at >= NOW() - INTERVAL '1 day' * $2 + ORDER BY created_at DESC + LIMIT $3`, + [`%${channel}%`, days, limit] + ); + return result.rows; + } + // ============== Curated Resource Indexing ============== /** diff --git a/server/src/routes/addie-admin.ts b/server/src/routes/addie-admin.ts index 152481e53e..faa56795a7 100644 --- a/server/src/routes/addie-admin.ts +++ b/server/src/routes/addie-admin.ts @@ -21,6 +21,7 @@ import Anthropic from "@anthropic-ai/sdk"; import { getAddieBoltApp } from "../addie/bolt-app.js"; import { AddieRouter, type RoutingContext } from "../addie/router.js"; import { sanitizeInput } from "../addie/security.js"; +import { runSlackHistoryBackfill } from "../addie/jobs/slack-history-backfill.js"; const logger = createLogger("addie-admin-routes"); const addieDb = new AddieDatabase(); @@ -2166,5 +2167,140 @@ Be specific and actionable. Focus on patterns that could help improve Addie's be } }); + // ========================================================================= + // SLACK HISTORY BACKFILL API (mounted at /api/admin/addie/backfill) + // ========================================================================= + + // POST /api/admin/addie/backfill/slack - Trigger Slack history backfill + apiRouter.post("/backfill/slack", requireAuth, requireAdmin, async (req, res) => { + try { + const { + days_back, + max_messages_per_channel, + channel_ids, + include_private_channels, + exclude_channel_names, + include_thread_replies, + } = req.body; + + // Validate days_back + const daysBack = days_back ? parseInt(String(days_back), 10) : 90; + if (isNaN(daysBack) || daysBack < 1 || daysBack > 365) { + res.status(400).json({ error: "days_back must be between 1 and 365" }); + return; + } + + // Validate max_messages_per_channel + const maxMessagesPerChannel = max_messages_per_channel ? parseInt(String(max_messages_per_channel), 10) : 1000; + if (isNaN(maxMessagesPerChannel) || maxMessagesPerChannel < 1 || maxMessagesPerChannel > 10000) { + res.status(400).json({ error: "max_messages_per_channel must be between 1 and 10000" }); + return; + } + + // Validate channel_ids format if provided + if (channel_ids !== undefined) { + if (!Array.isArray(channel_ids)) { + res.status(400).json({ error: "channel_ids must be an array" }); + return; + } + for (const id of channel_ids) { + if (typeof id !== 'string' || !/^C[A-Z0-9]+$/i.test(id)) { + res.status(400).json({ error: `Invalid channel ID format: ${id}` }); + return; + } + } + } + + // Validate exclude_channel_names if provided + if (exclude_channel_names !== undefined) { + if (!Array.isArray(exclude_channel_names) || !exclude_channel_names.every(n => typeof n === 'string')) { + res.status(400).json({ error: "exclude_channel_names must be an array of strings" }); + return; + } + } + + logger.info({ + daysBack, + maxMessagesPerChannel, + channelIds: channel_ids, + includePrivateChannels: include_private_channels, + excludeChannelNames: exclude_channel_names, + includeThreadReplies: include_thread_replies, + }, "Starting Slack history backfill"); + + // Run backfill (this may take a while for large workspaces) + const result = await runSlackHistoryBackfill({ + daysBack, + maxMessagesPerChannel, + channelIds: channel_ids, + includePrivateChannels: include_private_channels ?? false, + excludeChannelNames: exclude_channel_names, + includeThreadReplies: include_thread_replies ?? true, + }); + + logger.info({ + channelsProcessed: result.channelsProcessed, + messagesIndexed: result.messagesIndexed, + threadRepliesIndexed: result.threadRepliesIndexed, + }, "Slack history backfill complete"); + + res.json({ + success: true, + result, + }); + } catch (error) { + logger.error({ err: error }, "Error running Slack history backfill"); + res.status(500).json({ + error: "Internal server error", + message: error instanceof Error ? error.message : "Unable to run backfill", + }); + } + }); + + // GET /api/admin/addie/backfill/slack/status - Get current Slack index status + apiRouter.get("/backfill/slack/status", requireAuth, requireAdmin, async (req, res) => { + try { + // Get count of indexed messages + const totalCount = await addieDb.getSlackMessageCount(); + + // Get channel breakdown + const channelResult = await query<{ slack_channel_name: string; count: string }>( + `SELECT slack_channel_name, COUNT(*)::text as count + FROM addie_knowledge + WHERE source_type = 'slack' AND is_active = TRUE + GROUP BY slack_channel_name + ORDER BY COUNT(*) DESC` + ); + + // Get date range + const dateResult = await query<{ oldest: string; newest: string }>( + `SELECT + MIN(created_at)::text as oldest, + MAX(created_at)::text as newest + FROM addie_knowledge + WHERE source_type = 'slack'` + ); + + res.json({ + success: true, + status: { + totalMessages: totalCount, + channels: channelResult.rows.map(r => ({ + name: r.slack_channel_name, + count: parseInt(r.count, 10), + })), + oldestIndexed: dateResult.rows[0]?.oldest || null, + newestIndexed: dateResult.rows[0]?.newest || null, + }, + }); + } catch (error) { + logger.error({ err: error }, "Error getting Slack index status"); + res.status(500).json({ + error: "Internal server error", + message: error instanceof Error ? error.message : "Unable to get status", + }); + } + }); + return { pageRouter, apiRouter }; } diff --git a/server/src/slack/client.ts b/server/src/slack/client.ts index 71ebe80483..e4faa766cb 100644 --- a/server/src/slack/client.ts +++ b/server/src/slack/client.ts @@ -722,3 +722,108 @@ export async function getUserChannels(userId: string): Promise { logger.debug({ userId, channelCount: channelIds.length }, 'Fetched user channel memberships'); return channelIds; } + +/** + * Message from conversations.history + */ +export interface SlackHistoryMessage { + type: string; + user?: string; + bot_id?: string; + text?: string; + ts: string; + thread_ts?: string; + subtype?: string; + reply_count?: number; // Number of replies in thread (for parent messages) +} + +/** + * Get channel message history (conversations.history) + * Returns messages from a channel, paginated + * + * @param channelId - The channel ID to fetch history from + * @param options - Pagination and filtering options + * @returns Array of messages and pagination info + */ +export async function getChannelHistory( + channelId: string, + options: { + oldest?: string; // Unix timestamp - only messages after this time + latest?: string; // Unix timestamp - only messages before this time + limit?: number; // Max messages per request (default 100, max 1000) + cursor?: string; // Pagination cursor + } = {} +): Promise<{ messages: SlackHistoryMessage[]; hasMore: boolean; nextCursor?: string }> { + try { + const response = await slackRequest<{ + messages: SlackHistoryMessage[]; + has_more: boolean; + response_metadata?: { next_cursor?: string }; + }>('conversations.history', { + channel: channelId, + oldest: options.oldest, + latest: options.latest, + limit: options.limit ?? 100, + cursor: options.cursor, + }); + + return { + messages: response.messages ?? [], + hasMore: response.has_more ?? false, + nextCursor: response.response_metadata?.next_cursor, + }; + } catch (error) { + logger.error({ error, channelId }, 'Failed to get channel history'); + return { messages: [], hasMore: false }; + } +} + +/** + * Get all messages from a channel within a time range + * Handles pagination automatically with rate limiting + * + * @param channelId - The channel ID to fetch history from + * @param options - Time range and limit options + * @returns Array of all messages in the time range + */ +export async function getFullChannelHistory( + channelId: string, + options: { + oldest?: string; // Unix timestamp - only messages after this time + latest?: string; // Unix timestamp - only messages before this time + maxMessages?: number; // Stop after this many messages (default: no limit) + onProgress?: (count: number) => void; // Callback for progress updates + } = {} +): Promise { + const allMessages: SlackHistoryMessage[] = []; + let cursor: string | undefined; + const maxMessages = options.maxMessages ?? Infinity; + + do { + const result = await getChannelHistory(channelId, { + oldest: options.oldest, + latest: options.latest, + limit: 200, // Fetch in larger batches for efficiency + cursor, + }); + + allMessages.push(...result.messages); + + if (options.onProgress) { + options.onProgress(allMessages.length); + } + + if (allMessages.length >= maxMessages) { + break; + } + + cursor = result.nextCursor; + + if (cursor) { + await sleep(RATE_LIMIT_DELAY_MS); + } + } while (cursor); + + logger.debug({ channelId, messageCount: allMessages.length }, 'Fetched full channel history'); + return allMessages.slice(0, maxMessages); +}