From 62ee045a4728da2d756be3d2e40d13e592a0b5a9 Mon Sep 17 00:00:00 2001 From: Dotta Date: Mon, 12 Oct 2026 08:30:42 -0500 Subject: [PATCH] refactor(chat): isolate Slack identity connect workflow Co-Authored-By: Paperclip --- .../2026-10-10-chat-channels-refactor.md | 16 +++ .../chat-channels/github/identity-links.ts | 7 ++ .../services/chat-channels/identity-links.ts | 92 +++----------- .../chat-channels/slack/identity-links.ts | 113 ++++++++++++++++++ 4 files changed, 151 insertions(+), 77 deletions(-) create mode 100644 server/src/services/chat-channels/github/identity-links.ts create mode 100644 server/src/services/chat-channels/slack/identity-links.ts diff --git a/doc/plans/2026-10-10-chat-channels-refactor.md b/doc/plans/2026-10-10-chat-channels-refactor.md index 77ba3c0305..ce51837f23 100644 --- a/doc/plans/2026-10-10-chat-channels-refactor.md +++ b/doc/plans/2026-10-10-chat-channels-refactor.md @@ -187,3 +187,19 @@ transaction locks, token consumption, membership checks and post-commit notice. body substitution moves the runtime lookup behind the supplied live callback. All remaining statements are unchanged. Local PostgreSQL remains unavailable; record this slice's complete CI and review in its PR before continuing. + +### 2026-10-12 channel-folder reassessment: slice 3 + +Keep identity-token consumption, principal scope, membership checks, link locks +and revocation in shared `identity-links.ts`. Move Slack receipt queries, +self-service access requests and confirmation-notice staging into +`slack/identity-links.ts`. GitHub identity fields live in `github/identity-links.ts`. +The shared confirmation transaction still stages the Slack notice before commit +and schedules it only after commit. The live runtime and public URL callbacks +remain live. The Slack factory holds no cache or worker state. + +Four existing public-service cases passed before and after, including the full +Slack connect journey: discovery, preview, outsider access request, duplicate +request, member confirmation, provider notice, consumed-token rejection and +endpoint isolation. Receipt queries, access-request transaction and notice +payload retain their statements with explicit parameter substitutions. diff --git a/server/src/services/chat-channels/github/identity-links.ts b/server/src/services/chat-channels/github/identity-links.ts new file mode 100644 index 0000000000..c5a60c87e8 --- /dev/null +++ b/server/src/services/chat-channels/github/identity-links.ts @@ -0,0 +1,7 @@ +import type { chatExternalPrincipals } from "@paperclipai/db"; + +export function githubIdentityFields( + principal: Pick, +) { + return { githubUserId: principal.externalId, githubLogin: principal.handle }; +} diff --git a/server/src/services/chat-channels/identity-links.ts b/server/src/services/chat-channels/identity-links.ts index 2566966b11..47596405f3 100644 --- a/server/src/services/chat-channels/identity-links.ts +++ b/server/src/services/chat-channels/identity-links.ts @@ -1,33 +1,28 @@ +import { githubIdentityFields } from "./github/identity-links.js"; +import { createSlackIdentityLinks, type SlackIdentityEffectStager } from "./slack/identity-links.js"; import { createHash, randomBytes } from "node:crypto"; -import { and, asc, desc, eq, gt, inArray, ne, or, sql } from "drizzle-orm"; +import { and, asc, eq, gt, inArray, ne, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; -import { authUsers, chatActions, chatEndpoints, chatExternalPrincipals, chatIdentityLinks, companies, companyMemberships, invites, joinRequests } from "@paperclipai/db"; +import { authUsers, chatEndpoints, chatExternalPrincipals, chatIdentityLinks, companies, companyMemberships } from "@paperclipai/db"; import { conflict, forbidden, notFound, unprocessable } from "../../errors.js"; -import { logActivity } from "../activity-log.js"; - type EndpointRow = typeof chatEndpoints.$inferSelect; -type DbOrTransaction = Db | Parameters[0]>[0]; type ConfirmationRuntime = { credentialFingerprint: string; generation: number }; type ChatIdentityLinkOptions = { endpointRecord: (endpointId: string) => Promise<{ endpoint: EndpointRow } | null>; getPublicBaseUrl: () => string | null; confirmationRuntimeFor: (endpoint: EndpointRow) => Promise; - stageProviderEffect: (database: DbOrTransaction, input: { - endpoint: EndpointRow; principalId: string; providerActionId: string; - payload: { version: 1; effect: "ephemeral_message"; authorizationMode: "safe_notice"; threadId: string; userId: string; text: string; settleDelivery: false }; - runtimeContext: ConfirmationRuntime; - }) => Promise<{ id: string } | null>; + stageProviderEffect: SlackIdentityEffectStager; scheduleProviderEffect: (actionId: string) => void; }; export function createChatIdentityLinks(db: Db, options: ChatIdentityLinkOptions) { const { endpointRecord, getPublicBaseUrl, confirmationRuntimeFor, stageProviderEffect, scheduleProviderEffect } = options; - + const slackIdentityLinks = createSlackIdentityLinks(db, { previewIdentityLink, stageProviderEffect }); async function listPrincipals(endpointId: string) { const record = await endpointRecord(endpointId); @@ -50,16 +45,7 @@ export function createChatIdentityLinks(db: Db, options: ChatIdentityLinkOptions .select() .from(chatIdentityLinks) .where(eq(chatIdentityLinks.endpointId, endpointId)); - const connects = record.endpoint.provider === "slack" ? await db - .select({ principalId: chatActions.principalId, lastConnectAt: sql`max(${chatActions.createdAt})::text` }) - .from(chatActions) - .where(and( - eq(chatActions.companyId, record.endpoint.companyId), - eq(chatActions.endpointId, endpointId), - eq(chatActions.kind, "slack_connect"), - eq(chatActions.status, "processed"), - )) - .groupBy(chatActions.principalId) : []; + const connects = record.endpoint.provider === "slack" ? await slackIdentityLinks.listConnects(record.endpoint) : []; const connectByPrincipal = new Map(connects.map((row) => [row.principalId, new Date(row.lastConnectAt).toISOString()])); const userIds = links.flatMap((link) => link.paperclipUserId ? [link.paperclipUserId] : [], @@ -86,7 +72,7 @@ export function createChatIdentityLinks(db: Db, options: ChatIdentityLinkOptions return { id: link?.id ?? principal.id, principalId: principal.id, - ...(record.endpoint.provider === "github" ? { githubUserId: principal.externalId, githubLogin: principal.handle } : {}), + ...(record.endpoint.provider === "github" ? githubIdentityFields(principal) : {}), externalLabel: principal.displayName ?? principal.handle ?? principal.externalId, externalDetail: principal.handle @@ -216,11 +202,7 @@ export function createChatIdentityLinks(db: Db, options: ChatIdentityLinkOptions throw unprocessable("This identity-link request is invalid or expired"); } if (!["active", "verifying"].includes(row.endpoint.status)) throw unprocessable("This connection is not available"); - const selfService = Boolean(await db.select({ id: chatActions.id }).from(chatActions).where(and( - eq(chatActions.companyId, row.link.companyId), eq(chatActions.endpointId, row.link.endpointId), - eq(chatActions.principalId, row.link.principalId), eq(chatActions.kind, "slack_connect"), - sql`${chatActions.payload}->>'identityLinkHash' = ${tokenHash}`, - )).limit(1).then((rows) => rows[0])); + const selfService = await slackIdentityLinks.isSelfService(row.link, tokenHash); const membership = userId ? await db.select({ status: companyMemberships.status }).from(companyMemberships).where(and( eq(companyMemberships.companyId, row.link.companyId), eq(companyMemberships.principalType, "user"), eq(companyMemberships.principalId, userId), )).then((rows) => rows[0]) : null; @@ -245,43 +227,6 @@ export function createChatIdentityLinks(db: Db, options: ChatIdentityLinkOptions }; } - async function requestIdentityAccess(token: string, userId: string, requestIp: string) { - const preview = await previewIdentityLink(token, userId); - if (!preview.selfService) throw forbidden("This identity link cannot request access"); - if (preview.canConfirm) return { status: "member" as const }; - return db.transaction(async (tx) => { - await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${`chat-join:${preview.companyId}:${userId}`}, 0))`); - const tokenHash = createHash("sha256").update(token).digest("hex"); - const validLink = await tx.select({ id: chatIdentityLinks.id }).from(chatIdentityLinks).where(and( - eq(chatIdentityLinks.companyId, preview.companyId), eq(chatIdentityLinks.endpointId, preview.endpointId), - eq(chatIdentityLinks.confirmationTokenHash, tokenHash), eq(chatIdentityLinks.status, "pending"), gt(chatIdentityLinks.expiresAt, new Date()), - )).for("update").then((rows) => rows[0]); - if (!validLink) throw unprocessable("This identity-link request is invalid or expired"); - const user = await tx.select({ email: authUsers.email }).from(authUsers).where(eq(authUsers.id, userId)).then((rows) => rows[0]); - if (!user) throw forbidden("Sign in to request company access"); - const existing = await tx.select({ id: joinRequests.id }).from(joinRequests).where(and( - eq(joinRequests.companyId, preview.companyId), eq(joinRequests.requestType, "human"), eq(joinRequests.status, "pending_approval"), - or(eq(joinRequests.requestingUserId, userId), sql`lower(${joinRequests.requestEmailSnapshot}) = ${user.email.toLowerCase()}`), - )).then((rows) => rows[0]); - if (existing) return { status: "pending_approval" as const }; - const now = new Date(); - const [invite] = await tx.insert(invites).values({ - companyId: preview.companyId, tokenHash: createHash("sha256").update(randomBytes(32)).digest("hex"), - inviteType: "company_join", allowedJoinTypes: "human", acceptedAt: now, expiresAt: now, - defaultsPayload: { human: { role: "operator" }, source: "slack_identity_link" }, - }).returning({ id: invites.id }); - const [request] = await tx.insert(joinRequests).values({ - inviteId: invite.id, companyId: preview.companyId, requestType: "human", status: "pending_approval", - requestIp, requestingUserId: userId, requestEmailSnapshot: user.email, - }).returning({ id: joinRequests.id }); - await logActivity(tx as unknown as Db, { - companyId: preview.companyId, actorType: "user", actorId: userId, action: "join.requested", - entityType: "join_request", entityId: request.id, details: { requestType: "human", source: "slack_identity_link", endpointId: preview.endpointId }, - }); - return { status: "pending_approval" as const }; - }); - } - async function confirmIdentityLink(token: string, paperclipUserId: string) { const tokenHash = createHash("sha256").update(token).digest("hex"); const link = await db @@ -298,9 +243,7 @@ export function createChatIdentityLinks(db: Db, options: ChatIdentityLinkOptions throw unprocessable("This identity-link request is invalid or expired"); const endpointRecordForLink = await endpointRecord(link.endpointId); if (!endpointRecordForLink || !["active", "verifying"].includes(endpointRecordForLink.endpoint.status)) throw unprocessable("This connection is not available"); - const connectReceipt = endpointRecordForLink.endpoint.provider === "slack" ? await db.select({ payload: chatActions.payload }).from(chatActions).where(and( - eq(chatActions.endpointId, link.endpointId), eq(chatActions.principalId, link.principalId), eq(chatActions.kind, "slack_connect"), - )).orderBy(desc(chatActions.createdAt)).limit(1).then((rows) => rows[0]) : null; + const connectReceipt = endpointRecordForLink.endpoint.provider === "slack" ? await slackIdentityLinks.latestConnect(link) : null; const confirmationRuntime = connectReceipt ? await confirmationRuntimeFor(endpointRecordForLink.endpoint) : null; let confirmationEffectId: string | null = null; const confirmed = await db.transaction(async (tx) => { @@ -395,16 +338,11 @@ export function createChatIdentityLinks(db: Db, options: ChatIdentityLinkOptions .then((rows) => rows[0] ?? null); if (confirmedLink && connectReceipt && confirmationRuntime && typeof connectReceipt.payload.channelId === "string" && typeof connectReceipt.payload.userId === "string") { - const effect = await stageProviderEffect(tx, { - endpoint: endpointRecordForLink.endpoint, principalId: link.principalId, - providerActionId: `provider_effect:identity_linked:${link.id}:${tokenHash}`, - payload: { version: 1, effect: "ephemeral_message", authorizationMode: "safe_notice", - threadId: connectReceipt.payload.channelId, userId: connectReceipt.payload.userId, - text: "Your Slack account is connected to Paperclip. Future messages use your Paperclip permissions.", settleDelivery: false }, - runtimeContext: confirmationRuntime, + confirmationEffectId = await slackIdentityLinks.stageConfirmation(tx, { + endpoint: endpointRecordForLink.endpoint, link, tokenHash, + channelId: connectReceipt.payload.channelId, userId: connectReceipt.payload.userId, + confirmationRuntime, }); - if (!effect) throw conflict("The Slack connection changed; try confirming again"); - confirmationEffectId = effect.id; } return confirmedLink; }); @@ -455,5 +393,5 @@ export function createChatIdentityLinks(db: Db, options: ChatIdentityLinkOptions if (!revoked) throw notFound("Identity link not found"); } - return { listPrincipals, createLinkIntent, previewIdentityLink, requestIdentityAccess, confirmIdentityLink, revokeLink }; + return { listPrincipals, createLinkIntent, previewIdentityLink, requestIdentityAccess: slackIdentityLinks.requestIdentityAccess, confirmIdentityLink, revokeLink }; } diff --git a/server/src/services/chat-channels/slack/identity-links.ts b/server/src/services/chat-channels/slack/identity-links.ts new file mode 100644 index 0000000000..1a37771fa0 --- /dev/null +++ b/server/src/services/chat-channels/slack/identity-links.ts @@ -0,0 +1,113 @@ +import { createHash, randomBytes } from "node:crypto"; +import { and, desc, eq, gt, or, sql } from "drizzle-orm"; + +import type { Db } from "@paperclipai/db"; + +import { authUsers, chatActions, chatEndpoints, chatIdentityLinks, invites, joinRequests } from "@paperclipai/db"; + +import { conflict, forbidden, unprocessable } from "../../../errors.js"; + +import { logActivity } from "../../activity-log.js"; + +type EndpointRow = typeof chatEndpoints.$inferSelect; +type DbOrTransaction = Db | Parameters[0]>[0]; +type ConfirmationRuntime = { credentialFingerprint: string; generation: number }; +export type SlackIdentityEffectStager = (database: DbOrTransaction, input: { + endpoint: EndpointRow; principalId: string; providerActionId: string; + payload: { version: 1; effect: "ephemeral_message"; authorizationMode: "safe_notice"; threadId: string; userId: string; text: string; settleDelivery: false }; + runtimeContext: ConfirmationRuntime; +}) => Promise<{ id: string } | null>; + +type SlackIdentityLinkOptions = { + previewIdentityLink: (token: string, userId: string) => Promise<{ selfService: boolean; canConfirm: boolean; companyId: string; endpointId: string }>; + stageProviderEffect: SlackIdentityEffectStager; +}; + +export function createSlackIdentityLinks(db: Db, { previewIdentityLink, stageProviderEffect }: SlackIdentityLinkOptions) { + function listConnects(endpoint: EndpointRow) { + const endpointId = endpoint.id; + return db + .select({ principalId: chatActions.principalId, lastConnectAt: sql`max(${chatActions.createdAt})::text` }) + .from(chatActions) + .where(and( + eq(chatActions.companyId, endpoint.companyId), + eq(chatActions.endpointId, endpointId), + eq(chatActions.kind, "slack_connect"), + eq(chatActions.status, "processed"), + )) + .groupBy(chatActions.principalId); + } + + async function isSelfService(link: typeof chatIdentityLinks.$inferSelect, tokenHash: string) { + const selfService = Boolean(await db.select({ id: chatActions.id }).from(chatActions).where(and( + eq(chatActions.companyId, link.companyId), eq(chatActions.endpointId, link.endpointId), + eq(chatActions.principalId, link.principalId), eq(chatActions.kind, "slack_connect"), + sql`${chatActions.payload}->>'identityLinkHash' = ${tokenHash}`, + )).limit(1).then((rows) => rows[0])); + return selfService; + } + + function latestConnect(link: typeof chatIdentityLinks.$inferSelect) { + return db.select({ payload: chatActions.payload }).from(chatActions).where(and( + eq(chatActions.endpointId, link.endpointId), eq(chatActions.principalId, link.principalId), eq(chatActions.kind, "slack_connect"), + )).orderBy(desc(chatActions.createdAt)).limit(1).then((rows) => rows[0]); + } + + async function stageConfirmation(tx: DbOrTransaction, input: { + endpoint: EndpointRow; link: typeof chatIdentityLinks.$inferSelect; tokenHash: string; + channelId: string; userId: string; confirmationRuntime: ConfirmationRuntime; + }) { + const { endpoint, link, tokenHash, channelId, userId, confirmationRuntime } = input; + const effect = await stageProviderEffect(tx, { + endpoint: endpoint, principalId: link.principalId, + providerActionId: `provider_effect:identity_linked:${link.id}:${tokenHash}`, + payload: { + version: 1, effect: "ephemeral_message", authorizationMode: "safe_notice", + threadId: channelId, userId: userId, + text: "Your Slack account is connected to Paperclip. Future messages use your Paperclip permissions.", settleDelivery: false + }, + runtimeContext: confirmationRuntime, + }); + if (!effect) throw conflict("The Slack connection changed; try confirming again"); + return effect.id; + } + + async function requestIdentityAccess(token: string, userId: string, requestIp: string) { + const preview = await previewIdentityLink(token, userId); + if (!preview.selfService) throw forbidden("This identity link cannot request access"); + if (preview.canConfirm) return { status: "member" as const }; + return db.transaction(async (tx) => { + await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${`chat-join:${preview.companyId}:${userId}`}, 0))`); + const tokenHash = createHash("sha256").update(token).digest("hex"); + const validLink = await tx.select({ id: chatIdentityLinks.id }).from(chatIdentityLinks).where(and( + eq(chatIdentityLinks.companyId, preview.companyId), eq(chatIdentityLinks.endpointId, preview.endpointId), + eq(chatIdentityLinks.confirmationTokenHash, tokenHash), eq(chatIdentityLinks.status, "pending"), gt(chatIdentityLinks.expiresAt, new Date()), + )).for("update").then((rows) => rows[0]); + if (!validLink) throw unprocessable("This identity-link request is invalid or expired"); + const user = await tx.select({ email: authUsers.email }).from(authUsers).where(eq(authUsers.id, userId)).then((rows) => rows[0]); + if (!user) throw forbidden("Sign in to request company access"); + const existing = await tx.select({ id: joinRequests.id }).from(joinRequests).where(and( + eq(joinRequests.companyId, preview.companyId), eq(joinRequests.requestType, "human"), eq(joinRequests.status, "pending_approval"), + or(eq(joinRequests.requestingUserId, userId), sql`lower(${joinRequests.requestEmailSnapshot}) = ${user.email.toLowerCase()}`), + )).then((rows) => rows[0]); + if (existing) return { status: "pending_approval" as const }; + const now = new Date(); + const [invite] = await tx.insert(invites).values({ + companyId: preview.companyId, tokenHash: createHash("sha256").update(randomBytes(32)).digest("hex"), + inviteType: "company_join", allowedJoinTypes: "human", acceptedAt: now, expiresAt: now, + defaultsPayload: { human: { role: "operator" }, source: "slack_identity_link" }, + }).returning({ id: invites.id }); + const [request] = await tx.insert(joinRequests).values({ + inviteId: invite.id, companyId: preview.companyId, requestType: "human", status: "pending_approval", + requestIp, requestingUserId: userId, requestEmailSnapshot: user.email, + }).returning({ id: joinRequests.id }); + await logActivity(tx as unknown as Db, { + companyId: preview.companyId, actorType: "user", actorId: userId, action: "join.requested", + entityType: "join_request", entityId: request.id, details: { requestType: "human", source: "slack_identity_link", endpointId: preview.endpointId }, + }); + return { status: "pending_approval" as const }; + }); + } + + return { listConnects, isSelfService, latestConnect, stageConfirmation, requestIdentityAccess }; +}