fix(telegram): webhook hang - tests and fix (openclaw#26933) thanks @huntharo

Verified:
- pnpm build
- pnpm check
- pnpm test:macmini

Co-authored-by: huntharo <5617868+huntharo@users.noreply.github.com>
Co-authored-by: Tak Hoffman <781889+Takhoffman@users.noreply.github.com>

src/telegram/webhook.ts

@@ -3,7 +3,7 @@ import { webhookCallback } from "grammy";
 import type { OpenClawConfig } from "../config/config.js";
 import { isDiagnosticsEnabled } from "../infra/diagnostic-events.js";
 import { formatErrorMessage } from "../infra/errors.js";
-import { installRequestBodyLimitGuard } from "../infra/http-body.js";
+import { readJsonBodyWithLimit } from "../infra/http-body.js";
 import {
   logWebhookError,
   logWebhookProcessed,
@@ -21,6 +21,59 @@ const TELEGRAM_WEBHOOK_MAX_BODY_BYTES = 1024 * 1024;
 const TELEGRAM_WEBHOOK_BODY_TIMEOUT_MS = 30_000;
 const TELEGRAM_WEBHOOK_CALLBACK_TIMEOUT_MS = 10_000;
 
+async function listenHttpServer(params: {
+  server: ReturnType<typeof createServer>;
+  port: number;
+  host: string;
+}) {
+  await new Promise<void>((resolve, reject) => {
+    const onError = (err: Error) => {
+      params.server.off("error", onError);
+      reject(err);
+    };
+    params.server.once("error", onError);
+    params.server.listen(params.port, params.host, () => {
+      params.server.off("error", onError);
+      resolve();
+    });
+  });
+}
+
+function resolveWebhookPublicUrl(params: {
+  configuredPublicUrl?: string;
+  server: ReturnType<typeof createServer>;
+  path: string;
+  host: string;
+  port: number;
+}) {
+  if (params.configuredPublicUrl) {
+    return params.configuredPublicUrl;
+  }
+  const address = params.server.address();
+  if (address && typeof address !== "string") {
+    const resolvedHost =
+      params.host === "0.0.0.0" || address.address === "0.0.0.0" || address.address === "::"
+        ? "localhost"
+        : address.address;
+    return `http://${resolvedHost}:${address.port}${params.path}`;
+  }
+  const fallbackHost = params.host === "0.0.0.0" ? "localhost" : params.host;
+  return `http://${fallbackHost}:${params.port}${params.path}`;
+}
+
+async function initializeTelegramWebhookBot(params: {
+  bot: ReturnType<typeof createTelegramBot>;
+  runtime: RuntimeEnv;
+  abortSignal?: AbortSignal;
+}) {
+  const initSignal = params.abortSignal as Parameters<(typeof params.bot)["init"]>[0];
+  await withTelegramApiErrorLogging({
+    operation: "getMe",
+    runtime: params.runtime,
+    fn: () => params.bot.init(initSignal),
+  });
+}
+
 export async function startTelegramWebhook(opts: {
   token: string;
   accountId?: string;
@@ -55,7 +108,12 @@ export async function startTelegramWebhook(opts: {
     config: opts.config,
     accountId: opts.accountId,
   });
-  const handler = webhookCallback(bot, "http", {
+  await initializeTelegramWebhookBot({
+    bot,
+    runtime,
+    abortSignal: opts.abortSignal,
+  });
+  const handler = webhookCallback(bot, "callback", {
     secretToken: secret,
     onTimeout: "return",
     timeoutMilliseconds: TELEGRAM_WEBHOOK_CALLBACK_TIMEOUT_MS,
@@ -66,6 +124,14 @@ export async function startTelegramWebhook(opts: {
   }
 
   const server = createServer((req, res) => {
+    const respondText = (statusCode: number, text = "") => {
+      if (res.headersSent || res.writableEnded) {
+        return;
+      }
+      res.writeHead(statusCode, { "Content-Type": "text/plain; charset=utf-8" });
+      res.end(text);
+    };
+
     if (req.url === healthPath) {
       res.writeHead(200);
       res.end("ok");
@@ -80,69 +146,125 @@ export async function startTelegramWebhook(opts: {
     if (diagnosticsEnabled) {
       logWebhookReceived({ channel: "telegram", updateType: "telegram-post" });
     }
-    const guard = installRequestBodyLimitGuard(req, res, {
-      maxBytes: TELEGRAM_WEBHOOK_MAX_BODY_BYTES,
-      timeoutMs: TELEGRAM_WEBHOOK_BODY_TIMEOUT_MS,
-      responseFormat: "text",
-    });
-    if (guard.isTripped()) {
-      return;
-    }
-    const handled = handler(req, res);
-    if (handled && typeof handled.catch === "function") {
-      void handled
-        .then(() => {
-          if (diagnosticsEnabled) {
-            logWebhookProcessed({
-              channel: "telegram",
-              updateType: "telegram-post",
-              durationMs: Date.now() - startTime,
-            });
-          }
-        })
-        .catch((err) => {
-          if (guard.isTripped()) {
-            return;
-          }
-          const errMsg = formatErrorMessage(err);
-          if (diagnosticsEnabled) {
-            logWebhookError({
-              channel: "telegram",
-              updateType: "telegram-post",
-              error: errMsg,
-            });
-          }
-          runtime.log?.(`webhook handler failed: ${errMsg}`);
-          if (!res.headersSent) {
-            res.writeHead(500);
-          }
-          res.end();
-        })
-        .finally(() => {
-          guard.dispose();
+    void (async () => {
+      const body = await readJsonBodyWithLimit(req, {
+        maxBytes: TELEGRAM_WEBHOOK_MAX_BODY_BYTES,
+        timeoutMs: TELEGRAM_WEBHOOK_BODY_TIMEOUT_MS,
+        emptyObjectOnEmpty: false,
+      });
+      if (!body.ok) {
+        if (body.code === "PAYLOAD_TOO_LARGE") {
+          respondText(413, body.error);
+          return;
+        }
+        if (body.code === "REQUEST_BODY_TIMEOUT") {
+          respondText(408, body.error);
+          return;
+        }
+        if (body.code === "CONNECTION_CLOSED") {
+          respondText(400, body.error);
+          return;
+        }
+        respondText(400, body.error);
+        return;
+      }
+
+      let replied = false;
+      const reply = async (json: string) => {
+        if (replied) {
+          return;
+        }
+        replied = true;
+        if (res.headersSent || res.writableEnded) {
+          return;
+        }
+        res.writeHead(200, { "Content-Type": "application/json; charset=utf-8" });
+        res.end(json);
+      };
+      const unauthorized = async () => {
+        if (replied) {
+          return;
+        }
+        replied = true;
+        respondText(401, "unauthorized");
+      };
+      const secretHeaderRaw = req.headers["x-telegram-bot-api-secret-token"];
+      const secretHeader = Array.isArray(secretHeaderRaw) ? secretHeaderRaw[0] : secretHeaderRaw;
+
+      await handler(body.value, reply, secretHeader, unauthorized);
+      if (!replied) {
+        respondText(200);
+      }
+
+      if (diagnosticsEnabled) {
+        logWebhookProcessed({
+          channel: "telegram",
+          updateType: "telegram-post",
+          durationMs: Date.now() - startTime,
         });
-      return;
+      }
+    })().catch((err) => {
+      const errMsg = formatErrorMessage(err);
+      if (diagnosticsEnabled) {
+        logWebhookError({
+          channel: "telegram",
+          updateType: "telegram-post",
+          error: errMsg,
+        });
+      }
+      runtime.log?.(`webhook handler failed: ${errMsg}`);
+      respondText(500);
+    });
+  });
+
+  await listenHttpServer({
+    server,
+    port,
+    host,
+  });
+
+  const publicUrl = resolveWebhookPublicUrl({
+    configuredPublicUrl: opts.publicUrl,
+    server,
+    path,
+    host,
+    port,
+  });
+
+  try {
+    await withTelegramApiErrorLogging({
+      operation: "setWebhook",
+      runtime,
+      fn: () =>
+        bot.api.setWebhook(publicUrl, {
+          secret_token: secret,
+          allowed_updates: resolveTelegramAllowedUpdates(),
+        }),
+    });
+  } catch (err) {
+    server.close();
+    void bot.stop();
+    if (diagnosticsEnabled) {
+      stopDiagnosticHeartbeat();
     }
-    guard.dispose();
-  });
+    throw err;
+  }
 
-  const publicUrl =
-    opts.publicUrl ?? `http://${host === "0.0.0.0" ? "localhost" : host}:${port}${path}`;
-
-  await withTelegramApiErrorLogging({
-    operation: "setWebhook",
-    runtime,
-    fn: () =>
-      bot.api.setWebhook(publicUrl, {
-        secret_token: secret,
-        allowed_updates: resolveTelegramAllowedUpdates(),
-      }),
-  });
-
-  await new Promise<void>((resolve) => server.listen(port, host, resolve));
   runtime.log?.(`webhook listening on ${publicUrl}`);
 
+  let shutDown = false;
   const shutdown = () => {
+    if (shutDown) {
+      return;
+    }
+    shutDown = true;
+    void withTelegramApiErrorLogging({
+      operation: "deleteWebhook",
+      runtime,
+      fn: () => bot.api.deleteWebhook({ drop_pending_updates: false }),
+    }).catch(() => {
+      // withTelegramApiErrorLogging has already emitted the failure.
+    });
     server.close();
     void bot.stop();
     if (diagnosticsEnabled) {