diff --git a/docs-site/content/docs/integrations/context-sources.mdx b/docs-site/content/docs/integrations/context-sources.mdx index 213f5a3d7..a5d8b03df 100644 --- a/docs-site/content/docs/integrations/context-sources.mdx +++ b/docs-site/content/docs/integrations/context-sources.mdx @@ -1,6 +1,6 @@ --- title: Context Sources -description: Ingest semantic context from dbt, MetricFlow, LookML, Metabase, Looker, and Notion. +description: Ingest semantic context from dbt, MetricFlow, LookML, Metabase, Looker, Notion, and Slack. --- Context sources feed your existing analytics tooling into **ktx**. During ingestion, **ktx** extracts metadata from each source and uses a reconciliation agent to reconcile it with your existing semantic layer and knowledge base - preserving accepted edits rather than overwriting. @@ -27,7 +27,7 @@ LookML uses top-level `repoUrl`, and MetricFlow uses nested | Field | Required | Description | |-------|----------|-------------| -| `driver` | Yes | Source connector: `dbt`, `metricflow`, `lookml`, `metabase`, `looker`, or `notion` | +| `driver` | Yes | Source connector: `dbt`, `metricflow`, `lookml`, `metabase`, `looker`, `notion`, or `slack` | | `source_dir` | For local file sources | Absolute or project-relative source directory | | `repo_url` | For Git-hosted dbt sources | Git repository URL | | `repoUrl` | For Git-hosted LookML sources | Git repository URL | @@ -374,6 +374,47 @@ Create an integration at [notion.so/my-integrations](https://www.notion.so/my-in - Incremental sync cursors are stored in `.ktx/db.sqlite`; don't add `last_successful_cursor` to `ktx.yaml` +--- + +## Slack + +Ingests messages from allowlisted Slack channels as wiki context. Slack is knowledge-only and does not produce semantic layer sources. + +### What it provides + +- Markdown pages for messages in configured channels +- Channel allowlisting so ingest never scans full workspace history + +### Connection config + +```yaml title="ktx.yaml" +connections: + team-slack: + driver: slack + bot_token_ref: env:SLACK_BOT_TOKEN + channel_ids: + - C0123456789 + max_messages_per_channel: 1000 +``` + +### Authentication + +| Method | Config | +|--------|--------| +| Bot token | `bot_token_ref: env:SLACK_BOT_TOKEN` | + +The Slack app needs `channels:history` for public channels and `groups:history` for private channels that the app is added to. + +### What gets ingested + +- `wiki/global/slack//.md` for channel messages + +### Notes + +- `channel_ids` is required and must not be empty +- Slack ingest reads only the configured `channel_ids`, never every workspace channel +- `max_messages_per_channel` caps each channel fetch in one ingest run + ## Common errors | Error or symptom | Likely cause | Recovery | @@ -382,4 +423,5 @@ Create an integration at [notion.so/my-integrations](https://www.notion.so/my-in | Private repo/API authentication fails | Token env var or secret file is missing | Export the env var or update `auth_token_ref` to a readable file | | Ingest creates duplicate context | Existing source names or wiki pages do not match imported terminology | Review the diff, rename duplicates, and add wiki pages with canonical names | | Notion ingest skips pages | Integration lacks access or root ids are missing | Share pages with the Notion integration and set `root_page_ids` or use `all_accessible` carefully | +| Slack ingest reads no messages | The app is not in the channel or the channel has no readable messages | Invite the app to the configured channel and verify the token has the matching history scope | | Generated semantic sources fail validation | Tool metadata does not match the live warehouse schema | Map BI/source databases to primary warehouse connections and rerun validation | diff --git a/packages/cli/src/context/connections/slack-config.test.ts b/packages/cli/src/context/connections/slack-config.test.ts new file mode 100644 index 000000000..7dfcf3965 --- /dev/null +++ b/packages/cli/src/context/connections/slack-config.test.ts @@ -0,0 +1,71 @@ +import { mkdtemp, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { parseSlackConnectionConfig, resolveSlackBotToken, slackConnectionToPullConfig } from './slack-config.js'; + +describe('standalone Slack connection config', () => { + let tempDir: string; + + beforeEach(async () => { + tempDir = await mkdtemp(join(tmpdir(), 'ktx-slack-config-')); + }); + + afterEach(async () => { + await rm(tempDir, { recursive: true, force: true }); + }); + + it('parses Slack config with a required channel allowlist', () => { + expect( + parseSlackConnectionConfig({ + driver: 'slack', + bot_token_ref: 'env:SLACK_BOT_TOKEN', + channel_ids: ['C2', 'C1', 'C1'], + max_messages_per_channel: 25, + }), + ).toEqual({ + driver: 'slack', + bot_token: null, + bot_token_ref: 'env:SLACK_BOT_TOKEN', + channel_ids: ['C1', 'C2'], + max_messages_per_channel: 25, + }); + }); + + it('rejects Slack config without channels', () => { + expect(() => + parseSlackConnectionConfig({ + driver: 'slack', + bot_token_ref: 'env:SLACK_BOT_TOKEN', + channel_ids: [], + }), + ).toThrow('Slack connection config requires at least one channel_id'); + }); + + it('resolves Slack env and file token references', async () => { + const tokenPath = join(tempDir, 'slack-token.txt'); + await writeFile(tokenPath, 'xoxb-file-token\n', 'utf-8'); + + await expect(resolveSlackBotToken('env:SLACK_BOT_TOKEN', { env: { SLACK_BOT_TOKEN: 'xoxb-env-token' } })).resolves.toBe( + 'xoxb-env-token', + ); + await expect(resolveSlackBotToken(`file:${tokenPath}`)).resolves.toBe('xoxb-file-token'); + }); + + it('converts Slack config into adapter pull config', async () => { + await expect( + slackConnectionToPullConfig( + parseSlackConnectionConfig({ + driver: 'slack', + bot_token_ref: 'env:SLACK_BOT_TOKEN', + channel_ids: ['C1'], + }), + { env: { SLACK_BOT_TOKEN: 'xoxb-env-token' } }, + ), + ).resolves.toEqual({ + authToken: 'xoxb-env-token', + channelIds: ['C1'], + maxMessagesPerChannel: 1000, + }); + }); +}); diff --git a/packages/cli/src/context/connections/slack-config.ts b/packages/cli/src/context/connections/slack-config.ts new file mode 100644 index 000000000..2353f6017 --- /dev/null +++ b/packages/cli/src/context/connections/slack-config.ts @@ -0,0 +1,138 @@ +import { readFile } from 'node:fs/promises'; +import { homedir } from 'node:os'; +import { resolve } from 'node:path'; +import { slackPullConfigSchema, type SlackPullConfig } from '../ingest/adapters/slack/types.js'; +import type { KtxProjectConnectionConfig } from '../project/config.js'; + +type RawKtxSlackConnectionConfig = Extract; + +export type KtxSlackConnectionConfig = Omit< + RawKtxSlackConnectionConfig, + 'bot_token' | 'bot_token_ref' | 'channel_ids' | 'max_messages_per_channel' +> & { + driver: 'slack'; + bot_token: string | null; + bot_token_ref: string | null; + channel_ids: string[]; + max_messages_per_channel: number; +}; + +interface ResolveSlackTokenOptions { + env?: Record; + readTextFile?: (path: string) => Promise; +} + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} + +function optionalString(value: unknown): string | null { + return typeof value === 'string' && value.trim().length > 0 ? value.trim() : null; +} + +function stringArray(value: unknown): string[] { + if (!Array.isArray(value)) { + return []; + } + return value.flatMap((item): string[] => { + if (typeof item !== 'string') { + return []; + } + const trimmed = item.trim(); + return trimmed ? [trimmed] : []; + }); +} + +function integerWithFallback(value: unknown, fallback: number, name: string): number { + if (value === undefined || value === null) { + return fallback; + } + if (typeof value !== 'number' || !Number.isInteger(value)) { + throw new Error(`${name} must be an integer`); + } + return value; +} + +function boundedInteger(value: unknown, fallback: number, name: string, min: number, max: number): number { + const parsed = integerWithFallback(value, fallback, name); + if (parsed < min || parsed > max) { + throw new Error(`${name} must be between ${min} and ${max}`); + } + return parsed; +} + +function expandHome(path: string): string { + return path === '~' || path.startsWith('~/') ? resolve(homedir(), path.slice(2)) : path; +} + +export function parseSlackConnectionConfig(raw: unknown): KtxSlackConnectionConfig { + if (!isRecord(raw)) { + throw new Error('Slack connection config must be an object'); + } + if (raw.driver !== 'slack') { + throw new Error('Slack connection config requires driver: slack'); + } + const botToken = optionalString(raw.bot_token); + const botTokenRef = optionalString(raw.bot_token_ref); + if (!botToken && !botTokenRef) { + throw new Error('Slack connection config requires bot_token or bot_token_ref'); + } + if (botTokenRef && !botTokenRef.startsWith('env:') && !botTokenRef.startsWith('file:')) { + throw new Error('Slack bot_token_ref must use env:NAME or file:/path'); + } + const channelIds = stringArray(raw.channel_ids); + if (channelIds.length === 0) { + throw new Error('Slack connection config requires at least one channel_id'); + } + + return { + driver: 'slack', + bot_token: botToken, + bot_token_ref: botTokenRef, + channel_ids: [...new Set(channelIds)].sort((left, right) => left.localeCompare(right)), + max_messages_per_channel: boundedInteger( + raw.max_messages_per_channel, + 1000, + 'max_messages_per_channel', + 1, + 10_000, + ), + }; +} + +/** @internal */ +export async function resolveSlackBotToken( + botTokenRef: string, + options: ResolveSlackTokenOptions = {}, +): Promise { + if (botTokenRef.startsWith('env:')) { + const envName = botTokenRef.slice('env:'.length); + const value = (options.env ?? process.env)[envName]; + if (!value) { + throw new Error(`Slack token environment variable ${envName} is not set`); + } + return value.trim(); + } + if (botTokenRef.startsWith('file:')) { + const path = expandHome(botTokenRef.slice('file:'.length)); + const readTextFile = options.readTextFile ?? ((filePath: string) => readFile(filePath, 'utf-8')); + const value = (await readTextFile(path)).trim(); + if (!value) { + throw new Error(`Slack token file is empty: ${path}`); + } + return value; + } + throw new Error('Slack bot_token_ref must use env:NAME or file:/path'); +} + +export async function slackConnectionToPullConfig( + config: KtxSlackConnectionConfig, + options: ResolveSlackTokenOptions = {}, +): Promise { + const authToken = config.bot_token ?? (await resolveSlackBotToken(config.bot_token_ref ?? '', options)); + return slackPullConfigSchema.parse({ + authToken, + channelIds: config.channel_ids, + maxMessagesPerChannel: config.max_messages_per_channel, + }); +} diff --git a/packages/cli/src/context/ingest/adapters/slack/chunk.ts b/packages/cli/src/context/ingest/adapters/slack/chunk.ts new file mode 100644 index 000000000..575eded5d --- /dev/null +++ b/packages/cli/src/context/ingest/adapters/slack/chunk.ts @@ -0,0 +1,80 @@ +import { createHash } from 'node:crypto'; +import { readdir, readFile } from 'node:fs/promises'; +import { join, relative } from 'node:path'; +import type { ChunkResult, DiffSet, ScopeDescriptor, WorkUnit } from '../../types.js'; +import { slackManifestSchema } from './types.js'; + +async function walk(root: string): Promise { + const entries = await readdir(root, { withFileTypes: true, recursive: true }); + return entries + .filter((entry) => entry.isFile()) + .map((entry) => relative(root, join(entry.parentPath, entry.name)).replace(/\\/g, '/')) + .sort(); +} + +function safeUnitKey(path: string): string { + return `slack-${path + .replace(/^wiki\/global\//, '') + .replace(/\.md$/, '') + .replace(/[^a-zA-Z0-9]+/g, '-') + .replace(/^-+|-+$/g, '')}`; +} + +async function readManifest(stagedDir: string) { + try { + return slackManifestSchema.parse(JSON.parse(await readFile(join(stagedDir, 'manifest.json'), 'utf-8'))); + } catch (error) { + throw new Error(`Invalid Slack manifest: ${error instanceof Error ? error.message : String(error)}`); + } +} + +export async function chunkSlackStagedDir(stagedDir: string, diffSet?: DiffSet): Promise { + const files = await walk(stagedDir); + const manifest = await readManifest(stagedDir); + const touched = diffSet ? new Set([...diffSet.added, ...diffSet.modified]) : null; + const markdownFiles = files.filter((path) => path.startsWith('wiki/global/') && path.endsWith('.md')); + const workUnits: WorkUnit[] = []; + + for (const markdownPath of markdownFiles) { + if (touched && !touched.has(markdownPath)) { + continue; + } + const dependencyPaths = ['manifest.json'].filter((path) => path !== markdownPath); + workUnits.push({ + unitKey: safeUnitKey(markdownPath), + displayLabel: markdownPath.replace(/^wiki\/global\//, ''), + rawFiles: [markdownPath], + dependencyPaths, + peerFileIndex: markdownFiles.filter((path) => path !== markdownPath).sort(), + notes: + 'Synthesize durable wiki knowledge from this allowlisted Slack source. Treat Slack as wiki-only context unless another mapped source is confirmed with discover_data.', + }); + } + + return { + workUnits, + eviction: diffSet && diffSet.deleted.length > 0 ? { deletedRawPaths: [...diffSet.deleted].sort() } : undefined, + reconcileNotes: [ + `Slack channels fetched: ${manifest.channelIds.join(', ')}`, + `Slack maxMessagesPerChannel=${manifest.maxMessagesPerChannel}`, + 'Slack ingest reads only configured allowlisted channels.', + ], + contextReport: { + warnings: manifest.warnings, + }, + }; +} + +export async function describeSlackScope(stagedDir: string): Promise { + const manifest = await readManifest(stagedDir); + const scopeKey = JSON.stringify({ + channelIds: [...manifest.channelIds].sort(), + maxMessagesPerChannel: manifest.maxMessagesPerChannel, + }); + const fingerprint = createHash('sha256').update(scopeKey).digest('hex'); + return { + fingerprint, + isPathInScope: (rawPath) => + rawPath === 'manifest.json' || rawPath.startsWith('wiki/global/slack/'), + }; +} diff --git a/packages/cli/src/context/ingest/adapters/slack/detect.ts b/packages/cli/src/context/ingest/adapters/slack/detect.ts new file mode 100644 index 000000000..0a3ad53d4 --- /dev/null +++ b/packages/cli/src/context/ingest/adapters/slack/detect.ts @@ -0,0 +1,21 @@ +import { readFile, readdir } from 'node:fs/promises'; +import { join } from 'node:path'; +import { SLACK_SOURCE_KEY } from './types.js'; + +export async function detectSlackStagedDir(stagedDir: string): Promise { + try { + const manifest = JSON.parse(await readFile(join(stagedDir, 'manifest.json'), 'utf-8')) as { source?: unknown }; + if (manifest.source === SLACK_SOURCE_KEY) { + return true; + } + } catch { + // Fall through to structural detection for manually staged dirs. + } + + try { + const entries = await readdir(join(stagedDir, 'wiki', 'global'), { withFileTypes: true, recursive: true }); + return entries.some((entry) => entry.isFile() && entry.name.endsWith('.md')); + } catch { + return false; + } +} diff --git a/packages/cli/src/context/ingest/adapters/slack/fetch.ts b/packages/cli/src/context/ingest/adapters/slack/fetch.ts new file mode 100644 index 000000000..c2f22c3c7 --- /dev/null +++ b/packages/cli/src/context/ingest/adapters/slack/fetch.ts @@ -0,0 +1,78 @@ +import { mkdir, writeFile } from 'node:fs/promises'; +import { dirname, join } from 'node:path'; +import { SlackWebApiClient, type SlackApi } from './slack-client.js'; +import { + SLACK_SOURCE_KEY, + slackPullConfigSchema, + type SlackChannelMessage, +} from './types.js'; + +export interface FetchSlackSnapshotParams { + client?: SlackApi; + config: unknown; + stagedDir: string; + now?: () => Date; +} + +async function writeText(path: string, value: string): Promise { + await mkdir(dirname(path), { recursive: true }); + await writeFile(path, value.endsWith('\n') ? value : `${value}\n`, 'utf-8'); +} + +function safeSlug(value: string): string { + return value + .toLowerCase() + .replace(/[^a-z0-9]+/g, '-') + .replace(/^-+|-+$/g, ''); +} + +function messageMarkdown(message: SlackChannelMessage): string { + const lines = [ + `# Slack message ${message.channelId}/${message.ts}`, + '', + `- Channel: ${message.channelId}`, + `- Message timestamp: ${message.ts}`, + message.user ? `- Author: ${message.user}` : null, + message.username ? `- Username: ${message.username}` : null, + message.botId ? `- Bot ID: ${message.botId}` : null, + message.subtype ? `- Subtype: ${message.subtype}` : null, + message.threadTs ? `- Thread timestamp: ${message.threadTs}` : null, + '', + '## Message', + '', + message.text.trim().length > 0 ? message.text.trim() : '_No text content._', + ].filter((line): line is string => line !== null); + return `${lines.join('\n')}\n`; +} + +export async function fetchSlackSnapshot(params: FetchSlackSnapshotParams): Promise { + const config = slackPullConfigSchema.parse(params.config); + const client = params.client ?? new SlackWebApiClient(config.authToken); + const fetchedAt = (params.now ?? (() => new Date()))().toISOString(); + let messageCount = 0; + + for (const channelId of config.channelIds) { + const messages = await client.listChannelMessages(channelId, config.maxMessagesPerChannel); + messageCount += messages.length; + for (const message of messages) { + const path = join(params.stagedDir, 'wiki', 'global', 'slack', safeSlug(channelId), `${safeSlug(message.ts)}.md`); + await writeText(path, messageMarkdown(message)); + } + } + + await writeText( + join(params.stagedDir, 'manifest.json'), + JSON.stringify( + { + source: SLACK_SOURCE_KEY, + fetchedAt, + channelIds: config.channelIds, + maxMessagesPerChannel: config.maxMessagesPerChannel, + messageCount, + warnings: [], + }, + null, + 2, + ), + ); +} diff --git a/packages/cli/src/context/ingest/adapters/slack/slack-client.test.ts b/packages/cli/src/context/ingest/adapters/slack/slack-client.test.ts new file mode 100644 index 000000000..21749a5dd --- /dev/null +++ b/packages/cli/src/context/ingest/adapters/slack/slack-client.test.ts @@ -0,0 +1,83 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { SlackWebApiClient } from './slack-client.js'; + +describe('SlackWebApiClient', () => { + afterEach(() => { + vi.unstubAllGlobals(); + }); + + it('normalizes allowlisted channel messages from conversations.history', async () => { + const fetchMock = vi.fn().mockResolvedValueOnce({ + json: async () => ({ + ok: true, + messages: [ + { + ts: '1768000000.000100', + user: 'U-AUTHOR', + text: 'Allowlisted launch rule', + thread_ts: '1768000000.000100', + }, + ], + has_more: false, + }), + }); + vi.stubGlobal('fetch', fetchMock); + + await expect(new SlackWebApiClient('xoxb-token').listChannelMessages('C123', 1000)).resolves.toEqual([ + { + channelId: 'C123', + ts: '1768000000.000100', + user: 'U-AUTHOR', + username: null, + botId: null, + subtype: null, + threadTs: '1768000000.000100', + text: 'Allowlisted launch rule', + }, + ]); + expect(fetchMock).toHaveBeenCalledWith( + 'https://slack.com/api/conversations.history', + expect.objectContaining({ + method: 'POST', + headers: { + authorization: 'Bearer xoxb-token', + 'content-type': 'application/json; charset=utf-8', + }, + body: JSON.stringify({ channel: 'C123', limit: 200 }), + }), + ); + }); + + it('follows history cursors until maxMessages is reached', async () => { + const fetchMock = vi + .fn() + .mockResolvedValueOnce({ + json: async () => ({ + ok: true, + messages: [{ ts: '1.000001', text: 'first' }], + has_more: true, + response_metadata: { next_cursor: 'cursor-1' }, + }), + }) + .mockResolvedValueOnce({ + json: async () => ({ + ok: true, + messages: [{ ts: '2.000002', text: 'second' }], + has_more: false, + }), + }); + vi.stubGlobal('fetch', fetchMock); + + await expect(new SlackWebApiClient('xoxb-token').listChannelMessages('C123', 2)).resolves.toEqual([ + expect.objectContaining({ ts: '1.000001', text: 'first' }), + expect.objectContaining({ ts: '2.000002', text: 'second' }), + ]); + expect(fetchMock).toHaveBeenNthCalledWith( + 2, + 'https://slack.com/api/conversations.history', + expect.objectContaining({ + body: JSON.stringify({ channel: 'C123', limit: 1, cursor: 'cursor-1' }), + }), + ); + }); +}); diff --git a/packages/cli/src/context/ingest/adapters/slack/slack-client.ts b/packages/cli/src/context/ingest/adapters/slack/slack-client.ts new file mode 100644 index 000000000..c288aefd7 --- /dev/null +++ b/packages/cli/src/context/ingest/adapters/slack/slack-client.ts @@ -0,0 +1,103 @@ +import type { SlackChannelMessage } from './types.js'; + +export interface SlackApi { + listChannelMessages(channelId: string, maxMessages: number): Promise; +} + +type SlackApiResponse = { ok?: unknown; error?: unknown }; + +interface SlackConversationsHistoryResponse extends SlackApiResponse { + messages?: unknown; + has_more?: unknown; + response_metadata?: { + next_cursor?: unknown; + }; +} + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} + +function stringValue(value: unknown): string | null { + return typeof value === 'string' && value.trim().length > 0 ? value : null; +} + +function normalizeSlackError(method: string, payload: SlackApiResponse): Error { + const error = typeof payload.error === 'string' && payload.error ? payload.error : 'unknown_error'; + return new Error(`Slack ${method} failed: ${error}`); +} + +function headers(authToken: string): Record { + return { + authorization: `Bearer ${authToken}`, + 'content-type': 'application/json; charset=utf-8', + }; +} + +export class SlackWebApiClient implements SlackApi { + constructor(private readonly authToken: string) {} + + async listChannelMessages(channelId: string, maxMessages: number): Promise { + const messages: SlackChannelMessage[] = []; + let cursor: string | null = null; + + do { + const pageLimit = Math.max(1, Math.min(200, maxMessages - messages.length)); + const payload = await this.fetchHistoryPage(channelId, pageLimit, cursor); + messages.push(...payload.messages); + cursor = payload.nextCursor; + } while (cursor && messages.length < maxMessages); + + return messages.slice(0, maxMessages); + } + + private async fetchHistoryPage( + channelId: string, + limit: number, + cursor: string | null, + ): Promise<{ messages: SlackChannelMessage[]; nextCursor: string | null }> { + const response = await fetch('https://slack.com/api/conversations.history', { + method: 'POST', + headers: headers(this.authToken), + body: JSON.stringify({ + channel: channelId, + limit, + ...(cursor ? { cursor } : {}), + }), + }); + const payload = (await response.json()) as SlackConversationsHistoryResponse; + if (payload.ok !== true) { + throw normalizeSlackError('conversations.history', payload); + } + + return { + messages: Array.isArray(payload.messages) + ? payload.messages.flatMap((message): SlackChannelMessage[] => { + if (!isRecord(message)) { + return []; + } + const ts = stringValue(message.ts); + if (!ts) { + return []; + } + return [ + { + channelId, + ts, + user: stringValue(message.user), + username: stringValue(message.username), + botId: stringValue(message.bot_id), + subtype: stringValue(message.subtype), + threadTs: stringValue(message.thread_ts), + text: stringValue(message.text) ?? '', + }, + ]; + }) + : [], + nextCursor: + payload.has_more === true && typeof payload.response_metadata?.next_cursor === 'string' + ? payload.response_metadata.next_cursor || null + : null, + }; + } +} diff --git a/packages/cli/src/context/ingest/adapters/slack/slack.adapter.test.ts b/packages/cli/src/context/ingest/adapters/slack/slack.adapter.test.ts new file mode 100644 index 000000000..a65edb32b --- /dev/null +++ b/packages/cli/src/context/ingest/adapters/slack/slack.adapter.test.ts @@ -0,0 +1,102 @@ +import { mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import type { SlackApi } from './slack-client.js'; +import { SlackSourceAdapter } from './slack.adapter.js'; + +describe('SlackSourceAdapter', () => { + let stagedDir: string; + let client: SlackApi; + let adapter: SlackSourceAdapter; + + beforeEach(async () => { + stagedDir = await mkdtemp(join(tmpdir(), 'slack-adapter-')); + client = { + listChannelMessages: vi.fn().mockResolvedValue([ + { + channelId: 'C123', + ts: '1768000000.000100', + user: 'U-AUTHOR', + username: null, + botId: null, + subtype: null, + threadTs: null, + text: 'Allowlisted launch rule', + }, + ]), + }; + adapter = new SlackSourceAdapter({ + client, + now: () => new Date('2026-05-23T00:00:00.000Z'), + }); + }); + + afterEach(async () => { + await rm(stagedDir, { recursive: true, force: true }); + }); + + it('declares Slack source behavior', () => { + expect(adapter.source).toBe('slack'); + expect(adapter.skillNames).toEqual(['notion_synthesize']); + expect(adapter.reconcileSkillNames).toEqual([]); + expect(adapter.evidenceIndexing).toBe('documents'); + expect(adapter.triageSupported).toBe(false); + }); + + it('fetches only configured channels into wiki/global markdown', async () => { + await adapter.fetch( + { + authToken: 'xoxb-token', + channelIds: ['C123'], + maxMessagesPerChannel: 250, + }, + stagedDir, + { connectionId: 'slack', sourceKey: 'slack' }, + ); + + expect(client.listChannelMessages).toHaveBeenCalledWith('C123', 250); + await expect(readFile(join(stagedDir, 'wiki/global/slack/c123/1768000000-000100.md'), 'utf-8')).resolves.toContain( + 'Allowlisted launch rule', + ); + await expect(adapter.detect(stagedDir)).resolves.toBe(true); + }); + + it('chunks staged Slack markdown files as wiki-only work units', async () => { + await mkdir(join(stagedDir, 'wiki/global/slack/c123'), { recursive: true }); + await writeFile( + join(stagedDir, 'manifest.json'), + JSON.stringify({ + source: 'slack', + fetchedAt: '2026-05-23T00:00:00.000Z', + channelIds: ['C123'], + maxMessagesPerChannel: 1000, + messageCount: 1, + warnings: [], + }), + 'utf-8', + ); + await writeFile(join(stagedDir, 'wiki/global/slack/c123/1.md'), '# Slack note\n', 'utf-8'); + + const result = await adapter.chunk(stagedDir); + + expect(result.workUnits).toEqual([ + expect.objectContaining({ + unitKey: 'slack-slack-c123-1', + rawFiles: ['wiki/global/slack/c123/1.md'], + dependencyPaths: ['manifest.json'], + }), + ]); + expect(result.reconcileNotes).toContain('Slack ingest reads only configured allowlisted channels.'); + }); + + it('rejects empty channel allowlists before calling Slack', async () => { + await expect( + adapter.fetch({ authToken: 'xoxb-token', channelIds: [] }, stagedDir, { + connectionId: 'slack', + sourceKey: 'slack', + }), + ).rejects.toThrow(); + expect(client.listChannelMessages).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/cli/src/context/ingest/adapters/slack/slack.adapter.ts b/packages/cli/src/context/ingest/adapters/slack/slack.adapter.ts new file mode 100644 index 000000000..6efcc2380 --- /dev/null +++ b/packages/cli/src/context/ingest/adapters/slack/slack.adapter.ts @@ -0,0 +1,41 @@ +import type { ChunkResult, DiffSet, FetchContext, ScopeDescriptor, SourceAdapter } from '../../types.js'; +import { chunkSlackStagedDir, describeSlackScope } from './chunk.js'; +import { detectSlackStagedDir } from './detect.js'; +import { fetchSlackSnapshot } from './fetch.js'; +import type { SlackApi } from './slack-client.js'; + +export interface SlackSourceAdapterDeps { + client?: SlackApi; + now?: () => Date; +} + +export class SlackSourceAdapter implements SourceAdapter { + readonly source = 'slack'; + readonly skillNames = ['notion_synthesize']; + readonly reconcileSkillNames: string[] = []; + readonly evidenceIndexing = 'documents' as const; + readonly triageSupported = false; + + constructor(private readonly deps: SlackSourceAdapterDeps = {}) {} + + detect(stagedDir: string): Promise { + return detectSlackStagedDir(stagedDir); + } + + fetch(pullConfig: unknown, stagedDir: string, _ctx: FetchContext): Promise { + return fetchSlackSnapshot({ + config: pullConfig, + stagedDir, + ...(this.deps.client ? { client: this.deps.client } : {}), + ...(this.deps.now ? { now: this.deps.now } : {}), + }); + } + + chunk(stagedDir: string, diffSet?: DiffSet): Promise { + return chunkSlackStagedDir(stagedDir, diffSet); + } + + describeScope(stagedDir: string): Promise { + return describeSlackScope(stagedDir); + } +} diff --git a/packages/cli/src/context/ingest/adapters/slack/types.ts b/packages/cli/src/context/ingest/adapters/slack/types.ts new file mode 100644 index 000000000..e4ce80966 --- /dev/null +++ b/packages/cli/src/context/ingest/adapters/slack/types.ts @@ -0,0 +1,30 @@ +import { z } from 'zod'; + +export const SLACK_SOURCE_KEY = 'slack'; + +export const slackPullConfigSchema = z.object({ + authToken: z.string().min(1), + channelIds: z.array(z.string().min(1)).min(1), + maxMessagesPerChannel: z.number().int().min(1).max(10_000).default(1000), +}); +export type SlackPullConfig = z.infer; + +export const slackManifestSchema = z.object({ + source: z.literal(SLACK_SOURCE_KEY), + fetchedAt: z.string().datetime(), + channelIds: z.array(z.string()), + maxMessagesPerChannel: z.number().int().min(1), + messageCount: z.number().int().min(0), + warnings: z.array(z.string()).default([]), +}); + +export interface SlackChannelMessage { + channelId: string; + ts: string; + user: string | null; + username: string | null; + botId: string | null; + subtype: string | null; + threadTs: string | null; + text: string; +} diff --git a/packages/cli/src/context/ingest/local-adapters.test.ts b/packages/cli/src/context/ingest/local-adapters.test.ts index 5f61c7396..071365477 100644 --- a/packages/cli/src/context/ingest/local-adapters.test.ts +++ b/packages/cli/src/context/ingest/local-adapters.test.ts @@ -46,6 +46,7 @@ describe('local ingest adapters', () => { 'looker', 'metricflow', 'notion', + 'slack', ]); expect(adapters.find((adapter) => adapter.source === 'metabase')?.fetch).toBeTypeOf('function'); }); @@ -704,4 +705,26 @@ describe('local ingest adapters', () => { parsedTargetTables: {}, }); }); + + it('resolves Slack bot_token_ref and keeps the channel allowlist', async () => { + const slackProject = projectWithConnections({ + slack_docs: { + driver: 'slack', + bot_token_ref: 'env:SLACK_BOT_TOKEN', + channel_ids: ['C1'], + max_messages_per_channel: 50, + }, + } as never); + const slack = createDefaultLocalIngestAdapters(slackProject).find((candidate) => candidate.source === 'slack'); + + await expect( + localPullConfigForAdapter(slackProject, slack!, 'slack_docs', { + looker: { env: { SLACK_BOT_TOKEN: 'xoxb-env-token' } as NodeJS.ProcessEnv }, + }), + ).resolves.toEqual({ + authToken: 'xoxb-env-token', + channelIds: ['C1'], + maxMessagesPerChannel: 50, + }); + }); }); diff --git a/packages/cli/src/context/ingest/local-adapters.ts b/packages/cli/src/context/ingest/local-adapters.ts index ea7556e7e..b7170d9a2 100644 --- a/packages/cli/src/context/ingest/local-adapters.ts +++ b/packages/cli/src/context/ingest/local-adapters.ts @@ -1,6 +1,7 @@ import { join } from 'node:path'; import { localConnectionToWarehouseDescriptor } from '../../context/connections/local-warehouse-descriptor.js'; import { notionConnectionToPullConfig, parseNotionConnectionConfig } from '../../context/connections/notion-config.js'; +import { parseSlackConnectionConfig, slackConnectionToPullConfig } from '../../context/connections/slack-config.js'; import { resolveKtxConfigReference } from '../core/config-reference.js'; import { ktxLocalStateDbPath } from '../../context/project/local-state-db.js'; import type { KtxLocalProject } from '../../context/project/project.js'; @@ -42,6 +43,7 @@ import { pullConfigFromMetricflowIntegration } from './adapters/metricflow/pull- import { LocalNotionRuntimeStore } from './adapters/notion/local-state-store.js'; import { NotionSourceAdapter } from './adapters/notion/notion.adapter.js'; import type { NotionFetchLogger } from './adapters/notion/fetch.js'; +import { SlackSourceAdapter } from './adapters/slack/slack.adapter.js'; import { seedLocalMappingStateFromKtxYaml } from './local-mapping-reconcile.js'; import type { SourceAdapter } from './types.js'; @@ -123,6 +125,7 @@ export function createDefaultLocalIngestAdapters( }, ...(options.logger ? { logger: options.logger } : {}), }), + new SlackSourceAdapter(), ]; if (options.historicSql) { @@ -309,6 +312,11 @@ export async function localPullConfigForAdapter( lastSuccessfulCursor: await localNotionRuntimeStore(project).readCursor(connectionId), }; } + if (adapter.source === 'slack') { + return slackConnectionToPullConfig(parseSlackConnectionConfig(connection), { + env: options.looker?.env ?? process.env, + }); + } if (adapter.source === 'metricflow') { const metricflow = connection.metricflow; const metricflowConfig = diff --git a/packages/cli/src/context/project/driver-schemas.test.ts b/packages/cli/src/context/project/driver-schemas.test.ts index 898625461..cdb437d7c 100644 --- a/packages/cli/src/context/project/driver-schemas.test.ts +++ b/packages/cli/src/context/project/driver-schemas.test.ts @@ -115,6 +115,31 @@ describe('connectionConfigSchema - notion / dbt / metricflow', () => { ).toThrow(); }); + it('parses a slack connection with an allowlisted channel', () => { + const parsed = connectionConfigSchema.parse({ + driver: 'slack', + bot_token_ref: 'env:SLACK_BOT_TOKEN', + channel_ids: ['C0123456789'], + max_messages_per_channel: 200, + }); + expect(parsed).toMatchObject({ + driver: 'slack', + bot_token_ref: 'env:SLACK_BOT_TOKEN', + channel_ids: ['C0123456789'], + max_messages_per_channel: 200, + }); + }); + + it('rejects slack without channel allowlist', () => { + expect(() => + connectionConfigSchema.parse({ + driver: 'slack', + bot_token_ref: 'env:SLACK_BOT_TOKEN', + channel_ids: [], + }), + ).toThrow(); + }); + it('parses a dbt connection from a local source_dir', () => { const parsed = connectionConfigSchema.parse({ driver: 'dbt', diff --git a/packages/cli/src/context/project/driver-schemas.ts b/packages/cli/src/context/project/driver-schemas.ts index a3b71bff1..ab08c9bc7 100644 --- a/packages/cli/src/context/project/driver-schemas.ts +++ b/packages/cli/src/context/project/driver-schemas.ts @@ -170,6 +170,29 @@ const notionConnectionSchema = z }) .describe('Notion context-source connection.'); +const slackConnectionSchema = z + .looseObject({ + driver: z.literal('slack'), + bot_token: z.string().min(1).optional().describe('Literal Slack bot token. Prefer bot_token_ref.'), + bot_token_ref: z + .string() + .min(1) + .optional() + .describe('Reference to Slack bot token (e.g. env:SLACK_BOT_TOKEN).'), + channel_ids: z + .array(z.string().min(1)) + .min(1) + .describe('Allowlisted Slack channel IDs to inspect for messages.'), + max_messages_per_channel: z + .number() + .int() + .min(1) + .max(10000) + .optional() + .describe('Maximum messages to fetch per configured channel in one ingest run.'), + }) + .describe('Slack context-source connection.'); + const dbtConnectionSchema = z .looseObject({ driver: z.literal('dbt'), @@ -204,6 +227,7 @@ export const connectionConfigSchema = z.discriminatedUnion('driver', [ lookerConnectionSchema, lookmlConnectionSchema, notionConnectionSchema, + slackConnectionSchema, dbtConnectionSchema, metricflowConnectionSchema, ]);