import { GITHUB_RECOVERY_STATE_KEY, GITHUB_RECOVERY_INTERVAL_MS, GITHUB_RECOVERY_WINDOW_MS, GITHUB_RECOVERY_SCAN_LEASE_MS, githubRecoveryWindow, type GitHubRecoveryWindow } from "./webhook-recovery-state.js"; import { githubWebhookRecoverySchedule } from "./webhook-recovery-schedule.js"; import { registerGitHubWebhookRecoveryWork } from "./webhook-recovery-work.js"; import { notifyGitHubWebhookRecovery } from "./webhook-recovery-notifications.js"; import { createChatSdkStatePersistence } from "../sdk-state.js"; import type { ChatSdkStateCompareAndSetInput } from "../../chat-sdk-state.js"; import type { createDeliveryWorkCoordinator } from "../../delivery-work-coordinator.js"; import type { EndpointRow, DbTransaction, DbOrTransaction, RuntimeContext, LifecycleRuntimeFence } from "../runtime-types.js"; import { githubRepositoryInventoryItemFromPayload } from "./resources.js"; import { dueChatAction, PROVIDER_EFFECT_STALE_MS } from "../../chat-action-work.js"; import { notifyChatDeliveryWork, notifyChatActionWork } from "../../chat-work-notifications.js"; import { chatCredentialMutationLease } from "../../chat-credential-mutation-lease.js"; import { githubBodyMentionsBot, } from "../../chat-github-events.js"; import { githubAppJwt } from "../../chat-github-client.js"; import { createHash, createHmac, timingSafeEqual, } from "node:crypto"; import { and, asc, eq, isNull, lte, or, sql, } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { chatActions, chatDeliveries, chatEndpointResources, chatEndpoints, chatSdkState, toolConnections, } from "@paperclipai/db"; import type { ChatEndpointSetupState, ToolCredentialSecretRef, } from "@paperclipai/shared"; import { conflict, } from "../../../errors.js"; import { logger } from "../../../middleware/logger.js"; import { type ChatSdkProvider, } from "../../chat-sdk-runtime.js"; import type { ChatSdkStatePersistence, } from "../../chat-sdk-state.js"; import { listGitHubAppWebhookDeliveries, getGitHubAppWebhookDelivery, readGitHubAppWebhookConfig, requestGitHubAppWebhookRedelivery, getGitHubRecoveryComment, type GitHubAppWebhookDelivery, type GitHubAppWebhookDeliveryDetail, } from "../../chat-github-webhook-config.js"; import type { EndpointRecord } from "../runtime-types.js"; export type GitHubIngressStageResult = | { kind: "invalid_signature" } | { kind: "ignored" } | { actionId: string; kind: "staged" }; export type GitHubRecoveryReceipt = GitHubRecoveryWindow & { original: GitHubAppWebhookDeliveryDetail; }; export type GitHubWebhookIngressPayload = { version: 1; body: string; bodySha256: string; contentType: string; deliveryId: string; eventType: string; runtimeGeneration: number; credentialFingerprint: string; }; export type RedactedGitHubWebhookIngressPayload = Omit< GitHubWebhookIngressPayload, "body" | "contentType" > & { redacted: true; }; export function githubWebhookIngressPayload( payload: Record, ): GitHubWebhookIngressPayload | null { if ( payload.version !== 1 || typeof payload.body !== "string" || typeof payload.bodySha256 !== "string" || !/^[a-f0-9]{64}$/.test(payload.bodySha256) || typeof payload.contentType !== "string" || typeof payload.deliveryId !== "string" || !payload.deliveryId || typeof payload.eventType !== "string" || !payload.eventType || typeof payload.runtimeGeneration !== "number" || !Number.isSafeInteger(payload.runtimeGeneration) || typeof payload.credentialFingerprint !== "string" ) { return null; } if ( createHash("sha256").update(payload.body).digest("hex") !== payload.bodySha256 ) { return null; } return payload as GitHubWebhookIngressPayload; } export function redactedGitHubWebhookIngressPayload( payload: GitHubWebhookIngressPayload, ): RedactedGitHubWebhookIngressPayload { return { version: 1, bodySha256: payload.bodySha256, deliveryId: payload.deliveryId, eventType: payload.eventType, runtimeGeneration: payload.runtimeGeneration, credentialFingerprint: payload.credentialFingerprint, redacted: true, }; } type GitHubWebhookIngressOptions = { endpointRecord: (endpointId: string) => Promise; runtimeContextForRecord: (record: EndpointRecord) => RuntimeContext; resolveCredentialRefs: (endpoint: EndpointRow, refs: ToolCredentialSecretRef[], database?: DbOrTransaction) => Promise>; runtimeGeneration: (setup: ChatEndpointSetupState) => number; credentialFingerprint: (refs: ToolCredentialSecretRef[]) => string; getWebhookPublicBaseUrl: () => string | null; redactError: (error: unknown) => string; handleWebhook: (publicId: string, provider: ChatSdkProvider, request: Request, context?: { githubIngressActionId: string; githubIngressFence: LifecycleRuntimeFence; superseded?: boolean }) => Promise; runtimeCallbackEndpoint: (tx: DbTransaction, endpointId: string, context: LifecycleRuntimeFence, statuses: EndpointRow["status"][]) => Promise; fetchImpl: typeof fetch; persistence: ChatSdkStatePersistence; withCredentialMutationLease: ReturnType; isShuttingDown: () => boolean; githubWebhookReplayBarrier?: () => Promise; supportedWebhookEvents: ReadonlySet; }; export function createGitHubWebhookIngress(db: Db, options: GitHubWebhookIngressOptions) { const { endpointRecord, runtimeContextForRecord, resolveCredentialRefs, runtimeGeneration, credentialFingerprint, getWebhookPublicBaseUrl, redactError, handleWebhook, runtimeCallbackEndpoint, fetchImpl, persistence, withCredentialMutationLease, isShuttingDown, supportedWebhookEvents: SUPPORTED_GITHUB_WEBHOOK_EVENTS } = options; function githubSignatureIsValid( body: string, signature: string | null, webhookSecret: string | undefined, ): boolean { if (!signature || !webhookSecret) return false; const expected = `sha256=${createHmac("sha256", webhookSecret).update(body).digest("hex")}`; try { return timingSafeEqual(Buffer.from(signature), Buffer.from(expected)); } catch { return false; } } function githubWebhookContainsUserContent(eventType: string): boolean { return ( eventType === "issue_comment" || eventType === "issues" || eventType === "pull_request" || eventType === "pull_request_review" || eventType === "pull_request_review_comment" ); } function githubWebhookAddressesBot( payload: unknown, botUsername: string | null, ): boolean { if (!botUsername || !payload || typeof payload !== "object") return false; const comment = (payload as { comment?: unknown }).comment; if (!comment || typeof comment !== "object") return false; const body = (comment as { body?: unknown }).body; if (typeof body !== "string") return false; return githubBodyMentionsBot(body, botUsername); } function githubIngressAllowedInEndpointState( status: EndpointRow["status"], eventType: string, ): boolean { if (status === "verifying" || status === "active") return true; if (eventType === "ping" && (status === "draft" || status === "attention")) return true; if ( status === "attention" && [ "github_app_authorization", "installation", "installation_repositories", ].includes(eventType) ) return true; return status === "revoked" && eventType === "installation"; } async function stageGitHubWebhookIngress( endpoint: EndpointRow, request: Request, ): Promise { const body = await request.text(); const signature = request.headers.get("x-hub-signature-256"); const eventType = request.headers.get("x-github-event")?.trim() ?? ""; const providerDeliveryId = request.headers.get("x-github-delivery")?.trim() || `body:${createHash("sha256").update(body).digest("hex")}`; if (!SUPPORTED_GITHUB_WEBHOOK_EVENTS.has(eventType)) return { kind: "ignored" }; let parsedPayload: unknown = null; try { parsedPayload = JSON.parse(body) as unknown; } catch { // A valid signature is still required before a malformed body may be // retained. The native adapter will settle the authenticated envelope. } // Resolve from a captured ref set, then prove that exact ref set and // generation are still authoritative in the insertion transaction. This // keeps webhook authentication outside the potentially long credential // mutation lease without letting an obsolete secret stage new work. for (let attempt = 0; attempt < 3; attempt += 1) { const record = await endpointRecord(endpoint.id); if (!record || record.endpoint.provider !== "github") return { kind: "ignored" }; if ( !githubIngressAllowedInEndpointState(record.endpoint.status, eventType) ) { return { kind: "ignored" }; } const context = runtimeContextForRecord(record); let credentials: Record; try { credentials = await resolveCredentialRefs( record.endpoint, record.credentialSecretRefs, ); } catch (error) { const latest = await endpointRecord(endpoint.id); if ( latest && (runtimeGeneration(latest.endpoint.setup) !== context.generation || credentialFingerprint(latest.credentialSecretRefs) !== context.credentialFingerprint) ) { continue; } throw error; } if (!githubSignatureIsValid(body, signature, credentials.webhookSecret)) { const latest = await endpointRecord(endpoint.id); if ( latest && (runtimeGeneration(latest.endpoint.setup) !== context.generation || credentialFingerprint(latest.credentialSecretRefs) !== context.credentialFingerprint) ) { continue; } return { kind: "invalid_signature" }; } const incomingInstallationId = parsedPayload && typeof parsedPayload === "object" && !Array.isArray(parsedPayload) ? (parsedPayload as { installation?: { id?: unknown } }).installation ?.id : undefined; if ( incomingInstallationId !== undefined && String(incomingInstallationId) !== credentials.installationId && eventType !== "installation" ) { return { kind: "ignored" }; } const bodySha256 = createHash("sha256").update(body).digest("hex"); const staged = await db.transaction(async (tx) => { const currentEndpoint = await tx .select() .from(chatEndpoints) .where(eq(chatEndpoints.id, endpoint.id)) .for("update") .then((rows) => rows[0] ?? null); const currentConnection = await tx .select({ refs: toolConnections.credentialSecretRefs }) .from(toolConnections) .where( and( eq(toolConnections.companyId, record.endpoint.companyId), eq(toolConnections.id, record.endpoint.connectionId), ), ) .for("update") .then((rows) => rows[0] ?? null); if (!currentEndpoint || !currentConnection) return "retry" as const; if ( runtimeGeneration(currentEndpoint.setup) !== context.generation || credentialFingerprint(currentConnection.refs) !== context.credentialFingerprint ) { return "retry" as const; } if ( !githubIngressAllowedInEndpointState( currentEndpoint.status, eventType, ) ) { return "ignored" as const; } // Recovery history is denial-only provenance. A genuine callback // queued by GitHub before pause/reconnect must not be authenticated // into a later generation, even if the provider signs with its current // secret. Include archived endpoints so reusing an App cannot move a // delayed callback into another endpoint/company. const recoveryReceipts = credentials.appId ? await tx .select({ endpointId: chatActions.endpointId, payload: chatActions.payload, status: chatActions.status, }) .from(chatActions) .innerJoin( chatEndpoints, eq(chatEndpoints.id, chatActions.endpointId), ) .where( and( eq(chatEndpoints.provider, "github"), eq(chatEndpoints.botExternalId, credentials.appId), eq(chatActions.kind, "github_webhook_recovery"), eq( chatActions.providerActionId, `github_webhook_recovery:${providerDeliveryId}`, ), ), ) : []; for (const receipt of recoveryReceipts) { const fence = githubRecoveryWindow(receipt.payload); const original = (receipt.payload as GitHubRecoveryReceipt).original; const incoming = parsedPayload as { action?: unknown; installation?: { id?: unknown }; repository?: { id?: unknown; full_name?: unknown }; issue?: { id?: unknown; number?: unknown }; pull_request?: { id?: unknown; number?: unknown }; sender?: { id?: unknown }; comment?: { id?: unknown; body?: unknown; user?: { id?: unknown }; created_at?: unknown; updated_at?: unknown; }; } | null; const comment = incoming?.comment; if ( !fence || receipt.status === "cancelled" || receipt.endpointId !== currentEndpoint.id || fence.generation !== context.generation || fence.credentialFingerprint !== context.credentialFingerprint || fence.webhookUrl !== `${getWebhookPublicBaseUrl()}/api/chat-webhooks/${currentEndpoint.publicId}/github` || !original?.payload?.comment || original.event !== eventType || incoming?.action !== "created" || String(incoming?.installation?.id) !== original.payload.installationId || String(incoming?.repository?.id) !== original.payload.repositoryId || incoming?.repository?.full_name !== original.payload.repositoryFullName || String(incoming?.sender?.id) !== original.payload.senderId || (eventType === "issue_comment" ? String(incoming?.issue?.id) !== original.payload.issueId || String(incoming?.issue?.number) !== original.payload.issueNumber : String(incoming?.pull_request?.id) !== original.payload.pullRequestId || String(incoming?.pull_request?.number) !== original.payload.pullRequestNumber) || String(comment?.id) !== original.payload.comment.id || String(comment?.user?.id) !== original.payload.comment.userId || comment?.created_at !== original.payload.comment.createdAt || comment?.updated_at !== original.payload.comment.updatedAt || typeof comment?.body !== "string" || createHash("sha256").update(comment.body).digest("hex") !== original.payload.comment.bodySha256 ) { return "ignored" as const; } } // Signed App webhooks can cover every installed repository. Do not // retain comment bodies for destinations the operator did not enable. if (githubWebhookContainsUserContent(eventType)) { const item = githubRepositoryInventoryItemFromPayload(parsedPayload); if (!item) return "ignored" as const; const providerRepositoryId = typeof item.metadata?.providerRepositoryId === "string" ? item.metadata.providerRepositoryId : null; const matchingResource = await tx .select({ enabled: chatEndpointResources.enabled, id: chatEndpointResources.id, }) .from(chatEndpointResources) .where( and( eq(chatEndpointResources.companyId, currentEndpoint.companyId), eq(chatEndpointResources.endpointId, currentEndpoint.id), eq(chatEndpointResources.type, "repository"), eq(chatEndpointResources.availability, "available"), or( eq( chatEndpointResources.providerResourceId, item.providerResourceId, ), providerRepositoryId ? sql`${chatEndpointResources.metadata}->>'providerRepositoryId' = ${providerRepositoryId}` : undefined, ), ), ) .limit(1) .then((rows) => rows[0] ?? null); if (!matchingResource) return "ignored" as const; if (!matchingResource.enabled) { const action = parsedPayload && typeof parsedPayload === "object" && !Array.isArray(parsedPayload) ? (parsedPayload as { action?: unknown }).action : null; const ignoreDisabledResource = async () => { // Record only an authenticated, content-free admission decision. // Do not make operators infer a working callback from silence, // or retain the comment/actor/thread from a disabled repository. const providerEventId = `github:filtered_ingress:${createHash( "sha256", ) .update(providerDeliveryId) .digest("hex")}`; const eventKind = action === "edited" ? "message_updated" : action === "deleted" ? "message_deleted" : "message"; await notifyChatDeliveryWork(tx); await tx .insert(chatDeliveries) .values({ companyId: currentEndpoint.companyId, endpointId: currentEndpoint.id, providerEventId, deduplicationKey: providerEventId, eventKind, state: "filtered", processedAt: new Date(), redactedError: "Destination is not enabled in Paperclip", normalizedEvent: { providerEventId, kind: eventKind, filtering: { contentRetained: false, reason: "destination_not_enabled", resourceId: matchingResource.id, }, }, }) .onConflictDoNothing(); return "ignored" as const; }; if ( currentEndpoint.status !== "verifying" || action !== "created" || !githubWebhookAddressesBot( parsedPayload, currentEndpoint.botUsername, ) ) { return ignoreDisabledResource(); } const enabledResourceCount = await tx .select({ count: sql`count(*)::int` }) .from(chatEndpointResources) .where( and( eq(chatEndpointResources.endpointId, currentEndpoint.id), eq(chatEndpointResources.type, "repository"), eq(chatEndpointResources.enabled, true), eq(chatEndpointResources.availability, "available"), ), ) .then((rows) => rows[0]?.count ?? 0); if (enabledResourceCount !== 0) return ignoreDisabledResource(); } } const payload: GitHubWebhookIngressPayload = { version: 1, body, bodySha256, contentType: request.headers.get("content-type") ?? "application/json", deliveryId: providerDeliveryId, eventType, runtimeGeneration: context.generation, credentialFingerprint: context.credentialFingerprint, }; await notifyChatActionWork(tx, "github_webhook_ingress"); const inserted = await tx .insert(chatActions) .values({ companyId: currentEndpoint.companyId, endpointId: currentEndpoint.id, kind: "github_webhook_ingress", providerActionId: `github_webhook_ingress:${providerDeliveryId}`, payload, status: "received", }) .onConflictDoNothing() .returning({ id: chatActions.id }); if (inserted[0]?.id) { return { actionId: inserted[0].id, kind: "staged" } as const; } // An automatic callback is never authorization to reset the local // worker's terminal retry budget. The original may have arrived in the // interval between our remote scan and GitHub's asynchronous callback. if (recoveryReceipts.length > 0) return "ignored" as const; // GitHub preserves X-GitHub-Delivery when an operator uses Recent // Deliveries to retry a failed callback. Re-arm only the exact same, // currently authenticated body after a terminal processing failure. // Successful, cancelled, retryable, and in-flight receipts remain // immutable deduplication fences. await notifyChatActionWork(tx, "github_webhook_ingress"); const [rearmed] = await tx .update(chatActions) .set({ payload, // Keep the lifetime attempt counter monotonic. Besides bounding // repeated manual redeliveries, this prevents a late settlement // from an older processing claim from matching a fresh claim // after re-arm (an attempts=1 ABA). result: sql`jsonb_build_object( 'attempts', coalesce((${chatActions.result}->>'attempts')::int, 0) )`, status: "received", updatedAt: new Date(), }) .where( and( eq(chatActions.endpointId, currentEndpoint.id), eq( chatActions.providerActionId, `github_webhook_ingress:${providerDeliveryId}`, ), eq(chatActions.kind, "github_webhook_ingress"), eq(chatActions.status, "failed"), sql`coalesce(${chatActions.result}->>'retryable', 'false') = 'false'`, sql`${chatActions.payload}->>'bodySha256' = ${bodySha256}`, sql`${chatActions.payload}->>'eventType' = ${eventType}`, ), ) .returning({ id: chatActions.id }); const actionId = rearmed?.id ?? (await tx .select({ id: chatActions.id }) .from(chatActions) .where( and( eq(chatActions.endpointId, currentEndpoint.id), eq( chatActions.providerActionId, `github_webhook_ingress:${providerDeliveryId}`, ), ), ) .then((rows) => rows[0]?.id ?? null)); return actionId ? ({ actionId, kind: "staged" } as const) : ("retry" as const); }); if (staged === "retry") continue; if (staged === "ignored") return { kind: "ignored" }; return staged; } throw conflict("GitHub credentials changed during webhook authentication", { code: "chat_endpoint_credentials_changed", }); } function githubIngressAcceptedResponse() { return new Response("accepted", { status: 202 }); } async function settleGitHubWebhookIngress( actionId: string, input: { attempts: number; payload: GitHubWebhookIngressPayload; response?: Response; error?: unknown; cancelledCode?: string; }, ): Promise { const terminalResponse = input.response !== undefined && input.response.status < 500; const terminalFailure = input.attempts >= 5; const cancelled = Boolean(input.cancelledCode); const processed = Boolean(input.response?.ok); const terminal = processed || terminalResponse || terminalFailure || cancelled; const retryAt = new Date( Date.now() + Math.min(60_000, 1_000 * 2 ** Math.max(0, input.attempts - 1)), ); await db.transaction(async (tx) => { await notifyChatActionWork(tx, "github_webhook_ingress"); return tx .update(chatActions) .set({ status: cancelled ? "cancelled" : processed ? "processed" : "failed", payload: terminal ? redactedGitHubWebhookIngressPayload(input.payload) : input.payload, result: { attempts: input.attempts, ...(input.cancelledCode ? { code: input.cancelledCode } : {}), ...(input.response ? { httpStatus: input.response.status } : { error: redactError(input.error) }), retryable: !terminal, ...(!terminal ? { retryAt: retryAt.toISOString() } : {}), }, updatedAt: new Date(), }) .where( and( eq(chatActions.id, actionId), eq(chatActions.kind, "github_webhook_ingress"), eq(chatActions.status, "processing"), sql`(${chatActions.result}->>'attempts')::int = ${input.attempts}`, ), ); }); } async function processGitHubWebhookIngress( actionId: string, propagateProcessingFailure = false, ): Promise { const candidate = await db .select({ id: chatActions.id, status: chatActions.status, result: chatActions.result, updatedAt: chatActions.updatedAt, }) .from(chatActions) .where( and( eq(chatActions.id, actionId), eq(chatActions.kind, "github_webhook_ingress"), ), ) .then((rows) => rows[0] ?? null); if (!candidate) return null; if (["processed", "cancelled"].includes(candidate.status)) return githubIngressAcceptedResponse(); if (candidate.status === "failed" && candidate.result?.retryable !== true) return githubIngressAcceptedResponse(); const now = new Date(); const staleBefore = new Date(now.getTime() - PROVIDER_EFFECT_STALE_MS); if (candidate.status === "processing" && candidate.updatedAt > staleBefore) { return null; } if (candidate.status === "failed") { const retryAt = typeof candidate.result?.retryAt === "string" ? new Date(candidate.result.retryAt) : null; if (retryAt && retryAt > now) return null; } else if (candidate.status !== "received" && candidate.status !== "processing") { return null; } const attempts = (typeof candidate.result?.attempts === "number" ? candidate.result.attempts : 0) + 1; const previousAttempts = attempts - 1; const claimGuard = candidate.status === "processing" ? and( eq(chatActions.status, "processing"), lte(chatActions.updatedAt, staleBefore), sql`coalesce((${chatActions.result}->>'attempts')::int, 0) = ${previousAttempts}`, ) : candidate.status === "failed" ? and( eq(chatActions.status, "failed"), sql`coalesce((${chatActions.result}->>'attempts')::int, 0) = ${previousAttempts}`, ) : eq(chatActions.status, "received"); const [claimed] = await db.transaction(async (tx) => { await notifyChatActionWork(tx, "github_webhook_ingress"); return tx .update(chatActions) .set({ status: "processing", result: { attempts }, updatedAt: now, }) .where( and( eq(chatActions.id, candidate.id), eq(chatActions.kind, "github_webhook_ingress"), claimGuard, ), ) .returning({ id: chatActions.id, endpointId: chatActions.endpointId, payload: chatActions.payload, }); }); if (!claimed) return null; const action = claimed; // Only the successful claimant needs the replay body. Validate the row // returned by the claim so a competing retry cannot replace it underneath us. const payload = githubWebhookIngressPayload(action.payload); if (!payload) { await db.transaction(async (tx) => { await notifyChatActionWork(tx, "github_webhook_ingress"); return tx .update(chatActions) .set({ status: "failed", payload: {}, result: { code: "github_webhook_ingress_payload_invalid", retryable: false, }, updatedAt: new Date(), }) .where( and( eq(chatActions.id, action.id), eq(chatActions.kind, "github_webhook_ingress"), eq(chatActions.status, "processing"), sql`(${chatActions.result}->>'attempts')::int = ${attempts}`, ), ); }); return githubIngressAcceptedResponse(); } const record = await endpointRecord(action.endpointId); if ( !record || record.endpoint.provider !== "github" || !githubIngressAllowedInEndpointState( record.endpoint.status, payload.eventType, ) || runtimeGeneration(record.endpoint.setup) !== payload.runtimeGeneration || credentialFingerprint(record.credentialSecretRefs) !== payload.credentialFingerprint ) { await settleGitHubWebhookIngress(action.id, { attempts, payload, cancelledCode: "github_webhook_ingress_runtime_superseded", }); return githubIngressAcceptedResponse(); } try { const credentials = await resolveCredentialRefs( record.endpoint, record.credentialSecretRefs, ); const recovery = await db .select({ payload: chatActions.payload, status: chatActions.status }) .from(chatActions) .where( and( eq(chatActions.endpointId, record.endpoint.id), eq(chatActions.kind, "github_webhook_recovery"), eq( chatActions.providerActionId, `github_webhook_recovery:${payload.deliveryId}`, ), ), ) .then((rows) => rows[0] ?? null); if ( recovery && (recovery.status === "cancelled" || !(await githubRecoverySourceIsCurrent( (recovery.payload as GitHubRecoveryReceipt).original, credentials, ))) ) { await settleGitHubWebhookIngress(action.id, { attempts, payload, cancelledCode: "github_webhook_recovery_source_changed", }); return githubIngressAcceptedResponse(); } const signature = `sha256=${createHmac( "sha256", credentials.webhookSecret, ) .update(payload.body) .digest("hex")}`; await options.githubWebhookReplayBarrier?.(); const replayContext = { githubIngressActionId: action.id, githubIngressFence: { generation: payload.runtimeGeneration, credentialFingerprint: payload.credentialFingerprint, }, superseded: false, }; const response = await handleWebhook( record.endpoint.publicId, "github", new Request( `https://paperclip.internal/api/chat-webhooks/${record.endpoint.publicId}/github`, { method: "POST", headers: { "content-type": payload.contentType, "x-github-delivery": payload.deliveryId, "x-github-event": payload.eventType, "x-hub-signature-256": signature, }, body: payload.body, }, ), replayContext, ); if (replayContext.superseded) { await settleGitHubWebhookIngress(action.id, { attempts, payload, cancelledCode: "github_webhook_ingress_runtime_superseded", }); return githubIngressAcceptedResponse(); } if (response.status === 401) { const latest = await endpointRecord(action.endpointId); if ( !latest || runtimeGeneration(latest.endpoint.setup) !== payload.runtimeGeneration || credentialFingerprint(latest.credentialSecretRefs) !== payload.credentialFingerprint ) { await settleGitHubWebhookIngress(action.id, { attempts, payload, cancelledCode: "github_webhook_ingress_runtime_superseded", }); return githubIngressAcceptedResponse(); } } await settleGitHubWebhookIngress(action.id, { attempts, payload, response, }); return response.status >= 500 && !propagateProcessingFailure ? githubIngressAcceptedResponse() : response; } catch (error) { await settleGitHubWebhookIngress(action.id, { attempts, payload, error, }); logger.warn( { actionId: action.id, endpointId: action.endpointId, error: redactError(error), }, "durable GitHub webhook processing deferred", ); if (propagateProcessingFailure) throw error; return githubIngressAcceptedResponse(); } } async function githubRecoveryResource( database: DbOrTransaction, endpoint: EndpointRow, repositoryId: string, repositoryFullName?: string | null, ) { return database .select({ id: chatEndpointResources.id }) .from(chatEndpointResources) .where( and( eq(chatEndpointResources.companyId, endpoint.companyId), eq(chatEndpointResources.endpointId, endpoint.id), eq(chatEndpointResources.type, "repository"), eq(chatEndpointResources.enabled, true), eq(chatEndpointResources.availability, "available"), sql`${chatEndpointResources.metadata}->>'providerRepositoryId' = ${repositoryId}`, repositoryFullName ? eq( chatEndpointResources.providerResourceId, repositoryFullName.toLowerCase(), ) : undefined, ), ) .limit(1) .then((rows) => rows[0] ?? null); } async function githubRecoveryAlreadyReceived( database: DbOrTransaction, endpoint: EndpointRow, guid: string, ) { const action = await database .select({ id: chatActions.id }) .from(chatActions) .where( and( eq(chatActions.companyId, endpoint.companyId), eq(chatActions.endpointId, endpoint.id), eq(chatActions.providerActionId, `github_webhook_ingress:${guid}`), ), ) .limit(1) .then((rows) => rows[0]); if (action) return true; // Disabled destinations intentionally retain no ingress body. Their // content-free filtered receipt must also remain an admission fence. return database .select({ id: chatDeliveries.id }) .from(chatDeliveries) .where( and( eq(chatDeliveries.companyId, endpoint.companyId), eq(chatDeliveries.endpointId, endpoint.id), eq( chatDeliveries.providerEventId, `github:filtered_ingress:${createHash("sha256").update(guid).digest("hex")}`, ), ), ) .limit(1) .then((rows) => rows.length > 0); } async function rememberGitHubRecoverySkip( endpoint: EndpointRow, context: LifecycleRuntimeFence, window: GitHubRecoveryWindow, original: GitHubAppWebhookDelivery, code: string, ) { await db.transaction(async (tx) => { const current = await runtimeCallbackEndpoint(tx, endpoint.id, context, [ "active", ]); if ( !current || !original.repositoryId || !(await githubRecoveryResource(tx, current, original.repositoryId)) ) return; // Only the deny tombstone is needed, not the rejected detail or source // contents. Remembering it lets the next bounded scan reach older valid // messages instead of repeatedly inspecting the same five bot callbacks. await tx .insert(chatActions) .values({ companyId: current.companyId, endpointId: current.id, kind: "github_webhook_recovery", providerActionId: `github_webhook_recovery:${original.guid}`, payload: { ...window, original: { id: original.id, guid: original.guid }, }, status: "cancelled", result: { attempts: 0, code }, }) .onConflictDoUpdate({ target: [chatActions.endpointId, chatActions.providerActionId], set: { status: "cancelled", updatedAt: new Date(), result: sql`coalesce(${chatActions.result}, '{}'::jsonb) || jsonb_build_object('code', ${code}::text)`, }, setWhere: and( eq(chatActions.kind, "github_webhook_recovery"), sql`(${chatActions.payload}->>'generation')::int = ${context.generation}`, sql`${chatActions.payload}->>'credentialFingerprint' = ${context.credentialFingerprint}`, ), }); }); } async function githubRecoverySourceIsCurrent( detail: GitHubAppWebhookDeliveryDetail, credentials: Record, sourceFetch: typeof fetch = fetchImpl, ): Promise { const comment = detail?.payload?.comment; if ( !comment || !detail.payload.repositoryFullName || !["issue_comment", "pull_request_review_comment"].includes(detail.event) ) return false; const current = await getGitHubRecoveryComment({ fetch: sourceFetch, appToken: githubAppJwt(credentials.appId, credentials.privateKey), installationId: credentials.installationId, repositoryFullName: detail.payload.repositoryFullName, event: detail.event as "issue_comment" | "pull_request_review_comment", commentId: comment.id, }).catch((error) => { if (error?.statusCode === 404 || error?.statusCode === 410) return null; throw error; }); if (!current) return false; return ( current.id === comment.id && current.userId === comment.userId && current.userType === "User" && current.bodySha256 === comment.bodySha256 && current.createdAt === comment.createdAt && current.updatedAt === comment.updatedAt && (detail.event === "issue_comment" ? current.issueNumber === detail.payload.issueNumber : current.pullRequestNumber === detail.payload.pullRequestNumber) ); } /** * GitHub does not retry callbacks that fail before our durable inbox. Ask * GitHub to redeliver a bounded recent created-comment envelope, never turn * delivery-history JSON into a locally signed or directly admitted request. * A missing/changed epoch starts at now; historical messages are not adopted * into a new credential/generation merely because the App can still see them. */ async function processFailedGitHubWebhookDeliveries( limit = 5, onlyEndpointId?: string, ) { if (isShuttingDown() || !getWebhookPublicBaseUrl()?.startsWith("https://")) return 0; const now = new Date(); const rows = await db .select({ endpoint: chatEndpoints }) .from(chatEndpoints) .innerJoin( toolConnections, and( eq(toolConnections.id, chatEndpoints.connectionId), eq(toolConnections.companyId, chatEndpoints.companyId), ), ) .leftJoin( chatSdkState, and( eq(chatSdkState.endpointId, chatEndpoints.id), eq(chatSdkState.stateKey, GITHUB_RECOVERY_STATE_KEY), ), ) .where( and( eq(chatEndpoints.provider, "github"), eq(chatEndpoints.status, "active"), // Cloud Apps target the durable Cloud relay, not the direct tenant // callback checked by this recovery worker. The Cloud inbox owns // transport retries for those registrations. sql`nullif(${chatEndpoints.setup}#>>'{github,cloudRegistrationId}', '') is null`, eq(toolConnections.enabled, true), eq(toolConnections.status, "active"), onlyEndpointId ? eq(chatEndpoints.id, onlyEndpointId) : undefined, or( isNull(chatSdkState.id), // Backoff belongs to one runtime epoch. Reconnect or callback // changes must establish a fresh floor immediately, even when // the superseded epoch was rate limited for several hours. sql`${chatSdkState.value}->>'generation' is distinct from coalesce(${chatEndpoints.setup}->>'runtimeGeneration', '0')`, sql`${chatSdkState.value}->>'appId' is distinct from ${chatEndpoints.botExternalId}`, sql`${chatSdkState.value}->>'webhookUrl' is distinct from (${getWebhookPublicBaseUrl()} || '/api/chat-webhooks/' || ${chatEndpoints.publicId} || '/github')`, sql`(${chatSdkState.value}->>'nextScanAt')::timestamptz <= ${now.toISOString()}::timestamptz`, ), ), ) .orderBy(asc(chatSdkState.updatedAt), asc(chatEndpoints.id)) .limit(Math.min(5, Math.max(1, limit))); for (const { endpoint } of rows) await recoverGitHubWebhookEndpoint(endpoint.id); return rows.length; } async function saveRecoveryWindow(input: ChatSdkStateCompareAndSetInput) { return db.transaction(async tx => { await notifyGitHubWebhookRecovery(tx, { kind: "endpoint", id: input.endpointId }); return createChatSdkStatePersistence(tx as unknown as Db).compareAndSet(input); }); } async function recoverGitHubWebhookEndpoint( endpointId: string, canRun: () => boolean = () => true, signal?: AbortSignal, ) { const admitted = () => !isShuttingDown() && canRun() && !signal?.aborted; if (!admitted() || !getWebhookPublicBaseUrl()?.startsWith("https://")) return; const assertAdmitted = () => { if (!admitted()) throw signal?.reason ?? new Error("GitHub webhook recovery stopped"); }; const recoveryFetch: typeof fetch = (input, init) => { assertAdmitted(); return fetchImpl(input, init); }; const now = new Date(); const record = await endpointRecord(endpointId); if (!record || !record.endpoint.botExternalId) return; const endpoint = record.endpoint; const context = runtimeContextForRecord(record); const webhookUrl = `${getWebhookPublicBaseUrl()}/api/chat-webhooks/${endpoint.publicId}/github`; const scope = { companyId: endpoint.companyId, endpointId: endpoint.id }; const stored = await persistence.read(scope, GITHUB_RECOVERY_STATE_KEY); const previous = githubRecoveryWindow(stored?.value); const epochMatches = previous && previous.generation === context.generation && previous.credentialFingerprint === context.credentialFingerprint && previous.webhookUrl === webhookUrl && previous.appId === record.endpoint.botExternalId; if (epochMatches && Date.parse(previous.nextScanAt) > now.getTime()) return; const window: GitHubRecoveryWindow = { generation: context.generation, credentialFingerprint: context.credentialFingerprint, appId: record.endpoint.botExternalId, webhookUrl, floor: epochMatches ? previous.floor : now.toISOString(), nextScanAt: new Date( now.getTime() + GITHUB_RECOVERY_SCAN_LEASE_MS, ).toISOString(), }; if ( !(await saveRecoveryWindow({ ...scope, key: GITHUB_RECOVERY_STATE_KEY, expectedVersion: stored?.version ?? null, value: window, expiresAt: null, })) ) return; const claimedVersion = (stored?.version ?? 0) + 1; let nextDelay = GITHUB_RECOVERY_INTERVAL_MS; let outcome = "scanned"; let requested = 0; let inspected = 0; try { if (!epochMatches) { outcome = "epoch_initialized"; return; } const credentials = await resolveCredentialRefs( record.endpoint, record.credentialSecretRefs, ); if (credentials.appId !== window.appId || !credentials.installationId) return; const appToken = githubAppJwt( credentials.appId, credentials.privateKey, ); const config = await readGitHubAppWebhookConfig({ fetch: recoveryFetch, appToken, }); if ( config.url !== webhookUrl || config.contentType !== "json" || config.insecureSsl !== "0" ) { outcome = "callback_mismatch"; nextDelay = 5 * 60_000; return; } const attempts: GitHubAppWebhookDelivery[] = []; const floor = Math.max( Date.parse(window.floor), now.getTime() - GITHUB_RECOVERY_WINDOW_MS, ); let cursor: string | undefined; let truncated = false; let previousTimestamp = Number.POSITIVE_INFINITY; for (let page = 0; page < 3; page += 1) { const result = await listGitHubAppWebhookDeliveries({ fetch: recoveryFetch, appToken, cursor, }); for (const attempt of result.deliveries) { const timestamp = Date.parse(attempt.deliveredAt); if (timestamp > previousTimestamp) throw new Error("GitHub recovery history ordering changed"); previousTimestamp = timestamp; if (timestamp >= floor) attempts.push(attempt); } if (previousTimestamp < floor) break; if (!result.nextCursor) break; if (result.nextCursor === cursor) throw new Error("GitHub recovery pagination did not advance"); cursor = result.nextCursor; truncated = page === 2; } if (truncated) { outcome = "history_limit_reached"; nextDelay = 5 * 60_000; return; } const groups = new Map(); for (const attempt of attempts) { const group = groups.get(attempt.guid) ?? []; group.push(attempt); groups.set(attempt.guid, group); } for (const [guid, group] of groups) { if ( inspected >= 5 || isShuttingDown() || Date.now() >= Date.parse(window.nextScanAt) ) break; if ( group.some( (item) => item.statusCode !== null && item.statusCode >= 200 && item.statusCode < 400, ) ) continue; const originals = group.filter((item) => !item.redelivery); if (originals.length !== 1) continue; const original = originals[0]!; if ( Date.parse(original.deliveredAt) < floor || Date.parse(original.deliveredAt) > now.getTime() - 10_000 || original.statusCode === null || original.statusCode < 500 || original.action !== "created" || !["issue_comment", "pull_request_review_comment"].includes( original.event, ) || original.installationId !== credentials.installationId || !original.repositoryId || !(await githubRecoveryResource(db, endpoint, original.repositoryId)) ) continue; const recoveryId = `github_webhook_recovery:${guid}`; if (await githubRecoveryAlreadyReceived(db, endpoint, guid)) continue; const receipt = await db .select() .from(chatActions) .where( and( eq(chatActions.endpointId, endpoint.id), eq(chatActions.providerActionId, recoveryId), ), ) .then((items) => items[0] ?? null); const count = typeof receipt?.result?.attempts === "number" ? receipt.result.attempts : 0; const receiptWindow = githubRecoveryWindow(receipt?.payload); const latest = [...group].sort( (left, right) => Date.parse(right.deliveredAt) - Date.parse(left.deliveredAt), )[0]!; if ( latest.statusCode === null || latest.statusCode < 500 || latest.throttledAt !== null || Date.parse(latest.deliveredAt) > now.getTime() - 10_000 ) continue; if ( receipt && (receipt.status === "cancelled" || !receiptWindow || receiptWindow.generation !== window.generation || receiptWindow.credentialFingerprint !== window.credentialFingerprint || receiptWindow.webhookUrl !== window.webhookUrl || count >= 3 || receipt.result?.latestAttemptId === latest.id || !latest.redelivery || latest.statusCode === null || latest.statusCode < 500 || Date.parse(latest.deliveredAt) < Date.parse(String(receipt.result?.requestedAt)) || Date.parse(String(receipt.result?.retryAt)) > now.getTime()) ) continue; inspected += 1; const detail = await getGitHubAppWebhookDelivery({ fetch: recoveryFetch, appToken, deliveryId: original.id, }); const comment = detail.payload.comment; if ( detail.id !== original.id || detail.guid !== guid || detail.redelivery || detail.url !== webhookUrl || detail.event !== original.event || detail.deliveredAt !== original.deliveredAt || detail.statusCode !== original.statusCode || detail.installationId !== original.installationId || detail.repositoryId !== original.repositoryId || detail.action !== "created" || detail.payload.action !== "created" || detail.payload.installationId !== credentials.installationId || detail.payload.repositoryId !== original.repositoryId || !comment || comment.userType !== "User" || detail.payload.senderId !== comment.userId || Date.parse(comment.createdAt) < floor || comment.createdAt !== comment.updatedAt || !(await githubRecoveryResource( db, endpoint, original.repositoryId, detail.payload.repositoryFullName, )) ) { await rememberGitHubRecoverySkip( endpoint, context, window, original, "not_eligible_for_recovery", ); continue; } if (!(await githubRecoverySourceIsCurrent(detail, credentials, recoveryFetch))) { await rememberGitHubRecoverySkip( endpoint, context, window, original, "source_changed_or_unavailable", ); continue; } await withCredentialMutationLease(endpoint, async (lease) => { await lease.assertOwned(); const currentConfig = await readGitHubAppWebhookConfig({ fetch: recoveryFetch, appToken, }); if ( currentConfig.url !== webhookUrl || currentConfig.contentType !== "json" || currentConfig.insecureSsl !== "0" ) return; const requestedAt = new Date().toISOString(); const claimed = await db.transaction(async (tx) => { await lease.assertOwned(tx); const current = await runtimeCallbackEndpoint( tx, endpoint.id, context, ["active"], ); if ( !current || !(await githubRecoveryResource( tx, current, original.repositoryId!, detail.payload.repositoryFullName, )) ) return null; const connection = await tx .select({ id: toolConnections.id }) .from(toolConnections) .where( and( eq(toolConnections.id, current.connectionId), eq(toolConnections.companyId, current.companyId), eq(toolConnections.enabled, true), eq(toolConnections.status, "active"), ), ) .then((items) => items[0]); const scan = await tx .select({ id: chatSdkState.id }) .from(chatSdkState) .where( and( eq(chatSdkState.endpointId, endpoint.id), eq(chatSdkState.stateKey, GITHUB_RECOVERY_STATE_KEY), eq(chatSdkState.version, claimedVersion), ), ) .then((items) => items[0]); const local = await githubRecoveryAlreadyReceived( tx, current, guid, ); if ( !connection || !scan || local || isShuttingDown() || Date.now() >= Date.parse(window.nextScanAt) ) return null; const result = { attempts: count + 1, latestAttemptId: latest.id, requestedAt, retryAt: new Date( Date.now() + 60_000 * 5 ** count, ).toISOString(), }; const payload: GitHubRecoveryReceipt = { ...window, original: detail, }; const changes = receipt ? await tx .update(chatActions) .set({ status: "processing", result, updatedAt: new Date(), }) .where( and( eq(chatActions.id, receipt.id), eq(chatActions.updatedAt, receipt.updatedAt), ), ) .returning({ id: chatActions.id }) : await tx .insert(chatActions) .values({ companyId: endpoint.companyId, endpointId: endpoint.id, kind: "github_webhook_recovery", providerActionId: recoveryId, payload, status: "processing", result, }) .onConflictDoNothing() .returning({ id: chatActions.id }); return changes[0]?.id ?? null; }); if (!claimed) return; const ownsAttempt = and( eq(chatActions.id, claimed), eq(chatActions.status, "processing"), sql`(${chatActions.result}->>'attempts')::int = ${count + 1}`, sql`${chatActions.result}->>'latestAttemptId' = ${latest.id}`, sql`${chatActions.result}->>'requestedAt' = ${requestedAt}`, ); let transportStarted = false; try { await lease.assertOwned(); await requestGitHubAppWebhookRedelivery({ fetch: (input, init) => { assertAdmitted(); transportStarted = true; requested += 1; return fetchImpl(input, init); }, appToken, deliveryId: original.id, }); await db .update(chatActions) .set({ status: "processed", updatedAt: new Date() }) .where(ownsAttempt); } catch (error) { if (transportStarted) { await db .update(chatActions) .set({ status: "delivery_unknown", updatedAt: new Date() }) .where(ownsAttempt); } else { await db.transaction(async (tx) => { await notifyGitHubWebhookRecovery(tx, { kind: "endpoint", id: endpoint.id }); if (receipt) { await tx .update(chatActions) .set({ status: receipt.status, result: receipt.result, updatedAt: receipt.updatedAt }) .where(ownsAttempt); } else { await tx.delete(chatActions).where(ownsAttempt); } }); } throw error; } }); } } catch (error) { outcome = "scan_failed"; nextDelay = Math.max( 5 * 60_000, Math.min( 24 * 60 * 60_000, Number((error as { retryAfterMs?: unknown })?.retryAfterMs) || 0, ), ); logger.warn( { endpointId: endpoint.id, error: redactError(error) }, "GitHub webhook recovery deferred", ); } finally { await saveRecoveryWindow({ ...scope, key: GITHUB_RECOVERY_STATE_KEY, expectedVersion: claimedVersion, value: { ...window, nextScanAt: new Date(Date.now() + nextDelay).toISOString(), outcome, requested, }, expiresAt: null, }); } } const recoverySchedule = githubWebhookRecoverySchedule(db, { runtimeGeneration, credentialFingerprint, getWebhookPublicBaseUrl, recover: recoverGitHubWebhookEndpoint, }); function registerWebhookRecoveryWork(coordinator: ReturnType, canRun: () => boolean) { return registerGitHubWebhookRecoveryWork(coordinator, recoverySchedule, () => canRun() && !isShuttingDown()); } async function processPendingGitHubWebhookIngress( limit = 25, onlyActionId?: string, ) { const now = new Date(); const actions = await db .select({ id: chatActions.id }) .from(chatActions) .where(and( onlyActionId ? eq(chatActions.id, onlyActionId) : undefined, dueChatAction("github_webhook_ingress", now.getTime()), )) .orderBy(asc(chatActions.createdAt)) .limit(limit); for (const action of actions) await processGitHubWebhookIngress(action.id); return actions.length; } return { registerWebhookRecoveryWork, stageGitHubWebhookIngress, githubIngressAcceptedResponse, processGitHubWebhookIngress, processFailedGitHubWebhookDeliveries, processPendingGitHubWebhookIngress }; }