diff --git a/doc/plans/2026-10-10-chat-channels-refactor.md b/doc/plans/2026-10-10-chat-channels-refactor.md index 9657164c415..2b6765a72ea 100644 --- a/doc/plans/2026-10-10-chat-channels-refactor.md +++ b/doc/plans/2026-10-10-chat-channels-refactor.md @@ -150,3 +150,25 @@ function, generation reader and error text limit. It creates no worker or cache. - Local baseline and after selections could not start PostgreSQL and skipped the suite. They are not passes. Server typecheck and module boundaries pass. - Require complete chat CI on this slice before starting the next extraction. + +### 2026-10-12 channel-folder reassessment: slice 2 + +PR #15823 was merged at the user's request. Rebase this slice onto the merged +master, then retain the remaining four PRs for review in dependency order. + +- GitHub payload parsing, repository coordinate repair, webhook recovery, + repository pagination and bulk-selection entry points live in + `chat-channels/github/resources.ts`. +- Telegram migration preparation and conversation migration live in + `telegram/resources.ts`. They receive the existing transaction and timestamp; + the shared coordinator still upserts the destination between those steps. +- Slack fallback labels, Teams conversation IDs and Photon participant parsing + live in each provider's `resources.ts`. +- Shared inventory replacement, availability settlement and audited selection + retain their transaction and credential-lease ownership. Small provider + dispatch predicates remain next to the shared operation they select. +- All eight moved declarations and three transaction blocks retain their logic. + The GitHub provider guard stays in the caller. No factory starts background work. +- Existing public-service tests: 15 resource cases and the Telegram migration + case passed before; the same 16 cases passed after. The name filter excludes + unrelated cases. No live third-party provider calls were made. diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index 133288af6c6..a0b5c93324c 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -1,4 +1,8 @@ -import { createChatResourceReconciliation, githubRepositoryInventoryItemFromPayload, baseTeamsConversationId, slackResourceLabelIsFallback, telegramResourceLabelIsFallback } from "./chat-channels/resource-reconciliation.js"; +import { createChatResourceReconciliation } from "./chat-channels/resource-reconciliation.js"; +import { githubRepositoryInventoryItemFromPayload } from "./chat-channels/github/resources.js"; +import { baseTeamsConversationId } from "./chat-channels/microsoft-teams/resources.js"; +import { slackResourceLabelIsFallback } from "./chat-channels/slack/resources.js"; +import { telegramResourceLabelIsFallback } from "./chat-channels/telegram/resources.js"; import { MAX_INBOUND_TEXT, stableExternalPrincipalId, diff --git a/server/src/services/chat-channels/github/resources.ts b/server/src/services/chat-channels/github/resources.ts new file mode 100644 index 00000000000..b719053d6c0 --- /dev/null +++ b/server/src/services/chat-channels/github/resources.ts @@ -0,0 +1,278 @@ +import { and,asc,eq,sql } from "drizzle-orm"; +import { chatConversations,chatEndpointResources,chatEndpoints,type Db } from "@paperclipai/db"; +import { conflict,notFound } from "../../../errors.js"; +import type { CredentialMutationLeaseGuard } from "../../chat-credential-mutation-lease.js"; +import type { ChatProviderResourceInventoryItem } from "../../chat-provider-inventory.js"; +type EndpointRow=typeof chatEndpoints.$inferSelect; +type DbTransaction=Parameters[0]>[0]; + +export function githubRepositoryInventoryItemFromPayload( + payload: unknown, +): ChatProviderResourceInventoryItem|null { + if(!payload||typeof payload!=="object"||Array.isArray(payload)) + return null; + const repository=(payload as { repository?: unknown }).repository; + if( + !repository|| + typeof repository!=="object"|| + Array.isArray(repository) + ) + return null; + const value=repository as { + id?: unknown; + full_name?: unknown; + html_url?: unknown; + name?: unknown; + owner?: { id?: unknown; login?: unknown }; + private?: unknown; + }; + const repositoryId= + typeof value.id==="number"&&Number.isSafeInteger(value.id) + ? String(value.id) + :typeof value.id==="string"&&/^\d+$/.test(value.id) + ? value.id + :null; + const owner= + typeof value.owner?.login==="string"? value.owner.login.trim():""; + const name=typeof value.name==="string"? value.name.trim():""; + const fullName= + typeof value.full_name==="string"&&value.full_name.includes("/") + ? value.full_name.trim() + :owner&&name + ? `${owner}/${name}` + :""; + if(!repositoryId||!fullName) return null; + const ownerId= + typeof value.owner?.id==="number"&&Number.isSafeInteger(value.owner.id) + ? String(value.owner.id) + :typeof value.owner?.id==="string"&&/^\d+$/.test(value.owner.id) + ? value.owner.id + :undefined; + return { + providerResourceId: fullName.toLowerCase(), + parentProviderResourceId: ownerId, + type: "repository", + label: fullName, + providerUrl: + typeof value.html_url==="string"&&value.html_url.length>0 + ? value.html_url + :`https://github.com/${fullName}`, + metadata: { + providerRepositoryId: repositoryId, + fullName, + ...(owner? { owner }:{}), + private: value.private===true, + source: "provider_webhook", + }, + }; +} + +function githubRepositoryStableId( + item: ChatProviderResourceInventoryItem, +): string|null { + const value=item.metadata?.providerRepositoryId; + return typeof value==="string"&&/^\d+$/.test(value)? value:null; +} + +export async function reconcileGitHubRepositoryCoordinates(tx: DbTransaction,endpoint: EndpointRow,item: ChatProviderResourceInventoryItem) { + const stableGitHubId= + githubRepositoryStableId(item); + if(stableGitHubId) { + await tx.execute( + sql`select pg_advisory_xact_lock(hashtextextended(${`chat-github-repository:${endpoint.id}:${stableGitHubId}`}, 0))`, + ); + const stableResource=await tx + .select() + .from(chatEndpointResources) + .where( + and( + eq(chatEndpointResources.companyId,endpoint.companyId), + eq(chatEndpointResources.endpointId,endpoint.id), + eq(chatEndpointResources.type,"repository"), + sql`${chatEndpointResources.metadata}->>'providerRepositoryId' = ${stableGitHubId}`, + ), + ) + .for("update") + .then((rows) => rows[0]??null); + if( + stableResource&& + stableResource.providerResourceId!==item.providerResourceId + ) { + const coordinateResource=await tx + .select() + .from(chatEndpointResources) + .where( + and( + eq(chatEndpointResources.companyId,endpoint.companyId), + eq(chatEndpointResources.endpointId,endpoint.id), + eq(chatEndpointResources.type,"repository"), + eq( + chatEndpointResources.providerResourceId, + item.providerResourceId, + ), + ), + ) + .for("update") + .then((rows) => rows[0]??null); + if(coordinateResource&&coordinateResource.id!==stableResource.id) { + const coordinateConversation=await tx + .select({ id: chatConversations.id }) + .from(chatConversations) + .where( + and( + eq(chatConversations.endpointId,endpoint.id), + eq(chatConversations.resourceId,coordinateResource.id), + ), + ) + .limit(1) + .then((rows) => rows[0]??null); + if(coordinateConversation) { + throw conflict( + "GitHub repository identity is already bound to two coordinates", + { code: "chat_github_repository_identity_conflict" }, + ); + } + await tx + .delete(chatEndpointResources) + .where(eq(chatEndpointResources.id,coordinateResource.id)); + } + + const oldRepository=stableResource.providerResourceId; + const oldThreadPrefix=`github:${oldRepository}`; + const newThreadPrefix=`github:${item.providerResourceId}`; + const oldProviderUrl=`https://github.com/${oldRepository}`; + const newProviderUrl=`https://github.com/${item.providerResourceId}`; + await tx + .update(chatEndpointResources) + .set({ + providerResourceId: item.providerResourceId, + parentProviderResourceId: item.parentProviderResourceId??null, + label: item.label, + providerUrl: item.providerUrl??null, + availability: "available", + enabled: + stableResource.enabled||coordinateResource?.enabled===true, + metadata: item.metadata??{}, + updatedAt: new Date(), + }) + .where(eq(chatEndpointResources.id,stableResource.id)); + await tx + .update(chatConversations) + .set({ + externalConversationId: sql`case + when lower(${chatConversations.externalConversationId}) = ${oldRepository} then ${item.providerResourceId} + when lower(${chatConversations.externalConversationId}) = ${oldThreadPrefix} then ${newThreadPrefix} + else ${chatConversations.externalConversationId} + end`, + externalThreadId: sql`case + when lower(${chatConversations.externalThreadId}) = ${oldThreadPrefix} then ${newThreadPrefix} + when lower(${chatConversations.externalThreadId}) like ${`${oldThreadPrefix}:%`} + then ${newThreadPrefix} || substring( + ${chatConversations.externalThreadId} + from char_length(${oldThreadPrefix}) + 1 + ) + else ${chatConversations.externalThreadId} + end`, + externalLabel: item.label, + providerUrl: sql`case + when lower(${chatConversations.providerUrl}) like ${`${oldProviderUrl}/%`} + then ${newProviderUrl} || substring( + ${chatConversations.providerUrl} + from char_length(${oldProviderUrl}) + 1 + ) + else ${chatConversations.providerUrl} + end`, + updatedAt: new Date(), + }) + .where( + and( + eq(chatConversations.companyId,endpoint.companyId), + eq(chatConversations.endpointId,endpoint.id), + eq(chatConversations.resourceId,stableResource.id), + ), + ); + return { id: stableResource.id }; + } + } + + return null; +} + +type GitHubResourceOptions={ + endpointRecord: (endpointId: string) => Promise<{ endpoint: EndpointRow }|null>; + upsertProviderResourceRow: (tx: DbTransaction,endpoint: EndpointRow,item: ChatProviderResourceInventoryItem) => Promise<{ id: string }|null>; + updateResourceSelection: (endpointId: string,updates: Array<{ id: string; enabled: boolean }>,actorUserId?: string|null,options?: { allGitHubRepositoriesEnabled?: boolean }) => Promise; +}; + +export function createGitHubResources(db: Db,options: GitHubResourceOptions) { + const { endpointRecord,upsertProviderResourceRow,updateResourceSelection }=options; + async function reconcileGitHubWebhookRepository( + endpoint: EndpointRow, + payload: unknown, + credentialLease: CredentialMutationLeaseGuard, + ) { + const item=githubRepositoryInventoryItemFromPayload(payload); + if(!item) return; + await db.transaction(async (tx) => { + await credentialLease.assertOwned(tx); + const resource=await upsertProviderResourceRow(tx,endpoint,item); + if(!resource) return; + // A correctly signed repository callback is current provider proof. Keep + // Paperclip's enabled choice, and reopen only conversations previously + // quarantined for provider availability loss. + await tx + .update(chatConversations) + .set({ state: "active",updatedAt: new Date() }) + .where( + and( + eq(chatConversations.companyId,endpoint.companyId), + eq(chatConversations.endpointId,endpoint.id), + eq(chatConversations.resourceId,resource.id), + eq(chatConversations.state,"unavailable"), + ), + ); + await credentialLease.assertOwned(tx); + }); + } + + async function listGitHubRepositories( + endpointId: string, + query: { limit: number; offset: number; search: string }, + ) { + const record=await endpointRecord(endpointId); + if(!record||record.endpoint.provider!=="github") throw notFound("GitHub bot not found"); + const scope=and( + eq(chatEndpointResources.companyId,record.endpoint.companyId), + eq(chatEndpointResources.endpointId,endpointId), + eq(chatEndpointResources.type,"repository"), + ); + // Literal, case-insensitive substring search (%, _ and backslashes are + // ordinary characters). Counts always describe the entire connection. + const search=query.search + ? sql`strpos(lower(${chatEndpointResources.label}), lower(${query.search})) > 0` + :undefined; + const [rows,[counts]]=await Promise.all([ + db.select().from(chatEndpointResources).where(and(scope,search)) + .orderBy(asc(chatEndpointResources.label),asc(chatEndpointResources.id)) + .offset(query.offset).limit(query.limit+1), + db.select({ + totalCount: sql`count(*)::integer`, + enabledCount: sql`count(*) filter (where ${chatEndpointResources.enabled})::integer`, + availableCount: sql`count(*) filter (where ${chatEndpointResources.availability} = 'available')::integer`, + }).from(chatEndpointResources).where(scope), + ]); + return { + items: rows.slice(0,query.limit), + nextOffset: rows.length>query.limit? query.offset+query.limit:null, + totalCount: counts!.totalCount, + enabledCount: counts!.enabledCount, + availableCount: counts!.availableCount, + }; + } + + async function toggleAllGitHubRepositories(endpointId: string,enabled: boolean,actorUserId?: string|null) { + await updateResourceSelection(endpointId,[],actorUserId,{ allGitHubRepositoriesEnabled: enabled }); + return { success: true as const }; + } + return { reconcileGitHubWebhookRepository,listGitHubRepositories,toggleAllGitHubRepositories }; +} diff --git a/server/src/services/chat-channels/imessage-photon/resources.ts b/server/src/services/chat-channels/imessage-photon/resources.ts new file mode 100644 index 00000000000..a6ee2e58596 --- /dev/null +++ b/server/src/services/chat-channels/imessage-photon/resources.ts @@ -0,0 +1,14 @@ +import type { chatEndpointResources } from "@paperclipai/db"; + +export function imessageResourceParticipants(resource: typeof chatEndpointResources.$inferSelect) { + return resource.providerResourceId.startsWith("imessage-photon:")&& + Array.isArray(resource.metadata.participants) + ? resource.metadata.participants.flatMap((participant) => { + const address= + participant&&typeof participant==="object" + ? (participant as Record).address + :null; + return typeof address==="string"? [address]:[]; + }) + :undefined; +} diff --git a/server/src/services/chat-channels/microsoft-teams/resources.ts b/server/src/services/chat-channels/microsoft-teams/resources.ts new file mode 100644 index 00000000000..675245b00e5 --- /dev/null +++ b/server/src/services/chat-channels/microsoft-teams/resources.ts @@ -0,0 +1,3 @@ +export function baseTeamsConversationId(value: string): string { + return value.replace(/;messageid=[^;]+/i, ""); +} diff --git a/server/src/services/chat-channels/resource-reconciliation.ts b/server/src/services/chat-channels/resource-reconciliation.ts index b7f53a0ef0d..ac0d1069d73 100644 --- a/server/src/services/chat-channels/resource-reconciliation.ts +++ b/server/src/services/chat-channels/resource-reconciliation.ts @@ -1,3 +1,8 @@ +import { imessageResourceParticipants } from "./imessage-photon/resources.js"; +import { createGitHubResources, reconcileGitHubRepositoryCoordinates } from "./github/resources.js"; +import { baseTeamsConversationId } from "./microsoft-teams/resources.js"; +import { slackResourceLabelIsFallback } from "./slack/resources.js"; +import { telegramResourceLabelIsFallback, prepareTelegramResourceMigration, migrateTelegramResourceConversations } from "./telegram/resources.js"; import { and, asc, eq, inArray, ne, sql } from "drizzle-orm"; import { chatConversations, chatDeliveries, chatEndpointResources, chatEndpoints, toolConnections, type Db } from "@paperclipai/db"; import type { ChatEndpointSetupState } from "@paperclipai/shared"; @@ -20,226 +25,17 @@ type ResourceReconciliationOptions = { maxErrorText: number; }; -export function githubRepositoryInventoryItemFromPayload( - payload: unknown, -): ChatProviderResourceInventoryItem | null { - if (!payload || typeof payload !== "object" || Array.isArray(payload)) - return null; - const repository = (payload as { repository?: unknown }).repository; - if ( - !repository || - typeof repository !== "object" || - Array.isArray(repository) - ) - return null; - const value = repository as { - id?: unknown; - full_name?: unknown; - html_url?: unknown; - name?: unknown; - owner?: { id?: unknown; login?: unknown }; - private?: unknown; - }; - const repositoryId = - typeof value.id === "number" && Number.isSafeInteger(value.id) - ? String(value.id) - : typeof value.id === "string" && /^\d+$/.test(value.id) - ? value.id - : null; - const owner = - typeof value.owner?.login === "string" ? value.owner.login.trim() : ""; - const name = typeof value.name === "string" ? value.name.trim() : ""; - const fullName = - typeof value.full_name === "string" && value.full_name.includes("/") - ? value.full_name.trim() - : owner && name - ? `${owner}/${name}` - : ""; - if (!repositoryId || !fullName) return null; - const ownerId = - typeof value.owner?.id === "number" && Number.isSafeInteger(value.owner.id) - ? String(value.owner.id) - : typeof value.owner?.id === "string" && /^\d+$/.test(value.owner.id) - ? value.owner.id - : undefined; - return { - providerResourceId: fullName.toLowerCase(), - parentProviderResourceId: ownerId, - type: "repository", - label: fullName, - providerUrl: - typeof value.html_url === "string" && value.html_url.length > 0 - ? value.html_url - : `https://github.com/${fullName}`, - metadata: { - providerRepositoryId: repositoryId, - fullName, - ...(owner ? { owner } : {}), - private: value.private === true, - source: "provider_webhook", - }, - }; -} - -export function baseTeamsConversationId(value: string): string { - return value.replace(/;messageid=[^;]+/i, ""); -} - -export function slackResourceLabelIsFallback( - providerResourceId: string, - label: string, -): boolean { - const candidate = label.trim(); - return ( - candidate === providerResourceId || - candidate === `slack:${providerResourceId}` - ); -} - -export function telegramResourceLabelIsFallback( - providerResourceId: string, - label: string, -): boolean { - const candidate = label.trim(); - return ( - candidate === providerResourceId || - candidate === `telegram:${providerResourceId}` - ); -} - export function createChatResourceReconciliation(db: Db, options: ResourceReconciliationOptions) { const { endpointRecord, withCredentialMutationLease, runtimeGeneration, maxErrorText: MAX_ERROR_TEXT } = options; - function githubRepositoryStableId( - item: ChatProviderResourceInventoryItem, - ): string | null { - const value = item.metadata?.providerRepositoryId; - return typeof value === "string" && /^\d+$/.test(value) ? value : null; - } - async function upsertProviderResourceRow( tx: DbTransaction, endpoint: EndpointRow, item: ChatProviderResourceInventoryItem, ) { - const stableGitHubId = - endpoint.provider === "github" ? githubRepositoryStableId(item) : null; - if (stableGitHubId) { - await tx.execute( - sql`select pg_advisory_xact_lock(hashtextextended(${`chat-github-repository:${endpoint.id}:${stableGitHubId}`}, 0))`, - ); - const stableResource = await tx - .select() - .from(chatEndpointResources) - .where( - and( - eq(chatEndpointResources.companyId, endpoint.companyId), - eq(chatEndpointResources.endpointId, endpoint.id), - eq(chatEndpointResources.type, "repository"), - sql`${chatEndpointResources.metadata}->>'providerRepositoryId' = ${stableGitHubId}`, - ), - ) - .for("update") - .then((rows) => rows[0] ?? null); - if ( - stableResource && - stableResource.providerResourceId !== item.providerResourceId - ) { - const coordinateResource = await tx - .select() - .from(chatEndpointResources) - .where( - and( - eq(chatEndpointResources.companyId, endpoint.companyId), - eq(chatEndpointResources.endpointId, endpoint.id), - eq(chatEndpointResources.type, "repository"), - eq( - chatEndpointResources.providerResourceId, - item.providerResourceId, - ), - ), - ) - .for("update") - .then((rows) => rows[0] ?? null); - if (coordinateResource && coordinateResource.id !== stableResource.id) { - const coordinateConversation = await tx - .select({ id: chatConversations.id }) - .from(chatConversations) - .where( - and( - eq(chatConversations.endpointId, endpoint.id), - eq(chatConversations.resourceId, coordinateResource.id), - ), - ) - .limit(1) - .then((rows) => rows[0] ?? null); - if (coordinateConversation) { - throw conflict( - "GitHub repository identity is already bound to two coordinates", - { code: "chat_github_repository_identity_conflict" }, - ); - } - await tx - .delete(chatEndpointResources) - .where(eq(chatEndpointResources.id, coordinateResource.id)); - } - - const oldRepository = stableResource.providerResourceId; - const oldThreadPrefix = `github:${oldRepository}`; - const newThreadPrefix = `github:${item.providerResourceId}`; - const oldProviderUrl = `https://github.com/${oldRepository}`; - const newProviderUrl = `https://github.com/${item.providerResourceId}`; - await tx - .update(chatEndpointResources) - .set({ - providerResourceId: item.providerResourceId, - parentProviderResourceId: item.parentProviderResourceId ?? null, - label: item.label, - providerUrl: item.providerUrl ?? null, - availability: "available", - enabled: - stableResource.enabled || coordinateResource?.enabled === true, - metadata: item.metadata ?? {}, - updatedAt: new Date(), - }) - .where(eq(chatEndpointResources.id, stableResource.id)); - await tx - .update(chatConversations) - .set({ - externalConversationId: sql`case - when lower(${chatConversations.externalConversationId}) = ${oldRepository} then ${item.providerResourceId} - when lower(${chatConversations.externalConversationId}) = ${oldThreadPrefix} then ${newThreadPrefix} - else ${chatConversations.externalConversationId} - end`, - externalThreadId: sql`case - when lower(${chatConversations.externalThreadId}) = ${oldThreadPrefix} then ${newThreadPrefix} - when lower(${chatConversations.externalThreadId}) like ${`${oldThreadPrefix}:%`} - then ${newThreadPrefix} || substring( - ${chatConversations.externalThreadId} - from char_length(${oldThreadPrefix}) + 1 - ) - else ${chatConversations.externalThreadId} - end`, - externalLabel: item.label, - providerUrl: sql`case - when lower(${chatConversations.providerUrl}) like ${`${oldProviderUrl}/%`} - then ${newProviderUrl} || substring( - ${chatConversations.providerUrl} - from char_length(${oldProviderUrl}) + 1 - ) - else ${chatConversations.providerUrl} - end`, - updatedAt: new Date(), - }) - .where( - and( - eq(chatConversations.companyId, endpoint.companyId), - eq(chatConversations.endpointId, endpoint.id), - eq(chatConversations.resourceId, stableResource.id), - ), - ); - return { id: stableResource.id }; - } + if (endpoint.provider === "github") { + const renamed = await reconcileGitHubRepositoryCoordinates(tx, endpoint, item); + if (renamed) return renamed; } const [resource] = await tx @@ -280,35 +76,6 @@ export function createChatResourceReconciliation(db: Db, options: ResourceReconc return resource ?? null; } - async function reconcileGitHubWebhookRepository( - endpoint: EndpointRow, - payload: unknown, - credentialLease: CredentialMutationLeaseGuard, - ) { - const item = githubRepositoryInventoryItemFromPayload(payload); - if (!item) return; - await db.transaction(async (tx) => { - await credentialLease.assertOwned(tx); - const resource = await upsertProviderResourceRow(tx, endpoint, item); - if (!resource) return; - // A correctly signed repository callback is current provider proof. Keep - // Paperclip's enabled choice, and reopen only conversations previously - // quarantined for provider availability loss. - await tx - .update(chatConversations) - .set({ state: "active", updatedAt: new Date() }) - .where( - and( - eq(chatConversations.companyId, endpoint.companyId), - eq(chatConversations.endpointId, endpoint.id), - eq(chatConversations.resourceId, resource.id), - eq(chatConversations.state, "unavailable"), - ), - ); - await credentialLease.assertOwned(tx); - }); - } - async function reconcileProviderResourceRows( endpoint: EndpointRow, inventory: ChatProviderInventoryResult, @@ -390,55 +157,10 @@ export function createChatResourceReconciliation(db: Db, options: ResourceReconc .orderBy(asc(chatEndpointResources.label)); return resources.map((resource) => ({ ...resource, - participants: - resource.providerResourceId.startsWith("imessage-photon:") && - Array.isArray(resource.metadata.participants) - ? resource.metadata.participants.flatMap((participant) => { - const address = - participant && typeof participant === "object" - ? (participant as Record).address - : null; - return typeof address === "string" ? [address] : []; - }) - : undefined, + participants: imessageResourceParticipants(resource), })); } - async function listGitHubRepositories( - endpointId: string, - query: { limit: number; offset: number; search: string }, - ) { - const record = await endpointRecord(endpointId); - if (!record || record.endpoint.provider !== "github") throw notFound("GitHub bot not found"); - const scope = and( - eq(chatEndpointResources.companyId, record.endpoint.companyId), - eq(chatEndpointResources.endpointId, endpointId), - eq(chatEndpointResources.type, "repository"), - ); - // Literal, case-insensitive substring search (%, _ and backslashes are - // ordinary characters). Counts always describe the entire connection. - const search = query.search - ? sql`strpos(lower(${chatEndpointResources.label}), lower(${query.search})) > 0` - : undefined; - const [rows, [counts]] = await Promise.all([ - db.select().from(chatEndpointResources).where(and(scope, search)) - .orderBy(asc(chatEndpointResources.label), asc(chatEndpointResources.id)) - .offset(query.offset).limit(query.limit + 1), - db.select({ - totalCount: sql`count(*)::integer`, - enabledCount: sql`count(*) filter (where ${chatEndpointResources.enabled})::integer`, - availableCount: sql`count(*) filter (where ${chatEndpointResources.availability} = 'available')::integer`, - }).from(chatEndpointResources).where(scope), - ]); - return { - items: rows.slice(0, query.limit), - nextOffset: rows.length > query.limit ? query.offset + query.limit : null, - totalCount: counts!.totalCount, - enabledCount: counts!.enabledCount, - availableCount: counts!.availableCount, - }; - } - async function replaceResources( endpointId: string, updates: Array<{ id: string; enabled: boolean }>, @@ -449,11 +171,6 @@ export function createChatResourceReconciliation(db: Db, options: ResourceReconc return listResources(endpointId); } - async function toggleAllGitHubRepositories(endpointId: string, enabled: boolean, actorUserId?: string | null) { - await updateResourceSelection(endpointId, [], actorUserId, { allGitHubRepositoriesEnabled: enabled }); - return { success: true as const }; - } - async function updateResourceSelection( endpointId: string, updates: Array<{ id: string; enabled: boolean }>, @@ -671,47 +388,9 @@ export function createChatResourceReconciliation(db: Db, options: ResourceReconc currentEndpoint.provider === "telegram" ? effect.previousProviderResourceId : undefined; - const previousResource = previousProviderResourceId - ? await tx - .select() - .from(chatEndpointResources) - .where( - and( - eq(chatEndpointResources.endpointId, currentEndpoint.id), - eq(chatEndpointResources.type, "chat"), - eq( - chatEndpointResources.providerResourceId, - previousProviderResourceId, - ), - ), - ) - .for("update") - .then((rows) => rows[0] ?? null) - : null; - const migratedLabel = - previousResource && - telegramResourceLabelIsFallback(providerResourceId, effect.label) - ? previousResource.label - : effect.label; - const migratedEnabled = previousResource?.enabled === true; - if ( - previousResource && - previousProviderResourceId !== providerResourceId - ) { - await tx - .update(chatEndpointResources) - .set({ - availability: "unavailable", - enabled: false, - metadata: { - ...previousResource.metadata, - source: "chat_migration", - migratedTo: providerResourceId, - }, - updatedAt: now, - }) - .where(eq(chatEndpointResources.id, previousResource.id)); - } + const { previousResource, migratedLabel, migratedEnabled } = await prepareTelegramResourceMigration( + tx, currentEndpoint, effect, providerResourceId, previousProviderResourceId, now, + ); const preserveProviderLabel = (currentEndpoint.provider === "slack" && slackResourceLabelIsFallback( @@ -769,37 +448,7 @@ export function createChatResourceReconciliation(db: Db, options: ResourceReconc previousResource && previousProviderResourceId !== providerResourceId ) { - const previousThreadPrefix = `telegram:${previousProviderResourceId}`; - const migratedThreadPrefix = `telegram:${providerResourceId}`; - await tx - .update(chatConversations) - .set({ - resourceId: resource.id, - // Telegram SDK threads use the namespaced channel id. Keep - // the migrated row in that canonical shape so the first - // reply from the replacement supergroup finds this same - // task instead of creating a second conversation. - externalConversationId: migratedThreadPrefix, - externalThreadId: sql`case - when ${chatConversations.externalThreadId} = ${previousThreadPrefix} then ${migratedThreadPrefix} - when ${chatConversations.externalThreadId} like ${`${previousThreadPrefix}:%`} - then ${migratedThreadPrefix} || substring( - ${chatConversations.externalThreadId} - from char_length(${previousThreadPrefix}) + 1 - ) - else ${chatConversations.externalThreadId} - end`, - externalLabel: migratedLabel, - state: "active", - updatedAt: now, - }) - .where( - and( - eq(chatConversations.companyId, currentEndpoint.companyId), - eq(chatConversations.endpointId, currentEndpoint.id), - eq(chatConversations.resourceId, previousResource.id), - ), - ); + await migrateTelegramResourceConversations(tx, currentEndpoint, previousResource, resource, previousProviderResourceId, providerResourceId, migratedLabel, now); } if (resource) { await tx @@ -943,5 +592,7 @@ export function createChatResourceReconciliation(db: Db, options: ResourceReconc }); } + const { reconcileGitHubWebhookRepository, listGitHubRepositories, toggleAllGitHubRepositories } = createGitHubResources(db, { endpointRecord, upsertProviderResourceRow, updateResourceSelection }); + return { reconcileGitHubWebhookRepository, reconcileProviderResourceRows, listResources, listGitHubRepositories, replaceResources, toggleAllGitHubRepositories, hasNewerProcessedProviderLifecycleEffect, applyProviderAvailability }; } diff --git a/server/src/services/chat-channels/slack/resources.ts b/server/src/services/chat-channels/slack/resources.ts new file mode 100644 index 00000000000..1d7773f750a --- /dev/null +++ b/server/src/services/chat-channels/slack/resources.ts @@ -0,0 +1,10 @@ +export function slackResourceLabelIsFallback( + providerResourceId: string, + label: string, +): boolean { + const candidate = label.trim(); + return ( + candidate === providerResourceId || + candidate === `slack:${providerResourceId}` + ); +} diff --git a/server/src/services/chat-channels/telegram/resources.ts b/server/src/services/chat-channels/telegram/resources.ts new file mode 100644 index 00000000000..724b782dc96 --- /dev/null +++ b/server/src/services/chat-channels/telegram/resources.ts @@ -0,0 +1,104 @@ +import { and,eq,sql } from "drizzle-orm"; +import { chatConversations,chatEndpointResources,chatEndpoints,type Db } from "@paperclipai/db"; +import type { ChatProviderLifecycleEffect } from "../../chat-provider-lifecycle.js"; +type DbTransaction=Parameters[0]>[0]; +type EndpointRow=typeof chatEndpoints.$inferSelect; + +export function telegramResourceLabelIsFallback( + providerResourceId: string, + label: string, +): boolean { + const candidate=label.trim(); + return ( + candidate===providerResourceId|| + candidate===`telegram:${providerResourceId}` + ); +} + +export async function prepareTelegramResourceMigration( + tx: DbTransaction,currentEndpoint: EndpointRow, + effect: Extract, + providerResourceId: string,previousProviderResourceId: string|undefined,now: Date, +) { + const previousResource=previousProviderResourceId + ? await tx + .select() + .from(chatEndpointResources) + .where( + and( + eq(chatEndpointResources.endpointId,currentEndpoint.id), + eq(chatEndpointResources.type,"chat"), + eq( + chatEndpointResources.providerResourceId, + previousProviderResourceId, + ), + ), + ) + .for("update") + .then((rows) => rows[0]??null) + :null; + const migratedLabel= + previousResource&& + telegramResourceLabelIsFallback(providerResourceId,effect.label) + ? previousResource.label + :effect.label; + const migratedEnabled=previousResource?.enabled===true; + if( + previousResource&& + previousProviderResourceId!==providerResourceId + ) { + await tx + .update(chatEndpointResources) + .set({ + availability: "unavailable", + enabled: false, + metadata: { + ...previousResource.metadata, + source: "chat_migration", + migratedTo: providerResourceId, + }, + updatedAt: now, + }) + .where(eq(chatEndpointResources.id,previousResource.id)); + } + return { previousResource,migratedLabel,migratedEnabled }; +} + +export async function migrateTelegramResourceConversations( + tx: DbTransaction,currentEndpoint: EndpointRow, + previousResource: typeof chatEndpointResources.$inferSelect, + resource: { id: string },previousProviderResourceId: string|undefined, + providerResourceId: string,migratedLabel: string,now: Date, +) { + const previousThreadPrefix=`telegram:${previousProviderResourceId}`; + const migratedThreadPrefix=`telegram:${providerResourceId}`; + await tx + .update(chatConversations) + .set({ + resourceId: resource.id, + // Telegram SDK threads use the namespaced channel id. Keep + // the migrated row in that canonical shape so the first + // reply from the replacement supergroup finds this same + // task instead of creating a second conversation. + externalConversationId: migratedThreadPrefix, + externalThreadId: sql`case + when ${chatConversations.externalThreadId} = ${previousThreadPrefix} then ${migratedThreadPrefix} + when ${chatConversations.externalThreadId} like ${`${previousThreadPrefix}:%`} + then ${migratedThreadPrefix} || substring( + ${chatConversations.externalThreadId} + from char_length(${previousThreadPrefix}) + 1 + ) + else ${chatConversations.externalThreadId} + end`, + externalLabel: migratedLabel, + state: "active", + updatedAt: now, + }) + .where( + and( + eq(chatConversations.companyId,currentEndpoint.companyId), + eq(chatConversations.endpointId,currentEndpoint.id), + eq(chatConversations.resourceId,previousResource.id), + ), + ); +}