Integrating ClickHouse with Slack

Sync Slack into ClickHouse to query conversation metadata, users, messages, and thread replies alongside product or business data.

The app registry installs three raw tables, full scans for users and conversations, and checkpointed message history and threads as editable TypeScript source. The integration preserves the complete objects returned by Slack; add SQL views for the fields and relationships needed by the project.

Property	Included behavior
Registry app	slack (v0.2.0)
Install command	bunx chkit add slack
Source directory	src/integrations/slack
Authentication	Slack OAuth bearer token (user token recommended)
Environment variables	SLACK_API_TOKEN
ClickHouse	>=25.3.0
Coverage	3 synced resources, 0 derived views
Sync strategy	Full scans, Cursor checkpoints

Install the Slack integration

Run these commands in a TypeScript project with a package.json. Use a chkit release that includes the registry and ingestion plugin; these examples use the beta release.

bun add -d chkit@beta
bunx chkit registry inspect slack
bunx chkit add slack --dry-run
bunx chkit add slack

The installer copies source files, installs compatible dependencies, registers the ingestion plugin, and connects the exported schemas and pipeline to the project config. See installation behavior for config wiring and conflicts. Subsequent runs execute the installed code without contacting the registry.

Configure the ClickHouse connection

For a new project, the generated clickhouse.config.ts has this shape:

import { defineConfig } from '@chkit/core'
import { ingest } from '@chkit/plugin-ingest'

export default defineConfig({
  entry: './src/integrations/slack/index.ts',
  plugins: [ingest()],
  clickhouse: {
    url: process.env.CLICKHOUSE_URL ?? 'http://localhost:8123',
    username: process.env.CLICKHOUSE_USER ?? 'default',
    password: process.env.CLICKHOUSE_PASSWORD ?? '',
    database: process.env.CLICKHOUSE_DB ?? 'default',
  },
})

Set the connection variables for the intended destination. Preserve existing schema entries and plugins when adapting an existing project. Ingestion requires a direct ClickHouse connection and ClickHouse 25.3 or newer for native JSON.

The copied src/integrations/slack/config.ts separately sets the database containing the Slack tables. Set its database too; changing CLICKHOUSE_DB alone does not rename these schema objects.

Create a Slack access token

Use a Slack app installed in the intended workspace. A user OAuth token is recommended for reading public-channel history and threads. Create the app through Slack app settings, then configure its scopes and installation:

1. Create a Slack app for the workspace at api.slack.com/apps. Open OAuth & Permissions.
2. Add channels:read, channels:history, and users:read under User Token Scopes for the default three streams. Add users:read.email only when emails are needed.
3. For private channels add groups:read and groups:history; for DMs add im:read and im:history; for group DMs add mpim:read and mpim:history. Enable the matching conversationTypes in config.ts.
4. Install or reinstall the app to the workspace and copy the User OAuth Token into SLACK_API_TOKEN in the project .env or scheduler secret environment. A bot token only reads conversations it can access; enabled thread replies must also be permitted.
5. Keep tokens outside version control. Supply token refresh separately when using expiring tokens; the integration does not manage OAuth issuance or refresh.

Slack credential setup

The default public-channel integration requires the scopes below. No write scopes are needed.

Resource	Required scopes
Conversations	channels:read
Users	users:read
Messages and thread replies	channels:read, channels:history

Add scopes for each additional conversation type selected in config.ts:

Conversation type	Discovery scope	Message and reply scope
public_channel	channels:read	channels:history
private_channel	groups:read	groups:history
im	im:read	im:history
mpim	mpim:read	mpim:history

The users stream needs users:read. Add users:read.email when email addresses should appear in returned profiles; missing email fields are retained as missing. See Slack scopes and users.list.

Token type and conversation access

User tokens with the appropriate history scope read public channels and private conversations the user can access. Bot history access requires the bot to belong to the conversation. A bot may list public channels it has not joined, so the default channels: undefined can fail on history access. For a bot token, select IDs of conversations the bot has joined. See conversations.history.

Thread reads also require the appropriate history scope and conversation access; see conversations.replies. A user token is recommended for broad history and thread access. If an enabled thread request fails for the supplied token, the messages stream fails visibly. Fix the token’s access or explicitly set includeReplies: false to ingest history without fetching threads.

Set the access token in the project’s .env, replacing the placeholder:

SLACK_API_TOKEN=replace-with-the-slack-access-token

Bun loads the project’s .env. A scheduler must supply SLACK_API_TOKEN and the ClickHouse connection variables through its own environment or secret configuration. .env.example lists variable names and does not load credentials. Keep real tokens out of source control.

The reader obtains the token when a request runs and sends it as an Authorization: Bearer header. Schema imports, migration generation, and ingest list need no Slack token. This integration does not issue or refresh OAuth tokens; manage that lifecycle separately using Slack authentication.

