diff --git a/src/channels/telegram/bot.ts b/src/channels/telegram/bot.ts index 427c465..51cfc22 100644 --- a/src/channels/telegram/bot.ts +++ b/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; diff --git a/src/channels/telegram/messages.ts b/src/channels/telegram/messages.ts index dd9cda1..6374f6f 100644 --- a/src/channels/telegram/messages.ts +++ b/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( diff --git a/src/channels/telegram/updates.ts b/src/channels/telegram/updates.ts index 577685e..a2ff57f 100644 --- a/src/channels/telegram/updates.ts +++ b/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.", + ); }); } diff --git a/src/domain/ai-drafts.ts b/src/domain/ai-drafts.ts index 5cc5430..ef90d7b 100644 --- a/src/domain/ai-drafts.ts +++ b/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; + } } diff --git a/src/domain/conversation-expiry.ts b/src/domain/conversation-expiry.ts index 6898499..9efbad9 100644 --- a/src/domain/conversation-expiry.ts +++ b/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 { 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; } } diff --git a/src/domain/conversations.ts b/src/domain/conversations.ts index 0b3c56c..268d96f 100644 --- a/src/domain/conversations.ts +++ b/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 | 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; + } } diff --git a/src/domain/deliveries.ts b/src/domain/deliveries.ts index 14a5b7b..6a2d625 100644 --- a/src/domain/deliveries.ts +++ b/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; + } } diff --git a/src/runtime/main.ts b/src/runtime/main.ts index 4f9028d..7b4b28a 100644 --- a/src/runtime/main.ts +++ b/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 { 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 { } 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 { pollingBot = false; } -process.once("SIGINT", async () => { - logger.info("Stopping InboxBridge after SIGINT."); - await stopRuntime(); - handle.client.close(); - process.exit(0); +const rejectionWindow = new Map(); +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(); diff --git a/src/runtime/web-console.ts b/src/runtime/web-console.ts index ae9c4fd..46ad870 100644 --- a/src/runtime/web-console.ts +++ b/src/runtime/web-console.ts @@ -170,12 +170,39 @@ export interface ConsoleStatus { issues: string[]; } +export interface MetricsSnapshot { + messages: { + inbound_total: number; + outbound_total: number; + internal_total: number; + }; + deliveries: { + pending: number; + sent: number; + failed: number; + permanent_failure: number; + }; + conversations: { + open: number; + closed: number; + }; + ai_drafts: { + pending: number; + ready: number; + failed: number; + }; + uptime_seconds: number; + timestamp: string; +} + export interface WebConsoleOptions { settings: AppSettingsService; port: number; getStatus: () => ConsoleStatus; onConfigSaved: () => Promise; telegramWebhook?: (req: IncomingMessage, res: ServerResponse) => Promise; + dbHealthCheck: () => boolean; + collectMetrics: () => MetricsSnapshot; } type SessionKind = "password" | "setup"; @@ -202,6 +229,30 @@ export async function startWebConsole(options: WebConsoleOptions): Promise true; +const stubMetrics = () => ({ + messages: { inbound_total: 0, outbound_total: 0, internal_total: 0 }, + deliveries: { pending: 0, sent: 0, failed: 0, permanent_failure: 0 }, + conversations: { open: 0, closed: 0 }, + ai_drafts: { pending: 0, ready: 0, failed: 0 }, + uptime_seconds: 0, + timestamp: new Date().toISOString(), +}); + beforeEach(async () => { tempDir = await mkdtemp(join(tmpdir(), "inboxbridge-")); handle = createDb(`file:${join(tempDir, "test.sqlite")}`); @@ -154,6 +165,8 @@ describe("web console", () => { port: 0, getStatus: () => ({ bot: "stopped", issues: [] }), onConfigSaved: async () => {}, + dbHealthCheck: noopDbHealthCheck, + collectMetrics: stubMetrics, }); const port = (server.address() as AddressInfo).port; @@ -192,6 +205,8 @@ describe("web console", () => { port: 0, getStatus: () => ({ bot: "stopped", issues: [] }), onConfigSaved: async () => {}, + dbHealthCheck: noopDbHealthCheck, + collectMetrics: stubMetrics, }); const port = (server.address() as AddressInfo).port; @@ -215,6 +230,8 @@ describe("web console", () => { port: 0, getStatus: () => ({ bot: "stopped", issues: [] }), onConfigSaved: async () => {}, + dbHealthCheck: noopDbHealthCheck, + collectMetrics: stubMetrics, telegramWebhook: async () => { throw new Error("webhook failed"); }, @@ -246,6 +263,8 @@ describe("web console", () => { port, getStatus: () => ({ bot: "stopped", issues: [] }), onConfigSaved: async () => {}, + dbHealthCheck: noopDbHealthCheck, + collectMetrics: stubMetrics, }), /EADDRINUSE/, ); @@ -253,6 +272,112 @@ describe("web console", () => { await new Promise((resolve) => blocker.close(() => resolve())); } }); + + it("exposes /healthz without authentication", async () => { + const settings = new AppSettingsService(handle.db); + settings.setMany({ WEB_CONSOLE_PASSWORD_HASH: "bad:hash" }); + const server = await startWebConsole({ + settings, + port: 0, + getStatus: () => ({ bot: "stopped", issues: [] }), + onConfigSaved: async () => {}, + dbHealthCheck: () => true, + collectMetrics: stubMetrics, + }); + const port = (server.address() as AddressInfo).port; + try { + const res = await fetch(`http://127.0.0.1:${port}/healthz`); + assert.equal(res.status, 503); + const body = (await res.json()) as { status: string; bot: string; db: string }; + assert.equal(body.status, "degraded"); + assert.equal(body.bot, "stopped"); + assert.equal(body.db, "reachable"); + } finally { + server.close(); + } + }); + + it("returns 200 from /healthz when bot is running and db is reachable", async () => { + const settings = new AppSettingsService(handle.db); + settings.setMany({ WEB_CONSOLE_PASSWORD_HASH: "bad:hash" }); + const server = await startWebConsole({ + settings, + port: 0, + getStatus: () => ({ bot: "running", issues: [] }), + onConfigSaved: async () => {}, + dbHealthCheck: () => true, + collectMetrics: stubMetrics, + }); + const port = (server.address() as AddressInfo).port; + try { + const res = await fetch(`http://127.0.0.1:${port}/healthz`); + assert.equal(res.status, 200); + const body = (await res.json()) as { status: string }; + assert.equal(body.status, "ok"); + } finally { + server.close(); + } + }); + + it("redirects /metrics to /login without authentication", async () => { + const settings = new AppSettingsService(handle.db); + settings.setMany({ WEB_CONSOLE_PASSWORD_HASH: "bad:hash" }); + const server = await startWebConsole({ + settings, + port: 0, + getStatus: () => ({ bot: "stopped", issues: [] }), + onConfigSaved: async () => {}, + dbHealthCheck: noopDbHealthCheck, + collectMetrics: stubMetrics, + }); + const port = (server.address() as AddressInfo).port; + try { + const res = await fetch(`http://127.0.0.1:${port}/metrics`, { redirect: "manual" }); + assert.equal(res.status, 302); + assert.equal(res.headers.get("location"), "/login"); + } finally { + server.close(); + } + }); + + it("returns metrics JSON after authentication", async () => { + const settings = new AppSettingsService(handle.db); + settings.setMany({ WEB_CONSOLE_PASSWORD_HASH: "bad:hash" }); + const server = await startWebConsole({ + settings, + port: 0, + getStatus: () => ({ bot: "running", issues: [] }), + onConfigSaved: async () => {}, + dbHealthCheck: noopDbHealthCheck, + collectMetrics: () => ({ + messages: { inbound_total: 5, outbound_total: 3, internal_total: 1 }, + deliveries: { pending: 0, sent: 3, failed: 1, permanent_failure: 0 }, + conversations: { open: 2, closed: 1 }, + ai_drafts: { pending: 0, ready: 1, failed: 0 }, + uptime_seconds: 42, + timestamp: "2026-01-01T00:00:00.000Z", + }), + }); + const port = (server.address() as AddressInfo).port; + try { + const loginRes = await fetch(`http://127.0.0.1:${port}/login`, { + method: "POST", + body: new URLSearchParams({ setupToken: "setup-token" }), + redirect: "manual", + }); + const cookie = loginRes.headers.get("set-cookie")?.split(";")[0]; + assert.ok(cookie); + const res = await fetch(`http://127.0.0.1:${port}/metrics`, { + headers: { cookie }, + }); + assert.equal(res.status, 200); + const body = (await res.json()) as { messages: { inbound_total: number }; conversations: { open: number } }; + assert.equal(body.messages.inbound_total, 5); + assert.equal(body.conversations.open, 2); + } finally { + server.close(); + } + }); }); describe("conversation service", () => { @@ -400,6 +525,37 @@ describe("conversation service", () => { assert.equal((await service.recentMessages(bundle.conversation.id, 10)).length, 0); assert.equal((await service.getOrCreateConversation({ platform: "telegram", externalUserId: "42" })).contact.id, bundle.contact.id); }); + + it("aggregates conversation and message stats", async () => { + const service = new ConversationService(handle.db, 30); + const a = await service.getOrCreateConversation({ platform: "telegram", externalUserId: "1", displayName: "A" }); + const b = await service.getOrCreateConversation({ platform: "telegram", externalUserId: "2", displayName: "B" }); + await service.setConversationStatus(b.conversation.id, "closed"); + await service.createMessage({ + conversationId: a.conversation.id, + contactId: a.contact.id, + direction: "inbound", + platform: "telegram", + messageType: "text", + text: "hi", + }); + await service.createMessage({ + conversationId: a.conversation.id, + contactId: a.contact.id, + direction: "outbound", + platform: "telegram", + messageType: "text", + text: "hello", + }); + + const convStats = service.conversationStats(); + assert.equal(convStats.open, 1); + assert.equal(convStats.closed, 1); + + const msgStats = service.messageStats(); + assert.equal(msgStats.inbound, 1); + assert.equal(msgStats.outbound, 1); + }); }); describe("permissions and rate limits", () => { @@ -489,4 +645,20 @@ describe("telegram helpers", () => { assert.ok(adminBotCommands.some((command) => command.command === "delete")); assert.ok(adminBotCommands.every((command) => !command.command.startsWith("/"))); }); + + it("aggregates delivery stats by status", async () => { + const deliveries = new DeliveryService(handle.db); + const id1 = await deliveries.createPending(undefined, "telegram-user:1"); + const id2 = await deliveries.createPending(undefined, "telegram-user:2"); + const id3 = await deliveries.createPending(undefined, "telegram-user:3"); + await deliveries.markSent(id1); + await deliveries.markFailed(id2, "error", 1); + await deliveries.markPermanentFailure(id3, "fatal"); + + const stats = deliveries.stats(); + assert.equal(stats.pending, 0); + assert.equal(stats.sent, 1); + assert.equal(stats.failed, 1); + assert.equal(stats.permanentFailure, 1); + }); });