diff --git a/doc/plans/2026-10-10-chat-channels-refactor.md b/doc/plans/2026-10-10-chat-channels-refactor.md new file mode 100644 index 00000000000..c1db38d8abf --- /dev/null +++ b/doc/plans/2026-10-10-chat-channels-refactor.md @@ -0,0 +1,103 @@ +# Chat channel service extraction sequence + +## Scope and finish line + +Apply the refactor-large-file workflow to `server/src/services/chat-channels.ts`. +Map five responsibilities, then extract and verify one at a time. This slice +extracts provider message normalization without changing provider behavior, +public service exports, database contracts, authentication, or worker ownership. +Subsequent slices require reassessment against the preceding result. No merge or +deployment is part of this slice. + +## Baseline + +- Base: `1100682261` on a clean, detached task worktree. +- File: 39,132 lines / 1,502,701 bytes; imports occupy 421 lines. +- `chatChannelService`: 36,152 lines, including nested functions and wiring. + This is not a measurement of orchestration alone. The first slice removes + top-level normalization; it does not simplify that closure yet. +- Largest nested functions: `processMessage` (2,164 lines), `handleAction` + (1,074), `applyProviderLifecycleEffect` (895), `configureWithCredentialLease` + (868), `processSelectedPublication` (728), `handleModalSubmit` (708), + `handleWebhook` (684), `failedChatRetrySource` (677), and + `processLifecycleDelivery` (657). +- Existing siblings already own SDK runtimes/state, provider inventory reads, + resource lifecycle parsing, GitHub app management, registration, publication + projection/streaming, Teams transfers, Telegram media, and interaction forms. + Extend those boundaries where appropriate instead of duplicating them. +- Callers include application assembly, chat routes, issue/agent routes, and + existing unit/integration tests. Keep their imports and service API stable. + +## Ranked extraction candidates + +Estimates are source lines moved / net reduction after wiring. Later estimates +are provisional: explicit dependency contracts must precede each extraction. +New paths below are relative to `server/src/services/`. + +| Rank / responsibility | Destination | Approx. move / net | Inputs and outputs | State and effects owned | Risk and verification | +| --- | --- | --- | --- | --- | --- | +| 1. Provider message normalization | `chat-channels/provider-messages.ts` | 540 / 520 | Provider payload, author, Request, Teams parser, or retained delivery → actor, lifecycle event, message ID, revision digest, timestamp | No mutable service state or DB writes; consumes the supplied GitHub Request body | Preserve exact revision hashes, limits, native IDs, malformed-input handling, and raw clock provenance. Existing public service integration tests cover all four providers. | +| 2. Resource inventory reconciliation and availability | `chat-channels/resource-reconciliation.ts`, alongside `chat-provider-inventory.ts` and `chat-provider-lifecycle.ts` | 1,300–1,500 / 1,200–1,400 | Endpoint, inventory/lifecycle effect, credential lease, DB and explicit runtime callbacks → reconciled resources and availability | Company-scoped resource/conversation transactions; GitHub stable repository identity and provider-order checks | Keep advisory locks, lease checks and the complete availability transaction together. Verify repository rename, removal/reinstall, resource auditing, Telegram migration and stale callbacks. | +| 3. Identity linking | `chat-channels/identity-links.ts` | 430–600 / 390–550 | DB, endpoint lookup, live public URL getter, principal, token, authenticated user → link preview/confirmation/revocation | Token hashes/expiry, membership checks, join requests and transactional audit records | No cached authority or URL snapshot. Preserve endpoint/company binding, replay behavior and revoked/conflicting links; run identity route and DB integration tests. | +| 4. Publication replacement and supersession | `chat-channels/publication-replacement.ts` (the existing reconciliation module owns wake signals, a separate responsibility) | 1,000–1,150 / 950–1,100 | DB, publication, current endpoint/conversation/run → replacement candidate or suppression reason | Scoped reads and existing issue/run authorization snapshots; no transport or queue ownership | Preserve exact outbound link ownership, interaction precedence, terminal failure replacement, and issue→run lock order. Run publication reconciliation, progress and integration fixtures. | +| 5. Failed chat run retry coordination | `chat-channels/failed-run-retry.ts` | 1,600–1,900 / 1,400–1,700 | DB, source/action IDs, explicit admission/heartbeat/publication hooks → authorized retry, claim and completion | Per-service in-flight retry promises and authority registration/cleanup, persisted source evidence | Higher risk: preserve single retry ownership, no replay of provider calls, live source revalidation and shutdown ordering. Run reconstructed-service, edited-source, revocation and concurrent retry integration fixtures. | + +Start with candidate 1: it establishes a domain folder and removes an independent +responsibility without introducing a mutable context object. It is deliberately +smaller than the later stateful slices. Resource lifecycle effects remain in the +existing parser; message edits/deletes are a different normalization contract. + +Do not move runtime leases, webhook authentication, conversation admission, +delivery transactions, transport attempts, or shutdown piecemeal. The service +must continue to show who admits work and who releases resources. Candidate 2 +needs a narrow callback contract; candidate 5 must preserve the same service +lifetime for registration and in-flight promises. Candidates are sequential, +not independent concurrent changes to this file. + +## Verification and progress + +Before the production move, run existing service integration cases for message +lifecycle, file revision identity, provider timestamps, Telegram zero-message +handling and rich content. These use disposable PostgreSQL and fixture provider +transports, not live third-party execution. Keep those tests at their existing +public entry points. Add characterization only if inspection exposes a gap. + +After the move, repeat that selection, compare extracted declarations and the +remaining file mechanically against the base, then run the relevant adjacent +suites and repository typecheck, full test and build gates. Record blockers and +unverified checks explicitly; do not treat skipped integration tests as passes. + +- Dependencies: installed with the repository-pinned pnpm 9.15.4. The host's + newer pnpm ignored legacy patch settings, so verification uses the pinned tool. +- Original-base integration selection: 48 passed before and 48 passed after + extraction. Existing tests cover the boundary; no new test-only exports or + mock-only characterization were needed. +- Rebased onto `b1ebe3de3c` after master added fast-response acknowledgements + and action queue notifications. The only conflict was the import block; + retain upstream imports and add the normalization import. All 21 extracted + declarations match the new base, ignoring added exports and two narrowed + input types. Every remaining non-import declaration, including the entire + service body, matches the new base exactly. +- Current size: 39,155 lines / 1,509,090 bytes before extraction; 38,633 lines / + 1,491,422 bytes after. Net reduction: 522 lines / 17,668 bytes. Destination: + 552 lines / 19,028 bytes, including imports and the shared text-limit constant. +- The module takes only the Teams parser method and the two delivery fields it + reads, rather than requiring an entire runtime or database row. These are + type-only changes. It imports sibling implementations directly, never the + service entry point. Public service exports remain unchanged. +- Initial full-test/typecheck runs were interrupted to rebase and are not + passes. Rebased integration selection: 48 passed (the other 1,185 cases were + excluded by the test-name filter). Module-boundary and whitespace checks pass. + Workspace `pnpm -r typecheck` and `pnpm build`: passed. Full `pnpm test:run`: + pending. + +Reproduce the focused service-path check with the repository-pinned pnpm: + +```sh +pnpm exec vitest run server/src/__tests__/chat-channels.integration.test.ts \ + -t 'records message edits and deletes durably|redacts lifecycle edits|Telegram edit|Teams edit, soft-delete|verified Telegram edited_message|verified GitHub comment edit|provider timestamp provenance|Telegram retained zero-message admission|Telegram current rich inbound compatibility|deduplicates Slack file-only revisions|preserves exact Discord source authority' +``` + +Branch: `codex/chat-channel-provider-messages`. The user authorized publishing +this slice to `paperclipai/paperclip`; subsequent extractions are not part of +this PR. diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index d81f4ea8744..32e11ec69c7 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -1,3 +1,24 @@ +import { + MAX_INBOUND_TEXT, + stableExternalPrincipalId, + type GitHubLifecycleEvent, + type LifecycleActor, + lifecycleActorFromAuthor, + slackLifecycleFilesDigest, + discordLifecycleFilesDigest, + microsoftTeamsLifecycleEventFromPayload, + telegramLifecycleEventFromPayload, + isTelegramEditedMessageRaw, + isTelegramMigrationRaw, + telegramMessageSequence, + telegramMessageId, + telegramZeroMessageId, + telegramDeliveryHasZeroMessageId, + telegramMessageSentAt, + slackMessageSentAt, + teamsActivitySentAt, + githubLifecycleEventFromRequest, +} from "./chat-channels/provider-messages.js"; import { dueChatAction, nextChatActionAt, SLACK_COMMAND_POST_STALE_MS, SLACK_COMMAND_EXPLICIT_RETRY_STALE_MS, SLACK_COMMAND_ADMISSION_STALE_MS, PROVIDER_EFFECT_STALE_MS } from "./chat-action-work.js"; import type { ScheduledChatAction } from "./chat-work-notifications.js"; import { notifyChatDeliveryWork, notifyChatPublicationWork, notifyChatEndpointWork, notifyChatActionWork, notifyChatVerificationWork } from "./chat-work-notifications.js"; @@ -381,7 +402,6 @@ import { } from "./chat-inbound-wakeup-publications.js"; import { normalizeMicrosoftTeamsCredentialIds, - normalizeMicrosoftTeamsExternalPrincipalId, } from "./chat-teams-credentials.js"; import { isTeamsInlineImageContentType, @@ -847,7 +867,6 @@ const SUPPLIED_CREDENTIAL_KEYS: Record = { telegram: ["botToken"], }; -const MAX_INBOUND_TEXT = 100_000; const MAX_ERROR_TEXT = 2_000; const DELIVERY_PROCESSING_STALE_MS = 60_000; // Reactions can beat the transaction that persists a just-sent provider @@ -1737,19 +1756,6 @@ function telegramGuidanceCommand( return "unknown"; } -function stableExternalPrincipalId( - provider: ChatProvider, - author: Author, - raw?: unknown, -): string { - if (provider !== "microsoft-teams" || !raw || typeof raw !== "object") - return author.userId; - const from = (raw as { from?: unknown }).from; - if (!from || typeof from !== "object") return author.userId; - const aadObjectId = (from as { aadObjectId?: unknown }).aadObjectId; - return normalizeMicrosoftTeamsExternalPrincipalId(aadObjectId, author.userId); -} - function safeTitle(text: string, fallback: string): string { const line = text .replace(/<@[A-Z0-9]+>/gi, "") @@ -2037,534 +2043,6 @@ function isSupportedTelegramWebhookBaseUrl(value: string | null): boolean { } } -type GitHubLifecycleEvent = { - actor?: LifecycleActor; - isBotMessage?: true; - eventKind: "message_updated" | "message_deleted"; - messageId: string; - providerEventId?: string; - providerMessageSequence: number | null; - providerSentAt: string | null; - revision: string; - text: string; - threadId: string; -}; - -type TelegramLifecycleEvent = { - actor: LifecycleActor; - eventKind: "message_updated"; - messageId: string; - providerEventId?: string; - providerMessageSequence: number; - providerSentAt: string; - providerUpdateId: number | null; - revision: string; - text: string; - threadId: string; -}; - -type LifecycleActor = { - displayName: string; - externalId: string; - handle: string; -}; - -type MicrosoftTeamsLifecycleEvent = { - actor?: LifecycleActor; - eventKind: "message_updated" | "message_deleted" | "message_restored"; - messageId: string; - providerSentAt: string | null; - raw?: unknown; - revision: string; - text: string; - threadId: string; -}; - -function lifecycleActorFromAuthor( - provider: ChatProvider, - author: Author, - raw?: unknown, -): LifecycleActor { - return { - externalId: stableExternalPrincipalId(provider, author, raw), - displayName: author.fullName, - handle: author.userName, - }; -} - -function slackLifecycleFilesDigest(raw: unknown): string { - const files = - raw && typeof raw === "object" && !Array.isArray(raw) - ? (raw as Record).files - : undefined; - // Match the pinned Slack adapter's content-change projection. Transport - // locators/unfurls are not content identity and must never enter durable keys. - const projection = Array.isArray(files) - ? files.map((file) => { - const row = - file && typeof file === "object" && !Array.isArray(file) - ? (file as Record) - : {}; - return [ - ...["id", "name", "mimetype"].map((key) => - typeof row[key] === "string" ? row[key] : null, - ), - ...["size", "original_w", "original_h"].map((key) => - typeof row[key] === "number" && Number.isFinite(row[key]) - ? row[key] - : null, - ), - ]; - }) - : []; - return createHash("sha256").update(JSON.stringify(projection)).digest("hex"); -} - -function discordLifecycleFilesDigest(raw: unknown): string { - const attachments = - raw && typeof raw === "object" && !Array.isArray(raw) - ? (raw as Record).attachments - : undefined; - // Match the pinned Gateway adapter's flattened, ordered file projection. - // Rotating CDN URLs are transport authority, never source revision identity. - const projection = Array.isArray(attachments) - ? attachments.map((attachment) => { - const row = - attachment && - typeof attachment === "object" && - !Array.isArray(attachment) - ? (attachment as Record) - : {}; - return [ - ...["id", "filename", "content_type"].map((key) => - typeof row[key] === "string" ? row[key] : null, - ), - typeof row.size === "number" && Number.isFinite(row.size) - ? row.size - : null, - ]; - }) - : []; - return createHash("sha256").update(JSON.stringify(projection)).digest("hex"); -} - -function telegramLifecycleActor(message: { - from?: unknown; - sender_chat?: unknown; -}): LifecycleActor | null { - const candidate = - message.from && - typeof message.from === "object" && - !Array.isArray(message.from) - ? (message.from as Record) - : message.sender_chat && - typeof message.sender_chat === "object" && - !Array.isArray(message.sender_chat) - ? (message.sender_chat as Record) - : null; - if (!candidate) return null; - const id = candidate.id; - if (typeof id !== "string" && typeof id !== "number") return null; - const handle = - typeof candidate.username === "string" ? candidate.username : ""; - const displayName = - [candidate.first_name, candidate.last_name] - .filter( - (value): value is string => typeof value === "string" && Boolean(value), - ) - .join(" ") || - (typeof candidate.title === "string" ? candidate.title : "") || - handle || - String(id); - return { externalId: String(id), displayName, handle }; -} - -function microsoftTeamsLifecycleEventFromPayload( - payload: unknown, - endpointRuntime: ChatEndpointRuntime, -): MicrosoftTeamsLifecycleEvent | null { - if (!payload || typeof payload !== "object" || Array.isArray(payload)) - return null; - const activity = payload as { - channelData?: { eventType?: unknown }; - timestamp?: unknown; - type?: unknown; - }; - const eventType = activity.channelData?.eventType; - const isEdit = - activity.type === "messageUpdate" && eventType === "editMessage"; - const isRestore = - activity.type === "messageUpdate" && eventType === "undeleteMessage"; - const isDelete = - activity.type === "messageDelete" && eventType === "softDeleteMessage"; - if (!isEdit && !isRestore && !isDelete) return null; - const message = endpointRuntime.parseMicrosoftTeamsMessage(payload); - if (!message?.id || !message.threadId) return null; - const body = message.text.slice(0, MAX_INBOUND_TEXT); - const providerSentAt = - typeof activity.timestamp === "string" && - Number.isFinite(Date.parse(activity.timestamp)) - ? new Date(activity.timestamp).toISOString() - : null; - const bodyHash = createHash("sha256").update(body).digest("hex"); - const revision = providerSentAt ? `${providerSentAt}:${bodyHash}` : bodyHash; - return { - ...(!isDelete - ? { - actor: lifecycleActorFromAuthor( - "microsoft-teams", - message.author, - message.raw, - ), - } - : {}), - eventKind: isDelete - ? "message_deleted" - : isRestore - ? "message_restored" - : "message_updated", - messageId: message.id, - providerSentAt, - raw: message.raw, - revision, - text: isDelete - ? "An external message in this conversation was deleted." - : isRestore - ? `An external message was restored:\n\n${body}` - : `An external message was edited:\n\n${body}`, - threadId: message.threadId, - }; -} - -function telegramLifecycleEventFromPayload( - payload: unknown, -): TelegramLifecycleEvent | null { - if (!payload || typeof payload !== "object" || Array.isArray(payload)) - return null; - const update = payload as { - edited_message?: unknown; - update_id?: unknown; - }; - const edited = update.edited_message; - if (!edited || typeof edited !== "object" || Array.isArray(edited)) - return null; - const message = edited as { - caption?: unknown; - chat?: { id?: unknown }; - edit_date?: unknown; - from?: unknown; - message_id?: unknown; - message_thread_id?: unknown; - sender_chat?: unknown; - text?: unknown; - rich_message?: unknown; - }; - const chatId = message.chat?.id; - const messageId = message.message_id; - const editDate = message.edit_date; - if ( - (typeof chatId !== "string" && typeof chatId !== "number") || - typeof messageId !== "number" || - !Number.isSafeInteger(messageId) || - messageId < 0 || - typeof editDate !== "number" || - !Number.isSafeInteger(editDate) || - editDate <= 0 - ) - return null; - const topicId = message.message_thread_id; - if (topicId !== undefined && typeof topicId !== "number") return null; - const rich = normalizeTelegramRichMessage(message); - const body = - rich?.text ?? - (typeof message.text === "string" - ? message.text - : typeof message.caption === "string" - ? message.caption - : ""); - const chat = String(chatId); - const providerUpdateId = - typeof update.update_id === "number" && - Number.isSafeInteger(update.update_id) && - update.update_id >= 0 - ? update.update_id - : null; - const bodyHash = createHash("sha256").update(body).digest("hex"); - const actor = telegramLifecycleActor(message); - if (!actor) return null; - return { - actor, - eventKind: "message_updated", - messageId: `${chat}:${messageId}`, - ...(providerUpdateId === null - ? {} - : { providerEventId: `telegram:update:${providerUpdateId}` }), - providerMessageSequence: messageId, - providerSentAt: new Date(editDate * 1_000).toISOString(), - providerUpdateId, - // Bot API timestamps have one-second resolution. The update id is the - // authoritative identity when present; retaining a content hash keeps the - // fallback path from collapsing two distinct edits in that same second. - revision: `${editDate}:${bodyHash}${rich ? `:${rich.mediaDigest}` : ""}`, - text: `An external message was edited:\n\n${body.slice(0, MAX_INBOUND_TEXT)}`, - threadId: - topicId === undefined - ? `telegram:${chat}` - : `telegram:${chat}:${topicId}`, - }; -} - -function isTelegramEditedMessageRaw(raw: unknown): boolean { - return ( - !!raw && - typeof raw === "object" && - !Array.isArray(raw) && - typeof (raw as { edit_date?: unknown }).edit_date === "number" - ); -} - -function isTelegramMigrationRaw(raw: unknown): boolean { - if (!raw || typeof raw !== "object" || Array.isArray(raw)) return false; - const value = raw as { - migrate_from_chat_id?: unknown; - migrate_to_chat_id?: unknown; - }; - return ( - value.migrate_from_chat_id !== undefined || - value.migrate_to_chat_id !== undefined - ); -} - -function telegramMessageSequence(raw: unknown): number | null { - if (!raw || typeof raw !== "object" || Array.isArray(raw)) return null; - const value = (raw as { message_id?: unknown }).message_id; - return typeof value === "number" && Number.isSafeInteger(value) && value > 0 - ? value - : null; -} - -function telegramMessageId(raw: unknown): string | null { - if (!raw || typeof raw !== "object" || Array.isArray(raw)) return null; - const value = raw as { - chat?: { id?: unknown }; - message_id?: unknown; - }; - const messageId = value.message_id; - const chatId = value.chat?.id; - if ( - typeof messageId !== "number" || - !Number.isSafeInteger(messageId) || - messageId <= 0 - ) - return null; - const normalizedChatId = - typeof chatId === "number" && Number.isSafeInteger(chatId) - ? String(chatId) - : typeof chatId === "string" && /^-?\d+$/.test(chatId) - ? chatId - : null; - return normalizedChatId ? `${normalizedChatId}:${messageId}` : null; -} - -function telegramZeroMessageId(value: unknown): boolean { - // The pinned adapter qualifies native message IDs with the numeric chat ID. - // Keep legacy synthetic/non-numeric IDs compatible; zero is specifically - // the provider's non-ordinary message identity, never an admitted source. - return typeof value === "string" && /^(?:-?\d+:)?0$/.test(value); -} - -function telegramDeliveryHasZeroMessageId( - delivery: DeliveryRow, - provider: ChatProvider, -): boolean { - if (provider !== "telegram") return false; - const normalized = delivery.normalizedEvent as { - message?: { - providerMessageId?: unknown; - providerMessageSequence?: unknown; - }; - conversation?: { externalThreadId?: unknown }; - }; - if ( - telegramZeroMessageId(normalized.message?.providerMessageId) || - normalized.message?.providerMessageSequence === 0 - ) - return true; - const threadId = normalized.conversation?.externalThreadId; - // Check the independently retained event key only inside its exact thread - // namespace, not arbitrary IDs ending in ':0' from synthetic commands. - return ( - typeof threadId === "string" && - delivery.providerEventId.startsWith(`${threadId}:`) && - telegramZeroMessageId(delivery.providerEventId.slice(threadId.length + 1)) - ); -} - -function telegramMessageSentAt(raw: unknown): Date | null { - if (!raw || typeof raw !== "object" || Array.isArray(raw)) return null; - const value = (raw as { date?: unknown }).date; - if (typeof value !== "number" || !Number.isSafeInteger(value) || value <= 0) - return null; - const sentAt = new Date(value * 1_000); - return Number.isFinite(sentAt.getTime()) ? sentAt : null; -} - -function slackMessageSentAt(raw: unknown, messageId: string): Date | null { - if (!raw || typeof raw !== "object" || Array.isArray(raw)) return null; - const ts = (raw as { ts?: unknown }).ts; - if ( - ts !== messageId || - typeof ts !== "string" || - !/^[1-9][0-9]{0,12}\.[0-9]{6}$/.test(ts) - ) - return null; - const date = new Date(Number(ts) * 1_000); - return Number.isFinite(date.getTime()) ? date : null; -} - -function teamsActivitySentAt(raw: unknown): Date | null { - if (!raw || typeof raw !== "object" || Array.isArray(raw)) return null; - const value = (raw as { timestamp?: unknown }).timestamp; - // The SDK may supply a Date, but its missing-timestamp display fallback is - // only on Message.metadata. Never use that fallback as provider chronology. - if (value instanceof Date) - return Number.isFinite(value.getTime()) ? new Date(value.getTime()) : null; - if ( - typeof value !== "string" || - value.length > 64 || - !/^\d{4}-\d{2}-\d{2}T(?:[01]\d|2[0-3]):[0-5]\d:[0-5]\d(?:\.\d{1,7})?(?:Z|[+-](?:[01]\d|2[0-3]):[0-5]\d)$/.test( - value, - ) - ) - return null; - // Date.parse otherwise normalizes nonexistent dates such as February 30. - const day = value.slice(0, 10); - const calendar = new Date(`${day}T00:00:00.000Z`); - if ( - !Number.isFinite(calendar.getTime()) || - calendar.toISOString().slice(0, 10) !== day - ) - return null; - const sentAt = new Date(value); - return Number.isFinite(sentAt.getTime()) ? sentAt : null; -} - -async function githubLifecycleEventFromRequest( - request: Request, -): Promise { - const eventType = request.headers.get("x-github-event"); - const deliveryId = request.headers.get("x-github-delivery")?.trim() || null; - if ( - eventType !== "issue_comment" && - eventType !== "pull_request_review_comment" - ) - return null; - const payload = (await request.json()) as { - action?: unknown; - comment?: { - id?: unknown; - in_reply_to_id?: unknown; - body?: unknown; - updated_at?: unknown; - user?: { type?: unknown }; - }; - issue?: { number?: unknown; pull_request?: unknown }; - pull_request?: { number?: unknown }; - repository?: { name?: unknown; owner?: { login?: unknown } }; - sender?: { id?: unknown; login?: unknown; type?: unknown }; - }; - if (payload.action !== "edited" && payload.action !== "deleted") return null; - const owner = payload.repository?.owner?.login; - const repo = payload.repository?.name; - const messageId = payload.comment?.id; - if ( - typeof owner !== "string" || - typeof repo !== "string" || - (typeof messageId !== "string" && typeof messageId !== "number") - ) - return null; - let threadId: string; - if (eventType === "issue_comment") { - const number = payload.issue?.number; - if (typeof number !== "number") return null; - threadId = payload.issue?.pull_request - ? `github:${owner}/${repo}:${number}` - : `github:${owner}/${repo}:issue:${number}`; - } else { - const number = payload.pull_request?.number; - const rootCommentId = - payload.comment?.in_reply_to_id ?? payload.comment?.id; - if ( - typeof number !== "number" || - (typeof rootCommentId !== "string" && typeof rootCommentId !== "number") - ) - return null; - threadId = `github:${owner}/${repo}:${number}:rc:${rootCommentId}`; - } - const eventKind = - payload.action === "edited" ? "message_updated" : "message_deleted"; - const body = - typeof payload.comment?.body === "string" - ? payload.comment.body.slice(0, MAX_INBOUND_TEXT) - : ""; - const providerRevision = - typeof payload.comment?.updated_at === "string" - ? payload.comment.updated_at - : payload.action; - const parsedProviderSentAt = new Date(providerRevision); - const providerSentAt = Number.isFinite(parsedProviderSentAt.getTime()) - ? parsedProviderSentAt.toISOString() - : null; - const numericMessageId = - typeof messageId === "number" - ? messageId - : typeof messageId === "string" && /^\d+$/.test(messageId) - ? Number(messageId) - : null; - const providerMessageSequence = - numericMessageId !== null && - Number.isSafeInteger(numericMessageId) && - numericMessageId >= 0 - ? numericMessageId - : null; - const senderId = payload.sender?.id; - const senderLogin = payload.sender?.login; - const actor = - (typeof senderId === "string" || typeof senderId === "number") && - typeof senderLogin === "string" && - senderLogin.length > 0 - ? { - externalId: String(senderId), - displayName: senderLogin, - handle: senderLogin, - } - : undefined; - return { - ...(actor ? { actor } : {}), - ...(payload.comment?.user?.type === "Bot" || payload.sender?.type === "Bot" - ? { isBotMessage: true as const } - : {}), - eventKind, - messageId: String(messageId), - ...(deliveryId ? { providerEventId: `github:delivery:${deliveryId}` } : {}), - providerMessageSequence, - providerSentAt, - // GitHub's updated_at value can have coarser resolution than a quick - // sequence of edits. Include the normalized body so distinct edits at the - // same timestamp remain durable while an exact webhook redelivery still - // deduplicates. - revision: - eventKind === "message_updated" - ? `${providerRevision}:${createHash("sha256").update(body).digest("hex")}` - : providerRevision, - text: - eventKind === "message_updated" - ? `An external message was edited:\n\n${body}` - : "An external message in this conversation was deleted.", - threadId, - }; -} - function githubRepositoryInventoryItemFromPayload( payload: unknown, ): ChatProviderResourceInventoryItem | null { diff --git a/server/src/services/chat-channels/provider-messages.ts b/server/src/services/chat-channels/provider-messages.ts new file mode 100644 index 00000000000..2b7d38c0328 --- /dev/null +++ b/server/src/services/chat-channels/provider-messages.ts @@ -0,0 +1,552 @@ +import { createHash } from "node:crypto"; +import type { chatDeliveries } from "@paperclipai/db"; +import type { ChatProvider } from "@paperclipai/shared"; +import type { Author } from "chat"; +import { normalizeMicrosoftTeamsExternalPrincipalId } from "../chat-teams-credentials.js"; +import { normalizeTelegramRichMessage } from "../chat-telegram-rich-intake.js"; +import type { ChatEndpointRuntime } from "../voice/voice-runtime.js"; + +// Provider message identity and chronology only. Callers retain authentication, +// authorization, durable admission, and delivery ownership. +export const MAX_INBOUND_TEXT = 100_000; + +export function stableExternalPrincipalId( + provider: ChatProvider, + author: Author, + raw?: unknown, +): string { + if (provider !== "microsoft-teams" || !raw || typeof raw !== "object") + return author.userId; + const from = (raw as { from?: unknown }).from; + if (!from || typeof from !== "object") return author.userId; + const aadObjectId = (from as { aadObjectId?: unknown }).aadObjectId; + return normalizeMicrosoftTeamsExternalPrincipalId(aadObjectId, author.userId); +} + +export type GitHubLifecycleEvent = { + actor?: LifecycleActor; + isBotMessage?: true; + eventKind: "message_updated" | "message_deleted"; + messageId: string; + providerEventId?: string; + providerMessageSequence: number | null; + providerSentAt: string | null; + revision: string; + text: string; + threadId: string; +}; + +type TelegramLifecycleEvent = { + actor: LifecycleActor; + eventKind: "message_updated"; + messageId: string; + providerEventId?: string; + providerMessageSequence: number; + providerSentAt: string; + providerUpdateId: number | null; + revision: string; + text: string; + threadId: string; +}; + +export type LifecycleActor = { + displayName: string; + externalId: string; + handle: string; +}; + +type MicrosoftTeamsLifecycleEvent = { + actor?: LifecycleActor; + eventKind: "message_updated" | "message_deleted" | "message_restored"; + messageId: string; + providerSentAt: string | null; + raw?: unknown; + revision: string; + text: string; + threadId: string; +}; + +export function lifecycleActorFromAuthor( + provider: ChatProvider, + author: Author, + raw?: unknown, +): LifecycleActor { + return { + externalId: stableExternalPrincipalId(provider, author, raw), + displayName: author.fullName, + handle: author.userName, + }; +} + +export function slackLifecycleFilesDigest(raw: unknown): string { + const files = + raw && typeof raw === "object" && !Array.isArray(raw) + ? (raw as Record).files + : undefined; + // Match the pinned Slack adapter's content-change projection. Transport + // locators/unfurls are not content identity and must never enter durable keys. + const projection = Array.isArray(files) + ? files.map((file) => { + const row = + file && typeof file === "object" && !Array.isArray(file) + ? (file as Record) + : {}; + return [ + ...["id", "name", "mimetype"].map((key) => + typeof row[key] === "string" ? row[key] : null, + ), + ...["size", "original_w", "original_h"].map((key) => + typeof row[key] === "number" && Number.isFinite(row[key]) + ? row[key] + : null, + ), + ]; + }) + : []; + return createHash("sha256").update(JSON.stringify(projection)).digest("hex"); +} + +export function discordLifecycleFilesDigest(raw: unknown): string { + const attachments = + raw && typeof raw === "object" && !Array.isArray(raw) + ? (raw as Record).attachments + : undefined; + // Match the pinned Gateway adapter's flattened, ordered file projection. + // Rotating CDN URLs are transport authority, never source revision identity. + const projection = Array.isArray(attachments) + ? attachments.map((attachment) => { + const row = + attachment && + typeof attachment === "object" && + !Array.isArray(attachment) + ? (attachment as Record) + : {}; + return [ + ...["id", "filename", "content_type"].map((key) => + typeof row[key] === "string" ? row[key] : null, + ), + typeof row.size === "number" && Number.isFinite(row.size) + ? row.size + : null, + ]; + }) + : []; + return createHash("sha256").update(JSON.stringify(projection)).digest("hex"); +} + +function telegramLifecycleActor(message: { + from?: unknown; + sender_chat?: unknown; +}): LifecycleActor | null { + const candidate = + message.from && + typeof message.from === "object" && + !Array.isArray(message.from) + ? (message.from as Record) + : message.sender_chat && + typeof message.sender_chat === "object" && + !Array.isArray(message.sender_chat) + ? (message.sender_chat as Record) + : null; + if (!candidate) return null; + const id = candidate.id; + if (typeof id !== "string" && typeof id !== "number") return null; + const handle = + typeof candidate.username === "string" ? candidate.username : ""; + const displayName = + [candidate.first_name, candidate.last_name] + .filter( + (value): value is string => typeof value === "string" && Boolean(value), + ) + .join(" ") || + (typeof candidate.title === "string" ? candidate.title : "") || + handle || + String(id); + return { externalId: String(id), displayName, handle }; +} + +export function microsoftTeamsLifecycleEventFromPayload( + payload: unknown, + endpointRuntime: Pick, +): MicrosoftTeamsLifecycleEvent | null { + if (!payload || typeof payload !== "object" || Array.isArray(payload)) + return null; + const activity = payload as { + channelData?: { eventType?: unknown }; + timestamp?: unknown; + type?: unknown; + }; + const eventType = activity.channelData?.eventType; + const isEdit = + activity.type === "messageUpdate" && eventType === "editMessage"; + const isRestore = + activity.type === "messageUpdate" && eventType === "undeleteMessage"; + const isDelete = + activity.type === "messageDelete" && eventType === "softDeleteMessage"; + if (!isEdit && !isRestore && !isDelete) return null; + const message = endpointRuntime.parseMicrosoftTeamsMessage(payload); + if (!message?.id || !message.threadId) return null; + const body = message.text.slice(0, MAX_INBOUND_TEXT); + const providerSentAt = + typeof activity.timestamp === "string" && + Number.isFinite(Date.parse(activity.timestamp)) + ? new Date(activity.timestamp).toISOString() + : null; + const bodyHash = createHash("sha256").update(body).digest("hex"); + const revision = providerSentAt ? `${providerSentAt}:${bodyHash}` : bodyHash; + return { + ...(!isDelete + ? { + actor: lifecycleActorFromAuthor( + "microsoft-teams", + message.author, + message.raw, + ), + } + : {}), + eventKind: isDelete + ? "message_deleted" + : isRestore + ? "message_restored" + : "message_updated", + messageId: message.id, + providerSentAt, + raw: message.raw, + revision, + text: isDelete + ? "An external message in this conversation was deleted." + : isRestore + ? `An external message was restored:\n\n${body}` + : `An external message was edited:\n\n${body}`, + threadId: message.threadId, + }; +} + +export function telegramLifecycleEventFromPayload( + payload: unknown, +): TelegramLifecycleEvent | null { + if (!payload || typeof payload !== "object" || Array.isArray(payload)) + return null; + const update = payload as { + edited_message?: unknown; + update_id?: unknown; + }; + const edited = update.edited_message; + if (!edited || typeof edited !== "object" || Array.isArray(edited)) + return null; + const message = edited as { + caption?: unknown; + chat?: { id?: unknown }; + edit_date?: unknown; + from?: unknown; + message_id?: unknown; + message_thread_id?: unknown; + sender_chat?: unknown; + text?: unknown; + rich_message?: unknown; + }; + const chatId = message.chat?.id; + const messageId = message.message_id; + const editDate = message.edit_date; + if ( + (typeof chatId !== "string" && typeof chatId !== "number") || + typeof messageId !== "number" || + !Number.isSafeInteger(messageId) || + messageId < 0 || + typeof editDate !== "number" || + !Number.isSafeInteger(editDate) || + editDate <= 0 + ) + return null; + const topicId = message.message_thread_id; + if (topicId !== undefined && typeof topicId !== "number") return null; + const rich = normalizeTelegramRichMessage(message); + const body = + rich?.text ?? + (typeof message.text === "string" + ? message.text + : typeof message.caption === "string" + ? message.caption + : ""); + const chat = String(chatId); + const providerUpdateId = + typeof update.update_id === "number" && + Number.isSafeInteger(update.update_id) && + update.update_id >= 0 + ? update.update_id + : null; + const bodyHash = createHash("sha256").update(body).digest("hex"); + const actor = telegramLifecycleActor(message); + if (!actor) return null; + return { + actor, + eventKind: "message_updated", + messageId: `${chat}:${messageId}`, + ...(providerUpdateId === null + ? {} + : { providerEventId: `telegram:update:${providerUpdateId}` }), + providerMessageSequence: messageId, + providerSentAt: new Date(editDate * 1_000).toISOString(), + providerUpdateId, + // Bot API timestamps have one-second resolution. The update id is the + // authoritative identity when present; retaining a content hash keeps the + // fallback path from collapsing two distinct edits in that same second. + revision: `${editDate}:${bodyHash}${rich ? `:${rich.mediaDigest}` : ""}`, + text: `An external message was edited:\n\n${body.slice(0, MAX_INBOUND_TEXT)}`, + threadId: + topicId === undefined + ? `telegram:${chat}` + : `telegram:${chat}:${topicId}`, + }; +} + +export function isTelegramEditedMessageRaw(raw: unknown): boolean { + return ( + !!raw && + typeof raw === "object" && + !Array.isArray(raw) && + typeof (raw as { edit_date?: unknown }).edit_date === "number" + ); +} + +export function isTelegramMigrationRaw(raw: unknown): boolean { + if (!raw || typeof raw !== "object" || Array.isArray(raw)) return false; + const value = raw as { + migrate_from_chat_id?: unknown; + migrate_to_chat_id?: unknown; + }; + return ( + value.migrate_from_chat_id !== undefined || + value.migrate_to_chat_id !== undefined + ); +} + +export function telegramMessageSequence(raw: unknown): number | null { + if (!raw || typeof raw !== "object" || Array.isArray(raw)) return null; + const value = (raw as { message_id?: unknown }).message_id; + return typeof value === "number" && Number.isSafeInteger(value) && value > 0 + ? value + : null; +} + +export function telegramMessageId(raw: unknown): string | null { + if (!raw || typeof raw !== "object" || Array.isArray(raw)) return null; + const value = raw as { + chat?: { id?: unknown }; + message_id?: unknown; + }; + const messageId = value.message_id; + const chatId = value.chat?.id; + if ( + typeof messageId !== "number" || + !Number.isSafeInteger(messageId) || + messageId <= 0 + ) + return null; + const normalizedChatId = + typeof chatId === "number" && Number.isSafeInteger(chatId) + ? String(chatId) + : typeof chatId === "string" && /^-?\d+$/.test(chatId) + ? chatId + : null; + return normalizedChatId ? `${normalizedChatId}:${messageId}` : null; +} + +export function telegramZeroMessageId(value: unknown): boolean { + // The pinned adapter qualifies native message IDs with the numeric chat ID. + // Keep legacy synthetic/non-numeric IDs compatible; zero is specifically + // the provider's non-ordinary message identity, never an admitted source. + return typeof value === "string" && /^(?:-?\d+:)?0$/.test(value); +} + +export function telegramDeliveryHasZeroMessageId( + delivery: Pick, + provider: ChatProvider, +): boolean { + if (provider !== "telegram") return false; + const normalized = delivery.normalizedEvent as { + message?: { + providerMessageId?: unknown; + providerMessageSequence?: unknown; + }; + conversation?: { externalThreadId?: unknown }; + }; + if ( + telegramZeroMessageId(normalized.message?.providerMessageId) || + normalized.message?.providerMessageSequence === 0 + ) + return true; + const threadId = normalized.conversation?.externalThreadId; + // Check the independently retained event key only inside its exact thread + // namespace, not arbitrary IDs ending in ':0' from synthetic commands. + return ( + typeof threadId === "string" && + delivery.providerEventId.startsWith(`${threadId}:`) && + telegramZeroMessageId(delivery.providerEventId.slice(threadId.length + 1)) + ); +} + +export function telegramMessageSentAt(raw: unknown): Date | null { + if (!raw || typeof raw !== "object" || Array.isArray(raw)) return null; + const value = (raw as { date?: unknown }).date; + if (typeof value !== "number" || !Number.isSafeInteger(value) || value <= 0) + return null; + const sentAt = new Date(value * 1_000); + return Number.isFinite(sentAt.getTime()) ? sentAt : null; +} + +export function slackMessageSentAt(raw: unknown, messageId: string): Date | null { + if (!raw || typeof raw !== "object" || Array.isArray(raw)) return null; + const ts = (raw as { ts?: unknown }).ts; + if ( + ts !== messageId || + typeof ts !== "string" || + !/^[1-9][0-9]{0,12}\.[0-9]{6}$/.test(ts) + ) + return null; + const date = new Date(Number(ts) * 1_000); + return Number.isFinite(date.getTime()) ? date : null; +} + +export function teamsActivitySentAt(raw: unknown): Date | null { + if (!raw || typeof raw !== "object" || Array.isArray(raw)) return null; + const value = (raw as { timestamp?: unknown }).timestamp; + // The SDK may supply a Date, but its missing-timestamp display fallback is + // only on Message.metadata. Never use that fallback as provider chronology. + if (value instanceof Date) + return Number.isFinite(value.getTime()) ? new Date(value.getTime()) : null; + if ( + typeof value !== "string" || + value.length > 64 || + !/^\d{4}-\d{2}-\d{2}T(?:[01]\d|2[0-3]):[0-5]\d:[0-5]\d(?:\.\d{1,7})?(?:Z|[+-](?:[01]\d|2[0-3]):[0-5]\d)$/.test( + value, + ) + ) + return null; + // Date.parse otherwise normalizes nonexistent dates such as February 30. + const day = value.slice(0, 10); + const calendar = new Date(`${day}T00:00:00.000Z`); + if ( + !Number.isFinite(calendar.getTime()) || + calendar.toISOString().slice(0, 10) !== day + ) + return null; + const sentAt = new Date(value); + return Number.isFinite(sentAt.getTime()) ? sentAt : null; +} + +export async function githubLifecycleEventFromRequest( + request: Request, +): Promise { + const eventType = request.headers.get("x-github-event"); + const deliveryId = request.headers.get("x-github-delivery")?.trim() || null; + if ( + eventType !== "issue_comment" && + eventType !== "pull_request_review_comment" + ) + return null; + const payload = (await request.json()) as { + action?: unknown; + comment?: { + id?: unknown; + in_reply_to_id?: unknown; + body?: unknown; + updated_at?: unknown; + user?: { type?: unknown }; + }; + issue?: { number?: unknown; pull_request?: unknown }; + pull_request?: { number?: unknown }; + repository?: { name?: unknown; owner?: { login?: unknown } }; + sender?: { id?: unknown; login?: unknown; type?: unknown }; + }; + if (payload.action !== "edited" && payload.action !== "deleted") return null; + const owner = payload.repository?.owner?.login; + const repo = payload.repository?.name; + const messageId = payload.comment?.id; + if ( + typeof owner !== "string" || + typeof repo !== "string" || + (typeof messageId !== "string" && typeof messageId !== "number") + ) + return null; + let threadId: string; + if (eventType === "issue_comment") { + const number = payload.issue?.number; + if (typeof number !== "number") return null; + threadId = payload.issue?.pull_request + ? `github:${owner}/${repo}:${number}` + : `github:${owner}/${repo}:issue:${number}`; + } else { + const number = payload.pull_request?.number; + const rootCommentId = + payload.comment?.in_reply_to_id ?? payload.comment?.id; + if ( + typeof number !== "number" || + (typeof rootCommentId !== "string" && typeof rootCommentId !== "number") + ) + return null; + threadId = `github:${owner}/${repo}:${number}:rc:${rootCommentId}`; + } + const eventKind = + payload.action === "edited" ? "message_updated" : "message_deleted"; + const body = + typeof payload.comment?.body === "string" + ? payload.comment.body.slice(0, MAX_INBOUND_TEXT) + : ""; + const providerRevision = + typeof payload.comment?.updated_at === "string" + ? payload.comment.updated_at + : payload.action; + const parsedProviderSentAt = new Date(providerRevision); + const providerSentAt = Number.isFinite(parsedProviderSentAt.getTime()) + ? parsedProviderSentAt.toISOString() + : null; + const numericMessageId = + typeof messageId === "number" + ? messageId + : typeof messageId === "string" && /^\d+$/.test(messageId) + ? Number(messageId) + : null; + const providerMessageSequence = + numericMessageId !== null && + Number.isSafeInteger(numericMessageId) && + numericMessageId >= 0 + ? numericMessageId + : null; + const senderId = payload.sender?.id; + const senderLogin = payload.sender?.login; + const actor = + (typeof senderId === "string" || typeof senderId === "number") && + typeof senderLogin === "string" && + senderLogin.length > 0 + ? { + externalId: String(senderId), + displayName: senderLogin, + handle: senderLogin, + } + : undefined; + return { + ...(actor ? { actor } : {}), + ...(payload.comment?.user?.type === "Bot" || payload.sender?.type === "Bot" + ? { isBotMessage: true as const } + : {}), + eventKind, + messageId: String(messageId), + ...(deliveryId ? { providerEventId: `github:delivery:${deliveryId}` } : {}), + providerMessageSequence, + providerSentAt, + // GitHub's updated_at value can have coarser resolution than a quick + // sequence of edits. Include the normalized body so distinct edits at the + // same timestamp remain durable while an exact webhook redelivery still + // deduplicates. + revision: + eventKind === "message_updated" + ? `${providerRevision}:${createHash("sha256").update(body).digest("hex")}` + : providerRevision, + text: + eventKind === "message_updated" + ? `An external message was edited:\n\n${body}` + : "An external message in this conversation was deleted.", + threadId, + }; +}