import { getCredentialContext } from "@agent-native/core/server";
import { accessFilter, assertAccess } from "@agent-native/core/sharing";
import { and, desc, eq, inArray, isNull, lte, or } from "drizzle-orm";

import type {
  BrainCaptureKind,
  BrainSourceProvider,
} from "../../shared/types.js";
import { getDb, schema } from "../db/index.js";
import { listAccessibleAudienceIds } from "./audiences.js";
import {
  BrainCaptureBlockedError,
  createCapture,
  nanoid,
  nowIso,
  parseJson,
  readBrainSettings,
  serializeCapture,
  stableJson,
} from "./brain.js";
import { resolveMeetingMemberEmails } from "./meeting-audience.js";
import {
  ensureSlackPublicChannelMembership,
  type SlackChannelJoinResponse,
} from "./slack-channel-membership.js";
import {
  inspectSourceCredentialAvailability,
  resolveSourceCredential,
} from "./source-credentials.js";

export interface ConnectorSyncResult {
  runId: string;
  sourceId: string;
  provider: BrainSourceProvider;
  status: "success" | "error";
  capturesCreated: number;
  captures: Array<ReturnType<typeof serializeCapture>>;
  message: string;
  stats?: Record<string, unknown>;
}

async function createConnectorCapture(
  values: Parameters<typeof createCapture>[0],
) {
  try {
    return { capture: await createCapture(values), blocked: false };
  } catch (error) {
    if (error instanceof BrainCaptureBlockedError) {
      return { capture: null, blocked: true };
    }
    throw error;
  }
}

export interface SlackPilotOptions {
  readHistory?: boolean;
  channelRefs?: string[];
  resolveNames?: boolean;
  historyLimit?: number;
  maxChannels?: number;
  permalinkLimit?: number;
  recentDays?: number;
  oldest?: string;
}

export interface SlackPilotReport {
  sourceId: string;
  sourceTitle: string;
  ok: boolean;
  status: "validated" | "blocked" | "synced" | "error";
  historyRead: boolean;
  credential: {
    ok: boolean;
    team?: string | null;
    teamId?: string | null;
    workspaceUrl?: string | null;
    botUser?: string | null;
    error?: string | null;
  };
  guardrails: {
    historyReadRequested: boolean;
    maxChannels: number;
    historyLimit: number;
    pagesPerChannel: number;
    permalinkLimit: number;
    autoSync: false;
    oldest?: string;
  };
  channelValidation: {
    requested: number;
    checked: number;
    ok: number;
    excluded: number;
    missing: number;
    skipped: number;
    channels: Array<{
      ref: string;
      id?: string;
      name?: string;
      status: "ok" | "excluded" | "missing" | "skipped";
      message: string;
      directExcluded?: boolean;
      archived?: boolean;
      privateChannel?: boolean;
    }>;
  };
  sync?: {
    runId: string;
    status: "success" | "error";
    message: string;
    stats?: Record<string, unknown>;
  };
  capturesCreated: number;
  captures: Array<{
    id: string;
    title: string;
    capturedAt: string;
    sourceUrl?: string | null;
  }>;
  proposals: {
    total: number;
    pending: number;
    recent: Array<{
      id: string;
      title: string;
      status: string;
      createdAt: string;
    }>;
  };
  currentKnowledge: {
    total: number;
    published: number;
    draft: number;
    redacted: number;
    archived: number;
    recent: Array<{
      id: string;
      title: string;
      status: string;
      updatedAt: string;
    }>;
  };
  privacyExclusions: string[];
  nextSteps: string[];
}

type SourceRow = typeof schema.brainSources.$inferSelect;

interface Connector {
  sync(source: SourceRow): Promise<ConnectorSyncResult>;
}

interface RetryCursor {
  provider: BrainSourceProvider;
  endpoint: string;
  retryAfterSeconds: number;
  retryAfterAt: string;
  encounteredAt: string;
}

interface SlackChannelCursor {
  latestTs?: string;
  pageCursor?: string;
  pendingLatestTs?: string;
}

interface SlackSyncCursor {
  channels?: Record<string, SlackChannelCursor>;
  publicChannelOffset?: number;
  retry?: RetryCursor;
  lastRunAt?: string;
}

interface GranolaSyncCursor {
  cursor?: string | null;
  updatedAfter?: string;
  retry?: RetryCursor;
  lastRunAt?: string;
}

interface SlackChannel {
  id: string;
  name?: string;
  is_im?: boolean;
  is_mpim?: boolean;
  is_channel?: boolean;
  is_group?: boolean;
  is_archived?: boolean;
  is_private?: boolean;
  is_member?: boolean;
}

interface SlackMessage {
  type?: string;
  subtype?: string;
  user?: string;
  username?: string;
  bot_id?: string;
  text?: string;
  ts?: string;
  thread_ts?: string;
  reactions?: Array<{ name?: string; count?: number }>;
}

interface SlackHistoryResponse {
  messages?: SlackMessage[];
  has_more?: boolean;
  response_metadata?: { next_cursor?: string };
}

interface SlackRepliesResponse {
  messages?: SlackMessage[];
}

interface SlackInfoResponse {
  channel?: SlackChannel;
}

interface SlackListResponse {
  channels?: SlackChannel[];
  response_metadata?: { next_cursor?: string };
}

interface SlackMembersResponse {
  members?: string[];
  response_metadata?: { next_cursor?: string };
}

interface SlackUserInfoResponse {
  user?: { profile?: { email?: string } };
}

type SlackUserEmailCache = Map<string, string | null>;

const SLACK_USER_LOOKUP_CONCURRENCY = 4;
const SLACK_THREAD_CAPTURE_CONCURRENCY = 4;
const GRANOLA_NOTE_CAPTURE_CONCURRENCY = 4;
const CONNECTOR_SYNC_LEASE_MS = 10 * 60 * 1_000;
const CONNECTOR_SYNC_HEARTBEAT_MS = Math.floor(CONNECTOR_SYNC_LEASE_MS / 3);

interface SlackPermalinkResponse {
  permalink?: string;
}

interface SlackAuthTestResponse {
  url?: string;
  team?: string;
  user?: string;
  team_id?: string;
  user_id?: string;
  bot_id?: string;
}

interface GranolaListNote {
  id?: string;
  title?: string;
  created_at?: string;
  updated_at?: string;
  web_url?: string;
  owner?: unknown;
}

interface GranolaListResponse {
  notes?: GranolaListNote[];
  hasMore?: boolean;
  has_more?: boolean;
  cursor?: string | null;
}

interface GranolaNote extends GranolaListNote {
  summary_text?: string;
  summary_markdown?: string;
  summary?: unknown;
  transcript?: unknown;
  attendees?: unknown;
  calendar_event?: unknown;
  folder_membership?: unknown;
}

interface GitHubRepoCursor {
  updatedAfter?: string;
}

interface GitHubSyncCursor {
  repositories?: Record<string, GitHubRepoCursor>;
  retry?: RetryCursor;
  lastRunAt?: string;
}

interface GitHubLabel {
  name?: string;
  color?: string;
  description?: string | null;
}

interface GitHubLinkedRef {
  repo: string;
  number: number;
  type?: "issue" | "pull";
  url: string;
  linkedFrom?: {
    sourceId: string;
    captureId: string;
    captureTitle: string;
    sourceUrl?: string | null;
  };
}

interface GitHubIssueOrPull {
  id?: number;
  number?: number;
  title?: string;
  body?: string | null;
  html_url?: string;
  state?: string;
  locked?: boolean;
  created_at?: string;
  updated_at?: string;
  closed_at?: string | null;
  user?: {
    login?: string;
    html_url?: string;
  } | null;
  labels?: Array<string | GitHubLabel>;
  pull_request?: {
    html_url?: string;
    merged_at?: string | null;
    diff_url?: string;
    patch_url?: string;
    url?: string;
  };
}

interface GitHubPullDetails {
  html_url?: string;
  merged?: boolean;
  merged_at?: string | null;
  draft?: boolean;
  mergeable_state?: string | null;
  comments?: number;
  review_comments?: number;
  commits?: number;
  additions?: number;
  deletions?: number;
  changed_files?: number;
}

interface GitHubIssueComment {
  id?: number;
  body?: string | null;
  html_url?: string;
  created_at?: string;
  updated_at?: string;
  user?: {
    login?: string;
    html_url?: string;
  } | null;
}

interface GitHubPullReview {
  id?: number;
  state?: string;
  body?: string | null;
  html_url?: string;
  submitted_at?: string;
  user?: {
    login?: string;
    html_url?: string;
  } | null;
}

interface GitHubContext {
  pull?: GitHubPullDetails | null;
  comments: GitHubIssueComment[];
  reviews: GitHubPullReview[];
}

class ConnectorRateLimitError extends Error {
  constructor(
    public provider: BrainSourceProvider,
    public endpoint: string,
    public retryAfterSeconds: number,
  ) {
    super(
      `${provider} rate limited ${endpoint}; retry after ${retryAfterSeconds}s`,
    );
  }
}

function transcriptItems(config: Record<string, unknown>) {
  const raw = config.transcripts ?? config.sampleTranscripts ?? config.messages;
  if (!Array.isArray(raw)) return [];
  return raw
    .map((item) => {
      if (typeof item === "string") {
        return { title: "Imported capture", content: item };
      }
      if (!item || typeof item !== "object") return null;
      const record = item as Record<string, unknown>;
      const content =
        typeof record.content === "string"
          ? record.content
          : typeof record.text === "string"
            ? record.text
            : "";
      if (!content.trim()) return null;
      return {
        externalId:
          typeof record.externalId === "string" ? record.externalId : undefined,
        title:
          typeof record.title === "string" ? record.title : "Imported capture",
        content,
        kind:
          typeof record.kind === "string"
            ? (record.kind as BrainCaptureKind)
            : "transcript",
        capturedAt:
          typeof record.capturedAt === "string" ? record.capturedAt : undefined,
        metadata:
          typeof record.metadata === "object" && record.metadata
            ? (record.metadata as Record<string, unknown>)
            : {},
      };
    })
    .filter(Boolean) as Array<{
    title: string;
    content: string;
    kind: BrainCaptureKind;
    capturedAt?: string;
    metadata: Record<string, unknown>;
    externalId?: string;
  }>;
}

function objectValue(value: unknown): Record<string, unknown> {
  return value && typeof value === "object"
    ? (value as Record<string, unknown>)
    : {};
}

