mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-25 09:41:04 +01:00
refactor(chat): isolate Telegram, Slack, GitHub and Discord workflows (#15976)
## Thinking Path > - Paperclip manages AI agents and their work. > - Chat channels connect that work to external conversations. > - The shared chat service still contains provider-specific workflows. > - Those workflows are hard to inspect in a 34,000-line file. > - This stack extracts Telegram, Slack, GitHub and Discord workflows into their channel folders. > - Shared authorization and service lifetime remain explicit. ## Linked Issues or Issue Description Refs #15823, #15834, #15836, #15844 and #15845. Root of the second behavior-preserving channel-folder refactor stack. The dependent PRs #15979, #15981, #15985 and #15987 have merged into this branch; this PR lands all five extractions on master. ## What Changed - Move Telegram maintenance and subscription repair into `chat-channels/telegram/maintenance.ts`. - Move Slack session stopping and task starts into `chat-channels/slack/session-stop.ts` and `task-start.ts`. - Move GitHub webhook intake and recovery into `chat-channels/github/webhook-ingress.ts`. - Move Discord command registration and handling into `chat-channels/discord/commands.ts`. - Share type-only runtime records while preserving shared authorization, transactions, live lease checks and worker ownership. - Reduce the original service from 34,603 to 29,897 lines, removing 4,706 lines (13.6%). ## Verification - All 205 selected integration cases passed before and after their respective extractions: Telegram 41, Slack session stop 2 multi-step scenarios, GitHub 84, Slack task start 25 and Discord 53. Each dependent PR records its focused reproduction command. - Server typecheck, module boundaries and whitespace checks passed. - Per-slice syntax comparisons passed for moved declarations and remaining service statements, allowing only the documented live callback substitutions. - Tests exercise the public service with disposable databases and fixture provider transports. They do not claim live third-party execution. - Cumulative workspace typecheck and build passed on the final stack. - Current-head CI passed on the combined stack commit `d5b37dbe512e3e199932b6608cdc17639eda555a`; Greptile is 5/5 with no unresolved threads. - The [original serialized shard](https://github.com/paperclipai/paperclip/actions/runs/38266518817/job/114870022249) timed out in unchanged `adapter-auth-signal-routes.test.ts`. The exact case passed locally without changes, and the [single job retry](https://github.com/paperclipai/paperclip/actions/runs/38266518817/job/114874019933) passed. The original timeout remains recorded. Reproduce the original Telegram slice’s focused suite: ```sh pnpm exec vitest run server/src/__tests__/chat-channels.integration.test.ts -t 'Telegram durable private draft Stop|Telegram maintenance|Telegram webhook|Telegram.*(pause|remove)|Telegram.*commands' ``` - Cumulative local `pnpm test:run` exited 1: the first server group had **18,770 passed, 105 skipped, and one timeout** in unchanged `server/src/services/workspace-git-snapshot-streaming.test.ts:56` (300,000 ms on macOS). The fail-fast runner did not reach its later workspace and serialized groups. An unchanged isolated rerun passed using the existing timeout; this does not turn the original full run green. The test and its Git utility code are outside this refactor; all current-head CI jobs passed. - Reproduce that diagnostic alone: `pnpm exec vitest run server/src/services/workspace-git-snapshot-streaming.test.ts`. ## Risks - Credential lease ownership and provider-confirmation ordering must remain unchanged. Existing rotation, lease-loss, recovery and durable Stop cases pass. - The channel factories create no workers or timers and preserve live runtime, shutdown and gateway-ownership reads. ## Model Used OpenAI Codex, GPT-6 family, with reasoning and repository tools. The exact runtime model ID and context-window size are not exposed in this session. No separate agent was used. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [ ] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
a9830414c7
commit
a2ac5c2715
8 files changed
+5254
-4721
No files matched your search
@@ -0,0 +1,119 @@
|
||||
# Chat channel workflow refactor, batch two
|
||||
|
||||
Base: `37913e4907324838ae2c8c94dc8155002b6d4223` (fresh master after the previous stack merged). The previous five PRs
|
||||
(#15823, #15834, #15836, #15844, #15845) are merged. The service remains
|
||||
34,603 lines. This batch creates five dependent PRs; merging is separate work.
|
||||
|
||||
| Order / responsibility | Destination | Approximate lines moved | Contract and ownership | Verification |
|
||||
| --- | --- | --- | --- | --- |
|
||||
| 1. Telegram maintenance and Stop subscription repair | `chat-channels/telegram/maintenance.ts` | 850 | Endpoint lookup, credential lease, credential resolution, runtime fence, live webhook URL, fetch. Own the per-service subscription scan cursor. Preserve claim/confirmation/settlement ordering. | Existing maintenance, rotation, lost-lease, subscription recovery and durable draft Stop integration cases before/after. |
|
||||
| 2. Slack session stopping | `chat-channels/slack/session-stop.ts` | 750 | Stage and consume durable stop receipts; call shared principal authorization and task-control publication. Preserve captured target, event chronology and transaction locks. | Existing Slack stop, stale-session, authorization and failure-recovery cases before/after. |
|
||||
| 3. GitHub durable webhook ingress and recovery | `chat-channels/github/webhook-ingress.ts` | 1,400 | Persist verified ingress, replay through the shared webhook entry point, recover provider failures. Keep persistence lease, live shutdown check and current endpoint/credential fences. | Existing signed webhook, restart/replay, recovery and stale-source integration cases before/after. |
|
||||
| 4. Slack task-start admission | `chat-channels/slack/task-start.ts` | 920 | Recover task starter receipt, revalidate principal/resource/runtime and admit via the shared message path. Preserve lease ownership, synthetic message identity and claim barrier. | Existing slash-task concurrency, partial-post and recovery cases before/after. |
|
||||
| 5. Discord native commands | `chat-channels/discord/commands.ts` | 550 | Reconcile command registration and admit native controls; shared service retains gateway ownership and authorization. Read gateway ownership live. | Existing command registration, stale runtime, gateway ownership and native command cases before/after. |
|
||||
|
||||
Keep message admission, shared control authority, provider runtime startup,
|
||||
publication transport and service shutdown together. Each factory receives named,
|
||||
typed dependencies. Do not copy mutable maps or snapshot live values. Shared
|
||||
runtime record types may move to a type-only sibling module; do not introduce a
|
||||
runtime import back to the original service.
|
||||
|
||||
For each slice: baseline existing public-service tests, mechanical extraction,
|
||||
syntax/body parity, same tests afterward, server typecheck and module boundaries,
|
||||
then PR checks and current-head review. Run broad local gates once at final
|
||||
handoff. Record failures honestly; a retry does not erase the original failure.
|
||||
|
||||
## Slice 1: Telegram maintenance
|
||||
|
||||
- Service: 34,603 lines / 1,337,182 bytes → 33,718 lines / 1,306,801 bytes.
|
||||
- `telegram/maintenance.ts`: 931 lines / 32,792 bytes; type-only shared runtime
|
||||
records: 38 lines / 1,200 bytes.
|
||||
- Twelve moved declarations and six shared types preserve token-level syntax;
|
||||
all remaining service statements are unchanged. The module owns the existing
|
||||
scan cursor and creates no timers or background work. Its live callbacks retain
|
||||
credential-lease checks and runtime fencing around every provider mutation.
|
||||
- The same 41 existing Telegram public-service cases passed before and after.
|
||||
Server typecheck, module boundaries and whitespace checks passed. Full local
|
||||
repository gates are deferred to the cumulative handoff; remote CI/review will
|
||||
be recorded in this slice's PR.
|
||||
- Reassessment: keep Telegram draft Stop admission and publication transport in
|
||||
the orchestrator. Slack durable session stopping remains the next cohesive cut.
|
||||
|
||||
## Slice 2: Slack session stopping
|
||||
|
||||
- Service: 33,718 lines / 1,306,801 bytes → 32,928 lines / 1,279,554 bytes.
|
||||
- `slack/session-stop.ts`: 858 lines / 30,166 bytes.
|
||||
- All six moved declarations and all remaining service statements preserve
|
||||
token-level syntax. A getter retains the original live heartbeat property
|
||||
access. Shared principal authorization and publication staging stay in the
|
||||
caller; the entire captured-target claim and settlement transactions move.
|
||||
- Both existing public-service scenarios passed before and after. They cover
|
||||
exact-run cancellation, retry/deduplication, guest denial, delayed events,
|
||||
queued wakeups, a wakeup promoted after Stop, and cancellation losing to run
|
||||
completion. Server typecheck passes. This factory owns no mutable state.
|
||||
- Reassessment: GitHub ingress/recovery has a coherent durable inbox and replay
|
||||
contract; preserve its shutdown getter and persistence compare-and-set lease.
|
||||
|
||||
## Slice 3: GitHub webhook ingress and recovery
|
||||
|
||||
- Service: 32,928 lines / 1,279,554 bytes → 31,438 lines / 1,225,374 bytes.
|
||||
- `github/webhook-ingress.ts`: 1,579 lines / 58,233 bytes.
|
||||
- All 26 moved declarations and remaining statements preserve syntax after the
|
||||
explicit shutdown getter substitution. The replay barrier remains a live
|
||||
getter. Persistence and credential leases remain the same service instances.
|
||||
- The same 84 existing public-service cases passed before and after, including
|
||||
authentication before retention, disabled-repository filtering, restart,
|
||||
recovery epochs, concurrent scans, ambiguous redelivery, operator retries,
|
||||
canonical-source changes and stale-runtime replay. Server typecheck passes.
|
||||
- Reassessment: Slack task starter confirmation, recovery and failure settlement
|
||||
form one boundary. Keep normal message admission shared and pass its exact
|
||||
existing call contract; do not duplicate provider-visible posts during recovery.
|
||||
|
||||
## Slice 4: Slack task starts
|
||||
|
||||
- Service: 31,438 lines / 1,225,374 bytes → 30,449 lines / 1,190,643 bytes.
|
||||
- `slack/task-start.ts`: 1,075 lines / 39,262 bytes.
|
||||
- Nine moved declarations and all remaining statements preserve syntax. The
|
||||
existing positional message-admission contract is unchanged. The claim barrier
|
||||
remains live through a getter; no worker or mutable per-service state moves.
|
||||
- The same 25 public-service cases passed before and after, covering slash-task
|
||||
admission, provider-confirmed recovery without reposting, reclaimed admission,
|
||||
stale credential ownership, ambiguous sends and explicit operator retries.
|
||||
Server typecheck passes.
|
||||
- Reassessment: Discord command registration and native control admission can
|
||||
share a channel module. Expose only a live lease-key/token lookup; retain the
|
||||
gateway ownership map, renewal, partial acquisition and shutdown in the service.
|
||||
|
||||
## Slice 5: Discord commands
|
||||
|
||||
- Service: 30,449 lines / 1,190,643 bytes → 29,897 lines / 1,170,080 bytes.
|
||||
- `discord/commands.ts`: 639 lines / 25,036 bytes.
|
||||
- Both moved function bodies and all remaining statements preserve syntax after
|
||||
replacing the map read with a live lease-key/token callback. Gateway map,
|
||||
renewal, runtime identity, partial startup and shutdown ownership stay shared.
|
||||
- The same 53 registration and native-command integration cases passed before
|
||||
and after. They cover registration failure, stale/runtime/gateway authority,
|
||||
command receipts, exact target replay and current principal/resource access.
|
||||
Server typecheck passes. Network work remains outside command transactions.
|
||||
|
||||
## Reassessment after batch two
|
||||
|
||||
The service falls from 34,603 lines / 1,337,182 bytes to 29,897 lines /
|
||||
1,170,080 bytes: 4,706 lines and 167,102 bytes removed (13.6% of source lines).
|
||||
The five provider modules total 5,082 lines, plus 38 shared type lines.
|
||||
|
||||
The service function itself spans 27,835 lines. Its 212 nested functions account
|
||||
for 27,146 lines; 51 state/wiring/registration/public API statements account for
|
||||
435 lines. Imports account for 423 lines. This batch reduces provider workflow
|
||||
mixing; it does not resolve the large shared admission functions.
|
||||
|
||||
The largest remaining functions are `processMessage` (2,166 lines),
|
||||
`handleAction` (1,074), `configureWithCredentialLease` (868),
|
||||
`processSelectedPublication` (729), `handleModalSubmit` (708), `handleWebhook`
|
||||
(685), `processLifecycleDelivery` (657), and `applyProviderLifecycleEffect` (593).
|
||||
Future candidates include Teams file-transfer authority/protocol and shared
|
||||
inbound persistence/drain ownership. Preserve the shared transaction and live
|
||||
message contracts before attempting those cuts. Stop this batch at five PRs.
|
||||
|
||||
Cumulative verification: `pnpm -r typecheck` and `pnpm build` passed. Full local
|
||||
test suite results and current-head CI/review are recorded in the final PR.
|
||||
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