feat(runtime): stability foundation - global handlers, pino injection, health/metrics

src/channels/telegram/bot.ts

@@ -1,6 +1,7 @@
 import { createHash, timingSafeEqual } from "node:crypto";
 import type { IncomingMessage, ServerResponse } from "node:http";
 import { Bot, webhookCallback } from "grammy";
+import type { Logger } from "pino";
 import type { AppConfig } from "../../runtime/config.js";
 import { AiDraftService } from "../../domain/ai-drafts.js";
 import { ConversationService } from "../../domain/conversations.js";
@@ -11,7 +12,7 @@ import type { Database } from "../../storage/client.js";
 import { registerTelegramMenu } from "./menu.js";
 import { registerTelegramUpdates } from "./updates.js";
 
-export function createTelegramBot(config: AppConfig, db: Database): Bot {
+export function createTelegramBot(config: AppConfig, db: Database, logger: Logger): Bot {
   const bot = new Bot(config.TELEGRAM_BOT_TOKEN);
   const conversations = new ConversationService(
     db,
@@ -25,6 +26,7 @@ export function createTelegramBot(config: AppConfig, db: Database): Bot {
     permissions: new PermissionService(config.TELEGRAM_ADMIN_USER_IDS),
     rateLimit: new RateLimitService(config.RATE_LIMIT_WINDOW_SECONDS, config.RATE_LIMIT_MAX_MESSAGES),
     aiDrafts: new AiDraftService(db, conversations, config),
+    logger: logger.child({ module: "telegram.messages" }),
   };
   registerTelegramUpdates(bot, deps);
   return bot;

src/channels/telegram/messages.ts

@@ -1,4 +1,5 @@
 import type { Context } from "grammy";
+import type { Logger } from "pino";
 import type { AppConfig } from "../../runtime/config.js";
 import type { AiDraftService } from "../../domain/ai-drafts.js";
 import type { ConversationService } from "../../domain/conversations.js";
@@ -16,6 +17,7 @@ export interface TelegramMessageDeps {
   permissions: PermissionService;
   rateLimit: RateLimitService;
   aiDrafts: AiDraftService;
+  logger: Logger;
 }
 
 function fullName(user: { first_name?: string; last_name?: string }): string | undefined {
@@ -240,23 +242,29 @@ export async function handlePrivateMessage(ctx: Context, deps: TelegramMessageDe
         );
         return;
       } catch (retryError) {
-        console.error("Inbound delivery retry after topic recreation failed", {
-          messageId: savedMessage.id,
-          conversationId: bundle.conversation.id,
-          contactId: bundle.contact.id,
-          previousTopicThreadId: topic.messageThreadId,
-          error: retryError,
-        });
+        deps.logger.error(
+          {
+            messageId: savedMessage.id,
+            conversationId: bundle.conversation.id,
+            contactId: bundle.contact.id,
+            previousTopicThreadId: topic.messageThreadId,
+            err: retryError,
+          },
+          "Inbound delivery retry after topic recreation failed.",
+        );
       }
     }
 
-    console.error("Inbound delivery to management chat failed", {
-      messageId: savedMessage.id,
-      conversationId: bundle.conversation.id,
-      contactId: bundle.contact.id,
-      topicThreadId: topic.messageThreadId,
-      error,
-    });
+    deps.logger.error(
+      {
+        messageId: savedMessage.id,
+        conversationId: bundle.conversation.id,
+        contactId: bundle.contact.id,
+        topicThreadId: topic.messageThreadId,
+        err: error,
+      },
+      "Inbound delivery to management chat failed.",
+    );
     await ctx.reply("消息已保存，但转发到管理群失败。请稍后再试。");
     try {
       await ctx.api.sendMessage(

src/channels/telegram/updates.ts

@@ -39,6 +39,9 @@ export function registerTelegramUpdates(bot: Bot, deps: TelegramMessageDeps): vo
 
   bot.catch((error) => {
     const ctx = error.ctx;
-    console.error(`Telegram update ${ctx.update.update_id} failed`, error.error);
+    deps.logger.error(
+      { updateId: ctx.update.update_id, err: error.error },
+      "Telegram update processing failed.",
+    );
   });
 }

src/domain/ai-drafts.ts

@@ -87,4 +87,17 @@ export class AiDraftService {
       return { status: "failed", error: message };
     }
   }
+
+  stats(): { pending: number; ready: number; failed: number } {
+    const rows = this.db
+      .prepare("SELECT status, COUNT(*) AS cnt FROM ai_drafts GROUP BY status")
+      .all() as Array<{ status: string; cnt: number }>;
+    const result = { pending: 0, ready: 0, failed: 0 };
+    for (const row of rows) {
+      if (row.status === "pending") result.pending = row.cnt;
+      else if (row.status === "ready") result.ready = row.cnt;
+      else if (row.status === "failed") result.failed = row.cnt;
+    }
+    return result;
+  }
 }

src/domain/conversation-expiry.ts

@@ -1,4 +1,5 @@
 import type { Api } from "grammy";
+import type { Logger } from "pino";
 import type { Database } from "../storage/client.js";
 import { ConversationService } from "./conversations.js";
 
@@ -7,6 +8,7 @@ export async function sweepExpiredConversations(input: {
   db: Database;
   messageRetentionDays: number;
   defaultConversationRetentionDays: number | null;
+  logger: Logger;
 }): Promise<number> {
   const conversations = new ConversationService(
     input.db,
@@ -25,11 +27,14 @@ export async function sweepExpiredConversations(input: {
     } catch (error) {
       const message = error instanceof Error ? error.message : String(error);
       if (!message.toLowerCase().includes("message thread not found")) {
-        console.error("Failed to delete expired Telegram topic", {
-          conversationId: item.conversation.id,
-          messageThreadId: item.topic.messageThreadId,
-          error,
-        });
+        input.logger.error(
+          {
+            conversationId: item.conversation.id,
+            messageThreadId: item.topic.messageThreadId,
+            err: error,
+          },
+          "Failed to delete expired Telegram topic.",
+        );
         continue;
       }
     }

src/domain/conversations.ts

@@ -467,4 +467,29 @@ export class ConversationService {
     const row = this.db.prepare("SELECT * FROM tags WHERE name = ?").get(name) as Record<string, unknown> | undefined;
     return row ? tagFromRow(row) : undefined;
   }
+
+  conversationStats(): { open: number; closed: number } {
+    const rows = this.db
+      .prepare("SELECT status, COUNT(*) AS cnt FROM conversations GROUP BY status")
+      .all() as Array<{ status: string; cnt: number }>;
+    const result = { open: 0, closed: 0 };
+    for (const row of rows) {
+      if (row.status === "open") result.open = row.cnt;
+      else if (row.status === "closed") result.closed = row.cnt;
+    }
+    return result;
+  }
+
+  messageStats(): { inbound: number; outbound: number; internal: number } {
+    const rows = this.db
+      .prepare("SELECT direction, COUNT(*) AS cnt FROM messages GROUP BY direction")
+      .all() as Array<{ direction: string; cnt: number }>;
+    const result = { inbound: 0, outbound: 0, internal: 0 };
+    for (const row of rows) {
+      if (row.direction === "inbound") result.inbound = row.cnt;
+      else if (row.direction === "outbound") result.outbound = row.cnt;
+      else if (row.direction === "internal") result.internal = row.cnt;
+    }
+    return result;
+  }
 }

src/domain/deliveries.ts

@@ -66,4 +66,23 @@ export class DeliveryService {
       )
       .run(error, nowIso(), deliveryId);
   }
+
+  stats(): {
+    pending: number;
+    sent: number;
+    failed: number;
+    permanentFailure: number;
+  } {
+    const rows = this.db
+      .prepare("SELECT status, COUNT(*) AS cnt FROM deliveries GROUP BY status")
+      .all() as Array<{ status: string; cnt: number }>;
+    const result = { pending: 0, sent: 0, failed: 0, permanentFailure: 0 };
+    for (const row of rows) {
+      if (row.status === "pending") result.pending = row.cnt;
+      else if (row.status === "sent") result.sent = row.cnt;
+      else if (row.status === "failed") result.failed = row.cnt;
+      else if (row.status === "permanent_failure") result.permanentFailure = row.cnt;
+    }
+    return result;
+  }
 }

src/runtime/main.ts

@@ -14,6 +14,7 @@ import { RetentionService } from "../domain/retention.js";
 import { startDeliveryRetryWorker } from "../domain/delivery-retry.js";
 import { DeliveryService } from "../domain/deliveries.js";
 import { ConversationService } from "../domain/conversations.js";
+import { AiDraftService } from "../domain/ai-drafts.js";
 import { AppSettingsService } from "../domain/app-settings.js";
 import { ensureSetupToken, startWebConsole } from "./web-console.js";
 import type { IncomingMessage, ServerResponse } from "node:http";
@@ -46,6 +47,7 @@ async function runExpirySweep(): Promise<void> {
     db: handle.db,
     messageRetentionDays: config.MESSAGE_RETENTION_DAYS,
     defaultConversationRetentionDays: config.DEFAULT_CONVERSATION_RETENTION_DAYS,
+    logger: logger.child({ module: "conversation-expiry" }),
   });
   if (cleaned > 0) {
     logger.info({ cleaned }, "Expired conversations cleaned.");
@@ -78,7 +80,7 @@ async function restartRuntimeUnlocked(): Promise<void> {
   }
 
   const config = loadConfigFromSources(settings.all());
-  const bot = createTelegramBot(config, handle.db);
+  const bot = createTelegramBot(config, handle.db, logger);
   activeBot = bot;
   pollingBot = config.TELEGRAM_UPDATE_MODE === "polling";
   telegramWebhook = pollingBot ? undefined : createTelegramWebhookHandler(bot, config);
@@ -165,20 +167,84 @@ async function stopRuntime(): Promise<void> {
   pollingBot = false;
 }
 
-process.once("SIGINT", async () => {
-  logger.info("Stopping InboxBridge after SIGINT.");
-  await stopRuntime();
-  handle.client.close();
-  process.exit(0);
+const rejectionWindow = new Map<string, { count: number; firstAt: number }>();
+const REJECTION_DEDUP_WINDOW_MS = 60_000;
+const REJECTION_DEDUP_THRESHOLD = 5;
+
+process.on("unhandledRejection", (reason) => {
+  const key = reason instanceof Error ? reason.message : String(reason);
+  const now = Date.now();
+  const entry = rejectionWindow.get(key);
+  if (entry) {
+    entry.count += 1;
+    if (entry.count > REJECTION_DEDUP_THRESHOLD && now - entry.firstAt < REJECTION_DEDUP_WINDOW_MS) {
+      return;
+    }
+    if (entry.count === REJECTION_DEDUP_THRESHOLD) {
+      logger.warn({ key, count: entry.count }, "Unhandled rejection repeated; suppressing further logs for 60s.");
+    } else {
+      logger.error({ reason }, "Unhandled promise rejection.");
+    }
+  } else {
+    rejectionWindow.set(key, { count: 1, firstAt: now });
+    logger.error({ reason }, "Unhandled promise rejection.");
+  }
+  for (const [k, v] of rejectionWindow) {
+    if (now - v.firstAt >= REJECTION_DEDUP_WINDOW_MS) rejectionWindow.delete(k);
+  }
 });
 
-process.once("SIGTERM", async () => {
-  logger.info("Stopping InboxBridge after SIGTERM.");
-  await stopRuntime();
-  handle.client.close();
-  process.exit(0);
+let shuttingDown = false;
+
+process.on("uncaughtException", (error) => {
+  logger.fatal({ error }, "Uncaught exception; initiating controlled shutdown.");
+  if (shuttingDown) return;
+  shuttingDown = true;
+  void stopRuntime()
+    .catch((err) => {
+      logger.error({ error: err }, "Error during uncaughtException shutdown.");
+    })
+    .finally(() => {
+      handle.client.close();
+      process.exit(1);
+    });
+  setTimeout(() => {
+    logger.warn("Graceful shutdown timed out after 10s; forcing exit.");
+    handle.client.close();
+    process.exit(1);
+  }, 10_000).unref();
 });
 
+function gracefulShutdown(signal: string): void {
+  if (shuttingDown) return;
+  shuttingDown = true;
+  logger.info({ signal }, "Shutting down InboxBridge.");
+  let exited = false;
+  const forceExit = () => {
+    if (exited) return;
+    exited = true;
+    logger.warn("Graceful shutdown timed out after 10s; forcing exit.");
+    handle.client.close();
+    process.exit(1);
+  };
+  const timer = setTimeout(forceExit, 10_000);
+  timer.unref();
+  void stopRuntime()
+    .catch((error) => {
+      logger.error({ error }, "Error during graceful shutdown.");
+    })
+    .finally(() => {
+      if (exited) return;
+      exited = true;
+      clearTimeout(timer);
+      handle.client.close();
+      process.exit(0);
+    });
+}
+
+process.once("SIGINT", () => gracefulShutdown("SIGINT"));
+process.once("SIGTERM", () => gracefulShutdown("SIGTERM"));
+
 await startWebConsole({
   settings,
   port: databaseConfig.WEB_CONSOLE_PORT,
@@ -195,6 +261,47 @@ await startWebConsole({
     res.end("Telegram webhook is not configured.");
     return Promise.resolve();
   },
+  dbHealthCheck: () => {
+    handle.db.prepare("SELECT 1").get();
+    return true;
+  },
+  collectMetrics: () => {
+    const conversations = new ConversationService(
+      handle.db,
+      Number(loadConfigFromSources(settings.all()).MESSAGE_RETENTION_DAYS) || 30,
+      loadConfigFromSources(settings.all()).DEFAULT_CONVERSATION_RETENTION_DAYS ?? 30,
+    );
+    const deliveries = new DeliveryService(handle.db);
+    const aiDrafts = new AiDraftService(handle.db, conversations, loadConfigFromSources(settings.all()));
+    const msgStats = conversations.messageStats();
+    const convStats = conversations.conversationStats();
+    const delStats = deliveries.stats();
+    const draftStats = aiDrafts.stats();
+    return {
+      messages: {
+        inbound_total: msgStats.inbound,
+        outbound_total: msgStats.outbound,
+        internal_total: msgStats.internal,
+      },
+      deliveries: {
+        pending: delStats.pending,
+        sent: delStats.sent,
+        failed: delStats.failed,
+        permanent_failure: delStats.permanentFailure,
+      },
+      conversations: {
+        open: convStats.open,
+        closed: convStats.closed,
+      },
+      ai_drafts: {
+        pending: draftStats.pending,
+        ready: draftStats.ready,
+        failed: draftStats.failed,
+      },
+      uptime_seconds: Math.round(process.uptime()),
+      timestamp: new Date().toISOString(),
+    };
+  },
 });
 logger.info({ port: databaseConfig.WEB_CONSOLE_PORT }, "InboxBridge web console started.");
 await restartRuntime();