mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-28 12:04:41 +01:00
refactor(chat): extract channel status, receipts, drafts and authority workflows (#15995)
## Thinking Path > - Paperclip manages AI agents and their work. > - Chat channels connect that work to provider conversations. > - Provider workflows still fill the shared service. > - Their authority and recovery rules need clear homes. > - This PR lands five reviewed extractions into channel folders. > - Shared authorization, state and worker lifetime stay explicit. ## Linked Issues or Issue Description Refs #15976, #15996, #15999, #16001 and #16002. This root PR now contains all five extractions from batch three. The four upper PRs were merged into their parents. The combined tree matches reviewed stack head `789e918b5af49ec1958076ce4d22d31006205302` exactly. ## What Changed - Extract Slack session status and accepted-file receipt recovery. - Extract Telegram publication drafts and Stop handling. - Extract Photon events and native responses. - Extract Teams file-transfer authority. - Reduce `chat-channels.ts` from 29,897 to 26,461 lines (3,436 lines, 11.5%). - Preserve function bodies, runtime-map identity, leases and transaction boundaries. ## Verification - All 117 selected public-service cases passed before and after their respective extractions. - All 30 moved declarations and remaining service statements preserve normalized syntax across the five slices. - Server typecheck, module boundaries and whitespace checks passed. - Cumulative `pnpm -r typecheck` and `pnpm build` passed. Combined-head CI and review are pending on `5de1b0b99d71049a72440e8c131bf4bb30bdf3f3`. - Fixtures exercise assembled services and disposable databases, not live third-party providers. - Cumulative local `pnpm test:run` on `789e918b5af49ec1958076ce4d22d31006205302` exited 1: 32,414 passed, 125 skipped, one failure. `company-skills-routes.test.ts:1691` (`stars, forks, and comments on skills through company-scoped endpoints`) failed with `socket hang up`; that test and its route are unchanged by this stack. The same 58-case suite then passed in isolation without code changes. The original command did not reach 111 later serialized suites; its failure is retained, so the full-local-test checklist remains unchecked. ```sh pnpm exec vitest run server/src/__tests__/chat-channels.integration.test.ts -t 'Slack session|Slack turns to Idle|signed Slack session stop|denies guest Slack stops' ``` - Combined-head CI attempt 1: five runners received a shutdown signal at 19:13:08 UTC (server shard 10/12, serialized shards 2/9 and 7/9, E2E shards 3/8 and 7/8). These jobs stopped during execution without a failing test assertion. [Original workflow](https://github.com/paperclipai/paperclip/actions/runs/38293566028). The affected jobs will receive one unchanged retry after the other jobs finish. ## Risks - Changes cross durable provider recovery and authority boundaries. The moved bodies preserve syntax, and 117 focused public-service cases passed before and after their extractions. - Shared workers, timers, runtime maps and generic authorization remain in the service. - The full local test run failed in an unchanged company-skills route test; its result and the unrun suites are recorded above. ## 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 - [ ] All Paperclip CI gates are green - [ ] 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
cc88b61c09
commit
0d0e138dad
7 files changed
+3955
-3454
No files matched your search
@@ -0,0 +1,124 @@
|
||||
# Chat channel transport workflows, batch three
|
||||
|
||||
Base: `a2ac5c27157ab033a2e0e2731b1b8e35bf20e7b9`. The preceding stack
|
||||
(#15976, #15979, #15981, #15985, #15987) is merged. The original service is
|
||||
29,897 lines / 1,170,080 bytes. This batch creates five stacked PRs; merging is
|
||||
separate work.
|
||||
|
||||
| Order / boundary | Destination | Approximate lines moved / net reduction | Inputs and outputs | State and effects | Verification |
|
||||
| --- | --- | --- | --- | --- | --- |
|
||||
| 1. Slack session status | `chat-channels/slack/session-status.ts` | 342 / 330 | Endpoint lookup, credential lease, runtime fence, publication run identity, credentials and fetch; return processed count | Claim and settle existing durable status actions. Keep selected revision, attempt ownership and live test barrier. No worker or new state. | Existing public-service status, restart, rate-limit, concurrent selection, stop and Idle cases before/after. |
|
||||
| 2. Slack accepted-file receipts | `chat-channels/slack/file-receipts.ts` | 900 / 840 | Runtime/credential authorization, publication eligibility and endpoint lookup; record/recover accepted upload receipts | Preserve same-attempt leases, exact receipt identities, publication links and transaction boundaries. Keep generic publication transport shared. | Existing upload recovery, competing retry deadline, revoked reach and no-duplicate-upload cases before/after. |
|
||||
| 3. Telegram publication drafts | `chat-channels/telegram/publication-drafts.ts` | 394 / 370 | Current runtime and callback fence, source publication authority; reserve draft or handle stopped generation | Preserve runtime map identity, stop scope, draft revision and transactional invalidation. Keep maintenance and shared publication admission in their existing owners. | Existing durable private draft Stop, stale generation and publication recovery cases before/after. |
|
||||
| 4. Photon incoming events and responses | `chat-channels/imessage-photon/events.ts` | 820 / 780 | Runtime callbacks, principal authorization, question delivery and shared inbound admission | Preserve receiver sequence/lease checks, native interaction binding, transaction ordering and deferred publication. No receiver/startup or credential-inspection ownership move. | Existing Photon public-service integration suite before/after. |
|
||||
| 5. Teams file-transfer authority | `chat-channels/microsoft-teams/file-transfer-authority.ts` | 940 / 900 | Source proof, runtime/credential fence, principal authorization; return current recipient and immutable file authority | Keep source, recipient, Board intent and byte identity proofs together. Preserve map lifetimes and no-wait locks. Transport, maintenance cooldowns and cleanup stay with the publication worker. | Existing Teams current file authority, consent, composed Board transfer and stale-runtime cases before/after. |
|
||||
|
||||
Use named, typed contracts and channel folders. Keep shared message admission,
|
||||
provider startup/shutdown, generic publication transport and shared authorization
|
||||
in the service. Do not introduce a mutable context bag or a runtime import back
|
||||
to the service. Reconsider a candidate if its actual dependency contract proves
|
||||
too broad.
|
||||
|
||||
For each slice, inspect its execution path and callers, run baseline tests, move
|
||||
mechanically, compare normalized syntax, rerun the same public entry-point tests,
|
||||
and check server types and module boundaries. Finish CI and current-head review
|
||||
before opening the next PR. Run broad local gates once at the cumulative handoff.
|
||||
Record failures and skipped groups accurately; a retry does not erase a failed
|
||||
run. Fixtures exercise assembled services and persistence, not live providers.
|
||||
|
||||
The starting service function is 27,835 lines. Its 212 nested functions span
|
||||
27,146 lines; 51 wiring/state/API statements span 435 lines. Imports span 423
|
||||
lines. The largest shared function remains `processMessage` at 2,166 lines.
|
||||
Remap these measurements after the fifth extraction and stop at five PRs.
|
||||
|
||||
## Slice 1: Slack session status
|
||||
|
||||
- Service: 29,897 lines / 1,170,080 bytes → 29,553 lines / 1,156,944 bytes.
|
||||
- Destination: 408 lines / 15,730 bytes. The durable claim, provider status call
|
||||
and conditional settlement remain in one function. No state or worker moves.
|
||||
- All six selected public-service cases passed before and after. They exercise
|
||||
restart/rate-limit recovery, stale selection, Stop and publication-driven Idle.
|
||||
Server typecheck, module boundaries, whitespace and normalized syntax parity
|
||||
passed. The selection hook remains live through an options getter.
|
||||
- Next: accepted-file receipt recovery. Keep the generic outbound publication
|
||||
link writer shared and pass it as a typed callback to the Slack module.
|
||||
|
||||
## Slice 2: Slack accepted-file receipts
|
||||
|
||||
- Service: 29,553 lines / 1,156,944 bytes → 28,620 lines / 1,122,992 bytes.
|
||||
- Destination: 1,000 lines / 36,807 bytes. Thirteen declarations move together:
|
||||
receipt validation, recording, authorization, recovery, settlement and due-work
|
||||
selection. The generic outbound-link writer remains shared.
|
||||
- All nine selected public-service cases passed before and after. They cover
|
||||
accepted upload recovery without reposting, original-attempt ownership,
|
||||
competing retry deadlines, revoked reach and selected file publication.
|
||||
Server typecheck, module boundaries and syntax parity passed. No state or
|
||||
worker is added. Credential lease identity and all transactions are unchanged.
|
||||
- Next: Telegram private draft reservation and stopped-generation callbacks,
|
||||
retaining shared runtime-map identity and source-publication authorization.
|
||||
|
||||
## Slice 3: Telegram publication drafts
|
||||
|
||||
- Service: 28,620 lines / 1,122,992 bytes → 28,219 lines / 1,108,711 bytes.
|
||||
- Destination: 461 lines / 16,860 bytes. Both functions preserve their original
|
||||
bodies. Runtime maps remain the same per-service objects; the source authority
|
||||
callback and credential guard are unchanged.
|
||||
- All 34 durable private-draft Stop cases passed before and after, including
|
||||
restart, bot-global identity allocation, concurrent Stop, transaction failure,
|
||||
uncertain final delivery and retained stale callbacks. Server typecheck,
|
||||
boundaries, whitespace and normalized syntax parity passed.
|
||||
- Reassessment: Photon event handling can move without moving its receiver or
|
||||
setup inspection. For Teams, extract source/recipient/file authority as a
|
||||
coherent transaction-level API; keep transport protocol and scheduler lifetime
|
||||
in the service to avoid passing unrelated runtime/worker dependencies.
|
||||
|
||||
## Slice 4: Photon events
|
||||
|
||||
- Service: 28,219 lines / 1,108,711 bytes → 27,398 lines / 1,079,439 bytes.
|
||||
- Destination: 903 lines / 33,080 bytes. Provider failure, incoming event and
|
||||
native response handling preserve all three function bodies. The receiver,
|
||||
lease ownership, checkpoint callbacks and setup inspection remain in the
|
||||
service; question delivery uses the original service object.
|
||||
- All 18 Photon channel integration cases passed before and after. They use the
|
||||
assembled Chat SDK runtime, provider fixture streams and a disposable database
|
||||
to cover poll/confirmation responses, separate participant drafts, stale
|
||||
controls, receiver lease fencing, groups, echoes, restart and continuation.
|
||||
Server typecheck, boundaries, whitespace and syntax parity passed.
|
||||
- Next: Teams file-transfer authority, with no transport or timer ownership move.
|
||||
|
||||
## Slice 5: Teams file-transfer authority
|
||||
|
||||
- Service: 27,398 lines / 1,079,439 bytes → 26,461 lines / 1,046,711 bytes.
|
||||
- Destination: 1,041 lines / 36,532 bytes. Eleven declarations move into a
|
||||
transaction-level authority factory exposing Board intent staging and fresh
|
||||
authority derivation. It takes no database instance and starts no work.
|
||||
- The original per-service runtime maps and retry-proof callbacks are passed
|
||||
after the retry factory initializes. No-wait row locks, identity advisory
|
||||
locks, source ordering and byte/recipient proof hashes preserve their bodies.
|
||||
- All 50 selected Teams public-service cases passed before and after, including
|
||||
exact recipient admission, retained source continuation, composed Board file
|
||||
transfers, stale runtime callbacks and consent/file receipt recovery. Server
|
||||
typecheck, module boundaries, whitespace and normalized syntax parity passed.
|
||||
|
||||
## Reassessment after batch three
|
||||
|
||||
The service falls from 29,897 lines / 1,170,080 bytes to 26,461 lines /
|
||||
1,046,711 bytes: 3,436 lines (11.5%) and 123,369 bytes removed. The five new
|
||||
modules total 3,813 lines / 139,009 bytes. All 117 selected cases passed before
|
||||
and after their respective extractions.
|
||||
|
||||
The service function spans 24,484 lines. Its 191 nested functions span 23,829
|
||||
lines; 54 state/wiring/API statements span 425 lines. Imports span 412 lines.
|
||||
The largest remaining functions are `processMessage` (2,166 lines),
|
||||
`handleAction` (1,074), `configureWithCredentialLease` (868),
|
||||
`processSelectedPublication` (729), `handleModalSubmit` (708), `handleWebhook`
|
||||
(685), `processLifecycleDelivery` (657), `applyProviderLifecycleEffect` (593),
|
||||
and `ingestAttachments` (574).
|
||||
|
||||
Future candidates include receipt-reaction recovery, provider-effect dispatch,
|
||||
attachment ingestion and inbound delivery persistence/drain ownership. Teams
|
||||
transport protocol and scheduling can be reconsidered after their lifetime
|
||||
contract is explicit; this batch deliberately moved only authority proofs.
|
||||
Do not move a large shared admission function wholesale merely to reduce LOC.
|
||||
Stop this batch at five PRs. Cumulative local gates and current-head CI/review
|
||||
results are recorded in the final PR; per-slice PRs retain their own evidence.
|
||||
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,903 @@
|
||||
import type { EndpointRow, DbTransaction, RuntimeContext, LifecycleRuntimeFence } from "../../chat-channels/runtime-types.js";
|
||||
|
||||
import { notifyChatPublicationWork } from "../../chat-work-notifications.js";
|
||||
|
||||
import { nativeSha256 } from "../../native-runtime/canonical.js";
|
||||
|
||||
import { projectSafeChatPublicationText } from "../../chat-publication-projection.js";
|
||||
import { PhotonAnswerValidationError, nativePhotonInteraction, photonResponseCommand, parsePhotonQuestionAnswer, type PhotonPromptReceipt, type PhotonInteractionBinding, type PhotonDraft } from "../../photon/interactions.js";
|
||||
import { validateNativeQuestionResponseInput } from "../../native-runtime/native-question-bridge.js";
|
||||
import type { AskUserQuestionsAnswer, AskUserQuestionsInteraction, IssueThreadInteraction } from "@paperclipai/shared";
|
||||
import { PhotonError, photonFailure } from "../../photon/cloud.js";
|
||||
import { PhotonChatAdapter, photonThreadId } from "../../photon/adapter.js";
|
||||
|
||||
import type { LiveEvent as PhotonEvent } from "@photon-ai/advanced-imessage";
|
||||
|
||||
import {
|
||||
and,
|
||||
eq,
|
||||
inArray,
|
||||
isNotNull,
|
||||
sql,
|
||||
} from "drizzle-orm";
|
||||
|
||||
import type { Db } from "@paperclipai/db";
|
||||
|
||||
import {
|
||||
chatActions,
|
||||
chatConversations,
|
||||
chatEndpointResources,
|
||||
chatEndpoints,
|
||||
chatExternalPrincipals,
|
||||
chatPublications,
|
||||
issues,
|
||||
toolConnections,
|
||||
} from "@paperclipai/db";
|
||||
|
||||
import {
|
||||
HttpError,
|
||||
forbidden,
|
||||
} from "../../../errors.js";
|
||||
|
||||
import {
|
||||
type ChatSdkMessageCallbackEvent,
|
||||
} from "../../chat-sdk-runtime.js";
|
||||
|
||||
import {
|
||||
logActivity,
|
||||
} from "../../activity-log.js";
|
||||
|
||||
import { issueThreadInteractionService } from "../../issue-thread-interactions.js";
|
||||
import {
|
||||
questionResponseDeliveryService,
|
||||
} from "../../question-response-delivery.js";
|
||||
|
||||
import type {
|
||||
Author,
|
||||
} from "chat";
|
||||
|
||||
import type { EndpointRecord } from "../runtime-types.js";
|
||||
|
||||
type PhotonEventsOptions = {
|
||||
runtimeCallbackRecord: (endpointId: string, context: RuntimeContext, statuses: EndpointRow["status"][]) => Promise<EndpointRecord | null>;
|
||||
runtimeCallbackEndpoint: (tx: DbTransaction, endpointId: string, context: LifecycleRuntimeFence, statuses: EndpointRow["status"][]) => Promise<EndpointRow | null>;
|
||||
invalidateRuntime: (endpointId: string) => Promise<boolean>;
|
||||
ensurePrincipal: (endpoint: EndpointRow, author: Author, raw?: unknown) => Promise<{ principal: Pick<typeof chatExternalPrincipals.$inferSelect, "id" | "kind" | "isBot">; userId: string | null; linkedDenied: boolean }>;
|
||||
requireCurrentExternalActionAuthorization: (tx: DbTransaction, input: { conversationId: string; endpointId: string; expectedUserId: string; principalId: string; runtimeContext: LifecycleRuntimeFence }) => Promise<void>;
|
||||
isExternalActionAuthorizationChange: (error: unknown) => boolean;
|
||||
scheduleMessageProcessing: (task: () => Promise<void>) => void;
|
||||
processPendingPublications: () => Promise<unknown>;
|
||||
questionResponses: Pick<ReturnType<typeof questionResponseDeliveryService>, "deliver">;
|
||||
redactError: (error: unknown) => string;
|
||||
renewDiscordGatewayOwnershipForMessageAdmission: (tx: DbTransaction, endpointId: string, context: RuntimeContext) => Promise<{ kind: "owned" | "lost" }>;
|
||||
handleSdkMessage: (event: ChatSdkMessageCallbackEvent, runtimeContext?: RuntimeContext, messageOptions?: { receiptReactionSupported?: boolean; slackSlashControl?: boolean }) => Promise<void>;
|
||||
};
|
||||
|
||||
export function createPhotonEvents(db: Db, options: PhotonEventsOptions) {
|
||||
const { runtimeCallbackRecord, runtimeCallbackEndpoint, invalidateRuntime, ensurePrincipal, requireCurrentExternalActionAuthorization, isExternalActionAuthorizationChange, scheduleMessageProcessing, processPendingPublications, questionResponses, redactError, renewDiscordGatewayOwnershipForMessageAdmission, handleSdkMessage } = options;
|
||||
|
||||
|
||||
async function handlePhotonFailure(
|
||||
endpointId: string,
|
||||
error: unknown,
|
||||
context: RuntimeContext,
|
||||
) {
|
||||
if (error instanceof PhotonError && error.code === "attachment_not_ready")
|
||||
return;
|
||||
error = photonFailure(error);
|
||||
const fatal =
|
||||
error instanceof PhotonError &&
|
||||
[
|
||||
"credentials",
|
||||
"line_unavailable",
|
||||
"history_gap",
|
||||
"invalid_response",
|
||||
].includes(error.code);
|
||||
const message =
|
||||
error instanceof PhotonError
|
||||
? error.message
|
||||
: "Photon connection interrupted; reconnecting";
|
||||
await db.transaction(async (tx) => {
|
||||
const endpoint = await runtimeCallbackEndpoint(tx, endpointId, context, [
|
||||
"verifying",
|
||||
"active",
|
||||
"attention",
|
||||
]);
|
||||
if (!endpoint) return;
|
||||
await tx
|
||||
.update(chatEndpoints)
|
||||
.set({
|
||||
status: "attention",
|
||||
healthMessage: message,
|
||||
lastError: message,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(chatEndpoints.id, endpointId));
|
||||
if (fatal)
|
||||
await tx
|
||||
.update(toolConnections)
|
||||
.set({
|
||||
enabled: false,
|
||||
healthStatus: "error",
|
||||
healthMessage: message,
|
||||
})
|
||||
.where(eq(toolConnections.id, endpoint.connectionId));
|
||||
});
|
||||
// Retirement is queued outside the receiver callback to avoid self-joining.
|
||||
void invalidateRuntime(endpointId).catch(() => undefined);
|
||||
}
|
||||
|
||||
async function processPhotonResponse(
|
||||
endpoint: EndpointRow,
|
||||
event: PhotonEvent,
|
||||
threadId: string,
|
||||
adapter: PhotonChatAdapter,
|
||||
context: RuntimeContext,
|
||||
): Promise<boolean> {
|
||||
const command =
|
||||
event.type === "message.received"
|
||||
? photonResponseCommand(event.message.content.text ?? "")
|
||||
: null;
|
||||
const replyGuid =
|
||||
event.type === "message.received" ? event.message.replyTargetGuid : null;
|
||||
const prompt =
|
||||
event.type === "poll.changed"
|
||||
? await adapter.state.read<PhotonPromptReceipt>(
|
||||
`poll-message:${event.pollMessageGuid}`,
|
||||
)
|
||||
: replyGuid
|
||||
? ((await adapter.state.read<PhotonPromptReceipt>(
|
||||
`prompt-message:${replyGuid}`,
|
||||
)) ??
|
||||
(await adapter.state.read<PhotonPromptReceipt>(
|
||||
`poll-message:${replyGuid}`,
|
||||
)))
|
||||
: null;
|
||||
const reference = command?.reference ?? prompt?.reference;
|
||||
if (!reference) {
|
||||
if (
|
||||
!event.isFromMe &&
|
||||
(replyGuid ||
|
||||
(event.type === "poll.changed" &&
|
||||
(event.delta.type === "voted" || event.delta.type === "unvoted")))
|
||||
) {
|
||||
// A vote can beat the local poll-binding write. Keep it behind the
|
||||
// checkpoint while a publication in this chat is in flight.
|
||||
const pending = await db
|
||||
.select({ id: chatPublications.id })
|
||||
.from(chatPublications)
|
||||
.innerJoin(
|
||||
chatConversations,
|
||||
eq(chatConversations.id, chatPublications.conversationId),
|
||||
)
|
||||
.where(
|
||||
and(
|
||||
eq(chatPublications.endpointId, endpoint.id),
|
||||
eq(chatConversations.externalThreadId, threadId),
|
||||
inArray(chatPublications.state, [
|
||||
"pending",
|
||||
"retry",
|
||||
"streaming",
|
||||
"delivery_unknown",
|
||||
]),
|
||||
isNotNull(sql`${chatPublications.payload}->>'interactionId'`),
|
||||
),
|
||||
)
|
||||
.limit(1);
|
||||
if (pending.length)
|
||||
throw new PhotonError(
|
||||
"attachment_not_ready",
|
||||
"Waiting for Photon poll binding",
|
||||
);
|
||||
}
|
||||
return (
|
||||
event.type === "poll.changed" ||
|
||||
(event.type === "message.received" &&
|
||||
/^\/(answer|submit)\b/i.test(event.message.content.text ?? ""))
|
||||
);
|
||||
}
|
||||
const action = await db
|
||||
.select()
|
||||
.from(chatActions)
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.companyId, endpoint.companyId),
|
||||
eq(chatActions.endpointId, endpoint.id),
|
||||
eq(chatActions.kind, "photon_interaction"),
|
||||
eq(chatActions.providerActionId, `photon:${reference}`),
|
||||
),
|
||||
)
|
||||
.then((rows) => rows[0]);
|
||||
if (!action || action.status !== "issued" || !action.conversationId)
|
||||
return true;
|
||||
const binding = action.payload as unknown as PhotonInteractionBinding;
|
||||
if (
|
||||
binding.version !== 1 ||
|
||||
binding.reference !== reference ||
|
||||
!Number.isFinite(Date.parse(binding.expiresAt)) ||
|
||||
Date.parse(binding.expiresAt) <= Date.now()
|
||||
)
|
||||
return true;
|
||||
const conversation = await db
|
||||
.select()
|
||||
.from(chatConversations)
|
||||
.where(
|
||||
and(
|
||||
eq(chatConversations.companyId, endpoint.companyId),
|
||||
eq(chatConversations.endpointId, endpoint.id),
|
||||
eq(chatConversations.id, action.conversationId),
|
||||
),
|
||||
)
|
||||
.then((rows) => rows[0]);
|
||||
if (
|
||||
!conversation ||
|
||||
!["active", "waiting"].includes(conversation.state) ||
|
||||
conversation.sessionGeneration !== binding.sessionGeneration ||
|
||||
conversation.externalThreadId !== threadId
|
||||
)
|
||||
return true;
|
||||
const publication = await db
|
||||
.select()
|
||||
.from(chatPublications)
|
||||
.where(
|
||||
and(
|
||||
eq(chatPublications.companyId, endpoint.companyId),
|
||||
eq(chatPublications.endpointId, endpoint.id),
|
||||
eq(chatPublications.id, binding.publicationId),
|
||||
eq(chatPublications.conversationId, conversation.id),
|
||||
),
|
||||
)
|
||||
.then((rows) => rows[0]);
|
||||
if (
|
||||
!publication ||
|
||||
publication.payload.interactionId !== binding.interactionId ||
|
||||
new Date(event.occurredAt).getTime() < action.createdAt.getTime()
|
||||
)
|
||||
return true;
|
||||
if (
|
||||
["pending", "retry", "streaming", "delivery_unknown"].includes(
|
||||
publication.state,
|
||||
)
|
||||
)
|
||||
throw new PhotonError(
|
||||
"attachment_not_ready",
|
||||
"Waiting for Photon prompt publication",
|
||||
);
|
||||
if (publication.state !== "published") return true;
|
||||
const interaction = (
|
||||
await issueThreadInteractionService(db).listForIssue(conversation.issueId)
|
||||
).find((candidate) => candidate.id === binding.interactionId);
|
||||
if (
|
||||
!interaction ||
|
||||
interaction.status !== "pending" ||
|
||||
interaction.companyId !== endpoint.companyId ||
|
||||
interaction.createdByAgentId !== endpoint.assignedAgentId ||
|
||||
!nativePhotonInteraction(interaction)
|
||||
)
|
||||
return true;
|
||||
const actor =
|
||||
event.type === "message.received" ? event.message.sender : event.actor;
|
||||
if (event.isFromMe || !actor?.address || actor.service !== "iMessage")
|
||||
return true;
|
||||
const chat = await adapter.chatInfo(threadId);
|
||||
if (
|
||||
chat.isArchived ||
|
||||
!chat.participants.some(
|
||||
(person) =>
|
||||
person.address === actor.address && person.service === actor.service,
|
||||
)
|
||||
)
|
||||
return true;
|
||||
const principal = await ensurePrincipal(
|
||||
endpoint,
|
||||
{
|
||||
userId: `${actor.service}:${actor.address}`,
|
||||
userName: actor.address,
|
||||
fullName: actor.address,
|
||||
isMe: false,
|
||||
isBot: false,
|
||||
},
|
||||
event,
|
||||
);
|
||||
if (
|
||||
!principal.userId ||
|
||||
principal.linkedDenied ||
|
||||
principal.principal.kind !== "user" ||
|
||||
principal.principal.isBot
|
||||
)
|
||||
return true;
|
||||
const assertCurrent = async () =>
|
||||
db.transaction((tx) =>
|
||||
requireCurrentExternalActionAuthorization(tx, {
|
||||
conversationId: conversation.id,
|
||||
endpointId: endpoint.id,
|
||||
expectedUserId: principal.userId!,
|
||||
principalId: principal.principal.id,
|
||||
runtimeContext: context,
|
||||
}),
|
||||
);
|
||||
try {
|
||||
await assertCurrent();
|
||||
} catch (error) {
|
||||
if (isExternalActionAuthorizationChange(error)) return true;
|
||||
throw error;
|
||||
}
|
||||
const draftKey = `draft:${reference}:${principal.principal.id}:${principal.userId}`;
|
||||
let draft = (await adapter.state.read<PhotonDraft>(draftKey)) ?? {
|
||||
schema: 1,
|
||||
interactionId: interaction.id,
|
||||
principalId: principal.principal.id,
|
||||
userId: principal.userId,
|
||||
answers: [],
|
||||
lastSequence: 0,
|
||||
};
|
||||
if (draft.lastSequence > event.sequence) return true;
|
||||
const questionIndex = command?.questionIndex ?? prompt?.questionIndex ?? 0;
|
||||
let answerValue =
|
||||
command?.value ??
|
||||
(event.type === "message.received"
|
||||
? (event.message.content.text ?? "")
|
||||
: "");
|
||||
let pollChoice: string | undefined;
|
||||
if (event.type === "poll.changed") {
|
||||
if (!prompt || event.pollMessageGuid !== prompt.pollMessageGuid)
|
||||
return true;
|
||||
if (event.delta.type !== "voted" && event.delta.type !== "unvoted")
|
||||
return true;
|
||||
pollChoice = prompt.options[event.delta.optionIdentifier];
|
||||
if (!pollChoice) return true;
|
||||
if (event.delta.type === "unvoted") {
|
||||
const questionId =
|
||||
interaction.kind === "ask_user_questions"
|
||||
? interaction.payload.questions[questionIndex]?.id
|
||||
: null;
|
||||
await adapter.state.update<PhotonDraft>(draftKey, () => ({
|
||||
...draft,
|
||||
answers: draft.answers.filter(
|
||||
(answer) =>
|
||||
answer.questionId !== questionId ||
|
||||
!answer.optionIds.includes(pollChoice!),
|
||||
),
|
||||
decision: draft.decision === pollChoice ? undefined : draft.decision,
|
||||
lastSequence: event.sequence,
|
||||
}));
|
||||
return true;
|
||||
}
|
||||
}
|
||||
const enqueueResponse = async (key: string, text: string) => {
|
||||
await assertCurrent();
|
||||
await db.transaction(async (tx) => {
|
||||
await tx
|
||||
.insert(chatActions)
|
||||
.values({
|
||||
companyId: endpoint.companyId,
|
||||
endpointId: endpoint.id,
|
||||
conversationId: conversation.id,
|
||||
kind: "photon_response_notice",
|
||||
providerActionId: `photon-notice:${key}`,
|
||||
payload: {
|
||||
reference,
|
||||
questionIndex,
|
||||
publicationId: binding.publicationId,
|
||||
},
|
||||
status: "processed",
|
||||
})
|
||||
.onConflictDoNothing();
|
||||
await notifyChatPublicationWork(tx);
|
||||
await tx
|
||||
.insert(chatPublications)
|
||||
.values({
|
||||
companyId: endpoint.companyId,
|
||||
endpointId: endpoint.id,
|
||||
conversationId: conversation.id,
|
||||
issueId: conversation.issueId,
|
||||
idempotencyKey: key,
|
||||
payload: {
|
||||
...publication.payload,
|
||||
card: undefined,
|
||||
text: projectSafeChatPublicationText(text),
|
||||
},
|
||||
state: "pending",
|
||||
})
|
||||
.onConflictDoNothing();
|
||||
});
|
||||
scheduleMessageProcessing(() =>
|
||||
processPendingPublications().then(() => undefined),
|
||||
);
|
||||
};
|
||||
const notice = (text: string) =>
|
||||
enqueueResponse(`photon-response:${reference}:${event.sequence}`, text);
|
||||
try {
|
||||
if (interaction.kind === "ask_user_questions") {
|
||||
if (command?.command !== "submit") {
|
||||
let answer: AskUserQuestionsAnswer;
|
||||
if (pollChoice) {
|
||||
const question = interaction.payload.questions[questionIndex];
|
||||
if (
|
||||
!question ||
|
||||
!question.options.some((option) => option.id === pollChoice)
|
||||
)
|
||||
return true;
|
||||
answer = { questionId: question.id, optionIds: [pollChoice] };
|
||||
} else
|
||||
answer = parsePhotonQuestionAnswer(
|
||||
interaction,
|
||||
questionIndex,
|
||||
answerValue,
|
||||
);
|
||||
draft = {
|
||||
...draft,
|
||||
answers: [
|
||||
...draft.answers.filter(
|
||||
(entry) => entry.questionId !== answer.questionId,
|
||||
),
|
||||
answer,
|
||||
],
|
||||
lastSequence: event.sequence,
|
||||
};
|
||||
}
|
||||
const answers = interaction.payload.questions.flatMap((question) =>
|
||||
draft.answers.filter((answer) => answer.questionId === question.id),
|
||||
);
|
||||
if (
|
||||
interaction.payload.questions.length > 1 &&
|
||||
command?.command !== "submit"
|
||||
) {
|
||||
await adapter.state.update<PhotonDraft>(draftKey, () => draft);
|
||||
const next = interaction.payload.questions.findIndex(
|
||||
(question) =>
|
||||
!answers.some((answer) => answer.questionId === question.id),
|
||||
);
|
||||
if (next >= 0)
|
||||
await enqueueResponse(
|
||||
`photon-question:${reference}:${next}`,
|
||||
"Input needed",
|
||||
);
|
||||
else
|
||||
await notice(
|
||||
`Your answers are saved. Send /submit ${reference} to submit them.`,
|
||||
);
|
||||
return true;
|
||||
}
|
||||
await validateNativeQuestionResponseInput(interaction, { answers });
|
||||
// Canonical service rechecks required fields, options, audience, and
|
||||
// pending -> answered atomically with the response-delivery receipt.
|
||||
const issue = await db
|
||||
.select()
|
||||
.from(issues)
|
||||
.where(
|
||||
and(
|
||||
eq(issues.companyId, endpoint.companyId),
|
||||
eq(issues.id, conversation.issueId),
|
||||
),
|
||||
)
|
||||
.then((rows) => rows[0]);
|
||||
if (!issue) return true;
|
||||
const answered = await issueThreadInteractionService(
|
||||
db,
|
||||
).answerQuestions(
|
||||
issue,
|
||||
interaction.id,
|
||||
{ answers },
|
||||
{ userId: principal.userId },
|
||||
{
|
||||
beforeResolveInTransaction: async (tx) => {
|
||||
await requireCurrentExternalActionAuthorization(tx, {
|
||||
conversationId: conversation.id,
|
||||
endpointId: endpoint.id,
|
||||
expectedUserId: principal.userId!,
|
||||
principalId: principal.principal.id,
|
||||
runtimeContext: context,
|
||||
});
|
||||
const current = await tx
|
||||
.select()
|
||||
.from(chatConversations)
|
||||
.where(eq(chatConversations.id, conversation.id))
|
||||
.then((rows) => rows[0]);
|
||||
if (
|
||||
current?.sessionGeneration !== binding.sessionGeneration ||
|
||||
Date.parse(binding.expiresAt) <= Date.now()
|
||||
)
|
||||
throw forbidden(
|
||||
"Photon prompt expired or its task generation changed",
|
||||
);
|
||||
},
|
||||
afterResolveInTransaction: async (tx, resolved) => {
|
||||
const claimed = await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
principalId: principal.principal.id,
|
||||
status: "processed",
|
||||
result: {
|
||||
code: "photon_question_answered",
|
||||
interactionId: resolved.id,
|
||||
interactionStatus: resolved.status,
|
||||
answersSha256: nativeSha256(
|
||||
(resolved as AskUserQuestionsInteraction).result?.answers,
|
||||
),
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, action.id),
|
||||
eq(chatActions.status, "issued"),
|
||||
),
|
||||
)
|
||||
.returning({ id: chatActions.id });
|
||||
if (!claimed.length)
|
||||
throw forbidden("Photon prompt has already been resolved");
|
||||
await logActivity(tx as unknown as Db, {
|
||||
companyId: endpoint.companyId,
|
||||
actorType: "user",
|
||||
actorId: principal.userId!,
|
||||
action: "issue.thread_interaction_answered",
|
||||
entityType: "issue",
|
||||
entityId: issue.id,
|
||||
details: {
|
||||
source: "external_chat",
|
||||
provider: endpoint.provider,
|
||||
endpointId: endpoint.id,
|
||||
conversationId: conversation.id,
|
||||
interactionId: resolved.id,
|
||||
},
|
||||
});
|
||||
},
|
||||
},
|
||||
);
|
||||
scheduleMessageProcessing(async () => {
|
||||
await questionResponses.deliver(answered.id);
|
||||
await processPendingPublications();
|
||||
});
|
||||
} else {
|
||||
if (command?.command === "submit") return true;
|
||||
const matched = /^(accept|reject|1|2)(?:\s+([\s\S]+))?$/i.exec(
|
||||
answerValue.trim(),
|
||||
);
|
||||
const decision =
|
||||
pollChoice === "accept" || pollChoice === "reject"
|
||||
? pollChoice
|
||||
: matched
|
||||
? /^(accept|1)$/i.test(matched[1])
|
||||
? "accept"
|
||||
: "reject"
|
||||
: draft.decision;
|
||||
if (!decision)
|
||||
throw new PhotonAnswerValidationError(
|
||||
`Send /answer ${reference} Accept or /answer ${reference} Reject <reason>.`,
|
||||
);
|
||||
const reason =
|
||||
matched?.[2]?.trim() ??
|
||||
(draft.decision === "reject" && !pollChoice
|
||||
? answerValue.trim()
|
||||
: "");
|
||||
if (
|
||||
decision === "reject" &&
|
||||
interaction.payload.rejectRequiresReason &&
|
||||
!reason
|
||||
) {
|
||||
await adapter.state.update<PhotonDraft>(draftKey, () => ({
|
||||
...draft,
|
||||
decision,
|
||||
lastSequence: event.sequence,
|
||||
}));
|
||||
await notice(
|
||||
`Give a rejection reason with /answer ${reference} Reject <reason>.`,
|
||||
);
|
||||
return true;
|
||||
}
|
||||
const issue = await db
|
||||
.select()
|
||||
.from(issues)
|
||||
.where(
|
||||
and(
|
||||
eq(issues.companyId, endpoint.companyId),
|
||||
eq(issues.id, conversation.issueId),
|
||||
),
|
||||
)
|
||||
.then((rows) => rows[0]);
|
||||
if (!issue) return true;
|
||||
const mutation = {
|
||||
beforeResolveInTransaction: async (tx: DbTransaction) => {
|
||||
await requireCurrentExternalActionAuthorization(tx, {
|
||||
conversationId: conversation.id,
|
||||
endpointId: endpoint.id,
|
||||
expectedUserId: principal.userId!,
|
||||
principalId: principal.principal.id,
|
||||
runtimeContext: context,
|
||||
});
|
||||
const current = await tx
|
||||
.select()
|
||||
.from(chatConversations)
|
||||
.where(eq(chatConversations.id, conversation.id))
|
||||
.then((rows) => rows[0]);
|
||||
if (
|
||||
current?.sessionGeneration !== binding.sessionGeneration ||
|
||||
Date.parse(binding.expiresAt) <= Date.now()
|
||||
)
|
||||
throw forbidden(
|
||||
"Photon prompt expired or its task generation changed",
|
||||
);
|
||||
},
|
||||
afterResolveInTransaction: async (
|
||||
tx: DbTransaction,
|
||||
resolved: IssueThreadInteraction,
|
||||
) => {
|
||||
const claimed = await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
principalId: principal.principal.id,
|
||||
status: "processed",
|
||||
result: {
|
||||
code: "photon_confirmation_resolved",
|
||||
interactionId: resolved.id,
|
||||
interactionStatus: resolved.status,
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, action.id),
|
||||
eq(chatActions.status, "issued"),
|
||||
),
|
||||
)
|
||||
.returning({ id: chatActions.id });
|
||||
if (!claimed.length)
|
||||
throw forbidden("Photon prompt has already been resolved");
|
||||
await logActivity(tx as unknown as Db, {
|
||||
companyId: endpoint.companyId,
|
||||
actorType: "user",
|
||||
actorId: principal.userId!,
|
||||
action:
|
||||
decision === "accept"
|
||||
? "issue.thread_interaction_accepted"
|
||||
: "issue.thread_interaction_rejected",
|
||||
entityType: "issue",
|
||||
entityId: issue.id,
|
||||
details: {
|
||||
source: "external_chat",
|
||||
provider: endpoint.provider,
|
||||
endpointId: endpoint.id,
|
||||
conversationId: conversation.id,
|
||||
interactionId: resolved.id,
|
||||
},
|
||||
});
|
||||
},
|
||||
};
|
||||
if (decision === "accept")
|
||||
await issueThreadInteractionService(db).acceptInteraction(
|
||||
issue,
|
||||
interaction.id,
|
||||
{},
|
||||
{ userId: principal.userId },
|
||||
mutation,
|
||||
);
|
||||
else
|
||||
await issueThreadInteractionService(db).rejectInteraction(
|
||||
issue,
|
||||
interaction.id,
|
||||
{ reason },
|
||||
{ userId: principal.userId },
|
||||
mutation,
|
||||
);
|
||||
scheduleMessageProcessing(async () => {
|
||||
await processPendingPublications();
|
||||
});
|
||||
}
|
||||
} catch (error) {
|
||||
if (error instanceof PhotonError) throw error;
|
||||
if (isExternalActionAuthorizationChange(error)) return true;
|
||||
if (error instanceof HttpError && [403, 404, 409].includes(error.status))
|
||||
return true;
|
||||
// Only validator errors become correction copy. Infrastructure errors
|
||||
// leave the checkpoint unchanged and retry through durable recovery.
|
||||
if (
|
||||
(error instanceof HttpError && error.status === 422) ||
|
||||
error instanceof PhotonAnswerValidationError
|
||||
) {
|
||||
const missingIndex = interaction.kind === "ask_user_questions" && command?.command === "submit"
|
||||
? interaction.payload.questions.findIndex((question) =>
|
||||
question.required !== false && !draft.answers.some((answer) => answer.questionId === question.id))
|
||||
: -1;
|
||||
await notice(
|
||||
`${redactError(error)} Send /answer ${reference}.${(missingIndex >= 0 ? missingIndex : questionIndex) + 1} <answer> to correct it.`,
|
||||
);
|
||||
return true;
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
async function handlePhotonEvent(
|
||||
endpointId: string,
|
||||
event: PhotonEvent,
|
||||
context: RuntimeContext,
|
||||
) {
|
||||
const record = await runtimeCallbackRecord(endpointId, context, [
|
||||
"verifying",
|
||||
"active",
|
||||
"attention",
|
||||
]);
|
||||
if (!record)
|
||||
throw new Error("Photon receiver no longer owns this endpoint");
|
||||
const adapter = context.endpointRuntime!.getProviderAdapter();
|
||||
if (!(adapter instanceof PhotonChatAdapter))
|
||||
throw new Error("Photon adapter unavailable");
|
||||
if (adapter.authentication.identity.allocation === "shared" && event.type === "group.changed") return;
|
||||
if (
|
||||
event.type === "group.changed" &&
|
||||
(event.change.type === "participantRemoved" ||
|
||||
event.change.type === "participantLeft") &&
|
||||
event.change.participant.address === record.endpoint.botExternalId
|
||||
) {
|
||||
const removedThreadId = photonThreadId({
|
||||
lineId: adapter.authentication.identity.lineId,
|
||||
chatGuid: event.chatGuid,
|
||||
isGroup: true,
|
||||
});
|
||||
await db.transaction(async (tx) => {
|
||||
if (
|
||||
!(await runtimeCallbackEndpoint(tx, endpointId, context, [
|
||||
"verifying",
|
||||
"active",
|
||||
]))
|
||||
)
|
||||
throw new Error("Photon receiver is no longer current");
|
||||
if (
|
||||
(
|
||||
await renewDiscordGatewayOwnershipForMessageAdmission(
|
||||
tx,
|
||||
endpointId,
|
||||
context,
|
||||
)
|
||||
).kind !== "owned"
|
||||
)
|
||||
throw new Error("Photon receiver lease was replaced");
|
||||
// Removal can be the first retained event after activation. Preserve a
|
||||
// tombstone even when the group has never been discovered locally.
|
||||
await tx
|
||||
.insert(chatEndpointResources)
|
||||
.values({
|
||||
companyId: record.endpoint.companyId,
|
||||
endpointId,
|
||||
providerResourceId: removedThreadId,
|
||||
type: "group_chat",
|
||||
label: "Unavailable iMessage group",
|
||||
enabled: false,
|
||||
availability: "unavailable",
|
||||
metadata: { photonRemoved: true },
|
||||
})
|
||||
.onConflictDoUpdate({
|
||||
target: [
|
||||
chatEndpointResources.endpointId,
|
||||
chatEndpointResources.type,
|
||||
chatEndpointResources.providerResourceId,
|
||||
],
|
||||
set: {
|
||||
availability: "unavailable",
|
||||
metadata: sql`jsonb_set(${chatEndpointResources.metadata}, '{photonRemoved}', 'true'::jsonb)`,
|
||||
updatedAt: new Date(),
|
||||
},
|
||||
});
|
||||
});
|
||||
await adapter.endTyping(removedThreadId);
|
||||
return;
|
||||
}
|
||||
const groupIdentity = photonThreadId({ lineId: adapter.authentication.identity.lineId, chatGuid: event.chatGuid, isGroup: true });
|
||||
const removedGroup = await db.select({ metadata: chatEndpointResources.metadata }).from(chatEndpointResources).where(and(eq(chatEndpointResources.companyId, record.endpoint.companyId), eq(chatEndpointResources.endpointId, endpointId), eq(chatEndpointResources.providerResourceId, groupIdentity))).limit(1).then((rows) => rows[0]);
|
||||
const readded = event.type === "group.changed" && event.change.type === "participantAdded" && event.change.participant.address === record.endpoint.botExternalId;
|
||||
// Late events from a removed chat must still advance recovery without a
|
||||
// now-forbidden chat lookup holding every other conversation behind them.
|
||||
if (removedGroup?.metadata.photonRemoved === true && !readded) return;
|
||||
const chat = await adapter.client.chats
|
||||
.get(event.chatGuid)
|
||||
.catch((error) => {
|
||||
throw photonFailure(error);
|
||||
});
|
||||
if (chat.guid !== event.chatGuid || chat.service !== "iMessage" || (chat.isGroup && adapter.authentication.identity.allocation === "shared")) return;
|
||||
const threadId = photonThreadId({
|
||||
lineId: adapter.authentication.identity.lineId,
|
||||
chatGuid: chat.guid,
|
||||
isGroup: chat.isGroup,
|
||||
});
|
||||
const removed =
|
||||
event.type === "group.changed" &&
|
||||
(event.change.type === "participantRemoved" ||
|
||||
event.change.type === "participantLeft") &&
|
||||
event.change.participant.address === record.endpoint.botExternalId;
|
||||
const available = await db.transaction(async (tx) => {
|
||||
if (
|
||||
!(await runtimeCallbackEndpoint(tx, endpointId, context, [
|
||||
"verifying",
|
||||
"active",
|
||||
]))
|
||||
)
|
||||
throw new Error("Photon receiver is no longer current");
|
||||
const ownership = await renewDiscordGatewayOwnershipForMessageAdmission(
|
||||
tx,
|
||||
endpointId,
|
||||
context,
|
||||
);
|
||||
if (ownership.kind !== "owned")
|
||||
throw new Error("Photon receiver lease was replaced");
|
||||
const existing = await tx
|
||||
.select()
|
||||
.from(chatEndpointResources)
|
||||
.where(
|
||||
and(
|
||||
eq(chatEndpointResources.endpointId, endpointId),
|
||||
eq(chatEndpointResources.providerResourceId, threadId),
|
||||
eq(
|
||||
chatEndpointResources.type,
|
||||
chat.isGroup ? "group_chat" : "direct_message",
|
||||
),
|
||||
),
|
||||
)
|
||||
.then((rows) => rows[0]);
|
||||
const added =
|
||||
event.type === "group.changed" &&
|
||||
event.change.type === "participantAdded" &&
|
||||
event.change.participant.address === record.endpoint.botExternalId;
|
||||
const stillRemoved =
|
||||
removed || (existing?.metadata.photonRemoved === true && !added);
|
||||
const availability =
|
||||
stillRemoved || chat.isArchived ? "unavailable" : "available";
|
||||
const metadata = {
|
||||
photonRemoved: stillRemoved,
|
||||
participants: chat.participants
|
||||
.slice(0, 256)
|
||||
.map((p) => ({ address: p.address, service: p.service })),
|
||||
};
|
||||
const label = (
|
||||
chat.displayName || chat.participants.map((p) => p.address).join(", ")
|
||||
).slice(0, 512);
|
||||
await tx
|
||||
.insert(chatEndpointResources)
|
||||
.values({
|
||||
companyId: record.endpoint.companyId,
|
||||
endpointId,
|
||||
providerResourceId: threadId,
|
||||
type: chat.isGroup ? "group_chat" : "direct_message",
|
||||
label,
|
||||
enabled: !chat.isGroup,
|
||||
availability,
|
||||
metadata,
|
||||
})
|
||||
.onConflictDoUpdate({
|
||||
target: [
|
||||
chatEndpointResources.endpointId,
|
||||
chatEndpointResources.type,
|
||||
chatEndpointResources.providerResourceId,
|
||||
],
|
||||
set: { label, availability, metadata, updatedAt: new Date() },
|
||||
});
|
||||
return availability === "available";
|
||||
});
|
||||
if (!available) return;
|
||||
if (
|
||||
!event.isFromMe &&
|
||||
(await processPhotonResponse(
|
||||
record.endpoint,
|
||||
event,
|
||||
threadId,
|
||||
adapter,
|
||||
context,
|
||||
))
|
||||
)
|
||||
return;
|
||||
if (event.type === "message.received" && !event.isFromMe) {
|
||||
await handleSdkMessage(
|
||||
{
|
||||
provider: "imessage-photon",
|
||||
endpointId,
|
||||
thread: context.endpointRuntime!.thread(threadId),
|
||||
message: adapter.normalize(event.message, chat),
|
||||
trigger: chat.isGroup ? "unaddressed_message" : "direct_message",
|
||||
},
|
||||
context,
|
||||
);
|
||||
}
|
||||
}
|
||||
return { handlePhotonFailure, handlePhotonEvent };
|
||||
}
|
||||
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,1000 @@
|
||||
import type { EndpointRow, DbTransaction, DbOrTransaction, LifecycleRuntimeFence } from "../../chat-channels/runtime-types.js";
|
||||
|
||||
import { notifyChatPublicationWork } from "../../chat-work-notifications.js";
|
||||
|
||||
import type { ChatEndpointRuntime } from "../../voice/voice-runtime.js";
|
||||
|
||||
import { chatCredentialMutationLease, type CredentialMutationLeaseGuard } from "../../chat-credential-mutation-lease.js";
|
||||
|
||||
import {
|
||||
randomUUID,
|
||||
} from "node:crypto";
|
||||
import {
|
||||
and,
|
||||
asc,
|
||||
eq,
|
||||
inArray,
|
||||
isNull,
|
||||
lte,
|
||||
notExists,
|
||||
or,
|
||||
sql,
|
||||
} from "drizzle-orm";
|
||||
|
||||
import type { Db } from "@paperclipai/db";
|
||||
|
||||
import {
|
||||
chatActions,
|
||||
chatConversations,
|
||||
chatEndpointResources,
|
||||
chatEndpoints,
|
||||
chatMessageLinks,
|
||||
chatPublications,
|
||||
} from "@paperclipai/db";
|
||||
|
||||
import {
|
||||
isUuidLike,
|
||||
} from "@paperclipai/shared";
|
||||
|
||||
import {
|
||||
type SlackFileUploadAcceptedReceipt,
|
||||
} from "../../chat-sdk-runtime.js";
|
||||
|
||||
import {
|
||||
stageSlackSessionSync,
|
||||
} from "../../chat-slack-session-outbox.js";
|
||||
|
||||
import { classifyChatPublicationError } from "../../chat-publication-errors.js";
|
||||
|
||||
import type { EndpointRecord } from "../runtime-types.js";
|
||||
|
||||
type SlackFileUploadReceiptPayload = {
|
||||
version: 1;
|
||||
publicationId: string;
|
||||
publicationAttempt: number;
|
||||
threadId: string;
|
||||
fileIds: string[];
|
||||
botExternalId: string;
|
||||
runtimeGeneration: number;
|
||||
credentialFingerprint: string;
|
||||
};
|
||||
|
||||
function slackFileUploadReceiptPayload(
|
||||
payload: Record<string, unknown>,
|
||||
): SlackFileUploadReceiptPayload | null {
|
||||
if (
|
||||
payload.version !== 1 ||
|
||||
typeof payload.publicationId !== "string" ||
|
||||
!isUuidLike(payload.publicationId) ||
|
||||
typeof payload.publicationAttempt !== "number" ||
|
||||
!Number.isSafeInteger(payload.publicationAttempt) ||
|
||||
payload.publicationAttempt < 1 ||
|
||||
typeof payload.threadId !== "string" ||
|
||||
payload.threadId.length < 1 ||
|
||||
payload.threadId.length > 2_048 ||
|
||||
!payload.threadId.startsWith("slack:") ||
|
||||
typeof payload.botExternalId !== "string" ||
|
||||
payload.botExternalId.length < 1 ||
|
||||
payload.botExternalId.length > 255 ||
|
||||
typeof payload.runtimeGeneration !== "number" ||
|
||||
!Number.isSafeInteger(payload.runtimeGeneration) ||
|
||||
payload.runtimeGeneration < 0 ||
|
||||
typeof payload.credentialFingerprint !== "string" ||
|
||||
!/^[a-f0-9]{64}$/.test(payload.credentialFingerprint) ||
|
||||
!Array.isArray(payload.fileIds) ||
|
||||
payload.fileIds.length < 1 ||
|
||||
payload.fileIds.length > 20 ||
|
||||
payload.fileIds.some(
|
||||
(fileId) =>
|
||||
typeof fileId !== "string" || !/^F[A-Z0-9]{1,254}$/.test(fileId),
|
||||
) ||
|
||||
new Set(payload.fileIds).size !== payload.fileIds.length
|
||||
) {
|
||||
return null;
|
||||
}
|
||||
return payload as SlackFileUploadReceiptPayload;
|
||||
}
|
||||
|
||||
function operatorConfirmedSlackFileReceipt(
|
||||
result: Record<string, unknown> | null,
|
||||
payload: SlackFileUploadReceiptPayload,
|
||||
): boolean {
|
||||
return (
|
||||
result?.operatorConfirmedPublicationId === payload.publicationId &&
|
||||
result.operatorConfirmedAttempt === payload.publicationAttempt &&
|
||||
typeof result.operatorConfirmedAt === "string" &&
|
||||
Number.isFinite(Date.parse(result.operatorConfirmedAt))
|
||||
);
|
||||
}
|
||||
|
||||
function sameSlackFileUploadReceipt(
|
||||
left: SlackFileUploadReceiptPayload,
|
||||
right: SlackFileUploadReceiptPayload,
|
||||
): boolean {
|
||||
return (
|
||||
left.publicationId === right.publicationId &&
|
||||
left.publicationAttempt === right.publicationAttempt &&
|
||||
left.threadId === right.threadId &&
|
||||
left.botExternalId === right.botExternalId &&
|
||||
left.runtimeGeneration === right.runtimeGeneration &&
|
||||
left.credentialFingerprint === right.credentialFingerprint &&
|
||||
left.fileIds.length === right.fileIds.length &&
|
||||
left.fileIds.every((fileId, index) => fileId === right.fileIds[index])
|
||||
);
|
||||
}
|
||||
const SLACK_FILE_RECEIPT_MAX_AGE_MS = 24 * 60 * 60_000;
|
||||
const SLACK_FILE_RECEIPT_MAX_ATTEMPTS = 12;
|
||||
const SLACK_FILE_RECEIPT_STALE_MS = 60_000;
|
||||
type ConversationRow = typeof chatConversations.$inferSelect;
|
||||
type SlackFileReceiptsOptions = {
|
||||
endpointRecord: (id: string) => Promise<EndpointRecord | null>;
|
||||
withCredentialMutationLease: ReturnType<typeof chatCredentialMutationLease>;
|
||||
runtimeCallbackEndpoint: (tx: DbTransaction, endpointId: string, context: LifecycleRuntimeFence, statuses: EndpointRow["status"][]) => Promise<EndpointRow | null>;
|
||||
nonDirectDestinationAllowed: (endpoint: EndpointRow, resource: typeof chatEndpointResources.$inferSelect | null | undefined) => boolean;
|
||||
runtimeFor: (endpoint: EndpointRow) => Promise<ChatEndpointRuntime>;
|
||||
isExplicitOperatorPublication: (publication: typeof chatPublications.$inferSelect) => boolean;
|
||||
recordExactOutboundPublicationLink: (tx: DbOrTransaction, input: { publication: typeof chatPublications.$inferSelect; providerMessageId: string }) => Promise<boolean>;
|
||||
};
|
||||
|
||||
export function createSlackFileReceipts(db: Db, options: SlackFileReceiptsOptions) {
|
||||
const { endpointRecord, withCredentialMutationLease, runtimeCallbackEndpoint, nonDirectDestinationAllowed, runtimeFor, isExplicitOperatorPublication, recordExactOutboundPublicationLink } = options;
|
||||
|
||||
|
||||
async function recordSlackFileUploadReceipt(input: {
|
||||
credentialLease: CredentialMutationLeaseGuard;
|
||||
conversation: ConversationRow;
|
||||
endpoint: EndpointRow;
|
||||
publication: typeof chatPublications.$inferSelect;
|
||||
receipt: SlackFileUploadAcceptedReceipt;
|
||||
runtimeContext: LifecycleRuntimeFence;
|
||||
}): Promise<string> {
|
||||
const publicationAttempt = input.publication.attempts + 1;
|
||||
const botExternalId = input.endpoint.botExternalId;
|
||||
const expectedThreadId = `slack:${input.receipt.channelId}:${input.receipt.threadTs ?? ""}`;
|
||||
if (
|
||||
input.endpoint.provider !== "slack" ||
|
||||
!botExternalId ||
|
||||
input.receipt.version !== 1 ||
|
||||
expectedThreadId !== input.conversation.externalThreadId ||
|
||||
input.receipt.fileIds.length !==
|
||||
(input.publication.payload.attachmentIds?.length ?? 0)
|
||||
) {
|
||||
throw new Error(
|
||||
"Slack file upload receipt did not match its publication",
|
||||
);
|
||||
}
|
||||
const payload: SlackFileUploadReceiptPayload = {
|
||||
version: 1,
|
||||
publicationId: input.publication.id,
|
||||
publicationAttempt,
|
||||
threadId: expectedThreadId,
|
||||
fileIds: [...input.receipt.fileIds],
|
||||
botExternalId,
|
||||
runtimeGeneration: input.runtimeContext.generation,
|
||||
credentialFingerprint: input.runtimeContext.credentialFingerprint,
|
||||
};
|
||||
if (!slackFileUploadReceiptPayload(payload)) {
|
||||
throw new Error("Slack file upload receipt was malformed");
|
||||
}
|
||||
const providerActionId = `slack-file-receipt:${input.publication.id}:${publicationAttempt}`;
|
||||
const recordedAt = new Date();
|
||||
return db.transaction(async (tx) => {
|
||||
await input.credentialLease.assertOwned(tx);
|
||||
const publication = await tx
|
||||
.select({ id: chatPublications.id })
|
||||
.from(chatPublications)
|
||||
.where(
|
||||
and(
|
||||
eq(chatPublications.id, input.publication.id),
|
||||
eq(chatPublications.companyId, input.publication.companyId),
|
||||
eq(chatPublications.endpointId, input.publication.endpointId),
|
||||
eq(
|
||||
chatPublications.conversationId,
|
||||
input.publication.conversationId,
|
||||
),
|
||||
eq(chatPublications.state, "streaming"),
|
||||
eq(chatPublications.attempts, publicationAttempt),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
const endpoint = publication
|
||||
? await runtimeCallbackEndpoint(
|
||||
tx,
|
||||
input.endpoint.id,
|
||||
input.runtimeContext,
|
||||
["verifying", "active"],
|
||||
)
|
||||
: null;
|
||||
const conversation = endpoint
|
||||
? await tx
|
||||
.select({
|
||||
externalThreadId: chatConversations.externalThreadId,
|
||||
id: chatConversations.id,
|
||||
})
|
||||
.from(chatConversations)
|
||||
.where(
|
||||
and(
|
||||
eq(chatConversations.id, input.conversation.id),
|
||||
eq(chatConversations.companyId, input.publication.companyId),
|
||||
eq(chatConversations.endpointId, input.endpoint.id),
|
||||
inArray(chatConversations.state, [
|
||||
"active",
|
||||
"waiting",
|
||||
"completed",
|
||||
]),
|
||||
),
|
||||
)
|
||||
.for("no key update")
|
||||
.then((rows) => rows[0] ?? null)
|
||||
: null;
|
||||
if (
|
||||
!publication ||
|
||||
endpoint?.provider !== "slack" ||
|
||||
endpoint.botExternalId !== botExternalId ||
|
||||
conversation?.externalThreadId !== expectedThreadId
|
||||
) {
|
||||
throw new Error(
|
||||
"Slack file upload ownership changed before receipt persistence",
|
||||
);
|
||||
}
|
||||
await notifyChatPublicationWork(tx);
|
||||
const inserted = await tx
|
||||
.insert(chatActions)
|
||||
.values({
|
||||
companyId: input.publication.companyId,
|
||||
endpointId: input.publication.endpointId,
|
||||
conversationId: input.publication.conversationId,
|
||||
kind: "slack_file_upload_receipt",
|
||||
providerActionId,
|
||||
payload,
|
||||
status: "received",
|
||||
result: { code: "slack_file_upload_identity_pending", attempts: 0 },
|
||||
createdAt: recordedAt,
|
||||
updatedAt: recordedAt,
|
||||
})
|
||||
.onConflictDoNothing()
|
||||
.returning({ id: chatActions.id });
|
||||
if (inserted[0]) {
|
||||
await input.credentialLease.assertOwned(tx);
|
||||
return inserted[0].id;
|
||||
}
|
||||
const existing = await tx
|
||||
.select({ id: chatActions.id })
|
||||
.from(chatActions)
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.endpointId, input.publication.endpointId),
|
||||
eq(chatActions.providerActionId, providerActionId),
|
||||
eq(chatActions.kind, "slack_file_upload_receipt"),
|
||||
sql`${chatActions.payload} = ${JSON.stringify(payload)}::jsonb`,
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!existing) {
|
||||
throw new Error(
|
||||
"Slack file upload receipt conflicted with durable state",
|
||||
);
|
||||
}
|
||||
await input.credentialLease.assertOwned(tx);
|
||||
return existing.id;
|
||||
});
|
||||
}
|
||||
|
||||
async function stageSlackFilePublicationSuccess(
|
||||
tx: DbTransaction,
|
||||
input: {
|
||||
committedAt: Date;
|
||||
endpoint: EndpointRow;
|
||||
publication: typeof chatPublications.$inferSelect;
|
||||
providerMessageId: string;
|
||||
runtimeContext: LifecycleRuntimeFence;
|
||||
stageSessionSync: boolean;
|
||||
},
|
||||
): Promise<void> {
|
||||
if (
|
||||
!(await recordExactOutboundPublicationLink(tx, {
|
||||
publication: input.publication,
|
||||
providerMessageId: input.providerMessageId,
|
||||
}))
|
||||
) {
|
||||
throw new Error(
|
||||
"Slack file upload message identity conflicts with another publication",
|
||||
);
|
||||
}
|
||||
await tx
|
||||
.update(chatEndpoints)
|
||||
.set({
|
||||
lastPublicationAt: input.committedAt,
|
||||
updatedAt: input.committedAt,
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatEndpoints.id, input.endpoint.id),
|
||||
inArray(chatEndpoints.status, ["verifying", "active"]),
|
||||
sql`coalesce((${chatEndpoints.setup}->>'runtimeGeneration')::integer, 0) = ${input.runtimeContext.generation}`,
|
||||
),
|
||||
);
|
||||
if (input.stageSessionSync) {
|
||||
await stageSlackSessionSync(tx, {
|
||||
companyId: input.publication.companyId,
|
||||
endpointId: input.publication.endpointId,
|
||||
conversationId: input.publication.conversationId,
|
||||
runtimeGeneration: input.runtimeContext.generation,
|
||||
credentialFingerprint: input.runtimeContext.credentialFingerprint,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
async function lockSlackFileReceiptAuthorization(
|
||||
tx: DbTransaction,
|
||||
input: {
|
||||
actionId: string;
|
||||
payload: SlackFileUploadReceiptPayload;
|
||||
},
|
||||
): Promise<
|
||||
| {
|
||||
action: typeof chatActions.$inferSelect;
|
||||
conversation: ConversationRow;
|
||||
endpoint: EndpointRow;
|
||||
mode: "delivery_unknown" | "operator_confirmed";
|
||||
publication: typeof chatPublications.$inferSelect;
|
||||
}
|
||||
| { action: typeof chatActions.$inferSelect; mode: "defer" | "settled" }
|
||||
| null
|
||||
> {
|
||||
const publication = await tx
|
||||
.select()
|
||||
.from(chatPublications)
|
||||
.where(
|
||||
and(
|
||||
eq(chatPublications.id, input.payload.publicationId),
|
||||
eq(chatPublications.attempts, input.payload.publicationAttempt),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
const action = publication
|
||||
? await tx
|
||||
.select()
|
||||
.from(chatActions)
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, input.actionId),
|
||||
eq(chatActions.companyId, publication.companyId),
|
||||
eq(chatActions.endpointId, publication.endpointId),
|
||||
eq(chatActions.conversationId, publication.conversationId),
|
||||
eq(chatActions.kind, "slack_file_upload_receipt"),
|
||||
eq(
|
||||
chatActions.providerActionId,
|
||||
`slack-file-receipt:${publication.id}:${publication.attempts}`,
|
||||
),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null)
|
||||
: null;
|
||||
const currentPayload = action
|
||||
? slackFileUploadReceiptPayload(action.payload)
|
||||
: null;
|
||||
if (
|
||||
!publication ||
|
||||
!action ||
|
||||
!publication.payload.attachmentIds?.length ||
|
||||
publication.payload.card ||
|
||||
publication.payload.interactionId ||
|
||||
publication.idempotencyKey.startsWith("control:") ||
|
||||
!currentPayload ||
|
||||
!sameSlackFileUploadReceipt(currentPayload, input.payload)
|
||||
) {
|
||||
return null;
|
||||
}
|
||||
if (publication.state === "streaming") {
|
||||
return { action, mode: "defer" };
|
||||
}
|
||||
if (publication.state === "published" && publication.providerMessageId) {
|
||||
const exactLink = await tx
|
||||
.select({ id: chatMessageLinks.id })
|
||||
.from(chatMessageLinks)
|
||||
.where(
|
||||
and(
|
||||
eq(chatMessageLinks.companyId, publication.companyId),
|
||||
eq(chatMessageLinks.endpointId, publication.endpointId),
|
||||
eq(chatMessageLinks.conversationId, publication.conversationId),
|
||||
eq(chatMessageLinks.publicationId, publication.id),
|
||||
publication.commentId
|
||||
? eq(chatMessageLinks.commentId, publication.commentId)
|
||||
: isNull(chatMessageLinks.commentId),
|
||||
eq(
|
||||
chatMessageLinks.providerMessageId,
|
||||
publication.providerMessageId,
|
||||
),
|
||||
eq(chatMessageLinks.direction, "outbound"),
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
return exactLink ? { action, mode: "settled" } : null;
|
||||
}
|
||||
const mode =
|
||||
publication.state === "delivery_unknown"
|
||||
? "delivery_unknown"
|
||||
: publication.state === "published" &&
|
||||
!publication.providerMessageId &&
|
||||
operatorConfirmedSlackFileReceipt(action.result, input.payload)
|
||||
? "operator_confirmed"
|
||||
: null;
|
||||
if (!mode) return null;
|
||||
const endpoint = await runtimeCallbackEndpoint(
|
||||
tx,
|
||||
publication.endpointId,
|
||||
{
|
||||
generation: input.payload.runtimeGeneration,
|
||||
credentialFingerprint: input.payload.credentialFingerprint,
|
||||
},
|
||||
["verifying", "active"],
|
||||
);
|
||||
const conversation = endpoint
|
||||
? await tx
|
||||
.select()
|
||||
.from(chatConversations)
|
||||
.where(
|
||||
and(
|
||||
eq(chatConversations.id, publication.conversationId),
|
||||
eq(chatConversations.companyId, publication.companyId),
|
||||
eq(chatConversations.endpointId, publication.endpointId),
|
||||
eq(chatConversations.externalThreadId, input.payload.threadId),
|
||||
inArray(chatConversations.state, [
|
||||
"active",
|
||||
"waiting",
|
||||
"completed",
|
||||
]),
|
||||
),
|
||||
)
|
||||
.for("no key update")
|
||||
.then((rows) => rows[0] ?? null)
|
||||
: null;
|
||||
const resource =
|
||||
endpoint && conversation?.resourceId && !conversation.isDirectMessage
|
||||
? await tx
|
||||
.select()
|
||||
.from(chatEndpointResources)
|
||||
.where(
|
||||
and(
|
||||
eq(chatEndpointResources.id, conversation.resourceId),
|
||||
eq(chatEndpointResources.companyId, endpoint.companyId),
|
||||
eq(chatEndpointResources.endpointId, endpoint.id),
|
||||
),
|
||||
)
|
||||
.for("no key update")
|
||||
.then((rows) => rows[0] ?? null)
|
||||
: null;
|
||||
if (
|
||||
endpoint?.provider !== "slack" ||
|
||||
endpoint.botExternalId !== input.payload.botExternalId ||
|
||||
!conversation ||
|
||||
!(conversation.isDirectMessage
|
||||
? endpoint.allowDirectMessages
|
||||
: nonDirectDestinationAllowed(endpoint, resource))
|
||||
) {
|
||||
return null;
|
||||
}
|
||||
return { action, conversation, endpoint, mode, publication };
|
||||
}
|
||||
|
||||
async function processPendingSlackFileUploadReceipts(limit = 25) {
|
||||
const selectedAt = new Date();
|
||||
const actions = await db
|
||||
.select()
|
||||
.from(chatActions)
|
||||
.where(
|
||||
slackReceiptEligibility(selectedAt),
|
||||
)
|
||||
.orderBy(asc(chatActions.updatedAt), asc(chatActions.id))
|
||||
.limit(limit);
|
||||
let processed = 0;
|
||||
for (const selected of actions) {
|
||||
const payload = slackFileUploadReceiptPayload(selected.payload);
|
||||
if (!payload) {
|
||||
await db.transaction(async (tx) => {
|
||||
await notifyChatPublicationWork(tx);
|
||||
return tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "failed",
|
||||
result: {
|
||||
code: "slack_file_upload_receipt_payload_invalid",
|
||||
retryable: false,
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, selected.id),
|
||||
eq(chatActions.kind, selected.kind),
|
||||
eq(chatActions.status, selected.status),
|
||||
sql`${chatActions.payload} = ${JSON.stringify(selected.payload)}::jsonb`,
|
||||
selected.result === null
|
||||
? isNull(chatActions.result)
|
||||
: sql`${chatActions.result} = ${JSON.stringify(selected.result)}::jsonb`,
|
||||
),
|
||||
);
|
||||
});
|
||||
continue;
|
||||
}
|
||||
const record = await endpointRecord(selected.endpointId);
|
||||
if (!record) {
|
||||
await db.transaction(async (tx) => {
|
||||
await notifyChatPublicationWork(tx);
|
||||
return tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "cancelled",
|
||||
result: { code: "slack_file_upload_receipt_endpoint_removed" },
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, selected.id),
|
||||
eq(chatActions.kind, selected.kind),
|
||||
eq(chatActions.status, selected.status),
|
||||
sql`${chatActions.payload} = ${JSON.stringify(selected.payload)}::jsonb`,
|
||||
selected.result === null
|
||||
? isNull(chatActions.result)
|
||||
: sql`${chatActions.result} = ${JSON.stringify(selected.result)}::jsonb`,
|
||||
),
|
||||
);
|
||||
});
|
||||
continue;
|
||||
}
|
||||
if (["paused", "attention"].includes(record.endpoint.status)) continue;
|
||||
const ownerToken = randomUUID();
|
||||
let attempt = 1;
|
||||
try {
|
||||
await withCredentialMutationLease(record.endpoint, async (lease) => {
|
||||
const claim = await db.transaction(async (tx) => {
|
||||
await lease.assertOwned(tx);
|
||||
const locked = await lockSlackFileReceiptAuthorization(tx, {
|
||||
actionId: selected.id,
|
||||
payload,
|
||||
});
|
||||
const decisionAt = new Date();
|
||||
const priorAttempts =
|
||||
typeof locked?.action.result?.attempts === "number" &&
|
||||
Number.isSafeInteger(locked.action.result.attempts)
|
||||
? locked.action.result.attempts
|
||||
: 0;
|
||||
attempt = priorAttempts + 1;
|
||||
const retryAt =
|
||||
typeof locked?.action.result?.retryAt === "string"
|
||||
? Date.parse(locked.action.result.retryAt)
|
||||
: Number.NaN;
|
||||
const eligible = locked
|
||||
? locked.action.status === "received" ||
|
||||
(locked.action.status === "failed" &&
|
||||
locked.action.result?.retryable === true &&
|
||||
Number.isFinite(retryAt) &&
|
||||
retryAt <= decisionAt.getTime()) ||
|
||||
(locked.action.status === "processing" &&
|
||||
locked.action.updatedAt.getTime() <=
|
||||
decisionAt.getTime() - SLACK_FILE_RECEIPT_STALE_MS)
|
||||
: false;
|
||||
if (locked && !eligible) return null;
|
||||
if (locked?.mode === "defer") {
|
||||
if (locked.action.status === "processing") {
|
||||
await notifyChatPublicationWork(tx);
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "received",
|
||||
result: {
|
||||
code: "slack_file_upload_identity_pending",
|
||||
attempts: priorAttempts,
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(chatActions.id, locked.action.id));
|
||||
}
|
||||
return null;
|
||||
}
|
||||
if (locked?.mode === "settled") {
|
||||
await notifyChatPublicationWork(tx);
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "processed",
|
||||
result: {
|
||||
code: "slack_file_upload_identity_confirmed",
|
||||
attempts: priorAttempts,
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(chatActions.id, locked.action.id));
|
||||
return null;
|
||||
}
|
||||
if (
|
||||
!locked ||
|
||||
decisionAt.getTime() -
|
||||
(locked?.action.createdAt ?? selected.createdAt).getTime() >
|
||||
SLACK_FILE_RECEIPT_MAX_AGE_MS ||
|
||||
attempt > SLACK_FILE_RECEIPT_MAX_ATTEMPTS
|
||||
) {
|
||||
await notifyChatPublicationWork(tx);
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "cancelled",
|
||||
result: {
|
||||
code: locked
|
||||
? "slack_file_upload_receipt_expired"
|
||||
: "slack_file_upload_receipt_authorization_changed",
|
||||
attempts: priorAttempts,
|
||||
},
|
||||
updatedAt: decisionAt,
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, selected.id),
|
||||
inArray(chatActions.status, [
|
||||
"received",
|
||||
"failed",
|
||||
"processing",
|
||||
]),
|
||||
),
|
||||
);
|
||||
return null;
|
||||
}
|
||||
if (
|
||||
!["received", "failed", "processing"].includes(
|
||||
locked.action.status,
|
||||
)
|
||||
) {
|
||||
return null;
|
||||
}
|
||||
await notifyChatPublicationWork(tx);
|
||||
const [claimed] = await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "processing",
|
||||
result: {
|
||||
attempts: attempt,
|
||||
ownerToken,
|
||||
...(locked.mode === "operator_confirmed"
|
||||
? {
|
||||
operatorConfirmedPublicationId:
|
||||
locked.action.result?.operatorConfirmedPublicationId,
|
||||
operatorConfirmedAttempt:
|
||||
locked.action.result?.operatorConfirmedAttempt,
|
||||
operatorConfirmedAt:
|
||||
locked.action.result?.operatorConfirmedAt,
|
||||
}
|
||||
: {}),
|
||||
},
|
||||
updatedAt: decisionAt,
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, locked.action.id),
|
||||
eq(chatActions.status, locked.action.status),
|
||||
),
|
||||
)
|
||||
.returning({ id: chatActions.id });
|
||||
await lease.assertOwned(tx);
|
||||
return claimed ? locked : null;
|
||||
});
|
||||
if (!claim || !("endpoint" in claim)) return;
|
||||
processed += 1;
|
||||
const endpointRuntime = await runtimeFor(claim.endpoint);
|
||||
await lease.assertOwned();
|
||||
const providerMessageId =
|
||||
await endpointRuntime.resolveSlackFileUploadReceipt(
|
||||
payload.threadId,
|
||||
payload.fileIds,
|
||||
);
|
||||
if (!providerMessageId) {
|
||||
await db.transaction(async (tx) => {
|
||||
await lease.assertOwned(tx);
|
||||
await notifyChatPublicationWork(tx);
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "failed",
|
||||
result: sql`coalesce(${chatActions.result}, '{}'::jsonb) || ${JSON.stringify(
|
||||
{
|
||||
attempts: attempt,
|
||||
code: "slack_file_upload_identity_pending",
|
||||
retryable: true,
|
||||
retryAt: new Date(
|
||||
Date.now() + Math.min(60_000, 1_000 * 2 ** attempt),
|
||||
).toISOString(),
|
||||
},
|
||||
)}::jsonb`,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, selected.id),
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`${chatActions.result}->>'ownerToken' = ${ownerToken}`,
|
||||
),
|
||||
);
|
||||
});
|
||||
return;
|
||||
}
|
||||
await db.transaction(async (tx) => {
|
||||
await lease.assertOwned(tx);
|
||||
const locked = await lockSlackFileReceiptAuthorization(tx, {
|
||||
actionId: selected.id,
|
||||
payload,
|
||||
});
|
||||
if (
|
||||
!locked ||
|
||||
locked.mode === "defer" ||
|
||||
locked.mode === "settled" ||
|
||||
!("publication" in locked) ||
|
||||
locked.action.status !== "processing" ||
|
||||
locked.action.result?.ownerToken !== ownerToken
|
||||
) {
|
||||
if (!locked) {
|
||||
await notifyChatPublicationWork(tx);
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "cancelled",
|
||||
result: {
|
||||
attempts: attempt,
|
||||
code: "slack_file_upload_receipt_authorization_changed",
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, selected.id),
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`${chatActions.result}->>'ownerToken' = ${ownerToken}`,
|
||||
),
|
||||
);
|
||||
} else if (locked.mode === "defer") {
|
||||
await notifyChatPublicationWork(tx);
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "received",
|
||||
result: {
|
||||
attempts: attempt,
|
||||
code: "slack_file_upload_identity_pending",
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, selected.id),
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`${chatActions.result}->>'ownerToken' = ${ownerToken}`,
|
||||
),
|
||||
);
|
||||
} else {
|
||||
await notifyChatPublicationWork(tx);
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status:
|
||||
locked.mode === "settled" ? "processed" : "cancelled",
|
||||
result: {
|
||||
attempts: attempt,
|
||||
code:
|
||||
locked.mode === "settled"
|
||||
? "slack_file_upload_identity_confirmed"
|
||||
: "slack_file_upload_receipt_superseded",
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, selected.id),
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`${chatActions.result}->>'ownerToken' = ${ownerToken}`,
|
||||
),
|
||||
);
|
||||
}
|
||||
return;
|
||||
}
|
||||
const committedAt = new Date();
|
||||
if (
|
||||
!(await recordExactOutboundPublicationLink(tx, {
|
||||
publication: locked.publication,
|
||||
providerMessageId,
|
||||
}))
|
||||
) {
|
||||
await notifyChatPublicationWork(tx);
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "failed",
|
||||
result: {
|
||||
attempts: attempt,
|
||||
code: "slack_file_upload_receipt_message_conflict",
|
||||
retryable: false,
|
||||
},
|
||||
updatedAt: committedAt,
|
||||
})
|
||||
.where(eq(chatActions.id, selected.id));
|
||||
return;
|
||||
}
|
||||
if (locked.mode === "delivery_unknown") {
|
||||
await notifyChatPublicationWork(tx);
|
||||
const [settled] = await tx
|
||||
.update(chatPublications)
|
||||
.set({
|
||||
state: "published",
|
||||
providerMessageId,
|
||||
publishedAt: committedAt,
|
||||
nextAttemptAt: null,
|
||||
redactedError: null,
|
||||
updatedAt: committedAt,
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatPublications.id, locked.publication.id),
|
||||
eq(chatPublications.state, "delivery_unknown"),
|
||||
eq(chatPublications.attempts, payload.publicationAttempt),
|
||||
),
|
||||
)
|
||||
.returning({ id: chatPublications.id });
|
||||
if (!settled) {
|
||||
throw new Error(
|
||||
"Slack file upload publication changed before receipt commit",
|
||||
);
|
||||
}
|
||||
await stageSlackFilePublicationSuccess(tx, {
|
||||
committedAt,
|
||||
endpoint: locked.endpoint,
|
||||
publication: locked.publication,
|
||||
providerMessageId,
|
||||
runtimeContext: {
|
||||
generation: payload.runtimeGeneration,
|
||||
credentialFingerprint: payload.credentialFingerprint,
|
||||
},
|
||||
stageSessionSync:
|
||||
!isExplicitOperatorPublication(locked.publication) &&
|
||||
!locked.publication.idempotencyKey.startsWith(
|
||||
"control:status:",
|
||||
),
|
||||
});
|
||||
} else {
|
||||
await notifyChatPublicationWork(tx);
|
||||
const [enriched] = await tx
|
||||
.update(chatPublications)
|
||||
.set({ providerMessageId })
|
||||
.where(
|
||||
and(
|
||||
eq(chatPublications.id, locked.publication.id),
|
||||
eq(chatPublications.state, "published"),
|
||||
eq(chatPublications.attempts, payload.publicationAttempt),
|
||||
isNull(chatPublications.providerMessageId),
|
||||
),
|
||||
)
|
||||
.returning({ id: chatPublications.id });
|
||||
if (!enriched) {
|
||||
throw new Error(
|
||||
"Slack file upload publication changed before identity enrichment",
|
||||
);
|
||||
}
|
||||
}
|
||||
await notifyChatPublicationWork(tx);
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "processed",
|
||||
result: {
|
||||
attempts: attempt,
|
||||
code: "slack_file_upload_identity_confirmed",
|
||||
},
|
||||
updatedAt: committedAt,
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, selected.id),
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`${chatActions.result}->>'ownerToken' = ${ownerToken}`,
|
||||
),
|
||||
);
|
||||
await lease.assertOwned(tx);
|
||||
});
|
||||
});
|
||||
} catch (error) {
|
||||
const disposition = classifyChatPublicationError(error, attempt);
|
||||
const retryable =
|
||||
disposition.kind === "retry" ||
|
||||
disposition.kind === "delivery_unknown";
|
||||
const retryMs =
|
||||
disposition.kind === "retry"
|
||||
? disposition.retryAfterMs
|
||||
: Math.min(60_000, 1_000 * 2 ** Math.min(attempt, 8));
|
||||
await db.transaction(async (tx) => {
|
||||
await notifyChatPublicationWork(tx);
|
||||
return tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: retryable ? "failed" : "cancelled",
|
||||
result: sql`coalesce(${chatActions.result}, '{}'::jsonb) || ${JSON.stringify(
|
||||
{
|
||||
attempts: attempt,
|
||||
code: retryable
|
||||
? "slack_file_upload_receipt_lookup_retry"
|
||||
: "slack_file_upload_receipt_lookup_rejected",
|
||||
retryable,
|
||||
...(retryable
|
||||
? { retryAt: new Date(Date.now() + retryMs).toISOString() }
|
||||
: {}),
|
||||
},
|
||||
)}::jsonb`,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, selected.id),
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`${chatActions.result}->>'ownerToken' = ${ownerToken}`,
|
||||
),
|
||||
);
|
||||
});
|
||||
}
|
||||
}
|
||||
return processed;
|
||||
}
|
||||
|
||||
function slackReceiptEligibility(selectedAt?: Date) {
|
||||
return and(
|
||||
eq(chatActions.kind, "slack_file_upload_receipt"),
|
||||
notExists(
|
||||
db
|
||||
.select({ id: chatEndpoints.id })
|
||||
.from(chatEndpoints)
|
||||
.where(
|
||||
and(
|
||||
eq(chatEndpoints.id, chatActions.endpointId),
|
||||
inArray(chatEndpoints.status, ["paused", "attention"]),
|
||||
),
|
||||
),
|
||||
),
|
||||
// The normal publication owner is still polling Slack. Receipt
|
||||
// recovery must neither contend with it nor let its not-yet-due
|
||||
// action occupy the bounded recovery page.
|
||||
notExists(
|
||||
db
|
||||
.select({ id: chatPublications.id })
|
||||
.from(chatPublications)
|
||||
.where(
|
||||
and(
|
||||
sql`${chatPublications.id}::text = ${chatActions.payload}->>'publicationId'`,
|
||||
sql`${chatPublications.attempts}::text = ${chatActions.payload}->>'publicationAttempt'`,
|
||||
eq(chatPublications.state, "streaming"),
|
||||
),
|
||||
),
|
||||
),
|
||||
or(
|
||||
eq(chatActions.status, "received"),
|
||||
and(
|
||||
eq(chatActions.status, "failed"),
|
||||
sql`coalesce(${chatActions.result}->>'retryable', 'false') = 'true'`,
|
||||
selectedAt ? sql`(${chatActions.result}->>'retryAt')::timestamptz <= ${selectedAt.toISOString()}::timestamptz` : sql`${chatActions.result}->>'retryAt' is not null`,
|
||||
),
|
||||
and(
|
||||
eq(chatActions.status, "processing"),
|
||||
selectedAt ? lte(chatActions.updatedAt, new Date(selectedAt.getTime() - SLACK_FILE_RECEIPT_STALE_MS)) : undefined,
|
||||
),
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
async function nextSlackReceiptAt(): Promise<number | null> {
|
||||
const [row] = await db.select({ at: sql<string | null>`min(case
|
||||
when ${chatActions.status} = 'processing' then ${chatActions.updatedAt} + ${SLACK_FILE_RECEIPT_STALE_MS} * interval '1 millisecond'
|
||||
when ${chatActions.status} = 'failed' then (${chatActions.result}->>'retryAt')::timestamptz
|
||||
else now() end)` }).from(chatActions).where(slackReceiptEligibility());
|
||||
return row?.at == null ? null : new Date(row.at).getTime();
|
||||
}
|
||||
return { recordSlackFileUploadReceipt, stageSlackFilePublicationSuccess, processPendingSlackFileUploadReceipts, nextSlackReceiptAt };
|
||||
}
|
||||
@@ -0,0 +1,408 @@
|
||||
import type { EndpointRow, DbTransaction, DbOrTransaction, LifecycleRuntimeFence } from "../../chat-channels/runtime-types.js";
|
||||
|
||||
import { dueChatAction } from "../../chat-action-work.js";
|
||||
|
||||
import { notifyChatActionWork } from "../../chat-work-notifications.js";
|
||||
|
||||
import { chatCredentialMutationLease } from "../../chat-credential-mutation-lease.js";
|
||||
|
||||
import {
|
||||
randomUUID,
|
||||
} from "node:crypto";
|
||||
import {
|
||||
and,
|
||||
asc,
|
||||
desc,
|
||||
eq,
|
||||
gt,
|
||||
inArray,
|
||||
lte,
|
||||
or,
|
||||
sql,
|
||||
} from "drizzle-orm";
|
||||
|
||||
import type { Db } from "@paperclipai/db";
|
||||
|
||||
import {
|
||||
chatActions,
|
||||
chatConversations,
|
||||
chatEndpointResources,
|
||||
chatEndpoints,
|
||||
chatPublications,
|
||||
heartbeatRuns,
|
||||
} from "@paperclipai/db";
|
||||
|
||||
import { logger } from "../../../middleware/logger.js";
|
||||
|
||||
import {
|
||||
setSlackSessionStatus,
|
||||
slackSessionStatusForPublication,
|
||||
} from "../../chat-slack-sessions.js";
|
||||
import {
|
||||
slackSessionSyncPayload,
|
||||
} from "../../chat-slack-session-outbox.js";
|
||||
|
||||
import { classifyChatPublicationError } from "../../chat-publication-errors.js";
|
||||
|
||||
import type { EndpointRecord } from "../runtime-types.js";
|
||||
|
||||
type SlackSessionStatusOptions = {
|
||||
endpointRecord: (id: string) => Promise<EndpointRecord | null>;
|
||||
withCredentialMutationLease: ReturnType<typeof chatCredentialMutationLease>;
|
||||
runtimeCallbackEndpoint: (tx: DbTransaction, endpointId: string, context: LifecycleRuntimeFence, statuses: EndpointRow["status"][]) => Promise<EndpointRow | null>;
|
||||
nonDirectDestinationAllowed: (endpoint: EndpointRow, resource: typeof chatEndpointResources.$inferSelect | null | undefined) => boolean;
|
||||
runIdFromMilestonePublication: (publication: typeof chatPublications.$inferSelect) => string | null;
|
||||
resolveCredentials: (endpoint: EndpointRow, database?: DbOrTransaction) => Promise<Record<string, string>>;
|
||||
fetchImpl: typeof globalThis.fetch;
|
||||
redactError: (error: unknown) => string;
|
||||
slackSessionSyncSelectionBarrier?: () => Promise<void>;
|
||||
};
|
||||
|
||||
export function createSlackSessionStatus(db: Db, options: SlackSessionStatusOptions) {
|
||||
const { endpointRecord, withCredentialMutationLease, runtimeCallbackEndpoint, nonDirectDestinationAllowed, runIdFromMilestonePublication, resolveCredentials, fetchImpl, redactError } = options;
|
||||
|
||||
|
||||
async function processPendingSlackSessionSyncs(
|
||||
limit = 25,
|
||||
onlyActionId?: string,
|
||||
) {
|
||||
const now = new Date();
|
||||
const actions = await db
|
||||
.select({ action: chatActions })
|
||||
.from(chatActions)
|
||||
.leftJoin(chatEndpoints, and(eq(chatEndpoints.id, chatActions.endpointId), eq(chatEndpoints.companyId, chatActions.companyId)))
|
||||
.where(and(
|
||||
onlyActionId ? eq(chatActions.id, onlyActionId) : undefined,
|
||||
dueChatAction("slack_session_sync", now.getTime()),
|
||||
))
|
||||
.orderBy(asc(chatActions.updatedAt))
|
||||
.limit(limit)
|
||||
.then(rows => rows.map(row => row.action));
|
||||
await options.slackSessionSyncSelectionBarrier?.();
|
||||
for (const action of actions) {
|
||||
const payload = slackSessionSyncPayload(action.payload);
|
||||
if (!payload || !action.conversationId) {
|
||||
await db.transaction(async (tx) => {
|
||||
await notifyChatActionWork(tx, "slack_session_sync");
|
||||
return tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "failed",
|
||||
result: { code: "slack_session_payload_invalid" },
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, action.id),
|
||||
eq(chatActions.updatedAt, action.updatedAt),
|
||||
),
|
||||
);
|
||||
});
|
||||
continue;
|
||||
}
|
||||
const revisionWhere = and(
|
||||
eq(chatActions.id, action.id),
|
||||
sql`(${chatActions.payload}->>'revision')::bigint = ${payload.revision}`,
|
||||
);
|
||||
const selectedSnapshotWhere = and(
|
||||
revisionWhere,
|
||||
eq(chatActions.status, action.status),
|
||||
eq(chatActions.updatedAt, action.updatedAt),
|
||||
);
|
||||
const record = await endpointRecord(action.endpointId);
|
||||
if (record && ["paused", "attention"].includes(record.endpoint.status)) continue;
|
||||
if (!record || record.endpoint.provider !== "slack") {
|
||||
await db.transaction(async (tx) => {
|
||||
await notifyChatActionWork(tx, "slack_session_sync");
|
||||
return tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "cancelled",
|
||||
result: { code: "slack_session_endpoint_unavailable" },
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(selectedSnapshotWhere);
|
||||
});
|
||||
continue;
|
||||
}
|
||||
const attempts =
|
||||
(typeof action.result?.attempts === "number"
|
||||
? action.result.attempts
|
||||
: 0) + 1;
|
||||
const ownerToken = randomUUID();
|
||||
let ownsAttempt = false;
|
||||
try {
|
||||
await withCredentialMutationLease(record.endpoint, async (lease) => {
|
||||
const claim = await db.transaction(async (tx) => {
|
||||
await lease.assertOwned(tx);
|
||||
const endpoint = await runtimeCallbackEndpoint(
|
||||
tx,
|
||||
action.endpointId,
|
||||
{
|
||||
generation: payload.runtimeGeneration,
|
||||
credentialFingerprint: payload.credentialFingerprint,
|
||||
},
|
||||
["active", "verifying"],
|
||||
);
|
||||
const conversation = endpoint
|
||||
? await tx
|
||||
.select()
|
||||
.from(chatConversations)
|
||||
.where(
|
||||
and(
|
||||
eq(chatConversations.id, action.conversationId!),
|
||||
eq(chatConversations.endpointId, endpoint.id),
|
||||
eq(chatConversations.companyId, endpoint.companyId),
|
||||
inArray(chatConversations.state, [
|
||||
"active",
|
||||
"waiting",
|
||||
"completed",
|
||||
]),
|
||||
),
|
||||
)
|
||||
.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, endpoint.companyId),
|
||||
eq(chatEndpointResources.endpointId, endpoint.id),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null)
|
||||
: null;
|
||||
const superseded = conversation
|
||||
? await tx
|
||||
.select({ id: chatConversations.id })
|
||||
.from(chatConversations)
|
||||
.where(
|
||||
and(
|
||||
eq(chatConversations.companyId, conversation.companyId),
|
||||
eq(chatConversations.endpointId, conversation.endpointId),
|
||||
eq(
|
||||
chatConversations.externalThreadId,
|
||||
conversation.externalThreadId,
|
||||
),
|
||||
gt(
|
||||
chatConversations.sessionGeneration,
|
||||
conversation.sessionGeneration,
|
||||
),
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
.then((rows) => rows.length > 0)
|
||||
: false;
|
||||
if (
|
||||
!endpoint ||
|
||||
!conversation ||
|
||||
superseded ||
|
||||
!(conversation.isDirectMessage
|
||||
? endpoint.allowDirectMessages
|
||||
: nonDirectDestinationAllowed(endpoint, resource))
|
||||
) {
|
||||
await notifyChatActionWork(tx, "slack_session_sync");
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "cancelled",
|
||||
result: { code: "slack_session_destination_changed" },
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(selectedSnapshotWhere);
|
||||
return null;
|
||||
}
|
||||
// Recompute at dispatch: a delayed processing retry must not
|
||||
// overwrite a newer final response or a later question card.
|
||||
const publication = await tx
|
||||
.select()
|
||||
.from(chatPublications)
|
||||
.where(
|
||||
and(
|
||||
eq(chatPublications.companyId, endpoint.companyId),
|
||||
eq(chatPublications.endpointId, endpoint.id),
|
||||
eq(chatPublications.conversationId, conversation.id),
|
||||
eq(chatPublications.state, "published"),
|
||||
sql`${chatPublications.idempotencyKey} not like 'explicit%'`,
|
||||
sql`${chatPublications.idempotencyKey} not like 'control:status:%'`,
|
||||
sql`${chatPublications.idempotencyKey} not like 'wake:%'`,
|
||||
),
|
||||
)
|
||||
.orderBy(
|
||||
desc(chatPublications.publishedAt),
|
||||
desc(chatPublications.createdAt),
|
||||
desc(chatPublications.id),
|
||||
)
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!publication) {
|
||||
await notifyChatActionWork(tx, "slack_session_sync");
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "cancelled",
|
||||
result: { code: "slack_session_no_published_state" },
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(selectedSnapshotWhere);
|
||||
return null;
|
||||
}
|
||||
let status = slackSessionStatusForPublication(
|
||||
publication.payload,
|
||||
conversation.state === "completed",
|
||||
);
|
||||
const runId = runIdFromMilestonePublication(publication);
|
||||
if (status === "processing" && runId) {
|
||||
const run = await tx
|
||||
.select({ status: heartbeatRuns.status })
|
||||
.from(heartbeatRuns)
|
||||
.where(
|
||||
and(
|
||||
eq(heartbeatRuns.id, runId),
|
||||
eq(heartbeatRuns.companyId, endpoint.companyId),
|
||||
eq(heartbeatRuns.agentId, endpoint.assignedAgentId),
|
||||
),
|
||||
)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
// Stop may finish before its confirmation publication. Never
|
||||
// resurrect processing from that run's earlier working receipt.
|
||||
if (!run || !["queued", "running"].includes(run.status))
|
||||
status = "active";
|
||||
}
|
||||
await notifyChatActionWork(tx, "slack_session_sync");
|
||||
const claimed = await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "processing",
|
||||
result: { attempts, ownerToken },
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
selectedSnapshotWhere,
|
||||
or(
|
||||
and(
|
||||
eq(chatActions.status, "received"),
|
||||
sql`(${chatActions.result}->>'retryAt' is null or (${chatActions.result}->>'retryAt')::timestamptz <= ${new Date().toISOString()}::timestamptz)`,
|
||||
),
|
||||
and(
|
||||
eq(chatActions.status, "processing"),
|
||||
lte(chatActions.updatedAt, new Date(Date.now() - 60_000)),
|
||||
),
|
||||
),
|
||||
),
|
||||
)
|
||||
.returning({ id: chatActions.id });
|
||||
await lease.assertOwned(tx);
|
||||
return claimed.length ? { endpoint, conversation, status } : null;
|
||||
});
|
||||
if (!claim) return;
|
||||
ownsAttempt = true;
|
||||
const credentials = await resolveCredentials(claim.endpoint);
|
||||
await lease.assertOwned();
|
||||
const outcome = await setSlackSessionStatus({
|
||||
botToken: credentials.botToken,
|
||||
threadId: claim.conversation.externalThreadId,
|
||||
status: claim.status,
|
||||
fetch: fetchImpl,
|
||||
});
|
||||
await db.transaction(async (tx) => {
|
||||
await lease.assertOwned(tx);
|
||||
// Refresh long work before Slack's one-hour processing timeout.
|
||||
// Unsupported sessions settle until a new publication restages
|
||||
// them; completed threads must not create perpetual polling work.
|
||||
const retryMs =
|
||||
outcome === "updated" && claim.status === "processing"
|
||||
? 30 * 60_000
|
||||
: null;
|
||||
await notifyChatActionWork(tx, "slack_session_sync");
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: retryMs === null ? "processed" : "received",
|
||||
result: {
|
||||
attempts,
|
||||
outcome,
|
||||
sessionStatus: claim.status,
|
||||
...(retryMs === null
|
||||
? {}
|
||||
: {
|
||||
retryAt: new Date(Date.now() + retryMs).toISOString(),
|
||||
}),
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
revisionWhere,
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`${chatActions.result}->>'ownerToken' = ${ownerToken}`,
|
||||
),
|
||||
);
|
||||
});
|
||||
});
|
||||
} catch (error) {
|
||||
const disposition = classifyChatPublicationError(error, attempts);
|
||||
// Unlike posting a message, repeating setStatus cannot duplicate any
|
||||
// content. Ambiguous transport/receipt failures remain retryable, but
|
||||
// a definite scope/auth/destination rejection waits for new activity.
|
||||
const retryable =
|
||||
disposition.kind === "retry" ||
|
||||
disposition.kind === "delivery_unknown";
|
||||
const retryMs =
|
||||
disposition.kind === "retry"
|
||||
? disposition.retryAfterMs
|
||||
: Math.min(5 * 60_000, 1_000 * 2 ** Math.min(attempts, 8));
|
||||
await db.transaction(async (tx) => {
|
||||
await notifyChatActionWork(tx, "slack_session_sync");
|
||||
return tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: retryable ? "received" : "failed",
|
||||
result: {
|
||||
attempts,
|
||||
code: retryable
|
||||
? "slack_session_sync_retry"
|
||||
: "slack_session_sync_rejected",
|
||||
...(retryable
|
||||
? { retryAt: new Date(Date.now() + retryMs).toISOString() }
|
||||
: {}),
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
revisionWhere,
|
||||
ownsAttempt
|
||||
? and(
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`${chatActions.result}->>'ownerToken' = ${ownerToken}`,
|
||||
)
|
||||
: and(
|
||||
eq(chatActions.status, "received"),
|
||||
eq(chatActions.updatedAt, action.updatedAt),
|
||||
),
|
||||
),
|
||||
);
|
||||
});
|
||||
logger.warn(
|
||||
{
|
||||
endpointId: action.endpointId,
|
||||
actionId: action.id,
|
||||
retryable,
|
||||
error: redactError(error),
|
||||
},
|
||||
"Slack session status update did not complete; message delivery is unchanged",
|
||||
);
|
||||
}
|
||||
}
|
||||
return actions.length;
|
||||
}
|
||||
return { processPendingSlackSessionSyncs };
|
||||
}
|
||||
@@ -0,0 +1,461 @@
|
||||
import type { EndpointRow, DbTransaction, DbOrTransaction, RuntimeContext, LifecycleRuntimeFence } from "../../chat-channels/runtime-types.js";
|
||||
|
||||
import { notifyChatPublicationWork } from "../../chat-work-notifications.js";
|
||||
|
||||
import { type ChatRuntime } from "../../chat-runtime.js";
|
||||
import type { ChatEndpointRuntime } from "../../voice/voice-runtime.js";
|
||||
|
||||
import { type CredentialMutationLeaseGuard } from "../../chat-credential-mutation-lease.js";
|
||||
|
||||
import {
|
||||
and,
|
||||
eq,
|
||||
sql,
|
||||
} from "drizzle-orm";
|
||||
|
||||
import type { Db } from "@paperclipai/db";
|
||||
|
||||
import {
|
||||
chatActions,
|
||||
chatConversations,
|
||||
chatPublications,
|
||||
toolConnections,
|
||||
} from "@paperclipai/db";
|
||||
import type {
|
||||
ToolCredentialSecretRef,
|
||||
} from "@paperclipai/shared";
|
||||
|
||||
import {
|
||||
TELEGRAM_DRAFT_ACTION_KIND,
|
||||
TELEGRAM_DRAFT_STOPPED_REASON,
|
||||
parseTelegramDraftBinding,
|
||||
telegramDraftTextSha256,
|
||||
telegramGenerationStoppedReceipt,
|
||||
telegramPrivateDraftDestination,
|
||||
type TelegramDraftControl,
|
||||
type TelegramGenerationStoppedProof,
|
||||
} from "../../chat-telegram-draft-stop.js";
|
||||
|
||||
import {
|
||||
type ChatSdkCallbackEvent,
|
||||
type ChatSdkRuntime,
|
||||
} from "../../chat-sdk-runtime.js";
|
||||
|
||||
import {
|
||||
renderPublicationTransportText,
|
||||
} from "../../chat-publication-text-parts.js";
|
||||
|
||||
import type { EndpointRecord } from "../runtime-types.js";
|
||||
|
||||
type ConversationRow = typeof chatConversations.$inferSelect;
|
||||
type TelegramPublicationDraftsOptions = {
|
||||
runtimeCallbackRecord: (endpointId: string, context: RuntimeContext, statuses: EndpointRow["status"][]) => Promise<EndpointRecord | null>;
|
||||
runtimeCallbackEndpoint: (tx: DbTransaction, endpointId: string, context: LifecycleRuntimeFence, statuses: EndpointRow["status"][]) => Promise<EndpointRow | null>;
|
||||
runtimeFor: (endpoint: EndpointRow) => Promise<ChatEndpointRuntime>;
|
||||
runtime: Pick<ChatSdkRuntime | ChatRuntime, "get">;
|
||||
runtimeContexts: WeakMap<object, RuntimeContext>;
|
||||
credentialFingerprint: (refs: ToolCredentialSecretRef[]) => string;
|
||||
authorizeRetainedChatSourcePublication: (tx: DbOrTransaction, publication: typeof chatPublications.$inferSelect) => Promise<boolean>;
|
||||
};
|
||||
|
||||
export function createTelegramPublicationDrafts(db: Db, options: TelegramPublicationDraftsOptions) {
|
||||
const { runtimeCallbackRecord, runtimeCallbackEndpoint, runtimeFor, runtime, runtimeContexts, credentialFingerprint, authorizeRetainedChatSourcePublication } = options;
|
||||
|
||||
|
||||
async function handleTelegramGenerationStopped(
|
||||
callback: ChatSdkCallbackEvent<TelegramGenerationStoppedProof>,
|
||||
context: RuntimeContext,
|
||||
): Promise<void> {
|
||||
const receipt = telegramGenerationStoppedReceipt(callback.event);
|
||||
if (callback.provider !== "telegram" || !receipt) return;
|
||||
const record = await runtimeCallbackRecord(callback.endpointId, context, [
|
||||
"verifying",
|
||||
"active",
|
||||
]);
|
||||
if (
|
||||
record?.endpoint.provider !== "telegram" ||
|
||||
record.endpoint.botExternalId !== receipt.botUserId
|
||||
)
|
||||
return;
|
||||
const candidates = await db
|
||||
.select()
|
||||
.from(chatActions)
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.companyId, record.endpoint.companyId),
|
||||
eq(chatActions.endpointId, callback.endpointId),
|
||||
eq(chatActions.kind, TELEGRAM_DRAFT_ACTION_KIND),
|
||||
sql`${chatActions.payload}->>'botUserId' = ${receipt.botUserId}`,
|
||||
sql`${chatActions.payload}->>'draftId' = ${String(receipt.draftId)}`,
|
||||
),
|
||||
)
|
||||
.limit(2);
|
||||
if (candidates.length !== 1) return;
|
||||
const candidate = candidates[0]!;
|
||||
const binding = parseTelegramDraftBinding(candidate.payload);
|
||||
if (
|
||||
!binding ||
|
||||
binding.chatId !== receipt.chatId ||
|
||||
binding.messageThreadId !== receipt.messageThreadId ||
|
||||
binding.runtimeGeneration !== context.generation ||
|
||||
binding.credentialFingerprint !== context.credentialFingerprint
|
||||
)
|
||||
return;
|
||||
await db.transaction(async (tx) => {
|
||||
// The sender holds a renewable credential lease during draft HTTP. Do
|
||||
// not acquire that lease here: a Stop can arrive inside the first RPC.
|
||||
// Both Stop and final-send use the publication→endpoint→draft lock order.
|
||||
const publication = await tx
|
||||
.select()
|
||||
.from(chatPublications)
|
||||
.where(
|
||||
and(
|
||||
eq(chatPublications.companyId, record.endpoint.companyId),
|
||||
eq(chatPublications.endpointId, callback.endpointId),
|
||||
eq(chatPublications.conversationId, binding.conversationId),
|
||||
eq(chatPublications.id, binding.publicationId),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
const endpoint = await runtimeCallbackEndpoint(
|
||||
tx,
|
||||
callback.endpointId,
|
||||
context,
|
||||
["verifying", "active"],
|
||||
);
|
||||
if (
|
||||
!publication ||
|
||||
!endpoint ||
|
||||
endpoint.provider !== "telegram" ||
|
||||
endpoint.botExternalId !== receipt.botUserId ||
|
||||
runtime.get(endpoint.id) !== context.endpointRuntime ||
|
||||
publication.attempts !== binding.publicationAttempt ||
|
||||
telegramDraftTextSha256(
|
||||
renderPublicationTransportText(publication.payload),
|
||||
) !== binding.textSha256
|
||||
)
|
||||
return;
|
||||
const connection = await tx
|
||||
.select()
|
||||
.from(toolConnections)
|
||||
.where(
|
||||
and(
|
||||
eq(toolConnections.companyId, endpoint.companyId),
|
||||
eq(toolConnections.id, endpoint.connectionId),
|
||||
),
|
||||
)
|
||||
.for("share")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (
|
||||
!connection?.enabled ||
|
||||
connection.status !== "active" ||
|
||||
credentialFingerprint(connection.credentialSecretRefs) !==
|
||||
binding.credentialFingerprint
|
||||
)
|
||||
return;
|
||||
const conversation = await tx
|
||||
.select()
|
||||
.from(chatConversations)
|
||||
.where(
|
||||
and(
|
||||
eq(chatConversations.companyId, endpoint.companyId),
|
||||
eq(chatConversations.endpointId, endpoint.id),
|
||||
eq(chatConversations.id, binding.conversationId),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (
|
||||
!conversation ||
|
||||
!conversation.isDirectMessage ||
|
||||
conversation.sessionGeneration !== binding.sessionGeneration ||
|
||||
conversation.externalThreadId !==
|
||||
`telegram:${binding.chatId}${binding.messageThreadId === null ? "" : `:${binding.messageThreadId}`}`
|
||||
)
|
||||
return;
|
||||
const action = await tx
|
||||
.select()
|
||||
.from(chatActions)
|
||||
.where(eq(chatActions.id, candidate.id))
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (
|
||||
!action ||
|
||||
JSON.stringify(action.payload) !== JSON.stringify(candidate.payload)
|
||||
)
|
||||
return;
|
||||
// final_sending is an irreversible local I/O claim, not a delivery
|
||||
// receipt. A late Stop must not claim that an in-flight final was undone.
|
||||
if (
|
||||
action.result?.phase !== "drafting" ||
|
||||
action.status !== "processing" ||
|
||||
!["streaming", "delivery_unknown"].includes(publication.state)
|
||||
)
|
||||
return;
|
||||
const stoppedAt = new Date();
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "cancelled",
|
||||
result: {
|
||||
phase: "stopped",
|
||||
updateId: receipt.updateId,
|
||||
stoppedAt: stoppedAt.toISOString(),
|
||||
},
|
||||
updatedAt: stoppedAt,
|
||||
})
|
||||
.where(eq(chatActions.id, action.id));
|
||||
await notifyChatPublicationWork(tx);
|
||||
await tx
|
||||
.update(chatPublications)
|
||||
.set({
|
||||
state: "cancelled",
|
||||
nextAttemptAt: null,
|
||||
redactedError: TELEGRAM_DRAFT_STOPPED_REASON,
|
||||
updatedAt: stoppedAt,
|
||||
})
|
||||
.where(eq(chatPublications.id, publication.id));
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "cancelled",
|
||||
result: {
|
||||
code: "telegram_draft_presentation_stopped",
|
||||
attempts: publication.attempts,
|
||||
},
|
||||
updatedAt: stoppedAt,
|
||||
})
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.companyId, publication.companyId),
|
||||
eq(chatActions.endpointId, publication.endpointId),
|
||||
eq(
|
||||
chatActions.providerActionId,
|
||||
`task-control-authorization:${publication.id}`,
|
||||
),
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`${chatActions.result}->>'attempts' = ${String(publication.attempts)}`,
|
||||
),
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
async function reserveTelegramPublicationDraft(input: {
|
||||
endpoint: EndpointRow;
|
||||
conversation: ConversationRow;
|
||||
publication: typeof chatPublications.$inferSelect;
|
||||
text: string;
|
||||
credentialLease: CredentialMutationLeaseGuard;
|
||||
runtimeContext: RuntimeContext;
|
||||
}): Promise<{ actionId: string; control: TelegramDraftControl }> {
|
||||
const endpointRuntime = await runtimeFor(input.endpoint);
|
||||
const registeredContext = runtimeContexts.get(endpointRuntime as object);
|
||||
if (
|
||||
!registeredContext ||
|
||||
registeredContext.generation !== input.runtimeContext.generation ||
|
||||
registeredContext.credentialFingerprint !==
|
||||
input.runtimeContext.credentialFingerprint ||
|
||||
registeredContext.version !== input.runtimeContext.version
|
||||
)
|
||||
throw new Error("Telegram draft runtime authority changed");
|
||||
const draftRuntimeContext = { ...input.runtimeContext, endpointRuntime };
|
||||
const destination = telegramPrivateDraftDestination(
|
||||
input.conversation.externalThreadId,
|
||||
);
|
||||
if (
|
||||
!destination ||
|
||||
!input.conversation.isDirectMessage ||
|
||||
!input.endpoint.botExternalId
|
||||
)
|
||||
throw new Error("Telegram private draft destination is invalid");
|
||||
const bindingBase = {
|
||||
version: 1 as const,
|
||||
publicationId: input.publication.id,
|
||||
publicationAttempt: input.publication.attempts + 1,
|
||||
conversationId: input.conversation.id,
|
||||
sessionGeneration: input.conversation.sessionGeneration,
|
||||
runtimeGeneration: input.runtimeContext.generation,
|
||||
credentialFingerprint: input.runtimeContext.credentialFingerprint,
|
||||
botUserId: input.endpoint.botExternalId,
|
||||
...destination,
|
||||
textSha256: telegramDraftTextSha256(input.text),
|
||||
};
|
||||
const reserved = await db.transaction(async (tx) => {
|
||||
await input.credentialLease.assertOwned(tx);
|
||||
const publication = await tx
|
||||
.select()
|
||||
.from(chatPublications)
|
||||
.where(
|
||||
and(
|
||||
eq(chatPublications.id, input.publication.id),
|
||||
eq(chatPublications.state, "streaming"),
|
||||
eq(chatPublications.attempts, bindingBase.publicationAttempt),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!publication)
|
||||
throw new Error("Telegram draft publication ownership changed");
|
||||
const providerActionId = `telegram-draft-publication:${publication.id}:${publication.attempts}`;
|
||||
const existing = await tx
|
||||
.select({ id: chatActions.id })
|
||||
.from(chatActions)
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.endpointId, input.endpoint.id),
|
||||
eq(chatActions.providerActionId, providerActionId),
|
||||
),
|
||||
)
|
||||
.limit(1);
|
||||
// A second invocation of the same attempt cannot create another draft or
|
||||
// replay an uncertain final. A new explicit publication attempt gets a
|
||||
// fresh, never-before-used bot-global ID.
|
||||
if (existing.length)
|
||||
throw new Error("Telegram draft attempt already exists");
|
||||
// NO CYCLE and no company/endpoint ownership: rollback, endpoint deletion
|
||||
// and bot re-binding cannot recycle a provider-visible draft identity.
|
||||
// Exhaustion is a refusal, never a fallback to a process-local/random ID.
|
||||
const allocated = await tx.execute(
|
||||
sql`select nextval('chat_telegram_draft_ids')::integer as "draftId"`,
|
||||
);
|
||||
const binding = parseTelegramDraftBinding({
|
||||
...bindingBase,
|
||||
draftId: allocated[0]?.draftId,
|
||||
});
|
||||
if (!binding)
|
||||
throw new Error("Telegram draft identity allocation was invalid");
|
||||
const [action] = await tx
|
||||
.insert(chatActions)
|
||||
.values({
|
||||
companyId: input.endpoint.companyId,
|
||||
endpointId: input.endpoint.id,
|
||||
conversationId: input.conversation.id,
|
||||
kind: TELEGRAM_DRAFT_ACTION_KIND,
|
||||
providerActionId,
|
||||
payload: binding,
|
||||
status: "issued",
|
||||
result: { phase: "reserved" },
|
||||
})
|
||||
.returning();
|
||||
return { action: action!, binding };
|
||||
});
|
||||
const before = (final: boolean): Promise<boolean> =>
|
||||
db.transaction(async (tx) => {
|
||||
await input.credentialLease.assertOwned(tx);
|
||||
const publication = await tx
|
||||
.select()
|
||||
.from(chatPublications)
|
||||
.where(
|
||||
and(
|
||||
eq(chatPublications.companyId, input.endpoint.companyId),
|
||||
eq(chatPublications.id, input.publication.id),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
const endpoint = await runtimeCallbackEndpoint(
|
||||
tx,
|
||||
input.endpoint.id,
|
||||
draftRuntimeContext,
|
||||
["verifying", "active"],
|
||||
);
|
||||
if (
|
||||
!endpoint ||
|
||||
endpoint.provider !== "telegram" ||
|
||||
!endpoint.allowDirectMessages ||
|
||||
endpoint.botExternalId !== reserved.binding.botUserId ||
|
||||
runtime.get(endpoint.id) !== endpointRuntime
|
||||
)
|
||||
throw new Error("Telegram draft endpoint authority changed");
|
||||
const connection = await tx
|
||||
.select()
|
||||
.from(toolConnections)
|
||||
.where(
|
||||
and(
|
||||
eq(toolConnections.companyId, endpoint.companyId),
|
||||
eq(toolConnections.id, endpoint.connectionId),
|
||||
),
|
||||
)
|
||||
.for("share")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (
|
||||
!connection?.enabled ||
|
||||
connection.status !== "active" ||
|
||||
credentialFingerprint(connection.credentialSecretRefs) !==
|
||||
reserved.binding.credentialFingerprint
|
||||
)
|
||||
throw new Error("Telegram draft credential authority changed");
|
||||
const conversation = await tx
|
||||
.select()
|
||||
.from(chatConversations)
|
||||
.where(
|
||||
and(
|
||||
eq(chatConversations.companyId, endpoint.companyId),
|
||||
eq(chatConversations.endpointId, endpoint.id),
|
||||
eq(chatConversations.id, input.conversation.id),
|
||||
),
|
||||
)
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
const action = await tx
|
||||
.select()
|
||||
.from(chatActions)
|
||||
.where(eq(chatActions.id, reserved.action.id))
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (
|
||||
!action ||
|
||||
JSON.stringify(action.payload) !==
|
||||
JSON.stringify(reserved.action.payload) ||
|
||||
!publication ||
|
||||
publication.attempts !== reserved.binding.publicationAttempt
|
||||
)
|
||||
throw new Error("Telegram draft attempt authority changed");
|
||||
if (
|
||||
action.status === "cancelled" &&
|
||||
action.result?.phase === "stopped" &&
|
||||
publication.state === "cancelled"
|
||||
)
|
||||
return false;
|
||||
if (
|
||||
publication.state !== "streaming" ||
|
||||
!conversation ||
|
||||
telegramDraftTextSha256(
|
||||
renderPublicationTransportText(publication.payload),
|
||||
) !== reserved.binding.textSha256 ||
|
||||
conversation.sessionGeneration !==
|
||||
reserved.binding.sessionGeneration ||
|
||||
conversation.externalThreadId !==
|
||||
input.conversation.externalThreadId ||
|
||||
!["active", "waiting", "completed"].includes(conversation.state) ||
|
||||
!["reserved", "drafting"].includes(String(action.result?.phase)) ||
|
||||
!["issued", "processing"].includes(action.status) ||
|
||||
!(await authorizeRetainedChatSourcePublication(tx, publication))
|
||||
)
|
||||
throw new Error("Telegram draft source authority changed");
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({
|
||||
status: "processing",
|
||||
result: {
|
||||
phase: final ? "final_sending" : "drafting",
|
||||
},
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(chatActions.id, action.id));
|
||||
await input.credentialLease.assertOwned(tx);
|
||||
return true;
|
||||
});
|
||||
return {
|
||||
actionId: reserved.action.id,
|
||||
control: {
|
||||
version: 1,
|
||||
draftId: reserved.binding.draftId,
|
||||
beforeDraft: () => before(false),
|
||||
beforeFinal: () => before(true),
|
||||
},
|
||||
};
|
||||
}
|
||||
return { handleTelegramGenerationStopped, reserveTelegramPublicationDraft };
|
||||
}
|
||||
Reference in new issue
Block a user