feat(api): Telegram — темы, очередь повторов, «Система», /setup

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: всё окружение в одном месте.
This commit is contained in:
2026-10-17 16:07:26 +03:00
parent 5d970be228
commit 667c6a6e6d
15 changed files with 1316 additions and 22 deletions
+4 -1
View File
@@ -122,10 +122,13 @@ describe('POST /api/lead', () => {
}); });
it('500 server: исключение при записи', async () => { it('500 server: исключение при записи', async () => {
const { post, db } = setup(); const errors: unknown[] = [];
const { post, db } = setup({ onServerError: (err) => void errors.push(err) });
db.close(); db.close();
const res = await post(validBody()); const res = await post(validBody());
expect(res.status).toBe(500); expect(res.status).toBe(500);
expect(await res.json()).toEqual({ ok: false, error: 'server' }); expect(await res.json()).toEqual({ ok: false, error: 'server' });
// onServerError → «Система»
expect(errors).toHaveLength(1);
}); });
}); });
+3
View File
@@ -17,6 +17,8 @@ export type AppDeps = {
/** Адреса, от которых верим X-Real-IP, помимо loopback. */ /** Адреса, от которых верим X-Real-IP, помимо loopback. */
trustedProxies?: readonly string[]; trustedProxies?: readonly string[];
version?: string; version?: string;
/** 500: сообщить в «Систему». Ответ клиенту от этого не зависит. */
onServerError?: (err: unknown) => void;
}; };
export const MAX_BODY_BYTES = 16 * 1024; export const MAX_BODY_BYTES = 16 * 1024;
@@ -41,6 +43,7 @@ export function createApp(deps: AppDeps) {
app.onError((err) => { app.onError((err) => {
console.error(err); console.error(err);
deps.onServerError?.(err);
return reply({ ok: false, error: 'server' }, 500); return reply({ ok: false, error: 'server' }, 500);
}); });
+59
View File
@@ -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<string, string | undefined>;
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'),
},
};
}
+47 -21
View File
@@ -1,45 +1,71 @@
import { serve } from '@hono/node-server'; import { serve } from '@hono/node-server';
import { createApp } from './app.js'; import { createApp } from './app.js';
import { DEFAULT_DB_PATH, openDb } from './db/index.js'; import { loadConfig } from './config.js';
import { noopNotifier } from './notifier.js'; import { openDb, type Db } from './db/index.js';
import { noopNotifier, type Notifier } from './notifier.js';
import { createRateLimiter } from './rateLimit.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 config = loadConfig();
const db = openDb(process.env.DB_PATH ?? DEFAULT_DB_PATH); 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({ const app = createApp({
db, db,
now: () => new Date(), now,
rateLimiter: createRateLimiter(), rateLimiter: createRateLimiter(),
notifier: noopNotifier, notifier,
// В docker сокет nginx приходит не с 127.0.0.1, а со шлюза сети контейнера — его задаём явно. trustedProxies: config.trustedProxies,
trustedProxies: (process.env.TRUSTED_PROXIES ?? '') version: config.version,
.split(',') onServerError: (err) => void system.report(`500 ${err instanceof Error ? err.message : ''}`.slice(0, 120), `500: ${shortError(err)}`),
.map((s) => s.trim())
.filter(Boolean),
version: process.env.APP_VERSION,
}); });
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}`); console.log(`api listening on :${info.port}`);
outbox?.start();
bot?.start();
void system.report('started', `api запущен, версия ${config.version}`);
}); });
let stopping = false; let stopping = false;
function shutdown(signal: string) { async function shutdown(signal: string) {
if (stopping) return; if (stopping) return;
stopping = true; stopping = true;
console.log(`${signal}: останавливаемся`); console.log(`${signal}: останавливаемся`);
setTimeout(() => {
console.error('не дождались остановки, выходим');
process.exit(1);
}, 10_000).unref();
await bot?.stop();
await outbox?.stop();
// Сначала дожидаемся текущих запросов, потом закрываем базу — иначе запись оборвётся на полуслове. // Сначала дожидаемся текущих запросов, потом закрываем базу — иначе запись оборвётся на полуслове.
server.close(() => { server.close(() => {
db.close(); db.close();
process.exit(0); process.exit(0);
}); });
setTimeout(() => {
console.error('не дождались запросов, выходим');
db.close();
process.exit(1);
}, 10_000).unref();
} }
process.on('SIGTERM', () => shutdown('SIGTERM')); process.on('SIGTERM', () => void shutdown('SIGTERM'));
process.on('SIGINT', () => shutdown('SIGINT')); process.on('SIGINT', () => void shutdown('SIGINT'));
+129
View File
@@ -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<typeof fakeTelegram>[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]);
});
});
+162
View File
@@ -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<void>;
start(): void;
stop(): Promise<void>;
};
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<void> | 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<void>((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;
},
};
}
+110
View File
@@ -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<unknown>): Promise<TelegramError> {
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: неверный формат$/);
});
});
+143
View File
@@ -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<Response>;
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<typeof createTelegramClient>;
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<T>(
method: string,
params: Record<string, unknown> | FormData = {},
opts: { timeoutMs?: number; signal?: AbortSignal } = {},
): Promise<T> {
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<TgUser>('getMe'),
sendMessage: (params: SendMessageParams) => call<TgMessage>('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<TgMessage>('sendDocument', form, { timeoutMs: Math.max(timeoutMs, 60_000) });
},
getUpdates: (
params: { offset?: number; timeout: number; allowed_updates: string[] },
signal?: AbortSignal,
) => call<TgUpdate[]>('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<TgChat>('getChat', { chat_id }),
getChatMember: (chat_id: number, user_id: number) => call<TgChatMember>('getChatMember', { chat_id, user_id }),
answerCallbackQuery: (callback_query_id: string, text?: string) =>
call<boolean>('answerCallbackQuery', { callback_query_id, text }),
editMessageText: (params: { chat_id: number; message_id: number; text: string; reply_markup?: InlineKeyboard }) =>
call<TgMessage | boolean>('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<Uint8Array> {
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;
},
};
}
+36
View File
@@ -0,0 +1,36 @@
// Фейковый Telegram для тестов: записывает вызовы, ответы и сбои задаются по методу.
import type { TelegramClient } from './client.js';
type Method = Exclude<keyof TelegramClient, 'call'>;
type Handler = (params: any) => unknown;
export function fakeTelegram(handlers: Partial<Record<Method, Handler>> = {}) {
const calls: { method: Method; params: any }[] = [];
let messageId = 0;
const defaults: Partial<Record<Method, Handler>> = {
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<string, unknown>, {
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 };
}
+78
View File
@@ -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: '<b>Аня</b> & Ко', comment: '<a href="x">клик</a>' }, NOW);
expect(text).toContain('Имя: &lt;b&gt;Аня&lt;/b&gt; &amp; Ко');
expect(text).toContain('Комментарий: &lt;a href="x"&gt;клик&lt;/a&gt;');
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<b>', interest: 'question', comment: 'x', changed: ['name'] }),
).toBe('🟨 Уточнение к заявке #3\nИмя: A&lt;b&gt;');
});
});
it('escapeHtml', () => {
expect(escapeHtml('a & b < c > d')).toBe('a &amp; b &lt; c &gt; d');
});
+77
View File
@@ -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('&', '&amp;').replaceAll('<', '&lt;').replaceAll('>', '&gt;');
}
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');
}
+158
View File
@@ -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<typeof validBody>[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);
});
});
+183
View File
@@ -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<void>;
start(): void;
stop(): Promise<void>;
};
/** Пауза перед попыткой 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<TelegramClient, 'sendMessage'>;
chatId?: number;
threadId?: number;
now: () => Date;
system: SystemReporter;
intervalMs?: number;
}): Outbox {
const enabled = client !== undefined && chatId !== undefined;
let chain: Promise<void> = 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<boolean> {
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<void> {
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<void> {
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;
},
};
}
+63
View File
@@ -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;
+64
View File
@@ -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<void>;
};
export function createSystemReporter({
client,
chatId,
threadId,
now,
log = (line: string) => console.log(line),
}: {
client?: Pick<TelegramClient, 'sendMessage'>;
chatId?: number;
threadId?: number;
now: () => Date;
log?: (line: string) => void;
}): SystemReporter {
// Одинаковые события не чаще раза в 10 минут; пропущенные копим и показываем с ближайшим отправленным.
const throttle = new Map<string, { sentAt: number; skipped: number }>();
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');
}