// Telegram plugin module implements client fetch behavior. import type { ApiClientOptions } from "grammy"; import { responseWithRelease } from "openclaw/plugin-sdk/fetch-runtime"; import { normalizeOptionalLowercaseString } from "openclaw/plugin-sdk/string-coerce-runtime"; import type { TelegramTransport } from "./fetch.js"; import { isTelegramMisdirectedRequestError, tagTelegramNetworkError, TelegramRequestNotStartedError, } from "./network-errors.js"; import { resolveTelegramRequestTimeoutMs } from "./request-timeouts.js"; type TelegramFetchInput = Parameters>[0]; type TelegramFetchInit = Parameters>[1]; type TelegramClientFetch = NonNullable; type TelegramCompatFetch = ( input: TelegramFetchInput, init?: TelegramFetchInit, ) => Promise; type TelegramAbortSignalLike = { aborted: boolean; reason?: unknown; addEventListener: (type: "abort", listener: () => void, options?: { once?: boolean }) => void; removeEventListener: (type: "abort", listener: () => void) => void; }; export function asTelegramClientFetch( fetchImpl: TelegramCompatFetch | typeof globalThis.fetch, ): TelegramClientFetch { return fetchImpl as unknown as TelegramClientFetch; } function asTelegramCompatFetch(fetchImpl: TelegramClientFetch): TelegramCompatFetch { return fetchImpl as unknown as TelegramCompatFetch; } function isTelegramAbortSignalLike(value: unknown): value is TelegramAbortSignalLike { return ( typeof value === "object" && value !== null && "aborted" in value && typeof (value as { aborted?: unknown }).aborted === "boolean" && typeof (value as { addEventListener?: unknown }).addEventListener === "function" && typeof (value as { removeEventListener?: unknown }).removeEventListener === "function" ); } function readRequestUrl(input: TelegramFetchInput): string | null { if (typeof input === "string") { return input; } if (input instanceof URL) { return input.toString(); } if (input instanceof Request) { return input.url; } return null; } function extractTelegramApiMethod(input: TelegramFetchInput): string | null { const url = readRequestUrl(input); if (!url) { return null; } try { const pathname = new URL(url).pathname; const segments = pathname.split("/").filter(Boolean); const method = segments.length > 0 ? (segments.at(-1) ?? null) : null; return normalizeOptionalLowercaseString(method) ?? null; } catch { return null; } } const TELEGRAM_TIMEOUT_FALLBACK_METHODS = new Set([ "deletemycommands", "deletewebhook", "getme", "sendchataction", "setmycommands", "setwebhook", ]); function shouldRetryTimedOutTelegramControlRequest(method: string | null): boolean { return method !== null && TELEGRAM_TIMEOUT_FALLBACK_METHODS.has(method); } export function resolveTelegramClientTimeoutSeconds(params: { value: unknown; minimum?: number; }): number | undefined { const { value, minimum } = params; if (typeof value !== "number" || !Number.isFinite(value)) { return undefined; } const configured = Math.max(1, Math.floor(value)); if (typeof minimum !== "number" || !Number.isFinite(minimum)) { return configured; } return Math.max(configured, Math.max(1, Math.floor(minimum))); } export function resolveTelegramClientTimeoutMinimumSeconds( values: readonly (number | undefined)[], ) { let minimum: number | undefined; for (const value of values) { if (typeof value !== "number" || !Number.isFinite(value)) { continue; } const normalized = Math.max(1, Math.ceil(value)); minimum = minimum === undefined ? normalized : Math.max(minimum, normalized); } return minimum; } export function resolveTelegramOutboundClientTimeoutFloorSeconds(timeoutSeconds: unknown) { const timeoutMs = resolveTelegramRequestTimeoutMs("sendmessage", timeoutSeconds); return timeoutMs === undefined ? undefined : timeoutMs / 1000; } export function createTelegramClientFetch(params: { fetchImpl?: TelegramClientFetch; timeoutSeconds?: unknown; shutdownSignal?: unknown; transport?: Pick; }): TelegramCompatFetch | undefined { if (!params.fetchImpl && !params.shutdownSignal) { return undefined; } const callFetch = asTelegramCompatFetch( params.fetchImpl ?? asTelegramClientFetch(globalThis.fetch), ); const wrappedFetch = async (input: TelegramFetchInput, init?: TelegramFetchInit) => { const method = extractTelegramApiMethod(input); const requestTimeoutMs = resolveTelegramRequestTimeoutMs(method, params.timeoutSeconds); const shutdownSignal = isTelegramAbortSignalLike(params.shutdownSignal) ? params.shutdownSignal : undefined; const requestSignal = isTelegramAbortSignalLike(init?.signal) ? init.signal : undefined; const canForceTransportFallback = (reason: string) => !shutdownSignal?.aborted && !requestSignal?.aborted && params.transport?.forceFallback?.(reason) === true; const runFetch = async (allowMisdirectedFallback = false): Promise => { const controller = new AbortController(); const abortWith = (signal: Pick) => controller.abort(signal.reason); const onShutdown = () => { if (shutdownSignal) { abortWith(shutdownSignal); } }; let requestTimeout: ReturnType | undefined; let onRequestAbort: (() => void) | undefined; let requestTimedOut = false; const timeoutError = requestTimeoutMs !== undefined ? new Error(`Telegram ${method} timed out after ${requestTimeoutMs}ms`) : undefined; if (shutdownSignal?.aborted) { abortWith(shutdownSignal); } else if (shutdownSignal) { shutdownSignal.addEventListener("abort", onShutdown, { once: true }); } if (requestSignal) { if (requestSignal.aborted) { abortWith(requestSignal); } else { onRequestAbort = () => abortWith(requestSignal); requestSignal.addEventListener("abort", onRequestAbort); } } if (requestTimeoutMs && timeoutError) { requestTimeout = setTimeout(() => { requestTimedOut = true; controller.abort(timeoutError); }, requestTimeoutMs); requestTimeout.unref?.(); } const releaseRequest = async () => { if (requestTimeout) { clearTimeout(requestTimeout); } shutdownSignal?.removeEventListener("abort", onShutdown); if (requestSignal && onRequestAbort) { requestSignal.removeEventListener("abort", onRequestAbort); } }; try { const response = await callFetch(input, { ...init, signal: controller.signal, }); if (response.status === 421) { const retry = allowMisdirectedFallback && canForceTransportFallback("misdirected-request"); // HTTP 421 permits retrying a non-idempotent request; // arbitrary thrown 421 shapes do not own that fact. await response.body?.cancel().catch(() => undefined); if (retry) { await releaseRequest(); return runFetch(); } throw new TelegramRequestNotStartedError(); } // grammY consumes JSON after fetch resolves; keep its deadline and // cancellation linked until the response body settles. return responseWithRelease(response, releaseRequest); } catch (err) { await releaseRequest(); if (requestTimedOut && timeoutError) { throw timeoutError; } throw err; } }; try { return await runFetch(true); } catch (err) { if ( requestTimeoutMs && shouldRetryTimedOutTelegramControlRequest(method) && canForceTransportFallback("request-timeout") ) { return await runFetch(); } if ( isTelegramMisdirectedRequestError(err) && canForceTransportFallback("misdirected-request") ) { return await runFetch(); } throw err; } }; return (input: TelegramFetchInput, init?: TelegramFetchInit) => { return Promise.resolve(wrappedFetch(input, init)).catch((err: unknown) => { try { tagTelegramNetworkError(err, { method: extractTelegramApiMethod(input), url: readRequestUrl(input), }); } catch { // Tagging is best-effort; preserve the original fetch failure if the // error object cannot accept extra metadata. } throw err; }); }; }