Choose conversations and destination names

The installed config.ts starts with:

export const slackConfig: SlackConfig = {
  sourceId: 'slack.primary',
  database: 'default',
  tablePrefix: 'slack',
  channels: undefined,
  conversationTypes: ['public_channel'],
  includeReplies: true,
  pageSize: 200,
  messagePageSize: 15,
  requestIntervalMs: 3000,
  messageIntervalMs: 60000,
  historyFrom: '0.000000',
  overlapMs: 86400000,
  reconcileIntervalMs: 604800000,
}

Setting	Effect
sourceId	Stable, non-secret identity used in stream and row IDs
database	Database containing the three raw tables
tablePrefix	Prefix before _channels_raw, _users_raw, and _messages_raw
channels	Conversation IDs; undefined selects all discovered conversations; [] selects none
conversationTypes	Types passed to discovery; default ['public_channel']; also accepts private_channel, im, and mpim
includeReplies	Fetch complete threads for discovered roots with replies; default true
pageSize	Channel and user page size, integer 1–999; default 200
messagePageSize	History and reply page size, integer 1–999 subject to Slack’s app-specific limit; conservative default 15
requestIntervalMs	Delay before each authentication, conversation, and user request; default 3000 ms
messageIntervalMs	Delay before each history and reply request; default 60000 ms
historyFrom	First message timestamp for bootstrap/reconciliation; exact string, default 0.000000
overlapMs	Recent root overlap after the completed watermark; default one day
reconcileIntervalMs	Interval between full historical reads; default seven days; zero means every completed run

Set channels: ['C012EXAMPLE', 'C034EXAMPLE'] to restrict conversation metadata and messages. Selection uses Slack IDs, not channel names. Unknown IDs, inaccessible conversations, or IDs excluded by conversationTypes fail visibly. User collection remains workspace-wide.

Discovery includes archived conversations returned by Slack. Selecting private channels or direct messages changes the requested coverage and requires matching scopes and token access. Keep sourceId stable after ingestion begins. For another workspace, use a distinct source identity and a separate table prefix or database; copying the directory alone does not isolate pipeline and destination names.

Create tables and run the first sync

bunx chkit check
bunx chkit ingest list --tag provider:slack
bunx chkit generate --name add-slack
bunx chkit migrate

Review the generated SQL, apply it, and run ingestion:

bunx chkit migrate --apply
bunx chkit ingest run --tag provider:slack
bunx chkit query "SELECT count() FROM default.slack_messages_raw FINAL"

Queries in this guide use the default database and table prefix. Substitute configured names when they differ.

Included resources

Each resource has its own stream and raw table. The data field in every row retains the complete returned provider object, including fields not listed here and fields introduced by Slack later.

Resource	Default ClickHouse table	Records synced	API reference
Conversations (channels)	slack_channels_raw	Selected accessible conversations, including archived channels, with complete returned topics, purpose, membership flags, and shared-channel metadata. Public channels by default; private channels, DMs, and group DMs are opt-in.	GET/conversations.list
Users (users)	slack_users_raw	Complete returned workspace member and bot profiles, IDs, names, status, timezone, and deactivation flags. Email requires the optional users:read.email scope.	GET/users.list
Messages and thread replies (messages)	slack_messages_raw	Raw messages and replies selected by checkpointed timestamp windows, with retained thread work and periodic full historical reconciliation. Slack timestamps remain exact strings.	GET/conversations.history, GET/conversations.replies

The channels stream retains the conversation objects returned by conversations.list, including names, topics, purposes, flags, and other available metadata. These are Slack’s list responses; the reader does not expand them with conversations.info or fetch membership lists.

The users stream preserves returned profiles, custom fields, bots, and deleted-user flags. Channel selection does not filter users. The messages stream retains message text, timestamps, subtype, author references, thread fields, blocks, attachments, reactions, and file references when present. A file reference does not download the file. No separate reactions, files, pins, canvases, or search stream is included.

Requests and pagination

Every API URL is relative to https://slack.com/api/. Each reader calls POST auth.test to obtain the authenticated workspace’s team_id; the data methods use GET requests.

Resource	Requests and selection
channels	conversations.list with configured types, archived conversations included, then configured ID selection
users	users.list across the accessible workspace
messages	Independent conversation discovery, then conversations.history for each selected conversation; conversations.replies for roots with reply_count > 0 when replies are enabled

Metadata follows response_metadata.next_cursor until absent, null, or empty. Messages use exact oldest and latest time boundaries, preserving progress independently of temporary cursors. Empty message pages may use a live cursor; expiration restarts the unfinished interval. Repeated cursors and responses without a safe continuation fail visibly. Each stream discovers its workspace and parents independently.