function configuredList(
  config: Record<string, unknown>,
  keys: string[],
  nestedKey?: string,
): string[] {
  const values: string[] = [];
  const configs = nestedKey
    ? [config, objectValue(config[nestedKey])]
    : [config];

  for (const itemConfig of configs) {
    for (const key of keys) {
      const raw = itemConfig[key];
      if (typeof raw === "string") {
        values.push(...raw.split(","));
      } else if (Array.isArray(raw)) {
        for (const item of raw) {
          if (typeof item === "string") values.push(item);
        }
      }
    }
  }

  return Array.from(
    new Set(
      values
        .map((value) => value.trim())
        .filter(Boolean)
        .map((value) => value.replace(/^#/, "")),
    ),
  );
}

export function slackChannelRefsFromConfig(config: Record<string, unknown>) {
  return configuredList(
    config,
    [
      "channelIds",
      "channels",
      "allowedChannels",
      "allowlistedChannels",
      "allowList",
    ],
    "slack",
  );
}

function configuredNumber(
  config: Record<string, unknown>,
  keys: string[],
  fallback: number,
  options: { min: number; max: number; nestedKey?: string },
): number {
  const configs = options.nestedKey
    ? [config, objectValue(config[options.nestedKey])]
    : [config];
  for (const itemConfig of configs) {
    for (const key of keys) {
      const raw = itemConfig[key];
      const value =
        typeof raw === "number"
          ? raw
          : typeof raw === "string"
            ? Number(raw)
            : Number.NaN;
      if (Number.isFinite(value)) {
        return Math.max(options.min, Math.min(options.max, Math.floor(value)));
      }
    }
  }
  return fallback;
}

function clampInteger(
  value: number | undefined,
  fallback: number,
  min: number,
  max: number,
): number {
  if (typeof value !== "number" || !Number.isFinite(value)) return fallback;
  return Math.max(min, Math.min(max, Math.floor(value)));
}

function daysAgoIso(days: number): string {
  return new Date(Date.now() - days * 24 * 60 * 60 * 1000).toISOString();
}

function removeSlackHistoryConfig(config: Record<string, unknown>) {
  const next = { ...config };
  for (const key of [
    "channelIds",
    "channels",
    "allowedChannels",
    "allowlistedChannels",
    "allowList",
    "transcripts",
    "sampleTranscripts",
    "messages",
    "fixture",
    "testFixture",
    "useConfiguredItems",
  ]) {
    delete next[key];
  }
  const nestedSlack = objectValue(next.slack);
  for (const key of [
    "channelIds",
    "channels",
    "allowedChannels",
    "allowlistedChannels",
    "allowList",
  ]) {
    delete nestedSlack[key];
  }
  next.slack = nestedSlack;
  return next;
}

function sourceUrlFromCapture(
  capture: ReturnType<typeof serializeCapture>,
): string | null {
  const raw = capture.metadata as Record<string, unknown>;
  return typeof raw.sourceUrl === "string"
    ? raw.sourceUrl
    : typeof raw.permalink === "string"
      ? raw.permalink
      : null;
}

async function summarizeSlackPilotSource(sourceId: string) {
  const db = getDb();
  const [knowledgeRows, proposalRows] = await Promise.all([
    db
      .select()
      .from(schema.brainKnowledge)
      .where(
        and(
          accessFilter(schema.brainKnowledge, schema.brainKnowledgeShares),
          eq(schema.brainKnowledge.sourceId, sourceId),
        ),
      ),
    db
      .select()
      .from(schema.brainProposals)
      .where(
        and(
          accessFilter(schema.brainProposals, schema.brainProposalShares),
          eq(schema.brainProposals.sourceId, sourceId),
        ),
      ),
  ]);

  const sortRecent = <T extends { updatedAt?: unknown; createdAt?: unknown }>(
    rows: T[],
  ) =>
    [...rows].sort((a, b) =>
      String(b.updatedAt ?? b.createdAt ?? "").localeCompare(
        String(a.updatedAt ?? a.createdAt ?? ""),
      ),
    );

  return {
    proposals: {
      total: proposalRows.length,
      pending: proposalRows.filter((row) => row.status === "pending").length,
      recent: sortRecent(proposalRows)
        .slice(0, 3)
        .map((row) => ({
          id: String(row.id),
          title: String(row.title),
          status: String(row.status),
          createdAt: String(row.createdAt),
        })),
    },
    currentKnowledge: {
      total: knowledgeRows.length,
      published: knowledgeRows.filter((row) => row.status === "published")
        .length,
      draft: knowledgeRows.filter((row) => row.status === "draft").length,
      redacted: knowledgeRows.filter((row) => row.status === "redacted").length,
      archived: knowledgeRows.filter((row) => row.status === "archived").length,
      recent: sortRecent(knowledgeRows)
        .slice(0, 3)
        .map((row) => ({
          id: String(row.id),
          title: String(row.title),
          status: String(row.status),
          updatedAt: String(row.updatedAt),
        })),
    },
  };
}

function isFixtureConfig(config: Record<string, unknown>) {
  return (
    config.fixture === true ||
    config.testFixture === true ||
    config.useConfiguredItems === true
  );
}

async function requireConnectorCredential(
  key: string,
  label: string,
  provider: BrainSourceProvider,
  config?: Record<string, unknown>,
): Promise<string> {
  const ctx = getCredentialContext();
  if (!ctx) {
    throw new Error(
      `${label} sync requires an authenticated credential context`,
    );
  }
  const workspaceConnectionId =
    typeof config?.workspaceConnectionId === "string"
      ? config.workspaceConnectionId
      : undefined;
  const value = await resolveSourceCredential({
    provider,
    key,
    ctx,
    workspaceConnectionId,
  });
  if (!value) {
    if (workspaceConnectionId?.trim()) {
      const availability = await inspectSourceCredentialAvailability({
        provider,
        key,
        ctx,
        workspaceConnectionId,
      });
      throw new Error(
        availability.missingMessage ??
          `${label} credential ${key} is not configured`,
      );
    }
    throw new Error(`${label} credential ${key} is not configured`);
  }
  return value;
}

function retryCursor(
  error: ConnectorRateLimitError,
  provider: BrainSourceProvider,
): RetryCursor {
  const now = Date.now();
  return {
    provider,
    endpoint: error.endpoint,
    retryAfterSeconds: error.retryAfterSeconds,
    retryAfterAt: new Date(now + error.retryAfterSeconds * 1000).toISOString(),
    encounteredAt: new Date(now).toISOString(),
  };
}

function retryAfterSeconds(headers: Headers): number {
  const raw = headers.get("retry-after");
  const parsed = raw ? Number(raw) : Number.NaN;
  return Number.isFinite(parsed) && parsed > 0 ? Math.ceil(parsed) : 60;
}

function buildUrl(
  base: string,
  params: Record<string, string | number | boolean | null | undefined>,
) {
  const url = new URL(base);
  for (const [key, value] of Object.entries(params)) {
    if (value === undefined || value === null || value === "") continue;
    url.searchParams.set(key, String(value));
  }
  return url;
}

function slackTsFromDateish(value: string | undefined): string | undefined {
  if (!value) return undefined;
  if (/^\d+(\.\d+)?$/.test(value)) return value;
  const ms = Date.parse(value);
  if (!Number.isFinite(ms)) return undefined;
  return (ms / 1000).toFixed(6);
}

function isoFromSlackTs(ts: string | undefined): string | undefined {
  if (!ts) return undefined;
  const seconds = Number(ts.split(".")[0]);
  if (!Number.isFinite(seconds)) return undefined;
  return new Date(seconds * 1000).toISOString();
}

function newestSlackTs(messages: SlackMessage[]): string | undefined {
  const timestamps = messages
    .map((message) => message.ts)
    .filter((ts): ts is string => typeof ts === "string" && ts.length > 0)
    .sort((a, b) => Number(b) - Number(a));
  return timestamps[0];
}

function readableJson(value: unknown): string {
  if (typeof value === "string") return value.trim();
  if (value == null) return "";
  if (typeof value !== "object") return String(value);
  const record = value as Record<string, unknown>;
  for (const key of ["markdown", "text", "content", "summary"]) {
    const candidate = record[key];
    if (typeof candidate === "string" && candidate.trim()) {
      return candidate.trim();
    }
  }
  return JSON.stringify(value, null, 2);
}

function slackDirectConversationFromRef(
  channelRef: string,
): SlackChannel | null {
  const normalized = channelRef.trim().replace(/^#/, "");
  if (!/^D[A-Z0-9]+$/i.test(normalized)) return null;
  return { id: normalized.toUpperCase(), is_im: true };
}

export function isSlackDirectConversation(channel: SlackChannel): boolean {
  return channel.is_im === true || channel.is_mpim === true;
}

function isUsableSlackChannel(channel: SlackChannel): boolean {
  return (
    !isSlackDirectConversation(channel) &&
    channel.is_archived !== true &&
    (channel.is_channel === true || channel.is_group === true)
  );
}

function slackReactionCount(message: SlackMessage) {
  return (message.reactions ?? []).reduce(
    (total, reaction) => total + Math.max(0, Number(reaction.count) || 0),
    0,
  );
}

function isSlackTextMessage(message: SlackMessage) {
  return (
    message.type === "message" &&
    typeof message.text === "string" &&
    message.text.trim().length > 0 &&
    typeof message.ts === "string"
  );
}

export interface SlackThreadCapture {
  externalId: string;
  title: string;
  content: string;
  capturedAt: string;
  metadata: Record<string, unknown>;
}

/**
 * Builds the only Slack representation Brain persists. It deliberately omits
 * Slack user ids, display names, and the raw Events/Web API payload.
 */
export function normalizeSlackThreadCapture(input: {
  channel: SlackChannel;
  messages: SlackMessage[];
  permalink?: string | null;
  syncRunId?: string;
}): SlackThreadCapture | null {
  const messages = input.messages
    .filter(isSlackTextMessage)
    .sort((left, right) => Number(left.ts) - Number(right.ts));
  const root = messages[0];
  if (!root?.ts) return null;

  let offset = 0;
  const safeSegments = messages.map((message, index) => {
    const capturedAt = isoFromSlackTs(message.ts) ?? message.ts!;
    const text = `${capturedAt}\n${message.text!.trim()}`;
    const startOffset = offset;
    offset += text.length;
    if (index < messages.length - 1) offset += 2;
    return {
      id: `slack:${input.channel.id}:${message.ts}`,
      capturedAt,
      sourceUrl: input.permalink ?? undefined,
      text,
      startOffset,
      endOffset: startOffset + text.length,
      reactionCount: slackReactionCount(message),
    };
  });
  const content = safeSegments.map((segment) => segment.text).join("\n\n");
  const threadTs = root.thread_ts ?? root.ts;
  return {
    externalId: `slack:${input.channel.id}:${threadTs}`,
    title: `#${input.channel.name ?? input.channel.id} thread ${isoFromSlackTs(threadTs) ?? threadTs}`,
    content,
    capturedAt: isoFromSlackTs(threadTs) ?? nowIso(),
    metadata: {
      provider: "slack",
      connector: "slack",
      ...(input.syncRunId ? { syncRunId: input.syncRunId } : {}),
      channelId: input.channel.id,
      channelName: input.channel.name,
      threadTs,
      messageCount: safeSegments.length,
      reactionCount: safeSegments.reduce(
        (total, segment) => total + segment.reactionCount,
        0,
      ),
      sourceUrl: input.permalink ?? null,
      permalink: input.permalink ?? null,
      // These offsets are against `content`, the safe persisted capture, never
      // against a provider payload.
      safeSegments,
    },
  };
}

async function slackApi<T>(
  token: string,
  method: string,
  params: Record<string, string | number | boolean | null | undefined> = {},
  options: {
    httpMethod?: "GET" | "POST";
    toleratedErrors?: readonly string[];
  } = {},
): Promise<T> {
  const httpMethod = options.httpMethod ?? "GET";
  const url = buildUrl(
    `https://slack.com/api/${method}`,
    httpMethod === "GET" ? params : {},
  );
  const body = new URLSearchParams();
  if (httpMethod === "POST") {
    for (const [key, value] of Object.entries(params)) {
      if (value !== null && value !== undefined) body.set(key, String(value));
    }
  }
  const response = await fetch(url.toString(), {
    method: httpMethod,
    headers: {
      Authorization: `Bearer ${token}`,
      ...(httpMethod === "POST"
        ? { "Content-Type": "application/x-www-form-urlencoded" }
        : {}),
    },
    ...(httpMethod === "POST" ? { body } : {}),
  });
  if (response.status === 429) {
    throw new ConnectorRateLimitError(
      "slack",
      method,
      retryAfterSeconds(response.headers),
    );
  }
  if (!response.ok) {
    throw new Error(`Slack ${method} failed (${response.status})`);
  }
  const data = (await response.json()) as {
    ok?: boolean;
    error?: string;
    needed?: string;
    provided?: string;
  };
  if (data.ok === false) {
    if (data.error === "ratelimited") {
      throw new ConnectorRateLimitError("slack", method, 60);
    }
    if (data.error && options.toleratedErrors?.includes(data.error)) {
      return data as T;
    }
    const details = [
      data.needed ? `needed: ${data.needed}` : null,
      data.provided ? `provided: ${data.provided}` : null,
    ].filter(Boolean);
    throw new Error(
      `Slack ${method} failed: ${data.error ?? "unknown"}${details.length ? ` (${details.join("; ")})` : ""}`,
    );
  }
  return data as T;
}

async function resolveSlackChannelByName(
  token: string,
  name: string,
): Promise<SlackChannel | null> {
  let cursor: string | undefined;
  for (let page = 0; page < 10; page += 1) {
    const data = await slackApi<SlackListResponse>(
      token,
      "conversations.list",
      {
        types: "public_channel,private_channel",
        exclude_archived: true,
        limit: 200,
        cursor,
      },
    );
    const match = (data.channels ?? []).find(
      (channel) => channel.name?.toLowerCase() === name.toLowerCase(),
    );
    if (match) return match;
    cursor = data.response_metadata?.next_cursor;
    if (!cursor) break;
  }
  return null;
}

function matchesPublicChannelExclusion(
  channel: SlackChannel,
  patterns: string[],
) {
  return patterns.some((pattern) => {
    try {
      const expression = new RegExp(pattern, "i");
      return expression.test(channel.id) || expression.test(channel.name ?? "");
    } catch {
      return false;
    }
  });
}

async function listPublicSlackChannels(
  token: string,
  exclusionPatterns: string[],
): Promise<{ channels: SlackChannel[]; excluded: number }> {
  const channels: SlackChannel[] = [];
  const seenCursors = new Set<string>();
  let cursor: string | undefined;
  let excluded = 0;
  for (;;) {
    const page = await slackApi<SlackListResponse>(
      token,
      "conversations.list",
      {
        types: "public_channel",
        exclude_archived: true,
        limit: 200,
        cursor,
      },
    );
    for (const channel of page.channels ?? []) {
      if (
        channel.is_archived ||
        channel.is_private ||
        channel.is_channel === false
      ) {
        continue;
      }
      if (matchesPublicChannelExclusion(channel, exclusionPatterns)) {
        excluded += 1;
        continue;
      }
      channels.push(channel);
    }
    cursor = page.response_metadata?.next_cursor;
    if (!cursor || seenCursors.has(cursor)) break;
    seenCursors.add(cursor);
  }
  return { channels, excluded };
}

async function resolveSlackChannel(
  token: string,
  channelRef: string,
): Promise<SlackChannel | null> {
  const directConversation = slackDirectConversationFromRef(channelRef);
  if (directConversation) return directConversation;

  const looksLikeId = /^[CG][A-Z0-9]+$/i.test(channelRef);
  if (looksLikeId) {
    const data = await slackApi<SlackInfoResponse>(
      token,
      "conversations.info",
      { channel: channelRef },
    );
    return data.channel ?? null;
  }
  const byName = await resolveSlackChannelByName(token, channelRef);
  if (!byName) return null;
  const data = await slackApi<SlackInfoResponse>(token, "conversations.info", {
    channel: byName.id,
  });
  return data.channel ?? byName;
}

async function slackPermalink(
  token: string,
  channelId: string,
  ts: string | undefined,
): Promise<string | null> {
  if (!ts) return null;
  const data = await slackApi<SlackPermalinkResponse>(
    token,
    "chat.getPermalink",
    { channel: channelId, message_ts: ts },
  );
  return data.permalink ?? null;
}

async function slackPrivateChannelMemberEmails(
  token: string,
  channelId: string,
  userEmailCache: SlackUserEmailCache,
): Promise<string[] | null> {
  const memberIds = new Set<string>();
  const seenCursors = new Set<string>();
  let cursor: string | undefined;
  while (true) {
    const response = await slackApi<SlackMembersResponse>(
      token,
      "conversations.members",
      { channel: channelId, limit: 1_000, cursor },
    );
    for (const memberId of response.members ?? []) {
      if (memberId) memberIds.add(memberId);
    }
    const nextCursor = response.response_metadata?.next_cursor?.trim();
    if (!nextCursor) break;
    if (seenCursors.has(nextCursor)) return null;
    seenCursors.add(nextCursor);
    cursor = nextCursor;
  }
  const userIds = Array.from(memberIds);
  if (!userIds.length) return null;
  if (
    userIds.some(
      (userId) => userEmailCache.has(userId) && !userEmailCache.get(userId),
    )
  ) {
    return null;
  }

  const uncachedUserIds = userIds.filter(
    (userId) => !userEmailCache.has(userId),
  );
  for (
    let offset = 0;
    offset < uncachedUserIds.length;
    offset += SLACK_USER_LOOKUP_CONCURRENCY
  ) {
    const batch = uncachedUserIds.slice(
      offset,
      offset + SLACK_USER_LOOKUP_CONCURRENCY,
    );
    await Promise.all(
      batch.map(async (userId) => {
        const user = await slackApi<SlackUserInfoResponse>(
          token,
          "users.info",
          { user: userId },
        );
        const email = user.user?.profile?.email?.trim().toLowerCase() ?? null;
        userEmailCache.set(userId, email || null);
      }),
    );
    if (batch.some((userId) => !userEmailCache.get(userId))) return null;
  }

  const emails: string[] = [];
  for (const userId of userIds) {
    const email = userEmailCache.get(userId);
    if (!email) return null;
    emails.push(email);
  }
  return Array.from(new Set(emails)).sort();
}

function configuredSlackChannelIds(config: Record<string, unknown>) {
  return new Set(
    slackChannelRefsFromConfig(config).filter((value) =>
      /^[CG][A-Z0-9]+$/i.test(value),
    ),
  );
}

async function createSlackThreadCapture(input: {
  source: SourceRow;
  channel: SlackChannel;
  messages: SlackMessage[];
  permalink: string | null;
  syncRunId?: string;
  memberEmails?: string[] | null;
}) {
  const normalized = normalizeSlackThreadCapture({
    channel: input.channel,
    messages: input.messages,
    permalink: input.permalink,
    syncRunId: input.syncRunId,
  });
  if (!normalized) return { capture: null, blocked: false };
  try {
    const capture = await createCapture({
      sourceId: input.source.id,
      externalId: normalized.externalId,
      title: normalized.title,
      kind: "message",
      content: normalized.content,
      capturedAt: normalized.capturedAt,
      metadata: normalized.metadata,
      audience: {
        kind: input.channel.is_private ? "slack-private-channel" : "org",
        memberEmails: input.memberEmails ?? undefined,
        upstreamRefHash: input.channel.id,
      },
    });
    return { capture, blocked: false };
  } catch (error) {
    if (error instanceof BrainCaptureBlockedError) {
      return { capture: null, blocked: true };
    }
    throw error;
  }
}

export async function testSlackConnection(
  options: {
    channelRefs?: string[];
    resolveNames?: boolean;
    workspaceConnectionId?: string;
  } = {},
) {
  const token = await requireConnectorCredential(
    "SLACK_BOT_TOKEN",
    "Slack",
    "slack",
    { workspaceConnectionId: options.workspaceConnectionId },
  );
  const auth = await slackApi<SlackAuthTestResponse>(token, "auth.test");
  const channelRefs = Array.from(
    new Set(
      (options.channelRefs ?? [])
        .map((ref) => ref.trim().replace(/^#/, ""))
        .filter(Boolean),
    ),
  ).slice(0, 25);
  const channels = [];

  for (const ref of channelRefs) {
    const directConversation = slackDirectConversationFromRef(ref);
    if (directConversation) {
      channels.push({
        ref,
        id: directConversation.id,
        status: "excluded" as const,
        message: "DMs and MPIMs are structurally excluded.",
        directExcluded: true,
        archived: false,
        privateChannel: false,
      });
      continue;
    }

    const looksLikeId = /^[CG][A-Z0-9]+$/i.test(ref);
    if (!looksLikeId && !options.resolveNames) {
      channels.push({
        ref,
        status: "skipped" as const,
        message:
          "Name resolution was disabled. Pass a Slack channel ID or set resolveNames=true.",
      });
      continue;
    }

    const channel = await resolveSlackChannel(token, ref);
    if (!channel) {
      channels.push({
        ref,
        status: "missing" as const,
        message: "Slack did not return a matching channel.",
      });
      continue;
    }

    const directExcluded = isSlackDirectConversation(channel);
    const usable = isUsableSlackChannel(channel);
    channels.push({
      ref,
      id: channel.id,
      name: channel.name,
      status: usable ? ("ok" as const) : ("excluded" as const),
      directExcluded,
      archived: channel.is_archived === true,
      privateChannel: channel.is_group === true,
      message: usable
        ? "Channel can be allow-listed for Brain sync."
        : directExcluded
          ? "DMs and MPIMs are structurally excluded."
          : "Channel is archived or not a supported channel type.",
    });
  }

  return {
    ok: true,
    team: auth.team ?? null,
    teamId: auth.team_id ?? null,
    workspaceUrl: auth.url ?? null,
    botUser: auth.user ?? null,
    botUserId: auth.user_id ?? null,
    botId: auth.bot_id ?? null,
    checkedChannels: channels.length,
    channels,
    historyRead: false,
  };
}

function slackPilotPrivacyExclusions(
  validation: SlackPilotReport["channelValidation"],
  readHistory: boolean,
) {
  const exclusions = [
    "DMs and MPIMs are structurally excluded before any history read.",
    "Only explicitly allow-listed public or private channels are eligible.",
  ];
  if (validation.excluded > 0) {
    exclusions.push(
      `${validation.excluded} channel reference(s) were excluded as unsupported or archived.`,
    );
  }
  if (validation.missing > 0) {
    exclusions.push(
      `${validation.missing} channel reference(s) were not found by Slack.`,
    );
  }
  if (validation.skipped > 0) {
    exclusions.push(
      `${validation.skipped} channel name(s) were skipped because name resolution was not enabled.`,
    );
  }
  if (!readHistory) {
    exclusions.push("No Slack message history was read for this pilot report.");
  }
  return exclusions;
}

function slackPilotNextSteps(report: {
  status: SlackPilotReport["status"];
  historyRead: boolean;
  channelValidation: SlackPilotReport["channelValidation"];
  capturesCreated: number;
}) {
  if (report.channelValidation.requested === 0) {
    return [
      "Add one or two Slack channel IDs to the source allow-list.",
      "Run the pilot again before attempting any history sync.",
    ];
  }
  if (report.channelValidation.ok === 0) {
    return [
      "Fix the channel allow-list or enable name resolution for channel names.",
      "Confirm the Slack bot is a member of each private channel.",
    ];
  }
  if (!report.historyRead) {
    return [
      "Review the validated channel list.",
      "Run Pilot sync only if these channels should read a tiny recent sample.",
    ];
  }
  if (report.capturesCreated > 0) {
    return [
      "Review the new raw captures and distill only durable company knowledge.",
      "Keep regular sync disabled until the source rules look right.",
    ];
  }
  return [
    "No new messages were captured; check channel membership and recent activity.",
    "Widen the source window deliberately if the pilot needs older messages.",
  ];
}

export async function runSlackPilot(
  source: SourceRow,
  options: SlackPilotOptions = {},
): Promise<SlackPilotReport> {
  const config = parseJson<Record<string, unknown>>(source.configJson, {});
  const requestedRefs = Array.from(
    new Set(
      (options.channelRefs?.length
        ? options.channelRefs
        : slackChannelRefsFromConfig(config)
      )
        .map((ref) => ref.trim().replace(/^#/, ""))
        .filter(Boolean),
    ),
  );
  const maxChannels = clampInteger(options.maxChannels, 2, 1, 2);
  const historyLimit = clampInteger(options.historyLimit, 10, 1, 10);
  const permalinkLimit = clampInteger(options.permalinkLimit, 10, 0, 10);
  const recentDays = clampInteger(options.recentDays, 14, 1, 30);
  const oldest = options.oldest ?? daysAgoIso(recentDays);
  const baseGuardrails = {
    historyReadRequested: options.readHistory === true,
    maxChannels,
    historyLimit,
    pagesPerChannel: 1,
    permalinkLimit,
    autoSync: false as const,
    oldest,
  };
  const sourceSummary = await summarizeSlackPilotSource(source.id);

  let credential;
  try {
    credential = await testSlackConnection({
      channelRefs: requestedRefs,
      resolveNames: options.resolveNames === true,
      workspaceConnectionId:
        typeof config.workspaceConnectionId === "string"
          ? config.workspaceConnectionId
          : undefined,
    });
  } catch (err) {
    const message = err instanceof Error ? err.message : String(err);
    const channelValidation = {
      requested: requestedRefs.length,
      checked: 0,
      ok: 0,
      excluded: 0,
      missing: 0,
      skipped: 0,
      channels: [],
    };
    return {
      sourceId: source.id,
      sourceTitle: source.title,
      ok: false,
      status: "error",
      historyRead: false,
      credential: { ok: false, error: message },
      guardrails: baseGuardrails,
      channelValidation,
      capturesCreated: 0,
      captures: [],
      ...sourceSummary,
      privacyExclusions: slackPilotPrivacyExclusions(channelValidation, false),
      nextSteps: [
        "Fix the Slack credential before validating channel allow-lists.",
        "Then rerun the pilot with history reads still off.",
      ],
    };
  }

  const channels = credential.channels;
  const channelValidation = {
    requested: requestedRefs.length,
    checked: credential.checkedChannels,
    ok: channels.filter((channel) => channel.status === "ok").length,
    excluded: channels.filter((channel) => channel.status === "excluded")
      .length,
    missing: channels.filter((channel) => channel.status === "missing").length,
    skipped: channels.filter((channel) => channel.status === "skipped").length,
    channels,
  };

  if (requestedRefs.length === 0 || !options.readHistory) {
    const blocked = requestedRefs.length === 0;
    const partial = {
      status: blocked ? ("blocked" as const) : ("validated" as const),
      historyRead: false,
      channelValidation,
      capturesCreated: 0,
    };
    return {
      sourceId: source.id,
      sourceTitle: source.title,
      ok: !blocked,
      ...partial,
      credential: {
        ok: true,
        team: credential.team,
        teamId: credential.teamId,
        workspaceUrl: credential.workspaceUrl,
        botUser: credential.botUser,
      },
      guardrails: baseGuardrails,
      captures: [],
      ...sourceSummary,
      privacyExclusions: slackPilotPrivacyExclusions(channelValidation, false),
      nextSteps: slackPilotNextSteps(partial),
    };
  }

  const validatedChannelIds = channels
    .filter(
      (channel): channel is (typeof channels)[number] & { id: string } =>
        channel.status === "ok" && typeof channel.id === "string",
    )
    .map((channel) => channel.id)
    .slice(0, maxChannels);

  if (validatedChannelIds.length === 0) {
    const partial = {
      status: "blocked" as const,
      historyRead: false,
      channelValidation,
      capturesCreated: 0,
    };
    return {
      sourceId: source.id,
      sourceTitle: source.title,
      ok: false,
      ...partial,
      credential: {
        ok: true,
        team: credential.team,
        teamId: credential.teamId,
        workspaceUrl: credential.workspaceUrl,
        botUser: credential.botUser,
      },
      guardrails: baseGuardrails,
      captures: [],
      ...sourceSummary,
      privacyExclusions: slackPilotPrivacyExclusions(channelValidation, false),
      nextSteps: slackPilotNextSteps(partial),
    };
  }

  const boundedConfig = {
    ...removeSlackHistoryConfig(config),
    channelIds: validatedChannelIds,
    historyLimit,
    maxChannelsPerSync: maxChannels,
    pagesPerChannel: 1,
    permalinkLimit,
    autoSync: false,
    oldest,
    slack: {
      ...objectValue(removeSlackHistoryConfig(config).slack),
      channelIds: validatedChannelIds,
      historyLimit,
      maxChannelsPerSync: maxChannels,
      pagesPerChannel: 1,
      permalinkLimit,
      autoSync: false,
      oldest,
    },
  };
  const sync = await runConnectorSync({
    ...source,
    configJson: stableJson(boundedConfig),
  });
  const updatedSummary = await summarizeSlackPilotSource(source.id);
  const captures = sync.captures.map((capture) => ({
    id: capture.id,
    title: capture.title,
    capturedAt: capture.capturedAt,
    sourceUrl: sourceUrlFromCapture(capture),
  }));
  const partial = {
    status:
      sync.status === "success" ? ("synced" as const) : ("error" as const),
    historyRead: true,
    channelValidation,
    capturesCreated: sync.capturesCreated,
  };

  return {
    sourceId: source.id,
    sourceTitle: source.title,
    ok: sync.status === "success",
    ...partial,
    credential: {
      ok: true,
      team: credential.team,
      teamId: credential.teamId,
      workspaceUrl: credential.workspaceUrl,
      botUser: credential.botUser,
    },
    guardrails: baseGuardrails,
    sync: {
      runId: sync.runId,
      status: sync.status,
      message: sync.message,
      stats: sync.stats,
    },
    captures,
    ...updatedSummary,
    privacyExclusions: slackPilotPrivacyExclusions(channelValidation, true),
    nextSteps: slackPilotNextSteps(partial),
  };
}

function granolaSpeakerLabel(item: Record<string, unknown>): string {
  const speaker = objectValue(item.speaker);
  const label =
    speaker.diarization_label ??
    speaker.name ??
    speaker.source ??
    item.speaker ??
    "speaker";
  return String(label);
}

function granolaTranscriptLines(transcript: unknown): string[] {
  if (!Array.isArray(transcript)) return [];
  return transcript
    .map((item) => {
      if (typeof item === "string") return item.trim();
      if (!item || typeof item !== "object") return "";
      const record = item as Record<string, unknown>;
      const text = typeof record.text === "string" ? record.text.trim() : "";
      if (!text) return "";
      const start =
        typeof record.start_time === "string" ? ` ${record.start_time}` : "";
      return `[${granolaSpeakerLabel(record)}${start}] ${text}`;
    })
    .filter(Boolean);
}

export function normalizeGranolaNote(note: GranolaNote) {
  const calendar = objectValue(note.calendar_event);
  const title =
    note.title ??
    (typeof calendar.event_title === "string"
      ? calendar.event_title
      : "Granola meeting note");
  const summary =
    note.summary_markdown ??
    note.summary_text ??
    readableJson(note.summary).trim();
  const transcriptLines = granolaTranscriptLines(note.transcript);
  const body = [
    summary ? `Summary\n${summary}` : "",
    transcriptLines.length ? `Transcript\n${transcriptLines.join("\n")}` : "",
  ]
    .filter(Boolean)
    .join("\n\n")
    .trim();

  return {
    externalId: note.id ? `granola:${note.id}` : undefined,
    title,
    content: body || summary || title,
    capturedAt:
      (typeof calendar.scheduled_start_time === "string"
        ? calendar.scheduled_start_time
        : undefined) ??
      note.created_at ??
      nowIso(),
    metadata: {
      provider: "granola",
      granolaNoteId: note.id,
      createdAt: note.created_at,
      updatedAt: note.updated_at,
      scheduledStartTime:
        typeof calendar.scheduled_start_time === "string"
          ? calendar.scheduled_start_time
          : undefined,
      attendeesCount: Array.isArray(note.attendees)
        ? note.attendees.length
        : undefined,
      transcriptSegmentCount: Array.isArray(note.transcript)
        ? note.transcript.length
        : undefined,
      hasOwner: Boolean(note.owner),
      hasFolders: Boolean(note.folder_membership),
    },
  };
}

async function granolaApi<T>(
  token: string,
  path: string,
  params: Record<string, string | number | boolean | null | undefined> = {},
): Promise<T> {
  const url = buildUrl(`https://public-api.granola.ai/v1${path}`, params);
  const response = await fetch(url.toString(), {
    headers: { Authorization: `Bearer ${token}` },
  });
  if (response.status === 429) {
    throw new ConnectorRateLimitError(
      "granola",
      path,
      retryAfterSeconds(response.headers),
    );
  }
  if (!response.ok) {
    throw new Error(`Granola ${path} failed (${response.status})`);
  }
  return (await response.json()) as T;
}

function githubRepoFromValue(value: string): string | null {
  const trimmed = value.trim().replace(/\.git$/, "");
  if (!trimmed) return null;
  const withoutProtocol = trimmed
    .replace(/^https?:\/\/github\.com\//i, "")
    .replace(/^git@github\.com:/i, "");
  const [owner, repo] = withoutProtocol.split("/");
  if (!owner || !repo) return null;
  const cleanRepo = repo.split(/[?#]/)[0];
  if (
    !/^[A-Za-z0-9_.-]+$/.test(owner) ||
    !/^[A-Za-z0-9_.-]+$/.test(cleanRepo)
  ) {
    return null;
  }
  return `${owner}/${cleanRepo}`;
}

function githubReposFromConfig(config: Record<string, unknown>): string[] {
  return configuredList(config, ["repositories", "repos"], "github")
    .map(githubRepoFromValue)
    .filter((repo): repo is string => Boolean(repo));
}

function githubLinkedSourceIdsFromConfig(config: Record<string, unknown>) {
  return configuredList(
    config,
    ["linkedSourceIds", "linkedSlackSourceIds", "slackSourceIds", "sourceIds"],
    "github",
  );
}

function configuredBoolean(
  config: Record<string, unknown>,
  keys: string[],
  fallback: boolean,
  nestedKey?: string,
): boolean {
  const configs = nestedKey
    ? [config, objectValue(config[nestedKey])]
    : [config];
  for (const itemConfig of configs) {
    for (const key of keys) {
      const raw = itemConfig[key];
      if (typeof raw === "boolean") return raw;
      if (typeof raw === "string") {
        if (raw.toLowerCase() === "true") return true;
        if (raw.toLowerCase() === "false") return false;
      }
    }
  }
  return fallback;
}

function githubStateFromConfig(config: Record<string, unknown>) {
  const raw =
    (typeof config.state === "string" ? config.state : undefined) ??
    (typeof objectValue(config.github).state === "string"
      ? String(objectValue(config.github).state)
      : undefined);
  return raw === "open" || raw === "closed" || raw === "all" ? raw : "all";
}

function githubRetryAfterSeconds(headers: Headers): number {
  const retryAfter = retryAfterSeconds(headers);
  if (retryAfter !== 60) return retryAfter;
  const reset = headers.get("x-ratelimit-reset");
  const resetMs = reset ? Number(reset) * 1000 : Number.NaN;
  if (Number.isFinite(resetMs) && resetMs > Date.now()) {
    return Math.ceil((resetMs - Date.now()) / 1000);
  }
  return retryAfter;
}

function githubLabels(labels: GitHubIssueOrPull["labels"]): string[] {
  if (!Array.isArray(labels)) return [];
  return labels
    .map((label) => (typeof label === "string" ? label : label.name))
    .filter((label): label is string => Boolean(label));
}

function textExcerpt(value: string | null | undefined, maxLength: number) {
  const trimmed = (value ?? "").trim();
  if (trimmed.length <= maxLength) return trimmed;
  return `${trimmed.slice(0, Math.max(0, maxLength - 3)).trimEnd()}...`;
}

function sourceUrlFromMetadataRecord(metadata: Record<string, unknown>) {
  for (const key of ["sourceUrl", "url", "permalink", "webUrl", "web_url"]) {
    const value = metadata[key];
    if (typeof value === "string" && value.trim()) return value.trim();
  }
  return null;
}

function summarizeGitHubComments(comments: GitHubIssueComment[]) {
  return comments
    .filter((comment) => comment.body?.trim())
    .slice(0, 5)
    .map((comment) => {
      const author = comment.user?.login ?? "unknown";
      const at = comment.updated_at ?? comment.created_at ?? "unknown time";
      return `- ${author} at ${at}: ${textExcerpt(comment.body, 280)}`;
    });
}

function summarizeGitHubReviews(reviews: GitHubPullReview[]) {
  return reviews.slice(0, 5).map((review) => {
    const author = review.user?.login ?? "unknown";
    const state = review.state ?? "reviewed";
    const at = review.submitted_at ?? "unknown time";
    const body = textExcerpt(review.body, 220);
    return `- ${author} ${state.toLowerCase()} at ${at}${body ? `: ${body}` : ""}`;
  });
}

function gitHubRefKey(ref: GitHubLinkedRef) {
  return `${ref.repo}:${ref.number}`;
}

function githubRefsFromText(
  text: string,
  linkedFrom?: GitHubLinkedRef["linkedFrom"],
): GitHubLinkedRef[] {
  const refs: GitHubLinkedRef[] = [];
  const pattern =
    /https:\/\/github\.com\/([A-Za-z0-9_.-]+)\/([A-Za-z0-9_.-]+)\/(issues|pull)\/(\d+)/gi;
  for (const match of text.matchAll(pattern)) {
    const repo = githubRepoFromValue(`${match[1]}/${match[2]}`);
    const number = Number(match[4]);
    if (!repo || !Number.isInteger(number) || number <= 0) continue;
    refs.push({
      repo,
      number,
      type: match[3]?.toLowerCase() === "pull" ? "pull" : "issue",
      url: `https://github.com/${repo}/${match[3]}/${number}`,
      linkedFrom,
    });
  }
  return refs;
}

function normalizeGitHubIssue(
  repo: string,
  item: GitHubIssueOrPull,
  context: GitHubContext = { comments: [], reviews: [] },
  linkedFrom?: GitHubLinkedRef["linkedFrom"],
) {
  const isPullRequest = Boolean(item.pull_request);
  const number = item.number ?? item.id ?? "unknown";
  const labels = githubLabels(item.labels);
  const sourceUrl = item.html_url ?? item.pull_request?.html_url ?? null;
  const merged =
    context.pull?.merged ??
    Boolean(item.pull_request?.merged_at ?? context.pull?.merged_at);
  const mergedAt = context.pull?.merged_at ?? item.pull_request?.merged_at;
  const title =
    item.title ??
    `${repo} ${isPullRequest ? "pull request" : "issue"} #${number}`;
  const author = item.user?.login ?? "unknown";
  const bodyExcerpt = textExcerpt(item.body, 4000);
  const commentSummary = summarizeGitHubComments(context.comments);
  const reviewSummary = summarizeGitHubReviews(context.reviews);
  const header = [
    `GitHub ${repo} ${isPullRequest ? "pull request" : "issue"} #${number}`,
    `State: ${item.state ?? "unknown"}`,
    isPullRequest
      ? `Merged: ${merged ? "yes" : "no"}${mergedAt ? ` (${mergedAt})` : ""}`
      : "",
    `Author: ${author}`,
    labels.length ? `Labels: ${labels.join(", ")}` : "",
    item.created_at ? `Created: ${item.created_at}` : "",
    item.updated_at ? `Updated: ${item.updated_at}` : "",
    item.closed_at ? `Closed: ${item.closed_at}` : "",
    sourceUrl ? `URL: ${sourceUrl}` : "",
    linkedFrom
      ? `Linked from Slack capture: ${linkedFrom.captureTitle} (${linkedFrom.captureId})`
      : "",
    linkedFrom?.sourceUrl ? `Slack URL: ${linkedFrom.sourceUrl}` : "",
  ].filter(Boolean);
  const content = [
    header.join("\n"),
    "",
    `Title\n${title}`,
    "",
    bodyExcerpt ? `Body excerpt\n${bodyExcerpt}` : "Body excerpt\n(No body)",
    commentSummary.length
      ? `Comment summary\n${commentSummary.join("\n")}`
      : "",
    reviewSummary.length ? `Review summary\n${reviewSummary.join("\n")}` : "",
  ]
    .filter(Boolean)
    .join("\n\n")
    .trim();

  return {
    externalId: `github:${repo}:${isPullRequest ? "pull" : "issue"}:${number}`,
    title,
    kind: isPullRequest ? ("document" as const) : ("note" as const),
    content,
    capturedAt: item.updated_at ?? item.created_at ?? nowIso(),
    metadata: {
      provider: "github",
      connector: "github",
      repository: repo,
      number: item.number,
      githubId: item.id,
      type: isPullRequest ? "pull_request" : "issue",
      state: item.state,
      merged: isPullRequest ? merged : undefined,
      mergedAt,
      author,
      authorUrl: item.user?.html_url,
      labels,
      bodyExcerpt,
      sourceUrl,
      htmlUrl: item.html_url,
      createdAt: item.created_at,
      updatedAt: item.updated_at,
      closedAt: item.closed_at,
      linkedFrom,
      pullRequest: {
        ...item.pull_request,
        details: context.pull,
      },
      comments: context.comments.map((comment) => ({
        id: comment.id,
        author: comment.user?.login,
        authorUrl: comment.user?.html_url,
        sourceUrl: comment.html_url,
        createdAt: comment.created_at,
        updatedAt: comment.updated_at,
        bodyExcerpt: textExcerpt(comment.body, 800),
      })),
      reviews: context.reviews.map((review) => ({
        id: review.id,
        author: review.user?.login,
        authorUrl: review.user?.html_url,
        sourceUrl: review.html_url,
        state: review.state,
        submittedAt: review.submitted_at,
        bodyExcerpt: textExcerpt(review.body, 800),
      })),
    },
  };
}

async function githubApi<T>(
  token: string,
  path: string,
  params: Record<string, string | number | boolean | null | undefined> = {},
): Promise<T> {
  const url = buildUrl(`https://api.github.com${path}`, params);
  const response = await fetch(url.toString(), {
    headers: {
      Authorization: `Bearer ${token}`,
      Accept: "application/vnd.github+json",
      "X-GitHub-Api-Version": "2022-11-28",
      "User-Agent": "agent-native-brain",
    },
  });
  const remaining = response.headers.get("x-ratelimit-remaining");
  if (
    response.status === 429 ||
    (response.status === 403 &&
      (remaining === "0" || response.headers.has("retry-after")))
  ) {
    throw new ConnectorRateLimitError(
      "github",
      path,
      githubRetryAfterSeconds(response.headers),
    );
  }
  if (!response.ok) {
    let suffix = "";
    try {
      const data = (await response.json()) as { message?: string };
      suffix = data.message ? `: ${data.message}` : "";
    } catch {
      suffix = "";
    }
    throw new Error(`GitHub ${path} failed (${response.status})${suffix}`);
  }
  return (await response.json()) as T;
}

async function fetchGitHubContext(
  token: string,
  repo: string,
  item: GitHubIssueOrPull,
  options: {
    commentLimit: number;
    reviewLimit: number;
  },
): Promise<GitHubContext> {
  const number = item.number;
  if (!number) return { comments: [], reviews: [] };
  const isPullRequest = Boolean(item.pull_request);
  const [comments, pull, reviews] = await Promise.all([
    options.commentLimit > 0
      ? githubApi<GitHubIssueComment[]>(
          token,
          `/repos/${repo}/issues/${number}/comments`,
          { per_page: options.commentLimit },
        )
      : Promise.resolve([]),
    isPullRequest
      ? githubApi<GitHubPullDetails>(token, `/repos/${repo}/pulls/${number}`)
      : Promise.resolve(null),
    isPullRequest && options.reviewLimit > 0
      ? githubApi<GitHubPullReview[]>(
          token,
          `/repos/${repo}/pulls/${number}/reviews`,
          { per_page: options.reviewLimit },
        )
      : Promise.resolve([]),
  ]);
  return { comments, pull, reviews };
}

async function githubRefsFromLinkedSources(
  config: Record<string, unknown>,
  options: {
    captureLimit: number;
    refLimit: number;
    stats: Record<string, unknown>;
  },
): Promise<GitHubLinkedRef[]> {
  const sourceIds = githubLinkedSourceIdsFromConfig(config);
  options.stats.linkedSourceIds = sourceIds.length;
  if (!sourceIds.length) return [];

  const db = getDb();
  const refsByKey = new Map<string, GitHubLinkedRef>();

  for (const sourceId of sourceIds) {
    if (refsByKey.size >= options.refLimit) break;
    const access = await assertAccess("brain-source", sourceId, "viewer");
    if (access.resource.provider !== "slack") {
      options.stats.linkedSourcesSkipped =
        Number(options.stats.linkedSourcesSkipped) + 1;
      continue;
    }
    const audienceIds = await listAccessibleAudienceIds([sourceId]);
    if (!audienceIds.length) continue;
    const captures = await db
      .select({
        id: schema.brainRawCaptures.id,
        title: schema.brainRawCaptures.title,
        content: schema.brainRawCaptures.content,
        metadataJson: schema.brainRawCaptures.metadataJson,
      })
      .from(schema.brainRawCaptures)
      .innerJoin(
        schema.brainCaptureAudiences,
        eq(schema.brainCaptureAudiences.captureId, schema.brainRawCaptures.id),
      )
      .innerJoin(
        schema.brainAudiences,
        eq(schema.brainAudiences.id, schema.brainCaptureAudiences.audienceId),
      )
      .where(
        and(
          eq(schema.brainRawCaptures.sourceId, sourceId),
          eq(schema.brainRawCaptures.sensitivityDisposition, "allowed"),
          inArray(schema.brainCaptureAudiences.audienceId, audienceIds),
          eq(schema.brainAudiences.kind, "org"),
        ),
      )
      .orderBy(desc(schema.brainRawCaptures.capturedAt))
      .limit(options.captureLimit);

    options.stats.linkedCapturesScanned =
      Number(options.stats.linkedCapturesScanned) + captures.length;

    for (const capture of captures) {
      if (refsByKey.size >= options.refLimit) break;
      const metadata = parseJson<Record<string, unknown>>(
        capture.metadataJson,
        {},
      );
      const linkedFrom = {
        sourceId,
        captureId: capture.id,
        captureTitle: capture.title,
        sourceUrl: sourceUrlFromMetadataRecord(metadata),
      };
      const candidates = [
        capture.content,
        typeof metadata.sourceUrl === "string" ? metadata.sourceUrl : "",
        typeof metadata.permalink === "string" ? metadata.permalink : "",
      ].join("\n");

      for (const ref of githubRefsFromText(candidates, linkedFrom)) {
        refsByKey.set(gitHubRefKey(ref), ref);
        if (refsByKey.size >= options.refLimit) break;
      }
    }
  }

  options.stats.linkedRefsFound = refsByKey.size;
  return Array.from(refsByKey.values());
}

interface ConnectorSyncRunLease {
  runId: string;
  leaseToken: string;
}

class ConnectorSyncAlreadyRunningError extends Error {
  constructor(public readonly runId: string) {
    super("A sync is already running for this Brain source");
    this.name = "ConnectorSyncAlreadyRunningError";
  }
}

async function createRun(
  source: SourceRow,
  options: { exclusive?: boolean } = {},
): Promise<ConnectorSyncRunLease> {
  const db = getDb();
  const exclusive = options.exclusive !== false;
  const runId = nanoid();
  const leaseToken = nanoid();
  const now = nowIso();
  const staleStartedAt = new Date(
    Date.now() - CONNECTOR_SYNC_LEASE_MS,
  ).toISOString();
  await db
    .update(schema.brainSyncRuns)
    .set({
      activeSourceId: null,
      status: "error",
      error: "Sync lease expired before completion",
      leaseToken: null,
      leaseExpiresAt: null,
      completedAt: now,
    })
    .where(
      and(
        eq(schema.brainSyncRuns.sourceId, source.id),
        eq(schema.brainSyncRuns.status, "running"),
        or(
          lte(schema.brainSyncRuns.leaseExpiresAt, now),
          and(
            isNull(schema.brainSyncRuns.leaseExpiresAt),
            lte(schema.brainSyncRuns.startedAt, staleStartedAt),
          ),
        ),
      ),
    );
  try {
    await db.insert(schema.brainSyncRuns).values({
      id: runId,
      sourceId: source.id,
      activeSourceId: exclusive ? source.id : null,
      provider: source.provider,
      status: "running",
      statsJson: "{}",
      error: null,
      leaseToken,
      leaseExpiresAt: new Date(
        Date.now() + CONNECTOR_SYNC_LEASE_MS,
      ).toISOString(),
      startedAt: now,
      completedAt: null,
    });
  } catch (error) {
    if (!exclusive) throw error;
    const [active] = await db
      .select({ id: schema.brainSyncRuns.id })
      .from(schema.brainSyncRuns)
      .where(eq(schema.brainSyncRuns.activeSourceId, source.id))
      .limit(1);
    if (active) throw new ConnectorSyncAlreadyRunningError(active.id);
    throw error;
  }
  return { runId, leaseToken };
}

async function renewRunLease(run: ConnectorSyncRunLease) {
  const renewed = await getDb()
    .update(schema.brainSyncRuns)
    .set({
      leaseExpiresAt: new Date(
        Date.now() + CONNECTOR_SYNC_LEASE_MS,
      ).toISOString(),
    })
    .where(
      and(
        eq(schema.brainSyncRuns.id, run.runId),
        eq(schema.brainSyncRuns.leaseToken, run.leaseToken),
        eq(schema.brainSyncRuns.status, "running"),
      ),
    );
  if (renewed.rowsAffected === 0) {
    throw new Error("Brain source sync lease was lost");
  }
}

function startRunLeaseHeartbeat(run: ConnectorSyncRunLease) {
  let stopped = false;
  let failure: unknown;
  let pendingRenewal: Promise<void> | null = null;
  const timer = setInterval(() => {
    if (stopped || failure || pendingRenewal) return;
    pendingRenewal = renewRunLease(run)
      .catch((error) => {
        failure = error;
        clearInterval(timer);
      })
      .finally(() => {
        pendingRenewal = null;
      });
  }, CONNECTOR_SYNC_HEARTBEAT_MS);
  timer.unref?.();

  return {
    assertHealthy() {
      if (failure) throw failure;
    },
    async stop() {
      stopped = true;
      clearInterval(timer);
      await pendingRenewal;
    },
  };
}

async function finishRun(
  run: ConnectorSyncRunLease,
  status: "success" | "error",
  stats: Record<string, unknown>,
  error?: string | null,
) {
  const finished = await getDb()
    .update(schema.brainSyncRuns)
    .set({
      activeSourceId: null,
      status,
      statsJson: stableJson(stats),
      error: error ?? null,
      leaseToken: null,
      leaseExpiresAt: null,
      completedAt: nowIso(),
    })
    .where(
      and(
        eq(schema.brainSyncRuns.id, run.runId),
        eq(schema.brainSyncRuns.leaseToken, run.leaseToken),
      ),
    );
  if (finished.rowsAffected === 0) {
    throw new Error("Brain source sync lease was lost");
  }
}

async function syncFromConfiguredItems(
  source: SourceRow,
  emptyMessage: string,
): Promise<ConnectorSyncResult> {
  const run = await createRun(source);
  const { runId } = run;
  const config = parseJson<Record<string, unknown>>(source.configJson, {});
  const items = transcriptItems(config);
  const captures = [];
  let sensitivityBlocked = 0;
  const heartbeat = startRunLeaseHeartbeat(run);

  try {
    for (const item of items) {
      const captureResult = await createConnectorCapture({
        sourceId: source.id,
        externalId: item.externalId,
        title: item.title,
        kind: item.kind,
        content: item.content,
        capturedAt: item.capturedAt,
        metadata: {
          ...item.metadata,
          connector: source.provider,
          syncRunId: runId,
        },
      });
      if (captureResult.capture)
        captures.push(serializeCapture(captureResult.capture));
      if (captureResult.blocked) sensitivityBlocked += 1;
      heartbeat.assertHealthy();
    }
    await heartbeat.stop();
    heartbeat.assertHealthy();
    await renewRunLease(run);
    await getDb()
      .update(schema.brainSources)
      .set({
        lastSyncedAt: nowIso(),
        lastError: null,
        status: "active",
        updatedAt: nowIso(),
      })
      .where(
        and(
          accessFilter(schema.brainSources, schema.brainSourceShares),
          eq(schema.brainSources.id, source.id),
        ),
      );
    await finishRun(run, "success", {
      capturesCreated: captures.length,
      sensitivityBlocked,
    });
    return {
      runId,
      sourceId: source.id,
      provider: source.provider as BrainSourceProvider,
      status: "success",
      capturesCreated: captures.length,
      captures,
      message: captures.length ? "Imported configured captures" : emptyMessage,
    };
  } catch (err) {
    await heartbeat.stop();
    const message = err instanceof Error ? err.message : String(err);
    await renewRunLease(run);
    await getDb()
      .update(schema.brainSources)
      .set({ lastError: message, status: "error", updatedAt: nowIso() })
      .where(
        and(
          accessFilter(schema.brainSources, schema.brainSourceShares),
          eq(schema.brainSources.id, source.id),
        ),
      );
    await finishRun(
      run,
      "error",
      { capturesCreated: captures.length },
      message,
    );
    return {
      runId,
      sourceId: source.id,
      provider: source.provider as BrainSourceProvider,
      status: "error",
      capturesCreated: captures.length,
      captures,
      message,
    };
  }
}

async function syncSlack(source: SourceRow): Promise<ConnectorSyncResult> {
  const config = parseJson<Record<string, unknown>>(source.configJson, {});
  if (isFixtureConfig(config) || transcriptItems(config).length > 0) {
    return syncFromConfiguredItems(
      source,
      "Slack fixture source has no configured messages.",
    );
  }

  const run = await createRun(source);
  const { runId } = run;
  const db = getDb();
  const cursor = parseJson<SlackSyncCursor>(source.cursorJson, {});
  const nextCursor: SlackSyncCursor = {
    ...cursor,
    channels: { ...(cursor.channels ?? {}) },
    retry: undefined,
    lastRunAt: nowIso(),
  };
  const channelRefs = configuredList(
    config,
    [
      "channelIds",
      "channels",
      "allowedChannels",
      "allowlistedChannels",
      "allowList",
    ],
    "slack",
  );
  const includePublicChannels = configuredBoolean(
    config,
    ["includePublicChannels"],
    false,
    "slack",
  );
  const limit = configuredNumber(config, ["historyLimit", "limit"], 15, {
    min: 1,
    max: 15,
    nestedKey: "slack",
  });
  const maxChannels = configuredNumber(
    config,
    ["maxChannelsPerSync", "channelBudget"],
    3,
    { min: 1, max: 25, nestedKey: "slack" },
  );
  const pagesPerChannel = configuredNumber(
    config,
    ["pagesPerChannel", "pageBudget"],
    1,
    { min: 1, max: 5, nestedKey: "slack" },
  );
  const permalinkLimit = configuredNumber(
    config,
    ["permalinkLimit", "citationLinkLimit"],
    15,
    { min: 0, max: 50, nestedKey: "slack" },
  );
  const threadCaptureConcurrency = configuredNumber(
    config,
    ["threadCaptureConcurrency", "threadConcurrency"],
    SLACK_THREAD_CAPTURE_CONCURRENCY,
    { min: 1, max: 8, nestedKey: "slack" },
  );
  const initialOldest = slackTsFromDateish(
    typeof config.oldest === "string"
      ? config.oldest
      : typeof config.updatedAfter === "string"
        ? config.updatedAfter
        : typeof objectValue(config.slack).oldest === "string"
          ? String(objectValue(config.slack).oldest)
          : undefined,
  );

  const captures = [];
  const stats: Record<string, unknown> = {
    configuredChannels: channelRefs.length,
    includePublicChannels,
    discoveredPublicChannels: 0,
    excludedPublicChannels: 0,
    eligibleChannels: 0,
    scannedChannels: 0,
    rejectedChannels: 0,
    missingChannels: 0,
    messagesSeen: 0,
    capturesCreated: 0,
    threadsFetched: 0,
    threadCaptureConcurrency,
    sensitivityBlocked: 0,
    publicChannelsJoined: 0,
    publicChannelsAlreadyJoined: 0,
    rateLimited: false,
  };

  try {
    const token = await requireConnectorCredential(
      "SLACK_BOT_TOKEN",
      "Slack",
      "slack",
      config,
    );
    if (!channelRefs.length && !includePublicChannels) {
      throw new Error(
        "Slack source must configure channelIds, channels, allowedChannels, or includePublicChannels",
      );
    }

    const settings = await readBrainSettings();
    const publicExclusionPatterns =
      settings.publicChannelExclusionPatterns ?? [];
    const channelsById = new Map<string, SlackChannel>();
    for (const channelRef of channelRefs) {
      const channel = await resolveSlackChannel(token, channelRef);
      if (!channel) {
        stats.missingChannels = Number(stats.missingChannels) + 1;
        continue;
      }
      if (!isUsableSlackChannel(channel)) {
        stats.rejectedChannels = Number(stats.rejectedChannels) + 1;
        continue;
      }
      const isExplicitId = /^[CG][A-Z0-9]+$/i.test(channelRef);
      if (channel.is_private && !isExplicitId) {
        stats.rejectedChannels = Number(stats.rejectedChannels) + 1;
        continue;
      }
      if (
        !channel.is_private &&
        matchesPublicChannelExclusion(channel, publicExclusionPatterns)
      ) {
        stats.excludedPublicChannels = Number(stats.excludedPublicChannels) + 1;
        continue;
      }
      channelsById.set(channel.id, channel);
    }
    if (includePublicChannels) {
      const discovered = await listPublicSlackChannels(
        token,
        publicExclusionPatterns,
      );
      stats.discoveredPublicChannels = discovered.channels.length;
      stats.excludedPublicChannels =
        Number(stats.excludedPublicChannels) + discovered.excluded;
      for (const channel of discovered.channels) {
        channelsById.set(channel.id, channel);
      }
    }
    const channels = [...channelsById.values()];
    stats.eligibleChannels = channels.length;
    const userEmailCache: SlackUserEmailCache = new Map();

    const channelsToScan = includePublicChannels
      ? (() => {
          const ordered = [...channels].sort((left, right) =>
            left.id.localeCompare(right.id),
          );
          if (!ordered.length) return ordered;
          const offset =
            Math.max(0, cursor.publicChannelOffset ?? 0) % ordered.length;
          const rotated = [
            ...ordered.slice(offset),
            ...ordered.slice(0, offset),
          ];
          nextCursor.publicChannelOffset =
            (offset + Math.min(maxChannels, rotated.length)) % ordered.length;
          return rotated.slice(0, maxChannels);
        })()
      : channels.slice(0, maxChannels);

    let permalinkCalls = 0;

    for (const channel of channelsToScan) {
      const membership = await ensureSlackPublicChannelMembership(
        channel,
        (channelId) =>
          slackApi<SlackChannelJoinResponse>(
            token,
            "conversations.join",
            { channel: channelId },
            {
              httpMethod: "POST",
              toleratedErrors: ["already_in_channel"],
            },
          ),
      );
      if (membership === "joined") {
        stats.publicChannelsJoined = Number(stats.publicChannelsJoined) + 1;
      } else if (membership === "already_member") {
        stats.publicChannelsAlreadyJoined =
          Number(stats.publicChannelsAlreadyJoined) + 1;
      }
      const privateMemberEmails = channel.is_private
        ? await slackPrivateChannelMemberEmails(
            token,
            channel.id,
            userEmailCache,
          )
        : null;
      if (channel.is_private && !privateMemberEmails?.length) {
        stats.rejectedChannels = Number(stats.rejectedChannels) + 1;
        continue;
      }

      stats.scannedChannels = Number(stats.scannedChannels) + 1;
      const channelCursor = nextCursor.channels?.[channel.id] ?? {};
      const pendingLatest =
        channelCursor.pendingLatestTs ?? channelCursor.latestTs;

      for (let page = 0; page < pagesPerChannel; page += 1) {
        const params: Record<
          string,
          string | number | boolean | null | undefined
        > = {
          channel: channel.id,
          limit,
        };
        if (channelCursor.pageCursor) {
          params.cursor = channelCursor.pageCursor;
        } else if (channelCursor.latestTs) {
          params.oldest = channelCursor.latestTs;
          params.inclusive = false;
        } else if (initialOldest) {
          params.oldest = initialOldest;
          params.inclusive = false;
        }

        const data = await slackApi<SlackHistoryResponse>(
          token,
          "conversations.history",
          params,
        );
        const messages = (data.messages ?? []).filter(isSlackTextMessage);
        stats.messagesSeen = Number(stats.messagesSeen) + messages.length;

        const newest = newestSlackTs(messages);
        if (!channelCursor.pendingLatestTs && newest) {
          channelCursor.pendingLatestTs = newest;
        }

        const fetchedThreadTs = new Set<string>();
        const threadWork: Array<{
          message: SlackMessage;
          threadTs: string;
          fetchPermalink: boolean;
        }> = [];
        for (const message of messages) {
          const threadTs = message.thread_ts ?? message.ts;
          if (!threadTs || fetchedThreadTs.has(threadTs)) continue;
          fetchedThreadTs.add(threadTs);
          const fetchPermalink = permalinkCalls < permalinkLimit;
          if (fetchPermalink) {
            permalinkCalls += 1;
          }
          threadWork.push({ message, threadTs, fetchPermalink });
        }

        for (
          let offset = 0;
          offset < threadWork.length;
          offset += threadCaptureConcurrency
        ) {
          const batch = threadWork.slice(
            offset,
            offset + threadCaptureConcurrency,
          );
          const results = await Promise.allSettled(
            batch.map(async ({ message, threadTs, fetchPermalink }) => {
              const permalink = fetchPermalink
                ? await slackPermalink(token, channel.id, threadTs)
                : null;
              const thread = await slackApi<SlackRepliesResponse>(
                token,
                "conversations.replies",
                { channel: channel.id, ts: threadTs, limit: 1_000 },
              );
              stats.threadsFetched = Number(stats.threadsFetched) + 1;
              return createSlackThreadCapture({
                source,
                channel,
                messages: thread.messages?.length ? thread.messages : [message],
                permalink,
                syncRunId: runId,
                memberEmails: privateMemberEmails,
              });
            }),
          );
          let batchFailed = false;
          let batchError: unknown;
          for (const result of results) {
            if (result.status === "rejected") {
              if (!batchFailed) batchError = result.reason;
              batchFailed = true;
              continue;
            }
            const captureResult = result.value;
            if (captureResult.capture) {
              captures.push(serializeCapture(captureResult.capture));
            }
            if (captureResult.blocked) {
              stats.sensitivityBlocked = Number(stats.sensitivityBlocked) + 1;
            }
          }
          if (batchFailed) throw batchError;
          await renewRunLease(run);
        }

        const nextPage = data.response_metadata?.next_cursor;
        if (data.has_more && nextPage) {
          channelCursor.pageCursor = nextPage;
          nextCursor.channels![channel.id] = channelCursor;
          if (page + 1 < pagesPerChannel) continue;
          break;
        }

        channelCursor.pageCursor = undefined;
        channelCursor.latestTs =
          channelCursor.pendingLatestTs ??
          pendingLatest ??
          channelCursor.latestTs;
        channelCursor.pendingLatestTs = undefined;
        nextCursor.channels![channel.id] = channelCursor;
        break;
      }
    }

    stats.capturesCreated = captures.length;
    await renewRunLease(run);
    await db
      .update(schema.brainSources)
      .set({
        cursorJson: stableJson(nextCursor),
        lastSyncedAt: nowIso(),
        lastError: null,
        status: "active",
        updatedAt: nowIso(),
      })
      .where(
        and(
          accessFilter(schema.brainSources, schema.brainSourceShares),
          eq(schema.brainSources.id, source.id),
        ),
      );
    await finishRun(run, "success", stats);
    return {
      runId,
      sourceId: source.id,
      provider: "slack",
      status: "success",
      capturesCreated: captures.length,
      captures,
      stats,
      message: captures.length
        ? `Imported ${captures.length} Slack messages`
        : "Slack sync completed with no new channel messages",
    };
  } catch (err) {
    const message = err instanceof Error ? err.message : String(err);
    const isRateLimit = err instanceof ConnectorRateLimitError;
    const failedCursor: SlackSyncCursor = {
      ...cursor,
      ...nextCursor,
      retry: isRateLimit ? retryCursor(err, "slack") : cursor.retry,
      lastRunAt: nowIso(),
    };
    stats.capturesCreated = captures.length;
    stats.rateLimited = isRateLimit;
    await renewRunLease(run);
    await db
      .update(schema.brainSources)
      .set({
        cursorJson: stableJson(failedCursor),
        lastError: message,
        status: isRateLimit ? "active" : "error",
        updatedAt: nowIso(),
      })
      .where(
        and(
          accessFilter(schema.brainSources, schema.brainSourceShares),
          eq(schema.brainSources.id, source.id),
        ),
      );
    await finishRun(
      run,
      isRateLimit ? "success" : "error",
      stats,
      isRateLimit ? null : message,
    );
    return {
      runId,
      sourceId: source.id,
      provider: "slack",
      status: isRateLimit ? "success" : "error",
      capturesCreated: captures.length,
      captures,
      stats,
      message,
    };
  }
}

export async function refreshSlackThreadCapture(
  source: SourceRow,
  payloadJson: string,
) {
  const payload = parseJson<Record<string, unknown>>(payloadJson, {});
  const channelId =
    typeof payload.channelId === "string" ? payload.channelId.trim() : "";
  const threadTs =
    typeof payload.threadTs === "string" ? payload.threadTs.trim() : "";
  if (!/^[CG][A-Z0-9]+$/i.test(channelId) || !slackTsFromDateish(threadTs)) {
    throw new Error("Slack thread refresh requires a channelId and threadTs");
  }

  const config = parseJson<Record<string, unknown>>(source.configJson, {});
  const includePublicChannels = configuredBoolean(
    config,
    ["includePublicChannels"],
    false,
    "slack",
  );
  const explicitlyConfigured = configuredSlackChannelIds(config).has(channelId);
  if (
    !explicitlyConfigured &&
    !(includePublicChannels && channelId.startsWith("C"))
  ) {
    throw new Error(
      `Slack thread refresh rejected for channel ${channelId}: it is not in this source's public or explicit channel scope`,
    );
  }
  const run = await createRun(source, { exclusive: false });
  const { runId } = run;
  try {
    const token = await requireConnectorCredential(
      "SLACK_BOT_TOKEN",
      "Slack",
      "slack",
      config,
    );
    const channel = await resolveSlackChannel(token, channelId);
    if (!channel || !isUsableSlackChannel(channel)) {
      throw new Error(`Slack channel ${channelId} is unavailable for refresh`);
    }
    if (
      !channel.is_private &&
      matchesPublicChannelExclusion(
        channel,
        (await readBrainSettings()).publicChannelExclusionPatterns ?? [],
      )
    ) {
      throw new Error(
        `Slack thread refresh rejected for channel ${channelId}: it matches a Brain public-channel exclusion`,
      );
    }
    const memberEmails = channel.is_private
      ? await slackPrivateChannelMemberEmails(token, channel.id, new Map())
      : null;
    if (channel.is_private && !memberEmails?.length) {
      throw new Error(
        `Slack private channel ${channel.id} has no resolvable member emails; refusing to refresh without an ACL`,
      );
    }
    const [thread, permalink] = await Promise.all([
      slackApi<SlackRepliesResponse>(token, "conversations.replies", {
        channel: channel.id,
        ts: threadTs,
        limit: 1_000,
      }),
      slackPermalink(token, channel.id, threadTs),
    ]);
    const captureResult = await createSlackThreadCapture({
      source,
      channel,
      messages: thread.messages ?? [],
      permalink,
      syncRunId: runId,
      memberEmails,
    });
    const stats = {
      channelId: channel.id,
      threadTs,
      threadsFetched: 1,
      capturesCreated: captureResult.capture ? 1 : 0,
      sensitivityBlocked: captureResult.blocked ? 1 : 0,
    };
    await renewRunLease(run);
    await getDb()
      .update(schema.brainSources)
      .set({
        lastSyncedAt: nowIso(),
        lastError: null,
        status: "active",
        updatedAt: nowIso(),
      })
      .where(
        and(
          accessFilter(schema.brainSources, schema.brainSourceShares),
          eq(schema.brainSources.id, source.id),
        ),
      );
    await finishRun(run, "success", stats);
    return {
      runId,
      capture: captureResult.capture
        ? serializeCapture(captureResult.capture)
        : null,
    };
  } catch (error) {
    const message = error instanceof Error ? error.message : String(error);
    await finishRun(run, "error", { capturesCreated: 0 }, message);
    throw error;
  }
}

async function syncGranola(source: SourceRow): Promise<ConnectorSyncResult> {
  const config = parseJson<Record<string, unknown>>(source.configJson, {});
  if (isFixtureConfig(config) || transcriptItems(config).length > 0) {
    return syncFromConfiguredItems(
      source,
      "Granola fixture source has no configured transcripts.",
    );
  }

  const run = await createRun(source);
  const { runId } = run;
  const db = getDb();
  const cursor = parseJson<GranolaSyncCursor>(source.cursorJson, {});
  const pageSize = configuredNumber(config, ["pageSize", "limit"], 10, {
    min: 1,
    max: 30,
    nestedKey: "granola",
  });
  const pagesPerSync = configuredNumber(
    config,
    ["pagesPerSync", "pageBudget"],
    1,
    { min: 1, max: 5, nestedKey: "granola" },
  );
  const noteCaptureConcurrency = configuredNumber(
    config,
    ["noteCaptureConcurrency", "captureConcurrency"],
    GRANOLA_NOTE_CAPTURE_CONCURRENCY,
    { min: 1, max: 8, nestedKey: "granola" },
  );
  const configuredUpdatedAfter =
    typeof config.updatedAfter === "string"
      ? config.updatedAfter
      : typeof objectValue(config.granola).updatedAfter === "string"
        ? String(objectValue(config.granola).updatedAfter)
        : undefined;
  const updatedAfter = cursor.updatedAfter ?? configuredUpdatedAfter;

  const captures = [];
  const stats: Record<string, unknown> = {
    notesSeen: 0,
    notesFetched: 0,
    noteCaptureConcurrency,
    capturesCreated: 0,
    rateLimited: false,
  };

  try {
    const token = await requireConnectorCredential(
      "GRANOLA_API_KEY",
      "Granola",
      "granola",
      config,
    );
    let nextPageCursor = cursor.cursor ?? undefined;
    let maxUpdatedAt = updatedAfter;

    for (let page = 0; page < pagesPerSync; page += 1) {
      const list = await granolaApi<GranolaListResponse>(token, "/notes", {
        page_size: pageSize,
        cursor: nextPageCursor,
        updated_after: nextPageCursor ? undefined : updatedAfter,
      });
      const notes = list.notes ?? [];
      stats.notesSeen = Number(stats.notesSeen) + notes.length;

      const fetchableNotes = notes.filter(
        (listed): listed is GranolaListNote & { id: string } =>
          Boolean(listed.id),
      );
      for (
        let offset = 0;
        offset < fetchableNotes.length;
        offset += noteCaptureConcurrency
      ) {
        const batch = fetchableNotes.slice(
          offset,
          offset + noteCaptureConcurrency,
        );
        const results = await Promise.allSettled(
          batch.map(async (listed) => {
            const note = await granolaApi<GranolaNote>(
              token,
              `/notes/${encodeURIComponent(listed.id)}`,
              { include: "transcript" },
            );
            stats.notesFetched = Number(stats.notesFetched) + 1;
            const combinedNote = {
              ...listed,
              ...note,
            };
            const normalized = normalizeGranolaNote(combinedNote);
            const calendar = objectValue(combinedNote.calendar_event);
            const memberEmails = resolveMeetingMemberEmails(
              [
                combinedNote.attendees,
                combinedNote.owner,
                calendar.invitees,
                calendar.organiser,
              ],
              source.ownerEmail,
            );
            const captureResult = await createConnectorCapture({
              sourceId: source.id,
              externalId: normalized.externalId,
              title: normalized.title,
              kind: "transcript",
              content: normalized.content,
              capturedAt: normalized.capturedAt,
              metadata: {
                ...normalized.metadata,
                connector: "granola",
                syncRunId: runId,
              },
              audience: {
                kind: "meeting",
                memberEmails,
                upstreamRefHash: normalized.externalId ?? listed.id,
              },
            });
            return {
              captureResult,
              candidateUpdatedAt: note.updated_at ?? listed.updated_at,
            };
          }),
        );
        let batchFailed = false;
        let batchError: unknown;
        for (const result of results) {
          if (result.status === "rejected") {
            if (!batchFailed) batchError = result.reason;
            batchFailed = true;
            continue;
          }
          const { captureResult, candidateUpdatedAt } = result.value;
          if (captureResult.capture) {
            captures.push(serializeCapture(captureResult.capture));
          }
          if (captureResult.blocked) {
            stats.sensitivityBlocked =
              Number(stats.sensitivityBlocked ?? 0) + 1;
          }
          if (
            candidateUpdatedAt &&
            (!maxUpdatedAt ||
              Date.parse(candidateUpdatedAt) > Date.parse(maxUpdatedAt))
          ) {
            maxUpdatedAt = candidateUpdatedAt;
          }
        }
        if (batchFailed) throw batchError;
        await renewRunLease(run);
      }

      const hasMore = list.hasMore ?? list.has_more ?? false;
      nextPageCursor = list.cursor ?? undefined;
      if (!hasMore || !nextPageCursor) {
        nextPageCursor = undefined;
        break;
      }
    }

    const nextCursor: GranolaSyncCursor = {
      cursor: nextPageCursor ?? null,
      updatedAfter: nextPageCursor
        ? updatedAfter
        : (maxUpdatedAt ?? new Date().toISOString()),
      retry: undefined,
      lastRunAt: nowIso(),
    };

    stats.capturesCreated = captures.length;
    await renewRunLease(run);
    await db
      .update(schema.brainSources)
      .set({
        cursorJson: stableJson(nextCursor),
        lastSyncedAt: nowIso(),
        lastError: null,
        status: "active",
        updatedAt: nowIso(),
      })
      .where(
        and(
          accessFilter(schema.brainSources, schema.brainSourceShares),
          eq(schema.brainSources.id, source.id),
        ),
      );
    await finishRun(run, "success", stats);
    return {
      runId,
      sourceId: source.id,
      provider: "granola",
      status: "success",
      capturesCreated: captures.length,
      captures,
      stats,
      message: captures.length
        ? `Imported ${captures.length} Granola notes`
        : "Granola sync completed with no new notes",
    };
  } catch (err) {
    const message = err instanceof Error ? err.message : String(err);
    const isRateLimit = err instanceof ConnectorRateLimitError;
    const nextCursor: GranolaSyncCursor = {
      ...cursor,
      retry: isRateLimit ? retryCursor(err, "granola") : cursor.retry,
      lastRunAt: nowIso(),
    };
    stats.capturesCreated = captures.length;
    stats.rateLimited = isRateLimit;
    await renewRunLease(run);
    await db
      .update(schema.brainSources)
      .set({
        cursorJson: stableJson(nextCursor),
        lastError: message,
        status: isRateLimit ? "active" : "error",
        updatedAt: nowIso(),
      })
      .where(
        and(
          accessFilter(schema.brainSources, schema.brainSourceShares),
          eq(schema.brainSources.id, source.id),
        ),
      );
    await finishRun(
      run,
      isRateLimit ? "success" : "error",
      stats,
      isRateLimit ? null : message,
    );
    return {
      runId,
      sourceId: source.id,
      provider: "granola",
      status: isRateLimit ? "success" : "error",
      capturesCreated: captures.length,
      captures,
      stats,
      message,
    };
  }
}

async function syncGitHub(source: SourceRow): Promise<ConnectorSyncResult> {
  const config = parseJson<Record<string, unknown>>(source.configJson, {});
  if (isFixtureConfig(config) || transcriptItems(config).length > 0) {
    return syncFromConfiguredItems(
      source,
      "GitHub fixture source has no configured captures.",
    );
  }

  const run = await createRun(source);
  const { runId } = run;
  const db = getDb();
  const cursor = parseJson<GitHubSyncCursor>(source.cursorJson, {});
  const nextCursor: GitHubSyncCursor = {
    ...cursor,
    repositories: { ...(cursor.repositories ?? {}) },
    retry: undefined,
    lastRunAt: nowIso(),
  };
  const repositories = githubReposFromConfig(config);
  const state = githubStateFromConfig(config);
  const limit = configuredNumber(config, ["limit", "pageSize", "perPage"], 25, {
    min: 1,
    max: 50,
    nestedKey: "github",
  });
  const detailLimit = configuredNumber(
    config,
    ["detailLimit", "repoDetailLimit", "maxContextItems"],
    0,
    { min: 0, max: 20, nestedKey: "github" },
  );
  const linkedDetailLimit = configuredNumber(
    config,
    ["linkedDetailLimit", "contextLimit", "maxLinkedContextItems"],
    10,
    { min: 0, max: 20, nestedKey: "github" },
  );
  const commentLimit = configuredNumber(
    config,
    ["commentLimit", "commentsPerItem"],
    5,
    { min: 0, max: 25, nestedKey: "github" },
  );
  const reviewLimit = configuredNumber(
    config,
    ["reviewLimit", "reviewsPerPullRequest"],
    5,
    { min: 0, max: 25, nestedKey: "github" },
  );
  const linkedCaptureLimit = configuredNumber(
    config,
    ["linkedCaptureLimit", "slackCaptureLimit"],
    100,
    { min: 1, max: 500, nestedKey: "github" },
  );
  const linkedRefLimit = configuredNumber(
    config,
    ["linkedRefLimit", "maxLinkedRefs"],
    20,
    { min: 1, max: 100, nestedKey: "github" },
  );
  const maxRepositories = configuredNumber(
    config,
    ["maxRepositories", "repoLimit"],
    5,
    { min: 1, max: 10, nestedKey: "github" },
  );
  const includeIssues = configuredBoolean(
    config,
    ["includeIssues"],
    true,
    "github",
  );
  const includePullRequests = configuredBoolean(
    config,
    ["includePullRequests", "includePRs"],
    true,
    "github",
  );

  const captures: any[] = [];
  const stats: Record<string, unknown> = {
    configuredRepositories: repositories.length,
    scannedRepositories: 0,
    itemsSeen: 0,
    issuesSeen: 0,
    pullRequestsSeen: 0,
    linkedSourceIds: 0,
    linkedSourcesSkipped: 0,
    linkedCapturesScanned: 0,
    linkedRefsFound: 0,
    linkedRefsImported: 0,
    detailsFetched: 0,
    commentsFetched: 0,
    reviewsFetched: 0,
    capturesCreated: 0,
    state,
    includeIssues,
    includePullRequests,
    rateLimited: false,
  };
  const heartbeat = startRunLeaseHeartbeat(run);

  try {
    const token = await requireConnectorCredential(
      "GITHUB_TOKEN",
      "GitHub",
      "github",
      config,
    );
    const linkedRefs = await githubRefsFromLinkedSources(config, {
      captureLimit: linkedCaptureLimit,
      refLimit: linkedRefLimit,
      stats,
    });

    if (!repositories.length && !linkedRefs.length) {
      throw new Error(
        "GitHub source must configure repositories/repos or linkedSlackSourceIds with Slack captures that contain GitHub issue or PR links",
      );
    }
    if (!includeIssues && !includePullRequests) {
      await heartbeat.stop();
      heartbeat.assertHealthy();
      await renewRunLease(run);
      await db
        .update(schema.brainSources)
        .set({
          cursorJson: stableJson(nextCursor),
          lastSyncedAt: nowIso(),
          lastError: null,
          status: "active",
          updatedAt: nowIso(),
        })
        .where(
          and(
            accessFilter(schema.brainSources, schema.brainSourceShares),
            eq(schema.brainSources.id, source.id),
          ),
        );
      await finishRun(run, "success", stats);
      return {
        runId,
        sourceId: source.id,
        provider: "github",
        status: "success",
        capturesCreated: 0,
        captures,
        stats,
        message:
          "GitHub sync skipped because includeIssues and includePullRequests are both false",
      };
    }

    const importedExternalIds = new Set<string>();
    let linkedDetailBudget = linkedDetailLimit;
    let detailBudget = detailLimit;

    for (const ref of linkedRefs) {
      const item = await githubApi<GitHubIssueOrPull>(
        token,
        `/repos/${ref.repo}/issues/${ref.number}`,
      );
      const isPullRequest = Boolean(item.pull_request);
      if (isPullRequest) {
        stats.pullRequestsSeen = Number(stats.pullRequestsSeen) + 1;
        if (!includePullRequests) continue;
      } else {
        stats.issuesSeen = Number(stats.issuesSeen) + 1;
        if (!includeIssues) continue;
      }

      let context: GitHubContext = { comments: [], reviews: [] };
      if (linkedDetailBudget > 0) {
        context = await fetchGitHubContext(token, ref.repo, item, {
          commentLimit,
          reviewLimit,
        });
        linkedDetailBudget -= 1;
        stats.detailsFetched = Number(stats.detailsFetched) + 1;
        stats.commentsFetched =
          Number(stats.commentsFetched) + context.comments.length;
        stats.reviewsFetched =
          Number(stats.reviewsFetched) + context.reviews.length;
      }

      const normalized = normalizeGitHubIssue(
        ref.repo,
        item,
        context,
        ref.linkedFrom,
      );
      importedExternalIds.add(normalized.externalId);
      const captureResult = await createConnectorCapture({
        sourceId: source.id,
        externalId: normalized.externalId,
        title: normalized.title,
        kind: normalized.kind,
        content: normalized.content,
        capturedAt: normalized.capturedAt,
        metadata: {
          ...normalized.metadata,
          syncRunId: runId,
        },
      });
      if (captureResult.capture)
        captures.push(serializeCapture(captureResult.capture));
      if (captureResult.blocked) {
        stats.sensitivityBlocked = Number(stats.sensitivityBlocked ?? 0) + 1;
      }
      stats.itemsSeen = Number(stats.itemsSeen) + 1;
      stats.linkedRefsImported = Number(stats.linkedRefsImported) + 1;
      heartbeat.assertHealthy();
    }

    for (const repo of repositories.slice(0, maxRepositories)) {
      stats.scannedRepositories = Number(stats.scannedRepositories) + 1;
      const repoCursor = nextCursor.repositories?.[repo] ?? {};
      const items = await githubApi<GitHubIssueOrPull[]>(
        token,
        `/repos/${repo}/issues`,
        {
          state,
          sort: "updated",
          direction: "desc",
          per_page: limit,
          since: repoCursor.updatedAfter,
        },
      );
      stats.itemsSeen = Number(stats.itemsSeen) + items.length;

      let maxUpdatedAt = repoCursor.updatedAfter;
      for (const item of items) {
        const isPullRequest = Boolean(item.pull_request);
        if (isPullRequest) {
          stats.pullRequestsSeen = Number(stats.pullRequestsSeen) + 1;
          if (!includePullRequests) continue;
        } else {
          stats.issuesSeen = Number(stats.issuesSeen) + 1;
          if (!includeIssues) continue;
        }
        const externalId = `github:${repo}:${isPullRequest ? "pull" : "issue"}:${item.number ?? item.id ?? "unknown"}`;
        if (importedExternalIds.has(externalId)) continue;

        let context: GitHubContext = { comments: [], reviews: [] };
        if (detailBudget > 0) {
          context = await fetchGitHubContext(token, repo, item, {
            commentLimit,
            reviewLimit,
          });
          detailBudget -= 1;
          stats.detailsFetched = Number(stats.detailsFetched) + 1;
          stats.commentsFetched =
            Number(stats.commentsFetched) + context.comments.length;
          stats.reviewsFetched =
            Number(stats.reviewsFetched) + context.reviews.length;
        }

        const normalized = normalizeGitHubIssue(repo, item, context);
        importedExternalIds.add(normalized.externalId);
        const captureResult = await createConnectorCapture({
          sourceId: source.id,
          externalId: normalized.externalId,
          title: normalized.title,
          kind: normalized.kind,
          content: normalized.content,
          capturedAt: normalized.capturedAt,
          metadata: {
            ...normalized.metadata,
            syncRunId: runId,
          },
        });
        if (captureResult.capture)
          captures.push(serializeCapture(captureResult.capture));
        if (captureResult.blocked) {
          stats.sensitivityBlocked = Number(stats.sensitivityBlocked ?? 0) + 1;
        }
        heartbeat.assertHealthy();

        const candidateUpdatedAt = item.updated_at;
        if (
          candidateUpdatedAt &&
          (!maxUpdatedAt ||
            Date.parse(candidateUpdatedAt) > Date.parse(maxUpdatedAt))
        ) {
          maxUpdatedAt = candidateUpdatedAt;
        }
      }
      nextCursor.repositories![repo] = {
        updatedAfter: maxUpdatedAt ?? nowIso(),
      };
    }

    stats.capturesCreated = captures.length;
    await heartbeat.stop();
    heartbeat.assertHealthy();
    await renewRunLease(run);
    await db
      .update(schema.brainSources)
      .set({
        cursorJson: stableJson(nextCursor),
        lastSyncedAt: nowIso(),
        lastError: null,
        status: "active",
        updatedAt: nowIso(),
      })
      .where(
        and(
          accessFilter(schema.brainSources, schema.brainSourceShares),
          eq(schema.brainSources.id, source.id),
        ),
      );
    await finishRun(run, "success", stats);
    return {
      runId,
      sourceId: source.id,
      provider: "github",
      status: "success",
      capturesCreated: captures.length,
      captures,
      stats,
      message: captures.length
        ? `Imported ${captures.length} GitHub issues and pull requests`
        : "GitHub sync completed with no new issues or pull requests",
    };
  } catch (err) {
    await heartbeat.stop();
    const message = err instanceof Error ? err.message : String(err);
    const isRateLimit = err instanceof ConnectorRateLimitError;
    const failedCursor: GitHubSyncCursor = {
      ...cursor,
      ...nextCursor,
      retry: isRateLimit ? retryCursor(err, "github") : cursor.retry,
      lastRunAt: nowIso(),
    };
    stats.capturesCreated = captures.length;
    stats.rateLimited = isRateLimit;
    await renewRunLease(run);
    await db
      .update(schema.brainSources)
      .set({
        cursorJson: stableJson(failedCursor),
        lastError: message,
        status: isRateLimit ? "active" : "error",
        updatedAt: nowIso(),
      })
      .where(
        and(
          accessFilter(schema.brainSources, schema.brainSourceShares),
          eq(schema.brainSources.id, source.id),
        ),
      );
    await finishRun(
      run,
      isRateLimit ? "success" : "error",
      stats,
      isRateLimit ? null : message,
    );
    return {
      runId,
      sourceId: source.id,
      provider: "github",
      status: isRateLimit ? "success" : "error",
      capturesCreated: captures.length,
      captures,
      stats,
      message,
    };
  }
}

const manualConnector: Connector = {
  sync: (source) =>
    syncFromConfiguredItems(
      source,
      "Manual sources do not have a remote sync. Use import-capture or import-transcript.",
    ),
};

const slackConnector: Connector = {
  sync: syncSlack,
};

const granolaConnector: Connector = {
  sync: syncGranola,
};

const githubConnector: Connector = {
  sync: syncGitHub,
};

const clipsConnector: Connector = {
  sync: (source) =>
    syncFromConfiguredItems(
      source,
      "Clips connector configured. Export from Clips or add transcript payloads to source config.",
    ),
};

const connectors: Record<BrainSourceProvider, Connector> = {
  manual: manualConnector,
  generic: manualConnector,
  clips: clipsConnector,
  slack: slackConnector,
  granola: granolaConnector,
  github: githubConnector,
};

export async function runConnectorSync(source: SourceRow) {
  await assertAccess("brain-source", source.id, "editor");
  try {
    return await connectors[source.provider as BrainSourceProvider].sync(
      source,
    );
  } catch (error) {
    if (!(error instanceof ConnectorSyncAlreadyRunningError)) throw error;
    return {
      runId: error.runId,
      sourceId: source.id,
      provider: source.provider as BrainSourceProvider,
      status: "success" as const,
      capturesCreated: 0,
      captures: [],
      stats: { alreadyRunning: true },
      message: error.message,
    };
  }
}