"""Slack workspace connector.
Syncs workspace info, channels, and recent messages from the Slack Web API.
Uses cursor-based pagination and stores data in SlackChannel and ExternalSyncRecord.
"""
from __future__ import annotations
import logging
import time
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
import requests
from . import OAuthConnector, register_connector, _REGISTRY
logger = logging.getLogger(__name__)
# -- Registration metadata ---------------------------------------------------
register_connector(
"slack",
{
"service": "slack",
"name": "Slack",
"category": "messaging",
"description": "Workspace channels and recent messages via Slack Web API.",
"auth_type": "oauth2",
"auth_fields": ["access_token", "bot_token", "refresh_token"],
"capabilities": ["channels", "messages", "workspace_info"],
"rate_limit": "1 req/s per token (free), higher on paid plans",
"docs_url": "https://api.slack.com/methods",
},
)
class SlackConnector(OAuthConnector):
"""Slack workspace integration — pulls workspace info, channels, messages."""
_SERVICE = "slack"
_BASE_URL = "https://slack.com/api"
PAGE_LIMIT = 200
# Max messages to sync per channel per run (avoids huge pulls)
MAX_MESSAGES_PER_CHANNEL = 100
# -- OAuth 2.0 ------------------------------------------------------------
OAUTH_AUTHORIZE_URL = "https://slack.com/oauth/v2/authorize"
OAUTH_TOKEN_URL = "https://slack.com/api/oauth.v2.access"
OAUTH_SCOPES = [
"channels:read",
"chat:write",
"users:read",
]
@property
def _oauth_client_id(self) -> str:
from flask import current_app
return current_app.config.get("OAUTH_SLACK_CLIENT_ID", "")
@property
def _oauth_client_secret(self) -> str:
from flask import current_app
return current_app.config.get("OAUTH_SLACK_CLIENT_SECRET", "")
def oauth_authorize_url(self, state: str) -> str:
"""Build the Slack OAuth v2 authorization URL."""
from urllib.parse import urlencode
params = {
"client_id": self._oauth_client_id,
"redirect_uri": self._get_redirect_uri(),
"scope": ",".join(self.OAUTH_SCOPES),
"state": state,
}
return f"{self.OAUTH_AUTHORIZE_URL}?{urlencode(params)}"
def exchange_code_for_tokens(self, code: str, **kwargs) -> Dict[str, Any]:
"""Exchange authorization code for tokens via Slack OAuth v2."""
client_id = self._oauth_client_id
client_secret = self._oauth_client_secret
if not client_id or not client_secret:
raise ValueError(
"Slack OAuth credentials not configured. "
"Set OAUTH_SLACK_CLIENT_ID and OAUTH_SLACK_CLIENT_SECRET."
)
response = requests.post(
self.OAUTH_TOKEN_URL,
data={
"code": code,
"redirect_uri": self._get_redirect_uri(),
"client_id": client_id,
"client_secret": client_secret,
},
headers={"Content-Type": "application/x-www-form-urlencoded"},
timeout=15,
)
response.raise_for_status()
data = response.json()
if not data.get("ok", True):
raise ValueError(f"Slack OAuth error: {data.get('error', 'unknown')}")
return {
"access_token": data.get("access_token", ""),
"bot_token": data.get("bot_access_token", ""),
"refresh_token": data.get("refresh_token", ""),
"scope": data.get("scope", ""),
"team_id": data.get("team", {}).get("id", "") if isinstance(data.get("team"), dict) else data.get("team_id", ""),
"bot_user_id": data.get("bot_user_id", ""),
"expires_in": data.get("expires_in", ""),
"token_type": data.get("token_type", "bearer"),
}
def refresh_access_token(self, refresh_token: str) -> Dict[str, str]:
"""Refresh expired access token via Slack OAuth v2."""
client_id = self._oauth_client_id
client_secret = self._oauth_client_secret
if not client_id or not client_secret:
raise ValueError("Slack OAuth credentials not configured.")
response = requests.post(
self.OAUTH_TOKEN_URL,
data={
"grant_type": "refresh_token",
"refresh_token": refresh_token,
"client_id": client_id,
"client_secret": client_secret,
},
headers={"Content-Type": "application/x-www-form-urlencoded"},
timeout=15,
)
response.raise_for_status()
data = response.json()
if not data.get("ok", True):
raise ValueError(f"Slack token refresh error: {data.get('error', 'unknown')}")
return {
"access_token": data.get("access_token", ""),
"bot_token": data.get("bot_access_token", ""),
"refresh_token": data.get("refresh_token", ""),
"scope": data.get("scope", ""),
"team_id": data.get("team", {}).get("id", "") if isinstance(data.get("team"), dict) else data.get("team_id", ""),
}
# -- connect / disconnect --------------------------------------------------
def connect(self) -> Dict[str, Any]:
"""Validate the Slack Bot Token via GET /auth.test."""
self._log(event_type="connect_attempt", status="pending")
start = time.monotonic()
try:
bot_token = self.config.get("bot_token", "")
if not bot_token:
raise ValueError("Slack bot_token is required")
response = self._retry(self._api_get, "auth.test")
self._connected = True
duration_ms = int((time.monotonic() - start) * 1000)
self._log(
event_type="connect_success",
status="success",
duration_ms=duration_ms,
details={
"user_id": response.get("user_id"),
"team_id": response.get("team", {}).get("id"),
"team_name": response.get("team", {}).get("name"),
"bot_id": response.get("bot_id"),
},
)
return {
"status": "connected",
"service": "slack",
"team_id": response.get("team", {}).get("id"),
"team_name": response.get("team", {}).get("name"),
"user_id": response.get("user_id"),
"bot_id": response.get("bot_id"),
"duration_ms": duration_ms,
}
except Exception as exc:
duration_ms = int((time.monotonic() - start) * 1000)
self._log(
event_type="connect_error",
status="error",
duration_ms=duration_ms,
error_message=str(exc),
)
return {"status": "error", "error": str(exc)}
def disconnect(self) -> Dict[str, Any]:
"""Clear connection state and credentials."""
self._connected = False
# Wipe the stored bot token from config so it's not persisted
if "bot_token" in self.config:
self.config["bot_token"] = ""
self._log(event_type="disconnect", status="success")
return {"status": "disconnected", "service": "slack"}
# -- sync ------------------------------------------------------------------
def sync(self) -> Dict[str, Any]:
"""Pull workspace info, channels, and recent messages from Slack."""
self._log(event_type="sync_start", status="pending")
start = time.monotonic()
total_records = 0
try:
results: Dict[str, Any] = {}
# 1. Fetch workspace info
workspace_info = self._fetch_workspace_info()
results["workspace"] = workspace_info
# 2. Sync channels (public and private the bot can access)
channels = self._sync_channels()
total_records += channels
results["channels"] = channels
# 3. Sync recent messages for each channel
messages = self._sync_messages()
total_records += messages
results["messages"] = messages
duration_ms = int((time.monotonic() - start) * 1000)
self._log(
event_type="sync_complete",
status="success",
record_count=total_records,
duration_ms=duration_ms,
details=results,
)
return {
"status": "success",
"record_count": total_records,
"duration_ms": duration_ms,
"details": results,
"last_sync_at": datetime.now(timezone.utc).isoformat(),
}
except Exception as exc:
duration_ms = int((time.monotonic() - start) * 1000)
self._log(
event_type="sync_error",
status="error",
record_count=total_records,
duration_ms=duration_ms,
error_message=str(exc),
)
return {
"status": "error",
"error": str(exc),
"record_count": total_records,
}
# -- Workspace info --------------------------------------------------------
def _fetch_workspace_info(self) -> Dict[str, Any]:
"""Get workspace/team info via auth.test (already validated)."""
try:
resp = self._retry(self._api_get, "auth.test")
team = resp.get("team", {})
return {
"team_id": team.get("id"),
"team_name": team.get("name"),
"url": team.get("url"),
"enterprise_id": resp.get("team", {}).get("enterprise", {}).get("id"),
"enterprise_name": resp.get("team", {}).get("enterprise", {}).get("name"),
}
except Exception as exc:
logger.warning("Slack workspace info fetch failed: %s", exc)
return {}
# -- Channel sync -----------------------------------------------------------
def _sync_channels(self) -> int:
"""List all conversations and merge into SlackChannel model."""
from app.models import db, SlackChannel
all_channels = self._fetch_all_conversations()
count = 0
for raw in all_channels:
channel_id = raw.get("id", "")
if not channel_id:
continue
created_ts = raw.get("created")
created_at_ts: Optional[datetime] = None
if created_ts:
try:
created_at_ts = datetime.fromtimestamp(
float(created_ts), tz=timezone.utc
)
except (ValueError, TypeError, OSError):
pass
count += self._merge_channel(
SlackChannel,
self.company_id,
channel_id,
{
"name": raw.get("name", ""),
"is_private": raw.get("is_private", False),
"is_archived": raw.get("is_archived", False),
"purpose": raw.get("purpose", {}).get("value", ""),
"topic": raw.get("topic", {}).get("value", ""),
"created_at_ts": created_at_ts,
"metadata_json": {
k: v for k, v in raw.items()
if k not in ("id", "name", "is_private", "is_archived",
"purpose", "topic", "created")
},
},
)
db.session.commit()
logger.info("Slack channels synced: %d records merged", count)
return count
@staticmethod
def _merge_channel(model, company_id: str, external_id: str,
defaults: Dict[str, Any]) -> int:
"""Find or create a SlackChannel by (company_id, external_id) and update."""
from app.models import db
record = model.query.filter_by(
company_id=company_id,
external_id=external_id,
).first()
if record:
for key, value in defaults.items():
setattr(record, key, value)
else:
record = model(
company_id=company_id,
external_id=external_id,
**defaults,
)
db.session.add(record)
return 1
# -- Message sync ----------------------------------------------------------
def _sync_messages(self) -> int:
"""Pull recent messages from each channel into ExternalSyncRecord."""
from app.models import db, ExternalSyncRecord
all_channels = self._fetch_all_conversations()
count = 0
for raw in all_channels:
channel_id = raw.get("id", "")
channel_name = raw.get("name", "unknown")
if not channel_id:
continue
# Skip archived channels
if raw.get("is_archived", False):
continue
try:
messages = self._fetch_messages(channel_id, channel_name)
for msg in messages[:self.MAX_MESSAGES_PER_CHANNEL]:
ts = msg.get("ts", "")
if not ts:
continue
# Build a deterministic external_id
ext_id = f"channel:{channel_id}:msg:{ts}"
# Parse message timestamp
msg_ts: Optional[datetime] = None
try:
msg_ts = datetime.fromtimestamp(
float(ts), tz=timezone.utc
)
except (ValueError, TypeError, OSError):
pass
# Skip bot messages from ourselves to avoid echo
bot_id = msg.get("bot_id")
bot_profile = msg.get("bot_profile", {})
count += self._merge(
ExternalSyncRecord,
self.company_id,
ext_id,
{
"source_service": "slack",
"entity_type": "message",
"name": channel_name,
"status": "synced",
"properties_json": {
"channel_id": channel_id,
"channel_name": channel_name,
"user_id": msg.get("user", ""),
"user": msg.get("user", ""),
"text": msg.get("text", ""),
"ts": ts,
"thread_ts": msg.get("thread_ts", ""),
"reactions": msg.get("reactions", []),
"bot_id": bot_id,
"bot_name": bot_profile.get("name", ""),
"type": msg.get("type", ""),
"subtype": msg.get("subtype", ""),
"blocks": msg.get("blocks", []),
},
},
)
except Exception as exc:
logger.warning(
"Slack messages fetch failed for #%s: %s", channel_name, exc
)
continue
db.session.commit()
logger.info("Slack messages synced: %d records merged", count)
return count
@staticmethod
def _merge(model, company_id: str, external_id: str,
defaults: Dict[str, Any]) -> int:
"""Find or create a record by (company_id, external_id) and update."""
from app.models import db
record = model.query.filter_by(
company_id=company_id,
external_id=external_id,
).first()
if record:
for key, value in defaults.items():
setattr(record, key, value)
else:
record = model(
company_id=company_id,
external_id=external_id,
**defaults,
)
db.session.add(record)
return 1
# -- Paginated conversation list (cursor-based) ----------------------------
def _fetch_all_conversations(self) -> List[Dict[str, Any]]:
"""Paginate through conversations.list with cursor-based pagination."""
all_channels: List[Dict[str, Any]] = []
cursor: Optional[str] = None
while True:
self._check_rate_limit()
data = self._retry(
self._api_get,
"conversations.list",
limit=self.PAGE_LIMIT,
cursor=cursor,
types="public_channel,private_channel",
)
channels = data.get("channels", [])
if not channels:
break
all_channels.extend(channels)
logger.info(
"Slack conversations: fetched %d (total %d)",
len(channels), len(all_channels),
)
response_meta = data.get("response_metadata", {})
cursor = response_meta.get("next_cursor")
if not cursor:
break
# Small pause between pages to respect rate limits
time.sleep(0.15)
logger.info("Slack conversations: total %d channels", len(all_channels))
return all_channels
# -- Paginated message history (cursor-based) ------------------------------
def _fetch_messages(self, channel_id: str, channel_name: str) -> List[Dict[str, Any]]:
"""Paginate through conversations.history for a channel (oldest first)."""
all_messages: List[Dict[str, Any]] = []
oldest_cursor: Optional[str] = None
while True:
self._check_rate_limit()
data = self._retry(
self._api_get,
"conversations.history",
channel=channel_id,
limit=min(self.PAGE_LIMIT, self.MAX_MESSAGES_PER_CHANNEL),
oldest_cursor=oldest_cursor,
)
messages = data.get("messages", [])
if not messages:
break
all_messages.extend(messages)
logger.info(
"Slack #s messages: fetched %d (total %d)",
channel_name, len(messages), len(all_messages),
)
if len(all_messages) >= self.MAX_MESSAGES_PER_CHANNEL:
all_messages = all_messages[:self.MAX_MESSAGES_PER_CHANNEL]
break
response_meta = data.get("response_metadata", {})
oldest_cursor = response_meta.get("next_cursor")
if not oldest_cursor:
break
time.sleep(0.15)
return all_messages
# -- status ----------------------------------------------------------------
def status(self) -> Dict[str, Any]:
"""Check the health of the Slack connection via auth.test."""
if not self._connected:
return {
"service": "slack",
"connected": False,
"config_present": bool(self.config.get("bot_token")),
}
try:
response = self._retry(self._api_get, "auth.test")
team = response.get("team", {})
return {
"service": "slack",
"connected": True,
"team_id": team.get("id"),
"team_name": team.get("name"),
"user_id": response.get("user_id"),
"bot_id": response.get("bot_id"),
"last_sync_at": self.config.get("last_sync_at"),
}
except Exception as exc:
return {
"service": "slack",
"connected": False,
"error": str(exc),
}
# -- API helper ------------------------------------------------------------
def _api_get(self, method: str, **params) -> Dict[str, Any]:
"""Make a GET request to the Slack Web API with Bearer token auth."""
bot_token = self.config.get("bot_token", "")
if not bot_token:
raise ValueError("Slack bot_token not configured")
url = f"{self._BASE_URL}/{method}"
headers = {"Authorization": f"Bearer {bot_token}"}
# Filter out None/empty cursors — Slack rejects them
clean_params = {k: v for k, v in params.items() if v}
resp = requests.get(url, headers=headers, params=clean_params, timeout=30)
# Handle rate limiting
if resp.status_code == 429:
retry_after = resp.headers.get("Retry-After")
logger.warning(
"Slack rate limited: 429, Retry-After=%s", retry_after
)
if retry_after:
try:
time.sleep(int(retry_after))
except ValueError:
time.sleep(2)
# Don't raise — let the caller retry via _retry()
raise requests.HTTPError("Slack API rate limited (429)", response=resp)
resp.raise_for_status()
data = resp.json()
# Slack returns ok=false for API-level errors (e.g., invalid_auth)
if not data.get("ok", True):
error = data.get("error", "unknown_slack_error")
logger.error("Slack API error (%s): %s", method, error)
raise ValueError(f"Slack API error: {error}")
return data
# -- Register in the framework registry ---------------------------------------
_REGISTRY["slack"] = SlackConnector