For each channel, history advances newest first. Each loaded page stores its discovered thread work, then completes that bounded queue before requesting older history. Replies advance oldest first with an independent timestamp boundary. Threads retain the parent returned by Slack under its history identity. No history cursor waits during long thread reads, and no temporary cursor enters durable state.

How the sync works

Behavior	Details
Sync	Channels and users are full reads. Messages bootstrap retained history, then read overlapping time ranges. Each page checkpoints its timestamp frontier and pending threads after writes. Weekly full reconciliation revisits old roots and edits; temporary Slack cursors are never persisted.
Schedule	Run chkit ingest run through cron, CI, or an existing scheduler, with one process per project/target. Interrupted message windows resume their fixed bounds and pending threads; users/channels restart full reads. Conservative message pacing remains 15 items and one request per minute.
Deletions	Later observations replace stable row IDs, but deleted, inaccessible, or deselected records remain. The integration does not consume Slack events, propagate deletions, or maintain a complete change log.

Channels and users remain full metadata reads. Messages bootstrap from historyFrom to a fixed cutoff, then select overlapping ranges from their completed watermark. A full historical reconciliation every seven days revisits old edits, newly threaded roots, and late replies outside recent overlap. New channels bootstrap retained history. Coverage depends on retention, API visibility, scopes, and access.

Message checkpoints retain workspace/query scope, fixed bounds, channel position, exact history/reply frontiers, and a page-sized pending thread queue. State commits after its covered rows load. Failed or budget-limited runs resume those bounds and finish retained threads before advancing. The completed watermark moves only after all selected channels and child work finish.

Keep source identity, selection, reply policy, and time configuration stable. Changing them fails against saved state instead of reinterpreting progress. Earlier templates had no message checkpoint: upgrading to version 0.2.0 bootstraps once while preserving existing raw tables and identities.

Rate limits and scan duration

The default pipeline runs one stream, one fetch, and one load at a time. Metadata batches use a 500-row threshold; messages flush each source page so progress commits promptly. Requests wait before each attempt: metadata uses requestIntervalMs: 3000; history and replies use messageIntervalMs: 60000 with pages of 15 messages.

Slack documents a one-request-per-minute limit and a 15-object page limit for new commercially distributed applications outside the Slack Marketplace. Marketplace and internal customer-built apps retain Tier 3 limits. See the current conversations.history reference, conversations.replies reference, and rate-limit clarification.

For an internal app entitled to Tier 3, edit the copied config to use messagePageSize: 200 and messageIntervalMs: 1200. Confirm the app’s actual limits before changing these settings. Slack may still return fewer objects than requested or apply 429 responses.

A full history read with many threads can take hours or days at the conservative defaults. Restrict conversation IDs and select only the necessary streams before scheduling runs. The CLI’s default duration is 3600 seconds; set --max-duration for the expected scan. An interrupted message scan resumes its timestamp frontier and child work in the next run. Increasing duration allows more progress per execution.

Retries and failures

Requests have a 30-second timeout and run through the ingestion executor’s cancellation and retry context. The pipeline allows five retries after the initial attempt, with exponential backoff starting at one second and capped at 60 seconds. HTTP 429 responses honor Retry-After. Slack ratelimited and rate_limited responses also retry, with a 60-second fallback when no header is present. Network and server failures, including Slack internal_error, fatal_error, service_unavailable, and request_timeout, use the same retry budget.

Slack often reports errors in a successful HTTP response with ok: false. Authentication, permission, missing-scope, and unsupported-token errors fail with the Slack code and required scopes when supplied. Invalid payloads and identity fields also fail visibly. A failed request exhausts its budget before the stream fails; restarting the reader does not multiply its request retries.

Writes become visible batch by batch. A failed or interrupted run leaves already loaded rows in place. Repeated IDs reconcile in ClickHouse; this does not provide an atomic workspace snapshot or exactly-once delivery. A normal stream failure allows other selected streams to be attempted. ingest run exits with 0 only when all selected streams succeed.

Raw tables and identity

All three tables use these columns:

Column	ClickHouse type	Meaning
id	String	JSON-encoded composite source, resource, workspace, and entity identity
raw	JSON	Envelope with source_id, team_id, and the complete provider object under data; messages also include channel_id
_chkit_batch_id	String	Runtime batch identity
_chkit_run_id	String	Ingestion run identity
_chkit_ingested_at	DateTime64(6, 'UTC')	ClickHouse publication time, populated with now64(6)

Tables use ReplacingMergeTree(_chkit_ingested_at) with id as the primary and ordering key. Later ingestion time wins during replacement; source edits are not ordered by a Slack version field. Use FINAL to query the latest observed row while physical tables still contain earlier versions awaiting merges.

Resource	Composite identity
channels	[sourceId, 'channels', teamId, conversation.id]
users	[sourceId, 'users', teamId, user.id]
messages	[sourceId, 'messages', teamId, channelId, message.ts]

