mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 20:46:28 +01:00
refactor(chat): place provider resource rules in channel folders
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
296e4ea384
commit
5a7b888c5d
8 files changed
+451
-365
No files matched your search
@@ -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.
|
||||
@@ -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,
|
||||
|
||||
@@ -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<Parameters<Db["transaction"]>[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<string>`case
|
||||
when lower(${chatConversations.externalConversationId}) = ${oldRepository} then ${item.providerResourceId}
|
||||
when lower(${chatConversations.externalConversationId}) = ${oldThreadPrefix} then ${newThreadPrefix}
|
||||
else ${chatConversations.externalConversationId}
|
||||
end`,
|
||||
externalThreadId: sql<string>`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<string|null>`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<void>;
|
||||
};
|
||||
|
||||
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<number>`count(*)::integer`,
|
||||
enabledCount: sql<number>`count(*) filter (where ${chatEndpointResources.enabled})::integer`,
|
||||
availableCount: sql<number>`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 };
|
||||
}
|
||||
@@ -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<string,unknown>).address
|
||||
:null;
|
||||
return typeof address==="string"? [address]:[];
|
||||
})
|
||||
:undefined;
|
||||
}
|
||||
@@ -0,0 +1,3 @@
|
||||
export function baseTeamsConversationId(value: string): string {
|
||||
return value.replace(/;messageid=[^;]+/i, "");
|
||||
}
|
||||
@@ -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<string>`case
|
||||
when lower(${chatConversations.externalConversationId}) = ${oldRepository} then ${item.providerResourceId}
|
||||
when lower(${chatConversations.externalConversationId}) = ${oldThreadPrefix} then ${newThreadPrefix}
|
||||
else ${chatConversations.externalConversationId}
|
||||
end`,
|
||||
externalThreadId: sql<string>`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<string | null>`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<string, unknown>).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<number>`count(*)::integer`,
|
||||
enabledCount: sql<number>`count(*) filter (where ${chatEndpointResources.enabled})::integer`,
|
||||
availableCount: sql<number>`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<string>`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 };
|
||||
}
|
||||
@@ -0,0 +1,10 @@
|
||||
export function slackResourceLabelIsFallback(
|
||||
providerResourceId: string,
|
||||
label: string,
|
||||
): boolean {
|
||||
const candidate = label.trim();
|
||||
return (
|
||||
candidate === providerResourceId ||
|
||||
candidate === `slack:${providerResourceId}`
|
||||
);
|
||||
}
|
||||
@@ -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<Parameters<Db["transaction"]>[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<ChatProviderLifecycleEffect,{ kind: "resource" }>,
|
||||
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<string>`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),
|
||||
),
|
||||
);
|
||||
}
|
||||
Reference in new issue
Block a user