diff --git a/src/channels/telegram/bot.ts b/src/channels/telegram/bot.ts index 9f992ad..09b59dc 100644 --- a/src/channels/telegram/bot.ts +++ b/src/channels/telegram/bot.ts @@ -1,38 +1,9 @@ import { createHash, timingSafeEqual } from "node:crypto"; import type { IncomingMessage, ServerResponse } from "node:http"; -import { Bot, webhookCallback } from "grammy"; -import type { Logger } from "pino"; +import { webhookCallback, type Bot } from "grammy"; import type { AppConfig } from "../../runtime/config.js"; -import { AiDraftService } from "../../domain/ai-drafts.js"; -import { AuditService } from "../../domain/audit.js"; -import { ConversationService } from "../../domain/conversations.js"; -import { DeliveryService } from "../../domain/deliveries.js"; -import { PermissionService } from "../../domain/permissions.js"; -import { RateLimitService } from "../../domain/rate-limit.js"; -import type { Database } from "../../storage/client.js"; import { registerTelegramMenu } from "./menu.js"; -import { registerTelegramUpdates } from "./updates.js"; - -export function createTelegramBot(config: AppConfig, db: Database, logger: Logger): Bot { - const bot = new Bot(config.TELEGRAM_BOT_TOKEN); - const conversations = new ConversationService( - db, - config.MESSAGE_RETENTION_DAYS, - config.DEFAULT_CONVERSATION_RETENTION_DAYS, - ); - const deps = { - config, - conversations, - deliveries: new DeliveryService(db), - 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), - audit: new AuditService(db), - logger: logger.child({ module: "telegram.messages" }), - }; - registerTelegramUpdates(bot, deps); - return bot; -} +export { createTelegramBot } from "./factory.js"; export async function startTelegramBot(bot: Bot, config: AppConfig): Promise { await prepareTelegramBot(bot, config); diff --git a/src/channels/telegram/commands.ts b/src/channels/telegram/commands.ts index e9149cf..cc65034 100644 --- a/src/channels/telegram/commands.ts +++ b/src/channels/telegram/commands.ts @@ -183,7 +183,7 @@ export async function handleTopicCommand( } const updated = await deps.conversations.setConversationRetention(conversation.id, days); if (updated) { - deps.audit.log({ adminId, conversationId: conversation.id, action: "expire", detail: days === null ? "never" : String(days) }); + await deps.audit.log({ adminId, conversationId: conversation.id, action: "expire", detail: days === null ? "never" : String(days) }); } await ctx.reply(updated ? retentionText(updated) : "会话不存在,无法设置销毁策略。"); return true; @@ -212,7 +212,7 @@ export async function handleTopicCommand( return true; } await deps.conversations.addNote(conversation.id, adminId, args); - deps.audit.log({ adminId, conversationId: conversation.id, action: "note" }); + await deps.audit.log({ adminId, conversationId: conversation.id, action: "note" }); await ctx.reply("内部备注已保存。"); return true; } @@ -237,7 +237,7 @@ export async function handleTopicCommand( return true; } await deps.conversations.addTag(conversation.id, args); - deps.audit.log({ adminId, conversationId: conversation.id, action: "tag", detail: args }); + await deps.audit.log({ adminId, conversationId: conversation.id, action: "tag", detail: args }); await ctx.reply(`标签已添加:${args}`); return true; } @@ -247,7 +247,7 @@ export async function handleTopicCommand( return true; } await deps.conversations.removeTag(conversation.id, args); - deps.audit.log({ adminId, conversationId: conversation.id, action: "untag", detail: args }); + await deps.audit.log({ adminId, conversationId: conversation.id, action: "untag", detail: args }); await ctx.reply(`标签已移除:${args}`); return true; } @@ -262,7 +262,7 @@ export async function handleTopicCommand( return true; } await deps.conversations.setPriority(conversation.id, args as "low" | "normal" | "high" | "urgent"); - deps.audit.log({ adminId, conversationId: conversation.id, action: "priority", detail: args }); + await deps.audit.log({ adminId, conversationId: conversation.id, action: "priority", detail: args }); await ctx.reply(`优先级已设置为 ${args}。`); return true; } @@ -272,19 +272,19 @@ export async function handleTopicCommand( return true; } await deps.conversations.assign(conversation.id, args); - deps.audit.log({ adminId, conversationId: conversation.id, action: "assign", detail: args }); + await deps.audit.log({ adminId, conversationId: conversation.id, action: "assign", detail: args }); await ctx.reply(`已分配给 ${args}。`); return true; } case "ban": { await deps.conversations.blockContact(contact.id, adminId, args || undefined); - deps.audit.log({ adminId, conversationId: conversation.id, action: "ban", detail: args || undefined }); + await deps.audit.log({ adminId, conversationId: conversation.id, action: "ban", detail: args || undefined }); await ctx.reply("联系人已封禁,后续消息会被拒收。"); return true; } case "unban": { await deps.conversations.unblockContact(contact.id); - deps.audit.log({ adminId, conversationId: conversation.id, action: "unban" }); + await deps.audit.log({ adminId, conversationId: conversation.id, action: "unban" }); await ctx.reply("联系人已解除封禁。"); return true; } @@ -294,7 +294,7 @@ export async function handleTopicCommand( return true; } await ctx.api.deleteForumTopic(Number(topicContext.topic.managementChatId), topicContext.topic.messageThreadId); - deps.audit.log({ adminId, conversationId: conversation.id, action: "delete" }); + await deps.audit.log({ adminId, conversationId: conversation.id, action: "delete" }); await deps.conversations.deleteConversationData(conversation.id); return true; } @@ -304,20 +304,20 @@ export async function handleTopicCommand( return true; } await deps.conversations.resetConversation(conversation.id); - deps.audit.log({ adminId, conversationId: conversation.id, action: "reset" }); + await deps.audit.log({ adminId, conversationId: conversation.id, action: "reset" }); await ctx.reply("会话已重置。联系人映射和 Topic 已保留。"); return true; } case "close": { await deps.conversations.setConversationStatus(conversation.id, "closed"); - deps.audit.log({ adminId, conversationId: conversation.id, action: "close" }); + await deps.audit.log({ adminId, conversationId: conversation.id, action: "close" }); await ctx.reply("会话已关闭。"); return true; } case "open": case "reopen": { await deps.conversations.setConversationStatus(conversation.id, "open"); - deps.audit.log({ adminId, conversationId: conversation.id, action: "reopen" }); + await deps.audit.log({ adminId, conversationId: conversation.id, action: "reopen" }); await ctx.reply("会话已重新打开。"); return true; } @@ -328,14 +328,14 @@ export async function handleTopicCommand( return true; } await deps.conversations.mute(conversation.id, mutedUntil); - deps.audit.log({ adminId, conversationId: conversation.id, action: "mute", detail: mutedUntil }); + await deps.audit.log({ adminId, conversationId: conversation.id, action: "mute", detail: mutedUntil }); await ctx.reply(`已静音至 ${mutedUntil}。`); return true; } case "draft": { const subCommand = args.trim().split(/\s+/)[0]?.toLowerCase(); if (subCommand === "send") { - const draft = deps.aiDrafts.findReady(conversation.id); + const draft = await deps.aiDrafts.findReady(conversation.id); if (!draft || !draft.draftText) { await ctx.reply("当前没有可发送的草稿,使用 /draft 生成。"); return true; @@ -350,8 +350,8 @@ export async function handleTopicCommand( await deps.deliveries.markFailed(deliveryId, errMsg, 3); throw error; } - deps.aiDrafts.markSent(draft.id); - deps.audit.log({ adminId, conversationId: conversation.id, action: "draft_send", detail: String(draft.id) }); + await deps.aiDrafts.markSent(draft.id); + await deps.audit.log({ adminId, conversationId: conversation.id, action: "draft_send", detail: String(draft.id) }); await ctx.reply("草稿已发送给外部用户。"); } catch (error) { const msg = error instanceof Error ? error.message : String(error); @@ -360,18 +360,18 @@ export async function handleTopicCommand( return true; } if (subCommand === "discard") { - const draft = deps.aiDrafts.findReady(conversation.id); + const draft = await deps.aiDrafts.findReady(conversation.id); if (!draft) { await ctx.reply("当前没有可丢弃的草稿。"); return true; } - deps.aiDrafts.markDiscarded(draft.id); - deps.audit.log({ adminId, conversationId: conversation.id, action: "draft_discard", detail: String(draft.id) }); + await deps.aiDrafts.markDiscarded(draft.id); + await deps.audit.log({ adminId, conversationId: conversation.id, action: "draft_discard", detail: String(draft.id) }); await ctx.reply("草稿已丢弃。"); return true; } if (subCommand === "view") { - const draft = deps.aiDrafts.findReady(conversation.id); + const draft = await deps.aiDrafts.findReady(conversation.id); if (!draft || !draft.draftText) { await ctx.reply("当前没有草稿,使用 /draft 生成。"); return true; @@ -402,7 +402,7 @@ export async function handleTopicCommand( } case "ai_on": { await deps.conversations.setAiEnabled(conversation.id, true); - deps.audit.log({ adminId, conversationId: conversation.id, action: "ai_on" }); + await deps.audit.log({ adminId, conversationId: conversation.id, action: "ai_on" }); const globalEnabled = isAiConfigured(deps.config); const hint = globalEnabled ? "" : "\n\n提示:全局 AI 未开启,需先在控制台启用 AI_DRAFTS_ENABLED。"; await ctx.reply(`已对该会话开启 AI 草稿。${hint}`); @@ -410,12 +410,12 @@ export async function handleTopicCommand( } case "ai_off": { await deps.conversations.setAiEnabled(conversation.id, false); - deps.audit.log({ adminId, conversationId: conversation.id, action: "ai_off" }); + await deps.audit.log({ adminId, conversationId: conversation.id, action: "ai_off" }); await ctx.reply("已对该会话关闭 AI 草稿。该会话不再自动生成回复草稿。"); return true; } case "mine": { - const items = deps.conversations.listByAssignee(adminId, 20); + const items = await deps.conversations.listByAssignee(adminId, 20); if (items.length === 0) { await ctx.reply("当前没有分配给你的会话。使用 /assign <你的 user_id> 分配给自己。"); return true; @@ -434,7 +434,7 @@ export async function handleTopicCommand( await ctx.reply("用法:/search <关键词>,在当前会话历史消息中搜索。"); return true; } - const results = deps.conversations.searchMessagesInConversation(conversation.id, args, 20); + const results = await deps.conversations.searchMessagesInConversation(conversation.id, args, 20); if (results.length === 0) { await ctx.reply("未找到匹配的消息。"); return true; @@ -445,7 +445,7 @@ export async function handleTopicCommand( } case "audit": { const limit = parseLimit(args, 20, 50); - const logs = deps.audit.listByConversation(conversation.id, limit); + const logs = await deps.audit.listByConversation(conversation.id, limit); if (logs.length === 0) { await ctx.reply("当前会话暂无审计记录。"); return true; diff --git a/src/channels/telegram/factory.ts b/src/channels/telegram/factory.ts new file mode 100644 index 0000000..0c1e6b2 --- /dev/null +++ b/src/channels/telegram/factory.ts @@ -0,0 +1,32 @@ +import { Bot } from "grammy"; +import type { Logger } from "pino"; +import { AiDraftService } from "../../domain/ai-drafts.js"; +import { AuditService } from "../../domain/audit.js"; +import { ConversationService } from "../../domain/conversations.js"; +import { DeliveryService } from "../../domain/deliveries.js"; +import { PermissionService } from "../../domain/permissions.js"; +import { RateLimitService } from "../../domain/rate-limit.js"; +import type { Database } from "../../ports/database.js"; +import type { AppConfig } from "../../runtime/config.js"; +import { registerTelegramUpdates } from "./updates.js"; + +export function createTelegramBot(config: AppConfig, db: Database, logger: Logger): Bot { + const bot = new Bot(config.TELEGRAM_BOT_TOKEN); + const conversations = new ConversationService( + db, + config.MESSAGE_RETENTION_DAYS, + config.DEFAULT_CONVERSATION_RETENTION_DAYS, + ); + const deps = { + config, + conversations, + deliveries: new DeliveryService(db), + 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), + audit: new AuditService(db), + logger: logger.child({ module: "telegram.messages" }), + }; + registerTelegramUpdates(bot, deps); + return bot; +} diff --git a/src/channels/telegram/secrets.ts b/src/channels/telegram/secrets.ts new file mode 100644 index 0000000..8f67bbb --- /dev/null +++ b/src/channels/telegram/secrets.ts @@ -0,0 +1,8 @@ +import type { AppConfig } from "../../runtime/config.js"; + +export async function telegramWebhookSecret(config: AppConfig): Promise { + if (config.TELEGRAM_WEBHOOK_SECRET) return config.TELEGRAM_WEBHOOK_SECRET; + const data = new TextEncoder().encode(config.TELEGRAM_BOT_TOKEN); + const digest = await globalThis.crypto.subtle.digest("SHA-256", data); + return [...new Uint8Array(digest)].map((byte) => byte.toString(16).padStart(2, "0")).join(""); +} diff --git a/src/channels/telegram/worker-webhook.ts b/src/channels/telegram/worker-webhook.ts new file mode 100644 index 0000000..c8cbd89 --- /dev/null +++ b/src/channels/telegram/worker-webhook.ts @@ -0,0 +1,18 @@ +import { webhookCallback, type Bot } from "grammy"; + +export type WorkerTelegramWebhookHandler = (request: Request) => Promise; +export type WorkerWebhookCallbackFactory = (bot: Bot, adapter: "cloudflare-mod") => WorkerTelegramWebhookHandler; + +export function createWorkerTelegramWebhookHandler( + bot: Bot, + expectedSecret: string, + callbackFactory: WorkerWebhookCallbackFactory = webhookCallback as WorkerWebhookCallbackFactory, +): WorkerTelegramWebhookHandler { + const callback = callbackFactory(bot, "cloudflare-mod"); + return async (request) => { + if (request.headers.get("x-telegram-bot-api-secret-token") !== expectedSecret) { + return new Response("Forbidden", { status: 403 }); + } + return callback(request); + }; +} diff --git a/src/domain/ai-drafts.ts b/src/domain/ai-drafts.ts index 776eb67..0efa6da 100644 --- a/src/domain/ai-drafts.ts +++ b/src/domain/ai-drafts.ts @@ -1,6 +1,6 @@ import type { AppConfig } from "../runtime/config.js"; import { isAiConfigured } from "../runtime/config.js"; -import type { Database } from "../storage/client.js"; +import type { Database } from "../ports/database.js"; import { nowIso, type ConversationService } from "./conversations.js"; export interface DraftResult { @@ -72,13 +72,13 @@ export class AiDraftService { } const timestamp = nowIso(); - this.db + await this.db .prepare( `INSERT INTO ai_drafts (conversation_id, source_message_id, status, created_at, updated_at) VALUES (?, ?, 'pending', ?, ?)`, ) .run(conversationId, sourceMessageId ?? null, timestamp, timestamp); - const row = this.db.prepare("SELECT last_insert_rowid() AS id").get() as { id: number }; + const row = (await this.db.prepare("SELECT last_insert_rowid() AS id").get()) as { id: number }; const draftId = Number(row.id); try { @@ -136,7 +136,7 @@ export class AiDraftService { } const text = truncateText(rawText); - this.db + await this.db .prepare("UPDATE ai_drafts SET status = 'ready', draft_text = ?, updated_at = ? WHERE id = ?") .run(text, nowIso(), draftId); const elapsed = Date.now() - startTime; @@ -151,46 +151,46 @@ export class AiDraftService { } const message = lastError?.message ?? "AI draft generation failed."; - this.db + await this.db .prepare("UPDATE ai_drafts SET status = 'failed', error = ?, updated_at = ? WHERE id = ?") .run(message, nowIso(), draftId); return { status: "failed", error: message }; } catch (error) { const message = error instanceof Error ? error.message : String(error); - this.db + await this.db .prepare("UPDATE ai_drafts SET status = 'failed', error = ?, updated_at = ? WHERE id = ?") .run(message, nowIso(), draftId); return { status: "failed", error: message }; } } - findReady(conversationId: number): DraftRow | undefined { - const row = this.db + async findReady(conversationId: number): Promise { + const row = (await this.db .prepare( `SELECT * FROM ai_drafts WHERE conversation_id = ? AND status = 'ready' ORDER BY created_at DESC LIMIT 1`, ) - .get(conversationId) as Record | undefined; + .get(conversationId)) as Record | undefined; return row ? draftFromRow(row) : undefined; } - markSent(draftId: number): void { - this.db + async markSent(draftId: number): Promise { + await this.db .prepare("UPDATE ai_drafts SET status = 'sent', updated_at = ? WHERE id = ?") .run(nowIso(), draftId); } - markDiscarded(draftId: number): void { - this.db + async markDiscarded(draftId: number): Promise { + await this.db .prepare("UPDATE ai_drafts SET status = 'discarded', updated_at = ? WHERE id = ?") .run(nowIso(), draftId); } - stats(): { pending: number; ready: number; failed: number; sent: number; discarded: number } { - const rows = this.db + async stats(): Promise<{ pending: number; ready: number; failed: number; sent: number; discarded: number }> { + const rows = (await this.db .prepare("SELECT status, COUNT(*) AS cnt FROM ai_drafts GROUP BY status") - .all() as Array<{ status: string; cnt: number }>; + .all()) as Array<{ status: string; cnt: number }>; const result = { pending: 0, ready: 0, failed: 0, sent: 0, discarded: 0 }; for (const row of rows) { if (row.status === "pending") result.pending = row.cnt; diff --git a/src/domain/app-settings.ts b/src/domain/app-settings.ts index 5818b32..3e273cc 100644 --- a/src/domain/app-settings.ts +++ b/src/domain/app-settings.ts @@ -1,4 +1,5 @@ -import type { Database } from "../storage/client.js"; +import type { Database } from "../ports/database.js"; +import type { ConfigMap } from "../runtime/config.js"; export interface AppSetting { key: string; @@ -9,31 +10,31 @@ export interface AppSetting { export class AppSettingsService { constructor(private readonly db: Database) {} - all(): NodeJS.ProcessEnv { - const rows = this.db.prepare("SELECT key, value FROM app_settings").all() as Array<{ key: string; value: string }>; - const env: NodeJS.ProcessEnv = {}; + async all(): Promise { + const rows = (await this.db.prepare("SELECT key, value FROM app_settings").all()) as Array<{ key: string; value: string }>; + const env: ConfigMap = {}; for (const row of rows) env[row.key] = row.value; return env; } - get(key: string): string | undefined { - const row = this.db.prepare("SELECT value FROM app_settings WHERE key = ?").get(key) as { value: string } | undefined; + async get(key: string): Promise { + const row = (await this.db.prepare("SELECT value FROM app_settings WHERE key = ?").get(key)) as { value: string } | undefined; return row?.value; } - setMany(values: Record): void { + async setMany(values: Record): Promise { const updatedAt = new Date().toISOString(); const statement = this.db.prepare( `INSERT INTO app_settings (key, value, updated_at) VALUES (?, ?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at`, ); - this.db.exec("BEGIN"); + await this.db.exec("BEGIN"); try { - for (const [key, value] of Object.entries(values)) statement.run(key, value, updatedAt); - this.db.exec("COMMIT"); + for (const [key, value] of Object.entries(values)) await statement.run(key, value, updatedAt); + await this.db.exec("COMMIT"); } catch (error) { - this.db.exec("ROLLBACK"); + await this.db.exec("ROLLBACK"); throw error; } } diff --git a/src/domain/audit.ts b/src/domain/audit.ts index 75f92ec..a3449c6 100644 --- a/src/domain/audit.ts +++ b/src/domain/audit.ts @@ -1,4 +1,4 @@ -import type { DatabaseSync } from "node:sqlite"; +import type { Database } from "../ports/database.js"; export interface AuditLogEntry { id: number; @@ -36,11 +36,11 @@ function auditEntryFromRow(row: Record): AuditLogEntry { } export class AuditService { - constructor(private db: DatabaseSync) {} + constructor(private db: Database) {} - log(input: AuditLogInput): void { + async log(input: AuditLogInput): Promise { try { - this.db + await this.db .prepare( `INSERT INTO audit_logs (admin_id, conversation_id, action, detail, created_at) VALUES (?, ?, ?, ?, ?)`, @@ -57,7 +57,7 @@ export class AuditService { } } - list(opts: AuditListOptions): { items: AuditLogEntry[]; total: number } { + async list(opts: AuditListOptions): Promise<{ items: AuditLogEntry[]; total: number }> { const conditions: string[] = []; const params: Array = []; if (opts.conversationId !== undefined) { @@ -73,21 +73,21 @@ export class AuditService { params.push(opts.action); } const where = conditions.length > 0 ? `WHERE ${conditions.join(" AND ")}` : ""; - const rows = this.db + const rows = (await this.db .prepare( `SELECT id, admin_id, conversation_id, action, detail, created_at FROM audit_logs ${where} ORDER BY created_at DESC, id DESC LIMIT ? OFFSET ?`, ) - .all(...params, opts.limit, opts.offset) as Array>; - const countRow = this.db + .all(...params, opts.limit, opts.offset)) as Array>; + const countRow = (await this.db .prepare(`SELECT COUNT(*) AS cnt FROM audit_logs ${where}`) - .get(...params) as { cnt: number }; + .get(...params)) as { cnt: number }; return { items: rows.map(auditEntryFromRow), total: countRow.cnt }; } - listByConversation(conversationId: number, limit: number): AuditLogEntry[] { - return this.list({ conversationId, limit, offset: 0 }).items; + async listByConversation(conversationId: number, limit: number): Promise { + return (await this.list({ conversationId, limit, offset: 0 })).items; } } diff --git a/src/domain/conversation-expiry.ts b/src/domain/conversation-expiry.ts index 9efbad9..8bddc85 100644 --- a/src/domain/conversation-expiry.ts +++ b/src/domain/conversation-expiry.ts @@ -1,6 +1,6 @@ import type { Api } from "grammy"; import type { Logger } from "pino"; -import type { Database } from "../storage/client.js"; +import type { Database } from "../ports/database.js"; import { ConversationService } from "./conversations.js"; export async function sweepExpiredConversations(input: { diff --git a/src/domain/conversations.ts b/src/domain/conversations.ts index 24657b9..a1de191 100644 --- a/src/domain/conversations.ts +++ b/src/domain/conversations.ts @@ -1,4 +1,4 @@ -import type { Database } from "../storage/client.js"; +import type { Database } from "../ports/database.js"; import type { Contact, Conversation, Message, Tag, TelegramTopic } from "../storage/schema.js"; export interface ContactInput { @@ -168,36 +168,36 @@ export class ConversationService { async getOrCreateConversation(input: ContactInput): Promise { const timestamp = nowIso(); - let contact = this.findContact(input.platform, input.externalUserId); + let contact = await this.findContact(input.platform, input.externalUserId); if (!contact) { - this.db + await this.db .prepare( `INSERT INTO contacts (platform, external_user_id, username, display_name, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?)`, ) .run(input.platform, input.externalUserId, input.username ?? null, input.displayName ?? null, timestamp, timestamp); - contact = this.findContact(input.platform, input.externalUserId); + contact = await this.findContact(input.platform, input.externalUserId); } else { - this.db + await this.db .prepare("UPDATE contacts SET username = ?, display_name = ?, updated_at = ? WHERE id = ?") .run(input.username ?? contact.username, input.displayName ?? contact.displayName, timestamp, contact.id); - contact = this.getContactSync(contact.id); + contact = await this.getContactSync(contact.id); } if (!contact) throw new Error("Failed to create or load contact."); - let conversation = this.findLatestConversation(contact.id); + let conversation = await this.findLatestConversation(contact.id); if (!conversation) { const expiresAt = this.defaultConversationRetentionDays === null ? null : addDaysIso(this.defaultConversationRetentionDays); - this.db + await this.db .prepare( `INSERT INTO conversations ( contact_id, retention_days, expires_at, created_at, updated_at, last_message_at ) VALUES (?, ?, ?, ?, ?, ?)`, ) .run(contact.id, this.defaultConversationRetentionDays, expiresAt, timestamp, timestamp, timestamp); - conversation = this.findLatestConversation(contact.id); + conversation = await this.findLatestConversation(contact.id); } if (!conversation) throw new Error("Failed to create or load conversation."); @@ -205,69 +205,69 @@ export class ConversationService { } async isBlocked(contactId: number): Promise { - const row = this.db.prepare("SELECT id FROM blocks WHERE contact_id = ?").get(contactId); + const row = await this.db.prepare("SELECT id FROM blocks WHERE contact_id = ?").get(contactId); return Boolean(row); } async blockContact(contactId: number, createdBy: string, reason?: string): Promise { - this.db + await this.db .prepare( `INSERT INTO blocks (contact_id, reason, created_by, created_at) VALUES (?, ?, ?, ?) ON CONFLICT(contact_id) DO UPDATE SET reason = excluded.reason, created_by = excluded.created_by, created_at = excluded.created_at`, ) .run(contactId, reason ?? null, createdBy, nowIso()); - this.db.prepare("UPDATE contacts SET status = 'blocked', updated_at = ? WHERE id = ?").run(nowIso(), contactId); + await this.db.prepare("UPDATE contacts SET status = 'blocked', updated_at = ? WHERE id = ?").run(nowIso(), contactId); } async unblockContact(contactId: number): Promise { - this.db.prepare("DELETE FROM blocks WHERE contact_id = ?").run(contactId); - this.db.prepare("UPDATE contacts SET status = 'active', updated_at = ? WHERE id = ?").run(nowIso(), contactId); + await this.db.prepare("DELETE FROM blocks WHERE contact_id = ?").run(contactId); + await this.db.prepare("UPDATE contacts SET status = 'active', updated_at = ? WHERE id = ?").run(nowIso(), contactId); } async setConversationStatus(conversationId: number, status: "open" | "closed"): Promise { - this.db.prepare("UPDATE conversations SET status = ?, updated_at = ? WHERE id = ?").run(status, nowIso(), conversationId); + await this.db.prepare("UPDATE conversations SET status = ?, updated_at = ? WHERE id = ?").run(status, nowIso(), conversationId); } async setPriority(conversationId: number, priority: "low" | "normal" | "high" | "urgent"): Promise { - this.db.prepare("UPDATE conversations SET priority = ?, updated_at = ? WHERE id = ?").run(priority, nowIso(), conversationId); + await this.db.prepare("UPDATE conversations SET priority = ?, updated_at = ? WHERE id = ?").run(priority, nowIso(), conversationId); } async assign(conversationId: number, adminUserId: string): Promise { - this.db + await this.db .prepare("UPDATE conversations SET assigned_admin_id = ?, updated_at = ? WHERE id = ?") .run(adminUserId, nowIso(), conversationId); } async mute(conversationId: number, mutedUntil: string): Promise { - this.db + await this.db .prepare("UPDATE conversations SET muted_until = ?, updated_at = ? WHERE id = ?") .run(mutedUntil, nowIso(), conversationId); } async setConversationRetention(conversationId: number, days: number | null): Promise { const expiresAt = days === null ? null : addDaysIso(days); - this.db + await this.db .prepare("UPDATE conversations SET retention_days = ?, expires_at = ?, updated_at = ? WHERE id = ?") .run(days, expiresAt, nowIso(), conversationId); return this.getConversation(conversationId); } async setAiEnabled(conversationId: number, enabled: boolean): Promise { - this.db + await this.db .prepare("UPDATE conversations SET ai_enabled = ?, updated_at = ? WHERE id = ?") .run(enabled ? 1 : 0, nowIso(), conversationId); } async getAiEnabled(conversationId: number): Promise { - const row = this.db + const row = (await this.db .prepare("SELECT ai_enabled FROM conversations WHERE id = ?") - .get(conversationId) as { ai_enabled?: number } | undefined; + .get(conversationId)) as { ai_enabled?: number } | undefined; return row?.ai_enabled === undefined ? true : Boolean(row.ai_enabled); } async addNote(conversationId: number, adminUserId: string, note: string): Promise { - this.db + await this.db .prepare("INSERT INTO admin_notes (conversation_id, admin_user_id, note, created_at) VALUES (?, ?, ?, ?)") .run(conversationId, adminUserId, note, nowIso()); } @@ -275,12 +275,12 @@ export class ConversationService { async addTag(conversationId: number, name: string): Promise { const cleanName = name.trim().toLowerCase(); if (!cleanName) return; - this.db + await this.db .prepare("INSERT INTO tags (name, created_at) VALUES (?, ?) ON CONFLICT(name) DO NOTHING") .run(cleanName, nowIso()); - const tag = this.findTag(cleanName); + const tag = await this.findTag(cleanName); if (!tag) return; - this.db + await this.db .prepare( `INSERT INTO conversation_tags (conversation_id, tag_id, created_at) VALUES (?, ?, ?) @@ -290,13 +290,13 @@ export class ConversationService { } async removeTag(conversationId: number, name: string): Promise { - const tag = this.findTag(name.trim().toLowerCase()); + const tag = await this.findTag(name.trim().toLowerCase()); if (!tag) return; - this.db.prepare("DELETE FROM conversation_tags WHERE conversation_id = ? AND tag_id = ?").run(conversationId, tag.id); + await this.db.prepare("DELETE FROM conversation_tags WHERE conversation_id = ? AND tag_id = ?").run(conversationId, tag.id); } async listTags(conversationId: number): Promise { - const rows = this.db + const rows = (await this.db .prepare( `SELECT tags.* FROM tags @@ -304,61 +304,61 @@ export class ConversationService { WHERE conversation_tags.conversation_id = ? ORDER BY tags.name ASC`, ) - .all(conversationId) as Array>; + .all(conversationId)) as Array>; return rows.map(tagFromRow); } async recentNotes(conversationId: number, limit: number): Promise { - const rows = this.db + const rows = (await this.db .prepare("SELECT * FROM admin_notes WHERE conversation_id = ? ORDER BY created_at DESC LIMIT ?") - .all(conversationId, limit) as Array>; + .all(conversationId, limit)) as Array>; return rows.map(noteFromRow); } async deleteConversationData(conversationId: number): Promise { - this.db.exec("BEGIN"); + await this.db.exec("BEGIN"); try { - this.db + await this.db .prepare( `DELETE FROM deliveries WHERE source_message_id IN (SELECT id FROM messages WHERE conversation_id = ?)`, ) .run(conversationId); - this.db.prepare("DELETE FROM ai_drafts WHERE conversation_id = ?").run(conversationId); - this.db.prepare("DELETE FROM conversation_tags WHERE conversation_id = ?").run(conversationId); - this.db.prepare("DELETE FROM admin_notes WHERE conversation_id = ?").run(conversationId); - this.db.prepare("DELETE FROM telegram_topics WHERE conversation_id = ?").run(conversationId); - this.db.prepare("DELETE FROM messages WHERE conversation_id = ?").run(conversationId); - this.db.prepare("DELETE FROM conversations WHERE id = ?").run(conversationId); - this.db.exec("COMMIT"); + await this.db.prepare("DELETE FROM ai_drafts WHERE conversation_id = ?").run(conversationId); + await this.db.prepare("DELETE FROM conversation_tags WHERE conversation_id = ?").run(conversationId); + await this.db.prepare("DELETE FROM admin_notes WHERE conversation_id = ?").run(conversationId); + await this.db.prepare("DELETE FROM telegram_topics WHERE conversation_id = ?").run(conversationId); + await this.db.prepare("DELETE FROM messages WHERE conversation_id = ?").run(conversationId); + await this.db.prepare("DELETE FROM conversations WHERE id = ?").run(conversationId); + await this.db.exec("COMMIT"); } catch (error) { - this.db.exec("ROLLBACK"); + await this.db.exec("ROLLBACK"); throw error; } } async resetConversation(conversationId: number): Promise { - this.db.exec("BEGIN"); + await this.db.exec("BEGIN"); try { - this.db + await this.db .prepare( `DELETE FROM deliveries WHERE source_message_id IN (SELECT id FROM messages WHERE conversation_id = ?)`, ) .run(conversationId); - this.db.prepare("DELETE FROM ai_drafts WHERE conversation_id = ?").run(conversationId); - this.db.prepare("DELETE FROM conversation_tags WHERE conversation_id = ?").run(conversationId); - this.db.prepare("DELETE FROM admin_notes WHERE conversation_id = ?").run(conversationId); - this.db.prepare("DELETE FROM messages WHERE conversation_id = ?").run(conversationId); - this.db.exec("COMMIT"); + await this.db.prepare("DELETE FROM ai_drafts WHERE conversation_id = ?").run(conversationId); + await this.db.prepare("DELETE FROM conversation_tags WHERE conversation_id = ?").run(conversationId); + await this.db.prepare("DELETE FROM admin_notes WHERE conversation_id = ?").run(conversationId); + await this.db.prepare("DELETE FROM messages WHERE conversation_id = ?").run(conversationId); + await this.db.exec("COMMIT"); } catch (error) { - this.db.exec("ROLLBACK"); + await this.db.exec("ROLLBACK"); throw error; } } async expiredConversations(now = nowIso()): Promise> { - const rows = this.db + const rows = (await this.db .prepare( `SELECT conversations.id AS c_id, @@ -384,7 +384,7 @@ export class ConversationService { WHERE conversations.expires_at IS NOT NULL AND conversations.expires_at <= ? ORDER BY conversations.expires_at ASC`, ) - .all(now) as Array>; + .all(now)) as Array>; return rows.map((row) => ({ conversation: conversationFromRow({ @@ -424,7 +424,7 @@ export class ConversationService { }): Promise { const timestamp = nowIso(); const expiresAt = input.direction === "internal" ? null : addDaysIso(this.retentionDays); - this.db + await this.db .prepare( `INSERT INTO messages ( conversation_id, contact_id, direction, platform, message_type, text, raw_payload, @@ -443,24 +443,24 @@ export class ConversationService { timestamp, expiresAt, ); - this.db + await this.db .prepare("UPDATE conversations SET last_message_at = ?, updated_at = ? WHERE id = ?") .run(timestamp, timestamp, input.conversationId); - const row = this.db.prepare("SELECT * FROM messages WHERE id = last_insert_rowid()").get() as Record; + const row = (await this.db.prepare("SELECT * FROM messages WHERE id = last_insert_rowid()").get()) as Record; return messageFromRow(row); } async getTopicByThread(managementChatId: string, messageThreadId: number): Promise { - const row = this.db + const row = (await this.db .prepare("SELECT * FROM telegram_topics WHERE management_chat_id = ? AND message_thread_id = ?") - .get(managementChatId, messageThreadId) as Record | undefined; + .get(managementChatId, messageThreadId)) as Record | undefined; return row ? topicFromRow(row) : undefined; } async getTopicByConversation(conversationId: number): Promise { - const row = this.db + const row = (await this.db .prepare("SELECT * FROM telegram_topics WHERE conversation_id = ?") - .get(conversationId) as Record | undefined; + .get(conversationId)) as Record | undefined; return row ? topicFromRow(row) : undefined; } @@ -471,7 +471,7 @@ export class ConversationService { topicName: string; }): Promise { const timestamp = nowIso(); - this.db + await this.db .prepare( `INSERT INTO telegram_topics (conversation_id, management_chat_id, message_thread_id, topic_name, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?) @@ -496,54 +496,54 @@ export class ConversationService { } async recentMessages(conversationId: number, limit: number): Promise { - const rows = this.db + const rows = (await this.db .prepare("SELECT * FROM messages WHERE conversation_id = ? ORDER BY created_at DESC LIMIT ?") - .all(conversationId, limit) as Array>; + .all(conversationId, limit)) as Array>; return rows.map(messageFromRow); } async getMessage(messageId: number): Promise { - const row = this.db + const row = (await this.db .prepare("SELECT * FROM messages WHERE id = ?") - .get(messageId) as Record | undefined; + .get(messageId)) as Record | undefined; return row ? messageFromRow(row) : undefined; } - private findContact(platform: string, externalUserId: string): Contact | undefined { - const row = this.db + private async findContact(platform: string, externalUserId: string): Promise { + const row = (await this.db .prepare("SELECT * FROM contacts WHERE platform = ? AND external_user_id = ?") - .get(platform, externalUserId) as Record | undefined; + .get(platform, externalUserId)) as Record | undefined; return row ? contactFromRow(row) : undefined; } - private getContactSync(contactId: number): Contact | undefined { - const row = this.db.prepare("SELECT * FROM contacts WHERE id = ?").get(contactId) as Record | undefined; + private async getContactSync(contactId: number): Promise { + const row = (await this.db.prepare("SELECT * FROM contacts WHERE id = ?").get(contactId)) as Record | undefined; return row ? contactFromRow(row) : undefined; } - private findLatestConversation(contactId: number): Conversation | undefined { - const row = this.db + private async findLatestConversation(contactId: number): Promise { + const row = (await this.db .prepare("SELECT * FROM conversations WHERE contact_id = ? ORDER BY created_at DESC LIMIT 1") - .get(contactId) as Record | undefined; + .get(contactId)) as Record | undefined; return row ? conversationFromRow(row) : undefined; } - private getConversationSync(conversationId: number): Conversation | undefined { - const row = this.db + private async getConversationSync(conversationId: number): Promise { + const row = (await this.db .prepare("SELECT * FROM conversations WHERE id = ?") - .get(conversationId) as Record | undefined; + .get(conversationId)) as Record | undefined; return row ? conversationFromRow(row) : undefined; } - private findTag(name: string): Tag | undefined { - const row = this.db.prepare("SELECT * FROM tags WHERE name = ?").get(name) as Record | undefined; + private async findTag(name: string): Promise { + const row = (await 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 + async conversationStats(): Promise<{ open: number; closed: number }> { + const rows = (await this.db .prepare("SELECT status, COUNT(*) AS cnt FROM conversations GROUP BY status") - .all() as Array<{ status: string; cnt: number }>; + .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; @@ -552,10 +552,10 @@ export class ConversationService { return result; } - messageStats(): { inbound: number; outbound: number; internal: number } { - const rows = this.db + async messageStats(): Promise<{ inbound: number; outbound: number; internal: number }> { + const rows = (await this.db .prepare("SELECT direction, COUNT(*) AS cnt FROM messages GROUP BY direction") - .all() as Array<{ direction: string; cnt: number }>; + .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; @@ -565,12 +565,12 @@ export class ConversationService { return result; } - listConversations(opts: { + async listConversations(opts: { status?: "open" | "closed"; assignedTo?: string; limit: number; offset: number; - }): { items: ConversationListItem[]; total: number } { + }): Promise<{ items: ConversationListItem[]; total: number }> { const conditions: string[] = []; const params: Array = []; if (opts.status) { @@ -582,7 +582,7 @@ export class ConversationService { params.push(opts.assignedTo); } const where = conditions.length > 0 ? `WHERE ${conditions.join(" AND ")}` : ""; - const rows = this.db + const rows = (await this.db .prepare( `SELECT c.id, c.status, c.priority, c.assigned_admin_id, c.created_at, c.last_message_at, ct.display_name, ct.username, t.topic_name, t.message_thread_id @@ -593,36 +593,36 @@ export class ConversationService { ORDER BY c.last_message_at DESC NULLS LAST LIMIT ? OFFSET ?`, ) - .all(...params, opts.limit, opts.offset) as Array>; - const countRow = this.db + .all(...params, opts.limit, opts.offset)) as Array>; + const countRow = (await this.db .prepare(`SELECT COUNT(*) AS cnt FROM conversations c ${where}`) - .get(...params) as { cnt: number }; + .get(...params)) as { cnt: number }; return { items: rows.map(conversationListItemFromRow), total: countRow.cnt }; } - listByAssignee(adminId: string, limit: number): ConversationListItem[] { - return this.listConversations({ assignedTo: adminId, limit, offset: 0 }).items; + async listByAssignee(adminId: string, limit: number): Promise { + return (await this.listConversations({ assignedTo: adminId, limit, offset: 0 })).items; } - searchMessagesInConversation(conversationId: number, query: string, limit: number): Message[] { + async searchMessagesInConversation(conversationId: number, query: string, limit: number): Promise { const pattern = `%${query.replace(/[%_]/g, (m) => "\\" + m)}%`; - const rows = this.db + const rows = (await this.db .prepare( `SELECT * FROM messages WHERE conversation_id = ? AND text LIKE ? ESCAPE '\\' ORDER BY created_at DESC LIMIT ?`, ) - .all(conversationId, pattern, limit) as Array>; + .all(conversationId, pattern, limit)) as Array>; return rows.map(messageFromRow); } - searchMessages(opts: { + async searchMessages(opts: { query: string; conversationId?: number; limit: number; offset: number; - }): { items: MessageSearchResult[]; total: number } { + }): Promise<{ items: MessageSearchResult[]; total: number }> { const pattern = `%${opts.query.replace(/[%_]/g, (m) => "\\" + m)}%`; const conditions = ["m.text LIKE ? ESCAPE '\\'"]; const params: Array = [pattern]; @@ -631,7 +631,7 @@ export class ConversationService { params.push(opts.conversationId); } const where = `WHERE ${conditions.join(" AND ")}`; - const rows = this.db + const rows = (await this.db .prepare( `SELECT m.id, m.conversation_id, m.direction, m.message_type, m.text, m.created_at, ct.display_name, t.topic_name @@ -643,10 +643,10 @@ export class ConversationService { ORDER BY m.created_at DESC LIMIT ? OFFSET ?`, ) - .all(...params, opts.limit, opts.offset) as Array>; - const countRow = this.db + .all(...params, opts.limit, opts.offset)) as Array>; + const countRow = (await this.db .prepare(`SELECT COUNT(*) AS cnt FROM messages m ${where}`) - .get(...params) as { cnt: number }; + .get(...params)) as { cnt: number }; return { items: rows.map(messageSearchResultFromRow), total: countRow.cnt }; } } diff --git a/src/domain/deliveries.ts b/src/domain/deliveries.ts index 348bd9b..4f0ee81 100644 --- a/src/domain/deliveries.ts +++ b/src/domain/deliveries.ts @@ -1,4 +1,4 @@ -import type { Database } from "../storage/client.js"; +import type { Database } from "../ports/database.js"; import type { Delivery } from "../storage/schema.js"; import { nowIso } from "./conversations.js"; @@ -23,25 +23,25 @@ export class DeliveryService { async createPending(sourceMessageId: number | undefined, target: string): Promise { const timestamp = nowIso(); - this.db + await this.db .prepare( `INSERT INTO deliveries (source_message_id, target, status, created_at, updated_at) VALUES (?, ?, 'pending', ?, ?)`, ) .run(sourceMessageId ?? null, target, timestamp, timestamp); - const row = this.db.prepare("SELECT last_insert_rowid() AS id").get() as { id: number }; + const row = (await this.db.prepare("SELECT last_insert_rowid() AS id").get()) as { id: number }; return Number(row.id); } async markSent(deliveryId: number): Promise { - this.db + await this.db .prepare("UPDATE deliveries SET status = 'sent', last_error = NULL, updated_at = ? WHERE id = ?") .run(nowIso(), deliveryId); } async markFailed(deliveryId: number, error: string, attemptCount: number): Promise { const retryAt = new Date(Date.now() + Math.min(60_000, 2 ** attemptCount * 1000)).toISOString(); - this.db + await this.db .prepare( `UPDATE deliveries SET status = 'failed', attempt_count = ?, last_error = ?, next_retry_at = ?, updated_at = ? @@ -51,14 +51,14 @@ export class DeliveryService { } async dueFailed(now = nowIso()): Promise { - const rows = this.db + const rows = (await this.db .prepare("SELECT * FROM deliveries WHERE status = 'failed' AND next_retry_at <= ?") - .all(now) as Array>; + .all(now)) as Array>; return rows.map(deliveryFromRow); } async markPermanentFailure(deliveryId: number, error: string): Promise { - this.db + await this.db .prepare( `UPDATE deliveries SET status = 'permanent_failure', last_error = ?, next_retry_at = NULL, updated_at = ? @@ -67,15 +67,15 @@ export class DeliveryService { .run(error, nowIso(), deliveryId); } - stats(): { + async stats(): Promise<{ pending: number; sent: number; failed: number; permanentFailure: number; - } { - const rows = this.db + }> { + const rows = (await this.db .prepare("SELECT status, COUNT(*) AS cnt FROM deliveries GROUP BY status") - .all() as Array<{ status: string; cnt: number }>; + .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; @@ -86,26 +86,26 @@ export class DeliveryService { return result; } - listFailedDeliveries(opts: { + async listFailedDeliveries(opts: { limit: number; offset: number; - }): { items: Delivery[]; total: number } { - const rows = this.db + }): Promise<{ items: Delivery[]; total: number }> { + const rows = (await this.db .prepare( `SELECT * FROM deliveries WHERE status IN ('failed', 'permanent_failure') ORDER BY created_at ASC LIMIT ? OFFSET ?`, ) - .all(opts.limit, opts.offset) as Array>; - const countRow = this.db + .all(opts.limit, opts.offset)) as Array>; + const countRow = (await this.db .prepare(`SELECT COUNT(*) AS cnt FROM deliveries WHERE status IN ('failed', 'permanent_failure')`) - .get() as { cnt: number }; + .get()) as { cnt: number }; return { items: rows.map(deliveryFromRow), total: countRow.cnt }; } - scheduleRetry(deliveryId: number): void { - this.db + async scheduleRetry(deliveryId: number): Promise { + await this.db .prepare( `UPDATE deliveries SET next_retry_at = ?, updated_at = ? diff --git a/src/domain/retention.ts b/src/domain/retention.ts index 9640f0e..4a298a0 100644 --- a/src/domain/retention.ts +++ b/src/domain/retention.ts @@ -1,4 +1,4 @@ -import type { Database } from "../storage/client.js"; +import type { Database } from "../ports/database.js"; import { nowIso } from "./conversations.js"; import type { Logger } from "pino"; @@ -15,7 +15,7 @@ export class RetentionService { let cleaned = 0; const staleCutoff = new Date(Date.now() - STALE_PENDING_THRESHOLD_MS).toISOString(); - const staleResult = this.db + const staleResult = await this.db .prepare( `UPDATE ai_drafts SET status = 'failed', error = 'Draft generation timed out (process may have restarted)', updated_at = ? @@ -30,7 +30,7 @@ export class RetentionService { if (this.retentionDays > 0) { const retentionCutoff = new Date(Date.now() - this.retentionDays * 86400 * 1000).toISOString(); - const deleted = this.db + const deleted = await this.db .prepare( `DELETE FROM ai_drafts WHERE status IN ('sent', 'discarded', 'failed') AND created_at < ?`, @@ -38,7 +38,7 @@ export class RetentionService { .run(retentionCutoff); cleaned += Number(deleted.changes); - const softCleaned = this.db + const softCleaned = await this.db .prepare( `UPDATE ai_drafts SET draft_text = NULL, error = NULL, updated_at = ? @@ -48,7 +48,7 @@ export class RetentionService { cleaned += Number(softCleaned.changes); } - const result = this.db + const result = await this.db .prepare( `UPDATE messages SET text = NULL, raw_payload = NULL diff --git a/src/ports/database.ts b/src/ports/database.ts new file mode 100644 index 0000000..75a48b2 --- /dev/null +++ b/src/ports/database.ts @@ -0,0 +1,21 @@ +export type SqlValue = string | number | bigint | null | Uint8Array; + +export interface StatementResult { + changes: number | bigint; + lastInsertRowid?: number | bigint; +} + +export interface PreparedStatement { + run(...params: SqlValue[]): Promise; + get(...params: SqlValue[]): Promise; + all(...params: SqlValue[]): Promise; +} + +export interface Database { + prepare(sql: string): PreparedStatement; + exec(sql: string): Promise; +} + +export interface ClosableDatabase extends Database { + close(): void; +} diff --git a/src/runtime/config.ts b/src/runtime/config.ts index 0b1d606..97d0dbe 100644 --- a/src/runtime/config.ts +++ b/src/runtime/config.ts @@ -2,6 +2,8 @@ import { existsSync, readFileSync } from "node:fs"; import { resolve } from "node:path"; import { z } from "zod"; +export type ConfigMap = Record; + const booleanFromString = z .string() .optional() @@ -83,7 +85,7 @@ const databaseEnvSchema = z.object({ export type AppConfig = z.infer; export type DatabaseConfig = z.infer; -export function loadConfig(env: NodeJS.ProcessEnv = loadEnv()): AppConfig { +export function loadConfig(env: ConfigMap = loadEnv()): AppConfig { const parsed = envSchema.parse(env); if (parsed.TELEGRAM_UPDATE_MODE === "webhook" && !parsed.TELEGRAM_WEBHOOK_URL) { throw new Error("TELEGRAM_WEBHOOK_URL is required when TELEGRAM_UPDATE_MODE=webhook"); @@ -94,11 +96,11 @@ export function loadConfig(env: NodeJS.ProcessEnv = loadEnv()): AppConfig { return parsed; } -export function loadConfigFromSources(storedEnv: NodeJS.ProcessEnv, env: NodeJS.ProcessEnv = process.env): AppConfig { +export function loadConfigFromSources(storedEnv: ConfigMap, env: ConfigMap = process.env): AppConfig { return loadConfig({ ...storedEnv, ...env }); } -export function configIssues(storedEnv: NodeJS.ProcessEnv, env: NodeJS.ProcessEnv = process.env): string[] { +export function configIssues(storedEnv: ConfigMap, env: ConfigMap = process.env): string[] { try { loadConfigFromSources(storedEnv, env); return []; @@ -108,12 +110,12 @@ export function configIssues(storedEnv: NodeJS.ProcessEnv, env: NodeJS.ProcessEn } } -export function loadDatabaseConfig(env: NodeJS.ProcessEnv = loadEnv()): DatabaseConfig { +export function loadDatabaseConfig(env: ConfigMap = loadEnv()): DatabaseConfig { return databaseEnvSchema.parse(env); } -export function loadEnv(env: NodeJS.ProcessEnv = process.env, envPath = ".env"): NodeJS.ProcessEnv { - const merged: NodeJS.ProcessEnv = { ...env }; +export function loadEnv(env: ConfigMap = process.env, envPath = ".env"): ConfigMap { + const merged: ConfigMap = { ...env }; const fullPath = resolve(process.cwd(), envPath); if (!existsSync(fullPath)) return merged; diff --git a/src/runtime/main.ts b/src/runtime/main.ts index 3710f35..165bfcc 100644 --- a/src/runtime/main.ts +++ b/src/runtime/main.ts @@ -9,14 +9,13 @@ import { startTelegramBot, startTelegramPolling, } from "../channels/telegram/bot.js"; -import { sweepExpiredConversations } from "../domain/conversation-expiry.js"; -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 { AuditService } from "../domain/audit.js"; import { AppSettingsService } from "../domain/app-settings.js"; +import { runConversationExpiryJob, runMessageRetentionJob } from "./maintenance.js"; import { ensureSetupToken, startWebConsole } from "./web-console.js"; import type { IncomingMessage, ServerResponse } from "node:http"; @@ -26,7 +25,7 @@ const databaseConfig = loadDatabaseConfig(); const handle = createDb(databaseConfig.DATABASE_URL); await migrate(handle.client); const settings = new AppSettingsService(handle.db); -const setupToken = ensureSetupToken(settings); +const setupToken = await ensureSetupToken(settings); if (setupToken) { logger.info({ setupToken }, "Open the web console and use this setup token to finish InboxBridge configuration."); } @@ -42,13 +41,12 @@ let restartQueue = Promise.resolve(); async function runExpirySweep(): Promise { if (!activeBot) return; - const config = loadConfigFromSources(settings.all()); - const cleaned = await sweepExpiredConversations({ + const config = loadConfigFromSources(await settings.all()); + const cleaned = await runConversationExpiryJob({ api: activeBot.api, db: handle.db, - messageRetentionDays: config.MESSAGE_RETENTION_DAYS, - defaultConversationRetentionDays: config.DEFAULT_CONVERSATION_RETENTION_DAYS, - logger: logger.child({ module: "conversation-expiry" }), + config, + logger, }); if (cleaned > 0) { logger.info({ cleaned }, "Expired conversations cleaned."); @@ -56,9 +54,13 @@ async function runExpirySweep(): Promise { } async function runMessageRetentionSweep(): Promise { - const config = loadConfigFromSources(settings.all()); - const retention = new RetentionService(handle.db, config.MESSAGE_RETENTION_DAYS, logger.child({ module: "retention" })); - const cleaned = await retention.cleanupExpired(); + const config = loadConfigFromSources(await settings.all()); + const cleaned = await runMessageRetentionJob({ + api: activeBot?.api ?? ({} as never), + db: handle.db, + config, + logger, + }); if (cleaned > 0) { logger.info({ cleaned }, "Message retention sweep cleaned expired message content."); } else { @@ -74,14 +76,14 @@ async function restartRuntime(): Promise { async function restartRuntimeUnlocked(): Promise { await stopRuntime(); - const issues = configIssues(settings.all()); + const issues = configIssues(await settings.all()); if (issues.length > 0) { lastRuntimeError = issues.join("; "); logger.warn({ issues }, "InboxBridge bot is waiting for complete configuration."); return; } - const config = loadConfigFromSources(settings.all()); + const config = loadConfigFromSources(await settings.all()); const bot = createTelegramBot(config, handle.db, logger); activeBot = bot; pollingBot = config.TELEGRAM_UPDATE_MODE === "polling"; @@ -250,9 +252,9 @@ process.once("SIGTERM", () => gracefulShutdown("SIGTERM")); await startWebConsole({ settings, port: databaseConfig.WEB_CONSOLE_PORT, - getStatus: () => ({ + getStatus: async () => ({ bot: activeBot ? "running" : "stopped", - issues: lastRuntimeError ? [lastRuntimeError] : configIssues(settings.all()), + issues: lastRuntimeError ? [lastRuntimeError] : configIssues(await settings.all()), }), onConfigSaved: restartRuntime, telegramWebhook: (req, res) => { @@ -263,22 +265,23 @@ await startWebConsole({ res.end("Telegram webhook is not configured."); return Promise.resolve(); }, - dbHealthCheck: () => { - handle.db.prepare("SELECT 1").get(); + dbHealthCheck: async () => { + await handle.db.prepare("SELECT 1").get(); return true; }, - collectMetrics: () => { + collectMetrics: async () => { + const config = loadConfigFromSources(await settings.all()); const conversations = new ConversationService( handle.db, - Number(loadConfigFromSources(settings.all()).MESSAGE_RETENTION_DAYS) || 30, - loadConfigFromSources(settings.all()).DEFAULT_CONVERSATION_RETENTION_DAYS ?? 30, + Number(config.MESSAGE_RETENTION_DAYS) || 30, + config.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(); + const aiDrafts = new AiDraftService(handle.db, conversations, config); + const msgStats = await conversations.messageStats(); + const convStats = await conversations.conversationStats(); + const delStats = await deliveries.stats(); + const draftStats = await aiDrafts.stats(); return { messages: { inbound_total: msgStats.inbound, @@ -304,8 +307,8 @@ await startWebConsole({ timestamp: new Date().toISOString(), }; }, - collectOperationsOverview: () => { - const config = loadConfigFromSources(settings.all()); + collectOperationsOverview: async () => { + const config = loadConfigFromSources(await settings.all()); const conversations = new ConversationService( handle.db, config.MESSAGE_RETENTION_DAYS, @@ -313,10 +316,10 @@ await startWebConsole({ ); const deliveries = new DeliveryService(handle.db); const aiDrafts = new AiDraftService(handle.db, conversations, config); - const msgStats = conversations.messageStats(); - const convStats = conversations.conversationStats(); - const delStats = deliveries.stats(); - const draftStats = aiDrafts.stats(); + const msgStats = await conversations.messageStats(); + const convStats = await conversations.conversationStats(); + const delStats = await deliveries.stats(); + const draftStats = await aiDrafts.stats(); return { messages: { inboundTotal: msgStats.inbound, outboundTotal: msgStats.outbound, internalTotal: msgStats.internal }, deliveries: { pending: delStats.pending, sent: delStats.sent, failed: delStats.failed, permanentFailure: delStats.permanentFailure }, @@ -325,13 +328,14 @@ await startWebConsole({ uptimeSeconds: Math.round(process.uptime()), }; }, - listConversations: (opts) => { + listConversations: async (opts) => { + const config = loadConfigFromSources(await settings.all()); const conversations = new ConversationService( handle.db, - loadConfigFromSources(settings.all()).MESSAGE_RETENTION_DAYS, - loadConfigFromSources(settings.all()).DEFAULT_CONVERSATION_RETENTION_DAYS, + config.MESSAGE_RETENTION_DAYS, + config.DEFAULT_CONVERSATION_RETENTION_DAYS, ); - const result = conversations.listConversations({ + const result = await conversations.listConversations({ status: opts.status === "open" || opts.status === "closed" ? opts.status : undefined, assignedTo: opts.assignedTo || undefined, limit: opts.pageSize, @@ -353,9 +357,9 @@ await startWebConsole({ total: result.total, }; }, - listFailedDeliveries: (opts) => { + listFailedDeliveries: async (opts) => { const deliveries = new DeliveryService(handle.db); - const result = deliveries.listFailedDeliveries({ + const result = await deliveries.listFailedDeliveries({ limit: opts.pageSize, offset: (opts.page - 1) * opts.pageSize, }); @@ -376,11 +380,11 @@ await startWebConsole({ }, scheduleRetry: async (deliveryId: number) => { const deliveries = new DeliveryService(handle.db); - deliveries.scheduleRetry(deliveryId); + await deliveries.scheduleRetry(deliveryId); }, - listAuditLogs: (opts) => { + listAuditLogs: async (opts) => { const audit = new AuditService(handle.db); - const result = audit.list({ + const result = await audit.list({ adminId: opts.adminId || undefined, action: opts.action || undefined, limit: opts.pageSize, @@ -398,13 +402,14 @@ await startWebConsole({ total: result.total, }; }, - searchMessages: (opts) => { + searchMessages: async (opts) => { + const config = loadConfigFromSources(await settings.all()); const conversations = new ConversationService( handle.db, - loadConfigFromSources(settings.all()).MESSAGE_RETENTION_DAYS, - loadConfigFromSources(settings.all()).DEFAULT_CONVERSATION_RETENTION_DAYS, + config.MESSAGE_RETENTION_DAYS, + config.DEFAULT_CONVERSATION_RETENTION_DAYS, ); - const result = conversations.searchMessages({ + const result = await conversations.searchMessages({ query: opts.query, limit: opts.pageSize, offset: (opts.page - 1) * opts.pageSize, diff --git a/src/runtime/maintenance.ts b/src/runtime/maintenance.ts new file mode 100644 index 0000000..e3e746a --- /dev/null +++ b/src/runtime/maintenance.ts @@ -0,0 +1,47 @@ +import type { Bot } from "grammy"; +import type { Logger } from "pino"; +import { sweepExpiredConversations as defaultSweepExpiredConversations } from "../domain/conversation-expiry.js"; +import { RetentionService } from "../domain/retention.js"; +import type { Database } from "../ports/database.js"; +import type { AppConfig } from "./config.js"; + +export interface MaintenanceJobInput { + api: Bot["api"]; + db: Database; + config: AppConfig; + logger: Logger; +} + +export interface MaintenanceJobsInput extends MaintenanceJobInput { + sweepExpiredConversations?: (input: MaintenanceJobInput) => Promise; + cleanupExpiredMessages?: (input: MaintenanceJobInput) => Promise; +} + +export interface MaintenanceJobSummary { + expiredConversations: number; + expiredMessages: number; +} + +export async function runConversationExpiryJob(input: MaintenanceJobInput): Promise { + return defaultSweepExpiredConversations({ + api: input.api, + db: input.db, + messageRetentionDays: input.config.MESSAGE_RETENTION_DAYS, + defaultConversationRetentionDays: input.config.DEFAULT_CONVERSATION_RETENTION_DAYS, + logger: input.logger.child({ module: "conversation-expiry" }), + }); +} + +export async function runMessageRetentionJob(input: MaintenanceJobInput): Promise { + return new RetentionService( + input.db, + input.config.MESSAGE_RETENTION_DAYS, + input.logger.child({ module: "retention" }), + ).cleanupExpired(); +} + +export async function runMaintenanceJobs(input: MaintenanceJobsInput): Promise { + const expiredConversations = await (input.sweepExpiredConversations ?? runConversationExpiryJob)(input); + const expiredMessages = await (input.cleanupExpiredMessages ?? runMessageRetentionJob)(input); + return { expiredConversations, expiredMessages }; +} diff --git a/src/runtime/web-console.ts b/src/runtime/web-console.ts index 7cd4181..7f97282 100644 --- a/src/runtime/web-console.ts +++ b/src/runtime/web-console.ts @@ -271,29 +271,31 @@ export const auditActionOptions = [ export interface WebConsoleOptions { settings: AppSettingsService; port: number; - getStatus: () => ConsoleStatus; + getStatus: () => ConsoleStatus | Promise; onConfigSaved: () => Promise; telegramWebhook?: (req: IncomingMessage, res: ServerResponse) => Promise; - dbHealthCheck: () => boolean; - collectMetrics: () => MetricsSnapshot; - collectOperationsOverview: () => OperationsOverview; - listConversations: (opts: { page: number; status?: string; assignedTo?: string; pageSize: number }) => { items: ConversationListItemView[]; total: number }; - listFailedDeliveries: (opts: { page: number; pageSize: number }) => { items: DeliveryView[]; total: number }; + dbHealthCheck: () => boolean | Promise; + collectMetrics: () => MetricsSnapshot | Promise; + collectOperationsOverview: () => OperationsOverview | Promise; + listConversations: (opts: { page: number; status?: string; assignedTo?: string; pageSize: number }) => { items: ConversationListItemView[]; total: number } | Promise<{ items: ConversationListItemView[]; total: number }>; + listFailedDeliveries: (opts: { page: number; pageSize: number }) => { items: DeliveryView[]; total: number } | Promise<{ items: DeliveryView[]; total: number }>; scheduleRetry: (deliveryId: number) => Promise; - listAuditLogs: (opts: { page: number; adminId?: string; action?: string; pageSize: number }) => { items: AuditLogView[]; total: number }; - searchMessages: (opts: { query: string; page: number; pageSize: number }) => { items: MessageSearchView[]; total: number }; + listAuditLogs: (opts: { page: number; adminId?: string; action?: string; pageSize: number }) => { items: AuditLogView[]; total: number } | Promise<{ items: AuditLogView[]; total: number }>; + searchMessages: (opts: { query: string; page: number; pageSize: number }) => { items: MessageSearchView[]; total: number } | Promise<{ items: MessageSearchView[]; total: number }>; } type SessionKind = "password" | "setup"; +type OperationsTab = "overview" | "conversations" | "deliveries" | "audit" | "search"; +export type WebConsoleSessionStore = Map; class FormBodyTooLargeError extends Error {} -export function ensureSetupToken(settings: AppSettingsService): string | undefined { - if (settings.get(passwordHashKey)) return undefined; - const existing = settings.get(setupTokenKey); +export async function ensureSetupToken(settings: AppSettingsService): Promise { + if (await settings.get(passwordHashKey)) return undefined; + const existing = await settings.get(setupTokenKey); if (existing) return existing; const token = randomBytes(16).toString("hex"); - settings.setMany({ [setupTokenKey]: token }); + await settings.setMany({ [setupTokenKey]: token }); return token; } @@ -311,11 +313,11 @@ export async function startWebConsole(options: WebConsoleOptions): Promise 0) { await options.scheduleRetry(deliveryId); } - redirect(res, "/operations/deliveries?action=retryed"); - return; - } - - if (url.pathname === "/operations/audit" && req.method === "GET") { - const pageNum = Math.max(1, Number(url.searchParams.get("page") ?? "1")); - const adminId = url.searchParams.get("admin_id") ?? undefined; - const action = url.searchParams.get("action") ?? undefined; - const data = options.listAuditLogs({ page: pageNum, adminId, action, pageSize: 50 }); - renderAuditLogs(res, data, pageNum, adminId, action); - return; - } - - if (url.pathname === "/operations/search" && req.method === "GET") { - const query = url.searchParams.get("q") ?? ""; - const pageNum = Math.max(1, Number(url.searchParams.get("page") ?? "1")); - if (!query.trim()) { - renderSearch(res, { items: [], total: 0 }, query, pageNum); - return; - } - const data = options.searchMessages({ query: query.trim(), page: pageNum, pageSize: 50 }); - renderSearch(res, data, query, pageNum); - return; - } - - if (url.pathname === "/" && req.method === "GET") { - renderDashboard(res, options.settings, options.getStatus(), url.searchParams.get("saved") === "1"); + redirect(res, "/operations/deliveries?retryed=1"); return; } if (url.pathname === "/config" && req.method === "POST") { const form = await readForm(req); - const values = configValuesFromForm(options.settings, form); + const values = await configValuesFromForm(options.settings, form); const password = form.get("WEB_CONSOLE_PASSWORD")?.trim(); if (sessionKind === "setup" && !password) { send(res, 400, "text/plain", "首次配置必须设置控制台密码。"); @@ -435,9 +389,25 @@ export async function startWebConsole(options: WebConsoleOptions): Promise { + try { + const url = new URL(request.url); + const res = new FetchResponseSink(); + + if (url.pathname === "/healthz" && request.method === "GET") { + let dbReachable = false; + try { + dbReachable = await options.dbHealthCheck(); + } catch { + dbReachable = false; + } + const botRunning = (await options.getStatus()).bot === "running"; + const healthy = dbReachable && botRunning; + return jsonResponse( + { + status: healthy ? "ok" : "degraded", + bot: botRunning ? "running" : "stopped", + db: dbReachable ? "reachable" : "unreachable", + }, + healthy ? 200 : 503, + ); + } + + if (url.pathname === "/login" && request.method === "GET") { + await renderLogin(res.asServerResponse(), options.settings); + return res.toResponse(); + } + + const sessionKind = authenticatedFetchSessionKind(request, sessions); + if (!sessionKind) return redirectResponse("/login"); + + return textResponse("页面不存在", 404); + } catch (error) { + if (error instanceof FormBodyTooLargeError) return textResponse("请求体过大。", 413); + return textResponse(`控制台处理失败:${error instanceof Error ? error.message : String(error)}`, 500); + } +} + +class FetchResponseSink { + statusCode = 200; + private readonly headers = new Headers(); + private body = ""; + + asServerResponse(): ServerResponse { + const sink = this; + return { + setHeader: (name: string, value: number | string | readonly string[]) => { + if (Array.isArray(value)) sink.headers.set(name, value.join(", ")); + else sink.headers.set(name, String(value)); + }, + end: (body?: string | Buffer) => { + if (body !== undefined) sink.body = Buffer.isBuffer(body) ? body.toString("utf8") : String(body); + }, + get statusCode() { + return sink.statusCode; + }, + set statusCode(value: number) { + sink.statusCode = value; + }, + } as unknown as ServerResponse; + } + + toResponse(): Response { + return new Response(this.body, { status: this.statusCode, headers: this.headers }); + } +} + +function authenticatedFetchSessionKind(request: Request, sessions: WebConsoleSessionStore): SessionKind | undefined { + const cookie = request.headers.get("cookie") ?? ""; + const token = cookie + .split(";") + .map((part) => part.trim()) + .find((part) => part.startsWith(`${sessionCookie}=`)) + ?.slice(sessionCookie.length + 1); + return token ? sessions.get(token) : undefined; +} + +function jsonResponse(body: unknown, status: number): Response { + return new Response(JSON.stringify(body), { + status, + headers: { "content-type": "application/json" }, + }); +} + +function textResponse(body: string, status: number): Response { + return new Response(body, { status, headers: { "content-type": "text/plain" } }); +} + +function redirectResponse(location: string): Response { + return new Response(null, { status: 302, headers: { location } }); +} + +async function renderLogin(res: ServerResponse, settings: AppSettingsService, error = ""): Promise { + const hasPassword = Boolean(await settings.get(passwordHashKey)); const label = hasPassword ? "控制台密码" : "首次设置令牌"; const name = hasPassword ? "password" : "setupToken"; send( @@ -488,28 +555,64 @@ function renderLogin(res: ServerResponse, settings: AppSettingsService, error = - +
+ +
`), ); } -function renderDashboard(res: ServerResponse, settings: AppSettingsService, status: ConsoleStatus, saved: boolean): void { - const stored = settings.all(); - const needsPassword = !settings.get(passwordHashKey); +const configGroupSlugs: Array<{ slug: string; title: string }> = [ + { slug: "security", title: "访问安全" }, + { slug: "telegram", title: "Telegram 基础配置" }, + { slug: "runtime", title: "运行方式" }, + { slug: "retention", title: "数据保留与自动清理" }, + { slug: "ratelimit", title: "限流设置" }, + { slug: "delivery", title: "投递可靠性" }, + { slug: "ai-drafts", title: "AI 草稿" }, +]; + +async function renderOverview( + res: ServerResponse, + options: WebConsoleOptions, + saved: boolean, +): Promise { + const status = await options.getStatus(); + const stored = await options.settings.all(); const completedRequired = ["TELEGRAM_BOT_TOKEN", "TELEGRAM_MANAGEMENT_CHAT_ID", "TELEGRAM_ADMIN_USER_IDS"].filter( (key) => Boolean(stored[key]), ).length; const issueList = status.issues.length ? `
    ${status.issues.map((issue) => `
  • ${translateIssue(issue)}
  • `).join("")}
` : `

配置完整,bot 可以正常启动。

`; - const groups = fieldGroups.map((group) => renderGroup(group, stored)).join("\n"); - const navigation = ["访问安全", ...fieldGroups.map((group) => group.title)] - .map( - (title, index) => - ``, - ) - .join(""); const statusRunning = status.bot === "running"; + + let opsMetrics = ""; + try { + const data = await options.collectOperationsOverview(); + opsMetrics = ` +
+

消息

+
入站${data.messages.inboundTotal}
+
出站${data.messages.outboundTotal}
+
+

投递

+
sent${data.deliveries.sent}
+
failed${data.deliveries.failed}
+
+

会话

+
open${data.conversations.open}
+
closed${data.conversations.closed}
+
+
`; + } catch { + opsMetrics = `

Bot 未完成配置,运维数据暂不可用。

`; + } + + const savedBanner = saved + ? `` + : ""; + send( res, 200, @@ -518,12 +621,12 @@ function renderDashboard(res: ServerResponse, settings: AppSettingsService, stat

InboxBridge

-

控制台

-

集中管理 bot 连接、管理员白名单、数据保留、限流和 AI 草稿。常规部署无需手写环境变量。

+

控制台概览

+

运行状态、配置进度和关键指标一览。

${statusRunning ? "Bot 运行中" : "Bot 待配置"}
- ${saved ? `` : ""} + ${savedBanner}
-
-
-
+
+
+
-

访问安全

-

${needsPassword ? "设置控制台密码" : "修改控制台密码"}

-

${needsPassword ? "首次配置必须设置密码,后续登录使用该密码。" : "留空表示保持现有密码。"}

+

关键指标

+

实时数据

- 本地保存 -
-
-

控制台密码只保存在本地 SQLite 数据库中,用于保护配置页面。

- - + 查看详情
+ ${opsMetrics}
- ${groups} -
-
- 保存后立即生效 -

系统会重新读取配置;配置完整时 bot 会自动启动。

-
- -
- +
`), ); } -function renderGroup(group: (typeof fieldGroups)[number], stored: NodeJS.ProcessEnv): string { - const index = fieldGroups.indexOf(group) + 1; - return ``; +
+

控制台密码只保存在本地 SQLite 数据库中,用于保护配置页面。

+ + +
`; + } else { + const group = fieldGroups[currentIndex - 1]; + formBody = ` +
+
+

配置分组 ${currentIndex}

+

${escapeHtml(group.title)}

+

${escapeHtml(group.description)}

+
+ ${group.fields.length} 项 +
+
+ ${group.fields.map((field) => renderField(field, stored[field.key] ?? "")).join("\n")} +
`; + } + + const savedBanner = saved + ? `` + : ""; + + const pageTitle = configGroupSlugs[currentIndex].title; + + send( + res, + 200, + "text/html; charset=utf-8", + page(`配置 · ${pageTitle} - InboxBridge`, ` + +
+
+ 概览 +

配置仪表盘

+

${escapeHtml(pageTitle)}

+

管理 bot 连接、管理员白名单、数据保留、限流和 AI 草稿。

+
+ ${statusRunning ? "Bot 运行中" : "Bot 待配置"} +
+ ${savedBanner} +
+ +
+ ${formBody} +
+
+
+ 保存后立即生效 +

系统会重新读取配置;配置完整时 bot 会自动启动。

+
+ +
+
`), + ); +} + +const opsRoutes: Array<{ pattern: RegExp; key: OperationsTab }> = [ + { pattern: /^\/operations\/?$/, key: "overview" }, + { pattern: /^\/operations\/overview\/?$/, key: "overview" }, + { pattern: /^\/operations\/conversations\/?$/, key: "conversations" }, + { pattern: /^\/operations\/deliveries\/?$/, key: "deliveries" }, + { pattern: /^\/operations\/audit\/?$/, key: "audit" }, + { pattern: /^\/operations\/search\/?$/, key: "search" }, +]; + +async function renderOperationsPage( + res: ServerResponse, + options: WebConsoleOptions, + pathname: string, + searchParams: URLSearchParams, +): Promise { + const route = opsRoutes.find((r) => r.pattern.test(pathname)); + if (!route) { + send(res, 404, "text/plain", "运维页面不存在"); + return; + } + const tab = route.key; + const retryed = searchParams.get("retryed") === "1"; + + let opsBody: string; + try { + if (tab === "overview") { + opsBody = renderOperationsOverviewBody(await options.collectOperationsOverview(), retryed); + } else if (tab === "conversations") { + const pageNum = Math.max(1, Number(searchParams.get("page") ?? "1")); + const convStatus = searchParams.get("status") ?? undefined; + const assignedTo = searchParams.get("assigned_to") ?? undefined; + const data = await options.listConversations({ page: pageNum, status: convStatus, assignedTo, pageSize: 50 }); + opsBody = renderConversationsBody(data, pageNum, convStatus, assignedTo); + } else if (tab === "deliveries") { + const pageNum = Math.max(1, Number(searchParams.get("page") ?? "1")); + const data = await options.listFailedDeliveries({ page: pageNum, pageSize: 50 }); + opsBody = renderDeliveriesBody(data, pageNum); + } else if (tab === "audit") { + const pageNum = Math.max(1, Number(searchParams.get("page") ?? "1")); + const adminId = searchParams.get("admin_id") ?? undefined; + const action = searchParams.get("action") ?? undefined; + const data = await options.listAuditLogs({ page: pageNum, adminId, action, pageSize: 50 }); + opsBody = renderAuditLogsBody(data, pageNum, adminId, action); + } else { + const query = searchParams.get("q") ?? ""; + const pageNum = Math.max(1, Number(searchParams.get("page") ?? "1")); + const data = query.trim() + ? await options.searchMessages({ query: query.trim(), page: pageNum, pageSize: 50 }) + : { items: [], total: 0 }; + opsBody = renderSearchBody(data, query, pageNum); + } + } catch { + opsBody = `
Bot 未完成配置,运维数据暂不可用。请先在配置控制台填写 Bot Token、管理群 ID 和管理员白名单。
`; + } + + const navItems: Array<{ key: OperationsTab; href: string; label: string }> = [ + { key: "overview", href: "/operations", label: "概览" }, + { key: "conversations", href: "/operations/conversations", label: "会话" }, + { key: "deliveries", href: "/operations/deliveries", label: "投递" }, + { key: "audit", href: "/operations/audit", label: "审计" }, + { key: "search", href: "/operations/search", label: "搜索" }, + ]; + const nav = navItems + .map((item) => `${item.label}`) + .join("\n"); + const pageTitle = navItems.find((n) => n.key === tab)?.label ?? "运维"; + + send( + res, + 200, + "text/html; charset=utf-8", + page(`运维 · ${pageTitle} - InboxBridge`, ` + +
+
+ 概览 +

运维仪表盘

+

${escapeHtml(pageTitle)}

+

监控会话、投递、审计与消息检索。

+
+
+
${opsBody}
+ `), + ); } function renderField(field: FieldMeta, value: string): string { @@ -630,8 +885,8 @@ function translateIssue(issue: string): string { return escapeHtml(issue); } -function configValuesFromForm(settings: AppSettingsService, form: URLSearchParams): Record { - const current = settings.all(); +async function configValuesFromForm(settings: AppSettingsService, form: URLSearchParams): Promise> { + const current = await settings.all(); const values: Record = {}; for (const key of editableConfigKeys) { const value = form.get(key)?.trim() ?? ""; @@ -641,10 +896,11 @@ function configValuesFromForm(settings: AppSettingsService, form: URLSearchParam return values; } -function loginSessionKind(settings: AppSettingsService, form: URLSearchParams): SessionKind | undefined { - const hash = settings.get(passwordHashKey); +async function loginSessionKind(settings: AppSettingsService, form: URLSearchParams): Promise { + const hash = await settings.get(passwordHashKey); if (hash) return verifyPassword(form.get("password") ?? "", hash) ? "password" : undefined; - return settings.get(setupTokenKey) && form.get("setupToken") === settings.get(setupTokenKey) ? "setup" : undefined; + const setupToken = await settings.get(setupTokenKey); + return setupToken && form.get("setupToken") === setupToken ? "setup" : undefined; } function authenticatedSessionKind(req: IncomingMessage, sessions: Map): SessionKind | undefined { @@ -685,27 +941,27 @@ function verifyPassword(password: string, stored: string): boolean { function page(title: string, body: string): string { const fingerprint = createHash("sha256").update(title).digest("hex").slice(0, 8); - return `${escapeHtml(title)}
${body}
InboxBridge ${fingerprint}