Slack message timestamps remain strings throughout ingestion. Channel ID is part of message identity because a timestamp alone does not identify a message across channels. Names, email addresses, and message text are never row keys.

The integration packages raw tables only. Create projections in the project’s schema when typed fields or views are needed; the retained payloads support new SQL projections without another API fetch.

Query messages and threads

Inspect latest observed messages:

WITH toJSONString(raw) AS payload
SELECT
  JSONExtractString(payload, 'team_id') AS team_id,
  JSONExtractString(payload, 'channel_id') AS channel_id,
  JSONExtractString(payload, 'data', 'ts') AS ts,
  JSONExtractString(payload, 'data', 'thread_ts') AS thread_ts,
  JSONExtractString(payload, 'data', 'user') AS user_id,
  JSONExtractString(payload, 'data', 'text') AS text,
  _chkit_ingested_at
FROM default.slack_messages_raw FINAL
LIMIT 20;

Preserve source_id and team_id when joining messages to users or conversations. Use channel_id and the string thread_ts to group replies with a root; a root without thread_ts uses its own ts as the thread identity.

Scheduling, verification, and recovery

Schedule a finite ingestion command through cron, CI, or a job runner. The scheduler supplies the project directory, credentials, cadence, execution budget, and concurrency control. Run one ingestion process per project and ClickHouse target at a time; the pipeline’s limits apply within a process.

For a narrower messages run:

bunx chkit ingest list --tag provider:slack --tag resource:messages
bunx chkit ingest run --tag provider:slack --tag resource:messages --max-duration 86400 --json
bunx chkit ingest status --tag provider:slack --json

Repeated tags use AND matching. With the default source ID, --tag stream:slack.primary.messages selects the same stream. status reports message checkpoints; full metadata reads have no incremental bookmark. Verify the execution’s stream outcomes and query known retained messages rather than relying only on a row count.

After the first run, compare a known message and a reply in an old thread with their retained raw.data. Add a reply to that old thread, then run a messages-only backfill with a new ID and --from before the root’s timestamp to confirm it is collected. A normal incremental rerun revisits old roots during the next scheduled full reconciliation.

After fixing credentials, permissions, or destination failures, rerun ingestion; messages resume saved work and metadata scans restart. Messages support isolated bounded backfills through --backfill <id> --from <date> --to <date>. Unfinished bounds must match the saved backfill; use another ID for a different range. Channels and users do not interpret date bounds. Historical observations loaded later can replace newer raw payloads because versioning uses ingestion time.

Coverage and consistency limits

Deleted messages, conversations removed from selection, and data made inaccessible remain in ClickHouse. Full reads update returned identities and do not clear the destination. Add deliberate reconciliation if an exact current snapshot is required; a failed or permission-limited scan does not establish that unseen messages were deleted.

Slack retention and access restrictions limit historical coverage. Changes during pagination may cause a scan to miss or repeat objects. Raw replacement tables retain the latest observed payload and do not provide a complete permanent edit history. There are no webhooks, live change capture, OAuth refresh, or file downloads in this integration.

Customize the installed integration

File	Responsibility
config.ts	Source identity, destination names, conversation selection, replies, page sizes, pacing
client.ts	Authentication, payload validation, pagination, request pacing, and row identity
state.ts	Validated message windows, exact timestamp arithmetic, and source binding
sources/channels.ts	Conversation reader and raw table
sources/users.ts	User reader and raw table
sources/messages.ts	History and thread reader and raw messages table
pipeline.ts	Enabled streams, tags, batching, retries, and concurrency
index.ts	Discoverable pipeline and schema exports
tests/slack.test.ts, tests/checkpoints.test.ts, tests/fixtures.ts	Portable payload, restart, boundary, and reconciliation tests

Remove a stream from pipeline.ts to stop collecting a resource. Keep its schema export to preserve the existing table under migration management. Removing a schema export can generate a drop operation; review that migration. Revisit pacing when changing concurrency or page sizes and run fixture tests for customized readers.

Install and run fixture tests

bunx chkit add slack --with-tests
bun test src/integrations/slack/tests/slack.test.ts src/integrations/slack/tests/checkpoints.test.ts

The optional tests mock Slack responses and use an in-memory destination. They need Bun, with no Slack token or ClickHouse server. Adjust the test path for a custom --path. Installed files belong to the project; reinstalling does not overwrite local edits or merge newer integration versions automatically.

Related pages

• Integration list: available apps and dedicated sync guides.
• chkit registry: inspect app coverage and installable files.
• chkit add: installation flags, config wiring, and ownership.
• chkit ingest: stream selection and execution budgets.
• Destinations and transformations: raw JSON, replacement semantics, and deletion handling.
• Scheduling and recovery: scheduling, retries, and outcomes.
• Test a source: fixture checks for customized readers.