From 667c6a6e6dfde5751c49386c619c58e352be438d Mon Sep 17 00:00:00 2001 From: Nikolay Unuchek Date: Sat, 17 Oct 2026 15:48:04 +0300 Subject: [PATCH] =?UTF-8?q?feat(api):=20Telegram=20=E2=80=94=20=D1=82?= =?UTF-8?q?=D0=B5=D0=BC=D1=8B,=20=D0=BE=D1=87=D0=B5=D1=80=D0=B5=D0=B4?= =?UTF-8?q?=D1=8C=20=D0=BF=D0=BE=D0=B2=D1=82=D0=BE=D1=80=D0=BE=D0=B2,=20?= =?UTF-8?q?=C2=AB=D0=A1=D0=B8=D1=81=D1=82=D0=B5=D0=BC=D0=B0=C2=BB,=20/setu?= =?UTF-8?q?p?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit telegram/client: Bot API на fetch с таймаутом 5 с; ошибки делятся на повторяемые (сеть, таймаут, 429, 5xx) и конфигурационные (400/403…), токен не попадает ни в тексты ошибок, ни в tg_last_error. format: заявка и уточнение по шаблону плана, HTML с экранированием, источник из utm / referrer, первый визит. outbox: Notifier с отправкой после ответа и тиком раз в минуту, пауза 1, 2, 4… мин, уточнение только после своей заявки, после 10 попыток — в «Систему». system: тема «Система» с троттлингом ×N за период. bot: long polling, только TELEGRAM_ADMIN_ID, /setup создаёт темы и присылает id в личку. config: всё окружение в одном месте. --- apps/api/src/app.test.ts | 5 +- apps/api/src/app.ts | 3 + apps/api/src/config.ts | 59 +++++++++ apps/api/src/index.ts | 68 +++++++--- apps/api/src/telegram/bot.test.ts | 129 +++++++++++++++++++ apps/api/src/telegram/bot.ts | 162 ++++++++++++++++++++++++ apps/api/src/telegram/client.test.ts | 110 ++++++++++++++++ apps/api/src/telegram/client.ts | 143 +++++++++++++++++++++ apps/api/src/telegram/fake.ts | 36 ++++++ apps/api/src/telegram/format.test.ts | 78 ++++++++++++ apps/api/src/telegram/format.ts | 77 +++++++++++ apps/api/src/telegram/outbox.test.ts | 158 +++++++++++++++++++++++ apps/api/src/telegram/outbox.ts | 183 +++++++++++++++++++++++++++ apps/api/src/telegram/system.test.ts | 63 +++++++++ apps/api/src/telegram/system.ts | 64 ++++++++++ 15 files changed, 1316 insertions(+), 22 deletions(-) create mode 100644 apps/api/src/config.ts create mode 100644 apps/api/src/telegram/bot.test.ts create mode 100644 apps/api/src/telegram/bot.ts create mode 100644 apps/api/src/telegram/client.test.ts create mode 100644 apps/api/src/telegram/client.ts create mode 100644 apps/api/src/telegram/fake.ts create mode 100644 apps/api/src/telegram/format.test.ts create mode 100644 apps/api/src/telegram/format.ts create mode 100644 apps/api/src/telegram/outbox.test.ts create mode 100644 apps/api/src/telegram/outbox.ts create mode 100644 apps/api/src/telegram/system.test.ts create mode 100644 apps/api/src/telegram/system.ts diff --git a/apps/api/src/app.test.ts b/apps/api/src/app.test.ts index 532ebb5..ee6937a 100644 --- a/apps/api/src/app.test.ts +++ b/apps/api/src/app.test.ts @@ -122,10 +122,13 @@ describe('POST /api/lead', () => { }); it('500 server: исключение при записи', async () => { - const { post, db } = setup(); + const errors: unknown[] = []; + const { post, db } = setup({ onServerError: (err) => void errors.push(err) }); db.close(); const res = await post(validBody()); expect(res.status).toBe(500); expect(await res.json()).toEqual({ ok: false, error: 'server' }); + // onServerError → «Система» + expect(errors).toHaveLength(1); }); }); diff --git a/apps/api/src/app.ts b/apps/api/src/app.ts index 5f6302d..014918c 100644 --- a/apps/api/src/app.ts +++ b/apps/api/src/app.ts @@ -17,6 +17,8 @@ export type AppDeps = { /** Адреса, от которых верим X-Real-IP, помимо loopback. */ trustedProxies?: readonly string[]; version?: string; + /** 500: сообщить в «Систему». Ответ клиенту от этого не зависит. */ + onServerError?: (err: unknown) => void; }; export const MAX_BODY_BYTES = 16 * 1024; @@ -41,6 +43,7 @@ export function createApp(deps: AppDeps) { app.onError((err) => { console.error(err); + deps.onServerError?.(err); return reply({ ok: false, error: 'server' }, 500); }); diff --git a/apps/api/src/config.ts b/apps/api/src/config.ts new file mode 100644 index 0000000..4da7140 --- /dev/null +++ b/apps/api/src/config.ts @@ -0,0 +1,59 @@ +// Всё окружение api читается здесь. Telegram-переменные необязательны: без них api +// принимает заявки, а отправки копятся в pending до настройки. +import { DEFAULT_DB_PATH } from './db/index.js'; + +export type TelegramConfig = { + token?: string; + /** id группы, у супергрупп отрицательный: -100… */ + chatId?: number; + leadsThreadId?: number; + systemThreadId?: number; + adminId?: number; +}; + +export type Config = { + port: number; + dbPath: string; + version: string; + trustedProxies: string[]; + telegram: TelegramConfig; +}; + +type Env = Record; + +function optional(env: Env, name: string): string | undefined { + const value = env[name]?.trim(); + return value ? value : undefined; +} + +function optionalInt(env: Env, name: string): number | undefined { + const raw = optional(env, name); + if (raw === undefined) return undefined; + if (!/^-?\d+$/.test(raw) || !Number.isSafeInteger(Number(raw))) { + throw new Error(`${name}: ожидается целое число`); + } + return Number(raw); +} + +export function loadConfig(env: Env = process.env): Config { + const port = optionalInt(env, 'PORT') ?? 3001; + const token = optional(env, 'TELEGRAM_BOT_TOKEN'); + // Значение токена в текст ошибки не попадает — только факт, что формат не тот. + if (token !== undefined && !/^\d+:[\w-]+$/.test(token)) throw new Error('TELEGRAM_BOT_TOKEN: неверный формат'); + return { + port, + dbPath: optional(env, 'DB_PATH') ?? DEFAULT_DB_PATH, + version: optional(env, 'APP_VERSION') ?? 'dev', + trustedProxies: (optional(env, 'TRUSTED_PROXIES') ?? '') + .split(',') + .map((s) => s.trim()) + .filter(Boolean), + telegram: { + token, + chatId: optionalInt(env, 'TELEGRAM_CHAT_ID'), + leadsThreadId: optionalInt(env, 'TELEGRAM_LEADS_THREAD_ID'), + systemThreadId: optionalInt(env, 'TELEGRAM_SYSTEM_THREAD_ID'), + adminId: optionalInt(env, 'TELEGRAM_ADMIN_ID'), + }, + }; +} diff --git a/apps/api/src/index.ts b/apps/api/src/index.ts index 35968c1..e9bd156 100644 --- a/apps/api/src/index.ts +++ b/apps/api/src/index.ts @@ -1,45 +1,71 @@ import { serve } from '@hono/node-server'; import { createApp } from './app.js'; -import { DEFAULT_DB_PATH, openDb } from './db/index.js'; -import { noopNotifier } from './notifier.js'; +import { loadConfig } from './config.js'; +import { openDb, type Db } from './db/index.js'; +import { noopNotifier, type Notifier } from './notifier.js'; import { createRateLimiter } from './rateLimit.js'; +import { createBot, type Bot } from './telegram/bot.js'; +import { createTelegramClient } from './telegram/client.js'; +import { createOutbox, type Outbox } from './telegram/outbox.js'; +import { createSystemReporter, shortError } from './telegram/system.js'; -const port = Number(process.env.PORT ?? 3001); -const db = openDb(process.env.DB_PATH ?? DEFAULT_DB_PATH); +const config = loadConfig(); +const tg = config.telegram; +const now = () => new Date(); +const client = tg.token ? createTelegramClient({ token: tg.token }) : undefined; +const system = createSystemReporter({ client, chatId: tg.chatId, threadId: tg.systemThreadId, now }); + +let db: Db; +try { + db = openDb(config.dbPath); +} catch (err) { + // Best effort: база не открылась или миграция упала — сообщаем и выходим, docker перезапустит. + await system.report('startup', `Сбой старта api ${config.version}: ${shortError(err)}`); + process.exit(1); +} + +let outbox: Outbox | undefined; +let bot: Bot | undefined; +if (client) { + outbox = createOutbox({ getDb: () => db, client, chatId: tg.chatId, threadId: tg.leadsThreadId, now, system }); + bot = createBot({ client, telegram: tg, system }); +} +const notifier: Notifier = outbox ?? noopNotifier; const app = createApp({ db, - now: () => new Date(), + now, rateLimiter: createRateLimiter(), - notifier: noopNotifier, - // В docker сокет nginx приходит не с 127.0.0.1, а со шлюза сети контейнера — его задаём явно. - trustedProxies: (process.env.TRUSTED_PROXIES ?? '') - .split(',') - .map((s) => s.trim()) - .filter(Boolean), - version: process.env.APP_VERSION, + notifier, + trustedProxies: config.trustedProxies, + version: config.version, + onServerError: (err) => void system.report(`500 ${err instanceof Error ? err.message : ''}`.slice(0, 120), `500: ${shortError(err)}`), }); -const server = serve({ fetch: app.fetch, port }, (info) => { +const server = serve({ fetch: app.fetch, port: config.port }, (info) => { console.log(`api listening on :${info.port}`); + outbox?.start(); + bot?.start(); + void system.report('started', `api запущен, версия ${config.version}`); }); let stopping = false; -function shutdown(signal: string) { +async function shutdown(signal: string) { if (stopping) return; stopping = true; console.log(`${signal}: останавливаемся`); + setTimeout(() => { + console.error('не дождались остановки, выходим'); + process.exit(1); + }, 10_000).unref(); + await bot?.stop(); + await outbox?.stop(); // Сначала дожидаемся текущих запросов, потом закрываем базу — иначе запись оборвётся на полуслове. server.close(() => { db.close(); process.exit(0); }); - setTimeout(() => { - console.error('не дождались запросов, выходим'); - db.close(); - process.exit(1); - }, 10_000).unref(); } -process.on('SIGTERM', () => shutdown('SIGTERM')); -process.on('SIGINT', () => shutdown('SIGINT')); +process.on('SIGTERM', () => void shutdown('SIGTERM')); +process.on('SIGINT', () => void shutdown('SIGINT')); diff --git a/apps/api/src/telegram/bot.test.ts b/apps/api/src/telegram/bot.test.ts new file mode 100644 index 0000000..82038e9 --- /dev/null +++ b/apps/api/src/telegram/bot.test.ts @@ -0,0 +1,129 @@ +import { describe, expect, it } from 'vitest'; +import type { TelegramConfig } from '../config.js'; +import { createBot, LEADS_TOPIC, parseCommand, SYSTEM_TOPIC } from './bot.js'; +import { TelegramError, type TgUpdate } from './client.js'; +import { fakeTelegram } from './fake.js'; +import type { SystemReporter } from './system.js'; + +const ADMIN = 42; +const GROUP = -1001234; + +function groupMessage(text: string, fromId = ADMIN): TgUpdate { + return { + update_id: 1, + message: { + message_id: 10, + from: { id: fromId, is_bot: false, first_name: 'A' }, + chat: { id: GROUP, type: 'supergroup', title: 'Кварта' }, + text, + }, + }; +} + +function setup(handlers: Parameters[0] = {}, telegram: TelegramConfig = { adminId: ADMIN }) { + const tg = fakeTelegram(handlers); + const reports: string[] = []; + const system: SystemReporter = { report: async (_k, text) => void reports.push(text) }; + const bot = createBot({ client: tg.client, telegram: { token: '1:x', ...telegram }, system, retryPauseMs: 1 }); + const toGroup = () => tg.sent().filter((p) => p.chat_id === GROUP).map((p) => p.text as string); + const toAdmin = () => tg.sent().filter((p) => p.chat_id === ADMIN).map((p) => p.text as string); + return { tg, bot, reports, toGroup, toAdmin }; +} + +describe('parseCommand', () => { + it('команда, упоминание бота, аргументы', () => { + expect(parseCommand('/setup', 'kvarta_bot')).toEqual({ command: 'setup', args: [] }); + expect(parseCommand('/setup@Kvarta_Bot force', 'kvarta_bot')).toEqual({ command: 'setup', args: ['force'] }); + expect(parseCommand('/setup@other_bot', 'kvarta_bot')).toBeNull(); + expect(parseCommand('привет', 'kvarta_bot')).toBeNull(); + }); +}); + +describe('/setup', () => { + it('от чужого — тишина', async () => { + const { tg, bot } = setup(); + await bot.handleUpdate(groupMessage('/setup', 999)); + expect(tg.calls).toEqual([]); + }); + + it('без TELEGRAM_ADMIN_ID бот молчит всем', async () => { + const { tg, bot } = setup({}, {}); + await bot.handleUpdate(groupMessage('/setup')); + expect(tg.calls).toEqual([]); + }); + + it('от админа: создаёт обе темы и присылает id в личку, в группе id нет', async () => { + const { tg, bot, toGroup, toAdmin } = setup(); + await bot.handleUpdate(groupMessage('/setup@kvarta_bot')); + + expect(tg.calls.filter((c) => c.method === 'createForumTopic').map((c) => c.params)).toEqual([ + [GROUP, LEADS_TOPIC], + [GROUP, SYSTEM_TOPIC], + ]); + const lines = toAdmin().at(-1)!; + expect(lines).toContain('/root/kvarta_site/api.env'); + expect(lines).toContain(`TELEGRAM_CHAT_ID=${GROUP}`); + expect(lines).toContain('TELEGRAM_LEADS_THREAD_ID=101'); + expect(lines).toContain('TELEGRAM_SYSTEM_THREAD_ID=102'); + expect(toGroup()).toEqual(['Темы созданы, настройки отправил администратору в личку.']); + expect(toGroup().join()).not.toContain(String(GROUP)); + }); + + it('темы выключены — подсказка', async () => { + const { tg, bot, toGroup } = setup({ getChat: (id) => ({ id, type: 'supergroup', is_forum: false }) }); + await bot.handleUpdate(groupMessage('/setup')); + expect(toGroup()[0]).toMatch(/выключены темы/); + expect(tg.calls.some((c) => c.method === 'createForumTopic')).toBe(false); + }); + + it('нет права управлять темами — подсказка', async () => { + const { tg, bot, toGroup } = setup({ getChatMember: () => ({ status: 'administrator', can_manage_topics: false }) }); + await bot.handleUpdate(groupMessage('/setup')); + expect(toGroup()[0]).toMatch(/«Управление темами»/); + expect(tg.calls.some((c) => c.method === 'createForumTopic')).toBe(false); + }); + + it('админ не нажал /start — говорим в группе, темы не создаём', async () => { + const { tg, bot, toGroup } = setup({ + sendMessage: (p) => { + if (p.chat_id === ADMIN) throw new TelegramError('sendMessage: 403 Forbidden', 'config', 403); + return { message_id: 1, chat: { id: p.chat_id, type: 'supergroup' } }; + }, + }); + await bot.handleUpdate(groupMessage('/setup')); + expect(toGroup()).toEqual(['Не могу написать вам в личку. Откройте бота, нажмите /start и повторите /setup.']); + expect(tg.calls.some((c) => c.method === 'createForumTopic')).toBe(false); + }); + + it('уже настроено — отказ с подсказкой про force; force создаёт заново', async () => { + const configured = { adminId: ADMIN, chatId: GROUP, leadsThreadId: 1, systemThreadId: 2 }; + const { tg, bot, toGroup } = setup({}, configured); + await bot.handleUpdate(groupMessage('/setup')); + expect(toGroup()[0]).toMatch(/\/setup force/); + expect(tg.calls.some((c) => c.method === 'createForumTopic')).toBe(false); + + await bot.handleUpdate(groupMessage('/setup force')); + expect(tg.calls.filter((c) => c.method === 'createForumTopic')).toHaveLength(2); + }); +}); + +describe('polling', () => { + it('сдвигает offset, ошибки не роняют цикл, stop останавливает', async () => { + let n = 0; + const offsets: (number | undefined)[] = []; + const { bot } = setup({ + getUpdates: (p) => { + offsets.push(p.offset); + n++; + if (n === 1) return [groupMessage('/setup', 999)]; + if (n === 2) throw new TelegramError('getUpdates: таймаут', 'retryable'); + if (n === 3) return [{ ...groupMessage('привет', 999), update_id: 5 }]; + return new Promise((r) => setTimeout(() => r([]), 5)); + }, + }); + bot.start(); + await new Promise((r) => setTimeout(r, 50)); + await bot.stop(); + expect(offsets.slice(0, 4)).toEqual([undefined, 2, 2, 6]); + }); +}); diff --git a/apps/api/src/telegram/bot.ts b/apps/api/src/telegram/bot.ts new file mode 100644 index 0000000..cf96c72 --- /dev/null +++ b/apps/api/src/telegram/bot.ts @@ -0,0 +1,162 @@ +// Бот администратора: long polling getUpdates — единственный потребитель обновлений этого токена +// (ADR 0002). Слушаемся только TELEGRAM_ADMIN_ID, остальным не отвечаем вовсе. +import type { TelegramConfig } from '../config.js'; +import { TelegramError, type TelegramClient, type TgMessage, type TgUpdate, type TgUser } from './client.js'; +import { shortError, type SystemReporter } from './system.js'; + +export const LEADS_TOPIC = 'Kvarta Cafe: Заявки'; +export const SYSTEM_TOPIC = 'Kvarta Cafe: Система'; +const POLL_TIMEOUT_SEC = 25; +const RETRY_PAUSE_MS = 5000; + +export type Bot = { + handleUpdate(update: TgUpdate): Promise; + start(): void; + stop(): Promise; +}; + +type Client = Pick< + TelegramClient, + 'getMe' | 'getUpdates' | 'sendMessage' | 'getChat' | 'getChatMember' | 'createForumTopic' +>; + +/** `/setup@kvarta_bot force` → { command: 'setup', args: ['force'] }; чужое упоминание — null. */ +export function parseCommand(text: string | undefined, botUsername?: string) { + const match = text?.trim().match(/^\/(\w+)(?:@(\w+))?(?:\s+(.*))?$/s); + if (!match) return null; + const [, command, mention, rest] = match; + if (mention && botUsername && mention.toLowerCase() !== botUsername.toLowerCase()) return null; + return { command: command!.toLowerCase(), args: rest ? rest.trim().split(/\s+/) : [] }; +} + +export function createBot({ + client, + telegram, + system, + retryPauseMs = RETRY_PAUSE_MS, +}: { + client: Client; + telegram: TelegramConfig; + system: SystemReporter; + retryPauseMs?: number; +}): Bot { + let me: TgUser | undefined; + let offset: number | undefined; + let running = false; + let abort: AbortController | undefined; + let loop: Promise | undefined; + + const reply = (msg: TgMessage, text: string) => + client.sendMessage({ chat_id: msg.chat.id, message_thread_id: msg.message_thread_id, text }); + + async function setup(msg: TgMessage, force: boolean) { + if (msg.chat.type === 'private') { + await reply(msg, 'Команду /setup нужно отправить в группе, где бот будет присылать заявки.'); + return; + } + const { chatId, leadsThreadId, systemThreadId } = telegram; + if (chatId !== undefined && leadsThreadId !== undefined && systemThreadId !== undefined && !force) { + await reply( + msg, + 'Темы уже настроены в api.env. Чтобы создать их заново, отправьте /setup force — старые темы останутся, их можно удалить вручную.', + ); + return; + } + + const chat = await client.getChat(msg.chat.id); + if (!chat.is_forum) { + await reply(msg, 'В группе выключены темы. Включите их: настройки группы → «Темы», и повторите /setup.'); + return; + } + const member = await client.getChatMember(msg.chat.id, (me ??= await client.getMe()).id); + if (member.status !== 'administrator' || !member.can_manage_topics) { + await reply( + msg, + 'Боту не хватает прав. Сделайте его администратором группы с правом «Управление темами» и повторите /setup.', + ); + return; + } + + // Сначала проверяем личку: если админ не нажал /start, id некуда прислать — тогда и темы не создаём. + try { + await client.sendMessage({ chat_id: telegram.adminId!, text: `Настраиваю группу «${chat.title ?? chat.id}»…` }); + } catch (err) { + if (err instanceof TelegramError && err.kind === 'config') { + await reply(msg, 'Не могу написать вам в личку. Откройте бота, нажмите /start и повторите /setup.'); + return; + } + throw err; + } + + const leads = await client.createForumTopic(msg.chat.id, LEADS_TOPIC); + const sys = await client.createForumTopic(msg.chat.id, SYSTEM_TOPIC); + await client.sendMessage({ + chat_id: telegram.adminId!, + text: [ + 'Темы созданы. Впишите строки в /root/kvarta_site/api.env и перезапустите api:', + '', + `TELEGRAM_CHAT_ID=${msg.chat.id}`, + `TELEGRAM_LEADS_THREAD_ID=${leads.message_thread_id}`, + `TELEGRAM_SYSTEM_THREAD_ID=${sys.message_thread_id}`, + ].join('\n'), + }); + await reply(msg, 'Темы созданы, настройки отправил администратору в личку.'); + } + + async function handleUpdate(update: TgUpdate) { + const msg = update.message; + if (!msg?.from || telegram.adminId === undefined || msg.from.id !== telegram.adminId) return; + me ??= await client.getMe(); + const cmd = parseCommand(msg.text, me?.username); + if (!cmd) return; + if (cmd.command === 'setup') await setup(msg, cmd.args[0] === 'force'); + } + + const pause = (ms: number) => + new Promise((resolve) => { + const t = setTimeout(resolve, ms); + abort?.signal.addEventListener('abort', () => (clearTimeout(t), resolve()), { once: true }); + }); + + async function poll() { + while (running) { + try { + me ??= await client.getMe(); + const updates = await client.getUpdates( + { offset, timeout: POLL_TIMEOUT_SEC, allowed_updates: ['message', 'callback_query', 'my_chat_member'] }, + abort?.signal, + ); + for (const update of updates) { + // offset сдвигаем до обработки: упавшая команда не должна повторяться бесконечно. + offset = update.update_id + 1; + try { + await handleUpdate(update); + } catch (err) { + await system.report(`bot ${err instanceof Error ? err.message : ''}`.slice(0, 100), `Сбой команды бота: ${shortError(err)}`); + } + } + } catch (err) { + if (!running) return; + if (err instanceof TelegramError && err.kind === 'config') { + await system.report(`polling ${err.code}`, `Бот не получает обновления: ${err.message}`); + } + await pause(err instanceof TelegramError && err.retryAfterSec ? err.retryAfterSec * 1000 : retryPauseMs); + } + } + } + + return { + handleUpdate, + start() { + if (running) return; + running = true; + abort = new AbortController(); + loop = poll(); + }, + async stop() { + running = false; + abort?.abort(); + await loop; + }, + }; +} diff --git a/apps/api/src/telegram/client.test.ts b/apps/api/src/telegram/client.test.ts new file mode 100644 index 0000000..ab7a33f --- /dev/null +++ b/apps/api/src/telegram/client.test.ts @@ -0,0 +1,110 @@ +import { describe, expect, it } from 'vitest'; +import { loadConfig } from '../config.js'; +import { createTelegramClient, TelegramError, type FetchLike } from './client.js'; + +const TOKEN = '123456:SECRET-token_value'; + +function client(fetch: FetchLike, timeoutMs?: number) { + return createTelegramClient({ token: TOKEN, fetch, timeoutMs }); +} + +async function caught(p: Promise): Promise { + try { + await p; + } catch (err) { + expect(err).toBeInstanceOf(TelegramError); + return err as TelegramError; + } + throw new Error('ожидалась ошибка'); +} + +const json = (status: number, body: unknown) => new Response(JSON.stringify(body), { status }); + +describe('telegram client', () => { + it('sendMessage: POST JSON на метод, возвращает result', async () => { + const seen: { url: string; init: RequestInit }[] = []; + const tg = client(async (url, init) => { + seen.push({ url, init }); + return json(200, { ok: true, result: { message_id: 7 } }); + }); + const res = await tg.sendMessage({ chat_id: -1, message_thread_id: 3, text: 'hi' }); + expect(res).toEqual({ message_id: 7 }); + expect(seen[0]!.url).toBe(`https://api.telegram.org/bot${TOKEN}/sendMessage`); + expect(JSON.parse(seen[0]!.init.body as string)).toEqual({ chat_id: -1, message_thread_id: 3, text: 'hi' }); + }); + + it('sendDocument: multipart с файлом', async () => { + let form: FormData | undefined; + const tg = client(async (_url, init) => { + form = init.body as FormData; + return json(200, { ok: true, result: { message_id: 1 } }); + }); + await tg.sendDocument({ chat_id: 42, file: { name: 'kvarta.sqlite.gz', data: new Uint8Array([1, 2, 3]) }, caption: 'c' }); + expect(form!.get('chat_id')).toBe('42'); + const doc = form!.get('document') as File; + expect(doc.name).toBe('kvarta.sqlite.gz'); + expect(new Uint8Array(await doc.arrayBuffer())).toEqual(new Uint8Array([1, 2, 3])); + }); + + it('429 → повторяемая с retry_after; 5xx → повторяемая; 400/403 → конфигурационная', async () => { + const respond = (status: number, extra = {}) => + client(async () => json(status, { ok: false, error_code: status, description: 'Oops', ...extra })); + + const tooMany = await caught(respond(429, { parameters: { retry_after: 17 } }).sendMessage({ chat_id: 1, text: '' })); + expect(tooMany).toMatchObject({ kind: 'retryable', code: 429, retryAfterSec: 17 }); + expect((await caught(respond(502).getMe())).kind).toBe('retryable'); + expect(await caught(respond(400).getMe())).toMatchObject({ kind: 'config', code: 400, message: 'getMe: 400 Oops' }); + expect((await caught(respond(403).getMe())).kind).toBe('config'); + }); + + it('токен не попадает в текст ошибки: ни при сбое сети, ни в описании ответа', async () => { + const network = await caught( + client(async (url) => { + throw new TypeError(`fetch failed: ${url}`); + }).getMe(), + ); + expect(network).toMatchObject({ kind: 'retryable', message: 'getMe: сеть (TypeError)' }); + + const echoed = await caught(client(async () => json(404, { ok: false, description: `Not Found: ${TOKEN}` })).getMe()); + expect(echoed.message).not.toContain(TOKEN); + expect(echoed.message).not.toContain('SECRET'); + }); + + it('таймаут 5 с через AbortSignal → повторяемая', async () => { + const hang: FetchLike = (_url, init) => + new Promise((_resolve, reject) => init.signal!.addEventListener('abort', () => reject(init.signal!.reason))); + const err = await caught(client(hang, 20).getMe()); + expect(err).toMatchObject({ kind: 'retryable', message: 'getMe: таймаут' }); + }); + + it('downloadFile: отказывает при превышении лимита', async () => { + const tg = client(async () => new Response(new Uint8Array(10))); + expect((await tg.downloadFile('documents/a.gz', 100)).byteLength).toBe(10); + expect((await caught(tg.downloadFile('documents/a.gz', 5))).kind).toBe('config'); + }); +}); + +describe('loadConfig', () => { + it('значения по умолчанию и разбор Telegram-переменных', () => { + expect(loadConfig({})).toEqual({ + port: 3001, + dbPath: './data/kvarta.sqlite', + version: 'dev', + trustedProxies: [], + telegram: {}, + }); + const cfg = loadConfig({ + TELEGRAM_BOT_TOKEN: TOKEN, + TELEGRAM_CHAT_ID: '-1001234', + TELEGRAM_LEADS_THREAD_ID: '5', + TRUSTED_PROXIES: '172.18.0.1, 10.0.0.1', + }); + expect(cfg.telegram).toMatchObject({ token: TOKEN, chatId: -1001234, leadsThreadId: 5 }); + expect(cfg.trustedProxies).toEqual(['172.18.0.1', '10.0.0.1']); + }); + + it('ошибки называют переменную, но не показывают токен', () => { + expect(() => loadConfig({ TELEGRAM_CHAT_ID: 'abc' })).toThrow('TELEGRAM_CHAT_ID'); + expect(() => loadConfig({ TELEGRAM_BOT_TOKEN: 'bad token SECRET' })).toThrow(/^TELEGRAM_BOT_TOKEN: неверный формат$/); + }); +}); diff --git a/apps/api/src/telegram/client.ts b/apps/api/src/telegram/client.ts new file mode 100644 index 0000000..5eb775a --- /dev/null +++ b/apps/api/src/telegram/client.ts @@ -0,0 +1,143 @@ +// Тонкий клиент Bot API на fetch. URL содержит токен, поэтому ни URL, ни исходные +// ошибки fetch наружу не отдаём: тексты TelegramError собираются только из метода и ответа API. + +export type TelegramErrorKind = 'retryable' | 'config'; + +export class TelegramError extends Error { + constructor( + message: string, + readonly kind: TelegramErrorKind, + /** HTTP-код ответа API; нет у сетевых ошибок и таймаута */ + readonly code?: number, + readonly retryAfterSec?: number, + ) { + super(message); + this.name = 'TelegramError'; + } +} + +export type FetchLike = (url: string, init: RequestInit) => Promise; + +export type TgUser = { id: number; is_bot: boolean; first_name: string; username?: string }; +export type TgChat = { id: number; type: 'private' | 'group' | 'supergroup' | 'channel'; title?: string; is_forum?: boolean }; +export type TgDocument = { file_id: string; file_name?: string; file_size?: number }; +export type TgMessage = { + message_id: number; + message_thread_id?: number; + from?: TgUser; + chat: TgChat; + text?: string; + document?: TgDocument; +}; +export type TgCallbackQuery = { id: string; from: TgUser; message?: TgMessage; data?: string }; +export type TgUpdate = { + update_id: number; + message?: TgMessage; + callback_query?: TgCallbackQuery; + my_chat_member?: { chat: TgChat; from: TgUser }; +}; +export type TgChatMember = { status: string; can_manage_topics?: boolean }; +export type InlineKeyboard = { inline_keyboard: { text: string; callback_data: string }[][] }; + +export type SendMessageParams = { + chat_id: number; + text: string; + message_thread_id?: number; + parse_mode?: 'HTML'; + reply_markup?: InlineKeyboard; + disable_web_page_preview?: boolean; +}; + +export type TelegramClient = ReturnType; + +const API = 'https://api.telegram.org'; + +export function createTelegramClient({ + token, + fetch = globalThis.fetch, + timeoutMs = 5000, +}: { + token: string; + fetch?: FetchLike; + timeoutMs?: number; +}) { + async function request(method: string, url: string, init: RequestInit, timeout: number, signal?: AbortSignal) { + const timeoutSignal = AbortSignal.timeout(timeout); + try { + return await fetch(url, { ...init, signal: signal ? AbortSignal.any([signal, timeoutSignal]) : timeoutSignal }); + } catch (err) { + if (signal?.aborted) throw new TelegramError(`${method}: отменено`, 'retryable'); + const reason = timeoutSignal.aborted ? 'таймаут' : `сеть (${err instanceof Error ? err.name : 'unknown'})`; + throw new TelegramError(`${method}: ${reason}`, 'retryable'); + } + } + + async function call( + method: string, + params: Record | FormData = {}, + opts: { timeoutMs?: number; signal?: AbortSignal } = {}, + ): Promise { + const init: RequestInit = + params instanceof FormData + ? { method: 'POST', body: params } + : { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify(params) }; + const res = await request(method, `${API}/bot${token}/${method}`, init, opts.timeoutMs ?? timeoutMs, opts.signal); + + let body: { ok?: boolean; result?: T; description?: string; parameters?: { retry_after?: number } }; + try { + body = (await res.json()) as typeof body; + } catch { + body = {}; + } + if (res.ok && body.ok) return body.result as T; + + const description = (body.description ?? res.statusText).replaceAll(token, '***'); + const message = `${method}: ${res.status} ${description}`; + if (res.status === 429) { + throw new TelegramError(message, 'retryable', 429, body.parameters?.retry_after ?? 1); + } + if (res.status >= 500) throw new TelegramError(message, 'retryable', res.status); + // 400/401/403/404/409: дело в настройке (чат, тема, права, токен) — повтор сам не поможет. + throw new TelegramError(message, 'config', res.status); + } + + return { + call, + getMe: () => call('getMe'), + sendMessage: (params: SendMessageParams) => call('sendMessage', params), + sendDocument(params: { + chat_id: number; + file: { name: string; data: Uint8Array }; + caption?: string; + message_thread_id?: number; + }) { + const form = new FormData(); + form.set('chat_id', String(params.chat_id)); + if (params.message_thread_id !== undefined) form.set('message_thread_id', String(params.message_thread_id)); + if (params.caption) form.set('caption', params.caption); + form.set('document', new Blob([new Uint8Array(params.file.data)]), params.file.name); + // Файл бэкапа может идти дольше обычного сообщения. + return call('sendDocument', form, { timeoutMs: Math.max(timeoutMs, 60_000) }); + }, + getUpdates: ( + params: { offset?: number; timeout: number; allowed_updates: string[] }, + signal?: AbortSignal, + ) => call('getUpdates', params, { timeoutMs: (params.timeout + 10) * 1000, signal }), + createForumTopic: (chat_id: number, name: string) => + call<{ message_thread_id: number; name: string }>('createForumTopic', { chat_id, name }), + getChat: (chat_id: number) => call('getChat', { chat_id }), + getChatMember: (chat_id: number, user_id: number) => call('getChatMember', { chat_id, user_id }), + answerCallbackQuery: (callback_query_id: string, text?: string) => + call('answerCallbackQuery', { callback_query_id, text }), + editMessageText: (params: { chat_id: number; message_id: number; text: string; reply_markup?: InlineKeyboard }) => + call('editMessageText', params), + getFile: (file_id: string) => call<{ file_id: string; file_path?: string; file_size?: number }>('getFile', { file_id }), + async downloadFile(filePath: string, maxBytes: number): Promise { + const res = await request('downloadFile', `${API}/file/bot${token}/${filePath}`, {}, 60_000); + if (!res.ok) throw new TelegramError(`downloadFile: ${res.status}`, res.status >= 500 ? 'retryable' : 'config', res.status); + const data = new Uint8Array(await res.arrayBuffer()); + if (data.byteLength > maxBytes) throw new TelegramError('downloadFile: файл больше лимита', 'config'); + return data; + }, + }; +} diff --git a/apps/api/src/telegram/fake.ts b/apps/api/src/telegram/fake.ts new file mode 100644 index 0000000..72bd691 --- /dev/null +++ b/apps/api/src/telegram/fake.ts @@ -0,0 +1,36 @@ +// Фейковый Telegram для тестов: записывает вызовы, ответы и сбои задаются по методу. +import type { TelegramClient } from './client.js'; + +type Method = Exclude; +type Handler = (params: any) => unknown; + +export function fakeTelegram(handlers: Partial> = {}) { + const calls: { method: Method; params: any }[] = []; + let messageId = 0; + const defaults: Partial> = { + getMe: () => ({ id: 777, is_bot: true, first_name: 'Kvarta', username: 'kvarta_bot' }), + sendMessage: (p) => ({ message_id: ++messageId, chat: { id: p.chat_id, type: 'private' }, text: p.text }), + sendDocument: (p) => ({ message_id: ++messageId, chat: { id: p.chat_id, type: 'private' } }), + getChat: (p) => ({ id: p, type: 'supergroup', title: 'Кварта', is_forum: true }), + getChatMember: () => ({ status: 'administrator', can_manage_topics: true }), + createForumTopic: (_chat: number) => ({ message_thread_id: 100 + calls.filter((c) => c.method === 'createForumTopic').length, name: '' }), + answerCallbackQuery: () => true, + editMessageText: () => true, + getUpdates: () => [], + }; + + const client = new Proxy({} as Record, { + get(_t, method: string) { + return async (...args: unknown[]) => { + const params = args.length > 1 && method !== 'getUpdates' ? args : args[0]; + calls.push({ method: method as Method, params }); + const handler = handlers[method as Method] ?? defaults[method as Method]; + if (!handler) throw new Error(`fake telegram: нет обработчика ${method}`); + return handler(params); + }; + }, + }) as unknown as TelegramClient; + + const sent = (method: Method = 'sendMessage') => calls.filter((c) => c.method === method).map((c) => c.params); + return { client, calls, sent }; +} diff --git a/apps/api/src/telegram/format.test.ts b/apps/api/src/telegram/format.test.ts new file mode 100644 index 0000000..664eae6 --- /dev/null +++ b/apps/api/src/telegram/format.test.ts @@ -0,0 +1,78 @@ +import { describe, expect, it } from 'vitest'; +import { escapeHtml, formatClarification, formatLead, touchSource, type LeadForMessage } from './format.js'; + +const NOW = new Date('2026-10-17T12:00:00.000Z'); + +const lead: LeadForMessage = { + id: 12, + phone: '+375291234567', + name: 'Аня', + interest: 'subscription', + comment: 'Под эспрессо', + firstTouch: { utm_source: 'instagram', landingPage: '/', at: '2026-10-08T10:00:00.000Z' }, + lastTouch: { + utm_source: 'yandex', + utm_medium: 'cpc', + utm_campaign: 'osen', + landingPage: '/?utm_source=yandex', + submitPage: '/#form', + at: '2026-10-17T11:50:00.000Z', + }, +}; + +describe('formatLead', () => { + it('шаблон плана', () => { + expect(formatLead(lead, NOW)).toBe( + [ + '🟥 Заявка #12 · подписка', + 'Имя: Аня', + 'Телефон: +375291234567', + 'Комментарий: Под эспрессо', + 'Источник: yandex / cpc / osen (первый визит: instagram, 9 дн. назад)', + 'Страница: /?utm_source=yandex → /#form', + ].join('\n'), + ); + }); + + it('экранирует ввод посетителя', () => { + const text = formatLead({ ...lead, name: 'Аня & Ко', comment: 'клик' }, NOW); + expect(text).toContain('Имя: <b>Аня</b> & Ко'); + expect(text).toContain('Комментарий: <a href="x">клик</a>'); + expect(text).not.toMatch(/<(?!\/?b>)/); + }); + + it('без комментария строки нет; первый визит не показываем, если источник тот же', () => { + const text = formatLead({ ...lead, comment: null, firstTouch: { ...lead.lastTouch, at: '2026-10-01T00:00:00Z' } }, NOW); + expect(text).not.toContain('Комментарий'); + expect(text).toContain('Источник: yandex / cpc / osen\n'); + }); + + it('первый визит сегодня', () => { + const text = formatLead({ ...lead, firstTouch: { landingPage: '/', at: '2026-10-17T09:00:00.000Z' } }, NOW); + expect(text).toContain('(первый визит: прямой заход, сегодня)'); + }); +}); + +describe('touchSource', () => { + it('utm → хост referrer → прямой заход', () => { + expect(touchSource({ utm_source: 'instagram', landingPage: '/', at: '' })).toBe('instagram'); + expect(touchSource({ referrer: 'https://www.google.com/search?q=x', landingPage: '/', at: '' })).toBe('www.google.com'); + expect(touchSource({ referrer: 'не url', landingPage: '/', at: '' })).toBe('не url'); + expect(touchSource({ landingPage: '/', at: '' })).toBe('прямой заход'); + }); +}); + +describe('formatClarification', () => { + it('только изменившиеся поля', () => { + expect( + formatClarification({ leadId: 12, name: 'Аня', interest: 'one_time', comment: null, changed: ['interest', 'comment'] }), + ).toBe(['🟨 Уточнение к заявке #12', 'Интерес: разовый заказ', 'Комментарий: — (убран)'].join('\n')); + expect( + formatClarification({ leadId: 3, name: 'A', interest: 'question', comment: 'x', changed: ['name'] }), + ).toBe('🟨 Уточнение к заявке #3\nИмя: A<b>'); + }); +}); + +it('escapeHtml', () => { + expect(escapeHtml('a & b < c > d')).toBe('a & b < c > d'); +}); diff --git a/apps/api/src/telegram/format.ts b/apps/api/src/telegram/format.ts new file mode 100644 index 0000000..7a67cc9 --- /dev/null +++ b/apps/api/src/telegram/format.ts @@ -0,0 +1,77 @@ +// Тексты сообщений менеджеру. parse_mode HTML: всё, что ввёл посетитель, экранируем. +import { INTEREST_LABELS, type Interest, type Touch } from '@kvarta/shared/lead'; +import type { ClarifiedField } from '../notifier.js'; + +export function escapeHtml(s: string): string { + return s.replaceAll('&', '&').replaceAll('<', '<').replaceAll('>', '>'); +} + +export type LeadForMessage = { + id: number; + phone: string; + name: string; + interest: Interest; + comment: string | null; + firstTouch: Touch | null; + lastTouch: Touch; +}; + +export type ClarificationForMessage = { + leadId: number; + name: string; + interest: Interest; + comment: string | null; + changed: ClarifiedField[]; +}; + +const DAY_MS = 24 * 60 * 60 * 1000; + +/** «yandex / cpc / osen», иначе хост referrer, иначе «прямой заход». */ +export function touchSource(touch: Touch): string { + const utm = [touch.utm_source, touch.utm_medium, touch.utm_campaign].filter(Boolean); + if (utm.length > 0) return utm.join(' / '); + if (touch.referrer) { + try { + return new URL(touch.referrer).hostname || touch.referrer; + } catch { + return touch.referrer; + } + } + return 'прямой заход'; +} + +function daysAgo(at: string, now: Date): string { + const days = Math.floor((now.getTime() - Date.parse(at)) / DAY_MS); + if (!Number.isFinite(days) || days <= 0) return 'сегодня'; + return `${days} дн. назад`; +} + +function sourceLine(lead: LeadForMessage, now: Date): string { + const last = touchSource(lead.lastTouch); + const first = lead.firstTouch ? touchSource(lead.firstTouch) : null; + if (!first || !lead.firstTouch || first === last) return last; + return `${last} (первый визит: ${first}, ${daysAgo(lead.firstTouch.at, now)})`; +} + +export function formatLead(lead: LeadForMessage, now: Date): string { + const t = lead.lastTouch; + const page = t.submitPage && t.submitPage !== t.landingPage ? `${t.landingPage} → ${t.submitPage}` : t.landingPage; + const lines = [ + `🟥 Заявка #${lead.id} · ${INTEREST_LABELS[lead.interest]}`, + `Имя: ${escapeHtml(lead.name)}`, + `Телефон: ${escapeHtml(lead.phone)}`, + ]; + if (lead.comment) lines.push(`Комментарий: ${escapeHtml(lead.comment)}`); + lines.push(`Источник: ${escapeHtml(sourceLine(lead, now))}`, `Страница: ${escapeHtml(page)}`); + return lines.join('\n'); +} + +export function formatClarification(c: ClarificationForMessage): string { + const lines = [`🟨 Уточнение к заявке #${c.leadId}`]; + for (const field of c.changed) { + if (field === 'name') lines.push(`Имя: ${escapeHtml(c.name)}`); + if (field === 'interest') lines.push(`Интерес: ${INTEREST_LABELS[c.interest]}`); + if (field === 'comment') lines.push(`Комментарий: ${c.comment ? escapeHtml(c.comment) : '— (убран)'}`); + } + return lines.join('\n'); +} diff --git a/apps/api/src/telegram/outbox.test.ts b/apps/api/src/telegram/outbox.test.ts new file mode 100644 index 0000000..5f2931c --- /dev/null +++ b/apps/api/src/telegram/outbox.test.ts @@ -0,0 +1,158 @@ +import { describe, expect, it } from 'vitest'; +import { validateLead, type ValidLead } from '@kvarta/shared/lead'; +import { openDb } from '../db/index.js'; +import { saveLead } from '../leads.js'; +import { validBody } from '../testing.js'; +import { TelegramError } from './client.js'; +import { fakeTelegram } from './fake.js'; +import { backoffMs, createOutbox, MAX_ATTEMPTS } from './outbox.js'; +import type { SystemReporter } from './system.js'; + +const MINUTE = 60 * 1000; + +function lead(overrides: Parameters[0] = {}): ValidLead { + const r = validateLead(validBody(overrides)); + if (!r.ok) throw new Error('invalid'); + return r.data; +} + +function setup({ failures = [] as unknown[], chatId = -100 as number | null } = {}) { + const db = openDb(':memory:'); + const clock = { now: new Date('2026-10-17T12:00:00.000Z') }; + const queue = [...failures]; + const tg = fakeTelegram({ + sendMessage: (p) => { + const failure = queue.shift(); + if (failure) throw failure; + return { message_id: 1, chat: { id: p.chat_id, type: 'supergroup' } }; + }, + }); + const reports: { kind: string; text: string }[] = []; + const system: SystemReporter = { report: async (kind, text) => void reports.push({ kind, text }) }; + const outbox = createOutbox({ getDb: () => db, client: tg.client, chatId: chatId ?? undefined, threadId: 5, now: () => clock.now, system }); + const advance = (ms: number) => (clock.now = new Date(clock.now.getTime() + ms)); + const save = (l: ValidLead) => saveLead(db, l, { now: clock.now, userAgent: null }); + const row = (table: string, id = 1) => + db.prepare(`SELECT tg_status, tg_attempts, tg_last_error, tg_next_at FROM ${table} WHERE id = ?`).get(id) as { + tg_status: string; + tg_attempts: number; + tg_last_error: string | null; + tg_next_at: string | null; + }; + return { db, tg, outbox, reports, advance, save, row, clock }; +} + +const network = () => new TelegramError('sendMessage: таймаут', 'retryable'); + +describe('outbox', () => { + it('успех: сообщение в тему «Заявки», строка sent', async () => { + const { tg, outbox, save, row } = setup(); + const saved = save(lead()); + await outbox.leadSaved(saved.kind === 'created' ? saved.event : (null as never)); + + expect(tg.sent()).toHaveLength(1); + expect(tg.sent()[0]).toMatchObject({ chat_id: -100, message_thread_id: 5, parse_mode: 'HTML' }); + expect(tg.sent()[0].text).toMatch(/^🟥 Заявка #1 · подписка/); + expect(row('leads')).toMatchObject({ tg_status: 'sent', tg_attempts: 0 }); + }); + + it('неудача: failed, пауза 1, 2, 4 мин, потом отправка', async () => { + const { tg, outbox, save, row, advance, clock } = setup({ failures: [network(), network(), network()] }); + save(lead()); + + await outbox.tick(); + expect(row('leads')).toMatchObject({ + tg_status: 'failed', + tg_attempts: 1, + tg_last_error: 'sendMessage: таймаут', + tg_next_at: new Date(clock.now.getTime() + MINUTE).toISOString(), + }); + // До паузы не повторяем. + await outbox.tick(); + expect(tg.sent()).toHaveLength(1); + + advance(MINUTE); + await outbox.tick(); + expect(row('leads').tg_next_at).toBe(new Date(clock.now.getTime() + 2 * MINUTE).toISOString()); + advance(2 * MINUTE); + await outbox.tick(); + expect(row('leads').tg_next_at).toBe(new Date(clock.now.getTime() + 4 * MINUTE).toISOString()); + advance(4 * MINUTE); + await outbox.tick(); + expect(row('leads')).toMatchObject({ tg_status: 'sent', tg_attempts: 3, tg_last_error: null }); + expect(tg.sent()).toHaveLength(4); + }); + + it('429: ждём не меньше retry_after', async () => { + const { outbox, save, row, clock } = setup({ failures: [new TelegramError('429', 'retryable', 429, 300)] }); + save(lead()); + await outbox.tick(); + expect(row('leads').tg_next_at).toBe(new Date(clock.now.getTime() + 300_000).toISOString()); + }); + + it('уточнение не уходит раньше своей заявки', async () => { + const { tg, outbox, save, row, advance } = setup({ failures: [network()] }); + save(lead()); + advance(MINUTE); + save(lead({ comment: 'Лучше в зёрнах' })); + + await outbox.tick(); + // Заявка упала — уточнение тоже ждёт, хотя ему пауза не назначена. + expect(tg.sent()).toHaveLength(1); + expect(row('lead_clarifications')).toMatchObject({ tg_status: 'pending', tg_attempts: 0 }); + + advance(MINUTE); + await outbox.tick(); + expect(tg.sent().map((p) => p.text.split('\n')[0])).toEqual([ + '🟥 Заявка #1 · подписка', + '🟥 Заявка #1 · подписка', + '🟨 Уточнение к заявке #1', + ]); + expect(tg.sent()[2].text).toBe('🟨 Уточнение к заявке #1\nКомментарий: Лучше в зёрнах'); + expect(row('lead_clarifications').tg_status).toBe('sent'); + }); + + it('после 10 неудач — сообщение в «Систему» без персональных данных и больше не пробуем', async () => { + const { tg, outbox, save, row, advance, reports } = setup({ + failures: Array.from({ length: MAX_ATTEMPTS }, network), + }); + save(lead()); + for (let i = 1; i <= MAX_ATTEMPTS; i++) { + await outbox.tick(); + advance(backoffMs(i)); + } + expect(row('leads')).toMatchObject({ tg_status: 'failed', tg_attempts: MAX_ATTEMPTS }); + expect(reports).toEqual([{ kind: 'outbox leads 1', text: 'Заявка #1 не ушла в Telegram после 10 попыток' }]); + + advance(24 * 60 * MINUTE); + await outbox.tick(); + expect(tg.sent()).toHaveLength(MAX_ATTEMPTS); + }); + + it('400/403 от Telegram — в «Систему»', async () => { + const { outbox, save, reports } = setup({ + failures: [new TelegramError('sendMessage: 400 Bad Request: message thread not found', 'config', 400)], + }); + save(lead()); + await outbox.tick(); + expect(reports).toEqual([ + { kind: 'telegram 400', text: 'Telegram отказал: sendMessage: 400 Bad Request: message thread not found' }, + ]); + }); + + it('без TELEGRAM_CHAT_ID ничего не шлём, записи остаются pending', async () => { + const { tg, outbox, save, row } = setup({ chatId: null }); + save(lead()); + await outbox.tick(); + expect(tg.calls).toEqual([]); + expect(row('leads')).toMatchObject({ tg_status: 'pending', tg_attempts: 0 }); + }); + + it('параллельные вызовы не отправляют одно и то же дважды', async () => { + const { tg, outbox, save } = setup(); + save(lead()); + save(lead({ phone: '+375447654321' })); + await Promise.all([outbox.tick(), outbox.tick(), outbox.tick()]); + expect(tg.sent()).toHaveLength(2); + }); +}); diff --git a/apps/api/src/telegram/outbox.ts b/apps/api/src/telegram/outbox.ts new file mode 100644 index 0000000..05cbb2b --- /dev/null +++ b/apps/api/src/telegram/outbox.ts @@ -0,0 +1,183 @@ +// Очередь отправки заявок и уточнений в тему «Заявки». Состояние живёт в строках (tg_*), +// поэтому рестарт или потерянное событие ничего не теряют: следующий тик доберёт pending. +import type { Interest, Touch } from '@kvarta/shared/lead'; +import type { Db } from '../db/index.js'; +import type { ClarifiedField, LeadEvent, Notifier } from '../notifier.js'; +import { TelegramError, type TelegramClient } from './client.js'; +import { formatClarification, formatLead } from './format.js'; +import type { SystemReporter } from './system.js'; + +export const MAX_ATTEMPTS = 10; +export const TICK_MS = 60 * 1000; + +type Table = 'leads' | 'lead_clarifications'; +type Due = { table: Table; id: number; lead_id: number; created_at: string; tg_attempts: number }; + +type LeadRow = { + id: number; + phone: string; + name: string; + interest: Interest; + comment: string | null; + first_touch: string | null; + last_touch: string; +}; +type VersionRow = { name: string; interest: Interest; comment: string | null }; + +export type Outbox = Notifier & { + /** Отправить всё, что пора. Вызовы не пересекаются: следующий ждёт предыдущий. */ + tick(): Promise; + start(): void; + stop(): Promise; +}; + +/** Пауза перед попыткой n+1 после n неудач: 1, 2, 4… минуты. */ +export function backoffMs(failedAttempts: number): number { + return 2 ** (failedAttempts - 1) * 60 * 1000; +} + +export function createOutbox({ + getDb, + client, + chatId, + threadId, + now, + system, + intervalMs = TICK_MS, +}: { + getDb: () => Db; + client?: Pick; + chatId?: number; + threadId?: number; + now: () => Date; + system: SystemReporter; + intervalMs?: number; +}): Outbox { + const enabled = client !== undefined && chatId !== undefined; + let chain: Promise = Promise.resolve(); + let timer: NodeJS.Timeout | undefined; + + function nextDue(db: Db, nowIso: string): Due | undefined { + const ready = (t: string) => `${t}.tg_status IN ('pending','failed') AND ${t}.tg_attempts < ${MAX_ATTEMPTS} + AND (${t}.tg_next_at IS NULL OR ${t}.tg_next_at <= @now)`; + const lead = db + .prepare( + `SELECT 'leads' AS "table", id, id AS lead_id, created_at, tg_attempts FROM leads + WHERE ${ready('leads')} ORDER BY id LIMIT 1`, + ) + .get({ now: nowIso }) as Due | undefined; + // Уточнение без отправленной заявки менеджеру непонятно — ждём заявку. + const clarification = db + .prepare( + `SELECT 'lead_clarifications' AS "table", c.id, c.lead_id, c.created_at, c.tg_attempts + FROM lead_clarifications c JOIN leads l ON l.id = c.lead_id + WHERE l.tg_status = 'sent' AND ${ready('c')} + ORDER BY c.id LIMIT 1`, + ) + .get({ now: nowIso }) as Due | undefined; + if (!lead) return clarification; + if (!clarification) return lead; + return clarification.created_at < lead.created_at ? clarification : lead; + } + + function render(db: Db, item: Due): string { + if (item.table === 'leads') { + const row = db.prepare('SELECT * FROM leads WHERE id = ?').get(item.id) as LeadRow; + return formatLead( + { + id: row.id, + phone: row.phone, + name: row.name, + interest: row.interest, + comment: row.comment, + firstTouch: row.first_touch ? (JSON.parse(row.first_touch) as Touch) : null, + lastTouch: JSON.parse(row.last_touch) as Touch, + }, + now(), + ); + } + const current = db + .prepare('SELECT name, interest, comment FROM lead_clarifications WHERE id = ?') + .get(item.id) as VersionRow; + // Изменения считаем по базе, а не из события: тик после рестарта события не видел. + const previous = + (db + .prepare( + `SELECT name, interest, comment FROM lead_clarifications + WHERE lead_id = ? AND id < ? ORDER BY id DESC LIMIT 1`, + ) + .get(item.lead_id, item.id) as VersionRow | undefined) ?? + (db.prepare('SELECT name, interest, comment FROM leads WHERE id = ?').get(item.lead_id) as VersionRow); + const changed = (['name', 'interest', 'comment'] as ClarifiedField[]).filter((f) => current[f] !== previous[f]); + return formatClarification({ leadId: item.lead_id, ...current, changed }); + } + + async function sendOne(db: Db, item: Due): Promise { + try { + await client!.sendMessage({ + chat_id: chatId!, + message_thread_id: threadId, + text: render(db, item), + parse_mode: 'HTML', + disable_web_page_preview: true, + }); + db.prepare(`UPDATE ${item.table} SET tg_status = 'sent', tg_last_error = NULL, tg_next_at = NULL WHERE id = ?`).run( + item.id, + ); + return true; + } catch (err) { + const attempts = item.tg_attempts + 1; + const tgErr = err instanceof TelegramError ? err : undefined; + // В tg_last_error только наш текст: у TelegramError он без URL и токена. + const message = tgErr ? tgErr.message : err instanceof Error ? `${err.name}: ${err.message}` : String(err); + const waitMs = Math.max(backoffMs(attempts), (tgErr?.retryAfterSec ?? 0) * 1000); + db.prepare( + `UPDATE ${item.table} SET tg_status = 'failed', tg_attempts = ?, tg_last_error = ?, tg_next_at = ? WHERE id = ?`, + ).run(attempts, message.slice(0, 500), new Date(now().getTime() + waitMs).toISOString(), item.id); + + const what = item.table === 'leads' ? `Заявка #${item.id}` : `Уточнение к заявке #${item.lead_id}`; + if (tgErr?.kind === 'config') await system.report(`telegram ${tgErr.code}`, `Telegram отказал: ${tgErr.message}`); + if (attempts >= MAX_ATTEMPTS) { + await system.report( + `outbox ${item.table} ${item.id}`, + `${what} не ${item.table === 'leads' ? 'ушла' : 'ушло'} в Telegram после ${MAX_ATTEMPTS} попыток`, + ); + } + return false; + } + } + + async function run(): Promise { + if (!enabled) return; + const db = getDb(); + for (;;) { + const item = nextDue(db, now().toISOString()); + if (!item) return; + // После неудачи останавливаемся: остальное подождёт следующего тика, порядок не перемешается. + if (!(await sendOne(db, item))) return; + } + } + + function tick(): Promise { + chain = chain.then(run).catch((err: unknown) => console.error('outbox tick failed', err)); + return chain; + } + + return { + leadSaved(_event: LeadEvent) { + return tick(); + }, + tick, + start() { + if (!enabled || timer) return; + timer = setInterval(() => void tick(), intervalMs); + timer.unref(); + void tick(); + }, + async stop() { + if (timer) clearInterval(timer); + timer = undefined; + await chain; + }, + }; +} diff --git a/apps/api/src/telegram/system.test.ts b/apps/api/src/telegram/system.test.ts new file mode 100644 index 0000000..c2334bb --- /dev/null +++ b/apps/api/src/telegram/system.test.ts @@ -0,0 +1,63 @@ +import { describe, expect, it } from 'vitest'; +import { fakeTelegram } from './fake.js'; +import { createSystemReporter, SYSTEM_THROTTLE_MS } from './system.js'; + +function setup(overrides: { threadId?: number | undefined } = { threadId: 2 }) { + const tg = fakeTelegram(); + const clock = { now: new Date('2026-10-17T12:00:00.000Z') }; + const lines: string[] = []; + const system = createSystemReporter({ + client: tg.client, + chatId: -100, + threadId: overrides.threadId, + now: () => clock.now, + log: (l) => lines.push(l), + }); + const advance = (ms: number) => (clock.now = new Date(clock.now.getTime() + ms)); + return { tg, system, lines, advance }; +} + +describe('reportSystem', () => { + it('шлёт в тему «Система» и дублирует в stdout', async () => { + const { tg, system, lines } = setup(); + await system.report('started', 'api запущен, версия abc'); + expect(tg.sent()).toEqual([{ chat_id: -100, message_thread_id: 2, text: 'api запущен, версия abc' }]); + expect(lines).toEqual(['[system] started: api запущен, версия abc']); + }); + + it('одинаковый kind — не чаще раза в 10 минут, пропущенные идут счётчиком', async () => { + const { tg, system, advance } = setup(); + await system.report('500 boom', 'первый'); + advance(MINUTE); + await system.report('500 boom', 'второй'); + await system.report('500 boom', 'третий'); + await system.report('другое', 'другое событие'); + advance(SYSTEM_THROTTLE_MS); + await system.report('500 boom', 'четвёртый'); + advance(SYSTEM_THROTTLE_MS); + await system.report('500 boom', 'пятый'); + + expect(tg.sent().map((p) => p.text)).toEqual(['первый', 'другое событие', '×3 за период · четвёртый', 'пятый']); + }); + + it('без TELEGRAM_SYSTEM_THREAD_ID — только stdout', async () => { + const { tg, system, lines } = setup({ threadId: undefined }); + await system.report('started', 'x'); + expect(tg.calls).toEqual([]); + expect(lines).toHaveLength(1); + }); + + it('сбой отправки не бросает', async () => { + const tg = fakeTelegram({ + sendMessage: () => { + throw new Error('down'); + }, + }); + const lines: string[] = []; + const system = createSystemReporter({ client: tg.client, chatId: 1, threadId: 2, now: () => new Date(), log: (l) => lines.push(l) }); + await expect(system.report('k', 'x')).resolves.toBeUndefined(); + expect(lines.at(-1)).toContain('не отправлено'); + }); +}); + +const MINUTE = 60 * 1000; diff --git a/apps/api/src/telegram/system.ts b/apps/api/src/telegram/system.ts new file mode 100644 index 0000000..5b8787b --- /dev/null +++ b/apps/api/src/telegram/system.ts @@ -0,0 +1,64 @@ +// Тема «Kvarta Cafe: Система»: сбои и служебные события без персональных данных. +// Всё дублируется в stdout — это единственный канал, если Telegram недоступен или тема не задана. +import type { TelegramClient } from './client.js'; + +export const SYSTEM_THROTTLE_MS = 10 * 60 * 1000; + +export type SystemReporter = { + /** Никогда не бросает: отчёт о сбое не должен порождать новый сбой. */ + report(kind: string, text: string): Promise; +}; + +export function createSystemReporter({ + client, + chatId, + threadId, + now, + log = (line: string) => console.log(line), +}: { + client?: Pick; + chatId?: number; + threadId?: number; + now: () => Date; + log?: (line: string) => void; +}): SystemReporter { + // Одинаковые события не чаще раза в 10 минут; пропущенные копим и показываем с ближайшим отправленным. + const throttle = new Map(); + + return { + async report(kind, text) { + log(`[system] ${kind}: ${text}`); + if (!client || chatId === undefined || threadId === undefined) return; + + const t = now().getTime(); + const state = throttle.get(kind); + if (state && t - state.sentAt < SYSTEM_THROTTLE_MS) { + state.skipped++; + return; + } + const count = (state?.skipped ?? 0) + 1; + throttle.set(kind, { sentAt: t, skipped: 0 }); + try { + await client.sendMessage({ + chat_id: chatId, + message_thread_id: threadId, + text: count > 1 ? `×${count} за период · ${text}` : text, + }); + } catch (err) { + log(`[system] не отправлено в Telegram: ${err instanceof Error ? err.message : String(err)}`); + } + }, + }; +} + +/** Короткий stack без путей до node_modules: в «Систему» и stdout. */ +export function shortError(err: unknown): string { + if (!(err instanceof Error)) return String(err); + const frames = (err.stack ?? '') + .split('\n') + .slice(1) + .filter((l) => !l.includes('node_modules') && !l.includes('node:')) + .slice(0, 3) + .map((l) => l.trim()); + return [`${err.name}: ${err.message}`, ...frames].join('\n'); +}