mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
Merge origin/master into task workspace ownership
Co-Authored-By: Paperclip <noreply@paperclip.ing> * origin/master: fix(docker): skip the full-tree ownership walk on routine boots (#15992) refactor(chat): isolate Telegram, Slack, GitHub and Discord workflows (#15976)
This commit is contained in:
DottaandPaperclip committed
commit
d11cf62985
11 files changed
+5453
-4735
No files matched your search
@@ -32,7 +32,20 @@ function writeStub(name: string, body: string) {
|
||||
writeFileSync(path, `#!/bin/sh\n${body}\n`, { mode: 0o755 });
|
||||
}
|
||||
|
||||
function installStubs(ids: { uid: number; gid: number; nodeUid?: number; nodeGid?: number; homeMismatch?: boolean }) {
|
||||
type StubOptions = {
|
||||
uid: number;
|
||||
gid: number;
|
||||
nodeUid?: number;
|
||||
nodeGid?: number;
|
||||
/** Mismatch anywhere: both the shallow and the full-tree probe find it. */
|
||||
homeMismatch?: boolean;
|
||||
/** Mismatch only below the shallow probe depth: just the full-tree probe finds it. */
|
||||
deepMismatch?: boolean;
|
||||
/** The full-tree find exits non-zero without printing a mismatch (e.g. an I/O error). */
|
||||
fullProbeFails?: boolean;
|
||||
};
|
||||
|
||||
function installStubs(ids: StubOptions) {
|
||||
writeStub(
|
||||
"id",
|
||||
[
|
||||
@@ -46,12 +59,21 @@ function installStubs(ids: { uid: number; gid: number; nodeUid?: number; nodeGid
|
||||
for (const cmd of ["usermod", "groupmod", "chown"]) {
|
||||
writeStub(cmd, `echo "${cmd} $*" >> "${logFile}"`);
|
||||
}
|
||||
// The entrypoint's ownership probe is a first-mismatch find over the app
|
||||
// home. An empty result models a fully node-owned tree (image-baked dir,
|
||||
// healthy volume); a path models any uid OR gid mismatch anywhere in the
|
||||
// tree (fresh root-owned mount, root-owned descendant, stale group after
|
||||
// a GID-only remap).
|
||||
writeStub("find", ids.homeMismatch ? `echo "$1/mismatched-entry"` : `true`);
|
||||
// The entrypoint's ownership probes are first-mismatch finds over the app
|
||||
// home: a shallow one bounded by -maxdepth, and a full-tree one. An empty
|
||||
// result models a node-owned tree (image-baked dir, healthy volume); a
|
||||
// path models a uid OR gid mismatch (fresh root-owned mount, root-owned
|
||||
// descendant, stale group after a GID-only remap). Each call is logged as
|
||||
// "find-shallow" or "find-full" so tests can assert which walks ran.
|
||||
writeStub(
|
||||
"find",
|
||||
[
|
||||
`case " $* " in`,
|
||||
` *" -maxdepth "*) echo "find-shallow $1" >> "${logFile}"; ${ids.homeMismatch ? `echo "$1/mismatched-entry"` : "true"} ;;`,
|
||||
` *) echo "find-full $1" >> "${logFile}"; ${ids.homeMismatch || ids.deepMismatch ? `echo "$1/deep/mismatched-entry"` : ids.fullProbeFails ? "exit 1" : "true"} ;;`,
|
||||
`esac`,
|
||||
].join("\n"),
|
||||
);
|
||||
writeStub("gosu", `echo "gosu $*" >> "${logFile}"\nshift\nexec "$@"`);
|
||||
}
|
||||
|
||||
@@ -151,6 +173,102 @@ describe("docker-entrypoint.sh", () => {
|
||||
expect(calls).toContain(`chown -R node:node ${stubDir}`);
|
||||
});
|
||||
|
||||
describe("ownership probe cost", () => {
|
||||
const marker = () => join(stubDir, ".paperclip-ownership-verified");
|
||||
|
||||
it("skips the full-tree walk when ownership is correct and the marker is current", async () => {
|
||||
installStubs({ uid: 0, gid: 0 });
|
||||
writeFileSync(marker(), "v1 1000:1000\n");
|
||||
|
||||
const { stdout, calls } = await runEntrypoint({ PAPERCLIP_HOME: stubDir });
|
||||
|
||||
expect(calls).toContain(`find-shallow ${stubDir}`);
|
||||
expect(calls).not.toContain("find-full");
|
||||
expect(calls).not.toContain("chown");
|
||||
expect(stdout).toContain("ENTRYPOINT-CMD-RAN");
|
||||
});
|
||||
|
||||
it("skips the full-tree walk when the marker is current, even with a deep mismatch", async () => {
|
||||
// The marker is the trade-off that keeps boots O(1): once a tree has
|
||||
// been verified, only the shallow levels are re-checked.
|
||||
installStubs({ uid: 0, gid: 0, deepMismatch: true });
|
||||
writeFileSync(marker(), "v1 1000:1000\n");
|
||||
|
||||
const { calls } = await runEntrypoint({ PAPERCLIP_HOME: stubDir });
|
||||
|
||||
expect(calls).not.toContain("find-full");
|
||||
expect(calls).not.toContain("chown");
|
||||
});
|
||||
|
||||
it("runs one full-tree walk without a marker, then records the marker", async () => {
|
||||
// First boot after upgrading from an image without the marker.
|
||||
installStubs({ uid: 0, gid: 0 });
|
||||
|
||||
const { calls } = await runEntrypoint({ PAPERCLIP_HOME: stubDir });
|
||||
|
||||
expect(calls).toContain(`find-shallow ${stubDir}`);
|
||||
expect(calls).toContain(`find-full ${stubDir}`);
|
||||
expect(calls).not.toContain("chown");
|
||||
expect(readFileSync(marker(), "utf8")).toBe("v1 1000:1000\n");
|
||||
});
|
||||
|
||||
it("does not record the marker when the full-tree walk fails without finding a mismatch", async () => {
|
||||
installStubs({ uid: 0, gid: 0, fullProbeFails: true });
|
||||
|
||||
const { stdout, calls } = await runEntrypoint({ PAPERCLIP_HOME: stubDir });
|
||||
|
||||
expect(calls).toContain(`find-full ${stubDir}`);
|
||||
expect(calls).not.toContain("chown");
|
||||
expect(existsSync(marker())).toBe(false);
|
||||
expect(stdout).toContain("failed; it will run again on the next boot");
|
||||
expect(stdout).toContain("ENTRYPOINT-CMD-RAN");
|
||||
});
|
||||
|
||||
it("repairs a deep mismatch found by the full-tree walk when there is no marker", async () => {
|
||||
installStubs({ uid: 0, gid: 0, deepMismatch: true });
|
||||
|
||||
const { calls } = await runEntrypoint({ PAPERCLIP_HOME: stubDir });
|
||||
|
||||
expect(calls).toContain(`find-full ${stubDir}`);
|
||||
expect(calls).toContain(`chown -R node:node ${stubDir}`);
|
||||
expect(readFileSync(marker(), "utf8")).toBe("v1 1000:1000\n");
|
||||
});
|
||||
|
||||
it("repairs a root-owned top level without a full-tree probe, even with a current marker", async () => {
|
||||
installStubs({ uid: 0, gid: 0, homeMismatch: true });
|
||||
writeFileSync(marker(), "v1 1000:1000\n");
|
||||
|
||||
const { calls } = await runEntrypoint({ PAPERCLIP_HOME: stubDir });
|
||||
|
||||
expect(calls).toContain(`find-shallow ${stubDir}`);
|
||||
expect(calls).not.toContain("find-full");
|
||||
expect(calls).toContain(`chown -R node:node ${stubDir}`);
|
||||
});
|
||||
|
||||
it("re-runs the full-tree walk when the marker was written for another UID/GID", async () => {
|
||||
installStubs({ uid: 0, gid: 0, nodeUid: 1001, nodeGid: 1001, deepMismatch: true });
|
||||
writeFileSync(marker(), "v1 1000:1000\n");
|
||||
|
||||
const { calls } = await runEntrypoint({ USER_UID: "1001", USER_GID: "1001", PAPERCLIP_HOME: stubDir });
|
||||
|
||||
expect(calls).toContain(`find-full ${stubDir}`);
|
||||
expect(calls).toContain(`chown -R node:node ${stubDir}`);
|
||||
expect(readFileSync(marker(), "utf8")).toBe("v1 1001:1001\n");
|
||||
});
|
||||
|
||||
it("logs timestamped start, repair duration, and pre-exec lines", async () => {
|
||||
installStubs({ uid: 0, gid: 0, homeMismatch: true });
|
||||
|
||||
const { stdout } = await runEntrypoint({ PAPERCLIP_HOME: stubDir });
|
||||
const stamp = "\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}Z docker-entrypoint\\.sh: ";
|
||||
|
||||
expect(stdout).toMatch(new RegExp(`^${stamp}starting$`, "m"));
|
||||
expect(stdout).toMatch(new RegExp(`^${stamp}ownership repair finished in \\d+s$`, "m"));
|
||||
expect(stdout).toMatch(new RegExp(`^${stamp}starting app as node$`, "m"));
|
||||
expect(stdout.indexOf("starting app as node")).toBeLessThan(stdout.indexOf("ENTRYPOINT-CMD-RAN"));
|
||||
});
|
||||
});
|
||||
|
||||
it("execs directly and silently when already running as the requested user (restricted PodSecurity)", async () => {
|
||||
installStubs({ uid: 1000, gid: 1000 });
|
||||
|
||||
|
||||
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,639 @@
|
||||
import type { EndpointRow, DbTransaction, DbOrTransaction, RuntimeContext, LifecycleRuntimeFence } from "../runtime-types.js";
|
||||
|
||||
import { chatCredentialMutationLease, type CredentialMutationLeaseGuard } from "../../chat-credential-mutation-lease.js";
|
||||
|
||||
import {
|
||||
and,
|
||||
desc,
|
||||
eq,
|
||||
gt,
|
||||
} from "drizzle-orm";
|
||||
|
||||
import {
|
||||
readChatControlChronology,
|
||||
} from "../../chat-control-chronology.js";
|
||||
|
||||
import type { Db } from "@paperclipai/db";
|
||||
|
||||
import {
|
||||
chatActions,
|
||||
chatConversations,
|
||||
chatEndpointLeases,
|
||||
chatEndpointResources,
|
||||
chatEndpoints,
|
||||
issues,
|
||||
toolConnections,
|
||||
} from "@paperclipai/db";
|
||||
import type {
|
||||
ChatEndpointSetupState,
|
||||
SafeChatPublicationPayload,
|
||||
ToolCredentialSecretRef,
|
||||
} from "@paperclipai/shared";
|
||||
|
||||
import {
|
||||
conflict,
|
||||
notFound,
|
||||
} from "../../../errors.js";
|
||||
import { logger } from "../../../middleware/logger.js";
|
||||
|
||||
import {
|
||||
type ChatSdkCallbackEvent,
|
||||
type DiscordNativeCommandResponse,
|
||||
} from "../../chat-sdk-runtime.js";
|
||||
import {
|
||||
parseDiscordNativeCommand,
|
||||
parseDiscordNativeCommandReceipt,
|
||||
type DiscordNativeCommandTarget,
|
||||
} from "../../chat-discord-native-commands.js";
|
||||
|
||||
import {
|
||||
logActivity,
|
||||
} from "../../activity-log.js";
|
||||
|
||||
import { projectSafeChatPublication } from "../../chat-publication-projection.js";
|
||||
|
||||
import {
|
||||
readRegisteredDiscordCommandRegistration,
|
||||
reconcileStoredDiscordCommandRegistration,
|
||||
} from "../../chat-discord-command-registration-store.js";
|
||||
|
||||
import type {
|
||||
Author,
|
||||
SlashCommandEvent,
|
||||
} from "chat";
|
||||
|
||||
import type { EndpointRecord } from "../runtime-types.js";
|
||||
|
||||
type DiscordCommandsOptions = {
|
||||
endpointRecord: (endpointId: string) => Promise<EndpointRecord | null>;
|
||||
runtimeContextForRecord: (record: EndpointRecord) => RuntimeContext;
|
||||
runtimeGeneration: (setup: ChatEndpointSetupState) => number;
|
||||
credentialFingerprint: (refs: ToolCredentialSecretRef[]) => string;
|
||||
resolveCredentialRefs: (endpoint: EndpointRow, refs: ToolCredentialSecretRef[], database?: DbOrTransaction) => Promise<Record<string, string>>;
|
||||
fetchImpl: typeof fetch;
|
||||
withCredentialMutationLease: ReturnType<typeof chatCredentialMutationLease>;
|
||||
runtimeCallbackRecord: (endpointId: string, context: RuntimeContext, statuses: EndpointRow["status"][]) => Promise<EndpointRecord | null>;
|
||||
ensurePrincipal: (endpoint: EndpointRow, author: Author, raw?: unknown) => Promise<{ principal: { id: string }; userId: string | null }>;
|
||||
runtimeCallbackEndpoint: (tx: DbTransaction, endpointId: string, context: LifecycleRuntimeFence, statuses: EndpointRow["status"][]) => Promise<EndpointRow | null>;
|
||||
nonDirectDestinationAllowed: (endpoint: EndpointRow, resource: typeof chatEndpointResources.$inferSelect | null | undefined) => boolean;
|
||||
lockCurrentPrincipalAuthorization: (tx: DbOrTransaction, endpoint: EndpointRow, principalId: string) => Promise<{ allowed: boolean; userId: string | null }>;
|
||||
getDiscordGatewayOwnership: (endpointId: string) => { leaseKey: string; token: string } | undefined;
|
||||
discordGatewayRuntimeIsCurrent: (endpointId: string, context: RuntimeContext) => boolean;
|
||||
stageAuthorizedTaskControlPublication: (tx: DbOrTransaction, input: { companyId: string; conversationId: string; endpointId: string; idempotencyKey: string; issueId: string; payload: SafeChatPublicationPayload; principalId: string }) => Promise<{ id: string }>;
|
||||
scheduleMessageProcessing: (task: () => Promise<void>) => void;
|
||||
processPendingPublications: () => Promise<unknown>;
|
||||
};
|
||||
|
||||
export function createDiscordCommands(db: Db, options: DiscordCommandsOptions) {
|
||||
const { endpointRecord, runtimeContextForRecord, runtimeGeneration, credentialFingerprint, resolveCredentialRefs, fetchImpl, withCredentialMutationLease, runtimeCallbackRecord, ensurePrincipal, runtimeCallbackEndpoint, nonDirectDestinationAllowed, lockCurrentPrincipalAuthorization, getDiscordGatewayOwnership, discordGatewayRuntimeIsCurrent, stageAuthorizedTaskControlPublication, scheduleMessageProcessing, processPendingPublications } = options;
|
||||
|
||||
|
||||
async function reconcileDiscordCommands(
|
||||
endpointId: string,
|
||||
existingLease?: CredentialMutationLeaseGuard,
|
||||
force = false,
|
||||
): Promise<void> {
|
||||
const initial = await endpointRecord(endpointId);
|
||||
if (!initial || initial.endpoint.provider !== "discord") return;
|
||||
const reconcile = async (lease: CredentialMutationLeaseGuard) => {
|
||||
const record = await endpointRecord(endpointId);
|
||||
if (
|
||||
!record ||
|
||||
record.endpoint.provider !== "discord" ||
|
||||
!["verifying", "active", "attention"].includes(
|
||||
record.endpoint.status,
|
||||
) ||
|
||||
!record.endpoint.botExternalId ||
|
||||
!record.endpoint.providerAccountId
|
||||
)
|
||||
return;
|
||||
const scope = {
|
||||
companyId: record.endpoint.companyId,
|
||||
endpointId,
|
||||
applicationId: record.endpoint.botExternalId,
|
||||
guildId: record.endpoint.providerAccountId,
|
||||
};
|
||||
const fence = runtimeContextForRecord(record);
|
||||
const authorize = async (tx: DbTransaction) => {
|
||||
const endpoint = await tx
|
||||
.select()
|
||||
.from(chatEndpoints)
|
||||
.where(
|
||||
and(
|
||||
eq(chatEndpoints.companyId, scope.companyId),
|
||||
eq(chatEndpoints.id, scope.endpointId),
|
||||
),
|
||||
)
|
||||
.for("no key update")
|
||||
.then((rows) => rows[0]);
|
||||
if (
|
||||
!endpoint ||
|
||||
endpoint.provider !== "discord" ||
|
||||
!["verifying", "active", "attention"].includes(endpoint.status) ||
|
||||
endpoint.botExternalId !== scope.applicationId ||
|
||||
endpoint.providerAccountId !== scope.guildId ||
|
||||
runtimeGeneration(endpoint.setup) !== fence.generation
|
||||
)
|
||||
throw conflict("Discord command registration authority changed");
|
||||
const connection = await tx
|
||||
.select()
|
||||
.from(toolConnections)
|
||||
.where(
|
||||
and(
|
||||
eq(toolConnections.companyId, scope.companyId),
|
||||
eq(toolConnections.id, endpoint.connectionId),
|
||||
),
|
||||
)
|
||||
.for("no key update")
|
||||
.then((rows) => rows[0]);
|
||||
if (
|
||||
!connection?.enabled ||
|
||||
connection.status !== "active" ||
|
||||
credentialFingerprint(connection.credentialSecretRefs) !==
|
||||
fence.credentialFingerprint
|
||||
)
|
||||
throw conflict("Discord command registration authority changed");
|
||||
await lease.assertOwned(tx);
|
||||
};
|
||||
const setCapability = async (tx: DbTransaction, registered: boolean) => {
|
||||
const current = await tx
|
||||
.select()
|
||||
.from(chatEndpoints)
|
||||
.where(
|
||||
and(
|
||||
eq(chatEndpoints.companyId, scope.companyId),
|
||||
eq(chatEndpoints.id, endpointId),
|
||||
),
|
||||
)
|
||||
.then((rows) => rows[0]);
|
||||
if (!current) throw notFound("Chat endpoint not found");
|
||||
if (
|
||||
current.capabilities.slashCommands === registered &&
|
||||
current.capabilities.ephemeralMessages === registered
|
||||
)
|
||||
return;
|
||||
await tx
|
||||
.update(chatEndpoints)
|
||||
.set({
|
||||
capabilities: {
|
||||
...current.capabilities,
|
||||
slashCommands: registered,
|
||||
ephemeralMessages: registered,
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatEndpoints.companyId, scope.companyId),
|
||||
eq(chatEndpoints.id, endpointId),
|
||||
),
|
||||
);
|
||||
await logActivity(tx as unknown as Db, {
|
||||
companyId: scope.companyId,
|
||||
actorType: "system",
|
||||
actorId: "system",
|
||||
action: registered
|
||||
? "chat_endpoint.commands_registered"
|
||||
: "chat_endpoint.commands_unavailable",
|
||||
entityType: "tool_connection",
|
||||
entityId: current.connectionId,
|
||||
details: { endpointId, provider: "discord" },
|
||||
});
|
||||
};
|
||||
await lease.assertOwned();
|
||||
const credentials = await resolveCredentialRefs(
|
||||
record.endpoint,
|
||||
record.credentialSecretRefs,
|
||||
);
|
||||
if (
|
||||
credentials.applicationId !== scope.applicationId ||
|
||||
credentials.guildId !== scope.guildId
|
||||
)
|
||||
throw conflict("Discord command registration identity changed");
|
||||
const result = await reconcileStoredDiscordCommandRegistration(db, {
|
||||
scope,
|
||||
runtimeFence: {
|
||||
generation: fence.generation,
|
||||
credentialFingerprint: fence.credentialFingerprint,
|
||||
},
|
||||
botToken: credentials.botToken,
|
||||
fetch: fetchImpl,
|
||||
authorize,
|
||||
force,
|
||||
onState: async (tx, state) =>
|
||||
setCapability(tx, state.phase === "registered"),
|
||||
});
|
||||
if (result.kind === "deferred") return;
|
||||
await db.transaction(async (tx) => {
|
||||
await authorize(tx);
|
||||
const registered =
|
||||
result.kind === "registered"
|
||||
? await readRegisteredDiscordCommandRegistration(tx, scope)
|
||||
: null;
|
||||
await setCapability(tx, registered !== null);
|
||||
await lease.assertOwned(tx);
|
||||
});
|
||||
};
|
||||
try {
|
||||
if (existingLease) await reconcile(existingLease);
|
||||
else await withCredentialMutationLease(initial.endpoint, reconcile);
|
||||
} catch {
|
||||
// Commands are additive. Do not fail an otherwise working mention/thread
|
||||
// connection or expose provider errors/tokens when registration is busy.
|
||||
logger.warn(
|
||||
{ endpointId },
|
||||
"Discord command registration remains unconfirmed",
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
async function handleDiscordNativeCommand(
|
||||
event: ChatSdkCallbackEvent<SlashCommandEvent>,
|
||||
runtimeContext: RuntimeContext,
|
||||
): Promise<DiscordNativeCommandResponse> {
|
||||
const denied: DiscordNativeCommandResponse = { kind: "denied" };
|
||||
const record = await runtimeCallbackRecord(
|
||||
event.endpointId,
|
||||
runtimeContext,
|
||||
["active"],
|
||||
);
|
||||
if (
|
||||
!record ||
|
||||
record.endpoint.provider !== "discord" ||
|
||||
record.endpoint.capabilities.slashCommands !== true ||
|
||||
event.transport !== "discord_gateway" ||
|
||||
runtimeContext.discordGatewayOwned !== true
|
||||
)
|
||||
return denied;
|
||||
const scope = {
|
||||
companyId: record.endpoint.companyId,
|
||||
endpointId: record.endpoint.id,
|
||||
applicationId: record.endpoint.botExternalId ?? "",
|
||||
guildId: record.endpoint.providerAccountId ?? "",
|
||||
};
|
||||
const invocation = parseDiscordNativeCommand(event, scope);
|
||||
if (!invocation) return denied;
|
||||
const initialRegistration = await readRegisteredDiscordCommandRegistration(
|
||||
db,
|
||||
scope,
|
||||
);
|
||||
if (
|
||||
initialRegistration?.receipt.commandId !== invocation.registeredCommandId
|
||||
)
|
||||
return denied;
|
||||
const principal = await ensurePrincipal(
|
||||
record.endpoint,
|
||||
event.event.user,
|
||||
event.event.raw,
|
||||
);
|
||||
const authorityChanged = new Error("Discord command authority changed");
|
||||
let publish = false;
|
||||
try {
|
||||
const response = await db.transaction(
|
||||
async (tx): Promise<DiscordNativeCommandResponse> => {
|
||||
const endpoint = await runtimeCallbackEndpoint(
|
||||
tx,
|
||||
scope.endpointId,
|
||||
runtimeContext,
|
||||
["active"],
|
||||
);
|
||||
if (
|
||||
!endpoint ||
|
||||
endpoint.companyId !== scope.companyId ||
|
||||
endpoint.provider !== "discord" ||
|
||||
endpoint.botExternalId !== scope.applicationId ||
|
||||
endpoint.providerAccountId !== scope.guildId ||
|
||||
endpoint.capabilities.slashCommands !== true
|
||||
)
|
||||
return denied;
|
||||
const connection = await tx
|
||||
.select()
|
||||
.from(toolConnections)
|
||||
.where(
|
||||
and(
|
||||
eq(toolConnections.companyId, scope.companyId),
|
||||
eq(toolConnections.id, endpoint.connectionId),
|
||||
),
|
||||
)
|
||||
.for("no key update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (
|
||||
!connection ||
|
||||
!connection.enabled ||
|
||||
connection.status !== "active" ||
|
||||
credentialFingerprint(connection.credentialSecretRefs) !==
|
||||
runtimeContext.credentialFingerprint
|
||||
)
|
||||
return denied;
|
||||
const registration = await readRegisteredDiscordCommandRegistration(
|
||||
tx,
|
||||
scope,
|
||||
true,
|
||||
);
|
||||
if (
|
||||
registration?.receipt.commandId !== invocation.registeredCommandId
|
||||
)
|
||||
return denied;
|
||||
const providerActionId = `discord-native-command:${invocation.interactionId}`;
|
||||
const prior = await tx
|
||||
.select()
|
||||
.from(chatActions)
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.companyId, scope.companyId),
|
||||
eq(chatActions.endpointId, scope.endpointId),
|
||||
eq(chatActions.providerActionId, providerActionId),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
const receipt = prior
|
||||
? parseDiscordNativeCommandReceipt(prior, invocation, scope)
|
||||
: null;
|
||||
if (
|
||||
prior &&
|
||||
(!receipt ||
|
||||
receipt.principalId !== principal.principal.id ||
|
||||
receipt.runtimeFence.generation !== runtimeContext.generation ||
|
||||
receipt.runtimeFence.credentialFingerprint !==
|
||||
runtimeContext.credentialFingerprint)
|
||||
)
|
||||
return denied;
|
||||
// A replay selects only its original target, including a completed
|
||||
// generation. Never reinterpret an old close as a new task's command.
|
||||
const conversation = receipt
|
||||
? receipt.target
|
||||
? await tx
|
||||
.select()
|
||||
.from(chatConversations)
|
||||
.where(
|
||||
and(
|
||||
eq(chatConversations.companyId, scope.companyId),
|
||||
eq(chatConversations.endpointId, scope.endpointId),
|
||||
eq(chatConversations.id, receipt.target.conversationId),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null)
|
||||
: null
|
||||
: invocation.sourceKind !== "guild_channel"
|
||||
? await tx
|
||||
.select()
|
||||
.from(chatConversations)
|
||||
.where(
|
||||
and(
|
||||
eq(chatConversations.companyId, scope.companyId),
|
||||
eq(chatConversations.endpointId, scope.endpointId),
|
||||
eq(
|
||||
chatConversations.externalConversationId,
|
||||
invocation.channelId,
|
||||
),
|
||||
eq(
|
||||
chatConversations.externalThreadId,
|
||||
invocation.threadId,
|
||||
),
|
||||
),
|
||||
)
|
||||
.orderBy(desc(chatConversations.sessionGeneration))
|
||||
.limit(1)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null)
|
||||
: null;
|
||||
if (
|
||||
(receipt?.target && !conversation) ||
|
||||
(conversation &&
|
||||
(conversation.externalConversationId !== invocation.channelId ||
|
||||
conversation.externalThreadId !== invocation.threadId ||
|
||||
conversation.isDirectMessage !==
|
||||
(invocation.sourceKind === "direct_message") ||
|
||||
!["active", "waiting", "completed"].includes(
|
||||
conversation.state,
|
||||
) ||
|
||||
(receipt?.target &&
|
||||
(conversation.issueId !== receipt.target.issueId ||
|
||||
conversation.sessionGeneration !==
|
||||
receipt.target.sessionGeneration))))
|
||||
)
|
||||
return denied;
|
||||
const resource =
|
||||
invocation.sourceKind === "direct_message"
|
||||
? null
|
||||
: await tx
|
||||
.select()
|
||||
.from(chatEndpointResources)
|
||||
.where(
|
||||
and(
|
||||
eq(chatEndpointResources.companyId, scope.companyId),
|
||||
eq(chatEndpointResources.endpointId, scope.endpointId),
|
||||
eq(
|
||||
chatEndpointResources.providerResourceId,
|
||||
invocation.providerResourceId,
|
||||
),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (
|
||||
invocation.sourceKind === "direct_message"
|
||||
? !endpoint.allowDirectMessages
|
||||
: !nonDirectDestinationAllowed(endpoint, resource) ||
|
||||
(conversation && conversation.resourceId !== resource?.id)
|
||||
)
|
||||
return denied;
|
||||
const authorization = await lockCurrentPrincipalAuthorization(
|
||||
tx,
|
||||
endpoint,
|
||||
principal.principal.id,
|
||||
);
|
||||
if (
|
||||
!authorization.allowed ||
|
||||
authorization.userId !== principal.userId
|
||||
)
|
||||
return denied;
|
||||
const issue = conversation
|
||||
? await tx
|
||||
.select()
|
||||
.from(issues)
|
||||
.where(
|
||||
and(
|
||||
eq(issues.companyId, scope.companyId),
|
||||
eq(issues.id, conversation.issueId),
|
||||
),
|
||||
)
|
||||
.for("share")
|
||||
.then((rows) => rows[0] ?? null)
|
||||
: null;
|
||||
if (conversation && !issue) return denied;
|
||||
const ownership = getDiscordGatewayOwnership(scope.endpointId);
|
||||
if (
|
||||
!ownership ||
|
||||
!discordGatewayRuntimeIsCurrent(scope.endpointId, runtimeContext)
|
||||
)
|
||||
return denied;
|
||||
const lease = await tx
|
||||
.select()
|
||||
.from(chatEndpointLeases)
|
||||
.where(
|
||||
and(
|
||||
eq(chatEndpointLeases.companyId, scope.companyId),
|
||||
eq(chatEndpointLeases.endpointId, scope.endpointId),
|
||||
eq(chatEndpointLeases.leaseKey, ownership.leaseKey),
|
||||
eq(chatEndpointLeases.token, ownership.token),
|
||||
gt(chatEndpointLeases.expiresAt, new Date()),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (
|
||||
!lease ||
|
||||
!discordGatewayRuntimeIsCurrent(scope.endpointId, runtimeContext)
|
||||
)
|
||||
return denied;
|
||||
if (receipt)
|
||||
return { kind: "accepted", content: receipt.result.content };
|
||||
const target: DiscordNativeCommandTarget | null = conversation
|
||||
? {
|
||||
conversationId: conversation.id,
|
||||
issueId: conversation.issueId,
|
||||
sessionGeneration: conversation.sessionGeneration,
|
||||
}
|
||||
: null;
|
||||
const active =
|
||||
conversation &&
|
||||
issue &&
|
||||
["active", "waiting"].includes(conversation.state) &&
|
||||
(invocation.sourceKind !== "direct_message" ||
|
||||
!["done", "cancelled"].includes(issue.status));
|
||||
let content: string;
|
||||
let publicationId: string | null = null;
|
||||
if (invocation.command === "status") {
|
||||
content =
|
||||
issue && (invocation.sourceKind !== "direct_message" || active)
|
||||
? `${issue.identifier}: ${issue.title} — ${issue.status}${conversation?.state === "completed" ? " (conversation closed)" : ""}`.slice(
|
||||
0,
|
||||
2000,
|
||||
)
|
||||
: invocation.sourceKind === "guild_channel"
|
||||
? "Open the Discord task thread to view its Paperclip status."
|
||||
: "No task is active in this conversation.";
|
||||
} else if (
|
||||
invocation.command === "new" &&
|
||||
invocation.sourceKind !== "direct_message"
|
||||
) {
|
||||
content =
|
||||
"Open a new Discord thread by mentioning this agent in a new channel message. This thread stays bound to its current Paperclip task.";
|
||||
} else if (!active || !conversation || !issue) {
|
||||
content =
|
||||
invocation.sourceKind === "direct_message"
|
||||
? "No task is active. Send a message to start a new Paperclip task."
|
||||
: "No active task is bound here. Open its Discord thread to manage it.";
|
||||
} else if (
|
||||
(await readChatControlChronology(
|
||||
tx,
|
||||
record.endpoint,
|
||||
{
|
||||
id: conversation.externalThreadId,
|
||||
channelId: conversation.externalConversationId,
|
||||
},
|
||||
{
|
||||
receivedAt: new Date(),
|
||||
normalizedEvent: {
|
||||
message: { providerMessageId: invocation.interactionId },
|
||||
},
|
||||
},
|
||||
)) === "before_or_unproven"
|
||||
) {
|
||||
content =
|
||||
"This command predates an already completed chat control. Send a new command for the current conversation.";
|
||||
} else {
|
||||
const publicText =
|
||||
invocation.command === "new"
|
||||
? "Send your request to start a new Paperclip task."
|
||||
: invocation.sourceKind === "direct_message"
|
||||
? "This chat conversation is closed. Send another message to start a new task."
|
||||
: "This chat conversation is closed. A later message here will continue the same Paperclip task.";
|
||||
const publication = await stageAuthorizedTaskControlPublication(
|
||||
tx,
|
||||
{
|
||||
companyId: scope.companyId,
|
||||
endpointId: scope.endpointId,
|
||||
conversationId: conversation.id,
|
||||
issueId: issue.id,
|
||||
principalId: principal.principal.id,
|
||||
idempotencyKey: `control:${invocation.command}:discord:${scope.endpointId}:${invocation.interactionId}`,
|
||||
payload: projectSafeChatPublication({
|
||||
classification: "external",
|
||||
source: "task_control",
|
||||
text: publicText,
|
||||
}),
|
||||
},
|
||||
);
|
||||
publicationId = publication.id;
|
||||
content = `${invocation.command === "new" ? "New-task" : "Close"} request recorded. Watch this conversation for confirmation.`;
|
||||
}
|
||||
const inserted = await tx
|
||||
.insert(chatActions)
|
||||
.values({
|
||||
companyId: scope.companyId,
|
||||
endpointId: scope.endpointId,
|
||||
conversationId: target?.conversationId ?? null,
|
||||
principalId: principal.principal.id,
|
||||
kind: "discord_native_command",
|
||||
providerActionId,
|
||||
payload: {
|
||||
version: 1,
|
||||
invocation,
|
||||
runtimeFence: {
|
||||
generation: runtimeContext.generation,
|
||||
credentialFingerprint: runtimeContext.credentialFingerprint,
|
||||
},
|
||||
target,
|
||||
},
|
||||
status: "processed",
|
||||
result: {
|
||||
kind: "discord_native_command_recorded",
|
||||
content,
|
||||
publicationId,
|
||||
},
|
||||
})
|
||||
.onConflictDoNothing()
|
||||
.returning({ id: chatActions.id });
|
||||
if (!inserted.length) throw authorityChanged;
|
||||
await logActivity(tx as unknown as Db, {
|
||||
companyId: scope.companyId,
|
||||
actorType: authorization.userId ? "user" : "system",
|
||||
actorId: authorization.userId ?? `chat:${principal.principal.id}`,
|
||||
action: "chat.discord_command_recorded",
|
||||
entityType: "chat_endpoint",
|
||||
entityId: scope.endpointId,
|
||||
details: {
|
||||
command: invocation.command,
|
||||
interactionId: invocation.interactionId,
|
||||
conversationId: target?.conversationId ?? null,
|
||||
sessionGeneration: target?.sessionGeneration ?? null,
|
||||
publicationId,
|
||||
},
|
||||
});
|
||||
// No network work is performed under these locks. A stop while any DB
|
||||
// await was in flight rolls back the command and its staged outbox row.
|
||||
if (
|
||||
!discordGatewayRuntimeIsCurrent(scope.endpointId, runtimeContext) ||
|
||||
lease.expiresAt.getTime() <= Date.now()
|
||||
)
|
||||
throw authorityChanged;
|
||||
publish = publicationId !== null;
|
||||
return { kind: "accepted", content };
|
||||
},
|
||||
);
|
||||
if (publish)
|
||||
scheduleMessageProcessing(async () => {
|
||||
await processPendingPublications();
|
||||
});
|
||||
return response;
|
||||
} catch (error) {
|
||||
if (error === authorityChanged) return denied;
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
return { reconcileDiscordCommands, handleDiscordNativeCommand };
|
||||
}
|
||||
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,38 @@
|
||||
import type { Db, chatEndpoints } from "@paperclipai/db";
|
||||
import type { ToolCredentialSecretRef } from "@paperclipai/shared";
|
||||
import type { ChatEndpointRuntime } from "../voice/voice-runtime.js";
|
||||
|
||||
export type EndpointRow = typeof chatEndpoints.$inferSelect;
|
||||
|
||||
export type DbTransaction = Parameters<Parameters<Db["transaction"]>[0]>[0];
|
||||
|
||||
export type DbOrTransaction = Db | DbTransaction;
|
||||
|
||||
export type RuntimeContext = {
|
||||
credentialFingerprint: string;
|
||||
discordGatewayEventQueue?: Promise<void>;
|
||||
discordGatewaySequence?: number;
|
||||
discordGatewayOwned?: boolean;
|
||||
/** Cache identity only; durable command admission rechecks ownership. */
|
||||
discordCommandId?: string;
|
||||
endpointRuntime?: ChatEndpointRuntime;
|
||||
generation: number;
|
||||
localEpoch: number;
|
||||
version: string;
|
||||
};
|
||||
|
||||
export type LifecycleRuntimeFence = Pick<
|
||||
RuntimeContext,
|
||||
"credentialFingerprint" | "generation"
|
||||
>;
|
||||
|
||||
export type InboundRuntimeContext = LifecycleRuntimeFence &
|
||||
Pick<RuntimeContext, "discordGatewayOwned" | "endpointRuntime">;
|
||||
|
||||
export type EndpointRecord = {
|
||||
endpoint: EndpointRow;
|
||||
assignedAgentName: string;
|
||||
connectionName: string;
|
||||
applicationId: string;
|
||||
credentialSecretRefs: ToolCredentialSecretRef[];
|
||||
};
|
||||
@@ -0,0 +1,858 @@
|
||||
import type { EndpointRow, DbTransaction, DbOrTransaction, RuntimeContext, LifecycleRuntimeFence } from "../runtime-types.js";
|
||||
|
||||
import { dueChatAction, PROVIDER_EFFECT_STALE_MS } from "../../chat-action-work.js";
|
||||
|
||||
import { notifyChatActionWork } from "../../chat-work-notifications.js";
|
||||
|
||||
import {
|
||||
and,
|
||||
asc,
|
||||
desc,
|
||||
eq,
|
||||
inArray,
|
||||
isNull,
|
||||
lte,
|
||||
or,
|
||||
sql,
|
||||
} from "drizzle-orm";
|
||||
|
||||
import type { Db } from "@paperclipai/db";
|
||||
|
||||
import {
|
||||
agentWakeupRequests,
|
||||
agents,
|
||||
chatActions,
|
||||
chatConversations,
|
||||
chatEndpointResources,
|
||||
chatEndpoints,
|
||||
chatExternalPrincipals,
|
||||
heartbeatRuns,
|
||||
issues,
|
||||
} from "@paperclipai/db";
|
||||
import type {
|
||||
SafeChatPublicationPayload,
|
||||
} from "@paperclipai/shared";
|
||||
import {
|
||||
isUuidLike,
|
||||
} from "@paperclipai/shared";
|
||||
|
||||
import { logger } from "../../../middleware/logger.js";
|
||||
|
||||
import {
|
||||
logActivity,
|
||||
} from "../../activity-log.js";
|
||||
|
||||
import { projectSafeChatPublication } from "../../chat-publication-projection.js";
|
||||
|
||||
import {
|
||||
type SlackSessionStop,
|
||||
} from "../../chat-slack-sessions.js";
|
||||
|
||||
import {
|
||||
enqueueChatRunMilestones,
|
||||
safeMilestoneText,
|
||||
} from "../../chat-run-publications.js";
|
||||
|
||||
export type SlackSessionStopTarget =
|
||||
{ id: string; kind: "run" } | { id: string; kind: "wakeup" };
|
||||
|
||||
export type SlackSessionStopPayload = {
|
||||
version: 1;
|
||||
assignedAgentId: string;
|
||||
eventTimestamp: string;
|
||||
issueId: string | null;
|
||||
occurredAt: string;
|
||||
sessionGeneration: number | null;
|
||||
target: SlackSessionStopTarget | null;
|
||||
threadId: string;
|
||||
userId: string;
|
||||
};
|
||||
|
||||
export function slackSessionStopPayload(
|
||||
value: Record<string, unknown>,
|
||||
): SlackSessionStopPayload | null {
|
||||
const target = value.target;
|
||||
const parsedTarget =
|
||||
target === null
|
||||
? null
|
||||
: target &&
|
||||
typeof target === "object" &&
|
||||
!Array.isArray(target) &&
|
||||
((target as { kind?: unknown }).kind === "run" ||
|
||||
(target as { kind?: unknown }).kind === "wakeup") &&
|
||||
typeof (target as { id?: unknown }).id === "string" &&
|
||||
isUuidLike((target as { id: string }).id)
|
||||
? (target as SlackSessionStopTarget)
|
||||
: undefined;
|
||||
if (
|
||||
value.version !== 1 ||
|
||||
typeof value.assignedAgentId !== "string" ||
|
||||
!isUuidLike(value.assignedAgentId) ||
|
||||
typeof value.eventTimestamp !== "string" ||
|
||||
!value.eventTimestamp ||
|
||||
(value.issueId !== null &&
|
||||
(typeof value.issueId !== "string" || !isUuidLike(value.issueId))) ||
|
||||
typeof value.occurredAt !== "string" ||
|
||||
!Number.isFinite(new Date(value.occurredAt).getTime()) ||
|
||||
(value.sessionGeneration !== null &&
|
||||
(typeof value.sessionGeneration !== "number" ||
|
||||
!Number.isSafeInteger(value.sessionGeneration) ||
|
||||
value.sessionGeneration < 1)) ||
|
||||
(parsedTarget !== null &&
|
||||
(value.issueId === null || value.sessionGeneration === null)) ||
|
||||
parsedTarget === undefined ||
|
||||
typeof value.threadId !== "string" ||
|
||||
!value.threadId.startsWith("slack:") ||
|
||||
typeof value.userId !== "string" ||
|
||||
!value.userId
|
||||
) {
|
||||
return null;
|
||||
}
|
||||
return {
|
||||
version: 1,
|
||||
assignedAgentId: value.assignedAgentId,
|
||||
eventTimestamp: value.eventTimestamp,
|
||||
issueId: value.issueId,
|
||||
occurredAt: value.occurredAt,
|
||||
sessionGeneration: value.sessionGeneration,
|
||||
target: parsedTarget,
|
||||
threadId: value.threadId,
|
||||
userId: value.userId,
|
||||
};
|
||||
}
|
||||
type SlackSessionStopOptions = {
|
||||
runtimeCallbackEndpoint: (tx: DbTransaction, endpointId: string, context: LifecycleRuntimeFence, statuses: EndpointRow["status"][]) => Promise<EndpointRow | null>;
|
||||
lockCurrentPrincipalAuthorization: (tx: DbOrTransaction, endpoint: EndpointRow, principalId: string) => Promise<{ allowed: boolean; userId: string | null }>;
|
||||
nonDirectDestinationAllowed: (endpoint: EndpointRow, resource: typeof chatEndpointResources.$inferSelect | null | undefined) => boolean;
|
||||
stageAuthorizedTaskControlPublication: (tx: DbOrTransaction, input: { companyId: string; conversationId: string; endpointId: string; idempotencyKey: string; issueId: string; payload: SafeChatPublicationPayload; principalId: string }) => Promise<{ id: string }>;
|
||||
getPublicBaseUrl: () => string | null;
|
||||
heartbeat: { cancelRun?: (runId: string, reason?: string, options?: { errorCode?: string; eventMessage?: string; eventPayload?: Record<string, unknown>; resultJson?: Record<string, unknown> }) => Promise<unknown> };
|
||||
redactError: (error: unknown) => string;
|
||||
};
|
||||
|
||||
export function createSlackSessionStop(db: Db, options: SlackSessionStopOptions) {
|
||||
const { runtimeCallbackEndpoint, lockCurrentPrincipalAuthorization, nonDirectDestinationAllowed, stageAuthorizedTaskControlPublication, getPublicBaseUrl, redactError } = options;
|
||||
|
||||
|
||||
async function stageSlackSessionStop(
|
||||
endpoint: EndpointRow,
|
||||
event: SlackSessionStop,
|
||||
runtimeContext: RuntimeContext,
|
||||
): Promise<string | null> {
|
||||
return db.transaction(async (tx) => {
|
||||
const currentEndpoint = await runtimeCallbackEndpoint(
|
||||
tx,
|
||||
endpoint.id,
|
||||
runtimeContext,
|
||||
["active"],
|
||||
);
|
||||
if (!currentEndpoint || currentEndpoint.provider !== "slack") return null;
|
||||
|
||||
const conversation = await tx
|
||||
.select()
|
||||
.from(chatConversations)
|
||||
.where(
|
||||
and(
|
||||
eq(chatConversations.companyId, currentEndpoint.companyId),
|
||||
eq(chatConversations.endpointId, currentEndpoint.id),
|
||||
eq(chatConversations.externalThreadId, event.threadId),
|
||||
inArray(chatConversations.state, ["active", "waiting"]),
|
||||
),
|
||||
)
|
||||
.orderBy(desc(chatConversations.sessionGeneration))
|
||||
.limit(1)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
const principal = await tx
|
||||
.select()
|
||||
.from(chatExternalPrincipals)
|
||||
.where(
|
||||
and(
|
||||
eq(chatExternalPrincipals.companyId, currentEndpoint.companyId),
|
||||
eq(chatExternalPrincipals.provider, "slack"),
|
||||
eq(
|
||||
chatExternalPrincipals.providerAccountId,
|
||||
currentEndpoint.providerAccountId ?? "unknown",
|
||||
),
|
||||
eq(chatExternalPrincipals.externalId, event.userId),
|
||||
eq(chatExternalPrincipals.kind, "user"),
|
||||
eq(chatExternalPrincipals.isBot, false),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
const issue = conversation
|
||||
? await tx
|
||||
.select()
|
||||
.from(issues)
|
||||
.where(
|
||||
and(
|
||||
eq(issues.companyId, currentEndpoint.companyId),
|
||||
eq(issues.id, conversation.issueId),
|
||||
eq(issues.assigneeAgentId, currentEndpoint.assignedAgentId),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null)
|
||||
: null;
|
||||
|
||||
// Slack carries fractional event timestamps. Treat that exact instant as
|
||||
// the cancellation boundary so a run admitted even milliseconds after
|
||||
// Stop was clicked can never be mistaken for the stopped session.
|
||||
const eventCutoff = event.occurredAt;
|
||||
const futureEvent = event.occurredAt.getTime() > Date.now() + 60_000;
|
||||
let target: SlackSessionStopTarget | null = null;
|
||||
if (issue && !futureEvent && issue.executionRunId) {
|
||||
const run = await tx
|
||||
.select({
|
||||
agentId: heartbeatRuns.agentId,
|
||||
companyId: heartbeatRuns.companyId,
|
||||
contextSnapshot: heartbeatRuns.contextSnapshot,
|
||||
createdAt: heartbeatRuns.createdAt,
|
||||
id: heartbeatRuns.id,
|
||||
startedAt: heartbeatRuns.startedAt,
|
||||
})
|
||||
.from(heartbeatRuns)
|
||||
.where(
|
||||
and(
|
||||
eq(heartbeatRuns.id, issue.executionRunId),
|
||||
eq(heartbeatRuns.companyId, currentEndpoint.companyId),
|
||||
eq(heartbeatRuns.agentId, currentEndpoint.assignedAgentId),
|
||||
inArray(heartbeatRuns.status, [
|
||||
"queued",
|
||||
"running",
|
||||
"scheduled_retry",
|
||||
]),
|
||||
lte(heartbeatRuns.createdAt, eventCutoff),
|
||||
or(
|
||||
isNull(heartbeatRuns.startedAt),
|
||||
lte(heartbeatRuns.startedAt, eventCutoff),
|
||||
),
|
||||
or(
|
||||
sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issue.id}`,
|
||||
sql`${heartbeatRuns.contextSnapshot}->>'taskId' = ${issue.id}`,
|
||||
),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (run) target = { id: run.id, kind: "run" };
|
||||
} else if (issue && !futureEvent && !issue.executionRunId) {
|
||||
const wakeup = await tx
|
||||
.select({ id: agentWakeupRequests.id })
|
||||
.from(agentWakeupRequests)
|
||||
.where(
|
||||
and(
|
||||
eq(agentWakeupRequests.companyId, currentEndpoint.companyId),
|
||||
eq(agentWakeupRequests.agentId, currentEndpoint.assignedAgentId),
|
||||
inArray(agentWakeupRequests.status, [
|
||||
"queued",
|
||||
"deferred_issue_execution",
|
||||
]),
|
||||
isNull(agentWakeupRequests.runId),
|
||||
lte(agentWakeupRequests.requestedAt, eventCutoff),
|
||||
or(
|
||||
sql`${agentWakeupRequests.payload}->>'issueId' = ${issue.id}`,
|
||||
sql`${agentWakeupRequests.payload}->>'taskId' = ${issue.id}`,
|
||||
sql`${agentWakeupRequests.payload}->'_paperclipWakeContext'->>'issueId' = ${issue.id}`,
|
||||
sql`${agentWakeupRequests.payload}->'_paperclipWakeContext'->>'taskId' = ${issue.id}`,
|
||||
),
|
||||
),
|
||||
)
|
||||
.orderBy(desc(agentWakeupRequests.requestedAt))
|
||||
.limit(1)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (wakeup) target = { id: wakeup.id, kind: "wakeup" };
|
||||
}
|
||||
|
||||
const payload: SlackSessionStopPayload = {
|
||||
version: 1,
|
||||
assignedAgentId: currentEndpoint.assignedAgentId,
|
||||
eventTimestamp: event.eventTimestamp,
|
||||
issueId: issue?.id ?? null,
|
||||
occurredAt: event.occurredAt.toISOString(),
|
||||
sessionGeneration: conversation?.sessionGeneration ?? null,
|
||||
target,
|
||||
threadId: event.threadId,
|
||||
userId: event.userId,
|
||||
};
|
||||
await notifyChatActionWork(tx, "slack_session_stop");
|
||||
const [inserted] = await tx
|
||||
.insert(chatActions)
|
||||
.values({
|
||||
companyId: currentEndpoint.companyId,
|
||||
endpointId: currentEndpoint.id,
|
||||
conversationId: conversation?.id ?? null,
|
||||
principalId: principal?.id ?? null,
|
||||
kind: "slack_session_stop",
|
||||
providerActionId: `slack_session_stop:${event.providerEventId}`,
|
||||
payload,
|
||||
status: "received",
|
||||
})
|
||||
.onConflictDoNothing()
|
||||
.returning({ id: chatActions.id });
|
||||
return (
|
||||
inserted?.id ??
|
||||
(await tx
|
||||
.select({ id: chatActions.id })
|
||||
.from(chatActions)
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.endpointId, currentEndpoint.id),
|
||||
eq(
|
||||
chatActions.providerActionId,
|
||||
`slack_session_stop:${event.providerEventId}`,
|
||||
),
|
||||
eq(chatActions.kind, "slack_session_stop"),
|
||||
),
|
||||
)
|
||||
.then((rows) => rows[0]?.id ?? null))
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
async function processSlackSessionStop(actionId: string): Promise<void> {
|
||||
const now = new Date();
|
||||
const staleBefore = new Date(now.getTime() - PROVIDER_EFFECT_STALE_MS);
|
||||
const action = await db
|
||||
.select()
|
||||
.from(chatActions)
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, actionId),
|
||||
eq(chatActions.kind, "slack_session_stop"),
|
||||
),
|
||||
)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!action || ["processed", "cancelled"].includes(action.status)) return;
|
||||
if (action.status === "processing" && action.updatedAt > staleBefore)
|
||||
return;
|
||||
if (action.status === "failed") {
|
||||
if (action.result?.retryable !== true) return;
|
||||
const retryAt =
|
||||
typeof action.result.retryAt === "string"
|
||||
? new Date(action.result.retryAt)
|
||||
: null;
|
||||
if (retryAt && retryAt > now) return;
|
||||
} else if (action.status !== "received" && action.status !== "processing") {
|
||||
return;
|
||||
}
|
||||
const payload = slackSessionStopPayload(action.payload);
|
||||
const previousAttempts =
|
||||
typeof action.result?.attempts === "number" &&
|
||||
Number.isSafeInteger(action.result.attempts)
|
||||
? Math.max(0, action.result.attempts)
|
||||
: 0;
|
||||
const attempts = previousAttempts + 1;
|
||||
if (!payload) {
|
||||
await db.transaction(async (tx) => {
|
||||
await notifyChatActionWork(tx, "slack_session_stop");
|
||||
return tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "failed",
|
||||
result: {
|
||||
attempts,
|
||||
code: "slack_session_stop_payload_invalid",
|
||||
retryable: false,
|
||||
},
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(eq(chatActions.id, action.id));
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
type StopClaim =
|
||||
| { kind: "cancel_run"; runId: string; userId: string }
|
||||
| { kind: "settled" }
|
||||
| { kind: "retry" };
|
||||
const claim = await db.transaction(async (tx): Promise<StopClaim> => {
|
||||
const current = await tx
|
||||
.select()
|
||||
.from(chatActions)
|
||||
.where(eq(chatActions.id, action.id))
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
const currentRetryAt =
|
||||
typeof current?.result?.retryAt === "string"
|
||||
? new Date(current.result.retryAt)
|
||||
: null;
|
||||
const currentRetryIsDue =
|
||||
currentRetryAt === null ||
|
||||
(Number.isFinite(currentRetryAt.getTime()) && currentRetryAt <= now);
|
||||
const eligible =
|
||||
current?.status === "received" ||
|
||||
(current?.status === "processing" &&
|
||||
current.updatedAt <= staleBefore &&
|
||||
(current.result?.attempts ?? 0) === previousAttempts) ||
|
||||
(current?.status === "failed" &&
|
||||
current.result?.retryable === true &&
|
||||
currentRetryIsDue &&
|
||||
(current.result?.attempts ?? 0) === previousAttempts);
|
||||
if (!current || !eligible) return { kind: "settled" };
|
||||
|
||||
const settle = async (
|
||||
code: string,
|
||||
actorUserId: string | null,
|
||||
targetKind: "run" | "wakeup" | "none",
|
||||
): Promise<StopClaim> => {
|
||||
await notifyChatActionWork(tx, "slack_session_stop");
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status:
|
||||
code === "slack_session_stop_cancelled"
|
||||
? "processed"
|
||||
: "cancelled",
|
||||
result: { attempts, code, retryable: false },
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(eq(chatActions.id, current.id));
|
||||
await logActivity(tx as unknown as Db, {
|
||||
companyId: current.companyId,
|
||||
actorType: actorUserId ? "user" : "system",
|
||||
actorId: actorUserId ?? "slack-session-stop",
|
||||
action:
|
||||
code === "slack_session_stop_cancelled"
|
||||
? "chat.slack_session_stopped"
|
||||
: "chat.slack_session_stop_filtered",
|
||||
entityType: "chat_action",
|
||||
entityId: current.id,
|
||||
details: {
|
||||
endpointId: current.endpointId,
|
||||
conversationId: current.conversationId,
|
||||
eventTimestamp: payload.eventTimestamp,
|
||||
targetKind,
|
||||
resultCode: code,
|
||||
},
|
||||
});
|
||||
return { kind: "settled" };
|
||||
};
|
||||
|
||||
if (
|
||||
!payload.target ||
|
||||
!payload.issueId ||
|
||||
!payload.sessionGeneration ||
|
||||
!current.conversationId ||
|
||||
!current.principalId
|
||||
) {
|
||||
return settle("slack_session_stop_no_current_work", null, "none");
|
||||
}
|
||||
const endpoint = await tx
|
||||
.select()
|
||||
.from(chatEndpoints)
|
||||
.where(
|
||||
and(
|
||||
eq(chatEndpoints.id, current.endpointId),
|
||||
eq(chatEndpoints.companyId, current.companyId),
|
||||
eq(chatEndpoints.provider, "slack"),
|
||||
eq(chatEndpoints.status, "active"),
|
||||
eq(chatEndpoints.assignedAgentId, payload.assignedAgentId),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
const conversation = endpoint
|
||||
? await tx
|
||||
.select()
|
||||
.from(chatConversations)
|
||||
.where(
|
||||
and(
|
||||
eq(chatConversations.id, current.conversationId),
|
||||
eq(chatConversations.companyId, current.companyId),
|
||||
eq(chatConversations.endpointId, current.endpointId),
|
||||
eq(chatConversations.issueId, payload.issueId),
|
||||
eq(chatConversations.externalThreadId, payload.threadId),
|
||||
eq(
|
||||
chatConversations.sessionGeneration,
|
||||
payload.sessionGeneration,
|
||||
),
|
||||
inArray(chatConversations.state, ["active", "waiting"]),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null)
|
||||
: null;
|
||||
const resource =
|
||||
endpoint && conversation?.resourceId
|
||||
? await tx
|
||||
.select()
|
||||
.from(chatEndpointResources)
|
||||
.where(
|
||||
and(
|
||||
eq(chatEndpointResources.id, conversation.resourceId),
|
||||
eq(chatEndpointResources.companyId, current.companyId),
|
||||
eq(chatEndpointResources.endpointId, current.endpointId),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null)
|
||||
: null;
|
||||
const principal = endpoint
|
||||
? await tx
|
||||
.select()
|
||||
.from(chatExternalPrincipals)
|
||||
.where(
|
||||
and(
|
||||
eq(chatExternalPrincipals.id, current.principalId),
|
||||
eq(chatExternalPrincipals.companyId, current.companyId),
|
||||
eq(chatExternalPrincipals.provider, "slack"),
|
||||
eq(
|
||||
chatExternalPrincipals.providerAccountId,
|
||||
endpoint.providerAccountId ?? "unknown",
|
||||
),
|
||||
eq(chatExternalPrincipals.externalId, payload.userId),
|
||||
eq(chatExternalPrincipals.kind, "user"),
|
||||
eq(chatExternalPrincipals.isBot, false),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null)
|
||||
: null;
|
||||
const authorization =
|
||||
endpoint && principal
|
||||
? await lockCurrentPrincipalAuthorization(tx, endpoint, principal.id)
|
||||
: null;
|
||||
const destinationAllowed =
|
||||
endpoint && conversation
|
||||
? conversation.isDirectMessage
|
||||
? endpoint.allowDirectMessages
|
||||
: nonDirectDestinationAllowed(endpoint, resource)
|
||||
: false;
|
||||
const issue =
|
||||
endpoint && conversation
|
||||
? await tx
|
||||
.select()
|
||||
.from(issues)
|
||||
.where(
|
||||
and(
|
||||
eq(issues.id, payload.issueId),
|
||||
eq(issues.companyId, current.companyId),
|
||||
eq(issues.assigneeAgentId, payload.assignedAgentId),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null)
|
||||
: null;
|
||||
if (
|
||||
!endpoint ||
|
||||
!conversation ||
|
||||
!principal ||
|
||||
!authorization?.allowed ||
|
||||
!authorization.userId ||
|
||||
!destinationAllowed ||
|
||||
!issue
|
||||
) {
|
||||
return settle(
|
||||
"slack_session_stop_no_longer_authorized",
|
||||
null,
|
||||
payload.target.kind,
|
||||
);
|
||||
}
|
||||
|
||||
const eventCutoff = new Date(payload.occurredAt);
|
||||
let runId: string | null = null;
|
||||
if (payload.target.kind === "run") {
|
||||
runId = payload.target.id;
|
||||
} else {
|
||||
const wakeup = await tx
|
||||
.select()
|
||||
.from(agentWakeupRequests)
|
||||
.where(
|
||||
and(
|
||||
eq(agentWakeupRequests.id, payload.target.id),
|
||||
eq(agentWakeupRequests.companyId, current.companyId),
|
||||
eq(agentWakeupRequests.agentId, payload.assignedAgentId),
|
||||
lte(agentWakeupRequests.requestedAt, eventCutoff),
|
||||
or(
|
||||
sql`${agentWakeupRequests.payload}->>'issueId' = ${issue.id}`,
|
||||
sql`${agentWakeupRequests.payload}->>'taskId' = ${issue.id}`,
|
||||
sql`${agentWakeupRequests.payload}->'_paperclipWakeContext'->>'issueId' = ${issue.id}`,
|
||||
sql`${agentWakeupRequests.payload}->'_paperclipWakeContext'->>'taskId' = ${issue.id}`,
|
||||
),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (
|
||||
wakeup &&
|
||||
!wakeup.runId &&
|
||||
["queued", "deferred_issue_execution"].includes(wakeup.status) &&
|
||||
!issue.executionRunId
|
||||
) {
|
||||
await tx
|
||||
.update(agentWakeupRequests)
|
||||
.set({
|
||||
status: "cancelled",
|
||||
finishedAt: now,
|
||||
error: "Cancelled from the bound Slack agent session",
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(agentWakeupRequests.id, wakeup.id),
|
||||
inArray(agentWakeupRequests.status, [
|
||||
"queued",
|
||||
"deferred_issue_execution",
|
||||
]),
|
||||
isNull(agentWakeupRequests.runId),
|
||||
),
|
||||
);
|
||||
const agentName = await tx
|
||||
.select({ name: agents.name })
|
||||
.from(agents)
|
||||
.where(
|
||||
and(
|
||||
eq(agents.companyId, current.companyId),
|
||||
eq(agents.id, payload.assignedAgentId),
|
||||
),
|
||||
)
|
||||
.then((rows) => rows[0]?.name ?? "Paperclip agent");
|
||||
await stageAuthorizedTaskControlPublication(tx, {
|
||||
companyId: current.companyId,
|
||||
conversationId: conversation.id,
|
||||
endpointId: current.endpointId,
|
||||
idempotencyKey: `control:stop:${current.id}`,
|
||||
issueId: issue.id,
|
||||
payload: projectSafeChatPublication({
|
||||
classification: "external",
|
||||
progressState: "failed",
|
||||
source: "safe_milestone",
|
||||
text: safeMilestoneText({
|
||||
agentName,
|
||||
errorCode: "slack_session_stopped",
|
||||
issueId: issue.id,
|
||||
milestone: "failed",
|
||||
publicBaseUrl: getPublicBaseUrl(),
|
||||
}),
|
||||
}),
|
||||
principalId: principal.id,
|
||||
});
|
||||
return settle(
|
||||
"slack_session_stop_cancelled",
|
||||
authorization.userId,
|
||||
"wakeup",
|
||||
);
|
||||
}
|
||||
runId = wakeup?.runId ?? null;
|
||||
if (!runId && wakeup?.status === "claimed") {
|
||||
await notifyChatActionWork(tx, "slack_session_stop");
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "failed",
|
||||
result: {
|
||||
attempts,
|
||||
code: "slack_session_stop_wakeup_in_flight",
|
||||
retryable: true,
|
||||
retryAt: new Date(Date.now() + 1_000).toISOString(),
|
||||
},
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(eq(chatActions.id, current.id));
|
||||
return { kind: "retry" };
|
||||
}
|
||||
}
|
||||
|
||||
const run = runId
|
||||
? await tx
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(
|
||||
and(
|
||||
eq(heartbeatRuns.id, runId),
|
||||
eq(heartbeatRuns.companyId, current.companyId),
|
||||
eq(heartbeatRuns.agentId, payload.assignedAgentId),
|
||||
payload.target.kind === "wakeup"
|
||||
? eq(heartbeatRuns.wakeupRequestId, payload.target.id)
|
||||
: and(
|
||||
lte(heartbeatRuns.createdAt, eventCutoff),
|
||||
or(
|
||||
isNull(heartbeatRuns.startedAt),
|
||||
lte(heartbeatRuns.startedAt, eventCutoff),
|
||||
),
|
||||
),
|
||||
or(
|
||||
sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issue.id}`,
|
||||
sql`${heartbeatRuns.contextSnapshot}->>'taskId' = ${issue.id}`,
|
||||
),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null)
|
||||
: null;
|
||||
if (!run) {
|
||||
return settle(
|
||||
"slack_session_stop_target_superseded",
|
||||
authorization.userId,
|
||||
payload.target.kind,
|
||||
);
|
||||
}
|
||||
if (run.status === "cancelled") {
|
||||
return settle(
|
||||
"slack_session_stop_cancelled",
|
||||
authorization.userId,
|
||||
"run",
|
||||
);
|
||||
}
|
||||
if (
|
||||
issue.executionRunId !== run.id ||
|
||||
!["queued", "running", "scheduled_retry"].includes(run.status)
|
||||
) {
|
||||
return settle(
|
||||
"slack_session_stop_target_superseded",
|
||||
authorization.userId,
|
||||
"run",
|
||||
);
|
||||
}
|
||||
await notifyChatActionWork(tx, "slack_session_stop");
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "processing",
|
||||
result: {
|
||||
attempts,
|
||||
authorizedUserId: authorization.userId,
|
||||
runId: run.id,
|
||||
},
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(eq(chatActions.id, current.id));
|
||||
return {
|
||||
kind: "cancel_run",
|
||||
runId: run.id,
|
||||
userId: authorization.userId,
|
||||
};
|
||||
});
|
||||
if (claim.kind !== "cancel_run") return;
|
||||
|
||||
try {
|
||||
if (!options.heartbeat.cancelRun) {
|
||||
throw new Error("Heartbeat cancellation is unavailable");
|
||||
}
|
||||
await options.heartbeat.cancelRun(
|
||||
claim.runId,
|
||||
"Stopped from the bound Slack agent session",
|
||||
{
|
||||
errorCode: "slack_session_stopped",
|
||||
eventMessage: "run cancelled from Slack",
|
||||
resultJson: {
|
||||
cancelledByActorType: "user",
|
||||
cancelledByUserId: claim.userId,
|
||||
},
|
||||
eventPayload: {
|
||||
endpointId: action.endpointId,
|
||||
conversationId: action.conversationId,
|
||||
provider: "slack",
|
||||
},
|
||||
},
|
||||
);
|
||||
await enqueueChatRunMilestones(db, { publicBaseUrl: getPublicBaseUrl() });
|
||||
const authoritativeRun = await db
|
||||
.select({ status: heartbeatRuns.status })
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.id, claim.runId))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
const cancellationWon = authoritativeRun?.status === "cancelled";
|
||||
await db.transaction(async (tx) => {
|
||||
await notifyChatActionWork(tx, "slack_session_stop");
|
||||
const [settled] = await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: cancellationWon ? "processed" : "cancelled",
|
||||
result: {
|
||||
attempts,
|
||||
code: cancellationWon
|
||||
? "slack_session_stop_cancelled"
|
||||
: "slack_session_stop_target_superseded",
|
||||
runId: claim.runId,
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, action.id),
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`(${chatActions.result}->>'attempts')::int = ${attempts}`,
|
||||
sql`${chatActions.result}->>'runId' = ${claim.runId}`,
|
||||
),
|
||||
)
|
||||
.returning({ id: chatActions.id });
|
||||
if (!settled) return;
|
||||
await logActivity(tx as unknown as Db, {
|
||||
companyId: action.companyId,
|
||||
actorType: "user",
|
||||
actorId: claim.userId,
|
||||
action: cancellationWon
|
||||
? "chat.slack_session_stopped"
|
||||
: "chat.slack_session_stop_filtered",
|
||||
entityType: "chat_action",
|
||||
entityId: action.id,
|
||||
details: {
|
||||
endpointId: action.endpointId,
|
||||
conversationId: action.conversationId,
|
||||
eventTimestamp: payload.eventTimestamp,
|
||||
targetKind: "run",
|
||||
resultCode: cancellationWon
|
||||
? "slack_session_stop_cancelled"
|
||||
: "slack_session_stop_target_superseded",
|
||||
},
|
||||
});
|
||||
});
|
||||
} catch (error) {
|
||||
const retryable = attempts < 5;
|
||||
await db.transaction(async (tx) => {
|
||||
await notifyChatActionWork(tx, "slack_session_stop");
|
||||
return tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "failed",
|
||||
result: {
|
||||
attempts,
|
||||
code: "slack_session_stop_cancellation_failed",
|
||||
retryable,
|
||||
...(retryable
|
||||
? {
|
||||
retryAt: new Date(
|
||||
Date.now() + Math.min(60_000, 1_000 * 2 ** (attempts - 1)),
|
||||
).toISOString(),
|
||||
}
|
||||
: {}),
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, action.id),
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`(${chatActions.result}->>'attempts')::int = ${attempts}`,
|
||||
sql`${chatActions.result}->>'runId' = ${claim.runId}`,
|
||||
),
|
||||
);
|
||||
});
|
||||
logger.warn(
|
||||
{
|
||||
actionId: action.id,
|
||||
endpointId: action.endpointId,
|
||||
attempts,
|
||||
error: redactError(error),
|
||||
},
|
||||
"Slack session stop cancellation deferred",
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
async function processPendingSlackSessionStops(
|
||||
limit = 25,
|
||||
onlyActionId?: string,
|
||||
) {
|
||||
const actions = await db.select({ id: chatActions.id }).from(chatActions)
|
||||
.where(and(dueChatAction("slack_session_stop"), onlyActionId ? eq(chatActions.id, onlyActionId) : undefined))
|
||||
.orderBy(asc(chatActions.createdAt)).limit(limit);
|
||||
for (const action of actions) await processSlackSessionStop(action.id);
|
||||
return actions.length;
|
||||
}
|
||||
return { stageSlackSessionStop, processSlackSessionStop, processPendingSlackSessionStops };
|
||||
}
|
||||
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,931 @@
|
||||
import type { EndpointRow, DbTransaction, LifecycleRuntimeFence } from "../runtime-types.js";
|
||||
|
||||
import { PROVIDER_EFFECT_STALE_MS } from "../../chat-action-work.js";
|
||||
|
||||
import type { chatCredentialMutationLease, CredentialMutationLeaseGuard } from "../../chat-credential-mutation-lease.js";
|
||||
|
||||
import {
|
||||
createHash,
|
||||
} from "node:crypto";
|
||||
import {
|
||||
and,
|
||||
asc,
|
||||
eq,
|
||||
gt,
|
||||
lte,
|
||||
or,
|
||||
sql,
|
||||
} from "drizzle-orm";
|
||||
|
||||
import type { Db } from "@paperclipai/db";
|
||||
|
||||
import {
|
||||
chatActions,
|
||||
chatEndpoints,
|
||||
toolConnections,
|
||||
} from "@paperclipai/db";
|
||||
import type {
|
||||
ChatEndpointSetupState,
|
||||
ToolCredentialSecretRef,
|
||||
} from "@paperclipai/shared";
|
||||
|
||||
import {
|
||||
telegramStopSubscriptionPlan,
|
||||
telegramStopSubscriptionConfirmed,
|
||||
} from "../../chat-telegram-stop-subscription.js";
|
||||
|
||||
import {
|
||||
notFound,
|
||||
} from "../../../errors.js";
|
||||
import { logger } from "../../../middleware/logger.js";
|
||||
|
||||
import {
|
||||
logActivity,
|
||||
} from "../../activity-log.js";
|
||||
|
||||
import { classifyChatPublicationError } from "../../chat-publication-errors.js";
|
||||
|
||||
import type { EndpointRecord } from "../runtime-types.js";
|
||||
|
||||
export const TELEGRAM_COMMANDS = [
|
||||
{ command: "task", description: "Start or continue a Paperclip task" },
|
||||
{ command: "status", description: "Show the active Paperclip task" },
|
||||
{ command: "new", description: "Start a new task after the current one" },
|
||||
{ command: "close", description: "Close the active chat conversation" },
|
||||
] as const;
|
||||
|
||||
export type TelegramMaintenancePayload = {
|
||||
version: 1;
|
||||
operation: "register_commands" | "remove_endpoint" | "stop_subscription";
|
||||
runtimeGeneration: number;
|
||||
credentialFingerprint: string;
|
||||
webhookUrlSha256?: string;
|
||||
botUserId?: string;
|
||||
};
|
||||
|
||||
export function telegramMaintenancePayload(
|
||||
payload: Record<string, unknown>,
|
||||
): TelegramMaintenancePayload | null {
|
||||
if (
|
||||
payload.version !== 1 ||
|
||||
(payload.operation !== "register_commands" &&
|
||||
payload.operation !== "remove_endpoint" &&
|
||||
payload.operation !== "stop_subscription") ||
|
||||
typeof payload.runtimeGeneration !== "number" ||
|
||||
!Number.isSafeInteger(payload.runtimeGeneration) ||
|
||||
typeof payload.credentialFingerprint !== "string" ||
|
||||
(payload.operation === "stop_subscription" &&
|
||||
(typeof payload.webhookUrlSha256 !== "string" ||
|
||||
!/^[a-f0-9]{64}$/.test(payload.webhookUrlSha256) ||
|
||||
typeof payload.botUserId !== "string" ||
|
||||
!/^[1-9][0-9]{0,15}$/.test(payload.botUserId)))
|
||||
) {
|
||||
return null;
|
||||
}
|
||||
return payload as TelegramMaintenancePayload;
|
||||
}
|
||||
type TelegramMaintenanceOptions = {
|
||||
endpointRecord: (endpointId: string) => Promise<EndpointRecord | null>;
|
||||
runtimeContextForRecord: (record: EndpointRecord) => LifecycleRuntimeFence;
|
||||
withCredentialMutationLease: ReturnType<typeof chatCredentialMutationLease>;
|
||||
resolveCredentials: (endpoint: EndpointRow) => Promise<Record<string, string>>;
|
||||
clearCredentials: (endpoint: EndpointRow, lease: CredentialMutationLeaseGuard) => Promise<void>;
|
||||
runtimeCallbackEndpoint: (tx: DbTransaction, endpointId: string, context: LifecycleRuntimeFence, statuses: EndpointRow["status"][]) => Promise<EndpointRow | null>;
|
||||
getWebhookPublicBaseUrl: () => string | null;
|
||||
fetchImpl: typeof fetch;
|
||||
runtimeGeneration: (setup: ChatEndpointSetupState) => number;
|
||||
credentialFingerprint: (refs: ToolCredentialSecretRef[]) => string;
|
||||
redactError: (error: unknown) => string;
|
||||
credentialCheckTimeoutMs: number;
|
||||
};
|
||||
|
||||
export function createTelegramMaintenance(db: Db, options: TelegramMaintenanceOptions) {
|
||||
const { endpointRecord, runtimeContextForRecord, withCredentialMutationLease, resolveCredentials, clearCredentials, runtimeCallbackEndpoint, getWebhookPublicBaseUrl, fetchImpl, runtimeGeneration, credentialFingerprint, redactError, credentialCheckTimeoutMs: PROVIDER_CREDENTIAL_CHECK_TIMEOUT_MS } = options;
|
||||
|
||||
let telegramSubscriptionScanAfter: string | null = null;
|
||||
|
||||
async function stageTelegramMaintenance(
|
||||
endpoint: EndpointRow,
|
||||
operation: TelegramMaintenancePayload["operation"],
|
||||
) {
|
||||
const record = await endpointRecord(endpoint.id);
|
||||
if (!record) throw notFound("Chat endpoint not found");
|
||||
const context = runtimeContextForRecord(record);
|
||||
const subscriptionScope =
|
||||
operation === "stop_subscription"
|
||||
? telegramStopSubscriptionScope(record.endpoint, context)
|
||||
: null;
|
||||
if (operation === "stop_subscription" && !subscriptionScope) return null;
|
||||
const operationGeneration = subscriptionScope
|
||||
? `${context.generation}:${context.credentialFingerprint}:${subscriptionScope.botUserId}:${subscriptionScope.webhookUrlSha256}`
|
||||
: operation === "register_commands"
|
||||
? `${context.generation}:${record.endpoint.setup.testStartedAt ?? record.endpoint.updatedAt.toISOString()}`
|
||||
: String(context.generation);
|
||||
const providerActionId = `telegram_maintenance:${operation}:${operationGeneration}`;
|
||||
const [inserted] = await db
|
||||
.insert(chatActions)
|
||||
.values({
|
||||
companyId: endpoint.companyId,
|
||||
endpointId: endpoint.id,
|
||||
kind: "telegram_maintenance",
|
||||
providerActionId,
|
||||
payload: {
|
||||
version: 1,
|
||||
operation,
|
||||
runtimeGeneration: context.generation,
|
||||
credentialFingerprint: context.credentialFingerprint,
|
||||
...(subscriptionScope
|
||||
? {
|
||||
webhookUrlSha256: subscriptionScope.webhookUrlSha256,
|
||||
botUserId: subscriptionScope.botUserId,
|
||||
}
|
||||
: {}),
|
||||
} satisfies TelegramMaintenancePayload,
|
||||
status: "received",
|
||||
})
|
||||
.onConflictDoNothing()
|
||||
.returning();
|
||||
return (
|
||||
inserted ??
|
||||
(await db
|
||||
.select()
|
||||
.from(chatActions)
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.endpointId, endpoint.id),
|
||||
eq(chatActions.providerActionId, providerActionId),
|
||||
),
|
||||
)
|
||||
.then((rows) => rows[0] ?? null))
|
||||
);
|
||||
}
|
||||
|
||||
async function telegramMaintenanceRequest(
|
||||
botToken: string,
|
||||
method: "deleteWebhook" | "deleteMyCommands" | "setMyCommands",
|
||||
body: Record<string, unknown>,
|
||||
): Promise<void> {
|
||||
const response = await fetchImpl(
|
||||
`https://api.telegram.org/bot${encodeURIComponent(botToken)}/${method}`,
|
||||
{
|
||||
method: "POST",
|
||||
signal: AbortSignal.timeout(PROVIDER_CREDENTIAL_CHECK_TIMEOUT_MS),
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify(body),
|
||||
},
|
||||
);
|
||||
const result = (await response.json()) as {
|
||||
ok?: boolean;
|
||||
description?: string;
|
||||
parameters?: { retry_after?: number };
|
||||
};
|
||||
if (response.ok && result.ok) return;
|
||||
throw Object.assign(
|
||||
new Error(
|
||||
`Telegram ${method} failed: ${result.description ?? response.status}`,
|
||||
),
|
||||
response.status === 429
|
||||
? {
|
||||
adapter: "telegram",
|
||||
code: "RATE_LIMITED",
|
||||
name: "AdapterRateLimitError",
|
||||
retryAfter: result.parameters?.retry_after,
|
||||
response: { status: response.status, headers: response.headers },
|
||||
}
|
||||
: {
|
||||
adapter: "telegram",
|
||||
status: response.status,
|
||||
response: { status: response.status, headers: response.headers },
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
async function processTelegramMaintenance(
|
||||
actionId: string,
|
||||
force = false,
|
||||
credentialLease?: CredentialMutationLeaseGuard,
|
||||
): Promise<"processed" | "pending" | "failed"> {
|
||||
if (!credentialLease) {
|
||||
const initialAction = await db
|
||||
.select({ endpointId: chatActions.endpointId })
|
||||
.from(chatActions)
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, actionId),
|
||||
eq(chatActions.kind, "telegram_maintenance"),
|
||||
),
|
||||
)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!initialAction) return "failed";
|
||||
const initialEndpoint = await endpointRecord(initialAction.endpointId);
|
||||
if (!initialEndpoint) return "failed";
|
||||
// Recovery sweeps run independently from endpoint configuration. Hold
|
||||
// the same durable credential lease as reconnect, pause, and removal so
|
||||
// a stale bot token cannot mutate Telegram after its final runtime fence
|
||||
// but before transport.
|
||||
return withCredentialMutationLease(initialEndpoint.endpoint, (lease) =>
|
||||
processTelegramMaintenance(actionId, force, lease),
|
||||
);
|
||||
}
|
||||
let action = await db
|
||||
.select()
|
||||
.from(chatActions)
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, actionId),
|
||||
eq(chatActions.kind, "telegram_maintenance"),
|
||||
),
|
||||
)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!action) return "failed";
|
||||
if (action.status === "processed") return "processed";
|
||||
const payload = telegramMaintenancePayload(action.payload);
|
||||
if (!payload) {
|
||||
await db
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "failed",
|
||||
result: {
|
||||
code: "telegram_maintenance_payload_invalid",
|
||||
retryable: false,
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(chatActions.id, action.id));
|
||||
return "failed";
|
||||
}
|
||||
if (action.status === "processing") {
|
||||
if (action.updatedAt > new Date(Date.now() - PROVIDER_EFFECT_STALE_MS))
|
||||
return "pending";
|
||||
await db
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "failed",
|
||||
result: {
|
||||
...(action.result ?? {}),
|
||||
retryable: true,
|
||||
retryAt: new Date().toISOString(),
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, action.id),
|
||||
eq(chatActions.status, "processing"),
|
||||
eq(chatActions.updatedAt, action.updatedAt),
|
||||
),
|
||||
);
|
||||
action = await db
|
||||
.select()
|
||||
.from(chatActions)
|
||||
.where(eq(chatActions.id, action.id))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!action) return "failed";
|
||||
}
|
||||
if (action.status === "failed") {
|
||||
if (action.result?.retryable !== true && !force) return "failed";
|
||||
const retryAt =
|
||||
typeof action.result?.retryAt === "string"
|
||||
? new Date(action.result.retryAt)
|
||||
: null;
|
||||
if (!force && retryAt && retryAt > new Date()) return "pending";
|
||||
} else if (action.status !== "received") {
|
||||
return "failed";
|
||||
}
|
||||
const attempt =
|
||||
(typeof action.result?.attempts === "number"
|
||||
? action.result.attempts
|
||||
: 0) + 1;
|
||||
let providerConfirmed = action.result?.providerConfirmed === true;
|
||||
const [claimed] = await db.transaction(async (tx) => {
|
||||
await credentialLease.assertOwned(tx);
|
||||
const rows = await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "processing",
|
||||
result: { attempts: attempt, providerConfirmed },
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, action!.id),
|
||||
eq(chatActions.status, action!.status),
|
||||
),
|
||||
)
|
||||
.returning();
|
||||
await credentialLease.assertOwned(tx);
|
||||
return rows;
|
||||
});
|
||||
if (!claimed) return "pending";
|
||||
|
||||
try {
|
||||
const record = await endpointRecord(action.endpointId);
|
||||
if (!record || record.endpoint.provider !== "telegram") {
|
||||
throw Object.assign(new Error("Telegram endpoint is unavailable"), {
|
||||
code: "CHAT_PROVIDER_PRETRANSPORT_REJECTED",
|
||||
});
|
||||
}
|
||||
if (
|
||||
payload.operation === "remove_endpoint" &&
|
||||
record.credentialSecretRefs.length === 0
|
||||
) {
|
||||
// Recovery may observe the narrow crash window after provider cleanup
|
||||
// and credential clearing but before the action receipt was committed.
|
||||
// Empty refs prove removal cleanup crossed its final local boundary;
|
||||
// settle the idempotent action instead of misclassifying it as a
|
||||
// credential rotation.
|
||||
await db.transaction(async (tx) => {
|
||||
await credentialLease.assertOwned(tx);
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "processed",
|
||||
result: { attempts: attempt, providerConfirmed: true },
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, action!.id),
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`(${chatActions.result}->>'attempts')::int = ${attempt}`,
|
||||
),
|
||||
);
|
||||
await credentialLease.assertOwned(tx);
|
||||
});
|
||||
return "processed";
|
||||
}
|
||||
if (
|
||||
credentialFingerprint(record.credentialSecretRefs) !==
|
||||
payload.credentialFingerprint ||
|
||||
(payload.operation !== "remove_endpoint" &&
|
||||
runtimeGeneration(record.endpoint.setup) !==
|
||||
payload.runtimeGeneration)
|
||||
) {
|
||||
await db.transaction(async (tx) => {
|
||||
await credentialLease.assertOwned(tx);
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "cancelled",
|
||||
result: {
|
||||
attempts: attempt,
|
||||
code: "telegram_maintenance_credentials_superseded",
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, action!.id),
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`(${chatActions.result}->>'attempts')::int = ${attempt}`,
|
||||
),
|
||||
);
|
||||
await credentialLease.assertOwned(tx);
|
||||
});
|
||||
return "failed";
|
||||
}
|
||||
if (!providerConfirmed || payload.operation === "stop_subscription") {
|
||||
const credentials = await resolveCredentials(record.endpoint);
|
||||
if (!credentials.botToken) {
|
||||
throw Object.assign(new Error("Telegram bot token is unavailable"), {
|
||||
code: "CHAT_PROVIDER_PRETRANSPORT_REJECTED",
|
||||
});
|
||||
}
|
||||
if (payload.operation === "stop_subscription") {
|
||||
await repairTelegramStopSubscription(
|
||||
record.endpoint,
|
||||
payload,
|
||||
credentialLease,
|
||||
);
|
||||
providerConfirmed = true;
|
||||
} else if (payload.operation === "register_commands") {
|
||||
await credentialLease.assertOwned();
|
||||
await telegramMaintenanceRequest(
|
||||
credentials.botToken,
|
||||
"setMyCommands",
|
||||
{ commands: TELEGRAM_COMMANDS },
|
||||
);
|
||||
providerConfirmed = true;
|
||||
} else {
|
||||
await credentialLease.assertOwned();
|
||||
await telegramMaintenanceRequest(
|
||||
credentials.botToken,
|
||||
"deleteWebhook",
|
||||
{ drop_pending_updates: false },
|
||||
);
|
||||
await credentialLease.assertOwned();
|
||||
await telegramMaintenanceRequest(
|
||||
credentials.botToken,
|
||||
"deleteMyCommands",
|
||||
{},
|
||||
);
|
||||
providerConfirmed = true;
|
||||
}
|
||||
await credentialLease.assertOwned();
|
||||
await db.transaction(async (tx) => {
|
||||
await credentialLease.assertOwned(tx);
|
||||
const confirmed = await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
result: { attempts: attempt, providerConfirmed: true },
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, action.id),
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`(${chatActions.result}->>'attempts')::int = ${attempt}`,
|
||||
),
|
||||
)
|
||||
.returning({ id: chatActions.id });
|
||||
if (!confirmed.length) {
|
||||
throw new Error(
|
||||
"Telegram maintenance ownership changed before provider confirmation",
|
||||
);
|
||||
}
|
||||
await credentialLease.assertOwned(tx);
|
||||
});
|
||||
}
|
||||
if (payload.operation === "remove_endpoint") {
|
||||
await credentialLease.assertOwned();
|
||||
await clearCredentials(record.endpoint, credentialLease);
|
||||
}
|
||||
await db.transaction(async (tx) => {
|
||||
await credentialLease.assertOwned(tx);
|
||||
const processed = await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "processed",
|
||||
result: { attempts: attempt, providerConfirmed: true },
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, action.id),
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`(${chatActions.result}->>'attempts')::int = ${attempt}`,
|
||||
),
|
||||
)
|
||||
.returning({ id: chatActions.id });
|
||||
if (!processed.length) {
|
||||
throw new Error(
|
||||
"Telegram maintenance ownership changed before success settlement",
|
||||
);
|
||||
}
|
||||
await credentialLease.assertOwned(tx);
|
||||
});
|
||||
await logActivity(db, {
|
||||
companyId: record.endpoint.companyId,
|
||||
actorType: "system",
|
||||
actorId: "chat-maintenance",
|
||||
action: `chat_endpoint.telegram_${payload.operation}_completed`,
|
||||
entityType: "tool_connection",
|
||||
entityId: record.endpoint.connectionId,
|
||||
details: { endpointId: record.endpoint.id, attempt },
|
||||
});
|
||||
return "processed";
|
||||
} catch (error) {
|
||||
const disposition = classifyChatPublicationError(error, attempt);
|
||||
// Webhook/menu mutations are idempotent provider operations. A lost
|
||||
// response is safe to retry, and an explicit provider Retry-After must
|
||||
// remain durable rather than occupying an in-memory timer.
|
||||
const retryable =
|
||||
disposition.kind === "retry" || disposition.kind === "delivery_unknown";
|
||||
const retryAfterMs =
|
||||
disposition.kind === "retry"
|
||||
? disposition.retryAfterMs
|
||||
: Math.min(60_000, 1_000 * 2 ** Math.max(0, attempt - 1));
|
||||
const retryAt = retryable
|
||||
? new Date(Date.now() + retryAfterMs).toISOString()
|
||||
: null;
|
||||
const diagnostic = redactError(error);
|
||||
await db
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "failed",
|
||||
result: {
|
||||
attempts: attempt,
|
||||
code: `telegram_maintenance_${disposition.kind}`,
|
||||
providerConfirmed:
|
||||
providerConfirmed || action.result?.providerConfirmed === true,
|
||||
redactedError: diagnostic,
|
||||
retryable,
|
||||
retryAt,
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, action.id),
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`(${chatActions.result}->>'attempts')::int = ${attempt}`,
|
||||
),
|
||||
);
|
||||
await logActivity(db, {
|
||||
companyId: action.companyId,
|
||||
actorType: "system",
|
||||
actorId: "chat-maintenance",
|
||||
action: `chat_endpoint.telegram_${payload.operation}_deferred`,
|
||||
entityType: "tool_connection",
|
||||
entityId:
|
||||
(await endpointRecord(action.endpointId))?.endpoint.connectionId ??
|
||||
action.endpointId,
|
||||
details: {
|
||||
endpointId: action.endpointId,
|
||||
attempt,
|
||||
disposition: disposition.kind,
|
||||
retryAt,
|
||||
},
|
||||
}).catch((auditError) => {
|
||||
logger.error(
|
||||
{ endpointId: action!.endpointId, error: redactError(auditError) },
|
||||
"could not persist Telegram maintenance deferral audit",
|
||||
);
|
||||
});
|
||||
logger.warn(
|
||||
{
|
||||
endpointId: action.endpointId,
|
||||
operation: payload.operation,
|
||||
retryAt,
|
||||
},
|
||||
"Telegram maintenance was deferred",
|
||||
);
|
||||
return retryable ? "pending" : "failed";
|
||||
}
|
||||
}
|
||||
|
||||
async function processPendingTelegramMaintenance(limit = 25) {
|
||||
// Keyset scanning bounds each sweep without starving endpoints behind a
|
||||
// page whose subscriptions were already confirmed. No provider I/O here.
|
||||
const candidates = await db
|
||||
.select({ endpoint: chatEndpoints })
|
||||
.from(chatEndpoints)
|
||||
.innerJoin(
|
||||
toolConnections,
|
||||
and(
|
||||
eq(toolConnections.id, chatEndpoints.connectionId),
|
||||
eq(toolConnections.companyId, chatEndpoints.companyId),
|
||||
),
|
||||
)
|
||||
.where(
|
||||
and(
|
||||
eq(chatEndpoints.provider, "telegram"),
|
||||
eq(chatEndpoints.status, "active"),
|
||||
eq(toolConnections.status, "active"),
|
||||
eq(toolConnections.enabled, true),
|
||||
telegramSubscriptionScanAfter
|
||||
? gt(chatEndpoints.id, telegramSubscriptionScanAfter)
|
||||
: undefined,
|
||||
),
|
||||
)
|
||||
.orderBy(asc(chatEndpoints.id))
|
||||
.limit(Math.max(1, Math.min(limit, 100)));
|
||||
telegramSubscriptionScanAfter =
|
||||
candidates.length >= Math.max(1, Math.min(limit, 100))
|
||||
? candidates.at(-1)!.endpoint.id
|
||||
: null;
|
||||
for (const { endpoint } of candidates) {
|
||||
try {
|
||||
const record = await endpointRecord(endpoint.id);
|
||||
if (
|
||||
record &&
|
||||
!(await hasTelegramStopSubscription(
|
||||
record.endpoint,
|
||||
runtimeContextForRecord(record),
|
||||
))
|
||||
)
|
||||
await stageTelegramMaintenance(record.endpoint, "stop_subscription");
|
||||
} catch {
|
||||
logger.warn(
|
||||
{ endpointId: endpoint.id },
|
||||
"Telegram subscription maintenance could not be staged",
|
||||
);
|
||||
}
|
||||
}
|
||||
const now = new Date();
|
||||
const staleBefore = new Date(now.getTime() - PROVIDER_EFFECT_STALE_MS);
|
||||
const actions = await db
|
||||
.select({ id: chatActions.id })
|
||||
.from(chatActions)
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.kind, "telegram_maintenance"),
|
||||
sql`(${chatActions.payload}->>'operation' <> 'stop_subscription' or exists (
|
||||
select 1 from ${chatEndpoints} inner join ${toolConnections}
|
||||
on ${toolConnections.id} = ${chatEndpoints.connectionId} and ${toolConnections.companyId} = ${chatEndpoints.companyId}
|
||||
where ${chatEndpoints.id} = ${chatActions.endpointId} and ${chatEndpoints.companyId} = ${chatActions.companyId}
|
||||
and ${chatEndpoints.provider} = 'telegram' and ${chatEndpoints.status} = 'active'
|
||||
and ${toolConnections.status} = 'active' and ${toolConnections.enabled} = true
|
||||
))`,
|
||||
or(
|
||||
eq(chatActions.status, "received"),
|
||||
and(
|
||||
eq(chatActions.status, "processing"),
|
||||
lte(chatActions.updatedAt, staleBefore),
|
||||
),
|
||||
and(
|
||||
eq(chatActions.status, "failed"),
|
||||
sql`coalesce(${chatActions.result}->>'retryable', 'false') = 'true'`,
|
||||
sql`(${chatActions.result}->>'retryAt' is null or (${chatActions.result}->>'retryAt')::timestamptz <= ${now.toISOString()}::timestamptz)`,
|
||||
),
|
||||
),
|
||||
),
|
||||
)
|
||||
.orderBy(asc(chatActions.createdAt))
|
||||
.limit(limit);
|
||||
for (const action of actions) await processTelegramMaintenance(action.id);
|
||||
return actions.length;
|
||||
}
|
||||
|
||||
function telegramStopSubscriptionScope(
|
||||
endpoint: EndpointRow,
|
||||
context: LifecycleRuntimeFence,
|
||||
) {
|
||||
if (
|
||||
endpoint.provider !== "telegram" ||
|
||||
!endpoint.botExternalId ||
|
||||
!getWebhookPublicBaseUrl()
|
||||
)
|
||||
return null;
|
||||
const webhookUrlSha256 = createHash("sha256")
|
||||
.update(
|
||||
`${getWebhookPublicBaseUrl()}/api/chat-webhooks/${endpoint.publicId}/telegram`,
|
||||
)
|
||||
.digest("hex");
|
||||
return {
|
||||
version: 1,
|
||||
botUserId: endpoint.botExternalId,
|
||||
runtimeGeneration: context.generation,
|
||||
credentialFingerprint: context.credentialFingerprint,
|
||||
webhookUrlSha256,
|
||||
};
|
||||
}
|
||||
|
||||
async function hasTelegramStopSubscription(
|
||||
endpoint: EndpointRow,
|
||||
context: LifecycleRuntimeFence,
|
||||
): Promise<boolean> {
|
||||
const scope = telegramStopSubscriptionScope(endpoint, context);
|
||||
if (!scope) return false;
|
||||
const [receipt] = await db
|
||||
.select()
|
||||
.from(chatActions)
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.companyId, endpoint.companyId),
|
||||
eq(chatActions.endpointId, endpoint.id),
|
||||
eq(chatActions.kind, "telegram_stop_subscription"),
|
||||
eq(
|
||||
chatActions.providerActionId,
|
||||
`telegram-stop-subscription:${scope.runtimeGeneration}:${scope.credentialFingerprint}:${scope.webhookUrlSha256}`,
|
||||
),
|
||||
eq(chatActions.status, "processed"),
|
||||
),
|
||||
)
|
||||
.limit(1);
|
||||
return (
|
||||
!!receipt &&
|
||||
receipt.result?.code === "telegram_stop_subscription_confirmed" &&
|
||||
Object.keys(receipt.payload).length === Object.keys(scope).length &&
|
||||
Object.entries(scope).every(
|
||||
([key, value]) => receipt.payload[key] === value,
|
||||
)
|
||||
);
|
||||
}
|
||||
|
||||
async function recordTelegramStopSubscription(
|
||||
endpoint: EndpointRow,
|
||||
context: LifecycleRuntimeFence,
|
||||
lease: CredentialMutationLeaseGuard,
|
||||
): Promise<void> {
|
||||
const scope = telegramStopSubscriptionScope(endpoint, context);
|
||||
if (!scope)
|
||||
throw new Error("Telegram Stop subscription scope is unavailable");
|
||||
await db.transaction(async (tx) => {
|
||||
await lease.assertOwned(tx);
|
||||
const current = await runtimeCallbackEndpoint(tx, endpoint.id, context, [
|
||||
"verifying",
|
||||
"active",
|
||||
]);
|
||||
if (
|
||||
!current ||
|
||||
current.botExternalId !== scope.botUserId ||
|
||||
current.publicId !== endpoint.publicId
|
||||
)
|
||||
throw new Error("Telegram Stop subscription ownership changed");
|
||||
const [connection] = await tx
|
||||
.select()
|
||||
.from(toolConnections)
|
||||
.where(
|
||||
and(
|
||||
eq(toolConnections.id, current.connectionId),
|
||||
eq(toolConnections.companyId, current.companyId),
|
||||
),
|
||||
)
|
||||
.for("share");
|
||||
if (
|
||||
!connection?.enabled ||
|
||||
connection.status !== "active" ||
|
||||
credentialFingerprint(connection.credentialSecretRefs) !==
|
||||
context.credentialFingerprint
|
||||
)
|
||||
throw new Error(
|
||||
"Telegram Stop subscription credential ownership changed",
|
||||
);
|
||||
await tx
|
||||
.insert(chatActions)
|
||||
.values({
|
||||
companyId: endpoint.companyId,
|
||||
endpointId: endpoint.id,
|
||||
kind: "telegram_stop_subscription",
|
||||
providerActionId: `telegram-stop-subscription:${scope.runtimeGeneration}:${scope.credentialFingerprint}:${scope.webhookUrlSha256}`,
|
||||
payload: scope,
|
||||
status: "processed",
|
||||
result: { code: "telegram_stop_subscription_confirmed" },
|
||||
})
|
||||
.onConflictDoUpdate({
|
||||
target: [chatActions.endpointId, chatActions.providerActionId],
|
||||
set: {
|
||||
payload: scope,
|
||||
status: "processed",
|
||||
result: { code: "telegram_stop_subscription_confirmed" },
|
||||
updatedAt: new Date(),
|
||||
},
|
||||
});
|
||||
await lease.assertOwned(tx);
|
||||
});
|
||||
}
|
||||
|
||||
async function repairTelegramStopSubscription(
|
||||
endpoint: EndpointRow,
|
||||
payload: TelegramMaintenancePayload,
|
||||
lease: CredentialMutationLeaseGuard,
|
||||
): Promise<void> {
|
||||
const plainObject = (value: unknown): Record<string, unknown> =>
|
||||
value !== null && typeof value === "object" && !Array.isArray(value)
|
||||
? (value as Record<string, unknown>)
|
||||
: {};
|
||||
const context = {
|
||||
generation: payload.runtimeGeneration,
|
||||
credentialFingerprint: payload.credentialFingerprint,
|
||||
};
|
||||
const scope = telegramStopSubscriptionScope(endpoint, context);
|
||||
const reject = () =>
|
||||
Object.assign(
|
||||
new Error(
|
||||
"Telegram subscription cannot be safely upgraded under current ownership",
|
||||
),
|
||||
{
|
||||
code: "CHAT_PROVIDER_PRETRANSPORT_REJECTED",
|
||||
},
|
||||
);
|
||||
if (
|
||||
!scope ||
|
||||
scope.webhookUrlSha256 !== payload.webhookUrlSha256 ||
|
||||
scope.botUserId !== payload.botUserId
|
||||
)
|
||||
throw reject();
|
||||
const assertCurrent = async () =>
|
||||
db.transaction(async (tx) => {
|
||||
await lease.assertOwned(tx);
|
||||
const current = await runtimeCallbackEndpoint(
|
||||
tx,
|
||||
endpoint.id,
|
||||
context,
|
||||
["active"],
|
||||
);
|
||||
if (
|
||||
!current ||
|
||||
current.companyId !== endpoint.companyId ||
|
||||
current.botExternalId !== scope.botUserId ||
|
||||
current.publicId !== endpoint.publicId
|
||||
)
|
||||
throw reject();
|
||||
const [connection] = await tx
|
||||
.select()
|
||||
.from(toolConnections)
|
||||
.where(
|
||||
and(
|
||||
eq(toolConnections.id, current.connectionId),
|
||||
eq(toolConnections.companyId, current.companyId),
|
||||
),
|
||||
)
|
||||
.for("share");
|
||||
if (
|
||||
!connection?.enabled ||
|
||||
connection.status !== "active" ||
|
||||
credentialFingerprint(connection.credentialSecretRefs) !==
|
||||
context.credentialFingerprint
|
||||
)
|
||||
throw reject();
|
||||
await lease.assertOwned(tx);
|
||||
});
|
||||
await assertCurrent();
|
||||
// A recovered providerConfirmed flag is not a subscription capability.
|
||||
// Only the separately committed exact current receipt can avoid fresh I/O.
|
||||
if (await hasTelegramStopSubscription(endpoint, context)) return;
|
||||
const credentials = await resolveCredentials(endpoint);
|
||||
if (
|
||||
!credentials.botToken ||
|
||||
typeof credentials.webhookSecret !== "string" ||
|
||||
!/^[A-Za-z0-9_-]{1,256}$/.test(credentials.webhookSecret)
|
||||
)
|
||||
throw reject();
|
||||
const request = async (
|
||||
method: "getMe" | "getWebhookInfo" | "setWebhook",
|
||||
body: Record<string, unknown> = {},
|
||||
) => {
|
||||
await assertCurrent();
|
||||
let response: Response;
|
||||
try {
|
||||
response = await fetchImpl(
|
||||
`https://api.telegram.org/bot${encodeURIComponent(credentials.botToken!)}/${method}`,
|
||||
{
|
||||
method: "POST",
|
||||
redirect: "error",
|
||||
signal: AbortSignal.timeout(PROVIDER_CREDENTIAL_CHECK_TIMEOUT_MS),
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify(body),
|
||||
},
|
||||
);
|
||||
} catch {
|
||||
// Network errors can embed a token-bearing URL. Keep maintenance
|
||||
// diagnostics closed rather than forwarding provider/transport prose.
|
||||
throw new Error(
|
||||
"Telegram subscription transport did not confirm completion",
|
||||
);
|
||||
}
|
||||
let value: unknown;
|
||||
try {
|
||||
value = await response.json();
|
||||
} catch {
|
||||
throw new Error("Telegram subscription returned an invalid response");
|
||||
}
|
||||
const result = plainObject(value);
|
||||
if (!response.ok || result.ok !== true) {
|
||||
const retryAfter = plainObject(result.parameters).retry_after;
|
||||
throw Object.assign(
|
||||
new Error("Telegram subscription request was not confirmed"),
|
||||
response.status === 429
|
||||
? {
|
||||
adapter: "telegram",
|
||||
code: "RATE_LIMITED",
|
||||
name: "AdapterRateLimitError",
|
||||
retryAfter:
|
||||
typeof retryAfter === "number" && Number.isFinite(retryAfter)
|
||||
? retryAfter
|
||||
: undefined,
|
||||
response: {
|
||||
status: response.status,
|
||||
headers: response.headers,
|
||||
},
|
||||
}
|
||||
: {
|
||||
adapter: "telegram",
|
||||
status: response.status,
|
||||
response: {
|
||||
status: response.status,
|
||||
headers: response.headers,
|
||||
},
|
||||
},
|
||||
);
|
||||
}
|
||||
return result.result;
|
||||
};
|
||||
const identity = plainObject(await request("getMe"));
|
||||
if (
|
||||
identity.is_bot !== true ||
|
||||
!Number.isSafeInteger(identity.id) ||
|
||||
String(identity.id) !== scope.botUserId
|
||||
)
|
||||
throw reject();
|
||||
const url = `${getWebhookPublicBaseUrl()}/api/chat-webhooks/${endpoint.publicId}/telegram`;
|
||||
const plan = telegramStopSubscriptionPlan(
|
||||
await request("getWebhookInfo"),
|
||||
url,
|
||||
);
|
||||
if (!plan) throw reject();
|
||||
// No URL repoint, credential rotation, certificate guess, IP pin, or queue
|
||||
// dropping. A successful SET proves the stored secret; GET cannot do so.
|
||||
if (
|
||||
(await request("setWebhook", {
|
||||
url,
|
||||
secret_token: credentials.webhookSecret,
|
||||
...plan,
|
||||
drop_pending_updates: false,
|
||||
})) !== true
|
||||
)
|
||||
throw new Error("Telegram subscription mutation was not confirmed");
|
||||
if (
|
||||
!telegramStopSubscriptionConfirmed(
|
||||
await request("getWebhookInfo"),
|
||||
url,
|
||||
plan,
|
||||
)
|
||||
)
|
||||
throw new Error("Telegram subscription verification changed");
|
||||
await assertCurrent();
|
||||
await recordTelegramStopSubscription(endpoint, context, lease);
|
||||
}
|
||||
return { stageTelegramMaintenance, processTelegramMaintenance, processPendingTelegramMaintenance, hasTelegramStopSubscription, recordTelegramStopSubscription };
|
||||
}
|
||||
Reference in new issue
Block a user